Let turn origins and child processes own absence behavior

src/agent_comms/backend.py

@@ -395,9 +395,7 @@ def committable_message(self, text):
         return self.started_input and self.accepts_output and self.output.matches_message(text)
 
     def ensure_input_forwarding(self):
-        if self.native.proc.stdin is not None:
-            self.stdin = self.native.proc.stdin
-            self.native.forward_input(partial(self.inputs.forward, self))
+        self.native.forward_input(partial(self.inputs.forward, self))
 
     def observe_input_during_abort(self, event):
         if self.accepts_output:
@@ -413,32 +411,29 @@ def permits_live_status(self) -> bool:
 
     async def abort_stalled_rpc(self) -> None:
         await self.native.finish_input()
-        if self.native.proc.stdin is not None and self.native.proc.returncode is None:
-            abort_deadline = self.loop.time() + max(0.0, self.rpc_abort_grace)
-            try:
-                self.native.proc.stdin.write(
-                    self.native.reader.encode(commands.Abort(id=RequestId("agent-comms-watchdog")))
+        abort_deadline = self.loop.time() + max(0.0, self.rpc_abort_grace)
+        try:
+            remaining = max(0.0, abort_deadline - self.loop.time())
+            self.native.send(commands.Abort(id=RequestId("agent-comms-watchdog")))
+            await asyncio.wait_for(self.native.drain(), timeout=remaining)
+            while remaining := max(0.0, abort_deadline - self.loop.time()):
+                response = await asyncio.wait_for(
+                    self.native.reader.readline(), timeout=remaining
                 )
-                remaining = max(0.0, abort_deadline - self.loop.time())
-                await asyncio.wait_for(self.native.proc.stdin.drain(), timeout=remaining)
-                while remaining := max(0.0, abort_deadline - self.loop.time()):
-                    response = await asyncio.wait_for(
-                        self.native.reader.readline(), timeout=remaining
-                    )
-                    if not response:
-                        break
-                    try:
-                        abort_payload = PiRpcChannel.decode_record(response)
-                    except (json.JSONDecodeError, UnicodeDecodeError, MalformedRecord):
-                        continue
-                    abort_payload.observe_abort(self)
-                    if (
-                        isinstance(abort_payload, pi.Response)
-                        and abort_payload.command is commands.Abort
-                    ):
-                        break
-            except TimeoutError:
-                pass
+                if not response:
+                    break
+                try:
+                    abort_payload = PiRpcChannel.decode_record(response)
+                except (json.JSONDecodeError, UnicodeDecodeError, MalformedRecord):
+                    continue
+                abort_payload.observe_abort(self)
+                if (
+                    isinstance(abort_payload, pi.Response)
+                    and abort_payload.command is commands.Abort
+                ):
+                    break
+        except (TimeoutError, BrokenPipeError):
+            pass
         await self.native.shutdown()
 
     def context_info(self) -> events.ContextInfo:
@@ -490,12 +485,11 @@ async def run(self) -> AsyncGenerator[events.AgentEvent, None]:
                 )
                 async with self.native.failures():
                     self.output.sensitive |= self.native.sensitive_diagnostics
-                    if self.native.proc.stdin is not None:
-                        try:
-                            self.native.proc.stdin.write(self.native.reader.encode(self.native.attestation.request))
-                            await self.native.proc.stdin.drain()
-                        except (BrokenPipeError, ConnectionResetError):
-                            pass
+                    try:
+                        self.native.send(self.native.attestation.request)
+                        await self.native.drain()
+                    except (BrokenPipeError, ConnectionResetError):
+                        pass
                     await self.initialize_rpc()
                     while True:
                         self.skip = False

src/agent_comms/child_process.py

