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)