Recover exact returned native Host failures

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

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,17 +566,23 @@ 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()}
                if legacy else {'host_attempt_id': attempt}),
             'wake_turn_id': receipt['turn_id'], 'receipt_path': str(path),
             'receipt_sha256': hashlib.sha256(raw).hexdigest(),
             'source_phase_deadline_epoch': phase, 'session_profile_digest': current.digest,
+            **({'returned_failure': returned} if returned is not None else {}),
             'host_failure': failure, 'retry_failed': True, 'native': native, 'allowance': allowance}

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()