diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 44666f8d..e7d16e44 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/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(`force=True` 用于测试与手动)。当天对每个打开的 +他人知识 PR:当前 head 已通过并在合并流程中的不再评审;未通过(`failed`/`human`/`gate_failed`/上下文失效/head 变化等)的每天重新评审, +作者推送多少次都只在下一次每日检查时评审。每次评审(含本地门禁拒绝)都签发 `post_findings`:发布器在该 PR 上只维护一条带 +`` 标记、由自己账号发出的评论,已有则编辑、没有才创建,从不编辑他人伪造的同标记评论。未通过的外部 PR 不进 +人工队列(作者负责修改),也从不合并;影子模式与私有上游不发布评论;跨多个仓库的 PR 经其中一个 +公开且为 `auto_merge` 的仓库发布评论(只要其中有私有上游就不发布)。`post_findings` 失败时不回执,每轮重试直至成功;每条发现评论以签发时间为修订号, +发布器记住每个 PR 已发布的最新修订,较旧的项不会覆盖较新的评论;服务在账本(全局 `findings:`,每个 PR 一条,与其涉及的仓库 +范围无关)记下最新一条直至发布器确认送达,过期仍未送达的以相同内容与原修订号重新签发(与 PR 当前状态无关),因此旧发现永远 +不会覆盖新的;发布器的修订记录原子写入,损坏时视为无记录。 diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md index 35607035..003dfcfa 100644 --- a/doc/features/knowledge-service.md +++ b/doc/features/knowledge-service.md @@ -64,7 +64,8 @@ infermatrix-copilot kb rollback --to # 回指到较早的快照(下一 合并流程(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`。 diff --git a/src/infermatrix_copilot/kb_service/external.py b/src/infermatrix_copilot/kb_service/external.py index 75bbc181..f4826f15 100644 --- a/src/infermatrix_copilot/kb_service/external.py +++ b/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 = "" +# + 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" diff --git a/src/infermatrix_copilot/kb_service/ledger.py b/src/infermatrix_copilot/kb_service/ledger.py index 693314ff..e5dbcb6d 100644 --- a/src/infermatrix_copilot/kb_service/ledger.py +++ b/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( diff --git a/src/infermatrix_copilot/kb_service/merge.py b/src/infermatrix_copilot/kb_service/merge.py index 36131586..cfba6080 100644 --- a/src/infermatrix_copilot/kb_service/merge.py +++ b/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]) diff --git a/src/infermatrix_copilot/kb_service/outbox.py b/src/infermatrix_copilot/kb_service/outbox.py index 88c32482..cabc42cf 100644 --- a/src/infermatrix_copilot/kb_service/outbox.py +++ b/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, } diff --git a/src/infermatrix_copilot/kb_service/publisher.py b/src/infermatrix_copilot/kb_service/publisher.py index b2f51240..2c460524 100644 --- a/src/infermatrix_copilot/kb_service/publisher.py +++ b/src/infermatrix_copilot/kb_service/publisher.py @@ -57,8 +57,9 @@ HOLD_LABEL = "kb:hold" PUSHING = frozenset({"open_pr", "open_companion_pr"}) # retried until done, never acked as failed: stops (pause, close) and reports -# (open_issue: a transient GitHub error must not lose a sweep report) -RETRIED = frozenset({"pause", "close", "open_issue"}) +# (open_issue, post_findings: a transient GitHub error must not lose a sweep +# report or an author's findings) +RETRIED = frozenset({"pause", "close", "open_issue", "post_findings"}) class MergeUncertain(RuntimeError): @@ -671,6 +672,44 @@ def _cancel_pending_merge(self, number: int) -> bool: return after.get("state") == "OPEN" and not after.get("mergeQueueEntry") \ and not after.get("autoMergeRequest") + def _login(self) -> str: + if not getattr(self, "_me", ""): + self._me = json.loads(self.github.gh("api", "user"))["login"] + return self._me + + def _do_post_findings(self, item: OutboxItem) -> dict: + """ONE findings comment per PR, edited in place on every daily check: + the comment of ours that carries the marker is updated, else created.""" + number, marker = int(item.body["pr"]), str(item.body["marker"]) + comment = str(item.body["comment"]) + if marker not in comment: + raise PublishError("a findings comment must carry its marker") + revision = float(item.body.get("revision") or item.issued_at) + posted = self.state_dir / "findings" / f"{number}.json" + try: + latest = float(json.loads(posted.read_text(encoding="utf-8"))["revision"]) if posted.is_file() else 0.0 + except (OSError, ValueError, KeyError, TypeError): + latest = 0.0 # unreadable (written atomically, so only by hand): this item decides + self._trace(event="findings_record_unreadable", pr=number) + if latest > revision: + return {"pr": number, "superseded": True} # a newer findings comment is already there + existing = json.loads(self.github.gh("api", "--paginate", "--slurp", + f"repos/{self.github.repository}/issues/{number}/comments")) + flat = [c for page in existing for c in (page if isinstance(page, list) else [page])] + mine = [c for c in flat if marker in str(c.get("body") or "") + and (c.get("user") or {}).get("login") == self._login()] + if mine: + self.github.gh("api", "-X", "PATCH", f"repos/{self.github.repository}/issues/comments/{mine[0]['id']}", + "-F", "body=@-", input=comment) + else: + self.github.gh("pr", "comment", str(number), "--repo", self.github.repository, "--body-file", "-", + input=comment) + posted.parent.mkdir(parents=True, exist_ok=True) + tmp = posted.with_name(f".{posted.name}.{os.getpid()}.tmp") + tmp.write_text(json.dumps({"revision": revision, "item": item.id}), encoding="utf-8") + os.replace(tmp, posted) # atomic: a crash leaves the old record or the new one + return {"pr": number} + def _do_close(self, item: OutboxItem) -> dict: number = str(item.body["pr"]) if self.github.pr(int(number)).get("state") == "OPEN": diff --git a/src/infermatrix_copilot/kb_service/refine.py b/src/infermatrix_copilot/kb_service/refine.py index 875cf58b..42489fa3 100644 --- a/src/infermatrix_copilot/kb_service/refine.py +++ b/src/infermatrix_copilot/kb_service/refine.py @@ -26,8 +26,12 @@ def reasons_of(changeset: dict) -> list[str]: """Everything the gate said against this change, most specific first.""" detail = changeset["detail"] - reasons = [str(p) for p in detail.get("gate_problems") or []] - decision = detail.get("decision") or {} + return list(dict.fromkeys([str(p) for p in detail.get("gate_problems") or []] + + reasons_from_decision(detail.get("decision") or {})))[:30] + + +def reasons_from_decision(decision: dict) -> list[str]: + reasons: list[str] = [] for issue in decision.get("l1_issues") or []: reasons.append(f"L1 {issue.get('code')} {issue.get('path', '')} {issue.get('detail', '')}".strip()) for block in decision.get("blocks") or []: diff --git a/test/test_kb_external.py b/test/test_kb_external.py index 77a5b24c..bcbf7d1c 100644 --- a/test/test_kb_external.py +++ b/test/test_kb_external.py @@ -80,6 +80,15 @@ def _add_rule_files(): base, [Op("add", PAGE, "DEMO-2a", _rule("DEMO-2a", "PR #11"))], release="v1", today="2026-09-28").files.items()} +def _daily(rt): + """One daily pass (forced: in real time a day passes between two).""" + return poll_external(rt, force=True) + + +def _findings(tmp_path, number): + return [item["body"]["comment"] for item in _items(tmp_path, "post_findings") if item["body"]["pr"] == number] + + def _verdict(tmp_path, rt): envelope = _items(tmp_path, "merge")[-1]["body"]["verdict"] return verify("kb-gate-verdict", envelope, load_public_key(public_key_text(rt.outbox._key.public_key()))) @@ -88,11 +97,11 @@ def _verdict(tmp_path, rt): def test_a_human_knowledge_pr_passes_the_auto_gate_and_gets_a_verdict(tmp_path): rt, lifecycle, origin = _setup(tmp_path) head = _open_human_pr(rt, origin, 7, _add_rule_files()) - (event,) = poll_external(rt) + (event,) = _daily(rt) assert "auto pr_open" in event (changeset,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] assert changeset["pr_number"] == 7 and changeset["head_sha"] == head - assert poll_external(rt) == [] # judged once per head + assert _daily(rt) == [] # judged once per head assert merge.advance(rt, lifecycle) == [f"merge issued {changeset['id']}"] verdict = _verdict(tmp_path, rt) merge_base = _git(origin, "merge-base", "main", head) @@ -104,19 +113,20 @@ def test_a_human_knowledge_pr_passes_the_auto_gate_and_gets_a_verdict(tmp_path): def test_paths_outside_the_governed_pages_need_a_maintainer(tmp_path): rt, lifecycle, origin = _setup(tmp_path) head = _open_human_pr(rt, origin, 8, {"knowledge/tools/check.md": "tool notes\n"}) - (event,) = poll_external(rt) + (event,) = _daily(rt) assert event.endswith("to people") - assert "kb:human-approved" in rt.ledger.human_queue("demo")[-1]["reason"] - assert poll_external(rt) == [] # once per head + assert rt.ledger.human_queue("demo") == [] # the author is told, not people's queue + assert "kb:human-approved" in _findings(tmp_path, 8)[-1] and "not passed" in _findings(tmp_path, 8)[-1] + assert len(_daily(rt)) == 1 # not passed: checked again the next day # a maintainer approves the CURRENT head and labels it: a human-approved verdict rt.github.open[8]["labels"] = [{"name": "kb:human-approved"}] rt.github.reviews[8] = [{"id": 99, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] head2 = _open_human_pr(rt, origin, 8, {"knowledge/tools/check.md": "tool notes v2\n"}, labels=("kb:human-approved",)) rt.github.reviews[8] = [{"id": 99, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] - assert poll_external(rt) == [] # approval is for the old head: wait + assert _daily(rt) == [] # approval is for the old head: wait rt.github.reviews[8].append({"id": 100, "state": "APPROVED", "commit_id": head2, "user": {"login": "alice"}}) - (event,) = poll_external(rt) + (event,) = _daily(rt) assert "human-approved pr_open" in event (changeset,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] merge.advance(rt, lifecycle) @@ -132,7 +142,7 @@ def test_approvals_by_non_maintainers_or_later_retracted_do_not_count(tmp_path): rt.github.reviews[9] = [{"id": 1, "state": "APPROVED", "commit_id": head, "user": {"login": "mallory"}}, {"id": 2, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}, {"id": 3, "state": "DISMISSED", "commit_id": head, "user": {"login": "alice"}}] - assert poll_external(rt) == [] + assert _daily(rt) == [] def test_a_pr_branched_before_main_changed_the_same_page_must_be_rebased(tmp_path): @@ -142,22 +152,22 @@ def test_a_pr_branched_before_main_changed_the_same_page_must_be_rebased(tmp_pat page.write_text(page.read_text() + "\n\n") _git(origin, "add", "-A") _git(origin, "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-q", "-m", "main moves") - (event,) = poll_external(rt) - assert event.endswith("to people") and "rebased" in rt.ledger.human_queue("demo")[-1]["reason"] + (event,) = _daily(rt) + assert event.endswith("to people") and "rebased" in _findings(tmp_path, 10)[-1] def test_a_new_head_is_judged_again_and_drafts_and_our_own_prs_are_skipped(tmp_path): rt, lifecycle, origin = _setup(tmp_path) _open_human_pr(rt, origin, 11, _add_rule_files(), draft=True) - assert poll_external(rt) == [] + assert _daily(rt) == [] rt.github.open[11]["draft"] = False - assert len(poll_external(rt)) == 1 + assert len(_daily(rt)) == 1 (first,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] head2 = _open_human_pr(rt, origin, 11, {**_add_rule_files(), "knowledge/repos/demo/core/_index.md": "# core\n\n- [rules](rules.md)\n"}) assert merge.advance(rt, lifecycle) == [f"head_changed {first['id']}"] assert rt.ledger.human_queue("demo") == [] # an author pushing is not an incident - (event,) = poll_external(rt) + (event,) = _daily(rt) assert "PR #11" in event assert rt.ledger.changesets("demo", ("pr_open",))[-1]["head_sha"] == head2 # a PR the service opened itself is never judged as external @@ -166,7 +176,7 @@ def test_a_new_head_is_judged_again_and_drafts_and_our_own_prs_are_skipped(tmp_p human_reason="", drafted_events=[]) rt.ledger.update_changeset(ours, pr_number=12) _open_human_pr(rt, origin, 12, _add_rule_files()) - assert all("PR #12" not in event for event in poll_external(rt)) + assert all("PR #12" not in event for event in _daily(rt)) def _local_gate(tmp_path, rt, number): @@ -196,13 +206,13 @@ def test_verdicts_for_external_prs_pass_the_publishers_local_gate(tmp_path): rt, lifecycle, origin = _setup(tmp_path) rt.clock = __import__("time").time # verdict windows are checked against the real clock _open_human_pr(rt, origin, 7, _add_rule_files()) - poll_external(rt) + _daily(rt) merge.advance(rt, lifecycle) assert _local_gate(tmp_path, rt, 7) == [] head = _open_human_pr(rt, origin, 8, {"knowledge/tools/check.md": "tool notes\n"}, labels=("kb:human-approved",)) rt.github.reviews[8] = [{"id": 100, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] - assert [e for e in poll_external(rt) if "PR #7" in e] == [] # a merge in flight is not judged again + assert [e for e in _daily(rt) if "PR #7" in e and "findings" not in e] == [] # not judged again rt.ledger.update_changeset(rt.ledger.changesets("demo", ("merge_requested",))[0]["id"], status="merged") merge.advance(rt, lifecycle) with pytest.raises(VerdictError, match="only auto verdicts merge"): @@ -217,7 +227,7 @@ def test_a_retraction_on_a_later_page_of_reviews_is_seen(tmp_path): rt.github.reviews[20] = [{"id": 1, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}, *noise, {"id": 2, "state": "CHANGES_REQUESTED", "commit_id": head, "user": {"login": "alice"}}] - assert poll_external(rt) == [] + assert _daily(rt) == [] def test_an_auto_verdict_needs_a_current_judge_calibration(tmp_path): @@ -228,25 +238,25 @@ def test_an_auto_verdict_needs_a_current_judge_calibration(tmp_path): rt, lifecycle, origin = _setup(tmp_path) rt.registry["demo"] = replace(rt.registry["demo"], calibration_set="missing") _open_human_pr(rt, origin, 21, _add_rule_files()) - (event,) = poll_external(rt) + (event,) = _daily(rt) assert event.endswith("calibration_required") - assert poll_external(rt) == [] # not re-judged (no paid judge calls) until calibrated + assert _daily(rt) == [] # not re-judged (no paid judge calls) until calibrated _calibrate(tmp_path, rt, rt.registry["demo"]) - (event,) = poll_external(rt) + (event,) = _daily(rt) assert event.endswith("auto pr_open") def test_a_head_sent_to_people_can_be_approved_without_a_new_push(tmp_path): rt, lifecycle, origin = _setup(tmp_path) head = _open_human_pr(rt, origin, 30, {"knowledge/tools/y.md": "y\n"}) - (event,) = poll_external(rt) + (event,) = _daily(rt) assert event.endswith("to people") - assert poll_external(rt) == [] + assert len(_daily(rt)) == 1 # still not passing: checked again the next day rt.github.open[30]["labels"] = [{"name": "kb:human-approved"}] rt.github.reviews[30] = [{"id": 7, "state": "APPROVED", "commit_id": head, "user": {"login": "alice"}}] - (event,) = poll_external(rt) + (event,) = _daily(rt) assert "human-approved pr_open" in event - assert poll_external(rt) == [] + assert _daily(rt) == [] def test_rules_an_external_pr_retires_enter_the_retirement_ledger(tmp_path): @@ -258,7 +268,7 @@ def test_rules_an_external_pr_retires_enter_the_retirement_ledger(tmp_path): evidence="PR #13")], release="v1", today="2026-09-28").files _open_human_pr(rt, origin, 41, {f"knowledge/{rel}": text for rel, text in retired.items()}) rt.registry["demo"] = __import__("dataclasses").replace(rt.registry["demo"], retire_ratio=1.0) - events = poll_external(rt) + events = _daily(rt) assert "auto pr_open" in events[0], (events, rt.ledger.human_queue("demo")) (changeset,) = [cs for cs in rt.ledger.changesets("demo", ("pr_open",)) if cs["kind"] == "external"] assert changeset["detail"]["retirements"] == [{"rule_id": "DEMO-1a", "page": PAGE}] @@ -278,5 +288,65 @@ def test_our_own_prs_are_never_judged_as_external_in_any_status(tmp_path): human_reason="", drafted_events=[]) rt.ledger.update_changeset(ours, pr_number=number) _open_human_pr(rt, origin, number, _add_rule_files()) - assert poll_external(rt) == [] + assert _daily(rt) == [] assert not _items(tmp_path, "merge") + + +def test_other_peoples_prs_are_checked_once_a_day_not_on_every_push(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + _open_human_pr(rt, origin, 40, {"knowledge/tools/z.md": "z\n"}) + judged = lambda: [e for e in poll_external(rt) if e.startswith("external ")] # noqa: E731 + assert len(judged()) == 1 # the first pass of the day + _open_human_pr(rt, origin, 40, {"knowledge/tools/z.md": "z2\n"}) # the author pushes + assert judged() == [] # not until the next day + start = rt.clock() + rt.clock = lambda: start + 24 * 3600 + assert len(judged()) == 1 + + +def test_one_findings_comment_says_why_a_pr_is_not_merged(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + _open_human_pr(rt, origin, 41, _add_rule_files()) + (event,) = _daily(rt) + assert "auto pr_open" in event + (comment,) = _findings(tmp_path, 41) + assert comment.startswith("") and "**passed**" in comment + assert not _items(tmp_path, "post_verdict") + + +def test_a_pass_that_fails_half_way_never_checks_a_pr_twice_that_day(tmp_path): + import pytest + + from infermatrix_copilot.kb_service import external + + rt, lifecycle, origin = _setup(tmp_path) + _open_human_pr(rt, origin, 50, {"knowledge/tools/a.md": "a\n"}) + _open_human_pr(rt, origin, 51, _add_rule_files()) + real = external.run_gate + external.run_gate = lambda **kw: (_ for _ in ()).throw(RuntimeError("judge crashed")) + try: + with pytest.raises(RuntimeError): + poll_external(rt) # PR 50 was checked, then PR 51 crashed + finally: + external.run_gate = real + events = poll_external(rt) # the scheduler's next tick, same day + assert [e for e in events if "PR #50" in e] == [] and any("PR #51" in e for e in events) + + +def test_a_pr_spanning_repositories_is_told_through_one_that_publishes(tmp_path): + from dataclasses import replace + + rt, lifecycle, origin = _setup(tmp_path) + rt.registry["other"] = replace(lifecycle, repo="other", full_name="org/other", knowledge_dir="repos/other") + rt.ledger.ensure_repo("other", "auto_merge") + files = {**_add_rule_files(), "knowledge/repos/other/x.md": "# x\n"} + _open_human_pr(rt, origin, 60, files) + (event,) = _daily(rt) + assert event.endswith("to people") + (comment,) = _findings(tmp_path, 60) + assert "spans several repositories" in comment + # with a private upstream among its scopes nothing public may be written + rt.registry["other"] = replace(rt.registry["other"], upstream_visibility="private", mode="shadow") + _open_human_pr(rt, origin, 61, files) + _daily(rt) + assert _findings(tmp_path, 61) == [] diff --git a/test/test_kb_publisher.py b/test/test_kb_publisher.py index d4564b5c..71128cc9 100644 --- a/test/test_kb_publisher.py +++ b/test/test_kb_publisher.py @@ -41,6 +41,8 @@ def __init__(self, origin: Path): self.fail_disable_auto = 0 # the next N `--disable-auto` calls fail self.fail_dequeue = 0 # the next N dequeue mutations fail self.raise_on_query = 0 # the next N GraphQL queries crash the command (OSError) + self.issue_comments: dict[int, list[dict]] = {} + self.fail_comments = 0 # the next N comment writes fail def _by_branch(self, branch: str) -> list[dict]: return [pr for pr in self.prs.values() if pr["headRefName"] == branch and pr["state"] == "OPEN"] @@ -72,8 +74,24 @@ def __call__(self, argv, *, cwd=None, input=None, stdout=None, stderr=None, chec "baseRefName": pr.get("baseRefName", "main"), "mergeCommit": pr.get("mergeCommit"), "autoMergeRequest": {"mergeMethod": "MERGE"} if pr.get("auto") else None, "labels": [{"name": name} for name in pr["labels"]]}) + elif args[:2] == ["pr", "comment"] and self.fail_comments: + self.fail_comments -= 1 + return subprocess.CompletedProcess(argv, 1, b"", b"HTTP 502") elif args[:2] == ["pr", "comment"]: self.comments.append((int(args[2]), input.decode())) + thread = self.issue_comments.setdefault(int(args[2]), []) + thread.append({"id": 1000 + len(self.comments), "body": input.decode(), "user": {"login": "tzhouam"}}) + elif args[:2] == ["api", "user"]: + out = json.dumps({"login": "tzhouam"}) + elif args[:3] == ["api", "--paginate", "--slurp"]: + number = int(args[3].split("/issues/")[1].split("/")[0]) + out = json.dumps([self.issue_comments.get(number, [])]) + elif args[:3] == ["api", "-X", "PATCH"]: + comment_id = int(args[3].rsplit("/", 1)[1]) + for thread in self.issue_comments.values(): + for comment in thread: + if comment["id"] == comment_id: + comment["body"] = input.decode() elif args[:2] == ["pr", "ready"]: if "--undo" in args and "undo" in self.fail: self.fail.discard("undo") # fails once @@ -602,3 +620,112 @@ def test_acks_carry_the_documented_fields(tmp_path): assert envelope["purpose"] == "kb-ack" assert set(envelope["payload"]) == {"item_id", "kind", "changeset_id", "ok", "pr", "head_sha", "branch", "error"} assert sign # the envelope format is the shared signing module's + + +def test_the_findings_comment_is_created_once_and_then_edited_in_place(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + _collect(rt) + gh.issue_comments[42] = [{"id": 7, "body": " spoofed by someone else", + "user": {"login": "mallory"}}] + from infermatrix_copilot.kb_service.external import post_findings + post_findings(rt, lifecycle, 42, "a" * 40, "failed", ["L1 x"]) + assert pub.run_once()["performed"] == 1 + ours = [c for c in gh.issue_comments[42] if c["user"]["login"] == "tzhouam"] + assert len(ours) == 1 and "- L1 x" in ours[0]["body"] + assert gh.issue_comments[42][0]["body"].endswith("spoofed by someone else") # never edited + post_findings(rt, lifecycle, 42, "b" * 40, "pr_open", []) + assert pub.run_once()["performed"] == 1 + ours = [c for c in gh.issue_comments[42] if c["user"]["login"] == "tzhouam"] + assert len(ours) == 1 and "**passed**" in ours[0]["body"] # the same comment, edited + + +def test_a_findings_comment_that_fails_to_post_is_retried_not_dropped(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + _collect(rt) + from infermatrix_copilot.kb_service.external import post_findings + post_findings(rt, lifecycle, 42, "a" * 40, "failed", ["L1 x"]) + gh.fail_comments = 1 + assert pub.run_once()["failed"] == 1 + assert not list((tmp_path / "state" / "inbox" / "acks").glob("*.json")) # not answered: kept + assert pub.run_once()["performed"] == 1 + assert any("- L1 x" in c["body"] for c in gh.issue_comments[42]) + + +def test_an_older_findings_comment_never_overwrites_a_newer_one(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + _collect(rt) + from infermatrix_copilot.kb_service.external import post_findings + now = rt.clock() + post_findings(rt, lifecycle, 42, "a" * 40, "pr_open", []) # older: passed + rt.clock = lambda: now + 60 + post_findings(rt, lifecycle, 42, "a" * 40, "gate_failed", ["L1 newer"]) # newer: refused + gh.fail_comments = 1 # the older one fails first + pub.run_once() + pub.run_once() # the older one retried + ours = [c for c in gh.issue_comments[42] if c["user"]["login"] == "tzhouam"] + assert len(ours) == 1 and "- L1 newer" in ours[0]["body"] + + +def test_undelivered_findings_are_issued_again_after_they_expire(tmp_path): + from infermatrix_copilot.kb_service.external import post_findings, renew_findings + + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + _collect(rt) + post_findings(rt, lifecycle, 42, "a" * 40, "failed", ["L1 x"]) + assert renew_findings(rt) == [] # not expired yet + now = rt.clock() + rt.clock = pub.clock = rt.outbox._clock = lambda: now + 25 * 3600 # the publisher was down for a day + rt.outbox.refresh_control() + assert renew_findings(rt) == ["findings reissued PR #42"] + pub.run_once() + assert any("- L1 x" in c["body"] for c in gh.issue_comments[42]) + _collect(rt) + rt.clock = pub.clock = rt.outbox._clock = lambda: now + 50 * 3600 + assert renew_findings(rt) == [] # delivered: never again + + +def test_renewed_findings_keep_their_revision_across_a_scope_change(tmp_path): + from dataclasses import replace + + from infermatrix_copilot.kb_service.external import post_findings, renew_findings + + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + _collect(rt) + other = replace(lifecycle, repo="other", full_name="org/other", knowledge_dir="repos/other") + rt.registry["other"] = other + rt.ledger.ensure_repo("other", "auto_merge") + pub.repo_flags = {**pub.repo_flags, "other": (True, True)} + rt.outbox.refresh_control() + now = rt.clock() + post_findings(rt, other, 42, "a" * 40, "failed", ["old: spans A+B"]) # never delivered + rt.clock = rt.outbox._clock = pub.clock = lambda: now + 60 + post_findings(rt, lifecycle, 42, "b" * 40, "gate_failed", ["new: only B"]) # the PR narrowed its scope + for path in (tmp_path / "state" / "outbox").glob("[0-9]*.json"): # the old item was lost + if "old: spans" in path.read_text(): + path.unlink() + pub.run_once() + _collect(rt) + rt.clock = rt.outbox._clock = pub.clock = lambda: now + 30 * 3600 + rt.outbox.refresh_control() + assert renew_findings(rt) == [] # one record per PR: the newest, already delivered + ours = [c for c in gh.issue_comments[42] if c["user"]["login"] == "tzhouam"] + assert len(ours) == 1 and "new: only B" in ours[0]["body"] + + +def test_an_unreadable_revision_record_does_not_block_findings(tmp_path): + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + pub.run_once() + _collect(rt) + record = tmp_path / "publisher" / "findings" / "42.json" + record.parent.mkdir(parents=True, exist_ok=True) + record.write_text('{"revision": 17') # truncated by hand + from infermatrix_copilot.kb_service.external import post_findings + post_findings(rt, lifecycle, 42, "a" * 40, "failed", ["L1 x"]) + assert pub.run_once()["performed"] == 1 + assert any("- L1 x" in c["body"] for c in gh.issue_comments[42]) + assert json.loads(record.read_text())["revision"] > 17 # rewritten whole diff --git a/test/test_kb_refine.py b/test/test_kb_refine.py index ebf28ca8..24b28355 100644 --- a/test/test_kb_refine.py +++ b/test/test_kb_refine.py @@ -127,7 +127,7 @@ def crash_once(cs_id, **fields): if cs["kind"] == "refine"]) == 1 -def test_an_external_pr_refused_by_the_local_gate_goes_to_people(tmp_path): +def test_an_external_pr_refused_by_the_local_gate_is_told_why(tmp_path): rt, lifecycle = _flow_runtime(tmp_path) owner = rt.ledger.acquire_lease("t") cs = rt.ledger.new_changeset_id("demo", "external") @@ -138,7 +138,9 @@ def test_an_external_pr_refused_by_the_local_gate_goes_to_people(tmp_path): merge.apply_acks(rt, [{"item_id": "m1", "kind": "merge", "changeset_id": cs, "ok": False, "error": "local gate: L1 x", "problems": ["L1 x"]}]) assert rt.ledger.changeset(cs)["status"] == "gate_failed" - assert rt.ledger.human_queue("demo") + assert rt.ledger.human_queue("demo") == [] # its author is told, and it is checked again daily + (findings,) = [item["body"] for item in _items(tmp_path, "post_findings")] + assert findings["pr"] == 7 and "- L1 x" in findings["comment"] assert json.dumps(rt.ledger.changeset(cs)["detail"]["gate_problems"]) == '["L1 x"]'