From 7721ef538e7e0ecf36fb8b902f10e33c0dcdadf9 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 29 Sep 2026 00:25:08 +0800 Subject: [PATCH] Confirm pauses on GitHub and alert people when they are not The design's procedure P counts a pause only once it is observed: the service must see the PR out of the queue and a draft. A pause not confirmed within 5 minutes goes to people with the commands to do it by hand. Until now an ack alone marked a PR paused, and a pause that never ran went unnoticed. - pause_open_prs records each pause request's time and reason. - A "paused" change set (acked) is confirmed only when GitHub shows the PR as a draft (a draft cannot be in, or re-enter, the merge queue). Otherwise the pause is issued again, and a confirmed PR observed un-drafted later counts as unconfirmed again (with a fresh alert clock). Pause tracking runs before head checks, so a paused PR whose head moves stays watched. A completed re-pause (its ack, or a confirmation) clears the pending item, so a later undo is re-paused at once instead of waiting out the old item. - A request neither acked nor confirmed 5 minutes later goes to people once, with "remove it from the merge queue" and `gh pr ready --undo`. The escalation does not depend on GitHub answering: if reading a PR fails, only that change set is skipped, and an overdue pause still goes to people. - kb status reports pause_unconfirmed per repository: a breaker or rollback is not complete while it is non-zero. kb resume clears the bookkeeping, also for requests that were never acked. Co-Authored-By: Claude Opus 5.5 --- doc/architecture/SPEC/kb_service.md | 6 + src/infermatrix_copilot/kb_service/cli.py | 10 +- src/infermatrix_copilot/kb_service/merge.py | 87 ++++++++++-- test/test_kb_pause_confirm.py | 141 ++++++++++++++++++++ 4 files changed, 235 insertions(+), 9 deletions(-) create mode 100644 test/test_kb_pause_confirm.py diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 1c951cdf..1e7c5621 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -140,3 +140,9 @@ stdlib + PyYAML + `cryptography`（`kb` extra）+ `.adapters` + `.knowledge_serv `holds_server`（`kb holds-server`）只以 `GET/HEAD /holds.json` 提供 `/public/holds.json`（`Cache-Control: no-store`）， 其余路径一律 404，不列目录、不跟随路径、不接受写入；清单尚不存在时返回 503（门禁视为不可达并失败关闭）。默认只绑定 `127.0.0.1`，由 bot 主机的反向代理加 TLS 后对外，URL 写入 `.github/kb-gate/config.json` 的 `holds_url`。 + +## 2026-09-29 暂停以观测确认 +`pause_open_prs` 记录每个暂停请求的时间与原因。回执只代表发布器的说法：状态 `paused` 的变更集只有在 GitHub 显示该 PR +为 draft 时才记为已确认（draft 不能在合并队列中，也不能再次入队）；否则重发 `pause`。请求 5 分钟后仍未回执或未确认 → +转人工一次，附出队与 `gh pr ready --undo` 命令。`kb status` 报告每个仓库 `pause_unconfirmed` 数量，熔断/回滚在其归零 +前不算完成；`kb resume` 清除这些记录。 diff --git a/src/infermatrix_copilot/kb_service/cli.py b/src/infermatrix_copilot/kb_service/cli.py index 72247e98..d069d640 100644 --- a/src/infermatrix_copilot/kb_service/cli.py +++ b/src/infermatrix_copilot/kb_service/cli.py @@ -85,6 +85,12 @@ def _public(registry) -> set[str]: return {name for name, lifecycle in registry.items() if lifecycle.publishes} +def _unconfirmed(ledger, repo: str) -> int: + from .merge import pause_unconfirmed + + return pause_unconfirmed(ledger, repo) + + def _refresh(state_dir: Path, ledger, registry) -> None: _outbox(state_dir, ledger).transition(lambda: None, public_repos=_public(registry)) @@ -170,7 +176,9 @@ def main(argv: list[str] | None = None) -> int: {**row, "configured": row["repo"] in registry or row["repo"] == "*", "open_changesets": len(ledger.changesets(row["repo"], ("open", "pr_open", "signed", "queued"))) if row["repo"] != "*" else None, - "human_queue": len(ledger.human_queue(row["repo"])) if row["repo"] != "*" else None} + "human_queue": len(ledger.human_queue(row["repo"])) if row["repo"] != "*" else None, + # a breaker/rollback is not done while any of these remain + "pause_unconfirmed": _unconfirmed(ledger, row["repo"]) if row["repo"] != "*" else None} for row in ledger.all_repo_states() ], } diff --git a/src/infermatrix_copilot/kb_service/merge.py b/src/infermatrix_copilot/kb_service/merge.py index 6da6346d..60776ead 100644 --- a/src/infermatrix_copilot/kb_service/merge.py +++ b/src/infermatrix_copilot/kb_service/merge.py @@ -29,6 +29,8 @@ RESIGN_AFTER = 48 * 3600 # margin for the publisher's clock when deciding an item has certainly expired CLOCK_SKEW = 5 * 60 +# a pause not confirmed on GitHub by then goes to people (procedure P, step 4) +PAUSE_CONFIRM = 5 * 60 # a queued PR that neither merged nor left the queue by then was dropped by a # failing merge group (GitHub does not tell us): re-sign and re-enqueue QUEUE_STALL = 2 * 3600 @@ -93,6 +95,11 @@ def apply_acks(rt, acks: list[dict]) -> None: # cannot slip between reading the pause state and recording the ack with rt.outbox.publication_lock(): current = rt.ledger.changeset(changeset_id) + if ack.get("ok") and current["status"] == "paused": + # a re-pause retry completed: nothing is pending any more + if (current.get("pending_item") or {}).get("id") == ack.get("item_id"): + rt.ledger.update_changeset(changeset_id, pending_item=None) + continue if not ack.get("ok") or current["status"] not in ("pr_open", "verdict_posted", "queued"): continue if current["detail"].get("withdrawn"): @@ -325,11 +332,18 @@ def advance(rt, lifecycle) -> list[str]: 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")] - for changeset in rt.ledger.changesets(lifecycle.repo, (*IN_FLIGHT, "superseding")): + for changeset in rt.ledger.changesets(lifecycle.repo, (*IN_FLIGHT, "superseding", "paused")): number = changeset["pr_number"] if number is None: continue - pr: dict[str, Any] = _pr_state(rt, int(number)) + try: + pr: dict[str, Any] = _pr_state(rt, int(number)) + except Exception: # noqa: BLE001 - GitHub down or rate-limited + # the overdue-pause escalation must not depend on GitHub answering + if changeset["status"] == "paused" or paused: + _pause_unconfirmed(rt, lifecycle.repo, changeset, events) + events.append(f"observe_failed {changeset['id']}") + continue if changeset["status"] == "superseding": # the replaced PR stays tracked until GitHub shows it closed; the # close item is re-issued whenever the previous one expired @@ -351,6 +365,32 @@ def advance(rt, lifecycle) -> list[str]: _outcome(rt, changeset, "closed_unmerged") # overturned by people: a negative label events.append(f"closed {changeset['id']}") continue + # pause tracking comes before head checks: a paused PR stays watched + if changeset["status"] == "paused": + # an ack is the publisher's word; the pause counts once GitHub shows + # the PR as a draft (a draft cannot be in, or re-enter, the queue) + if pr.get("draft"): + if not changeset["detail"].get("pause_confirmed"): + # a pending retry is done with: a later undo gets a fresh pause + rt.ledger.update_changeset(changeset["id"], pending_item=None, detail={ + **changeset["detail"], "pause_confirmed": True}) + events.append(f"pause_confirmed {changeset['id']}") + else: + if changeset["detail"].get("pause_confirmed"): # un-drafted since: not paused any more + rt.ledger.update_changeset(changeset["id"], pending_item=None, detail={ + **changeset["detail"], "pause_confirmed": False, "pause_alerted": False, + "pause_requested_at": rt.clock()}) + changeset = rt.ledger.changeset(changeset["id"]) + issue_once(rt, lifecycle.repo, changeset, "pause", { + "changeset_id": changeset["id"], "pr": int(number), + "reason": "re-pausing: GitHub does not show the PR as a draft"}) + _pause_unconfirmed(rt, lifecycle.repo, changeset, events) + continue + if paused: + # the pause itself is issued by `kb pause` / breakers; one that has + # not even been acked in time goes to people + _pause_unconfirmed(rt, lifecycle.repo, changeset, events) + continue head = str((pr.get("head") or {}).get("sha") or "") if head != changeset["head_sha"] and changeset["kind"] == "external": # authors push: the new head is simply judged again at the next poll @@ -363,8 +403,6 @@ def advance(rt, lifecycle) -> list[str]: rt.ledger.enqueue_human(lifecycle.repo, f"PR #{number} head changed after signing", changeset["id"]) events.append(f"head_changed {changeset['id']}") continue - if paused: - continue # the pause itself is issued by `kb pause` / breakers if changeset["kind"] == "external" and changeset["detail"].get("source") == "human-approved" \ and changeset["status"] in ("verdict_posted", "queued"): from .external import approval_stands @@ -488,29 +526,62 @@ def record_retirements(rt, changeset: dict) -> None: def pause_open_prs(ledger, outbox, repo: str, reason: str) -> int: """Issue always-executable pause items (dequeue + draft) for every open - knowledge PR of ``repo``; they are marked paused once the publisher acks.""" + knowledge PR of ``repo``; they are marked paused once the publisher acks, + and confirmed once GitHub shows them as drafts.""" count = 0 if outbox is None: return count for changeset in ledger.changesets(repo, ("pr_open", "verdict_posted", "queued")): if changeset["pr_number"] is None: continue - outbox.issue(repo, "pause", {"changeset_id": changeset["id"], "pr": int(changeset["pr_number"]), - "reason": reason}) + item = outbox.issue(repo, "pause", {"changeset_id": changeset["id"], "pr": int(changeset["pr_number"]), + "reason": reason}) + ledger.update_changeset(changeset["id"], detail={ + **changeset["detail"], "pause_requested_at": item.issued_at, "pause_reason": reason}) count += 1 return count +def _pause_unconfirmed(rt, repo: str, changeset: dict, events: list[str]) -> None: + """Alert people once when a requested pause is not confirmed in time.""" + detail = changeset["detail"] + requested = float(detail.get("pause_requested_at") or 0) + if not requested or detail.get("pause_alerted") or detail.get("pause_confirmed") \ + or rt.clock() - requested < PAUSE_CONFIRM: + return + number = changeset["pr_number"] + rt.ledger.update_changeset(changeset["id"], detail={**detail, "pause_alerted": True}) + rt.ledger.enqueue_human(repo, ( + f"pause of PR #{number} not confirmed after {PAUSE_CONFIRM // 60} minutes " + f"({detail.get('pause_reason', '')}): remove it from the merge queue and run " + f"`gh pr ready {number} --undo -R {knowledge_repository()}`"), changeset["id"]) + events.append(f"pause_unconfirmed {changeset['id']}") + + +def pause_unconfirmed(ledger, repo: str) -> int: + """How many of ``repo``'s PRs are paused on paper but not yet on GitHub.""" + pending = [cs for cs in ledger.changesets(repo, ("pr_open", "verdict_posted", "queued")) + if cs["detail"].get("pause_requested_at")] + acked = [cs for cs in ledger.changesets(repo, ("paused",)) if not cs["detail"].get("pause_confirmed")] + return len(pending) + len(acked) + + def resume_paused_prs(ledger, repo: str) -> int: """After ``kb resume``: paused knowledge PRs go back to ``pr_open``. The resumed repository has a NEW generation, so the next pass signs a fresh verdict for the PR's current head and re-enqueues (the publisher's enqueue marks the draft PR ready first); a head that moved goes to people.""" count = 0 + tracking = ("pause_requested_at", "pause_reason", "pause_alerted", "pause_confirmed") + for changeset in ledger.changesets(repo, ("pr_open", "verdict_posted", "queued")): + if any(key in changeset["detail"] for key in tracking): # requested but never acked + ledger.update_changeset(changeset["id"], detail={ + k: v for k, v in changeset["detail"].items() if k not in tracking}) for changeset in ledger.changesets(repo, ("paused",)): if changeset["pr_number"] is None: continue - ledger.update_changeset(changeset["id"], status="pr_open", pending_item=None) + detail = {k: v for k, v in changeset["detail"].items() if k not in tracking} + ledger.update_changeset(changeset["id"], status="pr_open", pending_item=None, detail=detail) count += 1 return count diff --git a/test/test_kb_pause_confirm.py b/test/test_kb_pause_confirm.py new file mode 100644 index 00000000..4de1c689 --- /dev/null +++ b/test/test_kb_pause_confirm.py @@ -0,0 +1,141 @@ +"""A pause counts once GitHub shows it; unconfirmed pauses go to people.""" + +from __future__ import annotations + +from infermatrix_copilot.kb_service import merge +from test_kb_flow import _ack, _flow_runtime, _items, _open_pr + +HEAD = "d" * 40 + + +def _paused_repo(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=HEAD) + rt.github.prs[42] = {"state": "open", "merged": False, "draft": False, "head": {"sha": HEAD}} + rt.outbox.transition(lambda: rt.ledger.bump_generation(lifecycle.repo, pause=True, reason="drill"), + public_repos={lifecycle.repo}) + assert merge.pause_open_prs(rt.ledger, rt.outbox, lifecycle.repo, "drill") == 1 + return rt, lifecycle, changeset_id, publisher + + +def _later(rt, seconds): + start = rt.clock() + rt.clock = lambda: start + seconds + + +def test_an_unacked_pause_goes_to_people_once_after_five_minutes(tmp_path): + rt, lifecycle, changeset_id, _ = _paused_repo(tmp_path) + assert merge.advance(rt, lifecycle) == [] + assert merge.pause_unconfirmed(rt.ledger, "demo") == 1 + _later(rt, merge.PAUSE_CONFIRM + 1) + assert merge.advance(rt, lifecycle) == [f"pause_unconfirmed {changeset_id}"] + (alert,) = [item for item in rt.ledger.human_queue("demo") if "not confirmed" in item["reason"]] + assert "gh pr ready 42 --undo" in alert["reason"] and "drill" in alert["reason"] + assert merge.advance(rt, lifecycle) == [] # once + + +def test_an_acked_pause_is_confirmed_only_when_github_shows_a_draft(tmp_path): + rt, lifecycle, changeset_id, publisher = _paused_repo(tmp_path) + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset_id, ok=True) + assert rt.ledger.changeset(changeset_id)["status"] == "paused" + pauses = len(_items(tmp_path, "pause")) + assert merge.advance(rt, lifecycle) == [] # not a draft on GitHub: pause again + assert len(_items(tmp_path, "pause")) == pauses + 1 + assert merge.pause_unconfirmed(rt.ledger, "demo") == 1 + _later(rt, merge.PAUSE_CONFIRM + 1) + assert merge.advance(rt, lifecycle) == [f"pause_unconfirmed {changeset_id}"] + rt.github.prs[42]["draft"] = True + assert merge.advance(rt, lifecycle) == [f"pause_confirmed {changeset_id}"] + assert merge.pause_unconfirmed(rt.ledger, "demo") == 0 + assert merge.advance(rt, lifecycle) == [] + + +def test_resume_clears_the_pause_bookkeeping(tmp_path): + rt, lifecycle, changeset_id, publisher = _paused_repo(tmp_path) + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset_id, ok=True) + rt.github.prs[42]["draft"] = True + merge.advance(rt, lifecycle) + rt.ledger.resume("demo") + merge.resume_paused_prs(rt.ledger, "demo") + detail = rt.ledger.changeset(changeset_id)["detail"] + assert not {"pause_requested_at", "pause_alerted", "pause_confirmed"} & set(detail) + assert merge.pause_unconfirmed(rt.ledger, "demo") == 0 + + +def test_kb_status_reports_unconfirmed_pauses(tmp_path, monkeypatch, capsys): + import json + + from infermatrix_copilot.kb_service import cli + + rt, lifecycle, changeset_id, _ = _paused_repo(tmp_path) + monkeypatch.setattr(cli, "_registry", lambda: rt.registry) + monkeypatch.delenv("KB_SIGNING_KEY", raising=False) + assert cli.main(["--state-dir", str(tmp_path / "state"), "status"]) == 0 + report = json.loads(capsys.readouterr().out) + (demo,) = [row for row in report["repos"] if row["repo"] == "demo"] + assert demo["pause_unconfirmed"] == 1 + + +def test_a_confirmed_pause_that_is_undone_counts_as_unconfirmed_again(tmp_path): + rt, lifecycle, changeset_id, publisher = _paused_repo(tmp_path) + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset_id, ok=True) + rt.github.prs[42]["draft"] = True + merge.advance(rt, lifecycle) + assert merge.pause_unconfirmed(rt.ledger, "demo") == 0 + rt.github.prs[42]["draft"] = False # someone marked it ready again + merge.advance(rt, lifecycle) + assert merge.pause_unconfirmed(rt.ledger, "demo") == 1 + _later(rt, merge.PAUSE_CONFIRM + 1) + assert merge.advance(rt, lifecycle) == [f"pause_unconfirmed {changeset_id}"] + + +def test_resume_before_the_ack_clears_the_request(tmp_path): + rt, lifecycle, changeset_id, _ = _paused_repo(tmp_path) + rt.ledger.resume("demo") + merge.resume_paused_prs(rt.ledger, "demo") + assert merge.pause_unconfirmed(rt.ledger, "demo") == 0 + assert "pause_requested_at" not in rt.ledger.changeset(changeset_id)["detail"] + + +def test_a_paused_pr_whose_head_moves_stays_watched(tmp_path): + rt, lifecycle, changeset_id, publisher = _paused_repo(tmp_path) + rt.github.prs[42]["head"]["sha"] = "e" * 40 + _later(rt, merge.PAUSE_CONFIRM + 1) + assert merge.advance(rt, lifecycle) == [f"pause_unconfirmed {changeset_id}"] + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset_id, ok=True) + assert merge.pause_unconfirmed(rt.ledger, "demo") == 1 + rt.github.prs[42]["draft"] = True + assert merge.advance(rt, lifecycle) == [f"pause_confirmed {changeset_id}"] + + +def test_a_completed_retry_never_blocks_the_next_re_pause(tmp_path): + rt, lifecycle, changeset_id, publisher = _paused_repo(tmp_path) + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset_id, ok=True) + merge.advance(rt, lifecycle) # not a draft: a retry is issued + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset_id, ok=True) # the retry ran + assert rt.ledger.changeset(changeset_id)["pending_item"] is None + rt.github.prs[42]["draft"] = True + merge.advance(rt, lifecycle) # confirmed + rt.github.prs[42]["draft"] = False # ready again + before = len(_items(tmp_path, "pause")) + merge.advance(rt, lifecycle) + assert len(_items(tmp_path, "pause")) == before + 1 # a fresh pause at once + + +def test_overdue_pauses_escalate_even_when_github_cannot_be_read(tmp_path): + rt, lifecycle, changeset_id, publisher = _paused_repo(tmp_path) + + def down(url): + raise RuntimeError("GitHub 503") + + rt.github._fetch = down + _later(rt, merge.PAUSE_CONFIRM + 1) + events = merge.advance(rt, lifecycle) # unacked pause, GitHub down + assert f"pause_unconfirmed {changeset_id}" in events and f"observe_failed {changeset_id}" in events + + rt2, lifecycle2, changeset2, publisher2 = _paused_repo(tmp_path / "acked") + _ack(tmp_path / "acked", rt2, publisher2, kind="pause", changeset_id=changeset2, ok=True) + rt2.github._fetch = down + _later(rt2, merge.PAUSE_CONFIRM + 1) + assert f"pause_unconfirmed {changeset2}" in merge.advance(rt2, lifecycle2) # acked, never confirmed