v1.46.0: map builders can fork one shared session

docs/CHANGELOG.md

@@ -1,3 +1,76 @@
+## v1.46.0 — 2026-10-18
+
+**Map builders can fork one shared session.** Codex R169 and R179 (PM
+1.45.1, N16 grouped, Sol 6.1 high): each cold group builder started a fresh
+session and read the same contracts, rubric, world API and checks again. A
+builder took 178-261 s for 4 regions. Native built all 16 regions in about
+350 s, and its builders start from the pilot's context.
+
+- A map takes `fork: {"from": "<agent node>"}`. When that node has a recorded
+  session, each item's code builder forks it on its first visit
+  (`payload.fork_from`). A repair visit resumes the builder's own fork.
+  Judges never fork. Validation refuses a `fork` that is not
+  `{"from": <agent node>}`.
+- `fork_from` has the shape of `resume_from`. The resolved record carries
+  `mode: "fork"`. `resume_from` wins when both are present.
+- Codex runs `codex exec fork <id>`. It takes no thread lease, because a
+  fork leaves the source thread as it was. Any number of tasks can fork one
+  thread. The fork gets the full build contract (no resume delta), the
+  write sandbox, and `--disable memories --disable multi_agent`.
+- The receipt records the fork's own `thread_id`. Attempt usage comes from
+  the fork's own rollout. The SDK `turn.completed` total includes the copied
+  source turns, so it stays `session_cumulative`.
+- Claude Code already resumes with `--fork-session`, so a fork needs no
+  adapter change there.
+- Live probe: the fork's rollout counts 70,921 input tokens (43,264 cached)
+  where the SDK total says 126,420. The cache covers only the system prefix,
+  so each fork pays once for the copied history at the uncached rate.
+
+**Metadata that a live writer keeps in motion is source contention.** On a
+starved Windows runner, 6 of 200 attach loops failed with `source metadata
+did not stabilize`. `source_stamp` takes 8 samples of size, LastWriteTime
+and ChangeTime. When a writer moves them on every sample, it raises an
+`OSError`. That error had no contention mark, so it escaped every retry
+path. In the parent, attach raised it at once. In the helper, the session
+ended with `source changed or unavailable`. A worker then exited 1, not 75,
+so the supervisor did not respawn it and the job failed.
+
+- The error now carries `source_open_contention`, like a share violation.
+  Attach and launch bindings retry it inside their budget. Ordinary reads
+  still fail fast.
+- `_transient_attach_failure` counts source contention as transient, so a
+  worker that still meets it at the end of the budget exits 75.
+- The helper retries source contention only before its session opens.
+  After the open, the parent sends statements and `release`, not a new
+  `open`. Before, a share violation after the open made the helper wait for
+  an `open` and crash with `KeyError` on the next request.
+- The other failures on that runner were correct results after the 25 s
+  attach budget: 57 `ReadTimeout` (Windows admits one reader per store at a
+  time) and 21 `source changed` that the helper could not prove to be a
+  write. A worker respawns on both. Synthetic CPU, disk-flush and parallel
+  load (5 lanes, 0 failures) did not reproduce that runner.
+
+**Two Windows CI tests wait like the clients they model.** Product code is
+unchanged.
+
+- `test_concurrent_replay` failed on a Windows CI run with
+  `StoreMetadataDrift`. The first open saw a write and retried. On the retry
+  the ChangeTime of `state.sqlite3` moved, but its size and LastWriteTime did
+  not, and the metadata fence refused the binding. A Windows probe saw the same
+  ChangeTime-only move under live writers, with no rename or chmod. The fence
+  cannot tell this move from a rename away and back, so it stays. A worker
+  that hits it exits 75, and the supervisor respawns it (1.27.40). The test
+  runs 8 workers in one process, so each writer now respawns its attach like
+  the supervisor does (`_ATTACH_RESPAWN_LIMIT`).
+- `test_cli_json_defaults_to_the_mcp_compact_body` got `status: unavailable`
+  after `--timeout-seconds 1`: no store read was available in 1 s on a loaded
+  runner. The CLI await tests now wait 10 s, like the MCP await tests (1.33.5).
+  The job is complete, so the read returns at once.
+- The compact body of a job that is not terminal keeps `summary: ""`. The
+  TypeScript client types `summary` as a string and reads it on a timeout.
+- An unproven `source changed` still fails closed. A proven same-store write
+  already has a bounded retry in the helper.
+
 ## v1.45.1 — 2026-10-17
 
 **A map item's builder owns only its own verdict.** In the Codex R163 study

