From e9dd6199487f70e5bb83b9d241d69184aa5d6053 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 29 Sep 2026 01:45:25 +0800 Subject: [PATCH] Run the release audit in each sweep and draft the adapter-baseline companion Sweeps already accepted audit hints, but nothing ran the repository's release auditor for them. The design also has the service draft the adapter-baseline update as a companion PR that runs in parallel with the sweep and never blocks it. - sweep_audit.audit_sweep runs the adapter's release.auditor plugin (generated-baseline mode) for the sweep's upstream SHA pair before the sweep; a fallback re-check with no upstream movement is skipped. It reads one fetched revision of the knowledge repository, materialized with git archive (never a checkout or snapshot that may lag main), and the sweep then runs on that same revision. Knowledge document issues become per-page hints for the sweep generator. An audit that fails goes to people once per sweep, and the sweep runs without hints. - Reconciliation findings (the committed adapter baseline lags the audited upstream) become one companion PR per sweep: the plugin names its baseline file (new baseline_file field, relative to the adapter; vllm-omni: release_baseline.yaml). Only the upstream: and inventories: blocks are replaced with the generated values, so everything else, comments included, is kept. A block runs to the next top-level key: a column-0 comment inside it does not end it, and comments just before the next key stay with that key. It is published as a draft through open_companion_pr and goes to people. The sweep never waits for it: the companion uses the sweep's revision (no extra fetch), and any failure drafting it is traced and sent to people once per sweep while the sweep still runs. The scheduler's recovery pass for staged companions is isolated per change set too, so one that keeps failing to publish never blocks later ticks. - The sweep summary reports the number of reconciliation findings. Co-Authored-By: Claude Opus 5.5 --- adapters/vllm_omni/release_audit.py | 4 +- doc/architecture/SPEC/kb_service.md | 7 + .../kb_service/companion.py | 14 +- src/infermatrix_copilot/kb_service/report.py | 2 + .../kb_service/scheduler.py | 25 ++- src/infermatrix_copilot/kb_service/sources.py | 11 ++ src/infermatrix_copilot/kb_service/sweep.py | 12 +- .../kb_service/sweep_audit.py | 115 +++++++++++ test/test_kb_sweep_audit.py | 179 ++++++++++++++++++ test/test_release_audit.py | 2 + 10 files changed, 359 insertions(+), 12 deletions(-) create mode 100644 src/infermatrix_copilot/kb_service/sweep_audit.py create mode 100644 test/test_kb_sweep_audit.py diff --git a/adapters/vllm_omni/release_audit.py b/adapters/vllm_omni/release_audit.py index f2d69421..cc7a06c8 100644 --- a/adapters/vllm_omni/release_audit.py +++ b/adapters/vllm_omni/release_audit.py @@ -745,7 +745,7 @@ def audit_for_knowledge( ) -> dict[str, Any]: """Knowledge-service plugin entry (repo-neutral ``ReleaseAuditor`` contract): generated-baseline audit using this adapter's baseline and manifest.""" - return audit_release( + data = audit_release( upstream_repo=upstream_repo, from_ref=from_ref, to_ref=to_ref, @@ -755,6 +755,8 @@ def audit_for_knowledge( project_root=project_root, baseline_source="generated", ).data + # which adapter file the baseline companion PR updates (relative to this adapter) + return {**data, "baseline_file": "release_baseline.yaml"} def render_summary(report: ReleaseAuditReport) -> str: diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index bae6da13..0c95dc55 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -167,3 +167,10 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝 标签，进入人工队列；发布器拒绝在其上执行 `post_verdict`/`enqueue`（ready 与合并只能由人完成），且拒绝白名单外路径 （`knowledge/`、`.github/`、`src/`、`tools/`）。知识变更集处于 `companion_pending`（规则保持 active）；配套 PR 合并后，调度器 在当前 main 上重建它（外部引用已清零 → 正常过门与发布）；配套 PR 被关闭则知识变更转人工，且不计入熔断。 + +## 2026-09-29 巡检中的发版审计与基线配套 PR +配置了 `release.auditor` 的仓库，调度器在每次巡检前以 generated-baseline 模式运行审计插件（上游 SHA 对未变的回退巡检 +跳过）：知识文档问题按页面转为巡检生成器的提示；审计失败每次巡检只转人工一次，巡检照常进行（无提示）。reconciliation +（提交的适配器基线落后于审计的上游）不阻塞巡检，而是起草一个配套 PR：插件以 `baseline_file`（相对适配器目录）指明基线 +文件，只替换其中的 `upstream:` 与 `inventories:` 块（其余内容与注释保留），经 `open_companion_pr` 以 draft 发布并转人工； +每次巡检最多一个。巡检汇总列出 reconciliation 数量。 diff --git a/src/infermatrix_copilot/kb_service/companion.py b/src/infermatrix_copilot/kb_service/companion.py index 4b1c8872..27a8df2f 100644 --- a/src/infermatrix_copilot/kb_service/companion.py +++ b/src/infermatrix_copilot/kb_service/companion.py @@ -87,10 +87,16 @@ def publish_companion(rt, lifecycle, companion_id: str) -> str: 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."), + "title": (f"knowledge({lifecycle.repo}): update references for {changeset['detail']['for_changeset']}" + if changeset["detail"].get("for_changeset") + else f"adapter({lifecycle.repo}): update the release baseline " + f"({changeset['detail'].get('baseline_for', '')})"), + "body": (("Companion to a knowledge change that retires or replaces rules cited here " + f"(`{changeset['detail']['for_changeset']}`). The knowledge change is rebuilt once this merges.") + if changeset["detail"].get("for_changeset") + else "The committed adapter baseline lags the audited upstream release (reconciliation " + "findings of the release audit); this updates its upstream pin and inventories.") + + " Drafted by the Copilot knowledge service; review, mark ready and merge it by hand.", }) rt.ledger.update_changeset(companion_id, status="pr_requested") return "pr_requested" diff --git a/src/infermatrix_copilot/kb_service/report.py b/src/infermatrix_copilot/kb_service/report.py index 20a6fde5..f8ca9ffd 100644 --- a/src/infermatrix_copilot/kb_service/report.py +++ b/src/infermatrix_copilot/kb_service/report.py @@ -90,6 +90,8 @@ def summary(rt, lifecycle, report: dict) -> tuple[str, str]: f"({', '.join(f'{k}: {v}' for k, v in sorted(statuses.items())) or 'none'}) |", f"| operations | {', '.join(f'{k}: {v}' for k, v in sorted(ops.items())) or 'none'} |", f"| waiting for people | {len(rt.ledger.human_queue(lifecycle.repo))} |", + f"| release audit reconciliation findings | " + f"{'not run' if report.get('reconciliation') is None else report['reconciliation']} |", f"| model calls | {calls} ({tokens['input']} in / {tokens['output']} out tokens, {seconds:.0f}s) |", ] if report.get("breaker"): diff --git a/src/infermatrix_copilot/kb_service/scheduler.py b/src/infermatrix_copilot/kb_service/scheduler.py index 5fa28199..fe27617f 100644 --- a/src/infermatrix_copilot/kb_service/scheduler.py +++ b/src/infermatrix_copilot/kb_service/scheduler.py @@ -31,6 +31,7 @@ from .report import flush_reports from .companion import publish_companion from .external import poll_external +from .sweep_audit import audit_sweep, stage_baseline_companion from .runtime import collect_events, publish, run_intake from .sweep import UpstreamRepo, detect_release, run_sweep @@ -129,7 +130,11 @@ def _repo_tick(self, lifecycle) -> None: 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"]) + try: # a companion that cannot be published yet never blocks the rest of the tick + status = publish_companion(rt, lifecycle, changeset["id"]) + except Exception as exc: + self._record(lifecycle.repo, "error", error=f"companion {changeset['id']}: {exc!r}") + continue 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) @@ -141,7 +146,23 @@ def _repo_tick(self, lifecycle) -> None: upstream = UpstreamRepo(rt.state_dir / "upstream" / f"{lifecycle.repo}.git", lifecycle.full_name) sweep = detect_release(rt, lifecycle, upstream) if sweep is not None: - report = run_sweep(rt, lifecycle, rt.lease_owner, sweep, upstream) + # the release audit feeds the sweep hints; baseline drift becomes a + # companion PR that runs in parallel and never blocks the sweep + base_sha = rt.knowledge.fetch() # one revision for the audit and the sweep + hints, audit = audit_sweep(rt, lifecycle, sweep, upstream, base_sha) + try: # the baseline companion never blocks the sweep + stage_baseline_companion(rt, lifecycle, rt.lease_owner, sweep, audit, base_sha) + except Exception as exc: + self._record(lifecycle.repo, "error", error=f"baseline companion: {exc!r}", + trace=traceback.format_exc()[-2000:]) + key = f"baseline_companion_failed:{sweep.get('tag')}:{sweep.get('to_sha')}" + if not rt.ledger.get_cursor(lifecycle.repo, key): + rt.ledger.set_cursor(lifecycle.repo, key, "1") + rt.ledger.enqueue_human(lifecycle.repo, + f"could not draft the adapter baseline update: {exc}") + report = run_sweep(rt, lifecycle, rt.lease_owner, sweep, upstream, audit_hints=hints, + reconciliation=len(audit.reconciliation) if audit is not None else None, + base_sha=base_sha) for changeset_id in report["changesets"]: publish(rt, lifecycle, changeset_id) diff --git a/src/infermatrix_copilot/kb_service/sources.py b/src/infermatrix_copilot/kb_service/sources.py index b6432aac..0155e61e 100644 --- a/src/infermatrix_copilot/kb_service/sources.py +++ b/src/infermatrix_copilot/kb_service/sources.py @@ -92,6 +92,17 @@ def show(self, rev: str, path: str) -> str | None: except SourceError: return None + def export(self, rev: str, dest: str | Path) -> Path: + """Materialize the tree at ``rev`` under ``dest`` (no checkout involved).""" + import io + import tarfile + + dest = Path(dest) + dest.mkdir(parents=True, exist_ok=True) + with tarfile.open(fileobj=io.BytesIO(self._git("archive", "--format=tar", rev))) as tar: + tar.extractall(dest, filter="data") + return dest + def raw_manifest(self, old: str, new: str) -> list[dict]: """Every changed path as a kb-gate manifest entry (renames are D + A), the exact form the verifier recomputes.""" diff --git a/src/infermatrix_copilot/kb_service/sweep.py b/src/infermatrix_copilot/kb_service/sweep.py index 91b97f7c..54c6089f 100644 --- a/src/infermatrix_copilot/kb_service/sweep.py +++ b/src/infermatrix_copilot/kb_service/sweep.py @@ -277,7 +277,8 @@ def _progress(rt, lifecycle, sweep: dict) -> dict: def run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo, - audit_hints: dict[str, list[dict]] | None = None) -> dict: + audit_hints: dict[str, list[dict]] | None = None, reconciliation: int | None = None, + base_sha: str | None = None) -> dict: """Traced: every model call and decision of the sweep carries its release.""" from ..trace_store import trace_context @@ -289,7 +290,7 @@ def run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo, run_id = sweep_run_id(lifecycle.repo, sweep, progress) with trace_context(playbook="kb-sweep", repo=lifecycle.repo, release=str(sweep.get("tag") or ""), run_id=run_id): - report = _run_sweep(rt, lifecycle, owner, sweep, upstream, audit_hints) + report = _run_sweep(rt, lifecycle, owner, sweep, upstream, audit_hints, reconciliation, base_sha) report["run_id"] = run_id return report @@ -304,7 +305,8 @@ def draft_key_for_page(repo: str, sweep: dict, page: str) -> str: def _run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo, - audit_hints: dict[str, list[dict]] | None = None) -> dict: + audit_hints: dict[str, list[dict]] | None = None, reconciliation: int | None = None, + base_sha: str | None = None) -> dict: from ..trace_store import accepted_key, trace_context accepted: dict[str, str] = {} # page -> the accepted generator call's key @@ -317,13 +319,13 @@ def _run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo, to re-check rules against.""" from .runtime import gate_and_stage - base_sha = rt.knowledge.fetch() + base_sha = base_sha or rt.knowledge.fetch() # the revision the audit read, when there was one base = rt.knowledge.knowledge_files(base_sha) external = rt.knowledge.external_texts(base_sha) release, today = sweep["tag"] or rt.release_for(lifecycle.repo), rt.today() progress = _progress(rt, lifecycle, sweep) report = {"sweep": sweep, "t1": structural_report(base, lifecycle.knowledge_dir), - "changesets": [], "skipped_pages": [], "failed_pages": []} + "changesets": [], "skipped_pages": [], "failed_pages": [], "reconciliation": reconciliation} staged: list[tuple[list, object, list, str]] = [] # (ops, result, evidence, page key) t1 = report["t1"] if not progress["t1_done"]: diff --git a/src/infermatrix_copilot/kb_service/sweep_audit.py b/src/infermatrix_copilot/kb_service/sweep_audit.py new file mode 100644 index 00000000..95f486c7 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/sweep_audit.py @@ -0,0 +1,115 @@ +"""The release audit inside each sweep, and the adapter-baseline companion PR. + +A repository whose adapter configures ``release.auditor`` has its plugin run +(generated-baseline mode) for the sweep's upstream SHA pair: + +* knowledge-document issues become per-page hints for the sweep's generator; +* reconciliation findings (the committed adapter baseline lags the audited + upstream) do not block the sweep. Instead the service drafts the baseline + update as a companion PR: the plugin names its baseline file + (``baseline_file``, relative to the adapter) and the audit supplies the + generated values. Only the ``upstream:`` and ``inventories:`` blocks are + replaced, so everything else in the file (comments included) is kept. + People review and merge it; the sweep never waits for it. + +An audit that fails goes to people once per sweep; the sweep runs without hints. +""" + +from __future__ import annotations + +import tempfile + +import yaml + +from ..knowledge_service.release_audit import ReleaseAuditPluginError, run_release_audit + +BLOCKS = ("upstream", "inventories") + + +def audit_sweep(rt, lifecycle, sweep: dict, upstream, base_sha: str) -> tuple[dict[str, list[dict]], object | None]: + """(hints per knowledge-relative page, the audit result or None). The audit + reads ``base_sha`` of the knowledge repository, materialized for it: the + same revision the sweep then runs on (never a checkout or snapshot that + may lag main).""" + if not lifecycle.release.auditor or lifecycle.adapter_dir is None: + return {}, None + if not sweep.get("from_sha") or sweep.get("from_sha") == sweep.get("to_sha"): + return {}, None # nothing moved upstream (a fallback re-check) + try: + with tempfile.TemporaryDirectory(prefix="kb-audit-") as scratch: + project = rt.knowledge.export(base_sha, scratch) + result = run_release_audit( + lifecycle.adapter_dir, lifecycle.release.auditor, upstream_repo=upstream.path, + from_ref=sweep["from_sha"], to_ref=sweep["to_sha"], knowledge_root=project / "knowledge", + project_root=project) + except (ReleaseAuditPluginError, Exception) as exc: # noqa: BLE001 - never blocks the sweep + key = f"audit_failed:{sweep.get('tag')}:{sweep.get('to_sha')}" + if not rt.ledger.get_cursor(lifecycle.repo, key): + rt.ledger.set_cursor(lifecycle.repo, key, "1") + rt.ledger.enqueue_human(lifecycle.repo, f"release audit for {sweep.get('tag')} failed: {exc}") + return {}, None + hints: dict[str, list[dict]] = {} + for issue in result.issues: + page = str(issue.get("document") or "").removeprefix("knowledge/") + if page: + hints.setdefault(page, []).append(issue) + return hints, result + + +def _replace_block(text: str, key: str, value) -> str: + """Replace one top-level YAML block. It runs to the next top-level key; a + column-0 comment inside it belongs to it, but comments and blank lines + right before the next key belong to that key and are kept.""" + dumped = yaml.safe_dump({key: value}, sort_keys=False, default_flow_style=False) + lines = text.splitlines(keepends=True) + start = next((i for i, line in enumerate(lines) if line.startswith(f"{key}:")), None) + if start is None: + return text.rstrip("\n") + "\n\n" + dumped + end = next((i for i in range(start + 1, len(lines)) + if lines[i][:1] not in (" ", "\t", "#", "\n", "\r", "")), len(lines)) + while end - 1 > start and (not lines[end - 1].strip() or lines[end - 1].startswith("#")): + end -= 1 + tail = "".join(lines[end:]) + return "".join(lines[:start]) + dumped + ("\n" if tail and not tail.startswith("\n") else "") + tail + + +def baseline_text(committed: str, generated: dict) -> str: + text = committed + for key in BLOCKS: + if key in generated: + text = _replace_block(text, key, generated[key]) + return text + + +def stage_baseline_companion(rt, lifecycle, owner: str, sweep: dict, result, base_sha: str) -> str | None: + """Draft the adapter-baseline update once per sweep; None when not needed.""" + from .companion import publish_companion + + if result is None or not result.reconciliation or lifecycle.adapter_dir is None: + return None + baseline_file = str(result.data.get("baseline_file") or "") + if not baseline_file or ".." in baseline_file.split("/") or baseline_file.startswith("/"): + return None + key = f"baseline:{sweep.get('tag')}:{sweep.get('to_sha')}" + for changeset in rt.ledger.changesets(lifecycle.repo, ("companion_staged", "pr_requested", "companion_open", + "merged", "closed", "shadow_recorded")): + if changeset["kind"] == "companion" and changeset["detail"].get("baseline_for") == key: + return None # already drafted for this sweep + path = f"adapters/{lifecycle.adapter_dir.name}/{baseline_file}" + main = base_sha # the revision the audit and the sweep use + committed = rt.knowledge.show(main, path) + if committed is None: + return None + updated = baseline_text(committed, result.generated_baseline) + if updated == committed: + return None + companion_id = rt.ledger.new_changeset_id(lifecycle.repo, "companion") + rt.save_changeset_files(companion_id, {"base_sha": main, "files": {path: updated}, "deleted": [], + "evidence": []}) + rt.ledger.stage_intake(owner, lifecycle.repo, companion_id, kind="companion", status="companion_staged", + verdicts=[], drafted_events=[], + human_reason=f"review and merge the adapter baseline update for {sweep.get('tag')}", + detail={"for_changeset": "", "baseline_for": key, "base_sha": main, + "reconciliation": list(result.reconciliation)[:50]}) + publish_companion(rt, lifecycle, companion_id) + return companion_id diff --git a/test/test_kb_sweep_audit.py b/test/test_kb_sweep_audit.py new file mode 100644 index 00000000..4064063c --- /dev/null +++ b/test/test_kb_sweep_audit.py @@ -0,0 +1,179 @@ +"""The release audit in each sweep: page hints and the adapter-baseline companion.""" + +from __future__ import annotations + +import subprocess +from dataclasses import replace + +import yaml + +from infermatrix_copilot.kb_service.config import ReleaseConfig +from infermatrix_copilot.kb_service.sweep_audit import audit_sweep, baseline_text, stage_baseline_companion +from infermatrix_copilot.knowledge_service.release_audit import ReleaseAuditResult +from test_kb_flow import _flow_runtime, _items + +COMMITTED = """schema_version: 1 + +upstream: + repository: org/up + audited_sha: aaaa + +# fingerprints are generated +inventories: + models: + count: 1 + sha256: old + +path_owners: + core: + - core/ +""" +SWEEP = {"tag": "v2", "from_sha": "a" * 40, "to_sha": "b" * 40, "reason": "release"} + + +class Upstream: + path = "/nonexistent" + + +def test_only_the_generated_blocks_change_and_comments_survive(): + text = baseline_text(COMMITTED, {"upstream": {"repository": "org/up", "audited_sha": "bbbb"}, + "inventories": {"models": {"count": 2, "sha256": "new"}}}) + assert "# fingerprints are generated" in text and "path_owners:\n core:\n - core/\n" in text + data = yaml.safe_load(text) + assert data["upstream"]["audited_sha"] == "bbbb" and data["inventories"]["models"]["count"] == 2 + assert data["path_owners"] == {"core": ["core/"]} and data["schema_version"] == 1 + + +def _with_plugin(tmp_path, body: str): + rt, lifecycle = _flow_runtime(tmp_path) + adapter = tmp_path / "demoadapter" + adapter.mkdir() + (adapter / "plugin.py").write_text(body) + lifecycle = replace(lifecycle, adapter_dir=adapter, release=ReleaseConfig(trigger="github_release", + auditor="plugin.py")) + rt.registry["demo"] = lifecycle + return rt, lifecycle + + +PLUGIN = ''' +def audit_for_knowledge(*, upstream_repo, from_ref, to_ref, knowledge_root, project_root): + return {"upstream": {"from": {"sha": from_ref}, "to": {"sha": to_ref}}, + "issues": [{"kind": "stale_knowledge_source", "document": "knowledge/repos/demo/core/rules.md", + "detail": "gone"}], + "reconciliation": [{"kind": "baseline_pin_mismatch"}], + "generated_baseline": {"upstream": {"repository": "org/up", "audited_sha": to_ref}, + "inventories": {"models": {"count": 2, "sha256": "new"}}}, + "baseline_file": "release_baseline.yaml"} +''' + + +def test_audit_issues_become_page_hints(tmp_path): + rt, lifecycle = _with_plugin(tmp_path, PLUGIN) + hints, result = audit_sweep(rt, lifecycle, SWEEP, Upstream(), rt.knowledge.fetch()) + assert list(hints) == ["repos/demo/core/rules.md"] and hints["repos/demo/core/rules.md"][0]["detail"] == "gone" + assert len(result.reconciliation) == 1 + assert audit_sweep(rt, lifecycle, {**SWEEP, "from_sha": "b" * 40}, Upstream(), rt.knowledge.fetch()) == ({}, None) # nothing moved + + +def test_a_failing_audit_goes_to_people_once_and_never_blocks(tmp_path): + rt, lifecycle = _with_plugin(tmp_path, "def audit_for_knowledge(**kw):\n raise RuntimeError('boom')\n") + assert audit_sweep(rt, lifecycle, SWEEP, Upstream(), rt.knowledge.fetch()) == ({}, None) + assert audit_sweep(rt, lifecycle, SWEEP, Upstream(), rt.knowledge.fetch()) == ({}, None) + assert [i for i in rt.ledger.human_queue("demo") if "release audit" in i["reason"]].__len__() == 1 + + +def test_baseline_drift_becomes_one_companion_pr_per_sweep(tmp_path): + rt, lifecycle = _with_plugin(tmp_path, PLUGIN) + origin = tmp_path / "origin" + target = origin / "adapters" / "demoadapter" / "release_baseline.yaml" + target.parent.mkdir(parents=True) + target.write_text(COMMITTED) + subprocess.run(["git", "-C", str(origin), "add", "-A"], check=True) + subprocess.run(["git", "-C", str(origin), "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-qm", "b"], + check=True) + rt.lease_owner = rt.ledger.acquire_lease("scheduler") + _hints, result = audit_sweep(rt, lifecycle, SWEEP, Upstream(), rt.knowledge.fetch()) + companion_id = stage_baseline_companion(rt, lifecycle, rt.lease_owner, SWEEP, result, rt.knowledge.fetch()) + assert companion_id is not None + (item,) = _items(tmp_path, "open_companion_pr") + (path,) = item["body"]["files"] + assert path == "adapters/demoadapter/release_baseline.yaml" + assert f"audited_sha: {'b' * 40}" in item["body"]["files"][path] and "# fingerprints are generated" in item["body"]["files"][path] + assert "release baseline" in item["body"]["title"] + assert stage_baseline_companion(rt, lifecycle, rt.lease_owner, SWEEP, result, rt.knowledge.fetch()) is None # once per sweep + assert any("baseline update" in i["reason"] for i in rt.ledger.human_queue("demo")) + + +def test_no_reconciliation_means_no_companion(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + clean = ReleaseAuditResult("a" * 40, "b" * 40, (), (), {}, {"baseline_file": "release_baseline.yaml"}) + assert stage_baseline_companion(rt, lifecycle, "owner", SWEEP, clean, "0" * 40) is None + + +def test_the_audit_reads_the_fetched_revision_not_a_stale_checkout(tmp_path): + plugin = PLUGIN.replace('"detail": "gone"', + '"detail": (__import__("pathlib").Path(knowledge_root) / "repos/demo/core/rules.md").read_text()[-30:]') + rt, lifecycle = _with_plugin(tmp_path, plugin) + origin = tmp_path / "origin" + page = origin / "knowledge" / "repos" / "demo" / "core" / "rules.md" + page.write_text(page.read_text() + "\n\n") + subprocess.run(["git", "-C", str(origin), "-c", "user.name=t", "-c", "user.email=t@e", "commit", "-qam", "m"], + check=True) + hints, _result = audit_sweep(rt, lifecycle, SWEEP, Upstream(), rt.knowledge.fetch()) + assert "only on remote main" in hints["repos/demo/core/rules.md"][0]["detail"] + + +def test_comments_inside_a_block_do_not_split_it(): + committed = "upstream:\n repository: org/up\n# pinned by hand\n audited_sha: aaaa\n\n# next\ninventories: {}\n" + text = baseline_text(committed, {"upstream": {"repository": "org/up", "audited_sha": "bbbb"}}) + assert "aaaa" not in text and yaml.safe_load(text)["upstream"]["audited_sha"] == "bbbb" + assert "# next\ninventories: {}" in text + + +def test_a_failing_baseline_companion_never_blocks_the_sweep(tmp_path, monkeypatch): + from infermatrix_copilot.kb_service import scheduler as scheduler_module + from infermatrix_copilot.kb_service.scheduler import Scheduler + + rt, lifecycle = _with_plugin(tmp_path, PLUGIN) + lifecycle = replace(lifecycle, full_name="org/up") + rt.registry["demo"] = lifecycle + ran = [] + + def boom(*args, **kwargs): + raise RuntimeError("github down") + + monkeypatch.setattr(scheduler_module, "detect_release", lambda *a, **k: dict(SWEEP)) + monkeypatch.setattr(scheduler_module, "audit_sweep", lambda *a, **k: ({"p.md": []}, None)) + monkeypatch.setattr(scheduler_module, "stage_baseline_companion", boom) + monkeypatch.setattr(scheduler_module, "run_sweep", lambda *a, **k: ran.append(k) or { + "changesets": [], "breaker": "", "complete": False}) + Scheduler(rt, intake_every=10**9)._repo_tick(lifecycle) + assert ran and ran[0]["audit_hints"] == {"p.md": []} + assert any("baseline update" in i["reason"] for i in rt.ledger.human_queue("demo")) + + +def test_a_companion_that_keeps_failing_to_publish_never_blocks_later_sweeps(tmp_path, monkeypatch): + from infermatrix_copilot.kb_service import scheduler as scheduler_module + from infermatrix_copilot.kb_service.scheduler import Scheduler + + rt, lifecycle = _with_plugin(tmp_path, PLUGIN) + lifecycle = replace(lifecycle, full_name="org/up") + rt.registry["demo"] = lifecycle + rt.lease_owner = rt.ledger.acquire_lease("scheduler") + staged = rt.ledger.new_changeset_id("demo", "companion") + rt.ledger.stage_intake(rt.lease_owner, "demo", staged, kind="companion", status="companion_staged", + verdicts=[], human_reason="", drafted_events=[], detail={"for_changeset": ""}) + ran = [] + + def boom(*args, **kwargs): + raise RuntimeError("outbox unavailable") + + monkeypatch.setattr(scheduler_module, "publish_companion", boom) + monkeypatch.setattr(scheduler_module, "detect_release", lambda *a, **k: dict(SWEEP)) + monkeypatch.setattr(scheduler_module, "audit_sweep", lambda *a, **k: ({}, None)) + monkeypatch.setattr(scheduler_module, "run_sweep", lambda *a, **k: ran.append(1) or { + "changesets": [], "breaker": "", "complete": False}) + scheduler = Scheduler(rt, intake_every=10**9, release_every=0) + scheduler._repo_tick(lifecycle) + scheduler._repo_tick(lifecycle) # the second tick after the failure + assert len(ran) == 2 diff --git a/test/test_release_audit.py b/test/test_release_audit.py index 5cf00f26..91cec03b 100644 --- a/test/test_release_audit.py +++ b/test/test_release_audit.py @@ -451,6 +451,8 @@ def test_repo_neutral_plugin_loader(tmp_path): project_root=fixture["project"], ) assert result.to_sha == fixture["new"] + assert result.data["baseline_file"] == "release_baseline.yaml" # the baseline companion's target + assert (adapter / result.data["baseline_file"]).is_file() kinds = {i["kind"] for i in result.issues} # the real adapter baseline is used, so its owner documents are absent # from this fixture project; only knowledge-document kinds are enforced