From 39290009853d2ba352d735fca161169f53f134c0 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 29 Sep 2026 12:49:41 +0800 Subject: [PATCH] Merge knowledge PRs through the publisher's local gate (design v8) The owner decided against GitHub's merge queue and rulesets. The publisher is now the only thing that merges knowledge PRs, and it runs the gate itself on the exact tree that would land. - service: at pr_open it signs a verdict bound to the PR's exact head and issues a merge item carrying it, at most one per repository. There is no verdict comment, no wait for a GitHub kb-gate status and no enqueue. Acks: - merged: the merge SHA and the post-merge check result are recorded, and a failed post-merge check pauses the repository and goes to people. - refused because context moved, pages changed, the PR does not merge cleanly or the manifest differs: rebuild_needed, rebuilt on current main by the scheduler. - refused for a transient reason (main kept moving, control changed, git/gh trouble): back to pr_open, signed again. - refused because of the change itself: gate_failed, people decide. It is never retried as is. An unanswered merge item returns to pr_open when it expires. v7 verdict_posted/queued records are signed again as pr_open. A receipt still counts when GitHub showed the merge first. A post-merge check of "unknown" (the publisher could not run it, or the merge was recovered after a crash) is run by the service on its own clone (verify_merged) and retried until it has an answer. - local_gate: on the publisher's installed Copilot version it checks the verdict (signature, binding, auto only), that every path is a governed page of the item's own repository, runs verify_change on merge-tree(main, head) with the verdict handed over and no hold list, and checks the merged tree for new tree-level issues. post_merge_problems repeats this on the merge commit GitHub actually made. - publisher merge action, under a cross-process flock for the whole round (a second instance does nothing): - verify the verdict; the PR must be open, target main and still have the signed head; - run the local gate, again if main moves (at most 3 times); - re-read the control record (a pause voids the item); - write a merge intent, then gh pr merge --merge --match-head-commit; - run the post-merge check. Nothing after the merge becomes a failure ack. Each round first recovers intents that have no completion record, so a lost receipt is re-sent and the PR is never merged twice. When gh pr merge itself errors, the outcome is unknown: the intent is kept and nothing is recorded or acked; recovery settles it from GitHub. A kb:hold label left by a v7 pause is removed (and confirmed gone) before the gate. Only a PR that GitHub shows as MERGED counts: if main has a merge queue or auto-merge (gh pr merge only enqueues), the PR is dequeued and auto-merge disabled, and the item is refused only once GitHub (GraphQL: mergeQueueEntry, autoMergeRequest) confirms neither is left; any failure while cancelling counts as unresolved; until then the intent stays and recovery keeps cancelling (or picks up a merge that happened meanwhile). An item with an unsettled intent is skipped until recovery hears from GitHub. - gate_verifier.verify_change takes verdict= and check_holds= (the bundle is regenerated). - external: its status set is derived from merge.IN_FLIGHT, so a PR whose merge is in flight is not judged again. Tests for the v7 queue, status and enqueue mechanics are rewritten for the merge path, and the human-approved withdrawal/queue-slot tests are removed: a human-approved verdict never merges in v8, and that code goes in the removal PR. New end-to-end tests with real git: a merge only after the local gate passes, a refused change is never merged, a lost receipt is recovered without a second merge, a second publisher does nothing, and a pause during the gate stops the merge. Also tested: a delayed receipt with a failed post-merge check, a merge reported as failed that did merge, pause -> resume -> merge, a merge that only queued (queue membership modelled apart from auto-merge; a failing dequeue; a command error while confirming), and a recovery lookup that fails before it succeeds. Co-Authored-By: Claude Opus 5.5 --- .github/kb-gate/MANIFEST.sha256 | 2 +- .github/kb-gate/kb_gate.py | 13 +- doc/architecture/SPEC/kb_service.md | 23 ++ doc/architecture/SPEC/knowledge_service.md | 6 +- doc/features/knowledge-service.md | 5 +- .../kb_service/external.py | 11 +- .../kb_service/local_gate.py | 107 ++++++++ src/infermatrix_copilot/kb_service/merge.py | 221 ++++++++++------ src/infermatrix_copilot/kb_service/outbox.py | 4 +- .../kb_service/publisher.py | 235 ++++++++++++++++- .../knowledge_service/gate_verifier.py | 13 +- test/test_kb_external.py | 216 ++-------------- test/test_kb_flow.py | 132 +++++++--- test/test_kb_publisher.py | 242 ++++++++++++++++-- test/test_kb_rebuild.py | 136 ++-------- 15 files changed, 907 insertions(+), 459 deletions(-) create mode 100644 src/infermatrix_copilot/kb_service/local_gate.py 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]