合并 staging，并去掉已删组件留下的样式名单条目

knownUnstyled 不再保留 personal-report-additional-charts。它只被删掉的报告组件用过。其它白名单没有同类残留。业务代码没改。

自测：类名合同和两条否定合同 6 通过，失败 0，取消 0。tsc 0 错，lint 0 error。

Co-Authored-By: Grok (agent loop)

tests/test_api_scratch_retention.py

@@ -0,0 +1,314 @@
+"""Retention, chart-cache keys, and VedAstro audit switch."""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import threading
+import time
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+import pytest
+
+import api_scratch_retention as retention
+import jyotish_api_server as api
+import vedastro_service_adapter as adapter
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _hex(label: str) -> str:
+    return hashlib.sha256(label.encode("utf-8")).hexdigest() + ".json"
+
+
+def _isolate(monkeypatch, tmp_path: Path) -> None:
+    chart = tmp_path / "chart"
+    jobs = tmp_path / "jobs"
+    chart.mkdir()
+    jobs.mkdir()
+    monkeypatch.setattr(api, "_api_chart_cache_dir", lambda: chart)
+    monkeypatch.setattr(api, "_async_job_dir", lambda scope: jobs)
+    monkeypatch.setattr(api, "_async_job_backend", lambda: "file")
+    monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path / "adapter")
+    monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR", str(tmp_path / "snap"))
+    monkeypatch.setenv("JYOTISH_VEDASTRO_SNAPSHOT_NEGATIVE_CACHE_DIR", str(tmp_path / "neg"))
+    monkeypatch.setenv("VEDASTRO_GATEWAY_QUEUE_DIR", str(tmp_path / "gw"))
+    monkeypatch.setenv("JYOTISH_CACHE_DIR", str(tmp_path / "cache"))
+    monkeypatch.setenv("JYOTISH_API_CHART_CACHE_TTL_SECONDS", "900")
+    monkeypatch.delenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", raising=False)
+
+
+def _write(path: Path, payload) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload), encoding="utf-8")
+
+
+def test_prune_all_drops_expired_and_old_bad_chart_cache_only(monkeypatch, tmp_path):
+    _isolate(monkeypatch, tmp_path)
+    chart = api._api_chart_cache_dir()
+    now = time.time()
+    fresh = chart / _hex("fresh")
+    expired = chart / _hex("expired")
+    bad_old = chart / _hex("bad-old")
+    bad_young = chart / _hex("bad-young")
+    stranger = chart / "notes.json"
+    _write(fresh, {"cache_key": "a", "created_at": now, "payload": {}})
+    _write(expired, {"cache_key": "b", "created_at": now - 5000, "payload": {}})
+    bad_old.write_text("{", encoding="utf-8")
+    bad_young.write_text("{", encoding="utf-8")
+    stranger.write_text("{}", encoding="utf-8")
+    old = now - (25 * 3600)
+    os.utime(bad_old, (old, old))
+    os.utime(stranger, (old, old))
+
+    result = retention.prune_all(now=now)
+
+    assert fresh.exists()
+    assert bad_young.exists()
+    assert stranger.exists()
+    assert not expired.exists()
+    assert not bad_old.exists()
+    assert result["directories"]["api_chart_cache"]["removed"] == 2
+    assert result["errors"] == 0
+
+
+def test_job_cleanup_still_removes_only_expired_records(monkeypatch, tmp_path):
+    _isolate(monkeypatch, tmp_path)
+    jobs = api._async_job_dir("any")
+    expired = jobs / "expired.json"
+    active = jobs / "active.json"
+    bad = jobs / "bad.json"
+    expired.write_text(json.dumps({"expires_at_unix": time.time() - 1}), encoding="utf-8")
+    active.write_text(json.dumps({"expires_at_unix": time.time() + 60}), encoding="utf-8")
+    bad.write_text("{", encoding="utf-8")
+
+    result = api.prune_expired_async_jobs()
+
+    assert result["removed"] == 1
+    assert not expired.exists()
+    assert active.exists()
+    assert bad.exists()
+
+
+def test_interval_zero_does_not_start_a_thread():
+    before = [item.name for item in threading.enumerate()]
+    assert retention.start_periodic_prune(0) is None
+    assert [item.name for item in threading.enumerate()] == before
+
+
+def test_prune_counts_unlink_errors_without_raising(monkeypatch, tmp_path):
+    _isolate(monkeypatch, tmp_path)
+    chart = api._api_chart_cache_dir()
+    target = chart / _hex("stuck")
+    _write(target, {"created_at": time.time() - 5000, "payload": {}})
+    real_unlink = Path.unlink
+
+    def _boom(self, *args, **kwargs):
+        if self == target:
+            raise OSError("denied")
+        return real_unlink(self, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "unlink", _boom)
+    result = retention.prune_all(now=time.time())
+    assert result["errors"] > 0
+    assert target.exists()
+
+
+def _chart_key(body):
+    return api._api_chart_cache_key(api._build_api_chart_cache_payload(body))
+
+
+def test_numeric_birth_spellings_share_one_chart_cache_key():
+    base = {
+        "year": 1990,
+        "month": 1,
+        "day": 1,
+        "minute": 0,
+        "lat": 28.6139,
+        "lon": 77.2090,
+        "tz": 5.5,
+        "today": "2026-10-14",
+    }
+    keys = {_chart_key(dict(base, hour=hour)) for hour in (12, 12.0, "12")}
+    assert len(keys) == 1
+    assert _chart_key(dict(base, hour=12, second=0)) in keys
+    assert _chart_key(dict(base, hour=12, lat="28.6139")) in keys
+    assert _chart_key(dict(base, hour=12, tz="5.5")) in keys
+    assert _chart_key(dict(base, hour=12, tz="Asia/Kolkata")) not in keys
+    skipped = _chart_key(dict(base, hour=12, skip_vedastro_main_entry_overview=True))
+    assert skipped not in keys
+    assert api._build_api_chart_cache_payload(base)["cache_schema_version"] == 6
+
+
+def test_audit_artifact_default_off_and_compact_when_on(monkeypatch, tmp_path):
+    monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path)
+    monkeypatch.delenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", raising=False)
+    manifest = {"requests": [{"section": "chart_core"}], "reference_date": "2026-10-14"}
+    quiet = adapter._normalize_official_full_snapshot_success(
+        "https://example.test/api",
+        manifest,
+        {"chart_core": {"Status": "Pass"}},
+        {"chart_core": "ok"},
+        1,
+        [],
+    )
+    assert quiet["source_metadata"]["artifact_path"] is None
+    assert list(tmp_path.rglob("*.json")) == []
+
+    monkeypatch.setenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", "1")
+    written = adapter._normalize_official_full_snapshot_success(
+        "https://example.test/api",
+        manifest,
+        {"chart_core": {"Status": "Pass"}},
+        {"chart_core": "ok"},
+        1,
+        [],
+    )
+    rel = written["source_metadata"]["artifact_path"]
+    path = Path(rel)
+    if not path.is_absolute():
+        path = adapter.ROOT / rel
+    assert path.exists()
+    # Linux reports the mode bit. Windows keeps the file writable and does not surface 0600.
+    if os.name != "nt":
+        assert path.stat().st_mode & 0o777 == 0o600
+    text = path.read_text(encoding="utf-8")
+    assert "\n" not in text
+    assert json.loads(text)["status"] == "ok"
+
+
+def test_failed_snapshot_expires_before_success(monkeypatch, tmp_path):
+    monkeypatch.setattr(adapter, "ARTIFACT_DIR", tmp_path)
+    monkeypatch.setenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", "604800")
+    monkeypatch.delenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_NEGATIVE_TTL_SECONDS", raising=False)
+    calls = {"fail": 0, "ok": 0}
+
+    def fake_run(case, case_id="user_chart"):
+        if case.get("day") == 2:
+            calls["fail"] += 1
+            return {"available": False, "status": "timeout", "source_metadata": {}}
+        calls["ok"] += 1
+        return {"available": True, "status": "ok", "source_metadata": {}}
+
+    monkeypatch.setattr(adapter, "_run_official_full_snapshot_case", fake_run)
+    fail_case = {"year": 1990, "month": 1, "day": 2, "hour": 12, "lat": 1.0, "lon": 2.0, "tz": 5.5}
+    ok_case = {"year": 1990, "month": 1, "day": 3, "hour": 12, "lat": 1.0, "lon": 2.0, "tz": 5.5}
+    adapter.run_official_full_snapshot_for_case(fail_case, case_id="fail")
+    adapter.run_official_full_snapshot_for_case(ok_case, case_id="ok")
+    cache_dir = tmp_path / "official_full_snapshot_cache"
+    aged = time.time() - 301
+    for path in cache_dir.glob("*.json"):
+        record = json.loads(path.read_text(encoding="utf-8"))
+        record["created_at"] = aged
+        path.write_text(json.dumps(record), encoding="utf-8")
+        if record.get("negative") is True:
+            assert record["negative"] is True
+    adapter.run_official_full_snapshot_for_case(fail_case, case_id="fail")
+    adapter.run_official_full_snapshot_for_case(ok_case, case_id="ok")
+    assert calls["fail"] == 2
+    assert calls["ok"] == 1
+
+
+def test_gateway_status_reports_cache_ttl_fallback(monkeypatch):
+    from scripts import vedastro_gateway
+
+    monkeypatch.setenv("JYOTISH_SKIP_LOCAL_ENV", "1")
+    monkeypatch.delenv("VEDASTRO_OFFICIAL_FULL_SNAPSHOT_CACHE_TTL_SECONDS", raising=False)
+    monkeypatch.setenv("VEDASTRO_CACHE_TTL_SECONDS", "604800")
+    status = vedastro_gateway.gateway_status()
+    assert status["official_full_snapshot_cache_ttl_seconds"] == 604800
+    assert status["official_full_snapshot_negative_cache_ttl_seconds"] == 300
+
+
+def _day(offset: int) -> str:
+    return (datetime.now(timezone.utc).date() + timedelta(days=offset)).isoformat()
+
+
+def test_snapshot_negative_gateway_and_varga_prune_rules(monkeypatch, tmp_path):
+    _isolate(monkeypatch, tmp_path)
+    now = time.time()
+    today = datetime.fromtimestamp(now, timezone.utc).date()
+    snap = tmp_path / "snap"
+    neg = tmp_path / "neg"
+    gw = tmp_path / "gw"
+    varga = tmp_path / "cache" / "varga_segments"
+    keep_ref = (today - timedelta(days=7)).isoformat()
+    drop_ref = (today - timedelta(days=9)).isoformat()
+    _write(snap / _hex("snap-keep"), {"schema": "vedastro_snapshot_cache.v1", "reference_date": keep_ref})
+    _write(snap / _hex("snap-drop"), {"schema": "vedastro_snapshot_cache.v1", "reference_date": drop_ref})
+    _write(snap / "notes.json", {"reference_date": "1990-01-01"})
+    _write(neg / _hex("neg-today"), {"schema": "vedastro_snapshot_negative_cache.v1", "reference_date": today.isoformat()})
+    _write(neg / _hex("neg-yesterday"), {"schema": "vedastro_snapshot_negative_cache.v1", "reference_date": (today - timedelta(days=1)).isoformat()})
+    _write(neg / "stray.json", {"reference_date": "1990-01-01"})
+    job_id = "vgw_" + "ab" * 10
+    old_stamp = datetime.fromtimestamp(now - 90000, timezone.utc).isoformat()
+    fresh_stamp = datetime.fromtimestamp(now, timezone.utc).isoformat()
+    _write(gw / f"{job_id}.json", {"job_id": job_id, "status": "completed", "updated_at": old_stamp})
+    (gw / f"{job_id}.official_raw_response.json").write_text("{}", encoding="utf-8")
+    running = "vgw_" + "cd" * 10
+    _write(gw / f"{running}.json", {"job_id": running, "status": "running", "updated_at": old_stamp})
+    failed = "vgw_" + "ef" * 10
+    _write(gw / f"{failed}.json", {"job_id": failed, "status": "failed", "updated_at": fresh_stamp})
+    (gw / "notes.json").write_text("{}", encoding="utf-8")
+    old_varga = varga / _hex("old-segment")
+    young_varga = varga / _hex("young-segment")
+    _write(old_varga, {"window_key": "x"})
+    _write(young_varga, {"window_key": "y"})
+    _write(varga / "notes.json", {"window_key": "z"})
+    os.utime(old_varga, (now - 8 * 86400, now - 8 * 86400))
+
+    monkeypatch.setenv("VEDASTRO_WRITE_AUDIT_ARTIFACTS", "0")
+    artifact = tmp_path / "adapter" / "official_full_snapshot-0123456789ab-0123456789ab.json"
+    artifact.parent.mkdir(parents=True, exist_ok=True)
+    artifact.write_text("{}", encoding="utf-8")
+    os.utime(artifact, (now - 25 * 3600, now - 25 * 3600))
+    kept_artifact = tmp_path / "adapter" / "method_catalog_snapshot.json"
+    kept_artifact.write_text("{}", encoding="utf-8")
+    os.utime(kept_artifact, (now - 25 * 3600, now - 25 * 3600))
+
+    retention.prune_all(now=now)
+
+    assert (snap / _hex("snap-keep")).exists()
+    assert not (snap / _hex("snap-drop")).exists()
+    assert (snap / "notes.json").exists()
+    assert (neg / _hex("neg-today")).exists()
+    assert not (neg / _hex("neg-yesterday")).exists()
+    assert (neg / "stray.json").exists()
+    assert not (gw / f"{job_id}.json").exists()
+    assert not (gw / f"{job_id}.official_raw_response.json").exists()
+    assert (gw / f"{running}.json").exists()
+    assert (gw / f"{failed}.json").exists()
+    assert (gw / "notes.json").exists()
+    assert not old_varga.exists()
+    assert young_varga.exists()
+    assert (varga / "notes.json").exists()
+    assert not artifact.exists()
+    assert kept_artifact.exists()
+
+
+def _scratch_local_names() -> set[str]:
+    names: set[str] = set()
+    slash = re.compile(r"scratch/local/([A-Za-z0-9_.]+)")
+    joined = re.compile(
+        r"""['\"]scratch['\"]\s*/\s*['\"]local['\"]\s*/\s*(?:f)?['\"]([A-Za-z0-9_.]+)['\"]"""
+    )
+    after_local = re.compile(
+        r"""['\"]local['\"]\s*\)*\s*/\s*['\"]([A-Za-z0-9_.]+)['\"]"""
+    )
+    for path in (ROOT / "scripts").rglob("*.py"):
+        text = path.read_text(encoding="utf-8")
+        names.update(slash.findall(text))
+        names.update(joined.findall(text))
+        names.update(after_local.findall(text))
+    return names
+
+
+def test_new_scratch_local_dirs_are_registered_or_whitelisted():
+    allowed = set(retention.REGISTERED_DIR_NAMES) | set(retention.OFFLINE_SCRATCH_NAMES)
+    found = _scratch_local_names()
+    unexpected = sorted(found - allowed)
+    assert unexpected == [], unexpected
+    assert "api_chart_cache" in found
+    assert "consult_full_data_cache" in retention.OFFLINE_SCRATCH_NAMES

