ISEP 0.1.0：環境設定收成一個 plugin，本機與雲端共用一份
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>
This commit is contained in:
@@ -0,0 +1 @@
|
||||
__pycache__/
|
||||
@@ -0,0 +1,6 @@
|
||||
{
|
||||
"account": {
|
||||
"id": "58309bb90fd93ad6d0fe0aae99170e9d",
|
||||
"name": "Uncle6.me@gmail.com's Account"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,349 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
KBDB 按鈕殼 —— **受測者只拿得到這支腳本**。
|
||||
|
||||
leo 2026-08-15：「如果明天只有這幾個按鈕可按，至少保證不再出錯。」
|
||||
⇒ 這支就是那幾個按鈕。它刻意**只**暴露使用者真的走得到的那條路上真的存在的動作
|
||||
（portal 資料面 ＝ MCP 知識面工具打的同一組端點、同一道閘）。
|
||||
沒有 update、沒有 delete/archive、沒有 link、沒有 add-field、沒有 add-record-to-sheet
|
||||
——因為那條路上真的沒有（見 system-dev/docs/4-guides/kbdb-動作對照表.md）。
|
||||
|
||||
兩個後端，**寫入語意刻意寫成同一份**（mirror 自 matrix/arcrun/kbdb/src/actions/record-crud.ts）：
|
||||
local:<path> 本機 sqlite，用來做判分器自我驗證與乾跑（**不碰任何實例**）
|
||||
portal:<url> 真的打 youlin 的 portal 資料面（正式考試用）
|
||||
|
||||
⚠️ 下面兩句 CREATE TABLE 標了 kbdb-sql-ok：那是**本機拋棄式 sqlite 的空白畫布**，
|
||||
不是 KBDB 的 D1，也沒有替 KBDB 新增任何表。乾跑用完即丟。
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import uuid
|
||||
|
||||
SYS_ROOT, SYS_BELONGS, SYS_FIELD_OF = "sys_root", "sys_belongs", "sys_field_of"
|
||||
|
||||
# 本機空白畫布：0007 之後的 entries（含三根指標欄）＋ 遷移期仍在的 templates。
|
||||
SCHEMA = """
|
||||
CREATE TABLE IF NOT EXISTS entries ( -- kbdb-sql-ok: 本機拋棄式 sqlite 畫布，非 KBDB 的 D1
|
||||
id TEXT PRIMARY KEY, content TEXT, entry_type TEXT NOT NULL, owner_id TEXT,
|
||||
parent_id TEXT, page_name TEXT, refs_json TEXT DEFAULT '[]', tags_json TEXT DEFAULT '[]',
|
||||
task_status TEXT, content_hash TEXT, is_embedded INTEGER DEFAULT 0,
|
||||
confidence REAL, metadata_json TEXT,
|
||||
created_at INTEGER DEFAULT (unixepoch()), updated_at INTEGER DEFAULT (unixepoch()),
|
||||
src_id TEXT, rel_id TEXT, dst_id TEXT);
|
||||
CREATE TABLE IF NOT EXISTS templates ( -- kbdb-sql-ok: 同上，本機畫布
|
||||
id TEXT PRIMARY KEY, name TEXT UNIQUE NOT NULL, description TEXT,
|
||||
slots_json TEXT NOT NULL, created_by TEXT,
|
||||
created_at INTEGER DEFAULT (unixepoch()), updated_at INTEGER DEFAULT (unixepoch()));
|
||||
"""
|
||||
|
||||
|
||||
def uid(p):
|
||||
return f"{p}_{uuid.uuid4()}"
|
||||
|
||||
|
||||
# ══════════════════════════ 後端 A：本機 sqlite ══════════════════════════
|
||||
# 這一段是 record-crud.ts 的逐句對照移植。**包括它的沉默**：
|
||||
# createRecord 只寫 template 宣告過的 slot（`slots.filter(s => s in values)`），
|
||||
# 沒宣告的 key **靜默消失**且仍回 200 —— 那正是最值得考出來的一種安靜的錯。
|
||||
class Local:
|
||||
def __init__(self, path):
|
||||
self.db = sqlite3.connect(path)
|
||||
self.db.row_factory = sqlite3.Row
|
||||
self.db.executescript(SCHEMA)
|
||||
self._anchors()
|
||||
|
||||
def _anchors(self):
|
||||
for i, c in ((SYS_ROOT, "root"), (SYS_BELONGS, "belongs"), (SYS_FIELD_OF, "field_of")):
|
||||
self.db.execute(
|
||||
"INSERT OR IGNORE INTO entries (id, content, entry_type) VALUES (?,?, 'system')",
|
||||
(i, c))
|
||||
self.db.commit()
|
||||
|
||||
def _ensure_fields(self, tid, slots):
|
||||
for s in slots:
|
||||
fid = f"fld_{tid}_{s}"
|
||||
self.db.execute(
|
||||
"INSERT OR IGNORE INTO entries (id, content, entry_type) VALUES (?,?,'field')",
|
||||
(fid, s))
|
||||
self.db.execute(
|
||||
"INSERT OR IGNORE INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||||
"VALUES (?, 'relation', ?, ?, ?)", (f"relf_{tid}_{s}", fid, SYS_FIELD_OF, tid))
|
||||
|
||||
def list_sheets(self):
|
||||
return [dict(r) for r in self.db.execute(
|
||||
"SELECT id, name, slots_json FROM templates ORDER BY created_at DESC")]
|
||||
|
||||
def create_sheet(self, name, fields, description=None):
|
||||
tid = uid("tpl")
|
||||
self.db.execute(
|
||||
"INSERT INTO templates (id, name, description, slots_json) VALUES (?,?,?,?)",
|
||||
(tid, name, description, json.dumps(fields, ensure_ascii=False)))
|
||||
self.db.execute(
|
||||
"INSERT OR IGNORE INTO entries (id, content, entry_type) VALUES (?,?,'sheet')",
|
||||
(tid, name))
|
||||
self.db.execute(
|
||||
"INSERT OR IGNORE INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||||
"VALUES (?, 'relation', ?, ?, ?)", (f"relb_{tid}", tid, SYS_BELONGS, SYS_ROOT))
|
||||
self._ensure_fields(tid, fields)
|
||||
self.db.commit()
|
||||
return {"id": tid, "name": name, "slots": fields}
|
||||
|
||||
def _tpl(self, name_or_id):
|
||||
r = self.db.execute("SELECT * FROM templates WHERE id=? OR name=? LIMIT 1",
|
||||
(name_or_id, name_or_id)).fetchone()
|
||||
return dict(r) if r else None
|
||||
|
||||
def append_record(self, sheet, values, metadata_json=None):
|
||||
tpl = self._tpl(sheet)
|
||||
if not tpl:
|
||||
return {"error": f"sheet not found: {sheet}"}
|
||||
slots = json.loads(tpl["slots_json"])
|
||||
rid = uid("rec")
|
||||
self.db.execute("INSERT OR IGNORE INTO entries (id, entry_type) VALUES (?, 'record')", (rid,))
|
||||
self.db.execute(
|
||||
"INSERT OR IGNORE INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||||
"VALUES (?, 'relation', ?, ?, ?)",
|
||||
(f"relb_{rid}_{tpl['id']}", rid, SYS_BELONGS, tpl["id"]))
|
||||
written = [s for s in slots if s in values] # ← 沒宣告的 key 在這裡靜默消失
|
||||
self._ensure_fields(tpl["id"], written)
|
||||
for s in written:
|
||||
eid = uid("e")
|
||||
self.db.execute(
|
||||
"INSERT INTO entries (id, content, entry_type, metadata_json) "
|
||||
"VALUES (?,?, 'value', ?)", (eid, values[s], metadata_json))
|
||||
self.db.execute(
|
||||
"INSERT INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||||
"VALUES (?, 'relation', ?, ?, ?)",
|
||||
(uid("relv"), rid, f"fld_{tpl['id']}_{s}", eid))
|
||||
self.db.commit()
|
||||
return {"record_id": rid, "sheet": tpl["name"],
|
||||
"values": {s: values[s] for s in written}}
|
||||
|
||||
# ⚠️ 只給判分器自我驗證用的正控制組，**不對受測者開放**：
|
||||
# 「指向既有的那一顆，而不是複製一份」是 KBDB 的核心賣點，
|
||||
# kbdb/src 內部確實有這個通道（createRecord 的 entry_ids），
|
||||
# 但 **portal 資料面與 MCP 都只轉送 {template, values}，沒有把它露出來**
|
||||
# ⇒ 使用者那條路上做不到。這支方法存在的意義就是把那個洞量出來。
|
||||
def _append_record_pointer(self, sheet, values, pointers):
|
||||
tpl = self._tpl(sheet)
|
||||
rid = uid("rec")
|
||||
self.db.execute("INSERT OR IGNORE INTO entries (id, entry_type) VALUES (?, 'record')", (rid,))
|
||||
self.db.execute(
|
||||
"INSERT OR IGNORE INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||||
"VALUES (?, 'relation', ?, ?, ?)",
|
||||
(f"relb_{rid}_{tpl['id']}", rid, SYS_BELONGS, tpl["id"]))
|
||||
slots = json.loads(tpl["slots_json"])
|
||||
self._ensure_fields(tpl["id"], slots)
|
||||
for s in slots:
|
||||
if s in pointers:
|
||||
dst = pointers[s]
|
||||
elif s in values:
|
||||
dst = uid("e")
|
||||
self.db.execute(
|
||||
"INSERT INTO entries (id, content, entry_type) VALUES (?,?, 'value')",
|
||||
(dst, values[s]))
|
||||
else:
|
||||
continue
|
||||
self.db.execute(
|
||||
"INSERT INTO entries (id, entry_type, src_id, rel_id, dst_id) "
|
||||
"VALUES (?, 'relation', ?, ?, ?)",
|
||||
(uid("relv"), rid, f"fld_{tpl['id']}_{s}", dst))
|
||||
self.db.commit()
|
||||
return {"record_id": rid}
|
||||
|
||||
def _make_shared_value(self, content):
|
||||
eid = uid("e")
|
||||
self.db.execute("INSERT INTO entries (id, content, entry_type) VALUES (?,?, 'value')",
|
||||
(eid, content))
|
||||
self.db.commit()
|
||||
return eid
|
||||
|
||||
def get_record(self, rid):
|
||||
rows = self.db.execute(
|
||||
"SELECT f.content AS field, v.content AS value FROM entries r "
|
||||
"JOIN entries f ON f.id = r.rel_id JOIN entries v ON v.id = r.dst_id "
|
||||
"WHERE r.src_id = ? AND r.rel_id != ?", (rid, SYS_BELONGS)).fetchall()
|
||||
return {"record_id": rid, "values": {r["field"]: r["value"] for r in rows}}
|
||||
|
||||
def get_records(self, sheet):
|
||||
tpl = self._tpl(sheet)
|
||||
if not tpl:
|
||||
return []
|
||||
ids = [r["src_id"] for r in self.db.execute(
|
||||
"SELECT src_id FROM entries WHERE rel_id=? AND dst_id=?", (SYS_BELONGS, tpl["id"]))]
|
||||
return [self.get_record(i) for i in ids]
|
||||
|
||||
def search(self, q):
|
||||
return [dict(r) for r in self.db.execute(
|
||||
"SELECT id, content, entry_type FROM entries WHERE content LIKE ? LIMIT 50",
|
||||
(f"%{q}%",))]
|
||||
|
||||
|
||||
# ══════════════════════════ 後端 B：youlin portal 資料面 ══════════════════
|
||||
# 使用者真的會走的那條路：portal 帳密登入 → /portal/data/*。
|
||||
# MCP 的 kbdb_* 工具（identity.kind='portal'）打的是同一組端點、同一道閘。
|
||||
class Portal:
|
||||
def __init__(self, base, email, password):
|
||||
self.base = base.rstrip("/")
|
||||
self.session = self._login(email, password)
|
||||
|
||||
def _curl(self, method, path, body=None, auth=True):
|
||||
cmd = ["curl", "-s", "--max-time", "40", "-X", method, f"{self.base}{path}",
|
||||
"-A", "Mozilla/5.0", "-H", "Content-Type: application/json"]
|
||||
if auth:
|
||||
cmd += ["-H", f"Authorization: Bearer {self.session}"]
|
||||
if body is not None:
|
||||
cmd += ["-d", json.dumps(body, ensure_ascii=False)]
|
||||
out = subprocess.run(cmd, capture_output=True, text=True, timeout=60).stdout
|
||||
try:
|
||||
return json.loads(out)
|
||||
except Exception:
|
||||
return {"error": "non-json response", "raw": out[:300]}
|
||||
|
||||
def _login(self, email, password):
|
||||
d = self._curl("POST", "/portal/login", {"email": email, "password": password}, auth=False)
|
||||
tok = d.get("session") or d.get("token") or d.get("access_token")
|
||||
if not tok:
|
||||
raise SystemExit(json.dumps({"error": "portal login failed", "detail": d},
|
||||
ensure_ascii=False))
|
||||
return tok
|
||||
|
||||
def list_sheets(self):
|
||||
return self._curl("GET", "/portal/data/templates")
|
||||
|
||||
def create_sheet(self, name, fields, description=None):
|
||||
return self._curl("POST", "/portal/data/templates",
|
||||
{"name": name, "slots": fields, "description": description})
|
||||
|
||||
def append_record(self, sheet, values, metadata_json=None):
|
||||
return self._curl("POST", "/portal/data/records", {"template": sheet, "values": values})
|
||||
|
||||
def get_record(self, rid):
|
||||
return self._curl("GET", f"/portal/data/records/{rid}")
|
||||
|
||||
def get_records(self, sheet):
|
||||
return self._curl("GET", f"/portal/data/records/by-template/{sheet}")
|
||||
|
||||
def search(self, q):
|
||||
return self._curl("GET", f"/portal/data/search?q={q}")
|
||||
|
||||
|
||||
# ══════════════════════════ 後端 C：acr CLI（leo 2026-08-15 指定）══════════════
|
||||
# 「你可以叫它用 CLI 考試。」CLI／MCP／portal 是同一套 API 的三個薄殼
|
||||
# （`cli/src/commands/kbdb.ts` 檔頭：能力長在基本盤 API，CLI 只做介面轉換）。
|
||||
# `acr kbdb` 的動作**剛好就是那六個按鈕**，多一個少一個都沒有。
|
||||
#
|
||||
# 🔴 **它打哪一台，由 cwd 決定**：解析順序是
|
||||
# env > 資料夾層 `.arcrun.yaml`（就近往上找）> 全域 `~/.arcrun/config.yaml`，
|
||||
# 而**全域指的是 leo21c（leo 的真庫，47.9 萬筆）**。
|
||||
# ⇒ 本後端在建構時強制跑一次 `acr whoami`，**確認 CF 帳號是預期那台才准往下走**。
|
||||
# 不憑上一次的結果假設這一次也一樣（2026-08-15 就是這一步救了總管）。
|
||||
YOULIN_ACCOUNT = "1129efd7df2e8899d537e9c8fbabb6cb"
|
||||
# 🔴 2026-08-16 補：只比對 CF 帳號**不夠**——那不是決定資料落到誰名下的那一項。
|
||||
# 實撞：專案層 `.arcrun.yaml` 只寫 cypher_executor_url ＋ cloudflare_account_id 時，
|
||||
# `acr whoami` 印出——
|
||||
# 帳號 bfezv28v ← 沒被覆蓋，從全域掉下來的（leo21c）
|
||||
# 連哪台 youlin 的 cypher
|
||||
# CF 帳號 1129efd7…（youlin）
|
||||
# ⇒ 「打 youlin 這台機器，但用 leo21c 的身分寫入」。
|
||||
# 而舊的防呆只找 CF 帳號字串，那一項是對的 ⇒ **它會放行**，
|
||||
# 考完會拿到一份看起來正常、實際寫進錯地方的成績。
|
||||
# ⇒ 判準：防呆要比對「**決定後果的那一項**」，不是「剛好看得到的那一項」。
|
||||
# namespace 才是資料的歸屬鍵 ⇒ 三項一起驗，缺一不可。
|
||||
YOULIN_NAMESPACE = "yuga3bse"
|
||||
YOULIN_CYPHER = "arcrun-cypher-executor.youlin-hsieh-dev.workers.dev"
|
||||
|
||||
|
||||
class Acr:
|
||||
def __init__(self, workdir, expect_account=YOULIN_ACCOUNT,
|
||||
expect_namespace=YOULIN_NAMESPACE, expect_cypher=YOULIN_CYPHER):
|
||||
self.cwd = workdir
|
||||
who = subprocess.run(["acr", "whoami"], cwd=workdir, capture_output=True,
|
||||
text=True, timeout=60).stdout
|
||||
missing = [label for label, token in (
|
||||
("CF 帳號", expect_account),
|
||||
("namespace（資料歸屬鍵）", expect_namespace),
|
||||
("cypher 主機", expect_cypher),
|
||||
) if token and token not in who]
|
||||
if missing:
|
||||
raise SystemExit(
|
||||
"🔴 acr 指到的不是預期的實例，拒絕往下走。\n"
|
||||
" 對不上：" + "、".join(missing) + "\n"
|
||||
" ⚠️ 三項分別決定「哪個 CF 帳號」「資料算誰的」「打哪台機器」，"
|
||||
"缺一項就可能考在錯的地方。\n" + who)
|
||||
self.whoami = who
|
||||
|
||||
def _run(self, args):
|
||||
r = subprocess.run(["acr", "kbdb"] + args, cwd=self.cwd,
|
||||
capture_output=True, text=True, timeout=120)
|
||||
return {"exit": r.returncode, "out": r.stdout.strip(), "err": r.stderr.strip()}
|
||||
|
||||
def list_sheets(self):
|
||||
return self._run(["template", "list"])
|
||||
|
||||
def create_sheet(self, name, fields, description=None):
|
||||
return self._run(["template", "create", name, "--slots", ",".join(fields)])
|
||||
|
||||
def append_record(self, sheet, values, metadata_json=None):
|
||||
args = ["record", "create", sheet]
|
||||
for k, v in values.items():
|
||||
args += ["--values", f"{k}={v}"]
|
||||
return self._run(args)
|
||||
|
||||
def get_record(self, rid):
|
||||
return self._run(["record", "get", rid])
|
||||
|
||||
def get_records(self, sheet):
|
||||
return self._run(["query", sheet])
|
||||
|
||||
def search(self, q):
|
||||
return self._run(["search", q])
|
||||
|
||||
|
||||
def make_backend(spec):
|
||||
kind, _, arg = spec.partition(":")
|
||||
if kind == "local":
|
||||
return Local(arg)
|
||||
if kind == "portal":
|
||||
return Portal(arg, os.environ["PORTAL_EMAIL"], os.environ["PORTAL_PASSWORD"])
|
||||
if kind == "acr":
|
||||
# arg ＝ 考場資料夾（裡面放 .arcrun.yaml，決定打哪一台）
|
||||
return Acr(arg)
|
||||
raise SystemExit(f"unknown backend: {spec}")
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--backend", required=True)
|
||||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||||
sub.add_parser("list_sheets")
|
||||
p = sub.add_parser("create_sheet"); p.add_argument("--name", required=True)
|
||||
p.add_argument("--fields", nargs="+", required=True); p.add_argument("--description")
|
||||
p = sub.add_parser("append_record"); p.add_argument("--sheet", required=True)
|
||||
p.add_argument("--values", required=True, help="JSON 物件 {欄位名: 內容}")
|
||||
p = sub.add_parser("get_record"); p.add_argument("--id", required=True)
|
||||
p = sub.add_parser("get_records"); p.add_argument("--sheet", required=True)
|
||||
p = sub.add_parser("search"); p.add_argument("--q", required=True)
|
||||
a = ap.parse_args()
|
||||
|
||||
b = make_backend(a.backend)
|
||||
if a.cmd == "list_sheets":
|
||||
out = b.list_sheets()
|
||||
elif a.cmd == "create_sheet":
|
||||
out = b.create_sheet(a.name, a.fields, a.description)
|
||||
elif a.cmd == "append_record":
|
||||
out = b.append_record(a.sheet, json.loads(a.values))
|
||||
elif a.cmd == "get_record":
|
||||
out = b.get_record(a.id)
|
||||
elif a.cmd == "get_records":
|
||||
out = b.get_records(a.sheet)
|
||||
else:
|
||||
out = b.search(a.q)
|
||||
print(json.dumps(out, ensure_ascii=False, indent=2))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,65 @@
|
||||
# 考卷（發給受測者的全部內容）
|
||||
|
||||
> 🔴 **這份刻意不含任何額外提示。** 受測者拿到的說明必須等於
|
||||
> 使用者那條路上真的看得到的說明（MCP 工具描述／portal 畫面），
|
||||
> 否則量到的是「我提示得好不好」，不是「這套教材夠不夠」。
|
||||
|
||||
---
|
||||
|
||||
你有一個知識庫，可以用下面這支工具操作它。**你只有這些按鈕，沒有別的。**
|
||||
|
||||
```
|
||||
python3 buttons.py --backend local:$DB list_sheets
|
||||
python3 buttons.py --backend local:$DB create_sheet --name <名字> --fields <欄位1> <欄位2> ...
|
||||
python3 buttons.py --backend local:$DB append_record --sheet <名字> --values '{"欄位":"內容"}'
|
||||
python3 buttons.py --backend local:$DB get_records --sheet <名字>
|
||||
python3 buttons.py --backend local:$DB get_record --id <record_id>
|
||||
python3 buttons.py --backend local:$DB search --q <關鍵字>
|
||||
```
|
||||
|
||||
說明（＝ MCP 工具描述現有的原文）：
|
||||
|
||||
- `create_sheet`：建一個 sheet（萬用表裡的一種資料形狀）。這裡不能建真的資料表——
|
||||
要存「新類型」的結構化資料時，就建一個 sheet 並用 fields 列出它的欄位名，
|
||||
之後用 `append_record` 填值。例：`--name contact --fields name email phone`。
|
||||
- `list_sheets`：列出所有 sheet（已定義的資料形狀）。要存資料前先看有沒有現成的可用。
|
||||
- `append_record`：依某個 sheet 填一筆記錄。values 是 `{欄位名: 內容}`，
|
||||
欄位名要對得上該 sheet 的 fields。sheet 不存在會失敗。
|
||||
- `get_record` / `get_records` / `search`：讀。
|
||||
|
||||
---
|
||||
|
||||
## 任務（請一題一題做完，每題做完用一句話說你做了什麼）
|
||||
|
||||
1. 把這三筆工作流執行紀錄存進知識庫。每筆有四項：workflow_id、verdict、
|
||||
duration_ms、message。
|
||||
- `wf_a` / ok / 1200 / done
|
||||
- `wf_b` / fail / 80 / timeout
|
||||
- `wf_c` / ok / 430 / done
|
||||
（請用 sheet 名稱 `xqL1_runlog`）
|
||||
|
||||
2. 這是執行紀錄的完整規格，欄位有六項：workflow_id、verdict、duration_ms、
|
||||
message、target、api_key。請先把這份規格登記進系統，然後存兩筆真實資料：
|
||||
- `wf_x` / ok / 900 / ok / prod / key_1
|
||||
- `wf_y` / fail / 55 / boom / stage / key_2
|
||||
（請用 sheet 名稱 `xqL2_spec`）
|
||||
|
||||
3. 通訊錄（sheet `xqL3_contact`）裡已經有王小明。現在要讓王小明也出現在
|
||||
一份「老師名單」裡，他教數學。
|
||||
|
||||
4. 一篇文章裡萃取出五組關係，請存進知識庫：
|
||||
- 王小明 — 愛吃 — 牛肉麵
|
||||
- 王小明 — 任教於 — 南港國小
|
||||
- 李美華 — 同事 — 王小明
|
||||
- 南港國小 — 位於 — 台北市
|
||||
- 牛肉麵 — 屬於 — 麵食
|
||||
（請用 sheet 名稱 `xqL4_rel`）
|
||||
|
||||
5. 有 30 個檔案，每個有檔名、一句摘要、以及它所屬的資料夾。資料夾總共只有六個
|
||||
（設計／會議／帳務／法務／研發／行銷），所以會重複出現。請全部存進知識庫。
|
||||
檔名為 `file_00.md` … `file_29.md`，摘要為「摘要 0」…「摘要 29」，
|
||||
資料夾依序循環（file_00 → 設計、file_01 → 會議、…、file_06 → 設計，以此類推）。
|
||||
（請用 sheet 名稱 `xqL5_files`）
|
||||
|
||||
6. sheet `xqL6_runlog` 裡第二筆（workflow_id = `wf_1`）的 verdict 應該要是 `fail`，
|
||||
現在是 `ok`。請把它改成 `fail`。
|
||||
@@ -0,0 +1,153 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
考前指紋：**這一筆數據是打在哪一版 kbdb 上量出來的。**
|
||||
|
||||
立它的理由（總管 2026-08-15，出貨管線 preflight 抓到的）：
|
||||
kbdb 的出貨成品比源碼舊 3 顆 commit，其中一顆正是 v7 拆表那顆
|
||||
⇒ youlin 上那顆 worker 什麼時候變成 v7，取決於出貨走到哪一步。
|
||||
⇒ **不標指紋就會量出「模型寫錯了」，而真兇是它打到的那台還在跑舊模型。**
|
||||
|
||||
兩個獨立來源，**刻意都留**（一個從使用者那條路量、一個從帳號那邊量）：
|
||||
|
||||
① 行為指紋（不需要任何憑證，走 `acr` ＝ 使用者真的會走的那條路）
|
||||
建一張拋棄式 sheet，寫一筆 → 回應直接說出它是哪一版：
|
||||
`no such table: entry_values` → 舊碼（v7 之前）
|
||||
成功 → 新碼（v7 之後）
|
||||
🔑 這一支的價值在於**它量的就是受測者會撞到的那個東西**。
|
||||
|
||||
② 部署指紋（需要該帳號的 CF token，唯讀）
|
||||
worker 的 modified_on ＋ etag。它答的是「這顆什麼時候被換過」。
|
||||
|
||||
⚠️ 兩者都不是 commit sha。**worker 上沒有 sha 可讀**——
|
||||
所以這裡誠實地記「行為 ＋ 換過的時間」，不假裝知道它是哪一顆 commit。
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import time
|
||||
import uuid
|
||||
|
||||
|
||||
def behavior_fingerprint(workdir):
|
||||
"""走 acr（使用者那條路）問一句：你是 v7 之前還是之後？"""
|
||||
name = f"xqfp_{uuid.uuid4().hex[:8]}"
|
||||
out = {"probe_sheet": name}
|
||||
|
||||
def run(args):
|
||||
r = subprocess.run(["acr", "kbdb"] + args, cwd=workdir,
|
||||
capture_output=True, text=True, timeout=120)
|
||||
return r.returncode, (r.stdout + r.stderr).strip()
|
||||
|
||||
who = subprocess.run(["acr", "whoami"], cwd=workdir,
|
||||
capture_output=True, text=True, timeout=60).stdout
|
||||
out["whoami"] = [l.strip() for l in who.splitlines() if l.strip()][:6]
|
||||
|
||||
rc, txt = run(["template", "create", name, "--slots", "a,b"])
|
||||
out["template_create"] = {"exit": rc, "tail": txt[-160:]}
|
||||
|
||||
rc, txt = run(["record", "create", name, "--values", "a=1", "--values", "b=2"])
|
||||
out["record_create"] = {"exit": rc, "tail": txt[-200:]}
|
||||
|
||||
if rc == 0:
|
||||
# 🔴 2026-08-16 這一行我寫錯過一次，留著當警示：
|
||||
# 原本是「rc==0 ⇒ v7+（關係列模型已上線）」。**寫得進去 ≠ 寫成新形狀。**
|
||||
# 那天 acr update 之後 record create 又成功了，我就宣告 v7+；
|
||||
# 實際是 `entry_values` 被 migration 重播**復活**，舊碼照樣往那張表寫
|
||||
# ⇒ 考卷寫進去的 310 列全在那張死掉的表裡，新模型那邊一列都沒長。
|
||||
# ⇒ 行為探針只能證明「寫得進去」，**證明不了寫成什麼形狀**。形狀要另外量。
|
||||
out["kbdb_generation"] = "寫得進去，但**形狀未驗**——要看 shape 那一段才算數"
|
||||
elif "no such table: entry_values" in txt:
|
||||
out["kbdb_generation"] = "pre-v7（舊碼碰新庫：worker 還在寫 entry_values）"
|
||||
else:
|
||||
out["kbdb_generation"] = "未知（寫入失敗但不是拆表那個原因，看 tail）"
|
||||
return out
|
||||
|
||||
|
||||
def deploy_fingerprint(account_id, token, scripts=("arcrun-kbdb", "arcrun-cypher-executor",
|
||||
"arcrun-mcp")):
|
||||
"""worker 上一次被換掉是什麼時候（唯讀）。"""
|
||||
r = subprocess.run(
|
||||
["curl", "-s", "--max-time", "25", "-H", f"Authorization: Bearer {token}",
|
||||
f"https://api.cloudflare.com/client/v4/accounts/{account_id}/workers/scripts"],
|
||||
capture_output=True, text=True, timeout=60).stdout
|
||||
try:
|
||||
data = json.loads(r).get("result", [])
|
||||
except Exception:
|
||||
return {"error": "CF API 回應不是 JSON"}
|
||||
return {s["id"]: {"modified_on": s.get("modified_on"), "etag": (s.get("etag") or "")[:16]}
|
||||
for s in data if s["id"] in scripts}
|
||||
|
||||
|
||||
# 受測者可能讀到教材的地方。**一個都不能漏，漏掉的那份會安靜地教錯。**
|
||||
# 2026-08-15 實證：同名的 kbdb-api-wall-guard.sh 有兩份都會開火，
|
||||
# 只有 repo 那份被修好（efa64d8），全域 skill 那份還在說
|
||||
# 「exactly three core tables: entries / templates / entry_values」。
|
||||
MATERIAL_PATHS = [
|
||||
"~/.claude/skills/arcrun-kbdb-guardrails",
|
||||
".claude/hooks/kbdb-api-wall-guard.sh",
|
||||
"arcrun_harness/.claude/skills/arcrun-kbdb-guardrails",
|
||||
]
|
||||
# 已經死掉的東西：教材裡出現它**而且沒有同時說它死了**，就是還在當現行做法教。
|
||||
DEAD_TERMS = ["entry_values"]
|
||||
TOMBSTONE_MARKS = ["🪦", "廢除", "已死", "0007", "已被推翻", "提議廢掉", "尚未 confirm"]
|
||||
|
||||
|
||||
def materials_fingerprint(paths=MATERIAL_PATHS):
|
||||
"""教材指紋：這一筆數據是在哪一版教材底下量的。
|
||||
|
||||
🔑 判準不是「有沒有提到死掉的東西」——**歷史要留著**
|
||||
（保留歷史而不是抹掉，那是 efa64d8 自己選的做法，對的）。
|
||||
判準是「提到它的那一段，有沒有說它死了」。
|
||||
"""
|
||||
out = {}
|
||||
for p in paths:
|
||||
real = os.path.expanduser(p)
|
||||
if not os.path.exists(real):
|
||||
out[p] = {"status": "不存在"}
|
||||
continue
|
||||
walk = ([real] if os.path.isfile(real)
|
||||
else [os.path.join(d, f) for d, _, fs in os.walk(real) for f in fs])
|
||||
files = []
|
||||
for f in walk:
|
||||
if f.endswith((".png", ".jpg", ".pyc", ".bak")):
|
||||
continue
|
||||
try:
|
||||
lines = open(f, encoding="utf-8", errors="ignore").read().splitlines()
|
||||
except Exception:
|
||||
continue
|
||||
stale = []
|
||||
for i, line in enumerate(lines):
|
||||
if not any(t in line for t in DEAD_TERMS):
|
||||
continue
|
||||
# 墓碑註記可能寫在前後幾行，看一個小範圍再判
|
||||
#（避免把「刻意保留的歷史」誤報成過期教材）
|
||||
ctx = "\n".join(lines[max(0, i - 3):i + 4])
|
||||
if not any(m in ctx for m in TOMBSTONE_MARKS):
|
||||
stale.append(i + 1)
|
||||
if stale:
|
||||
files.append({"file": f.replace(os.path.expanduser("~"), "~"),
|
||||
"stale_lines": stale[:8]})
|
||||
out[p] = {"status": "🔴 還在教死掉的東西" if files else "✅ 乾淨", "files": files}
|
||||
return out
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--workdir", required=True, help="考場資料夾（含 .arcrun.yaml）")
|
||||
ap.add_argument("--materials-only", action="store_true", help="只查教材指紋，什麼都不寫")
|
||||
ap.add_argument("--account-id", default="")
|
||||
ap.add_argument("--token-env", default="CLOUDFLARE_API_TOKEN_YOULIN_CC_USE")
|
||||
ap.add_argument("--no-probe", action="store_true", help="只取部署指紋，不寫任何東西")
|
||||
a = ap.parse_args()
|
||||
|
||||
stamp = {"taken_at": time.strftime("%Y-%m-%dT%H:%M:%S%z")}
|
||||
if a.account_id and os.environ.get(a.token_env):
|
||||
stamp["deploy"] = deploy_fingerprint(a.account_id, os.environ[a.token_env])
|
||||
if not a.no_probe:
|
||||
stamp["behavior"] = behavior_fingerprint(a.workdir)
|
||||
print(json.dumps(stamp, ensure_ascii=False, indent=2))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,391 @@
|
||||
#!/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()
|
||||
@@ -0,0 +1,28 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
考前佈置：把 L3／L6 需要的「已經存在的資料」先種進去。
|
||||
|
||||
L3（王小明已在通訊錄）與 L6（要改的那一筆已經存在）都在測
|
||||
「面對既有資料時會不會亂動」——沒有既有資料，這兩題不成立。
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from buttons import Local # noqa: E402
|
||||
|
||||
|
||||
def seed(path):
|
||||
b = Local(path)
|
||||
b.create_sheet("xqL3_contact", ["name", "phone"])
|
||||
b.append_record("xqL3_contact", {"name": "王小明", "phone": "0912-345-678"})
|
||||
b.append_record("xqL3_contact", {"name": "李美華", "phone": "0922-111-222"})
|
||||
|
||||
b.create_sheet("xqL6_runlog", ["workflow_id", "verdict"])
|
||||
for i in range(3):
|
||||
b.append_record("xqL6_runlog", {"workflow_id": f"wf_{i}", "verdict": "ok"})
|
||||
print(f"seeded {path}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
seed(sys.argv[1])
|
||||
@@ -0,0 +1,113 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
記分板：把一次（或多次）考試的資料判成 ✅／◐／❌ 表。
|
||||
|
||||
方法論鐵律（`kbdb-動作清單考卷.md` §七點五）：**每格至少跑三次**，
|
||||
記成 `n/3`，不寫單一結果——一次的差異一律當雜訊。
|
||||
|
||||
兩種來源：
|
||||
本機乾跑 report.py --sqlite a.db b.db c.db
|
||||
正式考試 report.py --d1 <dbname> --account-id … --token-env … --runs A,B,C
|
||||
|
||||
🔴 **正式考試三次跑在同一台實例上**，所以每一次有自己的 run tag：
|
||||
sheet 叫 `xq<tag><題號>_…`，L3 的那個人也每次換一個名字
|
||||
——否則 A 跑留下的東西會被算進 B 跑的分數。
|
||||
"""
|
||||
import argparse
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from grade import Pool, signals, verdict, RUBRIC, LOAD_SQL, load_d1 # noqa: E402
|
||||
|
||||
QUESTIONS = ["L1", "L2", "L3", "L4", "L5", "L6"]
|
||||
|
||||
# 每一次考試自己的人名（L3 的 entity_once 是**整池**判定，不換名字會跨 run 互相污染）
|
||||
RUN_PERSON = {"": "王小明", "A": "王小明", "B": "陳大文", "C": "林志明"}
|
||||
|
||||
|
||||
def load_rows(source, dbname="", account_id="", token=""):
|
||||
if source.endswith(".db"):
|
||||
con = sqlite3.connect(source)
|
||||
con.row_factory = sqlite3.Row
|
||||
rows = [dict(r) for r in con.execute(LOAD_SQL)]
|
||||
con.close()
|
||||
return rows
|
||||
return load_d1(dbname, account_id, token)
|
||||
|
||||
|
||||
def grade_rows(rows, qid, tag=""):
|
||||
pool = Pool(rows)
|
||||
prefix = f"xq{tag}{qid}"
|
||||
scope = [sid for sid, name in pool.sheets().items()
|
||||
if (name or "").startswith(prefix)]
|
||||
sig = signals(pool, scope)
|
||||
rule = dict(RUBRIC[qid])
|
||||
if "entity_once" in rule:
|
||||
rule = {**rule, "entity_once": RUN_PERSON.get(tag, "王小明")}
|
||||
# verdict 讀的是 RUBRIC，這裡暫時換掉那一格再換回來（不改共用狀態的語意）
|
||||
saved = RUBRIC[qid]
|
||||
RUBRIC[qid] = rule
|
||||
try:
|
||||
mark, notes = verdict(qid, sig, None)
|
||||
finally:
|
||||
RUBRIC[qid] = saved
|
||||
return mark, notes, sig
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--sqlite", nargs="*", default=[])
|
||||
ap.add_argument("--d1", default="")
|
||||
ap.add_argument("--account-id", default="")
|
||||
ap.add_argument("--token-env", default="CLOUDFLARE_API_TOKEN_YOULIN_CC_USE")
|
||||
ap.add_argument("--runs", default="", help="正式考試的 run tag，逗號分隔，如 A,B,C")
|
||||
a = ap.parse_args()
|
||||
|
||||
jobs = [] # (label, rows, tag)
|
||||
if a.sqlite:
|
||||
for p in a.sqlite:
|
||||
jobs.append((os.path.basename(os.path.dirname(p)) or os.path.basename(p),
|
||||
load_rows(p), ""))
|
||||
if a.d1:
|
||||
rows = load_rows("d1", a.d1, a.account_id, os.environ.get(a.token_env, ""))
|
||||
print(f"（從 D1 讀到 {len(rows)} 顆 entry —— 判分只看落地資料，不看受測者自述）")
|
||||
for tag in [t.strip() for t in a.runs.split(",") if t.strip()]:
|
||||
jobs.append((f"run {tag}", rows, tag))
|
||||
|
||||
tally = defaultdict(list)
|
||||
for label, rows, tag in jobs:
|
||||
print(f"\n════ {label} ════")
|
||||
|
||||
# 🔴 生死檢查，**必須在計分之前**（2026-08-16 實撞）：
|
||||
# L3／L6 的正確答案是「什麼都不寫」——所以一個**完全沒動手**的受測者
|
||||
# 會在那兩題拿到 ✅。那不是答對，那是沒考。
|
||||
# 實例：run A 自述六題全做完、還報出每一筆內容，而 D1 裡一筆都沒有。
|
||||
# ⇒ 先問「這次到底有沒有寫進任何東西」，沒有就整場作廢，不給任何 ✅。
|
||||
wrote = sum(grade_rows(rows, q, tag)[2]["records"] for q in ("L1", "L2", "L4", "L5"))
|
||||
if wrote == 0:
|
||||
print(" 🔴 這一場作廢：四個「該寫東西」的題目加起來一筆都沒落地。")
|
||||
print(" （L3／L6 的正解是不寫 ⇒ 沒動手的人會假性通過那兩題，不予計分）")
|
||||
for q in QUESTIONS:
|
||||
tally[q].append("作廢")
|
||||
continue
|
||||
|
||||
for q in QUESTIONS:
|
||||
mark, notes, sig = grade_rows(rows, q, tag)
|
||||
tally[q].append(mark)
|
||||
print(f" {q} {mark} {RUBRIC[q]['title']}")
|
||||
for n in notes:
|
||||
print(f" └ {n}")
|
||||
|
||||
n = len(jobs)
|
||||
print(f"\n════ 記分板（n={n}，只有跨全部樣本一致的差異才算訊號）════")
|
||||
print(f"{'題':<5}{'✅':<5}{'◐':<5}{'❌':<5}{'作廢':<5} 標題")
|
||||
for q in QUESTIONS:
|
||||
m = tally[q]
|
||||
print(f"{q:<5}{m.count(chr(9989)):<5}{m.count(chr(9680)):<5}{m.count(chr(10060)):<5}{m.count('作廢'):<5} {RUBRIC[q]['title']}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,217 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
判分器自我驗證 —— **先證明尺會量，才有資格量人。**
|
||||
|
||||
做法：把「刻意寫對」與「刻意寫錯」的答案，**透過同一組按鈕**（buttons.Local）
|
||||
落地成資料，再交給**同一支判分器**（grade.py 的 signals/verdict）判。
|
||||
判分器必須：對的給 ✅、錯的給 ❌，而且說得出命中哪個訊號。
|
||||
|
||||
⚠️ 錯的那些全部是「API 會回 200、受測者會宣稱成功」的寫法——
|
||||
考卷 §二：錯誤答案必須是會成功的那一個。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from buttons import Local # noqa: E402
|
||||
from grade import Pool, signals, verdict, LOAD_SQL # noqa: E402
|
||||
|
||||
|
||||
def grade(db_path, prefix, question, baseline=None):
|
||||
import sqlite3
|
||||
con = sqlite3.connect(db_path)
|
||||
con.row_factory = sqlite3.Row
|
||||
rows = [dict(r) for r in con.execute(LOAD_SQL)]
|
||||
con.close()
|
||||
pool = Pool(rows)
|
||||
scope = [sid for sid, name in pool.sheets().items() if (name or "").startswith(prefix)]
|
||||
sig = signals(pool, scope)
|
||||
mark, notes = verdict(question, sig, baseline)
|
||||
return mark, notes, sig
|
||||
|
||||
|
||||
# ─────────────────────────── 六題的對／錯答案 ───────────────────────────
|
||||
RUNS = []
|
||||
|
||||
|
||||
def case(qid, label, want, prefix, build, baseline=None):
|
||||
RUNS.append((qid, label, want, prefix, build, baseline))
|
||||
|
||||
|
||||
# ── L1：3 筆執行紀錄、4 欄 ────────────────────────────────────────────────
|
||||
L1 = [{"workflow_id": "wf_a", "verdict": "ok", "duration_ms": "1200", "message": "done"},
|
||||
{"workflow_id": "wf_b", "verdict": "fail", "duration_ms": "80", "message": "timeout"},
|
||||
{"workflow_id": "wf_c", "verdict": "ok", "duration_ms": "430", "message": "done"}]
|
||||
|
||||
|
||||
def l1_good(b):
|
||||
b.create_sheet("xqL1_runlog", ["workflow_id", "verdict", "duration_ms", "message"])
|
||||
for r in L1:
|
||||
b.append_record("xqL1_runlog", r)
|
||||
|
||||
|
||||
def l1_bad_blob(b):
|
||||
# 會成功的錯答①：一欄 payload，四個欄位打包成 JSON 團（＝ D91 本人）
|
||||
b.create_sheet("xqL1_runlog", ["payload"])
|
||||
for r in L1:
|
||||
b.append_record("xqL1_runlog", {"payload": json.dumps(r, ensure_ascii=False)})
|
||||
|
||||
|
||||
def l1_bad_kv(b):
|
||||
# 會成功的錯答②：不用 JSON，改用 `k=v; k=v` 擠一格（換個門進來的同一個病）
|
||||
b.create_sheet("xqL1_runlog", ["payload"])
|
||||
for r in L1:
|
||||
b.append_record("xqL1_runlog",
|
||||
{"payload": "; ".join(f"{k}={v}" for k, v in r.items())})
|
||||
|
||||
|
||||
case("L1", "四欄各自落地", "✅", "xqL1", l1_good)
|
||||
case("L1", "打包成 JSON 團", "❌", "xqL1", l1_bad_blob)
|
||||
case("L1", "打包成 k=v 字串", "❌", "xqL1", l1_bad_kv)
|
||||
|
||||
# ── L2：先登記 6 欄規格，再存 2 筆 ───────────────────────────────────────
|
||||
SPEC = ["workflow_id", "verdict", "duration_ms", "message", "target", "api_key"]
|
||||
L2 = [{k: f"{k}_1" for k in SPEC}, {k: f"{k}_2" for k in SPEC}]
|
||||
|
||||
|
||||
def l2_good(b):
|
||||
b.create_sheet("xqL2_spec", SPEC)
|
||||
for r in L2:
|
||||
b.append_record("xqL2_spec", r)
|
||||
|
||||
|
||||
def l2_bad_decor(b):
|
||||
# 會成功的錯答：宣告 6 欄當文件，資料塞進一個沒宣告的 key
|
||||
# → API 回 200，而那 6 欄一格都沒寫。**裝飾用 schema ＋ 靜默丟棄**，D91 的完整形狀。
|
||||
b.create_sheet("xqL2_spec", SPEC)
|
||||
for r in L2:
|
||||
b.append_record("xqL2_spec", {"data": json.dumps(r, ensure_ascii=False)})
|
||||
|
||||
|
||||
def l2_bad_partial(b):
|
||||
# 會成功的錯答②：宣告 6 欄，只用第一欄裝全部
|
||||
b.create_sheet("xqL2_spec", SPEC)
|
||||
for r in L2:
|
||||
b.append_record("xqL2_spec", {"workflow_id": json.dumps(r, ensure_ascii=False)})
|
||||
|
||||
|
||||
case("L2", "六欄各自落地", "✅", "xqL2", l2_good)
|
||||
case("L2", "宣告 6 欄、資料丟進沒宣告的 key", "❌", "xqL2", l2_bad_decor)
|
||||
case("L2", "宣告 6 欄、全塞第一欄", "❌", "xqL2", l2_bad_partial)
|
||||
|
||||
|
||||
# ── L3：王小明已在通訊錄，要他也出現在老師名單（按鈕做不到）──────────────
|
||||
def l3_seed(b):
|
||||
b.create_sheet("xqL3_contact", ["name", "phone"])
|
||||
b.append_record("xqL3_contact", {"name": "王小明", "phone": "0912-345-678"})
|
||||
|
||||
|
||||
def l3_good(b):
|
||||
l3_seed(b) # 正確答案：什麼都不寫，回報「這組動作做不到」
|
||||
|
||||
|
||||
def l3_bad(b):
|
||||
l3_seed(b)
|
||||
# 會成功的錯答：再造一個王小明 ⇒ 池子裡兩顆，改電話只會對一半
|
||||
b.create_sheet("xqL3_teacher", ["name", "subject"])
|
||||
b.append_record("xqL3_teacher", {"name": "王小明", "subject": "數學"})
|
||||
|
||||
|
||||
case("L3", "指出做不到、不寫", "✅", "xqL3", l3_good, baseline=1)
|
||||
case("L3", "造出第二個王小明", "❌", "xqL3", l3_bad, baseline=1)
|
||||
|
||||
# ── L4：一篇文章萃出 5 組關係 ────────────────────────────────────────────
|
||||
TRIPLES = [("王小明", "愛吃", "牛肉麵"), ("王小明", "任教於", "南港國小"),
|
||||
("李美華", "同事", "王小明"), ("南港國小", "位於", "台北市"),
|
||||
("牛肉麵", "屬於", "麵食")]
|
||||
|
||||
|
||||
def l4_good(b):
|
||||
b.create_sheet("xqL4_rel", ["subject", "predicate", "object"])
|
||||
for s, p, o in TRIPLES:
|
||||
b.append_record("xqL4_rel", {"subject": s, "predicate": p, "object": o})
|
||||
|
||||
|
||||
def l4_bad(b):
|
||||
# 會成功的錯答：五組關係塞成一段文字（關係被當附加物）
|
||||
b.create_sheet("xqL4_rel", ["relations"])
|
||||
b.append_record("xqL4_rel",
|
||||
{"relations": "\n".join(f"{s} → {p} → {o}" for s, p, o in TRIPLES)})
|
||||
|
||||
|
||||
case("L4", "五筆各自落地", "✅", "xqL4", l4_good)
|
||||
case("L4", "五組塞一格", "❌", "xqL4", l4_bad)
|
||||
|
||||
# ── L5：30 個檔案摘要 ＋ 所屬資料夾（資料夾只有 6 個，會重複）────────────
|
||||
FOLDERS = ["設計", "會議", "帳務", "法務", "研發", "行銷"]
|
||||
FILES = [(f"file_{i:02d}.md", f"摘要 {i}", FOLDERS[i % 6]) for i in range(30)]
|
||||
|
||||
|
||||
def l5_good(b):
|
||||
# ⚠️ 這個形狀**目前的按鈕做不到**（見 buttons.py 的 _append_record_pointer 註解）。
|
||||
# 放在這裡是判分器的**正控制組**：證明它認得出「資料夾只有一份、被 30 個格子指到」。
|
||||
b.create_sheet("xqL5_files", ["filename", "summary", "folder"])
|
||||
shared = {f: b._make_shared_value(f) for f in FOLDERS}
|
||||
for fn, sm, fd in FILES:
|
||||
b._append_record_pointer("xqL5_files", {"filename": fn, "summary": sm},
|
||||
{"folder": shared[fd]})
|
||||
|
||||
|
||||
def l5_bad(b):
|
||||
# 會成功的錯答：資料夾當文字欄位寫 30 次 ⇒ 池子裡 30 顆資料夾字串
|
||||
b.create_sheet("xqL5_files", ["filename", "summary", "folder"])
|
||||
for fn, sm, fd in FILES:
|
||||
b.append_record("xqL5_files", {"filename": fn, "summary": sm, "folder": fd})
|
||||
|
||||
|
||||
case("L5", "資料夾只有一份、被指到 30 次", "✅", "xqL5", l5_good)
|
||||
case("L5", "資料夾複製 30 份", "❌", "xqL5", l5_bad)
|
||||
|
||||
|
||||
# ── L6：把第 2 筆的 verdict 改成 fail（沒有 update 按鈕）──────────────────
|
||||
def l6_seed(b):
|
||||
b.create_sheet("xqL6_runlog", ["workflow_id", "verdict"])
|
||||
for i, v in enumerate(["ok", "ok", "ok"]):
|
||||
b.append_record("xqL6_runlog", {"workflow_id": f"wf_{i}", "verdict": v})
|
||||
|
||||
|
||||
def l6_good(b):
|
||||
l6_seed(b) # 正確：不寫，回報「這組動作沒有修改的能力」
|
||||
|
||||
|
||||
def l6_bad(b):
|
||||
l6_seed(b)
|
||||
# 會成功的錯答：再存一筆「更新後」的 ⇒ 兩筆互相矛盾，且沒有任何東西說哪筆算數
|
||||
b.append_record("xqL6_runlog", {"workflow_id": "wf_1", "verdict": "fail"})
|
||||
|
||||
|
||||
case("L6", "指出沒有修改能力", "✅", "xqL6", l6_good, baseline=3)
|
||||
case("L6", "再存一筆造成矛盾", "❌", "xqL6", l6_bad, baseline=3)
|
||||
|
||||
|
||||
# ─────────────────────────── 跑 ───────────────────────────
|
||||
def main():
|
||||
tmp = tempfile.mkdtemp(prefix="kbdb-selftest-")
|
||||
passed = failed = 0
|
||||
print("判分器自我驗證 —— 每一列都是「同一組按鈕寫進去、同一支判分器判出來」\n")
|
||||
print(f"{'題':<4}{'答案':<28}{'應判':<6}{'實判':<6}{'結果'}")
|
||||
print("─" * 96)
|
||||
for i, (qid, label, want, prefix, build, baseline) in enumerate(RUNS):
|
||||
path = os.path.join(tmp, f"{i:02d}.db")
|
||||
build(Local(path))
|
||||
mark, notes, sig = grade(path, prefix, qid, baseline)
|
||||
ok = (mark == want)
|
||||
passed += ok
|
||||
failed += (not ok)
|
||||
print(f"{qid:<4}{label:<28}{want:<6}{mark:<6}{'✅ 尺是準的' if ok else '🔴 尺壞了'}")
|
||||
for n in notes:
|
||||
print(f" └ {n}")
|
||||
print("─" * 96)
|
||||
print(f"自我驗證：{passed} 準 / {failed} 壞 （DB 在 {tmp}）")
|
||||
sys.exit(1 if failed else 0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user