fix(context): share native continuity and bounded complete recovery

apps/presentation/dashboard/smoke/chat-model-menu-smoke.mjs

@@ -3,12 +3,12 @@ import { createElement } from "react";
 import { renderToStaticMarkup } from "react-dom/server";
 import { ChatModelMenu } from "../src/features/personal-workspace/chat-model-menu.tsx";
 import { ChannelHeader } from "../src/features/personal-workspace/channel-header.tsx";
+import { ComposerRuntimeControls } from "../src/features/personal-workspace/composer-controls.tsx";
 import { WorkspaceI18nProvider } from "../src/features/personal-workspace/i18n.tsx";
 
 // Presentation semantics only. Real width, focus, IME and native delivery still
 // belong to the packaged browser and production HTTP/provider checks.
 const originalWindow = globalThis.window;
-let assertions = 0;
 try {
   for (const locale of ["zh-CN", "en"]) {
     globalThis.window = { localStorage: { getItem: () => locale } };
@@ -61,7 +61,6 @@ try {
     assert.doesNotMatch(header("error"), /<button[^>]*disabled=""/, "A failed refresh can be retried");
     const stewardProps = {
       agents: [{ agentId: "codex", label: "Codex", available: true }], selectedAgentId: "codex",
-      selectedAgentModelReadback: { agent_model_id: "goal-only-model", agent_reasoning_effort: "high", agent_service_tier: "priority" },
       managerChannelBinding: { executor_endpoint: "codex", executor_kind: "individual", model: "gpt-6.1-sol", reasoning_effort: "xhigh", selection_policy: "preferred", allocation_reason: "product_default" },
       managerRuntime: { runtime_profile: "trusted_owner", sandbox: "danger-full-access", status: "ready" },
       onOpenManagerSettings() {},
@@ -73,9 +72,24 @@ try {
     assert.doesNotMatch(steward, /goal-only-model|Fast|personal-agent-select/, "The steward has one model owner, without Goal runtime/default controls");
     const restricted = header("idle", { ...stewardProps, managerRuntime: { runtime_profile: "restricted", sandbox: "read-only", status: "ready" } });
     assert.match(restricted, /只读|Read-only/);
-    const goal = header("idle", { ...stewardProps, selectedGoal: { goalId: "public-case", title: "Public Goal", agentLanes: [] } });
-    assert.doesNotMatch(goal, /goal-only-model|Fast|personal-chat-model-menu/, "The Goal reading header leaves sending choices in the single composer");
-    assert.doesNotMatch(goal, /gpt-6\.1-sol/, "Goal readback does not inherit the steward model");
+    const goalHeader = header("idle", { ...stewardProps, selectedGoal: { goalId: "public-case", title: "Public Goal", agentLanes: [] } });
+    assert.doesNotMatch(goalHeader, /goal-only-model|Fast|personal-chat-model-menu|personal-agent-select/, "Goal sending controls belong to the composer, and its header never inherits the steward binding");
+    assert.doesNotMatch(goalHeader, /gpt-6\.1-sol/, "Goal readback does not inherit the steward model");
+    const composer = (overrides = {}) => renderToStaticMarkup(createElement(WorkspaceI18nProvider, null,
+      createElement(ComposerRuntimeControls, { agents: stewardProps.agents, agentId: "codex", goalId: "public-case",
+        original: false, model: "next-model", effort: "low", onAgent() {}, onModel() {}, onEffort() {}, onDefaults() {},
+        readback: { agent_model_id: "goal-only-model", agent_reasoning_effort: "high", agent_service_tier: "priority" }, ...overrides })));
+    const goal = composer();
+    assert.match(goal, /goal-only-model/);
+    assert.match(goal, /Fast/);
+    assert.match(goal, new RegExp(defaultsLabel), "Goal defaults stay discoverable beside the actual model readback");
+    const composerSummary = goal.slice(goal.indexOf("<summary"), goal.indexOf("</summary>"));
+    assert.doesNotMatch(composerSummary, /next-model|low/, "The shipped composer forwards actual Session readback separately from pending configuration");
+    const statusComposer = composer({ agentId: "status-only" });
+    assert.doesNotMatch(statusComposer, /goal-only-model|Fast|personal-chat-model-menu/, "Status-only has no model Session to report or configure");
+    const originalComposer = composer({ original: true });
+    assert.match(originalComposer, /goal-only-model/, "Original conversation controls still show their observed Session");
+    assert.doesNotMatch(originalComposer, new RegExp(`${defaultsLabel}|${locale === "zh-CN" ? "新对话配置" : "New conversation settings"}|personal-agent-select`), "Original conversation controls do not offer another executor or ordinary conversation defaults");
     const linked = header("idle", { readOnlySourceLabel: "Historical run", selectedGoal: { goalId: "source-case", title: "Historical run", agentLanes: [] } });
     // The persistent badge is its trigger name and visible text. Its disclosure
     // panel is opened on demand and may explain the same source once.
@@ -87,7 +101,6 @@ try {
     assert.doesNotMatch(badge, /Historical run/, "The read-only source badge does not repeat the visible Goal title");
     const distinct = header("idle", { readOnlySourceLabel: "Remote build", selectedGoal: { goalId: "source-case", title: "Public Goal", agentLanes: [] } });
     assert.match(distinct, /class="personal-read-only-source-label"[^>]*>Remote build</, "A source the Goal title does not name stays visible in the badge");
-    assertions += 33;
   }
-  console.log(`Chat model/header: ${assertions} bilingual model, defaults and accessible refresh assertions passed (SSR only)`);
+  console.log("Chat model/header/composer: bilingual model, defaults and accessible refresh assertions passed (SSR only)");
 } finally { globalThis.window = originalWindow; }

benchmark/runtime/context_delivery.py

@@ -10,14 +10,13 @@
 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.
+    Keep the original source/hash. The product's lifecycle owner chooses the
+    full/delta view when the native tool actually presents it.
     """
     from loopx.command_invocation import loopx_cli_argv
 
@@ -31,29 +30,24 @@ def preflight_context_instruction(env: dict[str, str], path: Path, digest: str,
         + "\n```\nConsume any returned page content, then follow next_command through eof. "
         "If no page content is returned, run command and follow 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, "
+        "Normal completed reads use the explicit native-hook continuity assumption. "
+        "If projection is unavailable, the baseline was invalidated, 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_context import (
+        CONTEXT_LIFECYCLE_HOOKS, 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)
+    root.mkdir(mode=0o700, exist_ok=True)
+    info = root.lstat()
+    if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077:
+        raise ValueError("context delivery requires a private, physical trial directory")
     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"]])
@@ -82,7 +76,7 @@ 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) != 2 or path.read_text() != content:
+    if len(hooks) != len(CONTEXT_LIFECYCLE_HOOKS) or path.read_text() != content:
         raise ValueError("native host did not discover the exact trial context hooks")
     entries = []
     for hook in hooks:
@@ -94,5 +88,6 @@ def prepare_context_delivery(env: dict[str, str], home: Path) -> None:
     # 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"])])
+    native_args = ["-c", "hooks.state={" + ",".join(entries) + "}",
+        "-c", "developer_instructions=" + json.dumps(configuration["developer_instructions"])]
+    env["LOOPX_CODEX_CONTEXT_TRUST_ARGS"] = json.dumps(native_args)

benchmark/runtime/worker.py

@@ -150,6 +150,9 @@ def child_process(command: list[str], *, env: dict[str, str], stdout, stderr,
                 stderr=stderr,
                 text=True,
                 start_new_session=True,
+                # Opted-in replay checkpoints require owner-only native writes.
+                # This affects the isolated child, never the parent or off mode.
+                umask=0o077 if os.name == "posix" and env.get("LOOPX_CONTEXT_DELIVERY") == "on" else -1,
             )
         yield process
     finally:
@@ -514,19 +517,16 @@ def run_once(env: dict[str, str]) -> dict:
         if execution.context_delivery != "off":
             from benchmark.runtime.context_delivery import prepare_context_delivery
 
-            prior_trace = env.get("CODEX_ROLLOUT_TRACE_ROOT")
             try:
-                prepare_context_delivery(env, home)
+                context_env = dict(env)
+                prepare_context_delivery(context_env, home)
+                env.update(context_env)
                 receipt["context_transport"] = {"available": True}
             except Exception as exc:
                 # Optional presentation cannot block an otherwise admitted
-                # trial. Unsupported hooks/diagnostics keep ordinary full IO.
+                # trial. Unavailable native hooks keep ordinary full IO.
                 env.pop("LOOPX_CODEX_CONTEXT_DIR", None)
                 env.pop("LOOPX_CODEX_CONTEXT_TRUST_ARGS", None)
-                if prior_trace is None:
-                    env.pop("CODEX_ROLLOUT_TRACE_ROOT", None)
-                else:
-                    env["CODEX_ROLLOUT_TRACE_ROOT"] = prior_trace
                 receipt["context_transport"] = {"available": False, **_error_tail(exc)}
         body = "Finish the task."
         if stage == "plan":

benchmark/tests/test_native_checkpoint.py

@@ -930,6 +930,21 @@ def test_interrupted_cursor_requires_exact_unreleased_final_guard(tmp_path, monk
     rendered[-1]["items"][-1]["command"] = shlex.join(["/bin/bash", "-lc",
         "LOOPX_TURN=turn-1\n" + shlex.join(argv).replace("--turn-instance-id turn-1", '--turn-instance-id "${LOOPX_TURN:?}"')])
     _verify_interrupted_guard(rendered, guard)
+    for setting in ("off", "on"):
+        explicit = copy.deepcopy(turns)
+        explicit[-1]["items"][-1]["command"] = shlex.join([
+            "/bin/bash", "-lc", shlex.join(argv + ["--context-delivery", setting])])
+        _verify_interrupted_guard(explicit, {**guard, "context_delivery": setting})
+    with pytest.raises(ValueError, match="routes or identity"):
+        _verify_interrupted_guard(turns, {**guard, "context_delivery": "on"})
+    with pytest.raises(ValueError, match="routes or identity"):
+        _verify_interrupted_guard(turns, {**guard, "context_delivery": None})
+    # Equivalent implicit/explicit off commands cannot hide an earlier call.
+    repeated = copy.deepcopy(turns)
+    repeated[-1]["items"].append({**item, "id": "later-call", "command": shlex.join([
+        "/bin/bash", "-lc", shlex.join(argv + ["--context-delivery", "off"])])})
+    with pytest.raises(ValueError, match="ambiguous"):
+        _verify_interrupted_guard(repeated, guard)
     # A caller cannot change the captured presentation/authority mode on restore.
     opposite = list(argv)
     if "--turn-envelope" in opposite:

benchmark/tests/test_shared_codex_runtime.py

@@ -1134,6 +1134,24 @@ def unexpected(*args):
     assert not (Path(env["LOOPX_CODEX_HOME"]) / "hooks.json").exists()
 
 
+@pytest.mark.parametrize("setting,mode", [("off", 0o664), ("on", 0o600)])
+@pytest.mark.skipif(os.name != "posix", reason="Native POSIX file creation permissions")
+def test_context_native_writer_is_private_without_changing_off_umask(tmp_path, setting, mode):
+    from benchmark.runtime import worker
+
+    original = os.umask(0o002)
+    try:
+        command = [sys.executable, "-c", "from pathlib import Path; Path('native-history').write_bytes(b'fixture')"]
+        with worker.child_process(command, env={**os.environ, "LOOPX_PROJECT": str(tmp_path),
+                "LOOPX_CONTEXT_DELIVERY": setting}, stdout=subprocess.PIPE, stderr=subprocess.PIPE) as child:
+            child.communicate(timeout=10)
+            assert child.returncode == 0
+        assert (tmp_path / "native-history").stat().st_mode & 0o777 == mode
+        assert os.umask(0o002) == 0o002
+    finally:
+        os.umask(original)
+
+
 def test_context_selection_is_fixed_by_trial_home(tmp_path):
     args = settings(tmp_path, execution=Execution(mode="heartbeat", context_delivery="on"))
     prepare_codex_home(**args)
@@ -1151,7 +1169,7 @@ def test_context_setup_preserves_unsafe_or_unowned_inputs(tmp_path, kind):
         protected = home / "hooks.json"
         protected.write_text('{"operator": "unchanged"}')
     else:
-        protected = home / "context-traces"
+        protected = home / "loopx-context"
         if kind == "public_directory":
             protected.mkdir(mode=0o755)
             protected.chmod(0o755)
@@ -1166,15 +1184,17 @@ def test_context_setup_preserves_unsafe_or_unowned_inputs(tmp_path, kind):
     assert "LOOPX_CODEX_CONTEXT_TRUST_ARGS" not in env
 
 
-def test_context_discovery_keeps_solver_configuration_and_unrelated_hooks(tmp_path, monkeypatch):
+@pytest.mark.parametrize("provider", ["openai", "operator-provider"])
+def test_context_discovery_keeps_solver_configuration_and_unrelated_hooks(tmp_path, monkeypatch, provider):
     from contextlib import contextmanager
     from benchmark.runtime.context_delivery import prepare_context_delivery
     from loopx.control_plane.turn_driver.codex_stdio import StdioNativeGoalTransport
 
     home = tmp_path / "home"
     home.mkdir()
     config = home / "config.toml"
-    original_config = b"[features]\nplugins=true\nremote_plugin=true\nplugin_sharing=true\n"
+    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 = [], []
 
@@ -1187,7 +1207,7 @@ def request(method, params):
                 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(2)]
+                    "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"),
                 "key": "unrelated", "currentHash": "never-trust-this"}]}]}
         yield SimpleNamespace(request=request, notify=lambda *args: None)
@@ -1205,6 +1225,11 @@ def request(method, params):
     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)
     assert config.read_bytes() == original_config
+    overrides = {trusted[i + 1].split("=", 1)[0]: trusted[i + 1].split("=", 1)[1]
+                 for i, value in enumerate(trusted) if value == "-c"}
+    assert not any(key.startswith("model_provider") for key in overrides)
+    assert "CODEX_ROLLOUT_TRACE_ROOT" not in env
+    assert not (home / "context-traces").exists()
 
 
 @pytest.mark.parametrize("value", [True, False, "1", "codex_benchmark_heartbeat", "codex_app_heartbeat", "all"])