#!/usr/bin/env python3
"""
XKB Pipeline Health Check

檢查五件事：
1. wiki 單一真實來源：兩個路徑是否指向同一位置（symlink 正確）
2. recall 命中的 wiki 來源路徑是否正確
3. recall 端到端是否真的跑得起來（不是只看檔案在不在）
4. recall telemetry 是否還在寫入（太久沒紀錄 = 靜默故障）
5. cron 執行後 summary / vector 是否真的有增量更新

Usage:
    python3 scripts/health_check_pipeline.py
    python3 scripts/health_check_pipeline.py --json
"""

from __future__ import annotations

import argparse
import json
import os
import re
import shutil
import sys
from pathlib import Path
from datetime import datetime, timedelta, timezone

sys.path.insert(0, str(Path(__file__).resolve().parent))
import xkb_paths

WORKSPACE = xkb_paths.WORKSPACE
_SKILL_DIR = xkb_paths.SKILL_DIR
WIKI_DIR = xkb_paths.WIKI_DIR
WIKI_TOPICS_DIR = xkb_paths.WIKI_TOPICS_DIR
BOOKMARKS_DIR = xkb_paths.BOOKMARKS_DIR
INDEX_FILE = xkb_paths.INDEX_FILE
VECTOR_FILE = xkb_paths.VECTOR_FILE
CARDS_DIR = xkb_paths.CARDS_DIR

WORKSPACE_WIKI = WORKSPACE / "wiki"  # 若存在，必須指向 WIKI_DIR

OK = "✅"
WARN = "⚠️ "
FAIL = "❌"


def check_wiki_canonical() -> dict:
    """檢查 workspace/wiki 是否是指向 skill wiki 的 symlink。"""
    result = {"name": "wiki_canonical", "checks": []}

    # 1. skill wiki 是否存在
    if WIKI_DIR.exists():
        result["checks"].append({"ok": True, "msg": f"Skill wiki exists: {WIKI_DIR}"})
    else:
        result["checks"].append({"ok": False, "msg": f"Skill wiki MISSING: {WIKI_DIR}"})
        return result

    # 2. workspace/wiki 是否是 symlink
    if WORKSPACE_WIKI.is_symlink():
        target = Path(os.readlink(WORKSPACE_WIKI))
        if not target.is_absolute():
            target = (WORKSPACE_WIKI.parent / target).resolve()
        canonical = WIKI_DIR.resolve()
        if target.resolve() == canonical:
            result["checks"].append({"ok": True, "msg": f"workspace/wiki → {WIKI_DIR} (symlink correct)"})
        else:
            result["checks"].append({"ok": False, "msg": f"workspace/wiki symlink points to {target}, expected {canonical}"})
    elif WORKSPACE_WIKI.exists():
        # Windows junction / bind mount 不會被 is_symlink() 認出來，但解析後仍是同一個地方
        if WORKSPACE_WIKI.resolve() == WIKI_DIR.resolve():
            result["checks"].append({"ok": True, "msg": f"workspace/wiki → {WIKI_DIR} (link correct)"})
        else:
            result["checks"].append({"ok": False, "msg": "workspace/wiki is a real directory (not symlink) — dual-wiki risk!"})
    else:
        # 這個 symlink 是 VPS 上 OpenClaw 的擺法，不是 XKB 的必要條件。
        # 沒有它就沒有第二份 wiki，也就沒有這一項要防的風險。
        result["checks"].append({"ok": True, "msg": "workspace/wiki absent — no dual-wiki risk"})

    # 3. topic 數量
    topics = list(WIKI_TOPICS_DIR.glob("*.md")) if WIKI_TOPICS_DIR.exists() else []
    result["checks"].append({"ok": len(topics) > 0, "msg": f"Wiki topics: {len(topics)} pages"})

    return result


def check_recall_wiki_source() -> dict:
    """確認 WIKI_TOPICS_DIR 路徑是 skill wiki，而非舊的 workspace wiki。"""
    result = {"name": "recall_wiki_source", "checks": []}

    canonical = WIKI_DIR.resolve()
    actual = WIKI_TOPICS_DIR.parent.resolve()

    if actual == canonical:
        result["checks"].append({"ok": True, "msg": f"Recall reads from canonical wiki: {canonical}"})
    else:
        result["checks"].append({"ok": False, "msg": f"Recall reads from {actual}, expected {canonical}"})

    # 測試一個 sample 查詢能否命中 wiki
    if WIKI_TOPICS_DIR.exists():
        topics = list(WIKI_TOPICS_DIR.glob("*.md"))
        result["checks"].append({"ok": True, "msg": f"Wiki topics accessible: {[t.stem for t in topics]}"})
    else:
        result["checks"].append({"ok": False, "msg": "WIKI_TOPICS_DIR not found — recall wiki layer will return empty"})

    return result


def check_recall_live() -> dict:
    """真的跑一次 recall，而不是只確認檔案在不在。

    2026-05-04 的 script 清理移除了 recall_router 依賴的模組，recall 從此每次都
    崩潰；但當時的檢查只看目錄存不存在，所以整整 12 週沒有人發現。這個檢查是那次
    事故的直接產物：唯一能證明 recall 還活著的方法，就是把它叫起來一次。
    """
    result = {"name": "recall_live", "checks": []}

    sys.path.insert(0, str(Path(__file__).resolve().parent))
    try:
        import xkb_recall_server as server
    except Exception as e:
        result["checks"].append({"ok": False, "msg": f"cannot import xkb_recall_server: {e}"})
        return result

    if not server.ROUTER_SCRIPT.exists():
        result["checks"].append({"ok": False, "msg": f"router missing: {server.ROUTER_SCRIPT}"})
        return result

    probe = server._run_recall_structured("XKB 的知識管線是怎麼設計的")
    status = probe.get("status", "ok")
    if status == "failed":
        result["checks"].append({"ok": False, "msg": f"recall FAILED end-to-end: {probe.get('error')}"})
    else:
        result["checks"].append({
            "ok": True,
            "msg": f"recall ran end-to-end: {len(probe.get('results') or [])} result(s), "
                   f"confidence {probe.get('confidence')}"
        })

    return result


