快照新鲜度守卫：旧快照不许当今天的计划下发

实测两条路比对：代码序列一模一样，但候选卡名次、判决、依据有实质差异。根子在节奏——
出计划是前一晚 23:10 跑的，数据基座凌晨才更新证据线，快照天生比实时装配旧半天。

两条判据：代码版本要与当前一致、生成时刻要晚于今天早六点。都不满足就退回实时装配，
也就是今天的行为。守卫要松开靠的是把出计划挪到数据基座之后，不是把界调低。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

plan.py

@@ -855,6 +855,50 @@ def render_md(d: dict) -> str:
     return "\n".join(L)
 
 
+class StaleSnapshot(RuntimeError):
+    """快照还在，但太旧、不能拿来当今天的计划用。调用方接住它回落到实时装配。"""
+
+
+def snapshot_fresh_enough(snap: dict) -> tuple:
+    """这份快照够不够新，能不能当今天的计划直接下发。返回 (能不能, 一句原因)。
+
+    为什么需要这道守卫（2026-09-10，台账 057）: 出计划这一步现在是**前一晚**跑的
+    （实测 09-09 23:10 出的 09-09 那份），而数据基座凌晨才更新公司深度评析、催化事件
+    这些证据线。所以快照天生比实时装配旧半天 —— 实测两条路比对，代码序列一模一样，
+    但候选卡名次、判决、依据这些字段有实质差异。直接拿旧快照下发，等于让下游用昨晚的判决。
+
+    两条判据，都不满足就退回实时装配（也就是今天的行为，不会变坏）:
+      · 代码版本要和现在跑的一致 —— 版本变了说明判决逻辑可能变了，昨天算的不算数
+      · 生成时刻要晚于今天的界（默认早上六点）—— 保证它是在数据基座凌晨那批任务之后出的
+
+    **这道守卫要能松开，靠的是把出计划挪到数据基座之后**（比如早上七点十分，代码注释里
+    本来就是这么设计的），不是把界调低。界调低只会让旧快照蒙混过关。
+    """
+    import version
+    min_hour = int(os.environ.get("PLAN_SNAPSHOT_MIN_HOUR", "6"))
+    gen = str(snap.get("generated_at") or "")
+    ver = str(snap.get("plan_version") or "")
+    now_ver = ""
+    try:
+        now_ver = str(version.git_short_rev() or "")
+    except Exception:  # noqa: BLE001 —— 取不到版本就不拿这条卡人
+        now_ver = ""
+    if now_ver and ver and ver != now_ver:
+        return False, (f"快照是用代码版本 {ver} 出的，现在跑的是 {now_ver}；"
+                       f"判决逻辑可能已经变了，本次按实时装配")
+    if not gen:
+        return False, "快照没有生成时刻，判不了新旧"
+    try:
+        g = dt.datetime.fromisoformat(gen)
+    except ValueError:
+        return False, f"快照的生成时刻认不出: {gen}"
+    today_line = dt.datetime.now().replace(hour=min_hour, minute=0, second=0, microsecond=0)
+    if g < today_line:
+        return False, (f"快照出于 {gen}，早于今天 {min_hour:02d}:00；"
+                       f"数据基座凌晨那批证据线更新它没赶上，本次按实时装配")
+    return True, ""
+
+
 def pick_rows(rows, n: int, theme_cap: int) -> list:
     """从**已按分数排好序**的行里取 n 条，每个传导主题最多 theme_cap 条（0=不设限）。
 
@@ -902,6 +946,9 @@ def from_snapshot(date: str | None = None, top: int = 20, obs_top: int = 10,
     full_main, full_obs = snap.get("main"), snap.get("observe")
     if not isinstance(full_main, list) or not isinstance(full_obs, list):
         raise RuntimeError(f"快照缺全量主榜或观察档: {path}")
+    ok, why = snapshot_fresh_enough(snap)
+    if not ok:
+        raise StaleSnapshot(why)
 
     data = {k: v for k, v in snap.items()
             # 前三个是落盘那次的裁剪结果与参数，与本次请求无关；

test_plan_snapshot.py

@@ -113,6 +113,42 @@ def test_from_snapshot(tmpdir=None):
         t("快照不存在要抛异常", isinstance(e, (FileNotFoundError, RuntimeError)), type(e).__name__)
 
 
+def test_freshness_guard():
+    print("[新鲜度守卫 —— 旧快照不许蒙混过关]")
+    import datetime as _dt
+    now = _dt.datetime.now()
+    import version
+    cur = ""
+    try:
+        cur = str(version.git_short_rev() or "")
+    except Exception:  # noqa: BLE001
+        cur = ""
+
+    def snap(gen, ver=None):
+        return {"generated_at": gen, "plan_version": cur if ver is None else ver}
+
+    fresh = now.replace(hour=7, minute=30, second=0, microsecond=0)
+    if now < fresh:                      # 早上七点半之前跑测试时，用「一分钟前」当新鲜样本
+        fresh = now - _dt.timedelta(minutes=1)
+    ok, why = plan.snapshot_fresh_enough(snap(fresh.isoformat(timespec="seconds")))
+    t("今天出的、版本对得上的快照可用", ok, why)
+
+    y = (now - _dt.timedelta(days=1)).replace(hour=23, minute=10)
+    ok2, why2 = plan.snapshot_fresh_enough(snap(y.isoformat(timespec="seconds")))
+    t("昨晚出的快照不可用（数据基座凌晨那批它没赶上）", not ok2 and "早于今天" in why2, why2)
+
+    if cur:
+        ok3, why3 = plan.snapshot_fresh_enough(snap(fresh.isoformat(timespec="seconds"), "deadbee"))
+        t("代码版本对不上就不可用（判决逻辑可能变了）", not ok3 and "代码版本" in why3, why3)
+
+    ok4, _ = plan.snapshot_fresh_enough(snap(""))
+    t("没有生成时刻的快照不可用", not ok4)
+    ok5, _ = plan.snapshot_fresh_enough(snap("不是时间"))
+    t("生成时刻认不出的快照不可用", not ok5)
+
+    t("守卫用的是专门的异常类型，调用方接得住", issubclass(plan.StaleSnapshot, RuntimeError))
+
+
 def test_api_falls_back():
     print("[接口回落：读不到快照不能让接口挂掉]")
     src = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "api.py"),
@@ -131,6 +167,7 @@ def main():
     test_pick_rules()
     test_replay_against_real_snapshot()
     test_from_snapshot()
+    test_freshness_guard()
     test_api_falls_back()
     print("-" * 70)
     if FAILED: