From 7fc5956dc5536f83a1f6abfb362396118eba3cf5 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 11 Nov 2026 00:01:12 +0800
Subject: [PATCH 1/3] fix(benchmark): recover exact returned native Host
 failures

Bind an explicit scheduler retry to the original failed attempt, process return, session, and matching native error. Observe the retained native store in isolation and continue through the existing typed retry and admission rules.

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/runtime/interruption_recovery.py    |  65 ++++++-
 benchmark/runtime/scheduler.py                |   9 +-
 .../tests/test_failed_host_continuation.py    | 164 ++++++++++++++++--
 loopx/hosts/codex_checkpoint.py               |  84 +++++++--
 4 files changed, 292 insertions(+), 30 deletions(-)

diff --git a/benchmark/runtime/interruption_recovery.py b/benchmark/runtime/interruption_recovery.py
index 049bd9f52f..ab827c98ae 100644
--- a/benchmark/runtime/interruption_recovery.py
+++ b/benchmark/runtime/interruption_recovery.py
@@ -359,6 +359,47 @@ 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 {}
+    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 <= _number(before.get('observed_at_unix')) <= 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,
+            '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 +480,14 @@ 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)
+        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')):
+            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 +497,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 +520,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 +559,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=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 +577,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..582073e19b 100644
--- a/benchmark/tests/test_failed_host_continuation.py
+++ b/benchmark/tests/test_failed_host_continuation.py
@@ -112,6 +112,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 +125,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 +134,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 +163,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 +175,89 @@ 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'])
+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=[])]
+    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
+    calls = []
+    class Transport:
+        def request(self, method, params):
+            calls.append(method)
+            if method == 'thread/read':
+                return {'thread': {'id': 'original', 'cwd': str(tmp_path)}}
+            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, process_stopped_at=125, **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()
+        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', '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,6 +265,13 @@ 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)
+    (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
 history=pathlib.Path('native-history.json')
@@ -201,6 +291,7 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path
     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)
+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():
@@ -208,10 +299,14 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path
     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,
+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':now,'completedAt':now,
     'items':[{'type':'commandExecution','status':'completed'}]}))
-print(json.dumps({'type':'turn.failed','error':{'codexErrorInfo':'usageLimitExceeded'}}),flush=True)
+print(json.dumps({'type':'turn.failed','error':error}),flush=True)
 sys.exit(1)
 ''')
     for key, value in env.items():
@@ -224,13 +319,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 +337,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 +353,23 @@ 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 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 +385,26 @@ 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
     assert resume() == str(phase)
     if fault is None:
         from benchmark.runtime import codex_readiness
@@ -292,7 +426,15 @@ 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)
+        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 +456,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..b9967625cf 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,24 +212,52 @@ 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; it never classifies an unknown error from
+    its message. 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
 
-    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:
+    original = (_native_observation_fingerprint(Path(env['CODEX_HOME']), deadline=deadline)
+                if returned_error_sha256 is not None else None)
+    observer = (_idle_native_reader(env, timeout=timeout, deadline=deadline) if original 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:
@@ -216,24 +268,34 @@ def request(method, params):
         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
+        if (tail.get("status") != "failed"
+                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)):
             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}
+                   if error_digest is not None else {}),
                 "visible_history_sha256": hashlib.sha256(encoded).hexdigest()}
+    if original is not None and _native_observation_fingerprint(Path(env['CODEX_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]:

From 4357fa360d3951409a8463ff794fb8aea96f308e Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 11 Nov 2026 00:18:12 +0800
Subject: [PATCH 2/3] fix(benchmark): bind returned failure to its native Turn

Require one failed cursor from the provider-selected rollout inside the observed return window. Confirm the app-server tail names that cursor before accepting the original CLI error proof.

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/runtime/interruption_recovery.py    | 13 +++-
 .../tests/test_failed_host_continuation.py    | 60 +++++++++++++++--
 loopx/hosts/codex_checkpoint.py               | 38 ++++++++---
 loopx/hosts/codex_rollout_file.py             | 67 +++++++++++++++++++
 4 files changed, 157 insertions(+), 21 deletions(-)

diff --git a/benchmark/runtime/interruption_recovery.py b/benchmark/runtime/interruption_recovery.py
index ab827c98ae..7d99e8ae8d 100644
--- a/benchmark/runtime/interruption_recovery.py
+++ b/benchmark/runtime/interruption_recovery.py
@@ -368,10 +368,11 @@ def _returned_failure_witness(path: Path, receipt: dict, session_id: str,
 
     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 <= _number(before.get('observed_at_unix')) <= stopped
+            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):
@@ -395,6 +396,7 @@ def _returned_failure_witness(path: Path, receipt: dict, session_id: str,
         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)}
@@ -483,10 +485,15 @@ def digest(value):
     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 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',
@@ -562,7 +569,7 @@ def read_failed_host(state: dict, *, state_file: Path, env: dict[str, str],
     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'):
diff --git a/benchmark/tests/test_failed_host_continuation.py b/benchmark/tests/test_failed_host_continuation.py
index 582073e19b..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
@@ -176,7 +177,9 @@ def read():
 
 
 @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'])
+    '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()
@@ -190,6 +193,20 @@ def test_returned_unknown_failure_binds_error_and_isolated_stable_native_store(t
     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':
@@ -204,12 +221,15 @@ def test_returned_unknown_failure_binds_error_and_isolated_stable_native_store(t
         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)}}
+                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:
@@ -229,7 +249,7 @@ def reader(env, **kwargs):
     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, process_stopped_at=125, **kwargs)
+            'original', failure_kind='unknown', process_started_at=100.25, process_stopped_at=120.25, **kwargs)
     if fault:
         with pytest.raises(ValueError):
             read()
@@ -240,6 +260,9 @@ def read():
         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':
@@ -254,7 +277,7 @@ def read():
 ] + [
     ('unknown', 1, fault) for fault in
     (None, 'stop', 'profile', 'session', 'source_digest', 'exhausted', 'unknown_allowance',
-     'stop_after_prepare', 'no_return', 'missing_error', 'native_error', 'pending_native',
+     '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):
@@ -270,11 +293,14 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path
     (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)
@@ -284,12 +310,13 @@ 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()
@@ -298,14 +325,21 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path
     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())
+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':now,'completedAt':now,
+    'error':native_error,'startedAt':int(started),'completedAt':int(completed),
     'items':[{'type':'commandExecution','status':'completed'}]}))
+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)
 ''')
