mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 05:06:59 +00:00
feat(channels): select the model per conversation via /model (#6534)
* 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 <name> 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 <willem.jiang@gmail.com>
This commit is contained in:
Storm
GitHub
parent
16877b92fc
commit
061a7dfc14
@ -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 <name>` | 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 <name>` 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 <name>` 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
|
||||
|
||||
@ -573,11 +573,14 @@ DINGTALK_CLIENT_SECRET=your_client_secret
|
||||
| `/new` | 开启新对话 |
|
||||
| `/status` | 查看当前 thread 信息 |
|
||||
| `/models` | 列出可用模型 |
|
||||
| `/model [name\|default]` | 查看或固定当前会话使用的模型 |
|
||||
| `/memory` | 查看 memory |
|
||||
| `/help` | 查看帮助 |
|
||||
|
||||
> 没有命令前缀的消息会被当作普通聊天处理。DeerFlow 会自动创建 thread，并以对话方式回复。
|
||||
|
||||
模型选择是会话级的：`/model <name>` 将模型固定到**当前**会话——会按调用方可见的模型列表校验，选择写入 thread 元数据（Gateway 重启后仍有效），从下一条消息开始生效，无需新开对话。`/model` 查看当前生效模型及其来源，`/model default` 清除固定，`/models` 会报告当前固定的模型。
|
||||
|
||||
#### 请求链路关联
|
||||
|
||||
每个 Gateway HTTP 响应都携带 `X-Trace-Id` 响应头。若调用方传入了入站 `X-Trace-Id` 则继承之，否则自动生成，代理或上游服务可以借此跨服务固定同一个 id。该行为无需配置，也无法关闭。
|
||||
|
||||
@ -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 <name>` 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/<name>/chats/<thread_id>`; `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 <name>` 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/<name>/chats/<thread_id>`; `lead_agent` omits it and stays on the ordinary chat route.
|
||||
`/model <name>` 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.
|
||||
|
||||
|
||||
@ -16,6 +16,7 @@ KNOWN_CHANNEL_COMMANDS: frozenset[str] = frozenset(
|
||||
"/new",
|
||||
"/status",
|
||||
"/models",
|
||||
"/model",
|
||||
"/memory",
|
||||
"/help",
|
||||
}
|
||||
|
||||
@ -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 <name> — Start a new conversation with an agent\n"
|
||||
@ -2914,6 +3356,154 @@ class ChannelManager:
|
||||
|
||||
return "Usage: /agent list or /agent use <name>"
|
||||
|
||||
async def _stale_pin_recovery(
|
||||
self,
|
||||
msg: InboundMessage,
|
||||
exc: BaseException,
|
||||
*,
|
||||
expected_pin: str | None,
|
||||
thread_id: str,
|
||||
) -> StalePinRecovery | None:
|
||||
"""Turn a stale ``/model`` pin admission failure into an actionable reply.
|
||||
|
||||
Run admission rejects any run whose carried ``model_name`` is not in
|
||||
the configured allowlist. ``expected_pin`` is the selection the failed
|
||||
run actually submitted (read from its run context), so recovery only
|
||||
ever clears *that* selection: a delayed rejection for an old pin must
|
||||
not erase a newer selection committed while the rejected run was in
|
||||
flight. ``thread_id`` is the caller-authoritative conversation: every
|
||||
call site already holds it, so it is never re-derived from *msg* — a
|
||||
store-mapping rewrite between dispatch and recovery can never make
|
||||
this clear a different conversation's pin. Returns a
|
||||
:class:`StalePinRecovery` when *exc* was handled here (the caller owns
|
||||
outbound delivery); returns None when the error is not the allowlist
|
||||
rejection or no safe recovery exists.
|
||||
"""
|
||||
if expected_pin is None or not _is_model_allowlist_error(exc):
|
||||
return None
|
||||
client = self._get_client()
|
||||
current: str | None = self._thread_model_names.get(thread_id)
|
||||
if current is None:
|
||||
try:
|
||||
current = await self._load_thread_model(client, msg, thread_id)
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"[Manager] failed to load model pin while recovering from an admission error (thread_id=%s)",
|
||||
thread_id,
|
||||
exc_info=True,
|
||||
)
|
||||
return None
|
||||
if current == expected_pin:
|
||||
cleared = False
|
||||
async with self._model_pin_lock:
|
||||
# Re-check under the lock: a /model write may have committed
|
||||
# between the read above and lock acquisition.
|
||||
if self._thread_model_names.get(thread_id) == expected_pin:
|
||||
# Any truthy return means the durable write failed.
|
||||
persist_failed = await self._persist_model_pin(client, msg, thread_id, None, "__persist_failed__")
|
||||
if not persist_failed:
|
||||
self._remember_thread_model(thread_id, None)
|
||||
cleared = True
|
||||
if cleared:
|
||||
return StalePinRecovery(
|
||||
f"The model '{expected_pin}' selected for this conversation is no longer available, "
|
||||
"so I reset the model selection. This conversation now follows the configured default. "
|
||||
f"Use /model <name> to pick a different model. {_STALE_PIN_RETRY_NOTE}",
|
||||
True,
|
||||
)
|
||||
current = self._thread_model_names.get(thread_id)
|
||||
if current == expected_pin:
|
||||
# State unchanged but the durable write failed.
|
||||
return StalePinRecovery(
|
||||
f"The model '{expected_pin}' is no longer available, but I could not reset the selection automatically. Send /model default, then pick another model with /model <name>. {_STALE_PIN_RETRY_NOTE}",
|
||||
False,
|
||||
)
|
||||
# The rejected selection is no longer current (replaced by a newer
|
||||
# /model, already reset, or never pinned — e.g. the rejection came
|
||||
# from session configuration): reassure without touching the current
|
||||
# selection.
|
||||
if current:
|
||||
return StalePinRecovery(
|
||||
f"That message used model '{expected_pin}', which is no longer available. The current selection '{current}' is unchanged. {_STALE_PIN_RETRY_NOTE}",
|
||||
True,
|
||||
)
|
||||
# No pin is held for this thread, so the rejected selection came from
|
||||
# the session configuration — which still carries it, so the next
|
||||
# message would fail identically. Name the real source and the fix
|
||||
# instead of claiming a default takeover that did not happen.
|
||||
return StalePinRecovery(
|
||||
f"That message used model '{expected_pin}' from the session configuration, which is no longer available. Update the configured model, or pin a working one for this conversation with /model <name>. {_STALE_PIN_RETRY_NOTE}",
|
||||
False,
|
||||
)
|
||||
|
||||
async def _handle_model_command(self, msg: InboundMessage, args: str) -> str:
|
||||
"""Show, pin, or reset the model used by the current conversation.
|
||||
|
||||
Unlike ``/agent use`` (which starts a fresh conversation because a
|
||||
thread's checkpoint lineage belongs to its original agent), a model
|
||||
pin applies to the *current* conversation from the next message: the
|
||||
model is a per-run context value resolved anew on every turn.
|
||||
"""
|
||||
parts = args.split()
|
||||
thread_id = await self._lookup_thread_id(msg)
|
||||
|
||||
if not parts:
|
||||
pinned: str | None = None
|
||||
if thread_id:
|
||||
try:
|
||||
pinned = await self._load_thread_model(self._get_client(), msg, thread_id)
|
||||
except Exception:
|
||||
logger.warning("[Manager] failed to load model pin for /model status (thread_id=%s)", thread_id, exc_info=True)
|
||||
return "Current model: unknown. Use /models to list available models."
|
||||
if pinned:
|
||||
return f"Current model: {pinned} (pinned for this conversation). Use /model default to reset."
|
||||
configured, source = await self._resolve_configured_model_name(self._get_client(), msg, thread_id)
|
||||
if configured:
|
||||
# "session" covers default, channel, and user layers alike —
|
||||
# naming one layer would misreport the other two.
|
||||
origin = "agent configuration" if source == "agent" else "session configuration"
|
||||
return f"Current model: {configured} (from {origin})."
|
||||
return "Current model: server default. Use /models to list available models."
|
||||
|
||||
if len(parts) > 1:
|
||||
return "Usage: /model [<name>|default]"
|
||||
|
||||
if thread_id is None:
|
||||
return "No active conversation. Send a message first, then use /model."
|
||||
|
||||
client = self._get_client()
|
||||
name = parts[0]
|
||||
|
||||
if name.lower() == "default":
|
||||
async with self._model_pin_lock:
|
||||
if failure := await self._persist_model_pin(client, msg, thread_id, None, "Failed to reset the model selection."):
|
||||
return failure
|
||||
self._remember_thread_model(thread_id, None)
|
||||
return "Model selection reset. This conversation now follows the configured default."
|
||||
|
||||
try:
|
||||
data = await self._fetch_gateway_json("/api/models", msg=msg)
|
||||
except Exception:
|
||||
logger.exception("[Manager] failed to fetch the model list for /model")
|
||||
return "Failed to fetch the model list."
|
||||
# /api/models is role-filtered per caller, so list membership doubles
|
||||
# as the authorization check for user-owned connections. Channels
|
||||
# without owner binding (allowed_users mode) fall back to internal
|
||||
# auth, where /api/models returns the full configured list, so the
|
||||
# membership check admits any configured model there. That matches
|
||||
# the pre-existing /models command's exposure, but /model newly turns
|
||||
# it into model *selection* on such channels.
|
||||
names = [m["name"] for m in data.get("models", []) if isinstance(m, Mapping) and isinstance(m.get("name"), str)]
|
||||
if name not in names:
|
||||
listing = "\n".join(f"• {n}" for n in names) if names else "(none)"
|
||||
return f"Unknown model '{name}'. Available models:\n{listing}"
|
||||
|
||||
async with self._model_pin_lock:
|
||||
if failure := await self._persist_model_pin(client, msg, thread_id, name, "Failed to save the model selection."):
|
||||
return failure
|
||||
self._remember_thread_model(thread_id, name)
|
||||
return f"Model '{name}' selected. It applies to this conversation from the next message."
|
||||
|
||||
async def _goal_request(
|
||||
self,
|
||||
method: str,
|
||||
@ -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)
|
||||
|
||||
@ -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))
|
||||
|
||||
@ -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 = []
|
||||
|
||||
@ -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"
|
||||
|
||||
|
||||
@ -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")
|
||||
File diff suppressed because it is too large
Load Diff
@ -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 ────────────────────────────────────────────────────
|
||||
|
||||
|
||||
|
||||
@ -8,6 +8,7 @@
|
||||
"goal",
|
||||
"help",
|
||||
"memory",
|
||||
"model",
|
||||
"models",
|
||||
"new",
|
||||
"status"
|
||||
|
||||
@ -16,6 +16,7 @@ export const RESERVED_SLASH_SKILL_NAMES = new Set([
|
||||
"goal",
|
||||
"help",
|
||||
"memory",
|
||||
"model",
|
||||
"models",
|
||||
"new",
|
||||
"status",
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user