Merge pull request #229 from JiusiServe/kb/local-external-daily

Check other people's knowledge PRs daily and tell them why (design v8)

Author: tzhouam

doc/architecture/SPEC/kb_service.md

@@ -229,3 +229,15 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝
 服务自己的 PR 在任何状态下（`Ledger.changesets_for_pr`）都不会被当作他人的 PR 重新评审。
 外部 PR 被本地门禁拒绝仍为 `gate_failed` 转人工（他人 PR 的处理见后续）；仓库设置导致的拒绝（`main` 必须允许发布器直接合并）
 也转人工，精炼无法修复。
+
+## 2026-09-29 v8：他人的知识 PR 每日复检与发现评论
+`poll_external` 按 PR 限频：每个 PR 每天至多检查一次，检查完成即在账本记下（`external_checked:<PR>`），重启或中途失败都不会
+当天重复检查已完成的 PR（`force=True` 用于测试与手动）。当天对每个打开的
+他人知识 PR：当前 head 已通过并在合并流程中的不再评审；未通过（`failed`/`human`/`gate_failed`/上下文失效/head 变化等）的每天重新评审，
+作者推送多少次都只在下一次每日检查时评审。每次评审（含本地门禁拒绝）都签发 `post_findings`：发布器在该 PR 上只维护一条带
+`<!-- kb-findings:v1 -->` 标记、由自己账号发出的评论，已有则编辑、没有才创建，从不编辑他人伪造的同标记评论。未通过的外部 PR 不进
+人工队列（作者负责修改），也从不合并；影子模式与私有上游不发布评论；跨多个仓库的 PR 经其中一个
+公开且为 `auto_merge` 的仓库发布评论（只要其中有私有上游就不发布）。`post_findings` 失败时不回执，每轮重试直至成功；每条发现评论以签发时间为修订号，
+发布器记住每个 PR 已发布的最新修订，较旧的项不会覆盖较新的评论；服务在账本（全局 `findings:<PR>`，每个 PR 一条，与其涉及的仓库
+范围无关）记下最新一条直至发布器确认送达，过期仍未送达的以相同内容与原修订号重新签发（与 PR 当前状态无关），因此旧发现永远
+不会覆盖新的；发布器的修订记录原子写入，损坏时视为无记录。

doc/features/knowledge-service.md

@@ -64,7 +64,8 @@ infermatrix-copilot kb rollback --to <sha>   # 回指到较早的快照（下一
 合并流程（v8）：`open_pr` 回执 → 为该 PR 的精确 head 签发判定，随 `merge` 项交给 GPU 盒发布器 → 发布器在
 `main` 与 PR head 的合并树上运行本地门禁，通过才 `gh pr merge`（每仓库同时至多 1 个）→ 合并后激活新快照。
 本地门禁拒绝的变更从不合并：上下文变化则在当前 main 上重建；服务自己的变更被质量门或本地门禁拒绝时，由生成器按
-理由精炼后整体复检，至多 2 轮，仍失败则留在人工队列（不合并）。发版巡检在上游新 release 后运行
+理由精炼后整体复检，至多 2 轮，仍失败则留在人工队列（不合并）。他人的知识 PR 每天检查一次：结果写在该 PR 上唯一一条
+“知识质量门”评论里（原地更新），未通过则不合并，作者修改后次日复检。发版巡检在上游新 release 后运行
 T1/T2/T3 与 purge，每个规则页一个变更集，全部经过质量门。
 
 评审服务读取知识：将 `KNOWLEDGE_ROOT` 指向 `$KB_STATE_DIR/active`。

src/infermatrix_copilot/kb_service/external.py

@@ -12,10 +12,12 @@
   the full quality gate (L1, L2, consistency) against CURRENT main with the
   PR's changes applied; a pass gets an ``auto`` verdict.
 
