From d72f43229d1b228c542355629ed988c30b2202b8 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 8 Nov 2026 17:09:23 +0800 Subject: [PATCH] fix(quota): preserve explicit selection transport before SessionStart Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/cli_commands/quota.py | 2 +- loopx/cli_commands/quota_capture.py | 13 ++- .../control_plane/test_replan_staged_read.py | 16 +++ .../test_saved_context_project.py | 102 ++++++++++++++---- 4 files changed, 112 insertions(+), 21 deletions(-) diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index 22ad811942..7ce08cb724 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -368,7 +368,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 94d407a79d..5c8d0cdbf1 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -471,6 +471,8 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str full_reference = _capture_read_reference(projected["recovery"], page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv) reference = dict(full_reference) + else: + bind_capture_selection_transport(projected, args, capture_root=root) native_view = native_context_capture_view(projected, cache, context_directory=directory, hook_session=hook_session, progressive_enabled=activation["progressive_enabled"], initial_read_plan=prepared.initial_read_plan, metadata={"recovery": full_reference, "capture_sha256": args.capture_sha256, @@ -596,6 +598,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) @@ -720,13 +726,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 41305dc3ed..9a962bcc7f 100644 --- a/tests/control_plane/test_replan_staged_read.py +++ b/tests/control_plane/test_replan_staged_read.py @@ -245,6 +245,13 @@ def observe(command, **kwargs): original = Path(argv[argv.index("--capture-file") + 1]) assert original == captured_dir / "decision.json" and original.is_relative_to(wake) assert original.read_bytes() == raw and str(source_path) not in body + if replan: + selection = source["interaction_contract"]["cli_channel"]["selection_command"] + followup = shlex.split(selection["command_args_template"]) + assert followup[followup.index("--decision-output-root") + 1] == str(captured_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 not list(native.iterdir()), "Host preflight must not alter native continuity" session_id = "checkpoint-preflight-session" codex_context_hook(str(native), {"hook_event_name": "SessionStart", "source": "startup", @@ -307,6 +314,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(captured_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 3cc0ff9667..300c0fcdbc 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/tests/control_plane/test_saved_context_project.py @@ -421,9 +421,12 @@ 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-before-session"), + (True, False, "explicit-root"), +], ids=["off", "on", "on-after-compaction", "alias-before-session", "explicit-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,6 +437,9 @@ 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) @@ -462,13 +468,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 +503,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 not in {"alias-before-session", "explicit-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 +519,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) @@ -497,11 +528,17 @@ def consume(wire): rc, page = _run_cli(registry, runtime, *argv[argv.index("quota"):], cwd=project) assert rc == 0, page + page_limit = 8000 if placement == "explicit-before-session" else 32000 if enabled: - command.extend(("--max-page-output-bytes", "32000")) + command.extend(("--max-page-output-bytes", str(page_limit))) + expected_root = source.parent.parent + if placement in {"explicit-root", "explicit-before-session"}: + 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 + assert len(json.dumps(cold, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= page_limit presented = consume(cold["navigation"]) if compacted: codex_context_hook(str(native), {**event, "hook_event_name": "PreCompact"}) @@ -513,20 +550,47 @@ 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 == (page_limit if enabled 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 - current = Path(machine_reference_oracle(bound)["recovery"]["path"]) - captured = json.loads(current.with_name("decision.json").read_bytes()) - required = json.loads(current.read_bytes()) - assert json.dumps(required["quota"], sort_keys=True) == json.dumps(captured["quota"], sort_keys=True) - 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 + assert len(json.dumps(bound, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= page_limit + [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 in {"explicit-root", "explicit-before-session"}: + assert list(source.parent.parent.glob("*/decision.json")) == [source] + if placement == "explicit-before-session": + assert bound["mode"] == "full", bound + assert machine_reference_oracle(cold)["recovery"]["sha256"] == hashlib.sha256(raw).hexdigest() + for key in ("first_read_command", "command"): + rc, first_page = _run_cli(registry, runtime, *shlex.split(bound[key])[1:], cwd=project) + assert rc == 0 and first_page["ok"], first_page + assert len(json.dumps(first_page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= page_limit + if lifecycle: + assert bound["mode"] == ("full" if compacted else "delta"), bound + current = Path(machine_reference_oracle(bound)["recovery"]["path"]) + captured = json.loads(current.with_name("decision.json").read_bytes()) + required = json.loads(current.read_bytes()) + assert json.dumps(required["quota"], sort_keys=True) == json.dumps(captured["quota"], sort_keys=True) + delivered = consume(bound) + if not compacted: + assert delivered["base"]["capture_id"] == machine_reference_oracle(cold["navigation"])["target"]["capture_id"] else: assert bound["mode"] == "should-run" and not list(native.iterdir()) assert _heartbeat_receipt_count(runtime, turn) == 2 and _spend_run_count(runtime) == 0