fix(codex): share resume sessions across benchmark planning and execution

benchmark/LHTB/README.md

@@ -113,8 +113,8 @@ not replayed; bootstrap and Todo lifecycle follow the installed product version.
 ## Shared execution configuration
 
 `LOOPX_EXECUTION_MODE` selects `plain`, `native-goal`, `heartbeat` (default),
-`turn` or `loopx-goal`. `LOOPX_ITERATION_CONTEXT` defaults to `fresh`; only Turn
-accepts `resume-if-available`. Turn also requires `LOOPX_VALIDATION_COMMAND_JSON`,
+`turn` or `loopx-goal`. `LOOPX_ITERATION_CONTEXT` defaults to `fresh`; heartbeat and Turn
+accept `resume`, sharing the same Goal/Agent session across planning, wakes and Todos. Turn also requires `LOOPX_VALIDATION_COMMAND_JSON`,
 an argv array for the independently protected task validator. No generic
 benchmark scoring or hidden-verifier feedback is introduced.
 

benchmark/runtime/RUNTIME.md

@@ -30,15 +30,28 @@ agents:
 | --- | --- | --- |
 | `plain` | One Codex exec, native Goals disabled | Absent |
 | `native-goal` | Installed native Goal transport; objective `Finish the task.` | Absent |
-| `heartbeat` | Product thin heartbeat + external scheduler, fresh each wake | Present |
+| `heartbeat` | Product thin heartbeat + external scheduler, fresh or same-session resume | Present |
 | `turn` | Public Turn CLI, typed result, independent validation, settlement | Present |
 | `loopx-goal` | Product Goal body + installed native Goal transport | Present |
 
-Only `turn` accepts `iteration_context: resume-if-available`. Core session
-compatibility determines whether it actually resumes, including after changing
-Todo. Native Goal continuation stays with Codex; blocked Goals are not
-automatically unblocked. Plain exec versus Goal app-server also changes
-transport; it does not isolate the continuation effect alone.
+`heartbeat` and `turn` accept `iteration_context: resume`. The first invocation
+creates one Codex conversation; later planning checkpoints and execution wakes
+resume that exact native session ID for the trial's Goal and Agent, including
+when the selected Todo changes. Both drivers share the product's agent-scoped
+Codex session store. `fresh` remains the runner default and starts a new session
+on each invocation. The former context name is rejected, with no alias.
+
+Resume never falls back to a new conversation when a binding is corrupt, the
+trial home/workspace/model/settings change, or Codex returns a different ID.
+Repair the configuration or explicitly select `fresh`; the next observed fresh
+session replaces the binding. A timeout preserves an observed ID without
+claiming progress. Wakes remain serialized by the outer controller. Private
+wake receipts record the requested action and confirmed native session ID;
+aggregate trajectories copy each native session once.
+
+Native Goal continuation stays with Codex; blocked Goals are not automatically
+unblocked. Plain exec versus Goal app-server also changes transport; it does
+not isolate the continuation effect alone.
 
 `turn` requires `validation_command`, an argv list for an independently
 protected validator available inside the task environment. It receives the
@@ -67,10 +80,13 @@ planning process or missing result fails the entry; it never falls back to a
 generic Todo. A blocked entry retains the referenced blockers and starts no
 execution driver. Readback proves state and ownership, not semantic plan quality.
 
-Planning uses a separate fresh `codex exec` session with native Goals disabled
-for that call. Its session is not inserted into core Turn session bindings or
-resumed by the subsequent execution. This is a planning-contract ablation, not
-an exact reproduction of same-conversation interactive `$loopx` startup.
+Planning follows the chosen context policy for heartbeat and Turn. With
+`resume`, planning and execution share the same conversation, including later
+phase planning; each checkpoint still renders fresh public task inputs and
+validates actual Todo readback. With `fresh`, each planning/execution invocation
+starts a new conversation. LoopX Goal planning uses a separate exec conversation
+because native Goal execution owns its app-server thread lifecycle. The runner
+does not claim exact equivalence to interactive `$loopx` startup.
 The default `planning_timeout_sec` is 300; planning and preparation consume the
 same `scheduler_timeout_sec` phase budget as execution. Planning sessions are
 included in native session/token aggregation. No planning checkpoint is counted

benchmark/runtime/codex.py

@@ -10,7 +10,7 @@
 
 
 MODES = ("plain", "native-goal", "heartbeat", "turn", "loopx-goal")
-CONTEXTS = ("fresh", "resume-if-available")
+CONTEXTS = ("fresh", "resume")
 TASK_ENTRIES = ("seeded-todo", "loopx-planned")
 SANDBOXES = ("read-only", "workspace-write", "danger-full-access")
 
@@ -31,8 +31,8 @@ def __post_init__(self) -> None:
             raise ValueError("unsupported task entry")
         if self.task_entry == "loopx-planned" and not self.uses_loopx:
             raise ValueError("loopx-planned requires a LoopX execution mode")
