Integrate #2096 #2105 #1950 #2109

plans/commander-native-plan.md

@@ -48,8 +48,8 @@
 | 15 | Standing rules broken right now | `rotate.py broken_rules` | `StandingRule` members derive violations on read | `control_plane` |
 | 16 | Rule review | `rules.py`; `rules.jsonl`; maximum age | Rule instruction segments in the commander's turn context, with provenance | `control_plane` |
 | 17 | Load and forking | `agent-comms-commander capacity`: load and live fork reports; ten-minute beat age | Reporting-only `commander.capacity.Load.of`; no capacity refusal or automatic fork policy | `control_plane` |
-| 18 | Growth alarm | `growth_watch.py`: merges growing src by more than 40 net lines | `commander.config.Repo.main_changes` → `commander_measurements.MergeGrowth.read` | `control_plane` |
-| 19 | Storage hygiene | `storage.py`: checkout limit, cleanup under 30 GB, allocation floor 5 GB | `UnitStorage`, derived from worktree paths and free disk | `control_plane` |
+| 18 | Growth alarm | `StandingRule.broken` growth delivery (formerly `growth_watch.py`) | `commander.standing.MergeGrowth.violations`: captured first-parent growth | `control_plane` |
+| 19 | Storage hygiene | `StandingRule.broken` storage delivery (formerly `storage.py`) | `UnitStorage.violations`, `DiskLow.violations`; live `StorageRule` thresholds | `control_plane` |
 | 20 | Commander reaches units only through orders | `send_guard.py` (PreToolUse hook) | Harness hook checking Core order tools | `control_plane` |
 | 21 | Paths, repositories, commander identity | `CommanderConfig` from `commander.json` | Core project and registry configuration | `cutover` |
 | 22 | Seeing the fleet: board, strikes, queue, story, alarms | `fleet-board.md`, `order.py board`, `rotate.py`, `judge.py` | Toad commander page rendering Core through `ui_model` | `commander_page` |

pyproject.toml

@@ -60,6 +60,8 @@ agent-comms-commander = "agent_comms.commander.cli:main"
 agent-comms-watch-guard = "agent_comms_hooks.watch:main"
 agent-comms-commander-skill = "agent_comms_hooks.skill:main"
 agent-comms-owner-directive = "agent_comms_hooks.directive:main"
+agent-comms-claude-session-started = "agent_comms_hooks.session:started"
+agent-comms-claude-session-ended = "agent_comms_hooks.session:ended"
 agent-comms-connector = "agent_comms.connector:main"
 opencode-comms = "agent_comms.cli_bridges.opencode_launch:main"
 

src/agent_comms/child_process.py

@@ -263,6 +263,20 @@ def owns_turn(self, turn: ActiveTurn) -> bool:
     def signal(self, signum: int) -> None:
         Platform.current().send(self, signum)
 
+    def require_ancestor_of(self, pid: int) -> None:
+        """A supplied process is authority only for its actual descendant."""
+        platform = Platform.current()
+        platform.require(self)
+        current = platform.identity(pid)
+        while current.pid > 1:
+            if current == self:
+                return
+            current = platform.parent_identity(current.pid)
+        raise IdentityMismatchError('The declared process is not an ancestor of this hook')
+
+    async def wait_exited(self) -> None:
+        await Platform.current().wait_exited(self)
+
     def require_release_by(self, process: ProcessIdentity) -> None:
         if self != process:
             raise RelationViolationError("Only the registered owner may release itself.")
@@ -428,6 +442,12 @@ def force_group(self, identity: ProcessIdentity) -> None:
     @abstractmethod
     def identity(self, pid: int) -> ProcessIdentity: ...
 
+    def parent_identity(self, pid: int) -> ProcessIdentity:
+        raise NotImplementedError('This platform does not expose exact parent identities')
+
+    async def wait_exited(self, identity: ProcessIdentity) -> None:
+        raise NotImplementedError('This platform does not expose process exit observation')
+
     def matches(self, identity: ProcessIdentity) -> bool:
         try:
             return self.identity(identity.pid) == identity
@@ -494,9 +514,16 @@ def send(self, identity: ProcessIdentity, signum: int) -> None:
         os.kill(identity.pid, signum)
 
 
+@dataclass(frozen=True)
+class LinuxProcessRecord:
+    identity: ProcessIdentity
+    group: int
+    parent: int
+
+
 class LinuxPlatform(NamespaceContainment, PidfdHandles, PosixPlatform):
     @staticmethod
-    def _stat(pid: int) -> tuple[ProcessIdentity, int]:
+    def _stat(pid: int) -> LinuxProcessRecord:
         try:
             # comm can contain whitespace and ')'; fields after its final ')'
             # start at field 3. Kernel starttime is field 22, pgrp is field 5.
@@ -506,19 +533,45 @@ def _stat(pid: int) -> tuple[ProcessIdentity, int]:
         if fields[0] in {"Z", "X"}:
             raise ProcessLookupError(pid)
         group = int(fields[2])
