pr-debug: post landed-fix records to the knowledge intake mailbox issue

doc/RFC-knowledge-intake.md

@@ -204,11 +204,19 @@ adopted by pinning a new wheel version.
 2. ReviewBot: `KNOWLEDGE_INTAKE_ENABLED=true` — merged-PR recording plus daily
    shadow batches; inspect typed curation results, artifacts and
    `knowledge-batch` output.
-3. Wire the bugfix channel: point this repo's `KNOWLEDGE_INTAKE_DIR` and
-   the reviewbot's `COPILOT_INTAKE_DIR` at the **same directory** (v1
-   assumes a shared host). Without this pair, everything else works but
-   `pr_debug` learnings are silently absent from batches — both values
-   default to empty/off.
+3. Wire the bugfix channel. Same host: point this repo's
+   `KNOWLEDGE_INTAKE_DIR` and the reviewbot's `COPILOT_INTAKE_DIR` at the
+   **same directory**. Different hosts (the production case): set this
+   repo's `KNOWLEDGE_INTAKE_ISSUE` (plus `ALLOW_POST=1`, the outward-write
+   flag every post needs) and the reviewbot's
+   `KNOWLEDGE_INTAKE_ISSUE` to the same locked, pinned mailbox issue
+   (`owner/repo#N`, JiusiServe/InferMatrixCopilot#135 in production); the
+   harvest step posts each drop record there as a comment carrying
+   `<!-- infermatrix-copilot:bugfix-record:v1 -->` and one fenced JSON
+   block, and the reviewbot records collaborator-authored comments whose
+   `repo` names its watched repository. Without one of these pairs,
+   everything else works but `pr_debug` learnings are silently absent from
+   batches — all values default to empty/off.
 4. Create the bot's fork of this repo; set `KNOWLEDGE_FORK_SLUG`.
 5. `KNOWLEDGE_PR_ENABLED=true` under `POST_MODE=review` — PRs start; every
    one is reviewed like any other knowledge edit.

doc/architecture/SPEC/config.md

@@ -1,6 +1,6 @@
 # config.py —— 规范
 
-<!-- verified-against: 2026-08-28 -->
+<!-- verified-against: 2026-09-08 -->
 
 `LOC ~658 · 配置 · refactor-status: oversized`
 
@@ -19,7 +19,9 @@ MCP `repo_path` 授权面与 idempotency 保留期：
 checkout —— 最小权限；放宽是**运维**决定，不是调用方的）与
 `idem_retention_days`（默认 30，圈住 `.idem/` 索引）；以及
 知识回流落盘目录 `knowledge_intake_dir`（`KNOWLEDGE_INTAKE_DIR`，默认空 = 关闭；
-消费方见 `doc/RFC-knowledge-intake.md`）。
+消费方见 `doc/RFC-knowledge-intake.md`）与跨主机发布目标 `knowledge_intake_issue`
+（`KNOWLEDGE_INTAKE_ISSUE`，`owner/repo#N`，默认空 = 关闭：harvest step 把同一条记录
+以带标记的 JSON 评论贴到该锁定的 mailbox issue，供另一台主机上的 reviewbot 摄取）。
 
 ## 公开契约
 带全部可调项的 `Settings`；`reviewer` / `intent`（回退到 `agent_model`）；

doc/architecture/SPEC/engine/steps/pr.md

@@ -1,6 +1,6 @@
 # engine/steps/pr/ —— 规范
 
-<!-- verified-against: 2026-08-28 -->
+<!-- verified-against: 2026-09-08 -->
 
 `LOC ~1390（6 个文件） · step 库（PR） · refactor-status: ok`
 
@@ -61,8 +61,15 @@
 - `pr.post_review` 是**双闸**的（**C5**）。
 - `pr.harvest_debug_knowledge` **只在真实推送后落盘**（push 输出存在且非 dry-run，
   且至少一组修复同时有 root_cause 与 verification —— 与 debug memory 同一门槛），
-  写入 `settings.knowledge_intake_dir`（空 = 关闭，默认）。它是**刻意 fail-open**
-  的：目录未配、dry-run、无已验证修复、写失败（trace
+  投递到两个**彼此独立**的 sink：`settings.knowledge_intake_dir`（同主机落盘）与
+  `settings.knowledge_intake_issue`（跨主机 mailbox），各自配置了才尝试，任一失败
+  不压制另一个；两者都空 = 关闭（默认）。mailbox 投递（`_publish_intake_record`）把同一条记录以
+  `<!-- infermatrix-copilot:bugfix-record:v1 -->` + 一个 fenced JSON 块通过 `gh api`
+  贴为该 issue 的评论；它是对外写入，按不变量 5 双闸：配置了 mailbox **且**
+  `ALLOW_POST=1`，否则只 trace `knowledge_intake_publish_skipped` 作 dry-run；贴失败
+  trace `knowledge_intake_publish_failed` 并写 outputs `intake_publish_error`，成功写
+  `intake_comment`。它是**刻意 fail-open**
+  的：两处都未配、dry-run、无已验证修复、写失败（trace
   `knowledge_intake_write_failed` 后吞掉）都返回 ok —— 关闭学习回路
   **绝不能**让已落地的修复失败。它消费 executor 维护的 `state["outputs"]`，
   该表在 resume 时由 checkpoint 恢复（见 `test_knowledge_harvest.py` 的

src/infermatrix_copilot/config.py

@@ -194,6 +194,11 @@ class Settings(BaseSettings):
     # (e.g. the reviewbot's daily distillation batch). Empty disables the
     # drop; the step then no-ops without failing the run.
     knowledge_intake_dir: str = ""
+    # Cross-host publisher for the same record (KNOWLEDGE_INTAKE_ISSUE,
+    # "owner/repo#N"): the harvest step also posts it as a marked JSON
+    # comment on that locked mailbox issue, so a reviewbot on another host
+    # can ingest it. Operator opt-in; empty = off. Fail-open like the drop.
+    knowledge_intake_issue: str = ""
 
     # Engine
     run_root: Path = Path.home() / ".infermatrix-copilot" / "runs"

src/infermatrix_copilot/engine/steps/pr/debug.py

@@ -9,6 +9,7 @@
 from __future__ import annotations
 
 import json
+import re
 from datetime import datetime, timezone
 from pathlib import Path
 
@@ -199,9 +200,11 @@ async def _pr_harvest_debug_knowledge(ctx: StepContext) -> StepResult:
     is traced and swallowed (`knowledge_intake_write_failed`) — closing the
     learning loop must never fail the landed fix."""
     intake_dir = str(ctx.settings.knowledge_intake_dir or "").strip()
-    if not intake_dir:
+    intake_issue = str(ctx.settings.knowledge_intake_issue or "").strip()
+    if not intake_dir and not intake_issue:
         return StepResult(True, summary="knowledge intake disabled "
-                                        "(knowledge_intake_dir unset)")
+                                        "(knowledge_intake_dir and "
+                                        "knowledge_intake_issue unset)")
     outputs_map = ctx.state.get("outputs") or {}
     push_outputs = outputs_map.get("push")
     if push_outputs is None or push_outputs.get("dry_run"):
@@ -242,23 +245,92 @@ async def _pr_harvest_debug_knowledge(ctx: StepContext) -> StepResult:
         "groups": fixes,
         "created_at": datetime.now(timezone.utc).isoformat(),
     }
-    try:
-        directory = Path(intake_dir).expanduser()
-        directory.mkdir(parents=True, exist_ok=True)
-        final = directory / f"{ctx.run_dir.name}.json"
-        tmp = final.with_suffix(".tmp")
-        tmp.write_text(json.dumps(record, ensure_ascii=False, indent=1),
-                       encoding="utf-8")
-        tmp.replace(final)
-    except OSError as exc:
-        ctx.trace.record("knowledge_intake_write_failed",
-                         error=str(exc)[:300])
-        return StepResult(True, summary=f"intake drop failed (traced): {exc}",
-                          outputs={"intake_error": str(exc)[:300]})
-    ctx.trace.record("knowledge_intake_dropped", path=str(final),
-                     fixes=len(fixes))
+    # Two independent sinks: the local drop (same-host consumer) and the
+    # mailbox issue (cross-host consumer). Each is attempted when
+    # configured, and neither one's failure suppresses the other.
+    outputs: dict = {"fixes": len(fixes)}
+    delivered = []
+    if intake_dir:
+        try:
+            directory = Path(intake_dir).expanduser()
+            directory.mkdir(parents=True, exist_ok=True)
+            final = directory / f"{ctx.run_dir.name}.json"
+            tmp = final.with_suffix(".tmp")
+            tmp.write_text(json.dumps(record, ensure_ascii=False, indent=1),
+                           encoding="utf-8")
+            tmp.replace(final)
+            ctx.trace.record("knowledge_intake_dropped", path=str(final),
+                             fixes=len(fixes))
+            outputs["intake_path"] = str(final)
+            delivered.append("drop file")
+        except OSError as exc:
+            ctx.trace.record("knowledge_intake_write_failed",
+                             error=str(exc)[:300])
+            outputs["intake_error"] = str(exc)[:300]
+    if intake_issue:
+        outputs.update(_publish_intake_record(ctx, record, intake_issue))
+        if "intake_comment" in outputs:
+            delivered.append("mailbox issue")
+    if not delivered:
+        failures = ", ".join(
+            v for k, v in outputs.items() if k.endswith("_error")
+        )
+        return StepResult(True,
+                          summary=f"intake delivery failed (traced): {failures}",
+                          outputs=outputs)
     return StepResult(True,
                       summary=f"dropped {len(fixes)} verified fix record(s) "
-                              "for knowledge intake",
-                      outputs={"intake_path": str(final),
-                               "fixes": len(fixes)})
+                              "for knowledge intake via "
+                              + " and ".join(delivered),
+                      outputs=outputs)
+
+
+BUGFIX_RECORD_MARKER = "<!-- infermatrix-copilot:bugfix-record:v1 -->"
+_INTAKE_ISSUE = re.compile(r"^(?P<slug>[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)#(?P<number>[1-9][0-9]*)$")
+
+
+def _publish_intake_record(ctx: StepContext, record: dict, target: str) -> dict:
+    """Post the harvested record as a marked JSON comment on the configured
+    mailbox issue (`settings.knowledge_intake_issue`, owner/repo#N), the
+    cross-host counterpart of the drop directory. It is an outward write, so
+    it is double-gated like every other one (invariant 5): the operator
+    configures the mailbox AND sets ALLOW_POST=1; without the flag the post
+    is a traced dry-run skip. Fail-open: any failure is traced
+    (`knowledge_intake_publish_failed`) and reported in outputs, never raised.
+    """
+    if not ctx.settings.allow_post:
+        ctx.trace.record("knowledge_intake_publish_skipped",
+                         reason="ALLOW_POST=0 (dry-run)", issue=target)
+        return {"intake_publish_skipped": "ALLOW_POST=0 (dry-run)"}
+    match = _INTAKE_ISSUE.match(target)
+    if match is None:
+        ctx.trace.record("knowledge_intake_publish_failed",
+                         error=f"KNOWLEDGE_INTAKE_ISSUE is not owner/repo#N: {target[:80]}")
+        return {"intake_publish_error": "bad KNOWLEDGE_INTAKE_ISSUE"}
+    body = (BUGFIX_RECORD_MARKER + "\n```json\n"
+            + json.dumps(record, ensure_ascii=False, indent=1) + "\n```\n")
+    body_path = ctx.run_dir / "knowledge-intake-comment.md"
+    try:
+        body_path.write_text(body, encoding="utf-8")
+    except OSError as exc:
+        ctx.trace.record("knowledge_intake_publish_failed", error=str(exc)[:300])
+        return {"intake_publish_error": str(exc)[:300]}
+    try:
+        code, out = _gh(["api", f"repos/{match.group('slug')}/issues/"
+                         f"{match.group('number')}/comments",
+                         "-X", "POST", "-F", f"body=@{body_path}"])
+    except Exception as exc:  # noqa: BLE001 — a timeout or launch error
+        # must not turn a landed fix into a BLOCKED step.
+        detail = f"{type(exc).__name__}: {exc}"[:300]
+        ctx.trace.record("knowledge_intake_publish_failed", error=detail)
+        return {"intake_publish_error": detail}
+    if code != 0:
+        ctx.trace.record("knowledge_intake_publish_failed",
+                         error=(out or f"gh exit {code}")[:300])
+        return {"intake_publish_error": (out or f"gh exit {code}")[:300]}
+    try:
+        url = str(json.loads(out).get("html_url") or "")
+    except (ValueError, AttributeError):
+        url = ""
+    ctx.trace.record("knowledge_intake_published", issue=target, url=url)
+    return {"intake_comment": url or target}

test/test_knowledge_harvest.py

@@ -89,13 +89,148 @@ def test_landed_fix_writes_one_record(registry, settings, trace, tmp_path):
     assert list((tmp_path / "intake").iterdir()) == [path]
 
 
+def _capture_gh(monkeypatch, *, code=0, out=None):
+    import infermatrix_copilot.engine.steps.pr.debug as debug_module
+    calls = []
+
+    def fake_gh(args, cwd=None):
+        calls.append(list(args))
+        return code, (json.dumps({"html_url": "https://github.com/c/copilot/"
+                                  "issues/135#issuecomment-9"})
+                      if out is None else out)
+
+    monkeypatch.setattr(debug_module, "_gh", fake_gh)
+    return calls
+
+
+def test_landed_fix_is_posted_to_the_mailbox_issue(
+    registry, settings, trace, tmp_path, monkeypatch
+):
+    settings.knowledge_intake_dir = str(tmp_path / "intake")
+    settings.knowledge_intake_issue = "c/copilot#135"
+    settings.allow_post = True
+    settings.repo_full_names = {"demo": "owner/demo"}
+    calls = _capture_gh(monkeypatch)
+    result = _run(registry, settings, tmp_path=tmp_path, trace=trace,
+                  state=_state())
+    assert result.ok and "via drop file and mailbox issue" in result.summary
+    assert result.outputs["intake_comment"].endswith("#issuecomment-9")
+    assert calls == [["api", "repos/c/copilot/issues/135/comments", "-X", "POST",
+                      "-F", f"body=@{tmp_path / 'run-x' / 'knowledge-intake-comment.md'}"]]
+    body = (tmp_path / "run-x" / "knowledge-intake-comment.md").read_text(
+        encoding="utf-8"
+    )
+    assert body.startswith("<!-- infermatrix-copilot:bugfix-record:v1 -->\n```json\n")
+    posted = json.loads(body.split("```json\n", 1)[1].rsplit("```", 1)[0])
+    # The comment carries exactly the record the drop file carries.
+    dropped = json.loads((tmp_path / "intake" / "run-x.json").read_text(encoding="utf-8"))
+    assert posted == dropped and posted["repo"] == "owner/demo"
+    assert list(trace.events("knowledge_intake_published"))
+
+
+def test_mailbox_post_failure_is_swallowed(
+    registry, settings, trace, tmp_path, monkeypatch
+):
+    settings.knowledge_intake_dir = str(tmp_path / "intake")
+    settings.knowledge_intake_issue = "c/copilot#135"
+    settings.allow_post = True
+    _capture_gh(monkeypatch, code=1, out="HTTP 403: locked")
+    result = _run(registry, settings, tmp_path=tmp_path, trace=trace,
+                  state=_state())
+    assert result.ok  # the landed fix and its drop file are not undone
+    assert (tmp_path / "intake" / "run-x.json").is_file()
+    assert "via drop file" in result.summary
+    assert result.outputs["intake_publish_error"] == "HTTP 403: locked"
+    assert "intake_comment" not in result.outputs
+    assert list(trace.events("knowledge_intake_publish_failed"))
+
+
+def test_mailbox_publisher_exception_is_swallowed(
+    registry, settings, trace, tmp_path, monkeypatch
+):
+    import subprocess
+
+    import infermatrix_copilot.engine.steps.pr.debug as debug_module
+
+    def hanging_gh(args, cwd=None):
+        raise subprocess.TimeoutExpired(cmd=["gh", *args], timeout=120)
+
+    monkeypatch.setattr(debug_module, "_gh", hanging_gh)
+    settings.knowledge_intake_dir = str(tmp_path / "intake")
+    settings.knowledge_intake_issue = "c/copilot#135"
+    settings.allow_post = True
+    result = _run(registry, settings, tmp_path=tmp_path, trace=trace,
+                  state=_state())
+    assert result.ok and "via drop file" in result.summary
+    assert result.outputs["intake_publish_error"].startswith("TimeoutExpired")
+    assert list(trace.events("knowledge_intake_publish_failed"))
+
+
+def test_mailbox_post_is_dry_run_without_allow_post(
+    registry, settings, trace, tmp_path, monkeypatch
+):
+    """Invariant 5: an outward write needs the env flag too."""
+    settings.knowledge_intake_dir = str(tmp_path / "intake")
+    settings.knowledge_intake_issue = "c/copilot#135"
+    settings.allow_post = False
+    calls = _capture_gh(monkeypatch)
+    result = _run(registry, settings, tmp_path=tmp_path, trace=trace,
+                  state=_state())
+    assert result.ok and calls == []
+    assert "via drop file" in result.summary
+    assert result.outputs["intake_publish_skipped"].startswith("ALLOW_POST=0")
+    assert list(trace.events("knowledge_intake_publish_skipped"))
+
+
+def test_mailbox_is_off_without_configuration(
+    registry, settings, trace, tmp_path, monkeypatch
+):
+    settings.knowledge_intake_dir = str(tmp_path / "intake")
+    calls = _capture_gh(monkeypatch)
+    result = _run(registry, settings, tmp_path=tmp_path, trace=trace,
+                  state=_state())
+    assert result.ok and calls == []
+    assert "intake_comment" not in result.outputs
+
+
+def test_mailbox_only_configuration_posts_without_a_drop_dir(
+    registry, settings, trace, tmp_path, monkeypatch
+):
+    """The production shape: no local consumer, only the mailbox."""
+    settings.knowledge_intake_issue = "c/copilot#135"
+    settings.allow_post = True
+    calls = _capture_gh(monkeypatch)
+    result = _run(registry, settings, tmp_path=tmp_path, trace=trace,
+                  state=_state())
+    assert result.ok and "via mailbox issue" in result.summary
+    assert len(calls) == 1 and "intake_comment" in result.outputs
+    assert "intake_path" not in result.outputs
+    assert not (tmp_path / "intake").exists()
+
+
+def test_drop_failure_does_not_suppress_the_mailbox(
+    registry, settings, trace, tmp_path, monkeypatch
+):
+    blocker = tmp_path / "intake"
+    blocker.write_text("a file where the directory should be")
+    settings.knowledge_intake_dir = str(blocker)
+    settings.knowledge_intake_issue = "c/copilot#135"
+    settings.allow_post = True
+    calls = _capture_gh(monkeypatch)
+    result = _run(registry, settings, tmp_path=tmp_path, trace=trace,
+                  state=_state())
+    assert result.ok and "via mailbox issue" in result.summary
+    assert "intake_error" in result.outputs and "intake_comment" in result.outputs
+    assert len(calls) == 1
+
+
 def test_write_failure_is_swallowed(registry, settings, trace, tmp_path):
     blocker = tmp_path / "intake"
     blocker.write_text("a file where the directory should be")
     settings.knowledge_intake_dir = str(blocker)
     result = _run(registry, settings, trace, tmp_path, _state())
     assert result.ok  # the landed fix must never be failed by the drop
-    assert "intake drop failed" in result.summary
+    assert "intake delivery failed" in result.summary
     assert "intake_error" in result.outputs