feat(weixin): show typing indicator during message processing

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()

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'

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,

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

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:

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()