From b1dc51b4e5ef4e47c2a6723334022135b229cfa4 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 05:28:47 +0800 Subject: [PATCH 1/4] fix(benchmark): preserve periodic capture with one explicit handoff Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/README.md | 22 ++++ benchmark/runtime/checkpoint.py | 61 ++++++++- benchmark/tests/qualify_checkpoint_joint.py | 2 +- .../tests/test_checkpoint_settled_once.py | 121 ++++++++++++++++++ .../tests/test_native_checkpoint_handoff.py | 37 ++++-- 5 files changed, 227 insertions(+), 16 deletions(-) create mode 100644 benchmark/tests/test_checkpoint_settled_once.py diff --git a/benchmark/README.md b/benchmark/README.md index 1e8edc0e35..b13235342a 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -276,6 +276,28 @@ and stride receivers, and another full capture/receiver. Its default remains feedback: they qualify mechanical reconstruction and adoption, not scientific benefit or authorization to stop a live campaign. +To request **one explicit settled-work handoff alongside normal replan capture**, +keep `trigger: "fresh_replan"` and the checkpoint-root `destination`, and add: + +```json +"settled_work_capture_once": { + "destination": "/logs/agent/checkpoints/upgrade-handoff", + "purpose": "Explicit upgrade handoff" +} +``` + +The handoff inherits the declared components and participant flags. It waits for +a successful execute wake with its original settled Turn and unchanged request; +restored-history gates still apply. It records its own `checkpoint_handoff` +result without replacing the pre-replan result. A missed or uncertain attempt +is consumed, including when its artifact index is unavailable. Later work cannot +backfill it. Normal distinct replan captures continue at the configured cadence. +Omit this field to retain the existing behavior. Use it only for a necessary +operator-declared handoff, account for its one additional capture within the +original phase, and do not turn it into per-wake captures or a preparation Todo. +The joint qualifier's `--capture-boundary settled_once` exercises this path with +the same real native session, SQLite state, controller and fresh receivers. + For pre-decision capture at a **fresh replan projected at the start of a heartbeat wake**, add `"trigger": "fresh_replan"` and set `destination` to the trial's `checkpoints` directory itself. The Host preloads that wake's original quota guard diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py index 9d218cdabb..a9c36019c0 100644 --- a/benchmark/runtime/checkpoint.py +++ b/benchmark/runtime/checkpoint.py @@ -44,7 +44,7 @@ def parse_checkpoint_request(raw: bytes) -> dict[str, Any]: """Validate the same explicit declaration before staging and before capture.""" request = json.loads(raw) if (not isinstance(request, dict) or - set(request) - {"destination", "components", "point_metadata", "native_session", "evaluation_opportunity", "loopx_state", "trigger", "controller_handoff"} or + set(request) - {"destination", "components", "point_metadata", "native_session", "evaluation_opportunity", "loopx_state", "trigger", "controller_handoff", "settled_work_capture_once"} or not {"destination", "components", "point_metadata"}.issubset(request)): raise ValueError("checkpoint request requires destination, components and point_metadata") if "native_session" in request and type(request["native_session"]) is not bool: @@ -65,6 +65,18 @@ def parse_checkpoint_request(raw: bytes) -> dict[str, Any]: raise ValueError("checkpoint requires explicit components") if not isinstance(request["point_metadata"], dict): raise ValueError("checkpoint point_metadata must be an object") + handoff = request.get("settled_work_capture_once") + if "settled_work_capture_once" in request: + if (request.get("trigger") != "fresh_replan" or not isinstance(handoff, dict) + or set(handoff) != {"destination", "purpose"} + or not isinstance(handoff["purpose"], str) or not handoff["purpose"].strip()): + raise ValueError("settled-work handoff requires fresh_replan plus destination and explicit purpose") + root = normalize_checkpoint_request_path(request["destination"]) + target = normalize_checkpoint_request_path(handoff["destination"]) + if PurePosixPath(target).parent != PurePosixPath(root): + raise ValueError("settled-work handoff must be a direct child of this trial's checkpoint root") + if "settled_work_handoff" in request["point_metadata"]: + raise ValueError("settled-work handoff metadata must not replace an operator field") return request @@ -139,6 +151,10 @@ def _write_checkpoint_attempt(path: Path, value: dict[str, str]) -> bool: return True +def _completed_attempt_path(destination: Path) -> Path: + return destination.parent / ".completed-attempts" / (hashlib.sha256(destination.name.encode()).hexdigest() + ".json") + + def _mark_completed_capture_attempt(destination: Path, effect_id: str, request_sha256: str, wake_root: Path, turn_id: str) -> bool: """Use the original wake witness when the artifact-side index was unavailable. @@ -174,7 +190,7 @@ def _mark_completed_capture_attempt(destination: Path, effect_id: str, request_s attempts.mkdir(parents=True, exist_ok=True, mode=0o700) if attempts.resolve() != attempts: raise ValueError("completed capture attempt directory must be physical") - marker = attempts / (hashlib.sha256(destination.name.encode()).hexdigest() + ".json") + marker = _completed_attempt_path(destination) return _write_checkpoint_attempt(marker, value) @@ -196,6 +212,9 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st request, raw = {}, None if declaration_error is None and request.get("trigger") != "fresh_replan": return body + if declaration_error is None and request.get("settled_work_capture_once") is not None: + receipt["checkpoint_handoff"] = {"pending": True, "turn_id": receipt["turn_id"], + "request_sha256": hashlib.sha256(raw).hexdigest()} if receipt.get("mode") != "heartbeat" or receipt.get("stage") != "execute": raise ValueError("fresh_replan checkpoint trigger requires a heartbeat execute wake") started = time.monotonic() @@ -320,6 +339,40 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st return preloaded_body +def _capture_declared_settled_handoff(env: dict[str, str], receipt: dict[str, Any]) -> None: + """One explicit handoff, not a periodic capture or a substitute replan point.""" + pending = receipt.get("checkpoint_handoff") + if not isinstance(pending, dict) or pending.get("pending") is not True: + return + if not receipt.get("ok") or receipt.get("stage") != "execute" or receipt.get("budget_exhausted"): + return # No eligible completed boundary; keep the request unconsumed. + result = {"captured": False, "research_point_qualified": False, "phase": "after_settled_work"} + try: + request, raw = read_checkpoint_request(env["LOOPX_CHECKPOINT_REQUEST"]) + if (pending.get("turn_id") != receipt.get("turn_id") + or pending.get("request_sha256") != hashlib.sha256(raw).hexdigest()): + raise ValueError("settled-work handoff request or original Turn changed") + handoff = request["settled_work_capture_once"] + if _completed_attempt_path(Path(handoff["destination"])).exists(): + result["reason"] = "destination_already_attempted_no_replay" + receipt["checkpoint_handoff"] = {"pending": False, **result} + return # A witness of an attempt is not proof of a qualified point. + from .checkpoint_continuation import restored_capture_gate + gate = restored_capture_gate(env, full_history=not request.get("native_session", False)) + if not gate["capture_allowed"]: + result["reason"] = gate["reason"] + else: + effective = {k: v for k, v in request.items() if k not in {"trigger", "settled_work_capture_once"}} + effective.update(destination=handoff["destination"], point_metadata={**request["point_metadata"], + "settled_work_handoff": {"purpose": handoff["purpose"]}}) + completed = {k: v for k, v in receipt.items() if k not in {"checkpoint", "checkpoint_handoff"}} + capture_after_wake(env, completed, supplied_request=(effective, raw), restore_gate=gate) + result.update(completed["checkpoint"]) + except Exception as error: + result.update(reason="capture_failed", error_kind=type(error).__name__, error=str(error)[:500]) + receipt["checkpoint_handoff"] = {"pending": False, **result} + + def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *, supplied_request: tuple[dict[str, Any], bytes] | None = None, pending_guard: Path | None = None, @@ -327,6 +380,8 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *, request_path = env.get("LOOPX_CHECKPOINT_REQUEST") if not request_path: return # Feature-off path performs no additional read or write. + if supplied_request is None: + _capture_declared_settled_handoff(env, receipt) if supplied_request is None and (receipt.get("checkpoint") or {}).get("phase") == "before_replan": return # The after-wake hook cannot replace the pre-decision point. receipt["checkpoint"] = {"captured": False, "research_point_qualified": False} @@ -387,7 +442,7 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *, if identity is None: raise ValueError("checkpoint settlement identity is unavailable") request_sha256 = hashlib.sha256(raw).hexdigest() - if request.get("controller_handoff") and supplied_request is None: + if pending_guard is None and (request.get("controller_handoff") or supplied_request is not None): if not _mark_completed_capture_attempt(destination, identity.effect_id, request_sha256, Path(env["LOOPX_WAKE_LOG_DIR"]), receipt["turn_id"]): receipt["checkpoint"]["reason"] = "destination_already_attempted_no_replay" diff --git a/benchmark/tests/qualify_checkpoint_joint.py b/benchmark/tests/qualify_checkpoint_joint.py index 7767415e26..c52d1e87ce 100644 --- a/benchmark/tests/qualify_checkpoint_joint.py +++ b/benchmark/tests/qualify_checkpoint_joint.py @@ -93,7 +93,7 @@ def main(): parser.add_argument("--image", default="python:3.12-slim") parser.add_argument("--context-delivery", choices=("off", "on"), default="off", help="Qualify the existing opt-in context transport through source and fresh receivers") - parser.add_argument("--capture-boundary", choices=("fresh_replan", "settled_work"), default="fresh_replan", + parser.add_argument("--capture-boundary", choices=("fresh_replan", "settled_work", "settled_once"), default="fresh_replan", help="Qualify the existing automatic pre-replan or explicit settled-wake full capture") parser.add_argument("--reuse-point", type=Path, help="Reuse an already sealed synthetic source; never count its source test as passed") parser.add_argument("--captured-config", type=Path, help="The reused point's matching native fixture configuration") diff --git a/benchmark/tests/test_checkpoint_settled_once.py b/benchmark/tests/test_checkpoint_settled_once.py new file mode 100644 index 0000000000..763c7f4a4d --- /dev/null +++ b/benchmark/tests/test_checkpoint_settled_once.py @@ -0,0 +1,121 @@ +"""An explicit handoff does not change periodic capture or reuse a missed point.""" +import hashlib +import json +from types import SimpleNamespace + +import pytest + +from benchmark.runtime import checkpoint + + +def declaration(tmp_path): + return {"trigger": "fresh_replan", "destination": str(tmp_path / "checkpoints"), + "components": [{"id": "workspace", "source_path": str(tmp_path / "workspace")}], + "point_metadata": {"purpose": "Retained operator metadata"}, + "settled_work_capture_once": {"destination": str(tmp_path / "checkpoints" / "handoff"), + "purpose": "Explicit upgrade handoff"}} + + +@pytest.mark.parametrize("change", [ + {"trigger": None}, {"settled_work_capture_once": None}, + {"settled_work_capture_once": {"destination": "/outside/point", "purpose": "handoff"}}, + {"settled_work_capture_once": {"destination": "/logs/checkpoints/point", "purpose": " "}}, + {"point_metadata": {"settled_work_handoff": "Do not overwrite me"}}, +]) +def test_explicit_handoff_rejects_ambiguous_declarations(tmp_path, change): + with pytest.raises(ValueError): + checkpoint.parse_checkpoint_request(json.dumps({**declaration(tmp_path), **change}).encode()) + + +def test_handoff_inherits_all_declared_participants_without_changing_default(tmp_path): + request = {**declaration(tmp_path), "native_session": True, "loopx_state": True, + "evaluation_opportunity": True, "controller_handoff": True} + assert checkpoint.parse_checkpoint_request(json.dumps(request).encode()) == request + del request["settled_work_capture_once"] + assert checkpoint.parse_checkpoint_request(json.dumps(request).encode()) == request + + +@pytest.fixture +def prepared(tmp_path, monkeypatch): + request = declaration(tmp_path) + path = tmp_path / "request.json" + path.write_text(json.dumps(request)) + env = {"LOOPX_CHECKPOINT_REQUEST": str(path), "LOOPX_WAKE_LOG_DIR": str(tmp_path / "wakes"), + "LOOPX_RUNTIME_ROOT": str(tmp_path / "runtime"), "LOOPX_REGISTRY": str(tmp_path / "registry"), + "LOOPX_GOAL_ID": "source", "LOOPX_AGENT_ID": "agent"} + gate_calls, settlements = [], [] + def gate(*args, **kwargs): + gate_calls.append(kwargs) + return {"capture_allowed": True} + def settled(*args, **kwargs): + settlements.append(kwargs["turn_instance_id"]) + return SimpleNamespace(progress={"state": "settled"}, + identity=SimpleNamespace(value=SimpleNamespace(effect_id=kwargs["turn_instance_id"]))) + monkeypatch.setattr("benchmark.runtime.checkpoint_continuation.restored_capture_gate", gate) + monkeypatch.setattr(checkpoint, "read_heartbeat_settlement", settled) + def receipt(turn="first", **extra): + return {"ok": True, "stage": "execute", "mode": "heartbeat", "turn_id": turn, + "checkpoint": {"phase": "before_replan", "captured": False, "reason": "no_fresh_replan"}, + "checkpoint_handoff": {"pending": True, "turn_id": turn, + "request_sha256": hashlib.sha256(path.read_bytes()).hexdigest()}, **extra} + return request, path, env, receipt, gate_calls, settlements + + +def test_default_and_unsuccessful_wakes_add_no_io_or_settlement(prepared): + _, _, env, receipt, gates, settlements = prepared + for extra in ({"checkpoint_handoff": None}, {"ok": False}, + {"budget_exhausted": True}, {"stage": "plan"}): + current = receipt(**extra) + original = dict(current["checkpoint"]) + checkpoint.capture_after_wake(env, current) + assert current["checkpoint"] == original + assert not gates and not settlements + + +@pytest.mark.parametrize("mutation", ["request", "turn"]) +def test_changed_request_or_turn_cannot_capture(prepared, mutation): + _, path, env, receipt, gates, settlements = prepared + current = receipt() + if mutation == "request": + path.write_bytes(path.read_bytes() + b"\n") + else: + current["turn_id"] = "later" + checkpoint.capture_after_wake(env, current) + assert current["checkpoint_handoff"]["reason"] == "capture_failed" + assert not gates and not settlements + + +def test_failed_attempt_is_consumed_without_later_source_read(prepared, monkeypatch, tmp_path): + request, _, env, receipt, gates, settlements = prepared + request["loopx_state"] = True + path = tmp_path / "request.json" + path.write_text(json.dumps(request)) + attempts = [] + def unavailable(*args): + attempts.append(True) + raise ValueError("Synthetic source unavailable before sealing") + monkeypatch.setattr(checkpoint, "research_participants", unavailable) + current = receipt() + original = dict(current["checkpoint"]) + checkpoint.capture_after_wake(env, current) + assert current["checkpoint_handoff"]["reason"] == "capture_failed" + assert current["ok"] and current["checkpoint"] == original + assert not (tmp_path / "checkpoints" / "handoff").exists() + witness = (tmp_path / "wakes/first/checkpoint-completed-request.json").read_bytes() + for turn in ("first", "later"): + later = receipt(turn) + checkpoint.capture_after_wake(env, later) + assert later["checkpoint_handoff"]["reason"] == "destination_already_attempted_no_replay" + assert later["checkpoint"] == original + assert attempts == [True] and len(gates) == len(settlements) == 1 + assert (tmp_path / "wakes/first/checkpoint-completed-request.json").read_bytes() == witness + + +def test_restore_gate_still_blocks_explicit_handoff(prepared, monkeypatch): + _, _, env, receipt, _, settlements = prepared + monkeypatch.setattr("benchmark.runtime.checkpoint_continuation.restored_capture_gate", + lambda *args, **kwargs: {"capture_allowed": False, "reason": "restored_replan_pending"}) + current = receipt() + checkpoint.capture_after_wake(env, current) + assert current["checkpoint_handoff"]["reason"] == "restored_replan_pending" + assert not settlements diff --git a/benchmark/tests/test_native_checkpoint_handoff.py b/benchmark/tests/test_native_checkpoint_handoff.py index 8f007fe3f9..a527468482 100644 --- a/benchmark/tests/test_native_checkpoint_handoff.py +++ b/benchmark/tests/test_native_checkpoint_handoff.py @@ -369,7 +369,7 @@ def test_midwake_native_profile_refusal_precedes_interruption(tmp_path, monkeypa @pytest.mark.skipif(not os.environ.get("LOOPX_NATIVE_CHECKPOINT_BIN") or os.environ.get("LOOPX_CHECKPOINT_EXTERNAL_CONTROLLER") != "1", reason="Requires the real native CLI and explicitly attached checkpoint supervisor") -@pytest.mark.parametrize("capture_boundary", ["fresh_replan", "settled_work"]) +@pytest.mark.parametrize("capture_boundary", ["fresh_replan", "settled_work", "settled_once"]) @pytest.mark.parametrize("explore_mode", ["off", "planning"]) @pytest.mark.parametrize("context_delivery", ["off", "on"]) def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, explore_mode, context_delivery, capture_boundary): @@ -500,7 +500,7 @@ def response(payload, index, message): "[(Path('candidate.txt').write_text(str(v))," "Path('evidence-'+str(v)+'.json').write_text(json.dumps({'value':v,'synthetic':True}))) for v in values]; " "assert Path('candidate.txt').read_text()==str(values[-1]); print('CANDIDATE_VERIFIED')") - if capture_boundary == "settled_work" and phase["values"] == [6]: + if capture_boundary != "fresh_replan" and phase["values"] == [6]: # The actual native tool settles its original work before the # completed-wake hook. No operator-side settlement is backdated. prefix = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry), @@ -541,7 +541,7 @@ def response(payload, index, message): return {"type": "function_call", "id": "fc_packet", "call_id": "call_packet", "name": "exec_command", "arguments": json.dumps({ "cmd": "cat " + shlex.quote(str(paths[0])), "max_output_tokens": 30000})} - if capture_boundary == "settled_work" and phase["kind"] == "work" and phase["values"] == [6]: + if capture_boundary != "fresh_replan" and phase["kind"] == "work" and phase["values"] == [6]: record_counterexample() message["content"][0]["text"] = "Original packet read." if phase["kind"] == "review" else "Synthetic work verified." return message @@ -551,7 +551,7 @@ def settle_work(result, expected): assert candidate.read_text() == str(expected) guard = json.loads((project / "work-guard.json").read_text()) assert guard["heartbeat_receipt"]["turn_instance_id"] == result["turn_id"] - if capture_boundary == "settled_work" and expected == 6: + if capture_boundary != "fresh_replan" and expected == 6: from loopx.control_plane.quota.settlement import read_heartbeat_settlement observed = read_heartbeat_settlement(runtime, goal_id=cli.GOAL_ID, agent_id=cli.AGENT_ID, todo_id=cli.TODO_ID, turn_instance_id=result["turn_id"], registry_path=registry) @@ -582,12 +582,16 @@ def settle_work(result, expected): "native_session": True, "loopx_state": True, "evaluation_opportunity": True, "controller_handoff": True} if capture_boundary == "fresh_replan": declaration["trigger"] = "fresh_replan" + elif capture_boundary == "settled_once": + declaration.update(trigger="fresh_replan", settled_work_capture_once={ + "destination": str(logs.parent / "checkpoints" / "settled-work"), + "purpose": "Synthetic explicit source handoff"}) else: declaration["destination"] += "/settled-work" request.write_text(json.dumps(declaration)) for value in range(1, 7): phase.update(kind="work", values=[value], tool_sent=False, context_started=False) - actual = {**env, "LOOPX_CHECKPOINT_REQUEST": str(request)} if capture_boundary == "settled_work" and value == 6 else env + actual = {**env, "LOOPX_CHECKPOINT_REQUEST": str(request)} if capture_boundary != "fresh_replan" and value == 6 else env prior = run_once(actual) settle_work(prior, value) assert len(set(work_turns)) == 6 @@ -597,19 +601,28 @@ def settle_work(result, expected): assert reader.deliveries[0]["mode"] == "full" assert any(row["mode"] == "delta" for row in reader.deliveries[1:]) review_request_index = len(requests) - if capture_boundary == "settled_work": - assert prior["checkpoint"]["captured"], prior - assert not prior["checkpoint"]["research_point_qualified"] + if capture_boundary != "fresh_replan": + captured = prior["checkpoint_handoff"] if capture_boundary == "settled_once" else prior["checkpoint"] + assert captured["captured"], prior + assert not captured["research_point_qualified"] assert cli._heartbeat_receipt_count(runtime, prior["turn_id"]) == 1 - assert not guards and not list(logs.glob("*/checkpoint-start-guard.json")) + assert not guards + assert len(list(logs.glob("*/checkpoint-start-guard.json"))) == (1 if capture_boundary == "settled_once" else 0) + if capture_boundary == "settled_once": + assert prior["checkpoint"]["reason"] == "no_fresh_replan" + assert captured["phase"] == "after_settled_work" + assert configured["after"]["execution_profile"]["replan_after_effective_turns"] == 6 assert not prior.get("checkpoint_handoffs") and "turn_envelope" not in prior from benchmark.runtime.checkpoint import capture_after_wake replay = dict(prior) call_count = len(calls) capture_after_wake(actual, replay) - assert replay["checkpoint"]["reason"] == "destination_already_attempted_no_replay" - assert len(calls) == call_count and prior["checkpoint"]["captured"] - (tmp_path / "controller-result.json").write_text(json.dumps({**prior["checkpoint"], + if capture_boundary == "settled_once": + assert replay == prior # Completed handoff is not retried, even on repeated host closeout. + else: + assert replay["checkpoint"]["reason"] == "destination_already_attempted_no_replay" + assert len(calls) == call_count and captured["captured"] + (tmp_path / "controller-result.json").write_text(json.dumps({**captured, "natural_work_turns": work_turns, "capture_boundary": capture_boundary, "context_delivery": context_delivery, "context_reads": reader.deliveries})) return From 9fd7c14cc44d5d66776241d332fb1de7375ce3d2 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 05:38:49 +0800 Subject: [PATCH 2/4] test(checkpoint): decode current native context in joint qualification Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/tests/test_native_checkpoint_handoff.py | 5 ++--- tests/control_plane/test_research_checkpoint_cli.py | 5 +++-- tests/control_plane/test_research_context_adoption.py | 3 ++- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/benchmark/tests/test_native_checkpoint_handoff.py b/benchmark/tests/test_native_checkpoint_handoff.py index a527468482..eaf6eced68 100644 --- a/benchmark/tests/test_native_checkpoint_handoff.py +++ b/benchmark/tests/test_native_checkpoint_handoff.py @@ -13,7 +13,7 @@ import pytest from loopx.hosts.codex_checkpoint import native_session_checkpoint -from tests.control_plane.codex_checkpoint_fixture import real_native_source +from tests.control_plane.codex_checkpoint_fixture import real_native_source, read_cli_policy_payload from loopx.capabilities.benchmark_toolkit.native_codex_goal import ( NativeGoalProtocolError, StdioNativeGoalTransport, ) @@ -469,8 +469,7 @@ def response(payload, index, message): if phase["kind"] == "work" and not phase["tool_sent"]: # Only after complete native input delivery, inspect the # original decision as an independent authority oracle. - source = Path(reader.deliveries[-1]["recovery"]["path"]).with_name("decision.json") - current = json.loads(source.read_text()) + current = read_cli_policy_payload(json.dumps(reader.deliveries[-1]), returncode=0) assert not current.get("autonomous_replan_obligation") assert current["interaction_contract"]["agent_channel"]["delivery_allowed"] (project / "work-guard.json").write_text(json.dumps(current)) diff --git a/tests/control_plane/test_research_checkpoint_cli.py b/tests/control_plane/test_research_checkpoint_cli.py index f76942d9b4..cc0cefae20 100644 --- a/tests/control_plane/test_research_checkpoint_cli.py +++ b/tests/control_plane/test_research_checkpoint_cli.py @@ -13,6 +13,7 @@ from loopx.cli import main from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession from loopx.control_plane.turn_driver.codex_sessions import _store_codex_cli_session +from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload from tests.control_plane.test_research_context_adoption import ( _ordinary_review_and_work, no_benchmark_dependency, # noqa: F401 -- fixture is active in this module ) @@ -29,7 +30,7 @@ def test_explicit_ordinary_goal_capture_preserves_context_without_activation(tmp prefix = ['--format', 'json', '--registry', str(registry), '--runtime-root', str(runtime)] def cli(*args, ok=True): code = main([*prefix, *args]) - value = json.loads(capsys.readouterr().out) + value = read_cli_policy_payload(capsys.readouterr().out, returncode=code) assert code == (0 if ok else 1), value assert value['ok'] is ok return value @@ -148,7 +149,7 @@ def native(env, session_id, **kwargs): branch_prefix = ['--format', 'json', '--registry', str(branch_registry), '--runtime-root', str(branch_root)] def branch(*args): code = main([*branch_prefix, *args]) - result = json.loads(capsys.readouterr().out) + result = read_cli_policy_payload(capsys.readouterr().out, returncode=code) assert code == 0, result return result branch('bootstrap', '--project', str(workspace), '--goal-id', 'branch', '--objective', 'Continue retained experiments.', diff --git a/tests/control_plane/test_research_context_adoption.py b/tests/control_plane/test_research_context_adoption.py index 8fcc072230..c5158145b6 100644 --- a/tests/control_plane/test_research_context_adoption.py +++ b/tests/control_plane/test_research_context_adoption.py @@ -15,6 +15,7 @@ from loopx.hosts.research_checkpoint import canonical_context, research_context, verify_context_sources, adopt_todo_context, _encoded from canonical_authority_fixture import assert_todo_read_unchanged, isolate_sqlite_runtime from loopx.control_plane.turn_driver.codex_sessions import _store_codex_cli_session +from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload @pytest.fixture(autouse=True) @@ -45,7 +46,7 @@ def run(*args): code = main([*prefix,*args]) output = capsys.readouterr().out assert code == 0, output - return json.loads(output) + return read_cli_policy_payload(output, returncode=code) run('bootstrap','--project',str(workspace),'--goal-id',name,'--objective','Investigate synthetic candidates.', '--adapter-kind','read_only_project_map_v0','--adapter-status','connected-read-only','--write-scope','**','--no-global-sync') run('configure-goal','--goal-id',name,'--registered-agent',agent,'--agent-work-mode',agent+'=active', From 35b404daa673c9a35baeafb100deea6850cceccd Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 05:45:29 +0800 Subject: [PATCH 3/4] test(checkpoint): read native context in fresh receiver qualification Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/tests/test_checkpoint_joint_receiver.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/benchmark/tests/test_checkpoint_joint_receiver.py b/benchmark/tests/test_checkpoint_joint_receiver.py index 83ee843925..8595eac57d 100644 --- a/benchmark/tests/test_checkpoint_joint_receiver.py +++ b/benchmark/tests/test_checkpoint_joint_receiver.py @@ -40,6 +40,7 @@ def test_captured_point_continues_with_fresh_authority( from benchmark.runtime.sessions import BenchmarkSessionWake from benchmark.runtime.worker import native_command from loopx.cli import main + from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload from loopx.control_plane.effect_runtime import ( effect_runtime_result, EffectRuntimeRemoteError, @@ -123,7 +124,7 @@ def cli(*args): code = main([*prefix, *args]) output = capsys.readouterr().out assert code == 0, output - return json.loads(output) + return read_cli_policy_payload(output, returncode=code) cli( "bootstrap", From 08dbb5e7077959a644960a64fd00605df5eea98f Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 11:32:37 +0800 Subject: [PATCH 4/4] fix(checkpoint): retain missed handoffs within the original phase Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- benchmark/runtime/checkpoint.py | 101 +++++++++++------- .../tests/test_checkpoint_settled_once.py | 99 ++++++++++++++++- 2 files changed, 162 insertions(+), 38 deletions(-) diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py index a9c36019c0..677aed262c 100644 --- a/benchmark/runtime/checkpoint.py +++ b/benchmark/runtime/checkpoint.py @@ -214,6 +214,7 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st return body if declaration_error is None and request.get("settled_work_capture_once") is not None: receipt["checkpoint_handoff"] = {"pending": True, "turn_id": receipt["turn_id"], + "destination": request["settled_work_capture_once"]["destination"], "request_sha256": hashlib.sha256(raw).hexdigest()} if receipt.get("mode") != "heartbeat" or receipt.get("stage") != "execute": raise ValueError("fresh_replan checkpoint trigger requires a heartbeat execute wake") @@ -347,40 +348,28 @@ def _capture_declared_settled_handoff(env: dict[str, str], receipt: dict[str, An if not receipt.get("ok") or receipt.get("stage") != "execute" or receipt.get("budget_exhausted"): return # No eligible completed boundary; keep the request unconsumed. result = {"captured": False, "research_point_qualified": False, "phase": "after_settled_work"} - try: - request, raw = read_checkpoint_request(env["LOOPX_CHECKPOINT_REQUEST"]) - if (pending.get("turn_id") != receipt.get("turn_id") - or pending.get("request_sha256") != hashlib.sha256(raw).hexdigest()): - raise ValueError("settled-work handoff request or original Turn changed") - handoff = request["settled_work_capture_once"] - if _completed_attempt_path(Path(handoff["destination"])).exists(): - result["reason"] = "destination_already_attempted_no_replay" - receipt["checkpoint_handoff"] = {"pending": False, **result} - return # A witness of an attempt is not proof of a qualified point. - from .checkpoint_continuation import restored_capture_gate - gate = restored_capture_gate(env, full_history=not request.get("native_session", False)) - if not gate["capture_allowed"]: - result["reason"] = gate["reason"] - else: - effective = {k: v for k, v in request.items() if k not in {"trigger", "settled_work_capture_once"}} - effective.update(destination=handoff["destination"], point_metadata={**request["point_metadata"], - "settled_work_handoff": {"purpose": handoff["purpose"]}}) - completed = {k: v for k, v in receipt.items() if k not in {"checkpoint", "checkpoint_handoff"}} - capture_after_wake(env, completed, supplied_request=(effective, raw), restore_gate=gate) - result.update(completed["checkpoint"]) - except Exception as error: - result.update(reason="capture_failed", error_kind=type(error).__name__, error=str(error)[:500]) + completed = {k: v for k, v in receipt.items() if k not in {"checkpoint", "checkpoint_handoff"}} + capture_after_wake(env, completed, settled_handoff=pending) + result.update(completed["checkpoint"]) receipt["checkpoint_handoff"] = {"pending": False, **result} +def _capture_destination(root: Path, value: str) -> Path: + destination = Path(value) + if not destination.is_absolute() or destination != destination.resolve() or destination.parent != root: + raise ValueError("checkpoint destination must be a normalized direct child of this trial's checkpoints directory") + return destination + + def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *, supplied_request: tuple[dict[str, Any], bytes] | None = None, pending_guard: Path | None = None, - restore_gate: dict[str, Any] | None = None) -> None: + restore_gate: dict[str, Any] | None = None, + settled_handoff: dict[str, Any] | None = None) -> None: request_path = env.get("LOOPX_CHECKPOINT_REQUEST") if not request_path: return # Feature-off path performs no additional read or write. - if supplied_request is None: + if supplied_request is None and settled_handoff is None: _capture_declared_settled_handoff(env, receipt) if supplied_request is None and (receipt.get("checkpoint") or {}).get("phase") == "before_replan": return # The after-wake hook cannot replace the pre-decision point. @@ -401,10 +390,51 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *, if original_deadline <= capture_started: receipt["checkpoint"]["reason"] = "capture_window_closed" return + root = Path(env["LOOPX_WAKE_LOG_DIR"]).parent / "checkpoints" + observed = None + if settled_handoff is not None: + expected_hash = settled_handoff.get("request_sha256") + if (settled_handoff.get("turn_id") != receipt.get("turn_id") + or not isinstance(expected_hash, str) or len(expected_hash) != 64): + raise ValueError("settled-work handoff request or original Turn changed") + destination = _capture_destination(root, settled_handoff["destination"]) + if _completed_attempt_path(destination).exists(): + receipt["checkpoint"]["reason"] = "destination_already_attempted_no_replay" + return + observed = read_heartbeat_settlement( + Path(env["LOOPX_RUNTIME_ROOT"]), goal_id=env["LOOPX_GOAL_ID"], + agent_id=env["LOOPX_AGENT_ID"], todo_id=None, + turn_instance_id=receipt["turn_id"], resolve_original_binding=True, + registry_path=Path(env["LOOPX_REGISTRY"]), + ) + if observed is None or (observed.progress or {}).get("state") != "settled": + receipt["checkpoint"]["reason"] = "owned_turn_not_settled" + return + identity = observed.identity.value + if identity is None: + raise ValueError("checkpoint settlement identity is unavailable") + # Reserve the preflight's exact point before rereading a declaration + # that may have disappeared. The existing wake witness survives an + # unavailable artifact index; later work must not backfill this point. + if not _mark_completed_capture_attempt(destination, identity.effect_id, expected_hash, + Path(env["LOOPX_WAKE_LOG_DIR"]), receipt["turn_id"]): + receipt["checkpoint"]["reason"] = "destination_already_attempted_no_replay" + return + if original_deadline is not None and original_deadline <= time.time(): + receipt["checkpoint"]["reason"] = "capture_window_closed" + return if supplied_request is None: request, raw = read_checkpoint_request(request_path) else: request, raw = supplied_request + if settled_handoff is not None: + handoff = request["settled_work_capture_once"] + if (hashlib.sha256(raw).hexdigest() != expected_hash + or handoff["destination"] != str(destination)): + raise ValueError("settled-work handoff request or original Turn changed") + request = {k: v for k, v in request.items() if k not in {"trigger", "settled_work_capture_once"}} + request.update(destination=handoff["destination"], point_metadata={**request["point_metadata"], + "settled_work_handoff": {"purpose": handoff["purpose"]}}) window = CHECKPOINT_CAPTURE_SECONDS if request.get("controller_handoff") else 30 deadline = capture_started + window if original_deadline is not None: @@ -422,19 +452,17 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *, receipt["checkpoint"]["reason"] = "pre_replan_hook_required" return # This is a private evidence artifact below the runner's log owner. - destination = Path(request["destination"]) - root = Path(env["LOOPX_WAKE_LOG_DIR"]).parent / "checkpoints" - if not destination.is_absolute() or destination != destination.resolve() or destination.parent != root: - raise ValueError("checkpoint destination must be a normalized direct child of this trial's checkpoints directory") + destination = _capture_destination(root, request["destination"]) if destination.exists(): receipt["checkpoint"]["reason"] = "destination_already_present_no_replay" return - observed = read_heartbeat_settlement( - Path(env["LOOPX_RUNTIME_ROOT"]), goal_id=env["LOOPX_GOAL_ID"], - agent_id=env["LOOPX_AGENT_ID"], todo_id=None, - turn_instance_id=receipt["turn_id"], resolve_original_binding=True, - registry_path=Path(env["LOOPX_REGISTRY"]), - ) + if observed is None: + observed = read_heartbeat_settlement( + Path(env["LOOPX_RUNTIME_ROOT"]), goal_id=env["LOOPX_GOAL_ID"], + agent_id=env["LOOPX_AGENT_ID"], todo_id=None, + turn_instance_id=receipt["turn_id"], resolve_original_binding=True, + registry_path=Path(env["LOOPX_REGISTRY"]), + ) if observed is None or (observed.progress or {}).get("state") != "settled": receipt["checkpoint"]["reason"] = "owned_turn_not_settled" return @@ -442,7 +470,8 @@ def capture_after_wake(env: dict[str, str], receipt: dict[str, Any], *, if identity is None: raise ValueError("checkpoint settlement identity is unavailable") request_sha256 = hashlib.sha256(raw).hexdigest() - if pending_guard is None and (request.get("controller_handoff") or supplied_request is not None): + if (settled_handoff is None and pending_guard is None + and (request.get("controller_handoff") or supplied_request is not None)): if not _mark_completed_capture_attempt(destination, identity.effect_id, request_sha256, Path(env["LOOPX_WAKE_LOG_DIR"]), receipt["turn_id"]): receipt["checkpoint"]["reason"] = "destination_already_attempted_no_replay" diff --git a/benchmark/tests/test_checkpoint_settled_once.py b/benchmark/tests/test_checkpoint_settled_once.py index 763c7f4a4d..645b7aa97e 100644 --- a/benchmark/tests/test_checkpoint_settled_once.py +++ b/benchmark/tests/test_checkpoint_settled_once.py @@ -57,6 +57,7 @@ def receipt(turn="first", **extra): return {"ok": True, "stage": "execute", "mode": "heartbeat", "turn_id": turn, "checkpoint": {"phase": "before_replan", "captured": False, "reason": "no_fresh_replan"}, "checkpoint_handoff": {"pending": True, "turn_id": turn, + "destination": request["settled_work_capture_once"]["destination"], "request_sha256": hashlib.sha256(path.read_bytes()).hexdigest()}, **extra} return request, path, env, receipt, gate_calls, settlements @@ -82,7 +83,8 @@ def test_changed_request_or_turn_cannot_capture(prepared, mutation): current["turn_id"] = "later" checkpoint.capture_after_wake(env, current) assert current["checkpoint_handoff"]["reason"] == "capture_failed" - assert not gates and not settlements + assert not gates + assert settlements == (["first"] if mutation == "request" else []) def test_failed_attempt_is_consumed_without_later_source_read(prepared, monkeypatch, tmp_path): @@ -118,4 +120,97 @@ def test_restore_gate_still_blocks_explicit_handoff(prepared, monkeypatch): current = receipt() checkpoint.capture_after_wake(env, current) assert current["checkpoint_handoff"]["reason"] == "restored_replan_pending" - assert not settlements + assert settlements == ["first"] + + +def test_unsettled_work_does_not_consume_a_handoff(prepared, monkeypatch, tmp_path): + _, _, env, receipt, gates, _ = prepared + monkeypatch.setattr(checkpoint, "read_heartbeat_settlement", lambda *args, **kwargs: None) + reads = [] + monkeypatch.setattr(checkpoint, "read_checkpoint_request", lambda *args: reads.append(True)) + current = receipt() + checkpoint.capture_after_wake(env, current) + assert current["checkpoint_handoff"]["reason"] == "owned_turn_not_settled" + assert not reads and not gates + assert not (tmp_path / "wakes").exists() + + +def test_handoff_without_original_destination_fails_before_declaration_io(prepared, monkeypatch): + _, _, env, receipt, gates, settlements = prepared + current = receipt() + del current["checkpoint_handoff"]["destination"] + reads = [] + monkeypatch.setattr(checkpoint, "read_checkpoint_request", lambda *args: reads.append(True)) + checkpoint.capture_after_wake(env, current) + assert current["checkpoint_handoff"]["reason"] == "capture_failed" + assert not reads and not gates and not settlements + + +@pytest.mark.parametrize("mutation", ["missing", "changed"]) +def test_lost_declaration_cannot_backfill_the_original_point(prepared, tmp_path, mutation): + _, path, env, receipt, _, _ = prepared + workspace = tmp_path / "workspace" + workspace.mkdir() + source = workspace / "candidate.txt" + source.write_text("original settled candidate") + raw = path.read_bytes() + first = receipt() + original_periodic = dict(first["checkpoint"]) + if mutation == "missing": + path.unlink() + else: + path.write_bytes(raw + b"\n") + checkpoint.capture_after_wake(env, first) + assert first["checkpoint_handoff"]["reason"] == "capture_failed" + assert first["ok"] and first["checkpoint"] == original_periodic + + # Restoring the same declaration must not label subsequent work as the + # missed point, even when real artifact probe/capture would now succeed. + path.write_bytes(raw) + source.write_text("later candidate must not replace the missed point") + later = receipt("later") + checkpoint.capture_after_wake(env, later) + assert later["checkpoint_handoff"].get("captured") is False + assert later["checkpoint_handoff"]["reason"] == "destination_already_attempted_no_replay" + assert later["checkpoint"] == original_periodic and later["ok"] + assert not (tmp_path / "checkpoints" / "handoff").exists() + witness = json.loads((tmp_path / "wakes/first/checkpoint-completed-request.json").read_text()) + assert witness["request_sha256"] == hashlib.sha256(raw).hexdigest() + assert witness["settlement_effect_id"] == "first" + + +@pytest.mark.parametrize("deadline", ["1", "nan", "inf"]) +def test_closed_original_phase_adds_no_handoff_reads(prepared, monkeypatch, deadline): + _, _, env, receipt, gates, settlements = prepared + current = receipt() + original_periodic = dict(current["checkpoint"]) + env["LOOPX_PHASE_DEADLINE_EPOCH"] = deadline + reads = [] + def forbidden_read(*args): + reads.append(True) + raise AssertionError("closed phase must not read a declaration") + monkeypatch.setattr(checkpoint, "read_checkpoint_request", forbidden_read) + checkpoint.capture_after_wake(env, current) + assert not reads and not gates and not settlements + expected = "capture_window_closed" if deadline == "1" else "capture_failed" + assert current["checkpoint_handoff"]["reason"] == expected + assert current["checkpoint"] == original_periodic and current["ok"] + + +def test_phase_expiring_during_settlement_read_stops_before_declaration(prepared, monkeypatch): + _, _, env, receipt, gates, settlements = prepared + current = receipt() + env["LOOPX_PHASE_DEADLINE_EPOCH"] = "2" + now = [1.0] + monkeypatch.setattr(checkpoint.time, "time", lambda: now[0]) + read_settlement = checkpoint.read_heartbeat_settlement + def settle_then_expire(*args, **kwargs): + result = read_settlement(*args, **kwargs) + now[0] = 3.0 + return result + monkeypatch.setattr(checkpoint, "read_heartbeat_settlement", settle_then_expire) + reads = [] + monkeypatch.setattr(checkpoint, "read_checkpoint_request", lambda *args: reads.append(True)) + checkpoint.capture_after_wake(env, current) + assert current["checkpoint_handoff"]["reason"] == "capture_window_closed" + assert settlements == ["first"] and not reads and not gates