diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md
index 2c5801b5..eb6c8f22 100644
--- a/doc/architecture/SPEC/kb_service.md
+++ b/doc/architecture/SPEC/kb_service.md
@@ -244,3 +244,17 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝
 - 测试：`test_kb_local_gate_verifier.py`（取代 `test_kb_gate_verifier.py`；本地门禁上的重放/过期/伪造/非 auto 来源、清单与块表、
   混合路径与可执行位、跨仓库、上下文失效、落地树中的悬空引用）；`test_kb_publisher.py`（暂停不触碰 PR、恢复后重签合并、
   draft 不合并、`close` 重试）；`test_kb_external.py`（受治理页面以外的路径从不合并）。
+
+## 2026-09-29 v8：上游事实证明与发布器复核
+- 服务：`run_gate(facts=observer)` 在 L1 之后、L2 之前对变更涉及的规则调用 `facts.attest`（观测者为
+  `upstream_facts.MirrorObserver`：服务自己的上游裸镜像 `<state_dir>/upstream/<repo>.git`（与发版巡检共用）+ 只读
+  GitHub reader）。必需声明不成立 → `fail`（理由 `upstream fact: …`，不调用付费评审，进入精炼）；上游不可读 → `human`。
+  事实与钉住的 `upstream` 写入 `decision`，`sign_verdict` 签入判定。没有上游（`general`）或不发布的仓库不做证明。
+  `KbRuntime.upstream_facts(lifecycle)` 提供观测者（`from_env` 装配；测试默认不证明）。
+- 发布器：`merge` 在本地门禁通过后、最后一次读控制记录前调用 `facts.recheck`，数据来源独立：发布器自己的镜像
+  `<publisher_state>/upstream/<repo>.git` 与 `gh api repos/<upstream>/pulls/<n>`；上游来自**发布器自己的**适配器配置
+  （`upstreams`，只含启用且发布的仓库）。任何不一致、未配置公开上游或上游不可读 → 本轮不合并（`not merging: upstream facts: …`）。
+- 服务收到这类拒绝：回到 `pr_open` 下一轮重签，记录 `facts_failing_since`；连续 24 小时仍失败 → `gate_failed` 并转人工。
+  其他任何回执结束这段连续失败。
+- 测试：`test_kb_upstream_facts.py`（声明抽取、符号定义、证明与复核的一致/不一致/不可达/伪造、真实 git 镜像、
+  门禁在 L2 前失败、发布器一致时合并、不一致或不可达时不合并并在 24 小时后转人工、未配置上游不合并）。
diff --git a/doc/architecture/SPEC/knowledge_service.md b/doc/architecture/SPEC/knowledge_service.md
index c7931209..e33bc2f9 100644
--- a/doc/architecture/SPEC/knowledge_service.md
+++ b/doc/architecture/SPEC/knowledge_service.md
@@ -63,3 +63,12 @@ merge group 逐段验证、暂停清单与 `human-approved` 审批核对已于 2
 ## 2026-09-29 verify_change 供本地门禁调用
 `verify_change` 只接受调用方已验签与绑定的判定（`verdict=`），不再从 PR 评论查找判定，也不读暂停清单；
 暂停由签名控制记录表达。
+
+## 2026-09-29 上游事实（facts）
+`facts` 从变更涉及的规则中抽取可解析的上游声明：`^[PR #N]` 引用（PR 已合并）、反引号中首段为上游仓库顶层条目的路径
+（文件或目录存在）与 `path::Symbol`（文件中以 def/class/赋值定义了每个点分量）；首段不是顶层条目的片段路径、`..` 与
+`...` 段不算声明。`active` 规则的声明必须成立；退役规则的声明只按观测记录，但退役证据中的 PR 必须已合并。
+`attest(claims, observer)` 在上游当前 head 上观测，返回 `upstream={repository, sha}`、事实列表与问题；超过 200 条声明拒绝。
+`recheck(upstream, facts, observer)` 由发布器调用：仓库一致、SHA 为 40 位、每条事实重新观测后逐字段相等，
+签名了不成立的必需事实或格式错误均为问题；上游不可读（未知 SHA、网络、API 错误）抛 `FactsError`，由调用方重试。
+仅标准库。`gate_verifier.verify_change` 不再拒绝带事实的判定（由发布器的 `recheck` 复核）。
diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md
index 717a5d41..208de2c1 100644
--- a/doc/features/knowledge-service.md
+++ b/doc/features/knowledge-service.md
@@ -81,6 +81,8 @@ T1/T2/T3 与 purge，每个规则页一个变更集，全部经过质量门。
 逐块与一致性判定、上下文未变），通过才合并，合并后再检查落地的提交。只接受 `auto` 判定，
 没有 `human-approved` 路径：改动受治理页面以外路径的知识 PR 不会被自动合并，作者会在发现评论里被要求拆分。
 绕过发布器直接合并或推送的知识变更由每轮审计发现、暂停并撤回（见上文）。
+规则对上游的声明（引用的 PR 已合并、写全的上游路径与 `path::Symbol` 存在）在质量门中对钉住的上游 SHA 证明并签入判定；
+发布器合并前用自己的上游镜像与 `gh api` 逐条复核，不一致或上游不可读则不合并，连续 24 小时失败转人工。
 
 切换前的负责人步骤：`main` 必须允许发布器直接合并（不开合并队列、不开自动合并）；
 按设计 §P3 的顺序关闭 v7 遗留的自动合并并卸载 v7 发布器。
diff --git a/src/infermatrix_copilot/kb_service/cli.py b/src/infermatrix_copilot/kb_service/cli.py
index 5d1bf890..ba2aee27 100644
--- a/src/infermatrix_copilot/kb_service/cli.py
+++ b/src/infermatrix_copilot/kb_service/cli.py
@@ -259,6 +259,8 @@ def _publish(args) -> int:
         state_dir=state,
         author=(author.group(1), author.group(2)),
         repo_flags={name: (lc.publishes, lc.auto_merge) for name, lc in registry.items() if lc.enabled},
+        upstreams={name: lc.full_name for name, lc in registry.items()
+                   if lc.enabled and lc.publishes and lc.full_name},
         allow_post=os.environ.get("ALLOW_POST") == "1",
         allow_push=os.environ.get("ALLOW_PUSH") == "1",
     )
diff --git a/src/infermatrix_copilot/kb_service/external.py b/src/infermatrix_copilot/kb_service/external.py
index d75b2a5d..ea174830 100644
--- a/src/infermatrix_copilot/kb_service/external.py
+++ b/src/infermatrix_copilot/kb_service/external.py
@@ -286,7 +286,8 @@ def _poll(rt, owner: str, *, force: bool = False) -> list[str]:
                             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)
