diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index a93b81e..d05580b 100644
--- a/docs/CHANGELOG.md
+++ b/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
diff --git a/docs/FLOWS.md b/docs/FLOWS.md
index 1cea094..1620054 100644
--- a/docs/FLOWS.md
+++ b/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
diff --git a/puppetmaster/__init__.py b/puppetmaster/__init__.py
index ec39e67..8cfa898 100644
--- a/puppetmaster/__init__.py
+++ b/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"
diff --git a/puppetmaster/adapters/codex.py b/puppetmaster/adapters/codex.py
index c04de24..e694153 100644
--- a/puppetmaster/adapters/codex.py
+++ b/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}"'])
diff --git a/puppetmaster/flow.py b/puppetmaster/flow.py
index 7c56fa0..a4f8663 100644
--- a/puppetmaster/flow.py
+++ b/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
 
 
diff --git a/puppetmaster/mcp_server.py b/puppetmaster/mcp_server.py
index f341952..2e4cd4e 100644
--- a/puppetmaster/mcp_server.py
+++ b/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 "
diff --git a/puppetmaster/readonly_worker.py b/puppetmaster/readonly_worker.py
index 1582bf4..19b1ef8 100644
--- a/puppetmaster/readonly_worker.py
+++ b/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)
diff --git a/puppetmaster/worker_resume.py b/puppetmaster/worker_resume.py
index 8444c42..ce8debd 100644
--- a/puppetmaster/worker_resume.py
+++ b/puppetmaster/worker_resume.py
@@ -6,6 +6,11 @@
 job/task/role in the same store) or ``resume_session_id`` (an id the caller
 kept itself). Resolution is stamped back as ``payload["resume"]`` once, so
 retries reuse the same record instead of re-resolving.
+
+``fork_from`` (same shape as ``resume_from``) asks for a new session that
+starts from a copy of the prior one: the record carries ``mode: "fork"``.
+Codex forks with ``codex exec fork`` and leaves the source thread as it was,
+so any number of tasks can fork one session. Claude Code always forks.
 """
 from __future__ import annotations
 
@@ -22,7 +27,11 @@
 
 def resume_requested(payload: Optional[dict]) -> bool:
     payload = payload or {}
-    return bool(payload.get("resume_from") or payload.get("resume_session_id"))
+    return bool(payload.get("resume_from") or payload.get("resume_session_id") or payload.get("fork_from"))
+
+
+def is_fork(record: Optional[dict]) -> bool:
+    return isinstance(record, dict) and record.get("mode") == "fork"
 
 
 def task_resume_record(payload: Optional[dict], adapter: str) -> Optional[dict]:
@@ -48,8 +57,9 @@ def task_resume_record(payload: Optional[dict], adapter: str) -> Optional[dict]:
 
 
 def claim_resumed_session(record: Optional[dict], claimed: dict, role: str) -> Optional[dict]:
-    """Allow one task per job to continue a codex thread; ``codex exec resume`` has no fork."""
-    if not isinstance(record, dict) or record.get("status") != "resolved" or record.get("adapter") != "codex":
+    """Allow one task per job to continue a codex thread in place; any number can fork it."""
+    if (not isinstance(record, dict) or record.get("status") != "resolved" or record.get("adapter") != "codex"
+            or is_fork(record)):
         return record
     session_id = str(record.get("session_id"))
     holder = claimed.get(session_id)
@@ -139,10 +149,13 @@ def resolve_worker_resume(store: Any, payload: Optional[dict], adapter: str) ->
         return existing
     if not resume_requested(payload):
         return None
+    fork = bool(payload.get("fork_from")) and not (payload.get("resume_from") or payload.get("resume_session_id"))
+    request = {**payload, "resume_from": payload["fork_from"]} if fork else payload
     try:
-        return _resolve(store, payload, str(adapter or ""))
+        record = _resolve(store, request, str(adapter or ""))
     except Exception as exc:
-        return _unavailable(payload, f"resume lookup failed: {type(exc).__name__}: {exc}")
+        record = _unavailable(request, f"resume lookup failed: {type(exc).__name__}: {exc}")
+    return {**record, "mode": "fork"} if fork else record
 
 
 def _request_echo(payload: dict) -> dict:
diff --git a/puppetmaster/worker_runtime.py b/puppetmaster/worker_runtime.py
index 24b7e1f..7c9f360 100644
--- a/puppetmaster/worker_runtime.py
+++ b/puppetmaster/worker_runtime.py
@@ -882,15 +882,16 @@ def _transient_attach_failure(exc: BaseException) -> bool:
     """Attach failures a fresh process can expect to get past.
 
     Metadata-only drift (macOS provenance xattr, same-mode chmod), lock or
