scripts/health_check_pipeline.py
Rama main · réplica de lectura del repositorio original.
#!/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())