這個系列從 Day 3 開始，我大概每隔幾天就會寫一次「每一次都落檔」。

那是被咬過的。Day 3 那篇稿子在校對時，有一段因為只跑在終端機裡沒落檔，被判成「無法核實」打回來重做。從那之後 experiments/ 底下每支腳本都會寫 results.jsonl，連原始 HTTP 回應都整份存。這習慣救過我很多次，這幾天寫 Day 23、24，page_61 的秒數和 finish_reason 全部是直接從那些 jsonl 撈出來的，一個都不用靠記憶。

但實驗腳本的 jsonl 跟 Harness 需要的紀錄，是兩種東西。

實驗的 jsonl 一支腳本一份，欄位隨我當天的心情。Harness 的紀錄要回答跨文件、跨時間的問題：R1 觸發以後，切出來的片段有幾成讀成功？負號錯誤都出在哪種區塊？人工說模型錯了的那些位置，logprob 當時是多少？

Day 20 寫 reason 欄位時就說過，規則編號是為了讓 Day 26 能算「R3 觸發了幾次、之後成功幾次」。昨天的飛輪也在等這些答案。

現在手上已經有兩份 JSONL

組完 Day 24，Harness 跑一份文件會留下兩份東西：

• commander.jsonl：Day 20 的 _record 寫的決策紀錄，每行一個 dispatch、result、decide、escalate 或 no_rule
• Recorder 收到的信封：Day 23 的 Envelope，每一次派工、每一份 SubagentResult 各一則

加上 Day 25 的 DiffEvent，和之後人工稽核的判定。四種東西，要一起查。

我的決定是 JSONL 跟 SQLite 都用，分工講得很死：

為什麼不只用 SQLite？它很可靠，但 schema 一改舊資料就得遷移，遷移腳本寫錯一行，舊紀錄就被改掉，而且你不一定知道。

JSONL 的笨是一種保護。每行寫下去就是那樣，三年後的程式讀不懂某個欄位，那一行至少還在。

一則信封落地長什麼樣

Recorder 寫的每一行，最外層固定三個欄位（schema_version、event_type、recorded_at），後面接 Envelope.model_dump(mode="json") 的內容。一則 Reader 回報撞上限的結果訊息，排開來看是這樣：

{
  "schema_version": 2,
  "event_type": "message",
  "recorded_at": "2026-09-25T13:00:11.516042+00:00",
  "msg_id": "9c1e...",
  "trace_id": "a41f...",
  "reply_to": "77d0...",
  "sender": "reader",
  "recipient": "commander",
  "created_at": "2026-09-25T13:00:11.498210+00:00",
  "body": {
    "type": "result",
    "result": {
      "task_id": "a41f...",
      "role": "reader",
      "outcome": "FAIL",
      "signals": {
        "finish_reason": "length",
        "completion_tokens": 3000,
        "elapsed_sec": 58.191,
        "text_layer_chars": 1822,
        "raw_output_path": "raw/2026-09-25/a41f.txt",
        "raw_output_sha256": "5be2..."
      },
      "model_id": "gemma-4-26b",
      "impl_version": "reader@day21"
    }
  }
}

這是示意，不是真實紀錄。signals 裡的數字抄自 page_61 整頁那次實測（58.191 秒、3000 token、length），1822 是 Day 20 引用的整頁文字層字數；id 和雜湊是假的。

schema_version 是 2，原因有點丟臉。寫 Recorder 的時候我想存原始輸出的 sha256，回頭一看，Day 21 的 SubagentResult 沒有這個欄位，Day 23 的 signals 約定表也沒列。檔案被人改過，紀錄上看不出來。我補了一個約定：Reader 在 signals 裡填 raw_output_path 和 raw_output_sha256。補之前寫出的紀錄就是版本 1，沒有雜湊，讀的程式看到版本號就知道「不要當成雜湊比對失敗」。

