From 05bcc28e37eb42d2c8c1eac8cb437ac97b8977f5 Mon Sep 17 00:00:00 2001 From: Ace Data Cloud Dev Date: Tue, 21 Jul 2026 17:47:03 +0800 Subject: [PATCH] fix(channels): poll Wisdom messages as fallback --- coding_bridge/channels/wechat/adapter.py | 170 +++++++++++++- coding_bridge/channels/wechat/client.py | 37 +++- tests/test_channels_wechat.py | 270 ++++++++++++++++++++++- 3 files changed, 461 insertions(+), 16 deletions(-) diff --git a/coding_bridge/channels/wechat/adapter.py b/coding_bridge/channels/wechat/adapter.py index 64c1582..02deabd 100644 --- a/coding_bridge/channels/wechat/adapter.py +++ b/coding_bridge/channels/wechat/adapter.py @@ -25,6 +25,9 @@ import json import logging import re +import time +from collections import deque +from datetime import datetime from typing import Any from urllib.parse import urlparse, urlunparse @@ -37,6 +40,8 @@ _INITIAL_BACKOFF_S = 0.5 _MAX_BACKOFF_S = 30.0 +_POLL_INTERVAL_S = 1.0 +_MESSAGE_SEEN_MAX = 5000 _HTTP_TO_WS = {"http": "ws", "https": "wss"} # Any query-string token or password segment is redacted before logging. _TOKEN_QS_RE = re.compile(r"([?&](?:token|api_token|password)=)([^&]+)", re.IGNORECASE) @@ -78,23 +83,71 @@ def _parse_incoming(payload: dict[str, Any]) -> IncomingMessage | None: if not isinstance(target, str) or not target: return None + message_id = data.get("msg_id") if isinstance(data.get("msg_id"), str) else None + upstream_id = f"{target}:{message_id}" if message_id else None return IncomingMessage( sender_id=str(data.get("sender_id") or target), sender_name=data.get("sender_name") if isinstance(data.get("sender_name"), str) else None, target=ChannelTarget( conversation_id=target, conversation_type=str(data.get("conversation_type") or "private"), - reply_to_id=data.get("msg_id") if isinstance(data.get("msg_id"), str) else None, + reply_to_id=message_id, ), text=text, msg_type=str(data.get("msg_type") or "text"), direction=direction, - upstream_id=data.get("msg_id") if isinstance(data.get("msg_id"), str) else None, + upstream_id=upstream_id, received_at_ms=int(data["timestamp"]) if isinstance(data.get("timestamp"), int) else None, raw=dict(data), ) +def _parse_polled_message( + data: dict[str, Any], + targets: dict[str, tuple[str, str]], +) -> IncomingMessage | None: + if data.get("direction") != "inbound": + return None + text = data.get("text") + if not isinstance(text, str) or not text: + return None + conversation_id = data.get("conversation_id") + if not isinstance(conversation_id, str) or not conversation_id: + return None + mapped_target = targets.get(conversation_id) + if mapped_target is None: + return None + target_name = data.get("conversation_name") + if not isinstance(target_name, str) or not target_name: + target_name = mapped_target[0] + message_id = str(data.get("id") or "") + if not message_id: + return None + sent_at = data.get("sent_at") if isinstance(data.get("sent_at"), str) else "" + upstream_id = f"{conversation_id}:{message_id}" + received_at_ms = None + if sent_at: + with contextlib.suppress(ValueError): + timestamp = datetime.fromisoformat(sent_at.replace("Z", "+00:00")).timestamp() + received_at_ms = int(timestamp * 1000) + conversation_type = mapped_target[1] + return IncomingMessage( + sender_id=str(data.get("sender_id") or conversation_id), + sender_name=data.get("sender_name") if isinstance(data.get("sender_name"), str) else None, + target=ChannelTarget( + conversation_id=conversation_id, + conversation_type=conversation_type, + extra={"send_target": target_name}, + ), + text=text, + msg_type=str(data.get("type") or "text"), + direction="inbound", + upstream_id=upstream_id, + received_at_ms=received_at_ms, + raw=dict(data), + ) + + class WeChatAdapter: """One WeChat gateway endpoint = one WeChat instance = one adapter. @@ -115,6 +168,7 @@ def __init__( client: WeChatClient | None = None, ws_connect: Any | None = None, stop_event: asyncio.Event | None = None, + poll_interval: float = _POLL_INTERVAL_S, ) -> None: if not instance_id: raise ValueError("instance_id must be a non-empty string") @@ -132,6 +186,9 @@ def __init__( self._stop = stop_event or asyncio.Event() self._handler: MessageHandler | None = None self._owns_client = client is None + self._poll_interval = poll_interval + self._seen_order: deque[str] = deque() + self._seen: set[str] = set() # Held for the lifetime of one WS session so ``aclose()`` can force # the receive loop to unblock without waiting for a gateway heartbeat. self._active_ws: Any | None = None @@ -142,6 +199,18 @@ def set_handler(self, handler: MessageHandler) -> None: async def run(self) -> None: if self._handler is None: raise RuntimeError("WeChatAdapter.run() called before set_handler()") + poll_task = asyncio.create_task( + self._poll_loop(), + name=f"wechat-poll-{self.instance_id}", + ) + try: + await self._run_ws() + finally: + poll_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await poll_task + + async def _run_ws(self) -> None: ws_url = _build_ws_url(self._base_url, self._token) backoff = _INITIAL_BACKOFF_S while not self._stop.is_set(): @@ -178,6 +247,94 @@ async def run(self) -> None: pass backoff = min(backoff * 2, _MAX_BACKOFF_S) + async def _poll_loop(self) -> None: + cursor = int(time.time()) - 1 + delay = self._poll_interval + while not self._stop.is_set(): + try: + await asyncio.wait_for(self._stop.wait(), timeout=delay) + return + except asyncio.TimeoutError: + pass + + try: + rows = await self._client.poll_messages(cursor, limit=500) + if not rows: + delay = self._poll_interval + continue + newest = cursor + for row in rows: + sent_at = row.get("sent_at") + if isinstance(sent_at, str): + with contextlib.suppress(ValueError): + newest = max( + newest, + int( + datetime.fromisoformat( + sent_at.replace("Z", "+00:00") + ).timestamp() + ), + ) + unseen_rows = [ + row + for row in rows + if ( + row.get("conversation_id") + and row.get("id") is not None + and f"{row['conversation_id']}:{row['id']}" not in self._seen + ) + ] + conversations = await self._client.list_conversations() if unseen_rows else [] + except asyncio.CancelledError: + raise + except Exception as exc: + logger.warning( + "wechat: poll error instance=%s err=%s.%s", + self.instance_id, + exc.__class__.__module__, + exc.__class__.__name__, + ) + delay = min(max(delay * 2, self._poll_interval), _MAX_BACKOFF_S) + continue + + delay = self._poll_interval + + targets = { + str(item["id"]): (str(item["name"]), str(item.get("type") or "private")) + for item in conversations + if item.get("id") and item.get("name") + } + parsed: list[IncomingMessage] = [] + for row in unseen_rows: + message = _parse_polled_message(row, targets) + if message is not None: + parsed.append(message) + + parsed.sort(key=lambda item: item.received_at_ms or 0) + for message in parsed: + await self._dispatch(message, source="poll") + + cursor = max(cursor, min(newest, int(time.time()) - 1)) + + async def _dispatch(self, message: IncomingMessage, *, source: str) -> None: + identity = message.upstream_id + if identity and identity in self._seen: + return + if identity: + self._seen.add(identity) + self._seen_order.append(identity) + while len(self._seen_order) > _MESSAGE_SEEN_MAX: + self._seen.discard(self._seen_order.popleft()) + assert self._handler is not None + try: + await self._handler(message, self) + except Exception: + logger.exception( + "wechat: %s handler error instance=%s", + source, + self.instance_id, + ) + async def _consume(self, ws: Any) -> None: async for raw in ws: if self._stop.is_set(): @@ -192,14 +349,7 @@ async def _consume(self, ws: Any) -> None: msg = _parse_incoming(payload) if msg is None: continue - assert self._handler is not None # invariant checked in run() - try: - await self._handler(msg, self) - except Exception: - # A single handler failure must not kill the receive loop — - # the dispatcher already logs its own errors, and abusive - # senders shouldn't be able to DoS the adapter. - logger.exception("wechat: handler error instance=%s", self.instance_id) + await self._dispatch(msg, source="WS") async def send( self, target: ChannelTarget, text: str, *, reply_to: str | None = None diff --git a/coding_bridge/channels/wechat/client.py b/coding_bridge/channels/wechat/client.py index aa1fc2f..0167ed2 100644 --- a/coding_bridge/channels/wechat/client.py +++ b/coding_bridge/channels/wechat/client.py @@ -20,6 +20,7 @@ # is treated as a delivery failure — the adapter surfaces the payload verbatim # to the caller so operators have real diagnostics in the log. _ACCEPTED_STATUSES = frozenset({200, 201, 202}) +_POLL_READ_TIMEOUT_S = 30.0 # Safe pattern for a gateway task id. The gateway generates UUID-like tokens; we # refuse anything with URL metacharacters so a caller can't inject a query @@ -63,8 +64,9 @@ async def aclose(self) -> None: async def send_message( self, target: ChannelTarget, text: str, *, reply_to: str | None = None ) -> SendResult: + send_target = target.extra.get("send_target") payload: dict[str, Any] = { - "target": target.conversation_id, + "target": send_target if isinstance(send_target, str) else target.conversation_id, "text": text, } if target.conversation_type: @@ -112,6 +114,39 @@ async def send_message( latency_ms=latency_ms, ) + async def poll_messages(self, since: int, *, limit: int = 100) -> list[dict[str, Any]]: + """Return messages newer than a Unix timestamp from Wisdom's WAL reader.""" + resp = await self._client.get( + "/api/messages/poll", + params={"since": since, "limit": limit}, + timeout=_POLL_READ_TIMEOUT_S, + ) + resp.raise_for_status() + body = resp.json() + if not isinstance(body, list): + return [] + return [item for item in body if isinstance(item, dict)] + + async def list_conversations(self) -> list[dict[str, Any]]: + """Return conversation ids, names, and types used to address replies.""" + conversations: list[dict[str, Any]] = [] + offset = 0 + while True: + resp = await self._client.get( + "/api/conversations", + params={"limit": 200, "offset": offset}, + timeout=_POLL_READ_TIMEOUT_S, + ) + resp.raise_for_status() + body = resp.json() + if not isinstance(body, dict) or not isinstance(body.get("conversations"), list): + return conversations + page = [item for item in body["conversations"] if isinstance(item, dict)] + conversations.extend(page) + if len(page) < 200 or offset >= 1000: + return conversations + offset += len(page) + async def get_task_status(self, task_id: str) -> dict[str, Any]: """Fetch delivery status for a task returned by ``send_message``. diff --git a/tests/test_channels_wechat.py b/tests/test_channels_wechat.py index b7e9888..7e27bcc 100644 --- a/tests/test_channels_wechat.py +++ b/tests/test_channels_wechat.py @@ -13,7 +13,12 @@ from coding_bridge.channels import ChannelTarget, IncomingMessage, SendResult from coding_bridge.channels.wechat import WeChatAdapter, WeChatClient -from coding_bridge.channels.wechat.adapter import _build_ws_url, _parse_incoming, _redact_url +from coding_bridge.channels.wechat.adapter import ( + _build_ws_url, + _parse_incoming, + _parse_polled_message, + _redact_url, +) # --------------------------------------------------------------------------- # Pure helpers: no network @@ -72,7 +77,7 @@ def test_parse_incoming_happy_path_private() -> None: assert msg.target.conversation_type == "private" assert msg.target.reply_to_id == "m1" assert msg.text == "/ask hi" - assert msg.upstream_id == "m1" + assert msg.upstream_id == "wxid_alice:m1" def test_parse_incoming_falls_back_to_target_when_no_sender_id() -> None: @@ -139,6 +144,27 @@ async def test_wechat_client_send_202_returns_ok_with_upstream_id(respx_mock): assert route.calls[0].request.headers["authorization"] == "Bearer tok" +@pytest.mark.asyncio +async def test_wechat_client_uses_polled_display_name_only_for_outbound_send(respx_mock): + route = respx_mock.post("http://host/api/messages/send").mock( + return_value=httpx.Response(202, json={"task_id": "t-2"}) + ) + client = WeChatClient("http://host", "tok") + target = ChannelTarget( + conversation_id="Msg_private_1", + conversation_type="private", + extra={"send_target": "Alice"}, + ) + try: + result = await client.send_message(target, "hi") + finally: + await client.aclose() + + assert result.ok is True + sent = json.loads(route.calls[0].request.content.decode()) + assert sent["target"] == "Alice" + + @pytest.mark.asyncio async def test_wechat_client_send_500_returns_error_and_never_leaks_token(respx_mock): respx_mock.post("http://host/api/messages/send").mock( @@ -172,6 +198,85 @@ async def test_wechat_client_send_transport_error_returns_non_raising_result(res assert "ConnectError" in (r.error or "") +@pytest.mark.asyncio +async def test_wechat_client_polls_messages_with_authenticated_cursor(respx_mock): + route = respx_mock.get("http://host/api/messages/poll").mock( + return_value=httpx.Response(200, json=[{"id": "1", "text": "hello"}]) + ) + client = WeChatClient("http://host", "tok") + try: + rows = await client.poll_messages(123, limit=25) + finally: + await client.aclose() + + assert rows == [{"id": "1", "text": "hello"}] + request = route.calls[0].request + assert request.url.params["since"] == "123" + assert request.url.params["limit"] == "25" + assert request.headers["authorization"] == "Bearer tok" + assert request.extensions["timeout"]["read"] == 30.0 + + +@pytest.mark.asyncio +async def test_wechat_client_lists_conversation_targets(respx_mock): + route = respx_mock.get("http://host/api/conversations").mock( + side_effect=[ + httpx.Response( + 200, + json={ + "total": 1, + "conversations": [ + {"id": f"Msg_{index}", "name": f"User {index}", "type": "private"} + for index in range(200) + ], + }, + ), + httpx.Response( + 200, + json={ + "total": 1, + "conversations": [{"id": "Msg_200", "name": "Team", "type": "group"}], + }, + ), + ] + ) + client = WeChatClient("http://host", "tok") + try: + rows = await client.list_conversations() + finally: + await client.aclose() + + assert len(rows) == 201 + assert rows[-1] == {"id": "Msg_200", "name": "Team", "type": "group"} + assert route.call_count == 2 + assert route.calls[1].request.url.params["offset"] == "200" + + +@pytest.mark.asyncio +async def test_wechat_client_stops_at_wisdom_maximum_offset(respx_mock): + page = [ + {"id": f"Msg_{index}", "name": f"User {index}", "type": "private"} for index in range(200) + ] + route = respx_mock.get("http://host/api/conversations").mock( + return_value=httpx.Response(200, json={"total": 1, "conversations": page}) + ) + client = WeChatClient("http://host", "tok") + try: + rows = await client.list_conversations() + finally: + await client.aclose() + + assert len(rows) == 1200 + assert [call.request.url.params["offset"] for call in route.calls] == [ + "0", + "200", + "400", + "600", + "800", + "1000", + ] + + @pytest.mark.asyncio async def test_wechat_client_transport_error_does_not_leak_token(respx_mock): """Defensive: even if httpx's exception grew to echo request headers, @@ -234,7 +339,11 @@ def _factory(_url: str) -> _FakeWS: return _factory -def _valid_frame(text: str = "/ask hi", target: str = "wxid_a") -> str: +def _valid_frame( + text: str = "/ask hi", + target: str = "wxid_a", + message_id: str = "m1", +) -> str: return json.dumps( { "event": "message.new", @@ -246,7 +355,7 @@ def _valid_frame(text: str = "/ask hi", target: str = "wxid_a") -> str: "sender_id": target, "target": target, "text": text, - "msg_id": "m1", + "msg_id": message_id, }, } ) @@ -283,6 +392,154 @@ async def _drive() -> None: assert seen[0].sender_id == "wxid_a" +@pytest.mark.asyncio +async def test_adapter_polls_when_wisdom_websocket_is_silent(): + seen: list[IncomingMessage] = [] + + async def handler(msg: IncomingMessage, _adapter) -> None: + seen.append(msg) + stop.set() + + class _PollingClient: + async def poll_messages(self, since: int, *, limit: int = 100): + return [ + { + "id": "1", + "conversation_id": "Msg_private_1", + "sender_id": None, + "sender_name": None, + "direction": "inbound", + "type": "text", + "text": "hello", + "sent_at": "2026-07-21T07:24:55Z", + } + ] + + async def list_conversations(self): + return [ + { + "id": "Msg_private_1", + "name": "Alice", + "type": "private", + } + ] + + async def send_message(self, target, text, *, reply_to=None): + return SendResult(ok=True) + + async def aclose(self) -> None: + return + + stop = asyncio.Event() + adapter = WeChatAdapter( + instance_id="wisdom", + base_url="https://wisdom.example", + token="tok", + client=_PollingClient(), + ws_connect=_fake_connect([], stop), + stop_event=stop, + poll_interval=0.01, + ) + adapter.set_handler(handler) + + await asyncio.wait_for(adapter.run(), timeout=1.0) + await adapter.aclose() + + assert len(seen) == 1 + assert seen[0].text == "hello" + assert seen[0].sender_id == "Msg_private_1" + assert seen[0].target.conversation_id == "Msg_private_1" + assert seen[0].target.conversation_type == "private" + assert seen[0].target.extra["send_target"] == "Alice" + assert seen[0].upstream_id == "Msg_private_1:1" + + +def test_ws_and_poll_messages_share_canonical_identity(): + ws_message = _parse_incoming(json.loads(_valid_frame(target="Msg_private_1"))) + poll_message = _parse_polled_message( + { + "id": "m1", + "conversation_id": "Msg_private_1", + "direction": "inbound", + "type": "text", + "text": "hello", + "sent_at": "2026-07-21T07:24:55Z", + }, + {"Msg_private_1": ("Alice", "private")}, + ) + + assert ws_message is not None + assert poll_message is not None + assert ws_message.upstream_id == poll_message.upstream_id == "Msg_private_1:m1" + + +def test_polled_message_without_authoritative_conversation_type_is_dropped(): + message = _parse_polled_message( + { + "id": "m1", + "conversation_id": "Msg_unknown", + "conversation_name": "Unknown chat", + "direction": "inbound", + "type": "text", + "text": "hello", + }, + {}, + ) + + assert message is None + + +@pytest.mark.asyncio +async def test_poll_cursor_overlaps_start_and_advances_for_ws_seen_rows(monkeypatch): + poll_cursors: list[int] = [] + now = 1_000 + + class _PollingClient: + async def poll_messages(self, since: int, *, limit: int = 100): + poll_cursors.append(since) + if len(poll_cursors) == 1: + return [ + { + "id": "m1", + "conversation_id": "Msg_1", + "direction": "inbound", + "type": "text", + "text": "already seen", + "sent_at": "1970-01-01T00:16:40Z", + } + ] + stop.set() + return [] + + async def list_conversations(self): + raise AssertionError("seen rows must not reload conversations") + + async def send_message(self, target, text, *, reply_to=None): + return SendResult(ok=True) + + async def aclose(self) -> None: + return + + monkeypatch.setattr("coding_bridge.channels.wechat.adapter.time.time", lambda: now) + stop = asyncio.Event() + adapter = WeChatAdapter( + instance_id="wisdom", + base_url="https://wisdom.example", + token="tok", + client=_PollingClient(), + ws_connect=_fake_connect([], stop), + stop_event=stop, + poll_interval=0.01, + ) + adapter._seen.add("Msg_1:m1") + adapter.set_handler(lambda *_args: None) + + await asyncio.wait_for(adapter.run(), timeout=1.0) + await adapter.aclose() + + assert poll_cursors == [now - 1, now - 1] + + @pytest.mark.asyncio async def test_adapter_skips_malformed_and_non_inbound_frames(): seen: list[IncomingMessage] = [] @@ -332,7 +589,10 @@ async def handler(_msg: IncomingMessage, _adapter) -> None: call_count += 1 raise RuntimeError("handler exploded") - frames = [_valid_frame(text="one"), _valid_frame(text="two")] + frames = [ + _valid_frame(text="one", message_id="m1"), + _valid_frame(text="two", message_id="m2"), + ] stop = asyncio.Event() adapter = WeChatAdapter( instance_id="cvm-bj",