Integrate #2007 #2000 #2003 #1961 #1999 #2002

src/agent_comms/backend.py

@@ -54,7 +54,7 @@
 from .prompt_state import UnwrittenPrompt
 from .request_progress import NO_REQUEST_OBSERVER, RequestObserver, RequestProgress
 from .native_tool_call import NativeToolMode
-from .native_tool_transport import ToolTransport
+from .native_tool_transport import ToolTransport, UnboundToolTransport
 from .mro_dispatch import MroDispatch, handles
 from .selected_tool_broker import SelectedToolDenied
 from .send_outcome import SendBoundary
@@ -382,8 +382,8 @@ def context_info(self) -> events.ContextInfo:
             context_size=self.usage.size,
         )
 
-    async def open_transport(self, resources: AsyncExitStack) -> None:
-        self.tool_socket = await ToolTransport.open(self, resources)
+    async def open_transport(self, resources: AsyncExitStack) -> UnboundToolTransport:
+        return await ToolTransport.open(self, resources)
 
     @contextmanager
     def native_acquisition(self):
@@ -398,9 +398,9 @@ async def acquire_native(self, resources: AsyncExitStack, *, reuse: bool) -> PiS
         with self.native_acquisition():
             async with AsyncExitStack() as transport:
                 with self.startup.measurements.operation("open_transport"):
-                    await self.open_transport(transport)
+                    unbound = await self.open_transport(transport)
                 native = await self.open_native(reuse=reuse)
-                self.tool_socket.bind_process(native.proc.pid)
+                self.tool_socket = await unbound.bind_process(native.proc.pid)
                 native.transport = transport.pop_all()
         self.owner = asyncio.current_task()
         if self.owner is not None:
