mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-21 02:05:45 +00:00
docs(backend): add IM channel connections and GitHub agents architecture diagrams (#4112)
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/<start), single-active-owner transfer, provider message routing, owner-scoped file storage. - backend/docs/GITHUB_AGENTS.md — sequence + graph diagrams for GitHub event-driven agents: webhook fan-out, preferred_thread_id = UUID5, GH token lifecycle, race recovery. Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
黄云龙
+5
-9
@@ -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 <code>` (or Telegram `/start <code>`) **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/`)
|
||||
|
||||
@@ -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<br/>channels.github:<br/> enabled: true<br/> default_mention_login"]:::operator
|
||||
AgentYaml["agents/{name}/config.yaml<br/>github:<br/> installation_id<br/> bot_login<br/> bindings: [{repo, triggers}]"]:::agent
|
||||
Registry["build_github_agent_registry()<br/>(mtime-cached, asyncio.to_thread)"]:::route
|
||||
Webhook["POST /api/webhooks/github<br/>(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<br/>(github_webhooks.py)
|
||||
participant Disp as fanout_event()<br/>(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<br/>/api/webhooks/github
|
||||
|
||||
GH->>Router: delivery (event, delivery_id, payload,<br/>X-Hub-Signature-256, X-GitHub-Event)
|
||||
Router->>Router: _verify_signature()<br/>hmac.compare_digest(sha256, secret)
|
||||
Router->>Disp: fanout_event(bus, event, delivery_id, payload,<br/>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()<br/>UUID5(repo, number, agent)
|
||||
Disp->>Bus: publish_inbound(InboundMessage(<br/>channel=github, chat_id=repo,<br/>topic_id="{number}:{agent}",<br/>owner_user_id=match.user_id,<br/>metadata.agent_name=agent,<br/>metadata.preferred_thread_id=...,<br/>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.<br/>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<br/>owner/name"]:::input
|
||||
Number["issue/PR number<br/>int"]:::input
|
||||
Agent["agent_name<br/>[A-Za-z0-9-]+"]:::input
|
||||
Seed["seed = '{repo}#{number}:{agent}'"]:::hash
|
||||
UUID5["uuid.uuid5(<br/> GITHUB_THREAD_NAMESPACE,<br/> seed)"]:::hash
|
||||
Thread["thread_id<br/>(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<br/>(per-event override)"]:::step
|
||||
B["2. github.bot_login<br/>(agent's App identity)"]:::step
|
||||
C["3. channels.github.default_mention_login<br/>(operator-wide default)"]:::step
|
||||
D["4. agent.name<br/>(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<br/>_handle_chat_on_thread
|
||||
participant Pol as inject_github_credentials()<br/>(github/run_policy.py)
|
||||
participant Auth as mint_installation_token<br/>(github/app_auth.py)
|
||||
participant GH as GitHub API<br/>(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<br/>(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<br/>(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(<br/>thread_id, assistant_id,<br/>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<br/>cmd="git push https://x-access-token:$GH_TOKEN@..."
|
||||
Bash->>Sandbox: execute_command(<br/>env={"GH_TOKEN": "...",<br/> "GITHUB_TOKEN": "..."})
|
||||
Note over Sandbox: AioSandbox: bash.exec(env=...) on fresh session<br/>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<br/>(delivery 1)
|
||||
participant Mgr2 as ChannelManager<br/>(delivery 2)
|
||||
participant Client as langgraph_sdk client
|
||||
participant Gateway as Gateway<br/>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<br/>(in sandbox)"]:::gh
|
||||
Issue["GitHub issue / PR"]:::gh
|
||||
Channel["GitHubChannel.send()<br/>(log-only)"]:::send
|
||||
Log["gateway.log<br/>(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)
|
||||
@@ -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 <code>` (Telegram uses `/start <code>` 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<br/>/api/channels/...
|
||||
participant Store as SQL store<br/>channel_oauth_states
|
||||
participant Worker as Provider worker<br/>(Telegram/Slack/...)
|
||||
participant Repo as ChannelConnection repo<br/>(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 <code> (or /start <code>) 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<br/>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<br/>connection_id=A<br/>status=connected"]:::prior
|
||||
New["New owner<br/>connection_id=B"]:::new
|
||||
Upsert["upsert_connection()<br/>(owner_user_id=B)"]:::new
|
||||
Idx["Partial unique index<br/>uq_channel_connection_active_identity<br/>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<br/>owner=B<br/>status=connected"]:::winner
|
||||
Old["connection_id=A<br/>owner=A<br/>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<br/>(Slack/Telegram/...)
|
||||
participant Worker as Provider worker
|
||||
participant Bus as MessageBus<br/>InboundMessage queue
|
||||
participant Mgr as ChannelManager
|
||||
participant Client as langgraph_sdk<br/>async client
|
||||
participant Gateway as Gateway<br/>/api/* routers
|
||||
|
||||
Platform->>Worker: inbound chat message<br/>(resolved to connection_id + owner_user_id)
|
||||
Worker->>Bus: publish_inbound(InboundMessage)
|
||||
Bus->>Mgr: msg = get_inbound()
|
||||
Mgr->>Mgr: _channel_storage_user_id(msg)<br/>→ owner-bound user_id
|
||||
Mgr->>Mgr: _get_bound_identity_rejection()<br/>(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=...)<br/>(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,<br/>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<br/>(channel, chat_id, text, files)"]:::sync
|
||||
Sync1["Slack"]:::sync
|
||||
Sync2["Discord"]:::sync
|
||||
Sync3["DingTalk"]:::sync
|
||||
Wait["runs.wait()<br/>→ 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)<br/>throttled"]:::stream
|
||||
Mid2["Telegram: edit placeholder message<br/>Feishu: patch running card<br/>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<br/>connection_id, owner_user_id, workspace_id"]:::owner
|
||||
Resolve["_channel_storage_user_id(msg)<br/>sanitized + fall back to safe(msg.user_id)"]:::resolve
|
||||
UserID["user_id = OWNER"]:::resolve
|
||||
|
||||
RunID["run_context['user_id']<br/>(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<br/>(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 `<uploaded_files>` block injected into its context.
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
autonumber
|
||||
participant IM as Provider message<br/>(file attachment)
|
||||
participant Worker as Provider worker
|
||||
participant Mgr as ChannelManager
|
||||
participant Ch as Channel impl<br/>.receive_file
|
||||
participant FS as Uploads directory<br/>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<br/>(WeCom: decrypt_file;<br/>WeChat: read_bytes; others: HTTP GET)
|
||||
Ch->>FS: write_upload_file_no_symlink(<br/>uploads/OWNER/.../<br/>, safe_name, data)
|
||||
Ch-->>Mgr: msg with text rewritten to include <uploaded_files>
|
||||
Mgr->>Mgr: _ingest_inbound_files(<br/>thread_id, msg, user_id=storage_user_id)
|
||||
Mgr->>Agent: HumanMessage with <uploaded_files> block<br/>(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:
|
||||
|
||||
|
||||
Reference in New Issue
Block a user