fix(benchmark): recover a stopped worker after a completed host wake

benchmark/edgebench/README.md

@@ -290,15 +290,14 @@ An Official worker without a checkpoint must retain its existing session and
 use its authorized runtime owner's idle credential handoff; this preparation
 helper does not provide a live credential-replacement transaction.
 
-An unchanged stopped heartbeat worker with an empty-output legacy Host failure
-can use the explicit [idle-host handoff](../runtime/RUNTIME.md) after independent
+An unchanged stopped heartbeat worker with a completed Host or an empty-output
+legacy failure can use the explicit [idle-host handoff](../runtime/RUNTIME.md) after independent
 native-history, source, current Goal and account qualification. Seal with
 `python -m benchmark.edgebench.stopped_recovery --idle-host-handoff` using the
 original source declaration; then select the original scheduler digest with
 `benchmark.runtime.scheduler --idle-host-state-sha256`. Both options are off by
-default. The original failure remains recorded. This route preserves the
-existing completed session and source rather than manufacturing an interrupted
-Turn. Native judge restoration, opportunity/cooldown conservation, the audited
+default. The original return, session, and source remain recorded.
+Native judge restoration, opportunity/cooldown conservation, the audited
 remaining allowance and a finite owner cutoff remain separate requirements.
 It does not apply to Official or native-Goal workers.
 

benchmark/edgebench/stopped_recovery.py

@@ -102,6 +102,47 @@ def validate_idle_host_return(state: dict, receipt: dict, *, stdout: bytes, stde
         raise ValueError('Idle handoff requires its original empty-output legacy Host return')
 
 
+
+def completed_host_window(state: dict, receipt: dict, *, lineage: dict, phase: float) -> tuple[float, float]:
+    """Verify a completed process at the retained clock without judging its work."""
+    from benchmark.runtime.interruption_recovery import _number
+
+    process = receipt.get("native_process") or {}
+    before = process.get("before_cleanup") or {}
+    after = process.get("after_cleanup") or {}
+    session = receipt.get("session") or {}
+    started = _number(process.get("started_at_unix"))
+    returned = _number(before.get("observed_at_unix"))
+    stopped = _number(after.get("observed_at_unix"))
+    try:
+        wake_phase = float(receipt.get("phase_deadline_epoch"))
+    except (TypeError, ValueError) as error:
+        raise ValueError("Completed Host requires its original phase clock") from error
+    if ("host_attempt" in state or state.get("last_wake_status") != "wake_ok"
+            or state.get("last_wake_failure_kind") is not None
+            or any(type(row.get(key)) is not int or row[key] != 0
+                   for row, key in ((state, "last_wake_rc"), (receipt, "return_code"),
+                                    (before, "return_code"), (after, "return_code")))
+            or receipt.get("mode") != "heartbeat" or receipt.get("stage") != "execute"
+            or receipt.get("ok") is not True or receipt.get("host_invoked") is not True
+            or receipt.get("timed_out") is not False or receipt.get("budget_exhausted")
+            or "host_failure" in receipt or receipt.get("error_kind") is not None
+            or any(receipt.get(key) != lineage[key] for key in ("goal_id", "agent_id"))
+            or not re.fullmatch(r"[a-f0-9]{32}", receipt.get("host_attempt_id") or "")
+            or session.get("binding_scope") != "agent"
+            or not isinstance(session.get("session_id"), str) or not session["session_id"].strip()
+            or process.get("schema_version") != "native_codex_process_observation_v1"
+            or type(process.get("pid")) is not int or process["pid"] <= 1
+            or process.get("cleanup_trigger") != "scope_completed"
+            or any(process.get(key) is not None for key in
+                   ("first_failure", "scope_error_type", "cleanup_error_type", "persistence_error_type"))
+            or before.get("signal") is not None or after.get("signal") is not None
+            or not 0 < started <= returned <= stopped <= phase
+            or not math.isfinite(phase) or wake_phase != phase):
+        raise ValueError("Stopped source lacks its original completed Host return")
+    return started, stopped
+
+
 def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
                              output: Path, timeout: float = 120, install: bool = False,
                              idle_host_handoff: bool = False) -> dict:
@@ -149,8 +190,12 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
         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}):
+        completed = idle_host_handoff and "host_attempt" not in state
+        pending = json.loads(copied['host_pending'].read_bytes())
+        if completed:
+            if pending != {'pending': False}:
+                raise ValueError('Completed Host handoff requires its cleared pending witness')
+        elif prior.get('status') not in {'started', 'failed', 'stopped'} or pending != {'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()
@@ -177,7 +222,7 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
                     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 idle_host_handoff:
+        if idle_host_handoff and not completed:
             from benchmark.runtime.worker import host_output_paths
 
             stdout, stderr = host_output_paths(wake.parent)
@@ -198,8 +243,8 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
                     or type(receipt.get('return_code')) is not int or receipt['return_code'] != 1
                     or receipt.get('budget_exhausted')):
                 raise ValueError('Stopped source lacks its exact returned quota failure')
