diff --git a/nanobot/channels/manager.py b/nanobot/channels/manager.py index 4b16774a8..f5ad7a04a 100644 --- a/nanobot/channels/manager.py +++ b/nanobot/channels/manager.py @@ -5,6 +5,7 @@ from __future__ import annotations import asyncio import hashlib import inspect +from collections import OrderedDict from collections.abc import Awaitable, Callable, Iterable, Mapping from contextlib import suppress from pathlib import Path @@ -60,6 +61,7 @@ def _default_webui_dist() -> Path | None: _SEND_RETRY_DELAYS = (1, 2, 4) _RESTART_NOTICE_START_TIMEOUT_S = 30.0 _RESTART_NOTICE_START_POLL_S = 0.25 +ORIGIN_REPLY_FINGERPRINTS_MAX_SIZE = 1000 _BOOL_CAMEL_ALIASES: dict[str, str] = { "send_progress": "sendProgress", @@ -137,7 +139,7 @@ class ChannelManager: self._channel_tasks: dict[str, asyncio.Task[None]] = {} self._dispatch_task: asyncio.Task[None] | None = None self._started = False - self._origin_reply_fingerprints: dict[tuple[str, str, str], str] = {} + self._origin_reply_fingerprints: OrderedDict[tuple[str, str, str], str] = OrderedDict() self._init_channels() @@ -669,6 +671,16 @@ class ChannelManager: normalized = " ".join(content.split()) return hashlib.sha1(normalized.encode("utf-8")).hexdigest() if normalized else "" + def _remember_origin_reply_fingerprint( + self, + key: tuple[str, str, str], + fingerprint: str, + ) -> None: + self._origin_reply_fingerprints[key] = fingerprint + self._origin_reply_fingerprints.move_to_end(key) + while len(self._origin_reply_fingerprints) > ORIGIN_REPLY_FINGERPRINTS_MAX_SIZE: + self._origin_reply_fingerprints.popitem(last=False) + def _should_suppress_outbound(self, msg: OutboundMessage) -> bool: metadata = msg.metadata or {} if isinstance(outbound_event_from_message(msg), ProgressEvent): @@ -681,13 +693,14 @@ class ChannelManager: if isinstance(origin_message_id, str) and origin_message_id: key = (msg.channel, msg.chat_id, origin_message_id) if self._origin_reply_fingerprints.get(key) == fingerprint: + self._origin_reply_fingerprints.move_to_end(key) return True - self._origin_reply_fingerprints[key] = fingerprint + self._remember_origin_reply_fingerprint(key, fingerprint) message_id = metadata.get("message_id") if isinstance(message_id, str) and message_id: key = (msg.channel, msg.chat_id, message_id) - self._origin_reply_fingerprints[key] = fingerprint + self._remember_origin_reply_fingerprint(key, fingerprint) return False diff --git a/tests/channels/test_channel_plugins.py b/tests/channels/test_channel_plugins.py index a223c37bb..6e5ac47c9 100644 --- a/tests/channels/test_channel_plugins.py +++ b/tests/channels/test_channel_plugins.py @@ -7,6 +7,7 @@ import json import subprocess import sys import tomllib +from collections import OrderedDict from dataclasses import replace from importlib.metadata import PackageNotFoundError from pathlib import Path @@ -32,7 +33,7 @@ from nanobot.channels.contracts import ( SetupRequirement, channel_default_config, ) -from nanobot.channels.manager import ChannelManager +from nanobot.channels.manager import ORIGIN_REPLY_FINGERPRINTS_MAX_SIZE, ChannelManager from nanobot.channels.plugin import ChannelPlugin, load_channel_package from nanobot.config.loader import load_config, save_config from nanobot.config.schema import ChannelsConfig, Config @@ -3179,7 +3180,7 @@ def test_outbound_duplicate_suppression_is_scoped_to_origin_message() -> None: mgr.bus = MessageBus() mgr.channels = {} mgr._dispatch_task = None - mgr._origin_reply_fingerprints = {} + mgr._origin_reply_fingerprints = OrderedDict() first = OutboundMessage( channel="feishu", @@ -3212,6 +3213,39 @@ def test_outbound_duplicate_suppression_is_scoped_to_origin_message() -> None: assert mgr._should_suppress_outbound(new_origin_content) is False +def test_outbound_duplicate_suppression_cache_is_bounded() -> None: + mgr = ChannelManager.__new__(ChannelManager) + mgr._origin_reply_fingerprints = OrderedDict() + + for index in range(ORIGIN_REPLY_FINGERPRINTS_MAX_SIZE): + msg = OutboundMessage( + channel="feishu", + chat_id="chat123", + content="Done", + metadata={"message_id": f"msg-{index}"}, + ) + assert mgr._should_suppress_outbound(msg) is False + + duplicate = OutboundMessage( + channel="feishu", + chat_id="chat123", + content="Done", + metadata={"origin_message_id": "msg-0"}, + ) + newest = OutboundMessage( + channel="feishu", + chat_id="chat123", + content="Done", + metadata={"message_id": f"msg-{ORIGIN_REPLY_FINGERPRINTS_MAX_SIZE}"}, + ) + + assert mgr._should_suppress_outbound(duplicate) is True + assert mgr._should_suppress_outbound(newest) is False + assert len(mgr._origin_reply_fingerprints) == ORIGIN_REPLY_FINGERPRINTS_MAX_SIZE + assert ("feishu", "chat123", "msg-0") in mgr._origin_reply_fingerprints + assert ("feishu", "chat123", "msg-1") not in mgr._origin_reply_fingerprints + + @pytest.mark.asyncio async def test_send_with_retry_propagates_cancelled_error(): """_send_with_retry should re-raise CancelledError for graceful shutdown."""