From f6ed20cc41fff23621227db7b049783ab324808e Mon Sep 17 00:00:00 2001 From: Cary Palmer Date: Mon, 19 Oct 2026 16:00:28 -0400 Subject: [PATCH 1/3] tests: a group reuses on a follow-up when its builders are an inner map The regions treatment puts one builder per region in an inner map inside each group. On a follow-up, unnamed groups are reused, and inside a named group only the regions that the request names build again. The group review still covers the whole group. --- tests/test_flow_reuse.py | 67 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/tests/test_flow_reuse.py b/tests/test_flow_reuse.py index 3d8ea5c..f864f20 100644 --- a/tests/test_flow_reuse.py +++ b/tests/test_flow_reuse.py @@ -332,5 +332,72 @@ def test_validation(self): flow.validate_graph(bad)) +def nested_graph(): + """One builder per region inside each group: the group's build is an inner map.""" + shaped = graph() + item_graph = shaped["nodes"][0]["graph"] + item_graph["nodes"][0] = { + "id": "regions", "kind": "map", "items": "{{item.regions}}", "concurrency": 4, + "graph": {"entry": "build", + "nodes": [{"id": "build", "kind": "agent", "task": "Build {{item}}.", + "files": ["regions/{{item}}.py"]}, + {"id": "done", "kind": "end", "status": "pass"}], + "edges": [{"from": "build", "to": "done"}]}} + item_graph["entry"] = "regions" + for edge in item_graph["edges"]: + for end in ("from", "to"): + if edge[end] == "build": + edge[end] = "regions" + return shaped + + +class NestedReuseTests(unittest.TestCase): + """The regions treatment: a group is reused although its builders sit in an inner map.""" + + setUp, write, revise, built = ReuseTests.setUp, ReuseTests.write, ReuseTests.revise, ReuseTests.built + + def execute(self, node, run, prev): + if node["kind"] != "agent": + return ReuseTests.execute(self, node, run, prev) + key = run.state.get("key") + with self.lock: + self.calls.append((key, node["id"])) + return NodeOutcome(ok=True, verdict="PASS", output="built", files=[f"regions/{key}.py"], + usage={"input_tokens": 100}) + + def executor(self): + route = {} + + def spawn(state_dir, run_id): + thread = threading.Thread(target=flow.walk, args=(state_dir, run_id), kwargs={"execute": route["map"]}) + thread.start() + self.addCleanup(thread.join, 10) + + executor = JobNodeExecutor(self.state, spawn=spawn, poll_seconds=0.02) + route["map"] = lambda node, run, prev: (executor(node, run, prev) if node["kind"] == "map" + else self.execute(node, run, prev)) + return route["map"] + + def nested_run(self, text="", prior=None): + self.calls.clear() + run = flow.new_run(self.state, nested_graph(), text, cwd=str(self.work), + continue_from=prior.run_id if prior else None) + return flow.walk(self.state, run.run_id, execute=self.executor()) + + def test_unchanged_groups_reuse_and_named_regions_rebuild(self): + cold = self.nested_run() + self.assertEqual(cold.status, "done", cold.reason) + self.revise("region_001", "region_007") + run = self.nested_run(FOLLOW_UP, prior=cold) + self.assertEqual(run.status, "done", run.reason) + found = run.results["groups"]["items"] + self.assertEqual({key: found[key]["reuse"]["reused"] for key in GROUPS}, + {"group_00": True, "group_01": False, "group_02": True, "group_03": False}) + # Inside a named group, only the regions the request names build again. + self.assertEqual(self.built(), ["region_001", "region_007"]) + self.assertEqual(sorted({key for key, node in self.calls if node == "review"}), ["group_01", "group_03"]) + self.assertEqual(found["group_01"]["files"], ["regions/region_001.py", "regions/region_005.py"]) + + if __name__ == "__main__": unittest.main() From 61db51b3fe63254c558fc8cd24bcdff53ba3170c Mon Sep 17 00:00:00 2001 From: Cary Palmer Date: Mon, 19 Oct 2026 20:43:24 -0400 Subject: [PATCH 2/3] flow: a nested builder's revise no longer blocks follow-up reuse Codex R196 (1.46.1, regions treatment): groups that the follow-up did not name ran again, because _reads_request kept the inner builder's revise in the searched text when the entry node is a map. {{prev}} now counts only in nodes that get the request as prev: the entry, the branches of an entry parallel, and the same nodes inside an entry map. An entry parallel branch that reads {{prev}} now counts too. Release 1.46.2. --- docs/CHANGELOG.md | 19 ++++++++++++++++++ docs/FLOWS.md | 6 ++++-- puppetmaster/__init__.py | 2 +- puppetmaster/flow.py | 42 +++++++++++++++++++++++++++++++--------- pyproject.toml | 2 +- tests/test_flow_reuse.py | 20 +++++++++++++++++++ 6 files changed, 78 insertions(+), 13 deletions(-) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 568f391..414e660 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -1,3 +1,22 @@ +## v1.46.2 — 2026-10-19 + +**A nested builder's `revise` no longer blocks follow-up reuse.** Codex R196 +(PM 1.46.1, one builder per region inside each group, map `reuse` on): the +follow-up named regions in groups 01 and 03 only. Groups 00 and 02 did not +reuse ("its item graph reads the request"), so 8 extra region builders ran. + +- `_reads_request` removed `revise` only from the entry node itself. When the + entry is a map, the inner builder's `revise` (with `{{prev}}`) stayed in + the searched text, and the item counted as one that reads the request. +- `{{prev}}` now counts only in a node that gets the request as its `prev`: + the entry node, the branches of an entry `parallel`, and the same nodes + inside an entry `map`. Later nodes in a nested graph read their own + previous node, and their `{{prev}}` no longer counts. +- An entry `parallel` whose branch task reads `{{prev}}` now counts as + reading the request. Before, the branch was not searched and the item + could be reused against a new request. +- `{{input}}` anywhere still counts. + ## v1.46.1 — 2026-10-19 **A map inside a map item knows the outer items' writers.** R179 replica, diff --git a/docs/FLOWS.md b/docs/FLOWS.md index e1de91e..ea004fa 100644 --- a/docs/FLOWS.md +++ b/docs/FLOWS.md @@ -366,8 +366,10 @@ of its own files. On the first map visit of a follow-up run: unchanged, it has at least one review, and each review is a PASS whose files are unchanged. Otherwise it runs. An item without a recorded fingerprint runs. -4. An item never reuses when its item graph reads the request (`{{input}}`, - or `{{prev}}` in its entry node). An item never reuses when a request +4. An item never reuses when its item graph reads the request: `{{input}}` + anywhere, or `{{prev}}` outside `revise` in a node that gets the request + (the entry node, the branches of an entry `parallel`, and the same nodes + inside an entry `map`). An item never reuses when a request cannot name it: a one-character or all-digit key with no `names` alias and no files. diff --git a/puppetmaster/__init__.py b/puppetmaster/__init__.py index aa8d2e9..0d71a4a 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.46.1" +__version__ = "1.46.2" diff --git a/puppetmaster/flow.py b/puppetmaster/flow.py index cac0dea..33b5234 100644 --- a/puppetmaster/flow.py +++ b/puppetmaster/flow.py @@ -2641,17 +2641,41 @@ def _task_fingerprint(child_graph: dict, item: Any, index: int) -> str: def _reads_request(child_graph: dict) -> bool: """An item graph whose templates read the request: its result follows each new request. - ``{{input}}`` anywhere, or ``{{prev}}`` in the entry node's task, puts the - request into the item's own task. ``revise`` does not count: it is the - message to a resumed session (a repair, or a follow-up where its ``prev`` - carries the request), and every builder that runs on a follow-up gets the - request in some form. Which items the request concerns is the naming rule. + ``{{input}}`` anywhere, or ``{{prev}}`` in a node that gets the request as + its ``prev``, puts the request into the item's own task. ``revise`` does not + count: it is the message to a resumed session (a repair, or a follow-up + where its ``prev`` carries the request), and every builder that runs on a + follow-up gets the request in some form. Which items the request concerns + is the naming rule. """ - entry = next((node for node in child_graph.get("nodes") or [] if node.get("id") == child_graph.get("entry")), - {}) - first = {name: value for name, value in entry.items() if name != "revise"} return bool(_READS_REQUEST.search(json.dumps(child_graph.get("nodes"), default=str)) - or re.search(r"\{\{\s*prev\b", json.dumps(first, default=str))) + or _entry_reads_prev(child_graph)) + + +_READS_PREV = re.compile(r"\{\{\s*prev\b") + + +def _entry_reads_prev(graph: Any) -> bool: + """``{{prev}}`` outside ``revise`` in a node that gets the request as ``prev``. + + That is the entry node, the branches of an entry ``parallel``, and the same + nodes inside an entry ``map`` (its items get the request as their input on + a first visit). Later nodes inside a nested graph read their own previous + node, so the nested graph is not searched as one text. + """ + if not isinstance(graph, dict): + return False + nodes = {node.get("id"): node for node in graph.get("nodes") or [] if isinstance(node, dict)} + entry = nodes.get(graph.get("entry")) or {} + own = {name: value for name, value in entry.items() if name not in ("revise", "graph", "node")} + if _READS_PREV.search(json.dumps(own, default=str)): + return True + if entry.get("kind") == "parallel": + branches = [nodes.get(branch) or {} for branch in entry.get("branches") or []] + return any(_entry_reads_prev({"entry": branch.get("id"), "nodes": [branch]}) for branch in branches) + if entry.get("kind") == "map": + return _entry_reads_prev(_child_graph_shape(entry)) + return False def _missing_files(cwd: str, scope: list[str]) -> list[str]: diff --git a/pyproject.toml b/pyproject.toml index 6850866..bc4b397 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "puppetmaster-ai" -version = "1.46.1" +version = "1.46.2" 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_reuse.py b/tests/test_flow_reuse.py index f864f20..fba09df 100644 --- a/tests/test_flow_reuse.py +++ b/tests/test_flow_reuse.py @@ -296,6 +296,25 @@ def shape(**fields): self.assertTrue(flow._reads_request(shape(task="Do {{prev}}"))) self.assertTrue(flow._reads_request(shape(revise="Fix {{input}}"))) + def test_prev_inside_an_entry_map_or_parallel(self): + def entry_map(**fields): + inner = {"entry": "build", "edges": [{"from": "build", "to": "check"}], + "nodes": [{"id": "build", "kind": "agent", "task": "t", **fields}, + {"id": "check", "kind": "judge", "task": "Review {{prev}}"}]} + return {"entry": "regions", "edges": [], + "nodes": [{"id": "regions", "kind": "map", "items": "{{item.regions}}", "graph": inner}]} + # R196: a nested builder's revise made every group "read the request". + self.assertFalse(flow._reads_request(entry_map(revise="Fix: {{prev}}"))) + self.assertTrue(flow._reads_request(entry_map(task="Do {{prev}}"))) + self.assertTrue(flow._reads_request(entry_map(revise="Fix {{input}}"))) + both = {"entry": "both", "edges": [], + "nodes": [{"id": "both", "kind": "parallel", "branches": ["a", "b"]}, + {"id": "a", "kind": "agent", "task": "t", "revise": "Fix: {{prev}}"}, + {"id": "b", "kind": "agent", "task": "Do {{prev}}"}]} + self.assertTrue(flow._reads_request(both)) + both["nodes"][2]["task"] = "t" + self.assertFalse(flow._reads_request(both)) + def test_an_item_a_request_cannot_name_runs(self): decisions = flow._reuse_decisions( {"id": "m", "kind": "map", "reuse": {"inputs": ["rubric.json"]}, @@ -340,6 +359,7 @@ def nested_graph(): "id": "regions", "kind": "map", "items": "{{item.regions}}", "concurrency": 4, "graph": {"entry": "build", "nodes": [{"id": "build", "kind": "agent", "task": "Build {{item}}.", + "revise": "Apply this to {{item}}: {{prev}}", "files": ["regions/{{item}}.py"]}, {"id": "done", "kind": "end", "status": "pass"}], "edges": [{"from": "build", "to": "done"}]}} From 81ab5f7dd8dbf9b02a3a98ec7a7e8afd98933af3 Mon Sep 17 00:00:00 2001 From: Cary Palmer Date: Mon, 19 Oct 2026 20:57:47 -0400 Subject: [PATCH 3/3] file_claims: cache the worktree root, so renewals stop starting git Each claim renewal (every ttl/3, about 0.7 s) ran two git processes for the repository identity. The second (--git-common-dir) was not used. A 4-worker probe counted 57-62 subprocesses per worker, growing with build time; after this change it is 24 and fixed. --- docs/CHANGELOG.md | 10 ++++++++++ puppetmaster/file_claims.py | 20 ++++++++++++-------- tests/test_file_claims.py | 18 ++++++++++++++++++ 3 files changed, 40 insertions(+), 8 deletions(-) diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 414e660..84cd057 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -17,6 +17,16 @@ reuse ("its item graph reads the request"), so 8 extra region builders ran. could be reused against a new request. - `{{input}}` anywhere still counts. +**A worker no longer starts git about three times a second while it builds.** +The edit-claim renewal runs each `ttl/3` (about 0.7 s), and each renewal +asked git twice for the repository identity. A 4-worker probe counted 57-62 +subprocesses per worker, and the count grew with build time. + +- The worktree root is cached for each directory in a process + (`file_claims._identity_of`). The unused `--git-common-dir` call is gone. +- Probe after the fix: 24 subprocesses per worker. The count no longer grows + with build time. Claim identity and claim behavior do not change. + ## v1.46.1 — 2026-10-19 **A map inside a map item knows the outer items' writers.** R179 replica, diff --git a/puppetmaster/file_claims.py b/puppetmaster/file_claims.py index 0ff253f..282b2c8 100644 --- a/puppetmaster/file_claims.py +++ b/puppetmaster/file_claims.py @@ -7,6 +7,7 @@ from contextlib import contextmanager from dataclasses import dataclass +import functools import math import os from pathlib import Path @@ -220,7 +221,7 @@ def live_writers(self, repo: PathLike, paths: Iterable[PathLike]) -> list[tuple[ return [(str(row["path"]), str(row["owner"])) for row in rows] def repository_identity(self, repo: PathLike) -> tuple[str, Path]: - """Return canonical Git common-dir identity and this worktree's root.""" + """Return this worktree's root as its identity, and the root.""" return _repository_identity(repo) def _claim_keys(self, repo: PathLike, paths: Iterable[PathLike]) -> tuple[str, list[str]]: @@ -370,21 +371,24 @@ def _git(directory: Path, *args: str) -> str: def _repository_identity(repo: PathLike) -> tuple[str, Path]: - candidate = Path(repo).expanduser().resolve() + return _identity_of(str(Path(repo).expanduser().resolve())) + + +@functools.lru_cache(maxsize=256) +def _identity_of(candidate_text: str) -> tuple[str, Path]: + # A worker renews its claims each ttl/3 (about 0.7 s), and each renewal + # asks for this identity. A worktree root does not move while a process + # runs, so one git call per directory is enough. + candidate = Path(candidate_text) try: root_text = _git(candidate, "rev-parse", "--show-toplevel") - common_text = _git(candidate, "rev-parse", "--git-common-dir") except (OSError, subprocess.CalledProcessError) as exc: # Non-Git workspaces still need a stable, canonical admission scope. if not candidate.exists() or not candidate.is_dir(): raise ValueError("repo must be an existing workspace directory") from exc return str(candidate), candidate + # The physical worktree is the isolation boundary, not the common Git dir. root = Path(root_text).resolve() - common = Path(common_text) - if not common.is_absolute(): - common = candidate / common - # The physical worktree is the isolation boundary. The common Git - # directory is shared metadata, not a shared write target. return str(root), root diff --git a/tests/test_file_claims.py b/tests/test_file_claims.py index e54295d..4685361 100644 --- a/tests/test_file_claims.py +++ b/tests/test_file_claims.py @@ -105,6 +105,24 @@ def test_repo_subdirectory_resolves_git_common_dir_from_candidate(self) -> None: self.assertEqual(self.registry.repository_identity(self.root), self.registry.repository_identity(subdirectory)) + def test_renewals_ask_git_for_the_root_once(self) -> None: + # A worker renews each ttl/3; two git processes per renewal was about + # three a second per worker for the whole build. + claim = self.registry.acquire(self.root, "file.txt", "worker", 60) + real = subprocess.check_output + calls = [] + + def counted(*args, **kwargs): + calls.append(args[0]) + return real(*args, **kwargs) + + from unittest import mock + with mock.patch("puppetmaster.file_claims.subprocess.check_output", counted): + for _ in range(5): + self.assertTrue(self.registry.renew(self.root, "file.txt", claim.claim_id, 60)) + self.registry.repository_identity(self.root) + self.assertEqual([], calls) + def test_root_and_git_internal_claims(self) -> None: root_claim = self.registry.acquire(self.root, ".", "worker-a", 60) with self.assertRaises(FileClaimConflict):