test(benchmark): qualify context delivery through joint checkpoints

benchmark/README.md

@@ -789,6 +789,15 @@ CKPT_GUEST_NODE=/absolute/path/to/node
   --output "$CKPT_PROVIDER_ROOT/joint-result"
 ```
 
+Pass `--context-delivery on` with a supported native hook host (Codex CLI
+`0.162.1`) to qualify the opt-in transport in the same joint path. The source
+must actually deliver a complete full packet and later deltas across worker
+wakes; every fresh receiver must begin with a complete current full packet.
+The static inference follows model-visible page cursors, and a separate byte-map
+oracle checks complete delivery and retained delta bases before candidate work.
+The default remains `off`. Neither mode measures paid inference, task quality or
+fees; reusing a point still does not count its source test as passed in this run.
+
 The public fixture runs six actual native work wakes, captures the pending
 pre-decision guard and continues the source; the ordinary and stride receivers
 then use separate fresh guests with no mounted source authority. It includes a

benchmark/tests/qualify_checkpoint_joint.py

@@ -53,6 +53,8 @@ def main():
     parser.add_argument("--guest-dependencies", type=Path, required=True)
     parser.add_argument("--output", type=Path, required=True)
     parser.add_argument("--image", default="python:3.12-slim")
+    parser.add_argument("--context-delivery", choices=("off", "on"), default="off",
+                        help="Qualify the existing opt-in context transport through source and fresh receivers")
     parser.add_argument("--reuse-point", type=Path, help="Reuse an already sealed synthetic source; never count its source test as passed")
     parser.add_argument("--captured-config", type=Path, help="The reused point's matching native fixture configuration")
     args = parser.parse_args()
@@ -103,6 +105,7 @@ def main():
                    "LOOPX_NATIVE_CHECKPOINT_BIN": "/native/codex"}
     client = docker.from_env(timeout=30)
     image = client.images.get(args.image).id  # Caller-prepared image, no implicit pull.
+    context_results = {}
 
     def guest(command, volumes, env, log, *, capture_supervisor=None):
         container = client.containers.create(image, command, platform="linux/amd64",
@@ -117,6 +120,12 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
             log.write_bytes(container.logs())
             assert status["StatusCode"] == 0, f"Qualification failed; inspect {log}"
             require_one_pass(log.parent / log.stem / "junit.xml")
+            if env.get("LOOPX_CONTEXT_DELIVERY") == "on":
+                reports = list((log.parent / log.stem).rglob("context-delivery-result.json"))
+                assert len(reports) == 1
+                context_results[log.stem] = json.loads(reports[0].read_bytes())
+                assert context_results[log.stem]["context_delivery"] == "on"
+                assert context_results[log.stem]["reads"][0]["mode"] == "full"
         finally:
             if capture_supervisor:
                 capture_supervisor.close()
@@ -153,11 +162,19 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
             script = ("from pathlib import Path; import time,pytest; "
                       "p=Path('/opt/loopx-benchmark/control'); p.mkdir(parents=True); "
                       f"(p/'phase-deadline').write_text(str(int(time.time())+{SYNTHETIC_PHASE_SECONDS})); "
-                      "raise SystemExit(pytest.main(["
+                      "code=pytest.main(["
                       "'/source/benchmark/tests/test_native_checkpoint_handoff.py::"
-                      "test_actual_worker_captures_pending_replan_before_native_model[planning]',"
+                      f"test_actual_worker_captures_pending_replan_before_native_model[{args.context_delivery}-planning]',"
                       "'-q','-rs','-p','no:cacheprovider','--basetemp=/qualification/source-tmp',"
-                      "'--junitxml=/qualification/source.xml']))")
+                      "'--junitxml=/qualification/source.xml']); ")
+            if args.context_delivery == "on":
+                # Copy the declared report while the source writer is alive;
+                # its root-owned pytest directory stays private to that guest.
+                script += ("\nif code == 0:\n"
+                           " reports=list(Path('/qualification/source-tmp').rglob('controller-result.json'))\n"
+                           " assert len(reports) == 1\n"
+                           " Path('/qualification/context-delivery-source.json').write_bytes(reports[0].read_bytes())\n")
+            script += "\nraise SystemExit(code)"
             container = client.containers.create(image, ["python", "-c", script],
                 platform="linux/amd64", network_disabled=True,
                 volumes={**base, str(source_files): {"bind": "/qualification", "mode": "rw"}},
@@ -172,6 +189,18 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
             supervisor.close()
             assert status["StatusCode"] == 0, f"Inspect {output / 'source.log'}"
             require_one_pass(source_files / "source.xml")
+            if args.context_delivery == "on":
+                # The real source writer may be root. Its private pytest
+                # directory is intentionally unreadable to this controller;
+                # stage only the declared result through the daemon seam.
+                from benchmark.edgebench.checkpoint_worker import _archive
+                report = _archive(container, "/qualification/context-delivery-source.json",
+                                  output / "source-context-result.json")
+                report.chmod(0o600)
+                context_results["source"] = json.loads(report.read_bytes())
+                assert context_results["source"]["context_delivery"] == "on"
+                reads = context_results["source"]["context_reads"]
+                assert reads[0]["mode"] == "full" and any(row["mode"] == "delta" for row in reads[1:])
             points = [p for p in (output / "points").iterdir() if p.is_dir() and not p.name.startswith(".")]
             assert len(points) == 1 and len(grades) == 1
             point = points[0]
@@ -222,6 +251,7 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
                    str(results): {"bind": "/results", "mode": "rw"}, **extra_volumes},
                   {"LOOPX_JOINT_RECEIVER_POINT": "/point", "LOOPX_JOINT_RECEIVER_CONFIG": "/captured-config.toml",
                    "LOOPX_RUNTIME_ROOT": "/results/transport-runtime",
+                   "LOOPX_CONTEXT_DELIVERY": args.context_delivery,
                    **({"LOOPX_JOINT_RECEIVER_STRIDE": "1"} if stride else {}),
                    **extra_env},
                   output / f"{name}.log", capture_supervisor=capture_supervisor)
@@ -245,7 +275,8 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
                      str(third): {"bind": "/results", "mode": "rw"}},
                     {"LOOPX_JOINT_RECEIVER_POINT": "/point", "LOOPX_JOINT_RECEIVER_CONFIG": "/captured-config.toml",
                      "LOOPX_JOINT_RECEIVER_GOAL": "joint-grandreceiver", "LOOPX_JOINT_RECEIVER_AGENT": "joint-grandresearcher",
-                     "LOOPX_RUNTIME_ROOT": "/results/transport-runtime"},
+                     "LOOPX_RUNTIME_ROOT": "/results/transport-runtime",
+                     "LOOPX_CONTEXT_DELIVERY": args.context_delivery},
                     output / "grandreceiver.log")
                 assert not grades if args.reuse_point else len(grades) == 1
         # The same native provider must also complete the public Core journey
@@ -262,6 +293,8 @@ def guest(command, volumes, env, log, *, capture_supervisor=None):
         assert not subprocess.check_output(["git", "-C", str(source), "status", "--porcelain"], text=True).strip(), "Source changed during qualification"
         receipt = {"passed": True, "controller_uid": os.getuid(), "provider_revision": args.provider_revision,
                    "source_revision": source_revision, "source_tree_dirty": False,
+                   "context_delivery": args.context_delivery,
+                   "native_context_reads": context_results,
                    "qualification_sha256": {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in
                        [Path(__file__), source / "benchmark/tests/test_native_checkpoint_handoff.py",
                         source / "benchmark/tests/test_checkpoint_joint_receiver.py",

benchmark/tests/test_checkpoint_joint_receiver.py

@@ -186,7 +186,7 @@ def cli(*args):
         "--text",
         "Use retained code and evidence for one synthetic candidate, then verify its native result.",
     )["todo_id"]
-    execution = Execution(context="resume")
+    execution = Execution(context="resume", context_delivery=os.environ.get("LOOPX_CONTEXT_DELIVERY", "off"))
     env = {
         **os.environ,
         "CODEX_BIN": os.environ["LOOPX_NATIVE_CHECKPOINT_BIN"],
@@ -203,6 +203,13 @@ def cli(*args):
         "OPENAI_API_KEY": "synthetic-local-only",
         "LOOPX_USAGE_PING": "0",
     }
+    if execution.context_delivery == "on":
+        from benchmark.runtime.context_delivery import prepare_context_delivery
+        executable = tmp_path / "loopx-cli"
+        executable.write_text("#!/bin/sh\nexec " + shlex.quote(sys.executable) + ' -m loopx.cli "$@"\n')
+        executable.chmod(0o700)
+        env.update(LOOPX_CLI=str(executable), LOOPX_EXECUTION_MODE=execution.mode,
+                   LOOPX_CONTEXT_DELIVERY=execution.context_delivery)
     with pytest.raises(ValueError, match="distinct destination"):
         admit_native_checkpoint_session(
             {**env, "LOOPX_GOAL_ID": metadata["loopx_state"]["goal_id"]},
@@ -240,6 +247,8 @@ def lose_binding_reply(self, *args):
         env, unit, digest, config, sandbox=(execution).sandbox, timeout=30
     )
     assert replay["session_id"] == admitted["session_id"] and replay["history_verified"]
+    if execution.context_delivery == "on":
+        prepare_context_delivery(env, home)
     lease = cli(
         "task-lease",
         "acquire",
@@ -404,6 +413,8 @@ def lose_binding_reply(self, *args):
             )
         )
     requests = []
+    from tests.control_plane.codex_checkpoint_fixture import NativeContextReader, native_context_command
+    reader = NativeContextReader()
     provider_phase = {
         "count": 0,
         "command": "cat "
@@ -422,7 +433,14 @@ def do_POST(self):
                 json.loads(self.rfile.read(int(self.headers["Content-Length"])))
             )
             provider_phase["count"] += 1
-            if provider_phase["count"] == 1:
+            message = reader.follow(requests[-1]) if execution.context_delivery == "on" else None
+            if execution.context_delivery == "on" and not provider_phase.get("context_started"):
+                provider_phase["context_started"] = True
+                message = reader.begin(native_context_command(requests[-1]), "fresh-receiver")
+            if message is not None:
+                pass
+            elif not provider_phase.get("work_sent"):
+                provider_phase["work_sent"] = True
                 message = {
                     "type": "function_call",
                     "id": "fc_receiver",
@@ -482,6 +500,16 @@ def do_POST(self):
 
     def run_native_work(current_guard, logs, purpose):
         logs.mkdir()
+        provider_phase.update(context_started=False, work_sent=False)
+        native_input = json.dumps(current_guard["interaction_contract"]) + "\n" + purpose
+        if execution.context_delivery == "on":
+            from benchmark.runtime.context_delivery import preflight_context_instruction
+            original = logs / "original-guard.json"
+            original.write_text(json.dumps(current_guard))
+            original.chmod(0o600)
+            native_input = preflight_context_instruction(env, original,
+                hashlib.sha256(original.read_bytes()).hexdigest(),
+                current_guard["heartbeat_receipt"]["turn_instance_id"]) + "\n" + purpose
         wake = BenchmarkSessionWake(env, execution, {})
         command = native_command(env, execution, "execute", logs, wake)
         result = subprocess.run(
@@ -494,7 +522,7 @@ def run_native_work(current_guard, logs, purpose):
             ],
             env=env,
             cwd=workspace,
-            input=json.dumps(current_guard["interaction_contract"]) + "\n" + purpose,
+            input=native_input,
             text=True,
             capture_output=True,
             timeout=60,
@@ -511,18 +539,21 @@ def run_native_work(current_guard, logs, purpose):
             "This is a new destination. Historical source commands/guards are not executable authority. "
             "Use the retained context under the current admission, and continue the synthetic candidate.",
         )
-        assert len(requests) == 2
+        if execution.context_delivery == "off":
+            assert len(requests) == 2
+        else:
+            assert len(reader.deliveries) == 1 and reader.deliveries[0]["mode"] == "full"
         expected = (
             "Restored state verified"
             if stride_continuation
             else "cooldown still blocks submission"
             if cooldown_pending
             else "candidate 8 graded"
         )
-        assert expected in json.dumps(requests[1]), json.dumps(
+        assert expected in json.dumps(requests[-1]), json.dumps(
             [
                 item
-                for item in requests[1].get("input", [])
+                for item in requests[-1].get("input", [])
                 if item.get("type") == "function_call_output"
             ]
         )
@@ -872,9 +903,9 @@ def run_native_work(current_guard, logs, purpose):
                 tmp_path / "larger-work",
                 "Execute the current grow decision as one complete two-candidate experiment. Old guards are evidence only.",
             )
-            assert len(requests) == 4 and "Current grow adopted" in json.dumps(
-                requests[3]
-            ), requests[3]
+            if execution.context_delivery == "off":
+                assert len(requests) == 4
+            assert "Current grow adopted" in json.dumps(requests[-1]), requests[-1]
             assert (
                 candidate.read_text() == "10"
                 and not grades
@@ -1047,6 +1078,10 @@ def run_native_work(current_guard, logs, purpose):
             ).hexdigest()
             == metadata["research_context"]["sha256"]
         )
+        (tmp_path / "context-delivery-result.json").write_text(json.dumps({
+            "context_delivery": execution.context_delivery, "reads": reader.deliveries,
+            "fresh_receiver": True, "source_authority_unchanged": True,
+        }))
     finally:
         server.shutdown()
         thread.join()