diff --git a/.github/kb-gate/MANIFEST.sha256 b/.github/kb-gate/MANIFEST.sha256 index ae9e709f..8ceb6885 100644 --- a/.github/kb-gate/MANIFEST.sha256 +++ b/.github/kb-gate/MANIFEST.sha256 @@ -1,6 +1,6 @@ 5f6e269fbbb951f9a49c40474a3d71c556ac389f5ebd5545f0ac762f85361723 README.md 77d88b9671927ab95bc6ac9ed57aaa24226423e151843381c2a3fa90cd0cafde config.json -7ae4a3ff0372e54d4a9e5efc3887c2a00e9004b138215901d4c4d8c1ae5a1588 kb_gate.py +dad76104329e3f0b23385d87700fc0aae03996c16fc0b2527b6ef51b50a6cbec 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 9a04b20a..6c60a5f5 100644 --- a/.github/kb-gate/kb_gate.py +++ b/.github/kb-gate/kb_gate.py @@ -408,13 +408,16 @@ def relevant(path: str) -> bool: def verify_change(ctx: Context, *, pr: dict, head_sha: str, pre: str, post: str, final: str, - effective_base: str, stage: str) -> list[str]: + effective_base: str, stage: str, verdict: dict | None = None, + check_holds: bool = True) -> list[str]: """Problems with one PR's knowledge change (empty when it passes). ``pre``/``post`` are the commits (or trees) around this PR's change as it is applied; ``final`` is the tree that would land; ``effective_base`` is the state the change is applied on (current main, or the first parent of - the merge-group segment).""" + the merge-group segment). The publisher's local gate passes the verdict it + was handed (its signature and binding already checked) and no hold list: + a pause there is the signed control record.""" number = int(pr["number"]) entries = ctx.git.raw_diff(pre, post) if not touches_knowledge(entries): @@ -442,7 +445,7 @@ def verify_change(ctx: Context, *, pr: dict, head_sha: str, pre: str, post: str, touched = {e["path"][len(KNOWLEDGE_PREFIX):] for e in entries if e["path"].startswith(KNOWLEDGE_PREFIX)} try: - holds = ctx.holds() + holds = ctx.holds() if check_holds else {} if holds.get("global"): problems.append("the knowledge service holds every repository (global pause)") if number in {int(p) for p in holds.get("prs") or []}: @@ -455,7 +458,9 @@ def verify_change(ctx: Context, *, pr: dict, head_sha: str, pre: str, post: str, except (GateError, SignatureError, LifecycleError, KeyError, TypeError, ValueError) as exc: problems.append(f"hold list: {exc}") - verdict, rejected = find_verdict(ctx, number, head_sha) + rejected: list[str] = [] + if verdict is None: + verdict, rejected = find_verdict(ctx, number, head_sha) if verdict is None: detail = f" (rejected: {'; '.join(rejected[:3])})" if rejected else "" problems.append(f"no valid signed verdict for PR #{number} at {head_sha[:12]}{detail}") diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 841093cb..bcd296b5 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -195,3 +195,26 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝 （不区分大小写）或别名精确匹配；歧义身份改名 `.quarantined` 隔离、从不改写；属于其他仓库的投放文件留待其处理。 邮箱是公开 issue，只接受 `KB_BUGFIX_AUTHORS` 列出的作者（未设置则不读邮箱），每个仓库各自维护评论游标。 事件 ID 与记录的 `event_id` 一致，重复投递幂等。 + +## 2026-09-29 v8：发布器本地门禁与合并 +- 服务不再发布判定评论、不再等待 GitHub `kb-gate` 状态、不再入合并队列：`pr_open` 时签发绑定 PR/head 的判定，随 + `merge` 项交给发布器（每仓库同时至多一个 `merge_requested`）。回执：成功 → `merged`（合并 SHA、落地后复核结果）； + 落地后复核失败 → 暂停该仓库并转人工；拒绝 → 按理由分为重建（上下文/页面哈希变化、无法干净合并、清单不符 → + `rebuild_needed`，由持有租约的调度器在当前 main 上重建）、瞬时（main 持续变化、控制记录变化、git/gh 问题 → 回到 + `pr_open` 重签）、变更本身的问题（→ `gate_failed`，转人工，从不按原样重试）。未回执的 `merge` 过期后回到 `pr_open`。 + v7 的 `verdict_posted`/`queued` 记录按 `pr_open` 重签。 +- 落地后复核：回执中的 `passed`/`failed` 为终态（`failed` 暂停该仓库并转人工），即使 GitHub 先显示已合并、回执后到也照常处理； + `unknown`（发布器无法运行，或崩溃后恢复的合并）由服务 `verify_merged` 在自己的克隆上重跑，无法运行则下一轮重试。 +- `local_gate`：发布器在自己安装的 Copilot 版本上，对 `main` tip 与 PR head 的合并树运行 `gate_verifier.verify_change` + （直接传入已验签的判定、不读暂停清单），另要求每个路径都在该项所属仓库的知识目录内、判定来源为 `auto`，并检查合并树 + 无新增树级问题；`post_merge_problems` 在 GitHub 实际生成的合并提交上重跑。 +- 发布器 `merge`：整轮持有跨进程 `flock`（第二个实例直接退出）；验签与绑定 → PR 打开且目标为 `main`、head 一致 → + 取 main 与 PR head 跑本地门禁 → main 变化则重来（至多 3 次）→ 重读控制记录（暂停/代际变化即放弃）→ 写合并意图 → + `gh pr merge --merge --match-head-commit` → 读取合并提交并做落地后复核。合并之后的任何异常都不会变成失败回执； + 每轮开始先恢复有意图无完成记录的项（已合并则补发回执，未合并则丢弃意图、下一轮重试）。 + `gh pr merge` 报错时无法确定是否已合并：保留意图、不写记录也不回执，由下一轮恢复向 GitHub 确认。遗留的 `kb:hold` + 标签（v7 暂停留下）在合并前移除并确认。 + 只有 GitHub 显示 `MERGED` 才算合并：若 `main` 要求合并队列或开启了自动合并，`gh pr merge` 只会入队——此时关闭自动合并并 + 拒绝（转人工，要求 `main` 允许发布器直接合并）；只有经 GraphQL 确认既不在合并队列中（必要时 `dequeuePullRequest`）也没有自动合并，才记为拒绝，否则保留意图，由恢复流程继续取消或接收 + 后来发生的合并。存在未结的合并意图时该项跳过，直到恢复流程从 GitHub 得到结果。 +- external 的状态集合由 `merge.IN_FLIGHT` 派生，合并中的外部 PR 不会被重复评审。 diff --git a/doc/architecture/SPEC/knowledge_service.md b/doc/architecture/SPEC/knowledge_service.md index 83325997..e3227ffa 100644 --- a/doc/architecture/SPEC/knowledge_service.md +++ b/doc/architecture/SPEC/knowledge_service.md @@ -1,6 +1,6 @@ # knowledge_service/ — provider curation components - + The provider owns knowledge curation beneath the public SDK v1 facade. `KnowledgeCurator` composes four domain components over one explicit work @@ -74,3 +74,7 @@ L1 的索引检查允许为目录中已存在但未列入索引的页面补链 ## 2026-09-28 human-approved：最新审阅 验证器除核对签名绑定的审阅仍为 APPROVED 外，还读取该 PR 的全部审阅：每位被引用审阅者的**最新**审阅必须仍批准当前 head； 之后的 changes requested 或 dismiss 使该审批失效（即使被绑定的那条审阅仍显示 APPROVED）。 + +## 2026-09-29 verify_change 供本地门禁调用 +`verify_change` 新增 `verdict=`（调用方已验签与绑定的判定，跳过从 PR 评论查找）与 `check_holds=`（本地门禁不读暂停清单， +暂停由签名控制记录表达）。其余检查不变。 diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md index 99518497..760ecc90 100644 --- a/doc/features/knowledge-service.md +++ b/doc/features/knowledge-service.md @@ -61,8 +61,9 @@ infermatrix-copilot kb activate # 立即激活知识仓库 main 的 infermatrix-copilot kb rollback --to # 回指到较早的快照（下一请求生效） ``` -合并流程：`open_pr` 回执 → 为该 PR 的精确 head 签发判定并发布评论 → PR 阶段 `kb-gate` -通过后入合并队列（每仓库至多 1 个）→ 合并后激活新快照。发版巡检在上游新 release 后运行 +合并流程（v8）：`open_pr` 回执 → 为该 PR 的精确 head 签发判定，随 `merge` 项交给 GPU 盒发布器 → 发布器在 +`main` 与 PR head 的合并树上运行本地门禁，通过才 `gh pr merge`（每仓库同时至多 1 个）→ 合并后激活新快照。 +本地门禁拒绝的变更从不合并：上下文变化则在当前 main 上重建，变更本身的问题转人工。发版巡检在上游新 release 后运行 T1/T2/T3 与 purge，每个规则页一个变更集，全部经过质量门。 评审服务读取知识：将 `KNOWLEDGE_ROOT` 指向 `$KB_STATE_DIR/active`。 diff --git a/src/infermatrix_copilot/kb_service/external.py b/src/infermatrix_copilot/kb_service/external.py index 82b5e0dd..27218d86 100644 --- a/src/infermatrix_copilot/kb_service/external.py +++ b/src/infermatrix_copilot/kb_service/external.py @@ -30,7 +30,7 @@ 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 .merge import IN_FLIGHT, knowledge_repository from .runtime import calibration_current KIND = "external" @@ -148,9 +148,12 @@ def _ours(rt, number: int) -> bool: 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") +# every status an external change set can have; the in-flight ones come from +# the merge flow itself, so a new merge state can never be judged twice +_ALL = tuple(dict.fromkeys(( + *IN_FLIGHT, "gated", "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]: diff --git a/src/infermatrix_copilot/kb_service/local_gate.py b/src/infermatrix_copilot/kb_service/local_gate.py new file mode 100644 index 00000000..a93be9c4 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/local_gate.py @@ -0,0 +1,107 @@ +"""The publisher's local gate: the last check before a knowledge PR is merged. + +The publisher is the only thing that merges knowledge PRs (design v8). Right +before each merge it verifies, on its own machine and with the Copilot version +it has installed (never code from the PR or from main), the exact tree that +would land: the current tip of ``main`` merged with the PR head. + + verdict signature + binding (repository, PR, head, issue window) + -> every changed path is a governed page of the item's own repository + -> the change as applied equals the signed manifest (base independent) + -> L1 on the change and on the merged tree, block coverage, consistency + hashes, context-base check (``gate_verifier.verify_change``) + -> no new tree-level issue on the merged tree + +The same checks run once more on the merge commit GitHub actually made +(``post_merge_problems``): ``main`` can move in the seconds between the check +and the merge. +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from ..knowledge_service.gate_verifier import ( + Context, GateError, Git, touches_knowledge, tree_problems, verify_change, +) +from ..knowledge_service.ops import repo_scope +from ..knowledge_service.signing import SignatureError, verify +from ..knowledge_service.verdict import VerdictError, check_binding + +GOVERNED_PREFIX = "knowledge/" + + +class LocalGateError(RuntimeError): + """The local gate could not run (git or GitHub trouble); retried later.""" + + +def _scope_problems(entries: list[dict], repo: str) -> list[str]: + """Every changed path must be inside the item's own repository scope.""" + problems = [] + for entry in entries: + path = entry["path"] + if not path.startswith(GOVERNED_PREFIX): + problems.append(f"{path}: outside knowledge/ (split the PR)") + continue + try: + scope = repo_scope(path[len(GOVERNED_PREFIX):]) + except ValueError: + problems.append(f"{path}: not a governed knowledge page") + continue + name = scope.split("/")[-1] + if name != repo: + problems.append(f"{path}: belongs to {name}, not to {repo} (one repository per PR)") + return problems + + +def check_verdict(envelope: Any, public_key, *, repository: str, pr: int, head_sha: str, + now: float) -> dict: + try: + verdict = verify("kb-gate-verdict", envelope, public_key) + check_binding(verdict, repository=repository, pr=pr, head_sha=head_sha, now=now) + except (SignatureError, VerdictError, KeyError, TypeError, ValueError) as exc: + raise VerdictError(f"verdict: {exc}") from exc + if verdict.get("source") != "auto": + raise VerdictError("only auto verdicts merge (there is no human-approved path)") + return verdict + + +def gate(clone: Path, *, repository: str, repo: str, pr: dict, head_sha: str, main_sha: str, + verdict: dict, public_key, now: float) -> list[str]: + """Problems with merging ``head_sha`` into ``main_sha`` (empty = pass).""" + git = Git(clone) + try: + base = git.merge_base(main_sha, head_sha) + entries = git.raw_diff(base, head_sha) + if not touches_knowledge(entries): + return ["the PR changes nothing under knowledge/"] + problems = _scope_problems(entries, repo) + final = git.merge_tree(main_sha, head_sha) + except GateError as exc: + return [str(exc)] + ctx = Context(git=git, github=None, repository=repository, public_key=public_key, + holds_loader=lambda: {}, codeowners=[], now=now) + problems += verify_change(ctx, pr=pr, head_sha=head_sha, pre=base, post=head_sha, final=final, + effective_base=main_sha, stage="local", verdict=verdict, check_holds=False) + problems += tree_problems(git, main_sha, final) + return problems + + +def post_merge_problems(clone: Path, *, repository: str, pr: dict, head_sha: str, merge_sha: str, + verdict: dict, public_key, now: float) -> list[str]: + """The same checks on the merge commit that landed, against its first parent.""" + git = Git(clone) + try: + parents = git.parents(merge_sha) + except GateError as exc: # not fetched yet, or git trouble: the check could not run + raise LocalGateError(str(exc)) from exc + if len(parents) != 2 or parents[1] != head_sha: + return [f"merge commit {merge_sha[:12]} is not a merge of the verified head"] + first = parents[0] + ctx = Context(git=git, github=None, repository=repository, public_key=public_key, + holds_loader=lambda: {}, codeowners=[], now=now) + landed = {**pr, "state": "open"} # it was open when it merged + problems = verify_change(ctx, pr=landed, head_sha=head_sha, pre=first, post=merge_sha, final=merge_sha, + effective_base=first, stage="local", verdict=verdict, check_holds=False) + return problems + tree_problems(git, first, merge_sha) diff --git a/src/infermatrix_copilot/kb_service/merge.py b/src/infermatrix_copilot/kb_service/merge.py index d9b75fcb..8f8b997e 100644 --- a/src/infermatrix_copilot/kb_service/merge.py +++ b/src/infermatrix_copilot/kb_service/merge.py @@ -1,15 +1,20 @@ -"""The merge flow after a change set is handed to the publisher. +"""The merge flow after a change set is handed to the publisher (design v8). - gated ─publish→ pr_requested ─ack(pr, head)→ pr_open ─sign→ verdict_posted - ─PR-stage kb-gate success→ queued ─merged→ merged ─activation→ activated + gated ─publish→ pr_requested ─ack(pr, head)→ pr_open ─sign→ merge_requested + ─ack(merged, merge sha)→ merged ─activation→ activated -Anything off the happy path goes to people or stops: a rejected publisher -action (``failed``), a head that no longer matches what was signed -(``head_changed`` → human), a PR closed unmerged (``closed``), a paused -repository (``paused``; the pause itself is performed by the publisher through -an always-executable ``pause`` item). At most one knowledge PR per repository -is in the merge queue at a time, which bounds what can land while the -publisher is unavailable. +The service signs a verdict for the PR's exact head and hands it to the +publisher in a ``merge`` item; the publisher, the only thing that merges +knowledge PRs, runs the gate locally on the exact tree that would land and +merges only if it passes. A refused merge is never retried as is: context that +moved or a PR that no longer applies is rebuilt on current main, a real +problem with the change goes to people, and a transient failure (main moving, +GitHub or git trouble) is signed again next pass. At most one merge per +repository is in flight. + +Anything else off the happy path goes to people or stops: a head that no +longer matches what was signed (``head_changed`` → human), a PR closed +unmerged (``closed``), a paused repository (no ``merge`` is issued). The service only READS GitHub; every write is an outbox item. """ @@ -24,7 +29,8 @@ from ..knowledge_service.verdict import build_verdict, manifest_from_files VERDICT_MARKER = "" -IN_FLIGHT = ("pr_requested", "pr_open", "verdict_posted", "queued") +# verdict_posted/queued: v7 records (merge queue), signed again as pr_open +IN_FLIGHT = ("pr_requested", "pr_open", "verdict_posted", "queued", "merge_requested", "rebuild_needed") # 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 @@ -42,6 +48,9 @@ RESIGN_GATE = ("outside its issue window", "no valid signed verdict") REBUILD_GATE = ("context changed since the verdict", "pages changed since the consistency judgement", "does not merge cleanly", "signed patch manifest does not match") +# a refused local-gate merge that is about the moment, not the change: sign again +TRANSIENT_MERGE = ("main kept moving", "not merging:", "head of PR", "head moved", "git ", "gh ", + "did not merge", "is not open", "verdict: ") KNOWLEDGE_REPO_ENV = "KB_KNOWLEDGE_REPOSITORY" DEFAULT_KNOWLEDGE_REPOSITORY = "JiusiServe/InferMatrixCopilot" @@ -56,6 +65,7 @@ def knowledge_repository() -> str: "open_companion_pr": ("pr_requested", "companion_open"), "post_verdict": ("pr_open", "verdict_posted"), "enqueue": ("verdict_posted", "queued"), + "merge": ("merge_requested", "merged"), } @@ -125,11 +135,21 @@ def apply_acks(rt, acks: list[dict]) -> None: rt.ledger.update_changeset(changeset_id, status="pr_open", pending_item=None) continue pending = changeset.get("pending_item") or {} + if kind == "merge" and ack.get("ok") and changeset["status"] == "merged" \ + and ack.get("merge_sha") and ack.get("merge_sha") == changeset.get("merge_sha"): + # GitHub showed the merge before its receipt arrived: the receipt's + # post-merge check still counts + _record_post_check(rt, rt.ledger.changeset(changeset_id), ack.get("post_check", ""), + ack.get("problems") or []) + continue if kind not in TRANSITIONS or pending.get("id") != ack.get("item_id"): continue before, after = TRANSITIONS[kind] if changeset["status"] != before: continue + if kind == "merge": + _apply_merge_ack(rt, changeset, ack) + continue if not ack.get("ok"): rt.ledger.update_changeset(changeset_id, status="failed", pending_item=None) rt.ledger.enqueue_human(changeset["repo"], f"publisher refused {kind}: {ack.get('error', '')}", @@ -144,6 +164,91 @@ def apply_acks(rt, acks: list[dict]) -> None: rt.ledger.update_changeset(changeset_id, **fields) +def _pause_repo(rt, repo: str, reason: str) -> None: + """Stop automatic merging for ``repo`` (a new generation voids every item).""" + public = {name for name, lc in rt.registry.items() if lc.publishes} + rt.outbox.transition(lambda: rt.ledger.bump_generation(repo, pause=True, reason=reason), public_repos=public) + + +def _record_post_check(rt, changeset: dict, result: str, problems: list) -> None: + """``passed`` / ``failed`` are final; anything else (``unknown``: the + publisher could not run it, or recovered the merge after a crash) is run + by the service itself on the next pass (``verify_merged``).""" + if changeset["detail"].get("post_check") in ("passed", "failed"): + return + rt.ledger.update_changeset(changeset["id"], detail={**changeset["detail"], "post_check": result or "unknown"}) + if result == "failed": + # main moved in the seconds before the merge and what landed does not + # pass: stop merging this repository; people decide on a revert + reason = (f"post-merge check failed on {str(changeset.get('merge_sha') or '')[:12]} " + f"(PR #{changeset['pr_number']}): " + "; ".join(map(str, problems))[:400]) + _pause_repo(rt, changeset["repo"], reason) + rt.ledger.enqueue_human(changeset["repo"], reason, changeset["id"]) + + +def verify_merged(rt, lifecycle) -> list[str]: + """Run the post-merge check for merges whose result is not known yet, on + the service's own clone. A check that cannot run is simply retried.""" + from pathlib import Path + + from . import local_gate + + events = [] + for changeset in rt.ledger.changesets(lifecycle.repo, ("merged",)): + detail = changeset["detail"] + if not detail.get("verdict_issued_at") or detail.get("post_check") in ("passed", "failed"): + continue # not a v8 merge, or already settled + merge_sha = str(changeset.get("merge_sha") or "") + if not merge_sha: + try: + merge_sha = str(_pr_state(rt, int(changeset["pr_number"])).get("merge_commit_sha") or "") + except Exception: # noqa: BLE001 - GitHub trouble: next pass + continue + if not merge_sha: + continue + rt.ledger.update_changeset(changeset["id"], merge_sha=merge_sha) + changeset = rt.ledger.changeset(changeset["id"]) + try: + rt.knowledge.fetch() + verdict = sign_verdict(rt, changeset)["payload"] + problems = local_gate.post_merge_problems( + Path(rt.knowledge.path), repository=knowledge_repository(), + pr={"number": int(changeset["pr_number"]), "state": "open", + "head": {"sha": changeset["head_sha"]}, "labels": []}, + head_sha=changeset["head_sha"], merge_sha=merge_sha, verdict=verdict, + public_key=rt.outbox._key.public_key(), now=rt.clock()) + except Exception as exc: # noqa: BLE001 - e.g. the clone cannot fetch: next pass + events.append(f"post_check_retry {changeset['id']}: {exc}") + continue + _record_post_check(rt, changeset, "failed" if problems else "passed", problems) + events.append(f"post_check {'failed' if problems else 'passed'} {changeset['id']}") + return events + + +def _apply_merge_ack(rt, changeset: dict, ack: dict) -> None: + changeset_id, repo = changeset["id"], changeset["repo"] + if ack.get("ok"): + merge_sha = str(ack.get("merge_sha") or "") + rt.ledger.update_changeset(changeset_id, status="merged", merge_sha=merge_sha or None, pending_item=None) + record_retirements(rt, changeset) + _outcome(rt, changeset, "merged", merge_sha=merge_sha) + _record_post_check(rt, rt.ledger.changeset(changeset_id), ack.get("post_check", ""), + ack.get("problems") or []) + return + text = " ".join([str(ack.get("error") or ""), *map(str, ack.get("problems") or [])]) + if any(m in text for m in REBUILD_GATE): + rt.ledger.update_changeset(changeset_id, status="rebuild_needed", pending_item=None, + detail={**changeset["detail"], "rebuild_because": text[:500]}) + elif any(m in text for m in TRANSIENT_MERGE) and not ack.get("problems"): + rt.ledger.update_changeset(changeset_id, status="pr_open", pending_item=None) + else: # the local gate refused the change itself: it is not merged + rt.ledger.update_changeset(changeset_id, status="gate_failed", pending_item=None, + detail={**changeset["detail"], "gate_problems": (ack.get("problems") or [])[:50]}) + rt.ledger.enqueue_human(repo, f"local gate refused PR #{changeset['pr_number']}: {text[:400]}", + changeset_id) + _outcome(rt, changeset, "gate_failed", problems=(ack.get("problems") or [])[:20]) + + def sign_verdict(rt, changeset: dict) -> dict: """The signed kb-gate verdict for this change set's PR at its recorded head.""" detail = changeset["detail"] @@ -329,12 +434,7 @@ 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, 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) or cs["detail"].get("withdrawn")] + merging = rt.ledger.changesets(lifecycle.repo, ("merge_requested",)) for changeset in rt.ledger.changesets(lifecycle.repo, (*IN_FLIGHT, "superseding", "paused", "companion_open")): number = changeset["pr_number"] @@ -399,7 +499,7 @@ def advance(rt, lifecycle) -> list[str]: record_retirements(rt, changeset) _outcome(rt, changeset, "merged", merge_sha=str(pr.get("merge_commit_sha") or "")) events.append(f"merged {changeset['id']}") - continue + continue # its post-merge check: the receipt, or verify_merged if pr.get("state") == "closed": rt.ledger.update_changeset(changeset["id"], status="closed") _outcome(rt, changeset, "closed_unmerged") # overturned by people: a negative label @@ -464,70 +564,37 @@ def advance(rt, lifecycle) -> list[str]: "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) - if changeset["status"] == "queued" and ( - stale_verdict or (queued_at and rt.clock() - queued_at > QUEUE_STALL)): - # never free the queue slot on a timer: dequeue (+ draft) first; the - # pause ack returns it to pr_open, then it is re-signed and re-queued - if issue_once(rt, lifecycle.repo, changeset, "pause", { - "changeset_id": changeset["id"], "pr": int(number), - "reason": "re-signing: the verdict is ageing or the PR stalled in the merge queue"}): - events.append(f"requeue {changeset['id']}") + if changeset["status"] == "merge_requested": + pending = changeset.get("pending_item") or {} + if pending.get("kind") == "merge" and float(pending.get("expires_at") or 0) + CLOCK_SKEW < rt.clock(): + # expired unanswered: the publisher never ran it (or its record + # is lost); sign again. A merge it did run still shows as merged. + rt.ledger.update_changeset(changeset["id"], status="pr_open", pending_item=None) + events.append(f"merge_expired {changeset['id']}") continue - if changeset["status"] == "verdict_posted" and _enqueue_outstanding(rt, changeset): - continue # its ack decides: queued (then the queued path applies) or refused - if changeset["status"] == "verdict_posted" and stale_verdict: - _resign(rt, changeset, "verdict close to its issue window") - events.append(f"resign {changeset['id']}") + if changeset["status"] == "rebuild_needed": + if 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 getattr(rt, "lease_owner", None): # only the scheduler (holding the lease) rebuilds + new_id = rebuild(rt, lifecycle, changeset, changeset["detail"].get("rebuild_because", "")) + events.append(f"rebuilt {changeset['id']} as {new_id}" if new_id + else f"rebuild_failed {changeset['id']}") continue - if changeset["status"] == "verdict_posted": - state, description = _gate(rt, changeset["head_sha"], since=signed_at) - if state == "failure" and not any(m in description for m in TRANSIENT_GATE): - 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 - new_id = rebuild(rt, lifecycle, changeset, description) - events.append(f"rebuilt {changeset['id']} as {new_id}" if new_id - else f"rebuild_failed {changeset['id']}") - else: # a real problem with the change itself: people decide - rt.ledger.update_changeset(changeset["id"], status="gate_failed") - rt.ledger.enqueue_human(lifecycle.repo, f"kb-gate failed: {description}", changeset["id"]) - events.append(f"gate_failed {changeset['id']}") - continue - if changeset["status"] == "pr_open": - pending = changeset.get("pending_item") - if pending and pending.get("kind") == "post_verdict" and float(pending["expires_at"]) > rt.clock(): - continue + if changeset["status"] in ("pr_open", "verdict_posted", "queued"): + if merging: + continue # one merge in flight per repository envelope = sign_verdict(rt, changeset) - if issue_once(rt, lifecycle.repo, changeset, "post_verdict", { - "changeset_id": changeset["id"], "pr": int(number), "head_sha": changeset["head_sha"], - "comment": f"{VERDICT_MARKER}\n```json\n{json.dumps(envelope, ensure_ascii=False)}\n```\n", - }): + if issue_once(rt, lifecycle.repo, changeset, "merge", { + "changeset_id": changeset["id"], "pr": int(number), "head_sha": changeset["head_sha"], + "verdict": envelope}): current = rt.ledger.changeset(changeset["id"]) - rt.ledger.update_changeset(changeset["id"], detail={ + rt.ledger.update_changeset(changeset["id"], status="merge_requested", detail={ **current["detail"], "verdict_issued_at": envelope["payload"]["issued_at"]}) - events.append(f"verdict issued {changeset['id']}") - elif changeset["status"] == "verdict_posted" and not queued: - 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 + merging = [changeset] + events.append(f"merge issued {changeset['id']}") + return events + verify_merged(rt, lifecycle) def _rule_ids(changeset: dict) -> list[str]: diff --git a/src/infermatrix_copilot/kb_service/outbox.py b/src/infermatrix_copilot/kb_service/outbox.py index 5e5be17a..88c32482 100644 --- a/src/infermatrix_copilot/kb_service/outbox.py +++ b/src/infermatrix_copilot/kb_service/outbox.py @@ -37,12 +37,12 @@ ITEM_KINDS = ( "open_pr", "open_companion_pr", "post_verdict", "enqueue", - "update_branch", "pause", "close", "open_issue", + "update_branch", "pause", "close", "open_issue", "merge", ) ALWAYS_EXECUTABLE = frozenset({"pause", "close"}) ITEM_TTL = { "open_pr": 24 * 3600, "open_companion_pr": 24 * 3600, - "post_verdict": 30 * 60, "enqueue": 30 * 60, + "post_verdict": 30 * 60, "enqueue": 30 * 60, "merge": 30 * 60, "update_branch": 24 * 3600, "pause": 24 * 3600, "close": 24 * 3600, "open_issue": 7 * 24 * 3600, } diff --git a/src/infermatrix_copilot/kb_service/publisher.py b/src/infermatrix_copilot/kb_service/publisher.py index c3304761..b2f51240 100644 --- a/src/infermatrix_copilot/kb_service/publisher.py +++ b/src/infermatrix_copilot/kb_service/publisher.py @@ -29,6 +29,7 @@ from __future__ import annotations +import fcntl import json import os import re @@ -60,6 +61,22 @@ RETRIED = frozenset({"pause", "close", "open_issue"}) +class MergeUncertain(RuntimeError): + """`gh pr merge` failed in a way that does not say whether GitHub merged: + the intent stays and the next round's recovery asks GitHub (no receipt now).""" + + +class GateRefused(RuntimeError): + """The local gate found problems: the PR is not merged (the service refines).""" + + def __init__(self, problems: list[str]): + super().__init__("local gate: " + "; ".join(problems)) + self.problems = problems + + +MAIN_RETRIES = 3 # main moved between the check and the merge: check again at most this often + + class PublishError(RuntimeError): """An item could not be performed; it is acked as failed with this reason.""" @@ -162,7 +179,8 @@ def gh(self, *args: str, input: str | None = None, ok_fail: bool = False) -> str def pr(self, number: int) -> dict: return json.loads(self.gh("pr", "view", str(number), "--repo", self.repository, - "--json", "number,state,isDraft,headRefOid,headRefName,id,labels")) + "--json", "number,state,isDraft,headRefOid,headRefName,baseRefName,id," + "labels,mergeCommit,autoMergeRequest")) # -- the publisher ------------------------------------------------------------------- @@ -240,7 +258,26 @@ def sync_archives(self) -> dict: return counts def run_once(self) -> dict: + """One round under a cross-process lock: a second publisher (a timer + firing while a manual run is going) does nothing at all.""" summary = {"performed": 0, "failed": 0, "resent": 0, "undelivered": 0, "dry_run": 0, "skipped": 0} + self.state_dir.mkdir(parents=True, exist_ok=True) + fd = os.open(self.state_dir / "publisher.lock", os.O_RDWR | os.O_CREAT, 0o600) + try: + try: + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + self._trace(event="locked", detail="another publisher is running") + return {**summary, "locked": 1} + try: + self._recover_merges() + return self._round(summary) + finally: + fcntl.flock(fd, fcntl.LOCK_UN) + finally: + os.close(fd) + + def _round(self, summary: dict) -> dict: try: control = json.loads(self.transport.read("outbox/control.json")) # a quick look before verifying (check_item verifies it for real) @@ -272,6 +309,12 @@ def run_once(self) -> dict: if done.with_suffix(".corrupt").exists(): summary["skipped"] += 1 # waits for a person to decide (see the trace) continue + if self._intent_path(item_id).is_file(): + # a merge whose outcome recovery could not settle yet: never + # perform it again until GitHub has said what happened + self._trace(event="awaiting_recovery", item=item_id) + summary["skipped"] += 1 + continue summary[self._handle(item_id, control)] += 1 return summary @@ -290,6 +333,48 @@ def _record_done(self, item_id: str, ack: dict) -> None: Path(tmp).unlink(missing_ok=True) raise + def _intent_path(self, item_id: str) -> Path: + return self.state_dir / "intents" / f"{item_id}.json" + + def _recover_merges(self) -> None: + """A merge intent without a completion record: the publisher stopped + between asking GitHub to merge and recording it. Ask GitHub what + happened and finish the record, so the service gets its receipt.""" + folder = self.state_dir / "intents" + if not folder.is_dir(): + return + for path in sorted(folder.glob("*.json")): + item_id = path.stem + if self._done_path(item_id).is_file(): + path.unlink(missing_ok=True) + continue + try: + intent = json.loads(path.read_text(encoding="utf-8")) + pr = self.github.pr(int(intent["pr"])) + except (PublishError, OSError, ValueError, KeyError, TypeError) as exc: + self._trace(event="recovery_pending", item=item_id, error=str(exc)) + continue + merged = str((pr.get("mergeCommit") or {}).get("oid") or "") + ack = {"item_id": item_id, "kind": "merge", "changeset_id": intent["changeset_id"], + "pr": int(intent["pr"]), "head_sha": intent["head_sha"], "branch": "", "error": "", + "merge_sha": "", "problems": [], "ok": True, "recovered": True, + "post_check": "unknown"} + if pr.get("state") == "MERGED" and merged and pr.get("headRefOid") == intent["head_sha"]: + ack["merge_sha"] = merged + elif pr.get("state") == "OPEN": + if not self._cancel_pending_merge(int(intent["pr"])): + self._trace(event="recovery_pending", item=item_id, error="still queued or auto-merging") + continue # GitHub may still merge it: keep the intent + path.unlink() # the merge did not happen: the item is simply tried again + self._trace(event="intent_dropped", item=item_id) + continue + else: + ack.update(ok=False, error=f"PR #{intent['pr']} is {pr.get('state')} (not merged by us)") + self._record_done(item_id, ack) + path.unlink(missing_ok=True) + self._trace(event="merge_recovered", item=item_id, ack=ack) + self._send_ack(ack) + def _handle(self, item_id: str, control: Any) -> str: try: envelope = json.loads(self.transport.read(f"outbox/{item_id}.json")) @@ -298,6 +383,7 @@ def _handle(self, item_id: str, control: Any) -> str: publishes, auto_merge = self.repo_flags.get(claimed, (False, False)) item = check_item(envelope, control, self.service_public_key, now=self.clock(), repo_publishes=publishes, repo_auto_merge=auto_merge) + self._envelope, self._flags = envelope, (publishes, auto_merge) except (OutboxError, OSError, ValueError, KeyError, TypeError) as exc: # not performed and not acked: an unverifiable item is never # answered, and a stale one is superseded by the service itself @@ -311,6 +397,11 @@ def _handle(self, item_id: str, control: Any) -> str: "ok": True, "pr": None, "head_sha": "", "branch": "", "error": ""} try: ack.update(getattr(self, f"_do_{item.kind}")(item)) + except MergeUncertain as exc: + self._trace(event="merge_uncertain", item=item.id, repo=item.repo, error=str(exc)) + return "failed" # no record, no ack: recovery settles it from GitHub + except GateRefused as exc: + ack.update(ok=False, error=str(exc)[:500], problems=exc.problems[:50]) except (PublishError, OSError, ValueError, KeyError, TypeError) as exc: ack.update(ok=False, error=str(exc)[:500]) if not ack["ok"] and item.kind in RETRIED: @@ -319,6 +410,7 @@ def _handle(self, item_id: str, control: Any) -> str: self._trace(event="retry", item=item.id, kind=item.kind, repo=item.repo, error=ack["error"]) return "failed" self._record_done(item.id, ack) + self._intent_path(item.id).unlink(missing_ok=True) # the completion record supersedes it self._trace(event="performed" if ack["ok"] else "failed", item=item.id, kind=item.kind, repo=item.repo, ack=ack) self._send_ack(ack) @@ -451,6 +543,134 @@ def _do_open_issue(self, item: OutboxItem) -> dict: number = out.strip().rsplit("/", 1)[-1] return {"pr": int(number) if number.isdigit() else None} + def _recheck_control(self) -> None: + """Right before merging: a pause, rollback or mode change issued while + the gate ran voids the item (the control record is re-read).""" + try: + control = json.loads(self.transport.read("outbox/control.json")) + publishes, auto_merge = self._flags + check_item(self._envelope, control, self.service_public_key, now=self.clock(), + repo_publishes=publishes, repo_auto_merge=auto_merge) + except (OutboxError, OSError, ValueError, KeyError, TypeError) as exc: + raise PublishError(f"not merging: {exc}") from exc + + def _do_merge(self, item: OutboxItem) -> dict: + """The local gate on the exact merged tree, then the merge (design v8 §8.2).""" + from . import local_gate + + body = item.body + number, head = int(body["pr"]), str(body["head_sha"]) + verdict = local_gate.check_verdict(body["verdict"], self.service_public_key, + repository=self.github.repository, pr=number, head_sha=head, + now=self.clock()) + pr = self.github.pr(number) + if pr.get("state") != "OPEN": + raise PublishError(f"PR #{number} is {pr.get('state')}") + if pr.get("baseRefName") != "main": + raise GateRefused([f"PR #{number} targets {pr.get('baseRefName')}, not main"]) + if pr.get("headRefOid") != head: + raise PublishError(f"PR #{number} head moved to {str(pr.get('headRefOid'))[:12]}") + self._refuse_companion(pr) + labels = {str(label.get("name")) for label in pr.get("labels") or []} + if HOLD_LABEL in labels: + # a v7 pause left it: a merge item exists only for an unpaused + # repository under its current generation, so the hold is over + self.github.gh("pr", "edit", str(number), "--repo", self.github.repository, + "--remove-label", HOLD_LABEL, ok_fail=True) + pr = self.github.pr(number) + if HOLD_LABEL in {str(label.get("name")) for label in pr.get("labels") or []}: + raise PublishError(f"PR #{number} still carries {HOLD_LABEL}; retried next round") + remote = self._ensure_clone() + rest = {"number": number, "state": "open", "head": {"sha": head}, "draft": bool(pr.get("isDraft")), + "labels": pr.get("labels") or []} + for _attempt in range(MAIN_RETRIES): + self._git("fetch", "--quiet", remote, "+refs/heads/main:refs/remotes/origin/main", + f"+refs/pull/{number}/head:refs/kb/pr-{number}", auth=True) + if self._git("rev-parse", f"refs/kb/pr-{number}") != head: + raise PublishError(f"fetched head of PR #{number} is not {head[:12]}") + main_sha = self._git("rev-parse", "refs/remotes/origin/main") + problems = local_gate.gate(self.clone, repository=self.github.repository, repo=item.repo, pr=rest, + head_sha=head, main_sha=main_sha, verdict=verdict, + public_key=self.service_public_key, now=self.clock()) + if problems: + raise GateRefused(problems) + self._git("fetch", "--quiet", remote, "+refs/heads/main:refs/remotes/origin/main", auth=True) + if self._git("rev-parse", "refs/remotes/origin/main") == main_sha: + break + else: + raise PublishError(f"main kept moving while PR #{number} was checked; retried next round") + self._recheck_control() + if pr.get("isDraft"): # a v7 pause left it a draft; GitHub will not merge a draft + self.github.gh("pr", "ready", str(number), "--repo", self.github.repository) + intent = self._intent_path(item.id) + intent.parent.mkdir(parents=True, exist_ok=True) + intent.write_text(json.dumps({"pr": number, "head_sha": head, "changeset_id": body["changeset_id"], + "main_sha": main_sha, "at": self.clock()}), encoding="utf-8") + try: + self.github.gh("pr", "merge", str(number), "--repo", self.github.repository, "--merge", + "--match-head-commit", head) + except PublishError as exc: + raise MergeUncertain(f"gh pr merge #{number}: {exc}") from exc + # only a PR GitHub shows as MERGED is a merge: with a merge queue or + # auto-merge on main, `gh pr merge` succeeds by queueing instead + try: + merged = self.github.pr(number) + except (PublishError, OSError, ValueError, KeyError, TypeError) as exc: + raise MergeUncertain(f"PR #{number} state after merging: {exc}") from exc + if merged.get("state") == "OPEN": + if not self._cancel_pending_merge(number): + # GitHub could still merge it later: keep the intent, so recovery + # keeps trying to cancel, or picks up the merge if it happens + raise MergeUncertain(f"PR #{number} was queued for auto-merge and could not be cancelled yet") + raise GateRefused([f"PR #{number} was queued or set to auto-merge, not merged: main must allow " + "a direct merge by the publisher (no merge queue); auto-merge was disabled"]) + if merged.get("state") != "MERGED": + raise PublishError(f"PR #{number} is {merged.get('state')} after merging") + # merged from here on: nothing below may turn this into a failure ack + try: + merge_sha = str((merged.get("mergeCommit") or {}).get("oid") or "") + if not merge_sha: + return {"pr": number, "head_sha": head, "merge_sha": "", "post_check": "unknown", + "problems": ["merged, but GitHub has not reported the merge commit yet"]} + self._git("fetch", "--quiet", remote, "+refs/heads/main:refs/remotes/origin/main", auth=True) + post = local_gate.post_merge_problems(self.clone, repository=self.github.repository, pr=rest, + head_sha=head, merge_sha=merge_sha, verdict=verdict, + public_key=self.service_public_key, now=self.clock()) + except (PublishError, local_gate.LocalGateError, OSError, ValueError, KeyError, TypeError) as exc: + return {"pr": number, "head_sha": head, "merge_sha": "", "post_check": "unknown", + "problems": [f"merged, but the post-merge check could not run: {exc}"]} + return {"pr": number, "head_sha": head, "merge_sha": merge_sha, + "post_check": "failed" if post else "passed", "problems": post[:50]} + + _PENDING_QUERY = ("query=query($owner:String!,$name:String!,$number:Int!){repository(owner:$owner,name:$name)" + "{pullRequest(number:$number){id state mergeQueueEntry{id} autoMergeRequest{enabledAt}}}}") + + def _pending_merge(self, number: int) -> dict: + owner, name = self.github.repository.split("/", 1) + data = json.loads(self.github.gh("api", "graphql", "-f", self._PENDING_QUERY, "-F", f"owner={owner}", + "-F", f"name={name}", "-F", f"number={number}")) + return data["data"]["repository"]["pullRequest"] + + def _cancel_pending_merge(self, number: int) -> bool: + """Take the PR out of the merge queue AND turn auto-merge off, then + confirm both on GitHub. True only when GitHub shows the PR open with + neither: any failure (a command, the network, a malformed answer) + leaves the merge unresolved.""" + try: + pr = self._pending_merge(number) + if pr.get("mergeQueueEntry"): + self.github.gh("api", "graphql", "-f", "query=mutation($id:ID!){dequeuePullRequest(" + "input:{id:$id}){clientMutationId}}", "-f", f"id={pr['id']}", ok_fail=True) + if pr.get("autoMergeRequest"): + self.github.gh("pr", "merge", str(number), "--repo", self.github.repository, "--disable-auto", + ok_fail=True) + after = self._pending_merge(number) + except (PublishError, OSError, ValueError, KeyError, TypeError) as exc: + self._trace(event="cancel_unconfirmed", pr=number, error=str(exc)) + return False + return after.get("state") == "OPEN" and not after.get("mergeQueueEntry") \ + and not after.get("autoMergeRequest") + def _do_close(self, item: OutboxItem) -> dict: number = str(item.body["pr"]) if self.github.pr(int(number)).get("state") == "OPEN": @@ -473,10 +693,8 @@ def _git(self, *args: str, env: Mapping[str, str] | None = None, input: bytes | raise PublishError(f"git {args[0]} failed: {proc.stderr.decode(errors='replace')[:300]}") return proc.stdout.decode().strip() - def _build_commit(self, base_sha: str, files: Mapping[str, str], deleted: list[str], title: str, - *, when: float) -> str: - """The change as ONE commit on ``base_sha``, built in a scratch index. - Deterministic (dates from the item), so a retry rebuilds the same SHA.""" + def _ensure_clone(self) -> str: + """The publisher's own bare clone (never a working tree); returns the remote.""" remote = self.git_remote or f"https://github.com/{self.github.repository}.git" if not (self.clone / ".git").exists() and not (self.clone / "HEAD").exists(): self.clone.parent.mkdir(parents=True, exist_ok=True) @@ -484,6 +702,13 @@ def _build_commit(self, base_sha: str, files: Mapping[str, str], deleted: list[s stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False, cwd=self.github.cwd) if proc.returncode != 0: raise PublishError(f"git clone failed: {proc.stderr.decode(errors='replace')[:300]}") + return remote + + def _build_commit(self, base_sha: str, files: Mapping[str, str], deleted: list[str], title: str, + *, when: float) -> str: + """The change as ONE commit on ``base_sha``, built in a scratch index. + Deterministic (dates from the item), so a retry rebuilds the same SHA.""" + remote = self._ensure_clone() self._git("fetch", "--quiet", remote, "+refs/heads/main:refs/remotes/origin/main") if self._git("cat-file", "-t", base_sha) != "commit": raise PublishError(f"base {base_sha[:12]} is not a commit") diff --git a/src/infermatrix_copilot/knowledge_service/gate_verifier.py b/src/infermatrix_copilot/knowledge_service/gate_verifier.py index fcd1b926..eee5d2b7 100644 --- a/src/infermatrix_copilot/knowledge_service/gate_verifier.py +++ b/src/infermatrix_copilot/knowledge_service/gate_verifier.py @@ -407,13 +407,16 @@ def relevant(path: str) -> bool: def verify_change(ctx: Context, *, pr: dict, head_sha: str, pre: str, post: str, final: str, - effective_base: str, stage: str) -> list[str]: + effective_base: str, stage: str, verdict: dict | None = None, + check_holds: bool = True) -> list[str]: """Problems with one PR's knowledge change (empty when it passes). ``pre``/``post`` are the commits (or trees) around this PR's change as it is applied; ``final`` is the tree that would land; ``effective_base`` is the state the change is applied on (current main, or the first parent of - the merge-group segment).""" + the merge-group segment). The publisher's local gate passes the verdict it + was handed (its signature and binding already checked) and no hold list: + a pause there is the signed control record.""" number = int(pr["number"]) entries = ctx.git.raw_diff(pre, post) if not touches_knowledge(entries): @@ -441,7 +444,7 @@ def verify_change(ctx: Context, *, pr: dict, head_sha: str, pre: str, post: str, touched = {e["path"][len(KNOWLEDGE_PREFIX):] for e in entries if e["path"].startswith(KNOWLEDGE_PREFIX)} try: - holds = ctx.holds() + holds = ctx.holds() if check_holds else {} if holds.get("global"): problems.append("the knowledge service holds every repository (global pause)") if number in {int(p) for p in holds.get("prs") or []}: @@ -454,7 +457,9 @@ def verify_change(ctx: Context, *, pr: dict, head_sha: str, pre: str, post: str, except (GateError, SignatureError, LifecycleError, KeyError, TypeError, ValueError) as exc: problems.append(f"hold list: {exc}") - verdict, rejected = find_verdict(ctx, number, head_sha) + rejected: list[str] = [] + if verdict is None: + verdict, rejected = find_verdict(ctx, number, head_sha) if verdict is None: detail = f" (rejected: {'; '.join(rejected[:3])})" if rejected else "" problems.append(f"no valid signed verdict for PR #{number} at {head_sha[:12]}{detail}") diff --git a/test/test_kb_external.py b/test/test_kb_external.py index b543ce27..5bb849dd 100644 --- a/test/test_kb_external.py +++ b/test/test_kb_external.py @@ -81,8 +81,7 @@ def _add_rule_files(): 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]) + envelope = _items(tmp_path, "merge")[-1]["body"]["verdict"] return verify("kb-gate-verdict", envelope, load_public_key(public_key_text(rt.outbox._key.public_key()))) @@ -94,7 +93,7 @@ def test_a_human_knowledge_pr_passes_the_auto_gate_and_gets_a_verdict(tmp_path): (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']}"] + assert merge.advance(rt, lifecycle) == [f"merge 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 @@ -170,49 +169,44 @@ def test_a_new_head_is_judged_again_and_drafts_and_our_own_prs_are_skipped(tmp_p 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.""" +def _local_gate(tmp_path, rt, number): + """Run the publisher's local gate on the verdict this service handed over.""" import time - from infermatrix_copilot.knowledge_service import gate_verifier as gv - from infermatrix_copilot.knowledge_service.signing import sign + from infermatrix_copilot.kb_service import local_gate - (comment,) = [item["body"]["comment"] for item in _items(tmp_path, "post_verdict") - if item["body"]["pr"] == number] - head = rt.github.open[number]["head"]["sha"] + (body,) = [item["body"] for item in _items(tmp_path, "merge") if item["body"]["pr"] == number] + key = rt.outbox._key.public_key() + verdict = local_gate.check_verdict(body["verdict"], key, repository=REPO, pr=number, + head_sha=body["head_sha"], now=time.time()) + clone = rt.knowledge.path + _git(clone, "fetch", "-q", "origin", "+refs/heads/main:refs/remotes/origin/main", + f"+refs/pull/{number}/head:refs/kb/pr-{number}") + pr = {"number": number, "state": "open", "head": {"sha": body["head_sha"]}, "labels": []} + return local_gate.gate(clone, repository=REPO, repo="demo", pr=pr, head_sha=body["head_sha"], + main_sha=_git(clone, "rev-parse", "refs/remotes/origin/main"), verdict=verdict, + public_key=key, now=time.time()) - 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}] +def test_verdicts_for_external_prs_pass_the_publishers_local_gate(tmp_path): + import pytest - 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 + from infermatrix_copilot.knowledge_service.verdict import VerdictError - -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) == [] + assert _local_gate(tmp_path, rt, 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) + rt.github.reviews[8] = [{"id": 100, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] + assert [e for e in poll_external(rt) if "PR #7" in e] == [] # a merge in flight is not judged again + rt.ledger.update_changeset(rt.ledger.changesets("demo", ("merge_requested",))[0]["id"], status="merged") merge.advance(rt, lifecycle) - assert _verifier_passes(tmp_path, rt, origin, 8, labels=("kb:human-approved",), reviews=(review,)) == [] + with pytest.raises(VerdictError, match="only auto verdicts merge"): + _local_gate(tmp_path, rt, 8) # v8: a human-approved verdict never merges def test_a_retraction_on_a_later_page_of_reviews_is_seen(tmp_path): @@ -255,30 +249,6 @@ def test_a_head_sent_to_people_can_be_approved_without_a_new_push(tmp_path): 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 @@ -298,139 +268,3 @@ def test_rules_an_external_pr_retires_enter_the_retirement_ledger(tmp_path): 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_flow.py b/test/test_kb_flow.py index 9607d50a..28faa55c 100644 --- a/test/test_kb_flow.py +++ b/test/test_kb_flow.py @@ -89,7 +89,8 @@ def _items(tmp_path, kind): # -- merge flow ------------------------------------------------------------------------- -def test_happy_path_signs_a_verdict_bound_to_the_pr_and_enqueues_once(tmp_path): + +def test_happy_path_signs_a_verdict_bound_to_the_pr_and_hands_it_to_the_publisher(tmp_path): rt, lifecycle = _flow_runtime(tmp_path) changeset_id, publisher = _open_pr(tmp_path, rt, lifecycle) head = "d" * 40 @@ -97,11 +98,11 @@ def test_happy_path_signs_a_verdict_bound_to_the_pr_and_enqueues_once(tmp_path): assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" rt.github.prs[42] = {"state": "open", "merged": False, "head": {"sha": head}} - assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] - comment = _items(tmp_path, "post_verdict")[0]["body"]["comment"] - assert comment.startswith(merge.VERDICT_MARKER) - envelope = json.loads(comment.split("```json\n", 1)[1].rsplit("\n```", 1)[0]) - verdict = verify("kb-gate-verdict", envelope, load_public_key(public_key_text(rt.outbox._key.public_key()))) + assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"] + assert rt.ledger.changeset(changeset_id)["status"] == "merge_requested" + body = _items(tmp_path, "merge")[0]["body"] + assert (body["changeset_id"], body["pr"], body["head_sha"]) == (changeset_id, 42, head) + verdict = verify("kb-gate-verdict", body["verdict"], load_public_key(public_key_text(rt.outbox._key.public_key()))) check_binding(verdict, repository="JiusiServe/InferMatrixCopilot", pr=42, head_sha=head, now=rt.clock()) base = rt.knowledge.knowledge_files(rt.ledger.changeset(changeset_id)["detail"]["base_sha"]) head_files = {**base, **rt.load_changeset_files(changeset_id)["files"]} @@ -110,16 +111,12 @@ def test_happy_path_signs_a_verdict_bound_to_the_pr_and_enqueues_once(tmp_path): assert manifests_equal(verdict["manifest"], manifest_from_files( {k: v for k, v in base.items() if k in touched}, {k: v for k, v in head_files.items() if k in touched})) assert all(b["verdict"] == "pass" for b in verdict["blocks"]) + assert not _items(tmp_path, "post_verdict") and not _items(tmp_path, "enqueue") - _ack(tmp_path, rt, publisher, kind="post_verdict", changeset_id=changeset_id, ok=True) - assert merge.advance(rt, lifecycle) == [] # PR-stage kb-gate not green yet - rt.github.statuses[head] = "success" - assert merge.advance(rt, lifecycle) == [f"enqueue issued {changeset_id}"] - assert _items(tmp_path, "enqueue")[0]["body"] == {"changeset_id": changeset_id, "pr": 42, "head_sha": head} - _ack(tmp_path, rt, publisher, kind="enqueue", changeset_id=changeset_id, ok=True) - rt.github.prs[42] = {"state": "closed", "merged": True, "merge_commit_sha": "e" * 40, "head": {"sha": head}} - assert merge.advance(rt, lifecycle) == [f"merged {changeset_id}"] - assert rt.ledger.changeset(changeset_id)["merge_sha"] == "e" * 40 + _ack(tmp_path, rt, publisher, kind="merge", changeset_id=changeset_id, ok=True, pr=42, head_sha=head, + merge_sha="e" * 40, post_check="passed", problems=[]) + changeset = rt.ledger.changeset(changeset_id) + assert changeset["status"] == "merged" and changeset["merge_sha"] == "e" * 40 def test_head_change_after_signing_goes_to_people(tmp_path): @@ -131,19 +128,19 @@ def test_head_change_after_signing_goes_to_people(tmp_path): assert rt.ledger.human_queue("demo") -def test_one_queued_knowledge_pr_per_repository(tmp_path): + +def test_one_merge_in_flight_per_repository(tmp_path): rt, lifecycle = _flow_runtime(tmp_path) first, publisher = _open_pr(tmp_path, rt, lifecycle) - rt.ledger.update_changeset(first, status="queued", pr_number=1, head_sha="a" * 40) + rt.ledger.update_changeset(first, status="merge_requested", pr_number=1, head_sha="a" * 40) rt.github.prs[1] = {"state": "open", "merged": False, "head": {"sha": "a" * 40}} second = rt.ledger.new_changeset_id("demo", "intake") - rt.ledger.stage_intake(rt.ledger.acquire_lease("t"), "demo", second, detail={}, status="verdict_posted", + rt.ledger.stage_intake(rt.ledger.acquire_lease("t"), "demo", second, detail={}, status="pr_open", verdicts=[], human_reason="", drafted_events=[]) rt.ledger.update_changeset(second, pr_number=2, head_sha="b" * 40) rt.github.prs[2] = {"state": "open", "merged": False, "head": {"sha": "b" * 40}} - rt.github.statuses["b" * 40] = "success" - assert merge.advance(rt, lifecycle) == [] # waits for the queued one to land - assert not _items(tmp_path, "enqueue") + assert merge.advance(rt, lifecycle) == [] # waits for the first merge to be answered + assert not _items(tmp_path, "merge") def test_refused_publisher_action_fails_and_is_queued_for_people(tmp_path): @@ -154,6 +151,62 @@ def test_refused_publisher_action_fails_and_is_queued_for_people(tmp_path): assert any("patch mismatch" in row["reason"] for row in rt.ledger.human_queue("demo")) +def test_a_refused_merge_goes_by_what_the_local_gate_said(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + changeset_id, publisher = _open_pr(tmp_path, rt, lifecycle) + head = "d" * 40 + _ack(tmp_path, rt, publisher, kind="open_pr", changeset_id=changeset_id, ok=True, pr=42, head_sha=head) + rt.github.prs[42] = {"state": "open", "merged": False, "head": {"sha": head}} + + def refused(**ack): + merge.advance(rt, lifecycle) + _ack(tmp_path, rt, publisher, kind="merge", changeset_id=changeset_id, ok=False, **ack) + return rt.ledger.changeset(changeset_id) + + # the moment, not the change: signed again next pass + assert refused(error="main kept moving while PR #42 was checked")["status"] == "pr_open" + # context moved: rebuilt on current main + assert refused(error="local gate: context changed since the verdict was judged: x", + problems=["context changed since the verdict was judged: x"])["status"] == "rebuild_needed" + rt.ledger.update_changeset(changeset_id, status="pr_open") + # the change itself: never merged, people decide + changeset = refused(error="local gate: L1 dangling_reference", problems=["L1 dangling_reference a b"]) + assert changeset["status"] == "gate_failed" + assert changeset["detail"]["gate_problems"] == ["L1 dangling_reference a b"] + assert any("local gate refused" in row["reason"] for row in rt.ledger.human_queue("demo")) + + +def test_a_failed_post_merge_check_pauses_the_repository(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + changeset_id, publisher = _open_pr(tmp_path, rt, lifecycle) + head = "d" * 40 + _ack(tmp_path, rt, publisher, kind="open_pr", changeset_id=changeset_id, ok=True, pr=42, head_sha=head) + rt.github.prs[42] = {"state": "open", "merged": False, "head": {"sha": head}} + merge.advance(rt, lifecycle) + _ack(tmp_path, rt, publisher, kind="merge", changeset_id=changeset_id, ok=True, pr=42, head_sha=head, + merge_sha="e" * 40, post_check="failed", problems=["L1 dangling_reference x y"]) + assert rt.ledger.changeset(changeset_id)["status"] == "merged" + assert rt.ledger.repo_state("demo")["paused"] == 1 + assert any("post-merge check failed" in row["reason"] for row in rt.ledger.human_queue("demo")) + + +def test_a_receipt_arriving_after_github_showed_the_merge_still_counts(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + changeset_id, publisher = _open_pr(tmp_path, rt, lifecycle) + head = "d" * 40 + _ack(tmp_path, rt, publisher, kind="open_pr", changeset_id=changeset_id, ok=True, pr=42, head_sha=head) + rt.github.prs[42] = {"state": "open", "merged": False, "head": {"sha": head}} + merge.advance(rt, lifecycle) + item = rt.ledger.changeset(changeset_id)["pending_item"]["id"] + rt.github.prs[42] = {"state": "closed", "merged": True, "merge_commit_sha": "e" * 40, "head": {"sha": head}} + assert f"merged {changeset_id}" in merge.advance(rt, lifecycle) # seen on GitHub first + assert rt.ledger.repo_state("demo")["paused"] == 0 + _ack(tmp_path, rt, publisher, kind="merge", changeset_id=changeset_id, ok=True, pr=42, head_sha=head, + merge_sha="e" * 40, post_check="failed", problems=["L1 dangling_reference x y"], item_id=item) + assert rt.ledger.repo_state("demo")["paused"] == 1 + assert rt.ledger.changeset(changeset_id)["detail"]["post_check"] == "failed" + + def test_merged_retirements_become_purge_candidates_next_release(tmp_path): rt, lifecycle = _flow_runtime(tmp_path) cs = rt.ledger.new_changeset_id("demo", "sweep") @@ -372,25 +425,25 @@ def advance(rt_, lc): # -- review regressions --------------------------------------------------------------------- + def test_pending_actions_are_not_reissued_and_stale_acks_change_nothing(tmp_path): rt, lifecycle = _flow_runtime(tmp_path) changeset_id, publisher = _open_pr(tmp_path, rt, lifecycle) head = "d" * 40 _ack(tmp_path, rt, publisher, kind="open_pr", changeset_id=changeset_id, ok=True, pr=42, head_sha=head) rt.github.prs[42] = {"state": "open", "merged": False, "head": {"sha": head}} - assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] - assert merge.advance(rt, lifecycle) == [] # still pending: no duplicate verdict item - assert len(_items(tmp_path, "post_verdict")) == 1 + assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"] + assert merge.advance(rt, lifecycle) == [] # still pending: no duplicate merge item + assert len(_items(tmp_path, "merge")) == 1 first = rt.ledger.changeset(changeset_id)["pending_item"]["id"] - _ack(tmp_path, rt, publisher, kind="post_verdict", changeset_id=changeset_id, ok=True) - rt.github.statuses[head] = "success" - merge.advance(rt, lifecycle) - _ack(tmp_path, rt, publisher, kind="enqueue", changeset_id=changeset_id, ok=True) - assert rt.ledger.changeset(changeset_id)["status"] == "queued" - # a late REJECTION of the earlier verdict item must not demote the queued change set - _ack(tmp_path, rt, publisher, kind="post_verdict", changeset_id=changeset_id, ok=False, + _ack(tmp_path, rt, publisher, kind="merge", changeset_id=changeset_id, ok=True, pr=42, head_sha=head, + merge_sha="e" * 40, post_check="passed", problems=[]) + assert rt.ledger.changeset(changeset_id)["status"] == "merged" + # a late REJECTION of the same item must not undo the merge + _ack(tmp_path, rt, publisher, kind="merge", changeset_id=changeset_id, ok=False, error="late duplicate", item_id=first) - assert rt.ledger.changeset(changeset_id)["status"] == "queued" + assert rt.ledger.changeset(changeset_id)["status"] == "merged" + def test_expired_pending_action_may_be_reissued(tmp_path): @@ -401,8 +454,10 @@ def test_expired_pending_action_may_be_reissued(tmp_path): rt.github.prs[42] = {"state": "open", "merged": False, "head": {"sha": head}} merge.advance(rt, lifecycle) now = rt.clock() - rt.clock = lambda: now + 31 * 60 - assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] + rt.clock = lambda: now + 36 * 60 # 30 min expiry + clock-skew margin + assert merge.advance(rt, lifecycle) == [f"merge_expired {changeset_id}"] + assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"] + assert len(_items(tmp_path, "merge")) == 2 def test_overturn_breaker_publishes_the_pause_at_once(tmp_path): @@ -580,6 +635,7 @@ def do_rollback(): assert json.loads(rt.ledger.get_cursor("*", "rollback_pin"))["main"] == second + def test_resume_returns_paused_prs_to_re_signing(tmp_path): rt, lifecycle = _flow_runtime(tmp_path) changeset_id, publisher = _open_pr(tmp_path, rt, lifecycle) @@ -591,7 +647,7 @@ def test_resume_returns_paused_prs_to_re_signing(tmp_path): merge.resume_paused_prs(rt.ledger, "demo") assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" rt.github.prs[42] = {"state": "open", "merged": False, "head": {"sha": "d" * 40}} - assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] # fresh verdict + assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"] # fresh verdict def test_gated_change_sets_are_published_after_an_interruption(tmp_path): @@ -616,20 +672,20 @@ def answer(role, prompt): assert rt.ledger.changeset(changeset_id)["status"] == "pr_requested" + def test_a_pause_ack_after_resume_sends_the_pr_back_to_re_signing(tmp_path): rt, lifecycle = _flow_runtime(tmp_path) changeset_id, publisher = _open_pr(tmp_path, rt, lifecycle) _ack(tmp_path, rt, publisher, kind="open_pr", changeset_id=changeset_id, ok=True, pr=42, head_sha="d" * 40) - rt.ledger.update_changeset(changeset_id, status="queued") rt.ledger.bump_generation("demo", pause=True, reason="drill") merge.pause_open_prs(rt.ledger, rt.outbox, "demo", "drill") rt.ledger.resume("demo") merge.resume_paused_prs(rt.ledger, "demo") # nothing paused yet: the ack is late - assert rt.ledger.changeset(changeset_id)["status"] == "queued" + assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset_id, ok=True, item_id="late") assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" # not stranded as paused rt.github.prs[42] = {"state": "open", "merged": False, "head": {"sha": "d" * 40}} - assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] + assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"] def test_global_pause_is_respected_by_acks_advance_and_publish(tmp_path): diff --git a/test/test_kb_publisher.py b/test/test_kb_publisher.py index bdac2014..d4564b5c 100644 --- a/test/test_kb_publisher.py +++ b/test/test_kb_publisher.py @@ -21,6 +21,10 @@ def _git(path: Path, *args: str) -> str: return _git_raw(path, *args).strip() +class Crash(BaseException): + """The publisher process dies (not an error it handles).""" + + class FakeGh: """Just enough of `gh` for the publisher; PR heads come from the git remote.""" @@ -30,6 +34,13 @@ def __init__(self, origin: Path): self.calls: list[list[str]] = [] self.comments: list[tuple[int, str]] = [] self.fail: set[str] = set() # "remove-label" / "undo": that call fails + self.crash_after_merge = False # the publisher dies right after GitHub merged + self.merge_error_after_merging = False # GitHub merged, but the CLI reports an error + self.queue_instead_of_merging = False # main requires a merge queue: gh only enqueues + self.fail_views = 0 # the next N `pr view` calls fail + self.fail_disable_auto = 0 # the next N `--disable-auto` calls fail + self.fail_dequeue = 0 # the next N dequeue mutations fail + self.raise_on_query = 0 # the next N GraphQL queries crash the command (OSError) def _by_branch(self, branch: str) -> list[dict]: return [pr for pr in self.prs.values() if pr["headRefName"] == branch and pr["state"] == "OPEN"] @@ -49,10 +60,17 @@ def __call__(self, argv, *, cwd=None, input=None, stdout=None, stderr=None, chec number = 42 + len(self.prs) self.prs[number] = {"number": number, "state": "OPEN", "isDraft": "--draft" in args, "headRefOid": head, "headRefName": branch, "id": f"PR_{number}", - "labels": [], "queued": False, "body": (input or b"").decode()} + "labels": [], "queued": False, "body": (input or b"").decode(), + "baseRefName": "main", "mergeCommit": None} + _git(self.origin, "update-ref", f"refs/pull/{number}/head", head) elif args[:2] == ["pr", "view"]: + if self.fail_views: + self.fail_views -= 1 + return subprocess.CompletedProcess(argv, 1, b"", b"HTTP 502") pr = self.prs[int(args[2])] out = json.dumps({**{k: pr[k] for k in ("number", "state", "isDraft", "headRefOid", "headRefName", "id")}, + "baseRefName": pr.get("baseRefName", "main"), "mergeCommit": pr.get("mergeCommit"), + "autoMergeRequest": {"mergeMethod": "MERGE"} if pr.get("auto") else None, "labels": [{"name": name} for name in pr["labels"]]}) elif args[:2] == ["pr", "comment"]: self.comments.append((int(args[2]), input.decode())) @@ -64,10 +82,25 @@ def __call__(self, argv, *, cwd=None, input=None, stdout=None, stderr=None, chec elif args[:2] == ["pr", "merge"]: pr = self.prs[int(args[2])] if "--disable-auto" in args: - pr["queued"] = False - else: + if self.fail_disable_auto: + self.fail_disable_auto -= 1 + return subprocess.CompletedProcess(argv, 1, b"", b"HTTP 502") + pr["queued"] = pr["auto"] = False + elif "--auto" in args: assert args[args.index("--match-head-commit") + 1] == pr["headRefOid"] pr["queued"] = True + elif self.queue_instead_of_merging: + pr["in_queue"] = True # a merge queue entry; auto-merge stays off + else: # a direct merge (design v8): a real merge commit on origin's main + assert args[args.index("--match-head-commit") + 1] == pr["headRefOid"] + assert not pr["isDraft"] + _git(self.origin, "-c", "user.name=gh", "-c", "user.email=gh@e", "merge", "-q", "--no-ff", + "-m", f"Merge pull request #{pr['number']}", pr["headRefOid"]) + pr["state"], pr["mergeCommit"] = "MERGED", {"oid": _git(self.origin, "rev-parse", "HEAD")} + if self.crash_after_merge: + raise Crash() + if self.merge_error_after_merging: + return subprocess.CompletedProcess(argv, 1, b"", b"HTTP 502 (but it merged)") elif args[:2] == ["pr", "edit"]: labels = self.prs[int(args[2])]["labels"] if "--remove-label" in args and "remove-label" in self.fail: @@ -78,10 +111,22 @@ def __call__(self, argv, *, cwd=None, input=None, stdout=None, stderr=None, chec labels.remove(args[args.index("--remove-label") + 1]) elif args[:2] == ["pr", "close"]: self.prs[int(args[2])]["state"] = "CLOSED" + elif args[:2] == ["api", "graphql"] and args[3].startswith("query=query("): + if self.raise_on_query: + self.raise_on_query -= 1 + raise OSError("gh could not be started") + pr = self.prs[int([a for a in args if a.startswith("number=")][0].split("=", 1)[1])] + out = json.dumps({"data": {"repository": {"pullRequest": { + "id": pr["id"], "state": pr["state"], + "mergeQueueEntry": {"id": "Q"} if pr.get("in_queue") else None, + "autoMergeRequest": {"enabledAt": "t"} if pr.get("auto") else None}}}}) elif args[:2] == ["api", "graphql"]: + if "dequeuePullRequest" in args[3] and self.fail_dequeue: + self.fail_dequeue -= 1 + return subprocess.CompletedProcess(argv, 1, b"", b"HTTP 502") for pr in self.prs.values(): if f"id={pr['id']}" in args: - pr["queued"] = False + pr["queued"] = pr["in_queue"] = False elif args[:3] == ["api", "-X", "PUT"]: pass else: @@ -228,9 +273,141 @@ def test_pause_is_performed_even_while_the_repository_is_paused(tmp_path): assert rt.ledger.changeset(changeset_id)["status"] == "paused" -def test_pause_resume_regate_and_enqueue_round_trip(tmp_path): - """pause (draft + kb:hold) -> kb resume -> fresh verdict lifts the label -> - the gate can pass -> enqueue readies the draft and queues it.""" +def _merge_ready(tmp_path): + """The service's PR is open and its merge item (with a signed verdict) is issued.""" + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + _collect(rt) + head = gh.prs[42]["headRefOid"] + rt.github.prs[42] = {"number": 42, "state": "open", "merged": False, "head": {"sha": head}} + assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"] + rt.outbox.refresh_control() + return rt, lifecycle, changeset_id, pub, gh, head + + +def test_the_publisher_merges_only_after_its_local_gate_passes(tmp_path): + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + assert pub.run_once()["performed"] == 1 + origin = tmp_path / "origin" + merge_sha = _git(origin, "rev-parse", "main") + assert _git(origin, "rev-parse", "main^2") == head # a merge of exactly the verified head + _collect(rt) + changeset = rt.ledger.changeset(changeset_id) + assert changeset["status"] == "merged" and changeset["merge_sha"] == merge_sha + assert changeset["detail"]["post_check"] == "passed" + assert not list((tmp_path / "publisher" / "intents").glob("*.json")) + + +def test_a_change_the_local_gate_refuses_is_never_merged(tmp_path): + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + origin = tmp_path / "origin" + before = _git(origin, "rev-parse", "main") + index = origin / "knowledge" / "repos" / "demo" / "core" / "_index.md" # main moves in the same owner dir + index.write_text(index.read_text() + "\nmain moved here\n") + _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-qam", "main moves") + moved = _git(origin, "rev-parse", "main") + summary = pub.run_once() + assert summary["failed"] == 1 and gh.prs[42]["state"] == "OPEN" + assert _git(origin, "rev-parse", "main") == moved != before + _collect(rt) + changeset = rt.ledger.changeset(changeset_id) + assert changeset["status"] == "rebuild_needed" + assert "context changed" in changeset["detail"]["rebuild_because"] + + +def test_a_merge_whose_receipt_was_lost_is_recovered_not_repeated(tmp_path): + from test_kb_publisher import Crash + + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + gh.crash_after_merge = True + with pytest.raises(Crash): + pub.run_once() # merged on GitHub, then the process died + assert list((tmp_path / "publisher" / "intents").glob("*.json")) + gh.crash_after_merge = False + merges = sum(1 for call in gh.calls if call[:2] == ["pr", "merge"]) + pub.run_once() + assert sum(1 for call in gh.calls if call[:2] == ["pr", "merge"]) == merges # not merged twice + _collect(rt) + changeset = rt.ledger.changeset(changeset_id) + assert changeset["status"] == "merged" + assert changeset["merge_sha"] == _git(tmp_path / "origin", "rev-parse", "main") + # a recovered receipt carries no post-merge check: the service runs it itself + assert changeset["detail"]["post_check"] == "unknown" + assert f"post_check passed {changeset_id}" in merge.advance(rt, lifecycle) + assert rt.ledger.changeset(changeset_id)["detail"]["post_check"] == "passed" + + +def test_a_merge_reported_as_failed_that_did_merge_is_settled_from_github(tmp_path): + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + gh.merge_error_after_merging = True + pub.run_once() + assert list((tmp_path / "publisher" / "intents").glob("*.json")) # kept: the outcome is unknown + assert not list((tmp_path / "state" / "inbox" / "acks").glob("*.json")) # and nothing was answered + gh.merge_error_after_merging = False + merges = sum(1 for call in gh.calls if call[:2] == ["pr", "merge"]) + pub.run_once() + assert sum(1 for call in gh.calls if call[:2] == ["pr", "merge"]) == merges + _collect(rt) + assert rt.ledger.changeset(changeset_id)["status"] == "merged" + + +def test_a_merge_that_only_queued_is_not_reported_as_merged(tmp_path): + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + gh.queue_instead_of_merging = True + assert pub.run_once()["failed"] == 1 + assert gh.prs[42]["state"] == "OPEN" and not gh.prs[42].get("in_queue") # dequeued + assert any("dequeuePullRequest" in " ".join(call) for call in gh.calls) + _collect(rt) + changeset = rt.ledger.changeset(changeset_id) + assert changeset["status"] == "gate_failed" + assert "merge queue" in changeset["detail"]["gate_problems"][0] + + +def test_a_queued_merge_whose_cancellation_fails_stays_unresolved(tmp_path): + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + gh.queue_instead_of_merging = True + gh.fail_dequeue = 2 # this round, and the next recovery attempt + pub.run_once() + assert gh.prs[42]["in_queue"] is True + assert list((tmp_path / "publisher" / "intents").glob("*.json")) # kept: GitHub may still merge it + assert not list((tmp_path / "state" / "inbox" / "acks").glob("*.json")) + pub.run_once() # recovery: cancelling fails again + assert list((tmp_path / "publisher" / "intents").glob("*.json")) + pub.run_once() # recovery dequeues it, the item is tried again and refused (main queues) + assert not gh.prs[42]["in_queue"] + _collect(rt) + assert rt.ledger.changeset(changeset_id)["status"] == "gate_failed" + + +def test_an_error_while_cancelling_leaves_the_merge_unresolved(tmp_path): + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + gh.queue_instead_of_merging = True + gh.raise_on_query = 1 # the confirmation query cannot even run + pub.run_once() + assert list((tmp_path / "publisher" / "intents").glob("*.json")) + assert not list((tmp_path / "state" / "inbox" / "acks").glob("*.json")) + assert rt.ledger.changeset(changeset_id)["status"] == "merge_requested" + + +def test_an_unsettled_recovery_blocks_the_item_until_github_answers(tmp_path): + from test_kb_publisher import Crash + + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + gh.crash_after_merge = True + with pytest.raises(Crash): + pub.run_once() + gh.crash_after_merge = False + gh.fail_views = 1 # recovery cannot ask GitHub this round + summary = pub.run_once() + assert summary["skipped"] == 1 and summary["failed"] == 0 + assert list((tmp_path / "publisher" / "intents").glob("*.json")) + assert not list((tmp_path / "state" / "inbox" / "acks").glob("*.json")) + pub.run_once() # now it can: the merge is settled from GitHub + _collect(rt) + assert rt.ledger.changeset(changeset_id)["status"] == "merged" + + +def test_a_paused_then_resumed_pr_merges(tmp_path): rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) pub.run_once() _collect(rt) @@ -242,22 +419,53 @@ def test_pause_resume_regate_and_enqueue_round_trip(tmp_path): pub.run_once() _collect(rt) assert gh.prs[42]["isDraft"] and "kb:hold" in gh.prs[42]["labels"] - assert rt.ledger.changeset(changeset_id)["status"] == "paused" def _resume(): rt.ledger.resume(lifecycle.repo) merge.resume_paused_prs(rt.ledger, lifecycle.repo) rt.outbox.transition(_resume, public_repos={lifecycle.repo}) - assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] - pub.run_once() - _collect(rt) - assert "kb:hold" not in gh.prs[42]["labels"] and gh.comments[-1][0] == 42 - rt.github.statuses[head] = "success" # the precheck the comment triggered - assert merge.advance(rt, lifecycle) == [f"enqueue issued {changeset_id}"] - pub.run_once() + rt.github.prs[42]["draft"] = True + assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"] + assert pub.run_once()["performed"] == 1 + assert gh.prs[42]["state"] == "MERGED" and "kb:hold" not in gh.prs[42]["labels"] _collect(rt) - assert gh.prs[42]["isDraft"] is False and gh.prs[42]["queued"] is True - assert rt.ledger.changeset(changeset_id)["status"] == "queued" + assert rt.ledger.changeset(changeset_id)["status"] == "merged" + + +def test_a_second_publisher_does_nothing_while_one_runs(tmp_path): + import fcntl + import os + + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + fd = os.open(tmp_path / "publisher" / "publisher.lock", os.O_RDWR | os.O_CREAT, 0o600) + fcntl.flock(fd, fcntl.LOCK_EX) + try: + calls = len(gh.calls) + assert pub.run_once().get("locked") == 1 + assert len(gh.calls) == calls and gh.prs[42]["state"] == "OPEN" + finally: + os.close(fd) + assert pub.run_once()["performed"] == 1 + + +def test_a_pause_issued_while_the_gate_ran_stops_the_merge(tmp_path): + rt, lifecycle, changeset_id, pub, gh, head = _merge_ready(tmp_path) + from infermatrix_copilot.kb_service import local_gate + + real = local_gate.gate + + def gate_then_pause(*args, **kwargs): + problems = real(*args, **kwargs) + rt.outbox.transition(lambda: rt.ledger.bump_generation(lifecycle.repo, pause=True, reason="drill"), + public_repos={lifecycle.repo}) + return problems + local_gate.gate = gate_then_pause + try: + assert pub.run_once()["failed"] == 1 + finally: + local_gate.gate = real + assert gh.prs[42]["state"] == "OPEN" + assert not any(call[:2] == ["pr", "merge"] for call in gh.calls) def test_a_failed_pause_is_retried_not_acknowledged(tmp_path): diff --git a/test/test_kb_rebuild.py b/test/test_kb_rebuild.py index 4ff211ed..528d6ab2 100644 --- a/test/test_kb_rebuild.py +++ b/test/test_kb_rebuild.py @@ -1,4 +1,4 @@ -"""Stale verdicts: re-sign, rebuild on current main, or hand to people.""" +"""Refused merges: sign again, rebuild on current main, or hand to people.""" from __future__ import annotations @@ -8,78 +8,54 @@ HEAD = "d" * 40 -class GateGitHub(KnowledgeGitHub): - """kb-gate statuses with descriptions (the verifier's first problem).""" - - def __init__(self): - super().__init__() - self.descriptions: dict[str, str] = {} - - def _answer(self, url): - if "/commits/" in url and url.endswith("/status"): - sha = url.split("/commits/")[1].split("/")[0] - state = self.statuses.get(sha) - return {"statuses": [{"context": "kb-gate", "state": state, - "description": self.descriptions.get(sha, "")}] if state else []} - return super()._answer(url) - - def _posted(tmp_path): + """A change set whose merge item the publisher's local gate is about to answer.""" rt, lifecycle = _flow_runtime(tmp_path) - github = GateGitHub() - github.prs = rt.github.prs - rt.github = github changeset_id, publisher = _open_pr(tmp_path, rt, lifecycle) _ack(tmp_path, rt, publisher, kind="open_pr", changeset_id=changeset_id, ok=True, pr=42, head_sha=HEAD) rt.github.prs[42] = {"state": "open", "merged": False, "head": {"sha": HEAD}} - assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] - _ack(tmp_path, rt, publisher, kind="post_verdict", changeset_id=changeset_id, ok=True) - assert rt.ledger.changeset(changeset_id)["status"] == "verdict_posted" + assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"] + assert rt.ledger.changeset(changeset_id)["status"] == "merge_requested" + rt._test = (tmp_path, publisher, changeset_id) return rt, lifecycle, changeset_id, publisher -def _fail(rt, description): - rt.github.statuses[HEAD] = "failure" - rt.github.descriptions[HEAD] = description +def _fail(rt, problem, *, transient=False): + """The publisher's local gate refused the merge with this problem.""" + tmp_path, publisher, changeset_id = rt._test + if transient: + _ack(tmp_path, rt, publisher, kind="merge", changeset_id=changeset_id, ok=False, error=problem) + else: + _ack(tmp_path, rt, publisher, kind="merge", changeset_id=changeset_id, ok=False, + error=f"local gate: {problem}", problems=[problem]) -def test_the_signing_time_is_recorded_and_an_old_verdict_is_resigned(tmp_path): +def test_the_signing_time_is_recorded_with_the_merge_item(tmp_path): rt, lifecycle, changeset_id, _ = _posted(tmp_path) - signed_at = rt.ledger.changeset(changeset_id)["detail"]["verdict_issued_at"] - assert signed_at == rt.clock() - start = rt.clock() - rt.clock = lambda: start + merge.RESIGN_AFTER + 1 - assert merge.advance(rt, lifecycle) == [f"resign {changeset_id}"] - assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] # a fresh verdict, same head - assert len(_items(tmp_path, "post_verdict")) >= 1 + assert rt.ledger.changeset(changeset_id)["detail"]["verdict_issued_at"] == rt.clock() -def test_an_expired_or_missing_verdict_is_resigned(tmp_path): +def test_transient_refusals_are_signed_again(tmp_path): rt, lifecycle, changeset_id, _ = _posted(tmp_path) - _fail(rt, "no valid signed verdict for PR #42 at dddddddddddd (rejected: verdict is outside its issue window)") - assert merge.advance(rt, lifecycle) == [f"resign {changeset_id}"] + _fail(rt, "main kept moving while PR #42 was checked; retried next round", transient=True) assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" - - -def test_transient_gate_failures_wait(tmp_path): - rt, lifecycle, changeset_id, _ = _posted(tmp_path) - _fail(rt, "hold list: hold list is stale (the knowledge service may be down)") - assert merge.advance(rt, lifecycle) == [] - assert rt.ledger.changeset(changeset_id)["status"] == "verdict_posted" + assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"] # a fresh verdict, same head + assert len(_items(tmp_path, "merge")) == 1 # the answered item left the outbox; this is the new one def test_a_real_problem_goes_to_people_once(tmp_path): rt, lifecycle, changeset_id, _ = _posted(tmp_path) _fail(rt, "L1 duplicate_rule_id DEMO-2a") - assert merge.advance(rt, lifecycle) == [f"gate_failed {changeset_id}"] assert rt.ledger.changeset(changeset_id)["status"] == "gate_failed" - assert rt.ledger.human_queue("demo") + assert len(rt.ledger.human_queue("demo")) == 1 assert merge.advance(rt, lifecycle) == [] + assert _items(tmp_path, "merge") == [] # never retried as is def test_a_context_change_rebuilds_on_current_main_and_replaces_the_pr(tmp_path): rt, lifecycle, changeset_id, _ = _posted(tmp_path) _fail(rt, "context changed since the verdict was judged: repos/demo/core/_index.md") + assert rt.ledger.changeset(changeset_id)["status"] == "rebuild_needed" assert merge.advance(rt, lifecycle) == [] # outside the scheduler: no lease, no rebuild rt.lease_owner = rt.ledger.acquire_lease("scheduler") events = merge.advance(rt, lifecycle) @@ -146,74 +122,8 @@ def test_a_rule_that_no_longer_applies_goes_to_people(tmp_path): detail["operations"] = [{"kind": "edit_same_meaning", "page": "repos/demo/core/rules.md", "rule_id": "DEMO-9z", "section_markdown": "## DEMO-9z — gone\n"}] rt.ledger.update_changeset(changeset_id, detail=detail) - _fail(rt, "does not merge cleanly into the current base") + _fail(rt, "the PR does not merge cleanly into the current base") rt.lease_owner = rt.ledger.acquire_lease("scheduler") assert merge.advance(rt, lifecycle) == [f"rebuild_failed {changeset_id}"] assert rt.ledger.changeset(changeset_id)["status"] == "rebuild_failed" assert rt.ledger.human_queue("demo") - - -def test_a_queued_pr_that_stalls_is_resigned_and_frees_the_queue_slot(tmp_path): - rt, lifecycle, changeset_id, publisher = _posted(tmp_path) - rt.github.statuses[HEAD] = "success" - assert merge.advance(rt, lifecycle) == [f"enqueue issued {changeset_id}"] - _ack(tmp_path, rt, publisher, kind="enqueue", changeset_id=changeset_id, ok=True) - assert rt.ledger.changeset(changeset_id)["status"] == "queued" - start = rt.clock() - rt.clock = lambda: start + merge.QUEUE_STALL + 1 - assert merge.advance(rt, lifecycle) == [f"requeue {changeset_id}"] - assert rt.ledger.changeset(changeset_id)["status"] == "queued" # the slot is held until dequeued - assert merge.advance(rt, lifecycle) == [] # the pause item is pending - (pause,) = _items(tmp_path, "pause") - assert pause["body"]["pr"] == 42 - _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset_id, ok=True) - assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" - assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] - - -def test_an_outstanding_enqueue_blocks_re_signing_until_its_ack(tmp_path): - rt, lifecycle, changeset_id, publisher = _posted(tmp_path) - rt.github.statuses[HEAD] = "success" - assert merge.advance(rt, lifecycle) == [f"enqueue issued {changeset_id}"] - detail = rt.ledger.changeset(changeset_id)["detail"] - rt.ledger.update_changeset(changeset_id, detail={**detail, "verdict_issued_at": rt.clock() - merge.RESIGN_AFTER - 1}) - _fail(rt, "no valid signed verdict for PR #42") - assert merge.advance(rt, lifecycle) == [] # not re-signed under a live enqueue - assert rt.ledger.changeset(changeset_id)["pending_item"]["kind"] == "enqueue" - _ack(tmp_path, rt, publisher, kind="enqueue", changeset_id=changeset_id, ok=True) # a delayed ack - assert rt.ledger.changeset(changeset_id)["status"] == "queued" - assert merge.advance(rt, lifecycle) == [f"requeue {changeset_id}"] # the queued path dequeues first - - -def test_our_own_hold_label_and_statuses_older_than_the_verdict_never_escalate(tmp_path): - rt, lifecycle, changeset_id, _ = _posted(tmp_path) - _fail(rt, "PR #42 carries kb:hold") - assert merge.advance(rt, lifecycle) == [] - _fail(rt, "L1 duplicate_rule_id DEMO-2a") - - class Old(GateGitHub): - def _answer(self, url): - answer = super()._answer(url) - for status in answer.get("statuses", []) if isinstance(answer, dict) else []: - status["updated_at"] = "2020-01-01T00:00:00Z" # posted before this verdict - return answer - - old = Old() - old.prs, old.statuses, old.descriptions = rt.github.prs, rt.github.statuses, rt.github.descriptions - rt.github = old - assert merge.advance(rt, lifecycle) == [] - assert rt.ledger.changeset(changeset_id)["status"] == "verdict_posted" - - -def test_an_outstanding_enqueue_holds_the_single_queue_slot(tmp_path): - rt, lifecycle, first, publisher = _posted(tmp_path) - rt.github.statuses[HEAD] = "success" - assert merge.advance(rt, lifecycle) == [f"enqueue issued {first}"] - second = rt.ledger.new_changeset_id("demo", "intake") - rt.ledger.stage_intake(rt.ledger.acquire_lease("t2"), "demo", second, detail={}, status="verdict_posted", - kind="intake", verdicts=[], human_reason="", drafted_events=[]) - rt.ledger.update_changeset(second, pr_number=43, head_sha="e" * 40) - rt.github.prs[43] = {"state": "open", "merged": False, "head": {"sha": "e" * 40}} - rt.github.statuses["e" * 40] = "success" - assert merge.advance(rt, lifecycle) == [] # the first's enqueue is still executable - assert [i["body"]["changeset_id"] for i in _items(tmp_path, "enqueue")] == [first]