diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 1e7c5621..d23760ea 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -146,3 +146,9 @@ stdlib + PyYAML + `cryptography`（`kb` extra）+ `.adapters` + `.knowledge_serv 为 draft 时才记为已确认（draft 不能在合并队列中，也不能再次入队）；否则重发 `pause`。请求 5 分钟后仍未回执或未确认 → 转人工一次，附出队与 `gh pr ready --undo` 命令。`kb status` 报告每个仓库 `pause_unconfirmed` 数量，熔断/回滚在其归零 前不算完成；`kb resume` 清除这些记录。 + +## 2026-09-29 巡检汇总 issue +每次巡检完成（`complete`）后，`report.publish_summary` 把汇总（T1 发现、修复的索引、保留/失败的页面、变更集及其状态、 +各类操作数、待人工处理数、模型调用次数/token/耗时）写入 `/reports/sweep--.md`；仅当仓库为 +`auto_merge` 且上游公开时，另通过新的 outbox 项 `open_issue`（7 天有效）在知识仓库发布一个 issue。发布器按标题复用已有 +issue（同一报告只开一个），标签不存在时不带标签重试，绝不丢报告。 diff --git a/src/infermatrix_copilot/kb_service/outbox.py b/src/infermatrix_copilot/kb_service/outbox.py index 227af429..5e5be17a 100644 --- a/src/infermatrix_copilot/kb_service/outbox.py +++ b/src/infermatrix_copilot/kb_service/outbox.py @@ -37,13 +37,14 @@ ITEM_KINDS = ( "open_pr", "open_companion_pr", "post_verdict", "enqueue", - "update_branch", "pause", "close", + "update_branch", "pause", "close", "open_issue", ) ALWAYS_EXECUTABLE = frozenset({"pause", "close"}) ITEM_TTL = { "open_pr": 24 * 3600, "open_companion_pr": 24 * 3600, "post_verdict": 30 * 60, "enqueue": 30 * 60, "update_branch": 24 * 3600, "pause": 24 * 3600, "close": 24 * 3600, + "open_issue": 7 * 24 * 3600, } CONTROL_MAX_AGE = 10 * 60 HOLDS_MAX_AGE = 10 * 60 diff --git a/src/infermatrix_copilot/kb_service/publisher.py b/src/infermatrix_copilot/kb_service/publisher.py index 9ee35da6..b9b88f93 100644 --- a/src/infermatrix_copilot/kb_service/publisher.py +++ b/src/infermatrix_copilot/kb_service/publisher.py @@ -48,7 +48,9 @@ BRANCH = re.compile(r"kb/[A-Za-z0-9._-]+/[A-Za-z0-9._-]+") HOLD_LABEL = "kb:hold" PUSHING = frozenset({"open_pr", "open_companion_pr"}) -RETRIED = frozenset({"pause", "close"}) # stops: retried until done, never acked as failed +# retried until done, never acked as failed: stops (pause, close) and reports +# (open_issue: a transient GitHub error must not lose a sweep report) +RETRIED = frozenset({"pause", "close", "open_issue"}) class PublishError(RuntimeError): @@ -366,6 +368,24 @@ def _do_update_branch(self, item: OutboxItem) -> dict: "-f", f"expected_head_sha={item.body['head_sha']}") return {"pr": int(item.body["pr"])} + def _do_open_issue(self, item: OutboxItem) -> dict: + """One issue per title (a sweep report): an existing one is reused.""" + title = str(item.body["title"]) + found = json.loads(self.github.gh("issue", "list", "--repo", self.github.repository, "--state", "all", + "--search", f'in:title "{title}"', "--json", "number,title")) + for issue in found: + if issue.get("title") == title: + return {"pr": int(issue["number"])} + args = ["issue", "create", "--repo", self.github.repository, "--title", title, "--body-file", "-"] + for label in item.body.get("labels") or []: + args += ["--label", str(label)] + out = self.github.gh(*args, input=str(item.body.get("body") or ""), ok_fail=True) + if not out.strip(): # a missing label must not lose the report + out = self.github.gh("issue", "create", "--repo", self.github.repository, "--title", title, + "--body-file", "-", input=str(item.body.get("body") or "")) + number = out.strip().rsplit("/", 1)[-1] + return {"pr": int(number) if number.isdigit() else None} + def _do_close(self, item: OutboxItem) -> dict: number = str(item.body["pr"]) if self.github.pr(int(number)).get("state") == "OPEN": diff --git a/src/infermatrix_copilot/kb_service/report.py b/src/infermatrix_copilot/kb_service/report.py new file mode 100644 index 00000000..20a6fde5 --- /dev/null +++ b/src/infermatrix_copilot/kb_service/report.py @@ -0,0 +1,110 @@ +"""The release-sweep summary: findings, fixes, people's queue and cost. + +Every settled sweep writes ``/reports/sweep--.md``. +For a repository in auto_merge with a public upstream, the same summary is +also published as one issue in the knowledge repository through the outbox +(``open_issue``); the publisher creates it once, keyed by its title. Shadow +and private repositories publish nothing. +""" + +from __future__ import annotations + +import json +from collections import Counter + +MARKER = "" +PENDING = "pending_sweep_reports" + + +def queue_report(ledger, repo: str, report: dict) -> None: + pending = json.loads(ledger.get_cursor(repo, PENDING) or "[]") + if not any(item.get("run_id") == report.get("run_id") for item in pending): + pending.append(report) + ledger.set_cursor(repo, PENDING, json.dumps(pending, sort_keys=True, default=str)) + + +def flush_reports(rt, lifecycle) -> list[str]: + """Publish every queued report; a failure keeps it (and the rest) queued.""" + pending = json.loads(rt.ledger.get_cursor(lifecycle.repo, PENDING) or "[]") + done: list[str] = [] + while pending: + publish_summary(rt, lifecycle, pending[0]) + done.append(str(pending[0].get("run_id"))) + pending = pending[1:] + rt.ledger.set_cursor(lifecycle.repo, PENDING, json.dumps(pending, sort_keys=True, default=str)) + return done + + +def report_id(rt, report: dict) -> str: + """Stable for one sweep (across its resumed attempts), distinct across + sweeps: a release sweep is its tag; a periodic fallback sweep re-checks the + same tag, so it also carries its start date.""" + import datetime as dt + + sweep = report["sweep"] + tag = sweep.get("tag") or "untagged" + if sweep.get("reason") == "release": + return tag + started = float(report.get("started_at") or rt.clock()) + return f"{tag}-{sweep.get('reason') or 'sweep'}-" \ + f"{dt.datetime.fromtimestamp(started, dt.timezone.utc).strftime('%Y%m%d')}" + + +def summary(rt, lifecycle, report: dict) -> tuple[str, str]: + sweep = report["sweep"] + tag = report_id(rt, report) + t1 = report.get("t1") or {} + statuses: Counter = Counter() + ops: Counter = Counter() + for changeset_id in report.get("changesets") or []: + changeset = rt.ledger.changeset(changeset_id) + statuses[changeset["status"]] += 1 + for op in changeset["detail"].get("operations") or []: + ops[op["kind"]] += 1 + tokens = {"input": 0, "output": 0} + seconds = 0.0 + calls = 0 + if getattr(rt, "traces", None) is not None and report.get("run_id"): + run_id = report["run_id"] # this sweep instance only (a fallback reuses the tag) + for record in rt.traces.query(kind="model_call", run_id=run_id, limit=100000): + calls += 1 + tokens["input"] += int((record.get("usage") or {}).get("input_tokens") or 0) + tokens["output"] += int((record.get("usage") or {}).get("output_tokens") or 0) + seconds += float(record.get("seconds") or 0) + title = f"Knowledge sweep report: {lifecycle.repo} {tag}" + lines = [ + MARKER, + f"Release sweep of `{lifecycle.repo}` for **{tag}** " + f"(`{str(sweep.get('from_sha') or '')[:12]}..{str(sweep.get('to_sha') or '')[:12]}`, " + f"trigger: {sweep.get('reason', '')}).", + "", + "| | |", "|---|---|", + f"| structural issues (T1) | {len(t1.get('issues') or [])} |", + f"| duplicate rule IDs | {len(t1.get('duplicates') or {})} |", + f"| pages over capacity | {len(t1.get('over_capacity') or {})} |", + f"| pages missing from an index (fixed) | {len(t1.get('unlisted') or [])} |", + f"| pages kept as they are | {len(report.get('kept_pages', report.get('skipped_pages')) or [])} |", + f"| pages that failed an evaluation | {len(report.get('failed_attempts') or set(report.get('failed_pages') or []))} " + f"({sum((report.get('failed_attempts') or {}).values())} failed attempts) |", + f"| change sets | {sum(statuses.values())} " + 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"| model calls | {calls} ({tokens['input']} in / {tokens['output']} out tokens, {seconds:.0f}s) |", + ] + if report.get("breaker"): + lines += ["", f"**Circuit breaker:** {report['breaker']}; every change set went to people."] + lines += ["", "Generated by the Copilot knowledge service."] + return title, "\n".join(lines) + "\n" + + +def publish_summary(rt, lifecycle, report: dict) -> str: + """Write the summary locally; open the issue only where publication is allowed.""" + title, body = summary(rt, lifecycle, report) + path = rt.state_dir / "reports" / f"sweep-{lifecycle.repo}-{report_id(rt, report)}.md" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(f"# {title}\n\n{body}", encoding="utf-8") + if rt.outbox is None or not lifecycle.auto_merge or not lifecycle.publishes: + return "recorded" + rt.outbox.issue(lifecycle.repo, "open_issue", {"title": title, "body": body, "labels": ["kb:sweep-report"]}) + return "issued" diff --git a/src/infermatrix_copilot/kb_service/scheduler.py b/src/infermatrix_copilot/kb_service/scheduler.py index fb3425ce..3b49f8f6 100644 --- a/src/infermatrix_copilot/kb_service/scheduler.py +++ b/src/infermatrix_copilot/kb_service/scheduler.py @@ -27,6 +27,7 @@ from . import merge from .activate import ActivationError, activate, activation_lock from .audit import audit_main +from .report import flush_reports from .external import poll_external from .runtime import collect_events, publish, run_intake from .sweep import UpstreamRepo, detect_release, run_sweep @@ -130,8 +131,11 @@ def _repo_tick(self, lifecycle) -> None: report = run_sweep(rt, lifecycle, rt.lease_owner, sweep, upstream) for changeset_id in report["changesets"]: publish(rt, lifecycle, changeset_id) + self._record(lifecycle.repo, "sweep", tag=sweep["tag"], reason=sweep["reason"], changesets=report["changesets"], breaker=report["breaker"]) + for run_id in flush_reports(rt, lifecycle): # queued by settled sweeps; retried until done + self._record(lifecycle.repo, "sweep_report", run_id=run_id) def _overturn_breaker(self, lifecycle) -> None: rt = self.rt diff --git a/src/infermatrix_copilot/kb_service/sweep.py b/src/infermatrix_copilot/kb_service/sweep.py index ed48c05b..91b97f7c 100644 --- a/src/infermatrix_copilot/kb_service/sweep.py +++ b/src/infermatrix_copilot/kb_service/sweep.py @@ -272,7 +272,7 @@ def _progress(rt, lifecycle, sweep: dict) -> dict: key = [sweep["tag"], sweep["from_sha"], sweep["to_sha"]] if not progress or progress.get("key") != key: progress = {"key": key, "done": [], "attempts": {}, "t1_done": False, "purge_done": False, - "held": []} + "held": [], "kept": [], "started_at": rt.clock()} return progress @@ -281,9 +281,21 @@ def run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo, """Traced: every model call and decision of the sweep carries its release.""" from ..trace_store import trace_context + # one identity per sweep instance, persisted before any work so every + # resumed attempt traces under it (a fallback sweep reuses the tag) + progress = _progress(rt, lifecycle, sweep) + if rt.ledger.get_cursor(lifecycle.repo, "sweep_progress") in (None, ""): + rt.ledger.set_cursor(lifecycle.repo, "sweep_progress", json.dumps(progress, sort_keys=True)) + 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=f"sweep-{lifecycle.repo}-{sweep.get('tag') or ''}"): - return _run_sweep(rt, lifecycle, owner, sweep, upstream, audit_hints) + run_id=run_id): + report = _run_sweep(rt, lifecycle, owner, sweep, upstream, audit_hints) + report["run_id"] = run_id + return report + + +def sweep_run_id(repo: str, sweep: dict, progress: dict) -> str: + return f"sweep-{repo}-{sweep.get('tag') or ''}-{int(float(progress.get('started_at') or 0))}" def draft_key_for_page(repo: str, sweep: dict, page: str) -> str: @@ -348,6 +360,7 @@ def _run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo, continue if outcome is None: report["skipped_pages"].append(page) + progress.setdefault("kept", []).append(page) progress["done"].append(page) continue operations, result = outcome @@ -382,7 +395,16 @@ def _run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo, report["changesets"] = list(progress.get("held", [])) report["breaker"] = breaker report["complete"] = settled + # the whole sweep, across every resumed attempt (not just this call) + report["kept_pages"] = list(progress.get("kept", [])) + report["failed_attempts"] = dict(progress.get("attempts", {})) + report["started_at"] = progress.get("started_at") if settled: + # the report is queued durably BEFORE the progress that produced it is + # cleared; the scheduler flushes it (and retries until it succeeds) + from .report import queue_report + + queue_report(rt.ledger, lifecycle.repo, {**report, "run_id": sweep_run_id(lifecycle.repo, sweep, progress)}) rt.ledger.set_cursor(lifecycle.repo, "release", sweep["tag"]) rt.ledger.set_cursor(lifecycle.repo, "sweep_baseline", sweep["to_sha"]) rt.ledger.set_cursor(lifecycle.repo, "last_sweep_at", str(rt.clock())) diff --git a/test/test_kb_report.py b/test/test_kb_report.py new file mode 100644 index 00000000..2da0ff53 --- /dev/null +++ b/test/test_kb_report.py @@ -0,0 +1,177 @@ +"""The release-sweep summary: always recorded, published only where allowed.""" + +from __future__ import annotations + +import json +import subprocess +from dataclasses import replace + +from infermatrix_copilot.kb_service.report import MARKER, publish_summary, summary +from test_kb_flow import _flow_runtime, _items + + +def _report(rt, changeset_ids): + return {"sweep": {"tag": "v0.30.0", "from_sha": "a" * 40, "to_sha": "b" * 40, "reason": "release"}, + "t1": {"issues": [{"code": "x"}], "duplicates": {}, "over_capacity": {}, "unlisted": ["p.md"]}, + "changesets": changeset_ids, "skipped_pages": ["k.md", "l.md"], "failed_pages": ["f.md", "f.md"], + "breaker": "", "complete": True} + + +def _staged(rt, status, ops): + changeset_id = rt.ledger.new_changeset_id("demo", "sweep") + rt.ledger.stage_intake(rt.ledger.acquire_lease("t"), "demo", changeset_id, kind="sweep", + detail={"operations": ops}, status=status, verdicts=[], human_reason="", + drafted_events=[]) + return changeset_id + + +def test_the_summary_counts_findings_fixes_people_and_cost(tmp_path): + from infermatrix_copilot.trace_store import TraceStore, trace_context + + rt, lifecycle = _flow_runtime(tmp_path) + rt.traces = TraceStore(tmp_path / "traces", environ={}) + with trace_context(run_id="sweep-demo-v0.30.0-1790000000", playbook="kb-sweep"): + rt.traces.append("model_call", inputs={"prompt": "p"}, outputs={"reply": "r"}, + model={"role": "generator"}, usage={"input_tokens": 100, "output_tokens": 7}, seconds=3) + ids = [_staged(rt, "gated", [{"kind": "retire", "page": "x", "rule_id": "A"}]), + _staged(rt, "human", [{"kind": "edit_same_meaning", "page": "x", "rule_id": "B"}])] + title, body = summary(rt, lifecycle, {**_report(rt, ids), "run_id": "sweep-demo-v0.30.0-1790000000"}) + assert title == "Knowledge sweep report: demo v0.30.0" and body.startswith(MARKER) + for expected in ("| structural issues (T1) | 1 |", "| pages missing from an index (fixed) | 1 |", + "| pages kept as they are | 2 |", "| pages that failed an evaluation | 1 (0 failed attempts) |", + "| change sets | 2 (gated: 1, human: 1) |", "edit_same_meaning: 1, retire: 1", + "| model calls | 1 (100 in / 7 out tokens, 3s) |"): + assert expected in body, expected + + +def test_auto_merge_publishes_the_summary_as_an_issue(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + assert publish_summary(rt, lifecycle, _report(rt, [])) == "issued" + (item,) = _items(tmp_path, "open_issue") + assert item["body"]["title"] == "Knowledge sweep report: demo v0.30.0" + assert (tmp_path / "state" / "reports" / "sweep-demo-v0.30.0.md").exists() + + +def test_shadow_and_private_repositories_only_record_it(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + for variant in (replace(lifecycle, mode="shadow"), replace(lifecycle, upstream_visibility="private")): + assert publish_summary(rt, variant, _report(rt, [])) == "recorded" + assert _items(tmp_path, "open_issue") == [] + assert (tmp_path / "state" / "reports" / "sweep-demo-v0.30.0.md").exists() + + +def test_the_publisher_opens_one_issue_per_report(tmp_path): + from test_kb_publisher import FakeGh, _setup + + rt, lifecycle, _changeset_id, pub, gh = _setup(tmp_path) + issues: list[dict] = [] + + def run(argv, **kw): + args = argv[1:] + if args[:2] == ["issue", "list"]: + return subprocess.CompletedProcess(argv, 0, json.dumps(issues).encode(), b"") + if args[:2] == ["issue", "create"]: + if "--label" in args: + return subprocess.CompletedProcess(argv, 1, b"", b"label not found") + issues.append({"number": 7, "title": args[args.index("--title") + 1]}) + return subprocess.CompletedProcess(argv, 0, b"https://github.com/o/r/issues/7\n", b"") + return FakeGh.__call__(gh, argv, **kw) + + pub.github.run = run + publish_summary(rt, lifecycle, _report(rt, [])) + pub.run_once() + publish_summary(rt, lifecycle, _report(rt, [])) # the same report again + pub.run_once() + assert len(issues) == 1 # reused by title + acks = [json.loads(p.read_text())["payload"] for p in (tmp_path / "state" / "inbox" / "acks").glob("*.json")] + assert all(a["ok"] and a["pr"] == 7 for a in acks if a["kind"] == "open_issue") + + +def test_fallback_sweeps_of_the_same_tag_get_their_own_report(tmp_path): + from infermatrix_copilot.kb_service.report import report_id + + rt, lifecycle = _flow_runtime(tmp_path) + first = {**_report(rt, []), "sweep": {"tag": "v0.30.0", "reason": "fallback"}, "started_at": 1790000000.0} + later = {**first, "started_at": 1790000000.0 + 31 * 86400} + assert report_id(rt, first) != report_id(rt, later) + assert report_id(rt, first) == report_id(rt, dict(first)) # stable across retries + assert report_id(rt, _report(rt, [])) == "v0.30.0" + publish_summary(rt, lifecycle, first) + publish_summary(rt, lifecycle, later) + titles = {item["body"]["title"] for item in _items(tmp_path, "open_issue")} + assert len(titles) == 2 + + +def test_a_resumed_sweep_reports_every_attempt(tmp_path): + from infermatrix_copilot.kb_service.sweep import _progress + + rt, lifecycle = _flow_runtime(tmp_path) + sweep = {"tag": "v1", "from_sha": "a" * 40, "to_sha": "b" * 40, "reason": "release"} + progress = _progress(rt, lifecycle, sweep) + progress["kept"] = ["repos/demo/core/a.md"] # kept in an earlier attempt + progress["attempts"] = {"repos/demo/core/b.md": 1} # failed once, settled later + report = {**_report(rt, []), "sweep": sweep, "kept_pages": progress["kept"], + "failed_attempts": progress["attempts"], "skipped_pages": [], "failed_pages": []} + _title, body = summary(rt, lifecycle, report) + assert "| pages kept as they are | 1 |" in body + assert "| pages that failed an evaluation | 1 (1 failed attempts) |" in body + + +def test_a_fallback_sweep_never_reports_the_release_sweeps_model_calls(tmp_path): + from infermatrix_copilot.trace_store import TraceStore, trace_context + + rt, lifecycle = _flow_runtime(tmp_path) + rt.traces = TraceStore(tmp_path / "traces", environ={}) + with trace_context(run_id="sweep-demo-v0.30.0-1", playbook="kb-sweep"): + rt.traces.append("model_call", inputs={"prompt": "p"}, outputs={"reply": "r"}, + model={"role": "generator"}, usage={"input_tokens": 9, "output_tokens": 9}) + fallback = {**_report(rt, []), "sweep": {"tag": "v0.30.0", "reason": "fallback"}, "run_id": "sweep-demo-v0.30.0-2"} + assert "| model calls | 0 (0 in / 0 out tokens, 0s) |" in summary(rt, lifecycle, fallback)[1] + + +def test_a_queued_report_survives_a_failure_and_is_flushed_later(tmp_path, monkeypatch): + from infermatrix_copilot.kb_service import report as report_module + + rt, lifecycle = _flow_runtime(tmp_path) + report_module.queue_report(rt.ledger, "demo", {**_report(rt, []), "run_id": "sweep-demo-v0.30.0-5"}) + report_module.queue_report(rt.ledger, "demo", {**_report(rt, []), "run_id": "sweep-demo-v0.30.0-5"}) # once + + def broken(*args, **kwargs): + raise OSError("disk full") + + monkeypatch.setattr(report_module, "publish_summary", broken) + import pytest + with pytest.raises(OSError): + report_module.flush_reports(rt, lifecycle) + monkeypatch.undo() + assert report_module.flush_reports(rt, lifecycle) == ["sweep-demo-v0.30.0-5"] + assert report_module.flush_reports(rt, lifecycle) == [] + assert len(_items(tmp_path, "open_issue")) == 1 + + +def test_a_report_issue_survives_a_transient_github_failure(tmp_path): + from test_kb_publisher import FakeGh, _setup + + rt, lifecycle, _changeset_id, pub, gh = _setup(tmp_path) + issues: list[dict] = [] + state = {"down": True} + + def run(argv, **kw): + args = argv[1:] + if args[:2] == ["issue", "list"]: + if state["down"]: + return subprocess.CompletedProcess(argv, 1, b"", b"HTTP 502") + return subprocess.CompletedProcess(argv, 0, json.dumps(issues).encode(), b"") + if args[:2] == ["issue", "create"]: + issues.append({"number": 8, "title": args[args.index("--title") + 1]}) + return subprocess.CompletedProcess(argv, 0, b"https://github.com/o/r/issues/8\n", b"") + return FakeGh.__call__(gh, argv, **kw) + + pub.github.run = run + publish_summary(rt, lifecycle, _report(rt, [])) + pub.run_once() + assert issues == [] and not [p for p in (tmp_path / "state" / "inbox" / "acks").glob("*.json") + if json.loads(p.read_text())["payload"]["kind"] == "open_issue"] + state["down"] = False + pub.run_once() # retried, not lost + assert len(issues) == 1