+                            retire_ratio=lifecycle.retire_ratio, max_files=lifecycle.max_files,
+                            facts=rt.upstream_facts(lifecycle))
         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]
diff --git a/src/infermatrix_copilot/kb_service/gate.py b/src/infermatrix_copilot/kb_service/gate.py
index e960df95..8a1ffb65 100644
--- a/src/infermatrix_copilot/kb_service/gate.py
+++ b/src/infermatrix_copilot/kb_service/gate.py
@@ -78,6 +78,8 @@ class GateDecision:
     l1: ChangesetResult
     blocks: list[BlockVerdict] = field(default_factory=list)
     consistency: list[dict] = field(default_factory=list)
+    upstream: dict = field(default_factory=dict)   # {repository, sha} the facts were observed on
+    facts: list[dict] = field(default_factory=list)
 
     def to_dict(self) -> dict:
         return {
@@ -86,6 +88,7 @@ def to_dict(self) -> dict:
             "external_refs": [list(ref) for ref in self.l1.external_refs],
             "blocks": [b.to_dict() for b in self.blocks],
             "consistency": self.consistency,
+            "upstream": self.upstream, "facts": self.facts,
         }
 
 
@@ -210,11 +213,26 @@ def run_gate(*, base: dict[str, str], head: dict[str, str], changes: list[Change
              external_texts: dict[str, str], evidence: list[dict], gateway: ModelGateway,
              judge: ModelRole, release: str, repo_dir: str,
              protected_rules: tuple[str, ...] = (),
-             retire_ratio: float = 0.10, max_files: int = 50) -> GateDecision:
+             retire_ratio: float = 0.10, max_files: int = 50, facts=None) -> GateDecision:
+    """``facts`` observes the repository's public upstream (None: the
+    repository has no upstream to attest, or it publishes nothing)."""
     l1 = check_changeset(base, head, changes, external_texts=external_texts, release=release)
     reasons: list[str] = []
     if not l1.ok:
         return GateDecision("fail", [f"L1: {i.code} {i.path} {i.detail}" for i in l1.issues], l1)
+    upstream, signed_facts = {}, []
+    if facts is not None:
+        from ..knowledge_service.facts import FactsError, attest
+        from .upstream_facts import change_claims
+
+        try:
+            claims = change_claims(head, l1.blocks, facts.top_level(facts.head()))
+            upstream, signed_facts, problems = attest(claims, facts) if claims else ({}, [], [])
+        except FactsError as exc:
+            return GateDecision("human", [f"upstream facts could not be checked: {exc}"], l1)
+        if problems:
+            return GateDecision("fail", [f"upstream fact: {p}" for p in problems], l1,
+                                upstream=upstream, facts=signed_facts)
     if l1.external_refs:
         reasons.append("references outside knowledge/ need a companion PR: "
                        + ", ".join(f"{rid}@{path}" for rid, path in l1.external_refs))
@@ -234,17 +252,20 @@ def run_gate(*, base: dict[str, str], head: dict[str, str], changes: list[Change
               for b in l1.blocks]
     if any(b.verdict == "fail" for b in blocks):
         return GateDecision("fail", reasons + [f"L2 rejected {b.block.rule_id or b.block.path}"
-                                               for b in blocks if b.verdict == "fail"], l1, blocks)
+                                               for b in blocks if b.verdict == "fail"], l1, blocks,
+                            upstream=upstream, facts=signed_facts)
     owner_dirs = sorted({_owner_dir(b.path) for b in l1.blocks})
     consistency = check_consistency(head, owner_dirs, gateway=gateway, judge=judge)
     if any(item["verdict"] == "conflict" for item in consistency):
-        return GateDecision("fail", reasons + ["change set is inconsistent"], l1, blocks, consistency)
+        return GateDecision("fail", reasons + ["change set is inconsistent"], l1, blocks, consistency,
+                            upstream, signed_facts)
     if any(b.verdict == "human" for b in blocks):
         reasons.append("L2 was not sure about " + ", ".join(
             b.block.rule_id or b.block.path for b in blocks if b.verdict == "human"))
     if any(item["verdict"] == "unsure" for item in consistency):
         reasons.append("consistency check was not sure")
-    return GateDecision("human" if reasons else "pass", reasons, l1, blocks, consistency)
+    return GateDecision("human" if reasons else "pass", reasons, l1, blocks, consistency,
+                        upstream, signed_facts)
 
 
 def _is_active(section) -> bool:
diff --git a/src/infermatrix_copilot/kb_service/merge.py b/src/infermatrix_copilot/kb_service/merge.py
index 41b6ecbf..f74aecb0 100644
--- a/src/infermatrix_copilot/kb_service/merge.py
+++ b/src/infermatrix_copilot/kb_service/merge.py
@@ -32,6 +32,10 @@
 IN_FLIGHT = ("pr_requested", "pr_open", "merge_requested", "rebuild_needed")
 # margin for the publisher's clock when deciding an item has certainly expired
 CLOCK_SKEW = 5 * 60
+# the publisher could not confirm the signed upstream facts: signed again and
+# retried each round, to people once it kept failing this long (design v8 §8.2)
+FACTS_MERGE = "upstream facts:"
+FACTS_ESCALATE_AFTER = 24 * 3600
 # local-gate refusals the service answers by rebuilding on current main
 REBUILD_GATE = ("context changed since the verdict", "pages changed since the consistency judgement",
                 "does not merge cleanly", "signed patch manifest does not match")
@@ -194,6 +198,22 @@ def _apply_merge_ack(rt, changeset: dict, ack: dict) -> None:
                            ack.get("problems") or [])
         return
     text = " ".join([str(ack.get("error") or ""), *map(str, ack.get("problems") or [])])
+    detail = {k: v for k, v in changeset["detail"].items() if k != "facts_failing_since"}
+    if FACTS_MERGE in text:
+        since = float(changeset["detail"].get("facts_failing_since") or rt.clock())
+        if rt.clock() - since >= FACTS_ESCALATE_AFTER:
+            rt.ledger.update_changeset(changeset_id, status="gate_failed", pending_item=None,
+                                       detail={**detail, "gate_problems": [text[:400]]})
+            rt.ledger.enqueue_human(repo, f"PR #{changeset['pr_number']}: the publisher could not confirm its "
+                                          f"upstream facts for 24 hours: {text[:300]}", changeset_id)
+            _outcome(rt, changeset, "gate_failed", problems=[text[:400]])
+        else:
+            rt.ledger.update_changeset(changeset_id, status="pr_open", pending_item=None,
+                                       detail={**detail, "facts_failing_since": since})
+        return
+    if detail != changeset["detail"]:  # any other answer ends a run of fact failures
+        rt.ledger.update_changeset(changeset_id, detail=detail)
+        changeset = rt.ledger.changeset(changeset_id)
     if any(m in text for m in REBUILD_GATE):
         rt.ledger.update_changeset(changeset_id, status="rebuild_needed", pending_item=None,
                                    detail={**changeset["detail"], "rebuild_because": text[:500]})
