Compare commits

..
Author SHA1 Message Date
jesse-ux 149e1ec4c3 fix(consult): drop traces, budget checkpoints, silent summary inherit
Independent Staging Quality Gate / validate (push) Canceled after 3m8s
Independent Staging Quality Gate / publish (push) Canceled after 0s
BUG-729: dropped history rounds leave an omission marker in the model-visible summary slot.
BUG-730: checkpoint threshold is 0.4 of historyBudgetChars (128k still 16,000).
BUG-731: session_full new chat copies owned context_summary on the server; clients send only continued_from_session_id.
2026-09-16 07:38:36 +08:00
jesse-ux e61535f464 fix(consultation): 外网证据按盘+日期缓存,超时取消前台任务
Independent Staging Quality Gate / validate (push) Canceled after 2m21s
Independent Staging Quality Gate / publish (push) Canceled after 0s
BUG-727:同日 VedAstro 快照零等待,跨日先用旧的并后台刷新;join 超时必须 cancel,budget 不超过 2×join。BUG-728:western_evidence_packet 无读取点,默认不再进咨询响应。jyotish_api_server.py 未增长(11334→11291)。
2026-09-16 07:36:09 +08:00
29 changed files with 1523 additions and 118 deletions
+8
View File
@@ -1,5 +1,13 @@
# 印度占星 Skill 更新日志
## 2026-09-16 — 对话写满后开新对话会静默带上之前的摘要
普通咨询写满、点「开新对话」之后,服务端会把上一场的会话摘要拷进新对话,界面上看不出差别。进不了窗口的更早几轮,模型会看到省略说明,不再当成没发生过。Skill 版本不变。
## 2026-09-16 — 同一天再问同一张盘,不再重复等外部证据
普通聊天同一张盘、同一天里再问,外部证据直接用已经取到的,不再每轮等外网。隔了一天会先用最多七天内的旧证据马上回答,后台再刷新;「深入看今日」仍然只要当天的。西洋盘整包不再塞进每一轮的回答里,压缩后的西洋层还在。Skill 版本不变。
## 2026-09-16 — 校正时系统没接上会直说,记下经历后没重算也会说明
生时校正里,如果这边没判断出你在答哪一问,会请你把刚才那句再发一次,而不是说你没说清。记下经历后如果这次没能重新比较候选,会告诉你经历已经记下、稍后再比,不会假装范围已经在收。Skill 版本不变。
+80 -1
View File
@@ -11285,7 +11285,6 @@
- 相关记录:BUG-473、BUG-249、BUG-250、BUG-260、BUG-617
- 复发自:BUG-473(影响面含本文件,但两条防复发只约束正在流的那一行)
- 修复版本:待发布
## BUG-726 | 一轮校正把同一份 Case 档案从数据库取多次
- 状态:resolved
@@ -11301,3 +11300,83 @@
- 相关记录:BUG-176
- 复发自:无
- 修复版本:待发布
## BUG-727 | 普通聊天每轮每域同步等外网,超时不取消导致线程池饱和
- 状态:resolved
- 首次发现:2026-09-15
- 最近更新:2026-09-16
- 影响面:`POST /api/consultation_workflow` 前台路径、`execute_consultation_workflow`、`vedastro_gateway`、前台 VedAstro 线程池
- 用户现象:普通聊天每发一条、每个问题域都要等外网;本地排盘只占约 3%。超时后技法表 VedAstro 云状态仍是 blocked,之后每一轮继续白等约 1.5 秒。
- 触发条件:网页咨询 `defer_optional_external_evidence: true`;同一张盘一天内多轮提问;前台线程池默认 2 个 worker。
- 根因:外网证据本就按「盘 + UTC 日期」组织,却没有按这个键缓存,每轮重新 TLS 握手打 `api.vedastro.org`。`_join_foreground_vedastro` 超时返回 `official_blocked` 但不 `cancel()` 后台任务,任务继续跑满预算;join 1.5 s、budget 8 s、workers 2 三个数不在同一处约束,平均每 4 秒一轮就把池子占死。
- 修复:新增独立模块 `scripts/vedastro_snapshot_cache.py`(目录与 key 函数都与 `_api_chart_cache` 分开)。同日命中直接用、零等待、不 submit;1–7 天前先用旧的并后台刷新;`entrypoint=daily_starlanguage` 只接受当天。超时必须 `future.cancel()`;join / budget / workers 成组声明,且 `budget ≤ 2 × join`。本单不是推翻 BUG-301:前台仍取官方层,只是用缓存去掉重复等待。
- 验证:`tests/test_vedastro_snapshot_cache.py`(同日第二次零网关调用、跨日 stale+刷新不等待、daily_starlanguage 不吃昨天、缓存目录≠chart 缓存、N+1 等待不超过 join、三常数同块且 budget 比例、超时取消排队任务);既有前台赶上/超时降级回归。
- 防复发:前台外部证据必须有「盘 + 日期」级缓存;任何有界等待都必须同时取消它等待的后台任务。
- 相关记录:BUG-161、BUG-301、BUG-718
- 复发自:无(BUG-301 是故意把 VedAstro 放回前台,本单用缓存同时满足 161 与 301)
- 修复版本:待发布
## BUG-728 | `western_evidence_packet` 无人读却每轮每域传约 122 KB
- 状态:resolved
- 首次发现:2026-09-15
- 最近更新:2026-09-16
- 影响面:`POST /api/consultation_workflow` 响应体、`consumer_context.western_spectrum`
- 用户现象:一轮咨询响应约 52 万字符,其中西洋证据整包约 122 KB;前端解析后丢掉。三个域就是约 366 KB 无效 JSON。
- 触发条件:普通聊天每个问题域调用 `/api/consultation_workflow`。
- 根因:西洋盘计算是 must-use 层,但整包被无条件放进前台响应。`frontend/src` 零读取点;被读的是 `consumer_context.western_spectrum` 压缩投影。
- 修复:默认不把 `western_evidence_packet` 放进咨询工作流响应,计算与 `western_spectrum` 仍保留。MCP、高严谨、显式 `include_western_evidence_packet` / `western_oracle_payload` 仍返回整包。`consultationWorkflowResponseSchema` 为 `.passthrough()`,去掉该键仍能解析。
- 验证:`tests/test_vedastro_snapshot_cache.py` 默认省略、显式请求保留、oracle payload 仍返回;`frontend/tests/consultation-workflow-contract.test.ts` 无该键仍 `safeParse` 成功。
- 防复发:工作流响应新增大字段前必须有读取点;没有读取点的字段不得进入前台响应。
- 相关记录:BUG-727
- 复发自:无
- 修复版本:待发布
## BUG-729 | 咨询历史丢掉整轮时模型看不见任何痕迹
- 状态:resolved
- 首次发现:2026-09-15
- 最近更新:2026-09-16
- 影响面:`consultationHistoryWindow`、`consultationUserTurnContent`、`POST /api/consult`
- 用户现象:历史超过模型预算后,追问「刚才你说的那个时间」时模型当成从没说过。单条超长会写「省略 N 字」,整轮被丢掉时什么都不留。
- 触发条件:普通咨询多轮之后,尾巴字符数超过当前模型的历史预算,窗口从最旧整条丢弃。
- 根因:`droppedCount` 算出来了,`route.ts` 只取 `.tail`。BUG-555 的防复发只写了「不得再按固定 12 条 × 头部截断静默砍结论」,整轮丢弃不在字面里,所以没拦住。
- 修复:`droppedCount > 0` 时在摘要槽追加与 `omissionMarker` 同风格的说明。有摘要时写「更早的 N 轮问答已并入上面的会话摘要」;没有摘要时诚实写结论尚未并入。`droppedCount === 0` 不加这句话。
- 验证:超预算历史的模型可见文本含丢弃说明且轮数等于 `droppedCount`;零丢弃不加这句话;源码合同断言 `route.ts` 读取 `historyWindow.droppedCount`。
- 防复发:咨询历史任何形式的丢弃(截断单条、丢整轮)都必须在模型可见文本里留痕。
- 相关记录:BUG-555
- 复发自:BUG-555(防复发只覆盖头部截断)
- 修复版本:待发布
## BUG-730 | 写摘要的阈值写死 16,000,追不上按窗口算出的历史预算
- 状态:resolved
- 首次发现:2026-09-15
- 最近更新:2026-09-16
- 影响面:`historyBudgetChars`、`consultationHistoryCheckpointChars`、`shouldCheckpoint`、`checkpointSessionContextSummary`
- 用户现象:后台上架中等上下文窗口的模型后,每轮静默丢掉最老的几轮问答,摘要却还没开始写。
- 触发条件:模型 `context_window` 低于约 70,667(例如 64k / 32k)。历史预算已经小于写死的 16,000 字阈值。
- 根因:`shouldCheckpoint` 用常量 16,000,`consultationHistoryWindow` 用 `historyBudgetChars()`。两个数各写各的,没有「阈值必须低于预算」的断言。
- 修复:阈值改为预算的 0.4(默认 128k 窗口仍是 16,000)。运行时断言阈值 < 预算。检查点把会话模型的 `contextWindow` 传进去。
- 验证:表驱动覆盖 200k / 128k / 64k / 32k / null,逐条 `checkpoint < budget`;128k 仍为 16,000。
- 防复发:触发摘要的阈值必须由历史预算派生,并由一条断言钉死「阈值 < 预算」在所有合法上下文窗口下成立。
- 相关记录:BUG-555
- 复发自:BUG-555(检查点阈值与窗口预算未绑在一起)
- 修复版本:待发布
## BUG-731 | 对话写满后开新对话不继承服务端已有的会话摘要
- 状态:resolved
- 首次发现:2026-09-15
- 最近更新:2026-09-16
- 影响面:`chatSessionCreateSchema`、`POST /api/sessions`、`continueInNewChat`、`startNewChat`
- 用户现象:这段对话已写满、点「开新对话」之后,模型对刚才的结论一无所知,用户被要求从零开始。
- 触发条件:`append_consultation_question` 返回 `session_full`,客户端走「开新对话」。
- 根因:新会话是空的。`context_summary` 不在列表 GET 列里,创建合同也不接收它。即便前端想带,也没有合法入口。
- 修复:创建合同增加可选 `continued_from_session_id`(保持 `.strict()`,不加 `context_summary`)。服务端按当前用户读源会话摘要,读到才写入新行;读不到、不属于该用户、或为空都静默跳过,仍返回 201。写满出口把当前会话 id 带进创建请求。界面不加「接着上次聊」之类提示。
- 验证:带来源 id 时新行摘要等于源会话;源会话属于别人时新行无摘要且 201;源码合同断言创建 schema 没有 `context_summary` 字段;写满后「开新对话」没有新增提示文案。
- 防复发:会话满员后的「开新对话」出口必须由服务端继承 `context_summary`;摘要文本任何时候都不得由客户端提供。不得把 `messages` 加回列表 GET。
- 相关记录:BUG-464、BUG-555
- 复发自:无
- 修复版本:待发布
@@ -0,0 +1,60 @@
# PROGRESS · 普通聊天的三条记忆缺口(2026-09-16)
工作树:`.worktrees/consultation-context-memory-20260915`
分支:`codex/consultation-context-memory-20260915`
任务书基线:`origin/staging` @ `6b3248bf`;开工时本 worktree 在 `11893c7f`(任务书已合入 staging)。
本机 Windows。Skill **未 bump**。未改 `frontend/src/app/page.tsx`、数据库、依赖。
开工核对:`docs/BUG_HISTORY.md` 最大号 **BUG-720**。校正四单预占 721–726、external-evidence-cache 预占 727/728,本单使用预占 **BUG-729 / 730 / 731**,无冲突。
## 任务状态
| 任务 | 状态 | BUG |
| --- | --- | --- |
| 5.2 写摘要阈值跟着预算走 | 完成 | BUG-730 |
| 5.1 丢整轮必须留痕 | 完成 | BUG-729 |
| 5.3 写满时静默继承摘要 | 完成 | BUG-731 |
| 5.4 Bug 历史 | 完成 | 729/730/731 |
## 实现要点
- **BUG-730**:删除写死的 `CONSULTATION_HISTORY_TAIL_MAX_CHARS = 16_000`。新函数 `consultationHistoryCheckpointChars(window)` = `floor(historyBudgetChars(window) × 0.4)`,并运行时断言阈值 < 预算。128k 仍是 16,000。`shouldCheckpoint` / `checkpointSessionContextSummary` 吃 `contextWindow`;consult 把会话模型窗口传进去。
- **BUG-729**:`droppedCount > 0` 时在 `consultationUserTurnContent` 的摘要槽追加与 `omissionMarker` 同风格的说明。有摘要:「更早的 N 轮问答已并入上面的会话摘要」;无摘要:「更早的 N 轮问答未能进入本轮上下文,结论尚未并入会话摘要」。`droppedCount === 0` 不加。`route.ts` 四条用户回合都读 `historyWindow.droppedCount`。
- **BUG-731**:`chatSessionCreateSchema` 增加可选 `continued_from_session_id`(保持 `.strict()`,**不加** `context_summary`)。`POST /api/sessions` 按当前用户读源会话摘要,读到才写入新行;读不到 / 别人的 / 空都静默跳过,仍 201。`continueInNewChat` 把当前会话 id 传给 `startNewChat`。界面无新文案。
未改摘要机制本身(800 汉字、15 秒 ref 计时器、乐观并发、结算后异步)。未用 `AbortSignal.timeout()` 替换摘要超时。未把 `messages` 加回列表 GET。未动数据库。
## 既有断言改动
| 文件 | 原值 | 新值 | 原因 |
| --- | --- | --- | --- |
| `session-context-summary.test.ts` | `CONSULTATION_HISTORY_TAIL_MAX_CHARS === 16_000`;`shouldCheckpoint` 只测 15_999 / 16_001 | `consultationHistoryCheckpointChars(128_000) === 16_000`;另测 64k 阈值 2_400 | BUG-730:阈值由预算派生,128k 行为与改前一致 |
| `chat-session-write.test.ts` 源码合同 | `const { id, updated_at: _ignoredClientClock, ...values } = parsed.data` | 另拆 `continued_from_session_id`,不随 values 插入 | 该字段不是表列 |
| `chat-session-url.test.ts` / `composer-isolation-contract.test.ts` | 切片起点 `async function startNewChat()` | `async function startNewChat(` | 可带来源会话 id |
| `consultation-context-cache-contract.test.ts` | 检查点不传窗口 | 传 `sessionContextWindow`;历史文件不得再出现字面量 `16_000` | BUG-730 |
128k 检查点阈值:**原值 16,000 / 新值 16,000 / 原因** 比例 0.4 × 预算 40,000。未弱化其它既有断言。
## 测试
| 命令 | 结果 |
| --- | --- |
| `./node_modules/.bin/tsc --noEmit` | **0 错** |
| `npm run lint` | **0 error**(119 条既有 warning;本单未新增 error) |
| 本单相关 `npx tsx --test`(history / summary / cache contract / chat-session-write / authority / url / composer-isolation / consultation-context) | **69 pass / 0 fail** |
| `npx tsx --test tests/consultation*.test.ts tests/chat-session-*.test.ts tests/session-*.test.ts` | 264 项 / 252 pass / **12 fail**:3 个文件级失败 + 9 条 methodology,全是 Windows `SkillPackageRegistryError`(EPERM symlink),与本单文件无关 |
| `npm test`(`tests/*.test.ts tests/*.test.tsx`) | `# tests 3110 / # pass 3023 / # fail 73 / # skipped 14`。失败清单不含本单文件。对照近期同机 `PROGRESS-chart-page-blocking-open-20260915.md` 的 fail **73**,**零新增失败** |
| `page.tsx` | 未改 |
| `npm run build` | compile + TypeScript 过;Collecting page data 死在既有 `SkillPackageRegistryError`(EPERM symlink `/api/daily-starlanguage`),与本单无关。未能从本机构建表确认 `/` 的 `○ Static` 与首屏 gzip。源码:`page.tsx` 无 `force-dynamic`;既有合同「home stays a client-read query on a static route」本单相关套件已绿 |
## 环境缺口
- 本机 Windows 无开发者模式 symlink:`skill-package-registry` 建 live runtime alias 报 EPERM。因此 `next build` 收集页面数据失败;若干 consult/methodology/skill-binding 测试整文件红。Linux CI / staging 不受影响。
- 无 Docker:`tests/database-*.test.ts` 照常红,与基线同类。
- 无登录态、无 Chrome:写满后「开新对话」的静默继承未做浏览器走查。界面无新文案,源码合同已锁「开新对话」按钮与禁止「接着上次聊」类句子。
## 收尾限制
- 摘要失败仍只 `console.warn("session context summary failed")`,不打印摘要正文。
- 侧栏「新建对话」不带 `continued_from_session_id`,只有写满出口的「开新对话」会静默继承。
- 源会话属于别人或没有摘要时新行 `context_summary` 为空,接口仍 201。
@@ -0,0 +1,93 @@
# PROGRESS · 普通聊天外网证据缓存(2026-09-15 / 2026-09-16)
工作树:`.worktrees/consultation-external-evidence-cache-20260915`
分支:`codex/consultation-external-evidence-cache-20260915`
任务书基线:`origin/staging` @ `6b3248bf`(任务书写)
开工时远端:`origin/staging` @ `11893c7f973573da78d6dbb48db89cde8b373ecc`
预占 BUG:**727 / 728**(基线最大号 720;721–726 为同日其它单预占,未冲突)
交付 commit:`d49bd9cd`
## 串行基线偏离(产品授权)
`TASK-api-server-decomposition-20260916` **未合入** staging。任务书要求以拆分后的 `scripts/jyotish_api_server.py` 为基线;产品负责人要求本单现在就做。因此:
- **没有增长** `scripts/jyotish_api_server.py`。新逻辑在 `scripts/vedastro_snapshot_cache.py`、`scripts/vedastro_foreground.py`;主文件只做薄导入、把 submit/join 换成 `start_foreground_vedastro` / `finish_foreground_vedastro`、按需挂 `western_evidence_packet`。
- 冻结合同:`wc -l` 口径开工 **11334** 行,交付 **11291** 行(少 43);上限 11063+300=**11363**。
未改 context-memory 的前端会话文件,未改 session-capacity 的 SQL 迁移。
## 任务对照
| 任务 | 状态 | BUG |
| --- | --- | --- |
| 5.1 盘 + UTC 日期缓存 | 完成 | BUG-727 |
| 5.2 超时 cancel + join/budget/workers 成组,`budget ≤ 2×join` | 完成 | BUG-727 |
| 5.3 `western_evidence_packet` 按需返回 | 完成 | BUG-728 |
| 5.4 staging 单域耗时清单 | 完成(环境缺口:无 staging 登录) | — |
| 5.5 BUG-727 / 728 | 完成 | — |
## 实现要点
- 缓存键:`sha256(年月日时分秒 + lat + lon + tz + ayanamsa + node + UTC reference_date)`,文件名只有哈希。目录 `scratch/local/vedastro_snapshot_cache`,与 `_api_chart_cache` 分开。
- 同日命中:直接用,不 submit。1–7 天:先用旧的(网关包上标 `snapshot_reference_date`),后台刷新只写缓存。`entrypoint=daily_starlanguage` 只接受当天。未命中才走 1.5 s join。
- 只缓存 `official_verified` 且带 raw 的包;超时/blocked 不落盘,避免把未交付层标成 executed(BUG-301)。
- `_join_foreground_vedastro` 超时 `cancel_event.set()` + `future.cancel()`。budget 默认改为 `2×join`(1.5 s → 3 s),不再默认跑满 8 s。
- 非前台路径(报告/高严谨)仍同步跑网关、不受 join 上限,成功后同样写缓存。
- 前台响应默认去掉 `western_evidence_packet`,仍算、仍投影 `consumer_context.western_spectrum`。MCP / 高严谨 / `include_western_evidence_packet` / `western_oracle_payload` 仍返回整包。
- 前端 `runConsultationWorkflow` 把 `entrypoint` 传给 Python,供 daily 入口拒绝隔日缓存。
## `western_evidence_packet` 检索
| 位置 | 处理 |
| --- | --- |
| `frontend/src` | **零命中**,按「无人读」从咨询响应去掉 |
| `scripts/jyotish_api_server.py` | 仍计算;默认不进响应 |
| `scripts/vedastro_foreground.py` | `should_include_western_evidence_packet` |
| `scripts/western_evidence_packet.py` / `western_chart_engine.py` / `western_oracle_adapter.py` | 保留计算 |
| `scripts/unified_consultation_orchestrator.py` | 仍传入做审计日志 |
| `scripts/cross_system_arbitrator.py` | 保留 |
| `mcp_server.py` | 保留;`surface=skill_mcp` 仍返回整包 |
| `tests/` | oracle / MCP / 按需返回的断言保留 |
## 响应体积
| | 字符 |
| --- | ---: |
| 任务书基线(career 同参第二轮) | 523 k,其中 `western_evidence_packet` 122 k |
| 本机改后(stub 工作流,默认响应) | 该键不存在 |
| 显式 `include_western_evidence_packet` | 键在,体积 ≥ 默认响应 |
未在 staging 重测 523 k 真包(无登录)。按基线,每域少约 122 k 字符。
## 测试
本机无 `G:\Ferti\Jyotisha\.venv`,用 `C:\Users\74082\anaconda3\python.exe`(3.11.7)。任务书写的是 Python 3.13 + swisseph。
| 命令 | 结果 |
| --- | --- |
| `pytest tests/test_vedastro_snapshot_cache.py` | **10 passed** |
| 上项 + 前台赶上/超时/oracle/growth | **18 passed** |
| `pytest tests/test_consultation_consumer_context.py tests/test_vedastro_runtime_ops.py tests/test_vedastro_snapshot_cache.py tests/test_api_server_growth_contract.py tests/test_consultation_workflow_domains.py` | **pass**(1 skip:本机未装 vedastro pin) |
| `pytest tests/test_api_server_security.py` + 上列相关 | **169 passed / 6 failed / 1 skipped**。6 条失败在 **HEAD 无本单改动时同样失败**(thematic `relationship_strict_evidence` / `月度主状态`;`shadbala` advanced:`timezone inference dependency unavailable`)。不是本单引入。 |
| `frontend`:`consultation-workflow-contract` + `consultation-spectrum-parity` | **13 passed / 0 failed** |
| `scripts/run_quality_gate.py --profile quick` | **compile 通过**(含新模块)。停在 `interpretation_source_inventory_gate.py`:`ModuleNotFoundError: No module named 'mcp'`(Anaconda 无 FastMCP)。未改依赖。 |
## 开工预检
`python scripts/pre_work_check.py --remote-timeout 8 --command-timeout 45`:**fail**。`remote_visibility=verified`,fragment/external engine 扫描通过;`tests/test_preflight_fragment_scan.py` 在本机找不到 `.workbuddy/skills/jyotish-vedic-astrology`(Windows 工作树无该镜像)。属 ERR-007 同类环境,未新开台账。不得把 `.workbuddy` 当运行主仓。
## 5.4 staging 耗时
清单:`docs/testing/consultation-external-evidence-cache-20260915.md`。无 staging 登录,**环境缺口**。改前数字用任务书本机表(career 928/865 ms)。改后 staging 两轮 `consultationToolDurationMs` 待产品回填。本单不改三域上限 3。
## 观察项(本单不修)
`reference_date` 缺省 UTC 当天,与 UTC+8「今天」早上八点前后会错开一天。
## 既有断言改动
| 文件 | 原值 | 新值 | 原因 |
| --- | --- | --- | --- |
| 咨询工作流默认响应 | 总有 `western_evidence_packet` | 默认无该键 | BUG-728 无读取点 |
| `_foreground_vedastro_budget_seconds` 默认 | 8 s(2–12) | `min(env, 2×join)`,join 默认 1.5 → budget 3 | 5.2 让步:预算不超过 join 两倍 |
| `tests/conftest.py` | 无缓存隔离 | autouse 把 snapshot cache 指到 tmp | 避免测试互相命中磁盘缓存 |
+2 -2
View File
@@ -238,8 +238,8 @@
| `TASK-rectification-failure-attribution-20260915.md` | — | **三处把系统故障说成别的东西(独占 `route.ts`)**:意图分类器两次异常返回的 `null` 与用户真的「说不清」共用一条分支,回一句「我不太确定这句是不是在回答上面的问题」,**用户这句里的经历直接丢弃且不写证据**(BUG-722,采集题分支早已改对、点选题分支没跟上);引擎 429(`ERR_COMPUTE_BUSY` + `Retry-After`)被压成 `engine_request_failed`,不重试不打日志,证据记下了但范围不动、模型照说「记下了」(BUG-723,**复发自 BUG-715**);attempt 210s × 2 = 420s > 路由 `maxDuration` 240s,重试必超预算(BUG-724,**复发自 BUG-059**,BUG-388 的防复发只写了单次尝试)。超时改成整轮一个预算,不砍 attempt 也不提 240。**产品 2026-09-15 决定:意图分类继续用会话选定的贵模型,不新增「工具模型」角色** | 待领取 | — |
| `TASK-rectification-settled-render-split-20260915.md` | — | **前端性能单(独占校正会话组件,可并行)**:`rectification-agentic-chat.tsx` 1973 行、`useMemo` 0 个、`memo` 0 个,`messages.map` 内联在组件体里且逐条新建时间轴数组与 choice card,`ChatMessageRow` 无 memo、结算态 Markdown 走没有缓存的 `renderProse`。流式每帧(~60/s)重渲整条会话并重跑每条已结算消息的 Markdown。BUG-473 在本文件只落地了 `stream-frame-buffer`,咨询面的 `SettledMessageList` + `HistoryMessageEntry` 拆分没有跟过来。**零行为变化**;验收必须有按帧驱动的渲染计数断言(照 `home-streaming-render-split.test.ts`)。BUG 段 725 | 待领取 | — |
| `TASK-rectification-request-dossier-cache-20260915.md` | — | **低风险单,串行在 failure-attribution 之后(同改 `route.ts`)**:一轮 Agent 对话实测取 3.44 次整份 Case 档案(点选题 2.07 次),全仓约 40 个调用点、请求内零缓存;档案是「最近 50 轮 turns + 全部 evidence + 合成收据」的大 jsonb。做法是包装 `accounting` 客户端做**写即失效**的请求作用域缓存(两个只读投影命中缓存,其余任何 RPC 先清空再转发),**零调用点改动**。不得做成「请求内只读一次」——档案在请求内会变。BUG 段 726 | 待领取 | — |
| `TASK-consultation-external-evidence-cache-20260915.md` | — | **普通聊天性能单(Python;2026-09-15 产品拍板改为**排在 api-server-decomposition **之前**)**:每轮每域同步等外网,cProfile 前三名全是 `api.vedastro.org` 的 HTTPS 往返(0.801 + 0.786 + 0.206 s),本地 swisseph 只有 0.022 s。三个护栏数字凑不齐:前台等 1.5 s、后台跑 8 s、线程池只有 2 个 worker,且超时**不 cancel** → 每 4 秒一轮就长期饱和,之后每轮白等再拿 `official_blocked`(BUG-727)。另 `western_evidence_packet` 122 KB 前端零读取点(BUG-728)。**产品定案**:按「出生数据+岁差+交点+UTC 日期」缓存(与引擎 `_official_snapshot_reference_date` 同键,否决自定 TTL),同日 0 等待 / 跨日先用旧的(≤7 天)后台刷新 / `daily_starlanguage` 要求当天 / 冷启动才走 1.5 s。**不许「干脆不调」——那会重开 BUG-301。** 另含 staging 单域耗时实测单(代码注释里的 21 s 与本机 0.5 s 差 40 倍,三域上限就是从它推的)。BUG 段 727–728 | 待领取 | — |
| `TASK-consultation-context-memory-20260915.md` | — | **记忆三缺口(TS,可并行)**:历史超预算时从最老整轮丢弃,`droppedCount` 算了却**全仓零读取点**,模型不知道少看了几轮——单条截断有「省略 N 字」标记,整轮丢弃没有(BUG-729,BUG-555 防复发只写了「头部截断」所以漏网);写摘要阈值写死 16,000,历史预算却是 `clamp((窗口−60k)×1.5, 4k, 40k)`,窗口 < **70,667** 时预算低于阈值 → 每轮静默丢(BUG-730,后台上架中等窗口模型即触发);写满时服务端存着摘要,`continueInNewChat` 只带问题不带摘要,而 `context_summary` 根本不在任何会话接口的列里(BUG-731)。**产品定案:静默继承**,且摘要文本永远不许由客户端提供(`chatSessionCreateSchema` 只收来源会话 uuid)。BUG 段 729–731 | 待领取 | — |
| `TASK-consultation-external-evidence-cache-20260915.md` | `PROGRESS-consultation-external-evidence-cache-20260915.md` | **普通聊天性能单(Python;2026-09-15 产品拍板改为排在 api-server-decomposition 之前)**:每轮每域同步等外网,cProfile 前三名全是 `api.vedastro.org` 的 HTTPS 往返(0.801 + 0.786 + 0.206 s),本地 swisseph 只有 0.022 s。三个护栏数字凑不齐:前台等 1.5 s、后台跑 8 s、线程池只有 2 个 worker,且超时**不 cancel** → 每 4 秒一轮就长期饱和,之后每轮白等再拿 `official_blocked`(BUG-727)。另 `western_evidence_packet` 122 KB 前端零读取点(BUG-728)。**产品定案**:按「出生数据+岁差+交点+UTC 日期」缓存(与引擎 `_official_snapshot_reference_date` 同键,否决自定 TTL),同日 0 等待 / 跨日先用旧的(≤7 天)后台刷新 / `daily_starlanguage` 要求当天 / 冷启动才走 1.5 s。**不许「干脆不调」——那会重开 BUG-301。** 另含 staging 单域耗时实测单(代码注释里的 21 s 与本机 0.5 s 差 40 倍,三域上限就是从它推的)。BUG 段 727–728 | 待验收 | `codex/consultation-external-evidence-cache-20260915` |
| `TASK-consultation-context-memory-20260915.md` | `PROGRESS-consultation-context-memory-20260915.md` | **记忆三缺口(TS,可并行)**:历史超预算时从最老整轮丢弃,`droppedCount` 算了却**全仓零读取点**,模型不知道少看了几轮——单条截断有「省略 N 字」标记,整轮丢弃没有(BUG-729,BUG-555 防复发只写了「头部截断」所以漏网);写摘要阈值写死 16,000,历史预算却是 `clamp((窗口−60k)×1.5, 4k, 40k)`,窗口 < **70,667** 时预算低于阈值 → 每轮静默丢(BUG-730,后台上架中等窗口模型即触发);写满时服务端存着摘要,`continueInNewChat` 只带问题不带摘要,而 `context_summary` 根本不在任何会话接口的列里(BUG-731)。**产品定案:静默继承**,且摘要文本永远不许由客户端提供(`chatSessionCreateSchema` 只收来源会话 uuid)。BUG 段 729–731 | 待验收 | `codex/consultation-context-memory-20260915` |
| `TASK-consultation-session-capacity-20260915.md` | — | **对话上限单(一份迁移,可并行;不碰 route.ts)**:`append_consultation_question` 的 200,000 字符额度里,`thinkingText`(≤4,000) + `thinkingSections`(实测 1,521/2,243/2,977) 占一半以上,而 `techniqueTruth`/`workflowReceipt`/`agentExecutionReceipt` 照样入库却不计入——同一条上限身兼二职且两职都没做好,约 **19 轮** 就「已写满」(200 条那档永远碰不到)。**产品定案:思考文本不计入**,额度只数用户读得到的正文(约 19 → 约 50 轮),另设一条按 `length(elem::text)` 把全部字段算全的物理上限(算式取 1,000,000,写进迁移注释)护住数据库行;两档都返回同一个 `session_full`。保留 advisory lock / 幂等 / 满员拒绝(BUG-464 防复发)。BUG 段 732 | 待领取 | — |
| `TASK-freeze-metric-change-20260915.md` | — | **规则单(后面两单的前置,无 BUG 号)**:两条增长冻结余量都用完(`page.tsx` 1,951/1,951 余 **0**;`jyotish_api_server.py` 11,334/11,363 余 **29**),冻结从「逼新代码往外走」退化成「拦路」。实证:`page.tsx` 行数砍 59% 但 `Home()` 的 `useState` 从 56 涨到 **66**(拆的是代码不是状态);api server **225 个类方法只有 12 处真碰 HTTP 上下文**,4 处 `__new__` 伪造空壳就是这么来的。**产品拍板换口径**:主门改成「`Home()` 的 useState/useRef 不得增长」与「类方法数 + `__new__` 计数不得增长」,行数降级为粗护栏;**同时推翻 §6「参数式 hook 内部保持 0 个 React hook」**(那正是状态搬不走的原因)。改 `AGENTS.md` §6 + 两个合同测试,不碰业务代码 | 待领取 | — |
| `TASK-home-state-lowering-20260915.md` | — | **page.tsx 状态下沉第一簇(串行在 freeze-metric-change + C2 + R3 之后)**:66 个 state 里 `rectification*` 占 **15** 个,而它们服务的 `<ConversationalBirthTimeRectification>` 本来就是 `dynamic()` 懒加载子树、挂着 24 个 props;`useRectificationSurface` 要解构约 56 个参数。把这簇搬进子树,`Home()` 的 useState 从 66 降到 ≤ 53。**零行为变化**;第一步必须先把 15 个逐个分类(只服务子树 / 外壳也要读)。产品否决了 Context Provider 与外部 store 两条路。不占 BUG 号 | 待领取 | — |
@@ -0,0 +1,41 @@
# 真人核对 · staging 单域咨询耗时(外网证据缓存,2026-09-15)
执行环境无 staging 登录态。下列条目留给有真实账号的人在 staging 走查。本单自动化不替代这些。
目的:同一张盘、同一个问题域连发两轮,记录 `consultationToolDurationMs`。第一轮冷(可能等外网),第二轮应命中「盘 + UTC 日期」缓存,接近本地计算。
代码注释里的域上限 3 是按「staging 实测三域共 62.9 秒」反推的,即一域约 21 秒。本机同样调用只要约 0.5 秒。**本单不改三域上限**;若单域已显著低于 21 秒,把数字写进进度记录,放宽另开单。
## 准备
1. 打开 `https://staging.jyotisha.chat/login` 并登录(用自己的受控账号)。
2. 确认 `https://staging.jyotisha.chat/api/health` 的 `deployment.gitCommit` 等于含本单代码的那次 staging 提交。
3. 选一张**当天还没问过**的盘(或等 UTC 零点后再测),避免误用旧缓存。不要把出生资料贴回对话。
## 冷启动(第一轮)
4. 新建对话,只问一个域(例如事业)。不要点「深入看今日」。
5. 等回答出来。记下:
- 主观等待 `____` 秒
- 若浏览器/服务端日志能看到 `consultationToolDurationMs`:`____` ms
- 技法表「VedAstro 云状态」是 executed 还是 blocked:`____`
## 同日第二轮(应 0 等待)
6. 同一条对话里,用同一张盘再问同一域(换一句问法即可)。
7. 记下:
- 主观等待 `____` 秒
- `consultationToolDurationMs`:`____` ms
- VedAstro 云状态:`____`
8. 第二轮应明显短于第一轮,且不应再出现「空等约 1.5 秒再 blocked」。
## 跨日(可选,UTC 过零点后)
9. 第二天用同一张盘再问同一域。第一句应马上有回答(用昨天的证据),证据里能看出是哪一天的快照;下一轮才换成当天。
10. 点「深入看今日」时,不得沿用昨天的快照;没有当天的就走原来的 1.5 秒有界等待。
## 回填后怎么处理
把 5、7 的两组数字交给验收方,写进 `docs/tasks/PROGRESS-consultation-external-evidence-cache-20260915.md`。若单域已稳定低于 21 秒,结论写成「三域上限可以放宽到 N」的依据,**不要在本单改上限**。
**在这份清单回填之前,任何人不得声称「staging 单域耗时已验证」。**
+7 -1
View File
@@ -330,6 +330,7 @@ export async function POST(request: Request) {
{ status: 503 },
);
}
const sessionContextWindow = sessionModel.contextWindow;
if (parsed.data.entrypoint === "birth_time_rectification") {
return NextResponse.json(
@@ -353,7 +354,7 @@ export async function POST(request: Request) {
// Client `history` stays in the request schema for old bundles and is not read.
const contextSummary = parseSessionContextSummary(chatSession.context_summary);
const historyWindow = consultationHistoryWindow(chatSession.messages, contextSummary, {
contextWindow: sessionModel.contextWindow,
contextWindow: sessionContextWindow,
});
const storedHistory = historyWindow.tail;
const userControlledPrompt = [
@@ -596,6 +597,7 @@ export async function POST(request: Request) {
await checkpointSessionContextSummary({
messages: sessionRow.messages,
summary: sessionRow.context_summary,
contextWindow: sessionContextWindow,
generateText: (prompt, signal) => generateSessionContextSummaryText(summaryModel, prompt, signal),
update: async (summary, seenUpdatedAt) => {
let query = supabase.from("chat_sessions")
@@ -875,6 +877,7 @@ export async function POST(request: Request) {
instruction: modeInstruction,
extra: generalDailyContextPrompt(generalDailyContext),
summaryText: historyWindow.summaryText,
droppedCount: historyWindow.droppedCount,
question: resolvedQuestion.modelQuestion,
}),
},
@@ -1271,6 +1274,7 @@ export async function POST(request: Request) {
instruction: generalNoMinuteInstruction(Boolean(generalDailyContext)),
extra: generalDailyContextPrompt(generalDailyContext),
summaryText: historyWindow.summaryText,
droppedCount: historyWindow.droppedCount,
question: resolvedQuestion.modelQuestion,
}),
},
@@ -1355,6 +1359,7 @@ export async function POST(request: Request) {
name,
instruction: "先用 3–6 句口语直接回答下面的问题,不要加标题;形状为一句结论、2–3 条短要点(每条完整句子、不超过 30 字)、一句下一步,总量不超过 400 字;然后再按 skill Level 2 骨架写:原始结构、六步宫位、Yoga 表、时机、综合、文末技法审计表,最后才是现代生活。骨架不可省略。星盘事实只使用系统里已经注入的计算结果,不要复述内部字段、JSON 或再跑一遍咨询流程。",
summaryText: historyWindow.summaryText,
droppedCount: historyWindow.droppedCount,
question: resolvedQuestion.modelQuestion,
}),
},
@@ -1379,6 +1384,7 @@ export async function POST(request: Request) {
name,
instruction: "先用 3–6 句口语直接回答下面的问题,不要加标题;形状为一句结论、2–3 条短要点(每条完整句子、不超过 30 字)、一句下一步,总量不超过 400 字;然后再按 skill Level 2 骨架写:原始结构、六步宫位、Yoga 表、时机、综合、文末技法审计表,最后才是现代生活。骨架不可省略。星盘事实只使用系统里已经注入的计算结果,不要复述内部字段、JSON 或再跑一遍咨询流程。",
summaryText: historyWindow.summaryText,
droppedCount: historyWindow.droppedCount,
question: resolvedQuestion.modelQuestion,
}),
},
+28 -7
View File
@@ -1,6 +1,12 @@
import { NextResponse } from "next/server";
import { chatSessionCreateSchema, ChatSessionBodyTooLargeError, readChatSessionJson } from "@/lib/chat-session-write-contract";
import {
chatSessionCreateInsertRow,
chatSessionCreateSchema,
ChatSessionBodyTooLargeError,
readChatSessionJson,
} from "@/lib/chat-session-write-contract";
import { consumeUserRequestRateLimit } from "@/lib/request-rate-limit";
import { resolveInheritedContextSummary } from "@/lib/session-context-summary";
import { isSupabaseConfigurationError } from "@/lib/supabase/config";
import { createServerSupabaseClient } from "@/lib/supabase/server";
import {
@@ -94,13 +100,28 @@ export async function POST(request: Request) {
}
const parsed = chatSessionCreateSchema.safeParse(await readChatSessionJson(request));
if (!parsed.success) return NextResponse.json({ error: "聊天记录格式不正确" }, { status: 400 });
const { id, updated_at: _ignoredClientClock, ...values } = parsed.data;
const { error } = await supabase.from("chat_sessions").insert({
id,
user_id: user.id,
...values,
updated_at: new Date().toISOString(),
const { id, updated_at: _ignoredClientClock, continued_from_session_id: continuedFromSessionId, ...values } = parsed.data;
const inheritedSummary = await resolveInheritedContextSummary({
continuedFromSessionId,
loadOwnedSummary: async (sourceId) => {
const { data } = await supabase
.from("chat_sessions")
.select("context_summary")
.eq("id", sourceId)
.eq("user_id", user.id)
.maybeSingle();
return data?.context_summary ?? null;
},
});
const { error } = await supabase.from("chat_sessions").insert(
chatSessionCreateInsertRow({
id,
userId: user.id,
values,
inheritedSummary,
updatedAt: new Date().toISOString(),
}),
);
if (error) return NextResponse.json({ error: "聊天记录暂时无法同步" }, { status: 500 });
return NextResponse.json({ ok: true }, { status: 201 });
} catch (error) {
+1 -1
View File
@@ -135,7 +135,7 @@ export type ConsultationRunParams = {
uiPreviewMode: MutableRefObject<string | null>;
persistSession: (session: ChatSession, mode?: "create" | "update") => Promise<void>;
updateSession: (sessionId: string, change: (session: ChatSession) => ChatSession) => void;
startNewChat: () => Promise<ChatSession | null>;
startNewChat: (options?: { continuedFromSessionId?: string }) => Promise<ChatSession | null>;
continueInNewChat: (prompt: { question: string; theme: Theme }) => Promise<void>;
refreshAccount: () => Promise<void>;
openAccountDialog: (dialog: AccountDialog, options?: HTMLButtonElement | null | OpenAccountDialogOptions) => void;
+20 -4
View File
@@ -139,7 +139,11 @@ export function useSessionManagement(params: SessionManagementParams) {
setSessions((current) => current.map((session) => (session.id === sessionId ? change(session) : session)));
}
async function persistSession(session: ChatSession, mode: "create" | "update" = "update") {
async function persistSession(
session: ChatSession,
mode: "create" | "update" = "update",
options?: { continuedFromSessionId?: string },
) {
if (!account) throw new Error("账户尚未加载完成");
if (process.env.NODE_ENV === "development" && uiPreview.current) return;
const values = mode === "create"
@@ -153,6 +157,9 @@ export function useSessionManagement(params: SessionManagementParams) {
chart_profile_id: session.chartProfileId,
chart_profile_name: session.chartProfileName,
chart_profile_role: session.chartProfileRole,
...(options?.continuedFromSessionId
? { continued_from_session_id: options.continuedFromSessionId }
: {}),
}
: {
title: session.title,
@@ -192,7 +199,10 @@ export function useSessionManagement(params: SessionManagementParams) {
async function continueInNewChat(prompt: { question: string; theme: Theme }) {
setSessionFullPrompt(null);
const created = await startNewChat();
const sourceSessionId = activeSessionId;
const created = await startNewChat(
sourceSessionId ? { continuedFromSessionId: sourceSessionId } : undefined,
);
if (!created) return;
setDraft(prompt.question);
setDraftTheme(prompt.theme);
@@ -294,7 +304,7 @@ export function useSessionManagement(params: SessionManagementParams) {
}
}
async function startNewChat(): Promise<ChatSession | null> {
async function startNewChat(options?: { continuedFromSessionId?: string }): Promise<ChatSession | null> {
if (!account || !modelCatalog || creatingSession) return null;
const nextSession = {
...createSession(modelCatalog.defaultModelId),
@@ -312,7 +322,13 @@ export function useSessionManagement(params: SessionManagementParams) {
setComposerNotice("");
setRequestError(null);
try {
await persistSession(nextSession, "create");
await persistSession(
nextSession,
"create",
options?.continuedFromSessionId
? { continuedFromSessionId: options.continuedFromSessionId }
: undefined,
);
return nextSession;
} catch (caught) {
setSessions((current) => current.filter((session) => session.id !== nextSession.id));
@@ -59,6 +59,7 @@ export const chatSessionCreateSchema = z.object({
messages: z.array(chatMessageSchema).max(0),
session_type: z.enum(["consultation", "birth_time_rectification"]),
rectification_case_id: z.string().uuid().nullable(),
continued_from_session_id: z.string().uuid().optional(),
...chartBindingSchema,
updated_at: z.string().datetime().optional(),
}).strict();
@@ -148,11 +149,41 @@ export type ChatSessionCreate = Readonly<{
messages: readonly [];
session_type: "consultation" | "birth_time_rectification";
rectification_case_id: string | null;
continued_from_session_id?: string;
chart_profile_id?: string | null;
chart_profile_name?: string | null;
chart_profile_role?: "self" | "other" | null;
}>;
export type ChatSessionCreateInsertValues = Readonly<{
title: string;
theme: ConsultationDomain;
model_id: string;
messages: readonly unknown[];
session_type: "consultation" | "birth_time_rectification";
rectification_case_id: string | null;
chart_profile_id?: string | null;
chart_profile_name?: string | null;
chart_profile_role?: "self" | "other" | null;
}>;
export function chatSessionCreateInsertRow(input: {
id: string;
userId: string;
values: ChatSessionCreateInsertValues;
inheritedSummary?: unknown;
updatedAt: string;
}): Record<string, unknown> {
const row: Record<string, unknown> = {
id: input.id,
user_id: input.userId,
...input.values,
updated_at: input.updatedAt,
};
if (input.inheritedSummary) row.context_summary = input.inheritedSummary;
return row;
}
export type ChatSessionWrite = Readonly<{
title: string;
theme: ConsultationDomain;
@@ -1,11 +1,14 @@
export const CONSULTATION_HISTORY_LIMIT = 12;
export const CONSULTATION_HISTORY_MESSAGE_CHARS = 12_000;
export const CONSULTATION_HISTORY_TAIL_MAX_CHARS = 16_000;
export const CONSULTATION_HISTORY_SYSTEM_RESERVE_TOKENS = 60_000;
export const CONSULTATION_HISTORY_CHAR_PER_TOKEN = 1.5;
export const CONSULTATION_HISTORY_BUDGET_MIN_CHARS = 4_000;
export const CONSULTATION_HISTORY_BUDGET_MAX_CHARS = 40_000;
export const DEFAULT_MODEL_CONTEXT_WINDOW = 128_000;
// 0.4 of the history budget. At the default 128k window the budget is 40,000,
// so the checkpoint fires at 16,000 — the previous hard-coded threshold.
// Smaller windows then checkpoint before the tail can exceed the budget.
export const CONSULTATION_HISTORY_CHECKPOINT_BUDGET_RATIO = 0.4;
export const SESSION_CONTEXT_SUMMARY_HEADING = "【会话摘要(服务端维护)】";
@@ -51,6 +54,15 @@ export function historyBudgetChars(contextWindow: number | null | undefined): nu
);
}
export function consultationHistoryCheckpointChars(contextWindow: number | null | undefined): number {
const budget = historyBudgetChars(contextWindow);
const threshold = Math.floor(budget * CONSULTATION_HISTORY_CHECKPOINT_BUDGET_RATIO);
if (!(threshold < budget)) {
throw new Error("consultation history checkpoint threshold must stay below the history budget");
}
return threshold;
}
export function parseSessionContextSummary(value: unknown): SessionContextSummaryV1 | null {
if (!value || typeof value !== "object" || Array.isArray(value)) return null;
const row = value as Record<string, unknown>;
@@ -74,6 +86,13 @@ export function omissionMarker(omittedChars: number): string {
return `……(以下省略 ${omittedChars} 字,结论已并入会话摘要)`;
}
export function droppedRoundsMarker(droppedCount: number, hasSummary: boolean): string {
if (droppedCount <= 0) return "";
return hasSummary
? `……(更早的 ${droppedCount} 轮问答已并入上面的会话摘要)`
: `……(更早的 ${droppedCount} 轮问答未能进入本轮上下文,结论尚未并入会话摘要)`;
}
export function clipConsultationHistoryText(text: string): string {
if (text.length <= CONSULTATION_HISTORY_MESSAGE_CHARS) return text;
const omitted = text.length - CONSULTATION_HISTORY_MESSAGE_CHARS;
@@ -165,16 +184,18 @@ export function consultationUserTurnContent(input: {
instruction: string;
extra?: string;
summaryText?: string | null;
droppedCount?: number;
question: string;
}): string {
const summary = input.summaryText?.trim() ?? "";
const dropped = droppedRoundsMarker(input.droppedCount ?? 0, Boolean(summary));
return [
input.currentTime,
input.name ? `用户称呼:${input.name}` : "",
input.instruction,
input.extra ?? "",
input.summaryText?.trim()
? `${SESSION_CONTEXT_SUMMARY_HEADING}\n${input.summaryText.trim()}`
: "",
summary ? `${SESSION_CONTEXT_SUMMARY_HEADING}\n${summary}` : "",
dropped,
input.question,
].filter(Boolean).join("\n");
}
+23 -5
View File
@@ -1,7 +1,7 @@
import { Agent } from "@mastra/core/agent";
import {
CONSULTATION_HISTORY_TAIL_MAX_CHARS,
consultationHistoryCheckpointChars,
lastConsultationPair,
parseSessionContextSummary,
storedConsultationTurns,
@@ -98,7 +98,7 @@ export function sanitizeSessionContextSummary(raw: string): string | null {
export function tailCharCount(
messages: unknown,
summary: SessionContextSummaryV1 | null,
options: { excludeRequestId?: string } = {},
options: { excludeRequestId?: string; contextWindow?: number | null } = {},
): number {
const turns = storedConsultationTurns(messages, { excludeRequestId: options.excludeRequestId });
const tail = summary
@@ -110,9 +110,10 @@ export function tailCharCount(
export function shouldCheckpoint(
messages: unknown,
summary: SessionContextSummaryV1 | null,
options: { excludeRequestId?: string } = {},
options: { excludeRequestId?: string; contextWindow?: number | null } = {},
): boolean {
return tailCharCount(messages, summary, options) > CONSULTATION_HISTORY_TAIL_MAX_CHARS;
return tailCharCount(messages, summary, options)
> consultationHistoryCheckpointChars(options.contextWindow);
}
export function messagesForSummaryInput(
@@ -215,6 +216,19 @@ export function nextSessionContextSummary(
};
}
export async function resolveInheritedContextSummary(input: {
continuedFromSessionId?: string | null;
loadOwnedSummary: (sessionId: string) => Promise<unknown>;
}): Promise<SessionContextSummaryV1 | null> {
const sourceId = input.continuedFromSessionId?.trim();
if (!sourceId) return null;
try {
return parseSessionContextSummary(await input.loadOwnedSummary(sourceId));
} catch {
return null;
}
}
export async function writeSessionContextSummary(input: {
seenUpdatedAt: string | null;
summary: SessionContextSummaryV1;
@@ -228,13 +242,17 @@ export async function checkpointSessionContextSummary(input: {
messages: unknown;
summary: unknown;
excludeRequestId?: string;
contextWindow?: number | null;
now?: () => Date;
generateText: (prompt: string, signal?: AbortSignal) => Promise<string>;
update: (summary: SessionContextSummaryV1, seenUpdatedAt: string | null) => Promise<boolean>;
timeoutMs?: number;
}): Promise<"written" | "skipped" | "abandoned" | "failed"> {
const previous = parseSessionContextSummary(input.summary);
if (!shouldCheckpoint(input.messages, previous, { excludeRequestId: input.excludeRequestId })) {
if (!shouldCheckpoint(input.messages, previous, {
excludeRequestId: input.excludeRequestId,
contextWindow: input.contextWindow,
})) {
return "skipped";
}
try {
@@ -676,6 +676,7 @@ export function createConsultationTools(ctx: ConsultationAgentContext) {
signal: context.abortSignal ?? ctx.abortSignal,
plan: domainPlan,
requestId: ctx.requestId,
entrypoint: ctx.entrypoint,
});
const guarded = applyBirthTimeModeToWorkflowContext(workflow, ctx.consultationMode, {
birthTimeSource: ctx.serverChart.truth.birthTimeSource,
+8 -1
View File
@@ -125,7 +125,13 @@ export function consultationWorkflowFailureCode(error: unknown): ConsultationWor
export async function runConsultationWorkflow(
input: ConsultationInput,
options?: { foreground?: boolean; signal?: AbortSignal; plan?: ConsultationPlan; requestId?: string },
options?: {
foreground?: boolean;
signal?: AbortSignal;
plan?: ConsultationPlan;
requestId?: string;
entrypoint?: string;
},
) {
const { entryMode, question, theme, ...workflowInput } = input;
const plan = options?.plan ?? createConsultationPlan({ userIntent: question, theme });
@@ -139,6 +145,7 @@ export async function runConsultationWorkflow(
...workflowInput,
// Correlates this call with the API access log when a run fails.
...(options?.requestId ? { request_id: options.requestId } : {}),
...(options?.entrypoint ? { entrypoint: options.entrypoint } : {}),
entry_mode: entryMode,
question: workflowRequest.question,
question_text: workflowRequest.question,
@@ -20,6 +20,10 @@ test("session list GET omits messages while detail GET returns them", () => {
/SESSION_LIST_COLUMNS = "id,title,theme,model_id,session_type,rectification_case_id,chart_profile_id,chart_profile_name,chart_profile_role,updated_at,pinned,archived_at"/,
);
assert.doesNotMatch(listRoute, /select\(SESSION_LIST_COLUMNS\)[\s\S]*messages/);
assert.doesNotMatch(
listRoute,
/SESSION_LIST_COLUMNS = "[^"]*context_summary/,
);
assert.match(itemRoute, /export async function GET/);
assert.match(
itemRoute,
@@ -52,6 +56,8 @@ test("consult appends the user question after reserve and returns session_full",
assert.match(sendSource, /caught\.code === "session_full"/);
assert.match(sendSource, /label: "开新对话"/);
assert.match(sendSource, /continueInNewChat\(\{ question: originalQuestion, theme \}\)/);
assert.match(page, /continuedFromSessionId: sourceSessionId/);
assert.doesNotMatch(page, /接着上次|继续上次聊|继承会话摘要|从上次对话继续/);
});
test("PATCH compatibility accepts and ignores a legacy messages write", () => {
+1 -1
View File
@@ -173,7 +173,7 @@ test("popstate to a missing session query reuses selectSession side effects for
});
test("creating and leaving a session keep the address bar in sync", () => {
const startNewChat = sourceBetween(page, "async function startNewChat()", "function selectSession(");
const startNewChat = sourceBetween(page, "async function startNewChat(", "function selectSession(");
assert.match(startNewChat, /writeSessionUrl\(nextSession\.id, "push"\)/);
assert.match(startNewChat, /window\.history\.replaceState\(null, "", previousHref\)/);
+87 -3
View File
@@ -1,7 +1,14 @@
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import test from "node:test";
import { chatSessionCreateSchema, chatSessionMetadataPatchSchema, chatSessionWriteSchema, writeChatSession, type ChatSessionWrite } from "../src/lib/chat-session-write-contract.ts";
import {
chatSessionCreateInsertRow,
chatSessionCreateSchema,
chatSessionMetadataPatchSchema,
chatSessionWriteSchema,
writeChatSession,
type ChatSessionWrite,
} from "../src/lib/chat-session-write-contract.ts";
import { metadataUpdateValues } from "../src/lib/session-metadata-update.ts";
import { homeSurface } from "./home-surface.ts";
@@ -41,6 +48,61 @@ test("create schema keeps the client-generated session id after transcript limit
assert.equal(parsed.messages.length, 0);
});
test("create schema accepts continued_from_session_id and rejects client-supplied context_summary", () => {
const parsed = chatSessionCreateSchema.parse({
id: sessionId,
...createValues,
continued_from_session_id: "22222222-2222-4222-8222-222222222222",
});
assert.equal(parsed.continued_from_session_id, "22222222-2222-4222-8222-222222222222");
assert.equal("context_summary" in parsed, false);
assert.equal("context_summary" in chatSessionCreateSchema.shape, false);
const withSummary = chatSessionCreateSchema.safeParse({
id: sessionId,
...createValues,
context_summary: { version: 1, text: "injected" },
});
assert.equal(withSummary.success, false);
const badId = chatSessionCreateSchema.safeParse({
id: sessionId,
...createValues,
continued_from_session_id: "not-a-uuid",
});
assert.equal(badId.success, false);
});
test("create insert copies an owned summary and omits a missing one", () => {
const source = {
version: 1 as const,
text: "已问过的问题\n事业时机",
throughRequestId: "a1",
throughMessageIndex: 3,
messageCount: 4,
updatedAt: "2026-09-15T00:00:00.000Z",
};
const copied = chatSessionCreateInsertRow({
id: sessionId,
userId: "owner",
values: createValues,
inheritedSummary: source,
updatedAt: "2026-09-16T00:00:00.000Z",
});
assert.deepEqual(copied.context_summary, source);
assert.equal("continued_from_session_id" in copied, false);
const skipped = chatSessionCreateInsertRow({
id: sessionId,
userId: "owner",
values: createValues,
inheritedSummary: null,
updatedAt: "2026-09-16T00:00:00.000Z",
});
assert.equal("context_summary" in skipped, false);
assert.equal(skipped.user_id, "owner");
});
test("create schema rejects a non-empty transcript", () => {
// Former value: create accepted up to CHAT_SESSION_MAX_MESSAGES and was a
// history-import path. Create is now only an empty session.
@@ -122,6 +184,19 @@ test("metadata patch accepts pin and archive fields without a transcript", () =>
assert.deepEqual(chatSessionMetadataPatchSchema.parse({ archived_at: null }), { archived_at: null });
});
test("create write may carry continued_from_session_id and never context_summary", async () => {
const calls: Array<Record<string, unknown>> = [];
await writeChatSession(sessionId, {
...createValues,
continued_from_session_id: "22222222-2222-4222-8222-222222222222",
}, "create", async (_url, init) => {
calls.push(JSON.parse(String(init?.body)));
return Response.json({ ok: true }, { status: 201 });
});
assert.equal(calls[0]?.continued_from_session_id, "22222222-2222-4222-8222-222222222222");
assert.equal("context_summary" in (calls[0] ?? {}), false);
});
test("chat session writes use same-origin API instead of browser-to-Supabase requests", async () => {
const calls: Array<{ url: string; init?: RequestInit }> = [];
await writeChatSession(sessionId, metadataPatch, "update", async (url, init) => {
@@ -205,8 +280,17 @@ test("session API owns create and update while answer UI keeps sync failures out
assert.match(itemRoute, /readChatSessionJson/);
assert.match(collectionRoute, /ChatSessionBodyTooLargeError/);
assert.match(itemRoute, /ChatSessionBodyTooLargeError/);
// Former value: `const { id, ...values } = parsed.data` trusted the client clock.
assert.match(collectionRoute, /const \{ id, updated_at: _ignoredClientClock, \.\.\.values \} = parsed\.data/);
// Former value: `const { id, updated_at: _ignoredClientClock, ...values } = parsed.data`
// New value: also peel `continued_from_session_id` so it is never inserted as a column.
assert.match(
collectionRoute,
/const \{ id, updated_at: _ignoredClientClock, continued_from_session_id: continuedFromSessionId, \.\.\.values \} = parsed\.data/,
);
assert.match(collectionRoute, /resolveInheritedContextSummary/);
assert.match(collectionRoute, /chatSessionCreateInsertRow/);
assert.match(collectionRoute, /\.eq\("user_id", user\.id\)/);
assert.match(collectionRoute, /status: 201/);
assert.doesNotMatch(collectionRoute, /context_summary: parsed/);
assert.match(contract, /function limitTranscriptSize<Output extends \{ messages: Array<\{ text: string; thinkingText\?: string; thinkingSections\?: unknown \}> \}>/);
assert.match(contract, /\): z\.ZodType<Output> \{/);
assert.match(page, /chartSnapshotForSession/);
@@ -91,7 +91,7 @@ test("every external draft writer keeps working through the page-owned setters",
// When: each existing write path is inspected.
const chooseSuggested = sourceBetween(pageSource, "function chooseSuggestedQuestion(", "async function startSuggestedConsultation");
const startNewChat = sourceBetween(pageSource, "async function startNewChat()", "function selectSession(");
const startNewChat = sourceBetween(pageSource, "async function startNewChat(", "function selectSession(");
const selectSession = sourceBetween(pageSource, "function selectSession(sessionId: string)", "async function selectSessionModel");
const saveOnboardingName = sourceBetween(pageSource, "async function saveOnboardingName()", "async function saveOnboardingBirth");
const stopRestore = sourceBetween(pageSource, "updateSession(pending.sessionId, () => pending.previousSession);", "function completeConsultationInterface");
@@ -14,7 +14,10 @@ test("consult history uses the checkpoint tail and puts the summary after the ti
assert.match(consultRoute, /consultationHistoryWindow\(chatSession\.messages, contextSummary/);
assert.match(consultRoute, /SESSION_CONTEXT_SUMMARY_HEADING|consultationUserTurnContent/);
assert.match(consultRoute, /summaryText: historyWindow\.summaryText/);
assert.match(consultRoute, /droppedCount: historyWindow\.droppedCount/);
assert.match(history, /【会话摘要(服务端维护)】/);
assert.match(history, /droppedRoundsMarker/);
assert.doesNotMatch(history, /16_000|16000/);
const helper = history.slice(history.indexOf("export function consultationUserTurnContent"));
const timeLine = helper.indexOf("input.currentTime");
const summaryLine = helper.indexOf("SESSION_CONTEXT_SUMMARY_HEADING");
@@ -33,9 +36,13 @@ test("consult retries once on context overflow with summary plus the last pair",
test("consult checkpoints the session summary after a successful completion", () => {
assert.match(consultRoute, /void checkpointConsultationContext\(\)/);
assert.match(consultRoute, /checkpointSessionContextSummary/);
assert.match(consultRoute, /contextWindow: sessionContextWindow/);
assert.match(consultRoute, /context_summary ->>updatedAt|context_summary->>updatedAt/);
assert.match(consultRoute, /console\.warn\("session context summary failed"/);
assert.doesNotMatch(consultRoute, /AbortSignal\.timeout\(\s*15/);
const summary = readFileSync(new URL("../src/lib/session-context-summary.ts", import.meta.url), "utf8");
assert.match(summary, /consultationHistoryCheckpointChars/);
assert.doesNotMatch(summary, /CONSULTATION_HISTORY_TAIL_MAX_CHARS/);
});
test("the context summary migration only adds one jsonb column", () => {
@@ -2,11 +2,14 @@ import assert from "node:assert/strict";
import test from "node:test";
import {
CONSULTATION_HISTORY_CHECKPOINT_BUDGET_RATIO,
CONSULTATION_HISTORY_MESSAGE_CHARS,
clipConsultationHistoryText,
consultationHistoryCheckpointChars,
consultationHistoryFromStoredMessages,
consultationHistoryWindow,
consultationUserTurnContent,
droppedRoundsMarker,
historyBudgetChars,
isContextOverflowError,
omissionMarker,
@@ -77,6 +80,61 @@ test("historyBudgetChars uses 64k, 32k, and null-as-128k windows", () => {
assert.equal(historyBudgetChars(undefined), 40_000);
});
test("checkpoint threshold stays below the history budget for legal context windows", () => {
assert.equal(CONSULTATION_HISTORY_CHECKPOINT_BUDGET_RATIO, 0.4);
const windows = [200_000, 128_000, 64_000, 32_000, null] as const;
for (const window of windows) {
const budget = historyBudgetChars(window);
const threshold = consultationHistoryCheckpointChars(window);
assert.ok(threshold < budget, `window=${window}`);
}
// 128k keeps the previous 16,000-character checkpoint.
assert.equal(consultationHistoryCheckpointChars(128_000), 16_000);
assert.equal(consultationHistoryCheckpointChars(null), 16_000);
assert.equal(consultationHistoryCheckpointChars(64_000), 2_400);
assert.equal(consultationHistoryCheckpointChars(32_000), 1_600);
});
test("dropped whole turns leave an omission marker in the user-turn summary slot", () => {
const body = "x".repeat(3_000);
const history = consultationHistoryWindow(numberedMessages(20, () => body), null, {
contextWindow: 128_000,
});
assert.ok(history.droppedCount > 0);
const withoutSummary = consultationUserTurnContent({
currentTime: "当前时间:2026-09-16 12:00(中国)",
instruction: "先加载 Jyotish Skill",
summaryText: history.summaryText,
droppedCount: history.droppedCount,
question: "刚才你说的那个时间",
});
assert.match(
withoutSummary,
new RegExp(`更早的 ${history.droppedCount} 轮问答未能进入本轮上下文,结论尚未并入会话摘要`),
);
assert.equal(withoutSummary.includes(SESSION_CONTEXT_SUMMARY_HEADING), false);
const withSummary = consultationUserTurnContent({
currentTime: "当前时间:2026-09-16 12:00(中国)",
instruction: "先加载 Jyotish Skill",
summaryText: "先前结论",
droppedCount: history.droppedCount,
question: "刚才你说的那个时间",
});
assert.match(withSummary, new RegExp(`更早的 ${history.droppedCount} 轮问答已并入上面的会话摘要`));
assert.ok(withSummary.indexOf(SESSION_CONTEXT_SUMMARY_HEADING) < withSummary.indexOf("刚才你说的那个时间"));
const kept = consultationUserTurnContent({
currentTime: "当前时间:2026-09-16 12:00(中国)",
instruction: "先加载 Jyotish Skill",
summaryText: "先前结论",
droppedCount: 0,
question: "刚才你说的那个时间",
});
assert.equal(kept.includes("更早的"), false);
assert.equal(droppedRoundsMarker(0, true), "");
});
test("stored consultation history can skip the in-flight request id", () => {
const history = consultationHistoryFromStoredMessages([
{ role: "user", text: "old", requestId: "keep" },
@@ -177,6 +177,19 @@ test("validates and emits non-sensitive workflow and execution receipts", () =>
assert.match(route, /streamAgentResponse/);
});
test("consultationWorkflowResponseSchema still parses without western_evidence_packet", () => {
const schema = workflow.slice(
workflow.indexOf("export const consultationWorkflowResponseSchema"),
workflow.indexOf("function record("),
);
assert.match(schema, /success: z\.boolean\(\)/);
assert.match(schema, /chart: z\.record\(z\.unknown\(\)\)/);
assert.match(schema, /routing: z\.record\(z\.unknown\(\)\)/);
assert.match(schema, /consumer_context:/);
assert.match(schema, /\)\.passthrough\(\);/);
assert.doesNotMatch(schema, /western_evidence_packet/);
});
test("carries commercial technique truth into the model contract", () => {
assert.match(workflow, /technique_truth/);
assert.match(mastra, /deterministic_claims_forbidden_for/);
+43 -3
View File
@@ -3,13 +3,14 @@ import { readFileSync } from "node:fs";
import test from "node:test";
import {
CONSULTATION_HISTORY_TAIL_MAX_CHARS,
consultationHistoryCheckpointChars,
} from "../src/lib/consultation-session-history.ts";
import {
buildSummaryPrompt,
checkpointSessionContextSummary,
generateSessionContextSummary,
messagesForSummaryInput,
resolveInheritedContextSummary,
sanitizeSessionContextSummary,
shouldCheckpoint,
writeSessionContextSummary,
@@ -24,10 +25,14 @@ function overBudgetConversation() {
];
}
test("checkpoint triggers only when the tail exceeds 16_000 characters", () => {
assert.equal(CONSULTATION_HISTORY_TAIL_MAX_CHARS, 16_000);
test("checkpoint triggers only when the tail exceeds the derived threshold", () => {
// Former value: hard-coded CONSULTATION_HISTORY_TAIL_MAX_CHARS = 16_000.
// 128k still checkpoints at 16_000 (0.4 × 40_000 budget).
assert.equal(consultationHistoryCheckpointChars(128_000), 16_000);
assert.equal(shouldCheckpoint([{ role: "user", text: "x".repeat(15_999) }], null), false);
assert.equal(shouldCheckpoint([{ role: "user", text: "x".repeat(16_001) }], null), true);
assert.equal(shouldCheckpoint([{ role: "user", text: "x".repeat(2_400) }], null, { contextWindow: 64_000 }), false);
assert.equal(shouldCheckpoint([{ role: "user", text: "x".repeat(2_401) }], null, { contextWindow: 64_000 }), true);
});
test("checkpoint prompt omits the last question-answer pair", () => {
@@ -108,6 +113,41 @@ test("writeSessionContextSummary abandons when updatedAt does not match", async
assert.equal(result, "abandoned");
});
test("inherited context summary copies owned text and skips foreign or empty sources", async () => {
const source = {
version: 1 as const,
text: "已问过的问题\n事业时机",
throughRequestId: "a1",
throughMessageIndex: 3,
messageCount: 4,
updatedAt: "2026-09-15T00:00:00.000Z",
};
const copied = await resolveInheritedContextSummary({
continuedFromSessionId: "11111111-1111-4111-8111-111111111111",
loadOwnedSummary: async () => source,
});
assert.deepEqual(copied, source);
const foreign = await resolveInheritedContextSummary({
continuedFromSessionId: "22222222-2222-4222-8222-222222222222",
loadOwnedSummary: async () => null,
});
assert.equal(foreign, null);
const empty = await resolveInheritedContextSummary({
continuedFromSessionId: "11111111-1111-4111-8111-111111111111",
loadOwnedSummary: async () => ({ version: 1, text: " " }),
});
assert.equal(empty, null);
const skipped = await resolveInheritedContextSummary({
loadOwnedSummary: async () => {
throw new Error("should not load");
},
});
assert.equal(skipped, null);
});
test("checkpoint writes a new summary when the tail is over budget", async () => {
const result = await checkpointSessionContextSummary({
messages: overBudgetConversation(),
+41 -84
View File
@@ -52,6 +52,18 @@ try:
MIN_SCORING_EVENTS,
)
from scripts.vedastro_runtime_context import temporary_timeout_seconds
from scripts.vedastro_foreground import (
_FOREGROUND_VEDASTRO_EXECUTOR,
_FOREGROUND_VEDASTRO_WORKERS,
_blocked_foreground_vedastro,
_foreground_vedastro_budget_seconds,
_foreground_vedastro_join_seconds,
_join_foreground_vedastro,
_run_foreground_vedastro_gateway,
finish_foreground_vedastro,
should_include_western_evidence_packet,
start_foreground_vedastro,
)
except ModuleNotFoundError: # pragma: no cover - script execution path
from local_env import load_local_env
from rectification_policy import (
@@ -63,6 +75,18 @@ except ModuleNotFoundError: # pragma: no cover - script execution path
MIN_SCORING_EVENTS,
)
from vedastro_runtime_context import temporary_timeout_seconds
from vedastro_foreground import (
_FOREGROUND_VEDASTRO_EXECUTOR,
_FOREGROUND_VEDASTRO_WORKERS,
_blocked_foreground_vedastro,
_foreground_vedastro_budget_seconds,
_foreground_vedastro_join_seconds,
_join_foreground_vedastro,
_run_foreground_vedastro_gateway,
finish_foreground_vedastro,
should_include_western_evidence_packet,
start_foreground_vedastro,
)
try:
from scripts.unified_consultation_orchestrator import FORMAL_DIVISIONS, UnifiedConsultationOrchestrator
except ModuleNotFoundError: # pragma: no cover - script execution path
@@ -122,11 +146,6 @@ _ASYNC_JOB_EXECUTOR = ThreadPoolExecutor(
max_workers=_ASYNC_JOB_WORKERS,
thread_name_prefix='jyotish-job',
)
_FOREGROUND_VEDASTRO_WORKERS = max(int(os.environ.get('JYOTISH_FOREGROUND_VEDASTRO_WORKERS', '2')), 1)
_FOREGROUND_VEDASTRO_EXECUTOR = ThreadPoolExecutor(
max_workers=_FOREGROUND_VEDASTRO_WORKERS,
thread_name_prefix='jyotish-vedastro-fg',
)
_ASYNC_JOB_CAPACITY = threading.BoundedSemaphore(_ASYNC_JOB_WORKERS + _ASYNC_JOB_QUEUE_SIZE)
_RATE_LIMIT_LOCK = threading.Lock()
_RATE_LIMIT_BUCKETS: dict[str, tuple[float, int]] = {}
@@ -442,6 +461,8 @@ def _evaluate_vedastro_minute_sensitive_pair(request, times: list[str]) -> dict:
try:
reports.append(future.result(timeout=remaining))
except FuturesTimeoutError:
for pending in futures:
pending.cancel()
return _unevaluated_vedastro_minute_sensitive('vedastro_minute_snapshot_timeout')
except Exception:
return _unevaluated_vedastro_minute_sensitive('vedastro_minute_snapshot_error')
@@ -2036,22 +2057,6 @@ _VEDASTRO_COMPACT_DENY_KEYS = {
}
def _foreground_vedastro_budget_seconds() -> float:
raw = str(os.environ.get('JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS', '8')).strip()
try:
return min(max(float(raw), 2.0), 12.0)
except ValueError:
return 8.0
def _foreground_vedastro_join_seconds() -> float:
raw = str(os.environ.get('JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS', '1.5')).strip()
try:
return min(max(float(raw), 0.0), 3.0)
except ValueError:
return 1.5
def _vedastro_natal_signs(raw) -> dict | None:
if not isinstance(raw, dict):
return None
@@ -2107,45 +2112,6 @@ def _compact_vedastro_cross_check(vedastro_official: dict, vedastro_gateway: dic
return compact
def _blocked_foreground_vedastro(*, reason: str, error_type: str | None = None) -> dict:
packet = {
'scope': 'vedastro_gateway_run',
'status': 'official_blocked',
'official_closure_state': 'official_blocked',
'official_closure_reason': reason,
}
if error_type:
packet['error_type'] = error_type
return packet
def _run_foreground_vedastro_gateway(handler, body: dict) -> dict:
try:
with temporary_timeout_seconds(_foreground_vedastro_budget_seconds()):
result = handler._compute_vedastro_gateway_run(body)
except Exception as exc:
return _blocked_foreground_vedastro(
reason='gateway_invocation_error',
error_type=type(exc).__name__,
)
return result if isinstance(result, dict) else _blocked_foreground_vedastro(reason='gateway_invocation_error')
def _join_foreground_vedastro(future, *, timeout: float) -> dict:
if future is None:
return _blocked_foreground_vedastro(reason='foreground_optional_evidence_timeout')
try:
result = future.result(timeout=timeout)
except FuturesTimeoutError:
return _blocked_foreground_vedastro(reason='foreground_optional_evidence_timeout')
except Exception as exc:
return _blocked_foreground_vedastro(
reason='gateway_invocation_error',
error_type=type(exc).__name__,
)
return result if isinstance(result, dict) else _blocked_foreground_vedastro(reason='gateway_invocation_error')
def _blocked_birth_time_sensitivity(*, error_type: str) -> dict:
return {
'schema': 'jyotish.report_birth_time_sensitivity.v1',
@@ -2266,7 +2232,7 @@ def execute_consultation_workflow(
western_evidence_packet=western_evidence_packet,
blind=bool(body.get('blind') or body.get('blind_technical_mode')),
)
if western_evidence_packet:
if western_evidence_packet and should_include_western_evidence_packet(body, surface=surface):
result['western_evidence_packet'] = western_evidence_packet
if body.get('return_high_rigor_shape'):
result['endpoint'] = 'high_rigor_workflow'
@@ -2288,13 +2254,14 @@ def execute_consultation_workflow(
rectification = {}
muhurta_panchanga = {}
computed_chart = bool(chart)
foreground_vedastro_future = None
if defer_optional_external_evidence:
# Overlap a bounded official gateway with local compute. Main-entry
# overview / range scan stay skipped so foreground cannot replay BUG-161.
foreground_vedastro_future = _FOREGROUND_VEDASTRO_EXECUTOR.submit(
_run_foreground_vedastro_gateway, handler, dict(body),
)
# Overlap a bounded official gateway with local compute. Main-entry
# overview / range scan stay skipped so foreground cannot replay BUG-161.
# Same-day snapshot cache skips the submit entirely (BUG-727).
vedastro_session = start_foreground_vedastro(
handler,
dict(body),
defer_optional_external_evidence=defer_optional_external_evidence,
)
for step in runtime_planner.get('sync_steps', []):
if step == 'run_prashna':
@@ -2375,21 +2342,8 @@ def execute_consultation_workflow(
})
executed_steps.append('run_thematic_report')
vedastro_gateway = rectification.get('vedastro_gateway') if isinstance(rectification, dict) else None
if defer_optional_external_evidence:
vedastro_gateway = _join_foreground_vedastro(
foreground_vedastro_future,
timeout=_foreground_vedastro_join_seconds(),
)
elif not isinstance(vedastro_gateway, dict):
try:
vedastro_gateway = handler._compute_vedastro_gateway_run(body)
except Exception as exc: # Gateway evidence must not block the local chart result.
vedastro_gateway = {
'scope': 'vedastro_gateway_run',
'status': 'official_blocked',
'official_closure_reason': 'gateway_invocation_error',
'error_type': type(exc).__name__,
}
if defer_optional_external_evidence or not isinstance(vedastro_gateway, dict):
vedastro_gateway = finish_foreground_vedastro(vedastro_session)
vedastro_official = handler._high_rigor_vedastro_official_summary(chart)
gateway_raw = (
@@ -2524,7 +2478,6 @@ def execute_consultation_workflow(
'interpretation_source_runtime_coverage': interpretation_source_runtime_coverage,
'machine_evidence_packet': machine_evidence_packet,
'consumer_context': consumer_context,
'western_evidence_packet': western_evidence_packet or {},
'real_case_calibration': real_case_calibration,
'birth_time_sensitivity': birth_time_sensitivity,
'runtime_evidence_log': runtime_evidence_log,
@@ -2535,6 +2488,10 @@ def execute_consultation_workflow(
'domain-relevant routes execute according to the configured sample/network limits.'
),
}
if should_include_western_evidence_packet(body, surface=surface):
result['western_evidence_packet'] = western_evidence_packet or {}
if getattr(vedastro_session, 'meta', None):
result['vedastro_snapshot_cache'] = vedastro_session.meta
if high_rigor:
result['high_rigor_external_parity'] = {
'status': 'pass' if external_parity_gate.get('status') == 'pass' else 'blocked',
+2
View File
@@ -75,6 +75,8 @@ CORE_PYTEST_TARGETS = [
"tests/test_birth_time_journey_contract.py",
# Freeze scripts/jyotish_api_server.py growth; new features must be modules.
"tests/test_api_server_growth_contract.py",
# Foreground VedAstro snapshot cache + join cancel (BUG-727 / BUG-728).
"tests/test_vedastro_snapshot_cache.py",
# Native seven-governors adapter and the three read-only chart endpoints.
"tests/test_qizheng_chart_engine.py",
"tests/test_qizheng_api_productization.py",
+250
View File
@@ -0,0 +1,250 @@
"""Foreground VedAstro wait, cancel, and snapshot-cache coordination.
Join, budget, and worker count bound the same wall clock, so they are
declared together and must be changed together.
join: how long the consultation thread waits for the official gateway
(default 1.5s, cap 3s).
budget: how long a worker may keep running after it has started.
workers: process-wide pool. Extra work queues; join still caps the wait.
A timed-out join must cancel its future so queued work never starts.
budget <= k * join (k=2) keeps leftover occupancy of an already-running
worker from stretching later requests past the join cap. The previous
default budget of 8s with a 1.5s join left cancelled-too-late workers
holding the two-thread pool for the rest of the 8s.
"""
from __future__ import annotations
import os
import threading
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from typing import Any
try:
from scripts.vedastro_runtime_context import temporary_timeout_seconds
from scripts.vedastro_snapshot_cache import (
annotate_stale_gateway,
is_cacheable_gateway,
lookup_snapshot,
official_snapshot_reference_date,
requires_today_snapshot,
store_snapshot,
)
except ModuleNotFoundError: # pragma: no cover - script execution path
from vedastro_runtime_context import temporary_timeout_seconds
from vedastro_snapshot_cache import (
annotate_stale_gateway,
is_cacheable_gateway,
lookup_snapshot,
official_snapshot_reference_date,
requires_today_snapshot,
store_snapshot,
)
# --- join / budget / workers (one declaration block; change together) ---
FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT = 1.5
FOREGROUND_VEDASTRO_JOIN_SECONDS_MAX = 3.0
FOREGROUND_VEDASTRO_BUDGET_JOIN_RATIO = 2.0
FOREGROUND_VEDASTRO_WORKERS_DEFAULT = 2
# ----------------------------------------------------------------------
_FOREGROUND_VEDASTRO_WORKERS = max(
int(os.environ.get("JYOTISH_FOREGROUND_VEDASTRO_WORKERS", str(FOREGROUND_VEDASTRO_WORKERS_DEFAULT))),
1,
)
_FOREGROUND_VEDASTRO_EXECUTOR = ThreadPoolExecutor(
max_workers=_FOREGROUND_VEDASTRO_WORKERS,
thread_name_prefix="jyotish-vedastro-fg",
)
def _foreground_vedastro_join_seconds() -> float:
raw = str(os.environ.get("JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS", str(FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT))).strip()
try:
return min(max(float(raw), 0.0), FOREGROUND_VEDASTRO_JOIN_SECONDS_MAX)
except ValueError:
return FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT
def _foreground_vedastro_budget_seconds() -> float:
join = _foreground_vedastro_join_seconds()
cap = FOREGROUND_VEDASTRO_BUDGET_JOIN_RATIO * join
raw = str(os.environ.get("JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS", "")).strip()
if not raw:
return cap
try:
budget = float(raw)
except ValueError:
return cap
if budget < 0:
return cap
return min(budget, cap)
def _blocked_foreground_vedastro(*, reason: str, error_type: str | None = None) -> dict:
packet = {
"scope": "vedastro_gateway_run",
"status": "official_blocked",
"official_closure_state": "official_blocked",
"official_closure_reason": reason,
}
if error_type:
packet["error_type"] = error_type
return packet
def _run_foreground_vedastro_gateway(handler, body: dict, cancel_event: threading.Event | None = None) -> dict:
if cancel_event is not None and cancel_event.is_set():
return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout")
try:
with temporary_timeout_seconds(_foreground_vedastro_budget_seconds()):
result = handler._compute_vedastro_gateway_run(body)
except Exception as exc:
return _blocked_foreground_vedastro(
reason="gateway_invocation_error",
error_type=type(exc).__name__,
)
return result if isinstance(result, dict) else _blocked_foreground_vedastro(reason="gateway_invocation_error")
def _run_foreground_vedastro_gateway_and_store(
handler,
body: dict,
cancel_event: threading.Event | None = None,
) -> dict:
result = _run_foreground_vedastro_gateway(handler, body, cancel_event)
if is_cacheable_gateway(result):
try:
store_snapshot(body, result)
except OSError:
pass
return result
def _join_foreground_vedastro(future, *, timeout: float, cancel_event: threading.Event | None = None) -> dict:
if future is None:
return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout")
try:
result = future.result(timeout=timeout)
except FuturesTimeoutError:
if cancel_event is not None:
cancel_event.set()
future.cancel()
return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout")
except Exception as exc:
return _blocked_foreground_vedastro(
reason="gateway_invocation_error",
error_type=type(exc).__name__,
)
return result if isinstance(result, dict) else _blocked_foreground_vedastro(reason="gateway_invocation_error")
def should_include_western_evidence_packet(body: dict | None, *, surface: str = "") -> bool:
payload = body if isinstance(body, dict) else {}
if payload.get("include_western_evidence_packet") or payload.get("return_western_evidence_packet"):
return True
if payload.get("return_high_rigor_shape"):
return True
if isinstance(payload.get("western_oracle_payload"), dict) or isinstance(payload.get("western_astrology_oracle"), dict):
return True
if isinstance(payload.get("western_evidence_packet"), dict):
return True
return str(surface or "") in {"skill_mcp", "professional_reading_web", "mcp"}
class ForegroundVedastroSession:
def __init__(self, handler, body: dict, *, defer: bool) -> None:
self.handler = handler
self.body = dict(body)
self.defer = bool(defer)
self.cached_gateway: dict | None = None
self.future = None
self.cancel_event = threading.Event()
self.meta: dict[str, Any] = {}
def start(self) -> "ForegroundVedastroSession":
today = official_snapshot_reference_date(self.body)
hit = lookup_snapshot(self.body, today=today)
if hit is not None:
record = hit["record"]
freshness = hit["freshness"]
gateway = record["gateway"]
if freshness == "stale":
gateway = annotate_stale_gateway(
gateway,
reference_date=str(record.get("reference_date") or ""),
served_on_utc_date=today,
)
refresh_body = dict(self.body)
refresh_body["reference_date"] = today
refresh_body["today"] = today
refresh_body["current_date"] = today
self.future = _FOREGROUND_VEDASTRO_EXECUTOR.submit(
_run_foreground_vedastro_gateway_and_store,
self.handler,
refresh_body,
None,
)
self.cached_gateway = gateway
self.meta = {
"freshness": freshness,
"reference_date": str(record.get("reference_date") or today),
"served_on_utc_date": today,
"refresh_submitted": freshness == "stale",
}
return self
self.meta = {
"freshness": "miss",
"reference_date": today,
"served_on_utc_date": today,
"refresh_submitted": False,
"require_today": requires_today_snapshot(self.body),
}
if self.defer:
self.future = _FOREGROUND_VEDASTRO_EXECUTOR.submit(
_run_foreground_vedastro_gateway_and_store,
self.handler,
dict(self.body),
self.cancel_event,
)
return self
def finish(self) -> dict:
if self.cached_gateway is not None:
return self.cached_gateway
if self.defer:
return _join_foreground_vedastro(
self.future,
timeout=_foreground_vedastro_join_seconds(),
cancel_event=self.cancel_event,
)
try:
result = self.handler._compute_vedastro_gateway_run(self.body)
except Exception as exc:
return _blocked_foreground_vedastro(
reason="gateway_invocation_error",
error_type=type(exc).__name__,
)
if not isinstance(result, dict):
return _blocked_foreground_vedastro(reason="gateway_invocation_error")
if is_cacheable_gateway(result):
try:
store_snapshot(self.body, result)
except OSError:
pass
return result
def start_foreground_vedastro(handler, body: dict, *, defer_optional_external_evidence: bool) -> ForegroundVedastroSession:
return ForegroundVedastroSession(
handler,
body,
defer=defer_optional_external_evidence,
).start()
def finish_foreground_vedastro(session: ForegroundVedastroSession | None) -> dict:
if session is None:
return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout")
return session.finish()
+228
View File
@@ -0,0 +1,228 @@
"""Disk cache for foreground VedAstro gateway snapshots.
Keyed by birth data + ayanamsa + node + UTC reference date. This is a
separate store from ``_api_chart_cache`` (BUG-161): different directory
and a different key function. Freshness is the reference date itself
plus a 7-day stale window — there is no independent TTL environment
variable.
Filenames are sha256 hex only. Cache files must not contain names,
emails, or user ids.
"""
from __future__ import annotations
import hashlib
import json
import os
import re
import tempfile
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
CACHE_SCHEMA = "vedastro_snapshot_cache.v1"
MAX_STALE_DAYS = 7
_SHA256_NAME = re.compile(r"^[0-9a-f]{64}\.json$")
_IDENTITY_KEYS = {
"name",
"email",
"user_id",
"userid",
"user_email",
"session_id",
"sessionid",
"full_name",
"display_name",
}
def snapshot_cache_dir() -> Path:
raw = str(os.environ.get("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR") or "").strip()
path = Path(raw) if raw else ROOT / "scratch" / "local" / "vedastro_snapshot_cache"
path.mkdir(parents=True, exist_ok=True)
return path
def official_snapshot_reference_date(case: dict[str, Any] | None) -> str:
"""Same date rule as ``vedastro_service_adapter._official_snapshot_reference_date``.
Duplicated so this module does not import the adapter (and its optional
network stack) at cache-lookup time.
"""
payload = case if isinstance(case, dict) else {}
for key in ("reference_date", "today", "transit_date", "current_date"):
value = payload.get(key)
if not value:
continue
raw = str(value)[:10]
try:
datetime.strptime(raw, "%Y-%m-%d")
return raw
except ValueError:
continue
return datetime.utcnow().strftime("%Y-%m-%d")
def requires_today_snapshot(body: dict[str, Any] | None) -> bool:
payload = body if isinstance(body, dict) else {}
entrypoint = str(payload.get("entrypoint") or payload.get("consult_entrypoint") or "").strip()
return entrypoint == "daily_starlanguage"
def snapshot_cache_key(body: dict[str, Any] | None, *, reference_date: str) -> str:
payload = body if isinstance(body, dict) else {}
ayanamsa = (
payload.get("ayanamsa")
or payload.get("ayanamsa_name")
or payload.get("ayanamsa_policy")
or "raman"
)
node = payload.get("node_mode") or payload.get("nodeMode") or "mean"
material = {
"year": payload.get("year"),
"month": payload.get("month"),
"day": payload.get("day"),
"hour": payload.get("hour"),
"minute": payload.get("minute"),
"second": payload.get("second", 0),
"lat": payload.get("lat"),
"lon": payload.get("lon"),
"tz": payload.get("tz"),
"ayanamsa_policy": str(ayanamsa).strip().lower(),
"node_policy": str(node).strip().lower(),
"reference_date": str(reference_date)[:10],
}
canonical = json.dumps(material, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
def snapshot_cache_path(cache_key: str) -> Path:
if not re.fullmatch(r"[0-9a-f]{64}", cache_key):
raise ValueError("vedastro snapshot cache key must be sha256 hex")
return snapshot_cache_dir() / f"{cache_key}.json"
def is_cacheable_gateway(packet: Any) -> bool:
if not isinstance(packet, dict):
return False
state = packet.get("official_closure_state") or packet.get("status")
if state != "official_verified":
return False
raw = packet.get("official_raw_response") or packet.get("raw_response")
return bool(raw)
def _strip_identity(value: Any) -> Any:
if isinstance(value, dict):
cleaned = {}
for key, item in value.items():
if str(key).strip().lower() in _IDENTITY_KEYS:
continue
cleaned[key] = _strip_identity(item)
return cleaned
if isinstance(value, list):
return [_strip_identity(item) for item in value]
return value
def _load_record(cache_key: str) -> dict[str, Any] | None:
path = snapshot_cache_path(cache_key)
if not _SHA256_NAME.match(path.name) or not path.is_file():
return None
try:
record = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return None
if not isinstance(record, dict) or record.get("schema") != CACHE_SCHEMA:
return None
gateway = record.get("gateway")
if not isinstance(gateway, dict):
return None
reference_date = str(record.get("reference_date") or "")[:10]
try:
datetime.strptime(reference_date, "%Y-%m-%d")
except ValueError:
return None
return record
def lookup_snapshot(
body: dict[str, Any] | None,
*,
today: str | None = None,
require_today: bool | None = None,
) -> dict[str, Any] | None:
payload = body if isinstance(body, dict) else {}
served = today or official_snapshot_reference_date(payload)
must_be_today = requires_today_snapshot(payload) if require_today is None else bool(require_today)
today_key = snapshot_cache_key(payload, reference_date=served)
record = _load_record(today_key)
if record is not None:
return {
"freshness": "fresh",
"record": record,
"served_on_utc_date": served,
}
if must_be_today:
return None
current = datetime.strptime(served, "%Y-%m-%d").date()
for days in range(1, MAX_STALE_DAYS + 1):
past = (current - timedelta(days=days)).isoformat()
record = _load_record(snapshot_cache_key(payload, reference_date=past))
if record is None:
continue
return {
"freshness": "stale",
"record": record,
"served_on_utc_date": served,
}
return None
def store_snapshot(
body: dict[str, Any] | None,
gateway: dict[str, Any],
*,
reference_date: str | None = None,
) -> dict[str, Any] | None:
if not is_cacheable_gateway(gateway):
return None
payload = body if isinstance(body, dict) else {}
stored_date = (reference_date or official_snapshot_reference_date(payload))[:10]
cache_key = snapshot_cache_key(payload, reference_date=stored_date)
record = {
"schema": CACHE_SCHEMA,
"cache_key": cache_key,
"reference_date": stored_date,
"stored_at": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
"gateway": _strip_identity(gateway),
}
path = snapshot_cache_path(cache_key)
path.parent.mkdir(parents=True, exist_ok=True)
encoded = json.dumps(record, ensure_ascii=False, sort_keys=True)
fd, tmp_name = tempfile.mkstemp(prefix=f"{cache_key}.", suffix=".tmp", dir=str(path.parent))
try:
with os.fdopen(fd, "w", encoding="utf-8") as handle:
handle.write(encoded)
os.replace(tmp_name, path)
except Exception:
try:
os.unlink(tmp_name)
except OSError:
pass
raise
return record
def annotate_stale_gateway(gateway: dict[str, Any], *, reference_date: str, served_on_utc_date: str) -> dict[str, Any]:
"""Shallow copy that shows which day a stale snapshot is from.
Does not change ``official_closure_state`` / ``status`` — a previously
delivered official layer stays whatever it was (BUG-301).
"""
packet = dict(gateway)
packet["snapshot_reference_date"] = reference_date
packet["snapshot_freshness"] = "stale"
packet["snapshot_served_on_utc_date"] = served_on_utc_date
return packet
+8
View File
@@ -5,6 +5,8 @@ from __future__ import annotations
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = str(ROOT / "scripts")
WORKBUDDY_SKILL_SCRIPTS = ".workbuddy/skills/jyotish-vedic-astrology/scripts"
@@ -38,6 +40,12 @@ def pytest_runtest_setup() -> None:
ensure_project_scripts_first()
@pytest.fixture(autouse=True)
def _isolate_vedastro_snapshot_cache(tmp_path_factory, monkeypatch) -> None:
cache_dir = tmp_path_factory.mktemp("vedastro_snapshot_cache")
monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR", str(cache_dir))
def pytest_collection_modifyitems(items) -> None:
for item in items:
if item.fspath.basename == "test_api_server_security.py" and item.name.startswith(SLOW_API_SECURITY_PREFIXES):
+350
View File
@@ -0,0 +1,350 @@
"""BUG-727 / BUG-728: foreground VedAstro snapshot cache, cancel, and on-demand western packet."""
from __future__ import annotations
import json
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import pytest
from scripts import jyotish_api_server
from scripts import vedastro_foreground as foreground
from scripts.jyotish_api_server import JyotishAPIHandler
from scripts.vedastro_snapshot_cache import (
lookup_snapshot,
snapshot_cache_dir,
snapshot_cache_key,
snapshot_cache_path,
store_snapshot,
)
def _handler() -> JyotishAPIHandler:
return JyotishAPIHandler.__new__(JyotishAPIHandler)
def _verified_gateway(*, request_id: str, reference_date: str) -> dict:
return {
"scope": "vedastro_gateway_run",
"status": "official_verified",
"official_closure_state": "official_verified",
"official_raw_response": {
"request_id": request_id,
"natal": {"sun": "Leo", "moon": "Taurus", "ascendant": "Cancer"},
"reference_date": reference_date,
},
}
def _fake_chart() -> dict:
return {
"success": True,
"birth_info": {"date": "1997-08-08", "time": "05:00", "tz": 8},
"ascendant": {"lon": 92.0, "sign": "Cancer", "sign_idx": 3},
"planets": {},
"dasha": {"periods": [{"lord": "Sun", "start": "2026-01-01", "end": "2027-01-01"}]},
"modules": {
"varga_full": {"D9": {}, "D10": {}},
"arudha_padas": {"A10": {}, "UL": {}},
"narayana_dasha": {"periods": []},
"ashtakavarga": {"sav": []},
"kp_cusps": {"houses": []},
},
"special_lagnas": {"precision": "sunrise_correct"},
}
def _stub_local(monkeypatch, handler: JyotishAPIHandler, gateway) -> None:
monkeypatch.setattr(handler, "_compute_chart", lambda body: _fake_chart())
monkeypatch.setattr(
handler,
"_compute_rectification_gate",
lambda body: {
"success": True,
"summary": {"recommended_events": [], "warned": [], "disabled": []},
},
)
monkeypatch.setattr(
handler,
"_compute_thematic_report",
lambda body: {"success": True, "endpoint": "thematic_report", "themes": {}},
)
monkeypatch.setattr(handler, "_compute_vedastro_gateway_run", gateway)
def _body(*, reference_date: str, extra: dict | None = None) -> dict:
payload = {
"entry_mode": "direct_chart",
"question": "请直接排盘并重点看事业",
"theme": ["career"],
"year": 1997,
"month": 8,
"day": 8,
"hour": 5,
"minute": 0,
"lat": 36.420487,
"lon": 114.209936,
"tz": 8,
"reference_date": reference_date,
"today": reference_date,
"current_date": reference_date,
"western_mode": False,
"defer_optional_external_evidence": True,
}
if extra:
payload.update(extra)
return payload
@pytest.fixture
def snapshot_cache(tmp_path, monkeypatch):
cache_dir = tmp_path / "vedastro_snapshot_cache"
monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR", str(cache_dir))
return cache_dir
def test_snapshot_cache_dir_is_not_api_chart_cache_dir(snapshot_cache) -> None:
chart_dir = jyotish_api_server._api_chart_cache_dir()
snap_dir = snapshot_cache_dir()
assert snap_dir.resolve() != chart_dir.resolve()
assert snap_dir.name == "vedastro_snapshot_cache"
assert chart_dir.name == "api_chart_cache"
assert snapshot_cache_key is not jyotish_api_server._api_chart_cache_key
def test_snapshot_cache_key_differs_from_api_chart_cache_key() -> None:
body = _body(reference_date="2026-09-15")
snap_key = snapshot_cache_key(body, reference_date="2026-09-15")
chart_key = jyotish_api_server._api_chart_cache_key(
jyotish_api_server._build_api_chart_cache_payload(body)
)
assert snap_key != chart_key
assert len(snap_key) == 64
assert "1997" not in snap_key
assert "36.420487" not in snap_key
assert snapshot_cache_path(snap_key).name == f"{snap_key}.json"
def test_cache_file_strips_identity_and_keeps_hash_filename(snapshot_cache) -> None:
body = _body(reference_date="2026-09-15")
packet = _verified_gateway(request_id="keep-me", reference_date="2026-09-15")
packet["name"] = "Secret Person"
packet["email"] = "user@example.com"
packet["user_id"] = "usr_123"
stored = store_snapshot(body, packet, reference_date="2026-09-15")
assert stored is not None
path = snapshot_cache_path(stored["cache_key"])
text = path.read_text(encoding="utf-8")
assert "Secret Person" not in text
assert "user@example.com" not in text
assert "usr_123" not in text
assert "keep-me" in text
assert path.name.endswith(".json")
assert "1997" not in path.name
def test_second_same_day_consultation_skips_gateway_and_urlopen(snapshot_cache, monkeypatch) -> None:
handler = _handler()
calls = {"gateway": 0, "urlopen": 0}
def counting_gateway(body):
calls["gateway"] += 1
return _verified_gateway(request_id="same-day", reference_date=body.get("reference_date"))
def counting_urlopen(*_args, **_kwargs):
calls["urlopen"] += 1
raise AssertionError("urlopen must not run when the gateway is stubbed")
_stub_local(monkeypatch, handler, counting_gateway)
monkeypatch.setattr("urllib.request.urlopen", counting_urlopen)
body = _body(reference_date="2026-09-15")
first = handler._compute_consultation_workflow(body)
second = handler._compute_consultation_workflow(body)
assert calls["gateway"] == 1
assert calls["urlopen"] == 0
assert first["vedastro_gateway"] == second["vedastro_gateway"]
assert first["vedastro_snapshot_cache"]["freshness"] == "miss"
assert second["vedastro_snapshot_cache"]["freshness"] == "fresh"
def test_next_day_serves_stale_and_refreshes_without_waiting(snapshot_cache, monkeypatch) -> None:
handler = _handler()
yesterday = "2026-09-14"
today = "2026-09-15"
store_snapshot(
_body(reference_date=yesterday),
_verified_gateway(request_id="yesterday", reference_date=yesterday),
reference_date=yesterday,
)
release = threading.Event()
calls = {"gateway": 0}
def slow_today(body):
calls["gateway"] += 1
release.wait(timeout=2.0)
return _verified_gateway(request_id="today", reference_date=today)
_stub_local(monkeypatch, handler, slow_today)
started = time.monotonic()
result = handler._compute_consultation_workflow(_body(reference_date=today))
elapsed = time.monotonic() - started
release.set()
assert elapsed < 1.0
assert result["vedastro_snapshot_cache"]["freshness"] == "stale"
assert result["vedastro_snapshot_cache"]["reference_date"] == yesterday
assert result["vedastro_gateway"]["snapshot_reference_date"] == yesterday
assert result["vedastro_gateway"]["official_closure_state"] == "official_verified"
assert result["vedastro_gateway"]["official_raw_response"]["request_id"] == "yesterday"
deadline = time.monotonic() + 2.0
while calls["gateway"] < 1 and time.monotonic() < deadline:
time.sleep(0.01)
assert calls["gateway"] >= 1
def test_daily_starlanguage_does_not_eat_yesterday_cache(snapshot_cache, monkeypatch) -> None:
handler = _handler()
yesterday = "2026-09-14"
today = "2026-09-15"
store_snapshot(
_body(reference_date=yesterday),
_verified_gateway(request_id="yesterday", reference_date=yesterday),
reference_date=yesterday,
)
calls = {"gateway": 0}
def today_gateway(body):
calls["gateway"] += 1
return _verified_gateway(request_id="today-live", reference_date=today)
_stub_local(monkeypatch, handler, today_gateway)
result = handler._compute_consultation_workflow(
_body(reference_date=today, extra={"entrypoint": "daily_starlanguage"})
)
assert calls["gateway"] == 1
assert result["vedastro_snapshot_cache"]["freshness"] == "miss"
assert result["vedastro_gateway"]["official_raw_response"]["request_id"] == "today-live"
assert lookup_snapshot(
_body(reference_date=today, extra={"entrypoint": "daily_starlanguage"}),
today=today,
)["freshness"] == "fresh"
def test_nth_plus_one_wait_stays_within_join_and_queued_work_is_cancelled(
snapshot_cache, monkeypatch
) -> None:
join = 0.2
monkeypatch.setenv("JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS", str(join))
monkeypatch.setenv("JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS", "8")
handler = _handler()
started: list[int] = []
lock = threading.Lock()
release = threading.Event()
def slow_gateway(body):
with lock:
started.append(1)
release.wait(timeout=8)
return _verified_gateway(request_id="slow", reference_date="2026-09-15")
_stub_local(monkeypatch, handler, slow_gateway)
workers = foreground._FOREGROUND_VEDASTRO_WORKERS
body = _body(reference_date="2026-09-15")
elapsed: list[float] = []
def run_one() -> None:
t0 = time.monotonic()
result = handler._compute_consultation_workflow(body)
elapsed.append(time.monotonic() - t0)
assert result["vedastro_gateway"]["official_closure_reason"] == "foreground_optional_evidence_timeout"
try:
with ThreadPoolExecutor(max_workers=workers + 3) as pool:
futs = [pool.submit(run_one) for _ in range(workers + 3)]
for fut in futs:
fut.result(timeout=8)
assert max(elapsed) <= join + 2.5
assert len(started) <= workers
finally:
release.set()
def test_join_budget_workers_declared_together_and_budget_respects_ratio(monkeypatch) -> None:
source = Path("scripts/vedastro_foreground.py").read_text(encoding="utf-8")
join_idx = source.index("FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT")
ratio_idx = source.index("FOREGROUND_VEDASTRO_BUDGET_JOIN_RATIO")
workers_idx = source.index("FOREGROUND_VEDASTRO_WORKERS_DEFAULT")
span = max(join_idx, ratio_idx, workers_idx) - min(join_idx, ratio_idx, workers_idx)
assert span < 400
assert "change together" in source
monkeypatch.delenv("JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS", raising=False)
monkeypatch.delenv("JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS", raising=False)
join = foreground._foreground_vedastro_join_seconds()
budget = foreground._foreground_vedastro_budget_seconds()
assert budget <= foreground.FOREGROUND_VEDASTRO_BUDGET_JOIN_RATIO * join + 1e-9
monkeypatch.setenv("JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS", "1.5")
monkeypatch.setenv("JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS", "8")
assert foreground._foreground_vedastro_budget_seconds() <= 3.0 + 1e-9
def test_default_consultation_omits_western_packet_and_schema_required_keys_remain(
snapshot_cache, monkeypatch
) -> None:
handler = _handler()
_stub_local(
monkeypatch,
handler,
lambda body: _verified_gateway(request_id="omit-western", reference_date="2026-09-15"),
)
omitted = handler._compute_consultation_workflow(_body(reference_date="2026-09-15"))
included = handler._compute_consultation_workflow(
_body(reference_date="2026-09-15", extra={"include_western_evidence_packet": True})
)
assert "western_evidence_packet" not in omitted
assert "western_spectrum" in omitted["consumer_context"]
assert omitted["success"] is True
assert isinstance(omitted["chart"], dict)
assert isinstance(omitted["routing"], dict)
assert isinstance(omitted["consumer_context"], dict)
assert "western_evidence_packet" in included
workflow_ts = Path("frontend/src/mastra/consultation-workflow.ts").read_text(encoding="utf-8")
schema = workflow_ts.split("export const consultationWorkflowResponseSchema", 1)[1]
assert "success: z.boolean()" in schema
assert "chart: z.record(z.unknown())" in schema
assert "routing: z.record(z.unknown())" in schema
assert "consumer_context:" in schema
assert ").passthrough()" in schema
omitted_chars = len(json.dumps(omitted, ensure_ascii=False))
included_chars = len(json.dumps(included, ensure_ascii=False))
assert included_chars >= omitted_chars
def test_western_oracle_payload_still_returns_packet(snapshot_cache, monkeypatch) -> None:
handler = _handler()
_stub_local(
monkeypatch,
handler,
lambda body: _verified_gateway(request_id="oracle", reference_date="2026-09-15"),
)
result = handler._compute_consultation_workflow(
_body(
reference_date="2026-09-15",
extra={
"western_oracle_payload": {
"source_engine": "kerykeion_external_json",
"natal": {"ascendant": "Virgo", "mc": "Gemini"},
}
},
)
)
assert result["western_evidence_packet"]["source_engine"] == "kerykeion_external_json"