From 8bc878b29c8b3eda2c6b0ade4c0ed496bb4476b7 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 06:00:42 +0800
Subject: [PATCH 1/2] fix(benchmark): explicitly continue repaired quota
 failures on retained sessions

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
(cherry picked from commit d442384c8259c56bbe63709b0b76106db07b5e04)
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/README.md                           |  51 ++++
 benchmark/edgebench/stopped_recovery.py       |  32 +-
 benchmark/runtime/codex_readiness.py          |  38 ++-
 benchmark/runtime/interruption_recovery.py    |  88 ++++++
 benchmark/runtime/scheduler.py                |  48 ++-
 .../tests/test_failed_host_continuation.py    | 273 ++++++++++++++++++
 benchmark/tests/test_interruption_recovery.py |   7 +-
 .../control_plane/effect_runtime_handlers.ts  |   1 +
 .../control_plane/turn_driver/host_failure.py |  14 +
 .../control_plane/turn_driver/host_failure.ts |  29 ++
 .../control_plane/turn_driver/turn_journal.ts |  26 +-
 loopx/hosts/codex_checkpoint.py               |  48 +++
 scripts/external_scheduler_worker.py          |   2 +-
 .../effect_runtime_handlers.test.ts           |  18 ++
 14 files changed, 630 insertions(+), 45 deletions(-)
 create mode 100644 benchmark/tests/test_failed_host_continuation.py

diff --git a/benchmark/README.md b/benchmark/README.md
index bcc666d16c..2d78d9b708 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -1480,3 +1480,54 @@ belongs in this infrastructure handoff, never a replacement solution Todo.
 Recovery settles process transport only. Preserved pending semantic replan
 attempts still require their original settlement owner; an interrupted replan
 is not silently declared complete or cleared to force dispatch.
