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