Confirm pauses on GitHub and alert people when they are not

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` 提供 `<state_dir>/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 <n> --undo` 命令。`kb status` 报告每个仓库 `pause_unconfirmed` 数量，熔断/回滚在其归零
+前不算完成；`kb resume` 清除这些记录。

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()
                 ],
             }

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
 

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