fix(backends,conversations,telegram,ui): a background subagent reports back as its own turn

src/beaver_gateway/backends/claude_sdk.py

@@ -37,6 +37,7 @@ from claude_agent_sdk import (
     PermissionResultDeny,
     ResultMessage,
     StreamEvent,
+    TaskNotificationMessage,
     TextBlock,
     ThinkingBlock,
     ToolResultBlock,
@@ -72,7 +73,14 @@ from beaver_gateway.events.stream import (
 )
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
+    from collections.abc import (
+        AsyncIterator,
+        Awaitable,
+        Callable,
+        Coroutine,
+        Iterable,
+        Sequence,
+    )
 
     from anthropic.types import MessageParam
     from claude_agent_sdk import (
@@ -201,8 +209,16 @@ class ClaudeSdkBackend:
         tool_server: Callable[[str, str], McpSdkServerConfig | None] | None = None,
         asker: Callable[[str, dict[str, Any]], Awaitable[str]] | None = None,
         audit_sink: Callable[[policy_mod.ToolAudit], Awaitable[None]] | None = None,
+        waker: Callable[[str, str], Coroutine[Any, Any, None]] | None = None,
     ) -> None:
+        """One backend per Claude agent.
+
+        ``waker(key, note)`` is called when a session starts a turn by itself
+        (a background task reported back); the caller then attaches.
+        """
         self._agent = agent
+        self.waker = waker
+        self._tasks: set[asyncio.Task[None]] = set()
         self._audit_sink = audit_sink
         self._store = session_store
         self._runner = runner or RunnerConfig()
@@ -256,6 +272,51 @@ class ClaudeSdkBackend:
     def live(self, key: str) -> Session | None:
         return self._pool.get(key)
 
+    def has_pending(self, key: str) -> bool:
+        """A turn the session started on its own is waiting for ``attach``."""
+        live = self._pool.get(key)
+        return live is not None and bool(live.pending)
+
+    def pending_note(self, key: str) -> str | None:
+        live = self._pool.get(key)
+        return live.note if live is not None else None
+
+    async def attach(
+        self,
+        *,
+        key: str,
+        capture: TurnCapture | None = None,
+        observer: Callable[[Any], None] | None = None,
+        turn_id: str | None = None,
+    ) -> AsyncIterator[MessageStreamEvent]:
+        """Consume the turn a live session started by itself as a normal turn.
+
+        The CLI runs one when a background task (a subagent) reports back:
+        no prompt goes out, the events are what the model said in reply to
+        the report, the turn ends on that result.
+        """
+        live = self._pool.get(key)
+        if live is None or not live.pending:
+            return
+        message_id = f"msg_{uuid.uuid4().hex}"
+        yield build_message_start(message_id=message_id, model=self._agent.model)
+        turn = _Turn()
+        async with live.lock:
+            live.running_turn = turn_id or message_id
+            live.last_used = time.monotonic()
+            async for event in self._run_turn(live, None, turn, observer, capture):
+                yield event
+            live.turns += 1
+            live.last_used = time.monotonic()
+            live.running_turn = None
+        usage, _ = await self._after_turn(
+            live, turn, conversation_id=key, history=[], capture=capture
+        )
+        yield build_message_delta(
+            stop_reason=turn.stop_reason, usage=_wire_usage(usage)
+        )
+        yield build_message_stop()
+
     async def repair_session(
         self, session_id: str, *, text: str = "interrupted"
     ) -> int:
@@ -404,70 +465,71 @@ class ClaudeSdkBackend:
     async def _run_turn(
         self,
         live: Session,
-        prompt: str | list[dict[str, Any]],
+        prompt: str | list[dict[str, Any]] | None,
         turn: _Turn,
         observer: Callable[[Any], None] | None = None,
         capture: TurnCapture | None = None,
     ) -> AsyncIterator[MessageStreamEvent]:
-        """Run one prompt, yielding wire events as they arrive.
+        """Run one prompt, or attach to a self-started one, yielding events.
 
         ``turn`` is filled in place (result, synthesized history, count of
         events already yielded) so the caller can finish bookkeeping - and
         decide whether a retry is still possible - after a failure. The
         session id lands in ``capture`` with the first frame, so a turn cut
         by a restart still leaves a resumable session behind.
+
+        A result whose ``origin`` is not the human's ends a turn the CLI ran
+        on its own; inside a prompted turn it is passed over, the events it
+        came with stay in this turn rather than getting lost.
         """
-        streaming = self._agent.options.include_partial_messages
         raw: list[Any] = []
-        next_index = 0
-        offset = 0
-        await live.client.query(prompt if isinstance(prompt, str) else _stream(prompt))
-        async for message in live.client.receive_response():
-            if observer is not None:
-                observer(message)
-            session_id = getattr(message, "session_id", None)
-            if isinstance(session_id, str) and session_id and live.session_id is None:
-                live.session_id = session_id
-                if capture is not None:
-                    capture.session_id = session_id
-            if isinstance(message, MirrorErrorMessage):
-                live.dirty = True
-                _log.error(
-                    "session %s: mirror error, marked dirty: %s",
-                    live.session_id,
-                    message.error,
+        cursor = _Cursor()
+        live.observer = observer
+        live.attached = True
+        for queued in live.pending:
+            live.inbox.put_nowait(queued)
+        live.pending.clear()
+        try:
+            if prompt is not None:
+                await live.client.query(
+                    prompt if isinstance(prompt, str) else _stream(prompt)
                 )
-                continue
-            if getattr(message, "parent_tool_use_id", None) is not None:
-                continue
-            if isinstance(message, StreamEvent):
-                event = message.event
-                if event.get("type") == "message_start":
-                    offset = next_index
-                    turn.context_tokens = _context_of(event.get("message"))
+            while True:
+                message = await live.inbox.get()
+                if message is _END:
+                    if prompt is not None:
+                        msg = f"claude exited before answering ({live.session_id})"
+                        raise RuntimeError(msg)
+                    break
+                if isinstance(message, BaseException):
+                    raise message
+                if observer is not None:
+                    observer(message)
+                if not self._note(live, message, capture):
                     continue
-                index = event.get("index")
-                if isinstance(index, int):
-                    next_index = max(next_index, offset + index + 1)
-                    if streaming:
-                        for out in _emit_stream_event(event, offset + index):
-                            turn.events += 1
-                            yield out
-            elif isinstance(message, AssistantMessage):
-                raw.append(message)
-                if not streaming:
-                    for block in message.content:
-                        for out in _emit_block(block, next_index):
-                            turn.events += 1
-                            yield out
-                        next_index += 1
-            elif isinstance(message, UserMessage):
-                raw.append(message)
-            elif isinstance(message, ResultMessage):
-                turn.result = message
-                turn.stop_reason = _STOP_REASONS.get(
-                    message.stop_reason or "", "end_turn"
-                )
+                if isinstance(message, ResultMessage):
+                    if prompt is not None and _self_started(message):
+                        _log.info(
+                            "session %s: a self-started turn ended inside a "
+                            "prompted one, folded in",
+                            live.session_id,
+                        )
+                        continue
+                    turn.result = message
+                    turn.stop_reason = _STOP_REASONS.get(
+                        message.stop_reason or "", "end_turn"
+                    )
+                    break
+                for out in self._emit(message, turn, cursor, raw):
+                    turn.events += 1
+                    yield out
+            if turn.events == 0:
+                for out in _emit_whole(raw, cursor):
+                    turn.events += 1
+                    yield out
+        finally:
+            live.attached = False
+            live.woken = False
         turn.synthesized = synthesize_turn_messages(raw)
         _log.info(
             "turn: agent=%s session=%s events=%d synthesized=%d stop=%s",
@@ -478,6 +540,88 @@ class ClaudeSdkBackend:
             turn.stop_reason,
         )
 
+    @staticmethod
+    def _note(live: Session, message: Any, capture: TurnCapture | None) -> bool:
+        """Session bookkeeping for one message; False if not the main thread's."""
+        session_id = getattr(message, "session_id", None)
+        if isinstance(session_id, str) and session_id and live.session_id is None:
+            live.session_id = session_id
+            if capture is not None:
+                capture.session_id = session_id
+        if isinstance(message, MirrorErrorMessage):
+            live.dirty = True
+            _log.error(
+                "session %s: mirror error, marked dirty: %s",
+                live.session_id,
+                message.error,
+            )
+            return False
+        if isinstance(message, TaskNotificationMessage):
+            live.note = _note_of(message)
+            return False
+        return getattr(message, "parent_tool_use_id", None) is None
+
+    def _emit(
+        self, message: Any, turn: _Turn, cursor: _Cursor, raw: list[Any]
+    ) -> Iterable[MessageStreamEvent]:
+        streaming = self._agent.options.include_partial_messages
+        if isinstance(message, StreamEvent):
+            event = message.event
+            if event.get("type") == "message_start":
+                cursor.offset = cursor.next_index
+                turn.context_tokens = _context_of(event.get("message"))
+                return
+            index = event.get("index")
+            if isinstance(index, int):
+                cursor.next_index = max(cursor.next_index, cursor.offset + index + 1)
+                if streaming:
+                    yield from _emit_stream_event(event, cursor.offset + index)
+        elif isinstance(message, AssistantMessage):
+            raw.append(message)
+            if not streaming:
+                for block in message.content:
+                    yield from _emit_block(block, cursor.next_index)
+                    cursor.next_index += 1
+        elif isinstance(message, UserMessage):
+            raw.append(message)
+
+    async def _read(self, live: Session) -> None:
+        """Read the client for the session's whole life.
+
+        During a turn every message goes to the turn's inbox. Between turns
+        a subagent still running is shown to the last observer, and the
+        first frame of the main thread means the CLI started a turn by
+        itself: it is buffered in ``pending`` and the waker is told once.
+        """
+        try:
+            async for message in live.client.receive_messages():
+                if live.attached:
+                    live.inbox.put_nowait(message)
+                    continue
+                if isinstance(message, TaskNotificationMessage):
+                    live.note = _note_of(message)
+                if _main_thread(message):
+                    live.pending.append(message)
+                    if not live.woken:
+                        live.woken = True
+                        self._wake(live)
+                elif live.observer is not None:
+                    live.observer(message)
+            live.inbox.put_nowait(_END)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:  # noqa: BLE001
+            _log.error("session %s: reader died: %s", live.session_id, exc)
+            live.dirty = True
+            live.inbox.put_nowait(exc)
+
+    def _wake(self, live: Session) -> None:
+        if self.waker is None:
+            return
+        task = asyncio.create_task(self.waker(live.key, live.note or ""))
+        self._tasks.add(task)
+        task.add_done_callback(self._tasks.discard)
+
     async def _acquire(
         self,
         key: str,
@@ -541,7 +685,7 @@ class ClaudeSdkBackend:
             spec.tools,
             self._runner.user,
         )
-        return Session(
+        live = Session(
             key=key,
             agent=self._agent.name,
             kind=spec.kind,
@@ -550,6 +694,8 @@ class ClaudeSdkBackend:
             resumed=resume is not None,
             pinned=spec.pinned,
         )
+        live.reader = asyncio.create_task(self._read(live))
+        return live
 
     def _default_factory(self, options: ClaudeAgentOptions) -> SessionClient:
         return _RunnerClient(options, uid=self._uid)
@@ -739,6 +885,15 @@ class _SessionSpec:
     tools: bool
 
 
+_END = object()
+
+
+@dataclass
+class _Cursor:
+    next_index: int = 0
+    offset: int = 0
+
+
 @dataclass
 class _Turn:
     events: int = 0
@@ -866,6 +1021,33 @@ async def _stream(content: list[dict[str, Any]]) -> AsyncIterator[dict[str, Any]
     }
 
 
+def _emit_whole(raw: Iterable[Any], cursor: _Cursor) -> Iterable[MessageStreamEvent]:
+    """Blocks of a turn that streamed no deltas, e.g. one reported whole."""
+    for message in raw:
+        if isinstance(message, AssistantMessage):
+            for block in message.content:
+                yield from _emit_block(block, cursor.next_index)
+                cursor.next_index += 1
+
+
+def _self_started(result: ResultMessage) -> bool:
+    origin = result.origin
+    return origin is not None and origin.get("kind") != "human"
+
+
+def _main_thread(message: Any) -> bool:
+    if getattr(message, "parent_tool_use_id", None) is not None:
+        return False
+    return isinstance(
+        message, (StreamEvent, AssistantMessage, UserMessage, ResultMessage)
+    )
+
+
+def _note_of(message: TaskNotificationMessage) -> str:
+    summary = message.summary or message.data.get("description") or message.task_id
+    return f"{summary} ({message.status})"
+
+
 def synthesize_turn_messages(raw: Iterable[Any]) -> list[dict[str, Any]]:
     out: list[dict[str, Any]] = []
     for message in raw:

src/beaver_gateway/backends/sessions.py

@@ -20,7 +20,13 @@ from typing import TYPE_CHECKING, Any, Protocol
 import psutil
 
 if TYPE_CHECKING:
-    from collections.abc import AsyncIterable, AsyncIterator, Iterator, Mapping
+    from collections.abc import (
+        AsyncIterable,
+        AsyncIterator,
+        Callable,
+        Iterator,
+        Mapping,
+    )
 
 __all__ = ["DEFAULT_TTL", "Session", "SessionClient", "SessionPool", "cgroup_limit"]
 
@@ -42,7 +48,7 @@ _RSS_HEADROOM = 0.8
 class SessionClient(Protocol):
     async def connect(self) -> None: ...
     async def query(self, prompt: str | AsyncIterable[dict[str, Any]]) -> None: ...
-    def receive_response(self) -> AsyncIterator[Any]: ...
+    def receive_messages(self) -> AsyncIterator[Any]: ...
     async def interrupt(self) -> None: ...
     async def disconnect(self) -> None: ...
 
@@ -67,6 +73,20 @@ class Session:
     state: dict[str, Any] = field(default_factory=dict)
     """Scratch for policy rules (``agents/policy.py``); dies with the process."""
 
+    reader: asyncio.Task[None] | None = None
+    """Reads the client for the session's whole life, not only during a turn."""
+    inbox: asyncio.Queue[Any] = field(default_factory=asyncio.Queue)
+    """Where the reader puts messages while a turn is attached."""
+    pending: list[Any] = field(default_factory=list)
+    """A turn the CLI started on its own (a background task reported back),
+    buffered until a consumer attaches."""
+    attached: bool = False
+    woken: bool = False
+    note: str | None = None
+    """Summary of the last task notification, the pending turn's prompt."""
+    observer: Callable[[Any], None] | None = None
+    """The last turn's observer; sees subagent traffic between turns."""
+
     @property
     def busy(self) -> bool:
         return self.lock.locked() or self.running_turn is not None
@@ -192,6 +212,8 @@ class SessionPool:
         if session is None:
             return
         _log.info("closing session %s (%s, %s)", session.session_id, session.kind, key)
+        if session.reader is not None and session.reader is not asyncio.current_task():
+            session.reader.cancel()
         try:
             await session.client.disconnect()
         except Exception:  # noqa: BLE001
@@ -238,6 +260,7 @@ class SessionPool:
                 "turns": s.turns,
                 "busy": s.busy,
                 "running_turn": s.running_turn,
+                "pending": len(s.pending),
                 "pending_question": s.pending_question,
                 "pinned": s.pinned,
                 "dirty": s.dirty,

src/beaver_gateway/conversations/state.py

@@ -127,6 +127,12 @@ class State:
         self._tasks = set()
         self._idle_task = None
         self.scheduler = None
+        for backend in backends.values():
+            if hasattr(backend, "waker"):
+                backend.waker = self._woken
+
+    async def _woken(self, key: str, note: str) -> None:
+        raise NotImplementedError
 
     @property
     def db(self) -> Database:

src/beaver_gateway/conversations/texts.py

@@ -77,6 +77,8 @@ class ConversationTexts:
     )
     inject_header: Callable[[injects.InjectContext], str] = injects.inject_header
     bundle_header: str = "[injects accumulated since {since}; not the user]"
+    task_prompt: str = "[subagent reported back: {note}]"
+    """Stands in for the user's words in mirrors of a turn a subagent started."""
     interrupted: str = "interrupted"
     answered: str = "The user answered: {answer}"
     unanswered: str = (

src/beaver_gateway/conversations/turns.py

@@ -54,8 +54,15 @@ class Turns(Seeds):
         tools: bool = True,
         turn_id: str | None = None,
         item_origin: str | None = None,
+        shown: str | None = None,
+        attach: bool = False,
     ) -> AsyncIterator[MessageStreamEvent]:
-        """Run one turn under the conversation's lock; the only path to the backend."""
+        """Run one turn under the conversation's lock; the only path to the backend.
+
+        ``shown`` is what frontends display as the turn's prompt (the user's
+        words, not the envelope). ``attach`` sends nothing: the turn is one
+        the live session started by itself, a subagent reporting back.
+        """
         row_id = cast("int", conv.id)
         runner = self._runner(row_id)
         backend = self._backend(conv.agent_name)
@@ -65,7 +72,7 @@ class Turns(Seeds):
         async with runner.lock:
             runner.turn_id = turn_id
             runner.origin = origin
-            runner.text = _prompt_preview(messages)
+            runner.text = shown if shown is not None else _prompt_preview(messages)
             runner.started_at = datetime.now(UTC)
             runner.tools = {}
             await self._mark_running(conv, turn_id)
@@ -81,18 +88,28 @@ class Turns(Seeds):
             stop = "error"
             cut = False
             try:
-                events = backend.complete(
-                    agent=self._claude_agent(conv.agent_name),
-                    messages=messages,
-                    conversation_id=conv.external_id,
-                    session_id=resume if use_session else None,
-                    reseed=not use_session,
-                    capture=capture,
-                    kind=conv.kind,
-                    pinned=conv.kind == "master",
-                    tools=tools,
-                    observer=self._observer(conv, runner, turn_id, origin),
-                    turn_id=turn_id,
+                observer = self._observer(conv, runner, turn_id, origin)
+                events = (
+                    backend.attach(
+                        key=conv.external_id,
+                        capture=capture,
+                        observer=observer,
+                        turn_id=turn_id,
+                    )
+                    if attach
+                    else backend.complete(
+                        agent=self._claude_agent(conv.agent_name),
+                        messages=messages,
+                        conversation_id=conv.external_id,
+                        session_id=resume if use_session else None,
+                        reseed=not use_session,
+                        capture=capture,
+                        kind=conv.kind,
+                        pinned=conv.kind == "master",
+                        tools=tools,
+                        observer=observer,
+                        turn_id=turn_id,
+                    )
                 )
                 async for event in events:
                     yield event
@@ -128,6 +145,8 @@ class Turns(Seeds):
         turn_id: str | None = None,
         item_origin: str | None = None,
         attachments: Sequence[dict[str, Any]] | None = None,
+        shown: str | None = None,
+        attach: bool = False,
     ) -> tuple[str, TurnCapture]:
         capture = TurnCapture()
         acc = StreamAccumulator()
@@ -140,6 +159,8 @@ class Turns(Seeds):
             tools=tools,
             turn_id=turn_id,
             item_origin=item_origin,
+            shown=shown,
+            attach=attach,
         ):
             acc.feed(event)
         message = acc.finalize(model=agent.model)
