From b8cee72004360c0833eb44e33de665b66474d69b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 6 Nov 2026 06:40:50 +0800 Subject: [PATCH 1/2] Resume explicitly continued benchmark phases after a no-host budget stop Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/RUNTIME.md | 9 ++ benchmark/runtime/harbor.py | 1 + benchmark/runtime/scheduler.py | 111 ++++++++++++++- benchmark/runtime/sforge.py | 3 +- .../tests/test_scheduler_phase_resume.py | 131 ++++++++++++++++++ 5 files changed, 253 insertions(+), 2 deletions(-) create mode 100644 benchmark/tests/test_scheduler_phase_resume.py diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 0612e59cf7..fb6074252e 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -365,6 +365,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. To compare entry policies, hold the execution mode, session policy, model, diff --git a/benchmark/runtime/harbor.py b/benchmark/runtime/harbor.py index 8b8597418a..72dcf9b974 100644 --- a/benchmark/runtime/harbor.py +++ b/benchmark/runtime/harbor.py @@ -846,6 +846,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 79e63e645b..068aeb5bd1 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -1,17 +1,106 @@ """The existing scheduler invocation shared by native benchmark transports.""" +import argparse +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. BUDGET_EXHAUSTED_EXIT_CODE = 75 +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 = Path(state_file).absolute() + if path.resolve() != path or path.is_symlink(): + raise ValueError("Scheduler phase state must be a physical path") + from loopx.file_lock import try_exclusive_file_lock + from scripts.external_scheduler_worker import HOST_LOCK_OPERATION, _host_attempt, _host_pending + + with try_exclusive_file_lock(path, agent_id=agent_id, operation=HOST_LOCK_OPERATION) as held: + if held is None: + return str(path) # The ordinary scheduler retains its mutex behavior. + try: + state = json.loads(path.read_text()) + except FileNotFoundError: + return str(path) + if not isinstance(state, dict): + raise ValueError("Scheduler phase state is invalid") + prior = state.get("host_attempt") + if not isinstance(prior, dict) or prior.get("status") != "stopped": + return str(path) + # 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): + raise ValueError("Captured scheduler scope is invalid") + _host_attempt(state, scope) + if (prior.get("stop_exit_code") != BUDGET_EXHAUSTED_EXIT_CODE + or "host_failure" in prior or _host_pending(path) is None + or state.get("last_wake_rc") != BUDGET_EXHAUSTED_EXIT_CODE + or state.get("last_wake_status") != "wake_requested_stop"): + return str(path) + # 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. + if wake_log_dir is None: + return str(path) + # 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. + stop_time = path.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 str(path) + latest = max(receipts, key=lambda item: item.stat().st_mtime_ns) + try: + result = json.loads(latest.read_text()) + except (OSError, UnicodeError, json.JSONDecodeError): + return str(path) + 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(path) + 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"], @@ -24,3 +113,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 c3b3a325c0..0ea6ca6ccc 100644 --- a/benchmark/runtime/sforge.py +++ b/benchmark/runtime/sforge.py @@ -481,7 +481,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..6e17bb86e0 --- /dev/null +++ b/benchmark/tests/test_scheduler_phase_resume.py @@ -0,0 +1,131 @@ +"""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" / "source" / "receipt.json" + receipt.parent.mkdir(parents=True) + receipt.write_text(json.dumps(dict(budget_exhausted=True, host_invoked=False, + mode="heartbeat", stage="execute"))) + path.write_text(json.dumps(dict(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.with_name(path.name + ".host-attempt").write_text('{"pending":false}') + + +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" + + +@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" / "source" / "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"] From 8ea4b737cdfe4f0c606272711c8ec51e6b6c4666 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 6 Nov 2026 21:36:15 +0800 Subject: [PATCH 2/2] fix(benchmark): continue only from the latest phase attempt Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/scheduler.py | 142 ++++++++++++++---- .../tests/test_scheduler_phase_resume.py | 70 ++++++++- 2 files changed, 174 insertions(+), 38 deletions(-) diff --git a/benchmark/runtime/scheduler.py b/benchmark/runtime/scheduler.py index 068aeb5bd1..bfdb2e7b97 100644 --- a/benchmark/runtime/scheduler.py +++ b/benchmark/runtime/scheduler.py @@ -1,6 +1,7 @@ """The existing scheduler invocation shared by native benchmark transports.""" import argparse +from contextlib import ExitStack import hashlib import json import math @@ -15,6 +16,84 @@ BUDGET_EXHAUSTED_EXIT_CODE = 75 +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. @@ -26,34 +105,45 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, 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 = Path(state_file).absolute() - if path.resolve() != path or path.is_symlink(): - raise ValueError("Scheduler phase state must be a physical path") + 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 try_exclusive_file_lock(path, agent_id=agent_id, operation=HOST_LOCK_OPERATION) as held: - if held is None: - return str(path) # The ordinary scheduler retains its mutex behavior. - try: - state = json.loads(path.read_text()) - except FileNotFoundError: + 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) - if not isinstance(state, dict): - raise ValueError("Scheduler phase state is invalid") + 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(path) + 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): - raise ValueError("Captured scheduler scope is invalid") - _host_attempt(state, 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(path) is None + 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(path) + 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] @@ -62,30 +152,16 @@ def resume_state_file(state_file: str, *, phase_deadline: float, agent_id: str, raise ValueError("Resumed scheduler state must not be a symlink") if target.exists(): return str(target) # Never replace a retry/failure in this phase. - if wake_log_dir is None: - return str(path) # 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. - stop_time = path.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 str(path) - latest = max(receipts, key=lambda item: item.stat().st_mtime_ns) - try: - result = json.loads(latest.read_text()) - except (OSError, UnicodeError, json.JSONDecodeError): - return str(path) + 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(path) + return str(predecessor) + _create_resume_state(target, predecessor) return str(target) diff --git a/benchmark/tests/test_scheduler_phase_resume.py b/benchmark/tests/test_scheduler_phase_resume.py index 6e17bb86e0..bd633cb8b1 100644 --- a/benchmark/tests/test_scheduler_phase_resume.py +++ b/benchmark/tests/test_scheduler_phase_resume.py @@ -13,15 +13,22 @@ def _stopped(path): - receipt = path.parent / "wakes" / "source" / "receipt.json" - receipt.parent.mkdir(parents=True) + 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"))) - path.write_text(json.dumps(dict(host_attempt=dict(status="stopped", scope="a" * 64, - attempt=1, stop_exit_code=75), last_wake_rc=75, last_wake_status="wake_requested_stop"))) + 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) @@ -41,6 +48,59 @@ def test_budget_phase_retries_keep_the_same_state_and_original_receipts(tmp_path 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" @@ -53,7 +113,7 @@ def test_phase_continuation_does_not_clear_host_recovery_or_other_stops(tmp_path elif change == "other_stop": state["host_attempt"]["stop_exit_code"] = 76 elif change == "interrupted": - receipt = tmp_path / "wakes" / "source" / "receipt.json" + 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.