diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index 1c66b065..841093cb 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -187,3 +187,11 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝 `verify_snapshot` 在其他检查之前核对快照声明的知识格式：未声明或不在本 Copilot 支持的 `SUPPORTED_FORMATS` 中 → `ActivationError`，`active` 保持不变（先升级 Copilot 再激活新格式的知识）。快照随顶层 Markdown 一起携带 `_format.yaml`。 + +## 2026-09-29 Copilot 运行经验（来源②） +`collect_events` 在 `intake.copilot_runs` 开启时，除 `inbox/lessons/` 外还读取 `pr_debug` 的已验证修复记录（v1/v2， +`engine/steps/pr/debug.py` 产出）：同机运行投放到 `KB_BUGFIX_DIR`（缺省 `/inbox/bugfix`，与运行侧 +`KNOWLEDGE_INTAKE_DIR` 配为同一目录），异机运行发到邮箱 issue `KB_BUGFIX_MAILBOX`（`owner/repo#N`）。仓库按完整名 +（不区分大小写）或别名精确匹配；歧义身份改名 `.quarantined` 隔离、从不改写；属于其他仓库的投放文件留待其处理。 +邮箱是公开 issue，只接受 `KB_BUGFIX_AUTHORS` 列出的作者（未设置则不读邮箱），每个仓库各自维护评论游标。 +事件 ID 与记录的 `event_id` 一致，重复投递幂等。 diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md index 63da9853..92904920 100644 --- a/doc/features/knowledge-service.md +++ b/doc/features/knowledge-service.md @@ -50,6 +50,9 @@ infermatrix-copilot kb calibrate --repo vllm-omni # 评审模 模型：`KB_GENERATOR`（默认 `claude-code:claude-opus-5-5`）、`KB_JUDGE`（默认 `codex:gpt-6-sol:medium`）。每次模型调用写入 `$KB_STATE_DIR/traces/model_calls.jsonl`。 +运行经验（来源②）：`pr_debug` 的已验证修复记录经 `KB_BUGFIX_DIR`（同机，与运行侧 `KNOWLEDGE_INTAKE_DIR` +配为同一目录）与 `KB_BUGFIX_MAILBOX`（异机，`owner/repo#N`，只接受 `KB_BUGFIX_AUTHORS` 列出的作者）进入 intake。 + ## 调度、合并、激活与巡检 ```bash diff --git a/src/infermatrix_copilot/kb_service/runtime.py b/src/infermatrix_copilot/kb_service/runtime.py index d1017898..1cac4d6f 100644 --- a/src/infermatrix_copilot/kb_service/runtime.py +++ b/src/infermatrix_copilot/kb_service/runtime.py @@ -22,7 +22,10 @@ from .intake import draft_changes, merge_drafts, operations_json from .ledger import Ledger from .models import ModelGateway, ModelRole, ModelUnavailable, roles_from_env -from .sources import GitHubReader, KnowledgeRepo, SourceError, load_lessons +from .sources import ( + GitHubReader, KnowledgeRepo, SourceError, bugfix_lesson, load_bugfix_drops, load_lessons, mailbox_records, + resolve_repository, +) DEFAULT_BACKFILL_DAYS = 1 @@ -150,12 +153,42 @@ def collect_events(rt: KbRuntime, lifecycle: RepoLifecycle) -> int: new += 1 rt.ledger.set_cursor(lifecycle.repo, "merged_since", f"{merged_at}|{number}") if lifecycle.intake.copilot_runs: - for lesson in load_lessons(rt.state_dir / "inbox", lifecycle.repo): + def record(lesson: dict) -> None: + nonlocal new evidence = {"source_reference": f"run {lesson['run_ref']}", "title": "Copilot run lesson", "body": lesson["summary"], "changed_files": [], "diff_excerpt": lesson["diff_excerpt"]} if rt.ledger.record_event(lifecycle.repo, "copilot_run", lesson["event_id"], evidence) is not None: new += 1 + + for lesson in load_lessons(rt.state_dir / "inbox", lifecycle.repo): + record(lesson) + Path(lesson["path"]).rename(Path(lesson["path"]).with_suffix(".consumed")) + # verified pr_debug fixes: dropped by runs on this host ... + drops = os.environ.get("KB_BUGFIX_DIR") or str(rt.state_dir / "inbox" / "bugfix") + for lesson in load_bugfix_drops(drops, rt.registry, lifecycle.repo): + record(lesson) Path(lesson["path"]).rename(Path(lesson["path"]).with_suffix(".consumed")) + # ... and posted to the mailbox issue by runs on other hosts + mailbox = os.environ.get("KB_BUGFIX_MAILBOX", "") + authors = {a.strip().lower() for a in os.environ.get("KB_BUGFIX_AUTHORS", "").split(",") if a.strip()} + if mailbox and authors: + full_name, _, number = mailbox.partition("#") + after = int(rt.ledger.get_cursor(lifecycle.repo, "bugfix_mailbox_after") or 0) + since = rt.ledger.get_cursor(lifecycle.repo, "bugfix_mailbox_since") or "" + comments = rt.github.issue_comments(full_name, int(number), after_id=after, since=since) + for _comment_id, raw in mailbox_records(comments, authors): + lesson = bugfix_lesson(raw) + if lesson is not None and resolve_repository(lesson["repo_key"], rt.registry) == lifecycle.repo: + record(lesson) + if comments: + rt.ledger.set_cursor(lifecycle.repo, "bugfix_mailbox_after", str(int(comments[-1]["id"]))) + if comments[-1].get("created_at"): + # ``since`` is exclusive and second-grained: step back one + # second so an unread comment sharing that second is read + # again; the id cursor drops what was already seen + last = dt.datetime.fromisoformat(str(comments[-1]["created_at"]).replace("Z", "+00:00")) + rt.ledger.set_cursor(lifecycle.repo, "bugfix_mailbox_since", + (last - dt.timedelta(seconds=1)).strftime("%Y-%m-%dT%H:%M:%SZ")) return new diff --git a/src/infermatrix_copilot/kb_service/sources.py b/src/infermatrix_copilot/kb_service/sources.py index 92f14173..44e4d6f4 100644 --- a/src/infermatrix_copilot/kb_service/sources.py +++ b/src/infermatrix_copilot/kb_service/sources.py @@ -268,6 +268,22 @@ def pull_request(self, full_name: str, number: int) -> PullRequest: changed_files=tuple(paths), diff_excerpt="".join(excerpt), ) + def issue_comments(self, full_name: str, number: int, *, after_id: int = 0, since: str = "", + max_pages: int = 20) -> list[dict]: + """Comments on an issue with an id above ``after_id``, oldest first, + updated after ``since`` (just before the last seen one's creation) so a caller + that advances both past what it read is never stuck behind a window of + ``max_pages`` pages.""" + out: list[dict] = [] + params = {"since": since} if since else {} + for page in range(1, max_pages + 1): + batch = self.get(f"/repos/{full_name}/issues/{int(number)}/comments", **params, + per_page=100, page=page) or [] + out.extend(c for c in batch if int(c.get("id") or 0) > after_id) + if len(batch) < 100: + break + return sorted(out, key=lambda c: int(c.get("id") or 0)) + def latest_release(self, full_name: str) -> dict | None: releases = self.get(f"/repos/{full_name}/releases", per_page=5) for release in releases: @@ -285,6 +301,90 @@ def _glob_to_regex(pattern: str) -> str: return "".join(".*" if ch == "*" else re.escape(ch) for ch in pattern) +BUGFIX_MARKERS = ("", + "") + + +def bugfix_lesson(record) -> dict | None: + """A verified pr_debug fix record (drop file or mailbox comment, v1/v2) as a + lesson; None when malformed. ``repo_key`` is its repository identity.""" + if not isinstance(record, dict): + return None + repo, run = str(record.get("repo") or ""), str(record.get("run_id") or "") + groups = record.get("groups") + if not repo or not run or not isinstance(groups, list) or not groups: + return None + lines = [str(record.get("title") or "Copilot debug run")] + for group in groups[:10]: + if not isinstance(group, dict) or not isinstance(group.get("files") or [], list): + return None + lines += [f"- failure: {str(group.get('signature', ''))[:300]}", + f" root cause: {str(group.get('root_cause', ''))[:2000]}", + f" fix: {str(group.get('fix_summary', ''))[:2000]}", + f" verified by: {str(group.get('verification', ''))[:1000]}", + " files: " + ", ".join(str(f)[:300] for f in (group.get("files") or [])[:20])] + event_id = str(record.get("event_id") or f"bugfix_run:{repo.casefold()}:{run}")[:200] + return {"repo_key": repo, "event_id": event_id, "run_ref": run, + "summary": "\n".join(lines)[:MAX_BODY_CHARS], "diff_excerpt": ""} + + +def resolve_repository(key: str, registry) -> str | None: + """The one configured repository ``key`` names (full name, case-insensitive, + or exact alias); None when none does; "?" when it is ambiguous.""" + matches = {name for name, lc in registry.items() + if (lc.full_name and lc.full_name.casefold() == key.casefold()) or name == key} + if len(matches) > 1: + return "?" + return next(iter(matches), None) + + +def load_bugfix_drops(directory: str | Path, registry, repo: str) -> list[dict]: + """Fix records that runs on this host dropped for ``repo``. Records naming + another configured repository stay for its turn; an ambiguous identity is + quarantined (renamed ``.quarantined``), never rewritten.""" + directory = Path(directory) + lessons = [] + if not directory.is_dir(): + return lessons + for path in sorted(directory.glob("*.json")): + try: + lesson = bugfix_lesson(json.loads(path.read_text(encoding="utf-8"))) + except (OSError, ValueError): + continue + if lesson is None: + continue # malformed: left for inspection + owner = resolve_repository(lesson["repo_key"], registry) + if owner == "?": + path.rename(path.with_suffix(".quarantined")) + continue + if owner == repo: + lessons.append({**lesson, "path": str(path)}) + return lessons + + +def mailbox_records(comments, authors: set[str]) -> list[tuple[int, dict]]: + """(comment id, record) of marked fix-record comments by allowed authors. + The mailbox is a public issue: anyone else's comment is ignored.""" + out = [] + for comment in comments: + login = str((comment.get("user") or {}).get("login") or "").lower() + body = str(comment.get("body") or "") + marker = next((m for m in BUGFIX_MARKERS if m in body), None) + if marker is None or login not in authors: + continue + rest = body.split(marker, 1)[1] + start = rest.find("```json") + if start < 0: + continue + rest = rest[start + len("```json"):].lstrip() + try: # decode the object itself: a fence inside a string is not the end + record, _ = json.JSONDecoder().raw_decode(rest) + out.append((int(comment["id"]), record)) + except (KeyError, TypeError, ValueError): + continue + return out + + def load_lessons(inbox: str | Path, repo: str) -> list[dict]: """Copilot run lessons dropped by runs on this host (``inbox/lessons//``). Each file: {event_id, repo, run_ref, summary, diff_excerpt?}. Malformed files diff --git a/test/test_kb_lessons.py b/test/test_kb_lessons.py new file mode 100644 index 00000000..560be322 --- /dev/null +++ b/test/test_kb_lessons.py @@ -0,0 +1,145 @@ +"""Copilot run lessons (source 2): verified pr_debug fixes become knowledge events.""" + +from __future__ import annotations + +import json +from dataclasses import replace +from urllib.parse import parse_qs, urlparse + +from infermatrix_copilot.kb_service.runtime import collect_events +from test_kb_intake_gate import FakeGitHub, ScriptedGateway, _generator_then_judge, _runtime + +MARKER = "" + + +def _record(repo="org/demo", run="run-20260929-000001-aaaaaa"): + """The shape engine/steps/pr/debug.py drops and posts (schema v2).""" + return {"schema_version": 2, "run_id": run, "repo": repo, "pr": 11, "kind": "bugfix_run", + "title": "pr_debug fixes for PR #11", + "groups": [{"signature": "test_queue_full", "jobs": ["unit"], "root_cause": "queue grew without bound", + "fix_summary": "reject when full", "verification": "unit test passes", + "files": ["demo/core/q.py"]}], + "created_at": "2026-09-29T00:00:00+00:00", + "event_id": f"bugfix_run:{repo.casefold()}:{run}", "repository": {"full_name": repo, "alias": "demo"}} + + +class MailboxGitHub(FakeGitHub): + def __init__(self, comments): + super().__init__() + self.comments = comments + + def _answer(self, url): + if "/issues/135/comments" in url: # GitHub: created-time order, updated strictly after ``since`` + query = parse_qs(urlparse(url).query) + since = query.get("since", [""])[0] + page = int(query["page"][0]) + rows = [c for c in self.comments if c["updated_at"] > since] + return rows[(page - 1) * 100:page * 100] + return super()._answer(url) + + +def comment(cid, login, record, second=None): + second = cid if second is None else second + stamp = f"2026-09-29T{second // 3600:02d}:{second // 60 % 60:02d}:{second % 60:02d}Z" + return {"id": cid, "created_at": stamp, "updated_at": stamp, + "user": {"login": login}, "body": f"{MARKER}\n```json\n{json.dumps(record)}\n```\n"} + + +def _rt(tmp_path, monkeypatch, comments=()): + rt, lifecycle = _runtime(tmp_path, ScriptedGateway(_generator_then_judge())) + rt.github = MailboxGitHub(list(comments)) + monkeypatch.setenv("KB_BUGFIX_DIR", str(tmp_path / "drops")) + return rt, lifecycle + + +def _events(rt): + return [e for e in rt.ledger.events("demo", "pending", limit=100) if e["source"] == "copilot_run"] + + +def test_a_dropped_fix_record_becomes_one_event(tmp_path, monkeypatch): + rt, lifecycle = _rt(tmp_path, monkeypatch) + drops = tmp_path / "drops" + drops.mkdir() + (drops / "run-1.json").write_text(json.dumps(_record())) + collect_events(rt, lifecycle) + (event,) = _events(rt) + assert "queue grew without bound" in event["payload"]["body"] and "reject when full" in event["payload"]["body"] + assert (drops / "run-1.consumed").exists() + (drops / "run-1.json").write_text(json.dumps(_record())) # delivered twice: idempotent by event id + collect_events(rt, lifecycle) + assert len(_events(rt)) == 1 + + +def test_records_for_other_or_ambiguous_repositories_are_not_taken(tmp_path, monkeypatch): + rt, lifecycle = _rt(tmp_path, monkeypatch) + rt.registry["org/demo"] = replace(lifecycle, repo="org/demo", full_name="other/x") # alias collides + drops = tmp_path / "drops" + drops.mkdir() + (drops / "mine.json").write_text(json.dumps(_record(repo="org/demo"))) + (drops / "theirs.json").write_text(json.dumps(_record(repo="someone/else"))) + (drops / "broken.json").write_text("{not json") + collect_events(rt, lifecycle) + assert _events(rt) == [] + assert (drops / "mine.quarantined").exists() # ambiguous: never guessed + assert (drops / "theirs.json").exists() and (drops / "broken.json").exists() + + +def test_the_mailbox_takes_only_allowed_authors_and_advances_its_cursor(tmp_path, monkeypatch): + comments = [comment(1, "tzhouam", _record(run="run-a")), comment(2, "mallory", _record(run="run-b")), + comment(3, "tzhouam", _record(repo="someone/else", run="run-c"))] + rt, lifecycle = _rt(tmp_path, monkeypatch, comments) + monkeypatch.setenv("KB_BUGFIX_MAILBOX", "org/infra#135") + collect_events(rt, lifecycle) + assert _events(rt) == [] # no author allowlist: mailbox ignored + monkeypatch.setenv("KB_BUGFIX_AUTHORS", "TZhouAM") + collect_events(rt, lifecycle) + (event,) = _events(rt) + assert event["external_id"] == "bugfix_run:org/demo:run-a" + assert rt.ledger.get_cursor("demo", "bugfix_mailbox_after") == "3" + rt.github.comments.append(comment(4, "tzhouam", _record(run="run-d"))) + collect_events(rt, lifecycle) + assert len(_events(rt)) == 2 + + +def test_the_mailbox_is_never_stuck_behind_a_page_window(tmp_path, monkeypatch): + comments = [comment(i, "someone", {}) for i in range(1, 2001)] + [comment(2001, "tzhouam", _record(run="late"))] + rt, lifecycle = _rt(tmp_path, monkeypatch, comments) + monkeypatch.setenv("KB_BUGFIX_MAILBOX", "org/infra#135") + monkeypatch.setenv("KB_BUGFIX_AUTHORS", "tzhouam") + collect_events(rt, lifecycle) # first poll reads the first 2,000 + collect_events(rt, lifecycle) # the next continues past them + (event,) = _events(rt) + assert event["external_id"] == "bugfix_run:org/demo:late" + + +def test_a_code_fence_inside_the_record_and_malformed_files_are_handled(tmp_path, monkeypatch): + fenced = _record(run="fenced") + fenced["groups"][0]["root_cause"] = "the trace:\n```\nQueueFull\n```\n" + bad = _record(run="bad") + bad["groups"][0]["files"] = {"not": "a list"} + rt, lifecycle = _rt(tmp_path, monkeypatch, [comment(1, "tzhouam", bad), comment(2, "tzhouam", fenced)]) + monkeypatch.setenv("KB_BUGFIX_MAILBOX", "org/infra#135") + monkeypatch.setenv("KB_BUGFIX_AUTHORS", "tzhouam") + drops = tmp_path / "drops" + drops.mkdir() + (drops / "a-bad.json").write_text(json.dumps(bad)) + (drops / "b-good.json").write_text(json.dumps(_record(run="good"))) + collect_events(rt, lifecycle) + assert {e["external_id"] for e in _events(rt)} == {"bugfix_run:org/demo:fenced", "bugfix_run:org/demo:good"} + assert "QueueFull" in next(e for e in _events(rt) if e["external_id"].endswith("fenced"))["payload"]["body"] + assert (drops / "a-bad.json").exists() # malformed: left for inspection + + +def test_comments_sharing_the_last_seen_second_are_not_skipped(tmp_path, monkeypatch): + # comments 2,000 and 2,001 share a second across the page window + comments = [comment(i, "someone", {}) for i in range(1, 2001)] + [comment(2001, "tzhouam", _record(run="same"), + second=2000)] + rt, lifecycle = _rt(tmp_path, monkeypatch, comments) + monkeypatch.setenv("KB_BUGFIX_MAILBOX", "org/infra#135") + monkeypatch.setenv("KB_BUGFIX_AUTHORS", "tzhouam") + collect_events(rt, lifecycle) + collect_events(rt, lifecycle) + (event,) = _events(rt) + assert event["external_id"] == "bugfix_run:org/demo:same" + collect_events(rt, lifecycle) # the overlap is deduplicated by id + assert len(_events(rt)) == 1