-    busy contention and helper timeouts are transient. A replaced store or a
-    missing schema is not: respawning would only bind the wrong store or fail
-    again.
+    busy contention, source contention (a share violation, or metadata that
+    a live writer kept in motion) and helper timeouts are transient. A
+    replaced store or a missing schema is not: respawning would only bind the
+    wrong store or fail again.
     """
     import sqlite3
 
     from puppetmaster.identity import StoreMetadataDrift
 
-    if isinstance(exc, StoreMetadataDrift):
+    if isinstance(exc, StoreMetadataDrift) or getattr(exc, "source_open_contention", False):
         return True
     return isinstance(exc, (sqlite3.OperationalError, TimeoutError, BlockingIOError, PermissionError))
 
diff --git a/pyproject.toml b/pyproject.toml
index ccb61bd..1a3f439 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "puppetmaster-ai"
-version = "1.45.1"
+version = "1.46.0"
 description = "Gunicorn-style agent swarms with Redis-like shared memory and artifact stitching. (Imported as `puppetmaster`; published as `puppetmaster-ai` because the bare PyPI name is held by an abandoned 2019 project, name-reassignment pending.)"
 readme = "README.md"
 requires-python = ">=3.9"
diff --git a/tests/test_attach_metadata_drift.py b/tests/test_attach_metadata_drift.py
index 8b83c38..1535fa6 100644
--- a/tests/test_attach_metadata_drift.py
+++ b/tests/test_attach_metadata_drift.py
@@ -71,6 +71,10 @@ def test_replacement_and_missing_schema_are_not_respawned(self) -> None:
             StoreIdentityError("store removed or replaced since selection; explicitly reopen")))
         self.assertFalse(_transient_attach_failure(SqliteSchemaError("no schema")))
         self.assertTrue(_transient_attach_failure(StoreMetadataDrift("drift")))
+        unstable = OSError("source metadata did not stabilize")
+        unstable.source_open_contention = True
+        self.assertTrue(_transient_attach_failure(unstable))
+        self.assertFalse(_transient_attach_failure(OSError("other")))
         import sqlite3
         self.assertTrue(_transient_attach_failure(sqlite3.OperationalError("database is locked")))
 
diff --git a/tests/test_attempt_ledger.py b/tests/test_attempt_ledger.py
index 3d5a42a..2a57cfa 100644
--- a/tests/test_attempt_ledger.py
+++ b/tests/test_attempt_ledger.py
@@ -18,7 +18,9 @@
 from puppetmaster.attempts import (
     ExecutionAttempt, UsageObservation, LedgerConflictError, canonical_record,
 )
+from puppetmaster.identity import StoreMetadataDrift
 from puppetmaster.models import AgentRun, Task, TaskStatus
+from puppetmaster.orchestrator import _ATTACH_RESPAWN_LIMIT
 from puppetmaster.sqlite_store import SQLiteSwarmStore, SqliteSchemaError
 from puppetmaster.store import SwarmStore
 
@@ -146,9 +148,18 @@ class SQLiteLedgerTests(LedgerContract, unittest.TestCase):
 
     def test_concurrent_replay(self):
         def write(_):
-            store = SQLiteSwarmStore(self.root)
-            store.attach()
-            return store.record_attempt(self.attempt)
+            # Each writer is a worker. Windows can move ChangeTime alone under
+            # live writers. Attach then raises StoreMetadataDrift, and the
+            # supervisor respawns the worker.
+            for respawn in range(_ATTACH_RESPAWN_LIMIT + 1):
+                store = SQLiteSwarmStore(self.root)
+                try:
+                    store.attach()
+                except StoreMetadataDrift:
+                    if respawn == _ATTACH_RESPAWN_LIMIT:
+                        raise
+                    continue
+                return store.record_attempt(self.attempt)
         with ThreadPoolExecutor(max_workers=4) as pool:
             inserted = list(pool.map(write, range(8)))
         self.assertEqual(sum(inserted), 1)
diff --git a/tests/test_await_compact.py b/tests/test_await_compact.py
index cfeebd0..95f358e 100644
--- a/tests/test_await_compact.py
+++ b/tests/test_await_compact.py
@@ -294,8 +294,11 @@ def _cli(self, job_id: str, *extra: str) -> tuple[int, str]:
         out = io.StringIO()
         with contextlib.redirect_stdout(out):
             rc = cli_main(
+                # Same patience as the MCP helper: the job is complete, so a
+                # read returns at once. A loaded Windows runner needed more than
+                # 1 s before the store read was available.
                 ["--state-dir", str(self.store.root), "--backend", "sqlite", "await", job_id,
-                 "--timeout-seconds", "1", *extra]
+                 "--timeout-seconds", "10", *extra]
             )
         return rc, out.getvalue()
 
diff --git a/tests/test_flow_fork.py b/tests/test_flow_fork.py
new file mode 100644
index 0000000..71f5e76
--- /dev/null
+++ b/tests/test_flow_fork.py
@@ -0,0 +1,238 @@
+"""Map builders fork one shared session; judges stay fresh.
+
+Codex R169/R179 (PM 1.45.1, N16 grouped, Sol 6.1 high): every cold group
+builder started a fresh session and read the same contracts, rubric, world
+API and checks again (178-261 s for 4 regions; native built all 16 in about
+350 s). Native multiworker builders start from the pilot's context. A map
+``fork: {"from": <agent>}`` gives PM builders the same start: one session
+reads the shared inputs once, and each item's code builder forks it. Judges
+never fork, so a review never shares a builder's context.
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+_HERMETIC_DIR = os.path.dirname(os.path.abspath(__file__))
+if _HERMETIC_DIR not in sys.path:
+    sys.path.insert(0, _HERMETIC_DIR)
+import hermetic_env  # noqa: F401  # process-wide host-env isolation
+
+import tempfile
+import threading
+import unittest
+from pathlib import Path
+from tempfile import TemporaryDirectory
+from unittest.mock import patch
+
+from puppetmaster import flow
+from puppetmaster.adapters import CodexAdapter
+from puppetmaster.adapters import codex as codex_module
+from puppetmaster.adapters.codex import build_codex_resume_command
+from puppetmaster.flow import JobNodeExecutor, NodeOutcome
+from puppetmaster.models import Task
+from puppetmaster.session_lease import acquire_codex_thread
+from puppetmaster.worker_resume import claim_resumed_session, resolve_worker_resume
+
+from test_worker_resume import FakeStore, _codex_events, _run_adapter, _task as _store_task
+
+_HOMES = tempfile.TemporaryDirectory()
+_ENV = patch.dict("os.environ", {
+    "CODEX_HOME": str(Path(_HOMES.name) / "codex"),
+    "CLAUDE_CONFIG_DIR": str(Path(_HOMES.name) / "claude"),
+    "PUPPETMASTER_HOME": str(Path(_HOMES.name) / "pm"),
+})
+
+
+def setUpModule() -> None:
+    _ENV.start()
+
+
+def tearDownModule() -> None:
+    _ENV.stop()
+    _HOMES.cleanup()
+
+
+SOURCE = "0199a1b2-c3d4-7e5f-8a9b-0c1d2e3f4a5b"
+FORKED = "0199a1b2-ffff-7e5f-8a9b-0c1d2e3f4a5b"
+
+
+class ForkResolutionTests(unittest.TestCase):
+    def setUp(self):
+        self.store = FakeStore()
+        self.store.add("job-prime", _store_task("t-prime", "prime", "codex", job_id="job-prime"),
+                       {"adapter": "codex", "ephemeral": False, "thread_id": SOURCE})
+
+    def test_fork_from_resolves_like_resume_from_and_says_fork(self):
+        record = resolve_worker_resume(self.store, {"fork_from": {"job_id": "job-prime", "task_id": "t-prime"}},
+                                       "codex")
+        self.assertEqual(record, {"status": "resolved", "adapter": "codex", "session_id": SOURCE,
+                                  "from_job_id": "job-prime", "from_task_id": "t-prime", "mode": "fork"})
+        missing = resolve_worker_resume(self.store, {"fork_from": {"job_id": "nope"}}, "codex")
+        self.assertEqual((missing["status"], missing["mode"]), ("unavailable", "fork"))
+
+    def test_any_number_of_tasks_can_fork_one_codex_thread(self):
+        record = {"status": "resolved", "adapter": "codex", "session_id": SOURCE, "mode": "fork"}
+        claimed: dict = {}
+        self.assertEqual(claim_resumed_session(record, claimed, "group_00"), record)
+        self.assertEqual(claim_resumed_session(record, claimed, "group_01"), record)
+        self.assertEqual(claimed, {})
+
+    def test_resume_from_wins_over_fork_from(self):
+        record = resolve_worker_resume(self.store, {"resume_from": {"job_id": "job-prime", "task_id": "t-prime"},
+                                                    "fork_from": {"job_id": "job-prime"}}, "codex")
+        self.assertNotIn("mode", record)
+
+
+class CodexForkTests(unittest.TestCase):
+    def task(self, **payload) -> Task:
+        return Task(job_id="job-b", role="build", instruction="Build regions/a.py.", adapter="codex",
+                    payload={"cwd": str(Path.cwd()), "model": "gpt-6.1-sol", "write_scope": ["regions/a.py"],
+                             **payload})
+
+    def record(self):
+        return {"status": "resolved", "adapter": "codex", "session_id": SOURCE,
+                "from_job_id": "job-prime", "from_task_id": "t-prime", "mode": "fork"}
+
+    def test_fork_argv(self):
+        command = build_codex_resume_command(executable="codex", session_id=SOURCE, fork=True,
+                                             sandbox="workspace-write", model="gpt-6.1-sol")
+        self.assertEqual(command[:5], ["codex", "exec", "fork", SOURCE, "--json"])
+        self.assertIn('sandbox_mode="workspace-write"', command)
+        self.assertEqual(command[-2:], ["--", "-"])
+
+    def test_a_fork_needs_no_thread_lease_sends_the_build_contract_and_records_its_own_thread(self):
+        held = acquire_codex_thread(SOURCE)  # another job continues the source in place
+        self.assertIsNotNone(held)
+        stdout = _codex_events(
+            {"type": "thread.started", "thread_id": FORKED},
+            {"type": "turn.started"},
+            {"type": "item.completed", "item": {"type": "agent_message", "text": "VERDICT: PASS - built"}},
+            {"type": "turn.completed", "usage": {"input_tokens": 126420, "cached_input_tokens": 84096,
+                                                 "output_tokens": 83}},
+        )
+        try:
+            artifacts, kwargs, enrich, _ = _run_adapter(CodexAdapter(), self.task(resume=self.record()), stdout)
+        finally:
+            held.release()
+        command = kwargs["command"]
+        self.assertEqual(command[:5], ["/usr/bin/codex", "exec", "fork", SOURCE, "--json"])
+        prompt = kwargs.get("stdin_data") or ""
+        self.assertIn("Build mode", prompt)
+        self.assertIn("Build regions/a.py.", prompt)
+        enrich.assert_not_called()  # the source session already holds the shared context
+        receipt = artifacts[0].payload
+        self.assertEqual(receipt["resume"]["mode"], "fork")
+        self.assertEqual(receipt["thread_id"], FORKED)
+        self.assertFalse(receipt["ephemeral"])  # a later repair resumes the fork itself
+        # The SDK total counts the copied source turns: never the attempt's own.
+        self.assertEqual(receipt["sdk_usage_scope"], "session_cumulative")
+
+    def test_fork_usage_comes_from_the_new_threads_rollout(self):
+        seen = []
+
+        def rollout_path(home, thread):
+            seen.append(thread)
+            return Path(f"/rollouts/{thread}.jsonl")
+
+        stdout = _codex_events({"type": "thread.started", "thread_id": FORKED})
+        with patch.object(codex_module.codex_rollout, "rollout_path", side_effect=rollout_path), \
+                patch.object(codex_module.codex_rollout, "attempt_usage",
+                             side_effect=lambda path, base: {"path": path, "base": sorted(base)}):
+            found = codex_module._rollout_attempt_usage(Path("/home"), self.record(),
+                                                        frozenset({"source-turn"}), stdout)
+        self.assertEqual(seen, [FORKED])
+        self.assertEqual(found, {"path": Path(f"/rollouts/{FORKED}.jsonl"), "base": ["source-turn"]})
+
+
+ITEM = {"entry": "build",
+        "nodes": [{"id": "build", "kind": "agent", "task": "Build {{item.id}}.", "files": "{{item.files}}"},
+                  {"id": "review", "kind": "judge", "task": "Review {{item.id}}.", "files": "{{item.files}}"},
+                  {"id": "done", "kind": "end", "status": "pass"}],
+        "edges": [{"from": "build", "to": "review"}, {"from": "review", "to": "done", "when": "PASS"}]}
+
+
+def graph(fork=True):
+    groups = {"id": "groups", "kind": "map", "concurrency": 2, "graph": ITEM,
+              "items": [{"id": key, "files": [f"regions/{key}.py"]} for key in ("group_00", "group_01")]}
+    if fork:
+        groups["fork"] = {"from": "prime"}
+    return {"id": "fork-flow", "entry": "prime", "defaults": {"adapter": "codex"},
+            "nodes": [{"id": "prime", "kind": "agent", "role": "explore", "task": "Read the shared inputs."},
+                      groups, {"id": "done", "kind": "end", "status": "pass"}],
+            "edges": [{"from": "prime", "to": "groups"}, {"from": "groups", "to": "done", "when": "ok"}]}
+
+
+class FlowForkTests(unittest.TestCase):
+    def setUp(self):
+        self._tmp = TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.state, self.work = Path(self._tmp.name) / "state", Path(self._tmp.name) / "work"
+        self.state.mkdir()
+        (self.work / "regions").mkdir(parents=True)
+
+    def test_validation(self):
+        bad = graph()
+        bad["nodes"][1]["fork"] = {"from": "done"}
+        self.assertIn("map 'groups' fork from 'done' is not an agent node", flow.validate_graph(bad))
+        bad["nodes"][1]["fork"] = {"source": "prime"}
+        self.assertIn("map 'groups' fork must be an object with from: the agent node to fork",
+                      flow.validate_graph(bad))
+        self.assertEqual(flow.validate_graph(graph()), [])
+
+    def child(self, fork_from, sessions=None):
+        parent = {"run_id": "flow_parent00000", "node": "groups", "key": "group_00", "depth": 1,
+                  "peer_scopes": ["regions/group_01.py"], "fork_from": fork_from}
+        run = flow.new_run(self.state, {**ITEM, "defaults": {"adapter": "codex"}}, "",
+                           state={"item": {"id": "group_00", "files": ["regions/group_00.py"]}},
+                           parent=parent, cwd=str(self.work))
+        run.sessions.update(sessions or {})
+        return run
+
+    def spec(self, run, node_id, visit=1):
+        node = next(node for node in run.graph["nodes"] if node["id"] == node_id)
+        return JobNodeExecutor(self.state)._spec(node_id, node, run, run.input, visit)
+
+    def test_builders_fork_and_judges_stay_fresh(self):
+        run = self.child({"job_id": "job-prime", "task_id": "t-prime"})
+        build = self.spec(run, "build")
+        self.assertEqual(build.payload["fork_from"], {"job_id": "job-prime", "task_id": "t-prime"})
+        self.assertNotIn("resume_from", build.payload)
+        self.assertIn("Build group_00.", build.instruction)  # the full task, not a delta
+        review = self.spec(run, "review")
+        self.assertNotIn("fork_from", review.payload)
+
+    def test_a_repair_visit_resumes_the_builders_own_session(self):
+        own = {"build": {"job_id": "job-b", "task_id": "t-b", "run_id": "flow_x", "visit": 1}}
+        build = self.spec(self.child({"job_id": "job-prime", "task_id": "t-prime"}, own), "build", visit=2)
+        self.assertEqual(build.payload["resume_from"], {"job_id": "job-b", "task_id": "t-b"})
+        self.assertNotIn("fork_from", build.payload)
+
+    def test_no_source_session_means_no_fork(self):
+        self.assertNotIn("fork_from", self.spec(self.child(None), "build").payload)
+
+    def test_the_map_hands_its_children_the_source_session(self):
+        def execute(node, run, prev):
+            if node["kind"] == "agent" and node.get("role") != "explore":
+                return NodeOutcome(ok=True, verdict="PASS", output="built", files=run.state["item"]["files"])
+            return NodeOutcome(ok=True, verdict="PASS", output="ok")
+
+        def spawn(state_dir, run_id):
+            thread = threading.Thread(target=flow.walk, args=(state_dir, run_id), kwargs={"execute": execute})
+            thread.start()
+            self.addCleanup(thread.join, 10)
+
+        executor = JobNodeExecutor(self.state, spawn=spawn, poll_seconds=0.02)
+        route = lambda node, run, prev: executor(node, run, prev) if node["kind"] == "map" else execute(node, run, prev)
+        run = flow.new_run(self.state, graph(), "", cwd=str(self.work))
+        run.sessions["prime"] = {"job_id": "job-prime", "task_id": "t-prime", "adapter": "codex"}
+        flow.save_run(self.state, run)
+        run = flow.walk(self.state, run.run_id, execute=route)
+        self.assertEqual(run.status, "done", run.reason)
+        for item in run.results["groups"]["items"].values():
+            child = flow.load_run(self.state, item["run_id"])
+            self.assertEqual(child.parent["fork_from"], {"job_id": "job-prime", "task_id": "t-prime"})
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_readonly_windows_sequence.py b/tests/test_readonly_windows_sequence.py
index ac42d68..ccf7e60 100644
--- a/tests/test_readonly_windows_sequence.py
+++ b/tests/test_readonly_windows_sequence.py
@@ -58,8 +58,62 @@ def test_unchanged_native_metadata_takes_one_sample(self):
         self.assertEqual(calls, 1)
 
     def test_native_metadata_retry_is_bounded(self):
