From 5115c6b7c3361d717225f9ba68a58376426d1d1e Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 18:02:54 +0800
Subject: [PATCH 1/3] fix(context): keep native readers on their selected
 runtime

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/required-work-context.md       | 10 +++
 loopx/cli_commands/quota_capture.py           | 24 +++--
 .../turn_driver/decision_capture.py           |  6 +-
 .../references/repair-patterns.md             |  1 +
 tests/control_plane/test_cli_program_route.py | 89 +++++++++++++++++++
 .../test_native_context_delivery_cost.py      | 16 +++-
 6 files changed, 133 insertions(+), 13 deletions(-)

diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index 06fcd2e1d2..b28aab1fd3 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -175,6 +175,13 @@ model retention. These separate boundaries prevent transport loss from becoming
 either a repeated full-input loop or permission to act on missing requirements.
 No token savings or accepted outcome gains follow from instruction delivery alone.
 
+Commands returned by a running `context-read` keep its explicit CLI selection,
+or its current Python interpreter when no CLI was specified. This applies to
+continuations and rejected-selection recovery. Capture bytes, digests, cursors
+and authority are unchanged; native full/delta byte comparisons include this
+reader command metadata. Existing in-process reference constructors keep their
+default command route.
+
 An enabled native **full or delta** reference for a recognized decision can include
 `first_read_command`, reading its current full capture without an older baseline.
 It batches current admission/stop facts, the current action, the selected task,
@@ -272,6 +279,9 @@ reader，不重新运行首读导航；后续导航改变不能替换原选读
 及显式文件首读游标在选读摘要变化时仍拒绝，沿同一完整原 capture 引用恢复，不重新
 准入或改写旧描述文件。退役这些 legacy reader 前仍须对账原未完成 Turn；本切片
 保留历史解码与完整捕获恢复。
+实际运行的 `context-read` 返回续页或选读拒绝恢复命令时，保留显式指定的 CLI；
+未指定 CLI 则使用当前 Python 解释器。捕获正文、摘要、游标和权限不变，原生
+full/delta 字节比较计入这些 reader 命令元数据；既有进程内引用构造器保留默认路由。
 首读可把已经完整选中的同级内容合并为其完整父节点，减少地址重复；任何未选中的
 可选或未知内容都会阻止合并；若数组合并会恢复原元素顺序、覆盖 owner 的读取优先级，
 该数组及其祖先也不合并。选中事实和义务不变。显式选读和索引仍保留原地址。
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 880e908e02..05b0d57cf4 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -46,7 +46,8 @@ def _capture_page(path: Path, raw: bytes, digest: str, *, offset: int, size: int
                   list_fields: bool = False,
                   list_depth: int | None = None,
                   initial_read: bool = False,
-                  reference_file: Path | None = None) -> dict[str, Any]:
+                  reference_file: Path | None = None,
+                  reader_argv: list[str] | None = None) -> dict[str, Any]:
     end = min(offset + size, len(raw))
     while end < len(raw) and raw[end] & 0xC0 == 0x80:
         end -= 1
@@ -94,7 +95,8 @@ def page(text: str) -> dict[str, Any]:
                 selection_sha256=selection_sha256 if reference_file is None else None,
                 list_fields=list_fields if reference_file is None else False,
                 list_depth=list_depth if reference_file is None else None,
-                initial_read=initial_read if reference_file is None else False, reference_file=reference_file)
+                initial_read=initial_read if reference_file is None else False, reference_file=reference_file,
+                reader_argv=reader_argv)
         return result
 
     def fits(result: dict[str, Any]) -> bool:
@@ -181,7 +183,8 @@ def _context_selection_bytes(payload: dict[str, Any], selections: list[dict[str,
 
 def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, Any] | None = None,
                          next_page_bytes: int | None = None,
-                         next_page_output_bytes: int | None = None) -> dict[str, Any]:
+                         next_page_output_bytes: int | None = None,
+                         reader_argv: list[str] | None = None) -> dict[str, Any]:
     """Bounded local IO over immutable JSON, not a new context/authority owner.
 
     Verify the complete byte digest on every page. The cursor addresses bytes
@@ -286,7 +289,8 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
                 parent = "/".join(pointer.split("/")[:rejection["parent_token_count"] + 1])
                 command = context_capture_read_command(path, digest, select=[parent], list_fields=True,
                     page_bytes=next_page_bytes or size,
-                    max_page_output_bytes=next_page_output_bytes or output_limit)
+                    max_page_output_bytes=next_page_output_bytes or output_limit,
+                    reader_argv=reader_argv)
                 return {"ok": False, "schema_version": "context_capture_page_v1",
                     "error_kind": "request_rejected", "diagnostic_code": rejection["code"],
                     "reason": rejection["message"], "recovery_command": command,
@@ -324,7 +328,8 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
             output_limit=output_limit, instruction=bool(context_dir),
             presentation=presentation, next_page_bytes=next_page_bytes,
             next_page_output_bytes=next_page_output_bytes, select=selectors, selection_sha256=selection_digest,
-            list_fields=list_fields, list_depth=list_depth, initial_read=initial_read, reference_file=reference_file)
+            list_fields=list_fields, list_depth=list_depth, initial_read=initial_read, reference_file=reference_file,
+            reader_argv=reader_argv)
     except (OSError, ValueError, RecursionError, EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc:
         # Recovery is presentation-only: keep failures separate from quota's
         # collection error path and retain the existing receipt/capture.
@@ -484,6 +489,7 @@ def _native_context_read_bytes(reference: dict[str, Any], raw: bytes, *,
     Stop once this candidate cannot win. This measures one complete wire read,
     not model tokens, retention or the caller's later selective/recovery reads.
     """
+    from ..command_invocation import current_loopx_cli_argv
     from ..control_plane.effect_runtime import MAX_REQUEST_BYTES
 
     if len(raw) > MAX_REQUEST_BYTES:
@@ -493,9 +499,11 @@ def _native_context_read_bytes(reference: dict[str, Any], raw: bytes, *,
     path = Path(reference["path"]).expanduser().absolute()
     path = path.parent.resolve(strict=True) / path.name
     total, offset = len(_native_output_bytes(reference)), 0
+    reader_argv = current_loopx_cli_argv()
     while True:
         page = _capture_page(path, raw, reference["sha256"],
-            offset=offset, size=page_bytes, output_limit=output_bytes, instruction=False)
+            offset=offset, size=page_bytes, output_limit=output_bytes, instruction=False,
+            reader_argv=reader_argv)
         total += len(_native_output_bytes(page))
         if page["eof"] or (stop_after is not None and total > stop_after):
             return total
@@ -1115,9 +1123,11 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca
         print_payload(payload, args.format, lambda value: str(value.get("command", value.get("reason", ""))))
         return 0 if payload["ok"] else 1
     if args.quota_command == "context-read":
+        from ..command_invocation import current_loopx_cli_argv
+
         # Presentation recovery reads the original capture, never current Goal
         # state. In particular it must not evaluate a guard or append a receipt.
-        payload = read_context_capture(args)
+        payload = read_context_capture(args, reader_argv=current_loopx_cli_argv())
         print_payload(payload, args.format, lambda value: str(value.get("content", value.get("reason", ""))))
         return 0 if payload["ok"] else 1
     return None
diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py
index def8d79c45..94cd30a70d 100644
--- a/loopx/control_plane/turn_driver/decision_capture.py
+++ b/loopx/control_plane/turn_driver/decision_capture.py
@@ -219,14 +219,16 @@ def context_capture_read_command(path: Path, digest: str, *, offset: int = 0,
                                  list_fields: bool = False,
                                  list_depth: int | None = None,
                                  initial_read: bool = False,
-                                 reference_file: Path | None = None) -> str:
+                                 reference_file: Path | None = None,
+                                 reader_argv: list[str] | None = None) -> str:
     from ...command_invocation import loopx_cli_argv
 
     if page_bytes is None:
         page_bytes = NATIVE_CONTEXT_PAGE_BYTES
         if max_page_output_bytes is None:
             max_page_output_bytes = NATIVE_CONTEXT_PAGE_OUTPUT_BYTES
