fix(webui): deliver late subagent results as new turns (#4992)

nanobot/agent/loop.py

@@ -33,16 +33,14 @@ from nanobot.agent.tools.file_state import FileStateStore, bind_file_states, res
 from nanobot.agent.tools.message import MessageTool
 from nanobot.agent.tools.registry import ToolRegistry
 from nanobot.agent.tools.self import MyTool
+from nanobot.agent.turn_delivery import (
+    TurnDelivery,
+    TurnDeliveryFactory,
+)
+from nanobot.agent.turn_delivery import TurnRoute as TurnRoute
 from nanobot.agent.turn_hooks import AgentTurnHookSpec, build_agent_turn_hook
 from nanobot.bus.events import InboundMessage, OutboundMessage
-from nanobot.bus.outbound_events import (
-    RetryWaitEvent,
-    StreamDeltaEvent,
-    StreamedResponseEvent,
-    StreamEndEvent,
-    outbound_message_for_event,
-)
-from nanobot.bus.progress import build_bus_progress_callback
+from nanobot.bus.outbound_events import StreamedResponseEvent
 from nanobot.bus.queue import MessageBus
 from nanobot.bus.runtime_events import (
     RuntimeEventBus,
@@ -116,15 +114,6 @@ class TurnKind(Enum):
     SYSTEM = auto()
 
 
-@dataclass(frozen=True)
-class TurnRoute:
-    """Where a turn response is delivered, separate from its execution input."""
-
-    channel: str
-    chat_id: str
-    metadata: dict[str, Any] = field(default_factory=dict)
-
-
 @dataclass
 class StateTraceEntry:
     state: TurnState
@@ -142,7 +131,7 @@ class TurnContext:
     turn_id: str
     runtime: LLMRuntime
     kind: TurnKind
-    route: TurnRoute
+    delivery: TurnDelivery
     original_user_text: str | None = None
     session: Session | None = None
 
@@ -157,7 +146,7 @@ class TurnContext:
     stop_reason: str = ""
     had_injections: bool = False
 
-    user_persisted_early: bool = False
+    input_persisted_early: bool = False
     save_skip: int = 0
 
     outbound: OutboundMessage | None = None
@@ -296,6 +285,7 @@ class AgentLoop:
         model_preset: str | None = None,
         preset_snapshot_loader: preset_helpers.PresetSnapshotLoader | None = None,
         runtime_events: RuntimeEventBus | None = None,
+        turn_delivery_factory: TurnDeliveryFactory | None = None,
         runtime_model_publisher: Callable[[str, str | None], None] | None = None,
         restart_mode: str = "auto",
         local_trigger_store: Any | None = None,
@@ -305,8 +295,20 @@ class AgentLoop:
         _tc = tools_config or ToolsConfig()
         defaults = AgentDefaults()
         self.bus = bus
-        self.runtime_events = runtime_events or RuntimeEventBus()
-        self.runtime_event_publisher = RuntimeEventPublisher(self.runtime_events)
+        if turn_delivery_factory is not None:
+            if turn_delivery_factory.bus is not bus:
+                raise ValueError("turn delivery factory must use the agent message bus")
+            if (
+                runtime_events is not None
+                and turn_delivery_factory.runtime_events is not runtime_events
+            ):
+                raise ValueError("turn delivery factory must use the agent runtime event bus")
+            self.turn_delivery_factory = turn_delivery_factory
+            self.runtime_events = turn_delivery_factory.runtime_events
+        else:
+            self.runtime_events = runtime_events or RuntimeEventBus()
+            self.turn_delivery_factory = TurnDeliveryFactory(bus, self.runtime_events)
+        self.runtime_event_publisher = self.turn_delivery_factory.runtime_event_publisher
         self.channels_config = channels_config
         self.restart_mode = restart_mode
         self._runtime_model_publisher = runtime_model_publisher
@@ -589,53 +591,6 @@ class AgentLoop:
         if provider not in self._runtime_context_providers:
             self._runtime_context_providers.append(provider)
 
-    @staticmethod
-    def _turn_route(msg: InboundMessage, session_key: str) -> TurnRoute:
-        """Resolve response routing without mixing it into execution metadata."""
-        if msg.channel != "system":
-            return TurnRoute(
-                channel=msg.channel,
-                chat_id=msg.chat_id,
-                metadata=dict(msg.metadata or {}),
-            )
-
-        channel, chat_id = (
-            msg.chat_id.split(":", 1) if ":" in msg.chat_id else ("cli", msg.chat_id)
-        )
-        metadata: dict[str, Any] = {}
-        if (
-            channel == "slack"
-            and session_key.startswith("slack:")
-            and session_key.count(":") >= 2
-        ):
-            metadata["slack"] = {"thread_ts": session_key.split(":", 2)[2]}
-        if origin_message_id := msg.metadata.get("origin_message_id"):
-            metadata["origin_message_id"] = origin_message_id
-        return TurnRoute(channel=channel, chat_id=chat_id, metadata=metadata)
-
-    async def _build_bus_progress_callback(
-        self, msg: InboundMessage
-    ) -> Callable[..., Awaitable[None]]:
-        """Build a progress callback that publishes to the message bus."""
-        return build_bus_progress_callback(self.bus, msg)
-
-    async def _build_retry_wait_callback(
-        self, msg: InboundMessage
-    ) -> Callable[[str], Awaitable[None]]:
-        """Build a retry-wait callback that publishes to the message bus."""
-
-        async def _on_retry_wait(content: str) -> None:
-            await self.bus.publish_outbound(
-                outbound_message_for_event(
-                    channel=msg.channel,
-                    chat_id=msg.chat_id,
-                    event=RetryWaitEvent(content=content),
-                    metadata=msg.metadata,
-                )
-            )
-
-        return _on_retry_wait
-
     def _runtime_events(self) -> RuntimeEventPublisher:
         return ensure_runtime_event_publisher(self)
 
@@ -696,15 +651,16 @@ class AgentLoop:
     def _build_initial_messages(self, ctx: TurnContext) -> list[dict[str, Any]]:
         """Build the initial message list for the LLM turn."""
         assert ctx.session is not None
-        is_subagent = ctx.kind is TurnKind.SYSTEM and ctx.msg.sender_id == "subagent"
         scope = self.workspace_scopes.for_message(ctx.msg, ctx.session.metadata)
         return self.context.build_messages(
             history=ctx.history,
-            current_message="" if is_subagent else ctx.msg.content,
+            current_message=ctx.msg.content,
             media=ctx.msg.media if ctx.kind is TurnKind.USER and ctx.msg.media else None,
-            channel=ctx.route.channel,
-            chat_id=str(ctx.msg.metadata.get("context_chat_id") or ctx.route.chat_id),
-            current_role="assistant" if is_subagent else "user",
+            channel=ctx.delivery.route.channel,
+            chat_id=str(
+                ctx.msg.metadata.get("context_chat_id") or ctx.delivery.route.chat_id
+            ),
+            current_role="user",
             sender_id=ctx.msg.sender_id,
             session_summary=ctx.pending_summary,
             session_metadata=ctx.session.metadata,
@@ -718,13 +674,13 @@ class AgentLoop:
     def _request_context_for_turn(self, ctx: TurnContext) -> RequestContext:
         assert ctx.session is not None
         scope = self.workspace_scopes.for_turn(
-            channel=ctx.route.channel,
+            channel=ctx.delivery.route.channel,
             message_metadata=ctx.msg.metadata,
             session_metadata=ctx.session.metadata,
         )
         return RequestContext(
-            channel=ctx.route.channel,
-            chat_id=ctx.route.chat_id,
+            channel=ctx.delivery.route.channel,
+            chat_id=ctx.delivery.route.chat_id,
             message_id=ctx.msg.metadata.get("message_id"),
             session_key=ctx.session_key,
             original_user_text=ctx.original_user_text,
@@ -1123,6 +1079,7 @@ class AgentLoop:
         lock = self._session_locks.setdefault(session_key, asyncio.Lock())
         gate = self._concurrency_gate or nullcontext()
 
+        delivery = self.turn_delivery_factory.unrouted(msg, session_key)
         pending: asyncio.Queue | None = None
         try:
             async with lock, gate:
@@ -1131,66 +1088,23 @@ class AgentLoop:
                 pending = asyncio.Queue(maxsize=20)
                 self._pending_queues[session_key] = pending
                 try:
-                    on_stream = on_stream_end = None
-                    if msg.metadata.get("_wants_stream"):
-                        # Split one answer into distinct stream segments.
-                        stream_base_id = f"{msg.session_key}:{time.time_ns()}"
-                        stream_segment = 0
-
-                        def _current_stream_id() -> str:
-                            return f"{stream_base_id}:{stream_segment}"
-
-                        async def on_stream(delta: str) -> None:
-                            await self.bus.publish_outbound(
-                                outbound_message_for_event(
-                                    channel=msg.channel,
-                                    chat_id=msg.chat_id,
-                                    event=StreamDeltaEvent(
-                                        content=delta,
-                                        stream_id=_current_stream_id(),
-                                    ),
-                                    metadata=msg.metadata,
-                                )
-                            )
-
-                        async def on_stream_end(*, resuming: bool = False) -> None:
-                            nonlocal stream_segment
-                            await self.bus.publish_outbound(
-                                outbound_message_for_event(
-                                    channel=msg.channel,
-                                    chat_id=msg.chat_id,
-                                    event=StreamEndEvent(
-                                        stream_id=_current_stream_id(),
-                                        resuming=resuming,
-                                    ),
-                                    metadata=msg.metadata,
-                                )
-                            )
-                            stream_segment += 1
-
+                    delivery = self.turn_delivery_factory.create(
+                        msg,
+                        session_key,
+                        enable_stream=True,
+                    )
                     response = await self._process_message(
-                        msg, on_stream=on_stream, on_stream_end=on_stream_end,
+                        msg,
+                        on_stream=delivery.on_stream,
+                        on_stream_end=delivery.on_stream_end,
                         pending_queue=pending,
+                        delivery=delivery,
                     )
-                    completed_channel = msg.channel
-                    completed_chat_id = msg.chat_id
-                    if response is not None:
-                        await self.bus.publish_outbound(response)
-                        completed_channel = response.channel
-                        completed_chat_id = response.chat_id
-                    elif msg.channel == "cli":
-                        await self.bus.publish_outbound(OutboundMessage(
-                            channel=msg.channel, chat_id=msg.chat_id,
-                            content="", metadata=msg.metadata or {},
-                        ))
                     continuing = turn_continuation.internal_continuation_pending(msg.metadata)
-                    if not continuing:
-                        await self._runtime_events().turn_completed(
-                            channel=completed_channel,
-                            chat_id=completed_chat_id,
-                            session_key=session_key,
-                            metadata=msg.metadata,
-                        )
+                    await delivery.complete(
+                        response,
+                        publish_completion=not continuing,
+                    )
                     for _, coordinator in self._automation_turn_coordinators:
                         coordinator.complete(msg, response=response)
                 except asyncio.CancelledError:
@@ -1223,17 +1137,11 @@ class AgentLoop:
                     raise
                 except Exception as exc:
                     logger.exception("Error processing message for session {}", session_key)
-                    await self.bus.publish_outbound(OutboundMessage(
-                        channel=msg.channel, chat_id=msg.chat_id,
-                        content="Sorry, I encountered an error.",
-                    ))
-                    if not turn_continuation.internal_continuation_pending(msg.metadata):
-                        await self._runtime_events().turn_completed(
-                            channel=msg.channel,
-                            chat_id=msg.chat_id,
-                            session_key=session_key,
-                            metadata=msg.metadata,
+                    await delivery.fail(
+                        publish_completion=not turn_continuation.internal_continuation_pending(
+                            msg.metadata
                         )
+                    )
                     for _, coordinator in self._automation_turn_coordinators:
                         coordinator.complete(msg, error=exc)
                 finally:
@@ -1262,17 +1170,11 @@ class AgentLoop:
                                 leftover, session_key,
                             )
                     if not turn_continuation.internal_continuation_pending(msg.metadata):
-                        await self._runtime_events().run_status_changed(
-                            msg, session_key, "idle"
-                        )
-                        self._runtime_events().clear_turn(session_key)
+                        await delivery.idle()
                     await self._publish_next_deferred_automation_turn(session_key)
         finally:
             if pending is None:
-                await self._runtime_events().run_status_changed(
-                    msg, session_key, "idle"
-                )
-                self._runtime_events().clear_turn(session_key)
+                await delivery.idle()
                 await self._publish_next_deferred_automation_turn(session_key)
 
     async def close_mcp(self) -> None:
@@ -1321,6 +1223,7 @@ class AgentLoop:
         hook_factories: list[AgentTurnHookFactory] | None = None,
         tools: ToolRegistry | None = None,
         runtime: LLMRuntime | None = None,
+        delivery: TurnDelivery | None = None,
     ) -> OutboundMessage | None:
         """Process a single inbound message and return the response."""
         if runtime is None:
@@ -1334,7 +1237,14 @@ class AgentLoop:
             key = session_key or msg.session_key_override or f"{destination[0]}:{destination[1]}"
         else:
             key = session_key or msg.session_key
-        route = self._turn_route(msg, key)
+        if delivery is None:
+            delivery = self.turn_delivery_factory.create(msg, key)
+        elif delivery.session_key != key:
+            raise ValueError("turn delivery session does not match the processing session")
+        if on_stream is None:
+            on_stream = delivery.on_stream
+        if on_stream_end is None:
+            on_stream_end = delivery.on_stream_end
         t0 = time.time()
         ctx = TurnContext(
             msg=msg,
@@ -1344,7 +1254,7 @@ class AgentLoop:
             turn_id=f"{key}:{time.time_ns()}",
             runtime=runtime,
             kind=kind,
-            route=route,
+            delivery=delivery,
             original_user_text=(
                 None
                 if kind is TurnKind.SYSTEM
@@ -1474,8 +1384,8 @@ class AgentLoop:
         # ensure it exists in case this handler is invoked independently.
         if ctx.session is None:
             ctx.session = self.sessions.get_or_create(ctx.session_key)
+        await ctx.delivery.started()
         if ctx.kind is TurnKind.USER:
-            await self._runtime_events().session_turn_started(msg, ctx.session_key)
             self.workspace_scopes.persist_message_scope(ctx.session, msg)
 
         if self._restore_runtime_checkpoint(ctx.session):
@@ -1530,7 +1440,7 @@ class AgentLoop:
             # them out of LLM context.  /new is excluded because it
             # intentionally clears the session.
             if cmd_ctx.raw.lower() != "/new":
-                ctx.user_persisted_early = self._persist_user_message_early(
+                ctx.input_persisted_early = self._persist_user_message_early(
                     ctx.msg, ctx.session, _command=True
                 )
                 ctx.session.add_message(
@@ -1552,9 +1462,6 @@ class AgentLoop:
                 replay_max_messages=replay_max_messages,
             )
         is_subagent = ctx.kind is TurnKind.SYSTEM and ctx.msg.sender_id == "subagent"
-        if is_subagent and self._persist_subagent_followup(ctx.session, ctx.msg):
-            logger.debug("Subagent result persisted for session {}", ctx.session_key)
-            self.sessions.save(ctx.session)
 
         if ctx.kind is TurnKind.USER and (message_tool := self.tools.get("message")):
             if isinstance(message_tool, MessageTool):
@@ -1566,39 +1473,40 @@ class AgentLoop:
             "extend_to_user": is_subagent,
         }
         ctx.history = ctx.session.get_history(**_hist_kwargs)
-        self._runtime_events().record_turn_runtime(
-            ctx.session_key,
-            ctx.runtime,
-        )
+        if is_subagent:
+            # Keep the durable internal delivery as an assistant record, but
+            # present this completion to the model as fresh follow-up input.
+            # Providers without assistant-prefill support drop trailing
+            # assistant messages, so using the persisted record as the current
+            # prompt would hide an independently dispatched subagent result.
+            if self._persist_subagent_followup(ctx.session, ctx.msg):
+                logger.debug("Subagent result persisted for session {}", ctx.session_key)
+                self.sessions.save(ctx.session)
+            ctx.input_persisted_early = True
+        ctx.delivery.record_runtime(ctx.runtime)
 
         ctx.request_context = self._request_context_for_turn(ctx)
         if ctx.kind is TurnKind.USER:
             ctx.runtime_context_blocks = await self._resolve_runtime_context_for_turn(ctx)
         ctx.initial_messages = self._build_initial_messages(ctx)
         if ctx.kind is TurnKind.USER:
-            ctx.user_persisted_early = self._persist_user_message_early(
+            ctx.input_persisted_early = self._persist_user_message_early(
                 ctx.msg,
                 ctx.session,
                 runtime_context_blocks=ctx.runtime_context_blocks,
             )
 
-            if ctx.on_progress is None:
-                ctx.on_progress = await self._build_bus_progress_callback(ctx.msg)
-            if ctx.on_retry_wait is None:
-                ctx.on_retry_wait = await self._build_retry_wait_callback(ctx.msg)
+        if ctx.on_progress is None:
+            ctx.on_progress = ctx.delivery.progress_callback()
+        if ctx.on_retry_wait is None:
+            ctx.on_retry_wait = ctx.delivery.retry_wait_callback()
 
         return "ok"
 
     async def _state_run(self, ctx: TurnContext) -> str:
         if ctx.visible_run_started_at is None:
             ctx.visible_run_started_at = time.time()
-        if ctx.kind is TurnKind.USER:
-            await self._runtime_events().run_status_changed(
-                ctx.msg,
-                ctx.session_key,
-                "running",
-                started_at=ctx.visible_run_started_at,
-            )
+        await ctx.delivery.running(started_at=ctx.visible_run_started_at)
         result = await self._run_agent_loop(
             ctx.initial_messages,
             runtime=ctx.runtime,
@@ -1607,8 +1515,8 @@ class AgentLoop:
             on_stream_end=ctx.on_stream_end,
             on_retry_wait=ctx.on_retry_wait,
             session=ctx.session,
-            channel=ctx.route.channel,
-            chat_id=ctx.route.chat_id,
+            channel=ctx.delivery.route.channel,
+            chat_id=ctx.delivery.route.chat_id,
             message_id=ctx.msg.metadata.get("message_id"),
             metadata=ctx.msg.metadata,
             session_key=ctx.session_key,
@@ -1656,10 +1564,7 @@ class AgentLoop:
             ctx.session, ctx.all_messages, ctx.save_skip,
             turn_latency_ms=ctx.turn_latency_ms,
         )
-        self._runtime_events().record_turn_latency(
-            ctx.session_key,
-            ctx.turn_latency_ms,
-        )
+        ctx.delivery.record_latency(ctx.turn_latency_ms)
         if not ctx.ephemeral:
             ctx.session.enforce_file_cap(
                 on_archive=partial(self.context.memory.raw_archive, session_key=ctx.session_key)
@@ -1683,11 +1588,11 @@ class AgentLoop:
             ctx.outbound = None
             return "ok"
         if ctx.kind is TurnKind.SYSTEM:
-            ctx.outbound = OutboundMessage(
-                channel=ctx.route.channel,
-                chat_id=ctx.route.chat_id,
-                content=ctx.final_content or "Background task completed.",
-                metadata=dict(ctx.route.metadata),
+            ctx.outbound = ctx.delivery.background_response(
+                ctx.final_content,
+                stop_reason=ctx.stop_reason,
+                streamed=ctx.on_stream is not None,
+                latency_ms=ctx.turn_latency_ms,
             )
             return "ok"
         ctx.outbound = self._assemble_outbound(

nanobot/agent/turn_delivery.py

@@ -0,0 +1,300 @@
+"""Route and publish the user-visible lifecycle of an agent turn."""
+
+from __future__ import annotations
+
+import dataclasses
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+from typing import Any
+
+from nanobot.bus.events import InboundMessage, OutboundMessage
+from nanobot.bus.outbound_events import (
+    RetryWaitEvent,
+    StreamDeltaEvent,
+    StreamedResponseEvent,
+    StreamEndEvent,
+    outbound_message_for_event,
+)
+from nanobot.bus.progress import build_bus_progress_callback
+from nanobot.bus.queue import MessageBus
+from nanobot.bus.runtime_events import RuntimeEventBus, RuntimeEventPublisher
+
+
+@dataclass(frozen=True)
+class TurnRoute:
+    """Turn delivery destination and lifecycle policy, separate from execution input."""
+
+    channel: str
+    chat_id: str
+    metadata: dict[str, Any] = field(default_factory=dict)
+    publish_lifecycle: bool = False
+
+
+TurnRoutePolicy = Callable[[InboundMessage, str, TurnRoute], TurnRoute]
+ProgressCallback = Callable[..., Awaitable[None]]
+StreamCallback = Callable[[str], Awaitable[None]]
+StreamEndCallback = Callable[..., Awaitable[None]]
+RetryWaitCallback = Callable[[str], Awaitable[None]]
+
+
+class TurnDeliveryFactory:
+    """Create per-turn delivery objects from an optional edge-owned route policy."""
+
+    def __init__(
+        self,
+        bus: MessageBus,
+        runtime_events: RuntimeEventBus,
+        route_policy: TurnRoutePolicy | None = None,
+    ) -> None:
+        self.bus = bus
+        self.runtime_events = runtime_events
+        self.runtime_event_publisher = RuntimeEventPublisher(runtime_events)
+        self.route_policy = route_policy
+
+    def create(
+        self,
+        msg: InboundMessage,
+        session_key: str,
+        *,
+        enable_stream: bool = False,
+    ) -> TurnDelivery:
+        route = self._default_route(msg, session_key)
+        if self.route_policy is not None:
+            route = self.route_policy(msg, session_key, route)
+            if not isinstance(route, TurnRoute):
+                raise TypeError("turn route policy must return TurnRoute")
+        return TurnDelivery(
+            bus=self.bus,
+            runtime_event_publisher=self.runtime_event_publisher,
+            input_message=msg,
+            session_key=session_key,
+            route=route,
+            enable_stream=enable_stream,
+        )
+
+    def unrouted(self, msg: InboundMessage, session_key: str) -> TurnDelivery:
+        """Create a lifecycle fallback without invoking edge routing policy."""
+        return TurnDelivery(
+            bus=self.bus,
+            runtime_event_publisher=self.runtime_event_publisher,
+            input_message=msg,
+            session_key=session_key,
+            route=TurnRoute(
+                channel=msg.channel,
+                chat_id=msg.chat_id,
+                metadata=dict(msg.metadata or {}),
+            ),
+        )
+
+    @staticmethod
+    def _default_route(msg: InboundMessage, session_key: str) -> TurnRoute:
+        if msg.channel != "system":
+            return TurnRoute(
+                channel=msg.channel,
+                chat_id=msg.chat_id,
+                metadata=dict(msg.metadata or {}),
+                publish_lifecycle=True,
+            )
+
+        channel, chat_id = (
+            msg.chat_id.split(":", 1) if ":" in msg.chat_id else ("cli", msg.chat_id)
+        )
+        metadata: dict[str, Any] = {}
+        if (
+            channel == "slack"
+            and session_key.startswith("slack:")
+            and session_key.count(":") >= 2
+        ):
+            metadata["slack"] = {"thread_ts": session_key.split(":", 2)[2]}
+        if origin_message_id := msg.metadata.get("origin_message_id"):
+            metadata["origin_message_id"] = origin_message_id
+        return TurnRoute(channel=channel, chat_id=chat_id, metadata=metadata)
+
+
+@dataclass
+class TurnDelivery:
+    """Own routing, callbacks, and lifecycle publication for one turn."""
+
+    bus: MessageBus
+    runtime_event_publisher: RuntimeEventPublisher
+    input_message: InboundMessage
+    session_key: str
+    route: TurnRoute
+    enable_stream: bool = False
+    delivery_message: InboundMessage = field(init=False)
+    lifecycle_message: InboundMessage = field(init=False)
+    _stream_base_id: str | None = field(init=False, default=None)
+    _stream_segment: int = field(init=False, default=0)
+
+    def __post_init__(self) -> None:
+        self.delivery_message = dataclasses.replace(
+            self.input_message,
+            channel=self.route.channel,
+            chat_id=self.route.chat_id,
+            metadata=dict(self.route.metadata),
+        )
+        self.lifecycle_message = (
+            self.delivery_message if self.route.publish_lifecycle else self.input_message
+        )
+        if self.enable_stream and self.delivery_message.metadata.get("_wants_stream"):
+            self._stream_base_id = f"{self.session_key}:{time.time_ns()}"
+
+    @property
+    def on_stream(self) -> StreamCallback | None:
+        return self._publish_stream if self._stream_base_id is not None else None
+
+    @property
+    def on_stream_end(self) -> StreamEndCallback | None:
+        return self._publish_stream_end if self._stream_base_id is not None else None
+
+    def progress_callback(self) -> ProgressCallback | None:
+        if not self.route.publish_lifecycle:
+            return None
+        return build_bus_progress_callback(self.bus, self.delivery_message)
+
+    def retry_wait_callback(self) -> RetryWaitCallback | None:
+        if not self.route.publish_lifecycle:
+            return None
+
+        async def _on_retry_wait(content: str) -> None:
+            await self.bus.publish_outbound(
+                outbound_message_for_event(
+                    channel=self.delivery_message.channel,
+                    chat_id=self.delivery_message.chat_id,
+                    event=RetryWaitEvent(content=content),
+                    metadata=self.delivery_message.metadata,
+                )
+            )
+
+        return _on_retry_wait
+
+    async def started(self) -> None:
+        if self.route.publish_lifecycle:
+            await self.runtime_event_publisher.session_turn_started(
+                self.delivery_message,
+                self.session_key,
+            )
+
+    async def running(self, *, started_at: float) -> None:
+        if self.route.publish_lifecycle:
+            await self.runtime_event_publisher.run_status_changed(
+                self.delivery_message,
+                self.session_key,
+                "running",
+                started_at=started_at,
+            )
+
+    def record_runtime(self, runtime: Any) -> None:
+        self.runtime_event_publisher.record_turn_runtime(self.session_key, runtime)
+
+    def record_latency(self, latency_ms: int | None) -> None:
+        self.runtime_event_publisher.record_turn_latency(self.session_key, latency_ms)
+
+    def background_response(
+        self,
+        content: str | None,
+        *,
+        stop_reason: str,
+        streamed: bool,
+        latency_ms: int | None,
+    ) -> OutboundMessage:
+        metadata = dict(self.route.metadata)
+        if self.route.publish_lifecycle and latency_ms is not None:
+            metadata["latency_ms"] = int(latency_ms)
+        event = (
+            StreamedResponseEvent()
+            if self.route.publish_lifecycle
+            and streamed
+            and stop_reason not in {"error", "tool_error"}
+            else None
+        )
+        return OutboundMessage(
+            channel=self.route.channel,
+            chat_id=self.route.chat_id,
+            content=content or "Background task completed.",
+            metadata=metadata,
+            event=event,
+        )
+
+    async def complete(
+        self,
+        response: OutboundMessage | None,
+        *,
+        publish_completion: bool,
+    ) -> None:
+        completed_channel = self.lifecycle_message.channel
+        completed_chat_id = self.lifecycle_message.chat_id
+        if response is not None:
+            await self.bus.publish_outbound(response)
+            completed_channel = response.channel
+            completed_chat_id = response.chat_id
+        elif self.lifecycle_message.channel == "cli":
+            await self.bus.publish_outbound(
+                OutboundMessage(
+                    channel=self.lifecycle_message.channel,
+                    chat_id=self.lifecycle_message.chat_id,
+                    content="",
+                    metadata=dict(self.lifecycle_message.metadata or {}),
+                )
+            )
+        if publish_completion:
+            await self.runtime_event_publisher.turn_completed(
+                channel=completed_channel,
+                chat_id=completed_chat_id,
+                session_key=self.session_key,
+                metadata=self.lifecycle_message.metadata,
+            )
+
+    async def fail(self, *, publish_completion: bool) -> None:
+        await self.bus.publish_outbound(
+            OutboundMessage(
+                channel=self.lifecycle_message.channel,
+                chat_id=self.lifecycle_message.chat_id,
+                content="Sorry, I encountered an error.",
+                metadata=dict(self.lifecycle_message.metadata or {}),
+            )
+        )
+        if publish_completion:
+            await self.runtime_event_publisher.turn_completed(
+                channel=self.lifecycle_message.channel,
+                chat_id=self.lifecycle_message.chat_id,
+                session_key=self.session_key,
+                metadata=self.lifecycle_message.metadata,
+            )
+
+    async def idle(self) -> None:
+        await self.runtime_event_publisher.run_status_changed(
+            self.lifecycle_message,
+            self.session_key,
+            "idle",
+        )
+        self.runtime_event_publisher.clear_turn(self.session_key)
+
+    def _stream_id(self) -> str:
+        assert self._stream_base_id is not None
+        return f"{self._stream_base_id}:{self._stream_segment}"
+
+    async def _publish_stream(self, delta: str) -> None:
+        await self.bus.publish_outbound(
+            outbound_message_for_event(
+                channel=self.delivery_message.channel,
+                chat_id=self.delivery_message.chat_id,
+                event=StreamDeltaEvent(content=delta, stream_id=self._stream_id()),
+                metadata=self.delivery_message.metadata,
+            )
+        )
+
+    async def _publish_stream_end(self, *, resuming: bool = False) -> None:
+        await self.bus.publish_outbound(
+            outbound_message_for_event(
+                channel=self.delivery_message.channel,
+                chat_id=self.delivery_message.chat_id,
+                event=StreamEndEvent(
+                    stream_id=self._stream_id(),
+                    resuming=resuming,
+                ),
+                metadata=self.delivery_message.metadata,
+            )
+        )
+        self._stream_segment += 1

nanobot/cli/commands.py

@@ -1606,6 +1606,7 @@ def _run_gateway(
 ) -> None:
     """Shared gateway runtime; ``open_browser_url`` opens a tab once channels are up."""
     from nanobot.agent.tools.message import MessageTool
+    from nanobot.agent.turn_delivery import TurnDeliveryFactory
     from nanobot.bus.queue import MessageBus
     from nanobot.bus.runtime_events import RuntimeEventBus
     from nanobot.channels.manager import ChannelManager
@@ -1617,7 +1618,7 @@ def _run_gateway(
     from nanobot.providers.factory import build_provider_snapshot, load_provider_snapshot
     from nanobot.providers.image_generation import image_gen_provider_configs
     from nanobot.session.manager import SessionManager
-    from nanobot.session.webui_turns import WebuiTurnCoordinator
+    from nanobot.session.webui_turns import WebuiTurnCoordinator, WebuiTurnRoutePolicy
     from nanobot.triggers.local_runner import run_local_trigger_queue
     from nanobot.triggers.local_store import LocalTriggerStore
     from nanobot.webui.token_usage import TokenUsageHook
@@ -1679,6 +1680,12 @@ def _run_gateway(
     cron = CronService(cron_store_path)
     trigger_store = LocalTriggerStore(config.workspace_path)
 
+    turn_delivery_factory = TurnDeliveryFactory(
+        bus,
+        runtime_events,
+        route_policy=WebuiTurnRoutePolicy(session_manager),
+    )
+
     # Create agent with cron service
     agent = AgentLoop.from_config(
         config, bus,
@@ -1690,16 +1697,18 @@ def _run_gateway(
         image_generation_provider_configs=image_gen_provider_configs(config),
         provider_snapshot_loader=load_provider_snapshot,
         runtime_events=runtime_events,
+        turn_delivery_factory=turn_delivery_factory,
         provider_signature=provider_snapshot.signature,
         hooks=[TokenUsageHook(timezone_name=config.agents.defaults.timezone)],
         local_trigger_store=trigger_store,
         hook_factories=[create_file_edit_activity_hook],
     )
-    WebuiTurnCoordinator(
+    webui_turn_coordinator = WebuiTurnCoordinator(
         bus=bus,
         sessions=session_manager,
         schedule_background=lambda coro: agent._schedule_background(coro),
-    ).subscribe(runtime_events)
+    )
+    webui_turn_coordinator.subscribe(runtime_events)
     from nanobot.bus.events import OutboundMessage
     from nanobot.session.keys import session_key_for_channel
 

nanobot/session/turn_continuation.py

@@ -148,7 +148,7 @@ def prepare_save_boundary(ctx: Any) -> None:
         message_metadata=ctx.msg.metadata,
         initial_message_count=len(ctx.initial_messages),
         history_count=len(ctx.history),
-        user_persisted_early=ctx.user_persisted_early,
+        input_persisted_early=ctx.input_persisted_early,
     )
 
 
@@ -183,7 +183,7 @@ def _save_skip_for_turn(
     message_metadata: Mapping[str, Any] | None,
     initial_message_count: int,
     history_count: int,
-    user_persisted_early: bool,
+    input_persisted_early: bool,
 ) -> int:
     """Return the persisted-message append boundary for this turn."""
     if message_metadata and message_metadata.get(SKIP_USER_PERSIST_META) is True:
@@ -193,7 +193,7 @@ def _save_skip_for_turn(
     # build_messages may merge the current message into a same-role history tail.
     # Runner-appended messages start at initial_message_count in either shape.
     has_standalone_current = initial_message_count > 1 + history_count
-    if has_standalone_current and not user_persisted_early:
+    if has_standalone_current and not input_persisted_early:
         return initial_message_count - 1
     return initial_message_count
 

nanobot/session/webui_turns.py

@@ -5,11 +5,13 @@ from __future__ import annotations
 import re
 import time
 from collections.abc import Awaitable, Callable
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from typing import Any
+from uuid import uuid4
 
 from loguru import logger
 
+from nanobot.agent.turn_delivery import TurnRoute
 from nanobot.bus import progress as bus_progress
 from nanobot.bus.events import InboundMessage
 from nanobot.bus.outbound_events import (
@@ -37,6 +39,7 @@ from nanobot.session.history_visibility import is_hidden_history_message
 from nanobot.session.manager import Session, SessionManager
 from nanobot.utils.helpers import strip_think, truncate_text
 from nanobot.utils.llm_runtime import LLMRuntime
+from nanobot.webui.metadata import WEBUI_TURN_METADATA_KEY
 
 WEBUI_SESSION_METADATA_KEY = "webui"
 WEBUI_TITLE_METADATA_KEY = "title"
@@ -235,6 +238,40 @@ async def publish_turn_run_status(
         ),
     )
 
+@dataclass(frozen=True)
+class WebuiTurnRoutePolicy:
+    """Expose independently dispatched late subagent turns to WebUI sessions."""
+
+    sessions: SessionManager
+
+    def __call__(
+        self,
+        msg: InboundMessage,
+        session_key: str,
+        route: TurnRoute,
+    ) -> TurnRoute:
+        """Make an independently dispatched late subagent result visible in WebUI."""
+        if (
+            msg.channel != "system"
+            or msg.sender_id != "subagent"
+            or msg.metadata.get("injected_event") != "subagent_result"
+            or route.channel != "websocket"
+        ):
+            return route
+
+        session = self.sessions.get_or_create(session_key)
+        if session.metadata.get(WEBUI_SESSION_METADATA_KEY) is not True:
+            return route
+
+        metadata = dict(route.metadata)
+        metadata.update({
+            WEBUI_SESSION_METADATA_KEY: True,
+            "_wants_stream": True,
+            WEBUI_TURN_METADATA_KEY: f"subagent:{uuid4().hex}",
+        })
+        return replace(route, metadata=metadata, publish_lifecycle=True)
+
+
 @dataclass
 class WebuiTurnCoordinator:
     """Translate generic runtime events into WebUI/WebSocket wire messages."""

tests/agent/test_document_extraction_toggle.py

@@ -5,7 +5,7 @@ from unittest.mock import AsyncMock, MagicMock
 
 import pytest
 
-from nanobot.agent.loop import AgentLoop, TurnContext, TurnKind, TurnRoute, TurnState
+from nanobot.agent.loop import AgentLoop, TurnContext, TurnKind, TurnState
 from nanobot.bus.events import InboundMessage
 from nanobot.bus.queue import MessageBus
 from nanobot.config.schema import ChannelsConfig
@@ -42,20 +42,21 @@ async def test_state_restore_extracts_documents_by_default(
 
     monkeypatch.setattr("nanobot.agent.loop.extract_documents", fake_extract_documents)
 
+    msg = InboundMessage(
+        channel="cli",
+        sender_id="u",
+        chat_id="c",
+        content="summarize",
+        media=[str(doc_path)],
+    )
     ctx = TurnContext(
-        msg=InboundMessage(
-            channel="cli",
-            sender_id="u",
-            chat_id="c",
-            content="summarize",
-            media=[str(doc_path)],
-        ),
+        msg=msg,
         session_key="cli:c",
         state=TurnState.RESTORE,
         turn_id="turn-1",
         runtime=loop.llm_runtime(),
         kind=TurnKind.USER,
-        route=TurnRoute(channel="cli", chat_id="c"),
+        delivery=loop.turn_delivery_factory.create(msg, "cli:c"),
     )
 
     assert await loop._state_restore(ctx) == "ok"
@@ -79,20 +80,21 @@ async def test_state_restore_references_documents_when_extraction_disabled(
 
     monkeypatch.setattr("nanobot.agent.loop.extract_documents", fail_extract_documents)
 
+    msg = InboundMessage(
+        channel="cli",
+        sender_id="u",
+        chat_id="c",
+        content="summarize",
+        media=[str(doc_path)],
+    )
     ctx = TurnContext(
-        msg=InboundMessage(
-            channel="cli",
-            sender_id="u",
-            chat_id="c",
-            content="summarize",
-            media=[str(doc_path)],
-        ),
+        msg=msg,
         session_key="cli:c",
         state=TurnState.RESTORE,
         turn_id="turn-1",
         runtime=loop.llm_runtime(),
         kind=TurnKind.USER,
-        route=TurnRoute(channel="cli", chat_id="c"),
+        delivery=loop.turn_delivery_factory.create(msg, "cli:c"),
     )
 
     assert await loop._state_restore(ctx) == "ok"

tests/agent/test_loop_progress.py

@@ -8,6 +8,7 @@ import pytest
 
 from nanobot.agent.hooks import create_file_edit_activity_hook
 from nanobot.agent.loop import AgentLoop
+from nanobot.agent.tools.context import current_request_context
 from nanobot.agent.tools.filesystem import WriteFileTool
 from nanobot.bus.events import InboundMessage
 from nanobot.bus.outbound_events import (
@@ -22,11 +23,12 @@ from nanobot.bus.outbound_events import (
 from nanobot.bus.queue import MessageBus
 from nanobot.providers.base import LLMResponse, ToolCallRequest
 from nanobot.providers.factory import ProviderSnapshot
-from nanobot.session.webui_turns import WebuiTurnCoordinator
+from nanobot.session.webui_turns import WebuiTurnCoordinator, WebuiTurnRoutePolicy
 from nanobot.utils.progress_events import (
     invoke_file_edit_progress,
     on_progress_accepts_file_edit_events,
 )
+from nanobot.webui.metadata import WEBUI_TURN_METADATA_KEY
 
 
 def _make_loop(tmp_path: Path) -> AgentLoop:
@@ -43,6 +45,7 @@ def _make_loop(tmp_path: Path) -> AgentLoop:
 
 
 def _attach_webui_runtime_events(loop: AgentLoop, bus: MessageBus) -> None:
+    loop.turn_delivery_factory.route_policy = WebuiTurnRoutePolicy(loop.sessions)
     coordinator = WebuiTurnCoordinator(
         bus=bus,
         sessions=loop.sessions,
@@ -347,12 +350,14 @@ class TestToolEventProgress:
             "status": "editing",
         }]
 
-        progress = await loop._build_bus_progress_callback(InboundMessage(
+        msg = InboundMessage(
             channel="telegram",
             sender_id="u1",
             chat_id="chat1",
             content="edit",
-        ))
+        )
+        progress = loop.turn_delivery_factory.create(msg, msg.session_key).progress_callback()
+        assert progress is not None
         assert on_progress_accepts_file_edit_events(progress) is True
         await invoke_file_edit_progress(progress, edit_events)
         outbound = await bus.consume_outbound()
@@ -529,6 +534,157 @@ class TestToolEventProgress:
         assert turn_end_msgs[0].content == ""
         provider.chat_with_retry.assert_not_awaited()
 
+    @pytest.mark.asyncio
+    async def test_independent_late_subagent_result_gets_complete_webui_turn(
+        self,
+        tmp_path: Path,
+    ) -> None:
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.supports_progress_deltas = True
+        provider.get_default_model.return_value = "openai-codex/gpt-5.5"
+        first_request_started = asyncio.Event()
+        release_first_request = asyncio.Event()
+        requests: list[list[dict]] = []
+        request_contexts = []
+        tool_call = ToolCallRequest(id="call1", name="custom_tool", arguments={})
+        responses = iter([
+            LLMResponse(content="Checking", tool_calls=[tool_call]),
+            LLMResponse(content="The late result is ready", tool_calls=[]),
+        ])
+
+        async def chat_stream_with_retry(*, messages, on_content_delta, **kwargs):
+            requests.append([dict(message) for message in messages])
+            response = next(responses)
+            if len(requests) == 1:
+                first_request_started.set()
+                await release_first_request.wait()
+            await on_content_delta(response.content or "")
+            return response
+
+        provider.chat_stream_with_retry = chat_stream_with_retry
+        provider.chat_with_retry = AsyncMock()
+        loop = AgentLoop(
+            bus=bus,
+            provider=provider,
+            workspace=tmp_path,
+            model="openai-codex/gpt-5.5",
+        )
+        _attach_webui_runtime_events(loop, bus)
+        loop.tools.get_definitions = MagicMock(return_value=[])
+        loop.tools.prepare_call = MagicMock(return_value=(None, {}, None))
+
+        async def execute_tool(*args, **kwargs):
+            request_contexts.append(current_request_context())
+            return "ok"
+
+        loop.tools.execute = execute_tool
+        loop.consolidator.maybe_consolidate_by_tokens = AsyncMock(  # type: ignore[method-assign]
+            return_value=False
+        )
+
+        session_key = "websocket:chat-a"
+        session = loop.sessions.get_or_create(session_key)
+        session.add_message("user", "Run this in the background")
+        session.metadata.update({"webui": True, "title": "Existing title"})
+        loop.sessions.save(session)
+        dispatch = asyncio.create_task(loop._dispatch(InboundMessage(
+            channel="system",
+            sender_id="subagent",
+            chat_id=session_key,
+            content="Background research completed",
+            session_key_override=session_key,
+            metadata={
+                "injected_event": "subagent_result",
+                "subagent_task_id": "sub-1",
+            },
+        )))
+
+        await asyncio.wait_for(first_request_started.wait(), timeout=1)
+        await loop._pending_queues[session_key].put(InboundMessage(
+            channel="websocket",
+            sender_id="user",
+            chat_id="chat-a",
+            content="Can you include the key detail?",
+            session_key_override=session_key,
+        ))
+        release_first_request.set()
+        await asyncio.wait_for(dispatch, timeout=2)
+
+        outbound = []
+        while bus.outbound_size > 0:
+            outbound.append(await bus.consume_outbound())
+
+        assert len(requests) == 2
+        assert requests[0][-1]["role"] == "user"
+        assert requests[0][-1]["content"].endswith("Background research completed")
+        assert any(
+            message.get("role") == "user"
+            and message.get("content") == "Can you include the key detail?"
+            for message in requests[1]
+        )
+        assert len(request_contexts) == 1
+        request_ctx = request_contexts[0]
+        assert request_ctx is not None
+        assert request_ctx.metadata == {
+            "injected_event": "subagent_result",
+            "subagent_task_id": "sub-1",
+        }
+        statuses = [
+            message.event.status
+            for message in outbound
+            if isinstance(message.event, GoalStatusEvent)
+        ]
+        assert statuses == ["running", "idle"]
+        assert [
+            message.content
+            for message in outbound
+            if isinstance(message.event, StreamDeltaEvent)
+        ] == ["Checking", "The late result is ready"]
+        assert any(isinstance(message.event, ProgressEvent) for message in outbound)
+        assert len([
+            message for message in outbound if isinstance(message.event, TurnEndEvent)
+        ]) == 1
+        assert len([
+            message for message in outbound
+            if isinstance(message.event, StreamedResponseEvent)
+        ]) == 1
+        visible_events = [
+            message
+            for message in outbound
+            if isinstance(
+                message.event,
+                GoalStatusEvent
+                | ProgressEvent
+                | StreamDeltaEvent
+                | StreamEndEvent
+                | StreamedResponseEvent
+                | TurnEndEvent,
+            )
+        ]
+        assert visible_events
+        turn_ids = {
+            message.metadata.get(WEBUI_TURN_METADATA_KEY)
+            for message in visible_events
+        }
+        assert len(turn_ids) == 1
+        turn_id = turn_ids.pop()
+        assert isinstance(turn_id, str)
+        assert turn_id.startswith("subagent:")
+        assert all(
+            (message.channel, message.chat_id) == ("websocket", "chat-a")
+            and message.metadata.get("webui") is True
+            and message.metadata.get("_wants_stream") is True
+            and set(message.metadata) <= {
+                "webui",
+                "_wants_stream",
+                WEBUI_TURN_METADATA_KEY,
+                "latency_ms",
+            }
+            for message in visible_events
+        )
+        provider.chat_with_retry.assert_not_awaited()
+
     @pytest.mark.asyncio
     async def test_stream_timeout_recovery_continues_in_new_segment(
         self,

tests/agent/test_loop_save_turn.py

@@ -1424,6 +1424,7 @@ async def test_system_subagent_followup_is_persisted_before_prompt_assembly(tmp_
     # rewritten with volatile ``[Message Time: ...]`` prefixes.
     assert "[Message Time:" not in non_system[0]["content"]
     assert "[Message Time:" not in non_system[1]["content"]
+    assert non_system[2]["role"] == "user"
     assert non_system[2]["content"].count("subagent result") == 1
     assert non_system[2]["content"] == "subagent result"
 
@@ -1580,10 +1581,11 @@ async def test_multiple_subagent_followups_all_persist_as_standalone_history(tmp
     ]
 
 
-def test_prompt_merge_does_not_replace_standalone_subagent_history_entry(tmp_path: Path) -> None:
+def test_subagent_followup_uses_user_model_input_and_assistant_history(tmp_path: Path) -> None:
     loop = _mk_loop()
     session = Session(key="cli:merge")
     session.add_message("assistant", "previous assistant")
+    history = session.get_history(max_messages=0)
 
     inserted = loop._persist_subagent_followup(
         session,
@@ -1600,16 +1602,18 @@ def test_prompt_merge_does_not_replace_standalone_subagent_history_entry(tmp_pat
 
     builder = ContextBuilder(tmp_path)
     projected = builder.build_messages(
-        history=session.get_history(max_messages=0),
-        current_message="",
-        current_role="assistant",
+        history=history,
+        current_message="subagent result",
+        current_role="user",
         channel="cli",
         chat_id="merge",
     )
 
     non_system = [m for m in projected if m.get("role") != "system"]
     assert len(non_system) == 2
+    assert non_system[-1]["role"] == "user"
     assert "subagent result" in non_system[-1]["content"]
+    assert session.messages[-1]["role"] == "assistant"
     assert session.messages[-1]["content"] == "subagent result"
     assert session.messages[-1]["injected_event"] == "subagent_result"
 

tests/agent/test_runner_injections.py

@@ -689,6 +689,8 @@ async def test_waiting_dispatch_does_not_replace_active_pending_queue(tmp_path):
     from nanobot.bus.events import InboundMessage
 
     loop = _make_loop(tmp_path)
+    route_policy = MagicMock(side_effect=lambda _msg, _key, route: route)
+    loop.turn_delivery_factory.route_policy = route_policy
     session_key = "cli:c"
     lock = loop._session_locks.setdefault(session_key, asyncio.Lock())
     await lock.acquire()
@@ -712,10 +714,12 @@ async def test_waiting_dispatch_does_not_replace_active_pending_queue(tmp_path):
         await asyncio.wait_for(waiting_at_lock.wait(), timeout=2.0)
 
     assert loop._pending_queues[session_key] is active_pending
+    route_policy.assert_not_called()
 
     waiting.cancel()
     with pytest.raises(asyncio.CancelledError):
         await waiting
+    route_policy.assert_not_called()
     lock.release()
 
 
@@ -746,6 +750,44 @@ async def test_followup_routed_to_pending_queue(tmp_path):
     assert queued_msg.session_key == UNIFIED_SESSION_KEY
 
 
+@pytest.mark.asyncio
+async def test_mid_turn_subagent_result_does_not_resolve_a_new_turn_route(tmp_path):
+    """Injected results stay inside the active turn instead of opening a side turn."""
+    from nanobot.bus.events import InboundMessage
+
+    loop = _make_loop(tmp_path)
+    loop._dispatch = AsyncMock()  # type: ignore[method-assign]
+    route_policy = MagicMock(side_effect=lambda _msg, _key, route: route)
+    loop.turn_delivery_factory.route_policy = route_policy
+
+    session_key = "websocket:chat-1"
+    pending = asyncio.Queue(maxsize=20)
+    loop._pending_queues[session_key] = pending
+
+    run_task = asyncio.create_task(loop.run())
+    msg = InboundMessage(
+        channel="system",
+        sender_id="subagent",
+        chat_id=session_key,
+        content="background result",
+        metadata={
+            "injected_event": "subagent_result",
+            "subagent_task_id": "sub-1",
+        },
+        session_key_override=session_key,
+    )
+    await loop.bus.publish_inbound(msg)
+
+    queued_msg = await asyncio.wait_for(pending.get(), timeout=2)
+
+    loop.stop()
+    await asyncio.wait_for(run_task, timeout=2)
+
+    assert queued_msg is msg
+    assert loop._dispatch.await_count == 0
+    route_policy.assert_not_called()
+
+
 @pytest.mark.asyncio
 async def test_cron_turn_deferred_while_session_active(tmp_path):
     """Cron turns wait for the active session instead of becoming injections."""

tests/agent/test_turn_delivery.py

@@ -0,0 +1,60 @@
+from pathlib import Path
+
+from nanobot.agent.turn_delivery import TurnDeliveryFactory
+from nanobot.bus.events import InboundMessage
+from nanobot.bus.queue import MessageBus
+from nanobot.bus.runtime_events import RuntimeEventBus
+from nanobot.session.manager import SessionManager
+from nanobot.session.webui_turns import WebuiTurnRoutePolicy
+from nanobot.webui.metadata import WEBUI_TURN_METADATA_KEY
+
+
+def test_late_subagent_route_requires_webui_owned_session(tmp_path: Path) -> None:
+    sessions = SessionManager(tmp_path)
+    factory = TurnDeliveryFactory(
+        MessageBus(),
+        RuntimeEventBus(),
+        route_policy=WebuiTurnRoutePolicy(sessions),
+    )
+    session_key = "websocket:chat-a"
+    msg = InboundMessage(
+        channel="system",
+        sender_id="subagent",
+        chat_id=session_key,
+        content="Background research completed",
+        session_key_override=session_key,
+        metadata={
+            "injected_event": "subagent_result",
+            "subagent_task_id": "sub-1",
+        },
+    )
+
+    hidden_route = factory.create(msg, session_key).route
+
+    assert hidden_route.channel == "websocket"
+    assert hidden_route.chat_id == "chat-a"
+    assert hidden_route.metadata == {}
+    assert hidden_route.publish_lifecycle is False
+
+    session = sessions.get_or_create(session_key)
+    session.metadata["webui"] = True
+    first_visible_route = factory.create(msg, session_key).route
+    second_visible_route = factory.create(msg, session_key).route
+
+    assert first_visible_route.publish_lifecycle is True
+    assert set(first_visible_route.metadata) == {
+        "webui",
+        "_wants_stream",
+        WEBUI_TURN_METADATA_KEY,
+    }
+    assert first_visible_route.metadata["webui"] is True
+    assert first_visible_route.metadata["_wants_stream"] is True
+    first_turn_id = first_visible_route.metadata[WEBUI_TURN_METADATA_KEY]
+    second_turn_id = second_visible_route.metadata[WEBUI_TURN_METADATA_KEY]
+    assert first_turn_id.startswith("subagent:")
+    assert second_turn_id.startswith("subagent:")
+    assert first_turn_id != second_turn_id
+    assert msg.metadata == {
+        "injected_event": "subagent_result",
+        "subagent_task_id": "sub-1",
+    }

tests/cli/test_commands.py

@@ -13,6 +13,7 @@ import pytest
 from typer.testing import CliRunner
 
 from nanobot.agent.memory import MemoryStore
+from nanobot.agent.turn_delivery import TurnDeliveryFactory
 from nanobot.bus.events import InboundMessage, OutboundMessage
 from nanobot.cli import commands as cli_commands
 from nanobot.cli.commands import app
@@ -24,6 +25,7 @@ from nanobot.cron.webui_metadata import cron_proactive_delivery_metadata
 from nanobot.providers.factory import ProviderSnapshot, make_provider, provider_signature
 from nanobot.providers.openai_codex_provider import _strip_model_prefix
 from nanobot.providers.registry import find_by_name
+from nanobot.session.webui_turns import WebuiTurnRoutePolicy
 from nanobot.webui.metadata import (
     WEBUI_MESSAGE_SOURCE_METADATA_KEY,
     WEBUI_TURN_METADATA_KEY,
@@ -2790,6 +2792,11 @@ def test_gateway_local_trigger_queue_submits_agent_turns(
     assert kwargs["submit_turn"] is agent.submit_local_trigger_turn
     assert kwargs["is_channel_enabled"]("websocket") is True
     assert kwargs["is_channel_enabled"]("telegram") is False
+    turn_delivery_factory = agent_kwargs["turn_delivery_factory"]
+    assert isinstance(turn_delivery_factory, TurnDeliveryFactory)
+    assert turn_delivery_factory.bus is bus
+    assert isinstance(turn_delivery_factory.route_policy, WebuiTurnRoutePolicy)
+    assert turn_delivery_factory.route_policy.sessions is agent.sessions
 
 
 def test_gateway_workspace_override_does_not_migrate_legacy_cron(

tests/session/test_turn_continuation.py

@@ -146,7 +146,7 @@ def test_save_skip_matches_prefix_when_current_message_merged():
         message_metadata=None,
         initial_message_count=2,  # [system, merged user]
         history_count=1,
-        user_persisted_early=True,
+        input_persisted_early=True,
     )
     assert skip == 2
 
@@ -157,11 +157,11 @@ def test_save_skip_unchanged_for_standalone_current_message():
         message_metadata=None,
         initial_message_count=3,
         history_count=1,
-        user_persisted_early=True,
+        input_persisted_early=True,
     ) == 3
     assert _save_skip_for_turn(
         message_metadata=None,
         initial_message_count=3,
         history_count=1,
-        user_persisted_early=False,
+        input_persisted_early=False,
     ) == 2