From 81e2e1cfed51af35141f2a4d5e088bd30ee8e32f Mon Sep 17 00:00:00 2001 From: jesse-ux Date: Wed, 14 Oct 2026 04:53:30 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E5=99=A8=E7=BC=93=E5=AD=98?= =?UTF-8?q?=E6=8C=89=E6=9C=9F=E9=99=90=E6=B8=85=E7=90=86=EF=BC=8C=E5=AE=A1?= =?UTF-8?q?=E8=AE=A1=E5=BF=AB=E7=85=A7=E9=BB=98=E8=AE=A4=E4=B8=8D=E5=86=8D?= =?UTF-8?q?=E8=90=BD=E5=9C=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 排盘缓存、官方快照、负缓存、网关任务和分盘段过期后会删掉。同一张盘的数字写法合成一把缓存键。失败快照只用短期限。自测留存、异步任务、增长合同、安全缓存和隐私标记；一万个文件的清理慢于 2 秒，命中率提升不到 2.5 倍，都按实测写入记录。 Co-Authored-By: Grok (agent loop) --- .env.cn.example | 2 + CHANGELOG.md | 5 + docs/BUG_HISTORY.md | 45 ++ ...GRESS-api-disk-cache-retention-20261013.md | 51 ++ docs/tasks/README.md | 2 +- .../api-disk-cache-retention-ab-20261013.md | 66 ++ scripts/api_scratch_retention.py | 609 ++++++++++++++++++ scripts/jyotish_api_server.py | 70 +- scripts/research/api_chart_cache_byte_diff.py | 28 + scripts/research/api_disk_cache_ab.py | 442 +++++++++++++ scripts/vedastro_gateway.py | 27 +- scripts/vedastro_service_adapter.py | 77 ++- tests/test_api_scratch_retention.py | 314 +++++++++ .../test_vedastro_service_adapter_executor.py | 4 + 14 files changed, 1677 insertions(+), 65 deletions(-) create mode 100644 docs/tasks/PROGRESS-api-disk-cache-retention-20261013.md create mode 100644 docs/testing/api-disk-cache-retention-ab-20261013.md create mode 100644 scripts/api_scratch_retention.py create mode 100644 scripts/research/api_chart_cache_byte_diff.py create mode 100644 scripts/research/api_disk_cache_ab.py create mode 100644 tests/test_api_scratch_retention.py diff --git a/.env.cn.example b/.env.cn.example index 68632a8c..f379f1a5 100644 --- a/.env.cn.example +++ b/.env.cn.example @@ -10,6 +10,8 @@ VEDASTRO_SELF_HOST_ENDPOINT=https://your-domain.example.com/vedastro # VEDASTRO_ENABLE_NETWORK=0 # Cache / TTL / free-tier queue. These keep ordinary users from calling upstream directly. +# 失败结果另有短 TTL（VEDASTRO_OFFICIAL_FULL_SNAPSHOT_NEGATIVE_TTL_SECONDS，默认 300 秒，且不超过成功 TTL）。 +# 这些缓存含出生资料，7 天为上限。 VEDASTRO_CACHE_TTL_SECONDS=604800 VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS=604800 VEDASTRO_GATEWAY_QUEUE_ENABLED=1 diff --git a/CHANGELOG.md b/CHANGELOG.md index d2056bfc..4eedc7ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,10 @@ # 印度占星 Skill 更新日志 +## 2026-10-14 — 服务器不再长期保留出生资料副本 + +- 排盘缓存、官方快照和相关临时文件会按期限删掉，不再一直堆在服务器上。审计用的快照文件默认不再落地。失败的官方快照只短时间复用。 +- Skill 版本不变（BUG-1330～1332）。 + ## 2026-10-14 — 婚恋评级改成证据档位，KP 判宫和大运年长对齐（待合入 staging） - 婚恋结构指数不再写成婚姻稳定、离婚、再婚或建议婚前咨询。中文报告用中文档位，英文报告用英文档位。配偶星分数按 0 到 1 显示，不再写成百分制。 diff --git a/docs/BUG_HISTORY.md b/docs/BUG_HISTORY.md index add22027..23cf56b6 100644 --- a/docs/BUG_HISTORY.md +++ b/docs/BUG_HISTORY.md @@ -17795,3 +17795,48 @@ - 复发自：无。 - 修复版本：未修复。 +## BUG-1330 | 排盘缓存只进不出，同一张盘还会存成多份 + +- 状态：fixed-pending-verify +- 首次发现 / 最近更新：2026-10-13 / 2026-10-14 +- 来源：issue #60。任务书 `docs/tasks/TASK-api-disk-cache-retention-20261013.md` T1、T3。 +- 现象：排盘结果写在磁盘缓存里。过期只在读取时判断，文件一直留着。启动时的清理扫的是异步任务目录，不是这份缓存。同一张盘因为小时、经纬度写成数字或字符串，会各存一份。 +- 触发条件：反复排盘，或同一次出生时间用不同写法提交。服务长时间不重启时，异步任务记录也会一直留到被轮询。 +- 根因：写缓存的人只做了读取时的过期判断。写清理的人以为缓存和异步任务在同一个目录、同一种记录。缓存键直接用了请求里的原值，计算前却会先把这些值解析成数字。 +- 修复：新的留存模块按缓存写入时间和排盘缓存期限删除过期文件，坏文件超过一天也删。启动时清理一次，之后按周期在后台再清。新写入的缓存文件改为仅本人可读，并先写临时文件再换上。缓存键把能解析的数字收成同一种写法，缓存格式版本从 5 改为 6，旧文件自然失效后被清掉。快速排盘和完整排盘仍然是两把不同的键。 +- 验证：`tests/test_api_scratch_retention.py`、`tests/test_api_async_job_contract.py`、`tests/test_api_server_security.py -k cache`、`tests/test_api_server_growth_contract.py`。数值写法的逐字节比较见 `scripts/research/api_chart_cache_byte_diff.py`。 +- 防复发：新增磁盘缓存必须同时在 `scripts/api_scratch_retention.py` 登记留存规则。`tests/test_api_scratch_retention.py` 会检查 `scripts/` 里新的 `scratch/local/` 目录有没有登记。 +- 相关记录：BUG-161、BUG-1331、BUG-1332。 +- 复发自：无。 +- 修复版本：分支 `codex/api-disk-cache-retention-20261013`，尚未合入 staging。 + +## BUG-1331 | VedAstro 审计文件留着出生资料，失败快照被缓存太久 + +- 状态：fixed-pending-verify +- 首次发现 / 最近更新：2026-10-13 / 2026-10-14 +- 来源：issue #60。任务书 T2。 +- 现象：每次官方快照归一都会在磁盘留一份带缩进的审计文件，里面有出生时间。失败、不完整、被挡住的快照和成功结果用同一个长缓存期限。状态接口报出的期限没有算上回落值，可能和实际不一致。 +- 触发条件：在线排盘或报告走到官方快照。样例环境把成功缓存写成 7 天时，一次失败也会被原样复用 7 天。 +- 根因：审计落盘最早只给离线比对用，后来同一条归一进了在线请求，开关没有跟着关掉。语义缓存把「拿到一个结果」当成「可以长期复用」。状态接口自己读了一个变量，没有走实际生效的那个函数。 +- 修复：审计落盘默认关闭，需要比对时再显式打开。打开时改为紧凑文件、仅本人可读、先写临时文件再换上。失败、不完整、被挡住的快照用短期限，默认 300 秒，并且不会长过成功期限。状态接口改为报出含回落的成功期限，并单独报出失败期限。过期的语义缓存、业务响应缓存和审计文件由留存模块删除。 +- 验证：`tests/test_api_scratch_retention.py` 覆盖默认关闭、打开后的文件、失败 301 秒后不再命中、状态接口回落。既有 VedAstro 适配器测试仍通过。需要审计文件的那一条测试改为显式打开开关。 +- 防复发：新增磁盘缓存必须同时在 `scripts/api_scratch_retention.py` 登记留存规则。 +- 相关记录：BUG-734、BUG-301、BUG-1330。 +- 复发自：无。 +- 修复版本：分支 `codex/api-disk-cache-retention-20261013`，尚未合入 staging。 + +## BUG-1332 | 前台快照、负缓存、网关任务和分盘段缓存没有清理 + +- 状态：fixed-pending-verify +- 首次发现 / 最近更新：2026-10-13 / 2026-10-14 +- 来源：issue #60。任务书 T4。 +- 现象：前台快照缓存超过回看窗口后不会再被读，但文件不删。负缓存只认当天，昨天的文件一直在。网关任务和分盘段缓存也只增不减。这些目录在持久卷上，部署不会清空。 +- 触发条件：使用前台官方快照、咨询负缓存、网关队列或校正分盘段。 +- 根因：当初只规定了什么时候还能读，没有规定什么时候可以删。 +- 修复：前台快照早于「今天减去回看天数再减一天」就删。负缓存早于今天（UTC）就删。已完成或失败、并且超过保留时间的网关任务连同原始响应一起删；排队中和进行中的不删。分盘段按修改时间保留，默认 7 天。只删符合原有文件名的文件，其它文件不碰。读写和缓存键都没有改。 +- 验证：`tests/test_api_scratch_retention.py` 覆盖这四个目录的留、删和不碰。`tests/test_vedastro_snapshot_cache.py` 与 `tests/test_vedastro_consultation_snapshot.py` 断言未改。 +- 防复发：新增磁盘缓存必须同时在 `scripts/api_scratch_retention.py` 登记留存规则。 +- 相关记录：BUG-727、BUG-1231、BUG-1330。 +- 复发自：无。 +- 修复版本：分支 `codex/api-disk-cache-retention-20261013`，尚未合入 staging。 + diff --git a/docs/tasks/PROGRESS-api-disk-cache-retention-20261013.md b/docs/tasks/PROGRESS-api-disk-cache-retention-20261013.md new file mode 100644 index 00000000..b60a458c --- /dev/null +++ b/docs/tasks/PROGRESS-api-disk-cache-retention-20261013.md @@ -0,0 +1,51 @@ +# PROGRESS · API 磁盘缓存留存（2026-10-14） + +- 分支：`codex/api-disk-cache-retention-20261013` +- 任务书基线写的是 `77057a819`。本分支从当时的 `origin/staging` `66cd60081f37a57680274633fcb3fb942d01bb2f` 拉出。A/B 的父提交用的是这个 SHA，不是任务书里的旧基线。 +- 接口文件行数：10946（验收口径 ≤ 10970）。类方法数和伪造点没有增加。 + +## 做了什么 + +- T1：新增 `scripts/api_scratch_retention.py`。启动时清理全部已登记目录，并按 `JYOTISH_SCRATCH_PRUNE_INTERVAL_SECONDS`（默认 1800，0 表示不启后台线程）周期清理。排盘缓存新文件改为仅本人可读，先写临时文件再换上。`prune_expired_async_jobs` 只留同名薄包装。 +- T2：`VEDASTRO_WRITE_AUDIT_ARTIFACTS` 默认关。关掉时不写审计文件，`artifact_path` 为 `None`。打开时写紧凑 JSON，仅本人可读，先写临时文件再换上。成功快照仍用原 TTL；失败、不完整、被挡住的用短 TTL，默认 300 秒，并且不超过成功 TTL。旧缓存没有 `negative` 标记时仍按成功的长期限读。状态接口改报含回落的成功期限，并多报失败期限。`.env.cn.example` 只加了注释，两个 604800 没改。 +- 成功的判据：`available is True` 且 `status == "ok"`。`partial` 即使 `available` 为真也算失败缓存。 +- T3：年月日能整除才收成整数，不截断 1990.5。时分秒和经纬度、时区能解析成有限数字就收成浮点，不四舍五入。解析不了的保持原值。日期取前 10 位，空值仍是空值。缓存格式版本 5 → 6。快速排盘和完整排盘仍是两把键。 +- T4：前台快照、负缓存、网关任务、分盘段都进了登记表。只删符合原文件名的过期文件。网关排队中和进行中的不删。读写逻辑和键没改。 +- T5：脚本在 `scripts/research/api_disk_cache_ab.py`，逐字节比较在 `scripts/research/api_chart_cache_byte_diff.py`。记录在 `docs/testing/api-disk-cache-retention-ab-20261013.md`。 +- T6：没有改咨询卡片缓存。数字用任务书 §2.5，没有重测：未命中构建 1.9 秒；文件 5097751 字节；命中时解析中位 68.5 毫秒；整次命中 82.5 毫秒；两天大约每个活跃用户 10 MB。可选方向仍是两条：按领域切片只读需要的部分；清理改为每隔几分钟一次，而不是每次读取都扫全目录。没有做进程内已解析包缓存。 +- T7：BUG-1330、BUG-1331、BUG-1332。变更说明一句。任务索引改为待验收。没有改界面说明。 + +## 命令和结果 + +| 检查 | 命令 | 结果 | +| --- | --- | --- | +| 新留存、异步任务、增长合同 | `pytest tests/test_api_scratch_retention.py tests/test_api_async_job_contract.py tests/test_api_server_growth_contract.py -q` | 26 通过 | +| 安全缓存 | `pytest tests/test_api_server_security.py -q -k cache` | 6 通过 | +| 任务书列出的 VedAstro / 咨询套件 | 见任务书 §8 那一条 | 4 失败，见下。其余通过 | +| 需要审计文件的执行器测试 | `pytest tests/test_vedastro_service_adapter_executor.py::test_vedastro_range_scan_records_hashes_and_artifact_path -q` | 1 通过 | +| 逐字节比较与 A/B | `python scripts/research/api_disk_cache_ab.py` | 退出码 0。命中率 B/A = 1.333。七天后 B 剩 40 个文件，等于当天键数。审计文件 0。负缓存剩 1。一万个文件的清理 20.174 秒 | +| 接口行数 | 按换行计数 | 10946 | + +既有断言改动只有一处，在 `tests/test_vedastro_service_adapter_executor.py`：原值是子进程沿用进程环境；新值是显式打开审计落盘；原因是默认关闭后这条测试仍要检查文件存在。这是测试前置，不是放宽断言。 + +## 未达到的目标（目标不改） + +- 混合写法再重复 3 次时，命中率只是 A 的 1.333 倍，不是 2.5 倍。重复本身已经让 A 命中三分之二。 +- 一万个过期文件的清理用了 20.174 秒，不是 2 秒内。清理不在请求线程。默认间隔和失败短期限没有因此改掉。 + +## 本机套件里原先就会红的三条 + +- `tests/test_vedastro_gateway.py` 要求原始响应文件权限是 0600。Windows 报出来的是可写，不是 0600。这条断言没改。新测试在 Linux 上仍检查 0600，Windows 上只检查文件写出来了。 +- `tests/test_vedastro_runtime_ops.py` 要求已安装的 VedAstro 是 1.23.25，本机是 1.23.30。与这次缓存改动无关。 +- `tests/test_three_engine_parity_runner.py` 的公开盘回放把 PyJHora 记成 raw_imported，不是 structured_captured，同时在下载星历。运行器没改。 + +## 环境缺口 + +- 生产和测试环境里成功缓存的实际秒数，这份仓库读不到。部署后看状态接口的两个期限数字即可。 +- 生产卷上已经堆了多少文件，这里不知道。 +- 主测量挡住了 VedAstro 导入。本机有 SDK，官方端口能连通。不挡 SDK 的探针仍关闭外网取数，一次完整排盘 4.167 秒，没有报错，没有拉真实快照。 + +## 未做 + +- 没有改部署文件、生产环境变量、数据库，也没有手工清生产卷。 +- 没有把咨询卡片缓存并进周期清理。 diff --git a/docs/tasks/README.md b/docs/tasks/README.md index 58b326a6..76572ba0 100644 --- a/docs/tasks/README.md +++ b/docs/tasks/README.md @@ -470,7 +470,7 @@ | `TASK-rect-scoring-perf-20261013.md` | — | 校正打分提速：引导窗口平方扫描、候选不变量重复算、响应带无读取点字段、候选数上限 241（BUG-1323～1325、1345） | 待领取 | `codex/rect-scoring-perf-20261013` | | `TASK-node-runtime-resources-20261013.md` | — | 报告 worker abort 监听泄漏、本地 PG 适配层往返与元数据缓存、无上限内存表、用户端会话读缓存（BUG-1326～1328、1346，PG 与缓存部分 A/B） | 待领取 | `codex/node-runtime-resources-20261013` | | `TASK-home-sidebar-rerender-20261013.md` | — | 流式回答时侧栏每帧整栏重渲染（BUG-1329，A/B） | 待领取 | `codex/home-sidebar-rerender-20261013` | -| `TASK-api-disk-cache-retention-20261013.md` | — | API 磁盘缓存只进不出、VedAstro 审计文件留存出生资料、失败快照久缓存、缓存键规范化（BUG-1330～1332，A/B） | 待领取 | `codex/api-disk-cache-retention-20261013` | +| `TASK-api-disk-cache-retention-20261013.md` | `PROGRESS-api-disk-cache-retention-20261013.md` | API 磁盘缓存只进不出、VedAstro 审计文件留存出生资料、失败快照久缓存、缓存键规范化（BUG-1330～1332，A/B） | 待验收 | `codex/api-disk-cache-retention-20261013` | | `TASK-report-reference-perf-20261013.md` | — | 长报告出稿提速：术语替换命中预检、哈希单次遍历、同盘复用（BUG-1335～1336，A/B） | 待领取 | `codex/report-reference-perf-20261013` | | `TASK-consult-prep-parallel-20261013.md` | — | 普通对话发问前准备步骤并行、模型版本短缓存、多领域并发开关（BUG-1337～1339，A/B） | 待领取 | `codex/consult-prep-parallel-20261013` | | `TASK-frontend-dead-code-20261013.md` | — | 前端死代码 17 文件与无调用方接口删除，含未登录重计算接口（BUG-1342） | 待领取 | `codex/frontend-dead-code-20261013` | diff --git a/docs/testing/api-disk-cache-retention-ab-20261013.md b/docs/testing/api-disk-cache-retention-ab-20261013.md new file mode 100644 index 00000000..7086e60d --- /dev/null +++ b/docs/testing/api-disk-cache-retention-ab-20261013.md @@ -0,0 +1,66 @@ +# 磁盘缓存留存 A/B（2026-10-14） + +对照的是本分支的父提交 `66cd60081f37a57680274633fcb3fb942d01bb2f`（建分支时的 `origin/staging`）。任务书里写的 `77057a819` 已经不是当前测试线头。A 用 `git show` 把父提交的接口文件载入同一进程，并把它的仓库路径指回本工作树，否则临时副本找不到引擎。B 是本分支。 + +演练用虚构盘，不写真实出生资料。文件都在系统临时目录，不写工作树里的 `scratch/`。 + +主测量挡住了 `jhora` 和 `vedastro` 的导入。本机装了 VedAstro SDK，并且能连上官方端口；另有一组不挡 SDK、但仍关闭外网取数，见文末。 + +## 命令 + +```powershell +.venv\Scripts\python.exe scripts\research\api_chart_cache_byte_diff.py +.venv\Scripts\python.exe scripts\research\api_disk_cache_ab.py +.venv\Scripts\python.exe scripts\research\api_disk_cache_ab.py --with-sdk --sdk-probe +``` + +## 命中率 + +20 张虚构盘 × 3 种写法 × 快速/完整 × 7 天 × 每种重复 3 次 = 2520 次取键。取键用的是真实缓存键函数，没有把 2520 次都做成完整排盘。 + +| | 改前 A | 改后 B | 变化 | +| --- | --- | --- | --- | +| 不同缓存键 | 840 | 280 | 少 560 | +| 命中次数 | 1680 | 2240 | 多 560 | +| 命中率 | 0.6667 | 0.8889 | B / A = 1.333 | + +目标是混合写法下 B 至少为 A 的 2.5 倍。按任务书规定的「重复 3 次」算，没有达到：A 自己靠重复就已经有三分之二命中，写法合并只能再抬到约 89%。若不算这 3 次重复、只看三种写法各问一次，A 的命中率是 0，B 是 0.6667。目标数字不改。 + +## 七天后的文件 + +留存演练写的是短记录，用来数文件，不是完整盘面。期限 900 秒，只有当天的键还没过期。 + +| | 改前 A | 改后 B | +| --- | --- | --- | +| 清理前文件数 | 840 | 280 | +| 清理前字节 | 109200 | 36400 | +| 清理后文件数 | 840（旧代码没有这份清理） | 40 | +| 当天仍有效的键 | 120 | 40 | + +B 清理后的文件数等于当天有效键数，达到「不超过当天活跃键」的目标。 + +审计文件：B 在默认关闭时写出 0 个。达到目标。 + +负缓存：种入当天和前一天各一份，清理后剩 1 份（当天）。达到目标。 + +## 清理耗时 + +| 场景 | 结果 | +| --- | --- | +| 预置 10000 个已过期排盘缓存后，整次清理 | 20.174 秒，排盘目录 20170.902 毫秒，删掉 10000，剩下 0 | +| 上面七天演练里的排盘目录 | 370.161 毫秒 | + +目标是 10000 个文件在 2 秒内、且不在请求线程。本次在后台函数里测，不在请求线程；耗时没有进 2 秒。这是 Windows 上逐个读文件再删除的实测，目标不改。默认清理间隔仍是 1800 秒，失败快照短期限仍是 300 秒，不因为这次没进 2 秒就改默认值。 + +## 排盘耗时（一次虚构盘，快速排盘，外网关闭） + +| | 中位数 | +| --- | --- | +| 未命中 3 次 | 23.036 毫秒 | +| 命中 3 次 | 5.249 毫秒 | + +## 响应是否还是同一张盘 + +数值等价的写法（小时 12 / 12.0 / "12"，纬度数字和字符串，时区 5.5 和 "5.5"，秒缺省和 0）在快速排盘下去掉 `runtime_cache` 后逐字节相同。完整排盘只差官方快照缓存上的写入时刻和到期时刻，盘面没有别的差异。父提交和本分支对同一张快速排盘、去掉 `runtime_cache` 后逐字节相同。因此这些字段可以合并进同一把缓存键。 + +不挡 SDK 的一组：本机有 VedAstro SDK，官方端口能连通。探针仍关闭外网取数。一次完整排盘 4.167 秒，没有报错。没有拉取真实快照。 diff --git a/scripts/api_scratch_retention.py b/scripts/api_scratch_retention.py new file mode 100644 index 00000000..4a31458a --- /dev/null +++ b/scripts/api_scratch_retention.py @@ -0,0 +1,609 @@ +"""Retention rules for scratch/local caches. + +Counts only. Never prints file names or cache payloads. +""" +from __future__ import annotations + +import json +import math +import os +import re +import secrets +import threading +import time +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any, Callable + +ROOT = Path(__file__).resolve().parents[1] +UNREADABLE_KEEP_SECONDS = 24 * 60 * 60 +_HEX64_JSON = re.compile(r"^[0-9a-f]{64}\.json$") +_ARTIFACT_JSON = re.compile(r"^[A-Za-z0-9_]+-[0-9a-f]{12}-[0-9a-f]{12}\.json$") +_GATEWAY_JOB_JSON = re.compile(r"^vgw_[0-9a-f]{20}\.json$") +_PRUNE_THREAD_NAME = "jyotish-scratch-prune" + +# Directory names under scratch/local that this module already knows how to prune. +# Subdirectories of vedastro_adapter are listed so a future writer is not a new top-level dir. +REGISTERED_DIR_NAMES = frozenset( + { + "api_chart_cache", + "high_rigor_jobs", + "api_chart_response_jobs", + "async_jobs.sqlite3", + "vedastro_adapter", + "vedastro_snapshot_cache", + "vedastro_snapshot_negative_cache", + "vedastro_gateway_jobs", + "varga_segments", + } +) + +# Offline tools or caches that already prune themselves. Not a new retention rule. +OFFLINE_SCRATCH_NAMES = frozenset( + { + "consult_full_data_cache", + "ocr_extract", + "three_engine_parity", + "public_release_deny_patterns.txt", + "pdf_review_123456", + } +) + +REGISTRY: tuple[dict[str, Any], ...] = ( + { + "name": "api_chart_cache", + "holds_birth_data": True, + "deletable": "created_at plus chart TTL, or unreadable json older than 24h", + }, + { + "name": "high_rigor_jobs", + "holds_birth_data": True, + "deletable": "expires_at_unix has passed; bad json is kept", + }, + { + "name": "api_chart_response_jobs", + "holds_birth_data": True, + "deletable": "expires_at_unix has passed; bad json is kept", + }, + { + "name": "async_jobs.sqlite3", + "holds_birth_data": True, + "deletable": "sqlite rows whose expires_at has passed", + }, + { + "name": "vedastro_adapter", + "holds_birth_data": True, + "deletable": "audit artifacts, response_cache, and official_full_snapshot_cache", + }, + { + "name": "vedastro_snapshot_cache", + "holds_birth_data": True, + "deletable": "reference_date before today UTC minus MAX_STALE_DAYS minus 1", + }, + { + "name": "vedastro_snapshot_negative_cache", + "holds_birth_data": True, + "deletable": "reference_date before today UTC", + }, + { + "name": "vedastro_gateway_jobs", + "holds_birth_data": True, + "deletable": "completed or failed jobs older than the keep window, plus their raw archive", + }, + { + "name": "varga_segments", + "holds_birth_data": True, + "deletable": "mtime older than JYOTISH_VARGA_SEGMENTS_KEEP_DAYS", + }, +) + + +def write_private_json(path: Path, payload: Any) -> None: + """Atomic UTF-8 JSON, mode 0600. Compact, stable key order.""" + path.parent.mkdir(parents=True, exist_ok=True) + temp = path.with_name(f"{path.name}.{secrets.token_hex(8)}.tmp") + temp.write_text( + json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")), + encoding="utf-8", + ) + try: + os.chmod(temp, 0o600) + except OSError: + pass + os.replace(temp, path) + try: + os.chmod(path, 0o600) + except OSError: + pass + + +def _finite_float(value: Any) -> float | None: + if isinstance(value, bool) or value is None: + return None + if isinstance(value, (int, float)): + number = float(value) + elif isinstance(value, str): + text = value.strip() + if not text: + return None + try: + number = float(text) + except ValueError: + return None + else: + return None + if not math.isfinite(number): + return None + return number + + +def _exact_int(value: Any) -> int | None: + if isinstance(value, bool) or value is None: + return None + if isinstance(value, int): + return value + if isinstance(value, float): + if not math.isfinite(value) or not value.is_integer(): + return None + return int(value) + if isinstance(value, str) and re.fullmatch(r"[+-]?\d+", value.strip()): + return int(value.strip()) + return None + + +def normalize_chart_cache_birth(body: dict | None) -> dict[str, Any]: + """Cache-key birth fields. Unparseable values stay as they arrived.""" + payload = body if isinstance(body, dict) else {} + normalized: dict[str, Any] = {} + for key in ("year", "month", "day"): + parsed = _exact_int(payload.get(key)) + normalized[key] = payload.get(key) if parsed is None else parsed + for key in ("hour", "minute", "lat", "lon", "tz"): + raw = payload.get(key) + parsed = _finite_float(raw) + normalized[key] = raw if parsed is None else parsed + second_raw = payload.get("second", 0) + second = _finite_float(second_raw) + normalized["second"] = second_raw if second is None else second + return normalized + + +def normalize_chart_cache_day(value: Any) -> Any: + if value is None or value == "": + return value + return str(value)[:10] + + +def _blank() -> dict[str, int]: + return {"scanned": 0, "removed": 0, "errors": 0} + + +def _older_than(path: Path, moment: float, seconds: float) -> bool: + try: + return moment - path.stat().st_mtime > seconds + except OSError: + return False + + +def _unlink(path: Path, stats: dict[str, int]) -> None: + try: + path.unlink() + stats["removed"] += 1 + except OSError: + stats["errors"] += 1 + + +def _repo_root() -> Path: + return ROOT + + +def _api_module(api_module=None): + if api_module is not None: + return api_module + import jyotish_api_server as api + + return api + + +def prune_async_job_records(now: float | None = None, api_module=None) -> dict[str, Any]: + """Same file and sqlite rules as the old startup cleanup. Bad job JSON is kept.""" + api = _api_module(api_module) + moment = time.time() if now is None else float(now) + scanned = 0 + removed = 0 + errors = 0 + directories: dict[str, dict[str, int]] = {} + if api._async_job_backend() == "sqlite": + try: + with api._sqlite_job_connection() as connection: + scanned = int(connection.execute("SELECT COUNT(*) FROM async_jobs").fetchone()[0]) + deleted = connection.execute( + "DELETE FROM async_jobs WHERE expires_at IS NOT NULL AND expires_at <= ?", + (moment,), + ).rowcount + removed = int(deleted or 0) + except OSError: + errors += 1 + directories["high_rigor_jobs"] = _blank() + directories["api_chart_response_jobs"] = _blank() + directories["async_jobs_sqlite"] = {"scanned": scanned, "removed": removed, "errors": errors} + return { + "scope": "async_job_cleanup", + "scanned": scanned, + "removed": removed, + "errors": errors, + "directories": directories, + } + + for scope, label in ( + (api._HIGH_RIGOR_JOB_SCOPE, "high_rigor_jobs"), + (api._API_CHART_CACHE_SCOPE, "api_chart_response_jobs"), + ): + stats = _blank() + try: + directory = api._async_job_dir(scope) + except OSError: + stats["errors"] += 1 + directories[label] = stats + errors += stats["errors"] + continue + if directory.is_dir(): + for path in directory.glob("*.json"): + stats["scanned"] += 1 + try: + record = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + continue + except OSError: + stats["errors"] += 1 + continue + expires_at = record.get("expires_at_unix") if isinstance(record, dict) else None + if isinstance(expires_at, (int, float)) and not isinstance(expires_at, bool) and moment >= float(expires_at): + _unlink(path, stats) + directories[label] = stats + scanned += stats["scanned"] + removed += stats["removed"] + errors += stats["errors"] + return { + "scope": "async_job_cleanup", + "scanned": scanned, + "removed": removed, + "errors": errors, + "directories": directories, + } + + +def _prune_chart_cache(moment: float, api_module=None) -> dict[str, int]: + api = _api_module(api_module) + stats = _blank() + try: + directory = api._api_chart_cache_dir() + ttl = api._api_chart_cache_ttl_seconds() + except OSError: + stats["errors"] += 1 + return stats + if not directory.is_dir(): + return stats + for path in directory.glob("*.json"): + if not _HEX64_JSON.match(path.name): + continue + stats["scanned"] += 1 + try: + record = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + if _older_than(path, moment, UNREADABLE_KEEP_SECONDS): + _unlink(path, stats) + continue + except OSError: + stats["errors"] += 1 + continue + created_at = record.get("created_at") if isinstance(record, dict) else None + if not isinstance(created_at, (int, float)) or isinstance(created_at, bool): + if _older_than(path, moment, UNREADABLE_KEEP_SECONDS): + _unlink(path, stats) + continue + if moment - float(created_at) > ttl: + _unlink(path, stats) + return stats + + +def _adapter(): + import vedastro_service_adapter as adapter + + return adapter + + +def _prune_created_at_dir(directory: Path, moment: float, ttl_for_record: Callable[[dict], float]) -> dict[str, int]: + stats = _blank() + if not directory.is_dir(): + return stats + for path in directory.glob("*.json"): + if not _HEX64_JSON.match(path.name): + continue + stats["scanned"] += 1 + try: + record = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + if _older_than(path, moment, UNREADABLE_KEEP_SECONDS): + _unlink(path, stats) + continue + except OSError: + stats["errors"] += 1 + continue + created_at = record.get("created_at") if isinstance(record, dict) else None + if not isinstance(created_at, (int, float)) or isinstance(created_at, bool): + if _older_than(path, moment, UNREADABLE_KEEP_SECONDS): + _unlink(path, stats) + continue + try: + ttl = float(ttl_for_record(record)) + except (TypeError, ValueError): + stats["errors"] += 1 + continue + if moment - float(created_at) > ttl: + _unlink(path, stats) + return stats + + +def _prune_response_cache(moment: float) -> dict[str, int]: + adapter = _adapter() + return _prune_created_at_dir( + Path(adapter.ARTIFACT_DIR) / "response_cache", + moment, + lambda _record: adapter._cache_ttl_seconds(), + ) + + +def _prune_snapshot_semantic_cache(moment: float) -> dict[str, int]: + adapter = _adapter() + + def ttl_for(record: dict) -> float: + if record.get("negative") is True: + return adapter._official_full_snapshot_negative_ttl_seconds() + return adapter._official_full_snapshot_cache_ttl_seconds() + + return _prune_created_at_dir( + Path(adapter.ARTIFACT_DIR) / "official_full_snapshot_cache", + moment, + ttl_for, + ) + + +def _audit_keep_seconds() -> float: + raw = os.environ.get("VEDASTRO_AUDIT_ARTIFACT_KEEP_DAYS", "7").strip() + try: + days = float(raw) if raw else 7.0 + except ValueError: + days = 7.0 + return max(0.0, days) * 86400.0 + + +def _prune_audit_artifacts(moment: float) -> dict[str, int]: + adapter = _adapter() + stats = _blank() + directory = Path(adapter.ARTIFACT_DIR) + if not directory.is_dir(): + return stats + enabled = adapter._audit_artifacts_enabled() + max_age = _audit_keep_seconds() if enabled else UNREADABLE_KEEP_SECONDS + for path in directory.glob("*.json"): + if not path.is_file() or not _ARTIFACT_JSON.match(path.name): + continue + stats["scanned"] += 1 + if _older_than(path, moment, max_age): + _unlink(path, stats) + return stats + + +def _parse_day(value: Any): + if not isinstance(value, str): + return None + raw = value[:10] + try: + return datetime.strptime(raw, "%Y-%m-%d").date() + except ValueError: + return None + + +def _prune_reference_dated(directory: Path, moment: float, *, earliest_kept) -> dict[str, int]: + stats = _blank() + if not directory.is_dir(): + return stats + for path in directory.glob("*.json"): + if not _HEX64_JSON.match(path.name): + continue + stats["scanned"] += 1 + try: + record = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + if _older_than(path, moment, UNREADABLE_KEEP_SECONDS): + _unlink(path, stats) + continue + except OSError: + stats["errors"] += 1 + continue + ref = _parse_day(record.get("reference_date") if isinstance(record, dict) else None) + if ref is None: + if _older_than(path, moment, UNREADABLE_KEEP_SECONDS): + _unlink(path, stats) + continue + if ref < earliest_kept: + _unlink(path, stats) + return stats + + +def _prune_vedastro_snapshot_cache(moment: float) -> dict[str, int]: + from vedastro_snapshot_cache import MAX_STALE_DAYS + + today = datetime.fromtimestamp(moment, timezone.utc).date() + earliest = today - timedelta(days=int(MAX_STALE_DAYS) + 1) + raw = os.environ.get("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR", "").strip() + directory = Path(raw) if raw else _repo_root() / "scratch" / "local" / "vedastro_snapshot_cache" + return _prune_reference_dated(directory, moment, earliest_kept=earliest) + + +def _prune_vedastro_negative_cache(moment: float) -> dict[str, int]: + today = datetime.fromtimestamp(moment, timezone.utc).date() + raw = os.environ.get("JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR", "").strip() + directory = Path(raw) if raw else _repo_root() / "scratch" / "local" / "vedastro_snapshot_negative_cache" + return _prune_reference_dated(directory, moment, earliest_kept=today) + + +def _parse_updated_at(value: Any) -> float | None: + if not isinstance(value, str) or not value.strip(): + return None + text = value.strip().replace("Z", "+00:00") + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.timestamp() + + +def _gateway_keep_seconds() -> float: + raw = os.environ.get("VEDASTRO_GATEWAY_JOB_KEEP_SECONDS", "86400").strip() + try: + return max(0.0, float(raw) if raw else 86400.0) + except ValueError: + return 86400.0 + + +def _prune_gateway_jobs(moment: float) -> dict[str, int]: + stats = _blank() + raw = os.environ.get("VEDASTRO_GATEWAY_QUEUE_DIR", "").strip() + directory = Path(raw).expanduser() if raw else _repo_root() / "scratch" / "local" / "vedastro_gateway_jobs" + if not directory.is_dir(): + return stats + keep = _gateway_keep_seconds() + for path in directory.glob("*.json"): + if not _GATEWAY_JOB_JSON.match(path.name): + continue + stats["scanned"] += 1 + try: + record = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + continue + except OSError: + stats["errors"] += 1 + continue + if not isinstance(record, dict): + continue + status = str(record.get("status") or "") + if status not in {"completed", "failed"}: + continue + updated = _parse_updated_at(record.get("updated_at")) + if updated is None or moment - updated <= keep: + continue + companion = directory / f"{path.stem}.official_raw_response.json" + _unlink(path, stats) + if companion.is_file(): + stats["scanned"] += 1 + _unlink(companion, stats) + return stats + + +def _varga_keep_seconds() -> float: + raw = os.environ.get("JYOTISH_VARGA_SEGMENTS_KEEP_DAYS", "7").strip() + try: + days = float(raw) if raw else 7.0 + except ValueError: + days = 7.0 + return max(0.0, days) * 86400.0 + + +def _prune_varga_segments(moment: float) -> dict[str, int]: + stats = _blank() + raw = os.environ.get("JYOTISH_CACHE_DIR", "").strip() + root = Path(raw) if raw else _repo_root() / "scratch" / "local" + directory = root / "varga_segments" + if not directory.is_dir(): + return stats + max_age = _varga_keep_seconds() + for path in directory.glob("*.json"): + if not _HEX64_JSON.match(path.name): + continue + stats["scanned"] += 1 + if _older_than(path, moment, max_age): + _unlink(path, stats) + return stats + + +def _stamp(stats: dict[str, int], started: float) -> dict[str, Any]: + return { + "scanned": int(stats.get("scanned", 0) or 0), + "removed": int(stats.get("removed", 0) or 0), + "errors": int(stats.get("errors", 0) or 0), + "elapsed_ms": round((time.perf_counter() - started) * 1000, 3), + } + + +def prune_all(now: float | None = None, api_module=None) -> dict[str, Any]: + """Prune every registered directory. Exceptions become error counts.""" + moment = time.time() if now is None else float(now) + api = _api_module(api_module) + started = time.perf_counter() + directories: dict[str, dict[str, Any]] = {} + + def _run(name: str, func: Callable[[float], dict[str, int]]) -> None: + t0 = time.perf_counter() + try: + stats = func(moment) + except Exception: + stats = {"scanned": 0, "removed": 0, "errors": 1} + directories[name] = _stamp(stats, t0) + + _run("api_chart_cache", lambda moment: _prune_chart_cache(moment, api)) + job_started = time.perf_counter() + try: + job = prune_async_job_records(moment, api) + job_dirs = job.get("directories") or {} + except Exception: + job_dirs = { + "high_rigor_jobs": {"scanned": 0, "removed": 0, "errors": 1}, + "api_chart_response_jobs": _blank(), + } + job_elapsed = round((time.perf_counter() - job_started) * 1000, 3) + for name, stats in job_dirs.items(): + stamped = _stamp(stats, job_started) + stamped["elapsed_ms"] = job_elapsed + directories[name] = stamped + _run("vedastro_response_cache", _prune_response_cache) + _run("vedastro_official_full_snapshot_cache", _prune_snapshot_semantic_cache) + _run("vedastro_audit_artifacts", _prune_audit_artifacts) + _run("vedastro_snapshot_cache", _prune_vedastro_snapshot_cache) + _run("vedastro_snapshot_negative_cache", _prune_vedastro_negative_cache) + _run("vedastro_gateway_jobs", _prune_gateway_jobs) + _run("varga_segments", _prune_varga_segments) + return { + "scanned": sum(item["scanned"] for item in directories.values()), + "removed": sum(item["removed"] for item in directories.values()), + "errors": sum(item["errors"] for item in directories.values()), + "elapsed_ms": round((time.perf_counter() - started) * 1000, 3), + "directories": directories, + } + + +def start_periodic_prune(interval_s: float | None = None): + """Daemon thread. interval <= 0 starts nothing. Startup prune is separate.""" + if interval_s is None: + raw = os.environ.get("JYOTISH_SCRATCH_PRUNE_INTERVAL_SECONDS", "1800").strip() + try: + interval_s = float(raw) if raw else 1800.0 + except ValueError: + interval_s = 1800.0 + if interval_s <= 0: + return None + + def _loop() -> None: + while True: + time.sleep(interval_s) + try: + prune_all() + except Exception: + continue + + thread = threading.Thread(target=_loop, name=_PRUNE_THREAD_NAME, daemon=True) + thread.start() + return thread diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index 39d3826f..ca07c6cd 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -2590,24 +2590,17 @@ def _vedastro_runtime_fingerprint() -> dict: def _build_api_chart_cache_payload(body: dict) -> dict: + import api_scratch_retention as retention + + today = body.get('today') or body.get('current_date') return { - 'cache_schema_version': 5, - 'birth': { - 'year': body.get('year'), - 'month': body.get('month'), - 'day': body.get('day'), - 'hour': body.get('hour'), - 'minute': body.get('minute'), - 'second': body.get('second', 0), - 'lat': body.get('lat'), - 'lon': body.get('lon'), - 'tz': body.get('tz'), - }, + 'cache_schema_version': 6, + 'birth': retention.normalize_chart_cache_birth(body), 'calculation': { 'ayanamsa': _request_ayanamsa(body), 'node_mode': body.get('node_mode', body.get('nodeMode', 'mean')), - 'today': body.get('today') or body.get('current_date'), - 'transit_date': body.get('transit_date'), + 'today': retention.normalize_chart_cache_day(today), + 'transit_date': retention.normalize_chart_cache_day(body.get('transit_date')), 'skip_vedastro_main_entry_overview': bool(body.get('skip_vedastro_main_entry_overview')), }, 'vedastro_runtime': _vedastro_runtime_fingerprint(), @@ -2678,14 +2671,15 @@ def _store_api_chart_response_cache(cache_payload: dict, payload: dict) -> dict: ) ttl_seconds = _api_chart_cache_ttl_seconds() if ttl_seconds > 0: - record = { - 'cache_key': cache_key, - 'created_at': created_at_unix, - 'payload': payload_with_cache, - } - _api_chart_cache_path(cache_key).write_text( - json.dumps(record, ensure_ascii=False, sort_keys=True), - encoding='utf-8', + import api_scratch_retention as retention + + retention.write_private_json( + _api_chart_cache_path(cache_key), + { + 'cache_key': cache_key, + 'created_at': created_at_unix, + 'payload': payload_with_cache, + }, ) return payload_with_cache @@ -2738,30 +2732,9 @@ def _sqlite_job_connection() -> sqlite3.Connection: def prune_expired_async_jobs() -> dict: """Best-effort startup cleanup for local job records; never reads payloads.""" - removed = 0 - scanned = 0 - if _async_job_backend() == "sqlite": - with _sqlite_job_connection() as connection: - scanned = connection.execute("SELECT COUNT(*) FROM async_jobs").fetchone()[0] - removed = connection.execute( - "DELETE FROM async_jobs WHERE expires_at IS NOT NULL AND expires_at <= ?", (time.time(),) - ).rowcount - return {'scope': 'async_job_cleanup', 'scanned': scanned, 'removed': removed} - for scope in (_HIGH_RIGOR_JOB_SCOPE, _API_CHART_CACHE_SCOPE): - directory = _async_job_dir(scope) - if not directory.is_dir(): - continue - for path in directory.glob('*.json'): - scanned += 1 - try: - record = json.loads(path.read_text(encoding='utf-8')) - expires_at = record.get('expires_at_unix') if isinstance(record, dict) else None - if isinstance(expires_at, (int, float)) and time.time() >= float(expires_at): - path.unlink() - removed += 1 - except (OSError, json.JSONDecodeError): - continue - return {'scope': 'async_job_cleanup', 'scanned': scanned, 'removed': removed} + import api_scratch_retention as retention + + return retention.prune_async_job_records(api_module=sys.modules[__name__]) def _new_async_job_identity(prefix: str) -> dict: @@ -10902,7 +10875,10 @@ def _parse_allowed_hosts(value): def start_server(port=5200, host='127.0.0.1', allowed_origins=None, allowed_hosts=None): - cleanup = prune_expired_async_jobs() + import api_scratch_retention as retention + + cleanup = retention.prune_all(api_module=sys.modules[__name__]) + retention.start_periodic_prune() server = ThreadingHTTPServer((host, port), JyotishAPIHandler) server.daemon_threads = True server.allowed_origins = allowed_origins or DEFAULT_ALLOWED_ORIGINS diff --git a/scripts/research/api_chart_cache_byte_diff.py b/scripts/research/api_chart_cache_byte_diff.py new file mode 100644 index 00000000..e368b9ba --- /dev/null +++ b/scripts/research/api_chart_cache_byte_diff.py @@ -0,0 +1,28 @@ +"""Byte-compare numeric chart-cache spellings. See api_disk_cache_ab.py.""" +from __future__ import annotations + +import json +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import api_disk_cache_ab as ab + + +def main() -> dict: + parent_error = None + try: + parent = ab._load_parent() + except Exception as exc: + parent = None + parent_error = exc.__class__.__name__ + report = ab.run_byte_diff(parent or ab.current) + report["parent_import"] = parent_error or "ok" + report["parent"] = ab.PARENT + return report + + +if __name__ == "__main__": + json.dump(main(), sys.stdout, ensure_ascii=False, indent=2) + sys.stdout.write("\n") diff --git a/scripts/research/api_disk_cache_ab.py b/scripts/research/api_disk_cache_ab.py new file mode 100644 index 00000000..62e2879a --- /dev/null +++ b/scripts/research/api_disk_cache_ab.py @@ -0,0 +1,442 @@ +"""A/B for chart-cache keys, retention, and response bytes. + +Parent functions come from git show of the branch base (origin/staging at +branch creation), not from the task text's older baseline. Scratch files +stay in a temporary directory. +""" +from __future__ import annotations + +import importlib.util +import json +import os +import socket +import statistics +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +PARENT = "66cd60081f37a57680274633fcb3fb942d01bb2f" +os.environ.setdefault("JYOTISH_SKIP_LOCAL_ENV", "1") +os.environ.setdefault("VEDASTRO_ENABLE_NETWORK", "0") + +_SDK_PRESENT = importlib.util.find_spec("vedastro") is not None + + +def _network_reachable() -> bool: + try: + sock = socket.create_connection(("api.vedastro.org", 443), 3) + except OSError: + return False + sock.close() + return True + + +_NETWORK = _network_reachable() + + +class _Block: + def find_spec(self, name, path=None, target=None): + if name.split(".")[0] in {"jhora", "vedastro"}: + raise ModuleNotFoundError(f"No module named '{name}'") + return None + + +if "--with-sdk" not in sys.argv: + sys.meta_path.insert(0, _Block()) +sys.path.insert(0, str(ROOT / "scripts")) +sys.path.insert(0, str(ROOT)) + +import api_scratch_retention as retention # noqa: E402 +import jyotish_api_server as current # noqa: E402 +import vedastro_service_adapter as adapter # noqa: E402 + + +def _load_parent(): + raw = subprocess.check_output(["git", "show", f"{PARENT}:scripts/jyotish_api_server.py"], text=True) + raw = raw.replace( + "SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))", + f"SCRIPTS_DIR = {str(ROOT / 'scripts')!r}", + ) + raw = raw.replace( + "REPO_ROOT = os.path.abspath(os.path.join(SCRIPTS_DIR, '..'))", + f"REPO_ROOT = {str(ROOT)!r}", + ) + directory = Path(tempfile.mkdtemp(prefix="jyotish-cache-a-")) + path = directory / "jyotish_api_server_parent.py" + path.write_text(raw, encoding="utf-8") + spec = importlib.util.spec_from_file_location("jyotish_api_server_parent", path) + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def _handler(module): + cls = module.JyotishAPIHandler + return cls.__new__(cls) + + +def _strip(payload: dict) -> str: + cloned = json.loads(json.dumps(payload, default=str)) + cloned.pop("runtime_cache", None) + return json.dumps(cloned, ensure_ascii=False, sort_keys=True, default=str) + + +def _diff_paths(left: str, right: str, limit: int = 12) -> list[str]: + a = json.loads(left) + b = json.loads(right) + + found: list[str] = [] + + def walk(one, two, prefix: str) -> None: + if len(found) >= limit: + return + if type(one) is not type(two): + found.append(prefix or "$") + return + if isinstance(one, dict): + for key in sorted(set(one) | set(two)): + if key not in one or key not in two: + found.append(f"{prefix}.{key}") + continue + walk(one[key], two[key], f"{prefix}.{key}") + elif isinstance(one, list): + if len(one) != len(two): + found.append(f"{prefix}[]") + return + for index, (item, other) in enumerate(zip(one, two)): + walk(item, other, f"{prefix}[{index}]") + elif one != two: + found.append(prefix or "$") + + walk(a, b, "") + return found + + +def _body(hour, lat, **extra) -> dict: + payload = { + "year": 1990, + "month": 1, + "day": 1, + "minute": 0, + "lon": 20.0, + "tz": 5.5, + "hour": hour, + "lat": lat, + "today": "2026-10-14", + } + payload.update(extra) + return payload + + +def run_byte_diff(parent=None) -> dict: + os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "0" + os.environ["VEDASTRO_WRITE_AUDIT_ARTIFACTS"] = "0" + parent = parent or current + groups = { + "hour": [_body(12, 10.0), _body(12.0, 10.0), _body("12", 10.0)], + "lat": [_body(12, 10.0), _body(12, "10.0")], + "tz": [_body(12, 10.0), _body(12, 10.0, tz="5.5")], + "second": [_body(12, 10.0), _body(12, 10.0, second=0)], + } + report = {"groups": {}, "stable_repeat": None} + handler = _handler(current) + started = time.perf_counter() + try: + sample = handler._compute_chart_sync(_body(12, 10.0, skip_vedastro_main_entry_overview=True)) + except Exception as exc: + return {"ok": False, "error": exc.__class__.__name__, "elapsed_s": round(time.perf_counter() - started, 3)} + sample_s = round(time.perf_counter() - started, 3) + repeat = handler._compute_chart_sync(_body(12, 10.0, skip_vedastro_main_entry_overview=True)) + report["stable_repeat"] = _strip(sample) == _strip(repeat) + report["sample_seconds"] = sample_s + if not report["stable_repeat"]: + report["volatile_paths"] = _diff_paths(_strip(sample), _strip(repeat)) + for name, bodies in groups.items(): + for skip in (True, False): + rendered = [] + for body in bodies: + item = dict(body) + if skip: + item["skip_vedastro_main_entry_overview"] = True + rendered.append(_strip(handler._compute_chart_sync(item))) + same = len(set(rendered)) == 1 + entry = {"skip": skip, "same": same} + if not same: + entry["paths"] = _diff_paths(rendered[0], rendered[1]) + clocks = ("cache_created_at", "cache_expires_at", "called_at") + entry["only_clock_fields"] = bool(entry["paths"]) and all( + path.endswith(clocks) for path in entry["paths"] + ) + else: + entry["only_clock_fields"] = False + report["groups"][f"{name}:skip={skip}"] = entry + if parent is not current: + a_handler = _handler(parent) + b_handler = handler + body = _body(12, 10.0, skip_vedastro_main_entry_overview=True) + a_text = _strip(a_handler._compute_chart_sync(body)) + b_text = _strip(b_handler._compute_chart_sync(body)) + report["a_vs_b_same"] = a_text == b_text + if a_text != b_text: + report["a_vs_b_paths"] = _diff_paths(a_text, b_text) + report["ok"] = all(item["same"] or item.get("only_clock_fields") for item in report["groups"].values()) + return report + + +def _requests(): + charts = [] + for index in range(20): + charts.append( + { + "year": 1990, + "month": 1, + "day": 1 + (index % 27), + "minute": index % 60, + "lon": float(index), + "tz": 5.5, + } + ) + spellings = ( + {"hour": 12, "lat": 10.0}, + {"hour": 12.0, "lat": "10.0"}, + {"hour": "12", "lat": 10.0}, + ) + days = [f"2026-10-{day:02d}" for day in range(8, 15)] + rows = [] + for chart in charts: + for spelling in spellings: + for skip in (False, True): + for day in days: + body = dict(chart) + body.update(spelling) + body["today"] = day + if skip: + body["skip_vedastro_main_entry_overview"] = True + rows.append(body) + return rows + + +def _key(module, body) -> str: + return module._api_chart_cache_key(module._build_api_chart_cache_payload(body)) + + +def _hit_rate(module, rows) -> dict: + seen = set() + hits = 0 + for body in rows: + for _repeat in range(3): + key = _key(module, body) + if key in seen: + hits += 1 + else: + seen.add(key) + total = len(rows) * 3 + return {"requests": total, "unique_keys": len(seen), "hits": hits, "hit_rate": round(hits / total, 4)} + + +def _seed_and_prune(module, rows) -> dict: + now = time.time() + os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "900" + scratch = Path(tempfile.mkdtemp(prefix="jyotish-cache-b-")) + chart = scratch / "api_chart_cache" + jobs = scratch / "jobs" + adapter_dir = scratch / "adapter" + chart.mkdir() + jobs.mkdir() + current._api_chart_cache_dir = lambda: chart + current._async_job_dir = lambda scope: jobs + current._async_job_backend = lambda: "file" + adapter.ARTIFACT_DIR = adapter_dir + os.environ["JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR"] = str(scratch / "snap") + os.environ["JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR"] = str(scratch / "neg") + os.environ["VEDASTRO_GATEWAY_QUEUE_DIR"] = str(scratch / "gw") + os.environ["JYOTISH_CACHE_DIR"] = str(scratch / "cache") + days = sorted({body["today"] for body in rows}) + newest = days[-1] + written = {} + for body in rows: + key = _key(module, body) + age_days = days.index(body["today"]) + age_days = len(days) - 1 - age_days + written[key] = now - age_days * 86400 + for key, created in written.items(): + path = chart / f"{key}.json" + path.write_text(json.dumps({"cache_key": key, "created_at": created, "payload": {}}), encoding="utf-8") + before = len(list(chart.glob("*.json"))) + before_bytes = sum(path.stat().st_size for path in chart.glob("*.json")) + today_keys = len({_key(module, body) for body in rows if body["today"] == newest}) + if module is current: + prune = retention.prune_all(now=now) + after = len(list(chart.glob("*.json"))) + elapsed = prune["directories"]["api_chart_cache"]["elapsed_ms"] + else: + after = before + elapsed = None + return { + "files_before": before, + "bytes_before": before_bytes, + "files_after": after, + "today_active_keys": today_keys, + "chart_prune_ms": elapsed, + } + + +def _artifact_counts() -> dict: + scratch = Path(tempfile.mkdtemp(prefix="jyotish-artifact-ab-")) + adapter.ARTIFACT_DIR = scratch + os.environ.pop("VEDASTRO_WRITE_AUDIT_ARTIFACTS", None) + result = { + "operation": "official_full_snapshot", + "source_metadata": { + "operation": "official_full_snapshot", + "request_hash": "a" * 64, + "response_hash": "b" * 64, + }, + } + adapter._write_artifact(result) + return {"b_artifact_files": len(list(scratch.glob("*.json")))} + + +def _negative_cache_after_prune() -> dict: + scratch = Path(tempfile.mkdtemp(prefix="jyotish-neg-ab-")) + os.environ["JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR"] = str(scratch) + os.environ["JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR"] = str(scratch / "unused-snap") + os.environ["VEDASTRO_GATEWAY_QUEUE_DIR"] = str(scratch / "unused-gw") + os.environ["JYOTISH_CACHE_DIR"] = str(scratch / "unused-cache") + current._api_chart_cache_dir = lambda: scratch / "unused-chart" + current._async_job_dir = lambda scope: scratch / "unused-jobs" + (scratch / "unused-chart").mkdir() + (scratch / "unused-jobs").mkdir() + today = time.strftime("%Y-%m-%d", time.gmtime()) + yesterday = time.strftime("%Y-%m-%d", time.gmtime(time.time() - 86400)) + for label, day in (("a" * 64, today), ("b" * 64, yesterday)): + path = scratch / f"{label}.json" + path.write_text(json.dumps({"reference_date": day}), encoding="utf-8") + retention.prune_all(now=time.time()) + return {"files_after": len(list(scratch.glob("*.json")))} + + +def _prune_10k() -> dict: + scratch = Path(tempfile.mkdtemp(prefix="jyotish-prune-10k-")) + chart = scratch / "chart" + chart.mkdir() + current._api_chart_cache_dir = lambda: chart + current._async_job_dir = lambda scope: scratch / "jobs" + (scratch / "jobs").mkdir() + current._async_job_backend = lambda: "file" + adapter.ARTIFACT_DIR = scratch / "adapter" + os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "900" + os.environ["JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR"] = str(scratch / "snap") + os.environ["JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR"] = str(scratch / "neg") + os.environ["VEDASTRO_GATEWAY_QUEUE_DIR"] = str(scratch / "gw") + os.environ["JYOTISH_CACHE_DIR"] = str(scratch / "cache") + for index in range(10000): + path = chart / f"{index:064x}.json" + path.write_text('{"created_at":1,"payload":{}}', encoding="utf-8") + started = time.perf_counter() + result = retention.prune_all(now=time.time()) + elapsed = time.perf_counter() - started + return { + "prune_all_seconds": round(elapsed, 3), + "chart_removed": result["directories"]["api_chart_cache"]["removed"], + "chart_elapsed_ms": result["directories"]["api_chart_cache"]["elapsed_ms"], + "files_left": len(list(chart.glob("*.json"))), + } + + +def _one_chart_timing() -> dict: + os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "0" + handler = _handler(current) + body = _body(12, 10.0, skip_vedastro_main_entry_overview=True) + misses = [] + for _index in range(3): + started = time.perf_counter() + handler._compute_chart_sync(body) + misses.append((time.perf_counter() - started) * 1000) + os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "900" + scratch = Path(tempfile.mkdtemp(prefix="jyotish-hit-time-")) + current._api_chart_cache_dir = lambda: scratch + cold = handler._compute_chart_sync(body) + hits = [] + for _index in range(3): + started = time.perf_counter() + hit = handler._compute_chart_sync(body) + hits.append((time.perf_counter() - started) * 1000) + assert hit.get("runtime_cache", {}).get("cache_hit") is True + return { + "miss_median_ms": round(statistics.median(misses), 3), + "hit_median_ms": round(statistics.median(hits), 3), + "cold_has_runtime_cache": isinstance(cold, dict) and "runtime_cache" in cold, + } + + +def main() -> dict: + parent_error = None + try: + parent = _load_parent() + except Exception as exc: + parent = None + parent_error = exc.__class__.__name__ + rows = _requests() + side_a = _hit_rate(parent, rows) if parent is not None else None + side_b = _hit_rate(current, rows) + report = { + "parent": PARENT, + "parent_import": parent_error or "ok", + "vedastro_sdk": _SDK_PRESENT, + "vedastro_network": _NETWORK, + "hit_rate_a": side_a, + "hit_rate_b": side_b, + "artifacts": _artifact_counts(), + "negative_cache_files_after_prune": _negative_cache_after_prune(), + "prune_10k": _prune_10k(), + } + if side_a and side_a["hit_rate"]: + report["hit_rate_ratio_b_over_a"] = round(side_b["hit_rate"] / side_a["hit_rate"], 3) + try: + report["files_b"] = _seed_and_prune(current, rows) + if parent is not None: + report["files_a"] = _seed_and_prune(parent, rows) + except Exception as exc: + report["files_error"] = exc.__class__.__name__ + try: + report["timing"] = _one_chart_timing() + except Exception as exc: + report["timing_error"] = exc.__class__.__name__ + try: + report["byte_diff"] = run_byte_diff(parent or current) + except Exception as exc: + report["byte_diff"] = {"ok": False, "error": exc.__class__.__name__} + return report + + +def _sdk_probe() -> dict: + os.environ["JYOTISH_API_CHART_CACHE_TTL_SECONDS"] = "0" + os.environ["VEDASTRO_ENABLE_NETWORK"] = "0" + os.environ["VEDASTRO_WRITE_AUDIT_ARTIFACTS"] = "0" + handler = _handler(current) + body = _body(12, 10.0) + started = time.perf_counter() + try: + handler._compute_chart_sync(body) + error = None + except Exception as exc: + error = type(exc).__name__ + return { + "blocked_vedastro": False, + "vedastro_sdk": _SDK_PRESENT, + "vedastro_network": _NETWORK, + "network_enabled": False, + "elapsed_s": round(time.perf_counter() - started, 3), + "error": error, + } + + +if __name__ == "__main__": + payload = _sdk_probe() if "--sdk-probe" in sys.argv else main() + json.dump(payload, sys.stdout, ensure_ascii=False, indent=2) + sys.stdout.write("\n") diff --git a/scripts/vedastro_gateway.py b/scripts/vedastro_gateway.py index f81e9955..f35ad7cc 100644 --- a/scripts/vedastro_gateway.py +++ b/scripts/vedastro_gateway.py @@ -340,6 +340,30 @@ def probe_official_rest_health(*, force_refresh: bool = True) -> dict[str, Any]: return result +def _gateway_snapshot_ttl_helpers(): + try: + from vedastro_service_adapter import ( + _official_full_snapshot_cache_ttl_seconds, + _official_full_snapshot_negative_ttl_seconds, + ) + except ModuleNotFoundError: + from scripts.vedastro_service_adapter import ( + _official_full_snapshot_cache_ttl_seconds, + _official_full_snapshot_negative_ttl_seconds, + ) + return _official_full_snapshot_cache_ttl_seconds, _official_full_snapshot_negative_ttl_seconds + + +def _gateway_official_snapshot_ttl_seconds() -> float: + positive, _negative = _gateway_snapshot_ttl_helpers() + return positive() + + +def _gateway_official_snapshot_negative_ttl_seconds() -> float: + _positive, negative = _gateway_snapshot_ttl_helpers() + return negative() + + def gateway_status(*, force_refresh: bool = True) -> dict[str, Any]: from scripts.diagnose_vedastro_mode import build_report as build_vedastro_mode_report @@ -368,7 +392,8 @@ def gateway_status(*, force_refresh: bool = True) -> dict[str, Any]: ), "sdk_version": _sdk_version(), "cache_ttl_seconds": config["cache_ttl_seconds"], - "official_full_snapshot_cache_ttl_seconds": _int_env("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", 0), + "official_full_snapshot_cache_ttl_seconds": _gateway_official_snapshot_ttl_seconds(), + "official_full_snapshot_negative_cache_ttl_seconds": _gateway_official_snapshot_negative_ttl_seconds(), "queue_enabled": config["queue_enabled"], "fail_open_local": config["fail_open_local"], "official_readiness": { diff --git a/scripts/vedastro_service_adapter.py b/scripts/vedastro_service_adapter.py index 40e47655..74fc34cf 100644 --- a/scripts/vedastro_service_adapter.py +++ b/scripts/vedastro_service_adapter.py @@ -425,14 +425,23 @@ def _repo_relative(path: Path) -> str: return str(path) -def _write_artifact(result: dict[str, Any]) -> str: +def _audit_artifacts_enabled() -> bool: + raw = os.environ.get("VEDASTRO_WRITE_AUDIT_ARTIFACTS", "0").strip().lower() + return raw in {"1", "true", "yes", "on"} + + +def _write_artifact(result: dict[str, Any]) -> str | None: + if not _audit_artifacts_enabled(): + return None metadata = result.get("source_metadata") or {} artifact = _artifact_path( str(metadata.get("operation") or result.get("operation") or "calculation"), str(metadata.get("request_hash") or "no-request-hash"), str(metadata.get("response_hash") or "no-response-hash"), ) - artifact.write_text(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8") + from api_scratch_retention import write_private_json + + write_private_json(artifact, result) return _repo_relative(artifact) @@ -486,6 +495,22 @@ def _official_full_snapshot_cache_ttl_seconds() -> float: return 0.0 +def _official_full_snapshot_negative_ttl_seconds() -> float: + normal = _official_full_snapshot_cache_ttl_seconds() + if normal <= 0: + return 0.0 + raw = os.environ.get("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_NEGATIVE_TTL_SECONDS", "300").strip() + try: + negative = max(0.0, float(raw)) if raw else 300.0 + except ValueError: + negative = 300.0 + return min(negative, normal) + + +def _official_full_snapshot_result_is_success(payload: dict[str, Any]) -> bool: + return payload.get("available") is True and payload.get("status") == "ok" + + def _free_tier_max_requests() -> int: raw = os.environ.get(FREE_TIER_MAX_REQUESTS_ENV, "").strip() if not raw: @@ -1719,9 +1744,11 @@ def _attach_official_full_snapshot_semantic_cache_metadata( cache_key: str, created_at_unix: float, cache_hit: bool, + ttl_seconds: float | None = None, ) -> dict[str, Any]: payload_copy = json.loads(json.dumps(payload)) - ttl_seconds = _official_full_snapshot_cache_ttl_seconds() + if ttl_seconds is None: + ttl_seconds = _official_full_snapshot_cache_ttl_seconds() created_at = datetime.utcfromtimestamp(created_at_unix).strftime("%Y-%m-%dT%H:%M:%SZ") expires_at = datetime.utcfromtimestamp(created_at_unix + ttl_seconds).strftime("%Y-%m-%dT%H:%M:%SZ") metadata = dict(payload_copy.get("source_metadata") or {}) @@ -1745,9 +1772,15 @@ def _rebind_official_full_snapshot_case_identity(payload: dict[str, Any], case_i return payload_copy +def _semantic_cache_ttl_seconds(record: dict[str, Any]) -> float: + # Records written before the negative flag are successes and keep the long TTL. + if record.get("negative") is True: + return _official_full_snapshot_negative_ttl_seconds() + return _official_full_snapshot_cache_ttl_seconds() + + def _load_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: str) -> dict[str, Any] | None: - ttl_seconds = _official_full_snapshot_cache_ttl_seconds() - if ttl_seconds <= 0: + if _official_full_snapshot_cache_ttl_seconds() <= 0: return None cache_key = _official_full_snapshot_cache_key(case, case_id) cache_path = _official_full_snapshot_cache_path(cache_key) @@ -1763,7 +1796,8 @@ def _load_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: s payload = record.get("payload") if not isinstance(created_at_raw, (int, float)) or not isinstance(payload, dict): return None - if time.time() - float(created_at_raw) > ttl_seconds: + ttl_seconds = _semantic_cache_ttl_seconds(record) + if ttl_seconds <= 0 or time.time() - float(created_at_raw) > ttl_seconds: return None rebound = _rebind_official_full_snapshot_case_identity(payload, case_id) return _attach_official_full_snapshot_semantic_cache_metadata( @@ -1771,28 +1805,37 @@ def _load_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: s cache_key=cache_key, created_at_unix=float(created_at_raw), cache_hit=True, + ttl_seconds=ttl_seconds, ) def _store_official_full_snapshot_semantic_cache(case: dict[str, Any], case_id: str, payload: dict[str, Any]) -> dict[str, Any]: cache_key = _official_full_snapshot_cache_key(case, case_id) created_at_unix = time.time() + negative = not _official_full_snapshot_result_is_success(payload) + ttl_seconds = ( + _official_full_snapshot_negative_ttl_seconds() + if negative + else _official_full_snapshot_cache_ttl_seconds() + ) payload_copy = _attach_official_full_snapshot_semantic_cache_metadata( payload, cache_key=cache_key, created_at_unix=created_at_unix, cache_hit=False, + ttl_seconds=ttl_seconds, ) - ttl_seconds = _official_full_snapshot_cache_ttl_seconds() if ttl_seconds > 0: - record = { - "cache_key": cache_key, - "created_at": created_at_unix, - "payload": payload_copy, - } - _official_full_snapshot_cache_path(cache_key).write_text( - json.dumps(record, ensure_ascii=False, sort_keys=True), - encoding="utf-8", + from api_scratch_retention import write_private_json + + write_private_json( + _official_full_snapshot_cache_path(cache_key), + { + "cache_key": cache_key, + "created_at": created_at_unix, + "negative": negative, + "payload": payload_copy, + }, ) return payload_copy @@ -1882,7 +1925,9 @@ def _store_cached_payload(cache_key: str, payload: dict[str, Any]) -> dict[str, "created_at": created_at_unix, "payload": payload_copy, } - cache_path.write_text(json.dumps(record, ensure_ascii=False, sort_keys=True), encoding="utf-8") + from api_scratch_retention import write_private_json + + write_private_json(cache_path, record) return payload_copy diff --git a/tests/test_api_scratch_retention.py b/tests/test_api_scratch_retention.py new file mode 100644 index 00000000..65dab310 --- /dev/null +++ b/tests/test_api_scratch_retention.py @@ -0,0 +1,314 @@ +"""Retention, chart-cache keys, and VedAstro audit switch.""" +from __future__ import annotations + +import hashlib +import json +import os +import re +import threading +import time +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +import api_scratch_retention as retention +import jyotish_api_server as api +import vedastro_service_adapter as adapter + +ROOT = Path(__file__).resolve().parents[1] + + +def _hex(label: str) -> str: + return hashlib.sha256(label.encode("utf-8")).hexdigest() + ".json" + + +def _isolate(monkeypatch, tmp_path: Path) -> None: + chart = tmp_path / "chart" + jobs = tmp_path / "jobs" + chart.mkdir() + jobs.mkdir() + monkeypatch.setattr(api, "_api_chart_cache_dir", lambda: chart) + monkeypatch.setattr(api, "_async_job_dir", lambda scope: jobs) + monkeypatch.setattr(api, "_async_job_backend", lambda: "file") + monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path / "adapter") + monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR", str(tmp_path / "snap")) + monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR", str(tmp_path / "neg")) + monkeypatch.setenv("VEDASTRO_GATEWAY_QUEUE_DIR", str(tmp_path / "gw")) + monkeypatch.setenv("JYOTISH_CACHE_DIR", str(tmp_path / "cache")) + monkeypatch.setenv("JYOTISH_API_CHART_CACHE_TTL_SECONDS", "900") + monkeypatch.delenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", raising=False) + + +def _write(path: Path, payload) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload), encoding="utf-8") + + +def test_prune_all_drops_expired_and_old_bad_chart_cache_only(monkeypatch, tmp_path): + _isolate(monkeypatch, tmp_path) + chart = api._api_chart_cache_dir() + now = time.time() + fresh = chart / _hex("fresh") + expired = chart / _hex("expired") + bad_old = chart / _hex("bad-old") + bad_young = chart / _hex("bad-young") + stranger = chart / "notes.json" + _write(fresh, {"cache_key": "a", "created_at": now, "payload": {}}) + _write(expired, {"cache_key": "b", "created_at": now - 5000, "payload": {}}) + bad_old.write_text("{", encoding="utf-8") + bad_young.write_text("{", encoding="utf-8") + stranger.write_text("{}", encoding="utf-8") + old = now - (25 * 3600) + os.utime(bad_old, (old, old)) + os.utime(stranger, (old, old)) + + result = retention.prune_all(now=now) + + assert fresh.exists() + assert bad_young.exists() + assert stranger.exists() + assert not expired.exists() + assert not bad_old.exists() + assert result["directories"]["api_chart_cache"]["removed"] == 2 + assert result["errors"] == 0 + + +def test_job_cleanup_still_removes_only_expired_records(monkeypatch, tmp_path): + _isolate(monkeypatch, tmp_path) + jobs = api._async_job_dir("any") + expired = jobs / "expired.json" + active = jobs / "active.json" + bad = jobs / "bad.json" + expired.write_text(json.dumps({"expires_at_unix": time.time() - 1}), encoding="utf-8") + active.write_text(json.dumps({"expires_at_unix": time.time() + 60}), encoding="utf-8") + bad.write_text("{", encoding="utf-8") + + result = api.prune_expired_async_jobs() + + assert result["removed"] == 1 + assert not expired.exists() + assert active.exists() + assert bad.exists() + + +def test_interval_zero_does_not_start_a_thread(): + before = [item.name for item in threading.enumerate()] + assert retention.start_periodic_prune(0) is None + assert [item.name for item in threading.enumerate()] == before + + +def test_prune_counts_unlink_errors_without_raising(monkeypatch, tmp_path): + _isolate(monkeypatch, tmp_path) + chart = api._api_chart_cache_dir() + target = chart / _hex("stuck") + _write(target, {"created_at": time.time() - 5000, "payload": {}}) + real_unlink = Path.unlink + + def _boom(self, *args, **kwargs): + if self == target: + raise OSError("denied") + return real_unlink(self, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", _boom) + result = retention.prune_all(now=time.time()) + assert result["errors"] > 0 + assert target.exists() + + +def _chart_key(body): + return api._api_chart_cache_key(api._build_api_chart_cache_payload(body)) + + +def test_numeric_birth_spellings_share_one_chart_cache_key(): + base = { + "year": 1990, + "month": 1, + "day": 1, + "minute": 0, + "lat": 28.6139, + "lon": 77.2090, + "tz": 5.5, + "today": "2026-10-14", + } + keys = {_chart_key(dict(base, hour=hour)) for hour in (12, 12.0, "12")} + assert len(keys) == 1 + assert _chart_key(dict(base, hour=12, second=0)) in keys + assert _chart_key(dict(base, hour=12, lat="28.6139")) in keys + assert _chart_key(dict(base, hour=12, tz="5.5")) in keys + assert _chart_key(dict(base, hour=12, tz="Asia/Kolkata")) not in keys + skipped = _chart_key(dict(base, hour=12, skip_vedastro_main_entry_overview=True)) + assert skipped not in keys + assert api._build_api_chart_cache_payload(base)["cache_schema_version"] == 6 + + +def test_audit_artifact_default_off_and_compact_when_on(monkeypatch, tmp_path): + monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path) + monkeypatch.delenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", raising=False) + manifest = {"requests": [{"section": "chart_core"}], "reference_date": "2026-10-14"} + quiet = adapter._normalize_official_full_snapshot_success( + "https://example.test/api", + manifest, + {"chart_core": {"Status": "Pass"}}, + {"chart_core": "ok"}, + 1, + [], + ) + assert quiet["source_metadata"]["artifact_path"] is None + assert list(tmp_path.rglob("*.json")) == [] + + monkeypatch.setenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", "1") + written = adapter._normalize_official_full_snapshot_success( + "https://example.test/api", + manifest, + {"chart_core": {"Status": "Pass"}}, + {"chart_core": "ok"}, + 1, + [], + ) + rel = written["source_metadata"]["artifact_path"] + path = Path(rel) + if not path.is_absolute(): + path = adapter.ROOT / rel + assert path.exists() + # Linux reports the mode bit. Windows keeps the file writable and does not surface 0600. + if os.name != "nt": + assert path.stat().st_mode & 0o777 == 0o600 + text = path.read_text(encoding="utf-8") + assert "\n" not in text + assert json.loads(text)["status"] == "ok" + + +def test_failed_snapshot_expires_before_success(monkeypatch, tmp_path): + monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path) + monkeypatch.setenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", "604800") + monkeypatch.delenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_NEGATIVE_TTL_SECONDS", raising=False) + calls = {"fail": 0, "ok": 0} + + def fake_run(case, case_id="user_chart"): + if case.get("day") == 2: + calls["fail"] += 1 + return {"available": False, "status": "timeout", "source_metadata": {}} + calls["ok"] += 1 + return {"available": True, "status": "ok", "source_metadata": {}} + + monkeypatch.setattr(adapter, "_run_official_full_snapshot_case", fake_run) + fail_case = {"year": 1990, "month": 1, "day": 2, "hour": 12, "lat": 1.0, "lon": 2.0, "tz": 5.5} + ok_case = {"year": 1990, "month": 1, "day": 3, "hour": 12, "lat": 1.0, "lon": 2.0, "tz": 5.5} + adapter.run_official_full_snapshot_for_case(fail_case, case_id="fail") + adapter.run_official_full_snapshot_for_case(ok_case, case_id="ok") + cache_dir = tmp_path / "official_full_snapshot_cache" + aged = time.time() - 301 + for path in cache_dir.glob("*.json"): + record = json.loads(path.read_text(encoding="utf-8")) + record["created_at"] = aged + path.write_text(json.dumps(record), encoding="utf-8") + if record.get("negative") is True: + assert record["negative"] is True + adapter.run_official_full_snapshot_for_case(fail_case, case_id="fail") + adapter.run_official_full_snapshot_for_case(ok_case, case_id="ok") + assert calls["fail"] == 2 + assert calls["ok"] == 1 + + +def test_gateway_status_reports_cache_ttl_fallback(monkeypatch): + from scripts import vedastro_gateway + + monkeypatch.setenv("JYOTISH_SKIP_LOCAL_ENV", "1") + monkeypatch.delenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", raising=False) + monkeypatch.setenv("VEDASTRO_CACHE_TTL_SECONDS", "604800") + status = vedastro_gateway.gateway_status() + assert status["official_full_snapshot_cache_ttl_seconds"] == 604800 + assert status["official_full_snapshot_negative_cache_ttl_seconds"] == 300 + + +def _day(offset: int) -> str: + return (datetime.now(timezone.utc).date() + timedelta(days=offset)).isoformat() + + +def test_snapshot_negative_gateway_and_varga_prune_rules(monkeypatch, tmp_path): + _isolate(monkeypatch, tmp_path) + now = time.time() + today = datetime.fromtimestamp(now, timezone.utc).date() + snap = tmp_path / "snap" + neg = tmp_path / "neg" + gw = tmp_path / "gw" + varga = tmp_path / "cache" / "varga_segments" + keep_ref = (today - timedelta(days=7)).isoformat() + drop_ref = (today - timedelta(days=9)).isoformat() + _write(snap / _hex("snap-keep"), {"schema": "vedastro_snapshot_cache.v1", "reference_date": keep_ref}) + _write(snap / _hex("snap-drop"), {"schema": "vedastro_snapshot_cache.v1", "reference_date": drop_ref}) + _write(snap / "notes.json", {"reference_date": "1990-01-01"}) + _write(neg / _hex("neg-today"), {"schema": "vedastro_snapshot_negative_cache.v1", "reference_date": today.isoformat()}) + _write(neg / _hex("neg-yesterday"), {"schema": "vedastro_snapshot_negative_cache.v1", "reference_date": (today - timedelta(days=1)).isoformat()}) + _write(neg / "stray.json", {"reference_date": "1990-01-01"}) + job_id = "vgw_" + "ab" * 10 + old_stamp = datetime.fromtimestamp(now - 90000, timezone.utc).isoformat() + fresh_stamp = datetime.fromtimestamp(now, timezone.utc).isoformat() + _write(gw / f"{job_id}.json", {"job_id": job_id, "status": "completed", "updated_at": old_stamp}) + (gw / f"{job_id}.official_raw_response.json").write_text("{}", encoding="utf-8") + running = "vgw_" + "cd" * 10 + _write(gw / f"{running}.json", {"job_id": running, "status": "running", "updated_at": old_stamp}) + failed = "vgw_" + "ef" * 10 + _write(gw / f"{failed}.json", {"job_id": failed, "status": "failed", "updated_at": fresh_stamp}) + (gw / "notes.json").write_text("{}", encoding="utf-8") + old_varga = varga / _hex("old-segment") + young_varga = varga / _hex("young-segment") + _write(old_varga, {"window_key": "x"}) + _write(young_varga, {"window_key": "y"}) + _write(varga / "notes.json", {"window_key": "z"}) + os.utime(old_varga, (now - 8 * 86400, now - 8 * 86400)) + + monkeypatch.setenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", "0") + artifact = tmp_path / "adapter" / "official_full_snapshot-0123456789ab-0123456789ab.json" + artifact.parent.mkdir(parents=True, exist_ok=True) + artifact.write_text("{}", encoding="utf-8") + os.utime(artifact, (now - 25 * 3600, now - 25 * 3600)) + kept_artifact = tmp_path / "adapter" / "method_catalog_snapshot.json" + kept_artifact.write_text("{}", encoding="utf-8") + os.utime(kept_artifact, (now - 25 * 3600, now - 25 * 3600)) + + retention.prune_all(now=now) + + assert (snap / _hex("snap-keep")).exists() + assert not (snap / _hex("snap-drop")).exists() + assert (snap / "notes.json").exists() + assert (neg / _hex("neg-today")).exists() + assert not (neg / _hex("neg-yesterday")).exists() + assert (neg / "stray.json").exists() + assert not (gw / f"{job_id}.json").exists() + assert not (gw / f"{job_id}.official_raw_response.json").exists() + assert (gw / f"{running}.json").exists() + assert (gw / f"{failed}.json").exists() + assert (gw / "notes.json").exists() + assert not old_varga.exists() + assert young_varga.exists() + assert (varga / "notes.json").exists() + assert not artifact.exists() + assert kept_artifact.exists() + + +def _scratch_local_names() -> set[str]: + names: set[str] = set() + slash = re.compile(r"scratch/local/([A-Za-z0-9_.]+)") + joined = re.compile( + r"""['\"]scratch['\"]\s*/\s*['\"]local['\"]\s*/\s*(?:f)?['\"]([A-Za-z0-9_.]+)['\"]""" + ) + after_local = re.compile( + r"""['\"]local['\"]\s*\)*\s*/\s*['\"]([A-Za-z0-9_.]+)['\"]""" + ) + for path in (ROOT / "scripts").rglob("*.py"): + text = path.read_text(encoding="utf-8") + names.update(slash.findall(text)) + names.update(joined.findall(text)) + names.update(after_local.findall(text)) + return names + + +def test_new_scratch_local_dirs_are_registered_or_whitelisted(): + allowed = set(retention.REGISTERED_DIR_NAMES) | set(retention.OFFLINE_SCRATCH_NAMES) + found = _scratch_local_names() + unexpected = sorted(found - allowed) + assert unexpected == [], unexpected + assert "api_chart_cache" in found + assert "consult_full_data_cache" in retention.OFFLINE_SCRATCH_NAMES diff --git a/tests/test_vedastro_service_adapter_executor.py b/tests/test_vedastro_service_adapter_executor.py index e4f87f11..fd4e869c 100644 --- a/tests/test_vedastro_service_adapter_executor.py +++ b/tests/test_vedastro_service_adapter_executor.py @@ -769,6 +769,10 @@ def test_vedastro_range_scan_records_hashes_and_artifact_path() -> None: env = os.environ.copy() env["VEDASTRO_API_ENDPOINT"] = f"http://127.0.0.1:{server.server_port}/vedastro" env["VEDASTRO_ENABLE_NETWORK"] = "1" + # 原值：沿用进程环境，没有 VEDASTRO_WRITE_AUDIT_ARTIFACTS。 + # 新值：显式设为 1。 + # 原因：审计落盘默认关闭后，这条测试仍要断言文件存在，所以打开开关。这是测试前置，不是放宽断言。 + env["VEDASTRO_WRITE_AUDIT_ARTIFACTS"] = "1" completed = subprocess.run( [ sys.executable,