From 316846426a0b95df09da002ddc42c0e5395dee1a Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 6 Oct 2026 15:55:33 +0800 Subject: [PATCH 1/2] Add verified supervised reconciliation for owner-merged knowledge --- doc/architecture/SPEC/kb_service.md | 20 ++ doc/architecture/SPEC/knowledge_service.md | 6 +- doc/features/knowledge-service.md | 24 ++ src/infermatrix_copilot/kb_service/audit.py | 16 +- src/infermatrix_copilot/kb_service/cli.py | 47 ++- src/infermatrix_copilot/kb_service/outbox.py | 6 + .../kb_service/reconcile.py | 278 ++++++++++++++++++ .../knowledge_service/signing.py | 3 +- test/test_kb_reconcile_reviewed.py | 200 +++++++++++++ 9 files changed, 591 insertions(+), 9 deletions(-) create mode 100644 src/infermatrix_copilot/kb_service/reconcile.py create mode 100644 test/test_kb_reconcile_reviewed.py diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 39319024..eb843ef5 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -25,6 +25,8 @@ 只有 auto_merge 仓库才有非停止类写动作，shadow 与 disabled 一律不发布； `close`/`open_revert_pr` 总可执行）。 - `cli`：`infermatrix-copilot kb keygen|status|pause|resume|control`。 +- `reconcile`：负责人已合并的初始化/管理知识历史的显式接收；固定目标与 active、GitHub PR/merger/当前检查、 + 完整 first-parent 提交清单、目标树的两个校验器及真实快照验证，签名计划经重新验证后才安装不可变回执。 ## 不变量 - 模块不出现任何仓库名；仓库差异只来自 adapter 配置。 @@ -296,6 +298,24 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝 - `kb status` 新增 `unknown_commits`：未处置的未知提交及其接受请求的状态。 - 测试：`test_kb_accept_unknown.py`。 +## 2026-10-06 负责人已合并历史的监督接收 + +- `kb reconcile-reviewed --plan FILE --target SHA --allow-merger LOGIN --require-check NAME --reason TEXT` + 核验当前 main 的完整 SHA 与 active 的连续 first-parent 历史；所有知识提交都绑定唯一真实已合并 PR、 + merge SHA、父提交、完整原始 diff 哈希、允许的 merger、认证操作员和实际 review 状态。 + 指定检查须在精确 PR head 上当前成功；同一 app/context 的较早失败重试不覆盖最新结果。 +- 在目标树的独立临时导出中执行两个必须存在且成功的知识校验器；导出不能遗漏或改变 git 树的任何知识 blob， + 被服务的 Markdown/YAML 必须为普通 UTF-8 文件。构建并加载临时快照，验证格式、所有文件哈希、树和路由。 +- 计划按 `kb-reconciliation-plan` 签名但不授予信任。`--apply FILE` 持有激活锁并重新核验所有证据， + main/active 或证据变化即拒绝；仅两个 pin 均未改变时重复 apply 复用同一回执。 +- 回执按不同用途 `kb-reviewed-reconciliation` 签名，包含操作员/理由/检查/校验器及快照哈希， + 在 `reconciliations/<内容哈希>.json` 通过 fsync 和硬链接原子创建，禁止覆盖和符号链接。 + 审计和激活每次重新验证签名、仓库及精确提交清单；控制记录直接读取回执，不在 `trusted_merges` 缓存监督信任。 + 缺失回执使提交恢复未知，损坏回执拒绝激活/控制续签，后续新增知识提交仍保持未知。 +- 保持 `accept-unknown` 和自动质量门不变；不伪造 `disposed:`，不更改暂停或发布模式。 + 有待异步接受、撤回或原始 intake 候选的提交拒绝进入此路径。`kb status` 显示已核验的 `supervised_commits`。 +- 测试：`test_kb_reconcile_reviewed.py`。 + ## 2026-09-29 一致性冲突只算本次改动的 一致性评审看整个 owner 目录，但只有**涉及本次改动的规则**（新增、编辑、退役、被替代的规则 ID）的冲突才使变更失败； 两条都未被改动的规则之间的冲突已经在 main 上，记入该目录的 `preexisting`，不计入变更（精炼也无法修复它，否则该目录 diff --git a/doc/architecture/SPEC/knowledge_service.md b/doc/architecture/SPEC/knowledge_service.md index c4f6000b..d3ac4c8f 100644 --- a/doc/architecture/SPEC/knowledge_service.md +++ b/doc/architecture/SPEC/knowledge_service.md @@ -40,7 +40,11 @@ reported as reconciliation. `signing` provides Ed25519 envelopes whose signature covers a purpose tag plus the canonical JSON payload, so a signature made for one purpose (gate verdict, -outbox item, control record, publisher ack) never verifies for another. It +outbox item, control record, publisher ack, reconciliation plan, supervised +reconciliation receipt) never verifies for another. Plans use +`kb-reconciliation-plan`; only revalidated receipts use the distinct +`kb-reviewed-reconciliation` purpose, and neither can be replayed as an +automatic gate verdict or publisher action. It needs only `cryptography`. ## 2026-09-28 verdict diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md index 724fcc31..745c5ed4 100644 --- a/doc/features/knowledge-service.md +++ b/doc/features/knowledge-service.md @@ -90,6 +90,30 @@ infermatrix-copilot kb rollback --to # 回指到较早的快照（下一 合并精确的撤回 PR 并 `kb resume` 后恢复，被撤回的内容会作为新候选重新过门禁；也可以用 `kb accept-unknown ` 让该提交本身完整通过门禁后保留它（通过才算处置，仍需 `kb resume`）。发版巡检在上游新 release 后运行 T1/T2/T3 与 purge，每个规则页一个变更集，全部经过质量门。 +初始化、增加 owner 目录、修改顶层入口或校验器时，负责人合并的 PR 可能包含普通自动规则门禁不接受的路径。 +`kb reconcile-reviewed` 是这种已合并历史的显式操作员接收流程：以服务密钥签名，先生成可检查的计划，再重新核验后写入不可变回执。 +要求当前 main 的完整 SHA、当前 active 的连续 first-parent 历史、每个知识提交对应的唯一真实已合并 PR、 +允许的负责人 merger 与认证操作员，以及精确 PR head 上指定的当前成功检查。保留实际 review 状态，不虚构 APPROVED。 +在固定目标树上必须通过两个知识校验器与真实快照加载、格式、哈希和路由检查；缺失、失败或进行中的检查均拒绝。 + +```bash +infermatrix-copilot kb reconcile-reviewed --plan reconciliation.json \ + --target --allow-merger --require-check suite \ + --reason 'Owner-authorized recovery of merged initialization and administration' +# 检查计划中的提交、PR、操作员、检查和校验哈希后，再明确执行： +infermatrix-copilot kb reconcile-reviewed --apply reconciliation.json +infermatrix-copilot kb status +infermatrix-copilot kb resume --all +infermatrix-copilot kb activate +``` + +需提供 `KB_GITHUB_READ_TOKEN` 以核验认证操作员（沿用负责人的现有只读 API 凭据）。计划本身不产生信任； +apply 重新核验所有证据，main 或 active 改变、证据变化时要求新计划。两个 pin 不变时重复 apply 返回同一回执。 +回执在 `$KB_STATE_DIR/reconciliations/`，应随恢复备份保留；审计、激活和控制记录每次直接校验其签名与完整提交范围， +不缓存此类信任。新知识提交仍会阻塞激活；删除或破坏回执不能续签旧信任。 +接收不写 `disposed:`，不解除暂停，不更改 shadow/auto_merge，也不替代自动候选的完整质量门。 +已有异步接受、撤回或待 intake 候选的提交必须先处理完，不能由此流程绕过。 + 评审服务读取知识：将 `KNOWLEDGE_ROOT` 指向 `$KB_STATE_DIR/active`。 diff --git a/src/infermatrix_copilot/kb_service/audit.py b/src/infermatrix_copilot/kb_service/audit.py index 9f173900..c35081bc 100644 --- a/src/infermatrix_copilot/kb_service/audit.py +++ b/src/infermatrix_copilot/kb_service/audit.py @@ -85,6 +85,9 @@ def disposals(rt) -> dict[str, str]: def _records(rt) -> dict: + from .reconcile import trusted_commits + + reviewed = trusted_commits(rt) merge_shas, merged_heads, pending = set(), set(), set() for lifecycle in rt.registry.values(): for changeset in rt.ledger.changesets(lifecycle.repo, _RECORDED): @@ -98,7 +101,7 @@ def _records(rt) -> dict: if changeset.get("head_sha"): pending.add(changeset["head_sha"]) disposed = disposals(rt) - return {"merges": merge_shas, "heads": merged_heads, "pending": pending, + return {"merges": merge_shas | reviewed, "reviewed": reviewed, "heads": merged_heads, "pending": pending, "disposed": disposed, "disposing": set(disposed.values())} @@ -126,8 +129,10 @@ def _knowledge_paths(rt, sha: str, parents: list[str]) -> list[str]: def open_unknown(rt) -> list[str]: """Unknown commits not disposed of yet.""" disposed = disposals(rt) + from .reconcile import trusted_commits + reviewed = trusted_commits(rt) return sorted(name[len(UNKNOWN):] for name in rt.ledger.cursors_with_prefix("*", UNKNOWN) - if name[len(UNKNOWN):] not in disposed) + if name[len(UNKNOWN):] not in disposed and name[len(UNKNOWN):] not in reviewed) def provenance_problems(rt, sha: str) -> list[str]: @@ -151,7 +156,9 @@ def provenance_problems(rt, sha: str) -> list[str]: 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:] + # Supervised trust is never cached: control publication verifies its + # durable receipts directly, so removal/corruption cannot renew stale trust. + trusted = sorted((recs["merges"] - recs["reviewed"]) | set(recs["disposed"]) | recs["disposing"])[-TRUSTED_KEEP:] rt.ledger.set_cursor("*", TRUSTED, json.dumps(trusted)) @@ -264,12 +271,13 @@ def _issue_pending_reverts(rt, main: str) -> None: if not getattr(rt, "lease_owner", None): return disposed = disposals(rt) + remaining = set(open_unknown(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: + if record.get("state") != "revert_pending" or lifecycle is None or sha in disposed or sha not in remaining: continue record.update(_issue_revert(rt, lifecycle, sha, record, main)) rt.ledger.set_cursor("*", name, json.dumps(record)) diff --git a/src/infermatrix_copilot/kb_service/cli.py b/src/infermatrix_copilot/kb_service/cli.py index 9b2cf0e7..784befba 100644 --- a/src/infermatrix_copilot/kb_service/cli.py +++ b/src/infermatrix_copilot/kb_service/cli.py @@ -38,6 +38,7 @@ import json import os import re +import subprocess import sys import time from pathlib import Path @@ -90,16 +91,17 @@ def _outbox(state_dir: Path, ledger): return Outbox(state_dir, _signing_key(), ledger, clock=time.time) -def _unknown_status(ledger) -> dict: +def _unknown_status(ledger, reviewed: set[str] | None = None) -> dict: """Unknown knowledge commits not disposed of, with any accept request.""" from .accept import ACCEPT from .audit import DISPOSED, UNKNOWN disposed = {n[len(DISPOSED):] for n in ledger.cursors_with_prefix("*", DISPOSED)} + reviewed = reviewed or set() out = {} for name, raw in ledger.cursors_with_prefix("*", UNKNOWN).items(): sha = name[len(UNKNOWN):] - if sha in disposed: + if sha in disposed or sha in reviewed: continue record = json.loads(raw) request = ledger.get_cursor("*", ACCEPT + sha) @@ -214,6 +216,14 @@ def main(argv: list[str] | None = None) -> int: sub.add_parser("control") accept = sub.add_parser("accept-unknown") accept.add_argument("sha") + reconcile = sub.add_parser("reconcile-reviewed", help="verify and admit an exact owner-merged knowledge history") + action = reconcile.add_mutually_exclusive_group(required=True) + action.add_argument("--plan", type=Path, help="write a signed plan; does not admit any commit") + action.add_argument("--apply", type=Path, help="revalidate and apply a previously inspected signed plan") + reconcile.add_argument("--target", help="exact current main SHA, required when preparing a plan") + reconcile.add_argument("--allow-merger", action="append", default=[]) + reconcile.add_argument("--require-check", action="append", default=[]) + reconcile.add_argument("--reason", default="") run = sub.add_parser("run") run.add_argument("--playbook", required=True) run.add_argument("--repo", required=True) @@ -333,6 +343,30 @@ def main(argv: list[str] | None = None) -> int: if args.command in ("init", "widen", "deepen"): # before any ledger is opened: kb init never touches the service's kb.db return _init_command(args, _state_dir(args.state_dir)) + if args.command == "reconcile-reviewed": + from ..config import Settings + from ..knowledge_service.signing import canonical_json + from .runtime import KbRuntime + from .reconcile import make_plan, apply_plan + + rt = KbRuntime.from_env(Settings(), state_dir=_state_dir(args.state_dir)) + try: + key = _signing_key() + if args.plan: + envelope = make_plan(rt, key, target=args.target or "", allow_mergers=args.allow_merger, + required_checks=args.require_check, reason=args.reason) + with args.plan.open("xb") as handle: + handle.write(canonical_json(envelope)) + print(f"verified reconciliation plan: {args.plan}; no commit admitted") + else: + receipt = apply_plan(rt, json.loads(args.apply.read_text()), key) + print(f"supervised reconciliation receipt: {receipt}; pause and publication modes unchanged") + return 0 + except (ValueError, OSError, KeyError, RuntimeError, subprocess.SubprocessError) as exc: + print(str(exc), file=sys.stderr) + return 1 + finally: + rt.ledger.close() state_dir = _state_dir(args.state_dir) ledger = _ledger(state_dir) @@ -341,6 +375,12 @@ def main(argv: list[str] | None = None) -> int: if _sync_repos(ledger, registry) and os.environ.get("KB_SIGNING_KEY"): _refresh(state_dir, ledger, registry) if args.command == "status": + from .reconcile import read_receipts + from .sources import KnowledgeRepo + + public = _signing_key().public_key() if (state_dir / "reconciliations").exists() else None + reviewed = read_receipts(KnowledgeRepo(Path(os.environ.get("KB_KNOWLEDGE_CLONE") or + state_dir / "knowledge-repo")), state_dir, public) report = { "state_dir": str(state_dir), "active_snapshot": ledger.active_snapshot(), @@ -351,7 +391,8 @@ def main(argv: list[str] | None = None) -> int: "human_queue": len(ledger.human_queue(row["repo"])) if row["repo"] != "*" else None} for row in ledger.all_repo_states() ], - "unknown_commits": _unknown_status(ledger), + "unknown_commits": _unknown_status(ledger, reviewed), + "supervised_commits": sorted(reviewed), } json.dump(report, sys.stdout, indent=2, sort_keys=True) print() diff --git a/src/infermatrix_copilot/kb_service/outbox.py b/src/infermatrix_copilot/kb_service/outbox.py index f0798d30..311ae946 100644 --- a/src/infermatrix_copilot/kb_service/outbox.py +++ b/src/infermatrix_copilot/kb_service/outbox.py @@ -179,6 +179,12 @@ def _control_payload(self) -> dict: # 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 "[]") + from .reconcile import read_receipts + from .sources import KnowledgeRepo + + knowledge = KnowledgeRepo(Path(os.environ.get("KB_KNOWLEDGE_CLONE") or self.root / "knowledge-repo")) + reviewed = read_receipts(knowledge, self.root, self._key.public_key()) + trusted = sorted(set(trusted) | reviewed) return {"issued_at": self._clock(), "repos": repos, "provenance": {"active": self._ledger.active_snapshot() or "", "trusted": trusted}} diff --git a/src/infermatrix_copilot/kb_service/reconcile.py b/src/infermatrix_copilot/kb_service/reconcile.py new file mode 100644 index 00000000..a7df0a03 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/reconcile.py @@ -0,0 +1,278 @@ +"""Explicit admission of owner-merged initialization/admin knowledge commits. + +Automatic candidates still use the full knowledge gate. This operator path +records independently checked GitHub merge evidence and a verified final tree; +it never calls a merge, disposes of an unknown, or changes publication modes. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import re +import subprocess +import sys +import tempfile +from pathlib import Path + +from ..knowledge_service.signing import canonical_json, sign, verify +from .activate import activation_lock, build_snapshot, verify_snapshot + +PLAN_PURPOSE = "kb-reconciliation-plan" +RECEIPT_PURPOSE = "kb-reviewed-reconciliation" + + +class ReconciliationError(ValueError): + """The evidence cannot admit the pinned knowledge history.""" + + +def _digest(value) -> str: + return hashlib.sha256(canonical_json(value)).hexdigest() + + +def _repository(knowledge) -> str: + remote = knowledge._git("remote", "get-url", knowledge.remote).decode().strip() + match = re.fullmatch(r"(?:https://github\.com/|git@github\.com:)([^/]+/[^/]+?)(?:\.git)?", remote) + if not match: + raise ReconciliationError("knowledge origin must identify a GitHub repository") + return match[1] + + +def _commits(knowledge, base: str, target: str) -> list[dict]: + if not base or not knowledge.is_ancestor(base, target): + raise ReconciliationError("target must descend from the active snapshot") + records = [] + previous = base + for sha, parents, _at, _message in knowledge.first_parent_commits(base, target): + if not parents or parents[0] != previous: + raise ReconciliationError("active snapshot is not on the target's first-parent chain") + manifest = knowledge.raw_manifest(parents[0], sha) + if any(e["path"].startswith("knowledge/") for e in manifest): + records.append({"sha": sha, "parents": parents, "manifest_sha256": _digest(manifest)}) + previous = sha + if previous != target: + raise ReconciliationError("target first-parent history is incomplete") + return records + + +def _pages(github, endpoint: str, *, field: str | None = None) -> list[dict]: + result = [] + for page in range(1, 101): + response = github.get(endpoint, per_page=100, page=page) + batch = response.get(field) if field else response + if not isinstance(batch, list): + raise ReconciliationError(f"malformed GitHub evidence: {endpoint}") + result.extend(batch) + if len(batch) < 100: + return result + raise ReconciliationError(f"incomplete GitHub evidence: {endpoint}") + + +def _merge_evidence(github, repository: str, commit: dict, allowed: set[str], required: set[str]) -> dict: + sha = commit["sha"] + root = f"/repos/{repository}" + prs = [p for p in _pages(github, f"{root}/commits/{sha}/pulls") + if p.get("merged_at") and p.get("merge_commit_sha") == sha + and (p.get("base", {}).get("repo") or {}).get("full_name", "").lower() == repository.lower()] + if len(prs) != 1: + raise ReconciliationError(f"{sha[:12]} must match exactly one merged PR") + pr = github.get(f"{root}/pulls/{prs[0]['number']}") + merger = (pr.get("merged_by") or {}).get("login", "") + head = (pr.get("head") or {}).get("sha", "") + if not pr.get("merged_at") or pr.get("merge_commit_sha") != sha or merger.lower() not in allowed: + raise ReconciliationError(f"PR #{pr.get('number')} has unmatched merge evidence or disallowed merger") + if len(commit["parents"]) == 2 and commit["parents"][1] != head: + raise ReconciliationError(f"PR #{pr['number']} head does not match the actual merge parent") + checks = _pages(github, f"{root}/commits/{head}/check-runs", field="check_runs") + statuses = _pages(github, f"{root}/commits/{head}/statuses") + observed = [{"kind": "check", "app_id": (r.get("app") or {}).get("id", 0), "name": r["name"], + "state": r.get("conclusion") if r.get("status") == "completed" else "pending", + "id": r["id"], "head_sha": r.get("head_sha")} for r in checks] + observed += [{"kind": "status", "app_id": 0, "name": r["context"], "state": r["state"], + "id": r["id"], "head_sha": head} + for r in statuses] + latest = {} + for row in sorted(observed, key=lambda r: r["id"]): + latest[row["kind"], row["app_id"], row["name"]] = row + observed = list(latest.values()) + # Combined commit status is 'pending' even when the statuses list is empty. + # Require named evidence instead; no absence or aggregate counts as success. + for name in required: + found = [r for r in observed if r["name"] == name] + if not found or any(r["state"] != "success" or r["head_sha"] != head for r in found): + raise ReconciliationError(f"PR #{pr['number']} required check {name!r} is missing or unsuccessful") + if any(r["state"] not in {"success", "neutral", "skipped"} for r in observed): + raise ReconciliationError(f"PR #{pr['number']} has failed or pending checks") + reviews = _pages(github, f"{root}/pulls/{pr['number']}/reviews") + return {"number": pr["number"], "url": pr["html_url"], "head_sha": head, + "merge_sha": sha, "merged_at": pr["merged_at"], "merged_by": merger, + "checks": sorted(observed, key=lambda r: (r["name"], r["id"])), + "reviews": [{"id": r["id"], "state": r["state"], "commit_id": r["commit_id"], + "user": (r.get("user") or {}).get("login", "")} for r in reviews]} + + +def _validate_target(knowledge, target: str) -> dict: + with tempfile.TemporaryDirectory(prefix="kb-reconcile-") as temporary: + root = Path(temporary) + source = knowledge.export(target, root / "source") + # Validators also inspect tracked local/ paths. Give the exported tree + # an exact temporary index without checking out or modifying the live + # clone; its object store is referenced read-only through alternates. + subprocess.run(["git", "init", "-q", str(source)], check=True, capture_output=True) + objects = knowledge._git("rev-parse", "--path-format=absolute", "--git-path", "objects").decode().strip() + (source / ".git" / "objects" / "info" / "alternates").write_text(objects + "\n") + subprocess.run(["git", "-C", str(source), "read-tree", target], check=True, capture_output=True) + subprocess.run(["git", "-C", str(source), "update-ref", "HEAD", target], check=True, capture_output=True) + # git archive attributes must not silently hide a knowledge file from + # the validators. Served text must be regular UTF-8, never a symlink. + listing = knowledge._git("ls-tree", "-r", "-z", target, "--", "knowledge/") + for entry in listing.split(b"\0"): + if not entry: + continue + metadata, raw_path = entry.split(b"\t", 1) + mode, kind, oid = metadata.decode().split() + path = raw_path.decode("utf-8") + exported = source / path + if kind != "blob" or not exported.is_file() or exported.is_symlink(): + raise ReconciliationError(f"knowledge file was omitted or is not regular: {path}") + actual = knowledge._git("hash-object", "--no-filters", str(exported)).decode().strip() + if actual != oid: + raise ReconciliationError(f"exported knowledge differs from the exact git tree: {path}") + if path.endswith((".md", ".yaml")): + if mode != "100644": + raise ReconciliationError(f"served knowledge has an unsupported mode: {path}") + exported.read_text(encoding="utf-8", errors="strict") + reports = [] + for name in ("check_knowledge_tree.py", "check_wiki_lint.py"): + tool = source / "knowledge" / "tools" / name + if not tool.is_file(): + raise ReconciliationError(f"mandatory validator missing: {name}") + run = subprocess.run([sys.executable, str(tool)], cwd=source, capture_output=True, timeout=180) + if run.returncode: + detail = (run.stdout + run.stderr).decode(errors="replace")[-2000:] + raise ReconciliationError(f"{name} failed: {detail}") + reports.append({"name": name, "code_sha256": hashlib.sha256(tool.read_bytes()).hexdigest(), + "output_sha256": hashlib.sha256(run.stdout + run.stderr).hexdigest(), "exit_code": 0}) + snapshot = build_snapshot(root / "state", target, knowledge.knowledge_files(target), + extra=knowledge.top_level_knowledge(target)) + verify_snapshot(snapshot) + return {"validators": reports, "snapshot_manifest_sha256": + _digest(json.loads((snapshot / "MANIFEST.json").read_text()))} + + +def prepare(rt, *, target: str, allow_mergers: list[str], required_checks: list[str], reason: str) -> dict: + if not re.fullmatch(r"[0-9a-f]{40}", target) or not reason.strip() or not allow_mergers or not required_checks: + raise ReconciliationError("a full target SHA, reason, allowlisted mergers and required checks are mandatory") + if rt.knowledge.fetch() != target: + raise ReconciliationError("main differs from the explicitly pinned target") + base = rt.ledger.active_snapshot() + repository = _repository(rt.knowledge) + allowed = {a.lower() for a in allow_mergers} + actor = rt.github.get("/user").get("login", "") + if actor.lower() not in allowed: + raise ReconciliationError("authenticated operator is not an allowlisted owner") + commits = _commits(rt.knowledge, base, target) + if not commits: + raise ReconciliationError("no knowledge commits need reconciliation") + for commit in commits: + unknown = rt.ledger.get_cursor("*", "unknown:" + commit["sha"]) + record = json.loads(unknown) if unknown else {} + acceptance = json.loads(rt.ledger.get_cursor("*", "accept_request:" + commit["sha"]) or "{}") + if acceptance.get("state") == "requested": + raise ReconciliationError("a pending acceptance must settle before supervised reconciliation") + if record.get("state") not in {None, "people", "open", "accepted"} or record.get("revert_changeset"): + raise ReconciliationError("a pending revert/acceptance must settle before supervised reconciliation") + for lifecycle in rt.registry.values(): + candidate = rt.ledger.event_by_external_id(lifecycle.repo, "unrecorded", commit["sha"]) + if candidate and candidate["status"] == "pending": + raise ReconciliationError("an unrecorded intake candidate must settle before reconciliation") + commit["pr"] = _merge_evidence(rt.github, repository, commit, allowed, set(required_checks)) + validation = _validate_target(rt.knowledge, target) + if rt.knowledge.fetch() != target or rt.ledger.active_snapshot() != base: + raise ReconciliationError("main or active snapshot changed during validation; prepare a new plan") + return {"schema_version": 1, "repository": repository, "active_snapshot": base, "target": target, + "target_tree": rt.knowledge._git("rev-parse", f"{target}^{{tree}}").decode().strip(), + "actor": actor, "reason": reason.strip(), "allow_mergers": sorted(allowed), + "required_checks": sorted(set(required_checks)), "commits": commits, + **validation, "prepared_at": rt.clock()} + + +def make_plan(rt, key, **options) -> dict: + return sign(PLAN_PURPOSE, prepare(rt, **options), key) + + +def apply_plan(rt, envelope: dict, key) -> Path: + planned = verify(PLAN_PURPOSE, envelope, key.public_key()) + with activation_lock(rt.state_dir): + fresh = prepare(rt, target=planned["target"], allow_mergers=planned["allow_mergers"], + required_checks=planned["required_checks"], reason=planned["reason"]) + if {k: v for k, v in fresh.items() if k != "prepared_at"} != { + k: v for k, v in planned.items() if k != "prepared_at"}: + raise ReconciliationError("reconciliation evidence changed; prepare a new plan") + directory = rt.state_dir / "reconciliations" + if directory.is_symlink(): + raise ReconciliationError("reconciliation directory must not be a symlink") + directory.mkdir(parents=True, exist_ok=True) + for existing in directory.glob("*.json"): + prior = json.loads(existing.read_text()) + payload = verify(RECEIPT_PURPOSE, prior, key.public_key()) + if existing.stem != _digest(prior): + raise ReconciliationError("existing reconciliation receipt was modified") + if payload.get("plan_sha256") == _digest(envelope): + return existing + receipt = sign(RECEIPT_PURPOSE, {**fresh, "admitted_at": rt.clock(), "plan_sha256": _digest(envelope)}, key) + path = directory / f"{_digest(receipt)}.json" + # Plans have a separate signature purpose and cannot be installed as + # receipts. A successfully applied receipt is never replaced in place. + fd, temporary = tempfile.mkstemp(prefix=".receipt-", dir=directory) + try: + with os.fdopen(fd, "wb") as handle: + handle.write(canonical_json(receipt)) + handle.flush() + os.fsync(handle.fileno()) + os.link(temporary, path) # atomic publication, refuses replacement + directory_fd = os.open(directory, os.O_RDONLY) + try: + os.fsync(directory_fd) + finally: + os.close(directory_fd) + finally: + os.unlink(temporary) + rt.trace("outcome", context={"playbook": "kb-reconcile-reviewed", "repo": fresh["repository"]}, + result={"outcome": "supervised_admission", "target": fresh["target"], "receipt": path.name, + "actor": fresh["actor"], "commits": len(fresh["commits"])}) + return path + + +def trusted_commits(rt) -> set[str]: + """Re-verify immutable receipts before any audit, activation or control trust.""" + public = rt.outbox._key.public_key() if rt.outbox is not None else None + return read_receipts(rt.knowledge, rt.state_dir, public) + + +def read_receipts(knowledge, state_dir: Path, public) -> set[str]: + directory = state_dir / "reconciliations" + if directory.is_symlink(): + raise ReconciliationError("reconciliation directory must not be a symlink") + if not directory.exists(): + return set() + if public is None: + raise ReconciliationError("supervised receipts require the configured service signing key") + trusted = set() + for path in sorted(directory.glob("*.json")): + if path.is_symlink() or not path.is_file(): + raise ReconciliationError("reconciliation receipts must be regular immutable files") + envelope = json.loads(path.read_text()) + if path.stem != _digest(envelope): + raise ReconciliationError("reconciliation receipt content differs from its immutable name") + payload = verify(RECEIPT_PURPOSE, envelope, public) + if payload.get("schema_version") != 1 or payload["repository"] != _repository(knowledge): + raise ReconciliationError("reconciliation receipt does not identify this knowledge repository") + actual = _commits(knowledge, payload["active_snapshot"], payload["target"]) + recorded = [{k: c[k] for k in ("sha", "parents", "manifest_sha256")} for c in payload["commits"]] + if recorded != actual or payload["target_tree"] != knowledge._git( + "rev-parse", f"{payload['target']}^{{tree}}").decode().strip(): + raise ReconciliationError("reconciliation receipt does not cover the exact pinned history") + trusted.update(c["sha"] for c in actual) + return trusted diff --git a/src/infermatrix_copilot/knowledge_service/signing.py b/src/infermatrix_copilot/knowledge_service/signing.py index 3f68a744..0430da7a 100644 --- a/src/infermatrix_copilot/knowledge_service/signing.py +++ b/src/infermatrix_copilot/knowledge_service/signing.py @@ -23,7 +23,8 @@ Ed25519PrivateKey, Ed25519PublicKey, ) -PURPOSES = ("kb-gate-verdict", "kb-outbox-item", "kb-control", "kb-ack") +PURPOSES = ("kb-gate-verdict", "kb-outbox-item", "kb-control", "kb-ack", + "kb-reconciliation-plan", "kb-reviewed-reconciliation") class SignatureError(ValueError): diff --git a/test/test_kb_reconcile_reviewed.py b/test/test_kb_reconcile_reviewed.py new file mode 100644 index 00000000..a53123a8 --- /dev/null +++ b/test/test_kb_reconcile_reviewed.py @@ -0,0 +1,200 @@ +"""Supervised owner-merged admission stays distinct from automatic gating.""" + +from __future__ import annotations + +import json +from types import SimpleNamespace + +import pytest + +from infermatrix_copilot.kb_service import reconcile +from infermatrix_copilot.kb_service.activate import activate +from infermatrix_copilot.kb_service.accept import request_accept, accept_pending +from infermatrix_copilot.kb_service.audit import audit_main, open_unknown, provenance_problems, publish_trusted +from infermatrix_copilot.knowledge_service.signing import SignatureError, verify +from test_kb_audit import _setup, _land, _git + + +class MergeGitHub: + def __init__(self, sha): + self.actor = "owner" + self.pr = {"number": 41, "html_url": "https://github.com/org/kb/pull/41", "merged_at": "2026-10-06T01:00:00Z", + "merge_commit_sha": sha, "merged_by": {"login": "owner"}, "head": {"sha": sha}, + "base": {"repo": {"full_name": "org/kb"}}} + self.checks = [{"id": 1, "name": "suite", "head_sha": sha, "status": "completed", "conclusion": "success"}] + + def get(self, path, **_params): + if path == "/user": + return {"login": self.actor} + if path.endswith("/pulls"): + return [self.pr] + if path.endswith("/pulls/41"): + return self.pr + if path.endswith("/check-runs"): + return {"check_runs": self.checks} + if path.endswith(("/statuses", "/reviews")): + return [] + raise AssertionError(path) + + +@pytest.fixture +def world(tmp_path, monkeypatch): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + _land(origin, "knowledge/AGENTS.md", "# Agent instructions\n", "baseline entry") + for name in ("check_knowledge_tree.py", "check_wiki_lint.py"): + _land(origin, f"knowledge/tools/{name}", "import subprocess\nsubprocess.run(['git','ls-files','--error-unmatch','knowledge/AGENTS.md'], check=True, capture_output=True)\nprint('fixture validation passed')\n", "baseline validator") + baseline = rt.knowledge.fetch() + # This fixture baseline precedes the recovery target, exactly as deployment. + rt.ledger.set_cursor("*", "audit_main_sha", baseline) + activate(rt, baseline) + sha = _land(origin, "knowledge/README.md", "# Knowledge\n\nHuman-maintained onboarding.\n", "owner administration (#41)") + audit_main(rt, scheduler._pause) + rt.github = MergeGitHub(sha) + monkeypatch.setattr(reconcile, "_repository", lambda knowledge: "org/kb") + monkeypatch.setenv("KB_KNOWLEDGE_CLONE", str(rt.knowledge.path)) + return SimpleNamespace(rt=rt, scheduler=scheduler, origin=origin, sha=sha, base=baseline, + key=rt.outbox._key, options=dict(target=sha, allow_mergers=["owner"], + required_checks=["suite"], reason="authorized owner recovery")) + + +def test_supervised_admission_passes_real_activation_without_disposing_or_changing_modes(world): + w = world + request_accept(w.rt.ledger, w.sha, at=w.rt.clock()) + assert "refused" in accept_pending(w.rt)[0] # automatic accept still refuses top-level pages + plan = reconcile.make_plan(w.rt, w.key, **w.options) + assert open_unknown(w.rt) == [w.sha] # preparing never grants trust + receipt = reconcile.apply_plan(w.rt, plan, w.key) + assert reconcile.apply_plan(w.rt, plan, w.key) == receipt + assert open_unknown(w.rt) == [] + assert w.rt.ledger.get_cursor("*", "disposed:" + w.sha) is None + assert w.rt.ledger.repo_state("*")["paused"] + assert w.rt.ledger.repo_state("demo")["mode"] == "auto_merge" + assert provenance_problems(w.rt, w.sha) == [] + assert activate(w.rt, w.sha).name == w.sha + payload = verify(reconcile.RECEIPT_PURPOSE, json.loads(receipt.read_text()), w.key.public_key()) + assert payload["actor"] == "owner" and payload["commits"][0]["pr"]["reviews"] == [] + + +@pytest.mark.parametrize("failure", ["missing_check", "pending_check", "failed_check", "wrong_head", "wrong_merge", "merger", "actor"]) +def test_unproven_github_evidence_cannot_create_receipts(world, failure): + w = world + if failure == "missing_check": + w.rt.github.checks = [] + elif failure in {"pending_check", "failed_check"}: + w.rt.github.checks[0]["conclusion"] = "failure" if failure == "failed_check" else None + w.rt.github.checks[0]["status"] = "completed" if failure == "failed_check" else "queued" + elif failure == "wrong_head": + w.rt.github.checks[0]["head_sha"] = "f" * 40 + elif failure == "wrong_merge": + w.rt.github.pr["merge_commit_sha"] = "f" * 40 + elif failure == "merger": + w.rt.github.pr["merged_by"]["login"] = "untrusted" + else: + w.rt.github.actor = "untrusted" + with pytest.raises(reconcile.ReconciliationError): + reconcile.make_plan(w.rt, w.key, **w.options) + assert open_unknown(w.rt) == [w.sha] + assert not (w.rt.state_dir / "reconciliations").exists() + + +def test_target_drift_and_plan_tampering_refuse_before_admission(world): + w = world + plan = reconcile.make_plan(w.rt, w.key, **w.options) + damaged = json.loads(json.dumps(plan)) + damaged["payload"]["reason"] = "another reason" + with pytest.raises(SignatureError): + reconcile.apply_plan(w.rt, damaged, w.key) + _land(w.origin, "src/app.py", "x=1\n", "new main") + with pytest.raises(reconcile.ReconciliationError, match="main differs"): + reconcile.apply_plan(w.rt, plan, w.key) + assert not (w.rt.state_dir / "reconciliations").exists() + + +def test_new_knowledge_merge_stays_unknown_after_receipt_and_nonknowledge_descendant_is_allowed(world): + w = world + reconcile.apply_plan(w.rt, reconcile.make_plan(w.rt, w.key, **w.options), w.key) + code = _land(w.origin, "src/app.py", "x=1\n", "code only") + assert w.rt.knowledge.fetch() == code and provenance_problems(w.rt, code) == [] + assert reconcile.trusted_commits(w.rt) == {w.sha} + newer = _land(w.origin, "knowledge/README.md", "# Changed later\n", "new knowledge") + audit_main(w.rt, w.scheduler._pause) + assert open_unknown(w.rt) == [newer] + assert any(newer[:12] in p for p in provenance_problems(w.rt, w.rt.knowledge.fetch())) + + +def test_corrupt_or_missing_receipt_never_renews_cached_control_trust(world): + w = world + receipt = reconcile.apply_plan(w.rt, reconcile.make_plan(w.rt, w.key, **w.options), w.key) + publish_trusted(w.rt) + assert w.sha not in json.loads(w.rt.ledger.get_cursor("*", "trusted_merges")) + assert w.sha in w.rt.outbox.refresh_control()["provenance"]["trusted"] + original = receipt.read_bytes() + receipt.write_text("{}") + with pytest.raises(reconcile.ReconciliationError): + open_unknown(w.rt) + with pytest.raises(reconcile.ReconciliationError): + w.rt.outbox.refresh_control() + receipt.write_bytes(original) + receipt.unlink() + assert open_unknown(w.rt) == [w.sha] + assert w.sha not in w.rt.outbox.refresh_control()["provenance"]["trusted"] + + +def test_plan_signature_cannot_be_installed_as_receipt(world): + w = world + plan = reconcile.make_plan(w.rt, w.key, **w.options) + directory = w.rt.state_dir / "reconciliations" + directory.mkdir() + (directory / (reconcile._digest(plan) + ".json")).write_text(json.dumps(plan)) + with pytest.raises(SignatureError): + reconcile.trusted_commits(w.rt) + + +def test_failing_validator_keeps_previous_active_snapshot_and_pause(world): + w = world + newer = _land(w.origin, "knowledge/tools/check_wiki_lint.py", "raise SystemExit(1)\n", "broken validator (#41)") + w.rt.github = MergeGitHub(newer) + # Keep one commit in the planned interval to isolate mandatory validation. + w.rt.ledger.record_activation(w.sha, {}) + with pytest.raises(reconcile.ReconciliationError, match="check_wiki_lint.py failed"): + reconcile.make_plan(w.rt, w.key, **{**w.options, "target": newer}) + assert w.rt.ledger.active_snapshot() == w.sha + assert w.rt.ledger.repo_state("*")["paused"] + + +def test_only_first_parent_baseline_is_eligible(world): + w = world + _git(w.origin, "checkout", "-q", "-b", "side", w.base) + side = _land(w.origin, "src/side.py", "x=1\n", "side") + _git(w.origin, "checkout", "-q", "main") + _git(w.origin, "-c", "user.name=t", "-c", "user.email=t@e", "merge", "-q", "--no-ff", side) + main = w.rt.knowledge.fetch() + assert w.rt.knowledge.is_ancestor(side, main) + with pytest.raises(reconcile.ReconciliationError, match="first-parent"): + reconcile._commits(w.rt.knowledge, side, main) + + +def test_pending_revert_requires_resolution_before_supervised_admission(world): + w = world + record = json.loads(w.rt.ledger.get_cursor("*", "unknown:" + w.sha)) + w.rt.ledger.set_cursor("*", "unknown:" + w.sha, json.dumps({**record, "state": "revert_pending"})) + with pytest.raises(reconcile.ReconciliationError, match="pending revert"): + reconcile.make_plan(w.rt, w.key, **w.options) + + +def test_only_latest_check_attempt_governs_reconciliation(world): + w = world + older = {**w.rt.github.checks[0], "id": 0, "conclusion": "failure"} + w.rt.github.checks.insert(0, older) + plan = reconcile.make_plan(w.rt, w.key, **w.options) + assert plan["payload"]["commits"][0]["pr"]["checks"][0]["id"] == 1 + w.rt.github.checks[-1]["conclusion"] = "cancelled" + with pytest.raises(reconcile.ReconciliationError, match="required check"): + reconcile.make_plan(w.rt, w.key, **w.options) + + +def test_pending_accept_request_requires_resolution(world): + w = world + request_accept(w.rt.ledger, w.sha, at=w.rt.clock()) + with pytest.raises(reconcile.ReconciliationError, match="pending acceptance"): + reconcile.make_plan(w.rt, w.key, **w.options) From 86e6ab5afd42324477dc652516ebe06e2f41a91d Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 6 Oct 2026 15:57:27 +0800 Subject: [PATCH 2/2] Keep reconciliation validation and tests lint clean --- src/infermatrix_copilot/kb_service/reconcile.py | 2 +- test/test_kb_reconcile_reviewed.py | 17 +++++++++++------ 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/src/infermatrix_copilot/kb_service/reconcile.py b/src/infermatrix_copilot/kb_service/reconcile.py index a7df0a03..7c389648 100644 --- a/src/infermatrix_copilot/kb_service/reconcile.py +++ b/src/infermatrix_copilot/kb_service/reconcile.py @@ -148,7 +148,7 @@ def _validate_target(knowledge, target: str) -> dict: tool = source / "knowledge" / "tools" / name if not tool.is_file(): raise ReconciliationError(f"mandatory validator missing: {name}") - run = subprocess.run([sys.executable, str(tool)], cwd=source, capture_output=True, timeout=180) + run = subprocess.run([sys.executable, str(tool)], cwd=source, capture_output=True, timeout=180, check=False) if run.returncode: detail = (run.stdout + run.stderr).decode(errors="replace")[-2000:] raise ReconciliationError(f"{name} failed: {detail}") diff --git a/test/test_kb_reconcile_reviewed.py b/test/test_kb_reconcile_reviewed.py index a53123a8..61b36c92 100644 --- a/test/test_kb_reconcile_reviewed.py +++ b/test/test_kb_reconcile_reviewed.py @@ -6,13 +6,18 @@ from types import SimpleNamespace import pytest +from test_kb_audit import _git, _land, _setup from infermatrix_copilot.kb_service import reconcile +from infermatrix_copilot.kb_service.accept import accept_pending, request_accept from infermatrix_copilot.kb_service.activate import activate -from infermatrix_copilot.kb_service.accept import request_accept, accept_pending -from infermatrix_copilot.kb_service.audit import audit_main, open_unknown, provenance_problems, publish_trusted +from infermatrix_copilot.kb_service.audit import ( + audit_main, + open_unknown, + provenance_problems, + publish_trusted, +) from infermatrix_copilot.knowledge_service.signing import SignatureError, verify -from test_kb_audit import _setup, _land, _git class MergeGitHub: @@ -39,7 +44,7 @@ def get(self, path, **_params): @pytest.fixture def world(tmp_path, monkeypatch): - rt, lifecycle, scheduler, origin = _setup(tmp_path) + rt, _lifecycle, scheduler, origin = _setup(tmp_path) _land(origin, "knowledge/AGENTS.md", "# Agent instructions\n", "baseline entry") for name in ("check_knowledge_tree.py", "check_wiki_lint.py"): _land(origin, f"knowledge/tools/{name}", "import subprocess\nsubprocess.run(['git','ls-files','--error-unmatch','knowledge/AGENTS.md'], check=True, capture_output=True)\nprint('fixture validation passed')\n", "baseline validator") @@ -53,8 +58,8 @@ def world(tmp_path, monkeypatch): monkeypatch.setattr(reconcile, "_repository", lambda knowledge: "org/kb") monkeypatch.setenv("KB_KNOWLEDGE_CLONE", str(rt.knowledge.path)) return SimpleNamespace(rt=rt, scheduler=scheduler, origin=origin, sha=sha, base=baseline, - key=rt.outbox._key, options=dict(target=sha, allow_mergers=["owner"], - required_checks=["suite"], reason="authorized owner recovery")) + key=rt.outbox._key, options={"target": sha, "allow_mergers": ["owner"], + "required_checks": ["suite"], "reason": "authorized owner recovery"}) def test_supervised_admission_passes_real_activation_without_disposing_or_changing_modes(world):