From 5c8275e564b16f730b4b88b8c8fa4ae806609d07 Mon Sep 17 00:00:00 2001 From: Hyu Date: Sat, 10 Oct 2026 22:50:51 +0800 Subject: [PATCH] feat(weixin): show typing indicator during message processing --- .../openclaw_weixin_api/typing_indicator.py | 120 +++++++++++++++ src/langbot/pkg/pipeline/pipelinemgr.py | 5 +- src/langbot/pkg/platform/botmgr.py | 28 ++-- .../pkg/platform/processing_indicator.py | 13 ++ .../pkg/platform/sources/openclaw_weixin.py | 26 ++++ .../platform/test_openclaw_weixin_typing.py | 138 ++++++++++++++++++ 6 files changed, 318 insertions(+), 12 deletions(-) create mode 100644 src/langbot/libs/openclaw_weixin_api/typing_indicator.py create mode 100644 src/langbot/pkg/platform/processing_indicator.py create mode 100644 tests/unit_tests/platform/test_openclaw_weixin_typing.py diff --git a/src/langbot/libs/openclaw_weixin_api/typing_indicator.py b/src/langbot/libs/openclaw_weixin_api/typing_indicator.py new file mode 100644 index 000000000..9334fde0d --- /dev/null +++ b/src/langbot/libs/openclaw_weixin_api/typing_indicator.py @@ -0,0 +1,120 @@ +"""Best-effort, per-peer typing leases for the iLink transport.""" + +import asyncio +import contextlib +import logging +import time +from collections import OrderedDict +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + + +@dataclass +class Lease: + users: int = 0 + task: asyncio.Task | None = None + ticket: str = '' + + +class TypingIndicator: + INTERVAL = 5.0 + TIMEOUT = 3.0 + TICKET_TTL = 600.0 + MAX_PEERS = 4096 + + def __init__(self, client, context_token): + self.client = client + self.context_token = context_token + self.leases: dict[str, Lease] = {} + self.cache = OrderedDict() + self.closed = False + + async def _ticket(self, peer): + now = time.monotonic() + cached = self.cache.get(peer) + if cached and cached[1] > now: + return cached[0] + try: + response = await asyncio.wait_for( + self.client.get_config(peer, self.context_token(peer)), + self.TIMEOUT, + ) + if response.ret not in (None, 0): + raise ValueError('getconfig failed') + ticket = response.typing_ticket or '' + delay = self.TICKET_TTL if ticket else 30.0 + except Exception: + ticket = '' + delay = min((cached[2] * 2 if cached else 5.0), 300.0) + logger.debug('Weixin typing ticket unavailable; retry deferred') + self.cache[peer] = (ticket, now + delay, delay) + self.cache.move_to_end(peer) + while len(self.cache) > self.MAX_PEERS: + self.cache.popitem(last=False) + return ticket + + async def _run(self, peer, lease): + try: + while not self.closed and lease.users: + ticket = await self._ticket(peer) + if ticket: + # Keep the last ticket so cancellation can stop an in-flight start. + lease.ticket = ticket + try: + await asyncio.wait_for(self.client.send_typing(peer, ticket), self.TIMEOUT) + except Exception: + self.cache[peer] = ('', time.monotonic() + 30, 30) + logger.debug('Weixin typing start failed; retry deferred') + await asyncio.sleep(self.INTERVAL) + finally: + if lease.ticket: + try: + # Refresh expired tickets before cancelling a long-running turn. + ticket = await self._ticket(peer) or lease.ticket + await asyncio.wait_for(self.client.stop_typing(peer, ticket), self.TIMEOUT) + except Exception: + logger.debug('Weixin typing stop failed') + + @contextlib.asynccontextmanager + async def processing(self, peer): + lease = None + while not self.closed: + lease = self.leases.get(peer) + if lease is None: + if len(self.leases) >= self.MAX_PEERS: + lease = None + break + lease = Lease(users=1) + self.leases[peer] = lease + lease.task = asyncio.create_task(self._run(peer, lease)) + break + if lease.users: + lease.users += 1 + break + # Serialize stop/start for the same peer: an old stop must not cancel a new start. + await asyncio.gather(asyncio.shield(lease.task), return_exceptions=True) + if self.leases.get(peer) is lease: + self.leases.pop(peer, None) + lease = None + try: + yield + finally: + if lease is not None and lease.users: + lease.users -= 1 + if not lease.users: + if not lease.task.cancelling(): + lease.task.cancel() + await asyncio.gather(asyncio.shield(lease.task), return_exceptions=True) + if self.leases.get(peer) is lease: + self.leases.pop(peer, None) + + async def close(self): + self.closed = True + tasks = [lease.task for lease in self.leases.values()] + for task in tasks: + if not task.done() and not task.cancelling(): + task.cancel() + await asyncio.gather(*(asyncio.shield(task) for task in tasks), return_exceptions=True) + self.leases.clear() + self.cache.clear() diff --git a/src/langbot/pkg/pipeline/pipelinemgr.py b/src/langbot/pkg/pipeline/pipelinemgr.py index 3928f49ee..7ff0d2b4f 100644 --- a/src/langbot/pkg/pipeline/pipelinemgr.py +++ b/src/langbot/pkg/pipeline/pipelinemgr.py @@ -447,7 +447,10 @@ class RuntimePipeline: lane_outcome = 'success' lane_error = '' try: - return await self._process_query(query) + from ..platform.processing_indicator import processing_indicator + + async with processing_indicator(query.adapter, query.launcher_type.value.lower(), query.launcher_id): + return await self._process_query(query) except asyncio.CancelledError: lane_outcome = 'cancelled' lane_error = 'cancelled' diff --git a/src/langbot/pkg/platform/botmgr.py b/src/langbot/pkg/platform/botmgr.py index 7322de873..9ffc4cf57 100644 --- a/src/langbot/pkg/platform/botmgr.py +++ b/src/langbot/pkg/platform/botmgr.py @@ -1272,17 +1272,23 @@ class RuntimeBot: try: with trace_mod.scope(route_ref=self._route_ref(event_binding, target_type, target_uuid)): - async for _ in self.ap.agent_run_orchestrator.run( - envelope, - binding, - adapter_context={ - '_delivery_adapter': adapter, - '_platform_event': event, - '_execution_context': self.execution_context, - }, - ): - # Results are journaled by the orchestrator; platform sends require explicit actions. - pass + from .processing_indicator import processing_indicator + + reply_type, reply_id = (None, None) + if isinstance(event, platform_events.MessageReceivedEvent): + reply_type, reply_id, _ = self._infer_reply_target(event) + async with processing_indicator(adapter, reply_type, reply_id): + async for _ in self.ap.agent_run_orchestrator.run( + envelope, + binding, + adapter_context={ + '_delivery_adapter': adapter, + '_platform_event': event, + '_execution_context': self.execution_context, + }, + ): + # Results are journaled by the orchestrator; platform sends require explicit actions. + pass except Exception: return await self._record_event_route_trace( event_type=event_type, diff --git a/src/langbot/pkg/platform/processing_indicator.py b/src/langbot/pkg/platform/processing_indicator.py new file mode 100644 index 000000000..723efa356 --- /dev/null +++ b/src/langbot/pkg/platform/processing_indicator.py @@ -0,0 +1,13 @@ +"""Optional platform processing status around actual processor execution.""" + +import contextlib + + +@contextlib.asynccontextmanager +async def processing_indicator(adapter, target_type, target_id): + factory = getattr(type(adapter), 'processing_indicator', None) + if callable(factory) and target_type and target_id is not None: + async with factory(adapter, target_type, str(target_id)): + yield + else: + yield diff --git a/src/langbot/pkg/platform/sources/openclaw_weixin.py b/src/langbot/pkg/platform/sources/openclaw_weixin.py index 3991f5555..b4c9dc270 100644 --- a/src/langbot/pkg/platform/sources/openclaw_weixin.py +++ b/src/langbot/pkg/platform/sources/openclaw_weixin.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio import base64 +import contextlib import os import traceback import typing @@ -256,6 +257,8 @@ class OpenClawWeixinAdapter(abstract_platform_adapter.AbstractMessagePlatformAda # context_token cache: from_user_id -> context_token _context_tokens: dict[str, str] = pydantic.PrivateAttr(default_factory=dict) + _typing_indicator: typing.Any = pydantic.PrivateAttr(default=None) + _polling: bool = pydantic.PrivateAttr(default=False) _poll_task: typing.Optional[asyncio.Task] = pydantic.PrivateAttr(default=None) _bot_uuid: typing.Optional[str] = pydantic.PrivateAttr(default=None) @@ -347,6 +350,18 @@ class OpenClawWeixinAdapter(abstract_platform_adapter.AbstractMessagePlatformAda # Persist token to database so it survives restart await self._persist_config() + @contextlib.asynccontextmanager + async def processing_indicator(self, target_type, target_id): + if target_type not in ('person', 'friend'): + yield + return + from langbot.libs.openclaw_weixin_api.typing_indicator import TypingIndicator + + if self._typing_indicator is None: + self._typing_indicator = TypingIndicator(self.client, lambda peer: self._context_tokens.get(peer, '')) + async with self._typing_indicator.processing(target_id): + yield + async def send_message( self, target_type: str, @@ -479,6 +494,9 @@ class OpenClawWeixinAdapter(abstract_platform_adapter.AbstractMessagePlatformAda token=self.config.get('token', token), ) self.bot_account_id = self.config.get('account_id', 'openclaw-weixin') + from langbot.libs.openclaw_weixin_api.typing_indicator import TypingIndicator + + self._typing_indicator = TypingIndicator(self.client, lambda peer: self._context_tokens.get(peer, '')) self._polling = True # Start the long-poll loop @@ -489,6 +507,9 @@ class OpenClawWeixinAdapter(abstract_platform_adapter.AbstractMessagePlatformAda await self._poll_task except asyncio.CancelledError: pass + finally: + if self._typing_indicator is not None: + await self._typing_indicator.close() async def _poll_loop(self): """Long-poll loop: call getUpdates continuously. @@ -523,6 +544,9 @@ class OpenClawWeixinAdapter(abstract_platform_adapter.AbstractMessagePlatformAda await self.logger.error('OpenClaw WeChat session expired, attempting re-login...') try: await self._do_login() + if self._typing_indicator is not None: + await self._typing_indicator.close() + self._typing_indicator = None # Rebuild client with new credentials self.client = OpenClawWeixinClient( base_url=self.config.get('base_url', DEFAULT_BASE_URL), @@ -616,6 +640,8 @@ class OpenClawWeixinAdapter(abstract_platform_adapter.AbstractMessagePlatformAda async def kill(self) -> bool: """Stop the adapter.""" self._polling = False + if self._typing_indicator is not None: + await self._typing_indicator.close() if self._poll_task and not self._poll_task.done(): self._poll_task.cancel() try: diff --git a/tests/unit_tests/platform/test_openclaw_weixin_typing.py b/tests/unit_tests/platform/test_openclaw_weixin_typing.py new file mode 100644 index 000000000..61efe4135 --- /dev/null +++ b/tests/unit_tests/platform/test_openclaw_weixin_typing.py @@ -0,0 +1,138 @@ +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from langbot.libs.openclaw_weixin_api.typing_indicator import TypingIndicator +from langbot.pkg.platform.processing_indicator import processing_indicator + + +def indicator(): + client = SimpleNamespace( + get_config=AsyncMock(return_value=SimpleNamespace(ret=0, typing_ticket='ticket')), + send_typing=AsyncMock(), + stop_typing=AsyncMock(), + ) + manager = TypingIndicator(client, lambda peer: 'context') + manager.INTERVAL = 0.005 + return manager, client + + +async def until(predicate): + async with asyncio.timeout(1): + while not predicate(): + await asyncio.sleep(0.001) + + +@pytest.mark.asyncio +async def test_keepalive_shared_lease_and_cached_ticket(): + manager, client = indicator() + async with manager.processing('peer'): + async with manager.processing('peer'): + await until(lambda: client.send_typing.await_count >= 2) + assert len(manager.leases) == 1 + client.stop_typing.assert_not_awaited() + client.get_config.assert_awaited_once_with('peer', 'context') + client.stop_typing.assert_awaited_once_with('peer', 'ticket') + assert not manager.leases + + +@pytest.mark.asyncio +@pytest.mark.parametrize('failure', ['error', 'cancel']) +async def test_processing_failure_stops_typing(failure): + manager, client = indicator() + + async def work(): + async with manager.processing('peer'): + await until(lambda: client.send_typing.await_count) + if failure == 'error': + raise ValueError('processing failed') + await asyncio.Event().wait() + + task = asyncio.create_task(work()) + await until(lambda: client.send_typing.await_count) + if failure == 'cancel': + task.cancel() + with pytest.raises(ValueError if failure == 'error' else asyncio.CancelledError): + await task + client.stop_typing.assert_awaited_once() + assert not manager.leases + + +@pytest.mark.asyncio +async def test_ticket_failure_is_backed_off_without_interrupting_work(): + manager, client = indicator() + client.get_config.side_effect = OSError('offline') + async with manager.processing('peer'): + await asyncio.sleep(0.03) + client.get_config.assert_awaited_once() + client.send_typing.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_shutdown_and_peer_isolation(): + manager, client = indicator() + async with manager.processing('one'), manager.processing('two'): + await until(lambda: client.send_typing.await_count >= 2) + await manager.close() + assert client.stop_typing.await_count == 2 + count = client.send_typing.await_count + async with manager.processing('three'): + await asyncio.sleep(0.01) + assert client.send_typing.await_count == count + assert not manager.leases + + +@pytest.mark.asyncio +async def test_stop_refreshes_expired_ticket(): + manager, client = indicator() + async with manager.processing('peer'): + await until(lambda: client.send_typing.await_count) + manager.cache['peer'] = ('old', 0, 600) + client.get_config.return_value.typing_ticket = 'fresh' + client.stop_typing.assert_awaited_once_with('peer', 'fresh') + + +@pytest.mark.asyncio +async def test_shutdown_does_not_cancel_stop_already_in_progress(): + manager, client = indicator() + stopping = asyncio.Event() + release = asyncio.Event() + + async def stop(*args): + stopping.set() + await release.wait() + + client.stop_typing.side_effect = stop + + async def work(): + async with manager.processing('peer'): + await until(lambda: client.send_typing.await_count) + + task = asyncio.create_task(work()) + await asyncio.wait_for(stopping.wait(), 1) + shutdown = asyncio.create_task(manager.close()) + await asyncio.sleep(0) + assert not shutdown.done() + release.set() + await asyncio.wait_for(asyncio.gather(task, shutdown), 1) + client.stop_typing.assert_awaited_once() + assert not manager.leases + + +@pytest.mark.asyncio +async def test_optional_adapter_hook(): + async with processing_indicator(object(), 'person', 'peer'): + pass + from langbot.pkg.platform.sources.openclaw_weixin import OpenClawWeixinAdapter + + manager, client = indicator() + adapter = OpenClawWeixinAdapter.model_construct(client=client) + adapter._typing_indicator = manager + async with processing_indicator(adapter, 'group', 'peer'): + await asyncio.sleep(0) + client.get_config.assert_not_awaited() + async with processing_indicator(adapter, 'person', 'peer'): + await until(lambda: client.send_typing.await_count) + client.stop_typing.assert_awaited_once()