Take Copilot run lessons from pr_debug fix records into knowledge intake

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`（缺省 `<state_dir>/inbox/bugfix`，与运行侧
+`KNOWLEDGE_INTAKE_DIR` 配为同一目录），异机运行发到邮箱 issue `KB_BUGFIX_MAILBOX`（`owner/repo#N`）。仓库按完整名
+（不区分大小写）或别名精确匹配；歧义身份改名 `.quarantined` 隔离、从不改写；属于其他仓库的投放文件留待其处理。
+邮箱是公开 issue，只接受 `KB_BUGFIX_AUTHORS` 列出的作者（未设置则不读邮箱），每个仓库各自维护评论游标。
+事件 ID 与记录的 `event_id` 一致，重复投递幂等。

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

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
 
 

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 = ("<!-- infermatrix-copilot:bugfix-record:v2 -->",
+                  "<!-- infermatrix-copilot:bugfix-record:v1 -->")
+
+
+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/<repo>/``).
     Each file: {event_id, repo, run_ref, summary, diff_excerpt?}. Malformed files

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 = "<!-- infermatrix-copilot:bugfix-record:v2 -->"
+
+
+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