From 85b6bbd9624ed6ce404b1a8a62d529372f2ef09d Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 29 Sep 2026 01:21:09 +0800 Subject: [PATCH] Draft companion PRs for citations outside knowledge/ before a retirement lands When a knowledge change retires, replaces or purges rules that files outside knowledge/ still cite, the design has the service draft a companion PR that updates those citations. The knowledge change waits (its rules stay active, so nothing ever cites a retired rule) and resumes automatically once the companion merges. Until now such changes just went to people. - companion.stage_companion runs when the gate's only problem is external citations (not for held sweep change sets or forced-human ones). Each citing file is rewritten: a replaced rule's ID becomes its replacement; a list or table entry citing a retired rule is dropped; any other mention loses the ID (indentation kept, so nested YAML stays valid). Only the companion whitelist (skills, plugins, adapters, doc, playbooks) is touched. The knowledge change set waits as companion_pending. - The companion goes out through open_companion_pr as a draft labelled kb:companion, and into the human queue. The publisher refuses any path outside the companion whitelist (knowledge/, .github/, src/, tools/) and refuses post_verdict or enqueue on a companion, so only people can ready and merge it. Companions are recognized by their branch as well as the label, so a label that failed to apply cannot open that door (fail closed). A companion staged but not handed over (a crash) is published by the scheduler's recovery pass. - The scheduler follows the companion. On merge it rebuilds the waiting change on current main (the citations are gone, so it passes the gate and is published; a change set that never had a PR is superseded directly). The companion is marked merged only once every waiting change was handled, so a failed fetch or a crash retries the rebuild next pass. The rebuild waits while the repository (or everything) is paused. The change a companion belongs to comes from the companion itself (for_changeset), so a crash between staging the companion and linking the change still rebuilds it on merge. If the companion is closed unmerged, the change goes to people. Companion closures do not count toward the overturn breaker. Co-Authored-By: Claude Opus 5.5 --- doc/architecture/SPEC/kb_service.md | 8 + .../kb_service/companion.py | 96 +++++++ src/infermatrix_copilot/kb_service/merge.py | 48 +++- .../kb_service/publisher.py | 29 ++- src/infermatrix_copilot/kb_service/runtime.py | 6 + .../kb_service/scheduler.py | 7 +- test/test_kb_companion.py | 245 ++++++++++++++++++ 7 files changed, 429 insertions(+), 10 deletions(-) create mode 100644 src/infermatrix_copilot/kb_service/companion.py create mode 100644 test/test_kb_companion.py diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 8417248d..bae6da13 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -159,3 +159,11 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝 哈希）显式记录而非依赖时间戳：blob 不可变只归档一次，增长的记录文件再次归档，归档期间才落盘的 blob 进入下一个归档，并写 `.sha256` 清单（首行为包的哈希，其后每个成员一行；清单最后写入， 标志归档完整）。发布器（GPU 盒，经已有的 SSH 通道）每轮拉取本地没有的归档，只有哈希校验通过才保存到 `$KB_PUBLISHER_STATE/archive/`，否则记入 trace 并拒绝。恢复：按名称顺序解开到空目录后运行 `TraceStore.rebuild_index`。 + +## 2026-09-29 配套 PR（外部引用） +质量门唯一的问题是"知识目录外引用了本次退役/替换/删除的规则"时（非巡检暂存、非强制人工），`companion.stage_companion` +自动起草配套变更：被替换规则的 ID 改为新 ID；引用已退役规则的列表项/表格行删除，其余提及去掉 ID；只改配套白名单 +（`skills/`、`plugins/`、`adapters/`、`doc/`、`playbooks/`）。配套 PR 经 `open_companion_pr` 以 draft 创建并打 `kb:companion` +标签，进入人工队列；发布器拒绝在其上执行 `post_verdict`/`enqueue`（ready 与合并只能由人完成），且拒绝白名单外路径 +（`knowledge/`、`.github/`、`src/`、`tools/`）。知识变更集处于 `companion_pending`（规则保持 active）；配套 PR 合并后，调度器 +在当前 main 上重建它（外部引用已清零 → 正常过门与发布）；配套 PR 被关闭则知识变更转人工，且不计入熔断。 diff --git a/src/infermatrix_copilot/kb_service/companion.py b/src/infermatrix_copilot/kb_service/companion.py new file mode 100644 index 00000000..4b1c8872 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/companion.py @@ -0,0 +1,96 @@ +"""Companion PRs: update references outside knowledge/ before a retirement lands. + +When a knowledge change retires, replaces or purges rules that files outside +``knowledge/`` still cite (skills, plugins, adapters, docs, playbooks), the +gate sends it to people with "references outside knowledge/ need a companion +PR". Instead of stopping there, the service drafts that companion PR itself: +each citing file is rewritten (a replaced rule's ID becomes its replacement; +a list or table entry citing a retired rule is dropped; any other mention of +it is removed). The companion is opened ONLY as a draft labelled +``kb:companion`` through the restricted ``open_companion_pr`` path; people +review, ready and merge it. The knowledge change waits as +``companion_pending`` (its rules stay active meanwhile, so nothing ever cites +a retired rule), and is rebuilt on current main once the companion merged. +""" + +from __future__ import annotations + +import re + +COMPANION_REASON = "references outside knowledge/ need a companion PR" +COMPANION_DIRS = ("skills/", "plugins/", "adapters/", "doc/", "playbooks/") + + +def needs_companion(decision) -> bool: + """The change is otherwise fine: its only problem is external citations.""" + return (decision.status == "human" and len(decision.reasons) == 1 + and decision.reasons[0].startswith(COMPANION_REASON) and bool(decision.l1.external_refs)) + + +def rewrite(text: str, rule_id: str, replacement: str | None) -> str: + token = re.compile(rf"(? dict[str, str]: + replacements = {op.rule_id: op.new_rule_id for op in operations if op.kind == "replace" and op.new_rule_id} + files: dict[str, str] = {} + for rule_id, path in refs: + if not path.startswith(COMPANION_DIRS): + continue # never outside the companion whitelist + text = files.get(path, external[path]) + files[path] = rewrite(text, rule_id, replacements.get(rule_id)) + return {path: text for path, text in files.items() if text != external[path]} + + +def stage_companion(rt, lifecycle, owner: str, changeset_id: str, operations, decision, + external: dict[str, str], base_sha: str) -> str | None: + """Stage (and hand over) the companion for ``changeset_id``; the knowledge + change set then waits as companion_pending. None when nothing can be + rewritten automatically (it stays with people).""" + files = companion_files(external, decision.l1.external_refs, operations) + if not files: + return None + companion_id = rt.ledger.new_changeset_id(lifecycle.repo, "companion") + rt.save_changeset_files(companion_id, {"base_sha": base_sha, "files": files, "deleted": [], "evidence": []}) + rt.ledger.stage_intake(owner, lifecycle.repo, companion_id, kind="companion", status="companion_staged", + verdicts=[], human_reason=f"review and merge the companion PR for {changeset_id}", + drafted_events=[], detail={"for_changeset": changeset_id, "base_sha": base_sha, + "refs": [list(r) for r in decision.l1.external_refs]}) + waiting = rt.ledger.changeset(changeset_id) + rt.ledger.update_changeset(changeset_id, status="companion_pending", + detail={**waiting["detail"], "companion": companion_id}) + publish_companion(rt, lifecycle, companion_id) + return companion_id + + +def publish_companion(rt, lifecycle, companion_id: str) -> str: + changeset = rt.ledger.changeset(companion_id) + if changeset["status"] != "companion_staged": + return changeset["status"] + if rt.outbox is None or not lifecycle.auto_merge or not lifecycle.publishes: + rt.ledger.update_changeset(companion_id, status="shadow_recorded") + return "shadow_recorded" + data = rt.load_changeset_files(companion_id) + from .merge import issue_once + + issue_once(rt, lifecycle.repo, changeset, "open_companion_pr", { + "changeset_id": companion_id, "base_sha": data["base_sha"], + "branch": f"kb/{lifecycle.repo}/{companion_id}", "files": data["files"], "deleted": [], + "title": f"knowledge({lifecycle.repo}): update references for {changeset['detail']['for_changeset']}", + "body": ("Companion to a knowledge change that retires or replaces rules cited here " + f"(`{changeset['detail']['for_changeset']}`). Drafted by the Copilot knowledge service; " + "review, mark ready and merge it by hand. The knowledge change is rebuilt once this merges."), + }) + rt.ledger.update_changeset(companion_id, status="pr_requested") + return "pr_requested" diff --git a/src/infermatrix_copilot/kb_service/merge.py b/src/infermatrix_copilot/kb_service/merge.py index 60776ead..d9b75fcb 100644 --- a/src/infermatrix_copilot/kb_service/merge.py +++ b/src/infermatrix_copilot/kb_service/merge.py @@ -53,6 +53,7 @@ def knowledge_repository() -> str: # the transition each outbox item kind drives: (state before, state after) TRANSITIONS = { "open_pr": ("pr_requested", "pr_open"), + "open_companion_pr": ("pr_requested", "companion_open"), "post_verdict": ("pr_open", "verdict_posted"), "enqueue": ("verdict_posted", "queued"), } @@ -135,7 +136,7 @@ def apply_acks(rt, acks: list[dict]) -> None: changeset_id) continue fields = {"status": after, "pending_item": None} - if kind == "open_pr": + if kind in ("open_pr", "open_companion_pr"): fields.update(pr_number=int(ack["pr"]), head_sha=str(ack["head_sha"]), branch=str(ack.get("branch", ""))) elif kind == "enqueue": @@ -305,10 +306,12 @@ def to_people(reason: str) -> None: # the rebuild itself did not pass the gate: nothing replaces the PR to_people(f"rebuild {new_id} on current main did not pass the gate ({staged['status']})") return None - rt.ledger.update_changeset(changeset["id"], status="superseding", + no_pr = changeset.get("pr_number") is None + rt.ledger.update_changeset(changeset["id"], status="superseded" if no_pr else "superseding", detail={**detail, "rebuilt_as": new_id, "superseded_because": why}) _outcome(rt, changeset, "superseded", reason=why, replaced_by=new_id) - _request_close(rt, lifecycle.repo, rt.ledger.changeset(changeset["id"])) + if not no_pr: + _request_close(rt, lifecycle.repo, rt.ledger.changeset(changeset["id"])) publish(rt, lifecycle, new_id) return new_id @@ -332,7 +335,8 @@ def advance(rt, lifecycle) -> list[str]: queued = rt.ledger.changesets(lifecycle.repo, ("queued",)) + [ cs for cs in rt.ledger.changesets(lifecycle.repo, ("verdict_posted",)) if _enqueue_outstanding(rt, cs) or cs["detail"].get("withdrawn")] - for changeset in rt.ledger.changesets(lifecycle.repo, (*IN_FLIGHT, "superseding", "paused")): + for changeset in rt.ledger.changesets(lifecycle.repo, (*IN_FLIGHT, "superseding", "paused", + "companion_open")): number = changeset["pr_number"] if number is None: continue @@ -344,6 +348,42 @@ def advance(rt, lifecycle) -> list[str]: _pause_unconfirmed(rt, lifecycle.repo, changeset, events) events.append(f"observe_failed {changeset['id']}") continue + if changeset["status"] == "companion_open": + # people review and merge companions; the service only follows them + waiting = [cs for cs in rt.ledger.changesets(lifecycle.repo, ("companion_pending",)) + if cs["detail"].get("companion") == changeset["id"]] + target = changeset["detail"].get("for_changeset") + if target and not waiting: + # the companion names its change; a crash between staging the + # companion and linking the change leaves it "human" and unlinked + parent = rt.ledger.changeset(target) + if parent["status"] in ("companion_pending", "human") and not parent["detail"].get("rebuilt_as"): + waiting = [parent] + if pr.get("merged"): + if waiting and (paused or not getattr(rt, "lease_owner", None)): + continue # rebuild only unpaused and under the scheduler's lease: next pass + events.append(f"companion_merged {changeset['id']}") + retry = False + for knowledge in waiting: + try: + new_id = rebuild(rt, lifecycle, knowledge, "its companion PR merged") + except Exception as exc: # noqa: BLE001 - e.g. main not fetchable: next pass + retry = True + events.append(f"rebuild_retry {knowledge['id']}: {exc}") + continue + events.append(f"rebuilt {knowledge['id']} as {new_id}" if new_id + else f"rebuild_failed {knowledge['id']}") + if not retry: # only once every waiting change was handled + rt.ledger.update_changeset(changeset["id"], status="merged", + merge_sha=str(pr.get("merge_commit_sha") or "")) + elif pr.get("state") == "closed": + rt.ledger.update_changeset(changeset["id"], status="closed") + for knowledge in waiting: + rt.ledger.update_changeset(knowledge["id"], status="human") + rt.ledger.enqueue_human(lifecycle.repo, f"companion PR #{number} was closed unmerged", + knowledge["id"]) + events.append(f"companion_closed {changeset['id']}") + continue if changeset["status"] == "superseding": # the replaced PR stays tracked until GitHub shows it closed; the # close item is re-issued whenever the previous one expired diff --git a/src/infermatrix_copilot/kb_service/publisher.py b/src/infermatrix_copilot/kb_service/publisher.py index 0d3a368c..c3304761 100644 --- a/src/infermatrix_copilot/kb_service/publisher.py +++ b/src/infermatrix_copilot/kb_service/publisher.py @@ -47,6 +47,12 @@ ARCHIVE_NAME = re.compile(r"traces-\d{6}-\d{8}-\d{6}\.tar\.gz") GOVERNED_PATH = re.compile(r"knowledge/(?:repos/[A-Za-z0-9._-]+|general)/(?:[A-Za-z0-9._-]+/)*[A-Za-z0-9._-]+\.(?:md|yaml)") BRANCH = re.compile(r"kb/[A-Za-z0-9._-]+/[A-Za-z0-9._-]+") +# a companion may only touch these (never knowledge/, .github/, src/, tools/) +COMPANION_PATH = re.compile(r"(?:skills|plugins|adapters|doc|playbooks)/(?:[A-Za-z0-9._-]+/)*[A-Za-z0-9._-]+") +COMPANION_LABEL = "kb:companion" +# companions are recognized by their branch too (kb//-companion-), +# so a label that failed to apply cannot make one mergeable by automation +COMPANION_BRANCH = re.compile(r"kb/[A-Za-z0-9._-]+/[A-Za-z0-9._-]+-companion-[0-9a-f]+") HOLD_LABEL = "kb:hold" PUSHING = frozenset({"open_pr", "open_companion_pr"}) # retried until done, never acked as failed: stops (pause, close) and reports @@ -327,7 +333,8 @@ def _checked_head(self, item: OutboxItem) -> dict: raise PublishError(f"PR #{item.body['pr']} head moved to {str(pr.get('headRefOid'))[:12]}") return pr - def _do_open_pr(self, item: OutboxItem, *, draft: bool = False) -> dict: + def _do_open_pr(self, item: OutboxItem, *, draft: bool = False, allowed: re.Pattern = GOVERNED_PATH, + what: str = "governed knowledge pages") -> dict: body = item.body branch = str(body["branch"]) if not BRANCH.fullmatch(branch): @@ -335,8 +342,8 @@ def _do_open_pr(self, item: OutboxItem, *, draft: bool = False) -> dict: files: dict[str, str] = dict(body.get("files") or {}) deleted = [str(p) for p in body.get("deleted") or []] for path in [*files, *deleted]: - if not GOVERNED_PATH.fullmatch(path) or ".." in path.split("/"): - raise PublishError(f"refusing to write outside governed knowledge pages: {path}") + if not allowed.fullmatch(path) or ".." in path.split("/"): + raise PublishError(f"refusing to write outside {what}: {path}") commit = self._build_commit(str(body["base_sha"]), files, deleted, str(body["title"]), when=item.issued_at) existing = json.loads(self.github.gh("pr", "list", "--repo", self.github.repository, "--head", branch, @@ -365,13 +372,24 @@ def _do_open_pr(self, item: OutboxItem, *, draft: bool = False) -> dict: return {"pr": int(opened[0]["number"]), "head_sha": commit, "branch": branch} def _do_open_companion_pr(self, item: OutboxItem) -> dict: - return self._do_open_pr(item, draft=True) # companions are always drafts for people + """Always a draft, labelled kb:companion, and only within the companion + whitelist: people review, ready and merge it; automation never does.""" + result = self._do_open_pr(item, draft=True, allowed=COMPANION_PATH, + what="the companion whitelist (skills, plugins, adapters, doc, playbooks)") + self.github.gh("pr", "edit", str(result["pr"]), "--repo", self.github.repository, + "--add-label", COMPANION_LABEL, ok_fail=True) + return result + + def _refuse_companion(self, pr: dict) -> None: + labelled = COMPANION_LABEL in {str(label.get("name")) for label in pr.get("labels") or []} + if labelled or COMPANION_BRANCH.fullmatch(str(pr.get("headRefName") or "")): + raise PublishError("a companion PR is readied and merged only by people") def _do_post_verdict(self, item: OutboxItem) -> dict: """A verdict is only issued for an unpaused repository, under its current generation, so posting one also lifts an earlier pause's kb:hold label (first, so the precheck the comment triggers no longer sees it).""" - self._checked_head(item) + self._refuse_companion(self._checked_head(item)) number = str(item.body["pr"]) self.github.gh("pr", "edit", number, "--repo", self.github.repository, "--remove-label", HOLD_LABEL, ok_fail=True) # fails when the label is absent; what matters is checked next @@ -383,6 +401,7 @@ def _do_post_verdict(self, item: OutboxItem) -> dict: def _do_enqueue(self, item: OutboxItem) -> dict: pr = self._checked_head(item) + self._refuse_companion(pr) number = str(item.body["pr"]) if pr.get("isDraft"): # a resumed PR was turned into a draft by its pause self.github.gh("pr", "ready", number, "--repo", self.github.repository) diff --git a/src/infermatrix_copilot/kb_service/runtime.py b/src/infermatrix_copilot/kb_service/runtime.py index 5602e733..d1017898 100644 --- a/src/infermatrix_copilot/kb_service/runtime.py +++ b/src/infermatrix_copilot/kb_service/runtime.py @@ -297,6 +297,12 @@ def gate_and_stage(rt: KbRuntime, lifecycle: RepoLifecycle, owner: str, *, kind: "consistency": [{"owner_dir": c["owner_dir"], "verdict": c["verdict"]} for c in decision.consistency], "l1_issues": [i.to_dict() for i in decision.l1.issues]}) + from .companion import needs_companion, stage_companion + + if needs_companion(decision) and not hold and not force_human and operations: + # the change is fine except for citations outside knowledge/: draft the + # companion PR that updates them; this change waits for it + stage_companion(rt, lifecycle, owner, changeset_id, operations, decision, external, base_sha) return changeset_id diff --git a/src/infermatrix_copilot/kb_service/scheduler.py b/src/infermatrix_copilot/kb_service/scheduler.py index 5979bb22..5fa28199 100644 --- a/src/infermatrix_copilot/kb_service/scheduler.py +++ b/src/infermatrix_copilot/kb_service/scheduler.py @@ -29,6 +29,7 @@ from .archive import make_archive from .audit import audit_main from .report import flush_reports +from .companion import publish_companion from .external import poll_external from .runtime import collect_events, publish, run_intake from .sweep import UpstreamRepo, detect_release, run_sweep @@ -127,6 +128,9 @@ def _repo_tick(self, lifecycle) -> None: for changeset in rt.ledger.changesets(lifecycle.repo, ("gated",)): status = publish(rt, lifecycle, changeset["id"]) self._record(lifecycle.repo, "published", changeset=changeset["id"], status=status) + for changeset in rt.ledger.changesets(lifecycle.repo, ("companion_staged",)): + status = publish_companion(rt, lifecycle, changeset["id"]) + self._record(lifecycle.repo, "published", changeset=changeset["id"], status=status) if self._due(f"intake:{lifecycle.repo}", self.intake_every): new = collect_events(rt, lifecycle) changeset_id = run_intake(rt, lifecycle) @@ -150,7 +154,8 @@ def _overturn_breaker(self, lifecycle) -> None: rt = self.rt # only the service's own PRs: an author closing their own PR is no overturn recent = [cs for cs in rt.ledger.changesets(lifecycle.repo, ("closed",)) - if cs["kind"] != "external" and rt.clock() - float(cs["updated_at"]) < OVERTURN_WINDOW] + if cs["kind"] not in ("external", "companion") + and rt.clock() - float(cs["updated_at"]) < OVERTURN_WINDOW] state = rt.ledger.repo_state(lifecycle.repo) if len(recent) >= OVERTURN_LIMIT and not state["paused"]: self._pause(lifecycle.repo, f"overturn breaker: {len(recent)} knowledge PRs closed unmerged within 24h") diff --git a/test/test_kb_companion.py b/test/test_kb_companion.py new file mode 100644 index 00000000..4e526e4b --- /dev/null +++ b/test/test_kb_companion.py @@ -0,0 +1,245 @@ +"""Companion PRs update citations outside knowledge/ before a retirement lands.""" + +from __future__ import annotations + +import json +import subprocess +from dataclasses import replace + +from infermatrix_copilot.kb_service import merge +from infermatrix_copilot.kb_service.companion import rewrite +from infermatrix_copilot.kb_service.runtime import collect_events, run_intake +from test_kb_flow import _ack, _flow_runtime, _items +from test_kb_intake_gate import PAGE, ScriptedGateway, _judge_all + + +def test_rewrites_point_to_the_replacement_or_drop_the_retired_rule(): + text = "Intro cites DEMO-1a here.\n- DEMO-1a: bounded queue\n| DEMO-1a | x |\nKeep DEMO-1ab alone.\n" + assert rewrite(text, "DEMO-1a", "DEMO-9z") == ( + "Intro cites DEMO-9z here.\n- DEMO-9z: bounded queue\n| DEMO-9z | x |\nKeep DEMO-1ab alone.\n") + assert rewrite(text, "DEMO-1a", None) == "Intro cites here.\nKeep DEMO-1ab alone.\n" + import yaml + + nested = "tools:\n check:\n description: Consult DEMO-1a before merging\n" + rewritten = rewrite(nested, "DEMO-1a", None) + assert rewritten == "tools:\n check:\n description: Consult before merging\n" + assert yaml.safe_load(rewritten)["tools"]["check"]["description"] == "Consult before merging" + + +def _retiring_runtime(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + judge = _judge_all("yes") + + def answer(role, prompt): + if role.name == "generator": + return {"operations": [{"kind": "retire", "page": PAGE, "rule_id": "DEMO-1a", + "reason": "upstream-removed", "evidence": "PR #11"}], "rationale": "r"} + return judge(role, prompt) + + rt.gateway = ScriptedGateway(answer) + lifecycle = replace(lifecycle, retire_ratio=1.0) # retiring the only rule is the point here + rt.registry["demo"] = lifecycle + rt.lease_owner = rt.ledger.acquire_lease("scheduler") + return rt, lifecycle + + +def test_a_retirement_cited_by_a_skill_waits_for_its_companion_then_lands(tmp_path): + from infermatrix_copilot.knowledge_service.signing import generate_private_key, load_public_key, public_key_text + + rt, lifecycle = _retiring_runtime(tmp_path) + publisher = generate_private_key(tmp_path / "publisher.pem") + rt.publisher_public_key = load_public_key(public_key_text(publisher.public_key())) + collect_events(rt, lifecycle) + knowledge_id = run_intake(rt, lifecycle) + knowledge = rt.ledger.changeset(knowledge_id) + assert knowledge["status"] == "companion_pending" + companion_id = knowledge["detail"]["companion"] + (item,) = _items(tmp_path, "open_companion_pr") + assert set(item["body"]["files"]) == {"skills/x.md"} + assert "DEMO-1a" not in item["body"]["files"]["skills/x.md"] + assert _items(tmp_path, "open_pr") == [] # the retirement itself waits + # the publisher opened the draft; people merge it + _ack(tmp_path, rt, publisher, kind="open_companion_pr", changeset_id=companion_id, ok=True, + pr=50, head_sha="c" * 40) + assert rt.ledger.changeset(companion_id)["status"] == "companion_open" + rt.github.prs[50] = {"state": "open", "merged": False, "head": {"sha": "c" * 40}} + assert merge.advance(rt, lifecycle) == [] + origin = tmp_path / "origin" # the merged companion is on main + (origin / "skills" / "x.md").write_text(item["body"]["files"]["skills/x.md"]) + subprocess.run(["git", "-C", str(origin), "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-qam", + "companion"], check=True) + rt.github.prs[50] = {"state": "closed", "merged": True, "merge_commit_sha": "d" * 40, + "head": {"sha": "c" * 40}} + events = merge.advance(rt, lifecycle) + assert events[0] == f"companion_merged {companion_id}" and events[1].startswith(f"rebuilt {knowledge_id} as ") + new_id = events[1].rsplit(" ", 1)[1] + assert rt.ledger.changeset(knowledge_id)["status"] == "superseded" + assert rt.ledger.changeset(new_id)["status"] == "pr_requested" # gated and handed over + assert any(i["body"]["changeset_id"] == new_id for i in _items(tmp_path, "open_pr")) + + +def test_a_closed_companion_hands_the_retirement_to_people(tmp_path): + from infermatrix_copilot.knowledge_service.signing import generate_private_key, load_public_key, public_key_text + + rt, lifecycle = _retiring_runtime(tmp_path) + publisher = generate_private_key(tmp_path / "publisher.pem") + rt.publisher_public_key = load_public_key(public_key_text(publisher.public_key())) + collect_events(rt, lifecycle) + knowledge_id = run_intake(rt, lifecycle) + companion_id = rt.ledger.changeset(knowledge_id)["detail"]["companion"] + _ack(tmp_path, rt, publisher, kind="open_companion_pr", changeset_id=companion_id, ok=True, + pr=51, head_sha="c" * 40) + rt.github.prs[51] = {"state": "closed", "merged": False, "head": {"sha": "c" * 40}} + assert merge.advance(rt, lifecycle) == [f"companion_closed {companion_id}"] + assert rt.ledger.changeset(knowledge_id)["status"] == "human" + assert rt.ledger.changesets("demo", ("closed",))[0]["kind"] == "companion" # not an overturn + + +def test_the_publisher_keeps_companions_to_their_whitelist_and_never_merges_them(tmp_path): + from test_kb_publisher import _setup + + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + for path in list((tmp_path / "state" / "outbox").glob("[0-9]*.json")): + path.unlink() + base = rt.load_changeset_files(changeset_id)["base_sha"] + for i, bad in enumerate(("src/x.py", "knowledge/repos/demo/core/rules.md", ".github/x.yml", "tools/x.py")): + rt.outbox.issue("demo", "open_companion_pr", { + "changeset_id": f"c{i}", "base_sha": base, "branch": f"kb/demo/c{i}", "files": {bad: "x\n"}, + "deleted": [], "title": "t", "body": "b"}) + assert pub.run_once()["failed"] == 4 and gh.created() == 0 + rt.outbox.issue("demo", "open_companion_pr", { + "changeset_id": "cok", "base_sha": base, "branch": "kb/demo/cok", "files": {"skills/x.md": "ok\n"}, + "deleted": [], "title": "t", "body": "b"}) + assert pub.run_once()["performed"] == 1 + (number,) = [n for n, pr in gh.prs.items() if pr["headRefName"] == "kb/demo/cok"] + assert gh.prs[number]["isDraft"] is True and "kb:companion" in gh.prs[number]["labels"] + head = gh.prs[number]["headRefOid"] + rt.outbox.refresh_control() + rt.outbox.issue("demo", "enqueue", {"changeset_id": "cok", "pr": number, "head_sha": head}) + rt.outbox.issue("demo", "post_verdict", {"changeset_id": "cok", "pr": number, "head_sha": head, + "comment": "x"}) + assert pub.run_once()["failed"] == 2 + assert gh.prs[number]["isDraft"] is True and not gh.prs[number]["queued"] + acks = [json.loads(p.read_text())["payload"] for p in (tmp_path / "state" / "inbox" / "acks").glob("*.json")] + assert all("only by people" in a["error"] for a in acks if a["kind"] in ("enqueue", "post_verdict")) + + +def test_a_rebuild_that_fails_to_fetch_is_retried_and_the_companion_stays_open(tmp_path): + from infermatrix_copilot.knowledge_service.signing import generate_private_key, load_public_key, public_key_text + + rt, lifecycle = _retiring_runtime(tmp_path) + publisher = generate_private_key(tmp_path / "publisher.pem") + rt.publisher_public_key = load_public_key(public_key_text(publisher.public_key())) + collect_events(rt, lifecycle) + knowledge_id = run_intake(rt, lifecycle) + companion_id = rt.ledger.changeset(knowledge_id)["detail"]["companion"] + _ack(tmp_path, rt, publisher, kind="open_companion_pr", changeset_id=companion_id, ok=True, + pr=52, head_sha="c" * 40) + rt.github.prs[52] = {"state": "closed", "merged": True, "merge_commit_sha": "d" * 40, + "head": {"sha": "c" * 40}} + real_fetch = rt.knowledge.fetch + + def down(): + raise RuntimeError("git fetch failed") + + rt.knowledge.fetch = down + events = merge.advance(rt, lifecycle) + assert any(e.startswith(f"rebuild_retry {knowledge_id}") for e in events) + assert rt.ledger.changeset(companion_id)["status"] == "companion_open" # still followed + rt.knowledge.fetch = real_fetch + events = merge.advance(rt, lifecycle) + assert any(e.startswith(f"rebuilt {knowledge_id}") or e == f"rebuild_failed {knowledge_id}" for e in events) + assert rt.ledger.changeset(companion_id)["status"] == "merged" + + +def test_a_staged_companion_is_published_after_a_crash(tmp_path): + from infermatrix_copilot.kb_service import companion as companion_module + from infermatrix_copilot.kb_service.scheduler import Scheduler + + rt, lifecycle = _retiring_runtime(tmp_path) + real = companion_module.publish_companion + companion_module.publish_companion = lambda *a, **k: "crashed" # dies before handing over + try: + collect_events(rt, lifecycle) + knowledge_id = run_intake(rt, lifecycle) + finally: + companion_module.publish_companion = real + companion_id = rt.ledger.changeset(knowledge_id)["detail"]["companion"] + assert rt.ledger.changeset(companion_id)["status"] == "companion_staged" + assert _items(tmp_path, "open_companion_pr") == [] + Scheduler(rt)._repo_tick(lifecycle) # the next tick recovers it + assert rt.ledger.changeset(companion_id)["status"] == "pr_requested" + assert len(_items(tmp_path, "open_companion_pr")) == 1 + + +def test_an_unlabelled_companion_is_still_never_merged_by_automation(tmp_path): + from test_kb_publisher import _setup + + rt, lifecycle, changeset_id, pub, gh = _setup(tmp_path) + number = 77 + gh.prs[number] = {"number": number, "state": "OPEN", "isDraft": True, "headRefOid": "e" * 40, + "headRefName": "kb/demo/demo-companion-0123456789ab", "id": "PR_77", "labels": [], + "queued": False, "body": ""} + for path in list((tmp_path / "state" / "outbox").glob("[0-9]*.json")): + path.unlink() + rt.outbox.issue("demo", "enqueue", {"changeset_id": "x", "pr": number, "head_sha": "e" * 40}) + assert pub.run_once()["failed"] == 1 and gh.prs[number]["isDraft"] and not gh.prs[number]["queued"] + + +def _companion_open(tmp_path, rt, lifecycle, publisher, knowledge_id, pr): + companion_id = [cs for cs in rt.ledger.changesets("demo", ("pr_requested", "companion_staged")) + if cs["kind"] == "companion"][0]["id"] + _ack(tmp_path, rt, publisher, kind="open_companion_pr", changeset_id=companion_id, ok=True, + pr=pr, head_sha="c" * 40) + rt.github.prs[pr] = {"state": "closed", "merged": True, "merge_commit_sha": "d" * 40, + "head": {"sha": "c" * 40}} + return companion_id + + +def test_a_companion_merged_during_a_pause_is_rebuilt_after_resume(tmp_path): + from infermatrix_copilot.knowledge_service.signing import generate_private_key, load_public_key, public_key_text + + rt, lifecycle = _retiring_runtime(tmp_path) + publisher = generate_private_key(tmp_path / "publisher.pem") + rt.publisher_public_key = load_public_key(public_key_text(publisher.public_key())) + collect_events(rt, lifecycle) + knowledge_id = run_intake(rt, lifecycle) + companion_id = _companion_open(tmp_path, rt, lifecycle, publisher, knowledge_id, 53) + rt.outbox.transition(lambda: rt.ledger.bump_generation("demo", pause=True, reason="drill"), + public_repos={"demo"}) + assert merge.advance(rt, lifecycle) == [] + assert rt.ledger.changeset(knowledge_id)["status"] == "companion_pending" + rt.outbox.transition(lambda: rt.ledger.resume("demo"), public_repos={"demo"}) + events = merge.advance(rt, lifecycle) + assert any(e.startswith(f"rebuilt {knowledge_id}") or e == f"rebuild_failed {knowledge_id}" for e in events) + assert rt.ledger.changeset(companion_id)["status"] == "merged" + + +def test_a_crash_before_linking_still_rebuilds_the_change_when_the_companion_merges(tmp_path): + from infermatrix_copilot.knowledge_service.signing import generate_private_key, load_public_key, public_key_text + + rt, lifecycle = _retiring_runtime(tmp_path) + publisher = generate_private_key(tmp_path / "publisher.pem") + rt.publisher_public_key = load_public_key(public_key_text(publisher.public_key())) + real_update = rt.ledger.update_changeset + + def crash(changeset_id, **fields): + if fields.get("status") == "companion_pending": + raise RuntimeError("crash before linking") + return real_update(changeset_id, **fields) + + rt.ledger.update_changeset = crash + collect_events(rt, lifecycle) + try: + run_intake(rt, lifecycle) + except RuntimeError: + pass + rt.ledger.update_changeset = real_update + (knowledge,) = [cs for cs in rt.ledger.changesets("demo", ("human",)) if cs["kind"] == "intake"] + assert "companion" not in knowledge["detail"] + from infermatrix_copilot.kb_service.scheduler import Scheduler + Scheduler(rt)._repo_tick(lifecycle) # recovery publishes the staged companion + _companion_open(tmp_path, rt, lifecycle, publisher, knowledge["id"], 54) + events = merge.advance(rt, lifecycle) + assert any(e.startswith(f"rebuilt {knowledge['id']}") or e == f"rebuild_failed {knowledge['id']}" + for e in events)