finish_reason 一定要撈出來變成欄位，不能只留在 JSON 裡。Day 6 page_19 那次 length 讓我很在意這件事：被截斷的輸出拿去算 CER、覆蓋率，數字沒有意義。Day 14 MOPS 那四頁整頁測試全是 length，查詢時要是看不到這個欄位，那四頁的 CER 就會混進統計，把平均拉得亂七八糟。

commander.jsonl 我原樣保留，不重寫。它是 Day 20 的格式，Recorder 只在建索引時讀它。讀的時候發現一個小問題：Day 20 的 _record 沒有寫時間。同一個 task_id 的 dispatch 和 result 誰先誰後，只能靠行號。我沒去改 Day 20，索引時把行號存成 seq。

Tip：JSON 一定要 ensure_ascii=False，開檔一定指定 encoding="utf-8"。Day 11 那次 cp950 主控台印亂碼只是顯示問題；寫檔沒指定編碼，Python 在某些 Windows 設定下會用系統預設編碼寫，那就是資料真的壞了。

SQLite schema

五張表，每一張對應一個想回答的問題：

PRAGMA foreign_keys = ON;

CREATE TABLE IF NOT EXISTS documents (
    doc_id          TEXT PRIMARY KEY,
    source_path     TEXT NOT NULL,
    source_sha256   TEXT NOT NULL,
    sensitivity     TEXT NOT NULL DEFAULT 'internal',
    ingested_at     TEXT NOT NULL
);

-- Day 23 的信封，一則一列
CREATE TABLE IF NOT EXISTS messages (
    msg_id             TEXT PRIMARY KEY,
    trace_id           TEXT NOT NULL,
    reply_to           TEXT,
    sender             TEXT NOT NULL,
    recipient          TEXT NOT NULL,
    body_type          TEXT NOT NULL CHECK (body_type IN ('task', 'result')),
    created_at         TEXT NOT NULL,
    task_id            TEXT NOT NULL,
    task_kind          TEXT,             -- 只有 task 訊息有
    parent_task_id     TEXT,
    attempt            INTEGER,
    reason             TEXT,
    doc_id             TEXT,
    location           TEXT,
    dpi                INTEGER,
    outcome            TEXT,             -- 只有 result 訊息有
    finish_reason      TEXT,
    error_kind         TEXT,
    failed_layer       TEXT,
    elapsed_sec        REAL,
    model_id           TEXT,
    impl_version       TEXT,
    raw_output_sha256  TEXT
);
CREATE INDEX IF NOT EXISTS ix_msg_trace  ON messages(trace_id);
CREATE INDEX IF NOT EXISTS ix_msg_task   ON messages(task_id);
CREATE INDEX IF NOT EXISTS ix_msg_finish ON messages(finish_reason);

-- Day 20 的 commander.jsonl，一行一列
CREATE TABLE IF NOT EXISTS decisions (
    log_file   TEXT NOT NULL,
    seq        INTEGER NOT NULL,         -- 行號，Day 20 的紀錄沒有時間
    event      TEXT NOT NULL,
    task_id    TEXT NOT NULL,
    parent     TEXT,
    kind       TEXT,
    reason     TEXT,
    rule_id    TEXT,
    status     TEXT,
    PRIMARY KEY (log_file, seq)
);
CREATE INDEX IF NOT EXISTS ix_dec_rule ON decisions(rule_id);

-- Day 25 的 DiffEvent
CREATE TABLE IF NOT EXISTS diff_events (
    event_id    TEXT PRIMARY KEY,
    msg_id      TEXT NOT NULL REFERENCES messages(msg_id),
    comparison  TEXT NOT NULL CHECK (comparison IN ('tesseract', 'text_layer', 'human')),
    tag         TEXT NOT NULL,
    auto_label  TEXT NOT NULL,
    gt_start    INTEGER, gt_end  INTEGER,
    hyp_start   INTEGER, hyp_end INTEGER,
    gt_key      TEXT,
    hyp_key     TEXT,
    p1          REAL                     -- 當時那個位置的 logprob 機率，沒開就是 NULL
);
CREATE INDEX IF NOT EXISTS ix_diff_label ON diff_events(auto_label);

