v1.47.0: opt-in shared Codex host for a flow walk

docs/CHANGELOG.md

@@ -1,3 +1,39 @@
+## v1.47.0 — 2026-10-21
+
+**Opt-in: one shared Codex host for a flow walk.** A Codex worker starts its
+own `codex exec` process: the launcher, the native binary, a login-shell
+snapshot and git probes. Native Codex runs its subagents as threads in one
+process. Codex lab R200 measured about 114 s from `codex exec` launch to the
+first model turn for each cold builder, because the lab guards every process.
+R202 measured 0.05 s from a native subagent spawn to its own turn.
+
+- Set `"defaults": {"codex_host": true}` in the graph, or
+  `PUPPETMASTER_CODEX_HOST=1` for the walker. A graph value wins. Off by
+  default.
+- The top walker starts one `codex app-server` on a Unix socket
+  (`codex_host.py`) and stops it at walk end. Child walkers and workers find
+  it through `PUPPETMASTER_CODEX_HOST_RECORD`. The next walk of the run stops
+  a host that a dead walker left.
+- Each Codex agent and judge runs as a thread on the host
+  (`codex_session.run_host_turn`, stdlib WebSocket client `codex_ws.py`). A
+  repair resumes its thread. A cut interrupts only its own turn. An approval
+  request is declined and fails the turn.
+- The adapter reports file changes and commands as `codex exec --json`
+  items, so the write-scope gate, rollout usage and resume do not change.
+  Each worker gets its own shell environment through the thread's
+  `shell_environment_policy`.
+- These runs use `codex exec`: a fork, a bypass run, a native-steer run, a
+  custom `executable`, extra arguments other than `-c key=value`, a thread in
+  another Codex home, and Windows (no Unix sockets). A host that does not
+  start, or a connection that fails before the turn, also falls back.
+- Each host turn writes `codex_host_turn.json`: host start and ready times,
+  the request time, the thread start and the turn start.
+- Replica (R179 regions graph, Sol 6.1 high, 3 interleaved pairs): host on
+  548.6 / 292.2 / 370.6 s, host off 345.4 / 600.4 / 372.8 s. All runs
+  passed the judge. Host runs: 69 of 69 turns on the host, 0 `codex exec`
+  dispatches, request to own-turn start median 1.13 s. A local `codex exec`
+  start is about 1.7 s, so the wall time does not change on this Mac.
+
 ## v1.46.2 — 2026-10-19
 
 **A nested builder's `revise` no longer blocks follow-up reuse.** Codex R196

docs/FLOWS.md

@@ -444,5 +444,33 @@ a resumed walker kills the command the dead walker left running (when the pid
 still names that command) and runs it again. A pid that now names another
 process is left alone.
 
+### One shared Codex host
+
+A Codex worker usually starts its own `codex exec` process: the launcher, the
+native binary, a login-shell snapshot and git probes. Native Codex runs its
+subagents as threads in one process. With the host on, the top walker starts
+one `codex app-server` on a Unix socket. Every Codex agent and judge of the
+walk, in map items too, then runs as a thread on that host.
+
+- Turn it on with `"defaults": {"codex_host": true}` in the graph, or set
+  `PUPPETMASTER_CODEX_HOST=1` for the walker. A graph value wins over the
+  environment. `false` turns it off.
+- A repair resumes its thread on the host. A cut stops only its own turn.
+- Each worker gets its own shell environment (ports, worker flags) through the
+  thread's `shell_environment_policy`.
+- The write-scope gate, rollout usage and session resume work as with
+  `codex exec`. The adapter reports the host's file changes and commands as
+  `codex exec --json` items.
+- These runs use `codex exec`: a fork (`fork`), a bypass run, a native-steer
+  run, a custom `executable`, extra arguments other than `-c key=value`, and
+  a thread that another Codex home holds.
+- When the host does not start, or a connection fails before the turn, the
+  worker runs `codex exec`. The walker stops the host at walk end, and the
+  next walk of the run stops a host that a dead walker left.
+- POSIX only. Windows runs `codex exec`.
+
+Each host turn writes `codex_host_turn.json` in its capture directory: host
+start and ready times, the request time, the thread start and the turn start.
+
 Runs live under `<state_dir>/flows/runs/<run_id>/`. Pass `--cwd` (CLI) or
 `cwd` (MCP) so the run's state follows the workspace.

puppetmaster/__init__.py

