fix(benchmark): qualify returned legacy budget interruptions

benchmark/README.md

@@ -1438,6 +1438,18 @@ inert proof in that same container's control directory; omitting it is read-only
 Neither form starts the container, restores scoring authority or calls a model.
 Keep exclusive ownership of the source during preparation and continuation.
 
+The legacy source may still say `started`, or may already record a returned
+`stopped` attempt with exit code 75. The latter requires the unique latest wake
+receipt at or before that scheduler snapshot, with `host_invoked=true`,
+`budget_exhausted=true` and `timed_out=true`, no failure, and the original agent
+session. It is not the pre-dispatch budget shortcut. Preparation preserves the
+returned stop rather than changing it to `started` or inventing an attempt ID;
+the receiver must still verify the original native interrupted tail and current
+Goal/session admission. A later or ambiguous wake keeps the source held.
+For a returned stop, the native window ends at the receipt's recorded second,
+not a later maintenance stop of the container. The original hard deadline stays
+intact even when an older worker reserved cleanup time before it.
+
 After independently restoring the original native evaluator's history,
 opportunities and actual cooldown, restart only the same inert container and
 pass its returned `installed_proof_path` as `LOOPX_INTERRUPTED_SOURCE` to the
@@ -1480,3 +1492,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,29 +106,82 @@ 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'
+        returned_budget_stop = prior.get('status') == 'stopped'
+        if (prior.get('status') not in {'started', 'failed', 'stopped'}
                 or json.loads(copied['host_pending'].read_bytes()) != {'pending': True}):
             raise ValueError('Stopped legacy recovery requires the original unresolved Host attempt')
         wake = copied['wake_log_dir'] / declaration['wake_turn_id'] / 'receipt.json'
         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 returned_budget_stop:
+            from benchmark.runtime.scheduler import BUDGET_EXHAUSTED_EXIT_CODE
+
+            session = receipt.get('session', {})
+            if (prior.get('attempt_id') is not None or 'host_failure' in prior
+                    or type(prior.get('stop_exit_code')) is not int
+                    or prior['stop_exit_code'] != BUDGET_EXHAUSTED_EXIT_CODE
+                    or type(state.get('last_wake_rc')) is not int
+                    or state['last_wake_rc'] != BUDGET_EXHAUSTED_EXIT_CODE
+                    or state.get('last_wake_status') != 'wake_requested_stop'
+                    or state.get('last_wake_failure_kind') is not None
+                    or receipt.get('timed_out') is not True
+                    or type(receipt.get('return_code', BUDGET_EXHAUSTED_EXIT_CODE)) is not int
+                    or receipt.get('return_code', BUDGET_EXHAUSTED_EXIT_CODE) != BUDGET_EXHAUSTED_EXIT_CODE
+                    or session.get('binding_scope') != 'agent'
+                    or not isinstance(session.get('session_id'), str) or not session['session_id'].strip()
+                    or prior.get('session_id') not in (None, session['session_id'])
+                    or any(key in receipt and receipt[key] != declaration[key]
+                           for key in ('goal_id', 'agent_id'))):
+                raise ValueError('Stopped source lacks its exact returned budget stop')
+        if failed:
+            from loopx.control_plane.turn_driver.host_failure import build_host_failure_record, normalize_host_failure_record
+
+            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')
+        if failed or returned_budget_stop:
+            # Returned failures and budget stops are committed after their
+            # receipt. A still-started legacy attempt has the opposite order.
+            # Neither path may borrow an earlier wake or skip a later one.
+            all_receipts = list(copied['wake_log_dir'].glob('*/receipt.json'))
+            newest = max((p.stat().st_mtime for p in all_receipts), default=0)
+            if candidates or [p for p in all_receipts if p.stat().st_mtime == newest] != [wake]:
+                raise ValueError('Stopped source is not the last returned wake')
+            candidates = [wake]
         if candidates != [wake]:
             raise ValueError('Stopped source has no unique returned wake after its Host attempt')
         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:
+        # A returned stop proves the process was reaped before its receipt.
+        # Legacy workers could reserve cleanup time before the hard deadline;
+        # do not enlarge that native Turn's window to a later container stop.
+        process_stopped = math.floor(wake.stat().st_mtime) if returned_budget_stop else finished
+        if (not math.isfinite(phase) or not start <= min(phase, process_stopped)
+                or process_stopped > finished
+                or (not failed and not returned_budget_stop and phase > finished)):
             raise ValueError('Stopped source phase did not expire during the original wake')
         result = {'container_id': box.id, 'source_state': before, 'declaration': declaration,
                   'components': components, 'scheduler_state_sha256': hashlib.sha256(state_raw).hexdigest(),
                   'receipt_sha256': hashlib.sha256(raw).hexdigest(),
-                  'process_started_at': start, 'process_stopped_at': finished,
+                  'process_started_at': start, 'process_stopped_at': process_stopped,
                   'source_phase_deadline_epoch': phase, 'session': receipt.get('session'),
                   'execution_authority_granted': False}
         box.reload()

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.