-        with self.assertRaisesRegex(OSError, 'metadata did not stabilize'):
+        with self.assertRaisesRegex(OSError, 'metadata did not stabilize') as raised:
             self.native_stamp([(10, 20), (10, 21)] * 8, [(100, 10)] * 8)
+        # A writer moved the metadata on every sample: contention, not a new file.
+        self.assertTrue(readonly._source_open_contention(raised.exception))
+
+    def test_attach_retries_metadata_that_did_not_stabilize(self):
+        # Seen on a starved Windows runner: the parent's queued-source stamp
+        # raised this OSError, and attach failed outside every retry path.
+        with TemporaryDirectory() as tmp:
+            SQLiteSwarmStore(tmp).ensure_schema()
+            store = SQLiteSwarmStore(tmp)
+            original = readonly._source_stamp
+            calls = []
+
+            def unstable_once(target):
+                calls.append(target)
+                if len(calls) == 1:
+                    with patch.object(worker.os, 'name', 'nt'):
+                        self.native_stamp([(10, 20), (10, 21)] * 8, [(100, 10)] * 8)
+                return original(target)
+            with patch.object(readonly, '_source_stamp', unstable_once):
+                store.attach()
+            self.assertTrue(store._attached)
+            self.assertGreater(len(calls), 1)
+
+    def test_contention_after_open_ends_the_helper_session(self):
+        # Only an open can retry in one helper. A statement that meets moving
+        # metadata reports it, and the helper exits before the next request.
+        import contextlib
+        with TemporaryDirectory() as tmp:
+            SQLiteSwarmStore(tmp).ensure_schema()
+            path = Path(tmp) / 'state.sqlite3'
+            original = worker.source_stamp
+            descriptor_calls = []
+
+            def stamp(target=None, *, fd=None):
+                if fd is not None:
+                    descriptor_calls.append(fd)
+                    if len(descriptor_calls) > 1:
+                        error = OSError('source metadata did not stabilize')
+                        error.source_open_contention = True
+                        raise error
+                return original(target, fd=fd)
+            requests = io.StringIO(
+                json.dumps(dict(open=str(path))) + '\n' +
+                json.dumps(dict(sql='SELECT 1', parameters=[])) + '\n' +
+                json.dumps(dict(control='release', value=True)) + '\n')
+            out = io.StringIO()
+            with patch.object(worker, 'source_stamp', stamp), patch.object(sys, 'stdin', requests), \
+                    contextlib.redirect_stdout(out):
+                worker.serve()
+            replies = [json.loads(line) for line in out.getvalue().splitlines()]
+            self.assertEqual(replies[0], dict(ready=True))
+            self.assertIn('journal', replies[1])
+            self.assertIn('metadata did not stabilize', replies[2]['error'])
+            self.assertEqual(len(replies), 3)
 
     def test_numeric_lock_codes_are_authoritative(self):
         for code in (5, 6, 261, 262, 517):