-        return ProcessIdentity.in_group(pid, int(fields[19]), group), group
+        return LinuxProcessRecord(
+            ProcessIdentity.in_group(pid, int(fields[19]), group), group, int(fields[1]))
 
     def identity(self, pid: int) -> ProcessIdentity:
-        return self._stat(pid)[0]
+        return self._stat(pid).identity
+
+    def parent_identity(self, pid: int) -> ProcessIdentity:
+        return self.identity(self._stat(pid).parent)
+
+    async def wait_exited(self, identity: ProcessIdentity) -> None:
+        try:
+            descriptor = self.open_pidfd(identity.pid)
+        except ProcessLookupError:
+            return
+        try:
+            self.require(identity)
+            loop = asyncio.get_running_loop()
+            exited = loop.create_future()
+            def observe() -> None:
+                if not exited.done():
+                    exited.set_result(None)
+            loop.add_reader(descriptor, observe)
+            try:
+                await exited
+            finally:
+                loop.remove_reader(descriptor)
+        except ProcessLookupError:
+            return
+        finally:
+            os.close(descriptor)
 
     def members_of_group(self, pgid: int) -> tuple[ProcessIdentity, ...]:
         members = []
         for entry in Path("/proc").iterdir():
             if entry.name.isdecimal():
                 with suppress(ProcessLookupError):
-                    identity, group = self._stat(int(entry.name))
-                    if group == pgid:
-                        members.append(identity)
+                    record = self._stat(int(entry.name))
+                    if record.group == pgid:
+                        members.append(record.identity)
         return tuple(members)
 
     def send(self, identity: ProcessIdentity, signum: int) -> None:

src/agent_comms/cli_bridges/claude.py

@@ -12,9 +12,10 @@
 from agent_comms.coordination_database import CoordinationStore
 from agent_comms.private_path import FileRevision
 from agent_comms.wire_watch import WatchSource, WireWatch
+from agent_comms_hooks.session import SessionHook
 from .claude_protocol import CLAUDE_RECORD, ClaudeRecord, ClaudeSession, ClaudeMonitorFrame
 from .history import ClaudeHistory, JsonLinesFile, OfflineHistory, OnlineHistory
-from .shared import Binding, LineMonitorBridge, Options, run_bridge
+from .shared import Binding, Enrol, LineMonitorBridge, Options, run_bridge
 from agent_comms.runtime_updates import RuntimeUpdate, SessionFacts
 
 
@@ -104,23 +105,39 @@ def history_start(self, stream: BinaryIO, boundary: int) -> int:
 
     async def transcript(self) -> AsyncIterator[RuntimeUpdate]:
         session = self.start_state()
-        with self.source.open('rb') as stream:
-            original = FileRevision.from_file(stream)
-            async with aclosing(WireWatch.observations(WatchSource.modified_file(self.source))) as changes:
+        source = self.options.operation.transcript_source()
+        async with aclosing(WireWatch.observations(WatchSource.modified_file(source))) as changes:
+            await anext(changes)
+            while not source.exists():
+                await anext(changes)
+            with source.open('rb') as stream:
+                original = FileRevision.from_file(stream)
                 for update in self.decoded(stream, self.history_start(stream, original.size), original.size, session):
                     yield update
                 # Only the current status is published from the recent window.
                 for update in session.phase.window_ended(session):
                     yield update
                 async for _ in changes:
                     opened = FileRevision.from_file(stream)
-                    current = FileRevision.capture(self.source)
+                    current = FileRevision.capture(source)
                     if (current is None or current.identity != original.identity
                             or min(opened.size, current.size) < stream.tell()):
                         raise ValueError('Claude transcript was replaced or truncated')
                     for update in self.decoded(stream, stream.tell(), opened.size, session):
                         yield update
 
 
+class Connect(Enrol):
+    @classmethod
+    def serves(cls, bridge: type) -> bool:
+        return issubclass(bridge, ClaudeBridge)
+
+    async def execute(self, bridge: LineMonitorBridge) -> None:
+        await super().execute(bridge)
+        SessionHook.current(
+            bridge.binding.thread, str(RunningClaude(bridge.binding.thread).source)
+        ).start()
+
+
 def main() -> None:
     run_bridge(ClaudeBridge, Options.parse(__doc__, Path.home() / '.local/state/claude-comms-inbox', ClaudeBridge))

src/agent_comms/cli_bridges/claude_protocol.py

@@ -43,16 +43,16 @@
 from .shared import (
     MONITOR_FRAME,
     ForeignMonitorFrame,
-    IncomingFrame,
     MonitorFrame,
+    MonitorIncomingFrame,
     MonitorObservation,
     MonitorPart,
     MonitorProbe,
     ObservedMonitor,
 )
 
 
-class ClaudeMonitorFrame(IncomingFrame):
+class ClaudeMonitorFrame(MonitorIncomingFrame):
     """Claude Monitor caps lines at 500 and joined batches at 3000 UTF-16 units."""
     EVENT_CHARS = 500
     BATCH_CHARS = 3000