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

benchmark/edgebench/stopped_recovery.py

@@ -108,14 +108,35 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
         state = json.loads(state_raw)
         prior = state.get('host_attempt', {})
         failed = prior.get('status') == 'failed'
-        if (prior.get('status') not in {'started', 'failed'}
+        returned_budget_stop = prior.get('status') == 'stopped'
+        if (prior.get('status') not in {'started', 'failed', 'stopped'}
                 or json.loads(copied['host_pending'].read_bytes()) != {'pending': True}):
             raise ValueError('Stopped legacy recovery requires the original unresolved Host attempt')
         wake = copied['wake_log_dir'] / declaration['wake_turn_id'] / 'receipt.json'
         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
 
@@ -131,12 +152,14 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
                     or type(receipt.get('return_code')) is not int or receipt['return_code'] != 1
                     or receipt.get('budget_exhausted')):
                 raise ValueError('Stopped source lacks its exact returned quota failure')
-            # A returned failure is committed after its receipt. Do not borrow
-            # an earlier or later wake, or treat an uncertain started attempt as returned.
+        if failed or returned_budget_stop:
+            # Returned failures and budget stops are committed after their
+            # receipt. A still-started legacy attempt has the opposite order.
+            # Neither path may borrow an earlier wake or skip a later one.
             all_receipts = list(copied['wake_log_dir'].glob('*/receipt.json'))
             newest = max((p.stat().st_mtime for p in all_receipts), default=0)
             if candidates or [p for p in all_receipts if p.stat().st_mtime == newest] != [wake]:
-                raise ValueError('Stopped failure is not the last returned wake')
+                raise ValueError('Stopped source is not the last returned wake')
             candidates = [wake]
         if candidates != [wake]:
             raise ValueError('Stopped source has no unique returned wake after its Host attempt')
@@ -147,12 +170,18 @@ def prepare_stopped_recovery(client, *, container_id: str, declaration: dict,
             raise ValueError('Stopped source lacks an exact interrupted budget receipt')
         start = int(declaration['wake_turn_id'].split('-')[1]) / 1_000_000_000
         phase = float(copied['phase_clock'].read_text())
-        if not math.isfinite(phase) or not start <= min(phase, finished) or (not failed and phase > finished):
+        # A returned stop proves the process was reaped before its receipt.
+        # Legacy workers could reserve cleanup time before the hard deadline;
+        # do not enlarge that native Turn's window to a later container stop.
+        process_stopped = math.floor(wake.stat().st_mtime) if returned_budget_stop else finished
+        if (not math.isfinite(phase) or not start <= min(phase, process_stopped)
+                or process_stopped > finished
+                or (not failed and not returned_budget_stop and phase > finished)):
             raise ValueError('Stopped source phase did not expire during the original wake')
         result = {'container_id': box.id, 'source_state': before, 'declaration': declaration,
                   'components': components, 'scheduler_state_sha256': hashlib.sha256(state_raw).hexdigest(),
                   'receipt_sha256': hashlib.sha256(raw).hexdigest(),
-                  'process_started_at': start, 'process_stopped_at': finished,
+                  'process_started_at': start, 'process_stopped_at': process_stopped,
                   'source_phase_deadline_epoch': phase, 'session': receipt.get('session'),
                   'execution_authority_granted': False}
         box.reload()

benchmark/runtime/interruption_recovery.py

@@ -208,7 +208,12 @@ def read_interrupted_host(state: dict, *, state_file: Path, env: dict[str, str],
         phase = float(source['source_phase_deadline_epoch'] if legacy else receipt["phase_deadline_epoch"])
     except (KeyError, TypeError, ValueError) as error:
         raise ValueError("Original Host phase clock is unavailable") from error
-    if not math.isfinite(phase) or not started <= phase <= stopped:
+    returned_legacy_stop = source is not None and prior.get('status') == 'stopped'
+    if (returned_legacy_stop
+            and stopped != math.floor(path.stat().st_mtime)):
+        raise ValueError('Returned legacy Host window differs from its receipt')
+    if (not math.isfinite(phase) or not started <= phase
+            or (not returned_legacy_stop and phase > stopped)):
         raise ValueError("Original Host stopped outside its phase deadline")
     session = receipt.get("session") or {}
     if session.get("binding_scope") != "agent" or not session.get("session_id"):

benchmark/tests/test_interruption_recovery.py

@@ -82,13 +82,18 @@ def test_live_process_group_cannot_be_settled():
 
 
 @pytest.mark.skipif(os.name != "posix", reason="Owned process-group cleanup requires POSIX")
-@pytest.mark.parametrize("fault", [None, "profile", "stop", "wrong_session", "duplicate_wake"])
-def test_real_sqlite_scheduler_retains_interrupted_session_then_admits_next_turn(tmp_path, monkeypatch, fault):
+@pytest.mark.parametrize("fault", [None, "profile", "stop", "wrong_session", "duplicate_wake", "late_native"])
+@pytest.mark.parametrize('legacy_returned', [False, True])
+def test_real_sqlite_scheduler_retains_interrupted_session_then_admits_next_turn(tmp_path, monkeypatch, fault, legacy_returned):
     from benchmark.runtime import worker
     from benchmark.tests.test_worker_phase_budget import _phase_worker
     from tests import test_external_scheduler_worker as scheduler_fixture
 
     env, project, _runtime, _source, goal_state, _clock = _phase_worker(tmp_path, monkeypatch)
+    # The installed worker keeps its native home outside the submitted workspace.
+    # Match that layout: the home also contains a shared-skills symlink which is
+    # intentionally not a stopped-source participant.
+    env['LOOPX_CODEX_HOME'] = str(tmp_path / 'native-home')
     monkeypatch.setattr(worker, "time", time)
     binary = Path(env["CODEX_BIN"])
     binary.write_text(f"#!{sys.executable}\n" + '''
@@ -149,6 +154,44 @@ def interrupted(*args):
     assert first["host_invoked"] and first["budget_exhausted"]
     assert first["host_attempt_id"] == old["host_attempt"]["attempt_id"]
     assert first["native_process"]["after_cleanup"]["return_code"] is not None
+    if fault == 'late_native':
+        history = project / 'native-history.json'
+        value = json.loads(history.read_bytes())
+        value['completedAt'] = int(time.time()) + 60
+        history.write_text(json.dumps(value))
+    if legacy_returned:
+        from datetime import datetime, timezone
+        from benchmark.edgebench.stopped_recovery import prepare_stopped_recovery
+        from benchmark.tests.test_stopped_recovery import _fake_container
+
+        # Model the previous writer's actual persisted shape. These synthetic
+        # source bytes, including the returned stop, must remain unchanged.
+        old['host_attempt'].pop('attempt_id')
+        old['host_attempt']['session_id'] = None
+        original.write_text(json.dumps(old))
+        old_bytes = original.read_bytes()
+        for key in ('host_attempt_id', 'native_process', 'goal_id', 'agent_id'):
+            first.pop(key, None)
+        receipts[0].write_text(json.dumps(first))
+        returned_at = time.time()
+        os.utime(receipts[0], (returned_at, returned_at))
+        os.utime(original, (returned_at + .1, returned_at + .1))
+        home = Path(env['LOOPX_CODEX_HOME'])
+        for name in ('state_5.sqlite', 'thread_history_1.sqlite'):
+            (home / name).write_text('synthetic native observer state')
+        (home / 'sessions').mkdir(exist_ok=True)
+        (home / 'sessions/original.jsonl').write_text('{}\n')
+        phase_clock = tmp_path / 'original-phase-deadline'
+        phase_clock.write_text(os.environ['LOOPX_PHASE_DEADLINE_EPOCH'])
+        declaration = dict(goal_id=env['LOOPX_GOAL_ID'], agent_id=env['LOOPX_AGENT_ID'],
+            workspace=str(project), runtime_root=env['LOOPX_RUNTIME_ROOT'], native_home=str(home),
+            registry=env['LOOPX_REGISTRY'], scheduler_state=str(original),
+            wake_log_dir=env['LOOPX_WAKE_LOG_DIR'], phase_clock=str(phase_clock), wake_turn_id=first['turn_id'])
+        box, client = _fake_container(monkeypatch)
+        box.attrs['State']['FinishedAt'] = datetime.now(timezone.utc).isoformat()
+        proof = tmp_path / 'legacy-source-proof.json'
+        prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=proof)
+        monkeypatch.setenv('LOOPX_INTERRUPTED_SOURCE', str(proof))
     deadline = time.time() + 60
     monkeypatch.setenv("LOOPX_PHASE_DEADLINE_EPOCH", str(deadline))
     if fault == "profile":
@@ -171,6 +214,7 @@ def interrupted(*args):
         brief.write_text(json.dumps({'ok': True, 'handoff_text': 'Retained submission old-7 is now restored-7; read the inherited history. Goal and opportunities are unchanged.'}))
         monkeypatch.setenv('LOOPX_RECOVERY_HANDOFF', str(brief))
         monkeypatch.setenv('LOOPX_RECOVERY_HANDOFF_SHA256', hashlib.sha256(brief.read_bytes()).hexdigest())
+    original_receipt = receipts[0].read_bytes()
     new = resume_state_file(str(original), phase_deadline=deadline, agent_id=args.agent_id,
         wake_log_dir=env["LOOPX_WAKE_LOG_DIR"], env=dict(os.environ))
     if fault:
@@ -192,6 +236,7 @@ def interrupted(*args):
     argv = [json.loads(line) for line in (project / "native-argv").read_text().splitlines()]
     assert len(argv) == 2 and "resume" in argv[1] and "original-native" in argv[1]
     assert original.read_bytes() == old_bytes and goal_state.read_bytes() == initial_goal
+    assert receipts[0].read_bytes() == original_receipt
     assert json.loads(Path(new).read_bytes())["recovery_handoff_delivered"] is True
     assert scheduler_fixture.worker.run_worker(args) == 0
     inputs = [json.loads(line) for line in (project / 'native-inputs').read_text().splitlines()]

benchmark/tests/test_stopped_recovery.py

@@ -63,6 +63,73 @@ def archive(container, remote, destination, **kwargs):
     return box, SimpleNamespace(containers=SimpleNamespace(get=lambda identity: box))
 
 
+def _returned_budget_stop(declaration):
+    state = Path(declaration['scheduler_state'])
+    value = json.loads(state.read_bytes())
+    value['host_attempt'].update(status='stopped', stop_exit_code=75, session_id='original')
+    value.update(last_wake_rc=75, last_wake_status='wake_requested_stop', last_wake_failure_kind=None)
+    state.write_text(json.dumps(value))
+    os.utime(state, (111, 111))
+    receipt = Path(declaration['wake_log_dir']) / declaration['wake_turn_id'] / 'receipt.json'
+    value = json.loads(receipt.read_bytes())
+    value['timed_out'] = True
+    receipt.write_text(json.dumps(value))
+    os.utime(receipt, (110, 110))
+    return state, receipt
+
+
+@pytest.mark.parametrize('fault', [None, 'null_prior_session', 'same_mtime', 'cleanup_reserve', 'live', 'new_attempt',
+    'not_stop', 'wrong_exit', 'failure', 'unknown_failure', 'cleared_pending', 'wrong_session',
+    'missing_session', 'wrong_goal', 'wrong_agent', 'not_timed_out', 'not_invoked',
+    'not_budget', 'successful', 'success_exit', 'later_wake', 'tied_wake', 'receipt_after_state',
+    'outside_phase'])
+def test_returned_legacy_budget_stop_requires_its_last_receipt(tmp_path, monkeypatch, fault):
+    declaration = _source(tmp_path)
+    state, receipt = _returned_budget_stop(declaration)
+    value, returned = json.loads(state.read_bytes()), json.loads(receipt.read_bytes())
+    box, client = _fake_container(monkeypatch)
+    if fault == 'null_prior_session': value['host_attempt']['session_id'] = None
+    elif fault == 'live': box.attrs['State'].update(Status='running', Running=True, Pid=42)
+    elif fault == 'new_attempt': value['host_attempt']['attempt_id'] = 'b' * 32
+    elif fault == 'not_stop': value['host_attempt']['status'] = 'unknown'
+    elif fault == 'wrong_exit': value['host_attempt']['stop_exit_code'] = 1
+    elif fault == 'failure': value['host_attempt']['host_failure'] = {'kind': 'quota_exhausted'}
+    elif fault == 'unknown_failure': value['last_wake_failure_kind'] = 'unknown'
+    elif fault == 'cleared_pending': state.with_name(state.name + '.host-attempt').write_text('{"pending":false}')
+    elif fault == 'wrong_session': returned['session']['session_id'] = 'another'
+    elif fault == 'missing_session': returned['session']['session_id'] = ''
+    elif fault == 'wrong_goal': returned['goal_id'] = 'another-goal'
+    elif fault == 'wrong_agent': returned['agent_id'] = 'another-agent'
+    elif fault == 'not_timed_out': returned['timed_out'] = False
+    elif fault == 'not_invoked': returned['host_invoked'] = False
+    elif fault == 'not_budget': returned['budget_exhausted'] = False
+    elif fault == 'successful': returned['ok'] = True
+    elif fault == 'success_exit': returned['return_code'] = 0
+    elif fault == 'outside_phase': Path(declaration['phase_clock']).write_text('90')
+    elif fault == 'cleanup_reserve': Path(declaration['phase_clock']).write_text('130')
+    state.write_text(json.dumps(value)); os.utime(state, (111, 111))
+    receipt.write_text(json.dumps(returned))
+    stamp = 111 if fault == 'same_mtime' else 112 if fault == 'receipt_after_state' else 110
+    os.utime(receipt, (stamp, stamp))
+    if fault in {'later_wake', 'tied_wake'}:
+        other = receipt.parent.with_name('wake-109000000000-aaaaaaaaaaaa') / 'receipt.json'
+        other.parent.mkdir(); other.write_text('{}')
+        stamp = 110.5 if fault == 'later_wake' else 110
+        os.utime(other, (stamp, stamp))
+    original = state.read_bytes(), receipt.read_bytes()
+    output = tmp_path / 'source-proof.json'
+    if fault not in {None, 'null_prior_session', 'same_mtime', 'cleanup_reserve'}:
+        with pytest.raises(ValueError):
+            subject.prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=output)
+        assert not output.exists()
+    else:
+        result = subject.prepare_stopped_recovery(client, container_id=box.id, declaration=declaration, output=output)
+        assert result['session']['session_id'] == 'original'
+        assert result['process_stopped_at'] == stamp
+        assert result['execution_authority_granted'] is False
+    assert (state.read_bytes(), receipt.read_bytes()) == original
+
+
 @pytest.mark.parametrize("fault", [None, "live", "restart_command", "mount", "custom_hostname", "ambiguous_wake"])
 def test_prepare_stopped_source_keeps_original_state_and_requires_unique_return(tmp_path, monkeypatch, fault):
     declaration = _source(tmp_path)
@@ -138,13 +205,16 @@ def test_stopped_recovery_rejects_invalid_archive_timeout(tmp_path, timeout):
 
 
 @pytest.mark.skipif(os.environ.get("LOOPX_EDGEBENCH_DOCKER_SMOKE") != "1", reason="Explicit disposable Docker qualification")
-def test_real_stopped_docker_proof_installation_is_inert_and_read_back(tmp_path, monkeypatch):
+@pytest.mark.parametrize('returned', [False, True])
+def test_real_stopped_docker_proof_installation_is_inert_and_read_back(tmp_path, monkeypatch, returned):
     docker = pytest.importorskip("docker")
     client = docker.from_env(timeout=30)
     box = client.containers.create("python:3.12-slim", ["tail", "-f", "/dev/null"], network_disabled=True)
     local = tmp_path / "fixture"
     local.mkdir()
     declaration = _source(local)
+    if returned:
+        _returned_budget_stop(declaration)
     declaration = {key: value.replace(str(local), "/fixture") for key, value in declaration.items()}
     try:
         package = io.BytesIO()