fix(channels): snapshot outbound listeners per message (#6502)

* fix(channels): snapshot outbound listeners per message

* test(channels): cover listener registration during dispatch

* docs(channels): document outbound listener admission snapshot

* docs(channels): keep guidance under size limit

* docs(channels): clarify outbound snapshot unsubscribe semantics

* test(channels): pin outbound unsubscribe snapshot semantics

* docs(channels): keep outbound snapshot guidance within size limit

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/channels/AGENTS.md

@@ -8,7 +8,7 @@ Cursor writes drain before cancellation; token expiry drains cursor/auth clearin
 **Architecture**: Channels communicate with Gateway through the `langgraph-sdk` HTTP client (same as the frontend), ensuring threads are created and managed server-side. The internal SDK client injects process-local internal auth plus a matching CSRF cookie/header pair so Gateway accepts state-changing thread/run requests from channel workers without relying on browser session cookies.
 
 **Components**:
-- `message_bus.py` - Async pub/sub hub (`InboundMessage` → queue → dispatcher; `OutboundMessage` → callbacks → channels)
+- `message_bus.py` - Inbound queue → dispatcher; outbound uses a dispatch-start listener snapshot. Unsubscribe neither cancels nor drains captured callbacks.
 - `store.py` - JSON persistence for `channel_name:chat_id[:topic_id]` → `thread_id`. Guard every `_data` access with `_lock`; `list_entries()` snapshots under the lock and formats after release so concurrent writers cannot resize the dict during iteration.
 - `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout`
   Streaming `error` frames use the same failure path as transport exceptions. Publish the final outbound before releasing the inbound dedupe key, so provider redelivery can retry without overtaking the terminal reply. Log the frame's error name and message at WARNING with the thread ID; keep the synthetic error generic so payload details do not change reply selection.

backend/app/channels/message_bus.py

@@ -340,19 +340,27 @@ class MessageBus:
         self._outbound_listeners.append(callback)
 
     def unsubscribe_outbound(self, callback: OutboundCallback) -> None:
-        """Remove a previously registered outbound callback."""
+        """Remove a callback from subsequent outbound dispatches.
+
+        This does not cancel or wait for callbacks in an in-flight dispatch.
+        """
         self._outbound_listeners = [cb for cb in self._outbound_listeners if cb != callback]
 
     async def publish_outbound(self, msg: OutboundMessage) -> None:
-        """Dispatch an outbound message to all registered listeners."""
+        """Dispatch to listeners registered when dispatch begins.
+
+        Subscribing or unsubscribing during an in-flight dispatch affects
+        subsequent messages, not callbacks captured for the current message.
+        """
+        listeners = tuple(self._outbound_listeners)
         logger.info(
             "[Bus] outbound dispatching: channel=%s, chat_id=%s, listeners=%d, text_len=%d",
             msg.channel_name,
             msg.chat_id,
-            len(self._outbound_listeners),
+            len(listeners),
             len(msg.text),
         )
-        for callback in self._outbound_listeners:
+        for callback in listeners:
             try:
                 await callback(msg)
             except Exception:

backend/tests/test_message_bus_outbound_lifecycle.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from app.channels.message_bus import MessageBus, OutboundMessage
+
+
+@pytest.mark.asyncio
+async def test_outbound_dispatch_does_not_admit_listener_added_mid_message() -> None:
+    bus = MessageBus()
+    first_started = asyncio.Event()
+    allow_first = asyncio.Event()
+    seen: list[str] = []
+
+    async def first(_msg: OutboundMessage) -> None:
+        seen.append("first")
+        first_started.set()
+        await allow_first.wait()
+
+    async def late(_msg: OutboundMessage) -> None:
+        seen.append("late")
+
+    bus.subscribe_outbound(first)
+    message = OutboundMessage(
+        channel_name="wechat",
+        chat_id="chat-1",
+        thread_id="thread-1",
+        text="reply",
+    )
+
+    dispatch = asyncio.create_task(bus.publish_outbound(message))
+    try:
+        await asyncio.wait_for(first_started.wait(), timeout=1)
+        bus.subscribe_outbound(late)
+    finally:
+        allow_first.set()
+        await dispatch
+
+    assert seen == ["first"]
+
+    await bus.publish_outbound(message)
+    assert seen == ["first", "first", "late"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("removed", ["first", "second"])
+async def test_outbound_dispatch_retains_listener_removed_mid_message(removed: str) -> None:
+    bus = MessageBus()
+    first_started = asyncio.Event()
+    allow_first = asyncio.Event()
+    seen: list[tuple[str, str]] = []
+
+    async def first(msg: OutboundMessage) -> None:
+        first_started.set()
+        await allow_first.wait()
+        seen.append(("first", msg.text))
+
+    async def second(msg: OutboundMessage) -> None:
+        seen.append(("second", msg.text))
+
+    callbacks = {"first": first, "second": second}
+    for callback in callbacks.values():
+        bus.subscribe_outbound(callback)
+    message = OutboundMessage(
+        channel_name="wechat",
+        chat_id="chat-1",
+        thread_id="thread-1",
+        text="in-flight",
+    )
+
+    dispatch = asyncio.create_task(bus.publish_outbound(message))
+    try:
+        await asyncio.wait_for(first_started.wait(), timeout=1)
+        assert seen == [], "the second listener must still be waiting for its turn"
+        bus.unsubscribe_outbound(callbacks[removed])
+    finally:
+        allow_first.set()
+        await dispatch
+
+    assert seen == [("first", "in-flight"), ("second", "in-flight")]
+
+    next_message = OutboundMessage(
+        channel_name="wechat",
+        chat_id="chat-1",
+        thread_id="thread-1",
+        text="next",
+    )
+    await bus.publish_outbound(next_message)
+    remaining = "second" if removed == "first" else "first"
+    assert seen == [("first", "in-flight"), ("second", "in-flight"), (remaining, "next")]