Skip to content

Commit e3bc60d

Browse files
authored
Merge pull request #219 from JiusiServe/kb/sweep-report-issue
Publish a summary issue after each release sweep
2 parents bf73f93 + 4b09a1e commit e3bc60d

7 files changed

Lines changed: 345 additions & 5 deletions

File tree

‎doc/architecture/SPEC/kb_service.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -146,3 +146,9 @@ stdlib + PyYAML + `cryptography`(`kb` extra)+ `.adapters` + `.knowledge_serv
146146
为 draft 时才记为已确认(draft 不能在合并队列中,也不能再次入队);否则重发 `pause`。请求 5 分钟后仍未回执或未确认 →
147147
转人工一次,附出队与 `gh pr ready <n> --undo` 命令。`kb status` 报告每个仓库 `pause_unconfirmed` 数量,熔断/回滚在其归零
148148
前不算完成;`kb resume` 清除这些记录。
149+
150+
## 2026-09-29 巡检汇总 issue
151+
每次巡检完成(`complete`)后,`report.publish_summary` 把汇总(T1 发现、修复的索引、保留/失败的页面、变更集及其状态、
152+
各类操作数、待人工处理数、模型调用次数/token/耗时)写入 `<state_dir>/reports/sweep-<repo>-<tag>.md`;仅当仓库为
153+
`auto_merge` 且上游公开时,另通过新的 outbox 项 `open_issue`(7 天有效)在知识仓库发布一个 issue。发布器按标题复用已有
154+
issue(同一报告只开一个),标签不存在时不带标签重试,绝不丢报告。

‎src/infermatrix_copilot/kb_service/outbox.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -37,13 +37,14 @@
3737

3838
ITEM_KINDS = (
3939
"open_pr", "open_companion_pr", "post_verdict", "enqueue",
40-
"update_branch", "pause", "close",
40+
"update_branch", "pause", "close", "open_issue",
4141
)
4242
ALWAYS_EXECUTABLE = frozenset({"pause", "close"})
4343
ITEM_TTL = {
4444
"open_pr": 24 * 3600, "open_companion_pr": 24 * 3600,
4545
"post_verdict": 30 * 60, "enqueue": 30 * 60,
4646
"update_branch": 24 * 3600, "pause": 24 * 3600, "close": 24 * 3600,
47+
"open_issue": 7 * 24 * 3600,
4748
}
4849
CONTROL_MAX_AGE = 10 * 60
4950
HOLDS_MAX_AGE = 10 * 60

‎src/infermatrix_copilot/kb_service/publisher.py‎

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,9 @@
4848
BRANCH = re.compile(r"kb/[A-Za-z0-9._-]+/[A-Za-z0-9._-]+")
4949
HOLD_LABEL = "kb:hold"
5050
PUSHING = frozenset({"open_pr", "open_companion_pr"})
51-
RETRIED = frozenset({"pause", "close"}) # stops: retried until done, never acked as failed
51+
# retried until done, never acked as failed: stops (pause, close) and reports
52+
# (open_issue: a transient GitHub error must not lose a sweep report)
53+
RETRIED = frozenset({"pause", "close", "open_issue"})
5254

5355

5456
class PublishError(RuntimeError):
@@ -366,6 +368,24 @@ def _do_update_branch(self, item: OutboxItem) -> dict:
366368
"-f", f"expected_head_sha={item.body['head_sha']}")
367369
return {"pr": int(item.body["pr"])}
368370

371+
def _do_open_issue(self, item: OutboxItem) -> dict:
372+
"""One issue per title (a sweep report): an existing one is reused."""
373+
title = str(item.body["title"])
374+
found = json.loads(self.github.gh("issue", "list", "--repo", self.github.repository, "--state", "all",
375+
"--search", f'in:title "{title}"', "--json", "number,title"))
376+
for issue in found:
377+
if issue.get("title") == title:
378+
return {"pr": int(issue["number"])}
379+
args = ["issue", "create", "--repo", self.github.repository, "--title", title, "--body-file", "-"]
380+
for label in item.body.get("labels") or []:
381+
args += ["--label", str(label)]
382+
out = self.github.gh(*args, input=str(item.body.get("body") or ""), ok_fail=True)
383+
if not out.strip(): # a missing label must not lose the report
384+
out = self.github.gh("issue", "create", "--repo", self.github.repository, "--title", title,
385+
"--body-file", "-", input=str(item.body.get("body") or ""))
386+
number = out.strip().rsplit("/", 1)[-1]
387+
return {"pr": int(number) if number.isdigit() else None}
388+
369389
def _do_close(self, item: OutboxItem) -> dict:
370390
number = str(item.body["pr"])
371391
if self.github.pr(int(number)).get("state") == "OPEN":
Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,110 @@
1+
"""The release-sweep summary: findings, fixes, people's queue and cost.
2+
3+
Every settled sweep writes ``<state_dir>/reports/sweep-<repo>-<tag>.md``.
4+
For a repository in auto_merge with a public upstream, the same summary is
5+
also published as one issue in the knowledge repository through the outbox
6+
(``open_issue``); the publisher creates it once, keyed by its title. Shadow
7+
and private repositories publish nothing.
8+
"""
9+
10+
from __future__ import annotations
11+
12+
import json
13+
from collections import Counter
14+
15+
MARKER = "<!-- kb-sweep-report:v1 -->"
16+
PENDING = "pending_sweep_reports"
17+
18+
19+
def queue_report(ledger, repo: str, report: dict) -> None:
20+
pending = json.loads(ledger.get_cursor(repo, PENDING) or "[]")
21+
if not any(item.get("run_id") == report.get("run_id") for item in pending):
22+
pending.append(report)
23+
ledger.set_cursor(repo, PENDING, json.dumps(pending, sort_keys=True, default=str))
24+
25+
26+
def flush_reports(rt, lifecycle) -> list[str]:
27+
"""Publish every queued report; a failure keeps it (and the rest) queued."""
28+
pending = json.loads(rt.ledger.get_cursor(lifecycle.repo, PENDING) or "[]")
29+
done: list[str] = []
30+
while pending:
31+
publish_summary(rt, lifecycle, pending[0])
32+
done.append(str(pending[0].get("run_id")))
33+
pending = pending[1:]
34+
rt.ledger.set_cursor(lifecycle.repo, PENDING, json.dumps(pending, sort_keys=True, default=str))
35+
return done
36+
37+
38+
def report_id(rt, report: dict) -> str:
39+
"""Stable for one sweep (across its resumed attempts), distinct across
40+
sweeps: a release sweep is its tag; a periodic fallback sweep re-checks the
41+
same tag, so it also carries its start date."""
42+
import datetime as dt
43+
44+
sweep = report["sweep"]
45+
tag = sweep.get("tag") or "untagged"
46+
if sweep.get("reason") == "release":
47+
return tag
48+
started = float(report.get("started_at") or rt.clock())
49+
return f"{tag}-{sweep.get('reason') or 'sweep'}-" \
50+
f"{dt.datetime.fromtimestamp(started, dt.timezone.utc).strftime('%Y%m%d')}"
51+
52+
53+
def summary(rt, lifecycle, report: dict) -> tuple[str, str]:
54+
sweep = report["sweep"]
55+
tag = report_id(rt, report)
56+
t1 = report.get("t1") or {}
57+
statuses: Counter = Counter()
58+
ops: Counter = Counter()
59+
for changeset_id in report.get("changesets") or []:
60+
changeset = rt.ledger.changeset(changeset_id)
61+
statuses[changeset["status"]] += 1
62+
for op in changeset["detail"].get("operations") or []:
63+
ops[op["kind"]] += 1
64+
tokens = {"input": 0, "output": 0}
65+
seconds = 0.0
66+
calls = 0
67+
if getattr(rt, "traces", None) is not None and report.get("run_id"):
68+
run_id = report["run_id"] # this sweep instance only (a fallback reuses the tag)
69+
for record in rt.traces.query(kind="model_call", run_id=run_id, limit=100000):
70+
calls += 1
71+
tokens["input"] += int((record.get("usage") or {}).get("input_tokens") or 0)
72+
tokens["output"] += int((record.get("usage") or {}).get("output_tokens") or 0)
73+
seconds += float(record.get("seconds") or 0)
74+
title = f"Knowledge sweep report: {lifecycle.repo} {tag}"
75+
lines = [
76+
MARKER,
77+
f"Release sweep of `{lifecycle.repo}` for **{tag}** "
78+
f"(`{str(sweep.get('from_sha') or '')[:12]}..{str(sweep.get('to_sha') or '')[:12]}`, "
79+
f"trigger: {sweep.get('reason', '')}).",
80+
"",
81+
"| | |", "|---|---|",
82+
f"| structural issues (T1) | {len(t1.get('issues') or [])} |",
83+
f"| duplicate rule IDs | {len(t1.get('duplicates') or {})} |",
84+
f"| pages over capacity | {len(t1.get('over_capacity') or {})} |",
85+
f"| pages missing from an index (fixed) | {len(t1.get('unlisted') or [])} |",
86+
f"| pages kept as they are | {len(report.get('kept_pages', report.get('skipped_pages')) or [])} |",
87+
f"| pages that failed an evaluation | {len(report.get('failed_attempts') or set(report.get('failed_pages') or []))} "
88+
f"({sum((report.get('failed_attempts') or {}).values())} failed attempts) |",
89+
f"| change sets | {sum(statuses.values())} "
90+
f"({', '.join(f'{k}: {v}' for k, v in sorted(statuses.items())) or 'none'}) |",
91+
f"| operations | {', '.join(f'{k}: {v}' for k, v in sorted(ops.items())) or 'none'} |",
92+
f"| waiting for people | {len(rt.ledger.human_queue(lifecycle.repo))} |",
93+
f"| model calls | {calls} ({tokens['input']} in / {tokens['output']} out tokens, {seconds:.0f}s) |",
94+
]
95+
if report.get("breaker"):
96+
lines += ["", f"**Circuit breaker:** {report['breaker']}; every change set went to people."]
97+
lines += ["", "Generated by the Copilot knowledge service."]
98+
return title, "\n".join(lines) + "\n"
99+
100+
101+
def publish_summary(rt, lifecycle, report: dict) -> str:
102+
"""Write the summary locally; open the issue only where publication is allowed."""
103+
title, body = summary(rt, lifecycle, report)
104+
path = rt.state_dir / "reports" / f"sweep-{lifecycle.repo}-{report_id(rt, report)}.md"
105+
path.parent.mkdir(parents=True, exist_ok=True)
106+
path.write_text(f"# {title}\n\n{body}", encoding="utf-8")
107+
if rt.outbox is None or not lifecycle.auto_merge or not lifecycle.publishes:
108+
return "recorded"
109+
rt.outbox.issue(lifecycle.repo, "open_issue", {"title": title, "body": body, "labels": ["kb:sweep-report"]})
110+
return "issued"

‎src/infermatrix_copilot/kb_service/scheduler.py‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
from . import merge
2828
from .activate import ActivationError, activate, activation_lock
2929
from .audit import audit_main
30+
from .report import flush_reports
3031
from .external import poll_external
3132
from .runtime import collect_events, publish, run_intake
3233
from .sweep import UpstreamRepo, detect_release, run_sweep
@@ -130,8 +131,11 @@ def _repo_tick(self, lifecycle) -> None:
130131
report = run_sweep(rt, lifecycle, rt.lease_owner, sweep, upstream)
131132
for changeset_id in report["changesets"]:
132133
publish(rt, lifecycle, changeset_id)
134+
133135
self._record(lifecycle.repo, "sweep", tag=sweep["tag"], reason=sweep["reason"],
134136
changesets=report["changesets"], breaker=report["breaker"])
137+
for run_id in flush_reports(rt, lifecycle): # queued by settled sweeps; retried until done
138+
self._record(lifecycle.repo, "sweep_report", run_id=run_id)
135139

136140
def _overturn_breaker(self, lifecycle) -> None:
137141
rt = self.rt

‎src/infermatrix_copilot/kb_service/sweep.py‎

Lines changed: 25 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -272,7 +272,7 @@ def _progress(rt, lifecycle, sweep: dict) -> dict:
272272
key = [sweep["tag"], sweep["from_sha"], sweep["to_sha"]]
273273
if not progress or progress.get("key") != key:
274274
progress = {"key": key, "done": [], "attempts": {}, "t1_done": False, "purge_done": False,
275-
"held": []}
275+
"held": [], "kept": [], "started_at": rt.clock()}
276276
return progress
277277

278278

@@ -281,9 +281,21 @@ def run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo,
281281
"""Traced: every model call and decision of the sweep carries its release."""
282282
from ..trace_store import trace_context
283283

284+
# one identity per sweep instance, persisted before any work so every
285+
# resumed attempt traces under it (a fallback sweep reuses the tag)
286+
progress = _progress(rt, lifecycle, sweep)
287+
if rt.ledger.get_cursor(lifecycle.repo, "sweep_progress") in (None, ""):
288+
rt.ledger.set_cursor(lifecycle.repo, "sweep_progress", json.dumps(progress, sort_keys=True))
289+
run_id = sweep_run_id(lifecycle.repo, sweep, progress)
284290
with trace_context(playbook="kb-sweep", repo=lifecycle.repo, release=str(sweep.get("tag") or ""),
285-
run_id=f"sweep-{lifecycle.repo}-{sweep.get('tag') or ''}"):
286-
return _run_sweep(rt, lifecycle, owner, sweep, upstream, audit_hints)
291+
run_id=run_id):
292+
report = _run_sweep(rt, lifecycle, owner, sweep, upstream, audit_hints)
293+
report["run_id"] = run_id
294+
return report
295+
296+
297+
def sweep_run_id(repo: str, sweep: dict, progress: dict) -> str:
298+
return f"sweep-{repo}-{sweep.get('tag') or ''}-{int(float(progress.get('started_at') or 0))}"
287299

288300

289301
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,
348360
continue
349361
if outcome is None:
350362
report["skipped_pages"].append(page)
363+
progress.setdefault("kept", []).append(page)
351364
progress["done"].append(page)
352365
continue
353366
operations, result = outcome
@@ -382,7 +395,16 @@ def _run_sweep(rt, lifecycle, owner: str, sweep: dict, upstream: UpstreamRepo,
382395
report["changesets"] = list(progress.get("held", []))
383396
report["breaker"] = breaker
384397
report["complete"] = settled
398+
# the whole sweep, across every resumed attempt (not just this call)
399+
report["kept_pages"] = list(progress.get("kept", []))
400+
report["failed_attempts"] = dict(progress.get("attempts", {}))
401+
report["started_at"] = progress.get("started_at")
385402
if settled:
403+
# the report is queued durably BEFORE the progress that produced it is
404+
# cleared; the scheduler flushes it (and retries until it succeeds)
405+
from .report import queue_report
406+
407+
queue_report(rt.ledger, lifecycle.repo, {**report, "run_id": sweep_run_id(lifecycle.repo, sweep, progress)})
386408
rt.ledger.set_cursor(lifecycle.repo, "release", sweep["tag"])
387409
rt.ledger.set_cursor(lifecycle.repo, "sweep_baseline", sweep["to_sha"])
388410
rt.ledger.set_cursor(lifecycle.repo, "last_sweep_at", str(rt.clock()))

0 commit comments

Comments
 (0)