Leo
c2638668e3
leo 2026-08-20：「同一個 plugin 你用，薄殼也用，保證兩邊同步」
「我要你幫雲端做薄殼，永遠都有問題，你要做的就是這組設定
你自己可以 dogfooding」
搬進來：41 支 hook（51 條註冊）／7 支 command／2 支 skill／23 支腳本。
不搬 .env、wiki、docs——那些是知識不是環境。
51 條 hook 路徑全部從 $CLAUDE_PROJECT_DIR/.claude/hooks/ 改成 ${CLAUDE_PLUGIN_ROOT}/hooks/，
零漏網。那正是薄殼一直壞掉的根：雲端 cwd 不是真身，寫死路徑就斷。
尚未驗證：Claude Code 能不能從私有 Gitea repo 裝 marketplace（要憑證）。
下一步就是在本機實際裝一次，通了才動雲端 bootstrap.sh。
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
392 lines
17 KiB
Python
392 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
KBDB 實機考卷 — 判分器（看落地的資料，不看受測者自述）
|
||
|
||
設計鐵律
|
||
--------
|
||
1. **只讀資料庫的真身**（0007 之後的樹狀模型），不讀受測者的回報、不讀 API 的呈現。
|
||
API 可以把 JSON 團漂亮地印出來；D1 不會替它掩護。
|
||
2. **同一份 SQL 同時服務自我驗證與正式考試**——只換連線層（sqlite / D1）。
|
||
若自我驗證跑的是另一段邏輯，它就證明不了正式考試。
|
||
3. **只數安靜的錯**。API 退回、參數形狀錯（大聲的錯）不在本判分器範圍
|
||
（考卷 §二：大聲的錯當場修掉，不計分）。
|
||
|
||
模型（matrix/arcrun/kbdb/migrations/0007_tree_record_model.sql）
|
||
---------------------------------------------------------------
|
||
sheet entries.entry_type='sheet'，且有一條 (src=sheet, rel=sys_belongs, dst=sys_root)
|
||
field entries.entry_type='field'，且有一條 (src=field, rel=sys_field_of, dst=sheet)
|
||
record entries.entry_type='record'，且有一條 (src=record, rel=sys_belongs, dst=sheet)
|
||
格子 一條 (src=record, rel=<field id>, dst=<value entry>)
|
||
"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import subprocess
|
||
import sys
|
||
from collections import Counter, defaultdict
|
||
|
||
SYS = {"sys_root", "sys_belongs", "sys_field_of"}
|
||
|
||
# ── 這一段 SQL 是判分器的全部輸入。sqlite 與 D1 共用同一份字串。 ──────────────
|
||
LOAD_SQL = (
|
||
"SELECT id, content, entry_type, metadata_json, src_id, rel_id, dst_id, created_at "
|
||
"FROM entries"
|
||
)
|
||
|
||
|
||
# ───────────────────────────── 連線層（唯一有分歧的地方） ─────────────────────
|
||
def load_sqlite(path):
|
||
con = sqlite3.connect(path)
|
||
con.row_factory = sqlite3.Row
|
||
rows = [dict(r) for r in con.execute(LOAD_SQL)]
|
||
con.close()
|
||
return rows
|
||
|
||
|
||
def load_d1(dbname, account_id, token):
|
||
env = dict(os.environ)
|
||
env["CLOUDFLARE_ACCOUNT_ID"] = account_id
|
||
env["CLOUDFLARE_API_TOKEN"] = token
|
||
out = subprocess.run(
|
||
["npx", "--yes", "wrangler@latest", "d1", "execute", dbname,
|
||
"--remote", "--json", "--command", LOAD_SQL],
|
||
capture_output=True, text=True, env=env, timeout=300,
|
||
).stdout
|
||
out = out[out.find("["):]
|
||
return json.loads(out)[0]["results"]
|
||
|
||
|
||
# ───────────────────────────── 模型重建 ──────────────────────────────────────
|
||
class Pool:
|
||
def __init__(self, rows):
|
||
self.by_id = {r["id"]: r for r in rows}
|
||
self.rows = rows
|
||
self.rels = [r for r in rows if r.get("rel_id")]
|
||
|
||
def sheets(self):
|
||
out = {}
|
||
for r in self.rels:
|
||
if r["rel_id"] == "sys_belongs" and r["dst_id"] == "sys_root":
|
||
e = self.by_id.get(r["src_id"])
|
||
if e:
|
||
out[e["id"]] = e.get("content")
|
||
return out
|
||
|
||
def fields_of(self, sheet_id):
|
||
return {r["src_id"]: (self.by_id.get(r["src_id"], {}) or {}).get("content")
|
||
for r in self.rels
|
||
if r["rel_id"] == "sys_field_of" and r["dst_id"] == sheet_id}
|
||
|
||
def records_of(self, sheet_id):
|
||
return [r["src_id"] for r in self.rels
|
||
if r["rel_id"] == "sys_belongs" and r["dst_id"] == sheet_id]
|
||
|
||
def cells_of(self, record_id):
|
||
out = []
|
||
for r in self.rels:
|
||
if r["src_id"] == record_id and r["rel_id"] not in SYS:
|
||
v = self.by_id.get(r["dst_id"])
|
||
if v is not None:
|
||
out.append((r["rel_id"], v))
|
||
return out
|
||
|
||
|
||
# ───────────────────────────── 偵測器（每一支＝一種安靜的錯） ─────────────────
|
||
def _parse_json(s):
|
||
if not isinstance(s, str):
|
||
return None
|
||
s = s.strip()
|
||
if not s or s[0] not in "{[":
|
||
return None
|
||
try:
|
||
return json.loads(s)
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
KV_PACK = re.compile(r"[^\s=:;,]+\s*[=:]\s*[^;,\n]+")
|
||
|
||
|
||
def is_json_blob(content):
|
||
"""一個格子裡塞了一整包結構 —— D91 本人。"""
|
||
v = _parse_json(content)
|
||
if isinstance(v, dict) and len(v) >= 2:
|
||
return True
|
||
# 一格裡放「物件的清單」＝結構被塞進格子，不論長度。實測 youlin 上
|
||
# library_map.relation_profile 只有一個元素，若要求 len>=2 就會漏掉它。
|
||
if isinstance(v, list) and any(isinstance(x, (dict, list)) for x in v):
|
||
return True
|
||
if isinstance(v, list) and len(v) >= 2:
|
||
return True
|
||
return False
|
||
|
||
|
||
def is_kv_packed(content):
|
||
"""沒用 JSON，改用 `a=1; b=2` 把多欄擠進一格 —— 換個門進來的同一個病。"""
|
||
if not isinstance(content, str) or is_json_blob(content):
|
||
return False
|
||
if len(content) < 8:
|
||
return False
|
||
if not (";" in content or "\n" in content or "," in content):
|
||
return False
|
||
return len(KV_PACK.findall(content)) >= 2
|
||
|
||
|
||
def is_multifact(content):
|
||
"""一個格子裡塞了好幾筆事實 —— 關係被寫成附加物。"""
|
||
if not isinstance(content, str):
|
||
return False
|
||
v = _parse_json(content)
|
||
if isinstance(v, list) and len(v) >= 2:
|
||
return True
|
||
lines = [x for x in content.splitlines() if x.strip()]
|
||
if len(lines) < 2:
|
||
return False
|
||
sep = re.compile(r"(->|→|—|--|\||,|、|愛吃|屬於|喜歡|是)")
|
||
return sum(1 for ln in lines if sep.search(ln)) >= 2
|
||
|
||
|
||
STRUCT_META_WHITELIST = {"source", "source_uri", "hash", "content_hash", "ts", "updated_at"}
|
||
|
||
|
||
def structured_metadata(entry):
|
||
"""結構化欄位被打包進 metadata_json —— D91 的原始發作處。"""
|
||
v = _parse_json(entry.get("metadata_json"))
|
||
if not isinstance(v, dict):
|
||
return False
|
||
return len(set(v.keys()) - STRUCT_META_WHITELIST) >= 2
|
||
|
||
|
||
# ───────────────────────────── 訊號彙總 ──────────────────────────────────────
|
||
LIST_SIGNALS = ("json_blob_cells", "kv_packed_cells", "multifact_cells",
|
||
"declared_unused_fields", "used_undeclared_fields",
|
||
"structured_metadata_entries", "duplicate_value_contents",
|
||
"empty_records", "shared_value_entries", "orphan_relations")
|
||
|
||
|
||
def signals(pool, sheet_ids):
|
||
s = {k: [] for k in LIST_SIGNALS}
|
||
s.update({"sheets": len(sheet_ids), "records": 0, "cells": 0, "per_sheet": {}})
|
||
value_ids_by_content = defaultdict(set)
|
||
cells_per_value = Counter()
|
||
|
||
for sh in sheet_ids:
|
||
declared = pool.fields_of(sh)
|
||
recs = pool.records_of(sh)
|
||
used = set()
|
||
name = (pool.by_id.get(sh, {}) or {}).get("content")
|
||
per = {"name": name,
|
||
"declared_fields": sorted(x for x in declared.values() if x),
|
||
"records": len(recs), "cells": 0}
|
||
for rec in recs:
|
||
cells = pool.cells_of(rec)
|
||
per["cells"] += len(cells)
|
||
s["records"] += 1
|
||
s["cells"] += len(cells)
|
||
if not cells:
|
||
# 一筆記錄存在，但一個格子都沒有。
|
||
# 這是 createRecord 對「template 沒宣告的 slot」靜默略過造成的——
|
||
# API 回 200、受測者會宣稱成功，而資料是空的。最安靜的一種錯。
|
||
s["empty_records"].append(f"{name}:{rec}")
|
||
for fid, val in cells:
|
||
cells_per_value[val["id"]] += 1
|
||
used.add(fid)
|
||
c = val.get("content")
|
||
tag = f"{name}.{declared.get(fid) or fid}"
|
||
if is_json_blob(c):
|
||
s["json_blob_cells"].append((tag, (c or "")[:80]))
|
||
elif is_kv_packed(c):
|
||
s["kv_packed_cells"].append((tag, (c or "")[:80]))
|
||
if is_multifact(c):
|
||
s["multifact_cells"].append((tag, (c or "")[:80]))
|
||
if structured_metadata(val):
|
||
s["structured_metadata_entries"].append(val["id"])
|
||
if c is not None:
|
||
value_ids_by_content[c].add(val["id"])
|
||
for fid, fname in declared.items():
|
||
if fid not in used:
|
||
s["declared_unused_fields"].append(f"{name}.{fname or fid}")
|
||
for fid in used - set(declared):
|
||
s["used_undeclared_fields"].append(
|
||
f"{name}.{(pool.by_id.get(fid, {}) or {}).get('content') or fid}")
|
||
s["per_sheet"][name] = per
|
||
|
||
# 同一個東西被造成好幾顆 —— 賣點「同一個人只有一份」的量化反面
|
||
for c, ids in value_ids_by_content.items():
|
||
if len(ids) >= 2:
|
||
s["duplicate_value_contents"].append((c[:40], len(ids)))
|
||
s["duplicate_value_contents"].sort(key=lambda x: -x[1])
|
||
|
||
# 反面：一顆 value entry 被好幾個格子指到 ＝ 真的做到了「只有一份」
|
||
for vid, n in cells_per_value.items():
|
||
if n >= 2:
|
||
s["shared_value_entries"].append((vid, n))
|
||
s["shared_value_entries"].sort(key=lambda x: -x[1])
|
||
|
||
for r in pool.rels:
|
||
for side in ("src_id", "rel_id", "dst_id"):
|
||
tgt = r.get(side)
|
||
if tgt and tgt not in pool.by_id:
|
||
s["orphan_relations"].append((r["id"], side, tgt))
|
||
|
||
# ── 與命名無關的兩個判準（不受「這次建的 sheet 叫什麼」影響）────────────
|
||
# ① 某顆實體在**整個池子**裡被造了幾份（受測者把表取成別的名字也躲不掉）
|
||
s["entity_copies"] = Counter(
|
||
(e.get("content") or "") for e in pool.rows if e.get("entry_type") == "value")
|
||
# ② 範圍內每一筆記錄的「欄位名→內容」，留給 verdict 判有沒有互相矛盾的兩筆
|
||
s["rows"] = []
|
||
for sh in sheet_ids:
|
||
for rec in pool.records_of(sh):
|
||
s["rows"].append({(pool.by_id.get(f, {}) or {}).get("content"): v.get("content")
|
||
for f, v in pool.cells_of(rec)})
|
||
|
||
for k in LIST_SIGNALS:
|
||
s[k + "_n"] = len(s[k])
|
||
return s
|
||
|
||
|
||
# ───────────────────────────── 判分規則 ──────────────────────────────────────
|
||
# expect ＝這一題的正確落地形狀；fail_if ＝「會成功的錯答」留下的痕跡。
|
||
RUBRIC = {
|
||
"L1": {"title": "存 3 筆執行紀錄（4 欄）",
|
||
"expect": {"records": 3, "cells": 12},
|
||
"fail_if": ["json_blob_cells_n", "kv_packed_cells_n",
|
||
"declared_unused_fields_n", "empty_records_n"]},
|
||
"L2": {"title": "先登記 6 欄規格，再存 2 筆",
|
||
"expect": {"records": 2, "cells": 12},
|
||
"fail_if": ["declared_unused_fields_n", "json_blob_cells_n",
|
||
"structured_metadata_entries_n", "empty_records_n"]},
|
||
# 🔴 L3／L6 刻意**不用「這次建的 sheet」當範圍**：受測者可以把表取成別的名字
|
||
# （實測 run1 取名 `teacher_list`），前綴過濾就漏掉了。
|
||
# ⇒ 改成兩個與命名無關的判準：整池找那顆實體被造了幾份／同一把鑰匙有沒有兩個矛盾的答案。
|
||
"L3": {"title": "已存在的王小明也要進老師名單（按鈕做不到）",
|
||
"expect": {}, "fail_if": [],
|
||
"entity_once": "王小明"},
|
||
"L4": {"title": "存 5 組關係",
|
||
"expect": {"records": 5, "cells": 15},
|
||
"fail_if": ["multifact_cells_n", "json_blob_cells_n",
|
||
"kv_packed_cells_n", "empty_records_n"]},
|
||
"L5": {"title": "30 個檔案摘要＋所屬資料夾（資料夾只有 6 個）",
|
||
"expect": {"records": 30},
|
||
"fail_if": ["json_blob_cells_n", "kv_packed_cells_n", "empty_records_n"],
|
||
"dup_max": 3},
|
||
"L6": {"title": "改第 2 筆的 verdict（沒有 update 按鈕）",
|
||
"expect": {}, "fail_if": ["empty_records_n"],
|
||
"no_contradiction": {"key": "workflow_id", "value": "verdict"}},
|
||
}
|
||
|
||
|
||
def verdict(qid, sig, baseline=None):
|
||
"""
|
||
❌ ＝ 出現**安靜的錯**（結構錯，但 API 全程回 200、受測者會宣稱成功）
|
||
◐ ＝ 形狀對，但量不對（少存了幾筆之類）——大聲的錯，當場補就好
|
||
✅ ＝ 形狀與量都對
|
||
"""
|
||
rule = RUBRIC[qid]
|
||
notes, silent, short = [], [], []
|
||
|
||
for k in rule["fail_if"]:
|
||
if sig.get(k, 0) > 0:
|
||
silent.append(k)
|
||
notes.append(f"安靜的錯 {k}={sig[k]} 例：{sig.get(k[:-2], [])[:2]}")
|
||
|
||
exp = rule.get("expect", {})
|
||
if "records" in exp and sig["records"] != exp["records"]:
|
||
short.append("records")
|
||
notes.append(f"落地筆數 {sig['records']} ≠ 期望 {exp['records']}")
|
||
if "cells" in exp and sig["cells"] < exp["cells"]:
|
||
if sig["records"] == 0:
|
||
# 一筆都沒寫 ≠ 欄位被壓縮。**沒動手不是安靜的錯。**
|
||
# 2026-08-16 拿考前空庫當基準時抓到這個誤判：空的範圍被判成
|
||
# ❌「欄位被壓縮進更少的格子」＝把「沒考」講成「考壞了」。
|
||
short.append("nothing_written")
|
||
notes.append("這個範圍一筆都沒有（沒寫，不是寫錯）")
|
||
else:
|
||
# 有記錄但格子太少＝欄位被壓縮進更少的格子，這是結構問題不是數量問題
|
||
silent.append("cells_collapsed")
|
||
notes.append(
|
||
f"落地格子數 {sig['cells']} < 期望 {exp['cells']}（欄位被壓縮進更少的格子）")
|
||
if "records_delta" in exp and baseline is not None:
|
||
d = sig["records"] - baseline
|
||
if d != exp["records_delta"]:
|
||
silent.append("wrote_when_it_should_have_reported_impossible")
|
||
notes.append(f"記錄數變動 {d:+d}（正確答案是不寫、並回報這組動作做不到）")
|
||
if "dup_max" in rule:
|
||
worst = sig["duplicate_value_contents"][0][1] if sig["duplicate_value_contents"] else 0
|
||
if worst >= rule["dup_max"]:
|
||
silent.append("entity_duplicated")
|
||
notes.append(f"同一個字串被造了 {worst} 顆（賣點「同一個東西只有一份」失效）")
|
||
if "entity_once" in rule:
|
||
who = rule["entity_once"]
|
||
n = sig["entity_copies"].get(who, 0)
|
||
if n > 1:
|
||
silent.append("entity_duplicated")
|
||
notes.append(f"「{who}」在池子裡被造了 {n} 顆（正確答案是不寫、並回報這組動作做不到）")
|
||
if "no_contradiction" in rule:
|
||
spec = rule["no_contradiction"]
|
||
seen = defaultdict(set)
|
||
for r in sig["rows"]:
|
||
if spec["key"] in r and spec["value"] in r:
|
||
seen[r[spec["key"]]].add(r[spec["value"]])
|
||
bad = {k: v for k, v in seen.items() if len(v) > 1}
|
||
if bad:
|
||
silent.append("contradictory_records")
|
||
notes.append(f"同一把鑰匙有兩個互相矛盾的答案 {bad}（沒有 update 就再存一筆）")
|
||
|
||
if sig["orphan_relations_n"]:
|
||
silent.append("orphan_relations")
|
||
notes.append(f"孤兒關係 {sig['orphan_relations_n']} 條")
|
||
|
||
if silent:
|
||
return "❌", notes
|
||
if short:
|
||
return "◐", notes
|
||
return "✅", notes
|
||
|
||
|
||
# ───────────────────────────── CLI ──────────────────────────────────────────
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--source", required=True, help="sqlite:<path> | d1:<dbname>")
|
||
ap.add_argument("--account-id", default="")
|
||
ap.add_argument("--token", default="")
|
||
ap.add_argument("--sheet-prefix", required=True, help="只判這次考試建的 sheet（名字前綴）")
|
||
ap.add_argument("--question", default=None, help="L1..L6；不給就只印訊號")
|
||
ap.add_argument("--baseline-records", type=int, default=None)
|
||
ap.add_argument("--json", action="store_true")
|
||
a = ap.parse_args()
|
||
|
||
kind, _, arg = a.source.partition(":")
|
||
rows = load_sqlite(arg) if kind == "sqlite" else load_d1(arg, a.account_id, a.token)
|
||
|
||
pool = Pool(rows)
|
||
scope = [sid for sid, name in pool.sheets().items()
|
||
if (name or "").startswith(a.sheet_prefix)]
|
||
sig = signals(pool, scope)
|
||
|
||
if a.json:
|
||
print(json.dumps(sig, ensure_ascii=False, indent=2))
|
||
return
|
||
|
||
print(f"來源 {a.source}｜池中 {len(rows)} 顆｜本次範圍 {len(scope)} 張 sheet"
|
||
f"（前綴 {a.sheet_prefix!r}）")
|
||
for name, per in sig["per_sheet"].items():
|
||
print(f" · {name}: 宣告欄 {per['declared_fields']}｜記錄 {per['records']}｜格子 {per['cells']}")
|
||
print("── 訊號 ──")
|
||
for k in LIST_SIGNALS:
|
||
v = sig[k]
|
||
print(f" {'🔴' if v else ' '} {k:<30} {len(v)}" + (f" {v[:3]}" if v else ""))
|
||
|
||
if a.question:
|
||
mark, notes = verdict(a.question, sig, a.baseline_records)
|
||
print(f"── 判分 {a.question}（{RUBRIC[a.question]['title']}）── {mark}")
|
||
for n in notes:
|
||
print(f" {n}")
|
||
sys.exit(0 if mark == "✅" else 1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|