def check_recall_telemetry() -> dict:
    """Dead man's switch：太久沒有任何一筆 recall 紀錄，本身就是故障訊號。"""
    result = {"name": "recall_telemetry", "checks": []}

    stale_days = float(os.getenv("XKB_RECALL_STALE_DAYS", "7"))
    path = WORKSPACE / "memory" / "x-knowledge-base" / "recall-telemetry.jsonl"

    if not path.exists():
        result["checks"].append({"ok": False, "msg": f"recall telemetry not found: {path}"})
        return result

    age_days = (datetime.now().timestamp() - path.stat().st_mtime) / 86400
    result["checks"].append({
        "ok": age_days <= stale_days,
        "msg": f"last recall telemetry: {age_days:.1f} days ago (threshold {stale_days:.0f}d)"
    })

    # 最後一筆是不是失敗的，以及有沒有哪一層一直在退回 fallback。
    try:
        rows = []
        with path.open(encoding="utf-8") as fh:
            for line in fh:
                line = line.strip()
                if line:
                    try:
                        rows.append(json.loads(line))
                    except ValueError:
                        continue
        if rows:
            entry = rows[-1]
            result["checks"].append({
                "ok": True,
                "msg": f"last entry: recalled={entry.get('recalled')}, "
                       f"results={entry.get('result_count')}, ts={entry.get('ts', '')[:19]}"
            })
            result["checks"].extend(_degraded_recall_checks(rows))
    except Exception as e:
        result["checks"].append({"ok": False, "msg": f"telemetry unreadable: {e}"})

    return result


def _degraded_recall_checks(rows: list[dict]) -> list[dict]:
    """有哪一層壞著，而不是「這個主題沒有東西」。

    召回回空手有兩種原因，而它們原本在紀錄裡完全相同：recalled=false,
    result_count=0。2026-05-04 之後召回每次都崩潰，整整十二週沒有人發現——
    它一直有禮貌地回答「我不知道」。recall_router 現在會把這一次退回 fallback
    的層記進 degraded 欄位，這裡讀它。

    判斷用的是「持續多久」而不是「佔幾成」。語意後端偶爾逾時一次是抖動，天天
    逾時是壞了，而比率會被使用頻率影響：一天只查三次的話，一次逾時就是 33%。
    所以看的是同一層有沒有從幾小時前一直壞到最新這一次。
    """
    checks: list[dict] = []
    persist_hours = float(os.getenv("XKB_RECALL_DEGRADED_HOURS", "6"))

    # 舊紀錄沒有這個欄位。沒有欄位不等於沒壞掉，只是那時候還沒在記——
    # 對這些列保持沉默，不要假裝它們是綠的。
    known = [r for r in rows if "degraded" in r]
    if not known:
        checks.append({
            "ok": True,
            "msg": "recall telemetry 還沒有 degraded 欄位——下一次召回之後才判斷得出",
        })
        return checks

    latest = known[-1]
    now_broken = set(latest.get("degraded") or [])
    if not now_broken:
        blind = sum(1 for r in known if r.get("empty_because_broken"))
        checks.append({
            "ok": True,
            "msg": f"最近一次召回沒有任何一層退回 fallback"
                   f"（歷史上有 {blind} 次空手是因為壞掉，不是因為沒資料）",
        })
        return checks

    def _ts(row: dict) -> datetime | None:
        try:
            value = datetime.fromisoformat(str(row.get("ts", "")))
        except ValueError:
            return None
        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)

    latest_ts = _ts(latest)
    for layer in sorted(now_broken):
        # 從最新那一筆往回走，走到第一筆「這一層是好的」為止——那才是這一段
        # 連續壞掉的起點。
        #
        # 原本是從頭往後找第一次出現就 break，抓到的是「史上第一次」。於是
        # 三個月前抖一次、中間一千小時全好、剛剛又抖一次，會算出「已經連續壞
        # 了 2000h」——正是這個檢查的註解說要避免的那種誤報。而我為它寫的
        # 測試沒抓到，因為 fixture 裡沒有放更早的那一次。
        streak_start = None
        for row in reversed(known):
            if layer in (row.get("degraded") or []):
                streak_start = _ts(row)
            else:
                break
        if latest_ts and streak_start:
            persisted_hours = (latest_ts - streak_start).total_seconds() / 3600
        else:
            persisted_hours = 0.0
        checks.append({
            "ok": persisted_hours < persist_hours,
            "msg": (f"{layer}：最近一次退回 fallback（單次抖動，還不算故障）"
                    if persisted_hours < persist_hours else
                    f"{layer}：已經連續壞了 {persisted_hours:.0f}h ——"
                    f"召回這段期間回的「查無資料」不能當成真的沒資料"),
        })
    return checks


