feat(context): share whole-packet deltas and capture recovery

benchmark/edgebench/run.py

@@ -126,6 +126,8 @@ def main(argv=None):
                         help="New-run default: native public feedback; blind/best-only are explicit controls")
     parser.add_argument("--turn-envelope", action="store_true",
                         help="Opt-in short heartbeat context with same-invocation full captures")
+    parser.add_argument("--context-delivery", choices=("off", "on"), default="off",
+                        help="Experimental native Codex benchmark heartbeat reuse only; off by default")
     parser.add_argument("--native-goal-continue", action="store_true",
                         help="Opt-in: when a native Goal declares completion with budget left, start "
                              "continuation Goals until the budget edge (native-goal worker only)")
@@ -190,6 +192,8 @@ def main(argv=None):
         parser.error("--task-entry loopx-planned requires a heartbeat worker")
     if args.turn_envelope and args.worker not in {"heartbeat-resume", "heartbeat-explore"}:
         parser.error("--turn-envelope requires a heartbeat worker")
+    if args.context_delivery != "off" and not args.worker.startswith("heartbeat-"):
+        parser.error("--context-delivery requires a heartbeat worker")
     if args.native_goal_continue and args.worker != "native-goal":
         parser.error("--native-goal-continue requires --worker native-goal")
     if (args.replan_after_turns is not None or args.replan_after_todos is not None
@@ -320,6 +324,7 @@ def main(argv=None):
                          task_objective=task_objective,
                          task_entry=args.task_entry,
                          turn_envelope=args.turn_envelope,
+                         context_delivery=args.context_delivery,
                          native_goal_continue=args.native_goal_continue,
                          replan_after_turns=args.replan_after_turns,
                          replan_after_todos=args.replan_after_todos,
@@ -383,6 +388,7 @@ def main(argv=None):
         "loopx_commit": pins[0], "runner_commit": pins[1],
         "sforge_source": active_source_receipt(),
         **turn_envelope_receipt(explicit=args.turn_envelope),
+        **({"context_delivery": "on"} if args.context_delivery != "off" else {}),
         **({"native_goal_continue": True} if args.native_goal_continue else {}),
         "task_sha256": hashlib.sha256(task_file.read_bytes()).hexdigest(),
         "feedback": args.feedback, "internet": task.internet,

benchmark/runtime/RUNTIME.md

@@ -45,6 +45,86 @@ fail before execution. Set it back to `false` for the next run to roll back;
 do not change a running trial's treatment. Captures contain private Goal context.
 Transport tests do not establish lower token cost or better model decisions.
 
+For an explicitly selected native context-delivery treatment, add
+`context_delivery: "on"` to the same heartbeat agent `kwargs` (default `"off"`).
+Only the existing Codex heartbeat worker accepts `"on"`; booleans, other modes and
+scenario names are rejected. This is an experimental transport setting, separate
+from `turn_envelope`.
+The existing EdgeBench/SForge entry exposes the same selection through
+`--context-delivery on` on a heartbeat worker (`off` is the default).
+The worker preserves its existing `generic_cli` scheduler profile. Its execution
+owner requires `mode=heartbeat` and explicitly propagates `LOOPX_EXECUTION_MODE=heartbeat`
+and `LOOPX_CONTEXT_DELIVERY=on` into the native tool environment. Generic CLI
+without these worker declarations stays full. These are caller declarations,
+not host authentication or a new admission grant.
+It works with ordinary quota or the explicit TurnEnvelope view. The existing
+worker stages two product lifecycle hooks in the trial's private Codex home and
+queries native `hooks/list` without invoking a model. Only those exact native
+hashes receive per-process trust; workspace/plugin hooks are not trusted and no
+bypass flag or operator settings are used. Private native trace-v1 diagnostics
+are enabled only in this opted-in trial. Install and wake receipts record the
+selection; `context_transport.available` records setup, **not** warm eligibility
+or delivery/benefit. Unsupported native setup continues ordinary full output.
+
+The product heartbeat renderer carries `--codex-context-dir` and the original
+capture root into guard/reentry commands. The product observer proves intact
+baseline content in the current model input before warm reuse; uncertainty
+selects full. Startup/fork/compaction reset the epoch. Same-session resume clears
+the active call and keeps only an unconfirmed candidate; reuse requires the new
+call's complete actual input, never a saved session or file. Each worker wake
+uses a separate private trace directory, avoiding repeated scans of old wakes.
+The small recovery hook reads the original full capture after compaction, with
+no replacement admission or debit. Native worker and app-server code-mode tests
+cover resumed warm reuse for both packet views. The opted-in worker also carries
+a compact native developer instruction that survives compaction independently
+of hook output. If context was lost while continuing a captured decision,
+`quota context-read --codex-context-dir DIR --goal-id GOAL --agent-id AGENT`
+checks the current native session/workspace and selects the original capture.
+Follow its digest-pinned next commands; this read never reruns admission or
+spends quota. Missing/foreign references refuse lookup. Native tests omit or
+kill all subsequent hooks and still recover every byte through this route.
+Installed App adoption remains separate; deterministic transport tests are not autonomous
+solver outcomes. Include this persistent instruction in subsequent cost comparisons;
+earlier input replays did not include it. Count complete input, recovery reads, native discovery and trace
+IO in treatment cost, not just the delta bytes.
+
+Codex App setup uses the same product configuration renderer as the worker:
+`loopx --format json quota context-config --context-delivery on --codex-app --codex-context-root DIR --goal-id GOAL --agent-id AGENT`.
+It returns native hooks and persistent recovery instructions, without installing
+files or changing trust. A shared root isolates journals by native session;
+the trial-local single-session `--codex-context-dir` remains supported.
+See [heartbeat setup](../../docs/heartbeat-automation-prompt.md#opt-in-native-context-delivery)
+for activation, diagnostics, readback and rollback. Generated setup and a native
+app-server test do not by themselves prove installed desktop adoption.
+
+A simultaneous `fresh_replan` checkpoint listener retains ownership of the same
+original capture. It validates and holds the original full decision before any
+context projection, then releases its exact digest. Only afterward may the
+native observer project full/delta for an uninterrupted tool. A refused/lost
+release remains the original typed transport failure, without publishing an
+optimized result or rerunning the guard. An interrupted guard continues with
+the original full packet; no delta is required for checkpoint recovery. Native
+cursor verification also binds the context-directory/root option. The existing
+Darwin native-checkpoint refusal remains in force.
+
+The host-side checkpoint preflight still captures the complete original decision.
+With context delivery enabled, its prompt now supplies `quota context-project`
+with that original path/hash, Goal/Agent/Turn and private cache location. The
+native tool presents the saved decision using the same current-input observer;
+no guard, checkpoint, admission or debit is repeated. Cold delivery supplies the
+full body and its target; qualified warm delivery supplies the exact delta.
+The small response points to bounded reads and an original full-read fallback.
+No native setup, invalid cache or uncertain retention keeps full input. A bad
+source hash/scope is refused before publication. This does not renew decision
+freshness or establish autonomous solver benefit. Default-off preloading is
+unchanged. Include the projection command, descriptor, recovery and page reads
+in subsequent total-cost comparisons.
+Across distinct worker wakes, settle the prior Turn before selecting new work;
+reading an old prompt from resumed history cannot supply the next wake's guard.
+Do not modify an active/frozen trial. Set `context_delivery: "off"` for a new trial
+to roll back; default runs install no hooks, trust or trace settings. No scorer,
+budget, task semantics or launch authority changes with this option.
+
 Checkpoint capture is independent of this display experiment. A `fresh_replan`
 declaration may establish a private capture/handoff socket while
 `turn_envelope` remains false: the worker keeps the ordinary dispatcher and quota

benchmark/runtime/checkpoint.py

@@ -153,8 +153,14 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st
     preflight_seen = guard_path.exists()
     guard_path.write_text(guard_run.stdout, encoding="utf-8")
     guard_path.chmod(0o600)
-    preloaded_body = (f"Host checkpoint preflight has already executed this wake's original quota guard once. "
-        f"Read the complete packet at {guard_path} before work, then follow its interaction_contract. "
+    reading = f"Read the complete packet at {guard_path} before work, then follow its interaction_contract. "
+    if env.get("LOOPX_CONTEXT_DELIVERY") == "on" and env.get("LOOPX_CODEX_CONTEXT_DIR"):
+        from .context_delivery import preflight_context_instruction
+
+        reading = preflight_context_instruction(env, guard_path,
+            hashlib.sha256(guard_path.read_bytes()).hexdigest(), receipt["turn_id"])
+    preloaded_body = ("Host checkpoint preflight has already executed this wake's original quota guard once. "
+        + reading +
         "Do not repeat the initial guard below; use its selection or recovery command when required.\n\n" + body)
     receipt["checkpoint"] = {"captured": False, "research_point_qualified": False, "phase": "before_replan"}
     try:

benchmark/runtime/checkpoint_wake.py

@@ -136,6 +136,9 @@ def _capture(self, held, deadline):
                          "turn_envelope": self.receipt.get("turn_envelope", False),
                          "capture_root": str(self.capture_root), "socket_path": self.socket_path,
                          "decision_sha256": held["sha256"]}
+                if self.env.get("LOOPX_CODEX_CONTEXT_DIR"):
+                    guard["codex_context_dir"] = self.env["LOOPX_CODEX_CONTEXT_DIR"]
+                    guard["context_delivery"] = self.env.get("LOOPX_CONTEXT_DELIVERY", "off")
                 reader = readers.enter_context(native_checkpoint_reader(self.env, timeout=max(.1, window - time.time())))
                 native = native_session_checkpoint(self.env, self.receipt["session"]["session_id"],
                     timeout=max(.1, window - time.time()), interrupted_guard=guard, transport=reader,

benchmark/runtime/codex.py

@@ -28,8 +28,11 @@ class Execution:
     validation_command: tuple[str, ...] = ()
     task_entry: str | None = None
     turn_envelope: bool = False
+    context_delivery: str = "off"
 
     def __post_init__(self) -> None:
+        if self.context_delivery not in ("off", "on") or (self.context_delivery != "off" and self.mode != "heartbeat"):
+            raise ValueError("context_delivery supports only off or on with heartbeat execution")
         if not isinstance(self.turn_envelope, bool) or (self.turn_envelope and self.mode != "heartbeat"):
             raise ValueError("turn_envelope requires heartbeat execution and a boolean opt-in")
         if self.mode not in MODES or self.context not in CONTEXTS:
@@ -182,6 +185,7 @@ def prepare_codex_home(
             # policy explicitly so model tools cannot silently lose it.
             "[shell_environment_policy.set]",
             'LOOPX_USAGE_PING = "0"',
+            *(['LOOPX_CONTEXT_DELIVERY = "on"', 'LOOPX_EXECUTION_MODE = "heartbeat"'] if execution.context_delivery != "off" else []),
             "[features]",
             f"goals = {str(execution.native_goal).lower()}",
             "unified_exec = true",

benchmark/runtime/context_delivery.py

@@ -0,0 +1,98 @@
+"""Explicit trial-local consumer of the product's native context transport.
+
+No packet semantics live here. The existing worker stages exact product hooks
+and trusts only their native hashes; quota owns projection and full recovery.
+"""
+from __future__ import annotations
+
+import json
+import os
+import stat
+import shlex
+from pathlib import Path
+import tempfile
+
+
+def preflight_context_instruction(env: dict[str, str], path: Path, digest: str, turn_id: str) -> str:
+    """Delay presentation of the original preflight until its native tool call.
+
+    No model input exists at host preflight time. Keep the original source/hash
+    and let the product's current-input observer choose full/delta later.
+    """
+    from loopx.command_invocation import loopx_cli_argv
+
+    prefix = [*loopx_cli_argv(env["LOOPX_CLI"]), "--format", "json", "quota"]
+    source = ["--capture-file", str(path), "--capture-sha256", digest]
+    project = [*prefix, "context-project", *source,
+        "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"],
+        "--turn-instance-id", turn_id, "--context-delivery", "on", "--runtime-profile", "generic_cli", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"],
+        "--decision-output-root", str(path.parent / "decisions")]
+    full = [*prefix, "context-read", *source]
+    return ("Present that same immutable packet with:\n```sh\n" + shlex.join(project)
+        + "\n```\nFollow the returned command and all next_command pages; reconstruct any delta against "
+        "its exact retained baseline before following the original interaction_contract. "
+        "If projection is unavailable, the baseline is absent/uncertain, or a response is incomplete, "
+        "follow its full recovery reference. If the projection command itself fails, read the original "
+        "complete packet with:\n```sh\n" + shlex.join(full) + "\n```\nFollow every next_command. "
+        "Neither presentation nor recovery renews this decision's authority or freshness. ")
+
+
+def prepare_context_delivery(env: dict[str, str], home: Path) -> None:
+    from loopx.control_plane.turn_driver.codex_context import codex_context_configuration
+    from loopx.control_plane.turn_driver.codex_stdio import StdioNativeGoalTransport
+
+    root = home / "loopx-context"
+    traces = home / "context-traces"
+    for directory in (root, traces):
+        directory.mkdir(mode=0o700, exist_ok=True)
+        info = directory.lstat()
+        if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077:
+            raise ValueError("context diagnostics require private, physical trial directories")
+    # Opt-in diagnostics are private to this trial, never global or a retention
+    # assertion. Include all IO/model-input costs when comparing this treatment.
+    # Keep candidate references across resumed wakes, but inspect only this
+    # native invocation's diagnostics. Older traces are evidence, not a growing
+    # hot-path scan or a reason to distrust intact current model input.
+    env["CODEX_ROLLOUT_TRACE_ROOT"] = tempfile.mkdtemp(prefix="wake-", dir=traces)
+    env["LOOPX_CODEX_CONTEXT_DIR"] = str(root)
+    configuration = codex_context_configuration(str(root), goal_id=env["LOOPX_GOAL_ID"],
+        agent_id=env["LOOPX_AGENT_ID"], scenario="codex_benchmark_heartbeat", cli_argv=[env["LOOPX_CLI"], "--cli-bin", env["LOOPX_CLI"]])
+    content = json.dumps({"hooks": configuration["hooks"]})
+    path = home / "hooks.json"
+    if path.is_symlink() or (path.exists() and path.read_text() != content):
+        raise ValueError("native context hooks changed within a trial; use a new trial")
+    if not path.exists():
+        path.write_text(content)
+        path.chmod(0o600)
+    # The native host is the hash/key owner. Do not duplicate its hashing or
+    # bypass trust for unrelated workspace hooks. Discovery launches no model.
+    with (home / "context-hooks-stderr.log").open("a") as stderr:
+        with StdioNativeGoalTransport.spawn(
+            # This process only reads native hook identities. Catalog/plugin
+            # startup can clone repositories into an otherwise private home;
+            # it is unrelated to discovery and must not affect solver config.
+            [env["CODEX_BIN"], "-c", "features.plugins=false",
+             "-c", "features.remote_plugin=false", "-c", "features.plugin_sharing=false",
+             "app-server", "--listen", "stdio://"],
+            cwd=env["LOOPX_PROJECT"], env=env, stderr=stderr, response_timeout_sec=10,
+        ) as native:
+            native.request("initialize", {"clientInfo": {"name": "loopx_context_transport", "version": "0.1.0"},
+                                          "capabilities": {"experimentalApi": True}})
+            native.notify("initialized", {})
+            listed = native.request("hooks/list", {"cwds": [env["LOOPX_PROJECT"]]})
+    hooks = [hook for entry in listed["data"] for hook in entry["hooks"]
+             if hook.get("sourcePath") == str(path.resolve())]
+    if len(hooks) != 2 or path.read_text() != content:
+        raise ValueError("native host did not discover the exact trial context hooks")
+    entries = []
+    for hook in hooks:
+        key, digest = hook.get("key"), hook.get("currentHash")
+        if not isinstance(key, str) or not key or not isinstance(digest, str) or not digest:
+            raise ValueError("native host returned no context-hook trust identity")
+        entries.append(f"{json.dumps(key)} = {{ trusted_hash = {json.dumps(digest)} }}")
+    # CLI override keys are dot-separated; native keys contain paths/dots.
+    # Carry the entire table as TOML so native keys remain literal strings.
+    # This isolated trial's native developer instructions survive compaction;
+    # a failed SessionStart handler cannot erase the read-only recovery route.
+    env["LOOPX_CODEX_CONTEXT_TRUST_ARGS"] = json.dumps(["-c", "hooks.state={" + ",".join(entries) + "}",
+        "-c", "developer_instructions=" + json.dumps(configuration["developer_instructions"])])

benchmark/runtime/harbor.py

@@ -72,6 +72,7 @@ def __init__(
         task_entry=None,
         planning_timeout_sec=None,
         turn_envelope=False,
+        context_delivery="off",
         native_goal_continue=False,
         task_objective=None,
         execution_observation=None,
@@ -96,6 +97,7 @@ def __init__(
             validation_command if validation_command is not None else (),
             task_entry,
             turn_envelope,
+            context_delivery,
         )
         if planning_timeout_sec is not None:
             raise ValueError(
@@ -167,7 +169,8 @@ def __init__(
         super().__init__(*args, **kwargs)
 
     def _turn_envelope_receipt(self) -> dict:
-        return turn_envelope_receipt(explicit=self.execution.turn_envelope)
+        return {**turn_envelope_receipt(explicit=self.execution.turn_envelope),
+                **({"context_delivery": "on"} if self.execution.context_delivery != "off" else {})}
 
     def _planning_receipt(self) -> dict:
         return {"native_goal_continue": True} if self.native_goal_continue else {}
@@ -611,6 +614,7 @@ def _worker_env(self, *, cwd: str) -> dict[str, str]:
                 **({"LOOPX_CHECKPOINT_REQUEST": self.checkpoint_request_path}
                    if self.checkpoint_request_path is not None else {}),
                 **({"LOOPX_TURN_ENVELOPE": "1"} if self.execution.turn_envelope else {}),
+                **({"LOOPX_CONTEXT_DELIVERY": "on"} if self.execution.context_delivery != "off" else {}),
                 **({"LOOPX_NATIVE_GOAL_CONTINUE": "1"} if self.native_goal_continue else {}),
                 "LOOPX_ITERATION_CONTEXT": self.execution.context,
                 "LOOPX_CODEX_SANDBOX": self.execution.sandbox,