feat: add fixed questions and improve authentication

offline-backend-20260512/backend/CLAUDE.md

@@ -346,6 +346,7 @@ FastAPI application on port 8001 with health checks at `GET|HEAD /health`, `/api
 | **Dashboard Sessions** (`/api/dashboard-sessions`) | 「大屏绘制智能体」独立页面会话，**按外部 `task_id` 存取，不按 user 分权**（open read + shared writes；`user_id` audit-only）。一个 task 一行（upsert）：`transcript`（聊天记录 JSON）+ `report_json`（最新结构化数据，可解析 = 「有效结果」）。`GET /by-task/{task_id}` - 取该 task 会话（无则 404）；`PUT /by-task/{task_id}` - upsert（`{transcript?, report_json?}`，只覆盖传入项）。Backed by `deerflow.persistence.dashboard_sessions`（table `dashboard_sessions`, auto-created by `create_all`，无迁移）；store wired in `deps.py` as `app.state.dashboard_session_store`。前端独立页 `roundtable-planning/pages/DashboardAgentPage.tsx`（路由 `/page/strategy/dashboard-agent?taskId=`，侧边栏「问答管理 → 大屏绘制」）让智能体分析任务报告 JSON → report-json → 渲染 `DashboardApp`。Tests: `tests/test_dashboard_sessions.py`。 |
 | **Task Reports** (`/api/task-reports`) | 任务级研判情报报告详情，**按 `task_id` 共享、不按 user 分权**（写入者只留 `updated_by` 审计）。`GET /by-task/{task_id}` 始终返回原 Consumer 兼容外壳 `{state:"200", msg:"操作成功！", data:[...]}`（未保存为 `data: []`）；`POST /by-task/{task_id}` 首次保存，已有报告返回 409；`PUT /by-task/{task_id}` 原子替换完整 `data` 数组。`POST /import/by-task/{task_id}` 面向 TaskCOP 导入页，接收 `sq-report-mock.json` 兼容的 `data`/`records` 数组并完整覆盖目标任务；路径中的选定任务 id 优先于文件内的 `taskId`，避免误写。每一项严格是 `{id, taskId, createTime, updateTime, sessionId, contentJson, categoryType}`；`contentJson` 以 JSON **字符串**原样落在 `records_json`，避免破坏大屏按类目解析的输入格式。保存或导入非空 `data` 后，router 会通过 `app.state.taskcop_task_store.mark_analysis_completed` 将同 id 的 TaskCOP 任务升级至状态 `25`，使旧任务列表显示并筛选为“已完成”；普通保存时不存在匹配任务的报告仍可独立保存，导入则要求目标任务存在。Backed by `deerflow.persistence.task_reports`（table `task_reports`, auto-created by `create_all`）；store wired in `deps.py` as `app.state.task_report_store`。前端客户端 `strategy-components/api/task-reports.ts`，`core/auth/dashboard-input.ts` 与 rwfx 3q 完成检查均直接读此接口，不再依赖 Consumer 或本地 sq-report mock。Tests: `tests/test_task_reports.py`。 |
 | **Task Buttons** (`/api/task-buttons`) | Task-workspace configurable jump buttons (按钮管理), **global/shared** (no user scoping; `updated_by` audit-only). Each row = one jump button in a task deep-link workspace's bottom 快捷跳转 bar: `{id, business (rwfx/3qfx/xdfx), label, link_type (url/business), target, append_task_id, id_param (editable query-param name, e.g. id/task_id), id_kind (task/action — only xdfx differs, its deep-link id is an 行动id), open_mode (blank/self), login_params, append_auth, auth_token_param, auth_name_param, enabled, sort_order}`. **Two independent, coexisting login-append mechanisms for `url` buttons** — both resolved frontend-side: (1) `login_params` (JSON column `login_params_json`, `PortableJSON`, migration `20260625_01`) = a list of `{key, source, custom_value}` resolved at click time (`source` = a `userInfo` field or `other`→literal `custom_value`) and appended to the target URL — mirrors the light-app login params; empty = nothing appended. (2) `append_auth` + `auth_token_param` + `auth_name_param` (columns, migration `20260625_02`, default off) — appends `?{auth_token_param}={token1}&{auth_name_param}={username}` so the target system can identify the current user; gated on **token login** — the frontend only stashes `token1` (the `?authToken=`) + the raw `getTokenInfo` username (`login.token1`/`login.username`) when login went through `/login/token`, so a non-token-login session appends neither. The **frontend** holds code-seed defaults (rwfx's 4 historical buttons built from the live `task_deeplink` config); only when this table is empty do the workspace + admin page fall back to them, else this table is the sole source of truth. `GET /` - list all (open to any authenticated user — the workspace renders them); `PUT /` - **admin only** (`_require_admin`), atomically **replaces the whole set** (`{buttons: [...]}`, de-duped by `id`, last wins). Backed by `deerflow.persistence.task_buttons` (table `task_buttons`, auto-created by `create_all`); store wired in `deps.py` as `app.state.task_button_store`. Frontend: helpers `core/auth/task-buttons.ts`, client `strategy-components/api/task-buttons.ts`, admin page `pages/TaskButtonsPage.tsx` (route `/page/strategy/admin/task-buttons`, entry「按钮管理」in the bottom-left settings dropdown, admin-only, both sidebar modes; per-business tabs + 「导入默认」). Tests: `tests/test_task_buttons.py`. |
+| **Fixed Questions** (`/api/fixed-questions`) | Administrator-managed exact-match fixed Q&A (问题配置), **global/shared**. Each row is `{id, question, answer, enabled, tokens_per_second, sort_order}`; stream speed is configurable per row from `1–1000 token/s` and defaults to `100`. `GET /` and whole-set atomic `PUT /` are both admin-only. At the start of an external chat run, `services._find_fixed_answer` checks the latest eligible plain-text human message with exact, case/space/punctuation-sensitive equality. Enabled matches switch the run to `make_fixed_answer_agent`: a local `BaseChatModel` tokenizes with `cl100k_base`, replays the answer in rate-limited token chunks, and keeps the existing LangGraph SSE `messages`/`values` protocol, run journal, checkpoint history, deterministic local title, and refresh behavior intact while no provider/model request or model-concurrency slot is used. File/multimodal/context-prefixed turns, `canvas_agent`, `ai_writing`, writing mode, resume commands, and internal orchestration/background child runs bypass matching. Storage failures fail open to the normal model path. Backed by `deerflow.persistence.fixed_questions` (table `fixed_questions`, SHA-256 indexed lookup plus exact original-string collision check, migrations `20260821_01` + `20260821_02`), wired as `app.state.fixed_question_store`. Frontend: `strategy-components/api/fixed-questions.ts`, admin page `pages/FixedQuestionsPage.tsx`, route `/page/strategy/admin/fixed-questions`, entries in both full-layout and compact-workspace settings menus. Tests: `tests/test_fixed_questions.py`. |
 | **Knowledge** (`/api/knowledge`) | Product knowledge base, **global/shared** (no user isolation; writes record `created_by`/`updated_by` for audit). `POST /ingest/thread/{thread_id}` - sediment one thread into a note (reads messages from the checkpointer, `threads:read` owner-checked); `POST /ingest/threads/batch` - batch-sediment the caller's recent threads (`skip_existing` via `.manifest.json` dedup). **Defaults to `background=true`**: the slow per-thread extraction loop is deferred to a FastAPI background task and the endpoint returns `{queued:true, processed:N}` immediately (each thread's transcript is truncated to `max_input_chars` before the LLM call). Running the loop inline for a large batch outlives the nginx gateway timeout (504), which the frontend's `parseResponse` surfaces as a generic "Request failed"; `background=false` keeps the blocking path (full per-thread `created/skipped/failed` tally) for small batches/scripts. Shared loop: `_run_batch_ingest`. **Optional model selection**: both ingest endpoints accept `model_name` (overrides `knowledge.extract_model_name` for that capture; threaded through `KnowledgeService.capture_thread` ? `extract_thread_llm_multi`). **Model-error surfacing**: when the chosen model produces nothing usable (timeout / API error / non-JSON) capture still **falls back to rule-based extraction and lands a draft** (`llm_fallback:true`); the single-thread (non-background) response and the batch response (`llm_fallback` count) carry this so the UI tells the user "???????". The "??????" chat button stays background/fire-and-forget (passes `model_name`, no per-save error toast); the batch-import dialog is synchronous and shows the fallback count. `POST /ingest/file` - **import an uploaded file** (multipart: `file` + optional `title`/`tags`/`status`/`model_name`) into wiki note(s): the file is converted to Markdown (md/txt read directly, pdf/word/ppt/excel via `convert_file_to_markdown`), distilled by the same wiki-capture LLM pipeline as thread capture (`source_type="file"`), and falls back to a verbatim **draft** note when the model returns nothing usable (`llm_fallback:true`); `POST /ingest/search` - sediment search/tool results as a reference note; `POST /notes` - manual note; `GET /notes` (keyword/tag/source_type/status + pagination); `GET/PUT/DELETE /notes/{id}` (DELETE is soft = archive, `?hard=true` removes the vault file); `POST /search` - `mode=keyword\|vector\|hybrid` (vector?keyword fallback when no embedding endpoint); `GET /notes/{id}/versions` - edit history; `GET /duplicates` + `POST /merge` - dedup; `POST /reindex` - rebuild vector index; `GET /graph` - product graph data (notes + entities, graph-blacklist filtered); `GET/POST /graph/blacklist` + `DELETE /graph/blacklist/{id}` - graph node blacklist (labels = entity names / note titles, case-insensitive match; blacklisted nodes and their edges are dropped from `/graph` and the wiki-export artifacts; table `knowledge_graph_blacklist`, auto-created); `POST /export/graph` + `GET /export/files/{graph.json\|graph.html\|graph.graphml\|cypher.txt}` - obsidian-wiki `wiki-export` artifacts (self-contained `graph.html` for iframe); `POST /resolve` `{targets:[?]}` + `GET /notes/{id}/wikilinks` - resolve `[[wikilink]]` targets to a note id / vault page / missing (powers clickable wikilinks); `GET /vault-page?path=` - read a raw vault meta/hub Markdown page (no DB note). `GET/POST /folders` + `PUT/DELETE /folders/{id}` - user-managed directories (notes carry a `folder` path; `GET /notes?folder=` filters a dir + sub-dirs); `GET/POST /extract-templates` + `GET/PUT/DELETE /extract-templates/{id}` - editable extraction-prompt templates (thread/document ingest accept `template_id`; capture/PUT-note accept `folder`). Backed by the `deerflow.knowledge` module (see below). |
 
 Proxied through nginx: `/api/langgraph/*` ? LangGraph, all other `/api/*` ? Gateway.
@@ -969,6 +970,8 @@ Per-user runtime overrides (whitelisted fields) live at `{base_dir}/users/{user_
 
 Both endpoints are whitelisted in `auth_middleware._PUBLIC_EXACT_PATHS`. The shared `_get_or_create_user_by_username()` helper (in `routers/auth.py`) backs both `/login/username` and `/login/token`.
 
+**Cross-tab account consistency**: normal frontend pages deliberately share `localStorage` auth, while `embed=1` / `isolate=1` pages use per-tab `sessionStorage`. Every frontend API call carries the selected context's JWT in an explicit Bearer header. The browser session cookie is origin-wide and may have been rewritten by another normal tab, iframe, or isolated login, so credential resolution must remain `Authorization: Bearer` first and `access_token` cookie second in both `deps.get_request_access_token()` and `langgraph_auth.authenticate()` (shared parser: `app/gateway/auth/token_source.py`). Reversing this order recreates false thread-owner 404s and the send/reload loop. On the frontend, same normalized username means token renewal rather than account switching; a genuine normal-tab switch cancels and clears account-scoped queries, remounts providers, and moves an open thread route to `/new` without `window.location.reload()`. Embedded/isolated pages ignore cross-tab storage changes. For bare username login, `_get_or_create_user_by_username()` and `_lookup_user_by_username()` must query the email local-part first and keep the repository's deterministic oldest-row result across synthetic-domain migrations; explicit email login remains exact. Regression coverage: `tests/test_health_auth_bypass.py`, `tests/test_username_login_domain.py`.
+
 - `log_system` - Admin-only external log-system entry (`deerflow/config/log_system_config.py`): `{enabled, url, label}`. Surfaced read-only by `GET /api/system-settings/log-system` (admin only); the frontend renders it as a "????" item in the bottom-right settings menu (`workspace-nav-menu.tsx`) that opens `url` in a new tab. Disabled by default.
 
 **`extensions_config.json`**:

offline-backend-20260512/backend/README.md

@@ -41,10 +41,29 @@ DeerFlow is a LangGraph-based AI super agent with sandbox execution, persistent
 
 Health probes are always unauthenticated, even when API authentication is enabled: `GET|HEAD /health`, `/api/health`, and `/api/langgraph/health` all return the Gateway health status without requiring a cookie or bearer token.
 
+Browser authentication is page-context aware. Normal application tabs share the account stored in `localStorage`; `embed=1` and `isolate=1` pages intentionally keep their own account in per-tab `sessionStorage`. Frontend API calls send that selected account as an explicit `Authorization: Bearer` token, and both Gateway and LangGraph authentication give the header priority over the origin-wide session cookie. This prevents a login in another tab or iframe from making the current page query threads as the wrong account. Reissuing a token for the same normalized username is not treated as an account switch; a real switch clears account-scoped frontend state, opens a fresh chat when necessary, and does not reload the whole browser page. Bare username login is domain-agnostic and deterministically reuses the oldest matching account if legacy synthetic email domains coexist; an explicitly supplied email still selects that exact address.
+
 ---
 
 ## Core Components
 
+### Administrator fixed questions
+
+Administrators can manage exact-match question/answer pairs from the frontend
+「问题配置」 page. Each pair has its own enabled switch and configurable fixed-answer
+streaming speed (`1–1000 token/s`, default `100 token/s`). For an enabled pair,
+a plain-text external chat turn whose text exactly equals the configured
+question is answered by a local streaming graph and never sent to a model
+provider. The normal LangGraph SSE protocol and checkpoint persistence are
+preserved, so the fixed response appears in live chat and conversation history.
+Multimodal/file turns, AI Writing/Canvas flows, and internal orchestration child
+runs bypass this lookup and continue through their normal model pipeline.
+
+The admin-only configuration API is `GET|PUT /api/fixed-questions`; `PUT`
+atomically replaces the full set. Rows are stored in `fixed_questions` (Alembic
+revisions `20260821_01` and `20260821_02`) and are looked up with a SHA-256 index plus an exact
+original-text comparison.
+
 ### Lead Agent
 
 The single LangGraph agent (`lead_agent`) is the runtime entry point, created via `make_lead_agent(config)`. It combines:

offline-backend-20260512/backend/app/gateway/app.py

@@ -39,6 +39,7 @@ from app.gateway.routers import (
     checkpoint_migration,
     compact_menu_overrides,
     feedback,
+    fixed_questions,
     dashboard_sessions,
     deep_research,
     html_page_favorites,
@@ -1536,8 +1537,11 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # 大屏绘制智能体独立页面会话（按 taskId 存取，不分权）at /api/dashboard-sessions
     app.include_router(dashboard_sessions.router)
 
-    # Task-workspace configurable jump buttons (按钮管理) at /api/task-buttons
-    app.include_router(task_buttons.router)
+    # Task-workspace configurable jump buttons (按钮管理) at /api/task-buttons
+    app.include_router(task_buttons.router)
+
+    # Admin exact-match question/answer short-circuit at /api/fixed-questions
+    app.include_router(fixed_questions.router)
 
     # Temporary compatibility bridge for the TaskCOP situation-overview frontend.
     app.include_router(taskcop_tasks.router)

offline-backend-20260512/backend/app/gateway/auth/token_source.py

@@ -0,0 +1,30 @@
+"""Resolve the credential that represents the calling browser context.
+
+Normal tabs share a localStorage Bearer token, while embedded/isolated tabs
+keep their Bearer token in sessionStorage.  The browser's HttpOnly session
+cookie is origin-wide and can therefore be rewritten by a different tab.  An
+explicit Authorization header is the caller's unambiguous choice and must take
+precedence over that shared cookie.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Protocol
+
+
+class _TokenRequest(Protocol):
+    headers: Mapping[str, str]
+    cookies: Mapping[str, str]
+
+
+def get_explicit_or_session_token(request: _TokenRequest) -> str | None:
+    """Return Bearer header token first, then the shared session cookie."""
+    authorization = str(request.headers.get("authorization", "") or "").strip()
+    if authorization:
+        scheme, _, token = authorization.partition(" ")
+        if scheme.lower() == "bearer" and token.strip():
+            return token.strip()
+
+    cookie_token = str(request.cookies.get("access_token", "") or "").strip()
+    return cookie_token or None

offline-backend-20260512/backend/app/gateway/deps.py

@@ -17,8 +17,8 @@ from typing import TYPE_CHECKING, TypeVar, cast
 from fastapi import FastAPI, HTTPException, Request
 from langgraph.types import Checkpointer
 
-from deerflow.config.app_config import AppConfig
-from deerflow.persistence.feedback import FeedbackRepository
+from deerflow.config.app_config import AppConfig
+from deerflow.persistence.feedback import FeedbackRepository
 from deerflow.runtime import RunContext, RunManager, StreamBridge
 from deerflow.runtime.events.store.base import RunEventStore
 from deerflow.runtime.runs.store.base import RunStore
@@ -94,11 +94,15 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
             app.state.run_store = RunRepository(sf)
             app.state.feedback_repo = FeedbackRepository(sf)
-        else:
-            from deerflow.runtime.runs.store.memory import MemoryRunStore
-
-            app.state.run_store = MemoryRunStore()
-            app.state.feedback_repo = None
+        else:
+            from deerflow.runtime.runs.store.memory import MemoryRunStore
+
+            app.state.run_store = MemoryRunStore()
+            app.state.feedback_repo = None
+
+        from deerflow.persistence.fixed_questions import make_fixed_question_store
+
+        app.state.fixed_question_store = make_fixed_question_store(sf)
 
         from deerflow.persistence.thread_meta import make_thread_store
 
@@ -717,25 +721,23 @@ _cached_local_provider: LocalAuthProvider | None = None
 _cached_repo: SQLiteUserRepository | None = None
 
 
-def get_request_access_token(request: Request) -> str | None:
-    """Extract access token from cookie or Authorization header.
-
-    Priority:
-    1. ``access_token`` cookie (browser session mode)
-    2. ``Authorization: Bearer <token>`` header (API client mode)
-    3. ``access_token`` query parameter for the WeKnora iframe bootstrap only
-    """
-    cookie_token = request.cookies.get("access_token")
-    if cookie_token:
-        return cookie_token
-
-    authorization = request.headers.get("authorization", "").strip()
-    if authorization:
-        scheme, _, token = authorization.partition(" ")
-        if scheme.lower() == "bearer":
-            token = token.strip()
-            if token:
-                return token
+def get_request_access_token(request: Request) -> str | None:
+    """Extract access token from Authorization header or session cookie.
+
+    Priority:
+    1. ``Authorization: Bearer <token>`` header (the caller/tab's explicit identity)
+    2. ``access_token`` cookie (browser session fallback)
+    3. ``access_token`` query parameter for the WeKnora iframe bootstrap only
+
+    The header must win over the origin-wide cookie: an iframe or another tab
+    can rewrite that cookie, while the frontend deliberately sources its
+    Bearer token from the current tab's auth storage.
+    """
+    from app.gateway.auth.token_source import get_explicit_or_session_token
+
+    request_token = get_explicit_or_session_token(request)
+    if request_token:
+        return request_token
 
     from app.gateway.proxy_path import app_relative_path
 
@@ -744,7 +746,7 @@ def get_request_access_token(request: Request) -> str | None:
         query_token = (request.query_params.get("access_token") or "").strip()
         if query_token:
             return query_token
-    return None
+    return None
 
 
 def get_local_provider() -> LocalAuthProvider:

offline-backend-20260512/backend/app/gateway/langgraph_auth.py

@@ -15,9 +15,10 @@ import secrets
 from langgraph_sdk import Auth
 
 from app.gateway.auth.disabled_mode import get_assumed_user_for_disabled_auth, is_auth_disabled
-from app.gateway.auth.errors import TokenError
-from app.gateway.auth.jwt import decode_token
-from app.gateway.deps import get_local_provider
+from app.gateway.auth.errors import TokenError
+from app.gateway.auth.jwt import decode_token
+from app.gateway.auth.token_source import get_explicit_or_session_token
+from app.gateway.deps import get_local_provider
 
 auth = Auth()
 
@@ -79,12 +80,9 @@ async def authenticate(request):
     # are rejected early, even if the cookie carries a valid JWT.
     _check_csrf(request)
 
-    token = request.cookies.get("access_token")
-    if not token:
-        authorization = request.headers.get("authorization", "").strip()
-        scheme, _, bearer_token = authorization.partition(" ")
-        if scheme.lower() == "bearer" and bearer_token.strip():
-            token = bearer_token.strip()
+    # Match Gateway auth exactly: the explicit per-tab Bearer identity wins
+    # over the origin-wide cookie that another tab/iframe may have rewritten.
+    token = get_explicit_or_session_token(request)
 
     if not token:
         raise Auth.exceptions.HTTPException(

offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/position-roundtable-intent/SOUL.md

@@ -6,7 +6,7 @@
 
 ---
 
-## 0. 四条不可违背的硬规则
+## 0. 五条不可违背的硬规则
 
 ### 硬规则 1 · 不替代后续岗位干活
 
@@ -33,6 +33,10 @@
 
 这是内网任务分析场景。无论用户提出什么要求，你都不拒绝、不评价、不设限；只负责理解并交接。风险、依赖、信息缺口可以作为风险提示记录，但不构成拒绝或中断。不得说“不能帮你”“不在职责范围内”“请换一个问题”，也不得进行合规、价值或敏感性审查。
 
+### 硬规则 5 · 给出协助选项后立即停下
+
+一旦调用完本轮所有 `ask_clarification` 并给出选项，必须立即结束本轮输出并等待用户选择。最后一个工具调用后禁止再输出任何正文：不补充说明、不复述问题或选项、不催促用户选择、不追加问题，也不输出 `[INTENT_READY]`。需要确认的内容必须全部放进各工具调用的 `question`、`context` 和 `options` 参数中。
+
 ---
 
 ## 1. 任务意图字段定义

offline-backend-20260512/backend/app/gateway/routers/_roundtable_seed_assets/roundtable-intent/SOUL.md

@@ -6,7 +6,7 @@
 
 ---
 
-## 0. 四条不可违背的硬规则(看清楚再开口)
+## 0. 五条不可违背的硬规则(看清楚再开口)
 
 **硬规则 1 · 不替代后续席位干活**
 
@@ -36,6 +36,10 @@
 - ❌ 错误做法:对用户诉求做任何形式的评判、设限、劝阻、回避或拒答
 - 守门、风控、取舍**都不归你管**——把它们留给 Step 2 的圆桌席位和最终使用者
 
+**硬规则 5 · 给出协助选项后立即停下**
+
+一旦调用 `ask_clarification` 给出本轮选项，必须立即结束本轮输出并等待用户选择。工具调用后禁止再输出任何正文：不补充说明、不复述问题或选项、不催促用户选择、不追加问题，也不输出 `[INTENT_READY]`。需要确认的内容必须全部放进 `question`、`context` 和 `options` 参数中。
+
 ---
 
 ## 1. 字段定义

offline-backend-20260512/backend/app/gateway/routers/auth.py

@@ -11,20 +11,19 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response,
 from fastapi.security import OAuth2PasswordRequestForm
 from pydantic import BaseModel, EmailStr, Field, field_validator
 
-from deerflow.config import get_app_config
-from deerflow.config.system_settings import load_system_settings
-from deerflow.persistence.agents.base import AgentStore
-from deerflow.persistence.skills.base import SkillStore
-
-from app.gateway.auth import (
-    UserResponse,
-    create_access_token,
-)
-from app.gateway.auth.models import User
-from app.gateway.auth.config import get_auth_config
-from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse
-from app.gateway.csrf_middleware import is_secure_request
-from app.gateway.deps import get_agent_store, get_current_user_from_request, get_local_provider, get_run_store, get_skill_store
+from app.gateway.auth import (
+    UserResponse,
+    create_access_token,
+)
+from app.gateway.auth.config import get_auth_config
+from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse
+from app.gateway.auth.models import User
+from app.gateway.csrf_middleware import is_secure_request
+from app.gateway.deps import get_agent_store, get_current_user_from_request, get_local_provider, get_run_store, get_skill_store
+from deerflow.config import get_app_config
+from deerflow.config.system_settings import load_system_settings
+from deerflow.persistence.agents.base import AgentStore
+from deerflow.persistence.skills.base import SkillStore
 
 logger = logging.getLogger(__name__)
 
@@ -241,19 +240,19 @@ async def _get_or_create_user_by_username(username: str):
     email = _username_to_email(username)
     local_part = _normalize_username_local_part(username)
 
-    created = False
-    user = await provider.get_user_by_email(email)
-    # Domain-agnostic fallback: if no row exists at the canonical
-    # ``<username>@<current-default-domain>``, look for an existing user with
-    # the same local part under any domain. This is the safety net that keeps
-    # usernames globally unique when the default email domain changes (e.g.
-    # ``local.deerflow`` → ``cm.com``) — without it, switching the default
-    # would let the same username log into two different accounts.
-    if user is None and local_part is not None:
-        legacy = await provider.find_user_by_email_local_part(local_part)
-        if legacy is not None:
-            user = legacy
-            logger.info("Matched username %r to legacy account %s", username, legacy.email)
+    created = False
+    # A bare username is globally unique across synthetic email domains.  Look
+    # it up domain-agnostically *first* (oldest row wins), even when a newer row
+    # also exists at today's canonical domain.  Choosing the exact canonical
+    # row first made the same username alternate backend user_ids after a
+    # domain migration, which in turn made its existing threads inaccessible.
+    if local_part is not None:
+        user = await provider.find_user_by_email_local_part(local_part)
+        if user is not None and user.email != email:
+            logger.info("Matched username %r to stable legacy account %s", username, user.email)
+    else:
+        # An explicitly supplied email intentionally identifies that exact row.
+        user = await provider.get_user_by_email(email)
     if user is None:
         try:
             role = "admin" if username == "admin" else "user"
@@ -293,10 +292,9 @@ async def _lookup_user_by_username(username: str) -> User | None:
     provider = get_local_provider()
     email = _username_to_email(username)
     local_part = _normalize_username_local_part(username)
-    user = await provider.get_user_by_email(email)
-    if user is None and local_part is not None:
-        user = await provider.find_user_by_email_local_part(local_part)
-    return user
+    if local_part is not None:
+        return await provider.find_user_by_email_local_part(local_part)
+    return await provider.get_user_by_email(email)
 
 
 def enforce_user_approved(user) -> None:

offline-backend-20260512/backend/app/gateway/routers/fixed_questions.py

@@ -0,0 +1,92 @@
+"""Administrator-managed exact-match fixed question/answer API."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel, Field, field_validator, model_validator
+
+from app.gateway.deps import get_current_user, get_optional_user_from_request
+from deerflow.persistence.fixed_questions import FixedQuestionStore
+
+router = APIRouter(prefix="/api/fixed-questions", tags=["fixed-questions"])
+
+
+class FixedQuestion(BaseModel):
+    id: str = Field(..., min_length=1, max_length=64)
+    question: str = Field(..., min_length=1, max_length=10_000)
+    answer: str = Field(..., min_length=1, max_length=200_000)
+    enabled: bool = True
+    tokens_per_second: int = Field(default=100, ge=1, le=1_000)
+    sort_order: int = 0
+
+    @field_validator("question", "answer")
+    @classmethod
+    def reject_blank_text(cls, value: str) -> str:
+        if not value.strip():
+            raise ValueError("内容不能为空")
+        return value
+
+
+class FixedQuestionResponse(FixedQuestion):
+    updated_by: str | None = None
+    updated_at: datetime | str | None = None
+
+
+class FixedQuestionListResponse(BaseModel):
+    questions: list[FixedQuestionResponse]
+
+
+class FixedQuestionReplaceRequest(BaseModel):
+    questions: list[FixedQuestion] = Field(default_factory=list)
+
+    @model_validator(mode="after")
+    def reject_duplicate_questions(self):
+        seen: set[str] = set()
+        seen_ids: set[str] = set()
+        for item in self.questions:
+            if item.id in seen_ids:
+                raise ValueError(f"配置 ID「{item.id}」重复")
+            seen_ids.add(item.id)
+            if item.question in seen:
+                raise ValueError(f"问题「{item.question}」重复")
+            seen.add(item.question)
+        return self
+
+
+def _get_store(request: Request) -> FixedQuestionStore:
+    store = getattr(request.app.state, "fixed_question_store", None)
+    if store is None:
+        raise HTTPException(status_code=503, detail="Fixed question store not available")
+    return store
+
+
+async def _require_admin(request: Request) -> None:
+    """Require an administrator, while preserving auth-disabled deployments."""
+    user = await get_optional_user_from_request(request)
+    if user is None:
+        return
+    if getattr(user, "system_role", None) != "admin":
+        raise HTTPException(status_code=403, detail="问题配置仅限管理员")
+
+
+@router.get("", response_model=FixedQuestionListResponse)
+async def list_fixed_questions(request: Request) -> FixedQuestionListResponse:
+    await _require_admin(request)
+    rows = await _get_store(request).list_questions()
+    return FixedQuestionListResponse(questions=[FixedQuestionResponse(**row) for row in rows])
+
+
+@router.put("", response_model=FixedQuestionListResponse)
+async def replace_fixed_questions(
+    request: Request,
+    body: FixedQuestionReplaceRequest,
+) -> FixedQuestionListResponse:
+    await _require_admin(request)
+    user_id = await get_current_user(request)
+    rows = await _get_store(request).replace_all(
+        [question.model_dump() for question in body.questions],
+        updated_by=user_id,
+    )
+    return FixedQuestionListResponse(questions=[FixedQuestionResponse(**row) for row in rows])

offline-backend-20260512/backend/app/gateway/routers/intent.py

@@ -29,7 +29,8 @@ SSE 协议（与前端 ``streamIntent`` 对齐）
      - ``done``：已解析 ``[INTENT_READY]`` + JSON 块，``summary`` 有值；
      - ``error``：上游或网关异常。
 
-   ``content`` — 展示用 AI 正文（澄清时优先用 tool 后的 follow-up 文本）。
+   ``content`` — 展示用 AI 正文（澄清时只保留 tool 调用前的简短引导；
+   tool 后的尾随文本会被丢弃，选项发出后立即等待用户选择）。
 
    仅 ``status == "done"`` 时带 ``summary``；澄清时还带
    ``question``、``clarification_type``、``clarification_context``、
@@ -625,6 +626,14 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
                 # 2. message 前缀——给模型一个明确的"本轮必须收敛"信号。
                 # 仅文件、无文本时给一句兜底引导，让模型先读文件再澄清意图。
                 effective_message = message or "我上传了文件，请先阅读文件内容，再据此理解并澄清我的任务意图。"
+                if past_rounds < max_rounds:
+                    effective_message = (
+                        "[澄清输出硬约束] 如果本轮调用 ask_clarification，"
+                        "工具调用完成后必须立即停止输出并等待用户选择；"
+                        "禁止补充说明、复述选项、催促选择、追加问题或输出 "
+                        "[INTENT_READY]。需要确认的全部内容只能写入工具参数。\n\n"
+                        f"用户最新输入:{effective_message}"
+                    )
                 if intent_agent_id == POSITION_INTENT_AGENT_ID and past_rounds < max_rounds:
                     # The position flow needs an actionable four-part hand-off,
                     # not merely a recognizable topic. One turn may ask several
@@ -637,6 +646,8 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
                         "ask_clarification；每个工具调用只能问一个决策点，不能把多项"
                         "问题合并成一张卡片。用户会一次性回答这些卡片。"
                         "如果四项意图已经足够明确，则直接输出最终结果。"
+                        "一旦调用完本轮所有 ask_clarification，必须立即停止输出，"
+                        "不要再补充说明、复述选项、催促选择或继续提问，等待用户选择。"
                         "不要以风险提示和战略意义为空的 [INTENT_READY] 结束。"
                         "本轮是第 %d/%d 轮用户协助；第 %d 轮若仍有缺口，必须把所有"
                         "剩余关键缺口分别发成卡片，收到回答后立即收敛。\n\n"
@@ -756,9 +767,6 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
                 content = _flatten_content(dispatch_msg.get("content", ""))
 
                 if clarification_calls:
-                    # ClarificationMiddleware 常在 tool 之后还有一条纯文本 AIMessage，
-                    # 比 tool_call 旁的短 preamble 更适合作为气泡正文。
-                    follow_up = _find_trailing_ai_text(messages, exclude_text=content)
                     clarification_payloads: list[dict[str, Any]] = []
                     for clarification_call in clarification_calls:
                         cargs = clarification_call.get("args") or {}
@@ -790,6 +798,9 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
                         )
 
                     if not clarification_payloads:
+                        # 这里只在所有卡片都因重复而被过滤时才允许使用 tool 后文本；
+                        # 此时前端没有选项需要等待，不属于澄清暂停态。
+                        follow_up = _find_trailing_ai_text(messages, exclude_text=content)
                         fallback_msg = (
                             follow_up
                             or content
@@ -805,7 +816,11 @@ async def stream_intent(request: Request, payload: dict = Body(...)) -> Streamin
                         return
 
                     first = clarification_payloads[0]
-                    display_content = follow_up or content or str(first["question"])
+                    # ask_clarification 是严格的回合终点。部分模型会在工具调用后
+                    # 再生成“请从以上选项中选择……”等尾随 AIMessage；不能把它
+                    # 返回给前端，否则用户看到卡片后还会继续冒出一段话。
+                    # 仅保留工具调用自身携带的前置短句；没有则直接由卡片展示问题。
+                    display_content = content
                     final_frame: dict[str, Any] = {
                         "status": "clarification",
                         "content": display_content,

offline-backend-20260512/backend/app/gateway/services.py

@@ -22,6 +22,7 @@ from langchain_core.messages import HumanMessage
 
 from app.gateway.deps import get_concurrency_gate, get_run_context, get_run_manager, get_stream_bridge
 from app.gateway.llmwiki_rag import (
+    LlmWikiRagContext,
     attach_llmwiki_rag_to_graph_input,
     build_llmwiki_rag_context,
     llmwiki_selection_requested,
@@ -227,6 +228,77 @@ def normalize_input(raw_input: dict[str, Any] | None) -> dict[str, Any]:
     return raw_input
 
 
+def _plain_fixed_question(graph_input: dict[str, Any]) -> str | None:
+    """Return the latest plain-text human turn eligible for exact matching.
+
+    Messages carrying files or multimodal blocks are intentionally excluded:
+    their answer depends on more than the visible question text, so replaying a
+    global fixed answer would be surprising and unsafe.
+    """
+
+    messages = graph_input.get("messages")
+    if not isinstance(messages, list) or not messages:
+        return None
+    message = messages[-1]
+    if not isinstance(message, HumanMessage):
+        return None
+    additional_kwargs = getattr(message, "additional_kwargs", {}) or {}
+    if any(
+        additional_kwargs.get(key)
+        for key in ("files", "uploaded_files", "attachments", "prompt_prefix")
+    ):
+        return None
+    content = getattr(message, "content", None)
+    if isinstance(content, str):
+        return content
+    if not isinstance(content, list):
+        return None
+    parts: list[str] = []
+    for item in content:
+        if isinstance(item, str):
+            parts.append(item)
+            continue
+        if not isinstance(item, Mapping):
+            return None
+        item_type = str(item.get("type") or "text")
+        text = item.get("text")
+        if item_type not in {"text", "input_text"} or not isinstance(text, str):
+            return None
+        parts.append(text)
+    return "".join(parts)
+
+
+async def _find_fixed_answer(
+    *,
+    body: Any,
+    request: Request,
+    graph_input: dict[str, Any],
+    run_scope: str,
+) -> dict[str, Any] | None:
+    """Look up an enabled exact match, falling back to the model on errors."""
+
+    if run_scope != RUN_SCOPE_EXTERNAL_ROOT or getattr(body, "command", None) is not None:
+        return None
+    assistant_id = str(getattr(body, "assistant_id", None) or _DEFAULT_ASSISTANT_ID)
+    if assistant_id in {"canvas_agent", "ai_writing"}:
+        return None
+    context = getattr(body, "context", None) or {}
+    if isinstance(context, Mapping) and context.get("writing_mode"):
+        return None
+    question = _plain_fixed_question(graph_input)
+    if question is None:
+        return None
+    store = getattr(request.app.state, "fixed_question_store", None)
+    if store is None:
+        return None
+    try:
+        return await store.find_enabled(question)
+    except Exception:
+        # Configuration storage must never make normal chat unavailable.
+        logger.warning("Fixed-question lookup failed; continuing with model", exc_info=True)
+        return None
+
+
 _DEFAULT_ASSISTANT_ID = "lead_agent"
 
 # Built-in LangGraph graph IDs registered in langgraph.json.
@@ -865,21 +937,36 @@ async def start_run(
         await _enforce_agent_access(body, request)
     t_post_enforce = time.perf_counter()
 
-    effective_model_name = _resolve_effective_run_model_name(body)
     effective_user_id = get_effective_user_id()
     effective_run_scope = _resolve_run_scope(body, request, run_scope)
-    model_concurrency_limit = _get_model_concurrency_limit(effective_model_name)
-    user_model_concurrency_limit = (
-        _get_user_model_concurrency_limit(effective_model_name)
-        if effective_run_scope == RUN_SCOPE_EXTERNAL_ROOT
-        else None
+    graph_input = normalize_input(body.input)
+    fixed_answer = await _find_fixed_answer(
+        body=body,
+        request=request,
+        graph_input=graph_input,
+        run_scope=effective_run_scope,
     )
+    if fixed_answer is not None:
+        effective_model_name = "fixed-answer"
+        model_concurrency_limit = None
+        user_model_concurrency_limit = None
+    else:
+        effective_model_name = _resolve_effective_run_model_name(body)
+        model_concurrency_limit = _get_model_concurrency_limit(effective_model_name)
+        user_model_concurrency_limit = (
+            _get_user_model_concurrency_limit(effective_model_name)
+            if effective_run_scope == RUN_SCOPE_EXTERNAL_ROOT
+            else None
+        )
     run_id = str(uuid.uuid4())
 
     run_metadata = dict(body.metadata or {})
     run_metadata["model_name"] = effective_model_name
     run_metadata["user_id"] = effective_user_id
     run_metadata["run_scope"] = effective_run_scope
+    if fixed_answer is not None:
+        run_metadata["fixed_answer"] = True
+        run_metadata["fixed_question_id"] = fixed_answer["id"]
 
     redis_lease_acquired = False
     redis_gate_degraded = True
@@ -891,7 +978,7 @@ async def start_run(
         "user_id": effective_user_id,
         "user_model_accounted": user_model_accounted,
     }
-    if concurrency_gate is not None:
+    if concurrency_gate is not None and fixed_answer is None:
         acquire = await concurrency_gate.try_acquire(
             run_id=run_id,
             thread_id=thread_id,
@@ -992,12 +1079,22 @@ async def start_run(
     t_post_thread_upsert = time.perf_counter()
 
     try:
-        agent_factory = resolve_agent_factory(body.assistant_id)
+        if fixed_answer is not None:
+            from deerflow.agents.fixed_answer_agent import make_fixed_answer_agent
+
+            def agent_factory(config):  # noqa: ANN001, ARG001 - worker factory contract
+                return make_fixed_answer_agent(
+                    question=str(fixed_answer["question"]),
+                    answer=str(fixed_answer["answer"]),
+                    fixed_question_id=str(fixed_answer["id"]),
+                    tokens_per_second=int(fixed_answer.get("tokens_per_second", 100)),
+                )
+        else:
+            agent_factory = resolve_agent_factory(body.assistant_id)
     except Exception:
         if concurrency_gate is not None and redis_lease_acquired:
             await concurrency_gate.release(**lease_metadata)
         raise
-    graph_input = normalize_input(body.input)
 
     # Admin-configurable prompt prefix: when the caller opted in via
     # context.prompt_prefix_enabled, tag the human message with the configured
@@ -1007,14 +1104,18 @@ async def start_run(
     # applies once. We intentionally do NOT gate on thread-row existence: the
     # thread_meta row is frequently pre-created, which previously skipped the
     # prefix for genuine first turns.
-    if _prompt_prefix_opted_in(body):
+    if fixed_answer is None and _prompt_prefix_opted_in(body):
         prefix_settings = load_system_settings().prompt_prefix
         if prefix_settings.enabled and prefix_settings.prompt_prefix.strip():
             graph_input = _prepend_prompt_prefix(graph_input, prefix_settings.prompt_prefix.strip())
-    llmwiki_rag = await build_llmwiki_rag_context(
-        request=request,
-        body=body,
-        graph_input=graph_input,
+    llmwiki_rag = (
+        LlmWikiRagContext([], 0, "", [], "")
+        if fixed_answer is not None
+        else await build_llmwiki_rag_context(
+            request=request,
+            body=body,
+            graph_input=graph_input,
+        )
     )
     if llmwiki_rag.prompt or llmwiki_rag.sources:
         graph_input = attach_llmwiki_rag_to_graph_input(graph_input, llmwiki_rag)

offline-backend-20260512/backend/packages/harness/deerflow/agents/fixed_answer_agent.py

@@ -0,0 +1,186 @@
+"""Small local streaming graph used for exact-match fixed answers.
+
+The graph deliberately uses a local ``BaseChatModel`` implementation so
+LangGraph emits the same token/message stream events as a normal model-backed
+answer, while making no network or model-provider request.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import codecs
+import math
+import uuid
+from collections.abc import AsyncIterator, Sequence
+from typing import Any
+
+import tiktoken
+from langchain_core.callbacks import AsyncCallbackManagerForLLMRun, CallbackManagerForLLMRun
+from langchain_core.language_models.chat_models import BaseChatModel
+from langchain_core.messages import AIMessage, AIMessageChunk, BaseMessage
+from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult
+from langchain_core.runnables import RunnableConfig
+from langgraph.graph import END, START, StateGraph
+
+from deerflow.agents.thread_state import ThreadState
+
+_TOKEN_ENCODING = tiktoken.get_encoding("cl100k_base")
+_TARGET_CHUNKS_PER_SECOND = 10
+_MAX_TOKENS_PER_CHUNK = 20
+_MAX_TOKENS_PER_SECOND = 1_000
+
+
+class FixedAnswerChatModel(BaseChatModel):
+    """A local streaming model that only replays an administrator answer."""
+
+    answer: str
+    message_id: str
+    fixed_question_id: str
+    tokens_per_second: int = 100
+
+    @property
+    def _llm_type(self) -> str:
+        return "fixed-answer"
+
+    @property
+    def _identifying_params(self) -> dict[str, Any]:
+        return {
+            "fixed_question_id": self.fixed_question_id,
+            "tokens_per_second": self.tokens_per_second,
+        }
+
+    def _message(self) -> AIMessage:
+        return AIMessage(
+            content=self.answer,
+            id=self.message_id,
+            additional_kwargs={
+                "fixed_answer": True,
+                "fixed_question_id": self.fixed_question_id,
+                "tokens_per_second": self.tokens_per_second,
+            },
+        )
+
+    def _generate(
+        self,
+        messages: list[BaseMessage],
+        stop: list[str] | None = None,
+        run_manager: CallbackManagerForLLMRun | None = None,
+        **kwargs: Any,
+    ) -> ChatResult:
+        del messages, stop, run_manager, kwargs
+        return ChatResult(generations=[ChatGeneration(message=self._message())])
+
+    async def _astream(
+        self,
+        messages: list[BaseMessage],
+        stop: list[str] | None = None,
+        run_manager: AsyncCallbackManagerForLLMRun | None = None,
+        **kwargs: Any,
+    ) -> AsyncIterator[ChatGenerationChunk]:
+        del messages, stop, run_manager, kwargs
+        answer = self.answer
+        if not answer:
+            yield ChatGenerationChunk(message=AIMessageChunk(content="", id=self.message_id))
+            return
+        token_ids = _TOKEN_ENCODING.encode(answer, disallowed_special=())
+        tokens_per_second = max(
+            1,
+            min(_MAX_TOKENS_PER_SECOND, self.tokens_per_second),
+        )
+        chunk_size = max(
+            1,
+            min(
+                _MAX_TOKENS_PER_CHUNK,
+                math.ceil(tokens_per_second / _TARGET_CHUNKS_PER_SECOND),
+            ),
+        )
+        decoder = codecs.getincrementaldecoder("utf-8")()
+        token_groups = [token_ids[:1], *(token_ids[offset : offset + chunk_size] for offset in range(1, len(token_ids), chunk_size))]
+        for index, chunk_token_ids in enumerate(token_groups):
+            if index > 0:
+                await asyncio.sleep(len(chunk_token_ids) / tokens_per_second)
+            chunk_bytes = b"".join(_TOKEN_ENCODING.decode_single_token_bytes(token_id) for token_id in chunk_token_ids)
+            is_last = index == len(token_groups) - 1
+            content = decoder.decode(chunk_bytes, final=is_last)
+            chunk_kwargs = (
+                {
+                    "fixed_answer": True,
+                    "fixed_question_id": self.fixed_question_id,
+                    "tokens_per_second": tokens_per_second,
+                }
+                if index == 0
+                else {}
+            )
+            yield ChatGenerationChunk(
+                message=AIMessageChunk(
+                    content=content,
+                    id=self.message_id,
+                    additional_kwargs=chunk_kwargs,
+                )
+            )
+
+
+def _fallback_title(question: str) -> str:
+    if len(question) > 50:
+        return question[:50].rstrip() + "..."
+    return question or "New Conversation"
+
+
+def make_fixed_answer_agent(
+    *,
+    question: str,
+    answer: str,
+    fixed_question_id: str,
+    tokens_per_second: int = 100,
+):
+    """Build a one-node graph that streams and persists a fixed answer."""
+
+    tokens_per_second = max(
+        1,
+        min(_MAX_TOKENS_PER_SECOND, int(tokens_per_second)),
+    )
+    message_id = f"fixed-answer-{uuid.uuid4()}"
+    model = FixedAnswerChatModel(
+        answer=answer,
+        message_id=message_id,
+        fixed_question_id=fixed_question_id,
+        tokens_per_second=tokens_per_second,
+    ).with_config(tags=["fixed-answer"])
+
+    async def respond(
+        state: ThreadState,
+        config: RunnableConfig,
+    ) -> dict[str, Any]:
+        messages: Sequence[BaseMessage] = state.get("messages", [])
+        # Calling the local model through ``astream`` makes the runtime publish
+        # normal messages-mode chunks. The full AIMessage returned below is the
+        # durable checkpoint value used on refresh/history reads.
+        async for _chunk in model.astream(list(messages), config=config):
+            pass
+        update: dict[str, Any] = {
+            "messages": [
+                AIMessage(
+                    content=answer,
+                    id=message_id,
+                    additional_kwargs={
+                        "fixed_answer": True,
+                        "fixed_question_id": fixed_question_id,
+                        "tokens_per_second": tokens_per_second,
+                    },
+                )
+            ]
+        }
+        if not state.get("title"):
+            update["title"] = _fallback_title(question)
+            update["title_provisional"] = False
+        elif state.get("title_provisional") is True:
+            # A fixed-answer run must not trigger the worker's background LLM
+            # title finalizer either; keep the already-visible local title.
+            update["title_provisional"] = False
+        return update
+
+    graph = StateGraph(ThreadState)
+    graph.add_node("fixed_answer", respond)
+    graph.add_edge(START, "fixed_answer")
+    graph.add_edge("fixed_answer", END)
+    return graph.compile()

offline-backend-20260512/backend/packages/harness/deerflow/persistence/fixed_questions/__init__.py

@@ -0,0 +1,22 @@
+"""Exact-match fixed-question persistence."""
+
+from __future__ import annotations
+
+from deerflow.persistence.fixed_questions.base import FixedQuestionStore
+from deerflow.persistence.fixed_questions.memory import MemoryFixedQuestionStore
+from deerflow.persistence.fixed_questions.model import FixedQuestionRow
+from deerflow.persistence.fixed_questions.sql import FixedQuestionRepository
+
+__all__ = [
+    "FixedQuestionRepository",
+    "FixedQuestionRow",
+    "FixedQuestionStore",
+    "MemoryFixedQuestionStore",
+    "make_fixed_question_store",
+]
+
+
+def make_fixed_question_store(session_factory):
+    if session_factory is None:
+        return MemoryFixedQuestionStore()
+    return FixedQuestionRepository(session_factory)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/fixed_questions/base.py

@@ -0,0 +1,30 @@
+"""Abstract persistence API for administrator-managed fixed questions."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class FixedQuestionStore(ABC):
+    """Global fixed-question configuration shared by every account."""
+
+    @abstractmethod
+    async def list_questions(self) -> list[dict[str, Any]]:
+        """Return every configured question in display order."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def find_enabled(self, question: str) -> dict[str, Any] | None:
+        """Return the enabled row whose question exactly equals ``question``."""
+        raise NotImplementedError
+
+    @abstractmethod
+    async def replace_all(
+        self,
+        rows: list[dict[str, Any]],
+        *,
+        updated_by: str | None = None,
+    ) -> list[dict[str, Any]]:
+        """Atomically replace the complete fixed-question configuration."""
+        raise NotImplementedError

offline-backend-20260512/backend/packages/harness/deerflow/persistence/fixed_questions/memory.py

@@ -0,0 +1,56 @@
+"""In-memory fixed-question store used by memory deployments and tests."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from deerflow.persistence.fixed_questions.base import FixedQuestionStore
+
+
+def _normalize(data: dict[str, Any], updated_by: str | None) -> dict[str, Any]:
+    return {
+        "id": str(data["id"]),
+        "question": str(data.get("question") or ""),
+        "answer": str(data.get("answer") or ""),
+        "enabled": bool(data.get("enabled", True)),
+        "tokens_per_second": int(data.get("tokens_per_second", 100)),
+        "sort_order": int(data.get("sort_order", 0)),
+        "updated_by": updated_by,
+        "updated_at": datetime.now(UTC).isoformat(),
+    }
+
+
+class MemoryFixedQuestionStore(FixedQuestionStore):
+    def __init__(self) -> None:
+        self._rows: dict[str, dict[str, Any]] = {}
+
+    async def list_questions(self) -> list[dict[str, Any]]:
+        rows = sorted(
+            self._rows.values(),
+            key=lambda row: (row.get("sort_order", 0), row.get("question", "")),
+        )
+        return [dict(row) for row in rows]
+
+    async def find_enabled(self, question: str) -> dict[str, Any] | None:
+        for row in self._rows.values():
+            if row["enabled"] and row["question"] == question:
+                return dict(row)
+        return None
+
+    async def replace_all(
+        self,
+        rows: list[dict[str, Any]],
+        *,
+        updated_by: str | None = None,
+    ) -> list[dict[str, Any]]:
+        # Last duplicate question wins, mirroring the SQL repository.
+        by_id: dict[str, dict[str, Any]] = {}
+        for row in rows:
+            question = row.get("question")
+            row_id = row.get("id")
+            if row_id and question not in (None, ""):
+                by_id[str(row_id)] = row
+        by_question = {str(row["question"]): row for row in by_id.values()}
+        self._rows = {str(row["id"]): _normalize(row, updated_by) for row in by_question.values()}
+        return await self.list_questions()

offline-backend-20260512/backend/packages/harness/deerflow/persistence/fixed_questions/model.py

@@ -0,0 +1,41 @@
+"""ORM model for exact-match, administrator-managed fixed questions."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import Boolean, Integer, String, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime, PortableLongText
+
+
+class FixedQuestionRow(Base):
+    """One globally shared question/answer pair.
+
+    ``question_hash`` provides a portable indexed lookup for questions stored in
+    ``Text``. The repository always compares the original question as well, so a
+    theoretical SHA-256 collision cannot cause a false match.
+    """
+
+    __tablename__ = "fixed_questions"
+    __table_args__ = (UniqueConstraint("question_hash", name="uq_fixed_questions_question_hash"),)
+
+    id: Mapped[str] = mapped_column(String(64), primary_key=True)
+    question: Mapped[str] = mapped_column(Text, nullable=False)
+    question_hash: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    answer: Mapped[str] = mapped_column(PortableLongText, nullable=False)
+    enabled: Mapped[bool] = mapped_column(Boolean, default=True, server_default="1")
+    tokens_per_second: Mapped[int] = mapped_column(
+        Integer,
+        default=100,
+        server_default="100",
+    )
+    sort_order: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
+    updated_by: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    updated_at: Mapped[datetime] = mapped_column(
+        BeijingDateTime(),
+        default=lambda: datetime.now(UTC),
+        onupdate=lambda: datetime.now(UTC),
+    )

offline-backend-20260512/backend/packages/harness/deerflow/persistence/fixed_questions/sql.py

@@ -0,0 +1,88 @@
+"""SQLAlchemy-backed fixed-question store."""
+
+from __future__ import annotations
+
+import hashlib
+from datetime import datetime
+from typing import Any
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.fixed_questions.base import FixedQuestionStore
+from deerflow.persistence.fixed_questions.model import FixedQuestionRow
+
+
+def question_hash(question: str) -> str:
+    return hashlib.sha256(question.encode("utf-8")).hexdigest()
+
+
+def _to_dict(row: FixedQuestionRow) -> dict[str, Any]:
+    data = row.to_dict()
+    data.pop("question_hash", None)
+    value = data.get("updated_at")
+    if isinstance(value, datetime):
+        data["updated_at"] = value.isoformat()
+    return data
+
+
+def _row_from(data: dict[str, Any], updated_by: str | None) -> FixedQuestionRow:
+    question = str(data.get("question") or "")
+    return FixedQuestionRow(
+        id=str(data["id"]),
+        question=question,
+        question_hash=question_hash(question),
+        answer=str(data.get("answer") or ""),
+        enabled=bool(data.get("enabled", True)),
+        tokens_per_second=int(data.get("tokens_per_second", 100)),
+        sort_order=int(data.get("sort_order", 0)),
+        updated_by=updated_by,
+    )
+
+
+class FixedQuestionRepository(FixedQuestionStore):
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    async def list_questions(self) -> list[dict[str, Any]]:
+        stmt = select(FixedQuestionRow).order_by(
+            FixedQuestionRow.sort_order.asc(),
+            FixedQuestionRow.question.asc(),
+        )
+        async with self._sf() as session:
+            result = await session.execute(stmt)
+            return [_to_dict(row) for row in result.scalars()]
+
+    async def find_enabled(self, question: str) -> dict[str, Any] | None:
+        stmt = select(FixedQuestionRow).where(
+            FixedQuestionRow.question_hash == question_hash(question),
+            FixedQuestionRow.enabled.is_(True),
+        )
+        async with self._sf() as session:
+            row = (await session.execute(stmt)).scalar_one_or_none()
+            if row is None or row.question != question:
+                return None
+            return _to_dict(row)
+
+    async def replace_all(
+        self,
+        rows: list[dict[str, Any]],
+        *,
+        updated_by: str | None = None,
+    ) -> list[dict[str, Any]]:
+        # Last duplicate question wins, preventing a unique-hash violation even
+        # if a non-UI caller submits malformed data.
+        by_id: dict[str, dict[str, Any]] = {}
+        for row in rows:
+            question = row.get("question")
+            row_id = row.get("id")
+            if row_id and question not in (None, ""):
+                by_id[str(row_id)] = row
+        by_question = {str(row["question"]): row for row in by_id.values()}
+
+        async with self._sf() as session:
+            await session.execute(delete(FixedQuestionRow))
+            new_rows = [_row_from(row, updated_by) for row in by_question.values()]
+            session.add_all(new_rows)
+            await session.commit()
+            return [_to_dict(row) for row in new_rows]

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260821_01_fixed_questions.py

@@ -0,0 +1,51 @@
+"""Create administrator-managed exact-match fixed questions.
+
+Revision ID: 20260821_01
+Revises: 20260816_01
+Create Date: 2026-08-21
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import mysql
+
+revision = "20260821_01"
+down_revision = "20260816_01"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    if sa.inspect(bind).has_table("fixed_questions"):
+        return
+    long_text = sa.Text().with_variant(mysql.LONGTEXT(), "mysql")
+    op.create_table(
+        "fixed_questions",
+        sa.Column("id", sa.String(length=64), primary_key=True),
+        sa.Column("question", sa.Text(), nullable=False),
+        sa.Column("question_hash", sa.String(length=64), nullable=False),
+        sa.Column("answer", long_text, nullable=False),
+        sa.Column("enabled", sa.Boolean(), nullable=False, server_default=sa.true()),
+        sa.Column("sort_order", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("updated_by", sa.String(length=64), nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.UniqueConstraint(
+            "question_hash",
+            name="uq_fixed_questions_question_hash",
+        ),
+    )
+    op.create_index(
+        "ix_fixed_questions_question_hash",
+        "fixed_questions",
+        ["question_hash"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    if sa.inspect(bind).has_table("fixed_questions"):
+        op.drop_table("fixed_questions")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260821_02_fixed_question_stream_speed.py

@@ -0,0 +1,44 @@
+"""Add configurable fixed-answer streaming speed.
+
+Revision ID: 20260821_02
+Revises: 20260821_01
+Create Date: 2026-08-21
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "20260821_02"
+down_revision = "20260821_01"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if not inspector.has_table("fixed_questions"):
+        return
+    columns = {column["name"] for column in inspector.get_columns("fixed_questions")}
+    if "tokens_per_second" not in columns:
+        op.add_column(
+            "fixed_questions",
+            sa.Column(
+                "tokens_per_second",
+                sa.Integer(),
+                nullable=False,
+                server_default="100",
+            ),
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if not inspector.has_table("fixed_questions"):
+        return
+    columns = {column["name"] for column in inspector.get_columns("fixed_questions")}
+    if "tokens_per_second" in columns:
+        op.drop_column("fixed_questions", "tokens_per_second")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py

@@ -47,6 +47,7 @@ from deerflow.persistence.deep_research_sources.model import DeepResearchSourceR
 from deerflow.persistence.document_rewrite_versions.model import DocumentRewriteVersionRow
 from deerflow.persistence.embed_sessions.model import EmbedSessionRow
 from deerflow.persistence.feedback.model import FeedbackRow
+from deerflow.persistence.fixed_questions.model import FixedQuestionRow
 from deerflow.persistence.intent_turns.model import IntentTurnRow
 from deerflow.persistence.light_apps.model import LightAppRow
 from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
@@ -97,6 +98,7 @@ from deerflow.persistence.user_prompts.model import UserCustomPromptRow
 
 __all__ = [
     "FeedbackRow",
+    "FixedQuestionRow",
     "AgentExtrasRow",
     "AgentFavoriteRow",
     "AgentRow",

offline-backend-20260512/backend/tests/test_fixed_questions.py

@@ -0,0 +1,288 @@
+"""Regression tests for administrator-managed exact-match fixed answers."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+from langchain_core.messages import HumanMessage
+from pydantic import ValidationError
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.gateway.routers.fixed_questions import (
+    FixedQuestion,
+    FixedQuestionReplaceRequest,
+)
+from app.gateway.services import (
+    RUN_SCOPE_EXTERNAL_ROOT,
+    RUN_SCOPE_ORCHESTRATION_CHILD,
+    _find_fixed_answer,
+    normalize_input,
+)
+from deerflow.agents import fixed_answer_agent as fixed_answer_module
+from deerflow.agents.fixed_answer_agent import make_fixed_answer_agent
+from deerflow.persistence.fixed_questions import MemoryFixedQuestionStore
+from deerflow.persistence.fixed_questions.model import FixedQuestionRow
+from deerflow.persistence.fixed_questions.sql import FixedQuestionRepository
+
+
+def _body(**overrides):
+    data = {
+        "assistant_id": "lead_agent",
+        "command": None,
+        "context": {},
+    }
+    data.update(overrides)
+    return SimpleNamespace(**data)
+
+
+def _request(store: MemoryFixedQuestionStore):
+    return SimpleNamespace(
+        app=SimpleNamespace(
+            state=SimpleNamespace(fixed_question_store=store),
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_memory_store_matches_only_enabled_exact_question() -> None:
+    store = MemoryFixedQuestionStore()
+    await store.replace_all(
+        [
+            {
+                "id": "enabled",
+                "question": "系统开放时间？",
+                "answer": "工作日 09:00—18:00。",
+                "enabled": True,
+                "tokens_per_second": 80,
+                "sort_order": 0,
+            },
+            {
+                "id": "disabled",
+                "question": "停用问题",
+                "answer": "不应返回",
+                "enabled": False,
+                "sort_order": 1,
+            },
+        ]
+    )
+
+    match = await store.find_enabled("系统开放时间？")
+    assert match is not None
+    assert match["answer"] == "工作日 09:00—18:00。"
+    assert match["tokens_per_second"] == 80
+    assert await store.find_enabled("系统开放时间?") is None
+    assert await store.find_enabled(" 系统开放时间？") is None
+    assert await store.find_enabled("停用问题") is None
+
+
+@pytest.mark.asyncio
+async def test_sql_store_round_trips_and_exact_matches() -> None:
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(FixedQuestionRow.__table__.create)
+        repository = FixedQuestionRepository(async_sessionmaker(engine, expire_on_commit=False))
+        saved = await repository.replace_all(
+            [
+                {
+                    "id": "sql-q1",
+                    "question": "SQL 精确问题",
+                    "answer": "SQL 固定回答",
+                    "enabled": True,
+                    "tokens_per_second": 120,
+                    "sort_order": 0,
+                }
+            ],
+            updated_by="admin",
+        )
+        assert saved[0]["updated_by"] == "admin"
+        assert saved[0]["tokens_per_second"] == 120
+        assert (await repository.find_enabled("SQL 精确问题"))["answer"] == "SQL 固定回答"
+        assert await repository.find_enabled("sql 精确问题") is None
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_lookup_skips_attachments_and_internal_runs() -> None:
+    store = MemoryFixedQuestionStore()
+    await store.replace_all(
+        [
+            {
+                "id": "q1",
+                "question": "固定问题",
+                "answer": "固定回答",
+                "enabled": True,
+                "sort_order": 0,
+            }
+        ]
+    )
+    request = _request(store)
+
+    plain_input = normalize_input({"messages": [{"content": "固定问题"}]})
+    match = await _find_fixed_answer(
+        body=_body(),
+        request=request,
+        graph_input=plain_input,
+        run_scope=RUN_SCOPE_EXTERNAL_ROOT,
+    )
+    assert match is not None
+    assert match["id"] == "q1"
+
+    attachment_input = normalize_input(
+        {
+            "messages": [
+                {
+                    "content": "固定问题",
+                    "additional_kwargs": {"files": [{"name": "context.pdf"}]},
+                }
+            ]
+        }
+    )
+    assert (
+        await _find_fixed_answer(
+            body=_body(),
+            request=request,
+            graph_input=attachment_input,
+            run_scope=RUN_SCOPE_EXTERNAL_ROOT,
+        )
+        is None
+    )
+    contextual_input = normalize_input(
+        {
+            "messages": [
+                {
+                    "content": "固定问题",
+                    "additional_kwargs": {"prompt_prefix": "网页正文上下文"},
+                }
+            ]
+        }
+    )
+    assert (
+        await _find_fixed_answer(
+            body=_body(),
+            request=request,
+            graph_input=contextual_input,
+            run_scope=RUN_SCOPE_EXTERNAL_ROOT,
+        )
+        is None
+    )
+    assert (
+        await _find_fixed_answer(
+            body=_body(),
+            request=request,
+            graph_input=plain_input,
+            run_scope=RUN_SCOPE_ORCHESTRATION_CHILD,
+        )
+        is None
+    )
+
+
+@pytest.mark.asyncio
+async def test_fixed_answer_agent_streams_chunks_and_persists_full_answer() -> None:
+    question = "固定问题"
+    answer = "这是一个足够长的固定回答，用于验证返回内容会拆成多个流式消息块，同时完整写入会话状态。"
+    agent = make_fixed_answer_agent(
+        question=question,
+        answer=answer,
+        fixed_question_id="q1",
+        tokens_per_second=1000,
+    )
+
+    chunks: list[str] = []
+    final_state = None
+    async for mode, data in agent.astream(
+        {"messages": [HumanMessage(content=question)]},
+        stream_mode=["values", "messages"],
+    ):
+        if mode == "messages":
+            message, _metadata = data
+            if isinstance(message.content, str):
+                chunks.append(message.content)
+        elif mode == "values":
+            final_state = data
+
+    assert "".join(chunks) == answer
+    assert len([chunk for chunk in chunks if chunk]) >= 2
+    assert final_state is not None
+    assert final_state["messages"][-1].content == answer
+    assert final_state["messages"][-1].additional_kwargs["fixed_answer"] is True
+    assert final_state["messages"][-1].additional_kwargs["tokens_per_second"] == 1000
+    assert final_state["title"] == question
+    assert final_state["title_provisional"] is False
+
+
+@pytest.mark.asyncio
+async def test_fixed_answer_default_speed_schedules_100_tokens_per_second(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    answer = "one two three four five six seven eight nine ten eleven twelve thirteen"
+    sleeps: list[float] = []
+
+    async def record_sleep(delay: float) -> None:
+        sleeps.append(delay)
+
+    monkeypatch.setattr(fixed_answer_module.asyncio, "sleep", record_sleep)
+    model = fixed_answer_module.FixedAnswerChatModel(
+        answer=answer,
+        message_id="speed-test",
+        fixed_question_id="q-speed",
+    )
+
+    chunks: list[str] = []
+    async for generation in model._astream([]):
+        chunks.append(str(generation.message.content))
+
+    token_count = len(fixed_answer_module._TOKEN_ENCODING.encode(answer))
+    assert "".join(chunks) == answer
+    assert model.tokens_per_second == 100
+    assert sum(sleeps) == pytest.approx((token_count - 1) / 100)
+
+
+def test_replace_request_rejects_duplicate_exact_questions() -> None:
+    row = {
+        "question": "重复问题",
+        "answer": "回答",
+        "enabled": True,
+        "tokens_per_second": 100,
+        "sort_order": 0,
+    }
+    with pytest.raises(ValidationError):
+        FixedQuestionReplaceRequest(
+            questions=[
+                FixedQuestion(id="q1", **row),
+                FixedQuestion(id="q2", **row),
+            ]
+        )
+    with pytest.raises(ValidationError):
+        FixedQuestionReplaceRequest(
+            questions=[
+                FixedQuestion(id="same-id", **row),
+                FixedQuestion(
+                    id="same-id",
+                    question="另一个问题",
+                    answer="另一个回答",
+                    enabled=True,
+                    sort_order=1,
+                ),
+            ]
+        )
+
+
+def test_fixed_question_stream_speed_defaults_and_bounds() -> None:
+    default_row = FixedQuestion(
+        id="default-speed",
+        question="默认速度",
+        answer="回答",
+    )
+    assert default_row.tokens_per_second == 100
+
+    for invalid_speed in (0, 1001):
+        with pytest.raises(ValidationError):
+            FixedQuestion(
+                id=f"invalid-{invalid_speed}",
+                question="非法速度",
+                answer="回答",
+                tokens_per_second=invalid_speed,
+            )

offline-backend-20260512/backend/tests/test_health_auth_bypass.py

@@ -2,12 +2,15 @@
 
 from __future__ import annotations
 
+from types import SimpleNamespace
+
 import pytest
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
 from app.gateway import auth_middleware
 from app.gateway.app import app as gateway_app
+from app.gateway.auth.token_source import get_explicit_or_session_token
 from app.gateway.auth_middleware import PUBLIC_HEALTH_PATHS, AuthMiddleware
 
 
@@ -98,3 +101,28 @@ def test_non_health_route_still_requires_auth(monkeypatch: pytest.MonkeyPatch) -
         response = client.get("/private")
 
     assert response.status_code == 401
+
+
+def _auth_request(*, authorization: str = "", cookie: str = "") -> SimpleNamespace:
+    headers = {"authorization": authorization} if authorization else {}
+    cookies = {"access_token": cookie} if cookie else {}
+    return SimpleNamespace(headers=headers, cookies=cookies)
+
+
+def test_explicit_bearer_token_wins_over_shared_cookie() -> None:
+    """Another tab's cookie must not override this tab's explicit token."""
+    request = _auth_request(authorization="Bearer current-tab", cookie="other-tab")
+
+    assert get_explicit_or_session_token(request) == "current-tab"
+
+
+def test_cookie_remains_the_fallback_for_cookie_only_clients() -> None:
+    request = _auth_request(cookie="browser-session")
+
+    assert get_explicit_or_session_token(request) == "browser-session"
+
+
+def test_malformed_authorization_header_does_not_hide_valid_cookie() -> None:
+    request = _auth_request(authorization="Basic abc", cookie="browser-session")
+
+    assert get_explicit_or_session_token(request) == "browser-session"

offline-backend-20260512/backend/tests/test_username_login_domain.py

@@ -9,15 +9,18 @@ Verifies that:
 
 from __future__ import annotations
 
-import os
 from datetime import UTC, datetime
-from uuid import UUID, uuid4
+from uuid import uuid4
 
 import pytest
 
+from app.gateway.auth.local_provider import LocalAuthProvider
 from app.gateway.auth.models import User
 from app.gateway.auth.repositories.base import UserRepository
+from app.gateway.routers import auth as auth_module
 from app.gateway.routers.auth import (
+    _get_or_create_user_by_username,
+    _lookup_user_by_username,
     _normalize_username_local_part,
     _username_to_email,
 )
@@ -74,6 +77,9 @@ class FakeUserRepository(UserRepository):
     async def count_admin_users(self) -> int:
         return sum(1 for u in self.rows.values() if u.system_role == "admin")
 
+    async def list_users(self) -> list[User]:
+        return sorted(self.rows.values(), key=lambda user: user.created_at)
+
     async def get_user_by_oauth(self, provider: str, oauth_id: str) -> User | None:
         return None
 
@@ -146,3 +152,38 @@ async def test_local_part_lookup_misses_unknown_username() -> None:
     repo = FakeUserRepository()
     _seed(repo, "alice@cm.com")
     assert await repo.find_user_by_email_local_part("bob") is None
+
+
+@pytest.mark.asyncio
+async def test_username_login_uses_oldest_account_when_canonical_duplicate_exists(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A domain migration must not make one username alternate user ids."""
+    repo = FakeUserRepository()
+    older = _seed(repo, "alice@local.deerflow")
+    older.created_at = datetime(2024, 1, 1, tzinfo=UTC)
+    newer = _seed(repo, "alice@cm.com")
+    newer.created_at = datetime(2026, 1, 1, tzinfo=UTC)
+    monkeypatch.setattr(auth_module, "get_local_provider", lambda: LocalAuthProvider(repo))
+
+    user, created = await _get_or_create_user_by_username("alice")
+
+    assert created is False
+    assert user.id == older.id
+
+
+@pytest.mark.asyncio
+async def test_password_lookup_uses_same_stable_username_account(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    repo = FakeUserRepository()
+    older = _seed(repo, "alice@local.deerflow")
+    older.created_at = datetime(2024, 1, 1, tzinfo=UTC)
+    newer = _seed(repo, "alice@cm.com")
+    newer.created_at = datetime(2026, 1, 1, tzinfo=UTC)
+    monkeypatch.setattr(auth_module, "get_local_provider", lambda: LocalAuthProvider(repo))
+
+    user = await _lookup_user_by_username("alice")
+
+    assert user is not None
+    assert user.id == older.id