From e94b2a5d3325b1c162f1c48d0b8a07daf964ff0a Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 29 Sep 2026 00:00:05 +0800 Subject: [PATCH] Audit main daily for knowledge changes that landed without a record The gate constrains every merge path the service knows about. The design also requires a daily audit for the paths it does not (an admin bypass, a direct push, a merge the ledger missed): any unrecorded knowledge merge pauses auto-merge and alerts. - audit.audit_main walks main's first-parent history from the last audited commit (a ledger cursor; the first run only records the baseline). Every commit whose change against its first parent touches knowledge/ must be a recorded merge. The proof is structural, never a commit message: the commit's SHA is a change set's merge SHA, or it is a two-parent merge whose second parent is the exact head a merged change set recorded (what the merge queue lands) and that changed knowledge only as that head did (every knowledge path it changed was changed by the PR and holds exactly the head's content and file mode). A direct push titled "(#42)", or a merge of a recorded head that smuggles extra edits, is still caught. Anything else pauses the repositories it touched (everything, for knowledge outside a repository scope), dequeues their open knowledge PRs and goes to people. Repositories not in auto_merge only get an unrecorded_merge trace record. - Commits younger than a 2-hour grace period wait for the next run, so the audit never races the scheduler recording a merge. - The scheduler runs it every 24 hours. The pause logic is now Scheduler._pause, shared with the overturn breaker. - KnowledgeRepo gains first_parent_commits and changed_names. Co-Authored-By: Claude Opus 5.5 --- doc/architecture/SPEC/kb_service.md | 8 + src/infermatrix_copilot/kb_service/audit.py | 94 +++++++++++ .../kb_service/scheduler.py | 39 +++-- src/infermatrix_copilot/kb_service/sources.py | 35 ++++ test/test_kb_audit.py | 156 ++++++++++++++++++ 5 files changed, 320 insertions(+), 12 deletions(-) create mode 100644 src/infermatrix_copilot/kb_service/audit.py create mode 100644 test/test_kb_audit.py diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 7ccd8483..d0af455d 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/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`。 diff --git a/src/infermatrix_copilot/kb_service/audit.py b/src/infermatrix_copilot/kb_service/audit.py new file mode 100644 index 00000000..0e961239 --- /dev/null +++ b/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 diff --git a/src/infermatrix_copilot/kb_service/scheduler.py b/src/infermatrix_copilot/kb_service/scheduler.py index b5aa4187..fb3425ce 100644 --- a/src/infermatrix_copilot/kb_service/scheduler.py +++ b/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() diff --git a/src/infermatrix_copilot/kb_service/sources.py b/src/infermatrix_copilot/kb_service/sources.py index c78b6671..b6432aac 100644 --- a/src/infermatrix_copilot/kb_service/sources.py +++ b/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 -> " " 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) diff --git a/test/test_kb_audit.py b/test/test_kb_audit.py new file mode 100644 index 00000000..1f14eded --- /dev/null +++ b/test/test_kb_audit.py @@ -0,0 +1,156 @@ +"""The daily post-merge audit: unrecorded knowledge changes pause auto-merge.""" + +from __future__ import annotations + +import subprocess +import time +from dataclasses import replace + +from infermatrix_copilot.kb_service.audit import AUDIT_CURSOR, GRACE, audit_main +from infermatrix_copilot.kb_service.scheduler import Scheduler +from test_kb_flow import _flow_runtime +from test_kb_intake_gate import PAGE + + +def _git(cwd, *args): + return subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True, text=True).stdout.strip() + + +def _land(origin, rel: str, text: str, message: str) -> str: + target = origin / rel + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(text) + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", message) + return _git(origin, "rev-parse", "HEAD") + + +def _setup(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + rt.clock = lambda: time.time() + GRACE + 60 # every commit made in the test is old enough + scheduler = Scheduler(rt) + assert audit_main(rt, scheduler._pause) == [] # the first run records the baseline + assert rt.ledger.get_cursor("*", AUDIT_CURSOR) == rt.knowledge.main_sha() + return rt, lifecycle, scheduler, tmp_path / "origin" + + +def test_a_knowledge_change_without_a_record_pauses_auto_merge(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + page = origin / "knowledge" / PAGE + sha = _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", + "hotfix knowledge directly") + (finding,) = audit_main(rt, scheduler._pause) + assert sha[:12] in finding and "(no PR)" in finding + assert rt.ledger.repo_state("demo")["paused"] + assert any("daily audit" in item["reason"] for item in rt.ledger.human_queue("demo")) + assert rt.ledger.get_cursor("*", AUDIT_CURSOR) == sha + assert audit_main(rt, scheduler._pause) == [] # audited once + + +def test_recorded_merges_and_code_changes_pass(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + _land(origin, "src/app.py", "x = 1\n", "a code change (#40)") + page = origin / "knowledge" / PAGE + merged = _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", + "Merge pull request #42 from JiusiServe/kb/demo/x") + changeset = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.ledger.acquire_lease("t"), "demo", changeset, detail={}, status="merged", + verdicts=[], human_reason="", drafted_events=[]) + rt.ledger.update_changeset(changeset, pr_number=42, merge_sha=merged) + assert audit_main(rt, scheduler._pause) == [] + assert not rt.ledger.repo_state("demo")["paused"] + assert rt.ledger.get_cursor("*", AUDIT_CURSOR) == merged + + +def test_recent_commits_wait_for_the_next_run(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + before = rt.ledger.get_cursor("*", AUDIT_CURSOR) + page = origin / "knowledge" / PAGE + _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", "fresh") + rt.clock = time.time # the commit is younger than the grace period + assert audit_main(rt, scheduler._pause) == [] + assert rt.ledger.get_cursor("*", AUDIT_CURSOR) == before + + +def test_repositories_not_in_auto_merge_are_only_traced(tmp_path): + from infermatrix_copilot.trace_store import TraceStore + + rt, lifecycle, scheduler, origin = _setup(tmp_path) + rt.registry["demo"] = replace(rt.registry["demo"], mode="shadow") + rt.traces = TraceStore(tmp_path / "traces", environ={}) + page = origin / "knowledge" / PAGE + _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", "shadow change") + assert len(audit_main(rt, scheduler._pause)) == 1 + assert not rt.ledger.repo_state("demo")["paused"] + (record,) = rt.traces.query(kind="outcome") + assert record["result"]["outcome"] == "unrecorded_merge" and record["result"]["paused"] == [] + + +def test_knowledge_outside_a_repository_scope_pauses_everything(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + _land(origin, "knowledge/tools/check.md", "changed around the gate\n", "tools") + assert len(audit_main(rt, scheduler._pause)) == 1 + assert rt.ledger.repo_state("*")["paused"] + + +def test_a_commit_message_citing_a_recorded_pr_is_no_proof(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + page = origin / "knowledge" / PAGE + merged = _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", + "Merge pull request #42 from JiusiServe/kb/demo/x") + changeset = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.ledger.acquire_lease("t"), "demo", changeset, detail={}, status="merged", + verdicts=[], human_reason="", drafted_events=[]) + rt.ledger.update_changeset(changeset, pr_number=42, merge_sha=merged) + sneaky = _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", + "Follow-up correction (#42)") + (finding,) = audit_main(rt, scheduler._pause) + assert sneaky[:12] in finding and rt.ledger.repo_state("demo")["paused"] + + +def test_a_queue_merge_of_the_recorded_head_is_proof(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + _git(origin, "checkout", "-q", "-b", "kb/demo/y") + page = origin / "knowledge" / PAGE + head = _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", "knowledge change") + _git(origin, "checkout", "-q", "main") + _git(origin, "-c", "user.name=q", "-c", "user.email=q@e", "merge", "-q", "--no-ff", "-m", "queue merge", head) + changeset = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.ledger.acquire_lease("t"), "demo", changeset, detail={}, status="merged", + verdicts=[], human_reason="", drafted_events=[]) + rt.ledger.update_changeset(changeset, pr_number=43, head_sha=head, merge_sha="") + assert audit_main(rt, scheduler._pause) == [] + + +def test_a_merge_of_a_recorded_head_with_extra_edits_is_caught(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + _git(origin, "checkout", "-q", "-b", "kb/demo/z") + page = origin / "knowledge" / PAGE + head = _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", "knowledge change") + _git(origin, "checkout", "-q", "main") + _git(origin, "-c", "user.name=q", "-c", "user.email=q@e", "merge", "-q", "--no-ff", "--no-commit", head) + (origin / "knowledge" / "repos" / "demo" / "core" / "_index.md").write_text("# core\n\nsmuggled\n") + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=q", "-c", "user.email=q@e", "commit", "-q", "-m", "merge with extras") + changeset = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.ledger.acquire_lease("t"), "demo", changeset, detail={}, status="merged", + verdicts=[], human_reason="", drafted_events=[]) + rt.ledger.update_changeset(changeset, pr_number=44, head_sha=head, merge_sha="") + (finding,) = audit_main(rt, scheduler._pause) + assert "unrecorded knowledge change" in finding and rt.ledger.repo_state("demo")["paused"] + + +def test_a_merge_that_adds_an_executable_bit_is_caught(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + _git(origin, "checkout", "-q", "-b", "kb/demo/m") + page = origin / "knowledge" / PAGE + head = _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", "knowledge change") + _git(origin, "checkout", "-q", "main") + _git(origin, "-c", "user.name=q", "-c", "user.email=q@e", "merge", "-q", "--no-ff", "--no-commit", head) + _git(origin, "update-index", "--chmod=+x", f"knowledge/{PAGE}") + _git(origin, "-c", "user.name=q", "-c", "user.email=q@e", "commit", "-q", "-m", "merge with a mode change") + changeset = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.ledger.acquire_lease("t"), "demo", changeset, detail={}, status="merged", + verdicts=[], human_reason="", drafted_events=[]) + rt.ledger.update_changeset(changeset, pr_number=45, head_sha=head, merge_sha="") + assert len(audit_main(rt, scheduler._pause)) == 1