Audit main daily for knowledge changes that landed without a record

doc/architecture/SPEC/kb_service.md

@@ -127,3 +127,11 @@ stdlib + PyYAML + `cryptography`（`kb` extra）+ `.adapters` + `.knowledge_serv
   在 pause 回执成功前保持原状态（继续占用队列名额），回执后才记为 `approval_withdrawn`，剩余审批不会让同一 head 重新暂存。外部 PR 没有操作列表，L1 发现的退役/删除规则写入 `retirements`/`purges`，合并后同样进入
   退役账本（之后的发版巡检据此 purge）。作者关闭自己的 PR 不计入熔断。只触碰非受治理知识路径的 PR 归入 `general`（若其接收
   人工 PR），否则归入第一个接收人工 PR 的仓库。
+
+## 2026-09-28 每日合并审计
+`audit.audit_main`（调度器每 24 小时）沿 main 的 first-parent 历史从上次审计的提交向后检查：每个相对第一父改动了
+`knowledge/` 的提交，都必须是账本已记录合并的变更集。证明只看结构、从不信提交信息：提交 SHA 为某变更集的 merge SHA，或它是双亲
+合并、第二父正是某个已合并变更集记录的 head（合并队列落地的形态），且它对知识的改动恰好就是该 head 的改动（改动的每个
+知识路径都在该 PR 的改动内且内容与 head 相同；夹带额外改动的合并同样被发现）。否则（admin bypass、直接推送、账本漏记）对其涉及的仓库暂停自动合并（改动在仓库范围之外 → 全局暂停）、
+出队其打开的知识 PR 并转人工；不在 `auto_merge` 的仓库只写 `unrecorded_merge` trace。首次运行只记录基线；晚于
+2 小时宽限期的提交留到下次，避免与调度器记录合并赛跑。暂停逻辑与熔断共用 `Scheduler._pause`。

src/infermatrix_copilot/kb_service/audit.py

@@ -0,0 +1,94 @@
+"""Daily post-merge audit: nothing reaches knowledge/ on main without a record.
+
+The gate constrains every merge path the service knows about; this audit
+catches the ones it does not (an admin bypass, a direct push, a merge the
+ledger missed). It walks main's first-parent history from the last audited
+commit. Every commit whose change (against its first parent) touches
+``knowledge/`` must be the recorded merge of a knowledge change set: its SHA
+is a change set's merge SHA, or it merges a PR the ledger recorded as merged.
+Anything else pauses auto-merge for the repositories it touched (everything,
+when it touched knowledge outside a repository scope) and goes to people;
+repositories not in auto_merge only get a trace record.
+
+Commits younger than ``GRACE`` wait for the next run: the scheduler records a
+merge only when it next looks at the PR, and the audit must not race it.
+"""
+
+from __future__ import annotations
+
+import re
+
+from ..knowledge_service.ops import repo_scope
+
+AUDIT_CURSOR = "audit_main_sha"
+GRACE = 2 * 3600
+_PR_REF = re.compile(r"Merge pull request #(\d+)|\(#(\d+)\)\s*$", re.M)  # reporting only, never proof
+_RECORDED = ("merged",)
+
+
+def _scopes(paths: list[str]) -> set[str]:
+    scopes = set()
+    for path in paths:
+        try:
+            scopes.add(repo_scope(path[len("knowledge/"):]).split("/")[-1])
+        except Exception:  # noqa: BLE001 - knowledge/tools, top-level pages...
+            scopes.add("*")
+    return scopes
+
+
+def _merge_carries_only_the_head(rt, parents: list[str], sha: str, paths: list[str]) -> bool:
+    """A merge of a recorded head changed knowledge ONLY as that head did: every
+    knowledge path it changed was changed by the PR (merge base..head) and
+    holds exactly the head's content. Extra edits in the merge are unrecorded."""
+    first, head = parents
+    base = rt.knowledge.merge_base(first, head)
+    pr_paths = {p for p in rt.knowledge.changed_names(base, head) if p.startswith("knowledge/")}
+    if not set(paths) <= pr_paths:
+        return False
+    return rt.knowledge.blob_ids(sha, paths) == rt.knowledge.blob_ids(head, paths)
+
+
+def audit_main(rt, pause) -> list[str]:
+    """One audit pass; ``pause(scope, reason)`` pauses a repository ("*": all).
+    Returns findings (empty when everything that landed was recorded)."""
+    main = rt.knowledge.fetch()
+    last = rt.ledger.get_cursor("*", AUDIT_CURSOR)
+    if not last:
+        rt.ledger.set_cursor("*", AUDIT_CURSOR, main)  # first run: the baseline
+        return []
+    # proof is structural, never a commit message: the commit IS a recorded
+    # merge SHA, or it is a two-parent merge of the exact head a merged change
+    # set recorded (what the merge queue lands)
+    merge_shas, merged_heads = set(), set()
+    for lifecycle in rt.registry.values():
+        for changeset in rt.ledger.changesets(lifecycle.repo, _RECORDED):
+            if changeset.get("merge_sha"):
+                merge_shas.add(changeset["merge_sha"])
+            if changeset.get("head_sha"):
+                merged_heads.add(changeset["head_sha"])
+    findings: list[str] = []
+    audited = last
+    for sha, parents, committed_at, message in rt.knowledge.first_parent_commits(last, main):
+        if rt.clock() - committed_at < GRACE:
+            break  # too recent to judge: the next run starts here
+        audited = sha
+        paths = [p for p in rt.knowledge.changed_names(parents[0] if parents else "", sha)
+                 if p.startswith("knowledge/")]
+        if not paths:
+            continue
+        refs = [int(a or b) for a, b in _PR_REF.findall(message)]
+        if sha in merge_shas or (len(parents) == 2 and parents[1] in merged_heads
+                                 and _merge_carries_only_the_head(rt, parents, sha, paths)):
+            continue
+        scopes = _scopes(paths)
+        reason = (f"unrecorded knowledge change {sha[:12]}"
+                  f"{' (PR #' + str(refs[0]) + ')' if refs else ' (no PR)'} touching {paths[:3]}")
+        auto = [s for s in sorted(scopes) if s == "*" or getattr(rt.registry.get(s), "auto_merge", False)]
+        for scope in auto:
+            pause(scope, f"daily audit: {reason}")
+        rt.trace("outcome", context={"playbook": "kb-audit", "repo": ",".join(sorted(scopes))},
+                 result={"outcome": "unrecorded_merge", "sha": sha, "prs": refs, "paths": paths[:20],
+                         "paused": auto})
+        findings.append(reason)
+    rt.ledger.set_cursor("*", AUDIT_CURSOR, audited)
+    return findings

src/infermatrix_copilot/kb_service/scheduler.py

@@ -26,6 +26,7 @@
 
 from . import merge
 from .activate import ActivationError, activate, activation_lock
+from .audit import audit_main
 from .external import poll_external
 from .runtime import collect_events, publish, run_intake
 from .sweep import UpstreamRepo, detect_release, run_sweep
@@ -40,6 +41,7 @@ class Scheduler:
     tick_seconds: float = 60.0
     intake_every: float = 900.0
     release_every: float = 3600.0
+    audit_every: float = 24 * 3600.0
     log: list[dict] = field(default_factory=list)
     _last: dict[str, float] = field(default_factory=dict)
 
@@ -74,6 +76,12 @@ def tick(self) -> None:
                 self._repo_tick(lifecycle)
             except Exception as exc:  # isolate repositories from each other
                 self._record(lifecycle.repo, "error", error=repr(exc), trace=traceback.format_exc()[-2000:])
+        if self._due("audit", self.audit_every):
+            try:  # every knowledge change on main must have been recorded
+                for finding in audit_main(rt, self._pause):
+                    self._record("*", "audit", finding=finding)
+            except Exception as exc:
+                self._record("*", "error", error=repr(exc), trace=traceback.format_exc()[-2000:])
         if self._due("external", self.intake_every) and not merge.is_paused(rt.ledger, "*"):
             try:  # knowledge PRs the service did not open (source 4, human approvals)
                 for event in poll_external(rt):
@@ -132,18 +140,25 @@ def _overturn_breaker(self, lifecycle) -> None:
                   if cs["kind"] != "external" and rt.clock() - float(cs["updated_at"]) < OVERTURN_WINDOW]
         state = rt.ledger.repo_state(lifecycle.repo)
         if len(recent) >= OVERTURN_LIMIT and not state["paused"]:
-            reason = f"overturn breaker: {len(recent)} knowledge PRs closed unmerged within 24h"
-            # publish the pause in the same locked transition as the state change,
-            # so the publisher and the gate see it at once, then dequeue open PRs
-            if rt.outbox is not None:
-                rt.outbox.transition(
-                    lambda: rt.ledger.bump_generation(lifecycle.repo, pause=True, reason=reason),
-                    public_repos=self._public_repos())
-            else:
-                rt.ledger.bump_generation(lifecycle.repo, pause=True, reason=reason)
-            merge.pause_open_prs(rt.ledger, rt.outbox, lifecycle.repo, reason)
-            rt.ledger.enqueue_human(lifecycle.repo, reason)
-            self._record(lifecycle.repo, "paused", reason=reason)
+            self._pause(lifecycle.repo, f"overturn breaker: {len(recent)} knowledge PRs closed unmerged within 24h")
+
+    def _pause(self, repo: str, reason: str) -> None:
+        """Pause a repository ("*": all of them) and dequeue its open PRs."""
+        rt = self.rt
+        if rt.ledger.repo_state(repo)["paused"]:
+            rt.ledger.enqueue_human(repo if repo != "*" else next(iter(rt.registry)), reason)
+            return
+        # publish the pause in the same locked transition as the state change,
+        # so the publisher and the gate see it at once, then dequeue open PRs
+        if rt.outbox is not None:
+            rt.outbox.transition(lambda: rt.ledger.bump_generation(repo, pause=True, reason=reason),
+                                 public_repos=self._public_repos())
+        else:
+            rt.ledger.bump_generation(repo, pause=True, reason=reason)
+        for name in ([repo] if repo != "*" else [lc.repo for lc in rt.registry.values()]):
+            merge.pause_open_prs(rt.ledger, rt.outbox, name, reason)
+        rt.ledger.enqueue_human(repo if repo != "*" else next(iter(rt.registry)), reason)
+        self._record(repo, "paused", reason=reason)
 
     def serve(self, stop: threading.Event | None = None, *, once: bool = False) -> None:
         stop = stop or threading.Event()

src/infermatrix_copilot/kb_service/sources.py

@@ -107,6 +107,41 @@ def raw_manifest(self, old: str, new: str) -> list[dict]:
             })
         return entries
 
+    def first_parent_commits(self, old: str, new: str) -> list[tuple[str, list[str], float, str]]:
+        """(sha, parents, commit time, message) on new's first-parent chain
+        after old, oldest first."""
+        out = self._git("log", "--first-parent", "--reverse", "-z", "--format=%H %P%n%ct%n%B",
+                        f"{old}..{new}").decode("utf-8", "replace")
+        commits = []
+        for record in out.split("\0"):
+            if not record.strip():
+                continue
+            head, _, rest = record.lstrip("\n").partition("\n")
+            stamp, _, message = rest.partition("\n")
+            shas = head.split()
+            commits.append((shas[0], shas[1:], float(stamp or 0), message))
+        return commits
+
+    def blob_ids(self, rev: str, paths: list[str]) -> dict[str, str]:
+        """path -> "<mode> <type> <blob id>" at ``rev`` ("" for an absent path):
+        the mode is part of the identity (an executable bit is a change)."""
+        if not paths:
+            return {}
+        out = self._git("ls-tree", "-z", "--full-tree", rev, "--", *paths).split(b"\0")
+        found = {}
+        for entry in out:
+            if entry:
+                meta, _, name = entry.partition(b"\t")
+                found[name.decode("utf-8", "replace")] = meta.decode()
+        return {path: found.get(path, "") for path in paths}
+
+    def changed_names(self, old: str, new: str) -> list[str]:
+        if not old:  # a root commit
+            out = self._git("ls-tree", "-r", "-z", "--name-only", new)
+        else:
+            out = self._git("diff", "--name-only", "-z", "--no-renames", old, new)
+        return [name.decode("utf-8", "replace") for name in out.split(b"\0") if name]
+
     def knowledge_files(self, rev: str) -> dict[str, str]:
         """Knowledge-relative path -> text for the governed tree at ``rev``."""
         files = self._texts(rev, ("knowledge/repos/", "knowledge/general/"), KNOWLEDGE_SUFFIXES)