Merge pull request #232 from JiusiServe/kb/local-upstream-facts

Attest upstream facts in the gate and re-check them before merging (design v8 8.2)

Author: tzhouam

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 小时后转人工、未配置上游不合并）。

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` 复核）。

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 发布器。

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",
     )

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]

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:

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

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)

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