Release 1.46.2: nested revise reuse, no git polling in claim renewals

docs/CHANGELOG.md

@@ -1,3 +1,32 @@
+## 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.
+
+**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,

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.
 

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"

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
 
 

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]:

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"

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):

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"]},
@@ -332,5 +351,73 @@ 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}}.",
+                             "revise": "Apply this to {{item}}: {{prev}}",
+                             "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()