fix(context): preserve native reuse and budget complete reads

loopx/control_plane/turn_driver/codex_context.py

@@ -180,7 +180,12 @@ def codex_context_recovery_instructions(directory: str, *, goal_id: str, agent_i
         "(including compaction or a failed hook), recover before dependent work with: " + command + ". "
         "Follow the returned next_command through eof; verify complete bytes and digest. "
         "If lookup fails, restore the original capture/reference; do not guess, readmit or ask the owner. "
-        "Recovery never renews authority/freshness or proves retention. Independent authorized work can continue."
+        "Recovery never renews authority/freshness or proves retention. Independent authorized work can continue. "
+        "For code-mode reads, size batches against both direct-result and retained-history output limits; "
+        "max_output_tokens alone does not establish the latter. Start with at most two default pages when "
+        "the effective history limit is unknown. With a known sufficient limit, larger batches can reduce "
+        "round trips; emit each stdout separately and check complete model-visible bytes. This advice "
+        "grants no authority to change host limits or treat complete runtime output as retention proof."
     )
 
 

loopx/control_plane/turn_driver/codex_input_trace.py

@@ -45,6 +45,29 @@ def _encoded_native_document(value: object) -> str:
     return json.dumps(_canonical_native_json(value), sort_keys=True, ensure_ascii=False, allow_nan=False)
 
 
+def _code_mode_source_matches(model_source: object, executed_source: str) -> bool:
+    if not isinstance(model_source, str):
+        return False
+    if model_source == executed_source:
+        return True
+    # Native code mode removes its optional first-line JSON pragma before
+    # recording the executable cell. Preserve the exact remaining source;
+    # arbitrary comments, substrings and additional rewrites prove nothing.
+    first, newline, body = model_source.partition("\n")
+    first = first.lstrip(" \t")
+    prefix = "// @exec:"
+    if not newline or body != executed_source or not first.startswith(prefix):
+        return False
+    try:
+        options = json.loads(first[len(prefix):])
+        json.dumps(options, allow_nan=False)
+    except (ValueError, TypeError):
+        return False
+    # Settings and execution remain native-owned. This only binds the body
+    # recorded by the live cell to the model's exact tool call.
+    return isinstance(options, dict)
+
+
 def _native_tool_is_bound(tail: list[dict[str, Any]], completed: list[dict[str, Any]], tools: list[dict[str, Any]], output: dict[str, Any], tool_id: str) -> bool:
     tool = tools[0]["payload"]
     if tool.get("requester") == {"type": "model"}:
@@ -71,7 +94,7 @@ def _native_tool_is_bound(tail: list[dict[str, Any]], completed: list[dict[str,
                              and row["payload"].get("runtime_cell_id") == cell_id for row in tail)
                  and any(item.get("type") == "custom_tool_call" and item.get("name") == "exec"
                          and item.get("call_id") == cells[0]["payload"].get("model_visible_call_id")
-                         and item.get("input") == cells[0]["payload"].get("source_js")
+                         and _code_mode_source_matches(item.get("input"), cells[0]["payload"]["source_js"])
                          for item in output.get("output_items", [])))
     return bound
 

skills/loopx-self-repair/references/repair-patterns.md

@@ -5,6 +5,8 @@ teaches a reusable control-plane lesson.
 
 | Pattern | Symptoms | Evidence To Read | Likely Root | Durable Repair |
 | --- | --- | --- | --- | --- |
+| `native_code_cell_source_normalization` | An optional native context adapter always falls back to full after an output-budget pragma, despite complete retained input. | The same live model call, native cell source, child identity and complete model-visible capture bytes; compare the executable source separately from native framing. | Native removes the documented first-line execution pragma before recording the cell, while the adapter compares that body with the unprocessed model call. | Normalize only that first-line JSON-object pragma and retain exact remaining source, live call/cell ordering and complete-byte verification. Cover unchanged framing, altered bodies, malformed/extra headers and stale or partial input, plus real native resume and recovery. Preserve automatic same-capture full fallback; do not add an admission or owner question. |
+| `native_history_output_budget` | A complete CLI or code-cell result becomes a clipped or missing page in the next model input despite a larger direct-result budget. | Compare the CLI result, completed native cell, actual inference input and effective model history-output policy at the same native version. | Direct-result and retained-history limits are separate; an unknown loopback model may use a smaller byte-based fallback than the intended host's token-based policy. | Keep the original low-budget failures. Qualify a distinct matched workload under the intended declared history budget without changing live host limits or frozen criteria. Size batches within both budgets, retain complete-byte checks and same-capture recovery, and never count incomplete trials as savings or runtime output as retention proof. |
 | `archive_verification_byte_epoch_gap` | Backup preview or execution accepts compressed input rewritten after preflight; restoring it before the final hash conceals the change. | External manifest and preflight archive identity, the digest/count of bytes actually consumed by gzip/tar, copy observations and destination publication. | Separate mutable-file reads supply the hash and parser facts; a final hash cannot prove which bytes were parsed. | Keep early wrong-digest refusal. Generate a bounded compressed-byte witness in the parser's actual input stream, compare it with preflight before audit, and compare copy witnesses before publication. Preserve capacity checks, inert status and no execution authority. Cover preview, execute and ABA restoration with real producer archives and filesystem fault injection. |
 | `legacy_goal_command_current_work_gap` | A configured Goal command remains executable heartbeat guidance after current Todo selection, a quiet/replan route, or settlement has superseded it. | The native full decision and original capture, exact selected/adaptive work, final interaction channels, and command provenance. | Legacy recommendation precedence or copied refinements retain a command without binding it to current work; a later reentry can supersede an earlier projection again. | Keep legacy fields at their existing typed quota projection boundary and derive admission, action and spend policy from the canonical interaction owner on each reentry. Preserve source configuration, admitted unscoped command compatibility and exact governed-intent commands. Exercise real File/SQLite CLI selection, default/detail/Envelope views, capture, settlement and replay; never execute a stale command or create a human gate to resolve a projection contradiction. |
 | `outcome_review_source_capacity` | Repeated milestone reviews fail with `goal_acceptance_invalid_request` and `outcome sources exceeds its supported capacity` or `outcome source capacity reached`. | The exact failed operation/receipt, inspected original outcome sources and requirements, and its review evidence; verify whether refinement committed before a later writeback failure. | Requirement provenance is used as an append-only progress log; older native/CLI contracts required a source even for a review. | Keep the original outcome and all origins. Existing-outcome reviews/work links omit source (native: source_summary); put observations in review evidence/notes and owning materials. Creation/new requirements still need real provenance. On an older build, use an inspected existing source with exactly the same ref/summary for unchanged requirements only. Never prune origins, inflate limits, invent a new outcome to escape capacity, or silently relabel a new request. Preserve failed receipts and exact-operation recovery; qualify repeated reviews at the source limit and ownership/CAS/replay negative cases through real backends. |

tests/control_plane/test_codex_context_boundary.py

@@ -121,6 +121,12 @@ def _native_fixture(tmp_path):
     workspace.mkdir()
     env = {k: v for k, v in os.environ.items() if k in {
         "PATH", "HOME", "USER", "LOGNAME", "SHELL", "TMPDIR", "LANG", "LC_ALL", "SYSTEMROOT"}}
+    # Authority fixtures isolate temp files below tmp_path. Native release
+    # helpers refuse a CODEX_HOME inside their temp root, which otherwise
+    # breaks Linux sandbox dispatch. Keep both private, as sibling directories.
+    temporary = tmp_path / "native-tmp"
+    temporary.mkdir(mode=0o700)
+    env.update({key: str(temporary) for key in ("TMPDIR", "TEMP", "TMP")})
     env["CODEX_HOME"] = str(home)
     return binary, env, workspace, home
 

tests/control_plane/test_codex_context_capture.py

@@ -69,7 +69,7 @@ def _delivered(request, digest):
   + [(host, surface, False, True, False) for host in ("benchmark", "code_mode")
      for surface in ("quota_packet", "turn_envelope")]
   + [("code_mode", surface, False, True, True) for surface in ("quota_packet", "turn_envelope")])