@@ -243,7 +263,7 @@ def sign_verdict(rt, changeset: dict) -> dict:
     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,
-        release=detail["release"], upstream={}, facts=[],
+        release=detail["release"], upstream=decision.get("upstream") or {}, facts=decision.get("facts") or [],
         models={"generator": detail["generator"], "judge": detail["judge"]},
         source="auto", issued_at=rt.clock(),
     )
diff --git a/src/infermatrix_copilot/kb_service/publisher.py b/src/infermatrix_copilot/kb_service/publisher.py
index eab9607c..70bc0d37 100644
--- a/src/infermatrix_copilot/kb_service/publisher.py
+++ b/src/infermatrix_copilot/kb_service/publisher.py
@@ -199,6 +199,8 @@ class Publisher:
     git_remote: str = ""                         # default https://github.com/<repository>.git
     clock: Callable[[], float] = time.time
     log: list[dict] = field(default_factory=list)
+    upstreams: Mapping[str, str] = field(default_factory=dict)  # repo -> public upstream, from OUR config
+    observer: Callable[[str, str], Any] | None = None           # (repo, upstream) -> facts observer
 
     # local records -------------------------------------------------------------
     def _done_path(self, item_id: str) -> Path:
@@ -524,6 +526,32 @@ def _recheck_control(self) -> None:
         except (OutboxError, OSError, ValueError, KeyError, TypeError) as exc:
             raise PublishError(f"not merging: {exc}") from exc
 
+    def _recheck_facts(self, repo: str, verdict: dict) -> None:
+        """Observe every signed upstream fact again from the upstream itself (our
+        own mirror and ``gh api``, never the service's). Any difference, or an
+        upstream we cannot read, means no merge this round (design v8 §8.2)."""
+        from ..knowledge_service.facts import FactsError, recheck
+
+        facts = verdict.get("facts") or []
+        if not facts:
+            return
+        upstream = self.upstreams.get(repo, "")
+        if not upstream:
+            raise PublishError(f"not merging: upstream facts: {repo} has no public upstream in our configuration")
+        observer = (self.observer or self._mirror_observer)(repo, upstream)
+        try:
+            problems = recheck(verdict.get("upstream") or {}, facts, observer)
+        except FactsError as exc:
+            raise PublishError(f"not merging: upstream facts: {exc}") from exc
+        if problems:
+            raise PublishError("not merging: upstream facts: " + "; ".join(problems[:5]))
+
+    def _mirror_observer(self, repo: str, upstream: str):
+        from .upstream_facts import MirrorObserver
+
+        return MirrorObserver(self.state_dir / "upstream" / f"{repo}.git", upstream,
+                              lambda number: json.loads(self.github.gh("api", f"repos/{upstream}/pulls/{number}")))
+
     def _do_merge(self, item: OutboxItem) -> dict:
         """The local gate on the exact merged tree, then the merge (design v8 §8.2)."""
         from . import local_gate
@@ -566,6 +594,7 @@ def _do_merge(self, item: OutboxItem) -> dict:
                 break
         else:
             raise PublishError(f"main kept moving while PR #{number} was checked; retried next round")
+        self._recheck_facts(item.repo, verdict)
         self._recheck_control()
         intent = self._intent_path(item.id)
         intent.parent.mkdir(parents=True, exist_ok=True)
diff --git a/src/infermatrix_copilot/kb_service/runtime.py b/src/infermatrix_copilot/kb_service/runtime.py
index dc6ef3c1..0ba04bfb 100644
--- a/src/infermatrix_copilot/kb_service/runtime.py
+++ b/src/infermatrix_copilot/kb_service/runtime.py
@@ -46,6 +46,8 @@ class KbRuntime:
     publisher_public_key: object | None = None
     traces: object | None = None  # trace_store.TraceStore: decisions and outcomes (model calls via the gateway)
     release_label: Callable[[str], str] = field(default=lambda repo: "")
+    # the repository's upstream as the gate observes it (None: no facts attested)
+    upstream_facts: Callable[[RepoLifecycle], object | None] = field(default=lambda lifecycle: None)
 
     @classmethod
     def from_env(cls, settings, *, state_dir: Path | None = None) -> "KbRuntime":
@@ -74,13 +76,15 @@ def from_env(cls, settings, *, state_dir: Path | None = None) -> "KbRuntime":
         from ..trace_store import TraceStore
 
         traces = TraceStore(state_dir / "traces")
+        github = GitHubReader()
         return cls(
+            upstream_facts=lambda lifecycle: service_observer(state_dir, lifecycle, github),
             publisher_public_key=publisher_key, traces=traces,
             state_dir=state_dir, ledger=ledger, registry=registry,
             gateway=ModelGateway(settings, recorder=trace_recorder(traces)),
             generator=generator, judge=judge,
             knowledge=KnowledgeRepo(Path(os.environ.get("KB_KNOWLEDGE_CLONE") or state_dir / "knowledge-repo")),
-            github=GitHubReader(), outbox=outbox,
+            github=github, outbox=outbox,
         )
 
     # -- helpers -------------------------------------------------------------
@@ -296,7 +300,8 @@ def gate_and_stage(rt: KbRuntime, lifecycle: RepoLifecycle, owner: str, *, kind:
             base=base, head=head, changes=changes_between(base, head), external_texts=external,
             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)
+            retire_ratio=lifecycle.retire_ratio, max_files=lifecycle.max_files,
+            facts=rt.upstream_facts(lifecycle))
     if force_human and decision.status == "pass":
         decision.status = "human"
         decision.reasons.append(force_human)
@@ -392,6 +397,19 @@ def _pr_body(changeset_id: str, detail: dict) -> str:
     return "\n".join(lines)
 
 
