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