Nan Gao 085924a3d0
fix(channels): run each IM message once across instances, restarts and late redeliveries (#7104)
* fix(channels): run each IM message once across instances, restarts and late redeliveries

Inbound IM messages were deduplicated only by the 10-minute inbound dedupe
store (#7103). Discord direct messages had no key at all, a Discord thread
reply got an instance-dependent key, a Discord thread was created before
dedupe ran, a redelivery past the window started a second run, and two
instances racing on a conversation's first messages bound it to two
threads.

- Discord messages are keyed (discord, guild_id or "dm", posting channel,
  message_id); the posting channel is the same on every instance.
- MessageBus.claim_inbound lets an adapter record the manager's dedupe key
  before a side effect; Discord claims it on the Gateway loop before
  _create_thread, marks the published message dedupe_claimed, and gives the
  claim back when the message is never committed.
- runs.wait / runs.stream / runs.create send an Idempotency-Key derived from
  the dedupe key, and keyed human message ids are uuid5 of it (a uuid4 per
  attempt made every keyed retry a 409). A reused run's status-only /wait or
  stream gap is resolved to this message's own turn via threads.get_state;
  409s about the message's own keyed run keep the key instead of reading as
  a busy thread; an ended reused GitHub run gets no follow-up watcher.
- A conversation's first thread binding is claimed insert-if-absent
  (claim_thread_id on channel_conversations and channel_thread_bindings);
  the loser adopts the winner and deletes the empty thread it minted.
  /new and /agent use still rebind.

Docs: IM_CHANNEL_CONNECTIONS.md, CONFIGURATION.md, channels AGENTS.md,
CHANGELOG (en/zh).

* fix(channels): give back Discord dedupe claims for messages never processed

Review round 1 on #7104. A Discord claim records the manager's dedupe key
before the message is queued, so every path that drops the message unrun
must release it, or a peer refuses the redelivery for the rest of the TTL:

- The hand-off task releases the claim on cancellation too (stop() drains it
  while it resolves the connection identity), via await_drained so the
  cancellation still propagates.
- ChannelManager._discard_pending_inbound releases the key of a claimed
  message that shutdown drops from the queue, after the grace period and on
  a cancelled shutdown.
- The adapter tracks claims until they are committed. A release refused
  because stop() already closed cross-loop intake, or a hand-off cancelled
  before it ran, leaves the claim tracked, and stop() releases it last.

* fix(gateway): keep IM run idempotency stable when /new rebinds the conversation

Review round 2 on #7104. The Gateway scoped every Idempotency-Key by owner
and thread. /new rebinds an IM conversation to a fresh thread, so a late
redelivery of a message admitted on the old thread was requested on the new
one, got a different scoped key and started a second run.

- Keys from trusted internal callers (the IM channel manager) are scoped by
  owner only (`internal-run:` prefix); browser and API keys keep the
  owner-and-thread scope (`http-run:`).
- A key that resolves to a run of another thread or user raises
  RunIdempotencyKeyMismatch, a RuntimeError subclass so in-process launchers
  see the same type, and start_run answers it with 409 "Idempotency-Key
  already used with a different request" instead of a 500. The channel
  manager already treats that 409 as the message's admitted run.

Docs: API.md, IM_CHANNEL_CONNECTIONS.md, gateway AGENTS.md (reworded to stay
under its budget), CHANGELOG (en/zh).

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-10-19 22:57:19 +08:00

4064 lines
190 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""ChannelManager — consumes inbound messages and dispatches them to the DeerFlow agent via Gateway."""
from __future__ import annotations
import asyncio
import hashlib
import json
import logging
import math
import mimetypes
import re
import stat
import time
import uuid
from collections import OrderedDict
from collections.abc import Awaitable, Callable, Iterator, Mapping
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass
from html import escape
from pathlib import Path
from typing import Any, NamedTuple
from urllib.parse import quote, urlparse
import httpx
from fastapi import HTTPException
from langgraph_sdk.errors import ConflictError
from app.channels import buzz_run_policy as _buzz_run_policy # noqa: F401
from app.channels import feishu_run_policy as _feishu_run_policy # noqa: F401
from app.channels.capabilities import CHANNEL_CAPABILITIES
from app.channels.commands import KNOWN_CHANNEL_COMMANDS
from app.channels.connection_identity import lookup_thread_id
from app.channels.dedupe_store import InboundDedupeStore, MemoryInboundDedupeStore
from app.channels.message_bus import (
HUMAN_INPUT_REQUEST_METADATA_KEY,
HUMAN_INPUT_RESPONSE_METADATA_KEY,
INBOUND_FILE_CONTENT_KEY,
PENDING_CLARIFICATION_METADATA_KEY,
InboundMessage,
InboundMessageType,
MessageBus,
OutboundMessage,
ResolvedAttachment,
cleanup_inbound_files,
)
from app.channels.run_policy import CHANNEL_RUN_POLICY, ChannelRunPolicy
from app.channels.store import ChannelStore
from app.gateway.csrf_middleware import CSRF_COOKIE_NAME, CSRF_HEADER_NAME, generate_csrf_token
# Import built-in channel run-policy registrars eagerly so direct
# ChannelManager construction sees the same policy map as gateway bootstrap.
from app.gateway.github import run_policy as _github_run_policy # noqa: F401
from app.gateway.internal_auth import create_internal_auth_headers
from app.gateway.path_utils import resolve_outputs_confined_path
from deerflow.agents.human_input import read_human_input_response
from deerflow.config.agents_config import list_custom_agents, load_agent_config
from deerflow.config.paths import make_safe_user_id
from deerflow.runtime import END_SENTINEL, StreamBridge
from deerflow.runtime.goal import parse_goal_command
from deerflow.runtime.keyed_lock import AsyncKeyedLockTable
from deerflow.runtime.run_origin import DEERFLOW_ORIGIN_KEY, make_origin
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.skills.slash import parse_slash_skill_reference
from deerflow.skills.storage import get_or_new_skill_storage
from deerflow.skills.storage.skill_storage import SkillStorage
from deerflow.trace_context import ensure_trace_context
from deerflow.uploads.manager import apply_upload_sandbox_permits
from deerflow.utils.file_io import await_drained
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
logger = logging.getLogger(__name__)
DEFAULT_LANGGRAPH_URL = "http://localhost:8001/api"
DEFAULT_GATEWAY_URL = "http://localhost:8001"
DEFAULT_ASSISTANT_ID = "lead_agent"
DEFAULT_CHANNEL_MAX_CONCURRENCY = 5
DEFAULT_CHANNEL_SHUTDOWN_GRACE_PERIOD_SECONDS = 3.0
CUSTOM_AGENT_NAME_PATTERN = re.compile(r"^[A-Za-z0-9-]+$")
CHANNEL_AGENT_METADATA_KEY = "channel_agent_name"
THREAD_AGENT_METADATA_KEY = "agent_name"
CHANNEL_MODEL_METADATA_KEY = "channel_model_name"
# Bound for the manager's in-process per-thread caches (agent/model pins,
# synced-metadata bookkeeping). One constant so the clear-on-pressure bound
# cannot drift between call sites.
IN_PROCESS_CACHE_MAX_ENTRIES = 4096
MAX_CHANNEL_AGENT_LIST_ITEMS = 50
MAX_CHANNEL_AGENT_DESCRIPTION_CHARS = 120
# Lead-agent recursion budget (LangGraph super-steps for the lead graph only).
# This is independent of subagent depth: a `task()` dispatch runs the whole
# subagent inside ONE lead tools-node step, and subagents enforce their own
# limit via `subagents.max_turns` (see SubagentExecutor). Do not conflate this
# 100 with the general-purpose subagent's max_turns.
DEFAULT_RUN_CONFIG: dict[str, Any] = {"recursion_limit": 100}
DEFAULT_RUN_CONTEXT: dict[str, Any] = {
"thinking_enabled": True,
"is_plan_mode": False,
"subagent_enabled": False,
}
STREAM_UPDATE_MIN_INTERVAL_SECONDS = 1.0
STREAM_UPDATE_MIN_CHARS = 60 # flush immediately when this many chars accumulate
# Stream modes requested from the runtime, and the SSE event names under which
# the message-tuple stream may arrive: the embedded runtime (and LangGraph
# Platform) deliver the requested "messages-tuple" mode as event "messages".
STREAM_MODES = ["messages-tuple", "values"]
MESSAGE_STREAM_EVENTS = ("messages-tuple", "messages")
THREAD_BUSY_MESSAGE = "This conversation is already processing another request. Please wait for it to finish and try again."
# IM runs must outlive the request that created them. The manager holds a
# ``runs.wait`` / ``runs.stream`` request open for the whole run, and that
# request dies with this Gateway instance: a leader hand-off, a rolling
# restart, a crash. The Gateway's default ``on_disconnect="cancel"`` would then
# cancel the run itself; ``continue`` lets it finish and persist its thread
# state, so the user still finds the answer in the Web UI even though the IM
# reply is lost with the transport. ``runs.create`` (fire-and-forget) needs no
# such policy: its request returns as soon as the run is pending.
CHANNEL_RUN_ON_DISCONNECT = "continue"
# Run admission is idempotent on this key: a redelivery that outlived the
# inbound dedupe window (or a restart, or a fail-open dedupe store) gets the
# run the Gateway already admitted instead of a second one. The digest names
# the inbound dedupe key. The Gateway scopes an internal caller's key by
# owner only, so a redelivery requested on the thread ``/new`` bound later
# still meets the admitted run (a 409) instead of starting another one.
IDEMPOTENCY_KEY_HEADER = "Idempotency-Key"
_RUN_IDEMPOTENCY_KEY_PREFIX = "channel-inbound:v1:"
# The Gateway reuses a keyed run only for an identical input, so a keyed
# message's human message id is derived from the same key instead of uuid4.
_HUMAN_MESSAGE_ID_NAMESPACE = uuid.UUID("5b7c1e0e-8f5a-4d43-9a63-0c5b2f1d7e21")
# 409 details that name this message's own keyed run rather than another
# active run: the redelivery's input differs (re-ingested attachments get new
# names), or the run is live on a peer whose stream this instance cannot read.
_ADMITTED_RUN_CONFLICT_DETAILS = ("Idempotency-Key already used", "is not active on this worker and cannot be streamed")
ALREADY_ADMITTED_MESSAGE = "This message is already being processed. Its answer will appear in this conversation in DeerFlow."
_LIVE_RUN_STATUSES = frozenset({"pending", "running"})
@dataclass
class _RunAdmission:
"""How far one inbound message's run request got, for dedupe release.
``requested`` flips right before the run request leaves the manager;
``confirmed`` once the Gateway proved it admitted the run (a run id, a
stream frame, a completed response). See ``_keep_dedupe_key_after``.
"""
requested: bool = False
confirmed: bool = False
_RUN_ADMISSION: ContextVar[_RunAdmission | None] = ContextVar("deerflow_channel_run_admission", default=None)
# Transport failures that can only happen after the Gateway read the request:
# the run was admitted and continues (``on_disconnect: continue``), so a
# provider redelivery must stay a duplicate. Connection failures and HTTP
# status errors mean the Gateway never admitted it.
_RUN_REQUEST_REACHED_GATEWAY_ERRORS = (httpx.ReadTimeout, httpx.ReadError, httpx.RemoteProtocolError)
class _RunFailedFrame(RuntimeError):
"""The stream's ``error`` frame: the run ended, so a retry is not a duplicate."""
@contextmanager
def _run_admission_scope(*, fresh: bool = False) -> Iterator[_RunAdmission]:
"""Install the admission record for one inbound message.
The worker loop installs a fresh one per message; direct callers of
``_handle_message`` get one only when none is installed yet.
"""
existing = _RUN_ADMISSION.get()
if existing is not None and not fresh:
yield existing
return
admission = _RunAdmission()
token = _RUN_ADMISSION.set(admission)
try:
yield admission
finally:
_RUN_ADMISSION.reset(token)
def _note_run_requested() -> None:
if (admission := _RUN_ADMISSION.get()) is not None:
admission.requested = True
def _note_run_confirmed() -> None:
if (admission := _RUN_ADMISSION.get()) is not None:
admission.requested = True
admission.confirmed = True
# Recovery replies must say the turn was dropped: the run was rejected before
# starting, so the triggering message never reached the agent.
_STALE_PIN_RETRY_NOTE = "Your message was not processed — please send it again."
class StalePinRecovery(NamedTuple):
"""Outcome of a handled stale-pin admission failure.
``text`` is the user-facing reply; ``resolved`` tells requeueing callers
(the follow-up drain) whether the next attempt is free of *this*
rejection — True when the pin was cleared or a different selection is now
current, False when nothing was cleared and the next attempt rejects
identically (the durable reset failed, or the selection came from session
configuration, which still carries it).
"""
text: str
resolved: bool
BOUND_IDENTITY_REQUIRED_MESSAGE = "Connect this channel from DeerFlow Settings, complete the in-channel connect step, then send your message again."
BOUND_IDENTITY_UNAVAILABLE_MESSAGE = "Channel connection verification is temporarily unavailable. Please try again later or contact the DeerFlow operator."
# Inbound-redelivery dedup window. The dedupe state lives in
# ``self._inbound_dedupe_store``: the default in-process Memory store is
# local to this Gateway process (a recorded key survives only for the store's
# TTL / entry cap and is gone across a restart), while a Postgres-backed store
# is shared across pods. 10 minutes is a deliberately bounded window: long
# enough to absorb a near-term redelivery of the same event — whether a
# provider's own automatic retry or an operator resend — without keeping a
# growing ledger.
#
# For GitHub specifically: GitHub does NOT automatically retry or redeliver
# a failed delivery (non-2xx response, timeout, or connection error) — it
# is simply recorded as failed. See GitHub's own documentation:
# https://docs.github.com/en/webhooks/using-webhooks/handling-failed-webhook-deliveries.
# Every redelivery of the same ``X-GitHub-Delivery`` GUID is therefore an
# explicit action — the repo/App "Redeliver" button, the REST API, or an
# operator's own scheduled recovery script polling the failed-deliveries
# endpoint (the pattern GitHub's own docs recommend) — never an automatic
# GitHub-side retry. This TTL exists to absorb exactly those explicit
# near-term replays.
#
# At the boundary: a manual redelivery (e.g. GitHub's "Redeliver" button)
# clicked *after* the TTL has elapsed, or any redelivery following a Gateway
# restart, is no longer recognized as a duplicate here — the key has already
# been evicted, or never existed in the new process. Its run request still
# carries the ``Idempotency-Key`` derived from the same dedupe key, so the
# Gateway hands back the run it admitted for the first delivery (whatever
# its outcome) rather than running the turn again. Only messages without a
# dedupe key (no stable provider message id or workspace) can still run twice.
# Follow-up buffering for busy fire_and_forget threads (issue #4121 Slice 2).
# A ConflictError on a channel opted into ChannelRunPolicy.buffer_followups_on_busy
# buffers the triggering message per-thread instead of only logging it; a
# background watcher drains the buffer into a coalesced follow-up run once the
# busy run's StreamBridge stream reaches END_SENTINEL. See _buffer_followup,
# _drain_followups_for_thread, and _watch_run_and_drain_followups below.
FOLLOWUP_BUFFER_MAX_PER_THREAD = 20
FOLLOWUP_DRAIN_BATCH_SIZE = 10
FOLLOWUP_BLOCK_TAG = "followups-while-busy"
# Only server-stable provider message ids: client-generated ids (client_msg_id,
# client_id) are not guaranteed identical across a provider's own redelivery, so
# keying dedupe on them would miss exactly the retries we want to absorb.
INBOUND_DEDUPE_METADATA_KEYS = ("event_id", "message_id", "msg_id")
# Providers that persist connection.workspace_id = chat_id (telegram / feishu /
# wechat upsert_connection). Unbound inbound has no connection, so msg.workspace_id
# is unset; chat_id is still the tenant scope and is safe for the dedupe key.
# Slack is intentionally excluded: its channel ids are not globally unique.
CHAT_SCOPED_WORKSPACE_CHANNELS = frozenset({"telegram", "feishu", "wechat"})
InboundFileReader = Callable[[dict[str, Any], httpx.AsyncClient], Awaitable[bytes | None]]
# Cap for URL-based inbound attachments fetched by the generic reader (WeCom
# media today; the WeChat reader is path-only by design — see
# _read_wechat_inbound_file). The bytes are buffered in memory before being
# persisted, so an oversized attachment must be refused before it is fully
# read, not after — mirrors DingTalkChannel._download_by_code. 50 MB is a
# deliberate default, not the platform ceiling: published WeCom callback
# examples document files up to 100 MB, but the whole file is buffered (and
# decrypt_file allocates a second copy), so the bound matches the sibling
# channels' inbound caps (DingTalk's identically-sized 50 MB, WeChat's
# max_inbound_file_bytes) and halves worst-case per-message buffering; a
# legit-but-oversized file drops with a host-labeled warning naming the limit.
MAX_INBOUND_URL_FILE_BYTES = 50 * 1024 * 1024
# WeCom inbound media URLs come from the platform's WS frames (wecom.py passes
# ``payload.get("url")`` straight through), the same untrusted-input shape as
# WeChat's ``full_url``; the fetch is therefore gated to platform-owned hosts
# before streaming, mirroring WechatChannel._is_allowed_media_url. Two
# families: qq.com hosts, and the temporary signed COS links WeCom actually
# serves media from — ``ww-aibot-img-<APPID>.cos.<region>.myqcloud.com``
# (published callback examples; valid ~5 minutes). The COS numeric suffix is
# the owner's Tencent Cloud APPID and the bucket name is user-chosen, so any
# Tencent Cloud account could register a matching ``ww-aibot-img-*`` bucket:
# the shape alone proves nothing about ownership. Only the APPID observed in
# Tencent's published aibot callback examples (1258476243) is trusted by
# default; media from any other account — including a future WeCom rotation
# to a new APPID — goes through the operator suffix list
# ``channels.wecom.allowed_media_hosts``.
WECOM_ALLOWED_MEDIA_HOST_SUFFIXES = ("qq.com",)
_WECOM_MEDIA_COS_APPIDS = frozenset({"1258476243"})
_WECOM_MEDIA_COS_HOST_RE = re.compile(r"^ww-aibot-img-(?P<appid>\d+)\.cos\.[a-z0-9-]+\.myqcloud\.com$")
_METADATA_DROP_KEYS = frozenset({"raw_message", "ref_msg"})
def _slim_metadata(meta: dict[str, Any]) -> dict[str, Any]:
"""Return a shallow copy of *meta* with known-large keys removed."""
return {k: v for k, v in meta.items() if k not in _METADATA_DROP_KEYS}
INBOUND_FILE_READERS: dict[str, InboundFileReader] = {}
def register_inbound_file_reader(channel_name: str, reader: InboundFileReader) -> None:
INBOUND_FILE_READERS[channel_name] = reader
async def _read_http_inbound_file(file_info: dict[str, Any], client: httpx.AsyncClient) -> bytes | None:
url = file_info.get("url")
if not isinstance(url, str) or not url:
return None
chunks: list[bytes] = []
total = 0
# The transfer must stay undecoded: aiter_bytes() transparently decodes
# Content-Encoding, and the decoder allocates the whole decompressed body
# before yielding a single chunk — a compressed response from an admitted
# host would blow past the cap exactly like the unbounded read this
# reader exists to prevent. Identity is requested up front, any residual
# encoding is refused before reading, and aiter_raw() never decodes.
async with client.stream("GET", url, headers={"Accept-Encoding": "identity"}) as response:
response.raise_for_status()
encoding = (response.headers.get("content-encoding") or "").strip().lower()
if encoding and encoding != "identity":
logger.warning(
"[Manager] inbound file response uses Content-Encoding %r, dropping before decode: %s",
encoding,
_inbound_file_label(file_info, url),
)
return None
async for chunk in response.aiter_raw():
total += len(chunk)
if total > MAX_INBOUND_URL_FILE_BYTES:
logger.warning(
"[Manager] inbound file exceeds %d bytes download limit, dropping: %s",
MAX_INBOUND_URL_FILE_BYTES,
_inbound_file_label(file_info, url),
)
return None
chunks.append(chunk)
return b"".join(chunks)
def _url_host(url: str) -> str:
"""Best-effort host extraction for logging; never raises, never logs the URL.
Media URLs can carry access tokens in their query strings, so only the
host is surfaced in drop warnings.
"""
try:
return (urlparse(url).hostname or "").lower()
except ValueError:
return ""
def _inbound_file_label(file_info: dict[str, Any], url: str | None = None, idx: int | None = None) -> str:
"""Sanitized logging label for one inbound attachment: filename, else URL host.
Signed media URLs carry credentials in both the path and the query string
(WeCom COS links: ``/path?sign=...&q-signature=...``), so no part of the
URL itself may reach the logs — only the hostname. Filenames are webhook
supplied, so whitespace is collapsed and length capped to keep a crafted
name from forging log lines (mirrors dingtalk._display_filename).
"""
filename = file_info.get("filename")
if isinstance(filename, str) and filename.strip():
return re.sub(r"\s+", " ", filename).strip()[:80]
source = url if isinstance(url, str) else None
if source is None:
for key in ("url", "full_url"):
value = file_info.get(key)
if isinstance(value, str) and value.strip():
source = value
break
if source:
host = _url_host(source)
if host:
return f"host={host}"
return f"#{idx}" if idx is not None else "<unnamed>"
def _reader_error_summary(exc: BaseException) -> str:
"""Sanitized exception summary for inbound-media reader failures.
httpx exceptions format the full request URL into their message —
``HTTPStatusError`` includes the path and query, i.e. the signed download
credentials — and rendering the traceback (``logger.exception``) would
reproduce them verbatim, so only the class name and explicitly safe
fields ever reach the logs.
"""
summary = type(exc).__name__
if isinstance(exc, httpx.HTTPStatusError):
summary = f"{summary} ({exc.response.status_code})"
return summary
def _is_allowed_wecom_media_url(url: str, extra_suffixes: frozenset[str] | tuple[str, ...] | list[str] = ()) -> bool:
"""Platform-owned-host gate for WeCom inbound media fetches.
Matching semantics mirror ``WechatChannel._is_allowed_media_url``: http/https
only, and ``notqq.com`` / ``qq.com.evil.io`` never match a ``qq.com`` suffix.
The COS shape is matched exactly (see ``_WECOM_MEDIA_COS_HOST_RE``) AND its
numeric suffix must be one of the verified WeCom-owned APPIDs
(``_WECOM_MEDIA_COS_APPIDS``) — the suffix is a Tencent Cloud account
APPID and bucket names are user-chosen, so the shape alone would admit
any account that registers a lookalike bucket. Operator-supplied
``channels.wecom.allowed_media_hosts`` suffixes are merged in on top of the
hard-coded families (see ``_wecom_extra_media_host_suffixes``).
"""
try:
parsed = urlparse(url)
except ValueError:
return False
if parsed.scheme not in ("http", "https"):
return False
host = (parsed.hostname or "").lower()
if not host:
return False
if any(host == suffix or host.endswith(f".{suffix}") for suffix in (*WECOM_ALLOWED_MEDIA_HOST_SUFFIXES, *extra_suffixes)):
return True
cos_match = _WECOM_MEDIA_COS_HOST_RE.fullmatch(host)
return cos_match is not None and cos_match.group("appid") in _WECOM_MEDIA_COS_APPIDS
def _wecom_extra_media_host_suffixes() -> frozenset[str]:
"""Operator host suffixes from the live WeCom channel, if one is running.
The registry reader is module-level while ``channels.wecom.allowed_media_hosts``
is per-channel config, so the extras are resolved per call from the running
channel instance — the same service-reach-in ``_channel_supports_streaming``
already uses. Empty when no WeCom channel is up (direct calls, most tests),
leaving only the strict built-in families; a strict default plus an operator
escape hatch means a platform URL-shape change never requires widening the
hard-coded pattern for every deployment.
"""
try:
from app.channels.service import get_channel_service
service = get_channel_service()
channel = service.get_channel("wecom") if service is not None else None
except Exception:
return frozenset()
return frozenset(getattr(channel, "allowed_media_host_suffixes", ()) or ())
async def _read_wecom_inbound_file(file_info: dict[str, Any], client: httpx.AsyncClient) -> bytes | None:
url = file_info.get("url")
if isinstance(url, str) and url and not _is_allowed_wecom_media_url(url, _wecom_extra_media_host_suffixes()):
logger.warning(
"[Manager] WeCom inbound media URL host is not allowed, dropping file=%s host=%s",
_inbound_file_label(file_info, url),
_url_host(url),
)
return None
data = await _read_http_inbound_file(file_info, client)
if data is None:
return None
aeskey = file_info.get("aeskey") if isinstance(file_info.get("aeskey"), str) else None
if not aeskey:
return data
try:
from aibot.crypto_utils import decrypt_file
except Exception:
logger.exception("[Manager] failed to import WeCom decrypt_file")
return None
return decrypt_file(data, aeskey)
async def _read_wechat_inbound_file(file_info: dict[str, Any], client: httpx.AsyncClient) -> bytes | None:
raw_path = file_info.get("path")
if isinstance(raw_path, str) and raw_path.strip():
try:
return await await_drained(asyncio.to_thread(Path(raw_path).read_bytes))
except OSError:
logger.exception("[Manager] failed to read WeChat inbound file from local path: %s", raw_path)
return None
# No re-fetch fallback: the only producer (WechatChannel) always stages a
# local ``path`` and drops the attachment when staging fails, so a file
# dict without one has no legitimate fetch source. Fetching ``full_url``
# here would be the one ungated Gateway-host fetch left in the WeChat
# path — the channel-side ``channels.wechat.allowed_media_hosts`` gate
# cannot reach this module-level reader, and re-implementing it with
# divergent rules would drop media the operator explicitly allowed.
logger.debug("[Manager] WeChat inbound file has no staged local path, skipping")
return None
register_inbound_file_reader("wecom", _read_wecom_inbound_file)
register_inbound_file_reader("wechat", _read_wechat_inbound_file)
class InvalidChannelSessionConfigError(ValueError):
"""Raised when IM channel session overrides contain invalid agent config."""
class SlashSkillCommandResolutionError(RuntimeError):
"""Raised when IM slash-skill command resolution cannot complete safely."""
@dataclass(frozen=True, slots=True)
class _SlashSkillCommandResolution:
route_to_chat: bool = False
failure_message: str | None = None
@dataclass(frozen=True, slots=True)
class _BoundIdentityRejection:
message: str = BOUND_IDENTITY_REQUIRED_MESSAGE
# Server-side connection id that may be used only as an outbound routing
# hint for the rejection message. This is never copied from the inbound
# message; it comes from the repository re-read when available.
outbound_connection_id: str | None = None
# Server-side owner for the outbound routing connection above. It lets
# channel senders preserve per-connection context without trusting the
# rejected inbound identity assertion.
outbound_owner_user_id: str | None = None
@dataclass(slots=True)
class _SerializedThreadRunState:
"""Per-thread lock state for channels that queue same-thread turns."""
lock: asyncio.Lock
waiters: int = 0
@dataclass(slots=True)
class _FollowupEntry:
"""One inbound message's text, buffered because its thread was busy.
Routing/policy identity (channel_name, metadata, owner headers) for the
eventual drained run comes from a separate ``carrier_msg`` — see
``ChannelManager._drain_followups_for_thread`` — not from a per-entry
message, since every buffered entry for one thread_id shares that
identity already (thread_id is itself derived deterministically from
(repo, number, agent_name) for GitHub). Only the text needs to survive
per entry.
"""
dedupe_key: str
text: str
def _is_thread_busy_error(exc: BaseException | None) -> bool:
if exc is None:
return False
if isinstance(exc, ConflictError):
return not _is_admitted_run_conflict(exc)
return "already running a task" in str(exc)
def _is_admitted_run_conflict(exc: BaseException | None) -> bool:
"""True for a 409 about this message's own keyed run (see ``_ADMITTED_RUN_CONFLICT_DETAILS``)."""
return isinstance(exc, ConflictError) and any(detail in str(exc) for detail in _ADMITTED_RUN_CONFLICT_DETAILS)
def _is_run_status_result(result: Any) -> bool:
"""A ``/wait`` response carrying run status instead of thread state.
The Gateway answers a reused keyed run this way, because the thread head may
already belong to a later run. A fresh run that left no checkpoint looks the
same; both are resolved through ``_admitted_turn``.
"""
return isinstance(result, dict) and "messages" not in result and isinstance(result.get("status"), str)
def _turn_of_message(values: Any, message_id: str) -> dict[str, Any] | None:
"""Thread *values* cut after the turn that answered human message *message_id*.
The cut ends before the next visible human message, so the response, artifact
and clarification extractors (which read the turn after the last human
message) see this message's turn rather than a later run's.
"""
if not isinstance(values, dict) or not isinstance(messages := values.get("messages"), list):
return None
for index, message in enumerate(messages):
if isinstance(message, dict) and message.get("type") == "human" and message.get("id") == message_id:
end = next(
(later for later in range(index + 1, len(messages)) if isinstance(messages[later], dict) and messages[later].get("type") == "human" and not _is_hidden_human_control_message(messages[later])),
len(messages),
)
return {**values, "messages": messages[:end]}
return None
def _is_model_allowlist_error(exc: BaseException | None) -> bool:
"""True when *exc* is the run-admission rejection for an unlisted model.
``start_run`` rejects a run whose carried ``model_name`` is not in the
configured model allowlist with this message; a stale ``/model`` pin
surfaces here. Matched on the stable message substring, the same style
as ``_is_thread_busy_error``.
"""
if exc is None:
return False
return "is not in the configured model allowlist" in str(exc)
def _followup_dedupe_key(msg: InboundMessage) -> str:
"""Best-effort stable identifier for a buffered follow-up comment.
Mirrors ``_inbound_dedupe_key``'s provider-id preference order (a GitHub
webhook delivery id first, then the generic provider-message-id metadata
keys), but scoped to one thread's follow-up buffer rather than the
global cross-channel inbound dedupe map, and always returns a usable key
— falling back to an object-identity key — since the follow-up buffer
must still accept an entry even when a provider omits every known id
field (unlike ``_inbound_dedupe_key``, which returns ``None`` to skip
dedupe entirely in that case).
"""
metadata = msg.metadata or {}
gh = metadata.get("github")
if isinstance(gh, dict):
delivery_id = gh.get("delivery_id")
if delivery_id:
return f"github:delivery:{delivery_id}"
for key in INBOUND_DEDUPE_METADATA_KEYS:
value = metadata.get(key)
if value:
return f"{key}:{value}"
raw_message = metadata.get("raw_message")
if isinstance(raw_message, Mapping):
for key in INBOUND_DEDUPE_METADATA_KEYS:
value = raw_message.get(key)
if value:
return f"{key}:{value}"
# No stable provider id available: fall back to a per-message key so the
# entry is still buffered (just never deduped against a redelivery).
return f"__no_id__:{id(msg)}:{msg.created_at}"
def _format_followup_block(entries: list[_FollowupEntry]) -> str:
"""Coalesce buffered follow-up entries into one templated input block."""
lines = [
f"<{FOLLOWUP_BLOCK_TAG}>",
"The following messages arrived on this thread while a previous run was still in progress. They were queued and are now delivered together as one turn:",
"",
]
for idx, entry in enumerate(entries, start=1):
escaped_text = escape(entry.text, quote=False).replace(
"\n",
"\n ",
)
lines.append(f"{idx}. {escaped_text}")
lines.append(f"</{FOLLOWUP_BLOCK_TAG}>")
return "\n".join(lines)
def _as_dict(value: Any) -> dict[str, Any]:
return dict(value) if isinstance(value, Mapping) else {}
def _merge_dicts(*layers: Any) -> dict[str, Any]:
merged: dict[str, Any] = {}
for layer in layers:
if isinstance(layer, Mapping):
merged.update(layer)
return merged
def _select_carried_value(run_config: Mapping, run_context: Mapping, key: str) -> str | None:
"""Pick a session-carried scalar out of merged run carriers.
Carrier precedence mirrors the Gateway merge a run crosses: when the
merged request config carries a ``context`` section, ``build_run_config``
keeps it and *drops* ``configurable``, so ``config.context`` beats the
top-level ``context``; otherwise ``merge_run_context_overrides`` /
``_get_runtime_config`` let the top-level ``context`` carrier win over
``configurable``.
"""
if "context" in run_config:
candidates = (
_as_dict(run_config.get("context")).get(key),
run_context.get(key),
)
else:
candidates = (
run_context.get(key),
_as_dict(run_config.get("configurable")).get(key),
)
for candidate in candidates:
if isinstance(candidate, str) and candidate.strip():
return candidate.strip()
return None
def _select_carried_model_name(run_config: Mapping, run_context: Mapping) -> str | None:
"""Pick the session-configured model out of merged run carriers.
Single source of truth for the carrier precedence shared by the run
path and the ``/model`` status reply (see ``_select_carried_value`` for
the merge rule). Reordering or extending session layers must update this
rule and the layer merge in ``_resolve_run_params`` together.
"""
return _select_carried_value(run_config, run_context, "model_name")
def _select_carried_agent_name(run_config: Mapping, run_context: Mapping) -> str | None:
"""Pick the session-carried agent out of merged run carriers.
The effective agent rides the same three carriers across the Gateway
merge as ``model_name``, so the same precedence applies (the run path
only *defaults* an assistant-derived ``agent_name`` into the top-level
context with ``setdefault``, so a context-carried agent always beats it).
"""
return _select_carried_value(run_config, run_context, "agent_name")
def _normalize_custom_agent_name(raw_value: str) -> str:
"""Normalize legacy channel assistant IDs into valid custom agent names."""
normalized = raw_value.strip().lower().replace("_", "-")
if not normalized:
raise InvalidChannelSessionConfigError("Channel session assistant_id is empty. Use 'lead_agent' or a valid custom agent name.")
if not CUSTOM_AGENT_NAME_PATTERN.fullmatch(normalized):
raise InvalidChannelSessionConfigError(f"Invalid channel session assistant_id {raw_value!r}. Use 'lead_agent' or a custom agent name containing only letters, digits, and hyphens.")
return normalized
def _apply_explicit_agent_choice(
run_config: dict[str, Any],
run_context: dict[str, Any],
agent_name: str | None,
) -> None:
"""Pin or clear an explicit channel agent in every runtime carrier.
Gateway accepts ``agent_name`` from the request's top-level context and
from either RunnableConfig container. Its compatibility merge preserves
existing values with ``setdefault``, so an explicit ``/agent use`` choice
must normalize all three carriers before the request crosses that boundary.
``None`` represents an explicit reset to the default lead agent.
"""
carriers = [run_context]
for section in ("configurable", "context"):
value = run_config.get(section)
if isinstance(value, Mapping):
# Session layers own their nested dictionaries. Copy before changing
# one so selecting an agent for a conversation cannot mutate the
# manager's reusable channel configuration.
copied = dict(value)
run_config[section] = copied
carriers.append(copied)
for carrier in carriers:
if agent_name is None:
carrier.pop("agent_name", None)
else:
carrier["agent_name"] = agent_name
def _apply_explicit_model_choice(
run_config: dict[str, Any],
run_context: dict[str, Any],
model_name: str,
) -> None:
"""Pin a ``/model`` selection in every runtime carrier.
``merge_run_context_overrides`` fills ``config['configurable']`` and
``config['context']`` from ``body.context`` with ``setdefault``, so a
``model_name`` inherited from channel session config in either carrier
would silently win over a pin carried only by ``body.context``. Mirrors
``_apply_explicit_agent_choice``: normalize all three carriers.
"""
carriers = [run_context]
for section in ("configurable", "context"):
value = run_config.get(section)
if isinstance(value, Mapping):
copied = dict(value)
run_config[section] = copied
carriers.append(copied)
for carrier in carriers:
carrier["model_name"] = model_name
def _extract_response_text(result: dict | list) -> str:
"""Extract the last AI message text from a LangGraph runs.wait result.
``runs.wait`` returns the final state dict which contains a ``messages``
list. Each message is a dict with at least ``type`` and ``content``.
Handles special cases:
- Regular AI text responses
- Clarification interrupts (``ask_clarification`` tool messages)
"""
if isinstance(result, list):
messages = result
elif isinstance(result, dict):
messages = result.get("messages", [])
else:
return ""
# Walk backwards to find usable response text, but stop at the last
# human message to avoid returning text from a previous turn.
for msg in reversed(messages):
if not isinstance(msg, dict):
continue
msg_type = msg.get("type")
# Stop at the last human message — anything before it is a previous turn
if msg_type == "human":
if _is_hidden_human_control_message(msg):
continue
break
# Check for tool messages from ask_clarification (interrupt case)
if msg_type == "tool" and msg.get("name") == "ask_clarification":
content = msg.get("content", "")
if isinstance(content, str) and content:
return content
# Regular AI message with text content
if msg_type == "ai":
content = msg.get("content", "")
if isinstance(content, str) and content:
return content
# content can be a list of content blocks
if isinstance(content, list):
parts = []
for block in content:
if isinstance(block, dict) and block.get("type") == "text":
parts.append(block.get("text", ""))
elif isinstance(block, str):
parts.append(block)
text = "".join(parts)
if text:
return text
return ""
def _messages_from_result(result: dict | list) -> list[Any]:
if isinstance(result, list):
return result
if isinstance(result, dict):
messages = result.get("messages", [])
if isinstance(messages, list):
return messages
return []
def _current_turn_messages(result: dict | list) -> list[dict[str, Any]]:
messages = _messages_from_result(result)
current_turn: list[dict[str, Any]] = []
for msg in reversed(messages):
if not isinstance(msg, dict):
continue
if msg.get("type") == "human":
break
current_turn.append(msg)
current_turn.reverse()
return current_turn
def _has_current_turn_clarification(result: dict | list) -> bool:
"""Return True only when the current turn's final result is clarification."""
for msg in reversed(_current_turn_messages(result)):
msg_type = msg.get("type")
if msg_type == "tool":
return msg.get("name") == "ask_clarification"
if msg_type == "ai":
content = msg.get("content")
if isinstance(content, str):
if content:
return False
elif content:
return False
if msg.get("tool_calls"):
return False
return False
def _extract_human_input_payload(result: dict | list) -> dict[str, Any] | None:
"""Return the clarification's structured ``human_input`` payload, if present.
The clarification middleware stashes it under the tool message's
``artifact.human_input``. Only the current turn's trailing clarification
counts — a clarification answered earlier in the run must not leak into a
later final outbound.
"""
for msg in reversed(_current_turn_messages(result)):
if msg.get("type") == "tool" and msg.get("name") == "ask_clarification":
artifact = msg.get("artifact")
if isinstance(artifact, dict):
payload = artifact.get("human_input")
if isinstance(payload, dict):
return payload
return None
if msg.get("type") == "ai":
content = msg.get("content")
if isinstance(content, str) and content:
return None
if msg.get("tool_calls"):
return None
return None
def _response_metadata_with_clarification(
result: dict | list,
base_metadata: dict[str, Any],
*,
pending_clarification: bool,
) -> dict[str, Any]:
"""``_response_metadata`` plus the clarification's structured payload.
Channels that render richer clarification affordances (e.g. Feishu button
cards) read the payload from metadata; channels that do not simply ignore it.
"""
metadata = _response_metadata(base_metadata, pending_clarification=pending_clarification)
if pending_clarification:
payload = _extract_human_input_payload(result)
if payload is not None:
metadata[HUMAN_INPUT_REQUEST_METADATA_KEY] = payload
return metadata
def _response_metadata(base_metadata: dict[str, Any], *, pending_clarification: bool = False) -> dict[str, Any]:
metadata = _slim_metadata(base_metadata)
if pending_clarification:
metadata[PENDING_CLARIFICATION_METADATA_KEY] = True
return metadata
def _thread_channel_metadata(msg: InboundMessage) -> dict[str, Any]:
channel_source: dict[str, Any] = {
"type": "im_channel",
"provider": msg.channel_name,
"chat_id": msg.chat_id,
}
if msg.topic_id:
channel_source["topic_id"] = msg.topic_id
if msg.thread_ts:
channel_source["thread_ts"] = msg.thread_ts
if msg.connection_id:
channel_source["connection_id"] = msg.connection_id
return {"channel_source": channel_source}
def _extract_text_content(content: Any) -> str:
"""Extract text from a streaming payload content field."""
if isinstance(content, str):
return content
if isinstance(content, list):
parts: list[str] = []
for block in content:
if isinstance(block, str):
parts.append(block)
elif isinstance(block, Mapping):
text = block.get("text")
if isinstance(text, str):
parts.append(text)
else:
nested = block.get("content")
if isinstance(nested, str):
parts.append(nested)
return "".join(parts)
if isinstance(content, Mapping):
for key in ("text", "content"):
value = content.get(key)
if isinstance(value, str):
return value
return ""
def _merge_stream_text(existing: str, chunk: str) -> str:
"""Merge either delta text or cumulative text into a single snapshot."""
if not chunk:
return existing
if not existing:
return chunk
# Cumulative re-delivery: strictly longer and starts with existing.
if len(chunk) > len(existing) and chunk.startswith(existing):
return chunk
# Everything else is a delta — always append, even when the delta
# happens to match the buffer suffix (e.g. 'hel' + 'l') or equals
# the buffer (CJK reduplication: '谢' + '谢' = '谢谢'). Channels feed
# only delta ('messages-tuple') events to this function; 'values'
# snapshots are consumed via a separate branch, so a same-content
# delta (chunk == existing) still represents a fresh token to keep.
return existing + chunk
def _extract_stream_message_id(payload: Any, metadata: Any) -> str | None:
"""Best-effort extraction of the streamed AI message identifier."""
candidates = [payload, metadata]
if isinstance(payload, Mapping):
candidates.append(payload.get("kwargs"))
for candidate in candidates:
if not isinstance(candidate, Mapping):
continue
for key in ("id", "message_id"):
value = candidate.get(key)
if isinstance(value, str) and value:
return value
return None
def _stream_payload_type(payload: Mapping[str, Any]) -> str:
"""Resolve the message ``type`` of one ``messages-tuple`` payload.
Two payload shapes reach this function and they name the message type in
different places:
* The shape DeerFlow's own gateway emits (``runtime/serialization.py``
calls ``model_dump()``): ``type`` is the LangChain literal directly --
``"ai"`` / ``"AIMessageChunk"`` / ``"human"`` / ``"tool"`` / ``"system"``.
* LangChain's ``to_json()`` constructor shape, which
``_extract_stream_message_id`` and the content extraction below already
accommodate: the wrapper's own ``type`` is the literal string
``"constructor"`` and the real class name is the last element of the
``id`` path (``["langchain", "schema", "messages", "AIMessageChunk"]``),
with the constructor kwargs under ``kwargs``.
Reading only the top level would classify every constructor-shaped payload
as ``"constructor"``, which an allowlist rejects (safe) but which would
also mean hidden context and assistant output are treated identically --
so the class name is resolved properly instead of guessed.
"""
raw_type = payload.get("type")
if isinstance(raw_type, str) and raw_type and raw_type != "constructor":
return raw_type
kwargs = payload.get("kwargs")
if isinstance(kwargs, Mapping):
nested = kwargs.get("type")
if isinstance(nested, str) and nested:
return nested
lc_path = payload.get("id")
if isinstance(lc_path, (list, tuple)) and lc_path:
tail = lc_path[-1]
if isinstance(tail, str) and tail:
return tail
return raw_type if isinstance(raw_type, str) else ""
def _is_assistant_stream_type(payload_type: str) -> bool:
"""Is this message type assistant output, i.e. displayable in an IM channel?
An ALLOWLIST, deliberately. The previous denylist ("reject anything whose
type contains 'tool'") published every other message type, and DeerFlow
writes hidden model context into the ``messages`` channel as ordinary
messages: ``DynamicContextMiddleware`` injects the ``<memory>`` block as a
hidden ``HumanMessage`` (``type == "human"``) and rewrites the user's own
turn into a new ``HumanMessage``, and ``DurableContextMiddleware`` injects a
hidden ``<durable_context_data>`` ``HumanMessage``. LangGraph fans state
writes out on the ``messages-tuple`` stream, so all of those reached the
channel as if they were the assistant's reply -- proved live on a Buzz
relay, where each streaming update is an immutable public Nostr event and a
later corrective edit cannot unpublish the leaked one.
The accepted spellings are the ones assistant output actually carries:
LangChain serializes ``AIMessage.type`` as ``"ai"`` and
``AIMessageChunk.type`` as ``"AIMessageChunk"``; ``"assistant"`` is the
OpenAI-style spelling a foreign runtime may use. Matching is by prefix
rather than substring because a substring test is not safe here -- ordinary
English words contain "ai" ("chain", "domain"), so ``"ai" in type`` would
admit a future/foreign type name by accident, which is exactly the class of
mistake this allowlist exists to prevent. No LangChain message type other
than the AI ones begins with "ai" or "assistant".
"""
normalized = payload_type.strip().lower()
return normalized.startswith(("ai", "assistant"))
def _accumulate_stream_text(
buffers: dict[str, str],
current_message_id: str | None,
event_data: Any,
) -> tuple[str | None, str | None]:
"""Convert a ``messages-tuple`` event into the latest displayable AI text.
Only assistant output is displayable. Hidden human/system context (memory
facts, durable context, the middleware-rewritten echo of the user's own
message) and tool traffic must never be published to an IM channel; see
:func:`_is_assistant_stream_type`.
A bare ``str`` payload -- previously accepted here and buffered under the
current message id -- carries no type information at all, so it cannot be
attributed to the assistant. Nothing in DeerFlow produces it (the gateway
always serializes a ``messages-tuple`` chunk as ``[message_dict, metadata]``
via ``runtime/serialization.py::serialize_messages_tuple``), and a runtime
that did emit raw text deltas would emit hidden context the same way, with
no way to tell them apart. Under an allowlist an unattributable payload is
dropped rather than published.
"""
payload = event_data
metadata: Any = None
if isinstance(event_data, (list, tuple)):
if event_data:
payload = event_data[0]
if len(event_data) > 1:
metadata = event_data[1]
if not isinstance(payload, Mapping):
return None, current_message_id
if not _is_assistant_stream_type(_stream_payload_type(payload)):
return None, current_message_id
text = _extract_text_content(payload.get("content"))
if not text and isinstance(payload.get("kwargs"), Mapping):
text = _extract_text_content(payload["kwargs"].get("content"))
if not text:
return None, current_message_id
message_id = _extract_stream_message_id(payload, metadata) or current_message_id or "__default__"
buffers[message_id] = _merge_stream_text(buffers.get(message_id, ""), text)
return buffers[message_id], message_id
def _extract_artifacts(result: dict | list) -> list[str]:
"""Extract artifact paths from the last AI response cycle only.
Instead of reading the full accumulated ``artifacts`` state (which contains
all artifacts ever produced in the thread), this inspects the messages after
the last human message and collects file paths from ``present_files`` tool
calls. This ensures only newly-produced artifacts are returned.
"""
if isinstance(result, list):
messages = result
elif isinstance(result, dict):
messages = result.get("messages", [])
else:
return []
artifacts: list[str] = []
for msg in reversed(messages):
if not isinstance(msg, dict):
continue
# Stop at the last human message — anything before it is a previous turn
if msg.get("type") == "human":
if _is_hidden_human_control_message(msg):
continue
break
# Look for AI messages with present_files tool calls
if msg.get("type") == "ai":
for tc in msg.get("tool_calls", []):
if isinstance(tc, dict) and tc.get("name") == "present_files":
args = tc.get("args", {})
paths = args.get("filepaths", [])
if isinstance(paths, list):
artifacts.extend(p for p in paths if isinstance(p, str))
return artifacts
def _is_hidden_human_control_message(msg: Mapping[str, Any]) -> bool:
"""Distinguish internal control messages from real hidden user answers."""
if msg.get("type") != "human":
return False
additional_kwargs = msg.get("additional_kwargs")
if not isinstance(additional_kwargs, Mapping):
return False
return additional_kwargs.get("hide_from_ui") is True and read_human_input_response(additional_kwargs) is None
def _format_artifact_text(artifacts: list[str]) -> str:
"""Format artifact paths into a human-readable text block listing filenames."""
import posixpath
filenames = [posixpath.basename(p) for p in artifacts]
if len(filenames) == 1:
return f"Created File: 📎 {filenames[0]}"
return "Created Files: 📎 " + "、".join(filenames)
def _unknown_command_reply(command: str | None = None) -> str:
available = " | ".join(sorted(KNOWN_CHANNEL_COMMANDS))
if command:
return f"Unknown command: /{command}. Available commands: {available}"
return f"Unknown command. Available commands: {available}"
def _human_input_message_with_channel_response(
content: str,
*,
metadata: Mapping[str, Any],
original_content: str | None = None,
files: list[dict[str, Any]] | None = None,
message_id: str | None = None,
) -> dict[str, Any]:
"""``_human_input_message`` plus a channel-synthesized human-input answer.
Card-driven channels (Feishu button clicks) carry the answer in inbound
metadata; it must land on the human message's additional_kwargs for the
Gateway's human_input protocol, hidden from transcripts like Web card
replies.
"""
message = _human_input_message(content, original_content=original_content, files=files, message_id=message_id)
response = metadata.get(HUMAN_INPUT_RESPONSE_METADATA_KEY)
if isinstance(response, Mapping):
additional_kwargs = message.setdefault("additional_kwargs", {})
additional_kwargs[HUMAN_INPUT_RESPONSE_METADATA_KEY] = dict(response)
additional_kwargs["hide_from_ui"] = True
return message
def _human_input_message(content: str, *, original_content: str | None = None, files: list[dict[str, Any]] | None = None, message_id: str | None = None) -> dict[str, Any]:
# The Gateway records run input as sent, but add_messages gives an id-less
# message a fresh uuid in the checkpoint. A web client reconnecting to the
# run rebuilds its input from both copies and can only match them by id.
# A keyed inbound message passes an id derived from its idempotency key.
message: dict[str, Any] = {"id": message_id or str(uuid.uuid4()), "role": "human", "content": content}
if original_content is not None and original_content != content or files:
additional_kwargs: dict[str, Any] = {}
if original_content is not None and original_content != content:
additional_kwargs[ORIGINAL_USER_CONTENT_KEY] = original_content
if files:
additional_kwargs["files"] = files
message["additional_kwargs"] = additional_kwargs
return message
def _auth_disabled_owner_user_id() -> str | None:
try:
from app.gateway.auth_disabled import AUTH_DISABLED_USER_ID, is_auth_disabled
except Exception:
logger.debug("Unable to inspect auth-disabled mode for channel owner fallback", exc_info=True)
return None
return AUTH_DISABLED_USER_ID if is_auth_disabled() else None
def _effective_owner_user_id(msg: InboundMessage) -> str | None:
return _auth_disabled_owner_user_id() or msg.owner_user_id
def _apply_effective_owner(msg: InboundMessage) -> InboundMessage:
owner_user_id = _effective_owner_user_id(msg)
if owner_user_id:
msg.owner_user_id = owner_user_id
return msg
def _run_origin(msg: InboundMessage) -> dict[str, str] | None:
"""Server-owned ``deerflow_origin`` for a run this manager starts.
The Gateway honours it only from internal callers (this manager's client
carries the internal token); a provider name outside the origin shape is
dropped rather than failing the message.
"""
try:
return make_origin("github" if msg.channel_name == "github" else "im_channel", provider=msg.channel_name)
except ValueError:
return None
def _apply_run_origin(run_kwargs: dict[str, Any], msg: InboundMessage) -> None:
if (origin := _run_origin(msg)) is not None:
run_kwargs["metadata"] = {DEERFLOW_ORIGIN_KEY: origin}
def _owner_headers(msg: InboundMessage) -> dict[str, str] | None:
owner_user_id = _effective_owner_user_id(msg)
if not owner_user_id:
return None
return create_internal_auth_headers(owner_user_id=owner_user_id)
def _run_request_headers(msg: InboundMessage, idempotency_key: str | None) -> dict[str, str] | None:
headers = dict(_owner_headers(msg) or {})
if idempotency_key:
headers[IDEMPOTENCY_KEY_HEADER] = idempotency_key
return headers or None
def _safe_user_id_for_run(raw_user_id: str) -> str:
from deerflow.config.paths import get_paths
try:
return get_paths().prepare_user_dir_for_raw_id(raw_user_id)
except Exception:
logger.exception("Failed to prepare channel run user directory")
return make_safe_user_id(raw_user_id)
def _channel_storage_user_id(msg: InboundMessage) -> str | None:
"""Resolve the canonical DeerFlow user id for a channel-triggered message.
Single source of truth for both the agent **run identity**
(``_resolve_run_params`` → ``run_context["user_id"]``) and the **file/artifact
storage bucket** (``receive_file`` / ``_ingest_inbound_files`` /
``_prepare_artifact_delivery``), so the bucket the agent reads/writes always
matches where channel files are staged. Prefer the bound DeerFlow owner,
otherwise fall back to the sanitized raw platform user id. Without that
fallback, an unbound auth-enabled channel would run under ``safe(msg.user_id)``
but stage files under ``get_effective_user_id()`` (the dispatcher task's unset
contextvar → ``"default"``), so uploads would land in ``users/default/...``
while the agent reads ``users/{safe_platform_user_id}/...``. Returns ``None``
only when neither identity is available, leaving the caller to fall back to the
contextvar/default user.
Distinct from :func:`_owner_headers`, which deliberately sends the *raw* owner
id (no sanitize, no platform fallback) over HTTP for gateway to re-resolve;
this helper is the in-process, sanitized, filesystem-facing identity.
"""
owner_user_id = _effective_owner_user_id(msg)
if owner_user_id:
return _safe_user_id_for_run(owner_user_id)
if msg.user_id:
return _safe_user_id_for_run(msg.user_id)
return None
def _resolve_slash_skill_command(
text: str,
available_skills: set[str] | None = None,
storage: SkillStorage | Callable[[], SkillStorage] | None = None,
) -> _SlashSkillCommandResolution | None:
reference = parse_slash_skill_reference(text)
if reference is None:
return None
try:
resolved_storage = storage() if callable(storage) else storage or get_or_new_skill_storage()
skills = resolved_storage.load_skills(enabled_only=False)
skill = next((candidate for candidate in skills if candidate.name == reference.name), None)
if skill is None:
return None
if not skill.enabled:
return _SlashSkillCommandResolution(failure_message=f"Skill `/{reference.name}` is installed but disabled. Enable it before using slash activation.")
if available_skills is not None and reference.name not in available_skills:
return _SlashSkillCommandResolution(failure_message=f"Skill `/{reference.name}` is not available for this agent.")
return _SlashSkillCommandResolution(route_to_chat=True)
except Exception as exc:
logger.exception("[Manager] failed to resolve slash skill command")
raise SlashSkillCommandResolutionError("Failed to resolve slash skill command. Please check the skill configuration.") from exc
def _resolve_attachments(thread_id: str, artifacts: list[str], *, user_id: str | None = None) -> list[ResolvedAttachment]:
"""Resolve virtual artifact paths to host filesystem paths with metadata.
Only paths under ``/mnt/user-data/outputs/`` are accepted; any other
virtual path is rejected with a warning to prevent exfiltrating uploads
or workspace files via IM channels.
Skips artifacts that cannot be resolved (missing files, invalid paths)
and logs warnings for them.
"""
attachments: list[ResolvedAttachment] = []
effective_user_id = user_id or get_effective_user_id()
for virtual_path in artifacts:
# Security: only files under the agent outputs directory may leave the
# thread. The shared helper rejects sibling ``uploads/``/``workspace/``
# paths both lexically (``..``) and after symlink resolution, so this
# rule cannot drift from the artifact editor's.
try:
actual = resolve_outputs_confined_path(thread_id, virtual_path, user_id=effective_user_id)
except HTTPException as exc:
logger.warning("[Manager] rejected artifact path outside outputs: %s (%s)", virtual_path, exc.detail)
continue
try:
if not actual.is_file():
logger.warning("[Manager] artifact not found on disk: %s -> %s", virtual_path, actual)
continue
mime, _ = mimetypes.guess_type(str(actual))
mime = mime or "application/octet-stream"
attachments.append(
ResolvedAttachment(
virtual_path=virtual_path,
actual_path=actual,
filename=actual.name,
mime_type=mime,
size=actual.stat().st_size,
is_image=mime.startswith("image/"),
)
)
except (ValueError, OSError) as exc:
logger.warning("[Manager] failed to resolve artifact %s: %s", virtual_path, exc)
return attachments
def _prepare_artifact_delivery(
thread_id: str,
response_text: str,
artifacts: list[str],
*,
user_id: str | None = None,
) -> tuple[str, list[ResolvedAttachment]]:
"""Resolve attachments and append filename fallbacks to the text response."""
attachments: list[ResolvedAttachment] = []
if not artifacts:
return response_text, attachments
attachments = _resolve_attachments(thread_id, artifacts, user_id=user_id)
resolved_virtuals = {attachment.virtual_path for attachment in attachments}
unresolved = [path for path in artifacts if path not in resolved_virtuals]
if unresolved:
artifact_text = _format_artifact_text(unresolved)
response_text = (response_text + "\n\n" + artifact_text) if response_text else artifact_text
# Always include resolved attachment filenames as a text fallback so files
# remain discoverable even when the upload is skipped or fails.
if attachments:
resolved_text = _format_artifact_text([attachment.virtual_path for attachment in attachments])
response_text = (response_text + "\n\n" + resolved_text) if response_text else resolved_text
return response_text, attachments
def _make_inbound_file_sandbox_readable(file_path: Path) -> None:
"""Make a channel-downloaded upload readable by the sandbox process.
The gateway writes inbound files as root with 0o600; in AIO/Docker sandbox
mode the sandbox runs as a non-root user on the bind-mounted path and
cannot read the file without group/other read bits. Delegates to the shared
apply_upload_sandbox_permits helper so the permission change stays bound to
the validated upload inode (O_NOFOLLOW + fchmod) and cannot be redirected
through a symlink swapped in after validation.
"""
apply_upload_sandbox_permits(file_path, stat.S_IRGRP | stat.S_IROTH)
async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id: str | None = None) -> list[dict[str, Any]]:
try:
return await _ingest_inbound_files_impl(thread_id, msg, user_id=user_id)
finally:
await cleanup_inbound_files(msg.files)
async def _ingest_inbound_files_impl(thread_id: str, msg: InboundMessage, *, user_id: str | None = None) -> list[dict[str, Any]]:
if not msg.files:
return []
from deerflow.uploads.manager import (
UnsafeUploadPathError,
claim_unique_filename,
ensure_uploads_dir,
normalize_filename,
write_upload_file_no_symlink,
)
def _prepare_uploads_dir() -> tuple[Path, set[str]]:
# Worker thread: ensure_uploads_dir's mkdir and the iterdir enumeration are
# blocking filesystem IO that must stay off the event loop.
target = ensure_uploads_dir(thread_id, user_id=user_id)
existing = {entry.name for entry in target.iterdir() if entry.is_file()}
return target, existing
uploads_dir, seen_names = await asyncio.to_thread(_prepare_uploads_dir)
created: list[dict[str, Any]] = []
file_reader = INBOUND_FILE_READERS.get(msg.channel_name, _read_http_inbound_file)
async with httpx.AsyncClient(timeout=httpx.Timeout(20.0)) as client:
for idx, f in enumerate(msg.files):
if not isinstance(f, dict):
continue
ftype = f.get("type") if isinstance(f.get("type"), str) else "file"
filename = f.get("filename") if isinstance(f.get("filename"), str) else ""
inline_content = f.pop(INBOUND_FILE_CONTENT_KEY, None)
if isinstance(inline_content, bytes):
data = inline_content
elif isinstance(inline_content, (bytearray, memoryview)):
data = bytes(inline_content)
else:
try:
data = await file_reader(f, client)
except Exception as exc:
# Sanitized on purpose: the URL-bearing exception message
# and traceback must not reach the logs (see
# _reader_error_summary).
logger.warning(
"[Manager] failed to read inbound file: channel=%s, file=%s, error=%s",
msg.channel_name,
_inbound_file_label(f, idx=idx),
_reader_error_summary(exc),
)
continue
if data is None:
logger.warning(
"[Manager] inbound file reader returned no data: channel=%s, file=%s",
msg.channel_name,
_inbound_file_label(f, idx=idx),
)
continue
if not filename:
ext = ".bin"
if ftype == "image":
ext = ".png"
filename = f"{msg.thread_ts or 'msg'}_{idx}{ext}"
try:
safe_name = claim_unique_filename(normalize_filename(filename), seen_names)
except ValueError:
logger.warning(
"[Manager] skipping inbound file with unsafe filename: channel=%s, file=%r",
msg.channel_name,
filename,
)
continue
dest = uploads_dir / safe_name
try:
while True:
try:
dest = await asyncio.to_thread(write_upload_file_no_symlink, uploads_dir, safe_name, data, exclusive=True)
break
except FileExistsError:
# Another upload can claim a name after the directory
# snapshot, including while this attachment downloads.
safe_name = claim_unique_filename(safe_name, seen_names)
# Root-written 0o600 files are unreadable to the non-root
# sandbox; grant group/other read like the HTTP upload path.
await asyncio.to_thread(_make_inbound_file_sandbox_readable, dest)
except UnsafeUploadPathError:
logger.warning("[Manager] skipping inbound file with unsafe destination: %s", safe_name)
continue
except Exception:
logger.exception("[Manager] failed to write inbound file: %s", dest)
continue
created.append(
{
"filename": safe_name,
"size": len(data),
"path": f"/mnt/user-data/uploads/{safe_name}",
"is_image": ftype == "image",
}
)
return created
class ChannelManager:
"""Core dispatcher that bridges IM channels to the DeerFlow agent.
It reads from the MessageBus inbound queue, creates/reuses threads on
Gateway's LangGraph-compatible API, sends messages via ``runs.wait``, and publishes
outbound responses back through the bus.
"""
def __init__(
self,
bus: MessageBus,
store: ChannelStore,
*,
max_concurrency: int = DEFAULT_CHANNEL_MAX_CONCURRENCY,
shutdown_grace_period_seconds: float = DEFAULT_CHANNEL_SHUTDOWN_GRACE_PERIOD_SECONDS,
langgraph_url: str = DEFAULT_LANGGRAPH_URL,
gateway_url: str = DEFAULT_GATEWAY_URL,
assistant_id: str = DEFAULT_ASSISTANT_ID,
default_session: dict[str, Any] | None = None,
channel_sessions: dict[str, Any] | None = None,
connection_repo: Any | None = None,
require_bound_identity: bool = False,
inbound_dedupe_store: InboundDedupeStore | None = None,
get_stream_bridge: Callable[[], StreamBridge | None] | None = None,
) -> None:
if isinstance(max_concurrency, bool) or not isinstance(max_concurrency, int) or max_concurrency <= 0:
raise ValueError("max_concurrency must be a positive integer")
if isinstance(shutdown_grace_period_seconds, bool) or not isinstance(shutdown_grace_period_seconds, (int, float)) or not math.isfinite(shutdown_grace_period_seconds) or shutdown_grace_period_seconds < 0:
raise ValueError("shutdown_grace_period_seconds must be a non-negative finite number")
self.bus = bus
self.store = store
self._max_concurrency = max_concurrency
self._shutdown_grace_period_seconds = float(shutdown_grace_period_seconds)
self._langgraph_url = langgraph_url
self._gateway_url = gateway_url
self._assistant_id = assistant_id
self._default_session = _as_dict(default_session)
self._channel_sessions = dict(channel_sessions or {})
self._connection_repo = connection_repo
self._require_bound_identity = require_bound_identity
# Zero-arg accessor for the FastAPI app's StreamBridge singleton,
# threaded in from app.py's lifespan via start_channel_service() ->
# ChannelService.__init__ (mirrors how ScheduledTaskService gets a
# launch_run closure over `app` in the same lifespan function). None
# when not wired (e.g. a ChannelManager constructed directly in
# tests) — follow-up buffering still works, but no watcher is
# spawned to auto-drain it (see _maybe_spawn_followup_watcher).
self._get_stream_bridge = get_stream_bridge
self._client = None # lazy init — langgraph_sdk async client
self._channel_metadata_synced: set[str] = set()
# Explicit /agent selections are pinned to the newly-created thread.
# Cache the durable thread metadata so the hot path does not GET the
# same thread before every turn; None distinguishes a checked default
# thread from a thread that has not been inspected yet.
self._thread_agent_names: dict[str, str | None] = {}
# Explicit /model selections are pinned to the current thread. Same
# cache semantics as _thread_agent_names: None marks a thread that was
# checked and has no pin. Unlike agent pins (written only when a
# thread is created and immutable afterwards), model pins mutate on
# live threads, so this cache is consistent only within one process:
# a pod that cached an entry will not see pins set by other pods
# until the cache clears or the pod restarts. IM providers hold a
# single long-lived connection, so the pin writer and reader run in
# the same Gateway process in practice; revisit if pins ever need to
# be read across a shared store.
self._thread_model_names: dict[str, str | None] = {}
# Bumped on every model-cache mutation (publish or clear-on-pressure
# eviction). A cold load captures it before its in-flight threads.get
# and refetches when it moved — an eviction must not let an older
# snapshot overwrite a successfully committed pin (see _load_thread_model).
self._model_cache_generation = 0
# Serializes cold pin loads against persist+cache publication: without
# it, a slow threads.get that captured pre-pin metadata can land after
# a concurrent /model write and overwrite the fresh cache entry with
# the stale value (commands are not run-serialized).
self._model_pin_lock = asyncio.Lock()
# Waiter-aware per-conversation locks prevent concurrent inbound messages
# from creating duplicate threads. Participants are checked out before
# they wait, so failure or cancellation of the current creator cannot let
# a late caller bypass an already-queued creator through a new lock generation.
self._thread_create_locks = AsyncKeyedLockTable[tuple[str, str, str | None]]()
# Per-thread run locks for channels that want in-manager serialization
# instead of surfacing the runtime's generic busy reply.
self._serialized_thread_runs: dict[tuple[str, str], _SerializedThreadRunState] = {}
self._skill_storage: SkillStorage | None = None
self._csrf_token = generate_csrf_token()
self._running = False
# Distinct from self._running: that flag is also False before the
# very first start() (so tests that call internal drain/handler
# methods directly without going through start()/stop() keep
# working unchanged). self._stopped tracks specifically whether
# stop() has run, for the follow-up drain guard below.
self._stopped = False
# Fixed, long-lived workers own message handling end to end. The pool
# size is the only number of handler coroutines that can exist; inbound
# bursts remain as bounded queue entries instead of semaphore-waiting
# task-per-message fan-out.
self._worker_tasks: set[asyncio.Task[None]] = set()
# Inbound webhook dedupe store. Defaults to the in-process Memory store
# (pre-#4120 behavior). Multi-pod deployments inject a shared store so
# duplicate deliveries landing on different pods are collapsed.
self._inbound_dedupe_store = inbound_dedupe_store if inbound_dedupe_store is not None else MemoryInboundDedupeStore()
# Adapters claim the same key through the bus before platform side
# effects a duplicate must not repeat (Discord thread creation).
self.bus.set_inbound_claim(self._claim_inbound, self._release_inbound_claim)
# Per-thread follow-up buffers for busy fire_and_forget channels that
# opted into ChannelRunPolicy.buffer_followups_on_busy (issue #4121
# Slice 2). Keyed by thread_id -> OrderedDict[dedupe_key -> entry],
# oldest-first, mirroring the dedupe store's shape but scoped
# per-thread with a hard cap instead of a global TTL (see
# _buffer_followup / _enforce_followup_cap).
self._followup_buffers: dict[str, OrderedDict[str, _FollowupEntry]] = {}
# Background watcher tasks spawned by _maybe_spawn_followup_watcher,
# tracked so stop() can cancel+await them instead of leaving them as
# orphaned fire-and-forget tasks that could still fire a follow-up
# run after this manager has been shut down. Discarded via the same
# task's done-callback (see _maybe_spawn_followup_watcher).
self._followup_watcher_tasks: set[asyncio.Task] = set()
@staticmethod
def _channel_supports_streaming(channel_name: str) -> bool:
from .service import get_channel_service
service = get_channel_service()
if service:
channel = service.get_channel(channel_name)
if channel is not None:
return channel.supports_streaming
return CHANNEL_CAPABILITIES.get(channel_name, {}).get("supports_streaming", False)
def _resolve_session_layer(self, msg: InboundMessage) -> tuple[dict[str, Any], dict[str, Any]]:
channel_layer = _as_dict(self._channel_sessions.get(msg.channel_name))
users_layer = _as_dict(channel_layer.get("users"))
user_layer = _as_dict(users_layer.get(msg.user_id))
return channel_layer, user_layer
def _resolve_message_assistant_id(self, msg: InboundMessage) -> str | None:
"""Per-message agent override (e.g. GitHub webhook fan-out: multiple
agents may bind the same repo, each gets its own inbound message with
its own agent_name in metadata). Honors the same shape as channel/user
session config: the bare agent name routes through the lead_agent +
agent_name context pattern in ``_resolve_run_params``."""
msg_metadata = msg.metadata if isinstance(msg.metadata, dict) else {}
meta_assistant_id = msg_metadata.get("assistant_id") or msg_metadata.get("agent_name")
if isinstance(meta_assistant_id, str) and meta_assistant_id.strip():
return meta_assistant_id
return None
def _resolve_assistant_id(
self,
*,
message_assistant_id: str | None,
thread_assistant_id: str | None,
user_layer: Mapping,
channel_layer: Mapping,
) -> str:
"""Resolve the assistant a run on this thread would use.
Single source of truth for assistant precedence, shared by the run
path and the ``/model`` status reply: per-message override, thread
pin, then session layers (user > channel > default), falling back
to the manager default.
"""
assistant_id = message_assistant_id or thread_assistant_id or user_layer.get("assistant_id") or channel_layer.get("assistant_id") or self._default_session.get("assistant_id") or self._assistant_id
if not isinstance(assistant_id, str) or not assistant_id.strip():
assistant_id = self._assistant_id
return assistant_id
def _begin_serialized_thread_run(
self,
*,
channel_name: str,
thread_id: str,
) -> tuple[_SerializedThreadRunState | None, bool]:
policy = CHANNEL_RUN_POLICY.get(channel_name)
if policy is None or not policy.serialize_thread_runs:
return None, False
key = (channel_name, thread_id)
state = self._serialized_thread_runs.get(key)
if state is None:
state = _SerializedThreadRunState(lock=asyncio.Lock())
self._serialized_thread_runs[key] = state
queued = state.lock.locked()
state.waiters += 1
return state, queued
def _finish_serialized_thread_run(
self,
*,
channel_name: str,
thread_id: str,
state: _SerializedThreadRunState | None,
lock_acquired: bool,
) -> None:
if state is None:
return
if lock_acquired:
state.lock.release()
state.waiters -= 1
if state.waiters == 0 and not state.lock.locked():
self._serialized_thread_runs.pop((channel_name, thread_id), None)
# -- follow-up buffering for busy fire_and_forget threads (issue #4121) --
def _resolve_stream_bridge(self) -> StreamBridge | None:
"""Resolve the current StreamBridge via the injected accessor, if any."""
if self._get_stream_bridge is None:
return None
try:
return self._get_stream_bridge()
except Exception:
logger.exception("[Manager] get_stream_bridge callable raised; follow-up watch disabled for this run")
return None
def _enforce_followup_cap(self, thread_id: str, buffer: OrderedDict[str, _FollowupEntry]) -> None:
"""Drop the OLDEST buffered entries once *buffer* exceeds the per-thread cap.
Dropping the oldest (rather than the newest, incoming) entry means a
thread that is deep enough in the backlog to hit the cap still keeps
the most recent activity — a better signal for the eventual coalesced
turn than the stalest queued comment. No reaction/acknowledgment is
sent on drop (out of scope for this slice); a WARNING is logged so
operators can see it in gateway.log.
"""
while len(buffer) > FOLLOWUP_BUFFER_MAX_PER_THREAD:
dropped_key, _ = buffer.popitem(last=False)
logger.warning(
"[Manager] follow-up buffer overflow for thread_id=%s (cap=%d); dropped oldest buffered comment (dedupe_key=%s)",
thread_id,
FOLLOWUP_BUFFER_MAX_PER_THREAD,
dropped_key,
)
def _buffer_followup(self, thread_id: str, msg: InboundMessage) -> None:
"""Append *msg* to thread_id's follow-up buffer (ConflictError path).
Dedupe mirrors ``_is_duplicate_inbound``'s OrderedDict idiom, scoped
per-thread instead of global: a redelivered webhook for a comment
already buffered (same dedupe key) is a no-op instead of a second
entry.
"""
key = _followup_dedupe_key(msg)
buffer = self._followup_buffers.setdefault(thread_id, OrderedDict())
if key in buffer:
logger.info(
"[Manager] duplicate follow-up ignored for thread_id=%s (dedupe_key=%s)",
thread_id,
key,
)
return
buffer[key] = _FollowupEntry(dedupe_key=key, text=msg.text)
self._enforce_followup_cap(thread_id, buffer)
logger.info(
"[Manager] buffered follow-up for busy thread_id=%s (dedupe_key=%s, buffered=%d)",
thread_id,
key,
len(buffer),
)
def _pop_followup_batch(self, thread_id: str, *, limit: int) -> list[_FollowupEntry]:
"""Pop up to *limit* buffered entries FIFO (oldest first)."""
buffer = self._followup_buffers.get(thread_id)
if not buffer:
return []
batch: list[_FollowupEntry] = []
for _ in range(min(limit, len(buffer))):
_, entry = buffer.popitem(last=False)
batch.append(entry)
if not buffer:
self._followup_buffers.pop(thread_id, None)
return batch
def _requeue_followups(self, thread_id: str, entries: list[_FollowupEntry]) -> None:
"""Put a popped batch back at the front of the buffer (oldest-first).
Used when the drain's own ``runs.create`` call itself fails —
including the ``ConflictError`` edge case where something this
manager did not create (a manual Web UI turn, a scheduled run) is
occupying the thread. The entries are not lost: the next time *any*
run this manager creates on this thread completes, its watcher will
attempt another drain and find them still buffered.
"""
if not entries:
return
existing = self._followup_buffers.get(thread_id, OrderedDict())
merged: OrderedDict[str, _FollowupEntry] = OrderedDict()
for entry in entries:
merged[entry.dedupe_key] = entry
for key, entry in existing.items():
merged.setdefault(key, entry)
self._enforce_followup_cap(thread_id, merged)
self._followup_buffers[thread_id] = merged
def _maybe_spawn_followup_watcher(
self,
thread_id: str,
run_result: Any,
carrier_msg: InboundMessage,
) -> None:
"""Spawn a background watcher for a just-created run, if wired up.
No-ops (spawns nothing) when no ``get_stream_bridge`` accessor was
threaded in — e.g. a ``ChannelManager`` constructed directly without
going through ``start_channel_service()`` — so tests and any
not-yet-wired deployment never see a dangling background task for
this. When wired, mirrors the worker-task error-reporting pattern:
``asyncio.create_task`` + ``add_done_callback(self._log_task_error)``
so an unexpected watcher failure is surfaced in the logs instead of
silently vanishing. The task is also tracked in
``self._followup_watcher_tasks`` (discarded via its own done-callback)
so ``stop()`` can cancel+await any watcher still in flight instead of
leaving it to fire a follow-up run after shutdown.
"""
if self._get_stream_bridge is None or self._stopped:
return
run_id = run_result.get("run_id") if isinstance(run_result, dict) else None
if not run_id:
logger.warning(
"[Manager] runs.create returned no run_id for thread_id=%s; cannot watch for follow-up drain",
thread_id,
)
return
task = asyncio.create_task(self._watch_run_and_drain_followups(thread_id, run_id, carrier_msg))
self._followup_watcher_tasks.add(task)
task.add_done_callback(self._followup_watcher_tasks.discard)
task.add_done_callback(self._log_task_error)
async def _watch_run_and_drain_followups(
self,
thread_id: str,
run_id: str,
carrier_msg: InboundMessage,
) -> None:
"""Watch *run_id* until it ends, then attempt to drain thread_id's buffer.
Subscribes to the StreamBridge the same way existing consumers do
(``entry is END_SENTINEL``, see ``app/gateway/services.py``). Runs
for as long as the underlying run does — GitHub coding runs
routinely take several minutes, so this deliberately does not apply
an artificial timeout, mirroring why the dispatch path itself uses
``runs.create`` instead of ``runs.wait`` in the first place. Draining
is a no-op when the buffer is empty, which is the common case (most
runs never hit a busy-thread conflict).
"""
stream_bridge = self._resolve_stream_bridge()
if stream_bridge is None:
logger.warning(
"[Manager] no stream bridge available; cannot watch run_id=%s for thread_id=%s follow-up drain (any buffered follow-ups will be drained by a later watched run on this thread)",
run_id,
thread_id,
)
return
try:
async for entry in stream_bridge.subscribe(run_id):
if entry is END_SENTINEL:
break
except Exception:
logger.exception(
"[Manager] error watching run_id=%s for thread_id=%s follow-up drain",
run_id,
thread_id,
)
return
client = self._get_client()
await self._drain_followups_for_thread(client, thread_id, carrier_msg)
async def _drain_followups_for_thread(
self,
client,
thread_id: str,
carrier_msg: InboundMessage,
) -> None:
"""Coalesce up to one batch of buffered follow-ups into a fresh run.
``carrier_msg`` supplies routing/policy identity (channel_name,
metadata, owner headers) for the drained run — it is safe to reuse
across an entire drain chain because every buffered entry for one
thread_id shares that identity (thread_id itself is derived
deterministically from (repo, number, agent_name) for GitHub).
A batch larger than ``FOLLOWUP_DRAIN_BATCH_SIZE`` is intentionally
NOT drained in one shot: only the oldest batch is popped here, and
the run created for it is itself watched (via
``_maybe_spawn_followup_watcher``), so a deeper backlog chains into
another drain cycle once this run ends, rather than growing one
unbounded coalesced input block.
If anything from here through ``runs.create`` fails — resolving run
params, applying channel policy, or ``runs.create`` itself
(including ``ConflictError`` from something this manager did not
create racing onto the same thread) — the popped batch is requeued
(not lost) and this coroutine returns without raising or looping:
the next run this manager successfully creates and watches on this
thread will attempt the drain again.
No-ops if the manager has already been ``stop()``-ped: a watcher
task that slips past its own cancellation and reaches this point
after shutdown must not fire a brand new run into a stopped manager.
(Deliberately keyed on ``self._stopped``, not ``self._running`` —
the latter is also ``False`` before the very first ``start()``,
which would otherwise make this guard fire for callers that invoke
the drain directly without going through the dispatch lifecycle.)
"""
if self._stopped:
logger.info(
"[Manager] skipping follow-up drain for thread_id=%s; manager is stopped",
thread_id,
)
return
entries = self._pop_followup_batch(thread_id, limit=FOLLOWUP_DRAIN_BATCH_SIZE)
if not entries:
return
logger.info(
"[Manager] draining %d buffered follow-up(s) for thread_id=%s",
len(entries),
thread_id,
)
# Initialized before the try: ``_resolve_run_params`` itself is covered
# by the except below, and a failure there has no submitted selection
# for stale-pin recovery to compare (``expected_pin=None`` short-circuits
# it), so the batch simply requeues as before.
expected_pin: str | None = None
try:
# Everything from here through runs.create is covered by the
# same except below: a pre-create failure (e.g. the target agent
# config was removed mid-run, or channel-policy/credential
# resolution raises) must requeue the popped batch exactly like
# a runs.create failure does — none of these steps get to
# silently drop entries that were already popped off the buffer.
assistant_id, run_config, run_context = self._resolve_run_params(carrier_msg, thread_id)
# Same stale-pin contract as ``_handle_chat_on_thread``: the drained
# run submits the pinned model too, so an allowlist rejection here
# must clear the pin instead of requeueing the batch into the same
# failure on every drain cycle.
pinned_model_value = run_context.get("model_name")
expected_pin = pinned_model_value.strip() if isinstance(pinned_model_value, str) and pinned_model_value.strip() else None
await self._apply_channel_policy(carrier_msg, run_context)
human_message = _human_input_message(_format_followup_block(entries))
run_kwargs: dict[str, Any] = {
"input": {"messages": [human_message]},
"config": run_config,
"context": run_context,
"multitask_strategy": "reject",
}
_apply_run_origin(run_kwargs, carrier_msg)
if owner_headers := _owner_headers(carrier_msg):
run_kwargs["headers"] = owner_headers
result = await client.runs.create(thread_id, assistant_id, **run_kwargs)
except Exception as exc:
if _is_thread_busy_error(exc):
logger.warning(
"[Manager] follow-up drain hit a busy thread_id=%s (a run this manager did not create is active); re-buffering %d entries",
thread_id,
len(entries),
)
elif recovery := await self._stale_pin_recovery(carrier_msg, exc, expected_pin=expected_pin, thread_id=thread_id):
# Log structured facts only: the recovery text ends with a
# "please resend" note meant for chat replies, but the buffered
# messages are retried automatically — echoing it here would
# invite double-processing. GitHub, the only buffering channel
# today, delivers outbound sends log-only anyway.
if recovery.resolved:
# Pin cleared, or a newer selection already current: the
# requeued batch goes out on the next cycle. The verb
# stays neutral — ``resolved`` covers both sub-cases, and
# in the replacement case nothing was actually cleared.
logger.info(
"[Manager] follow-up drain recovered from stale model %r (thread_id=%s); re-buffering %d entries for automatic retry",
expected_pin,
thread_id,
len(entries),
)
else:
# Nothing was cleared — the durable reset failed, or the
# selection came from session configuration, which still
# carries it — so the next drain cycle rejects identically.
# Keep an operator-visible signal instead of a per-cycle
# INFO line; the batch is still requeued, never dropped.
logger.warning(
"[Manager] follow-up drain could not self-recover from stale model %r (thread_id=%s); re-buffering %d entries into a repeated admission failure",
expected_pin,
thread_id,
len(entries),
)
else:
logger.exception(
"[Manager] follow-up drain failed for thread_id=%s; re-buffering %d entries",
thread_id,
len(entries),
)
self._requeue_followups(thread_id, entries)
return
self._maybe_spawn_followup_watcher(thread_id, result, carrier_msg)
async def _publish_progress_update(self, msg: InboundMessage, thread_id: str, text: str) -> None:
await self.bus.publish_outbound(
OutboundMessage(
channel_name=msg.channel_name,
chat_id=msg.chat_id,
thread_id=thread_id,
text=text,
is_final=False,
thread_ts=msg.thread_ts,
connection_id=msg.connection_id,
owner_user_id=msg.owner_user_id,
metadata=_response_metadata(msg.metadata),
)
)
def _resolve_run_params(self, msg: InboundMessage, thread_id: str) -> tuple[str, dict[str, Any], dict[str, Any]]:
channel_layer, user_layer = self._resolve_session_layer(msg)
# Per-message agent override and the assistant precedence chain live
# in _resolve_message_assistant_id / _resolve_assistant_id, shared
# with the /model status reply so both paths resolve the same agent.
message_assistant_id = self._resolve_message_assistant_id(msg)
thread_assistant_id = self._thread_agent_names.get(thread_id)
assistant_id = self._resolve_assistant_id(
message_assistant_id=message_assistant_id,
thread_assistant_id=thread_assistant_id,
user_layer=user_layer,
channel_layer=channel_layer,
)
# Session layer merge order (default -> channel -> user, later wins)
# feeds _select_carried_model_name, the single carrier-precedence
# rule shared with the /model status reply
# (_resolve_configured_model_name). Keep both in sync when
# reordering layers.
run_config = _merge_dicts(
DEFAULT_RUN_CONFIG,
self._default_session.get("config"),
channel_layer.get("config"),
user_layer.get("config"),
)
configurable = run_config.get("configurable")
if isinstance(configurable, Mapping):
configurable = dict(configurable)
else:
configurable = {}
run_config["configurable"] = configurable
# Pin channel-triggered runs to the root graph namespace so follow-up
# turns continue from the same conversation checkpoint.
configurable["checkpoint_ns"] = ""
configurable["thread_id"] = thread_id
# ``user_id`` drives DeerFlow-owned memory, files, and thread buckets.
# For browser-connected IM channels, prefer the DeerFlow account that
# owns the connection. Preserve the raw platform user under
# ``channel_user_id`` for platform-facing lookups and audits.
run_context_identity: dict[str, Any] = {"thread_id": thread_id}
# ``channel_name`` lets in-graph code (e.g. ``_make_lead_agent``)
# decide whether a tool is safe to expose for this run. Webhook
# channels carry untrusted external prompts (GitHub comments,
# Telegram chats from non-owners, etc.), so admin-shaped tools
# like ``update_agent`` are dropped when the run was triggered
# via one. See ``_make_lead_agent`` for the gate.
run_context_identity["channel_name"] = msg.channel_name
# Single source of truth for the run identity: the same helper that scopes
# inbound files and outbound artifacts, so the bucket the agent reads/writes
# always matches where channel files are staged.
run_user_id = _channel_storage_user_id(msg)
if run_user_id:
run_context_identity["user_id"] = run_user_id
if msg.user_id:
run_context_identity["channel_user_id"] = msg.user_id
run_context = _merge_dicts(
DEFAULT_RUN_CONTEXT,
self._default_session.get("context"),
channel_layer.get("context"),
user_layer.get("context"),
run_context_identity,
)
# An explicit /model pin wins over every session-config layer. A
# custom agent's ``model`` still loses to it in the agent factory
# (``_resolve_model_name`` prefers the requested name), keeping
# explicit choices ahead of implicit configuration.
pinned_model_name = self._thread_model_names.get(thread_id)
if pinned_model_name:
_apply_explicit_model_choice(run_config, run_context, pinned_model_name)
explicit_agent_choice = message_assistant_id is not None or thread_assistant_id is not None
# Custom agents are implemented as lead_agent + agent_name context.
# Keep backward compatibility for channel configs that set
# assistant_id: <custom-agent-name> by routing through lead_agent.
if assistant_id != DEFAULT_ASSISTANT_ID:
normalized_agent_name = _normalize_custom_agent_name(assistant_id)
if explicit_agent_choice:
_apply_explicit_agent_choice(run_config, run_context, normalized_agent_name)
else:
run_context.setdefault("agent_name", normalized_agent_name)
assistant_id = DEFAULT_ASSISTANT_ID
elif explicit_agent_choice:
# An explicit lead_agent selection is also a real pin: discard a
# configured agent in every Gateway-supported carrier so
# /agent use lead_agent cannot claim to reset the conversation
# while silently routing elsewhere.
_apply_explicit_agent_choice(run_config, run_context, None)
# Apply per-channel run policy (recursion_limit bump for webhook
# channels, etc.). Looking the policy up by channel_name keeps
# GitHub-specific knobs out of this method — adding the next
# webhook channel is a one-row CHANNEL_RUN_POLICY entry, not a
# new if-branch here.
policy = CHANNEL_RUN_POLICY.get(msg.channel_name)
if policy is not None and policy.default_recursion_limit is not None:
# Per-message override (via msg.metadata[channel_name]) honors
# the operator's explicit per-agent recursion_limit verbatim —
# including values below the channel default. A safety-conscious
# ``github.recursion_limit: 50`` on a review-only agent now halts
# at 50 super-steps as documented in GitHubAgentConfig, instead
# of being silently clamped up to the channel default. When no
# override is present, the channel default acts as a floor over
# whatever session config supplied (the higher value wins).
channel_meta = (msg.metadata or {}).get(msg.channel_name, {})
override = channel_meta.get("recursion_limit") if isinstance(channel_meta, dict) else None
if isinstance(override, int) and override > 0:
run_config["recursion_limit"] = override
else:
run_config["recursion_limit"] = max(run_config.get("recursion_limit", 100), policy.default_recursion_limit)
return assistant_id, run_config, run_context
async def _apply_channel_policy(self, msg: InboundMessage, run_context: dict[str, Any]) -> ChannelRunPolicy | None:
"""Apply per-channel run policy that needs ``run_context`` access.
Run AFTER ``_resolve_run_params`` (which produced ``run_context``)
and BEFORE the agent runs. Covers:
* ``disable_clarification`` for non-interactive channels —
``ClarificationMiddleware`` would otherwise dead-end a webhook
run waiting for a synchronous reply that only arrives as a
later, separate webhook delivery.
* Channel-specific credentials provider — e.g. the GitHub channel
installs a token-mint callable so ``bash_tool`` can resolve a
fresh installation token on every invocation (longer than the
1h GitHub TTL).
``recursion_limit`` is applied inside :meth:`_resolve_run_params`
instead because it lives on ``run_config`` (not ``run_context``)
and the resolver already builds ``run_config``.
Returns the resolved :class:`ChannelRunPolicy` (or ``None`` when
the channel has no entry) so :meth:`_handle_chat` can branch on
flags like ``fire_and_forget`` without doing a second dict
lookup.
"""
policy = CHANNEL_RUN_POLICY.get(msg.channel_name)
if policy is None:
return None
if policy.interaction_mode is not None:
run_context["interaction_mode"] = policy.interaction_mode
# Keep legacy consumers (including sandbox network approval) aligned.
if policy.interaction_mode != "interactive":
run_context["disable_clarification"] = True
elif not policy.is_interactive:
run_context["disable_clarification"] = True
if policy.credentials_provider is not None:
try:
await policy.credentials_provider(msg, run_context)
except Exception:
# Credential failures must NOT drop the delivery — the
# provider's own logging records the cause; we keep the
# run going (read-only is better than no response).
logger.warning(
"[Manager] channel=%s credentials_provider raised; run proceeds without injected credentials",
msg.channel_name,
exc_info=True,
)
return policy
def _resolve_available_skill_names(self, msg: InboundMessage, thread_id: str) -> set[str] | None:
"""*thread_id* comes from ``_lookup_thread_id`` (``""`` when unmapped); never
re-read the JSON store here, which holds no mapping for bound messages."""
_, _, run_context = self._resolve_run_params(msg, thread_id)
if run_context.get("is_bootstrap"):
return {"bootstrap"}
agent_name = run_context.get("agent_name")
if not isinstance(agent_name, str) or not agent_name.strip():
return None
# Read the agent config from the same owner bucket the run uses:
# ``run_context["user_id"]`` is the resolved owner (``_channel_storage_user_id``),
# but without it ``load_agent_config`` falls back to the dispatch loop's unset
# contextvar (``"default"``), reading the wrong user's per-user custom agent.
agent_config = load_agent_config(_normalize_custom_agent_name(agent_name), user_id=run_context.get("user_id"))
if agent_config and agent_config.skills is not None:
return set(agent_config.skills)
return None
# -- LangGraph SDK client (lazy) ----------------------------------------
def _csrf_headers(self) -> dict[str, str]:
"""Return the double-submit CSRF pair for state-changing Gateway calls.
Internal auth does not exempt a request from ``CSRFMiddleware``, so every
Gateway mutation the manager sends must carry this pair.
"""
return {
CSRF_HEADER_NAME: self._csrf_token,
"Cookie": f"{CSRF_COOKIE_NAME}={self._csrf_token}",
}
def _get_client(self):
"""Return the ``langgraph_sdk`` async client, creating it on first use."""
if self._client is None:
from langgraph_sdk import get_client
self._client = get_client(
url=self._langgraph_url,
headers={**create_internal_auth_headers(), **self._csrf_headers()},
)
return self._client
def _get_skill_storage(self) -> SkillStorage:
if self._skill_storage is None:
self._skill_storage = get_or_new_skill_storage()
return self._skill_storage
# -- lifecycle ---------------------------------------------------------
async def start(self) -> None:
"""Open inbound admission and start the fixed message-worker pool."""
if self._running:
return
unfinished_workers = {task for task in self._worker_tasks if not task.done()}
if unfinished_workers:
raise RuntimeError("cannot restart ChannelManager while workers are still running")
self._worker_tasks.clear()
self._running = True
self._stopped = False
self.bus.open_inbound()
self._worker_tasks = {
asyncio.create_task(
self._worker_loop(worker_index),
name=f"deerflow-channel-worker-{worker_index}",
)
for worker_index in range(self._max_concurrency)
}
for task in self._worker_tasks:
task.add_done_callback(self._log_task_error)
task.add_done_callback(self._worker_tasks.discard)
logger.info(
"ChannelManager started (workers=%d, inbound_queue_maxsize=%d, shutdown_grace=%.1fs)",
self._max_concurrency,
self.bus.inbound_queue_maxsize,
self._shutdown_grace_period_seconds,
)
def begin_shutdown(self) -> int:
"""Close admission while leaving workers alive to drain accepted work."""
self._running = False
self._stopped = True
return self.bus.close_inbound()
async def stop(self) -> None:
"""Drain accepted work, then cancel and await every owned task.
Admission closes immediately, but workers keep processing the accepted
queue for ``shutdown_grace_period_seconds`` so provider acknowledgments
are normally followed by a final response. Once that grace expires,
workers are cancelled. A successful return means no worker or follow-up
watcher can continue using channel resources. The Gateway lifespan owns
the process-level timeout; if it cancels this coroutine, transports stay
attached to the service and shutdown can be retried.
"""
invalidated_reservations = self.begin_shutdown()
loop = asyncio.get_running_loop()
grace_deadline = loop.time() + self._shutdown_grace_period_seconds
worker_tasks = list(self._worker_tasks)
watcher_tasks = list(self._followup_watcher_tasks)
# Watchers can create follow-up runs and are not acknowledgments for
# this accepted queue. Stop them immediately; active workers are still
# allowed to finish their current/queued messages during the grace.
for task in watcher_tasks:
task.cancel()
join_task = asyncio.create_task(self.bus.join_inbound(), name="deerflow-channel-inbound-drain")
drained = False
try:
done, _ = await asyncio.wait({join_task}, timeout=max(0.0, grace_deadline - loop.time()))
drained = join_task in done and not join_task.cancelled() and join_task.exception() is None
except asyncio.CancelledError:
for task in worker_tasks:
task.cancel()
discarded_messages = await self._discard_pending_inbound()
if not join_task.done():
join_task.cancel()
if invalidated_reservations or discarded_messages:
logger.warning(
"[Manager] cancelled shutdown rejected pending inbound work: reservations=%d, queued_messages=%d",
invalidated_reservations,
discarded_messages,
)
raise
if not drained:
logger.warning(
"[Manager] graceful shutdown period expired after %.1fs; cancelling active handlers",
self._shutdown_grace_period_seconds,
)
for task in worker_tasks:
task.cancel()
# Anything still queued never started and must not keep Queue.join()
# pending after its workers have been cancelled.
discarded_messages = await self._discard_pending_inbound()
if not join_task.done():
join_task.cancel()
owned_tasks = {task for task in (*worker_tasks, *watcher_tasks, join_task) if not task.done()}
if owned_tasks:
await asyncio.gather(*owned_tasks, return_exceptions=True)
for task in tuple(self._worker_tasks):
if task.done():
self._worker_tasks.discard(task)
for task in tuple(self._followup_watcher_tasks):
if task.done():
self._followup_watcher_tasks.discard(task)
if invalidated_reservations or discarded_messages:
logger.warning(
"[Manager] shutdown rejected pending inbound work: reservations=%d, queued_messages=%d",
invalidated_reservations,
discarded_messages,
)
logger.info("ChannelManager stopped")
async def _discard_pending_inbound(self) -> int:
"""Release attachments, and adapter dedupe claims, of messages shutdown never dispatched.
An adapter-claimed message (``dedupe_claimed``) recorded its dedupe key
before it was queued; dropped unprocessed, it must give the key back so
a redelivery on a peer is not refused for the rest of the TTL.
"""
discarded = 0
while True:
try:
msg = self.bus.get_inbound_nowait()
except asyncio.QueueEmpty:
return discarded
try:
await cleanup_inbound_files(msg.files)
if msg.dedupe_claimed:
try:
await self._release_inbound_dedupe_key(msg)
except Exception:
logger.exception("[Manager] failed to release the dedupe claim of a discarded inbound message")
finally:
self.bus.inbound_task_done()
discarded += 1
# -- worker pool -------------------------------------------------------
async def _worker_loop(self, worker_index: int) -> None:
logger.info("[Manager] inbound worker %d started", worker_index)
# Closing admission flips ``_running`` to False, but queued messages
# were already accepted (and providers may already have acknowledged
# them). Keep draining until the queue is empty; stop() then cancels
# workers that are idle in get_inbound().
while self._running or not self.bus.inbound_queue.empty():
try:
msg = await self.bus.get_inbound()
except asyncio.CancelledError:
raise
# Inbound IM messages are a non-HTTP entry point: channels hold
# long-lived provider connections, so no ASGI middleware ever runs
# for them. Scope one trace id per message here -- the worker task
# is long-lived and reused, so the scope must close with the
# message rather than leak into the next one.
with ensure_trace_context(), _run_admission_scope(fresh=True):
dedupe_recorded = False
try:
# Dedupe before logging "received" so a provider retrying an
# event N times does not log N accepts. Provider ack side
# effects may still happen before this manager-level dedupe,
# unless the adapter claimed the key ahead of them.
if not msg.dedupe_claimed and await self._is_duplicate_inbound(msg):
continue
dedupe_recorded = self._inbound_dedupe_key(msg) is not None
logger.info(
"[Manager] received inbound: channel=%s, chat_id=%s, type=%s, text_len=%d, files=%d",
msg.channel_name,
msg.chat_id,
msg.msg_type.value,
len(msg.text or ""),
len(msg.files),
)
# Deliberately awaited inline: never create a task per message.
await self._handle_message(msg)
except asyncio.CancelledError as exc:
# A cancellation after dedupe admission must make provider
# redelivery retryable rather than retaining a TTL-long key for
# work that never completed -- unless the Gateway may already
# be executing the run, which outlives this worker.
if dedupe_recorded and not self._keep_dedupe_key_after(msg, exc):
try:
await self._release_inbound_dedupe_key(msg)
except Exception:
logger.exception("[Manager] failed to release inbound dedupe key during worker cancellation")
raise
except Exception as exc:
logger.exception(
"[Manager] inbound worker %d failed handling channel=%s chat_id=%s",
worker_index,
msg.channel_name,
msg.chat_id,
)
if dedupe_recorded and not self._keep_dedupe_key_after(msg, exc):
try:
await self._release_inbound_dedupe_key(msg)
except Exception:
# A dedupe backend outage must not shrink the fixed
# worker pool by letting cleanup escape this loop.
logger.exception("[Manager] failed to release inbound dedupe key after worker error")
finally:
try:
await cleanup_inbound_files(msg.files)
finally:
self.bus.inbound_task_done()
@staticmethod
def _inbound_dedupe_key(msg: InboundMessage) -> tuple[str, str, str, str] | None:
metadata = msg.metadata or {}
message_id = None
for key in INBOUND_DEDUPE_METADATA_KEYS:
value = metadata.get(key)
if value:
message_id = str(value)
break
if message_id is None:
raw_message = metadata.get("raw_message")
if isinstance(raw_message, Mapping):
for key in INBOUND_DEDUPE_METADATA_KEYS:
value = raw_message.get(key)
if value:
message_id = str(value)
break
if message_id is None:
return None
if msg.channel_name == "discord":
# Message ids are global snowflakes. ``channel_id`` is where the
# message was posted, which every instance sees the same way;
# ``chat_id`` depends on this instance's thread tracking (a reply
# in an untracked thread is routed as an orphan). A direct message
# has no guild.
return ("discord", str(metadata.get("guild_id") or "dm"), str(metadata.get("channel_id") or msg.chat_id), message_id)
# Fail closed: without a workspace/team/guild identifier we cannot tell two
# workspaces apart (e.g. Slack channel ids are not globally unique), so
# skip dedupe rather than risk collapsing distinct workspaces' messages.
# Both fallbacks are appended last and gated on every earlier source being
# absent, so they can only turn "no key" into a key — never change one.
# A conversation_id not reused across a provider's own redelivery would
# degrade to today's no-dedupe behaviour, never collapse two conversations
# (chat_id and message_id stay in the tuple). conversation_id covers
# DingTalk (group + P2P); chat-scoped providers fall back to chat_id.
workspace_id = msg.workspace_id or metadata.get("workspace_id") or metadata.get("team_id") or metadata.get("guild_id") or metadata.get("aibotid") or metadata.get("conversation_id")
if not workspace_id and msg.channel_name in CHAT_SCOPED_WORKSPACE_CHANNELS:
workspace_id = msg.chat_id or None
if not workspace_id:
return None
return (msg.channel_name, str(workspace_id), msg.chat_id, message_id)
@staticmethod
def _run_idempotency_key(msg: InboundMessage) -> str | None:
"""``Idempotency-Key`` for *msg*'s run, or ``None`` when it has no dedupe key."""
key = ChannelManager._inbound_dedupe_key(msg)
if key is None:
return None
digest = hashlib.sha256(json.dumps(list(key), ensure_ascii=False, separators=(",", ":")).encode()).hexdigest()
return f"{_RUN_IDEMPOTENCY_KEY_PREFIX}{digest}"
@staticmethod
def _human_message_id(msg: InboundMessage) -> str | None:
"""Human message id stable across redeliveries of *msg*, or ``None`` when unkeyed."""
idempotency_key = ChannelManager._run_idempotency_key(msg)
return str(uuid.uuid5(_HUMAN_MESSAGE_ID_NAMESPACE, idempotency_key)) if idempotency_key else None
async def _claim_inbound(self, msg: InboundMessage) -> bool:
"""Record *msg*'s dedupe key before the adapter's platform side effects.
Installed as the bus's inbound claim. ``False`` means another delivery
(a peer instance's session, an earlier redelivery) owns the message.
Once recorded, ``msg.dedupe_claimed`` tells the worker not to record
it again; the adapter copies the flag onto the message it publishes.
"""
if self._inbound_dedupe_key(msg) is None:
return True
if await self._is_duplicate_inbound(msg):
return False
msg.dedupe_claimed = True
return True
async def _release_inbound_claim(self, msg: InboundMessage) -> None:
"""Undo ``_claim_inbound`` for a message the adapter never handed off."""
if msg.dedupe_claimed:
await self._release_inbound_dedupe_key(msg)
async def _is_duplicate_inbound(self, msg: InboundMessage) -> bool:
key = self._inbound_dedupe_key(msg)
if key is None:
return False
# Delegated to the shared/per-pod dedupe store. The store owns TTL eviction
# and capacity bounds; try_record returns True when the key was already
# present (i.e. this is a duplicate delivery to drop).
is_duplicate = await self._inbound_dedupe_store.try_record(key)
if is_duplicate:
logger.info(
"[Manager] duplicate inbound ignored: channel=%s, chat_id=%s, message_id=%s",
msg.channel_name,
msg.chat_id,
key[-1],
)
return is_duplicate
async def _release_inbound_dedupe_key(self, msg: InboundMessage) -> None:
"""Drop a recorded dedupe key so a provider redelivery can be reprocessed.
Called only on transient/unexpected handling failures: the key was
recorded on receipt so retries arriving *while* the message is being
handled are still deduped, but if handling fails we must not turn a
recoverable error into a TTL-long black hole for the same message_id.
Generic failure paths ask ``_keep_dedupe_key_after`` first; sites that
know the Gateway rejected the run (busy thread, stale pin) release
unconditionally.
"""
key = self._inbound_dedupe_key(msg)
if key is not None:
await self._inbound_dedupe_store.release(key)
@staticmethod
def _keep_dedupe_key_after(msg: InboundMessage, exc: BaseException | None) -> bool:
"""Whether *msg*'s dedupe key must outlive a failed or cancelled handling.
Runs are created with ``on_disconnect: continue``, so once the Gateway
may have admitted this message's run, the run outlives the request and
a provider redelivery must stay a duplicate: a second run would repeat
the first one's side effects. The key is released, so the redelivery
can retry, only while the manager still owns the outcome: before the
run request leaves it, when the Gateway provably rejected the request
(an HTTP status error, a connection that never opened), or after a
terminal ``error`` frame ended the run. ``/wait`` confirms admission
only with its final response, so a cancellation or a post-request
transport failure while it is in flight counts as admitted rather than
risking a duplicate run.
"""
admission = _RUN_ADMISSION.get()
if admission is None or not admission.requested or isinstance(exc, _RunFailedFrame):
return False
if admission.confirmed or isinstance(exc, (asyncio.CancelledError, *_RUN_REQUEST_REACHED_GATEWAY_ERRORS)):
logger.info(
"[Manager] keeping inbound dedupe key, the run may continue on the Gateway: channel=%s, chat_id=%s, cause=%s",
msg.channel_name,
msg.chat_id,
type(exc).__name__ if exc is not None else "completed",
)
return True
return False
@staticmethod
def _log_task_error(task: asyncio.Task) -> None:
"""Surface unhandled exceptions from background tasks."""
if task.cancelled():
return
exc = task.exception()
if exc:
logger.error("[Manager] unhandled error in message task: %s", exc, exc_info=exc)
async def _handle_message(self, msg: InboundMessage) -> None:
try:
with _run_admission_scope():
await self._dispatch_message(msg)
finally:
await cleanup_inbound_files(msg.files)
async def _dispatch_message(self, msg: InboundMessage) -> None:
msg = _apply_effective_owner(msg)
try:
# Commands are handled below because provider adapters consume
# binding commands before manager dispatch, and _handle_command()
# applies its own admission gate for manager-level commands.
bound_identity_rejection = None
if msg.msg_type != InboundMessageType.COMMAND:
bound_identity_rejection = await self._get_bound_identity_rejection(msg)
if bound_identity_rejection is not None:
await self._reject_unbound_channel_message(msg, bound_identity_rejection=bound_identity_rejection)
return
if msg.msg_type == InboundMessageType.COMMAND:
await self._handle_command(msg)
else:
await self._handle_chat(msg, bound_identity_checked=True)
except InvalidChannelSessionConfigError as exc:
logger.warning(
"Invalid channel session config for %s (chat=%s): %s",
msg.channel_name,
msg.chat_id,
exc,
)
await self._send_error(msg, str(exc))
except SlashSkillCommandResolutionError as exc:
logger.warning(
"Slash skill command resolution failed for %s (chat=%s): %s",
msg.channel_name,
msg.chat_id,
exc,
)
await self._send_error(msg, str(exc))
except Exception as exc:
logger.exception(
"Error handling message from %s (chat=%s)",
msg.channel_name,
msg.chat_id,
)
# Transient/unexpected failure: release the dedupe key so a provider
# redelivery of the same message can recover instead of being dropped
# for the dedupe TTL -- unless the run may already be executing.
# Stale /model pin admission failures are handled at the
# run-creation sites, which know the selection the run used.
if not self._keep_dedupe_key_after(msg, exc):
await self._release_inbound_dedupe_key(msg)
await self._send_error(msg, "An internal error occurred. Please try again.")
# -- chat handling -----------------------------------------------------
async def _get_bound_identity_rejection(self, msg: InboundMessage) -> _BoundIdentityRejection | None:
"""Return None when *msg* may proceed; otherwise return rejection routing hints.
The returned object means the message lacks a verified bound identity.
Its fields are intentionally limited to server-side values re-read from
the connection repository, so rejection outbounds never trust a rejected
inbound message's asserted connection metadata.
"""
if not self._require_bound_identity:
return None
# Webhook-authenticated channels (GitHub) opt out via
# ChannelRunPolicy.requires_bound_identity=False. Authenticity is
# enforced at the webhook route by HMAC, and the "sender → DeerFlow
# user" binding is encoded in the agent's config.yaml ownership, not
# in the channel-connections table — there is no per-sender
# /connect handshake to perform.
policy = CHANNEL_RUN_POLICY.get(msg.channel_name)
if policy is not None and not policy.requires_bound_identity:
return None
if _auth_disabled_owner_user_id():
return None
has_connection = bool(msg.connection_id)
has_owner = bool(msg.owner_user_id)
if not (has_connection and has_owner):
return _BoundIdentityRejection()
if self._connection_repo is None:
return _BoundIdentityRejection(message=BOUND_IDENTITY_UNAVAILABLE_MESSAGE)
# The manager is the run-creation security boundary, so it does not
# trust mutable InboundMessage identity fields by themselves. Re-read
# the binding by provider identity before creating DeerFlow threads or
# runs. If the asserted identity does not match, keep only the
# server-side connection fields as outbound routing hints.
connection = await self._connection_repo.find_connection_by_external_identity(
provider=msg.channel_name,
external_account_id=msg.user_id,
workspace_id=msg.workspace_id or None,
)
if connection is None:
return _BoundIdentityRejection()
connection_id = connection.get("id")
owner_user_id = connection.get("owner_user_id")
if connection_id == msg.connection_id and owner_user_id == msg.owner_user_id:
return None
return _BoundIdentityRejection(outbound_connection_id=connection_id, outbound_owner_user_id=owner_user_id)
async def _reject_unbound_channel_message(
self,
msg: InboundMessage,
*,
bound_identity_rejection: _BoundIdentityRejection,
) -> None:
logger.info(
"[Manager] rejecting unbound channel message: channel=%s, chat_id=%s",
msg.channel_name,
msg.chat_id,
)
outbound = OutboundMessage(
channel_name=msg.channel_name,
chat_id=msg.chat_id,
thread_id="",
text=bound_identity_rejection.message,
thread_ts=msg.thread_ts,
connection_id=bound_identity_rejection.outbound_connection_id,
owner_user_id=bound_identity_rejection.outbound_owner_user_id,
metadata=_slim_metadata(msg.metadata),
)
await self.bus.publish_outbound(outbound)
async def _lookup_thread_id(self, msg: InboundMessage) -> str | None:
return await lookup_thread_id(msg, repo=self._connection_repo, store=self.store)
async def _store_thread_id(self, msg: InboundMessage, thread_id: str) -> None:
if msg.connection_id and msg.owner_user_id and self._connection_repo is not None:
await self._connection_repo.set_thread_id(
connection_id=msg.connection_id,
owner_user_id=msg.owner_user_id,
provider=msg.channel_name,
external_conversation_id=msg.chat_id,
external_topic_id=msg.topic_id,
thread_id=thread_id,
)
return
await self.store.set_thread_id(
msg.channel_name,
msg.chat_id,
thread_id,
topic_id=msg.topic_id,
user_id=msg.user_id,
)
async def _claim_thread_id(self, msg: InboundMessage, thread_id: str) -> str:
"""Bind *msg*'s conversation to *thread_id* unless bound already; return the bound thread."""
if msg.connection_id and msg.owner_user_id and self._connection_repo is not None:
return await self._connection_repo.claim_thread_id(
connection_id=msg.connection_id,
owner_user_id=msg.owner_user_id,
provider=msg.channel_name,
external_conversation_id=msg.chat_id,
external_topic_id=msg.topic_id,
thread_id=thread_id,
)
return await self.store.claim_thread_id(msg.channel_name, msg.chat_id, thread_id, topic_id=msg.topic_id, user_id=msg.user_id)
def _remember_thread_agent(self, thread_id: str, agent_name: str | None) -> None:
if len(self._thread_agent_names) > IN_PROCESS_CACHE_MAX_ENTRIES:
self._thread_agent_names.clear()
self._thread_agent_names[thread_id] = agent_name
async def _load_thread_agent(self, client, msg: InboundMessage, thread_id: str) -> str | None:
"""Load an explicit channel agent selection from durable thread metadata."""
if thread_id in self._thread_agent_names:
return self._thread_agent_names[thread_id]
get_kwargs: dict[str, Any] = {}
if owner_headers := _owner_headers(msg):
get_kwargs["headers"] = owner_headers
thread = await client.threads.get(thread_id, **get_kwargs)
metadata = thread.get("metadata") if isinstance(thread, Mapping) else None
raw_agent_name = metadata.get(CHANNEL_AGENT_METADATA_KEY) if isinstance(metadata, Mapping) else None
agent_name: str | None = None
if isinstance(raw_agent_name, str) and raw_agent_name.strip():
if raw_agent_name.strip().lower() == DEFAULT_ASSISTANT_ID:
agent_name = DEFAULT_ASSISTANT_ID
else:
try:
agent_name = _normalize_custom_agent_name(raw_agent_name)
except InvalidChannelSessionConfigError as exc:
raise InvalidChannelSessionConfigError("This conversation has an invalid stored agent selection. Use /agent use <name> to start a valid conversation.") from exc
self._remember_thread_agent(thread_id, agent_name)
return agent_name
def _remember_thread_model(self, thread_id: str, model_name: str | None) -> None:
self._model_cache_generation += 1
if len(self._thread_model_names) > IN_PROCESS_CACHE_MAX_ENTRIES:
self._thread_model_names.clear()
self._thread_model_names[thread_id] = model_name
async def _load_thread_model(self, client, msg: InboundMessage, thread_id: str) -> str | None:
"""Load an explicit /model pin from durable thread metadata.
Reading is deliberately lenient: a stale pin (a model later removed
from the configured allowlist) is left in place and rejected at run
admission, where the run-creation paths route the failure to
``_stale_pin_recovery`` so the user gets an actionable reply instead
of the generic fallback. Anything that is not a non-empty string reads
as "no pin".
The ``threads.get`` deliberately runs *outside* ``_model_pin_lock``:
one slow GET must not serialize every conversation's preload and every
``/model`` write process-wide. Writes are the deliberate exception:
``/model`` writes and stale-pin resets hold the lock across the durable
``threads.update`` itself (see ``_persist_model_pin``), so the durable
write and the cache publication stay atomic against a cold load's
older snapshot — and the post-GET re-check below then sees the fresh
value, so the older snapshot loses.
"""
if thread_id in self._thread_model_names:
return self._thread_model_names[thread_id]
get_kwargs: dict[str, Any] = {}
if owner_headers := _owner_headers(msg):
get_kwargs["headers"] = owner_headers
for attempt in (0, 1):
generation = self._model_cache_generation
thread = await client.threads.get(thread_id, **get_kwargs)
metadata = thread.get("metadata") if isinstance(thread, Mapping) else None
raw_model_name = metadata.get(CHANNEL_MODEL_METADATA_KEY) if isinstance(metadata, Mapping) else None
model_name = raw_model_name.strip() if isinstance(raw_model_name, str) and raw_model_name.strip() else None
async with self._model_pin_lock:
# A /model write that committed while the GET was in flight
# already published under the lock; its value wins over this
# older snapshot.
if thread_id in self._thread_model_names:
return self._thread_model_names[thread_id]
if attempt == 0 and generation != self._model_cache_generation:
# A write or a clear-on-pressure eviction landed while the
# GET was in flight — the cache entry may have been
# evicted after the write, so the in-cache re-check above
# cannot see it. The snapshot may be stale; refetch once
# instead of publishing it over the committed selection.
continue
self._remember_thread_model(thread_id, model_name)
return model_name
async def _persist_model_pin(
self,
client,
msg: InboundMessage,
thread_id: str,
model_name: str | None,
failure_reply: str,
) -> str | None:
"""Persist (or, with ``None``, clear) the ``/model`` pin.
Returns ``failure_reply`` when the thread update fails so callers can
return it verbatim; returns ``None`` on success. Callers hold
``_model_pin_lock`` so the durable write and the cache publication
stay atomic against a cold load's older ``threads.get`` snapshot.
"""
update_kwargs: dict[str, Any] = {"metadata": {CHANNEL_MODEL_METADATA_KEY: model_name}}
if owner_headers := _owner_headers(msg):
update_kwargs["headers"] = owner_headers
try:
await client.threads.update(thread_id, **update_kwargs)
except Exception:
logger.exception(
"[Manager] failed to persist model pin (model_name=%s) for thread_id=%s",
model_name,
thread_id,
)
return failure_reply
return None
async def _resolve_configured_model_name(
self,
client,
msg: InboundMessage,
thread_id: str | None,
) -> tuple[str | None, str | None]:
"""Resolve the model the next run would use without a ``/model`` pin.
Returns ``(model_name, source)`` with source ``"session"`` (a session
config layer carried the model) or ``"agent"`` (a custom agent's
configured ``model`` applies as the runtime fallback). The layer
merge reuses the exact ``_merge_dicts`` shapes of
``_resolve_run_params`` (a later layer replaces a nested carrier
wholesale), and ``_select_carried_model_name`` is the single
definition of which carrier wins — no third replication of the
Gateway precedence. When no layer carries a model, the fallback
mirrors the agent factory's ``_resolve_model_name(requested or
agent_model_name)``: the effective agent is resolved from the same
merged carriers the run would use (explicit message/thread choice,
then carrier ``agent_name``, then session ``assistant_id``), and its
configured ``model`` is the next run's model.
"""
channel_layer, user_layer = self._resolve_session_layer(msg)
merged_config = _merge_dicts(
DEFAULT_RUN_CONFIG,
self._default_session.get("config"),
channel_layer.get("config"),
user_layer.get("config"),
)
merged_context = _merge_dicts(
DEFAULT_RUN_CONTEXT,
self._default_session.get("context"),
channel_layer.get("context"),
user_layer.get("context"),
)
carried = _select_carried_model_name(merged_config, merged_context)
if carried:
return carried, "session"
thread_assistant_id: str | None = None
if thread_id:
try:
thread_assistant_id = await self._load_thread_agent(client, msg, thread_id)
except Exception:
logger.warning(
"[Manager] failed to load thread agent for /model status (thread_id=%s)",
thread_id,
exc_info=True,
)
# Resolve the effective agent the way the run path and the runtime
# would: an explicit per-message or thread-pinned choice overwrites
# every carrier (``_apply_explicit_agent_choice``); otherwise the
# merged carriers' ``agent_name`` wins, because the run path only
# *defaults* the assistant-derived name into the top-level context
# with ``setdefault``; otherwise the session-layer ``assistant_id``
# applies.
agent_name: str | None = None
explicit_assistant = self._resolve_message_assistant_id(msg) or thread_assistant_id
if explicit_assistant is not None:
if explicit_assistant != DEFAULT_ASSISTANT_ID:
agent_name = explicit_assistant
else:
agent_name = _select_carried_agent_name(merged_config, merged_context)
if agent_name is None:
assistant_id = self._resolve_assistant_id(
message_assistant_id=None,
thread_assistant_id=None,
user_layer=user_layer,
channel_layer=channel_layer,
)
if assistant_id != DEFAULT_ASSISTANT_ID:
agent_name = assistant_id
if agent_name:
try:
agent_config = await asyncio.to_thread(
load_agent_config,
_normalize_custom_agent_name(agent_name),
user_id=_channel_storage_user_id(msg),
)
except Exception:
logger.warning(
"[Manager] failed to load agent config %r for /model status",
agent_name,
exc_info=True,
)
else:
agent_model = agent_config.model if agent_config else None
if isinstance(agent_model, str) and agent_model.strip():
return agent_model.strip(), "agent"
return None, None
async def _create_thread(
self,
client,
msg: InboundMessage,
*,
agent_name: str | None = None,
) -> str:
"""Create a new thread through Gateway and rebind the conversation to it (``/new``)."""
thread_id, _ = await self._create_gateway_thread(client, msg, agent_name=agent_name)
await self._store_thread_id(msg, thread_id)
self._remember_thread_agent(thread_id, agent_name)
logger.info("[Manager] new thread created through Gateway: thread_id=%s for chat_id=%s topic_id=%s", thread_id, msg.chat_id, msg.topic_id)
return thread_id
async def _create_first_thread(self, client, msg: InboundMessage) -> tuple[str, bool]:
"""Create the conversation's first thread unless another instance bound one first.
Returns ``(thread_id, created)``. Two instances that both missed the
lookup each create a Gateway thread; the binding is claimed
insert-if-absent, so the loser adopts the winner's thread (and deletes
the one it created, which nothing references) instead of overwriting
the binding and splitting the conversation.
"""
thread_id, minted = await self._create_gateway_thread(client, msg)
bound = await self._claim_thread_id(msg, thread_id)
if bound != thread_id:
logger.info(
"[Manager] chat_id=%s topic_id=%s was bound to thread_id=%s concurrently; adopting it instead of thread_id=%s",
msg.chat_id,
msg.topic_id,
bound,
thread_id,
)
if minted:
await self._delete_unbound_thread(client, msg, thread_id)
return bound, False
self._remember_thread_agent(thread_id, None)
logger.info("[Manager] new thread created through Gateway: thread_id=%s for chat_id=%s topic_id=%s", thread_id, msg.chat_id, msg.topic_id)
return thread_id, True
async def _delete_unbound_thread(self, client, msg: InboundMessage, thread_id: str) -> None:
delete_kwargs: dict[str, Any] = {}
if owner_headers := _owner_headers(msg):
delete_kwargs["headers"] = owner_headers
try:
await client.threads.delete(thread_id, **delete_kwargs)
except Exception:
logger.warning("[Manager] failed to delete thread_id=%s after losing the conversation binding; it stays empty", thread_id, exc_info=True)
async def _create_gateway_thread(
self,
client,
msg: InboundMessage,
*,
agent_name: str | None = None,
) -> tuple[str, bool]:
"""Create a thread through Gateway; ``(thread_id, minted)``.
``minted`` is False for a channel-supplied preferred id: ``POST
/threads`` returns an existing thread for it, which may hold history
and must never be deleted as a lost first-binding race.
"""
metadata = _thread_channel_metadata(msg)
if agent_name is not None:
metadata[CHANNEL_AGENT_METADATA_KEY] = agent_name
# Web thread search returns metadata but no run context. Persist the
# canonical key consumed by ``pathOfThread`` so opening this IM
# conversation in the browser keeps the same custom agent. The lead
# agent deliberately has no canonical key: it uses the ordinary chat
# route rather than a non-existent custom-agent route.
if agent_name != DEFAULT_ASSISTANT_ID:
metadata[THREAD_AGENT_METADATA_KEY] = agent_name
owner_headers = _owner_headers(msg)
# Some channels (notably GitHub) supply a deterministic preferred
# thread id so a (repo, PR/issue number) always lands on the same
# LangGraph thread, even after a store wipe. When absent, Gateway
# mints a random id as before.
meta = msg.metadata if isinstance(msg.metadata, dict) else {}
preferred_thread_id = meta.get("preferred_thread_id")
create_kwargs: dict[str, Any] = {"metadata": metadata}
if isinstance(preferred_thread_id, str) and preferred_thread_id:
create_kwargs["thread_id"] = preferred_thread_id
if owner_headers:
create_kwargs["headers"] = owner_headers
try:
thread = await client.threads.create(**create_kwargs)
except ConflictError as exc:
# True race: two webhook deliveries for the same (repo, number)
# land within ms with the same preferred_thread_id. The Gateway
# ``POST /threads`` route is idempotent on sequential reads (it
# returns the existing record when present), so this branch only
# fires for a real concurrent-create conflict that the underlying
# store surfaced as 409.
#
# Narrow the recovery to ConflictError specifically: any other
# exception (transient DB outage, network error, 5xx) used to
# land here too and silently wrote ``preferred_thread_id`` into
# the store, mapping subsequent webhooks to a thread that was
# never created — every later run would 404 forever with no
# retry path. Those non-conflict failures now propagate so the
# caller fails the delivery cleanly.
if not (isinstance(preferred_thread_id, str) and preferred_thread_id):
# Without a preferred id we cannot deterministically recover.
raise
# Verify the racing-write target actually exists before we
# cache the mapping. If ConflictError fires but threads.get
# also rejects, the store underneath is in an inconsistent
# state and we surface the failure rather than poisoning the
# mapping for every future delivery on this issue/PR.
try:
get_kwargs: dict[str, Any] = {}
if owner_headers:
get_kwargs["headers"] = owner_headers
await client.threads.get(preferred_thread_id, **get_kwargs)
except Exception as verify_exc:
logger.warning(
"[Manager] threads.create raced on preferred_thread_id=%s (%s) but follow-up threads.get failed (%s); not caching the mapping",
preferred_thread_id,
exc.__class__.__name__,
verify_exc.__class__.__name__,
)
raise
logger.info(
"[Manager] threads.create raced on preferred_thread_id=%s (%s); reusing the deterministic id",
preferred_thread_id,
exc.__class__.__name__,
)
return preferred_thread_id, False
return thread["thread_id"], "thread_id" not in create_kwargs
async def _get_or_create_thread(self, client, msg: InboundMessage) -> tuple[str, bool]:
"""Return ``(thread_id, created)``, creating a thread only if needed.
Each inbound message is dispatched on its own task, so two messages that
arrive close together for the same chat would both look up a missing
thread and then both create one — the second store silently overwrites
the first, orphaning a Gateway thread and splitting the conversation.
Serialize the create path per conversation and re-check inside the lock
so only the first message creates a thread and the rest reuse it; across
instances the binding claim in ``_create_first_thread`` converges them.
"""
thread_id = await self._lookup_thread_id(msg)
if thread_id:
return thread_id, False
key = (msg.channel_name, msg.chat_id, msg.topic_id)
async with self._thread_create_locks.hold(key):
# A concurrent message for the same chat may have created the
# thread while we were waiting on the lock.
thread_id = await self._lookup_thread_id(msg)
if thread_id:
return thread_id, False
return await self._create_first_thread(client, msg)
async def _update_thread_channel_metadata(self, client, msg: InboundMessage, thread_id: str) -> None:
"""Best-effort source metadata backfill for existing IM-created threads."""
# The metadata (provider/chat/topic) is constant for a thread, so one
# successful backfill per manager lifetime is enough — skip the
# redundant PATCH on every subsequent inbound message.
if thread_id in self._channel_metadata_synced:
return
update_kwargs: dict[str, Any] = {"metadata": _thread_channel_metadata(msg)}
if owner_headers := _owner_headers(msg):
update_kwargs["headers"] = owner_headers
try:
await client.threads.update(thread_id, **update_kwargs)
except Exception:
logger.debug("[Manager] failed to update channel metadata for thread_id=%s", thread_id, exc_info=True)
return
if len(self._channel_metadata_synced) > IN_PROCESS_CACHE_MAX_ENTRIES:
self._channel_metadata_synced.clear()
self._channel_metadata_synced.add(thread_id)
async def _handle_chat(
self,
msg: InboundMessage,
extra_context: dict[str, Any] | None = None,
*,
bound_identity_checked: bool = False,
) -> None:
# Normal entry paths already run the bound-identity check in
# _handle_message() or _handle_command(). Keep this default False so
# direct callers and future internal paths still fail closed.
bound_identity_rejection = None if bound_identity_checked else await self._get_bound_identity_rejection(msg)
if bound_identity_rejection is not None:
await self._reject_unbound_channel_message(msg, bound_identity_rejection=bound_identity_rejection)
return
client = self._get_client()
storage_user_id = _channel_storage_user_id(msg)
# Look up the existing DeerFlow thread, creating one if this is the
# first message for the chat. topic_id may be None (e.g. Telegram
# private chats) — the store handles this by using the "channel:chat_id"
# key without a topic suffix.
thread_id, created = await self._get_or_create_thread(client, msg)
if not created:
logger.info("[Manager] reusing thread: thread_id=%s for topic_id=%s", thread_id, msg.topic_id)
await self._update_thread_channel_metadata(client, msg, thread_id)
await self._load_thread_agent(client, msg, thread_id)
try:
await self._load_thread_model(client, msg, thread_id)
except Exception:
# A transient threads.get failure must not turn an ordinary
# chat message into the generic error reply: nothing is cached
# on failure, so the pin stays unloaded and is retried next turn.
logger.warning(
"[Manager] failed to preload model pin for thread_id=%s; continuing without it",
thread_id,
exc_info=True,
)
serial_state, queued = self._begin_serialized_thread_run(
channel_name=msg.channel_name,
thread_id=thread_id,
)
serial_lock_acquired = False
try:
if queued:
await self._publish_progress_update(
msg,
thread_id,
"Queued behind another request in this conversation. I’ll start working on this as soon as it finishes.",
)
if serial_state is not None:
await serial_state.lock.acquire()
serial_lock_acquired = True
if queued:
await self._publish_progress_update(msg, thread_id, "thinking...")
await self._handle_chat_on_thread(
client,
msg,
thread_id,
extra_context=extra_context,
storage_user_id=storage_user_id,
)
finally:
self._finish_serialized_thread_run(
channel_name=msg.channel_name,
thread_id=thread_id,
state=serial_state,
lock_acquired=serial_lock_acquired,
)
async def _handle_chat_on_thread(
self,
client,
msg: InboundMessage,
thread_id: str,
*,
extra_context: dict[str, Any] | None = None,
storage_user_id: str | None = None,
) -> None:
if storage_user_id is None:
storage_user_id = _channel_storage_user_id(msg)
# Derived before ``receive_file`` may rewrite the message.
idempotency_key = self._run_idempotency_key(msg)
human_message_id = self._human_message_id(msg)
assistant_id, run_config, run_context = self._resolve_run_params(msg, thread_id)
# The pin this run actually submits — stale-pin recovery compares the
# rejected selection against it so a delayed rejection for an old pin
# can never erase a newer /model choice committed mid-flight.
pinned_model_value = run_context.get("model_name")
expected_pin = pinned_model_value.strip() if isinstance(pinned_model_value, str) and pinned_model_value.strip() else None
# Apply per-channel policy: credentials provider (e.g. GitHub
# installation-token mint) and the non-interactive flag for
# webhook channels. Driven by CHANNEL_RUN_POLICY so each new
# webhook channel is a one-row registration, not a fresh
# if-branch here.
policy = await self._apply_channel_policy(msg, run_context)
# If the inbound message contains file attachments, let the channel
# materialize (download) them and update msg.text to include sandbox file paths.
# This enables downstream models to access user-uploaded files by path.
# Channels that do not support file download will simply return the original message.
if msg.files:
from .service import get_channel_service
service = get_channel_service()
channel = service.get_channel(msg.channel_name) if service else None
logger.info("[Manager] preparing receive file context for %d attachments", len(msg.files))
msg = await channel.receive_file(msg, thread_id, user_id=storage_user_id) if channel else msg
if extra_context:
run_context.update(extra_context)
original_text = msg.text
uploaded = await _ingest_inbound_files(thread_id, msg, user_id=storage_user_id)
human_message = _human_input_message_with_channel_response(msg.text, metadata=msg.metadata, original_content=original_text, files=uploaded or None, message_id=human_message_id)
if self._channel_supports_streaming(msg.channel_name):
await self._handle_streaming_chat(
client,
msg,
thread_id,
assistant_id,
run_config,
run_context,
human_message,
storage_user_id=storage_user_id,
idempotency_key=idempotency_key,
)
return
run_kwargs: dict[str, Any] = {
"input": {"messages": [human_message]},
"config": run_config,
"context": run_context,
"multitask_strategy": "reject",
}
_apply_run_origin(run_kwargs, msg)
if request_headers := _run_request_headers(msg, idempotency_key):
run_kwargs["headers"] = request_headers
if policy is not None and policy.fire_and_forget:
# Fire-and-forget path: the channel does its own outbound
# during the run (GitHub agents post to the issue/PR via the
# ``gh`` CLI from inside the sandbox), so there is nothing
# for the manager to ferry back. Use ``runs.create`` — a
# short POST that returns once the run is ``pending`` — to
# avoid the SDK's 300s ``httpx.ReadTimeout`` on legitimately
# long autonomous runs, and the false "internal error"
# outbound that follows when it fires. ``ConflictError`` is
# still raised synchronously by ``start_run`` if a previous
# run on this thread is still active, so the existing
# busy-thread path is preserved.
logger.info(
"[Manager] invoking runs.create(thread_id=%s, text_len=%d) [fire_and_forget]",
thread_id,
len(msg.text or ""),
)
_note_run_requested()
try:
# Capturing the return value is new (issue #4121 Slice 2):
# it carries ``run_id``, which the follow-up watcher below
# needs to subscribe to this run's StreamBridge stream. When
# ``buffer_followups_on_busy`` is off this is otherwise
# behaviorally identical to the previous bare ``await``.
result = await client.runs.create(thread_id, assistant_id, **run_kwargs)
except Exception as exc:
if _is_admitted_run_conflict(exc):
# An earlier delivery's run; it posts its own outcome.
_note_run_confirmed()
logger.info("[Manager] run already admitted for this message: thread_id=%s (%s)", thread_id, exc)
return
if _is_thread_busy_error(exc):
logger.warning("[Manager] thread busy (concurrent run rejected): thread_id=%s", thread_id)
if policy.buffer_followups_on_busy:
self._buffer_followup(thread_id, msg)
else:
# Swallowed like the generic handler would not be: release the
# key so the provider's redelivery can retry once the thread
# frees, instead of being dropped for the dedupe TTL.
await self._release_inbound_dedupe_key(msg)
await self._send_error(msg, THREAD_BUSY_MESSAGE)
return
if recovery := await self._stale_pin_recovery(msg, exc, expected_pin=expected_pin, thread_id=thread_id):
await self._send_error(msg, recovery.text)
await self._release_inbound_dedupe_key(msg)
return
raise
_note_run_confirmed()
# A redelivery gets an already-finished run back; its stream may be
# gone, so only a live run is worth watching.
status = result.get("status") if isinstance(result, dict) else None
if policy.buffer_followups_on_busy and (status is None or status in _LIVE_RUN_STATUSES):
self._maybe_spawn_followup_watcher(thread_id, result, msg)
return
# Only now: ``runs.create`` above has no ``on_disconnect`` parameter.
run_kwargs["on_disconnect"] = CHANNEL_RUN_ON_DISCONNECT
logger.info("[Manager] invoking runs.wait(thread_id=%s, text_len=%d)", thread_id, len(msg.text or ""))
_note_run_requested()
try:
result = await client.runs.wait(
thread_id,
assistant_id,
**run_kwargs,
)
except Exception as exc:
if _is_admitted_run_conflict(exc):
_note_run_confirmed()
logger.info("[Manager] run already admitted for this message: thread_id=%s (%s)", thread_id, exc)
await self._send_error(msg, ALREADY_ADMITTED_MESSAGE)
return
if _is_thread_busy_error(exc):
logger.warning("[Manager] thread busy (concurrent run rejected): thread_id=%s", thread_id)
# Same reason as the fire-and-forget branch above: this error is
# handled here rather than re-raised, so release explicitly.
await self._release_inbound_dedupe_key(msg)
await self._send_error(msg, THREAD_BUSY_MESSAGE)
return
if recovery := await self._stale_pin_recovery(msg, exc, expected_pin=expected_pin, thread_id=thread_id):
await self._release_inbound_dedupe_key(msg)
await self._send_error(msg, recovery.text)
return
raise
_note_run_confirmed()
if _is_run_status_result(result):
# A redelivery attached to the run admitted for the first delivery.
if result["status"] in _LIVE_RUN_STATUSES:
logger.info("[Manager] run already admitted for this message is still %s: thread_id=%s", result["status"], thread_id)
await self._send_error(msg, ALREADY_ADMITTED_MESSAGE)
return
result = await self._admitted_turn(client, msg, thread_id) or result
response_text = _extract_response_text(result)
pending_clarification = _has_current_turn_clarification(result)
artifacts = _extract_artifacts(result)
logger.info(
"[Manager] agent response received: thread_id=%s, response_len=%d, artifacts=%d",
thread_id,
len(response_text) if response_text else 0,
len(artifacts),
)
# Reuse the storage owner cached at the top of _handle_chat so uploads and
# artifact delivery always resolve to the same bucket, even if a future
# channel.receive_file returns a rewritten InboundMessage.
response_text, attachments = _prepare_artifact_delivery(thread_id, response_text, artifacts, user_id=storage_user_id)
if not response_text:
if attachments:
response_text = _format_artifact_text([a.virtual_path for a in attachments])
else:
response_text = "(No response from agent)"
outbound = OutboundMessage(
channel_name=msg.channel_name,
chat_id=msg.chat_id,
thread_id=thread_id,
text=response_text,
artifacts=artifacts,
attachments=attachments,
thread_ts=msg.thread_ts,
connection_id=msg.connection_id,
owner_user_id=msg.owner_user_id,
metadata=_response_metadata_with_clarification(result, msg.metadata, pending_clarification=pending_clarification),
)
logger.info("[Manager] publishing outbound message to bus: channel=%s, chat_id=%s", msg.channel_name, msg.chat_id)
await self.bus.publish_outbound(outbound)
async def _admitted_turn(self, client, msg: InboundMessage, thread_id: str) -> dict[str, Any] | None:
"""This message's turn from the thread's durable state, for a reused run.
A reused run's ``/wait`` answer carries no state, and its replayed
stream may be gone (``gap``). The keyed human message id locates the
turn the admitted run answered, even when later runs followed it.
"""
message_id = self._human_message_id(msg)
if message_id is None:
return None
get_kwargs: dict[str, Any] = {}
if owner_headers := _owner_headers(msg):
get_kwargs["headers"] = owner_headers
try:
state = await client.threads.get_state(thread_id, **get_kwargs)
except Exception:
logger.warning("[Manager] failed to read thread state for an admitted run: thread_id=%s", thread_id, exc_info=True)
return None
return _turn_of_message(state.get("values") if isinstance(state, dict) else None, message_id)
async def _handle_streaming_chat(
self,
client,
msg: InboundMessage,
thread_id: str,
assistant_id: str,
run_config: dict[str, Any],
run_context: dict[str, Any],
human_message: dict[str, Any],
storage_user_id: str | None = None,
idempotency_key: str | None = None,
) -> None:
logger.info("[Manager] invoking runs.stream(thread_id=%s, text_len=%d)", thread_id, len(msg.text or ""))
last_values: dict[str, Any] | list | None = None
streamed_buffers: dict[str, str] = {}
current_message_id: str | None = None
latest_text = ""
last_published_text = ""
last_published_len = 0
last_publish_at = 0.0
stream_error: BaseException | None = None
replay_gap = False
pin_recovery_text: str | None = None
pinned_model_value = run_context.get("model_name")
expected_pin = pinned_model_value.strip() if isinstance(pinned_model_value, str) and pinned_model_value.strip() else None
stream_kwargs: dict[str, Any] = {
"input": {"messages": [human_message]},
"config": run_config,
"context": run_context,
"stream_mode": list(STREAM_MODES),
"multitask_strategy": "reject",
"on_disconnect": CHANNEL_RUN_ON_DISCONNECT,
}
_apply_run_origin(stream_kwargs, msg)
if request_headers := _run_request_headers(msg, idempotency_key):
stream_kwargs["headers"] = request_headers
_note_run_requested()
try:
async for chunk in client.runs.stream(
thread_id,
assistant_id,
**stream_kwargs,
):
# Any frame proves the Gateway admitted the run.
_note_run_confirmed()
event = getattr(chunk, "event", "")
data = getattr(chunk, "data", None)
if event in MESSAGE_STREAM_EVENTS:
accumulated_text, current_message_id = _accumulate_stream_text(streamed_buffers, current_message_id, data)
if accumulated_text:
latest_text = accumulated_text
elif event == "error":
error_data = data if isinstance(data, dict) else {}
logger.warning(
"[Manager] stream error frame: thread_id=%s, error=%s: %s",
thread_id,
error_data.get("name", "Error"),
error_data.get("message", "unknown"),
)
# The v1 SDK yields backend errors as frames, not exceptions.
# Keep draining the stream and use the existing failure path.
stream_error = _RunFailedFrame("Agent run failed.")
elif event == "gap":
# A reused run whose events are no longer replayable.
replay_gap = True
elif event == "values" and isinstance(data, (dict, list)):
last_values = data
# Clarification text is only in the values snapshot;
# publish it so the user sees the question mid-stream.
if _has_current_turn_clarification(data):
clarification_text = _extract_response_text(data)
if clarification_text and clarification_text != latest_text:
latest_text = clarification_text
if not latest_text or latest_text == last_published_text:
continue
now = time.monotonic()
new_chars = len(latest_text) - last_published_len
# OR logic: flush when interval elapsed OR enough chars accumulated
if last_published_text:
if now - last_publish_at < STREAM_UPDATE_MIN_INTERVAL_SECONDS and new_chars < STREAM_UPDATE_MIN_CHARS:
continue
display_text = latest_text + " ▉"
await self.bus.publish_outbound(
OutboundMessage(
channel_name=msg.channel_name,
chat_id=msg.chat_id,
thread_id=thread_id,
text=display_text,
is_final=False,
thread_ts=msg.thread_ts,
connection_id=msg.connection_id,
owner_user_id=msg.owner_user_id,
metadata=_response_metadata(msg.metadata),
)
)
last_published_text = latest_text
last_published_len = len(latest_text)
last_publish_at = now
except Exception as exc:
stream_error = exc
if _is_admitted_run_conflict(exc):
_note_run_confirmed()
logger.info("[Manager] run already admitted for this message: thread_id=%s (%s)", thread_id, exc)
elif _is_thread_busy_error(exc):
logger.warning("[Manager] thread busy (concurrent run rejected): thread_id=%s", thread_id)
else:
logger.exception("[Manager] streaming error: thread_id=%s", thread_id)
# start_run's 400 surfaces here before any SSE response, so
# stale-pin recovery must run in this path — _handle_message's
# generic handler never sees streaming failures. The text is
# delivered as the single final outbound in the finally block.
pin_recovery = await self._stale_pin_recovery(msg, exc, expected_pin=expected_pin, thread_id=thread_id)
pin_recovery_text = pin_recovery.text if pin_recovery else None
finally:
if replay_gap and last_values is None and not latest_text and stream_error is None:
last_values = await self._admitted_turn(client, msg, thread_id)
result = last_values if last_values is not None else {"messages": [{"type": "ai", "content": latest_text}]}
response_text = _extract_response_text(result)
pending_clarification = _has_current_turn_clarification(result)
artifacts = _extract_artifacts(result)
# Reuse the storage owner resolved by _handle_chat so artifact delivery
# matches the upload bucket and we avoid re-running _safe_user_id_for_run
# (and its possible filesystem touch) on the streaming-error path.
response_text, attachments = _prepare_artifact_delivery(thread_id, response_text, artifacts, user_id=storage_user_id)
if not response_text:
if attachments:
response_text = _format_artifact_text([attachment.virtual_path for attachment in attachments])
elif stream_error:
if _is_admitted_run_conflict(stream_error):
response_text = ALREADY_ADMITTED_MESSAGE
elif _is_thread_busy_error(stream_error):
response_text = THREAD_BUSY_MESSAGE
else:
response_text = pin_recovery_text or "An error occurred while processing your request. Please try again."
else:
response_text = latest_text or "(No response from agent)"
logger.info(
"[Manager] streaming response completed: thread_id=%s, response_len=%d, artifacts=%d, error=%s",
thread_id,
len(response_text),
len(artifacts),
stream_error,
)
await self.bus.publish_outbound(
OutboundMessage(
channel_name=msg.channel_name,
chat_id=msg.chat_id,
thread_id=thread_id,
text=response_text,
artifacts=artifacts,
attachments=attachments,
is_final=True,
thread_ts=msg.thread_ts,
connection_id=msg.connection_id,
owner_user_id=msg.owner_user_id,
metadata=_response_metadata_with_clarification(result, msg.metadata, pending_clarification=pending_clarification),
)
)
if stream_error is not None and not self._keep_dedupe_key_after(msg, stream_error):
# This path swallows its own errors, so _handle_message's generic
# handler never runs and never releases the key. Release only
# after publishing the final outbound so a provider redelivery
# cannot overtake this attempt's terminal reply. A stream that
# broke after the run was admitted keeps the key: the run goes on.
await self._release_inbound_dedupe_key(msg)
# -- command handling --------------------------------------------------
async def _handle_command(self, msg: InboundMessage) -> None:
# Commands are the other run-creation entry point besides chat: /new
# calls _create_thread() directly, and /bootstrap routes into
# _handle_chat(). Apply the same bound-identity admission boundary here
# so unbound platform users cannot create unowned threads/checkpoints or
# query Gateway state via commands. Provider-level binding flows
# (/connect <code>, /start <code>) are consumed by the provider adapter
# before the message reaches the manager, so they are unaffected.
bound_identity_rejection = await self._get_bound_identity_rejection(msg)
if bound_identity_rejection is not None:
await self._reject_unbound_channel_message(msg, bound_identity_rejection=bound_identity_rejection)
return
raw_text = msg.text
text = raw_text.strip()
parts = text.split(maxsplit=1)
reply: str | None = None
if not parts:
command = None
reply = _unknown_command_reply()
else:
command = parts[0].lower().removeprefix("/")
if reply is None and not raw_text.startswith("/"):
reply = _unknown_command_reply(command)
if reply is None and command == "bootstrap":
from dataclasses import replace as _dc_replace
chat_text = parts[1] if len(parts) > 1 else "Initialize workspace"
chat_msg = _dc_replace(msg, text=chat_text, msg_type=InboundMessageType.CHAT)
await self._handle_chat(chat_msg, extra_context={"is_bootstrap": True}, bound_identity_checked=True)
return
if reply is None and command == "new":
# Create a new thread through Gateway
client = self._get_client()
await self._create_thread(client, msg)
reply = "New conversation started."
elif reply is None and command == "status":
thread_id = await self._lookup_thread_id(msg)
reply = f"Active thread: {thread_id}" if thread_id else "No active conversation."
elif reply is None and command == "models":
# The pin footer decorates only a successful listing: appended to
# the failure string it would advertise a pin next to an error
# that lists nothing.
try:
data = await self._fetch_gateway_json("/api/models", msg=msg)
except Exception:
logger.exception("Failed to fetch models from gateway")
reply = "Failed to fetch models information."
else:
names = [m["name"] for m in data.get("models", []) if isinstance(m, Mapping) and isinstance(m.get("name"), str)]
reply = ("Available models:\n" + "\n".join(f"• {n}" for n in names)) if names else "No models configured."
model_thread_id = await self._lookup_thread_id(msg)
if model_thread_id:
try:
pinned_model = await self._load_thread_model(self._get_client(), msg, model_thread_id)
except Exception:
logger.warning("[Manager] failed to load model pin for /models footer (thread_id=%s)", model_thread_id, exc_info=True)
pinned_model = None
if pinned_model:
reply += f"\nCurrent conversation model: {pinned_model} (pinned via /model)"
elif reply is None and command == "model":
reply = await self._handle_model_command(msg, parts[1] if len(parts) > 1 else "")
elif reply is None and command == "memory":
reply = await self._fetch_gateway("/api/memory", "memory", msg=msg)
elif reply is None and command == "agent":
reply = await self._handle_agent_command(msg, parts[1] if len(parts) > 1 else "")
elif reply is None and command == "goal":
reply = await self._handle_goal_command(msg, parts[1] if len(parts) > 1 else "")
if reply is None:
return
elif reply is None and command == "help":
reply = (
"Available commands:\n"
"/bootstrap — Start a bootstrap session (enables agent setup)\n"
"/goal [condition|clear] — Set, show, or clear an active goal\n"
"/new — Start a new conversation\n"
"/status — Show current thread info\n"
"/models — List available models\n"
"/model [name|default] — Show or pin this conversation's model\n"
"/memory — Show memory status\n"
"/agent list — List your Custom Agents\n"
"/agent use <name> — Start a new conversation with an agent\n"
"/<skill-name> <task> — Activate an enabled skill for one turn\n"
"/help — Show this help"
)
elif reply is None:
thread_id = await self._lookup_thread_id(msg)
if thread_id:
await self._load_thread_agent(self._get_client(), msg, thread_id)
slash_resolution = await asyncio.to_thread(
lambda: _resolve_slash_skill_command(
raw_text,
self._resolve_available_skill_names(msg, thread_id or ""),
self._get_skill_storage,
)
)
if slash_resolution and slash_resolution.failure_message:
reply = slash_resolution.failure_message
elif slash_resolution and slash_resolution.route_to_chat:
from dataclasses import replace as _dc_replace
chat_msg = _dc_replace(msg, msg_type=InboundMessageType.CHAT)
await self._handle_chat(chat_msg, bound_identity_checked=True)
return
else:
reply = _unknown_command_reply(command)
outbound = OutboundMessage(
channel_name=msg.channel_name,
chat_id=msg.chat_id,
thread_id=await self._lookup_thread_id(msg) or "",
text=reply,
thread_ts=msg.thread_ts,
connection_id=msg.connection_id,
owner_user_id=msg.owner_user_id,
metadata=_slim_metadata(msg.metadata),
)
await self.bus.publish_outbound(outbound)
async def _handle_agent_command(self, msg: InboundMessage, args: str) -> str:
"""List owner-scoped agents or pin one to a fresh conversation."""
parts = args.split()
if len(parts) == 1 and parts[0].lower() == "list":
user_id = _channel_storage_user_id(msg)
try:
agents = await asyncio.to_thread(list_custom_agents, user_id=user_id)
except Exception:
logger.exception("Failed to list custom agents for channel command")
return "Failed to list agents."
rows = ["• lead_agent — Default agent"]
sorted_agents = sorted(agents, key=lambda agent: agent.name)
for agent in sorted_agents[:MAX_CHANNEL_AGENT_LIST_ITEMS]:
description = " ".join((agent.description or "").split())[:MAX_CHANNEL_AGENT_DESCRIPTION_CHARS]
rows.append(f"• {agent.name} — {description}" if description else f"• {agent.name}")
if len(sorted_agents) > MAX_CHANNEL_AGENT_LIST_ITEMS:
rows.append(f"… and {len(sorted_agents) - MAX_CHANNEL_AGENT_LIST_ITEMS} more")
return "Available agents:\n" + "\n".join(rows)
if len(parts) == 2 and parts[0].lower() == "use":
raw_name = parts[1]
if raw_name.lower() == DEFAULT_ASSISTANT_ID:
agent_name = DEFAULT_ASSISTANT_ID
display_name = DEFAULT_ASSISTANT_ID
else:
try:
agent_name = _normalize_custom_agent_name(raw_name)
except InvalidChannelSessionConfigError:
return "Invalid agent name. Use letters, digits, and hyphens only."
try:
await asyncio.to_thread(
load_agent_config,
agent_name,
user_id=_channel_storage_user_id(msg),
)
except FileNotFoundError:
return f"Agent '{agent_name}' was not found. Use /agent list to see available agents."
except Exception:
logger.exception("Failed to load custom agent for channel command")
return f"Failed to select agent '{agent_name}'."
display_name = agent_name
await self._create_thread(self._get_client(), msg, agent_name=agent_name)
return f"Agent '{display_name}' selected. New conversation started."
return "Usage: /agent list or /agent use <name>"
async def _stale_pin_recovery(
self,
msg: InboundMessage,
exc: BaseException,
*,
expected_pin: str | None,
thread_id: str,
) -> StalePinRecovery | None:
"""Turn a stale ``/model`` pin admission failure into an actionable reply.
Run admission rejects any run whose carried ``model_name`` is not in
the configured allowlist. ``expected_pin`` is the selection the failed
run actually submitted (read from its run context), so recovery only
ever clears *that* selection: a delayed rejection for an old pin must
not erase a newer selection committed while the rejected run was in
flight. ``thread_id`` is the caller-authoritative conversation: every
call site already holds it, so it is never re-derived from *msg* — a
store-mapping rewrite between dispatch and recovery can never make
this clear a different conversation's pin. Returns a
:class:`StalePinRecovery` when *exc* was handled here (the caller owns
outbound delivery); returns None when the error is not the allowlist
rejection or no safe recovery exists.
"""
if expected_pin is None or not _is_model_allowlist_error(exc):
return None
client = self._get_client()
current: str | None = self._thread_model_names.get(thread_id)
if current is None:
try:
current = await self._load_thread_model(client, msg, thread_id)
except Exception:
logger.warning(
"[Manager] failed to load model pin while recovering from an admission error (thread_id=%s)",
thread_id,
exc_info=True,
)
return None
if current == expected_pin:
cleared = False
async with self._model_pin_lock:
# Re-check under the lock: a /model write may have committed
# between the read above and lock acquisition.
if self._thread_model_names.get(thread_id) == expected_pin:
# Any truthy return means the durable write failed.
persist_failed = await self._persist_model_pin(client, msg, thread_id, None, "__persist_failed__")
if not persist_failed:
self._remember_thread_model(thread_id, None)
cleared = True
if cleared:
return StalePinRecovery(
f"The model '{expected_pin}' selected for this conversation is no longer available, "
"so I reset the model selection. This conversation now follows the configured default. "
f"Use /model <name> to pick a different model. {_STALE_PIN_RETRY_NOTE}",
True,
)
current = self._thread_model_names.get(thread_id)
if current == expected_pin:
# State unchanged but the durable write failed.
return StalePinRecovery(
f"The model '{expected_pin}' is no longer available, but I could not reset the selection automatically. Send /model default, then pick another model with /model <name>. {_STALE_PIN_RETRY_NOTE}",
False,
)
# The rejected selection is no longer current (replaced by a newer
# /model, already reset, or never pinned — e.g. the rejection came
# from session configuration): reassure without touching the current
# selection.
if current:
return StalePinRecovery(
f"That message used model '{expected_pin}', which is no longer available. The current selection '{current}' is unchanged. {_STALE_PIN_RETRY_NOTE}",
True,
)
# No pin is held for this thread, so the rejected selection came from
# the session configuration — which still carries it, so the next
# message would fail identically. Name the real source and the fix
# instead of claiming a default takeover that did not happen.
return StalePinRecovery(
f"That message used model '{expected_pin}' from the session configuration, which is no longer available. Update the configured model, or pin a working one for this conversation with /model <name>. {_STALE_PIN_RETRY_NOTE}",
False,
)
async def _handle_model_command(self, msg: InboundMessage, args: str) -> str:
"""Show, pin, or reset the model used by the current conversation.
Unlike ``/agent use`` (which starts a fresh conversation because a
thread's checkpoint lineage belongs to its original agent), a model
pin applies to the *current* conversation from the next message: the
model is a per-run context value resolved anew on every turn.
"""
parts = args.split()
thread_id = await self._lookup_thread_id(msg)
if not parts:
pinned: str | None = None
if thread_id:
try:
pinned = await self._load_thread_model(self._get_client(), msg, thread_id)
except Exception:
logger.warning("[Manager] failed to load model pin for /model status (thread_id=%s)", thread_id, exc_info=True)
return "Current model: unknown. Use /models to list available models."
if pinned:
return f"Current model: {pinned} (pinned for this conversation). Use /model default to reset."
configured, source = await self._resolve_configured_model_name(self._get_client(), msg, thread_id)
if configured:
# "session" covers default, channel, and user layers alike —
# naming one layer would misreport the other two.
origin = "agent configuration" if source == "agent" else "session configuration"
return f"Current model: {configured} (from {origin})."
return "Current model: server default. Use /models to list available models."
if len(parts) > 1:
return "Usage: /model [<name>|default]"
if thread_id is None:
return "No active conversation. Send a message first, then use /model."
client = self._get_client()
name = parts[0]
if name.lower() == "default":
async with self._model_pin_lock:
if failure := await self._persist_model_pin(client, msg, thread_id, None, "Failed to reset the model selection."):
return failure
self._remember_thread_model(thread_id, None)
return "Model selection reset. This conversation now follows the configured default."
try:
data = await self._fetch_gateway_json("/api/models", msg=msg)
except Exception:
logger.exception("[Manager] failed to fetch the model list for /model")
return "Failed to fetch the model list."
# /api/models is role-filtered per caller, so list membership doubles
# as the authorization check for user-owned connections. Channels
# without owner binding (allowed_users mode) fall back to internal
# auth, where /api/models returns the full configured list, so the
# membership check admits any configured model there. That matches
# the pre-existing /models command's exposure, but /model newly turns
# it into model *selection* on such channels.
names = [m["name"] for m in data.get("models", []) if isinstance(m, Mapping) and isinstance(m.get("name"), str)]
if name not in names:
listing = "\n".join(f"• {n}" for n in names) if names else "(none)"
return f"Unknown model '{name}'. Available models:\n{listing}"
async with self._model_pin_lock:
if failure := await self._persist_model_pin(client, msg, thread_id, name, "Failed to save the model selection."):
return failure
self._remember_thread_model(thread_id, name)
return f"Model '{name}' selected. It applies to this conversation from the next message."
async def _goal_request(
self,
method: str,
thread_id: str,
*,
headers: dict[str, str],
json: dict[str, Any] | None = None,
) -> dict[str, Any]:
async with httpx.AsyncClient() as http:
request = getattr(http, method.lower())
kwargs: dict[str, Any] = {"timeout": 10, "headers": {**headers, **self._csrf_headers()}}
if json is not None:
kwargs["json"] = json
response = await request(f"{self._gateway_url}/api/threads/{quote(thread_id, safe='')}/goal", **kwargs)
response.raise_for_status()
return response.json() or {}
async def _handle_goal_command(self, msg: InboundMessage, args: str) -> str | None:
command = parse_goal_command(args)
thread_id = await self._lookup_thread_id(msg)
headers = _owner_headers(msg) or create_internal_auth_headers()
if command.kind == "status":
if not thread_id:
return "No active goal."
try:
goal = (await self._goal_request("get", thread_id, headers=headers)).get("goal")
except Exception:
logger.exception("Failed to fetch goal from gateway")
return "Failed to fetch goal information."
return f"Goal: {goal.get('objective')}" if goal else "No active goal."
if command.kind == "clear":
if not thread_id:
return "Goal cleared."
try:
await self._goal_request("delete", thread_id, headers=headers)
except Exception:
logger.exception("Failed to clear goal through gateway")
return "Failed to clear goal."
return "Goal cleared."
if not thread_id:
thread_id, _ = await self._create_first_thread(self._get_client(), msg)
try:
await self._goal_request("put", thread_id, headers=headers, json={"objective": command.objective})
except Exception:
logger.exception("Failed to set goal through gateway")
return "Failed to set goal."
from dataclasses import replace as _dc_replace
chat_msg = _dc_replace(msg, text=command.objective, msg_type=InboundMessageType.CHAT)
await self._handle_chat(chat_msg, bound_identity_checked=True)
return None
async def _fetch_gateway_json(self, path: str, *, msg: InboundMessage | None = None) -> dict[str, Any]:
"""GET a Gateway API path and return the parsed JSON body.
Raises on transport/HTTP errors; callers rendering user-facing
replies should catch and degrade (see ``_fetch_gateway``).
"""
headers = _owner_headers(msg) if msg is not None else None
async with httpx.AsyncClient() as http:
resp = await http.get(
f"{self._gateway_url}{path}",
timeout=10,
headers=headers or create_internal_auth_headers(),
)
resp.raise_for_status()
return resp.json()
async def _fetch_gateway(self, path: str, kind: str, *, msg: InboundMessage | None = None) -> str:
"""Fetch data from the Gateway API for command responses."""
try:
data = await self._fetch_gateway_json(path, msg=msg)
except Exception:
logger.exception("Failed to fetch %s from gateway", kind)
return f"Failed to fetch {kind} information."
if kind == "memory":
facts = data.get("facts", [])
return f"Memory contains {len(facts)} fact(s)."
return str(data)
# -- error helper ------------------------------------------------------
async def _send_error(self, msg: InboundMessage, error_text: str) -> None:
outbound = OutboundMessage(
channel_name=msg.channel_name,
chat_id=msg.chat_id,
thread_id=await self._lookup_thread_id(msg) or "",
text=error_text,
thread_ts=msg.thread_ts,
connection_id=msg.connection_id,
owner_user_id=msg.owner_user_id,
metadata=_slim_metadata(msg.metadata),
)
await self.bus.publish_outbound(outbound)