From a0e223e20cebcce3479accb60f6aec44d1caadda Mon Sep 17 00:00:00 2001 From: jesse-ux Date: Wed, 14 Oct 2026 07:23:58 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8F=91=E9=97=AE=E5=87=86=E5=A4=87=E5=8F=AF?= =?UTF-8?q?=E4=BB=A5=E5=B9=B6=E8=A1=8C=EF=BC=8C=E4=B8=89=E4=B8=AA=E5=BC=80?= =?UTF-8?q?=E5=85=B3=E5=85=88=E4=BF=9D=E6=8C=81=E5=85=B3=E9=97=AD=EF=BC=88?= =?UTF-8?q?BUG-1337=EF=BD=9E1339=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 开流前的注销检查可以和读会话、解析模型同时进行；审核改到扣点前等待，不再挡住第一行；模型版本可以短缓存；两个领域可以一起算。本机没有数据库中位数，也没有带星盘网络库的镜像，所以默认仍是关闭、缓存 0、领域逐个算。 自测：tsc 0 错；lint 0 error；相关测试失败 0、取消 0。test:db 未跑。 Co-Authored-By: Grok (agent loop) --- BLOCKED.md | 7 + docs/BUG_HISTORY.md | 45 ++++ ...PROGRESS-consult-prep-parallel-20261013.md | 58 +++++ docs/tasks/README.md | 2 +- .../consult-prep-parallel-ab-20261013.md | 36 +++ .../research/consult-domain-concurrency-ab.py | 24 ++ .../research/consult-prep-parallel-ab.mts | 79 ++++++ frontend/src/app/api/consult/route.ts | 246 +++++++++++------- frontend/src/lib/consult/prep-flags.ts | 62 +++++ frontend/src/lib/consult/prep-log.ts | 34 +++ frontend/src/lib/consult/turn-prep.ts | 195 ++++++++++++++ frontend/src/lib/model-catalog.ts | 53 +++- frontend/src/mastra/consultation-tools.ts | 158 +++++++---- ...t-first-frame-and-pacing-20260928.test.tsx | 13 +- .../tests/consult-prep-gate-20261013.test.ts | 164 ++++++++++++ .../tests/consult-turn-prep-20261013.test.ts | 152 +++++++++++ .../consultation-agentic-runtime.test.ts | 59 ++++- frontend/tests/content-moderation.test.ts | 11 +- frontend/tests/model-catalog.test.ts | 13 +- ...ctification-session-composer-guard.test.ts | 18 +- frontend/tests/support/server-only-empty.mjs | 1 + frontend/tests/support/stub-server-only.mjs | 8 + 22 files changed, 1276 insertions(+), 162 deletions(-) create mode 100644 docs/tasks/PROGRESS-consult-prep-parallel-20261013.md create mode 100644 docs/testing/consult-prep-parallel-ab-20261013.md create mode 100644 frontend/scripts/research/consult-domain-concurrency-ab.py create mode 100644 frontend/scripts/research/consult-prep-parallel-ab.mts create mode 100644 frontend/src/lib/consult/prep-flags.ts create mode 100644 frontend/src/lib/consult/prep-log.ts create mode 100644 frontend/src/lib/consult/turn-prep.ts create mode 100644 frontend/tests/consult-prep-gate-20261013.test.ts create mode 100644 frontend/tests/consult-turn-prep-20261013.test.ts create mode 100644 frontend/tests/support/server-only-empty.mjs create mode 100644 frontend/tests/support/stub-server-only.mjs diff --git a/BLOCKED.md b/BLOCKED.md index ce413644..2f00689d 100644 --- a/BLOCKED.md +++ b/BLOCKED.md @@ -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`。 diff --git a/docs/BUG_HISTORY.md b/docs/BUG_HISTORY.md index add22027..5e7256ca 100644 --- a/docs/BUG_HISTORY.md +++ b/docs/BUG_HISTORY.md @@ -17795,3 +17795,48 @@ - 复发自：无。 - 修复版本：未修复。 +## BUG-1337 | 发问前互不依赖的准备步骤仍一个一个等 + +- 状态：fixed-pending-verify +- 首次发现 / 最近更新：2026-10-14 / 2026-10-14 +- 来源：issue #62。任务书 T1、T2、T4。 +- 现象：登录之后，注销检查、读请求、读会话、解析模型、输入审核、再读一次标题，彼此能同时做的地方仍排成一队。打开远程审核时，第一行最多晚约 1.5 秒。 +- 触发条件：普通对话发问。 +- 根因：这些步骤按当时的目的写成了串行。先建流只把更后面的模型工作搬进了流里。 +- 修复：准备步骤收进一个函数。并行开关打开时，注销检查和读请求、读会话、解析模型同时进行，失败仍按原来的状态码顺序返回。审核在扣点前等待，不再挡住第一行。标题在问题保存成功后就开始读。开关默认关闭，因为本机没有量到数据库中位数。 +- 验证：`consult-turn-prep-20261013.test.ts` 两种开关的状态码和正文一致，假时钟并行 160、串行 210。审核挂起 1.5 秒时第一小段仍先写出。拦住时不扣点。 +- 防复发：冻结账号仍在扣点前返回；被拦的问题不调用扣点。开启默认值前必须有本机数据库中位数，并且首行至少快两成。 +- 相关记录：BUG-1074、BUG-976、BUG-1338。 +- 复发自：无。 +- 修复版本：分支 `codex/consult-prep-parallel-20261013`，尚未合入 staging。默认开关仍关。 + +## BUG-1338 | 会话绑定的模型版本每次都重新查库并检查地址 + +- 状态：fixed-pending-verify +- 首次发现 / 最近更新：2026-10-14 / 2026-10-14 +- 来源：issue #62。任务书 T3。 +- 现象：普通对话每轮都按模型和版本查一次库、解密、对兼容供应商做地址检查。同一文件里没绑版本的目录已经缓存 15 秒。 +- 触发条件：会话带了模型版本号。 +- 根因：版本解析是后加的，没有沿用目录缓存。已发布或已退役版本用的供应商代码、类型和地址不能改，但启用开关和密钥仍能改，所以不能缓存太久，也不能缓存失败。 +- 修复：成功的解析可以缓存，上限 15 秒，和目录一样。失败、空结果、地址检查失败都不缓存。管理端保存供应商、发布、回滚时本来就会清目录缓存，现在一并清掉版本缓存。地址检查函数没改。缓存时间默认 0，因为没有量到「准备模型」快了一半。 +- 验证：行为测试在 15 秒内第二次不再查库、不再做地址检查；清缓存或关掉缓存后会再查；失败和空结果之后的下一次仍会查。 +- 防复发：缓存时间不得超过目录的 15 秒。缓存失败会把停用或地址拒绝钉住，因此禁止。 +- 相关记录：BUG-148、BUG-1337。 +- 复发自：无。 +- 修复版本：分支 `codex/consult-prep-parallel-20261013`，尚未合入 staging。默认缓存时间仍是 0。 + +## BUG-1339 | 一次问到的多个领域仍逐个计算 + +- 状态：closed_by_design +- 首次发现 / 最近更新：2026-10-14 / 2026-10-14 +- 来源：issue #62。任务书 T5。 +- 现象：一次咨询如果要算两个领域，第二个要等第一个算完。 +- 触发条件：普通对话的计算计划里有两个领域。 +- 根因：领域上限来自单次时限。并行会占满两个重计算名额，热缓存时第二个领域本来能复用第一个的排盘。任务书起草试算大约只快一成，热缓存时更慢。 +- 修复：加了开关。默认仍是逐个算。打开时两个领域一起算，结果仍按计划顺序拼，任一失败整次失败，429 仍是原来的错误码。本轮没有拿带星盘网络库的镜像重跑，所以不把默认改成 2。 +- 验证：后完成的领域仍排在计划后面；一个 429 时错误码不变；开关为 1 时同时进行的领域数是 1。镜像墙钟、争用 429 次数和逐字节差分都没有跑。 +- 防复发：BUG-257 仍在。没有「至少快两成五、429 不增加、差分一致」之前，默认保持 1。领域个数上限和逐个时的时间预算公式没改。 +- 相关记录：BUG-257、BUG-723、BUG-1142。 +- 复发自：无。 +- 修复版本：未改默认行为。分支 `codex/consult-prep-parallel-20261013`。 + diff --git a/docs/tasks/PROGRESS-consult-prep-parallel-20261013.md b/docs/tasks/PROGRESS-consult-prep-parallel-20261013.md new file mode 100644 index 00000000..5a8ae7f4 --- /dev/null +++ b/docs/tasks/PROGRESS-consult-prep-parallel-20261013.md @@ -0,0 +1,58 @@ +# 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，写明没有跑镜像 | diff --git a/docs/tasks/README.md b/docs/tasks/README.md index 58b326a6..9fd08cec 100644 --- a/docs/tasks/README.md +++ b/docs/tasks/README.md @@ -472,7 +472,7 @@ | `TASK-home-sidebar-rerender-20261013.md` | — | 流式回答时侧栏每帧整栏重渲染（BUG-1329，A/B） | 待领取 | `codex/home-sidebar-rerender-20261013` | | `TASK-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` | — | 长报告出稿提速：术语替换命中预检、哈希单次遍历、同盘复用（BUG-1335～1336，A/B） | 待领取 | `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-20261013.md` | `PROGRESS-consult-prep-parallel-20261013.md` | 普通对话发问前准备步骤并行、模型版本短缓存、多领域并发开关（BUG-1337～1339，A/B） | 待验收 | `codex/consult-prep-parallel-20261013` | | `TASK-frontend-dead-code-20261013.md` | — | 前端死代码 17 文件与无调用方接口删除，含未登录重计算接口（BUG-1342） | 待领取 | `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` | | `TASK-answer-effort-ab-20261013.md` | — | 答题步推理强度开关 + A/B（需模型 key） | 待领取（排队：consult-prep-parallel） | `codex/answer-effort-ab-20261013` | diff --git a/docs/testing/consult-prep-parallel-ab-20261013.md b/docs/testing/consult-prep-parallel-ab-20261013.md new file mode 100644 index 00000000..ece10232 --- /dev/null +++ b/docs/testing/consult-prep-parallel-ab-20261013.md @@ -0,0 +1,36 @@ +# 发问准备并行：本轮量到的和没量到的（2026-10-14） + +开关都还关着。用户发问后的第一行，这轮不会变快。 + +## 进程内假时钟 + +命令：`node node_modules/tsx/dist/cli.mjs scripts/research/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 不增加，响应除约定字段外一致 | diff --git a/frontend/scripts/research/consult-domain-concurrency-ab.py b/frontend/scripts/research/consult-domain-concurrency-ab.py new file mode 100644 index 00000000..8b36af3f --- /dev/null +++ b/frontend/scripts/research/consult-domain-concurrency-ab.py @@ -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()) diff --git a/frontend/scripts/research/consult-prep-parallel-ab.mts b/frontend/scripts/research/consult-prep-parallel-ab.mts new file mode 100644 index 00000000..55bc491f --- /dev/null +++ b/frontend/scripts/research/consult-prep-parallel-ab.mts @@ -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("../../tests/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((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)); diff --git a/frontend/src/app/api/consult/route.ts b/frontend/src/app/api/consult/route.ts index 06ab53f2..e9e6a056 100644 --- a/frontend/src/app/api/consult/route.ts +++ b/frontend/src/app/api/consult/route.ts @@ -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 | 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 { + 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>>; type ReservationResult = { success: boolean; credits: number | null; error_code: string | null }; type ModelSelection = Awaited>>; @@ -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 => { let prepared: PreparedConsultationRoute; + const prepareStarted = Date.now(); try { prepared = await prepareConsultationRoute({ 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 | 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 { diff --git a/frontend/src/lib/consult/prep-flags.ts b/frontend/src/lib/consult/prep-flags.ts new file mode 100644 index 00000000..b1b52a24 --- /dev/null +++ b/frontend/src/lib/consult/prep-flags.ts @@ -0,0 +1,62 @@ +/** Call-time switches for consult preparation. Read on each request, not at import. */ + +const warned = new Set(); + +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>, 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> = 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> = 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> = 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(); +} diff --git a/frontend/src/lib/consult/prep-log.ts b/frontend/src/lib/consult/prep-log.ts new file mode 100644 index 00000000..6888671a --- /dev/null +++ b/frontend/src/lib/consult/prep-log.ts @@ -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, + }; +} diff --git a/frontend/src/lib/consult/turn-prep.ts b/frontend/src/lib/consult/turn-prep.ts new file mode 100644 index 00000000..fe79768c --- /dev/null +++ b/frontend/src/lib/consult/turn-prep.ts @@ -0,0 +1,195 @@ +import { NextResponse } from "next/server"; + +export type ConsultPrepTimings = { + authMs: number; + deletionMs: number; + sessionMs: number; + modelMs: number; +}; + +type AuthResult = { + data: { user: User | null }; + error: unknown; +}; + +type BodyResult = + | { success: true; data: Parsed } + | { success: false; error: { flatten: () => unknown } }; + +type SessionResult = { + 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 = { + parallel: boolean; + getUser: () => Promise>; + readPendingDeletion: (userId: string) => Promise; + readBody: () => Promise>; + readSession: (userId: string, parsed: Parsed) => Promise>; + resolveModel: (chatSession: Session) => Promise; +}; + +export type ConsultTurnHead = + | { 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 = + | { 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, +): Promise> { + 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> => { + 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 }; +} diff --git a/frontend/src/lib/model-catalog.ts b/frontend/src/lib/model-catalog.ts index 9d68e15b..aa5a8ada 100644 --- a/frontend/src/lib/model-catalog.ts +++ b/frontend/src/lib/model-catalog.ts @@ -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 | 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; +}; 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 { - 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(` + return catalogIo.queryAdminRows(` 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 { + 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(); +} diff --git a/frontend/src/mastra/consultation-tools.ts b/frontend/src/mastra/consultation-tools.ts index 96c2681e..47d9e899 100644 --- a/frontend/src/mastra/consultation-tools.ts +++ b/frontend/src/mastra/consultation-tools.ts @@ -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. diff --git a/frontend/tests/consult-first-frame-and-pacing-20260928.test.tsx b/frontend/tests/consult-first-frame-and-pacing-20260928.test.tsx index d639dd2c..a1e36050 100644 --- a/frontend/tests/consult-first-frame-and-pacing-20260928.test.tsx +++ b/frontend/tests/consult-first-frame-and-pacing-20260928.test.tsx @@ -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 =>"); 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 [ diff --git a/frontend/tests/consult-prep-gate-20261013.test.ts b/frontend/tests/consult-prep-gate-20261013.test.ts new file mode 100644 index 00000000..5acfdb0b --- /dev/null +++ b/frontend/tests/consult-prep-gate-20261013.test.ts @@ -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((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); +}); diff --git a/frontend/tests/consult-turn-prep-20261013.test.ts b/frontend/tests/consult-turn-prep-20261013.test.ts new file mode 100644 index 00000000..28ace9cd --- /dev/null +++ b/frontend/tests/consult-turn-prep-20261013.test.ts @@ -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 = {}) { + 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 = {}) { + 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((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); +}); diff --git a/frontend/tests/consultation-agentic-runtime.test.ts b/frontend/tests/consultation-agentic-runtime.test.ts index cab14bc8..13204d55 100644 --- a/frontend/tests/consultation-agentic-runtime.test.ts +++ b/frontend/tests/consultation-agentic-runtime.test.ts @@ -138,7 +138,7 @@ type PlanResult = Record & { async function runDomainPlan( domains: string[], - runWorkflow: (theme: string) => ReturnType, + runWorkflow: (theme: string) => ReturnType | Promise>, 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(); diff --git a/frontend/tests/content-moderation.test.ts b/frontend/tests/content-moderation.test.ts index 97699890..dfc1126d 100644 --- a/frontend/tests/content-moderation.test.ts +++ b/frontend/tests/content-moderation.test.ts @@ -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/); diff --git a/frontend/tests/model-catalog.test.ts b/frontend/tests/model-catalog.test.ts index 68a84dac..83591a87 100644 --- a/frontend/tests/model-catalog.test.ts +++ b/frontend/tests/model-catalog.test.ts @@ -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/); }); diff --git a/frontend/tests/rectification-session-composer-guard.test.ts b/frontend/tests/rectification-session-composer-guard.test.ts index ba8c90c5..3715acbf 100644 --- a/frontend/tests/rectification-session-composer-guard.test.ts +++ b/frontend/tests/rectification-session-composer-guard.test.ts @@ -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/); }); diff --git a/frontend/tests/support/server-only-empty.mjs b/frontend/tests/support/server-only-empty.mjs new file mode 100644 index 00000000..cb0ff5c3 --- /dev/null +++ b/frontend/tests/support/server-only-empty.mjs @@ -0,0 +1 @@ +export {}; diff --git a/frontend/tests/support/stub-server-only.mjs b/frontend/tests/support/stub-server-only.mjs new file mode 100644 index 00000000..8683ef09 --- /dev/null +++ b/frontend/tests/support/stub-server-only.mjs @@ -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); +}