mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-09-03 17:52:00 +03:00
fix(channels): bound origin reply fingerprint cache
Signed-off-by: Lanre Shittu <136805224+Shizoqua@users.noreply.github.com>
This commit is contained in:
@@ -5,6 +5,7 @@ from __future__ import annotations
|
|||||||
import asyncio
|
import asyncio
|
||||||
import hashlib
|
import hashlib
|
||||||
import inspect
|
import inspect
|
||||||
|
from collections import OrderedDict
|
||||||
from collections.abc import Awaitable, Callable, Iterable, Mapping
|
from collections.abc import Awaitable, Callable, Iterable, Mapping
|
||||||
from contextlib import suppress
|
from contextlib import suppress
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
@@ -60,6 +61,7 @@ def _default_webui_dist() -> Path | None:
|
|||||||
_SEND_RETRY_DELAYS = (1, 2, 4)
|
_SEND_RETRY_DELAYS = (1, 2, 4)
|
||||||
_RESTART_NOTICE_START_TIMEOUT_S = 30.0
|
_RESTART_NOTICE_START_TIMEOUT_S = 30.0
|
||||||
_RESTART_NOTICE_START_POLL_S = 0.25
|
_RESTART_NOTICE_START_POLL_S = 0.25
|
||||||
|
ORIGIN_REPLY_FINGERPRINTS_MAX_SIZE = 1000
|
||||||
|
|
||||||
_BOOL_CAMEL_ALIASES: dict[str, str] = {
|
_BOOL_CAMEL_ALIASES: dict[str, str] = {
|
||||||
"send_progress": "sendProgress",
|
"send_progress": "sendProgress",
|
||||||
@@ -137,7 +139,7 @@ class ChannelManager:
|
|||||||
self._channel_tasks: dict[str, asyncio.Task[None]] = {}
|
self._channel_tasks: dict[str, asyncio.Task[None]] = {}
|
||||||
self._dispatch_task: asyncio.Task[None] | None = None
|
self._dispatch_task: asyncio.Task[None] | None = None
|
||||||
self._started = False
|
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()
|
self._init_channels()
|
||||||
|
|
||||||
@@ -669,6 +671,16 @@ class ChannelManager:
|
|||||||
normalized = " ".join(content.split())
|
normalized = " ".join(content.split())
|
||||||
return hashlib.sha1(normalized.encode("utf-8")).hexdigest() if normalized else ""
|
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:
|
def _should_suppress_outbound(self, msg: OutboundMessage) -> bool:
|
||||||
metadata = msg.metadata or {}
|
metadata = msg.metadata or {}
|
||||||
if isinstance(outbound_event_from_message(msg), ProgressEvent):
|
if isinstance(outbound_event_from_message(msg), ProgressEvent):
|
||||||
@@ -681,13 +693,14 @@ class ChannelManager:
|
|||||||
if isinstance(origin_message_id, str) and origin_message_id:
|
if isinstance(origin_message_id, str) and origin_message_id:
|
||||||
key = (msg.channel, msg.chat_id, origin_message_id)
|
key = (msg.channel, msg.chat_id, origin_message_id)
|
||||||
if self._origin_reply_fingerprints.get(key) == fingerprint:
|
if self._origin_reply_fingerprints.get(key) == fingerprint:
|
||||||
|
self._origin_reply_fingerprints.move_to_end(key)
|
||||||
return True
|
return True
|
||||||
self._origin_reply_fingerprints[key] = fingerprint
|
self._remember_origin_reply_fingerprint(key, fingerprint)
|
||||||
|
|
||||||
message_id = metadata.get("message_id")
|
message_id = metadata.get("message_id")
|
||||||
if isinstance(message_id, str) and message_id:
|
if isinstance(message_id, str) and message_id:
|
||||||
key = (msg.channel, msg.chat_id, 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
|
return False
|
||||||
|
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import json
|
|||||||
import subprocess
|
import subprocess
|
||||||
import sys
|
import sys
|
||||||
import tomllib
|
import tomllib
|
||||||
|
from collections import OrderedDict
|
||||||
from dataclasses import replace
|
from dataclasses import replace
|
||||||
from importlib.metadata import PackageNotFoundError
|
from importlib.metadata import PackageNotFoundError
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
@@ -32,7 +33,7 @@ from nanobot.channels.contracts import (
|
|||||||
SetupRequirement,
|
SetupRequirement,
|
||||||
channel_default_config,
|
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.channels.plugin import ChannelPlugin, load_channel_package
|
||||||
from nanobot.config.loader import load_config, save_config
|
from nanobot.config.loader import load_config, save_config
|
||||||
from nanobot.config.schema import ChannelsConfig, 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.bus = MessageBus()
|
||||||
mgr.channels = {}
|
mgr.channels = {}
|
||||||
mgr._dispatch_task = None
|
mgr._dispatch_task = None
|
||||||
mgr._origin_reply_fingerprints = {}
|
mgr._origin_reply_fingerprints = OrderedDict()
|
||||||
|
|
||||||
first = OutboundMessage(
|
first = OutboundMessage(
|
||||||
channel="feishu",
|
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
|
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
|
@pytest.mark.asyncio
|
||||||
async def test_send_with_retry_propagates_cancelled_error():
|
async def test_send_with_retry_propagates_cancelled_error():
|
||||||
"""_send_with_retry should re-raise CancelledError for graceful shutdown."""
|
"""_send_with_retry should re-raise CancelledError for graceful shutdown."""
|
||||||
|
|||||||
Reference in New Issue
Block a user