merge origin/staging into frontend dead code (issue #63)

Docs-only catch-up; agent-loop rule A. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LutgzyJqNhPpZy1nFsJtN3
2026-10-14 11:20:59 +08:00
parent 392c1f7218 4130d1dba9
commit 400af57e86
24 changed files with 37342 additions and 18192 deletions
@@ -1,5 +1,12 @@
 # BLOCKED

+## TASK-consult-prep-parallel-20261013：审核真实延迟、Postgres 中位数、多领域镜像（2026-10-14）
+
+- **阿里云审核**：本机没有审核密钥。1500 毫秒挡首帧的结论来自测试里的假等待，不是真实供应商。开关 `CONSULT_PREP_PARALLEL` 保持关闭。
+- **本机 Postgres**：没有量 `getUser`、注销检查、读会话、按版本读模型各 30 次的中位数。因此不能判断首帧是否真的快了两成，也不能判断模型准备是否快了一半。两个相关开关保持关闭（并行关、版本缓存 0 毫秒）。
+- **多领域并发**：没有拿带星盘网络库的镜像重跑。任务书里的起草试算大约只快一成，不到两成五的开启线。`CONSULT_DOMAIN_CONCURRENCY` 保持 1。
+- **staging 前后对比**：要等这支分支部署之后才能看真实首行时间。本轮不改默认值。
+
 ## TASK-rectification-explicit-entry-20261012：T0 日志、T5 模型（BUG-1306，2026-10-12）

 - **T0**：本 worktree 没有 staging 主机，读不到 web 容器 `docker logs`，也没有数据库。无法核对 10-11 那一轮是否调用了 `offer-birth-time-rectification`。不阻塞实现。产品可照做的查法写在 `docs/tasks/PROGRESS-rectification-explicit-entry-20261012.md`，只看工具名、status，以及该会话 messages 里有没有 `rectificationOffer`。
@@ -0,0 +1,70 @@
+# PROGRESS · 发问准备并行（2026-10-14）
+
+- 分支：`codex/consult-prep-parallel-20261013`
+- 基线：`origin/staging` `66cd60081f37a57680274633fcb3fb942d01bb2f`（任务书写的 `77057a819` 的后代；计划评论里的 `2994588ad` 更早，不因此停工）
+- 个人长报告写作开关仍是关闭，本轮没有改报告章节
+- 编号：本分支上的 `docs/BUG_HISTORY.md` 最大号是 BUG-1322。BUG-1323～1336 被其它未合并任务书占用。本单用 BUG-1337～1339
+
+## 开关
+
+三个都保持代码里的默认，没有单独提交去打开：
+
+| 开关 | 默认 | 原因 |
+| --- | --- | --- |
+| `CONSULT_PREP_PARALLEL` | 关 | 没有本机 Postgres 中位数，不满足首帧快两成 |
+| `MODEL_VERSION_CACHE_TTL_MS` | 0 | 没有量到「准备模型」快一半 |
+| `CONSULT_DOMAIN_CONCURRENCY` | 1 | 没有带星盘网络库的镜像重测；起草试算预计达不到快两成五 |
+
+因此没有写 `CHANGELOG.md`，也没有改 `frontend/DESIGN.md`。
+
+## 做了什么
+
+- T1：开流前的登录、注销检查、请求体、会话、模型收进 `loadConsultTurnHead`。并行时注销检查和后面三条同时开始，失败仍按旧的先后返回。423 仍在任何扣点之前。
+- T2：并行打开时，审核在开流前启动、在扣点前等待。拦住时不扣点，并且比资料错误优先。关掉时仍是开流前等待审核。
+- T3：按模型和版本缓存成功的解析，最长 15 秒；失败和空结果不缓存；管理端清目录缓存时一并清掉。地址检查函数本身没改。已发布版本的供应商代码、类型和地址不可变，但启用开关和密钥仍可变，所以不能超过目录缓存的 15 秒。
+- T4：问题保存成功后立刻读标题，和分类同时进行，用到标题时再等。分类仍在扣点和保存之后。
+- T5：领域并发开关。为 1 时仍是原来的逐个循环，含时间预算。为 2 时一起算，结果按计划顺序组装，不再逐个套时间预算，领域个数上限不变。
+- T6：准备阶段记一条 `consultation-prep-v1`，只有阶段名、毫秒和状态，没有问题原文。
+- T7：见 `docs/testing/consult-prep-parallel-ab-20261013.md`。假时钟并行 160、串行 210。这不是开启依据。
+- 测试注入：模型目录增加 `setModelCatalogIoForTests`，只给测试替换查库、解密和地址检查。生产路径仍调用原来的函数。
+
+## 改过的既有断言
+
+| 文件 | 原值 | 新值 | 原因 |
+| --- | --- | --- | --- |
+| `content-moderation.test.ts` | 源码含 `await moderate(visibleQuestion, "input"` 且早于 `reserve_consultation_usage` | `await settleInputModeration()` 早于扣点；准备服务里 `beforeReserve` 早于 `input.reserve()`；拦住时扣点次数是 0 | 审核挪到流内、扣点前。关掉开关时开流前的等待还在 |
+| `model-catalog.test.ts` | `return await readVersion(modelId, configVersion)` | `const model = await readVersion(...)`，只有非空才写入短缓存 | 绑版本的解析可以缓存，失败和空结果不写 |
+| `model-catalog.test.ts` | 咨询路由里 `if (!sessionModel)` 后面是 503 | 准备函数里 `if (!sessionModel)` 返回 `modelUnavailable()`，状态 503 | 模型不可用仍是开流前的 503，文案随准备步骤搬走 |
+| `consult-first-frame-and-pacing-20260928.test.tsx` | 401、请求体、会话不存在、会话类型、模型已变这些字面量在路由的 `runTurn` 之前 | 字面量在 `turn-prep.ts`；路由在 `runTurn` 之前调用 `loadConsultTurnHead` 并返回失败 | 开流前的判断收进同一个函数，状态码仍在开流前 |
+| `rectification-session-composer-guard.test.ts` | 咨询路由里会话缺失返回 404，类型不对返回 `session_not_consultation` | 同样两句在 `turn-prep.ts` | 会话判断搬走，文案和状态码没变 |
+
+`consultation-agentic-runtime.test.ts` 的领域循环帮助函数现在也接受异步的假计算，以便断言「后完成的领域仍排在计划后面」。原有断言没有改松。
+
+## 并发为 2 时的时间预算
+
+领域数仍先被原来的上限截断（现在是 2）。开关为 2 时，这两个领域一起开始，不再逐个问「剩下的时间还够不够下一个」。开关为 1 时，原来的逐个判断还在。
+
+## 自测
+
+test:db 未跑：本轮没有改查询语句，任务书也不要求。前端全量和 Python 快速门留给验收。`npm run lint` 0 error，原有 warning 没动。
+
+| 检查 | 命令 | 结果 |
+| --- | --- | --- |
+| 类型 | `node frontend/node_modules/typescript/bin/tsc --noEmit -p frontend` | 0 错 |
+| 规范 | `npm run lint --prefix frontend` | 0 error，125 条原有 warning |
+| 准备与审核 | `tsx --test` 下列文件：`consult-turn-prep-20261013`、`consult-prep-gate-20261013`、`content-moderation`、`model-catalog`、`consult-first-frame-and-pacing-20260928`、`rectification-session-composer-guard` | 准备函数 3 通过；门禁文件 5 通过；其余在同一次 41 例里通过（当时缓存测试还在改，后来单文件 5 通过、失败 0、取消 0） |
+| 计费、注销、标题、寒暄、领域 | 同上命令：`application-billing-contract`、`account-deletion`、`session-title-persist-contract`、`consultation-smalltalk`、`consultation-agentic-runtime` | 136 通过，失败 0，取消 0 |
+| 假时钟 | `consult-prep-parallel-ab.mts` | 并行 160，串行 210，同时步数 2 对 1 |
+| 领域镜像 | `consult-domain-concurrency-ab.py` | 退出 0，写明没有跑镜像 |
+
+## 第 1 次打回
+
+合并最新 `origin/staging`。业务代码没改，三个开关默认值没改。
+
+| 文件 | 原值 | 新值 | 原因 |
+| --- | --- | --- | --- |
+| `chat-session-authority.test.ts` | 在 `route.ts` 里找 `if (!chatSession)` 后的 404 | 在 `turn-prep.ts` 里锁 `sessionMissing()` 的 404，和分开的 `notConsultation()` 409 | 判定搬走了。仍不许并成一个判断 |
+| `consultation-workflow-contract.test.ts` | `beforeReserve: ({ consultationMode }) => createConsultationPlan({`，以及 `reserve: () => reserveConsultationModel(` | `beforeReserve` 的 async 函数体里调用 `createConsultationPlan({`，并且早于 `reserve`；`reserve` 的 async 函数体里调用 `reserveConsultationModel(` | 改成了 async。没有放宽成「文件里出现过函数名」 |
+| `consult-prep-parallel-ab.mts` | `frontend/scripts/research/`，从 `frontend/tests` 导入 | `frontend/tests/consult-prep-parallel-ab.mts` | 镜像会复制 scripts、不复制 tests。另一个新增的领域脚本是 Python，不导入测试目录 |
+
+这次自测：上述合同测试加出生时间模式、镜像导入检查，40 通过，失败 0，取消 0。假时钟脚本挪走后仍是并行 160、串行 210。`tsc --noEmit` 0 错。`npm run lint` 0 error，125 条原有 warning。全量 `npm test` 留给验收。`git diff a0e223e2` 到本修复会带上合并进来的测试环境前端文件；相对那次合并的 `origin/staging`，`frontend/src` 没有新增改动。
@@ -475,8 +475,8 @@
 | `TASK-api-disk-cache-retention-20261013.md` | `PROGRESS-api-disk-cache-retention-20261013.md` | API 磁盘缓存只进不出、VedAstro 审计文件留存出生资料、失败快照久缓存、缓存键规范化（BUG-1330～1332，A/B） | 待验收 | `codex/api-disk-cache-retention-20261013` |
 | `TASK-report-reference-perf-20261013.md` | `PROGRESS-report-reference-perf-20261013.md` | 长报告出稿提速：术语替换命中预检、哈希单次遍历、同盘复用（BUG-1335～1336，A/B） | 待验收 | `codex/report-reference-perf-20261013` |
 | `TASK-report-reference-perf-fix-20261014.md` | `PROGRESS-report-reference-perf-20261013.md` | **#61 第 1 次打回**：`69c739fb` 本单全过；与清理规则合并后，长报告缓存目录补进自行清理名单 | 待验收 | `codex/report-reference-perf-20261013` |
-| `TASK-consult-prep-parallel-20261013.md` | — | 普通对话发问前准备步骤并行、模型版本短缓存、多领域并发开关（BUG-1337～1339，A/B） | 待领取 | `codex/consult-prep-parallel-20261013` |
-| `TASK-consult-prep-parallel-fix-20261014.md` | — | **#62 第 1 次打回**：`a0e223e2` 业务逻辑通过；两条源码文本断言未跟改（§7.8）、A/B 脚本从 `frontend/tests` 导入（BUG-1140） | 待领取 | `codex/consult-prep-parallel-20261013` |
+| `TASK-consult-prep-parallel-20261013.md` | `PROGRESS-consult-prep-parallel-20261013.md` | 普通对话发问前准备步骤并行、模型版本短缓存、多领域并发开关（BUG-1337～1339，A/B） | 待验收 | `codex/consult-prep-parallel-20261013` |
+| `TASK-consult-prep-parallel-fix-20261014.md` | `PROGRESS-consult-prep-parallel-20261013.md` | **#62 第 1 次打回**：两条源码文本检查改认新位置，对照脚本挪进测试目录 | 待验收 | `codex/consult-prep-parallel-20261013` |
 | `TASK-frontend-dead-code-20261013.md` | `PROGRESS-frontend-dead-code-20261013.md` | 前端死代码 17 文件与无调用方接口删除，含未登录重计算接口（BUG-1342） | 待验收 | `codex/frontend-dead-code-20261013` |
 | `TASK-frontend-dead-code-fix-20261014.md` | `PROGRESS-frontend-dead-code-20261013.md` | **#63 第 1 次打回**：类名白名单去掉已删组件留下的条目 | 待验收 | `codex/frontend-dead-code-20261013` |
 | `TASK-api-chart-subprocess-gate-20261013.md` | — | 合盘不带 skip 串行排盘、报告无效子进程、名额轻重混用（只模拟）（BUG-1333～1334，A/B） | 待领取（排队：api-disk-cache-retention + report-reference-perf） | `codex/api-chart-subprocess-gate-20261013` |
@@ -0,0 +1,36 @@
+# 发问准备并行：本轮量到的和没量到的（2026-10-14）
+
+开关都还关着。用户发问后的第一行，这轮不会变快。
+
+## 进程内假时钟
+
+命令：`node node_modules/tsx/dist/cli.mjs tests/consult-prep-parallel-ab.mts`（在 `frontend` 目录）。
+
+假延迟：注销检查 50 毫秒，读会话 80 毫秒，解析模型 80 毫秒。这不是数据库实测。
+
+| | 逻辑耗时 | 同时进行的步数 |
+| --- | --- | --- |
+| 串行 | 210 | 1 |
+| 并行 | 160 | 2 |
+
+并行比这三条假延迟的相加更短。它不能代替「首帧中位数下降两成」的开启条件，因为本机没有量 Postgres。
+
+审核密钥不在这台机器上。测试里把审核挂起 1500 毫秒，第一小段在 400 毫秒内已经写出，然后才收到拦截。这是假等待，不是供应商的真实时间。
+
+## 版本缓存
+
+行为测试：15 秒内第二次解析不再查库、不再做地址检查；清缓存后会再查；缓存关掉后每次都查；查库失败和空结果不会被记住。没有量到真实「准备模型」那一段快了多少，所以缓存时间保持 0。
+
+## 两个领域一起算
+
+命令：`.venv\Scripts\python.exe frontend\scripts\research\consult-domain-concurrency-ab.py`
+
+本轮没有启动计算服务，也没有带星盘网络库的镜像。任务书起草时的试算大约只快一成，热缓存时反而更慢，达不到快两成五。默认仍是一个一个算。
+
+## 仍然关闭的原因
+
+| 开关 | 现在 | 要打开还缺什么 |
+| --- | --- | --- |
+| 发问准备并行 | 关 | 本机 Postgres 上鉴权、注销检查、读会话、读模型各 30 次的中位数，并且首帧中位数至少快两成 |
+| 模型版本缓存 | 0 毫秒 | 同上，准备模型那一段至少快一半 |
+| 两个领域一起算 | 1 | 带星盘网络库的镜像上墙钟至少快两成五，争用时 429 不增加，响应除约定字段外一致 |
@@ -0,0 +1,24 @@
+"""Two-domain concurrency bench.
+
+The vedastro image and the local API were not started in this round.
+The task book's draft on one public chart was about 10 percent faster when
+cold and slower when the chart was already cached. The switch opens only
+when the vedastro image is at least 25 percent faster, contention 429s do
+not rise, and the byte diff matches. Default stays 1.
+"""
+
+from __future__ import annotations
+
+
+def main() -> int:
+    print("CONSULT_DOMAIN_CONCURRENCY default stays 1")
+    print("vedastro image bench: not run")
+    print("local API bench: not run")
+    print("contention 429 count: not run")
+    print("byte diff: not run")
+    print("reason: this round has no vedastro measurement; the published draft missed the 25 percent gate")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -10,7 +10,15 @@ import {
  runConsultationWorkflow,
 } from "@/mastra";
 import { blocksPromptExtraction } from "@/lib/consult-safety";
+import { consultDomainConcurrency, consultPrepParallelEnabled, modelVersionCacheTtlMs } from "@/lib/consult/prep-flags";
+import {
+  ConsultationInputBlockedError,
+  consultPrepObservabilityEvent,
+  type ConsultPrepPhase,
+} from "@/lib/consult/prep-log";
+import { loadConsultTurnHead } from "@/lib/consult/turn-prep";
 import { inputBlockedCopy, moderate, MODERATION_OUTPUT_REPLACED_COPY } from "@/lib/moderation";
+import type { ModerationResult } from "@/lib/moderation";
 import { recordModerationEvent } from "@/lib/moderation/log";
 import { consultationDomainSchema } from "@/lib/consultation-domain-registry";
 import { parseAgentReply } from "@/lib/agent-reply";
@@ -304,78 +312,43 @@ export async function POST(request: Request) {
    return jsonForSupabaseSetupFailure(error, "POST /api/consult");
  }

-  const {
-    data: { user },
-    error: authError,
-  } = await supabase.auth.getUser();
-  if (authError || !user) {
-    return NextResponse.json(
-      { error: "请先登录", message: "登录后才能开始咨询。" },
-      { status: 401 },
-    );
+  const prepParallel = consultPrepParallelEnabled();
+  const prepPhases: ConsultPrepPhase[] = [];
+  const head = await loadConsultTurnHead({
+    parallel: prepParallel,
+    getUser: () => supabase.auth.getUser(),
+    readPendingDeletion: (userId) => refuseWhenAccountDeletionPending(userId),
+    readBody: async () => chatRequestSchema.safeParse(await request.json().catch(() => null)),
+    readSession: async (userId, parsed) => await supabase
+      .from("chat_sessions")
+      .select("id,model_id,model_config_version,session_type,messages,title,theme,chart_profile_id,chart_profile_name,chart_profile_role,context_summary")
+      .eq("id", parsed.sessionId)
+      .eq("user_id", userId)
+      .maybeSingle(),
+    resolveModel: async (chatSession) => {
+      if (!chatSession.model_id) return null;
+      return await resolveSessionLanguageModel(
+        chatSession.model_id,
+        chatSession.model_config_version,
+      );
+    },
+  });
+  if (!head.ok) {
+    // A pending account deletion freezes the account: nothing new is charged.
+    const frozen = head.frozen ? head.response : null;
+    if (frozen) return frozen;
+    return head.response;
  }
-  // A pending account deletion freezes the account: nothing new is charged.
-  const frozen = await refuseWhenAccountDeletionPending(user.id);
-  if (frozen) return frozen;
-
-  const parsed = chatRequestSchema.safeParse(
-    await request.json().catch(() => null),
+  const { user, parsed, chatSession, sessionModel } = head;
+  prepPhases.push(
+    { phase: "prep.auth", durationMs: head.timings.authMs, status: "completed" },
+    { phase: "prep.deletion", durationMs: head.timings.deletionMs, status: "completed" },
+    { phase: "prep.session", durationMs: head.timings.sessionMs, status: "completed" },
+    { phase: "prep.model", durationMs: head.timings.modelMs, status: "completed" },
+    { phase: prepParallel ? "prep.flag.parallel-on" : "prep.flag.parallel-off", durationMs: 0, status: "completed" },
+    { phase: modelVersionCacheTtlMs() > 0 ? "prep.flag.cache-on" : "prep.flag.cache-off", durationMs: 0, status: "completed" },
+    { phase: consultDomainConcurrency() === 2 ? "prep.flag.domains-2" : "prep.flag.domains-1", durationMs: 0, status: "completed" },
  );
-  if (!parsed.success) {
-    return NextResponse.json(
-      { error: "出生资料或问题格式不正确", details: parsed.error.flatten() },
-      { status: 400 },
-    );
-  }
-
-  const { data: chatSession, error: chatSessionError } = await supabase
-    .from("chat_sessions")
-    .select("id,model_id,model_config_version,session_type,messages,title,theme,chart_profile_id,chart_profile_name,chart_profile_role,context_summary")
-    .eq("id", parsed.data.sessionId)
-    .eq("user_id", user.id)
-    .maybeSingle();
-  if (chatSessionError) {
-    return NextResponse.json(
-      { error: "暂时无法读取咨询会话", message: "请稍后重试。" },
-      { status: 503 },
-    );
-  }
-  if (!chatSession) {
-    return NextResponse.json(
-      { error: "咨询会话不存在", message: "请重新进入咨询。" },
-      { status: 404 },
-    );
-  }
-  if (chatSession.session_type !== "consultation") {
-    return NextResponse.json(
-      {
-        error: "这是生时校正会话",
-        code: "session_not_consultation",
-        message: "请在生时校正里继续。",
-      },
-      { status: 409 },
-    );
-  }
-  if (!chatSession.model_id || chatSession.model_id !== parsed.data.modelId) {
-    return NextResponse.json(
-      { error: "会话模型已经变化", message: "请刷新会话后重新发送，本次不会扣点。" },
-      { status: 409 },
-    );
-  }
-  const sessionModel = await resolveSessionLanguageModel(
-    chatSession.model_id,
-    chatSession.model_config_version,
-  );
-
-  if (!sessionModel) {
-    return NextResponse.json(
-      {
-        error: "模型暂不可用",
-        message: "当前会话绑定的数据库模型配置不可用，请联系管理员，本次不会扣点。",
-      },
-      { status: 503 },
-    );
-  }
  const sessionContextWindow = sessionModel.contextWindow;

  if (parsed.data.entrypoint === "birth_time_rectification") {
@@ -421,14 +394,58 @@ export async function POST(request: Request) {
  // Content moderation, input side (2026-09-30): before any reservation, so a
  // blocked question is never sent to a model and never charged.
  const moderationContext = { route: "consult", userId: user.id, requestId: parsed.data.requestId, modelId: chatSession.model_id };
-  const inputModeration = await moderate(visibleQuestion, "input", moderationContext);
-  if (inputModeration.verdict !== "pass") await recordModerationEvent(accounting, inputModeration, "input", moderationContext);
-  if (inputModeration.verdict === "block") {
+  let inputModerationPending: Promise<ModerationResult> | null = null;
+  let inputModerationRecorded = false;
+  let moderationTimed = false;
+  let prepLogged = false;
+  const moderationStartedAt = Date.now();
+  function logPrep() {
+    if (prepLogged) return;
+    prepLogged = true;
+    logAgentObservability(consultPrepObservabilityEvent({
+      requestId,
+      sessionId,
+      phases: prepPhases,
+    }));
+  }
+  async function settleInputModeration(): Promise<ModerationResult | null> {
+    if (!inputModerationPending) return null;
+    const result = await inputModerationPending;
+    if (!moderationTimed) {
+      moderationTimed = true;
+      prepPhases.push({
+        phase: "prep.moderation",
+        durationMs: Math.max(0, Date.now() - moderationStartedAt),
+        status: result.verdict === "block" ? "failed" : "completed",
+      });
+    }
+    if (result.verdict !== "pass" && !inputModerationRecorded) {
+      inputModerationRecorded = true;
+      await recordModerationEvent(accounting, result, "input", moderationContext);
+    }
+    return result;
+  }
+  function contentBlockedResponse(result: ModerationResult) {
    return NextResponse.json(
-      { error: "无法处理该请求", code: "content_blocked", message: inputBlockedCopy(inputModeration) },
+      { error: "无法处理该请求", code: "content_blocked", message: inputBlockedCopy(result) },
      { status: 400 },
    );
  }
+  if (!prepParallel) {
+    const inputModeration = await moderate(visibleQuestion, "input", moderationContext);
+    prepPhases.push({
+      phase: "prep.moderation",
+      durationMs: Math.max(0, Date.now() - moderationStartedAt),
+      status: inputModeration.verdict === "block" ? "failed" : "completed",
+    });
+    if (inputModeration.verdict !== "pass") await recordModerationEvent(accounting, inputModeration, "input", moderationContext);
+    if (inputModeration.verdict === "block") {
+      logPrep();
+      return contentBlockedResponse(inputModeration);
+    }
+  } else {
+    inputModerationPending = moderate(visibleQuestion, "input", moderationContext);
+  }
  type SelectedModel = NonNullable<Awaited<ReturnType<typeof resolveSessionLanguageModel>>>;
  type ReservationResult = { success: boolean; credits: number | null; error_code: string | null };
  type ModelSelection = Awaited<ReturnType<typeof reserveConsultationModel<SelectedModel, ReservationResult>>>;
@@ -443,6 +460,7 @@ export async function POST(request: Request) {
  // pre-stream order because its text/plain replies cannot ride an NDJSON stream.
  const runTurn = async (): Promise<Response> => {
  let prepared: PreparedConsultationRoute<ModelSelection, ConsultationPlan>;
+  const prepareStarted = Date.now();
  try {
    prepared = await prepareConsultationRoute<ModelSelection, ConsultationPlan>({
      userId,
@@ -475,13 +493,25 @@ export async function POST(request: Request) {
        return data;
      },
      loadCandidateRange: (profileUserId) => loadReportCandidateRange(accounting, { userId: profileUserId }),
-      beforeReserve: ({ consultationMode }) => createConsultationPlan({
-        userIntent: parsed.data.question,
-        theme: consultationTheme,
-        consultationMode,
-        modelCreditCost: sessionModel.creditCost,
-      }),
-      reserve: () => reserveConsultationModel(
+      beforeReserve: async ({ consultationMode }) => {
+        prepPhases.push({
+          phase: "prep.subject",
+          durationMs: Math.max(0, Date.now() - prepareStarted),
+          status: "completed",
+        });
+        const inputModeration = await settleInputModeration();
+        if (inputModeration?.verdict === "block") throw new ConsultationInputBlockedError(inputModeration);
+        return createConsultationPlan({
+          userIntent: parsed.data.question,
+          theme: consultationTheme,
+          consultationMode,
+          modelCreditCost: sessionModel.creditCost,
+        });
+      },
+      reserve: async () => {
+        const reserveStarted = Date.now();
+        try {
+          return await reserveConsultationModel(
        chatSession.model_id,
        (modelId) => sessionModel?.id === modelId ? sessionModel : null,
        async (model) => {
@@ -503,9 +533,22 @@ export async function POST(request: Request) {
            error_code: reservation.data.reason,
          };
        },
-      ),
+        );
+        } finally {
+          prepPhases.push({
+            phase: "prep.reserve",
+            durationMs: Math.max(0, Date.now() - reserveStarted),
+            status: "completed",
+          });
+        }
+      },
    });
  } catch (error) {
+    const blocked = await settleInputModeration();
+    if (blocked?.verdict === "block") {
+      logPrep();
+      return contentBlockedResponse(blocked);
+    }
    if (error instanceof FeaturePricingError) {
      console.error(
        `[billing] reservation failed request=${requestId} reason=${error.code}`,
@@ -626,6 +669,7 @@ export async function POST(request: Request) {
  }

  let appendedQuestion: { success: boolean; error_code?: string | null };
+  const appendStarted = Date.now();
  try {
    appendedQuestion = await retryDetachedSettlement(async () => {
      const { data, error } = await accounting.rpc("append_consultation_question", {
@@ -640,7 +684,18 @@ export async function POST(request: Request) {
      }
      return parsedAppend.data;
    });
+    prepPhases.push({
+      phase: "prep.append",
+      durationMs: Math.max(0, Date.now() - appendStarted),
+      status: "completed",
+    });
  } catch {
+    prepPhases.push({
+      phase: "prep.append",
+      durationMs: Math.max(0, Date.now() - appendStarted),
+      status: "failed",
+    });
+    logPrep();
    await cancel();
    return NextResponse.json(
      { error: "暂时无法保存问题", message: "请稍后重试。" },
@@ -670,6 +725,14 @@ export async function POST(request: Request) {
  let classificationOutcome: string | undefined;
  let classificationDurationMs: number | null = null;
  let latencyState: ReturnType<typeof createConsultationRuntimeState> | null = null;
+  const titleStarted = Date.now();
+  const titleRead = supabase
+    .from("chat_sessions")
+    .select("title")
+    .eq("id", sessionId)
+    .eq("user_id", userId)
+    .maybeSingle();
+  const classifyStarted = Date.now();
  const turn = parsed.data.entrypoint === undefined
    ? await classifyConsultationTurn({
      model: selectedModel,
@@ -705,8 +768,14 @@ export async function POST(request: Request) {
      },
    })
    : { kind: "consult" as const };
+  prepPhases.push({
+    phase: "prep.classify",
+    durationMs: classificationDurationMs ?? Math.max(0, Date.now() - classifyStarted),
+    status: "completed",
+  });

  if (turn.kind === "smalltalk") {
+    logPrep();
    return streamSmalltalkResponse({
      requestId,
      reply: turn.reply,
@@ -738,12 +807,13 @@ export async function POST(request: Request) {
  }

  const expectedTitle = typeof chatSession.title === "string" ? chatSession.title : "";
-  const { data: titleRow } = await supabase
-    .from("chat_sessions")
-    .select("title")
-    .eq("id", sessionId)
-    .eq("user_id", userId)
-    .maybeSingle();
+  const { data: titleRow } = await titleRead;
+  prepPhases.push({
+    phase: "prep.title",
+    durationMs: Math.max(0, Date.now() - titleStarted),
+    status: "completed",
+  });
+  logPrep();

  async function checkpointConsultationContext() {
    try {
@@ -0,0 +1,62 @@
+/** Call-time switches for consult preparation. Read on each request, not at import. */
+
+const warned = new Set<string>();
+
+function warnOnce(name: string, raw: string): void {
+  if (warned.has(name)) return;
+  warned.add(name);
+  console.warn(`[consult-prep] ${name} 的值无法识别，按默认处理`);
+  void raw;
+}
+
+function readRaw(env: Readonly<Record<string, string | undefined>>, name: string): string | undefined {
+  const raw = env[name];
+  if (raw === undefined) return undefined;
+  const trimmed = raw.trim();
+  return trimmed ? trimmed : undefined;
+}
+
+export function consultPrepParallelEnabled(
+  env: Readonly<Record<string, string | undefined>> = process.env,
+): boolean {
+  const raw = readRaw(env, "CONSULT_PREP_PARALLEL");
+  if (raw === undefined) return false;
+  const value = raw.toLowerCase();
+  if (value === "on") return true;
+  if (value === "off") return false;
+  warnOnce("CONSULT_PREP_PARALLEL", raw);
+  return false;
+}
+
+/** 0 closes the cache. Anything outside 0–15000 stays on the default, which is off. */
+export function modelVersionCacheTtlMs(
+  env: Readonly<Record<string, string | undefined>> = process.env,
+): number {
+  const raw = readRaw(env, "MODEL_VERSION_CACHE_TTL_MS");
+  if (raw === undefined) return 0;
+  if (!/^\d+$/.test(raw)) {
+    warnOnce("MODEL_VERSION_CACHE_TTL_MS", raw);
+    return 0;
+  }
+  const parsed = Number(raw);
+  if (parsed > 15_000) {
+    warnOnce("MODEL_VERSION_CACHE_TTL_MS", raw);
+    return 0;
+  }
+  return parsed;
+}
+
+/** 1 is today's serial loop. 2 starts the capped domains together. */
+export function consultDomainConcurrency(
+  env: Readonly<Record<string, string | undefined>> = process.env,
+): 1 | 2 {
+  const raw = readRaw(env, "CONSULT_DOMAIN_CONCURRENCY");
+  if (raw === undefined || raw === "1") return 1;
+  if (raw === "2") return 2;
+  warnOnce("CONSULT_DOMAIN_CONCURRENCY", raw);
+  return 1;
+}
+
+export function resetConsultPrepFlagWarningsForTests(): void {
+  warned.clear();
+}
@@ -0,0 +1,34 @@
+import type { ModerationResult } from "@/lib/moderation";
+
+export const CONSULT_PREP_AGENT_VERSION = "consultation-prep-v1";
+
+export class ConsultationInputBlockedError extends Error {
+  readonly result: ModerationResult;
+
+  constructor(result: ModerationResult) {
+    super("content_blocked");
+    this.name = "ConsultationInputBlockedError";
+    this.result = result;
+  }
+}
+
+export type ConsultPrepPhase = {
+  phase: string;
+  durationMs: number;
+  status: "completed" | "failed" | "skipped";
+};
+
+/** Closed observability event. The question text is not a field. */
+export function consultPrepObservabilityEvent(input: {
+  requestId: string;
+  sessionId: string;
+  phases: readonly ConsultPrepPhase[];
+}) {
+  return {
+    requestId: input.requestId,
+    sessionId: input.sessionId,
+    agentVersion: CONSULT_PREP_AGENT_VERSION,
+    contractPhases: input.phases.map((phase) => ({ ...phase })),
+    billingSettlementResult: "not_applicable" as const,
+  };
+}
@@ -0,0 +1,195 @@
+import { NextResponse } from "next/server";
+
+export type ConsultPrepTimings = {
+  authMs: number;
+  deletionMs: number;
+  sessionMs: number;
+  modelMs: number;
+};
+
+type AuthResult<User> = {
+  data: { user: User | null };
+  error: unknown;
+};
+
+type BodyResult<Parsed> =
+  | { success: true; data: Parsed }
+  | { success: false; error: { flatten: () => unknown } };
+
+type SessionResult<Session> = {
+  data: Session | null;
+  error: unknown;
+};
+
+type SessionRow = {
+  model_id: string | null;
+  session_type: string | null;
+  model_config_version?: number | null;
+  messages?: unknown;
+  title?: unknown;
+  theme?: unknown;
+  chart_profile_id?: unknown;
+  chart_profile_name?: unknown;
+  chart_profile_role?: unknown;
+  context_summary?: unknown;
+};
+
+type ParsedRequest = {
+  sessionId: string;
+  modelId: string;
+};
+
+export type ConsultTurnHeadDeps<User extends { id: string }, Parsed extends ParsedRequest, Session extends SessionRow, Model> = {
+  parallel: boolean;
+  getUser: () => Promise<AuthResult<User>>;
+  readPendingDeletion: (userId: string) => Promise<Response | null>;
+  readBody: () => Promise<BodyResult<Parsed>>;
+  readSession: (userId: string, parsed: Parsed) => Promise<SessionResult<Session>>;
+  resolveModel: (chatSession: Session) => Promise<Model | null>;
+};
+
+export type ConsultTurnHead<User, Parsed, Session, Model> =
+  | { ok: false; response: Response; frozen: boolean }
+  | {
+    ok: true;
+    user: User;
+    parsed: { success: true; data: Parsed };
+    chatSession: Session & { model_id: string };
+    sessionModel: Model;
+    timings: ConsultPrepTimings;
+  };
+
+function elapsedSince(started: number): number {
+  return Math.max(0, Date.now() - started);
+}
+
+function loginRequired(): Response {
+  return NextResponse.json(
+    { error: "请先登录", message: "登录后才能开始咨询。" },
+    { status: 401 },
+  );
+}
+
+function badBody(error: { flatten: () => unknown }): Response {
+  return NextResponse.json(
+    { error: "出生资料或问题格式不正确", details: error.flatten() },
+    { status: 400 },
+  );
+}
+
+function sessionUnreadable(): Response {
+  return NextResponse.json(
+    { error: "暂时无法读取咨询会话", message: "请稍后重试。" },
+    { status: 503 },
+  );
+}
+
+function sessionMissing(): Response {
+  return NextResponse.json(
+    { error: "咨询会话不存在", message: "请重新进入咨询。" },
+    { status: 404 },
+  );
+}
+
+function notConsultation(): Response {
+  return NextResponse.json(
+    {
+      error: "这是生时校正会话",
+      code: "session_not_consultation",
+      message: "请在生时校正里继续。",
+    },
+    { status: 409 },
+  );
+}
+
+function modelChanged(): Response {
+  return NextResponse.json(
+    { error: "会话模型已经变化", message: "请刷新会话后重新发送，本次不会扣点。" },
+    { status: 409 },
+  );
+}
+
+function modelUnavailable(): Response {
+  return NextResponse.json(
+    {
+      error: "模型暂不可用",
+      message: "当前会话绑定的数据库模型配置不可用，请联系管理员，本次不会扣点。",
+    },
+    { status: 503 },
+  );
+}
+
+type Chain<Parsed, Session, Model> =
+  | { ok: false; response: Response }
+  | {
+    ok: true;
+    parsed: { success: true; data: Parsed };
+    chatSession: Session & { model_id: string };
+    sessionModel: Model;
+  };
+
+/**
+ * Auth, then either the old serial checks or deletion in parallel with
+ * body → session → model. Failures are still reported in the old order.
+ */
+export async function loadConsultTurnHead<
+  User extends { id: string },
+  Parsed extends ParsedRequest,
+  Session extends SessionRow,
+  Model,
+>(
+  deps: ConsultTurnHeadDeps<User, Parsed, Session, Model>,
+): Promise<ConsultTurnHead<User, Parsed, Session, Model>> {
+  const timings: ConsultPrepTimings = { authMs: 0, deletionMs: 0, sessionMs: 0, modelMs: 0 };
+  const authStarted = Date.now();
+  const auth = await deps.getUser();
+  timings.authMs = elapsedSince(authStarted);
+  const user = auth.data.user;
+  if (auth.error || !user) return { ok: false, response: loginRequired(), frozen: false };
+
+  const readChain = async (): Promise<Chain<Parsed, Session, Model>> => {
+    const body = await deps.readBody();
+    if (!body.success) return { ok: false, response: badBody(body.error) };
+    const sessionStarted = Date.now();
+    const session = await deps.readSession(user.id, body.data);
+    timings.sessionMs = elapsedSince(sessionStarted);
+    if (session.error) return { ok: false, response: sessionUnreadable() };
+    if (!session.data) return { ok: false, response: sessionMissing() };
+    const chatSession = session.data;
+    if (chatSession.session_type !== "consultation") return { ok: false, response: notConsultation() };
+    if (!chatSession.model_id || chatSession.model_id !== body.data.modelId) {
+      return { ok: false, response: modelChanged() };
+    }
+    const modelStarted = Date.now();
+    const sessionModel = await deps.resolveModel(chatSession);
+    timings.modelMs = elapsedSince(modelStarted);
+    if (!sessionModel) return { ok: false, response: modelUnavailable() };
+    return {
+      ok: true,
+      parsed: body,
+      chatSession: chatSession as Session & { model_id: string },
+      sessionModel,
+    };
+  };
+
+  if (!deps.parallel) {
+    const deletionStarted = Date.now();
+    const frozen = await deps.readPendingDeletion(user.id);
+    timings.deletionMs = elapsedSince(deletionStarted);
+    if (frozen) return { ok: false, response: frozen, frozen: true };
+    const chain = await readChain();
+    if (!chain.ok) return { ok: false, response: chain.response, frozen: false };
+    return { ...chain, user, timings };
+  }
+
+  const deletionStarted = Date.now();
+  const deletion = deps.readPendingDeletion(user.id).then((frozen) => {
+    timings.deletionMs = elapsedSince(deletionStarted);
+    return frozen;
+  });
+  const chain = readChain();
+  const [frozen, settled] = await Promise.all([deletion, chain]);
+  if (frozen) return { ok: false, response: frozen, frozen: true };
+  if (!settled.ok) return { ok: false, response: settled.response, frozen: false };
+  return { ...settled, user, timings };
+}
@@ -3,8 +3,22 @@ import "server-only";
 import type { MastraModelConfig } from "@mastra/core/llm";

 import { queryAdminRows } from "@/lib/admin/database";
+import { modelVersionCacheTtlMs } from "@/lib/consult/prep-flags";
 import { assertAllowedModelProviderUrl } from "@/lib/epay/gateway-policy";
 import { decryptModelProviderApiKey, type ModelProviderType } from "@/lib/model-provider-policy";
+
+/** Tests replace the database, decrypt and DNS steps. Production leaves these as the real functions. */
+const catalogIo = {
+  queryAdminRows,
+  assertAllowedModelProviderUrl,
+  decryptModelProviderApiKey,
+};
+
+export function setModelCatalogIoForTests(next: Partial<typeof catalogIo> | null): void {
+  catalogIo.queryAdminRows = next?.queryAdminRows ?? queryAdminRows;
+  catalogIo.assertAllowedModelProviderUrl = next?.assertAllowedModelProviderUrl ?? assertAllowedModelProviderUrl;
+  catalogIo.decryptModelProviderApiKey = next?.decryptModelProviderApiKey ?? decryptModelProviderApiKey;
+}
 import type { LanguageModelCatalog, ResolvedLanguageModel } from "@/mastra/model";

 const cacheTtlMs = 15_000;
@@ -17,7 +31,11 @@ type PublishedModelRow = {
  context_window: number | null;
 };
 type Cache = { expiresAt: number; catalog: LanguageModelCatalog };
-const state = globalThis as typeof globalThis & { jyotishaModelCatalogCache?: Cache };
+type VersionHit = { expiresAt: number; model: ResolvedLanguageModel };
+const state = globalThis as typeof globalThis & {
+  jyotishaModelCatalogCache?: Cache;
+  jyotishaModelVersionCache?: Map<string, VersionHit>;
+};

 function isSecretSettingKey(key: string) {
  const normalized = key.toLowerCase().replace(/[^a-z0-9]/g, "");
@@ -37,7 +55,7 @@ function publicModel(model: ResolvedLanguageModel) {
  return { id: model.id, label: model.label, description: model.description, creditCost: model.creditCost, isDefault: model.isDefault };
 }
 async function resolveRow(row: PublishedModelRow): Promise<ResolvedLanguageModel | null> {
-  const apiKey = decryptModelProviderApiKey({ encryptedApiKey: row.encrypted_api_key });
+  const apiKey = catalogIo.decryptModelProviderApiKey({ encryptedApiKey: row.encrypted_api_key });
  if (!apiKey) return null;
  let model: MastraModelConfig;
  if (row.provider_type === "openai") {
@@ -45,7 +63,7 @@ async function resolveRow(row: PublishedModelRow): Promise<ResolvedLanguageModel
  } else if (row.provider_type === "anthropic") {
    model = { providerId: "anthropic", modelId: row.provider_model.replace(/^anthropic\//, ""), apiKey };
  } else {
-    model = { providerId: row.provider_code, modelId: row.provider_model, url: (await assertAllowedModelProviderUrl(row.base_url ?? "")).url.toString(), apiKey };
+    model = { providerId: row.provider_code, modelId: row.provider_model, url: (await catalogIo.assertAllowedModelProviderUrl(row.base_url ?? "")).url.toString(), apiKey };
  }
  return {
    id: row.model_id, label: row.label, description: row.description, creditCost: row.credit_cost,
@@ -55,7 +73,7 @@ async function resolveRow(row: PublishedModelRow): Promise<ResolvedLanguageModel
  };
 }
 async function queryCatalog(where: string, values: readonly unknown[] = []) {
-  return queryAdminRows<PublishedModelRow>(`
+  return catalogIo.queryAdminRows<PublishedModelRow>(`
    select c.model_id,v.version,v.label,v.description,v.provider_model,v.credit_cost,v.is_default,
      p.code provider_code,p.provider_type,p.base_url,p.encrypted_api_key,
      v.input_cost_microusd_per_million input_cost,v.output_cost_microusd_per_million output_cost,
@@ -86,9 +104,32 @@ async function readVersion(modelId: string, version: number) {
  const rows = await queryCatalog("c.model_id=$1 and v.version=$2 and v.status in ('published','retired') and v.enabled and p.enabled", [modelId, version]);
  return rows[0] ? resolveRow(rows[0]) : null;
 }
+function versionCache(): Map<string, VersionHit> {
+  return state.jyotishaModelVersionCache ??= new Map();
+}
+
 export async function resolveSessionLanguageModel(modelId: string, configVersion: number | null | undefined) {
-  if (configVersion !== null && configVersion !== undefined) { try { return await readVersion(modelId, configVersion); } catch { return null; } }
+  if (configVersion !== null && configVersion !== undefined) {
+    const ttl = modelVersionCacheTtlMs();
+    const key = `${modelId}:${configVersion}`;
+    const cache = versionCache();
+    if (ttl > 0) {
+      const hit = cache.get(key);
+      if (hit && hit.expiresAt > Date.now()) return hit.model;
+    }
+    try {
+      const model = await readVersion(modelId, configVersion);
+      // Failures and null stay uncached so a disabled provider is not pinned.
+      if (model && ttl > 0) cache.set(key, { expiresAt: Date.now() + ttl, model });
+      return model;
+    } catch {
+      return null;
+    }
+  }
  const catalog = await loadLanguageModelCatalog();
  return catalog.models.find((model) => model.id === modelId) ?? null;
 }
-export function invalidateLanguageModelCatalog() { delete state.jyotishaModelCatalogCache; }
+export function invalidateLanguageModelCatalog() {
+  delete state.jyotishaModelCatalogCache;
+  state.jyotishaModelVersionCache?.clear();
+}
@@ -27,6 +27,7 @@ import { normalizeTechniqueAuditRows } from "../lib/consultation-technique-audit
 import type { AgentModelFinishReason, EvidenceCardTelemetry } from "../lib/agent-observability.ts";
 import { agentGenerationSettings, AGENT_SLICE_ANSWER_OUTPUT_TOKENS, AGENT_SLICE_THINKING_OUTPUT_TOKENS } from "../lib/agent-generation-settings.ts";
 import { chartCalculationProgressLabel, evidenceLookupActivityLabel } from "../lib/consultation-activity-labels.ts";
+import { consultDomainConcurrency } from "../lib/consult/prep-flags.ts";
 import {
  buildEvidenceCard,
  citedEvidenceCardFields,
@@ -1227,57 +1228,114 @@ export function createConsultationTools(ctx: ConsultationAgentContext) {
        try {
          const userIntent = ctx.plan?.userIntent ?? input.question;
          const executions: DomainExecution[] = [];
-          for (let index = 0; index < domains.length; index += 1) {
-            const domain = domains[index];
-            if (!domain) continue;
-            // Every domain shares the run's single abort deadline, so a plan
-            // that runs long would abort mid-loop and lose the domains already
-            // calculated. Stop while there is still time to answer instead.
-            if (!domainFitsRunBudget(now() - startedAt, executions.length)) break;
-            await context.writer?.custom({
-              type: "data-jyotish-activity",
-              data: {
-                phase: "chart-calculation",
-                label: chartCalculationProgressLabel(index + 1, domains.length),
-              },
-            });
-            const domainPlan = ctx.plan
-              && domains.length === 1
-              && ctx.plan.requestedDomains.length === 1
-              && ctx.plan.requestedDomains[0] === domain
-              ? ctx.plan
-              : createConsultationPlan({
-                userIntent,
-                theme: domain,
-                consultationMode: ctx.consultationMode,
-                ...(ctx.plan ? { modelCreditCost: ctx.plan.maxCreditCost, depth: ctx.plan.depth } : {}),
-                ...(domain === "timing" || domain === "annual"
-                  ? { timingHorizon: ctx.plan?.timingHorizon ?? undefined }
-                  : { timingHorizon: null }),
+          if (consultDomainConcurrency() === 1) {
+            for (let index = 0; index < domains.length; index += 1) {
+              const domain = domains[index];
+              if (!domain) continue;
+              // Every domain shares the run's single abort deadline, so a plan
+              // that runs long would abort mid-loop and lose the domains already
+              // calculated. Stop while there is still time to answer instead.
+              if (!domainFitsRunBudget(now() - startedAt, executions.length)) break;
+              await context.writer?.custom({
+                type: "data-jyotish-activity",
+                data: {
+                  phase: "chart-calculation",
+                  label: chartCalculationProgressLabel(index + 1, domains.length),
+                },
              });
-            const toolInput = consultationInputSchema.parse({
-              ...ctx.serverChart.toolInput,
-              entryMode: "direct_chart",
-              question: userIntent,
-              theme: domain,
-            });
-            const workflow = await (ctx.runWorkflow ?? runConsultationWorkflow)(toolInput, {
-              foreground: true,
-              signal: context.abortSignal ?? ctx.abortSignal,
-              plan: domainPlan,
-              requestId: ctx.requestId,
-              entrypoint: ctx.entrypoint,
-            });
-            const guarded = applyBirthTimeModeToWorkflowContext(workflow, ctx.consultationMode, {
-              birthTimeSource: ctx.serverChart.truth.birthTimeSource,
-            });
-            const agentContext = toAgentConsultationContext(guarded);
-            executions.push({
-              domain,
-              context: agentContext,
-              modelOutput: toModelOutput(agentContext, domainPlan, domain),
-              receipt: consultationWorkflowReceipt(guarded),
-            });
+              const domainPlan = ctx.plan
+                && domains.length === 1
+                && ctx.plan.requestedDomains.length === 1
+                && ctx.plan.requestedDomains[0] === domain
+                ? ctx.plan
+                : createConsultationPlan({
+                  userIntent,
+                  theme: domain,
+                  consultationMode: ctx.consultationMode,
+                  ...(ctx.plan ? { modelCreditCost: ctx.plan.maxCreditCost, depth: ctx.plan.depth } : {}),
+                  ...(domain === "timing" || domain === "annual"
+                    ? { timingHorizon: ctx.plan?.timingHorizon ?? undefined }
+                    : { timingHorizon: null }),
+                });
+              const toolInput = consultationInputSchema.parse({
+                ...ctx.serverChart.toolInput,
+                entryMode: "direct_chart",
+                question: userIntent,
+                theme: domain,
+              });
+              const workflow = await (ctx.runWorkflow ?? runConsultationWorkflow)(toolInput, {
+                foreground: true,
+                signal: context.abortSignal ?? ctx.abortSignal,
+                plan: domainPlan,
+                requestId: ctx.requestId,
+                entrypoint: ctx.entrypoint,
+              });
+              const guarded = applyBirthTimeModeToWorkflowContext(workflow, ctx.consultationMode, {
+                birthTimeSource: ctx.serverChart.truth.birthTimeSource,
+              });
+              const agentContext = toAgentConsultationContext(guarded);
+              executions.push({
+                domain,
+                context: agentContext,
+                modelOutput: toModelOutput(agentContext, domainPlan, domain),
+                receipt: consultationWorkflowReceipt(guarded),
+              });
+            }
+          } else {
+            for (let index = 0; index < domains.length; index += 1) {
+              const domain = domains[index];
+              if (!domain) continue;
+              await context.writer?.custom({
+                type: "data-jyotish-activity",
+                data: {
+                  phase: "chart-calculation",
+                  label: chartCalculationProgressLabel(index + 1, domains.length),
+                },
+              });
+            }
+            const parallelExecutions = await Promise.all(domains.map(async (domain) => {
+              if (!domain) return null;
+              const domainPlan = ctx.plan
+                && domains.length === 1
+                && ctx.plan.requestedDomains.length === 1
+                && ctx.plan.requestedDomains[0] === domain
+                ? ctx.plan
+                : createConsultationPlan({
+                  userIntent,
+                  theme: domain,
+                  consultationMode: ctx.consultationMode,
+                  ...(ctx.plan ? { modelCreditCost: ctx.plan.maxCreditCost, depth: ctx.plan.depth } : {}),
+                  ...(domain === "timing" || domain === "annual"
+                    ? { timingHorizon: ctx.plan?.timingHorizon ?? undefined }
+                    : { timingHorizon: null }),
+                });
+              const toolInput = consultationInputSchema.parse({
+                ...ctx.serverChart.toolInput,
+                entryMode: "direct_chart",
+                question: userIntent,
+                theme: domain,
+              });
+              const workflow = await (ctx.runWorkflow ?? runConsultationWorkflow)(toolInput, {
+                foreground: true,
+                signal: context.abortSignal ?? ctx.abortSignal,
+                plan: domainPlan,
+                requestId: ctx.requestId,
+                entrypoint: ctx.entrypoint,
+              });
+              const guarded = applyBirthTimeModeToWorkflowContext(workflow, ctx.consultationMode, {
+                birthTimeSource: ctx.serverChart.truth.birthTimeSource,
+              });
+              const agentContext = toAgentConsultationContext(guarded);
+              return {
+                domain,
+                context: agentContext,
+                modelOutput: toModelOutput(agentContext, domainPlan, domain),
+                receipt: consultationWorkflowReceipt(guarded),
+              };
+            }));
+            for (const execution of parallelExecutions) {
+              if (execution) executions.push(execution);
+            }
          }
          // Domains the cap refused and domains the clock ran out on are the
          // same disclosure to the caller: requested but not calculated.
@@ -7,6 +7,7 @@ const listRoute = readFileSync(new URL("../src/app/api/sessions/route.ts", impor
 const listFilter = readFileSync(new URL("../src/lib/session-list-filter.ts", import.meta.url), "utf8");
 const itemRoute = readFileSync(new URL("../src/app/api/sessions/[id]/route.ts", import.meta.url), "utf8");
 const consultRoute = readFileSync(new URL("../src/app/api/consult/route.ts", import.meta.url), "utf8");
+const turnPrep = readFileSync(new URL("../src/lib/consult/turn-prep.ts", import.meta.url), "utf8");
 const sql = readFileSync(
  new URL("../supabase/migrations/20260901010000_append_consultation_question.sql", import.meta.url),
  "utf8",
@@ -59,14 +60,18 @@ test("session list GET omits messages while detail GET returns them", () => {
 });

 test("consult POST on a rectification session is 409, a missing id is 404", () => {
-  // 原值：!chatSession || session_type !== consultation 一律 404「咨询会话不存在」
-  // 新值：查不到 404；类型不对 409 session_not_consultation
-  // 原因：BUG-925 前端无法区分不存在与类型不对
-  assert.match(consultRoute, /if \(!chatSession\) \{[\s\S]*status: 404/);
-  assert.match(consultRoute, /code: "session_not_consultation"/);
-  assert.match(consultRoute, /status: 409/);
+  // 原值：consult route 里 /if (!chatSession) {[\s\S]*status: 404/，以及同文件的 session_not_consultation 与 409
+  // 新值：turn-prep.ts 里 sessionMissing() 为 404，notConsultation() 为 409 session_not_consultation，两处调用分开
+  // 原因：判定从 route.ts 挪进 turn-prep.ts；仍是查不到 404、类型不对 409，不得并成一个判断
+  assert.match(turnPrep, /function sessionMissing\(\)[\s\S]*status: 404/);
+  assert.match(turnPrep, /if \(!session\.data\) return \{ ok: false, response: sessionMissing\(\) \}/);
+  assert.match(turnPrep, /function notConsultation\(\)[\s\S]*code: "session_not_consultation"[\s\S]*status: 409/);
+  assert.match(
+    turnPrep,
+    /if \(chatSession\.session_type !== "consultation"\) return \{ ok: false, response: notConsultation\(\) \}/,
+  );
  assert.doesNotMatch(
-    consultRoute,
+    turnPrep,
    /if \(!chatSession \|\| chatSession\.session_type !== "consultation"\)/,
  );
 });
@@ -255,18 +255,23 @@ test("a turn that throws before producing any response ends the stream with a ge

 test("the consult route opens the stream before subject reads, reservation, question append and classification", () => {
  const route = source("../src/app/api/consult/route.ts");
+  const head = source("../src/lib/consult/turn-prep.ts");
  const openIndex = route.indexOf("const runTurn = async (): Promise<Response> =>");
  assert.ok(openIndex > 0);
-  // Stays in front of the stream: auth, body, session existence and type, the session model.
+  // 原值: 401、请求体、会话不存在、会话类型、模型已变这些字面量都在 route.ts 的 runTurn 之前。
+  // 新值: 这些字面量在 turn-prep.ts；route.ts 在 runTurn 之前调用 loadConsultTurnHead 并返回失败响应。提示词拦截仍在 route.ts。
+  // 原因: 开流前的独立读取收进同一个函数，两种开关的状态码仍在开流前返回。
+  assert.ok(route.indexOf("loadConsultTurnHead") < openIndex);
+  assert.ok(route.indexOf("if (!head.ok)") < openIndex);
+  assert.ok(route.indexOf("if (blocksPromptExtraction(userControlledPrompt))") < openIndex);
  for (const marker of [
    "status: 401",
-    '{ error: "出生资料或问题格式不正确", details: parsed.error.flatten() }',
+    '{ error: "出生资料或问题格式不正确", details: error.flatten() }',
    'error: "咨询会话不存在"',
    'code: "session_not_consultation"',
    'error: "会话模型已经变化"',
-    "if (blocksPromptExtraction(userControlledPrompt))",
  ]) {
-    assert.ok(route.indexOf(marker) < openIndex, `${marker} must answer with an HTTP status`);
+    assert.ok(head.includes(marker), `${marker} stays an HTTP response before the stream`);
  }
  // Inside the stream: the slow part, and every rejection it can raise.
  for (const marker of [
@@ -0,0 +1,164 @@
+import assert from "node:assert/strict";
+import { readFileSync } from "node:fs";
+import { register } from "node:module";
+import test from "node:test";
+
+import { agentObservabilityEventSchema } from "../src/lib/agent-observability.ts";
+import { ConsultationInputBlockedError, consultPrepObservabilityEvent } from "../src/lib/consult/prep-log.ts";
+import { inputBlockedCopy, MODERATION_SELF_HARM_COPY } from "../src/lib/moderation/index.ts";
+import type { ModerationResult } from "../src/lib/moderation/types.ts";
+import { streamFirstResponse } from "../src/lib/stream-first-response.ts";
+
+const blocked: ModerationResult = {
+  verdict: "block",
+  categories: ["self_harm"],
+  ruleIds: ["lexicon"],
+  providerId: "local",
+};
+
+test("a blocked question records once and never reaches reserve", async () => {
+  let reserves = 0;
+  let records = 0;
+  const pending = Promise.resolve(blocked);
+  async function beforeReserve() {
+    const result = await pending;
+    if (result.verdict !== "pass") records += 1;
+    if (result.verdict === "block") throw new ConsultationInputBlockedError(result);
+    reserves += 1;
+  }
+  await assert.rejects(beforeReserve, (error: unknown) => {
+    assert.ok(error instanceof ConsultationInputBlockedError);
+    assert.equal(inputBlockedCopy(error.result), MODERATION_SELF_HARM_COPY);
+    return true;
+  });
+  assert.equal(reserves, 0);
+  assert.equal(records, 1);
+  const route = readFileSync(new URL("../src/app/api/consult/route.ts", import.meta.url), "utf8");
+  const blockedAt = route.indexOf('blocked?.verdict === "block"');
+  const profileAt = route.indexOf("error instanceof ConsultationProfileTruthError");
+  assert.ok(blockedAt > 0 && blockedAt < profileAt);
+});
+
+test("moderation that takes 1.5s does not hold the first frame", async () => {
+  let release: (result: ModerationResult) => void = () => {};
+  const pending = new Promise<ModerationResult>((resolve) => {
+    release = resolve;
+  });
+  const started = Date.now();
+  const response = streamFirstResponse({
+    requestId: "prep-frame",
+    run: async () => {
+      const result = await pending;
+      return Response.json(
+        { error: "无法处理该请求", code: "content_blocked", message: inputBlockedCopy(result) },
+        { status: 400 },
+      );
+    },
+  });
+  const reader = response.body!.getReader();
+  const first = await reader.read();
+  assert.ok(Date.now() - started < 400, "first frame waited for moderation");
+  assert.ok(first.value && first.value.byteLength > 0);
+  setTimeout(() => release(blocked), 1_500);
+  const chunks: string[] = [];
+  const decoder = new TextDecoder();
+  while (true) {
+    const next = await reader.read();
+    if (next.value) chunks.push(decoder.decode(next.value));
+    if (next.done) break;
+  }
+  const events = chunks.join("").trim().split("\n").map((line) => JSON.parse(line));
+  const rejected = events.find((event) => event.type === "run.failed");
+  assert.equal(rejected.message, MODERATION_SELF_HARM_COPY);
+  assert.equal(rejected.reason, "content_blocked");
+  assert.equal(rejected.status, 400);
+});
+
+test("prep observability accepts the closed schema and drops the question", () => {
+  const question = "这句问题原文不该出现在日志里";
+  const event = consultPrepObservabilityEvent({
+    requestId: "req-prep",
+    sessionId: "sess-prep",
+    phases: [
+      { phase: "prep.model", durationMs: 3, status: "completed" },
+      { phase: "prep.moderation", durationMs: 12, status: "completed" },
+    ],
+  });
+  assert.deepEqual(agentObservabilityEventSchema.parse(event), event);
+  assert.equal(JSON.stringify(event).includes(question), false);
+  assert.equal(event.agentVersion, "consultation-prep-v1");
+});
+
+test("readVersion cache skips a second database and DNS lookup inside the ttl", async () => {
+  await register(new URL("./support/stub-server-only.mjs", import.meta.url).href, import.meta.url);
+  let queries = 0;
+  let dns = 0;
+  const row = {
+    model_id: "model-a",
+    version: 4,
+    label: "Model A",
+    description: "fixture",
+    provider_model: "demo",
+    credit_cost: 1,
+    is_default: false,
+    provider_code: "demo",
+    provider_type: "openai-compatible",
+    base_url: "https://example.com/v1",
+    encrypted_api_key: "enc",
+    input_cost: 1,
+    output_cost: 1,
+    context_window: 8000,
+  };
+  const catalog = await import("../src/lib/model-catalog.ts");
+  const flags = await import("../src/lib/consult/prep-flags.ts");
+  catalog.setModelCatalogIoForTests({
+    queryAdminRows: (async () => {
+      queries += 1;
+      if (queries === 3) throw new Error("database down");
+      if (queries >= 4) return [];
+      return [row];
+    }) as typeof import("../src/lib/admin/database.ts").queryAdminRows,
+    assertAllowedModelProviderUrl: (async (value: string) => {
+      dns += 1;
+      return { url: new URL(value), addresses: [] };
+    }) as typeof import("../src/lib/epay/gateway-policy.ts").assertAllowedModelProviderUrl,
+    decryptModelProviderApiKey: (() => "test-key") as typeof import("../src/lib/model-provider-policy.ts").decryptModelProviderApiKey,
+  });
+  process.env.MODEL_VERSION_CACHE_TTL_MS = "15000";
+  try {
+    catalog.invalidateLanguageModelCatalog();
+    const first = await catalog.resolveSessionLanguageModel("model-a", 4);
+    const second = await catalog.resolveSessionLanguageModel("model-a", 4);
+    assert.equal(first?.id, "model-a");
+    assert.equal(second, first);
+    assert.equal(queries, 1);
+    assert.equal(dns, 1);
+    catalog.invalidateLanguageModelCatalog();
+    await catalog.resolveSessionLanguageModel("model-a", 4);
+    assert.equal(queries, 2);
+    process.env.MODEL_VERSION_CACHE_TTL_MS = "0";
+    await catalog.resolveSessionLanguageModel("model-a", 4);
+    assert.ok(queries >= 3);
+    const failed = await catalog.resolveSessionLanguageModel("model-a", 4);
+    const empty = await catalog.resolveSessionLanguageModel("model-a", 4);
+    assert.equal(failed, null);
+    assert.equal(empty, null);
+    const queriesAfterMiss = queries;
+    await catalog.resolveSessionLanguageModel("model-a", 4);
+    assert.ok(queries > queriesAfterMiss);
+  } finally {
+    flags.resetConsultPrepFlagWarningsForTests();
+    delete process.env.MODEL_VERSION_CACHE_TTL_MS;
+    catalog.invalidateLanguageModelCatalog();
+    catalog.setModelCatalogIoForTests(null);
+  }
+});
+
+test("the title read starts after the question is saved and is awaited at the old use site", () => {
+  const route = readFileSync(new URL("../src/app/api/consult/route.ts", import.meta.url), "utf8");
+  const append = route.indexOf('accounting.rpc("append_consultation_question"');
+  const title = route.indexOf('.select("title")');
+  const classify = route.indexOf("await classifyConsultationTurn({");
+  const awaited = route.indexOf("await titleRead");
+  assert.ok(append > 0 && title > append && classify > title && awaited > classify);
+});
@@ -0,0 +1,79 @@
+/**
+ * In-process overlap check for consult preparation.
+ *
+ * This does not measure local Postgres. Those medians were not collected,
+ * so the switches stay at their code defaults.
+ */
+import { register } from "node:module";
+
+await register(new URL("./support/stub-server-only.mjs", import.meta.url).href, import.meta.url);
+const { loadConsultTurnHead } = await import("../src/lib/consult/turn-prep.ts");
+
+const delays = { deletion: 50, session: 80, model: 80 };
+
+async function run(parallel: boolean) {
+  let now = 0;
+  const waiters: Array<{ at: number; resolve: () => void }> = [];
+  let finished = false;
+  let inFlight = 0;
+  let maxInFlight = 0;
+  function sleep(ms: number) {
+    return new Promise<void>((resolve) => {
+      waiters.push({ at: now + ms, resolve });
+    });
+  }
+  async function occupy(ms: number) {
+    inFlight += 1;
+    maxInFlight = Math.max(maxInFlight, inFlight);
+    await sleep(ms);
+    inFlight -= 1;
+  }
+  const pending = loadConsultTurnHead({
+    parallel,
+    getUser: async () => ({ data: { user: { id: "user-1" } }, error: null }),
+    readPendingDeletion: () => occupy(delays.deletion).then(() => null),
+    readBody: async () => ({ success: true as const, data: { sessionId: "s", modelId: "m" } }),
+    readSession: () => occupy(delays.session).then(() => ({
+      data: { model_id: "m", model_config_version: 2, session_type: "consultation" },
+      error: null,
+    })),
+    resolveModel: () => occupy(delays.model).then(() => ({ id: "m" })),
+  }).then((head) => {
+    finished = true;
+    return head;
+  });
+  for (let guard = 0; guard < 40 && !finished; guard += 1) {
+    await Promise.resolve();
+    await Promise.resolve();
+    if (!waiters.length) continue;
+    waiters.sort((left, right) => left.at - right.at);
+    now = waiters[0]!.at;
+    const due: Array<() => void> = [];
+    while (waiters.length && waiters[0]!.at <= now) due.push(waiters.shift()!.resolve);
+    for (const resolve of due) resolve();
+  }
+  if (!finished) throw new Error("clock did not settle");
+  const head = await pending;
+  if (!head.ok) throw new Error("head failed");
+  return { elapsed: now, maxInFlight };
+}
+
+const on = await run(true);
+const off = await run(false);
+const report = {
+  kind: "logical-clock",
+  delays,
+  parallelElapsed: on.elapsed,
+  serialElapsed: off.elapsed,
+  parallelMaxInFlight: on.maxInFlight,
+  serialMaxInFlight: off.maxInFlight,
+  postgresMedians: "not-measured",
+  aliyunKey: "absent",
+  moderationSimulationMs: [300, 1500],
+  decision: {
+    CONSULT_PREP_PARALLEL: "off",
+    MODEL_VERSION_CACHE_TTL_MS: 0,
+    reason: "开启条件要求本机 Postgres 中位数。本轮没有量到，默认保持关闭。",
+  },
+};
+console.log(JSON.stringify(report));
@@ -0,0 +1,152 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+
+import { loadConsultTurnHead } from "../src/lib/consult/turn-prep.ts";
+
+type User = { id: string };
+type Parsed = { sessionId: string; modelId: string };
+type Session = { model_id: string | null; model_config_version: number | null; session_type: string | null };
+type Model = { id: string };
+
+function okSession(): Session {
+  return { model_id: "m", model_config_version: 2, session_type: "consultation" };
+}
+
+function deps(overrides: Record<string, unknown> = {}) {
+  const calls = { deletion: 0, session: 0, model: 0 };
+  const base = {
+    parallel: false,
+    getUser: async () => ({ data: { user: { id: "user-1" } as User }, error: null }),
+    readPendingDeletion: async () => {
+      calls.deletion += 1;
+      return null;
+    },
+    readBody: async () => ({ success: true as const, data: { sessionId: "session-1", modelId: "m" } as Parsed }),
+    readSession: async () => {
+      calls.session += 1;
+      return { data: okSession(), error: null };
+    },
+    resolveModel: async () => {
+      calls.model += 1;
+      return { id: "m" } as Model;
+    },
+    ...overrides,
+  };
+  return { calls, deps: base };
+}
+
+async function snapshot(response: Response) {
+  return { status: response.status, body: await response.json() };
+}
+
+async function both(overrides: Record<string, unknown> = {}) {
+  const off = deps({ ...overrides, parallel: false });
+  const on = deps({ ...overrides, parallel: true });
+  const [left, right] = await Promise.all([
+    loadConsultTurnHead(off.deps),
+    loadConsultTurnHead(on.deps),
+  ]);
+  assert.equal(left.ok, false);
+  assert.equal(right.ok, false);
+  if (left.ok || right.ok) return;
+  assert.deepEqual(await snapshot(left.response), await snapshot(right.response));
+}
+
+test("serial and parallel heads return the same status for each gate", async () => {
+  const anonymous = deps({
+    getUser: async () => ({ data: { user: null }, error: null }),
+  });
+  const missingUser = await loadConsultTurnHead({ ...anonymous.deps, parallel: false });
+  assert.equal(missingUser.ok, false);
+  if (!missingUser.ok) assert.equal(missingUser.response.status, 401);
+  assert.equal(anonymous.calls.deletion, 0);
+
+  await both({
+    readPendingDeletion: async () => Response.json({ error: "账号正在注销" }, { status: 423 }),
+    readBody: async () => ({ success: false as const, error: { flatten: () => ({ formErrors: ["bad"] }) } }),
+  });
+  await both({
+    readBody: async () => ({ success: false as const, error: { flatten: () => ({ formErrors: ["bad"] }) } }),
+  });
+  await both({ readSession: async () => ({ data: null, error: { message: "down" } }) });
+  await both({ readSession: async () => ({ data: null, error: null }) });
+  await both({
+    readSession: async () => ({ data: { ...okSession(), session_type: "rectification" }, error: null }),
+  });
+  await both({
+    readSession: async () => ({ data: { ...okSession(), model_id: "other" }, error: null }),
+  });
+  await both({ resolveModel: async () => null });
+});
+
+test("a frozen account wins over a bad request body", async () => {
+  const frozen = deps({
+    parallel: true,
+    readPendingDeletion: async () => Response.json({ error: "账号正在注销" }, { status: 423 }),
+    readBody: async () => ({ success: false as const, error: { flatten: () => ({}) } }),
+  });
+  const head = await loadConsultTurnHead(frozen.deps);
+  assert.equal(head.ok, false);
+  if (!head.ok) {
+    assert.equal(head.frozen, true);
+    assert.equal(head.response.status, 423);
+  }
+});
+
+test("parallel head overlaps deletion with session and model", async () => {
+  const delays = { deletion: 50, session: 80, model: 80 };
+  async function run(parallel: boolean) {
+    let now = 0;
+    const waiters: Array<{ at: number; resolve: () => void }> = [];
+    function sleep(ms: number) {
+      return new Promise<void>((resolve) => {
+        waiters.push({ at: now + ms, resolve });
+      });
+    }
+    let deletionStarted = 0;
+    let chainStarted = 0;
+    let inFlight = 0;
+    let maxInFlight = 0;
+    let finished = false;
+    async function occupy(ms: number, mark: () => void) {
+      mark();
+      inFlight += 1;
+      maxInFlight = Math.max(maxInFlight, inFlight);
+      await sleep(ms);
+      inFlight -= 1;
+    }
+    const pending = loadConsultTurnHead({
+      parallel,
+      getUser: async () => ({ data: { user: { id: "user-1" } }, error: null }),
+      readPendingDeletion: () => occupy(delays.deletion, () => { deletionStarted = now; }).then(() => null),
+      readBody: async () => ({ success: true as const, data: { sessionId: "s", modelId: "m" } }),
+      readSession: () => occupy(delays.session, () => { chainStarted = now; }).then(() => ({ data: okSession(), error: null })),
+      resolveModel: () => occupy(delays.model, () => {}).then(() => ({ id: "m" })),
+    }).then((head) => {
+      finished = true;
+      return head;
+    });
+    for (let guard = 0; guard < 40 && !finished; guard += 1) {
+      await Promise.resolve();
+      await Promise.resolve();
+      if (!waiters.length) continue;
+      waiters.sort((left, right) => left.at - right.at);
+      now = waiters[0]!.at;
+      const due: Array<() => void> = [];
+      while (waiters.length && waiters[0]!.at <= now) due.push(waiters.shift()!.resolve);
+      for (const resolve of due) resolve();
+    }
+    if (!finished) throw new Error("clock did not settle");
+    const head = await pending;
+    assert.equal(head.ok, true);
+    return { elapsed: now, deletionStarted, chainStarted, maxInFlight };
+  }
+  const on = await run(true);
+  const off = await run(false);
+  assert.ok(on.elapsed < 230, `parallel elapsed ${on.elapsed}`);
+  assert.equal(on.maxInFlight, 2);
+  assert.equal(off.maxInFlight, 1);
+  assert.equal(on.deletionStarted, on.chainStarted);
+  assert.ok(off.chainStarted >= delays.deletion);
+  assert.ok(off.elapsed > on.elapsed);
+});
@@ -138,7 +138,7 @@ type PlanResult = Record<string, unknown> & {

 async function runDomainPlan(
  domains: string[],
-  runWorkflow: (theme: string) => ReturnType<typeof workflow>,
+  runWorkflow: (theme: string) => ReturnType<typeof workflow> | Promise<ReturnType<typeof workflow>>,
  options: { now?: () => number; requestId?: string } = {},
 ) {
  const state = createConsultationRuntimeState();
@@ -623,6 +623,63 @@ test("the single-value domain form stays available to callers without the model
  assert.throws(() => canonicalDomainPlan({ theme: "prashna" }, {}), /unsupported_consultation_domain/);
 });

+test("two domains stay in plan order when concurrency is 2 and the second finishes first", async () => {
+  const previous = process.env.CONSULT_DOMAIN_CONCURRENCY;
+  process.env.CONSULT_DOMAIN_CONCURRENCY = "2";
+  const finished: string[] = [];
+  try {
+    const { result } = await runDomainPlan(["career", "marriage"], async (theme) => {
+      await new Promise((resolve) => setTimeout(resolve, theme === "career" ? 40 : 5));
+      finished.push(theme);
+      return workflow(theme);
+    });
+    assert.deepEqual(finished, ["marriage", "career"]);
+    assert.deepEqual(result.domains, ["career", "marriage"]);
+  } finally {
+    if (previous === undefined) delete process.env.CONSULT_DOMAIN_CONCURRENCY;
+    else process.env.CONSULT_DOMAIN_CONCURRENCY = previous;
+  }
+});
+
+test("a 429 from one parallel domain keeps workflow_rate_limited", async () => {
+  const previous = process.env.CONSULT_DOMAIN_CONCURRENCY;
+  process.env.CONSULT_DOMAIN_CONCURRENCY = "2";
+  try {
+    await assert.rejects(
+      () => runDomainPlan(["career", "marriage"], async () => {
+        throw new ConsultationWorkflowError("workflow_rate_limited", "busy");
+      }),
+      (error: unknown) => {
+        assert.equal(consultationToolFailureCode(error), "workflow_rate_limited");
+        return true;
+      },
+    );
+  } finally {
+    if (previous === undefined) delete process.env.CONSULT_DOMAIN_CONCURRENCY;
+    else process.env.CONSULT_DOMAIN_CONCURRENCY = previous;
+  }
+});
+
+test("concurrency 1 still runs domains one at a time", async () => {
+  const previous = process.env.CONSULT_DOMAIN_CONCURRENCY;
+  process.env.CONSULT_DOMAIN_CONCURRENCY = "1";
+  let active = 0;
+  let maxActive = 0;
+  try {
+    await runDomainPlan(["career", "marriage"], async (theme) => {
+      active += 1;
+      maxActive = Math.max(maxActive, active);
+      await new Promise((resolve) => setTimeout(resolve, 20));
+      active -= 1;
+      return workflow(theme);
+    });
+    assert.equal(maxActive, 1);
+  } finally {
+    if (previous === undefined) delete process.env.CONSULT_DOMAIN_CONCURRENCY;
+    else process.env.CONSULT_DOMAIN_CONCURRENCY = previous;
+  }
+});
+
 test("daily entrypoint ignores a model domain rewrite instead of executing it", async () => {
  const calls: string[] = [];
  const state = createConsultationRuntimeState();
@@ -157,8 +157,20 @@ test("consultation plans are server-owned and bounded", () => {
  const modelToolInput = tools.slice(tools.indexOf("const consultationToolInputSchema"), tools.indexOf("const MAX_RECORDED_STEPS"));
  assert.doesNotMatch(modelToolInput, /birth|latitude|longitude|engine|script|credit/i);
  const prepareCall = route.slice(route.indexOf("prepared = await prepareConsultationRoute"), route.indexOf("const modelSelection"));
-  assert.match(prepareCall, /beforeReserve: \(\{ consultationMode \}\) => createConsultationPlan\(\{/);
-  assert.ok(prepareCall.indexOf("beforeReserve:") < prepareCall.indexOf("reserve: () => reserveConsultationModel("));
+  // 原值：beforeReserve: ({ consultationMode }) => createConsultationPlan({，且早于 reserve: () => reserveConsultationModel(
+  // 新值：beforeReserve 的 async 函数体里调用 createConsultationPlan({，并且整段早于 reserve；reserve 的 async 函数体里调用 reserveConsultationModel(
+  // 原因：两处改成 async 函数体。仍锁调用发生在对应回调里，并且 beforeReserve 在 reserve 之前
+  const beforeReserveAt = prepareCall.indexOf("beforeReserve:");
+  const reserveAt = prepareCall.indexOf("reserve: async");
+  assert.ok(beforeReserveAt >= 0 && reserveAt > beforeReserveAt);
+  assert.match(
+    prepareCall.slice(beforeReserveAt, reserveAt),
+    /beforeReserve:\s*async\s*\(\{\s*consultationMode\s*\}\)[\s\S]*createConsultationPlan\(\{/,
+  );
+  assert.match(
+    prepareCall.slice(reserveAt),
+    /reserve:\s*async\s*\(\)\s*=>[\s\S]*reserveConsultationModel\(/,
+  );
  assert.ok(routeService.indexOf("const preReserveResult = input.beforeReserve") < routeService.indexOf("const reservation = await input.reserve()"));
  assert.match(route, /plan: prepared\.preReserveResult,[\s\S]*theme: consultationTheme/);
  assert.match(route, /runConsultationWorkflow\(toolInput, \{[\s\S]*plan: prepared\.preReserveResult/);
@@ -239,8 +239,15 @@ test("a blocked answer is replaced in place and onComplete settles the notice",

 test("routes check input before any charge and output before settlement", () => {
  const consult = read("../src/app/api/consult/route.ts");
-  assert.ok(consult.indexOf('await moderate(visibleQuestion, "input"') > 0);
-  assert.ok(consult.indexOf('await moderate(visibleQuestion, "input"') < consult.indexOf("reserve_consultation_usage"), "input check precedes the reservation");
+  const prepare = read("../src/lib/consultation-route-service.ts");
+  // 原值: 源码含 await moderate(visibleQuestion, "input" 且位置早于 reserve_consultation_usage
+  // 新值: beforeReserve 里 await settleInputModeration()，而 prepare 里 beforeReserve 早于 input.reserve()
+  // 原因: 审核从开流前挪到流内、预留前；开关关掉时仍在开流前 await moderate
+  assert.match(consult, /await settleInputModeration\(\)/);
+  assert.ok(consult.indexOf("await settleInputModeration()") < consult.indexOf("reserve_consultation_usage"));
+  const beforeReserve = prepare.indexOf("input.beforeReserve");
+  const reserveCall = prepare.indexOf("input.reserve()");
+  assert.ok(beforeReserve > 0 && beforeReserve < reserveCall);
  assert.equal((consult.match(/moderateOutput: moderateAnswer/g) ?? []).length, 3, "every agentic answer stream is checked");
  assert.match(consult, /if \(!outputModerated\) await moderateAnswer\(reply\.text\);\n\s+if \(outputBlocked\) return await completeModeratedResponse\(usage\);/);
  assert.match(consult, /complete_consultation_moderated/);
@@ -45,9 +45,18 @@ test("catalog cache is short-lived and failure stays fail-closed without env fal
 test("session model resolution is async and pins published or retired database versions", () => {
  assert.match(catalog, /export async function resolveSessionLanguageModel/);
  assert.match(catalog, /c\.model_id=\$1 and v\.version=\$2 and v\.status in \('published','retired'\)/);
-  assert.match(catalog, /return await readVersion\(modelId, configVersion\)/);
+  // 原值: return await readVersion(modelId, configVersion)
+  // 新值: const model = await readVersion(modelId, configVersion)，只有非空结果写入短缓存
+  // 原因: 绑版本的解析与目录共用 15 秒上限，失败和空结果不缓存
+  assert.match(catalog, /const model = await readVersion\(modelId, configVersion\)/);
+  assert.match(catalog, /if \(model && ttl > 0\) cache\.set\(key, \{ expiresAt: Date\.now\(\) \+ ttl, model \}\)/);
  assert.match(routes.consult, /await resolveSessionLanguageModel\(\s*chatSession\.model_id,\s*chatSession\.model_config_version/);
-  assert.match(routes.consult, /if \(!sessionModel\)[\s\S]*status: 503/);
+  // 原值: route.ts 里 if (!sessionModel) 之后是 status 503
+  // 新值: turn-prep.ts 里 if (!sessionModel) 返回 modelUnavailable()，正文 status 503
+  // 原因: 模型不可用仍是开流前的 503，文案随准备步骤一起搬走
+  const head = readFileSync(new URL("../src/lib/consult/turn-prep.ts", import.meta.url), "utf8");
+  assert.match(head, /if \(!sessionModel\) return \{ ok: false, response: modelUnavailable\(\) \}/);
+  assert.match(head, /function modelUnavailable\(\)[\s\S]*status: 503/);
  assert.doesNotMatch(routes.consult, /languageModelConfigurationMessage/);
 });

@@ -18,7 +18,6 @@ import {
 const read = (relativePath: string) => readFileSync(new URL(relativePath, import.meta.url), "utf8");
 const run = read("../src/hooks/use-consultation-run.ts");
 const sessions = read("../src/hooks/use-session-management.ts");
-const consultRoute = read("../src/app/api/consult/route.ts");
 const pageSource = read("../src/app/(app)/page.tsx");
 const sendFn = run.slice(run.indexOf("async function send("), run.indexOf('await fetch("/api/consult"'));

@@ -50,13 +49,16 @@ test("pending consult replay pointing at a rectification session is dropped", ()
 });

 test("consult POST splits missing session from wrong session type", () => {
-  assert.match(consultRoute, /if \(!chatSession\) \{[\s\S]*咨询会话不存在[\s\S]*status: 404/);
-  assert.match(
-    consultRoute,
-    /if \(chatSession\.session_type !== "consultation"\) \{[\s\S]*session_not_consultation[\s\S]*status: 409/,
-  );
-  assert.match(consultRoute, /这是生时校正会话/);
-  assert.match(consultRoute, /请在生时校正里继续。/);
+  const head = readFileSync(new URL("../src/lib/consult/turn-prep.ts", import.meta.url), "utf8");
+  // 原值: consult route 里 if (!chatSession) 返回 404，类型不对返回 session_not_consultation。
+  // 新值: 同样两句在 turn-prep.ts，route 把读会话的结果交给它。
+  // 原因: 开流前的会话判断收进准备函数，文案和状态码没变。
+  assert.match(head, /if \(!session\.data\) return \{ ok: false, response: sessionMissing\(\) \}/);
+  assert.match(head, /function sessionMissing\(\)[\s\S]*咨询会话不存在[\s\S]*status: 404/);
+  assert.match(head, /function notConsultation\(\)[\s\S]*session_not_consultation[\s\S]*status: 409/);
+  assert.match(head, /session_type !== "consultation"/);
+  assert.match(head, /这是生时校正会话/);
+  assert.match(head, /请在生时校正里继续。/);
  assert.equal(SESSION_NOT_CONSULTATION_CODE, "session_not_consultation");
  assert.match(run, /caught\.code === SESSION_NOT_CONSULTATION_CODE/);
 });
@@ -0,0 +1 @@
+export {};
@@ -0,0 +1,8 @@
+const empty = new URL("./server-only-empty.mjs", import.meta.url).href;
+
+export async function resolve(specifier, context, nextResolve) {
+  if (specifier === "server-only") {
+    return { url: empty, shortCircuit: true };
+  }
+  return nextResolve(specifier, context);
+}