P241: integrate main with shared native load and growth owners

docs/p241-native-measurements.md

@@ -0,0 +1,59 @@
+# P241: shared commander load and growth
+
+Owner: pi-cn-phase2. PR #1965. Rebased onto requested main `15ba6edce`.
+
+## Ownership
+
+- `commander/capacity.py:Load` owns native and CLI load, the weighting formula,
+  and selection of receiving child commanders with retained orders. The CLI
+  uses its supplied commander context; it does not construct a competing one.
+- `turn_digests.py:OrderRead.for_commander` owns the shared order/holder selection
+  and incarnation check. Both full edit reads and count-only reads consume it.
+  `OrderEditCountRead.count` serves load and the unjudged standing rule. The
+  competing `CommsUnjudgedTool.readers/count` methods are deleted.
+- The heartbeat reader/writer, its CLI write and its config-directory creation
+  are deleted. Registry parentage/status and held orders supply that fact.
+- `commander/standing.py:MergeGrowth` reads one captured first-parent Git history
+  with numstat, not a fetch followed by two subprocesses per revision.
+  `repo.py:Repo.git` owns bounded checked execution and remote discovery.
+- Main's `CommanderContext`, argument parser, order lifecycle and recorded-check
+  progress implementations are preserved. `ToolCaller.acting_name` selects the
+  explicitly bound caller or the context's configured commander. The CLI uses
+  the existing concrete caller states and FieldCodec normalization, not a nullable
+  name or a second context. Checked order reads obtain their cutoff from the order.
+
+Before-rebase AST evidence uses the existing audit Package/Repository parser
+across Core, tools and tests: `p241-2200-ast-before.txt` in this unit's scratch.
+Receiver bindings are syntactic; dynamic resolution and state freshness are not
+proved by the census. The previous retained source evidence is also preserved.
+
+## Verification and limitations
+
+The 20:52 checkpoint was broken: a CLI FieldCodec reference lacked its import.
+Its retained tests reported 11 passed, one failure and 36 errors. Both ratchets
+exited 2 because the command omitted the required `--head`; those were not
+successful checks. The current rebase removes that obsolete CLI identity change.
+
+Current checks use the shared interpreter, background `timeout 300` commands,
+and the unchanged checker against the requested main. Results and application
+receipts are retained under `/home/ts/.cache/agent-scratch/pi-cn-phase2/` with
+prefix `p241-2200-`. No installation or provider input is performed here.
+
+Final source checkpoint `d00ab51d7`: **49 tests passed** under background
+`timeout 300`; source ratchet **1 solely SourceLines +8**, with no introduced
+findings; tools ratchet **0**. Both ran separately against `15ba6edce`, with
+`--head` supplied. Before/after AST acquisition reported no parse omissions.
+
+Actual configured saved-state native handler read **32 open / 30 overdue /
+1,860 unjudged / total 278.0**. The real CLI then read **32 / 30 / 1,861**;
+the count increased by one between reads, so this is not a frozen parity sample. The
+focused test checks exact native/CLI equality on one retained fixture. This is
+not an installed native-transport or provider-process verification.
+
+The 22:10 readiness deadline was missed: the final fork argument representation
+needed correction. No failed or running check was represented as successful.
+
+This delivers the shared load/growth checkpoint. Broader timing/alert parity and
+P244's comparisons are not claimed complete. Earlier native/CLI samples are
+historical evidence only, not proof of the rebased artifact. No blocked goal is
+reset, and no uncertain operation is replayed.

src/agent_comms/commander/capacity.py

@@ -1,43 +1,37 @@
-"""Report commander load and live forks; no command refuses on capacity."""
+"""One commander load report for native tools and the commander's CLI."""
 from __future__ import annotations
 
-import os, time
+import time
 from dataclasses import dataclass
-from pathlib import Path
 
 from ..comms import Comms
-
-BEAT_AGE = 10 * 60
-
-
-def live_forks(forks: Path) -> list[str]:
-    os.makedirs(forks, exist_ok=True)
-    return [n[:-5] for n in os.listdir(forks) if n.endswith('.beat') and time.time() - os.path.getmtime(f'{forks}/{n}') < BEAT_AGE]
-
-
-def beat(forks: Path, name: str) -> None:
-    os.makedirs(forks, exist_ok=True)
-    with open(f'{forks}/{name}.beat', 'a'):
-        os.utime(f'{forks}/{name}.beat')
+from ..field_codec import projected
+from ..tool_result import ToolResult
+from ..turn_digests import OrderEditCountRead
 
 
 @dataclass(frozen=True)