docs/FLOWS.md

@@ -264,6 +264,37 @@ nothing when no flow walks, and it does nothing in a Puppetmaster worker. It
 takes about 40 ms per edit and makes no model call. An edit hook cannot see
 a file that a shell command writes. Claim such a file yourself.
 
+### Builders fork one shared session
+
+A fresh builder in each item reads the same shared inputs (contracts, rubric,
+API, checks) before it writes a line. A map can opt in to `fork`: one agent
+node reads the shared inputs once, and each item's builder starts from a copy
+of that session.
+
+```json
+{"id": "prime", "kind": "agent", "role": "explore",
+ "task": "Read brief.md, the contracts and the checks. Do not edit files. Summarize what a builder needs."},
+{"id": "groups", "kind": "map", "items": "{{state.groups}}", "fork": {"from": "prime"},
+ "graph": {"...": "..."}}
+```
+
+With an edge `prime -> groups`:
+
+1. `fork.from` must name an agent node. Use a read-only role (`explore`) so
+   the source session writes no files.
+2. On its first visit, each item's code builder forks the source session.
+   Codex runs `codex exec fork`, which leaves the source thread as it was,
+   so all items can fork it at the same time. Claude Code always forks.
+3. A repair visit resumes the builder's own forked session, not the source.
+4. Judges never fork. A review never shares a builder's context.
+5. When the source node has no recorded session (it did not run, or it ran
+   on an adapter without sessions), builders start fresh as before.
+
+The forked session counts its own turn only. The copied history is not
+cached for the new session, so each fork pays once for the source's input
+tokens at the uncached rate. Keep the source turn short: read and summarize,
+do not explore.
+
 ## Reviews bind to the final source
 
 A judge's PASS is about the files that it read. When the judge or the code

puppetmaster/__init__.py

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

puppetmaster/adapters/codex.py

@@ -13,7 +13,7 @@
 from puppetmaster.usage import selected_token_usage
 from puppetmaster.session_lease import SessionLease, acquire_codex_thread
 from puppetmaster.worker_attribution import attribution_payload, codex_event_references
-from puppetmaster.worker_resume import resolved_resume, task_resume_record
+from puppetmaster.worker_resume import is_fork, resolved_resume, task_resume_record
 
 from . import codex_rollout
 
