diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 150e95f8b4..5657ba18f6 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -170,8 +170,11 @@ native projector present full/delta for an uninterrupted tool. A refused/lost release remains the original typed transport failure, without publishing an optimized result or rerunning the guard. An interrupted guard continues with the original full packet; no delta is required for checkpoint recovery. Native -cursor verification also binds the context-directory/root option. The existing -Darwin native-checkpoint refusal remains in force. +cursor verification binds the context-directory/root option and the worker's +original `--complete-turn` and `--turn-envelope` choices. A checkpoint without +a complete-mode binding cannot verify a pending command that uses +`--complete-turn`. Historical ordinary and envelope guards remain readable. +The existing Darwin native-checkpoint refusal remains in force. The host-side checkpoint preflight captures the complete original decision. With context delivery enabled, its prompt supplies `quota context-project` diff --git a/benchmark/runtime/checkpoint_wake.py b/benchmark/runtime/checkpoint_wake.py index dd62124ac9..127d3151fc 100644 --- a/benchmark/runtime/checkpoint_wake.py +++ b/benchmark/runtime/checkpoint_wake.py @@ -136,6 +136,7 @@ def _capture(self, held, deadline): "goal_id": self.env["LOOPX_GOAL_ID"], "agent_id": self.env["LOOPX_AGENT_ID"], "turn_instance_id": held["quota"]["heartbeat_receipt"]["turn_instance_id"], "turn_envelope": self.receipt.get("turn_envelope", False), + "complete_turn": self.receipt.get("complete_turn", False), "capture_root": str(self.capture_root), "socket_path": self.socket_path, "decision_sha256": held["sha256"]} if self.env.get("LOOPX_CODEX_CONTEXT_DIR"): diff --git a/benchmark/tests/test_native_checkpoint.py b/benchmark/tests/test_native_checkpoint.py index 7975fe6d17..0dc9528cfe 100644 --- a/benchmark/tests/test_native_checkpoint.py +++ b/benchmark/tests/test_native_checkpoint.py @@ -881,9 +881,11 @@ def forbidden_fork(*args, **kwargs): assert settled.progress["state"] == "settled" -@pytest.mark.parametrize("turn_envelope", [None, False, True], ids=["legacy", "ordinary", "envelope"]) +@pytest.mark.parametrize("turn_envelope,complete_turn", [(None, None), (False, False), (True, False), (False, True)], + ids=["legacy", "ordinary", "envelope", "complete"]) @pytest.mark.parametrize("context_option", [None, "codex_context_dir", "codex_context_root"]) -def test_interrupted_cursor_requires_exact_unreleased_final_guard(tmp_path, monkeypatch, turn_envelope, context_option): +def test_interrupted_cursor_requires_exact_unreleased_final_guard( + tmp_path, monkeypatch, turn_envelope, complete_turn, context_option): """A stopped command is eligible only at the declared pre-output boundary.""" import copy import shlex @@ -901,6 +903,10 @@ def test_interrupted_cursor_requires_exact_unreleased_final_guard(tmp_path, monk guard["turn_envelope"] = turn_envelope if turn_envelope is False: argv.remove("--turn-envelope") + if complete_turn is not None: + guard["complete_turn"] = complete_turn + if complete_turn: + argv.append("--complete-turn") if context_option: guard[context_option] = "/fixture/context" argv += ["--" + context_option.replace("_", "-"), guard[context_option]] @@ -914,7 +920,7 @@ def test_interrupted_cursor_requires_exact_unreleased_final_guard(tmp_path, monk captured = native_session_checkpoint(env, "fixture-session", timeout=10, interrupted_guard=guard) assert captured["metadata"]["turn_ids"] == ["retained-compaction", "pending"] assert "completed_turn_id" not in captured["metadata"] - assert captured["metadata"]["interrupted_guard"]["native_call_id"] == "guard-call" + assert captured["metadata"]["interrupted_guard"] == {**guard, "native_call_id": "guard-call"} assert "thread/fork" not in calls rendered = copy.deepcopy(turns) @@ -941,14 +947,33 @@ def test_interrupted_cursor_requires_exact_unreleased_final_guard(tmp_path, monk if "--turn-envelope" in opposite: opposite.remove("--turn-envelope") else: + if "--complete-turn" in opposite: + opposite.remove("--complete-turn") opposite.append("--turn-envelope") changed = copy.deepcopy(turns) changed[-1]["items"][-1]["command"] = shlex.join(["/bin/bash", "-lc", shlex.join(opposite)]) with pytest.raises(ValueError, match="routes or identity"): _verify_interrupted_guard(changed, guard) + if turn_envelope is False: + # Complete and ordinary commands have the same old envelope flag but + # different settlement routes. Neither may stand in for the other. + alternate = [word for word in argv if word != "--complete-turn"] if complete_turn else [*argv, "--complete-turn"] + switched = copy.deepcopy(turns) + switched[-1]["items"][-1]["command"] = shlex.join(["/bin/bash", "-lc", shlex.join(alternate)]) + with pytest.raises(ValueError, match="routes or identity"): + _verify_interrupted_guard(switched, guard) + # A previous invocation of the other mode is not this held command. + switched[-1]["items"].append(item) + _verify_interrupted_guard(switched, guard) + if complete_turn: + with pytest.raises(ValueError, match="incompatible"): + _verify_interrupted_guard(turns, {**guard, "turn_envelope": True}) + with pytest.raises(ValueError, match="routes or identity"): + _verify_interrupted_guard(turns, {k: v for k, v in guard.items() if k != "complete_turn"}) for invalid in [None, 0, 1, "false"]: - with pytest.raises(ValueError, match="must be boolean"): - _verify_interrupted_guard(turns, {**guard, "turn_envelope": invalid}) + for flag in ("turn_envelope", "complete_turn"): + with pytest.raises(ValueError, match="must be boolean"): + _verify_interrupted_guard(turns, {**guard, flag: invalid}) variants = [] for option in ("--codex-context-dir", "--codex-context-root"): altered = list(argv) @@ -988,3 +1013,108 @@ def test_interrupted_cursor_requires_exact_unreleased_final_guard(tmp_path, monk _verify_interrupted_guard(changed, guard) with pytest.raises(ValueError, match="call identity"): _verify_interrupted_guard(turns, {**guard, "native_call_id": "different"}) + + +@pytest.mark.skipif(os.name != "posix", reason="Exercises the private Host Unix rendezvous") +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("mode_matches", [True, False], ids=["retained", "switched"]) +def test_held_complete_guard_checkpoint_binds_original_settlement_mode(tmp_path, monkeypatch, provider, mode_matches): + """Exercise the worker, real quota guard and checkpoint store with fixed native history.""" + from copy import deepcopy + from pathlib import Path + import shlex + import subprocess + import sys + + from benchmark.runtime.checkpoint import held_quota_decision + from benchmark.runtime.checkpoint_wake import CheckpointWake + from benchmark.runtime.codex import turn_envelope_receipt + from benchmark.runtime.worker import heartbeat_body + from loopx.control_plane.effect_runtime import effect_runtime_result + 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 loopx.hosts.codex_checkpoint import _guard_cli_arguments + from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime + from tests.control_plane.test_quota_settlement_cli import ( + AGENT_ID, GOAL_ID, TURN_ID, _configure_autonomous_replan_fixture, _heartbeat_receipt_count, + _initialize_fixture_authority, _spend_run_count, _write_fixture, + ) + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path) + _configure_autonomous_replan_fixture(project, runtime, registry) + _initialize_fixture_authority(project, runtime, provider=provider) + code = project / "candidate.txt" + code.write_text("synthetic candidate\n") + source = Path(sys.modules["loopx"].__file__).resolve().parent.parent + cli = source / "scripts/loopx" + pages = [] + native, calls = fixture(project, monkeypatch, pages) + declaration = tmp_path / "request.json" + declaration.write_text(json.dumps({ + "destination": str(tmp_path / "checkpoints"), "trigger": "fresh_replan", + "point_metadata": {}, "native_session": True, + "components": [{"id": "candidate", "source_path": str(code)}], + })) + env = {**os.environ, **native, "CODEX_BIN": sys.executable, "MODEL_NAME": "fixture", + "REASONING_EFFORT": "high", "LOOPX_CODEX_SANDBOX": "read-only", + "LOOPX_PYTHON": sys.executable, "PYTHONPATH": str(source), + "LOOPX_REGISTRY": str(registry), "LOOPX_RUNTIME_ROOT": str(runtime), "LOOPX_CLI": str(cli), + "LOOPX_GOAL_ID": GOAL_ID, "LOOPX_AGENT_ID": AGENT_ID, + "LOOPX_WAKE_LOG_DIR": str(tmp_path / "wakes"), "LOOPX_CHECKPOINT_REQUEST": str(declaration), + "LOOPX_CONTEXT_DELIVERY": "off", "LOOPX_TURN_ENVELOPE": "0", + "LOOPX_COMPLETE_TURN": "1"} + session = CodexAgentSession(env, context="resume", sandbox="read-only", receipt={}) + _store_codex_cli_session(runtime, lineage=session.lineage, session_scope="agent", + session_id="fixture-session", goal_ref=session.goal_ref, session_profile_digest=session.digest) + receipt = {"mode": "heartbeat", "stage": "execute", "turn_id": TURN_ID, + "session": {"session_id": "fixture-session"}, + **turn_envelope_receipt(explicit=False, complete=True)} + (tmp_path / "wakes" / TURN_ID).mkdir(parents=True) + + with CheckpointWake(env, receipt) as host: + body = heartbeat_body(env, TURN_ID, native_goal=False, decision_handoff_socket=host.socket_path) + block = body.split("```sh\n", 1)[1].split("\n```", 1)[0] + command = shlex.join(["/bin/bash", "-lc", block]) + original = _guard_cli_arguments(command, {"cli": str(cli), "turn_instance_id": TURN_ID}) + assert "--complete-turn" in original and "--turn-envelope" not in original + observed = original if mode_matches else [word for word in original if word != "--complete-turn"] + item = {"type": "commandExecution", "id": "pending-guard", "cwd": str(project), + "command": shlex.join(["/bin/bash", "-lc", shlex.join([str(cli), *observed])]), + "status": "failed", "exitCode": 137, "aggregatedOutput": None} + history = {"data": [{"id": "interrupted-turn", "status": "interrupted", "items": [item]}], + "nextCursor": None} + pages.extend((history, deepcopy(history))) + host.listener.settimeout(30) + with subprocess.Popen([str(cli), *original], cwd=project, env=env, + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) as command_process: + with held_quota_decision(host.listener, host.capture_root, goal_id=GOAL_ID, + agent_id=AGENT_ID, turn_instance_id=TURN_ID, + deadline=time.monotonic() + 30) as held: + assert held["quota"]["ok"] is True and held["intent"]["capture"] is True + quota = held["quota"]["heartbeat_receipt"] + assert quota["turn_instance_id"] == TURN_ID + digest = hashlib.sha256(held["path"].read_bytes()).hexdigest() + assert digest == held["sha256"] + point = host.root / held["intent"]["obligation_id"] + if mode_matches: + captured = host._capture(held, time.monotonic() + 30) + result = effect_runtime_result("runtime.checkpoint_artifacts", { + "action": "verify", "root": str(point), "expected_sha256": captured["sha256"], + }) + assert result["sha256"] == captured["sha256"] + manifest = json.loads((point / "checkpoint.json").read_bytes())["point_metadata"] + stored = manifest["native_session"]["interrupted_guard"] + assert stored["complete_turn"] is True and stored["turn_envelope"] is False + assert stored["turn_instance_id"] == TURN_ID and stored["native_call_id"] == item["id"] + assert stored["decision_sha256"] == digest + assert (point / "components" / manifest["held_decision_component_id"]).read_bytes() == held["path"].read_bytes() + else: + with pytest.raises(ValueError, match="routes or identity"): + host._capture(held, time.monotonic() + 30) + assert not point.exists() + stdout, stderr = command_process.communicate(timeout=30) + assert command_process.returncode == 0, (stdout, stderr) + assert _heartbeat_receipt_count(runtime, TURN_ID) == 1 + assert _spend_run_count(runtime) == 0 + assert "thread/fork" not in calls diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py index 2110efa1b2..572d89cfc2 100644 --- a/loopx/hosts/codex_checkpoint.py +++ b/loopx/hosts/codex_checkpoint.py @@ -368,16 +368,21 @@ def arguments(command): except (SystemExit, ValueError) as error: raise ValueError("native pending guard arguments are invalid") from error args = arguments(item.get("command", "")) - # Older sealed interrupted guards were captured with mandatory envelopes. - # New captures bind the explicit execution mode; neither format opts in. + # Older sealed guards could omit the then-mandatory envelope flag. + # Missing complete mode is historical off, never inferred from the command. turn_envelope = guard.get("turn_envelope", True) + complete_turn = guard.get("complete_turn", False) if not isinstance(turn_envelope, bool): raise ValueError("native pending guard envelope mode must be boolean") + if not isinstance(complete_turn, bool): + raise ValueError("native pending guard complete mode must be boolean") + if turn_envelope and complete_turn: + raise ValueError("native pending guard has incompatible envelope and complete modes") expected = {"command": "quota", "quota_command": "should-run", "format": "json", "registry": guard["registry"], "runtime_root": guard["runtime_root"], "goal_id": guard["goal_id"], "agent_id": guard["agent_id"], "turn_instance_id": guard["turn_instance_id"], "turn_envelope": turn_envelope, - "decision_output_root": guard["capture_root"], + "complete_turn": complete_turn, "decision_output_root": guard["capture_root"], "decision_output_dir": None, "decision_handoff_socket": guard["socket_path"], "codex_context_dir": guard.get("codex_context_dir"), "codex_context_root": guard.get("codex_context_root"),