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 dad55a96eec56fe5e5c9ae8f489bddacc806dc8d Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Sat, 7 Nov 2026 18:20:31 +0800
Subject: [PATCH 2/2] fix(benchmark): hold explicit retries without qualified
 Host state

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 benchmark/runtime/scheduler.py                |  65 +++++++--
 .../tests/test_failed_host_continuation.py    | 131 +++++++++++++++---
 2 files changed, 168 insertions(+), 28 deletions(-)

diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py
index cc6e0e4a90..9b7008877f 100644
--- a/benchmark/runtime/scheduler.py
+++ b/benchmark/runtime/scheduler.py
@@ -110,7 +110,38 @@ def _create_resume_state(target: Path, predecessor: Path, *, recovered: dict | N
             stream.flush()
             os.fsync(stream.fileno())
     except FileExistsError:
-        return
+        if failed_recovery:
+            raise ValueError("Explicit failed continuation phase already exists") from None
+
+
+def _verify_failed_phase_replay(phase: Path, state: dict, recovery: dict, *,
+                                states: list[Path], digest: str) -> None:
+    from .interruption_recovery import _physical_read
+    from loopx.control_plane.turn_driver.host_failure import normalize_host_failure_record
+
+    original_ref = state.get("resume_from")
+    original = Path(original_ref) if isinstance(original_ref, str) else None
+    if original is None or original == phase or original not in states:
+        raise ValueError("Explicit failed continuation requires the original replay source")
+    raw = _physical_read(original)
+    if hashlib.sha256(raw).hexdigest() != digest:
+        raise ValueError("Explicit failed continuation replay source changed")
+    saved = json.loads(raw)
+    prior = saved.get("host_attempt") if isinstance(saved, dict) else None
+    native = recovery.get("native")
+    if not isinstance(prior, dict) or not isinstance(native, dict):
+        raise ValueError("Explicit failed continuation requires the original Host and native session")
+    failure = normalize_host_failure_record(prior.get("host_failure"))
+    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 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
+            or type(state.get("recovered_attempt_count")) is not int
+            or state["recovered_attempt_count"] != failure["attempt"]):
+        raise ValueError("Explicit failed continuation replay differs from the original returned failure")
 
 
 def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str,
@@ -133,25 +164,30 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str,
     from loopx.file_lock import try_exclusive_file_lock
     from scripts.external_scheduler_worker import HOST_LOCK_OPERATION, _host_attempt, _host_pending
 
+    def hold(state: Path, reason: str) -> str:
+        if retry_failed_state_sha256 is not None:
+            raise ValueError(f"Explicit failed continuation {reason}")
+        return str(state)
+
     with ExitStack() as locks:
         # The root serializes phase selection. Locking the existing phase files
         # as well keeps a running/failed phase from moving under that decision.
         if locks.enter_context(try_exclusive_file_lock(
                 path, agent_id=agent_id, operation=HOST_LOCK_OPERATION)) is None:
-            return str(path)
+            return hold(path, "requires the scheduler root lock")
         for phase in _phase_states(path):
             if phase == path:
                 continue
             if locks.enter_context(try_exclusive_file_lock(
                     phase, agent_id=agent_id, operation=HOST_LOCK_OPERATION)) is None:
-                return str(phase)
+                return hold(phase, "requires the retained phase lock")
         states = _phase_states(path)
         if not states:
-            return str(path)
+            return hold(path, "requires the retained scheduler state")
         predecessor = _latest_phase_state(path, states)
         state = _state_object(predecessor)
         if state is None:
-            return str(predecessor)
+            return hold(predecessor, "requires a readable retained scheduler state")
         prior = state.get("host_attempt")
         recovery = state.get("interruption_recovery") or state.get('failed_host_recovery')
         can_continue_stopped_phase = (isinstance(prior, dict) and prior.get("status") == "stopped"
@@ -168,29 +204,34 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str,
                 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):
