fix(channels): preserve legacy stream hook signatures

2026-07-01 20:17:00 +08:00
parent c757c5466c
commit 03be51ade5
2 changed files with 216 additions and 25 deletions
@@ -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)

@@ -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."""