feat(im): Add user-owned IM channel connections (#3487)

* Add user-owned IM channel connections * Fix dev startup and channel connect popup * Use async channel connect flow * Harden dev service daemon startup * Support local IM channel connections * Align IM connections with local channels * Fix safe user id digest algorithm * Address Copilot IM channel feedback * Address IM channel review comments * Support all integrated IM channel connections * Format additional channel connection tests * Keep unavailable channel connect buttons clickable * Fix IM channel provider icons * Add runtime setup for enabled IM channels * Guard global shortcut key handling * Keep configured IM channels editable * Avoid password autofill for channel secrets * Make channel threads visible to connection owners * Persist IM runtime config locally * Allow disconnecting runtime IM channels * Route no-auth channel sessions to local user * Use default user for auth-disabled local mode * Show IM channel source on threads * Prefill IM channel runtime config * Reflect IM channel runtime health * Ignore Feishu message read events * Ignore Feishu non-content message events * Let setup wizard enable IM channels * Fix frontend formatting after merge * Stabilize backend tests without local config * Isolate channel runtime config tests * Address channel connection review comments * Use sha256 user buckets with legacy migration * Ensure runtime IM channels are ready after restart * Persist disconnected IM channel state * Address channel connection review comments * Address channel connection review findings Frontend connect flow: - Open the runtime-config dialog only when a provider still needs credentials; configured providers go straight to the connect flow, so the binding-code/deep-link path is reachable from the UI again. - After saving credentials, continue into the connect flow when a user binding is still required (multi-user mode) instead of stopping at a "Connected" toast. - Extract shared provider-state helpers to core/channels/provider-state and add unit + e2e coverage for the direct-connect and configure-then-connect paths. Provider status semantics: - Report connection_status from the user's newest connection row; with no binding it is not_connected, except in auth-disabled local mode where a configured running channel is effectively connected. Concurrency and event-loop correctness: - Offload ChannelRuntimeConfigStore construction and writes, channel service construction, and Slack connection replies to threads; add a tests/blocking_io/ anchor for the runtime-config handlers. - Consume binding codes with a conditional UPDATE so a code can only be used once under concurrent workers; retry upsert_connection as an update when a concurrent insert wins the unique constraint. - Serialize ensure_channel_ready per channel so concurrent provider polls cannot double-start a channel worker. Config and migration hardening: - Stop mutating the get_app_config()-cached Telegram provider config; the runtime store now owns the UI-entered bot username. - Register channel_connections in STARTUP_ONLY_FIELDS with the standardized startup-only Field description. - Match the legacy unsafe-id bucket by recomputing its exact SHA-1 name so another user's same-prefix bucket can never be migrated. - Remove the unused Telegram process_webhook_update path and document src/core/channels in the frontend docs. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Address PR review comments on authz scoping and channel runtime Security (review feedback from ShenAC-SAC): - Scope internal-token callers to the connection owner carried in X-DeerFlow-Owner-User-Id instead of bypassing owner checks outright, in both require_permission(owner_check=True) and the stateless run endpoints. Internal callers keep access to their own and shared/legacy threads, and may claim a default-owned channel thread for its real owner, but a leaked internal token no longer grants cross-user thread access. - Require admin privileges for POST/DELETE /api/channels/{provider}/ runtime-config: runtime credentials and channel workers are instance-wide shared state (same model as the MCP config API). Read-only provider listing stays available to all users. Performance (review feedback from willem-bd): - Skip the redundant thread channel-metadata PATCH after the first successful backfill per thread. - Reuse the per-connection Slack WebClient until its token changes instead of constructing one per outbound message. - Reconcile channel readiness for all providers concurrently in GET /api/channels/providers. Also resolve the code-quality unused-import flag in the blocking-io anchor by pre-importing the channel service via importlib. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Fix prettier formatting in provider-state test Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * Reconcile UI runtime channel config with config reload on restart Main now reloads a channel's config.yaml entry on restart_channel() (#3514, issue #3497). Adapt the user-owned connection flow to coexist: - configure_channel() restarts with reload_config=False — the caller just supplied the authoritative config (browser-entered credentials that are never written to config.yaml), so a file reload must not clobber it with the stale on-disk entry. - _load_channel_config() re-applies the UI runtime-store overlay used at startup, so an operator-triggered restart keeps browser-entered credentials for channels without a config.yaml entry and does not resurrect a channel disconnected from the UI. - Offload the reload's disk IO (config.yaml + runtime store) with asyncio.to_thread, matching the blocking-IO policy on this branch. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-06-13 19:06:01 +00:00 · 2026-06-12 15:24:58 +08:00
parent b8f5ed360f
commit aa015462a7
96 changed files with 8585 additions and 277 deletions
@@ -11,6 +11,7 @@ from pydantic import BaseModel, ConfigDict, Field, field_validator

 from deerflow.config.acp_config import ACPAgentConfig, load_acp_config_from_dict
 from deerflow.config.agents_api_config import AgentsApiConfig, load_agents_api_config_from_dict
+from deerflow.config.channel_connections_config import ChannelConnectionsConfig
 from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
 from deerflow.config.database_config import DatabaseConfig
 from deerflow.config.extensions_config import ExtensionsConfig
@@ -116,6 +117,13 @@ class AppConfig(BaseModel):
    subagents: SubagentsAppConfig = Field(default_factory=SubagentsAppConfig, description="Subagent runtime configuration")
    guardrails: GuardrailsConfig = Field(default_factory=GuardrailsConfig, description="Guardrail middleware configuration")
    circuit_breaker: CircuitBreakerConfig = Field(default_factory=CircuitBreakerConfig, description="LLM circuit breaker configuration")
+    channel_connections: ChannelConnectionsConfig = Field(
+        default_factory=ChannelConnectionsConfig,
+        description=format_field_description(
+            "channel_connections",
+            field_doc="User-facing IM channel connection configuration.",
+        ),
+    )
    loop_detection: LoopDetectionConfig = Field(default_factory=LoopDetectionConfig, description="Loop detection middleware configuration")
    safety_finish_reason: SafetyFinishReasonConfig = Field(default_factory=SafetyFinishReasonConfig, description="Provider safety-filter finish_reason interception middleware configuration")
    model_config = ConfigDict(extra="allow")
@@ -0,0 +1,61 @@
+"""Configuration for user-owned IM channel connections."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class SlackChannelConnectionConfig(BaseModel):
+    enabled: bool = False
+
+    @property
+    def configured(self) -> bool:
+        return True
+
+
+class TelegramChannelConnectionConfig(BaseModel):
+    enabled: bool = False
+    bot_username: str = ""
+
+    @property
+    def configured(self) -> bool:
+        return bool(self.bot_username)
+
+
+class DiscordChannelConnectionConfig(BaseModel):
+    enabled: bool = False
+
+    @property
+    def configured(self) -> bool:
+        return True
+
+
+class BindingCodeChannelConnectionConfig(BaseModel):
+    enabled: bool = False
+
+    @property
+    def configured(self) -> bool:
+        return True
+
+
+class ChannelConnectionsConfig(BaseModel):
+    """Top-level config for browser-connectable IM channels."""
+
+    enabled: bool = False
+    slack: SlackChannelConnectionConfig = Field(default_factory=SlackChannelConnectionConfig)
+    telegram: TelegramChannelConnectionConfig = Field(default_factory=TelegramChannelConnectionConfig)
+    discord: DiscordChannelConnectionConfig = Field(default_factory=DiscordChannelConnectionConfig)
+    feishu: BindingCodeChannelConnectionConfig = Field(default_factory=BindingCodeChannelConnectionConfig)
+    dingtalk: BindingCodeChannelConnectionConfig = Field(default_factory=BindingCodeChannelConnectionConfig)
+    wechat: BindingCodeChannelConnectionConfig = Field(default_factory=BindingCodeChannelConnectionConfig)
+    wecom: BindingCodeChannelConnectionConfig = Field(default_factory=BindingCodeChannelConnectionConfig)
+
+    def provider_status(self, provider: str) -> dict[str, bool]:
+        config = getattr(self, provider, None)
+        if config is None:
+            return {"enabled": False, "configured": False}
+        enabled = bool(config.enabled)
+        return {
+            "enabled": enabled,
+            "configured": enabled and bool(config.configured),
+        }
@@ -1,4 +1,5 @@
 import hashlib
+import logging
 import os
 import re
 import shutil
@@ -14,6 +15,8 @@ _SAFE_USER_ID_RE = re.compile(r"^[A-Za-z0-9_\-]+$")
 _UNSAFE_USER_ID_CHAR_RE = re.compile(r"[^A-Za-z0-9_\-]")
 _SAFE_USER_ID_DIGEST_HEX_LEN = 16

+logger = logging.getLogger(__name__)
+

 def _default_local_base_dir() -> Path:
    """Return the caller project's writable DeerFlow state directory."""
@@ -47,7 +50,13 @@ def make_safe_user_id(raw: str) -> str:
    sanitized = _UNSAFE_USER_ID_CHAR_RE.sub("-", raw)
    if sanitized == raw:
        return raw
-    digest = hashlib.sha1(raw.encode("utf-8")).hexdigest()[:_SAFE_USER_ID_DIGEST_HEX_LEN]
+    digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:_SAFE_USER_ID_DIGEST_HEX_LEN]
+    return f"{sanitized}-{digest}"
+
+
+def _legacy_safe_user_id(raw: str, sanitized: str) -> str:
+    """Bucket name produced by the previous (SHA-1) digest revision for ``raw``."""
+    digest = hashlib.sha1(raw.encode("utf-8"), usedforsecurity=False).hexdigest()[:_SAFE_USER_ID_DIGEST_HEX_LEN]
    return f"{sanitized}-{digest}"


@@ -172,6 +181,32 @@ class Paths:
        """Directory for a specific user: `{base_dir}/users/{user_id}/`."""
        return self.base_dir / "users" / _validate_user_id(user_id)

+    def prepare_user_dir_for_raw_id(self, raw_user_id: str) -> str:
+        """Return the safe user ID and migrate this ID's legacy unsafe-id bucket.
+
+        A previous branch revision used SHA-1 for unsafe external user IDs.
+        New IDs use SHA-256; the legacy bucket name is recomputed from the same
+        raw ID, so only this user's own old bucket can ever be moved — a
+        different raw ID sharing the sanitized prefix produces a different
+        legacy digest and is never touched.
+        """
+        safe_user_id = make_safe_user_id(raw_user_id)
+        sanitized = _UNSAFE_USER_ID_CHAR_RE.sub("-", raw_user_id)
+        if safe_user_id == raw_user_id:
+            return safe_user_id
+
+        users_dir = self.base_dir / "users"
+        target_dir = users_dir / safe_user_id
+        legacy_dir = users_dir / _legacy_safe_user_id(raw_user_id, sanitized)
+        try:
+            if target_dir.exists() or not legacy_dir.is_dir():
+                return safe_user_id
+            legacy_dir.rename(target_dir)
+            logger.info("Migrated legacy unsafe-id user directory to the current digest format")
+        except OSError:
+            logger.exception("Failed to migrate legacy unsafe-id user directory")
+        return safe_user_id
+
    def user_memory_file(self, user_id: str) -> Path:
        """Per-user memory file: `{base_dir}/users/{user_id}/memory.json`."""
        return self.user_dir(user_id) / "memory.json"
@@ -56,6 +56,9 @@ STARTUP_ONLY_FIELDS: dict[str, str] = {
    # startup and the live channel clients are not rebuilt on
    # config.yaml edits.
    "channels": ("start_channel_service() is invoked once during startup; the live IM channel clients (Feishu, Slack, Telegram, DingTalk) are not rebuilt when channels.* changes."),
+    "channel_connections": (
+        "start_channel_service() wires the connection repository and channel workers once at startup, and the channel-connections router caches the merged provider config on app.state; channel_connections.* edits need a restart."
+    ),
 }


@@ -0,0 +1,21 @@
+"""User-owned IM channel connection persistence."""
+
+from deerflow.persistence.channel_connections.model import (
+    ChannelConnectionRow,
+    ChannelConversationRow,
+    ChannelCredentialRow,
+    ChannelOAuthStateRow,
+)
+from deerflow.persistence.channel_connections.sql import (
+    ChannelConnectionRepository,
+    ChannelCredentialCipher,
+)
+
+__all__ = [
+    "ChannelConnectionRepository",
+    "ChannelConnectionRow",
+    "ChannelConversationRow",
+    "ChannelCredentialCipher",
+    "ChannelCredentialRow",
+    "ChannelOAuthStateRow",
+]
@@ -0,0 +1,111 @@
+"""ORM models for user-owned IM channel connections."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import JSON, DateTime, ForeignKey, Index, Integer, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+
+
+def _utc_now() -> datetime:
+    return datetime.now(UTC)
+
+
+class ChannelConnectionRow(Base):
+    __tablename__ = "channel_connections"
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    owner_user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    provider: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
+    status: Mapped[str] = mapped_column(String(32), nullable=False, default="connected")
+
+    external_account_id: Mapped[str] = mapped_column(String(128), nullable=False, default="")
+    external_account_name: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    workspace_id: Mapped[str] = mapped_column(String(128), nullable=False, default="")
+    workspace_name: Mapped[str | None] = mapped_column(String(256), nullable=True)
+    bot_user_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+
+    scopes_json: Mapped[list] = mapped_column(JSON, default=list)
+    capabilities_json: Mapped[dict] = mapped_column(JSON, default=dict)
+    metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
+
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=_utc_now)
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=_utc_now, onupdate=_utc_now)
+    last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    last_error_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+
+    __table_args__ = (
+        UniqueConstraint(
+            "owner_user_id",
+            "provider",
+            "external_account_id",
+            "workspace_id",
+            name="uq_channel_connection_owner_provider_identity",
+        ),
+        Index("idx_channel_connections_event_lookup", "provider", "workspace_id", "bot_user_id"),
+    )
+
+
+class ChannelCredentialRow(Base):
+    __tablename__ = "channel_credentials"
+
+    connection_id: Mapped[str] = mapped_column(
+        String(64),
+        ForeignKey("channel_connections.id", ondelete="CASCADE"),
+        primary_key=True,
+    )
+    encrypted_access_token: Mapped[str | None] = mapped_column(Text, nullable=True)
+    encrypted_refresh_token: Mapped[str | None] = mapped_column(Text, nullable=True)
+    token_type: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    refresh_expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    encrypted_extra_json: Mapped[str | None] = mapped_column(Text, nullable=True)
+    version: Mapped[int] = mapped_column(Integer, nullable=False, default=1)
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=_utc_now, onupdate=_utc_now)
+
+
+class ChannelOAuthStateRow(Base):
+    __tablename__ = "channel_oauth_states"
+
+    state_hash: Mapped[str] = mapped_column(String(128), primary_key=True)
+    owner_user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    provider: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
+    code_verifier_encrypted: Mapped[str | None] = mapped_column(Text, nullable=True)
+    nonce_hash: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    redirect_after: Mapped[str | None] = mapped_column(Text, nullable=True)
+    requested_scopes_json: Mapped[list] = mapped_column(JSON, default=list)
+    metadata_json: Mapped[dict] = mapped_column(JSON, default=dict)
+    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    consumed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=_utc_now)
+
+
+class ChannelConversationRow(Base):
+    __tablename__ = "channel_conversations"
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    connection_id: Mapped[str] = mapped_column(
+        String(64),
+        ForeignKey("channel_connections.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    owner_user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    provider: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
+    external_conversation_id: Mapped[str] = mapped_column(String(128), nullable=False)
+    external_topic_id: Mapped[str] = mapped_column(String(128), nullable=False, default="")
+    thread_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=_utc_now)
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=_utc_now, onupdate=_utc_now)
+
+    __table_args__ = (
+        UniqueConstraint(
+            "connection_id",
+            "external_conversation_id",
+            "external_topic_id",
+            name="uq_channel_conversation_connection_external",
+        ),
+    )
@@ -0,0 +1,387 @@
+"""SQL repository for user-owned IM channel connections."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import json
+import logging
+import uuid
+from datetime import UTC, datetime
+from typing import Any
+
+from cryptography.fernet import Fernet, InvalidToken
+from sqlalchemy import delete, func, select, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.channel_connections.model import (
+    ChannelConnectionRow,
+    ChannelConversationRow,
+    ChannelCredentialRow,
+    ChannelOAuthStateRow,
+)
+from deerflow.utils.time import coerce_iso
+
+logger = logging.getLogger(__name__)
+
+
+class ChannelCredentialCipher:
+    """Encrypts provider credentials before they are persisted."""
+
+    def __init__(self, fernet: Fernet) -> None:
+        self._fernet = fernet
+
+    @classmethod
+    def from_key(cls, key: str) -> ChannelCredentialCipher:
+        digest = hashlib.sha256(key.encode("utf-8")).digest()
+        return cls(Fernet(base64.urlsafe_b64encode(digest)))
+
+    def encrypt_text(self, value: str | None) -> str | None:
+        if value is None:
+            return None
+        return "fernet:v1:" + self._fernet.encrypt(value.encode("utf-8")).decode("ascii")
+
+    def decrypt_text(self, value: str | None) -> str | None:
+        if value is None:
+            return None
+        token = value.removeprefix("fernet:v1:")
+        return self._fernet.decrypt(token.encode("ascii")).decode("utf-8")
+
+
+class ChannelConnectionRepository:
+    """Persistence facade for channel connections, credentials, and conversations."""
+
+    def __init__(
+        self,
+        session_factory: async_sessionmaker[AsyncSession],
+        *,
+        cipher: ChannelCredentialCipher | None = None,
+    ) -> None:
+        self.session_factory = session_factory
+        self._cipher = cipher
+
+    async def close(self) -> None:
+        from deerflow.persistence.engine import close_engine
+
+        await close_engine()
+
+    @staticmethod
+    def _new_id() -> str:
+        return uuid.uuid4().hex
+
+    @staticmethod
+    def _normalize_optional_identity(value: str | None) -> str:
+        return value or ""
+
+    @staticmethod
+    def _coerce_datetime(value: datetime | None) -> datetime | None:
+        if value is None or value.tzinfo is not None:
+            return value
+        return value.replace(tzinfo=UTC)
+
+    def _encrypt_optional_secret(self, value: str | None) -> str | None:
+        if value is None:
+            return None
+        if self._cipher is None:
+            raise RuntimeError("channel connection encryption key is required")
+        return self._cipher.encrypt_text(value)
+
+    @staticmethod
+    def _connection_to_dict(row: ChannelConnectionRow) -> dict[str, Any]:
+        data = row.to_dict()
+        data["external_account_id"] = data["external_account_id"] or None
+        data["workspace_id"] = data["workspace_id"] or None
+        data["scopes"] = data.pop("scopes_json") or []
+        data["capabilities"] = data.pop("capabilities_json") or {}
+        data["metadata"] = data.pop("metadata_json") or {}
+        for key in ("created_at", "updated_at", "last_seen_at", "last_error_at"):
+            value = data.get(key)
+            if isinstance(value, datetime):
+                data[key] = coerce_iso(value)
+        return data
+
+    async def upsert_connection(
+        self,
+        *,
+        owner_user_id: str,
+        provider: str,
+        external_account_id: str | None = None,
+        external_account_name: str | None = None,
+        workspace_id: str | None = None,
+        workspace_name: str | None = None,
+        bot_user_id: str | None = None,
+        scopes: list[str] | None = None,
+        capabilities: dict[str, Any] | None = None,
+        metadata: dict[str, Any] | None = None,
+        status: str = "connected",
+    ) -> dict[str, Any]:
+        external_account_id_value = self._normalize_optional_identity(external_account_id)
+        workspace_id_value = self._normalize_optional_identity(workspace_id)
+
+        def _apply(row: ChannelConnectionRow) -> None:
+            row.status = status
+            row.external_account_name = external_account_name
+            row.workspace_name = workspace_name
+            row.bot_user_id = bot_user_id
+            row.scopes_json = list(scopes or [])
+            row.capabilities_json = dict(capabilities or {})
+            row.metadata_json = dict(metadata or {})
+
+        stmt = select(ChannelConnectionRow).where(
+            ChannelConnectionRow.owner_user_id == owner_user_id,
+            ChannelConnectionRow.provider == provider,
+            ChannelConnectionRow.external_account_id == external_account_id_value,
+            ChannelConnectionRow.workspace_id == workspace_id_value,
+        )
+        async with self.session_factory() as session:
+            row = (await session.execute(stmt)).scalar_one_or_none()
+            if row is None:
+                row = ChannelConnectionRow(
+                    id=self._new_id(),
+                    owner_user_id=owner_user_id,
+                    provider=provider,
+                    external_account_id=external_account_id_value,
+                    workspace_id=workspace_id_value,
+                )
+                session.add(row)
+
+            _apply(row)
+            try:
+                await session.commit()
+            except IntegrityError:
+                # A concurrent writer inserted the same identity first; retry as
+                # an update of that row.
+                await session.rollback()
+                row = (await session.execute(stmt)).scalar_one()
+                _apply(row)
+                await session.commit()
+            await session.refresh(row)
+            return self._connection_to_dict(row)
+
+    async def list_connections(self, owner_user_id: str) -> list[dict[str, Any]]:
+        async with self.session_factory() as session:
+            result = await session.execute(select(ChannelConnectionRow).where(ChannelConnectionRow.owner_user_id == owner_user_id).order_by(ChannelConnectionRow.updated_at.desc(), ChannelConnectionRow.id.desc()))
+            return [self._connection_to_dict(row) for row in result.scalars()]
+
+    async def disconnect_connection(self, *, connection_id: str, owner_user_id: str) -> bool:
+        async with self.session_factory() as session:
+            row = await session.get(ChannelConnectionRow, connection_id)
+            if row is None or row.owner_user_id != owner_user_id:
+                return False
+
+            row.status = "revoked"
+            credential = await session.get(ChannelCredentialRow, connection_id)
+            if credential is not None:
+                await session.delete(credential)
+            await session.commit()
+            return True
+
+    async def store_credentials(
+        self,
+        connection_id: str,
+        *,
+        access_token: str | None,
+        refresh_token: str | None = None,
+        token_type: str | None = None,
+        expires_at: datetime | None = None,
+        refresh_expires_at: datetime | None = None,
+        extra: dict[str, Any] | None = None,
+    ) -> None:
+        if self._cipher is None:
+            raise RuntimeError("channel connection encryption key is required")
+        async with self.session_factory() as session:
+            row = await session.get(ChannelCredentialRow, connection_id)
+            if row is None:
+                row = ChannelCredentialRow(connection_id=connection_id)
+                session.add(row)
+            row.encrypted_access_token = self._cipher.encrypt_text(access_token)
+            row.encrypted_refresh_token = self._cipher.encrypt_text(refresh_token)
+            row.token_type = token_type
+            row.expires_at = expires_at
+            row.refresh_expires_at = refresh_expires_at
+            row.encrypted_extra_json = self._cipher.encrypt_text(json.dumps(extra or {}, ensure_ascii=False))
+            row.version = (row.version or 0) + 1
+            await session.commit()
+
+    async def get_credentials(self, connection_id: str) -> dict[str, Any] | None:
+        if self._cipher is None:
+            return None
+        async with self.session_factory() as session:
+            row = await session.get(ChannelCredentialRow, connection_id)
+            if row is None:
+                return None
+            try:
+                extra_raw = self._cipher.decrypt_text(row.encrypted_extra_json)
+                return {
+                    "connection_id": row.connection_id,
+                    "access_token": self._cipher.decrypt_text(row.encrypted_access_token),
+                    "refresh_token": self._cipher.decrypt_text(row.encrypted_refresh_token),
+                    "token_type": row.token_type,
+                    "expires_at": self._coerce_datetime(row.expires_at),
+                    "refresh_expires_at": self._coerce_datetime(row.refresh_expires_at),
+                    "extra": json.loads(extra_raw) if extra_raw else {},
+                }
+            except (InvalidToken, UnicodeError, json.JSONDecodeError):
+                logger.warning(
+                    "Unable to decrypt channel connection credentials; treating credentials as unavailable",
+                    exc_info=True,
+                )
+                return None
+
+    @staticmethod
+    def hash_state(state: str) -> str:
+        return hashlib.sha256(state.encode("utf-8")).hexdigest()
+
+    async def create_oauth_state(
+        self,
+        *,
+        owner_user_id: str,
+        provider: str,
+        state: str,
+        expires_at: datetime,
+        code_verifier: str | None = None,
+        nonce_hash: str | None = None,
+        redirect_after: str | None = None,
+        requested_scopes: list[str] | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> None:
+        row = ChannelOAuthStateRow(
+            state_hash=self.hash_state(state),
+            owner_user_id=owner_user_id,
+            provider=provider,
+            code_verifier_encrypted=self._encrypt_optional_secret(code_verifier),
+            nonce_hash=nonce_hash,
+            redirect_after=redirect_after,
+            requested_scopes_json=list(requested_scopes or []),
+            metadata_json=dict(metadata or {}),
+            expires_at=expires_at,
+        )
+        async with self.session_factory() as session:
+            session.add(row)
+            await session.commit()
+
+    async def count_oauth_states(self, *, owner_user_id: str, provider: str) -> int:
+        async with self.session_factory() as session:
+            result = await session.execute(
+                select(func.count())
+                .select_from(ChannelOAuthStateRow)
+                .where(
+                    ChannelOAuthStateRow.owner_user_id == owner_user_id,
+                    ChannelOAuthStateRow.provider == provider,
+                )
+            )
+            return int(result.scalar_one())
+
+    async def consume_oauth_state(
+        self,
+        *,
+        provider: str,
+        state: str,
+        now: datetime | None = None,
+    ) -> dict[str, Any] | None:
+        current_time = now or datetime.now(UTC)
+        state_hash = self.hash_state(state)
+        async with self.session_factory() as session:
+            await session.execute(delete(ChannelOAuthStateRow).where(ChannelOAuthStateRow.expires_at < current_time))
+            row = await session.get(ChannelOAuthStateRow, state_hash)
+            if row is None or row.provider != provider or row.consumed_at is not None:
+                await session.commit()
+                return None
+            expires_at = self._coerce_datetime(row.expires_at)
+            if expires_at is not None and expires_at < current_time:
+                await session.commit()
+                return None
+
+            # Conditional UPDATE so two concurrent workers cannot both consume
+            # the same binding code: only the writer that flips consumed_at
+            # from NULL wins.
+            result = await session.execute(
+                update(ChannelOAuthStateRow)
+                .where(
+                    ChannelOAuthStateRow.state_hash == state_hash,
+                    ChannelOAuthStateRow.consumed_at.is_(None),
+                )
+                .values(consumed_at=current_time)
+            )
+            await session.commit()
+            if result.rowcount != 1:
+                return None
+            return {
+                "owner_user_id": row.owner_user_id,
+                "provider": row.provider,
+                "requested_scopes": row.requested_scopes_json or [],
+                "metadata": row.metadata_json or {},
+                "redirect_after": row.redirect_after,
+            }
+
+    async def find_connection_by_external_identity(
+        self,
+        *,
+        provider: str,
+        external_account_id: str,
+        workspace_id: str | None = None,
+    ) -> dict[str, Any] | None:
+        async with self.session_factory() as session:
+            result = await session.execute(
+                select(ChannelConnectionRow)
+                .where(
+                    ChannelConnectionRow.provider == provider,
+                    ChannelConnectionRow.external_account_id == self._normalize_optional_identity(external_account_id),
+                    ChannelConnectionRow.workspace_id == self._normalize_optional_identity(workspace_id),
+                    ChannelConnectionRow.status == "connected",
+                )
+                .order_by(ChannelConnectionRow.updated_at.desc(), ChannelConnectionRow.id.desc())
+                .limit(1)
+            )
+            row = result.scalar_one_or_none()
+            return self._connection_to_dict(row) if row is not None else None
+
+    async def set_thread_id(
+        self,
+        *,
+        connection_id: str,
+        owner_user_id: str,
+        provider: str,
+        external_conversation_id: str,
+        thread_id: str,
+        external_topic_id: str | None = None,
+    ) -> None:
+        topic_id = external_topic_id or ""
+        async with self.session_factory() as session:
+            stmt = select(ChannelConversationRow).where(
+                ChannelConversationRow.connection_id == connection_id,
+                ChannelConversationRow.external_conversation_id == external_conversation_id,
+                ChannelConversationRow.external_topic_id == topic_id,
+            )
+            row = (await session.execute(stmt)).scalar_one_or_none()
+            if row is None:
+                row = ChannelConversationRow(
+                    id=self._new_id(),
+                    connection_id=connection_id,
+                    owner_user_id=owner_user_id,
+                    provider=provider,
+                    external_conversation_id=external_conversation_id,
+                    external_topic_id=topic_id,
+                    thread_id=thread_id,
+                )
+                session.add(row)
+            else:
+                row.thread_id = thread_id
+                row.owner_user_id = owner_user_id
+                row.provider = provider
+            await session.commit()
+
+    async def get_thread_id(
+        self,
+        connection_id: str,
+        external_conversation_id: str,
+        external_topic_id: str | None = None,
+    ) -> str | None:
+        async with self.session_factory() as session:
+            stmt = select(ChannelConversationRow.thread_id).where(
+                ChannelConversationRow.connection_id == connection_id,
+                ChannelConversationRow.external_conversation_id == external_conversation_id,
+                ChannelConversationRow.external_topic_id == (external_topic_id or ""),
+            )
+            return (await session.execute(stmt)).scalar_one_or_none()
@@ -14,10 +14,26 @@ its storage implementation lives in ``deerflow.runtime.events.store.db`` and
 there is no matching entity directory.
 """

+from deerflow.persistence.channel_connections.model import (
+    ChannelConnectionRow,
+    ChannelConversationRow,
+    ChannelCredentialRow,
+    ChannelOAuthStateRow,
+)
 from deerflow.persistence.feedback.model import FeedbackRow
 from deerflow.persistence.models.run_event import RunEventRow
 from deerflow.persistence.run.model import RunRow
 from deerflow.persistence.thread_meta.model import ThreadMetaRow
 from deerflow.persistence.user.model import UserRow

-__all__ = ["FeedbackRow", "RunEventRow", "RunRow", "ThreadMetaRow", "UserRow"]
+__all__ = [
+    "ChannelConnectionRow",
+    "ChannelConversationRow",
+    "ChannelCredentialRow",
+    "ChannelOAuthStateRow",
+    "FeedbackRow",
+    "RunEventRow",
+    "RunRow",
+    "ThreadMetaRow",
+    "UserRow",
+]
@@ -71,6 +71,15 @@ class ThreadMetaStore(abc.ABC):
        """
        pass

+    @abc.abstractmethod
+    async def update_owner(self, thread_id: str, owner_user_id: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None:
+        """Move a thread metadata row to a new owner.
+
+        Intended for trusted internal repair/migration paths. No-op if the
+        row does not exist or the caller fails the owner check.
+        """
+        pass
+
    @abc.abstractmethod
    async def check_access(self, thread_id: str, user_id: str, *, require_existing: bool = False) -> bool:
        """Check if ``user_id`` has access to ``thread_id``."""
@@ -127,6 +127,14 @@ class MemoryThreadMetaStore(ThreadMetaStore):
        record["updated_at"] = now_iso()
        await self._store.aput(THREADS_NS, thread_id, record)

+    async def update_owner(self, thread_id: str, owner_user_id: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None:
+        record = await self._get_owned_record(thread_id, user_id, "MemoryThreadMetaStore.update_owner")
+        if record is None:
+            return
+        record["user_id"] = owner_user_id
+        record["updated_at"] = now_iso()
+        await self._store.aput(THREADS_NS, thread_id, record)
+
    async def delete(self, thread_id: str, *, user_id: str | None | _AutoSentinel = AUTO) -> None:
        record = await self._get_owned_record(thread_id, user_id, "MemoryThreadMetaStore.delete")
        if record is None:
@@ -211,6 +211,21 @@ class ThreadMetaRepository(ThreadMetaStore):
            row.updated_at = datetime.now(UTC)
            await session.commit()

+    async def update_owner(
+        self,
+        thread_id: str,
+        owner_user_id: str,
+        *,
+        user_id: str | None | _AutoSentinel = AUTO,
+    ) -> None:
+        """Move a thread metadata row to ``owner_user_id``."""
+        resolved_user_id = resolve_user_id(user_id, method_name="ThreadMetaRepository.update_owner")
+        async with self._sf() as session:
+            if not await self._check_ownership(session, thread_id, resolved_user_id):
+                return
+            await session.execute(update(ThreadMetaRow).where(ThreadMetaRow.thread_id == thread_id).values(user_id=owner_user_id, updated_at=datetime.now(UTC)))
+            await session.commit()
+
    async def delete(
        self,
        thread_id: str,
@@ -83,6 +83,7 @@ class RunRecord:
    multitask_strategy: str = "reject"
    metadata: dict = field(default_factory=dict)
    kwargs: dict = field(default_factory=dict)
+    user_id: str | None = None
    created_at: str = ""
    updated_at: str = ""
    task: asyncio.Task | None = field(default=None, repr=False)
@@ -124,7 +125,7 @@ class RunManager:

    @staticmethod
    def _store_put_payload(record: RunRecord, *, error: str | None = None) -> dict[str, Any]:
-        return {
+        payload = {
            "thread_id": record.thread_id,
            "assistant_id": record.assistant_id,
            "status": record.status.value,
@@ -135,6 +136,9 @@ class RunManager:
            "created_at": record.created_at,
            "model_name": record.model_name,
        }
+        if record.user_id is not None:
+            payload["user_id"] = record.user_id
+        return payload

    async def _call_store_with_retry(
        self,
@@ -241,6 +245,7 @@ class RunManager:
            kwargs=row.get("kwargs") or {},
            created_at=row.get("created_at") or "",
            updated_at=row.get("updated_at") or "",
+            user_id=row.get("user_id"),
            error=row.get("error"),
            model_name=row.get("model_name"),
            store_only=True,
@@ -320,6 +325,7 @@ class RunManager:
        metadata: dict | None = None,
        kwargs: dict | None = None,
        multitask_strategy: str = "reject",
+        user_id: str | None = None,
    ) -> RunRecord:
        """Create a new pending run and register it."""
        run_id = str(uuid.uuid4())
@@ -333,6 +339,7 @@ class RunManager:
            multitask_strategy=multitask_strategy,
            metadata=metadata or {},
            kwargs=kwargs or {},
+            user_id=user_id,
            created_at=now,
            updated_at=now,
        )
@@ -504,6 +511,7 @@ class RunManager:
        kwargs: dict | None = None,
        multitask_strategy: str = "reject",
        model_name: str | None = None,
+        user_id: str | None = None,
    ) -> RunRecord:
        """Atomically check for inflight runs and create a new one.

@@ -546,6 +554,7 @@ class RunManager:
                multitask_strategy=multitask_strategy,
                metadata=metadata or {},
                kwargs=kwargs or {},
+                user_id=user_id,
                created_at=now,
                updated_at=now,
                model_name=model_name,