+            if (env is None or not wake_log_dir or env.get("LOOPX_EXECUTION_MODE") != "heartbeat"
+                    or env.get("LOOPX_ITERATION_CONTEXT") != "resume"):
+                raise ValueError("Explicit failed continuation requires original native and wake routes")
+            failed_phase = state.get("failed_host_recovery")
+            if (same_phase and isinstance(failed_phase, dict)
+                    and failed_phase.get("source_state_sha256") == retry_failed_state_sha256):
+                _verify_failed_phase_replay(predecessor, state, failed_phase,
+                                            states=states, digest=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)
+            return hold(predecessor, "requires the original Host attempt")
         # Validate the captured record with its existing transport owner.
         scope = prior.get("scope")
         if not isinstance(scope, str) or not re.fullmatch(r"[0-9a-f]{64}", scope):
-            return str(predecessor)
+            return hold(predecessor, "requires the original Host attempt scope")
         try:
             _host_attempt(state, scope)
         except (TypeError, ValueError):
-            return str(predecessor)
+            return hold(predecessor, "requires a valid original Host attempt")
         # A returned interrupted wake is distinct from a pre-dispatch stop.
         # Explicit continuation only reconciles transport; all Goal/Stop,
         # replan, lease and launch decisions remain with the normal next wake.
         recovered = 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')
+            assert env is not None and wake_log_dir is not None
             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)
diff --git a/benchmark/tests/test_failed_host_continuation.py b/benchmark/tests/test_failed_host_continuation.py
index 0fd79d20bc..c41dc58fa3 100644
--- a/benchmark/tests/test_failed_host_continuation.py
+++ b/benchmark/tests/test_failed_host_continuation.py
@@ -8,6 +8,7 @@
 import os
 from pathlib import Path
 import shlex
+import subprocess
 import sys
 import time
 
@@ -20,6 +21,81 @@
 from loopx.hosts import codex_checkpoint as native
 
 
+def _run_scheduler_probe(tmp_path, state, *, deadline, digest):
+    marker = tmp_path / "worker-invoked"
+    worker = tmp_path / "scripts" / "external_scheduler_worker.py"
+    worker.parent.mkdir()
+    worker.write_text(f"from pathlib import Path\nPath({str(marker)!r}).touch()\n")
+    env = {key: value for key, value in os.environ.items() if not key.startswith("LOOPX_")}
+    env.update(LOOPX_PHASE_DEADLINE_EPOCH=str(deadline), LOOPX_AGENT_ID="fixture-agent",
+        LOOPX_GOAL_ID="fixture-goal", LOOPX_EXECUTION_MODE="heartbeat", LOOPX_ITERATION_CONTEXT="resume",
+        LOOPX_CLI=str(tmp_path / "fixture-cli"), LOOPX_REGISTRY=str(tmp_path / "registry.json"),
+        LOOPX_RUNTIME_ROOT=str(tmp_path / "runtime"), LOOPX_WAKE_LOG_DIR=str(tmp_path / "wakes"))
+    command = [sys.executable, "-m", "benchmark.runtime.scheduler",
+        "--python", sys.executable, "--source", str(tmp_path), "--state-file", str(state), "--host-timeout", "10"]
+    if digest is not None:
+        command.extend(["--retry-failed-state-sha256", digest])
+    result = subprocess.run(command, cwd=Path(__file__).resolve().parents[2], env=env,
+                            capture_output=True, text=True, timeout=30, check=False)
+    return result, marker.exists()
+
+
+@pytest.mark.parametrize("source", [
+    pytest.param(None, id="missing"),
+    pytest.param(b"\xff", id="unreadable"),
+    pytest.param(b"{}", id="no-attempt"),
+    pytest.param(b'{"host_attempt":{"scope":"unknown"}}', id="invalid-scope"),
+    pytest.param(json.dumps({"host_attempt": {"scope": "a" * 64, "attempt": False}}).encode(),
+                 id="invalid-attempt"),
+])
+def test_explicit_scheduler_never_launches_without_qualified_source(tmp_path, source):
+    state = tmp_path / "scheduler.json"
+    if source is not None:
+        state.write_bytes(source)
+    digest = hashlib.sha256(source).hexdigest() if source is not None else "a" * 64
+    result, invoked = _run_scheduler_probe(tmp_path, state, deadline=time.time() + 60, digest=digest)
+    assert result.returncode != 0 and "Explicit failed continuation" in result.stderr
+    assert not invoked and not list(tmp_path.glob("scheduler.phase-*.json"))
+    assert (state.read_bytes() if state.exists() else None) == source
+
+
+def test_ordinary_scheduler_can_launch_without_explicit_source(tmp_path):
+    state = tmp_path / "scheduler.json"
+    result, invoked = _run_scheduler_probe(tmp_path, state, deadline=time.time() + 60, digest=None)
+    assert result.returncode == 0 and invoked and not state.exists()
+
+
+@pytest.mark.parametrize("locked_file", ["scheduler.json", "scheduler.phase-other.json"])
+def test_explicit_scheduler_does_not_launch_when_phase_selection_is_locked(tmp_path, locked_file):
+    from loopx.file_lock import try_exclusive_file_lock
+    from scripts.external_scheduler_worker import HOST_LOCK_OPERATION
+
+    held = tmp_path / locked_file
+    held.write_bytes(b"{}")
+    with try_exclusive_file_lock(held, agent_id="fixture-agent", operation=HOST_LOCK_OPERATION) as lock:
+        assert lock is not None
+        result, invoked = _run_scheduler_probe(tmp_path, tmp_path / "scheduler.json",
+            deadline=time.time() + 60, digest=hashlib.sha256(b"{}").hexdigest())
+    assert result.returncode != 0 and "Explicit failed continuation" in result.stderr
+    assert not invoked and held.read_bytes() == b"{}"
+
+
+def test_explicit_scheduler_requires_original_binding_for_same_phase_replay(tmp_path):
+    state = tmp_path / "scheduler.json"
+    state.write_bytes(b"{}")
+    digest = hashlib.sha256(b"{}").hexdigest()
+    deadline = time.time() + 60
+    suffix = hashlib.sha256(deadline.hex().encode()).hexdigest()[:20]
+    phase = tmp_path / f"scheduler.phase-{suffix}.json"
+    phase.write_text(json.dumps({"resume_from": str(state), "failed_host_recovery": {
+        "phase_deadline_epoch": deadline, "source_state_sha256": digest,
+    }}))
+    before = phase.read_bytes()
+    result, invoked = _run_scheduler_probe(tmp_path, state, deadline=deadline, digest=digest)
+    assert result.returncode != 0 and "Explicit failed continuation" in result.stderr
+    assert not invoked and phase.read_bytes() == before and state.read_bytes() == b"{}"
+
+
 @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),
