Phase 3: 攝入管線 — kb.py 共用模組 + ingest.py（重試/fallback/branch+PR）+ selfcheck

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tools/ingest.py

@@ -0,0 +1,307 @@
+"""攝入管線（AGENTS.md §6）：讀 raw/converted 的 Markdown，呼叫本地 Ollama
+產出/更新 wiki 頁，重建 index.md、追加 log.md，並以 git branch 準備 PR。
+
+用法：
+  python tools/ingest.py raw/converted/xxx.md [...]   # 攝入指定來源
+  python tools/ingest.py --all-pending                # 攝入所有尚無 summary 的來源
+  共用選項：[--force] [--dry-run] [--root DIR]
+
+注意：--all-pending 以「目前分支」的 wiki 為準；尚未合併的 ingest PR 不會被看到。
+攝入腳本永遠不直接 commit 到 main（AGENTS.md §1.5）。
+"""
+import argparse
+import datetime
+import json
+import pathlib
+import subprocess
+import sys
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+import kb
+
+SUMMARY_PROMPT = """你是金融電子支付 QA 部門知識庫的攝入引擎。閱讀以下來源文件，只輸出一個 JSON 物件，不得有任何其他文字。
+
+規則：
+- 文字內容用繁體中文；slug 用英文小寫 kebab-case。
+- description 為一句話摘要；tags 為 2 至 6 個英文小寫 kebab-case 標籤。
+- summary_markdown：文件重點摘要（目的、關鍵事實、結論、教訓）。不要逐步驟複製測試案例內容——單檔可 grep 的細節不進 wiki。
+- knowledge_items：值得建頁或更新的項目。entity＝系統/模組/API/法規條目/專案；concept＝跨來源的缺陷模式/測試策略/法遵準則。沒有就給空陣列。
+- facts_markdown：本文件中關於該項目的事實要點（markdown bullet），供合併至該項目頁面。
+
+JSON 格式：
+{{"title": "...", "description": "...", "slug": "...", "tags": ["..."],
+  "summary_markdown": "...",
+  "knowledge_items": [{{"type": "entity 或 concept", "slug": "...", "title": "...",
+                        "description": "...", "tags": ["..."], "facts_markdown": "..."}}]}}
+
+檔名：{filename}
+文件內容：
+<<<
+{content}
+>>>"""
+
+CHUNK_PROMPT = """以下是一份長文件的第 {i}/{n} 段。請以繁體中文摘要此段重點（markdown bullet），只輸出 JSON：{{"summary_markdown": "..."}}
+
+<<<
+{content}
+>>>"""
+
+MERGE_PROMPT = """你是知識庫維護引擎。以下是既有 wiki 頁面內文，以及來自新來源的事實要點。
+請把新事實整合進頁面（就地編輯：合併、去重；若與既有內容矛盾，兩種說法並陳並標註「⚠ 待查證」）。
+只輸出 JSON：{{"updated_markdown": "...", "updated_description": "..."}}
+規則：繁體中文；保留原有結構與仍然有效的內容。
+
+頁面標題：{title}
+既有內文：
+<<<
+{body}
+>>>
+新事實（來源 {source_ref}）：
+<<<
+{facts}
+>>>"""
+
+
+def _s(o, k):
+    return isinstance(o.get(k), str) and o[k].strip()
+
+
+def _tags_ok(o):
+    return isinstance(o.get("tags"), list) and o["tags"] and all(isinstance(t, str) for t in o["tags"])
+
+
+def validate_summary(obj):
+    for k in ("title", "description", "slug", "summary_markdown"):
+        if not _s(obj, k):
+            return f"欄位 {k} 缺漏或非字串"
+    if not _tags_ok(obj):
+        return "tags 須為非空字串列表"
+    if not isinstance(obj.get("knowledge_items"), list):
+        return "knowledge_items 須為列表"
+    for it in obj["knowledge_items"]:
+        if not isinstance(it, dict) or it.get("type") not in ("entity", "concept"):
+            return "knowledge_item 須為物件且 type 為 entity|concept"
+        for k in ("slug", "title", "description", "facts_markdown"):
+            if not _s(it, k):
+                return f"knowledge_item 欄位 {k} 缺漏或非字串"
+        if not _tags_ok(it):
+            return "knowledge_item.tags 須為非空字串列表"
+    return None
+
+
+def validate_chunk(obj):
+    return None if _s(obj, "summary_markdown") else "欄位 summary_markdown 缺漏"
+
+
+def validate_merge(obj):
+    for k in ("updated_markdown", "updated_description"):
+        if not _s(obj, k):
+            return f"欄位 {k} 缺漏或非字串"
+    return None
+
+
+def split_chunks(text, limit):
+    """依段落邊界切塊，每塊不超過 limit 字元（單一超長段落則硬切）。"""
+    chunks, cur = [], ""
+    for para in text.split("\n\n"):
+        while len(para) > limit:  # 單段超長：硬切
+            chunks.append(para[:limit])
+            para = para[limit:]
+        if len(cur) + len(para) + 2 > limit and cur:
+            chunks.append(cur)
+            cur = para
+        else:
+            cur = f"{cur}\n\n{para}" if cur else para
+    if cur:
+        chunks.append(cur)
+    return chunks
+
+
+def summarize(cfg, filename, text):
+    limit = cfg["limits"]["ingest_chunk_max_chars"]
+    if len(text) > limit:
+        parts = split_chunks(text, limit)
+        partials = []
+        for i, part in enumerate(parts, 1):
+            obj, _ = kb.llm_json(cfg, "ingest",
+                                 CHUNK_PROMPT.format(i=i, n=len(parts), content=part),
+                                 validate_chunk)
+            partials.append(f"### 分段 {i} 摘要\n{obj['summary_markdown']}")
+        text = "（以下為長文件的分段摘要，請據此綜合）\n\n" + "\n\n".join(partials)
+    return kb.llm_json(cfg, "ingest",
+                       SUMMARY_PROMPT.format(filename=filename, content=text),
+                       validate_summary)
+
+
+def _unique_path(d, slug, sha8):
+    p = d / f"{slug}.md"
+    return p if not p.exists() else d / f"{slug}-{sha8}.md"
+
+
+def write_summary_page(root, obj, source_ref, sha8, today):
+    d = root / kb.PAGE_DIRS["summary"]
+    p = _unique_path(d, kb.safe_slug(obj["slug"], f"doc-{sha8}"), sha8)
+    meta = {"type": "summary", "title": obj["title"], "description": obj["description"],
+            "tags": obj["tags"], "timestamp": today, "sources": [source_ref],
+            "status": "draft"}
+    p.write_text(kb.dump_page(meta, obj["summary_markdown"]), encoding="utf-8")
+    return p
+
+
+def upsert_item(root, cfg, item, source_ref, today, changed):
+    slug = kb.safe_slug(item["slug"], f"item-{source_ref[-8:]}")
+    existing = None
+    for t in ("entity", "concept"):  # 既有頁優先（就地編輯），不管本次 LLM 判的型別
+        p = root / kb.PAGE_DIRS[t] / f"{slug}.md"
+        if p.exists():
+            existing = p
+            break
+    if existing:
+        meta, body = kb.parse_page(existing.read_text(encoding="utf-8"))
+        obj, _ = kb.llm_json(cfg, "ingest",
+                             MERGE_PROMPT.format(title=meta["title"], body=body,
+                                                 source_ref=source_ref,
+                                                 facts=item["facts_markdown"]),
+                             validate_merge)
+        meta["description"] = obj["updated_description"]
+        meta["tags"] = sorted(set(meta["tags"]) | set(item["tags"]))
+        meta["timestamp"] = today
+        meta["status"] = "draft"  # 內容變動須重新人審
+        if source_ref not in meta["sources"]:
+            meta["sources"].append(source_ref)
+        existing.write_text(kb.dump_page(meta, obj["updated_markdown"]), encoding="utf-8")
+        changed.append(f"更新 {existing.relative_to(root).as_posix()}")
+    else:
+        p = root / kb.PAGE_DIRS[item["type"]] / f"{slug}.md"
+        meta = {"type": item["type"], "title": item["title"],
+                "description": item["description"], "tags": item["tags"],
+                "timestamp": today, "sources": [source_ref], "status": "draft"}
+        p.write_text(kb.dump_page(meta, item["facts_markdown"]), encoding="utf-8")
+        changed.append(f"新增 {p.relative_to(root).as_posix()}")
+
+
+# ---------- git ----------
+
+def run_git(root, *args, capture=False):
+    return subprocess.run(["git", *args], cwd=root, check=True, text=True,
+                          encoding="utf-8", capture_output=capture)
+
+
+def git_preflight(root):
+    branch = run_git(root, "branch", "--show-current", capture=True).stdout.strip()
+    if branch != "main":
+        raise SystemExit(f"須在 main 分支執行（目前：{branch}）")
+    dirty = [l for l in run_git(root, "status", "--porcelain", capture=True)
+             .stdout.splitlines() if l.strip() and not l[3:].startswith("raw/")]
+    if dirty:
+        raise SystemExit("工作區有 raw/ 以外的未提交變更，請先處理：\n" + "\n".join(dirty))
+
+
+def referenced_sha8s(root):
+    refs = set()
+    for p in kb.iter_pages(root):
+        meta, _ = kb.parse_page(p.read_text(encoding="utf-8"))
+        for s in meta.get("sources", []):
+            refs.add(str(s).rsplit("#", 1)[-1])
+    return refs
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__,
+                                 formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("paths", nargs="*", help="raw/converted 下的 Markdown 路徑")
+    ap.add_argument("--all-pending", action="store_true")
+    ap.add_argument("--force", action="store_true", help="已攝入過（hash 相同）也重跑")
+    ap.add_argument("--dry-run", action="store_true", help="只列計畫，不呼叫 LLM、不寫檔")
+    ap.add_argument("--root", default=str(kb.ROOT))
+    a = ap.parse_args(argv)
+    root = pathlib.Path(a.root).resolve()
+    cfg = kb.load_config(root)
+    manifest = json.loads((root / "raw" / "manifest.json").read_text(encoding="utf-8"))
+    conv = {k: v for k, v in manifest["files"].items() if v["kind"] == "converted"}
+
+    if a.paths:
+        targets = []
+        for p in a.paths:
+            rel = pathlib.Path(p)
+            rel = (rel.relative_to(root) if rel.is_absolute() else rel).as_posix()
+            if rel not in conv:
+                raise SystemExit(f"{rel} 不在 manifest 的 converted 條目中，請先跑 convert.py")
+            targets.append(rel)
+    elif a.all_pending:
+        done = referenced_sha8s(root)
+        targets = [k for k, v in conv.items() if v["sha256"][:8] not in done]
+    else:
+        ap.error("請指定來源路徑或 --all-pending")
+    if not a.force:
+        done = referenced_sha8s(root)
+        skipped = [t for t in targets if conv[t]["sha256"][:8] in done]
+        targets = [t for t in targets if conv[t]["sha256"][:8] not in done]
+        for t in skipped:
+            print(f"skip: {t} 已有 summary 引用（--force 可重跑）")
+    if not targets:
+        print("沒有待攝入的來源。")
+        return
+
+    today = datetime.date.today().isoformat()
+    stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
+    branch = f"ingest/{stamp}-{kb.safe_slug(pathlib.Path(targets[0]).stem, 'batch')[:30]}"
+
+    if a.dry_run:
+        limit = cfg["limits"]["ingest_chunk_max_chars"]
+        print(f"[dry-run] 分支：{branch}，模型：{cfg['tasks']['ingest']['model']}")
+        for t in targets:
+            n = len((root / t).read_text(encoding="utf-8"))
+            print(f"[dry-run] {t}: {n} chars，{max(1, -(-n // limit))} 段")
+        return
+
+    git_preflight(root)
+    run_git(root, "checkout", "-q", "-b", branch)
+    ok, failed, changed = [], [], []
+    try:
+        for t in targets:
+            try:
+                text = (root / t).read_text(encoding="utf-8")
+                sha8 = conv[t]["sha256"][:8]
+                source_ref = f"{t}#{sha8}"
+                obj, model = summarize(cfg, pathlib.Path(conv[t]["original_path"]).name, text)
+                p = write_summary_page(root, obj, source_ref, sha8, today)
+                changed.append(f"新增 {p.relative_to(root).as_posix()}")
+                for item in obj["knowledge_items"]:
+                    upsert_item(root, cfg, item, source_ref, today, changed)
+                ok.append((t, model))
+                print(f"ingested: {t}（model={model}）")
+            except Exception as e:
+                failed.append((t, e))
+                print(f"error: {t}: {e}", file=sys.stderr)
+        if not ok:
+            run_git(root, "checkout", "-q", "main")
+            run_git(root, "branch", "-q", "-D", branch)
+            raise SystemExit("全部來源攝入失敗，已還原至 main。")
+        kb.rebuild_index(root)
+        kb.append_log(root, "ingest", f"攝入 {len(ok)} 份來源（branch {branch}）",
+                      changed + [f"失敗：{t}（{e}）" for t, e in failed])
+        run_git(root, "add", "wiki", "index.md", "log.md", "raw")
+        run_git(root, "commit", "-q", "-m",
+                f"ingest: {len(ok)} 份來源\n\n" + "\n".join(f"- {c}" for c in changed))
+        try:
+            run_git(root, "push", "-q", "-u", "origin", branch)
+            remote = run_git(root, "remote", "get-url", "origin",
+                             capture=True).stdout.strip().removesuffix(".git")
+            print(f"PR 建立網址：{remote}/compare/main...{branch}")
+        except subprocess.CalledProcessError:
+            print(f"push 失敗（無 remote 或離線）。分支 {branch} 保留於本地，請手動 push 後開 PR。")
+        run_git(root, "checkout", "-q", "main")
+        print(f"完成：{len(ok)} 成功、{len(failed)} 失敗。變更在分支 {branch}，經 PR 審核後合併。")
+    except SystemExit:
+        raise
+    except Exception:
+        print(f"攝入中斷。目前在分支 {branch}，工作區可能有未提交變更，"
+              "請人工檢查（不自動清除以免遺失資料）。", file=sys.stderr)
+        raise
+    if failed:
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()