@@ -356,6 +390,10 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path
     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':
@@ -405,6 +443,9 @@ def resume(**kwargs):
         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
@@ -433,6 +474,11 @@ def unexpected_observer(*args, **kwargs):
             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)
diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py
index b9967625cf..2110efa1b2 100644
--- a/loopx/hosts/codex_checkpoint.py
+++ b/loopx/hosts/codex_checkpoint.py
@@ -238,11 +238,12 @@ def native_failed_turn(env: dict[str, str], session_id: str, *, failure_kind: st
 
     The caller independently proves the exact Host returned and explicitly
     requests recovery after an operator repair. Non-quota failures also require
-    the exact returned error digest; it never classifies an unknown error from
-    its message. This observation starts no Turn and supplies no checkpoint.
+    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
 
     require_host_failure_kind(failure_kind)
     if ((returned_error_sha256 is None and failure_kind != "quota_exhausted") or not session_id
@@ -253,9 +254,9 @@ def native_failed_turn(env: dict[str, str], session_id: str, *, failure_kind: st
             or process_started_at > process_stopped_at or timeout <= 0):
         raise ValueError("Failed native recovery requires its returned failure and process window")
     deadline = time.monotonic() + timeout
-    original = (_native_observation_fingerprint(Path(env['CODEX_HOME']), deadline=deadline)
-                if returned_error_sha256 is not None else None)
-    observer = (_idle_native_reader(env, timeout=timeout, deadline=deadline) if original is not None
+    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):
@@ -264,7 +265,19 @@ def request(method, params):
                 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")
@@ -273,12 +286,14 @@ def request(method, params):
                             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")
+        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"}
@@ -290,10 +305,10 @@ def request(method, params):
             raise ValueError("Failed native history advanced during readback")
         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}
-                   if error_digest is not None else {}),
+                **({'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 original is not None and _native_observation_fingerprint(Path(env['CODEX_HOME']), deadline=deadline) != original:
+    if home is not None and _native_observation_fingerprint(home, deadline=deadline) != original:
         raise ValueError('Failed native source changed during isolated readback')
     return result
 
@@ -385,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")
@@ -393,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()}

From 224885d2f14251977471de19dc066dad638d10dd Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Wed, 11 Nov 2026 00:23:48 +0800
Subject: [PATCH 3/3] docs(benchmark): explain exact returned failure recovery

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/README.md | 19 +++++++++++--------
 1 file changed, 11 insertions(+), 8 deletions(-)

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