@@ -160,6 +181,47 @@ class Turns(Seeds):
         runner = self._runners.get(cast("int", conv.id))
         return runner.origin if runner is not None and runner.turn_id else None
 
+    async def _woken(self, key: str, note: str) -> None:
+        """The live session of ``key`` started a turn by itself; run it next."""
+        conv = await self.get(key)
+        if conv is None:
+            _log.warning("self-started turn on unknown conversation %s: %s", key, note)
+            return
+        _log.info("conversation %s woke by itself: %s", key, note)
+        self._ensure_worker(cast("int", conv.id))
+
+    async def _run_pending(self, conv: Conversation) -> None:
+        turn_id = f"turn_{uuid4().hex[:12]}"
+        note = self._backend(conv.agent_name).pending_note(conv.external_id) or ""
+        try:
+            text, _ = await self.run_text_turn(
+                conv,
+                note,
+                origin="task",
+                turn_id=turn_id,
+                item_origin="task",
+                shown=note,
+                attach=True,
+            )
+        except Exception:  # noqa: BLE001
+            _log.exception(
+                "self-started turn %s on %s failed", turn_id, conv.external_id
+            )
+            return
+        if not text.strip():
+            return
+        self._bus.publish(
+            "reply",
+            conversation_id=conv.external_id,
+            turn_id=turn_id,
+            item=None,
+            item_origin="task",
+            source="task",
+            prompt=self._texts.task_prompt.format(note=note),
+            user_text=None,
+            text=text,
+        )
+
     def _ensure_worker(self, row_id: int) -> None:
         runner = self._runner(row_id)
         runner.wake.set()
