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