-class Load:
-    """The commander load on one Comms: its open and overdue orders, its unjudged edits, live forks."""
+class Load(ToolResult):
+    """Order and edit debt, with receiving children that themselves command orders."""
     owed: int
     overdue: int
     unjudged: int
     forks: tuple[str, ...]
 
     @classmethod
-    def of(cls, comms: Comms, commander: str, forks: Path) -> Load:
-        from ..tools import CommsUnjudgedTool
-        actor, now = comms.registry.require(commander).incarnation, time.time()
-        owed = [t for t in comms.goals.todos.held_for_commander(actor) if t.pending]
+    def of(cls, comms: Comms, commander: str) -> Load:
+        snapshot, todos, now = comms.registry.snapshot(), comms.goals.todos, time.time()
+        owner = snapshot.require(commander)
+        owed = tuple(t for t in todos.held_for_commander(owner.incarnation) if t.pending)
+        forks = tuple(sorted(
+            thread.name for thread in snapshot.threads.values()
+            if thread.parent == owner.name and snapshot.statuses[thread.name].receiving
+            and todos.held_for_commander(thread.incarnation)
+        ))
         return cls(len(owed), sum(t.overdue(now) for t in owed),
-                   CommsUnjudgedTool.count(comms, commander), tuple(live_forks(forks)))
+                   OrderEditCountRead.count(comms, owner.name), forks)
 
-    @property
+    @projected(view="wire")
     def total(self) -> float:
         return self.owed + 2 * self.overdue + self.unjudged / 10
 
@@ -47,4 +41,4 @@ def __str__(self) -> str:
 
 
 def main(a, ctx):
-    print(Load.of(ctx.comms, ctx.commander, ctx.state / 'forks'))
+    print(Load.of(ctx.comms, ctx.commander))

src/agent_comms/commander/config.py

@@ -96,7 +96,7 @@ def write(self, home: Path) -> Path:
         record.write_text(self.model_dump_json(indent=1) + '\n')
         path.unlink(missing_ok=True)
         path.symlink_to(record)
-        for directory in (self.state, self.state / 'forks', self.state / 'orders', self.worktrees,
+        for directory in (self.state, self.state / 'orders', self.worktrees,
                           self.scratch, self.board.parent):
             directory.mkdir(parents=True, exist_ok=True)
         return path

src/agent_comms/commander/order.py

@@ -76,7 +76,7 @@
     VoidStrikeTodo,
     WithdrawTodo,
 )
-from ..tool_caller import ThreadCaller
+from ..tool_caller import NoThreadCaller, ThreadCaller
 from ..tools import (
     CommsChangeProblemTool,
     CommsFireTool,
@@ -89,7 +89,6 @@
     CommsTodoListTool,
     TodoList,
 )
-from . import capacity
 from .config import CommanderConfig, CommanderRole
 from .send_guard import CommanderSession
 
@@ -765,19 +764,19 @@ def fire(self, unit: str, why: str) -> None:
 @dataclass(frozen=True, kw_only=True)
 class CommanderVerb(CommanderRole, DeclaredFamily, DeclaredOptions, affix="Verb"):
     """One commander verb: its options are its fields, its parser comes from them, and run() does the work."""
-    fork: str | None = option("--fork", default=None, help="a forked commander's name; keeps its beat fresh")
+    fork: ThreadCaller | NoThreadCaller = option(
+        "--fork", type=ThreadCaller, normalize=FieldCodec.encode,
+        default=NoThreadCaller(), help="the acting commander's name")
 
     def who(self, ctx: Commander) -> str:
-        return self.fork or ctx.commander
+        return self.fork.acting_name(ctx.commander)
 
     def apply(self, ctx: Commander) -> None:
         self.gate(ctx)
         self.run(ctx)
 
     def gate(self, ctx: Commander) -> None:
         self.require_commander(ctx.comms.registry.require(self.who(ctx)))
-        if self.fork:
-            capacity.beat(ctx.state / 'forks', self.fork)
 
     @abstractmethod
     def run(self, ctx: Commander) -> None: ...

src/agent_comms/commander/standing.py

@@ -4,12 +4,10 @@
 
 import asyncio
 import logging
-import subprocess
 import time
 from collections.abc import Callable, Iterable, Iterator
 from contextlib import AbstractContextManager
 from dataclasses import dataclass
-from pathlib import Path
 from typing import ClassVar
 
 from ..comms import Comms
@@ -25,11 +23,8 @@
 )
 from ..thread_identity import StallWatcherIncarnation, ThreadIncarnation
 from ..threads import Thread
