From 061a7dfc14560e9ac29459a07d89dc833586e328 Mon Sep 17 00:00:00 2001 From: Storm <39829981+StormTian@users.noreply.github.com> Date: Sat, 10 Oct 2026 09:45:17 +0800 Subject: [PATCH] feat(channels): select the model per conversation via /model (#6534) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(channels): select the model per conversation via /model IM channels could list models with /models but had no way to choose one: the model for a conversation came only from channel/session config or the server default, so a user who needed a stronger or cheaper model for one task had to reconfigure the whole channel or move to the web UI. - /model pins a model to the current conversation, validated against the caller-visible /api/models list (list membership doubles as the authorization check), persisted in thread metadata so it survives a Gateway restart and applies from the next message in place — unlike /agent use, which must start a new conversation. - The pin is written to all three runtime carriers (run context, configurable, context) after the session layers merge, so it wins over channel/session configured models; like any explicitly requested model it also takes precedence over a Custom Agent's configured model. - /model (no args) reports the effective model and its source through one resolver that mirrors the run merge order (default -> channel -> user, context carrier over configurable, both session-config carriers scanned); /model default clears the pin; /models names the pinned model in a footer. - Pin lookups degrade instead of failing messages: the status query reports "Current model: unknown" on a lookup failure (matching the /models footer), and the reuse-thread preload leaves the in-process cache untouched so the next turn retries a transient threads.get. - Reserve "model" in the slash-skill reserved-name contract (backend, contract fixture, frontend), register the Telegram handler, and document the command in /help, README, and README_zh. * fix(channels): fence model pin cache and mirror runtime model resolution Review follow-up for /model: - Serialize cold pin loads with persist+cache publication under a per-manager lock, so a slow threads.get that captured pre-pin metadata can never overwrite a concurrently committed selection (commands are not run-serialized). - _resolve_configured_model_name now replicates the run path's shallow session merges and applies the real runtime precedence — config.context (preserved by build_run_config, unbeatable by the setdefault-filled body context) > top-level context > configurable — instead of scanning per-layer carriers; a later layer replacing a nested carrier wholesale no longer leaks a stale value into the status reply. - /api/models resolves the trusted bound owner for internal channel calls and filters the list with the owner's role, so /model selection gates on the same principal that run admission admits for that account. Tests: pin-write-vs-in-flight-load race, status compared against the real merged runtime config (parametrized across carriers), nested-carrier replacement, and owner-role filtering with the real RBAC provider; each verified red by reverting exactly its guard. * fix(channels): share model resolvers with the run path and harden the pin lifecycle The /model status reply re-implemented the session-layer merge and the Gateway carrier precedence, and it ignored the custom-agent model fallback — on a thread pinned to a custom agent whose config sets ``model:`` it reported "server default" while the run actually used the agent's model. Review also found the pin lock spanned a network call and a stale pin bricked its conversation behind the generic error fallback. - Extract ``_select_carried_model_name`` as the single carrier-precedence rule; the status reply reuses the run path's exact ``_merge_dicts`` layer shapes, so a future layering change has one place to touch. - Extract ``_resolve_message_assistant_id`` / ``_resolve_assistant_id`` and share the assistant precedence chain between the run path and the status reply. - Fall back to the resolved custom agent's configured ``model`` when no session layer carries one, mirroring the agent factory's ``_resolve_model_name(requested or agent_model_name)``; the reply now distinguishes "(from channel configuration)" and "(from agent configuration)". - Run the cold ``threads.get`` in ``_load_thread_model`` outside ``_model_pin_lock``; only cache publication takes the lock, with a post-GET re-check so an in-flight /model write still wins over the older snapshot. - Detect the run-admission allowlist rejection in ``_handle_message`` and recover: clear the stale pin (durable write plus cache under the lock) and reply with the reset and the /model recovery instead of the generic internal-error fallback. - Document /model and the ``channel_model_name`` pin semantics in backend/app/channels/AGENTS.md. * fix(channels): fence stale-pin recovery and cover streaming admission failures Review round 4 found the stale-pin recovery itself unsafe and incomplete: - Recovery cleared whatever pin was current when the delayed admission rejection arrived, so a rejection for an old pin could erase a replacement selection committed mid-flight (and tell the user the new model was unavailable). ``_stale_pin_recovery`` now takes the failed run's own selection (``expected_pin``, read from its run context) and clears only while that selection is still current under the lock; a replaced or already-reset selection gets a non-destructive reply instead. - Feishu/Telegram/DingTalk streaming turns never reached the recovery hook because ``_handle_streaming_chat`` swallows start_run's 400 into its terminal-error reply. The recovery now runs in the streaming terminal-error path too, delivered as the single final outbound with the same dedupe cleanup; blocking, streaming, and fire-and-forget run-creation paths all handle the rejection at the site that knows the submitted selection. * docs(channels): fit the /model entry within the agent-guidance size budget The base file sits one byte under the AGENTS.md hard limit, so documenting /model requires net-zero growth: keep the command-inventory addition and a compact pin-semantics sentence, and losslessly tighten the surrounding prose (the file was already at the limit). * fix(channels): apply the review suggestions on /model replies Three non-blocking findings from the latest review: - The /models pin footer decorated the failure string too, advertising a pin next to an error that lists nothing. The command now fetches the listing through ``_fetch_gateway_json`` and appends the footer only on a successful fetch. - The /model status reply labeled every session-layer model "from channel configuration", misreporting user- and default-layer models. The label is now "session configuration" for every session layer. - Stale-pin recovery replies read as if the turn went through, but the run was rejected before starting and the triggering message never reached the agent. All recovery variants now end with an explicit "Your message was not processed — please send it again." * fix(channels): recover stale model pins on the follow-up drain path The drain is the fourth run-creation path and the only one without stale-pin recovery: a /model pin that left the allowlist while follow-ups were buffered made every drained runs.create fail admission, requeueing the batch into the same 400 on every cycle with no user-facing recovery — contradicting the documented "stale pins self-clear" invariant. - Compute expected_pin from the drain's run context (mirroring _handle_chat_on_thread, initialized None since _resolve_run_params is itself inside the guarded try) and run _stale_pin_recovery on drain failure before requeueing; the pin self-clears and the requeued batch succeeds on the next cycle. - Pass the caller-authoritative thread_id into _stale_pin_recovery: every call site already holds it, so recovery never re-derives it from the message — a store-mapping rewrite between dispatch and recovery can never clear the wrong conversation's pin, and the lookup's failure mode (and its guard) disappears entirely. - Return StalePinRecovery(text, resolved) so requeueing callers can tell a cleared pin from a repeat failure: the drain logs INFO when the next cycle is free of this rejection and WARNING when recovery could not remove it (persist failed, or the selection came from session configuration, which still carries it) — no silent infinite requeue loop. Drain logs carry structured facts only: the reply text ends with a "please resend" note that would double-process buffered messages retried automatically (GitHub, the only buffering channel, sends log-only anyway). - Reword the config-carried recovery reply: with no pin held, the rejected selection came from session configuration — name that source and the fix instead of claiming a default takeover that would not happen. - Correct the _load_thread_model lock docstring (writes deliberately hold _model_pin_lock across threads.update) and read the agent config for /model status off the event loop via asyncio.to_thread, with a Blockbuster-gated blocking_io regression test. - Pin the under-lock re-check in _stale_pin_recovery with a concurrency test (a /model committed during the lock-wait window wins; recovery persists nothing), mutation-verified — the drain is the only caller running on a background task without the serialized-run lock, so it is the path most exposed to that window. * Revert "fix(channels): recover stale model pins on the follow-up drain path" This reverts commit 0b4652c689afb56791e5e670901c97a4bf44a156. The drain recovery work moves to a dedicated PR stacked on this branch, keeping #6534 scoped to the /model feature itself. * fix(channels): recover stale model pins on the follow-up drain path The drain is the fourth run-creation path and the only one without stale-pin recovery: a /model pin that left the allowlist while follow-ups were buffered made every drained runs.create fail admission, requeueing the batch into the same 400 on every cycle with no user-facing recovery — contradicting the documented "stale pins self-clear" invariant. - Compute expected_pin from the drain's run context (mirroring _handle_chat_on_thread, initialized None since _resolve_run_params is itself inside the guarded try) and run _stale_pin_recovery on drain failure before requeueing; the pin self-clears and the requeued batch succeeds on the next cycle. - Pass the caller-authoritative thread_id into _stale_pin_recovery: every call site already holds it, so recovery never re-derives it from the message — a store-mapping rewrite between dispatch and recovery can never clear the wrong conversation's pin, and the lookup's failure mode (and its guard) disappears entirely. - Return StalePinRecovery(text, resolved) so requeueing callers can tell a cleared pin from a repeat failure: the drain logs INFO when the next cycle is free of this rejection and WARNING when recovery could not remove it (persist failed, or the selection came from session configuration, which still carries it) — no silent infinite requeue loop. Drain logs carry structured facts only: the reply text ends with a "please resend" note that would double-process buffered messages retried automatically (GitHub, the only buffering channel, sends log-only anyway). - Reword the config-carried recovery reply: with no pin held, the rejected selection came from session configuration — name that source and the fix instead of claiming a default takeover that would not happen. - Correct the _load_thread_model lock docstring (writes deliberately hold _model_pin_lock across threads.update) and read the agent config for /model status off the event loop via asyncio.to_thread, with a Blockbuster-gated blocking_io regression test. - Pin the under-lock re-check in _stale_pin_recovery with a concurrency test (a /model committed during the lock-wait window wins; recovery persists nothing), mutation-verified — the drain is the only caller running on a background task without the serialized-run lock, so it is the path most exposed to that window. (cherry picked from commit 0b4652c689afb56791e5e670901c97a4bf44a156) * fix(channels): apply the third review round on /model command replies - Delete the dead kind == "models" branch in _fetch_gateway: /models has rendered the listing inline since the pin-footer fix, so the helper's copy was unreachable and had already diverged (unguarded m["name"], no pin footer). The internal-auth-header test now drives the surviving /memory kind. - Guard the inline /models listing comprehension like the _handle_model_command sibling (isinstance(m, Mapping) and isinstance(m.get("name"), str)), so a malformed payload entry can no longer turn the command into the generic internal-error reply; regression: test_models_command_skips_malformed_entries. - Drop the unused user parameter from _resolve_bound_owner_for_model_list: the resolved owner comes solely from the trusted header, and the parameter implied the caller's own principal shaped the filter. * fix(channels): apply the fourth review round on the /model lifecycle - Preserve fresh model pins across cache eviction: a cold threads.get racing a successful /model write could still publish its stale snapshot after clear-on-pressure evicted the fresh entry — the in-cache re-check cannot see an eviction. The model-pin cache now carries a write generation; a load whose generation moved during the GET refetches once instead of publishing over the committed selection. Regression: test_fresh_model_pin_survives_cache_eviction_during_cold_load (blocked GET, write, eviction, then the refetch wins), mutation-verified. - Resolve context-carried agents before reporting their model: /model status now mirrors the run path's effective-agent rule — explicit message/thread choice overwrites every carrier, then the merged carriers' agent_name (new _select_carried_agent_name, sharing the model-name precedence via _select_carried_value), then the session-layer assistant_id that only setdefaults into context. A session with context.agent_name=coder now reports coder-model instead of server default, and assistant_id=writer alongside context.agent_name=coder reports coder-model instead of writer-model. Regressions: context-carried, context-beats-assistant_id, and thread-pin-beats-context status tests. --------- Co-authored-by: Willem Jiang --- README.md | 3 + README_zh.md | 3 + backend/app/channels/AGENTS.md | 6 +- backend/app/channels/commands.py | 1 + backend/app/channels/manager.py | 667 +++++++- backend/app/channels/telegram.py | 1 + backend/app/gateway/routers/models.py | 26 +- .../packages/harness/deerflow/skills/slash.py | 2 +- .../test_model_status_agent_config_offloop.py | 53 + backend/tests/test_channels.py | 1383 ++++++++++++++++- backend/tests/test_models_authorization.py | 80 + contracts/slash_skill_contract.json | 1 + frontend/src/core/skills/slash.ts | 1 + 13 files changed, 2182 insertions(+), 45 deletions(-) create mode 100644 backend/tests/blocking_io/test_model_status_agent_config_offloop.py 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",