From b8dff3d67e841be2ceab911e79b56a9da314da63 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Mon, 28 Sep 2026 23:23:30 +0800 Subject: [PATCH] Gate knowledge PRs the service did not open, and issue human-approved verdicts Two paths from the design that had no service side yet. Human-written knowledge PRs (source 4) need a signed verdict from the gate. PRs that the auto gate cannot pass (knowledge/tools, skills, ...) need a knowledge maintainer's approval turned into a human-approved verdict. Until now every such PR failed kb-gate forever. - external.poll_external runs from the scheduler (lease held, not globally paused). It takes each open, non-draft PR that touches knowledge/ and was not opened by the service, and judges it once per head and path (auto or human), so a head sent to people on the auto path is judged again as soon as a maintainer approves it, without a new push: - kb:human-approved with a knowledge maintainer's (CODEOWNERS of .github/kb-gate/ on main) standing approval of the CURRENT head: L1 on the governed pages (the approval replaces the whitelist and L2, never the rest of L1), then a signed human-approved verdict bound to those reviews. A later review by the same person replaces an earlier one, read across every page of reviews; - otherwise, all paths governed: the full gate (L1, L2, consistency) on current main plus the PR's changes; a pass gets an auto verdict. The PR's pre-images must equal main's, or the author is asked to rebase; like the service's own PRs, an auto verdict needs a current passing judge calibration (calibration_required otherwise, re-judged only once calibration is current, so no repeated paid judge calls); - anything else goes to people once per head. - The verdict manifest comes straight from git (the full merge-base..head raw diff), so it covers non-governed paths; context_base_sha is main at judging. Tests run the real kb-gate verifier on both kinds of verdict. - An external change set follows the ordinary merge flow. A new head (head_changed) or a moved context (stale_context) is judged again at the next poll instead of going to people, and an author closing their own PR does not trip the overturn breaker. - A withdrawn approval stops the PR. After signing, every pass re-checks that each named maintainer's latest review still approves the signed head; if not, it goes to people and a pause item dequeues it and makes it a draft. It keeps its status, and so the repository's queue slot, until that pause is acked (a failed pause is retried), and only then becomes approval_withdrawn. Until then it keeps holding the queue slot even if it was verdict_posted with an enqueue in flight, since it may have been queued meanwhile. A pause acked while an earlier signed enqueue could still run (the publisher may execute them in either order) does not complete it: the pause is repeated once that enqueue has certainly expired (plus a clock-skew margin). Completion is decided by when the acked pause was ISSUED (recorded on the pending item), never by when its ack arrived. The enqueue expiry is recovered at withdrawal from every durable record (the recorded value, the pending item, and unacked signed enqueue items still in the outbox), so a restart or a pre-upgrade change set cannot lose it. Remaining approvals never re-stage that head. The kb-gate verifier (bundle regenerated) now also checks each reviewer's latest review, not only that the bound review still reads APPROVED. - External change sets carry no operations, so the rules L1 finds them retiring or purging are stored and entered in the retirement ledger on merge, like our own, so later release sweeps can purge them. - KnowledgeRepo gains fetch_pull (a PR head fetched as data, never checked out), merge_base, show and raw_manifest. Co-Authored-By: Claude Opus 5.5 --- .github/kb-gate/MANIFEST.sha256 | 2 +- .github/kb-gate/kb_gate.py | 13 + doc/architecture/SPEC/kb_service.md | 19 + doc/architecture/SPEC/knowledge_service.md | 4 + .../kb_service/external.py | 283 ++++++++++++ src/infermatrix_copilot/kb_service/merge.py | 97 +++- .../kb_service/scheduler.py | 10 +- src/infermatrix_copilot/kb_service/sources.py | 29 ++ .../knowledge_service/gate_verifier.py | 13 + test/test_kb_external.py | 436 ++++++++++++++++++ test/test_kb_gate_verifier.py | 2 + 11 files changed, 900 insertions(+), 8 deletions(-) create mode 100644 src/infermatrix_copilot/kb_service/external.py create mode 100644 test/test_kb_external.py diff --git a/.github/kb-gate/MANIFEST.sha256 b/.github/kb-gate/MANIFEST.sha256 index 88a019f3..ae9e709f 100644 --- a/.github/kb-gate/MANIFEST.sha256 +++ b/.github/kb-gate/MANIFEST.sha256 @@ -1,6 +1,6 @@ 5f6e269fbbb951f9a49c40474a3d71c556ac389f5ebd5545f0ac762f85361723 README.md 77d88b9671927ab95bc6ac9ed57aaa24226423e151843381c2a3fa90cd0cafde config.json -e8157d90270e9f437b364e9ef95bc0603c80c6421fa591953525387cc4f5d4e0 kb_gate.py +7ae4a3ff0372e54d4a9e5efc3887c2a00e9004b138215901d4c4d8c1ae5a1588 kb_gate.py 5743c4a6943c291073c5344d408eaec28e8bc7440a8ee113365588728ded5e8a l1.py 39f69aa816b83b673443601d38e57040521664c5aca407154ba40a8a35bc61d9 lifecycle.py b0d0fab0210870f989da80f7ffba3978dbf723f4f087025524d1963c4081a009 ops.py diff --git a/.github/kb-gate/kb_gate.py b/.github/kb-gate/kb_gate.py index 417ca858..9a04b20a 100644 --- a/.github/kb-gate/kb_gate.py +++ b/.github/kb-gate/kb_gate.py @@ -544,6 +544,19 @@ def _human_approval_problems(ctx: Context, pr: dict, verdict: dict, head_sha: st problems.append(f"review {review_id} is by someone the verdict does not name") else: approvers.add(login) + if approvers: + # a later review by the same person (changes requested, dismissal) + # withdraws the approval even though the bound review still reads APPROVED + latest: dict[str, dict] = {} + for review in ctx.github.get_all(f"/repos/{ctx.repository}/pulls/{number}/reviews"): + login = str((review.get("user") or {}).get("login") or "").lower() + if login and review.get("state") in ("APPROVED", "CHANGES_REQUESTED", "DISMISSED"): + latest[login] = review + for login in sorted(approvers): + review = latest.get(login) or {} + if review.get("state") != "APPROVED" or review.get("commit_id") != head_sha: + problems.append(f"{login}'s latest review no longer approves this head") + approvers.discard(login) # knowledge maintainers = the CODEOWNERS of the gate itself. Governed pages # deliberately have no code owners (that would force a review on every auto # PR); paths outside the whitelist get GitHub's native code-owner review. diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 1d5f77d4..7ccd8483 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -108,3 +108,22 @@ stdlib + PyYAML + `cryptography`（`kb` extra）+ `.adapters` + `.knowledge_serv 中断后再次执行会找到它而不会重复暂存）。只有重建**通过**质量门才取代旧 PR：旧变更集进入 `superseding`，`close` 项 过期即重发，直到观测到 PR 已关闭才记为 `superseded`（不计入被人工推翻的熔断）；重建未通过、操作无法再应用或没有 可重放的操作 → `rebuild_failed` 并转人工，旧 PR 保持打开。其他失败（变更本身有问题）→ `gate_failed` 并转人工一次。 + +## 2026-09-28 外部知识 PR（来源④）与 human-approved 判定 +`external.poll_external`（调度器按 intake 间隔、全局未暂停时调用，需持有租约）处理知识仓库中**非服务创建**、非 draft、 +触碰 `knowledge/` 的打开 PR，每个 head 在每条路径（auto / human）上只评一次（auto 路径转人工的 head 在维护者审批后 +无需新推送即可按 human 路径重新评判）： +- `kb:human-approved` 且知识维护者（main 上 `.github/kb-gate/` 的 CODEOWNERS）对**当前 head** 的审批仍有效（同一人的 + 后续审阅覆盖之前的）→ 只在受治理页面上跑 L1（审批替代白名单与 L2，不替代其余 L1）→ 签发 `human-approved` 判定， + 绑定这些审阅。 +- 否则全部路径须为受治理页面 → 以**当前 main 加上 PR 改动**完整过质量门（L1、L2、一致性）→ 通过则签发 `auto` 判定。 + PR 的前像必须等于 main（否则队列落地的改动与签名不同），不等则请作者 rebase。 + 与服务自己的 PR 一样，`auto` 判定要求当前通过的评审校准；否则记为 `calibration_required`，校准恢复后才重新评判 + （期间不重复调用付费评审）。维护者审批读取全部分页的审阅，后续页上的撤回同样生效。 +- 其余情况（跨多个仓库、白名单外路径未经审批、L1 失败、质量门未通过）转人工，每个 head 一次。 +- 判定清单直接取自 git（`merge-base..head` 的完整原始 diff），覆盖白名单外路径；`context_base_sha` 为评判时的 main。 +- 暂存为 `external` 变更集后走普通合并流程。作者推送新 head → `head_changed`（不转人工）；上下文失效 → `stale_context`； + 两者都在下次轮询时重新评判。签发后维护者撤回审批（最新审阅不再批准该 head）→ 转人工并下发 `pause`（出队 + draft）； + 在 pause 回执成功前保持原状态（继续占用队列名额），回执后才记为 `approval_withdrawn`，剩余审批不会让同一 head 重新暂存。外部 PR 没有操作列表，L1 发现的退役/删除规则写入 `retirements`/`purges`，合并后同样进入 + 退役账本（之后的发版巡检据此 purge）。作者关闭自己的 PR 不计入熔断。只触碰非受治理知识路径的 PR 归入 `general`（若其接收 + 人工 PR），否则归入第一个接收人工 PR 的仓库。 diff --git a/doc/architecture/SPEC/knowledge_service.md b/doc/architecture/SPEC/knowledge_service.md index eec2f95c..83325997 100644 --- a/doc/architecture/SPEC/knowledge_service.md +++ b/doc/architecture/SPEC/knowledge_service.md @@ -70,3 +70,7 @@ L1 的索引检查允许为目录中已存在但未列入索引的页面补链 只接受单侧改动的三方合并（更严格，不更宽松）。 - 测试：`test_kb_gate_verifier.py`（重放/过期/伪造、清单漏项、块表不全、混合路径与可执行位、暂停清单、 上下文失效、状态发布、人工审批撤销、merge group 逐段与前序同目录 PR、验证包新鲜与自包含、独立运行）。 + +## 2026-09-28 human-approved：最新审阅 +验证器除核对签名绑定的审阅仍为 APPROVED 外，还读取该 PR 的全部审阅：每位被引用审阅者的**最新**审阅必须仍批准当前 head； +之后的 changes requested 或 dismiss 使该审批失效（即使被绑定的那条审阅仍显示 APPROVED）。 diff --git a/src/infermatrix_copilot/kb_service/external.py b/src/infermatrix_copilot/kb_service/external.py new file mode 100644 index 00000000..82b5e0dd --- /dev/null +++ b/src/infermatrix_copilot/kb_service/external.py @@ -0,0 +1,283 @@ +"""Knowledge PRs the service did not open (source ④), and the human-approved path. + +The scheduler polls the knowledge repository's open PRs. For every non-draft +PR that touches ``knowledge/`` and is not one of the service's own: + +* **human-approved**: the PR carries ``kb:human-approved`` and a knowledge + maintainer (a CODEOWNERS owner of ``.github/kb-gate/`` on main) approved + its CURRENT head. L1 must pass on the governed pages (the maintainer's + approval replaces the auto-merge whitelist and L2, never the rest of L1); + the service then signs a ``human-approved`` verdict naming those reviews. +* **auto**: every changed path is a governed page. The change is judged with + the full quality gate (L1, L2, consistency) against CURRENT main with the + PR's changes applied; a pass gets an ``auto`` verdict. + +Anything else goes to people once per head (with what would make it pass). +A staged external change set then follows the ordinary merge flow (verdict +comment, PR-stage kb-gate, merge queue). A new head, or a verdict whose +context moved, is simply judged again at the next poll. + +Judging "on current main" requires the PR's pre-images to equal main's: +otherwise the merge queue would apply different changes than were signed, and +the author is asked to rebase. +""" + +from __future__ import annotations + +from typing import Any + +from ..knowledge_service.gate_verifier import MAINTAINER_PATH, WHITELIST_CODES, code_owners, parse_codeowners +from ..knowledge_service.l1 import Change, check_changeset +from ..knowledge_service.ops import repo_scope +from .gate import run_gate +from .merge import knowledge_repository +from .runtime import calibration_current + +KIND = "external" +HUMAN_LABEL = "kb:human-approved" +GOVERNED = ("knowledge/repos/", "knowledge/general/") +SUFFIXES = (".md", ".yaml") +# statuses after which a head is judged again (the verdict can no longer verify) +REJUDGE = ("head_changed", "stale_context", "superseded") + + +def _governed(path: str) -> bool: + return path.startswith(GOVERNED) and path.endswith(SUFFIXES) + + +def _pulls(rt) -> list[dict]: + out: list[dict] = [] + for page in range(1, 11): + batch = rt.github.get(f"/repos/{knowledge_repository()}/pulls", state="open", per_page=100, page=page) + out.extend(batch or []) + if len(batch or []) < 100: + break + return out + + +def _maintainers(rt, main_sha: str) -> set[str]: + text = rt.knowledge.show(main_sha, ".github/CODEOWNERS") or "" + return set(code_owners(parse_codeowners(text), MAINTAINER_PATH)) + + +def approval_stands(rt, changeset: dict) -> bool: + """Whether a human-approved external change set's approvals still stand + (each named maintainer's latest review approves the signed head).""" + detail = changeset["detail"] + ids, _names = _approvals(rt, int(changeset["pr_number"]), changeset["head_sha"], + {str(n).lower() for n in detail.get("reviewers") or []}) + return bool(ids) and set(detail.get("reviewers") or []) <= set(_names) + + +def _approvals(rt, number: int, head: str, maintainers: set[str]) -> tuple[list[int], list[str]]: + """Reviews still standing as approvals of THIS head by maintainers (a later + review by the same person replaces an earlier one).""" + reviews: list[dict] = [] + for page in range(1, 51): # every review: a later one on a further page may retract an approval + batch = rt.github.get(f"/repos/{knowledge_repository()}/pulls/{number}/reviews", + per_page=100, page=page) or [] + reviews.extend(batch) + if len(batch) < 100: + break + latest: dict[str, dict] = {} + for review in reviews: + login = str((review.get("user") or {}).get("login") or "").lower() + if login and review.get("state") in ("APPROVED", "CHANGES_REQUESTED", "DISMISSED"): + latest[login] = review + ids, names = [], [] + for login, review in sorted(latest.items()): + if review.get("state") == "APPROVED" and review.get("commit_id") == head and login in maintainers: + ids.append(int(review["id"])) + names.append(login) + return ids, names + + +def _scope(paths: list[str], registry: dict) -> tuple[Any, str]: + """The one lifecycle a change belongs to, or (None, why not).""" + scopes = set() + for path in paths: + rel = path[len("knowledge/"):] + try: + scopes.add(repo_scope(rel).split("/")[-1]) + except Exception: # noqa: BLE001 - outside repos/ and general/ + continue + if len(scopes) > 1: + return None, f"spans several repositories: {sorted(scopes)}" + if not scopes: + # only non-governed knowledge (knowledge/tools, skills, ...): owned by + # the cross-repository scope when it takes PRs, else the first + # repository that does (the ledger is partitioned by repository) + takers = [lc for name, lc in sorted(registry.items()) if lc.enabled and lc.intake.human_prs] + general = registry.get("general") + chosen = general if general in takers else (takers[0] if takers else None) + if chosen is None: + return None, "no repository takes knowledge PRs through the service" + return chosen, "" + lifecycle = registry.get(scopes.pop()) + if lifecycle is None or not lifecycle.enabled or not lifecycle.intake.human_prs: + return None, "its repository does not take knowledge PRs through the service" + return lifecycle, "" + + +def _path_of(changeset: dict) -> str: + detail = changeset["detail"] + return detail.get("path") or ("human" if detail.get("source") == "human-approved" else "auto") + + +def _judged(rt, number: int, head: str, path: str) -> bool: + """Whether this head was already judged on this path. The key includes the + path: a head sent to people on the auto path is judged again once a + maintainer approves it (no push needed), while the paid auto judge still + runs once per head.""" + for lifecycle in rt.registry.values(): + for changeset in rt.ledger.changesets(lifecycle.repo, _ALL): + if changeset["kind"] == KIND and changeset["pr_number"] == number and _path_of(changeset) == path \ + and changeset["head_sha"] == head and changeset["status"] not in REJUDGE: + if changeset["status"] == "calibration_required" and calibration_current(rt, lifecycle): + rt.ledger.update_changeset(changeset["id"], status="superseded") + return False # the judge is calibrated now: judge this head again + return True + return False + + +def _ours(rt, number: int) -> bool: + for lifecycle in rt.registry.values(): + for changeset in rt.ledger.changesets(lifecycle.repo, _ALL): + if changeset["pr_number"] == number and changeset["kind"] != KIND: + return True + return False + + +_ALL = ("gated", "pr_requested", "pr_open", "verdict_posted", "queued", "merged", "closed", "failed", "human", + "head_changed", "stale_context", "superseded", "superseding", "gate_failed", "rebuild_failed", + "paused", "shadow_recorded", "calibration_required", "approval_withdrawn") + + +def poll_external(rt) -> list[str]: + """One pass over open knowledge PRs; needs the scheduler's lease.""" + owner = getattr(rt, "lease_owner", None) + if rt.outbox is None or not owner: + return [] + events: list[str] = [] + main_sha = rt.knowledge.fetch() + maintainers: set[str] | None = None + for pr in _pulls(rt): + number = int(pr["number"]) + head = str((pr.get("head") or {}).get("sha") or "") + labels = {str(label.get("name")) for label in pr.get("labels") or []} + human = HUMAN_LABEL in labels + if pr.get("draft") or not head or _ours(rt, number) or _judged(rt, number, head, "human" if human else "auto"): + continue + if rt.knowledge.fetch_pull(number) != head: + continue # pushed while we looked: next poll + merge_base = rt.knowledge.merge_base(main_sha, head) + manifest = rt.knowledge.raw_manifest(merge_base, head) + paths = [entry["path"] for entry in manifest] + if not any(p.startswith("knowledge/") for p in paths): + continue + path = "human" if human else "auto" + lifecycle, why = _scope([p for p in paths if p.startswith("knowledge/")], rt.registry) + if lifecycle is None: + events.append(_to_people(rt, None, number, head, f"PR #{number} {why}", path)) + continue + if human: + maintainers = maintainers if maintainers is not None else _maintainers(rt, main_sha) + review_ids, reviewers = _approvals(rt, number, head, maintainers) + if not review_ids: + continue # labeled but not (yet) approved on this head by a maintainer: wait + elif not all(_governed(p) for p in paths): + events.append(_to_people(rt, lifecycle, number, head, + f"PR #{number} changes paths outside the governed knowledge pages; " + f"a knowledge maintainer can approve it and add {HUMAN_LABEL}", path)) + continue + base_main = rt.knowledge.knowledge_files(main_sha) + pre = rt.knowledge.knowledge_files(merge_base) + post = rt.knowledge.knowledge_files(head) + governed = [e for e in manifest if _governed(e["path"])] + rels = [e["path"][len("knowledge/"):] for e in governed] + moved = [rel for rel in rels if base_main.get(rel) != pre.get(rel)] + if moved: + events.append(_to_people(rt, lifecycle, number, head, + f"PR #{number} must be rebased: main changed {moved[:3]} since it branched", path)) + continue + head_files = dict(base_main) + for rel in rels: + if rel in post: + head_files[rel] = post[rel] + else: + head_files.pop(rel, None) + changes = [Change(e["path"], e["status"] if e["status"] in "AMD" else "M", e["old_mode"], e["new_mode"]) + for e in (governed if human else manifest)] + release = rt.release_for(lifecycle.repo) + detail: dict[str, Any] = {"operations": [], "event_ids": [], "base_sha": main_sha, "release": release, + "manifest": manifest, "external_pr": number, + "title": str(pr.get("title") or ""), "author": str((pr.get("user") or {}).get("login") or "")} + if human: + result = check_changeset(base_main, head_files, changes, + external_texts=rt.knowledge.external_texts(main_sha), release=release) + issues = [i for i in result.issues if i.code not in WHITELIST_CODES] + if issues: + events.append(_to_people(rt, lifecycle, number, head, f"PR #{number} fails L1: " + "; ".join( + f"{i.code} {i.path} {i.detail}" for i in issues[:3]), path)) + continue + detail.update(source="human-approved", review_ids=review_ids, reviewers=reviewers, + generator="human", judge="human", decision={"blocks": [], "consistency": []}, + **_lifecycle_bookkeeping(result, head_files)) + status = "pr_open" + else: + evidence = [{"source_reference": f"PR #{number}", "title": detail["title"], + "body": str(pr.get("body") or "")[:3000], "author": detail["author"]}] + decision = run_gate(base=base_main, head=head_files, changes=changes, + external_texts=rt.knowledge.external_texts(main_sha), evidence=evidence, + gateway=rt.gateway, judge=rt.judge, release=release, + repo_dir=lifecycle.knowledge_dir, protected_rules=lifecycle.protected_rules, + retire_ratio=lifecycle.retire_ratio, max_files=lifecycle.max_files) + detail.update(source="auto", generator="human", judge=rt.judge.label(), decision=decision.to_dict(), + **_lifecycle_bookkeeping(decision.l1, head_files)) + status = {"pass": "pr_open", "fail": "failed", "human": "human"}[decision.status] + if status == "pr_open" and lifecycle.auto_merge and not calibration_current(rt, lifecycle): + # an auto verdict is only as good as the judge (as for our own PRs) + status = "calibration_required" + detail["decision"]["reasons"] = [*detail["decision"].get("reasons", []), + "judge calibration missing or outdated"] + if status == "pr_open" and not lifecycle.auto_merge: + status = "shadow_recorded" + changeset_id = rt.ledger.new_changeset_id(lifecycle.repo, KIND) + rt.save_changeset_files(changeset_id, {"base_sha": main_sha, "files": {r: head_files[r] for r in rels + if r in head_files}, + "deleted": [r for r in rels if r not in head_files], "evidence": []}) + rt.ledger.stage_intake(owner, lifecycle.repo, changeset_id, kind=KIND, status=status, verdicts=[], + human_reason="" if status in ("pr_open", "shadow_recorded") + else f"external PR #{number}: " + "; ".join(detail["decision"].get("reasons", [])), + drafted_events=[], detail=detail) + rt.ledger.update_changeset(changeset_id, pr_number=number, head_sha=head) + rt.trace("decision", context={"repo": lifecycle.repo, "changeset_id": changeset_id, "pr": number, + "playbook": "kb-external", "step": "gate"}, + result={"status": status, "source": detail["source"], + "reviewers": detail.get("reviewers", [])}) + events.append(f"external {changeset_id} PR #{number} {detail['source']} {status}") + return events + + +def _lifecycle_bookkeeping(result, head_files: dict[str, str]) -> dict: + """The rules this change retires and purges, as the merge flow records them + (our own change sets derive this from their operations).""" + from ..knowledge_service.ops import all_rule_ids + + pages = {rid: page for page, text in sorted(head_files.items()) if page.endswith(".md") + for rid in all_rule_ids({page: text})} + return {"retirements": [{"rule_id": rid, "page": pages.get(rid, "")} for rid in result.retired], + "purges": list(result.purged)} + + +def _to_people(rt, lifecycle, number: int, head: str, reason: str, path: str) -> str: + """Record a head that people must handle, once (as a human change set).""" + repo = lifecycle.repo if lifecycle is not None else next(iter(rt.registry)) + changeset_id = rt.ledger.new_changeset_id(repo, KIND) + rt.save_changeset_files(changeset_id, {"base_sha": "", "files": {}, "deleted": [], "evidence": []}) + rt.ledger.stage_intake(rt.lease_owner, repo, changeset_id, kind=KIND, status="human", verdicts=[], + human_reason=reason, drafted_events=[], + detail={"external_pr": number, "reason": reason, "path": path}) + rt.ledger.update_changeset(changeset_id, pr_number=number, head_sha=head) + return f"external {changeset_id} PR #{number} to people" + diff --git a/src/infermatrix_copilot/kb_service/merge.py b/src/infermatrix_copilot/kb_service/merge.py index 3a615a1e..6da6346d 100644 --- a/src/infermatrix_copilot/kb_service/merge.py +++ b/src/infermatrix_copilot/kb_service/merge.py @@ -27,6 +27,8 @@ IN_FLIGHT = ("pr_requested", "pr_open", "verdict_posted", "queued") # a verdict must first verify within 72 hours: re-sign well before that RESIGN_AFTER = 48 * 3600 +# margin for the publisher's clock when deciding an item has certainly expired +CLOCK_SKEW = 5 * 60 # a queued PR that neither merged nor left the queue by then was dropped by a # failing merge group (GitHub does not tell us): re-sign and re-enqueue QUEUE_STALL = 2 * 3600 @@ -68,7 +70,7 @@ def issue_once(rt, repo: str, changeset: dict, kind: str, body: dict) -> bool: return False item = rt.outbox.issue(repo, kind, body) rt.ledger.update_changeset(changeset["id"], pending_item={ - "id": item.id, "kind": kind, "expires_at": item.expires_at}) + "id": item.id, "kind": kind, "expires_at": item.expires_at, "issued_at": item.issued_at}) return True @@ -93,7 +95,21 @@ def apply_acks(rt, acks: list[dict]) -> None: current = rt.ledger.changeset(changeset_id) if not ack.get("ok") or current["status"] not in ("pr_open", "verdict_posted", "queued"): continue - if is_paused(rt.ledger, current["repo"]): + if current["detail"].get("withdrawn"): + live_until = float(current["detail"].get("enqueue_expires_at") or 0) + CLOCK_SKEW + # decided by when the ACKED pause was issued, never by when its + # ack arrived: only a pause issued after every enqueue died is final + pending = current.get("pending_item") or {} + acked_issued = float(pending.get("issued_at") or 0) \ + if pending.get("kind") == "pause" and pending.get("id") == ack.get("item_id") else 0.0 + if acked_issued < live_until: + # an enqueue signed earlier could still run after this + # pause: pause again once it has certainly expired + rt.ledger.update_changeset(changeset_id, pending_item=None, detail={ + **current["detail"], "pause_again_after": live_until}) + else: + rt.ledger.update_changeset(changeset_id, status="approval_withdrawn", pending_item=None) + elif is_paused(rt.ledger, current["repo"]): rt.ledger.update_changeset(changeset_id, status="paused", pending_item=None) else: # resumed before this (late) pause landed: the PR is now @@ -131,16 +147,22 @@ def sign_verdict(rt, changeset: dict) -> dict: touched = set(data["files"]) | set(data.get("deleted", [])) manifest = manifest_from_files({k: v for k, v in base.items() if k in touched}, {k: v for k, v in head.items() if k in touched}) + # an external PR's manifest comes from git and covers every changed path + manifest = detail.get("manifest") or manifest decision = detail["decision"] blocks = [{k: b[k] for k in ("block_id", "kind", "path", "rule_id", "op", "sha256", "verdict", "model")} for b in decision["blocks"]] consistency = [{"owner_dir": c["owner_dir"], "verdict": c["verdict"], "pages": c["pages"]} for c in decision["consistency"]] + human = detail.get("source") == "human-approved" verdict = build_verdict( repository=knowledge_repository(), pr=int(changeset["pr_number"]), head_sha=changeset["head_sha"], - context_base_sha=detail["base_sha"], manifest=manifest, blocks=blocks, consistency=consistency, + context_base_sha=detail["base_sha"], manifest=manifest, blocks=[] if human else blocks, + consistency=[] if human else consistency, release=detail["release"], upstream={}, facts=[], models={"generator": detail["generator"], "judge": detail["judge"]}, + source="human-approved" if human else "auto", + review_ids=tuple(detail.get("review_ids") or ()), reviewers=tuple(detail.get("reviewers") or ()), issued_at=rt.clock(), ) return sign("kb-gate-verdict", verdict, rt.outbox._key) @@ -178,6 +200,27 @@ def _epoch(value) -> float: return 0.0 +def _live_enqueue_until(rt, changeset: dict) -> float: + """The latest expiry of any enqueue signed for this change set that the + publisher could still execute, from every durable record: the recorded + value, the pending item, and unacknowledged signed items still in the + outbox (an item leaves the outbox only once its ack is collected).""" + latest = float(changeset["detail"].get("enqueue_expires_at") or 0) + pending = changeset.get("pending_item") or {} + if pending.get("kind") == "enqueue": + latest = max(latest, float(pending.get("expires_at") or 0)) + for path in sorted(rt.outbox.outbox_dir.glob("*.json")): + if path.name == "control.json": + continue + try: + payload = json.loads(path.read_text(encoding="utf-8"))["payload"] + except (OSError, ValueError, KeyError, TypeError): + continue + if payload.get("kind") == "enqueue" and (payload.get("body") or {}).get("changeset_id") == changeset["id"]: + latest = max(latest, float(payload.get("expires_at") or 0)) + return latest + + def _enqueue_outstanding(rt, changeset: dict) -> bool: """A signed enqueue that the publisher may still execute (unexpired).""" pending = changeset.get("pending_item") or {} @@ -276,10 +319,12 @@ def advance(rt, lifecycle) -> list[str]: if rt.outbox is None: return events paused = is_paused(rt.ledger, lifecycle.repo) - # the one queue slot per repository is taken by a queued PR AND by any PR - # whose signed enqueue is still executable (not yet acked) + # the one queue slot per repository is taken by a queued PR, by any PR whose + # signed enqueue is still executable (not yet acked), and by a withdrawn PR + # until its pause (dequeue) is acked: it may have been queued meanwhile queued = rt.ledger.changesets(lifecycle.repo, ("queued",)) + [ - cs for cs in rt.ledger.changesets(lifecycle.repo, ("verdict_posted",)) if _enqueue_outstanding(rt, cs)] + cs for cs in rt.ledger.changesets(lifecycle.repo, ("verdict_posted",)) + if _enqueue_outstanding(rt, cs) or cs["detail"].get("withdrawn")] for changeset in rt.ledger.changesets(lifecycle.repo, (*IN_FLIGHT, "superseding")): number = changeset["pr_number"] if number is None: @@ -307,6 +352,11 @@ def advance(rt, lifecycle) -> list[str]: events.append(f"closed {changeset['id']}") continue head = str((pr.get("head") or {}).get("sha") or "") + if head != changeset["head_sha"] and changeset["kind"] == "external": + # authors push: the new head is simply judged again at the next poll + rt.ledger.update_changeset(changeset["id"], status="head_changed", pending_item=None) + events.append(f"head_changed {changeset['id']}") + continue if head != changeset["head_sha"]: _outcome(rt, changeset, "head_changed", head_sha=head) rt.ledger.update_changeset(changeset["id"], status="head_changed") @@ -315,6 +365,27 @@ def advance(rt, lifecycle) -> list[str]: continue if paused: continue # the pause itself is issued by `kb pause` / breakers + if changeset["kind"] == "external" and changeset["detail"].get("source") == "human-approved" \ + and changeset["status"] in ("verdict_posted", "queued"): + from .external import approval_stands + + withdrawn = bool(changeset["detail"].get("withdrawn")) + if withdrawn or not approval_stands(rt, changeset): + # a withdrawn approval must stop the PR: dequeue + draft. The + # change set keeps its status (and its queue slot) until the + # pause is acked; only then is it approval_withdrawn + if not withdrawn: + rt.ledger.update_changeset(changeset["id"], detail={ + **changeset["detail"], "withdrawn": True, + "enqueue_expires_at": _live_enqueue_until(rt, changeset)}) + rt.ledger.enqueue_human(lifecycle.repo, f"PR #{number}: approval withdrawn", changeset["id"]) + changeset = rt.ledger.changeset(changeset["id"]) + events.append(f"approval_withdrawn {changeset['id']}") + if rt.clock() >= float(changeset["detail"].get("pause_again_after") or 0): + issue_once(rt, lifecycle.repo, changeset, "pause", { + "changeset_id": changeset["id"], "pr": int(number), + "reason": "a knowledge maintainer withdrew the approval"}) + continue signed_at = float(changeset["detail"].get("verdict_issued_at") or 0) stale_verdict = bool(signed_at) and rt.clock() - signed_at > RESIGN_AFTER queued_at = float(changeset["detail"].get("queued_at") or 0) @@ -339,6 +410,10 @@ def advance(rt, lifecycle) -> list[str]: if any(m in description for m in RESIGN_GATE): _resign(rt, changeset, description) events.append(f"resign {changeset['id']}") + elif any(m in description for m in REBUILD_GATE) and changeset["kind"] == "external": + # nothing of ours to rebuild: judge the PR again on current main + rt.ledger.update_changeset(changeset["id"], status="stale_context", pending_item=None) + events.append(f"stale_context {changeset['id']}") elif any(m in description for m in REBUILD_GATE): if not getattr(rt, "lease_owner", None): continue # only the scheduler (holding the lease) rebuilds @@ -367,6 +442,11 @@ def advance(rt, lifecycle) -> list[str]: if _gate_status(rt, changeset["head_sha"]) == "success": if issue_once(rt, lifecycle.repo, changeset, "enqueue", { "changeset_id": changeset["id"], "pr": int(number), "head_sha": changeset["head_sha"]}): + current = rt.ledger.changeset(changeset["id"]) + # remembered past any later pending item: a withdrawal must + # outlast every enqueue that could still execute + rt.ledger.update_changeset(changeset["id"], detail={ + **current["detail"], "enqueue_expires_at": current["pending_item"]["expires_at"]}) events.append(f"enqueue issued {changeset['id']}") queued = [changeset] # at most one queued knowledge PR per repository return events @@ -399,6 +479,11 @@ def record_retirements(rt, changeset: dict) -> None: "reason": op.get("reason", "")}) elif op["kind"] == "purge": rt.ledger.mark_purged(changeset["repo"], op["rule_id"]) + # external PRs carry no operations: what L1 found them retiring/purging + for item in changeset["detail"].get("retirements", []): + rt.ledger.record_retirement(changeset["repo"], item["rule_id"], item["page"], release) + for rule_id in changeset["detail"].get("purges", []): + rt.ledger.mark_purged(changeset["repo"], rule_id) def pause_open_prs(ledger, outbox, repo: str, reason: str) -> int: diff --git a/src/infermatrix_copilot/kb_service/scheduler.py b/src/infermatrix_copilot/kb_service/scheduler.py index 9038cffe..b5aa4187 100644 --- a/src/infermatrix_copilot/kb_service/scheduler.py +++ b/src/infermatrix_copilot/kb_service/scheduler.py @@ -26,6 +26,7 @@ from . import merge from .activate import ActivationError, activate, activation_lock +from .external import poll_external from .runtime import collect_events, publish, run_intake from .sweep import UpstreamRepo, detect_release, run_sweep @@ -73,6 +74,12 @@ def tick(self) -> None: self._repo_tick(lifecycle) except Exception as exc: # isolate repositories from each other self._record(lifecycle.repo, "error", error=repr(exc), trace=traceback.format_exc()[-2000:]) + if self._due("external", self.intake_every) and not merge.is_paused(rt.ledger, "*"): + try: # knowledge PRs the service did not open (source 4, human approvals) + for event in poll_external(rt): + self._record("*", "external", detail=event) + except Exception as exc: + self._record("*", "error", error=repr(exc), trace=traceback.format_exc()[-2000:]) try: sha = rt.knowledge.fetch() with activation_lock(rt.state_dir): # the pin cannot change under this decision @@ -120,8 +127,9 @@ def _repo_tick(self, lifecycle) -> None: def _overturn_breaker(self, lifecycle) -> None: rt = self.rt + # only the service's own PRs: an author closing their own PR is no overturn recent = [cs for cs in rt.ledger.changesets(lifecycle.repo, ("closed",)) - if rt.clock() - float(cs["updated_at"]) < OVERTURN_WINDOW] + if cs["kind"] != "external" and rt.clock() - float(cs["updated_at"]) < OVERTURN_WINDOW] state = rt.ledger.repo_state(lifecycle.repo) if len(recent) >= OVERTURN_LIMIT and not state["paused"]: reason = f"overturn breaker: {len(recent)} knowledge PRs closed unmerged within 24h" diff --git a/src/infermatrix_copilot/kb_service/sources.py b/src/infermatrix_copilot/kb_service/sources.py index d3d98fea..c78b6671 100644 --- a/src/infermatrix_copilot/kb_service/sources.py +++ b/src/infermatrix_copilot/kb_service/sources.py @@ -78,6 +78,35 @@ def _texts(self, rev: str, prefixes: tuple[str, ...], suffixes: tuple[str, ...] continue return texts + def fetch_pull(self, number: int) -> str: + """Fetch a pull request's head as data (never checked out); its SHA.""" + self._git("fetch", "--quiet", self.remote, f"+refs/pull/{int(number)}/head:refs/kb/pull/{int(number)}") + return self._git("rev-parse", f"refs/kb/pull/{int(number)}").decode().strip() + + def merge_base(self, a: str, b: str) -> str: + return self._git("merge-base", a, b).decode().strip() + + def show(self, rev: str, path: str) -> str | None: + try: + return self._git("show", f"{rev}:{path}").decode("utf-8", "replace") + except SourceError: + return None + + def raw_manifest(self, old: str, new: str) -> list[dict]: + """Every changed path as a kb-gate manifest entry (renames are D + A), + the exact form the verifier recomputes.""" + out = self._git("diff", "--raw", "-z", "--no-renames", "--no-abbrev", old, new).split(b"\0") + entries, zero = [], "0" * 40 + for index in range(0, len(out) - 1, 2): + old_mode, new_mode, old_blob, new_blob, status = out[index].decode()[1:].split() + entries.append({ + "path": out[index + 1].decode("utf-8", "replace"), "status": status[0], + "old_blob": "" if old_blob == zero else old_blob, "new_blob": "" if new_blob == zero else new_blob, + "old_mode": "" if old_mode == "000000" else old_mode, + "new_mode": "" if new_mode == "000000" else new_mode, + }) + return entries + def knowledge_files(self, rev: str) -> dict[str, str]: """Knowledge-relative path -> text for the governed tree at ``rev``.""" files = self._texts(rev, ("knowledge/repos/", "knowledge/general/"), KNOWLEDGE_SUFFIXES) diff --git a/src/infermatrix_copilot/knowledge_service/gate_verifier.py b/src/infermatrix_copilot/knowledge_service/gate_verifier.py index d90224a5..fcd1b926 100644 --- a/src/infermatrix_copilot/knowledge_service/gate_verifier.py +++ b/src/infermatrix_copilot/knowledge_service/gate_verifier.py @@ -543,6 +543,19 @@ def _human_approval_problems(ctx: Context, pr: dict, verdict: dict, head_sha: st problems.append(f"review {review_id} is by someone the verdict does not name") else: approvers.add(login) + if approvers: + # a later review by the same person (changes requested, dismissal) + # withdraws the approval even though the bound review still reads APPROVED + latest: dict[str, dict] = {} + for review in ctx.github.get_all(f"/repos/{ctx.repository}/pulls/{number}/reviews"): + login = str((review.get("user") or {}).get("login") or "").lower() + if login and review.get("state") in ("APPROVED", "CHANGES_REQUESTED", "DISMISSED"): + latest[login] = review + for login in sorted(approvers): + review = latest.get(login) or {} + if review.get("state") != "APPROVED" or review.get("commit_id") != head_sha: + problems.append(f"{login}'s latest review no longer approves this head") + approvers.discard(login) # knowledge maintainers = the CODEOWNERS of the gate itself. Governed pages # deliberately have no code owners (that would force a review on every auto # PR); paths outside the whitelist get GitHub's native code-owner review. diff --git a/test/test_kb_external.py b/test/test_kb_external.py new file mode 100644 index 00000000..b543ce27 --- /dev/null +++ b/test/test_kb_external.py @@ -0,0 +1,436 @@ +"""Knowledge PRs the service did not open: auto gate, human approval, people.""" + +from __future__ import annotations + +import json +import subprocess +import uuid + +from infermatrix_copilot.kb_service import merge +from infermatrix_copilot.kb_service.external import poll_external +from infermatrix_copilot.knowledge_service.signing import load_public_key, public_key_text, verify +from infermatrix_copilot.knowledge_service.ops import KnowledgeOperation as Op, apply_operations +from test_kb_flow import KnowledgeGitHub, _flow_runtime, _items +from test_kb_intake_gate import PAGE, _rule, _tree + +REPO = "JiusiServe/InferMatrixCopilot" + + +def _git(cwd, *args): + return subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True, text=True).stdout.strip() + + +class PullsGitHub(KnowledgeGitHub): + def __init__(self): + super().__init__() + self.open: dict[int, dict] = {} + self.reviews: dict[int, list[dict]] = {} + + def _answer(self, url): + path = url.split("?", 1)[0] + if path.endswith(f"/repos/{REPO}/pulls"): + return list(self.open.values()) if "page=1" in url else [] + if "/reviews" in path: + reviews = self.reviews.get(int(path.split("/pulls/")[1].split("/")[0]), []) + from urllib.parse import parse_qs, urlsplit + page = int(parse_qs(urlsplit(url).query).get("page", ["1"])[0]) + return reviews[(page - 1) * 100: page * 100] + return super()._answer(url) + + +def _setup(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + github = PullsGitHub() + github.prs = rt.github.prs + rt.github = github + rt.lease_owner = rt.ledger.acquire_lease("scheduler") + origin = tmp_path / "origin" + (origin / ".github").mkdir(exist_ok=True) + (origin / ".github" / "CODEOWNERS").write_text("/.github/ @alice\n") + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "owners") + return rt, lifecycle, origin + + +def _open_human_pr(rt, origin, number, files: dict[str, str | None], *, labels=(), draft=False): + _git(origin, "checkout", "-q", "-B", f"human-{number}", "main") + for rel, text in files.items(): + target = origin / rel + if text is None: + target.unlink() + else: + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(text) + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=h", "-c", "user.email=h@e", "commit", "-q", "--allow-empty", + "-m", f"pr {number} {uuid.uuid4().hex}") # a distinct head even for an identical tree + head = _git(origin, "rev-parse", "HEAD") + _git(origin, "update-ref", f"refs/pull/{number}/head", head) + _git(origin, "checkout", "-q", "main") + rt.github.open[number] = {"number": number, "draft": draft, "head": {"sha": head}, "title": "Add a rule", + "body": "from a person", "user": {"login": "dev"}, + "labels": [{"name": name} for name in labels]} + rt.github.prs[number] = {"state": "open", "merged": False, "head": {"sha": head}} + return head + + +def _add_rule_files(): + base = _tree() + return {f"knowledge/{rel}": text for rel, text in apply_operations( + base, [Op("add", PAGE, "DEMO-2a", _rule("DEMO-2a", "PR #11"))], release="v1", today="2026-09-28").files.items()} + + +def _verdict(tmp_path, rt): + comment = _items(tmp_path, "post_verdict")[-1]["body"]["comment"] + envelope = json.loads(comment.split("```json\n", 1)[1].rsplit("\n```", 1)[0]) + return verify("kb-gate-verdict", envelope, load_public_key(public_key_text(rt.outbox._key.public_key()))) + + +def test_a_human_knowledge_pr_passes_the_auto_gate_and_gets_a_verdict(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + head = _open_human_pr(rt, origin, 7, _add_rule_files()) + (event,) = poll_external(rt) + assert "auto pr_open" in event + (changeset,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] + assert changeset["pr_number"] == 7 and changeset["head_sha"] == head + assert poll_external(rt) == [] # judged once per head + assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset['id']}"] + verdict = _verdict(tmp_path, rt) + merge_base = _git(origin, "merge-base", "main", head) + assert verdict["source"] == "auto" and verdict["pr"] == 7 and verdict["head_sha"] == head + assert verdict["manifest"] == rt.knowledge.raw_manifest(merge_base, head) + assert verdict["blocks"] and all(b["verdict"] == "pass" for b in verdict["blocks"]) + + +def test_paths_outside_the_governed_pages_need_a_maintainer(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + head = _open_human_pr(rt, origin, 8, {"knowledge/tools/check.md": "tool notes\n"}) + (event,) = poll_external(rt) + assert event.endswith("to people") + assert "kb:human-approved" in rt.ledger.human_queue("demo")[-1]["reason"] + assert poll_external(rt) == [] # once per head + # a maintainer approves the CURRENT head and labels it: a human-approved verdict + rt.github.open[8]["labels"] = [{"name": "kb:human-approved"}] + rt.github.reviews[8] = [{"id": 99, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] + head2 = _open_human_pr(rt, origin, 8, {"knowledge/tools/check.md": "tool notes v2\n"}, + labels=("kb:human-approved",)) + rt.github.reviews[8] = [{"id": 99, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] + assert poll_external(rt) == [] # approval is for the old head: wait + rt.github.reviews[8].append({"id": 100, "state": "APPROVED", "commit_id": head2, "user": {"login": "alice"}}) + (event,) = poll_external(rt) + assert "human-approved pr_open" in event + (changeset,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] + merge.advance(rt, lifecycle) + verdict = _verdict(tmp_path, rt) + assert verdict["source"] == "human-approved" and verdict["review_ids"] == [100] + assert verdict["reviewers"] == ["alice"] and verdict["blocks"] == [] + assert changeset["head_sha"] == head2 + + +def test_approvals_by_non_maintainers_or_later_retracted_do_not_count(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + head = _open_human_pr(rt, origin, 9, {"knowledge/tools/x.md": "x\n"}, labels=("kb:human-approved",)) + rt.github.reviews[9] = [{"id": 1, "state": "APPROVED", "commit_id": head, "user": {"login": "mallory"}}, + {"id": 2, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}, + {"id": 3, "state": "DISMISSED", "commit_id": head, "user": {"login": "alice"}}] + assert poll_external(rt) == [] + + +def test_a_pr_branched_before_main_changed_the_same_page_must_be_rebased(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + _open_human_pr(rt, origin, 10, _add_rule_files()) + page = origin / "knowledge" / PAGE + page.write_text(page.read_text() + "\n\n") + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "main moves") + (event,) = poll_external(rt) + assert event.endswith("to people") and "rebased" in rt.ledger.human_queue("demo")[-1]["reason"] + + +def test_a_new_head_is_judged_again_and_drafts_and_our_own_prs_are_skipped(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + _open_human_pr(rt, origin, 11, _add_rule_files(), draft=True) + assert poll_external(rt) == [] + rt.github.open[11]["draft"] = False + assert len(poll_external(rt)) == 1 + (first,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] + head2 = _open_human_pr(rt, origin, 11, {**_add_rule_files(), "knowledge/repos/demo/core/_index.md": + "# core\n\n- [rules](rules.md)\n"}) + assert merge.advance(rt, lifecycle) == [f"head_changed {first['id']}"] + assert rt.ledger.human_queue("demo") == [] # an author pushing is not an incident + (event,) = poll_external(rt) + assert "PR #11" in event + assert rt.ledger.changesets("demo", ("pr_open",))[-1]["head_sha"] == head2 + # a PR the service opened itself is never judged as external + ours = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.lease_owner, "demo", ours, detail={}, status="pr_open", verdicts=[], + human_reason="", drafted_events=[]) + rt.ledger.update_changeset(ours, pr_number=12) + _open_human_pr(rt, origin, 12, _add_rule_files()) + assert all("PR #12" not in event for event in poll_external(rt)) + + +def _verifier_passes(tmp_path, rt, origin, number, labels=(), reviews=()): + """Run the real kb-gate PR-stage verifier on the verdict this service posted.""" + import time + + from infermatrix_copilot.knowledge_service import gate_verifier as gv + from infermatrix_copilot.knowledge_service.signing import sign + + (comment,) = [item["body"]["comment"] for item in _items(tmp_path, "post_verdict") + if item["body"]["pr"] == number] + head = rt.github.open[number]["head"]["sha"] + + class VerifierGitHub: + def get(self, path): + if path.endswith(f"/pulls/{number}"): + return {"number": number, "state": "open", "draft": False, "head": {"sha": head}, + "labels": [{"name": name} for name in labels]} + return {r["id"]: r for r in reviews}[int(path.rsplit("/", 1)[1])] + + def get_all(self, path): + return list(reviews) if path.endswith("/reviews") else [{"body": comment}] + + holds = {"issued_at": time.time(), "sequence": 1, "global": False, "repos": [], "prs": []} + ctx = gv.Context(git=gv.Git(rt.knowledge.path), github=VerifierGitHub(), repository=REPO, + public_key=rt.outbox._key.public_key(), holds_loader=lambda: sign("kb-holds", holds, rt.outbox._key), + codeowners=gv.parse_codeowners((origin / ".github" / "CODEOWNERS").read_text()), now=time.time()) + _head, problems = gv.verify_pr(ctx, number) + return problems + + +def test_verdicts_for_external_prs_pass_the_real_kb_gate_verifier(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + rt.clock = __import__("time").time # verdict windows are checked against the real clock + _open_human_pr(rt, origin, 7, _add_rule_files()) + poll_external(rt) + merge.advance(rt, lifecycle) + assert _verifier_passes(tmp_path, rt, origin, 7) == [] + + head = _open_human_pr(rt, origin, 8, {"knowledge/tools/check.md": "tool notes\n"}, labels=("kb:human-approved",)) + review = {"id": 100, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}} + rt.github.reviews[8] = [review] + poll_external(rt) + merge.advance(rt, lifecycle) + assert _verifier_passes(tmp_path, rt, origin, 8, labels=("kb:human-approved",), reviews=(review,)) == [] + + +def test_a_retraction_on_a_later_page_of_reviews_is_seen(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + head = _open_human_pr(rt, origin, 20, {"knowledge/tools/x.md": "x\n"}, labels=("kb:human-approved",)) + noise = [{"id": 1000 + i, "state": "COMMENTED", "commit_id": head, "user": {"login": f"u{i}"}} + for i in range(120)] + rt.github.reviews[20] = [{"id": 1, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}, + *noise, + {"id": 2, "state": "CHANGES_REQUESTED", "commit_id": head, "user": {"login": "alice"}}] + assert poll_external(rt) == [] + + +def test_an_auto_verdict_needs_a_current_judge_calibration(tmp_path): + from dataclasses import replace + + from test_kb_intake_gate import _calibrate + + rt, lifecycle, origin = _setup(tmp_path) + rt.registry["demo"] = replace(rt.registry["demo"], calibration_set="missing") + _open_human_pr(rt, origin, 21, _add_rule_files()) + (event,) = poll_external(rt) + assert event.endswith("calibration_required") + assert poll_external(rt) == [] # not re-judged (no paid judge calls) until calibrated + _calibrate(tmp_path, rt, rt.registry["demo"]) + (event,) = poll_external(rt) + assert event.endswith("auto pr_open") + + +def test_a_head_sent_to_people_can_be_approved_without_a_new_push(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + head = _open_human_pr(rt, origin, 30, {"knowledge/tools/y.md": "y\n"}) + (event,) = poll_external(rt) + assert event.endswith("to people") + assert poll_external(rt) == [] + rt.github.open[30]["labels"] = [{"name": "kb:human-approved"}] + rt.github.reviews[30] = [{"id": 7, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] + (event,) = poll_external(rt) + assert "human-approved pr_open" in event + assert poll_external(rt) == [] + + +def test_a_withdrawn_approval_after_signing_stops_the_pr(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + rt.clock = __import__("time").time # the verifier checks the verdict window in real time + head = _open_human_pr(rt, origin, 40, {"knowledge/tools/z.md": "z\n"}, labels=("kb:human-approved",)) + rt.github.reviews[40] = [{"id": 5, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] + poll_external(rt) + (changeset,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] + merge.advance(rt, lifecycle) + rt.ledger.update_changeset(changeset["id"], status="verdict_posted", pending_item=None) + rt.github.reviews[40].append({"id": 6, "state": "CHANGES_REQUESTED", "commit_id": head, + "user": {"login": "alice"}}) + assert merge.advance(rt, lifecycle) == [f"approval_withdrawn {changeset['id']}"] + assert any(item["body"]["pr"] == 40 for item in _items(tmp_path, "pause")) + assert rt.ledger.human_queue("demo") + # until the pause is acked it keeps its state (and so its queue slot) + assert rt.ledger.changeset(changeset["id"])["status"] == "verdict_posted" + assert merge.advance(rt, lifecycle) == [] + assert poll_external(rt) == [] # never re-staged on the remaining approvals + # the verifier sees it too, even though review 5 still reads APPROVED + problems = _verifier_passes(tmp_path, rt, origin, 40, labels=("kb:human-approved",), + reviews=tuple(rt.github.reviews[40])) + assert any("latest review no longer approves" in p for p in problems), problems + + +def test_rules_an_external_pr_retires_enter_the_retirement_ledger(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + (origin / "skills" / "x.md").write_text("no citations here\n") # else it needs a companion PR + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "drop citation") + retired = apply_operations(_tree(), [Op("retire", PAGE, "DEMO-1a", reason="upstream-removed", + evidence="PR #13")], release="v1", today="2026-09-28").files + _open_human_pr(rt, origin, 41, {f"knowledge/{rel}": text for rel, text in retired.items()}) + rt.registry["demo"] = __import__("dataclasses").replace(rt.registry["demo"], retire_ratio=1.0) + events = poll_external(rt) + assert "auto pr_open" in events[0], (events, rt.ledger.human_queue("demo")) + (changeset,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] + assert changeset["detail"]["retirements"] == [{"rule_id": "DEMO-1a", "page": PAGE}] + rt.ledger.update_changeset(changeset["id"], status="verdict_posted") + rt.github.prs[41] = {"state": "closed", "merged": True, "merge_commit_sha": "f" * 40, + "head": {"sha": changeset["head_sha"]}} + merge.advance(rt, lifecycle) + row = rt.ledger._conn.execute("SELECT rule_id, page FROM retirements WHERE repo='demo'").fetchall() + assert [tuple(r) for r in row] == [("DEMO-1a", PAGE)] + + +def test_a_withdrawn_queued_pr_holds_the_slot_until_its_pause_is_acked(tmp_path): + from test_kb_flow import _ack + from infermatrix_copilot.knowledge_service.signing import generate_private_key, load_public_key, public_key_text + + rt, lifecycle, origin = _setup(tmp_path) + publisher = generate_private_key(tmp_path / "pub.pem") + rt.publisher_public_key = load_public_key(public_key_text(publisher.public_key())) + head = _open_human_pr(rt, origin, 50, {"knowledge/tools/q.md": "q\n"}, labels=("kb:human-approved",)) + rt.github.reviews[50] = [{"id": 1, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}, + {"id": 2, "state": "APPROVED", "commit_id": head, "user": {"login": "bob"}}] + (origin / ".github" / "CODEOWNERS").write_text("/.github/ @alice @bob\n") + _git(origin, "add", "-A") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "two maintainers") + rt.github.open[50]["head"]["sha"] = head + poll_external(rt) + (changeset,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] + rt.ledger.update_changeset(changeset["id"], status="queued", pending_item=None) + rt.github.reviews[50].append({"id": 3, "state": "CHANGES_REQUESTED", "commit_id": head, + "user": {"login": "alice"}}) + assert merge.advance(rt, lifecycle) == [f"approval_withdrawn {changeset['id']}"] + assert rt.ledger.changeset(changeset["id"])["status"] == "queued" # slot still held + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset["id"], ok=False) # dequeue failed + assert rt.ledger.changeset(changeset["id"])["status"] == "queued" + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset["id"], ok=True) + assert rt.ledger.changeset(changeset["id"])["status"] == "approval_withdrawn" + assert poll_external(rt) == [] # bob's approval alone does not re-stage it + + +def test_a_withdrawal_during_an_outstanding_enqueue_keeps_the_queue_slot(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + head = _open_human_pr(rt, origin, 60, {"knowledge/tools/w.md": "w\n"}, labels=("kb:human-approved",)) + rt.github.reviews[60] = [{"id": 1, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] + poll_external(rt) + (first,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] + merge.advance(rt, lifecycle) + rt.ledger.update_changeset(first["id"], status="verdict_posted", pending_item=None) + rt.github.statuses[head] = "success" + assert merge.advance(rt, lifecycle) == [f"enqueue issued {first['id']}"] + rt.github.reviews[60].append({"id": 2, "state": "DISMISSED", "commit_id": head, "user": {"login": "alice"}}) + assert merge.advance(rt, lifecycle) == [f"approval_withdrawn {first['id']}"] + # a second, fully eligible PR must not be enqueued while the first may be queued + second = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.lease_owner, "demo", second, detail={}, status="verdict_posted", verdicts=[], + human_reason="", drafted_events=[]) + rt.ledger.update_changeset(second, pr_number=61, head_sha="6" * 40) + rt.github.prs[61] = {"state": "open", "merged": False, "head": {"sha": "6" * 40}} + rt.github.statuses["6" * 40] = "success" + assert merge.advance(rt, lifecycle) == [] + assert [i["body"]["pr"] for i in _items(tmp_path, "enqueue")] == [60] + + + +def test_a_withdrawal_completes_only_after_every_enqueue_could_have_run(tmp_path): + """The publisher may run the pause before a still-valid enqueue: the pause + is repeated once that enqueue has certainly expired.""" + from test_kb_flow import _ack + from infermatrix_copilot.knowledge_service.signing import generate_private_key, load_public_key, public_key_text + + rt, lifecycle, origin = _setup(tmp_path) + publisher = generate_private_key(tmp_path / "pub.pem") + rt.publisher_public_key = load_public_key(public_key_text(publisher.public_key())) + head = _open_human_pr(rt, origin, 70, {"knowledge/tools/v.md": "v\n"}, labels=("kb:human-approved",)) + rt.github.reviews[70] = [{"id": 1, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] + poll_external(rt) + (cs,) = [c for c in rt.ledger.changesets("demo", ("pr_open",)) if c["kind"] == "external"] + merge.advance(rt, lifecycle) + rt.ledger.update_changeset(cs["id"], status="verdict_posted", pending_item=None) + rt.github.statuses[head] = "success" + assert merge.advance(rt, lifecycle) == [f"enqueue issued {cs['id']}"] + rt.github.reviews[70].append({"id": 2, "state": "DISMISSED", "commit_id": head, "user": {"login": "alice"}}) + merge.advance(rt, lifecycle) + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=cs["id"], ok=True) # ran before the enqueue + assert rt.ledger.changeset(cs["id"])["status"] == "verdict_posted" # not complete yet + pauses = len(_items(tmp_path, "pause")) + assert merge.advance(rt, lifecycle) == [] and len(_items(tmp_path, "pause")) == pauses + start = rt.clock() + rt.clock = lambda: start + 30 * 60 + merge.CLOCK_SKEW + 1 + rt.outbox._clock = rt.clock # items are stamped by the outbox clock # the enqueue is dead now + merge.advance(rt, lifecycle) + assert len(_items(tmp_path, "pause")) == pauses + 1 + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=cs["id"], ok=True) + assert rt.ledger.changeset(cs["id"])["status"] == "approval_withdrawn" + + +def test_a_late_ack_of_an_early_pause_does_not_complete_the_withdrawal(tmp_path): + from test_kb_flow import _ack + from infermatrix_copilot.knowledge_service.signing import generate_private_key, load_public_key, public_key_text + + rt, lifecycle, origin = _setup(tmp_path) + publisher = generate_private_key(tmp_path / "pub.pem") + rt.publisher_public_key = load_public_key(public_key_text(publisher.public_key())) + head = _open_human_pr(rt, origin, 80, {"knowledge/tools/u.md": "u\n"}, labels=("kb:human-approved",)) + rt.github.reviews[80] = [{"id": 1, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] + poll_external(rt) + (cs,) = [c for c in rt.ledger.changesets("demo", ("pr_open",)) if c["kind"] == "external"] + merge.advance(rt, lifecycle) + rt.ledger.update_changeset(cs["id"], status="verdict_posted", pending_item=None) + rt.github.statuses[head] = "success" + merge.advance(rt, lifecycle) # enqueue issued + rt.github.reviews[80].append({"id": 2, "state": "DISMISSED", "commit_id": head, "user": {"login": "alice"}}) + merge.advance(rt, lifecycle) # the first pause, issued now + start = rt.clock() + rt.clock = lambda: start + 30 * 60 + merge.CLOCK_SKEW + 1 + rt.outbox._clock = rt.clock # items are stamped by the outbox clock + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=cs["id"], ok=True) # its ack arrives late + assert rt.ledger.changeset(cs["id"])["status"] == "verdict_posted" + merge.advance(rt, lifecycle) # a pause issued after the boundary + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=cs["id"], ok=True) + assert rt.ledger.changeset(cs["id"])["status"] == "approval_withdrawn" + + +def test_a_missing_enqueue_expiry_is_recovered_from_the_outbox(tmp_path): + from test_kb_flow import _ack + from infermatrix_copilot.knowledge_service.signing import generate_private_key, load_public_key, public_key_text + + rt, lifecycle, origin = _setup(tmp_path) + publisher = generate_private_key(tmp_path / "pub.pem") + rt.publisher_public_key = load_public_key(public_key_text(publisher.public_key())) + head = _open_human_pr(rt, origin, 90, {"knowledge/tools/t.md": "t\n"}, labels=("kb:human-approved",)) + rt.github.reviews[90] = [{"id": 1, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] + poll_external(rt) + (cs,) = [c for c in rt.ledger.changesets("demo", ("pr_open",)) if c["kind"] == "external"] + merge.advance(rt, lifecycle) + rt.ledger.update_changeset(cs["id"], status="verdict_posted", pending_item=None) + rt.github.statuses[head] = "success" + merge.advance(rt, lifecycle) # enqueue issued + detail = rt.ledger.changeset(cs["id"])["detail"] + detail.pop("enqueue_expires_at") # a restart lost the recorded expiry + rt.ledger.update_changeset(cs["id"], detail=detail, pending_item=None) + rt.github.reviews[90].append({"id": 2, "state": "DISMISSED", "commit_id": head, "user": {"login": "alice"}}) + merge.advance(rt, lifecycle) + assert rt.ledger.changeset(cs["id"])["detail"]["enqueue_expires_at"] > 0 # from the signed outbox item + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=cs["id"], ok=True) + assert rt.ledger.changeset(cs["id"])["status"] == "verdict_posted" # not final yet diff --git a/test/test_kb_gate_verifier.py b/test/test_kb_gate_verifier.py index cc6b9343..acbc6109 100644 --- a/test/test_kb_gate_verifier.py +++ b/test/test_kb_gate_verifier.py @@ -169,6 +169,8 @@ def get(self, path: str): parts = path.strip("/").split("/") if parts[-2:] == ["comments"] or parts[-1] == "comments": return self.comments.get(int(parts[-2]), []) + if len(parts) == 6 and parts[5] == "reviews": + return [r for (n, _id), r in sorted(self.reviews.items()) if n == int(parts[4])] if len(parts) == 7 and parts[5] == "reviews": return self.reviews[(int(parts[4]), int(parts[6]))] if parts[-1] == "pulls":