Publish a summary issue after each release sweep

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 <n> --undo` 命令。`kb status` 报告每个仓库 `pause_unconfirmed` 数量，熔断/回滚在其归零
 前不算完成；`kb resume` 清除这些记录。
+
+## 2026-09-29 巡检汇总 issue
+每次巡检完成（`complete`）后，`report.publish_summary` 把汇总（T1 发现、修复的索引、保留/失败的页面、变更集及其状态、
+各类操作数、待人工处理数、模型调用次数/token/耗时）写入 `<state_dir>/reports/sweep-<repo>-<tag>.md`；仅当仓库为
+`auto_merge` 且上游公开时，另通过新的 outbox 项 `open_issue`（7 天有效）在知识仓库发布一个 issue。发布器按标题复用已有
+issue（同一报告只开一个），标签不存在时不带标签重试，绝不丢报告。

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

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

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 ``<state_dir>/reports/sweep-<repo>-<tag>.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 = "<!-- kb-sweep-report:v1 -->"
+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"

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

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