-- 人工判定
CREATE TABLE IF NOT EXISTS audits (
    audit_id        INTEGER PRIMARY KEY AUTOINCREMENT,
    event_id        TEXT NOT NULL REFERENCES diff_events(event_id),
    auditor         TEXT NOT NULL,
    decision        TEXT NOT NULL CHECK (decision IN
                        ('model_wrong', 'model_right', 'gt_defect', 'unclear')),
    corrected_text  TEXT,
    audited_at      TEXT NOT NULL
);

派工和結果我塞在同一張 messages，用 body_type 分。NULL 會很多，但幾乎每個查詢都要把派工和結果配對，拆兩張就多一次 JOIN，划算。

decisions 跟 messages 靠 task_id 接起來。前者知道「為什麼有這個任務」（reason、rule_id），後者知道「這個任務做出了什麼」（outcome、finish_reason）。Day 20 那個「R3 觸發幾次、之後成功幾次」的問題，要兩張一起查才答得出來。

diff_events.comparison 對應 Day 25 那三種對照組。這是我最在意的欄位，因為飛輪那條規矩（只有人工修正可以拿來調門檻）要靠它才守得住。

audits.decision 有一個 gt_defect，給 Day 25 講的那種情況用：GT 寫 ㇐、模型寫 一，模型是對的，錯的是 PDF 文字層。這種事件不能算進模型的錯誤率，也不能算成「模型是對的」去稀釋誤報率。

寫入程式

from __future__ import annotations

import hashlib
import json
import os
import sqlite3
from datetime import datetime, timezone
from pathlib import Path

SCHEMA_VERSION = 2   # v2 起 Reader 的 signals 帶 raw_output_sha256
DDL = open(Path(__file__).with_name("schema.sql"), encoding="utf-8").read()  # 上面那段 SQL


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