-    return shlex.join([*loopx_cli_argv(), "--format", "json", "quota", "context-read",
+    return shlex.join([*(reader_argv if reader_argv is not None else loopx_cli_argv()),
+        "--format", "json", "quota", "context-read",
         *(["--reference-file", str(reference_file)] if reference_file is not None
           else ["--capture-file", str(path), "--capture-sha256", digest]),
         "--offset", str(offset), "--page-bytes", str(page_bytes),
diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md
index 6b55a12b0f..7d1ac3c537 100644
--- a/skills/loopx-self-repair/references/repair-patterns.md
+++ b/skills/loopx-self-repair/references/repair-patterns.md
@@ -6,6 +6,7 @@ teaches a reusable control-plane lesson.
 | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair |
 | --- | --- | --- | --- | --- |
 | `saved_projection_goal_route_coupling` | A saved context projection fails on conflicting default runtime roots even with native transport disabled, while the same capture remains readable. | Run the shipped projection and returned reader against one digest-bound synthetic capture, with implicit, missing and global registry selectors; retain digest/identity refusals and unchanged runtime snapshots. | Shared CLI registry discovery runs before the saved-view owner validates its source. | Classify the saved presentation at the existing CLI route owner; preserve machine-configuration routing, live authority conflict checks and source/session validation. Reuse the registry-independence integration tests rather than adding a second recovery path or repairing live state. |
+| `native_reader_program_route_loss` | A native context read succeeds through a selected CLI or interpreter, but its returned continuation or recovery command invokes another runtime. | Actual lifecycle capture, generated recovery instructions, unchanged execution of emitted commands with an isolated competing program, complete source/digest readback and installed-package provenance. | The reader used its generic command-renderer default after the caller had already selected the runtime. | Pass the current program argv through the existing reader and command renderer for continuation and rejected-selection recovery; retain constructor defaults and shell quoting. Include emitted command metadata in native byte-cost estimates. Cover raw, selected and index pagination, quoted aliases, refusal recovery and source-state immutability through real CLI and isolated wheel entrypoints. Routing preserves the selected program; it does not pin an installation version or prove Host adoption. |
 | `host_bridge_reader_orphan` | A rejected bridge observation leaves its reader thread and output pipe alive after the child has exited. | Original framing error, cleanup result, child exit, bounded-queue backpressure and reader/pipe state across repeated failures. | Cancellation rejects another malformed frame; the sole consumer leaves while the producer blocks on a full queue, and joining alone cannot release it. | Stop publication, release the waiting producer, join and close through the existing transport helper. Retain cleanup-unconfirmed results and original Host drain ownership; reclaiming a reader does not prove Host exit. Cover malformed JSON and invalid typed observations with real subprocesses, plus existing real Host cancellation and success paths. Do not increase timeouts or fabricate a successful result. |
 | `codex_strict_result_schema_gap` | A Codex Turn fails with invalid_json_schema before execution, or a strict-schema stop result fails internal evidence validation. | Generated Codex schema, required properties, unsupported conditional keywords, and both CLI/app-server result boundaries. | Optional evidence and conditional schema requirements conflict with strict Structured Outputs; internal stop results prohibit evidence fields. | Require nullable evidence fields on the Codex wire and remove unsupported conditionals. Normalize only stop null placeholders in both Codex transports; reject missing/null profile material evidence and preserve non-null evidence for the existing validator. Cover legacy absence, invalid arrays and an isolated schema-acceptance probe without changing other hosts. |
 | `initial_preference_view_omission` | A fulfilled preference hook leaves corrections, retirement or advisory instructions absent from the recommended initial read. | The preference owner's complete current scoped view, actual selected reader/native input, empty and unavailable states, and same-capture continuation/recovery. | A hand-picked projection uses incomplete or stale child field names after the owner has already delivered the current view. | Reuse the complete owned current view in the existing initial selector; retain provenance, advisory authority, exact-scope freshness and immutable capture checks. Cover real File/SQLite lifecycle reads, empty replacements, unknown values and native consumer readback. Preserve changed-profile refusal and fresh recovery; disclose restored bytes as information cost, not compression or model benefit. |
diff --git a/tests/control_plane/test_cli_program_route.py b/tests/control_plane/test_cli_program_route.py
index f0687fabda..f3f3ff9fed 100644
--- a/tests/control_plane/test_cli_program_route.py
+++ b/tests/control_plane/test_cli_program_route.py
@@ -1,6 +1,7 @@
 """An explicitly selected runtime must execute its own generated control loop."""
 from __future__ import annotations
 
+import json
 import os
 import shlex
 import subprocess
@@ -146,3 +147,91 @@ def run(argv, *, expected_code=0):
     settled = run(guard_args)
     assert settled["effective_action"] == "heartbeat_settled_skip"
     assert not marker.exists(), "A generated command escaped to the global runtime"
+
+
+@pytest.mark.skipif(os.name == "nt", reason="Exercises a POSIX console alias and poisoned PATH")
+@pytest.mark.parametrize("program", ["module", "candidate's cli"])
+@pytest.mark.parametrize("output_limit", [8192, 64000])
+@pytest.mark.parametrize("selection", ["raw", "selected", "index", "rejected"])
+def test_native_reader_continuation_and_recovery_execute_the_same_runtime(
+    tmp_path, monkeypatch, program, output_limit, selection,
+):
+    from loopx.cli_commands.quota_capture import native_context_capture_view, _native_context_read_bytes
+    from loopx.control_plane.turn_driver import codex_context as context
+    from test_codex_context_lifecycle import project, setup
+
+    setup(tmp_path, monkeypatch)
+    payload = {"ok": True, "goal_id": "fixture", "requirements": "Keep the complete 原始条件。\n" * 8000,
+               "items": [{"source": "original"} for _ in range(1000)]}
+    _, wire = project(tmp_path, payload=payload)
+    view = native_context_capture_view(wire, Path(wire["recovery"]["path"]).parent,
+                                      context_directory=str(tmp_path))
+    capture = Path(view["recovery"]["path"])
+    original, state = capture.read_bytes(), (tmp_path / "session.json").read_bytes()
+    bin_dir = tmp_path / "candidate bin"
+    bin_dir.mkdir()
+    marker = tmp_path / "wrong-runtime"
+    poison = bin_dir / "loopx"
+    poison.write_text(f"#!{sys.executable}\nfrom pathlib import Path\n"
+                      f"Path({str(marker)!r}).touch()\nraise SystemExit(97)\n")
+    poison.chmod(0o755)
+    monkeypatch.setenv("PATH", str(bin_dir) + os.pathsep + os.environ.get("PATH", ""))
+    prefix = [sys.executable, "-m", "loopx.entrypoint"]
+    if program != "module":
+        alias = bin_dir / program
+        alias.symlink_to(Path(sys.executable).parent / "loopx")
+        prefix = [str(alias), "--cli-bin", str(alias)]
+    instructions = context.codex_context_recovery_instructions(str(tmp_path), goal_id="fixture",
+        agent_id="agent", cli_argv=prefix)
+    command = instructions.split("Read missing necessary facts from this same capture with: ", 1)[1]
+    words = shlex.split(command.split(" --select '<pointer>'.", 1)[0])
+    for option in ("--page-bytes", "--max-page-output-bytes"):
+        words[words.index(option) + 1] = str(output_limit)
+    if selection == "selected":
+        words += ["--select", "/requirements"]
+    elif selection == "index":
+        words += ["--select", "/items", "--list"]
+    elif selection == "rejected":
+        words += ["--select", "/missing"]
+
+    content, pages, page_bytes = b"", 0, 0
+    while True:
+        assert words[:len(prefix)] == prefix
+        # Execute generated commands unchanged; do not strip or repair their launcher.
+        result = subprocess.run(words, cwd=tmp_path, capture_output=True, timeout=30)
+        page = json.loads(result.stdout)
+        assert len(result.stdout) <= output_limit
+        assert not marker.exists(), "The reader escaped to the global installation"
+        if not page["ok"]:
+            assert selection == "rejected" and pages == 0 and result.returncode == 1
+            assert "content" not in page
+            words = shlex.split(page["recovery_command"])
+            continue
+        assert result.returncode == 0 and page["offset"] == len(content), page
+        content += page["content"].encode()
+        pages += 1
+        page_bytes += len(result.stdout)
+        if page["eof"]:
+            break
+        assert page["next_offset"] == len(content)
+        words = shlex.split(page["next_command"])
+
+    if selection == "raw":
+        assert content == original and pages > 1
+        # Compare the real public printer against the production full/delta cost oracle.
+        from loopx.command_invocation import loopx_cli_route
+        with loopx_cli_route(None if program == "module" else prefix[0]):
+            reference_bytes = len((json.dumps(view["recovery"], ensure_ascii=False,
+                separators=(",", ":")) + "\n").encode())
+            assert _native_context_read_bytes(view["recovery"], original,
+                page_bytes=output_limit, output_bytes=output_limit) == reference_bytes + page_bytes
+    else:
+        [selected] = json.loads(content)["selections"]
+        if selection == "selected":
+            assert selected == {"pointer": "/requirements", "value": payload["requirements"]} and pages > 1
+        elif selection == "index":
+            assert [field["pointer"] for field in selected["fields"]] == [f"/items/{i}" for i in range(1000)]
+            assert pages > 1
+        else:
+            assert {field["pointer"] for field in selected["fields"]} == {"/ok", "/goal_id", "/requirements", "/items"}
+    assert capture.read_bytes() == original and (tmp_path / "session.json").read_bytes() == state
diff --git a/tests/control_plane/test_native_context_delivery_cost.py b/tests/control_plane/test_native_context_delivery_cost.py
index f370622331..996095af67 100644
--- a/tests/control_plane/test_native_context_delivery_cost.py
+++ b/tests/control_plane/test_native_context_delivery_cost.py
@@ -10,6 +10,7 @@
 import pytest
 
 from loopx.cli import build_parser
+from loopx.command_invocation import current_loopx_cli_argv
 from loopx.cli_commands.quota_capture import native_context_capture_view, read_context_capture
 from loopx.control_plane.quota.error_codes import QuotaDecisionTransportError
 from loopx.control_plane.quota.turn_envelope import prepare_context_delivery, restore_context_delivery
@@ -27,14 +28,21 @@ def _payload(repeats=24):
     }
 
 
+def _reader_args(command):
+    words, prefix = shlex.split(command), current_loopx_cli_argv()
+    if words[:len(prefix)] == prefix:
+        return build_parser().parse_args(words[len(prefix):])
+    assert words[0] == "loopx"  # existing in-process reference constructors
+    return build_parser().parse_args(words[1:])
+
+
 def _pages(reference):
     """Follow the actual advertised CLI reader, including its page budgets."""
-    parser = build_parser()
     command = reference["command"]
     pages = []
     while True:
-        args = parser.parse_args(shlex.split(command)[1:])
-        page = read_context_capture(args)
+        args = _reader_args(command)
+        page = read_context_capture(args, reader_argv=current_loopx_cli_argv())
         assert page["ok"], page
         pages.append(page)
         if page["eof"]:
@@ -296,7 +304,7 @@ def test_native_cost_matches_public_pages_through_a_parent_alias(tmp_path):
     assert len(pages) > 1 and "read_instruction" in pages[0]
     # The first command preserves the alias, but real continuation commands use
     # the longer physical parent. Cost all of them with the public reader.
-    continuation = build_parser().parse_args(shlex.split(pages[0]["next_command"])[1:])
+    continuation = _reader_args(pages[0]["next_command"])
     assert Path(continuation.capture_file) == capture
     assert b"".join(page["content"].encode() for page in pages) == raw
     assert _native_context_read_bytes(

From 1c3cce391d03bece79c8f34d8d1b851adc21aa51 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 18:51:46 +0800
Subject: [PATCH 2/3] fix(context): retain reader program through native
 navigation and result details

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 docs/reference/required-work-context.md       |  18 +--
 loopx/cli_commands/quota.py                   |  13 +-
 loopx/cli_commands/quota_capture.py           |  64 ++++++----
 .../references/repair-patterns.md             |   2 +-
 tests/control_plane/test_cli_program_route.py | 116 ++++++++++++++++--
 .../test_context_delivery_cli.py              |  15 ++-
 .../test_goal_state_context_transport.py      |   7 +-
 7 files changed, 176 insertions(+), 59 deletions(-)

diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md
index b28aab1fd3..c587374c35 100644
--- a/docs/reference/required-work-context.md
+++ b/docs/reference/required-work-context.md
@@ -175,12 +175,13 @@ model retention. These separate boundaries prevent transport loss from becoming
 either a repeated full-input loop or permission to act on missing requirements.
 No token savings or accepted outcome gains follow from instruction delivery alone.
 
-Commands returned by a running `context-read` keep its explicit CLI selection,
-or its current Python interpreter when no CLI was specified. This applies to
-continuations and rejected-selection recovery. Capture bytes, digests, cursors
-and authority are unchanged; native full/delta byte comparisons include this
-reader command metadata. Existing in-process reference constructors keep their
-default command route.
+Reader commands emitted by native quota, saved-context projection, bounded
+settlement results and a running `context-read` keep the explicit CLI selection,
+or the current Python interpreter when no CLI was specified. This covers initial
+navigation, complete reads, continuations and rejected-selection recovery.
+Capture bytes, digests, cursors and authority are unchanged; native full/delta
+byte comparisons include this reader command metadata. Existing in-process
+reference constructors keep their default command route.
 
 An enabled native **full or delta** reference for a recognized decision can include
 `first_read_command`, reading its current full capture without an older baseline.
@@ -279,8 +280,9 @@ reader，不重新运行首读导航；后续导航改变不能替换原选读
 及显式文件首读游标在选读摘要变化时仍拒绝，沿同一完整原 capture 引用恢复，不重新
 准入或改写旧描述文件。退役这些 legacy reader 前仍须对账原未完成 Turn；本切片
 保留历史解码与完整捕获恢复。
-实际运行的 `context-read` 返回续页或选读拒绝恢复命令时，保留显式指定的 CLI；
-未指定 CLI 则使用当前 Python 解释器。捕获正文、摘要、游标和权限不变，原生
+原生 quota、已保存上下文投影、有界结算结果和实际运行的 `context-read` 生成读取
+命令时，保留显式指定的 CLI；未指定 CLI 则使用当前 Python 解释器。首读导航、
+完整读取、续页及选读拒绝恢复均沿用该程序。捕获正文、摘要、游标和权限不变，原生
 full/delta 字节比较计入这些 reader 命令元数据；既有进程内引用构造器保留默认路由。
 首读可把已经完整选中的同级内容合并为其完整父节点，减少地址重复；任何未选中的
 可选或未知内容都会阻止合并；若数组合并会恢复原元素顺序、覆盖 owner 的读取优先级，
diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py
index 2f5afb956b..8d05efbc4c 100644
--- a/loopx/cli_commands/quota.py
+++ b/loopx/cli_commands/quota.py
@@ -321,6 +321,9 @@ def _emit_quota_result(
     print_payload: PrintPayload,
 ) -> int:
     """Capture the full decision before projecting and printing the CLI view."""
+    from ..command_invocation import current_loopx_cli_argv
+
+    reader_argv = current_loopx_cli_argv()
     # Compose optional capability guidance after recovery has chosen the plan,
     # before capture/compaction. The quota decision owner stays capability-neutral.
     cli = (payload.get("interaction_contract") or {}).get("cli_channel")
@@ -341,7 +344,7 @@ def _emit_quota_result(
             and getattr(args, "max_page_output_bytes", None) is not None):
         try:
             view = budgeted_settlement_result(payload, runtime_root=context.runtime_root,
-                output_bytes=args.max_page_output_bytes)
+                output_bytes=args.max_page_output_bytes, reader_argv=reader_argv)
         except QuotaDecisionTransportError as exc:
             # Transport failure cannot replace an already evaluated refusal or
             # committed receipt. Preserve unknown controls as well as identity;
@@ -367,10 +370,8 @@ def _emit_quota_result(
     delivery_request = None
     decision_identity = ("", "")
     if capture_directory is not None and payload.get("replan_action_packet"):
-        from ..command_invocation import current_loopx_cli_argv
-
         payload = {**payload, "replan_read_path": str(capture_directory / "decision.json"),
-            "replan_read_argv": current_loopx_cli_argv()}
+            "replan_read_argv": reader_argv}
     if args.quota_command == "should-run" and context is not None:
         # Capture the one liveness observation with the decision that produced
         # it. Presentation and evidence recovery must not record it again.
@@ -452,8 +453,8 @@ def _emit_quota_result(
             # The original authority decision remains separately captured.
             payload = native_context_capture_view(payload, capture_directory, context_directory=directory,
                 metadata={"capture_sha256": capture_id, **({"recovery": recovery} if recovery is not None else {})},
-                max_page_output_bytes=getattr(args, "max_page_output_bytes", None))
-            payload = model_context_reference(payload)
+                max_page_output_bytes=getattr(args, "max_page_output_bytes", None), reader_argv=reader_argv)
+            payload = model_context_reference(payload, reader_argv=reader_argv)
         except (QuotaDecisionTransportError, OSError, ValueError) as exc:
             if not isinstance(exc, QuotaDecisionTransportError):
                 exc = QuotaDecisionTransportError(
diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py
index 05b0d57cf4..37a31d584f 100644
--- a/loopx/cli_commands/quota_capture.py
+++ b/loopx/cli_commands/quota_capture.py
@@ -350,7 +350,7 @@ def read_context_capture(args: argparse.Namespace, *, presentation: dict[str, An
 
 
 def _capture_read_reference(value: dict[str, Any], *, page_bytes: int,
-                            output_bytes: int) -> dict[str, str]:
+                            output_bytes: int, reader_argv: list[str] | None = None) -> dict[str, str]:
     """Keep file identity and the budgeted reader at the transport boundary.
 
     Navigation lives on the outer view; complete paging obligations accompany
@@ -358,10 +358,10 @@ def _capture_read_reference(value: dict[str, Any], *, page_bytes: int,
     """
     return {"path": value["path"], "sha256": value["sha256"],
         "command": context_capture_read_command(Path(value["path"]), value["sha256"],
-            page_bytes=page_bytes, max_page_output_bytes=output_bytes)}
+            page_bytes=page_bytes, max_page_output_bytes=output_bytes, reader_argv=reader_argv)}
 
 
-def model_context_reference(view: dict[str, Any]) -> dict[str, Any]:
+def model_context_reference(view: dict[str, Any], *, reader_argv: list[str] | None = None) -> dict[str, Any]:
     """Present native navigation; machine references and journal bindings stay full.
 
     The descriptor holds complete digests locally. Legacy explicit readers and
@@ -381,7 +381,8 @@ def reference(value: dict[str, Any], *, navigation: bool = False) -> dict[str, s
         path = Path(value["path"])
         reader = write_context_read_reference(path, value["sha256"], navigation=value if navigation else None)
         return {"reference_file": str(reader), "command": context_capture_read_command(path, value["sha256"],
-            page_bytes=page_bytes, max_page_output_bytes=output_bytes, reference_file=reader)}
+            page_bytes=page_bytes, max_page_output_bytes=output_bytes, reference_file=reader,
+            reader_argv=reader_argv)}
 
     result = {"ok": True, "schema_version": "context_capture_navigation_v1", "mode": view["mode"],
         **reference(view, navigation=True), "total_bytes": view["total_bytes"], "recovery": reference(view["recovery"]),
@@ -395,7 +396,7 @@ def reference(value: dict[str, Any], *, navigation: bool = False) -> dict[str, s
         recovery = view["recovery"]
         result["first_read_command"] = context_capture_read_command(Path(recovery["path"]), recovery["sha256"],
             page_bytes=page_bytes, max_page_output_bytes=output_bytes, initial_read=True,
-            reference_file=Path(result["recovery"]["reference_file"]))
+            reference_file=Path(result["recovery"]["reference_file"]), reader_argv=reader_argv)
     result["instruction"] += (" Append --select '<pointer>' or --list to recovery.command for current fields. "
         "Reader references keep complete digests locally; the CLI checks them on every page. "
         "A reference or digest grants no authority or freshness.")
@@ -409,26 +410,29 @@ def _native_context_reference(projected: dict[str, Any], reference: dict[str, An
                               total_bytes: int, metadata: dict[str, Any] | None,
                               field_index: list[dict[str, Any]] | None,
                               initial_read_pointers: list[str] | None,
-                              page_bytes: int, output_bytes: int) -> dict[str, Any]:
+                              page_bytes: int, output_bytes: int,
+                              reader_argv: list[str] | None = None) -> dict[str, Any]:
     """Build the bounded native reference without publishing or registering it."""
     target = projected.get("target") if projected.get("schema_version") == "context_delivery_v1" else None
-    reference = _capture_read_reference(reference, page_bytes=page_bytes, output_bytes=output_bytes)
+    reference = _capture_read_reference(reference, page_bytes=page_bytes, output_bytes=output_bytes,
+        reader_argv=reader_argv)
     result = {"ok": True, **(metadata or {}), "mode": mode, **reference,
         **({"target": target} if target else {})}
     if isinstance(result.get("recovery"), dict):
         result["recovery"] = _capture_read_reference(result["recovery"],
-            page_bytes=page_bytes, output_bytes=output_bytes)
+            page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv)
     current_path = Path(current["path"])
     result.setdefault("recovery", _capture_read_reference(current,
-        page_bytes=page_bytes, output_bytes=output_bytes))
+        page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv))
     if mode != "full" and projected.get("continuity_basis") == "captured_file":
         result["continuity_basis"] = "captured_file"
         result["baseline_reference"] = dict(projected["baseline_reference"])
     result["total_bytes"] = total_bytes
     result["index_command"] = context_capture_read_command(current_path, current["sha256"],
-        page_bytes=min(page_bytes, 24000), max_page_output_bytes=min(output_bytes, 24000), list_fields=True)
+        page_bytes=min(page_bytes, 24000), max_page_output_bytes=min(output_bytes, 24000), list_fields=True,
+        reader_argv=reader_argv)
     result["select_command"] = context_capture_read_command(current_path, current["sha256"],
-        page_bytes=page_bytes, max_page_output_bytes=output_bytes) + " --select '<pointer>'"
+        page_bytes=page_bytes, max_page_output_bytes=output_bytes, reader_argv=reader_argv) + " --select '<pointer>'"
     # A complete scalar preview helps orientation without inventing a second
     # decision owner. Oversized or structured values remain discoverable in the
     # index; no prose is truncated or treated as an authorization summary.
@@ -449,7 +453,7 @@ def fits() -> bool:
             "without reading an older baseline. "
             "Read each complete optional attachment from its captured step before using it. " + instruction)
         result["first_read_command"] = context_capture_read_command(current_path, current["sha256"],
-            page_bytes=page_bytes, max_page_output_bytes=output_bytes, initial_read=True)
+            page_bytes=page_bytes, max_page_output_bytes=output_bytes, initial_read=True, reader_argv=reader_argv)
         if not fits():
             result.pop("first_read_command")  # A complete optional selection or the existing generic recovery.
             result["overview"] = overview
@@ -483,7 +487,8 @@ def _native_output_bytes(value: dict[str, Any]) -> bytes:
 
 def _native_context_read_bytes(reference: dict[str, Any], raw: bytes, *,
                                page_bytes: int, output_bytes: int,
-                               stop_after: int | None = None) -> int:
+                               stop_after: int | None = None,
+                               reader_argv: list[str] | None = None) -> int:
     """Count the reference plus its explicit-file reader before publication.
 
     Stop once this candidate cannot win. This measures one complete wire read,
@@ -499,7 +504,7 @@ def _native_context_read_bytes(reference: dict[str, Any], raw: bytes, *,
     path = Path(reference["path"]).expanduser().absolute()
     path = path.parent.resolve(strict=True) / path.name
     total, offset = len(_native_output_bytes(reference)), 0
-    reader_argv = current_loopx_cli_argv()
+    reader_argv = current_loopx_cli_argv() if reader_argv is None else reader_argv
     while True:
         page = _capture_page(path, raw, reference["sha256"],
             offset=offset, size=page_bytes, output_limit=output_bytes, instruction=False,
@@ -513,7 +518,8 @@ def _native_context_read_bytes(reference: dict[str, Any], raw: bytes, *,
 def native_context_capture_view(projected: dict[str, Any], directory: Path, *,
                                 context_directory: str | None = None,
                                 metadata: dict[str, Any] | None = None,
-                                max_page_output_bytes: int | None = None) -> dict[str, Any]:
+                                max_page_output_bytes: int | None = None,
+                                reader_argv: list[str] | None = None) -> dict[str, Any]:
     """Publish the cheaper readable native transport for an already-decided view.
 
     TS owns valid deltas and their bindings. This IO adapter can only fall back
@@ -561,7 +567,7 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
         return _native_context_reference(projected, ref, mode=kind, current=current,
             current_document=current_document, total_bytes=len(body), metadata=metadata,
             field_index=field_index, initial_read_pointers=initial_read_pointers,
-            page_bytes=page_bytes, output_bytes=output_bytes)
+            page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv)
 
     if typed and mode == "delta":
         result = None
@@ -570,7 +576,7 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
         try:
             result = view(mode, reference, raw)
             delta_cost = _native_context_read_bytes(result, raw,
-                page_bytes=page_bytes, output_bytes=output_bytes)
+                page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv)
         except (QuotaDecisionTransportError, ValueError) as exc:
             result, delta_error = None, exc
         try:
@@ -579,7 +585,8 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]:
             # This lower bound avoids paging a large body when delta wins already.
             if delta_cost is None or len(_native_output_bytes(full)) + len(current_raw) <= delta_cost:
                 full_cost = _native_context_read_bytes(full, current_raw,
-                    page_bytes=page_bytes, output_bytes=output_bytes, stop_after=delta_cost)
+                    page_bytes=page_bytes, output_bytes=output_bytes, stop_after=delta_cost,
+                    reader_argv=reader_argv)
                 if delta_cost is None or full_cost <= delta_cost:
                     result, publish_wire, ancestry = full, False, 0
         except (QuotaDecisionTransportError, ValueError):
@@ -611,6 +618,9 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]:
     with that same reference, never a replacement guard.
     Invalid run-mode opt-ins fail before that optional transport path.
     """
+    from ..command_invocation import current_loopx_cli_argv
+
+    reader_argv = current_loopx_cli_argv()
     try:
         activation = native_context_activation(args)
         path, raw, payload = _read_capture(args.capture_file, args.capture_sha256)
@@ -630,7 +640,7 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]:
             "recovery": "Restore the original digest-bound decision; no guard, admission or spend was performed."}
 
     reference = {"path": str(path), "sha256": args.capture_sha256,
-        "command": context_capture_read_command(path, args.capture_sha256)}
+        "command": context_capture_read_command(path, args.capture_sha256, reader_argv=reader_argv)}
     full_reference = dict(reference)
     mode = "full"
     target = None
@@ -650,7 +660,7 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]:
             page_bytes = min(page_bytes, output_bytes)
 
         full_reference = _capture_read_reference(full_reference,
-            page_bytes=page_bytes, output_bytes=output_bytes)
+            page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv)
         reference = dict(full_reference)
         directory = codex_context_directory(directory=args.codex_context_dir,
             root=args.codex_context_root, create=True)
