From 43997e8a56f1f2f7faf743027e40efc23906b3e4 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 03:17:28 +0800 Subject: [PATCH 1/4] fix(benchmark): continue verified interrupted native hosts without rewinding state Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/README.md | 60 ++++++ benchmark/edgebench/stopped_recovery.py | 186 ++++++++++++++++ benchmark/runtime/interruption_recovery.py | 204 ++++++++++++++++++ benchmark/runtime/scheduler.py | 50 ++++- benchmark/runtime/worker.py | 14 +- benchmark/tests/test_interruption_recovery.py | 186 ++++++++++++++++ benchmark/tests/test_stopped_recovery.py | 135 ++++++++++++ loopx/hosts/codex_checkpoint.py | 56 ++++- scripts/external_scheduler_worker.py | 8 +- 9 files changed, 884 insertions(+), 15 deletions(-) create mode 100644 benchmark/edgebench/stopped_recovery.py create mode 100644 benchmark/runtime/interruption_recovery.py create mode 100644 benchmark/tests/test_interruption_recovery.py create mode 100644 benchmark/tests/test_stopped_recovery.py diff --git a/benchmark/README.md b/benchmark/README.md index d219238a4f..85d8c8e559 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -1377,3 +1377,63 @@ This used real Codex processes and native HTTP with deterministic local inferenc and synthetic grading. It does **not** qualify scientific research utility, matched strategy results, or the frontend restore journey. Actual task/profile admission and measured continuation outcomes remain required. + +### Continue an interrupted Host in its retained filesystem + +The explicit `benchmark.runtime.scheduler` budget-continuation entrypoint can +settle a returned, budget-interrupted native Host before the next normal wake. +It retains the original scheduler attempt, wake receipt, Goal, Todos and native +conversation. A new append-only phase records the exact source and one finite +deadline; retrying that phase cannot refill time. The next wake still performs +ordinary quota, Stop, current-profile and session checks. No tool is replayed, +work marked complete, or replan obligation discarded by this settlement. + +New scheduler-owned attempts carry an attempt identity and passive process +cleanup observations. Recovery requires one matching returned budget receipt, +a terminated original process group, the unchanged session binding, and stable +native history ending in an interrupted Turn within that process window. +Unknown failures, live processes, pending native effects, ambiguous receipts, +changed profiles and stopped Goals remain held. Ordinary scheduler invocation +does not opt into continuation. Completed-checkpoint admission stays unchanged. + +An older worker without the process observation can instead retain its exact +stopped Docker filesystem. Prepare evidence through the existing Docker archive +provider; do not substitute an older sealed checkpoint or copy files over the +current workspace. The declaration names the original `goal_id`, `agent_id`, +`workspace`, `runtime_root`, `native_home`, `registry`, `scheduler_state`, +`wake_log_dir`, `phase_clock` and `wake_turn_id` (absolute guest paths for paths). +For example: + +```sh +python -m benchmark.edgebench.stopped_recovery \ + --source-container FULL_CONTAINER_ID \ + --declaration /private/recovery/source.json \ + --output /private/recovery/source-proof.json --install +``` + +The command refuses a running container, non-inert restart command or external +mounts. It inventories the current workspace, runtime, registry, original wake, +native history and SQLite WAL/journal companions, then verifies the container +remained stopped. `--install` places and reads back only a content-addressed +inert proof in that same container's control directory; omitting it is read-only. +Neither form starts the container, restores scoring authority or calls a model. +Keep exclusive ownership of the source during preparation and continuation. + +After independently restoring the original native evaluator's history, +opportunities and actual cooldown, restart only the same inert container and +pass its returned `installed_proof_path` as `LOOPX_INTERRUPTED_SOURCE` to the +explicit scheduler continuation. Preserve all original guest paths, native +configuration and session. Hot-upgraded LoopX source belongs outside inventoried +directories. Supply the separately admitted remaining allowance through the +existing `LOOPX_PHASE_DEADLINE_EPOCH`; leave the original phase-clock file intact +for source verification. Any changed participant, including a missing WAL, +refuses settlement. A held reason is printed and the original phase remains +selected; inspect the original evidence rather than delete it or select fresh. + +This provider path qualifies transport recovery, not a new sealed research point +or cross-container restore. It never grants evaluator registration, budget, +execution authorization or scientific comparability. Keep the prior failed +segment and new segment separately attributable, and verify actual model activity +and native feedback after dispatch. Disposable tests exercise real Docker archive +and proof installation; SQLite/process tests exercise interrupted work followed +by the same session's next Turn without paid model calls. diff --git a/benchmark/edgebench/stopped_recovery.py b/benchmark/edgebench/stopped_recovery.py new file mode 100644 index 0000000000..916d65a583 --- /dev/null +++ b/benchmark/edgebench/stopped_recovery.py @@ -0,0 +1,186 @@ +"""Prepare inert evidence from one stopped worker's unchanged filesystem. + +The Docker provider proves namespace termination. The runtime recovery owner +checks these exact bytes and the native session before any new admission. +This command never starts a container, restores a judge, or invokes a model. +""" +from __future__ import annotations + +import argparse +from datetime import datetime +import hashlib +import io +import json +import os +from pathlib import Path +import re +import tarfile +import tempfile +import time + +from .checkpoint_worker import _archive_component, _guest_path + + +def file_inventory(root: Path) -> list[dict]: + entries = [root] if root.is_file() else sorted(root.rglob('*')) + result = [] + for path in entries: + if path.is_symlink(): + raise ValueError('Interrupted source inventory cannot contain symlinks') + if path.name.endswith('-shm'): + continue # SQLite rebuilds this reader-only index. + if path.is_file(): + h = hashlib.sha256() + with path.open('rb') as stream: + while data := stream.read(1024 * 1024): + h.update(data) + result.append({'path': '' if path == root else str(path.relative_to(root)), + 'bytes': path.stat().st_size, 'sha256': h.hexdigest()}) + elif not path.is_dir(): + raise ValueError('Interrupted source inventory contains a special file') + return result + + +def source_paths(declaration: dict) -> dict[str, str]: + required = {'goal_id', 'agent_id', 'workspace', 'runtime_root', 'native_home', + 'registry', 'scheduler_state', 'wake_log_dir', 'phase_clock', 'wake_turn_id'} + if set(declaration) != required: + raise ValueError('Stopped recovery requires its exact lane and original source paths') + for key in required - {'goal_id', 'agent_id', 'wake_turn_id'}: + _guest_path(declaration[key]) + if not re.fullmatch(r'wake-[0-9]+-[a-f0-9]{12}', declaration['wake_turn_id']): + raise ValueError('Stopped recovery requires the original worker wake identity') + home = declaration['native_home'] + # Explicit native participants exclude auth, caches and credential stores. + return {**{k: declaration[k] for k in ('workspace', 'runtime_root', 'registry', + 'scheduler_state', 'wake_log_dir', 'phase_clock')}, + 'host_pending': declaration['scheduler_state'] + '.host-attempt', + 'native_sessions': home + '/sessions', 'native_state': home + '/state_5.sqlite', + 'native_history': home + '/thread_history_1.sqlite', 'native_config': home + '/config.toml'} + + +def prepare_stopped_recovery(client, *, container_id: str, declaration: dict, + output: Path, timeout: float = 120, install: bool = False) -> dict: + if not re.fullmatch(r'[a-f0-9]{64}', container_id): + raise ValueError('Stopped recovery requires one full container identity') + paths = source_paths(declaration) + if output.exists() or output.is_symlink() or output.parent.resolve() != output.parent: + raise ValueError('Stopped recovery output must be fresh and physical') + box = client.containers.get(container_id) + box.reload() + before = dict(box.attrs['State']) + if (box.id != container_id or before.get('Status') != 'exited' + or before.get('Running') is not False or before.get('Pid') != 0 + or before.get('Restarting') or before.get('Paused') or before.get('Dead')): + raise ValueError('Original worker namespace is not demonstrably stopped') + # This source can be restarted as an inert shell, never by replaying its + # original install/run command. Actual restart remains the caller's action. + if box.attrs['Config'].get('Entrypoint') or box.attrs['Config'].get('Cmd') != ['tail', '-f', '/dev/null']: + raise ValueError('Original worker must have an inert restart command') + if box.attrs.get('Mounts'): + raise ValueError('Stopped worker has externally writable mounts') + finished = datetime.fromisoformat(before['FinishedAt'].replace('Z', '+00:00')).timestamp() + deadline = time.time() + timeout + components = [] + with tempfile.TemporaryDirectory(prefix='.stopped-source-', dir=output.parent) as temporary: + copied = {} + for key, remote in paths.items(): + local = _archive_component(box, remote, Path(temporary) / key, deadline=deadline) + copied[key] = local + # SQLite companions are declared separately: a WAL is committed + # source data, not an optional display/cache file. + for suffix in ('', '-wal', '-journal') if local.is_file() else ('',): + member = Path(str(local) + suffix) + if member.exists(): + components.append({'source_path': remote + suffix, + 'files': file_inventory(member), 'directory': member.is_dir()}) + state_raw = copied['scheduler_state'].read_bytes() + state = json.loads(state_raw) + if (state.get('host_attempt', {}).get('status') != 'started' + or json.loads(copied['host_pending'].read_bytes()) != {'pending': True}): + raise ValueError('Stopped legacy recovery requires the original unresolved Host attempt') + wake = copied['wake_log_dir'] / declaration['wake_turn_id'] / 'receipt.json' + raw = wake.read_bytes() + receipt = json.loads(raw) + candidates = [p for p in copied['wake_log_dir'].glob('*/receipt.json') + if p.stat().st_mtime > copied['scheduler_state'].stat().st_mtime] + if candidates != [wake]: + raise ValueError('Stopped source has no unique returned wake after its Host attempt') + if (receipt.get('turn_id') != declaration['wake_turn_id'] or receipt.get('host_invoked') is not True + or receipt.get('budget_exhausted') is not True or receipt.get('mode') != 'heartbeat' + or receipt.get('stage') != 'execute' or receipt.get('ok') is not False + or 'host_failure' in receipt): + raise ValueError('Stopped source lacks an exact interrupted budget receipt') + start = int(declaration['wake_turn_id'].split('-')[1]) / 1_000_000_000 + phase = float(copied['phase_clock'].read_text()) + if not start <= phase <= finished: + raise ValueError('Stopped source phase did not expire during the original wake') + result = {'container_id': box.id, 'source_state': before, 'declaration': declaration, + 'components': components, 'scheduler_state_sha256': hashlib.sha256(state_raw).hexdigest(), + 'receipt_sha256': hashlib.sha256(raw).hexdigest(), + 'process_started_at': start, 'process_stopped_at': finished, + 'source_phase_deadline_epoch': phase, 'session': receipt.get('session'), + 'execution_authority_granted': False} + box.reload() + if box.attrs['State'] != before: + raise ValueError('Original worker changed during stopped-source readback') + if install: + from docker.errors import NotFound + + raw = json.dumps(result, sort_keys=True).encode() + name = 'interruption-source-' + hashlib.sha256(raw).hexdigest() + '.json' + parent = Path(declaration['scheduler_state']).parent + if any(parent == Path(p) or Path(p) in parent.parents for p in + (declaration['workspace'], declaration['runtime_root'], declaration['wake_log_dir'], declaration['native_home'])): + raise ValueError('Recovery proof must not mutate its inventoried source directories') + # A stopped worker's inert control directory is the sole destination; + # never place the proof in a different container or overwrite old state. + package = io.BytesIO() + with tarfile.open(fileobj=package, mode='w') as archive: + member = tarfile.TarInfo(name) + member.mode, member.size = 0o600, len(raw) + archive.addfile(member, io.BytesIO(raw)) + box.reload() + if box.attrs['State'] != before: + raise ValueError('Original worker changed before proof installation') + with tempfile.TemporaryDirectory(prefix='.proof-existing-', dir=output.parent) as directory: + try: + existing = _archive_component(box, str(parent / name), Path(directory) / 'proof', deadline=time.time() + timeout) + except NotFound: + existing = None + if existing is not None and existing.read_bytes() != raw: + raise ValueError('Original worker recovery proof path is already occupied') + if existing is None and not box.put_archive(str(parent), package.getvalue()): + raise ValueError('Original worker did not accept its inert recovery proof') + with tempfile.TemporaryDirectory(prefix='.proof-readback-', dir=output.parent) as directory: + copied = _archive_component(box, str(parent / name), Path(directory) / 'proof', deadline=time.time() + timeout) + if copied.read_bytes() != raw: + raise ValueError('Original worker recovery proof readback differs') + box.reload() + if box.attrs['State'] != before: + raise ValueError('Original worker changed during proof installation') + result['installed_proof_path'] = str(parent / name) + with os.fdopen(os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), 'w') as stream: + json.dump(result, stream, indent=2) + stream.flush() + os.fsync(stream.fileno()) + return result + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--source-container', required=True) + parser.add_argument('--declaration', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--install', action='store_true', help='Place the inert proof only in this same stopped worker') + args = parser.parse_args() + import docker + result = prepare_stopped_recovery(docker.from_env(timeout=30), + container_id=args.source_container, declaration=json.loads(args.declaration.read_bytes()), + output=args.output, install=args.install) + print(json.dumps({'container_id': result['container_id'], 'output': str(args.output), + 'installed_proof_path': result.get('installed_proof_path'), 'execution_authority_granted': False})) + + +if __name__ == '__main__': + main() diff --git a/benchmark/runtime/interruption_recovery.py b/benchmark/runtime/interruption_recovery.py new file mode 100644 index 0000000000..8ac96bea8a --- /dev/null +++ b/benchmark/runtime/interruption_recovery.py @@ -0,0 +1,204 @@ +"""Reconcile one interrupted benchmark Host, without certifying its work. + +This is Python provider/process IO around the existing Codex session, native +history and external scheduler owners. It does not decide Goal/Todo recovery, +grant budget, replay a tool, restore judge authority or manufacture a checkpoint. +""" +from __future__ import annotations + +import hashlib +import json +import math +import os +from pathlib import Path +import re + + +class ProcessObservers: + """Preserve request-admission cleanup while retaining process observations.""" + def __init__(self, *observers): + self.observers = [observer for observer in observers if observer is not None] + + def started(self, process): + for observer in self.observers: + observer.started(process) + + def before_cleanup(self, process, error): + for observer in self.observers: + observer.before_cleanup(process, error) + + def signal_requested(self, number): + for observer in self.observers: + observer.signal_requested(number) + + def after_cleanup(self, process, error): + for observer in self.observers: + observer.after_cleanup(process, error) + + +def _physical_read(path: Path) -> bytes: + if path.is_symlink() or path.resolve(strict=True) != path or not path.is_file(): + raise ValueError("Interruption evidence must be a physical regular file") + return path.read_bytes() + + +def _number(value): + if type(value) not in (int, float) or not math.isfinite(value): + raise ValueError("Interruption evidence requires finite process times") + return value + + +def _process_window(receipt): + process = receipt.get("native_process", {}) + after = process.get("after_cleanup") or {} + if (process.get("schema_version") != "native_codex_process_observation_v1" + or type(process.get("pid")) is not int or process["pid"] <= 1 + or type(after.get("return_code")) is not int + or process.get("cleanup_error_type") is not None + or process.get("persistence_error_type") is not None): + raise ValueError("Original native process termination is unproven") + started = _number(process.get("started_at_unix")) + stopped = _number(after.get("observed_at_unix")) + if started > stopped: + raise ValueError("Original native process window is invalid") + # The worker owns and reaps this process group. A still-live group is a + # conservative hold even if the PID has since been reused. + if os.name != "posix": + raise ValueError("Original process-group termination is unavailable") + try: + os.killpg(process["pid"], 0) + except ProcessLookupError: + return started, stopped + except PermissionError as error: + raise ValueError("Original process-group termination is unavailable") from error + raise ValueError("Original native process group is still alive") + + +def _stopped_source(state_file: Path, state: dict, env: dict[str, str], wake_log_dir: str): + """Consume the Docker provider's exact, explicitly prepared legacy source. + + This is only valid on that unchanged filesystem. A proof cannot be combined + with an older checkpoint or with a newly created Goal/native home. + """ + from benchmark.edgebench.stopped_recovery import file_inventory, source_paths + + proof_path = Path(env['LOOPX_INTERRUPTED_SOURCE']).absolute() + raw = _physical_read(proof_path) + proof = json.loads(raw) + declaration = proof['declaration'] + stopped = proof['source_state'] + expected = {'goal_id': env['LOOPX_GOAL_ID'], 'agent_id': env['LOOPX_AGENT_ID'], + 'workspace': env['LOOPX_PROJECT'], 'runtime_root': env['LOOPX_RUNTIME_ROOT'], + 'native_home': env['CODEX_HOME'], 'registry': env['LOOPX_REGISTRY'], + 'scheduler_state': str(state_file), 'wake_log_dir': wake_log_dir} + if (not re.fullmatch(r'[a-f0-9]{64}', proof.get('container_id', '')) + or stopped.get('Status') != 'exited' or stopped.get('Running') is not False + or stopped.get('Pid') != 0 or stopped.get('Restarting') or stopped.get('Paused') + or stopped.get('Dead') or proof.get('execution_authority_granted') is not False + or any(declaration.get(k) != v for k, v in expected.items()) + or hashlib.sha256(_physical_read(state_file)).hexdigest() != proof['scheduler_state_sha256']): + raise ValueError('Stopped recovery proof differs from the original worker') + required = set(source_paths(declaration).values()) + supplied = {part['source_path'] for part in proof['components']} + if not required <= supplied or any(p not in required and not any( + p == base + suffix for base in required for suffix in ('-wal', '-journal')) for p in supplied): + raise ValueError('Stopped recovery proof omits or adds source participants') + if len(supplied) != len(proof['components']): + raise ValueError('Stopped recovery proof repeats source participants') + for base in required: + for suffix in ('-wal', '-journal'): + if Path(base + suffix).exists() != (base + suffix in supplied): + raise ValueError('Stopped recovery source SQLite companions changed') + for component in proof['components']: + root = Path(component['source_path']) + if root.resolve(strict=True) != root or root.is_symlink() or root.is_dir() != component['directory']: + raise ValueError('Stopped recovery source participant moved') + if file_inventory(root) != component['files']: + raise ValueError('Stopped recovery source bytes changed') + path = Path(wake_log_dir) / declaration['wake_turn_id'] / 'receipt.json' + receipt_raw = _physical_read(path) + if hashlib.sha256(receipt_raw).hexdigest() != proof['receipt_sha256']: + raise ValueError('Stopped recovery original wake changed') + receipt = json.loads(receipt_raw) + return path, receipt_raw, receipt, proof + + +def read_interrupted_host(state: dict, *, state_file: Path, env: dict[str, str], wake_log_dir: str) -> dict: + """Qualify an exact returned wake, then inspect its existing native session. + + No caller-supplied terminal boolean or inferred last score authorizes this + path. Missing legacy evidence remains a recovery gap, not a new attempt. + """ + from .scheduler import BUDGET_EXHAUSTED_EXIT_CODE + from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession + from loopx.hosts.codex_checkpoint import native_interrupted_turn + from .codex import process_environment + from scripts.external_scheduler_worker import _host_pending + + # Match the worker's original isolated home, not an operator's ambient one. + env = process_environment(Path(env["LOOPX_CODEX_HOME"]), base=env) + if _host_pending(state_file) is not True: + raise ValueError("Original Host pending witness is unresolved or already cleared") + + prior = state.get("host_attempt", {}) + attempt = prior.get("attempt_id", "") + legacy = bool(env.get('LOOPX_INTERRUPTED_SOURCE')) and not attempt + if ((not legacy and not re.fullmatch(r"[0-9a-f]{32}", attempt)) + or prior.get("status") not in {"started", "stopped"} + or "host_failure" in prior + or (prior["status"] == "stopped" and prior.get("stop_exit_code") != BUDGET_EXHAUSTED_EXIT_CODE)): + raise ValueError("Original Host interruption identity is unavailable") + source = None + if legacy: + path, raw, receipt, source = _stopped_source(state_file, state, env, wake_log_dir) + else: + matches = [] + root = Path(wake_log_dir).absolute() + for path in root.glob("*/receipt.json"): + raw = _physical_read(path) + try: + receipt = json.loads(raw) + except ValueError: + continue + if isinstance(receipt, dict) and receipt.get("host_attempt_id") == attempt: + matches.append((path, raw, receipt)) + if len(matches) != 1: + raise ValueError("Original Host interruption must have one exact wake receipt") + path, raw, receipt = matches[0] + if (receipt.get("turn_id") != path.parent.name or receipt.get("mode") != "heartbeat" + or receipt.get("stage") != "execute" or receipt.get("host_invoked") is not True + or receipt.get("budget_exhausted") is not True or receipt.get("ok") is not False + or "host_failure" in receipt + or (not legacy and any(receipt.get(key) != env["LOOPX_" + key.upper()] + for key in ("goal_id", "agent_id")))): + raise ValueError("Original Host receipt is not this lane's interrupted budget stop") + started, stopped = ((_number(source['process_started_at']), _number(source['process_stopped_at'])) + if legacy else _process_window(receipt)) + try: + phase = float(source['source_phase_deadline_epoch'] if legacy else receipt["phase_deadline_epoch"]) + except (KeyError, TypeError, ValueError) as error: + raise ValueError("Original Host phase clock is unavailable") from error + if not math.isfinite(phase) or not started <= phase <= stopped: + raise ValueError("Original Host stopped outside its phase deadline") + session = receipt.get("session") or {} + if session.get("binding_scope") != "agent" or not session.get("session_id"): + raise ValueError("Original Host session is unavailable") + current = CodexAgentSession(env, context="resume", + sandbox=env.get("LOOPX_CODEX_SANDBOX", "danger-full-access"), receipt={}) + with current.admission.launch_lifetime(operation="benchmark_interrupted_host_recovery"): + pass # Also checks legacy source/mirror activation, unlike require_current alone. + if (not current.binding or current.session_id != session["session_id"] + or prior.get("session_id") not in (None, current.session_id)): + raise ValueError("Original Host session binding changed") + native = native_interrupted_turn(env, current.session_id, + process_started_at=started, process_stopped_at=stopped) + with current.admission.launch_lifetime(operation="benchmark_interrupted_host_recovery"): + pass # A concurrent Stop or registry change cannot authorize the new phase. + if _physical_read(path) != raw: + raise ValueError("Original Host receipt changed during readback") + return {**({'source_container_id': source['container_id'], + 'source_proof_sha256': hashlib.sha256(_physical_read(Path(env['LOOPX_INTERRUPTED_SOURCE']))).hexdigest()} + if legacy else {'host_attempt_id': attempt}), "wake_turn_id": receipt["turn_id"], + "receipt_path": str(path), "receipt_sha256": hashlib.sha256(raw).hexdigest(), + "source_phase_deadline_epoch": phase, "session_profile_digest": current.digest, + "native": native} diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index 822cab9f99..f189651975 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -9,6 +9,7 @@ from pathlib import Path import re import shlex +import sys import time # Private process transport between the benchmark wake and its shell scheduler. @@ -89,19 +90,26 @@ def _phase_receipt(state_file: Path, wake_log_dir: str | None) -> dict | None: return value if isinstance(value, dict) else None -def _create_resume_state(target: Path, predecessor: Path) -> None: +def _create_resume_state(target: Path, predecessor: Path, *, recovered: dict | None = None, + replan_attempts=None) -> None: if target.is_symlink(): raise ValueError("Resumed scheduler state must not be a symlink") try: - with target.open("x", encoding="utf-8") as stream: - json.dump({"resume_from": str(predecessor)}, stream) + with os.fdopen(os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w", encoding="utf-8") as stream: + value = {"resume_from": str(predecessor)} + if recovered is not None: + value.update(interruption_recovery=recovered, + recovered_session_id=recovered["native"]["session_id"], + replan_attempts=replan_attempts or []) + json.dump(value, stream) + stream.flush() + os.fsync(stream.fileno()) except FileExistsError: return - target.chmod(0o600) def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, - wake_log_dir: str | None = None) -> str: + wake_log_dir: str | None = None, env: dict[str, str] | None = None) -> str: """Select a new transport phase only after an explicit budget continuation. Kernel state and native sessions stay in place. The original scheduler @@ -135,7 +143,12 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, if state is None: return str(predecessor) prior = state.get("host_attempt") - if not isinstance(prior, dict) or prior.get("status") != "stopped": + recovery = state.get("interruption_recovery") + if (recovery is not None and recovery.get("phase_deadline_epoch") != phase_deadline + and not (isinstance(prior, dict) and prior.get("status") == "stopped" + and prior.get("stop_exit_code") == BUDGET_EXHAUSTED_EXIT_CODE)): + raise ValueError("Recovered phase deadline is immutable; inspect the current phase before continuing") + if not isinstance(prior, dict): return str(predecessor) # Validate the captured record with its existing transport owner. scope = prior.get("scope") @@ -145,7 +158,27 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, _host_attempt(state, scope) except (TypeError, ValueError): return str(predecessor) - if (prior.get("stop_exit_code") != BUDGET_EXHAUSTED_EXIT_CODE + # A returned interrupted wake is distinct from a pre-dispatch stop. + # Explicit continuation only reconciles transport; all Goal/Stop, + # replan, lease and launch decisions remain with the normal next wake. + recovered = None + if env is not None and wake_log_dir and _host_pending(predecessor) is not None: + from .interruption_recovery import read_interrupted_host + + try: + recovered = read_interrupted_host(state, state_file=predecessor, env=env, wake_log_dir=wake_log_dir) + except (OSError, ValueError, RuntimeError, TimeoutError, KeyError, TypeError) as error: + print(f"Interrupted Host remains held: {type(error).__name__}: {error}", file=sys.stderr) + if recovered is not None: + suffix = hashlib.sha256(float(phase_deadline).hex().encode()).hexdigest()[:20] + target = path.with_name(f"{path.stem}.phase-{suffix}{path.suffix}") + recovered.update(source_state_sha256=hashlib.sha256(predecessor.read_bytes()).hexdigest(), + phase_deadline_epoch=phase_deadline) + _create_resume_state(target, predecessor, recovered=recovered, + replan_attempts=state.get("replan_attempts")) + return str(target) + if (prior.get("status") != "stopped" + or prior.get("stop_exit_code") != BUDGET_EXHAUSTED_EXIT_CODE or "host_failure" in prior or _host_pending(predecessor) is None or state.get("last_wake_rc") != BUDGET_EXHAUSTED_EXIT_CODE or state.get("last_wake_status") != "wake_requested_stop"): @@ -206,7 +239,8 @@ def main() -> int: args = parser.parse_args() state = resume_state_file(args.state_file, phase_deadline=float(os.environ["LOOPX_PHASE_DEADLINE_EPOCH"]), - agent_id=os.environ["LOOPX_AGENT_ID"], wake_log_dir=os.environ.get("LOOPX_WAKE_LOG_DIR")) + agent_id=os.environ["LOOPX_AGENT_ID"], wake_log_dir=os.environ.get("LOOPX_WAKE_LOG_DIR"), + env=dict(os.environ)) command = worker_command(dict(os.environ), python=args.python, source=args.source, state_file=state, host_timeout=args.host_timeout) os.execvpe(command[0], command, os.environ) diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index 968539ed90..f805b71527 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -207,11 +207,18 @@ def native_child_process(command, *, env, stdout, stderr, session_wake, receipt, guard = ProviderRequestAdmission(env, session_wake, receipt, wake, timeout=timeout, original_quota=original_quota) scope = guard.serve(command) + observation = guard + if env.get("LOOPX_HOST_ATTEMPT_ID") and session_wake is not None: + from loopx.capabilities.benchmark_toolkit.native_codex_process import NativeCodexProcessObservation + from .interruption_recovery import ProcessObservers + + native_observation = NativeCodexProcessObservation(lambda value: receipt.update(native_process=value)) + observation = ProcessObservers(native_observation, guard) with scope as routed: with child_process(routed, env=env, stdout=stdout, stderr=stderr, goal_admission=session_wake.admission if session_wake is not None else None, check_current_launch=heartbeat_launch_check(env, session_wake, original_quota=original_quota), - **({"process_observation": guard} if guard is not None else {})) as process: + **({"process_observation": observation} if observation is not None else {})) as process: yield process if guard is not None and guard.refusal_code is not None: from loopx.control_plane.goals.first_party_host_admission import FirstPartyHostRuntimeRejected @@ -548,6 +555,11 @@ def run_once(env: dict[str, str]) -> dict: "ok": False, "timed_out": False, } + if env.get("LOOPX_HOST_ATTEMPT_ID"): + # The scheduler supplies the transport identity, never model prose. + receipt.update(host_attempt_id=env["LOOPX_HOST_ATTEMPT_ID"], + goal_id=env.get("LOOPX_GOAL_ID"), agent_id=env.get("LOOPX_AGENT_ID"), + phase_deadline_epoch=env.get("LOOPX_PHASE_DEADLINE_EPOCH")) session_wake = None phase_work_deadline = None checkpoint_context = ExitStack() diff --git a/benchmark/tests/test_interruption_recovery.py b/benchmark/tests/test_interruption_recovery.py new file mode 100644 index 0000000000..838a5a973e --- /dev/null +++ b/benchmark/tests/test_interruption_recovery.py @@ -0,0 +1,186 @@ +"""Retained interrupted work resumes through ordinary admission, never replay.""" +from __future__ import annotations + +from contextlib import contextmanager +import copy +import json +import os +from pathlib import Path +import shlex +import sys +import time + +import pytest + +from benchmark.runtime.interruption_recovery import _process_window +from benchmark.runtime.scheduler import resume_state_file +from loopx.hosts import codex_checkpoint as native + + +def _history(monkeypatch, tmp_path, *, change=None): + tail = dict(id="last", status="interrupted", error=None, startedAt=110, completedAt=120, + items=[dict(type="commandExecution", status="failed")]) + turns = [dict(id="first", status="completed", items=[]), tail] + if change == "in_progress": + tail["status"] = "inProgress" + elif change == "pending_effect": + tail["items"][0]["status"] = "inProgress" + elif change == "outside_window": + tail["startedAt"] = 90 + elif change == "completed": + tail["status"] = "completed" + elif change == "duplicate": + turns[0]["id"] = "last" + calls = [] + + class Transport: + def request(self, method, params): + calls.append(method) + if method == "thread/read": + return {"thread": {"id": "other" if change == "wrong_session" else "session", + "cwd": str(tmp_path)}} + assert method == "thread/turns/list" # No native resume/interrupt/model method. + result = copy.deepcopy(turns) + if change == "changed_history" and calls.count(method) > 1: + result[0]["items"] = [dict(type="agentMessage", text="later")] + return dict(data=result, nextCursor=None) + + @contextmanager + def reader(*args, **kwargs): + yield Transport() + + monkeypatch.setattr(native, "native_checkpoint_reader", reader) + return calls + + +@pytest.mark.parametrize("change", ["in_progress", "pending_effect", "outside_window", "completed", + "duplicate", "wrong_session", "changed_history"]) +def test_native_interruption_rejects_unproven_or_changed_tail(monkeypatch, tmp_path, change): + _history(monkeypatch, tmp_path, change=change) + with pytest.raises(ValueError): + native.native_interrupted_turn({"LOOPX_PROJECT": str(tmp_path)}, "session", + process_started_at=100, process_stopped_at=125) + + +def test_interruption_readback_does_not_relax_completed_checkpoint(monkeypatch, tmp_path): + calls = _history(monkeypatch, tmp_path) + result = native.native_interrupted_turn({"LOOPX_PROJECT": str(tmp_path)}, "session", + process_started_at=100, process_stopped_at=125) + assert result["turn_count"] == 2 and result["turn_id"] == "last" + assert calls == ["thread/read", "thread/turns/list", "thread/turns/list"] + with native.native_checkpoint_reader({}, timeout=10) as transport: + with pytest.raises(ValueError, match="completed final cursor"): + native._complete_history(transport.request, "session") + + +@pytest.mark.skipif(os.name != "posix", reason="Process-group termination requires POSIX") +def test_live_process_group_cannot_be_settled(): + evidence = dict(native_process=dict(schema_version="native_codex_process_observation_v1", + pid=os.getpgrp(), started_at_unix=100, after_cleanup=dict(return_code=0, observed_at_unix=110))) + with pytest.raises(ValueError, match="still alive"): + _process_window(evidence) + + +@pytest.mark.skipif(os.name != "posix", reason="Owned process-group cleanup requires POSIX") +@pytest.mark.parametrize("fault", [None, "profile", "stop", "wrong_session", "duplicate_wake"]) +def test_real_sqlite_scheduler_retains_interrupted_session_then_admits_next_turn(tmp_path, monkeypatch, fault): + from benchmark.runtime import worker + from benchmark.tests.test_worker_phase_budget import _phase_worker + from tests import test_external_scheduler_worker as scheduler_fixture + + env, project, _runtime, _source, goal_state, _clock = _phase_worker(tmp_path, monkeypatch) + monkeypatch.setattr(worker, "time", time) + binary = Path(env["CODEX_BIN"]) + binary.write_text(f"#!{sys.executable}\n" + ''' +import json, os, pathlib, signal, sys, time +history=pathlib.Path('native-history.json') +if 'app-server' in sys.argv: + for line in sys.stdin: + message=json.loads(line) + if 'id' not in message: continue + method=message['method'] + if method=='initialize': result={} + elif method=='thread/read': result={'thread':{'id':'original-native','cwd':str(pathlib.Path.cwd())}} + elif method=='thread/turns/list': result={'data':[json.loads(history.read_text())],'nextCursor':None} + else: raise AssertionError(method) + print(json.dumps({'jsonrpc':'2.0','id':message['id'],'result':result}),flush=True) + sys.exit(0) +with pathlib.Path('native-argv').open('a') as f: f.write(json.dumps(sys.argv[1:])+'\\n') +print(json.dumps({'type':'thread.started','thread_id':'original-native'}),flush=True) +start=int(time.time()) +def interrupted(*args): + history.write_text(json.dumps({'id':'interrupted-work','status':'interrupted','error':None, + 'startedAt':start,'completedAt':int(time.time()), + 'items':[{'type':'commandExecution','status':'failed'}]})) + sys.exit(130) +signal.signal(signal.SIGINT,interrupted) +sys.stdin.read() +if pathlib.Path('continue').exists(): + print(json.dumps({'type':'turn.completed'}),flush=True) +else: + time.sleep(60) +''') + entry = tmp_path / "worker-entry.py" + entry.write_text("from benchmark.runtime import worker\n" + "worker.heartbeat_body=lambda *a,**kw:'synthetic authorized work'\n" + "raise SystemExit(worker.main())\n") + for key, value in env.items(): + monkeypatch.setenv(key, value) + args, _ = scheduler_fixture._guarded_host(tmp_path / "scheduler") + args.once = True + args.registry, args.runtime_root = env["LOOPX_REGISTRY"], env["LOOPX_RUNTIME_ROOT"] + args.goal_id, args.agent_id = env["LOOPX_GOAL_ID"], env["LOOPX_AGENT_ID"] + args.wake_cmd = shlex.join([sys.executable, str(entry)]) + args.wake_stop_exit_code, args.wake_timeout_seconds = 75, 30 + monkeypatch.setattr(scheduler_fixture.worker, "run_quota_should_run", + lambda *a, **kw: scheduler_fixture._hint_payload(should_run=True)) + initial_goal = goal_state.read_bytes() + monkeypatch.setenv("LOOPX_PHASE_DEADLINE_EPOCH", str(time.time() + 8)) + assert scheduler_fixture.worker.run_worker(args) == 0 + original = Path(args.state_file) + old_bytes = original.read_bytes() + old = json.loads(old_bytes) + assert old["host_attempt"]["status"] == "stopped" + receipts = list(Path(env["LOOPX_WAKE_LOG_DIR"]).glob("*/receipt.json")) + assert len(receipts) == 1 + first = json.loads(receipts[0].read_bytes()) + assert first["host_invoked"] and first["budget_exhausted"] + assert first["host_attempt_id"] == old["host_attempt"]["attempt_id"] + assert first["native_process"]["after_cleanup"]["return_code"] is not None + deadline = time.time() + 60 + monkeypatch.setenv("LOOPX_PHASE_DEADLINE_EPOCH", str(deadline)) + if fault == "profile": + monkeypatch.setenv("MODEL_NAME", "different-model") + elif fault == "stop": + from loopx.control_plane.goals.activation_service import set_goal_activation_state + result = set_goal_activation_state(registry_path=Path(args.registry), goal_id=args.goal_id, + state="stopped", reason="Owner stopped this work.", actor_kind="owner", execute=True) + assert result["ok"] + elif fault == "wrong_session": + first["session"]["session_id"] = "another-native" + receipts[0].write_text(json.dumps(first)) + elif fault == "duplicate_wake": + second_receipt = receipts[0].parent.with_name("duplicate") / "receipt.json" + second_receipt.parent.mkdir() + second_receipt.write_bytes(receipts[0].read_bytes()) + new = resume_state_file(str(original), phase_deadline=deadline, agent_id=args.agent_id, + wake_log_dir=env["LOOPX_WAKE_LOG_DIR"], env=dict(os.environ)) + if fault: + assert new == str(original) and original.read_bytes() == old_bytes + assert len((project / "native-argv").read_text().splitlines()) == 1 + return + assert new != str(original) + settlement = json.loads(Path(new).read_bytes()) + assert settlement["recovered_session_id"] == "original-native" + assert settlement["interruption_recovery"]["native"]["turn_id"] == "interrupted-work" + assert resume_state_file(str(original), phase_deadline=deadline, agent_id=args.agent_id, + wake_log_dir=env["LOOPX_WAKE_LOG_DIR"], env=dict(os.environ)) == new + with pytest.raises(ValueError, match="immutable"): + resume_state_file(str(original), phase_deadline=deadline + 60, agent_id=args.agent_id, + wake_log_dir=env["LOOPX_WAKE_LOG_DIR"], env=dict(os.environ)) + (project / "continue").touch() + args.state_file = new + assert scheduler_fixture.worker.run_worker(args) == 0 + argv = [json.loads(line) for line in (project / "native-argv").read_text().splitlines()] + assert len(argv) == 2 and "resume" in argv[1] and "original-native" in argv[1] + assert original.read_bytes() == old_bytes and goal_state.read_bytes() == initial_goal diff --git a/benchmark/tests/test_stopped_recovery.py b/benchmark/tests/test_stopped_recovery.py new file mode 100644 index 0000000000..7fcd52850d --- /dev/null +++ b/benchmark/tests/test_stopped_recovery.py @@ -0,0 +1,135 @@ +"""A stopped source proof observes existing bytes, and grants no execution.""" +from __future__ import annotations + +from datetime import datetime, timezone +import io +import json +import os +from pathlib import Path +import shutil +import tarfile +import time +from types import SimpleNamespace + +import pytest + +from benchmark.edgebench import stopped_recovery as subject +from benchmark.runtime.interruption_recovery import _stopped_source + + +def _source(root): + paths = dict(goal_id="fixture-goal", agent_id="fixture-agent", workspace=str(root / "workspace"), + runtime_root=str(root / "runtime"), native_home=str(root / "native"), + registry=str(root / "control/registry.json"), scheduler_state=str(root / "control/state.json"), + wake_log_dir=str(root / "wakes"), phase_clock=str(root / "control/deadline"), + wake_turn_id="wake-100000000000-deadbeefabcd") + for path in [Path(paths[k]) for k in ("workspace", "runtime_root", "native_home", "wake_log_dir")]: + path.mkdir(parents=True) + (root / "control").mkdir() + for file in ("workspace/answer", "runtime/goal", "native/state_5.sqlite", + "native/state_5.sqlite-wal", "native/thread_history_1.sqlite", "native/config.toml"): + (root / file).write_bytes(b"synthetic retained bytes") + (root / "native/sessions").mkdir() + (root / "native/sessions/session.jsonl").write_text("{}\n") + (root / "control/registry.json").write_text("{}") + (root / "control/deadline").write_text("105") + state = Path(paths["scheduler_state"]) + state.write_text(json.dumps(dict(host_attempt=dict(status="started", scope="a" * 64, attempt=1)))) + state.with_name(state.name + '.host-attempt').write_text('{"pending":true}') + os.utime(state, (99, 99)) + receipt = Path(paths["wake_log_dir"]) / paths["wake_turn_id"] / "receipt.json" + receipt.parent.mkdir() + receipt.write_text(json.dumps(dict(turn_id=paths["wake_turn_id"], host_invoked=True, budget_exhausted=True, + mode="heartbeat", stage="execute", ok=False, session=dict(binding_scope="agent", session_id="original")))) + os.utime(receipt, (110, 110)) + return paths + + +def _fake_container(monkeypatch): + box = SimpleNamespace(id="a" * 64, attrs={"State": dict(Status="exited", Running=False, Pid=0, + FinishedAt="1970-01-01T00:02:00Z"), "Config": dict(Cmd=["tail", "-f", "/dev/null"]), "Mounts": []}) + box.reload = lambda: None + def archive(container, remote, destination, **kwargs): + destination.mkdir() + source = Path(remote) + if source.is_dir(): + return Path(shutil.copytree(source, destination / source.name)) + copied = Path(shutil.copy2(source, destination / source.name)) + for suffix in ("-wal", "-journal"): + if Path(remote + suffix).exists(): + shutil.copy2(remote + suffix, str(copied) + suffix) + return copied + monkeypatch.setattr(subject, "_archive_component", archive) + return box, SimpleNamespace(containers=SimpleNamespace(get=lambda identity: box)) + + +@pytest.mark.parametrize("fault", [None, "live", "restart_command", "mount", "ambiguous_wake"]) +def test_prepare_stopped_source_keeps_original_state_and_requires_unique_return(tmp_path, monkeypatch, fault): + declaration = _source(tmp_path) + box, client = _fake_container(monkeypatch) + original = Path(declaration["scheduler_state"]).read_bytes() + if fault == "live": box.attrs["State"].update(Status="running", Running=True, Pid=42) + elif fault == "restart_command": box.attrs["Config"]["Cmd"] = ["run-original-job"] + elif fault == "mount": box.attrs["Mounts"] = [dict(Source="/writable")] + elif fault == "ambiguous_wake": + second = tmp_path / "wakes/second/receipt.json" + second.parent.mkdir() + second.write_text("{}") + output = tmp_path / "source-proof.json" + if fault: + with pytest.raises(ValueError): + subject.prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=output) + assert not output.exists() + else: + result = subject.prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=output) + assert result["execution_authority_granted"] is False + assert any(p["source_path"].endswith("state_5.sqlite-wal") for p in result["components"]) + assert Path(declaration["scheduler_state"]).read_bytes() == original + + +@pytest.mark.parametrize("fault", ["workspace", "wal_removed", "new_journal", "session_removed", "state", "wrong_lane"]) +def test_consumer_rejects_stale_or_incomplete_latest_source(tmp_path, monkeypatch, fault): + declaration = _source(tmp_path) + box, client = _fake_container(monkeypatch) + proof = tmp_path / "source-proof.json" + subject.prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=proof) + env = dict(LOOPX_INTERRUPTED_SOURCE=str(proof), LOOPX_GOAL_ID=declaration["goal_id"], + LOOPX_AGENT_ID=declaration["agent_id"], LOOPX_PROJECT=declaration["workspace"], + LOOPX_RUNTIME_ROOT=declaration["runtime_root"], CODEX_HOME=declaration["native_home"], + LOOPX_REGISTRY=declaration["registry"]) + if fault == "workspace": (tmp_path / "workspace/answer").write_text("different") + elif fault == "wal_removed": (tmp_path / "native/state_5.sqlite-wal").unlink() + elif fault == "new_journal": (tmp_path / "native/state_5.sqlite-journal").write_text("different") + elif fault == "session_removed": (tmp_path / "native/sessions/session.jsonl").unlink() + elif fault == "state": (tmp_path / "control/state.json").write_text("{}") + elif fault == "wrong_lane": env["LOOPX_AGENT_ID"] = "other" + with pytest.raises(ValueError): + _stopped_source(Path(declaration["scheduler_state"]), {}, env, declaration["wake_log_dir"]) + + +@pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1", reason="Explicit disposable Docker qualification") +def test_real_stopped_docker_proof_installation_is_inert_and_read_back(tmp_path): + docker = pytest.importorskip("docker") + client = docker.from_env(timeout=30) + box = client.containers.create("python:3.12-slim", ["tail", "-f", "/dev/null"], network_disabled=True) + local = tmp_path / "fixture" + local.mkdir() + declaration = _source(local) + declaration = {key: value.replace(str(local), "/fixture") for key, value in declaration.items()} + try: + package = io.BytesIO() + with tarfile.open(fileobj=package, mode="w") as archive: + archive.add(local, arcname="fixture") + assert box.put_archive("/", package.getvalue()) + box.start() + box.stop(timeout=2) + box.reload() + original = dict(box.attrs["State"]) + proof = subject.prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, + output=tmp_path / "proof.json", install=True) + assert proof["installed_proof_path"].startswith("/fixture/control/interruption-source-") + assert proof["execution_authority_granted"] is False + box.reload() + assert box.attrs["State"] == original + finally: + box.remove(force=True) diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py index c033ccf537..248b6abb80 100644 --- a/loopx/hosts/codex_checkpoint.py +++ b/loopx/hosts/codex_checkpoint.py @@ -25,7 +25,7 @@ from loopx.control_plane.turn_driver.codex_stdio import StdioNativeGoalTransport -def _complete_history(request, session_id: str, *, interrupted_guard: dict | None = None) -> list[dict[str, Any]]: +def _retained_history(request, session_id: str) -> list[dict[str, Any]]: turns = [] cursor = None seen = set() @@ -45,17 +45,65 @@ def _complete_history(request, session_id: str, *, interrupted_guard: dict | Non seen.add(cursor) if not turns or any(t.get("status") not in {"completed", "failed", "interrupted"} or not t.get("id") for t in turns): raise ValueError("native checkpoint requires complete retained turns") + ids = [t["id"] for t in turns] + if len(set(ids)) != len(ids): + raise ValueError("native checkpoint history contains duplicate turns") + return turns + + +def _complete_history(request, session_id: str, *, interrupted_guard: dict | None = None) -> list[dict[str, Any]]: + turns = _retained_history(request, session_id) if interrupted_guard is None: if turns[-1].get("status") != "completed": raise ValueError("native checkpoint requires a completed final cursor") else: _verify_interrupted_guard(turns, interrupted_guard) - ids = [t["id"] for t in turns] - if len(set(ids)) != len(ids): - raise ValueError("native checkpoint history contains duplicate turns") return turns +def native_interrupted_turn(env: dict[str, str], session_id: str, *, + process_started_at: float, process_stopped_at: float, + timeout: float = 30) -> dict[str, Any]: + """Read a stable interrupted tail; this is not checkpoint admission. + + The caller proves original process termination and current session ownership. + No resume, fork, interruption, model call, history edit or command replay occurs. + A completed research-checkpoint cursor remains a separate, stricter contract. + """ + if (not session_id or any(type(v) not in (int, float) or not math.isfinite(v) + for v in (process_started_at, process_stopped_at, timeout)) + or process_started_at > process_stopped_at or timeout <= 0): + raise ValueError("Native interruption requires its original process window") + deadline = time.monotonic() + timeout + with native_checkpoint_reader(env, timeout=timeout) as transport: + def request(method, params): + remaining = deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("Native interruption readback window closed") + transport.response_timeout_sec = remaining + return transport.request(method, params) + _native_thread_readback(request, env, session_id) + turns = _retained_history(request, session_id) + tail = turns[-1] + start, end = tail.get("startedAt"), tail.get("completedAt") + if (tail.get("status") != "interrupted" or tail.get("error") is not None + or any(type(value) not in (int, float) or not math.isfinite(value) + for value in (start, end)) + # Native Turn timestamps have second resolution. + or not math.floor(process_started_at) <= start <= end <= math.ceil(process_stopped_at)): + raise ValueError("Native interruption differs from the original process window") + if any(item.get("status") in {"inProgress", "running", "pending"} + for item in tail.get("items", [])): + raise ValueError("Native interruption still has pending effects") + encoded = json.dumps(turns, sort_keys=True, separators=(",", ":")).encode() + reread = _retained_history(request, session_id) + if json.dumps(reread, sort_keys=True, separators=(",", ":")).encode() != encoded: + raise ValueError("Native interruption advanced during readback") + return {"session_id": session_id, "turn_id": tail["id"], + "visible_history_sha256": hashlib.sha256(encoded).hexdigest(), + "turn_count": len(turns)} + + def _guard_cli_arguments(command: str, guard: dict) -> list[str]: """Parse the provider shell wrapper plus the renderer's optional Turn assignment.""" outer = shlex.split(command) diff --git a/scripts/external_scheduler_worker.py b/scripts/external_scheduler_worker.py index 42ced4e060..f2510222fc 100644 --- a/scripts/external_scheduler_worker.py +++ b/scripts/external_scheduler_worker.py @@ -30,6 +30,7 @@ import signal import sys import time +import uuid from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path @@ -921,12 +922,15 @@ def save(fields: dict[str, Any]) -> None: # state write or the Host invocation can be interrupted. _save_host_pending(state_path, True) state["host_attempt"] = {"scope": scope, "status": "started", + "attempt_id": uuid.uuid4().hex, "attempt": prior.get("attempt", 0) + 1, - "session_id": prior.get("session_id"), + "session_id": prior.get("session_id") or state.get("recovered_session_id"), **({"replan_obligation_id": attempt["obligation_id"]} if attempt is not None else {})} save({}) - environment = {**os.environ, "LOOPX_HOST_RETRY_SESSION_ID": prior.get("session_id") or ""} + environment = {**os.environ, + "LOOPX_HOST_ATTEMPT_ID": state["host_attempt"]["attempt_id"], + "LOOPX_HOST_RETRY_SESSION_ID": state["host_attempt"]["session_id"] or ""} wake_result = _run_wake( args.wake_cmd, timeout_seconds=getattr(args, "wake_timeout_seconds", None), From 301a755acf55ca588d6ebb7e61b8f39cd35503bb Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 03:40:25 +0800 Subject: [PATCH 2/4] fix(benchmark): bind current worker and deliver retained recovery handoff Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/README.md | 24 +++++++ benchmark/edgebench/stopped_recovery.py | 8 ++- benchmark/runtime/interruption_recovery.py | 30 ++++++++ benchmark/runtime/scheduler.py | 13 +++- benchmark/runtime/worker.py | 8 ++- benchmark/tests/test_interruption_recovery.py | 65 ++++++++++++++++- .../tests/test_scheduler_phase_resume.py | 22 ++++++ benchmark/tests/test_stopped_recovery.py | 70 +++++++++++++++++-- scripts/external_scheduler_worker.py | 15 ++++ 9 files changed, 243 insertions(+), 12 deletions(-) diff --git a/benchmark/README.md b/benchmark/README.md index 85d8c8e559..60141c5635 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -1437,3 +1437,27 @@ segment and new segment separately attributable, and verify actual model activit and native feedback after dispatch. Disposable tests exercise real Docker archive and proof installation; SQLite/process tests exercise interrupted work followed by the same session's next Turn without paid model calls. + +The legacy stopped-source path accepts only Docker's default container-ID hostname. +The receiver derives the current UTS hostname independently and refuses copied +proofs in a different worker, including custom-hostname sources. This is an +accidental-cross-worker boundary under the trusted Docker operator, not a +cryptographic defense against that operator spoofing hostnames. The installed +proof initially has root ownership and mode `0600`; a non-root worker's operator +must assign that one inert file to the original execution user before consumption. +Only verified SQLite `-shm` companions are omitted as rebuildable indexes; arbitrary +workspace files with that suffix remain protected. + +An explicitly continued phase can carry a hash-bound content handoff through +`LOOPX_RECOVERY_HANDOFF` and `LOOPX_RECOVERY_HANDOFF_SHA256`. Use the existing +`handoff restore --input FILE --format json` content envelope (`ok: true` and +`handoff_text`), without ownership arguments. The normal heartbeat presents it; +a successful same-session Host receipt confirms its digest and stops repeated +presentation. An uncertain attempt retains the same input, and missing or changed +bytes fail closed. A legitimate later budget phase can bind its own new handoff; +the earlier phase and digest remain immutable. Native submission-ID remapping +belongs in this infrastructure handoff, never a replacement solution Todo. + +Recovery settles process transport only. Preserved pending semantic replan +attempts still require their original settlement owner; an interrupted replan +is not silently declared complete or cleared to force dispatch. diff --git a/benchmark/edgebench/stopped_recovery.py b/benchmark/edgebench/stopped_recovery.py index 916d65a583..0ecc7352d2 100644 --- a/benchmark/edgebench/stopped_recovery.py +++ b/benchmark/edgebench/stopped_recovery.py @@ -28,7 +28,11 @@ def file_inventory(root: Path) -> list[dict]: if path.is_symlink(): raise ValueError('Interrupted source inventory cannot contain symlinks') if path.name.endswith('-shm'): - continue # SQLite rebuilds this reader-only index. + database = path.with_name(path.name[:-4]) + if database.is_file() and not database.is_symlink(): + with database.open('rb') as source: + if source.read(16) == b'SQLite format 3\0': + continue # Only a SQLite participant's reader index is rebuilt. if path.is_file(): h = hashlib.sha256() with path.open('rb') as stream: @@ -77,6 +81,8 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict, # original install/run command. Actual restart remains the caller's action. if box.attrs['Config'].get('Entrypoint') or box.attrs['Config'].get('Cmd') != ['tail', '-f', '/dev/null']: raise ValueError('Original worker must have an inert restart command') + if box.attrs['Config'].get('Hostname') != box.id[:12]: + raise ValueError('Stopped recovery requires the original default Docker hostname') if box.attrs.get('Mounts'): raise ValueError('Stopped worker has externally writable mounts') finished = datetime.fromisoformat(before['FinishedAt'].replace('Z', '+00:00')).timestamp() diff --git a/benchmark/runtime/interruption_recovery.py b/benchmark/runtime/interruption_recovery.py index 8ac96bea8a..2e01364649 100644 --- a/benchmark/runtime/interruption_recovery.py +++ b/benchmark/runtime/interruption_recovery.py @@ -12,6 +12,30 @@ import os from pathlib import Path import re +import stat + + +def recovery_handoff(env: dict[str, str]) -> tuple[dict | None, str]: + """Read content through the existing handoff owner; never adopt ownership.""" + path, digest = env.get('LOOPX_RECOVERY_HANDOFF'), env.get('LOOPX_RECOVERY_HANDOFF_SHA256') + if path is None and digest is None: + return None, '' + if not path or not re.fullmatch(r'[0-9a-f]{64}', digest or ''): + raise ValueError('Recovery handoff requires its exact path and digest') + source = Path(path) + if not source.is_absolute() or source.resolve() != source or source.is_symlink(): + raise ValueError('Recovery handoff must be a physical absolute file') + # Reject special files before reading, including a raced FIFO replacement. + descriptor = os.open(source, os.O_RDONLY | getattr(os, 'O_NONBLOCK', 0) | getattr(os, 'O_NOFOLLOW', 0)) + with os.fdopen(descriptor, 'rb') as stream: + if not stat.S_ISREG(os.fstat(stream.fileno()).st_mode): + raise ValueError('Recovery handoff must be a regular file') + raw = stream.read(65537) + if len(raw) > 65536 or hashlib.sha256(raw).hexdigest() != digest: + raise ValueError('Recovery handoff bytes changed or exceed the transport bound') + from loopx.cli_commands.handoff_restore import restore_handoff_input + text = restore_handoff_input(raw.decode('utf-8'), input_format='json') + return {'path': path, 'sha256': digest}, text class ProcessObservers: @@ -98,6 +122,12 @@ def _stopped_source(state_file: Path, state: dict, env: dict[str, str], wake_log or any(declaration.get(k) != v for k, v in expected.items()) or hashlib.sha256(_physical_read(state_file)).hexdigest() != proof['scheduler_state_sha256']): raise ValueError('Stopped recovery proof differs from the original worker') + # Derive identity from the current UTS namespace, never a caller env value. + # The Docker producer only accepts the original default hostname. This + # prevents accidentally consuming copied evidence in a fresh worker; the + # trusted host/operator remains responsible for Docker namespace authority. + if os.name != 'posix' or os.uname().nodename != proof['container_id'][:12]: + raise ValueError('Stopped recovery is only valid in the original container namespace') required = set(source_paths(declaration).values()) supplied = {part['source_path'] for part in proof['components']} if not required <= supplied or any(p not in required and not any( diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index f189651975..9df2861ddc 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -91,7 +91,7 @@ def _phase_receipt(state_file: Path, wake_log_dir: str | None) -> dict | None: def _create_resume_state(target: Path, predecessor: Path, *, recovered: dict | None = None, - replan_attempts=None) -> None: + replan_attempts=None, handoff: dict | None = None) -> None: if target.is_symlink(): raise ValueError("Resumed scheduler state must not be a symlink") try: @@ -101,6 +101,8 @@ def _create_resume_state(target: Path, predecessor: Path, *, recovered: dict | N value.update(interruption_recovery=recovered, recovered_session_id=recovered["native"]["session_id"], replan_attempts=replan_attempts or []) + if handoff is not None: + value["recovery_handoff"] = handoff json.dump(value, stream) stream.flush() os.fsync(stream.fileno()) @@ -119,6 +121,8 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, if (isinstance(phase_deadline, bool) or not isinstance(phase_deadline, (int, float)) or not math.isfinite(phase_deadline) or phase_deadline <= time.time()): raise ValueError("Budget continuation requires an open finite phase deadline") + from .interruption_recovery import recovery_handoff + handoff, _ = recovery_handoff(env or {}) path = _physical_state(Path(state_file).absolute()) from loopx.file_lock import try_exclusive_file_lock from scripts.external_scheduler_worker import HOST_LOCK_OPERATION, _host_attempt, _host_pending @@ -144,6 +148,11 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, return str(predecessor) prior = state.get("host_attempt") recovery = state.get("interruption_recovery") + can_continue_stopped_phase = (isinstance(prior, dict) and prior.get("status") == "stopped" + and prior.get("stop_exit_code") == BUDGET_EXHAUSTED_EXIT_CODE) + if (recovery is not None and state.get("recovery_handoff") != handoff + and (recovery.get("phase_deadline_epoch") == phase_deadline or not can_continue_stopped_phase)): + raise ValueError("Recovered phase handoff is immutable; preserve its exact input") if (recovery is not None and recovery.get("phase_deadline_epoch") != phase_deadline and not (isinstance(prior, dict) and prior.get("status") == "stopped" and prior.get("stop_exit_code") == BUDGET_EXHAUSTED_EXIT_CODE)): @@ -175,7 +184,7 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, recovered.update(source_state_sha256=hashlib.sha256(predecessor.read_bytes()).hexdigest(), phase_deadline_epoch=phase_deadline) _create_resume_state(target, predecessor, recovered=recovered, - replan_attempts=state.get("replan_attempts")) + replan_attempts=state.get("replan_attempts"), handoff=handoff) return str(target) if (prior.get("status") != "stopped" or prior.get("stop_exit_code") != BUDGET_EXHAUSTED_EXIT_CODE diff --git a/benchmark/runtime/worker.py b/benchmark/runtime/worker.py index f805b71527..f7673de2ef 100644 --- a/benchmark/runtime/worker.py +++ b/benchmark/runtime/worker.py @@ -126,7 +126,9 @@ def heartbeat_body(env: dict[str, str], turn_id: str, *, native_goal: bool, ) from .research_probe import runtime_execution_guidance - return body + runtime_execution_guidance(env) + from .interruption_recovery import recovery_handoff + _, handoff = recovery_handoff(env) + return body + runtime_execution_guidance(env) + ("\nRecovery handoff (content only):\n" + handoff if handoff else "") @contextmanager @@ -567,6 +569,10 @@ def run_once(env: dict[str, str]) -> dict: Path(env.get("LOOPX_RUNTIME_ROOT", str(home))) / "benchmark-pending-turn.json" ) try: + from .interruption_recovery import recovery_handoff + handoff, _ = recovery_handoff(env) + if handoff is not None: + receipt["recovery_handoff_sha256"] = handoff["sha256"] if stage == "execute" and env.get("LOOPX_PHASE_DEADLINE_EPOCH"): remaining = float(env["LOOPX_PHASE_DEADLINE_EPOCH"]) - time.time() # Heartbeat work uses the whole remaining phase. Cleanup is bounded diff --git a/benchmark/tests/test_interruption_recovery.py b/benchmark/tests/test_interruption_recovery.py index 838a5a973e..451f5ea341 100644 --- a/benchmark/tests/test_interruption_recovery.py +++ b/benchmark/tests/test_interruption_recovery.py @@ -114,7 +114,8 @@ def interrupted(*args): 'items':[{'type':'commandExecution','status':'failed'}]})) sys.exit(130) signal.signal(signal.SIGINT,interrupted) -sys.stdin.read() +payload=sys.stdin.read() +with pathlib.Path('native-inputs').open('a') as f:f.write(json.dumps(payload)+'\\n') if pathlib.Path('continue').exists(): print(json.dumps({'type':'turn.completed'}),flush=True) else: @@ -122,7 +123,8 @@ def interrupted(*args): ''') entry = tmp_path / "worker-entry.py" entry.write_text("from benchmark.runtime import worker\n" - "worker.heartbeat_body=lambda *a,**kw:'synthetic authorized work'\n" + "from benchmark.runtime.interruption_recovery import recovery_handoff\n" + "worker.heartbeat_body=lambda env,*a,**kw:'synthetic authorized work'+recovery_handoff(env)[1]\n" "raise SystemExit(worker.main())\n") for key, value in env.items(): monkeypatch.setenv(key, value) @@ -163,6 +165,12 @@ def interrupted(*args): second_receipt = receipts[0].parent.with_name("duplicate") / "receipt.json" second_receipt.parent.mkdir() second_receipt.write_bytes(receipts[0].read_bytes()) + if fault is None: + import hashlib + brief = tmp_path / 'recovery-handoff.json' + brief.write_text(json.dumps({'ok': True, 'handoff_text': 'Retained submission old-7 is now restored-7; read the inherited history. Goal and opportunities are unchanged.'})) + monkeypatch.setenv('LOOPX_RECOVERY_HANDOFF', str(brief)) + monkeypatch.setenv('LOOPX_RECOVERY_HANDOFF_SHA256', hashlib.sha256(brief.read_bytes()).hexdigest()) new = resume_state_file(str(original), phase_deadline=deadline, agent_id=args.agent_id, wake_log_dir=env["LOOPX_WAKE_LOG_DIR"], env=dict(os.environ)) if fault: @@ -184,3 +192,56 @@ def interrupted(*args): argv = [json.loads(line) for line in (project / "native-argv").read_text().splitlines()] assert len(argv) == 2 and "resume" in argv[1] and "original-native" in argv[1] assert original.read_bytes() == old_bytes and goal_state.read_bytes() == initial_goal + assert json.loads(Path(new).read_bytes())["recovery_handoff_delivered"] is True + assert scheduler_fixture.worker.run_worker(args) == 0 + inputs = [json.loads(line) for line in (project / 'native-inputs').read_text().splitlines()] + assert len(inputs) == 3 + assert 'restored-7' in inputs[1] and 'restored-7' not in inputs[2] + + + +def test_recovery_handoff_missing_changed_and_default_parity(tmp_path): + import hashlib + from benchmark.runtime.interruption_recovery import recovery_handoff + assert recovery_handoff({}) == (None, '') + path = tmp_path / 'handoff.json' + raw = json.dumps({'ok': True, 'handoff_text': 'Read retained history; no ownership change.'}).encode() + env = dict(LOOPX_RECOVERY_HANDOFF=str(path), LOOPX_RECOVERY_HANDOFF_SHA256=hashlib.sha256(raw).hexdigest()) + with pytest.raises(FileNotFoundError): recovery_handoff(env) + path.write_bytes(raw) + first = recovery_handoff(env) + assert recovery_handoff(env) == first + path.write_bytes(raw + b' ') + with pytest.raises(ValueError, match='bytes changed'): recovery_handoff(env) + with pytest.raises(ValueError): recovery_handoff({'LOOPX_RECOVERY_HANDOFF': str(path)}) + + +@pytest.mark.parametrize('status', ['started', 'unknown', 'returned']) +def test_new_transport_phase_does_not_clear_unsettled_semantic_replan(tmp_path, monkeypatch, status): + from tests.test_external_scheduler_worker import _guarded_host, worker + from tests.test_external_scheduler_replan_retry import _replan + from benchmark.runtime.scheduler import _create_resume_state + args, count = _guarded_host(tmp_path, kind='provider_capacity', recover_at=2) + args.once = True + monkeypatch.setattr(worker, 'run_quota_should_run', lambda *a, **kw: _replan(tmp_path)) + assert worker.run_worker(args, clock=lambda: 1000.0) == 1 + old = Path(args.state_file) + state = json.loads(old.read_bytes()) + attempts = state['replan_attempts'] + attempts[0]['status'] = status + target = tmp_path / 'continued.json' + _create_resume_state(target, old, recovered={'native': {'session_id': 'same-session'}}, + replan_attempts=attempts) + args.state_file = str(target) + assert worker.run_worker(args, clock=lambda: 1031.0) in (0, 2) + assert count.read_text() == '1' + assert json.loads(target.read_bytes())['replan_attempts'] == attempts + + +@pytest.mark.skipif(not hasattr(os, 'mkfifo'), reason='FIFO boundary requires POSIX') +def test_recovery_handoff_rejects_fifo_without_waiting_for_a_writer(tmp_path): + from benchmark.runtime.interruption_recovery import recovery_handoff + pipe = tmp_path / 'handoff.fifo' + os.mkfifo(pipe) + with pytest.raises(ValueError, match='regular file'): + recovery_handoff(dict(LOOPX_RECOVERY_HANDOFF=str(pipe), LOOPX_RECOVERY_HANDOFF_SHA256='a' * 64)) diff --git a/benchmark/tests/test_scheduler_phase_resume.py b/benchmark/tests/test_scheduler_phase_resume.py index bd633cb8b1..d7f2a218be 100644 --- a/benchmark/tests/test_scheduler_phase_resume.py +++ b/benchmark/tests/test_scheduler_phase_resume.py @@ -189,3 +189,25 @@ def test_only_explicit_continuation_wraps_the_existing_scheduler(): assert continued[1:3] == ["-m", "benchmark.runtime.scheduler"] assert worker_command({**env, "LOOPX_EXECUTION_MODE": "native-goal"}, **options, resume_budget_stop=True) == ["/python", "-m", "benchmark.runtime.worker"] + + +def test_later_authorized_phase_can_bind_a_new_handoff_without_rewriting_prior(tmp_path, monkeypatch): + import hashlib + from benchmark.runtime import interruption_recovery + root = tmp_path / 'scheduler.json' + _stopped(root) + source = json.loads(root.read_bytes()) + old_deadline = time.time() - 10 + source.update(interruption_recovery={'phase_deadline_epoch': old_deadline}, + recovery_handoff={'path': '/old/handoff', 'sha256': 'a' * 64}) + root.write_text(json.dumps(source)) + old = root.read_bytes() + brief = tmp_path / 'new-handoff.json' + brief.write_text(json.dumps({'ok': True, 'handoff_text': 'New phase, same objective and opportunity.'})) + env = dict(LOOPX_RECOVERY_HANDOFF=str(brief), LOOPX_RECOVERY_HANDOFF_SHA256=hashlib.sha256(brief.read_bytes()).hexdigest()) + monkeypatch.setattr(interruption_recovery, 'read_interrupted_host', + lambda *a, **kw: {'native': {'session_id': 'original'}}) + deadline = time.time() + 100 + selected = Path(resume_state_file(str(root),phase_deadline=deadline,agent_id='agent',wake_log_dir=str(tmp_path/'wakes'),env=env)) + assert selected != root and root.read_bytes() == old + assert json.loads(selected.read_bytes())['recovery_handoff']['sha256'] == env['LOOPX_RECOVERY_HANDOFF_SHA256'] diff --git a/benchmark/tests/test_stopped_recovery.py b/benchmark/tests/test_stopped_recovery.py index 7fcd52850d..822f0f7af3 100644 --- a/benchmark/tests/test_stopped_recovery.py +++ b/benchmark/tests/test_stopped_recovery.py @@ -28,7 +28,7 @@ def _source(root): (root / "control").mkdir() for file in ("workspace/answer", "runtime/goal", "native/state_5.sqlite", "native/state_5.sqlite-wal", "native/thread_history_1.sqlite", "native/config.toml"): - (root / file).write_bytes(b"synthetic retained bytes") + (root / file).write_bytes((b"SQLite format 3\0" if file.endswith(".sqlite") else b"") + b"synthetic retained bytes") (root / "native/sessions").mkdir() (root / "native/sessions/session.jsonl").write_text("{}\n") (root / "control/registry.json").write_text("{}") @@ -47,7 +47,7 @@ def _source(root): def _fake_container(monkeypatch): box = SimpleNamespace(id="a" * 64, attrs={"State": dict(Status="exited", Running=False, Pid=0, - FinishedAt="1970-01-01T00:02:00Z"), "Config": dict(Cmd=["tail", "-f", "/dev/null"]), "Mounts": []}) + FinishedAt="1970-01-01T00:02:00Z"), "Config": dict(Cmd=["tail", "-f", "/dev/null"], Hostname="a" * 12), "Mounts": []}) box.reload = lambda: None def archive(container, remote, destination, **kwargs): destination.mkdir() @@ -60,10 +60,11 @@ def archive(container, remote, destination, **kwargs): shutil.copy2(remote + suffix, str(copied) + suffix) return copied monkeypatch.setattr(subject, "_archive_component", archive) + monkeypatch.setattr(os, "uname", lambda: SimpleNamespace(nodename="a" * 12)) return box, SimpleNamespace(containers=SimpleNamespace(get=lambda identity: box)) -@pytest.mark.parametrize("fault", [None, "live", "restart_command", "mount", "ambiguous_wake"]) +@pytest.mark.parametrize("fault", [None, "live", "restart_command", "mount", "custom_hostname", "ambiguous_wake"]) def test_prepare_stopped_source_keeps_original_state_and_requires_unique_return(tmp_path, monkeypatch, fault): declaration = _source(tmp_path) box, client = _fake_container(monkeypatch) @@ -71,6 +72,7 @@ def test_prepare_stopped_source_keeps_original_state_and_requires_unique_return( if fault == "live": box.attrs["State"].update(Status="running", Running=True, Pid=42) elif fault == "restart_command": box.attrs["Config"]["Cmd"] = ["run-original-job"] elif fault == "mount": box.attrs["Mounts"] = [dict(Source="/writable")] + elif fault == "custom_hostname": box.attrs["Config"]["Hostname"] = "custom-worker" elif fault == "ambiguous_wake": second = tmp_path / "wakes/second/receipt.json" second.parent.mkdir() @@ -87,7 +89,7 @@ def test_prepare_stopped_source_keeps_original_state_and_requires_unique_return( assert Path(declaration["scheduler_state"]).read_bytes() == original -@pytest.mark.parametrize("fault", ["workspace", "wal_removed", "new_journal", "session_removed", "state", "wrong_lane"]) +@pytest.mark.parametrize("fault", [None, "workspace", "wal_removed", "new_journal", "session_removed", "state", "wrong_lane", "copied_proof", "wrong_container"]) def test_consumer_rejects_stale_or_incomplete_latest_source(tmp_path, monkeypatch, fault): declaration = _source(tmp_path) box, client = _fake_container(monkeypatch) @@ -103,8 +105,15 @@ def test_consumer_rejects_stale_or_incomplete_latest_source(tmp_path, monkeypatc elif fault == "session_removed": (tmp_path / "native/sessions/session.jsonl").unlink() elif fault == "state": (tmp_path / "control/state.json").write_text("{}") elif fault == "wrong_lane": env["LOOPX_AGENT_ID"] = "other" - with pytest.raises(ValueError): - _stopped_source(Path(declaration["scheduler_state"]), {}, env, declaration["wake_log_dir"]) + elif fault == "copied_proof": monkeypatch.setattr(os, "uname", lambda: SimpleNamespace(nodename="b" * 12)) + elif fault == "wrong_container": + value = json.loads(proof.read_bytes()); value["container_id"] = "b" * 64; proof.write_text(json.dumps(value)) + if fault: + with pytest.raises(ValueError): + _stopped_source(Path(declaration["scheduler_state"]), {}, env, declaration["wake_log_dir"]) + else: + first = _stopped_source(Path(declaration["scheduler_state"]), {}, env, declaration["wake_log_dir"]) + assert _stopped_source(Path(declaration["scheduler_state"]), {}, env, declaration["wake_log_dir"]) == first @pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1", reason="Explicit disposable Docker qualification") @@ -131,5 +140,54 @@ def test_real_stopped_docker_proof_installation_is_inert_and_read_back(tmp_path) assert proof["execution_authority_granted"] is False box.reload() assert box.attrs["State"] == original + # Install only public source outside the inventoried fixture and consume + # the proof inside the same original namespace, including a replay. + code = Path(__file__).resolve().parents[2] + package = io.BytesIO() + with tarfile.open(fileobj=package, mode="w") as archive: + for directory in ("benchmark", "loopx"): + for path in (code / directory).rglob("*"): + if path.is_file() and path.suffix in {".py", ".json"}: + archive.add(path, arcname="source/" + str(path.relative_to(code))) + assert box.put_archive("/", package.getvalue()) + env = dict(LOOPX_INTERRUPTED_SOURCE=proof["installed_proof_path"], + LOOPX_GOAL_ID=declaration["goal_id"], LOOPX_AGENT_ID=declaration["agent_id"], + LOOPX_PROJECT=declaration["workspace"], LOOPX_RUNTIME_ROOT=declaration["runtime_root"], + CODEX_HOME=declaration["native_home"], LOOPX_REGISTRY=declaration["registry"]) + script = ("import sys,json;sys.path.insert(0,'/source');from pathlib import Path;" + "from benchmark.runtime.interruption_recovery import _stopped_source;" + f"a=(Path({declaration['scheduler_state']!r}),{{}},{env!r},{declaration['wake_log_dir']!r});" + "first=_stopped_source(*a);assert _stopped_source(*a)==first;print('same namespace replay verified')") + box.start() + result = box.exec_run(["python", "-c", script]) + assert result.exit_code == 0, result.output.decode() + # A new worker with identical source bytes cannot use this proof. + other = client.containers.create("python:3.12-slim", ["tail", "-f", "/dev/null"], network_disabled=True) + try: + for remote in ("/fixture", "/source"): + stream, _ = box.get_archive(remote) + assert other.put_archive("/", b"".join(stream)) + other.start() + rejected = other.exec_run(["python", "-c", script]) + assert rejected.exit_code != 0 + assert b"original container namespace" in rejected.output + finally: + other.remove(force=True) finally: box.remove(force=True) + + +def test_inventory_preserves_arbitrary_shm_names_and_omits_only_sqlite_index(tmp_path): + answer = tmp_path / 'answer-shm' + answer.write_text('original candidate') + before = subject.file_inventory(tmp_path) + answer.write_text('different candidate') + assert subject.file_inventory(tmp_path) != before + (tmp_path / 'database').write_bytes(b'SQLite format 3\0' + b'synthetic database') + index = tmp_path / 'database-shm' + index.write_bytes(b'rebuildable index') + before = subject.file_inventory(tmp_path) + index.write_bytes(b'rebuilt reader index') + assert subject.file_inventory(tmp_path) == before + (tmp_path / 'database').write_bytes(b'not a database') + assert any(row['path'] == 'database-shm' for row in subject.file_inventory(tmp_path)) diff --git a/scripts/external_scheduler_worker.py b/scripts/external_scheduler_worker.py index f2510222fc..739449a359 100644 --- a/scripts/external_scheduler_worker.py +++ b/scripts/external_scheduler_worker.py @@ -931,6 +931,16 @@ def save(fields: dict[str, Any]) -> None: environment = {**os.environ, "LOOPX_HOST_ATTEMPT_ID": state["host_attempt"]["attempt_id"], "LOOPX_HOST_RETRY_SESSION_ID": state["host_attempt"]["session_id"] or ""} + # The explicit continuation phase may carry one frozen + # content handoff. It grants no work/ownership authority. + # Keep it for uncertain retries; stop presenting it only + # after the same native Host returns a successful receipt. + for key in ("LOOPX_RECOVERY_HANDOFF", "LOOPX_RECOVERY_HANDOFF_SHA256"): + environment.pop(key, None) + handoff = state.get("recovery_handoff") + if handoff is not None and not state.get("recovery_handoff_delivered"): + environment.update(LOOPX_RECOVERY_HANDOFF=handoff["path"], + LOOPX_RECOVERY_HANDOFF_SHA256=handoff["sha256"]) wake_result = _run_wake( args.wake_cmd, timeout_seconds=getattr(args, "wake_timeout_seconds", None), @@ -966,6 +976,11 @@ def save(fields: dict[str, Any]) -> None: failure = _host_result(wake_result, state["host_attempt"], clock()) if failure is None: state.pop("host_attempt", None) + if handoff is not None and not state.get("recovery_handoff_delivered"): + receipt = json.loads(wake_result.stdout) + if receipt.get("recovery_handoff_sha256") != handoff["sha256"]: + raise ValueError("Native Host did not confirm the recovery handoff input") + state["recovery_handoff_delivered"] = True else: state["host_attempt"] = failure wake_failed = True From 8633d29ef8a58d9089ef522ca6c3a110d11d36ba Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 03:43:57 +0800 Subject: [PATCH 3/4] fix(benchmark): retain recovery handoffs across pre-dispatch continuation Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/scheduler.py | 9 ++++++--- benchmark/tests/test_scheduler_phase_resume.py | 18 ++++++++++++++++++ 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index 9df2861ddc..235d4152a8 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -150,8 +150,11 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, recovery = state.get("interruption_recovery") can_continue_stopped_phase = (isinstance(prior, dict) and prior.get("status") == "stopped" and prior.get("stop_exit_code") == BUDGET_EXHAUSTED_EXIT_CODE) - if (recovery is not None and state.get("recovery_handoff") != handoff - and (recovery.get("phase_deadline_epoch") == phase_deadline or not can_continue_stopped_phase)): + requested_suffix = hashlib.sha256(float(phase_deadline).hex().encode()).hexdigest()[:20] + same_phase = predecessor.name == f"{path.stem}.phase-{requested_suffix}{path.suffix}" + if ((recovery is not None or state.get("resume_from")) + and state.get("recovery_handoff") != handoff + and (same_phase or not can_continue_stopped_phase)): raise ValueError("Recovered phase handoff is immutable; preserve its exact input") if (recovery is not None and recovery.get("phase_deadline_epoch") != phase_deadline and not (isinstance(prior, dict) and prior.get("status") == "stopped" @@ -209,7 +212,7 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, or result.get("host_invoked") is not False or result.get("mode") != "heartbeat" or result.get("stage") != "execute"): return str(predecessor) - _create_resume_state(target, predecessor) + _create_resume_state(target, predecessor, handoff=handoff) return str(target) diff --git a/benchmark/tests/test_scheduler_phase_resume.py b/benchmark/tests/test_scheduler_phase_resume.py index d7f2a218be..52d1410241 100644 --- a/benchmark/tests/test_scheduler_phase_resume.py +++ b/benchmark/tests/test_scheduler_phase_resume.py @@ -211,3 +211,21 @@ def test_later_authorized_phase_can_bind_a_new_handoff_without_rewriting_prior(t selected = Path(resume_state_file(str(root),phase_deadline=deadline,agent_id='agent',wake_log_dir=str(tmp_path/'wakes'),env=env)) assert selected != root and root.read_bytes() == old assert json.loads(selected.read_bytes())['recovery_handoff']['sha256'] == env['LOOPX_RECOVERY_HANDOFF_SHA256'] + + +def test_pre_dispatch_continuation_binds_and_replays_its_optional_handoff(tmp_path): + import hashlib + root = tmp_path / 'scheduler.json' + _stopped(root) + brief = tmp_path / 'handoff.json' + brief.write_text(json.dumps({'ok': True, 'handoff_text': 'Keep the original goal.'})) + env = dict(LOOPX_RECOVERY_HANDOFF=str(brief), LOOPX_RECOVERY_HANDOFF_SHA256=hashlib.sha256(brief.read_bytes()).hexdigest()) + deadline = time.time() + 100 + def select(): + return Path(resume_state_file(str(root),phase_deadline=deadline,agent_id='agent',wake_log_dir=str(tmp_path/'wakes'),env=env)) + phase = select() + assert json.loads(phase.read_bytes())['recovery_handoff']['sha256'] == env['LOOPX_RECOVERY_HANDOFF_SHA256'] + assert select() == phase + brief.write_text(json.dumps({'ok': True, 'handoff_text': 'Different material.'})) + env['LOOPX_RECOVERY_HANDOFF_SHA256'] = hashlib.sha256(brief.read_bytes()).hexdigest() + with pytest.raises(ValueError, match='immutable'): select() From b54ca62c892599ba623940fd1f83e0a59fccb849 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 16:06:43 +0800 Subject: [PATCH 4/4] fix(benchmark): share the stopped recovery archive deadline Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/edgebench/stopped_recovery.py | 17 +++-- benchmark/tests/test_stopped_recovery.py | 81 +++++++++++++++++++----- 2 files changed, 78 insertions(+), 20 deletions(-) diff --git a/benchmark/edgebench/stopped_recovery.py b/benchmark/edgebench/stopped_recovery.py index 0ecc7352d2..18c3d8593b 100644 --- a/benchmark/edgebench/stopped_recovery.py +++ b/benchmark/edgebench/stopped_recovery.py @@ -11,6 +11,7 @@ import hashlib import io import json +import math import os from pathlib import Path import re @@ -18,6 +19,7 @@ import tempfile import time +from .checkpoint_controller import _remaining from .checkpoint_worker import _archive_component, _guest_path @@ -65,6 +67,9 @@ def source_paths(declaration: dict) -> dict[str, str]: def prepare_stopped_recovery(client, *, container_id: str, declaration: dict, output: Path, timeout: float = 120, install: bool = False) -> dict: + if not math.isfinite(timeout) or timeout <= 0: + raise ValueError('Stopped recovery timeout must be finite and positive') + deadline = time.monotonic() + timeout if not re.fullmatch(r'[a-f0-9]{64}', container_id): raise ValueError('Stopped recovery requires one full container identity') paths = source_paths(declaration) @@ -86,7 +91,6 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict, if box.attrs.get('Mounts'): raise ValueError('Stopped worker has externally writable mounts') finished = datetime.fromisoformat(before['FinishedAt'].replace('Z', '+00:00')).timestamp() - deadline = time.time() + timeout components = [] with tempfile.TemporaryDirectory(prefix='.stopped-source-', dir=output.parent) as temporary: copied = {} @@ -151,21 +155,24 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict, raise ValueError('Original worker changed before proof installation') with tempfile.TemporaryDirectory(prefix='.proof-existing-', dir=output.parent) as directory: try: - existing = _archive_component(box, str(parent / name), Path(directory) / 'proof', deadline=time.time() + timeout) + existing = _archive_component(box, str(parent / name), Path(directory) / 'proof', deadline=deadline) except NotFound: existing = None if existing is not None and existing.read_bytes() != raw: raise ValueError('Original worker recovery proof path is already occupied') - if existing is None and not box.put_archive(str(parent), package.getvalue()): - raise ValueError('Original worker did not accept its inert recovery proof') + if existing is None: + _remaining(deadline) + if not box.put_archive(str(parent), package.getvalue()): + raise ValueError('Original worker did not accept its inert recovery proof') with tempfile.TemporaryDirectory(prefix='.proof-readback-', dir=output.parent) as directory: - copied = _archive_component(box, str(parent / name), Path(directory) / 'proof', deadline=time.time() + timeout) + copied = _archive_component(box, str(parent / name), Path(directory) / 'proof', deadline=deadline) if copied.read_bytes() != raw: raise ValueError('Original worker recovery proof readback differs') box.reload() if box.attrs['State'] != before: raise ValueError('Original worker changed during proof installation') result['installed_proof_path'] = str(parent / name) + _remaining(deadline) with os.fdopen(os.open(output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), 'w') as stream: json.dump(result, stream, indent=2) stream.flush() diff --git a/benchmark/tests/test_stopped_recovery.py b/benchmark/tests/test_stopped_recovery.py index 822f0f7af3..fc4049b0fc 100644 --- a/benchmark/tests/test_stopped_recovery.py +++ b/benchmark/tests/test_stopped_recovery.py @@ -1,7 +1,6 @@ """A stopped source proof observes existing bytes, and grants no execution.""" from __future__ import annotations -from datetime import datetime, timezone import io import json import os @@ -69,10 +68,14 @@ def test_prepare_stopped_source_keeps_original_state_and_requires_unique_return( declaration = _source(tmp_path) box, client = _fake_container(monkeypatch) original = Path(declaration["scheduler_state"]).read_bytes() - if fault == "live": box.attrs["State"].update(Status="running", Running=True, Pid=42) - elif fault == "restart_command": box.attrs["Config"]["Cmd"] = ["run-original-job"] - elif fault == "mount": box.attrs["Mounts"] = [dict(Source="/writable")] - elif fault == "custom_hostname": box.attrs["Config"]["Hostname"] = "custom-worker" + if fault == "live": + box.attrs["State"].update(Status="running", Running=True, Pid=42) + elif fault == "restart_command": + box.attrs["Config"]["Cmd"] = ["run-original-job"] + elif fault == "mount": + box.attrs["Mounts"] = [dict(Source="/writable")] + elif fault == "custom_hostname": + box.attrs["Config"]["Hostname"] = "custom-worker" elif fault == "ambiguous_wake": second = tmp_path / "wakes/second/receipt.json" second.parent.mkdir() @@ -99,15 +102,24 @@ def test_consumer_rejects_stale_or_incomplete_latest_source(tmp_path, monkeypatc LOOPX_AGENT_ID=declaration["agent_id"], LOOPX_PROJECT=declaration["workspace"], LOOPX_RUNTIME_ROOT=declaration["runtime_root"], CODEX_HOME=declaration["native_home"], LOOPX_REGISTRY=declaration["registry"]) - if fault == "workspace": (tmp_path / "workspace/answer").write_text("different") - elif fault == "wal_removed": (tmp_path / "native/state_5.sqlite-wal").unlink() - elif fault == "new_journal": (tmp_path / "native/state_5.sqlite-journal").write_text("different") - elif fault == "session_removed": (tmp_path / "native/sessions/session.jsonl").unlink() - elif fault == "state": (tmp_path / "control/state.json").write_text("{}") - elif fault == "wrong_lane": env["LOOPX_AGENT_ID"] = "other" - elif fault == "copied_proof": monkeypatch.setattr(os, "uname", lambda: SimpleNamespace(nodename="b" * 12)) + if fault == "workspace": + (tmp_path / "workspace/answer").write_text("different") + elif fault == "wal_removed": + (tmp_path / "native/state_5.sqlite-wal").unlink() + elif fault == "new_journal": + (tmp_path / "native/state_5.sqlite-journal").write_text("different") + elif fault == "session_removed": + (tmp_path / "native/sessions/session.jsonl").unlink() + elif fault == "state": + (tmp_path / "control/state.json").write_text("{}") + elif fault == "wrong_lane": + env["LOOPX_AGENT_ID"] = "other" + elif fault == "copied_proof": + monkeypatch.setattr(os, "uname", lambda: SimpleNamespace(nodename="b" * 12)) elif fault == "wrong_container": - value = json.loads(proof.read_bytes()); value["container_id"] = "b" * 64; proof.write_text(json.dumps(value)) + value = json.loads(proof.read_bytes()) + value["container_id"] = "b" * 64 + proof.write_text(json.dumps(value)) if fault: with pytest.raises(ValueError): _stopped_source(Path(declaration["scheduler_state"]), {}, env, declaration["wake_log_dir"]) @@ -116,8 +128,17 @@ def test_consumer_rejects_stale_or_incomplete_latest_source(tmp_path, monkeypatc assert _stopped_source(Path(declaration["scheduler_state"]), {}, env, declaration["wake_log_dir"]) == first +@pytest.mark.parametrize("timeout", [0, -1, float("inf"), -float("inf"), float("nan")]) +def test_stopped_recovery_rejects_invalid_archive_timeout(tmp_path, timeout): + output = tmp_path / "proof.json" + with pytest.raises(ValueError, match="timeout must be finite and positive"): + subject.prepare_stopped_recovery(None, container_id="a" * 64, + declaration=_source(tmp_path), output=output, timeout=timeout) + assert not output.exists() + + @pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1", reason="Explicit disposable Docker qualification") -def test_real_stopped_docker_proof_installation_is_inert_and_read_back(tmp_path): +def test_real_stopped_docker_proof_installation_is_inert_and_read_back(tmp_path, monkeypatch): docker = pytest.importorskip("docker") client = docker.from_env(timeout=30) box = client.containers.create("python:3.12-slim", ["tail", "-f", "/dev/null"], network_disabled=True) @@ -134,9 +155,39 @@ def test_real_stopped_docker_proof_installation_is_inert_and_read_back(tmp_path) box.stop(timeout=2) box.reload() original = dict(box.attrs["State"]) + archive_requests = [] + get = client.api._get + + def observe_archive_timeout(url, **options): + if url.endswith("/archive"): + archive_requests.append((options["params"]["path"], options["timeout"])) + return get(url, **options) + + monkeypatch.setattr(client.api, "_get", observe_archive_timeout) + archive_component = subject._archive_component + expired_output = tmp_path / "expired-proof.json" + with monkeypatch.context() as slow_reader: + def enter_after_deadline(*args, **kwargs): + time.sleep(0.02) + return archive_component(*args, **kwargs) + + slow_reader.setattr(subject, "_archive_component", enter_after_deadline) + with pytest.raises(TimeoutError, match="capture window closed"): + subject.prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, + output=expired_output, timeout=0.001, install=True) + assert not expired_output.exists() and not archive_requests + box.reload() + assert box.attrs["State"] == original + + allowance = 30 proof = subject.prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, - output=tmp_path / "proof.json", install=True) + output=tmp_path / "proof.json", timeout=allowance, install=True) assert proof["installed_proof_path"].startswith("/fixture/control/interruption-source-") + assert len([path for path, _ in archive_requests if path == proof["installed_proof_path"]]) == 2 + remaining = [seconds for _, seconds in archive_requests] + assert all(0 < seconds <= allowance for seconds in remaining) + assert remaining == sorted(remaining, reverse=True) + monkeypatch.setattr(client.api, "_get", get) assert proof["execution_authority_granted"] is False box.reload() assert box.attrs["State"] == original