deer-flow/backend/app/channels/manager.py

"""ChannelManager — consumes inbound messages and dispatches them to the DeerFlow agent via Gateway."""

from __future__ import annotations

import asyncio
import logging
import math
import mimetypes
import re
import stat
import time
import uuid
from collections import OrderedDict
from collections.abc import Awaitable, Callable, Mapping
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 (
    INBOUND_FILE_CONTENT_KEY,
    PENDING_CLARIFICATION_METADATA_KEY,
    InboundMessage,
    InboundMessageType,
    MessageBus,
    OutboundMessage,
    ResolvedAttachment,
)
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.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.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."
# 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 — the key has already been
# evicted, or never existed in the new process — so the agent runs again
# and may repeat a real side effect (e.g. a duplicate PR comment on
# GitHub). This is parity with every other IM channel's dedupe (same
# mechanism, same TTL), not a channel-specific gap. True idempotency against
# a late/manual redelivery would require persisting the dedupe key in
# ``ChannelStore`` instead, which is not implemented here.
# 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 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 True
    return "already running a task" in str(exc)


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 _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:
    """Return whether a human message is an internal control message hidden from UI."""
    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


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(content: str, *, original_content: str | None = None, files: list[dict[str, Any]] | 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.
    message: dict[str, Any] = {"id": 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 _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]]:
    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()
        # 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 = self.bus.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 = self.bus.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")

    # -- 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():
                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.
                    if 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:
                    # A cancellation after dedupe admission must make provider
                    # redelivery retryable rather than retaining a TTL-long key for
                    # work that never completed.
                    if dedupe_recorded:
                        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:
                    logger.exception(
                        "[Manager] inbound worker %d failed handling channel=%s chat_id=%s",
                        worker_index,
                        msg.channel_name,
                        msg.chat_id,
                    )
                    if dedupe_recorded:
                        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:
                    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

        # 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)

    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.
        """
        key = self._inbound_dedupe_key(msg)
        if key is not None:
            await self._inbound_dedupe_store.release(key)

    @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:
        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:
            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. Stale /model pin admission failures are handled
            # at the run-creation sites, which know the selection the run used.
            await self._release_inbound_dedupe_key(msg)
            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

        self.store.set_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 store the mapping."""
        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__,
            )
            await self._store_thread_id(msg, preferred_thread_id)
            self._remember_thread_agent(preferred_thread_id, agent_name)
            return preferred_thread_id
        thread_id = thread["thread_id"]
        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 _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.
        """
        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_thread(client, msg), True

    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)

        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(msg.text, original_content=original_text, files=uploaded or None)

        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,
            )
            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 owner_headers := _owner_headers(msg):
            run_kwargs["headers"] = owner_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 ""),
            )
            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_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
            if policy.buffer_followups_on_busy:
                self._maybe_spawn_followup_watcher(thread_id, result, msg)
            return

        logger.info("[Manager] invoking runs.wait(thread_id=%s, text_len=%d)", thread_id, len(msg.text or ""))
        try:
            result = await client.runs.wait(
                thread_id,
                assistant_id,
                **run_kwargs,
            )
        except Exception as exc:
            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

        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(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 _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,
    ) -> 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
        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",
        }
        _apply_run_origin(stream_kwargs, msg)
        if owner_headers := _owner_headers(msg):
            stream_kwargs["headers"] = owner_headers

        try:
            async for chunk in client.runs.stream(
                thread_id,
                assistant_id,
                **stream_kwargs,
            ):
                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 = RuntimeError("Agent run failed.")
                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_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:
            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_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(msg.metadata, pending_clarification=pending_clarification),
                )
            )
            if stream_error is not None:
                # 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.
                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_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)