tools/kb.py

@@ -0,0 +1,186 @@
+"""共用模組：config 載入驗證、本地 Ollama 呼叫（重試 + fallback）、
+wiki 頁 frontmatter 解析/寫出、index.md 重建、log.md 追加。
+
+供 ingest.py / lint.py / search.py / mcp/server.py 使用。
+"""
+import datetime
+import json
+import pathlib
+import os
+import re
+import urllib.request
+
+import yaml
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+PAGE_DIRS = {"summary": "wiki/summaries", "entity": "wiki/entities", "concept": "wiki/concepts"}
+PAGE_TYPES = set(PAGE_DIRS)
+STATUSES = {"draft", "reviewed", "stale"}
+REQUIRED_FM = ("type", "title", "description", "tags", "timestamp", "sources", "status")
+
+
+# ---------- config ----------
+
+def load_config(root=ROOT):
+    """讀取 config/models.yaml。缺欄位報錯，不使用隱含預設值（AGENTS.md §1.2）。"""
+    path = root / "config" / "models.yaml"
+    cfg = yaml.safe_load(path.read_text(encoding="utf-8"))
+    try:
+        cfg["ollama"]["base_url"]
+        cfg["ollama"]["timeout_seconds"]
+        for task in ("ingest", "lint"):
+            for key in ("model", "temperature", "max_retries"):
+                cfg["tasks"][task][key]
+        cfg["tasks"]["fallback"]["model"]
+        cfg["limits"]["mcp_response_max_tokens"]
+        cfg["limits"]["ingest_chunk_max_chars"]
+    except (KeyError, TypeError) as e:
+        raise SystemExit(f"config/models.yaml 缺欄位：{e}")
+    for task, spec in cfg["tasks"].items():
+        if spec["model"] == "CHANGE_ME":
+            raise SystemExit(f"config/models.yaml：tasks.{task}.model 尚未設定")
+        if spec["model"].endswith(":cloud"):
+            raise SystemExit(f"tasks.{task}.model 為 :cloud 模型，違反資料落地（AGENTS.md §1.1）")
+    # 環境變數是唯一允許的覆寫來源（AGENTS.md §1.2）
+    cfg["ollama"]["base_url"] = os.environ.get("OLLAMA_BASE_URL", cfg["ollama"]["base_url"])
+    return cfg
+
+
+# ---------- Ollama ----------
+
+def ollama_chat(cfg, model, prompt, temperature, json_format=False):
+    """單次呼叫本地 Ollama /api/chat（stdlib urllib，不需額外依賴）。"""
+    body = {
+        "model": model,
+        "messages": [{"role": "user", "content": prompt}],
+        "stream": False,
+        "options": {"temperature": temperature},
+    }
+    if json_format:
+        body["format"] = "json"
+    req = urllib.request.Request(
+        cfg["ollama"]["base_url"].rstrip("/") + "/api/chat",
+        data=json.dumps(body).encode("utf-8"),
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(req, timeout=cfg["ollama"]["timeout_seconds"]) as r:
+        return json.loads(r.read())["message"]["content"]
+
+
+def llm_json(cfg, task, prompt, validate):
+    """要求 JSON 輸出的呼叫：主模型重試 max_retries 次，仍失敗切 fallback 模型。
+
+    本地模型輸出不穩定是預期情況（AGENTS.md 前提），故每次失敗把錯誤附回
+    prompt 再試。validate(obj) 回傳錯誤訊息字串或 None。
+    回傳 (obj, 實際使用的 model)。全部失敗拋 RuntimeError。
+    """
+    spec = cfg["tasks"][task]
+    attempts = []
+    for model in (spec["model"], cfg["tasks"]["fallback"]["model"]):
+        hint = ""
+        for _ in range(spec["max_retries"]):
+            try:
+                raw = ollama_chat(cfg, model, prompt + hint, spec["temperature"],
+                                  json_format=True)
+                obj = json.loads(raw)
+                err = validate(obj)
+                if err is None:
+                    return obj, model
+            except Exception as e:
+                err = f"{type(e).__name__}: {e}"
+            attempts.append(f"{model}: {err}")
+            hint = f"\n\n注意：上次輸出無效（{err}）。請只輸出符合規格的 JSON，不得有其他文字。"
+    raise RuntimeError("LLM 輸出驗證失敗（含 fallback）：" + " | ".join(attempts[-4:]))
+
+
+# ---------- wiki 頁 ----------
+
+def parse_page(text):
+    """回傳 (frontmatter dict, body)。格式不符拋 ValueError。"""
+    m = re.match(r"^---\r?\n(.*?)\r?\n---\r?\n(.*)$", text, re.S)
+    if not m:
+        raise ValueError("缺少 YAML frontmatter")
+    meta = yaml.safe_load(m.group(1))
+    if not isinstance(meta, dict):
+        raise ValueError("frontmatter 不是 mapping")
+    return meta, m.group(2).strip()
+
+
+def dump_page(meta, body):
+    fm = yaml.safe_dump(meta, allow_unicode=True, sort_keys=False)
+    return f"---\n{fm}---\n\n{body.strip()}\n"
+
+
+def validate_meta(meta):
+    """依 AGENTS.md §3.2 檢查 frontmatter，回傳錯誤訊息列表。"""
+    errs = [f"缺欄位 {k}" for k in REQUIRED_FM if k not in meta]
+    if errs:
+        return errs
+    if meta["type"] not in PAGE_TYPES:
+        errs.append(f"type 不合法：{meta['type']}")
+    if meta["status"] not in STATUSES:
+        errs.append(f"status 不合法：{meta['status']}")
+    if not isinstance(meta["tags"], list) or not meta["tags"]:
+        errs.append("tags 須為非空列表")
+    if not isinstance(meta["sources"], list) or not meta["sources"]:
+        errs.append("sources 須為非空列表")
+    for s in meta.get("sources") or []:
+        if not re.match(r"^raw/(converted|originals)/.+#[0-9a-f]{8}$", str(s)):
+            errs.append(f"source_ref 格式不符：{s}")
+    if not re.match(r"^\d{4}-\d{2}-\d{2}$", str(meta["timestamp"])):
+        errs.append(f"timestamp 須為 YYYY-MM-DD：{meta['timestamp']}")
+    return errs
+
+
+def safe_slug(s, fallback):
+    s = re.sub(r"[^a-z0-9]+", "-", str(s).lower()).strip("-")[:60]
+    return s or fallback
+
+
+def iter_pages(root=ROOT):
+    for sub in PAGE_DIRS.values():
+        d = root / sub
+        if d.is_dir():
+            yield from sorted(d.glob("*.md"))
+
+
+# ---------- index.md / log.md ----------
+
+def rebuild_index(root=ROOT):
+    """由全部 wiki 頁 frontmatter 決定性重建 index.md（AGENTS.md §10）。"""
+    groups = {"summary": [], "entity": [], "concept": []}
+    for p in iter_pages(root):
+        meta, _ = parse_page(p.read_text(encoding="utf-8"))
+        rel = p.relative_to(root).as_posix()
+        tags = " ".join(f"#{t}" for t in meta["tags"])
+        groups[meta["type"]].append(
+            (str(meta["title"]), f"- [{meta['title']}]({rel}) — {meta['description']} ｜ {tags}"))
+    def section(key):
+        lines = [line for _, line in sorted(groups[key])]
+        return "\n".join(lines) if lines else "（尚無頁面）"
+    text = f"""# 知識庫索引
+
+<!-- 目錄式導航（AGENTS.md §10）：每頁一行，由 ingest 管線自動重建，格式：
+     - [標題](路徑) — 一句摘要 ｜ #tag1 #tag2
+     查詢工作流一律先讀本檔定位候選頁（最多 10 頁）。 -->
+
+## 摘要頁（summaries）
+
+{section('summary')}
+
+## 實體頁（entities）
+
+{section('entity')}
+
+## 概念頁（concepts）
+
+{section('concept')}
+"""
+    (root / "index.md").write_text(text, encoding="utf-8")
+
+
+def append_log(root, op, title, lines):
+    entry = f"\n## [{datetime.date.today().isoformat()}] {op} | {title}\n\n"
+    entry += "".join(f"- {l}\n" for l in lines)
+    with open(root / "log.md", "a", encoding="utf-8") as f:
+        f.write(entry)

tools/selfcheck_ingest.py

@@ -0,0 +1,144 @@
+"""Phase 3 自檢：以假 LLM + 沙盒 git repo 驗證攝入管線的完整 plumbing——
+分支建立、頁面產出、frontmatter 合規、index/log 更新、冪等跳過、就地編輯合併，
+以及 kb.llm_json 的重試與 fallback 切換。不需要 Ollama 在線。
+
+執行：.venv/Scripts/python tools/selfcheck_ingest.py
+"""
+import hashlib
+import pathlib
+import shutil
+import subprocess
+import sys
+import tempfile
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+import kb
+import ingest
+
+CFG = """ollama: {base_url: "http://localhost:11434", timeout_seconds: 5}
+tasks:
+  ingest: {model: "fake:test", temperature: 0.2, max_retries: 2}
+  lint: {model: "fake:test", temperature: 0.1, max_retries: 2}
+  fallback: {model: "fake:fb"}
+limits: {mcp_response_max_tokens: 2000, ingest_chunk_max_chars: 200}
+"""
+
+
+def fake_llm(cfg, task, prompt, validate):
+    if prompt.startswith("以下是一份長文件的第"):
+        obj = {"summary_markdown": "- 分段重點"}
+    elif '"updated_markdown"' in prompt:
+        obj = {"updated_markdown": "- 壓測逾時\n- 新來源事實",
+               "updated_description": "支付閘道模組（已更新）"}
+    else:
+        obj = {"title": "支付閘道壓測報告", "description": "壓測發現逾時缺陷",
+               "slug": "gateway-load-test", "tags": ["load-test", "gateway"],
+               "summary_markdown": "- 逾時缺陷於壓測重現",
+               "knowledge_items": [{"type": "entity", "slug": "payment-gateway",
+                                    "title": "支付閘道", "description": "支付閘道模組",
+                                    "tags": ["gateway"], "facts_markdown": "- 壓測逾時"}]}
+    err = validate(obj)  # 假輸出也要通過真 validator，確保契約一致
+    assert err is None, err
+    return obj, "fake:test"
+
+
+def git(root, *args):
+    return subprocess.run(["git", *args], cwd=root, check=True, text=True,
+                          encoding="utf-8", capture_output=True).stdout.strip()
+
+
+def add_doc(root, name, text):
+    p = root / "raw" / "converted" / name
+    p.write_text(text, encoding="utf-8")
+    sha = hashlib.sha256(text.encode("utf-8")).hexdigest()
+    import json
+    mp = root / "raw" / "manifest.json"
+    m = json.loads(mp.read_text(encoding="utf-8"))
+    m["files"][f"raw/converted/{name}"] = {
+        "kind": "converted", "sha256": sha, "original_path": f"raw/originals/{name}",
+        "original_sha256": sha, "converter": "from_docx", "converted_at": "2026-07-14T00:00:00"}
+    mp.write_text(json.dumps(m, ensure_ascii=False, indent=2), encoding="utf-8")
+    return sha
+
+
+def main():
+    tmp = pathlib.Path(tempfile.mkdtemp(prefix="ppqa-ingest-"))
+    real_llm, real_chat = kb.llm_json, kb.ollama_chat
+    try:
+        root = tmp / "repo"
+        for sub in ("config", "raw/originals", "raw/converted",
+                    "wiki/summaries", "wiki/entities", "wiki/concepts"):
+            (root / sub).mkdir(parents=True)
+        (root / "config" / "models.yaml").write_text(CFG, encoding="utf-8")
+        (root / "raw" / "manifest.json").write_text('{"version": 1, "files": {}}',
+                                                    encoding="utf-8")
+        (root / "index.md").write_text("# 知識庫索引\n", encoding="utf-8")
+        (root / "log.md").write_text("# 操作日誌\n", encoding="utf-8")
+        git(root, "init", "-q", "-b", "main")
+        git(root, "config", "user.name", "selfcheck")
+        git(root, "config", "user.email", "selfcheck@local")
+        text = "支付閘道於壓力測試下出現逾時缺陷，交易峰值時重試機制未生效。\n\n" * 12
+        add_doc(root, "doc1.md", text)  # > 200 chars → 走分段路徑
+        git(root, "add", "-A")
+        git(root, "commit", "-q", "-m", "init")
+
+        # 1) 攝入：建分支、產頁、commit、回到 main
+        kb.llm_json = fake_llm
+        ingest.main(["raw/converted/doc1.md", "--root", str(root)])
+        assert git(root, "branch", "--show-current") == "main"
+        branch = git(root, "branch", "--list", "ingest/*").strip("* ").strip()
+        assert branch, "未建立 ingest 分支"
+        assert not git(root, "status", "--porcelain"), "工作區不乾淨"
+        git(root, "checkout", "-q", branch)
+        summary = list((root / "wiki/summaries").glob("*.md"))
+        entity = root / "wiki/entities/payment-gateway.md"
+        assert len(summary) == 1 and entity.is_file()
+        meta, _ = kb.parse_page(entity.read_text(encoding="utf-8"))
+        assert kb.validate_meta(meta) == [], kb.validate_meta(meta)
+        assert "payment-gateway" in (root / "index.md").read_text(encoding="utf-8")
+        assert "] ingest |" in (root / "log.md").read_text(encoding="utf-8")
+        print("PASS: 攝入 → 分支 + summary/entity 頁 + index/log")
+
+        # 2) 模擬 PR 合併後：冪等跳過；新來源觸發既有頁就地編輯
+        git(root, "checkout", "-q", "main")
+        git(root, "merge", "-q", branch)
+        ingest.main(["raw/converted/doc1.md", "--root", str(root)])  # 應 skip
+        assert git(root, "branch", "--show-current") == "main"
+        add_doc(root, "doc2.md", "第二份文件：支付閘道逾時已修復並通過回歸測試。")
+        git(root, "add", "-A")
+        git(root, "commit", "-q", "-m", "doc2")
+        ingest.main(["raw/converted/doc2.md", "--root", str(root)])
+        b2 = [b.strip("* ").strip() for b in
+              git(root, "branch", "--list", "ingest/*").splitlines() if branch not in b]
+        git(root, "checkout", "-q", b2[0])
+        meta, body = kb.parse_page(entity.read_text(encoding="utf-8"))
+        assert len(meta["sources"]) == 2 and "已更新" in meta["description"], meta
+        assert "新來源事實" in body
+        git(root, "checkout", "-q", "main")
+        print("PASS: 冪等跳過 + 既有 entity 頁就地編輯（sources 追加）")
+
+        # 3) kb.llm_json：重試後切 fallback；全失敗拋錯
+        kb.llm_json = real_llm
+        cfg = kb.load_config(root)
+        calls = {"n": 0}
+        def flaky(cfg_, model, prompt, temperature, json_format=False):
+            calls["n"] += 1
+            return "not-json" if calls["n"] < 3 else '{"summary_markdown": "ok"}'
+        kb.ollama_chat = flaky
+        obj, model = kb.llm_json(cfg, "ingest", "p", ingest.validate_chunk)
+        assert obj["summary_markdown"] == "ok" and model == "fake:fb" and calls["n"] == 3
+        kb.ollama_chat = lambda *a, **k: "junk"
+        try:
+            kb.llm_json(cfg, "ingest", "p", ingest.validate_chunk)
+            raise AssertionError("應拋 RuntimeError")
+        except RuntimeError as e:
+            assert "fallback" in str(e)
+        print("PASS: 重試 + fallback 切換 + 全失敗報錯")
+        print("ALL PASS")
+    finally:
+        kb.llm_json, kb.ollama_chat = real_llm, real_chat
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    main()