P205: start an idle Claude turn through its registered input frame

docs/cli-bridges/claude-code.md

@@ -2,7 +2,37 @@
 
 Claude Code's Monitor injects stdout lines into its conversation. The adapter
 is `agent_comms.cli_bridges.claude` (command `agent-comms-claude-bridge`); all Core delivery behavior comes from
-`agent_comms.cli_bridges.shared`, shared with Codex. No Claude settings hook is needed.
+`agent_comms.cli_bridges.shared`, shared with Codex. No persistent Claude settings change is needed;
+managed sessions receive their lifecycle hooks through per-process settings.
+
+## Start a session that can wake from its first bus message
+
+Monitor needs a tool call first, so it cannot wake a never-prompted session.
+Instead, explicitly launch a fresh Claude session with its stdin owned by the bridge:
+
+```sh
+agent-comms-claude-bridge run \
+  --root /path/to/private-bus --bus-thread claude-code \
+  --session NEW_UUID --state /path/to/private-bridge-state
+```
+
+Declare the external bus thread first. Run launches the real `claude` executable
+in that thread's worktree, with stream-JSON input/output and the supplied session
+UUID. It sends no bootstrap prompt. The first incoming conversation message is
+its first user input; subsequent bus messages use the same process and session.
+Claude's real SessionStart hook starts the same session-owned daemon used by
+Monitor. Run only attaches a managed-input frontend to that daemon; it owns no
+registry entry, inbox watch or transcript reader. Registration retains the
+`IncomingFrame` family's turn-start behavior with the daemon's live endpoint. Delivery is acknowledged only when Claude replays that
+input's UUID, not when the pipe flushes. The session's owner process exit stops the daemon and its bus watch, leaving
+uncertain attempts in the existing journal without replay.
+
+`--claude PATH` selects the executable; repeat `--claude-arg=ARG` to pass Claude
+options. The bridge does not override the configured model, tools or permission
+policy. Raw Claude stream-JSON output goes to stdout; progress goes to stderr.
+The existing transcript reader publishes Claude's actual saved history to Core.
+This is opt-in and starts a new managed session; it does not take over an
+already-running interactive Claude session or replace its Monitor transport.
 
 ## Start the monitor
 
@@ -42,8 +72,9 @@ and cancellation; an internal startup timer cannot safely guess its latency.
 - `DeliveryJournal` records message IDs and transport results in
   `~/.local/state/claude-comms-inbox/delivery.sqlite3`. Core still owns the inbox.
 - `CliBridge` owns complete message formatting, delivery tracking and errors.
-- `Inbox` owns the delivery lifetime and lock; `Watch` also starts transcript
-  publication. `Send` owns a required target, body and kind. Argument decoding
+- The SessionStart-owned `Serve` daemon owns registration, inbox delivery and
+  transcript publication. `Watch` attaches Monitor; `Run` attaches native stdin
+  and holds the same reader lock. `Send` owns a required target, body and kind. Argument decoding
   constructs the operation; execution does not compare command strings.
 - `ClaudeMonitorFrame` owns Monitor's 500-character event cap. It splits a body
   into numbered parts; each begins `Comms in · sender · id · k/n`. The encoded
@@ -56,10 +87,10 @@ and cancellation; an internal startup timer cannot safely guess its latency.
   publication. Monitor delivery therefore requires `watch`; an inbox-only
   command cannot supply session evidence and is not offered. Claude's completed
   JSONL blocks supply replies, visible reasoning,
-  tool calls/results and current activity without a plugin or settings hook.
+  tool calls/results and current activity without a transcript plugin.
   Hidden thinking supplies activity only. Unknown records fail visibly.
 - Startup reads at most the latest 4 MiB, publishing completed items and only
-  the latest status. The bridge then reads appended complete lines every 0.2 s;
+  the latest status. Filesystem notifications admit appended complete lines;
   a partial final line waits for its newline. Replacement or truncation fails.
   No offset is saved. Core makes repeated completed items idempotent.
 

pyproject.toml

@@ -58,6 +58,10 @@ agent-comms-codex-bridge = "agent_comms.cli_bridges.codex:main"
 agent-comms-opencode-bridge = "agent_comms.cli_bridges.opencode:main"
 agent-comms-commander = "agent_comms.commander.cli:main"
 agent-comms-watch-guard = "agent_comms_hooks.watch:main"
+agent-comms-claude-precompact = "agent_comms_hooks.compaction:starting"
+agent-comms-claude-compacted = "agent_comms_hooks.compaction:finished"
+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

@@ -34,6 +34,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar
 
+from pydantic import ConfigDict, TypeAdapter
+
 from .declared_family import DeclaredFamily
 from .errors import RelationViolationError
 from .field_codec import FieldCodec, NullMember, RecordMember, SubclassUnion
@@ -263,6 +265,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 +444,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 +516,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 +535,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:
@@ -1339,6 +1394,15 @@ class StreamingChildStdio(ChildStdio):
     limit: int = 65536
     stderr: int | io.IOBase = subprocess.PIPE
 
+    _output: ClassVar[TypeAdapter[asyncio.StreamReader]] = TypeAdapter(
+        asyncio.StreamReader, config=ConfigDict(arbitrary_types_allowed=True)
+    )
+
+    @classmethod
+    def output(cls, process: asyncio.subprocess.Process) -> asyncio.StreamReader:
+        """Decode the external subprocess result against this declared PIPE contract."""
+        return cls._output.validate_python(process.stdout)
+
     @property
     def options(self) -> dict:
         return {