fix(consultation): 外网证据按盘+日期缓存，超时取消前台任务

BUG-727：同日 VedAstro 快照零等待，跨日先用旧的并后台刷新；join 超时必须 cancel，budget 不超过 2×join。BUG-728：western_evidence_packet 无读取点，默认不再进咨询响应。jyotish_api_server.py 未增长（11334→11291）。

scripts/jyotish_api_server.py

@@ -52,6 +52,18 @@ try:
         MIN_SCORING_EVENTS,
     )
     from scripts.vedastro_runtime_context import temporary_timeout_seconds
+    from scripts.vedastro_foreground import (
+        _FOREGROUND_VEDASTRO_EXECUTOR,
+        _FOREGROUND_VEDASTRO_WORKERS,
+        _blocked_foreground_vedastro,
+        _foreground_vedastro_budget_seconds,
+        _foreground_vedastro_join_seconds,
+        _join_foreground_vedastro,
+        _run_foreground_vedastro_gateway,
+        finish_foreground_vedastro,
+        should_include_western_evidence_packet,
+        start_foreground_vedastro,
+    )
 except ModuleNotFoundError:  # pragma: no cover - script execution path
     from local_env import load_local_env
     from rectification_policy import (
@@ -63,6 +75,18 @@ except ModuleNotFoundError:  # pragma: no cover - script execution path
         MIN_SCORING_EVENTS,
     )
     from vedastro_runtime_context import temporary_timeout_seconds
+    from vedastro_foreground import (
+        _FOREGROUND_VEDASTRO_EXECUTOR,
+        _FOREGROUND_VEDASTRO_WORKERS,
+        _blocked_foreground_vedastro,
+        _foreground_vedastro_budget_seconds,
+        _foreground_vedastro_join_seconds,
+        _join_foreground_vedastro,
+        _run_foreground_vedastro_gateway,
+        finish_foreground_vedastro,
+        should_include_western_evidence_packet,
+        start_foreground_vedastro,
+    )
 try:
     from scripts.unified_consultation_orchestrator import FORMAL_DIVISIONS, UnifiedConsultationOrchestrator
 except ModuleNotFoundError:  # pragma: no cover - script execution path
@@ -122,11 +146,6 @@ _ASYNC_JOB_EXECUTOR = ThreadPoolExecutor(
     max_workers=_ASYNC_JOB_WORKERS,
     thread_name_prefix='jyotish-job',
 )
-_FOREGROUND_VEDASTRO_WORKERS = max(int(os.environ.get('JYOTISH_FOREGROUND_VEDASTRO_WORKERS', '2')), 1)
-_FOREGROUND_VEDASTRO_EXECUTOR = ThreadPoolExecutor(
-    max_workers=_FOREGROUND_VEDASTRO_WORKERS,
-    thread_name_prefix='jyotish-vedastro-fg',
-)
 _ASYNC_JOB_CAPACITY = threading.BoundedSemaphore(_ASYNC_JOB_WORKERS + _ASYNC_JOB_QUEUE_SIZE)
 _RATE_LIMIT_LOCK = threading.Lock()
 _RATE_LIMIT_BUCKETS: dict[str, tuple[float, int]] = {}
@@ -442,6 +461,8 @@ def _evaluate_vedastro_minute_sensitive_pair(request, times: list[str]) -> dict:
         try:
             reports.append(future.result(timeout=remaining))
         except FuturesTimeoutError:
+            for pending in futures:
+                pending.cancel()
             return _unevaluated_vedastro_minute_sensitive('vedastro_minute_snapshot_timeout')
         except Exception:
             return _unevaluated_vedastro_minute_sensitive('vedastro_minute_snapshot_error')
@@ -2036,22 +2057,6 @@ _VEDASTRO_COMPACT_DENY_KEYS = {
 }
 
 