-Anything else goes to people once per head (with what would make it pass).
-A staged external change set then follows the ordinary merge flow (verdict
-comment, PR-stage kb-gate, merge queue). A new head, or a verdict whose
-context moved, is simply judged again at the next poll.
+A PR that does not pass is never merged (design v8, D12): the author is told
+why in ONE findings comment on the PR, updated in place, and the PR is checked
+again once a day (not on every push) until it passes. A passing PR follows the
+ordinary merge flow: a signed verdict in a ``merge`` item, the publisher's
+local gate, the merge. A passing head is not judged again; a new head, or a
+merge refused meanwhile, is judged at the next daily pass.
 
 Judging "on current main" requires the PR's pre-images to equal main's:
 otherwise the merge queue would apply different changes than were signed, and
@@ -24,6 +26,7 @@
 
 from __future__ import annotations
 
+import json
 from typing import Any
 
 from ..knowledge_service.gate_verifier import MAINTAINER_PATH, WHITELIST_CODES, code_owners, parse_codeowners
@@ -37,8 +40,15 @@
 HUMAN_LABEL = "kb:human-approved"
 GOVERNED = ("knowledge/repos/", "knowledge/general/")
 SUFFIXES = (".md", ".yaml")
-# statuses after which a head is judged again (the verdict can no longer verify)
-REJUDGE = ("head_changed", "stale_context", "superseded")
+# statuses after which a head is judged again at the next daily pass: it did
+# not pass, or its verdict can no longer verify
+REJUDGE = ("head_changed", "stale_context", "superseded", "failed", "human", "gate_failed", "rebuild_failed")
+EXTERNAL_EVERY = 24 * 3600  # other people's PRs are checked once a day
+CHECKED_CURSOR = "external_checked:{number}"  # per PR: a failed pass never re-judges finished ones
+FINDINGS_MARKER = "<!-- kb-findings:v1 -->"
+# + PR number, kept GLOBALLY (a comment belongs to the PR, whatever
+# repository scope it touches now): the latest findings until delivered
+FINDINGS_CURSOR = "findings:"
 
 
 def _governed(path: str) -> bool:
@@ -157,11 +167,117 @@ def _ours(rt, number: int) -> bool:
     "paused", "shadow_recorded", "calibration_required", "approval_withdrawn")))
 
 
-def poll_external(rt) -> list[str]:
-    """One pass over open knowledge PRs; needs the scheduler's lease."""
+def poll_external(rt, *, force: bool = False) -> list[str]:
+    """A pass over open knowledge PRs; needs the scheduler's lease. Each PR is
+    checked at most once per ``EXTERNAL_EVERY``, recorded per PR in the ledger
+    as soon as it was checked: a restart, or a pass that fails half-way, never
+    checks a PR twice in a day. ``force`` ignores the daily limit."""
     owner = getattr(rt, "lease_owner", None)
     if rt.outbox is None or not owner:
         return []
+    return renew_findings(rt) + _poll(rt, owner, force=force)
+
+
+def _due(rt, number: int, force: bool) -> bool:
+    last = float(rt.ledger.get_cursor("*", CHECKED_CURSOR.format(number=number)) or 0)
+    return force or rt.clock() - last >= EXTERNAL_EVERY
+
+
+def _checked(rt, number: int) -> None:
+    rt.ledger.set_cursor("*", CHECKED_CURSOR.format(number=number), str(rt.clock()))
+
+
+def findings_text(number: int, head: str, status: str, reasons: list[str], today: str) -> str:
+    """The one comment the service keeps on another person's knowledge PR."""
+    lines = [FINDINGS_MARKER, f"**Knowledge quality gate** · checked {today} at `{head[:12]}` "
+             "(checked again once a day)", ""]
+    if status in ("pr_open", "merged"):
+        lines.append("Result: **passed**. The publisher merges it after its final check on the exact merge result.")
+    elif status == "shadow_recorded":
+        lines.append("Result: **passed** (this repository is in shadow mode: nothing is merged automatically yet).")
+    else:
+        lines.append("Result: **not passed**. This PR will not be merged until the problems below are fixed; "
+                     "push a fix and it is checked again at the next daily pass.")
+        lines += ["", "Problems:"] + [f"- {r}" for r in (reasons or ["the change needs a person's decision"])[:20]]
+    return "\n".join(lines) + "\n"
+
+
+def _findings_lifecycle(rt, paths: list[str]):
+    """Which repository may carry the findings of a PR that belongs to none
+    (it spans several): only when every scope it touches publishes (a private
+    upstream's name must never appear), through one in auto_merge."""
+    scopes = set()
+    for path in paths:
+        try:
+            scopes.add(repo_scope(path[len("knowledge/"):]).split("/")[-1])
+        except Exception:  # noqa: BLE001 - outside repos/ and general/
+            continue
+    lifecycles = [rt.registry.get(name) for name in sorted(scopes)]
+    if not lifecycles or any(lc is None or not lc.publishes for lc in lifecycles):
+        return None
+    return next((lc for lc in lifecycles if lc.auto_merge), None)
+
+
+def post_findings(rt, lifecycle, number: int, head: str, status: str, reasons: list[str]) -> None:
+    """Hand the findings comment to the publisher (never in shadow mode, never
+    for a private upstream: those publish nothing)."""
+    if lifecycle is None or not lifecycle.auto_merge or not lifecycle.publishes:
+        return
+    comment = findings_text(number, head, status, reasons, rt.today())
+    _issue_findings(rt, lifecycle.repo, number, head, comment)
+
+
+def _issue_findings(rt, repo: str, number: int, head: str, comment: str, revision: float | None = None) -> None:
+    """Issue the findings and remember them (one record per PR) until the
+    publisher confirms delivery. ``revision`` is when the findings were
+    written: the publisher drops an item older than what it already posted,
+    and a renewal keeps the original revision, so it can never overtake newer
+    findings."""
+    revision = rt.clock() if revision is None else revision
+    item = rt.outbox.issue(repo, "post_findings", {
+        "pr": number, "head_sha": head, "marker": FINDINGS_MARKER, "comment": comment, "revision": revision})
+    rt.ledger.set_cursor("*", f"{FINDINGS_CURSOR}{number}", json.dumps({
+        "repo": repo, "item_id": item.id, "expires_at": item.expires_at, "head_sha": head, "comment": comment,
+        "revision": revision, "delivered": False}))
+
+
+def findings_delivered(rt, ack: dict) -> None:
+    """The publisher's ack for a findings item: only the latest one counts."""
+    name = f"{FINDINGS_CURSOR}{int(ack.get('pr') or 0)}"
+    raw = rt.ledger.get_cursor("*", name)
+    if not raw or not ack.get("ok"):
+        return
+    record = json.loads(raw)
+    if record.get("item_id") == ack.get("item_id"):
+        rt.ledger.set_cursor("*", name, json.dumps({**record, "delivered": True}))
+
+
+def renew_findings(rt) -> list[str]:
+    """Findings that expired undelivered (the publisher was down or GitHub
+    failed for a day) are issued again with the same text: an author is
+    never left without them, whatever the PR's status."""
+    events = []
+    for name, raw in rt.ledger.cursors_with_prefix("*", FINDINGS_CURSOR).items():
+        record = json.loads(raw)
+        if record.get("delivered") or float(record.get("expires_at") or 0) > rt.clock():
+            continue
+        lifecycle = rt.registry.get(record.get("repo"))
+        if lifecycle is None or not lifecycle.auto_merge or not lifecycle.publishes:
+            continue  # its repository publishes nothing any more
+        number = int(name[len(FINDINGS_CURSOR):])
+        _issue_findings(rt, lifecycle.repo, number, record["head_sha"], record["comment"],
+                        revision=float(record.get("revision") or 0))
+        events.append(f"findings reissued PR #{number}")
+    return events
+
+
+def _decision_reasons(decision: dict) -> list[str]:
+    from .refine import reasons_from_decision
+
+    return reasons_from_decision(decision)
+
+
+def _poll(rt, owner: str, *, force: bool = False) -> list[str]:
     events: list[str] = []
     main_sha = rt.knowledge.fetch()
     maintainers: set[str] | None = None
@@ -172,6 +288,8 @@ def poll_external(rt) -> list[str]:
         human = HUMAN_LABEL in labels
         if pr.get("draft") or not head or _ours(rt, number) or _judged(rt, number, head, "human" if human else "auto"):
             continue
+        if not _due(rt, number, force):
+            continue  # checked today already: once a day, not on every push
         if rt.knowledge.fetch_pull(number) != head:
             continue  # pushed while we looked: next poll
         merge_base = rt.knowledge.merge_base(main_sha, head)
@@ -182,7 +300,9 @@ def poll_external(rt) -> list[str]:
         path = "human" if human else "auto"
         lifecycle, why = _scope([p for p in paths if p.startswith("knowledge/")], rt.registry)
         if lifecycle is None:
-            events.append(_to_people(rt, None, number, head, f"PR #{number} {why}", path))
+            events.append(_to_people(rt, None, number, head, f"PR #{number} {why}", path,
+                                     findings_via=_findings_lifecycle(rt, paths)))
+            _checked(rt, number)
             continue
         if human:
             maintainers = maintainers if maintainers is not None else _maintainers(rt, main_sha)
@@ -193,6 +313,7 @@ def poll_external(rt) -> list[str]:
             events.append(_to_people(rt, lifecycle, number, head,
                                      f"PR #{number} changes paths outside the governed knowledge pages; "
                                      f"a knowledge maintainer can approve it and add {HUMAN_LABEL}", path))
+            _checked(rt, number)
             continue
         base_main = rt.knowledge.knowledge_files(main_sha)
         pre = rt.knowledge.knowledge_files(merge_base)
@@ -203,6 +324,7 @@ def poll_external(rt) -> list[str]:
         if moved:
             events.append(_to_people(rt, lifecycle, number, head,
                                      f"PR #{number} must be rebased: main changed {moved[:3]} since it branched", path))
+            _checked(rt, number)
             continue
         head_files = dict(base_main)
         for rel in rels:
@@ -223,6 +345,7 @@ def poll_external(rt) -> list[str]:
             if issues:
                 events.append(_to_people(rt, lifecycle, number, head, f"PR #{number} fails L1: " + "; ".join(
                     f"{i.code} {i.path} {i.detail}" for i in issues[:3]), path))
+                _checked(rt, number)
                 continue
             detail.update(source="human-approved", review_ids=review_ids, reviewers=reviewers,
                           generator="human", judge="human", decision={"blocks": [], "consistency": []},
@@ -251,14 +374,17 @@ def poll_external(rt) -> list[str]:
                                                                               if r in head_files},
                                                "deleted": [r for r in rels if r not in head_files], "evidence": []})
         rt.ledger.stage_intake(owner, lifecycle.repo, changeset_id, kind=KIND, status=status, verdicts=[],
-                               human_reason="" if status in ("pr_open", "shadow_recorded")
-                               else f"external PR #{number}: " + "; ".join(detail["decision"].get("reasons", [])),
+                               # the author fixes it (told in the findings comment): not people's queue
+                               human_reason="",
                                drafted_events=[], detail=detail)
         rt.ledger.update_changeset(changeset_id, pr_number=number, head_sha=head)
         rt.trace("decision", context={"repo": lifecycle.repo, "changeset_id": changeset_id, "pr": number,
                                       "playbook": "kb-external", "step": "gate"},
                  result={"status": status, "source": detail["source"],
                          "reviewers": detail.get("reviewers", [])})
+        post_findings(rt, lifecycle, number, head, status,
+                      [] if status in ("pr_open", "shadow_recorded") else _decision_reasons(detail["decision"]))
+        _checked(rt, number)
         events.append(f"external {changeset_id} PR #{number} {detail['source']} {status}")
     return events
 
@@ -274,14 +400,15 @@ def _lifecycle_bookkeeping(result, head_files: dict[str, str]) -> dict:
             "purges": list(result.purged)}
 
 
-def _to_people(rt, lifecycle, number: int, head: str, reason: str, path: str) -> str:
-    """Record a head that people must handle, once (as a human change set)."""
+def _to_people(rt, lifecycle, number: int, head: str, reason: str, path: str, findings_via=None) -> str:
+    """Record a head the gate cannot take as is; the author is told why."""
     repo = lifecycle.repo if lifecycle is not None else next(iter(rt.registry))
     changeset_id = rt.ledger.new_changeset_id(repo, KIND)
     rt.save_changeset_files(changeset_id, {"base_sha": "", "files": {}, "deleted": [], "evidence": []})
     rt.ledger.stage_intake(rt.lease_owner, repo, changeset_id, kind=KIND, status="human", verdicts=[],
-                           human_reason=reason, drafted_events=[],
+                           human_reason="", drafted_events=[],
                            detail={"external_pr": number, "reason": reason, "path": path})
     rt.ledger.update_changeset(changeset_id, pr_number=number, head_sha=head)
+    post_findings(rt, lifecycle or findings_via, number, head, "human", [reason])
     return f"external {changeset_id} PR #{number} to people"
 

src/infermatrix_copilot/kb_service/ledger.py

@@ -282,6 +282,12 @@ def get_cursor(self, repo: str, name: str) -> str | None:
             "SELECT value FROM cursors WHERE repo=? AND name=?", (repo, name)).fetchone()
         return row["value"] if row else None
 
+    def cursors_with_prefix(self, repo: str, prefix: str) -> dict[str, str]:
+        rows = self._conn.execute("SELECT name, value FROM cursors WHERE repo=? AND name LIKE ? ESCAPE '\\'",
+                                  (repo, prefix.replace("\\", "\\\\").replace("%", "\\%")
+                                   .replace("_", "\\_") + "%")).fetchall()
+        return {row["name"]: row["value"] for row in rows}
+
     def set_cursor(self, repo: str, name: str, value: str) -> None:
         with self.tx() as cur:
             cur.execute(

src/infermatrix_copilot/kb_service/merge.py

@@ -96,6 +96,11 @@ def apply_acks(rt, acks: list[dict]) -> None:
     late ack for a superseded duplicate, or one arriving after the change set
     moved on, changes nothing."""
     for ack in acks:
+        if ack.get("kind") == "post_findings" and "invalid" not in ack:
+            from .external import findings_delivered
+
+            findings_delivered(rt, ack)
+            continue
         changeset_id = str(ack.get("changeset_id") or "")
         if not changeset_id or "invalid" in ack:
             continue
@@ -255,8 +260,15 @@ def _apply_merge_ack(rt, changeset: dict, ack: dict) -> None:
         else:
             rt.ledger.update_changeset(changeset_id, status="gate_failed", pending_item=None,
                                        detail={**changeset["detail"], "gate_problems": problems})
-            rt.ledger.enqueue_human(repo, f"local gate refused PR #{changeset['pr_number']}: {text[:400]}",
-                                    changeset_id)
+            if changeset["kind"] == "external" and not any(m in text for m in CONFIG_MERGE):
+                # someone else's PR: its author is told why (checked again daily)
+                from .external import post_findings
+
+                post_findings(rt, rt.registry.get(repo), int(changeset["pr_number"]), str(changeset["head_sha"]),
+                              "gate_failed", [str(p) for p in problems] or [text[:400]])
+            else:
+                rt.ledger.enqueue_human(repo, f"local gate refused PR #{changeset['pr_number']}: {text[:400]}",
+                                        changeset_id)
         _outcome(rt, changeset, "gate_failed", problems=problems[:20])
 
 

src/infermatrix_copilot/kb_service/outbox.py

@@ -37,12 +37,12 @@
 
 ITEM_KINDS = (
     "open_pr", "open_companion_pr", "post_verdict", "enqueue",
-    "update_branch", "pause", "close", "open_issue", "merge",
+    "update_branch", "pause", "close", "open_issue", "merge", "post_findings",
 )
 ALWAYS_EXECUTABLE = frozenset({"pause", "close"})
 ITEM_TTL = {
     "open_pr": 24 * 3600, "open_companion_pr": 24 * 3600,
-    "post_verdict": 30 * 60, "enqueue": 30 * 60, "merge": 30 * 60,
+    "post_verdict": 30 * 60, "enqueue": 30 * 60, "merge": 30 * 60, "post_findings": 24 * 3600,
     "update_branch": 24 * 3600, "pause": 24 * 3600, "close": 24 * 3600,
     "open_issue": 7 * 24 * 3600,
 }