fix(channels): buffer GitHub follow-ups during busy runs (#4133)

Queue comments received while a run is active, then submit one deduplicated follow-up after it finishes. Failed drains are requeued and watcher tasks stop cleanly with the channel manager.

backend/AGENTS.md

@@ -654,7 +654,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk
 4. For chat: look up/create thread through Gateway's LangGraph-compatible API
 5. Feishu/Telegram chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`)
 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound
-6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound. The agent posts its own reply mid-run via `gh` from the sandbox. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub).
+6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound. The agent posts its own reply mid-run via `gh` from the sandbox. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub); when the channel's policy also sets `buffer_followups_on_busy=True` (GitHub's default — see "Follow-up buffering while busy" below), the triggering message is additionally captured into a per-thread buffer instead of only logged, so a concurrent comment is not silently dropped.
 7. Feishu channel sends one running reply card up front, then patches the same card for each outbound update (card JSON sets `config.update_multi=true` for Feishu's patch API requirement). Messages already sent inside an existing Feishu topic carry a compact source-message preview in that card, and queued same-thread follow-ups patch their own source message's card from queued → running → final without falling back to the generic busy reply.
 8. Telegram streaming: the "Working on it..." placeholder message is registered as the stream target; non-final updates `editMessageText` it in place (channel-side throttle: 1s in private chats, 3s in groups due to Telegram's 20 msg/min group cap; 4096-char truncation; rate-limited updates dropped); the final update performs the last edit and splits >4096 texts into follow-up messages
 9. DingTalk AI Card mode (when `card_template_id` configured): `runs.stream()` → create card with initial text → stream updates via `PUT /v1.0/card/streaming` → finalize on `is_final=True`. Falls back to `sampleMarkdown` if card creation or streaming fails
@@ -691,6 +691,7 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
 **GitHub event-driven agents** (webhook-driven IM channel):
 - Custom agents declare a `github:` block in their `config.yaml` to bind to repos and event triggers; the webhook route is fail-closed by default (mounted only when `GITHUB_WEBHOOK_SECRET` is set) and exempt from auth/CSRF because authenticity is enforced by HMAC.
 - Outbound is **log-only** by design: each agent posts its own reply mid-run via the `gh` CLI from its sandbox, so the manager uses `fire_and_forget=True` and `runs.create()` returns once pending.
+- **Follow-up buffering while busy** (issue #4121): because outbound is log-only, the pre-existing `THREAD_BUSY_MESSAGE` reply on a `ConflictError` was invisible to the commenter — a comment posted while a run was already active looked like it had been silently ignored. When `ChannelRunPolicy.buffer_followups_on_busy=True` (GitHub's default), a `ConflictError` on `runs.create()` now also appends the triggering message to a per-thread, in-memory buffer (`ChannelManager._followup_buffers`) — deduped by GitHub delivery id, capped at `FOLLOWUP_BUFFER_MAX_PER_THREAD` (20, oldest dropped with a WARNING log on overflow). The first successful `runs.create()` on a thread now captures its `run_id` and spawns a background watcher that subscribes to that run's `StreamBridge` stream; once it observes `END_SENTINEL`, the watcher drains up to `FOLLOWUP_DRAIN_BATCH_SIZE` (10) buffered entries into one `<followups-while-busy>`-wrapped input and fires a follow-up `runs.create()` — itself watched the same way, so a backlog deeper than one batch chains into further drain cycles instead of growing one unbounded input. If that follow-up `runs.create()` itself hits `ConflictError` (e.g. a manual Web UI turn or a scheduled run raced onto the same thread), the batch is requeued rather than lost, and is retried whenever this manager next successfully creates and watches a run on that thread. Reactions/acknowledgment (e.g. GitHub's `eyes`/`confused` reaction API) on buffered comments are intentionally **out of scope** for this mechanism and left to a follow-up — comments are coalesced silently. **Plumbing**: the watcher needs the Gateway's `StreamBridge` singleton, which `ChannelManager` did not previously have access to; it is threaded from `app.py`'s lifespan (where `app.state.stream_bridge` is already set by `langgraph_runtime`) through `start_channel_service(get_stream_bridge=...)` → `ChannelService.__init__` → `ChannelManager.__init__`, as a zero-arg closure mirroring the existing `launch_run=lambda **kwargs: launch_scheduled_thread_run(app=app, **kwargs)` pattern used for `ScheduledTaskService` in the same lifespan function. A `ChannelManager` constructed without it (e.g. directly in a test) still buffers safely — it just has no watcher to auto-drain. **Scope limitation**: the buffer and watcher state are per-process, in-memory. Under `GATEWAY_WORKERS>1` or multi-pod, a follow-up comment routed to a different worker process than the one running the busy thread's agent will not see that buffer. This is a known, deliberately deferred limitation with the same shape as the cross-pod gap described for issue #4120 (a shared buffer store or IM-leader election would be needed to close it) — single-process/single-pod deployments, the safe default, see no correctness issue from this, only the documented per-process scope.
 - See [backend/docs/GITHUB_AGENTS.md](docs/GITHUB_AGENTS.md) for the architecture diagrams: webhook → fan-out → `InboundMessage` dispatch, `preferred_thread_id = UUID5(repo, number, agent_name)` thread determinism, mention-handle precedence chain, GH token lifecycle via `GH_TOKEN`/`GITHUB_TOKEN` per-call `extra_env`, and the narrow `ConflictError` (HTTP 409) thread-create race recovery.
 
 

backend/app/channels/manager.py

@@ -10,6 +10,7 @@ import time
 from collections import OrderedDict
 from collections.abc import Awaitable, Callable, Mapping
 from dataclasses import dataclass
+from html import escape
 from pathlib import Path
 from typing import Any
 from urllib.parse import quote
@@ -37,6 +38,7 @@ from app.gateway.github import run_policy as _github_run_policy  # noqa: F401
 from app.gateway.internal_auth import create_internal_auth_headers
 from deerflow.config.agents_config import load_agent_config
 from deerflow.config.paths import make_safe_user_id
+from deerflow.runtime import END_SENTINEL, StreamBridge
 from deerflow.runtime.goal import parse_goal_command
 from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.skills.slash import parse_slash_skill_reference
@@ -102,6 +104,15 @@ BOUND_IDENTITY_UNAVAILABLE_MESSAGE = "Channel connection verification is tempora
 # mechanism, same TTL), not a channel-specific gap. True idempotency against
 # a late/manual redelivery would require persisting the dedupe key in
 # ``ChannelStore`` instead, which is not implemented here.
+# Follow-up buffering for busy fire_and_forget threads (issue #4121 Slice 2).
+# A ConflictError on a channel opted into ChannelRunPolicy.buffer_followups_on_busy
+# buffers the triggering message per-thread instead of only logging it; a
+# background watcher drains the buffer into a coalesced follow-up run once the
+# busy run's StreamBridge stream reaches END_SENTINEL. See _buffer_followup,
+# _drain_followups_for_thread, and _watch_run_and_drain_followups below.
+FOLLOWUP_BUFFER_MAX_PER_THREAD = 20
+FOLLOWUP_DRAIN_BATCH_SIZE = 10
+FOLLOWUP_BLOCK_TAG = "followups-while-busy"
 INBOUND_DEDUPE_TTL_SECONDS = 10 * 60
 INBOUND_DEDUPE_MAX_ENTRIES = 4096
 # Only server-stable provider message ids: client-generated ids (client_msg_id,
@@ -225,6 +236,23 @@ class _SerializedThreadRunState:
     waiters: int = 0
 
 
+@dataclass(slots=True)
+class _FollowupEntry:
+    """One inbound message's text, buffered because its thread was busy.
+
+    Routing/policy identity (channel_name, metadata, owner headers) for the
+    eventual drained run comes from a separate ``carrier_msg`` — see
+    ``ChannelManager._drain_followups_for_thread`` — not from a per-entry
+    message, since every buffered entry for one thread_id shares that
+    identity already (thread_id is itself derived deterministically from
+    (repo, number, agent_name) for GitHub). Only the text needs to survive
+    per entry.
+    """
+
+    dedupe_key: str
+    text: str
+
+
 def _is_thread_busy_error(exc: BaseException | None) -> bool:
     if exc is None:
         return False
@@ -233,6 +261,59 @@ def _is_thread_busy_error(exc: BaseException | None) -> bool:
     return "already running a task" in str(exc)
 
 
+def _followup_dedupe_key(msg: InboundMessage) -> str:
+    """Best-effort stable identifier for a buffered follow-up comment.
+
+    Mirrors ``_inbound_dedupe_key``'s provider-id preference order (a GitHub
+    webhook delivery id first, then the generic provider-message-id metadata
+    keys), but scoped to one thread's follow-up buffer rather than the
+    global cross-channel inbound dedupe map, and always returns a usable key
+    — falling back to an object-identity key — since the follow-up buffer
+    must still accept an entry even when a provider omits every known id
+    field (unlike ``_inbound_dedupe_key``, which returns ``None`` to skip
+    dedupe entirely in that case).
+    """
+    metadata = msg.metadata or {}
+    gh = metadata.get("github")
+    if isinstance(gh, dict):
+        delivery_id = gh.get("delivery_id")
+        if delivery_id:
+            return f"github:delivery:{delivery_id}"
+
+    for key in INBOUND_DEDUPE_METADATA_KEYS:
+        value = metadata.get(key)
+        if value:
+            return f"{key}:{value}"
+
+    raw_message = metadata.get("raw_message")
+    if isinstance(raw_message, Mapping):
+        for key in INBOUND_DEDUPE_METADATA_KEYS:
+            value = raw_message.get(key)
+            if value:
+                return f"{key}:{value}"
+
+    # No stable provider id available: fall back to a per-message key so the
+    # entry is still buffered (just never deduped against a redelivery).
+    return f"__no_id__:{id(msg)}:{msg.created_at}"
+
+
+def _format_followup_block(entries: list[_FollowupEntry]) -> str:
+    """Coalesce buffered follow-up entries into one templated input block."""
+    lines = [
+        f"<{FOLLOWUP_BLOCK_TAG}>",
+        "The following messages arrived on this thread while a previous run was still in progress. They were queued and are now delivered together as one turn:",
+        "",
+    ]
+    for idx, entry in enumerate(entries, start=1):
+        escaped_text = escape(entry.text, quote=False).replace(
+            "\n",
+            "\n   ",
+        )
+        lines.append(f"{idx}. {escaped_text}")
+    lines.append(f"</{FOLLOWUP_BLOCK_TAG}>")
+    return "\n".join(lines)
+
+
 def _as_dict(value: Any) -> dict[str, Any]:
     return dict(value) if isinstance(value, Mapping) else {}
 
@@ -829,6 +910,7 @@ class ChannelManager:
         channel_sessions: dict[str, Any] | None = None,
         connection_repo: Any | None = None,
         require_bound_identity: bool = False,
+        get_stream_bridge: Callable[[], StreamBridge | None] | None = None,
     ) -> None:
         self.bus = bus
         self.store = store
@@ -840,6 +922,14 @@ class ChannelManager:
         self._channel_sessions = dict(channel_sessions or {})
         self._connection_repo = connection_repo
         self._require_bound_identity = require_bound_identity
+        # Zero-arg accessor for the FastAPI app's StreamBridge singleton,
+        # threaded in from app.py's lifespan via start_channel_service() ->
+        # ChannelService.__init__ (mirrors how ScheduledTaskService gets a
+        # launch_run closure over `app` in the same lifespan function). None
+        # when not wired (e.g. a ChannelManager constructed directly in
+        # tests) — follow-up buffering still works, but no watcher is
+        # spawned to auto-drain it (see _maybe_spawn_followup_watcher).
+        self._get_stream_bridge = get_stream_bridge
         self._client = None  # lazy init — langgraph_sdk async client
         self._channel_metadata_synced: set[str] = set()
         # Per-conversation locks so concurrent inbound messages for the same
@@ -852,11 +942,30 @@ class ChannelManager:
         self._csrf_token = generate_csrf_token()
         self._semaphore: asyncio.Semaphore | None = None
         self._running = False
+        # Distinct from self._running: that flag is also False before the
+        # very first start() (so tests that call internal drain/handler
+        # methods directly without going through start()/stop() keep
+        # working unchanged). self._stopped tracks specifically whether
+        # stop() has run, for the follow-up drain guard below.
+        self._stopped = False
         self._task: asyncio.Task | None = None
         # Insertion order == chronological (keys are never re-inserted), so an
         # OrderedDict lets us evict expired/overflow entries from the front in
         # O(k) instead of scanning all entries on every inbound message.
         self._recent_inbound_events: OrderedDict[tuple[str, str, str, str], float] = OrderedDict()
+        # Per-thread follow-up buffers for busy fire_and_forget channels that
+        # opted into ChannelRunPolicy.buffer_followups_on_busy (issue #4121
+        # Slice 2). Keyed by thread_id -> OrderedDict[dedupe_key -> entry],
+        # oldest-first, mirroring _recent_inbound_events's shape but scoped
+        # per-thread with a hard cap instead of a global TTL (see
+        # _buffer_followup / _enforce_followup_cap).
+        self._followup_buffers: dict[str, OrderedDict[str, _FollowupEntry]] = {}
+        # Background watcher tasks spawned by _maybe_spawn_followup_watcher,
+        # tracked so stop() can cancel+await them instead of leaving them as
+        # orphaned fire-and-forget tasks that could still fire a follow-up
+        # run after this manager has been shut down. Discarded via the same
+        # task's done-callback (see _maybe_spawn_followup_watcher).
+        self._followup_watcher_tasks: set[asyncio.Task] = set()
 
     @staticmethod
     def _channel_supports_streaming(channel_name: str) -> bool:
@@ -911,6 +1020,271 @@ class ChannelManager:
         if state.waiters == 0 and not state.lock.locked():
             self._serialized_thread_runs.pop((channel_name, thread_id), None)
 
+    # -- follow-up buffering for busy fire_and_forget threads (issue #4121) --
+
+    def _resolve_stream_bridge(self) -> StreamBridge | None:
+        """Resolve the current StreamBridge via the injected accessor, if any."""
+        if self._get_stream_bridge is None:
+            return None
+        try:
+            return self._get_stream_bridge()
+        except Exception:
+            logger.exception("[Manager] get_stream_bridge callable raised; follow-up watch disabled for this run")
+            return None
+
+    def _enforce_followup_cap(self, thread_id: str, buffer: OrderedDict[str, _FollowupEntry]) -> None:
+        """Drop the OLDEST buffered entries once *buffer* exceeds the per-thread cap.
+
+        Dropping the oldest (rather than the newest, incoming) entry means a
+        thread that is deep enough in the backlog to hit the cap still keeps
+        the most recent activity — a better signal for the eventual coalesced
+        turn than the stalest queued comment. No reaction/acknowledgment is
+        sent on drop (out of scope for this slice); a WARNING is logged so
+        operators can see it in gateway.log.
+        """
+        while len(buffer) > FOLLOWUP_BUFFER_MAX_PER_THREAD:
+            dropped_key, _ = buffer.popitem(last=False)
+            logger.warning(
+                "[Manager] follow-up buffer overflow for thread_id=%s (cap=%d); dropped oldest buffered comment (dedupe_key=%s)",
+                thread_id,
+                FOLLOWUP_BUFFER_MAX_PER_THREAD,
+                dropped_key,
+            )
+
+    def _buffer_followup(self, thread_id: str, msg: InboundMessage) -> None:
+        """Append *msg* to thread_id's follow-up buffer (ConflictError path).
+
+        Dedupe mirrors ``_is_duplicate_inbound``'s OrderedDict idiom, scoped
+        per-thread instead of global: a redelivered webhook for a comment
+        already buffered (same dedupe key) is a no-op instead of a second
+        entry.
+        """
+        key = _followup_dedupe_key(msg)
+        buffer = self._followup_buffers.setdefault(thread_id, OrderedDict())
+        if key in buffer:
+            logger.info(
+                "[Manager] duplicate follow-up ignored for thread_id=%s (dedupe_key=%s)",
+                thread_id,
+                key,
+            )
+            return
+
+        buffer[key] = _FollowupEntry(dedupe_key=key, text=msg.text)
+        self._enforce_followup_cap(thread_id, buffer)
+        logger.info(
+            "[Manager] buffered follow-up for busy thread_id=%s (dedupe_key=%s, buffered=%d)",
+            thread_id,
+            key,
+            len(buffer),
+        )
+
+    def _pop_followup_batch(self, thread_id: str, *, limit: int) -> list[_FollowupEntry]:
+        """Pop up to *limit* buffered entries FIFO (oldest first)."""
+        buffer = self._followup_buffers.get(thread_id)
+        if not buffer:
+            return []
+
+        batch: list[_FollowupEntry] = []
+        for _ in range(min(limit, len(buffer))):
+            _, entry = buffer.popitem(last=False)
+            batch.append(entry)
+
+        if not buffer:
+            self._followup_buffers.pop(thread_id, None)
+        return batch
+
+    def _requeue_followups(self, thread_id: str, entries: list[_FollowupEntry]) -> None:
+        """Put a popped batch back at the front of the buffer (oldest-first).
+
+        Used when the drain's own ``runs.create`` call itself fails —
+        including the ``ConflictError`` edge case where something this
+        manager did not create (a manual Web UI turn, a scheduled run) is
+        occupying the thread. The entries are not lost: the next time *any*
+        run this manager creates on this thread completes, its watcher will
+        attempt another drain and find them still buffered.
+        """
+        if not entries:
+            return
+
+        existing = self._followup_buffers.get(thread_id, OrderedDict())
+        merged: OrderedDict[str, _FollowupEntry] = OrderedDict()
+        for entry in entries:
+            merged[entry.dedupe_key] = entry
+        for key, entry in existing.items():
+            merged.setdefault(key, entry)
+
+        self._enforce_followup_cap(thread_id, merged)
+        self._followup_buffers[thread_id] = merged
+
+    def _maybe_spawn_followup_watcher(
+        self,
+        thread_id: str,
+        run_result: Any,
+        carrier_msg: InboundMessage,
+    ) -> None:
+        """Spawn a background watcher for a just-created run, if wired up.
+
+        No-ops (spawns nothing) when no ``get_stream_bridge`` accessor was
+        threaded in — e.g. a ``ChannelManager`` constructed directly without
+        going through ``start_channel_service()`` — so tests and any
+        not-yet-wired deployment never see a dangling background task for
+        this. When wired, mirrors the existing ``_dispatch_loop`` pattern:
+        ``asyncio.create_task`` + ``add_done_callback(self._log_task_error)``
+        so an unexpected watcher failure is surfaced in the logs instead of
+        silently vanishing. The task is also tracked in
+        ``self._followup_watcher_tasks`` (discarded via its own done-callback)
+        so ``stop()`` can cancel+await any watcher still in flight instead of
+        leaving it to fire a follow-up run after shutdown.
+        """
+        if self._get_stream_bridge is None:
+            return
+
+        run_id = run_result.get("run_id") if isinstance(run_result, dict) else None
+        if not run_id:
+            logger.warning(
+                "[Manager] runs.create returned no run_id for thread_id=%s; cannot watch for follow-up drain",
+                thread_id,
+            )
+            return
+
+        task = asyncio.create_task(self._watch_run_and_drain_followups(thread_id, run_id, carrier_msg))
+        self._followup_watcher_tasks.add(task)
+        task.add_done_callback(self._followup_watcher_tasks.discard)
+        task.add_done_callback(self._log_task_error)
+
+    async def _watch_run_and_drain_followups(
+        self,
+        thread_id: str,
+        run_id: str,
+        carrier_msg: InboundMessage,
+    ) -> None:
+        """Watch *run_id* until it ends, then attempt to drain thread_id's buffer.
+
+        Subscribes to the StreamBridge the same way existing consumers do
+        (``entry is END_SENTINEL``, see ``app/gateway/services.py``). Runs
+        for as long as the underlying run does — GitHub coding runs
+        routinely take several minutes, so this deliberately does not apply
+        an artificial timeout, mirroring why the dispatch path itself uses
+        ``runs.create`` instead of ``runs.wait`` in the first place. Draining
+        is a no-op when the buffer is empty, which is the common case (most
+        runs never hit a busy-thread conflict).
+        """
+        stream_bridge = self._resolve_stream_bridge()
+        if stream_bridge is None:
+            logger.warning(
+                "[Manager] no stream bridge available; cannot watch run_id=%s for thread_id=%s follow-up drain (any buffered follow-ups will be drained by a later watched run on this thread)",
+                run_id,
+                thread_id,
+            )
+            return
+
+        try:
+            async for entry in stream_bridge.subscribe(run_id):
+                if entry is END_SENTINEL:
+                    break
+        except Exception:
+            logger.exception(
+                "[Manager] error watching run_id=%s for thread_id=%s follow-up drain",
+                run_id,
+                thread_id,
+            )
+            return
+
+        client = self._get_client()
+        await self._drain_followups_for_thread(client, thread_id, carrier_msg)
+
+    async def _drain_followups_for_thread(
+        self,
+        client,
+        thread_id: str,
+        carrier_msg: InboundMessage,
+    ) -> None:
+        """Coalesce up to one batch of buffered follow-ups into a fresh run.
+
+        ``carrier_msg`` supplies routing/policy identity (channel_name,
+        metadata, owner headers) for the drained run — it is safe to reuse
+        across an entire drain chain because every buffered entry for one
+        thread_id shares that identity (thread_id itself is derived
+        deterministically from (repo, number, agent_name) for GitHub).
+
+        A batch larger than ``FOLLOWUP_DRAIN_BATCH_SIZE`` is intentionally
+        NOT drained in one shot: only the oldest batch is popped here, and
+        the run created for it is itself watched (via
+        ``_maybe_spawn_followup_watcher``), so a deeper backlog chains into
+        another drain cycle once this run ends, rather than growing one
+        unbounded coalesced input block.
+
+        If anything from here through ``runs.create`` fails — resolving run
+        params, applying channel policy, or ``runs.create`` itself
+        (including ``ConflictError`` from something this manager did not
+        create racing onto the same thread) — the popped batch is requeued
+        (not lost) and this coroutine returns without raising or looping:
+        the next run this manager successfully creates and watches on this
+        thread will attempt the drain again.
+
+        No-ops if the manager has already been ``stop()``-ped: a watcher
+        task that slips past its own cancellation and reaches this point
+        after shutdown must not fire a brand new run into a stopped manager.
+        (Deliberately keyed on ``self._stopped``, not ``self._running`` —
+        the latter is also ``False`` before the very first ``start()``,
+        which would otherwise make this guard fire for callers that invoke
+        the drain directly without going through the dispatch lifecycle.)
+        """
+        if self._stopped:
+            logger.info(
+                "[Manager] skipping follow-up drain for thread_id=%s; manager is stopped",
+                thread_id,
+            )
+            return
+
+        entries = self._pop_followup_batch(thread_id, limit=FOLLOWUP_DRAIN_BATCH_SIZE)
+        if not entries:
+            return
+
+        logger.info(
+            "[Manager] draining %d buffered follow-up(s) for thread_id=%s",
+            len(entries),
+            thread_id,
+        )
+        try:
+            # Everything from here through runs.create is covered by the
+            # same except below: a pre-create failure (e.g. the target agent
+            # config was removed mid-run, or channel-policy/credential
+            # resolution raises) must requeue the popped batch exactly like
+            # a runs.create failure does — none of these steps get to
+            # silently drop entries that were already popped off the buffer.
+            assistant_id, run_config, run_context = self._resolve_run_params(carrier_msg, thread_id)
+            await self._apply_channel_policy(carrier_msg, run_context)
+
+            human_message = _human_input_message(_format_followup_block(entries))
+            run_kwargs: dict[str, Any] = {
+                "input": {"messages": [human_message]},
+                "config": run_config,
+                "context": run_context,
+                "multitask_strategy": "reject",
+            }
+            if owner_headers := _owner_headers(carrier_msg):
+                run_kwargs["headers"] = owner_headers
+
+            result = await client.runs.create(thread_id, assistant_id, **run_kwargs)
+        except Exception as exc:
+            if _is_thread_busy_error(exc):
+                logger.warning(
+                    "[Manager] follow-up drain hit a busy thread_id=%s (a run this manager did not create is active); re-buffering %d entries",
+                    thread_id,
+                    len(entries),
+                )
+            else:
+                logger.exception(
+                    "[Manager] follow-up drain failed for thread_id=%s; re-buffering %d entries",
+                    thread_id,
+                    len(entries),
+                )
+            self._requeue_followups(thread_id, entries)
+            return
+
+        self._maybe_spawn_followup_watcher(thread_id, result, carrier_msg)
+
     async def _publish_progress_update(self, msg: InboundMessage, thread_id: str, text: str) -> None:
         await self.bus.publish_outbound(
             OutboundMessage(
@@ -1113,6 +1487,7 @@ class ChannelManager:
         if self._running:
             return
         self._running = True
+        self._stopped = False
         self._semaphore = asyncio.Semaphore(self._max_concurrency)
         self._task = asyncio.create_task(self._dispatch_loop())
         logger.info("ChannelManager started (max_concurrency=%d)", self._max_concurrency)
@@ -1120,6 +1495,7 @@ class ChannelManager:
     async def stop(self) -> None:
         """Stop the dispatch loop."""
         self._running = False
+        self._stopped = True
         if self._task:
             self._task.cancel()
             try:
@@ -1127,6 +1503,25 @@ class ChannelManager:
             except asyncio.CancelledError:
                 pass
             self._task = None
+
+        # Follow-up watchers are long-lived background tasks (they await a
+        # run's full stream, which can take minutes) started outside the
+        # dispatch loop, so cancelling self._task above does not touch them.
+        # Left unmanaged, one still subscribed to a run that ends AFTER this
+        # point would drain its buffer and fire a brand new runs.create()
+        # into a manager that has already been stopped.
+        watcher_tasks = list(self._followup_watcher_tasks)
+        for task in watcher_tasks:
+            task.cancel()
+        for task in watcher_tasks:
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+            except Exception:
+                logger.exception("[Manager] follow-up watcher task raised during stop()")
+        self._followup_watcher_tasks.clear()
+
         logger.info("ChannelManager stopped")
 
     # -- dispatch loop -----------------------------------------------------
@@ -1647,17 +2042,27 @@ class ChannelManager:
                 len(msg.text or ""),
             )
             try:
-                await client.runs.create(thread_id, assistant_id, **run_kwargs)
+                # Capturing the return value is new (issue #4121 Slice 2):
+                # it carries ``run_id``, which the follow-up watcher below
+                # needs to subscribe to this run's StreamBridge stream. When
+                # ``buffer_followups_on_busy`` is off this is otherwise
+                # behaviorally identical to the previous bare ``await``.
+                result = await client.runs.create(thread_id, assistant_id, **run_kwargs)
             except Exception as exc:
                 if _is_thread_busy_error(exc):
                     logger.warning("[Manager] thread busy (concurrent run rejected): thread_id=%s", thread_id)
-                    # Swallowed like the generic handler would not be: release the
-                    # key so the provider's redelivery can retry once the thread
-                    # frees, instead of being dropped for the dedupe TTL.
-                    self._release_inbound_dedupe_key(msg)
+                    if policy.buffer_followups_on_busy:
+                        self._buffer_followup(thread_id, msg)
+                    else:
+                        # Swallowed like the generic handler would not be: release the
+                        # key so the provider's redelivery can retry once the thread
+                        # frees, instead of being dropped for the dedupe TTL.
+                        self._release_inbound_dedupe_key(msg)
                     await self._send_error(msg, THREAD_BUSY_MESSAGE)
                     return
                 raise
+            if policy.buffer_followups_on_busy:
+                self._maybe_spawn_followup_watcher(thread_id, result, msg)
             return
 
         logger.info("[Manager] invoking runs.wait(thread_id=%s, text_len=%d)", thread_id, len(msg.text or ""))

backend/app/channels/run_policy.py

@@ -84,6 +84,21 @@ class ChannelRunPolicy:
             unrelated DeerFlow threads continue concurrently. Defaults
             to False so existing channels keep the runtime's native
             multitask behavior unless they opt in explicitly.
+        buffer_followups_on_busy: When True, a ``ConflictError`` on the
+            ``fire_and_forget`` dispatch path (see
+            :meth:`ChannelManager._handle_chat_on_thread`) does more than
+            log + reply with the generic busy message: the triggering
+            message is appended to a per-thread follow-up buffer, and a
+            background watcher subscribes to the active run's
+            ``StreamBridge`` stream so it can coalesce the buffer into a
+            follow-up run as soon as that run ends. This targets
+            ``fire_and_forget`` channels whose ``send`` is otherwise the
+            only feedback a busy sender gets (e.g. GitHub, where ``send``
+            is log-only) — without it, a concurrent comment is silently
+            dropped from the sender's point of view. Defaults to False so
+            channels that have not opted in keep the exact old
+            silent-drop-with-log behavior; see
+            ``app.gateway.github.run_policy`` for GitHub's opt-in.
     """
 
     is_interactive: bool = True
@@ -92,6 +107,7 @@ class ChannelRunPolicy:
     requires_bound_identity: bool = True
     fire_and_forget: bool = False
     serialize_thread_runs: bool = False
+    buffer_followups_on_busy: bool = False
 
 
 # Channel name → policy. Channels absent from this map fall through to

backend/app/channels/service.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 import asyncio
 import logging
 import os
+from collections.abc import Callable
 from typing import TYPE_CHECKING, Any
 
 from app.channels.base import Channel
@@ -18,6 +19,7 @@ logger = logging.getLogger(__name__)
 if TYPE_CHECKING:
     from deerflow.config.app_config import AppConfig
     from deerflow.config.channel_connections_config import ChannelConnectionsConfig
+    from deerflow.runtime import StreamBridge
 
 # Channel name → import path for lazy loading
 _CHANNEL_REGISTRY: dict[str, str] = {
@@ -97,10 +99,12 @@ class ChannelService:
         *,
         connection_repo: Any | None = None,
         require_bound_identity: bool = False,
+        get_stream_bridge: Callable[[], StreamBridge | None] | None = None,
     ) -> None:
         self.bus = MessageBus()
         self.store = ChannelStore()
         self._connection_repo = connection_repo
+        self._get_stream_bridge = get_stream_bridge
         config = dict(channels_config or {})
         langgraph_url = _resolve_service_url(config, "langgraph_url", _CHANNELS_LANGGRAPH_URL_ENV, DEFAULT_LANGGRAPH_URL)
         gateway_url = _resolve_service_url(config, "gateway_url", _CHANNELS_GATEWAY_URL_ENV, DEFAULT_GATEWAY_URL)
@@ -115,6 +119,7 @@ class ChannelService:
             channel_sessions=channel_sessions,
             connection_repo=connection_repo,
             require_bound_identity=require_bound_identity,
+            get_stream_bridge=get_stream_bridge,
         )
         self._channels: dict[str, Any] = {}  # name -> Channel instance
         self._config = config
@@ -122,8 +127,19 @@ class ChannelService:
         self._readiness_locks: dict[str, asyncio.Lock] = {}
 
     @classmethod
-    def from_app_config(cls, app_config: AppConfig | None = None) -> ChannelService:
-        """Create a ChannelService from the application config."""
+    def from_app_config(
+        cls,
+        app_config: AppConfig | None = None,
+        *,
+        get_stream_bridge: Callable[[], StreamBridge | None] | None = None,
+    ) -> ChannelService:
+        """Create a ChannelService from the application config.
+
+        ``get_stream_bridge`` is threaded straight through to the
+        ``ChannelManager`` (see its docstring); it is optional so direct
+        callers (including most tests) that don't need follow-up-buffer
+        auto-draining can omit it.
+        """
         if app_config is None:
             from deerflow.config.app_config import get_app_config
 
@@ -141,6 +157,7 @@ class ChannelService:
             channels_config=channels_config,
             connection_repo=_make_connection_repo(connection_config),
             require_bound_identity=require_bound_identity,
+            get_stream_bridge=get_stream_bridge,
         )
 
     async def start(self) -> None:
@@ -407,14 +424,27 @@ def get_channel_service() -> ChannelService | None:
     return _channel_service
 
 
-async def start_channel_service(app_config: AppConfig | None = None) -> ChannelService:
-    """Create and start the global ChannelService from app config."""
+async def start_channel_service(
+    app_config: AppConfig | None = None,
+    *,
+    get_stream_bridge: Callable[[], StreamBridge | None] | None = None,
+) -> ChannelService:
+    """Create and start the global ChannelService from app config.
+
+    ``get_stream_bridge`` is threaded through to ``ChannelService.from_app_config``
+    -> ``ChannelManager`` so fire_and_forget channels that opt into
+    ``ChannelRunPolicy.buffer_followups_on_busy`` (currently GitHub) can watch
+    a run's completion and auto-drain buffered follow-ups. ``app.py``'s
+    lifespan passes a closure over ``app.state.stream_bridge`` here, the same
+    pattern it already uses for ``ScheduledTaskService``'s ``launch_run``.
+    """
     global _channel_service
     if _channel_service is not None:
         return _channel_service
     # from_app_config reads the JSON channel store and runtime config files;
-    # keep that disk IO off the event loop.
-    _channel_service = await asyncio.to_thread(ChannelService.from_app_config, app_config)
+    # keep that disk IO off the event loop. asyncio.to_thread forwards both
+    # args and kwargs to the target callable.
+    _channel_service = await asyncio.to_thread(ChannelService.from_app_config, app_config, get_stream_bridge=get_stream_bridge)
     await _channel_service.start()
     return _channel_service
 

backend/app/gateway/app.py

@@ -250,7 +250,19 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         try:
             from app.channels.service import start_channel_service
 
-            channel_service = await start_channel_service(startup_config)
+            # Closure over `app` (mirrors ScheduledTaskService's `launch_run`
+            # below) rather than resolving `app.state.stream_bridge` here
+            # directly: `stream_bridge` is a STARTUP_ONLY_FIELDS singleton set
+            # once, above, by `langgraph_runtime(app, startup_config)`, so
+            # either shape is safe by construction — the closure is just the
+            # more defensive/consistent-with-precedent form, and it is what
+            # ChannelManager's follow-up-drain watcher (issue #4121 Slice 2)
+            # uses to reach the same StreamBridge every other run consumer
+            # goes through `get_stream_bridge(request)` for.
+            channel_service = await start_channel_service(
+                startup_config,
+                get_stream_bridge=lambda: getattr(app.state, "stream_bridge", None),
+            )
             logger.info("Channel service started: %s", channel_service.get_status())
         except Exception:
             logger.exception("No IM channels configured or channel service failed to start")

backend/app/gateway/github/run_policy.py

@@ -129,6 +129,15 @@ def register_policy() -> None:
         # still raised synchronously by ``start_run`` before the run is
         # accepted, so the busy-thread path is preserved.
         fire_and_forget=True,
+        # GitHub's ``send`` is log-only (agents post via ``gh`` themselves),
+        # so a busy-thread ``THREAD_BUSY_MESSAGE`` is otherwise invisible to
+        # the commenter — a concurrent comment is silently dropped from
+        # their point of view (issue #4121). Buffering + draining follow-ups
+        # once the busy run ends directly fixes that for the one channel
+        # where it is a real, routinely-triggered problem; other
+        # fire_and_forget channels keep the dataclass default (False) until
+        # they have the same log-only-send shape and opt in explicitly.
+        buffer_followups_on_busy=True,
     )
 
 

backend/docs/GITHUB_AGENTS.md

@@ -11,6 +11,7 @@ This document covers the **architecture** of that pipeline:
 - GH token lifecycle (`GITHUB_APP_ID` + `PRIVATE_KEY` → `run_context["github_token"]` → sandbox `GH_TOKEN`/`GITHUB_TOKEN`)
 - `ConflictError` (HTTP 409) thread-create race recovery
 - Why **outbound is log-only** (agents post via `gh` from their sandbox)
+- Follow-up buffering while busy (issue #4121): buffer → watch → drain, and the single-process scope limit
 
 ## Overview
 
@@ -218,6 +219,49 @@ The recovery is **narrow**: only `langgraph_sdk.errors.ConflictError` (HTTP 409)
 
 The follow-up `threads.get(preferred_thread_id)` is itself verified before caching — if it also rejects, the store underneath is in an inconsistent state and the failure surfaces.
 
+## Follow-up Buffering While Busy (#4121)
+
+A **different** conflict from the one above: not two deliveries racing to *create* a thread, but a new comment arriving while a *run* is already active on an existing thread. `runs.create()` raises `ConflictError` for that too, and until this fix the manager only logged it and replied with `THREAD_BUSY_MESSAGE` — invisible to the commenter, since `GitHubChannel.send()` is log-only (see below). The comment looked silently ignored.
+
+```mermaid
+sequenceDiagram
+    autonumber
+    participant C1 as Comment 1
+    participant C2 as Comment 2 (while busy)
+    participant Mgr as ChannelManager
+    participant Client as langgraph_sdk client
+    participant SB as StreamBridge
+
+    C1->>Mgr: InboundMessage
+    Mgr->>Client: runs.create() [fire_and_forget]
+    Client-->>Mgr: {run_id: run-1, status: pending}
+    Mgr->>SB: subscribe(run-1) (background watcher)
+
+    C2->>Mgr: InboundMessage (same thread_id)
+    Mgr->>Client: runs.create()
+    Client-->>Mgr: 409 ConflictError
+    Mgr->>Mgr: _buffer_followup(thread_id, msg)<br/>(deduped by delivery_id, capped at 20)
+    Mgr-->>C2: THREAD_BUSY_MESSAGE (log-only on GitHub)
+
+    Note over SB: run-1 completes
+    SB-->>Mgr: END_SENTINEL
+    Mgr->>Mgr: _drain_followups_for_thread()<br/>(pop up to 10, oldest first)
+    Mgr->>Client: runs.create() with <followups-while-busy> input
+    Client-->>Mgr: {run_id: run-2, status: pending}
+    Mgr->>SB: subscribe(run-2) (watch again — chains if >10 queued)
+```
+
+Key properties:
+
+- **Dedupe**: buffering keys on the GitHub webhook delivery id (falling back to the generic provider-message-id metadata keys, mirroring `_inbound_dedupe_key`), so a redelivered webhook for a comment already buffered is a no-op rather than a duplicate entry.
+- **Cap**: 20 entries per thread. Overflow drops the *oldest* buffered entry (not the newest) with a WARNING log — recent activity is a more useful signal than the stalest queued comment once a thread is deep enough in the backlog to hit the cap. No reaction/acknowledgment is sent on drop; see the reactions note below.
+- **Batching**: a drain coalesces at most 10 entries into one `<followups-while-busy>` input block. A backlog deeper than 10 is not force-fit into a single turn — the drained run is itself watched, so its own completion triggers another drain cycle for the remainder.
+- **Drain-conflict edge case**: if the drain's own `runs.create()` also hits `ConflictError` — e.g. a manual Web UI turn or a scheduled run raced onto the same thread — the popped batch is requeued (not lost, not retried in a tight loop). It is picked up again the next time this manager successfully creates *and watches* a run on that thread, which is guaranteed to eventually happen once whatever is occupying the thread finishes and any subsequent trigger succeeds.
+- **Reactions are out of scope for this slice.** GitHub's reaction API (`eyes`/`confused` acknowledgment) has no existing integration in this codebase and needs its own design pass; buffered comments are coalesced silently, with no per-comment acknowledgment.
+- **Config**: gated behind `ChannelRunPolicy.buffer_followups_on_busy` (default `False`); GitHub's own registration in `app/gateway/github/run_policy.py` opts in, since it is exactly the fire_and_forget + log-only-send channel this was designed for. Any other channel that adopts `fire_and_forget=True` in the future keeps the old silent-drop-with-log behavior unless it opts in too.
+- **Plumbing**: the watcher subscribes to the Gateway's `StreamBridge` singleton (`app.state.stream_bridge`), which `ChannelManager` previously had no way to reach — every other consumer gets it via `get_stream_bridge(request)`, which needs an HTTP `Request` the bus-consumer loop doesn't have. It is threaded as a zero-arg closure from `app.py`'s lifespan (`get_stream_bridge=lambda: getattr(app.state, "stream_bridge", None)`) through `start_channel_service()` → `ChannelService.__init__` → `ChannelManager.__init__`, mirroring the existing `launch_run=lambda **kwargs: launch_scheduled_thread_run(app=app, **kwargs)` closure already used for `ScheduledTaskService` in the same lifespan function.
+- **Single-process scope (known limitation)**: the buffer and watcher tasks live in one `ChannelManager` instance's process memory. Under `GATEWAY_WORKERS>1` or a multi-pod deployment, a follow-up comment that a load balancer or webhook fan-out routes to a *different* worker than the one that created the busy run will not see that worker's buffer. This mirrors the cross-pod gap already documented for issue #4120 (IM leader election or a shared buffer store would close it) and is deliberately deferred — single-process/single-pod deployments, the safe default, are unaffected.
+
 ## Outbound is Log-only
 
 ```mermaid
@@ -248,9 +292,10 @@ This is also why the GitHub channel registers `ChannelRunPolicy.fire_and_forget=
 
 ## Cross-references
 
-- [AGENTS.md](../AGENTS.md) → "GitHub event-driven agents" — the index view in `backend/AGENTS.md` (binding shape, per-event triggers, mention precedence, token env summary)
+- [AGENTS.md](../AGENTS.md) → "GitHub event-driven agents" — the index view in `backend/AGENTS.md` (binding shape, per-event triggers, mention precedence, token env summary, follow-up buffering summary)
 - [IM_CHANNEL_CONNECTIONS.md](IM_CHANNEL_CONNECTIONS.md) — interactive IM channels (Telegram/Slack/etc.) for the full `_handle_chat` and owner-scoped file storage flow
 - `app/gateway/github/dispatcher.py` — `fanout_event`, `_is_self_event`, mention precedence chain
+- `app/channels/manager.py` — `_buffer_followup`, `_drain_followups_for_thread`, `_watch_run_and_drain_followups` (follow-up buffering while busy, issue #4121)
 - `app/gateway/github/identity.py` — `resolve_thread_id` (UUID5), `extract_target`
 - `app/gateway/github/triggers.py` — `event_should_fire`, `DEFAULT_TRIGGERS`
 - `app/gateway/github/run_policy.py` — `inject_github_credentials`, `register_policy`

backend/tests/test_channels.py

@@ -1360,12 +1360,21 @@ class TestChannelManager:
         _run(go())
 
     def test_fire_and_forget_thread_busy_releases_dedupe_key(self, tmp_path):
-        """Same invariant on the third swallow site: runs.create's busy branch."""
+        """Same invariant for a fire-and-forget channel that does not buffer follow-ups."""
         import httpx
         from langgraph_sdk.errors import ConflictError
 
-        import app.gateway.github.run_policy  # noqa: F401 — register policy
         from app.channels.manager import THREAD_BUSY_MESSAGE, ChannelManager
+        from app.channels.run_policy import CHANNEL_RUN_POLICY, ChannelRunPolicy
+
+        channel_name = "test-fire-and-forget-retry"
+        original = CHANNEL_RUN_POLICY.get(channel_name)
+        CHANNEL_RUN_POLICY[channel_name] = ChannelRunPolicy(
+            is_interactive=False,
+            fire_and_forget=True,
+            requires_bound_identity=False,
+            buffer_followups_on_busy=False,
+        )
 
         async def go():
             bus = MessageBus()
@@ -1390,7 +1399,7 @@ class TestChannelManager:
 
             def _inbound() -> InboundMessage:
                 return InboundMessage(
-                    channel_name="github",
+                    channel_name=channel_name,
                     chat_id="owner/repo",
                     user_id="dev",
                     owner_user_id="agent-owner-1",
@@ -1410,7 +1419,13 @@ class TestChannelManager:
 
             assert manager._client.runs.create.call_count == 2
 
-        _run(go())
+        try:
+            _run(go())
+        finally:
+            if original is None:
+                CHANNEL_RUN_POLICY.pop(channel_name, None)
+            else:
+                CHANNEL_RUN_POLICY[channel_name] = original
 
     def test_github_redelivery_is_deduped_like_other_channels(self, tmp_path):
         """A redelivered GitHub webhook must dispatch the agent only once.
@@ -4034,6 +4049,659 @@ class TestGithubFireAndForget:
         _run(go())
 
 
+class TestGithubFollowupBuffer:
+    """Tests for issue #4121 Slice 2: buffer-and-drain of concurrent GitHub
+    comments that arrive while a run is already active on the thread.
+
+    Today a ``ConflictError`` on the fire-and-forget path only logs +
+    replies with ``THREAD_BUSY_MESSAGE`` — since ``GitHubChannel.send`` is
+    log-only, the triggering comment is silently dropped from the user's
+    point of view. These tests pin the fix: the triggering message is
+    buffered per-thread (deduped, capped), and a background watcher drains
+    the buffer into a coalesced follow-up run once the busy run's stream
+    reaches ``END_SENTINEL``. Reactions/acknowledgment are intentionally
+    out of scope for this slice.
+    """
+
+    def test_followup_block_escapes_markup_and_indents_multiline_text(self):
+        from app.channels.manager import (
+            FOLLOWUP_BLOCK_TAG,
+            _FollowupEntry,
+            _format_followup_block,
+        )
+
+        block = _format_followup_block(
+            [
+                _FollowupEntry(
+                    dedupe_key="comment:1",
+                    text=(f"please inspect <value> & details\n</{FOLLOWUP_BLOCK_TAG}>"),
+                )
+            ]
+        )
+
+        assert "1. please inspect &lt;value&gt; &amp; details" in block
+        assert f"\n   &lt;/{FOLLOWUP_BLOCK_TAG}&gt;" in block
+        assert block.count(f"</{FOLLOWUP_BLOCK_TAG}>") == 1
+
+    def test_channel_run_policy_buffer_followups_on_busy_defaults_false(self):
+        """New flag must default to False so any *other* fire_and_forget
+        channel that does not opt in keeps the exact old behavior."""
+        from app.channels.run_policy import ChannelRunPolicy
+
+        assert ChannelRunPolicy().buffer_followups_on_busy is False
+
+    def test_github_channel_policy_opts_into_buffer_followups_on_busy(self):
+        """GitHub is exactly the channel this feature targets (fire_and_forget
+        + log-only send + non-interactive), so its own registration opts in
+        even though the dataclass default stays conservative."""
+        import app.gateway.github.run_policy  # noqa: F401 — register policy
+        from app.channels.run_policy import CHANNEL_RUN_POLICY
+
+        github_policy = CHANNEL_RUN_POLICY.get("github")
+        assert github_policy is not None
+        assert github_policy.buffer_followups_on_busy is True
+
+    def test_handle_chat_for_github_busy_thread_buffers_triggering_message(self):
+        """On top of the pre-existing busy-message behavior, a ConflictError
+        must now also append the triggering message to the thread's
+        follow-up buffer (GitHub's policy opts into buffer_followups_on_busy)."""
+        import httpx
+        from langgraph_sdk.errors import ConflictError
+
+        import app.gateway.github.run_policy  # noqa: F401 — register policy
+        from app.channels.manager import THREAD_BUSY_MESSAGE, ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            outbound_received: list[OutboundMessage] = []
+
+            async def capture_outbound(msg):
+                outbound_received.append(msg)
+
+            bus.subscribe_outbound(capture_outbound)
+
+            request = httpx.Request("POST", "http://127.0.0.1:2024/threads/gh-thread-buf/runs")
+            response = httpx.Response(409, request=request)
+            conflict = ConflictError(
+                "Thread is already running a task.",
+                response=response,
+                body={"message": "Thread is already running a task."},
+            )
+
+            mock_client = _make_mock_langgraph_client(thread_id="gh-thread-buf")
+            mock_client.runs.create = AsyncMock(side_effect=conflict)
+            manager._client = mock_client
+
+            await manager.start()
+            try:
+                await manager._handle_chat(
+                    InboundMessage(
+                        channel_name="github",
+                        chat_id="zhfeng/llm-gateway",
+                        user_id="zhfeng",
+                        owner_user_id="agent-owner-1",
+                        text="please also update the README",
+                        metadata={"github": {"delivery_id": "delivery-buf-1"}},
+                    )
+                )
+                await _wait_for(lambda: any(m.text == THREAD_BUSY_MESSAGE for m in outbound_received))
+            finally:
+                await manager.stop()
+
+            assert "gh-thread-buf" in manager._followup_buffers
+            buffered = list(manager._followup_buffers["gh-thread-buf"].values())
+            assert len(buffered) == 1
+            assert buffered[0].text == "please also update the README"
+
+        _run(go())
+
+    def test_conflict_error_does_not_buffer_when_flag_disabled(self):
+        """A fire_and_forget channel that has NOT opted into
+        buffer_followups_on_busy must keep the exact old silent-drop-with-log
+        behavior: busy message still emitted, nothing buffered."""
+        import httpx
+        from langgraph_sdk.errors import ConflictError
+
+        from app.channels.manager import THREAD_BUSY_MESSAGE, ChannelManager
+        from app.channels.run_policy import CHANNEL_RUN_POLICY, ChannelRunPolicy
+
+        original = CHANNEL_RUN_POLICY.get("test-fire-and-forget-no-buffer")
+        CHANNEL_RUN_POLICY["test-fire-and-forget-no-buffer"] = ChannelRunPolicy(
+            is_interactive=False,
+            fire_and_forget=True,
+            requires_bound_identity=False,
+            buffer_followups_on_busy=False,
+        )
+        try:
+
+            async def go():
+                bus = MessageBus()
+                store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+                manager = ChannelManager(bus=bus, store=store)
+
+                outbound_received: list[OutboundMessage] = []
+
+                async def capture_outbound(msg):
+                    outbound_received.append(msg)
+
+                bus.subscribe_outbound(capture_outbound)
+
+                request = httpx.Request("POST", "http://127.0.0.1:2024/threads/no-buf-thread/runs")
+                response = httpx.Response(409, request=request)
+                conflict = ConflictError("busy", response=response, body={"message": "busy"})
+
+                mock_client = _make_mock_langgraph_client(thread_id="no-buf-thread")
+                mock_client.runs.create = AsyncMock(side_effect=conflict)
+                manager._client = mock_client
+
+                await manager.start()
+                try:
+                    await manager._handle_chat(
+                        InboundMessage(
+                            channel_name="test-fire-and-forget-no-buffer",
+                            chat_id="c1",
+                            user_id="u1",
+                            text="hello while busy",
+                        )
+                    )
+                    await _wait_for(lambda: any(m.text == THREAD_BUSY_MESSAGE for m in outbound_received))
+                finally:
+                    await manager.stop()
+
+                assert manager._followup_buffers == {}
+
+            _run(go())
+        finally:
+            if original is None:
+                CHANNEL_RUN_POLICY.pop("test-fire-and-forget-no-buffer", None)
+            else:
+                CHANNEL_RUN_POLICY["test-fire-and-forget-no-buffer"] = original
+
+    def test_duplicate_delivery_id_does_not_double_buffer(self):
+        """A redelivered webhook for the same comment (same delivery_id) must
+        not be buffered twice."""
+        from app.channels.manager import ChannelManager
+
+        manager = ChannelManager(bus=MessageBus(), store=ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json"))
+        thread_id = "gh-thread-dedup"
+        msg = InboundMessage(
+            channel_name="github",
+            chat_id="c",
+            user_id="u",
+            text="please look at this",
+            metadata={"github": {"delivery_id": "dupe-1"}},
+        )
+
+        manager._buffer_followup(thread_id, msg)
+        manager._buffer_followup(thread_id, msg)  # redelivery of the same comment
+
+        assert len(manager._followup_buffers[thread_id]) == 1
+
+    def test_followup_buffer_overflow_drops_oldest_and_warns(self, caplog):
+        """At the per-thread cap, overflow must drop the OLDEST buffered
+        comment (not the newest) and log a warning — recent activity is a
+        more useful signal than the stalest queued item once a thread is
+        deep enough in the backlog to hit the cap."""
+        from app.channels.manager import FOLLOWUP_BUFFER_MAX_PER_THREAD, ChannelManager
+
+        manager = ChannelManager(bus=MessageBus(), store=ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json"))
+        thread_id = "gh-thread-overflow"
+
+        with caplog.at_level(logging.WARNING):
+            for i in range(FOLLOWUP_BUFFER_MAX_PER_THREAD + 5):
+                msg = InboundMessage(
+                    channel_name="github",
+                    chat_id="c",
+                    user_id="u",
+                    text=f"comment {i}",
+                    metadata={"github": {"delivery_id": f"d{i}"}},
+                )
+                manager._buffer_followup(thread_id, msg)
+
+        buffer = manager._followup_buffers[thread_id]
+        assert len(buffer) == FOLLOWUP_BUFFER_MAX_PER_THREAD
+        kept_texts = {entry.text for entry in buffer.values()}
+        for i in range(5):
+            assert f"comment {i}" not in kept_texts
+        assert f"comment {FOLLOWUP_BUFFER_MAX_PER_THREAD + 4}" in kept_texts
+        assert any("overflow" in r.message.lower() for r in caplog.records)
+
+    def test_drain_batches_at_most_ten_entries_per_cycle(self):
+        """A queue deeper than the drain batch size must only coalesce the
+        oldest N entries in one cycle, leaving the rest buffered — this is
+        what lets a >10 backlog chain into a second drain cycle instead of
+        growing one unbounded input block."""
+        from app.channels.manager import FOLLOWUP_DRAIN_BATCH_SIZE, ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            carrier_msg = InboundMessage(
+                channel_name="github",
+                chat_id="zhfeng/llm-gateway",
+                user_id="zhfeng",
+                owner_user_id="agent-owner-1",
+                text="carrier",
+            )
+            thread_id = "gh-thread-batch"
+            for i in range(15):
+                entry_msg = InboundMessage(
+                    channel_name="github",
+                    chat_id="zhfeng/llm-gateway",
+                    user_id="zhfeng",
+                    owner_user_id="agent-owner-1",
+                    text=f"comment {i}",
+                    metadata={"github": {"delivery_id": f"d{i}"}},
+                )
+                manager._buffer_followup(thread_id, entry_msg)
+
+            assert len(manager._followup_buffers[thread_id]) == 15
+
+            mock_client = _make_mock_langgraph_client(thread_id=thread_id)
+            mock_client.runs.create = AsyncMock(return_value={"run_id": "run-drain-1", "status": "pending"})
+
+            await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg)
+
+            mock_client.runs.create.assert_called_once()
+            drained_text = mock_client.runs.create.call_args[1]["input"]["messages"][0]["content"]
+            for i in range(FOLLOWUP_DRAIN_BATCH_SIZE):
+                assert f"comment {i}" in drained_text
+            for i in range(FOLLOWUP_DRAIN_BATCH_SIZE, 15):
+                assert f"comment {i}" not in drained_text
+
+            assert len(manager._followup_buffers[thread_id]) == 15 - FOLLOWUP_DRAIN_BATCH_SIZE
+
+        _run(go())
+
+    def test_drain_conflict_requeues_entries_without_losing_them(self):
+        """If the drain's own runs.create hits ConflictError (a real edge
+        case — e.g. a manual/scheduled trigger raced onto the same thread),
+        the popped batch must be put back rather than lost, and the drain
+        must not raise."""
+        import httpx
+        from langgraph_sdk.errors import ConflictError
+
+        from app.channels.manager import ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            carrier_msg = InboundMessage(
+                channel_name="github",
+                chat_id="zhfeng/llm-gateway",
+                user_id="zhfeng",
+                owner_user_id="agent-owner-1",
+                text="queued comment",
+            )
+            thread_id = "gh-thread-9"
+            manager._buffer_followup(thread_id, carrier_msg)
+            assert len(manager._followup_buffers[thread_id]) == 1
+
+            request = httpx.Request("POST", "http://127.0.0.1:2024/threads/gh-thread-9/runs")
+            response = httpx.Response(409, request=request)
+            conflict = ConflictError("busy", response=response, body={"message": "busy"})
+
+            mock_client = MagicMock()
+            mock_client.runs.create = AsyncMock(side_effect=conflict)
+
+            # Must not raise.
+            await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg)
+
+            assert thread_id in manager._followup_buffers
+            assert len(manager._followup_buffers[thread_id]) == 1
+            mock_client.runs.create.assert_called_once()
+
+        _run(go())
+
+    def test_drain_non_conflict_error_also_requeues_without_crashing(self):
+        """A non-busy exception from the drain's runs.create (network error,
+        5xx, ...) must also be swallowed-and-requeued rather than crashing
+        the watcher task or losing the buffered comments."""
+        from app.channels.manager import ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            carrier_msg = InboundMessage(
+                channel_name="github",
+                chat_id="zhfeng/llm-gateway",
+                user_id="zhfeng",
+                owner_user_id="agent-owner-1",
+                text="queued comment",
+            )
+            thread_id = "gh-thread-neterr"
+            manager._buffer_followup(thread_id, carrier_msg)
+
+            mock_client = MagicMock()
+            mock_client.runs.create = AsyncMock(side_effect=RuntimeError("connection reset"))
+
+            await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg)
+
+            assert len(manager._followup_buffers[thread_id]) == 1
+
+        _run(go())
+
+    def test_drain_resolve_run_params_failure_requeues_entries_without_losing_them(self, monkeypatch):
+        """If a step BETWEEN the buffer pop and runs.create raises -- e.g.
+        _resolve_run_params blows up because the target agent config was
+        removed mid-run -- the already-popped batch must still end up back
+        in the buffer instead of vanishing, and the drain must not raise
+        (mirrors the existing runs.create requeue-on-failure guarantee)."""
+        from app.channels.manager import ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            carrier_msg = InboundMessage(
+                channel_name="github",
+                chat_id="zhfeng/llm-gateway",
+                user_id="zhfeng",
+                owner_user_id="agent-owner-1",
+                text="queued comment",
+            )
+            thread_id = "gh-thread-resolve-fail"
+            manager._buffer_followup(thread_id, carrier_msg)
+            assert len(manager._followup_buffers[thread_id]) == 1
+
+            def _boom(*args, **kwargs):
+                raise RuntimeError("agent config missing mid-run")
+
+            monkeypatch.setattr(manager, "_resolve_run_params", _boom)
+
+            mock_client = MagicMock()
+            mock_client.runs.create = AsyncMock(return_value={"run_id": "should-not-be-created", "status": "pending"})
+
+            # Must not raise -- the failure must be swallowed and requeued.
+            await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg)
+
+            assert thread_id in manager._followup_buffers
+            assert len(manager._followup_buffers[thread_id]) == 1
+            mock_client.runs.create.assert_not_called()
+
+        _run(go())
+
+    def test_drain_apply_channel_policy_failure_requeues_entries_without_losing_them(self, monkeypatch):
+        """Same guarantee one step later: if _apply_channel_policy raises
+        (e.g. channel-policy/credential resolution blows up instead of
+        following its documented degrade-and-continue path), the popped
+        batch must still be requeued rather than lost."""
+        from app.channels.manager import ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            carrier_msg = InboundMessage(
+                channel_name="github",
+                chat_id="zhfeng/llm-gateway",
+                user_id="zhfeng",
+                owner_user_id="agent-owner-1",
+                text="queued comment",
+            )
+            thread_id = "gh-thread-apply-fail"
+            manager._buffer_followup(thread_id, carrier_msg)
+
+            async def _boom(*args, **kwargs):
+                raise RuntimeError("channel policy blew up")
+
+            monkeypatch.setattr(manager, "_apply_channel_policy", _boom)
+
+            mock_client = MagicMock()
+            mock_client.runs.create = AsyncMock(return_value={"run_id": "should-not-be-created", "status": "pending"})
+
+            await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg)
+
+            assert thread_id in manager._followup_buffers
+            assert len(manager._followup_buffers[thread_id]) == 1
+            mock_client.runs.create.assert_not_called()
+
+        _run(go())
+
+    def test_run_watcher_drains_buffer_on_end_sentinel(self):
+        """End-to-end mechanism test: a busy-thread follow-up gets buffered,
+        and once the ORIGINAL run's stream reaches END_SENTINEL, the watcher
+        drains the buffer into a new coalesced runs.create call. That
+        drained run is itself watched too, so an empty buffer at its own
+        END_SENTINEL is a clean no-op (the chain terminates)."""
+        import httpx
+        from langgraph_sdk.errors import ConflictError
+
+        import app.gateway.github.run_policy  # noqa: F401 — register policy
+        from app.channels.manager import FOLLOWUP_BLOCK_TAG, ChannelManager
+        from deerflow.runtime import MemoryStreamBridge
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            bridge = MemoryStreamBridge()
+            manager = ChannelManager(bus=bus, store=store, get_stream_bridge=lambda: bridge)
+
+            request = httpx.Request("POST", "http://127.0.0.1:2024/threads/gh-thread-watch/runs")
+            response = httpx.Response(409, request=request)
+            conflict = ConflictError("busy", response=response, body={"message": "busy"})
+
+            mock_client = _make_mock_langgraph_client(thread_id="gh-thread-watch")
+            mock_client.runs.create = AsyncMock(
+                side_effect=[
+                    {"run_id": "run-1", "status": "pending"},
+                    conflict,
+                    {"run_id": "run-2", "status": "pending"},
+                ]
+            )
+            manager._client = mock_client
+
+            # First message: no active run yet -> succeeds, watcher spawned for run-1.
+            await manager._handle_chat(
+                InboundMessage(
+                    channel_name="github",
+                    chat_id="zhfeng/llm-gateway",
+                    user_id="zhfeng",
+                    owner_user_id="agent-owner-1",
+                    text="first comment",
+                    metadata={"github": {"delivery_id": "d1"}},
+                )
+            )
+            # Second message: thread is busy -> ConflictError -> buffered.
+            await manager._handle_chat(
+                InboundMessage(
+                    channel_name="github",
+                    chat_id="zhfeng/llm-gateway",
+                    user_id="zhfeng",
+                    owner_user_id="agent-owner-1",
+                    text="second comment while busy",
+                    metadata={"github": {"delivery_id": "d2"}},
+                )
+            )
+
+            assert len(manager._followup_buffers["gh-thread-watch"]) == 1
+
+            # The busy run completes -> watcher observes END_SENTINEL -> drains.
+            await bridge.publish_end("run-1")
+            await _wait_for(lambda: mock_client.runs.create.call_count == 3, timeout=2.0)
+
+            drain_call = mock_client.runs.create.call_args_list[2]
+            assert drain_call[0][0] == "gh-thread-watch"
+            coalesced_text = drain_call[1]["input"]["messages"][0]["content"]
+            assert "second comment while busy" in coalesced_text
+            assert f"<{FOLLOWUP_BLOCK_TAG}>" in coalesced_text
+
+            await _wait_for(lambda: "gh-thread-watch" not in manager._followup_buffers)
+
+            # The drained run (run-2) also gets watched. Ending it with an
+            # empty buffer must be a clean no-op — no 4th runs.create call.
+            await bridge.publish_end("run-2")
+            await asyncio.sleep(0.2)
+            assert mock_client.runs.create.call_count == 3
+
+        _run(go())
+
+    def test_stop_cancels_inflight_followup_watcher_task(self):
+        """A follow-up watcher spawned for a run that is still active must be
+        tracked and actually cancelled+awaited by manager.stop() rather than
+        left running as an orphaned task -- otherwise a run that ends AFTER
+        shutdown would still fire a brand new runs.create() into a manager
+        that has already been stopped."""
+        from app.channels.manager import ChannelManager
+        from deerflow.runtime import MemoryStreamBridge
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            bridge = MemoryStreamBridge()
+            manager = ChannelManager(bus=bus, store=store, get_stream_bridge=lambda: bridge)
+
+            carrier_msg = InboundMessage(
+                channel_name="github",
+                chat_id="zhfeng/llm-gateway",
+                user_id="zhfeng",
+                owner_user_id="agent-owner-1",
+                text="carrier",
+            )
+            thread_id = "gh-thread-stop-watcher"
+            # Something buffered, so a slipped-through drain would have a
+            # non-empty batch to (wrongly) fire a run for.
+            manager._buffer_followup(
+                thread_id,
+                InboundMessage(
+                    channel_name="github",
+                    chat_id="c",
+                    user_id="u",
+                    text="queued while busy",
+                    metadata={"github": {"delivery_id": "d-stop-1"}},
+                ),
+            )
+
+            mock_client = _make_mock_langgraph_client(thread_id=thread_id)
+            mock_client.runs.create = AsyncMock(return_value={"run_id": "run-should-not-fire", "status": "pending"})
+            manager._client = mock_client
+
+            await manager.start()
+
+            # Spawn a watcher for a run whose stream never ends -- it sits
+            # suspended awaiting stream_bridge.subscribe(), exactly like a
+            # real in-flight watcher for a long-running GitHub coding run.
+            manager._maybe_spawn_followup_watcher(thread_id, {"run_id": "run-being-watched"}, carrier_msg)
+            await asyncio.sleep(0.05)
+
+            assert len(manager._followup_watcher_tasks) == 1
+            watcher_task = next(iter(manager._followup_watcher_tasks))
+            assert not watcher_task.done()
+
+            await manager.stop()
+
+            assert watcher_task.done()
+            assert watcher_task.cancelled()
+            assert watcher_task not in manager._followup_watcher_tasks
+
+            # A late "run completed" signal for the (cancelled) watched run
+            # must not resurrect a drain -- nothing is subscribed anymore.
+            await bridge.publish_end("run-being-watched")
+            await asyncio.sleep(0.1)
+            mock_client.runs.create.assert_not_called()
+
+        _run(go())
+
+    def test_drain_after_stop_does_not_create_run(self):
+        """Belt-and-suspenders guard: even if a drain call reaches
+        _drain_followups_for_thread after the manager has been stopped (e.g.
+        a watcher that had already slipped past its own cancellation point
+        mid-drain), it must not fire client.runs.create against the stopped
+        manager, and the buffered entries must remain untouched (not popped,
+        not lost)."""
+        from app.channels.manager import ChannelManager
+
+        async def go():
+            bus = MessageBus()
+            store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
+            manager = ChannelManager(bus=bus, store=store)
+
+            carrier_msg = InboundMessage(
+                channel_name="github",
+                chat_id="zhfeng/llm-gateway",
+                user_id="zhfeng",
+                owner_user_id="agent-owner-1",
+                text="carrier",
+            )
+            thread_id = "gh-thread-post-stop-drain"
+            manager._buffer_followup(
+                thread_id,
+                InboundMessage(
+                    channel_name="github",
+                    chat_id="c",
+                    user_id="u",
+                    text="queued",
+                    metadata={"github": {"delivery_id": "d-post-stop-1"}},
+                ),
+            )
+
+            await manager.start()
+            await manager.stop()
+            assert manager._running is False
+
+            mock_client = MagicMock()
+            mock_client.runs.create = AsyncMock(return_value={"run_id": "should-not-fire", "status": "pending"})
+
+            await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg)
+
+            mock_client.runs.create.assert_not_called()
+            assert len(manager._followup_buffers[thread_id]) == 1
+
+        _run(go())
+
+    def test_channel_manager_get_stream_bridge_threaded_from_service(self):
+        """The app.py -> service.py -> manager.py plumbing: ChannelService
+        must forward get_stream_bridge through to its ChannelManager."""
+        from app.channels.service import ChannelService
+
+        sentinel = object()
+        service = ChannelService(channels_config={}, get_stream_bridge=lambda: sentinel)
+
+        assert service.manager._get_stream_bridge() is sentinel
+
+    def test_start_channel_service_forwards_get_stream_bridge(self):
+        """The module-level singleton entrypoint must also thread the
+        callable through to ChannelService.from_app_config."""
+        import app.channels.service as service_module
+
+        captured: dict[str, object] = {}
+
+        class _FakeService:
+            async def start(self):
+                return None
+
+            def get_status(self):
+                return {}
+
+        def fake_from_app_config(app_config=None, *, get_stream_bridge=None):
+            captured["get_stream_bridge"] = get_stream_bridge
+            return _FakeService()
+
+        async def go():
+            service_module._channel_service = None
+            with patch.object(service_module.ChannelService, "from_app_config", staticmethod(fake_from_app_config)):
+                sentinel = object()
+                await service_module.start_channel_service(get_stream_bridge=lambda: sentinel)
+
+            assert captured["get_stream_bridge"]() is sentinel
+
+        try:
+            _run(go())
+        finally:
+            service_module._channel_service = None
+
+
 class _BoundIdentityRepo:
     def __init__(self, connections: list[dict[str, str | None]] | None = None) -> None:
         self.connections = list(connections or [])

backend/tests/test_gateway_lifespan_shutdown.py

@@ -42,7 +42,7 @@ async def _run_lifespan_with_hanging_stop() -> float:
     fake_service = MagicMock()
     fake_service.get_status = MagicMock(return_value={})
 
-    async def fake_start(_startup_config):
+    async def fake_start(_startup_config, **_kwargs):
         return fake_service
 
     close_oidc_service = AsyncMock()
@@ -90,7 +90,7 @@ async def _run_lifespan_with_upload_staging_cleanup():
     close_oidc_service = AsyncMock()
     stop_channel_service = AsyncMock()
 
-    async def fake_start(_startup_config):
+    async def fake_start(_startup_config, **_kwargs):
         return fake_service
 
     with (
@@ -142,7 +142,7 @@ async def _run_lifespan_with_memory_flush(*, enabled: bool, flush_return: bool)
     close_oidc_service = AsyncMock()
     stop_channel_service = AsyncMock()
 
-    async def fake_start(_startup_config):
+    async def fake_start(_startup_config, **_kwargs):
         return fake_service
 
     manager = MagicMock()
@@ -217,7 +217,7 @@ async def _run_lifespan_with_warm_return(warm_return: bool | None) -> MagicMock:
     close_oidc_service = AsyncMock()
     stop_channel_service = AsyncMock()
 
-    async def fake_start(_startup_config):
+    async def fake_start(_startup_config, **_kwargs):
         return fake_service
 
     manager = MagicMock()

backend/tests/test_monocle_tracing.py

@@ -330,7 +330,7 @@ def test_gateway_lifespan_initializes_monocle():
     fake_service = MagicMock()
     fake_service.get_status = MagicMock(return_value={})
 
-    async def fake_start(_startup_config):
+    async def fake_start(_startup_config, **_kwargs):
         return fake_service
 
     setup_spy = MagicMock(return_value=False)
@@ -372,7 +372,7 @@ def test_gateway_lifespan_survives_monocle_setup_failure(caplog):
     fake_service = MagicMock()
     fake_service.get_status = MagicMock(return_value={})
 
-    async def fake_start(_startup_config):
+    async def fake_start(_startup_config, **_kwargs):
         return fake_service
 
     setup_spy = MagicMock(side_effect=ValueError("MONOCLE_EXPORTERS has unknown exporter(s): fle."))