-        if self.context != "fresh" and self.mode != "turn":
-            raise ValueError("resume-if-available currently requires mode=turn")
+        if self.context != "fresh" and self.mode not in {"turn", "heartbeat"}:
+            raise ValueError("resume requires mode=turn or mode=heartbeat")
         if self.sandbox not in SANDBOXES:
             raise ValueError("unsupported Codex sandbox")
         if not math.isfinite(self.timeout_seconds) or self.timeout_seconds <= 0:

benchmark/runtime/sessions.py

@@ -0,0 +1,117 @@
+"""Trial-local conversation continuity through the Codex session owner."""
+
+from __future__ import annotations
+
+import json
+import shutil
+from pathlib import Path
+from typing import Any
+
+from .codex import Execution
+
+from loopx.control_plane.goals.first_party_host_admission import (
+    FirstPartyHostGoalAdmission,
+    capture_first_party_host_goal_ref,
+)
+from loopx.control_plane.turn_driver.codex_cli import codex_cli_event_session_id
+from loopx.control_plane.turn_driver.codex_sessions import (
+    _store_codex_cli_session,
+    codex_session_profile_digest,
+    require_codex_session_profile,
+    select_codex_cli_session,
+)
+
+
+class BenchmarkSessionWake:
+    """Planning and heartbeat use the same Goal/Agent binding as governed Turns.
+
+    The outer controller serializes wakes. A native ID must be observed even on
+    timeout; neither missing history nor a failed resume permits a fresh fork.
+    """
+
+    def __init__(
+        self, env: dict[str, str], execution: Execution, receipt: dict[str, Any],
+    ) -> None:
+        self.root = Path(env["LOOPX_RUNTIME_ROOT"])
+        self.lineage = {
+            "goal_id": env["LOOPX_GOAL_ID"],
+            "agent_id": env["LOOPX_AGENT_ID"],
+        }
+        self.goal_ref = capture_first_party_host_goal_ref(
+            registry_path=Path(env["LOOPX_REGISTRY"]),
+            goal_id=self.lineage["goal_id"],
+        )
+        self.admission = FirstPartyHostGoalAdmission.for_plan(
+            registry_path=Path(env["LOOPX_REGISTRY"]),
+            goal_id=self.lineage["goal_id"],
+            planned_goal_ref=self.goal_ref,
+        )
+        if execution.context == "fresh":
+            self.admission.require_current()
+        binary = shutil.which(env["CODEX_BIN"])
+        if binary is None:
+            raise ValueError("Codex CLI executable is unavailable")
+        self.digest = codex_session_profile_digest(
+            project=Path(env["LOOPX_PROJECT"]),
+            codex_bin=binary,
+            home=Path(env["CODEX_HOME"]),
+            model=env["MODEL_NAME"],
+            reasoning_effort=env["REASONING_EFFORT"],
+            sandbox=execution.sandbox,
+        )
+        binding = (
+            select_codex_cli_session(
+                self.root,
+                lineage=self.lineage,
+                session_scope="agent",
+                goal_admission=self.admission,
+            )
+            if execution.context == "resume"
+            else None
+        )
+        if binding:
+            require_codex_session_profile(binding, self.digest)
+            if binding.get("operation_transport"):
+                raise ValueError("session requires its original managed transport")
+        self.session_id = binding["session_id"] if binding else None
+        self.receipt = receipt
+        receipt["session"] = {
+            "binding_scope": "agent",
+            "action": "resume" if binding else "start_new",
+            "session_id": self.session_id,
+        }
+
+    def observe(self, path: Path) -> None:
+        if not path.exists():
+            return  # No process was launched.
+        ids = set()
+        with path.open() as stream:
+            for line in stream:
+                try:
+                    event = json.loads(line)
+                except ValueError:
+                    continue
+                candidate = (
+                    codex_cli_event_session_id(event)
+                    if isinstance(event, dict)
+                    else None
+                )
+                if candidate:
+                    ids.add(candidate)
+        if len(ids) != 1 or (
+            self.session_id is not None and self.session_id not in ids
+        ):
+            self.receipt.update(ok=False, error_kind="session_identity_unconfirmed")
+            raise RuntimeError("Codex did not confirm the expected session identity")
+        observed = ids.pop()
+        self.admission.accept_result(
+            lambda: _store_codex_cli_session(
+                self.root,
+                lineage=self.lineage,
+                session_scope="agent",
+                session_id=observed,
+                goal_ref=self.goal_ref,
+                session_profile_digest=self.digest,
+            )
+        )
+        self.receipt["session"]["session_id"] = observed

benchmark/runtime/worker.py

@@ -134,6 +134,10 @@ def turn_command(
         ),
         "--iteration-context",
         execution.context,
