From f85ee590e132c804fb12b995e3626fe886c0c71c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E4=BA=91=E9=BE=99?= <76432572+nankingjing@users.noreply.github.com> Date: Sun, 12 Jul 2026 11:42:55 +0800 Subject: [PATCH] docs(backend): add IM channel connections and GitHub agents architecture diagrams (#4112) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds two focused architecture docs and references from backend/AGENTS.md: - backend/docs/IM_CHANNEL_CONNECTIONS.md — sequence + graph diagrams for user-owned IM channel connections: bind-code flow (/connect/ --- backend/AGENTS.md | 14 +- backend/docs/GITHUB_AGENTS.md | 258 +++++++++++++++++++++++++ backend/docs/IM_CHANNEL_CONNECTIONS.md | 239 ++++++++++++++++++++++- 3 files changed, 501 insertions(+), 10 deletions(-) create mode 100644 backend/docs/GITHUB_AGENTS.md diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 6158244aa..0b6ee3ed0 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -536,16 +536,12 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_ - Telegram, Slack, Discord, Feishu/Lark, DingTalk, WeChat, and WeCom workers resolve incoming platform identities to connection records before reaching `ChannelManager`. - **Connect-code ordering vs `allowed_users`**: inbound workers consume a valid `/connect ` (or Telegram `/start `) **before** applying the `allowed_users` filter, so a newly allowlisted-but-unbound user can bootstrap their first bind via the browser flow. Consequence: `allowed_users` is **not** a bind-time defense — any sender who possesses a valid code can consume it (not only allowlisted users). The bind security model rests on the code's confidentiality: `secrets.token_urlsafe(16)`, 600 s TTL, one-time `consume_oauth_state`, and codes surfaced only in the initiating browser (never echoed to chat). `allowed_users` still gates ordinary (non-bind) messages. - **Single-active-owner transfer semantics**: an external identity is keyed by `(provider, external_account_id, workspace_id)`. The latest successful bind wins — `upsert_connection` revokes other owners' active rows for the same identity (ownership transfer). This invariant is enforced at the DB layer by the partial unique index `uq_channel_connection_active_identity` (`WHERE status != 'revoked'`), so concurrent connects from different owners cannot both end `connected`; the losing writer retries against the now-visible state. `find_connection_by_external_identity` therefore resolves deterministically. -- See `backend/docs/IM_CHANNEL_CONNECTIONS.md` for provider setup and operational notes. +- See `backend/docs/IM_CHANNEL_CONNECTIONS.md` for provider setup, operational notes, and the architecture diagrams (connect-code flow, single-active-owner transfer, sync vs streaming dispatch, owner-scoped file storage pipeline). -**GitHub event-driven agents**: -- Configure agent-level bindings in a custom agent's `config.yaml` under `github:`. The global `config.yaml` `channels.github` block is only for the operator kill-switch (`enabled`) and the default mention login; per-agent `installation_id`, `bot_login`, repo bindings, and triggers live with the custom agent. -- Bindings are opt-in by event. `DEFAULT_TRIGGERS` only supplies per-event field defaults for events a binding declared. `GitHubAgentConfig` enforces a single binding per repo per agent; merge trigger maps instead of duplicating a repo. -- Threading is deterministic: fan-out sets `metadata["preferred_thread_id"]` from UUID5 over `(repo, PR/issue number, agent_name)`, and `ChannelManager._create_thread` passes it to `client.threads.create(thread_id=...)`. Different agents on the same PR intentionally get different LangGraph threads. ChannelStore uses `topic_id = f"{number}:{agent_name}"` so each agent's cached mapping is independent. -- Thread-create race recovery is narrow by design: only `langgraph_sdk.errors.ConflictError` (HTTP 409) is treated as a concurrent-create collision and followed by `threads.get(preferred_thread_id)` verification. Other create failures propagate so the delivery can fail/retry rather than caching an unverified mapping. -- Mention-handle precedence for `require_mention` triggers is `trigger.mention_login` → `github.bot_login` → `channels.github.default_mention_login` → `agent.name`. Whitespace-only defaults are treated as unset. -- Set `GITHUB_APP_ID` and `GITHUB_APP_PRIVATE_KEY_PATH` (or `GITHUB_APP_PRIVATE_KEY`) to enable installation-token minting. `ChannelManager` mints a short-lived installation token from the binding's `installation_id` on the bus-consumer side and passes the token string in `run_context["github_token"]`; the bash tool exposes it to sandbox commands as `GH_TOKEN` / `GITHUB_TOKEN` via per-call `extra_env`. No global `os.environ` mutation is used, so concurrent GitHub runs for different repos do not clobber each other. -- Tokens are not auto-refreshed past GitHub's 1h TTL. Long-running agents may need to finish GitHub writes before expiry until refresh is reintroduced. If minting fails, the agent still runs without push/write credentials. +**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. +- 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. ### Memory System (`packages/harness/deerflow/agents/memory/`) diff --git a/backend/docs/GITHUB_AGENTS.md b/backend/docs/GITHUB_AGENTS.md new file mode 100644 index 000000000..801d74c90 --- /dev/null +++ b/backend/docs/GITHUB_AGENTS.md @@ -0,0 +1,258 @@ +# GitHub Event-Driven Agents + +GitHub is a **webhook-push** channel: there is no long-polling worker. Every GitHub App / repository delivery lands at `POST /api/webhooks/github`, where it is HMAC-verified, fan-out'd to one `InboundMessage` per matching custom-agent binding, and shipped to the rest of DeerFlow through the same `ChannelManager` that handles Feishu/Slack/Telegram. For the high-level orientation, see [AGENTS.md](../AGENTS.md) → "GitHub event-driven agents". + +This document covers the **architecture** of that pipeline: + +- Per-agent bindings (`config.yaml` → `github:` block) +- Webhook → fan-out → `InboundMessage` dispatch +- Mention-handle precedence for `require_mention` triggers +- `preferred_thread_id = UUID5(repo, number, agent_name)` thread determinism +- 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) + +## Overview + +GitHub bindings are declared **per custom agent** in `users/{owner_user_id}/agents/{agent_name}/config.yaml` under a `github:` block. The global `config.yaml` `channels.github` block is intentionally minimal — only the operator kill-switch (`enabled`) and `default_mention_login` live there. Everything that identifies "which agent handles which repo" lives next to the agent that owns it. + +```mermaid +graph LR + classDef operator fill:#D8CFC4,stroke:#6E6259,color:#2F2A26 + classDef agent fill:#E5D2C4,stroke:#806A5B,color:#30251E + classDef route fill:#C9D7D2,stroke:#5D706A,color:#21302C + + OperatorYaml["config.yaml
channels.github:
enabled: true
default_mention_login"]:::operator + AgentYaml["agents/{name}/config.yaml
github:
installation_id
bot_login
bindings: [{repo, triggers}]"]:::agent + Registry["build_github_agent_registry()
(mtime-cached, asyncio.to_thread)"]:::route + Webhook["POST /api/webhooks/github
(HMAC verify)"]:::route + + OperatorYaml --> Registry + AgentYaml --> Registry + Registry --> Webhook +``` + +Each agent binding lists the **events it cares about** under `triggers:`. Events absent from `triggers:` are not delivered to that agent — the dispatcher never loads the agent for them. `DEFAULT_TRIGGERS` only supplies **field-level defaults** (e.g. `require_mention: true`) for events a binding did declare; it is no longer an enablement list. + +## Webhook → Fan-out → Dispatch + +The webhook handler stays cheap — no LangGraph calls — so GitHub's 10-second delivery timeout is never at risk. Verification, fan-out, and the bus publish are all in-process and bounded. + +```mermaid +sequenceDiagram + autonumber + participant GH as GitHub + participant Router as POST /api/webhooks/github
(github_webhooks.py) + participant Disp as fanout_event()
(github/dispatcher.py) + participant Reg as build_github_agent_registry + participant Trg as event_should_fire + participant Bus as MessageBus + participant Mgr as ChannelManager + participant Client as langgraph_sdk client + participant Gateway as Gateway
/api/webhooks/github + + GH->>Router: delivery (event, delivery_id, payload,
X-Hub-Signature-256, X-GitHub-Event) + Router->>Router: _verify_signature()
hmac.compare_digest(sha256, secret) + Router->>Disp: fanout_event(bus, event, delivery_id, payload,
operator_default_mention_login) + Disp->>Reg: build_github_agent_registry() (to_thread) + Reg-->>Disp: agents bound to (repo, event) + loop each matched agent + Disp->>Disp: _is_self_event(sender.login)? + alt self event + Disp-->>Disp: skipped (self_event) + else trigger filter + Disp->>Trg: event_should_fire(event, payload, trigger, default_mention_login) + Trg-->>Disp: (fire, reason) + opt fire + Disp->>Disp: build_prompt() + resolve_thread_id()
UUID5(repo, number, agent) + Disp->>Bus: publish_inbound(InboundMessage(
channel=github, chat_id=repo,
topic_id="{number}:{agent}",
owner_user_id=match.user_id,
metadata.agent_name=agent,
metadata.preferred_thread_id=...,
metadata.github={...})) + end + end + end + Disp-->>Router: summary {matched, fired, skipped} + Router-->>GH: 200 OK + + Note over Bus,Gateway: Bus consumer side + Bus->>Mgr: msg = get_inbound() + Mgr->>Client: client.threads.create(thread_id=preferred_thread_id) + Client->>Gateway: threads.create (with owner headers) + Gateway-->>Client: thread_id (or 409) + Mgr->>Client: runs.create() [fire_and_forget=True] + Client->>Gateway: start run + Gateway-->>Client: pending + Note over Mgr: Manager returns immediately.
Agent posts to GitHub via gh CLI. +``` + +## `preferred_thread_id = UUID5(...)` Thread Determinism + +`resolve_thread_id(repo, issue_or_pr_number, agent_name)` builds a deterministic LangGraph thread id so a `(repo, PR/issue number)` always lands on the same thread — even after a store wipe, even across gateway replicas (same UUID5 namespace). + +```mermaid +graph LR + classDef input fill:#D8CFC4,stroke:#6E6259,color:#2F2A26 + classDef hash fill:#D7D3E8,stroke:#6B6680,color:#29263A + classDef thread fill:#C9D7D2,stroke:#5D706A,color:#21302C + + Repo["repo
owner/name"]:::input + Number["issue/PR number
int"]:::input + Agent["agent_name
[A-Za-z0-9-]+"]:::input + Seed["seed = '{repo}#{number}:{agent}'"]:::hash + UUID5["uuid.uuid5(
GITHUB_THREAD_NAMESPACE,
seed)"]:::hash + Thread["thread_id
(same across replicas + restarts)"]:::thread + + Repo --> Seed + Number --> Seed + Agent --> Seed + Seed --> UUID5 --> Thread +``` + +Different agents on the same PR (coder + reviewer) **deliberately** get different thread ids — `agent_name` is part of the seed. Sharing a thread would couple their message histories and checkpoints, and `multitask_strategy="reject"` would silently drop one run on every dual-mention. Each agent owns its own thread; cross-agent coordination flows through GitHub (PR comments, review threads), the source of truth humans see. + +`ChannelStore` uses `topic_id = f"{number}:{agent_name}"` as its cache key, so each agent's cached mapping is independent — a coder's mapping is invisible to a reviewer on the same PR. + +## Mention-handle Precedence + +For bindings that declare `require_mention: true` on a given event, the dispatcher must resolve **which** mention login gates the trigger. The precedence chain is: + +```mermaid +graph LR + classDef step fill:#C9D7D2,stroke:#5D706A,color:#21302C + classDef fallback fill:#D7D3E8,stroke:#6B6680,color:#29263A + + A["1. trigger.mention_login
(per-event override)"]:::step + B["2. github.bot_login
(agent's App identity)"]:::step + C["3. channels.github.default_mention_login
(operator-wide default)"]:::step + D["4. agent.name
(last-resort fallback)"]:::fallback + Use["effective @mention"]:::step + + A -->|"non-empty"| Use + B -->|"non-empty"| Use + C -->|"non-empty"| Use + D --> Use +``` + +Whitespace-only values at every level are treated as unset, so the chain falls through cleanly. The `_is_self_event` gate uses the same precedence (with the agent's whole `bindings[*].triggers[*].mention_login` aggregated across all bindings, plus an `agent.name` fallback) so the self-loop gate and the mention gate stay coherent. + +## GH Token Lifecycle + +GitHub Agents get push/write credentials as **per-call installation tokens**, not as inherited environment. The minted token string is bound into `run_context["github_token"]` on the bus-consumer side and exposed to the agent's sandbox commands as both `GH_TOKEN` and `GITHUB_TOKEN` via per-call `extra_env` on `execute_command`. No `os.environ` mutation, no cross-repo bleed. + +```mermaid +sequenceDiagram + autonumber + participant Bus as MessageBus + participant Mgr as ChannelManager
_handle_chat_on_thread + participant Pol as inject_github_credentials()
(github/run_policy.py) + participant Auth as mint_installation_token
(github/app_auth.py) + participant GH as GitHub API
(POST /app/installations/{id}/access_tokens) + participant Run as runs.create + participant GW as Gateway runtime + participant Bash as bash_tool + participant Sandbox as Sandbox process + participant Cmd as Sandbox command
(gh / git push) + + Bus->>Mgr: msg (channel=github, metadata.github.installation_id) + Mgr->>Pol: _apply_channel_policy(msg, run_context) + Pol->>Auth: mint_installation_token(installation_id) + Auth->>GH: exchange installation_id for token
(JWT signed with PRIVATE_KEY) + GH-->>Auth: {token, expires_at} + Auth-->>Pol: token string (cached ~55min) + Pol->>Mgr: run_context["github_token"] = token + Mgr->>Run: client.runs.create(
thread_id, assistant_id,
context={..., github_token}) + Run->>GW: POST /threads/{id}/runs + GW-->>Run: pending + Run-->>Mgr: returns once pending + Note over Mgr: Manager returns immediately (fire_and_forget) + + Note over GW,Bash: Harness side + GW->>Bash: bash_tool call
cmd="git push https://x-access-token:$GH_TOKEN@..." + Bash->>Sandbox: execute_command(
env={"GH_TOKEN": "...",
"GITHUB_TOKEN": "..."}) + Note over Sandbox: AioSandbox: bash.exec(env=...) on fresh session
LocalSandbox: subprocess.run(env=...) + Sandbox->>Cmd: gh pr comment / git push / etc. + Cmd->>GH: authenticated GitHub API call +``` + +Why a string and not a closure: `run_context` is JSON-encoded by the `langgraph_sdk` HTTP client before reaching Gateway. A Python callable does not survive that serialization. The harness side (`_github_env_from_runtime`) already accepts either shape, but only `str` round-trips through the SDK transport. + +**Token TTL caveat**: GitHub installation tokens are valid for ~1 hour. Most agent runs finish well inside that window. Truly long coder runs (multi-hour refactors at the higher `recursion_limit=250` ceiling) may see a 401 on a late `git push` / `gh pr create`. Auto-refresh past the 1h TTL is intentionally deferred — it requires registering a token-provider lookup on the harness side, which crosses the harness/app boundary (`tests/test_harness_boundary.py`). Until refresh ships, long runs should finish GitHub writes before expiry or accept the loss. If minting fails (bad App id, wrong installation_id, missing private key), the agent still runs without push/write credentials — read-only is better than no response. + +## Thread-create Race Recovery + +Two webhook deliveries for the same `(repo, number)` can land within milliseconds of each other and race on `threads.create(thread_id=preferred_thread_id)`. The recovery is narrow by design. + +```mermaid +sequenceDiagram + autonumber + participant Mgr1 as ChannelManager
(delivery 1) + participant Mgr2 as ChannelManager
(delivery 2) + participant Client as langgraph_sdk client + participant Gateway as Gateway
threads.create + + par concurrent deliveries + Mgr1->>Client: client.threads.create(thread_id=preferred) + Client->>Gateway: POST /threads {thread_id} + and + Mgr2->>Client: client.threads.create(thread_id=preferred) + Client->>Gateway: POST /threads {thread_id} + end + + Gateway-->>Client: 200 OK (first writer wins) + Gateway-->>Client: 409 ConflictError (second writer) + + Client-->>Mgr1: thread (created) + Client-->>Mgr2: ConflictError + + Note over Mgr2: Recovery branch + Mgr2->>Client: threads.get(preferred_thread_id) + alt existing + Client-->>Mgr2: thread + Mgr2->>Mgr2: _store_thread_id(msg, preferred_thread_id) + Mgr2-->>Mgr2: reuse deterministic id + else also missing + Client-->>Mgr2: error + Mgr2->>Mgr2: raise (do NOT cache the mapping) + end +``` + +The recovery is **narrow**: only `langgraph_sdk.errors.ConflictError` (HTTP 409) is treated as a concurrent-create collision. Any other failure (transient DB outage, network error, 5xx) propagates so the delivery can fail/retry rather than silently caching `preferred_thread_id` into the store and mapping every future webhook on this issue/PR to a thread that was never created (every later run would 404 forever with no retry path). + +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. + +## Outbound is Log-only + +```mermaid +graph LR + classDef agent fill:#E5D2C4,stroke:#806A5B,color:#30251E + classDef send fill:#C9D7D2,stroke:#5D706A,color:#21302C + classDef gh fill:#D7D3E8,stroke:#6B6680,color:#29263A + + Run["GitHub agent run"]:::agent + GhCLI["gh CLI
(in sandbox)"]:::gh + Issue["GitHub issue / PR"]:::gh + Channel["GitHubChannel.send()
(log-only)"]:::send + Log["gateway.log
(INFO line)"]:::send + + Run -->|"mid-run intentional writeback"| GhCLI --> Issue + Run -->|"final assistant message"| Channel --> Log +``` + +GitHub agents post to GitHub themselves via the `gh` CLI from inside their sandbox (`gh issue comment`, `gh pr comment`, `gh pr create`, etc.). The channel's `send()` is **log-only** by design — the agent's final assistant message is logged at INFO for visibility but never auto-posted. + +Why: + +- **Multiple agents can bind the same event.** coder + reviewer on a mention would each auto-post a reply, producing two replies per mention even when only one had useful work. Letting the LLM call `gh` mid-run means silence is just "the LLM did not call `gh`". +- **The agent often wants to post intermediate updates** (an issue comment linking the PR, a sub-issue comment, a PR description edit). The auto-post-the-final-message contract didn't model that and forced the final message to play double duty. +- **The dispatcher's per-agent `_is_self_event` gate** already prevents comments the LLM posts via `gh` from looping the webhook back into a new run for the same agent. + +This is also why the GitHub channel registers `ChannelRunPolicy.fire_and_forget=True`: the manager calls `runs.create()` and returns once the run is `pending`, no outbound ferrying, no SDK 300s `httpx.ReadTimeout` on a legitimate long coder run. + +## 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) +- [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/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` +- `app/gateway/routers/github_webhooks.py` — HMAC verify, route mount predicate +- `app/channels/github.py` — `GitHubChannel` (log-only outbound) \ No newline at end of file diff --git a/backend/docs/IM_CHANNEL_CONNECTIONS.md b/backend/docs/IM_CHANNEL_CONNECTIONS.md index 823f98750..06107cd6a 100644 --- a/backend/docs/IM_CHANNEL_CONNECTIONS.md +++ b/backend/docs/IM_CHANNEL_CONNECTIONS.md @@ -4,7 +4,244 @@ DeerFlow supports user-owned IM channel bindings for Telegram, Slack, Discord, F No public IP, OAuth callback URL, or provider webhook is required in this implementation. -## Configuration +This document covers both **architecture** (how the bind / dispatch / file pipeline fits together) and **configuration / operations** (the existing `config.yaml` knobs and security notes). For the high-level orientation, see [AGENTS.md](../AGENTS.md) → "IM Channels System". + +--- + +## Architecture Overview + +A user-owned IM channel connection is a **per-DeerFlow-user bind layer** layered on top of the existing provider bot credentials in `channels.*`. The connection layer adds three things the bot credentials alone cannot give you: + +1. **Owner identity** — each `(provider, external account, workspace)` maps to exactly one DeerFlow account (`owner_user_id`). Every run created from that connection runs in the owner's bucket (memory, uploads, outputs, custom agent). +2. **One-time bind codes** — the browser Connect flow mints a short-lived `secrets.token_urlsafe(16)` code (600 s TTL, single-use) and surfaces it only in the initiating user's browser. The platform worker consumes `/connect ` (Telegram uses `/start ` over a deep link) before applying any `allowed_users` filter, so a not-yet-allowlisted user can complete their first bind. +3. **Strict ownership transfer** — the latest successful bind wins; `upsert_connection` revokes other owners' active rows for the same external identity. The DB-enforced partial unique index `uq_channel_connection_active_identity` (`WHERE status != 'revoked'`) makes the invariant race-free across concurrent writers. + +Connect codes are deliberately **bind-time defenses**, not chat-time defenses. After binding, ordinary `allowed_users` continue to gate regular messages exactly as before. + +## Connect-code Flow + +The browser initiates; the provider worker consumes the code; the manager never sees the code itself. + +```mermaid +sequenceDiagram + autonumber + participant Browser as Browser (Settings) + participant Gateway as Gateway
/api/channels/... + participant Store as SQL store
channel_oauth_states + participant Worker as Provider worker
(Telegram/Slack/...) + participant Repo as ChannelConnection repo
(upsert_connection) + + Browser->>Gateway: POST /api/channels/{provider}/connect + Gateway->>Store: insert code (token_urlsafe(16), TTL=600s, single-use) + Gateway-->>Browser: code + (Telegram: deep-link URL) + + Note over Browser,Worker: User sends /connect (or /start ) to the provider bot + + Worker->>Store: consume_oauth_state(code) + alt valid + unexpired + Store-->>Worker: ok (state consumed once) + Worker->>Repo: upsert_connection(provider, external_account_id, workspace_id, owner_user_id) + Repo-->>Worker: connection row (active) + Worker-->>Browser: success reply (via channel callback) + else invalid / expired / used + Store-->>Worker: reject + Worker-->>Browser: rejected (no reply in chat) + end + + Note over Repo: Partial unique index uq_channel_connection_active_identity
revokes prior owner's active row for the same identity +``` + +## Single-active-owner Transfer + +The partial unique index is the source of truth — application code never has to "revoke the previous owner" explicitly because the upsert that re-uses an identity fails on conflict and the loser retries against the now-visible revoked state. + +```mermaid +graph LR + classDef prior fill:#E5D2C4,stroke:#806A5B,color:#30251E + classDef new fill:#C9D7D2,stroke:#5D706A,color:#21302C + classDef db fill:#D7D3E8,stroke:#6B6680,color:#29263A + + Prior["Prior owner
connection_id=A
status=connected"]:::prior + New["New owner
connection_id=B"]:::new + Upsert["upsert_connection()
(owner_user_id=B)"]:::new + Idx["Partial unique index
uq_channel_connection_active_identity
WHERE status != 'revoked'"]:::db + + Prior -->|"loser: revoke"| Idx + New -->|"winner: insert"| Idx + Upsert -->|"trigger"| Idx + Idx -->|"returns"| Prior + Idx -.->|"retry against new state"| Upsert +``` + +After the dust settles: + +```mermaid +graph LR + classDef winner fill:#C9D7D2,stroke:#5D706A,color:#21302C + classDef loser fill:#D7D3E8,stroke:#6B6680,color:#29263A + + New["connection_id=B
owner=B
status=connected"]:::winner + Old["connection_id=A
owner=A
status=revoked"]:::loser + + New --- Old +``` + +The same invariant protects the `find_connection_by_external_identity` lookup used by `ChannelManager._get_bound_identity_rejection` — a non-revoked row can resolve to exactly one owner at any time. + +## Provider Message Flow Once Bound + +After a connection is bound, every inbound message walks the same path through `ChannelManager`. Slack/Discord (no streaming) and Feishu/Telegram (streaming) diverge only at the run boundary. + +```mermaid +sequenceDiagram + autonumber + participant Platform as Provider
(Slack/Telegram/...) + participant Worker as Provider worker + participant Bus as MessageBus
InboundMessage queue + participant Mgr as ChannelManager + participant Client as langgraph_sdk
async client + participant Gateway as Gateway
/api/* routers + + Platform->>Worker: inbound chat message
(resolved to connection_id + owner_user_id) + Worker->>Bus: publish_inbound(InboundMessage) + Bus->>Mgr: msg = get_inbound() + Mgr->>Mgr: _channel_storage_user_id(msg)
→ owner-bound user_id + Mgr->>Mgr: _get_bound_identity_rejection()
(re-check identity by provider+ext+ws) + Mgr->>Client: _get_or_create_thread(thread_id or new) + Client->>Gateway: threads.create(metadata={channel_source}) + Gateway-->>Client: thread_id + Mgr->>Mgr: receive_file(msg, thread_id, user_id=...)
(owner-bound bucket) + Mgr->>Mgr: _ingest_inbound_files(thread_id, user_id=...) + + alt channel supports streaming + Mgr->>Client: runs.stream(messages-tuple + values) + loop each chunk + Client-->>Mgr: delta / values snapshot + Mgr->>Bus: publish_outbound(is_final=False) + end + else Slack/Discord (no streaming) + Mgr->>Client: runs.wait() + Client-->>Mgr: final state + end + + Mgr->>Bus: publish_outbound(is_final=True) + Bus->>Worker: outbound callback + Worker->>Platform: post reply (Telegram editMessageText,
Feishu patch card, etc.) +``` + +## Sync vs Streaming Channels + +The two paths split on `ChannelRunPolicy.supports_streaming` (per-channel registration in `CHANNEL_CAPABILITIES`): + +```mermaid +graph TB + classDef sync fill:#E5D2C4,stroke:#806A5B,color:#30251E + classDef stream fill:#C9D7D2,stroke:#5D706A,color:#21302C + + Msg["InboundMessage
(channel, chat_id, text, files)"]:::sync + Sync1["Slack"]:::sync + Sync2["Discord"]:::sync + Sync3["DingTalk"]:::sync + Wait["runs.wait()
→ extract final AI text"]:::sync + Out1["publish_outbound(is_final=True)"]:::sync + + Stream1["Feishu"]:::stream + Stream2["Telegram"]:::stream + Stream3["WeCom (AI card)"]:::stream + Stream["runs.stream(messages-tuple + values)"]:::stream + Mid1["publish_outbound(is_final=False)
throttled"]:::stream + Mid2["Telegram: edit placeholder message
Feishu: patch running card
WeCom: PUT /v1.0/card/streaming"]:::stream + Final["publish_outbound(is_final=True)"]:::stream + + Msg --> Sync1 --> Wait --> Out1 + Msg --> Sync2 --> Wait --> Out1 + Msg --> Sync3 --> Wait --> Out1 + + Msg --> Stream1 --> Stream --> Mid1 --> Mid2 --> Final + Msg --> Stream2 --> Stream --> Mid1 --> Mid2 --> Final + Msg --> Stream3 --> Stream --> Mid1 --> Mid2 --> Final +``` + +For the special GitHub case (`fire_and_forget=True` channel policy), the manager calls `runs.create()` and returns once the run is `pending` — no outbound reply, because GitHub agents post via the `gh` CLI from inside their sandbox. See [GITHUB_AGENTS.md](GITHUB_AGENTS.md) for the full GitHub flow. + +## Owner-scoped File Storage + +`ChannelManager` resolves the storage owner **once** at the top of `_handle_chat` via `_channel_storage_user_id(msg)` and threads that value through the entire file pipeline. The same identity is used as the run `user_id` in `run_context` and as the bucket for memory, uploads, and outputs — so the bucket the agent reads/writes is always the bucket where channel files were staged. + +```mermaid +flowchart TB + classDef owner fill:#D8CFC4,stroke:#6E6259,color:#2F2A26 + classDef resolve fill:#C9D7D2,stroke:#5D706A,color:#21302C + classDef bucket fill:#D7D3E8,stroke:#6B6680,color:#29263A + classDef agent fill:#E5D2C4,stroke:#806A5B,color:#30251E + + Inbound["InboundMessage
connection_id, owner_user_id, workspace_id"]:::owner + Resolve["_channel_storage_user_id(msg)
sanitized + fall back to safe(msg.user_id)"]:::resolve + UserID["user_id = OWNER"]:::resolve + + RunID["run_context['user_id']
(run identity)"]:::agent + RunUploads["ensure_uploads_dir(thread_id, user_id=OWNER)"]:::bucket + Ingest["_ingest_inbound_files(user_id=OWNER)"]:::bucket + Receive["Channel.receive_file(msg, thread_id, user_id=OWNER)"]:::bucket + Resolved["_resolve_attachments(user_id=OWNER)"]:::bucket + Artifact["_prepare_artifact_delivery(user_id=OWNER)"]:::bucket + Memory["_resolve_memory_user_id
(make_safe_user_id match)"]:::bucket + + Bucket["backend/.deer-flow/users/OWNER/.../user-data/{uploads,outputs}"]:::bucket + + Inbound --> Resolve --> UserID + UserID --> RunID + UserID --> Receive + UserID --> Ingest + UserID --> RunUploads + UserID --> Resolved + UserID --> Artifact + UserID --> Memory + RunUploads --> Bucket + Ingest --> Bucket + Receive --> Bucket + Resolved --> Bucket + Artifact --> Bucket +``` + +The cached value is reused across the blocking (`runs.wait`) and streaming (`_handle_streaming_chat`) paths — even if a future `Channel.receive_file` returns a rewritten `InboundMessage`, uploads and artifact delivery still target the same bucket. + +## IM File Attachment Pipeline + +Inbound files (images, documents) walk through `Channel.receive_file` for materialization, then `_ingest_inbound_files` for owner-bound staging. The agent sees the staged path via the `` block injected into its context. + +```mermaid +sequenceDiagram + autonumber + participant IM as Provider message
(file attachment) + participant Worker as Provider worker + participant Mgr as ChannelManager + participant Ch as Channel impl
.receive_file + participant FS as Uploads directory
users/OWNER/.../uploads/ + participant Agent as Agent run + + IM->>Worker: message with file URL/bytes + Worker->>Mgr: InboundMessage(files=[...], connection_id, owner_user_id) + Mgr->>Mgr: storage_user_id = _channel_storage_user_id(msg) + Mgr->>Ch: receive_file(msg, thread_id, user_id=storage_user_id) + Note over Ch: provider-specific download
(WeCom: decrypt_file;
WeChat: read_bytes; others: HTTP GET) + Ch->>FS: write_upload_file_no_symlink(
uploads/OWNER/.../
, safe_name, data) + Ch-->>Mgr: msg with text rewritten to include + Mgr->>Mgr: _ingest_inbound_files(
thread_id, msg, user_id=storage_user_id) + Mgr->>Agent: HumanMessage with block
(paths under /mnt/user-data/uploads/) + Agent->>FS: read_file / view_image (sandbox) +``` + +## Cross-references + +- [AGENTS.md](../AGENTS.md) → "IM Channels System" — the index view in `backend/AGENTS.md` (configuration knobs, message flow, component list) +- [GITHUB_AGENTS.md](GITHUB_AGENTS.md) — webhook-driven GitHub channel, agent bindings, fan-out, token lifecycle +- `app/channels/manager.py` — dispatcher, `_channel_storage_user_id`, `_handle_chat`, `_handle_streaming_chat` +- `deerflow.persistence.channel_connections` — SQL tables (`channel_connections`, `channel_oauth_states`, `channel_conversations`, `channel_credentials`) and `upsert_connection` / `consume_oauth_state` / `find_connection_by_external_identity` + +--- + +# Configuration Configure the actual IM bots under the existing `channels` block: