From c6bf5a956632bfc0f50bcc102808857b629b7d08 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 29 Sep 2026 15:35:40 +0800 Subject: [PATCH] Attest upstream facts in the gate and re-check them before merging (design v8 8.2) Rules state facts about their upstream: the PRs they cite were merged, and the files and symbols they name exist. Until now nothing checked them. The service signed facts=[], and the verifier refused any verdict that carried facts. - New knowledge_service/facts.py, shared by both sides. It extracts the claims that can be resolved from a rule: - ^[PR #N] citations; - backticked paths whose first segment is a top-level entry of the upstream tree; - path::Symbol, where every dotted part must be defined by a def, class or assignment. Fragments such as config/x.py, and paths with .. or ... segments, are not claims. An active rule's claims must hold. A retired rule's claims are only recorded, but the PR named as retirement evidence must be merged. attest() observes the claims on the upstream head and returns the pinned {repository, sha}, the facts and the problems. recheck() observes every signed fact again and compares it field by field. An unreadable upstream raises FactsError. - Service: run_gate(facts=observer) attests right after L1, so a false claim fails before any paid L2 call and goes to refine. An unreadable upstream sends the change to people. The facts and the pinned upstream go into the decision and are signed into the verdict. The observer (kb_service/upstream_facts.MirrorObserver) uses the service's own bare mirror, shared with the release sweep, and the read-only GitHub reader. KbRuntime.upstream_facts supplies it. A repository without an upstream, or one that publishes nothing, attests nothing. - Publisher: after the local gate passes and before the final control re-read, it re-checks every fact from independent sources: its own mirror and `gh api`, using upstreams from its own adapter config. Any difference, a missing public upstream, or an unreadable upstream means no merge this round. - The service treats such a refusal as transient and signs again next round, recording facts_failing_since. After 24 consecutive hours the change goes to gate_failed and people are told. Any other answer ends the run of failures. - gate_verifier no longer refuses verdicts that carry facts: the publisher re-checks them. Measured on the live vllm-omni knowledge against upstream main: 710 active rules, 68 path/symbol claims, 7 rules (1%) cite a path that no longer exists. Those rules fail the gate when they are next edited, which is the intent. Tests: test_kb_upstream_facts.py covers claim extraction, symbol definitions, and attest/recheck when the upstream agrees, disagrees, is unreachable, has an unknown SHA, is another repository, or the facts are forged. It also covers a real git mirror and a gate that fails before L2. For the publisher: it merges when the upstream agrees; it does not merge when the upstream disagrees or is unreachable, and escalates after 24 hours; it does not merge without a configured public upstream. Co-Authored-By: Claude Opus 5.5 --- doc/architecture/SPEC/kb_service.md | 14 + doc/architecture/SPEC/knowledge_service.md | 9 + doc/features/knowledge-service.md | 2 + src/infermatrix_copilot/kb_service/cli.py | 2 + .../kb_service/external.py | 3 +- src/infermatrix_copilot/kb_service/gate.py | 29 +- src/infermatrix_copilot/kb_service/merge.py | 22 +- .../kb_service/publisher.py | 29 ++ src/infermatrix_copilot/kb_service/runtime.py | 22 +- .../kb_service/upstream_facts.py | 102 +++++++ .../knowledge_service/facts.py | 232 +++++++++++++++ .../knowledge_service/gate_verifier.py | 2 - test/test_kb_upstream_facts.py | 271 ++++++++++++++++++ 13 files changed, 729 insertions(+), 10 deletions(-) create mode 100644 src/infermatrix_copilot/kb_service/upstream_facts.py create mode 100644 src/infermatrix_copilot/knowledge_service/facts.py create mode 100644 test/test_kb_upstream_facts.py 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`：服务自己的上游裸镜像 `/upstream/.git`（与发版巡检共用）+ 只读 + GitHub reader）。必需声明不成立 → `fail`（理由 `upstream fact: …`，不调用付费评审，进入精炼）；上游不可读 → `human`。 + 事实与钉住的 `upstream` 写入 `decision`，`sign_verdict` 签入判定。没有上游（`general`）或不发布的仓库不做证明。 + `KbRuntime.upstream_facts(lifecycle)` 提供观测者（`from_env` 装配；测试默认不证明）。 +- 发布器：`merge` 在本地门禁通过后、最后一次读控制记录前调用 `facts.recheck`，数据来源独立：发布器自己的镜像 + `/upstream/.git` 与 `gh api repos//pulls/`；上游来自**发布器自己的**适配器配置 + （`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/.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\d{1,7})\s*\]") +EVIDENCE_PR = re.compile(r"\bPR\s*#(?P\d{1,7})\b") +CODE_SPAN = re.compile(r"`(?P[^`\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