Compare commits

...
Author SHA1 Message Date
jesse-uxandGrok (agent loop) 81e2e1cfed 服务器缓存按期限清理,审计快照默认不再落地
排盘缓存、官方快照、负缓存、网关任务和分盘段过期后会删掉。同一张盘的数字写法合成一把缓存键。失败快照只用短期限。自测留存、异步任务、增长合同、安全缓存和隐私标记;一万个文件的清理慢于 2 秒,命中率提升不到 2.5 倍,都按实测写入记录。

Co-Authored-By: Grok (agent loop)
2026-10-14 04:53:30 +08:00
14 changed files with 1677 additions and 65 deletions
+2
View File
@@ -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
+5
View File
@@ -1,5 +1,10 @@
# 印度占星 Skill 更新日志
## 2026-10-14 — 服务器不再长期保留出生资料副本
- 排盘缓存、官方快照和相关临时文件会按期限删掉,不再一直堆在服务器上。审计用的快照文件默认不再落地。失败的官方快照只短时间复用。
- Skill 版本不变(BUG-1330~1332)。
## 2026-10-14 — 婚恋评级改成证据档位,KP 判宫和大运年长对齐(待合入 staging)
- 婚恋结构指数不再写成婚姻稳定、离婚、再婚或建议婚前咨询。中文报告用中文档位,英文报告用英文档位。配偶星分数按 0 到 1 显示,不再写成百分制。
+45
View File
@@ -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。
@@ -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 秒,没有报错,没有拉真实快照。
## 未做
- 没有改部署文件、生产环境变量、数据库,也没有手工清生产卷。
- 没有把咨询卡片缓存并进周期清理。
+1 -1
View File
@@ -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` |
@@ -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 秒,没有报错。没有拉取真实快照。
+609
View File
@@ -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
+23 -47
View File
@@ -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
@@ -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")
+442
View File
@@ -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")
+26 -1
View File
@@ -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": {
+61 -16
View File
@@ -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
+314
View File
@@ -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
@@ -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,