-        if failed or returned_budget_stop:
-            # Returned failures and budget stops are committed after their
+        if failed or returned_budget_stop or completed:
+            # Returned Hosts 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'))
@@ -209,27 +254,33 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
             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 (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 (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
+        wake_started = int(declaration['wake_turn_id'].split('-')[1]) / 1_000_000_000
         phase = float(copied['phase_clock'].read_text())
-        # 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 or idle_host_handoff 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')
+        if completed:
+            start, process_stopped = completed_host_window(state, receipt, lineage=declaration, phase=phase)
+            if (receipt.get('turn_id') != declaration['wake_turn_id'] or start < wake_started
+                    or process_stopped > min(finished, math.floor(wake.stat().st_mtime) + 1)):
+                raise ValueError('Completed Host process lies outside the retained wake')
+        else:
+            if (receipt.get('turn_id') != declaration['wake_turn_id'] or receipt.get('host_invoked') is not True
+                    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 (not failed and 'host_failure' in receipt)):
+                raise ValueError('Stopped source lacks an exact interrupted budget receipt')
+            start = wake_started
+            # Legacy workers could reserve cleanup time before the hard deadline.
+            process_stopped = math.floor(wake.stat().st_mtime) if returned_budget_stop or idle_host_handoff 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': process_stopped,
                   'source_phase_deadline_epoch': phase, 'session': receipt.get('session'),
                   **({'idle_host_handoff': True} if idle_host_handoff else {}),
+                  **({'completed_host': True} if completed else {}),
                   'execution_authority_granted': False}
         box.reload()
         if box.attrs['State'] != before:
@@ -287,7 +338,7 @@ def main():
     parser.add_argument('--output', type=Path, required=True)
     parser.add_argument('--install', action='store_true', help='Place the inert proof only in this same stopped worker')
     parser.add_argument('--idle-host-handoff', action='store_true',
-                        help='Explicitly seal an empty-output legacy Host failure for separate idle-history qualification')
+                        help='Explicitly seal a returned Host for separate idle-history qualification')
     args = parser.parse_args()
     import docker
     result = prepare_stopped_recovery(docker.from_env(timeout=30),

benchmark/runtime/RUNTIME.md

@@ -440,17 +440,18 @@ Budget exhaustion is distinct from task completion: this clock does not override
 provider failures, Stop, blocked planning, or a valid terminal LoopX decision.
 Campaign duration coverage must still classify an early exit as missing time.
 
-An operator may explicitly hand off an unchanged stopped legacy heartbeat Host
-whose failed attempt produced no stdout/stderr and returned a phase timeout,
-while the native session remained idle. This is disabled by default. First run
-`benchmark.edgebench.stopped_recovery` with `--idle-host-handoff` and, when
-needed, `--install` to place its inert proof in that same stopped worker. The
-producer requires the exact failed scheduler, pending witness, latest unique
-wake, original workspace, registry, runtime and native stores, including committed
-SQLite WALs. It preserves the unknown failure and does not infer a quota failure
-from diagnostic text. The proof grants no execution or evaluator authority.
-The empty-output witness uses the worker's `stdout.jsonl` and `stderr.log` files;
-the producer and receiver share that writer-owned layout.
+An operator may explicitly hand off an unchanged stopped heartbeat worker when
+its native session is idle. First run `benchmark.edgebench.stopped_recovery`
+with `--idle-host-handoff` and, when needed, `--install` to place its inert
+proof in that same stopped worker. The proof includes the latest unique wake,
+the scheduler and its independent pending witness, the original workspace,
+registry, runtime and native stores, including committed SQLite WALs.
+This path is disabled by default and grants no execution or evaluator authority.
+
+A completed Host requires a successful process receipt at the recorded phase
+deadline and a cleared pending witness. The producer also supports a legacy
+empty-output phase timeout with a still-pending witness. It checks the worker's
+`stdout.jsonl` and `stderr.log` and preserves the recorded unknown failure.
 
 After the owner restores compatible native evaluator authority separately,
 restart only the original inert container and retain its original environment.
@@ -460,8 +461,9 @@ authorized finite `LOOPX_PHASE_DEADLINE_EPOCH`. Invoke
 `--idle-host-state-sha256 ORIGINAL_SCHEDULER_SHA256`. The consumer verifies the
 same namespace and complete unchanged source, current Goal/session/profile and
 fresh native account allowance. The native history owner must return a stable
-complete history whose last completed cursor precedes the failed attempt, with
-no pending effects. Its observer uses a disposable copy of the native store and
+complete history with no pending effects. For a completed Host, the last
+native Turn must fall inside the original process window. For a legacy timeout,
+it must precede the attempted launch. The observer uses a disposable copy of the native store and
 committed WALs because native read APIs can update their own history index; it
 copies no authentication file. Every copied native base participant must match
 the producer's sealed set; an extra archived-session/index store or database
@@ -474,9 +476,9 @@ existing lock owner's process/token witnesses; registry contents, SQLite/WALs,
 work files and unrelated lock files remain protected.
 
 An accepted handoff creates a new finite transport phase linked to the original
-failed record. It retains the session, Goal, Todos, work and failure evidence;
-it does not settle an old Turn, refill native opportunities/cooldown, or create
-another task budget. Repeating the exact phase retains its clock and any later
+scheduler record. It retains the session, Goal, Todos, work, and any pending
+semantic settlement. The existing owners still enforce settlement, native
+opportunities, cooldown and the total task budget. Repeating the exact phase retains its clock and any later
 failure; a changed deadline cannot replay the original handoff. The owner still
 enforces its audited remaining solver allowance and external cutoff, including
 setup and cleanup. Omit both new opt-ins to keep the ordinary refusal path.