Fix paging and numeric fidelity for saved native quota context

loopx/cli_commands/quota_capture.py

@@ -805,7 +805,8 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str
         activation = native_context_activation(args)
         saved_page = bool(getattr(args, "initial_read", False) or getattr(args, "select", None))
         if saved_page and (args.format != "json" or not _context_page_output_enabled(args, activation) or args.offset != 0
-                or args.reference_file or args.selection_sha256 or args.page_bytes is not None
+                or args.reference_file or args.selection_sha256
+                or (args.page_bytes is not None and not activation["enabled"])
                 or (args.initial_read and args.select)):
             raise ValueError("Use JSON context-project at offset zero with native delivery or progressive --initial-read. "
                 "Choose either --initial-read or --select. Use context-read for replay or continuation.")

loopx/cli_commands/quota_presentation.py

@@ -5,7 +5,7 @@
 from collections.abc import Callable, Mapping
 
 from ..control_plane.effect_runtime import EffectRuntimeRejected
-from ..control_plane.quota.turn_envelope import build_turn_envelope
+from ..control_plane.quota.turn_envelope import build_turn_envelope, project_captured_replan_cli_view
 
 
 def _quota_renderer(
@@ -84,8 +84,7 @@ def _project_quota_cli_payload(
             return compact_quota_plan_cli_payload(payload, detail_sections=detail_sections)
         if args.quota_command == "should-run":
             if captured_decision_path and "vision" not in detail_sections:
-                from ..control_plane.quota.turn_envelope import _turn_envelope_result
-                payload = _turn_envelope_result("replan_cli_view", payload=payload)
+                payload = project_captured_replan_cli_view(payload)
             return compact_quota_should_run_cli_payload(
                 payload,
                 include_todo_summary_detail="agent-todos" in detail_sections,

loopx/control_plane/goals/vision_checkpoint.ts

@@ -104,10 +104,10 @@ export function visionAuthoringContract(autonomousReplan = false): JsonObject {
       "Replace todo_<id> with a verified Todo ID after normal authoring/readback. Entries record plans, not Todo mutations, claims or execution authority. " +
       "For ordinary CLI writeback, pass the packet with --agent-vision-json <file>. " +
       (autonomousReplan
-        ? "JSON replaces the complete packet; the sparse minimal_example is not an update to an existing Vision. " +
-          "Start from the current Agent's complete packet and full vision_patch, then replace only deliberately reviewed fields and the fresh path_delta. " +
-          "Preserve every other field and the current state; path_delta.retained text does not merge omitted fields. " +
-          "Reuse the known current packet, or read it from canonical status/history if unavailable; never reconstruct missing intent from summaries. "
+        ? "JSON replaces the complete packet. For an existing Vision, the sparse minimal_example shows syntax. " +
+          "Use the current Agent's complete packet and full vision_patch. Read canonical status/history if unavailable. Never reconstruct intent from summaries. " +
+          "Edit only deliberately reviewed fields and supply a fresh path_delta. Keep the current state and all other fields. " +
+          "path_delta.retained text does not merge omitted fields. "
         : "") +
       "checkpoint-context is only for recovery after the original committed Turn writeback; " +
       "follow its returned same-Turn recovery action, not a fresh-turn preflight.",

loopx/control_plane/quota/turn_envelope.py

@@ -181,6 +181,17 @@ def bind_turn_decision_capture(
     )
 
 
+def project_captured_replan_cli_view(payload: dict[str, Any]) -> dict[str, Any]:
+    """Stage replan reads without round-tripping unchanged source values."""
+    if not isinstance(payload.get("replan_read_path"), str):
+        return payload
+    from .cli_projection import _apply_context_projection_changes
+
+    result = _turn_envelope_result("replan_cli_view", payload=payload, source_edits=True,
+        source_number_tokens=source_number_tokens(payload))
+    return _apply_context_projection_changes(payload, result)
+
+
 def prepare_context_delivery(
     context: Mapping[str, Any], *, scope: Mapping[str, str], capture_id: str,
     baseline: Mapping[str, Any] | None = None,

loopx/control_plane/quota/turn_envelope.ts

@@ -1737,7 +1737,12 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
   const operation = request.operation;
   if (operation === "replan_view") return replanView(request);
   if (operation === "context_select") return capturedDecisionSelection(request);
-  if (operation === "replan_cli_view") return replanCliView(requireJsonObject(request.payload, "replan CLI payload"));
+  if (operation === "replan_cli_view") {
+    const source = requireJsonObject(request.payload, "replan CLI payload");
+    if (request.source_edits !== true) return replanCliView(source);
+    return contextProjectionChanges(source,
+      replanCliView(source, contextSourceIdentity(source, request.source_number_tokens)));
+  }
   if (operation === "selected_work_todo") {
     const source = requireJsonObject(request.payload, "Turn work selection source");
     const historical = scalarString(request.historical_primary_text, "historical primary Todo");

tests/control_plane/test_replan_staged_read.py

@@ -231,19 +231,32 @@ 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:
-            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)
-        source, view = json.loads(source_path.read_bytes()), json.loads(view_path.read_bytes())
-        assert str(view_path) in body and str(source_path) not in body
+        source_path = wake / "checkpoint-start-guard.json"
+        raw = source_path.read_bytes()
+        source = json.loads(raw)
         assert body.endswith("original heartbeat instructions")
         if context_delivery:
-            assert "--defer-context-presentation" in commands[0]
-            assert "--context-delivery" in commands[0]
+            from loopx.control_plane.turn_driver.codex_context import codex_context_hook
+
+            assert "--prepared-capture-reference" in commands[0] and "--defer-context-presentation" in commands[0]
+            assert "--decision-output-root" not in commands[0]
+            captured_dir = Path(commands[0][commands[0].index("--decision-output-dir") + 1])
+            argv = shlex.split(body.split("```sh\n", 1)[1].split("\n```", 1)[0])
+            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
             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",
+                "session_id": session_id, "cwd": str(fixture["project"])})
+            env["CODEX_THREAD_ID"] = session_id
+            projected = run(argv, cwd=fixture["project"], env=env, capture_output=True, text=True, timeout=60)
+            assert projected.returncode == 0, projected.stdout
+            presentation = reference_metadata_oracle(json.loads(projected.stdout))
+            assert presentation["ok"] and presentation["mode"] == "full"
+            assert presentation["target"]["capture_id"] == argv[argv.index("--capture-sha256") + 1]
+            view = read_cli_policy_payload(projected.stdout, returncode=projected.returncode)
+            assert original.read_bytes() == raw and source_path.read_bytes() == raw
             assert not any(read.get("source") == "goal_state" for read in
                 view["interaction_contract"]["agent_channel"]["required_reads"])
             delivered = [row["content"] for row in view["interaction_contract"]["agent_channel"]
@@ -256,18 +269,14 @@ def observe(command, **kwargs):
             else:
                 assert len(delivered) == 1 and delivered[0]["full_source"] is True
                 assert delivered[0]["text"] == fixture["state"].read_text()
-            # The optional full/delta transport must present the staged CLI
-            # view too. Its full recovery must not restore old inline hints.
-            command = body.split("```sh\n", 1)[1].split("\n```", 1)[0]
-            projected = run(shlex.split(command), cwd=fixture["project"], env=env,
-                capture_output=True, text=True, timeout=60)
-            assert projected.returncode == 0, projected.stdout
-            presentation = reference_metadata_oracle(json.loads(projected.stdout))
-            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
+        else:
+            assert "--decision-output-root" in commands[0]
+            view_path = wake / "checkpoint-start-view.json"
+            assert str(view_path) in body and str(source_path) not in body
+            view = json.loads(view_path.read_bytes())
         assert source["heartbeat_receipt"]["turn_instance_id"] == TURN
         assert source["heartbeat_receipt"] == view["heartbeat_receipt"]
+        assert json.dumps(source["quota"], sort_keys=True) == json.dumps(view["quota"], sort_keys=True)
         assert receipt["checkpoint"]["reason"] == "synthetic_no_native_capture"
         selection = view["interaction_contract"]["cli_channel"].get("selection_command")
         if selection:
@@ -627,10 +636,13 @@ def test_large_acceptance_is_not_repeated_in_cli_or_envelope_work_context(tmp_pa
 
 
 def test_staged_entry_replays_goal_and_changed_horizon_with_existing_delta_owner(tmp_path):
+    from loopx.control_plane.quota.turn_envelope import project_captured_replan_cli_view
     from replan_packet_fixture import populated_replan_source
 
     source = populated_replan_source(tmp_path, acceptance=True)
-    entry = _turn_envelope_result("replan_cli_view", payload=source)
+    source["measurements"] = [1.0, -0.0, 9007199254740993]
+    entry = project_captured_replan_cli_view(source)
+    assert json.dumps(entry["measurements"]) == json.dumps(source["measurements"])
     assert entry["action_portfolio"] == source["action_portfolio"]
     scope = {"goal_id": source["goal_id"], "agent_id": source["agent_identity"]["agent_id"],
         "session_id": "synthetic-session", "context_epoch": "one"}
@@ -639,7 +651,7 @@ def test_staged_entry_replays_goal_and_changed_horizon_with_existing_delta_owner
         current = copy.deepcopy(source)
         if changed:
             current["planning_horizon"]["work_items"][0].update(priority="P0", successor_todo_ids=["todo-new-dependency"])
-        value = _turn_envelope_result("replan_cli_view", payload=current)
+        value = project_captured_replan_cli_view(current)
         candidate = prepare_context_delivery(value, scope=scope, capture_id="entry-b", surface="quota_packet",
             baseline=full, retained={"complete": True, "target": full["target"]})
         # Even without native delta, the entry has one immutable reader rather
@@ -663,7 +675,7 @@ def test_staged_entry_replays_goal_and_changed_horizon_with_existing_delta_owner
                   {"effective_action": "governed_capability_intent"},
                   {"replan_action_packet": {**source["replan_action_packet"], "settlement_only": True}}):
         legacy = {**source, **patch}
-        assert _turn_envelope_result("replan_cli_view", payload=legacy) == legacy
+        assert project_captured_replan_cli_view(legacy) == legacy
 
 
 def test_native_reader_keeps_its_installation_across_acceptance_and_writeback(tmp_path, monkeypatch):

tests/control_plane/test_saved_context_project.py

@@ -520,6 +520,10 @@ def consume(wire):
     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"]
@@ -949,11 +953,18 @@ def response(request, number, message):
             original_command = original["interaction_contract"]["cli_channel"]["selection_command"]["command_args_template"]
             words = shlex.split(selected["command_args_template"])
             original_words = shlex.split(original_command)
-            for option in ("--context-delivery", "--codex-context-dir", "--max-page-output-bytes"):
-                for command_words in (words, original_words):
-                    if option in command_words:
-                        index = command_words.index(option)
-                        del command_words[index:index + 2]
+            transport = {
+                "--decision-output-root": str(home / "loopx-context"),
+                "--context-delivery": "on",
+                "--delta-delivery": "auto",
+                "--codex-context-dir": str(home / "loopx-context"),
+                "--max-page-output-bytes": "32000",
+            }
+            for option, value in transport.items():
+                assert option not in original_words and words.count(option) == 1
+                index = words.index(option)
+                assert words[index + 1] == value
+                del words[index:index + 2]
             assert words == original_words
             selected["command_args_template"] = original_command
             # Main already deduplicates identical Todo rows within this exact
@@ -1423,31 +1434,46 @@ def cli(*args):
         return _run_cli(registry, runtime, *args, cwd=tmp_path)
     initial_files = set(context.rglob("*"))
     for invalid in (["--initial-read", "--select", "/instructions"], ["--offset", "1", "--select", "/instructions"],
-                    ["--context-delivery", "off", "--initial-read"], ["--goal-id", "foreign", "--initial-read"]):
+                    ["--context-delivery", "off", "--initial-read"], ["--goal-id", "foreign", "--initial-read"],
+                    ["--delta-delivery", "off", "--initial-read", "--page-bytes", "3200"]):
         code, error = cli(*project, *invalid)
         assert code and not error["ok"] and "content" not in error
+    code, error = cli(*project, "--select", "/instructions", "--page-bytes", "255")
+    assert code and not error["ok"] and "--page-bytes must be between 256 and" in error["reason"]
     assert set(context.rglob("*")) == initial_files
-    code, first = cli(*project, "--select", "/instructions")
+    window = 3200
+    code, first = cli(*project, "--select", "/instructions", "--page-bytes", str(window))
     assert code == 0 and first["offset"] == 0 and not first["eof"]
     assert first["navigation"]["mode"] == "full"
     codex_context_hook(str(context), {**event, "hook_event_name": "PreCompact"})
     journal = (context / "session.json").read_bytes()
     files = set(context.rglob("*"))
-    def read_reference():
-        ref = first["reference_file"]
+    def read_selection(*, resume):
         result = bytearray()
-        args = ["quota", "context-read", "--reference-file", ref, "--max-page-output-bytes", "6400"]
+        if resume:
+            result.extend(first["content"].encode())
+            assert len(result) <= window
+            words = shlex.split(first["next_command"])
+            args = words[words.index("quota"):]
+        else:
+            args = ["quota", "context-read", "--reference-file", first["reference_file"],
+                    "--max-page-output-bytes", "6400"]
         while True:
             code, page = cli(*args)
             assert code == 0 and page["offset"] == len(result)
-            result.extend(page["content"].encode())
+            content = page["content"].encode()
+            if resume:
+                assert len(content) <= window
+            result.extend(content)
             if page["eof"]:
                 assert len(result) == page["total_bytes"]
                 return json.loads(result)["selections"][0]["value"]
             words = shlex.split(page["next_command"])
             args = words[words.index("quota"):]
     with ThreadPoolExecutor(2) as pool:
-        assert list(pool.map(lambda _: read_reference(), range(2))) == [value, value]
+        replay = pool.submit(read_selection, resume=False)
+        continuation = pool.submit(read_selection, resume=True)
+        assert replay.result() == continuation.result() == value
     assert (context / "session.json").read_bytes() == journal and set(context.rglob("*")) == files
     assert hashlib.sha256(original.read_bytes()).hexdigest() == digest
     assert not registry.exists() and not runtime.exists()