diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index 741e45b729..9f73b313d3 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -374,7 +374,7 @@ def _emit_quota_result( # Preserve the original validation error when activation itself failed. context_delivery_enabled = False activation = {"progressive_enabled": False} - bind_capture_selection_transport(payload, args) + bind_capture_selection_transport(payload, args, capture_directory=capture_directory) delivery_request = None decision_identity = ("", "") if capture_directory is not None and payload.get("replan_action_packet"): diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 4bf1e7449a..50f83e11d9 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -1034,6 +1034,10 @@ def prepare_decision_capture(args: argparse.Namespace) -> Path | None: return Path(tempfile.mkdtemp(prefix="decision-", dir=parent)) assert requested is not None # root is absent; exactly one destination was checked above. directory = Path(requested).expanduser().absolute() + if getattr(args, "prepared_capture_reference", False): + # Bind the physical parent before admission: the saved selection + # uses it as an output root, which must not be a symlink. + directory = Path(os.path.realpath(directory.parent, strict=True)) / directory.name # A new directory per invocation prevents overwriting a prior observation, # including a prior invocation in the same Turn. Existing symlinks fail too. directory.mkdir(mode=0o700) @@ -1158,13 +1162,18 @@ def _capture_selection_transport_template(payload: dict[str, Any], args: argpars def bind_capture_selection_transport(payload: dict[str, Any], args: argparse.Namespace, *, - capture_root: Path | None = None) -> None: + capture_root: Path | None = None, + capture_directory: Path | None = None) -> None: """Retain caller display/capture options in the existing selection command. Only transport argv is composed here; the typed quota owner still selects the action and owns admission. Each selection invocation allocates a new capture. """ root = getattr(args, "decision_output_root", None) + # A prepared reference binds one immutable directory. Same-Turn + # selection uses its parent to create a fresh sibling capture. + if capture_directory is not None and getattr(args, "prepared_capture_reference", False): + capture_root = capture_directory.parent if root is None and capture_root is None: return interaction = payload.get("interaction_contract") diff --git a/tests/control_plane/test_replan_staged_read.py b/tests/control_plane/test_replan_staged_read.py index 05df95313d..6bb65081ec 100644 --- a/tests/control_plane/test_replan_staged_read.py +++ b/tests/control_plane/test_replan_staged_read.py @@ -231,13 +231,23 @@ def observe(command, **kwargs): try: body = checkpoint.capture_before_replan(env, receipt, "original heartbeat instructions") assert len(commands) == 1 and "--turn-envelope" not in commands[0] - assert "--decision-output-root" in commands[0] source_path, view_path = wake / "checkpoint-start-guard.json", wake / "checkpoint-start-view.json" if context_delivery: + assert "--prepared-capture-reference" in commands[0] argv = shlex.split(body.split("```sh\n", 1)[1].split("\n```", 1)[0]) view_path = Path(argv[argv.index("--capture-file") + 1]) assert view_path.parent != wake and view_path.is_relative_to(wake) + else: + assert "--decision-output-root" in commands[0] source, view = json.loads(source_path.read_bytes()), json.loads(view_path.read_bytes()) + if context_delivery and replan: + original_dir = Path(commands[0][commands[0].index("--decision-output-dir") + 1]) + selection = source["interaction_contract"]["cli_channel"]["selection_command"] + followup = shlex.split(selection["command_args_template"]) + assert followup[followup.index("--decision-output-root") + 1] == str(original_dir.parent.resolve()) + assert followup[followup.index("--context-delivery") + 1] == "on" + assert followup[followup.index("--codex-context-dir") + 1] == str(native) + assert "--decision-output-dir" not in followup and "--prepared-capture-reference" not in followup assert str(view_path) in body and str(source_path) not in body assert body.endswith("original heartbeat instructions") if context_delivery: @@ -266,6 +276,18 @@ def observe(command, **kwargs): assert presentation["ok"] and presentation["mode"] == "full" assert reference_metadata_oracle(presentation)["recovery"]["path"] == str(view_path) assert read_cli_policy_payload(projected.stdout, returncode=projected.returncode) == view + if replan: + # Before SessionStart the original guard is the full fallback. + # Check staged delivery on that same capture with its real lifecycle. + from loopx.control_plane.turn_driver.codex_context import codex_context_hook + + session = "staged-checkpoint-reader" + codex_context_hook(str(native), {"session_id": session, "cwd": str(fixture["project"]), + "hook_event_name": "SessionStart", "source": "startup"}) + projected = run(shlex.split(command), cwd=fixture["project"], env={**env, "CODEX_THREAD_ID": session}, + capture_output=True, text=True, timeout=60) + assert projected.returncode == 0, projected.stdout + view = read_cli_policy_payload(projected.stdout, returncode=projected.returncode) assert source["heartbeat_receipt"]["turn_instance_id"] == TURN assert source["heartbeat_receipt"] == view["heartbeat_receipt"] assert receipt["checkpoint"]["reason"] == "synthetic_no_native_capture" @@ -298,6 +320,15 @@ def observe(command, **kwargs): assert {key: fixture[key].read_bytes() for key in before} == before assert capture.read_bytes() == source_path.read_bytes() assert len(commands) == 1 # Reads did not re-enter checkpoint preflight. + if context_delivery: + selected = shlex.split(selection["route_prefix"]) + followup + selected[selected.index("{todo_id}")] = TODO + result = run(selected, cwd=fixture["project"], env=env, capture_output=True, text=True, timeout=60) + assert result.returncode == 0, result.stdout + [next_capture] = set(original_dir.parent.glob("*/decision.json")) - {capture} + identity = json.loads(next_capture.read_bytes())["heartbeat_receipt"]["settlement_identity"] + assert (identity["todo_id"], identity["turn_instance_id"]) == (TODO, TURN) + assert capture.read_bytes() == source_path.read_bytes() finally: restart_effect_runtime(wait_for_exit=True) diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index 6ccdcf8b83..d48b4c2429 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/tests/control_plane/test_saved_context_project.py @@ -421,9 +421,11 @@ def test_preflight_instruction_pins_both_projection_and_full_fallback(tmp_path): @pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("enabled,compacted", [(False, False), (True, False), (True, True)], - ids=["off", "on", "on-after-compaction"]) -def test_checkpoint_preflight_selection_inherits_native_transport(tmp_path, monkeypatch, provider, enabled, compacted): +@pytest.mark.parametrize("enabled,compacted,placement", [ + (False, False, "physical"), (True, False, "physical"), (True, True, "physical"), + (True, False, "alias-before-session"), (True, False, "explicit-root"), +], ids=["off", "on", "on-after-compaction", "alias-before-session", "explicit-root"]) +def test_checkpoint_preflight_selection_inherits_native_transport(tmp_path, monkeypatch, provider, enabled, compacted, placement): """The saved guard is presented after SessionStart, then really bound once.""" from benchmark.runtime.checkpoint import capture_before_replan from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos @@ -434,9 +436,18 @@ def test_checkpoint_preflight_selection_inherits_native_transport(tmp_path, monk _heartbeat_receipt_count, _spend_run_count, _write_fixture, ) + import subprocess + + tmp_path = tmp_path.resolve() isolate_sqlite_runtime(tmp_path, monkeypatch) project, runtime, registry = _write_fixture(tmp_path / "authority") _configure_selectable_alternative(project) + if enabled and placement == "physical": + # Keep an actual delta positive: the bound view preserves 0.5's spelling. + # A 1.0 -> 1 transition requires full recovery even within the same epoch. + registry_data = json.loads(registry.read_bytes()) + registry_data["goals"][0]["quota"]["compute"] = 0.5 # The fixture still declares two slots. + registry.write_text(json.dumps(registry_data)) state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" initialize_canonical_authority(runtime, GOAL_ID, build_todo_runtime_shadow_projection(goal_id=GOAL_ID, todos=read_fixture_todos(state), handoff_mode="soft_claim"), @@ -462,13 +473,34 @@ def test_checkpoint_preflight_selection_inherits_native_transport(tmp_path, monk "LOOPX_CONTEXT_DELIVERY": "on" if enabled else "off", "LOOPX_CODEX_CONTEXT_DIR": str(native)} env.pop("CODEX_THREAD_ID", None) receipt = {"turn_id": turn, "mode": "heartbeat", "stage": "execute"} - body = capture_before_replan(env, receipt, "Current work.") + alias = tmp_path / "capture parent alias" + if placement != "physical": + run = subprocess.run + + def through_alias(argv, **kwargs): + if "--prepared-capture-reference" in argv: + argv = list(argv) + i = argv.index("--decision-output-dir") + 1 + requested = Path(argv[i]) + alias.symlink_to(requested.parent, target_is_directory=True) + argv[i] = str(alias / requested.name) + return run(argv, **kwargs) + + # Change only how the CLI reaches the directory already allocated by + # the checkpoint runner. Guard, capture, projection and selection are real. + with monkeypatch.context() as scoped: + scoped.setattr("benchmark.runtime.checkpoint.subprocess.run", through_alias) + body = capture_before_replan(env, receipt, "Current work.") + else: + body = capture_before_replan(env, receipt, "Current work.") original_guard = (wake / "checkpoint-start-guard.json").read_bytes() if enabled: command = shlex.split(body.split("```sh\n", 1)[1].split("\n```", 1)[0]) source = Path(command[command.index("--capture-file") + 1]) assert source.parent != wake and source.is_relative_to(wake) assert command[command.index("--capture-sha256") + 1] == hashlib.sha256(source.read_bytes()).hexdigest() + if placement != "physical": + command[command.index("--capture-file") + 1] = str(alias / source.relative_to(source.parent.parent)) else: source = wake / "checkpoint-start-view.json" raw = source.read_bytes() @@ -476,10 +508,13 @@ def test_checkpoint_preflight_selection_inherits_native_transport(tmp_path, monk assert not list(native.iterdir()) # Host preflight must not publish before native SessionStart. event = {"session_id": "native-session", "cwd": str(project), "hook_event_name": "SessionStart", "source": "startup"} - monkeypatch.setenv("CODEX_THREAD_ID", event["session_id"]) + lifecycle = enabled and placement != "alias-before-session" monkeypatch.chdir(project) - if enabled: + if lifecycle: + monkeypatch.setenv("CODEX_THREAD_ID", event["session_id"]) codex_context_hook(str(native), event) + else: + monkeypatch.delenv("CODEX_THREAD_ID", raising=False) def consume(wire): data = bytearray() @@ -489,7 +524,8 @@ def consume(wire): while True: assert page["ok"] and page["offset"] == len(data) data.extend(page["content"].encode()) - codex_context_hook(str(native), {**event, "hook_event_name": "PostToolUse", "tool_response": json.dumps(page)}) + if lifecycle: + codex_context_hook(str(native), {**event, "hook_event_name": "PostToolUse", "tool_response": json.dumps(page)}) if page["eof"]: assert hashlib.sha256(data).hexdigest() == machine_reference_oracle(wire)["sha256"] return json.loads(data) @@ -499,6 +535,11 @@ def consume(wire): if enabled: command.extend(("--max-page-output-bytes", "32000")) + expected_root = source.parent.parent + if placement == "explicit-root": + expected_root = tmp_path / "explicit capture root" + expected_root.mkdir(mode=0o700) + command.extend(("--decision-output-root", str(expected_root))) rc, cold = _run_cli(registry, runtime, *command[command.index("quota"):], cwd=project) assert rc == 0 and cold["navigation"]["mode"] == "full" and cold["offset"] == 0, cold assert len(json.dumps(cold, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000 @@ -513,16 +554,41 @@ def consume(wire): parsed = build_parser().parse_args(argv[1:]) assert (parsed.context_delivery == "on") is enabled assert parsed.codex_context_dir == (str(native) if enabled else None) - assert parsed.max_page_output_bytes == (32000 if enabled else None) + assert parsed.max_page_output_bytes == (32000 if lifecycle else None) assert argv.count("--decision-output-root") == 1 + if placement == "alias-before-session": + rejected_argv = argv.copy() + rejected_argv[rejected_argv.index("--decision-output-root") + 1] = str(alias) + rc, rejected = _run_cli(registry, runtime, *rejected_argv[1:], cwd=project) + assert rc == 1 and "no guard was run" in rejected["reason"], rejected + assert list(expected_root.glob("*/decision.json")) == [source] + assert _heartbeat_receipt_count(runtime, turn) == 1 rc, bound = _run_cli(registry, runtime, *argv[1:], cwd=project) assert rc == 0, bound if enabled: - assert bound["mode"] == ("full" if compacted else "delta"), bound - assert len(json.dumps(bound, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000 - delivered = consume(bound) - if not compacted: - assert delivered["base"]["capture_id"] == machine_reference_oracle(cold["navigation"])["target"]["capture_id"] + assert Path(parsed.decision_output_root) == expected_root + [selected_capture] = set(expected_root.glob("*/decision.json")) - {source} + digest = machine_reference_oracle(bound)["capture_sha256"] + rc, page = _run_cli(registry, runtime, "quota", "context-read", "--capture-file", str(selected_capture), + "--capture-sha256", digest, "--select", "/heartbeat_receipt/settlement_identity", cwd=project) + assert rc == 0 and page["eof"], page + [identity] = json.loads(page["content"])["selections"] + selected_receipt = identity["value"] + assert (selected_receipt["goal_id"], selected_receipt["agent_id"], selected_receipt["turn_instance_id"], + selected_receipt["todo_id"]) == (GOAL_ID, AGENT_ID, turn, TODO_ID) + if placement == "explicit-root": + assert list(source.parent.parent.glob("*/decision.json")) == [source] + if lifecycle: + expected_mode = "full" if compacted or placement == "explicit-root" else "delta" + assert bound["mode"] == expected_mode, bound + assert len(json.dumps(bound, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000 + delivered = consume(bound) + if expected_mode == "delta": + assert delivered["base"]["capture_id"] == machine_reference_oracle(cold["navigation"])["target"]["capture_id"] + if placement == "explicit-root": + assert type(presented["quota"]["compute"]) is float + assert type(delivered["quota"]["compute"]) is int + assert delivered["quota"]["compute"] == presented["quota"]["compute"] == 1 else: assert bound["mode"] == "should-run" and not list(native.iterdir()) assert _heartbeat_receipt_count(runtime, turn) == 2 and _spend_run_count(runtime) == 0