@@ -682,15 +692,15 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]:
             # committed recovery even if optional wire/inline delivery fails;
             # the input capture digest below remains the original identity.
             full_reference = _capture_read_reference(projected["recovery"],
-                page_bytes=page_bytes, output_bytes=output_bytes)
+                page_bytes=page_bytes, output_bytes=output_bytes, reader_argv=reader_argv)
             reference = dict(full_reference)
         native_view = native_context_capture_view(projected, cache, context_directory=directory,
             metadata={"recovery": full_reference, "capture_sha256": args.capture_sha256,
                 "context_delivery": activation},
-            max_page_output_bytes=getattr(args, "max_page_output_bytes", None))
+            max_page_output_bytes=getattr(args, "max_page_output_bytes", None), reader_argv=reader_argv)
         reference = {key: native_view[key] for key in ("path", "sha256", "command")}
         mode, target = native_view["mode"], native_view.get("target")
-        native_view = model_context_reference(native_view)
+        native_view = model_context_reference(native_view, reader_argv=reader_argv)
     except (QuotaDecisionTransportError, OSError, ValueError, TypeError, RecursionError) as exc:
         if native_view is not None and not isinstance(exc, QuotaDecisionTransportError):
             # The native view and journal are already valid. Publishing the
@@ -710,7 +720,8 @@ def project_saved_context_capture(args: argparse.Namespace) -> dict[str, Any]:
                 page_bytes = min(page_bytes, output_bytes)
             recovery_index = {**full_reference, "command": context_capture_read_command(
                 Path(full_reference["path"]), full_reference["sha256"],
-                page_bytes=page_bytes, max_page_output_bytes=output_bytes, list_fields=True)}
+                page_bytes=page_bytes, max_page_output_bytes=output_bytes, list_fields=True,
+                reader_argv=reader_argv)}
             return {"ok": False, "error_code": exc.error_code,
                 "reason": " ".join(str(exc).split())[:500],
                 "heartbeat_receipt": receipt, "capture_sha256": args.capture_sha256,
@@ -824,6 +835,7 @@ def capture_decision(directory: Path | None, payload: dict[str, object]) -> None
 
 def budgeted_settlement_result(
     payload: dict[str, object], *, runtime_root: Path, output_bytes: int,
+    reader_argv: list[str] | None = None,
 ) -> dict[str, object]:
     """Present one evaluated result with its complete immutable diagnostic.
 
@@ -858,7 +870,7 @@ def fits(value: dict[str, object]) -> bool:
         details = {"reference_file": str(reference),
             "command": context_capture_read_command(saved, digest, reference_file=reference,
                 page_bytes=min(NATIVE_CONTEXT_PAGE_BYTES, output_bytes),
-                max_page_output_bytes=output_bytes)}
+                max_page_output_bytes=output_bytes, reader_argv=reader_argv)}
         deferred = [key for key in ("before", "after") if key in payload]
         view = {key: value for key, value in payload.items() if key not in deferred}
         view["result_details"] = {**details, "deferred_fields": deferred}
@@ -875,7 +887,7 @@ def fits(value: dict[str, object]) -> bool:
         view = {"ok": payload.get("ok"), "result_details": details,
             "command": context_capture_read_command(saved, digest, reference_file=reference, select=pointers,
                 page_bytes=min(NATIVE_CONTEXT_PAGE_BYTES, output_bytes),
-                max_page_output_bytes=output_bytes),
+                max_page_output_bytes=output_bytes, reader_argv=reader_argv),
             "instruction": "Read the complete selected result controls before acting. The operation was "
                 "already evaluated; preserve its original Turn/effect identity. No execution authority "
                 "or context baseline is granted. before/after remain available in the same immutable result."}
diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md
index 7d1ac3c537..8fcf13f220 100644
--- a/skills/loopx-self-repair/references/repair-patterns.md
+++ b/skills/loopx-self-repair/references/repair-patterns.md
@@ -6,7 +6,7 @@ teaches a reusable control-plane lesson.
 | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair |
 | --- | --- | --- | --- | --- |
 | `saved_projection_goal_route_coupling` | A saved context projection fails on conflicting default runtime roots even with native transport disabled, while the same capture remains readable. | Run the shipped projection and returned reader against one digest-bound synthetic capture, with implicit, missing and global registry selectors; retain digest/identity refusals and unchanged runtime snapshots. | Shared CLI registry discovery runs before the saved-view owner validates its source. | Classify the saved presentation at the existing CLI route owner; preserve machine-configuration routing, live authority conflict checks and source/session validation. Reuse the registry-independence integration tests rather than adding a second recovery path or repairing live state. |
-| `native_reader_program_route_loss` | A native context read succeeds through a selected CLI or interpreter, but its returned continuation or recovery command invokes another runtime. | Actual lifecycle capture, generated recovery instructions, unchanged execution of emitted commands with an isolated competing program, complete source/digest readback and installed-package provenance. | The reader used its generic command-renderer default after the caller had already selected the runtime. | Pass the current program argv through the existing reader and command renderer for continuation and rejected-selection recovery; retain constructor defaults and shell quoting. Include emitted command metadata in native byte-cost estimates. Cover raw, selected and index pagination, quoted aliases, refusal recovery and source-state immutability through real CLI and isolated wheel entrypoints. Routing preserves the selected program; it does not pin an installation version or prove Host adoption. |
+| `native_reader_program_route_loss` | A native producer or read uses a selected CLI or interpreter, but returned navigation, continuation or recovery invokes another runtime. | Actual lifecycle capture, generated first-read and recovery instructions, unchanged execution of emitted commands with an isolated competing program, complete source/digest readback and installed-package provenance. | A presentation layer used its generic command-renderer default after the caller had already selected the runtime. | Pass the current program argv through existing CLI, navigation, bounded-result and reader renderers; retain constructor defaults and shell quoting. Include emitted command metadata in native byte-cost estimates. Cover cold guard and saved-context projection before direct first reads, raw/selected/index pagination, quoted aliases, refusal details and source-state immutability through real CLI and isolated wheel entrypoints. Routing preserves the selected program; it does not pin an installation version or prove Host adoption. |
 | `host_bridge_reader_orphan` | A rejected bridge observation leaves its reader thread and output pipe alive after the child has exited. | Original framing error, cleanup result, child exit, bounded-queue backpressure and reader/pipe state across repeated failures. | Cancellation rejects another malformed frame; the sole consumer leaves while the producer blocks on a full queue, and joining alone cannot release it. | Stop publication, release the waiting producer, join and close through the existing transport helper. Retain cleanup-unconfirmed results and original Host drain ownership; reclaiming a reader does not prove Host exit. Cover malformed JSON and invalid typed observations with real subprocesses, plus existing real Host cancellation and success paths. Do not increase timeouts or fabricate a successful result. |
 | `codex_strict_result_schema_gap` | A Codex Turn fails with invalid_json_schema before execution, or a strict-schema stop result fails internal evidence validation. | Generated Codex schema, required properties, unsupported conditional keywords, and both CLI/app-server result boundaries. | Optional evidence and conditional schema requirements conflict with strict Structured Outputs; internal stop results prohibit evidence fields. | Require nullable evidence fields on the Codex wire and remove unsupported conditionals. Normalize only stop null placeholders in both Codex transports; reject missing/null profile material evidence and preserve non-null evidence for the existing validator. Cover legacy absence, invalid arrays and an isolated schema-acceptance probe without changing other hosts. |
 | `initial_preference_view_omission` | A fulfilled preference hook leaves corrections, retirement or advisory instructions absent from the recommended initial read. | The preference owner's complete current scoped view, actual selected reader/native input, empty and unavailable states, and same-capture continuation/recovery. | A hand-picked projection uses incomplete or stale child field names after the owner has already delivered the current view. | Reuse the complete owned current view in the existing initial selector; retain provenance, advisory authority, exact-scope freshness and immutable capture checks. Cover real File/SQLite lifecycle reads, empty replacements, unknown values and native consumer readback. Preserve changed-profile refusal and fresh recovery; disclose restored bytes as information cost, not compression or model benefit. |
diff --git a/tests/control_plane/test_cli_program_route.py b/tests/control_plane/test_cli_program_route.py
index f3f3ff9fed..3a5adbfb94 100644
--- a/tests/control_plane/test_cli_program_route.py
+++ b/tests/control_plane/test_cli_program_route.py
@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import json
+import hashlib
 import os
 import shlex
 import subprocess
@@ -16,6 +17,23 @@
 cli = journey.cli
 
 
+def _reader_program(tmp_path, monkeypatch, program):
+    bin_dir = tmp_path / "candidate bin"
+    bin_dir.mkdir()
+    marker = tmp_path / "wrong-runtime"
+    poison = bin_dir / "loopx"
+    poison.write_text(f"#!{sys.executable}\nfrom pathlib import Path\n"
+                      f"Path({str(marker)!r}).touch()\nraise SystemExit(97)\n")
+    poison.chmod(0o755)
+    monkeypatch.setenv("PATH", str(bin_dir) + os.pathsep + os.environ.get("PATH", ""))
+    prefix = [sys.executable, "-m", "loopx.entrypoint"]
+    if program != "module":
+        alias = bin_dir / program
+        alias.symlink_to(Path(sys.executable).parent / "loopx")
+        prefix = [str(alias), "--cli-bin", str(alias)]
+    return prefix, marker
+
+
 def test_manager_required_read_uses_the_same_invocation_route(tmp_path):
     from loopx.command_invocation import loopx_cli_route
     from loopx.capabilities.manager_context import turn_start_hook
@@ -168,19 +186,7 @@ def test_native_reader_continuation_and_recovery_execute_the_same_runtime(
                                       context_directory=str(tmp_path))
     capture = Path(view["recovery"]["path"])
     original, state = capture.read_bytes(), (tmp_path / "session.json").read_bytes()
-    bin_dir = tmp_path / "candidate bin"
-    bin_dir.mkdir()
-    marker = tmp_path / "wrong-runtime"
-    poison = bin_dir / "loopx"
-    poison.write_text(f"#!{sys.executable}\nfrom pathlib import Path\n"
-                      f"Path({str(marker)!r}).touch()\nraise SystemExit(97)\n")
-    poison.chmod(0o755)
-    monkeypatch.setenv("PATH", str(bin_dir) + os.pathsep + os.environ.get("PATH", ""))
-    prefix = [sys.executable, "-m", "loopx.entrypoint"]
-    if program != "module":
-        alias = bin_dir / program
-        alias.symlink_to(Path(sys.executable).parent / "loopx")
-        prefix = [str(alias), "--cli-bin", str(alias)]
+    prefix, marker = _reader_program(tmp_path, monkeypatch, program)
     instructions = context.codex_context_recovery_instructions(str(tmp_path), goal_id="fixture",
         agent_id="agent", cli_argv=prefix)
     command = instructions.split("Read missing necessary facts from this same capture with: ", 1)[1]
@@ -235,3 +241,87 @@ def test_native_reader_continuation_and_recovery_execute_the_same_runtime(
         else:
             assert {field["pointer"] for field in selected["fields"]} == {"/ok", "/goal_id", "/requirements", "/items"}
     assert capture.read_bytes() == original and (tmp_path / "session.json").read_bytes() == state
+
+
+@pytest.mark.skipif(os.name == "nt", reason="Exercises a POSIX console alias and poisoned PATH")
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("program", ["module", "candidate's cli"])
+@pytest.mark.parametrize("entry", ["guard", "project"])
+@pytest.mark.parametrize("output_limit", [8192, 64000])
+def test_native_cli_navigation_runs_initial_and_complete_reads_on_same_program(
+    tmp_path, monkeypatch, provider, program, entry, output_limit,
+):
+    from loopx.control_plane.turn_driver import codex_context as context
+
+    project, runtime, registry, _, _ = journey._source(tmp_path, provider=provider,
+        extra=f"claimed_by={cli.AGENT_ID}")
+    monkeypatch.setenv("LOOPX_USAGE_PING", "0")
+    monkeypatch.setenv("CODEX_THREAD_ID", "native-navigation")
+    root, captures = tmp_path / "context", tmp_path / "captures"
+    root.mkdir(mode=0o700)
+    captures.mkdir(mode=0o700)
+    directory = context.codex_context_directory(root=str(root), create=True)
+    context.codex_context_hook(directory, dict(session_id="native-navigation", cwd=str(project),
+        hook_event_name="SessionStart", source="startup"))
+    prefix, marker = _reader_program(tmp_path, monkeypatch, program)
+    options = ["--codex-app", "--goal-id", cli.GOAL_ID, "--agent-id", cli.AGENT_ID,
+        "--turn-instance-id", cli.TURN_ID, "--context-delivery", "on",
+        "--codex-context-root", str(root), "--decision-output-root", str(captures),
+        "--max-page-output-bytes", str(output_limit)]
+
+    def run(words, expected_code=0):
+        result = subprocess.run(words, cwd=project, capture_output=True, timeout=60)
+        assert result.returncode == expected_code, (result.returncode, result.stdout, result.stderr)
+        assert not marker.exists(), "Native navigation escaped to the global program"
+        return json.loads(result.stdout)
+
+    view = run([*prefix, "--format", "json", "--registry", str(registry),
+        "--runtime-root", str(runtime), "quota", "should-run", "--todo-id", cli.TODO_ID, *options])
+    [decision] = captures.glob("decision-*/decision.json")
+    original = decision.read_bytes()
+    assert cli._heartbeat_receipt_count(runtime, cli.TURN_ID) == 1 and cli._spend_run_count(runtime) == 0
+    if entry == "project":
+        context.codex_context_hook(directory, dict(session_id="native-navigation", cwd=str(project),
+            hook_event_name="PostCompact"))
+        view = run([*prefix, "--format", "json", "quota", "context-project", *options,
+            "--capture-file", str(decision), "--capture-sha256", hashlib.sha256(original).hexdigest()])
+    state = (Path(directory) / "session.json").read_bytes()
+
+    def read(command):
+        content = b""
+        while True:
+            words = shlex.split(command)
+            page = run(words)  # Execute exactly what the real CLI returned.
+            assert words[:len(prefix)] == prefix
+            assert page["offset"] == len(content)
+            content += page["content"].encode()
+            if page["eof"]:
+                return json.loads(content)
+            assert page["next_offset"] == len(content)
+            command = page["next_command"]
+
+    complete = read(view["recovery"]["command"])
+    assert read(view["command"]) == complete
+    first = read(view["first_read_command"])
+    assert any(row.get("pointer") == "/goal_id" and row.get("value") == cli.GOAL_ID
+               for row in first["selections"])
+    for row in first["selections"]:
+        if "value" not in row:
+            continue
+        expected = complete
+        for token in row["pointer"].split("/")[1:]:
+            token = token.replace("~1", "/").replace("~0", "~")
+            expected = expected[int(token)] if isinstance(expected, list) else expected[token]
+        assert row["value"] == expected
+    # A refused spend is still an evaluated result: read its bounded diagnostic,
+    # rather than replaying the operation or escaping to another installation.
+    refused = run([*prefix, "--format", "json", "--registry", str(registry),
+        "--runtime-root", str(runtime), "quota", "spend-slot", "--goal-id", cli.GOAL_ID,
+        "--agent-id", cli.AGENT_ID, "--todo-id", cli.TODO_ID, "--turn-instance-id", cli.TURN_ID,
+        "--execute", "--max-page-output-bytes", "4096"], expected_code=1)
+    assert refused["ok"] is False
+    result = read(refused["result_details"]["command"])
+    assert result["ok"] is False and result["goal_id"] == cli.GOAL_ID
+    assert result["mode"] == "spend-slot" and not result["appended"] and not result["registry_mutated"]
+    assert decision.read_bytes() == original and (Path(directory) / "session.json").read_bytes() == state
+    assert cli._heartbeat_receipt_count(runtime, cli.TURN_ID) == 1 and cli._spend_run_count(runtime) == 0
diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py
index ec190f28cc..ef987d1551 100644
--- a/tests/control_plane/test_context_delivery_cli.py
+++ b/tests/control_plane/test_context_delivery_cli.py
@@ -27,6 +27,15 @@
 )
 
 
+def _native_reader_args(command):
+    from loopx.cli import build_parser
+    from loopx.command_invocation import current_loopx_cli_argv
+
+    words, prefix = shlex.split(command), current_loopx_cli_argv()
+    assert words[:len(prefix)] == prefix
+    return build_parser().parse_args(words[len(prefix):])
+
+
 def test_relocated_current_source_overlay_crosses_the_real_typed_codec(tmp_path, monkeypatch):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     scope = dict(goal_id="fixture", agent_id="reader", session_id="native", context_epoch="epoch")
@@ -495,7 +504,7 @@ def consume(page):
             assert wire["mode"] == "full"
         # Autonomous readers can choose the current full file directly, whether
         # the advertised transport is full or delta. No hook/read ACK is needed.
-        args = build_parser().parse_args(shlex.split(wire["recovery"]["command"])[1:])
+        args = _native_reader_args(wire["recovery"]["command"])
         assert args.max_page_output_bytes == limit
         observed = consume(read_context_capture(args))
         recovery = read_context_read_reference(wire["recovery"]["reference_file"])[1]
@@ -508,7 +517,7 @@ def consume(page):
                 "--decision-output-root", str(projection_root), "--context-delivery", "on",
                 "--codex-context-dir", str(root), "--max-page-output-bytes", str(limit), cwd=project)
             assert rc == 0 and projected["ok"], projected
-            args = build_parser().parse_args(shlex.split(projected["recovery"]["command"])[1:])
+            args = _native_reader_args(projected["recovery"]["command"])
             assert consume(read_context_capture(args)) == observed
     assert _heartbeat_receipt_count(runtime, "native-full") == 1
     assert _spend_run_count(runtime) == 0 and state.read_bytes() == original
@@ -546,7 +555,7 @@ def guard(turn, cwd=project):
         return view
 
     def consume(view):
-        page = read_context_capture(build_parser().parse_args(shlex.split(view["recovery"]["command"])[1:]))
+        page = read_context_capture(_native_reader_args(view["recovery"]["command"]))
         observed = b""
         while True:
             assert page["offset"] == len(observed)
diff --git a/tests/control_plane/test_goal_state_context_transport.py b/tests/control_plane/test_goal_state_context_transport.py
index 883542bf22..575d392bc2 100644
--- a/tests/control_plane/test_goal_state_context_transport.py
+++ b/tests/control_plane/test_goal_state_context_transport.py
@@ -87,8 +87,11 @@ def guard():
     assert code == 0 and page["ok"], (page.get("reason"), page.get("error"), page.get("error_code"))
     digest = reference_metadata_oracle(page)["sha256"]
     if page.get("schema_version") != "context_capture_page_v1":
-        page = read_context_capture(build_parser().parse_args(
-            shlex.split(page["command"])[1:] + ["--page-bytes", "3072"]))
+        from test_context_delivery_cli import _native_reader_args
+
+        args = _native_reader_args(page["command"])
+        args.page_bytes = 3072
+        page = read_context_capture(args)
         assert page["ok"], page
     while True:
         assert page["offset"] == len(raw) and page["sha256"] == digest

From 810ac2af573fd14f8e9facc5464ad5cc0c8b988e Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 19:38:18 +0800
Subject: [PATCH 3/3] test(context): prove selected reader routes through
 native Codex tools

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Signed-off-by: huangruiteng <huangrt01@163.com>
---
 .../test_context_delivery_cli.py              | 30 ++++++++-
 .../test_saved_context_project.py             | 62 ++++++++++++++++---
 2 files changed, 83 insertions(+), 9 deletions(-)

diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py
index ef987d1551..aaddae4a2d 100644
--- a/tests/control_plane/test_context_delivery_cli.py
+++ b/tests/control_plane/test_context_delivery_cli.py
@@ -72,7 +72,35 @@ def visit(node):
 
 
 @pytest.mark.parametrize("body", ["plain text " * 900, '汉🙂"\\\t\n' * 700])
-@pytest.mark.parametrize("limit", [1600, 7000])
+def test_reader_rejects_budget_too_small_for_reader_program_metadata(
+    tmp_path, monkeypatch, body,
+):
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry = _write_fixture(tmp_path)
+    capture = tmp_path / "original.json"
+    raw = (json.dumps({"requirements": body}, ensure_ascii=False) + "\n").encode()
+    capture.write_bytes(raw)
+    digest = hashlib.sha256(raw).hexdigest()
+    limit = 1600
+    response = subprocess.run(
+        [sys.executable, "-m", "loopx.entrypoint", "--format", "json", "quota",
+         "context-read", "--capture-file", str(capture),
+         "--capture-sha256", digest, "--page-bytes", "8192",
+         "--max-page-output-bytes", str(limit)],
+        capture_output=True, cwd=Path(__file__).resolve().parents[2], check=False,
+    )
+    assert response.returncode == 1
+    assert len(response.stdout) <= limit
+    rejected = json.loads(response.stdout)
+    assert rejected["ok"] is False and "content" not in rejected
+    assert "same capture" in rejected["reason"]
+    assert capture.read_bytes() == raw
+    assert _heartbeat_receipt_count(runtime, "reader-budget") == 0
+    assert _spend_run_count(runtime) == 0
+
+
+@pytest.mark.parametrize("body", ["plain text " * 900, '汉🙂"\\\t\n' * 700])
+@pytest.mark.parametrize("limit", [4096, 7000])
 def test_reader_bounds_actual_json_stdout_and_preserves_every_byte(tmp_path, monkeypatch, body, limit):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     project, runtime, registry = _write_fixture(tmp_path)
diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py
index b826d9e48a..0756f4a688 100644
--- a/tests/control_plane/test_saved_context_project.py
+++ b/tests/control_plane/test_saved_context_project.py
@@ -500,7 +500,7 @@ def consume(wire):
 
 
 @pytest.mark.skipif(not os.environ.get("LOOPX_NATIVE_CHECKPOINT_BIN"), reason="Explicit native Codex; loopback inference only")
-@pytest.mark.parametrize("host", ["app_server", "benchmark", "benchmark_progressive", "app_machine", "app_code_mode", "app_progressive", "app_progressive_large"])
+@pytest.mark.parametrize("host", ["app_server", "benchmark", "benchmark_progressive", "app_machine", "app_code_mode", "app_progressive", "app_progressive_large", "app_program_route", "app_code_mode_program_route"])
 def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, monkeypatch, host):
     from canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime, read_fixture_todos
     from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
@@ -519,6 +519,10 @@ def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, mo
     benchmark = host in ("benchmark", "benchmark_progressive")
     progressive = host in ("app_progressive", "benchmark_progressive", "app_progressive_large")
     large_capture = host == "app_progressive_large"
+    routed_program = host.endswith("_program_route")
+    code_mode = host.startswith("app_code_mode")
+    if routed_program and os.name == "nt":
+        pytest.skip("POSIX console alias and controlled PATH")
     unicode_requirement = "保留原始目标、范围与停止条件🙂\n" * 120
     supporting_material = "资料🙂" * 390_000 if large_capture else ""
     binary, env, workspace, home = _native_fixture(tmp_path)
@@ -555,6 +559,20 @@ def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, mo
     cache.mkdir(mode=0o700)
     program = os.environ.get("LOOPX_NATIVE_CONTEXT_CLI") or str(Path(sys.executable).with_name("loopx"))
     cli = [program, "--cli-bin", program]
+    wrong_runtime = workspace / "wrong-runtime"
+    if routed_program:
+        bin_dir = workspace / "candidate bin"
+        bin_dir.mkdir()
+        poison = bin_dir / "loopx"
+        poison.write_text(f"#!{sys.executable}\nfrom pathlib import Path\n"
+                          f"Path({str(wrong_runtime)!r}).touch()\nraise SystemExit(97)\n")
+        poison.chmod(0o755)
+        candidate = bin_dir / "candidate's cli"
+        candidate.symlink_to(Path(program))
+        cli = ([sys.executable, "-m", "loopx.entrypoint"] if code_mode
+               else [str(candidate), "--cli-bin", str(candidate)])
+        cli += ["--registry", str(registry), "--runtime-root", str(runtime)]
+        env["PATH"] = str(bin_dir) + os.pathsep + env["PATH"]
     from loopx.control_plane.turn_driver.codex_context import codex_context_configuration
     configuration = codex_context_configuration(str(context), goal_id=GOAL_ID, agent_id=AGENT_ID, cli_argv=cli, scenario="codex_app_heartbeat")
     if host == "app_machine":
@@ -571,13 +589,27 @@ def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, mo
     read_offsets = {}
     actual_turn = None
     rejected_read = False
+    reading_initial = False
+    initial_content = {}
 
     def response(request, number, message):
-        nonlocal phase, queue, actual_turn, receipts, rejected_read
+        nonlocal phase, queue, actual_turn, receipts, rejected_read, reading_initial
         try:
             if continuation := native_process_continuation(request, f"projection_{number}",
-                    code_mode=host == "app_code_mode"):
+                    code_mode=code_mode):
                 return continuation
+            if reading_initial:
+                name = phase.removesuffix("_read")
+                page = last_context_page(request, allow_selection=True)
+                assert page and page["ok"] and page["offset"] == len(initial_content[name])
+                initial_content[name] += page["content"].encode()
+                assert len(initial_content[name]) == (page["total_bytes"] if page["eof"] else page["next_offset"])
+                if page["eof"]:
+                    from tests.control_plane.codex_checkpoint_fixture import selected_context_values
+                    assert selected_context_values(initial_content[name].decode())["/goal_id"] == GOAL_ID
+                    reading_initial = False
+                else:
+                    queue.insert(0, page["next_command"])
             if rejected_read:
                 output = [item["output"] for item in request["input"]
                     if item.get("type") in ("function_call_output", "custom_tool_call_output")][-1]
@@ -632,6 +664,13 @@ def response(request, number, message):
                 # guessing offsets from the capture size. This keeps each
                 # request's real page budget and avoids overlapping reads.
                 queue.append(navigation["command"])
+                if routed_program and descriptor["mode"] == "full":
+                    # Execute the returned initial navigation unchanged inside
+                    # real native tools, then retain the existing full-source
+                    # oracle and lifecycle proof. PATH's ambient CLI must fail.
+                    initial_content[name] = b""
+                    reading_initial = True
+                    queue.insert(0, navigation["first_read_command"])
                 if progressive:
                     # Exercise the actual small reference and selective public
                     # reader in native model input; never consume either file in full.
@@ -704,14 +743,19 @@ def response(request, number, message):
                     if benchmark:
                         return message
                     return response(request, number, message)
-            if host == "app_code_mode":
+            arguments = {"cmd": command, "max_output_tokens": 22000}
+            if routed_program:
+                # A login shell may replace the controlled PATH before it
+                # executes the otherwise unchanged generated reader command.
+                arguments["login"] = False
+            if code_mode:
                 return {"type": "custom_tool_call", "id": f"fc_{number}", "call_id": f"call_{number}",
                     "name": "exec", "input": '// @exec: {"max_output_tokens":24000}\ntext((await tools.exec_command('
-                        + json.dumps({"cmd": command, "max_output_tokens": 22000}) + ')).output);'}
+                        + json.dumps(arguments) + ')).output);'}
             # Match the generated native profile's 64000-byte page transport;
             # the old 3000-token tool cap truncated its valid JSON first page.
             return {"type": "function_call", "id": f"fc_{number}", "call_id": f"call_{number}",
-                "name": "exec_command", "arguments": json.dumps({"cmd": command, "max_output_tokens": 22000})}
+                "name": "exec_command", "arguments": json.dumps(arguments)}
         except Exception as error:
             errors.append(error)
             return message
@@ -759,11 +803,13 @@ def response(request, number, message):
             config.write_text(overrides + config.read_text())
             with (home / "config.toml").open("a") as config:
                 config.write("[sandbox_workspace_write]\nnetwork_access=true\n")
-            if host == "app_code_mode":
+            if code_mode:
                 config = home / "config.toml"
                 config.write_text(config.read_text().replace("[features]\n", "[features]\ncode_mode=true\n"))
             _app_server(binary, env, workspace, home, True, sandbox="workspace-write")
-    assert not errors and phase == "done", errors
+    assert not errors and phase == "done", (errors, phase, wrong_runtime.exists())
+    if routed_program:
+        assert set(initial_content) == {"cold"} and not wrong_runtime.exists()
     if benchmark:
         turns = [json.loads(raw)["heartbeat_receipt"]["turn_instance_id"] for _, raw in sources]
         assert len(set(turns)) == 2