|
| 1 | +"""Companion PRs: update references outside knowledge/ before a retirement lands. |
| 2 | +
|
| 3 | +When a knowledge change retires, replaces or purges rules that files outside |
| 4 | +``knowledge/`` still cite (skills, plugins, adapters, docs, playbooks), the |
| 5 | +gate sends it to people with "references outside knowledge/ need a companion |
| 6 | +PR". Instead of stopping there, the service drafts that companion PR itself: |
| 7 | +each citing file is rewritten (a replaced rule's ID becomes its replacement; |
| 8 | +a list or table entry citing a retired rule is dropped; any other mention of |
| 9 | +it is removed). The companion is opened ONLY as a draft labelled |
| 10 | +``kb:companion`` through the restricted ``open_companion_pr`` path; people |
| 11 | +review, ready and merge it. The knowledge change waits as |
| 12 | +``companion_pending`` (its rules stay active meanwhile, so nothing ever cites |
| 13 | +a retired rule), and is rebuilt on current main once the companion merged. |
| 14 | +""" |
| 15 | + |
| 16 | +from __future__ import annotations |
| 17 | + |
| 18 | +import re |
| 19 | + |
| 20 | +COMPANION_REASON = "references outside knowledge/ need a companion PR" |
| 21 | +COMPANION_DIRS = ("skills/", "plugins/", "adapters/", "doc/", "playbooks/") |
| 22 | + |
| 23 | + |
| 24 | +def needs_companion(decision) -> bool: |
| 25 | + """The change is otherwise fine: its only problem is external citations.""" |
| 26 | + return (decision.status == "human" and len(decision.reasons) == 1 |
| 27 | + and decision.reasons[0].startswith(COMPANION_REASON) and bool(decision.l1.external_refs)) |
| 28 | + |
| 29 | + |
| 30 | +def rewrite(text: str, rule_id: str, replacement: str | None) -> str: |
| 31 | + token = re.compile(rf"(?<![\w-]){re.escape(rule_id)}(?![\w-])") |
| 32 | + if replacement: |
| 33 | + return token.sub(replacement, text) |
| 34 | + out = [] |
| 35 | + for line in text.splitlines(keepends=True): |
| 36 | + if token.search(line): |
| 37 | + if line.lstrip().startswith(("- ", "* ", "|")): |
| 38 | + continue # a list or table entry about the retired rule |
| 39 | + indent = line[:len(line) - len(line.lstrip(" \t"))] # YAML nesting must survive |
| 40 | + line = indent + re.sub(r"[ \t]{2,}", " ", token.sub("", line[len(indent):])) |
| 41 | + out.append(line) |
| 42 | + return "".join(out) |
| 43 | + |
| 44 | + |
| 45 | +def companion_files(external: dict[str, str], refs, operations) -> dict[str, str]: |
| 46 | + replacements = {op.rule_id: op.new_rule_id for op in operations if op.kind == "replace" and op.new_rule_id} |
| 47 | + files: dict[str, str] = {} |
| 48 | + for rule_id, path in refs: |
| 49 | + if not path.startswith(COMPANION_DIRS): |
| 50 | + continue # never outside the companion whitelist |
| 51 | + text = files.get(path, external[path]) |
| 52 | + files[path] = rewrite(text, rule_id, replacements.get(rule_id)) |
| 53 | + return {path: text for path, text in files.items() if text != external[path]} |
| 54 | + |
| 55 | + |
| 56 | +def stage_companion(rt, lifecycle, owner: str, changeset_id: str, operations, decision, |
| 57 | + external: dict[str, str], base_sha: str) -> str | None: |
| 58 | + """Stage (and hand over) the companion for ``changeset_id``; the knowledge |
| 59 | + change set then waits as companion_pending. None when nothing can be |
| 60 | + rewritten automatically (it stays with people).""" |
| 61 | + files = companion_files(external, decision.l1.external_refs, operations) |
| 62 | + if not files: |
| 63 | + return None |
| 64 | + companion_id = rt.ledger.new_changeset_id(lifecycle.repo, "companion") |
| 65 | + rt.save_changeset_files(companion_id, {"base_sha": base_sha, "files": files, "deleted": [], "evidence": []}) |
| 66 | + rt.ledger.stage_intake(owner, lifecycle.repo, companion_id, kind="companion", status="companion_staged", |
| 67 | + verdicts=[], human_reason=f"review and merge the companion PR for {changeset_id}", |
| 68 | + drafted_events=[], detail={"for_changeset": changeset_id, "base_sha": base_sha, |
| 69 | + "refs": [list(r) for r in decision.l1.external_refs]}) |
| 70 | + waiting = rt.ledger.changeset(changeset_id) |
| 71 | + rt.ledger.update_changeset(changeset_id, status="companion_pending", |
| 72 | + detail={**waiting["detail"], "companion": companion_id}) |
| 73 | + publish_companion(rt, lifecycle, companion_id) |
| 74 | + return companion_id |
| 75 | + |
| 76 | + |
| 77 | +def publish_companion(rt, lifecycle, companion_id: str) -> str: |
| 78 | + changeset = rt.ledger.changeset(companion_id) |
| 79 | + if changeset["status"] != "companion_staged": |
| 80 | + return changeset["status"] |
| 81 | + if rt.outbox is None or not lifecycle.auto_merge or not lifecycle.publishes: |
| 82 | + rt.ledger.update_changeset(companion_id, status="shadow_recorded") |
| 83 | + return "shadow_recorded" |
| 84 | + data = rt.load_changeset_files(companion_id) |
| 85 | + from .merge import issue_once |
| 86 | + |
| 87 | + issue_once(rt, lifecycle.repo, changeset, "open_companion_pr", { |
| 88 | + "changeset_id": companion_id, "base_sha": data["base_sha"], |
| 89 | + "branch": f"kb/{lifecycle.repo}/{companion_id}", "files": data["files"], "deleted": [], |
| 90 | + "title": f"knowledge({lifecycle.repo}): update references for {changeset['detail']['for_changeset']}", |
| 91 | + "body": ("Companion to a knowledge change that retires or replaces rules cited here " |
| 92 | + f"(`{changeset['detail']['for_changeset']}`). Drafted by the Copilot knowledge service; " |
| 93 | + "review, mark ready and merge it by hand. The knowledge change is rebuilt once this merges."), |
| 94 | + }) |
| 95 | + rt.ledger.update_changeset(companion_id, status="pr_requested") |
| 96 | + return "pr_requested" |
0 commit comments