+def service_observer(state_dir: Path, lifecycle: RepoLifecycle, github: GitHubReader):
+    """The service's view of a public upstream: its own mirror (shared with the
+    release sweep) and the read-only GitHub reader. A repository without an
+    upstream, or one that publishes nothing, attests no facts."""
+    if not lifecycle.full_name or not lifecycle.publishes:
+        return None
+    from .upstream_facts import MirrorObserver
+
+    full_name = lifecycle.full_name
+    return MirrorObserver(state_dir / "upstream" / f"{lifecycle.repo}.git", full_name,
+                          lambda number: github.get(f"/repos/{full_name}/pulls/{number}"))
+
+
 def calibration_record(rt: KbRuntime, lifecycle: RepoLifecycle) -> dict | None:
     raw = rt.ledger.get_cursor(lifecycle.repo, "calibration")
     return json.loads(raw) if raw else None
diff --git a/src/infermatrix_copilot/kb_service/upstream_facts.py b/src/infermatrix_copilot/kb_service/upstream_facts.py
new file mode 100644
index 00000000..defe31a8
--- /dev/null
+++ b/src/infermatrix_copilot/kb_service/upstream_facts.py
@@ -0,0 +1,102 @@
+"""Upstream facts for the gate (service) and the local gate (publisher).
+
+``MirrorObserver`` reads one public upstream through a bare mirror (paths,
+symbols) and a PR lookup (the service's read-only GitHub reader, or the
+publisher's own ``gh api``): the two sides never share a data source, so the
+publisher's re-check is independent of what the service saw.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+from typing import Callable
+
+from ..knowledge_service.facts import Claim, FactsError, claims_in
+from ..knowledge_service.lifecycle import LifecycleError, Page
+
+
+class MirrorObserver:
+    def __init__(self, git_dir: Path, repository: str, pull: Callable[[int], dict], *,
+                 url: str | None = None):
+        self.git_dir = Path(git_dir)
+        self.repository = repository
+        self._pull = pull
+        self._url = url or f"https://github.com/{repository}.git"
+        self._synced = False
+
+    def _git(self, *args: str, ok: tuple[int, ...] = (0,)) -> subprocess.CompletedProcess:
+        try:
+            proc = subprocess.run(["git", "--git-dir", str(self.git_dir), *args], capture_output=True,
+                                  text=True, check=False, timeout=600)
+        except (OSError, subprocess.TimeoutExpired) as exc:
+            raise FactsError(f"git {args[0]}: {exc}") from exc
+        if proc.returncode not in ok:
+            raise FactsError(f"git {' '.join(args[:2])}: {proc.stderr.strip()[:300]}")
+        return proc
+
+    def sync(self) -> None:
+        """Fetch the upstream's branches once per observer."""
+        if self._synced:
+            return
+        if not self.git_dir.exists():
+            self.git_dir.parent.mkdir(parents=True, exist_ok=True)
+            try:
+                proc = subprocess.run(["git", "clone", "--bare", "--quiet", self._url, str(self.git_dir)],
+                                      capture_output=True, text=True, check=False, timeout=1800)
+            except (OSError, subprocess.TimeoutExpired) as exc:
+                raise FactsError(f"cannot clone {self.repository}: {exc}") from exc
+            if proc.returncode != 0:
+                raise FactsError(f"cannot clone {self.repository}: {proc.stderr.strip()[:300]}")
+        self._git("fetch", "--quiet", "origin", "+refs/heads/*:refs/heads/*")
+        self._synced = True
+
+    def head(self) -> str:
+        self.sync()
+        return self._git("rev-parse", "HEAD^{commit}").stdout.strip()
+
+    def _commit(self, sha: str) -> None:
+        self.sync()
+        if self._git("cat-file", "-e", f"{sha}^{{commit}}", ok=(0, 1, 128)).returncode != 0:
+            raise FactsError(f"{self.repository} has no commit {sha[:12]}")
+
+    def top_level(self, sha: str) -> set[str]:
+        self._commit(sha)
+        return set(self._git("ls-tree", "--name-only", sha).stdout.split())
+
+    def path_exists(self, sha: str, path: str) -> bool:
+        self._commit(sha)
+        return self._git("cat-file", "-e", f"{sha}:{path}", ok=(0, 1, 128)).returncode == 0
+
+    def file_text(self, sha: str, path: str) -> str | None:
+        self._commit(sha)
+        proc = self._git("cat-file", "blob", f"{sha}:{path}", ok=(0, 1, 128))
+        return proc.stdout if proc.returncode == 0 else None
+
+    def pull(self, number: int) -> dict:
+        try:
+            data = self._pull(number)
+        except FactsError:
+            raise
+        except Exception as exc:  # noqa: BLE001 - the API is down or rate-limited
+            raise FactsError(f"{self.repository} PR #{number}: {exc}") from exc
+        if not isinstance(data, dict):
+            raise FactsError(f"{self.repository} PR #{number}: unexpected answer")
+        return data
+
+
+def change_claims(head: dict[str, str], blocks, top_level: set[str]) -> list[Claim]:
+    """The claims of every rule the change adds, edits or retires (prose and
+    purged rules carry none)."""
+    claims: list[Claim] = []
+    for block in blocks:
+        if block.kind != "rule" or block.op == "purge" or block.path not in head:
+            continue
+        try:
+            section = Page.parse(head[block.path]).rule(block.rule_id)
+            footer = section.footer
+        except (LifecycleError, KeyError):
+            continue  # L1 already refused anything unparsable
+        claims += claims_in(section.body_without_footer, top_level, active=footer.status == "active",
+                            evidence=footer.evidence)
+    return claims
diff --git a/src/infermatrix_copilot/knowledge_service/facts.py b/src/infermatrix_copilot/knowledge_service/facts.py
new file mode 100644
index 00000000..098dbe82
--- /dev/null
+++ b/src/infermatrix_copilot/knowledge_service/facts.py
@@ -0,0 +1,232 @@
+"""Upstream fact attestations: what a changed rule claims about its upstream.
+
+A rule states facts about the upstream repository it describes: the PRs it
+cites (``^[PR #N]``) were merged, and the files (and ``path::Symbol``
+definitions) it names exist. The service observes those facts on a pinned
+upstream SHA when it gates a change and signs them into the verdict; the
+publisher observes every one again from the upstream itself (the GitHub API
+and its own mirror) right before merging, and does not merge on any
+difference (design v8 §7.1, §8.2).
+
+Only claims that can be resolved are facts: a backticked token is a path claim
+when its first segment is a top-level entry of the upstream tree (``pkg/
+x.py`` is, ``config/x.py`` is not). Active rules must hold every claim; a
+retired rule's claims are recorded as observed (the behaviour may have moved
+without the file going away), but the PR its retirement names as evidence
+must be merged.
+
+Standard library only.
+"""
+
+from __future__ import annotations
+
+import ast
+import re
+from dataclasses import dataclass
+from typing import Iterable, Protocol
+
+PR_CITATION = re.compile(r"\^\[\s*PR\s*#(?P<n>\d{1,7})\s*\]")
+EVIDENCE_PR = re.compile(r"\bPR\s*#(?P<n>\d{1,7})\b")
+CODE_SPAN = re.compile(r"`(?P<tok>[^`\n]{1,200})`")
+SEGMENT = re.compile(r"[A-Za-z0-9_.-]+")
+PATH = re.compile(r"(?:[A-Za-z0-9_.-]+/)+[A-Za-z0-9_.-]*")  # at least one slash
+SYMBOL = re.compile(r"[A-Za-z_][A-Za-z0-9_]*(?:\.[A-Za-z_][A-Za-z0-9_]*)*")
+MAX_FACTS = 200
+
+
+class FactsError(RuntimeError):
+    """The upstream could not be read (unreachable, unknown SHA, API error)."""
+
+
+class Observer(Protocol):
+    """Reads one upstream repository. Raises FactsError when it cannot tell."""
+
+    repository: str
+
+    def head(self) -> str: ...
+
+    def top_level(self, sha: str) -> set[str]: ...
+
+    def path_exists(self, sha: str, path: str) -> bool: ...
+
+    def file_text(self, sha: str, path: str) -> str | None: ...
+
+    def pull(self, number: int) -> dict: ...
+
+
+@dataclass(frozen=True)
+class Claim:
+    kind: str            # pr | path | symbol
+    must_hold: bool      # an active rule's claim (a retired rule's is only recorded)
+    pr: int = 0
+    path: str = ""
+    symbol: str = ""
+
+    @property
+    def key(self) -> tuple:
+        return (self.kind, self.pr, self.path, self.symbol)
+
+
+def claims_in(text: str, top_level: set[str], *, active: bool, evidence: str = "") -> list[Claim]:
+    """The resolvable claims of one rule's text (and of its retirement evidence)."""
+    out: list[Claim] = [Claim("pr", active, pr=int(m.group("n"))) for m in PR_CITATION.finditer(text)]
+    if not active:
+        out += [Claim("pr", True, pr=int(m.group("n"))) for m in EVIDENCE_PR.finditer(evidence)]
+    for m in CODE_SPAN.finditer(text):
+        token = m.group("tok").strip()
+        path, _, symbol = token.partition("::")
+        # a bare top-level name is a claim only with a symbol (`setup.py::main`):
+        # alone it is too often a word (`tests`, `docs`)
+        resolvable = PATH.fullmatch(path) or (symbol and SEGMENT.fullmatch(path))
+        if not resolvable or path.split("/", 1)[0] not in top_level:
+            continue
+        if any(part and set(part) == {"."} for part in path.split("/")):
+            continue  # `..` escapes, `...` placeholders
+        if symbol:
+            if SYMBOL.fullmatch(symbol) and not path.endswith("/"):
+                out.append(Claim("symbol", active, path=path, symbol=symbol))
+            continue
+        out.append(Claim("path", active, path=path.rstrip("/")))
+    return out
+
+
+def merge_claims(claims: Iterable[Claim]) -> list[Claim]:
+    """One claim per fact; it must hold when any rule requires it."""
+    merged: dict[tuple, Claim] = {}
+    for claim in claims:
+        seen = merged.get(claim.key)
+        if seen is None or (claim.must_hold and not seen.must_hold):
+            merged[claim.key] = claim
+    return sorted(merged.values(), key=lambda c: c.key)
+
+
+def defines(text: str, symbol: str, path: str = "") -> bool:
+    """The file defines ``symbol``. Python is parsed (text in strings and
+    comments never counts): ``A.b`` must be a qualified definition, a single
+    name may be defined at any depth. Other files, or Python this
+    interpreter cannot parse, fall back to line patterns."""
+    if path.endswith((".py", ".pyi")):
+        try:
+            names = _python_definitions(ast.parse(text))
+        except (SyntaxError, ValueError, RecursionError):
+            names = None
+        if names is not None:
+            return symbol in names or ("." not in symbol and any(n.rsplit(".", 1)[-1] == symbol for n in names))
+    for part in symbol.split("."):
+        name = re.escape(part)
+        if not re.search(rf"(?m)^\s*(?:async\s+def|def|class)\s+{name}\b|^\s*{name}\s*(?::[^=\n]*)?=", text):
+            return False
+    return True
+
+
+def _python_definitions(tree: ast.AST) -> set[str]:
+    """Qualified names of every def, class, assigned and imported name (``A.b`` for a member)."""
+    out: set[str] = set()
+
+    def visit(node: ast.AST, prefix: str) -> None:
+        for child in ast.iter_child_nodes(node):
+            if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+                out.add(prefix + child.name)
+                visit(child, prefix + child.name + ".")
+                continue
+            targets = child.targets if isinstance(child, ast.Assign) else \
+                [child.target] if isinstance(child, (ast.AnnAssign, ast.AugAssign)) else []
+            for target in targets:
+                out.update(prefix + name for name in _bound(target))
+            if isinstance(child, (ast.Import, ast.ImportFrom)):  # a re-export binds the name too
+                out.update(prefix + (a.asname or a.name.split(".")[0]) for a in child.names if a.name != "*")
+            visit(child, prefix)
+
+    visit(tree, "")
+    return out
+
+
+def _bound(target: ast.AST) -> list[str]:
+    """Names an assignment target binds: ``a``, ``a, (b, *c)``; never the names
+    it only reads (``d[k] = v``, ``obj.attr = v``)."""
+    if isinstance(target, ast.Name):
+        return [target.id]
+    if isinstance(target, ast.Starred):
+        return _bound(target.value)
+    if isinstance(target, (ast.Tuple, ast.List)):
+        return [name for element in target.elts for name in _bound(element)]
+    return []
+
+
+def observe(claim: Claim, observer: Observer, sha: str) -> dict:
+    """The fact as the upstream shows it at ``sha`` (raises FactsError)."""
+    fact: dict = {"kind": claim.kind, "repository": observer.repository, "must_hold": claim.must_hold}
+    if claim.kind == "pr":
+        pull = observer.pull(claim.pr)
+        fact.update(pr=claim.pr, merged=bool(pull.get("merged")),
+                    merge_commit_sha=str(pull.get("merge_commit_sha") or "") if pull.get("merged") else "")
+    elif claim.kind == "path":
+        fact.update(sha=sha, path=claim.path, exists=observer.path_exists(sha, claim.path))
+    else:
+        text = observer.file_text(sha, claim.path)
+        fact.update(sha=sha, path=claim.path, symbol=claim.symbol,
+                    defined=text is not None and defines(text, claim.symbol, claim.path))
+    return fact
+
+
+def holds(fact: dict) -> bool:
+    return bool(fact.get("merged") if fact["kind"] == "pr" else
+                fact.get("exists") if fact["kind"] == "path" else fact.get("defined"))
+
+
+def describe(fact: dict) -> str:
+    if fact["kind"] == "pr":
+        return f"{fact['repository']} PR #{fact['pr']} is not merged"
+    if fact["kind"] == "path":
+        return f"{fact['path']} does not exist at {fact['repository']}@{fact['sha'][:12]}"
+    return f"{fact['path']}::{fact['symbol']} is not defined at {fact['repository']}@{fact['sha'][:12]}"
+
+
+def attest(claims: list[Claim], observer: Observer) -> tuple[dict, list[dict], list[str]]:
+    """Observe every claim on the upstream's current head: (upstream, facts,
+    problems). A problem is a claim an active rule (or a retirement's
+    evidence) requires that does not hold."""
+    sha = observer.head()
+    claims = merge_claims(claims)
+    if len(claims) > MAX_FACTS:
+        raise FactsError(f"{len(claims)} upstream claims (limit {MAX_FACTS}); split the change")
+    facts = [observe(claim, observer, sha) for claim in claims]
+    problems = [describe(f) for f in facts if f["must_hold"] and not holds(f)]
+    return {"repository": observer.repository, "sha": sha}, facts, problems
+
+
+def recheck(upstream: dict, facts: list[dict], observer: Observer) -> list[str]:
+    """The publisher's side: every signed fact observed again, independently.
+    Raises FactsError when the upstream cannot be read (not a verdict on the
+    facts: retried)."""
+    if not facts:
+        return []
+    problems = []
+    if upstream.get("repository") != observer.repository:
+        return [f"facts are about {upstream.get('repository')}, not {observer.repository}"]
+    sha = str(upstream.get("sha") or "")
+    if not re.fullmatch(r"[0-9a-f]{40}", sha):
+        return ["the verdict names no pinned upstream SHA"]
+    for fact in facts:
+        try:
+            claim = Claim(str(fact["kind"]), bool(fact["must_hold"]), pr=int(fact.get("pr") or 0),
+                          path=str(fact.get("path") or ""), symbol=str(fact.get("symbol") or ""))
+        except (KeyError, TypeError, ValueError):
+            problems.append(f"malformed fact: {str(fact)[:120]}")
+            continue
+        if claim.kind not in ("pr", "path", "symbol") or fact.get("repository") != observer.repository \
+                or (claim.kind != "pr" and fact.get("sha") != sha):
+            problems.append(f"malformed fact: {str(fact)[:120]}")
+            continue
+        if claim.must_hold and not holds(fact):
+            problems.append("signed a fact that does not hold: " + describe(fact))
+            continue
+        actual = observe(claim, observer, sha)
+        if actual != fact:
+            problems.append(f"upstream disagrees: signed {_short(fact)}, observed {_short(actual)}")
+    return problems
+
+
+def _short(fact: dict) -> str:
+    keep = ("pr", "merged", "merge_commit_sha", "path", "symbol", "exists", "defined")
+    return ", ".join(f"{k}={fact[k]}" for k in keep if k in fact)
diff --git a/src/infermatrix_copilot/knowledge_service/gate_verifier.py b/src/infermatrix_copilot/knowledge_service/gate_verifier.py
index 69e9dfba..964acd40 100644
--- a/src/infermatrix_copilot/knowledge_service/gate_verifier.py
+++ b/src/infermatrix_copilot/knowledge_service/gate_verifier.py
@@ -266,8 +266,6 @@ def verify_change(ctx: Context, *, pr: dict, head_sha: str, pre: str, post: str,
         problems.append("only auto verdicts merge (there is no human-approved path)")
     if not manifests_equal(verdict.get("manifest") or [], entries):
         problems.append("the signed patch manifest does not match the change as applied")
-    if verdict.get("facts"):
-        problems.append("the verdict carries upstream fact attestations this verifier cannot re-check")
 
     result = check_changeset(pre_files, post_files, changes,
                              external_texts=ctx.git.external_texts(final),
diff --git a/test/test_kb_upstream_facts.py b/test/test_kb_upstream_facts.py
new file mode 100644
index 00000000..7615cb0d
--- /dev/null
+++ b/test/test_kb_upstream_facts.py
@@ -0,0 +1,271 @@
+"""Upstream fact attestations: the service signs them, the publisher observes them again."""
+
+from __future__ import annotations
+
+import subprocess
+
+import pytest
+
+from infermatrix_copilot.kb_service import merge
+from infermatrix_copilot.kb_service.gate import changes_between, run_gate
+from infermatrix_copilot.kb_service.upstream_facts import MirrorObserver
+from infermatrix_copilot.knowledge_service.facts import (
+    Claim, FactsError, attest, claims_in, defines, recheck,
+)
+from infermatrix_copilot.knowledge_service.ops import KnowledgeOperation as Op, apply_operations
+from test_kb_intake_gate import JUDGE, PAGE, ScriptedGateway, _judge_all, _rule, _tree
+
+UP = "org/upstream"
+SHA = "a" * 40
+
+
+class FakeUpstream:
+    """An upstream the test controls: files at one SHA and PR states."""
+
+    def __init__(self, files=None, pulls=None, *, sha=SHA, down=False):
+        self.repository = UP
+        self.sha = sha
+        self.files = dict(files or {"pkg/queue.py": "class Queue:\n    def put(self):\n        pass\n",
+                                    "README.md": "x\n"})
+        self.pulls = dict(pulls or {10: {"merged": True, "merge_commit_sha": "c" * 40}})
+        self.down = down
+
+    def _up(self):
+        if self.down:
+            raise FactsError("upstream unreachable")
+
+    def head(self):
+        self._up()
+        return self.sha
+
+    def _known(self, sha):
+        self._up()
+        if sha != self.sha:
+            raise FactsError(f"no commit {sha[:12]}")
+
+    def top_level(self, sha):
+        self._known(sha)
+        return {p.split("/", 1)[0] for p in self.files}
+
+    def path_exists(self, sha, path):
+        self._known(sha)
+        return path in self.files or any(p.startswith(path + "/") for p in self.files)
+
+    def file_text(self, sha, path):
+        self._known(sha)
+        return self.files.get(path)
+
+    def pull(self, number):
+        self._up()
+        return self.pulls.get(number, {"merged": False})
+
+
+# -- claims ------------------------------------------------------------------------
+
+def test_only_resolvable_tokens_are_claims():
+    text = ("见 `pkg/queue.py`、`pkg/queue.py::Queue.put`、`pkg/`、`config/x.py`、`/v1/chat`、"
+            "`pkg/../etc/passwd`、`pkg/...`、`origin/main`。 ^[PR #10] ^[PR #11]")
+    claims = claims_in(text, {"pkg", "README.md"}, active=True)
+    assert {(c.kind, c.pr, c.path, c.symbol) for c in claims} == {
+        ("pr", 10, "", ""), ("pr", 11, "", ""), ("path", 0, "pkg/queue.py", ""),
+        ("symbol", 0, "pkg/queue.py", "Queue.put"), ("path", 0, "pkg", "")}
+    assert all(c.must_hold for c in claims)
+
+
+def test_a_retired_rule_records_its_claims_but_must_cite_merged_evidence():
+    claims = claims_in("`pkg/gone.py` ^[PR #10]", {"pkg"}, active=False, evidence="PR #12")
+    assert {(c.kind, c.pr, c.path, c.must_hold) for c in claims} == {
+        ("pr", 10, "", False), ("path", 0, "pkg/gone.py", False), ("pr", 12, "", True)}
+
+
+def test_symbol_definitions():
+    text = "class Queue:\n    async def put(self):\n        pass\nLIMIT: int = 3\n"
+    for path in ("pkg/queue.py", "pkg/queue.txt"):              # parsed, and the line-pattern fallback
+        assert defines(text, "Queue.put", path) and defines(text, "LIMIT", path)
+        assert not defines(text, "Queue.get", path) and not defines(text, "Que", path)
+
+
+def test_definitions_inside_strings_do_not_count():
+    text = 'DOC = """\nclass Removed:\n    pass\n"""\n# def gone(): pass\ndef kept():\n    x = 1\n'
+    assert not defines(text, "Removed", "pkg/m.py") and not defines(text, "gone", "pkg/m.py")
+    assert defines(text, "kept", "pkg/m.py") and defines(text, "DOC", "pkg/m.py")
+    assert not defines(text, "kept.y", "pkg/m.py")
+    reads = "registry[Missing] = 1\nobj.Attr = 2\na, (b, *c) = 1, (2, 3)\n"
+    assert not defines(reads, "Missing", "pkg/m.py") and not defines(reads, "Attr", "pkg/m.py")
+    assert not defines(reads, "obj", "pkg/m.py") and not defines(reads, "registry", "pkg/m.py")
+    assert all(defines(reads, n, "pkg/m.py") for n in ("a", "b", "c"))
+    exports = "from .engine import Engine as Public, Other\nimport os.path\nfrom x import *\n"
+    assert all(defines(exports, n, "pkg/__init__.py") for n in ("Public", "Other", "os"))
+    assert not defines(exports, "Engine", "pkg/__init__.py")
+    _, facts, problems = attest(claims_in("`pkg/m.py::Removed`", {"pkg"}, active=True),
+                                FakeUpstream(files={"pkg/m.py": text}))
+    assert facts[0]["defined"] is False and problems
+
+
+def test_root_files_are_claims_only_with_a_symbol():
+    claims = claims_in("`setup.py::missing` `setup.py` `tests`", {"setup.py", "tests"}, active=True)
+    assert [(c.kind, c.path, c.symbol) for c in claims] == [("symbol", "setup.py", "missing")]
+    _, _, problems = attest(claims, FakeUpstream(files={"setup.py": "def main():\n    pass\n"}))
+    assert problems == [f"setup.py::missing is not defined at {UP}@{SHA[:12]}"]
+
+
+# -- attest and recheck --------------------------------------------------------------
+
+def _claims(*tokens, active=True, evidence=""):
+    return claims_in(" ".join(tokens), {"pkg", "README.md"}, active=active, evidence=evidence)
+
+
+def test_attest_signs_what_holds_and_fails_what_an_active_rule_gets_wrong():
+    upstream, facts, problems = attest(_claims("`pkg/queue.py::Queue.put`", "^[PR #10]"), FakeUpstream())
+    assert upstream == {"repository": UP, "sha": SHA} and problems == [] and len(facts) == 2
+    _, _, problems = attest(_claims("`pkg/nope.py`", "`pkg/queue.py::Queue.get`", "^[PR #99]"), FakeUpstream())
+    assert sorted(problems) == [f"{UP} PR #99 is not merged",
+                                f"pkg/nope.py does not exist at {UP}@{SHA[:12]}",
+                                f"pkg/queue.py::Queue.get is not defined at {UP}@{SHA[:12]}"]
+    _, facts, problems = attest(_claims("`pkg/nope.py`", active=False, evidence="PR #10"), FakeUpstream())
+    assert problems == [] and any(f["kind"] == "path" and f["exists"] is False for f in facts)
+
+
+def test_recheck_agrees_only_with_the_same_upstream_state():
+    upstream, facts, _ = attest(_claims("`pkg/queue.py`", "^[PR #10]"), FakeUpstream())
+    assert recheck(upstream, facts, FakeUpstream()) == []
+    assert recheck({}, [], FakeUpstream(down=True)) == []                   # nothing to check
+    moved = FakeUpstream(pulls={10: {"merged": True, "merge_commit_sha": "d" * 40}})
+    assert any("upstream disagrees" in p for p in recheck(upstream, facts, moved))
+    gone = FakeUpstream(files={"README.md": "x\n"})
+    assert any("upstream disagrees" in p for p in recheck(upstream, facts, gone))
+    with pytest.raises(FactsError):
+        recheck(upstream, facts, FakeUpstream(down=True))
+    with pytest.raises(FactsError):                                         # the pinned SHA is unknown
+        recheck(upstream, facts, FakeUpstream(sha="b" * 40))
+    other = FakeUpstream()
+    other.repository = "org/other"
+    assert recheck(upstream, facts, other) == [f"facts are about {UP}, not org/other"]
+
+
+def test_recheck_refuses_forged_or_false_facts():
+    upstream, facts, _ = attest(_claims("`pkg/queue.py`"), FakeUpstream())
+    false = [{**facts[0], "exists": False}]
+    assert recheck(upstream, false, FakeUpstream())[0].startswith("signed a fact that does not hold")
+    assert recheck(upstream, [{"kind": "path"}], FakeUpstream())[0].startswith("malformed fact")
+    assert recheck({"repository": UP, "sha": "HEAD"}, facts, FakeUpstream()) == [
+        "the verdict names no pinned upstream SHA"]
+
+
+# -- the mirror ----------------------------------------------------------------------
+
+def _git(cwd, *args):
+    return subprocess.run(["git", "-C", str(cwd), *args], check=True, capture_output=True, text=True).stdout.strip()
+
+
+def test_mirror_observer_reads_a_real_repository(tmp_path):
+    work = tmp_path / "up"
+    work.mkdir()
+    _git(work, "init", "-q", "-b", "main")
+    (work / "pkg").mkdir()
+    (work / "pkg" / "queue.py").write_text("class Queue:\n    pass\n")
+    _git(work, "add", "-A")
+    _git(work, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "x")
+    head = _git(work, "rev-parse", "HEAD")
+    observer = MirrorObserver(tmp_path / "mirror.git", UP, lambda n: {"merged": n == 10}, url=str(work))
+    assert observer.head() == head and observer.top_level(head) == {"pkg"}
+    assert observer.path_exists(head, "pkg/queue.py") and observer.path_exists(head, "pkg")
+    assert not observer.path_exists(head, "pkg/nope.py")
+    assert "class Queue" in observer.file_text(head, "pkg/queue.py")
+    assert observer.pull(10)["merged"] is True
+    with pytest.raises(FactsError):
+        observer.path_exists("f" * 40, "pkg/queue.py")
+    broken = MirrorObserver(tmp_path / "m2.git", UP, lambda n: 1 / 0, url=str(tmp_path / "missing"))
+    with pytest.raises(FactsError):
+        broken.head()
+    with pytest.raises(FactsError):
+        MirrorObserver(tmp_path / "mirror.git", UP, lambda n: 1 / 0, url=str(work)).pull(1)
+
+
+# -- the gate --------------------------------------------------------------------------
+
+def _gate(claim: str, observer):
+    base = _tree()
+    head = apply_operations(base, [Op("add", PAGE, "DEMO-2a", _rule("DEMO-2a", "PR #10", claim=claim))],
+                            release="v1", today="2026-09-29").files
+    head = {**base, **head}
+    return run_gate(base=base, head=head, changes=changes_between(base, head), external_texts={},
+                    evidence=[], gateway=ScriptedGateway(_judge_all("yes")), judge=JUDGE, release="v1",
+                    repo_dir="repos/demo", retire_ratio=1.0, facts=observer)
+
+
+def test_the_gate_signs_facts_and_fails_a_false_claim_before_paying_the_judge():
+    decision = _gate("`pkg/queue.py::Queue.put` 满时拒绝", FakeUpstream())
+    assert decision.status == "pass", decision.reasons
+    assert decision.upstream == {"repository": UP, "sha": SHA}
+    assert {f["kind"] for f in decision.facts} == {"pr", "symbol"}
+    assert decision.to_dict()["facts"] == decision.facts
+    decision = _gate("`pkg/removed.py` 满时拒绝", FakeUpstream())
+    assert decision.status == "fail" and decision.blocks == []              # no L2 call was made
+    assert decision.reasons == [f"upstream fact: pkg/removed.py does not exist at {UP}@{SHA[:12]}"]
+    decision = _gate("`pkg/queue.py` 满时拒绝", FakeUpstream(down=True))
+    assert decision.status == "human" and "could not be checked" in decision.reasons[0]
+    assert _gate("`pkg/queue.py` 满时拒绝", None).facts == []                 # no upstream: nothing attested
+
+
+# -- the publisher ---------------------------------------------------------------------
+
+def _with_facts(tmp_path, observer):
+    """The service's PR is open; its decision carries facts; the merge item is issued."""
+    from test_kb_publisher import _collect, _setup
+
+    rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path)
+    pub.run_once()
+    _collect(rt)
+    changeset = rt.ledger.changeset(changeset_id)
+    upstream, facts, _ = attest(_claims("`pkg/queue.py`", "^[PR #10]"), FakeUpstream())
+    rt.ledger.update_changeset(changeset_id, detail={**changeset["detail"], "decision": {
+        **changeset["detail"]["decision"], "upstream": upstream, "facts": facts}})
+    head = gh.prs[42]["headRefOid"]
+    rt.github.prs[42] = {"number": 42, "state": "open", "merged": False, "head": {"sha": head}}
+    assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"]
+    rt.outbox.refresh_control()
+    pub.upstreams = {lifecycle.repo: UP}
+    pub.observer = lambda repo, name: observer
+    return rt, lifecycle, changeset_id, pub, gh
+
+
+def test_the_publisher_merges_when_the_upstream_agrees(tmp_path):
+    from test_kb_publisher import _collect
+
+    rt, lifecycle, changeset_id, pub, gh = _with_facts(tmp_path, FakeUpstream())
+    assert pub.run_once()["performed"] == 1 and gh.prs[42]["state"] == "MERGED"
+    _collect(rt)
+    assert rt.ledger.changeset(changeset_id)["status"] == "merged"
+
+
+@pytest.mark.parametrize("observer", [FakeUpstream(pulls={10: {"merged": False}}), FakeUpstream(down=True)],
+                         ids=["disagrees", "unreachable"])
+def test_the_publisher_does_not_merge_and_people_hear_after_a_day(tmp_path, observer):
+    from test_kb_publisher import _collect
+
+    rt, lifecycle, changeset_id, pub, gh = _with_facts(tmp_path, observer)
+    assert pub.run_once()["failed"] == 1 and gh.prs[42]["state"] == "OPEN"
+    assert not any(call[:2] == ["pr", "merge"] for call in gh.calls)
+    _collect(rt)
+    changeset = rt.ledger.changeset(changeset_id)
+    assert changeset["status"] == "pr_open" and changeset["detail"]["facts_failing_since"]
+    assert rt.ledger.human_queue("demo") == []                               # retried first
+    now = rt.clock()
+    rt.clock = rt.outbox._clock = pub.clock = lambda: now + 25 * 3600
+    assert merge.advance(rt, lifecycle) == [f"merge issued {changeset_id}"]   # signed again
+    rt.outbox.refresh_control()
+    assert pub.run_once()["failed"] == 1
+    _collect(rt)
+    assert rt.ledger.changeset(changeset_id)["status"] == "gate_failed"
+    assert any("upstream facts" in row["reason"] for row in rt.ledger.human_queue("demo"))
+
+
+def test_facts_without_a_configured_public_upstream_never_merge(tmp_path):
+    rt, lifecycle, changeset_id, pub, gh = _with_facts(tmp_path, FakeUpstream())
+    pub.upstreams = {}
+    assert pub.run_once()["failed"] == 1 and gh.prs[42]["state"] == "OPEN"
+
+
+def test_a_claim_is_hashable_per_fact():
+    assert Claim("path", True, path="a/b").key == Claim("path", False, path="a/b").key