From 27d879a150ec060bda06965baaeb24167f4ea641 Mon Sep 17 00:00:00 2001 From: Cary Palmer Date: Sun, 18 Oct 2026 07:18:17 -0400 Subject: [PATCH 1/4] tests: two Windows CI tests wait like the clients they model test_concurrent_replay runs 8 workers in one process. On Windows the ChangeTime of state.sqlite3 can move alone under live writers. Attach then raises StoreMetadataDrift. The fence stays, because stat cannot tell this move from a rename away and back. A real worker exits 75 and the supervisor respawns it. Each test writer now respawns its attach the same way, up to _ATTACH_RESPAWN_LIMIT. The CLI await tests now wait 10 s, like the MCP await tests. The job is complete, so the read returns at once. With 1 s, a loaded Windows runner gave status unavailable before the first store read. 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. --- docs/CHANGELOG.md | 23 +++++++++++++++++++++++ tests/test_attempt_ledger.py | 17 ++++++++++++++--- tests/test_await_compact.py | 5 ++++- 3 files changed, 41 insertions(+), 4 deletions(-) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index a93b81e0..50dc9082 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -1,3 +1,26 @@ +## Unreleased + +**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/tests/test_attempt_ledger.py b/tests/test_attempt_ledger.py index 3d5a42ad..2a57cfa4 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 cfeebd05..95f358ef 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() From c8b503bfd5e01afdde58110c76a4e2b27c282a98 Mon Sep 17 00:00:00 2001 From: Cary Palmer Date: Sun, 18 Oct 2026 19:18:03 -0400 Subject: [PATCH 2/4] readonly: metadata that a live writer keeps in motion is contention On a starved Windows runner, source_stamp raised "source metadata did not stabilize" after 8 samples. The OSError had no contention mark, so it escaped every retry path. Attach raised it at once, the helper ended its session, and a worker exited 1. The supervisor did not respawn it. Mark the error with 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 exits 75 if the budget runs out. The helper retries contention only before its session opens. After the open, a contention error made the helper wait for a new open and crash with KeyError on the next statement or release. --- docs/CHANGELOG.md | 24 +++++++++++ puppetmaster/readonly_worker.py | 10 ++++- puppetmaster/worker_runtime.py | 9 ++-- tests/test_attach_metadata_drift.py | 4 ++ tests/test_readonly_windows_sequence.py | 56 ++++++++++++++++++++++++- 5 files changed, 96 insertions(+), 7 deletions(-) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 50dc9082..b3ce786c 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -1,5 +1,29 @@ ## Unreleased +**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. diff --git a/puppetmaster/readonly_worker.py b/puppetmaster/readonly_worker.py index 1582bf40..19b1ef8c 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_runtime.py b/puppetmaster/worker_runtime.py index 24b7e1fa..7c9f3605 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/tests/test_attach_metadata_drift.py b/tests/test_attach_metadata_drift.py index 8b83c38a..1535fa6d 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_readonly_windows_sequence.py b/tests/test_readonly_windows_sequence.py index ac42d68b..ccf7e60e 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): From b38ae3598ae7c6bc1c6dd7ca0e114b3728e090d1 Mon Sep 17 00:00:00 2001 From: Cary Palmer Date: Sun, 18 Oct 2026 23:35:06 -0400 Subject: [PATCH 3/4] flow: map builders can fork one shared session (1.46.0) A map takes fork: {"from": }. Each item's code builder forks that node's recorded session on its first visit; a repair resumes the builder's own fork; judges never fork. Codex runs codex exec fork with no thread lease, the full build contract and the write sandbox. The receipt records the fork's own thread, and attempt usage comes from its own rollout. --- docs/CHANGELOG.md | 28 +++- docs/FLOWS.md | 31 +++++ puppetmaster/__init__.py | 2 +- puppetmaster/adapters/codex.py | 28 +++- puppetmaster/flow.py | 23 +++- puppetmaster/mcp_server.py | 4 +- puppetmaster/worker_resume.py | 23 +++- pyproject.toml | 2 +- tests/test_flow_fork.py | 238 +++++++++++++++++++++++++++++++++ 9 files changed, 363 insertions(+), 16 deletions(-) create mode 100644 tests/test_flow_fork.py diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index b3ce786c..d05580b0 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -1,4 +1,30 @@ -## Unreleased +## 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": ""}`. 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": }`. +- `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 `. 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 diff --git a/docs/FLOWS.md b/docs/FLOWS.md index 1cea094a..1620054d 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 ec39e67f..8cfa8984 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 c04de246..e6941532 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 `` (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 7c56fa0a..a4f86638 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 f3419524..2e4cd4eb 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\": } 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/worker_resume.py b/puppetmaster/worker_resume.py index 8444c429..ce8debda 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/pyproject.toml b/pyproject.toml index ccb61bde..1a3f439b 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_flow_fork.py b/tests/test_flow_fork.py new file mode 100644 index 00000000..bcb59897 --- /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": }`` 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": str(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": 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() From fe94dd98249a782e29085737e95e118c381aa0c8 Mon Sep 17 00:00:00 2001 From: Cary Palmer Date: Sun, 18 Oct 2026 23:45:25 -0400 Subject: [PATCH 4/4] tests: compare the fork rollout path as a Path (Windows separators) --- tests/test_flow_fork.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_flow_fork.py b/tests/test_flow_fork.py index bcb59897..71f5e76a 100644 --- a/tests/test_flow_fork.py +++ b/tests/test_flow_fork.py @@ -138,11 +138,11 @@ def rollout_path(home, thread): 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": str(path), "base": sorted(base)}): + 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": f"/rollouts/{FORKED}.jsonl", "base": ["source-turn"]}) + self.assertEqual(found, {"path": Path(f"/rollouts/{FORKED}.jsonl"), "base": ["source-turn"]}) ITEM = {"entry": "build",