合并 staging,并把长报告缓存目录登记进自行清理名单
Independent Staging Quality Gate / validate (push) Successful in 14m32s
Independent Staging Quality Gate / publish (push) Successful in 4m18s

与最新 origin/staging 合并。report_reference_cache 写入 OFFLINE_SCRATCH_NAMES。清理仍由每次读取前删掉两天前的文件完成,两个缓存模块的行为没改。

自测:test_api_scratch_retention 10、test_report_reference_perf 15、test_consult_card_full_source 12,共 37 通过。

Co-Authored-By: Grok (agent loop)
This commit is contained in:
jesse-ux
2026-10-14 10:40:43 +08:00
co-authored by Grok (agent loop)
78 changed files with 67879 additions and 18059 deletions
+46 -7
View File
@@ -127,6 +127,7 @@ def _active_narayana(
birth_at: datetime,
event_at: datetime,
periods: list[dict[str, Any]] | None = None,
antardasha_by_md: dict[int, list[dict[str, Any]]] | None = None,
) -> tuple[int | None, int | None]:
if periods is None:
periods = narayana_dasha.calc_narayana_mahadasha(
@@ -134,10 +135,30 @@ def _active_narayana(
planet_longitudes,
)
age = max((event_at - birth_at).total_seconds() / (365.2425 * 86_400), 0.0)
active = narayana_dasha.get_current_narayana_dasha(periods, age)
major = active.get("md") or {}
minor = active.get("ad") or {}
return major.get("sign_idx"), minor.get("sign_idx")
if antardasha_by_md is None:
active = narayana_dasha.get_current_narayana_dasha(periods, age)
major = active.get("md") or {}
minor = active.get("ad") or {}
return major.get("sign_idx"), minor.get("sign_idx")
# BUG-1324: same MD/AD lookup as narayana_dasha.get_current_narayana_dasha
# lines 769-800 (total_cycle, cycle_start, age_in_cycle, MD loop, AD loop).
# PD (lines 801-812) is unused by scoring and is not computed here.
if not periods:
return None, None
total_cycle = sum(p["years"] for p in periods)
if total_cycle == 0:
return None, None
cycle_start = min(float(p.get("start_age", 0)) for p in periods)
age_in_cycle = ((age - cycle_start) % total_cycle) + cycle_start
for period in periods:
if period["start_age"] <= age_in_cycle < period["end_age"]:
md_sign = period.get("sign_idx")
ads = antardasha_by_md.get(period["sign_idx"]) or []
for ad in ads:
if ad["start_age"] <= age_in_cycle < ad["end_age"]:
return md_sign, ad.get("sign_idx")
return md_sign, None
return None, None
def _varga_chart(charts: dict, prefix: str) -> dict | None:
@@ -198,13 +219,15 @@ def _score_event(
vimshottari: tuple[str, str, str],
narayana: tuple[int | None, int | None],
arudha_padas: dict,
functional: dict | None = None,
) -> CandidateEvidence:
_, target_houses = DOMAIN_CONFIG[event["domain"]]
ascendant_index = int(natal_chart["ascendant"]["lon"] // 30)
target_lords = _house_lords(ascendant_index, target_houses)
functional = functional_benefics.derive_functional_benefic_malefic(
natal_chart["ascendant"].get("sign")
)
if functional is None:
functional = functional_benefics.derive_functional_benefic_malefic(
natal_chart["ascendant"].get("sign")
)
functional_benefics_set = set(functional.get("functional_benefics") or [])
functional_malefics_set = set(functional.get("functional_malefics") or [])
major_lord, minor_lord, pratyantar_lord = vimshottari
@@ -542,6 +565,18 @@ def build_candidate_static_context(
)
except (KeyError, TypeError, ValueError):
narayana_periods = None
functional_profile = functional_benefics.derive_functional_benefic_malefic(
chart["ascendant"].get("sign")
)
narayana_antardasha_by_md = None
if narayana_periods:
narayana_antardasha_by_md = {
period["sign_idx"]: narayana_dasha.calc_narayana_antardasha(
narayana_periods,
period["sign_idx"],
)
for period in narayana_periods
}
birth_info = chart.get("birth_info") if isinstance(chart.get("birth_info"), dict) else {}
kp_snapshot = observe_kp_cusps(
@@ -595,6 +630,8 @@ def build_candidate_static_context(
"shadbala_result": shadbala_result,
"vimshottari_timeline": vimshottari_timeline,
"narayana_periods": narayana_periods,
"functional_profile": functional_profile,
"narayana_antardasha_by_md": narayana_antardasha_by_md,
}
@@ -649,6 +686,7 @@ def _candidate_row(
candidate_at,
event_at,
context.get("narayana_periods"),
context.get("narayana_antardasha_by_md"),
)
except (KeyError, TypeError, ValueError):
missing_layers.append("Narayana_MD_AD")
@@ -664,6 +702,7 @@ def _candidate_row(
vimshottari=vimshottari,
narayana=narayana,
arudha_padas=arudha_padas,
functional=context.get("functional_profile"),
))
transit_rules = _controlled_transit_rules(
request,
+611
View File
@@ -0,0 +1,611 @@
"""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",
# 与 consult_full_data_cache 同走 get_or_build,每次调用先 prune_cache(name=…) 删两天前的文件,自己会清;含出生资料,保留期 2 天。
"report_reference_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
+60 -9
View File
@@ -4,7 +4,9 @@ from typing import Any, Mapping, Sequence
from uuid import NAMESPACE_URL, uuid5
from scripts.active_rectification_events import build_candidate_result_summary
from scripts.offline_compute_mixins import BadRequest
from scripts.rectification.candidate_feature_service import build_candidate_feature_snapshot
from scripts.rectification.candidate_window import enumerate_candidate_window
from scripts.rectification.contracts import (
EVENT_CONTRACT_VERSION,
RectificationRequest,
@@ -195,6 +197,35 @@ def _rectification_report(
}
MAX_CANDIDATES_PER_REQUEST = 241
class TooManyCandidates(BadRequest):
"""Rejected before any chart is computed.
The API server maps every BadRequest to HTTP 400 with error_code
ERR_BAD_REQUEST and this task must not edit that server. The token
too_many_candidates lives in the message, with the cap and the count.
"""
code = "too_many_candidates"
def __init__(self, count: int) -> None:
self.count = int(count)
self.limit = MAX_CANDIDATES_PER_REQUEST
super().__init__(f"too_many_candidates: limit={self.limit} count={self.count}")
def _candidate_count(request: RectificationRequest) -> int:
return len(enumerate_candidate_window(request))
def _reject_if_too_many_candidates(request: RectificationRequest) -> None:
count = _candidate_count(request)
if count > MAX_CANDIDATES_PER_REQUEST:
raise TooManyCandidates(count)
def engine_scoring_versions() -> dict[str, str]:
"""Identity fields also returned by `/api/rectification/v5/score`, without scoring."""
return {
@@ -219,7 +250,14 @@ def candidate_features(request: RectificationRequest) -> dict[str, Any]:
}
def score_candidates(request: RectificationRequest) -> dict[str, Any]:
def score_candidates(
request: RectificationRequest,
*,
include_internal: bool = False,
enforce_candidate_cap: bool = True,
) -> dict[str, Any]:
if enforce_candidate_cap:
_reject_if_too_many_candidates(request)
scoring_request = scoreable_request(request)
built = build_event_contribution_matrix(scoring_request)
rows = score_from_matrix(scoring_request, built)
@@ -272,7 +310,7 @@ def score_candidates(request: RectificationRequest) -> dict[str, Any]:
)
rectification_report["next_step_codes"] = candidate_summary["next_step_codes"]
candidate_summary["report"] = rectification_report
return {
payload = {
"result_id": result_id,
"algorithm_version": ALGORITHM_VERSION,
"event_contract_version": EVENT_CONTRACT_VERSION,
@@ -286,11 +324,9 @@ def score_candidates(request: RectificationRequest) -> dict[str, Any]:
"conflicting_event_ids": [item["event_id"] for item in row["evidence"] if item["points"] < 0],
} for row in rows],
"candidate_decisions": candidate_decisions,
"candidate_decision_receipt": decision_receipt,
"decision_receipt": decision_receipt,
"execution_ledger_version": EXECUTION_LEDGER_VERSION,
"execution_ledger": execution_ledger,
"event_contribution_matrix": built["matrix"],
"candidate_feature_snapshot": build_candidate_feature_snapshot(
scoring_request, spec_hash, built.get("static_contexts")
),
@@ -317,10 +353,20 @@ def score_candidates(request: RectificationRequest) -> dict[str, Any]:
"margin_percent": decision_receipt["margin_percent"],
"can_confirm_exact_minute": decision_receipt["confirmation_allowed"],
}
if include_internal:
payload["candidate_decision_receipt"] = decision_receipt
payload["event_contribution_matrix"] = built["matrix"]
return payload
def diagnostics(request: RectificationRequest) -> dict[str, Any]:
scored = score_candidates(request)
def diagnostics(
request: RectificationRequest,
*,
enforce_candidate_cap: bool = True,
) -> dict[str, Any]:
if enforce_candidate_cap:
_reject_if_too_many_candidates(request)
scored = score_candidates(request, enforce_candidate_cap=enforce_candidate_cap)
return {
"result_id": scored["result_id"],
"algorithm_version": scored["algorithm_version"],
@@ -328,7 +374,6 @@ def diagnostics(request: RectificationRequest) -> dict[str, Any]:
"decision_policy_version": scored["decision_policy_version"],
"calculation_spec_hash": scored["calculation_spec_hash"],
"candidate_decisions": scored["candidate_decisions"],
"candidate_decision_receipt": scored["candidate_decision_receipt"],
"decision_receipt": scored["decision_receipt"],
"execution_ledger_version": scored["execution_ledger_version"],
"execution_ledger": scored["execution_ledger"],
@@ -460,7 +505,11 @@ def _normalize_relative_support(raw: Sequence[float]) -> list[float]:
return shares
def block_scan(request: RectificationRequest) -> dict[str, Any]:
def block_scan(
request: RectificationRequest,
*,
enforce_candidate_cap: bool = True,
) -> dict[str, Any]:
"""Aggregate event scores into declared periods or caller-supplied sub-blocks."""
requested_step = request.get("minute_step")
if isinstance(requested_step, int) and requested_step > 1:
@@ -472,7 +521,9 @@ def block_scan(request: RectificationRequest) -> dict[str, Any]:
if step <= 1:
step = 10
scoring_request = {**request, "minute_step": step}
scored = score_candidates(scoring_request)
if enforce_candidate_cap:
_reject_if_too_many_candidates(scoring_request)
scored = score_candidates(scoring_request, enforce_candidate_cap=enforce_candidate_cap)
events_by_id = {
str(event.get("id") or ""): event
for event in request.get("events") or []
@@ -130,6 +130,8 @@ def merge_transition_proximity(
for context in static_contexts
if isinstance(context, dict) and (time := _context_time(context))
}
# BUG-1324: one unfiltered dasha walk per candidate, then filter by event year.
# Released when this function returns. Not a module-level cache.
vim_cache: dict[tuple[Any, ...], list[date]] = {}
pd_cache: dict[tuple[Any, ...], list[date]] = {}
narayana_cache: dict[tuple[Any, ...], list[date] | None] = {}
@@ -156,14 +158,15 @@ def merge_transition_proximity(
# Legacy time-only contexts still use the request's date.
candidate_at = context.get("candidate_at")
candidate_date = candidate_at.date().isoformat() if isinstance(candidate_at, datetime) else birth_date
vim_key = (candidate_date, round(float(moon), 6), lo, hi)
moon_key = round(float(moon), 6)
vim_key = (candidate_date, moon_key)
if vim_key not in vim_cache:
vim_cache[vim_key] = _vim_start_dates(candidate_date, float(moon), lo, hi)
vim_cache[vim_key] = _vim_start_dates(candidate_date, float(moon), 1, 9999)
pd_cache[vim_key] = _vim_start_dates(
candidate_date,
float(moon),
lo,
hi,
1,
9999,
include_pratyantar=True,
)
planets = context.get("planet_longitudes") or {}
@@ -171,25 +174,26 @@ def merge_transition_proximity(
narayana_key = (
candidate_date,
int(asc) if isinstance(asc, int) else None,
lo,
hi,
round(float(moon), 6),
moon_key,
)
if narayana_key not in narayana_cache:
narayana_cache[narayana_key] = (
_narayana_start_dates(int(asc), planets, candidate_date, lo, hi)
_narayana_start_dates(int(asc), planets, candidate_date, 1, 9999)
if isinstance(asc, int) and isinstance(planets, dict)
else None
)
ad_starts = vim_cache[vim_key]
ad_starts = [item for item in vim_cache[vim_key] if lo <= item.year <= hi]
ad_set = set(ad_starts)
pd_only = [item for item in pd_cache[vim_key] if item not in ad_set]
pd_only = [item for item in pd_cache[vim_key] if lo <= item.year <= hi and item not in ad_set]
narayana_starts = narayana_cache[narayana_key]
if narayana_starts is not None:
narayana_starts = [item for item in narayana_starts if lo <= item.year <= hi]
scored = score_transition_proximity(
event_date=event_date,
precision=precision,
vim_starts=ad_starts,
vim_pd_starts=pd_only,
narayana_starts=narayana_cache[narayana_key] or [],
narayana_starts=narayana_starts or [],
)
if scored["points"] <= 0 and not scored["rule_ids"]:
continue
+10 -3
View File
@@ -814,6 +814,7 @@ def _score_year(
vimshottari=vimshottari,
narayana=narayana,
arudha_padas=context.get("arudha_padas") or {},
functional=context.get("functional_profile"),
)
@@ -1913,8 +1914,12 @@ def guided_collect_windows(
vim_cache=vim_cache,
narayana_cache=narayana_cache,
)
# BUG-1323: gathered depends only on (track, boundary index). Prefetch once
# per candidate and skip a boundary after the one==two check has passed.
work_tracks = [starts_by_time.get(_context_time(context) or "") or {} for context in work]
rows: list[dict[str, Any]] = []
seen: set[tuple[int, int, int, str]] = set()
done: set[tuple[str, int]] = set()
for left, right in _representative_pairs(work):
left_time = str(_context_time(left) or "")
right_time = str(_context_time(right) or "")
@@ -1928,10 +1933,12 @@ def guided_collect_windows(
for index, (one, two) in enumerate(zip(left_dates, right_dates)):
if one == two:
continue
if (track, index) in done:
continue
done.add((track, index))
gathered: list[date] = []
for context in work:
time = _context_time(context)
dates = (starts_by_time.get(time or "") or {}).get(track) or []
for tracks in work_tracks:
dates = tracks.get(track) or []
if index < len(dates):
gathered.append(dates[index])
if len(gathered) < 2 or min(gathered) == max(gathered):
@@ -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")
@@ -50,7 +50,7 @@ def worker(root: Path, dated: bool):
request = make_request(case)
if dated:
request["candidate_intervals"] = [{"start_at": f'{request["birth_date"]}T{request["start_time"]}', "end_at": f'{request["birth_date"]}T{request["end_time"]}'}]
result.append(projection(score(request), width))
result.append(projection(score(request, include_internal=True), width))
print(json.dumps(result))
@@ -0,0 +1,257 @@
#!/usr/bin/env python3
"""Bench and canonical dump for the 2026-10-13 rectification scoring speedup.
Read-only aside from the dump files the caller asks for. Cross-process dumps
must be launched with PYTHONHASHSEED=0. The baseline tree has no
include_internal parameter; that call falls back to the old signature.
"""
from __future__ import annotations
import argparse
import hashlib
import json
import statistics
import sys
import time
from datetime import date
from pathlib import Path
from typing import Any, Callable
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from scripts.rectification.api_service import block_scan, diagnostics, score_candidates
from scripts.rectification.contracts import normalize_rectification_request
from scripts.rectification import event_probes
from scripts.rectification import scoring_service
from scripts import active_rectification_event_engine as event_engine
FROZEN_TODAY = date(2026, 9, 16)
EVENT_SPECS = (
("education", "education_start", "2008-01-01", "2008-12-31", "year"),
("career", "career_entry", "2012-06-01", "2012-06-30", "month"),
("relocation", "relocation", "2014-03-01", "2014-03-31", "month"),
("relationship", "relationship_commitment", "2016-10-01", "2016-10-31", "month"),
("career", "promotion", "2018-01-01", "2018-12-31", "year"),
("finance", "asset_change", "2019-05-01", "2019-05-31", "month"),
("family", "family_event", "2020-08-01", "2020-08-31", "month"),
("health", "self_health_event", "2022-01-01", "2022-12-31", "year"),
)
class _FrozenDate(date):
@classmethod
def today(cls) -> date:
return FROZEN_TODAY
def _events() -> list[dict[str, Any]]:
rows = []
for index, (domain, kind, start, end, precision) in enumerate(EVENT_SPECS, start=1):
rows.append({
"id": f"00000000-0000-4000-8000-{index:012d}",
"domain": domain,
"event_kind": kind,
"date_start": start,
"date_end": end,
"precision": precision,
"summary": "虚构",
})
return rows
def make_request(start: str, end: str, **extra: Any) -> dict[str, Any]:
body = {
"birth_date": "1990-01-01",
"start_time": start,
"end_time": end,
"lat": 39.9,
"lon": 116.4,
"tz": 8,
"events": _events(),
**extra,
}
return normalize_rectification_request(body, today=FROZEN_TODAY)
def _strip(value: Any, keys: set[str]) -> Any:
if isinstance(value, dict):
return {key: _strip(item, keys) for key, item in value.items() if key not in keys}
if isinstance(value, list):
return [_strip(item, keys) for item in value]
return value
def _canonical(value: Any) -> bytes:
return json.dumps(value, sort_keys=True, ensure_ascii=True, separators=(",", ":")).encode("ascii")
def _score(request: dict[str, Any]) -> dict[str, Any]:
try:
return score_candidates(request, include_internal=True)
except TypeError:
return score_candidates(request)
def _call(kind: str, request: dict[str, Any], *, internal: bool = False) -> dict[str, Any]:
if kind == "score":
# Bench times the product call. Dump asks for the internal matrix so
# the baseline tree, which always returned it, stays comparable.
if internal:
return _score(request)
return score_candidates(request)
if kind == "block":
return block_scan(request)
if kind == "diagnostics":
return diagnostics(request)
raise ValueError(kind)
def scenarios() -> list[dict[str, Any]]:
from tests.test_rectification_engine_memoization import public_score_request
rows = [
{"name": "12:00-12:20", "kind": "score", "request": make_request("12:00", "12:20")},
{"name": "11:45-12:15", "kind": "score", "request": make_request("11:45", "12:15")},
{"name": "11:00-13:00", "kind": "score", "request": make_request("11:00", "13:00")},
{"name": "23:20-00:40", "kind": "score", "request": make_request("23:20", "00:40")},
{"name": "declined-relationship-career", "kind": "score", "request": make_request(
"11:30", "12:30", declined_domains=["relationship", "career"],
)},
{"name": "refresh-probes-columns", "kind": "score", "request": make_request(
"11:30", "12:30", refresh_probes=True, column_times=["11:40", "12:10", "12:20"],
)},
{"name": "minute-step-2", "kind": "score", "request": make_request("10:00", "13:00", minute_step=2)},
{"name": "block-step-10", "kind": "block", "request": make_request("00:00", "23:59", minute_step=10)},
{"name": "block-step-5", "kind": "block", "request": make_request("10:00", "14:00", minute_step=5)},
{"name": "public-score-request", "kind": "score", "request": normalize_rectification_request(
public_score_request(), today=FROZEN_TODAY,
)},
{"name": "diagnostics-11:00-13:00", "kind": "diagnostics", "request": make_request("11:00", "13:00")},
]
return rows
def _strip_keys(kind: str) -> set[str]:
# diagnostics no longer repeats the receipt. Drop it on both sides so the
# remaining payload is what must stay byte-identical.
if kind == "diagnostics":
return {"column_compare_ms", "candidate_decision_receipt"}
return {"column_compare_ms"}
class _SegmentTimer:
def __init__(self) -> None:
self.totals = {
"guided_collect_windows": 0.0,
"merge_transition_proximity": 0.0,
"get_current_narayana_dasha": 0.0,
"derive_functional_benefic_malefic": 0.0,
"build_candidate_static_context": 0.0,
}
def wrap(self, owner: Any, name: str, key: str) -> Callable[[], None]:
original = getattr(owner, name)
def wrapped(*args: Any, **kwargs: Any) -> Any:
started = time.perf_counter()
try:
return original(*args, **kwargs)
finally:
self.totals[key] += time.perf_counter() - started
setattr(owner, name, wrapped)
def restore() -> None:
setattr(owner, name, original)
return restore
def _median(samples: list[float]) -> float:
return round(statistics.median(samples), 4)
def bench() -> dict[str, Any]:
jobs = [
("12:00-12:20", "score", make_request("12:00", "12:20")),
("11:45-12:15", "score", make_request("11:45", "12:15")),
("11:00-13:00", "score", make_request("11:00", "13:00")),
("block-step-10", "block", make_request("00:00", "23:59", minute_step=10)),
]
rows = []
event_probes.date = _FrozenDate
try:
for name, kind, request in jobs:
samples = []
segments = []
sizes = []
counts = []
for _ in range(5):
timer = _SegmentTimer()
restores = [
timer.wrap(event_probes, "guided_collect_windows", "guided_collect_windows"),
timer.wrap(scoring_service, "merge_transition_proximity", "merge_transition_proximity"),
timer.wrap(event_engine.narayana_dasha, "get_current_narayana_dasha", "get_current_narayana_dasha"),
timer.wrap(event_engine.functional_benefics, "derive_functional_benefic_malefic", "derive_functional_benefic_malefic"),
timer.wrap(event_engine, "build_candidate_static_context", "build_candidate_static_context"),
]
started = time.perf_counter()
try:
result = _call(kind, request)
finally:
for restore in restores:
restore()
elapsed = time.perf_counter() - started
samples.append(elapsed)
segments.append({key: round(value, 4) for key, value in timer.totals.items()})
sizes.append(len(_canonical(_strip(result, _strip_keys(kind)))))
counts.append(result.get("candidate_count") or len(result.get("candidate_scores") or []))
median_segments: dict[str, float] = {}
for key in segments[0]:
median_segments[key] = _median([item[key] for item in segments])
rows.append({
"window": name,
"candidate_count": counts[0],
"median_seconds": _median(samples),
"segment_median_seconds": median_segments,
"response_bytes": int(statistics.median(sizes)),
})
finally:
event_probes.date = date
return {"runs": 5, "rows": rows}
def dump(out_dir: Path) -> list[dict[str, str]]:
out_dir.mkdir(parents=True, exist_ok=True)
written = []
event_probes.date = _FrozenDate
try:
for row in scenarios():
result = _call(row["kind"], row["request"], internal=True)
payload = _strip(result, _strip_keys(row["kind"]))
raw = _canonical(payload)
digest = hashlib.sha256(raw).hexdigest()
path = out_dir / f"{row['name'].replace(':', '')}.json"
path.write_bytes(raw)
written.append({"name": row["name"], "sha256": digest, "bytes": str(len(raw))})
print(f"{row['name']}\t{digest}\t{len(raw)}")
finally:
event_probes.date = date
return written
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("command", choices=("bench", "dump"))
parser.add_argument("--out", type=Path, default=ROOT / "artifacts" / "rect-scoring-perf-20261013")
args = parser.parse_args()
if args.command == "bench":
print(json.dumps(bench(), ensure_ascii=True, indent=2))
else:
dump(args.out)
if __name__ == "__main__":
main()
+5 -2
View File
@@ -30,8 +30,11 @@ from scripts.research.sealed_holdout_rerun import (
OFFSETS = (-30, -20, -15, -10, -8, -5, -3, 0, 3, 5, 8, 10, 15, 20, 30)
RADII = (15, 30, 60)
MINUTE_STEP = 1
FREEZE = ROOT / "docs/research/reported_offset_ephemeris_nutation_2026_10_08.freeze.json"
REPORT = ROOT / "docs/research/reported_offset_ephemeris_nutation_2026_10_08.json"
# 原值: docs/research/reported_offset_ephemeris_nutation_2026_10_08.{freeze.json,json}
# 新值: docs/research/reported_offset_scoring_perf_2026_10_13.{freeze.json,json}
# 原因: BUG-1323~1325 改了冻结身份文件,输出不变,ERR-110 另存新记录
FREEZE = ROOT / "docs/research/reported_offset_scoring_perf_2026_10_13.freeze.json"
REPORT = ROOT / "docs/research/reported_offset_scoring_perf_2026_10_13.json"
LEGACY_REPORT = ROOT / "docs/research/reported_offset_2026_09_20.json"
+5 -2
View File
@@ -27,8 +27,11 @@ from scripts.minute_rectification_feature_facts_v4 import build_feature_fact_row
from scripts.minute_rectification_holdout_validator import validate
DATASET = ROOT / "references/real_case_calibration/minute_rectification_holdout_v3.json"
FREEZE = ROOT / "docs/research/sealed_holdout_rerun_ephemeris_nutation_2026_10_08.freeze.json"
REPORT = ROOT / "docs/research/sealed_holdout_rerun_ephemeris_nutation_2026_10_08.json"
# 原值: docs/research/sealed_holdout_rerun_ephemeris_nutation_2026_10_08.{freeze.json,json}
# 新值: docs/research/sealed_holdout_rerun_scoring_perf_2026_10_13.{freeze.json,json}
# 原因: BUG-1323~1325 改了冻结身份文件,输出不变,ERR-110 另存新记录
FREEZE = ROOT / "docs/research/sealed_holdout_rerun_scoring_perf_2026_10_13.freeze.json"
REPORT = ROOT / "docs/research/sealed_holdout_rerun_scoring_perf_2026_10_13.json"
LEGACY_REPORT = ROOT / "docs/research/sealed_holdout_rerun_2026_09_20.json"
ARCHIVE = ROOT / "docs/research/history/rectification_pre_cross_midnight_2026_09_20"
PRODUCTION_FILES = [
+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