-from ..todos import (
-    FlatTrend, OrderProgress, PullRequests, Todo,
-)
-from ..tools import CommsUnjudgedTool
-from ..turn_digests import TurnDigestRead, TurnsSince
+from ..todos import FlatTrend, OrderProgress, PullRequests, Todo
+from ..turn_digests import OrderEditCountRead, TurnDigestRead, TurnsSince
 from .config import CommanderConfig
 
 
@@ -202,7 +197,7 @@ class Unjudged(ReasonedStandingRule, UndatedStandingRule):
 
     @classmethod
     def reason(cls, comms: Comms, commander: str) -> str | None:
-        total = CommsUnjudgedTool.count(comms, commander)
+        total = OrderEditCountRead.count(comms, commander)
         return str(total) if total > cls.limit else None
 
 
@@ -220,37 +215,33 @@ class MergeGrowth(DatedStandingRule):
     src: tuple[int, int]
     tests: tuple[int, int]
 
-    @staticmethod
-    def git(checkout: Path, *args: str) -> str:
-        return subprocess.run(
-            ["git", "-C", str(checkout), *args], capture_output=True, text=True
-        ).stdout
-
-    @classmethod
-    def numstat(cls, checkout: Path, sha: str, path: str) -> tuple[int, int]:
-        diff = cls.git(checkout, "diff", "--numstat", f"{sha}^1", sha, "--", path)
-        counted = [
-            (int(a), int(d))
-            for a, d, _ in (row.split("\t", 2) for row in diff.splitlines())
-            if a.isdigit()
-        ]
-        return sum(a for a, _ in counted), sum(d for _, d in counted)
-
     @classmethod
     def merges(cls, repo: Repo) -> Iterator[MergeGrowth]:
-        """Each first-parent commit on the repository's main inside the window, newest first."""
-        cls.git(repo.checkout, "fetch", "-q", "origin", "main")
-        log = cls.git(
-            repo.checkout,
-            "log",
-            "--first-parent",
-            f"--since={cls.window}",
-            "--format=%H %ct %s",
-            "origin/main",
+        """Read first-parent growth from one captured, already-fetched Git history.
+
+        Fetch belongs to the checkout's update owner, not a standing-rule read.
+        Git emits each commit's first-parent diff in this same process.
+        """
+        log = repo.git(
+            repo.checkout, "log", "--first-parent", f"--since={cls.window}",
+            "--format=%x1e%H%x00%ct%x00%s", "--numstat", "--no-renames",
+            "--diff-merges=first-parent", "origin/main", "--", repo.source, "tests",
         )
-        for sha, at, subject in (line.split(" ", 2) for line in log.splitlines()):
-            src, tests = (cls.numstat(repo.checkout, sha, path) for path in (repo.source, "tests"))
-            yield cls(repo.github, sha, subject, float(at), src, tests)
+        for record in log.split("\x1e")[1:]:
+            header, _, diff = record.partition("\n")
+            sha, at, subject = header.split("\x00", 2)
+            src, tests = [0, 0], [0, 0]
+            for row in diff.splitlines():
+                if not row:
+                    continue
+                added, deleted, path = row.split("\t", 2)
+                if not (added.isdigit() and deleted.isdigit()):
+                    continue
+                for root, counts in ((repo.source, src), ("tests", tests)):
+                    if path == root or path.startswith(root.rstrip("/") + "/"):
+                        counts[0] += int(added)
+                        counts[1] += int(deleted)
+            yield cls(repo.github, sha, subject, float(at), (src[0], src[1]), (tests[0], tests[1]))
 
     @classmethod
     def violations(cls, comms: Comms, commander: str) -> tuple[MergeGrowth, ...]:

src/agent_comms/repo.py

@@ -28,14 +28,19 @@ def contains(self, sha: str) -> bool:
             result.check_returncode()
         return result.returncode == 0
 
-    @classmethod
-    def of(cls, checkout: Path) -> Repo:
-        """Its GitHub name comes from the checkout's own origin remote."""
-        url = subprocess.run(
-            ["git", "-C", str(checkout), "remote", "get-url", "origin"],
+    @staticmethod
+    def git(checkout: Path, *args: str) -> str:
+        return subprocess.run(
+            ["git", "-C", str(checkout), *args],
             capture_output=True,
             text=True,
             check=True,
-        ).stdout.strip()
+            timeout=30,
+        ).stdout
+
+    @classmethod
+    def of(cls, checkout: Path) -> Repo:
+        """Its GitHub name comes from the checkout's own origin remote."""
+        url = cls.git(checkout, "remote", "get-url", "origin").strip()
         github = url.removesuffix(".git").replace(":", "/").split("/")[-2:]
         return cls(checkout=checkout.resolve(), github="/".join(github))

src/agent_comms/tool_caller.py

@@ -38,6 +38,9 @@ def listing(self, comms: Comms, threads: tuple[ThreadRecord, ...]) -> ThreadList
     @abstractmethod
     def catalog(self, comms: Comms, tools: tuple[dict[str, object], ...]) -> ToolCatalogResult: ...
 
+    @abstractmethod
+    def acting_name(self, default: str) -> str: ...
+
     @abstractmethod
     def subject(self, named: str) -> str: ...
 
@@ -67,6 +70,9 @@ def catalog(self, comms: Comms, tools: tuple[dict[str, object], ...]) -> ToolCat
     def refuse_thread(self) -> NoReturn:
         raise CallingThreadUnavailable("This tool requires a thread caller.")
 
+    def acting_name(self, default: str) -> str:
+        return default
+
     def subject(self, named: str) -> str:
         if not named:
             return self.refuse_thread()
@@ -96,6 +102,9 @@ def listing(self, comms: Comms, threads: tuple[ThreadRecord, ...]) -> ThreadList
     def catalog(self, comms: Comms, tools: tuple[dict[str, object], ...]) -> ToolCatalogResult:
         return ThreadToolCatalogResult(tools, comms.registry.require(self.name).service_tier)
 
+    def acting_name(self, default: str) -> str:
+        return self.name
+
     def subject(self, named: str) -> str:
         return named or self.name
 

src/agent_comms/tools.py

@@ -12,6 +12,7 @@
 
 from pydantic import JsonValue
 
+from .commander.capacity import Load
 from .progress_claims import ProgressClaim
 from .pi_vocabulary import ServiceTier, ThinkingLevel, ThinkingLevelField
 from .cli_commands import (
@@ -79,7 +80,7 @@
     TaskScopeSelection, OriginalTaskChange,
 )
 from .order_work import OrderWork
-from .turn_digests import LastTurns, OrderEditCountRead, OrderEditsRead, ThreadTurns, TurnsSince, TurnDigestRead, UnjudgedEdits
+from .turn_digests import LastTurns, OrderEditsRead, ThreadTurns, TurnsSince, TurnDigestRead, UnjudgedEdits
 from .relationships import RelationshipEdit
 from .work_branches import WorkBranch, WorkBranches
 from .tool_output import MAX_INLINE_OUTPUT_BYTES
@@ -761,25 +762,18 @@ class CommsUnjudgedTool(ThreadReadTool):
     label = "Unjudged edits"
     description = "Read original edit calls since each open order, excluding that order's recorded verdicts."
 
-    @classmethod
-    def readers(cls, comms: Comms, commander: str, read_type=OrderEditsRead):
-        actor = comms.registry.require(commander).incarnation
-        for todo in comms.goals.todos.held_for_commander(actor):
-            assignment = todo.held()
-            holder = comms.registry.require(assignment.owner)
-            if holder.incarnation != assignment.owner_identity:
-                raise ToolRefusedError(
-                    f"Order {todo.id} holder incarnation no longer exists"
-                )
-            yield read_type(comms.root, holder, TurnsSince(todo.order.at.recorded_at("An order")), todo)
+    def apply(self, comms: Comms, caller: ToolCaller) -> UnjudgedEdits:
+        return UnjudgedEdits(tuple(edit for read in OrderEditsRead.for_commander(
+            comms, caller.subject(self.thread)) for edit in read.read()))
 
-    @classmethod
-    def count(cls, comms: Comms, commander: str) -> int:
-        return sum(read.read() for read in cls.readers(comms, commander, OrderEditCountRead))
 
-    def apply(self, comms: Comms, caller: ToolCaller) -> UnjudgedEdits:
-        return UnjudgedEdits(tuple(edit for read in self.readers(comms, caller.subject(self.thread))
-                                  for edit in read.read()))
+@dataclass(frozen=True, kw_only=True)
+class CommsCommanderLoadTool(ThreadReadTool):
+    label = "Measured commander load"
+    description = "Read canonical order/edit debt and receiving child commanders; never self-reported."
+
+    def apply(self, comms: Comms, caller: ToolCaller) -> Load:
+        return Load.of(comms, caller.subject(self.thread))
 
 
 @dataclass(frozen=True, kw_only=True)