From 09303ec4e2e560fa253db481f268864adfe53f79 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 29 Sep 2026 14:24:50 +0800 Subject: [PATCH] Catch, block and revert knowledge changes that bypassed the gate (design v8) Without GitHub-side enforcement, a person can still merge or push a knowledge change the gate never saw. The service now catches it on the next tick, keeps it from being served, and opens a revert. - audit (provenance) runs every scheduler tick, no longer daily. Each first-parent commit on main that touches knowledge/ is classified as: - trusted: a recorded merge SHA, or a two-parent merge of a recorded head carrying only that head's change; - pending: a merge of a head whose merge item still waits for its receipt; looked at again next tick; - disposed: an exact revert of it was merged; - unknown: anything else. An inexact revert merge is itself unknown. - For an unknown commit the service: - pauses the repositories it touched (everything outside a repository scope); - records unknown: and alerts people; - sends its content back to intake as a new candidate (event source "unrecorded"); - opens a revert PR that restores the governed paths to their first-parent content. The publisher builds it and it is NEVER merged automatically; open_revert_pr is always executable, since it is issued because the repository was paused. A path changed again since (the revert would conflict), or a non-governed path, goes to people instead. Without the lease the revert is recorded as revert_pending and issued by the next pass that holds it. - A merged revert disposes of the unknown commit only if it is exact: the PR merged the revert commit the publisher built, and on governed paths the merge changed exactly the reverted paths back to their pre-images. disposed: records the pair. A closed revert leaves it undisposed and alerts people. - Exactness compares tree entries (mode and content). A revert made by hand is recognised the same way. The set is every undisposed unknown commit sharing a knowledge path with the new commit, closed transitively (an earlier partial revert is itself one of them). The new commit disposes of the set when it changes only the set's paths and, after it, EVERY path the set touched, governed or not, holds its tree entry from before the earliest commit (ordered by ancestry): together they changed nothing. The unknown record keeps the complete path list and the entries of all its knowledge paths, uncapped, so conflict detection and exactness cover every path. A disposed commit is trusted only while its disposing revert is part of the history being checked, so main rewound past the revert brings it back. A revert PR or intake candidate is created only when EVERY repository scope the commit touches may publish; a commit touching a private upstream goes to people. A merged revert PR is settled only after fetching, so a fresh merge is never mistaken for another unknown change. - activate() itself, and so `kb activate`, refuses any SHA with an unknown or pending knowledge commit between the active snapshot and it (the first activation has no baseline). The scheduler checks the same before activating and records activation_blocked, and nothing is activated while an unknown commit is undisposed (the G5 exception in the design). - The intake event for an unknown commit carries its diff, not just its message and file names. - The signed control record now carries provenance: the active snapshot and the trusted list (merge SHAs plus disposal pairs). Before merging, the publisher checks that every governed knowledge commit on main since the active snapshot is trusted or one of its own merges; otherwise it refuses (transient: the service pauses and reverts). Not implemented yet: kb accept-unknown (accepting the unknown content instead of reverting). For now a merged exact revert, the service's or one made by hand, lifts the block. Tests: an unknown change is paused, blocks activation, is reverted and sent back to intake, and is handled once; an exact revert disposes of it and both commits become trusted; a revert merged with other changes disposes of nothing and is caught itself; a conflicting revert goes to people; nothing is activated while an unknown change is open, and activation resumes after the revert; the publisher refuses to merge on top of an unknown change; our own merge waiting for its receipt is not unknown; a revert merged with a mode change is not exact; a hand-made revert disposes of what it undoes; activate() refuses a commit that passed no gate; the intake event carries the diff; a revert is settled without the test pre-fetching; a revert must restore non-governed paths too (a partial revert keeps the block, and a second completes it); a hand revert of an AGENTS-only change clears it; a later edit to the 54th of 55 paths makes the revert conflict; rewinding main past a revert brings the unknown change back; a commit touching a private scope is never reverted in public. Co-Authored-By: Claude Opus 5.5 --- doc/architecture/SPEC/kb_service.md | 18 + doc/features/knowledge-service.md | 4 +- .../kb_service/activate.py | 15 +- src/infermatrix_copilot/kb_service/audit.py | 339 +++++++++++++++--- src/infermatrix_copilot/kb_service/merge.py | 23 +- src/infermatrix_copilot/kb_service/outbox.py | 13 +- .../kb_service/publisher.py | 34 +- .../kb_service/scheduler.py | 14 +- src/infermatrix_copilot/kb_service/sources.py | 18 + test/test_kb_audit.py | 34 +- test/test_kb_flow.py | 34 +- test/test_kb_provenance.py | 279 ++++++++++++++ 12 files changed, 738 insertions(+), 87 deletions(-) create mode 100644 test/test_kb_provenance.py diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index e7d16e44..8fcbed3b 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -241,3 +241,21 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝 发布器记住每个 PR 已发布的最新修订，较旧的项不会覆盖较新的评论；服务在账本（全局 `findings:`，每个 PR 一条，与其涉及的仓库 范围无关）记下最新一条直至发布器确认送达，过期仍未送达的以相同内容与原修订号重新签发（与 PR 当前状态无关），因此旧发现永远 不会覆盖新的；发布器的修订记录原子写入，损坏时视为无记录。 + +## 2026-09-29 v8：来历检查与撤回 +`audit`（原每日审计）改为每个调度 tick 运行，沿 main 的第一父链检查每个触碰 `knowledge/` 的提交：可信（账本记录的合并 SHA，或只携带 +已记录 head 知识变化的双亲合并）、待回执（合并项仍在等待回执的 head，下一 tick 再看）、已处置（精确撤回已合并）、未知。 +- 未知提交：暂停其涉及的仓库（触及仓库范围之外则全部暂停），记录 `unknown:`，告警，并把其内容作为新候选事件（`unrecorded`） + 送回 intake；对受治理路径生成撤回 PR（恢复为第一父内容，由发布器构建、永不自动合并；`open_revert_pr` 属于总可执行项）。 + 路径之后又被修改（会冲突）或涉及非受治理路径时不生成撤回、转人工。无租约时记为 `revert_pending`，由下一次持有租约的检查补发。 +- 撤回 PR 由人合并：只有 PR 合并的正是发布器构建的撤回提交、且合并提交在受治理路径上的变化恰为撤回路径并恢复到前像，才记录 + `disposed:` 处置配对；不精确的撤回合并本身也是未知提交。关闭撤回 PR 则保持未处置并告警。 +- 存在未处置的未知提交时不激活任何快照（G5 例外）；可信列表（合并 SHA 与处置配对）写入签名控制记录的 `provenance`，发布器合并前 + 核对自当前激活快照以来 main 上每个受治理知识提交都可信（或是其自己的合并），否则不合并（瞬时，服务随即暂停并撤回）。 +- 精确性按树条目（模式 + 内容）比较；手工撤回同样被识别：与某提交共享知识路径的未处置未知提交（传递闭包，先前的部分撤回也是其中之一） + 构成一组；若该提交只改动这组涉及的路径，且之后这组触及的每条知识路径（不限受治理路径）都恢复到最早一个提交之前的树条目 + （按祖先关系排序），则整组被处置。未知记录保存完整路径列表与全部知识路径的前像，冲突检查覆盖所有路径。已处置的未知提交只在其处置撤回属于被检查的历史时 + 才可信（main 回退到撤回之前会使其重新成为未知）；只要涉及的任一仓库范围为私有上游，就不开公开撤回 PR、不生成候选事件，只转人工。结算撤回前先拉取，避免把新合并的撤回误判为未知。 +- 激活本身（`activate()`，含 `kb activate`）核对从当前激活快照到目标 SHA 的每个知识提交都可信（首次激活除外），否则 + `ActivationError`；调度器在激活前做同样的检查并只记录 `activation_blocked`。入 intake 的候选事件带上该提交的 diff。 +- 尚未实现：`kb accept-unknown`（接受未知提交内容而不撤回），目前通过合并精确撤回（服务开出的或手工的）解除。 diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md index 003dfcfa..48aed1fa 100644 --- a/doc/features/knowledge-service.md +++ b/doc/features/knowledge-service.md @@ -65,7 +65,9 @@ infermatrix-copilot kb rollback --to # 回指到较早的快照（下一 `main` 与 PR head 的合并树上运行本地门禁，通过才 `gh pr merge`（每仓库同时至多 1 个）→ 合并后激活新快照。 本地门禁拒绝的变更从不合并：上下文变化则在当前 main 上重建；服务自己的变更被质量门或本地门禁拒绝时，由生成器按 理由精炼后整体复检，至多 2 轮，仍失败则留在人工队列（不合并）。他人的知识 PR 每天检查一次：结果写在该 PR 上唯一一条 -“知识质量门”评论里（原地更新），未通过则不合并，作者修改后次日复检。发版巡检在上游新 release 后运行 +“知识质量门”评论里（原地更新），未通过则不合并，作者修改后次日复检。 +有人绕过门禁直接合并/推送知识变更时，服务立即暂停该仓库、停止激活新快照、告警，并开出撤回 PR（不自动合并）； +合并精确的撤回 PR 并 `kb resume` 后恢复，被撤回的内容会作为新候选重新过门禁。发版巡检在上游新 release 后运行 T1/T2/T3 与 purge，每个规则页一个变更集，全部经过质量门。 评审服务读取知识：将 `KNOWLEDGE_ROOT` 指向 `$KB_STATE_DIR/active`。 diff --git a/src/infermatrix_copilot/kb_service/activate.py b/src/infermatrix_copilot/kb_service/activate.py index a36d0002..545ac538 100644 --- a/src/infermatrix_copilot/kb_service/activate.py +++ b/src/infermatrix_copilot/kb_service/activate.py @@ -135,11 +135,20 @@ def switch_active(state_dir: Path, snapshot: Path) -> None: os.replace(tmp, link) -def activate(rt, sha: str, *, locked: bool = False) -> Path: - """Build, verify and switch to the snapshot of knowledge at ``sha``.""" +def activate(rt, sha: str, *, locked: bool = False, verify_provenance: bool = True) -> Path: + """Build, verify and switch to the snapshot of knowledge at ``sha``. Every + knowledge commit from the active snapshot up to ``sha`` must have passed + the gate (design v8): nothing that bypassed it is ever served, whoever + asks (the scheduler, or `kb activate`).""" if not locked: with activation_lock(rt.state_dir): - return activate(rt, sha, locked=True) + return activate(rt, sha, locked=True, verify_provenance=verify_provenance) + if verify_provenance: + from .audit import provenance_problems + + problems = provenance_problems(rt, sha) + if problems: + raise ActivationError("provenance: " + "; ".join(problems[:5])) files = rt.knowledge.knowledge_files(sha) extra = rt.knowledge.top_level_knowledge(sha) snapshot = build_snapshot(rt.state_dir, sha, files, extra=extra) diff --git a/src/infermatrix_copilot/kb_service/audit.py b/src/infermatrix_copilot/kb_service/audit.py index 0e961239..9f173900 100644 --- a/src/infermatrix_copilot/kb_service/audit.py +++ b/src/infermatrix_copilot/kb_service/audit.py @@ -1,29 +1,61 @@ -"""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. +"""Provenance: nothing reaches knowledge/ on main without passing the gate (design v8). + +Without GitHub-side enforcement a person can still merge (or push) a knowledge +change the gate never saw. This module catches it and undoes it. Every +scheduler tick it walks main's first-parent history from the last audited +commit. A commit whose change (against its first parent) touches +``knowledge/`` is one of: + +* trusted: it IS a recorded merge of a knowledge change set (the publisher + merged it after its local gate), or a two-parent merge of a head such a + change set recorded that carries nothing but that head's knowledge change, + or the revert that disposed of an unknown commit; +* pending: a merge of a head whose merge item still waits for its receipt; +* disposed: an unknown commit an EXACT revert was merged for; +* unknown: anything else. + +An unknown commit pauses automatic merging for the repositories it touched +(everything, when it touched knowledge outside a repository scope), alerts +people, goes back into intake with its diff as evidence, and gets a revert +PR: governed paths restored to the commit's first-parent tree entries (mode +and content), built by the publisher and NEVER merged automatically. If a +path changed again after the commit, or the commit touched non-governed +paths, people revert it by hand. + +A revert disposes of unknown commits only if it is exact: it changes exactly +the governed paths they changed, back to exactly the tree entries they had +before the first of them (a revert PR the service opened must also be the +very commit the publisher built). A hand-made revert is recognised the same +way. The pairs are recorded; both commits are then trusted. + +No snapshot is activated past a commit that is unknown or still pending: +``provenance_problems`` checks every commit from the active snapshot up to +the exact SHA ``activate()`` is asked for (so ``kb activate`` cannot bypass +it). The trusted list reaches the publisher through the signed control record. """ from __future__ import annotations +import json import re from ..knowledge_service.ops import repo_scope AUDIT_CURSOR = "audit_main_sha" -GRACE = 2 * 3600 +GRACE = 0 # v8: acted on at once; our own merges waiting for a receipt are "pending", not unknown +UNKNOWN = "unknown:" # "*" cursors: one record per unknown commit +DISPOSED = "disposed:" # "*" cursors: unknown commit -> the exact revert that disposed of it +TRUSTED = "trusted_merges" # "*" cursor: the list the control record carries to the publisher +GOVERNED = ("knowledge/repos/", "knowledge/general/") +SUFFIXES = (".md", ".yaml") _PR_REF = re.compile(r"Merge pull request #(\d+)|\(#(\d+)\)\s*$", re.M) # reporting only, never proof _RECORDED = ("merged",) +TRUSTED_KEEP = 500 +DIFF_LIMIT = 8 * 1024 + + +def _governed(path: str) -> bool: + return path.startswith(GOVERNED) and path.endswith(SUFFIXES) def _scopes(paths: list[str]) -> set[str]: @@ -48,47 +80,264 @@ def _merge_carries_only_the_head(rt, parents: list[str], sha: str, paths: list[s 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() +def disposals(rt) -> dict[str, str]: + return {name[len(DISPOSED):]: value for name, value in rt.ledger.cursors_with_prefix("*", DISPOSED).items()} + + +def _records(rt) -> dict: + merge_shas, merged_heads, pending = set(), set(), set() for lifecycle in rt.registry.values(): for changeset in rt.ledger.changesets(lifecycle.repo, _RECORDED): + if changeset["kind"] == "revert" and not changeset["detail"].get("exact"): + continue # an inexact revert merge is itself unknown if changeset.get("merge_sha"): merge_shas.add(changeset["merge_sha"]) if changeset.get("head_sha"): merged_heads.add(changeset["head_sha"]) + for changeset in rt.ledger.changesets(lifecycle.repo, ("merge_requested",)): + if changeset.get("head_sha"): + pending.add(changeset["head_sha"]) + disposed = disposals(rt) + return {"merges": merge_shas, "heads": merged_heads, "pending": pending, + "disposed": disposed, "disposing": set(disposed.values())} + + +def _classify(rt, sha: str, parents: list[str], paths: list[str], recs: dict, target: str) -> str: + """``target``: the commit the history is checked up to. A disposed commit + is trusted only when the revert that disposed of it is part of that + history (main rewound to before the revert brings it back).""" + if not paths: + return "none" + if sha in recs["merges"] or sha in recs["disposing"]: + return "trusted" + if sha in recs["disposed"]: + return "trusted" if rt.knowledge.is_ancestor(recs["disposed"][sha], target) else "unknown" + if len(parents) == 2 and parents[1] in recs["heads"] and _merge_carries_only_the_head(rt, parents, sha, paths): + return "trusted" + if len(parents) == 2 and parents[1] in recs["pending"]: + return "pending" + return "unknown" + + +def _knowledge_paths(rt, sha: str, parents: list[str]) -> list[str]: + return [p for p in rt.knowledge.changed_names(parents[0] if parents else "", sha) if p.startswith("knowledge/")] + + +def open_unknown(rt) -> list[str]: + """Unknown commits not disposed of yet.""" + disposed = disposals(rt) + return sorted(name[len(UNKNOWN):] for name in rt.ledger.cursors_with_prefix("*", UNKNOWN) + if name[len(UNKNOWN):] not in disposed) + + +def provenance_problems(rt, sha: str) -> list[str]: + """Why ``sha`` may not be activated (empty: every knowledge commit from the + active snapshot up to it is trusted). The very first activation has no + trusted baseline and is allowed.""" + active = rt.ledger.active_snapshot() + if not active or active == sha: + return [] + if not rt.knowledge.is_ancestor(active, sha): + return [f"{sha[:12]} does not descend from the active snapshot {active[:12]}"] + recs = _records(rt) + problems = [] + for commit, parents, _at, _message in rt.knowledge.first_parent_commits(active, sha): + kind = _classify(rt, commit, parents, _knowledge_paths(rt, commit, parents), recs, sha) + if kind in ("unknown", "pending"): + problems.append(f"{commit[:12]} is {kind}") + return problems + + +def publish_trusted(rt) -> None: + """What the publisher may treat as trusted on main (control record).""" + recs = _records(rt) + trusted = sorted(recs["merges"] | set(recs["disposed"]) | recs["disposing"])[-TRUSTED_KEEP:] + rt.ledger.set_cursor("*", TRUSTED, json.dumps(trusted)) + + +def audit_main(rt, pause) -> list[str]: + """One pass; ``pause(scope, reason)`` pauses a repository ("*": all). + Returns findings (empty when everything that landed was trusted).""" + 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 + publish_trusted(rt) + return [] 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 + break + paths = _knowledge_paths(rt, sha, parents) + kind = _classify(rt, sha, parents, paths, _records(rt), main) + if kind == "pending": + break # our own merge; its receipt records it next tick: look again then 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)): + if kind in ("none", "trusted") or rt.ledger.get_cursor("*", UNKNOWN + sha): 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) + if _dispose_by_revert(rt, sha, parents, paths): + continue # an exact revert (by hand, or of a PR merged around the service) + findings.append(_unknown(rt, pause, sha, parents, message, paths, main)) rt.ledger.set_cursor("*", AUDIT_CURSOR, audited) + _issue_pending_reverts(rt, main) + publish_trusted(rt) return findings + + +def _dispose_by_revert(rt, sha: str, parents: list[str], paths: list[str]) -> list[str]: + """Whether ``sha`` completes the revert of open unknown commits. The set + is every open unknown commit sharing a knowledge path with ``sha``, and, + transitively, with those (a partial revert made earlier is itself one of + them). It is disposed of when ``sha`` changes only paths of the set and, + after it, EVERY path the set touched (governed or not) holds exactly the + tree entry it had before the earliest commit of the set: together they + changed nothing. Returns the commits disposed of.""" + import functools + + if not paths or not parents: + return [] + records = {u: json.loads(rt.ledger.get_cursor("*", UNKNOWN + u) or "{}") for u in open_unknown(rt)} + # history order (ancestors first): the earliest commit's pre-image wins + in_order = sorted(records, key=functools.cmp_to_key( + lambda a, b: -1 if rt.knowledge.is_ancestor(a, b) else (1 if rt.knowledge.is_ancestor(b, a) else 0))) + targets, touched = [], set(paths) + grew = True + while grew: + grew = False + for unknown in in_order: + if unknown not in targets and set(records[unknown].get("pre") or {}) & touched: + targets.append(unknown) + touched |= set(records[unknown].get("pre") or {}) + grew = True + if not targets: + return [] + covered: dict[str, str] = {} + for unknown in in_order: + if unknown in targets: + for path, entry in (records[unknown].get("pre") or {}).items(): + covered.setdefault(path, entry) + if not set(paths) <= set(covered) or rt.knowledge.blob_ids(sha, list(covered)) != covered: + return [] + for unknown in targets: + rt.ledger.set_cursor("*", DISPOSED + unknown, sha) + return targets + + +def _unknown(rt, pause, sha: str, parents: list[str], message: str, paths: list[str], main: str) -> str: + refs = [int(a or b) for a, b in _PR_REF.findall(message)] + 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"provenance: {reason}; nothing is activated until it is reverted") + # the COMPLETE path list and every path's entry before the commit: a + # revert is exact only if it restores all of them (reports stay short) + record = {"scopes": sorted(scopes), "paths": paths, "prs": refs, "paused": auto, "state": "open", + "parents": parents, "reason": reason, "at": rt.clock(), + "pre": rt.knowledge.blob_ids(parents[0], paths) if parents else {}} + revert_repo = next((s for s in auto if s != "*"), None) + lifecycle = rt.registry.get(revert_repo) if revert_repo else None + # a revert PR and a candidate both publish the commit's paths and content: + # only when EVERY repository scope it touches may publish + public = all(s == "*" or (rt.registry.get(s) is not None and rt.registry[s].publishes) for s in scopes) + if lifecycle is not None and lifecycle.publishes and public: + record.update(repo=lifecycle.repo, state="revert_pending") + # its content goes back through the gate as a candidate of its own + rt.ledger.record_event(lifecycle.repo, "unrecorded", sha, { + "source_reference": f"commit {sha}", "title": message.splitlines()[0] if message else sha, + "body": message[:3000], "changed_files": paths[:50], + "diff_excerpt": rt.knowledge.diff_text(parents[0] if parents else "", sha, paths)[:DIFF_LIMIT]}) + else: + record["state"] = "people" + rt.ledger.enqueue_human(revert_repo or next(iter(rt.registry)), f"{reason}: revert it by hand") + rt.ledger.set_cursor("*", UNKNOWN + sha, json.dumps(record)) + rt.trace("outcome", context={"playbook": "kb-audit", "repo": ",".join(sorted(scopes))}, + result={"outcome": "unrecorded_merge", "sha": sha, "prs": refs, "paths": paths[:20], "paused": auto}) + return reason + + +def _issue_pending_reverts(rt, main: str) -> None: + """Revert PRs for unknown commits that do not have one yet (staging needs + the scheduler's lease; without it they wait for the next pass).""" + if not getattr(rt, "lease_owner", None): + return + disposed = disposals(rt) + for name, raw in sorted(rt.ledger.cursors_with_prefix("*", UNKNOWN).items(), + key=lambda item: json.loads(item[1]).get("at", 0)): + record = json.loads(raw) + sha = name[len(UNKNOWN):] + lifecycle = rt.registry.get(record.get("repo") or "") + if record.get("state") != "revert_pending" or lifecycle is None or sha in disposed: + continue + record.update(_issue_revert(rt, lifecycle, sha, record, main)) + rt.ledger.set_cursor("*", name, json.dumps(record)) + + +def _issue_revert(rt, lifecycle, sha: str, record: dict, main: str) -> dict: + """A revert PR restoring every governed path to its tree entry before the + commit. Refused (people) when a path is not governed or changed again.""" + from .merge import issue_once + + reason, paths, parents = record["reason"], record["paths"], record.get("parents") or [] + first = parents[0] if parents else "" + other = [p for p in paths if not _governed(p)] + if other or not first: + rt.ledger.enqueue_human(lifecycle.repo, f"{reason}: it touches {other[:3] or 'a root commit'}, " + "which the service does not revert; revert it by hand") + return {"state": "people"} + landed, now = rt.knowledge.blob_ids(sha, paths), rt.knowledge.blob_ids(main, paths) + moved = [p for p in paths if landed[p] != now[p]] + if moved: + rt.ledger.enqueue_human(lifecycle.repo, f"{reason}: a revert would conflict ({moved[:3]} changed again " + "since); revert it by hand (an exact revert unblocks it)") + return {"state": "conflict", "moved": moved} + pre = record["pre"] + if any(entry and not entry.startswith("100644 blob ") for entry in pre.values()): + rt.ledger.enqueue_human(lifecycle.repo, f"{reason}: a path had a non-regular mode before it; revert it by hand") + return {"state": "people"} + files = {p: rt.knowledge.show(first, p) for p, entry in pre.items() if entry} + deleted = [p for p, entry in pre.items() if not entry] + changeset_id = rt.ledger.new_changeset_id(lifecycle.repo, "revert") + rel = lambda p: p[len("knowledge/"):] # noqa: E731 + rt.save_changeset_files(changeset_id, {"base_sha": main, "files": {rel(p): t for p, t in files.items()}, + "deleted": [rel(p) for p in deleted], + "evidence": [{"source_reference": f"commit {sha}", "title": reason}]}) + rt.ledger.stage_intake(rt.lease_owner, lifecycle.repo, changeset_id, kind="revert", status="pr_requested", + verdicts=[], human_reason=f"{reason}: review and merge revert PR (changeset {changeset_id})", + drafted_events=[], detail={"reverts": sha, "expected": pre, "operations": []}) + issue_once(rt, lifecycle.repo, rt.ledger.changeset(changeset_id), "open_revert_pr", { + "changeset_id": changeset_id, "base_sha": main, "branch": f"kb/revert/{sha[:12]}", + "files": files, "deleted": deleted, + "title": f"knowledge: revert {sha[:12]} (merged without passing the knowledge gate)", + "body": (f"{reason}.\n\nThis revert restores the governed knowledge pages to their content before " + f"`{sha}`. It is never merged automatically: merge it to dispose of the unrecorded change " + "(auto-merge then resumes after `kb resume`), or close it and decide otherwise. The reverted " + "content has been queued as a new knowledge candidate and goes through the gate.")}) + return {"state": "reverting", "revert_changeset": changeset_id} + + +def settle_revert(rt, changeset: dict, pr: dict) -> bool: + """A revert PR was merged: record the disposal only if it is exact (the + very commit the publisher built, merged with nothing else on governed + paths, every path back to its tree entry, mode included).""" + rt.knowledge.fetch() # the merge commit may be newer than our clone + merge_sha = str(pr.get("merge_commit_sha") or "") + detail = changeset["detail"] + expected: dict[str, str] = detail.get("expected") or {} + exact = False + if merge_sha and str((pr.get("head") or {}).get("sha") or "") == changeset.get("head_sha"): + parents = rt.knowledge.parents(merge_sha) + if parents: + changed = {p for p in rt.knowledge.changed_names(parents[0], merge_sha) if p.startswith("knowledge/")} + exact = changed == set(expected) and rt.knowledge.blob_ids(merge_sha, list(expected)) == expected + rt.ledger.update_changeset(changeset["id"], status="merged", merge_sha=merge_sha or None, + detail={**detail, "exact": exact}) + if exact: + rt.ledger.set_cursor("*", DISPOSED + detail["reverts"], merge_sha) + else: + rt.ledger.enqueue_human(changeset["repo"], ( + f"revert PR #{changeset['pr_number']} for {detail['reverts'][:12]} was merged but is not the exact " + "revert (modified or with other knowledge changes): the change stays undisposed"), changeset["id"]) + return exact diff --git a/src/infermatrix_copilot/kb_service/merge.py b/src/infermatrix_copilot/kb_service/merge.py index cfba6080..f41cf5bf 100644 --- a/src/infermatrix_copilot/kb_service/merge.py +++ b/src/infermatrix_copilot/kb_service/merge.py @@ -69,6 +69,7 @@ def knowledge_repository() -> str: "post_verdict": ("pr_open", "verdict_posted"), "enqueue": ("verdict_posted", "queued"), "merge": ("merge_requested", "merged"), + "open_revert_pr": ("pr_requested", "revert_open"), } @@ -164,7 +165,7 @@ def apply_acks(rt, acks: list[dict]) -> None: changeset_id) continue fields = {"status": after, "pending_item": None} - if kind in ("open_pr", "open_companion_pr"): + if kind in ("open_pr", "open_companion_pr", "open_revert_pr"): fields.update(pr_number=int(ack["pr"]), head_sha=str(ack["head_sha"]), branch=str(ack.get("branch", ""))) elif kind == "enqueue": @@ -463,7 +464,7 @@ def advance(rt, lifecycle) -> list[str]: paused = is_paused(rt.ledger, lifecycle.repo) merging = rt.ledger.changesets(lifecycle.repo, ("merge_requested",)) for changeset in rt.ledger.changesets(lifecycle.repo, (*IN_FLIGHT, "superseding", "paused", - "companion_open")): + "companion_open", "revert_open")): number = changeset["pr_number"] if number is None: continue @@ -511,6 +512,24 @@ def advance(rt, lifecycle) -> list[str]: knowledge["id"]) events.append(f"companion_closed {changeset['id']}") continue + if changeset["status"] == "revert_open": + # people merge reverts; an exact one disposes of the unknown change + from .audit import settle_revert + + if pr.get("merged"): + try: + exact = settle_revert(rt, changeset, pr) + except Exception as exc: # noqa: BLE001 - e.g. the clone cannot fetch yet: next pass + events.append(f"revert_settle_retry {changeset['id']}: {exc}") + continue + events.append(f"revert_merged {changeset['id']} exact={exact}") + elif pr.get("state") == "closed": + rt.ledger.update_changeset(changeset["id"], status="closed") + rt.ledger.enqueue_human(lifecycle.repo, ( + f"revert PR #{number} was closed: {changeset['detail'].get('reverts', '')[:12]} stays " + "undisposed and activation stays blocked until it is reverted"), changeset["id"]) + events.append(f"revert_closed {changeset['id']}") + continue if changeset["status"] == "superseding": # the replaced PR stays tracked until GitHub shows it closed; the # close item is re-issued whenever the previous one expired diff --git a/src/infermatrix_copilot/kb_service/outbox.py b/src/infermatrix_copilot/kb_service/outbox.py index cabc42cf..adfc9e6a 100644 --- a/src/infermatrix_copilot/kb_service/outbox.py +++ b/src/infermatrix_copilot/kb_service/outbox.py @@ -37,12 +37,15 @@ ITEM_KINDS = ( "open_pr", "open_companion_pr", "post_verdict", "enqueue", - "update_branch", "pause", "close", "open_issue", "merge", "post_findings", + "update_branch", "pause", "close", "open_issue", "merge", "post_findings", "open_revert_pr", ) -ALWAYS_EXECUTABLE = frozenset({"pause", "close"}) +# they only ever stop or undo: executable while a repository is paused (a +# revert is issued BECAUSE the repository was paused for an unknown change) +ALWAYS_EXECUTABLE = frozenset({"pause", "close", "open_revert_pr"}) ITEM_TTL = { "open_pr": 24 * 3600, "open_companion_pr": 24 * 3600, "post_verdict": 30 * 60, "enqueue": 30 * 60, "merge": 30 * 60, "post_findings": 24 * 3600, + "open_revert_pr": 24 * 3600, "update_branch": 24 * 3600, "pause": 24 * 3600, "close": 24 * 3600, "open_issue": 7 * 24 * 3600, } @@ -182,7 +185,11 @@ def _control_payload(self) -> dict: "paused": bool(row["paused"])} for row in self._ledger.all_repo_states() } - return {"issued_at": self._clock(), "repos": repos} + # what the publisher checks main against before every merge: every + # governed knowledge commit since the active snapshot must be trusted + trusted = json.loads(self._ledger.get_cursor("*", "trusted_merges") or "[]") + return {"issued_at": self._clock(), "repos": repos, + "provenance": {"active": self._ledger.active_snapshot() or "", "trusted": trusted}} def refresh_control(self) -> dict: with self.publication_lock(): diff --git a/src/infermatrix_copilot/kb_service/publisher.py b/src/infermatrix_copilot/kb_service/publisher.py index 2c460524..2c20f29a 100644 --- a/src/infermatrix_copilot/kb_service/publisher.py +++ b/src/infermatrix_copilot/kb_service/publisher.py @@ -55,7 +55,7 @@ # so a label that failed to apply cannot make one mergeable by automation COMPANION_BRANCH = re.compile(r"kb/[A-Za-z0-9._-]+/[A-Za-z0-9._-]+-companion-[0-9a-f]+") HOLD_LABEL = "kb:hold" -PUSHING = frozenset({"open_pr", "open_companion_pr"}) +PUSHING = frozenset({"open_pr", "open_companion_pr", "open_revert_pr"}) # retried until done, never acked as failed: stops (pause, close) and reports # (open_issue, post_findings: a transient GitHub error must not lose a sweep # report or an author's findings) @@ -384,7 +384,7 @@ def _handle(self, item_id: str, control: Any) -> str: publishes, auto_merge = self.repo_flags.get(claimed, (False, False)) item = check_item(envelope, control, self.service_public_key, now=self.clock(), repo_publishes=publishes, repo_auto_merge=auto_merge) - self._envelope, self._flags = envelope, (publishes, auto_merge) + self._envelope, self._flags, self._control = envelope, (publishes, auto_merge), control except (OutboxError, OSError, ValueError, KeyError, TypeError) as exc: # not performed and not acked: an unverifiable item is never # answered, and a stale one is superseded by the service itself @@ -464,6 +464,32 @@ def _do_open_pr(self, item: OutboxItem, *, draft: bool = False, allowed: re.Patt raise PublishError(f"the PR for {branch} could not be confirmed after creation") return {"pr": int(opened[0]["number"]), "head_sha": commit, "branch": branch} + def _do_open_revert_pr(self, item: OutboxItem) -> dict: + """A revert of an unknown change: governed pages only, never merged by us.""" + return self._do_open_pr(item) + + def _provenance_problems(self, main_sha: str) -> list[str]: + """Every governed knowledge commit on main since the active snapshot + must be trusted: listed in the signed control record, or a merge this + publisher made itself (its receipt may not have reached the service).""" + payload = (getattr(self, "_control", None) or {}).get("payload") or {} + provenance = payload.get("provenance") or {} + active = str(provenance.get("active") or "") + if not active: + return [] # nothing activated yet: no trusted baseline to compare with + trusted = set(provenance.get("trusted") or []) + done = self.state_dir / "done" + for path in (done.glob("*.json") if done.is_dir() else []): + try: + record = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + if record.get("kind") == "merge" and record.get("merge_sha"): + trusted.add(record["merge_sha"]) + commits = self._git("rev-list", "--first-parent", f"{active}..{main_sha}", "--", + "knowledge/repos", "knowledge/general").split() + return [sha for sha in commits if sha not in trusted] + def _do_open_companion_pr(self, item: OutboxItem) -> dict: """Always a draft, labelled kb:companion, and only within the companion whitelist: people review, ready and merge it; automation never does.""" @@ -590,6 +616,10 @@ def _do_merge(self, item: OutboxItem) -> dict: if self._git("rev-parse", f"refs/kb/pr-{number}") != head: raise PublishError(f"fetched head of PR #{number} is not {head[:12]}") main_sha = self._git("rev-parse", "refs/remotes/origin/main") + unknown = self._provenance_problems(main_sha) + if unknown: + raise PublishError(f"not merging: knowledge changes on main that passed no gate " + f"({', '.join(s[:12] for s in unknown[:3])}); the service pauses and reverts them") problems = local_gate.gate(self.clone, repository=self.github.repository, repo=item.repo, pr=rest, head_sha=head, main_sha=main_sha, verdict=verdict, public_key=self.service_public_key, now=self.clock()) diff --git a/src/infermatrix_copilot/kb_service/scheduler.py b/src/infermatrix_copilot/kb_service/scheduler.py index fe27617f..91688974 100644 --- a/src/infermatrix_copilot/kb_service/scheduler.py +++ b/src/infermatrix_copilot/kb_service/scheduler.py @@ -27,7 +27,7 @@ from . import merge from .activate import ActivationError, activate, activation_lock from .archive import make_archive -from .audit import audit_main +from .audit import audit_main, open_unknown, provenance_problems from .report import flush_reports from .companion import publish_companion from .external import poll_external @@ -45,7 +45,7 @@ class Scheduler: tick_seconds: float = 60.0 intake_every: float = 900.0 release_every: float = 3600.0 - audit_every: float = 24 * 3600.0 + audit_every: float = 0.0 # v8: provenance is checked every tick archive_every: float = 7 * 24 * 3600.0 log: list[dict] = field(default_factory=list) _last: dict[str, float] = field(default_factory=dict) @@ -108,8 +108,14 @@ def tick(self) -> None: if pinned_main == sha: pass # rolled back away from this main: stay there until main moves on elif sha != rt.ledger.active_snapshot(): - activate(rt, sha, locked=True) - self._record("*", "activated", snapshot=sha) + blocked = open_unknown(rt) or provenance_problems(rt, sha) + if blocked: + # a knowledge change that passed no gate is on main: nothing + # new is served until it is reverted (design v8, G5 exception) + self._record("*", "activation_blocked", unknown=[s[:12] for s in blocked]) + else: + activate(rt, sha, locked=True) + self._record("*", "activated", snapshot=sha) except ActivationError as exc: rt.ledger.enqueue_human("*", f"snapshot activation refused: {exc}") self._record("*", "activation_refused", error=str(exc)) diff --git a/src/infermatrix_copilot/kb_service/sources.py b/src/infermatrix_copilot/kb_service/sources.py index 44e4d6f4..71cab98b 100644 --- a/src/infermatrix_copilot/kb_service/sources.py +++ b/src/infermatrix_copilot/kb_service/sources.py @@ -86,6 +86,24 @@ def fetch_pull(self, number: int) -> str: def merge_base(self, a: str, b: str) -> str: return self._git("merge-base", a, b).decode().strip() + def is_ancestor(self, older: str, newer: str) -> bool: + proc = subprocess.run(["git", "-C", str(self.path), "merge-base", "--is-ancestor", older, newer], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False) + return proc.returncode == 0 + + def diff_text(self, old: str, new: str, paths: list[str]) -> str: + """The textual diff of ``paths`` (evidence for a candidate).""" + if not paths: + return "" + args = ["diff", old, new, "--", *paths] if old else ["show", "--format=", new, "--", *paths] + try: + return self._git(*args).decode("utf-8", "replace") + except SourceError: + return "" + + def parents(self, sha: str) -> list[str]: + return self._git("rev-list", "--parents", "-n", "1", sha).decode().split()[1:] + def show(self, rev: str, path: str) -> str | None: try: return self._git("show", f"{rev}:{path}").decode("utf-8", "replace") diff --git a/test/test_kb_audit.py b/test/test_kb_audit.py index 1f14eded..af05229a 100644 --- a/test/test_kb_audit.py +++ b/test/test_kb_audit.py @@ -1,4 +1,4 @@ -"""The daily post-merge audit: unrecorded knowledge changes pause auto-merge.""" +"""Provenance: unrecorded knowledge changes pause auto-merge, block activation and get reverted.""" from __future__ import annotations @@ -6,7 +6,7 @@ import time from dataclasses import replace -from infermatrix_copilot.kb_service.audit import AUDIT_CURSOR, GRACE, audit_main +from infermatrix_copilot.kb_service.audit import AUDIT_CURSOR, GRACE, audit_main # noqa: F401 from infermatrix_copilot.kb_service.scheduler import Scheduler from test_kb_flow import _flow_runtime from test_kb_intake_gate import PAGE @@ -28,6 +28,7 @@ def _land(origin, rel: str, text: str, message: str) -> str: 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 + rt.lease_owner = rt.ledger.acquire_lease("scheduler") 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() @@ -42,7 +43,7 @@ def test_a_knowledge_change_without_a_record_pauses_auto_merge(tmp_path): (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 any("provenance" 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 @@ -54,7 +55,7 @@ def test_recorded_merges_and_code_changes_pass(tmp_path): 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", + rt.ledger.stage_intake(rt.lease_owner, "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) == [] @@ -62,14 +63,21 @@ def test_recorded_merges_and_code_changes_pass(tmp_path): assert rt.ledger.get_cursor("*", AUDIT_CURSOR) == merged -def test_recent_commits_wait_for_the_next_run(tmp_path): +def test_our_own_merge_waiting_for_its_receipt_is_not_unknown(tmp_path): rt, lifecycle, scheduler, origin = _setup(tmp_path) before = rt.ledger.get_cursor("*", AUDIT_CURSOR) + _git(origin, "checkout", "-q", "-b", "kb/demo/r") 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 + 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", "merged by us", head) + changeset = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.lease_owner, "demo", changeset, detail={}, status="merge_requested", + verdicts=[], human_reason="", drafted_events=[]) + rt.ledger.update_changeset(changeset, pr_number=46, head_sha=head) + assert audit_main(rt, scheduler._pause) == [] # the receipt is on its way + assert not rt.ledger.repo_state("demo")["paused"] + assert rt.ledger.get_cursor("*", AUDIT_CURSOR) == before # looked at again next tick def test_repositories_not_in_auto_merge_are_only_traced(tmp_path): @@ -99,7 +107,7 @@ def test_a_commit_message_citing_a_recorded_pr_is_no_proof(tmp_path): 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", + rt.ledger.stage_intake(rt.lease_owner, "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", @@ -116,7 +124,7 @@ def test_a_queue_merge_of_the_recorded_head_is_proof(tmp_path): _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", + rt.ledger.stage_intake(rt.lease_owner, "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) == [] @@ -133,7 +141,7 @@ def test_a_merge_of_a_recorded_head_with_extra_edits_is_caught(tmp_path): _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", + rt.ledger.stage_intake(rt.lease_owner, "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) @@ -150,7 +158,7 @@ def test_a_merge_that_adds_an_executable_bit_is_caught(tmp_path): _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", + rt.ledger.stage_intake(rt.lease_owner, "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 diff --git a/test/test_kb_flow.py b/test/test_kb_flow.py index c50a6d21..1dfdc6d5 100644 --- a/test/test_kb_flow.py +++ b/test/test_kb_flow.py @@ -247,10 +247,10 @@ def test_activation_switches_atomically_and_rolls_back(tmp_path, monkeypatch): rt, lifecycle = _flow_runtime(tmp_path) _add_agents(rt) first = rt.knowledge.fetch() - activate(rt, first) + activate(rt, first, verify_provenance=False) assert Path(active_link(rt.state_dir)).resolve().name == first second = _commit_change(rt, "repos/demo/guide.md", "# guide\n") - activate(rt, second) + activate(rt, second, verify_provenance=False) monkeypatch.setenv("KNOWLEDGE_ROOT", str(active_link(rt.state_dir))) from infermatrix_copilot.knowledge_view import _load_view _load_view.cache_clear() @@ -265,11 +265,11 @@ def test_activation_refuses_a_broken_tree(tmp_path): rt, lifecycle = _flow_runtime(tmp_path) _add_agents(rt) good = rt.knowledge.fetch() - activate(rt, good) + activate(rt, good, verify_provenance=False) bad = _commit_change(rt, "repos/demo/_routes.yaml", "schema_version: 1\nowners:\n - owner: core\n path: repos/demo/missing.md\n") with pytest.raises(ActivationError): - activate(rt, bad) + activate(rt, bad, verify_provenance=False) assert Path(active_link(rt.state_dir)).resolve().name == good # unchanged @@ -484,9 +484,9 @@ def test_scheduler_does_not_undo_a_rollback_until_main_moves(tmp_path): rt, lifecycle = _flow_runtime(tmp_path) _add_agents(rt) first = rt.knowledge.fetch() - activate(rt, first) + activate(rt, first, verify_provenance=False) second = _commit_change(rt, "repos/demo/guide.md", "# guide\n") - activate(rt, second) + activate(rt, second, verify_provenance=False) rollback(rt, first) scheduler = Scheduler(rt) scheduler.intake_every = scheduler.release_every = 10 ** 9 @@ -494,6 +494,12 @@ def test_scheduler_does_not_undo_a_rollback_until_main_moves(tmp_path): scheduler.tick() assert Path(active_link(rt.state_dir)).resolve().name == first # rollback respected third = _commit_change(rt, "repos/demo/guide.md", "# guide v3\n") + owner = rt.ledger.acquire_lease("t") + for sha in (second, third): # both passed the gate + merged = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(owner, "demo", merged, detail={}, status="merged", + verdicts=[], human_reason="", drafted_events=[]) + rt.ledger.update_changeset(merged, merge_sha=sha) scheduler.tick() assert Path(active_link(rt.state_dir)).resolve().name == third # a newer main activates @@ -502,7 +508,7 @@ def test_activation_rejects_a_snapshot_whose_content_was_altered(tmp_path): rt, lifecycle = _flow_runtime(tmp_path) _add_agents(rt) sha = rt.knowledge.fetch() - snapshot = activate(rt, sha) + snapshot = activate(rt, sha, verify_provenance=False) (snapshot / "knowledge" / "AGENTS.md").write_text("# tampered\n", encoding="utf-8") with pytest.raises(ActivationError, match="does not match its manifest"): verify_snapshot(snapshot) @@ -610,9 +616,9 @@ def test_rollback_and_a_concurrent_tick_are_serialised(tmp_path, monkeypatch): rt, lifecycle = _flow_runtime(tmp_path) _add_agents(rt) first = rt.knowledge.fetch() - activate(rt, first) + activate(rt, first, verify_provenance=False) second = _commit_change(rt, "repos/demo/guide.md", "# guide\n") - activate(rt, second) + activate(rt, second, verify_provenance=False) original = activate_module.verify_snapshot entered = threading.Event() @@ -850,17 +856,17 @@ def test_activation_refuses_a_knowledge_format_this_copilot_does_not_read(tmp_pa rt, lifecycle = _flow_runtime(tmp_path) _add_agents(rt) good = rt.knowledge.fetch() - snapshot = activate(rt, good) + snapshot = activate(rt, good, verify_provenance=False) assert json.loads((snapshot / "MANIFEST.json").read_text())["knowledge_format"] == 2 newer = _commit_change(rt, "_format.yaml", "format_version: 3\n") with pytest.raises(ActivationError, match="knowledge format 3"): - activate(rt, newer) + activate(rt, newer, verify_provenance=False) origin = Path(rt.knowledge.path).parent / "origin" (origin / "knowledge" / "_format.yaml").unlink() _git(origin, "add", "-A") _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "undeclared") with pytest.raises(ActivationError, match="undeclared"): - activate(rt, rt.knowledge.fetch()) + activate(rt, rt.knowledge.fetch(), verify_provenance=False) assert Path(active_link(rt.state_dir)).resolve().name == good # still the good one @@ -875,8 +881,8 @@ def test_a_malformed_format_declaration_is_refused_not_coerced(tmp_path, declare rt, lifecycle = _flow_runtime(tmp_path) _add_agents(rt) good = rt.knowledge.fetch() - activate(rt, good) + activate(rt, good, verify_provenance=False) bad = _commit_change(rt, "_format.yaml", f"format_version: {declared}\n") with pytest.raises(ActivationError, match="undeclared"): - activate(rt, bad) + activate(rt, bad, verify_provenance=False) assert Path(active_link(rt.state_dir)).resolve().name == good diff --git a/test/test_kb_provenance.py b/test/test_kb_provenance.py new file mode 100644 index 00000000..789e91db --- /dev/null +++ b/test/test_kb_provenance.py @@ -0,0 +1,279 @@ +"""Unknown knowledge changes: revert PRs, exact disposal, blocked activation, the publisher's check.""" + +from __future__ import annotations + +import json + +from infermatrix_copilot.kb_service import merge +from infermatrix_copilot.kb_service.audit import ( + DISPOSED, TRUSTED, UNKNOWN, audit_main, open_unknown, +) +from test_kb_audit import _git, _land, _setup +from test_kb_flow import _items +from test_kb_intake_gate import PAGE + + +def _sneak(origin, text="\n\n"): + page = origin / "knowledge" / PAGE + before = page.read_text() + sha = _land(origin, f"knowledge/{PAGE}", before + text, "hotfix knowledge directly") + return sha, before + + +def _open_revert(rt, origin, sha, before, *, extra=None): + """What the publisher does for open_revert_pr, then its ack.""" + (item,) = _items(origin.parent, "open_revert_pr") + body = item["body"] + assert body["files"] == {f"knowledge/{PAGE}": before} and body["deleted"] == [] + _git(origin, "checkout", "-q", "-b", body["branch"]) + head = _land(origin, f"knowledge/{PAGE}", before, "revert") + if extra: + head = _land(origin, extra[0], extra[1], "and something else") + _git(origin, "checkout", "-q", "main") + merge.apply_acks(rt, [{"item_id": item["id"], "kind": "open_revert_pr", "changeset_id": body["changeset_id"], + "ok": True, "pr": 90, "head_sha": head, "branch": body["branch"]}]) + return body["changeset_id"], head + + +def _merge_revert(rt, origin, head, *, chmod=False): + """People merge the revert PR on GitHub (the service's clone has not seen it yet).""" + _git(origin, "-c", "user.name=h", "-c", "user.email=h@e", "merge", "-q", "--no-ff", "--no-commit", head) + if chmod: + _git(origin, "update-index", "--chmod=+x", f"knowledge/{PAGE}") + _git(origin, "-c", "user.name=h", "-c", "user.email=h@e", "commit", "-q", "-m", "merge revert") + merged = _git(origin, "rev-parse", "HEAD") + rt.github.prs[90] = {"state": "closed", "merged": True, "merge_commit_sha": merged, "head": {"sha": head}} + return merged + + +def test_an_unknown_change_is_paused_blocked_reverted_and_sent_back_to_intake(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + sha, before = _sneak(origin) + (finding,) = audit_main(rt, scheduler._pause) + assert sha[:12] in finding and rt.ledger.repo_state("demo")["paused"] + assert open_unknown(rt) == [sha] + record = json.loads(rt.ledger.get_cursor("*", UNKNOWN + sha)) + assert record["state"] == "reverting" + (revert,) = rt.ledger.changesets_of_kind("demo", "revert") + assert revert["status"] == "pr_requested" and revert["detail"]["reverts"] == sha + (item,) = _items(tmp_path, "open_revert_pr") + assert item["body"]["files"] == {f"knowledge/{PAGE}": before} + (event,) = rt.ledger.events("demo", "pending") + assert event["payload"]["source_reference"] == f"commit {sha}" + assert "+" in event["payload"]["diff_excerpt"] # the content, not a name + assert audit_main(rt, scheduler._pause) == [] # handled once + + +def test_an_exact_revert_disposes_of_the_change_and_both_become_trusted(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + sha, before = _sneak(origin) + audit_main(rt, scheduler._pause) + changeset_id, head = _open_revert(rt, origin, sha, before) + assert rt.ledger.changeset(changeset_id)["status"] == "revert_open" + merged = _merge_revert(rt, origin, head) + assert f"revert_merged {changeset_id} exact=True" in merge.advance(rt, lifecycle) + assert rt.ledger.get_cursor("*", DISPOSED + sha) == merged + assert open_unknown(rt) == [] + assert audit_main(rt, scheduler._pause) == [] # the revert merge is trusted too + assert {sha, merged} <= set(json.loads(rt.ledger.get_cursor("*", TRUSTED))) + + +def test_a_revert_merged_with_other_changes_disposes_of_nothing(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + sha, before = _sneak(origin) + audit_main(rt, scheduler._pause) + changeset_id, head = _open_revert(rt, origin, sha, before, + extra=("knowledge/repos/demo/core/_index.md", "# core\n\nsmuggled\n")) + merged = _merge_revert(rt, origin, head) + assert f"revert_merged {changeset_id} exact=False" in merge.advance(rt, lifecycle) + assert open_unknown(rt) == [sha] + assert any("not the exact revert" in row["reason"] for row in rt.ledger.human_queue("demo")) + (finding,) = audit_main(rt, scheduler._pause) # and the inexact merge is itself unknown + assert merged[:12] in finding + + +def test_a_revert_that_would_conflict_goes_to_people(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + page = origin / "knowledge" / PAGE + sha, _before = _sneak(origin) + _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", "a second direct push") + audit_main(rt, scheduler._pause) + record = json.loads(rt.ledger.get_cursor("*", UNKNOWN + sha)) + assert record["state"] == "conflict" + assert all(sha not in item["body"]["title"] for item in _items(tmp_path, "open_revert_pr")) + assert any("would conflict" in row["reason"] for row in rt.ledger.human_queue("demo")) + + +def test_nothing_is_activated_while_an_unknown_change_is_open(tmp_path): + from infermatrix_copilot.kb_service.activate import activate + from test_kb_flow import _add_agents + + rt, lifecycle, scheduler, origin = _setup(tmp_path) + from infermatrix_copilot.kb_service.audit import AUDIT_CURSOR + + _add_agents(rt) + activate(rt, rt.knowledge.fetch()) + rt.ledger.set_cursor("*", AUDIT_CURSOR, rt.knowledge.main_sha()) # the fixture's own setup commit + active = rt.ledger.active_snapshot() + sha, before = _sneak(origin) + scheduler._last = {f"intake:{lifecycle.repo}": rt.clock(), f"release:{lifecycle.repo}": rt.clock(), + "external": rt.clock(), "archive": rt.clock()} + scheduler.tick() + assert rt.ledger.active_snapshot() == active + assert any(e["event"] == "activation_blocked" for e in scheduler.log) + changeset_id, head = _open_revert(rt, origin, sha, before) + _merge_revert(rt, origin, head) + rt.ledger.resume("demo") + scheduler.tick() + assert rt.ledger.active_snapshot() == rt.knowledge.main_sha() != active + + +def test_the_publisher_refuses_to_merge_on_top_of_an_unknown_change(tmp_path): + from infermatrix_copilot.kb_service.activate import activate + from test_kb_flow import _add_agents + from test_kb_publisher import _collect, _merge_ready + + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + _add_agents(rt) + activate(rt, rt.knowledge.fetch()) + rt.outbox.refresh_control() # carries the active snapshot + trusted list + origin = tmp_path / "origin" + _git(origin, "checkout", "-q", "main") + sneaky, _ = _sneak(origin) + summary = pub.run_once() + assert summary["failed"] == 1 and gh.prs[42]["state"] == "OPEN" + _collect(rt) + assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" # transient: the service pauses next + + +def test_a_revert_merged_with_a_mode_change_is_not_exact(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + sha, before = _sneak(origin) + audit_main(rt, scheduler._pause) + changeset_id, head = _open_revert(rt, origin, sha, before) + _merge_revert(rt, origin, head, chmod=True) + assert f"revert_merged {changeset_id} exact=False" in merge.advance(rt, lifecycle) + assert open_unknown(rt) == [sha] + + +def test_a_revert_made_by_hand_disposes_of_what_it_undoes(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + page = origin / "knowledge" / PAGE + first, before = _sneak(origin) + second = _land(origin, f"knowledge/{PAGE}", page.read_text() + "\n\n", "second push") + audit_main(rt, scheduler._pause) + assert open_unknown(rt) == sorted([first, second]) + manual = _land(origin, f"knowledge/{PAGE}", before, "revert both by hand") + assert audit_main(rt, scheduler._pause) == [] # recognised as their exact revert + assert open_unknown(rt) == [] + assert rt.ledger.get_cursor("*", DISPOSED + first) == manual == rt.ledger.get_cursor("*", DISPOSED + second) + + +def test_activation_itself_refuses_a_commit_that_passed_no_gate(tmp_path): + import pytest + + from infermatrix_copilot.kb_service.activate import ActivationError, activate + from test_kb_flow import _add_agents + + rt, lifecycle, scheduler, origin = _setup(tmp_path) + _add_agents(rt) + activate(rt, rt.knowledge.fetch()) + _sneak(origin) # not even audited yet + with pytest.raises(ActivationError, match="provenance"): + activate(rt, rt.knowledge.fetch()) # what `kb activate` runs + + +def test_a_revert_must_restore_every_knowledge_path_not_just_governed_ones(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + page = origin / "knowledge" / PAGE + before = page.read_text() + page.write_text(before + "\n\n") + (origin / "knowledge" / "AGENTS.md").write_text("# agents, rewritten around the gate\n") + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "page + AGENTS") + both = _git(origin, "rev-parse", "HEAD") + audit_main(rt, scheduler._pause) + assert _items(tmp_path, "open_revert_pr") == [] # not all governed: people revert it + _land(origin, f"knowledge/{PAGE}", before, "revert only the page") + audit_main(rt, scheduler._pause) + assert both in open_unknown(rt) # AGENTS.md is still unreviewed + (origin / "knowledge" / "AGENTS.md").unlink() + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "revert AGENTS too") + audit_main(rt, scheduler._pause) + assert both not in open_unknown(rt) # the second revert completed it + + +def test_a_revert_by_hand_of_a_non_governed_change_clears_the_block(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + agents = _land(origin, "knowledge/AGENTS.md", "# agents around the gate\n", "AGENTS only") + audit_main(rt, scheduler._pause) + assert open_unknown(rt) == [agents] + (origin / "knowledge" / "AGENTS.md").unlink() + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "revert it") + assert audit_main(rt, scheduler._pause) == [] + assert open_unknown(rt) == [] + + +def test_a_later_edit_to_any_path_of_a_large_change_makes_the_revert_conflict(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + core = origin / "knowledge" / "repos" / "demo" / "core" + for n in range(55): + (core / f"p{n:02d}.md").write_text(f"# p{n}\n") + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "55 pages around the gate") + big = _git(origin, "rev-parse", "HEAD") + later = _land(origin, "knowledge/repos/demo/core/p53.md", "# p53, edited through the gate\n", "gated edit") + merged = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.lease_owner, "demo", merged, detail={}, status="merged", + verdicts=[], human_reason="", drafted_events=[]) + rt.ledger.update_changeset(merged, merge_sha=later) + audit_main(rt, scheduler._pause) + record = json.loads(rt.ledger.get_cursor("*", UNKNOWN + big)) + assert len(record["paths"]) == 55 + assert record["state"] == "conflict" and "knowledge/repos/demo/core/p53.md" in record["moved"] + assert _items(tmp_path, "open_revert_pr") == [] + + +def test_rewinding_main_past_a_revert_brings_the_unknown_change_back(tmp_path): + import pytest + + from infermatrix_copilot.kb_service.activate import ActivationError, activate + from infermatrix_copilot.kb_service.audit import AUDIT_CURSOR + from test_kb_flow import _add_agents + + rt, lifecycle, scheduler, origin = _setup(tmp_path) + _add_agents(rt) + activate(rt, rt.knowledge.fetch()) + rt.ledger.set_cursor("*", AUDIT_CURSOR, rt.knowledge.main_sha()) + sha, before = _sneak(origin) + audit_main(rt, scheduler._pause) + _land(origin, f"knowledge/{PAGE}", before, "revert by hand") + audit_main(rt, scheduler._pause) + assert open_unknown(rt) == [] # disposed of by the revert + _git(origin, "reset", "-q", "--hard", sha) # main rewound past the revert + with pytest.raises(ActivationError, match="provenance"): + activate(rt, rt.knowledge.fetch()) + + +def test_a_change_touching_a_private_scope_is_never_reverted_in_public(tmp_path): + from dataclasses import replace + + rt, lifecycle, scheduler, origin = _setup(tmp_path) + rt.registry["secret"] = replace(lifecycle, repo="secret", full_name="org/secret", knowledge_dir="repos/secret", + upstream_visibility="private", mode="shadow") + rt.ledger.ensure_repo("secret", "shadow") + page = origin / "knowledge" / PAGE + page.write_text(page.read_text() + "\n\n") + private = origin / "knowledge" / "repos" / "secret" / "notes.md" + private.parent.mkdir(parents=True, exist_ok=True) + private.write_text("# private upstream notes\n") + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "mixed scopes") + mixed = _git(origin, "rev-parse", "HEAD") + audit_main(rt, scheduler._pause) + assert _items(tmp_path, "open_revert_pr") == [] # nothing public about the private scope + assert rt.ledger.events("demo", "pending") == [] + assert json.loads(rt.ledger.get_cursor("*", UNKNOWN + mixed))["state"] == "people" + assert rt.ledger.repo_state("demo")["paused"] # still stopped until people revert it