From 23fe65c58830978e0d90de15a1537983eda4f8cd Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 07:00:56 +0800 Subject: [PATCH] fix(benchmark): qualify returned legacy budget interruptions Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> (cherry picked from commit b7f28a6096f6c3b76606776b2b8371b752b1eab5) Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/README.md | 12 ++++ benchmark/edgebench/stopped_recovery.py | 41 +++++++++-- benchmark/runtime/interruption_recovery.py | 7 +- benchmark/tests/test_interruption_recovery.py | 49 ++++++++++++- benchmark/tests/test_stopped_recovery.py | 72 ++++++++++++++++++- 5 files changed, 171 insertions(+), 10 deletions(-) diff --git a/benchmark/README.md b/benchmark/README.md index 2d78d9b708..9a00944151 100644 --- a/benchmark/README.md +++ b/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 diff --git a/benchmark/edgebench/stopped_recovery.py b/benchmark/edgebench/stopped_recovery.py index 2fa5f49587..779fe299d4 100644 --- a/benchmark/edgebench/stopped_recovery.py +++ b/benchmark/edgebench/stopped_recovery.py @@ -108,7 +108,8 @@ 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' @@ -116,6 +117,26 @@ 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 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() diff --git a/benchmark/runtime/interruption_recovery.py b/benchmark/runtime/interruption_recovery.py index 377a664af4..43042cea8f 100644 --- a/benchmark/runtime/interruption_recovery.py +++ b/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"): diff --git a/benchmark/tests/test_interruption_recovery.py b/benchmark/tests/test_interruption_recovery.py index 7d5ebc6f59..1d76b52ce8 100644 --- a/benchmark/tests/test_interruption_recovery.py +++ b/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()] diff --git a/benchmark/tests/test_stopped_recovery.py b/benchmark/tests/test_stopped_recovery.py index fc4049b0fc..2b4862a06d 100644 --- a/benchmark/tests/test_stopped_recovery.py +++ b/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()