From e61535f464cf43cd444c277243e05c2a9e640c4d Mon Sep 17 00:00:00 2001 From: jesse-ux Date: Wed, 16 Sep 2026 07:27:20 +0800 Subject: [PATCH] =?UTF-8?q?fix(consultation):=20=E5=A4=96=E7=BD=91?= =?UTF-8?q?=E8=AF=81=E6=8D=AE=E6=8C=89=E7=9B=98+=E6=97=A5=E6=9C=9F?= =?UTF-8?q?=E7=BC=93=E5=AD=98=EF=BC=8C=E8=B6=85=E6=97=B6=E5=8F=96=E6=B6=88?= =?UTF-8?q?=E5=89=8D=E5=8F=B0=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BUG-727：同日 VedAstro 快照零等待，跨日先用旧的并后台刷新；join 超时必须 cancel，budget 不超过 2×join。BUG-728：western_evidence_packet 无读取点，默认不再进咨询响应。jyotish_api_server.py 未增长（11334→11291）。 --- CHANGELOG.md | 4 + docs/BUG_HISTORY.md | 33 +- ...tation-external-evidence-cache-20260915.md | 93 +++++ docs/tasks/README.md | 2 +- ...tation-external-evidence-cache-20260915.md | 41 ++ frontend/src/mastra/consultation-tools.ts | 1 + frontend/src/mastra/consultation-workflow.ts | 9 +- .../consultation-workflow-contract.test.ts | 13 + scripts/jyotish_api_server.py | 125 ++----- scripts/run_quality_gate.py | 2 + scripts/vedastro_foreground.py | 250 +++++++++++++ scripts/vedastro_snapshot_cache.py | 228 ++++++++++++ tests/conftest.py | 8 + tests/test_vedastro_snapshot_cache.py | 350 ++++++++++++++++++ 14 files changed, 1072 insertions(+), 87 deletions(-) create mode 100644 docs/tasks/PROGRESS-consultation-external-evidence-cache-20260915.md create mode 100644 docs/testing/consultation-external-evidence-cache-20260915.md create mode 100644 scripts/vedastro_foreground.py create mode 100644 scripts/vedastro_snapshot_cache.py create mode 100644 tests/test_vedastro_snapshot_cache.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 52be9745..d5b58001 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,9 @@ # 印度占星 Skill 更新日志 +## 2026-09-16 — 同一天再问同一张盘，不再重复等外部证据 + +普通聊天同一张盘、同一天里再问，外部证据直接用已经取到的，不再每轮等外网。隔了一天会先用最多七天内的旧证据马上回答，后台再刷新；「深入看今日」仍然只要当天的。西洋盘整包不再塞进每一轮的回答里，压缩后的西洋层还在。Skill 版本不变。 + ## 2026-09-16 — 校正时系统没接上会直说，记下经历后没重算也会说明 生时校正里，如果这边没判断出你在答哪一问，会请你把刚才那句再发一次，而不是说你没说清。记下经历后如果这次没能重新比较候选，会告诉你经历已经记下、稍后再比，不会假装范围已经在收。Skill 版本不变。 diff --git a/docs/BUG_HISTORY.md b/docs/BUG_HISTORY.md index 0fa8bb83..183685d1 100644 --- a/docs/BUG_HISTORY.md +++ b/docs/BUG_HISTORY.md @@ -11285,7 +11285,6 @@ - 相关记录：BUG-473、BUG-249、BUG-250、BUG-260、BUG-617 - 复发自：BUG-473（影响面含本文件，但两条防复发只约束正在流的那一行） - 修复版本：待发布 - ## BUG-726 | 一轮校正把同一份 Case 档案从数据库取多次 - 状态：resolved @@ -11301,3 +11300,35 @@ - 相关记录：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 +- 复发自：无 +- 修复版本：待发布 diff --git a/docs/tasks/PROGRESS-consultation-external-evidence-cache-20260915.md b/docs/tasks/PROGRESS-consultation-external-evidence-cache-20260915.md new file mode 100644 index 00000000..b9c9b441 --- /dev/null +++ b/docs/tasks/PROGRESS-consultation-external-evidence-cache-20260915.md @@ -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 | 避免测试互相命中磁盘缓存 | diff --git a/docs/tasks/README.md b/docs/tasks/README.md index 9f50f30d..f9211371 100644 --- a/docs/tasks/README.md +++ b/docs/tasks/README.md @@ -238,7 +238,7 @@ | `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-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` | — | **记忆三缺口（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-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 + 两个合同测试，不碰业务代码 | 待领取 | — | diff --git a/docs/testing/consultation-external-evidence-cache-20260915.md b/docs/testing/consultation-external-evidence-cache-20260915.md new file mode 100644 index 00000000..0471c1d7 --- /dev/null +++ b/docs/testing/consultation-external-evidence-cache-20260915.md @@ -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 单域耗时已验证」。** diff --git a/frontend/src/mastra/consultation-tools.ts b/frontend/src/mastra/consultation-tools.ts index 2b10e0cc..2c4a820d 100644 --- a/frontend/src/mastra/consultation-tools.ts +++ b/frontend/src/mastra/consultation-tools.ts @@ -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, diff --git a/frontend/src/mastra/consultation-workflow.ts b/frontend/src/mastra/consultation-workflow.ts index 463df743..67d5b2f9 100644 --- a/frontend/src/mastra/consultation-workflow.ts +++ b/frontend/src/mastra/consultation-workflow.ts @@ -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, diff --git a/frontend/tests/consultation-workflow-contract.test.ts b/frontend/tests/consultation-workflow-contract.test.ts index e50f23f8..c092944c 100644 --- a/frontend/tests/consultation-workflow-contract.test.ts +++ b/frontend/tests/consultation-workflow-contract.test.ts @@ -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/); diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index 5e163d8a..f446031b 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -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', diff --git a/scripts/run_quality_gate.py b/scripts/run_quality_gate.py index 2e251ce5..877a27cb 100644 --- a/scripts/run_quality_gate.py +++ b/scripts/run_quality_gate.py @@ -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", diff --git a/scripts/vedastro_foreground.py b/scripts/vedastro_foreground.py new file mode 100644 index 00000000..9cfc85c5 --- /dev/null +++ b/scripts/vedastro_foreground.py @@ -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() diff --git a/scripts/vedastro_snapshot_cache.py b/scripts/vedastro_snapshot_cache.py new file mode 100644 index 00000000..1a7eaa97 --- /dev/null +++ b/scripts/vedastro_snapshot_cache.py @@ -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 diff --git a/tests/conftest.py b/tests/conftest.py index 21b15bca..aa2ed850 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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): diff --git a/tests/test_vedastro_snapshot_cache.py b/tests/test_vedastro_snapshot_cache.py new file mode 100644 index 00000000..89a0d751 --- /dev/null +++ b/tests/test_vedastro_snapshot_cache.py @@ -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"