fix(context): verify exact native hook trust

benchmark/runtime/RUNTIME.md

@@ -58,23 +58,29 @@ 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
+worker stages the product's generated lifecycle hooks (currently SessionStart,
+PreCompact, PostCompact, PostToolUse and Interrupt) in the trial's private Codex
+home and queries native `hooks/list` without invoking a model. Discovery must
+match every generated handler's event, command, matcher, timeout, async setting
+and enabled state, with unique native trust keys. Matching counts alone are
+insufficient; discovery order may differ. Only those exact native hashes receive
+per-process trust; workspace/plugin hooks are not trusted and no bypass flag or
+operator settings are used. Discovery-only plugin overrides do not reach the
+solver. Context delivery does not enable native input tracing. 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
+capture root into guard/reentry commands. Normal complete reader delivery establishes
+an explicit native-hook continuity assumption, not proof of later model retention.
+Known loss, incomplete delivery, startup, fork and compaction invalidate the
+baseline. A same-session resume keeps only a candidate until the native replay
+verifies every baseline reader page. Missing, changed or unsupported history
+falls back to full. Ordinary tools and turn endings retain continuity. An independent new
+decision reads its current full capture when the baseline is unavailable; do not
+pre-read an older capture just to seed a delta. A dependent continuation after
+context loss must recover its exact original capture through the small recovery
+reference, 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,
@@ -85,22 +91,22 @@ 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.
+earlier input replays did not include it. Count complete input and replay, recovery
+reads, native discovery and hook 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
+for activation, 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
+native projector present 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
@@ -110,11 +116,11 @@ 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;
+native tool presents the saved decision using the same native lifecycle projector;
 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.
-Uncertain retention keeps full input. If enabled native preparation cannot
+Invalidated or unavailable baselines keep full input. If enabled native preparation cannot
 establish safe recovery, the command returns a typed transport failure with the
 original full-read reference and receipt; use that explicit reference without
 rerunning admission. An optional cache write after recovery is established can

benchmark/runtime/context_delivery.py

@@ -38,9 +38,7 @@ def preflight_context_instruction(env: dict[str, str], path: Path, digest: str,
 
 
 def prepare_context_delivery(env: dict[str, str], home: Path) -> None:
-    from loopx.control_plane.turn_driver.codex_context import (
-        CONTEXT_LIFECYCLE_HOOKS, codex_context_configuration,
-    )
+    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"
@@ -76,13 +74,27 @@ def prepare_context_delivery(env: dict[str, str], home: Path) -> None:
             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) != len(CONTEXT_LIFECYCLE_HOOKS) or path.read_text() != content:
+    # Match the generated handler multiset, not just its size. Native discovery
+    # can reorder it; duplicates, disabled handlers and substitutions cannot
+    # stand in for an expected event/command/settings combination.
+    expected = [{"eventName": event[0].lower() + event[1:],
+                 "handlerType": handler["type"], "command": handler["command"],
+                 "matcher": group.get("matcher"), "timeoutSec": handler.get("timeout"),
+                 "async": handler.get("async", False)}
+                for event, groups in configuration["hooks"].items()
+                for group in groups for handler in group["hooks"]]
+    if not expected or len(hooks) != len(expected) or path.read_text() != content:
         raise ValueError("native host did not discover the exact trial context hooks")
-    entries = []
+    entries, keys = [], set()
     for hook in hooks:
+        actual = {field: hook.get(field) for field in expected[0]}
+        if hook.get("enabled") is not True or type(hook.get("async")) is not bool or actual not in expected:
+            raise ValueError("native host did not discover the exact trial context hooks")
+        expected.remove(actual)
         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")
+        if not isinstance(key, str) or not key or key in keys or not isinstance(digest, str) or not digest:
+            raise ValueError("native host returned no unique context-hook trust identity")
+        keys.add(key)
         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.

benchmark/tests/test_shared_codex_runtime.py

@@ -1185,7 +1185,11 @@ def test_context_setup_preserves_unsafe_or_unowned_inputs(tmp_path, kind):
 
 
 @pytest.mark.parametrize("provider", ["openai", "operator-provider"])
-def test_context_discovery_keeps_solver_configuration_and_unrelated_hooks(tmp_path, monkeypatch, provider):
+@pytest.mark.parametrize("mutation", [None, "additional_generated_handler", "missing", "extra",
+    "duplicate_handler", "duplicate_key", "event", "command", "matcher", "timeout", "async",
+    "async_number", "type", "disabled", "missing_enabled", "source", "missing_key", "missing_hash",
+    "malformed_key", "malformed_hash", "changed_file"])
+def test_context_discovery_keeps_solver_configuration_and_unrelated_hooks(tmp_path, monkeypatch, provider, mutation):
     from contextlib import contextmanager
     from benchmark.runtime.context_delivery import prepare_context_delivery
     from loopx.control_plane.turn_driver.codex_stdio import StdioNativeGoalTransport