def check_semantic_index() -> dict:
    """語意索引存不存在、跟得上不跟得上內容。

    向量是離線建的：wiki 或記憶檔改了但沒重建索引，召回會用舊向量算相似度，
    不會報錯，只會撈回過時的段落。二進位格式又沒有自我描述能力，
    keys 與 .bin 不同步時切出來的向量是錯位的——同樣不會拋錯。
    """
    result = {"name": "semantic_index", "checks": []}
    meta_path = Path(os.getenv("XKB_SEMANTIC_INDEX",
                               str(BOOKMARKS_DIR / "semantic_index.json")))
    bin_path = meta_path.with_suffix(".bin")

    if not (meta_path.exists() and bin_path.exists()):
        result["checks"].append({
            "ok": False,
            "msg": "semantic index missing — 召回會退回字串比對（中文命中率大幅下降）",
        })
        return result

    try:
        with meta_path.open(encoding="utf-8") as fh:
            meta = json.load(fh)
        indexed = int(meta.get("count", 0))
        expected_bytes = int(meta.get("bytes", 0))
    except (OSError, ValueError):
        result["checks"].append({"ok": False, "msg": f"semantic index meta unreadable: {meta_path}"})
        return result

    actual_bytes = bin_path.stat().st_size
    result["checks"].append({
        "ok": actual_bytes == expected_bytes,
        "msg": f"index integrity: {indexed} vectors, {actual_bytes} bytes"
               + ("" if actual_bytes == expected_bytes else f" — 應為 {expected_bytes}，keys 與 .bin 不同步"),
    })

    # 內容比索引新 = 索引過時
    newest = 0.0
    for directory in (WIKI_TOPICS_DIR, WORKSPACE / "memory"):
        if not directory.exists():
            continue
        for path in list(directory.glob("*.md"))[:400]:
            newest = max(newest, path.stat().st_mtime)

    if newest:
        lag_hours = (newest - bin_path.stat().st_mtime) / 3600
        max_lag = float(os.getenv("XKB_SEMANTIC_MAX_LAG_HOURS", "26"))
        result["checks"].append({
            "ok": lag_hours <= max_lag,
            "msg": f"index freshness: 內容比索引新 {max(lag_hours, 0):.1f}h (threshold {max_lag:.0f}h)"
                   + ("" if lag_hours <= max_lag else " — 重跑 build_vector_index.py --incremental"),
        })

    return result


def check_topic_map() -> dict:
    """卡片→wiki 的對應表是不是真的在對應東西。

    2026-05-04 的 productization cleanup 把真實對應表換成了 repo 裡的範本
    （your-category-slug → your-wiki-topic-slug）。之後 sync_cards_to_wiki
    每次都回報「No mapped cards found」就結束，離開碼 0，看起來完全正常——
    卡片進 wiki 這條路空轉了三個月沒有人發現。

    範本值是可以直接認出來的，認出來就該報錯。
    """
    result = {"name": "topic_map", "checks": []}
    path = WIKI_DIR / "topic-map.json"

    if not path.exists():
        result["checks"].append({"ok": False, "msg": f"topic-map.json not found: {path}"})
        return result

    try:
        with path.open(encoding="utf-8") as fh:
            mapping = (json.load(fh) or {}).get("mapping", {})
    except (OSError, ValueError) as exc:
        result["checks"].append({"ok": False, "msg": f"topic-map unreadable: {exc}"})
        return result

    placeholders = {"your-category-slug", "another-category", "topic-a", "topic-b",
                    "your-wiki-topic-slug"}
    found = [k for k in mapping if k in placeholders]
    if found:
        result["checks"].append({
            "ok": False,
            "msg": f"topic-map 仍是範本（{', '.join(found)}）— 卡片無法進 wiki，這條路等於停擺",
        })
        return result

    # 對應到的主題必須真的存在，否則吸收時會寫到不存在的頁面
    topics = {p.stem for p in WIKI_TOPICS_DIR.glob("*.md")} if WIKI_TOPICS_DIR.exists() else set()
    dangling = sorted({t for v in mapping.values()
                       for t in (v.get("topics", []) if isinstance(v, dict) else [])
                       if t not in topics})
    result["checks"].append({
        "ok": bool(mapping) and not dangling,
        "msg": f"topic-map: {len(mapping)} 個分類對應"
               + ("" if not dangling else f" — 指向不存在的主題: {', '.join(dangling)}"),
    })
    return result


def check_staging_backlog() -> dict:
    """待審候選有沒有在無聲累積。

    2026-04-07 到 07-16 累積了 146 個檔案、470 條候選，被勾選過的只有 1 條——
    對話裡談出來的結論三個半月都沒進知識庫，而且沒有任何地方會提到這件事。
    擷取層有健檢、召回層有健檢，消化層原本什麼都沒有。
    """
    result = {"name": "staging_backlog", "checks": []}
    staging_dir = WIKI_DIR / "_staging"

    if not staging_dir.exists():
        result["checks"].append({"ok": True, "msg": "no staging dir — nothing pending"})
        return result

    max_pending = int(os.getenv("XKB_STAGING_MAX_PENDING", "60"))
    max_age_days = int(os.getenv("XKB_STAGING_MAX_AGE_DAYS", "30"))

    # Read through xkb_review rather than parsing staging again here. The two
    # parsers had already drifted apart on which files they looked at, and a
    # second definition of "pending" is how this repository ends up with two
    # numbers for one question.
    try:
        import xkb_review
        promoted = xkb_review._promoted_ids(
            xkb_review.GOVERNANCE_DIR / "candidate-registry.jsonl")
        outstanding = [
            c for c in xkb_review.load_candidates(classify=False)
            # Keep this check's long-standing scope: top-level staging only,
            # not the pre-2026-07 archive.
            if "/" not in c.source_file
            and c.status == "pending"
            and c.candidate_id not in promoted
        ]
    except Exception as exc:
        result["checks"].append({"ok": False, "msg": f"staging counts unavailable: {exc}"})
        return result

    pending = len(outstanding)
    dates = [c.source_date for c in outstanding if c.source_date != "unknown"]
    oldest: str | None = min(dates) if dates else None

    # 問的是「追不追得上」，不是「積了幾筆」。
    #
    # 原本這一項在 pending > 60 時紅燈，訊息寫「用 xkb_review.py 審核」，而它被
    # 歸在「等你決定」底下。2026-09-12 實測：180 筆待審裡 106 筆是 safe_promotion
    # ——治理會自動吸收，一筆都不需要人。積起來的原因是治理每晚上限 20 筆，而候選
    # 進來得比 20 快，連續三天都打在上限上。
    #
    # 那是吞吐量設定，不是判斷。而「積了幾筆」這個問法在任何補跑期間都會紅燈，
    # 於是它每天叫人去審核一件系統自己會做的事——這則訊息會被停止閱讀就是這樣來的。
    #
    # 現在拿帳本裡的實際吸收速度跟候選進來的速度比。追得上就不是故障，不管積了幾筆；
    # 追不上才是，而且訊息會說「追不上」而不是「請你審核」。
    absorbed_per_day, arrived_per_day = _governance_rates()

    if absorbed_per_day is None or arrived_per_day is None:
        # 帳本還沒有足夠紀錄算速度（要三次治理執行）。
        #
        # 這種時候不要紅燈。「積了幾筆」這個問法正是我們要淘汰的那一個——它在
        # 補跑期間必定紅燈，而現在恰好就是補跑期間。拿一個已知會誤報的判準去填
        # 空窗，等於把噪音留在門口。
        #
        # 說出狀態、但不當故障：真正的判斷等帳本有資料再做。
        result["checks"].append({
            "ok": True,
            "msg": f"待審候選 {pending} 筆——還要 {3 - _governance_runs()} 次治理執行"
                   f"才算得出吸收速度，在那之前不判斷追不追得上",
        })
    else:
        keeping_up = absorbed_per_day >= arrived_per_day
        result["checks"].append({
            "ok": keeping_up,
            "msg": (f"待審候選 {pending} 筆；治理每天吸收 {absorbed_per_day:.0f} 筆、"
                    f"進來 {arrived_per_day:.0f} 筆"
                    + ("——追得上" if keeping_up else
                       "——追不上，調高 run_candidate_governance.sh 的 LIMIT")),
        })

    if oldest:
        age_days = (datetime.now(timezone.utc).date() - datetime.strptime(oldest, "%Y-%m-%d").date()).days
        # 年齡只在追不上的時候才有意義：追得上的話，舊的那批本來就排在前面，
        # 會自己被吸收掉。單看年齡會讓一個正在消化的佇列每天紅燈。
        stalled = (absorbed_per_day is not None and arrived_per_day is not None
                   and absorbed_per_day < arrived_per_day)
        result["checks"].append({
            "ok": not (stalled and age_days > max_age_days),
            "msg": f"最舊的待審候選：{oldest}（{age_days} 天前）"
                   + ("" if not stalled else f"，而且追不上——它不會自己消失"),
        })

    return result


def _governance_runs() -> int:
    """帳本裡有幾次治理紀錄。用來說「還要幾次才算得出速度」。"""
    try:
        import xkb_ledger
        return len([r for r in xkb_ledger.read("candidate-governance")
                    if isinstance(r.get("produced"), int)])
    except Exception:  # noqa: BLE001
        return 0


def _governance_rates() -> tuple[float | None, float | None]:
    """(每天吸收幾筆, 每天進來幾筆)。算不出來時回 (None, None)。

    吸收速度來自管線帳本——那是實際發生過的事，不是設定值。進來的速度用待審
    候選的日期分佈算：最舊到今天的天數除以總筆數，是平均到達率。

    兩邊都算不出來時不要猜。猜一個速度然後拿它判斷，會得到一個看起來精確的錯誤。
    """
    try:
        import xkb_ledger
        rows = xkb_ledger.read("candidate-governance")
    except Exception:  # noqa: BLE001
        return None, None

    absorbed = [r.get("produced") for r in rows if isinstance(r.get("produced"), int)]
    if len(absorbed) < 3:
        return None, None
    recent = absorbed[-7:]
    per_day = sum(recent) / len(recent)

    try:
        import xkb_review
        promoted = xkb_review._promoted_ids(
            xkb_review.GOVERNANCE_DIR / "candidate-registry.jsonl")
        outstanding = [c for c in xkb_review.load_candidates(classify=False)
                       if "/" not in c.source_file and c.status == "pending"
                       and c.candidate_id not in promoted]
    except Exception:  # noqa: BLE001
        return per_day, None

    dates = sorted(c.source_date for c in outstanding if c.source_date != "unknown")
    if len(dates) < 2:
        return per_day, 0.0
    span = (datetime.strptime(dates[-1], "%Y-%m-%d").date()
            - datetime.strptime(dates[0], "%Y-%m-%d").date()).days
    return per_day, len(outstanding) / max(span, 1)


def check_governance_actionable() -> dict:
    """Warn only when bounded governance backlog exceeds threshold or is overdue."""
    result = {"name": "governance_actionable", "checks": []}
    try:
        import xkb_review
        counts = xkb_review.governance_health_counts(int(os.getenv("XKB_GOVERNANCE_TTL_DAYS", "30")))
    except Exception as exc:
        result["checks"].append({"ok": False, "msg": f"governance counts unavailable: {exc}"})
        return result
    threshold = int(os.getenv("XKB_GOVERNANCE_MAX_PENDING", os.getenv("XKB_STAGING_MAX_PENDING", "60")))
    actionable = {key: counts.get(key, 0) for key in ("pending", "medium", "low", "proposal", "quarantine", "overdue", "safe_promotion")}
    actionable["ttl"] = actionable["quarantine"]
    result["actionable_counts"] = actionable
    warning = counts["pending"] > threshold or counts["overdue"] > 0
    result["checks"].append({"ok": not warning, "msg": "governance actionable counts: " + json.dumps(actionable, ensure_ascii=False) + (f" — threshold {threshold}" if warning else "")})
    return result


