diff --git a/doc/RFC-knowledge-intake.md b/doc/RFC-knowledge-intake.md index af586afdf..8962c95fe 100644 --- a/doc/RFC-knowledge-intake.md +++ b/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 + `` 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. diff --git a/doc/architecture/SPEC/config.md b/doc/architecture/SPEC/config.md index 44fe14fb7..9410d203c 100644 --- a/doc/architecture/SPEC/config.md +++ b/doc/architecture/SPEC/config.md @@ -1,6 +1,6 @@ # config.py —— 规范 - + `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`）； diff --git a/doc/architecture/SPEC/engine/steps/pr.md b/doc/architecture/SPEC/engine/steps/pr.md index c32d677fb..d931a2b78 100644 --- a/doc/architecture/SPEC/engine/steps/pr.md +++ b/doc/architecture/SPEC/engine/steps/pr.md @@ -1,6 +1,6 @@ # engine/steps/pr/ —— 规范 - + `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`）把同一条记录以 + `` + 一个 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` 的 diff --git a/src/infermatrix_copilot/config.py b/src/infermatrix_copilot/config.py index 6fa2f76ac..8ad0cdfff 100644 --- a/src/infermatrix_copilot/config.py +++ b/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" diff --git a/src/infermatrix_copilot/engine/steps/pr/debug.py b/src/infermatrix_copilot/engine/steps/pr/debug.py index 57a0bd1f6..31f8bb363 100644 --- a/src/infermatrix_copilot/engine/steps/pr/debug.py +++ b/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 = "" +_INTAKE_ISSUE = re.compile(r"^(?P[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)#(?P[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} diff --git a/test/test_knowledge_harvest.py b/test/test_knowledge_harvest.py index 5b0bb40db..e425c1124 100644 --- a/test/test_knowledge_harvest.py +++ b/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("\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