class Recorder:
    def __init__(self, root: str | Path):
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)
        self.db = sqlite3.connect(self.root / "index.sqlite3")
        self.db.execute("PRAGMA journal_mode=WAL")
        self.db.executescript(DDL)

    # ---- JSONL：唯一真相 ----
    def _append(self, event_type: str, body: dict) -> None:
        line = json.dumps({"schema_version": SCHEMA_VERSION, "event_type": event_type,
                           "recorded_at": _now(), **body}, ensure_ascii=False)
        path = self.root / f"events-{datetime.now(timezone.utc):%Y-%m-%d}.jsonl"
        with open(path, "a", encoding="utf-8", newline="\n") as f:
            f.write(line + "\n")
            f.flush()
            os.fsync(f.fileno())

    # ---- Day 24 HarnessDispatch 呼叫的入口 ----
    def write(self, env) -> None:
        rec = env.model_dump(mode="json")
        self._append("message", rec)
        try:
            self._index_message(rec)
        except sqlite3.Error as exc:
            # 索引失敗不擋主流程，但要留紀錄，之後重建
            self._append("index_error", {"msg_id": rec["msg_id"], "error": str(exc)})

    def _index_message(self, rec: dict) -> None:
        body = rec["body"]
        row = {k: None for k in (
            "task_kind", "parent_task_id", "attempt", "reason", "doc_id", "location", "dpi",
            "outcome", "finish_reason", "error_kind", "failed_layer", "elapsed_sec",
            "model_id", "impl_version", "raw_output_sha256")}
        if body["type"] == "task":
            t = body["task"]
            row.update(task_id=t["task_id"], task_kind=t["kind"],
                       parent_task_id=t["parent_task_id"], attempt=t["attempt"],
                       reason=t["reason"], doc_id=t["doc_id"], location=t["location"],
                       dpi=t["params"].get("dpi"))
        else:
            r = body["result"]
            s = r["signals"]
            row.update(task_id=r["task_id"], outcome=r["outcome"],
                       finish_reason=s.get("finish_reason"), error_kind=s.get("error_kind"),
                       failed_layer=s.get("failed_layer"), elapsed_sec=s.get("elapsed_sec"),
                       model_id=r["model_id"], impl_version=r["impl_version"],
                       raw_output_sha256=s.get("raw_output_sha256"))
        row.update(msg_id=rec["msg_id"], trace_id=rec["trace_id"], reply_to=rec["reply_to"],
                   sender=rec["sender"], recipient=rec["recipient"],
                   body_type=body["type"], created_at=rec["created_at"])
        cols = ", ".join(row)
        with self.db:
            self.db.execute(
                f"INSERT OR IGNORE INTO messages ({cols}) VALUES ({', '.join(':' + k for k in row)})",
                row)

    # ---- Day 20 的 commander.jsonl ----
    def ingest_commander_log(self, path: Path) -> None:
        rows = []
        with open(path, encoding="utf-8") as f:
            for seq, line in enumerate(f):
                e = json.loads(line)
                rows.append((path.name, seq, e["event"], e["task_id"], e.get("parent"),
                             e.get("kind"), e.get("reason"), e.get("rule_id"), e.get("status")))
        with self.db:
            self.db.executemany(
                "INSERT OR IGNORE INTO decisions VALUES (?,?,?,?,?,?,?,?,?)", rows)

    # ---- Day 25 的 DiffEvent ----
    def write_diffs(self, msg_id: str, comparison: str, events: list[dict],
                    p1_by_index: dict[int, float] | None = None) -> None:
        rows = []
        for n, ev in enumerate(events):
            event_id = hashlib.sha256(f"{msg_id}|{comparison}|{n}".encode()).hexdigest()[:20]
            row = {"event_id": event_id, "msg_id": msg_id, "comparison": comparison,
                   **ev, "p1": (p1_by_index or {}).get(n)}
            self._append("diff_event", row)
            rows.append(row)
        with self.db:
            self.db.executemany(
                """INSERT OR IGNORE INTO diff_events
                   (event_id, msg_id, comparison, tag, auto_label,
                    gt_start, gt_end, hyp_start, hyp_end, gt_key, hyp_key, p1)
                   VALUES (:event_id, :msg_id, :comparison, :tag, :auto_label,
                           :gt_start, :gt_end, :hyp_start, :hyp_end,
                           :gt_key, :hyp_key, :p1)""",
                rows,
            )

    # ---- SQLite 壞了就砍掉重來 ----
    def rebuild_index(self) -> None:
        for path in sorted(self.root.glob("events-*.jsonl")):
            with open(path, encoding="utf-8") as f:
                for line in f:
                    rec = json.loads(line)
                    if rec["event_type"] == "message":
                        self._index_message(rec)
                    # diff_event、audit 同理，略
        for log in self.root.glob("commander*.jsonl"):
            self.ingest_commander_log(log)

write_diffs 收的 events，是 Day 25 的 DiffEvent 攤平後加上 auto_label，gt_span 拆成 gt_start、gt_end。

幾個選擇的理由。

os.fsync 每行都做。會慢，但這條路徑上最慢的是 OCR，一個區塊動輒好幾秒。我在意的是 Harness 當掉時，最後幾則訊息有沒有真的在磁碟上。

event_id 用 msg_id、對照組、序號算雜湊，不用自增 id。同一批 diff 重播兩次，會算出同樣的 event_id，INSERT OR IGNORE 不會寫出重複的列。decisions 的主鍵是（檔名、行號），道理一樣。rebuild_index 可以放心跑很多次。

write 裡 SQLite 失敗不丟例外。JSONL 已經寫進去，只是索引落後；讓可以重建的東西拖垮不能重來的東西，不划算。