-def _foreground_vedastro_budget_seconds() -> float:
-    raw = str(os.environ.get('JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS', '8')).strip()
-    try:
-        return min(max(float(raw), 2.0), 12.0)
-    except ValueError:
-        return 8.0
-
-
-def _foreground_vedastro_join_seconds() -> float:
-    raw = str(os.environ.get('JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS', '1.5')).strip()
-    try:
-        return min(max(float(raw), 0.0), 3.0)
-    except ValueError:
-        return 1.5
-
-
 def _vedastro_natal_signs(raw) -> dict | None:
     if not isinstance(raw, dict):
         return None
@@ -2107,45 +2112,6 @@ def _compact_vedastro_cross_check(vedastro_official: dict, vedastro_gateway: dic
     return compact
 
 
-def _blocked_foreground_vedastro(*, reason: str, error_type: str | None = None) -> dict:
-    packet = {
-        'scope': 'vedastro_gateway_run',
-        'status': 'official_blocked',
-        'official_closure_state': 'official_blocked',
-        'official_closure_reason': reason,
-    }
-    if error_type:
-        packet['error_type'] = error_type
-    return packet
-
-
-def _run_foreground_vedastro_gateway(handler, body: dict) -> dict:
-    try:
-        with temporary_timeout_seconds(_foreground_vedastro_budget_seconds()):
-            result = handler._compute_vedastro_gateway_run(body)
-    except Exception as exc:
-        return _blocked_foreground_vedastro(
-            reason='gateway_invocation_error',
-            error_type=type(exc).__name__,
-        )
-    return result if isinstance(result, dict) else _blocked_foreground_vedastro(reason='gateway_invocation_error')
-
-
-def _join_foreground_vedastro(future, *, timeout: float) -> dict:
-    if future is None:
-        return _blocked_foreground_vedastro(reason='foreground_optional_evidence_timeout')
-    try:
-        result = future.result(timeout=timeout)
-    except FuturesTimeoutError:
-        return _blocked_foreground_vedastro(reason='foreground_optional_evidence_timeout')
-    except Exception as exc:
-        return _blocked_foreground_vedastro(
-            reason='gateway_invocation_error',
-            error_type=type(exc).__name__,
-        )
-    return result if isinstance(result, dict) else _blocked_foreground_vedastro(reason='gateway_invocation_error')
-
-
 def _blocked_birth_time_sensitivity(*, error_type: str) -> dict:
     return {
         'schema': 'jyotish.report_birth_time_sensitivity.v1',
@@ -2266,7 +2232,7 @@ def execute_consultation_workflow(
             western_evidence_packet=western_evidence_packet,
             blind=bool(body.get('blind') or body.get('blind_technical_mode')),
         )
-        if western_evidence_packet:
+        if western_evidence_packet and should_include_western_evidence_packet(body, surface=surface):
             result['western_evidence_packet'] = western_evidence_packet
         if body.get('return_high_rigor_shape'):
             result['endpoint'] = 'high_rigor_workflow'
@@ -2288,13 +2254,14 @@ def execute_consultation_workflow(
     rectification = {}
     muhurta_panchanga = {}
     computed_chart = bool(chart)
-    foreground_vedastro_future = None
-    if defer_optional_external_evidence:
-        # Overlap a bounded official gateway with local compute. Main-entry
-        # overview / range scan stay skipped so foreground cannot replay BUG-161.
-        foreground_vedastro_future = _FOREGROUND_VEDASTRO_EXECUTOR.submit(
-            _run_foreground_vedastro_gateway, handler, dict(body),
-        )
+    # Overlap a bounded official gateway with local compute. Main-entry
+    # overview / range scan stay skipped so foreground cannot replay BUG-161.
+    # Same-day snapshot cache skips the submit entirely (BUG-727).
+    vedastro_session = start_foreground_vedastro(
+        handler,
+        dict(body),
+        defer_optional_external_evidence=defer_optional_external_evidence,
+    )
 
     for step in runtime_planner.get('sync_steps', []):
         if step == 'run_prashna':
@@ -2375,21 +2342,8 @@ def execute_consultation_workflow(
         })
         executed_steps.append('run_thematic_report')
     vedastro_gateway = rectification.get('vedastro_gateway') if isinstance(rectification, dict) else None
-    if defer_optional_external_evidence:
-        vedastro_gateway = _join_foreground_vedastro(
-            foreground_vedastro_future,
-            timeout=_foreground_vedastro_join_seconds(),
-        )
-    elif not isinstance(vedastro_gateway, dict):
-        try:
-            vedastro_gateway = handler._compute_vedastro_gateway_run(body)
-        except Exception as exc:  # Gateway evidence must not block the local chart result.
-            vedastro_gateway = {
-                'scope': 'vedastro_gateway_run',
-                'status': 'official_blocked',
-                'official_closure_reason': 'gateway_invocation_error',
-                'error_type': type(exc).__name__,
-            }
+    if defer_optional_external_evidence or not isinstance(vedastro_gateway, dict):
+        vedastro_gateway = finish_foreground_vedastro(vedastro_session)
 
     vedastro_official = handler._high_rigor_vedastro_official_summary(chart)
     gateway_raw = (
@@ -2524,7 +2478,6 @@ def execute_consultation_workflow(
         'interpretation_source_runtime_coverage': interpretation_source_runtime_coverage,
         'machine_evidence_packet': machine_evidence_packet,
         'consumer_context': consumer_context,
-        'western_evidence_packet': western_evidence_packet or {},
         'real_case_calibration': real_case_calibration,
         'birth_time_sensitivity': birth_time_sensitivity,
         'runtime_evidence_log': runtime_evidence_log,
@@ -2535,6 +2488,10 @@ def execute_consultation_workflow(
             'domain-relevant routes execute according to the configured sample/network limits.'
         ),
     }
+    if should_include_western_evidence_packet(body, surface=surface):
+        result['western_evidence_packet'] = western_evidence_packet or {}
+    if getattr(vedastro_session, 'meta', None):
+        result['vedastro_snapshot_cache'] = vedastro_session.meta
     if high_rigor:
         result['high_rigor_external_parity'] = {
             'status': 'pass' if external_parity_gate.get('status') == 'pass' else 'blocked',

scripts/run_quality_gate.py

@@ -75,6 +75,8 @@ CORE_PYTEST_TARGETS = [
     "tests/test_birth_time_journey_contract.py",
     # Freeze scripts/jyotish_api_server.py growth; new features must be modules.
     "tests/test_api_server_growth_contract.py",
+    # Foreground VedAstro snapshot cache + join cancel (BUG-727 / BUG-728).
+    "tests/test_vedastro_snapshot_cache.py",
     # Native seven-governors adapter and the three read-only chart endpoints.
     "tests/test_qizheng_chart_engine.py",
     "tests/test_qizheng_api_productization.py",

scripts/vedastro_foreground.py

@@ -0,0 +1,250 @@
+"""Foreground VedAstro wait, cancel, and snapshot-cache coordination.
+
+Join, budget, and worker count bound the same wall clock, so they are
+declared together and must be changed together.
+
+join: how long the consultation thread waits for the official gateway
+      (default 1.5s, cap 3s).
+budget: how long a worker may keep running after it has started.
+workers: process-wide pool. Extra work queues; join still caps the wait.
+
+A timed-out join must cancel its future so queued work never starts.
+budget <= k * join (k=2) keeps leftover occupancy of an already-running
+worker from stretching later requests past the join cap. The previous
+default budget of 8s with a 1.5s join left cancelled-too-late workers
+holding the two-thread pool for the rest of the 8s.
+"""
+from __future__ import annotations
+
+import os
+import threading
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
+from typing import Any
+
+try:
+    from scripts.vedastro_runtime_context import temporary_timeout_seconds
+    from scripts.vedastro_snapshot_cache import (
+        annotate_stale_gateway,
+        is_cacheable_gateway,
+        lookup_snapshot,
+        official_snapshot_reference_date,
+        requires_today_snapshot,
+        store_snapshot,
+    )
+except ModuleNotFoundError:  # pragma: no cover - script execution path
+    from vedastro_runtime_context import temporary_timeout_seconds
+    from vedastro_snapshot_cache import (
+        annotate_stale_gateway,
+        is_cacheable_gateway,
+        lookup_snapshot,
+        official_snapshot_reference_date,
+        requires_today_snapshot,
+        store_snapshot,
+    )
+
+# --- join / budget / workers (one declaration block; change together) ---
+FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT = 1.5
+FOREGROUND_VEDASTRO_JOIN_SECONDS_MAX = 3.0
+FOREGROUND_VEDASTRO_BUDGET_JOIN_RATIO = 2.0
+FOREGROUND_VEDASTRO_WORKERS_DEFAULT = 2
+# ----------------------------------------------------------------------
+
+_FOREGROUND_VEDASTRO_WORKERS = max(
+    int(os.environ.get("JYOTISH_FOREGROUND_VEDASTRO_WORKERS", str(FOREGROUND_VEDASTRO_WORKERS_DEFAULT))),
+    1,
+)
+_FOREGROUND_VEDASTRO_EXECUTOR = ThreadPoolExecutor(
+    max_workers=_FOREGROUND_VEDASTRO_WORKERS,
+    thread_name_prefix="jyotish-vedastro-fg",
+)
+
+
+def _foreground_vedastro_join_seconds() -> float:
+    raw = str(os.environ.get("JYOTISH_FOREGROUND_VEDASTRO_JOIN_SECONDS", str(FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT))).strip()
+    try:
+        return min(max(float(raw), 0.0), FOREGROUND_VEDASTRO_JOIN_SECONDS_MAX)
+    except ValueError:
+        return FOREGROUND_VEDASTRO_JOIN_SECONDS_DEFAULT
+
+
+def _foreground_vedastro_budget_seconds() -> float:
+    join = _foreground_vedastro_join_seconds()
+    cap = FOREGROUND_VEDASTRO_BUDGET_JOIN_RATIO * join
+    raw = str(os.environ.get("JYOTISH_FOREGROUND_VEDASTRO_BUDGET_SECONDS", "")).strip()
+    if not raw:
+        return cap
+    try:
+        budget = float(raw)
+    except ValueError:
+        return cap
+    if budget < 0:
+        return cap
+    return min(budget, cap)
+
+
+def _blocked_foreground_vedastro(*, reason: str, error_type: str | None = None) -> dict:
+    packet = {
+        "scope": "vedastro_gateway_run",
+        "status": "official_blocked",
+        "official_closure_state": "official_blocked",
+        "official_closure_reason": reason,
+    }
+    if error_type:
+        packet["error_type"] = error_type
+    return packet
+
+
+def _run_foreground_vedastro_gateway(handler, body: dict, cancel_event: threading.Event | None = None) -> dict:
+    if cancel_event is not None and cancel_event.is_set():
+        return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout")
+    try:
+        with temporary_timeout_seconds(_foreground_vedastro_budget_seconds()):
+            result = handler._compute_vedastro_gateway_run(body)
+    except Exception as exc:
+        return _blocked_foreground_vedastro(
+            reason="gateway_invocation_error",
+            error_type=type(exc).__name__,
+        )
+    return result if isinstance(result, dict) else _blocked_foreground_vedastro(reason="gateway_invocation_error")
+
+
+def _run_foreground_vedastro_gateway_and_store(
+    handler,
+    body: dict,
+    cancel_event: threading.Event | None = None,
+) -> dict:
+    result = _run_foreground_vedastro_gateway(handler, body, cancel_event)
+    if is_cacheable_gateway(result):
+        try:
+            store_snapshot(body, result)
+        except OSError:
+            pass
+    return result
+
+
+def _join_foreground_vedastro(future, *, timeout: float, cancel_event: threading.Event | None = None) -> dict:
+    if future is None:
+        return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout")
+    try:
+        result = future.result(timeout=timeout)
+    except FuturesTimeoutError:
+        if cancel_event is not None:
+            cancel_event.set()
+        future.cancel()
+        return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout")
+    except Exception as exc:
+        return _blocked_foreground_vedastro(
+            reason="gateway_invocation_error",
+            error_type=type(exc).__name__,
+        )
+    return result if isinstance(result, dict) else _blocked_foreground_vedastro(reason="gateway_invocation_error")
+
+
+def should_include_western_evidence_packet(body: dict | None, *, surface: str = "") -> bool:
+    payload = body if isinstance(body, dict) else {}
+    if payload.get("include_western_evidence_packet") or payload.get("return_western_evidence_packet"):
+        return True
+    if payload.get("return_high_rigor_shape"):
+        return True
+    if isinstance(payload.get("western_oracle_payload"), dict) or isinstance(payload.get("western_astrology_oracle"), dict):
+        return True
+    if isinstance(payload.get("western_evidence_packet"), dict):
+        return True
+    return str(surface or "") in {"skill_mcp", "professional_reading_web", "mcp"}
+
+
+class ForegroundVedastroSession:
+    def __init__(self, handler, body: dict, *, defer: bool) -> None:
+        self.handler = handler
+        self.body = dict(body)
+        self.defer = bool(defer)
+        self.cached_gateway: dict | None = None
+        self.future = None
+        self.cancel_event = threading.Event()
+        self.meta: dict[str, Any] = {}
+
+    def start(self) -> "ForegroundVedastroSession":
+        today = official_snapshot_reference_date(self.body)
+        hit = lookup_snapshot(self.body, today=today)
+        if hit is not None:
+            record = hit["record"]
+            freshness = hit["freshness"]
+            gateway = record["gateway"]
+            if freshness == "stale":
+                gateway = annotate_stale_gateway(
+                    gateway,
+                    reference_date=str(record.get("reference_date") or ""),
+                    served_on_utc_date=today,
+                )
+                refresh_body = dict(self.body)
+                refresh_body["reference_date"] = today
+                refresh_body["today"] = today
+                refresh_body["current_date"] = today
+                self.future = _FOREGROUND_VEDASTRO_EXECUTOR.submit(
+                    _run_foreground_vedastro_gateway_and_store,
+                    self.handler,
+                    refresh_body,
+                    None,
+                )
+            self.cached_gateway = gateway
+            self.meta = {
+                "freshness": freshness,
+                "reference_date": str(record.get("reference_date") or today),
+                "served_on_utc_date": today,
+                "refresh_submitted": freshness == "stale",
+            }
+            return self
+        self.meta = {
+            "freshness": "miss",
+            "reference_date": today,
+            "served_on_utc_date": today,
+            "refresh_submitted": False,
+            "require_today": requires_today_snapshot(self.body),
+        }
+        if self.defer:
+            self.future = _FOREGROUND_VEDASTRO_EXECUTOR.submit(
+                _run_foreground_vedastro_gateway_and_store,
+                self.handler,
+                dict(self.body),
+                self.cancel_event,
+            )
+        return self
+
+    def finish(self) -> dict:
+        if self.cached_gateway is not None:
+            return self.cached_gateway
+        if self.defer:
+            return _join_foreground_vedastro(
+                self.future,
+                timeout=_foreground_vedastro_join_seconds(),
+                cancel_event=self.cancel_event,
+            )
+        try:
+            result = self.handler._compute_vedastro_gateway_run(self.body)
+        except Exception as exc:
+            return _blocked_foreground_vedastro(
+                reason="gateway_invocation_error",
+                error_type=type(exc).__name__,
+            )
+        if not isinstance(result, dict):
+            return _blocked_foreground_vedastro(reason="gateway_invocation_error")
+        if is_cacheable_gateway(result):
+            try:
+                store_snapshot(self.body, result)
+            except OSError:
+                pass
+        return result
+
+
+def start_foreground_vedastro(handler, body: dict, *, defer_optional_external_evidence: bool) -> ForegroundVedastroSession:
+    return ForegroundVedastroSession(
+        handler,
+        body,
+        defer=defer_optional_external_evidence,
+    ).start()
+
+
+def finish_foreground_vedastro(session: ForegroundVedastroSession | None) -> dict:
+    if session is None:
+        return _blocked_foreground_vedastro(reason="foreground_optional_evidence_timeout")
+    return session.finish()

scripts/vedastro_snapshot_cache.py

@@ -0,0 +1,228 @@
+"""Disk cache for foreground VedAstro gateway snapshots.
+
+Keyed by birth data + ayanamsa + node + UTC reference date. This is a
+separate store from ``_api_chart_cache`` (BUG-161): different directory
+and a different key function. Freshness is the reference date itself
+plus a 7-day stale window — there is no independent TTL environment
+variable.
+
+Filenames are sha256 hex only. Cache files must not contain names,
+emails, or user ids.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import tempfile
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+CACHE_SCHEMA = "vedastro_snapshot_cache.v1"
+MAX_STALE_DAYS = 7
+_SHA256_NAME = re.compile(r"^[0-9a-f]{64}\.json$")
+_IDENTITY_KEYS = {
+    "name",
+    "email",
+    "user_id",
+    "userid",
+    "user_email",
+    "session_id",
+    "sessionid",
+    "full_name",
+    "display_name",
+}
+
+
+def snapshot_cache_dir() -> Path:
+    raw = str(os.environ.get("JYOTISH_VEDASTRO_SNAPSHOT_CACHE_DIR") or "").strip()
+    path = Path(raw) if raw else ROOT / "scratch" / "local" / "vedastro_snapshot_cache"
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def official_snapshot_reference_date(case: dict[str, Any] | None) -> str:
+    """Same date rule as ``vedastro_service_adapter._official_snapshot_reference_date``.
+
+    Duplicated so this module does not import the adapter (and its optional
+    network stack) at cache-lookup time.
+    """
+    payload = case if isinstance(case, dict) else {}
+    for key in ("reference_date", "today", "transit_date", "current_date"):
+        value = payload.get(key)
+        if not value:
+            continue
+        raw = str(value)[:10]
+        try:
+            datetime.strptime(raw, "%Y-%m-%d")
+            return raw
+        except ValueError:
+            continue
+    return datetime.utcnow().strftime("%Y-%m-%d")
+
+
+def requires_today_snapshot(body: dict[str, Any] | None) -> bool:
+    payload = body if isinstance(body, dict) else {}
+    entrypoint = str(payload.get("entrypoint") or payload.get("consult_entrypoint") or "").strip()
+    return entrypoint == "daily_starlanguage"
+
+
+def snapshot_cache_key(body: dict[str, Any] | None, *, reference_date: str) -> str:
+    payload = body if isinstance(body, dict) else {}
+    ayanamsa = (
+        payload.get("ayanamsa")
+        or payload.get("ayanamsa_name")
+        or payload.get("ayanamsa_policy")
+        or "raman"
+    )
+    node = payload.get("node_mode") or payload.get("nodeMode") or "mean"
+    material = {
+        "year": payload.get("year"),
+        "month": payload.get("month"),
+        "day": payload.get("day"),
+        "hour": payload.get("hour"),
+        "minute": payload.get("minute"),
+        "second": payload.get("second", 0),
+        "lat": payload.get("lat"),
+        "lon": payload.get("lon"),
+        "tz": payload.get("tz"),
+        "ayanamsa_policy": str(ayanamsa).strip().lower(),
+        "node_policy": str(node).strip().lower(),
+        "reference_date": str(reference_date)[:10],
+    }
+    canonical = json.dumps(material, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+def snapshot_cache_path(cache_key: str) -> Path:
+    if not re.fullmatch(r"[0-9a-f]{64}", cache_key):
+        raise ValueError("vedastro snapshot cache key must be sha256 hex")
+    return snapshot_cache_dir() / f"{cache_key}.json"
+
+
+def is_cacheable_gateway(packet: Any) -> bool:
+    if not isinstance(packet, dict):
+        return False
+    state = packet.get("official_closure_state") or packet.get("status")
+    if state != "official_verified":
+        return False
+    raw = packet.get("official_raw_response") or packet.get("raw_response")
+    return bool(raw)
+
+
+def _strip_identity(value: Any) -> Any:
+    if isinstance(value, dict):
+        cleaned = {}
+        for key, item in value.items():
+            if str(key).strip().lower() in _IDENTITY_KEYS:
+                continue
+            cleaned[key] = _strip_identity(item)
+        return cleaned
+    if isinstance(value, list):
+        return [_strip_identity(item) for item in value]
+    return value
+
+
+def _load_record(cache_key: str) -> dict[str, Any] | None:
+    path = snapshot_cache_path(cache_key)
+    if not _SHA256_NAME.match(path.name) or not path.is_file():
+        return None
+    try:
+        record = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return None
+    if not isinstance(record, dict) or record.get("schema") != CACHE_SCHEMA:
+        return None
+    gateway = record.get("gateway")
+    if not isinstance(gateway, dict):
+        return None
+    reference_date = str(record.get("reference_date") or "")[:10]
+    try:
+        datetime.strptime(reference_date, "%Y-%m-%d")
+    except ValueError:
+        return None
+    return record
+
+
+def lookup_snapshot(
+    body: dict[str, Any] | None,
+    *,
+    today: str | None = None,
+    require_today: bool | None = None,
+) -> dict[str, Any] | None:
+    payload = body if isinstance(body, dict) else {}
+    served = today or official_snapshot_reference_date(payload)
+    must_be_today = requires_today_snapshot(payload) if require_today is None else bool(require_today)
+    today_key = snapshot_cache_key(payload, reference_date=served)
+    record = _load_record(today_key)
+    if record is not None:
+        return {
+            "freshness": "fresh",
+            "record": record,
+            "served_on_utc_date": served,
+        }
+    if must_be_today:
+        return None
+    current = datetime.strptime(served, "%Y-%m-%d").date()
+    for days in range(1, MAX_STALE_DAYS + 1):
+        past = (current - timedelta(days=days)).isoformat()
+        record = _load_record(snapshot_cache_key(payload, reference_date=past))
+        if record is None:
+            continue
+        return {
+            "freshness": "stale",
+            "record": record,
+            "served_on_utc_date": served,
+        }
+    return None
+
+
+def store_snapshot(
+    body: dict[str, Any] | None,
+    gateway: dict[str, Any],
+    *,
+    reference_date: str | None = None,
+) -> dict[str, Any] | None:
+    if not is_cacheable_gateway(gateway):
+        return None
+    payload = body if isinstance(body, dict) else {}
+    stored_date = (reference_date or official_snapshot_reference_date(payload))[:10]
+    cache_key = snapshot_cache_key(payload, reference_date=stored_date)
+    record = {
+        "schema": CACHE_SCHEMA,
+        "cache_key": cache_key,
+        "reference_date": stored_date,
+        "stored_at": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
+        "gateway": _strip_identity(gateway),
+    }
+    path = snapshot_cache_path(cache_key)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    encoded = json.dumps(record, ensure_ascii=False, sort_keys=True)
+    fd, tmp_name = tempfile.mkstemp(prefix=f"{cache_key}.", suffix=".tmp", dir=str(path.parent))
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            handle.write(encoded)
+        os.replace(tmp_name, path)
+    except Exception:
+        try:
+            os.unlink(tmp_name)
+        except OSError:
+            pass
+        raise
+    return record
+
+
+def annotate_stale_gateway(gateway: dict[str, Any], *, reference_date: str, served_on_utc_date: str) -> dict[str, Any]:
+    """Shallow copy that shows which day a stale snapshot is from.
+
+    Does not change ``official_closure_state`` / ``status`` — a previously
+    delivered official layer stays whatever it was (BUG-301).
+    """
+    packet = dict(gateway)
+    packet["snapshot_reference_date"] = reference_date
+    packet["snapshot_freshness"] = "stale"
+    packet["snapshot_served_on_utc_date"] = served_on_utc_date
+    return packet