From 03be51ade5951b8b0a3b4b34caebdfa7547fa000 Mon Sep 17 00:00:00 2001 From: Xubin Ren <52506698+Re-bin@users.noreply.github.com> Date: Wed, 1 Jul 2026 18:52:43 +0800 Subject: [PATCH] fix(channels): preserve legacy stream hook signatures --- nanobot/channels/manager.py | 111 ++++++++++++++++----- tests/channels/test_channel_plugins.py | 130 +++++++++++++++++++++++++ 2 files changed, 216 insertions(+), 25 deletions(-) diff --git a/nanobot/channels/manager.py b/nanobot/channels/manager.py index 5a2d0634..b2ac69b8 100644 --- a/nanobot/channels/manager.py +++ b/nanobot/channels/manager.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import hashlib +import inspect from collections.abc import Callable from contextlib import suppress from pathlib import Path @@ -381,23 +382,95 @@ class ChannelManager: except asyncio.CancelledError: break + @staticmethod + def _accepts_keyword(callable_obj: Callable[..., Any], name: str) -> bool: + try: + signature = inspect.signature(callable_obj) + except (TypeError, ValueError): + return True + return any( + parameter.kind is inspect.Parameter.VAR_KEYWORD or parameter.name == name + for parameter in signature.parameters.values() + ) + + @classmethod + async def _send_reasoning_delta(cls, channel: BaseChannel, msg: OutboundMessage, event: ProgressEvent) -> None: + metadata = msg.metadata + kwargs: dict[str, Any] = {} + if cls._accepts_keyword(channel.send_reasoning_delta, "stream_id"): + kwargs["stream_id"] = event.stream_id + else: + metadata = dict(metadata or {}) + metadata["_reasoning_delta"] = True + if event.stream_id is not None: + metadata["_stream_id"] = event.stream_id + await channel.send_reasoning_delta( + msg.chat_id, + msg.content, + metadata, + **kwargs, + ) + + @classmethod + async def _send_reasoning_end(cls, channel: BaseChannel, msg: OutboundMessage, event: ProgressEvent) -> None: + metadata = msg.metadata + kwargs: dict[str, Any] = {} + if cls._accepts_keyword(channel.send_reasoning_end, "stream_id"): + kwargs["stream_id"] = event.stream_id + else: + metadata = dict(metadata or {}) + metadata["_reasoning_end"] = True + if event.stream_id is not None: + metadata["_stream_id"] = event.stream_id + await channel.send_reasoning_end( + msg.chat_id, + metadata, + **kwargs, + ) + + @classmethod + async def _send_stream_event( + cls, + channel: BaseChannel, + msg: OutboundMessage, + event: StreamDeltaEvent | StreamEndEvent, + ) -> None: + metadata = msg.metadata + kwargs: dict[str, Any] = {} + if cls._accepts_keyword(channel.send_delta, "stream_id"): + kwargs["stream_id"] = event.stream_id + else: + metadata = dict(metadata or {}) + if event.stream_id is not None: + metadata["_stream_id"] = event.stream_id + + if isinstance(event, StreamEndEvent): + if cls._accepts_keyword(channel.send_delta, "stream_end"): + kwargs["stream_end"] = True + else: + metadata = dict(metadata or {}) + metadata["_stream_end"] = True + if cls._accepts_keyword(channel.send_delta, "resuming"): + kwargs["resuming"] = event.resuming + elif not kwargs: + metadata = dict(metadata or {}) + metadata["_stream_delta"] = True + + await channel.send_delta( + msg.chat_id, + msg.content, + metadata, + **kwargs, + ) + @staticmethod async def _send_once(channel: BaseChannel, msg: OutboundMessage) -> None: """Send one outbound message without retry policy.""" event = outbound_event_from_message(msg) if isinstance(event, ProgressEvent) and event.reasoning_end: - await channel.send_reasoning_end( - msg.chat_id, - msg.metadata, - stream_id=event.stream_id, - ) + await ChannelManager._send_reasoning_end(channel, msg, event) elif isinstance(event, ProgressEvent) and event.reasoning_delta: - await channel.send_reasoning_delta( - msg.chat_id, - msg.content, - msg.metadata, - stream_id=event.stream_id, - ) + await ChannelManager._send_reasoning_delta(channel, msg, event) elif isinstance(event, ProgressEvent) and event.reasoning: # BaseChannel translates one-shot reasoning to a single delta + # end pair so plugins only implement the streaming primitives. @@ -409,21 +482,9 @@ class ChannelManager: msg.metadata, ) elif isinstance(event, StreamDeltaEvent): - await channel.send_delta( - msg.chat_id, - msg.content, - msg.metadata, - stream_id=event.stream_id, - ) + await ChannelManager._send_stream_event(channel, msg, event) elif isinstance(event, StreamEndEvent): - await channel.send_delta( - msg.chat_id, - msg.content, - msg.metadata, - stream_id=event.stream_id, - stream_end=True, - resuming=event.resuming, - ) + await ChannelManager._send_stream_event(channel, msg, event) elif not isinstance(event, StreamedResponseEvent): await channel.send(msg) diff --git a/tests/channels/test_channel_plugins.py b/tests/channels/test_channel_plugins.py index ab8b3989..ccf8a4b8 100644 --- a/tests/channels/test_channel_plugins.py +++ b/tests/channels/test_channel_plugins.py @@ -10,8 +10,10 @@ import pytest from nanobot.bus.events import OutboundMessage from nanobot.bus.outbound_events import ( + ProgressEvent, StreamDeltaEvent, StreamedResponseEvent, + StreamEndEvent, outbound_message_for_event, ) from nanobot.bus.queue import MessageBus @@ -773,6 +775,134 @@ async def test_send_with_retry_calls_send_delta(): assert send_delta_called is True +@pytest.mark.asyncio +async def test_send_with_retry_supports_legacy_stream_delta_signature(): + """External plugins with the old send_delta signature should keep working.""" + calls: list[tuple[str, str, dict]] = [] + + class _LegacyStreamingChannel(BaseChannel): + name = "legacy_streaming" + display_name = "Legacy Streaming" + + async def start(self) -> None: + pass + + async def stop(self) -> None: + pass + + async def send(self, msg: OutboundMessage) -> None: + pass + + async def send_delta( + self, + chat_id: str, + delta: str, + metadata: dict | None = None, + ) -> None: + calls.append((chat_id, delta, dict(metadata or {}))) + + fake_config = SimpleNamespace( + channels=ChannelsConfig(send_max_retries=3), + providers=SimpleNamespace(groq=SimpleNamespace(api_key="")), + ) + mgr = ChannelManager.__new__(ChannelManager) + mgr.config = fake_config + mgr.bus = MessageBus() + mgr.channels = {"legacy_streaming": _LegacyStreamingChannel(fake_config, mgr.bus)} + mgr._dispatch_task = None + + await mgr._send_with_retry( + mgr.channels["legacy_streaming"], + outbound_message_for_event( + channel="legacy_streaming", + chat_id="123", + event=StreamDeltaEvent(content="hello", stream_id="s1"), + ), + ) + await mgr._send_with_retry( + mgr.channels["legacy_streaming"], + outbound_message_for_event( + channel="legacy_streaming", + chat_id="123", + event=StreamEndEvent(content="", stream_id="s1", resuming=True), + ), + ) + + assert calls == [ + ("123", "hello", {"_stream_id": "s1", "_stream_delta": True}), + ("123", "", {"_stream_id": "s1", "_stream_end": True}), + ] + + +@pytest.mark.asyncio +async def test_send_with_retry_supports_legacy_reasoning_signature(): + """External plugins with the old reasoning hook signature should keep working.""" + deltas: list[tuple[str, str, dict]] = [] + ends: list[tuple[str, dict]] = [] + + class _LegacyReasoningChannel(BaseChannel): + name = "legacy_reasoning" + display_name = "Legacy Reasoning" + + async def start(self) -> None: + pass + + async def stop(self) -> None: + pass + + async def send(self, msg: OutboundMessage) -> None: + pass + + async def send_reasoning_delta( + self, + chat_id: str, + delta: str, + metadata: dict | None = None, + ) -> None: + deltas.append((chat_id, delta, dict(metadata or {}))) + + async def send_reasoning_end( + self, + chat_id: str, + metadata: dict | None = None, + ) -> None: + ends.append((chat_id, dict(metadata or {}))) + + fake_config = SimpleNamespace( + channels=ChannelsConfig(send_max_retries=3), + providers=SimpleNamespace(groq=SimpleNamespace(api_key="")), + ) + mgr = ChannelManager.__new__(ChannelManager) + mgr.config = fake_config + mgr.bus = MessageBus() + mgr.channels = {"legacy_reasoning": _LegacyReasoningChannel(fake_config, mgr.bus)} + mgr._dispatch_task = None + + await mgr._send_with_retry( + mgr.channels["legacy_reasoning"], + outbound_message_for_event( + channel="legacy_reasoning", + chat_id="123", + event=ProgressEvent(content="thinking", reasoning_delta=True, stream_id="r1"), + ), + ) + await mgr._send_with_retry( + mgr.channels["legacy_reasoning"], + outbound_message_for_event( + channel="legacy_reasoning", + chat_id="123", + event=ProgressEvent(reasoning_end=True, stream_id="r1"), + ), + ) + + assert deltas == [ + ("123", "thinking", {"_reasoning_delta": True, "_stream_id": "r1"}), + ] + assert ends == [ + ("123", {"_reasoning_end": True, "_stream_id": "r1"}), + ] + + @pytest.mark.asyncio async def test_send_with_retry_skips_send_when_streamed(): """_send_with_retry should not call send for streamed response events."""