def check_external_dependencies() -> dict:
    """The two things XKB needs and does not ship.

    gbrain holds the card side of retrieval; ten scripts query it. openclaw
    copies the wiki into the memory vault for other agents. Both are meant to
    be here, and both fail quietly — gbrain by returning an empty result that
    is indistinguishable from "nothing on that subject", openclaw by a mirror
    that silently stops being written.
    """
    result = {"name": "external_dependencies", "checks": []}

    try:
        sys.path.insert(0, str(Path(__file__).resolve().parent))
        from xbrain_recall import GBRAIN_AVAILABLE, xbrain_query
    except Exception as err:  # noqa: BLE001
        result["checks"].append(
            {"ok": False, "msg": f"gbrain 橋接載不進來：{type(err).__name__}: {err}"})
        return result

    if not GBRAIN_AVAILABLE:
        result["checks"].append(
            {"ok": False, "msg": "找不到 gbrain — 卡片那一半的召回會靜靜回空"})
    else:
        # 一次真的查詢。「裝在那裡」跟「回答得出來」是兩件事。
        #
        # xbrain_query 不會拋例外——它把 timeout、找不到執行檔、非零離開碼
        # 全部接住並回 []。所以這裡不能寫成 try/except：那個分支永遠不會執行，
        # 而所有失敗都會被說成「索引可能是空的」，在一個專門為了「不要說錯
        # 原因」而寫的檢查裡說錯原因。空結果本身只能說明「查不到」。
        hits = xbrain_query("知識", limit=1)
        if hits:
            result["checks"].append({"ok": True, "msg": "gbrain 查得到東西"})
        else:
            result["checks"].append(
                {"ok": False,
                 "msg": "gbrain 一個常見詞查不到東西 — 可能是索引空的，也可能是它根本沒回應"
                        "（xbrain_query 兩種情況都回空，分不出來）"})

    # which() 只看呼叫者的 PATH，而健檢從純 cron 跑，PATH 通常只有
    # /usr/bin:/bin。openclaw 是使用者裝的，所以光靠 which 會每天早上
    # 通報一個正常運作的鏡像壞了。
    openclaw = shutil.which("openclaw") or next(
        (p for p in (Path("/usr/bin/openclaw"), Path("/usr/local/bin/openclaw"),
                     Path.home() / ".local" / "bin" / "openclaw") if p.exists()),
        None,
    )
    if openclaw:
        result["checks"].append({"ok": True, "msg": f"openclaw 在（wiki 鏡像需要它）"})
    else:
        result["checks"].append(
            {"ok": False,
             "msg": "找不到 openclaw — wiki 不會再被複製到記憶庫，agent 搜不到它。"
                    "XKB 自己的召回不受影響"})
    return result


def check_conversation_capture(days: int = 3) -> dict:
    """Conversations must actually reach the store, not just open a session.

    The hook opens a session and then records a turn. Between 2026-08-24 and
    08-29 it opened sessions and recorded nothing: stdin was decoded with the
    Windows codepage, the surrogates that produced raised UnicodeEncodeError
    inside turn_id, and that is a ValueError, which the fail-open handler is
    written to swallow. Exit code 0, no error anywhere, five days of Chinese
    conversations lost.

    Failing open is right — XKB must never block a conversation — but failing
    open is not the same as failing silently. A day whose sessions all hold
    zero turns is the signature, and it is visible from the store alone.
    """
    result = {"name": "conversation_capture", "checks": []}
    db_path = xkb_paths.SERVICE_DB
    if not db_path.exists():
        result["checks"].append({"ok": True, "msg": "no knowledge service store on this host"})
        return result

    since = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
    try:
        import sqlite3
        from contextlib import closing
        with closing(sqlite3.connect(f"file:{db_path}?mode=ro", uri=True)) as db:
            total, empty = db.execute(
                """SELECT COUNT(*),
                          SUM(CASE WHEN t.n IS NULL THEN 1 ELSE 0 END)
                     FROM sessions s
                     LEFT JOIN (SELECT session_id, COUNT(*) n FROM turns GROUP BY session_id) t
                       ON t.session_id = s.session_id
                    WHERE s.created_at >= ?""",
                (since,),
            ).fetchone()
    except Exception as exc:
        result["checks"].append({"ok": False, "msg": f"capture counts unavailable: {exc}"})
        return result

    total = total or 0
    empty = empty or 0
    if total == 0:
        result["checks"].append({"ok": True, "msg": f"no conversations in the last {days}d — nothing to capture"})
        return result

    # Every session empty means capture is broken, not that the conversations
    # were unremarkable. A mix is normal: a session can open and go unused.
    broken = empty == total
    result["checks"].append({
        "ok": not broken,
        "msg": (f"conversation capture: {total - empty}/{total} sessions recorded turns "
                f"(last {days}d)" + ("" if not broken else
                " — hook 開了 session 卻沒有記下任何 turn，召回與候選池都收不到東西")),
    })
    return result


