fix(benchmark): explicitly continue repaired quota failures on the original session

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.

benchmark/edgebench/stopped_recovery.py

@@ -106,24 +106,48 @@ 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'
         raw = wake.read_bytes()
         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(),

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.

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}