+
+### Explicitly continue after native account allowance is repaired
+
+A hard native quota failure is still held by default. Repair or replenish the
+authorized experiment account through its existing provider and verify its fresh
+allowance before requesting another phase. This does not change the desktop
+login, waive a Stop, refill evaluator opportunities, or turn hard quota failures
+into automatic timed retries.
+
+For a retained complete worker, the same scheduler entrypoint accepts
+`--retry-failed-state-sha256 SHA256` naming the exact original scheduler file.
+Supply the independently admitted finite remainder in
+`LOOPX_PHASE_DEADLINE_EPOCH`, preserving the original phase clock and failed
+receipt. The caller must retain the workspace, Goal/SQLite, session, evaluator
+history and cooldown; this entrypoint cannot reconstruct missing source data.
+For example, after these provider and source checks:
+
+```sh
+python -m benchmark.runtime.scheduler \
+  --python /runtime/bin/python --source /runtime/source \
+  --state-file /runtime/control/scheduler-state.json --host-timeout 600 \
+  --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 shared Turn retry rule decides explicit-retry eligibility. A new append-only
+phase records the original failure, digest, native cursor and deadline, retains
+the Host attempt count and semantic replan records, and starts through normal
+Goal/Stop/session/work admission. Repeating the request selects that same phase,
+including any later failure; it cannot clear the failure or extend its deadline.
+A later explicit repair must bind that new failure separately. Existing operator
+and account authority remain required; a digest is provenance, not permission.
+
+The explicit failed-phase path additionally uses the installed optional native
+Codex provider's existing availability policy (`loopx-codex-provider-routing`).
+It reads `account/read(refreshToken=false)` and `account/rateLimits/read` in a
+fresh disposable inspection home before creating the phase. Missing provider
+support, unknown allowance, exhausted allowance or a changed Goal/session keep
+the original failure held. Only a bounded availability receipt is retained;
+no account response, login values or token is emitted. The ordinary startup
+probe and automatic retry path keep their previous behavior. This observation
+is not a reservation of future provider capacity; a later failure is retained
+and requires another explicit repair.
diff --git a/benchmark/edgebench/stopped_recovery.py b/benchmark/edgebench/stopped_recovery.py
index 18c3d8593b..2fa5f49587 100644
--- a/benchmark/edgebench/stopped_recovery.py
+++ b/benchmark/edgebench/stopped_recovery.py
@@ -106,7 +106,9 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
                                        'files': file_inventory(member), 'directory': member.is_dir()})
         state_raw = copied['scheduler_state'].read_bytes()
         state = json.loads(state_raw)
-        if (state.get('host_attempt', {}).get('status') != 'started'
+        prior = state.get('host_attempt', {})
+        failed = prior.get('status') == 'failed'
+        if (prior.get('status') not in {'started', 'failed'}
                 or json.loads(copied['host_pending'].read_bytes()) != {'pending': True}):
             raise ValueError('Stopped legacy recovery requires the original unresolved Host attempt')
         wake = copied['wake_log_dir'] / declaration['wake_turn_id'] / 'receipt.json'
@@ -114,16 +116,38 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
         receipt = json.loads(raw)
         candidates = [p for p in copied['wake_log_dir'].glob('*/receipt.json')
                       if p.stat().st_mtime > copied['scheduler_state'].stat().st_mtime]
+        if failed:
+            from loopx.control_plane.turn_driver.host_failure import build_host_failure_record, normalize_host_failure_record
+
+            failure = normalize_host_failure_record(prior.get('host_failure'))
+            if (failure['kind'] != 'quota_exhausted' or prior.get('session_id') is None
+                    or failure['attempt'] != prior.get('attempt')
+                    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 normalize_host_failure_record(receipt.get('host_failure'))
+                        != build_host_failure_record(failure['kind'], attempt=1)
+                    or receipt.get('session', {}).get('session_id') != prior['session_id']
+                    or receipt.get('session', {}).get('binding_scope') != 'agent'
+                    or type(receipt.get('return_code')) is not int or receipt['return_code'] != 1
+                    or receipt.get('budget_exhausted')):
+                raise ValueError('Stopped source lacks its exact returned quota failure')
+            # A returned failure is committed after its receipt. Do not borrow
+            # an earlier or later wake, or treat an uncertain started attempt as returned.
+            all_receipts = list(copied['wake_log_dir'].glob('*/receipt.json'))
+            newest = max((p.stat().st_mtime for p in all_receipts), default=0)
+            if candidates or [p for p in all_receipts if p.stat().st_mtime == newest] != [wake]:
+                raise ValueError('Stopped failure is not the last returned wake')
+            candidates = [wake]
         if candidates != [wake]:
             raise ValueError('Stopped source has no unique returned wake after its Host attempt')
         if (receipt.get('turn_id') != declaration['wake_turn_id'] or receipt.get('host_invoked') is not True
-                or receipt.get('budget_exhausted') is not True or receipt.get('mode') != 'heartbeat'
+                or (not failed and receipt.get('budget_exhausted') is not True) or receipt.get('mode') != 'heartbeat'
                 or receipt.get('stage') != 'execute' or receipt.get('ok') is not False
-                or 'host_failure' in receipt):
+                or (not failed and 'host_failure' in receipt)):
             raise ValueError('Stopped source lacks an exact interrupted budget receipt')
         start = int(declaration['wake_turn_id'].split('-')[1]) / 1_000_000_000
         phase = float(copied['phase_clock'].read_text())
-        if not start <= phase <= finished:
+        if not math.isfinite(phase) or not start <= min(phase, finished) or (not failed and phase > finished):
             raise ValueError('Stopped source phase did not expire during the original wake')
         result = {'container_id': box.id, 'source_state': before, 'declaration': declaration,
                   'components': components, 'scheduler_state_sha256': hashlib.sha256(state_raw).hexdigest(),
diff --git a/benchmark/runtime/codex_readiness.py b/benchmark/runtime/codex_readiness.py
index 3d35830f41..b5e3e52619 100644
--- a/benchmark/runtime/codex_readiness.py
+++ b/benchmark/runtime/codex_readiness.py
@@ -64,9 +64,10 @@ def _remaining(deadline: float) -> float:
     return remaining
 
 
-def _observe_once(env: dict[str, str], deadline: float) -> dict:
+def _observe_once(env: dict[str, str], deadline: float, *, require_allowance: bool = False) -> dict:
     started = time.monotonic()
     stage = "inspection_home"
+    allowance = None
     try:
         _remaining(deadline)
         # Native app-server can install bundled skills during account-only
@@ -97,11 +98,30 @@ def request(method, params):
                 if not isinstance(usage, dict) or not usage:
                     raise _InvalidObservation("unavailable_rate_limits")
                 _remaining(deadline)
+                if require_allowance:
+                    # Reuse the optional native provider's typed availability
+                    # policy. A single observed receiver is not an account ring:
+                    # this read cannot switch login or enable API fallback.
+                    stage = "allowance"
+                    try:
+                        from loopx_codex_provider_routing.native import policy
+                    except ImportError:
+                        raise _InvalidObservation("allowance_provider_unavailable") from None
+                    observed_at = time.time()
+                    decision = policy({"now": observed_at,
+                        "policy": {"enabled": True, "slots": ["receiver"], "cooldown_seconds": 0},
+                        "active": "receiver", "observations": [{"slot": "receiver",
+                            "observed_at": observed_at, "status": "ok", "usage": usage}],
+                        "last_switch_at": None, "recovery_required": False, "login_matches": True})
+                    _remaining(deadline)
+                    if decision.get("action") != "ready":
+                        raise _InvalidObservation("allowance_unavailable")
+                    allowance = {"availability": "available", "observed_at": observed_at}
     except Exception as exc:
         error_kind, retryable = _failure(exc)
         return {"ok": False, "stage": stage, "error_kind": error_kind,
                 "retryable": retryable, "elapsed_seconds": round(time.monotonic() - started, 3)}
-    return {"ok": True, "stage": stage,
+    return {"ok": True, "stage": stage, **({"allowance": allowance} if allowance else {}),
             "elapsed_seconds": round(time.monotonic() - started, 3)}
 
 
@@ -132,6 +152,20 @@ def check_native_receiver(env: dict[str, str]) -> dict:
     return result
 
 
+def require_native_receiver_allowance(env: dict[str, str]) -> dict:
+    """Fresh, account-only evidence for an explicit failed-Host continuation.
+
+    The startup probe's default behavior stays unchanged. This stricter caller
+    refuses unknown/exhausted allowance and missing optional provider support;
+    it never refreshes login, switches accounts, or starts inference.
+    """
+    observation = _observe_once(env, time.monotonic() + 25, require_allowance=True)
+    if not observation["ok"]:
+        raise ValueError("Explicit Host continuation requires fresh native allowance: "
+                         + observation["stage"] + "/" + observation["error_kind"])
+    return observation["allowance"]
+
+
 def main() -> int:
     # Emit only bounded local stage/category/timing diagnostics. Never include
     # raw exception text, native stderr, login values or account/RPC payloads.
diff --git a/benchmark/runtime/interruption_recovery.py b/benchmark/runtime/interruption_recovery.py
index 2e01364649..f5cc59c1f6 100644
--- a/benchmark/runtime/interruption_recovery.py
+++ b/benchmark/runtime/interruption_recovery.py
@@ -232,3 +232,91 @@ def read_interrupted_host(state: dict, *, state_file: Path, env: dict[str, str],
             "receipt_path": str(path), "receipt_sha256": hashlib.sha256(raw).hexdigest(),
             "source_phase_deadline_epoch": phase, "session_profile_digest": current.digest,
             "native": native}
+
+
+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.
+
+    The digest binds a caller's fresh repair/allowance decision to this exact
+    failed attempt, not to every subsequent failure. The ordinary scheduler and
+    Turn launch still own current Goal, Stop, work and budget admission.
+    """
+    from loopx.control_plane.turn_driver.host_failure import (
+        build_host_failure_record, explicit_host_failure_retry_available, normalize_host_failure_record,
+    )
+    from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession
+    from loopx.hosts.codex_checkpoint import native_failed_turn
+    from .codex import process_environment
+    from .codex_readiness import require_native_receiver_allowance
+    from scripts.external_scheduler_worker import _host_pending
+
+    original_state = _physical_read(state_file)
+    if (env.get('LOOPX_EXECUTION_MODE') != 'heartbeat' or env.get('LOOPX_ITERATION_CONTEXT') != 'resume'
+            or not re.fullmatch(r'[a-f0-9]{64}', source_state_sha256)
+            or hashlib.sha256(original_state).hexdigest() != source_state_sha256
+            or _host_pending(state_file) is not True):
+        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'
+            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')
+    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
+    source = None
+    if legacy:
+        path, raw, receipt, source = _stopped_source(state_file, state, env, wake_log_dir)
+    else:
+        if not re.fullmatch(r'[0-9a-f]{32}', attempt):
+            raise ValueError('Failed Host attempt has no exact returned identity')
+        matches = []
+        for path in Path(wake_log_dir).absolute().glob('*/receipt.json'):
+            raw = _physical_read(path)
+            receipt = json.loads(raw)
+            if isinstance(receipt, dict) and receipt.get('host_attempt_id') == attempt:
+                matches.append((path, raw, receipt))
+        if len(matches) != 1:
+            raise ValueError('Failed Host must have one exact returned wake')
+        path, raw, receipt = matches[0]
+    if (receipt.get('turn_id') != path.parent.name or receipt.get('mode') != 'heartbeat'
+            or receipt.get('stage') != 'execute' or receipt.get('ok') is not False
+            or receipt.get('host_invoked') is not True or receipt.get('budget_exhausted')
+            or type(receipt.get('return_code')) is not int or receipt['return_code'] != 1
+            or normalize_host_failure_record(receipt.get('host_failure'))
+                != build_host_failure_record(failure['kind'], attempt=1)
+            or (not legacy and any(receipt.get(key) != env['LOOPX_' + key.upper()]
+                                   for key in ('goal_id', 'agent_id')))):
+        raise ValueError('Failed Host receipt differs from this lane and failure')
+    started, stopped = ((_number(source['process_started_at']), _number(source['process_stopped_at']))
+                        if legacy else _process_window(receipt))
+    phase = _number(float(source['source_phase_deadline_epoch'] if legacy else receipt['phase_deadline_epoch']))
+    if phase < started:
+        raise ValueError('Original failure phase predates its Host invocation')
+    session = receipt.get('session') or {}
+    current = CodexAgentSession(env, context='resume',
+        sandbox=env.get('LOOPX_CODEX_SANDBOX', 'danger-full-access'), receipt={})
+    with current.admission.launch_lifetime(operation='benchmark_failed_host_recovery'):
+        pass
+    if (session.get('binding_scope') != 'agent' or not current.binding
+            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')
+    native = native_failed_turn(env, current.session_id, failure_kind=failure['kind'],
+                               process_started_at=started, process_stopped_at=stopped)
+    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:
+        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,
+            'host_failure': failure, 'retry_failed': True, 'native': native, 'allowance': allowance}
diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py
index 235d4152a8..cc6e0e4a90 100644
--- a/benchmark/runtime/scheduler.py
+++ b/benchmark/runtime/scheduler.py
@@ -91,16 +91,19 @@ def _phase_receipt(state_file: Path, wake_log_dir: str | None) -> dict | None:
 
 
 def _create_resume_state(target: Path, predecessor: Path, *, recovered: dict | None = None,
-                         replan_attempts=None, handoff: dict | None = None) -> None:
+                         replan_attempts=None, handoff: dict | None = None,
+                         failed_recovery: bool = False) -> None:
     if target.is_symlink():
         raise ValueError("Resumed scheduler state must not be a symlink")
     try:
         with os.fdopen(os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "w", encoding="utf-8") as stream:
             value = {"resume_from": str(predecessor)}
             if recovered is not None:
-                value.update(interruption_recovery=recovered,
-                    recovered_session_id=recovered["native"]["session_id"],
+                value["failed_host_recovery" if failed_recovery else "interruption_recovery"] = recovered
+                value.update(recovered_session_id=recovered["native"]["session_id"],
                     replan_attempts=replan_attempts or [])
+                if failed_recovery:
+                    value['recovered_attempt_count'] = recovered['host_failure']['attempt']
             if handoff is not None:
                 value["recovery_handoff"] = handoff
             json.dump(value, stream)
@@ -111,7 +114,8 @@ def _create_resume_state(target: Path, predecessor: Path, *, recovered: dict | N
 
 
 def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str,
-                      wake_log_dir: str | None = None, env: dict[str, str] | None = None) -> str:
+                      wake_log_dir: str | None = None, env: dict[str, str] | None = None,
+                      retry_failed_state_sha256: str | None = None) -> str:
     """Select a new transport phase only after an explicit budget continuation.
 
     Kernel state and native sessions stay in place. The original scheduler
@@ -121,6 +125,8 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str,
     if (isinstance(phase_deadline, bool) or not isinstance(phase_deadline, (int, float))
             or not math.isfinite(phase_deadline) or phase_deadline <= time.time()):
         raise ValueError("Budget continuation requires an open finite phase deadline")
+    if retry_failed_state_sha256 is not None and not re.fullmatch(r'[a-f0-9]{64}', retry_failed_state_sha256):
+        raise ValueError('Explicit failed continuation requires its exact source-state digest')
     from .interruption_recovery import recovery_handoff
     handoff, _ = recovery_handoff(env or {})
     path = _physical_state(Path(state_file).absolute())
@@ -147,19 +153,26 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str,
         if state is None:
             return str(predecessor)
         prior = state.get("host_attempt")
-        recovery = state.get("interruption_recovery")
+        recovery = state.get("interruption_recovery") or state.get('failed_host_recovery')
         can_continue_stopped_phase = (isinstance(prior, dict) and prior.get("status") == "stopped"
                                       and prior.get("stop_exit_code") == BUDGET_EXHAUSTED_EXIT_CODE)
         requested_suffix = hashlib.sha256(float(phase_deadline).hex().encode()).hexdigest()[:20]
         same_phase = predecessor.name == f"{path.stem}.phase-{requested_suffix}{path.suffix}"
+        explicit_failed_phase = (retry_failed_state_sha256 is not None and not same_phase
+            and hashlib.sha256(predecessor.read_bytes()).hexdigest() == retry_failed_state_sha256)
         if ((recovery is not None or state.get("resume_from"))
                 and state.get("recovery_handoff") != handoff
-                and (same_phase or not can_continue_stopped_phase)):
+                and (same_phase or not (can_continue_stopped_phase or explicit_failed_phase))):
             raise ValueError("Recovered phase handoff is immutable; preserve its exact input")
         if (recovery is not None and recovery.get("phase_deadline_epoch") != phase_deadline
-                and not (isinstance(prior, dict) and prior.get("status") == "stopped"
-                         and prior.get("stop_exit_code") == BUDGET_EXHAUSTED_EXIT_CODE)):
+                and not (can_continue_stopped_phase or explicit_failed_phase)):
             raise ValueError("Recovered phase deadline is immutable; inspect the current phase before continuing")
+        if retry_failed_state_sha256 is not None:
+            if (same_phase and state.get('failed_host_recovery', {}).get('source_state_sha256')
+                    == retry_failed_state_sha256):
+                return str(predecessor)  # Replay never erases a later failure in this phase.
+            if not explicit_failed_phase:
+                raise ValueError('Explicit failed continuation source changed; inspect the retained phase')
         if not isinstance(prior, dict):
             return str(predecessor)
         # Validate the captured record with its existing transport owner.
@@ -174,7 +187,15 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str,
         # Explicit continuation only reconciles transport; all Goal/Stop,
         # replan, lease and launch decisions remain with the normal next wake.
         recovered = None
-        if env is not None and wake_log_dir and _host_pending(predecessor) is not None:
+        failed_recovery = False
+        if retry_failed_state_sha256 is not None:
+            if env is None or wake_log_dir is None:
+                raise ValueError('Explicit failed continuation requires original native and wake routes')
+            from .interruption_recovery import read_failed_host
+            recovered = read_failed_host(state, state_file=predecessor, env=env,
+                wake_log_dir=wake_log_dir, source_state_sha256=retry_failed_state_sha256)
+            failed_recovery = True
+        if recovered is None and env is not None and wake_log_dir and _host_pending(predecessor) is not None:
             from .interruption_recovery import read_interrupted_host
 
             try:
@@ -182,12 +203,15 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str,
             except (OSError, ValueError, RuntimeError, TimeoutError, KeyError, TypeError) as error:
                 print(f"Interrupted Host remains held: {type(error).__name__}: {error}", file=sys.stderr)
         if recovered is not None:
+            if failed_recovery and phase_deadline <= time.time():
+                raise ValueError('Explicit failed continuation exhausted its phase during qualification')
             suffix = hashlib.sha256(float(phase_deadline).hex().encode()).hexdigest()[:20]
             target = path.with_name(f"{path.stem}.phase-{suffix}{path.suffix}")
             recovered.update(source_state_sha256=hashlib.sha256(predecessor.read_bytes()).hexdigest(),
                              phase_deadline_epoch=phase_deadline)
             _create_resume_state(target, predecessor, recovered=recovered,
-                                 replan_attempts=state.get("replan_attempts"), handoff=handoff)
+                                 replan_attempts=state.get("replan_attempts"), handoff=handoff,
+                                 failed_recovery=failed_recovery)
             return str(target)
         if (prior.get("status") != "stopped"
                 or prior.get("stop_exit_code") != BUDGET_EXHAUSTED_EXIT_CODE
@@ -248,11 +272,13 @@ def main() -> int:
     parser.add_argument("--source", required=True)
     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')
     args = parser.parse_args()
     state = resume_state_file(args.state_file,
         phase_deadline=float(os.environ["LOOPX_PHASE_DEADLINE_EPOCH"]),
         agent_id=os.environ["LOOPX_AGENT_ID"], wake_log_dir=os.environ.get("LOOPX_WAKE_LOG_DIR"),
-        env=dict(os.environ))
+        env=dict(os.environ), retry_failed_state_sha256=args.retry_failed_state_sha256)
     command = worker_command(dict(os.environ), python=args.python, source=args.source,
                              state_file=state, host_timeout=args.host_timeout)
     os.execvpe(command[0], command, os.environ)
diff --git a/benchmark/tests/test_failed_host_continuation.py b/benchmark/tests/test_failed_host_continuation.py
new file mode 100644
index 0000000000..0fd79d20bc
--- /dev/null
+++ b/benchmark/tests/test_failed_host_continuation.py
@@ -0,0 +1,273 @@
+"""Explicit recovery preserves a failed Host and its native session/finite phase."""
+from __future__ import annotations
+
+from contextlib import contextmanager
+import copy
+import hashlib
+import json
+import os
+from pathlib import Path
+import shlex
+import sys
+import time
+
+import pytest
+
+from benchmark.runtime.scheduler import resume_state_file
+from loopx.control_plane.turn_driver.host_failure import (
+    build_host_failure_record, explicit_host_failure_retry_available, host_failure_retry_available,
+)
+from loopx.hosts import codex_checkpoint as native
+
+
+@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),
+    ('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):
+    failure = build_host_failure_record(kind, attempt=attempt)
+    original = copy.deepcopy(failure)
+    assert explicit_host_failure_retry_available(failure, retry_failed=explicit) is allowed
+    if kind == 'quota_exhausted':
+        assert host_failure_retry_available(failure) is False
+    assert failure == original
+
+
+@pytest.mark.parametrize('fault', [None, 'in_progress', 'pending_effect', 'wrong_session',
+    'wrong_workspace', 'wrong_error', 'prose_only', 'changed_history', 'outside_window', 'empty'])
+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'},
+        items=[{'type': 'commandExecution', 'status': 'completed'}])]
+    if fault == 'in_progress': turns[-1]['status'] = 'inProgress'
+    elif fault == 'pending_effect': turns[-1]['items'][0]['status'] = 'inProgress'
+    elif fault == 'wrong_error': turns[-1]['error']['codexErrorInfo'] = 'unknownFutureCode'
+    elif fault == 'prose_only': turns[-1]['error'] = {'message': 'usageLimitExceeded'}
+    elif fault == 'outside_window': turns[-1]['startedAt'] = 90
+    elif fault == 'empty': turns = []
+    calls = []
+    class Transport:
+        def request(self, method, params):
+            calls.append(method)
+            if method == 'thread/read':
+                return {'thread': {'id': 'different' if fault == 'wrong_session' else 'original',
+                    'cwd': '/different' if fault == 'wrong_workspace' else 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': 'new'}]
+            return {'data': rows, 'nextCursor': None}
+    @contextmanager
+    def reader(*a, **kw): yield Transport()
+    monkeypatch.setattr(native, 'native_checkpoint_reader', reader)
+    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:
+        with pytest.raises(ValueError): read()
+    else:
+        result = read()
+        assert result['status'] == 'failed' and result['turn_count'] == 2
+        assert result['failure_kind'] == 'quota_exhausted' and result['session_id'] == 'original'
+        with pytest.raises(ValueError, match='completed final cursor'):
+            native._complete_history(Transport().request, '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):
+    from benchmark.runtime import worker
+    from benchmark.tests.test_worker_phase_budget import _phase_worker
+    from tests import test_external_scheduler_worker as scheduler_fixture
+
+    env, project, runtime, _source, goal_state, _clock = _phase_worker(tmp_path, monkeypatch)
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / 'packages/loopx-codex-provider-routing/src'))
+    monkeypatch.setattr(worker, 'time', time)
+    Path(env['CODEX_BIN']).write_text(f'#!{sys.executable}\n' + '''
+import json,os,pathlib,sys,time
+history=pathlib.Path('native-history.json')
+if 'app-server' in sys.argv:
+    for line in sys.stdin:
+        message=json.loads(line)
+        if 'id' not in message: continue
+        method=message['method']
+        if method=='initialize': result={}
+        elif method=='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/turns/list': result={'data':[json.loads(history.read_text())],'nextCursor':None}
+        else: raise AssertionError(method)
+        print(json.dumps({'jsonrpc':'2.0','id':message['id'],'result':result}),flush=True)
+    sys.exit(0)
+with pathlib.Path('native-argv').open('a') as f: f.write(json.dumps(sys.argv[1:])+'\\n')
+print(json.dumps({'type':'thread.started','thread_id':'original-native'}),flush=True)
+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,
+    'items':[{'type':'commandExecution','status':'completed'}]}))
+print(json.dumps({'type':'turn.failed','error':{'codexErrorInfo':'usageLimitExceeded'}}),flush=True)
+sys.exit(1)
+''')
+    for key, value in env.items(): monkeypatch.setenv(key, value)
+    deadline = time.time() + 120
+    monkeypatch.setenv('LOOPX_PHASE_DEADLINE_EPOCH', str(deadline))
+    args, _ = scheduler_fixture._guarded_host(tmp_path / 'scheduler')
+    args.once = True
+    args.registry, args.runtime_root = env['LOOPX_REGISTRY'], env['LOOPX_RUNTIME_ROOT']
+    args.goal_id, args.agent_id = env['LOOPX_GOAL_ID'], env['LOOPX_AGENT_ID']
+    args.wake_cmd = shlex.join([sys.executable, '-m', 'benchmark.runtime.worker'])
+    args.wake_stop_exit_code, args.wake_timeout_seconds = 75, 90
+    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']['status'] == 'failed'
+    assert scheduler_fixture.worker.run_worker(args) == 2  # Still no automatic retry.
+    assert len((project / 'native-argv').read_text().splitlines()) == 1
+    receipts = list(Path(env['LOOPX_WAKE_LOG_DIR']).glob('*/receipt.json'))
+    assert len(receipts) == 1
+    receipt = json.loads(receipts[0].read_bytes())
+    assert receipt['native_process']['after_cleanup']['return_code'] == 1
+    digest = hashlib.sha256(original_bytes).hexdigest()
+    if fault == 'stop':
+        from loopx.control_plane.goals.activation_service import set_goal_activation_state
+        assert set_goal_activation_state(registry_path=Path(args.registry), goal_id=args.goal_id,
+            state='stopped', reason='Owner stopped work', actor_kind='owner', execute=True)['ok']
+    elif fault == 'profile': monkeypatch.setenv('MODEL_NAME', 'other-model')
+    elif fault == 'session':
+        receipt['session']['session_id'] = 'different-native'
+        receipts[0].write_text(json.dumps(receipt))
+    elif fault == 'source_digest': digest = 'a' * 64
+    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}}}))
+    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),
+            retry_failed_state_sha256=digest, **kwargs)
+    if fault and fault != 'stop_after_prepare':
+        with pytest.raises((ValueError, RuntimeError)): resume()
+        assert original.read_bytes() == original_bytes
+        assert not list(original.parent.glob('*.phase-*.json'))
+        return
+    goal_before = goal_state.read_bytes()
+    phase = Path(resume())
+    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['failed_host_recovery']['native']['status'] == 'failed'
+    assert phase_initial['failed_host_recovery']['phase_deadline_epoch'] == deadline
+    assert resume() == str(phase)
+    if fault == 'stop_after_prepare':
+        from loopx.control_plane.goals.activation_service import set_goal_activation_state
+        assert set_goal_activation_state(registry_path=Path(args.registry), goal_id=args.goal_id,
+            state='stopped', reason='Owner stopped before dispatch', actor_kind='owner', execute=True)['ok']
+        args.state_file = str(phase)
+        scheduler_fixture.worker.run_worker(args)
+        assert len((project / 'native-argv').read_text().splitlines()) == 1
+        assert original.read_bytes() == original_bytes
+        return
+    # Repeating the unrepaired failure stays held in this same phase and cannot
+    # consume the original explicit recovery a second time.
+    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 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.
+    (project / 'allowance-repaired').touch()
+    next_deadline = deadline + 30
+    next_phase = resume_state_file(str(original), phase_deadline=next_deadline, agent_id=args.agent_id,
+        wake_log_dir=env['LOOPX_WAKE_LOG_DIR'], env=dict(os.environ),
+        retry_failed_state_sha256=hashlib.sha256(failed_again).hexdigest())
+    monkeypatch.setenv('LOOPX_PHASE_DEADLINE_EPOCH', str(next_deadline))
+    args.state_file = next_phase
+    assert scheduler_fixture.worker.run_worker(args) == 0
+    argv = [json.loads(line) for line in (project / 'native-argv').read_text().splitlines()]
+    assert len(argv) == 3 and all('resume' in a and 'original-native' in a for a in argv[1:])
+    assert (project / 'continued-work').is_file()
+    assert original.read_bytes() == original_bytes and phase.read_bytes() == failed_again
+    assert goal_state.read_bytes() == goal_before
+
+
+@pytest.mark.parametrize('fault', [None, 'unknown', 'missing_session', 'other_receipt', 'late_wake', 'live', 'wrong_container'])
+def test_legacy_quota_failure_proof_retains_complete_same_container_source(tmp_path, monkeypatch, fault):
+    from benchmark.tests.test_stopped_recovery import _source, _fake_container
+    from benchmark.edgebench.stopped_recovery import prepare_stopped_recovery
+    from benchmark.runtime.interruption_recovery import _stopped_source
+    declaration = _source(tmp_path)
+    box, client = _fake_container(monkeypatch)
+    state_path = Path(declaration['scheduler_state'])
+    failure = build_host_failure_record('unknown' if fault == 'unknown' else 'quota_exhausted', attempt=2)
+    state = {'host_attempt': dict(status='failed', scope='a' * 64, attempt=2,
+        host_failure=failure, session_id=None if fault == 'missing_session' else 'original', not_before=111),
+        'last_wake_status': 'wake_failed', 'last_wake_rc': 1}
+    state_path.write_text(json.dumps(state))
+    os.utime(state_path, (111, 111))
+    receipt_path = Path(declaration['wake_log_dir']) / declaration['wake_turn_id'] / 'receipt.json'
+    receipt = json.loads(receipt_path.read_bytes())
+    receipt.pop('budget_exhausted')
+    receipt.update(return_code=1, host_failure=build_host_failure_record(failure['kind'], attempt=1))
+    if fault == 'other_receipt': receipt['session']['session_id'] = 'different'
+    receipt_path.write_text(json.dumps(receipt));os.utime(receipt_path, (110, 110))
+    if fault == 'late_wake':
+        late = receipt_path.parent.with_name('later') / 'receipt.json'
+        late.parent.mkdir();late.write_text('{}');os.utime(late, (112, 112))
+    if fault == 'live': box.attrs['State'].update(Status='running', Running=True, Pid=44)
+    before = state_path.read_bytes()
+    output = tmp_path / 'quota-failure-proof.json'
+    if fault not in (None, 'wrong_container'):
+        with pytest.raises(ValueError):
+            prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=output)
+        assert not output.exists()
+        return
+    proof = prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=output)
+    assert proof['execution_authority_granted'] is False
+    env = {'LOOPX_INTERRUPTED_SOURCE': str(output), 'LOOPX_GOAL_ID': declaration['goal_id'],
+        'LOOPX_AGENT_ID': declaration['agent_id'], 'LOOPX_PROJECT': declaration['workspace'],
+        'LOOPX_RUNTIME_ROOT': declaration['runtime_root'], 'CODEX_HOME': declaration['native_home'],
+        'LOOPX_REGISTRY': declaration['registry']}
+    if fault == 'wrong_container':
+        from types import SimpleNamespace
+        monkeypatch.setattr(os, 'uname', lambda: SimpleNamespace(nodename='b' * 12))
+        with pytest.raises(ValueError, match='original container'):
+            _stopped_source(state_path, state, env, declaration['wake_log_dir'])
+    else:
+        assert _stopped_source(state_path, state, env, declaration['wake_log_dir'])[2]['host_failure']['kind'] == 'quota_exhausted'
+    assert state_path.read_bytes() == before
+
+
+@pytest.mark.parametrize('available', [False, True])
+def test_strict_allowance_uses_isolated_account_only_owner(tmp_path, monkeypatch, available):
+    from benchmark.tests.test_codex_readiness import native_fixture
+    from benchmark.runtime.codex_readiness import require_native_receiver_allowance
+    env, trace, home, pending = native_fixture(tmp_path, 'success')
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2] / 'packages/loopx-codex-provider-routing/src'))
+    before = {p: p.read_bytes() for p in (home / 'auth.json', pending)}
+    if not available:
+        # Deliberately hide the optional provider, without a substitute policy.
+        monkeypatch.setitem(sys.modules, 'loopx_codex_provider_routing.native', None)
+        with pytest.raises(ValueError, match='allowance_provider_unavailable'):
+            require_native_receiver_allowance(env)
+    else:
+        observed = require_native_receiver_allowance(env)
+        assert observed['availability'] == 'available'
+        assert set(observed) == {'availability', 'observed_at'}
+    assert {p: p.read_bytes() for p in before} == before
+    methods = [json.loads(line)['method'] for line in trace.read_text().splitlines()]
+    assert methods == ['initialize', 'initialized', 'account/read', 'account/rateLimits/read']
+    assert not (home / 'skills').exists()
diff --git a/benchmark/tests/test_interruption_recovery.py b/benchmark/tests/test_interruption_recovery.py
index 451f5ea341..7d5ebc6f59 100644
--- a/benchmark/tests/test_interruption_recovery.py
+++ b/benchmark/tests/test_interruption_recovery.py
@@ -217,7 +217,8 @@ def test_recovery_handoff_missing_changed_and_default_parity(tmp_path):
 
 
 @pytest.mark.parametrize('status', ['started', 'unknown', 'returned'])
-def test_new_transport_phase_does_not_clear_unsettled_semantic_replan(tmp_path, monkeypatch, status):
+@pytest.mark.parametrize("failed_recovery", [False, True])
+def test_new_transport_phase_does_not_clear_unsettled_semantic_replan(tmp_path, monkeypatch, status, failed_recovery):
     from tests.test_external_scheduler_worker import _guarded_host, worker
     from tests.test_external_scheduler_replan_retry import _replan
     from benchmark.runtime.scheduler import _create_resume_state
@@ -230,8 +231,8 @@ def test_new_transport_phase_does_not_clear_unsettled_semantic_replan(tmp_path,
     attempts = state['replan_attempts']
     attempts[0]['status'] = status
     target = tmp_path / 'continued.json'
-    _create_resume_state(target, old, recovered={'native': {'session_id': 'same-session'}},
-                         replan_attempts=attempts)
+    _create_resume_state(target, old, recovered={'native': {'session_id': 'same-session'},
+        'host_failure': {'attempt': 1}}, replan_attempts=attempts, failed_recovery=failed_recovery)
     args.state_file = str(target)
     assert worker.run_worker(args, clock=lambda: 1031.0) in (0, 2)
     assert count.read_text() == '1'
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index af454f6393..cef5507e88 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -265,6 +265,7 @@ export function createEffectRuntimeHandlers(
       "turn_journal.inspect",
       lazyHandler(() => import("./turn_driver/turn_journal.ts"), ({interpretTurnJournal}) => (params) => interpretTurnJournal(turnJournalInspectionRequest(params))),
     ],
+    ["host_failure.retry", lazyHandler(() => import("./turn_driver/host_failure.ts"), ({hostFailureRetryRequest}) => hostFailureRetryRequest)],
     ["turn.controller", lazyHandler(() => import("./turn_driver/loop_controller.ts"), ({controllerRequest}) => controllerRequest)],
     ["turn_journal.write", lazyHandler(() => import("./turn_driver/turn_journal_effects.ts"), ({commitTurnJournal}) => commitTurnJournal)],
     ["turn_journal.find_settlement", lazyHandler(() => import("./turn_driver/turn_journal_query.ts"), ({findTurnJournalBySettlement}) => findTurnJournalBySettlement)],
diff --git a/loopx/control_plane/turn_driver/host_failure.py b/loopx/control_plane/turn_driver/host_failure.py
index ba9d7cba79..f7bfafa65a 100644
--- a/loopx/control_plane/turn_driver/host_failure.py
+++ b/loopx/control_plane/turn_driver/host_failure.py
@@ -127,6 +127,20 @@ def host_failure_retry_available(record: Mapping[str, Any]) -> bool:
     return int(normalized["attempt"]) < int(retry["max_attempts"])
 
 
+def explicit_host_failure_retry_available(record: Mapping[str, Any], *, retry_failed: bool) -> bool:
+    """Share the Turn Journal's explicit repair rule; no automatic retry change."""
+    from ..effect_runtime import effect_runtime_result
+
+    if type(retry_failed) is not bool:
+        raise ValueError("Host retry requires an explicit retry intent")
+    result = effect_runtime_result("host_failure.retry", {
+        "failure": normalize_host_failure_record(record), "retry_failed": retry_failed,
+    })
+    if not isinstance(result, dict) or type(result.get("allowed")) is not bool:
+        raise RuntimeError("Host retry decision is unavailable")
+    return result["allowed"]
+
+
 def record_host_failure(
     journal: dict[str, Any],
     *,
diff --git a/loopx/control_plane/turn_driver/host_failure.ts b/loopx/control_plane/turn_driver/host_failure.ts
index ff0d6debd2..fa3c3febff 100644
--- a/loopx/control_plane/turn_driver/host_failure.ts
+++ b/loopx/control_plane/turn_driver/host_failure.ts
@@ -45,3 +45,32 @@ export function normalizeHostFailure(value: unknown, integerAttempt?: string | n
     fail("host failure record contains unsupported fields");
   return expected;
 }
+
+/** The same retry rule for an explicit Turn retry and an outer Host phase.
+ * This decides failure eligibility only, never session, budget or launch authority.
+ */
+export function hostFailureRetryCheck(failure: JsonObject, retryFailed: boolean): JsonObject | null {
+  // Both callers normalize with their existing transport owner first. Journal
+  // decoding may retain a large integral attempt as its exact decimal string.
+  if (!retryFailed) return {kind: "host_retry_policy", outcome: "failed",
+    reason: "failed_retry_not_requested"};
+  if (failure.retryable === true) {
+    const retry = failure.retry as JsonObject;
+    return Number(failure.attempt) >= Number(retry.max_attempts)
+      ? {kind: "host_retry_policy", outcome: "failed", reason: "host_retry_budget_exhausted"}
+      : {kind: "host_retry_policy", outcome: "passed"};
+  }
+  // Preserve the Journal's explicit-repair behavior. Output exhaustion has no
+  // proved whole-Turn remainder, so even an explicit blind replay is refused.
+  return failure.kind === "output_budget_exhausted"
+    ? {kind: "host_retry_policy", outcome: "failed", reason: "host_retry_not_available"}
+    : null;
+}
+
+export function hostFailureRetryRequest(params: JsonObject): JsonObject {
+  if (typeof params.retry_failed !== "boolean" || Object.keys(params).some(
+    key => !["failure", "retry_failed"].includes(key)))
+    throw new EffectRuntimeRequestError("Host retry requires an explicit retry intent");
+  const check = hostFailureRetryCheck(normalizeHostFailure(params.failure), params.retry_failed);
+  return {allowed: check === null || check.outcome === "passed", check};
+}
diff --git a/loopx/control_plane/turn_driver/turn_journal.ts b/loopx/control_plane/turn_driver/turn_journal.ts
index 91e93362a4..ff8c708856 100644
--- a/loopx/control_plane/turn_driver/turn_journal.ts
+++ b/loopx/control_plane/turn_driver/turn_journal.ts
@@ -1,4 +1,4 @@
-import { normalizeHostFailure } from "./host_failure.ts";
+import { normalizeHostFailure, hostFailureRetryCheck } from "./host_failure.ts";
 import transactionContract from "../turn_transaction_contract.json" with {
   type: "json",
 };
@@ -326,32 +326,10 @@ function hostRetryPolicyCheck(journal: JsonObject): TurnRecoveryCheck | null {
     if (!(error instanceof EffectRuntimeRequestError)) throw error;
     return {kind: "host_retry_policy", outcome: "failed", reason: "host_retry_contract_invalid"};
   }
-  const retry = asObject(failure.retry);
   const attempt = failure.attempt;
-  const kind = String(failure.kind);
-  const retryable = failure.retryable;
   if (journal.host_attempt_count !== attempt)
     return {kind: "host_retry_policy", outcome: "failed", reason: "host_retry_contract_invalid"};
-  if (retryable === true && Number(attempt) >= Number(retry.max_attempts)) {
-    return {
-      kind: "host_retry_policy",
-      outcome: "failed",
-      reason: "host_retry_budget_exhausted",
-    };
-  }
-  if (retryable === true) {
-    return { kind: "host_retry_policy", outcome: "passed" };
-  }
-  // A max-token terminal has no proved whole-Turn remaining budget or
-  // final-response reserve. Unlike legacy explicitly retried terminal errors,
-  // repeating it would be a blind rerun of an expensive request.
-  return kind === "output_budget_exhausted"
-    ? {
-        kind: "host_retry_policy",
-        outcome: "failed",
-        reason: "host_retry_not_available",
-      }
-    : null;
+  return hostFailureRetryCheck(failure, true) as TurnRecoveryCheck | null;
 }
 
 function recoveryDecision(
diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py
index af34afe824..bad8caeb7b 100644
--- a/loopx/hosts/codex_checkpoint.py
+++ b/loopx/hosts/codex_checkpoint.py
@@ -104,6 +104,54 @@ def request(method, params):
                 "turn_count": len(turns)}
 
 
+def native_failed_turn(env: dict[str, str], session_id: str, *, failure_kind: str,
+                       process_started_at: float, process_stopped_at: float,
+                       timeout: float = 30) -> dict[str, Any]:
+    """Read a retained quota-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.
+    """
+    from loopx.control_plane.turn_driver.codex_cli import _codex_error_info_category
+
+    if (failure_kind != "quota_exhausted" or not session_id
+            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")
+    deadline = time.monotonic() + timeout
+    with native_checkpoint_reader(env, timeout=timeout, deadline=deadline) 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)
+        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)
+        start, end = tail.get("startedAt"), tail.get("completedAt")
+        if (tail.get("status") != "failed" or 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")
+        if any(item.get("status") in {"inProgress", "running", "pending"}
+               for item in tail.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",
+                "failure_kind": failure_kind, "turn_count": len(turns),
+                "visible_history_sha256": hashlib.sha256(encoded).hexdigest()}
+
+
 def _guard_cli_arguments(command: str, guard: dict) -> list[str]:
     """Parse the provider shell wrapper plus the renderer's optional Turn assignment."""
     outer = shlex.split(command)
diff --git a/scripts/external_scheduler_worker.py b/scripts/external_scheduler_worker.py
index 739449a359..588e067f74 100644
--- a/scripts/external_scheduler_worker.py
+++ b/scripts/external_scheduler_worker.py
@@ -923,7 +923,7 @@ def save(fields: dict[str, Any]) -> None:
                         _save_host_pending(state_path, True)
                         state["host_attempt"] = {"scope": scope, "status": "started",
                             "attempt_id": uuid.uuid4().hex,
-                            "attempt": prior.get("attempt", 0) + 1,
+                            "attempt": prior.get("attempt", state.get("recovered_attempt_count", 0)) + 1,
                             "session_id": prior.get("session_id") or state.get("recovered_session_id"),
                             **({"replan_obligation_id": attempt["obligation_id"]}
                                if attempt is not None else {})}
diff --git a/tests/control_plane_ts/effect_runtime_handlers.test.ts b/tests/control_plane_ts/effect_runtime_handlers.test.ts
index a7a1318c10..1dc0733f2e 100644
--- a/tests/control_plane_ts/effect_runtime_handlers.test.ts
+++ b/tests/control_plane_ts/effect_runtime_handlers.test.ts
@@ -254,3 +254,21 @@ for (const retired of [
       {code: "unsupported_method"});
   });
 }
+
+
+test("explicit Host retry is registered and requires a valid typed failure and intent", async () => {
+  const failure = {schema_version: "loopx_turn_host_failure_v0", kind: "quota_exhausted",
+    attempt: 1, retryable: false};
+  assert.deepEqual(await dispatchEffectRuntimeMethod(handlers, "host_failure.retry",
+    {failure, retry_failed: true}), {allowed: true, check: null});
+  const held = await dispatchEffectRuntimeMethod(handlers, "host_failure.retry",
+    {failure, retry_failed: false}) as Record<string, unknown>;
+  assert.equal(held.allowed, false);
+  for (const params of [{failure}, {failure, retry_failed: "true"},
+    {failure, retry_failed: true, extra: true}, {retry_failed: true},
+    {failure: {...failure, kind: "not_registered"}, retry_failed: true},
+    {failure: {...failure, attempt: true}, retry_failed: true},
+    {failure: {...failure, retryable: true}, retry_failed: true}]) {
+    await assert.rejects(dispatchEffectRuntimeMethod(handlers, "host_failure.retry", params));
+  }
+});

From 3e49d8832e0615947b4150cd1f8b4cc3163b0eae Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 07:00:56 +0800
Subject: [PATCH 2/2] fix(benchmark): qualify returned legacy budget
 interruptions

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
(cherry picked from commit b7f28a6096f6c3b76606776b2b8371b752b1eab5)
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/README.md                           | 12 ++++
 benchmark/edgebench/stopped_recovery.py       | 41 +++++++++--
 benchmark/runtime/interruption_recovery.py    |  7 +-
 benchmark/tests/test_interruption_recovery.py | 49 ++++++++++++-
 benchmark/tests/test_stopped_recovery.py      | 72 ++++++++++++++++++-
 5 files changed, 171 insertions(+), 10 deletions(-)

diff --git a/benchmark/README.md b/benchmark/README.md
index 2d78d9b708..9a00944151 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -1438,6 +1438,18 @@ inert proof in that same container's control directory; omitting it is read-only
 Neither form starts the container, restores scoring authority or calls a model.
 Keep exclusive ownership of the source during preparation and continuation.
 
+The legacy source may still say `started`, or may already record a returned
+`stopped` attempt with exit code 75. The latter requires the unique latest wake
+receipt at or before that scheduler snapshot, with `host_invoked=true`,
+`budget_exhausted=true` and `timed_out=true`, no failure, and the original agent
+session. It is not the pre-dispatch budget shortcut. Preparation preserves the
+returned stop rather than changing it to `started` or inventing an attempt ID;
+the receiver must still verify the original native interrupted tail and current
+Goal/session admission. A later or ambiguous wake keeps the source held.
+For a returned stop, the native window ends at the receipt's recorded second,
+not a later maintenance stop of the container. The original hard deadline stays
+intact even when an older worker reserved cleanup time before it.
+
 After independently restoring the original native evaluator's history,
 opportunities and actual cooldown, restart only the same inert container and
 pass its returned `installed_proof_path` as `LOOPX_INTERRUPTED_SOURCE` to the
diff --git a/benchmark/edgebench/stopped_recovery.py b/benchmark/edgebench/stopped_recovery.py
index 2fa5f49587..779fe299d4 100644
--- a/benchmark/edgebench/stopped_recovery.py
+++ b/benchmark/edgebench/stopped_recovery.py
@@ -108,7 +108,8 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
         state = json.loads(state_raw)
         prior = state.get('host_attempt', {})
         failed = prior.get('status') == 'failed'
-        if (prior.get('status') not in {'started', 'failed'}
+        returned_budget_stop = prior.get('status') == 'stopped'
+        if (prior.get('status') not in {'started', 'failed', 'stopped'}
                 or json.loads(copied['host_pending'].read_bytes()) != {'pending': True}):
             raise ValueError('Stopped legacy recovery requires the original unresolved Host attempt')
         wake = copied['wake_log_dir'] / declaration['wake_turn_id'] / 'receipt.json'
@@ -116,6 +117,26 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
         receipt = json.loads(raw)
         candidates = [p for p in copied['wake_log_dir'].glob('*/receipt.json')
                       if p.stat().st_mtime > copied['scheduler_state'].stat().st_mtime]
+        if returned_budget_stop:
+            from benchmark.runtime.scheduler import BUDGET_EXHAUSTED_EXIT_CODE
+
+            session = receipt.get('session', {})
+            if (prior.get('attempt_id') is not None or 'host_failure' in prior
+                    or type(prior.get('stop_exit_code')) is not int
+                    or prior['stop_exit_code'] != BUDGET_EXHAUSTED_EXIT_CODE
+                    or type(state.get('last_wake_rc')) is not int
+                    or state['last_wake_rc'] != BUDGET_EXHAUSTED_EXIT_CODE
+                    or state.get('last_wake_status') != 'wake_requested_stop'
+                    or state.get('last_wake_failure_kind') is not None
+                    or receipt.get('timed_out') is not True
+                    or type(receipt.get('return_code', BUDGET_EXHAUSTED_EXIT_CODE)) is not int
+                    or receipt.get('return_code', BUDGET_EXHAUSTED_EXIT_CODE) != BUDGET_EXHAUSTED_EXIT_CODE
+                    or session.get('binding_scope') != 'agent'
+                    or not isinstance(session.get('session_id'), str) or not session['session_id'].strip()
+                    or prior.get('session_id') not in (None, session['session_id'])
+                    or any(key in receipt and receipt[key] != declaration[key]
+                           for key in ('goal_id', 'agent_id'))):
+                raise ValueError('Stopped source lacks its exact returned budget stop')
         if failed:
             from loopx.control_plane.turn_driver.host_failure import build_host_failure_record, normalize_host_failure_record
 
@@ -131,12 +152,14 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
                     or type(receipt.get('return_code')) is not int or receipt['return_code'] != 1
                     or receipt.get('budget_exhausted')):
                 raise ValueError('Stopped source lacks its exact returned quota failure')
-            # A returned failure is committed after its receipt. Do not borrow
-            # an earlier or later wake, or treat an uncertain started attempt as returned.
+        if failed or returned_budget_stop:
+            # Returned failures and budget stops are committed after their
+            # receipt. A still-started legacy attempt has the opposite order.
+            # Neither path may borrow an earlier wake or skip a later one.
             all_receipts = list(copied['wake_log_dir'].glob('*/receipt.json'))
             newest = max((p.stat().st_mtime for p in all_receipts), default=0)
             if candidates or [p for p in all_receipts if p.stat().st_mtime == newest] != [wake]:
-                raise ValueError('Stopped failure is not the last returned wake')
+                raise ValueError('Stopped source is not the last returned wake')
             candidates = [wake]
         if candidates != [wake]:
             raise ValueError('Stopped source has no unique returned wake after its Host attempt')
@@ -147,12 +170,18 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
             raise ValueError('Stopped source lacks an exact interrupted budget receipt')
         start = int(declaration['wake_turn_id'].split('-')[1]) / 1_000_000_000
         phase = float(copied['phase_clock'].read_text())
-        if not math.isfinite(phase) or not start <= min(phase, finished) or (not failed and phase > finished):
+        # A returned stop proves the process was reaped before its receipt.
+        # Legacy workers could reserve cleanup time before the hard deadline;
+        # do not enlarge that native Turn's window to a later container stop.
+        process_stopped = math.floor(wake.stat().st_mtime) if returned_budget_stop else finished
+        if (not math.isfinite(phase) or not start <= min(phase, process_stopped)
+                or process_stopped > finished
+                or (not failed and not returned_budget_stop and phase > finished)):
             raise ValueError('Stopped source phase did not expire during the original wake')
         result = {'container_id': box.id, 'source_state': before, 'declaration': declaration,
                   'components': components, 'scheduler_state_sha256': hashlib.sha256(state_raw).hexdigest(),
                   'receipt_sha256': hashlib.sha256(raw).hexdigest(),
-                  'process_started_at': start, 'process_stopped_at': finished,
+                  'process_started_at': start, 'process_stopped_at': process_stopped,
                   'source_phase_deadline_epoch': phase, 'session': receipt.get('session'),
                   'execution_authority_granted': False}
         box.reload()
diff --git a/benchmark/runtime/interruption_recovery.py b/benchmark/runtime/interruption_recovery.py
index f5cc59c1f6..392af9aac9 100644
--- a/benchmark/runtime/interruption_recovery.py
+++ b/benchmark/runtime/interruption_recovery.py
@@ -208,7 +208,12 @@ def read_interrupted_host(state: dict, *, state_file: Path, env: dict[str, str],
         phase = float(source['source_phase_deadline_epoch'] if legacy else receipt["phase_deadline_epoch"])
     except (KeyError, TypeError, ValueError) as error:
         raise ValueError("Original Host phase clock is unavailable") from error
-    if not math.isfinite(phase) or not started <= phase <= stopped:
+    returned_legacy_stop = source is not None and prior.get('status') == 'stopped'
+    if (returned_legacy_stop
+            and stopped != math.floor(path.stat().st_mtime)):
+        raise ValueError('Returned legacy Host window differs from its receipt')
+    if (not math.isfinite(phase) or not started <= phase
+            or (not returned_legacy_stop and phase > stopped)):
         raise ValueError("Original Host stopped outside its phase deadline")
     session = receipt.get("session") or {}
     if session.get("binding_scope") != "agent" or not session.get("session_id"):
diff --git a/benchmark/tests/test_interruption_recovery.py b/benchmark/tests/test_interruption_recovery.py
index 7d5ebc6f59..1d76b52ce8 100644
--- a/benchmark/tests/test_interruption_recovery.py
+++ b/benchmark/tests/test_interruption_recovery.py
@@ -82,13 +82,18 @@ def test_live_process_group_cannot_be_settled():
 
 
 @pytest.mark.skipif(os.name != "posix", reason="Owned process-group cleanup requires POSIX")
-@pytest.mark.parametrize("fault", [None, "profile", "stop", "wrong_session", "duplicate_wake"])
-def test_real_sqlite_scheduler_retains_interrupted_session_then_admits_next_turn(tmp_path, monkeypatch, fault):
+@pytest.mark.parametrize("fault", [None, "profile", "stop", "wrong_session", "duplicate_wake", "late_native"])
+@pytest.mark.parametrize('legacy_returned', [False, True])
+def test_real_sqlite_scheduler_retains_interrupted_session_then_admits_next_turn(tmp_path, monkeypatch, fault, legacy_returned):
     from benchmark.runtime import worker
     from benchmark.tests.test_worker_phase_budget import _phase_worker
     from tests import test_external_scheduler_worker as scheduler_fixture
 
     env, project, _runtime, _source, goal_state, _clock = _phase_worker(tmp_path, monkeypatch)
+    # The installed worker keeps its native home outside the submitted workspace.
+    # Match that layout: the home also contains a shared-skills symlink which is
+    # intentionally not a stopped-source participant.
+    env['LOOPX_CODEX_HOME'] = str(tmp_path / 'native-home')
     monkeypatch.setattr(worker, "time", time)
     binary = Path(env["CODEX_BIN"])
     binary.write_text(f"#!{sys.executable}\n" + '''
@@ -149,6 +154,44 @@ def interrupted(*args):
     assert first["host_invoked"] and first["budget_exhausted"]
     assert first["host_attempt_id"] == old["host_attempt"]["attempt_id"]
     assert first["native_process"]["after_cleanup"]["return_code"] is not None
+    if fault == 'late_native':
+        history = project / 'native-history.json'
+        value = json.loads(history.read_bytes())
+        value['completedAt'] = int(time.time()) + 60
+        history.write_text(json.dumps(value))
+    if legacy_returned:
+        from datetime import datetime, timezone
+        from benchmark.edgebench.stopped_recovery import prepare_stopped_recovery
+        from benchmark.tests.test_stopped_recovery import _fake_container
+
+        # Model the previous writer's actual persisted shape. These synthetic
+        # source bytes, including the returned stop, must remain unchanged.
+        old['host_attempt'].pop('attempt_id')
+        old['host_attempt']['session_id'] = None
+        original.write_text(json.dumps(old))
+        old_bytes = original.read_bytes()
+        for key in ('host_attempt_id', 'native_process', 'goal_id', 'agent_id'):
+            first.pop(key, None)
+        receipts[0].write_text(json.dumps(first))
+        returned_at = time.time()
+        os.utime(receipts[0], (returned_at, returned_at))
+        os.utime(original, (returned_at + .1, returned_at + .1))
+        home = Path(env['LOOPX_CODEX_HOME'])
+        for name in ('state_5.sqlite', 'thread_history_1.sqlite'):
+            (home / name).write_text('synthetic native observer state')
+        (home / 'sessions').mkdir(exist_ok=True)
+        (home / 'sessions/original.jsonl').write_text('{}\n')
+        phase_clock = tmp_path / 'original-phase-deadline'
+        phase_clock.write_text(os.environ['LOOPX_PHASE_DEADLINE_EPOCH'])
+        declaration = dict(goal_id=env['LOOPX_GOAL_ID'], agent_id=env['LOOPX_AGENT_ID'],
+            workspace=str(project), runtime_root=env['LOOPX_RUNTIME_ROOT'], native_home=str(home),
+            registry=env['LOOPX_REGISTRY'], scheduler_state=str(original),
+            wake_log_dir=env['LOOPX_WAKE_LOG_DIR'], phase_clock=str(phase_clock), wake_turn_id=first['turn_id'])
+        box, client = _fake_container(monkeypatch)
+        box.attrs['State']['FinishedAt'] = datetime.now(timezone.utc).isoformat()
+        proof = tmp_path / 'legacy-source-proof.json'
+        prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=proof)
+        monkeypatch.setenv('LOOPX_INTERRUPTED_SOURCE', str(proof))
     deadline = time.time() + 60
     monkeypatch.setenv("LOOPX_PHASE_DEADLINE_EPOCH", str(deadline))
     if fault == "profile":
@@ -171,6 +214,7 @@ def interrupted(*args):
         brief.write_text(json.dumps({'ok': True, 'handoff_text': 'Retained submission old-7 is now restored-7; read the inherited history. Goal and opportunities are unchanged.'}))
         monkeypatch.setenv('LOOPX_RECOVERY_HANDOFF', str(brief))
         monkeypatch.setenv('LOOPX_RECOVERY_HANDOFF_SHA256', hashlib.sha256(brief.read_bytes()).hexdigest())
+    original_receipt = receipts[0].read_bytes()
     new = resume_state_file(str(original), phase_deadline=deadline, agent_id=args.agent_id,
         wake_log_dir=env["LOOPX_WAKE_LOG_DIR"], env=dict(os.environ))
     if fault:
@@ -192,6 +236,7 @@ def interrupted(*args):
     argv = [json.loads(line) for line in (project / "native-argv").read_text().splitlines()]
     assert len(argv) == 2 and "resume" in argv[1] and "original-native" in argv[1]
     assert original.read_bytes() == old_bytes and goal_state.read_bytes() == initial_goal
+    assert receipts[0].read_bytes() == original_receipt
     assert json.loads(Path(new).read_bytes())["recovery_handoff_delivered"] is True
     assert scheduler_fixture.worker.run_worker(args) == 0
     inputs = [json.loads(line) for line in (project / 'native-inputs').read_text().splitlines()]
diff --git a/benchmark/tests/test_stopped_recovery.py b/benchmark/tests/test_stopped_recovery.py
index fc4049b0fc..2b4862a06d 100644
--- a/benchmark/tests/test_stopped_recovery.py
+++ b/benchmark/tests/test_stopped_recovery.py
@@ -63,6 +63,73 @@ def archive(container, remote, destination, **kwargs):
     return box, SimpleNamespace(containers=SimpleNamespace(get=lambda identity: box))
 
 
+def _returned_budget_stop(declaration):
+    state = Path(declaration['scheduler_state'])
+    value = json.loads(state.read_bytes())
+    value['host_attempt'].update(status='stopped', stop_exit_code=75, session_id='original')
+    value.update(last_wake_rc=75, last_wake_status='wake_requested_stop', last_wake_failure_kind=None)
+    state.write_text(json.dumps(value))
+    os.utime(state, (111, 111))
+    receipt = Path(declaration['wake_log_dir']) / declaration['wake_turn_id'] / 'receipt.json'
+    value = json.loads(receipt.read_bytes())
+    value['timed_out'] = True
+    receipt.write_text(json.dumps(value))
+    os.utime(receipt, (110, 110))
+    return state, receipt
+
+
+@pytest.mark.parametrize('fault', [None, 'null_prior_session', 'same_mtime', 'cleanup_reserve', 'live', 'new_attempt',
+    'not_stop', 'wrong_exit', 'failure', 'unknown_failure', 'cleared_pending', 'wrong_session',
+    'missing_session', 'wrong_goal', 'wrong_agent', 'not_timed_out', 'not_invoked',
+    'not_budget', 'successful', 'success_exit', 'later_wake', 'tied_wake', 'receipt_after_state',
+    'outside_phase'])
+def test_returned_legacy_budget_stop_requires_its_last_receipt(tmp_path, monkeypatch, fault):
+    declaration = _source(tmp_path)
+    state, receipt = _returned_budget_stop(declaration)
+    value, returned = json.loads(state.read_bytes()), json.loads(receipt.read_bytes())
+    box, client = _fake_container(monkeypatch)
+    if fault == 'null_prior_session': value['host_attempt']['session_id'] = None
+    elif fault == 'live': box.attrs['State'].update(Status='running', Running=True, Pid=42)
+    elif fault == 'new_attempt': value['host_attempt']['attempt_id'] = 'b' * 32
+    elif fault == 'not_stop': value['host_attempt']['status'] = 'unknown'
+    elif fault == 'wrong_exit': value['host_attempt']['stop_exit_code'] = 1
+    elif fault == 'failure': value['host_attempt']['host_failure'] = {'kind': 'quota_exhausted'}
+    elif fault == 'unknown_failure': value['last_wake_failure_kind'] = 'unknown'
+    elif fault == 'cleared_pending': state.with_name(state.name + '.host-attempt').write_text('{"pending":false}')
+    elif fault == 'wrong_session': returned['session']['session_id'] = 'another'
+    elif fault == 'missing_session': returned['session']['session_id'] = ''
+    elif fault == 'wrong_goal': returned['goal_id'] = 'another-goal'
+    elif fault == 'wrong_agent': returned['agent_id'] = 'another-agent'
+    elif fault == 'not_timed_out': returned['timed_out'] = False
+    elif fault == 'not_invoked': returned['host_invoked'] = False
+    elif fault == 'not_budget': returned['budget_exhausted'] = False
+    elif fault == 'successful': returned['ok'] = True
+    elif fault == 'success_exit': returned['return_code'] = 0
+    elif fault == 'outside_phase': Path(declaration['phase_clock']).write_text('90')
+    elif fault == 'cleanup_reserve': Path(declaration['phase_clock']).write_text('130')
+    state.write_text(json.dumps(value)); os.utime(state, (111, 111))
+    receipt.write_text(json.dumps(returned))
+    stamp = 111 if fault == 'same_mtime' else 112 if fault == 'receipt_after_state' else 110
+    os.utime(receipt, (stamp, stamp))
+    if fault in {'later_wake', 'tied_wake'}:
+        other = receipt.parent.with_name('wake-109000000000-aaaaaaaaaaaa') / 'receipt.json'
+        other.parent.mkdir(); other.write_text('{}')
+        stamp = 110.5 if fault == 'later_wake' else 110
+        os.utime(other, (stamp, stamp))
+    original = state.read_bytes(), receipt.read_bytes()
+    output = tmp_path / 'source-proof.json'
+    if fault not in {None, 'null_prior_session', 'same_mtime', 'cleanup_reserve'}:
+        with pytest.raises(ValueError):
+            subject.prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=output)
+        assert not output.exists()
+    else:
+        result = subject.prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=output)
+        assert result['session']['session_id'] == 'original'
+        assert result['process_stopped_at'] == stamp
+        assert result['execution_authority_granted'] is False
+    assert (state.read_bytes(), receipt.read_bytes()) == original
+
+
 @pytest.mark.parametrize("fault", [None, "live", "restart_command", "mount", "custom_hostname", "ambiguous_wake"])
 def test_prepare_stopped_source_keeps_original_state_and_requires_unique_return(tmp_path, monkeypatch, fault):
     declaration = _source(tmp_path)
@@ -138,13 +205,16 @@ def test_stopped_recovery_rejects_invalid_archive_timeout(tmp_path, timeout):
 
 
 @pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1", reason="Explicit disposable Docker qualification")
-def test_real_stopped_docker_proof_installation_is_inert_and_read_back(tmp_path, monkeypatch):
+@pytest.mark.parametrize('returned', [False, True])
+def test_real_stopped_docker_proof_installation_is_inert_and_read_back(tmp_path, monkeypatch, returned):
     docker = pytest.importorskip("docker")
     client = docker.from_env(timeout=30)
     box = client.containers.create("python:3.12-slim", ["tail", "-f", "/dev/null"], network_disabled=True)
     local = tmp_path / "fixture"
     local.mkdir()
     declaration = _source(local)
+    if returned:
+        _returned_budget_stop(declaration)
     declaration = {key: value.replace(str(local), "/fixture") for key, value in declaration.items()}
     try:
         package = io.BytesIO()