@@ -40,12 +116,18 @@ def test_failed_native_readback_preserves_strict_checkpoint_and_unknown_error_ga
     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 = []
+    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):
@@ -59,13 +141,15 @@ def request(self, method, params):
                 rows[0]['items'] = [{'type': 'agentMessage', 'text': 'new'}]
             return {'data': rows, 'nextCursor': None}
     @contextmanager
-    def reader(*a, **kw): yield Transport()
+    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()
+        with pytest.raises(ValueError):
+            read()
     else:
         result = read()
         assert result['status'] == 'failed' and result['turn_count'] == 2
@@ -117,7 +201,8 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path
 print(json.dumps({'type':'turn.failed','error':{'codexErrorInfo':'usageLimitExceeded'}}),flush=True)
 sys.exit(1)
 ''')
-    for key, value in env.items(): monkeypatch.setenv(key, value)
+    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')
@@ -145,11 +230,13 @@ def test_real_sqlite_failed_host_can_explicitly_resume_original_session(tmp_path
         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 == '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 == '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}}}))
@@ -158,7 +245,8 @@ def resume(**kwargs):
             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()
+        with pytest.raises((ValueError, RuntimeError)):
+            resume()
         assert original.read_bytes() == original_bytes
         assert not list(original.parent.glob('*.phase-*.json'))
         return
@@ -171,6 +259,12 @@ def resume(**kwargs):
     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 is None:
+        original.write_bytes(original_bytes + b"\n")
+        with pytest.raises(ValueError, match="replay source changed"):
+            resume()
+        original.write_bytes(original_bytes)
+        assert json.loads(phase.read_bytes()) == phase_initial
     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,
@@ -222,12 +316,17 @@ def test_legacy_quota_failure_proof_retains_complete_same_container_source(tmp_p
     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 == '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)
+        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'):