def check_provenance_markers() -> dict:
    """Knowledge distilled from Pan's own notes must be marked as such.

    Recall down-weights self-derived knowledge so that collecting an idea,
    recalling it, discussing it and distilling it again cannot turn one
    opinion into a consensus. The penalty is applied by looking for a marker
    in the text, so a writer that invents its own annotation silently opts
    out of it — which is how 913 promoted claims ended up competing at full
    weight against the external sources they were reasoned from.
    """
    result = {"name": "provenance_markers", "checks": []}
    topics_dir = WIKI_DIR / "topics"
    if not topics_dir.exists():
        result["checks"].append({"ok": True, "msg": "no topics dir"})
        return result
    try:
        sys.path.insert(0, str(Path(__file__).resolve().parent))
        from xkb_provenance import MARKER, is_self_derived
    except Exception as exc:
        result["checks"].append({"ok": False, "msg": f"provenance module unavailable: {exc}"})
        return result

    # Three tools write knowledge into these pages and each annotates a line
    # its own way — one names a memory file, one a URL, one a staging entry.
    # That variety is fine; the annotations carry different facts. What must
    # hold for every one of them is that a reader can tell where the line came
    # from, because recall down-weights our own conclusions and cannot do that
    # for a line it fails to classify. A fourth writer inventing a fourth
    # format shows up here rather than in the ranking six months later.
    annotated = re.compile(r"^\s*-\s.*\*\(")
    external = re.compile(r"https?://")
    unclassified = 0
    files = set()
    for path in sorted(topics_dir.glob("*.md")):
        try:
            content = path.read_text(encoding="utf-8")
        except OSError:
            continue
        for line in content.splitlines():
            if not annotated.match(line):
                continue
            # Asked with the reader's own function, not a copy of its rule:
            # a check that decides for itself what counts is checking
            # something other than what happens at recall.
            if is_self_derived(line) or external.search(line):
                continue
            unclassified += 1
            files.add(path.name)

    result["checks"].append({
        "ok": unclassified == 0,
        "msg": ("every knowledge line names an external source or is marked self-derived"
                if unclassified == 0 else
                f"{unclassified} knowledge lines in {len(files)} page(s) name neither a "
                f"source URL nor '{MARKER}', so recall cannot weigh them — "
                f"跑 scripts/backfill_self_derived_marker.py --apply"),
    })
    return result


def check_pipeline_ledger() -> dict:
    """每個階段上一次執行做了什麼——照它自己的節奏判斷，不用手調門檻。

    這一項問的是別的檢查問不到的事。其他檢查看的都是現況（有幾張卡、索引裡
    有幾筆），而現況是可以重算的，重算不出歷史。2026-09-01 到 09-09 書籤轉卡
    每晚失敗、八天 0 產出時，所有現況的數字都還在、都「正常」，沒有任何一項
    檢查看得到那八個晚上。

    門檻用各階段自己的紀錄算：跑得多密由中位間隔決定，改排程不會變成誤報。
    寫死成 24 小時的話，一改排程就天天紅燈，而天天紅燈等於沒有檢查。
    """
    result = {"name": "pipeline_ledger", "checks": []}

    try:
        import xkb_ledger
    except Exception as err:  # noqa: BLE001
        result["checks"].append({"ok": False, "msg": f"帳本模組載入失敗：{err}"})
        return result

    rows = xkb_ledger.read()
    if not rows:
        # 帳本是新的，或這台機器還沒跑過任何階段。沒有紀錄不等於故障。
        result["checks"].append({
            "ok": True,
            "msg": f"帳本還沒有紀錄（{xkb_ledger.LEDGER_PATH.name}）——下一次排程執行後才會有",
        })
        return result

    now = datetime.now(timezone.utc)
    quiet_runs = int(os.getenv("XKB_LEDGER_QUIET_RUNS", "3"))

    for stage in sorted({r.get("stage", "") for r in rows if r.get("stage")}):
        stage_rows = [r for r in rows if r.get("stage") == stage]
        last = stage_rows[-1]
        last_ts = xkb_ledger._parse_ts(last)

        # 1. 上一次整批失敗
        if not last.get("ok", True):
            result["checks"].append({
                "ok": False,
                "msg": f"{stage}：上一次執行整批失敗——{last.get('reason') or '沒有記下原因'}",
            })
            continue

        # 2. 這個階段停了。快慢由它自己的紀錄決定。
        gap = xkb_ledger.typical_gap_seconds(stage_rows)
        if gap and last_ts:
            silent = (now - last_ts).total_seconds()
            if silent > gap * 3:
                result["checks"].append({
                    "ok": False,
                    "msg": (f"{stage}：平常每 {gap / 3600:.0f}h 跑一次，"
                            f"已經 {silent / 3600:.0f}h 沒有紀錄——排程可能沒在跑"),
                })
                continue

        # 3. 連續幾次都沒有產出，而且還有東西在排隊。
        recent = stage_rows[-quiet_runs:]
        produced_recent = [r.get("produced") for r in recent if "produced" in r]
        waiting = any((r.get("pending") or 0) > 0 or (r.get("failed") or 0) > 0
                      for r in recent)
        if (len(produced_recent) == quiet_runs
                and all(p == 0 for p in produced_recent) and waiting):
            reason = next((r.get("reason") for r in reversed(recent) if r.get("reason")), "")
            result["checks"].append({
                "ok": False,
                "msg": (f"{stage}：連續 {quiet_runs} 次執行都沒有產出，但還有東西在排隊"
                        + (f"——{reason}" if reason else "")),
            })
            continue

        bits = [f"{k}={last[k]}" for k in ("intake", "produced", "failed", "pending")
                if k in last]
        age = f"{(now - last_ts).total_seconds() / 3600:.0f}h 前" if last_ts else "時間不明"
        result["checks"].append({
            "ok": True,
            "msg": f"{stage}：{age}　" + " ".join(bits) if bits else f"{stage}：{age}",
        })

    return result


