From 4b09a1e953240bddd84658ef248ee8332eace8d2 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 29 Sep 2026 00:45:51 +0800 Subject: [PATCH] Publish a summary issue after each release sweep The design asks for a summary issue after each release sweep: findings, fixes, people's queue, cost. Sweeps ran, but nothing reported them. - Every sweep instance gets a run id (repository, tag, start time), persisted before any work so each resumed attempt traces under it. A settled sweep queues its report durably BEFORE its progress is cleared; the scheduler flushes queued reports every tick and keeps any that fail, so a failure after the sweep settles cannot lose the report. Model-call cost is summed for that run id only, so a fallback sweep never reports the release sweep's calls. - report.publish_summary writes /reports/sweep--.md: T1 findings, index fixes, pages kept or failed, change sets by status, operations by kind, the repository's human queue, and model calls with tokens and time (from the trace store). Kept and failed pages are aggregated across every resumed attempt through the persisted sweep progress, not just the final call. - Each report has a durable id, stable across the sweep's resumed attempts: a release sweep is its tag, and a periodic fallback sweep (which re-checks the same tag) adds its start date. Fallback reports therefore never overwrite, or get deduplicated into, an earlier one. - A repository in auto_merge with a public upstream also gets one issue in the knowledge repository, through a new outbox item kind open_issue (valid 7 days). Shadow and private repositories only record the file. - The publisher's open_issue reuses an existing issue with the same title (one issue per report), and retries without labels if a label is missing, so a report is never lost. open_issue is retried like pause and close: a transient GitHub failure is never acked as failed, so the item stays in the outbox until it succeeds (or expires after 7 days). Co-Authored-By: Claude Opus 5.5 --- doc/architecture/SPEC/kb_service.md | 6 + src/infermatrix_copilot/kb_service/outbox.py | 3 +- .../kb_service/publisher.py | 22 ++- src/infermatrix_copilot/kb_service/report.py | 110 +++++++++++ .../kb_service/scheduler.py | 4 + src/infermatrix_copilot/kb_service/sweep.py | 28 ++- test/test_kb_report.py | 177 ++++++++++++++++++ 7 files changed, 345 insertions(+), 5 deletions(-) create mode 100644 src/infermatrix_copilot/kb_service/report.py create mode 100644 test/test_kb_report.py 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