Compare commits

...
Author SHA1 Message Date
jesse-uxandGrok (agent loop) 360a667e4f fix(capacity): align two stale checks with the rate-limit edit.
The consultation contract still required await directly before runConsultationWorkflow(toolInput), and a new report test shifted three privacy allowlist lines. Keep the toolInput lock and move that test to the end. No product code change.

Co-Authored-By: Grok (agent loop)
2026-10-15 15:31:35 +08:00
jesse-uxandGrok (agent loop) 904dfe9f79 merge origin/staging into beta capacity (issue #76).
Bring in the bounce-1 fix brief without rewriting the pushed capacity commit.

Co-Authored-By: Grok (agent loop)
2026-10-15 15:24:52 +08:00
Jesse_ChenandClaude Opus 5.5 9fae294532 docs(tasks): beta capacity fix brief (agent loop, issue #76, bounce 1)
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LutgzyJqNhPpZy1nFsJtN3
2026-10-15 15:10:37 +08:00
jesse-uxandGrok (agent loop) a54f75e5e3 fix(capacity): limit each user and queue heavy compute briefly.
Consult, rectification turns, and report generation stop before a credit reservation when the product caps are hit. A full heavy-compute gate waits up to 8 seconds, and one consultation retries that 429 inside the same request.

Co-Authored-By: Grok (agent loop)
2026-10-15 13:33:06 +08:00
Claude ef49257b73 merge origin/staging into answer effort ab (issue #65)
Independent Staging Quality Gate / validate (push) Successful in 14m3s
Independent Staging Quality Gate / publish (push) Successful in 3m47s
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-15 11:15:23 +08:00
Claude 7175548531 merge origin/staging into api chart subprocess gate (issue #64)
Independent Staging Quality Gate / validate (push) Successful in 13m47s
Independent Staging Quality Gate / publish (push) Successful in 4m5s
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-15 10:44:32 +08:00
jesse-uxandGrok (agent loop) e21e167a73 fix(chart): lock synastry chart calls and stop overwriting the A/B note
The bounced contract still looked for the old chart calls. The research script now writes to a temp directory unless --out is set. Quota splits stay unimplemented.

Co-Authored-By: Grok (agent loop)
2026-10-15 10:19:41 +08:00
jesse-uxandGrok (agent loop) 18fdce6006 merge origin/staging into the chart subprocess gate branch
Keep the pushed fix commit and bring in the bounce brief plus the data-layer fix.

Co-Authored-By: Grok (agent loop)
2026-10-15 10:11:38 +08:00
jesse-uxGrok (agent loop)Cursor
336286e428 fix(chart): skip unused synastry overview and doomed report subprocesses
Synastry charts now skip the unused external overview and run together. Reports skip PyJHora subprocesses when that library is missing, and a full heavy-compute slot logs which paths hold it. Splitting quotas stays a simulation.

Gate, growth, and report pytest passed, as did the frontend route tests, tsc, lint, and the A/B script.

Co-Authored-By: Grok (agent loop)
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-10-15 08:02:39 +08:00
47 changed files with 2286 additions and 88 deletions
+10
View File
@@ -1,5 +1,15 @@
# 印度占星 Skill 更新日志
## 2026-10-15 — 内测容量:问得太密会停一下,人多时先排一小会儿
- 同一个人短时间问得太密、校正答得太密,或一小时里报告生成太多,会先停一下,这次不扣点。
- 同时算的人多过机器能接的两路时,先排一小会儿。排上了继续;排不上会说人有点多,并把这一轮预留的点数退回。
- Skill 版本不变(BUG-1355、BUG-1356)。
## 2026-10-15 — 合盘更快(待合入 staging)
- 合盘排两张盘时关掉用不上的外部概览,两张盘一起排。个人报告在没装参照引擎时不再为注定失败的回放反复起进程。轻重任务要不要分开限流,这一单只给了模拟表,还没改规则。
## 2026-10-15 — 星盘页上升行补上星宿,表头对齐(待合入 staging)
- 本命盘和月亮盘的上升行现在有星宿、宿主和 pada。状态写「不适用」,不再是空横杠。分盘等表里居中的那几列,表头也居中。当前大运不再画左侧那条线,浅色底还在。已经打开过的星盘会重新计算。Skill 版本不变(BUG-1352、BUG-1353)。
+1 -1
View File
@@ -37,7 +37,7 @@ The API image `WORKDIR` is `/app` (`deploy/railway-api.Dockerfile`). Runtime fil
The volume is not a backup domain. Chart cache is a TTL'd compute cache (default 900s) and is cheaper to rebuild than to archive. Async / high-rigor job state is short-lived (default TTL 3600s) and is not a restore unit. Restart persistence is the only reason the volume exists. Staging encrypted backups remain PostgreSQL-only via `deploy/backup-staging-postgres.sh`; do not add `api_scratch` to that helper. Project-prefixed volume names are `jyotisha-staging_api_scratch` and `jyotisha-production_api_scratch`.
Heavy rectification scans and high-rigor workflows share a process-wide fail-fast concurrency gate (`JYOTISH_HEAVY_COMPUTE_CONCURRENCY`, default `2` to match the 2 vCPU host). Saturated requests return HTTP 429 with `Retry-After` and are not queued. Health checks and other light routes do not take a slot.
Heavy rectification scans and high-rigor workflows share a process-wide concurrency gate (`JYOTISH_HEAVY_COMPUTE_CONCURRENCY`, default `2` to match the 2 vCPU host). A saturated request waits up to 8 seconds if fewer than 4 requests are already waiting, then returns HTTP 429 with `Retry-After`. Health checks and other light routes do not take a slot.
## DNS and Supabase Auth
+107
View File
@@ -17991,6 +17991,36 @@
- 复发自:无。
- 修复版本:分支 `codex/api-disk-cache-retention-20261013`,尚未合入 staging。
## BUG-1333 | 合盘排盘漏关外部概览,而且两张盘排队算
- 状态:fixed-pending-verify
- 首次发现 / 最近更新:2026-10-13 / 2026-10-15
- 来源:issue #64。任务书 `docs/tasks/TASK-api-chart-subprocess-gate-20261013.md` T1、T3、T4。
- 现象:合盘每次要排两张盘,都没关掉用不上的外部概览。每张盘多起两个后台进程,而且两张盘一张一张排。名额占满时,日志里看不出是谁占着。
- 触发条件:打开合盘。排盘接口本身默认仍走完整链,给报告和外部调用留着。
- 根因:合盘这条路由没跟上「只读盘面就关掉外部概览」的约定,也没有合同测试拦住。名额按路径一刀切,不带 skip 的排盘默认不占名额。
- 修复:合盘两次排盘都带上关闭外部概览,两张盘同时排;婚恋分支的两张分盘也同时排。前端里凡是发给排盘接口的请求体,都要带这个开关,由源码合同守着。另加一个默认关闭的开关:打开后,不带 skip 的同步排盘才占一个重计算名额。名额占满时记一行日志,只写被拒的路径和当前占用者的路径、已占秒数,不写请求正文。
- 验证:`frontend/tests/synastry-chart-skip.test.ts`、`frontend/tests/chart-request-skip-contract.test.ts`、`tests/test_api_chart_subprocess_gate.py`、`tests/test_api_heavy_compute_gate.py`。合盘耗时见 `docs/testing/api-chart-subprocess-gate-ab-20261013.md`。`tests/test_supabase_user_data_contract.py` 锁住合盘路由向 `/api/chart` 请求本人和对方两张盘,且两次都带 `skip_vedastro_main_entry_overview: true`。
- 防复发:源码合同要求 `frontend/src/app/api` 和 `frontend/src/lib` 里发往 `/api/chart` 的请求体带 `skip_vedastro_main_entry_overview: true`。Python 合同按路由原文锁住这两次调用和这个开关。`/api/chart` 仍不在重计算路径名单里。
- 相关记录:BUG-161、BUG-715、BUG-717、BUG-1302。
- 复发自:无。
- 修复版本:分支 `codex/api-chart-subprocess-gate-20261013`。编号是任务书预留的空号,夹在 BUG-1332 和 BUG-1335 之间。
## BUG-1334 | 个人报告在没有参照引擎时仍为每次报告启动子进程
- 状态:fixed-pending-verify
- 首次发现 / 最近更新:2026-10-13 / 2026-10-15
- 来源:issue #64。任务书 T2。
- 现象:每份个人报告会起大约 10 个后台进程。其中参照引擎没装时,年度回放注定失败,但仍要付启动开销。另一段静态校验和用户的盘无关,每份报告都重跑。
- 触发条件:生成个人长报告,且这台机器没有安装 PyJHora。装了参照引擎的开发机上,隔离用的子进程仍然要留着。
- 根因:子进程是为了把参照引擎隔离开,这是对的;父进程却没有先问「这台机器根本没装」。静态校验文件被挂在每份报告上,没有按文件变化复用。
- 修复:父进程先看一次能不能导入参照引擎,不能就不起子进程,返回和「子进程在没装时」相同的结果。解释器查找和静态校验各在进程内记一次;校验文件变了会重算。不缓存任何出生资料。装了参照引擎时仍走子进程。
- 验证:`tests/test_api_chart_subprocess_gate.py` 对照 `tests/fixtures/api-chart-subprocess-gate/`。生产上的收益大约是每份报告 0.5 到 0.9 秒、少掉那几个注定失败的进程,不是排查时在装了参照引擎的机器上看到的 5 秒。本机没有 PyJHora,实测见 `docs/testing/api-chart-subprocess-gate-ab-20261013.md`。
- 防复发:无参照引擎时测试断言不会调用 `subprocess.run`,返回值与固定样本一致。
- 相关记录:BUG-721、BUG-1265、BUG-1333。
- 复发自:无。
- 修复版本:分支 `codex/api-chart-subprocess-gate-20261013`。
## BUG-1335 | 个人长报告的中文出稿和溯源哈希重复劳动
- 状态:investigating
@@ -36044,6 +36074,36 @@
- 复发自:无。
- 修复版本:分支 `codex/api-disk-cache-retention-20261013`,尚未合入 staging。
## BUG-1333 | 合盘排盘漏关外部概览,而且两张盘排队算
- 状态:fixed-pending-verify
- 首次发现 / 最近更新:2026-10-13 / 2026-10-15
- 来源:issue #64。任务书 `docs/tasks/TASK-api-chart-subprocess-gate-20261013.md` T1、T3、T4。
- 现象:合盘每次要排两张盘,都没关掉用不上的外部概览。每张盘多起两个后台进程,而且两张盘一张一张排。名额占满时,日志里看不出是谁占着。
- 触发条件:打开合盘。排盘接口本身默认仍走完整链,给报告和外部调用留着。
- 根因:合盘这条路由没跟上「只读盘面就关掉外部概览」的约定,也没有合同测试拦住。名额按路径一刀切,不带 skip 的排盘默认不占名额。
- 修复:合盘两次排盘都带上关闭外部概览,两张盘同时排;婚恋分支的两张分盘也同时排。前端里凡是发给排盘接口的请求体,都要带这个开关,由源码合同守着。另加一个默认关闭的开关:打开后,不带 skip 的同步排盘才占一个重计算名额。名额占满时记一行日志,只写被拒的路径和当前占用者的路径、已占秒数,不写请求正文。
- 验证:`frontend/tests/synastry-chart-skip.test.ts`、`frontend/tests/chart-request-skip-contract.test.ts`、`tests/test_api_chart_subprocess_gate.py`、`tests/test_api_heavy_compute_gate.py`。合盘耗时见 `docs/testing/api-chart-subprocess-gate-ab-20261013.md`。`tests/test_supabase_user_data_contract.py` 锁住合盘路由向 `/api/chart` 请求本人和对方两张盘,且两次都带 `skip_vedastro_main_entry_overview: true`。
- 防复发:源码合同要求 `frontend/src/app/api` 和 `frontend/src/lib` 里发往 `/api/chart` 的请求体带 `skip_vedastro_main_entry_overview: true`。Python 合同按路由原文锁住这两次调用和这个开关。`/api/chart` 仍不在重计算路径名单里。
- 相关记录:BUG-161、BUG-715、BUG-717、BUG-1302。
- 复发自:无。
- 修复版本:分支 `codex/api-chart-subprocess-gate-20261013`。编号是任务书预留的空号,夹在 BUG-1332 和 BUG-1335 之间。
## BUG-1334 | 个人报告在没有参照引擎时仍为每次报告启动子进程
- 状态:fixed-pending-verify
- 首次发现 / 最近更新:2026-10-13 / 2026-10-15
- 来源:issue #64。任务书 T2。
- 现象:每份个人报告会起大约 10 个后台进程。其中参照引擎没装时,年度回放注定失败,但仍要付启动开销。另一段静态校验和用户的盘无关,每份报告都重跑。
- 触发条件:生成个人长报告,且这台机器没有安装 PyJHora。装了参照引擎的开发机上,隔离用的子进程仍然要留着。
- 根因:子进程是为了把参照引擎隔离开,这是对的;父进程却没有先问「这台机器根本没装」。静态校验文件被挂在每份报告上,没有按文件变化复用。
- 修复:父进程先看一次能不能导入参照引擎,不能就不起子进程,返回和「子进程在没装时」相同的结果。解释器查找和静态校验各在进程内记一次;校验文件变了会重算。不缓存任何出生资料。装了参照引擎时仍走子进程。
- 验证:`tests/test_api_chart_subprocess_gate.py` 对照 `tests/fixtures/api-chart-subprocess-gate/`。生产上的收益大约是每份报告 0.5 到 0.9 秒、少掉那几个注定失败的进程,不是排查时在装了参照引擎的机器上看到的 5 秒。本机没有 PyJHora,实测见 `docs/testing/api-chart-subprocess-gate-ab-20261013.md`。
- 防复发:无参照引擎时测试断言不会调用 `subprocess.run`,返回值与固定样本一致。
- 相关记录:BUG-721、BUG-1265、BUG-1333。
- 复发自:无。
- 修复版本:分支 `codex/api-chart-subprocess-gate-20261013`。
## BUG-1335 | 个人长报告的中文出稿和溯源哈希重复劳动
- 状态:investigating
@@ -36197,3 +36257,50 @@
- 复发自:无。
- 修复版本:未修。分支 `codex/answer-effort-ab-20261013` 只记录。原先写在 BUG-1351,该号已被数据层占用;按本文件规则保留较早的 BUG-1351,本条改为 BUG-1354,接在 BUG-1353 之后。
## BUG-1355 | 对话、校正回合和报告生成没有按用户限流
- 状态:mitigated
- 首次发现 / 最近更新:2026-10-14 / 2026-10-15
- 来源:issue #76。任务书 `docs/tasks/TASK-beta-capacity-20261014.md`。
- 现象:一个人可以连续打对话、校正回合或报告生成。报告只有每天 5 份的上限,没有按小时的上限。对话和校正没有按这个人计数。
- 触发条件:已登录后调用 `POST /api/consult`、`POST /api/rectification/agent` 或 `POST /api/reports`。开案 `POST /api/rectification/cases/open` 不计数。
- 根因:`requestRateLimits` 只覆盖地点搜索、时区、今日星语、星历、合盘和会话写入。重计算闸是全进程共享的,挡不住同一个人把模型调用打满。
- 修复:对话 10 次/分钟且 120 次/小时,校正回合 5 次/分钟且 40 次/小时,报告生成另加 3 次/小时(每天 5 份仍在)。两档都还有余量才放行;拒绝时不写入任何一档,也不预留点数。响应是 429、`code: rate_limited` 和 `Retry-After`。
- 验证:`frontend/tests/request-rate-limit.test.ts`;报告核心测试「hourly rate limit returns 429 and does not reserve」。
- 防复发:上述测试锁住桶的边界、拒绝时不记入另一档,以及三条路由在预留点数之前判断。开案路由源码不得调用这组限流。
- 相关记录:BUG-723(校正客户端已有的退避,本单不改)、BUG-1302(预热仍不占重计算名额)、BUG-1357(打回后补上的两条旧检查)。
- 复发自:无
- 修复版本:分支 `codex/beta-capacity-20261014`。
## BUG-1356 | 重计算满了立刻拒绝,对话也不再试
- 状态:mitigated
- 首次发现 / 最近更新:2026-10-14 / 2026-10-15
- 来源:issue #76。任务书 `docs/tasks/TASK-beta-capacity-20261014.md`。
- 现象:同时只有 2 个重计算名额。第 3 个请求马上收到 429。普通对话把这次 429 当成这一轮失败,不按 `Retry-After` 再试。
- 触发条件:已有 2 个请求占着 `/api/consultation_workflow` 或校正打分,又来一个对话取证。
- 根因:闸门用非阻塞信号量,满了就抛 `HeavyComputeBusy`。对话侧把 HTTP 429 记成 `workflow_rate_limited` 后直接失败。
- 修复:名额满时最多再等 8 秒,同时等待的人最多 4 个。等不到或等待的人已经满,仍抛 `HeavyComputeBusy`,`Retry-After` 仍是原来的秒数。对话对这种 429 最多再试 2 次,几次等待加起来不超过 15 秒。等待时进度行写「人有点多,排一下队」。仍失败则沿用原来的取消预留退点,只走一次,并告诉用户这一轮没排上、点数已退回。并发上限仍是 2。
- 验证:`tests/test_api_heavy_compute_gate.py` 的排队拿到名额、等待者满立刻拒绝、超时仍 429;`frontend/tests/consultation-workflow-retry.test.ts`。
- 防复发:上述测试。健康检查仍不进闸。校正客户端原有的退避(BUG-723)没有改。
- 相关记录:BUG-717、BUG-723、BUG-1302、BUG-1345、BUG-1355、BUG-1357。
- 复发自:无
- 修复版本:分支 `codex/beta-capacity-20261014`。
## BUG-1357 | 限流改完后两条旧检查没跟上
- 状态:resolved
- 首次发现:2026-10-15
- 最近更新:2026-10-15
- 影响面:`frontend/tests/consultation-voice-contract.test.ts`;`frontend/tests/personal-report-api.test.ts` 与 `tests/test_repo_privacy_markers.py` 的行号白名单。没有用户界面变化。
- 用户现象:同一个人问太密、校正太密、报告太多会先停一下,人多时会先排一小会儿。这些都对。全量检查多出 2 条失败,这一版上不了线。
- 触发条件:验收方对 `a54f75e5` 跑前端全量和 Python 全量。
- 根因:对话路由把 `runConsultationWorkflow(toolInput` 包进排队重试,源码合同仍要求 `await` 紧贴函数名。报告的小时限流测试插在文件中部,把隐私白名单按行号锁定的三行各下移 13 行。被保护的行为都还在,不是泄漏。上一轮只跑了定向测试,这两条不在那一组里。
- 修复:正则仍锁定 `runConsultationWorkflow(toolInput`,不再要求 `await` 紧贴。小时限流测试整段挪到文件末尾,断言原文不变,`SAFE_LINES` 不改。业务代码不改。
- 验证:`docs/tasks/PROGRESS-beta-capacity-fix-20261015.md`。语音合同、报告接口、镜像不引用测试共 75 通过、0 失败;隐私扫描 62 通过、0 失败;tsc 退出码 0;lint 0 错误、123 条原有警告。
- 防复发:改这三处路由的语句前,在 `frontend/tests` 和 `tests` 里 grep 被改写的原文。新测试不要插在 `SAFE_LINES` 锁定行的上面。
- 相关记录:BUG-933、BUG-1014、BUG-1355、BUG-1356
- 复发自:BUG-933(源码合同没在同一轮跟上)
- 修复版本:分支 `codex/beta-capacity-20261014`。
+21
View File
@@ -0,0 +1,21 @@
# 内测邀请码(用兑换码)
后台路径:`/admin/codes`(兑换码)。需要管理权限。不要把真实兑换码写进仓库、进度或聊天记录。完整码只在生成成功的那一次弹窗里出现,关掉就不能再看,列表里只剩掩码。
## 批量生成
1. 打开兑换码页,点「批量生成」。
2. 「每个点数」是这一批每个码到账的点数。表单初始值是 10,可以改。内测每码给多少点由当次生成的人填写,产品以后可以改这个数。
3. 「数量」一次 1 到 100 个。同一批的点数、到期时间、备注相同。
4. 到期时间可以空着。备注用来区分这一批,例如内测第几批,不要写别人的个人资料。
5. 生成后立刻把完整码交给对应的人。页面会提示关闭后无法再看。
## 查谁用了
列表有掩码、点数、状态、到期、备注、兑换账户、兑换时间、撤销时间。状态可以筛:可用、已过期、已兑换、已撤销。兑换账户和兑换时间就是谁用过。
## 作废
还没兑换、也没撤销的码,行上有「撤销」。撤销后不能再兑。已经兑换的不能从这里追回已经到账的点数。
支付在内测期间保持关闭。这页已经能批量生成、查看使用者和作废,本单不另加功能。
@@ -0,0 +1,90 @@
# PROGRESS · 合盘排盘与报告多余子进程(2026-10-15)
- 分支:`codex/api-chart-subprocess-gate-20261013`
- 实际基线:`origin/staging` `9274491681f6fcacbc87e3797441d2c5221de874`(任务书写的 `77057a819` 更早)
- `scripts/jyotish_api_server.py`:开工 10944 行,做完 10940 行
- BUG-1333、BUG-1334。`docs/BUG_HISTORY.md` 后半是同一份历史的副本,两处都补上了。
## 做了什么
1. 合盘两次排盘都带 `skip_vedastro_main_entry_overview: true`,不改 `synastryBirthPayload`。两张盘和婚恋分支的两张九分盘都改成同时发出。任一失败仍走原来的 `catch`。
2. 前端 `app/api` 和 `lib` 里发往 `/api/chart` 的请求体必须带这个开关。白名单是空的。
3. 父进程先看一次能不能导入 PyJHora。不能就不起年度回放和 Tajika 的子进程,返回和「子进程在没装时」相同的结果。样本在 `tests/fixtures/api-chart-subprocess-gate/`,来自本机真实子进程(当时 `find_spec("jhora")` 为 `None`)。装了之后仍走子进程。
4. 解释器查找(含「没找到」)和静态 oracle 校验在进程内各记一次。校验按文件路径、mtime、大小作键。不缓存出生资料。
5. 开关 `JYOTISH_GATE_CHART_OVERVIEW` 默认关。只有把它设为 `1` 时,不带 skip、也不是异步排队的 `/api/chart` 才占一个重计算名额。`/api/chart` 没有加进 `HEAVY_COMPUTE_PATHS`。预热咨询不占名额的判断挪进了闸门模块。
6. 名额占满时打一行日志:被拒路径、当前占用者路径、已占秒数。不写请求正文。
## 自测
| 检查 | 命令 | 结果 |
| --- | --- | --- |
| 闸门、增长护栏、本单新测试 | `.venv\Scripts\python.exe -m pytest tests/test_api_chart_subprocess_gate.py tests/test_api_heavy_compute_gate.py tests/test_api_server_growth_contract.py -q` | 21 passed,退出码 0 |
| 报告三份既有测试 | 同上解释器 `-m pytest tests/test_report_reader_main.py tests/test_report_longform_parity.py tests/test_report_density_facts.py -q` | 27 passed,退出码 0 |
| 前端本单与相关单文件 | 在 `frontend` 下 `node node_modules/tsx/dist/cli.mjs --test tests/synastry-chart-skip.test.ts tests/chart-request-skip-contract.test.ts tests/chart-view-route.test.ts tests/ephemeris-route.test.ts` | 22 pass,0 fail |
| 类型检查 | `frontend\node_modules\.bin\tsc --noEmit -p frontend` | 退出码 0 |
| lint | 在 `frontend` 下 `npm run lint` | 退出码 0,0 error,123 条原有 warning |
| A/B | `.venv\Scripts\python.exe scripts\research\api_chart_subprocess_ab.py` | 退出码 0。数字见下 |
`grep` 这三个名字在 `tests/` 里只命中本单新测试:`annual_pyjhora_replay`、`_oracle_progress_snapshot`、`_resolve_external_pyjhora_python`。
## A/B 数字
合盘:串行不带 skip 中位 8.0596 秒、2 次 `subprocess.run`;并发带 skip 中位 0.0166 秒、0 次。盘面摘要逐字节一致。耗时约为对照的 0.2%。
报告:强制仍起子进程中位 16.5322 秒、7 次;短路后 15.3235 秒、1 次。去掉易变字段后逐字节一致。少掉的 6 次是年度回放和 Tajika。
名额模拟(种子 64,一分钟,未改规则):
| 方案 | 咨询被拒 | 报告被拒 | 校正被拒 | 最高并发 | 碰到不得加池的那条 |
| --- | --- | --- | --- | --- | --- |
| O1 现状 | 40% | 0% | 17% | 2 | 否 |
| O2 咨询单独小池 | 0% | 0% | 17% | 3 | 是 |
| O3 咨询最多再等 1.5 秒 | 30% | 0% | 33% | 2 | 否 |
| O4 重任务最多占 1 个 | 0% | 0% | 67% | 2 | 否 |
全文在 `docs/testing/api-chart-subprocess-gate-ab-20261013.md`。
## 改既有断言
无。原值 / 新值 / 原因:没有改旧断言。
## 偏离
- `/api/health` 没有加 `heavy_compute` 字段。状态函数 `heavy_compute_status()` 在闸门模块里,接到健康检查会让主文件变长。任务书允许砍掉这一项。
- 星盘名额开关默认关。打开会让不带 skip 的外部调用可能拿到 429,任务书要求先有 A/B 再提议,本单不打开。
- O2、O3、O4 没有实现。
- 报告对照不是 staging 源码另导入,而是本分支把「已安装」强制打开。静态校验缓存两边都在,所以耗时差只反映那 6 个子进程。
- 合盘子进程是 2 → 0,不是任务书举例的 4 → 0。这台机器两张盘合计只数到 2 次 `subprocess.run`。
- 没有「装了 PyJHora」的计时。父进程强制当成已安装时,子进程仍会因为没装而失败,这一组用来代表「不短路就要付启动开销」。
## 环境缺口
- 本机 `.venv`:`sys.executable` 是工作树的 `python.exe`;`shutil.which("python3")` 是 Windows 应用商店占位;`python3.14` 没有;`find_spec("jhora")` 是 `None`。
- Docker 在,没有本地 `jyotisha-api` 镜像,镜像里的解释器路径没有核实。
- 生产装了 VedAstro、并且开了外网时,不带 skip 的排盘真实耗时本机测不到。
- 名额被占满后连带拒绝的真实频率,要等日志上了 staging 再看。
## 需拍板
轻重任务要不要分开限流,本单不改。模拟里,让咨询少被拒,校正就会多被拒;给咨询单独开池会让同时在跑的个数超过 2。要改规则需要另立一单。
## 默认值
- `JYOTISH_GATE_CHART_OVERVIEW` 默认关。
- 合盘 skip 和报告短路没有开关。
## 2026-10-15 打回修复
合盘路由的合同跟上了新写法:仍锁本人和对方各请求一次排盘,并锁住关闭外部概览。对照脚本默认写到系统临时目录,只有带 `--out` 才写指定文件,不再覆盖已入库的记录。业务代码没改。名额配额仍不实现。没有新 bug 编号,补在 BUG-1333 的两处副本上。
| 检查 | 命令 | 结果 |
| --- | --- | --- |
| 用户数据合同 | `.venv\Scripts\python.exe -m pytest tests/test_supabase_user_data_contract.py -q` | 8 passed,退出码 0 |
| 脚本默认路径 | 调用 `resolve_record_path` | 默认不在仓库内,也不是已入库的 json;`--out` 指向传入路径 |
| 源码文本检索 | `tests` 与 `frontend/tests` 里的合盘路由、`postPython("/api/chart"`、`postPython("/api/varga_full"` | 只有这一条合同在断言旧写法,已改。其它命中是在读路由文件,没有断言旧的两次调用。分盘前缀仍匹配,没改 |
| 类型检查 | 在 `frontend` 下 `node_modules\.bin\tsc --noEmit -p .` | 退出码 0 |
| lint | 在 `frontend` 下 `npm run lint` | 退出码 0,0 error,123 条原有 warning |
原值:`postPython("/api/chart", selfPayload)` 与 `postPython("/api/chart", partnerPayload)`。新值:两次调用都展开对应 payload,并带 `skip_vedastro_main_entry_overview: true`。原因:合盘改为同时排两张盘并关掉用不上的外部概览,合同要锁住这两次请求和这个开关,不放宽。
偏离:先合并了 `origin/staging` `1de3cc307be9134a3b8d1d606c738c1468414118`。相对打回 SHA `336286e4`,`frontend/src` 里多出来的是数据访问层那一单,不是这次修复改的。技能目录下的同名脚本没改,它的输出根不是仓库的 `docs/testing`。
@@ -0,0 +1,43 @@
# 进度 · 内测容量(2026-10-15)
- 分支:`codex/beta-capacity-20261014`
- 基线:开工时 `origin/staging` `ef49257b734678ff944a840076347baa02d0323c`
- 任务书:`docs/tasks/TASK-beta-capacity-20261014.md`
- 前置:`origin/codex/node-runtime-resources-20261013` 与 `origin/codex/rect-scoring-perf-20261013` 都是该 staging 的祖先(`merge-base --is-ancestor` 退出码 0)。限流文件里已有过期清理。
## 做了什么
- T1:对话 10 次/分钟 + 120 次/小时,校正回合 5 次/分钟 + 40 次/小时,报告 3 次/小时(每天 5 份不变)。拒绝发生在预留点数之前,429 + `rate_limited` + `Retry-After`。开案路由不计数。
- T2:重计算并发仍是 2。满了最多等 8 秒,同时等待最多 4 个;等不到或等待已满仍是 429,`Retry-After` 不变。对话对这种 429 最多再试 2 次,等待合计不超过 15 秒。进度行「人有点多,排一下队」,失败句「人有点多,这一轮没排上,点数已经退回。」退点仍走原来的取消预留,代理路径用已有的一次结算。
- T3:后台已能批量生成(1–100 个、每码点数自填、初始 10)、按兑换账户查看、撤销未使用的码。说明在 `docs/operations/beta-invite-codes.md`。没有扩功能。
- T4:脚本 `frontend/scripts/research/beta-load-smoke.mts`。不从 `frontend/tests` 导入。本机没有跑。
- T5:BUG-1355、BUG-1356,CHANGELOG、VOICE、DESIGN、真机两条。
## 断言
旧测试仍要求「名额被占着就拒绝」,没有改拒绝码和 `Retry-After` 的期望值。
| 位置 | 原值 | 新值 | 原因 |
| --- | --- | --- | --- |
| `tests/test_api_heavy_compute_gate.py` 立刻拒绝的夹具和单测 | `reset()` / `reset(limit=1, retry_after_seconds=3)` | 增加 `wait_seconds=0` | 默认改为最多等 8 秒。这些测试继续锁 429 和 `Retry-After`,排队另有新测试 |
| `tests/test_api_chart_subprocess_gate.py` 两处占着名额再请求 | `reset(limit=1)` | `reset(limit=1, wait_seconds=0)` | 同上。仍锁同步排盘被拒绝、日志不出现请求正文 |
## 偏离与环境缺口
- T0:仓库文档里的目标规格,staging 与生产都是 2 vCPU / 4 GB,代码默认并发 2。这台 Windows 没有 staging 的登录壳,没有采到机器上的实际核数、内存和 `JYOTISH_HEAVY_COMPUTE_CONCURRENCY`。不能把文档规格当成已经登主机核对过。
- T4:没有 `STAGING_TEST_OTP`,没有对 staging 或生产发压测。改前、改后的并发数字都没有。服务器 CPU / 内存同样没有采样。脚本留给验收方在有测试验证码的环境重跑。20 并发档没有跑。
- 旧版对话(不走代理运行时)会重试,但进度行要等回答流开始才有,所以那条路径等待时看不到「排一下队」。现在的对话走代理运行时,进度行会换字。任务书允许把排队文案收成只多等。
- 没有改数据库,没跑 `test:db`。没有跑前端全量 `npm test`,没有跑 Python 快速门。
- `deploy/README.md` 有一句仍写「满了不排队」。改成了最多等 8 秒、等待不超过 4 个,避免操作说明和代码相反。没有改部署配置。
## 自测
| 检查 | 命令 | 结果数字 |
| --- | --- | --- |
| Python 闸门 | `.venv\Scripts\python.exe -m pytest tests/test_api_heavy_compute_gate.py tests/test_api_chart_subprocess_gate.py -q` | 17 通过(8 + 9),0 失败 |
| 前端相关测试 | `node node_modules/tsx/dist/cli.mjs --test tests/request-rate-limit.test.ts tests/consultation-workflow-retry.test.ts tests/personal-report-api.test.ts tests/runtime-sources-no-test-imports.test.ts` | 76 通过,0 失败 |
| 类型 | `frontend\node_modules\.bin\tsc --noEmit -p .` | 退出码 0 |
| 前端 lint | `npm run lint` | 0 错误,123 条原有警告 |
| 压测 | 未执行 `frontend/scripts/research/beta-load-smoke.mts` | 本机没有 `STAGING_TEST_OTP`。没有改前/改后并发数字,没有服务器 CPU/内存 |
没有改数据库,没跑 `test:db`。没有跑前端全量 `npm test`,没有跑 Python 快速门。
@@ -0,0 +1,46 @@
# 进度 · 内测容量修复(2026-10-15,第 1 次打回)
- 分支:`codex/beta-capacity-20261014`
- 被验 SHA:`a54f75e5e33775d5dc91f3058f5505a56b498bf2`
- 修复单:`docs/tasks/TASK-beta-capacity-fix-20261015.md`
- 原任务书:`docs/tasks/TASK-beta-capacity-20261014.md`
- 开工时本分支最大 Bug 编号:BUG-1356。本轮新增 BUG-1357。BUG-1351、BUG-1352、BUG-1353 已在本分支文件里,没有重用。
## 做了什么
- 先把最新 `origin/staging` 合并进本分支(不改写已推送的 `a54f75e5`)。合并只带进修复单和任务索引的一行。
- 语音合同:仍锁「工作流调用的入参是 `toolInput`」,以及用户轮不塞工具 JSON。不再要求 `await` 紧贴函数名。
- 隐私白名单:采用修复单推荐的做法 a。把「hourly rate limit returns 429」整段挪到 `personal-report-api.test.ts` 末尾,断言原文不变。`SAFE_LINES` 不改。三行回到 832、860、890,且整行与白名单一致。
- 业务代码不改。
## 断言
| 位置 | 原值 | 新值 | 原因 |
| --- | --- | --- | --- |
| `consultation-voice-contract.test.ts`「consult user turn no longer dumps tool JSON」 | `/await runConsultationWorkflow\(toolInput/` | `/runConsultationWorkflow\(toolInput/` | 调用包进 `retryOnConsultationRateLimit`,`await` 不再紧贴函数名。仍锁入参是 `toolInput` |
| `personal-report-api.test.ts` 小时限流测试的位置 | 插在「429 at the daily limit」和「502 model_unavailable」之间 | 文件末尾,断言原文不变 | 插在中部让三行白名单各下移 13 行。挪走后行号复原,不改 `SAFE_LINES` |
## 源码文本再搜一遍
在 `frontend/tests` 和 `tests` 里核对三条路由被改写的语句:
- 只有语音合同把 `await runConsultationWorkflow(toolInput` 写成必须匹配。已按上表改。
- `consultation-workflow-contract.test.ts` 是在代理分支里断言**不**出现 `await runConsultationWorkflow`,与这次包装无关,没改。
- 「暂时无法生成解读」仍在非排队失败的分支里,`consult-first-frame-and-pacing-20260928.test.tsx` 继续对得上,没改。
- `toNextResponse` 多了响应头,没有测试按旧函数签名做源码断言。
- Python 侧没有按这三处被改写语句做源码文本断言。
## 偏离与环境缺口
- 没有改数据库,没跑 `test:db`。没有跑前端全量 `npm test`,没有跑 Python 快速门。
- 并发压测 `frontend/scripts/research/beta-load-smoke.mts` 未跑。没有测试验证码,修复单也写明本机和执行侧都连不上测试服务器,不要登预发。没有改前/改后并发数字,没有服务器占用采样。
## 自测
| 检查 | 命令 | 结果数字 |
| --- | --- | --- |
| 前端相关测试 | `node node_modules/tsx/dist/cli.mjs --test tests/consultation-voice-contract.test.ts tests/runtime-sources-no-test-imports.test.ts tests/personal-report-api.test.ts` | 75 通过,0 失败,0 取消 |
| 隐私扫描 | `.venv\Scripts\python.exe -m pytest tests/test_repo_privacy_markers.py -q` | 62 通过,0 失败 |
| 类型检查 | `frontend\node_modules\.bin\tsc --noEmit -p .` | 退出码 0 |
| 前端 lint | `npm run lint` | 0 错误,123 条原有警告 |
| 并发压测 | 未跑 `beta-load-smoke.mts` | 没有测试验证码,没有登预发 |
+3 -2
View File
@@ -340,6 +340,7 @@
| `TASK-release-path-production-fix2-20261014.md` | `PROGRESS-release-path-production-fix2-20261014.md` | **#74 修复单 2(产品选 A)**:恢复点要带属主与权限,旧格式不得回退;演练按备份里的授权比对,并在真实 PostgreSQL 17 上跑通一次 | 待验收 | `codex/release-path-production-20261014` |
| `TASK-beta-safety-net-20261014.md` | — | **内测安全网**:每日备份 + 恢复演练、探活告警、本地错误汇总、日志轮转、按 requestId 查日志。**等发布通路单合入、#58 合入后开工** | 待前置 | — |
| `TASK-beta-capacity-20261014.md` | — | **内测容量**:按用户限流、重计算短排队与对话重试、兑换码当邀请码、staging 10/20 并发压测。**等 #58 合入后开工,压测等 #57 合入** | 待前置 | — |
| `TASK-beta-capacity-fix-20261015.md` | — | **#76 第 1 次打回**:`a54f75e5` 限流与短排队逻辑通过;源码文本断言未跟改、插入新测试使隐私白名单行号错位 | 待领取 | `codex/beta-capacity-20261014` |
| `TASK-special-lagnas-native-20261008.md` | `PROGRESS-special-lagnas-native-20261008.md` | **特殊上升点原生计算(线上缺陷)**:sync6 U6 运行时调 PyJHora,生产镜像没有 PyJHora → staging 上 Hora/Ghati/Sree/Indu/Pranapada 全 blocked;原生实现、PyJHora 只作冻结对照、快速门禁止运行时 import jhora;最先做 | 已验收(已部署 staging `795b8845`,BUG-1273) | `codex/special-lagnas-native-20261008`(未上线;部署以 `/api/health` 为准) |
| `TASK-ephemeris-nutation-20261008.md` | `PROGRESS-ephemeris-nutation-20261008.md` | **恒星黄道少扣章动(BUG-1275)**:行星 = 含章动热带视位置 − 平岁差,全体偏 Δψ(≤17″),大运日期差 2–3 天;改标准算法、PyJHora 冻结对照、scoring-12、v5 重跑 | 已验收(已部署 staging `db0134f9`,BUG-1275) | `codex/ephemeris-nutation-20261008`(未推送,未部署) |
| `TASK-sync6-gate-red-20261007.md` | — | **sync6 后门禁仍红**:run 3213 快速门已过,「Frontend and database tests」开跑 20 秒后日志中断、15 分钟后判失败,后续步骤 failure 0 秒;疑运行机掉线;先 Re-run,红则本机带 Docker 跑 `CI=1 npm test` 对照 `0ed45a4b` | 已完成(Re-run 通过,BUG-1271,部署 `2df6ee72`) | — |
@@ -483,8 +484,8 @@
| `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` |
| `TASK-api-chart-subprocess-gate-fix-20261015.md` | — | **#64 第 1 次打回**:`336286e4` 内容通过(合盘 1.19→0.015 s、正文逐字节一致);Python 合同按源码检查合盘路由旧写法未跟改;A/B 脚本默认覆盖入库记录 | 待领取 | `codex/api-chart-subprocess-gate-20261013` |
| `TASK-api-chart-subprocess-gate-20261013.md` | `PROGRESS-api-chart-subprocess-gate-20261013.md` | 合盘不带 skip 串行排盘、报告无效子进程、名额轻重混用(只模拟)(BUG-1333~1334,A/B) | 第 1 次打回,见修复单 | `codex/api-chart-subprocess-gate-20261013` |
| `TASK-api-chart-subprocess-gate-fix-20261015.md` | `PROGRESS-api-chart-subprocess-gate-20261013.md` | **#64 第 1 次打回**:`336286e4` 内容通过(合盘 1.19→0.015 s、正文逐字节一致);Python 合同按源码检查合盘路由旧写法未跟改;A/B 脚本默认覆盖入库记录 | 待验收 | `codex/api-chart-subprocess-gate-20261013` |
| `TASK-answer-effort-ab-20261013.md` | `PROGRESS-answer-effort-ab-20261013.md` | 答题步推理强度开关 + A/B(需模型 key) | 待验收 | `codex/answer-effort-ab-20261013` |
| `TASK-answer-effort-ab-fix-20261015.md` | `PROGRESS-answer-effort-ab-fix-20261015.md` | **#65 第 1 次打回**:`ab7f66be` 开关逻辑通过(默认请求体不变);研究脚本从 `frontend/tests` 读文件(BUG-1140),已挪到 `frontend/tests/answer-effort-probe.mts` | 待验收 | `codex/answer-effort-ab-20261013` |
| `TASK-consult-route-split-20261013.md` | — | consult POST 拆分第一阶段:开流前加载 + 计费结算 | 待领取(排队:consult-prep-parallel、answer-effort-ab) | `codex/consult-route-split-20261013` |
@@ -0,0 +1,37 @@
# TASK · 内测容量修复单(issue #76,第 1 次打回)
- 日期:2026-10-15
- 原任务书:`docs/tasks/TASK-beta-capacity-20261014.md`
- 被验 SHA:`a54f75e5e33775d5dc91f3058f5505a56b498bf2`(`codex/beta-capacity-20261014`)
- 执行方:Grok,同一分支续做(先 merge 最新 `origin/staging`,不改写已推历史)
## 1. 验收结论:逻辑通过,全量多出 2 条失败
| 检查 | 结果 | 证据 |
| --- | --- | --- |
| 按用户限流 | 通过 | 问、校正每轮、报告每小时各有分钟 / 小时两档;任一档满就整体拒,拒绝不记账;都在扣点之前 |
| 重计算短排队 | 通过 | 并发仍是 2(BUG-717);最多 4 个等待、每个最多 8 秒,等不到仍 429(BUG-720 有上限) |
| 前端排队重试 | 通过 | 最多 2 次、总等待 ≤ 15 秒;最终失败走原取消路径 |
| tsc / lint | 通过 | — |
| **前端全量** | **未通过:新增 1 条** | `frontend/tests/consultation-voice-contract.test.ts`「consult user turn no longer dumps tool JSON that the model would parrot」:`/await runConsultationWorkflow\(toolInput/` 不再匹配——`route.ts` 改成了 `await retryOnConsultationRateLimit({ run: () => runConsultationWorkflow(toolInput, …) })` |
| **Python 全量** | **未通过:新增 1 条** | `tests/test_repo_privacy_markers.py::test_tracked_repository_has_no_privacy_markers`:`HOME_PREFIX frontend/tests/personal-report-api.test.ts lines=845,873,903` |
第 2 条不是真的隐私泄漏:这三行是已有的、经审过的虚构路径样本,`tests/test_repo_privacy_markers.py` 的 `SAFE_LINES` 按「文件 + 行号 + 整行」锁定它们。本单在它们上面插入了一条 13 行的新测试,行号整体下移,白名单就对不上了。
## 2. 修法要求
1. `consultation-voice-contract.test.ts`:按 §7.3 写三栏改断言,**意图不放宽**——仍要锁住「工作流调用的入参是 `toolInput`」与该测试原本关心的那段(用户轮不再塞工具 JSON);新正则匹配 `runConsultationWorkflow(toolInput` 即可,不得去掉对 `toolInput` 的锁定。
2. 隐私白名单二选一(推荐 a):
a. 把新加的「hourly rate limit returns 429」测试挪到文件末尾,让那三行回到原行号,`SAFE_LINES` 不动;
b. 更新 `SAFE_LINES` 里这三行的行号,按三栏写原值 / 新值 / 原因。
3. 再 grep 一遍 `route.ts` / `rectification/agent/route.ts` / `reports/route.ts` 被改写语句在 `frontend/tests`、`tests` 里的源码文本断言(§7.8)。
4. 交回前本机跑:`npx tsx --test tests/consultation-voice-contract.test.ts tests/runtime-sources-no-test-imports.test.ts` 与 `python -m pytest tests/test_repo_privacy_markers.py -q`。
5. 业务代码不改。
## 3. 环境缺口(不阻断)
`frontend/scripts/research/beta-load-smoke.mts` 并发压测未跑(无测试验证码、无预发主机)。合入部署后由 Claude 在验收清单里列给产品负责人,或另开一单在 staging 跑。
## 4. 让步顺序
无。第 1 次打回。
@@ -0,0 +1,53 @@
{
"charts": {
"serial_seconds_median": 8.0596,
"concurrent_skip_seconds_median": 0.0166,
"serial_subprocesses_median": 2,
"concurrent_skip_subprocesses_median": 0,
"summary_bytes_match": true
},
"report_forced_subprocess": {
"seconds_median": 16.5322,
"subprocesses_median": 7
},
"report_short_circuit": {
"seconds_median": 15.3235,
"subprocesses_median": 1
},
"report_bytes_match": true,
"schemes": [
{
"scheme": "O1",
"consult_429": "0.40",
"report_429": "0.00",
"score_429": "0.17",
"peak": 2,
"touches_bug_717": false
},
{
"scheme": "O2",
"consult_429": "0.00",
"report_429": "0.00",
"score_429": "0.17",
"peak": 3,
"touches_bug_717": true
},
{
"scheme": "O3",
"consult_429": "0.30",
"report_429": "0.00",
"score_429": "0.33",
"peak": 2,
"touches_bug_717": false
},
{
"scheme": "O4",
"consult_429": "0.00",
"report_429": "0.00",
"score_429": "0.67",
"peak": 2,
"touches_bug_717": false
}
],
"jhora_installed": false
}
@@ -0,0 +1,40 @@
# 合盘排盘与报告子进程 A/B(2026-10-15)
Windows 本机,工作树 `.venv` 的 `python.exe`。`importlib.util.find_spec("jhora")` 是 `None`。`python3` 指向 Windows 应用商店占位程序,`python3.14` 不存在。Docker 守护进程在,本机没有 `jyotisha-api` 镜像,所以没有进镜像再探一次。本机没有 VedAstro SDK,外网关闭(`VEDASTRO_ENABLE_NETWORK=0`)。没有「真 SDK」一组。
脚本:`scripts/research/api_chart_subprocess_ab.py`。每组 3 次,取中位数。盘是任务书里的虚构盘(1990-01-01 12:00 与 1992-01-02 12:00,新德里)。星盘缓存 TTL 设为 0。易变字段在比对前丢掉:`generated_at`、`created_at`、`timestamp`、`elapsed_seconds`、`elapsed_ms`、`duration_ms`、`request_id`。原始数字在同目录 `api-chart-subprocess-gate-ab-20261013.json`。
对照不是另一次 `git show` 出 staging 再导入。星盘对照是本分支的 `_compute_chart_sync`:不带 skip、两张盘串行。报告对照是本分支把「参照引擎已安装」强制打开,从而仍会起子进程;静态校验缓存和解释器查找缓存两边都开着。所以报告耗时差只含那 6 个注定失败的回放进程,不含静态校验少跑的那一次。静态校验「连调两次只跑一次」由单测覆盖。
## 合盘两张盘
| | 中位耗时 | 中位 `subprocess.run` 次数 | 盘面摘要 |
| --- | --- | --- | --- |
| 对照:不带 skip,串行 | 8.0596 秒 | 2 | — |
| 本单:带 skip,并发 | 0.0166 秒 | 0 | 与对照逐字节一致 |
0.0166 / 8.0596 ≈ 0.2%,低于「不超过对照 10%」的线。子进程从 2 降到 0。任务书写的「4 → 0」是按每张盘 2 个子进程、两张合计 4 个来估的。这台机器上两张盘合计只数到 2 次 `subprocess.run`。不带 skip 的 8 秒里,大部分是没装 SDK 时子进程空等,不是盘面计算本身。
## 个人长报告(完整数据版,中英,带事实表)
| | 中位耗时 | 中位 `subprocess.run` 次数 | 中英正文与事实表 |
| --- | --- | --- | --- |
| 对照:强制仍起回放子进程 | 16.5322 秒 | 7 | — |
| 本单:没装参照引擎就不起 | 15.3235 秒 | 1 | 去掉易变字段后逐字节一致 |
少 6 次,对应年度回放 3 次和 Tajika 3 次。剩下的 1 次两边都有,短路没有把它去掉。子进程数 1,低于「模拟生产不超过 3」的线。耗时 15.32 秒,不比对照的 16.53 秒差。本机没装 PyJHora,没有再跑「装了参照引擎」的一组;那种环境仍必须走子进程,单测用桩覆盖。
生产上的收益仍按任务书:大约每份报告 0.5 到 0.9 秒,不是装了参照引擎的机器上看到的约 5 秒。
## 名额方案模拟(没改规则)
固定一分钟:2 份报告(每次 13 秒)、20 次咨询(每次 0.2 秒)、6 次校正打分(每次 8 秒)。随机种子 64。并发上限按现状 2。O2、O3、O4 只在脚本里模拟。
| 方案 | 咨询被拒 | 报告被拒 | 校正被拒 | 同时在跑的最高个数 | 是否碰到「不得加池、不得提高并发」 |
| --- | --- | --- | --- | --- | --- |
| O1 现状 | 40% | 0% | 17% | 2 | 否 |
| O2 咨询单独小池 | 0% | 0% | 17% | 3 | 是,本单不实现 |
| O3 咨询最多再等 1.5 秒 | 30% | 0% | 33% | 2 | 否,本单不实现 |
| O4 重任务最多占 1 个名额 | 0% | 0% | 67% | 2 | 否,本单不实现 |
这是一分钟的草图,不是 staging 上的真实频率。咨询少被拒的时候,校正会多被拒(O3、O4)。O2 咨询不再被拒,但同时在跑的个数到了 3,越过了现在的上限。
+6
View File
@@ -0,0 +1,6 @@
# 内测容量 · 真机两条(2026-10-14)
用测试账号。不要把出生资料、姓名、邮箱写进记录。
1. 快速连发触发限流。登录后在一分钟里连续发送超过 10 次普通对话。预期:出现「这一会儿问得有点密,先停一下。」这一轮不扣点。等提示里的时间过去再发一次,可以正常问。
2. 排队提示出现与消失。需要两个人同时占着计算(或用压测脚本打满两个名额)时,第三个人发普通对话。预期:进度行先出现「人有点多,排一下队」,排上之后这句消失,回到正在计算。若一直排不上,出现「人有点多,这一轮没排上,点数已经退回。」余额与发送前一致。
+1 -1
View File
@@ -907,7 +907,7 @@ or user IDs.
Agent 的 live 标记只有 `InlineSpinner` 一种。曾经并存的 canvas 小球(`thinking-orbs`)已移除,不得再引入第二种 live 标记。
**普通对话发出即有反馈(2026-09-28,TASK-consult-first-frame-and-pacing,BUG-1074)。** 按下发送的那一帧,回答行就带着步骤栏,唯一一行是排队行「收到,正在看你的问题…」(`QUEUED_TIMELINE_ROW`,行内 `InlineSpinner` + shimmer,summary 写「正在分析」)。服务端流的第一个字节也是这句(`activity` 事件,phase `received`),所以那一帧到达时行不换字。不再有「分类结果回来之前整行静默」:只有服务端判定为寒暄的回复(`run.completed` 带 `responseKind: "smalltalk"`)才把步骤栏收掉、只留一句话,寒暄前会闪约一秒这一行,属于流式生成中的正常形态。首事件之后的行照旧由服务端事件推动;不新增第二种 live 标记,不加 spinner 面板或骨架。
**普通对话发出即有反馈(2026-09-28,TASK-consult-first-frame-and-pacing,BUG-1074)。** 按下发送的那一帧,回答行就带着步骤栏,唯一一行是排队行「收到,正在看你的问题…」(`QUEUED_TIMELINE_ROW`,行内 `InlineSpinner` + shimmer,summary 写「正在分析」)。服务端流的第一个字节也是这句(`activity` 事件,phase `received`),所以那一帧到达时行不换字。不再有「分类结果回来之前整行静默」:只有服务端判定为寒暄的回复(`run.completed` 带 `responseKind: "smalltalk"`)才把步骤栏收掉、只留一句话,寒暄前会闪约一秒这一行,属于流式生成中的正常形态。首事件之后的行照旧由服务端事件推动;不新增第二种 live 标记,不加 spinner 面板或骨架。重计算名额满了时,计算那一行先换成「人有点多,排一下队」(`activity` phase `compute-queue`,仍是原来的 live 行),排上之后回到「正在计算本命盘…」。排不上则整轮失败,文案是「人有点多,这一轮没排上,点数已经退回。」
校正面的所有等待复用行内等待:进入前的 hydration 在揭幕之前完成,进入后唯一的等待形态是时间线 live 行(含「正在准备下一个问题…」这一条独立 live 行)。打字回答的 live 行从发出那一帧起就是阶段句「收到,正在对照你的档案…」,按服务端 `turn.progress` 换成「正在记下这件事…」「正在重新对照盘面…」「正在准备下一个问题…」,属于流式生成中(BUG-1047)。区间交付卡只挂在最新那条采用旁白下面,不得留在更早的采集/区分题下。有未答的采集或选择题时卡仍在,「更像这个」置灰并写「先答完上面这道,再选时间」,不得整卡消失。卡上至多三列并排,相同性格句只写一次,点「更像这个」即采用该列分钟,按钮显示「正在采用…」或「已采用」。采用过程中整张卡留在原处,不得因 `busy` 卸掉。采用后若还有前事核对,服务端写一条自己的采用消息(「已按 HH:MM 排盘…」),核对题的四个选项挂在这条消息下,进行中那一行由它替换,不挂到上一题的回复下(BUG-1146)。采用后前事核对结束走 `verified_idle`:一行收尾文案跟在卡片下面、与助手列对齐,没有 live 行、没有重载、没有采用状态条。卡片与右栏细则见 §11、§12。
+4
View File
@@ -480,3 +480,7 @@ Jyotisha 的可见文案是产品的一部分。正确性红线(真实性、
- 不写健康诊断、投资建议、法律意见;不写「一定」「必然」「注定」。
- 出生时间没确认时不用宫位,只按月亮所在星座写,caution 固定为「出生时间尚未确认,这里只看大方向,不依赖具体分钟。」
## 人多与问得太密(2026-10-15)
不写「系统繁忙」「请稍后重试」。计算要等空位时,进度行只写「人有点多,排一下队」。这一轮最终没排上:「人有点多,这一轮没排上,点数已经退回。」对话太密:「这一会儿问得有点密,先停一下。」校正回合太密:「这一会儿校正问得有点密,先停一下。」这一小时报告太多:「这一小时的报告已经够多了,先停一下。」
@@ -0,0 +1,277 @@
/**
* Staging-only smoke for 10 or 20 concurrent consultations.
*
* Not imported by frontend/src. Reads STAGING_TEST_OTP from the environment
* and never prints it. Refuses a production host. Does not sample server
* CPU or memory; that needs a staging shell this process does not have.
*
* node node_modules/tsx/dist/cli.mjs scripts/research/beta-load-smoke.mts --users 10
*/
const usersArg = process.argv.find((arg) => arg.startsWith("--users="));
const usersFlag = process.argv.indexOf("--users");
const requestedUsers = Number(usersArg ? usersArg.slice("--users=".length) : process.argv[usersFlag + 1]);
const baseUrl = (process.env.STAGING_BASE_URL || "https://staging.jyotisha.chat").replace(/\/$/, "");
const otp = process.env.STAGING_TEST_OTP || "";
const emailDomain = process.env.STAGING_TEST_EMAIL_DOMAIN || "example.test";
function refuse(message: string): never {
console.error(message);
process.exit(2);
}
let host = "";
try {
host = new URL(baseUrl).host;
} catch {
refuse("STAGING_BASE_URL is not a URL");
}
if (host === "jyotisha.chat" || host === "www.jyotisha.chat" || host === "admin.jyotisha.chat") {
refuse("refusing production host");
}
if (!/^\d{6}$/.test(otp)) refuse("STAGING_TEST_OTP is not set");
if (requestedUsers !== 10 && requestedUsers !== 20) refuse("pass --users 10 or --users 20");
if (!/^[a-z0-9.-]+$/.test(emailDomain)) refuse("STAGING_TEST_EMAIL_DOMAIN is not a plain domain");
type Sample = {
kind: "consult" | "rectification";
status: number;
firstByteMs: number | null;
totalMs: number;
failed: boolean;
http429: boolean;
http5xx: boolean;
};
const samples: Sample[] = [];
const CONSULT_QUEUE_FAILED = "人有点多,这一轮没排上,点数已经退回。";
function percentile(values: number[], ratio: number): number | null {
if (values.length === 0) return null;
const sorted = [...values].sort((a, b) => a - b);
const index = Math.min(sorted.length - 1, Math.max(0, Math.ceil(sorted.length * ratio) - 1));
return sorted[index] ?? null;
}
async function request(jar: Map<string, string>, path: string, init: RequestInit = {}): Promise<Response> {
const headers = new Headers(init.headers);
if (jar.size > 0) headers.set("cookie", [...jar].map(([key, value]) => `${key}=${value}`).join("; "));
const response = await fetch(`${baseUrl}${path}`, { ...init, headers, redirect: "manual" });
const setCookies = typeof response.headers.getSetCookie === "function" ? response.headers.getSetCookie() : [];
for (const line of setCookies) {
const pair = line.split(";")[0] ?? "";
const eq = pair.indexOf("=");
if (eq > 0) jar.set(pair.slice(0, eq).trim(), pair.slice(eq + 1).trim());
}
return response;
}
async function timed(jar: Map<string, string>, path: string, body: unknown, kind: Sample["kind"]): Promise<void> {
const started = Date.now();
let status = 0;
let firstByteMs: number | null = null;
let failed = false;
try {
const response = await request(jar, path, {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify(body),
});
status = response.status;
const reader = response.body?.getReader();
if (reader) {
let text = "";
for (;;) {
const chunk = await reader.read();
if (firstByteMs === null && (chunk.value || chunk.done)) firstByteMs = Date.now() - started;
if (chunk.value) text += new TextDecoder().decode(chunk.value);
if (chunk.done) break;
}
failed = text.includes('"type":"run.failed"') || text.includes(CONSULT_QUEUE_FAILED);
} else {
firstByteMs = Date.now() - started;
failed = status >= 400;
}
} catch {
status = 0;
failed = true;
}
samples.push({
kind,
status,
firstByteMs,
totalMs: Date.now() - started,
failed,
http429: status === 429,
http5xx: status >= 500 || status === 0,
});
}
async function login(index: number): Promise<Map<string, string> | null> {
const jar = new Map<string, string>();
const email = `capacity-smoke-${index}@${emailDomain}`;
const sent = await request(jar, "/api/auth/email-otp/send-verification-otp", {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ email, type: "sign-in" }),
});
if (!sent.ok) return null;
const signed = await request(jar, "/api/auth/sign-in/email-otp", {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ email, otp }),
});
if (!signed.ok || jar.size === 0) return null;
return jar;
}
async function prepareAccount(jar: Map<string, string>): Promise<boolean> {
const saved = await request(jar, "/api/account", {
method: "PATCH",
headers: { "content-type": "application/json" },
body: JSON.stringify({
name: "容量冒烟",
birth_date: "1990-06-15",
reported_birth_time: "12:00",
birth_time_source: "family_exact",
birth_time_period: null,
birth_time_clue: null,
uncertainty_before_minutes: 0,
uncertainty_after_minutes: 0,
country_code: "CN",
province_code: "110000",
city_code: "110000",
district_code: "110101",
latitude: 39.9042,
longitude: 116.4074,
timezone_offset: 8,
}),
});
return saved.ok;
}
async function oneUser(index: number): Promise<string[]> {
const deleted: string[] = [];
const jar = await login(index);
if (!jar) {
samples.push({ kind: "consult", status: 0, firstByteMs: null, totalMs: 0, failed: true, http429: false, http5xx: true });
return deleted;
}
if (!await prepareAccount(jar)) {
samples.push({ kind: "consult", status: 0, firstByteMs: null, totalMs: 0, failed: true, http429: false, http5xx: true });
return deleted;
}
const models = await request(jar, "/api/models");
const catalog = models.ok ? await models.json().catch(() => null) as { defaultModelId?: string } | null : null;
const modelId = catalog?.defaultModelId;
if (!modelId) {
samples.push({ kind: "consult", status: 0, firstByteMs: null, totalMs: 0, failed: true, http429: false, http5xx: true });
return deleted;
}
const sessionId = crypto.randomUUID();
const created = await request(jar, "/api/sessions", {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
id: sessionId,
title: "容量冒烟",
theme: "general",
model_id: modelId,
messages: [],
session_type: "consultation",
rectification_case_id: null,
chart_profile_role: "self",
}),
});
if (!created.ok) {
samples.push({ kind: "consult", status: created.status, firstByteMs: null, totalMs: 0, failed: true, http429: created.status === 429, http5xx: created.status >= 500 });
return deleted;
}
const chart = {
year: 1990,
month: 6,
day: 15,
hour: 12,
minute: 0,
lat: 39.9042,
lon: 116.4074,
tz: 8,
city: "北京",
ayanamsa: "lahiri",
};
await timed(jar, "/api/consult", {
...chart,
requestId: crypto.randomUUID(),
sessionId,
modelId,
consultationMode: "verified_chart",
theme: "general",
question: "今年工作整体顺不顺?",
history: [],
}, "consult");
await timed(jar, "/api/consult", {
...chart,
requestId: crypto.randomUUID(),
sessionId,
modelId,
consultationMode: "verified_chart",
theme: "general",
question: "那接下来几个月呢?",
history: [],
}, "consult");
if (index < 3) {
const opened = await request(jar, "/api/rectification/cases/open", {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ intent: "homepage", requestId: crypto.randomUUID() }),
});
const openedBody = opened.ok ? await opened.json().catch(() => null) as { caseId?: string; sessionId?: string } | null : null;
if (openedBody?.caseId && openedBody.sessionId) {
for (const message of ["我小时候搬家过一次。", "那次大概是上小学之前。"]) {
await timed(jar, "/api/rectification/agent", {
caseId: openedBody.caseId,
sessionId: openedBody.sessionId,
requestId: crypto.randomUUID(),
action: "message",
message,
}, "rectification");
}
deleted.push(openedBody.sessionId);
} else {
samples.push({
kind: "rectification",
status: opened.status,
firstByteMs: null,
totalMs: 0,
failed: true,
http429: opened.status === 429,
http5xx: opened.status >= 500 || opened.status === 0,
});
}
}
const removed = await request(jar, `/api/sessions/${sessionId}`, { method: "DELETE" });
if (removed.ok) deleted.push(sessionId);
return deleted;
}
const startedAt = Date.now();
const cleaned = (await Promise.all(Array.from({ length: requestedUsers }, (_, index) => oneUser(index)))).flat();
const consults = samples.filter((sample) => sample.kind === "consult");
const finalFailures = consults.filter((sample) => sample.failed || sample.http429 || sample.http5xx);
const firstBytes = consults.map((sample) => sample.firstByteMs).filter((value): value is number => value !== null);
const totals = consults.map((sample) => sample.totalMs);
const summary = {
users: requestedUsers,
host,
requests: samples.length,
consults: consults.length,
http429: samples.filter((sample) => sample.http429).length,
http5xx: samples.filter((sample) => sample.http5xx).length,
finalFailures: finalFailures.length,
finalFailureRatio: consults.length === 0 ? null : Number((finalFailures.length / consults.length).toFixed(4)),
firstByteP95Ms: percentile(firstBytes, 0.95),
totalP95Ms: percentile(totals, 0.95),
sessionsDeleted: cleaned.length,
serverSample: "not taken; this process has no staging docker stats",
elapsedMs: Date.now() - startedAt,
};
console.log(JSON.stringify(summary));
+24 -8
View File
@@ -129,6 +129,12 @@ import {
import { persistGuardedSessionTitle } from "@/lib/session-title-guard";
import { generateSessionTitle, shouldGenerateSessionTitle } from "@/lib/session-title-agent";
import { z } from "zod";
import { CONSULTATION_QUEUE_FAILED_MESSAGE, isConsultationRateLimitFailure, retryOnConsultationRateLimit } from "@/lib/consultation-workflow-retry";
import {
consumeUserRateLimitSet,
userRateLimitDeniedResponse,
userRateLimitMessages,
} from "@/lib/request-rate-limit";
export const runtime = "nodejs";
export const maxDuration = 480;
@@ -367,6 +373,11 @@ export async function POST(request: Request) {
const currentDate = chinaCalendarDate(requestTime);
const userId = user.id;
const limited = consumeUserRateLimitSet(["consultMinute", "consultHour"], userId);
if (!limited.ok) {
const denied = userRateLimitDeniedResponse(limited.retryAfterSeconds, userRateLimitMessages.consult);
return NextResponse.json(denied.body, { status: denied.status, headers: denied.headers });
}
const requestId = parsed.data.requestId;
const sessionId = parsed.data.sessionId;
const consultationTheme = parsed.data.theme;
@@ -1786,9 +1797,11 @@ export async function POST(request: Request) {
theme: parsed.data.theme,
});
const workflowContext = applyBirthTimeModeToWorkflowContext(
await runConsultationWorkflow(toolInput, {
foreground: true,
plan: prepared.preReserveResult,
await retryOnConsultationRateLimit({
run: () => runConsultationWorkflow(toolInput, {
foreground: true,
plan: prepared.preReserveResult,
}),
}),
consultationMode,
{ birthTimeSource: prepared.serverChart.truth.birthTimeSource },
@@ -1894,12 +1907,15 @@ export async function POST(request: Request) {
console.error(
`[consult] generation failed request=${requestId} model=${selectedModel.id} reason=${reason}`,
);
const rateLimited = isConsultationRateLimitFailure(error);
return NextResponse.json(
{
error: "暂时无法生成解读",
message: "咨询服务暂时不可用,请稍后再试。",
recovery: "稍后重试,或换一个模型继续。",
},
rateLimited
? { error: CONSULTATION_QUEUE_FAILED_MESSAGE, message: CONSULTATION_QUEUE_FAILED_MESSAGE }
: {
error: "暂时无法生成解读",
message: "咨询服务暂时不可用,请稍后再试。",
recovery: "稍后重试,或换一个模型继续。",
},
{ status: 503 },
);
}
@@ -18,6 +18,11 @@ import {
replyToDeclaredBirthWindow,
} from "@/lib/rectification-agentic/v9/agent-route-typed-message";
import { respondWithRectificationAgentTurn } from "@/lib/rectification-agentic/v9/agent-route-agent-turn";
import {
consumeUserRateLimitSet,
userRateLimitDeniedResponse,
userRateLimitMessages,
} from "@/lib/request-rate-limit";
export const runtime = "nodejs";
export const maxDuration = 240;
@@ -66,6 +71,20 @@ export async function POST(request: Request) {
const frozen = await refuseWhenAccountDeletionPending(userId);
if (frozen) return frozen;
// Opening a case is POST /api/rectification/cases/open and is not counted.
// Every turn on this route is, and the check is before any credit reserve.
const limited = consumeUserRateLimitSet(
["rectificationTurnMinute", "rectificationTurnHour"],
userId,
);
if (!limited.ok) {
const denied = userRateLimitDeniedResponse(
limited.retryAfterSeconds,
userRateLimitMessages.rectificationTurn,
);
return NextResponse.json(denied.body, { status: denied.status, headers: denied.headers });
}
if (action === "opening") {
const consistency = await finalizeSuccessfulTurnExit({
accounting: accounting as never, userId, caseId, action, segmentOpening: true,
+10 -2
View File
@@ -5,6 +5,7 @@ import { LONGFORM_SNAPSHOT_SECTION_ID } from "@/lib/personal-report-longform-sna
import { resolveMissingBirthTimezoneOffset } from "@/lib/birth-profile-timezone";
import { stampReportBirthFingerprint } from "@/lib/report-birth-binding";
import { runConsultationWorkflow } from "@/mastra";
import { consumeUserRateLimitSet, userRateLimitMessages } from "@/lib/request-rate-limit";
import { createPersonalReportAgent } from "@/mastra/personal-report";
import { loadLanguageModelCatalog } from "@/lib/model-catalog";
import {
@@ -68,8 +69,8 @@ function sanitizedErrorCode(error: unknown): string {
return "UnknownError";
}
function toNextResponse(response: { status: number; body: Record<string, unknown> }) {
return NextResponse.json(response.body, { status: response.status });
function toNextResponse(response: { status: number; body: Record<string, unknown>; headers?: Readonly<Record<string, string>> }) {
return NextResponse.json(response.body, { status: response.status, headers: response.headers });
}
function listReportView(
@@ -308,6 +309,13 @@ export async function POST(request: Request) {
},
persistence,
model: defaultModel,
consumeReportRateLimit: userId
? () => {
const decision = consumeUserRateLimitSet(["reportGenerateHour"], userId);
if (decision.ok) return decision;
return { ...decision, message: userRateLimitMessages.reportGenerate };
}
: undefined,
billing: {
reserve: async ({ userId: billingUserId, requestId, modelId }) => {
try {
+18 -14
View File
@@ -183,8 +183,10 @@ export async function POST(request: Request) {
synastryBirthPayload(globalBirthProfileFromAccountRow(selfResolved.row)),
synastryBirthPayload(globalBirthProfileFromAccountRow(partnerResolved.row)),
]);
const selfChart = await postPython("/api/chart", selfPayload);
const partnerChart = await postPython("/api/chart", partnerPayload);
const [selfChart, partnerChart] = await Promise.all([
postPython("/api/chart", { ...selfPayload, skip_vedastro_main_entry_overview: true }),
postPython("/api/chart", { ...partnerPayload, skip_vedastro_main_entry_overview: true }),
]);
const relationshipType = parsed.data.relationshipType === "business" || parsed.data.relationshipType === "family" || parsed.data.relationshipType === "general"
? parsed.data.relationshipType
: "romance";
@@ -210,18 +212,20 @@ export async function POST(request: Request) {
message: "该关系类型尚无可验证的专用合盘计算合同;已保留问题草稿。",
});
}
const selfD9 = await postPython("/api/varga_full", {
...selfPayload,
planets: selfChart.planets,
ascendant: selfChart.ascendant,
divisions: ["D9"],
});
const partnerD9 = await postPython("/api/varga_full", {
...partnerPayload,
planets: partnerChart.planets,
ascendant: partnerChart.ascendant,
divisions: ["D9"],
});
const [selfD9, partnerD9] = await Promise.all([
postPython("/api/varga_full", {
...selfPayload,
planets: selfChart.planets,
ascendant: selfChart.ascendant,
divisions: ["D9"],
}),
postPython("/api/varga_full", {
...partnerPayload,
planets: partnerChart.planets,
ascendant: partnerChart.ascendant,
divisions: ["D9"],
}),
]);
const synastry = await postPython("/api/synastry", {
male_moon: moonLongitude(selfChart),
female_moon: moonLongitude(partnerChart),
@@ -61,6 +61,7 @@ export const ChatMessageRow = memo(function ChatMessageRow({
received: "working",
"loading-method": "searching",
"chart-calculation": "solving",
"compute-queue": "working",
"evidence-validation": "working",
"answer-composition": "composing",
} as const)[message.activity.phase]
@@ -4,6 +4,8 @@
* turn is classified or the chart is read (TASK-consult-first-frame-and-pacing-20260928 D1).
*/
export const CONSULTATION_RECEIVED_LABEL = "收到,正在看你的问题…";
export const CONSULTATION_QUEUE_LABEL = "人有点多,排一下队";
export const CONSULTATION_QUEUE_FAILED_MESSAGE = "人有点多,这一轮没排上,点数已经退回。";
export const CONSULTATION_LOADING_METHOD_LABEL = "正在读取印度占星分析规则…";
export const CONSULTATION_CHART_CALCULATION_LABEL = "正在计算本命盘…";
export const CONSULTATION_EVIDENCE_VALIDATION_LABEL = "正在核对可用证据…";
@@ -10,6 +10,9 @@ export const publicActivityPhaseSchema = z.enum([
"received",
"loading-method",
"chart-calculation",
// Heavy compute is full. The calculate row keeps this sentence while the
// consultation retries, then returns to the chart-calculation label.
"compute-queue",
"evidence-validation",
"answer-composition",
]);
@@ -70,7 +70,7 @@ export function reduceConsultationTimeline(
return applyCalculateHints(created);
}
if (event.type === "activity") {
if (event.phase === "chart-calculation") {
if (event.phase === "chart-calculation" || event.phase === "compute-queue") {
return patchRow(state, CALCULATE_ID, {
status: "live",
label: event.label,
@@ -0,0 +1,57 @@
import { ConsultationWorkflowError } from "../mastra/consultation-workflow.ts";
import {
CONSULTATION_QUEUE_FAILED_MESSAGE,
CONSULTATION_QUEUE_LABEL,
} from "./consultation-activity-labels.ts";
export { CONSULTATION_QUEUE_FAILED_MESSAGE, CONSULTATION_QUEUE_LABEL };
export const CONSULTATION_QUEUE_MAX_RETRIES = 2;
export const CONSULTATION_QUEUE_MAX_WAIT_MS = 15_000;
export function isConsultationRateLimitFailure(error: unknown): boolean {
if (error instanceof ConsultationWorkflowError) return error.code === "workflow_rate_limited";
if (error instanceof Error && error.cause !== undefined) return isConsultationRateLimitFailure(error.cause);
return false;
}
function retryDelaySeconds(error: unknown): number | undefined {
if (error instanceof ConsultationWorkflowError && error.code === "workflow_rate_limited") {
return error.retryAfterSeconds ?? 2;
}
if (error instanceof Error && error.cause !== undefined) return retryDelaySeconds(error.cause);
return undefined;
}
export function consultationRunFailureMessage(error: unknown, fallback: string): string {
return isConsultationRateLimitFailure(error) ? CONSULTATION_QUEUE_FAILED_MESSAGE : fallback;
}
/**
* Retry one consultation workflow call when the heavy-compute gate answers
* 429. At most two retries, and the sleeps together stay within 15 seconds.
*/
export async function retryOnConsultationRateLimit<T>(input: Readonly<{
run: () => Promise<T>;
onQueue?: () => void | Promise<void>;
sleep?: (ms: number) => Promise<void>;
}>): Promise<T> {
const sleep = input.sleep ?? ((ms: number) => new Promise((resolve) => setTimeout(resolve, ms)));
let retries = 0;
let waitedMs = 0;
for (;;) {
try {
return await input.run();
} catch (error) {
const retryAfterSeconds = retryDelaySeconds(error);
if (retryAfterSeconds === undefined || retries >= CONSULTATION_QUEUE_MAX_RETRIES) throw error;
const remainingMs = CONSULTATION_QUEUE_MAX_WAIT_MS - waitedMs;
if (remainingMs <= 0) throw error;
const waitMs = Math.min(retryAfterSeconds * 1000, remainingMs);
retries += 1;
waitedMs += waitMs;
await input.onQueue?.();
await sleep(waitMs);
}
}
}
+23 -1
View File
@@ -63,7 +63,11 @@ export const personalReportCreateRequestSchema = z.object({
export type PersonalReportCreateRequest = z.infer<typeof personalReportCreateRequestSchema>;
export type ReportRouteResponse = Readonly<{ status: number; body: Record<string, unknown> }>;
export type ReportRouteResponse = Readonly<{
status: number;
body: Record<string, unknown>;
headers?: Readonly<Record<string, string>>;
}>;
export type ReportServicePort = Pick<
PersonalReportService,
@@ -231,6 +235,10 @@ export type ReportCreateCoreDeps = Readonly<{
persistence: ReportServicePort;
model: Readonly<{ id: string; configVersion?: number; inputCostMicrousdPerMillion?: number; outputCostMicrousdPerMillion?: number }> | null;
billing?: ReportBillingPort;
/** Hourly per-user cap. Checked after a replay and the daily cap, before reserve. */
consumeReportRateLimit?: () =>
| { readonly ok: true }
| { readonly ok: false; readonly retryAfterSeconds: number; readonly message: string };
runWorkflow: (input: ConsultationInput) => Promise<unknown>;
createAgent: (model: Readonly<{ id: string }>) => ReportAgentPort;
skillSnapshot: SkillSnapshot;
@@ -387,6 +395,20 @@ export async function resolveReportCreate(deps: ReportCreateCoreDeps): Promise<R
let billingReserved = false;
if (deps.billing && selectedModel) {
if (deps.consumeReportRateLimit) {
const limited = deps.consumeReportRateLimit();
if (!limited.ok) {
return {
status: 429,
headers: { "Retry-After": String(limited.retryAfterSeconds) },
body: {
error: limited.message,
code: "rate_limited",
retryAfterSeconds: limited.retryAfterSeconds,
},
};
}
}
let authorization;
try {
authorization = await deps.billing.reserve({ userId, requestId: payload.requestId, modelId: selectedModel.id });
+69 -6
View File
@@ -23,6 +23,18 @@ export const requestRateLimits = {
ephemeris: { limit: 30, windowMs: 60_000 },
synastry: { limit: 8, windowMs: 60 * 60_000 },
sessionWrite: { limit: 40, windowMs: 60_000 },
// Two windows each: a minute burst and an hour budget (product 2026-10-14).
consultMinute: { limit: 10, windowMs: 60_000 },
consultHour: { limit: 120, windowMs: 60 * 60_000 },
rectificationTurnMinute: { limit: 5, windowMs: 60_000 },
rectificationTurnHour: { limit: 40, windowMs: 60 * 60_000 },
reportGenerateHour: { limit: 3, windowMs: 60 * 60_000 },
} as const;
export const userRateLimitMessages = {
consult: "这一会儿问得有点密,先停一下。",
rectificationTurn: "这一会儿校正问得有点密,先停一下。",
reportGenerate: "这一小时的报告已经够多了,先停一下。",
} as const;
export type RequestRateLimitBucket = keyof typeof requestRateLimits;
@@ -54,30 +66,41 @@ function sweepExpiredRateLimitKeys(
}
}
export function consumeRequestRateLimit(input: RequestRateLimitInput): RequestRateLimitDecision {
if (input.windowMs > seenMaxWindowMs) seenMaxWindowMs = input.windowMs;
consumeCalls += 1;
function evaluateRequestRateLimit(
input: RequestRateLimitInput,
record: boolean,
): RequestRateLimitDecision {
const now = input.now ?? Date.now();
const store = input.store ?? defaultStore();
if (consumeCalls % 1_000 === 0 || store.size > 5_000) {
sweepExpiredRateLimitKeys(store, now, seenMaxWindowMs);
if (record) {
if (input.windowMs > seenMaxWindowMs) seenMaxWindowMs = input.windowMs;
consumeCalls += 1;
if (consumeCalls % 1_000 === 0 || store.size > 5_000) {
sweepExpiredRateLimitKeys(store, now, seenMaxWindowMs);
}
}
const windowStart = now - input.windowMs;
const recent = (store.get(input.key) ?? []).filter((timestamp) => timestamp > windowStart);
if (recent.length >= input.limit) {
const retryAfterSeconds = Math.max(1, Math.ceil((recent[0]! + input.windowMs - now) / 1000));
store.set(input.key, recent);
if (record) store.set(input.key, recent);
return { ok: false, retryAfterSeconds };
}
if (!record) return { ok: true };
recent.push(now);
store.set(input.key, recent);
return { ok: true };
}
export function consumeRequestRateLimit(input: RequestRateLimitInput): RequestRateLimitDecision {
return evaluateRequestRateLimit(input, true);
}
export function consumeUserRequestRateLimit(
bucket: RequestRateLimitBucket,
userId: string,
now?: number,
store?: RequestRateLimitStore,
): RequestRateLimitDecision {
const spec = requestRateLimits[bucket];
return consumeRequestRateLimit({
@@ -85,5 +108,45 @@ export function consumeUserRequestRateLimit(
limit: spec.limit,
windowMs: spec.windowMs,
now,
store,
});
}
/**
* Allow the request only when every bucket still has room. A refusal writes
* nothing, so the bucket that still had room is not charged for the miss.
*/
export function consumeUserRateLimitSet(
buckets: readonly RequestRateLimitBucket[],
userId: string,
now?: number,
store?: RequestRateLimitStore,
): RequestRateLimitDecision {
let retryAfterSeconds = 0;
for (const bucket of buckets) {
const spec = requestRateLimits[bucket];
const decision = evaluateRequestRateLimit({
key: `${bucket}:${userId}`,
limit: spec.limit,
windowMs: spec.windowMs,
now,
store,
}, false);
if (!decision.ok) retryAfterSeconds = Math.max(retryAfterSeconds, decision.retryAfterSeconds);
}
if (retryAfterSeconds > 0) return { ok: false, retryAfterSeconds };
for (const bucket of buckets) consumeUserRequestRateLimit(bucket, userId, now, store);
return { ok: true };
}
export function userRateLimitDeniedResponse(retryAfterSeconds: number, message: string) {
return {
status: 429 as const,
headers: { "Retry-After": String(retryAfterSeconds) },
body: {
error: message,
code: "rate_limited" as const,
retryAfterSeconds,
},
};
}
+2 -1
View File
@@ -22,6 +22,7 @@ import {
} from "./consultation-step-timing.ts";
import { createVisibleTextTransformer } from "./stream-text-response.ts";
import { consultationWriteLabel } from "./consultation-activity-labels.ts";
import { consultationRunFailureMessage } from "./consultation-workflow-retry.ts";
import { logTruncatedReasoning } from "./consultation-budget.ts";
import {
classifyPass4,
@@ -1177,7 +1178,7 @@ export function streamAgentResponse(options: StreamAgentResponseOptions) {
send(controller, {
type: "run.failed",
code,
message: runFailedMessage(code),
message: consultationRunFailureMessage(error, runFailedMessage(code)),
...(failureReceipt ? { receipt: failureReceipt } : {}),
});
if (!disconnected) controller.close();
+42 -15
View File
@@ -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, type ProviderReasoningEffort } from "../lib/agent-generation-settings.ts";
import { chartCalculationProgressLabel, evidenceLookupActivityLabel } from "../lib/consultation-activity-labels.ts";
import { CONSULTATION_QUEUE_LABEL, retryOnConsultationRateLimit } from "../lib/consultation-workflow-retry.ts";
import { consultDomainConcurrency } from "../lib/consult/prep-flags.ts";
import {
buildEvidenceCard,
@@ -1309,6 +1310,36 @@ export function createConsultationTools(ctx: ConsultationAgentContext) {
try {
const userIntent = ctx.plan?.userIntent ?? input.question;
const executions: DomainExecution[] = [];
const runQueuedWorkflow = async (
toolInput: ReturnType<typeof consultationInputSchema.parse>,
domainPlan: ConsultationPlan,
progressLabel: string,
) => {
let queued = false;
const workflow = await retryOnConsultationRateLimit({
run: () => (ctx.runWorkflow ?? runConsultationWorkflow)(toolInput, {
foreground: true,
signal: context.abortSignal ?? ctx.abortSignal,
plan: domainPlan,
requestId: ctx.requestId,
entrypoint: ctx.entrypoint,
}),
onQueue: async () => {
queued = true;
await context.writer?.custom({
type: "data-jyotish-activity",
data: { phase: "compute-queue", label: CONSULTATION_QUEUE_LABEL },
});
},
});
if (queued) {
await context.writer?.custom({
type: "data-jyotish-activity",
data: { phase: "chart-calculation", label: progressLabel },
});
}
return workflow;
};
if (consultDomainConcurrency() === 1) {
for (let index = 0; index < domains.length; index += 1) {
const domain = domains[index];
@@ -1344,13 +1375,11 @@ export function createConsultationTools(ctx: ConsultationAgentContext) {
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 workflow = await runQueuedWorkflow(
toolInput,
domainPlan,
chartCalculationProgressLabel(index + 1, domains.length),
);
const guarded = applyBirthTimeModeToWorkflowContext(workflow, ctx.consultationMode, {
birthTimeSource: ctx.serverChart.truth.birthTimeSource,
});
@@ -1374,7 +1403,7 @@ export function createConsultationTools(ctx: ConsultationAgentContext) {
},
});
}
const parallelExecutions = await Promise.all(domains.map(async (domain) => {
const parallelExecutions = await Promise.all(domains.map(async (domain, domainIndex) => {
if (!domain) return null;
const domainPlan = ctx.plan
&& domains.length === 1
@@ -1396,13 +1425,11 @@ export function createConsultationTools(ctx: ConsultationAgentContext) {
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 workflow = await runQueuedWorkflow(
toolInput,
domainPlan,
chartCalculationProgressLabel(domainIndex + 1, domains.length),
);
const guarded = applyBirthTimeModeToWorkflowContext(workflow, ctx.consultationMode, {
birthTimeSource: ctx.serverChart.truth.birthTimeSource,
});
+11 -1
View File
@@ -102,14 +102,23 @@ export type ConsultationWorkflowFailureCode = typeof consultationWorkflowFailure
export class ConsultationWorkflowError extends Error {
readonly code: ConsultationWorkflowFailureCode;
readonly retryAfterSeconds?: number;
constructor(code: ConsultationWorkflowFailureCode, message: string) {
constructor(code: ConsultationWorkflowFailureCode, message: string, retryAfterSeconds?: number) {
super(message);
this.name = "ConsultationWorkflowError";
this.code = code;
if (retryAfterSeconds !== undefined) this.retryAfterSeconds = retryAfterSeconds;
}
}
function parseRetryAfterHeader(raw: string | null): number {
if (!raw) return 2;
const seconds = Number(raw.trim());
if (!Number.isFinite(seconds)) return 2;
return Math.max(1, Math.ceil(seconds));
}
function httpFailureCode(status: number): ConsultationWorkflowFailureCode {
if (status === 400) return "workflow_bad_request";
if (status === 403) return "workflow_forbidden";
@@ -174,6 +183,7 @@ export async function runConsultationWorkflow(
throw new ConsultationWorkflowError(
response.ok ? "workflow_empty_response" : httpFailureCode(response.status),
data?.error || data?.message || `Jyotish API returned ${response.status}`,
response.status === 429 ? parseRetryAfterHeader(response.headers.get("retry-after")) : undefined,
);
}
const parsed = consultationWorkflowResponseSchema.safeParse(data);
@@ -0,0 +1,44 @@
import assert from "node:assert/strict";
import { readdirSync, readFileSync, statSync } from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";
import test from "node:test";
const frontendRoot = fileURLToPath(new URL("../", import.meta.url));
const roots = ["src/app/api", "src/lib"];
// Empty on purpose. A call site that cannot send skip must be listed here with a reason.
const allowlist: string[] = [];
function walk(dir: string): string[] {
const found: string[] = [];
for (const name of readdirSync(dir)) {
const full = path.join(dir, name);
if (statSync(full).isDirectory()) found.push(...walk(full));
else if (name.endsWith(".ts") || name.endsWith(".tsx")) found.push(full);
}
return found;
}
test("python /api/chart bodies from app/api and lib send skip", () => {
const failures: string[] = [];
for (const root of roots) {
for (const file of walk(path.join(frontendRoot, root))) {
const rel = path.relative(frontendRoot, file).replaceAll("\\", "/");
if (allowlist.includes(rel)) continue;
const text = readFileSync(file, "utf8");
const re = /["']\/api\/chart["']/g;
let match: RegExpExecArray | null;
while ((match = re.exec(text))) {
const idx = match.index;
const before = text.slice(Math.max(0, idx - 120), idx);
if (!/postPython\s*\(|callEngine\s*\(|fetchEngine\s*\(|fetch\s*\(/.test(before)) continue;
const window = text.slice(Math.max(0, idx - 40), idx + 420);
const sendsSkip = window.includes("skip_vedastro_main_entry_overview: true")
|| window.includes("transitChartBody(")
|| (/["']\/api\/chart["']\s*,\s*payload/.test(window) && text.includes("skip_vedastro_main_entry_overview: true"));
if (!sendsSkip) failures.push(`${rel}:${text.slice(0, idx).split("\n").length}`);
}
}
}
assert.deepEqual(failures, []);
});
@@ -179,7 +179,10 @@ test("consult user turn no longer dumps tool JSON that the model would parrot",
assert.doesNotMatch(chartBranch, /JSON\.stringify\(toolInput\)/);
assert.doesNotMatch(chartBranch, /经过服务端校验的工具参数/);
assert.match(route, /文末技法审计表/);
assert.match(chartBranch, /await runConsultationWorkflow\(toolInput/);
// 原值: /await runConsultationWorkflow\(toolInput/
// 新值: /runConsultationWorkflow\(toolInput/
// 原因: 调用包进 retryOnConsultationRateLimit,await 不再紧贴函数名。仍锁工作流入参是 toolInput;用户轮不塞工具 JSON 的断言不动。
assert.match(chartBranch, /runConsultationWorkflow\(toolInput/);
});
// 原值: test "the opener is a shape — 反差, 谁推谁修, 扮演哪个象, then at most three actions"(锁 OPENER SHAPE 四步、≤ 20 字行动、降级路线「第 3 步照给」)
@@ -0,0 +1,102 @@
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import test from "node:test";
import { ConsultationWorkflowError } from "../src/mastra/consultation-workflow.ts";
import { CONSULTATION_QUEUE_LABEL } from "../src/lib/consultation-activity-labels.ts";
import {
CONSULTATION_QUEUE_FAILED_MESSAGE,
consultationRunFailureMessage,
isConsultationRateLimitFailure,
retryOnConsultationRateLimit,
} from "../src/lib/consultation-workflow-retry.ts";
import {
emptyConsultationTimeline,
reduceConsultationTimelineEvents,
} from "../src/lib/consultation-run-timeline.ts";
test("a 429 is retried once and the queue line is shown before success", async () => {
let calls = 0;
const queued: string[] = [];
const sleeps: number[] = [];
const result = await retryOnConsultationRateLimit({
run: async () => {
calls += 1;
if (calls === 1) throw new ConsultationWorkflowError("workflow_rate_limited", "busy", 2);
return "ok";
},
onQueue: () => { queued.push(CONSULTATION_QUEUE_LABEL); },
sleep: async (ms) => { sleeps.push(ms); },
});
assert.equal(result, "ok");
assert.equal(calls, 2);
assert.deepEqual(sleeps, [2000]);
assert.deepEqual(queued, ["人有点多,排一下队"]);
});
test("three 429s stop within two retries and 15 seconds", async () => {
let calls = 0;
let slept = 0;
await assert.rejects(
() => retryOnConsultationRateLimit({
run: async () => {
calls += 1;
throw new ConsultationWorkflowError("workflow_rate_limited", "busy", 2);
},
sleep: async (ms) => { slept += ms; },
}),
(error: unknown) => {
assert.equal(isConsultationRateLimitFailure(error), true);
return true;
},
);
assert.equal(calls, 3);
assert.equal(slept, 4_000);
assert.ok(slept <= 15_000);
assert.equal(
consultationRunFailureMessage(new ConsultationWorkflowError("workflow_rate_limited", "busy", 2), "其他失败"),
CONSULTATION_QUEUE_FAILED_MESSAGE,
);
assert.equal(consultationRunFailureMessage(new Error("other"), "其他失败"), "其他失败");
});
test("Retry-After cannot push the total wait past 15 seconds", async () => {
let calls = 0;
let slept = 0;
await assert.rejects(() => retryOnConsultationRateLimit({
run: async () => {
calls += 1;
throw new ConsultationWorkflowError("workflow_rate_limited", "busy", 20);
},
sleep: async (ms) => { slept += ms; },
}));
assert.equal(calls, 2);
assert.equal(slept, 15_000);
});
test("the queue sentence replaces the calculate row and the chart label comes back", () => {
const calculating = reduceConsultationTimelineEvents([
{ type: "tool.started", callId: "tool-1", tool: "run-jyotish-consultation", label: "正在计算个人星盘" },
{ type: "activity", phase: "chart-calculation", label: "正在计算本命盘…" },
], emptyConsultationTimeline());
const queued = reduceConsultationTimelineEvents([
{ type: "activity", phase: "compute-queue", label: CONSULTATION_QUEUE_LABEL },
], calculating);
assert.equal(queued.rows.find((row) => row.kind === "calculate")?.label, "人有点多,排一下队");
assert.equal(queued.rows.find((row) => row.kind === "calculate")?.status, "live");
const resumed = reduceConsultationTimelineEvents([
{ type: "activity", phase: "chart-calculation", label: "正在计算本命盘…" },
], queued);
assert.equal(resumed.rows.find((row) => row.kind === "calculate")?.label, "正在计算本命盘…");
});
test("a failed consultation still settles through the existing cancel path before the queue sentence", () => {
const stream = readFileSync(new URL("../src/lib/stream-agent-response.ts", import.meta.url), "utf8");
const consult = readFileSync(new URL("../src/app/api/consult/route.ts", import.meta.url), "utf8");
const onErrorAt = stream.indexOf("await options.onError?.");
const messageAt = stream.indexOf("consultationRunFailureMessage(error, runFailedMessage(code))");
assert.ok(onErrorAt > 0 && messageAt > onErrorAt);
assert.match(consult, /onError: \(error\) => settleRun\(\s*cancel/);
assert.match(consult, /else await cancel\(\)/);
assert.match(consult, /retryOnConsultationRateLimit/);
assert.match(consult, /CONSULTATION_QUEUE_FAILED_MESSAGE/);
});
@@ -1525,3 +1525,19 @@ test("core read: the detail response has no coverSubject, and never the fingerpr
assert.doesNotMatch(body, new RegExp(BIRTH_FP));
assert.doesNotMatch(body, /coverSubject|birthFingerprint|birth_date|latitude|longitude|timezone|birthPlaceLabel|北京/);
});
// 原值: 插在「429 at the daily limit」和「502 model_unavailable」之间。
// 新值: 整段挪到文件末尾,断言不变。
// 原因: 插在中部让 privacy 白名单按行号锁定的三行各下移 13 行。挪到末尾后那三行回到原行号,SAFE_LINES 不改。
test("core create: hourly rate limit returns 429 and does not reserve", async () => {
const billing = billingHarness();
const response = await resolveReportCreate(baseDeps({
billing: billing.billing,
consumeReportRateLimit: () => ({ ok: false, retryAfterSeconds: 12, message: "这一小时的报告已经够多了,先停一下。" }),
}));
assert.equal(response.status, 429);
assert.equal(response.body.code, "rate_limited");
assert.equal(response.body.retryAfterSeconds, 12);
assert.equal(response.headers?.["Retry-After"], "12");
assert.equal(billing.reservations, 0);
});
+100
View File
@@ -1,8 +1,13 @@
import assert from "node:assert/strict";
import test from "node:test";
import { readFileSync } from "node:fs";
import {
consumeRequestRateLimit,
consumeUserRateLimitSet,
requestRateLimits,
resetRequestRateLimitMaintenanceForTests,
userRateLimitDeniedResponse,
userRateLimitMessages,
} from "../src/lib/request-rate-limit.ts";
test("a user is allowed up to the window limit and then must wait", () => {
@@ -110,3 +115,98 @@ test("keys past the longest seen window are swept, and a one-hour key survives 6
assert.equal(mixed.has("ephemeris:person"), true);
resetRequestRateLimitMaintenanceForTests();
});
test("consult allows 10 per minute and refuses the 11th without charging the hour bucket further", () => {
const store = new Map<string, number[]>();
for (let index = 0; index < 10; index += 1) {
const decision = consumeUserRateLimitSet(
["consultMinute", "consultHour"],
"person",
1_000 + index,
store,
);
assert.deepEqual(decision, { ok: true });
}
const denied = consumeUserRateLimitSet(["consultMinute", "consultHour"], "person", 2_000, store);
assert.equal(denied.ok, false);
if (!denied.ok) assert.equal(denied.retryAfterSeconds, 59);
assert.equal(store.get("consultMinute:person")?.length, 10);
assert.equal(store.get("consultHour:person")?.length, 10);
assert.equal(requestRateLimits.consultMinute.limit, 10);
assert.equal(requestRateLimits.consultHour.limit, 120);
});
test("the hour bucket can refuse while the minute bucket still has room, and neither bucket records the miss", () => {
const store = new Map<string, number[]>();
for (let index = 0; index < 120; index += 1) {
consumeRequestRateLimit({
key: "consultHour:person",
limit: 120,
windowMs: 60 * 60_000,
now: index * 1_000,
store,
});
}
const denied = consumeUserRateLimitSet(["consultMinute", "consultHour"], "person", 130_000, store);
assert.equal(denied.ok, false);
assert.equal(store.has("consultMinute:person"), false);
assert.equal(store.get("consultHour:person")?.length, 120);
});
test("rectification turns allow 5 per minute and 40 per hour, reports allow 3 per hour", () => {
const store = new Map<string, number[]>();
for (let index = 0; index < 5; index += 1) {
assert.deepEqual(
consumeUserRateLimitSet(["rectificationTurnMinute", "rectificationTurnHour"], "person", 5_000 + index, store),
{ ok: true },
);
}
const denied = consumeUserRateLimitSet(
["rectificationTurnMinute", "rectificationTurnHour"],
"person",
5_100,
store,
);
assert.equal(denied.ok, false);
assert.equal(store.get("rectificationTurnMinute:person")?.length, 5);
assert.equal(requestRateLimits.rectificationTurnMinute.limit, 5);
assert.equal(requestRateLimits.rectificationTurnHour.limit, 40);
assert.equal(requestRateLimits.reportGenerateHour.limit, 3);
assert.equal(requestRateLimits.reportGenerateHour.windowMs, 60 * 60_000);
const reports = new Map<string, number[]>();
for (let index = 0; index < 3; index += 1) {
assert.deepEqual(
consumeUserRateLimitSet(["reportGenerateHour"], "person", 8_000 + index, reports),
{ ok: true },
);
}
const reportDenied = consumeUserRateLimitSet(["reportGenerateHour"], "person", 9_000, reports);
assert.equal(reportDenied.ok, false);
assert.equal(reports.get("reportGenerateHour:person")?.length, 3);
});
test("a denied response carries 429, rate_limited, and Retry-After", () => {
const denied = userRateLimitDeniedResponse(12, userRateLimitMessages.consult);
assert.equal(denied.status, 429);
assert.equal(denied.headers["Retry-After"], "12");
assert.deepEqual(denied.body, {
error: "这一会儿问得有点密,先停一下。",
code: "rate_limited",
retryAfterSeconds: 12,
});
});
test("consult, rectification turns, and report creation check the limit before they reserve", () => {
const consult = readFileSync(new URL("../src/app/api/consult/route.ts", import.meta.url), "utf8");
const rectification = readFileSync(new URL("../src/app/api/rectification/agent/route.ts", import.meta.url), "utf8");
const caseOpen = readFileSync(new URL("../src/app/api/rectification/cases/open/route.ts", import.meta.url), "utf8");
const reports = readFileSync(new URL("../src/app/api/reports/route.ts", import.meta.url), "utf8");
const reportCore = readFileSync(new URL("../src/lib/personal-report-route-core.ts", import.meta.url), "utf8");
assert.ok(consult.indexOf("consumeUserRateLimitSet(") < consult.indexOf("reserve_consultation_usage"));
assert.match(consult, /userRateLimitDeniedResponse\(/);
assert.ok(rectification.indexOf("consumeUserRateLimitSet(") < rectification.indexOf("respondWithRectificationAgentTurn("));
assert.equal(caseOpen.includes("consumeUserRateLimitSet"), false);
assert.match(reports, /consumeUserRateLimitSet\(\["reportGenerateHour"\]/);
assert.ok(reportCore.indexOf("if (deps.consumeReportRateLimit)") < reportCore.indexOf("deps.billing.reserve"));
});
+103
View File
@@ -0,0 +1,103 @@
import assert from "node:assert/strict";
import { spawnSync } from "node:child_process";
import { fileURLToPath } from "node:url";
import test from "node:test";
const frontendRoot = fileURLToPath(new URL("../", import.meta.url));
const partnerId = "11111111-1111-4111-8111-111111111111";
function runRoute(script: string): { romance: Record<string, unknown>; business: Record<string, unknown>; maxChart: number; maxVarga: number; chartSkip: boolean } {
const result = spawnSync(process.execPath, [
"--experimental-test-module-mocks",
"--import",
"tsx",
"--input-type=module",
"--eval",
script,
], { cwd: frontendRoot, encoding: "utf8", timeout: 20000 });
assert.equal(result.status, 0, result.stderr || result.stdout);
const line = result.stdout.trim().split("\n").at(-1) ?? "{}";
return JSON.parse(line);
}
test("synastry charts send skip concurrently and keep the stubbed JSON", () => {
const script = String.raw`
import { mock } from "node:test";
import { pathToFileURL } from "node:url";
const moduleUrl = (path) => pathToFileURL(process.cwd() + "/" + path).href;
mock.module("server-only", { namedExports: {} });
mock.module("@/lib/account-deletion-server", { namedExports: { refuseWhenAccountDeletionPending: async () => null } });
mock.module("@/lib/product-access", { namedExports: { isProductEnabled: async () => true } });
mock.module("@/lib/request-rate-limit", { namedExports: { consumeUserRequestRateLimit: () => ({ ok: true, retryAfterSeconds: 1 }) } });
mock.module("@/lib/supabase/server", { namedExports: { createServerSupabaseClient: async () => ({ auth: { getUser: async () => ({ data: { user: { id: "user-1" } }, error: null }) } }) } });
mock.module("@/lib/subject-birth", { namedExports: {
ConsultationSubjectError: class ConsultationSubjectError extends Error {},
loadSubjectBirth: async () => ({ chartable: true, row: { name: "fixture" } }),
} });
mock.module("@/lib/server-owned-birth-profile", { namedExports: { globalBirthProfileFromAccountRow: (row) => row } });
mock.module("@/lib/global-birth-payloads", { namedExports: {
synastryBirthPayload: async () => ({ year: 1990, month: 1, day: 1, hour: 12, minute: 0, lat: 28.6139, lon: 77.209, tz: 5.5, ayanamsa: "lahiri" }),
} });
const chartBodies = [];
const maxInflight = { chart: 0, varga: 0 };
const inflight = { chart: 0, varga: 0 };
const waiters = { chart: [], varga: [] };
globalThis.fetch = async (inputUrl, init) => {
const url = String(inputUrl);
const body = JSON.parse(String(init.body));
const kind = url.endsWith("/api/chart") ? "chart" : url.endsWith("/api/varga_full") ? "varga" : "other";
if (kind === "chart") chartBodies.push(body);
if (kind === "chart" || kind === "varga") {
inflight[kind] += 1;
maxInflight[kind] = Math.max(maxInflight[kind], inflight[kind]);
await new Promise((resolve) => {
waiters[kind].push(resolve);
if (waiters[kind].length >= 2) {
const pending = waiters[kind].splice(0, 2);
pending.forEach((fn) => fn());
}
});
inflight[kind] -= 1;
}
if (url.endsWith("/api/chart")) {
return Response.json({
planets: { Moon: { lon: 12.5, sign: "Aries", nakshatra: "Ashwini", nakshatra_pada: 1, nakshatra_lord: "Ketu" } },
ascendant: { sign: "Leo" },
});
}
if (url.endsWith("/api/varga_full")) {
const divisions = body.divisions || [];
if (divisions.includes("D9")) {
return Response.json({ source: "engine", result: { D9_Navamsa: { ascendant: { sign: "Virgo" }, planets: { Moon: { sign: "Cancer" }, Venus: { sign: "Taurus" }, Mars: { sign: "Aries" } } } } });
}
return Response.json({ source: "engine", result: {
D2_Hora: { planets: { Moon: { sign: "Taurus" } } },
D10_Dasamsa: { ascendant: { sign: "Libra" } },
D11_Rudramsa: { planets: { Sun: { sign: "Capricorn" } } },
} });
}
if (url.endsWith("/api/synastry")) return Response.json({ total_score: 18, max_score: 36 });
return Response.json({}, { status: 500 });
};
const { POST } = await import(moduleUrl("src/app/api/synastry/route.ts"));
const romanceRes = await POST(new Request("https://example.test/api/synastry", { method: "POST", body: JSON.stringify({ partnerChartProfileId: "${partnerId}", relationshipType: "romance" }) }));
const romance = await romanceRes.json();
const businessRes = await POST(new Request("https://example.test/api/synastry", { method: "POST", body: JSON.stringify({ partnerChartProfileId: "${partnerId}", relationshipType: "business" }) }));
const business = await businessRes.json();
console.log(JSON.stringify({ romance, business, maxChart: maxInflight.chart, maxVarga: maxInflight.varga, chartSkip: chartBodies.every((body) => body.skip_vedastro_main_entry_overview === true) && chartBodies.length === 4 }));
`;
const got = runRoute(script);
assert.equal(got.maxChart, 2);
assert.equal(got.maxVarga, 2);
assert.equal(got.chartSkip, true);
assert.equal(got.romance.status, "ok");
assert.equal(got.romance.method, "ashtakoot_plus_moon_nakshatra_d9");
assert.deepEqual(got.romance.selfChart, {
moon: { sign: "Aries", nakshatra: "Ashwini", pada: 1, lord: "Ketu", longitude: 12.5 },
d9: { ascendant: { sign: "Virgo" }, moon: { sign: "Cancer" }, venus: { sign: "Taurus" }, mars: { sign: "Aries" }, source: "engine" },
});
assert.equal((got.romance.relationshipReport as { scoreBand?: string }).scoreBand, "mixed");
assert.equal(got.business.status, "ok");
assert.equal(got.business.method, "d2_d10_d11_business_screening_partial");
assert.equal((got.business.relationshipReport as { scoreBand?: string }).scoreBand, "not_scored");
});
+49
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import contextlib
import hashlib
import importlib.util
import json
import subprocess
import sys
@@ -22,6 +23,34 @@ from typing import Any, Callable
ENGINE = "PyJHora/JHora"
REPO_ROOT = Path(__file__).resolve().parents[1]
_jhora_spec_cache: bool | None = None
def reset_jhora_import_cache() -> None:
"""Tests reset the process-local "is PyJHora installed?" answer."""
global _jhora_spec_cache
_jhora_spec_cache = None
def _jhora_importable() -> bool:
global _jhora_spec_cache
if _jhora_spec_cache is None:
_jhora_spec_cache = importlib.util.find_spec("jhora") is not None
return _jhora_spec_cache
def _no_jhora_tajaka_replay(settings: dict[str, Any]) -> dict[str, Any]:
"""Same dict the child returns when ``import jhora`` raises ModuleNotFoundError."""
copied = dict(settings)
return {
"status": "blocked",
"reason": "pyjhora_ayanamsa_configuration_failed:ModuleNotFoundError",
"settings": copied,
"node_mode": {
"requested": copied.get("node_mode"),
"status": "unsupported_by_replay_adapter",
},
}
_TAJAKA_YEAR_LORD_MODULE = "jhora.horoscope.transit.tajaka"
_YEAR_LORD_MODULES = (
"jhora.horoscope.dhasa.annual.mudda",
@@ -139,6 +168,10 @@ def _run_isolated_tajaka_year_lord(
"settings": settings,
}
request_hash = _request_hash(request)
if not _jhora_importable():
replay = _no_jhora_tajaka_replay(settings)
replay["request_hash"] = request_hash
return replay
command = [
sys.executable,
"-c",
@@ -276,6 +309,10 @@ def _run_isolated_annual_replay(
"settings": settings,
}
request_hash = _request_hash(request)
if not _jhora_importable():
replay = _no_jhora_annual_replay(settings)
replay["request_hash"] = request_hash
return replay
command = [
sys.executable,
"-c",
@@ -391,6 +428,18 @@ def _annual_worker_result(
}
def _no_jhora_annual_replay(settings: dict[str, Any]) -> dict[str, Any]:
"""Same dict the annual child returns when ``import jhora`` raises ModuleNotFoundError."""
copied = dict(settings)
node_mode = {"requested": copied.get("node_mode"), "status": "unsupported_by_replay_adapter"}
reason = "pyjhora_annual_adapter_setup_failed:ModuleNotFoundError"
return _annual_worker_result(
{key: {"status": "blocked", "reason": reason} for key in _ANNUAL_SURFACE_KEYS},
settings=copied,
node_mode=node_mode,
)
def _annual_replay_blocked(reason: str, request_hash: str, settings: dict[str, Any]) -> dict[str, Any]:
return _annual_worker_result(
{key: {"status": "blocked", "reason": reason} for key in _ANNUAL_SURFACE_KEYS},
+147 -18
View File
@@ -1,20 +1,28 @@
"""Fail-fast bounded concurrency for heavy Jyotish API compute endpoints.
"""Bounded concurrency for heavy Jyotish API compute endpoints.
Request threads that run rectification scans or high-rigor workflows share one
process-wide semaphore sized for the 2 vCPU production host. Saturated requests
return immediately; they are not queued. Health checks and other light routes
must not call this gate.
Request threads that run rectification scans or consultation workflows share one
process-wide semaphore sized for the 2 vCPU production host. A saturated request
waits up to 8 seconds when fewer than 4 threads are already waiting. If that
queue is full or the wait expires, the caller still gets HTTP 429 and
Retry-After. A thread is never parked without a deadline. Health checks and
other light routes must not call this gate.
"""
from __future__ import annotations
import logging
import os
import threading
import time
DEFAULT_CONCURRENCY = 2
DEFAULT_RETRY_AFTER_SECONDS = 2
DEFAULT_QUEUE_WAIT_SECONDS = 8.0
DEFAULT_MAX_WAITERS = 4
ENV_CONCURRENCY = "JYOTISH_HEAVY_COMPUTE_CONCURRENCY"
ENV_RETRY_AFTER = "JYOTISH_HEAVY_COMPUTE_RETRY_AFTER_SECONDS"
ENV_GATE_CHART_OVERVIEW = "JYOTISH_GATE_CHART_OVERVIEW"
_LOG = logging.getLogger("jyotish.heavy_compute")
HEAVY_COMPUTE_PATHS = frozenset(
{
@@ -46,10 +54,33 @@ HEAVY_COMPUTE_PATHS = frozenset(
_state_lock = threading.Lock()
_semaphore: threading.BoundedSemaphore | None = None
_retry_after_seconds = DEFAULT_RETRY_AFTER_SECONDS
_limit = DEFAULT_CONCURRENCY
_wait_seconds = DEFAULT_QUEUE_WAIT_SECONDS
_max_waiters = DEFAULT_MAX_WAITERS
_waiters = 0
_holders: list[dict] = []
_holder_seq = 0
class _HeldSlot:
"""One acquired heavy-compute permit, paired with the path that holds it."""
def __init__(self, semaphore: threading.BoundedSemaphore, holder_id: int) -> None:
self._semaphore = semaphore
self._holder_id = holder_id
self._released = False
def release(self) -> None:
with _state_lock:
if self._released:
return
self._released = True
_holders[:] = [item for item in _holders if item["id"] != self._holder_id]
self._semaphore.release()
class HeavyComputeBusy(RuntimeError):
"""No heavy-compute slot is free; callers must fail fast with HTTP 429."""
"""The short queue could not take this request; callers return HTTP 429."""
error_code = "ERR_COMPUTE_BUSY"
@@ -78,9 +109,11 @@ def reset_heavy_compute_gate(
*,
limit: int | None = None,
retry_after_seconds: int | None = None,
wait_seconds: float | None = None,
max_waiters: int | None = None,
) -> None:
"""Rebuild the process-wide semaphore. Tests must call this after env changes."""
global _semaphore, _retry_after_seconds
global _semaphore, _retry_after_seconds, _limit, _holders, _wait_seconds, _max_waiters, _waiters
resolved_limit = (
limit
if limit is not None
@@ -91,33 +124,129 @@ def reset_heavy_compute_gate(
if retry_after_seconds is not None
else _parse_positive_int(os.environ.get(ENV_RETRY_AFTER), DEFAULT_RETRY_AFTER_SECONDS)
)
resolved_wait = DEFAULT_QUEUE_WAIT_SECONDS if wait_seconds is None else max(0.0, float(wait_seconds))
resolved_waiters = DEFAULT_MAX_WAITERS if max_waiters is None else max(0, int(max_waiters))
with _state_lock:
_retry_after_seconds = resolved_retry
_limit = resolved_limit
_wait_seconds = resolved_wait
_max_waiters = resolved_waiters
_waiters = 0
_holders = []
_semaphore = threading.BoundedSemaphore(resolved_limit)
def _ensure_locked() -> tuple[threading.BoundedSemaphore, int]:
global _semaphore, _retry_after_seconds
global _semaphore, _retry_after_seconds, _limit
if _semaphore is None:
limit = _parse_positive_int(os.environ.get(ENV_CONCURRENCY), DEFAULT_CONCURRENCY)
_limit = _parse_positive_int(os.environ.get(ENV_CONCURRENCY), DEFAULT_CONCURRENCY)
_retry_after_seconds = _parse_positive_int(
os.environ.get(ENV_RETRY_AFTER), DEFAULT_RETRY_AFTER_SECONDS
)
_semaphore = threading.BoundedSemaphore(limit)
_semaphore = threading.BoundedSemaphore(_limit)
return _semaphore, _retry_after_seconds
def acquire_heavy_compute_slot(path: str) -> threading.BoundedSemaphore | None:
"""Acquire a slot for a gated path. Light paths return None. Fail-fast on saturation."""
if not is_heavy_compute_path(path):
return None
def _holder_view(now: float) -> list[dict]:
return [
{
"path": item["path"],
"held_seconds": round(max(now - item["started"], 0.0), 3),
}
for item in _holders
]
def _remember_locked(path: str, semaphore: threading.BoundedSemaphore) -> _HeldSlot:
"""Record a permit that this thread already holds. Caller owns ``_state_lock``."""
global _holder_seq
_holder_seq += 1
holder_id = _holder_seq
_holders.append({"id": holder_id, "path": path, "started": time.monotonic()})
return _HeldSlot(semaphore, holder_id)
def _reject_busy(path: str, retry_after: int) -> None:
with _state_lock:
snapshot = _holder_view(time.monotonic())
_LOG.warning(
"heavy_compute_busy rejected_path=%s holders=%s",
path,
snapshot,
)
raise HeavyComputeBusy(retry_after)
def _acquire_recorded(path: str) -> _HeldSlot:
"""Take one permit. Wait briefly when full; still raise once the queue cannot."""
global _waiters
with _state_lock:
semaphore, retry_after = _ensure_locked()
if not semaphore.acquire(blocking=False):
raise HeavyComputeBusy(retry_after)
return semaphore
if semaphore.acquire(blocking=False):
return _remember_locked(path, semaphore)
if _waiters >= _max_waiters:
queued = False
else:
_waiters += 1
queued = True
wait_seconds = _wait_seconds
if not queued:
_reject_busy(path, retry_after)
acquired = False
try:
acquired = bool(semaphore.acquire(blocking=True, timeout=wait_seconds))
finally:
with _state_lock:
_waiters -= 1
if not acquired:
_reject_busy(path, retry_after)
with _state_lock:
return _remember_locked(path, semaphore)
def release_heavy_compute_slot(slot: threading.BoundedSemaphore | None) -> None:
def heavy_compute_waiter_count() -> int:
"""Threads parked inside the short queue. Tests use this to avoid races."""
with _state_lock:
return _waiters
def acquire_heavy_compute_slot(path: str) -> _HeldSlot | None:
"""Acquire a slot for a gated path. Light paths return None."""
if not is_heavy_compute_path(path):
return None
return _acquire_recorded(path)
def _chart_overview_gate_enabled() -> bool:
return os.environ.get(ENV_GATE_CHART_OVERVIEW, "").strip() == "1"
def acquire_heavy_compute_slot_for_request(path: str, body: dict | None) -> _HeldSlot | None:
"""Path gate, plus the chart-overview switch and the warm-consultation exception.
``/api/chart`` stays out of ``HEAVY_COMPUTE_PATHS``. It takes a slot only when
``JYOTISH_GATE_CHART_OVERVIEW=1``, the body does not skip the VedAstro overview,
and the body is not an async enqueue. Default is off.
"""
payload = body if isinstance(body, dict) else {}
if path == "/api/consultation_workflow" and payload.get("consult_card_warm_only") is True:
return None
if path == "/api/chart" and _chart_overview_gate_enabled():
if payload.get("async") or payload.get("enqueue"):
return None
if payload.get("skip_vedastro_main_entry_overview"):
return None
return _acquire_recorded(path)
return acquire_heavy_compute_slot(path)
def heavy_compute_status() -> dict:
"""Process-local counts only. No request body and no user identity."""
with _state_lock:
_ensure_locked()
return {"limit": _limit, "in_use": len(_holders)}
def release_heavy_compute_slot(slot: _HeldSlot | None) -> None:
if slot is not None:
slot.release()
+6 -10
View File
@@ -122,13 +122,13 @@ except ModuleNotFoundError: # pragma: no cover - script execution path
try:
from scripts.api_heavy_compute_gate import (
HeavyComputeBusy,
acquire_heavy_compute_slot,
acquire_heavy_compute_slot_for_request,
release_heavy_compute_slot,
)
except ModuleNotFoundError: # pragma: no cover - script execution path
from api_heavy_compute_gate import (
HeavyComputeBusy,
acquire_heavy_compute_slot,
acquire_heavy_compute_slot_for_request,
release_heavy_compute_slot,
)
@@ -3357,14 +3357,10 @@ class JyotishAPIHandler(BaseHTTPRequestHandler, VedastroEvidenceMixin, SynastryM
try:
self._enforce_request_security(require_json=True)
body = self._read_json_body()
# Chart-save warm builds today's packet on the same route. It must
# not take a heavy-compute slot: the cap is 2, and a question asked
# right after save would otherwise come back 429.
warm_only = (
path == '/api/consultation_workflow'
and body.get('consult_card_warm_only') is True
)
compute_slot = None if warm_only else acquire_heavy_compute_slot(path)
# Warm chart-save must not take a slot (BUG-1302). Chart overview
# gating is off unless JYOTISH_GATE_CHART_OVERVIEW=1. Both live in
# the gate module so this file only keeps the call.
compute_slot = acquire_heavy_compute_slot_for_request(path, body)
if path == '/api/location/resolve':
city = str(body.get('city') or '').strip()
city_aliases = {'beijing': '北京', 'shanghai': '上海', 'guangzhou': '广州', 'shenzhen': '深圳'}
+26 -2
View File
@@ -10762,7 +10762,25 @@ def _planet_snapshot(planets, planet_name):
}
_oracle_progress_cache: dict = {}
def reset_oracle_progress_snapshot_cache():
"""Tests clear the static-file validation cache."""
_oracle_progress_cache.clear()
def _oracle_progress_cache_key(oracle_file):
try:
stat = oracle_file.stat()
except OSError:
return None
mtime = getattr(stat, 'st_mtime_ns', stat.st_mtime)
return (str(oracle_file.resolve()), mtime, stat.st_size)
def _oracle_progress_snapshot():
cache_key = None
try:
import json
import subprocess
@@ -10772,6 +10790,9 @@ def _oracle_progress_snapshot():
root = Path(__file__).resolve().parents[1]
oracle_file = root / 'references' / 'oracle' / 'dasha_shadbala_oracle_cases.json'
cache_key = _oracle_progress_cache_key(oracle_file)
if cache_key is not None and cache_key in _oracle_progress_cache:
return _oracle_progress_cache[cache_key]
with tempfile.TemporaryDirectory() as temp_dir:
queue_path = Path(temp_dir) / 'oracle-queue.json'
queue = subprocess.run(
@@ -10797,7 +10818,7 @@ def _oracle_progress_snapshot():
raise RuntimeError(validation.stderr.strip() or validation.stdout.strip())
report = json.loads(validation.stdout)
summary = report.get('summary', {})
return {
result = {
'scope': 'external_oracle_evidence_validation',
'collection_queue': 'external_oracle_collection_queue',
'total_packets': summary.get('total_packets', 0),
@@ -10809,7 +10830,7 @@ def _oracle_progress_snapshot():
'boundary': report.get('boundary', 'Dasha/Shadbala absolute values are not externally calibrated until enough packets pass validation.'),
}
except Exception as exc:
return {
result = {
'scope': 'external_oracle_evidence_validation',
'collection_queue': 'external_oracle_collection_queue',
'total_packets': 0,
@@ -10820,6 +10841,9 @@ def _oracle_progress_snapshot():
'promotion_rule': 'external_verified requires source_artifact, filled target values, and non-local-engine external evidence.',
'boundary': f'Oracle progress unavailable: {exc}',
}
if cache_key is not None:
_oracle_progress_cache[cache_key] = result
return result
# Combustion orbs of the full reading (Step 5 shadbala block). The Moon, Rahu
+291
View File
@@ -0,0 +1,291 @@
"""A/B for issue #64. A is this branch with the PyJHora short-circuit forced off
(the staging control flow: always spawn). B is the short-circuit on.
Chart skip/concurrency is measured on the real chart call. The gate schemes
O2/O3/O4 are simulated only and are not implemented.
Volatile fields dropped before the report byte compare:
generated_at, created_at, timestamp, elapsed_seconds, elapsed_ms, duration_ms, request_id.
"""
from __future__ import annotations
import argparse
import json
import os
import random
import statistics
import subprocess
import sys
import tempfile
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts"))
os.environ.setdefault("JYOTISH_API_CHART_CACHE_TTL_SECONDS", "0")
os.environ.setdefault("VEDASTRO_ENABLE_NETWORK", "0")
os.environ.setdefault("JYOTISH_GATE_CHART_OVERVIEW", "0")
import annual_pyjhora_replay as replay # noqa: E402
import tajika as tajika_top # noqa: E402
from scripts.jyotish_api_server import JyotishAPIHandler # noqa: E402
from scripts.jyotish_engine import reset_oracle_progress_snapshot_cache # noqa: E402
from scripts.professional_report_reference import build_professional_report_reference # noqa: E402
from scripts.tajika import reset_external_pyjhora_python_cache # noqa: E402
BIRTH = {
"year": 1990,
"month": 1,
"day": 1,
"hour": 12,
"minute": 0,
"second": 0,
"lat": 28.6139,
"lon": 77.2090,
"tz": 5.5,
"ayanamsa": "lahiri",
"node_mode": "true",
}
PARTNER = {**BIRTH, "year": 1992, "day": 2}
VOLATILE = {"generated_at", "created_at", "timestamp", "elapsed_seconds", "elapsed_ms", "duration_ms", "request_id"}
class _Counter:
def __init__(self) -> None:
self.n = 0
self._real = subprocess.run
self._lock = threading.Lock()
def install(self) -> None:
def _run(*args, **kwargs):
with self._lock:
self.n += 1
return self._real(*args, **kwargs)
subprocess.run = _run # type: ignore[assignment]
for module in list(sys.modules.values()):
bound = getattr(module, "subprocess", None)
if getattr(bound, "run", None) is self._real:
bound.run = _run # type: ignore[union-attr]
def reset(self) -> None:
with self._lock:
self.n = 0
def _median(values: list[float]) -> float:
return statistics.median(values) if values else 0.0
def _strip(value):
if isinstance(value, dict):
return {key: _strip(item) for key, item in value.items() if key not in VOLATILE}
if isinstance(value, list):
return [_strip(item) for item in value]
return value
def _canon(value) -> str:
return json.dumps(_strip(value), sort_keys=True, ensure_ascii=True, default=str)
def _chart_summary(chart: dict) -> dict:
planets = chart.get("planets") if isinstance(chart, dict) else None
moon = planets.get("Moon") if isinstance(planets, dict) else None
lon = None
if isinstance(moon, dict):
lon = moon.get("lon", moon.get("longitude"))
return {
"ascendant": chart.get("ascendant") if isinstance(chart, dict) else None,
"planets": planets,
"moon_longitude": lon,
}
def _one_chart(body: dict) -> tuple[float, dict]:
started = time.perf_counter()
chart = object.__new__(JyotishAPIHandler)._compute_chart_sync(body)
return time.perf_counter() - started, chart
def _measure_charts(counter: _Counter) -> dict:
serial_times = []
concurrent_times = []
serial_subs = []
concurrent_subs = []
same = True
for _ in range(3):
counter.reset()
started = time.perf_counter()
_, self_a = _one_chart(BIRTH)
_, partner_a = _one_chart(PARTNER)
serial_times.append(time.perf_counter() - started)
serial_subs.append(counter.n)
counter.reset()
started = time.perf_counter()
with ThreadPoolExecutor(max_workers=2) as pool:
self_b = pool.submit(_one_chart, {**BIRTH, "skip_vedastro_main_entry_overview": True})
partner_b = pool.submit(_one_chart, {**PARTNER, "skip_vedastro_main_entry_overview": True})
_, self_chart = self_b.result()
_, partner_chart = partner_b.result()
concurrent_times.append(time.perf_counter() - started)
concurrent_subs.append(counter.n)
same = same and _canon(_chart_summary(self_a)) == _canon(_chart_summary(self_chart))
same = same and _canon(_chart_summary(partner_a)) == _canon(_chart_summary(partner_chart))
return {
"serial_seconds_median": round(_median(serial_times), 4),
"concurrent_skip_seconds_median": round(_median(concurrent_times), 4),
"serial_subprocesses_median": _median(serial_subs),
"concurrent_skip_subprocesses_median": _median(concurrent_subs),
"summary_bytes_match": same,
}
def _measure_report(counter: _Counter, *, spawn: bool) -> dict:
times = []
subs = []
payload = None
body = {**BIRTH, "format": "markdown", "edition": "full_data", "languages": ["zh", "en"], "include_fact_tables": True}
for _ in range(3):
replay.reset_jhora_import_cache()
if spawn:
replay._jhora_spec_cache = True
reset_oracle_progress_snapshot_cache()
reset_external_pyjhora_python_cache()
tajika_top.reset_external_pyjhora_python_cache()
counter.reset()
started = time.perf_counter()
payload = build_professional_report_reference(
object.__new__(JyotishAPIHandler),
body,
)
times.append(time.perf_counter() - started)
subs.append(counter.n)
return {
"seconds_median": round(_median(times), 4),
"subprocesses_median": _median(subs),
"payload": payload,
}
def _simulate() -> list[dict]:
"""One fixed minute. O2 adds a second pool and is the only scheme that touches BUG-717."""
rng = random.Random(64)
arrivals = [("report", index * 30.0, 13.0) for index in range(2)]
arrivals += [("consult", rng.random() * 60.0, 0.2) for _ in range(20)]
arrivals += [("score", rng.random() * 60.0, 8.0) for _ in range(6)]
arrivals.sort(key=lambda item: item[1])
def run(mode: str) -> dict:
heavy: list[tuple[float, str]] = []
light: list[float] = []
rejected = {"consult": 0, "report": 0, "score": 0}
accepted = {"consult": 0, "report": 0, "score": 0}
peak = 0
def occupy(at: float) -> tuple[int, int]:
running = [item for item in heavy if item[0] > at]
lights = [item for item in light if item > at]
return len(running), len(lights)
for kind, arrival, duration in arrivals:
heavy_n, light_n = occupy(arrival)
heavy = [item for item in heavy if item[0] > arrival]
light = [item for item in light if item > arrival]
placed = False
stamp = arrival
if mode == "O2" and kind == "consult":
if light_n < 2:
light.append(arrival + duration)
placed = True
elif mode == "O3" and kind == "consult":
start = arrival
while start <= arrival + 1.5:
heavy_n, _light_n = occupy(start)
if heavy_n < 2:
heavy.append((start + duration, kind))
placed = True
stamp = start
break
nxt = min(item[0] for item in heavy)
if nxt > arrival + 1.5:
break
start = nxt
elif mode == "O4" and kind != "consult":
heavy_jobs = sum(1 for _end, slot in heavy if slot != "consult")
if heavy_jobs < 1 and heavy_n < 2:
heavy.append((arrival + duration, kind))
placed = True
else:
if heavy_n < 2:
heavy.append((arrival + duration, kind))
placed = True
if placed:
accepted[kind] += 1
else:
rejected[kind] += 1
running, lights = occupy(stamp)
peak = max(peak, running + lights)
def rate(kind: str) -> str:
total = accepted[kind] + rejected[kind]
return "0" if total == 0 else f"{rejected[kind] / total:.2f}"
return {
"scheme": mode,
"consult_429": rate("consult"),
"report_429": rate("report"),
"score_429": rate("score"),
"peak": peak,
"touches_bug_717": mode == "O2",
}
return [run(mode) for mode in ("O1", "O2", "O3", "O4")]
def resolve_record_path(explicit: str | None) -> Path:
"""Write the A/B JSON only where asked. The committed note stays untouched."""
if explicit:
return Path(explicit)
directory = Path(tempfile.mkdtemp(prefix="api-chart-subprocess-ab-"))
return directory / "api-chart-subprocess-gate-ab-20261013.json"
def main(argv: list[str] | None = None) -> None:
parser = argparse.ArgumentParser(description="A/B for issue #64 chart subprocess gate")
parser.add_argument(
"--out",
default=None,
help="Write the JSON record to this path. Omit to use a fresh temp directory, not the committed docs/testing file.",
)
args = parser.parse_args(argv)
counter = _Counter()
counter.install()
charts = _measure_charts(counter)
forced = _measure_report(counter, spawn=True)
native = _measure_report(counter, spawn=False)
report_match = _canon(forced["payload"]) == _canon(native["payload"])
schemes = _simulate()
out = {
"charts": charts,
"report_forced_subprocess": {key: value for key, value in forced.items() if key != "payload"},
"report_short_circuit": {key: value for key, value in native.items() if key != "payload"},
"report_bytes_match": report_match,
"schemes": schemes,
"jhora_installed": False,
}
text = json.dumps(out, ensure_ascii=False, indent=2)
dest = resolve_record_path(args.out)
dest.parent.mkdir(parents=True, exist_ok=True)
dest.write_text(text + "\n", encoding="utf-8")
print(dest, file=sys.stderr)
print(text)
if __name__ == "__main__":
main()
+21
View File
@@ -1501,7 +1501,28 @@ print(json.dumps({str(k): float(v) for k, v in dict(scores).items()}))
return None
_external_pyjhora_python_cached = False
_external_pyjhora_python_value: Optional[str] = None
def reset_external_pyjhora_python_cache() -> None:
"""Tests and a fresh process both start with no remembered interpreter."""
global _external_pyjhora_python_cached, _external_pyjhora_python_value
_external_pyjhora_python_cached = False
_external_pyjhora_python_value = None
def _resolve_external_pyjhora_python() -> Optional[str]:
"""Remember the lookup, including "not installed", for this process."""
global _external_pyjhora_python_cached, _external_pyjhora_python_value
if _external_pyjhora_python_cached:
return _external_pyjhora_python_value
_external_pyjhora_python_value = _lookup_external_pyjhora_python()
_external_pyjhora_python_cached = True
return _external_pyjhora_python_value
def _lookup_external_pyjhora_python() -> Optional[str]:
candidates = [
os.environ.get('JYOTISH_PYJHORA_PYTHON'),
shutil.which('python3.14'),
+6
View File
@@ -0,0 +1,6 @@
Generated 2026-10-15 by calling the isolated workers with PyJHora absent
(importlib.util.find_spec("jhora") is None on the Windows venv):
.venv\Scripts\python -c "import json; from scripts.annual_pyjhora_replay import _isolated_tajaka_worker, _isolated_annual_worker; req={'birth_julian_day':2447893.0,'place':{'name':'Delhi','latitude':28.6139,'longitude':77.209,'timezone':5.5,'elevation':0},'age':30,'settings':{'ayanamsa':'LAHIRI','node_mode':'mean'}}; print(json.dumps(_isolated_tajaka_worker(req), sort_keys=True)); print(json.dumps(_isolated_annual_worker(req), sort_keys=True, default=str))"
The parent adds request_hash after this body. The short-circuit must match this body.
@@ -0,0 +1,41 @@
{
"annual_chart_snapshot": {
"reason": "pyjhora_annual_adapter_setup_failed:ModuleNotFoundError",
"status": "blocked"
},
"annual_chart_speed_snapshot": {
"reason": "pyjhora_annual_adapter_setup_failed:ModuleNotFoundError",
"status": "blocked"
},
"effective_ayanamsa": null,
"evidence_scope": "pyjhora_behavior_only",
"mudda_dasha": {
"reason": "pyjhora_annual_adapter_setup_failed:ModuleNotFoundError",
"status": "blocked"
},
"node_mode": {
"requested": "mean",
"status": "unsupported_by_replay_adapter"
},
"parity_status": "not_multiengine_parity",
"patyayini_dasha": {
"reason": "pyjhora_annual_adapter_setup_failed:ModuleNotFoundError",
"status": "blocked"
},
"sahams": {
"reason": "pyjhora_annual_adapter_setup_failed:ModuleNotFoundError",
"status": "blocked"
},
"settings": {
"ayanamsa": "LAHIRI",
"node_mode": "mean"
},
"solar_return_boundary": {
"reason": "pyjhora_annual_adapter_setup_failed:ModuleNotFoundError",
"status": "blocked"
},
"tajika_yogas": {
"reason": "pyjhora_annual_adapter_setup_failed:ModuleNotFoundError",
"status": "blocked"
}
}
@@ -0,0 +1,12 @@
{
"node_mode": {
"requested": "mean",
"status": "unsupported_by_replay_adapter"
},
"reason": "pyjhora_ayanamsa_configuration_failed:ModuleNotFoundError",
"settings": {
"ayanamsa": "LAHIRI",
"node_mode": "mean"
},
"status": "blocked"
}
+203
View File
@@ -0,0 +1,203 @@
"""Issue #64: skip useless report subprocesses and observe the heavy-compute gate."""
from __future__ import annotations
import json
import logging
from pathlib import Path
import pytest
from scripts import api_heavy_compute_gate as gate
from scripts.annual_pyjhora_replay import (
_run_isolated_annual_replay,
_run_isolated_tajaka_year_lord,
reset_jhora_import_cache,
)
from scripts.jyotish_engine import (
_oracle_progress_snapshot,
reset_oracle_progress_snapshot_cache,
)
from scripts.tajika import (
_resolve_external_pyjhora_python,
reset_external_pyjhora_python_cache,
)
ROOT = Path(__file__).resolve().parents[1]
FIXTURES = ROOT / "tests" / "fixtures" / "api-chart-subprocess-gate"
class _Place:
name = "Delhi"
latitude = 28.6139
longitude = 77.209
timezone = 5.5
elevation = 0
_PROFILE = {"ayanamsa": "LAHIRI", "node_mode": "mean"}
def _without_hash(payload: dict) -> dict:
body = dict(payload)
body.pop("request_hash", None)
return body
def test_no_jhora_tajaka_skips_subprocess_and_matches_golden(monkeypatch) -> None:
reset_jhora_import_cache()
monkeypatch.setattr("scripts.annual_pyjhora_replay._jhora_importable", lambda: False)
def _forbid_subprocess(*_args, **_kwargs):
raise AssertionError("subprocess.run must not run when jhora cannot be imported")
monkeypatch.setattr("scripts.annual_pyjhora_replay.subprocess.run", _forbid_subprocess)
result = _run_isolated_tajaka_year_lord(_PROFILE, 2447893.0, _Place(), 30)
golden = json.loads((FIXTURES / "tajaka-no-jhora.json").read_text(encoding="utf-8"))
assert result["request_hash"]
assert _without_hash(result) == golden
def test_no_jhora_annual_skips_subprocess_and_matches_golden(monkeypatch) -> None:
reset_jhora_import_cache()
monkeypatch.setattr("scripts.annual_pyjhora_replay._jhora_importable", lambda: False)
def _forbid_subprocess(*_args, **_kwargs):
raise AssertionError("subprocess.run must not run when jhora cannot be imported")
monkeypatch.setattr("scripts.annual_pyjhora_replay.subprocess.run", _forbid_subprocess)
result = _run_isolated_annual_replay(_PROFILE, 2447893.0, _Place(), 30)
golden = json.loads((FIXTURES / "annual-no-jhora.json").read_text(encoding="utf-8"))
assert result["request_hash"]
assert _without_hash(result) == golden
def test_jhora_available_still_uses_subprocess(monkeypatch) -> None:
reset_jhora_import_cache()
monkeypatch.setattr("scripts.annual_pyjhora_replay._jhora_importable", lambda: True)
calls = {"n": 0}
class _Completed:
returncode = 0
stdout = json.dumps({"status": "partial_verified", "settings": {}})
def _fake_run(*_args, **_kwargs):
calls["n"] += 1
return _Completed()
monkeypatch.setattr("scripts.annual_pyjhora_replay.subprocess.run", _fake_run)
tajaka = _run_isolated_tajaka_year_lord(_PROFILE, 2447893.0, _Place(), 30)
annual = _run_isolated_annual_replay(_PROFILE, 2447893.0, _Place(), 30)
assert calls["n"] == 2
assert tajaka["request_hash"]
assert annual["request_hash"]
def test_pyjhora_interpreter_lookup_is_cached(monkeypatch) -> None:
reset_external_pyjhora_python_cache()
calls = {"n": 0}
def _missing():
calls["n"] += 1
return None
monkeypatch.setattr("scripts.tajika._lookup_external_pyjhora_python", _missing)
assert _resolve_external_pyjhora_python() is None
assert _resolve_external_pyjhora_python() is None
assert calls["n"] == 1
def test_oracle_progress_snapshot_reuses_until_the_file_changes(monkeypatch) -> None:
reset_oracle_progress_snapshot_cache()
calls = {"n": 0}
real_run = __import__("subprocess").run
def _counting_run(*args, **kwargs):
calls["n"] += 1
return real_run(*args, **kwargs)
monkeypatch.setattr("subprocess.run", _counting_run)
first = _oracle_progress_snapshot()
after_first = calls["n"]
assert after_first >= 1
second = _oracle_progress_snapshot()
assert calls["n"] == after_first
assert second == first
oracle = ROOT / "references" / "oracle" / "dasha_shadbala_oracle_cases.json"
stat = oracle.stat()
later = stat.st_mtime + 5
try:
oracle.touch()
import os
os.utime(oracle, (later, later))
third = _oracle_progress_snapshot()
assert calls["n"] > after_first
assert third["scope"] == first["scope"]
finally:
os_utime = __import__("os").utime
os_utime(oracle, (stat.st_atime, stat.st_mtime))
reset_oracle_progress_snapshot_cache()
def test_chart_overview_gate_is_off_by_default(monkeypatch) -> None:
monkeypatch.delenv(gate.ENV_GATE_CHART_OVERVIEW, raising=False)
gate.reset_heavy_compute_gate(limit=1)
held = gate.acquire_heavy_compute_slot("/api/high_rigor_workflow")
assert held is not None
assert gate.acquire_heavy_compute_slot_for_request("/api/chart", {"year": 1990}) is None
gate.release_heavy_compute_slot(held)
assert gate.heavy_compute_status()["in_use"] == 0
gate.reset_heavy_compute_gate()
def test_chart_overview_gate_only_sync_unskipped_requests(monkeypatch) -> None:
monkeypatch.setenv(gate.ENV_GATE_CHART_OVERVIEW, "1")
# 原值:reset(limit=1) 后第二次立刻 HeavyComputeBusy
# 新值:wait_seconds=0
# 原因:默认改为短排队;本测试仍锁「名额被占时同步排盘被拒绝」,等待行为不在这里
gate.reset_heavy_compute_gate(limit=1, wait_seconds=0)
body = {"year": 1990, "lat": 28.6, "lon": 77.2}
slot = gate.acquire_heavy_compute_slot_for_request("/api/chart", body)
assert slot is not None
assert gate.heavy_compute_status() == {"limit": 1, "in_use": 1}
with pytest.raises(gate.HeavyComputeBusy):
gate.acquire_heavy_compute_slot_for_request("/api/chart", {"year": 1991})
gate.release_heavy_compute_slot(slot)
assert gate.heavy_compute_status()["in_use"] == 0
assert gate.acquire_heavy_compute_slot_for_request(
"/api/chart", {"skip_vedastro_main_entry_overview": True}
) is None
assert gate.acquire_heavy_compute_slot_for_request("/api/chart", {"async": True}) is None
assert gate.acquire_heavy_compute_slot_for_request("/api/chart", {"enqueue": 1}) is None
assert "/api/chart" not in gate.HEAVY_COMPUTE_PATHS
gate.reset_heavy_compute_gate()
def test_busy_log_names_holders_and_not_the_body(caplog) -> None:
# 原值:reset(limit=1) 后拒绝日志立刻出现
# 新值:wait_seconds=0
# 原因:默认会先等 8 秒;本测试仍锁日志只有路径、没有请求正文
gate.reset_heavy_compute_gate(limit=1, wait_seconds=0)
held = gate.acquire_heavy_compute_slot("/api/professional_report_reference")
assert held is not None
secret_body = {"birth_date": "1990-01-01", "name": "must-not-appear"}
with caplog.at_level(logging.WARNING, logger="jyotish.heavy_compute"):
with pytest.raises(gate.HeavyComputeBusy):
gate.acquire_heavy_compute_slot_for_request("/api/professional_report_reference", secret_body)
text = caplog.text
assert "/api/professional_report_reference" in text
assert "1990-01-01" not in text
assert "must-not-appear" not in text
gate.release_heavy_compute_slot(held)
assert gate.heavy_compute_status()["in_use"] == 0
gate.reset_heavy_compute_gate()
def test_warm_consultation_still_skips_the_slot() -> None:
gate.reset_heavy_compute_gate(limit=1)
assert gate.acquire_heavy_compute_slot_for_request(
"/api/consultation_workflow", {"consult_card_warm_only": True}
) is None
assert gate.heavy_compute_status()["in_use"] == 0
gate.reset_heavy_compute_gate()
+89 -2
View File
@@ -4,6 +4,7 @@ from __future__ import annotations
import json
import threading
import time
from io import BytesIO
import pytest
@@ -14,6 +15,7 @@ from scripts.api_heavy_compute_gate import (
HEAVY_COMPUTE_PATHS,
HeavyComputeBusy,
acquire_heavy_compute_slot,
heavy_compute_waiter_count,
is_heavy_compute_path,
release_heavy_compute_slot,
reset_heavy_compute_gate,
@@ -93,7 +95,10 @@ def limit_one_gate(monkeypatch):
monkeypatch.setenv("JYOTISH_HEAVY_COMPUTE_CONCURRENCY", "1")
monkeypatch.setenv("JYOTISH_HEAVY_COMPUTE_RETRY_AFTER_SECONDS", "2")
monkeypatch.setenv("JYOTISH_API_RATE_LIMIT_PER_MINUTE", "0")
reset_heavy_compute_gate()
# 原值:reset() 使用默认「满了立刻拒绝」
# 新值:wait_seconds=0,本夹具里的 429 断言仍是立刻拒绝
# 原因:产品改为最多等 8 秒;排队成功、等待者满、超时另有测试
reset_heavy_compute_gate(wait_seconds=0)
yield
monkeypatch.delenv("JYOTISH_HEAVY_COMPUTE_CONCURRENCY", raising=False)
monkeypatch.delenv("JYOTISH_HEAVY_COMPUTE_RETRY_AFTER_SECONDS", raising=False)
@@ -119,7 +124,10 @@ def test_health_and_light_paths_are_not_gated() -> None:
def test_acquire_fail_fast_then_succeeds_after_release() -> None:
reset_heavy_compute_gate(limit=1, retry_after_seconds=3)
# 原值:reset(limit=1, retry_after_seconds=3) 后第二路立刻 HeavyComputeBusy
# 新值:多传 wait_seconds=0
# 原因:默认改为短排队;本测试仍锁 Retry-After=3 与 ERR_COMPUTE_BUSY,不锁「禁止等待」
reset_heavy_compute_gate(limit=1, retry_after_seconds=3, wait_seconds=0)
first = acquire_heavy_compute_slot("/api/high_rigor_workflow")
assert first is not None
with pytest.raises(HeavyComputeBusy) as caught:
@@ -248,8 +256,87 @@ def test_warm_consultation_does_not_consume_the_user_slot(limit_one_gate, monkey
assert "packet" not in warm_box["payload"]
def _wait_until(predicate, timeout: float = 2.0) -> bool:
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if predicate():
return True
time.sleep(0.01)
return False
def test_third_request_gets_the_slot_after_one_holder_releases() -> None:
reset_heavy_compute_gate(limit=2, retry_after_seconds=2, wait_seconds=2, max_waiters=4)
first = acquire_heavy_compute_slot("/api/consultation_workflow")
second = acquire_heavy_compute_slot("/api/rectification/v5/score")
assert first is not None and second is not None
found: dict = {}
def _third() -> None:
found["slot"] = acquire_heavy_compute_slot("/api/consultation_workflow")
worker = threading.Thread(target=_third)
worker.start()
assert _wait_until(lambda: heavy_compute_waiter_count() == 1)
release_heavy_compute_slot(first)
worker.join(timeout=3)
assert not worker.is_alive()
assert found["slot"] is not None
release_heavy_compute_slot(second)
release_heavy_compute_slot(found["slot"])
reset_heavy_compute_gate()
def test_waiter_overflow_rejects_immediately() -> None:
reset_heavy_compute_gate(limit=1, retry_after_seconds=2, wait_seconds=2, max_waiters=1)
held = acquire_heavy_compute_slot("/api/high_rigor_workflow")
assert held is not None
parked: dict = {}
def _wait() -> None:
try:
parked["slot"] = acquire_heavy_compute_slot("/api/consultation_workflow")
except HeavyComputeBusy as exc:
parked["error"] = exc
waiter = threading.Thread(target=_wait)
waiter.start()
assert _wait_until(lambda: heavy_compute_waiter_count() == 1)
started = time.monotonic()
with pytest.raises(HeavyComputeBusy) as caught:
acquire_heavy_compute_slot("/api/rectification/v5/score")
assert time.monotonic() - started < 0.3
assert caught.value.retry_after_seconds == 2
assert caught.value.error_code == "ERR_COMPUTE_BUSY"
release_heavy_compute_slot(held)
waiter.join(timeout=3)
assert not waiter.is_alive()
assert parked.get("slot") is not None
release_heavy_compute_slot(parked["slot"])
reset_heavy_compute_gate()
def test_queue_timeout_still_returns_busy() -> None:
reset_heavy_compute_gate(limit=1, retry_after_seconds=2, wait_seconds=0.2, max_waiters=4)
held = acquire_heavy_compute_slot("/api/professional_report_reference")
assert held is not None
started = time.monotonic()
with pytest.raises(HeavyComputeBusy) as caught:
acquire_heavy_compute_slot("/api/consultation_workflow")
elapsed = time.monotonic() - started
assert 0.1 <= elapsed < 2.0
assert caught.value.retry_after_seconds == 2
assert caught.value.error_code == "ERR_COMPUTE_BUSY"
light = acquire_heavy_compute_slot("/api/health")
assert light is None
release_heavy_compute_slot(held)
reset_heavy_compute_gate()
def test_quality_gate_runs_heavy_compute_gate() -> None:
from scripts.run_quality_gate import CORE_PYTEST_TARGETS
assert "tests/test_api_heavy_compute_gate.py" in CORE_PYTEST_TARGETS
assert gate.DEFAULT_CONCURRENCY == 2
assert gate.DEFAULT_QUEUE_WAIT_SECONDS == 8
assert gate.DEFAULT_MAX_WAITERS == 4
+5 -2
View File
@@ -298,8 +298,11 @@ def test_synastry_route_orchestrates_python_chart_and_ashtakoot() -> None:
route = SYNASTRY_ROUTE.read_text(encoding="utf-8")
for token in (
'const apiBase = process.env.JYOTISH_API_BASE ?? "http://127.0.0.1:5200"',
'postPython("/api/chart", selfPayload)',
'postPython("/api/chart", partnerPayload)',
# 原值: postPython("/api/chart", selfPayload) 与 postPython("/api/chart", partnerPayload)
# 新值: 两次调用都展开对应 payload,并带 skip_vedastro_main_entry_overview: true
# 原因: 合盘改为同时排本人和对方两张盘,并关掉用不上的外部概览。断言仍锁住这两次 /api/chart,同时锁住这个开关,不放宽。
'postPython("/api/chart", { ...selfPayload, skip_vedastro_main_entry_overview: true })',
'postPython("/api/chart", { ...partnerPayload, skip_vedastro_main_entry_overview: true })',
'postPython("/api/varga_full"',
'postPython("/api/synastry"',
"moonLongitude(selfChart)",