@@ -1,4 +1,4 @@
 """Puppetmaster: distributed agent workers with stitched shared memory."""
 
 __all__ = ["__version__"]
-__version__ = "1.46.2"
+__version__ = "1.47.0"

puppetmaster/adapters/codex.py

@@ -2,6 +2,7 @@
 
 import json
 import os
+import time
 from pathlib import Path
 from typing import Any, Mapping, Optional, Union
 
@@ -314,6 +315,14 @@ def _invoke_cli(
         cwd: Path,
         timeout_seconds: int,
     ) -> StreamedProcess:
+        host = _host_for(task, prepared)
+        if host is not None:
+            from puppetmaster.invocation import check_external_dispatch
+
+            check_external_dispatch()
+            hosted = self._invoke_host(task, prepared, cwd, timeout_seconds, host)
+            if hosted is not None:
+                return hosted
         if prepared.extras.get("resumed") or not bool((task.payload or {}).get("native_steer")):
             home = self._worker_home(prepared)
             from puppetmaster import codex_home
@@ -346,6 +355,92 @@ def _invoke_cli(
         except Exception:
             return super()._invoke_cli(task, prepared, cwd, timeout_seconds)
 
+    def _invoke_host(self, task: Task, prepared: CliInvocation, cwd: Path,
+                     timeout_seconds: int, host: dict) -> Optional[StreamedProcess]:
+        """Run the turn as a thread on the flow's shared host (see codex_host).
+
+        Returns None when the host fails before the turn starts: the caller
+        then runs ``codex exec``. The turn is reported as ``codex exec --json``
+        events, with the file changes and commands Codex made.
+        """
+        from puppetmaster.adapters.codex_session import run_host_turn
+        from puppetmaster.adapters.codex_ws import HostSocket
+        from puppetmaster.cancellation import JobCancelled, is_cancelled
+        from puppetmaster.fs_permissions import mkdir_private, write_private_text
+        from puppetmaster.invocation import current_attempt
+        from puppetmaster.swarm_reasoning import caller_pinned_effort
+        from ._streaming import _resolve_sidecar_state_dir, _write_dispatch_receipt, capture_dir
+
+        config = _host_config(task.payload.get("extra_args"))
+        prompt = str(prepared.extras.get("prompt") or task.instruction or "")
+        env = prepared.env
+        if env is None:
+            env = inject_worker_cli_env(apply_worktree_ports(os.environ.copy(), cwd))
+        config["shell_environment_policy"] = _host_shell_policy(env)
+        resume_record = prepared.extras.get("resume") if prepared.extras.get("resumed") else None
+        session_id = _resumed_session_id(resume_record)
+        home = Path(host["home"]) if host.get("home") else _codex_home().user_home(env)
+        baseline = _rollout_baseline(home, resume_record)
+        state_dir = _resolve_sidecar_state_dir()
+        log_dir = capture_dir(state_dir, task) if state_dir is not None else None
+        log_path = None
+        if log_dir is not None:
+            mkdir_private(log_dir)
+            log_path = log_dir / "codex_host.log"
+        receipt = _write_dispatch_receipt(
+            log_dir, task=task, sidecar_name="codex_host",
+            command=["codex-host", str(host.get("socket")), "resume" if session_id else "start"],
+            cwd=str(cwd), env=env, stdin_data=prompt, pid=int(host["pid"]))
+        budget = task.payload.get("max_output_bytes")
+        requested_at = time.time()
+        result = run_host_turn(
+            lambda: HostSocket(str(host["socket"]), timeout=10.0),
+            cwd, prompt,
+            model=str(prepared.extras.get("model") or "") or None,
+            sandbox=str(prepared.extras.get("sandbox") or "workspace-write"),
+            approval_policy=str(prepared.extras.get("approval_policy") or "never"),
+            config=config,
+            effort=caller_pinned_effort(task.payload),
+            ephemeral=bool(prepared.extras.get("ephemeral", True)),
+            resume_thread=session_id,
+            timeout=float(timeout_seconds),
+            cancellation_check=lambda: is_cancelled(task.job_id),
+            protocol_log_path=log_path,
+            max_output_bytes=budget if isinstance(budget, int) and not isinstance(budget, bool) and budget > 0 else None,
+        )
+        if result.thread_id is None and result.status == "failed":
+            return None
+        if log_dir is not None:
+            # Lab observation: host start, this request and its own turn start.
+            write_private_text(log_dir / "codex_host_turn.json", json.dumps({
+                "host_pid": host.get("pid"), "host_started_at": host.get("started_at"),
+                "host_ready_at": host.get("ready_at"), "requested_at": requested_at,
+                "thread_started_at": result.thread_started_at,
+                "turn_started_at": result.turn_started_at, "finished_at": time.time(),
+                "thread_id": result.thread_id, "resumed": session_id is not None,
+                "status": result.status}), lock=False)
+        if result.status == "cancelled":
+            raise JobCancelled(task.job_id)
+        stdout = _app_server_events(result)
+        usage = None
+        if session_id is not None or not prepared.extras.get("ephemeral", True):
+            usage = _rollout_attempt_usage(home, resume_record, baseline, stdout)
+        if usage is None and session_id is None:
+            usage = _app_server_attempt_usage(result)
+        output_limit = result.status == "output_limit"
+        return StreamedProcess(
+            returncode=0 if result.status == "completed" else 1,
+            stdout=stdout,
+            stderr=result.error or "",
+            timed_out=result.status == "timeout" or output_limit,
+            live_log_path=str(log_path) if log_path is not None else None,
+            spawn_error=None,
+            output_limit_hit=output_limit,
+            attempt_id=current_attempt(task),
+            dispatch_receipt=receipt,
+            attempt_usage=usage,
+        )
+
     def _invoke_app_server(self, task: Task, prepared: CliInvocation, cwd: Path,
                            timeout_seconds: int, cancelled) -> StreamedProcess:
         """Run the native_steer turn and report it as a ``codex exec --json`` stream."""
@@ -734,6 +829,7 @@ def _app_server_events(result: Any) -> str:
     if result.thread_id:
         events.append({"type": "thread.started", "thread_id": result.thread_id})
     events.append({"type": "turn.started"})
+    events.extend({"type": "item.completed", "item": item} for item in getattr(result, "items", ()))
     events.extend({"type": "item.completed", "item": {"type": "agent_message", "text": text}}
                   for text in result.messages)
     if result.status == "failed":
@@ -743,6 +839,58 @@ def _app_server_events(result: Any) -> str:
     return "\n".join(json.dumps(event) for event in events) + "\n"
 
 
+def _host_for(task: Task, prepared: CliInvocation) -> Optional[dict]:
+    """The flow's live shared host when this run can be a thread on it, else None.
+
+    A fork, a native-steer run, a bypass run, extra arguments other than
+    ``-c key=value`` and a thread held in another home run ``codex exec``.
+    """
+    from puppetmaster import codex_host
+
+    host = codex_host.current()
+    if host is None or prepared.extras.get("fork") or prepared.extras.get("bypass") \
+            or (task.payload or {}).get("executable"):
+        return None
+    if bool((task.payload or {}).get("native_steer")) or _host_config(task.payload.get("extra_args")) is None:
+        return None
+    resume = prepared.extras.get("resume") if prepared.extras.get("resumed") else None
+    session_id = _resumed_session_id(resume)
+    if session_id is not None:
+        home = _codex_home().home_for_session(session_id)
+        if home is None or str(home) != str(host.get("home")):
+            return None
+    return host
+
+
+def _host_config(extra_args: object) -> Optional[dict]:
+    """``-c key=value`` arguments as a thread config, or None for any other argument."""
+    parts = command_parts(extra_args) if extra_args else []
+    config: dict = {}
+    index = 0
+    while index < len(parts):
+        if parts[index] != "-c" or index + 1 >= len(parts) or "=" not in parts[index + 1]:
+            return None
+        key, raw = parts[index + 1].split("=", 1)
+        try:
+            value: Any = json.loads(raw)
+        except ValueError:
+            value = raw.strip("'")
+        config[key.strip()] = value
+        index += 2
+    return config
+
+
+def _host_shell_policy(env: Mapping[str, str]) -> dict:
+    """The thread shell environment: this worker's env where it differs from the host's."""
+    host_env = inject_worker_cli_env(os.environ.copy())
+    changed = {key: value for key, value in env.items() if host_env.get(key) != value}
+    policy: dict = {"inherit": "all", "set": changed}
+    dropped = sorted(key for key in host_env if key not in env)
+    if dropped:
+        policy["exclude"] = dropped
+    return policy
+
+
 def _app_server_attempt_usage(result: Any) -> Optional[dict]:
     """The app-server's token counts as attempt usage, or None when it sent none.