tests/test_report_reference_perf.py

@@ -0,0 +1,374 @@
+"""Byte-identical report shortcuts and the full_data markdown cache."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+import os
+import re
+import threading
+import time
+from typing import Any
+
+import pytest
+
+from scripts.calculation_profile_contract import hash_bound_result
+from scripts.pl9_full_data_export import render_full_data_markdown
+from scripts.pl9_language_terms import _thicken_zh_dasha_density, replace_natural_language_term
+from scripts.report_perf_flags import (
+    REFERENCE_CACHE_DEFAULT,
+    RENDER_FAST_DEFAULT,
+    report_reference_cache_enabled,
+    report_render_fast_enabled,
+)
+from scripts.professional_report_reference import build_professional_report_reference
+from tests.test_report_english_edition import CASES, _packet
+
+
+def _legacy_drop_key_recursive(payload: Any, key_name: str) -> Any:
+    if isinstance(payload, dict):
+        return {
+            key: _legacy_drop_key_recursive(value, key_name)
+            for key, value in payload.items()
+            if key != key_name
+        }
+    if isinstance(payload, list):
+        return [_legacy_drop_key_recursive(item, key_name) for item in payload]
+    return payload
+
+
+def _legacy_drop_dotted_path(payload: Any, path: tuple[str, ...]) -> Any:
+    if len(path) == 2 and path[0] == "**":
+        return _legacy_drop_key_recursive(payload, path[1])
+    if not path or not isinstance(payload, dict) or path[0] not in payload:
+        return payload
+    cloned = dict(payload)
+    key, *rest = path
+    if not rest:
+        cloned.pop(key, None)
+        return cloned
+    cloned[key] = _legacy_drop_dotted_path(payload[key], tuple(rest))
+    return cloned
+
+
+def _legacy_canonicalize(value: Any) -> Any:
+    if isinstance(value, dict):
+        normalized: dict[str, Any] = {}
+        for key, child in value.items():
+            normalized_key = key if isinstance(key, str) else f"__{type(key).__name__}__:{key}"
+            normalized[normalized_key] = _legacy_canonicalize(child)
+        return normalized
+    if isinstance(value, (list, tuple)):
+        return [_legacy_canonicalize(item) for item in value]
+    return value
+
+
+def _legacy_payload(result: dict, scope: dict) -> dict:
+    excluded_top = {"result_hash", "result_binding", "calculation_profile", "calculation_profile_id"} | {
+        str(key) for key in (scope.get("excluded_top_keys") or ())
+    }
+    payload = {key: value for key, value in result.items() if key not in excluded_top}
+    for raw_path in scope.get("excluded_paths") or ():
+        parts = tuple(part for part in str(raw_path).split(".") if part)
+        if parts:
+            payload = _legacy_drop_dotted_path(payload, parts)
+    return payload
+
+
+def _legacy_hash(result: dict, input_hash: str, scope: dict) -> str:
+    encoded = json.dumps(
+        _legacy_canonicalize({"input_hash": input_hash, "result": _legacy_payload(result, scope)}),
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+        default=str,
+    )
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+@pytest.fixture(scope="module")
+def day_views():
+    """One real day packet. ``delivered`` is the bound object; render uses a copy."""
+    from scripts.jyotish_engine import build_professional_report_reference_packet, cmd_full_reading
+    from scripts.professional_report_reference import _export_args
+
+    os.environ["JYOTISH_REPORT_RENDER_FAST"] = "0"
+    os.environ["JYOTISH_REPORT_REFERENCE_CACHE"] = "0"
+    try:
+        case = CASES["day"]
+        reading = cmd_full_reading(type("Args", (), dict(case))())
+        delivered = build_professional_report_reference_packet(reading, _export_args(dict(case)), [])
+        packet = dict(delivered)
+        packet["report_version"] = "pl9_personal_long_report.v3"
+        zh = render_full_data_markdown(copy.deepcopy(packet))
+        english = copy.deepcopy(packet)
+        english["report_language"] = "en"
+        en = render_full_data_markdown(english)
+    finally:
+        os.environ.pop("JYOTISH_REPORT_RENDER_FAST", None)
+        os.environ.pop("JYOTISH_REPORT_REFERENCE_CACHE", None)
+    return delivered, packet, zh, en
+
+
+def test_flags_follow_the_environment_at_call_time(monkeypatch) -> None:
+    monkeypatch.delenv("JYOTISH_REPORT_RENDER_FAST", raising=False)
+    monkeypatch.delenv("JYOTISH_REPORT_REFERENCE_CACHE", raising=False)
+    assert report_render_fast_enabled() is RENDER_FAST_DEFAULT
+    assert report_reference_cache_enabled() is REFERENCE_CACHE_DEFAULT
+    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "0")
+    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "0")
+    assert report_render_fast_enabled() is False
+    assert report_reference_cache_enabled() is False
+    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "1")
+    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "1")
+    assert report_render_fast_enabled() is True
+    assert report_reference_cache_enabled() is True
+
+
+def test_fast_render_matches_the_legacy_bytes(day_views, monkeypatch) -> None:
+    _delivered, packet, zh_off, en_off = day_views
+    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "1")
+    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "0")
+    zh_on = render_full_data_markdown(copy.deepcopy(packet))
+    english = copy.deepcopy(packet)
+    english["report_language"] = "en"
+    en_on = render_full_data_markdown(english)
+    assert zh_on == zh_off
+    assert en_on == en_off
+
+
+def _both(function, monkeypatch, *args):
+    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "0")
+    off = function(*args)
+    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "1")
+    on = function(*args)
+    assert on == off
+    return off
+
+
+def test_dasha_density_samples_come_from_the_real_render(day_views, monkeypatch) -> None:
+    _delivered, _packet_view, zh, _en = day_views
+    lines = [line for line in zh.splitlines() if line.strip()]
+    # This fictional chart never emits the density sentence. Fill the existing
+    # sanitizer sentence with lords that do appear in the render.
+    lords = [token for token in ("木星", "北交点", "土星", "金星") if token in zh]
+    assert len(lords) >= 2
+    hit = f"{lords[0]}主大运中的{lords[1]}子运，会把{lords[0]}的自然主题带入当前阶段。"
+    miss = next(line for line in lines if "子运" not in line and "`" not in line)
+    quoted = next(line for line in lines if "`" in line)
+    hit_out = _both(_thicken_zh_dasha_density, monkeypatch, hit + "\n")
+    assert "这一子运不是单独结论" in hit_out or "这个次子运只承担短周期细化" in hit_out
+    miss_out = _both(_thicken_zh_dasha_density, monkeypatch, miss + "\n")
+    assert miss_out == miss
+    quoted_out = _both(_thicken_zh_dasha_density, monkeypatch, quoted + "\n")
+    assert "`" in quoted_out
+    assert quoted in quoted_out
+
+
+def test_term_samples_come_from_the_real_render(day_views, monkeypatch) -> None:
+    _delivered, _packet_view, zh, _en = day_views
+    lines = zh.splitlines()
+    raw = "Ashtakavarga"
+    hit = next(line for line in lines if raw in re.sub(r"`[^`]*`", "", line))
+    miss = next(line for line in lines if line.strip() and raw not in line)
+    span = next(span for span in re.findall(r"`[^`]*`", zh) if raw not in span)
+    hit_out = _both(replace_natural_language_term, monkeypatch, hit, raw, "探针词")
+    assert "探针词" in hit_out or hit_out == hit
+    if re.search(rf"(?<![A-Za-z0-9_]){raw}(?![A-Za-z0-9_])", re.sub(r"`[^`]*`", "", hit)):
+        assert "探针词" in hit_out
+    miss_out = _both(replace_natural_language_term, monkeypatch, miss, raw, "探针词")
+    assert miss_out == miss
+    quoted_out = _both(replace_natural_language_term, monkeypatch, span, raw, "探针词")
+    assert quoted_out == span
+
+
+def test_fast_hash_matches_the_legacy_algorithm_on_a_real_packet(day_views, monkeypatch) -> None:
+    packet, _render_packet, _zh, _en = day_views
+    scope = packet["result_binding"]["binding_scope"]
+    input_hash = packet["calculation_profile"]["input_hash"]
+    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "1")
+    assert hash_bound_result(packet, input_hash, scope) == _legacy_hash(packet, input_hash, scope)
+    assert hash_bound_result(packet, input_hash, scope) == packet["result_hash"]
+    variant = copy.deepcopy(packet)
+    variant["probe_layer"] = {
+        "elapsed_seconds": 1.5,
+        "called_at": "2001-01-01T00:00:00Z",
+        "cache_created_at": "created",
+        "cache_expires_at": "expires",
+        "nested": {"elapsed_seconds": 9, (1, 2): {"called_at": "inside"}},
+    }
+    summary = variant.setdefault("ai_and_audit", {}).setdefault("summary", {})
+    summary["elapsed_seconds"] = 99
+    summary["kept"] = "yes"
+    assert hash_bound_result(variant, input_hash, scope) == _legacy_hash(variant, input_hash, scope)
+    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "0")
+    assert hash_bound_result(packet, input_hash) == packet["result_hash"]
+
+
+class _CountingHandler:
+    def __init__(self, birth: dict, packet: dict) -> None:
+        self.birth = birth
+        self.packet = packet
+        self.calls = 0
+        self.fail_once = False
+        self.delay = 0.0
+        self.lock = threading.Lock()
+
+    def _high_rigor_birth_payload(self, body: dict) -> dict:
+        payload = dict(self.birth)
+        if body.get("today"):
+            payload["today"] = body["today"]
+        payload["candidate_range"] = body.get("candidate_range")
+        return payload
+
+    def _compute_full_reading_for_thematic(self, _birth: dict) -> dict:
+        with self.lock:
+            self.calls += 1
+            call = self.calls
+        if self.delay:
+            time.sleep(self.delay)
+        if self.fail_once and call == 1:
+            raise RuntimeError("report_build_failed")
+        return {"stub": True}
+
+
+def _body(**extra) -> dict:
+    body = {
+        "format": "markdown",
+        "edition": "full_data",
+        "include_fact_tables": True,
+        "languages": ["zh", "en"],
+        "today": "2026-04-07",
+    }
+    body.update(extra)
+    return body
+
+
+@pytest.fixture
+def cached_report(monkeypatch, tmp_path, day_views):
+    import scripts.consult_full_data_cache as cache
+    import scripts.professional_report_reference as reference
+
+    def _dir(name: str = "consult_full_data_cache"):
+        path = tmp_path / name
+        path.mkdir(parents=True, exist_ok=True)
+        return path
+
+    monkeypatch.setattr(cache, "cache_dir", _dir)
+    monkeypatch.setattr(reference, "render_full_data_markdown", lambda packet: "en-body" if packet.get("report_language") == "en" else "zh-body")
+    monkeypatch.setattr(reference, "report_density_packet", lambda _packet: {"facts": 1})
+    monkeypatch.setattr(reference, "render_dasha_applicability", lambda _packet, language="zh": f"app-{language}")
+    monkeypatch.setattr(reference, "clean_reader_appendix_markdown", lambda text: text)
+    packet = day_views[1]
+    birth = {key: value for key, value in CASES["day"].items() if key != "name"}
+    handler = _CountingHandler(birth, packet)
+
+    class _Engine:
+        def build_professional_report_reference_packet(self, _reading, _args, _packs):
+            return copy.deepcopy(packet)
+
+        def render_pl9_markdown(self, _packet):
+            return "reference"
+
+    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "1")
+    monkeypatch.setenv("JYOTISH_REPORT_RENDER_FAST", "0")
+    monkeypatch.setenv("GITHUB_SHA", "report-cache-test")
+    return handler, _Engine(), _dir("report_reference_cache")
+
+
+def _dumps(result: dict) -> str:
+    return json.dumps(result, ensure_ascii=False, default=str)
+
+
+def test_second_call_reuses_the_cached_response(cached_report) -> None:
+    handler, engine, folder = cached_report
+    first = build_professional_report_reference(handler, _body(), engine=engine)
+    second = build_professional_report_reference(handler, _body(), engine=engine)
+    assert handler.calls == 1
+    assert _dumps(first) == _dumps(second)
+    assert list(folder.glob("*.json"))
+
+
+@pytest.mark.parametrize("extra", [
+    {"today": "2026-05-01"},
+    {"languages": ["zh"]},
+    {"include_fact_tables": False},
+    {"candidate_range": {"kind": "probe"}},
+])
+def test_changed_inputs_miss_the_report_cache(cached_report, extra) -> None:
+    handler, engine, _folder = cached_report
+    build_professional_report_reference(handler, _body(), engine=engine)
+    build_professional_report_reference(handler, _body(**extra), engine=engine)
+    assert handler.calls == 2
+
+
+def test_github_sha_is_part_of_the_report_cache_key(cached_report, monkeypatch) -> None:
+    handler, engine, _folder = cached_report
+    build_professional_report_reference(handler, _body(), engine=engine)
+    monkeypatch.setenv("GITHUB_SHA", "report-cache-other")
+    build_professional_report_reference(handler, _body(), engine=engine)
+    assert handler.calls == 2
+
+
+def test_json_format_and_disabled_cache_write_nothing(cached_report, monkeypatch) -> None:
+    handler, engine, folder = cached_report
+    for path in folder.glob("*.json"):
+        path.unlink()
+    build_professional_report_reference(
+        handler,
+        {"format": "json", "edition": "full_data", "today": "2026-04-07"},
+        engine=engine,
+    )
+    assert list(folder.glob("*.json")) == []
+    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "0")
+    build_professional_report_reference(handler, _body(), engine=engine)
+    assert list(folder.glob("*.json")) == []
+    monkeypatch.setenv("JYOTISH_REPORT_REFERENCE_CACHE", "1")
+    monkeypatch.setenv("GITHUB_SHA", "unknown")
+    build_professional_report_reference(handler, _body(), engine=engine)
+    assert list(folder.glob("*.json")) == []
+
+
+def test_concurrent_same_key_builds_once(cached_report) -> None:
+    handler, engine, _folder = cached_report
+    handler.delay = 0.2
+    barrier = threading.Barrier(2)
+    errors = []
+
+    def run() -> None:
+        try:
+            barrier.wait()
+            build_professional_report_reference(handler, _body(), engine=engine)
+        except Exception as exc:  # pragma: no cover - recorded for the assertion
+            errors.append(exc)
+
+    threads = [threading.Thread(target=run) for _ in range(2)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    assert errors == []
+    assert handler.calls == 1
+
+
+def test_failed_build_is_not_cached(cached_report) -> None:
+    handler, engine, folder = cached_report
+    handler.fail_once = True
+    with pytest.raises(RuntimeError):
+        build_professional_report_reference(handler, _body(), engine=engine)
+    assert list(folder.glob("*.json")) == []
+    build_professional_report_reference(handler, _body(), engine=engine)
+    assert handler.calls == 2
+    build_professional_report_reference(handler, _body(), engine=engine)
+    assert handler.calls == 2
+
+
+def test_report_cache_directory_prunes_old_files(cached_report) -> None:
+    handler, engine, folder = cached_report
+    stale = folder / "stale-probe.json"
+    stale.write_text("{}", encoding="utf-8")
+    old = time.time() - (3 * 24 * 60 * 60)
+    os.utime(stale, (old, old))
+    build_professional_report_reference(handler, _body(), engine=engine)
+    assert not stale.exists()

tests/test_vedastro_service_adapter_executor.py

@@ -769,6 +769,10 @@ def test_vedastro_range_scan_records_hashes_and_artifact_path() -> None:
         env = os.environ.copy()
         env["VEDASTRO_API_ENDPOINT"] = f"http://127.0.0.1:{server.server_port}/vedastro"
         env["VEDASTRO_ENABLE_NETWORK"] = "1"
+        # 原值：沿用进程环境，没有 VEDASTRO_WRITE_AUDIT_ARTIFACTS。
+        # 新值：显式设为 1。
+        # 原因：审计落盘默认关闭后，这条测试仍要断言文件存在，所以打开开关。这是测试前置，不是放宽断言。
+        env["VEDASTRO_WRITE_AUDIT_ARTIFACTS"] = "1"
         completed = subprocess.run(
             [
                 sys.executable,