Draft companion PRs for citations outside knowledge/ before a retirement lands

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 被关闭则知识变更转人工，且不计入熔断。

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"(?<![\w-]){re.escape(rule_id)}(?![\w-])")
+    if replacement:
+        return token.sub(replacement, text)
+    out = []
+    for line in text.splitlines(keepends=True):
+        if token.search(line):
+            if line.lstrip().startswith(("- ", "* ", "|")):
+                continue  # a list or table entry about the retired rule
+            indent = line[:len(line) - len(line.lstrip(" \t"))]  # YAML nesting must survive
+            line = indent + re.sub(r"[ \t]{2,}", " ", token.sub("", line[len(indent):]))
+        out.append(line)
+    return "".join(out)
+
+
+def companion_files(external: dict[str, str], refs, operations) -> 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"

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

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/<repo>/<repo>-companion-<id>),
+# 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,16 +333,17 @@ 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):
             raise PublishError(f"refusing branch name {branch!r}")
         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)

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
 
 

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")