Preserve same-Turn Todo selection after quota preflight

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"):

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")

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)
 

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,24 +473,48 @@ 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()
     assert _heartbeat_receipt_count(runtime, turn) == 1 and _spend_run_count(runtime) == 0
     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