本機我拿 Day 23 的 Envelope 和 Day 20 的 Task、Day 21 的 SubagentResult 各做了一則假訊息，寫進去再呼叫 rebuild_index() 重播。重播後 messages 還是兩列，沒有重複；finish_reason='length' 查得到。數字是我抄的，不是真的請求。

這些紀錄要回答的問題

schema 是從問題倒推的。三個例子。

Day 20 欠的那筆帳：每條規則觸發了幾次，它生出的任務後來有幾個成功？

SELECT d.rule_id,
       COUNT(*) AS fired,
       SUM(m.outcome IN ('PASS', 'N/A')) AS child_ok
FROM decisions d
JOIN decisions c  ON c.parent = d.task_id AND c.event = 'dispatch'
                 AND c.reason LIKE d.rule_id || ':%'
JOIN messages m   ON m.task_id = c.task_id AND m.body_type = 'result'
WHERE d.event = 'decide'
GROUP BY d.rule_id;

c.reason LIKE d.rule_id || ':%' 靠的就是 Day 20 那條慣例：reason 一定以規則編號開頭。當初覺得那只是方便人讀，現在它是 JOIN 條件。

負號錯誤都出在哪種任務、幾 DPI？

SELECT t.task_kind, t.dpi, d.comparison, COUNT(*) AS n
FROM diff_events d
JOIN messages r ON r.msg_id = d.msg_id
JOIN messages t ON t.task_id = r.task_id AND t.body_type = 'task'
WHERE d.auto_label = 'sign_lost'
GROUP BY 1, 2, 3
ORDER BY n DESC;

還有 Day 25 的 calibrate_min_prob 要吃的資料。gt_defect 和 unclear 排除在外：

rows = recorder.db.execute(
    """SELECT d.p1, a.decision = 'model_wrong'
       FROM diff_events d
       JOIN audits a USING (event_id)
       WHERE d.p1 IS NOT NULL
         AND a.decision IN ('model_wrong', 'model_right')"""
).fetchall()
threshold, false_alarm = calibrate_min_prob([(p, bool(e)) for p, e in rows])

這幾個查詢我只在測試資料上確認過語法跑得動，還沒有真的資料可以查。

隱私：紀錄本身也是敏感資料

寫到這裡我才意識到有點諷刺。標題有「隱私」兩個字，前面花好幾天做供應鏈審查、選本地模型，結果我的紀錄系統把文字片段一段段存進 gt_key、hyp_key，原始輸出整份放在 raw/。

年報是公開的，還好。同一套 Harness 要是拿去跑重大契約原稿、還沒公告的財報草稿，這些紀錄就是一份很完整的敏感資料副本。

目前的做法：

• documents.sensitivity 在文件進來時標好等級，後面所有紀錄都繼承它
• 原始輸出和 JSONL 只存在跑 OCR 的那台機器上，跟模型一樣不出內網
• diff_events 只存差異那一小段的比較鍵，要看上下文就回原始檔
• 保存期限、誰可以查、哪些欄位要遮，跟等級綁在一起

最後一條我只寫到「要綁在一起」。具體幾天、誰有權限，是政策問題，要跟用這套系統的單位一起定，不是我在程式裡寫個常數就算數。Day 29 講 Router 的資料敏感分級時會再碰這塊。

空著的那張表

五張表裡，前四張只要 Harness 一跑就會自己長。audits 不會。

它要人來填。可是 Day 25 那條規矩說，只有人工判定過的事件能拿來調門檻，所以整個飛輪的油門其實握在這張空表手上。

一份年報幾百個區塊，人不可能每塊都看。隨手 ORDER BY random() LIMIT 50，抽到的多半是本來就沒事的區塊，真正危險的混排區跟表格區反而不一定抽得到。那要抽幾塊才夠說「這份文件的錯誤率大概落在哪裡」？2% 夠不夠？5% 會不會太多？老實說，這個比例我到現在都還是憑感覺講的。