+        "--session-scope",
+        "agent",
+        "--codex-reasoning-effort",
+        env["REASONING_EFFORT"],
         "--codex-bin",
         env["CODEX_BIN"],
         "--codex-model",
@@ -200,6 +204,32 @@ def run_native_goal(
                 receipt["native_goal"] = compact_native_goal_receipt(observed[0])
 
 
+def native_command(env, execution, stage, wake, session_wake):
+    if session_wake is None:
+        # Preserve the baseline/Goal-planning entrypoint and its transport inputs.
+        return [
+            env["CODEX_BIN"], "exec", "--json", "--skip-git-repo-check",
+            "--sandbox", execution.sandbox, "--cd", env["LOOPX_PROJECT"],
+            *(["-c", "features.goals=false", "--output-schema", str(wake / "planning-schema.json"),
+               "--output-last-message", str(wake / "planning-result.json")] if stage == "plan" else []),
+            "-",
+        ]
+    from loopx.control_plane.turn_driver.codex_cli import _codex_command
+
+    command = _codex_command(
+        codex_bin=env["CODEX_BIN"], project=Path(env["LOOPX_PROJECT"]),
+        schema_path=wake / "planning-schema.json" if stage == "plan" else None,
+        output_path=wake / "planning-result.json" if stage == "plan" else None,
+        sandbox=execution.sandbox, model=env["MODEL_NAME"],
+        reasoning_effort=env["REASONING_EFFORT"],
+        session_id=session_wake.session_id if session_wake else None,
+        mcp_server=None,
+    )
+    if stage == "plan":
+        command[-1:-1] = ["-c", "features.goals=false"]
+    return command
+
+
 def run_once(env: dict[str, str]) -> dict:
     execution = Execution(
         mode=env.get("LOOPX_EXECUTION_MODE", "heartbeat"),
@@ -229,6 +259,7 @@ def run_once(env: dict[str, str]) -> dict:
         "ok": False,
         "timed_out": False,
     }
+    session_wake = None
     pending_path = (
         Path(env.get("LOOPX_RUNTIME_ROOT", str(home))) / "benchmark-pending-turn.json"
     )
@@ -266,6 +297,11 @@ def run_once(env: dict[str, str]) -> dict:
             body = heartbeat_body(env, turn_id, native_goal=execution.native_goal)
         elif execution.mode == "plain":
             body = Path(env["LOOPX_TASK_DOC"]).read_text(encoding="utf-8")
+        session_wake = None
+        if execution.mode in {"heartbeat", "turn"} and not (execution.mode == "turn" and stage == "execute"):
+            from benchmark.runtime.sessions import BenchmarkSessionWake
+
+            session_wake = BenchmarkSessionWake(env, execution, receipt)
         with (wake / "stderr.log").open("w") as stderr:
             if execution.native_goal and stage == "execute":
                 run_native_goal(env, execution, body, receipt, stderr)
@@ -286,22 +322,7 @@ def run_once(env: dict[str, str]) -> dict:
                         pending.get("resume_turn_key"),
                     )
                     if execution.mode == "turn" and stage == "execute"
-                    else [
-                        env["CODEX_BIN"],
-                        "exec",
-                        "--json",
-                        "--skip-git-repo-check",
-                        "--sandbox",
-                        execution.sandbox,
-                        "--cd",
-                        env["LOOPX_PROJECT"],
-                        *([
-                            "-c", "features.goals=false",
-                            "--output-schema", str(wake / "planning-schema.json"),
-                            "--output-last-message", str(wake / "planning-result.json"),
-                        ] if stage == "plan" else []),
-                        "-",
-                    ]
+                    else native_command(env, execution, stage, wake, session_wake)
                 )
                 with (wake / "stdout.jsonl").open("w") as stdout:
                     with child_process(
@@ -347,13 +368,19 @@ def run_once(env: dict[str, str]) -> dict:
         receipt["error_kind"] = type(exc).__name__
         raise
     finally:
-        if (home / "sessions").is_dir():
-            # One authoritative copy per native session; resume must not count
-            # the same prefix again in every wake's aggregate trajectory.
-            shutil.copytree(
-                home / "sessions", log_root.parent / "sessions", dirs_exist_ok=True
-            )
-        (wake / "receipt.json").write_text(json.dumps(receipt, indent=2) + "\n")
+        try:
+            if session_wake is not None:
+                session_wake.observe(wake / "stdout.jsonl")
+        except BaseException as exc:
+            receipt.update(ok=False, error_kind=type(exc).__name__)
+            raise
+        finally:
+            if (home / "sessions").is_dir():
+                # One authoritative copy; resumed prefixes are never double counted.
+                shutil.copytree(
+                    home / "sessions", log_root.parent / "sessions", dirs_exist_ok=True
+                )
+            (wake / "receipt.json").write_text(json.dumps(receipt, indent=2) + "\n")
     return receipt