def _newest_card() -> tuple[float | None, str]:
    """最新一張卡片的 mtime 與檔名；沒有卡片時回 (None, "")。"""
    newest_mtime: float | None = None
    newest_name = ""
    for card in xkb_paths.card_files():
        try:
            mtime = card.stat().st_mtime
        except OSError:
            continue
        if newest_mtime is None or mtime > newest_mtime:
            newest_mtime, newest_name = mtime, card.name
    return newest_mtime, newest_name


def check_card_production() -> dict:
    """書籤有沒有真的在變成知識卡。

    2026-09-01 到 09-09，排程每晚都跑、每晚都回報成功，但每一筆都以 404 失敗：
    八天、25 筆失敗、0 張卡。整條擷取管線是死的，而這份健檢一個字都沒提——
    它只看得到「索引檔很久沒被寫過」，因為沒有新卡片本來就不會寫索引。

    所以這裡問的是最上游那件事：東西進來了，有沒有變成知識。佇列本身變長不算
    故障（那是消化速度，limit 的事）；佇列不是空的、卻連一張新卡都生不出來，
    才是故障。
    """
    result = {"name": "card_production", "checks": []}

    try:
        from xkb_pending_work import pending_breakdown
        counts = pending_breakdown(BOOKMARKS_DIR, CARDS_DIR)
    except Exception as err:  # noqa: BLE001 — 健檢自己不能把整份報告弄掛
        result["checks"].append({
            "ok": False,
            "msg": f"讀不出待轉書籤數量，無法判斷擷取管線是否還活著：{err}",
        })
        return result

    # 卡住的項目不是故障，是一件等人決定的事：沒有任何排程會再碰 failed，
    # 所以它們既不會好轉也不會惡化，只會一直在那裡。說出來，但不要算成紅燈——
    # 每天亮著的紅燈會把真正的紅燈一起淹掉。
    if counts["stuck"]:
        result["checks"].append({
            "ok": True,
            "msg": f"{counts['stuck']} 筆卡在失敗狀態，不會自動重試（要重跑得手動重設佇列）",
        })

    # 「還會被處理的」才是判斷管線活著與否的依據。拿全部待轉去判斷的話，
    # 佇列裡只剩卡住的項目時，這一項會永遠紅燈。
    pending = counts["actionable"]

    stall_hours = int(os.getenv("XKB_CARD_STALL_HOURS", "48"))
    newest_mtime, newest_name = _newest_card()

    if newest_mtime is None:
        result["checks"].append({
            "ok": pending == 0,
            "msg": f"memory/cards/ 是空的，還有 {pending} 筆書籤排隊中",
        })
        return result

    idle_hours = (datetime.now().timestamp() - newest_mtime) / 3600

    # 佇列空的時候，多久沒產出新卡片都是正常的——那只代表你沒有新書籤。
    if pending == 0:
        result["checks"].append({
            "ok": True,
            "msg": f"書籤佇列已清空；最後一張卡 {newest_name}（{idle_hours:.0f}h 前）",
        })
        return result

    stalled = idle_hours > stall_hours
    result["checks"].append({
        "ok": not stalled,
        "msg": (f"{pending} 筆書籤排隊中，最後一張卡 {idle_hours:.0f}h 前產出"
                if not stalled else
                f"{pending} 筆書籤排隊中，但已經 {idle_hours:.0f}h 沒有任何新卡片"
                f"（超過 {stall_hours}h）——書籤轉卡可能每晚都在失敗，"
                f"看 /tmp/xkb-bookmark-batch.log"),
    })
    return result