@@ -1525,13 +1525,26 @@ async def start(
                 raise
         return child
 
-    async def write(self, data: bytes) -> None:
-        """Write to this still-running child's input and observe transport backpressure."""
+    def send(self, data: bytes) -> None:
+        """Write synchronously while this child's input still exists and it runs."""
         if self.stdin is None or self.returncode is not None:
             raise BrokenPipeError
         self.stdin.write(data)
+
+    async def drain(self) -> None:
         await self.stdin.drain()
 
+    async def write(self, data: bytes) -> None:
+        """Write to this still-running child's input and observe transport backpressure."""
+        self.send(data)
+        await self.drain()
+
+    async def send_fenced(self, data: bytes, boundary, *, measurements) -> None:
+        """Keep the independent raw writer on this child's original input."""
+        from .native_prompt_send import send_fenced_prompt
+
+        await send_fenced_prompt(self.stdin, data, boundary, measurements=measurements)
+
     @property
     def returncode(self) -> int | None:
         return self.process.returncode

src/agent_comms/input_drain.py

@@ -638,9 +638,9 @@ async def delivery_scope(self, session_id: str) -> InputDeliveryScope:
             return UNOBSERVED_DELIVERY_SCOPE
         keys = set(self.input_keys(session_id))
         keys.update(
-            self.dispositions.bus_key(turn.origin, owner)
+            key
             for turn in self.state(session_id).pending_turns
-            if turn.origin is not None
+            for key in turn.origin.bus_keys(self.dispositions, owner)
         )
         return OwnerQueueInputDeliveryScope(frozenset(keys))
 

src/agent_comms/messages.py

@@ -37,12 +37,14 @@
 from .mentions import ThreadMention
 from .message_reference import MessageReference
 from .page_traversal import LiveCursor, PageCursor
+from .routing import TurnOrigin
 from .task_sources import Decision, NoTaskAttachment, TaskAttachment
 from .thread_identity import AgentThreadRole, ThreadRole
 from .ui_model.message_category import InboundCategory, MessageCategory, OtherCategory
 from .wake_policy import BoundedTriageWake, PassiveWake
 
 if TYPE_CHECKING:
+    from .input_disposition import InputDispositions
     from .historical_views import HistorySource
     from .retained_task_facts import ExactTaskFact
     from .threads import Thread
@@ -361,7 +363,7 @@ class HistoricalMembershipNoticeRole(HistoricalRole, MembershipNoticeRole, decla
 
 
 @dataclass(frozen=True, slots=True)
-class Message(Addressed, BusCapability):
+class Message(Addressed, BusCapability, TurnOrigin):
     """Declares one inter-thread message.
 
     The bus assigns ``seq`` (a monotonically increasing log position) at send
@@ -560,6 +562,19 @@ def exact_reply_target(self) -> str:
         """The exact response destination owned by this opening message."""
         return CommandTarget.named(self.target).reply_target(self.sender)
 
+    autonomous: ClassVar[bool] = False
+
+    @property
+    def separate_turn(self) -> bool:
+        return self.role.response_policy(self).separate_turn
+
+    @property
+    def batch_target(self) -> str:
+        return self.exact_reply_target
+
+    def bus_keys(self, dispositions: InputDispositions, owner: Thread) -> tuple[str, ...]:
+        return (dispositions.bus_key(self, owner),)
+
     @property
     def reply_target(self) -> str | None:
         """An external sender may receive an automatic live reply."""

src/agent_comms/native_child.py

@@ -91,8 +91,21 @@ class PiChild:
     forwarding: InputForwarder = field(default_factory=NoInputForwarder, init=False, repr=False)
     _close_job: asyncio.Task[ChildOutcome | BaseException] | None = field(default=None, init=False, repr=False)
 
+    def send(self, command) -> asyncio.Future:
+        future = self.reader.track(command)
+        self.proc.send(self.reader.command_bytes(command))
+        return future
+
+    async def drain(self) -> None:
+        await self.proc.drain()
+
+    async def send_fenced(self, command, boundary, *, measurements) -> None:
+        self.reader.track(command)
+        await self.proc.send_fenced(self.reader.command_bytes(command), boundary, measurements=measurements)
+
     async def reply_ui(self, response) -> None:
-        await asyncio.wait_for(self.proc.write(self.reader.encode(response)), timeout=2)
+        self.send(response)
+        await asyncio.wait_for(self.drain(), timeout=2)
 
     @classmethod
     async def start_command(cls, argv, *, cwd, env=INHERITED_ENVIRONMENT, pass_fds=(),
@@ -104,7 +117,7 @@ async def request(self, command):
         if self.retired:
             raise RuntimeError('Pi child has retired')
         async with self.failures():
-            response = await self.reader.request(command, self.proc.stdin)
+            response = await self.reader.request(command, self)
         if not response.succeeded:
             raise RuntimeError(response.error.value_or("Pi gave no error text"))
         return response.data.require_payload()
@@ -302,7 +315,7 @@ class BorrowedNative(SuccessorNative):
     async def inspect(self, persistent, prepare, request):
         # The active TurnSession owns receive/correlation. Borrow its original
         # pending response instead of starting a competing reader.
-        async with request.pending_response(self.child.reader, self.child.proc.stdin) as response:
+        async with request.pending_response(self.child) as response:
             return (await response).data.require_payload()
 
 @dataclass

src/agent_comms/pi_commands.py

@@ -86,19 +86,20 @@ def invalidates_identity(cls, response: Response, session: TurnSession) -> bool:
         return False
 
     @asynccontextmanager
-    async def pending_response(self, channel, writer):
-        future = channel.track(self)
+    async def pending_response(self, child):
+        channel = child.reader
+        future = child.send(self)
         try:
-            writer.write(channel.command_bytes(self))
-            await writer.drain()
+            await child.drain()
             yield future
         finally:
             self.id.release_from(channel.pending, type(self))
             future.cancel()
 
-    async def exchange(self, channel, writer, *, strict=False, max_bytes=None, on_event=lambda event: None):
+    async def exchange(self, child, *, strict=False, max_bytes=None, on_event=lambda event: None):
         from .pi_events import Response
-        async with self.pending_response(channel, writer) as future:
+        channel = child.reader
+        async with self.pending_response(child) as future:
             while not future.done():
                 event = await channel.receive(strict=strict, max_bytes=max_bytes)
                 if event is None:
@@ -261,6 +262,7 @@ async def catalog(
         from .child_process import BoundedRun
         from .native_pi import NativePiRpcLaunch
         from .pi_rpc import PiRpcChannel
+        from .native_child import PiChild
 
         try:
             launch = await asyncio.to_thread(
@@ -274,7 +276,7 @@ async def catalog(
             async with BoundedRun.session(
                 launch.argv, cwd=launch.cwd, env=launch.env, timeout=CATALOG_TIMEOUT
             ) as child:
-                yield PiRpcChannel(child.stdout), child.stdin
+                yield PiChild(child, PiRpcChannel(child.stdout))
         except TimeoutError as error:
             raise NativePiUnavailable(
                 f"Native catalog discovery timed out: no catalog reply from {agent_bin!r} "
@@ -287,8 +289,8 @@ async def discover(
         self, agent_bin: str, arguments: Sequence[str], root: Path | str | None = None,
         *, package: VerifiedNativePackage,
     ) -> PiResponseData:
-        async with self.catalog(agent_bin, arguments, root=root, package=package) as (channel, writer):
-            response = await self.exchange(channel, writer)
+        async with self.catalog(agent_bin, arguments, root=root, package=package) as child:
+            response = await self.exchange(child)
             if response.succeeded:
                 return response.data.require_payload()
             raise NativePiUnavailable(response.error.text_or("Native catalog query was refused"))
@@ -342,11 +344,11 @@ async def configure(
         semantics cannot change a running turn, a saved transcript or its budget.
         """
         catalog = self.catalog(agent_bin, arguments, package=package, worktree=worktree, root=root)
-        async with catalog as (channel, writer):
-            response = await self.exchange(channel, writer)
+        async with catalog as child:
+            response = await self.exchange(child)
             if not response.succeeded:
                 raise NativePiUnavailable(response.error.text_or(self.error_message))
-            state = await GetState().exchange(channel, writer)
+            state = await GetState().exchange(child)
             if not state.succeeded:
                 raise NativePiUnavailable(
                     state.error.text_or("Native configuration was not reported")

src/agent_comms/pi_rpc.py

@@ -71,10 +71,10 @@ async def next_event(self):
         except asyncio.QueueShutDown:
             return reading.result()
 
-    async def request(self, command, writer):
+    async def request(self, command, child):
         if self._read_task is None:
-            return await command.exchange(self, writer, strict=True, on_event=self.event_type.unowned_event)
-        async with command.pending_response(self, writer) as future:
+            return await command.exchange(child, strict=True, on_event=self.event_type.unowned_event)
+        async with command.pending_response(child) as future:
             await asyncio.wait((future, self._read_task), return_when=asyncio.FIRST_COMPLETED)
             response = future.result() if future.done() else self._read_task.result()
             if response is None:
@@ -106,11 +106,6 @@ async def receive(
             return None
         return self.decode_record(raw, strict=strict, max_bytes=max_bytes, event_type=self.event_type)
 
-    def encode(self, command: PiCommand) -> bytes:
-        """Register before writing; correlated proof still requires its native input ID."""
-        self.track(command)
-        return self.command_bytes(command)
-
     def track(self, command: PiCommand) -> asyncio.Future[Response]:
         """Expose the same pending response to the request's lifecycle owner."""
         return command.id.track_in(self.pending, type(command), command)

src/agent_comms/private_send_claim.py

@@ -30,7 +30,7 @@
 from .native_entries import NativeEvidenceScope
 from .native_pi import NativeContextProof, NativePiTerminalFailure, NativeTurnResult
 from .native_prompt_binding import bind_expected_prompt
-from .native_prompt_send import PromptBoundary, send_fenced_prompt
+from .native_prompt_send import PromptBoundary
 from .native_session_reopen import NativeSessionIdentity
 from .native_startup import NativeStartup
 from .native_tool_call import NativeToolMode
@@ -271,11 +271,9 @@ def _start_once(self, identity: NativeSessionIdentity) -> None:
             # all of these operations have finished successfully.
 
     async def send_prompt(self, turn: TrackedTurnSession) -> None:
-        stdin = turn.native.proc.stdin
-        assert stdin is not None
         with turn.startup.measurements.operation("prompt_writer_join"):
-            await send_fenced_prompt(
-                stdin, turn.native.reader.encode(turn.command),
+            await turn.native.send_fenced(
+                turn.command,
                 partial(self.prompt_boundary, turn), measurements=turn.startup.measurements,
             )
 

src/agent_comms/prompt_state.py

@@ -67,7 +67,7 @@ def dispatch(self):
     async def deliver_to(self, inputs, session):
         def write():
             session.grant_prompt()
-            session.stdin.write(session.native.reader.encode(session.command))
+            session.native.send(session.command)
         async with MaintenanceBarrier(RegistryFile.path_in(session.startup.root)).ingress_async() as status:
             with status.phase.send_new(session.send_boundary, None, session.command.input_id, session.command.message) as outcome:
                 completion = outcome.dispatch(inputs, session, self, write)