@@ -170,6 +232,10 @@ class Turns(Seeds):
     async def _worker(self, row_id: int) -> None:
         runner = self._runner(row_id)
         while True:
+            conv = await self.get_row(row_id)
+            if conv is not None and self._has_pending(conv):
+                await self._run_pending(conv)
+                continue
             items = await self._queue.pending(row_id)
             batch, wait = self._pick(items)
             if batch is None:
@@ -233,6 +299,7 @@ class Turns(Seeds):
                 turn_id=turn_id,
                 item_origin=head.origin,
                 attachments=head.attachments if origin == "user" else None,
+                shown=head.text if origin == "user" else None,
             )
         except Exception:  # noqa: BLE001
             _log.exception("turn %s on %s failed", turn_id, conv.external_id)
@@ -254,6 +321,12 @@ class Turns(Seeds):
                 text=text,
             )
 
+    def _has_pending(self, conv: Conversation) -> bool:
+        try:
+            return self._backend(conv.agent_name).has_pending(conv.external_id)
+        except LookupError:
+            return False
+
     def _bundle(self, items: Sequence[InjectQueueItem]) -> str:
         lines = [self._texts.bundle_header.format(since=iso(items[0].created_at))]
         lines.extend(f"- [{i.origin}] {i.text}" for i in items)

src/beaver_gateway/frontends/telegram/drafts.py

@@ -47,6 +47,7 @@ class Draft:
         self._interval = interval
         self.status = status
         self.text = ""
+        self._paused = False
         self._dirty = True
         self._broken = False
         self._last_sent = 0.0
@@ -65,11 +66,22 @@ class Draft:
         if status != self.status:
             self.status = status
             self._dirty = True
+            self._paused = bool(self.text.strip())
 
     def append(self, text: str) -> None:
-        if text:
+        """Add streamed text; a tool call in between starts a new paragraph.
+
+        The model ends a sentence, calls a tool, then goes on without a
+        break - the same text the final message shows as paragraphs.
+        """
+        if not text:
+            return
+        if self._paused and not self.text.endswith("\n"):
+            self.text = self.text.rstrip() + "\n\n" + text.lstrip()
+        else:
             self.text += text
-            self._dirty = True
+        self._paused = False
+        self._dirty = True
 
     async def stop(self) -> None:
         """Stop pushing and wait for the push in flight.

src/beaver_gateway/frontends/telegram/frontend.py

@@ -777,7 +777,7 @@ class TelegramFrontend(Frontend):
             return
         match kind:
             case "turn.start":
-                if event.get("origin") == "user":
+                if event.get("origin") in ("user", "task"):
                     await self._open_draft(key, event, target)
             case "stream":
                 self._feed_draft(key, event)