def check_index_freshness() -> dict:
    """檢查 search_index 和 vector_index 的 summary 覆蓋率，以及有沒有漏掉卡片。"""
    result = {"name": "index_freshness", "checks": []}

    # search_index
    if not INDEX_FILE.exists():
        result["checks"].append({"ok": False, "msg": f"search_index.json not found: {INDEX_FILE}"})
        return result

    raw = json.loads(INDEX_FILE.read_text(encoding="utf-8"))
    items = raw.get("items", [])
    total = len(items)
    has_summary = sum(1 for i in items if (i.get("summary") or "").strip())
    enriched = sum(1 for i in items if i.get("enriched"))
    coverage = round(has_summary / total * 100) if total else 0

    result["checks"].append({
        "ok": coverage >= 70,
        "msg": f"search_index summary coverage: {has_summary}/{total} ({coverage}%) | enriched: {enriched}"
    })

    # 索引有沒有落後於卡片。
    #
    # 這裡原本問的是「索引檔 26 小時內有沒有被寫過」，而 sync_enriched_index
    # 只在真的有東西要寫時才寫檔。於是這一項實際上問的是「今天有沒有新知識
    # 進來」——安靜的一天會報壞，而 2026-09-01 到 09-09 書籤轉卡整個停擺時，
    # 它也只報得出「索引沒有跟上」，沒有一個字提到一張卡都沒產出。症狀的症狀。
    #
    # 改成比時間戳也不行：卡片會被去重、鏡像、回填之類的維護動到 mtime，那會
    # 讓「最新的卡比索引新」每天亮紅燈，換一種噪音而已。落後與否是名詞的問題
    # 不是時鐘的問題——哪幾張卡不在索引裡。剛產出的卡還在路上，所以只算那些
    # 已經過了一輪同步時間、卻仍然沒進索引的。
    mtime = INDEX_FILE.stat().st_mtime
    age_hours = (datetime.now().timestamp() - mtime) / 3600
    result["checks"].append({
        "ok": True,
        "msg": f"search_index last written {age_hours:.1f}h ago",
    })

    # 比對要用「來源」不是「檔名」。同一個來源在 memory/cards/ 可能有兩個
    # 檔名（github-x 與 github_star-x、卡片與書籤各一份），索引去重之後只留
    # 一列，留下的那一列指向另一個檔名。純比檔名會把這種重複檔報成「漏了、
    # 關鍵字查不到」——內容其實查得到，那是紅燈但不是故障，跟它要取代的
    # 誤報一樣沒用。
    grace_hours = int(os.getenv("XKB_INDEX_MISS_GRACE_HOURS", "26"))
    indexed_stems = set()
    indexed_urls = set()
    for item in items:
        rel = item.get("relative_path") or item.get("path") or ""
        if rel:
            indexed_stems.add(Path(rel).stem)
        url = (item.get("source_url") or "").strip()
        if url:
            indexed_urls.add(url)

    cutoff = datetime.now().timestamp() - grace_hours * 3600
    missed = []
    for card in xkb_paths.card_files():
        if card.stem in indexed_stems:
            continue
        try:
            if card.stat().st_mtime >= cutoff:
                continue  # 剛產出，還沒輪到同步
            text = card.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        source = re.search(r'^source_url:\s*"?([^"\n]+)"?\s*$', text, re.MULTILINE)
        if source and source.group(1).strip() in indexed_urls:
            continue  # 同一個來源已經以另一個檔名進了索引
        missed.append(card.name)

    result["checks"].append({
        "ok": not missed,
        "msg": ("every card older than the sync window is in the search index"
                if not missed else
                f"{len(missed)} 張卡片不在 search_index 裡（{grace_hours}h 以上）"
                f"，關鍵字查不到：{', '.join(sorted(missed)[:3])}"
                f"{' …' if len(missed) > 3 else ''} — 跑 scripts/build_search_index.sh --incremental"),
    })

    # vector_index：檢查新鮮度就好，不要把 62MB 讀進來數個數。
    # 健檢是每天自動跑的，成本要跟「確認狀態」相稱，不是重做一次工作。
    # 原本 json.loads 整份檔案在 VPS 上要 7 秒（冷快取時更久），
    # 而它換來的只是一個向量數字——那個數字在 semantic_index.json 的 meta 裡就有。
    if not VECTOR_FILE.exists():
        # 語意召回讀的是 semantic_index.bin，不是這一份。
        # 這份只是重建索引時的中間產物，缺了不影響召回。
        result["checks"].append({
            "ok": True,
            "msg": "vector_index.json absent — 召回讀 semantic_index，此檔僅為重建用中間產物",
        })
    else:
        v_mtime = VECTOR_FILE.stat().st_mtime
        size_mb = VECTOR_FILE.stat().st_size / 1e6
        v_age_hours = (datetime.now().timestamp() - v_mtime) / 3600
        # 同樣不看時鐘：這份也是有東西才寫。語意層真正的落後由 semantic_index
        # 那一節比對內容判斷，這裡只確認檔案不是空的。
        result["checks"].append({
            "ok": size_mb > 0.1,
            "msg": f"vector_index: {size_mb:.0f}MB, last written {v_age_hours:.1f}h ago",
        })

    # cards vs index coverage
    if CARDS_DIR.exists():
        card_count = len(xkb_paths.card_files())
        result["checks"].append({
            "ok": enriched >= card_count * 0.9,
            "msg": f"Cards in memory/cards/: {card_count} | enriched in index: {enriched}"
        })

    return result


def fmt_checks(section: dict) -> str:
    lines = [f"\n── {section['name']} ──"]
    for c in section["checks"]:
        icon = OK if c["ok"] else FAIL
        lines.append(f"  {icon}  {c['msg']}")
    return "\n".join(lines)


# 健檢包含哪些項目——唯一的定義。
#
# 原本這份清單有兩份：這支腳本一份，health_check_notify（排程每天跑的那支）
# 自己再寫一份。2026-09-11 的後果是：我在這裡加了 card_production 與
# pipeline_ledger，寫了測試確認「有註冊」，然後那兩項在每日訊息裡一次都沒出現
# 過——因為每天跑的是另一份清單。反過來 external_dependencies 只在 notify 那份
# 裡，所以手動跑這支時它不會被檢查。
#
# 兩份清單就是兩個會各自漂走的答案。新增檢查時只改這裡。
CHECKS = (
    check_wiki_canonical,
    check_recall_wiki_source,
    check_recall_live,
    check_recall_telemetry,
    check_semantic_index,
    check_topic_map,
    check_staging_backlog,
    check_governance_actionable,
    check_provenance_markers,
    check_conversation_capture,
    check_external_dependencies,
    check_card_production,
    check_pipeline_ledger,
    check_index_freshness,
)


def run_all() -> list[dict]:
    """跑完整份健檢。呼叫端不要自己列清單。"""
    return [check() for check in CHECKS]


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--json", action="store_true")
    args = parser.parse_args()

    sections = run_all()

    if args.json:
        print(json.dumps(sections, ensure_ascii=False, indent=2))
        # 離開碼要反映健康狀態，--json 也一樣：排程是靠離開碼判斷要不要告警的
        return 0 if all(c["ok"] for s in sections for c in s["checks"]) else 1

    print("🔍  XKB Pipeline Health Check")
    print(f"    {datetime.now().strftime('%Y-%m-%d %H:%M')}")

    all_ok = True
    for s in sections:
        print(fmt_checks(s))
        if any(not c["ok"] for c in s["checks"]):
            all_ok = False

    print()
    if all_ok:
        print(f"{OK}  All checks passed.")
    else:
        print(f"{FAIL}  Some checks failed — review above.")

    return 0 if all_ok else 1


import xkb_usage  # noqa: E402  — 量測誰在跑，見 scripts/xkb_usage.py

if __name__ == "__main__":
    xkb_usage.record(__file__)
    raise SystemExit(main())