diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 8179fe9ab..87b297497 100644 --- a/backend/AGENTS.md +++ b/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 ``-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. diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index d583c516b..a50c99c68 100644 --- a/backend/app/channels/manager.py +++ b/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"") + 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 "")) diff --git a/backend/app/channels/run_policy.py b/backend/app/channels/run_policy.py index 3e91a2290..963954dcd 100644 --- a/backend/app/channels/run_policy.py +++ b/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 diff --git a/backend/app/channels/service.py b/backend/app/channels/service.py index 0f861ef72..20740eddd 100644 --- a/backend/app/channels/service.py +++ b/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 diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index 90b3732ae..8fde90f27 100644 --- a/backend/app/gateway/app.py +++ b/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") diff --git a/backend/app/gateway/github/run_policy.py b/backend/app/gateway/github/run_policy.py index 2de72de1b..cf45cbd2d 100644 --- a/backend/app/gateway/github/run_policy.py +++ b/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, ) diff --git a/backend/docs/GITHUB_AGENTS.md b/backend/docs/GITHUB_AGENTS.md index 801d74c90..d46004cec 100644 --- a/backend/docs/GITHUB_AGENTS.md +++ b/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)
(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()
(pop up to 10, oldest first) + Mgr->>Client: runs.create() with 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 `` 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` diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 6c8f8ca54..b3352a370 100644 --- a/backend/tests/test_channels.py +++ b/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 & details\n"), + ) + ] + ) + + assert "1. please inspect <value> & details" in block + assert f"\n </{FOLLOWUP_BLOCK_TAG}>" in block + assert block.count(f"") == 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 []) diff --git a/backend/tests/test_gateway_lifespan_shutdown.py b/backend/tests/test_gateway_lifespan_shutdown.py index 28ba46df1..5a283baf4 100644 --- a/backend/tests/test_gateway_lifespan_shutdown.py +++ b/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() diff --git a/backend/tests/test_monocle_tracing.py b/backend/tests/test_monocle_tracing.py index 075897ab8..b6a10a312 100644 --- a/backend/tests/test_monocle_tracing.py +++ b/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."))