test(context): follow file references in worker and checkpoint observers

tests/control_plane/codex_checkpoint_fixture.py

@@ -101,12 +101,25 @@ def begin(self, command, label):
     def follow(self, request):
         if self.pending is None:
             return None
-        page = last_context_page(request)
+        page = last_context_page(request, allow_reference=True)
         assert page and page["ok"], "Native reader output must reach the model intact"
+        if page.get("schema_version") != "context_capture_page_v1":
+            assert self.pending["descriptor"] is None and "reference" not in self.pending
+            # This fixture voluntarily reads the complete saved input for its
+            # byte oracle. A model-visible reference alone proves no coverage.
+            self.pending["reference"] = page
+            return self.tool(page["command"])
         self.pending["page_calls"] += 1
         if self.pending["descriptor"] is None:
-            assert page["offset"] == 0 and page["mode"] in {"full", "delta"}
-            self.pending["descriptor"] = page
+            assert page["offset"] == 0
+            if reference := self.pending.get("reference"):
+                assert page["sha256"] == reference["sha256"]
+                # The generic saved-file reader returns bytes and cursors;
+                # delivery identity stays in the original visible reference.
+                self.pending["descriptor"] = {**reference, **page}
+            else:
+                assert page["mode"] in {"full", "delta"}
+                self.pending["descriptor"] = page
         first = self.pending["descriptor"]
         assert page["sha256"] == first["sha256"]
         if not page["eof"]:

tests/control_plane/test_context_capture_selection.py

@@ -6,7 +6,7 @@
 
 import pytest
 
-from tests.control_plane.codex_checkpoint_fixture import last_context_page
+from tests.control_plane.codex_checkpoint_fixture import NativeContextReader, last_context_page
 from loopx.cli_commands.quota_capture import read_context_capture
 from loopx.control_plane.turn_driver.decision_capture import capture_page_range
 from test_quota_settlement_cli import _run_cli
@@ -23,9 +23,17 @@ def test_native_saved_reference_is_not_reader_coverage(wrapped):
     request = {"input": [item]}
     assert last_context_page(request) is None
     assert last_context_page(request, allow_reference=True) == reference
+    reader = NativeContextReader()
+    reader.begin("admit current Turn", "synthetic")
+    call = reader.follow(request)
+    assert json.loads(call["arguments"])["cmd"] == reference["command"]
+    assert not reader.deliveries and not reader.retained
+    assert reader.pending["page_calls"] == 0
     # A damaged current response cannot borrow the earlier complete reference.
     request["input"].append({"type": "function_call_output", "output": '{"ok":true'})
     assert last_context_page(request, allow_reference=True) is None
+    with pytest.raises(AssertionError, match="output must reach the model intact"):
+        reader.follow(request)
 
 
 def fixture_capture(tmp_path):
@@ -42,6 +50,36 @@ def args(path, raw):
         offset=0, page_bytes=256, max_page_output_bytes=2000, format="json", select=["/rows/0"]))
 
 
+def test_native_reference_reader_uses_real_pages_before_recording_delivery(tmp_path):
+    path, raw, _ = fixture_capture(tmp_path)
+    options = args(path, raw)
+    options.select = None
+    reference = {"ok": True, "mode": "full", "path": str(path),
+        "sha256": options.capture_sha256, "command": "read saved capture",
+        "target": {"capture_id": "synthetic-source"}, "recovery": {"path": str(path)}}
+    request = {"input": [{"type": "function_call_output", "output": json.dumps(reference)}]}
+    reader = NativeContextReader()
+    reader.begin("admit current Turn", "source")
+    reader.follow(request)
+    page_count = 0
+    while True:
+        page = read_context_capture(options)
+        assert page["ok"] and "mode" not in page and "target" not in page
+        request["input"].append({"type": "function_call_output", "output": json.dumps(page)})
+        call = reader.follow(request)
+        page_count += 1
+        if page["eof"]:
+            assert call is None
+            break
+        assert not reader.deliveries and not reader.retained
+        assert json.loads(call["arguments"])["cmd"] == page["next_command"]
+        options.offset = page["next_offset"]
+    assert page_count > 1
+    assert reader.deliveries[0]["page_calls"] == page_count
+    assert reader.deliveries[0]["mode"] == "full"
+    assert reader.retained["synthetic-source"]["raw"] == raw.decode().rstrip("\n")
+
+
 def test_actual_cli_batch_pages_preserve_values_without_live_state(tmp_path):
     path, raw, original = fixture_capture(tmp_path)
     digest = hashlib.sha256(raw).hexdigest()

tests/control_plane/test_heartbeat_envelope_transport.py

@@ -243,7 +243,7 @@ def test_real_worker_generated_guard_selection_reentry_and_settle_once(
                LOOPX_WAKE_LOG_DIR=str(logs), LOOPX_TURN_ENVELOPE='1' if enabled else '0')
     if context_delivery:
         # No native proof in this CLI fixture: composed capture must still
-        # release ordinary full output and preserve reentry and settlement.
+        # save a full baseline and preserve reentry and settlement.
         (tmp_path / 'context').mkdir(mode=0o700)
         env.update(LOOPX_CONTEXT_DELIVERY='on', LOOPX_EXECUTION_MODE='heartbeat', LOOPX_CODEX_CONTEXT_DIR=str(tmp_path / 'context'))
         # Generated CLI reentry uses this process's environment; preserve the
@@ -260,6 +260,16 @@ def test_real_worker_generated_guard_selection_reentry_and_settle_once(
     first = json.loads(result.stdout)
     assert result.returncode == 0, first
     def complete_page(value):
+        if context_delivery:
+            # This compatibility test elects to read the full saved view.
+            # A manifest only supplies navigation; it is not delivered body.
+            assert value['mode'] == 'full'
+            digest = value['sha256']
+            read = subprocess.run(shlex.split(value['command']), env=env, cwd=project,
+                capture_output=True, text=True, check=True)
+            value = json.loads(read.stdout)
+            assert value['schema_version'] == 'context_capture_page_v1'
+            assert value['sha256'] == digest
         if value.get('schema_version') != 'context_capture_page_v1':
             return value
         digest, body = value['sha256'], b''