From 66690fdb0cd2a217c6ebd08438957a97dfd2d43d Mon Sep 17 00:00:00 2001 From: chengyongru <61816729+chengyongru@users.noreply.github.com> Date: Wed, 22 Jul 2026 23:04:36 +0800 Subject: [PATCH] fix(webui): deliver late subagent results as new turns (#4992) --- nanobot/agent/loop.py | 273 ++++++---------- nanobot/agent/turn_delivery.py | 300 ++++++++++++++++++ nanobot/cli/commands.py | 15 +- nanobot/session/turn_continuation.py | 6 +- nanobot/session/webui_turns.py | 39 ++- .../agent/test_document_extraction_toggle.py | 36 ++- tests/agent/test_loop_progress.py | 162 +++++++++- tests/agent/test_loop_save_turn.py | 12 +- tests/agent/test_runner_injections.py | 42 +++ tests/agent/test_turn_delivery.py | 60 ++++ tests/cli/test_commands.py | 7 + tests/session/test_turn_continuation.py | 6 +- 12 files changed, 740 insertions(+), 218 deletions(-) create mode 100644 nanobot/agent/turn_delivery.py create mode 100644 tests/agent/test_turn_delivery.py diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index ab353e4f..9328e28b 100644 --- a/nanobot/agent/loop.py +++ b/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( diff --git a/nanobot/agent/turn_delivery.py b/nanobot/agent/turn_delivery.py new file mode 100644 index 00000000..2eeb5aab --- /dev/null +++ b/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 diff --git a/nanobot/cli/commands.py b/nanobot/cli/commands.py index 963584fe..baaac49c 100644 --- a/nanobot/cli/commands.py +++ b/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 diff --git a/nanobot/session/turn_continuation.py b/nanobot/session/turn_continuation.py index 483f7fc0..97d9c941 100644 --- a/nanobot/session/turn_continuation.py +++ b/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 diff --git a/nanobot/session/webui_turns.py b/nanobot/session/webui_turns.py index 1de44213..d8683aed 100644 --- a/nanobot/session/webui_turns.py +++ b/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.""" diff --git a/tests/agent/test_document_extraction_toggle.py b/tests/agent/test_document_extraction_toggle.py index 0be59f96..29f8d6c1 100644 --- a/tests/agent/test_document_extraction_toggle.py +++ b/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" diff --git a/tests/agent/test_loop_progress.py b/tests/agent/test_loop_progress.py index d1ddbc98..3ea6d953 100644 --- a/tests/agent/test_loop_progress.py +++ b/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, diff --git a/tests/agent/test_loop_save_turn.py b/tests/agent/test_loop_save_turn.py index b09d8324..a87600a5 100644 --- a/tests/agent/test_loop_save_turn.py +++ b/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" diff --git a/tests/agent/test_runner_injections.py b/tests/agent/test_runner_injections.py index 8a8ad912..b3c39fc6 100644 --- a/tests/agent/test_runner_injections.py +++ b/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.""" diff --git a/tests/agent/test_turn_delivery.py b/tests/agent/test_turn_delivery.py new file mode 100644 index 00000000..24988a20 --- /dev/null +++ b/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", + } diff --git a/tests/cli/test_commands.py b/tests/cli/test_commands.py index cdb2fc40..9843f6d2 100644 --- a/tests/cli/test_commands.py +++ b/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( diff --git a/tests/session/test_turn_continuation.py b/tests/session/test_turn_continuation.py index 6eed8e6d..862d3415 100644 --- a/tests/session/test_turn_continuation.py +++ b/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