@@ -587,11 +587,12 @@ def __init__(self, launch, make_command: Callable[..., commands.Prompt], *,
         self.tool_mode = tool_mode
         self.make_command = make_command
 
-    async def open_transport(self, resources: AsyncExitStack) -> None:
-        self.tool_socket = await self.tool_mode.open_transport(self, resources)
-        self.command = self.make_command(owner_tools=self.tool_socket.connection)
+    async def open_transport(self, resources: AsyncExitStack) -> UnboundToolTransport:
+        unbound = await self.tool_mode.open_transport(self, resources)
+        self.command = self.make_command(owner_tools=unbound.listener.connection)
         del self.make_command
         self.output.sensitive = bool(self.command.images)
+        return unbound
 
     async def context_proof(self):
         if not self.started_input or not self.accepts_output:

src/agent_comms/channel_coding_tools.py

@@ -9,32 +9,30 @@
 
 import asyncio
 from abc import ABC, abstractmethod
-from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field, replace
 from pathlib import Path
 
 from agent_comms.coordinator import Coordination
 
-from .selected_claim import (
-    publish_selected_resource_claim,
-    verify_selected_wake,
-)
 from .comms import Comms
-from .field_codec import FieldCodec
 from .envelope_claim_transitions import (
     ClaimOwner,
-    FileClaimPath,
     WakeClaim,
 )
-from .native_tool_call import CodingCall, NativeToolMode, SelectedToolDenied
+from .field_codec import FieldCodec
 from .native_arguments import NativeArguments
-from .storage_room import NoRoomNeeded, RoomReservation
+from .native_tool_call import CodingCall, NativeToolMode, SelectedToolDenied
+from .native_tool_transport import ToolListener, ToolTransport, UnboundToolTransport
 from .native_tools import NativeToolDeclaration
-from .native_tool_transport import ToolTransport
+from .selected_claim import (
+    publish_selected_resource_claim,
+    verify_selected_wake,
+)
 from .selected_tool_broker import (
     consume_selected_slot,
     verify_sent_full_input,
 )
+from .storage_room import NoRoomNeeded, RoomReservation
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -131,14 +129,14 @@ def launch_arguments(self, package: Path) -> tuple[str, ...]:
         thread = self.owner.comms.registry.require(self.owner.owner_name)
         return thread.native_launch_arguments(NativeArguments.parse(arguments), package)
 
-    async def open_transport(self, turn, ownership) -> ToolTransport:
+    async def open_transport(self, turn, ownership) -> UnboundToolTransport:
         return await CodingToolSocket.open(turn, ownership, self.owner)
 
 class CodingToolSocket(ToolTransport[CodingCall]):
     """Multi-call native policy on the same peer-authenticated owner transport."""
 
-    def __init__(self, directory: Path, token: str, owner: CodingToolOwner) -> None:
-        super().__init__(directory, token)
+    def __init__(self, listener: ToolListener, owner: CodingToolOwner, *, pid: int) -> None:
+        super().__init__(listener, pid=pid)
         self.owner = owner
         self.completions = set()
 
@@ -152,7 +150,7 @@ def decode_request(self, raw: bytes) -> CodingCall:
         if (
             type(value) is not dict
             or set(value) != {"token", "call_id", "name", "arguments"}
-            or value["token"] != self.token
+            or value["token"] != self.listener.token
         ):
             raise SelectedToolDenied("Coding request is not authenticated")
         return CodingCall.decode(value["call_id"], value["name"], value["arguments"])
@@ -171,12 +169,12 @@ def tool_finished(self, event, input_id):
         task = asyncio.create_task(call.progress.finish_room())
         self.completions.add(task)
 
-    async def _close(self):
-        await super()._close()
+    async def _drain_calls(self):
         if self.completions:
             await asyncio.gather(*tuple(self.completions))
         for call in self.calls.values():
             await asyncio.to_thread(call.progress.room.abandon)
 
-    def failure_response(self, error: Exception) -> dict[str, object]:
+    @staticmethod
+    def failure_response(error: Exception) -> dict[str, object]:
         return {"ok": False, "error": str(error)}

src/agent_comms/checkpoint_seals.py

@@ -10,7 +10,7 @@
 from .declared_family import DeclaredFamily
 from .errors import CheckpointNeedsRepairError, RelationViolationError
 from .field_codec import Id128Text, Sha256Text
-from .private_path import FileRevision
+from .private_path import FileIdentity, FileRevision
 
 if TYPE_CHECKING:
     import os
@@ -66,7 +66,14 @@ def recover(
         path: Path,
         saved: PrefixWitness,
         info: os.stat_result,
-    ) -> PrefixWitness | None: ...
+    ) -> PrefixWitness:
+        return saved.index_tail(bus, marker, db, path, info)
+
+    @abstractmethod
+    def recovery_end(self, info: os.stat_result) -> int: ...
+
+    @abstractmethod
+    def chunk_complete(self, offset: int, size: int, rows: int) -> bool: ...
 
     def check_final(self, saved: PrefixWitness, db_path: Path) -> None:
         raise RelationViolationError("Private bus checkpoint index seal changed or is pending.")
@@ -100,8 +107,24 @@ def recover(
         path: Path,
         saved: PrefixWitness,
         info: os.stat_result,
-    ) -> None:
-        self.check_final(saved, path)
+    ) -> PrefixWitness:
+        if saved.revision.identity != FileIdentity.from_stat(info):
+            self.require_rebuilt(saved, saved)
+            copied = FileRevision.from_stat(path.stat())
+            if (copied.size, copied.mtime_ns) != (self.db_revision.size, self.db_revision.mtime_ns):
+                raise RelationViolationError("Private bus copied index content revision changed.")
+            saved = saved.rebind_source(bus, marker, db, info)
+        else:
+            self.check_final(saved, path)
+        if saved.revision == FileRevision.from_stat(info):
+            return saved
+        return super().recover(bus, marker, db, path, saved, info)
+
+    def recovery_end(self, info: os.stat_result) -> int:
+        return info.st_size
+
+    def chunk_complete(self, offset: int, size: int, rows: int) -> bool:
+        return size >= 16 * 1024 * 1024 or rows >= 10_000
 
     def require_rebuilt(self, saved: PrefixWitness, rebuilt: PrefixWitness) -> None:
         if self.witness != saved.seal():
@@ -129,15 +152,18 @@ def recover(
         saved: PrefixWitness,
         info: os.stat_result,
     ) -> PrefixWitness:
