diff --git a/benchmark/README.md b/benchmark/README.md index 9865607d94..8d6d26895f 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -1547,14 +1547,17 @@ python -m benchmark.runtime.scheduler \ --retry-failed-state-sha256 EXACT_FAILED_STATE_SHA256 ``` -The recovery provider requires one returned `quota_exhausted` wake, its original -session/profile binding, terminated native process group and a stable native -`failed` tail with the corresponding structured quota error and no pending tool -effects. Legacy workers can use the same stopped-container proof above, but only -when the exact last returned failure, original session and complete unchanged -filesystem are retained. Unknown errors, missing sessions and incomplete history -remain held. The ordinary completed-checkpoint gate is unchanged; a failed native -Turn is not rewritten or presented as completed. +The recovery provider requires the exact failed-state digest, the original +session/profile binding, a terminated native process group and a stable native +`failed` tail without pending effects. Quota recovery requires the corresponding +structured `quota_exhausted` error and accepts the stopped-container proof for +legacy workers with the exact last failure, original session and retained +filesystem verified byte-for-byte. For other failures, including `unknown`, the +provider-selected rollout must identify one failed Turn with start and completion +records within the original process window through its first observed return. The +native tail must name that Turn and match the original CLI's terminal error. +Missing or ambiguous evidence keeps the failure held. Idle handoff and completed +checkpoints use their own evidence gates. The shared Turn retry rule decides explicit-retry eligibility. A new append-only phase records the original failure, digest, native cursor and deadline, retains diff --git a/benchmark/runtime/interruption_recovery.py b/benchmark/runtime/interruption_recovery.py index 049bd9f52f..7d99e8ae8d 100644 --- a/benchmark/runtime/interruption_recovery.py +++ b/benchmark/runtime/interruption_recovery.py @@ -359,6 +359,49 @@ def verify_idle_host_replay(state_file: Path, recovery: dict, *, env: dict[str, raise ValueError('Idle handoff replay differs from the original qualified source') +def _returned_failure_witness(path: Path, receipt: dict, session_id: str, + started: float, stopped: float) -> dict: + """Seal one complete Codex returned error, not a diagnostic classification.""" + from loopx.control_plane.turn_driver.codex_cli import codex_cli_event_session_id + from loopx.hosts.codex_checkpoint import native_failure_error_digest + from .worker import host_output_paths + + process = receipt.get('native_process') or {} + before, after = process.get('before_cleanup') or {}, process.get('after_cleanup') or {} + returned_at = _number(before.get('observed_at_unix')) + if (receipt.get('timed_out') or receipt.get('error_kind') + or any(type(frame.get('return_code')) is not int or frame['return_code'] != 1 + for frame in (before, after)) + or not started <= returned_at <= stopped + or before.get('signal') is not None or after.get('signal') is not None + or process.get('cleanup_trigger') != 'scope_completed' + or process.get('scope_error_type') is not None): + raise ValueError('Explicit returned failure requires its original native return') + stdout, stderr = host_output_paths(path.parent) + raw, diagnostic = _physical_read(stdout), _physical_read(stderr) + try: + events = [json.loads(line) for line in raw.decode('utf-8').splitlines() if line.strip()] + except (ValueError, UnicodeError) as error: + raise ValueError('Explicit returned failure requires complete native stdout') from error + if not events or any(not isinstance(event, dict) for event in events): + raise ValueError('Explicit returned failure requires native event objects') + threads = [index for index, event in enumerate(events) + if event.get('type') in {'thread.started', 'thread_started'}] + sessions = [codex_cli_event_session_id(events[index]) for index in threads] + starts = [index for index, event in enumerate(events) if event.get('type') == 'turn.started'] + failures = [index for index, event in enumerate(events) if event.get('type') == 'turn.failed'] + if (sessions != [session_id] or len(starts) != 1 or failures != [len(events) - 1] + or not threads[0] < starts[0] < failures[0] + or any(event.get('type') == 'turn.completed' for event in events)): + raise ValueError('Explicit returned failure differs from its original native invocation') + error = events[-1].get('error') + return {'session_id': session_id, 'process_started_at': started, 'process_stopped_at': stopped, + 'process_returned_at': returned_at, + 'stdout_sha256': hashlib.sha256(raw).hexdigest(), 'stderr_sha256': hashlib.sha256(diagnostic).hexdigest(), + 'error_sha256': native_failure_error_digest(error), + 'error_projection_sha256': native_failure_error_digest(error, cli_projection=True)} + + def _verify_failed_wake(path: Path, receipt: dict, failure: dict, env: dict[str, str], *, legacy: bool) -> None: from loopx.control_plane.turn_driver.host_failure import ( @@ -439,6 +482,19 @@ def digest(value): or session.get('session_id') != prior['session_id'] or _number(recovery.get('source_phase_deadline_epoch')) != _number(float(source_deadline))): raise ValueError('Explicit failed continuation retained wake scope changed') + returned = recovery.get('returned_failure') + if failure['kind'] != 'quota_exhausted' or returned is not None: + started, stopped = _process_window(receipt) + cursor = native.get('rollout_cursor') + if (not isinstance(returned, dict) + or returned != _returned_failure_witness(path, receipt, prior['session_id'], started, stopped) + or not digest(native.get('error_sha256')) + or native.get('returned_error_sha256') != returned.get('error_projection_sha256') + or not isinstance(cursor, dict) or cursor.get('turn_id') != native['turn_id'] + or not digest(cursor.get('record_sha256')) + or not started <= _number(cursor.get('started_at')) <= _number(cursor.get('completed_at')) + <= returned['process_returned_at']): + raise ValueError('Explicit failed continuation returned failure proof changed') current_env = process_environment(Path(env['LOOPX_CODEX_HOME']), base=env) current = CodexAgentSession(current_env, context='resume', sandbox=env.get('LOOPX_CODEX_SANDBOX', 'danger-full-access'), receipt={}) @@ -448,7 +504,7 @@ def digest(value): def read_failed_host(state: dict, *, state_file: Path, env: dict[str, str], wake_log_dir: str, source_state_sha256: str) -> dict: - """Qualify an explicit same-session retry of one returned quota failure. + """Qualify an explicit same-session retry of one exact returned failure. The digest binds a caller's fresh repair/allowance decision to this exact failed attempt, not to every subsequent failure. The ordinary scheduler and @@ -471,12 +527,12 @@ def read_failed_host(state: dict, *, state_file: Path, env: dict[str, str], raise ValueError('Explicit Host recovery no longer names the original failed state') prior = state.get('host_attempt', {}) failure = normalize_host_failure_record(prior.get('host_failure')) - if (prior.get('status') != 'failed' or failure['kind'] != 'quota_exhausted' + if (prior.get('status') != 'failed' or state.get('last_wake_status') != 'wake_failed' or type(state.get('last_wake_rc')) is not int or state['last_wake_rc'] != 1 or prior.get('session_id') is None or not explicit_host_failure_retry_available(failure, retry_failed=True)): - raise ValueError('Original Host is not an explicitly repairable returned quota failure') + raise ValueError('Original Host is not an explicitly repairable returned failure') env = process_environment(Path(env['LOOPX_CODEX_HOME']), base=env) attempt = prior.get('attempt_id', '') legacy = bool(env.get('LOOPX_INTERRUPTED_SOURCE')) and not attempt @@ -510,12 +566,17 @@ def read_failed_host(state: dict, *, state_file: Path, env: dict[str, str], or not current.session_id or current.session_id != session.get('session_id') or prior['session_id'] != current.session_id): raise ValueError('Failed Host session binding changed') + returned = (_returned_failure_witness(path, receipt, current.session_id, started, stopped) + if failure['kind'] != 'quota_exhausted' else None) native = native_failed_turn(env, current.session_id, failure_kind=failure['kind'], - process_started_at=started, process_stopped_at=stopped) + process_started_at=started, process_stopped_at=returned['process_returned_at'] if returned else stopped, + **({'returned_error_sha256': returned['error_projection_sha256']} if returned is not None else {})) allowance = require_native_receiver_allowance(env) with current.admission.launch_lifetime(operation='benchmark_failed_host_recovery'): pass - if _physical_read(path) != raw or _physical_read(state_file) != original_state: + if (_physical_read(path) != raw or _physical_read(state_file) != original_state + or (returned is not None and returned != _returned_failure_witness( + path, receipt, current.session_id, started, stopped))): raise ValueError('Original failure changed during recovery readback') return {**({'source_container_id': source['container_id'], 'source_proof_sha256': hashlib.sha256(_physical_read(Path(env['LOOPX_INTERRUPTED_SOURCE']))).hexdigest()} @@ -523,4 +584,5 @@ def read_failed_host(state: dict, *, state_file: Path, env: dict[str, str], '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, + **({'returned_failure': returned} if returned is not None else {}), 'host_failure': failure, 'retry_failed': True, 'native': native, 'allowance': allowance} diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index 85952f31a4..0723d449e3 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -130,7 +130,9 @@ def _verify_failed_phase_replay(phase: Path, state: dict, recovery: dict, *, states: list[Path], digest: str, env: dict[str, str], wake_log_dir: str) -> None: from .interruption_recovery import _physical_read, verify_failed_host_replay_witness - from loopx.control_plane.turn_driver.host_failure import normalize_host_failure_record + from loopx.control_plane.turn_driver.host_failure import ( + explicit_host_failure_retry_available, normalize_host_failure_record, + ) from scripts.external_scheduler_worker import _host_attempt original_ref = state.get("resume_from") @@ -153,7 +155,8 @@ def _verify_failed_phase_replay(phase: Path, state: dict, recovery: dict, *, session = prior.get("session_id") if (prior.get("status") != "failed" or saved.get("last_wake_status") != "wake_failed" or type(saved.get("last_wake_rc")) is not int or saved["last_wake_rc"] != 1 - or failure["kind"] != "quota_exhausted" or failure != recovery.get("host_failure") + or not explicit_host_failure_retry_available(failure, retry_failed=True) + or failure != recovery.get("host_failure") or recovery.get("retry_failed") is not True or native.get("status") != "failed" or not isinstance(session, str) or not session or native.get("session_id") != session or state.get("recovered_session_id") != session @@ -379,7 +382,7 @@ def main() -> int: parser.add_argument("--state-file", required=True) parser.add_argument("--host-timeout", required=True, type=float) parser.add_argument('--retry-failed-state-sha256', - help='Explicitly continue only this exact returned quota failure after provider allowance repair') + 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') args = parser.parse_args() diff --git a/benchmark/tests/test_failed_host_continuation.py b/benchmark/tests/test_failed_host_continuation.py index 9df76293c2..f7c01525cd 100644 --- a/benchmark/tests/test_failed_host_continuation.py +++ b/benchmark/tests/test_failed_host_continuation.py @@ -3,6 +3,7 @@ from contextlib import contextmanager import copy +from datetime import datetime, timezone import hashlib import json import os @@ -112,6 +113,7 @@ def test_explicit_scheduler_requires_complete_same_phase_evidence(tmp_path, orig @pytest.mark.parametrize('kind,attempt,explicit,allowed', [ ('quota_exhausted', 1, False, False), ('quota_exhausted', 1, True, True), ('quota_exhausted', 7, True, True), ('output_budget_exhausted', 1, True, False), + ('unknown', 2, False, False), ('unknown', 2, True, True), ('provider_capacity', 2, True, True), ('provider_capacity', 3, True, False), ]) def test_explicit_retry_uses_shared_turn_rule_without_automatic_quota_retry(kind, attempt, explicit, allowed): @@ -124,7 +126,7 @@ def test_explicit_retry_uses_shared_turn_rule_without_automatic_quota_retry(kind @pytest.mark.parametrize('fault', [None, 'in_progress', 'pending_effect', 'wrong_session', - 'wrong_workspace', 'wrong_error', 'prose_only', 'changed_history', 'outside_window', 'empty']) + 'wrong_workspace', 'wrong_error', 'prose_only', 'changed_history', 'outside_window', 'empty', 'pending_prefix']) def test_failed_native_readback_preserves_strict_checkpoint_and_unknown_error_gate(tmp_path, monkeypatch, fault): turns = [dict(id='first', status='completed', items=[]), dict(id='last', status='failed', startedAt=110, completedAt=120, error={'codexErrorInfo': 'usageLimitExceeded'}, @@ -133,6 +135,8 @@ def test_failed_native_readback_preserves_strict_checkpoint_and_unknown_error_ga turns[-1]['status'] = 'inProgress' elif fault == 'pending_effect': turns[-1]['items'][0]['status'] = 'inProgress' + elif fault == 'pending_prefix': + turns[0]['items'] = [{'type': 'commandExecution', 'status': 'pending'}] elif fault == 'wrong_error': turns[-1]['error']['codexErrorInfo'] = 'unknownFutureCode' elif fault == 'prose_only': @@ -160,7 +164,7 @@ def reader(*a, **kw): def read(): return native.native_failed_turn({'LOOPX_PROJECT': str(tmp_path)}, 'original', failure_kind='quota_exhausted', process_started_at=100, process_stopped_at=125) - if fault: + if fault not in {None, 'pending_prefix'}: with pytest.raises(ValueError): read() else: @@ -172,9 +176,111 @@ def read(): assert not set(calls) - {'thread/read', 'thread/turns/list'} +@pytest.mark.parametrize('fault', [None, 'no_witness', 'missing_error', 'wrong_error', 'new_error_field', + 'structured_conflict', 'running', 'pending_prefix', 'outside_window', 'changed_history', 'changed_source', + 'missing_rollout', 'later_foreign_tail', 'same_second_foreign_tail', 'late_rollout', 'coarse_rollout', + 'ambiguous_rollout']) +def test_returned_unknown_failure_binds_error_and_isolated_stable_native_store(tmp_path, monkeypatch, fault): + home = tmp_path / 'native' + home.mkdir() + (home / 'config.toml').write_text('') + database = home / 'state_5.sqlite' + database.write_bytes(b'original retained native store') + original = database.read_bytes() + # Diagnostic-looking words never reclassify this retained typed unknown. + cli_error = {'message': 'Synthetic failure mentioning quota and idle'} + native_error = {**cli_error, 'codexErrorInfo': 'other', 'additionalDetails': None, 'misalignment': None} + turns = [dict(id='prefix', status='completed', items=[]), + dict(id='original-failed-turn', status='failed', startedAt=110, completedAt=120, + error=copy.deepcopy(native_error), items=[])] + rollout = home / 'sessions' / 'rollout-fixture-original.jsonl' + rollout.parent.mkdir() + markers = [ + (110.5, {'type': 'task_started', 'turn_id': 'original-failed-turn'}), + ({'late_rollout': 120.5, 'coarse_rollout': 120.0}.get(fault, 120.125), + {'type': 'task_complete', 'turn_id': 'original-failed-turn', 'error': cli_error}), + ] + if fault == 'ambiguous_rollout': + markers.insert(1, (112.0, {'type': 'task_started', 'turn_id': 'foreign-turn'})) + records = [{'type': 'session_meta', 'payload': {'id': 'original', 'cwd': str(tmp_path)}}] + records.extend({'type': 'event_msg', 'timestamp': datetime.fromtimestamp(stamp, timezone.utc).isoformat(), + 'payload': payload} for stamp, payload in markers) + if fault != 'missing_rollout': + rollout.write_text(''.join(json.dumps(record) + '\n' for record in records)) + if fault == 'missing_error': + turns[-1].pop('error') + elif fault == 'wrong_error': + turns[-1]['error']['message'] = 'A different failure' + elif fault == 'new_error_field': + turns[-1]['error']['futureField'] = 'additional cause' + elif fault == 'structured_conflict': + turns[-1]['error']['codexErrorInfo'] = 'usageLimitExceeded' + elif fault == 'running': + turns[-1]['status'] = 'inProgress' + elif fault == 'pending_prefix': + turns[0]['items'] = [{'type': 'commandExecution', 'status': 'pending'}] + elif fault == 'outside_window': + turns[-1]['startedAt'] = 99 + elif fault in ('later_foreign_tail', 'same_second_foreign_tail'): + stamp = 121 if fault == 'later_foreign_tail' else 120 + turns.append(dict(turns[-1], id='foreign-turn', startedAt=stamp, completedAt=stamp)) + calls = [] + class Transport: + def request(self, method, params): + calls.append(method) + if method == 'thread/read': + return {'thread': {'id': 'original', 'cwd': str(tmp_path), 'path': str(rollout)}} + assert method == 'thread/turns/list' + rows = copy.deepcopy(turns) + if fault == 'changed_history' and calls.count(method) > 1: + rows[0]['items'] = [{'type': 'agentMessage', 'text': 'later work'}] + return {'data': rows, 'nextCursor': None} + @contextmanager + def reader(env, **kwargs): + observer = Path(env['CODEX_HOME']) + assert observer != home and (observer / database.name).read_bytes() == original + # Native app-server may write its own index, never the retained source. + (observer / database.name).write_bytes(b'observer index update') + if fault == 'changed_source': + database.write_bytes(b'concurrent original writer') + yield Transport() + monkeypatch.setattr(native, 'native_checkpoint_reader', reader) + returned_digest = hashlib.sha256(json.dumps(cli_error, sort_keys=True, separators=(',', ':')).encode()).hexdigest() + kwargs = {} if fault == 'no_witness' else {'returned_error_sha256': returned_digest} + def read(): + return native.native_failed_turn({'LOOPX_PROJECT': str(tmp_path), 'CODEX_HOME': str(home)}, + 'original', failure_kind='unknown', process_started_at=100.25, process_stopped_at=120.25, **kwargs) + if fault: + with pytest.raises(ValueError): + read() + else: + result = read() + assert result['failure_kind'] == 'unknown' and result['status'] == 'failed' + assert result['turn_id'] == 'original-failed-turn' and result['turn_count'] == 2 + assert result['returned_error_sha256'] == returned_digest + assert result['error_sha256'] == hashlib.sha256(json.dumps(native_error, sort_keys=True, + separators=(',', ':')).encode()).hexdigest() + assert result['rollout_cursor'] == {'turn_id': 'original-failed-turn', 'started_at': 110.5, + 'completed_at': 120.125, 'record_sha256': hashlib.sha256( + b''.join(rollout.read_bytes().splitlines(keepends=True)[1:])).hexdigest()} + with pytest.raises(ValueError, match='completed final cursor'): + native._complete_history(Transport().request, 'original') + if fault != 'changed_source': + assert database.read_bytes() == original + assert not set(calls) - {'thread/read', 'thread/turns/list'} + + @pytest.mark.skipif(os.name != 'posix', reason='Original process group readback requires POSIX') -@pytest.mark.parametrize('fault', [None, 'stop', 'profile', 'session', 'source_digest', 'exhausted', 'unknown_allowance', 'stop_after_prepare']) -def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path, monkeypatch, fault): +@pytest.mark.parametrize('kind,prior_attempt,fault', [ + ('quota_exhausted', 0, fault) for fault in + (None, 'stop', 'profile', 'session', 'source_digest', 'exhausted', 'unknown_allowance', 'stop_after_prepare') +] + [ + ('unknown', 1, fault) for fault in + (None, 'stop', 'profile', 'session', 'source_digest', 'exhausted', 'unknown_allowance', + 'stop_after_prepare', 'no_return', 'after_return', 'missing_error', 'native_error', 'pending_native', + 'stdout_session', 'stdout_truncated') +]) +def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path, monkeypatch, kind, prior_attempt, 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 @@ -182,9 +288,19 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path env, project, runtime, _source, goal_state, _clock = _phase_worker(tmp_path, monkeypatch) monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / 'packages/loopx-codex-provider-routing/src')) monkeypatch.setattr(worker, 'time', time) + failure_error = ({'codexErrorInfo': 'usageLimitExceeded'} if kind == 'quota_exhausted' + else {'message': 'Synthetic native task failed'}) + (project / 'failure-error.json').write_text(json.dumps(failure_error)) + native_home = Path(env['LOOPX_CODEX_HOME']) + native_home.mkdir(parents=True, exist_ok=True) + rollout = native_home / 'sessions' / 'rollout-fixture-original-native.jsonl' + (native_home / 'state_5.sqlite').write_bytes(b'retained native fixture') + retained_store = (native_home / 'state_5.sqlite').read_bytes() Path(env['CODEX_BIN']).write_text(f'#!{sys.executable}\n' + ''' import json,os,pathlib,sys,time +from datetime import datetime, timezone history=pathlib.Path('native-history.json') +rollout=pathlib.Path(os.environ['LOOPX_CODEX_HOME'])/'sessions'/'rollout-fixture-original-native.jsonl' if 'app-server' in sys.argv: for line in sys.stdin: message=json.loads(line) @@ -194,24 +310,37 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path elif method=='account/read': result={'account':{'type':'chatgpt'}} elif method=='account/rateLimits/read': result=json.loads(pathlib.Path('observed-allowance.json').read_text()) if pathlib.Path('observed-allowance.json').exists() else {'rateLimits':{'primary':{'usedPercent':40}}} - elif method=='thread/read': result={'thread':{'id':'original-native','cwd':str(pathlib.Path.cwd())}} + elif method=='thread/read': result={'thread':{'id':'original-native','cwd':str(pathlib.Path.cwd()),'path':str(rollout)}} 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') +started=time.time() print(json.dumps({'type':'thread.started','thread_id':'original-native'}),flush=True) +print(json.dumps({'type':'turn.started'}),flush=True) payload=sys.stdin.read() with pathlib.Path('native-inputs').open('a') as f: f.write(json.dumps(payload)+'\\n') if pathlib.Path('allowance-repaired').exists(): pathlib.Path('continued-work').write_text('next candidate on retained session') print(json.dumps({'type':'turn.completed'}),flush=True) sys.exit(0) -now=int(time.time()) -history.write_text(json.dumps({'id':'quota-failed-work','status':'failed', - 'error':{'codexErrorInfo':'usageLimitExceeded'},'startedAt':now,'completedAt':now, +completed=time.time() +error=json.loads(pathlib.Path('failure-error.json').read_text()) +native_error={**error,'additionalDetails':None} +if 'codexErrorInfo' not in native_error: + native_error.update(codexErrorInfo='other',misalignment=None) +history.write_text(json.dumps({'id':'failed-work','status':'failed', + 'error':native_error,'startedAt':int(started),'completedAt':int(completed), 'items':[{'type':'commandExecution','status':'completed'}]})) -print(json.dumps({'type':'turn.failed','error':{'codexErrorInfo':'usageLimitExceeded'}}),flush=True) +rollout.parent.mkdir(parents=True, exist_ok=True) +meta={'type':'session_meta','payload':{'id':'original-native','cwd':str(pathlib.Path.cwd())}} +def marker(stamp, kind, **payload): + return {'type':'event_msg','timestamp':datetime.fromtimestamp(stamp, timezone.utc).isoformat(), + 'payload':{'type':kind,'turn_id':'failed-work',**payload}} +rollout.write_text(''.join(json.dumps(record)+'\\n' for record in (meta, + marker(started,'task_started'),marker(completed,'task_complete',error=error)))) +print(json.dumps({'type':'turn.failed','error':error}),flush=True) sys.exit(1) ''') for key, value in env.items(): @@ -224,13 +353,17 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path args.goal_id, args.agent_id = env['LOOPX_GOAL_ID'], env['LOOPX_AGENT_ID'] args.wake_cmd = shlex.join([sys.executable, '-m', 'benchmark.runtime.worker']) args.wake_stop_exit_code, args.wake_timeout_seconds = 75, 90 + if prior_attempt: + Path(args.state_file).parent.mkdir(parents=True, exist_ok=True) + Path(args.state_file).write_text(json.dumps({'recovered_attempt_count': prior_attempt})) monkeypatch.setattr(scheduler_fixture.worker, 'run_quota_should_run', lambda *a, **kw: scheduler_fixture._hint_payload(should_run=True)) assert scheduler_fixture.worker.run_worker(args) == 1 original = Path(args.state_file) original_bytes = original.read_bytes() old = json.loads(original_bytes) - assert old['host_attempt']['host_failure']['kind'] == 'quota_exhausted' + assert old['host_attempt']['host_failure']['kind'] == kind + assert old['host_attempt']['attempt'] == prior_attempt + 1 assert old['host_attempt']['status'] == 'failed' assert scheduler_fixture.worker.run_worker(args) == 2 # Still no automatic retry. assert len((project / 'native-argv').read_text().splitlines()) == 1 @@ -238,6 +371,7 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path assert len(receipts) == 1 receipt = json.loads(receipts[0].read_bytes()) assert receipt['native_process']['after_cleanup']['return_code'] == 1 + assert receipt['host_failure']['attempt'] == 1 digest = hashlib.sha256(original_bytes).hexdigest() if fault == 'stop': from loopx.control_plane.goals.activation_service import set_goal_activation_state @@ -253,6 +387,27 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path elif fault in ('exhausted', 'unknown_allowance'): (project / 'observed-allowance.json').write_text(json.dumps({'rateLimits': {}} if fault == 'unknown_allowance' else {'rateLimits': {'primary': {'usedPercent': 100, 'resetsAt': int(time.time()) + 1000}, 'credits': {'hasCredits': False}}})) + elif fault == 'no_return': + receipt['native_process']['before_cleanup']['return_code'] = None + receipts[0].write_text(json.dumps(receipt)) + elif fault == 'after_return': + task_start = json.loads(rollout.read_text().splitlines()[1])['timestamp'] + receipt['native_process']['before_cleanup']['observed_at_unix'] = datetime.fromisoformat(task_start).timestamp() + receipts[0].write_text(json.dumps(receipt)) + elif fault in ('missing_error', 'native_error', 'pending_native'): + history = json.loads((project / 'native-history.json').read_text()) + if fault == 'missing_error': + history.pop('error') + elif fault == 'native_error': + history['error']['message'] = 'A different failed native task' + else: + history['items'][0]['status'] = 'inProgress' + (project / 'native-history.json').write_text(json.dumps(history)) + elif fault in ('stdout_session', 'stdout_truncated'): + output = receipts[0].parent / 'stdout.jsonl' + raw_output = output.read_text() + output.write_text(raw_output.replace('original-native', 'another-native') + if fault == 'stdout_session' else raw_output[:-5]) def resume(**kwargs): return 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), @@ -268,9 +423,29 @@ def resume(**kwargs): assert phase != original phase_initial = json.loads(phase.read_bytes()) assert phase_initial['recovered_session_id'] == 'original-native' - assert phase_initial['recovered_attempt_count'] == 1 + assert phase_initial['recovered_attempt_count'] == prior_attempt + 1 assert phase_initial['failed_host_recovery']['native']['status'] == 'failed' assert phase_initial['failed_host_recovery']['phase_deadline_epoch'] == deadline + assert phase_initial['failed_host_recovery']['host_failure']['kind'] == kind + assert (native_home / 'state_5.sqlite').read_bytes() == retained_store + if kind == 'unknown': + proof = phase_initial['failed_host_recovery']['returned_failure'] + error_digest = hashlib.sha256(json.dumps(failure_error, sort_keys=True, + separators=(',', ':')).encode()).hexdigest() + expected_native_error = {**failure_error, 'codexErrorInfo': 'other', 'additionalDetails': None, + 'misalignment': None} + native_error_digest = hashlib.sha256(json.dumps(expected_native_error, sort_keys=True, + separators=(',', ':')).encode()).hexdigest() + assert proof['error_sha256'] == error_digest + assert proof['error_projection_sha256'] == error_digest + assert proof['stdout_sha256'] == hashlib.sha256( + (receipts[0].parent / 'stdout.jsonl').read_bytes()).hexdigest() + assert phase_initial['failed_host_recovery']['native']['turn_id'] == 'failed-work' + assert phase_initial['failed_host_recovery']['native']['error_sha256'] == native_error_digest + assert phase_initial['failed_host_recovery']['native']['returned_error_sha256'] == error_digest + cursor = phase_initial['failed_host_recovery']['native']['rollout_cursor'] + assert cursor['turn_id'] == 'failed-work' + assert proof['process_started_at'] <= cursor['started_at'] <= cursor['completed_at'] <= proof['process_returned_at'] assert resume() == str(phase) if fault is None: from benchmark.runtime import codex_readiness @@ -292,7 +467,20 @@ def unexpected_observer(*args, **kwargs): with pytest.raises(ValueError, match="wake receipt changed"): resume() receipt_path.write_bytes(returned_bytes) - for field in ("allowance", "native", "receipt_sha256", "session_profile_digest"): + if kind == 'unknown': + output = receipt_path.parent / 'stdout.jsonl' + returned_output = output.read_bytes() + output.write_bytes(returned_output + b'\n') + with pytest.raises(ValueError, match='returned failure'): + resume() + output.write_bytes(returned_output) + incomplete = copy.deepcopy(phase_initial) + del incomplete['failed_host_recovery']['native']['rollout_cursor'] + phase.write_text(json.dumps(incomplete)) + with pytest.raises(ValueError, match='returned failure'): + resume() + for field in ("allowance", "native", "receipt_sha256", "session_profile_digest", + *(("returned_failure",) if kind == 'unknown' else ())): incomplete = copy.deepcopy(phase_initial) del incomplete["failed_host_recovery"][field] phase.write_text(json.dumps(incomplete)) @@ -314,7 +502,7 @@ def unexpected_observer(*args, **kwargs): args.state_file = str(phase) assert scheduler_fixture.worker.run_worker(args) == 1 failed_again = phase.read_bytes() - assert json.loads(failed_again)['host_attempt']['attempt'] == 2 + assert json.loads(failed_again)['host_attempt']['attempt'] == prior_attempt + 2 assert resume() == str(phase) and phase.read_bytes() == failed_again assert scheduler_fixture.worker.run_worker(args) == 2 # A later explicit repair names the new failure, rather than erasing it. diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py index 9f729e5efe..2110efa1b2 100644 --- a/loopx/hosts/codex_checkpoint.py +++ b/loopx/hosts/codex_checkpoint.py @@ -117,6 +117,30 @@ def _native_store_components(home: Path) -> list[dict[str, str]]: return members +def _native_observation_fingerprint(home: Path, *, deadline: float) -> dict[str, str | None]: + """Bind an isolated observation to unchanged physical native-store bytes.""" + sources = [home / 'config.toml', *(Path(m['source_path']) for m in _native_store_components(home))] + for source in tuple(sources): + if source.suffix == '.sqlite': + sources.extend(p for suffix in ('-wal', '-journal') + if (p := Path(str(source) + suffix)).exists()) + fingerprint = {} + for source in sources: + for entry in [source, *sorted(source.rglob('*'))] if source.is_dir() else [source]: + if entry.is_symlink() or entry.resolve(strict=True) != entry: + raise ValueError('Native failure observation requires physical source members') + if time.monotonic() >= deadline: + raise TimeoutError('Failed native recovery readback window closed') + if entry.is_dir(): + fingerprint[str(entry.relative_to(home))] = None + elif entry.is_file(): + with entry.open('rb') as stream: + fingerprint[str(entry.relative_to(home))] = hashlib.file_digest(stream, 'sha256').hexdigest() + else: + raise ValueError('Native failure observation requires regular source members') + return fingerprint + + @contextmanager def _idle_native_reader(env: dict[str, str], *, timeout: float, deadline: float, sealed_store_paths: tuple[str, ...] | None = None): @@ -188,52 +212,105 @@ def request(method, params): "visible_history_sha256": hashlib.sha256(encoded).hexdigest()} +def native_failure_error_digest(error: Any, *, cli_projection: bool = False) -> str: + """Hash the Codex error object without interpreting or retaining its prose. + + The CLI can omit app-server's optional null fields and catch-all `other` + discriminator. Its comparison projection drops only those protocol + defaults; the raw native object is separately sealed without that projection. + """ + if not isinstance(error, dict) or not isinstance(error.get('message'), str) or not error['message'].strip(): + raise ValueError('Returned native failure requires its error object') + value = dict(error) + if cli_projection: + for key in ('codexErrorInfo', 'additionalDetails', 'misalignment'): + if value.get(key) is None or (key == 'codexErrorInfo' and value[key] == 'other'): + value.pop(key, None) + return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(',', ':'), + allow_nan=False).encode()).hexdigest() + + def native_failed_turn(env: dict[str, str], session_id: str, *, failure_kind: str, process_started_at: float, process_stopped_at: float, + returned_error_sha256: str | None = None, timeout: float = 30) -> dict[str, Any]: - """Read a retained quota-failed cursor without rewriting it as completed. + """Read a retained failed cursor without rewriting it as completed. The caller independently proves the exact Host returned and explicitly - requests recovery after repairing its provider allowance. This observation - neither starts/resumes a native Thread nor supplies a checkpoint certificate. + requests recovery after an operator repair. Non-quota failures also require + the exact returned error digest and a precise provider rollout cursor. + This observation starts no Turn and supplies no checkpoint. """ from loopx.control_plane.turn_driver.codex_cli import _codex_error_info_category + from loopx.control_plane.turn_driver.host_failure import require_host_failure_kind + from .codex_rollout_file import read_failed_turn_cursor - if (failure_kind != "quota_exhausted" or not session_id + require_host_failure_kind(failure_kind) + if ((returned_error_sha256 is None and failure_kind != "quota_exhausted") or not session_id + or (returned_error_sha256 is not None and (not isinstance(returned_error_sha256, str) + or re.fullmatch(r'[a-f0-9]{64}', returned_error_sha256) is None)) or any(type(x) not in (int, float) or not math.isfinite(x) for x in (process_started_at, process_stopped_at, timeout)) or process_started_at > process_stopped_at or timeout <= 0): - raise ValueError("Failed native recovery requires its quota failure and process window") + raise ValueError("Failed native recovery requires its returned failure and process window") deadline = time.monotonic() + timeout - with native_checkpoint_reader(env, timeout=timeout, deadline=deadline) as transport: + home = Path(env['CODEX_HOME']) if returned_error_sha256 is not None else None + original = _native_observation_fingerprint(home, deadline=deadline) if home is not None else {} + observer = (_idle_native_reader(env, timeout=timeout, deadline=deadline) if home is not None + else native_checkpoint_reader(env, timeout=timeout, deadline=deadline)) + with observer as transport: def request(method, params): remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError("Failed native recovery readback window closed") transport.response_timeout_sec = remaining return transport.request(method, params) - _native_thread_readback(request, env, session_id) + thread = _native_thread_readback(request, env, session_id) + cursor = None + if home is not None: + source = thread.get('path') + rollout = Path(source) if isinstance(source, str) else None + if rollout is None or not rollout.is_relative_to(home): + raise ValueError('Failed native rollout must be in the original store') + relative = rollout.relative_to(home) + if (not relative.parts or relative.parts[0] not in ('sessions', 'archived_sessions') + or original.get(str(relative)) is None): + raise ValueError('Failed native rollout must be a retained source file') + cursor = read_failed_turn_cursor(rollout, session_id=session_id, workspace=env['LOOPX_PROJECT'], + process_started_at=process_started_at, process_stopped_at=process_stopped_at, deadline=deadline) turns = _retained_history(request, session_id) tail = turns[-1] error = tail.get("error") - category, structured = _codex_error_info_category( - error.get("codexErrorInfo") if isinstance(error, dict) else None) + error_digest = native_failure_error_digest(error) if returned_error_sha256 is not None else None + error_projection = (native_failure_error_digest(error, cli_projection=True) + if returned_error_sha256 is not None else None) + category, structured = _codex_error_info_category(error.get('codexErrorInfo') if isinstance(error, dict) else None) start, end = tail.get("startedAt"), tail.get("completedAt") - if (tail.get("status") != "failed" or not structured or category != failure_kind + end_bound = process_stopped_at if cursor is not None else math.ceil(process_stopped_at) + if (tail.get("status") != "failed" + or (cursor is not None and tail.get('id') != cursor['turn_id']) + or (returned_error_sha256 is not None and error_projection != returned_error_sha256) + or (returned_error_sha256 is None and (not structured or category != failure_kind)) or any(type(value) not in (int, float) or not math.isfinite(value) for value in (start, end)) - or not math.floor(process_started_at) <= start <= end <= math.ceil(process_stopped_at)): + or not math.floor(process_started_at) <= start <= end <= end_bound): raise ValueError("Native failure differs from the original returned Host") + effect_turns = turns if returned_error_sha256 is not None else [tail] if any(item.get("status") in {"inProgress", "running", "pending"} - for item in tail.get("items", [])): + for turn in effect_turns for item in turn.get("items", [])): raise ValueError("Failed native cursor still has pending effects") encoded = json.dumps(turns, sort_keys=True, separators=(",", ":")).encode() if json.dumps(_retained_history(request, session_id), sort_keys=True, separators=(",", ":")).encode() != encoded: raise ValueError("Failed native history advanced during readback") - return {"session_id": session_id, "turn_id": tail["id"], "status": "failed", + result = {"session_id": session_id, "turn_id": tail["id"], "status": "failed", "failure_kind": failure_kind, "turn_count": len(turns), + **({'error_sha256': error_digest, 'returned_error_sha256': error_projection, + 'rollout_cursor': cursor} if cursor is not None else {}), "visible_history_sha256": hashlib.sha256(encoded).hexdigest()} + if home is not None and _native_observation_fingerprint(home, deadline=deadline) != original: + raise ValueError('Failed native source changed during isolated readback') + return result def _guard_cli_arguments(command: str, guard: dict) -> list[str]: @@ -323,7 +400,7 @@ def arguments(command): raise ValueError("native pending guard call identity differs") -def _native_thread_readback(request, env: dict[str, str], session_id: str) -> None: +def _native_thread_readback(request, env: dict[str, str], session_id: str) -> dict[str, Any]: readback = request("thread/read", {"threadId": session_id, "includeTurns": False}) if readback.get("thread", {}).get("id") != session_id: raise ValueError("native checkpoint session identity mismatch") @@ -331,6 +408,7 @@ def _native_thread_readback(request, env: dict[str, str], session_id: str) -> No if (not workspace.is_absolute() or workspace.resolve(strict=True) != workspace or readback["thread"].get("cwd") != str(workspace)): raise ValueError("native checkpoint session workspace differs from the declared code basis") + return readback['thread'] def native_completed_cursor(env: dict[str, str], session_id: str, *, diff --git a/loopx/hosts/codex_rollout_file.py b/loopx/hosts/codex_rollout_file.py index 8e6061aca9..766debe09b 100644 --- a/loopx/hosts/codex_rollout_file.py +++ b/loopx/hosts/codex_rollout_file.py @@ -2,11 +2,20 @@ from __future__ import annotations +from datetime import datetime +import hashlib +import json import os from pathlib import Path +import re +import time from typing import BinaryIO +_MAX_ROLLOUT_RECORD_BYTES = 16 * 1024 * 1024 +_ROLLOUT_TIMESTAMP = re.compile(r'\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(?:\.(\d{1,9}))?(?:Z|[+-]\d\d:\d\d)') + + def open_rollout(path: Path) -> BinaryIO: if os.name != "nt": return path.open("rb") @@ -45,3 +54,61 @@ def open_rollout(path: Path) -> BinaryIO: except BaseException: os.close(descriptor) raise + + +def _rollout_records(stream: BinaryIO, *, deadline: float): + while True: + if time.monotonic() >= deadline: + raise TimeoutError('Failed native rollout readback window closed') + raw = stream.readline(_MAX_ROLLOUT_RECORD_BYTES + 1) + if not raw: + return + if len(raw) > _MAX_ROLLOUT_RECORD_BYTES or not raw.endswith(b'\n'): + raise ValueError('Failed native rollout has an incomplete or oversized record') + row = json.loads(raw) + if not isinstance(row, dict) or not isinstance(row.get('payload'), dict): + raise ValueError('Failed native rollout has an invalid record') + yield raw, row, row['payload'] + + +def read_failed_turn_cursor(path: Path, *, session_id: str, workspace: str, + process_started_at: float, process_stopped_at: float, + deadline: float) -> dict[str, str | float]: + """Read one failed native Turn from the precise records of the observed process.""" + with open_rollout(path) as stream: + records = _rollout_records(stream, deadline=deadline) + _, first, meta = next(records, (None, {}, {})) + if (first.get('type') != 'session_meta' or meta.get('id') != session_id + or meta.get('cwd') != workspace): + raise ValueError('Failed native rollout differs from the bound session') + markers = [] + for raw, row, payload in records: + kind = payload.get('type') + if row.get('type') != 'event_msg' or kind not in ( + 'task_started', 'task_complete', 'task_completed', 'turn_aborted'): + continue + timestamp = row.get('timestamp') + match = _ROLLOUT_TIMESTAMP.fullmatch(timestamp) if isinstance(timestamp, str) else None + if match is None: + raise ValueError('Failed native rollout requires precise turn timestamps') + observed = datetime.fromisoformat(timestamp).timestamp() + # A timestamp truncated to its recorded precision must fit entirely + # within the observed process window. + latest = observed + 10 ** -min(len(match[1] or ''), 6) + if not markers and latest <= process_started_at: + continue + if observed < process_started_at or latest > process_stopped_at or len(markers) == 2: + raise ValueError('Native rollout has another turn in or after the returned process window') + markers.append((raw, kind, observed, payload)) + + if len(markers) != 2: + raise ValueError('Failed native rollout requires one returned turn') + started_raw, start_kind, started, start = markers[0] + ended_raw, end_kind, ended, end = markers[1] + turn_id = start.get('turn_id') + if (start_kind != 'task_started' or end_kind not in ('task_complete', 'task_completed') or started > ended + or not isinstance(turn_id, str) or not turn_id or end.get('turn_id') != turn_id + or not isinstance(end.get('error'), dict)): + raise ValueError('Native rollout did not record the exact failed turn') + return {'turn_id': turn_id, 'started_at': started, 'completed_at': ended, + 'record_sha256': hashlib.sha256(started_raw + ended_raw).hexdigest()}