diff --git a/benchmark/edgebench/README.md b/benchmark/edgebench/README.md index cf5997a900..35c034c6b8 100644 --- a/benchmark/edgebench/README.md +++ b/benchmark/edgebench/README.md @@ -290,15 +290,14 @@ An Official worker without a checkpoint must retain its existing session and use its authorized runtime owner's idle credential handoff; this preparation helper does not provide a live credential-replacement transaction. -An unchanged stopped heartbeat worker with an empty-output legacy Host failure -can use the explicit [idle-host handoff](../runtime/RUNTIME.md) after independent +An unchanged stopped heartbeat worker with a completed Host or an empty-output +legacy failure can use the explicit [idle-host handoff](../runtime/RUNTIME.md) after independent native-history, source, current Goal and account qualification. Seal with `python -m benchmark.edgebench.stopped_recovery --idle-host-handoff` using the original source declaration; then select the original scheduler digest with `benchmark.runtime.scheduler --idle-host-state-sha256`. Both options are off by -default. The original failure remains recorded. This route preserves the -existing completed session and source rather than manufacturing an interrupted -Turn. Native judge restoration, opportunity/cooldown conservation, the audited +default. The original return, session, and source remain recorded. +Native judge restoration, opportunity/cooldown conservation, the audited remaining allowance and a finite owner cutoff remain separate requirements. It does not apply to Official or native-Goal workers. diff --git a/benchmark/edgebench/stopped_recovery.py b/benchmark/edgebench/stopped_recovery.py index 16f59f3cfb..54556497b1 100644 --- a/benchmark/edgebench/stopped_recovery.py +++ b/benchmark/edgebench/stopped_recovery.py @@ -102,6 +102,47 @@ def validate_idle_host_return(state: dict, receipt: dict, *, stdout: bytes, stde raise ValueError('Idle handoff requires its original empty-output legacy Host return') + +def completed_host_window(state: dict, receipt: dict, *, lineage: dict, phase: float) -> tuple[float, float]: + """Verify a completed process at the retained clock without judging its work.""" + from benchmark.runtime.interruption_recovery import _number + + process = receipt.get("native_process") or {} + before = process.get("before_cleanup") or {} + after = process.get("after_cleanup") or {} + session = receipt.get("session") or {} + started = _number(process.get("started_at_unix")) + returned = _number(before.get("observed_at_unix")) + stopped = _number(after.get("observed_at_unix")) + try: + wake_phase = float(receipt.get("phase_deadline_epoch")) + except (TypeError, ValueError) as error: + raise ValueError("Completed Host requires its original phase clock") from error + if ("host_attempt" in state or state.get("last_wake_status") != "wake_ok" + or state.get("last_wake_failure_kind") is not None + or any(type(row.get(key)) is not int or row[key] != 0 + for row, key in ((state, "last_wake_rc"), (receipt, "return_code"), + (before, "return_code"), (after, "return_code"))) + or receipt.get("mode") != "heartbeat" or receipt.get("stage") != "execute" + or receipt.get("ok") is not True or receipt.get("host_invoked") is not True + or receipt.get("timed_out") is not False or receipt.get("budget_exhausted") + or "host_failure" in receipt or receipt.get("error_kind") is not None + or any(receipt.get(key) != lineage[key] for key in ("goal_id", "agent_id")) + or not re.fullmatch(r"[a-f0-9]{32}", receipt.get("host_attempt_id") or "") + or session.get("binding_scope") != "agent" + or not isinstance(session.get("session_id"), str) or not session["session_id"].strip() + or process.get("schema_version") != "native_codex_process_observation_v1" + or type(process.get("pid")) is not int or process["pid"] <= 1 + or process.get("cleanup_trigger") != "scope_completed" + or any(process.get(key) is not None for key in + ("first_failure", "scope_error_type", "cleanup_error_type", "persistence_error_type")) + or before.get("signal") is not None or after.get("signal") is not None + or not 0 < started <= returned <= stopped <= phase + or not math.isfinite(phase) or wake_phase != phase): + raise ValueError("Stopped source lacks its original completed Host return") + return started, stopped + + def prepare_stopped_recovery(client, *, container_id: str, declaration: dict, output: Path, timeout: float = 120, install: bool = False, idle_host_handoff: bool = False) -> dict: @@ -149,8 +190,12 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict, prior = state.get('host_attempt', {}) failed = prior.get('status') == 'failed' returned_budget_stop = prior.get('status') == 'stopped' - if (prior.get('status') not in {'started', 'failed', 'stopped'} - or json.loads(copied['host_pending'].read_bytes()) != {'pending': True}): + completed = idle_host_handoff and "host_attempt" not in state + pending = json.loads(copied['host_pending'].read_bytes()) + if completed: + if pending != {'pending': False}: + raise ValueError('Completed Host handoff requires its cleared pending witness') + elif prior.get('status') not in {'started', 'failed', 'stopped'} or pending != {'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() @@ -177,7 +222,7 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict, or any(key in receipt and receipt[key] != declaration[key] for key in ('goal_id', 'agent_id'))): raise ValueError('Stopped source lacks its exact returned budget stop') - if idle_host_handoff: + if idle_host_handoff and not completed: from benchmark.runtime.worker import host_output_paths stdout, stderr = host_output_paths(wake.parent) @@ -198,8 +243,8 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict, or type(receipt.get('return_code')) is not int or receipt['return_code'] != 1 or receipt.get('budget_exhausted')): raise ValueError('Stopped source lacks its exact returned quota failure') - if failed or returned_budget_stop: - # Returned failures and budget stops are committed after their + if failed or returned_budget_stop or completed: + # Returned Hosts are committed after their # receipt. A still-started legacy attempt has the opposite order. # Neither path may borrow an earlier wake or skip a later one. all_receipts = list(copied['wake_log_dir'].glob('*/receipt.json')) @@ -209,27 +254,33 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict, candidates = [wake] 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 (not failed and 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 (not failed and '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 + wake_started = int(declaration['wake_turn_id'].split('-')[1]) / 1_000_000_000 phase = float(copied['phase_clock'].read_text()) - # A returned stop proves the process was reaped before its receipt. - # Legacy workers could reserve cleanup time before the hard deadline; - # do not enlarge that native Turn's window to a later container stop. - process_stopped = math.floor(wake.stat().st_mtime) if returned_budget_stop or idle_host_handoff else finished - if (not math.isfinite(phase) or not start <= min(phase, process_stopped) - or process_stopped > finished - or (not failed and not returned_budget_stop and phase > finished)): - raise ValueError('Stopped source phase did not expire during the original wake') + if completed: + start, process_stopped = completed_host_window(state, receipt, lineage=declaration, phase=phase) + if (receipt.get('turn_id') != declaration['wake_turn_id'] or start < wake_started + or process_stopped > min(finished, math.floor(wake.stat().st_mtime) + 1)): + raise ValueError('Completed Host process lies outside the retained wake') + else: + if (receipt.get('turn_id') != declaration['wake_turn_id'] or receipt.get('host_invoked') is not True + or (not failed and 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 (not failed and 'host_failure' in receipt)): + raise ValueError('Stopped source lacks an exact interrupted budget receipt') + start = wake_started + # Legacy workers could reserve cleanup time before the hard deadline. + process_stopped = math.floor(wake.stat().st_mtime) if returned_budget_stop or idle_host_handoff else finished + if (not math.isfinite(phase) or not start <= min(phase, process_stopped) + or process_stopped > finished + or (not failed and not returned_budget_stop and 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': process_stopped, 'source_phase_deadline_epoch': phase, 'session': receipt.get('session'), **({'idle_host_handoff': True} if idle_host_handoff else {}), + **({'completed_host': True} if completed else {}), 'execution_authority_granted': False} box.reload() if box.attrs['State'] != before: @@ -287,7 +338,7 @@ def main(): 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') parser.add_argument('--idle-host-handoff', action='store_true', - help='Explicitly seal an empty-output legacy Host failure for separate idle-history qualification') + help='Explicitly seal a returned Host for separate idle-history qualification') args = parser.parse_args() import docker result = prepare_stopped_recovery(docker.from_env(timeout=30), diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 5657ba18f6..e58a08fc67 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -440,17 +440,18 @@ Budget exhaustion is distinct from task completion: this clock does not override provider failures, Stop, blocked planning, or a valid terminal LoopX decision. Campaign duration coverage must still classify an early exit as missing time. -An operator may explicitly hand off an unchanged stopped legacy heartbeat Host -whose failed attempt produced no stdout/stderr and returned a phase timeout, -while the native session remained idle. This is disabled by default. First run -`benchmark.edgebench.stopped_recovery` with `--idle-host-handoff` and, when -needed, `--install` to place its inert proof in that same stopped worker. The -producer requires the exact failed scheduler, pending witness, latest unique -wake, original workspace, registry, runtime and native stores, including committed -SQLite WALs. It preserves the unknown failure and does not infer a quota failure -from diagnostic text. The proof grants no execution or evaluator authority. -The empty-output witness uses the worker's `stdout.jsonl` and `stderr.log` files; -the producer and receiver share that writer-owned layout. +An operator may explicitly hand off an unchanged stopped heartbeat worker when +its native session is idle. First run `benchmark.edgebench.stopped_recovery` +with `--idle-host-handoff` and, when needed, `--install` to place its inert +proof in that same stopped worker. The proof includes the latest unique wake, +the scheduler and its independent pending witness, the original workspace, +registry, runtime and native stores, including committed SQLite WALs. +This path is disabled by default and grants no execution or evaluator authority. + +A completed Host requires a successful process receipt at the recorded phase +deadline and a cleared pending witness. The producer also supports a legacy +empty-output phase timeout with a still-pending witness. It checks the worker's +`stdout.jsonl` and `stderr.log` and preserves the recorded unknown failure. After the owner restores compatible native evaluator authority separately, restart only the original inert container and retain its original environment. @@ -460,8 +461,9 @@ authorized finite `LOOPX_PHASE_DEADLINE_EPOCH`. Invoke `--idle-host-state-sha256 ORIGINAL_SCHEDULER_SHA256`. The consumer verifies the same namespace and complete unchanged source, current Goal/session/profile and fresh native account allowance. The native history owner must return a stable -complete history whose last completed cursor precedes the failed attempt, with -no pending effects. Its observer uses a disposable copy of the native store and +complete history with no pending effects. For a completed Host, the last +native Turn must fall inside the original process window. For a legacy timeout, +it must precede the attempted launch. The observer uses a disposable copy of the native store and committed WALs because native read APIs can update their own history index; it copies no authentication file. Every copied native base participant must match the producer's sealed set; an extra archived-session/index store or database @@ -474,9 +476,9 @@ existing lock owner's process/token witnesses; registry contents, SQLite/WALs, work files and unrelated lock files remain protected. An accepted handoff creates a new finite transport phase linked to the original -failed record. It retains the session, Goal, Todos, work and failure evidence; -it does not settle an old Turn, refill native opportunities/cooldown, or create -another task budget. Repeating the exact phase retains its clock and any later +scheduler record. It retains the session, Goal, Todos, work, and any pending +semantic settlement. The existing owners still enforce settlement, native +opportunities, cooldown and the total task budget. Repeating the exact phase retains its clock and any later failure; a changed deadline cannot replay the original handoff. The owner still enforces its audited remaining solver allowance and external cutoff, including setup and cleanup. Omit both new opt-ins to keep the ordinary refusal path. diff --git a/benchmark/runtime/interruption_recovery.py b/benchmark/runtime/interruption_recovery.py index 1c99f9917b..5f6c5c6f70 100644 --- a/benchmark/runtime/interruption_recovery.py +++ b/benchmark/runtime/interruption_recovery.py @@ -273,13 +273,10 @@ def read_interrupted_host(state: dict, *, state_file: Path, env: dict[str, str], def read_idle_host(state: dict, *, state_file: Path, env: dict[str, str], wake_log_dir: str) -> dict: - """Explicitly hand off one stopped legacy Host to its unchanged idle session. - - An empty output and unknown failure do not establish idleness. The native - owner must independently return its complete stable history before the - original attempt. No failure classification or original receipt is changed. - """ - from benchmark.edgebench.stopped_recovery import native_source_paths, validate_idle_host_return + """Qualify the stopped Host's unchanged idle session and current authority.""" + from benchmark.edgebench.stopped_recovery import ( + completed_host_window, native_source_paths, validate_idle_host_return, + ) from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession from loopx.hosts.codex_checkpoint import native_idle_turn from .codex import process_environment @@ -288,20 +285,28 @@ def read_idle_host(state: dict, *, state_file: Path, env: dict[str, str], wake_l from .worker import host_output_paths env = process_environment(Path(env['LOOPX_CODEX_HOME']), base=env) - if _host_pending(state_file) is not True: - raise ValueError('Idle handoff requires its original pending Host witness') path, raw, receipt, source = _stopped_source(state_file, state, env, wake_log_dir) if source.get('idle_host_handoff') is not True: raise ValueError('Idle handoff requires its explicitly sealed stopped source') - stdout, stderr = host_output_paths(path.parent) - validate_idle_host_return(state, receipt, stdout=_physical_read(stdout), - stderr=_physical_read(stderr), - lineage={'goal_id': env['LOOPX_GOAL_ID'], 'agent_id': env['LOOPX_AGENT_ID']}) + completed = source.get('completed_host') is True + expected_pending = not completed + if _host_pending(state_file) is not expected_pending: + raise ValueError('Idle handoff differs from its original pending Host witness') + lineage = {'goal_id': env['LOOPX_GOAL_ID'], 'agent_id': env['LOOPX_AGENT_ID']} started = _number(source['process_started_at']) stopped = _number(source['process_stopped_at']) - if (receipt.get('turn_id') != path.parent.name or stopped != math.floor(path.stat().st_mtime) - or started > stopped): - raise ValueError('Idle handoff differs from its original returned Host window') + if completed: + clock = _number(source['source_phase_deadline_epoch']) + if completed_host_window(state, receipt, lineage=lineage, phase=clock) != (started, stopped): + raise ValueError('Idle handoff differs from its original completed Host window') + else: + stdout, stderr = host_output_paths(path.parent) + validate_idle_host_return(state, receipt, stdout=_physical_read(stdout), + stderr=_physical_read(stderr), lineage=lineage) + if stopped != math.floor(path.stat().st_mtime) or started > stopped: + raise ValueError('Idle handoff differs from its original returned Host window') + if receipt.get('turn_id') != path.parent.name: + raise ValueError('Idle handoff wake differs from its original receipt') session = receipt['session']['session_id'] current = CodexAgentSession(env, context='resume', sandbox=env.get('LOOPX_CODEX_SANDBOX', 'danger-full-access'), receipt={}) @@ -311,7 +316,8 @@ def read_idle_host(state: dict, *, state_file: Path, env: dict[str, str], wake_l raise ValueError('Idle handoff native session binding changed') allowance = require_native_receiver_allowance(env) native = native_idle_turn(env, session, process_started_at=started, - sealed_store_paths=tuple(native_source_paths(source['declaration']).values())) + sealed_store_paths=tuple(native_source_paths(source['declaration']).values()), + **({'process_stopped_at': stopped} if completed else {})) with current.admission.launch_lifetime(operation='benchmark_idle_host_handoff'): from loopx.file_lock import _effect_mutation_lock_path, lock_holder_paths from loopx.control_plane.goals.source_session_registry_state import guard_path @@ -334,12 +340,14 @@ def read_idle_host(state: dict, *, state_file: Path, env: dict[str, str], wake_l 'wake_turn_id': receipt['turn_id'], 'process_started_at': started, 'source_phase_deadline_epoch': source['source_phase_deadline_epoch'], 'session_profile_digest': current.digest, 'allowance': allowance, - 'original_host_failure': state['host_attempt']['host_failure'], 'native': native} + 'native': native, **({'process_stopped_at': stopped} if completed + else {'original_host_failure': state['host_attempt']['host_failure']})} def verify_idle_host_replay(state_file: Path, recovery: dict, *, env: dict[str, str]) -> None: """A repeated phase selection retains its original qualification and clock.""" - from benchmark.edgebench.stopped_recovery import validate_idle_host_return + from benchmark.edgebench.stopped_recovery import completed_host_window, validate_idle_host_return + from scripts.external_scheduler_worker import _host_pending_path from .worker import host_output_paths native, allowance = recovery.get('native'), recovery.get('allowance') @@ -352,14 +360,25 @@ def verify_idle_host_replay(state_file: Path, recovery: dict, *, env: dict[str, source = json.loads(_physical_read(state_file)) raw = _physical_read(Path(recovery['receipt_path'])) receipt = json.loads(raw) - stdout, stderr = host_output_paths(Path(recovery['receipt_path']).parent) - validate_idle_host_return(source, receipt, stdout=_physical_read(stdout), - stderr=_physical_read(stderr), - lineage={'goal_id': env['LOOPX_GOAL_ID'], 'agent_id': env['LOOPX_AGENT_ID']}) - if (hashlib.sha256(raw).hexdigest() != recovery['receipt_sha256'] - or source['host_attempt']['host_failure'] != recovery['original_host_failure'] + lineage = {'goal_id': env['LOOPX_GOAL_ID'], 'agent_id': env['LOOPX_AGENT_ID']} + started = _number(recovery.get('process_started_at')) + completed_at = _number(native.get('completed_at')) + if 'process_stopped_at' in recovery: + stopped = _number(recovery['process_stopped_at']) + clock = _number(recovery.get('source_phase_deadline_epoch')) + window_ok = ( + json.loads(_physical_read(_host_pending_path(state_file))) == {'pending': False} + and completed_host_window(source, receipt, lineage=lineage, phase=clock) == (started, stopped) + and math.floor(started) <= _number(native.get('started_at')) <= completed_at <= stopped + ) + else: + stdout, stderr = host_output_paths(Path(recovery['receipt_path']).parent) + validate_idle_host_return(source, receipt, stdout=_physical_read(stdout), + stderr=_physical_read(stderr), lineage=lineage) + window_ok = (source['host_attempt']['host_failure'] == recovery['original_host_failure'] + and completed_at < math.floor(started)) + if (not window_ok or hashlib.sha256(raw).hexdigest() != recovery['receipt_sha256'] or native.get('status') != 'completed' or native.get('session_id') != receipt['session']['session_id'] - or not _number(native.get('completed_at')) < math.floor(_number(recovery['process_started_at'])) or type(native.get('turn_count')) is not int or native['turn_count'] < 1 or not re.fullmatch(r'[a-f0-9]{64}', native.get('visible_history_sha256', ''))): raise ValueError('Idle handoff replay differs from the original qualified source') diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index 0723d449e3..d38fbf1fce 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -384,7 +384,7 @@ def main() -> int: parser.add_argument('--retry-failed-state-sha256', help='Explicitly continue only this exact returned failure after operator repair') parser.add_argument('--idle-host-state-sha256', - help='Explicitly hand off this sealed legacy empty-output Host to its verified idle native session') + help='Explicitly hand off this sealed returned Host to its verified idle native session') args = parser.parse_args() state = resume_state_file(args.state_file, phase_deadline=float(os.environ["LOOPX_PHASE_DEADLINE_EPOCH"]), diff --git a/benchmark/tests/test_idle_host_handoff.py b/benchmark/tests/test_idle_host_handoff.py index 80bbb9bd8f..66ac2fdd09 100644 --- a/benchmark/tests/test_idle_host_handoff.py +++ b/benchmark/tests/test_idle_host_handoff.py @@ -58,6 +58,77 @@ def _idle_return(declaration): return state, receipt + +def _completed_return(declaration): + state, receipt = _idle_return(declaration) + attempt = { + "goal_id": declaration["goal_id"], "agent_id": declaration["agent_id"], + "registry": declaration["registry"], "runtime_root": declaration["runtime_root"], + "obligation_id": "replan-0123456789abcdef", "status": "returned", "wake_rc": 0, + } + state.write_text(json.dumps({"last_wake_status": "wake_ok", "last_wake_rc": 0, + "replan_attempts": [attempt]})) + os.utime(state, (111, 111)) + state.with_name(state.name + ".host-attempt").write_text('{"pending": false}') + value = json.loads(receipt.read_text()) + for key in ("budget_exhausted", "error_kind", "error"): + value.pop(key, None) + value.update( + ok=True, return_code=0, timed_out=False, + goal_id=declaration["goal_id"], agent_id=declaration["agent_id"], + host_attempt_id="f" * 32, phase_deadline_epoch="200", + native_process={ + "schema_version": "native_codex_process_observation_v1", + "pid": 23, "started_at_unix": 101.1, "cleanup_trigger": "scope_completed", + "before_cleanup": {"return_code": 0, "observed_at_unix": 108.9}, + "after_cleanup": {"return_code": 0, "observed_at_unix": 109.1}, + }, + ) + receipt.write_text(json.dumps(value)) + os.utime(receipt, (110, 110)) + Path(declaration["phase_clock"]).write_text("200.0") + return state, receipt + + +@pytest.mark.parametrize("fault", [None, "default", "pending", "clock", "wrong_goal", "late_return", "later_wake"]) +def test_completed_host_source_requires_the_last_exact_return(tmp_path, monkeypatch, fault): + declaration = _source(tmp_path) + state, receipt = _completed_return(declaration) + box, client = _fake_container(monkeypatch) + if fault == "pending": + state.with_name(state.name + ".host-attempt").write_text('{"pending": true}') + elif fault == "clock": + Path(declaration["phase_clock"]).write_text("210") + elif fault in {"wrong_goal", "late_return"}: + value = json.loads(receipt.read_text()) + if fault == "wrong_goal": + value["goal_id"] = "other-goal" + else: + value["native_process"]["after_cleanup"]["observed_at_unix"] = 111.5 + receipt.write_text(json.dumps(value)) + os.utime(receipt, (110, 110)) + elif fault == "later_wake": + later = receipt.parent.with_name("later") / "receipt.json" + later.parent.mkdir() + later.write_bytes(receipt.read_bytes()) + os.utime(later, (111, 111)) + output = tmp_path / "proof.json" + before = state.read_bytes() + if fault: + with pytest.raises(ValueError): + prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, + output=output, idle_host_handoff=fault != "default") + assert not output.exists() + else: + proof = prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, + output=output, idle_host_handoff=True) + assert proof["completed_host"] is True + assert proof["execution_authority_granted"] is False + assert proof["process_started_at"] == 101.1 + assert proof["process_stopped_at"] == 109.1 + assert proof["source_phase_deadline_epoch"] == 200 + assert state.read_bytes() == before + def test_stopped_export_uses_output_created_by_the_real_worker(tmp_path, monkeypatch): from benchmark.runtime import worker from benchmark.tests.test_worker_phase_budget import _phase_worker @@ -301,8 +372,9 @@ def reader(*args, **kwargs): "unrelated_lock", ], ) +@pytest.mark.parametrize("completed_host", [False, True], ids=["legacy-empty", "completed-host"]) def test_real_sqlite_idle_handoff_keeps_source_and_runs_the_retained_session( - tmp_path, monkeypatch, fault + tmp_path, monkeypatch, fault, completed_host ): from benchmark.runtime import worker from benchmark.tests.test_worker_phase_budget import _phase_worker @@ -344,7 +416,10 @@ def test_real_sqlite_idle_handoff_keeps_source_and_runs_the_retained_session( elif method=='account/read': value={'account':{'type':'chatgpt'}} elif method=='account/rateLimits/read': value={'rateLimits':{'primary':{'usedPercent':100 if (markers/'exhausted').exists() else 40}}} elif method=='thread/read': value={'thread':{'id':'original-native','cwd':str(pathlib.Path.cwd())}} - elif method=='thread/turns/list': value={'data':[{'id':'last-completed','status':'completed','startedAt':80,'completedAt':int(time.time())+10 if (markers/'new-native-work').exists() else 90,'error':None,'items':[]}],'nextCursor':None} + elif method=='thread/turns/list': + started,completed=(102,108) if (markers/'completed-host').exists() else (80,90) + if (markers/'new-native-work').exists(): completed=int(time.time())+10 + value={'data':[{'id':'last-completed','status':'completed','startedAt':started,'completedAt':completed,'error':None,'items':[]}],'nextCursor':None} else: raise AssertionError(method) print(json.dumps({'jsonrpc':'2.0','id':request['id'],'result':value}),flush=True) sys.exit(0) @@ -389,17 +464,20 @@ def test_real_sqlite_idle_handoff_keeps_source_and_runs_the_retained_session( ) state.write_text("{}") state.with_name(state.name + ".host-attempt").write_text('{"pending":true}') - state, returned = _idle_return(declaration) + if completed_host: + (control / "completed-host").touch() + state, returned = _completed_return(declaration) + else: + state, returned = _idle_return(declaration) + clock.write_text("105") receipt = json.loads(returned.read_text()) receipt["session"]["session_id"] = "original-native" returned.write_text(json.dumps(receipt)) os.utime(returned, (110, 110)) - # The only latest source wake is the selected returned failure. + # The selected return must still be the latest source wake. for old in Path(env["LOOPX_WAKE_LOG_DIR"]).glob("*/receipt.json"): if old != returned: os.utime(old, (95, 95)) - # Readback fixtures use historical seconds, as does the retained Host proof. - clock.write_text("105") original_uts = os.uname() unrelated_lock = project / "research.lock" unrelated_lock.write_text("retained work evidence") @@ -482,8 +560,16 @@ def resume(): initial = phase.read_bytes() saved = json.loads(initial) assert saved["recovered_session_id"] == "original-native" - assert saved["idle_host_recovery"]["original_host_failure"]["kind"] == "unknown" - assert saved["idle_host_recovery"]["phase_deadline_epoch"] == deadline + recovery = saved["idle_host_recovery"] + if completed_host: + assert recovery["process_stopped_at"] == 109.1 + assert recovery["native"]["started_at"] == 102 + assert recovery["native"]["completed_at"] == 108 + assert "original_host_failure" not in recovery + assert saved["replan_attempts"] == json.loads(original)["replan_attempts"] + else: + assert recovery["original_host_failure"]["kind"] == "unknown" + assert recovery["phase_deadline_epoch"] == deadline assert resume() == str(phase) and phase.read_bytes() == initial for key in ("allowance", "native", "receipt_sha256", "session_profile_digest"): changed = copy.deepcopy(saved) @@ -492,6 +578,13 @@ def resume(): with pytest.raises(ValueError): resume() phase.write_bytes(initial) + if completed_host: + pending = state.with_name(state.name + ".host-attempt") + cleared = pending.read_bytes() + pending.write_text('{"pending": true}') + with pytest.raises(ValueError, match="replay differs"): + resume() + pending.write_bytes(cleared) assert worker.run_once(env)["ok"] assert ( project / "continued-candidate" diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py index 2f2f1e79f7..0c3c449023 100644 --- a/loopx/hosts/codex_checkpoint.py +++ b/loopx/hosts/codex_checkpoint.py @@ -173,15 +173,18 @@ def _idle_native_reader(env: dict[str, str], *, timeout: float, deadline: float, def native_idle_turn(env: dict[str, str], session_id: str, *, process_started_at: float, - timeout: float = 30, sealed_store_paths: tuple[str, ...] | None = None) -> dict[str, Any]: - """Observe a stable completed history predating an uncertain Host launch. + process_stopped_at: float | None = None, timeout: float = 30, + sealed_store_paths: tuple[str, ...] | None = None) -> dict[str, Any]: + """Observe a stable idle session at a stopped Host's process boundary. - The caller separately proves the original namespace stopped, seals its - source, and explicitly requests a finite handoff. This read neither retries - the failed Host nor turns its failure into a completed native Turn. + With a verified completed process window, require its last native Turn + inside that window. Otherwise require completed history before the + uncertain launch. The caller owns source termination and phase admission. """ if (not session_id or any(type(v) not in (int, float) or not math.isfinite(v) - for v in (process_started_at, timeout)) or timeout <= 0): + for v in (process_started_at, timeout)) or timeout <= 0 + or (process_stopped_at is not None and (type(process_stopped_at) not in (int, float) + or not math.isfinite(process_stopped_at) or process_stopped_at < process_started_at))): raise ValueError("Native idle readback requires its original process boundary") deadline = time.monotonic() + timeout with _idle_native_reader(env, timeout=timeout, deadline=deadline, @@ -194,14 +197,18 @@ def request(method, params): return transport.request(method, params) _native_thread_readback(request, env, session_id) turns = _complete_history(request, session_id) + tail = turns[-1] for turn in turns: start, end = turn.get("startedAt"), turn.get("completedAt") if (any(type(v) not in (int, float) or not math.isfinite(v) for v in (start, end)) - or not start <= end < math.floor(process_started_at) - or any(item.get("status") in {"inProgress", "running", "pending"} - for item in turn.get("items", []))): + or start > end or any(item.get("status") in {"inProgress", "running", "pending"} + for item in turn.get("items", []))): + raise ValueError("Native history is not idle before the original Host") + if turn is tail and process_stopped_at is not None: + if not math.floor(process_started_at) <= start <= end <= process_stopped_at: + raise ValueError("Native completed Turn is outside the original Host process") + elif end >= math.floor(process_started_at): raise ValueError("Native history is not idle before the original Host") - tail = turns[-1] if tail.get("error") is not None: raise ValueError("Native idle cursor retains an error") encoded = json.dumps(turns, sort_keys=True, separators=(",", ":")).encode() @@ -210,7 +217,8 @@ def request(method, params): raise ValueError("Native idle history advanced during readback") return {"session_id": session_id, "turn_id": tail["id"], "status": "completed", "completed_at": tail["completedAt"], "turn_count": len(turns), - "visible_history_sha256": hashlib.sha256(encoded).hexdigest()} + "visible_history_sha256": hashlib.sha256(encoded).hexdigest(), + **({"started_at": tail["startedAt"]} if process_stopped_at is not None else {})} def native_failure_error_digest(error: Any, *, cli_projection: bool = False) -> str: