diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a4a1a32ff..c89dfadde6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1247,6 +1247,26 @@ This release closes that milestone with **621 merged pull requests**. ### Fixed +- **channels:** Runs started from an IM channel no longer die with the Gateway + instance that received the message. The manager holds `runs.wait` / + `runs.stream` open for the whole run, and the Gateway cancels a run whose + creating request disconnects unless that request asked for + `on_disconnect: continue`; a leader hand-off, a rolling restart or a crash + of the receiving instance therefore cancelled every IM run in flight. + Channel runs now ask for `continue`, so the run finishes on the Gateway + executing it and its thread state is persisted — the answer stays readable + in the web UI even though the IM reply is lost with the old transport. The + same policy covers the SDK's 300s read timeout on a Slack/Discord + `runs.wait`: the chat still gets the error reply, but the run finishes in + the background instead of being cancelled. Because the run survives, the + manager now keeps the inbound dedupe key once the run request has left it + (a cancelled worker, a read timeout, a dropped connection): the provider's + redelivery on the new holder is dropped instead of executing the turn a + second time, while a request the Gateway rejected or a failure before + sending still releases the key for a retry. Fire-and-forget channels + (GitHub) are unchanged: `runs.create` returns as soon as the run is pending. + ([#6876]) + - **deploy:** The Gateway container now shuts down gracefully. The production compose `command`, both `CMD`s in `backend/Dockerfile` and the Helm chart started it as `sh -c "cd backend && … uv run … uvicorn …"`, which leaves `sh` @@ -11711,3 +11731,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6844]: https://github.com/bytedance/deer-flow/pull/6844 [#6852]: https://github.com/bytedance/deer-flow/pull/6852 [#6865]: https://github.com/bytedance/deer-flow/pull/6865 +[#6876]: https://github.com/bytedance/deer-flow/pull/6876 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 93fc0b591c..5ea8e271fa 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -1002,6 +1002,18 @@ ### 修复 +- **渠道：** 从 IM 渠道发起的运行不再随接收消息的 Gateway 实例一起消亡。manager 在整个 + 运行期间保持 `runs.wait` / `runs.stream` 请求打开，而 Gateway 会在创建请求断开时取消 + 运行，除非该请求声明了 `on_disconnect: continue`；因此 leader 切换、滚动重启或接收 + 实例崩溃都会取消所有进行中的 IM 运行。渠道运行现在声明 `continue`，运行会在执行它的 + Gateway 上完成并持久化对话线程状态——即使 IM 回复随旧传输丢失，答案仍可在 Web UI + 中查看。同一策略也覆盖 Slack/Discord `runs.wait` 遇到 SDK 300 秒读超时的情况：聊天里 + 仍会收到错误回复，但运行会在后台完成而不再被取消。由于运行会继续，manager 现在在运行 + 请求发出之后保留入站去重键（worker 被取消、读超时、连接断开）：provider 在新持有者上的 + 重投递会被丢弃，而不是把这一轮再执行一遍；Gateway 拒绝的请求或发送前的失败仍会释放该键 + 以便重试。即发即弃渠道（GitHub）不受影响：`runs.create` 在运行进入 pending 后即返回。 + ([#6876]) + - **前端：** 用户管理设置页不再在静态网站演示中发出注定失败的查询，非管理员的 `?settings=users` 深链接也不再打开仅管理员可见的面板。该页面此前只以 `system_role === "admin"` 判断，而演示用户名义上是管理员，静态响应层又没有 @@ -9345,3 +9357,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6844]: https://github.com/bytedance/deer-flow/pull/6844 [#6852]: https://github.com/bytedance/deer-flow/pull/6852 [#6865]: https://github.com/bytedance/deer-flow/pull/6865 +[#6876]: https://github.com/bytedance/deer-flow/pull/6876 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 76f6a2f617..f9280f0a89 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -14,13 +14,13 @@ Finalize asyncgens; keep primary errors; propagate lone Bot errors. **Components**: - `message_bus.py` - Inbound queue → dispatcher; outbound uses a dispatch-start listener snapshot. Unsubscribe neither cancels nor drains captured callbacks. - `store.py` - `ChannelStore` protocol (async `get_thread_id`/`set_thread_id`/`remove`/`list_entries`) for `channel_name:chat_id[:topic_id]` → `thread_id`. `SqlChannelStore` (`channel_thread_bindings`, harness `persistence/channel_thread_bindings/`; upsert keeps `created_at`, topic-less `remove` deletes by key prefix) for sqlite/postgres, so replicas share bindings; `JsonChannelStore` (`{base_dir}/channels/store.json`, loaded once per process, file I/O via `to_thread` under `_lock`) only for `database.backend: memory`. `resolve_channel_store` picks; `ChannelService.start()` runs `SqlChannelStore.import_legacy_json` once (file + empty table → `INSERT … ON CONFLICT DO NOTHING`, rename to `store.json.migrated`, a peer's rename tolerated; populated table → file kept). Feishu's sync lark callback (on the SDK's own loop) bridges `_resolve_routing_bounded` to the Gateway loop via `_run_store_call` (tracked future; refuses only a wait from the Gateway loop itself). Store calls run under `FEISHU_STORE_LOOKUP_TIMEOUT_SECONDS` (3s; the SDK loop also services ws keepalives); timeout or store error = *unresolved* (`StoreLookupUnresolved`), never a miss: the message joins a per-chat FIFO drained on the Gateway loop with backoff up to `FEISHU_STORE_RESOLVE_TOTAL_BUDGET_SECONDS` (30s), then is dropped with a warning, never routed to a new conversation; each deferred message holds a bus intake reservation (the `inbound_queue_maxsize` bound) and its 30s budget runs from arrival. `discord.py` reads `channels_dir`. Tests: `test_channel_thread_binding_store.py` (also under `blocking_io/`). -- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout` +- `manager.py` - Core dispatcher: creates threads via `client.threads.create()`, routes commands including `/goal` (setting a goal persists it through Gateway and then routes the objective as a chat turn) and `/agent` (`list` is owner-scoped; `use` creates a fresh thread and persists the Custom Agent selection under both the channel restart key and Web's canonical routing metadata so existing checkpoint lineage never changes agent across IM/Web), keeps Slack/Discord on `client.runs.wait()`, uses `client.runs.stream(["messages-tuple", "values"])` for Feishu/Telegram incremental outbound updates, serializes same-thread Feishu turns in-manager when the channel's `ChannelRunPolicy.serialize_thread_runs=True` so rapid follow-ups queue instead of tripping the runtime busy reply, and switches to `client.runs.create()` (fire-and-forget, returns once the run is `pending`) for channels whose `ChannelRunPolicy.fire_and_forget=True` so long autonomous runs do not hit the SDK default 300s `httpx.ReadTimeout`; `runs.wait`/`runs.stream` pass `on_disconnect: continue`, so a run outlives the instance that requested it (reply lost, thread persisted); hence once a run request is sent, cancellation or a post-request transport error keeps the inbound dedupe key (`_keep_dedupe_key_after`) and only Gateway rejections, pre-request failures and `error` frames release it Streaming `error` frames use the same failure path as transport exceptions. Publish the final outbound before releasing the inbound dedupe key, so provider redelivery can retry without overtaking the terminal reply. Log the frame's error name and message at WARNING with the thread ID; keep the synthetic error generic so payload details do not change reply selection. - **What may be published from the stream is an allowlist, not a denylist** (`_accumulate_stream_text`): only assistant message types — LangChain serializes `AIMessage.type` as `"ai"` and `AIMessageChunk.type` as `"AIMessageChunk"`, plus the OpenAI-style `"assistant"` spelling for foreign runtimes — become displayable text. The previous rule rejected only payloads whose `type` contained `"tool"` and therefore published everything else, which leaked DeerFlow's hidden model context to every streaming IM channel: `DynamicContextMiddleware` injects recalled memory as a hidden `HumanMessage` (`type == "human"`) and rewrites the user's own turn into a new `HumanMessage`, `DurableContextMiddleware` injects a hidden `` `HumanMessage`, and LangGraph fans those state writes out on the `messages-tuple` stream. Proved live on a Buzz relay (a `` fact block; a verbatim echo of the user's message as the reply). Matching is by prefix (`ai` / `assistant`), never substring, because ordinary words contain `"ai"` (`chain`, `domain`). The message type is resolved by `_stream_payload_type`, which handles both the `model_dump()` shape DeerFlow's own gateway emits and LangChain's `to_json()` constructor shape (whose top-level `type` is the literal `"constructor"`, with the class name at the tail of the `id` path). A bare `str` payload is no longer accepted at all: it carries no type information, so it cannot be attributed to the assistant, and nothing in DeerFlow produces one (`runtime/serialization.py::serialize_messages_tuple` always emits `[message_dict, metadata]`). + **What may be published from the stream is an allowlist, not a denylist** (`_accumulate_stream_text`): only assistant message types — LangChain serializes `AIMessage.type` as `"ai"` and `AIMessageChunk.type` as `"AIMessageChunk"`, plus the OpenAI-style `"assistant"` spelling for foreign runtimes — become displayable text. The previous rule rejected only payloads whose `type` contained `"tool"` and therefore published everything else, which leaked DeerFlow's hidden model context to every streaming IM channel: `DynamicContextMiddleware` injects recalled memory as a hidden `HumanMessage` (`type == "human"`) and rewrites the user's own turn into a new `HumanMessage`, `DurableContextMiddleware` injects a hidden `` `HumanMessage`, and LangGraph fans those state writes out on the `messages-tuple` stream. Proved live on a Buzz relay. Matching is by prefix (`ai` / `assistant`), never substring, because ordinary words contain `"ai"` (`chain`, `domain`). The message type is resolved by `_stream_payload_type`, which handles both the `model_dump()` shape DeerFlow's own gateway emits and LangChain's `to_json()` constructor shape (whose top-level `type` is the literal `"constructor"`, with the class name at the tail of the `id` path). A bare `str` payload is no longer accepted at all: it carries no type information, so it cannot be attributed to the assistant, and nothing in DeerFlow produces one (`runtime/serialization.py::serialize_messages_tuple` always emits `[message_dict, metadata]`). - `base.py` - Abstract `Channel` base class (start/stop/send lifecycle). Provider callbacks that submit coroutines from SDK threads must use `_submit_threadsafe_coroutine()` (or `_submit_threadsafe_coroutine_future()` to await the outcome through `asyncio.shield`): it creates and retains the real `asyncio.Task` on the owner loop instead of treating `run_coroutine_threadsafe()`'s proxy Future as a completion signal. Submission is closed atomically with shutdown, and `stop()` must call `_close_and_drain_threadsafe_futures()` before tearing down SDK resources. Blocking SDK teardown in `stop()` (a client `close()`, a thread `join()`) must run off the event loop through `asyncio.to_thread`: the shared Gateway loop also serves `POST /api/channels/{name}/restart`; Slack cleanup awaits connect then closes off-loop; a 5s deadline/cancellation retains it for retry. SDK close errors propagate. Anchored by `tests/blocking_io/test_channel_stop_offloop.py`. **Proactive push** (`send_notification`): WeCom, Feishu, WeChat and QQ C2C use the shared scheduled outbox; `capabilities.py` gates enqueue/UI. Unsupported rows fail terminally. Await ACK; `ChannelUnavailable` parks, business rejection counts. WeCom socket `RuntimeError` parks except `Reply ack error`. Feishu resolves owner p2p; explicit groups use connection IDs and a separate hook. QQ verifies owner/current-App C2C, never group member OpenIDs or passive reply context. WeChat uses encrypted owner/bot/private context, restored after restart; stale public metadata is not trusted. `store_connection_context` locks/checks the active identity before merging encrypted extra. The browser connection API strips legacy context tokens. Shared IM settings (0041) gate enqueue/send; disabling atomically closes pending/sending rows, and reenabling does not replay them. In-flight provider sends may finish. Binding lookup failures park; revoked targets fail before liveness. Terminal rows fence late writes. `_on_finalization` stages one notice per target with the outcome (stop > pause > finish > run outcome); recovery dedupes, idle ends use `idle:`, trials/interrupts stay silent. Localized text has no IDs/links. Lifespan starts delivery after channels, detaches enqueue if it cannot, and stops delivery first. Setup: [IM guide](../../docs/IM_CHANNEL_CONNECTIONS.md#proactive-notification-settings). - `wecom.py` - WeCom WebSocket channel with proactive `send_notification`; `_ws_transport_up` maps disconnects to `ChannelUnavailable` (no retry used). `capabilities.py` - SDK-free `CHANNEL_CAPABILITIES`. - `service.py` - Drain manager work before transports; retain timed-out instances for retry. Feishu/DingTalk `stop()` stop the SDK thread's private loop via `_request_*_shutdown` (Feishu disarms lark's `_auto_reconnect` first, else its reconnect branch eats the stop; DingTalk runs its own copy of the SDK connect loop), then join; a live join or Slack cleanup raises `ChannelStopTimeout`: retain, defer replacement/restart, log WARNING without traceback; wedged instances report not running. Reload returns snapshots; restart owns cached fallback/publication under the channel lock. `_stopping` fences starts/publication (including after pre-start I/O) until shutdown succeeds. Lock manager startup through stale cleanup; capture the shutdown generation before locking to fence queued starts. `stop()` stays unlocked to fence in-flight startup. Leader lease (`leadership.py`): only the `channel_leases` holder starts a non-GitHub transport (`_start_channel` checks `holds()`; callers acquire first); `_supervise` renews at a fixed rate, runs starts/stops per provider, stops a transport before its lease lapses. -- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, edits the "Working on it..." stream target in place via `editMessageText`, and can optionally send final Markdown replies as Rich Messages through `channels.telegram.rich_messages` — the Rich path only fires when `rich_messages` is on **and** the text contains a rich construct (fenced code, emphasis, a table, a task list, `
`, block math, or a link), so structured command/error replies stay plain text and their newlines and `` tokens survive; `discord.py` runs `_on_message` on discord.py's private loop, so connection-repository work (identity lookup with the reservation commit, `/connect` binding) goes to the Gateway loop through `_submit_threadsafe_coroutine` — the SQLAlchemy engine is bound there — and bind replies return through `_run_on_discord_loop`; typing registers before that hand-off settles, and a failed hand-off skips the ack and stops the indicator it started unless another message still relies on it; `discord.py` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; each tick sends the indicator with discord.py 2.x's `await channel.typing()` (first failed tick logs at WARNING, later ones at DEBUG); because `stop()` runs on the main loop while typing tasks belong to `_discord_loop`, normal cross-thread cancellation, awaiting, and map cleanup are scheduled there with a bounded wait, while `_run_client()` drains tasks in its `finally` block before an exception/disconnect can make that loop unusable; an already-stopped foreign loop must never have its tasks awaited from the main loop; to that end every outbound cross-loop call (`send` / `send_file` / `_get_channel_or_thread`) goes through `_run_on_discord_loop`, which bounds the await (`DISCORD_OUTBOUND_TIMEOUT_SECONDS` 30s; `DISCORD_UPLOAD_TIMEOUT_SECONDS` 120s for file uploads: slow uplink plus 429 retry-after) and fails fast with a `RuntimeError` when the client loop is missing or not running — a dead client becomes a logged send failure instead of a permanently hung `ChannelManager` worker — and `is_running` reports client-thread aliveness (like `feishu.py`) so `ensure_channel_ready` can restart the channel after `_run_client()` exits on a fatal error; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`) +- `slack.py` / `feishu.py` / `telegram.py` / `discord.py` / `dingtalk.py` - Platform-specific implementations (`feishu.py` tracks the running card `message_id` in memory and patches the same card in place; `telegram.py` accepts inbound text/photos/documents, preserves media captions, hands token-free attachment bytes to the shared upload pipeline, edits the "Working on it..." stream target in place via `editMessageText`, and can optionally send final Markdown replies as Rich Messages through `channels.telegram.rich_messages` — the Rich path only fires when `rich_messages` is on **and** the text contains a rich construct (fenced code, emphasis, a table, a task list, `
`, block math, or a link), so structured command/error replies stay plain text and their newlines and `` tokens survive; `discord.py` runs `_on_message` on discord.py's private loop, so connection-repository work (identity lookup with the reservation commit, `/connect` binding) goes to the Gateway loop through `_submit_threadsafe_coroutine` — the SQLAlchemy engine is bound there — and bind replies return through `_run_on_discord_loop`; typing registers before that hand-off settles, and a failed hand-off skips the ack and stops the indicator it started unless another message still relies on it; `discord.py` registers typing loops before inbound handling yields and `_start_typing()` refuses work once `_running` is false; each tick sends the indicator with discord.py 2.x's `await channel.typing()`; because `stop()` runs on the main loop while typing tasks belong to `_discord_loop`, normal cross-thread cancellation, awaiting, and map cleanup are scheduled there with a bounded wait, while `_run_client()` drains tasks in its `finally` block before an exception/disconnect can make that loop unusable; an already-stopped foreign loop must never have its tasks awaited from the main loop; to that end every outbound cross-loop call (`send` / `send_file` / `_get_channel_or_thread`) goes through `_run_on_discord_loop`, which bounds the await (`DISCORD_OUTBOUND_TIMEOUT_SECONDS` 30s; `DISCORD_UPLOAD_TIMEOUT_SECONDS` 120s for file uploads: slow uplink plus 429 retry-after) and fails fast with a `RuntimeError` when the client loop is missing or not running — a dead client becomes a logged send failure instead of a permanently hung `ChannelManager` worker — and `is_running` reports client-thread aliveness (like `feishu.py`) so `ensure_channel_ready` can restart the channel after `_run_client()` exits on a fatal error; `dingtalk.py` optionally uses AI Card streaming for in-place updates when `card_template_id` is configured, and overrides `receive_file` to download inbound images (`picture`/`richText`) and documents (`file`) by `downloadCode` into the thread uploads bucket, mirroring `feishu.py`) - `buzz.py` - Buzz (Nostr relay) implementation: one NIP-42-authenticated websocket, pubkey-allowlist + mention/DM/thread-follow gating (thread-follow reads via the manager's `connection_identity.lookup_thread_id`, so bound authors' repo-only threads count), streaming replies via in-place kind-40003 edits; requires the `buzz` dependency extra. Its durable seen-event replay guard uses async `aseen()` / `arecord()` / `aflush()` boundaries: initial JSON reads and coalesced atomic writes run off the Gateway event loop, and `stop()` awaits the final flush before returning. **Subscription model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model)): the relay fans kind-9 chat events out **only** to channel-scoped subscriptions, proved on a live relay: `REQ {"kinds":[9]}` is accepted and answered with `EOSE` but never receives an event, `REQ {"kinds":[9],"#h":[uuid]}` works, and a multi-value `#h` matches nothing, so it is strictly one REQ per channel. Every connection therefore rebuilds three kinds of subscription after NIP-42 auth: `buzz-discovery` (`{"kinds":[39000]}`, a historical query returning exactly the channels this identity belongs to, one stored event each then `EOSE` — adding `#p` returns zero, do not "narrow" it), `buzz-membership` (`{"kinds":[44100,44101],"#p":[us],"since":}`, the relay-signed member-added/member-removed notifications whose `p` tag names the affected member and `h` tag the channel), and one `buzz-chat-` per discovered channel. Chat subscriptions open as each kind-39000 arrives; the discovery `EOSE` is the completeness barrier that retries any that failed and warns when discovery found nothing. A kind-44100 for our pubkey subscribes to the new channel **live** (then re-issues discovery so its name/type reach the DM-detection cache, but only when that channel's metadata is actually missing — an unconditional refresh is how a burst of 44100s multiplied into one discovery pass each); a kind-44101 issues `buzz_nostr.close_frame` for exactly that channel's subscription and drops its metadata. **The membership subscription is scoped to LIVE events** and this is load-bearing: buzz-relay *stores* 44100/44101 and serves history newest-first (default limit 2000), so an unscoped filter replayed the whole membership history on every connect — every stored add read as live, re-running discovery once each (M+1 discovery passes × N stored kind-39000 events per connect, observed live as two `channel discovery complete` lines and one channel logged `` because the 44100 path subscribed before its metadata arrived), re-subscribing channels we have since been removed from, and letting a stored 44101 transiently unsubscribe a channel we are still in. `since` is anchored at the moment the socket opened (`_session_started_at`) minus `MEMBERSHIP_LOOKBACK_SECONDS` (60) of slack, which covers both relay clock skew and a membership change published *during* the connect/auth handshake; the slack can only cost an idempotent replay of the last minute. **A relay `CLOSED` frame is recovered** — every subscription on the socket fails silently when dropped, so `_handle_closed` re-issues it, bounded by `MAX_RESUBSCRIBE_ATTEMPTS` (3) per subscription id per connection *and per auth epoch* (`_resubscribe_attempts` is reset on session start, on re-auth, and by `stop()`, so pre-auth rejections — which the auth branch already recovers wholesale — never spend the authenticated session's budget). The first retry is immediate (a one-off hiccup is the common case); later ones back off 1s then 2s, awaited inline in the read loop, not in a background task that could outlive its socket. **`auth-required:` before this socket has completed its NIP-42 handshake is the expected bootstrap sequence, not a refusal**, and `_handle_closed` short-circuits it ahead of every recovery path: the connector opens its control REQs immediately in case the relay serves unauthenticated reads, a closed relay answers `auth-required:` plus an `AUTH` challenge, and the auth branch re-opens everything. That case is logged at DEBUG and consumes neither the permanent-refusal branch nor the retry budget (a chat subscription is still dropped from `_chat_subscriptions`, since it genuinely is not subscribed and discovery is what re-opens it). Treating it as permanent produced a false operator-facing warning that discovery/membership tracking was DOWN while discovery then completed normally. The boundary is the per-socket `_auth_completed` flag, set once the signed AUTH event has been sent and cleared on session entry, session exit, and `stop()`; the same reason *after* that stays loud, and says the subscription is down until the relay's next AUTH challenge or the next reconnect rather than borrowing the non-auth wording. Then `_is_transient_close` decides whether to retry at all: NIP-01/NIP-42 `auth-required:`/`restricted:`/`blocked:`/`mute:`/`invalid:`/`pow:` prefixes and buzz-relay's own removal/revocation prose are permanent (do not fight the relay over a channel that is no longer ours), `rate-limited:`/`error:`/no-reason-at-all and anything unrecognized are transient — the default resolves toward "keep listening" because going silently deaf is the failure this exists to remove, and the attempt budget bounds a wrong guess. A chat `CLOSED` is only ever recovered for a channel already in `_chat_subscriptions`: a `CLOSED` is relay-supplied, so acting on an unknown one would let a relay induce a subscription just by naming a channel. Every subscription that goes unlistened is logged at WARNING, never INFO. Subscription ids are deterministic per channel so one can be replaced or closed without disturbing the others, and `_chat_subscriptions` is per-socket state cleared on session end, on re-auth (a pre-auth REQ may have been rejected), and by `stop()`. Three bounds on remote-fed state: `MAX_CACHED_CHANNELS` (512) caps the kind-39000 metadata cache, the watermark map, and the resubscribe-attempt map; `MAX_CHANNEL_SUBSCRIPTIONS` (256, well under buzz-relay's own 1024-per-connection ceiling) caps live chat subscriptions — at the cap new channels are refused and named in a warning rather than evicting a working subscription. **Backlog paging:** freeze the read floor and track `_backlog_until`; stop/warn on stalled timestamps, never skip a tied second. Removal/live close clears paging and closes backlog; ignore stale frames. Reset cursors per socket. Bounds: [subscription model](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-subscription-model). **Trust model** (operator-facing version in [IM_CHANNEL_CONNECTIONS.md](../../docs/IM_CHANNEL_CONNECTIONS.md#buzz-trust-model)): every inbound `EVENT` is authenticated at the single `handle_relay_frame` choke point — the NIP-01 id is recomputed from the delivered payload and the BIP-340 Schnorr signature verified against the claimed `pubkey` (`buzz_nostr.verify_event`, pure and total: malformed input returns `False`, never raises) — so `ev["pubkey"]`, the authorization principal for both the allowlist and the `/connect` bind, cannot be forged by a relay the DeerFlow operator does not run. What remains trusted is the *authorship* of kind-39000 channel metadata: any member can sign one, and because per-channel subscriptions are now driven by discovery, a forged kind-39000 has two effects rather than one — it can mark a channel `type: "dm"` (relaxing `require_mention` for that channel) **and** it can induce a chat subscription for a channel of the forger's choosing, since the channels we listen to are exactly the channels we hold metadata for. Neither makes anything be *acted on*: `allowed_users` and per-event signature verification are independent gates, so an induced subscription only means the relay reads its own traffic back to a subscriber that drops it, bounded by `MAX_CHANNEL_SUBSCRIPTIONS` (which refuses rather than evicts, so it cannot displace a real channel). Same for a forged kind-44100, except its `p` tag is re-checked locally so it must at least name us. Closing this needs a configured trusted relay pubkey, which `relay_url` is not. `allowed_users` is deny-by-default (empty = nobody, unlike siblings' empty = everyone), so `start()` logs a WARNING when it is empty and each drop logs at DEBUG. The resubscribe cursor (`since`) is **per channel**, advances only for events that were actually processed, and never past `now + MAX_FUTURE_SKEW_SECONDS`, because it is peer-supplied (`created_at`) and a single future-dated event otherwise made the connector permanently deaf. Per channel rather than global is the safety-critical half: one shared cursor is the newest event seen in *any* channel, so a busy channel would drag it past a quiet channel's unread messages, skipping them on the next reconnect. Per-channel cursors can only ever cost duplicate delivery (absorbed by the manager's `event_id` dedupe), and an evicted cursor degrades to "no `since`", i.e. the relay's default backlog — both fail toward replay, never toward a miss. Streaming tracks every oversize chunk index (`_stream_targets` for chunk 0, `_stream_tails` for the rest), since the manager republishes cumulative text and reposting `chunks[1:]` per update flooded the channel; all of it is per-connection state cleared by `stop()`, and the remote-fed kind-39000 cache is capped at `MAX_CACHED_CHANNELS`. **`send()` refuses outright to publish text carrying a hidden model-context wrapper** (``, ``, `` — `_HIDDEN_CONTEXT_MARKERS`), logging at ERROR and clearing the stream bookkeeping on a blocked `is_final`. This is defense in depth behind the manager's allowlist, and it lives here rather than in a sibling connector because on Buzz a leak is permanent: every streaming update is an immutable public Nostr event, so a corrective edit only changes what clients render while the original leaked event stays on the relay. Matching is on the literal opening tag, so a reply that merely talks about memory is still published Buzz seen-event shutdown stays bounded and retryable: `aflush()` awaits any in-flight write, attempts at most one final snapshot, leaves a still-changing generation dirty for fail-open replay, and returns without a live persistence timer. A Gateway cancellation does not cancel the underlying worker-thread write; `BuzzChannel.stop()` tracks cleanup completion separately from transport admission so ChannelService can retry the retained channel without racing a second snapshot against that write. The store is quiesced before stop (including the already-stopped guard), so a timed-out relay task that records after stop only marks data dirty and cannot schedule detached file work; a repeated `stop()` still drains that dirty state, while `BuzzChannel.start()` explicitly resumes scheduling and flushes it automatically. - `github.py` - Webhook-driven GitHub channel. Inbound messages come from `POST /api/webhooks/github`; outbound is log-only because GitHub agents post explicitly with `gh` from their sandbox when they choose to comment or create a PR diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index 629e2956dc..4c306b4555 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -11,7 +11,9 @@ import stat import time import uuid from collections import OrderedDict -from collections.abc import Awaitable, Callable, Mapping +from collections.abc import Awaitable, Callable, Iterator, Mapping +from contextlib import contextmanager +from contextvars import ContextVar from dataclasses import dataclass from html import escape from pathlib import Path @@ -100,6 +102,74 @@ STREAM_UPDATE_MIN_CHARS = 60 # flush immediately when this many chars accumulat STREAM_MODES = ["messages-tuple", "values"] MESSAGE_STREAM_EVENTS = ("messages-tuple", "messages") THREAD_BUSY_MESSAGE = "This conversation is already processing another request. Please wait for it to finish and try again." + +# IM runs must outlive the request that created them. The manager holds a +# ``runs.wait`` / ``runs.stream`` request open for the whole run, and that +# request dies with this Gateway instance: a leader hand-off, a rolling +# restart, a crash. The Gateway's default ``on_disconnect="cancel"`` would then +# cancel the run itself; ``continue`` lets it finish and persist its thread +# state, so the user still finds the answer in the Web UI even though the IM +# reply is lost with the transport. ``runs.create`` (fire-and-forget) needs no +# such policy: its request returns as soon as the run is pending. +CHANNEL_RUN_ON_DISCONNECT = "continue" + + +@dataclass +class _RunAdmission: + """How far one inbound message's run request got, for dedupe release. + + ``requested`` flips right before the run request leaves the manager; + ``confirmed`` once the Gateway proved it admitted the run (a run id, a + stream frame, a completed response). See ``_keep_dedupe_key_after``. + """ + + requested: bool = False + confirmed: bool = False + + +_RUN_ADMISSION: ContextVar[_RunAdmission | None] = ContextVar("deerflow_channel_run_admission", default=None) + +# Transport failures that can only happen after the Gateway read the request: +# the run was admitted and continues (``on_disconnect: continue``), so a +# provider redelivery must stay a duplicate. Connection failures and HTTP +# status errors mean the Gateway never admitted it. +_RUN_REQUEST_REACHED_GATEWAY_ERRORS = (httpx.ReadTimeout, httpx.ReadError, httpx.RemoteProtocolError) + + +class _RunFailedFrame(RuntimeError): + """The stream's ``error`` frame: the run ended, so a retry is not a duplicate.""" + + +@contextmanager +def _run_admission_scope(*, fresh: bool = False) -> Iterator[_RunAdmission]: + """Install the admission record for one inbound message. + + The worker loop installs a fresh one per message; direct callers of + ``_handle_message`` get one only when none is installed yet. + """ + existing = _RUN_ADMISSION.get() + if existing is not None and not fresh: + yield existing + return + admission = _RunAdmission() + token = _RUN_ADMISSION.set(admission) + try: + yield admission + finally: + _RUN_ADMISSION.reset(token) + + +def _note_run_requested() -> None: + if (admission := _RUN_ADMISSION.get()) is not None: + admission.requested = True + + +def _note_run_confirmed() -> None: + if (admission := _RUN_ADMISSION.get()) is not None: + admission.requested = True + admission.confirmed = True + + # Recovery replies must say the turn was dropped: the run was rejected before # starting, so the triggering message never reached the agent. _STALE_PIN_RETRY_NOTE = "Your message was not processed — please send it again." @@ -2261,7 +2331,7 @@ class ChannelManager: # for them. Scope one trace id per message here -- the worker task # is long-lived and reused, so the scope must close with the # message rather than leak into the next one. - with ensure_trace_context(): + with ensure_trace_context(), _run_admission_scope(fresh=True): dedupe_recorded = False try: # Dedupe before logging "received" so a provider retrying an @@ -2280,24 +2350,25 @@ class ChannelManager: ) # Deliberately awaited inline: never create a task per message. await self._handle_message(msg) - except asyncio.CancelledError: + except asyncio.CancelledError as exc: # A cancellation after dedupe admission must make provider # redelivery retryable rather than retaining a TTL-long key for - # work that never completed. - if dedupe_recorded: + # work that never completed -- unless the Gateway may already + # be executing the run, which outlives this worker. + if dedupe_recorded and not self._keep_dedupe_key_after(msg, exc): try: await self._release_inbound_dedupe_key(msg) except Exception: logger.exception("[Manager] failed to release inbound dedupe key during worker cancellation") raise - except Exception: + except Exception as exc: logger.exception( "[Manager] inbound worker %d failed handling channel=%s chat_id=%s", worker_index, msg.channel_name, msg.chat_id, ) - if dedupe_recorded: + if dedupe_recorded and not self._keep_dedupe_key_after(msg, exc): try: await self._release_inbound_dedupe_key(msg) except Exception: @@ -2368,11 +2439,43 @@ class ChannelManager: recorded on receipt so retries arriving *while* the message is being handled are still deduped, but if handling fails we must not turn a recoverable error into a TTL-long black hole for the same message_id. + Generic failure paths ask ``_keep_dedupe_key_after`` first; sites that + know the Gateway rejected the run (busy thread, stale pin) release + unconditionally. """ key = self._inbound_dedupe_key(msg) if key is not None: await self._inbound_dedupe_store.release(key) + @staticmethod + def _keep_dedupe_key_after(msg: InboundMessage, exc: BaseException | None) -> bool: + """Whether *msg*'s dedupe key must outlive a failed or cancelled handling. + + Runs are created with ``on_disconnect: continue``, so once the Gateway + may have admitted this message's run, the run outlives the request and + a provider redelivery must stay a duplicate: a second run would repeat + the first one's side effects. The key is released, so the redelivery + can retry, only while the manager still owns the outcome: before the + run request leaves it, when the Gateway provably rejected the request + (an HTTP status error, a connection that never opened), or after a + terminal ``error`` frame ended the run. ``/wait`` confirms admission + only with its final response, so a cancellation or a post-request + transport failure while it is in flight counts as admitted rather than + risking a duplicate run. + """ + admission = _RUN_ADMISSION.get() + if admission is None or not admission.requested or isinstance(exc, _RunFailedFrame): + return False + if admission.confirmed or isinstance(exc, (asyncio.CancelledError, *_RUN_REQUEST_REACHED_GATEWAY_ERRORS)): + logger.info( + "[Manager] keeping inbound dedupe key, the run may continue on the Gateway: channel=%s, chat_id=%s, cause=%s", + msg.channel_name, + msg.chat_id, + type(exc).__name__ if exc is not None else "completed", + ) + return True + return False + @staticmethod def _log_task_error(task: asyncio.Task) -> None: """Surface unhandled exceptions from background tasks.""" @@ -2383,6 +2486,10 @@ class ChannelManager: logger.error("[Manager] unhandled error in message task: %s", exc, exc_info=exc) async def _handle_message(self, msg: InboundMessage) -> None: + with _run_admission_scope(): + await self._dispatch_message(msg) + + async def _dispatch_message(self, msg: InboundMessage) -> None: msg = _apply_effective_owner(msg) try: # Commands are handled below because provider adapters consume @@ -2415,7 +2522,7 @@ class ChannelManager: exc, ) await self._send_error(msg, str(exc)) - except Exception: + except Exception as exc: logger.exception( "Error handling message from %s (chat=%s)", msg.channel_name, @@ -2423,9 +2530,11 @@ class ChannelManager: ) # Transient/unexpected failure: release the dedupe key so a provider # redelivery of the same message can recover instead of being dropped - # for the dedupe TTL. Stale /model pin admission failures are handled - # at the run-creation sites, which know the selection the run used. - await self._release_inbound_dedupe_key(msg) + # for the dedupe TTL -- unless the run may already be executing. + # Stale /model pin admission failures are handled at the + # run-creation sites, which know the selection the run used. + if not self._keep_dedupe_key_after(msg, exc): + await self._release_inbound_dedupe_key(msg) await self._send_error(msg, "An internal error occurred. Please try again.") # -- chat handling ----------------------------------------------------- @@ -3005,6 +3114,7 @@ class ChannelManager: thread_id, len(msg.text or ""), ) + _note_run_requested() try: # Capturing the return value is new (issue #4121 Slice 2): # it carries ``run_id``, which the follow-up watcher below @@ -3029,11 +3139,15 @@ class ChannelManager: await self._release_inbound_dedupe_key(msg) return raise + _note_run_confirmed() if policy.buffer_followups_on_busy: self._maybe_spawn_followup_watcher(thread_id, result, msg) return + # Only now: ``runs.create`` above has no ``on_disconnect`` parameter. + run_kwargs["on_disconnect"] = CHANNEL_RUN_ON_DISCONNECT logger.info("[Manager] invoking runs.wait(thread_id=%s, text_len=%d)", thread_id, len(msg.text or "")) + _note_run_requested() try: result = await client.runs.wait( thread_id, @@ -3053,6 +3167,7 @@ class ChannelManager: await self._send_error(msg, recovery.text) return raise + _note_run_confirmed() response_text = _extract_response_text(result) pending_clarification = _has_current_turn_clarification(result) @@ -3121,17 +3236,21 @@ class ChannelManager: "context": run_context, "stream_mode": list(STREAM_MODES), "multitask_strategy": "reject", + "on_disconnect": CHANNEL_RUN_ON_DISCONNECT, } _apply_run_origin(stream_kwargs, msg) if owner_headers := _owner_headers(msg): stream_kwargs["headers"] = owner_headers + _note_run_requested() try: async for chunk in client.runs.stream( thread_id, assistant_id, **stream_kwargs, ): + # Any frame proves the Gateway admitted the run. + _note_run_confirmed() event = getattr(chunk, "event", "") data = getattr(chunk, "data", None) @@ -3149,7 +3268,7 @@ class ChannelManager: ) # The v1 SDK yields backend errors as frames, not exceptions. # Keep draining the stream and use the existing failure path. - stream_error = RuntimeError("Agent run failed.") + stream_error = _RunFailedFrame("Agent run failed.") elif event == "values" and isinstance(data, (dict, list)): last_values = data # Clarification text is only in the values snapshot; @@ -3241,11 +3360,12 @@ class ChannelManager: metadata=_response_metadata_with_clarification(result, msg.metadata, pending_clarification=pending_clarification), ) ) - if stream_error is not None: + if stream_error is not None and not self._keep_dedupe_key_after(msg, stream_error): # This path swallows its own errors, so _handle_message's generic # handler never runs and never releases the key. Release only # after publishing the final outbound so a provider redelivery - # cannot overtake this attempt's terminal reply. + # cannot overtake this attempt's terminal reply. A stream that + # broke after the run was admitted keeps the key: the run goes on. await self._release_inbound_dedupe_key(msg) # -- command handling -------------------------------------------------- diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index cd733bcd3d..9d33aa5324 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -814,7 +814,7 @@ Every IM channel except GitHub connects outbound to its platform (Telegram long - Runtime channel configuration does not cross instances yet: `POST /api/channels/{provider}/runtime-config` saves the values (in the shared `runtime-config.json`) and applies them only where the provider's transport runs — a non-holder accepts them without restarting the holder, which keeps its current credentials until it next starts the channel (an instance taking a lease over reloads the provider's saved config first). Make credential changes on a single instance or restart the holder afterwards. - The block is validated on every topology: an unknown key, a non-mapping value, `enabled` outside `auto | true | false`, `lease_seconds < 3` or `grace_seconds < 0` stop startup with the dotted path in the message. - Restart-required (`channels.leader_lease` in the reload-boundary registry): the gate evaluates it once and `ChannelService` captures it at construction. Change it on every instance and restart them together. -- Known limits: a leader change cancels the IM runs in flight on the old holder (their `on_disconnect` is `cancel`), and transport-local state such as Feishu running cards is lost on failover. +- Known limits: a leader change loses the IM replies in flight on the old holder. The runs themselves continue on the Gateway executing them (channel runs are created with `on_disconnect: continue`) and their thread state is persisted, so the answer stays readable in the web UI, but the transport-local state that would deliver it (a Telegram placeholder message, a Feishu running card, a DingTalk replier) is gone with the old holder; a run executing on the instance that died is recovered as an error like any run of a dead worker. ### Credentials encryption key (`DEER_FLOW_CREDENTIALS_KEY`) diff --git a/backend/docs/IM_CHANNEL_CONNECTIONS.md b/backend/docs/IM_CHANNEL_CONNECTIONS.md index 24526b9c4a..511472f9d5 100644 --- a/backend/docs/IM_CHANNEL_CONNECTIONS.md +++ b/backend/docs/IM_CHANNEL_CONNECTIONS.md @@ -631,7 +631,16 @@ platform connection to one instance: the holder of the provider's row in state, so the connect flow works on any instance; `GET /api/channels/` names each provider's `leader`. Runtime credentials saved on an instance that does not hold the provider apply on the holder only when it next starts the -channel. See +channel. Channel runs are created with `on_disconnect: continue`, so a run +keeps executing when the request that created it dies with its instance (a +leader hand-off, a rolling restart, a crash): its thread state is persisted +and the answer stays readable in the web UI, while the IM reply itself is +lost with the old holder's transport state. Because that run survives, the +manager keeps the inbound dedupe key once a run request has left it (a +cancelled worker, a read timeout, a dropped connection), so the provider's +redelivery of the same message on the new holder is dropped instead of +executing the turn twice; a request the Gateway rejected, or a failure before +the request was sent, still releases the key for a retry. See [IM channel leader lease](CONFIGURATION.md#im-channel-leader-lease-channelsleader_lease). ## Security Notes diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index a3188eec7a..0100b5e450 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -12,9 +12,11 @@ from pathlib import Path from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch +import httpx import pytest from app.channels.base import Channel +from app.channels.dedupe_store import MemoryInboundDedupeStore from app.channels.message_bus import ( PENDING_CLARIFICATION_METADATA_KEY, InboundMessage, @@ -1435,15 +1437,22 @@ class TestChannelManager: _run(go()) - def test_streaming_transient_failure_releases_dedupe_key(self, tmp_path, monkeypatch): - """Release a swallowed streaming error only after its final outbound. + @pytest.mark.parametrize("admitted", [False, True], ids=["before-admission", "after-admission"]) + def test_streaming_transient_failure_releases_dedupe_key_only_before_admission(self, tmp_path, monkeypatch, admitted): + """A swallowed streaming error releases the key only while the manager + still owns the outcome, and only after its final outbound. _release_inbound_dedupe_key lives in _handle_message's `except Exception` handler, but _handle_streaming_chat handles its own errors and never re-raises — so without an explicit release the key recorded on receipt survives the full dedupe TTL and the provider's redelivery (the retry - that would recover the failure) is silently dropped. Releasing before - the final outbound would let that retry overtake the terminal reply. + that would recover a failure) is silently dropped. Releasing before the + final outbound would let that retry overtake the terminal reply. + + Once the Gateway has admitted the run (a frame arrived), a broken + stream no longer releases the key: the run continues on the Gateway + (``on_disconnect: continue``), so the redelivery would execute the + same turn twice. """ monkeypatch.setattr("app.channels.manager.STREAM_UPDATE_MIN_INTERVAL_SECONDS", 0.0) from app.channels.manager import ChannelManager @@ -1465,10 +1474,11 @@ class TestChannelManager: def _failing_stream(*args, **kwargs): async def gen(): - yield _make_stream_part( - "messages-tuple", - [{"id": "ai-1", "content": "Partial", "type": "AIMessageChunk"}, {"langgraph_node": "agent"}], - ) + if admitted: + yield _make_stream_part( + "messages-tuple", + [{"id": "ai-1", "content": "Partial", "type": "AIMessageChunk"}, {"langgraph_node": "agent"}], + ) raise ConnectionError("stream broken") return gen() @@ -1494,10 +1504,12 @@ class TestChannelManager: # The provider redelivers the same message after the failure. await bus.publish_inbound(_inbound()) - await _wait_for(lambda: manager._client.runs.stream.call_count == 2) + await asyncio.wait_for(bus.join_inbound(), timeout=5) await manager.stop() - assert manager._client.runs.stream.call_count == 2 + # Before admission the retry runs; after admission it is a duplicate + # of a run the Gateway is still executing and must be dropped. + assert manager._client.runs.stream.call_count == (1 if admitted else 2) _run(go()) @@ -1947,7 +1959,7 @@ class TestChannelManager: history_by_checkpoint: dict[tuple[str, str], list[str]] = {} - async def _runs_wait(thread_id, assistant_id, *, input, config, context, multitask_strategy=None, metadata=None): + async def _runs_wait(thread_id, assistant_id, *, input, config, context, multitask_strategy=None, metadata=None, on_disconnect=None): del assistant_id, context # unused in this test, kept for signature parity checkpoint_ns = config.get("configurable", {}).get("checkpoint_ns") @@ -5548,6 +5560,10 @@ class TestGithubFireAndForget: # multitask_strategy must still be ``reject`` — concurrent runs on # the same GitHub thread are surfaced via ConflictError below. assert create_args[1]["multitask_strategy"] == "reject" + # ``runs.create`` returns as soon as the run is pending, so there is + # no long-lived request whose disconnect could cancel the run — and + # the SDK's ``create`` has no ``on_disconnect`` parameter at all. + assert "on_disconnect" not in create_args[1] _run(go()) @@ -5729,6 +5745,249 @@ async def _stale_pin_drain_setup(thread_id, *, model, pin=None, session_model=No return manager, carrier_msg, mock_client, bus +class TestChannelRunDisconnectPolicy: + """IM runs must outlive the HTTP request that created them. + + The manager holds ``runs.wait`` / ``runs.stream`` open for the whole run, + and that request dies with the receiving Gateway instance: a leader + hand-off, a rolling restart, a crash. The Gateway's default + ``on_disconnect`` is ``cancel``, which would cancel the run itself, so + every long-lived channel request asks for ``continue``: the run finishes + and persists its thread state even though the IM reply is lost with the + transport. + """ + + def test_runs_wait_asks_gateway_to_continue_on_disconnect(self): + from app.channels.manager import CHANNEL_RUN_ON_DISCONNECT, ChannelManager + + assert CHANNEL_RUN_ON_DISCONNECT == "continue" + + async def go(): + bus = MessageBus() + store = JsonChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store) + mock_client = _make_mock_langgraph_client(thread_id="slack-thread-1") + manager._client = mock_client + + await manager._handle_chat(InboundMessage(channel_name="slack", chat_id="C123", user_id="U1", text="hi")) + + mock_client.runs.wait.assert_called_once() + wait_kwargs = mock_client.runs.wait.call_args.kwargs + assert wait_kwargs["on_disconnect"] == "continue" + assert wait_kwargs["multitask_strategy"] == "reject" + + _run(go()) + + def test_runs_stream_asks_gateway_to_continue_on_disconnect(self): + from app.channels.manager import ChannelManager + + async def go(): + bus = MessageBus() + store = JsonChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + manager = ChannelManager(bus=bus, store=store) + captured: dict[str, object] = {} + + async def _stream(thread_id, assistant_id, **kwargs): # noqa: ARG001 + captured.update(kwargs) + yield _make_stream_part( + "values", + { + "messages": [ + {"type": "human", "content": "hi"}, + {"type": "ai", "content": "done"}, + ], + "artifacts": [], + }, + ) + + mock_client = _make_mock_langgraph_client(thread_id="feishu-thread-1") + mock_client.runs.stream = MagicMock(side_effect=_stream) + manager._client = mock_client + + await manager._handle_chat(InboundMessage(channel_name="feishu", chat_id="chat1", user_id="user1", text="hi")) + + mock_client.runs.stream.assert_called_once() + assert captured["on_disconnect"] == "continue" + assert captured["multitask_strategy"] == "reject" + + _run(go()) + + +class TestDedupeKeySurvivesAdmittedRun: + """A provider redelivery must not start a second run for a turn the + Gateway may already be executing. + + Channel runs are created with ``on_disconnect: continue`` (see + ``TestChannelRunDisconnectPolicy``), so a run outlives the manager worker + that requested it. Releasing the inbound dedupe key when that worker is + cancelled (a leader hand-off, a shutdown) would let the provider's + redelivery through on the replacement instance and execute the same turn + twice. The key is therefore released only while the manager still owns the + outcome: before the run request leaves it, or when the Gateway provably + rejected the request. + """ + + @staticmethod + def _message(channel: str) -> InboundMessage: + if channel == "github": + # Shaped like app.gateway.github.dispatcher.fanout_event output. + return InboundMessage( + channel_name="github", + chat_id="zhfeng/llm-gateway", + user_id="alice", + owner_user_id="alice", + text="@bot please review", + topic_id="7:reviewer", + workspace_id="zhfeng/llm-gateway", + metadata={"message_id": "d1:alice:reviewer", "agent_name": "reviewer"}, + ) + return InboundMessage(channel_name=channel, chat_id="C1", user_id="U1", text="hi", metadata={"event_id": "evt-1", "team_id": "T1"}) + + @staticmethod + def _manager(store: MemoryInboundDedupeStore, bus: MessageBus): + from app.channels.manager import ChannelManager + + return ChannelManager( + bus=bus, + store=JsonChannelStore(path=Path(tempfile.mkdtemp()) / "store.json"), + inbound_dedupe_store=store, + max_concurrency=1, + shutdown_grace_period_seconds=0, + ) + + @staticmethod + def _final_stream(*_args, **_kwargs): + async def gen(): + yield _make_stream_part( + "values", + {"messages": [{"type": "human", "content": "hi"}, {"type": "ai", "content": "done"}], "artifacts": []}, + ) + + return gen() + + async def _hand_off(self, channel: str, *, block: str) -> tuple[MagicMock, bool]: + """Cancel instance A while it is blocked at *block*, then redeliver on B. + + Returns B's LangGraph client and whether the dedupe key was still + recorded between A's cancellation and B's redelivery. + """ + import app.gateway.github.run_policy # noqa: F401 — register the fire-and-forget policy + from app.channels.manager import ChannelManager + + store = MemoryInboundDedupeStore() + blocked = asyncio.Event() + + async def _block(*_args, **_kwargs): + blocked.set() + await asyncio.Event().wait() + + def _blocking_stream(*_args, **_kwargs): + async def gen(): + yield _make_stream_part( + "messages-tuple", + [{"id": "ai-1", "content": "Partial", "type": "AIMessageChunk"}, {"langgraph_node": "agent"}], + ) + blocked.set() + await asyncio.Event().wait() + + return gen() + + bus_a = MessageBus() + manager_a = self._manager(store, bus_a) + client_a = _make_mock_langgraph_client(thread_id="thread-a") + if block == "thread_create": + client_a.threads.create = AsyncMock(side_effect=_block) + elif block == "wait": + client_a.runs.wait = AsyncMock(side_effect=_block) + elif block == "stream": + client_a.runs.stream = MagicMock(side_effect=_blocking_stream) + elif block == "create": + client_a.runs.create = AsyncMock(side_effect=_block) + else: + raise AssertionError(block) + manager_a._client = client_a + await manager_a.start() + await bus_a.publish_inbound(self._message(channel)) + await asyncio.wait_for(blocked.wait(), timeout=5) + # Grace period 0: stop() cancels the worker in the middle of the handler, + # exactly what a leader hand-off or a shutdown does. + await manager_a.stop() + + key = ChannelManager._inbound_dedupe_key(self._message(channel)) + assert key is not None + recorded_after_cancel = key in store._store + + bus_b = MessageBus() + manager_b = self._manager(store, bus_b) + client_b = _make_mock_langgraph_client(thread_id="thread-b") + client_b.runs.stream = MagicMock(side_effect=self._final_stream) + client_b.runs.create = AsyncMock(return_value={"run_id": "run-b", "status": "pending"}) + manager_b._client = client_b + await manager_b.start() + await bus_b.publish_inbound(self._message(channel)) + await asyncio.wait_for(bus_b.join_inbound(), timeout=5) + await manager_b.stop() + return client_b, recorded_after_cancel + + @staticmethod + def _run_requests(client: MagicMock) -> int: + return client.runs.wait.call_count + client.runs.stream.call_count + client.runs.create.call_count + + @pytest.mark.parametrize( + ("channel", "block"), + [("slack", "wait"), ("feishu", "stream"), ("github", "create")], + ids=["slack-wait", "feishu-stream", "github-create"], + ) + def test_cancel_after_the_run_request_keeps_the_key_so_redelivery_is_dropped(self, channel, block): + async def go(): + client_b, recorded_after_cancel = await self._hand_off(channel, block=block) + assert recorded_after_cancel is True + assert self._run_requests(client_b) == 0 + + _run(go()) + + @pytest.mark.parametrize("channel", ["slack", "feishu", "github"]) + def test_cancel_before_the_run_request_releases_the_key_so_redelivery_retries(self, channel): + async def go(): + client_b, recorded_after_cancel = await self._hand_off(channel, block="thread_create") + assert recorded_after_cancel is False + assert self._run_requests(client_b) == 1 + + _run(go()) + + @pytest.mark.parametrize( + ("error", "kept"), + [ + (httpx.ReadTimeout("read timed out"), True), + (httpx.RemoteProtocolError("peer closed connection"), True), + (httpx.ConnectError("connection refused"), False), + (RuntimeError("unexpected"), False), + ], + ids=["read-timeout-keeps", "protocol-error-keeps", "connect-error-releases", "unexpected-error-releases"], + ) + def test_wait_failure_releases_the_key_only_when_the_gateway_cannot_have_admitted_the_run(self, error, kept): + """``/wait`` confirms admission only with its final response, so a + failure after the request was read by the Gateway (a read timeout, a + dropped connection) must keep the key: the run is still executing.""" + + async def go(): + from app.channels.manager import ChannelManager + + store = MemoryInboundDedupeStore() + manager = self._manager(store, MessageBus()) + client = _make_mock_langgraph_client(thread_id="thread-1") + client.runs.wait = AsyncMock(side_effect=error) + manager._client = client + msg = self._message("slack") + assert await manager._is_duplicate_inbound(msg) is False # recorded on receipt, as the worker does + + await manager._handle_message(msg) + + assert (ChannelManager._inbound_dedupe_key(msg) in store._store) is kept + + _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. diff --git a/deploy/helm/deer-flow/README.md b/deploy/helm/deer-flow/README.md index 63b01f4391..2986ee7847 100644 --- a/deploy/helm/deer-flow/README.md +++ b/deploy/helm/deer-flow/README.md @@ -309,7 +309,8 @@ kubectl -n deer-flow exec deploy/deer-flow-provisioner -- curl -s localhost:8002 next heartbeat. The startup gate refuses `enabled: false` while channels are enabled. Credentials saved from the web UI reach the holding Pod only when it next starts the channel, so restart it after changing them, and a failover - cancels the IM replies in flight on the old holder. `gateway.replicas: 1` + loses the IM replies in flight on the old holder, though the runs themselves + finish and their threads stay readable in the web UI. `gateway.replicas: 1` stays the default because the WeChat QR login and browser tools are still single-instance. Before raising replicas, also set `persistence.home.accessMode: ReadWriteMany` on multi-node clusters (thread diff --git a/frontend/src/content/en/application/deployment-guide.mdx b/frontend/src/content/en/application/deployment-guide.mdx index c8f833631b..40b5a4fed2 100644 --- a/frontend/src/content/en/application/deployment-guide.mdx +++ b/frontend/src/content/en/application/deployment-guide.mdx @@ -156,7 +156,7 @@ The Helm chart ships `gateway.replicas: 1`. Raising it is supported for the web The chart exports `DEER_FLOW_MULTI_INSTANCE=true` when `gateway.replicas > 1`, and the Gateway refuses to start if any of these is missing. If you scale with `kubectl scale` or an HPA instead of `gateway.replicas`, set `gateway.multiInstance: true` first so the declaration, the PodDisruptionBudget and the required shared `AUTH_JWT_SECRET` do not keep their single-replica rendering (the PodDisruptionBudget then blocks node drains of the single Pod until the scale-up happens; disable it with `gateway.podDisruptionBudget.enabled: false` for that interim if needed). On multi-node clusters also set `persistence.home.accessMode: ReadWriteMany` (uploads, outputs, memory and `extensions_config.json` live on that volume; MCP and skill changes made through one Pod are then visible to the others on their next request) and `agent_storage.backend: db` so custom agents are visible on every Pod. -IM channels work on several replicas through `channels.leader_lease` in `config` (default `auto`): one Pod holds each platform connection, and when it dies another Pod takes over within `lease_seconds + grace_seconds` plus one renewal interval (50 seconds by default). The startup gate refuses `enabled: false` while channels are enabled. Channel credentials saved from the web UI reach the Pod holding that platform only when it next starts the channel, so restart that Pod after changing them; a takeover cancels the IM replies the old Pod was still generating. Still single-instance: the WeChat QR login and browser tools. Keep one replica while those are enabled. +IM channels work on several replicas through `channels.leader_lease` in `config` (default `auto`): one Pod holds each platform connection, and when it dies another Pod takes over within `lease_seconds + grace_seconds` plus one renewal interval (50 seconds by default). The startup gate refuses `enabled: false` while channels are enabled. Channel credentials saved from the web UI reach the Pod holding that platform only when it next starts the channel, so restart that Pod after changing them; a takeover loses the IM replies the old Pod was still delivering, although the runs themselves finish and their threads stay readable in the web UI. Still single-instance: the WeChat QR login and browser tools. Keep one replica while those are enabled. ### K8s Provisioner setup diff --git a/frontend/src/content/zh/application/deployment-guide.mdx b/frontend/src/content/zh/application/deployment-guide.mdx index 30597033a5..3fb3d9cc34 100644 --- a/frontend/src/content/zh/application/deployment-guide.mdx +++ b/frontend/src/content/zh/application/deployment-guide.mdx @@ -153,7 +153,7 @@ Helm chart 默认 `gateway.replicas: 1`。当所有副本共享同一个 Postgre 当 `gateway.replicas > 1` 时，chart 会导出 `DEER_FLOW_MULTI_INSTANCE=true`，缺少上述任一项时 Gateway 会拒绝启动。如果你通过 `kubectl scale` 或 HPA 而不是 `gateway.replicas` 扩容，请先设置 `gateway.multiInstance: true`：它们在不执行 Helm upgrade 的情况下改变副本数，否则该声明、PodDisruptionBudget 和必需的共享 `AUTH_JWT_SECRET` 都会停留在单副本的渲染结果上（在真正扩容之前，PodDisruptionBudget 会阻止驱逐唯一的 Pod 和排空节点；如有需要，可在此期间设置 `gateway.podDisruptionBudget.enabled: false`）。多节点集群还需要把 `persistence.home.accessMode` 设为 `ReadWriteMany`（上传、产物、记忆与 `extensions_config.json` 都在该卷上；在一个 Pod 上修改的 MCP 与技能配置会在其他 Pod 的下一次请求中生效），并设置 `agent_storage.backend: db` 让自定义 agent 在每个 Pod 上可见。 -IM 通道可以通过 `config` 中的 `channels.leader_lease`（默认 `auto`）在多副本下运行：每个平台的连接只由一个 Pod 持有，该 Pod 宕机后，另一个 Pod 会在 `lease_seconds + grace_seconds` 再加一个续约周期内接管（默认 50 秒）。启用通道时启动门禁会拒绝 `enabled: false`。在 Web UI 中保存的通道凭据要等持有该平台的 Pod 下次启动通道时才会生效，修改后请重启该 Pod；接管会取消旧 Pod 上仍在生成的 IM 回复。仍然只能单实例运行的部分：WeChat 扫码登录、浏览器工具。启用这些功能时请保持 1 个副本。 +IM 通道可以通过 `config` 中的 `channels.leader_lease`（默认 `auto`）在多副本下运行：每个平台的连接只由一个 Pod 持有，该 Pod 宕机后，另一个 Pod 会在 `lease_seconds + grace_seconds` 再加一个续约周期内接管（默认 50 秒）。启用通道时启动门禁会拒绝 `enabled: false`。在 Web UI 中保存的通道凭据要等持有该平台的 Pod 下次启动通道时才会生效，修改后请重启该 Pod；接管会丢失旧 Pod 上仍在投递的 IM 回复，但运行本身会继续完成，其对话线程仍可在 Web UI 中查看。仍然只能单实例运行的部分：WeChat 扫码登录、浏览器工具。启用这些功能时请保持 1 个副本。 ### K8s Provisioner 配置