diff --git a/benchmark/README.md b/benchmark/README.md index 855bac62d2..eb8f69ad71 100644 --- a/benchmark/README.md +++ b/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 diff --git a/benchmark/tests/qualify_checkpoint_joint.py b/benchmark/tests/qualify_checkpoint_joint.py index 88bb21d635..f51c34cc75 100644 --- a/benchmark/tests/qualify_checkpoint_joint.py +++ b/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", diff --git a/benchmark/tests/test_checkpoint_joint_receiver.py b/benchmark/tests/test_checkpoint_joint_receiver.py index 219d862018..8c7295d40e 100644 --- a/benchmark/tests/test_checkpoint_joint_receiver.py +++ b/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,7 +539,10 @@ 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 @@ -519,10 +550,10 @@ def run_native_work(current_guard, logs, purpose): 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() diff --git a/benchmark/tests/test_native_checkpoint_handoff.py b/benchmark/tests/test_native_checkpoint_handoff.py index d3c7eac0f8..53a0e303ed 100644 --- a/benchmark/tests/test_native_checkpoint_handoff.py +++ b/benchmark/tests/test_native_checkpoint_handoff.py @@ -370,7 +370,8 @@ def test_midwake_native_profile_refusal_precedes_interruption(tmp_path, monkeypa os.environ.get("LOOPX_CHECKPOINT_EXTERNAL_CONTROLLER") != "1", reason="Requires the real native CLI and explicitly attached checkpoint supervisor") @pytest.mark.parametrize("explore_mode", ["off", "planning"]) -def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, explore_mode): +@pytest.mark.parametrize("context_delivery", ["off", "on"]) +def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, explore_mode, context_delivery): """Six real settled work wakes -> pre-decision capture -> stride -> work. Inference is deterministic; native history/tools, canonical SQLite, quota, @@ -383,6 +384,7 @@ def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, exp from benchmark.runtime.worker import run_once from tests.control_plane import test_quota_settlement_cli as cli from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture + from tests.control_plane.codex_checkpoint_fixture import NativeContextReader, native_context_command project, runtime, registry = cli._write_fixture(tmp_path / "authority") document = json.loads(registry.read_text()) @@ -403,6 +405,7 @@ def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, exp skills.mkdir() requests, guards, work_turns = [], [], [] phase = {"kind": "initial", "values": [], "tool_sent": False} + reader = NativeContextReader() def run(*args): code, result = cli._run_cli(registry, runtime, *args, cwd=project) @@ -422,6 +425,22 @@ def run(*args): def response(payload, index, message): requests.append(payload) + if context_delivery == "on" and phase["kind"] != "initial": + follow = reader.follow(payload) + if follow: + return follow + if not phase.get("context_started"): + phase["context_started"] = True + return reader.begin(native_context_command(payload, + todo_id=cli.TODO_ID if phase["kind"] == "work" else None), phase["kind"]) + if phase["kind"] == "work" and not phase["tool_sent"]: + # Only after complete native input delivery, inspect the + # original decision as an independent authority oracle. + source = Path(reader.deliveries[-1]["recovery"]["path"]).with_name("decision.json") + current = json.loads(source.read_text()) + assert not current.get("autonomous_replan_obligation") + assert current["interaction_contract"]["agent_channel"]["delivery_allowed"] + (project / "work-guard.json").write_text(json.dumps(current)) if phase["kind"] == "work" and not phase["tool_sent"]: phase["tool_sent"] = True prefix = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry), @@ -440,6 +459,14 @@ def response(payload, index, message): "[(Path('candidate.txt').write_text(str(v))," "Path('evidence-'+str(v)+'.json').write_text(json.dumps({'value':v,'synthetic':True}))) for v in values]; " "assert Path('candidate.txt').read_text()==str(values[-1]); print('CANDIDATE_VERIFIED')") + if context_delivery == "on": + code = ("import json; from pathlib import Path; " + "g=json.loads(Path('work-guard.json').read_text()); " + "assert g['interaction_contract']['agent_channel']['delivery_allowed']; " + f"values={phase['values']!r}; " + "[(Path('candidate.txt').write_text(str(v))," + "Path('evidence-'+str(v)+'.json').write_text(json.dumps({'value':v,'synthetic':True}))) for v in values]; " + "assert Path('candidate.txt').read_text()==str(values[-1]); print('CANDIDATE_VERIFIED')") return {"type": "function_call", "id": f"fc_work_{index}", "call_id": f"call_work_{index}", "name": "exec_command", "arguments": json.dumps({ "cmd": shlex.quote(sys.executable) + " -c " + shlex.quote(code), @@ -465,6 +492,9 @@ def response(payload, index, message): assert not guard["interaction_contract"]["agent_channel"]["delivery_allowed"] # The sealed point precedes the policy decision and seventh work. assert candidate.read_text() == "6" + if context_delivery == "on": + message["content"][0]["text"] = "Original packet read completely through native context tools." + return message return {"type": "function_call", "id": "fc_packet", "call_id": "call_packet", "name": "exec_command", "arguments": json.dumps({ "cmd": "cat " + shlex.quote(str(paths[0])), "max_output_tokens": 30000})} @@ -500,11 +530,18 @@ def settle_work(result, expected): LOOPX_ITERATION_CONTEXT="resume", LOOPX_SHARED_SKILLS=str(skills), MODEL_NAME="fixture", REASONING_EFFORT="high", OPENAI_BASE_URL=base["model_providers"]["fixture"]["base_url"], OPENAI_API_KEY="synthetic-local-only", LOOPX_CODEX_TURN_TIMEOUT_SEC="60") + env["LOOPX_CONTEXT_DELIVERY"] = context_delivery for value in range(1, 7): - phase.update(kind="work", values=[value], tool_sent=False) + phase.update(kind="work", values=[value], tool_sent=False, context_started=False) prior = run_once(env) settle_work(prior, value) - assert len(set(work_turns)) == 6 and len(calls) == 13 + assert len(set(work_turns)) == 6 + if context_delivery == "off": + assert len(calls) == 13 + else: + assert reader.deliveries[0]["mode"] == "full" + assert any(row["mode"] == "delta" for row in reader.deliveries[1:]) + review_request_index = len(requests) if explore_mode == "planning": from loopx.capabilities.explore.result_log import ( append_explore_result_events, build_explore_node_event, build_explore_finding_event, @@ -525,19 +562,21 @@ def settle_work(result, expected): "point_metadata": {}, "trigger": "fresh_replan", "native_session": True, "loopx_state": True, "evaluation_opportunity": True, "controller_handoff": True})) - phase.update(kind="review", tool_sent=False) + phase.update(kind="review", tool_sent=False, context_started=False) result = run_once({**env, "LOOPX_CHECKPOINT_REQUEST": str(request)}) assert result["ok"] and result["checkpoint"]["captured"], result assert "turn_envelope" not in prior and "turn_envelope_source" not in prior assert "turn_envelope" not in result and "turn_envelope_source" not in result - assert 'Use TurnEnvelope:' not in json.dumps(requests[13]) + assert 'Use TurnEnvelope:' not in json.dumps(requests[review_request_index]) assert 'interaction_contract' in guards[0] and 'action_signature' not in guards[0] assert result["checkpoint"]["phase"] == "before_replan" assert result["checkpoint"]["source_turn_id"] == prior["turn_id"] assert result["session"]["session_id"] == prior["session"]["session_id"] - assert not result.get("checkpoint_handoffs") and len(calls) == 15 + assert not result.get("checkpoint_handoffs") + if context_delivery == "off": + assert len(calls) == 15 assert cli._heartbeat_receipt_count(runtime, result["turn_id"]) == 1 - assert "settlement_binding_owed" in json.dumps(requests[14]) and len(guards) == 1 + assert "settlement_binding_owed" in json.dumps(requests[-1]) and len(guards) == 1 # The original source now admits its review; capture has granted nothing. selected = run("quota", "should-run", "--goal-id", cli.GOAL_ID, "--agent-id", cli.AGENT_ID, @@ -580,14 +619,17 @@ def settle_work(result, expected): "--agent-vision-json", str(vision), "--no-global-sync", "--suppress-external-sinks") run("quota", "spend-slot", *binding, "--source", "heartbeat", "--execute") - phase.update(kind="work", values=[7, 8], tool_sent=False) + phase.update(kind="work", values=[7, 8], tool_sent=False, context_started=False) adopted = run_once(env) settle_work(adopted, 8) - assert len(calls) == 17 and len(work_turns) == 7 + assert len(work_turns) == 7 + if context_delivery == "off": + assert len(calls) == 17 assert all(json.loads((project / f"evidence-{v}.json").read_text())["value"] == v for v in (7, 8)) (tmp_path / "controller-result.json").write_text(json.dumps({**result["checkpoint"], "natural_work_turns": work_turns[:6], "stride_action": "grow", "explore_mode": explore_mode, - "source_followthrough_verified": True, "adopted_turn_id": adopted["turn_id"]})) + "source_followthrough_verified": True, "adopted_turn_id": adopted["turn_id"], + "context_delivery": context_delivery, "context_reads": reader.deliveries})) @pytest.mark.skipif(not os.environ.get("LOOPX_NATIVE_CHECKPOINT_BIN") or os.name != "posix", diff --git a/benchmark/tests/test_shared_codex_runtime.py b/benchmark/tests/test_shared_codex_runtime.py index 73274ad7c3..23c6ce659b 100644 --- a/benchmark/tests/test_shared_codex_runtime.py +++ b/benchmark/tests/test_shared_codex_runtime.py @@ -1125,6 +1125,29 @@ def test_native_context_opt_in_is_only_a_heartbeat_treatment(mode): Execution(context_delivery="1") +@pytest.mark.parametrize("declared_mode", [None, "heartbeat"]) +def test_context_preflight_accepts_the_resolved_heartbeat_mode(tmp_path, monkeypatch, declared_mode): + from benchmark.tests.test_resume_sessions import resume_env + from loopx.control_plane.turn_driver.codex_context import context_delivery_activation + + env = resume_env(tmp_path, monkeypatch) + if declared_mode is None: + env.pop("LOOPX_EXECUTION_MODE") + env["LOOPX_CONTEXT_DELIVERY"] = "on" + monkeypatch.setattr("benchmark.runtime.context_delivery.prepare_context_delivery", lambda *args: None) + + def preflight(current, *_args, **_kwargs): + activation = context_delivery_activation("on", runtime_profile="generic_cli", + benchmark_mode=current.get("LOOPX_EXECUTION_MODE")) + assert activation["scenario"] == "codex_benchmark_heartbeat" + return "Current Todo input." + + monkeypatch.setattr("benchmark.runtime.worker.heartbeat_body", preflight) + result = run_once(env) + assert result["ok"] and result["session"]["session_id"] + assert env.get("LOOPX_EXECUTION_MODE") == declared_mode + + def test_context_transport_unavailable_preserves_ordinary_worker(tmp_path, monkeypatch): from benchmark.tests.test_resume_sessions import resume_env env = resume_env(tmp_path, monkeypatch) diff --git a/tests/control_plane/codex_checkpoint_fixture.py b/tests/control_plane/codex_checkpoint_fixture.py index e967e463d8..daeade632b 100644 --- a/tests/control_plane/codex_checkpoint_fixture.py +++ b/tests/control_plane/codex_checkpoint_fixture.py @@ -1,12 +1,142 @@ """Deterministic local inference for real Codex checkpoint owner tests.""" import json import os +from pathlib import Path +import shlex import threading import time import uuid from contextlib import contextmanager from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +def delivered_context(request, digest): + """Independent byte-map oracle over actual native model input.""" + import hashlib + + delivered = {} + for item in request["input"]: + output = item.get("output") + texts = ([part["text"] for part in output if part.get("type") == "input_text"] + if item.get("type") == "custom_tool_call_output" and isinstance(output, list) + else [output]) + for output in texts: + if not isinstance(output, str): + continue + try: + page = json.loads(output.split("\nOutput:\n", 1)[-1]) + if isinstance(page, dict) and page.get("exit_code") == 0 and isinstance(page.get("output"), str): + page = json.loads(page["output"]) + except (ValueError, TypeError): + continue + if not isinstance(page, dict) or page.get("schema_version") != "context_capture_page_v1" or page.get("sha256") != digest: + continue + assert page["ok"] + data = page["content"].encode() + assert len(data) == (page["total_bytes"] if page["eof"] else page["next_offset"]) - page["offset"] + for offset, value in enumerate(data, page["offset"]): + assert offset not in delivered or delivered[offset] == value + delivered[offset] = value + if not delivered: + return "" + assert set(delivered) == set(range(len(delivered))) + raw = bytes(delivered[i] for i in range(len(delivered))) + assert hashlib.sha256(raw).hexdigest() == digest + return raw.decode().rstrip("\n") + + +def last_context_page(request): + for item in reversed(request["input"]): + output = item.get("output") + parts = [p.get("text", "") for p in output] if isinstance(output, list) else [output] + for part in reversed(parts): + if not isinstance(part, str): + continue + try: + value = json.loads(part.split("\nOutput:\n", 1)[-1]) + if isinstance(value, dict) and value.get("exit_code") == 0: + value = json.loads(value["output"]) + if isinstance(value, dict) and value.get("schema_version") == "context_capture_page_v1": + return value + except (ValueError, KeyError): + continue + return None + + +class NativeContextReader: + """Static inference which follows only model-visible native reader cursors. + + Expected files are inspected after complete input coverage, never to seed a + baseline or choose a page. Native hooks remain the production continuity owner. + """ + + def __init__(self): + self.pending = None + self.deliveries = [] + self.retained = {} + self.calls = 0 + self.identity = uuid.uuid4().hex + + def tool(self, command): + self.calls += 1 + return {"type": "function_call", "id": f"fc_context_{self.identity}_{self.calls}", + "call_id": f"call_context_{self.identity}_{self.calls}", "name": "exec_command", + "arguments": json.dumps({"cmd": command, "yield_time_ms": 10000, + "max_output_tokens": 30000})} + + def begin(self, command, label): + assert self.pending is None + self.pending = {"label": label, "descriptor": None, "page_calls": 0} + return self.tool(command) + + def follow(self, request): + if self.pending is None: + return None + page = last_context_page(request) + assert page and page["ok"], "Native reader output must reach the model intact" + 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 + first = self.pending["descriptor"] + assert page["sha256"] == first["sha256"] + if not page["eof"]: + return self.tool(page["next_command"]) + raw = delivered_context(request, first["sha256"]) + assert len(raw.encode()) in {page["total_bytes"], page["total_bytes"] - 1} + # The original reader owns digest verification. This separate oracle + # checks the actual complete model input against its declared source. + assert raw == Path(first["path"]).read_text().rstrip("\n") + packet = json.loads(raw) + if first["mode"] == "delta": + base = self.retained[packet["base"]["capture_id"]] + assert delivered_context(request, base["sha256"]) == base["raw"] + self.retained[first["target"]["capture_id"]] = {"raw": raw, "sha256": first["sha256"]} + self.deliveries.append({"label": self.pending["label"], "mode": first["mode"], + "body_bytes": page["total_bytes"], "page_calls": self.pending["page_calls"], + "sha256": first["sha256"], "recovery": first["recovery"], + "target": first["target"]}) + self.pending = None + return None + + +def native_context_command(request, *, todo_id=None): + """Use the actual worker prompt, including its original wake/transport pins.""" + latest = [item for item in request["input"] if item.get("role") == "user"][-1] + texts = [part.get("text", "") for part in latest.get("content", []) if isinstance(part, dict)] + text = "\n".join(texts) + projects = [line for line in text.splitlines() if " quota context-project " in line] + if projects: + return projects[-1] + command = [line for line in text.splitlines() + if " quota should-run " in line and "--codex-context-dir" in line][-1] + assignment = [line for line in text.splitlines() if line.startswith("LOOPX_TURN=")][-1] + turn = shlex.split(assignment)[0].split("=", 1)[1] + argv = [turn if arg == "${LOOPX_TURN:?}" else arg for arg in shlex.split(command)] + if todo_id: + argv += ["--todo-id", todo_id] + return shlex.join(argv) + @contextmanager def local_inference_provider(response_builder=None, usage_builder=None): """Static loopback Responses stream, retaining the actual native model input.""" diff --git a/tests/control_plane/test_codex_context_capture.py b/tests/control_plane/test_codex_context_capture.py index f6a28de082..eb84be30dd 100644 --- a/tests/control_plane/test_codex_context_capture.py +++ b/tests/control_plane/test_codex_context_capture.py @@ -19,7 +19,9 @@ from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection from loopx.control_plane.quota.turn_envelope import prepare_context_delivery, restore_context_delivery from test_quota_settlement_cli import AGENT_ID, GOAL_ID, TODO_ID, _spend_run_count, _write_fixture -from tests.control_plane.codex_checkpoint_fixture import local_inference_provider +from tests.control_plane.codex_checkpoint_fixture import ( + local_inference_provider, delivered_context as _delivered, last_context_page as _last_page, +) from tests.control_plane.test_heartbeat_envelope_transport import private_handoff # noqa: F401 -- shared private socket fixture from tests.control_plane.test_codex_context_boundary import _app_server, _configure_provider, _input_text, _native_fixture @@ -27,58 +29,6 @@ reason="Explicit native Codex binary required; inference uses loopback only") -def _delivered(request, digest): - # Independent byte-map oracle: recover every model-visible byte, including - # separate code-mode emissions and overlapping reads at smaller page sizes. - delivered = {} - for item in request["input"]: - output = item.get("output") - texts = ([part["text"] for part in output if part.get("type") == "input_text"] - if item.get("type") == "custom_tool_call_output" and isinstance(output, list) - else [output]) - for output in texts: - if not isinstance(output, str): - continue - try: - page = json.loads(output.split("\nOutput:\n", 1)[-1]) - if isinstance(page, dict) and page.get("exit_code") == 0 and isinstance(page.get("output"), str): - page = json.loads(page["output"]) - except (ValueError, TypeError): - continue - if not isinstance(page, dict) or page.get("schema_version") != "context_capture_page_v1" or page.get("sha256") != digest: - continue - assert page["ok"] - data = page["content"].encode() - assert len(data) == (page["total_bytes"] if page["eof"] else page["next_offset"]) - page["offset"] - for offset, value in enumerate(data, page["offset"]): - assert offset not in delivered or delivered[offset] == value - delivered[offset] = value - if not delivered: - return "" - assert set(delivered) == set(range(len(delivered))) - raw = bytes(delivered[i] for i in range(len(delivered))) - assert hashlib.sha256(raw).hexdigest() == digest - return raw.decode().rstrip("\n") - - -def _last_page(request): - for item in reversed(request["input"]): - output = item.get("output") - parts = [p.get("text", "") for p in output] if isinstance(output, list) else [output] - for part in reversed(parts): - if not isinstance(part, str): - continue - try: - value = json.loads(part.split("\nOutput:\n", 1)[-1]) - if isinstance(value, dict) and value.get("exit_code") == 0: - value = json.loads(value["output"]) - if isinstance(value, dict) and value.get("schema_version") == "context_capture_page_v1": - return value - except (ValueError, KeyError): - continue - return None - - @pytest.mark.parametrize("host,surface,failed_primary,resumed,adaptive", [ (host, surface, failed, False, False) for host, surface in [("exec", "quota_packet"), ("app_server", "turn_envelope"), ("code_mode", "quota_packet"), ("code_mode", "turn_envelope")]