@@ -152,7 +152,9 @@ def _prepare_cli_invocation(
         resume_record = task_resume_record(task.payload, "codex")
         resume = resolved_resume(resume_record, "codex")
         lease = None
-        if resume is not None:
+        fork = resume is not None and is_fork(resume)
+        if resume is not None and not fork:
+            # A fork leaves its source thread as it was; only an in-place resume holds it.
             resume_record, lease = _lease_codex_thread(resume_record)
             resume = resolved_resume(resume_record, "codex")
         # A flow revisit: the thread already holds the original task, its contract,
@@ -213,6 +215,7 @@ def _prepare_cli_invocation(
             command = build_codex_resume_command(
                 executable=command_base,
                 session_id=str(resume["session_id"]),
+                fork=fork,
                 model=model,
                 sandbox=sandbox,
                 approval_policy=approval_policy,
@@ -249,6 +252,7 @@ def _prepare_cli_invocation(
                 "ephemeral_source": ephemeral_source,
                 "resume": resume_record,
                 "resumed": resume is not None,
+                "fork": fork,
                 "session_lease": lease,
                 "write_capable": write_capable,
                 # Builders and verdict reviewers answer in free text, not findings JSON.
@@ -296,7 +300,8 @@ def _worker_home(prepared: CliInvocation) -> Optional[Path]:
                 codex_home.ensure_system_skills(home, prepared.command[:prepared.command.index("exec")])
             except Exception:
                 pass  # Best effort: stock Codex still installs the bundle itself.
-        if home is not None and not resuming and prepared.command[-2:] == list(STDIN_PROMPT):
+        if home is not None and (not resuming or prepared.extras.get("fork")) \
+                and prepared.command[-2:] == list(STDIN_PROMPT):
             # Bounded workers never fork subagents, and must not accrue memories.
             prepared.command = [*prepared.command[:-2], "--disable", "memories",
                                 "--disable", "multi_agent", *STDIN_PROMPT]
@@ -428,6 +433,8 @@ def _finalize_cli_run(
         resume_record = prepared.extras.get("resume")
         resume_fields = {"resume": resume_record} if resume_record else {}
         resumed = bool(prepared.extras.get("resumed"))
+        # A fork reports its own new thread, never the one it copied.
+        expected_thread = resume_record if resumed and not prepared.extras.get("fork") else None
         timeout_seconds = self.timeout_for(task)
         cwd = Path(task.payload.get("cwd") or ".").resolve()
 
@@ -442,7 +449,7 @@ def _finalize_cli_run(
             stderr = completed.stderr
             # A persisted thread survives the timeout; record it so a follow-up
             # can resume it. The resolver still checks the session is on disk.
-            timeout_thread_id = observed_thread_id(events, resume_record if resumed else None)
+            timeout_thread_id = observed_thread_id(events, expected_thread)
             stdout_capture = capture_subprocess_stdout(
                 text=stdout,
                 task=task,
@@ -535,7 +542,7 @@ def _finalize_cli_run(
             ),
             "",
         )
-        thread_id = observed_thread_id(events, resume_record if resumed else None)
+        thread_id = observed_thread_id(events, expected_thread)
 
         stdout_capture = capture_subprocess_stdout(
             text=completed.stdout,
@@ -794,6 +801,12 @@ def _rollout_attempt_usage(
     """
     sid = _resumed_session_id(resume_record)
     try:
+        if sid is not None and is_fork(resume_record):
+            # The fork's rollout copies the source's turns; its own is the new one.
+            path = codex_rollout.rollout_path(home, observed_thread_id(parse_codex_events(stdout or "")))
+            if path is None:
+                return codex_rollout.unlinked("fork_rollout_not_found")
+            return codex_rollout.attempt_usage(path, baseline or frozenset())
         if sid is not None:
             return codex_rollout.attempt_usage(
                 codex_rollout.rollout_path(home, sid), baseline or frozenset())
@@ -910,6 +923,7 @@ def build_codex_resume_command(
     *,
     executable: Union[str, list[str]] = "codex",
     session_id: str,
+    fork: bool = False,
     model: Optional[str] = None,
     sandbox: str = "workspace-write",
     approval_policy: str = "never",
@@ -923,9 +937,11 @@ def build_codex_resume_command(
     ``-c sandbox_mode`` override and cwd comes from the subprocess. It never
     takes ``--ephemeral``, so the resumed thread stays resumable for the next
     revision. The prompt is read from stdin via the trailing ``-- -``.
+    ``fork`` builds ``codex exec fork <thread_id>`` (same options): a new
+    thread that starts from a copy of the source, which stays as it was.
     """
     command = command_parts(executable)
-    command.extend(["exec", "resume", str(session_id), "--json"])
+    command.extend(["exec", "fork" if fork else "resume", str(session_id), "--json"])
     command.extend(["-c", f'approval_policy="{approval_policy}"'])
     if sandbox:
         command.extend(["-c", f'sandbox_mode="{sandbox}"'])

puppetmaster/flow.py

@@ -165,6 +165,9 @@ def validate_graph(graph: Any, *, child: bool = False, depth: int = 0,
     for node in by_id.values():
         if node.get("kind") == "parallel":
             problems.extend(_parallel_problems(node, by_id))
+        source = (node.get("fork") or {}).get("from") if node.get("kind") == "map" else None
+        if isinstance(source, str) and (by_id.get(source) or {}).get("kind") != "agent":
+            problems.append(f"map {node['id']!r} fork from {source!r} is not an agent node")
     edges = graph.get("edges", [])
     if not isinstance(edges, list):
         return problems + ["edges must be a list"]
@@ -337,6 +340,10 @@ def _map_problems(node: dict, defaults: dict, depth: int) -> list[str]:
             problems.append(f"map {nid!r} reuse needs inputs: the files each item's result depends on")
         elif "names" in reuse and not isinstance(reuse["names"], (str, list)):
             problems.append(f"map {nid!r} reuse names must be a template or a list")
+    if "fork" in node:
+        fork = node["fork"]
+        if not isinstance(fork, dict) or set(fork) != {"from"} or not isinstance(fork.get("from"), str):
+            problems.append(f"map {nid!r} fork must be an object with from: the agent node to fork")
     policy = node.get("pass", "all")
     if not (policy == "all" or (isinstance(policy, (int, float)) and not isinstance(policy, bool)
                                 and 0 < policy <= 1)):
@@ -1739,6 +1746,10 @@ def _spec(self, key: str, node: dict, run: FlowRun, prev: str, visit: int):
             frame = "" if read_only else _item_frame(run)
             if frame and frame not in payload["resume_prompt"]:
                 payload["resume_prompt"] = f"{frame}\n\n{payload['resume_prompt']}"
+        elif not read_only and (run.parent or {}).get("fork_from") and node.get("role", "code") == "code":
+            # A builder starts from the map's shared session (inputs already read);
+            # judges stay fresh, so a review never shares a builder's context.
+            payload["fork_from"] = dict(run.parent["fork_from"])
         return WorkerSpec(role=key, instruction=task,
                           adapter=node.get("adapter") or defaults["adapter"], payload=payload)
 
@@ -1938,6 +1949,7 @@ def _run_map(self, node: dict, run: FlowRun, prev: str) -> NodeOutcome:
         if interrupted and not children:
             return NodeOutcome(ok=False, error=interrupted)
         created = False
+        fork_from = _fork_source(node, run)
         for index, (key, item) in enumerate(zip(keys, items)):
             if interrupted or key not in inflight["chosen"] or key in children:
                 continue
@@ -1948,7 +1960,7 @@ def _run_map(self, node: dict, run: FlowRun, prev: str) -> NodeOutcome:
             child = new_run(self.state_dir, child_graph, child_input,
                             state={"item": item, "index": index, "key": key},
                             parent={"run_id": run.run_id, "node": node["id"], "key": key, "depth": depth,
-                                    "peer_scopes": peers},
+                                    "peer_scopes": peers, "fork_from": fork_from},
                             continue_from=continued, backend=run.backend, worker_mode=run.worker_mode)
             children[key] = child.run_id
             created = True
@@ -2360,6 +2372,15 @@ def _child_graph_shape(node: dict) -> Any:
 _LEFT_OPEN = re.compile(r"\{\{[^{}]*\}\}")
 
 
+def _fork_source(node: dict, run: FlowRun) -> Optional[dict]:
+    """The session a map's builders fork from: ``fork.from``'s, once that node has run."""
+    source = (node.get("fork") or {}).get("from")
+    session = run.sessions.get(source) if source else None
+    if not session or not session.get("task_id"):
+        return None
+    return {"job_id": session["job_id"], "task_id": session["task_id"]}
+
+
 _PEER_LIMIT = 20
 
 

puppetmaster/mcp_server.py

@@ -1756,7 +1756,9 @@ def _build_tools() -> list[McpTool]:
                 "interrupted or waiting at a gate; every agent/judge node is a durable "
                 "Puppetmaster job. Nodes re-entered by a back-edge, and nodes of a run started "
                 "with continue_from, resume their own provider session with only the delta. "
-                "A map node fans out per-item child flows with bounded concurrency. Actions: "
+                "A map node fans out per-item child flows with bounded concurrency; with fork: "
+                "{\"from\": <agent node>} each item builder starts from a fork of that node's "
+                "session (judges stay fresh). Actions: "
                 "validate, save, run, status, wait, resume (answer a gate, or restart=true), "
                 "stop, cut, list. Stop and cut are cooperative: their reply has settled=false "
                 "and open_work while workers can still write; call wait, which returns when "

puppetmaster/readonly_worker.py

@@ -86,7 +86,11 @@ class BasicInfo(ctypes.Structure):
             write = (last.write - epoch) * 100
             if (first.write, first.change) == (last.write, last.change) and st.st_mtime_ns == write:
                 return (st.st_dev, st.st_ino, st.st_size, write, (last.change - epoch) * 100)
-        raise OSError('source metadata did not stabilize')
+        # A live writer moved the metadata on every sample. That is source
+        # contention, which binding reads retry. It is not a new file.
+        error = OSError('source metadata did not stabilize')
+        error.source_open_contention = True
+        raise error
     finally:
         if owned:
             os.close(fd)
@@ -620,7 +624,9 @@ def session_output(message):
                     error='unable to open database: source changed or unavailable: ' + str(exc),
                     source_open_contention=getattr(exc, 'source_open_contention', False),
                 )
-                if response['source_open_contention']:
+                # Only an open can retry in this helper. After the open, the
+                # parent sends statements and release, not a new open.
+                if response['source_open_contention'] and not opened:
                     emit(response)
                 else:
                     output(response)