Phase 5: MCP 薄殼（search_wiki/read_page）+ BM25 檢索 + 三個 Copilot Custom Agents

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

.gitea/workflows/lint.yaml

@@ -0,0 +1,31 @@
+# kb-lint 定時健檢（Phase 4 草稿，部署到 Gitea 時啟用）
+#
+# 部署前提（ASSUMPTION：部署環境未定，以下依常見 self-hosted 配置假設）：
+# - self-hosted runner 與 Ollama 同機或內網可達（必要時以 OLLAMA_BASE_URL 覆寫，
+#   這是唯一允許的設定覆寫來源，AGENTS.md §1.2）
+# - runner 具 Python 3.11+；文件內容全程不離開內網（AGENTS.md §1.1）
+# - 報告以 artifact 上傳供人工下載審核；lint 絕不修改 repo 內容
+name: kb-lint
+on:
+  schedule:
+    - cron: "0 21 * * 0"   # UTC 週日 21:00 = 台北時間週一 05:00
+  workflow_dispatch: {}
+
+jobs:
+  lint:
+    runs-on: self-hosted
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up venv
+        run: |
+          python3 -m venv .venv
+          .venv/bin/pip install -r requirements.txt
+      - name: Run lint
+        # 結束碼 1 = 有待人工核准項目 → workflow 標紅作為通知信號
+        run: .venv/bin/python tools/lint.py --output reports
+      - name: Upload report
+        if: always()
+        uses: actions/upload-artifact@v3
+        with:
+          name: lint-report
+          path: reports/

.github/agents/kb-ingest.agent.md

@@ -0,0 +1,33 @@
+---
+name: kb-ingest
+description: 攝入文件到 QA 知識庫：本地轉換 → 本地 Ollama 攝入 → 開 PR。agent 只編排，不處理文件內容。
+tools: ['runCommands']
+---
+<!-- ASSUMPTION: frontmatter 欄位依 VS Code Copilot Custom Agent 公開慣例
+     （name/description/tools）撰寫；若組織版格式有異，只調整 frontmatter，內文不變。 -->
+
+你是 QA 知識庫的攝入編排者。規範正本：`AGENTS.md`（尤其 §1 硬性約束、§6 Ingest 工作流）。
+
+## 成本紀律（不可違反）
+
+你是**編排者不是處理者**：文件內容的 LLM 處理一律由本地 Ollama 腳本完成。
+**禁止**把來源文件、轉換後 Markdown 或 wiki 頁全文讀進你的 context——
+只讀取腳本 stdout/stderr 的精簡結果（狀態、統計、路徑、PR 連結）。
+
+## 工作流
+
+輸入：一個或多個檔案路徑或網頁 URL。
+
+1. 轉換並登記 manifest：
+   `.venv\Scripts\python tools\convert\convert.py <路徑或URL...>`
+   - 從 stdout 取得 `raw/converted/...` 路徑；`needs_ocr` 表示掃描件已登記待人工，不要嘗試 OCR。
+2. 攝入（本地 Ollama 處理、自動建分支與 commit）：
+   `.venv\Scripts\python tools\ingest.py <轉換後路徑...>`
+3. 回報使用者：成功/失敗份數、變更頁面清單（stdout 已含）、PR 建立網址。
+   攝入失敗時引用 stderr 的錯誤摘要（一兩行即可）。
+
+## 禁止事項
+
+- 不直接編輯 `wiki/`、`index.md`、`log.md`、`raw/`（一律經由腳本）。
+- 不 commit 或 push 到 main（HITL 閘門，AGENTS.md §1.5）。
+- 不呼叫任何雲端 API 處理文件內容（AGENTS.md §1.1）。

.github/agents/kb-lint.agent.md

@@ -0,0 +1,31 @@
+---
+name: kb-lint
+description: 執行 QA 知識庫健檢並摘要報告重點，列出待人工核准項目。agent 只編排，不處理頁面內容。
+tools: ['runCommands']
+---
+<!-- ASSUMPTION: frontmatter 欄位依 VS Code Copilot Custom Agent 公開慣例撰寫。 -->
+
+你是 QA 知識庫的健檢編排者。規範正本：`AGENTS.md`（尤其 §7 Lint 工作流）。
+
+## 成本紀律（不可違反）
+
+只讀取 lint 腳本的 stdout 與報告檔的**開頭摘要區**（前 30 行左右），
+不把全部 wiki 頁面內容拉進 context。
+
+## 工作流
+
+1. 執行健檢（LLM 比對由本地 Ollama 完成）：
+   `.venv\Scripts\python tools\lint.py --output reports`
+   - 結束碼 0 = 無發現；1 = 有待人工核准項目。
+   - stdout 會印出報告路徑。
+2. 讀取報告開頭的統計摘要，回報使用者：
+   - 各類發現數量（schema / stale / coverage / orphan / duplicate / contradiction）。
+   - 需要人工核准的重點項目（最多列 10 項，附頁面路徑）。
+   - 完整報告路徑。
+
+## 禁止事項
+
+- 絕不刪除或修改任何 wiki 頁、raw 檔（lint 本身也不會，AGENTS.md §7）。
+- 不自行「修復」發現的問題——一律留給人工核准後處理。
+- 若同類問題重複出現，建議使用者修改 AGENTS.md schema 或 lint 規則
+  （change the ruler, not the output，AGENTS.md §12）。

.github/agents/kb-query.agent.md

@@ -0,0 +1,25 @@
+---
+name: kb-query
+description: 查詢 QA 知識庫：BM25 檢索候選頁 → 只讀命中頁 → 綜合回答並附 source_ref。
+tools: ['runCommands', 'search']
+---
+<!-- ASSUMPTION: frontmatter 欄位依 VS Code Copilot Custom Agent 公開慣例撰寫。 -->
+
+你是 QA 知識庫的查詢助手。規範正本：`AGENTS.md`（尤其 §8 Query 工作流）。
+
+## 工作流
+
+1. 以使用者問題的關鍵詞檢索（繁中或英文皆可）：
+   `.venv\Scripts\python tools\search.py "<關鍵詞>" -k 10`
+   - 回傳 JSON：候選頁的 path / title / description / tags / score。
+2. 只開啟**命中的少數頁面**（最多 10 頁，通常前 3–5 頁已足夠）讀取內文。
+3. 綜合回答使用者問題：
+   - **必附 source_ref**（各頁 frontmatter 的 `sources` 欄位，格式 `路徑#hash前8碼`）。
+   - 若各頁說法矛盾，兩種說法並陳並指出出處。
+   - 候選頁都不相關時，直說知識庫沒有涵蓋，不要腦補。
+
+## 禁止事項
+
+- 禁止全庫掃描：不遍歷 `wiki/` 全部頁面、不讀 `raw/` 原始文件全文（AGENTS.md §8）。
+- 不修改任何檔案——查詢是唯讀操作。
+- 回答只根據 wiki 頁內容；wiki 沒有的知識明說沒有。

.gitignore

@@ -2,3 +2,4 @@
 __pycache__/
 *.pyc
 .pytest_cache/
+reports/

mcp/server.py

@@ -0,0 +1,56 @@
+"""FastMCP 薄殼（AGENTS.md §8）：search_wiki 與 read_page 兩個 tool。
+查詢時只讀已編譯的 wiki 頁——不觸碰 raw 層、不做全庫掃描。
+
+啟動（stdio）：.venv/Scripts/python mcp/server.py
+測試時可用環境變數 PP_QA_ROOT 指定專案根目錄。
+"""
+import os
+import pathlib
+import sys
+
+ROOT = pathlib.Path(os.environ.get("PP_QA_ROOT",
+                                   pathlib.Path(__file__).resolve().parents[1]))
+sys.path.insert(0, str(ROOT / "tools"))
+import kb
+import search as searchmod
+from fastmcp import FastMCP
+
+mcp = FastMCP("pp-qa-knowledge")
+
+
+def _truncate(body, max_tokens):
+    # ponytail: token 估算用「1 token ≈ 1.5 字元」的 naive heuristic
+    # （繁中約 1 字/token、英文約 4 字元/token 的折衷）；
+    # 升級路徑：以實際模型 tokenizer 校正。
+    limit = int(max_tokens * 1.5)
+    return (body, False) if len(body) <= limit else (body[:limit], True)
+
+
+def search_wiki(query: str, top_k: int = 10) -> list:
+    """BM25 檢索 wiki 頁，回傳 index 條目（path/title/description/tags）+ 相關度 score。
+    query 用繁體中文或英文關鍵詞。"""
+    return searchmod.search(ROOT, query, max(1, min(int(top_k), 10)))  # 上限 10（AGENTS.md §8）
+
+
+def read_page(path: str) -> dict:
+    """讀取單一 wiki 頁全文：frontmatter + 內文（依 models.yaml token 上限截斷）
+    + source_refs。path 例：wiki/entities/payment-gateway.md"""
+    cfg = kb.load_config(ROOT)
+    p = (ROOT / path).resolve()
+    wiki = (ROOT / "wiki").resolve()
+    # 信任邊界：擋路徑跳脫，只允許 wiki/ 下的 .md
+    if not p.is_relative_to(wiki) or p.suffix != ".md":
+        raise ValueError("path 須指向 wiki/ 下的 .md 頁面")
+    if not p.is_file():
+        raise FileNotFoundError(f"頁面不存在：{path}")
+    meta, body = kb.parse_page(p.read_text(encoding="utf-8"))
+    body, truncated = _truncate(body, cfg["limits"]["mcp_response_max_tokens"])
+    return {"frontmatter": meta, "body": body, "truncated": truncated,
+            "source_refs": [str(s) for s in meta.get("sources") or []]}
+
+
+mcp.tool(search_wiki)
+mcp.tool(read_page)
+
+if __name__ == "__main__":
+    mcp.run()

requirements.txt

@@ -0,0 +1,9 @@
+# 全部為本地程式庫，無任何雲端轉換服務（AGENTS.md §1.1）
+pyyaml           # config/models.yaml
+python-docx      # Word → Markdown
+openpyxl         # Excel → Markdown
+pymupdf          # PDF → Markdown
+python-pptx     # PowerPoint → Markdown
+trafilatura      # 網頁快照 → Markdown 正文萃取
+rank-bm25        # tools/search.py BM25（第一版不用向量庫）
+fastmcp          # mcp/server.py 薄殼

tools/convert/convert.py

@@ -0,0 +1,188 @@
+"""統一轉換入口：依副檔名/URL 分派轉換器，登記 SHA-256 至 raw/manifest.json。
+
+流程（AGENTS.md §6.1、§9）：
+  原始檔存入 raw/originals/ 並登記 hash → 轉換 → Markdown 存入 raw/converted/
+  並登記 hash 與原始檔對應 → 之後攝入管線只讀 converted 層。
+
+用法：python tools/convert/convert.py <檔案路徑或URL>... [--dry-run] [--root DIR]
+"""
+import argparse
+import datetime
+import hashlib
+import json
+import pathlib
+import re
+import sys
+import urllib.parse
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+import from_docx
+import from_pdf
+import from_pptx
+import from_web
+import from_xlsx
+
+ROOT = pathlib.Path(__file__).resolve().parents[2]
+CONVERTERS = {
+    ".docx": (from_docx.convert, "from_docx", "docx"),
+    ".xlsx": (from_xlsx.convert, "from_xlsx", "xlsx"),
+    ".pdf": (from_pdf.convert, "from_pdf", "pdf"),
+    ".pptx": (from_pptx.convert, "from_pptx", "pptx"),
+}
+
+
+def _now():
+    return datetime.datetime.now().astimezone().isoformat(timespec="seconds")
+
+
+def _slug(s):
+    s = re.sub(r"[^a-z0-9一-鿿]+", "-", s.lower()).strip("-")
+    return s or "page"
+
+
+def load_manifest(root):
+    p = root / "raw" / "manifest.json"
+    m = json.loads(p.read_text(encoding="utf-8"))
+    if m.get("version") != 1 or "files" not in m:
+        raise SystemExit(f"manifest 結構不符：{p}")
+    return m
+
+
+def save_manifest(root, m):
+    p = root / "raw" / "manifest.json"
+    tmp = p.with_suffix(".json.tmp")
+    tmp.write_text(json.dumps(m, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    tmp.replace(p)  # 原子替換，避免寫壞 manifest
+
+
+def _find_by_sha(m, sha):
+    for path, e in m["files"].items():
+        if e["kind"] == "original" and e["sha256"] == sha:
+            return path
+    return None
+
+
+def _dest_name(dir_, name, sha):
+    """既有同名檔且內容不同時，附 hash 前 8 碼避免覆蓋（raw 不可變）。"""
+    p = dir_ / name
+    if p.exists() and hashlib.sha256(p.read_bytes()).hexdigest() != sha:
+        stem, ext = pathlib.Path(name).stem, pathlib.Path(name).suffix
+        return f"{stem}-{sha[:8]}{ext}"
+    return name
+
+
+def _register_pair(m, root, orig_rel, orig_entry, md_rel, md_text, converter, dry):
+    if not dry:
+        md_path = root / md_rel
+        md_path.write_text(md_text, encoding="utf-8")
+        m["files"][orig_rel] = orig_entry
+        m["files"][md_rel] = {
+            "kind": "converted",
+            "sha256": hashlib.sha256(md_text.encode("utf-8")).hexdigest(),
+            "original_path": orig_rel,
+            "original_sha256": orig_entry["sha256"],
+            "converter": converter,
+            "converted_at": _now(),
+        }
+    print(f"converted: {orig_rel} -> {md_rel}" + (" [dry-run]" if dry else ""))
+
+
+def process_local(root, m, path, dry):
+    src = pathlib.Path(path).resolve()
+    if not src.is_file():
+        raise FileNotFoundError(src)
+    ext = src.suffix.lower()
+    if ext in (".html", ".htm"):
+        conv, conv_name, media = (
+            lambda p, assets_dir=None, md_dir=None: from_web.extract(
+                pathlib.Path(p).read_text(encoding="utf-8", errors="replace")),
+            "from_web", "html")
+    elif ext in CONVERTERS:
+        conv, conv_name, media = CONVERTERS[ext]
+    else:
+        raise ValueError(f"不支援的格式：{src.name}")
+
+    sha = hashlib.sha256(src.read_bytes()).hexdigest()
+    if _find_by_sha(m, sha):
+        print(f"skip: {src.name} 已登記（hash 相同）")
+        return
+
+    originals = root / "raw" / "originals"
+    if src.parent == originals.resolve():
+        dest = src
+    else:
+        dest = originals / _dest_name(originals, src.name, sha)
+        if not dry and not dest.exists():
+            dest.write_bytes(src.read_bytes())
+    orig_rel = f"raw/originals/{dest.name}"
+    orig_entry = {"kind": "original", "sha256": sha, "media_type": media,
+                  "added_at": _now(), "status": "converted"}
+
+    md_name = f"{dest.stem}-{sha[:8]}.md"
+    assets = None if dry else root / "raw" / "converted" / "assets" / f"{dest.stem}-{sha[:8]}"
+    md_text, status = conv(str(src), assets_dir=assets, md_dir=root / "raw" / "converted")
+    if status == "needs_ocr":
+        orig_entry["status"] = "needs_ocr"
+        if not dry:
+            m["files"][orig_rel] = orig_entry
+        print(f"needs_ocr: {orig_rel} 無文字層，已登記並跳過（不擅自 OCR）"
+              + (" [dry-run]" if dry else ""))
+        return
+    _register_pair(m, root, orig_rel, orig_entry, f"raw/converted/{md_name}",
+                   md_text, conv_name, dry)
+
+
+def process_url(root, m, url, dry):
+    html = from_web.fetch(url)
+    sha = hashlib.sha256(html.encode("utf-8")).hexdigest()
+    if _find_by_sha(m, sha):
+        print(f"skip: {url} 已登記（快照 hash 相同）")
+        return
+    u = urllib.parse.urlparse(url)
+    slug = _slug(f"{u.netloc}{u.path}")[:80]
+    originals = root / "raw" / "originals"
+    name = _dest_name(originals, f"{slug}.html", sha)
+    orig_rel = f"raw/originals/{name}"
+    orig_entry = {"kind": "original", "sha256": sha, "media_type": "html",
+                  "added_at": _now(), "status": "converted",
+                  "source_url": url, "fetched_at": _now()}
+    if not dry:
+        (originals / name).write_text(html, encoding="utf-8")  # 完整快照（來源可能消失）
+    md_text, status = from_web.extract(html)
+    if status != "ok":
+        orig_entry["status"] = "pending"
+        if not dry:
+            m["files"][orig_rel] = orig_entry
+        print(f"pending: {url} 萃取不到正文，僅保存快照" + (" [dry-run]" if dry else ""))
+        return
+    md_name = f"{pathlib.Path(name).stem}-{sha[:8]}.md"
+    _register_pair(m, root, orig_rel, orig_entry, f"raw/converted/{md_name}",
+                   md_text, "from_web", dry)
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("inputs", nargs="+", help="檔案路徑或 http(s) URL")
+    ap.add_argument("--dry-run", action="store_true")
+    ap.add_argument("--root", default=str(ROOT), help="專案根目錄（測試用）")
+    a = ap.parse_args(argv)
+    root = pathlib.Path(a.root).resolve()
+    m = load_manifest(root)
+    failed = []
+    for item in a.inputs:
+        try:
+            if item.startswith(("http://", "https://")):
+                process_url(root, m, item, a.dry_run)
+            else:
+                process_local(root, m, item, a.dry_run)
+        except Exception as e:  # 單檔失敗不中斷批次，最後彙總報錯
+            failed.append((item, e))
+            print(f"error: {item}: {e}", file=sys.stderr)
+    if not a.dry_run:
+        save_manifest(root, m)
+    if failed:
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/from_docx.py

@@ -0,0 +1,96 @@
+"""Word (.docx) → Markdown。保留標題層級與表格結構，圖片抽出為附檔並留佔位引用。
+
+可獨立執行：python tools/convert/from_docx.py <file.docx> [-o out.md] [--dry-run]
+"""
+import argparse
+import pathlib
+import re
+
+import docx
+from docx.oxml.ns import qn
+from docx.table import Table
+from docx.text.paragraph import Paragraph
+
+
+def _heading_level(par):
+    # ponytail: 只認 "Heading N" / "標題 N" 樣式名；其他語系的 Word 樣式名不涵蓋，
+    # 升級路徑：改讀 w:outlineLvl。
+    m = re.match(r"(?:Heading|標題)\s*(\d)", par.style.name or "")
+    return int(m.group(1)) if m else None
+
+
+def _table_md(tbl):
+    rows = [
+        [c.text.strip().replace("\n", " ").replace("|", "\\|") for c in row.cells]
+        for row in tbl.rows
+    ]
+    if not rows:
+        return ""
+    width = max(len(r) for r in rows)
+    rows = [r + [""] * (width - len(r)) for r in rows]
+    out = ["| " + " | ".join(rows[0]) + " |", "|" + " --- |" * width]
+    out += ["| " + " | ".join(r) + " |" for r in rows[1:]]
+    return "\n".join(out)
+
+
+def _par_md(par, doc, assets_dir, md_dir, counter):
+    parts = []
+    for run in par.runs:
+        for blip in run._element.findall(".//" + qn("a:blip")):
+            rid = blip.get(qn("r:embed"))
+            if not rid:
+                continue
+            part = doc.part.related_parts[rid]
+            ext = pathlib.Path(part.partname).suffix or ".png"
+            counter[0] += 1
+            name = f"img{counter[0]}{ext}"
+            if assets_dir is not None:
+                assets_dir.mkdir(parents=True, exist_ok=True)
+                (assets_dir / name).write_bytes(part.blob)
+                rel = (assets_dir / name).relative_to(md_dir).as_posix()
+            else:
+                rel = name  # dry-run：僅佔位
+            parts.append(f"![{name}]({rel})")
+        parts.append(run.text)
+    text = "".join(parts).strip()
+    lvl = _heading_level(par)
+    if lvl and text:
+        return "#" * lvl + " " + text
+    return text
+
+
+def convert(src, assets_dir=None, md_dir=None):
+    """回傳 (markdown, status)。status 恆為 'ok'（docx 必有文字層）。"""
+    src = pathlib.Path(src)
+    md_dir = pathlib.Path(md_dir) if md_dir else src.parent
+    doc = docx.Document(str(src))
+    counter = [0]
+    blocks = []
+    for child in doc.element.body.iterchildren():
+        if child.tag == qn("w:p"):
+            blocks.append(_par_md(Paragraph(child, doc), doc, assets_dir, md_dir, counter))
+        elif child.tag == qn("w:tbl"):
+            blocks.append(_table_md(Table(child, doc)))
+    md = "\n\n".join(b for b in blocks if b)
+    return md, "ok"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("src")
+    ap.add_argument("-o", "--output")
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    if a.dry_run:
+        md, status = convert(a.src)
+        print(f"[dry-run] {a.src}: status={status}, {len(md)} chars, "
+              f"{md.count(chr(10)) + 1} lines")
+        return
+    out = pathlib.Path(a.output) if a.output else pathlib.Path(a.src).with_suffix(".md")
+    md, _ = convert(a.src, assets_dir=out.parent / f"{out.stem}-assets", md_dir=out.parent)
+    out.write_text(md, encoding="utf-8")
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/from_pdf.py

@@ -0,0 +1,70 @@
+"""PDF → Markdown。無文字層（掃描件）回報 needs_ocr 並跳過，不擅自呼叫 OCR。
+
+可獨立執行：python tools/convert/from_pdf.py <file.pdf> [-o out.md] [--dry-run]
+"""
+import argparse
+import pathlib
+
+import fitz  # pymupdf
+
+# ponytail: 掃描件偵測用「平均每頁字元數 < 25」的 naive heuristic，
+# 混合型 PDF（部分頁掃描）會整份判為有文字層；升級路徑：逐頁判定 + 逐頁標記。
+_MIN_CHARS_PER_PAGE = 25
+
+
+def convert(src, assets_dir=None, md_dir=None):
+    """回傳 (markdown, status)。status ∈ {'ok', 'needs_ocr'}。
+
+    ponytail: 表格輸出為純文字行（不重建表格結構），升級路徑：page.find_tables()。
+    """
+    src = pathlib.Path(src)
+    md_dir = pathlib.Path(md_dir) if md_dir else src.parent
+    doc = fitz.open(str(src))
+    pages = [page.get_text("text").strip() for page in doc]
+    total = sum(len(p) for p in pages)
+    if total < _MIN_CHARS_PER_PAGE * max(len(pages), 1):
+        doc.close()
+        return "", "needs_ocr"
+
+    counter = 0
+    blocks = []
+    for i, page in enumerate(doc):
+        blocks.append(f"<!-- page {i + 1} -->\n\n{pages[i]}")
+        if assets_dir is None:
+            continue
+        for img in page.get_images(full=True):
+            xref = img[0]
+            pix = fitz.Pixmap(doc, xref)
+            if pix.n > 4:  # CMYK 等 → 轉 RGB
+                pix = fitz.Pixmap(fitz.csRGB, pix)
+            counter += 1
+            name = f"img{counter}.png"
+            assets_dir.mkdir(parents=True, exist_ok=True)
+            pix.save(str(assets_dir / name))
+            rel = (assets_dir / name).relative_to(md_dir).as_posix()
+            blocks.append(f"![{name}]({rel})")
+    doc.close()
+    return "\n\n".join(blocks), "ok"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("src")
+    ap.add_argument("-o", "--output")
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    if a.dry_run:
+        md, status = convert(a.src)
+        print(f"[dry-run] {a.src}: status={status}, {len(md)} chars")
+        return
+    out = pathlib.Path(a.output) if a.output else pathlib.Path(a.src).with_suffix(".md")
+    md, status = convert(a.src, assets_dir=out.parent / f"{out.stem}-assets", md_dir=out.parent)
+    if status == "needs_ocr":
+        print(f"needs_ocr: {a.src} 無文字層，已跳過（不擅自 OCR）")
+        raise SystemExit(2)
+    out.write_text(md, encoding="utf-8")
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/from_pptx.py

@@ -0,0 +1,74 @@
+"""PowerPoint (.pptx) → Markdown。每張投影片一節，含表格、圖片佔位與講者備註。
+
+可獨立執行：python tools/convert/from_pptx.py <file.pptx> [-o out.md] [--dry-run]
+"""
+import argparse
+import pathlib
+
+from pptx import Presentation
+from pptx.enum.shapes import MSO_SHAPE_TYPE
+
+
+def _table_md(tbl):
+    rows = [
+        [c.text.strip().replace("|", "\\|").replace("\n", " ") for c in row.cells]
+        for row in tbl.rows
+    ]
+    if not rows:
+        return ""
+    width = len(rows[0])
+    out = ["| " + " | ".join(rows[0]) + " |", "|" + " --- |" * width]
+    out += ["| " + " | ".join(r) + " |" for r in rows[1:]]
+    return "\n".join(out)
+
+
+def convert(src, assets_dir=None, md_dir=None):
+    """回傳 (markdown, status)。status 恆為 'ok'。"""
+    src = pathlib.Path(src)
+    md_dir = pathlib.Path(md_dir) if md_dir else src.parent
+    prs = Presentation(str(src))
+    counter = 0
+    sections = []
+    for i, slide in enumerate(prs.slides, 1):
+        blocks = [f"## Slide {i}"]
+        for shape in slide.shapes:
+            if shape.has_text_frame and shape.text_frame.text.strip():
+                blocks.append(shape.text_frame.text.strip())
+            elif shape.has_table:
+                blocks.append(_table_md(shape.table))
+            elif shape.shape_type == MSO_SHAPE_TYPE.PICTURE:
+                counter += 1
+                name = f"img{counter}.{shape.image.ext}"
+                if assets_dir is not None:
+                    assets_dir.mkdir(parents=True, exist_ok=True)
+                    (assets_dir / name).write_bytes(shape.image.blob)
+                    rel = (assets_dir / name).relative_to(md_dir).as_posix()
+                else:
+                    rel = name
+                blocks.append(f"![{name}]({rel})")
+        if slide.has_notes_slide:
+            notes = slide.notes_slide.notes_text_frame.text.strip()
+            if notes:
+                blocks.append(f"### 講者備註\n\n{notes}")
+        sections.append("\n\n".join(blocks))
+    return "\n\n".join(sections), "ok"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("src")
+    ap.add_argument("-o", "--output")
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    if a.dry_run:
+        md, status = convert(a.src)
+        print(f"[dry-run] {a.src}: status={status}, {len(md)} chars")
+        return
+    out = pathlib.Path(a.output) if a.output else pathlib.Path(a.src).with_suffix(".md")
+    md, _ = convert(a.src, assets_dir=out.parent / f"{out.stem}-assets", md_dir=out.parent)
+    out.write_text(md, encoding="utf-8")
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/from_web.py

@@ -0,0 +1,53 @@
+"""網頁 URL → 完整 HTML 快照 + Markdown 正文萃取（trafilatura，全程本地處理）。
+
+可獨立執行：python tools/convert/from_web.py <url> [-o out.md] [--snapshot out.html] [--dry-run]
+"""
+import argparse
+import pathlib
+
+import trafilatura
+
+
+def fetch(url):
+    """抓取 URL，回傳完整 HTML 字串。失敗時拋 RuntimeError。"""
+    html = trafilatura.fetch_url(url)
+    if not html:
+        raise RuntimeError(f"抓取失敗：{url}")
+    return html
+
+
+def extract(html):
+    """HTML → Markdown 正文。回傳 (markdown, status)。萃取不到正文時 status='empty'。"""
+    md = trafilatura.extract(
+        html, output_format="markdown", include_tables=True, include_links=True
+    )
+    if not md or not md.strip():
+        return "", "empty"
+    return md.strip(), "ok"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("url")
+    ap.add_argument("-o", "--output")
+    ap.add_argument("--snapshot", help="HTML 快照輸出路徑")
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    html = fetch(a.url)
+    md, status = extract(html)
+    if a.dry_run:
+        print(f"[dry-run] {a.url}: status={status}, snapshot {len(html)} chars, "
+              f"markdown {len(md)} chars")
+        return
+    if status != "ok":
+        print(f"empty: {a.url} 萃取不到正文")
+        raise SystemExit(2)
+    if a.snapshot:
+        pathlib.Path(a.snapshot).write_text(html, encoding="utf-8")
+    out = pathlib.Path(a.output) if a.output else pathlib.Path("page.md")
+    out.write_text(md, encoding="utf-8")
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/from_xlsx.py

@@ -0,0 +1,56 @@
+"""Excel (.xlsx) → Markdown。每個工作表一節，內容輸出為 Markdown 表格。
+
+可獨立執行：python tools/convert/from_xlsx.py <file.xlsx> [-o out.md] [--dry-run]
+"""
+import argparse
+import pathlib
+
+import openpyxl
+
+
+def _cell(v):
+    return "" if v is None else str(v).replace("|", "\\|").replace("\n", " ").strip()
+
+
+def convert(src, assets_dir=None, md_dir=None):
+    """回傳 (markdown, status)。status 恆為 'ok'。
+
+    ponytail: merged cell 只有左上角有值（openpyxl read_only 行為），不展開；
+    升級路徑：非 read_only 模式讀 merged_cells.ranges 補值。
+    """
+    wb = openpyxl.load_workbook(str(src), data_only=True, read_only=True)
+    sections = []
+    for ws in wb.worksheets:
+        rows = [[_cell(v) for v in row] for row in ws.iter_rows(values_only=True)]
+        rows = [r for r in rows if any(r)]  # 去除全空列
+        body = f"## 工作表：{ws.title}\n\n"
+        if not rows:
+            body += "（空白工作表）"
+        else:
+            width = max(len(r) for r in rows)
+            rows = [r + [""] * (width - len(r)) for r in rows]
+            lines = ["| " + " | ".join(rows[0]) + " |", "|" + " --- |" * width]
+            lines += ["| " + " | ".join(r) + " |" for r in rows[1:]]
+            body += "\n".join(lines)
+        sections.append(body)
+    wb.close()
+    return "\n\n".join(sections), "ok"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("src")
+    ap.add_argument("-o", "--output")
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    md, status = convert(a.src)
+    if a.dry_run:
+        print(f"[dry-run] {a.src}: status={status}, {len(md)} chars")
+        return
+    out = pathlib.Path(a.output) if a.output else pathlib.Path(a.src).with_suffix(".md")
+    out.write_text(md, encoding="utf-8")
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/selfcheck.py

@@ -0,0 +1,127 @@
+"""Phase 2 自檢：程式化產生各格式最小測試檔，驗證轉換器與 convert.py 端到端。
+
+執行：.venv/Scripts/python tools/convert/selfcheck.py
+邏輯壞掉時 assert 會失敗（AGENTS.md §13.4：一個可執行的檢查，不用框架）。
+"""
+import json
+import pathlib
+import shutil
+import sys
+import tempfile
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+import convert
+import from_docx
+import from_pdf
+import from_pptx
+import from_web
+import from_xlsx
+
+
+def make_fixtures(d):
+    import docx as docxlib
+    doc = docxlib.Document()
+    doc.add_heading("支付閘道測試報告", level=1)
+    doc.add_paragraph("逾時缺陷於壓力測試重現。")
+    t = doc.add_table(rows=2, cols=2)
+    t.rows[0].cells[0].text = "案例"
+    t.rows[0].cells[1].text = "結果"
+    t.rows[1].cells[0].text = "TC-001"
+    t.rows[1].cells[1].text = "FAIL"
+    doc.save(d / "report.docx")
+
+    import openpyxl
+    wb = openpyxl.Workbook()
+    ws = wb.active
+    ws.title = "測項"
+    ws.append(["編號", "說明"])
+    ws.append(["TC-001", "逾時 | 重試"])
+    wb.save(d / "cases.xlsx")
+
+    import fitz
+    pdf = fitz.open()
+    page = pdf.new_page()
+    page.insert_text((72, 72), "Payment gateway timeout defect reproduced under load test.")
+    pdf.save(d / "text.pdf")
+    pdf.close()
+    scanned = fitz.open()
+    scanned.new_page()  # 無文字層 → 應判 needs_ocr
+    scanned.save(d / "scanned.pdf")
+    scanned.close()
+
+    from pptx import Presentation
+    prs = Presentation()
+    slide = prs.slides.add_slide(prs.slide_layouts[1])
+    slide.shapes.title.text = "結案報告"
+    slide.notes_slide.notes_text_frame.text = "備註：法遵項目全數通過"
+    prs.save(d / "closing.pptx")
+
+    (d / "page.html").write_text(
+        "<html><body><article><h1>AML 測試指引</h1>"
+        + "<p>可疑交易監控測試需涵蓋大額交易與分散交易兩種樣態。</p>" * 8
+        + "</article></body></html>", encoding="utf-8")
+
+
+def main():
+    tmp = pathlib.Path(tempfile.mkdtemp(prefix="ppqa-selfcheck-"))
+    try:
+        fx = tmp / "fx"
+        fx.mkdir()
+        make_fixtures(fx)
+
+        md, st = from_docx.convert(fx / "report.docx")
+        assert st == "ok" and "# 支付閘道測試報告" in md and "| TC-001 | FAIL |" in md, md
+        md, st = from_xlsx.convert(fx / "cases.xlsx")
+        assert st == "ok" and "## 工作表：測項" in md and "逾時 \\| 重試" in md, md
+        md, st = from_pdf.convert(fx / "text.pdf")
+        assert st == "ok" and "<!-- page 1 -->" in md and "timeout defect" in md, md
+        _, st = from_pdf.convert(fx / "scanned.pdf")
+        assert st == "needs_ocr", st
+        md, st = from_pptx.convert(fx / "closing.pptx")
+        assert st == "ok" and "## Slide 1" in md and "法遵項目全數通過" in md, md
+        md, st = from_web.extract((fx / "page.html").read_text(encoding="utf-8"))
+        assert st == "ok" and "可疑交易監控" in md, md
+        print("PASS: 5 個轉換器 + needs_ocr 偵測")
+
+        # convert.py 端到端（沙盒 root）
+        root = tmp / "root"
+        for sub in ("raw/originals", "raw/converted"):
+            (root / sub).mkdir(parents=True)
+        (root / "raw/manifest.json").write_text('{"version": 1, "files": {}}', encoding="utf-8")
+
+        convert.main([str(fx / "report.docx"), str(fx / "scanned.pdf"),
+                      str(fx / "page.html"), "--root", str(root)])
+        m = json.loads((root / "raw/manifest.json").read_text(encoding="utf-8"))
+        origs = {k: v for k, v in m["files"].items() if v["kind"] == "original"}
+        convs = {k: v for k, v in m["files"].items() if v["kind"] == "converted"}
+        assert len(origs) == 3 and len(convs) == 2, m
+        assert origs["raw/originals/scanned.pdf"]["status"] == "needs_ocr"
+        for k, v in convs.items():
+            assert (root / k).is_file(), k
+            assert m["files"][v["original_path"]]["sha256"] == v["original_sha256"]
+        assert (root / "raw/originals/report.docx").is_file()
+        print("PASS: convert.py 端到端 + manifest 對應")
+
+        # 冪等：同檔再跑一次 → skip，manifest 不變
+        before = (root / "raw/manifest.json").read_text(encoding="utf-8")
+        convert.main([str(fx / "report.docx"), "--root", str(root)])
+        assert (root / "raw/manifest.json").read_text(encoding="utf-8") == before
+        print("PASS: 冪等（同 hash 跳過）")
+
+        # dry-run：全新沙盒不落地
+        root2 = tmp / "root2"
+        for sub in ("raw/originals", "raw/converted"):
+            (root2 / sub).mkdir(parents=True)
+        (root2 / "raw/manifest.json").write_text('{"version": 1, "files": {}}', encoding="utf-8")
+        convert.main([str(fx / "report.docx"), "--dry-run", "--root", str(root2)])
+        m2 = json.loads((root2 / "raw/manifest.json").read_text(encoding="utf-8"))
+        assert m2["files"] == {} and not list((root2 / "raw/originals").iterdir())
+        print("PASS: --dry-run 不落地")
+
+        print("ALL PASS")
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    main()

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/lint.py

@@ -0,0 +1,211 @@
+"""wiki 健檢（AGENTS.md §7）：schema 完整性、過期主張、覆蓋缺口、孤兒頁、
+重複/矛盾頁偵測（LLM）。產出 markdown 報告；絕不修改或刪除任何檔案，
+所有項目一律標記待人工核准。
+
+用法：python tools/lint.py [--output reports] [--no-llm] [--root DIR]
+結束碼：0 = 無發現，1 = 有發現（供 CI 判斷）。
+"""
+import argparse
+import datetime
+import difflib
+import itertools
+import json
+import pathlib
+import re
+import sys
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+import kb
+
+# ponytail: 過期門檻寫死 180 天（naive heuristic）；若各類頁面需要不同節奏，
+# 升級路徑：依 type/tags 分級門檻。
+STALE_DAYS = 180
+# ponytail: LLM 逐對比較上限 15 對，避免 O(n²) 呼叫爆量；升級路徑：以 embedding
+# 預篩（Phase 6 之後評估 EmbeddingGemma / snowflake-arctic-embed）。
+MAX_LLM_PAIRS = 15
+SIM_THRESHOLD = 0.6
+
+PAIR_PROMPT = """判斷以下兩個 QA 知識庫 wiki 頁是否「重複」（描述同一事物，應合併）或「矛盾」（對同一事實給出不相容的主張）。只輸出 JSON：
+{{"relation": "duplicate" 或 "contradiction" 或 "none", "explanation": "一句話理由（繁體中文）"}}
+
+頁 A：{a_title} — {a_desc}
+內文節錄：
+<<<
+{a_body}
+>>>
+
+頁 B：{b_title} — {b_desc}
+內文節錄：
+<<<
+{b_body}
+>>>"""
+
+
+def validate_pair(obj):
+    if obj.get("relation") not in ("duplicate", "contradiction", "none"):
+        return "relation 須為 duplicate|contradiction|none"
+    if not isinstance(obj.get("explanation"), str):
+        return "explanation 須為字串"
+    return None
+
+
+def load_pages(root, findings):
+    pages = []
+    for p in kb.iter_pages(root):
+        rel = p.relative_to(root).as_posix()
+        try:
+            meta, body = kb.parse_page(p.read_text(encoding="utf-8"))
+            pages.append((rel, meta, body))
+        except Exception as e:
+            findings.append(("schema", rel, f"無法解析：{e}"))
+    return pages
+
+
+def check_schema(pages, manifest, findings):
+    for rel, meta, _ in pages:
+        for err in kb.validate_meta(meta):
+            findings.append(("schema", rel, err))
+        for s in meta.get("sources") or []:
+            path, _, sha8 = str(s).rpartition("#")
+            entry = manifest["files"].get(path)
+            if entry is None:
+                findings.append(("schema", rel, f"source_ref 不在 manifest：{s}"))
+            elif not entry["sha256"].startswith(sha8):
+                findings.append(("schema", rel, f"source_ref hash 與 manifest 不符：{s}"))
+
+
+def check_stale(pages, findings):
+    today = datetime.date.today()
+    for rel, meta, _ in pages:
+        try:
+            ts = datetime.date.fromisoformat(str(meta.get("timestamp")))
+        except ValueError:
+            continue  # schema 檢查已報
+        age = (today - ts).days
+        if meta.get("status") == "stale":
+            findings.append(("stale", rel, "已標記 stale，待人工複審或更新"))
+        elif age > STALE_DAYS:
+            findings.append(("stale", rel, f"最後更新 {age} 天前，建議人工複審後標記 stale"))
+
+
+def check_coverage(pages, manifest, findings):
+    referenced = {str(s).rsplit("#", 1)[-1]
+                  for _, meta, _ in pages for s in meta.get("sources") or []}
+    for path, e in manifest["files"].items():
+        if e["kind"] == "converted" and e["sha256"][:8] not in referenced:
+            findings.append(("coverage", path, "已轉換但沒有任何 wiki 頁引用（攝入缺口）"))
+        if e["kind"] == "original" and e.get("status") == "needs_ocr":
+            findings.append(("coverage", path, "needs_ocr：掃描件待人工決定 OCR 方案"))
+
+
+def check_orphans(root, pages, findings):
+    # ponytail: 以「檔名出現在任何連結目標中」判斷被連結，不解析相對路徑；
+    # 升級路徑：正規化解析所有 markdown 連結。
+    link_targets = ""
+    idx = root / "index.md"
+    if idx.is_file():
+        link_targets += idx.read_text(encoding="utf-8")
+    for _, _, body in pages:
+        link_targets += body
+    linked = set(re.findall(r"\(([^)]+\.md)\)", link_targets))
+    linked_names = {pathlib.PurePosixPath(t).name for t in linked}
+    for rel, _, _ in pages:
+        if pathlib.PurePosixPath(rel).name not in linked_names:
+            findings.append(("orphan", rel, "不被 index.md 或任何其他頁連結"))
+
+
+def candidate_pairs(pages):
+    cands = {}
+    for (ra, ma, ba), (rb, mb, bb) in itertools.combinations(pages, 2):
+        sim = difflib.SequenceMatcher(
+            None, f"{ma['title']} {ma['description']}",
+            f"{mb['title']} {mb['description']}").ratio()
+        shared_tags = len(set(ma.get("tags") or []) & set(mb.get("tags") or []))
+        if sim >= SIM_THRESHOLD or shared_tags >= 2:
+            cands[(ra, rb)] = (sim, (ma, ba), (mb, bb))
+    ranked = sorted(cands.items(), key=lambda kv: -kv[1][0])
+    return ranked[:MAX_LLM_PAIRS]
+
+
+def check_pairs_llm(cfg, pages, findings):
+    for (ra, rb), (sim, (ma, ba), (mb, bb)) in candidate_pairs(pages):
+        try:
+            obj, _ = kb.llm_json(cfg, "lint", PAIR_PROMPT.format(
+                a_title=ma["title"], a_desc=ma["description"], a_body=ba[:1500],
+                b_title=mb["title"], b_desc=mb["description"], b_body=bb[:1500]),
+                validate_pair)
+        except RuntimeError as e:
+            findings.append(("llm-pair", f"{ra} ↔ {rb}", f"LLM 比對失敗：{e}"))
+            continue
+        if obj["relation"] == "duplicate":
+            findings.append(("duplicate", f"{ra} ↔ {rb}", f"疑似重複：{obj['explanation']}"))
+        elif obj["relation"] == "contradiction":
+            findings.append(("contradiction", f"{ra} ↔ {rb}", f"疑似矛盾：{obj['explanation']}"))
+
+
+SECTION_TITLES = {
+    "schema": "Schema 完整性", "stale": "過期主張（staleness）",
+    "coverage": "覆蓋缺口", "orphan": "孤兒頁",
+    "duplicate": "重複頁（LLM 判定）", "contradiction": "矛盾頁（LLM 判定）",
+    "llm-pair": "LLM 比對失敗（需重跑）",
+}
+
+
+def write_report(out_dir, findings, n_pages, llm_note):
+    now = datetime.datetime.now()
+    out_dir.mkdir(parents=True, exist_ok=True)
+    path = out_dir / f"lint-{now.strftime('%Y%m%d-%H%M%S')}.md"
+    counts = {}
+    for check, _, _ in findings:
+        counts[check] = counts.get(check, 0) + 1
+    lines = [f"# Lint 報告 — {now.strftime('%Y-%m-%d %H:%M')}", "",
+             f"- 檢查頁面數：{n_pages}",
+             f"- 發現項目：{len(findings)}（" + "、".join(
+                 f"{SECTION_TITLES[k]} {v}" for k, v in counts.items()) + "）"
+             if findings else "- 發現項目：0",
+             f"- LLM 比對：{llm_note}", "",
+             "> lint 絕不修改或刪除任何檔案；以下所有項目皆**待人工核准**後才處置。", ""]
+    for check in SECTION_TITLES:
+        items = [(w, msg) for c, w, msg in findings if c == check]
+        if not items:
+            continue
+        lines += [f"## {SECTION_TITLES[check]}", ""]
+        lines += [f"- [ ] `{w}` — {msg}" for w, msg in items]
+        lines.append("")
+    path.write_text("\n".join(lines), encoding="utf-8")
+    return path
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--output", default="reports")
+    ap.add_argument("--no-llm", 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"))
+
+    findings = []
+    pages = load_pages(root, findings)
+    check_schema(pages, manifest, findings)
+    check_stale(pages, findings)
+    check_coverage(pages, manifest, findings)
+    check_orphans(root, pages, findings)
+    if a.no_llm:
+        llm_note = "已略過（--no-llm）"
+    else:
+        llm_note = f"model={cfg['tasks']['lint']['model']}，上限 {MAX_LLM_PAIRS} 對"
+        check_pairs_llm(cfg, pages, findings)
+
+    out = pathlib.Path(a.output)
+    report = write_report(out if out.is_absolute() else root / out,
+                          findings, len(pages), llm_note)
+    print(f"報告：{report}")
+    print(f"發現 {len(findings)} 項" if findings else "無發現")
+    if findings:
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()

tools/search.py

@@ -0,0 +1,69 @@
+"""wiki BM25 全文檢索（第一版不用向量庫，AGENTS.md §0）。
+
+用法：python tools/search.py "查詢詞" [-k 10] [--root DIR]
+輸出：JSON 陣列（path、title、description、tags、score），供 agent 與 MCP 使用。
+"""
+import argparse
+import json
+import pathlib
+import re
+import sys
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+import kb
+from rank_bm25 import BM25Okapi
+
+
+def tokenize(text):
+    """英數字詞 + CJK bigram。
+
+    ponytail: bigram 斷詞無語意理解（同義詞不匹配），天花板是字面重疊；
+    升級路徑：Phase 6 後評估 EmbeddingGemma / snowflake-arctic-embed 向量檢索。
+    """
+    text = text.lower()
+    tokens = re.findall(r"[a-z0-9]+", text)
+    for run in re.findall(r"[一-鿿]+", text):
+        tokens += [run] if len(run) == 1 else [run[i:i + 2] for i in range(len(run) - 1)]
+    return tokens
+
+
+def build_corpus(root):
+    docs = []
+    for p in kb.iter_pages(root):
+        try:
+            meta, body = kb.parse_page(p.read_text(encoding="utf-8"))
+        except ValueError:
+            continue  # 壞頁由 lint 報，檢索直接略過
+        text = (f"{meta['title']} " * 3 + f"{meta['description']} " * 2
+                + " ".join(meta.get("tags") or []) + " " + body)
+        docs.append({"path": p.relative_to(root).as_posix(), "title": str(meta["title"]),
+                     "description": str(meta["description"]),
+                     "tags": [str(t) for t in meta.get("tags") or []],
+                     "tokens": tokenize(text)})
+    return docs
+
+
+def search(root, query, k=10):
+    docs = build_corpus(pathlib.Path(root))
+    if not docs:
+        return []
+    bm = BM25Okapi([d["tokens"] for d in docs])
+    scores = bm.get_scores(tokenize(query))
+    ranked = sorted(zip(docs, scores), key=lambda x: -x[1])[:k]
+    return [{"path": d["path"], "title": d["title"], "description": d["description"],
+             "tags": d["tags"], "score": round(float(s), 4)}
+            for d, s in ranked if s > 0]
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("query")
+    ap.add_argument("-k", type=int, default=10, help="回傳筆數上限（查詢工作流上限 10）")
+    ap.add_argument("--root", default=str(kb.ROOT))
+    a = ap.parse_args(argv)
+    hits = search(a.root, a.query, max(1, min(a.k, 10)))
+    print(json.dumps(hits, ensure_ascii=False, indent=2))
+
+
+if __name__ == "__main__":
+    main()

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()

tools/selfcheck_lint.py

@@ -0,0 +1,115 @@
+"""Phase 4 自檢：沙盒植入各類違規（schema 缺欄位、壞 source_ref、過期頁、
+攝入缺口、needs_ocr、孤兒頁、相似頁對），驗證 lint 全部抓到、產出報告、
+且不修改任何檔案。LLM 比對以假 LLM 驗證矛盾偵測路徑。
+
+執行：.venv/Scripts/python tools/selfcheck_lint.py
+"""
+import json
+import pathlib
+import shutil
+import sys
+import tempfile
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+import kb
+import lint
+
+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: 12000}
+"""
+SHA_A = "abcd1234" + "0" * 56
+SHA_B = "beef5678" + "0" * 56
+
+
+def page(root, rel, meta, body="內文。"):
+    p = root / rel
+    p.write_text(kb.dump_page(meta, body), encoding="utf-8")
+
+
+def build(root):
+    import datetime
+    today = datetime.date.today().isoformat()
+    for sub in ("config", "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(json.dumps({"version": 1, "files": {
+        "raw/converted/good-src.md": {"kind": "converted", "sha256": SHA_A,
+                                      "original_path": "raw/originals/g.docx",
+                                      "original_sha256": SHA_A, "converter": "from_docx",
+                                      "converted_at": "2026-07-01T00:00:00"},
+        "raw/converted/uncovered.md": {"kind": "converted", "sha256": SHA_B,
+                                       "original_path": "raw/originals/u.docx",
+                                       "original_sha256": SHA_B, "converter": "from_docx",
+                                       "converted_at": "2026-07-01T00:00:00"},
+        "raw/originals/scan.pdf": {"kind": "original", "sha256": SHA_B,
+                                   "media_type": "pdf", "added_at": "2026-07-01T00:00:00",
+                                   "status": "needs_ocr"},
+    }}), encoding="utf-8")
+    ref = f"raw/converted/good-src.md#{SHA_A[:8]}"
+    good = {"type": "entity", "title": "支付閘道", "description": "支付閘道模組",
+            "tags": ["gateway"], "timestamp": today, "sources": [ref], "status": "draft"}
+    page(root, "wiki/entities/good.md", good)
+    bad = {"type": "entity", "title": "壞頁", "description": "缺 tags",
+           "timestamp": today, "sources": ["raw/converted/ghost.md#deadbeef"],
+           "status": "draft", "type": "entity"}
+    page(root, "wiki/entities/bad-schema.md", bad)
+    old = dict(good, title="舊概念", type="concept", timestamp="2024-01-01")
+    page(root, "wiki/concepts/old.md", old)
+    dup_a = dict(good, title="支付閘道逾時缺陷", tags=["gateway", "timeout"],
+                 description="逾時缺陷模式")
+    dup_b = dict(good, title="支付閘道逾時缺陷模式", tags=["gateway", "timeout"],
+                 description="逾時的缺陷模式")
+    page(root, "wiki/entities/dup-a.md", dup_a, "重試機制在峰值失效。")
+    page(root, "wiki/entities/dup-b.md", dup_b, "重試機制在峰值一律生效。")
+    (root / "index.md").write_text(
+        "- [支付閘道](wiki/entities/good.md)\n- [舊概念](wiki/concepts/old.md)\n"
+        "- [A](wiki/entities/dup-a.md)\n- [B](wiki/entities/dup-b.md)\n", encoding="utf-8")
+
+
+def run_lint(args):
+    try:
+        lint.main(args)
+        return 0
+    except SystemExit as e:
+        return e.code
+
+
+def main():
+    tmp = pathlib.Path(tempfile.mkdtemp(prefix="ppqa-lint-"))
+    real = kb.llm_json
+    try:
+        root = tmp / "repo"
+        build(root)
+        before = {p: p.read_bytes() for p in root.rglob("*.md")}
+
+        code = run_lint(["--no-llm", "--root", str(root)])
+        assert code == 1, code
+        report = next((root / "reports").glob("lint-*.md")).read_text(encoding="utf-8")
+        for expected in ("缺欄位 tags", "source_ref 不在 manifest",
+                         "old.md", "建議人工複審", "uncovered.md", "攝入缺口",
+                         "needs_ocr", "bad-schema.md` — 不被 index.md"):
+            assert expected in report, f"報告缺少：{expected}\n{report}"
+        assert "good.md" not in report
+        after = {p: p.read_bytes() for p in root.rglob("*.md") if "reports" not in str(p)}
+        assert all(before[p] == after[p] for p in after), "lint 修改了檔案！"
+        print("PASS: schema/stale/coverage/orphan 全抓到，報告產出，檔案零修改")
+
+        kb.llm_json = lambda cfg, task, prompt, validate: (
+            {"relation": "contradiction", "explanation": "重試機制生效與否說法相反"}, "fake:test")
+        code = run_lint(["--root", str(root)])
+        assert code == 1
+        report = sorted((root / "reports").glob("lint-*.md"))[-1].read_text(encoding="utf-8")
+        assert "疑似矛盾" in report and "dup-a.md ↔ wiki/entities/dup-b.md" in report, report
+        print("PASS: LLM 矛盾頁偵測（Phase 6 成功標準的機制驗證）")
+        print("ALL PASS")
+    finally:
+        kb.llm_json = real
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    main()

tools/selfcheck_search.py

@@ -0,0 +1,88 @@
+"""Phase 5 自檢：BM25 檢索排序、MCP search_wiki / read_page、
+路徑跳脫防護（信任邊界）、token 上限截斷。
+
+執行：.venv/Scripts/python tools/selfcheck_search.py
+"""
+import os
+import pathlib
+import shutil
+import sys
+import tempfile
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+import kb
+import search
+
+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: 100, ingest_chunk_max_chars: 12000}
+"""
+
+
+def page(root, rel, title, desc, tags, body):
+    meta = {"type": {"summaries": "summary", "entities": "entity",
+                     "concepts": "concept"}[rel.split("/")[1]],
+            "title": title, "description": desc, "tags": tags,
+            "timestamp": "2026-07-14", "sources": ["raw/converted/x.md#abcd1234"],
+            "status": "draft"}
+    (root / rel).write_text(kb.dump_page(meta, body), encoding="utf-8")
+
+
+def main():
+    tmp = pathlib.Path(tempfile.mkdtemp(prefix="ppqa-search-"))
+    try:
+        root = tmp / "repo"
+        for sub in ("config", "wiki/summaries", "wiki/entities", "wiki/concepts"):
+            (root / sub).mkdir(parents=True)
+        (root / "config/models.yaml").write_text(CFG, encoding="utf-8")
+        page(root, "wiki/entities/payment-gateway.md", "支付閘道", "支付閘道模組",
+             ["gateway"], "壓力測試時出現逾時缺陷，重試機制未生效。")
+        page(root, "wiki/concepts/aml-testing.md", "AML 測試準則", "可疑交易監控測試綜合",
+             ["aml", "compliance"], "大額交易與分散交易樣態的監控測試要求。")
+        page(root, "wiki/summaries/long-doc.md", "長文件摘要", "截斷測試用",
+             ["test"], "逾時。" * 500)
+
+        hits = search.search(root, "支付閘道 逾時")
+        assert hits and hits[0]["path"] == "wiki/entities/payment-gateway.md", hits
+        assert all(h["score"] > 0 for h in hits)
+        hits2 = search.search(root, "可疑交易監控")
+        assert hits2[0]["path"] == "wiki/concepts/aml-testing.md", hits2
+        assert search.search(root, "zzz-nonexistent-term") == []
+        print("PASS: BM25 檢索排序（繁中 bigram）")
+
+        os.environ["PP_QA_ROOT"] = str(root)
+        import importlib
+        sys.path.insert(0, str(pathlib.Path(__file__).parents[1] / "mcp"))
+        server = importlib.import_module("server")
+        res = server.search_wiki("支付閘道 逾時", top_k=99)  # top_k 超限 → 收斂為 10
+        assert res[0]["path"] == "wiki/entities/payment-gateway.md"
+        pg = server.read_page("wiki/entities/payment-gateway.md")
+        assert pg["frontmatter"]["title"] == "支付閘道" and not pg["truncated"]
+        assert pg["source_refs"] == ["raw/converted/x.md#abcd1234"]
+        long = server.read_page("wiki/summaries/long-doc.md")
+        assert long["truncated"] and len(long["body"]) <= 150 and long["source_refs"]
+        print("PASS: MCP search_wiki + read_page（含 token 截斷、source_refs 保留）")
+
+        for bad in ("../AGENTS.md", "wiki/../config/models.yaml", "wiki/entities/x.txt"):
+            try:
+                server.read_page(bad)
+                raise AssertionError(f"應拒絕：{bad}")
+            except (ValueError, FileNotFoundError) as e:
+                assert isinstance(e, ValueError), f"{bad} 應為 ValueError"
+        try:
+            server.read_page("wiki/entities/no-such.md")
+            raise AssertionError("應拋 FileNotFoundError")
+        except FileNotFoundError:
+            pass
+        print("PASS: 路徑跳脫防護（信任邊界）")
+        print("ALL PASS")
+    finally:
+        os.environ.pop("PP_QA_ROOT", None)
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    main()