diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 7d018f2b1d..65ebdd8b4d 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -371,6 +371,15 @@ heartbeat/turn scheduler path, that receipt exits the wake with code 75 and the configured shell worker stops normally. It does not re-admit empty wakes, mark the task complete or spend quota. Direct plain/native-Goal calls retain exit 0 for normal budget exhaustion. + +An explicitly continued Harbor phase or SForge warm restore may select a new +shell transport record after that budget stop. The retained worker receipt must +prove the stop happened before native dispatch. The original stop record and IO +witness remain intact; the finite phase deadline names the successor record, so +retries in that same phase retain its failures. Missing proof, interrupted native +work, provider failures and other stops keep the original recovery path. This +does not reopen a terminal LoopX decision or change Goal/Agent, Todos, native +conversation, scoring history or checkpoint frequency. The deadline uses the task environment's clock, including remote Harbor backends. Budget exhaustion is distinct from task completion: this clock does not override provider failures, Stop, blocked planning, or a valid terminal LoopX decision. diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index 2b49ac1945..6ecd140fee 100644 --- a/benchmark/runtime/harbor.py +++ b/benchmark/runtime/harbor.py @@ -851,6 +851,7 @@ async def run( command = worker_command( env, python=f"{_PYTHON}/bin/python3", source=_SRC, state_file=_SCHEDULER_STATE, host_timeout=host_timeout, + resume_budget_stop=self._phase_number > 1, ) phase_log = f"/logs/agent/worker-phase-{self._phase_number:03d}.log" shell = ( diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index 37f9546931..822cab9f99 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -1,6 +1,15 @@ """The existing scheduler invocation shared by native benchmark transports.""" +import argparse +from contextlib import ExitStack +import hashlib +import json +import math +import os +from pathlib import Path +import re import shlex +import time # Private process transport between the benchmark wake and its shell scheduler. # The benchmark worker remains the owner of phase-budget exhaustion. @@ -13,11 +22,167 @@ def phase_work_reserve(mode: str) -> int: return 0 if mode == "heartbeat" else PHASE_WORK_RESERVE_SECONDS +def _physical_state(path: Path) -> Path: + if path.is_symlink() or path.resolve() != path: + raise ValueError("Scheduler phase state must be a physical path") + return path + + +def _phase_states(root: Path) -> list[Path]: + states = [root, *sorted(root.parent.glob(f"{root.stem}.phase-*{root.suffix}"))] + return [_physical_state(path) for path in states if path.exists()] + + +def _state_object(path: Path) -> dict | None: + try: + value = json.loads(path.read_text()) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + return value if isinstance(value, dict) else None + + +def _latest_phase_state(root: Path, states: list[Path]) -> Path: + """Follow explicit resume links, then use mtime only for legacy phase files.""" + documents = {path: state for path in states if (state := _state_object(path)) is not None} + children: dict[Path, list[Path]] = {} + for path, state in documents.items(): + parent = state.get("resume_from") + if not isinstance(parent, str): + continue + parent_path = Path(parent) + if (not parent_path.is_absolute() or parent_path.resolve() != parent_path + or parent_path not in documents): + continue + children.setdefault(parent_path, []).append(path) + current = root if root in documents else max(states, key=lambda path: path.stat().st_mtime_ns) + seen = set() + while current not in seen and children.get(current): + seen.add(current) + current = max(children[current], key=lambda path: path.stat().st_mtime_ns) + unreadable = [path for path in states if path not in documents] + if unreadable: + newest = max(unreadable, key=lambda path: path.stat().st_mtime_ns) + if newest.stat().st_mtime_ns > current.stat().st_mtime_ns: + current = newest # Unreadable current evidence must be held, not skipped. + return current + + +def _phase_receipt(state_file: Path, wake_log_dir: str | None) -> dict | None: + if wake_log_dir is None: + return None + # The scheduler writes its current stop after the worker receipt. Later + # phases share this log directory, so the state mtime remains the fence. + stop_time = state_file.stat().st_mtime_ns + receipts = [] + for receipt in Path(wake_log_dir).glob("*/receipt.json"): + if receipt.is_symlink() or receipt.resolve() != receipt: + raise ValueError("Source wake receipt must be a physical file") + if receipt.stat().st_mtime_ns <= stop_time: + receipts.append(receipt) + if not receipts: + return None + latest = max(receipts, key=lambda item: item.stat().st_mtime_ns) + try: + value = json.loads(latest.read_text()) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + return value if isinstance(value, dict) else None + + +def _create_resume_state(target: Path, predecessor: Path) -> None: + if target.is_symlink(): + raise ValueError("Resumed scheduler state must not be a symlink") + try: + with target.open("x", encoding="utf-8") as stream: + json.dump({"resume_from": str(predecessor)}, stream) + except FileExistsError: + return + target.chmod(0o600) + + +def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, + wake_log_dir: str | None = None) -> str: + """Select a new transport phase only after an explicit budget continuation. + + Kernel state and native sessions stay in place. The original scheduler + record and its independent IO witness are retained, including failures. + The caller owns authorization and the unchanged finite phase deadline. + """ + if (isinstance(phase_deadline, bool) or not isinstance(phase_deadline, (int, float)) + or not math.isfinite(phase_deadline) or phase_deadline <= time.time()): + raise ValueError("Budget continuation requires an open finite phase deadline") + path = _physical_state(Path(state_file).absolute()) + from loopx.file_lock import try_exclusive_file_lock + from scripts.external_scheduler_worker import HOST_LOCK_OPERATION, _host_attempt, _host_pending + + with ExitStack() as locks: + # The root serializes phase selection. Locking the existing phase files + # as well keeps a running/failed phase from moving under that decision. + if locks.enter_context(try_exclusive_file_lock( + path, agent_id=agent_id, operation=HOST_LOCK_OPERATION)) is None: + return str(path) + for phase in _phase_states(path): + if phase == path: + continue + if locks.enter_context(try_exclusive_file_lock( + phase, agent_id=agent_id, operation=HOST_LOCK_OPERATION)) is None: + return str(phase) + states = _phase_states(path) + if not states: + return str(path) + predecessor = _latest_phase_state(path, states) + state = _state_object(predecessor) + if state is None: + return str(predecessor) + prior = state.get("host_attempt") + if not isinstance(prior, dict) or prior.get("status") != "stopped": + return str(predecessor) + # Validate the captured record with its existing transport owner. + scope = prior.get("scope") + if not isinstance(scope, str) or not re.fullmatch(r"[0-9a-f]{64}", scope): + return str(predecessor) + try: + _host_attempt(state, scope) + except (TypeError, ValueError): + return str(predecessor) + if (prior.get("stop_exit_code") != BUDGET_EXHAUSTED_EXIT_CODE + or "host_failure" in prior or _host_pending(predecessor) is None + or state.get("last_wake_rc") != BUDGET_EXHAUSTED_EXIT_CODE + or state.get("last_wake_status") != "wake_requested_stop"): + return str(predecessor) + # Only the immutable phase clock names the new state. Todo/reset-token + # changes cannot clear a failure already recorded in this new phase. + suffix = hashlib.sha256(float(phase_deadline).hex().encode()).hexdigest()[:20] + target = path.with_name(f"{path.stem}.phase-{suffix}{path.suffix}") + if target.is_symlink(): + raise ValueError("Resumed scheduler state must not be a symlink") + if target.exists(): + return str(target) # Never replace a retry/failure in this phase. + # Only a stop before native dispatch admits this shortcut. An + # interrupted native Turn still needs its original recovery/settlement. + # Restrict legacy receipt discovery to the source scheduler snapshot; + # later phases share the log directory but cannot replace this proof. + result = _phase_receipt(predecessor, wake_log_dir) + if (not isinstance(result, dict) or result.get("budget_exhausted") is not True + or result.get("host_invoked") is not False or result.get("mode") != "heartbeat" + or result.get("stage") != "execute"): + return str(predecessor) + _create_resume_state(target, predecessor) + return str(target) + + def worker_command(env: dict[str, str], *, python: str, source: str, - state_file: str, host_timeout: float) -> list[str]: + state_file: str, host_timeout: float, + resume_budget_stop: bool = False) -> list[str]: + if type(resume_budget_stop) is not bool: + raise ValueError("resume_budget_stop must be a boolean") wake = [python, "-m", "benchmark.runtime.worker"] if env["LOOPX_EXECUTION_MODE"] not in {"heartbeat", "turn"}: return wake + if resume_budget_stop: + return [python, "-m", "benchmark.runtime.scheduler", "--python", python, + "--source", source, "--state-file", state_file, + "--host-timeout", str(host_timeout)] return [ python, f"{source}/scripts/external_scheduler_worker.py", "--cli-bin", env["LOOPX_CLI"], "--registry", env["LOOPX_REGISTRY"], @@ -30,3 +195,23 @@ def worker_command(env: dict[str, str], *, python: str, source: str, "--quota-timeout-seconds", "30", "--error-backoff-seconds", "15", *(["--wake-host-failure"] if env["LOOPX_EXECUTION_MODE"] == "heartbeat" else []), ] + + +def main() -> int: + parser = argparse.ArgumentParser(description="Resume one explicitly continued benchmark phase.") + parser.add_argument("--python", required=True) + parser.add_argument("--source", required=True) + parser.add_argument("--state-file", required=True) + parser.add_argument("--host-timeout", required=True, type=float) + args = parser.parse_args() + state = resume_state_file(args.state_file, + phase_deadline=float(os.environ["LOOPX_PHASE_DEADLINE_EPOCH"]), + agent_id=os.environ["LOOPX_AGENT_ID"], wake_log_dir=os.environ.get("LOOPX_WAKE_LOG_DIR")) + command = worker_command(dict(os.environ), python=args.python, source=args.source, + state_file=state, host_timeout=args.host_timeout) + os.execvpe(command[0], command, os.environ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmark/runtime/sforge.py b/benchmark/runtime/sforge.py index 82df81dc0b..cdbf3048bc 100644 --- a/benchmark/runtime/sforge.py +++ b/benchmark/runtime/sforge.py @@ -480,7 +480,8 @@ def format_run_cmd(self, prompt_path, *, model=None, cwd="", internet=True, resu if self.warm_start is not None: env["LOOPX_WARM_START"] = f"{_CONTROL}/warm-start.json" command = worker_command(env, python=f"{_PYTHON}/bin/python3", source=_SRC, - state_file=_SCHEDULER_STATE, host_timeout=self.turn_timeout) + state_file=_SCHEDULER_STATE, host_timeout=self.turn_timeout, + resume_budget_stop=self.warm_start is not None) if self.task_entry == "loopx-planned" and self.warm_start is None: env["LOOPX_PLANNING_RESULT"] = "/opt/loopx-benchmark/control/planning-phase-001.json" command = [f"{_PYTHON}/bin/python3", "-m", "benchmark.runtime.sforge_entry", *command] diff --git a/benchmark/tests/test_scheduler_phase_resume.py b/benchmark/tests/test_scheduler_phase_resume.py new file mode 100644 index 0000000000..bd633cb8b1 --- /dev/null +++ b/benchmark/tests/test_scheduler_phase_resume.py @@ -0,0 +1,191 @@ +"""An authorized new phase preserves old stops and current conversation state.""" +from __future__ import annotations + +import json +from pathlib import Path +import shlex +import sys +import time + +import pytest + +from benchmark.runtime.scheduler import resume_state_file, worker_command + + +def _stopped(path): + receipt = path.parent / "wakes" / path.stem / "receipt.json" + receipt.parent.mkdir(parents=True, exist_ok=True) + receipt.write_text(json.dumps(dict(budget_exhausted=True, host_invoked=False, + mode="heartbeat", stage="execute"))) + state = json.loads(path.read_text()) if path.exists() else {} + state.update(host_attempt=dict(status="stopped", scope="a" * 64, + attempt=1, stop_exit_code=75), last_wake_rc=75, last_wake_status="wake_requested_stop") + path.write_text(json.dumps(state)) + path.with_name(path.name + ".host-attempt").write_text('{"pending":false}') + + +def _resume(path, end): + return Path(resume_state_file(str(path), phase_deadline=end, agent_id="agent", + wake_log_dir=str(path.parent / "wakes"))) + + +def test_budget_phase_retries_keep_the_same_state_and_original_receipts(tmp_path): + old = tmp_path / "scheduler.json" + _stopped(old) + before = old.read_bytes() + witness = old.with_name(old.name + ".host-attempt").read_bytes() + end = time.time() + 1000 + new = resume_state_file(str(old), phase_deadline=end, agent_id="agent", wake_log_dir=str(tmp_path / "wakes")) + assert new != str(old) and Path(new).parent == old.parent + Path(new).write_text('{"host_attempt":{"status":"failed"}}') + assert resume_state_file(str(old), phase_deadline=end, agent_id="agent", wake_log_dir=str(tmp_path / "wakes")) == new + state = json.loads(old.read_bytes()) + state["reset_token"] = "new-todo-observation" + old.write_text(json.dumps(state)) + assert resume_state_file(str(old), phase_deadline=end, agent_id="agent", wake_log_dir=str(tmp_path / "wakes")) == new + assert json.loads(before)["host_attempt"] == json.loads(old.read_bytes())["host_attempt"] + assert old.with_name(old.name + ".host-attempt").read_bytes() == witness + assert json.loads(Path(new).read_bytes())["host_attempt"]["status"] == "failed" + + +def test_repeated_budget_stops_advance_from_the_actual_predecessor(tmp_path): + root = tmp_path / "scheduler.json" + _stopped(root) + second = _resume(root, time.time() + 1000) + _stopped(second) + third = _resume(root, time.time() + 2000) + assert third not in {root, second} + assert json.loads(third.read_text())["resume_from"] == str(second) + + +@pytest.mark.parametrize("change", ["hard_failure", "unknown", "started", "unreadable", "interrupted"]) +def test_later_phase_cannot_skip_the_latest_host_attempt(tmp_path, change): + from loopx.control_plane.turn_driver.host_failure import build_host_failure_record + + root = tmp_path / "scheduler.json" + _stopped(root) + second = _resume(root, time.time() + 1000) + _stopped(second) + state = json.loads(second.read_text()) + if change == "hard_failure": + state["host_attempt"] = { + "scope": "a" * 64, "status": "failed", "attempt": 1, + "host_failure": build_host_failure_record("quota_exhausted", attempt=1), + "not_before": time.time() + 100, + } + state["last_wake_rc"] = 1 + state["last_wake_status"] = "wake_failed" + elif change == "unknown": + state["host_attempt"] = { + "scope": "a" * 64, "status": "failed", "attempt": 1, + "host_failure": build_host_failure_record("unknown", attempt=1), + "not_before": time.time() + 100, + } + state["last_wake_rc"] = 1 + state["last_wake_status"] = "wake_failed" + elif change == "started": + state["host_attempt"]["status"] = "started" + elif change == "unreadable": + second.write_text("broken-state") + else: + receipt = tmp_path / "wakes" / second.stem / "receipt.json" + receipt.write_text(json.dumps(dict(budget_exhausted=True, host_invoked=True, + mode="heartbeat", stage="execute"))) + second.touch() + if change != "unreadable": + second.write_text(json.dumps(state)) + before = second.read_bytes() + later = _resume(root, time.time() + 2000) + assert later == second + assert second.read_bytes() == before + assert not list(tmp_path.glob("scheduler.phase-*.json"))[1:] + + +@pytest.mark.parametrize("change", ["started", "provider_failure", "other_stop", "interrupted", "unknown_witness"]) +def test_phase_continuation_does_not_clear_host_recovery_or_other_stops(tmp_path, change): + old = tmp_path / "scheduler.json" + _stopped(old) + state = json.loads(old.read_bytes()) + if change == "started": + state["host_attempt"]["status"] = "started" + elif change == "provider_failure": + state["host_attempt"].update(status="failed", host_failure={"kind": "unknown"}) + elif change == "other_stop": + state["host_attempt"]["stop_exit_code"] = 76 + elif change == "interrupted": + receipt = tmp_path / "wakes" / old.stem / "receipt.json" + receipt.write_text(json.dumps(dict(budget_exhausted=True, host_invoked=True, + mode="heartbeat", stage="execute"))) + # The stopped transport snapshot follows this interrupted receipt. + old.touch() + else: + old.with_name(old.name + ".host-attempt").write_text("incomplete") + old.write_text(json.dumps(state)) + before = old.read_bytes() + assert resume_state_file(str(old), phase_deadline=time.time() + 1000, agent_id="agent", wake_log_dir=str(tmp_path / "wakes")) == str(old) + assert old.read_bytes() == before and not list(tmp_path.glob("*.phase-*.json")) + + +@pytest.mark.parametrize("end", [0, float("inf"), float("nan"), True]) +def test_phase_selection_cannot_grant_an_expired_or_nonfinite_allowance(tmp_path, end): + old = tmp_path / "scheduler.json" + _stopped(old) + before = old.read_bytes() + with pytest.raises(ValueError, match="finite phase"): + resume_state_file(str(old), phase_deadline=end, agent_id="agent", wake_log_dir=str(tmp_path / "wakes")) + assert old.read_bytes() == before + + +def test_real_scheduler_resumes_original_native_session_after_budget_stop(tmp_path, monkeypatch): + from benchmark.runtime import worker + from benchmark.tests.test_worker_phase_budget import _phase_worker + from tests import test_external_scheduler_worker as fixture + + env, project, _runtime, _source, goal_state, _clock = _phase_worker( + tmp_path, monkeypatch, work_seconds=300, output="success") + original_goal = goal_state.read_bytes() + first = worker.run_once(env) + assert first["ok"] and first["session"]["session_id"] == "original-native" + entry = tmp_path / "native_worker.py" + entry.write_text("from benchmark.runtime import worker\n" + "worker.heartbeat_body=lambda *a,**kw:'synthetic work'\n" + "raise SystemExit(worker.main())\n") + for key, value in env.items(): + monkeypatch.setenv(key, value) + args, _ = fixture._guarded_host(tmp_path / "scheduler") + args.once = True + args.registry = env["LOOPX_REGISTRY"] + args.runtime_root = env["LOOPX_RUNTIME_ROOT"] + args.goal_id = env["LOOPX_GOAL_ID"] + args.agent_id = env["LOOPX_AGENT_ID"] + args.wake_cmd = shlex.join([sys.executable, str(entry)]) + args.wake_stop_exit_code = 75 + args.wake_timeout_seconds = None + monkeypatch.setattr(fixture.worker, "run_quota_should_run", + lambda *a, **kw: fixture._hint_payload(should_run=True)) + monkeypatch.setenv("LOOPX_PHASE_DEADLINE_EPOCH", "1") + assert fixture.worker.run_worker(args) == 0 # Actual worker returns budget stop. + old = Path(args.state_file) + end = time.time() + 1000 + monkeypatch.setenv("LOOPX_PHASE_DEADLINE_EPOCH", str(end)) + assert fixture.worker.run_worker(args) == 0 # Same old phase still cannot dispatch. + old_bytes = old.read_bytes() + args.state_file = resume_state_file(str(old), phase_deadline=end, agent_id="agent", wake_log_dir=env["LOOPX_WAKE_LOG_DIR"]) + assert fixture.worker.run_worker(args) == 0 + assert (project / "native-starts").read_text().splitlines() == ["native", "native"] + receipts = [json.loads(path.read_bytes()) for path in Path(env["LOOPX_WAKE_LOG_DIR"]).glob("*/receipt.json")] + invoked = [row for row in receipts if row.get("host_invoked")] + assert len(invoked) == 2 and all(row["session"]["session_id"] == "original-native" for row in invoked) + assert old.read_bytes() == old_bytes and goal_state.read_bytes() == original_goal + + +def test_only_explicit_continuation_wraps_the_existing_scheduler(): + env = dict(LOOPX_EXECUTION_MODE="heartbeat", LOOPX_CLI="/cli", LOOPX_REGISTRY="/registry", + LOOPX_RUNTIME_ROOT="/runtime", LOOPX_GOAL_ID="goal", LOOPX_AGENT_ID="agent") + options = dict(python="/python", source="/source", state_file="/state", host_timeout=1000) + ordinary = worker_command(env, **options) + assert ordinary[1] == "/source/scripts/external_scheduler_worker.py" + continued = worker_command(env, **options, resume_budget_stop=True) + assert continued[1:3] == ["-m", "benchmark.runtime.scheduler"] + assert worker_command({**env, "LOOPX_EXECUTION_MODE": "native-goal"}, **options, + resume_budget_stop=True) == ["/python", "-m", "benchmark.runtime.worker"]