diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index ccc5e1e4a..e3fb5fe55 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -57,6 +57,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_ - Telegram uses a deep-link `/start ` flow over the existing long-polling worker. Slack, Discord, Feishu/Lark, DingTalk, WeChat, and WeCom use `/connect ` over their existing outbound channel workers. - WeChat timing settings (`polling_timeout`, `polling_retry_delay`, `qrcode_poll_interval`, `qrcode_poll_timeout`) accept only positive finite seconds; invalid values fall back to their defaults so polling cannot enter a hot loop or sleep forever. - WeCom serializes `start()` and `stop()` for each channel instance. The SDK `connect()` task covers connection setup only; after the handshake, the SDK owns a separate receive task. Shutdown cancels an in-progress connection attempt and awaits the SDK's actual asynchronous receive-task/socket cleanup before releasing lifecycle state or allowing a restart. Cancellation of `stop()` still propagates, but only after owned cleanup finishes and lifecycle references are cleared; real connection failures remain reported by `_on_ws_task_done`. +- WeCom outbound content is capped at the protocol's 20480 UTF-8 bytes: stream replies clip on a character boundary with a truncation marker (one stream carries the whole reply and cannot split mid-way), while proactive pushes split into at most 10 sequential markdown messages per push, with the remaining tail clipped and marker-terminated. Both paths measure bytes, not characters. A per-chat send lock serializes each split batch end to end, because manager workers run concurrently and two long pushes to the same chat would otherwise interleave chunks; locks are reference-counted and reclaimed once no sender holds or waits on them, so the registry does not grow for the life of the Gateway. - Frontend APIs: `GET /api/channels/providers`, `GET /api/channels/connections`, `POST /api/channels/{provider}/connect`, and `DELETE /api/channels/connections/{connection_id}`. - Browser APIs remain protected by normal Gateway auth/CSRF. Provider messages arrive through the already-configured channel workers. - Provider-level `connection_status` reflects the user's newest connection row. With no binding it is `not_connected`, except in auth-disabled local mode where a configured running channel reports `connected` because all channel messages already route to the default user. diff --git a/backend/app/channels/wecom.py b/backend/app/channels/wecom.py index 10990f58c..abb97ad24 100644 --- a/backend/app/channels/wecom.py +++ b/backend/app/channels/wecom.py @@ -34,6 +34,63 @@ def _open_binary(path: str): return open(path, "rb") +# The WeCom bot protocol caps message content at 20480 UTF-8 bytes, for both +# passive stream replies and active markdown pushes. +_WECOM_MAX_CONTENT_BYTES = 20480 +_TRUNCATION_MARKER = "\n\n... (truncated)" +# One push must not flood the chat with an unbounded run of messages: keep the +# first few chunks and collapse the rest into one truncated tail. +_WECOM_MAX_CHUNK_BATCH = 10 + + +def _clip_to_byte_limit(text: str, limit: int) -> str: + """Clip text to a UTF-8 byte budget, never splitting a character.""" + if len(text.encode("utf-8")) <= limit: + return text + budget = limit - len(_TRUNCATION_MARKER.encode("utf-8")) + clipped = text.encode("utf-8")[:budget].decode("utf-8", errors="ignore") + return clipped + _TRUNCATION_MARKER + + +def _split_for_byte_limit(text: str, limit: int) -> list[str]: + """Split text into chunks within the UTF-8 byte limit. + + Prefers newline boundaries so markdown structure survives the split. + The batch cap applies inside the loop, so a pathological text is never + fully split just to be discarded. + """ + if len(text.encode("utf-8")) <= limit: + return [text] + chunks: list[str] = [] + remaining = text + while len(remaining.encode("utf-8")) > limit: + if len(chunks) >= _WECOM_MAX_CHUNK_BATCH - 1: + logger.warning( + "WeCom push of %d bytes exceeds %d messages, capping the batch", + len(text.encode("utf-8")), + _WECOM_MAX_CHUNK_BATCH, + ) + chunks.append(_clip_to_byte_limit(remaining, limit)) + return chunks + window = remaining.encode("utf-8")[:limit].decode("utf-8", errors="ignore") + cut = window.rfind("\n") + if cut <= 0: + cut = len(window) + else: + # Keep the delimiter on this chunk's tail: the sequential messages + # must round-trip to the original text exactly. + cut += 1 + if cut == 0: + # limit is narrower than one whole character; take it anyway so + # the loop always advances. + cut = 1 + chunks.append(remaining[:cut]) + remaining = remaining[cut:] + if remaining: + chunks.append(remaining) + return chunks + + class WeComChannel(Channel): def __init__(self, bus: MessageBus, config: dict[str, Any]) -> None: super().__init__(name="wecom", bus=bus, config=config) @@ -45,6 +102,9 @@ class WeComChannel(Channel): self._lifecycle_lock = asyncio.Lock() self._ws_frames: dict[str, dict[str, Any]] = {} self._ws_stream_ids: dict[str, str] = {} + self._ws_send_locks: dict[str, asyncio.Lock] = {} + self._ws_send_lock_users: dict[str, int] = {} + self._ws_send_locks_guard = asyncio.Lock() self._working_message = "Working on it..." @property @@ -462,19 +522,40 @@ class WeComChannel(Channel): return await self._send_with_retry( - lambda: self._ws_client.reply_stream(frame, stream_id, msg.text, bool(msg.is_final)), + lambda: self._ws_client.reply_stream(frame, stream_id, _clip_to_byte_limit(msg.text, _WECOM_MAX_CONTENT_BYTES), bool(msg.is_final)), max_retries=_max_retries, log_prefix="[WeCom]", operation_name="stream send", ) return - body = {"msgtype": "markdown", "markdown": {"content": msg.text}} - await self._send_with_retry( - lambda: self._ws_client.send_message(msg.chat_id, body), - max_retries=_max_retries, - log_prefix="[WeCom]", - ) + # No replyable frame (e.g. a scheduled-task push): a stream reply is one + # stream per reply and cannot split mid-way, but this path can, so the + # full text goes out as sequential markdown messages. Each send awaits, + # so hold a per-chat lock across the whole batch: manager workers run + # concurrently, and two long pushes to the same chat would otherwise + # interleave chunks (A1, B1, A2, B2) and break the sequential contract. + async with self._ws_send_locks_guard: + lock = self._ws_send_locks.setdefault(msg.chat_id, asyncio.Lock()) + self._ws_send_lock_users[msg.chat_id] = self._ws_send_lock_users.get(msg.chat_id, 0) + 1 + try: + async with lock: + for chunk in _split_for_byte_limit(msg.text, _WECOM_MAX_CONTENT_BYTES): + body = {"msgtype": "markdown", "markdown": {"content": chunk}} + await self._send_with_retry( + lambda body=body: self._ws_client.send_message(msg.chat_id, body), + max_retries=_max_retries, + log_prefix="[WeCom]", + ) + finally: + async with self._ws_send_locks_guard: + self._ws_send_lock_users[msg.chat_id] -= 1 + # Reclaim only while nobody else is queued on this chat's lock; + # the guard serializes the check so a waiter can never end up + # holding a fresh lock for a chat whose batch is mid-flight. + if self._ws_send_lock_users[msg.chat_id] == 0: + self._ws_send_lock_users.pop(msg.chat_id, None) + self._ws_send_locks.pop(msg.chat_id, None) async def _upload_media_ws( self, diff --git a/backend/tests/test_wecom_content_limit.py b/backend/tests/test_wecom_content_limit.py new file mode 100644 index 000000000..80a256a1f --- /dev/null +++ b/backend/tests/test_wecom_content_limit.py @@ -0,0 +1,345 @@ +"""Regression tests for the WeCom outbound content byte limit (#5140). + +Both outbound paths in ``WeComChannel._send_ws`` previously sent unbounded +text while the bot protocol caps content at 20480 UTF-8 bytes. The stream +reply path now clips on a character boundary with a truncation marker, and +the proactive push path splits into sequential markdown messages. +""" + +from __future__ import annotations + +import asyncio +from typing import Any +from unittest.mock import AsyncMock + +from app.channels.message_bus import MessageBus, OutboundMessage +from app.channels.wecom import ( + _TRUNCATION_MARKER, + _WECOM_MAX_CHUNK_BATCH, + _WECOM_MAX_CONTENT_BYTES, + WeComChannel, + _clip_to_byte_limit, + _split_for_byte_limit, +) + + +def _run(coro): + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + + +def _byte_len(text: str) -> int: + return len(text.encode("utf-8")) + + +class TestClipToByteLimit: + def test_short_text_passes_through(self): + assert _clip_to_byte_limit("hello", 100) == "hello" + + def test_exact_limit_passes_through(self): + text = "a" * _WECOM_MAX_CONTENT_BYTES + assert _clip_to_byte_limit(text, _WECOM_MAX_CONTENT_BYTES) == text + + def test_multibyte_cut_never_splits_a_character(self): + # One 3-byte character straddling the budget cut. + text = "a" * 100 + "汉" * 100 + clipped = _clip_to_byte_limit(text, 105) + assert clipped.endswith("(truncated)") + assert _byte_len(clipped) <= 105 + assert "汉" not in clipped + + def test_full_width_report_stays_within_protocol_cap(self): + text = "深度报告" * 10000 + clipped = _clip_to_byte_limit(text, _WECOM_MAX_CONTENT_BYTES) + assert _byte_len(clipped) <= _WECOM_MAX_CONTENT_BYTES + assert clipped.endswith("(truncated)") + + +class TestSplitForByteLimit: + def test_short_text_is_single_chunk(self): + assert _split_for_byte_limit("hello", 100) == ["hello"] + + def test_each_chunk_within_limit_and_content_preserved(self): + text = "\n".join(f"line {i} " + "字" * 50 for i in range(200)) + chunks = _split_for_byte_limit(text, _WECOM_MAX_CONTENT_BYTES) + assert len(chunks) > 1 + for chunk in chunks: + assert _byte_len(chunk) <= _WECOM_MAX_CONTENT_BYTES + # Exact round trip: the sequential messages must rebuild the original + # text byte for byte, delimiters included. + assert "".join(chunks) == text + + def test_boundary_newline_lands_on_chunk_tail(self): + # Regression for the review on #5148: a boundary delimiter used to be + # stripped by lstrip, so one newline per split silently vanished. + text = "ab\ncd\n" + "x" * (_WECOM_MAX_CONTENT_BYTES * 2) + chunks = _split_for_byte_limit(text, _WECOM_MAX_CONTENT_BYTES) + assert "".join(chunks) == text + + def test_leading_blank_lines_are_content_not_dropped(self): + text = "第一段\n\n\n" + "字" * (_WECOM_MAX_CONTENT_BYTES * 2) + chunks = _split_for_byte_limit(text, _WECOM_MAX_CONTENT_BYTES) + assert "".join(chunks) == text + + def test_no_newline_falls_back_to_hard_cut(self): + text = "x" * (_WECOM_MAX_CONTENT_BYTES * 2 + 500) + chunks = _split_for_byte_limit(text, _WECOM_MAX_CONTENT_BYTES) + assert len(chunks) == 3 + for chunk in chunks: + assert _byte_len(chunk) <= _WECOM_MAX_CONTENT_BYTES + assert "".join(chunks) == text + + def test_limit_narrower_than_one_character_still_advances(self): + # limit=3 cannot hold even one 4-byte emoji: the window decodes to an + # empty string and a hard cut of 0 would spin forever. The split must + # take the character anyway and terminate with content intact. + chunks = _split_for_byte_limit("😀" * 5, 3) + assert "".join(chunks) == "😀" * 5 + assert len(chunks) == 5 + + def test_split_caps_chunk_batch_with_truncation_marker(self): + text = "x" * (_WECOM_MAX_CONTENT_BYTES * 25) + chunks = _split_for_byte_limit(text, _WECOM_MAX_CONTENT_BYTES) + assert len(chunks) == _WECOM_MAX_CHUNK_BATCH + for chunk in chunks: + assert _byte_len(chunk) <= _WECOM_MAX_CONTENT_BYTES + assert chunks[-1].endswith(_TRUNCATION_MARKER) + # The kept prefix is verbatim; only the collapsed tail is clipped. + assert "".join(chunks[:-1]) == text[: _WECOM_MAX_CONTENT_BYTES * (_WECOM_MAX_CHUNK_BATCH - 1)] + + def test_split_under_cap_is_not_marked(self): + text = "x" * (_WECOM_MAX_CONTENT_BYTES * 3) + chunks = _split_for_byte_limit(text, _WECOM_MAX_CONTENT_BYTES) + assert len(chunks) == 3 + assert not chunks[-1].endswith(_TRUNCATION_MARKER) + + def test_cap_clips_the_unsplit_remainder(self, monkeypatch): + # The cap must apply inside the loop: the remainder past the kept + # chunks is clipped whole, never fully split just to be discarded. + import app.channels.wecom as wecom_module + + seen = {} + real_clip = wecom_module._clip_to_byte_limit + + def spy(text, limit): + seen["text"] = text + return real_clip(text, limit) + + monkeypatch.setattr(wecom_module, "_clip_to_byte_limit", spy) + text = "x" * (_WECOM_MAX_CONTENT_BYTES * 25) + chunks = _split_for_byte_limit(text, _WECOM_MAX_CONTENT_BYTES) + assert len(chunks) == _WECOM_MAX_CHUNK_BATCH + expected_tail = len(text) - _WECOM_MAX_CONTENT_BYTES * (_WECOM_MAX_CHUNK_BATCH - 1) + assert len(seen["text"]) == expected_tail + + +class TestSendWsContentLimit: + def _channel(self) -> WeComChannel: + ch = WeComChannel(bus=MessageBus(), config={}) + ch._ws_client = AsyncMock() + return ch + + def test_stream_reply_clips_overlong_snapshot(self): + ch = self._channel() + ch._ws_frames["t1"] = {"frame": 1} + ch._ws_stream_ids["t1"] = "stream-1" + msg = OutboundMessage( + channel_name="wecom", + chat_id="c1", + thread_id="th1", + text="报告" * 20000, + is_final=True, + thread_ts="t1", + ) + _run(ch._send_ws(msg)) + ch._ws_client.reply_stream.assert_called_once() + sent = ch._ws_client.reply_stream.call_args[0][2] + assert _byte_len(sent) <= _WECOM_MAX_CONTENT_BYTES + assert sent.endswith("(truncated)") + + def test_stream_reply_short_text_untouched(self): + ch = self._channel() + ch._ws_frames["t1"] = {"frame": 1} + ch._ws_stream_ids["t1"] = "stream-1" + msg = OutboundMessage( + channel_name="wecom", + chat_id="c1", + thread_id="th1", + text="short reply", + is_final=False, + thread_ts="t1", + ) + _run(ch._send_ws(msg)) + assert ch._ws_client.reply_stream.call_args[0][2] == "short reply" + + def test_proactive_push_splits_into_sequential_markdown_messages(self): + ch = self._channel() + msg: OutboundMessage = OutboundMessage( + channel_name="wecom", + chat_id="c1", + thread_id="th1", + text="推送内容\n" * 5000, + thread_ts=None, + ) + _run(ch._send_ws(msg)) + calls = ch._ws_client.send_message.call_args_list + assert len(calls) > 1 + for call in calls: + body: dict[str, Any] = call[0][1] + assert body["msgtype"] == "markdown" + assert _byte_len(body["markdown"]["content"]) <= _WECOM_MAX_CONTENT_BYTES + + def test_proactive_push_short_text_single_message(self): + ch = self._channel() + msg = OutboundMessage( + channel_name="wecom", + chat_id="c1", + thread_id="th1", + text="short push", + thread_ts=None, + ) + _run(ch._send_ws(msg)) + ch._ws_client.send_message.assert_called_once() + + +class TestSendWsChatSerialization: + """Review on #5148: manager workers run concurrently, and each chunk send + awaits, so two long pushes to the same chat used to interleave (A1, B1, + A2, B2). The per-chat lock must keep each batch contiguous. + """ + + @staticmethod + def _recording_channel(): + ch = WeComChannel(bus=MessageBus(), config={}) + sent: list[tuple[str, str]] = [] + + class RecordingClient: + async def send_message(self, chat_id, body): + sent.append((chat_id, body["markdown"]["content"])) + # Yield so a lockless batch would interleave with the other + # coroutine after every single chunk. + await asyncio.sleep(0) + + ch._ws_client = RecordingClient() + return ch, sent + + @staticmethod + def _push(chat_id: str, text: str) -> OutboundMessage: + return OutboundMessage( + channel_name="wecom", + chat_id=chat_id, + thread_id="th1", + text=text, + thread_ts=None, + ) + + def test_concurrent_batches_to_same_chat_stay_contiguous(self): + ch, sent = self._recording_channel() + text_a = "\n".join(f"报告甲 第{i}段 " + "字" * 50 for i in range(400)) + text_b = "\n".join(f"推送乙 第{i}段 " + "文" * 50 for i in range(400)) + chunks_a = _split_for_byte_limit(text_a, _WECOM_MAX_CONTENT_BYTES) + chunks_b = _split_for_byte_limit(text_b, _WECOM_MAX_CONTENT_BYTES) + assert len(chunks_a) > 1 and len(chunks_b) > 1 + + async def both(): + await asyncio.gather( + ch._send_ws(self._push("c1", text_a)), + ch._send_ws(self._push("c1", text_b)), + ) + + _run(both()) + contents = [content for _, content in sent] + # Either batch order is fine; what matters is no interleaving. + assert contents in (chunks_a + chunks_b, chunks_b + chunks_a) + + def test_staggered_waiter_keeps_one_lock_and_registry_drains(self): + # The interleaving this fix exists for: the waiter must queue on the + # same lock while the holder's cleanup runs, never end up holding a + # fresh lock mid-batch, and the registry must drain once both finish. + ch, sent = self._recording_channel() + text_a = "\n".join(f"先行批 第{i}段 " + "字" * 50 for i in range(400)) + text_b = "\n".join(f"后到批 第{i}段 " + "文" * 50 for i in range(400)) + chunks_a = _split_for_byte_limit(text_a, _WECOM_MAX_CONTENT_BYTES) + chunks_b = _split_for_byte_limit(text_b, _WECOM_MAX_CONTENT_BYTES) + assert len(chunks_a) > 1 and len(chunks_b) > 1 + + async def staggered(): + holder = asyncio.create_task(ch._send_ws(self._push("c1", text_a))) + # Let the holder get mid-batch, then queue the waiter on the same + # chat so its registration overlaps the holder's later chunks. + for _ in range(3): + await asyncio.sleep(0) + waiter = asyncio.create_task(ch._send_ws(self._push("c1", text_b))) + await asyncio.gather(holder, waiter) + + _run(staggered()) + contents = [content for _, content in sent] + # The holder started first and keeps the lock, so its batch is first. + assert contents == chunks_a + chunks_b + assert ch._ws_send_locks == {} + assert ch._ws_send_lock_users == {} + + def test_completed_chat_lock_is_reclaimed(self): + ch, _ = self._recording_channel() + _run(ch._send_ws(self._push("c1", "short push"))) + assert ch._ws_send_locks == {} + assert ch._ws_send_lock_users == {} + + def test_lock_is_reclaimed_after_a_capped_batch(self): + ch, sent = self._recording_channel() + text = "x" * (_WECOM_MAX_CONTENT_BYTES * 12) + _run(ch._send_ws(self._push("c1", text))) + assert sent # the batch went out + assert ch._ws_send_locks == {} + assert ch._ws_send_lock_users == {} + + def test_concurrent_senders_each_leave_no_locks(self): + ch, _ = self._recording_channel() + + async def many(): + await asyncio.gather(*(ch._send_ws(self._push(f"chat-{i}", f"msg {i}")) for i in range(20))) + + _run(many()) + assert ch._ws_send_locks == {} + assert ch._ws_send_lock_users == {} + + def test_different_chats_keep_their_own_order(self): + ch, sent = self._recording_channel() + text_a = "\n".join(f"给甲群 第{i}段 " + "字" * 50 for i in range(300)) + text_b = "\n".join(f"给乙群 第{i}段 " + "文" * 50 for i in range(300)) + + async def both(): + await asyncio.gather( + ch._send_ws(self._push("chat-a", text_a)), + ch._send_ws(self._push("chat-b", text_b)), + ) + + _run(both()) + # Different chats may interleave freely, but each chat's own chunks + # must arrive in order and complete. + assert [c for chat, c in sent if chat == "chat-a"] == _split_for_byte_limit(text_a, _WECOM_MAX_CONTENT_BYTES) + assert [c for chat, c in sent if chat == "chat-b"] == _split_for_byte_limit(text_b, _WECOM_MAX_CONTENT_BYTES) + + +class TestEmojiBoundaries: + def test_split_all_emoji_input_terminates_and_preserves(self): + # 4-byte emoji only: a byte cut lands mid-character, and the split must + # carry that character into the next chunk rather than dropping it. + text = "😀" * 10000 + chunks = _split_for_byte_limit(text, _WECOM_MAX_CONTENT_BYTES) + assert len(chunks) == 2 + assert "".join(chunks) == text + assert all(_byte_len(c) <= _WECOM_MAX_CONTENT_BYTES for c in chunks) + + def test_split_tiny_limit_with_emoji_never_loops_or_loses(self): + chunks = _split_for_byte_limit("😀" * 10, 9) + assert "".join(chunks) == "😀" * 10 + assert all(_byte_len(c) <= 9 for c in chunks) + + def test_clip_at_emoji_boundary_stays_within_budget(self): + out = _clip_to_byte_limit("😀" * 10000, 105) + assert _byte_len(out) <= 105 + assert out.endswith("(truncated)")