-        from .private_bus_checkpoint import _rebuild_checkpoint_unlocked
-
-        if (
-            saved.seal() not in (self.prior, self.expected)
-            or self.expected.revision != FileRevision.from_stat(info)
-            or self.prior.root_id != marker.root_id
-        ):
+        if (saved.seal() not in (self.prior, self.expected)
+                or not self.expected.revision.holds_cut(
+                    FileRevision.from_stat(info), self.expected.revision.size)
+                or self.prior.root_id != marker.root_id):
             raise RelationViolationError("Private bus checkpoint pending intent is inconsistent.")
-        return _rebuild_checkpoint_unlocked(bus, marker, db, path, info, self)
+        return super().recover(bus, marker, db, path, saved, info)
+
+    def recovery_end(self, info: os.stat_result) -> int:
+        return self.expected.revision.size
+
+    def chunk_complete(self, offset: int, size: int, rows: int) -> bool:
+        return offset == self.expected.revision.size
 
     def require_rebuilt(self, saved: PrefixWitness, rebuilt: PrefixWitness) -> None:
         if (

src/agent_comms/commander/config.py

@@ -81,6 +81,11 @@ def packaged_skill() -> Path:
     def installed(cls) -> CommanderConfig:
         return cls.model_validate_json(cls.path().read_text())
 
+    def repo_directory(self, name: str) -> Path:
+        """A configured GitHub repo name takes precedence over a directory spelling."""
+        return next((repo.checkout for repo in self.repos
+                     if repo.github.rsplit('/', 1)[-1] == name), Path(name)).resolve()
+
     def write(self, home: Path) -> Path:
         """Record the config and start each state file empty, so every tool reads a real file."""
         self.require_commander(Comms(self.root).registry.require(self.commander))

src/agent_comms/commander/order.py

@@ -19,7 +19,7 @@
 """
 import argparse, asyncio, json, os, re, subprocess, sys, time
 from abc import abstractmethod
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 from uuid import uuid4
 
@@ -177,6 +177,16 @@ def forwarders_added(diff, sources=()):
                if m.group(2) not in kept and not state_call.search(diff[m.start():m.end() + 200].split('\n+', 2)[1] if '\n+' in diff[m.start():m.end()+200] else '')])
 
 
+def read_check(line: str) -> OrderCheck:
+    """Resolve the commander's check boundary once; stored checks only carry directories."""
+    if not (check := OrderCheck.parse(line)):
+        sys.exit('Refused: Check reads "<repo> :: <shell command> -> <integer>"')
+    directory = CONFIG.repo_directory(check.directory)
+    if not directory.is_dir():
+        sys.exit(f'Refused: Check repo {check.directory} is not a directory')
+    return replace(check, directory=str(directory))
+
+
 def send(unit, due, body: str, clean=False, mechanical=False, reassign=None, new=None,
          traced=None, replaces=None, disk=False, finding=UnlocatedProblemFinding()):
     """Validate an order body and record it in Core: a new problem (new=(PID, TEXT)) or the listed open problems.
@@ -209,7 +219,8 @@ def send(unit, due, body: str, clean=False, mechanical=False, reassign=None, new
     fact = re.search(r'(?m)^Problems?:\s*(P\d+(?:\s*,\s*P\d+)*)\b', body)
     group = [new[0]] if new else re.findall(r'P\d+', fact[1]) if fact else []
     owner = re.search(r'(?m)^Owner:\s*(\S+):(\d+)\s*$', body)
-    check = (line := re.search(r'(?m)^Check:(.*)$', body)) and OrderCheck.parse(line[1])
+    line = re.search(r'(?m)^Check:(.*)$', body)
+    check = read_check(line[1] if line else '')
     listed = {t.id: t for t in todos()}
     if not group:
         problems.append('no Problem: P<n> line; record it first with `order problem` (every order owns a listed problem)')
@@ -219,14 +230,10 @@ def send(unit, due, body: str, clean=False, mechanical=False, reassign=None, new
         problems.append('no Owner: path:line of the existing owner the fix belongs on')
     elif not os.path.exists(owner[1]) or int(owner[2]) > sum(1 for _ in open(owner[1])):
         problems.append(f'Owner {owner[1]}:{owner[2]} does not exist; read the code and name the real owner')
-    if not check:
-        problems.append('no Check: <repo> :: <command> -> N; done runs it, so make it the real acceptance count')
-    elif not os.path.isdir(check.directory):
-        problems.append(f'Check repo {check.directory} is not a directory')
     # A check that can never print its target, or a premise naming lines that do something else, wastes the order.
     # An order's check must fail now (or it proves nothing), and every absolute
     # path:line it cites must exist.
-    if check and os.path.isdir(check.directory) and not problems:
+    if not problems:
         try:
             ran = subprocess.run(['bash', '-c', check.command], cwd=check.directory, capture_output=True, timeout=90)
             printed = CheckPrinted.decoded(ran.stdout)
@@ -240,7 +247,7 @@ def send(unit, due, body: str, clean=False, mechanical=False, reassign=None, new
     # P12 deleted acp_ingress.py, which Toad imports.
     deleted = {stem for edit in edits
                for stem in re.findall(r'(?i)\bdelete\s+`?(?:\S*/)?(\w+)\.py\b', edit)}
-    if check and deleted and (unread := importers(deleted, check)):
+    if deleted and (unread := importers(deleted, check)):
         problems.append(f'the order deletes {sorted(deleted)}, which {unread} import; '
                         'the check must also count their `from agent_comms` imports')
     if any(re.search(r'(?<![\w/])tools/', edit) for edit in edits) and not all(
@@ -328,10 +335,7 @@ def correct_check(pid: str, line: str, why: str) -> None:
     todo = COMMS.goals.todos.get(pid)
     if not todo.state.open or not todo.pending:
         sys.exit(f'Refused: {pid} has no pending order')
-    if not (check := OrderCheck.parse(line)):
-        sys.exit('Refused: --check reads "<repo> :: <shell command> -> <integer>"')
-    if not os.path.isdir(check.directory):
-        sys.exit(f'Refused: check repo {check.directory} is not a directory')
+    check = read_check(line)
     if len(why.split()) < 8:
         sys.exit('Refused: --why says what was wrong with the recorded check (8+ words)')
     def git(*args: str) -> str:

src/agent_comms/commander/standing.py

@@ -11,9 +11,13 @@
 
 from ..comms import Comms
 from ..goal_actions import OrderGoalAction
-from ..standing_rules import (DatedStandingRule, ReasonedStandingRule, StandingRule,
-                              UndatedStandingRule)
 from ..goals import Goal
+from ..standing_rules import (
+    DatedStandingRule,
+    ReasonedStandingRule,
+    StandingRule,
+    UndatedStandingRule,
+)
 from ..threads import Thread
 from ..todos import FlatTrend, OrderProgress, PullRequests, StrikeTodo, Todo
 from ..turn_digests import TurnDigestRead, TurnsSince
@@ -71,13 +75,13 @@ class WheelSpin(StandingRule, declared_name=None):
 
     @classmethod
     def idle(cls, comms: Comms, commander: str) -> list[Thread]:
-        """Units with a native turn in the last ten minutes and no order, parked threads aside."""
+        """Units with a recent turn in their own transcript and no order, parked threads aside."""
         held = {todo.holder for todo in cls.todos(comms, commander)}
         excluded = held | CONFIG.parked | {commander, ""}
         bound = TurnsSince(time.time() - cls.recent)
         return sorted((thread for thread in comms.registry.all_threads().values()
                        if thread.name not in excluded
-                       and TurnDigestRead(comms.root, thread, bound).native()),
+                       and TurnDigestRead(comms.root, thread, bound).read().turns),
                       key=lambda thread: thread.name)
 
     def assign(self, comms: Comms, commander: str, unit: Thread) -> None:
@@ -181,31 +185,42 @@ def git(checkout: Path, *args: str) -> str:
             ["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")
+        """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 = cls.git(
             repo.checkout,
             "log",
             "--first-parent",
             f"--since={cls.window}",
-            "--format=%H %ct %s",
+            "--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, ...]: