diff --git a/README.md b/README.md index ce737d7c56..0c2ccc7cda 100644 --- a/README.md +++ b/README.md @@ -1083,6 +1083,7 @@ Once a channel is connected, you can interact with DeerFlow directly from the ch | `/new` | Start a new conversation | | `/status` | Show current thread info | | `/models` | List available models | +| `/model [name\|default]` | Show or pin the current conversation's model | | `/memory` | View memory | | `/agent list` | List your Custom Agents | | `/agent use ` | Start a new conversation with a Custom Agent | @@ -1093,6 +1094,8 @@ Once a channel is connected, you can interact with DeerFlow directly from the ch Agent selection is conversation-scoped: `/agent use ` starts a fresh conversation and pins that Custom Agent in the thread metadata. Existing conversations never switch agents midway, the selection survives a Gateway restart, and opening the IM-created thread in the Web UI continues through the same Custom Agent. Use `/agent use lead_agent` to return to the default agent in a new conversation. +Model selection is conversation-scoped too: `/model ` pins a model to the *current* conversation — validated against the caller-visible model list, persisted in the thread metadata so it survives a Gateway restart, and applied from the next message without starting a new conversation. `/model` shows the effective model and its source, `/model default` clears the pin, and `/models` reports the pinned model. + #### Request Trace Correlation Every Gateway HTTP response carries an `X-Trace-Id` header. The id is inherited diff --git a/README_zh.md b/README_zh.md index fadd334169..aaa09cbdfc 100644 --- a/README_zh.md +++ b/README_zh.md @@ -573,11 +573,14 @@ DINGTALK_CLIENT_SECRET=your_client_secret | `/new` | 开启新对话 | | `/status` | 查看当前 thread 信息 | | `/models` | 列出可用模型 | +| `/model [name\|default]` | 查看或固定当前会话使用的模型 | | `/memory` | 查看 memory | | `/help` | 查看帮助 | > 没有命令前缀的消息会被当作普通聊天处理。DeerFlow 会自动创建 thread，并以对话方式回复。 +模型选择是会话级的：`/model ` 将模型固定到**当前**会话——会按调用方可见的模型列表校验，选择写入 thread 元数据（Gateway 重启后仍有效），从下一条消息开始生效，无需新开对话。`/model` 查看当前生效模型及其来源，`/model default` 清除固定，`/models` 会报告当前固定的模型。 + #### 请求链路关联 每个 Gateway HTTP 响应都携带 `X-Trace-Id` 响应头。若调用方传入了入站 `X-Trace-Id` 则继承之，否则自动生成，代理或上游服务可以借此跨服务固定同一个 id。该行为无需配置，也无法关闭。 diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index 9f25fe2d5f..933662179d 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -35,11 +35,13 @@ Cursor writes drain before cancellation; token expiry drains cursor/auth clearin - First-use creation is serialized by a waiter-aware keyed lock over `(channel_name, chat_id, topic_id)`. Holders and queued callers share one lock generation through failure and cancellation; idle entries are reclaimed only after the final participant leaves. This prevents a late inbound message from bypassing a queued creator and splitting one conversation across duplicate Gateway threads. 5. Feishu/Telegram chat: `runs.stream()` → accumulate AI text → publish multiple outbound updates (`is_final=False`) → publish final outbound (`is_final=True`) 6. Slack/Discord chat: `runs.wait()` → extract final response → publish outbound -6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound. The agent posts its own reply mid-run via `gh` from the sandbox. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub); when the channel's policy also sets `buffer_followups_on_busy=True` (GitHub's default — see "Follow-up buffering while busy" below), the triggering message is additionally captured into a per-thread buffer instead of only logged, so a concurrent comment is not silently dropped. +6b. GitHub chat (`ChannelRunPolicy.fire_and_forget=True`): `runs.create()` returns once the run is `pending`; the manager does not wait for the final state and does not publish an outbound; the agent posts its own reply via sandbox `gh`. `ConflictError` on a busy thread still trips the standard `THREAD_BUSY_MESSAGE` path (log-only on GitHub); with `buffer_followups_on_busy=True` (GitHub's default) the triggering message is captured into a per-thread buffer so a concurrent comment is not silently dropped (see "Follow-up buffering while busy"). 7. Feishu channel sends one running reply card up front, then patches the same card for each outbound update (card JSON sets `config.update_multi=true` for Feishu's patch API requirement). Messages already sent inside an existing Feishu topic carry a compact source-message preview in that card, and queued same-thread follow-ups patch their own source message's card from queued → running → final without falling back to the generic busy reply. 8. Telegram streaming: the "Working on it..." placeholder message is registered as the stream target; non-final updates `editMessageText` it in place (channel-side throttle: 1s in private chats, 3s in groups due to Telegram's 20 msg/min group cap; 4096-char truncation; rate-limited updates dropped); the final update performs the last edit and splits >4096 texts into follow-up messages 9. DingTalk AI Card mode (when `card_template_id` configured): `runs.stream()` → create card with initial text → stream updates via `PUT /v1.0/card/streaming` → finalize on `is_final=True`. Falls back to `sampleMarkdown` if card creation or streaming fails -10. For commands (`/new`, `/status`, `/models`, `/memory`, `/goal`, `/agent`, `/help`): handle locally or query Gateway API. `/agent list` reads only the effective owner's Custom Agents; `/agent use ` validates in the same owner bucket, creates a new thread, and persists `channel_agent_name` in its metadata. A Custom Agent also writes canonical `metadata.agent_name`, which makes thread-search results route Web continuation through `/workspace/agents//chats/`; `lead_agent` deliberately omits that canonical key and stays on the ordinary chat route. The manager caches `channel_agent_name` for the hot path and reloads it after restart before routing a resumed turn. An explicit selection also normalizes `agent_name` across top-level run context plus the existing RunnableConfig `context` and `configurable` carriers (or clears all three for `lead_agent`) before Gateway's `setdefault` compatibility merge, so stale channel defaults cannot silently win. +10. For commands (`/new`, `/status`, `/models`, `/model`, `/memory`, `/goal`, `/agent`, `/help`): handle locally or query Gateway API. `/agent list` reads the effective owner's Custom Agents; `/agent use ` validates in the same owner bucket, creates a new thread, and persists `channel_agent_name` in its metadata. A Custom Agent writes canonical `metadata.agent_name` so thread-search routes Web continuation to `/workspace/agents//chats/`; `lead_agent` omits it and stays on the ordinary chat route. +`/model ` persists `channel_model_name` in the *current* thread's metadata (in place, no fresh thread); `/model` reports the effective model, `/model default` clears it (stale pins self-clear). +The `channel_agent_name` cache reloads after restart before a resumed turn routes. An explicit selection also normalizes `agent_name` across the top-level run context and both RunnableConfig carriers (`context`, `configurable`) — cleared for `lead_agent` — before Gateway's `setdefault` merge, so stale channel defaults cannot silently win. 11. Outbound → channel callbacks → platform reply - GitHub is the exception: the channel logs the final assistant message and does **not** auto-post it to GitHub. Agents use the sandbox `gh` CLI (`gh issue comment`, `gh pr comment`, `gh pr create`, etc.) for intentional writeback, so silence is cheap when several agents fan out on the same event. diff --git a/backend/app/channels/commands.py b/backend/app/channels/commands.py index 5557b8ada1..6887e37b92 100644 --- a/backend/app/channels/commands.py +++ b/backend/app/channels/commands.py @@ -16,6 +16,7 @@ KNOWN_CHANNEL_COMMANDS: frozenset[str] = frozenset( "/new", "/status", "/models", + "/model", "/memory", "/help", } diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index cbeff25fbe..20b8c2b450 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -15,7 +15,7 @@ from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass from html import escape from pathlib import Path -from typing import Any +from typing import Any, NamedTuple from urllib.parse import quote, urlparse import httpx @@ -70,6 +70,11 @@ 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 @@ -92,6 +97,26 @@ STREAM_UPDATE_MIN_CHARS = 60 # flush immediately when this many chars accumulat 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 @@ -440,6 +465,19 @@ def _is_thread_busy_error(exc: BaseException | None) -> bool: 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. @@ -505,6 +543,54 @@ def _merge_dicts(*layers: Any) -> dict[str, Any]: 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("_", "-") @@ -546,6 +632,31 @@ def _apply_explicit_agent_choice( 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. @@ -1273,6 +1384,27 @@ class ChannelManager: # 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 @@ -1330,6 +1462,38 @@ class ChannelManager: 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, *, @@ -1592,6 +1756,11 @@ class ChannelManager: 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 @@ -1600,6 +1769,12 @@ class ChannelManager: # 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)) @@ -1621,6 +1796,35 @@ class ChannelManager: 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", @@ -1650,22 +1854,23 @@ class ChannelManager: 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 (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 below. - message_assistant_id: str | None = None - 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(): - message_assistant_id = meta_assistant_id - + # 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 = 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 + 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"), @@ -1713,6 +1918,14 @@ class ChannelManager: 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 @@ -2146,7 +2359,8 @@ class ChannelManager: ) # 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. + # 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.") @@ -2247,7 +2461,7 @@ class ChannelManager: ) def _remember_thread_agent(self, thread_id: str, agent_name: str | None) -> None: - if len(self._thread_agent_names) > 4096: + if len(self._thread_agent_names) > IN_PROCESS_CACHE_MAX_ENTRIES: self._thread_agent_names.clear() self._thread_agent_names[thread_id] = agent_name @@ -2274,6 +2488,180 @@ class ChannelManager: 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, @@ -2394,7 +2782,7 @@ class ChannelManager: 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) > 4096: + if len(self._channel_metadata_synced) > IN_PROCESS_CACHE_MAX_ENTRIES: self._channel_metadata_synced.clear() self._channel_metadata_synced.add(thread_id) @@ -2425,6 +2813,17 @@ class ChannelManager: 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, @@ -2471,6 +2870,11 @@ class ChannelManager: 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 @@ -2556,6 +2960,10 @@ class ChannelManager: 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) @@ -2576,8 +2984,11 @@ class ChannelManager: await self._release_inbound_dedupe_key(msg) await self._send_error(msg, THREAD_BUSY_MESSAGE) return - else: - raise + 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) @@ -2637,6 +3048,9 @@ class ChannelManager: 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, @@ -2714,6 +3128,12 @@ class ChannelManager: 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) @@ -2731,7 +3151,7 @@ class ChannelManager: if _is_thread_busy_error(stream_error): response_text = THREAD_BUSY_MESSAGE else: - response_text = "An error occurred while processing your request. Please try again." + 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)" @@ -2809,7 +3229,28 @@ class ChannelManager: 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": - reply = await self._fetch_gateway("/api/models", "models", msg=msg) + # 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": @@ -2826,6 +3267,7 @@ class ChannelManager: "/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 — Start a new conversation with an agent\n" @@ -2914,6 +3356,154 @@ class ChannelManager: return "Usage: /agent list or /agent use " + 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 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 . {_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 . {_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 [|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, @@ -2971,28 +3561,31 @@ class ChannelManager: 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.""" - import httpx - try: - 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() - data = resp.json() + 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 == "models": - names = [m["name"] for m in data.get("models", [])] - return ("Available models:\n" + "\n".join(f"• {n}" for n in names)) if names else "No models configured." - elif kind == "memory": + if kind == "memory": facts = data.get("facts", []) return f"Memory contains {len(facts)} fact(s)." return str(data) diff --git a/backend/app/channels/telegram.py b/backend/app/channels/telegram.py index de458e0c16..8f0c74f710 100644 --- a/backend/app/channels/telegram.py +++ b/backend/app/channels/telegram.py @@ -192,6 +192,7 @@ class TelegramChannel(Channel): app.add_handler(CommandHandler("new", self._cmd_generic)) app.add_handler(CommandHandler("status", self._cmd_generic)) app.add_handler(CommandHandler("models", self._cmd_generic)) + app.add_handler(CommandHandler("model", self._cmd_generic)) app.add_handler(CommandHandler("memory", self._cmd_generic)) app.add_handler(CommandHandler("agent", self._cmd_generic)) app.add_handler(CommandHandler("goal", self._cmd_generic)) diff --git a/backend/app/gateway/routers/models.py b/backend/app/gateway/routers/models.py index cc68cdc4b6..46d24a1ef2 100644 --- a/backend/app/gateway/routers/models.py +++ b/backend/app/gateway/routers/models.py @@ -11,6 +11,7 @@ from app.gateway.authz import ( resolve_model_authorization, ) from app.gateway.deps import get_config, get_optional_user_from_request +from app.gateway.internal_auth import get_trusted_internal_owner_user_id from deerflow.config.app_config import AppConfig from deerflow.config.model_config import ModelConfig from deerflow.models.reasoning import reasoning_capabilities_payload, resolve_reasoning_contract @@ -20,6 +21,23 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/api", tags=["models"]) +async def _resolve_bound_owner_for_model_list(request: Request) -> object | None: + """Resolve the trusted bound owner behind an internal channel call. + + Internal callers (IM channel workers, the scheduler) authenticate as a + synthetic user whose role is popped to ``default_role`` when the + authorization principal is built, so the filtered list reflects the + default role — not the account the conversation actually belongs to. + When the call carries a trusted ``X-DeerFlow-Owner-User-Id`` header, the + channel acts on behalf of that owner (same identity run admission + stamps), and the list must be filtered with the owner's role so commands + like ``/model`` gate on the same principal the run would. + """ + from app.gateway.services import resolve_trusted_internal_owner_for_attribution + + return await resolve_trusted_internal_owner_for_attribution(request, get_trusted_internal_owner_user_id(request)) + + class ReasoningEffortCapabilitiesResponse(BaseModel): """Effort values a model accepts, in display order.""" @@ -130,8 +148,14 @@ async def list_models( user = await get_optional_user_from_request(request) if user is not None: + # Internal channel calls authenticate as a synthetic principal that + # falls under default_role; when they carry a trusted bound owner, + # filter with the owner's role so the list matches what run admission + # would admit for that account. + owner = await _resolve_bound_owner_for_model_list(request) + principal_user = owner if owner is not None else user try: - provider, principal = resolve_model_authorization(user, is_internal=_is_internal_caller(request, user)) + provider, principal = resolve_model_authorization(principal_user, is_internal=_is_internal_caller(request, user)) except _AuthorizationUnavailable as exc: if exc.fail_closed: visible_models = [] diff --git a/backend/packages/harness/deerflow/skills/slash.py b/backend/packages/harness/deerflow/skills/slash.py index 75c3407921..fe7e6dc89c 100644 --- a/backend/packages/harness/deerflow/skills/slash.py +++ b/backend/packages/harness/deerflow/skills/slash.py @@ -14,7 +14,7 @@ from deerflow.skills.types import Skill #: (``tests/test_slash_skill_contract.py`` here, ``slash-contract.test.ts`` on #: the frontend), so a reserved command or grammar change in only one language #: fails CI. -RESERVED_SLASH_SKILL_NAMES = frozenset({"agent", "bootstrap", "context", "goal", "help", "memory", "models", "new", "status"}) +RESERVED_SLASH_SKILL_NAMES = frozenset({"agent", "bootstrap", "context", "goal", "help", "memory", "model", "models", "new", "status"}) _SLASH_SKILL_RE = re.compile(r"^/([a-z0-9]+(?:-[a-z0-9]+)*)(?:\s+|$)") _CONTEXT_COMPACT_ARGUMENT = "compact" diff --git a/backend/tests/blocking_io/test_model_status_agent_config_offloop.py b/backend/tests/blocking_io/test_model_status_agent_config_offloop.py new file mode 100644 index 0000000000..ac76ac98a1 --- /dev/null +++ b/backend/tests/blocking_io/test_model_status_agent_config_offloop.py @@ -0,0 +1,53 @@ +"""Regression anchor: ``/model`` status must not read agent configs on the event loop. + +``_resolve_configured_model_name`` resolves a custom-agent selection into its +configured model, which means reading the owner's agent config from disk. +Every run and channel shares the Gateway event loop, so that read must happen +in a worker thread (``asyncio.to_thread``), mirroring ``_handle_agent_command``. +If the ``to_thread`` wrapper is ever dropped, the strict Blockbuster gate raises +``BlockingError`` at the real filesystem read below. +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +from app.channels import manager as manager_module +from app.channels.manager import ChannelManager +from app.channels.message_bus import InboundMessage, MessageBus +from app.channels.store import ChannelStore + +pytestmark = pytest.mark.asyncio + + +async def test_resolve_configured_model_name_loads_agent_config_off_loop(monkeypatch, tmp_path) -> None: + config_file = tmp_path / "agent.yaml" + await asyncio.to_thread(config_file.write_text, "model: cfg-agent-model\n", encoding="utf-8") + + def spy_load_agent_config(name, *, user_id=None): + # A genuine filesystem read: if ``_resolve_configured_model_name`` + # calls the loader synchronously again, the Blockbuster gate fires here. + text = config_file.read_text(encoding="utf-8") + assert name == "cfg-agent" + return SimpleNamespace(model=text.split(":", 1)[1].strip()) + + # Module-level import (collection time): deriving the import path here + # would itself perform file reads under the Blockbuster gate. + monkeypatch.setattr(manager_module, "load_agent_config", spy_load_agent_config) + + # ChannelStore's constructor mkdirs: setup I/O stays off the loop, same + # as the file writes above — only the manager call is the unit under test. + store = await asyncio.to_thread(ChannelStore, path=tmp_path / "store.json") + manager = ChannelManager(bus=MessageBus(), store=store) + thread_id = "thread-1" + # Seeded cache: the assistant resolution needs no client round-trip. + manager._thread_agent_names[thread_id] = "cfg-agent" + msg = InboundMessage(channel_name="test", chat_id="chat1", user_id="platform-user", text="/model") + + resolved, source = await manager._resolve_configured_model_name(MagicMock(), msg, thread_id) + + assert (resolved, source) == ("cfg-agent-model", "agent") diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 32ccb227f4..84381adb44 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -38,6 +38,16 @@ def test_known_channel_command_detection_only_matches_control_commands(): assert not is_known_channel_command(" /new") +def test_model_command_is_a_known_channel_command(): + from app.channels.commands import KNOWN_CHANNEL_COMMANDS, is_known_channel_command + + assert "/model" in KNOWN_CHANNEL_COMMANDS + assert is_known_channel_command("/model gpt-4") + assert is_known_channel_command("/MODEL") + # Prefix collisions must not classify as the /model command. + assert not is_known_channel_command("/modeling stuff") + + def test_strip_leading_mentions_only_drops_flush_leading_mentions(): from app.channels.commands import is_known_channel_command, strip_leading_mentions @@ -712,6 +722,46 @@ def _make_mock_langgraph_client(thread_id="test-thread-123", run_result=None): return mock_client +def _mock_gateway_models(monkeypatch, model_names): + """Route GET /api/models through a stubbed httpx client returning *model_names*.""" + + class MockResponse: + def raise_for_status(self): + return None + + def json(self): + return {"models": [{"name": name} for name in model_names]} + + class MockAsyncClient: + def __init__(self, *args, **kwargs): + return None + + async def __aenter__(self): + return self + + async def __aexit__(self, exc_type, exc, tb): + return None + + async def get(self, url, **kwargs): + return MockResponse() + + monkeypatch.setattr("app.channels.manager.httpx.AsyncClient", MockAsyncClient) + + +def _make_allowlist_exc(model: str): + """Build the 400 admission error a stale model selection produces.""" + import httpx + from langgraph_sdk.errors import BadRequestError + + request = httpx.Request("POST", "http://gateway/api/runs") + response = httpx.Response(400, request=request) + return BadRequestError( + f"Model {model!r} is not in the configured model allowlist", + response=response, + body={"detail": f"Model {model!r} is not in the configured model allowlist"}, + ) + + async def _make_channel_connection_repo(tmp_path: Path): from deerflow.persistence.channel_connections import ChannelConnectionRepository, ChannelCredentialCipher from deerflow.persistence.engine import get_session_factory, init_engine @@ -929,7 +979,7 @@ class TestChannelManager: return None def json(self): - return {"models": [{"name": "default"}]} + return {"facts": [{"text": "a"}]} class MockAsyncClient: def __init__(self, *args, **kwargs): @@ -953,10 +1003,10 @@ class TestChannelManager: store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") manager = ChannelManager(bus=bus, store=store, gateway_url="http://gateway:8001") - reply = await manager._fetch_gateway("/api/models", "models") + reply = await manager._fetch_gateway("/api/memory", "memory") - assert reply == "Available models:\n• default" - assert calls[0]["url"] == "http://gateway:8001/api/models" + assert reply == "Memory contains 1 fact(s)." + assert calls[0]["url"] == "http://gateway:8001/api/memory" assert calls[0]["timeout"] == 10 assert calls[0]["headers"]["X-DeerFlow-Internal-Token"] @@ -2633,6 +2683,7 @@ class TestChannelManager: manager._get_or_create_thread = AsyncMock(return_value=(thread_id, False)) manager._update_thread_channel_metadata = AsyncMock() manager._load_thread_agent = AsyncMock(return_value=None) + manager._load_thread_model = AsyncMock(return_value=None) manager._publish_progress_update = AsyncMock(side_effect=asyncio.CancelledError()) manager._handle_chat_on_thread = AsyncMock() @@ -3650,6 +3701,1150 @@ class TestChannelManager: _run(go()) + def test_handle_command_model_pin_persists_to_thread_metadata(self, monkeypatch): + from app.channels.manager import ChannelManager + from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME + + _mock_gateway_models(monkeypatch, ["model-a", "model-b"]) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + owner_user_id="deerflow-user-1", + text="/model model-b", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "model-b") + + assert reply == "Model 'model-b' selected. It applies to this conversation from the next message." + update_kwargs = mock_client.threads.update.call_args.kwargs + assert update_kwargs["metadata"]["channel_model_name"] == "model-b" + assert update_kwargs["headers"][INTERNAL_OWNER_USER_ID_HEADER_NAME] == "deerflow-user-1" + _, _, run_context = manager._resolve_run_params(msg, "thread-1") + assert run_context["model_name"] == "model-b" + + _run(go()) + + @pytest.mark.parametrize("config_carrier", ["context", "configurable"]) + def test_model_pin_overrides_every_gateway_config_carrier(self, monkeypatch, config_carrier): + """A /model pin must win after the real Gateway config merge. + + ``merge_run_context_overrides`` copies body.context into both + RunnableConfig containers with ``setdefault``, so a session-configured + ``model_name`` in either container would shadow a pin that only lives + in body.context. + """ + from app.channels.manager import ChannelManager + from app.gateway.services import build_run_config, merge_run_context_overrides + from deerflow.agents.lead_agent.agent import _get_runtime_config + + _mock_gateway_models(monkeypatch, ["model-a", "model-b"]) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager( + bus=MessageBus(), + store=store, + channel_sessions={ + "test": { + "context": {"model_name": "session-model"}, + "config": {config_carrier: {"model_name": "configured-model"}}, + } + }, + ) + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model model-b", + msg_type=InboundMessageType.COMMAND, + ) + + await manager._handle_model_command(msg, "model-b") + assistant_id, run_config, run_context = manager._resolve_run_params(msg, "thread-1") + gateway_config = build_run_config("thread-1", run_config, None, assistant_id=assistant_id) + merge_run_context_overrides(gateway_config, run_context, internal=True) + + assert gateway_config["configurable"]["model_name"] == "model-b" + assert gateway_config["context"]["model_name"] == "model-b" + assert _get_runtime_config(gateway_config)["model_name"] == "model-b" + + _run(go()) + + def test_model_command_rejects_unknown_model(self, monkeypatch): + from app.channels.manager import ChannelManager + + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model nope", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "nope") + + assert reply == "Unknown model 'nope'. Available models:\n• model-a" + mock_client.threads.update.assert_not_called() + _, _, run_context = manager._resolve_run_params(msg, "thread-1") + assert "model_name" not in run_context + + _run(go()) + + def test_model_default_clears_pin(self): + from app.channels.manager import CHANNEL_MODEL_METADATA_KEY, ChannelManager + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + manager._thread_model_names["thread-1"] = "model-b" + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model default", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "default") + + assert reply == "Model selection reset. This conversation now follows the configured default." + update_kwargs = mock_client.threads.update.call_args.kwargs + assert update_kwargs["metadata"][CHANNEL_MODEL_METADATA_KEY] is None + _, _, run_context = manager._resolve_run_params(msg, "thread-1") + assert "model_name" not in run_context + + _run(go()) + + def test_model_command_requires_active_conversation(self): + from app.channels.manager import ChannelManager + + async def go(): + manager = ChannelManager( + bus=MessageBus(), + store=ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json"), + ) + mock_client = _make_mock_langgraph_client() + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model model-a", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "model-a") + + assert reply == "No active conversation. Send a message first, then use /model." + mock_client.threads.update.assert_not_called() + + _run(go()) + + def test_model_pin_restored_from_thread_metadata(self): + """After a Gateway restart the pin comes back from durable thread metadata.""" + from app.channels.manager import ChannelManager + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + mock_client.threads.get = AsyncMock(return_value={"thread_id": "thread-1", "metadata": {"channel_model_name": "model-b"}}) + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="hello", + ) + + pinned = await manager._load_thread_model(mock_client, msg, "thread-1") + + assert pinned == "model-b" + _, _, run_context = manager._resolve_run_params(msg, "thread-1") + assert run_context["model_name"] == "model-b" + # Second read hits the cache and does not re-fetch. + mock_client.threads.get.reset_mock() + assert await manager._load_thread_model(mock_client, msg, "thread-1") == "model-b" + mock_client.threads.get.assert_not_called() + + _run(go()) + + def test_model_status_reports_pin_and_sources(self): + from app.channels.manager import ChannelManager + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager( + bus=MessageBus(), + store=store, + channel_sessions={"test": {"context": {"model_name": "session-model"}}}, + ) + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + mock_client.threads.get = AsyncMock(return_value={"thread_id": "thread-1", "metadata": {"channel_model_name": "model-b"}}) + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: model-b (pinned for this conversation). Use /model default to reset." + + # Cleared pin -> the session-configured model is reported. + manager._remember_thread_model("thread-1", None) + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: session-model (from session configuration)." + + # No thread, no session config -> server default. + manager2 = ChannelManager( + bus=MessageBus(), + store=ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json"), + ) + reply = await manager2._handle_model_command(msg, "") + assert reply == "Current model: server default. Use /models to list available models." + + _run(go()) + + def test_model_status_reads_configurable_carrier(self): + """Session layers can pin model_name under config.configurable too; the + status reply must report it instead of falling back to 'server default'.""" + from app.channels.manager import ChannelManager + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager( + bus=MessageBus(), + store=store, + channel_sessions={"test": {"config": {"configurable": {"model_name": "configured-model"}}}}, + ) + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "") + + assert reply == "Current model: configured-model (from session configuration)." + + _run(go()) + + def test_model_status_mirrors_runtime_precedence(self): + """Layer order (default < channel < user) applies within each carrier, + and the context carrier beats configurable (_get_runtime_config).""" + from app.channels.manager import ChannelManager + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager( + bus=MessageBus(), + store=store, + default_session={ + "context": {"model_name": "default-ctx"}, + "config": {"configurable": {"model_name": "default-cfg"}}, + }, + channel_sessions={"test": {"context": {"model_name": "channel-ctx"}}}, + ) + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + # Within the context carrier, the channel layer beats the default layer. + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: channel-ctx (from session configuration)." + + # A user-layer configurable entry cannot shadow a channel-layer context entry. + manager._channel_sessions["test"].setdefault("users", {}).setdefault("platform-user", {})["config"] = {"configurable": {"model_name": "user-cfg"}} + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: channel-ctx (from session configuration)." + + # A user-layer context entry wins over every configurable entry. + manager._channel_sessions["test"]["users"]["platform-user"]["context"] = {"model_name": "user-ctx"} + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: user-ctx (from session configuration)." + + _run(go()) + + def test_fresh_model_pin_survives_cache_eviction_during_cold_load(self, monkeypatch): + """A cold metadata GET racing a successful /model write plus a + clear-on-pressure eviction must not publish its stale snapshot: the + generation guard refetches after invalidation, so the committed pin + survives in the cache exactly as it does in durable metadata.""" + from app.channels.manager import ChannelManager + + monkeypatch.setattr("app.channels.manager.IN_PROCESS_CACHE_MAX_ENTRIES", 1) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="hello", + ) + + get_started = asyncio.Event() + release_get = asyncio.Event() + responses = [ + {"thread_id": "thread-1", "metadata": {}}, # stale pre-pin snapshot + {"thread_id": "thread-1", "metadata": {"channel_model_name": "new-model"}}, + ] + + async def gated_get(tid, **kwargs): + if len(responses) > 1: + get_started.set() + await release_get.wait() + return responses.pop(0) + + mock_client.threads.get = AsyncMock(side_effect=gated_get) + + load = asyncio.create_task(manager._load_thread_model(mock_client, msg, "thread-1")) + await get_started.wait() + # A successful /model write commits, then clear-on-pressure evicts it. + manager._remember_thread_model("thread-1", "new-model") + manager._remember_thread_model("other-thread", "x") + manager._remember_thread_model("third-thread", "y") # cap=1: len 2 > 1 → clear() + assert "thread-1" not in manager._thread_model_names + release_get.set() + + assert await load == "new-model" + assert manager._thread_model_names["thread-1"] == "new-model" + assert mock_client.threads.get.await_count == 2 + + _run(go()) + + def test_model_status_resolves_context_carried_agent(self, monkeypatch): + """A session-carried agent_name (context carrier) resolves that agent's + configured model: the runtime honors the same carrier, so /model must + match the next run instead of reporting server default.""" + from app.channels.manager import ChannelManager + + def fake_load_agent_config(name, *, user_id=None): + assert name == "coder" + return SimpleNamespace(model="coder-model") + + monkeypatch.setattr("app.channels.manager.load_agent_config", fake_load_agent_config) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager( + bus=MessageBus(), + store=store, + channel_sessions={"test": {"context": {"agent_name": "coder"}}}, + ) + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: coder-model (from agent configuration)." + + _run(go()) + + def test_model_status_context_agent_beats_session_assistant_id(self, monkeypatch): + """assistant_id from a session layer loses to a context-carried + agent_name: the run path only defaults the assistant-derived name into + context (setdefault), so the carrier wins — and the status must too.""" + from app.channels.manager import ChannelManager + + configs = {"coder": "coder-model", "writer": "writer-model"} + + def fake_load_agent_config(name, *, user_id=None): + return SimpleNamespace(model=configs[name]) + + monkeypatch.setattr("app.channels.manager.load_agent_config", fake_load_agent_config) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager( + bus=MessageBus(), + store=store, + channel_sessions={"test": {"assistant_id": "writer", "context": {"agent_name": "coder"}}}, + ) + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: coder-model (from agent configuration)." + + _run(go()) + + def test_model_status_thread_agent_pin_beats_context_agent(self, monkeypatch): + """An explicit thread pin overwrites every carrier + (_apply_explicit_agent_choice), so it beats a context-carried + agent_name in the status reply too.""" + from app.channels.manager import ChannelManager + + configs = {"coder": "coder-model", "writer": "writer-model"} + + def fake_load_agent_config(name, *, user_id=None): + return SimpleNamespace(model=configs[name]) + + monkeypatch.setattr("app.channels.manager.load_agent_config", fake_load_agent_config) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager( + bus=MessageBus(), + store=store, + channel_sessions={"test": {"context": {"agent_name": "coder"}}}, + ) + manager._thread_agent_names["thread-1"] = "writer" + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: writer-model (from agent configuration)." + + _run(go()) + + def test_handle_chat_continues_when_model_pin_preload_fails(self): + """A transient pin-lookup failure on a reused thread must not surface as + the generic error reply; the message is processed without the pin.""" + from app.channels.manager import ChannelManager + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "test-thread-123", topic_id="topic1") + manager = ChannelManager(bus=bus, store=store) + manager._client = _make_mock_langgraph_client() + + async def raising_pin_load(client, msg, thread_id): + raise RuntimeError("transient gateway failure") + + manager._load_thread_model = raising_pin_load + outbound_received = [] + + async def capture_outbound(message): + outbound_received.append(message) + + bus.subscribe_outbound(capture_outbound) + await manager.start() + await bus.publish_inbound( + InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="user1", + text="hi", + topic_id="topic1", + ) + ) + await _wait_for(lambda: len(outbound_received) >= 1) + await manager.stop() + + manager._client.runs.wait.assert_called_once() + assert outbound_received[0].text == "Hello from agent!" + + _run(go()) + + def test_model_pin_write_wins_over_in_flight_cold_load(self, monkeypatch): + """A /model write committed while a cold threads.get is in flight must + survive: the older GET snapshot must never overwrite the fresh cache.""" + from app.channels.manager import ChannelManager + + _mock_gateway_models(monkeypatch, ["model-a", "model-b"]) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + + get_started = asyncio.Event() + release_get = asyncio.Event() + + async def delayed_get(tid, **kwargs): + get_started.set() + await release_get.wait() + # The GET returns metadata captured before the pin was written. + return {"thread_id": tid, "metadata": {}} + + mock_client.threads.get = delayed_get + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model model-b", + msg_type=InboundMessageType.COMMAND, + ) + + load_task = asyncio.create_task(manager._load_thread_model(mock_client, msg, "thread-1")) + await get_started.wait() + pin_task = asyncio.create_task(manager._handle_model_command(msg, "model-b")) + # Without serialization the pin commits now; releasing the GET then + # lets the stale snapshot clobber the cache. + await asyncio.sleep(0.05) + release_get.set() + loaded, pin_reply = await asyncio.gather(load_task, pin_task) + + assert pin_reply == "Model 'model-b' selected. It applies to this conversation from the next message." + assert manager._thread_model_names["thread-1"] == "model-b" + _, _, run_context = manager._resolve_run_params(msg, "thread-1") + assert run_context["model_name"] == "model-b" + + _run(go()) + + @pytest.mark.parametrize( + ("channel_layer", "expected"), + [ + # config.context survives build_run_config and beats every other carrier. + ( + { + "context": {"model_name": "ctx-model"}, + "config": { + "context": {"model_name": "cfg-ctx-model"}, + "configurable": {"model_name": "cfg-conf-model"}, + }, + }, + "cfg-ctx-model", + ), + # No config.context: the top-level context carrier beats configurable. + ( + { + "context": {"model_name": "ctx-model"}, + "config": {"configurable": {"model_name": "cfg-conf-model"}}, + }, + "ctx-model", + ), + # Only configurable: it is what the run uses. + ({"config": {"configurable": {"model_name": "cfg-conf-model"}}}, "cfg-conf-model"), + # Only config.context. + ({"config": {"context": {"model_name": "cfg-ctx-model"}}}, "cfg-ctx-model"), + ], + ) + def test_model_status_matches_merged_runtime_config(self, channel_layer, expected): + """The /model status reply must equal the model in the real merged runtime config.""" + from app.channels.manager import ChannelManager + from app.gateway.services import build_run_config, merge_run_context_overrides + from deerflow.agents.lead_agent.agent import _get_runtime_config + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager( + bus=MessageBus(), + store=store, + channel_sessions={"test": channel_layer}, + ) + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + reported, _ = await manager._resolve_configured_model_name(manager._client, msg, "thread-1") + + # Replay the manager's run params through the real Gateway merge. + _, run_config, run_context = manager._resolve_run_params(msg, "thread-1") + gateway_config = build_run_config("thread-1", run_config, None) + merge_run_context_overrides(gateway_config, run_context, internal=True) + runtime_model = _get_runtime_config(gateway_config).get("model_name") + + assert reported == runtime_model == expected + + _run(go()) + + def test_model_status_mirrors_nested_carrier_replacement(self): + """A later session layer carrying a carrier key replaces it wholesale + (shallow merge); the status must not resurrect the replaced value.""" + from app.channels.manager import ChannelManager + from app.gateway.services import build_run_config, merge_run_context_overrides + from deerflow.agents.lead_agent.agent import _get_runtime_config + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager( + bus=MessageBus(), + store=store, + channel_sessions={ + "test": { + "config": {"configurable": {"model_name": "chan-cfg"}}, + "users": {"platform-user": {"config": {"configurable": {}}}}, + } + }, + ) + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + reported, _ = await manager._resolve_configured_model_name(manager._client, msg, "thread-1") + + _, run_config, run_context = manager._resolve_run_params(msg, "thread-1") + gateway_config = build_run_config("thread-1", run_config, None) + merge_run_context_overrides(gateway_config, run_context, internal=True) + runtime_model = _get_runtime_config(gateway_config).get("model_name") + + assert reported is None and runtime_model is None + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: server default. Use /models to list available models." + + _run(go()) + + def test_model_status_reports_custom_agent_model(self, monkeypatch): + """A conversation pinned to a custom agent whose config sets ``model:`` + runs that model — the agent factory falls back to it when the request + carries none — so the status reply must report it, not 'server default'.""" + from types import SimpleNamespace + + from app.channels import manager as manager_module + from app.channels.manager import ChannelManager + + monkeypatch.setattr( + manager_module, + "load_agent_config", + lambda name, *, user_id=None: SimpleNamespace(model="agent-model"), + ) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store, channel_sessions={"test": {}}) + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + mock_client.threads.get = AsyncMock(return_value={"thread_id": "thread-1", "metadata": {"channel_agent_name": "coder"}}) + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "") + + assert reply == "Current model: agent-model (from agent configuration)." + # Session layer wins over the agent fallback (requested > agent_model_name). + manager._channel_sessions["test"]["context"] = {"model_name": "session-model"} + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: session-model (from session configuration)." + + _run(go()) + + def test_model_status_agent_without_configured_model(self, monkeypatch): + """A custom agent with no ``model:`` (or an unloadable config) leaves the + conversation on the server default; the status must say so, not crash.""" + from types import SimpleNamespace + + from app.channels import manager as manager_module + from app.channels.manager import ChannelManager + + monkeypatch.setattr( + manager_module, + "load_agent_config", + lambda name, *, user_id=None: SimpleNamespace(model=None), + ) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + mock_client.threads.get = AsyncMock(return_value={"thread_id": "thread-1", "metadata": {"channel_agent_name": "coder"}}) + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "") + + assert reply == "Current model: server default. Use /models to list available models." + + # An unloadable agent config (deleted agent) degrades the same way. + def _raise(name, *, user_id=None): + raise FileNotFoundError(name) + + monkeypatch.setattr(manager_module, "load_agent_config", _raise) + reply = await manager._handle_model_command(msg, "") + assert reply == "Current model: server default. Use /models to list available models." + + _run(go()) + + def test_model_pin_write_does_not_wait_for_in_flight_cold_load(self, monkeypatch): + """The cold threads.get runs outside _model_pin_lock: a /model write must + commit while a slow GET is still in flight instead of queueing behind it, + and the finished GET must then lose to the committed write.""" + from app.channels.manager import ChannelManager + + _mock_gateway_models(monkeypatch, ["new-model"]) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + get_started = asyncio.Event() + release_get = asyncio.Event() + + async def slow_get(thread_id, **kwargs): + get_started.set() + await release_get.wait() + return {"thread_id": thread_id, "metadata": {"channel_model_name": "old-model"}} + + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + mock_client.threads.get = AsyncMock(side_effect=slow_get) + mock_client.threads.update = AsyncMock(return_value={"thread_id": "thread-1"}) + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + load_task = asyncio.ensure_future(manager._load_thread_model(mock_client, msg, "thread-1")) + await get_started.wait() + # With the GET outside the pin lock this write completes without + # releasing the blocked GET. + reply = await asyncio.wait_for(manager._handle_model_command(msg, "new-model"), timeout=5) + assert reply == "Model 'new-model' selected. It applies to this conversation from the next message." + assert manager._thread_model_names["thread-1"] == "new-model" + # The stale snapshot arriving after the write must not overwrite it. + release_get.set() + assert await load_task == "new-model" + assert manager._thread_model_names["thread-1"] == "new-model" + + _run(go()) + + def test_models_command_omits_pin_footer_on_fetch_failure(self, monkeypatch): + """The pin footer decorates only a successful listing — a failed fetch + must not advertise a pin next to an error that lists nothing.""" + from app.channels.manager import ChannelManager + + async def _raise(path, *, msg=None): + raise RuntimeError("gateway down") + + monkeypatch.setattr(ChannelManager, "_fetch_gateway_json", _raise) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=bus, store=store, gateway_url="http://gateway:8001") + manager._thread_model_names["thread-1"] = "model-b" + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + outbound_received = [] + bus.subscribe_outbound(outbound_received.append) + + await manager._handle_command( + InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/models", + msg_type=InboundMessageType.COMMAND, + ) + ) + + assert outbound_received[0].text == "Failed to fetch models information." + + _run(go()) + + def test_stale_model_pin_admission_failure_resets_pin(self, monkeypatch): + """A run rejected because the pinned model left the allowlist must not + leave the conversation bricked: the rejected selection is cleared + (durable write plus cache) and the reply names the recovery.""" + from app.channels.manager import ChannelManager + + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + manager._remember_thread_model("thread-1", "gone-model") + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + mock_client.threads.update = AsyncMock(return_value={"thread_id": "thread-1"}) + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="hello", + ) + + recovery = await manager._stale_pin_recovery(msg, _make_allowlist_exc("gone-model"), expected_pin="gone-model", thread_id="thread-1") + + assert recovery is not None and recovery.resolved is True + text = recovery.text + assert "gone-model" in text and "/model" in text + assert "not processed" in text + assert manager._thread_model_names["thread-1"] is None + mock_client.threads.update.assert_awaited_once() + assert mock_client.threads.update.await_args.kwargs["metadata"] == {"channel_model_name": None} + + _run(go()) + + def test_stale_pin_recovery_does_not_erase_replacement_pin(self, monkeypatch): + """A delayed rejection for an old pin must not clear a newer selection + committed while the rejected run was in flight.""" + from app.channels.manager import ChannelManager + + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + # The replacement selection is current; the failed run used gone-model. + manager._remember_thread_model("thread-1", "model-b") + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + mock_client.threads.update = AsyncMock(return_value={"thread_id": "thread-1"}) + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="hello", + ) + + recovery = await manager._stale_pin_recovery(msg, _make_allowlist_exc("gone-model"), expected_pin="gone-model", thread_id="thread-1") + + assert recovery is not None and recovery.resolved is True + text = recovery.text + assert "gone-model" in text and "model-b" in text and "unchanged" in text + assert manager._thread_model_names["thread-1"] == "model-b" + mock_client.threads.update.assert_not_awaited() + + _run(go()) + + def test_stale_pin_recovery_rechecks_pin_under_the_lock(self, monkeypatch): + """The under-lock re-check is the last line of defense: a /model write + that commits while recovery waits on _model_pin_lock must win over the + stale rejection. Without it, recovery persists a clear that silently + erases the user's fresh selection from durable metadata. (The sibling + test covers replacement visible at read time; this one covers the + replacement landing in the lock-wait window — the only window the + re-check exists for, and the one the background drain task is most + exposed to since it holds no serialized-run lock.)""" + from app.channels.manager import ChannelManager + + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + manager._remember_thread_model("thread-1", "gone-model") + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + mock_client.threads.update = AsyncMock(return_value={"thread_id": "thread-1"}) + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="hello", + ) + + # Hold the lock like an in-flight /model write does; recovery + # reads the stale cache value, then parks on the lock. + await manager._model_pin_lock.acquire() + recovery_task = asyncio.create_task(manager._stale_pin_recovery(msg, _make_allowlist_exc("gone-model"), expected_pin="gone-model", thread_id="thread-1")) + # Everything before the lock is synchronous, so one yield parks + # recovery on the acquire; the simulated write then publishes its + # value under the lock exactly like the real /model path. + await asyncio.sleep(0) + manager._remember_thread_model("thread-1", "new-model") + manager._model_pin_lock.release() + + recovery = await recovery_task + + # The re-check saw the newer pin: recovery persisted nothing, and + # the user's fresh selection survives — cache and durable alike. + assert recovery is not None and recovery.resolved is True + mock_client.threads.update.assert_not_awaited() + assert manager._thread_model_names["thread-1"] == "new-model" + assert "new-model" in recovery.text and "unchanged" in recovery.text + + _run(go()) + + def test_stale_pin_recovery_ignores_other_failures(self, monkeypatch): + """Only the allowlist rejection for the run's own selection is handled: + other errors, unknown selections, and config-caused rejections are not.""" + from app.channels.manager import ChannelManager + + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="hello", + ) + + # Unrelated error. + assert await manager._stale_pin_recovery(msg, RuntimeError("boom"), expected_pin="gone-model", thread_id="thread-1") is None + # Allowlist rejection, but the caller cannot name the run's selection. + assert await manager._stale_pin_recovery(msg, _make_allowlist_exc("gone-model"), expected_pin=None, thread_id="thread-1") is None + # Allowlist rejection for a selection this thread does not hold + # (session-config-caused): nothing to clear, and the reply names + # the configuration source instead of claiming a default takeover. + recovery = await manager._stale_pin_recovery(msg, _make_allowlist_exc("cfg-model"), expected_pin="cfg-model", thread_id="thread-1") + assert recovery is not None and recovery.resolved is False + text = recovery.text + assert "cfg-model" in text and "session configuration" in text + assert "configured default" not in text + + _run(go()) + + def test_streaming_chat_recovers_stale_model_pin(self, monkeypatch): + """Streaming channels never reach _handle_message's generic handler: + start_run's 400 surfaces inside _handle_streaming_chat, which must clear + the stale pin and deliver the recovery as its single final outbound.""" + from app.channels.manager import ChannelManager + + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + bus = MessageBus() + outbounds: list[OutboundMessage] = [] + bus.subscribe_outbound(outbounds.append) + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=bus, store=store) + manager._remember_thread_model("thread-1", "gone-model") + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + mock_client.threads.update = AsyncMock(return_value={"thread_id": "thread-1"}) + + def failing_stream(*args, **kwargs): + async def gen(): + raise _make_allowlist_exc("gone-model") + yield # pragma: no cover + + return gen() + + mock_client.runs.stream = failing_stream + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="hello", + ) + + await manager._handle_streaming_chat( + mock_client, + msg, + "thread-1", + "lead_agent", + {"configurable": {"thread_id": "thread-1"}}, + {"model_name": "gone-model", "thread_id": "thread-1"}, + {"role": "user", "content": "hello"}, + ) + + finals = [o for o in outbounds if o.is_final] + assert len(finals) == 1 + assert "gone-model" in finals[0].text and "/model" in finals[0].text + assert "An error occurred while processing your request" not in finals[0].text + assert manager._thread_model_names["thread-1"] is None + mock_client.threads.update.assert_awaited_once() + assert mock_client.threads.update.await_args.kwargs["metadata"] == {"channel_model_name": None} + + _run(go()) + + def test_models_command_marks_current_pin(self, monkeypatch): + from app.channels.manager import ChannelManager + + _mock_gateway_models(monkeypatch, ["model-a", "model-b"]) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=bus, store=store, gateway_url="http://gateway:8001") + manager._thread_model_names["thread-1"] = "model-b" + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + outbound_received = [] + + async def capture_outbound(message): + outbound_received.append(message) + + bus.subscribe_outbound(capture_outbound) + + await manager._handle_command( + InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/models", + msg_type=InboundMessageType.COMMAND, + ) + ) + + assert outbound_received[0].text == ("Available models:\n• model-a\n• model-b\nCurrent conversation model: model-b (pinned via /model)") + + _run(go()) + + def test_models_command_skips_malformed_entries(self, monkeypatch): + """A malformed entry in the Gateway payload must not turn /models into + the generic internal-error reply: the listing renders the valid names, + mirroring the guarded comprehension in _handle_model_command.""" + from app.channels.manager import ChannelManager + + async def _payload(self, path, *, msg=None): + return {"models": [{"name": "model-a"}, {"broken": True}, "garbage", 42]} + + monkeypatch.setattr(ChannelManager, "_fetch_gateway_json", _payload) + + async def go(): + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=bus, store=store, gateway_url="http://gateway:8001") + manager._client = _make_mock_langgraph_client(thread_id="thread-1") + outbound_received = [] + + async def capture_outbound(message): + outbound_received.append(message) + + bus.subscribe_outbound(capture_outbound) + + await manager._handle_command( + InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/models", + msg_type=InboundMessageType.COMMAND, + ) + ) + + assert outbound_received[0].text == "Available models:\n• model-a" + + _run(go()) + + def test_model_command_usage_rejects_multiple_args(self): + """`/model a b` is rejected with usage text before any I/O happens.""" + from app.channels.manager import ChannelManager + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model model-a model-b", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "model-a model-b") + + assert reply == "Usage: /model [|default]" + mock_client.threads.update.assert_not_called() + + _run(go()) + + def test_model_status_degrades_when_pin_lookup_fails(self): + """A Gateway failure while reading the pin must not bubble into the + generic message error handler — the status reply degrades to unknown.""" + from app.channels.manager import ChannelManager + + async def go(): + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("test", "chat1", "thread-1") + manager = ChannelManager(bus=MessageBus(), store=store) + mock_client = _make_mock_langgraph_client(thread_id="thread-1") + mock_client.threads.get = AsyncMock(side_effect=RuntimeError("gateway down")) + manager._client = mock_client + msg = InboundMessage( + channel_name="test", + chat_id="chat1", + user_id="platform-user", + text="/model", + msg_type=InboundMessageType.COMMAND, + ) + + reply = await manager._handle_model_command(msg, "") + + assert reply == "Current model: unknown. Use /models to list available models." + mock_client.threads.update.assert_not_called() + + _run(go()) + def test_each_topic_creates_new_thread(self): """Messages with distinct topic_ids should each create a new DeerFlow thread.""" from app.channels.manager import ChannelManager @@ -4606,6 +5801,39 @@ class TestGithubFireAndForget: _run(go()) +def _stale_pin_drain_setup(thread_id, *, model, pin=None, session_model=None): + """Assemble the shared drain-recovery fixture: a manager with one buffered + follow-up whose ``runs.create`` rejects *model* with an allowlist 400. + + Returns ``(manager, carrier_msg, mock_client, bus)``; each test customizes + the durable-thread mocks it asserts on. + """ + from app.channels.manager import ChannelManager + + bus = MessageBus() + store = ChannelStore(path=Path(tempfile.mkdtemp()) / "store.json") + store.set_thread_id("github", "zhfeng/llm-gateway", thread_id) + manager = ChannelManager(bus=bus, store=store) + if pin is not None: + manager._remember_thread_model(thread_id, pin) + if session_model is not None: + manager._default_session = {"context": {"model_name": session_model}} + + carrier_msg = InboundMessage( + channel_name="github", + chat_id="zhfeng/llm-gateway", + user_id="zhfeng", + owner_user_id="agent-owner-1", + text="queued comment", + ) + manager._buffer_followup(thread_id, carrier_msg) + + mock_client = MagicMock() + mock_client.runs.create = AsyncMock(side_effect=_make_allowlist_exc(model)) + manager._client = mock_client + return manager, carrier_msg, mock_client, bus + + class TestGithubFollowupBuffer: """Tests for issue #4121 Slice 2: buffer-and-drain of concurrent GitHub comments that arrive while a run is already active on the thread. @@ -4949,6 +6177,153 @@ class TestGithubFollowupBuffer: _run(go()) + def test_drain_clears_stale_model_pin_and_requeues(self, monkeypatch, caplog): + """A drained run rejected for a stale /model pin must clear the pin — + the drain is a run-creation path like _handle_chat, so the self-clear + invariant covers it too. The batch is still requeued: with the pin + gone, the next drain cycle succeeds instead of looping on the same + admission failure.""" + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + thread_id = "gh-thread-stale-pin" + manager, carrier_msg, mock_client, bus = _stale_pin_drain_setup(thread_id, model="gone-model", pin="gone-model") + mock_client.threads.update = AsyncMock(return_value={"thread_id": thread_id}) + + outbounds = [] + + async def capture_outbound(message): + outbounds.append(message) + + bus.subscribe_outbound(capture_outbound) + + with caplog.at_level(logging.INFO): + await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg) + + # Pin self-cleared, batch requeued for the next cycle. The recovery + # is logged, NOT sent: the buffered messages are retried + # automatically, so a "please resend" reply would double-process + # them (and the only buffering channel, GitHub, is log-only anyway). + assert manager._thread_model_names[thread_id] is None + assert len(manager._followup_buffers[thread_id]) == 1 + mock_client.threads.update.assert_awaited_once() + assert mock_client.threads.update.await_args.kwargs["metadata"] == {"channel_model_name": None} + assert outbounds == [] + # Anchored to the resolved branch: the WARNING branch prints the + # rejected model too, so the model name alone would not prove + # which path ran. + assert "recovered from stale model" in caplog.text + assert "could not self-recover" not in caplog.text + + _run(go()) + + def test_drain_after_concurrent_model_replacement_persists_nothing(self, monkeypatch, caplog): + """A /model committed while the drained run was in flight makes the + rejected selection stale-but-already-replaced: recovery must NOT clear + anything (the newer choice wins) and the log verb must stay neutral — + "recovered", not "cleared" — because no reset ever happened.""" + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + thread_id = "gh-thread-replaced" + manager, carrier_msg, mock_client, _bus = _stale_pin_drain_setup(thread_id, model="gone-model", pin="gone-model") + mock_client.threads.update = AsyncMock() + + async def reject_after_replacement(*args, **kwargs): + # The user pins a working model while the run rejection is in flight. + manager._remember_thread_model(thread_id, "new-model") + raise _make_allowlist_exc("gone-model") + + mock_client.runs.create = AsyncMock(side_effect=reject_after_replacement) + + with caplog.at_level(logging.INFO): + await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg) + + # The replacement selection is untouched and no durable write ran… + assert manager._thread_model_names[thread_id] == "new-model" + mock_client.threads.update.assert_not_awaited() + # …but the batch is still requeued and the log does not claim a clear. + assert len(manager._followup_buffers[thread_id]) == 1 + assert "recovered from stale model" in caplog.text + assert "cleared" not in caplog.text + assert "could not self-recover" not in caplog.text + + _run(go()) + + def test_drain_recovery_uses_caller_thread_id_despite_lookup_failure(self, monkeypatch): + """Recovery takes the caller's authoritative thread_id and never + re-derives it from the message: even with the store lookup raising, + the stale pin is still cleared for the right conversation. (Before the + caller-thread_id contract, this same hiccup silently meant "no + recovery" — the failure mode no longer exists.)""" + from app.channels.manager import ChannelManager + + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + thread_id = "gh-thread-lookup-boom" + manager, carrier_msg, mock_client, _bus = _stale_pin_drain_setup(thread_id, model="gone-model", pin="gone-model") + mock_client.threads.update = AsyncMock(return_value={"thread_id": thread_id}) + + async def _boom(self, msg): + raise RuntimeError("connection repo down") + + monkeypatch.setattr(ChannelManager, "_lookup_thread_id", _boom) + + await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg) + + # Recovery never consults the store: the pin is cleared for the + # caller's thread, the batch is requeued, the durable reset ran. + assert manager._thread_model_names[thread_id] is None + assert len(manager._followup_buffers[thread_id]) == 1 + mock_client.threads.update.assert_awaited_once() + + _run(go()) + + def test_drain_logs_warning_when_stale_pin_persist_fails(self, monkeypatch, caplog): + """When the durable pin reset fails, recovery returns without clearing, + so the next drain cycle rejects identically: the drain must keep an + operator-visible WARNING instead of a per-cycle INFO "recovered" line.""" + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + thread_id = "gh-thread-persist-fails" + manager, carrier_msg, mock_client, _bus = _stale_pin_drain_setup(thread_id, model="gone-model", pin="gone-model") + mock_client.threads.update = AsyncMock(side_effect=RuntimeError("store down")) + + with caplog.at_level(logging.INFO): + await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg) + + # Nothing cleared: pin and batch both intact, WARNING emitted. + assert manager._thread_model_names[thread_id] == "gone-model" + assert len(manager._followup_buffers[thread_id]) == 1 + assert "could not self-recover from stale model" in caplog.text + + _run(go()) + + def test_drain_logs_warning_when_rejection_comes_from_session_config(self, monkeypatch, caplog): + """When the drained run's stale model came from session configuration + (no /model pin held), recovery has nothing to clear and the next cycle + rejects identically — the drain must surface that as WARNING, not as a + per-cycle INFO "recovered" line, while still requeueing the batch.""" + _mock_gateway_models(monkeypatch, ["model-a"]) + + async def go(): + thread_id = "gh-thread-session-cfg" + manager, carrier_msg, mock_client, _bus = _stale_pin_drain_setup(thread_id, model="cfg-model", session_model="cfg-model") + mock_client.threads.get = AsyncMock(return_value={"thread_id": thread_id, "metadata": {}}) + mock_client.threads.update = AsyncMock() + + with caplog.at_level(logging.INFO): + await manager._drain_followups_for_thread(mock_client, thread_id, carrier_msg) + + # Nothing to clear (no pin): batch requeued, WARNING emitted, no durable write. + assert len(manager._followup_buffers[thread_id]) == 1 + mock_client.threads.update.assert_not_awaited() + assert "could not self-recover from stale model" in caplog.text + + _run(go()) + def test_drain_resolve_run_params_failure_requeues_entries_without_losing_them(self, monkeypatch): """If a step BETWEEN the buffer pop and runs.create raises -- e.g. _resolve_run_params blows up because the target agent config was diff --git a/backend/tests/test_models_authorization.py b/backend/tests/test_models_authorization.py index 377064c5c8..4cb59cc3a9 100644 --- a/backend/tests/test_models_authorization.py +++ b/backend/tests/test_models_authorization.py @@ -230,6 +230,86 @@ def test_list_models_provider_error_fail_closed_vs_open(monkeypatch, fail_closed assert len(response.json()["models"]) == expected_count +def _internal_caller_app(app_config: AppConfig, internal_user) -> FastAPI: + """Models app whose request state mirrors AuthMiddleware's internal path.""" + + app = _make_models_app(app_config) + + @app.middleware("http") + async def _stamp_internal_state(request, call_next): + request.state.user = internal_user + return await call_next(request) + + return app + + +def test_list_models_internal_caller_with_bound_owner_uses_owner_role(monkeypatch): + """Internal channel call with a trusted owner header filters by the owner's role. + + Regression for the /model selection gate: with ``default_role=user`` + restricted to cheap models, a bound admin must still see premium models — + run admission accepts them for that account, so the list (which doubles as + the selection validator) must too. Uses the real RBAC provider so the + owner role flows through the same principal pipeline as production. + """ + provider = RbacAuthorizationProvider( + roles={ + "user": {"models": {"allow": ["cheap"]}}, + "admin": {"models": {"allow": "*"}}, + }, + ) + _enable_authorization(monkeypatch, provider, default_role="user") + + internal_user = _user(id="__internal__", system_role="internal", oauth_provider=None, oauth_id=None) + admin_owner = _user(id="admin-1", system_role="admin", oauth_provider=None, oauth_id=None) + monkeypatch.setattr( + "app.gateway.routers.models.get_optional_user_from_request", + AsyncMock(return_value=internal_user), + ) + monkeypatch.setattr( + "app.gateway.services.get_local_provider", + lambda: SimpleNamespace(get_user=AsyncMock(return_value=admin_owner)), + ) + + app_config = _make_app_config(["cheap", "premium"]) + app = _internal_caller_app(app_config, internal_user) + + with TestClient(app) as client: + response = client.get("/api/models", headers={"X-DeerFlow-Owner-User-Id": "admin-1"}) + + assert response.status_code == 200 + names = [m["name"] for m in response.json()["models"]] + assert names == ["cheap", "premium"] + + +def test_list_models_internal_caller_without_owner_still_uses_default_role(monkeypatch): + """Internal calls without a trusted owner keep the default_role filter. + + Pins the documented unbound-channel posture: no owner header → the + synthetic principal still falls under ``default_role``. + """ + provider = RbacAuthorizationProvider( + roles={"user": {"models": {"allow": ["cheap"]}}}, + ) + _enable_authorization(monkeypatch, provider, default_role="user") + + internal_user = _user(id="__internal__", system_role="internal", oauth_provider=None, oauth_id=None) + monkeypatch.setattr( + "app.gateway.routers.models.get_optional_user_from_request", + AsyncMock(return_value=internal_user), + ) + + app_config = _make_app_config(["cheap", "premium"]) + app = _internal_caller_app(app_config, internal_user) + + with TestClient(app) as client: + response = client.get("/api/models") + + assert response.status_code == 200 + names = [m["name"] for m in response.json()["models"]] + assert names == ["cheap"] + + # ── get_model tests ──────────────────────────────────────────────────── diff --git a/contracts/slash_skill_contract.json b/contracts/slash_skill_contract.json index 21cb4b282c..ca26e7639f 100644 --- a/contracts/slash_skill_contract.json +++ b/contracts/slash_skill_contract.json @@ -8,6 +8,7 @@ "goal", "help", "memory", + "model", "models", "new", "status" diff --git a/frontend/src/core/skills/slash.ts b/frontend/src/core/skills/slash.ts index dbb87dc789..5eb0feb5d4 100644 --- a/frontend/src/core/skills/slash.ts +++ b/frontend/src/core/skills/slash.ts @@ -16,6 +16,7 @@ export const RESERVED_SLASH_SKILL_NAMES = new Set([ "goal", "help", "memory", + "model", "models", "new", "status",