-def test_native_cli_capture_delta_and_compact_recovery(tmp_path, monkeypatch, host, surface, failed_primary, resumed, adaptive, hook_loss=None, shared_root=False, checkpoint_handoff=False, handoff_socket=None):
+def test_native_cli_capture_delta_and_compact_recovery(tmp_path, monkeypatch, host, surface, failed_primary, resumed, adaptive, hook_loss=None, shared_root=False, checkpoint_handoff=False, handoff_socket=None, native_pragma=False, native_read_batch=2, native_history_tokens=None):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     binary, env, workspace, home = _native_fixture(tmp_path)
     env["CODEX_ROLLOUT_TRACE_ROOT"] = str(workspace / "traces")
@@ -323,20 +323,21 @@ def response(request, number, message):
 
     def tool(cmd, number):
         if host == "code_mode":
+            pragma = '// @exec: {"max_output_tokens":12000}\n' if native_pragma else ""
             # Exercise the shipped default page size and bounded batching for
             # both output forms. One model round per page can exhaust the
             # unchanged native turn deadline before same-capture recovery ends.
             commands = [cmd]
-            if stages["pages"]:
+            while stages["pages"] and len(commands) < native_read_batch:
                 commands.append(stages["pages"].pop(0))
             if adaptive:
                 code = "".join("{const r=await tools.exec_command(" + json.dumps({
                     "cmd": command, "max_output_tokens": 2400})
                     + ");text(r.exit_code === 0 ? r.output : r);}" for command in commands)
                 return {"type": "custom_tool_call", "id": f"ctc_{number}", "call_id": f"call_{number}",
-                    "name": "exec", "input": code}
+                    "name": "exec", "input": pragma + code}
             return {"type": "custom_tool_call", "id": f"ctc_{number}", "call_id": f"call_{number}",
-                "name": "exec", "input": "".join("text(await tools.exec_command(" + json.dumps({
+                "name": "exec", "input": pragma + "".join("text(await tools.exec_command(" + json.dumps({
                     "cmd": command, "max_output_tokens": 2000}) + "));" for command in commands)}
         return {"type": "function_call", "id": f"fc_{number}", "call_id": f"call_{number}",
             "name": "exec_command", "arguments": json.dumps({"cmd": cmd, "max_output_tokens": 2000})}
@@ -360,6 +361,8 @@ def observed_response(request, number, message):
             if host == "code_mode":
                 config = home / "config.toml"
                 config.write_text(config.read_text().replace("[features]\n", "[features]\ncode_mode=true\n"))
+                if native_history_tokens is not None:
+                    config.write_text(f"tool_output_token_limit={native_history_tokens}\n" + config.read_text())
             if native_configuration:
                 config = home / "config.toml"
                 config.write_text("developer_instructions=" + json.dumps(native_configuration["developer_instructions"])
@@ -480,6 +483,21 @@ def held(*args, **kwargs):
     assert len(captured) == 2
     assert wire_paths["warm"].stat().st_size < wire_paths["cold"].stat().st_size
     assert len(calls) == len(observed) and stages["baseline_confirmed"]
+    if native_read_batch > 2:
+        visible_batches = []
+        for request in observed:
+            for item in request["input"]:
+                if item.get("type") != "custom_tool_call_output" or not isinstance(item.get("output"), list):
+                    continue
+                pages_in_output = 0
+                for part in item["output"]:
+                    try:
+                        page = json.loads(part.get("text", ""))
+                    except ValueError:
+                        continue
+                    pages_in_output += isinstance(page, dict) and page.get("schema_version") == "context_capture_page_v1"
+                visible_batches.append(pages_in_output)
+        assert max(visible_batches, default=0) > 2  # complete bytes were independently checked above
     if host == "benchmark":
         trace_roots = list((home / "context-traces").glob("wake-*"))
         assert len(trace_roots) == (2 if resumed else 1)
@@ -524,6 +542,22 @@ def test_native_missing_hook_recovers_from_persistent_cli_route(tmp_path, monkey
         host, surface, False, True, True, hook_loss)
 
 
+@pytest.mark.parametrize("surface", ["quota_packet", "turn_envelope"])
+def test_native_code_mode_pragma_retains_delta_and_recovers(tmp_path, monkeypatch, surface):
+    test_native_cli_capture_delta_and_compact_recovery(tmp_path, monkeypatch,
+        "code_mode", surface, False, True, True, native_pragma=True)
+
+
+@pytest.mark.parametrize("surface", ["quota_packet", "turn_envelope"])
+def test_native_known_history_budget_keeps_larger_batches_and_recovery(tmp_path, monkeypatch, surface):
+    # The unknown loopback model otherwise inherits a byte-based fallback.
+    # Match a declared 10k-token history budget only in this separate case;
+    # existing low-budget/truncation coverage and host defaults stay intact.
+    test_native_cli_capture_delta_and_compact_recovery(tmp_path, monkeypatch,
+        "code_mode", surface, False, True, True, "missing", shared_root=True,
+        native_pragma=True, native_read_batch=8, native_history_tokens=10000)
+
+
 @pytest.mark.parametrize("surface", ["quota_packet", "turn_envelope"])
 def test_native_shared_configuration_resume_and_lost_hook(tmp_path, monkeypatch, surface):
     test_native_cli_capture_delta_and_compact_recovery(tmp_path, monkeypatch,

tests/control_plane/test_codex_input_retention.py

@@ -188,13 +188,16 @@ def test_unknown_stale_lossy_or_unbound_observation_never_reuses(tmp_path, failu
     assert not codex_context_in_active_input(observation, baseline=baseline, session_id="session")
 
 
+@pytest.mark.parametrize("pragma", ["", '// @exec: {"max_output_tokens":12000}\n',
+    ' \t// @exec: {"yield_time_ms":1000}\r\n'])
 @pytest.mark.parametrize("failure", [None, "parent_missing", "parent_changed", "parent_ended", "parent_wrong_call",
-    "child_wrong_cell", "child_wrong_tool", "output_missing", "output_preview", "output_wrong_role"])
-def test_code_mode_requires_exact_live_parent_and_complete_model_visible_output(tmp_path, failure):
+    "child_wrong_cell", "child_wrong_tool", "output_missing", "output_preview", "output_wrong_role",
+    "malformed_pragma", "nonobject_pragma", "nonfinite_pragma", "other_comment", "second_pragma"])
+def test_code_mode_requires_exact_live_parent_and_complete_model_visible_output(tmp_path, failure, pragma):
     observation, baseline, request, rows = fixture(tmp_path)
     bundle = Path(observation["trace_bundle"])
     source = 'text(await tools.exec_command({cmd:"loopx quota should-run"}));'
-    output = {"output_items": [{"type": "custom_tool_call", "name": "exec", "call_id": "parent", "input": source}]}
+    output = {"output_items": [{"type": "custom_tool_call", "name": "exec", "call_id": "parent", "input": pragma + source}]}
     parent = {**rows[-1], "seq": 2, "payload": {"type": "code_cell_started", "runtime_cell_id": "cell",
         "model_visible_call_id": "parent", "source_js": source}}
     rows[-1]["seq"] = 3
@@ -225,6 +228,11 @@ def test_code_mode_requires_exact_live_parent_and_complete_model_visible_output(
         for item in request["input"]:
             item["type"] = "message"
             item["role"] = "assistant"
+    elif failure in ("malformed_pragma", "nonobject_pragma", "nonfinite_pragma", "other_comment", "second_pragma"):
+        prefix = {"malformed_pragma": "// @exec: {broken}\n", "nonobject_pragma": "// @exec: []\n",
+            "nonfinite_pragma": '// @exec: {"max_output_tokens":NaN}\n', "other_comment": "// other\n",
+            "second_pragma": "// @exec: {}\n// @exec: {}\n"}[failure]
+        output["output_items"][0]["input"] = prefix + source
     (bundle / "payloads/output.json").write_text(json.dumps(output))
     (bundle / "payloads/input.json").write_text(json.dumps(request))
     (bundle / "trace.jsonl").write_text("".join(json.dumps(row) + "\n" for row in rows))