@@ -1196,7 +1200,16 @@ def test_context_discovery_keeps_solver_configuration_and_unrelated_hooks(tmp_pa
     original_config = (f'model_provider={json.dumps(provider)}\n'
         + '[features]\nplugins=true\nremote_plugin=true\nplugin_sharing=true\n').encode()
     config.write_bytes(original_config)
-    methods, commands = [], []
+    methods, commands, discovered = [], [], []
+    if mutation == "additional_generated_handler":
+        from loopx.control_plane.turn_driver import codex_context
+        original = codex_context.codex_context_configuration
+        def extended(*args, **kwargs):
+            configuration = original(*args, **kwargs)
+            configuration["hooks"]["PostToolUse"][0]["hooks"].append(
+                {"type": "command", "command": "fixture-other-handler", "timeout": 15, "async": True})
+            return configuration
+        monkeypatch.setattr(codex_context, "codex_context_configuration", extended)
 
     @contextmanager
     def discovery(argv, **kwargs):
@@ -1206,21 +1219,60 @@ def request(method, params):
             if method == "initialize":
                 return {}
             assert method == "hooks/list", "discovery must not start model work"
-            own = [{"sourcePath": str((home / "hooks.json").resolve()),
-                    "key": f"own-{i}", "currentHash": f"native-hash-{i}"} for i in range(5)]
-            return {"data": [{"hooks": [*own, {"sourcePath": str(tmp_path / "other-hooks.json"),
+            path = home / "hooks.json"
+            configured = json.loads(path.read_text())["hooks"]
+            events = {"SessionStart": "sessionStart", "PreCompact": "preCompact", "PostCompact": "postCompact",
+                      "PostToolUse": "postToolUse", "Interrupt": "interrupt"}
+            assert set(configured) == set(events)
+            own = [{"sourcePath": str(path.resolve()), "eventName": events[event],
+                    "handlerType": handler["type"], "command": handler["command"],
+                    "timeoutSec": handler["timeout"], "matcher": group.get("matcher"),
+                    "async": handler.get("async", False), "enabled": True}
+                   for event, groups in configured.items() for group in groups for handler in group["hooks"]]
+            for i, hook in enumerate(own):
+                hook.update(key=f"own-{i}", currentHash=f"native-hash-{i}")
+            discovered.extend(dict(hook) for hook in own)
+            if mutation == "missing":
+                own.pop()
+            elif mutation == "extra":
+                own.append(own[-1] | {"key": "extra"})
+            elif mutation == "duplicate_handler":
+                own[-1] = own[0] | {"key": own[-1]["key"]}
+            elif mutation == "duplicate_key":
+                own[-1]["key"] = own[0]["key"]
+            elif mutation == "changed_file":
+                path.write_text('{}')
+            else:
+                changes = {"event": ("eventName", "preToolUse"), "command": ("command", "other"),
+                    "matcher": ("matcher", "Bash"), "timeout": ("timeoutSec", 999), "async": ("async", True),
+                    "async_number": ("async", 0), "type": ("handlerType", "prompt"), "disabled": ("enabled", False),
+                    "missing_enabled": ("enabled", None), "source": ("sourcePath", str(tmp_path / "wrong.json")),
+                    "missing_key": ("key", None), "missing_hash": ("currentHash", None),
+                    "malformed_key": ("key", 1), "malformed_hash": ("currentHash", [])}
+                if mutation in changes:
+                    field, value = changes[mutation]
+                    own[0][field] = value
+            # Native discovery has its own ordering, unrelated to config order.
+            return {"data": [{"hooks": [*reversed(own), {"sourcePath": str(tmp_path / "other-hooks.json"),
                 "key": "unrelated", "currentHash": "never-trust-this"}]}]}
         yield SimpleNamespace(request=request, notify=lambda *args: None)
 
     monkeypatch.setattr(StdioNativeGoalTransport, "spawn", discovery)
     env = {"LOOPX_CLI": "fixture-loopx", "LOOPX_GOAL_ID": "fixture", "LOOPX_AGENT_ID": "agent",
            "CODEX_BIN": "native-codex", "LOOPX_PROJECT": str(tmp_path)}
-    prepare_context_delivery(env, home)
+    if mutation in {None, "additional_generated_handler"}:
+        prepare_context_delivery(env, home)
+    else:
+        with pytest.raises(ValueError, match="exact trial context hooks|trust identity"):
+            prepare_context_delivery(env, home)
+        assert "LOOPX_CODEX_CONTEXT_TRUST_ARGS" not in env
+        assert config.read_bytes() == original_config
+        return
     assert methods == ["initialize", "hooks/list"]
     assignments = {commands[0][i + 1] for i, value in enumerate(commands[0]) if value == "-c"}
     assert assignments == {"features.plugins=false", "features.remote_plugin=false", "features.plugin_sharing=false"}
     trusted = json.loads(env["LOOPX_CODEX_CONTEXT_TRUST_ARGS"])
-    assert "native-hash-0" in trusted[1] and "native-hash-1" in trusted[1]
+    assert all(hook["key"] in trusted[1] and hook["currentHash"] in trusted[1] for hook in discovered)
     assert "unrelated" not in trusted[1] and "never-trust-this" not in trusted[1]
     assert all("features.plugins" not in value and "features.remote_plugin" not in value
                and "features.plugin_sharing" not in value for value in trusted)

loopx/control_plane/turn_driver/codex_context.py

@@ -262,8 +262,11 @@ def codex_context_configuration(directory: str, *, goal_id: str, agent_id: str,
         *(["--codex-app"] if scenario == "codex_app_heartbeat" else
           ["--runtime-profile", "codex_app_ssh_goal" if scenario == "codex_app_ssh_goal" else "generic_cli"]),
         option, directory]), "timeout": 10}
+    # Native Interrupt handlers have a 3s ceiling. Declare the effective value
+    # rather than letting discovery silently clamp a different configuration.
     return {"hooks": {event: [{**({"matcher": "^Bash$"} if event == "PostToolUse" else {}),
-                              "hooks": [handler]}] for event in CONTEXT_LIFECYCLE_HOOKS},
+                              "hooks": [{**handler, "timeout": 3} if event == "Interrupt" else handler]}]
+                      for event in CONTEXT_LIFECYCLE_HOOKS},
         "developer_instructions": codex_context_recovery_instructions(directory, goal_id=goal_id,
             agent_id=agent_id, cli_argv=cli_argv, session_root=session_root)}