diff --git a/doc/GUIDE.md b/doc/GUIDE.md
index be851fe..f2b5b18 100644
--- a/doc/GUIDE.md
+++ b/doc/GUIDE.md
@@ -426,7 +426,8 @@ pr.fetch_diff → pr.gate_check → agent.review_diff → [post] pr.post_review
 ```
 workspace.guard_clean → pr.checkout_branch → pr.fetch_ci_failures → pr.group_failures
   → [not report_only] agent.debug_group (foreach failure_groups)
-  → [not report_only] review.patch_gate → [not report_only] ci.push → report.final_summary
+  → [not report_only] review.patch_gate → [not report_only] ci.push
+  → [not report_only] pr.harvest_debug_knowledge → report.final_summary
 ```
 
 **`pr-rebase`** — push 只针对 PR head 分支，force 仅 with-lease。
diff --git a/doc/README.md b/doc/README.md
index 1690e51..667b06b 100644
--- a/doc/README.md
+++ b/doc/README.md
@@ -36,6 +36,7 @@
 | [`features/strict-review-deep-engine.md`](features/strict-review-deep-engine.md) | 已实现（默认开） |
 | [`features/review-recall.md`](features/review-recall.md) | 已实现（默认开）——v14/v15 召回攻坚 |
 | [`features/auto-run.md`](features/auto-run.md) | **draft，未实现**——GitHub 事件触发 |
+| [`RFC-knowledge-intake.md`](RFC-knowledge-intake.md) | 已实现（默认关，shadow 起步）——合并 PR 与 bugfix 学习经 reviewbot 蒸馏，以 fork PR 回流知识库，人审合并即晋升 |
 
 ## architecture/ —— 维护者
 
diff --git a/doc/RFC-knowledge-intake.md b/doc/RFC-knowledge-intake.md
new file mode 100644
index 0000000..d12f3c7
--- /dev/null
+++ b/doc/RFC-knowledge-intake.md
@@ -0,0 +1,167 @@
+# RFC — Knowledge intake: merged-PR and bugfix learnings return as reviewed knowledge PRs
+
+- Status: implemented, shadow-first and default-off — copilot side in
+  PR #110 (`pr.harvest_debug_knowledge`), reviewbot side in
+  zuiho-kai/omni-reviewbot#21 (intake ledger) and #22 (daily distiller)
+- Owner: knowledge plane (`knowledge/`, `AGENTS.md` contract) on this side;
+  `omni_reviewbot.knowledge_intake` / `knowledge_distiller` on the bot side
+- Evidence: `test/test_knowledge_harvest.py` (step + executor/resume e2e),
+  reviewbot `tests/test_knowledge_intake.py`, `tests/test_knowledge_distiller.py`,
+  `tests/test_e2e_knowledge_flow.py` (cross-repo contract pinned)
+
+## Motivation
+
+The knowledge tree has a written intake contract — PR-learning produces only
+executable rules in the nearest owner's `rules.md`, raw material is deleted,
+two validators plus the release audit gate every batch — but no automated
+feeder. Learnings evaporate at exactly the two moments they are freshest:
+
+1. **A vllm-omni PR merges.** The review discussion and final diff often
+   carry a generalizable lesson (a contract the reviewer had to rediscover,
+   a failure class the diff fixed), and today nobody distills it.
+2. **A copilot `pr_debug` run lands a fix.** The agent produced a verified
+   root cause and fix; `record_debug_memory` keeps it in the run-scoped
+   debug store, but nothing carries it into the human-curated tree.
+
+The constraint that shapes everything below is the knowledge plane's safety
+model: **agents propose, humans promote**. Any automation that writes the
+tree directly would break it. A pull request against this repo *is* the
+promotion gate — review and merge are the human act — so the design's job is
+to deliver well-formed, validator-clean proposals as PRs, and nothing else.
+
+## Design
+
+Three stages, deliberately decoupled so each can fail without losing events.
+
+### 1. Recording (cheap, every watch cycle)
+
+- **Every merged PR counts — literally.** The reviewbot scans GitHub's
+  closed pulls directly (drafts and never-tracked PRs included) with a
+  durable cursor in GitHub's own timestamp format; rows are unique per
+  source, so overlapping windows and crashes re-read but never drop or
+  duplicate. It does not trust its own `tracked_pulls`.
+- **Bugfix runs arrive as drop files.** This repo's
+  `pr.harvest_debug_knowledge` step (end of the `pr-debug` playbook, risk
+  `knowledge`) writes one JSON record per run into
+  `knowledge_intake_dir` (`KNOWLEDGE_INTAKE_DIR`) — but only when the fix
+  actually landed: a real, non-dry-run `ci.push` and at least one fix with
+  both root cause and verification (the same bar as debug memory). The
+  step is fail-open by design: unset directory, dry-run, zero verified
+  fixes, or a write error each no-op with an honest summary — closing the
+  learning loop must never fail the landed fix.
+- Records land in a `knowledge_intake` SQLite table, `pending` until a
+  batch consumes them.
+
+### 2. Distillation (expensive, at most once a day)
+
+- The model **only proposes**: it reads a dedicated work clone of this repo
+  (never the live checkout the Direct bridge imports) and returns
+  catalog-constrained structured output — target `rules.md` page from the
+  scanned catalog, a new auditable rule id, one complete section in the
+  page's own language and format, cited sources. Source events are fenced
+  as `<untrusted_data>`; instructions never come from them.
+- Code **applies mechanically and append-only**: new sections plus an
+  `updated:` frontmatter bump. Existing sections are never rewritten —
+  merging and pruning stay human/curator work. Proposals that miss the
+  catalog, collide with an existing rule id, or fail shape checks are
+  dropped and counted.
+- The **validators are the gate**: `check_knowledge_tree.py` and
+  `check_wiki_lint.py` run in the clone; a failure — or a missing
+  validator — fails the batch closed. Rows stay `pending` and retry on the
+  next daily batch, bounded by an attempt cap that parks poison rows
+  visibly as `error`. The vllm-omni release audit runs in CI on the PR, as
+  it does for any knowledge edit.
+
+### 3. Publication (fork-based, double-gated)
+
+- The date-scoped branch (`knowledge/intake-YYYY-MM-DD`) pushes to the
+  **bot's fork**; the PR targets this repo. The bot holds no write access
+  here — merging the PR is the human promotion gate, and the PR body lists
+  every source event plus the validator transcript.
+- Shadow is the default: the batch stays a local commit plus an artifact
+  under the bot's `state/artifacts/knowledge-intake/`. Posting requires
+  `POST_MODE=review` **and** `KNOWLEDGE_PR_ENABLED` with a configured
+  fork — the same double-gate shape as every outward write on both sides.
+- The automatic cadence yields at most one PR per day. The distiller
+  reuses the day-branch's **open** PR when one exists; if the day's PR was
+  already merged or closed, a manual same-day rerun with new rules opens a
+  follow-up PR on the same branch (showing only the new commits) — a
+  deliberate consequence of trusting GitHub's open-PR state over local
+  bookkeeping. Same-day reruns build on the day's branch (never
+  reset it) and re-attempt publication of anything committed but
+  unpublished — so a manual `knowledge-batch` retry after a failed push or
+  PR creation recovers the same branch, and a no-rules retry can never
+  mark rows done while their committed batch sits unpublished on the same
+  branch. **The recovery boundary is the day**: the automatic cadence
+  retries ~24h later under a fresh date branch, re-distilling the still-
+  pending rows from upstream state. Rows are never lost, but a prior
+  day's committed-yet-unpublished branch is not resumed — at worst it
+  survives as a dead branch on the fork (no PR references it). Cross-day
+  branch recovery was considered and skipped: it trades real complexity
+  for saving one day of latency on an already-daily loop.
+
+## The cross-repo contract
+
+The drop record is the only coupling between the two repos:
+
+```json
+{"run_id": "...", "repo": "owner/repo", "pr": 123, "kind": "bugfix_run",
+ "title": "...", "groups": [{"signature": "...", "jobs": [],
+ "root_cause": "...", "fix_summary": "...", "verification": "...",
+ "files": []}], "created_at": "..."}
+```
+
+- `repo` carries the full GitHub identity from `repo_full_names` when
+  configured; the reviewbot normalizes alias-form values to its watched
+  slug so rows can never strand under an unqueryable key.
+- Each side pins the shape in its own tests: the producer here in
+  `test_knowledge_harvest.py` (field-level assertions on the written
+  record), the consumer in the reviewbot's `tests/test_e2e_knowledge_flow.py`
+  via a **deliberately duplicated** fixture — the repos share no code, so
+  there is no single fixture both import. A change on one side breaks that
+  side's test but not the other's; the contract therefore changes only in
+  lockstep, with this section as the reference.
+
+## Fit with this repo's invariants
+
+- **Repo neutrality**: the harvest step contains no repo literal; the drop
+  directory is plain settings, empty by default.
+- **State via `state_updates` / executor restore**: the step consumes the
+  executor-maintained `outputs` map, which the resume path restores from
+  `progress.json` — pinned by the crash-then-resume e2e test.
+- **Read-wide/write-narrow**: the tree is still written only by humans
+  merging PRs; the automation's write surface is a drop directory and a
+  fork.
+
+## Alternatives considered
+
+- **Per-event PRs** — cleanest provenance, but several PRs a day on an
+  active repo turns the promotion gate into review spam. Daily batch with
+  skip-if-empty bounds the load; provenance lives in the PR body instead.
+- **Direct writes to the tree (or pushing to this repo)** — rejected
+  outright: it deletes the human promotion step the knowledge plane is
+  built on, and would require granting the bot write access here.
+- **Copilot-hosted distiller** — this repo has no daemon; the reviewbot
+  already runs a poll loop with maintenance gating, budgets, and crash
+  recovery, so the scheduled half lives there and this repo stays a
+  library plus one playbook step.
+- **Stopping at skill candidates / debug memory** — that is where learnings
+  already go today; both are run-scoped proposals with no path into the
+  human-curated tree. This design is that missing path, not a replacement
+  for them.
+
+## Rollout
+
+1. Reviewbot: `KNOWLEDGE_INTAKE_ENABLED=true` — merged-PR recording plus
+   daily shadow batches; inspect artifacts and `knowledge-batch` output.
+2. 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. Create the bot's fork of this repo; set `KNOWLEDGE_FORK_SLUG`.
+4. `KNOWLEDGE_PR_ENABLED=true` under `POST_MODE=review` — PRs start; every
+   one is reviewed like any other knowledge edit.
+
+Rollback at any stage is turning the flag off; nothing in the tree changes
+without a merged PR.
diff --git a/doc/architecture/SPEC/config.md b/doc/architecture/SPEC/config.md
index bf5f6bb..44fe14f 100644
--- a/doc/architecture/SPEC/config.md
+++ b/doc/architecture/SPEC/config.md
@@ -17,7 +17,9 @@ metrics 与升级，提供带类型字段和安全默认值；外加 PR4d 知识
 MCP `repo_path` 授权面与 idempotency 保留期：
 `mcp_allowed_repo_roots`（`MCP_ALLOWED_REPO_ROOTS`，空 = 只有已配置的
 checkout —— 最小权限；放宽是**运维**决定，不是调用方的）与
-`idem_retention_days`（默认 30，圈住 `.idem/` 索引）。
+`idem_retention_days`（默认 30，圈住 `.idem/` 索引）；以及
+知识回流落盘目录 `knowledge_intake_dir`（`KNOWLEDGE_INTAKE_DIR`，默认空 = 关闭；
+消费方见 `doc/RFC-knowledge-intake.md`）。
 
 ## 公开契约
 带全部可调项的 `Settings`；`reviewer` / `intent`（回退到 `agent_model`）；
diff --git a/doc/architecture/SPEC/engine/steps/pr.md b/doc/architecture/SPEC/engine/steps/pr.md
index 7574245..c32d677 100644
--- a/doc/architecture/SPEC/engine/steps/pr.md
+++ b/doc/architecture/SPEC/engine/steps/pr.md
@@ -2,7 +2,7 @@
 
 <!-- verified-against: 2026-08-28 -->
 
-`LOC ~1300（6 个文件） · step 库（PR） · refactor-status: ok`
+`LOC ~1390（6 个文件） · step 库（PR） · refactor-status: ok`
 
 ## 职责
 受守卫的推送、只读的 PR 抓取/门禁、PR rebase、PR debug、受门禁的评审发布。
@@ -15,13 +15,14 @@
 - `rebase.py` —— `pr.checkout_branch`、`pr.rebase_onto_base`、`pr.analyze_diff`、
   `agent.verify_module`。
 - `debug.py` —— `pr.fetch_ci_failures`（+ `_enrich_ci_logs`）、`pr.group_failures`、
-  `agent.debug_group`。
+  `agent.debug_group`、`pr.harvest_debug_knowledge`。
 - `publish.py` —— 对外写入（risk=push）：`ci.push`、`pr.post_review`。
 - `utils.py` —— 纯函数 `extract_signature`（及其正则）。
 
-## Steps（11 个）
+## Steps（12 个）
 `ci.push`（script/push）；`pr.fetch_diff`、`pr.gate_check`、`pr.checkout_branch`、
 `pr.analyze_diff`、`pr.fetch_ci_failures`、`pr.group_failures`（deterministic/read）；
+`pr.harvest_debug_knowledge`（deterministic/knowledge）；
 `pr.rebase_onto_base`、`agent.debug_group`（agent/write_workspace）；
 `agent.verify_module`（validation/read）；`pr.post_review`（script/push）。
 
@@ -58,6 +59,14 @@
 - `pr.fetch_ci_failures` 经 profile 选定的 CI provider 富化日志，否则记一条
   `capability_gap`（**E2**）；`pr.group_failures` 按**归一化后**的签名分组。
 - `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
+  `knowledge_intake_write_failed` 后吞掉）都返回 ok —— 关闭学习回路
+  **绝不能**让已落地的修复失败。它消费 executor 维护的 `state["outputs"]`，
+  该表在 resume 时由 checkpoint 恢复（见 `test_knowledge_harvest.py` 的
+  crash-then-resume 用例）。设计记录：`doc/RFC-knowledge-intake.md`。
 
 ## 边界 —— 不属于这里
 不含推送授权逻辑（那是 `push`）；不含 CI 日志抓取机制（那是 `ci/providers`）；
@@ -70,6 +79,7 @@
 ## 测试
 `test_pr_steps.py`（含钉 ref run 域隔离、head 移动检测、stale expected_head
 BLOCK、worktree 分键/拒外来树）、`test_push_and_steps.py`、
+`test_knowledge_harvest.py`（harvest step + executor crash-then-resume）、
 `test_ci_and_repo_map.py`（注意：`test_ci_and_repo_map` monkeypatch 的是
 `pr.debug._gh`，即 `pr.fetch_ci_failures` 绑定 `gh` 的那个子模块）；
 端到端：`test_thin_mcp_server.py`（评审跑在钉住的 worktree 上、head 移动
diff --git a/doc/architecture/SPEC/playbooks/PLAYBOOKS.md b/doc/architecture/SPEC/playbooks/PLAYBOOKS.md
index 1cbe871..9df86e8 100644
--- a/doc/architecture/SPEC/playbooks/PLAYBOOKS.md
+++ b/doc/architecture/SPEC/playbooks/PLAYBOOKS.md
@@ -1,6 +1,6 @@
 # playbooks/*.yaml —— 规范
 
-<!-- verified-against: 2026-08-25 -->
+<!-- verified-against: 2026-08-28 -->
 
 `8 个文件 · 声明式编排数据 · refactor-status: ok`
 
diff --git a/playbooks/pr-debug.yaml b/playbooks/pr-debug.yaml
index 8069bfe..f7cd59f 100644
--- a/playbooks/pr-debug.yaml
+++ b/playbooks/pr-debug.yaml
@@ -20,5 +20,9 @@ steps:
   - {id: debug, step: agent.debug_group, foreach: failure_groups, when: not report_only}
   - {id: gate, step: review.patch_gate, params: {pre_push: true}, when: not report_only}
   - {id: push, step: ci.push, when: not report_only}
+  # Landed-fix learning loop: drops a bugfix-run record for external
+  # knowledge intake; a no-op unless knowledge_intake_dir is configured
+  # and the push above was real (not dry-run).
+  - {id: harvest, step: pr.harvest_debug_knowledge, when: not report_only}
   - {id: report, step: report.final_summary}
 success: each failure group fixed at root cause or escalated; additive push only
diff --git a/src/infermatrix_copilot/config.py b/src/infermatrix_copilot/config.py
index 97dd9da..6fa2f76 100644
--- a/src/infermatrix_copilot/config.py
+++ b/src/infermatrix_copilot/config.py
@@ -188,6 +188,12 @@ class Settings(BaseSettings):
     # doc is reachable on demand via doc_search/doc_read.
     knowledge_dir: Path = _resource_dir("knowledge")
     knowledge_general_docs: list[str] = ["general/_index.md"]
+    # Drop directory for bugfix-run intake records (KNOWLEDGE_INTAKE_DIR):
+    # after a pr_debug run lands a real push, `pr.harvest_debug_knowledge`
+    # writes one JSON record here for an external knowledge-intake consumer
+    # (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 = ""
 
     # 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 8aa3605..57a0bd1 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
+from datetime import datetime, timezone
 from pathlib import Path
 
 from ....scopes import post_plan_scope
@@ -182,3 +183,82 @@ async def _pr_debug_group(ctx: StepContext) -> StepResult:
     else:
         result.summary = f"'{sig}': {result.summary[:250]}"
     return result
+
+
+@step("pr.harvest_debug_knowledge", "deterministic", "knowledge",
+      "Drop a bugfix-run record for external knowledge intake.")
+async def _pr_harvest_debug_knowledge(ctx: StepContext) -> StepResult:
+    """After a pr_debug run whose fixes actually landed (a real, non-dry-run
+    push), write one JSON record — failure signatures with their verified root
+    causes and fixes — into `settings.knowledge_intake_dir`, where an external
+    consumer (the reviewbot's knowledge-intake scanner) batches it into a
+    knowledge PR for human promotion.
+
+    Deliberately fail-open as a no-op: unset directory, dry-run push, or zero
+    verified fixes each return ok with an honest summary, and a write failure
+    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:
+        return StepResult(True, summary="knowledge intake disabled "
+                                        "(knowledge_intake_dir 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"):
+        return StepResult(True, summary="no landed fix (push missing or "
+                                        "dry-run) — nothing to harvest")
+    groups = ctx.state.get("failure_groups") or []
+    debug_outputs = outputs_map.get("debug") or {}
+    # foreach with one item returns the agent's outputs unmerged; more items
+    # arrive keyed by index (executor._merge).
+    indexed = ({"0": debug_outputs} if "root_cause" in debug_outputs
+               else {k: v for k, v in debug_outputs.items() if k.isdigit()})
+    fixes = []
+    for index, group in enumerate(groups):
+        out = indexed.get(str(index)) or {}
+        if not (out.get("root_cause") and out.get("verification")):
+            continue  # same bar as debug memory: unverified fixes don't teach
+        fixes.append({
+            "signature": str(group.get("signature", ""))[:300],
+            "jobs": [str(j)[:120] for j in (group.get("jobs") or [])[:10]],
+            "root_cause": str(out.get("root_cause", ""))[:2_000],
+            "fix_summary": str(out.get("fix_summary", ""))[:2_000],
+            "verification": str(out.get("verification", ""))[:1_000],
+            "files": [str(f)[:300]
+                      for f in (out.get("files_modified") or [])[:20]],
+        })
+    if not fixes:
+        return StepResult(True, summary="no verified fixes to harvest")
+    spec = ctx.state.get("task_spec") or {}
+    alias = str(spec.get("repo", ""))
+    record = {
+        "run_id": ctx.run_dir.name,
+        # Full GitHub identity when configured, so the consumer keys the
+        # event correctly; the local alias otherwise.
+        "repo": ctx.settings.repo_full_names.get(alias, alias),
+        "pr": spec.get("pr"),
+        "kind": "bugfix_run",
+        "title": f"pr_debug fixes for PR #{spec.get('pr')}",
+        "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))
+    return StepResult(True,
+                      summary=f"dropped {len(fixes)} verified fix record(s) "
+                              "for knowledge intake",
+                      outputs={"intake_path": str(final),
+                               "fixes": len(fixes)})
diff --git a/test/test_knowledge_harvest.py b/test/test_knowledge_harvest.py
new file mode 100644
index 0000000..5b0bb40
--- /dev/null
+++ b/test/test_knowledge_harvest.py
@@ -0,0 +1,211 @@
+"""Guardrails for `pr.harvest_debug_knowledge` — the landed-fix intake drop.
+
+Pinned behavior: the step is a strict no-op without a configured intake dir,
+without a real (non-dry-run) push, or without a verified fix; it writes one
+JSON record per run keyed by run id; and a write failure is traced and
+swallowed — it must never fail a run whose fix already landed."""
+
+import asyncio
+import json
+
+import pytest
+
+from infermatrix_copilot.engine.registry import StepRegistry
+from infermatrix_copilot.engine.step import StepContext
+from infermatrix_copilot.engine.steps import register_builtin_steps
+
+
+@pytest.fixture()
+def registry():
+    return register_builtin_steps(StepRegistry())
+
+
+def _run(registry, settings, trace, tmp_path, state):
+    handler = registry.get("pr.harvest_debug_knowledge").handler
+    ctx = StepContext(settings=settings, state=state, params={},
+                      run_dir=tmp_path / "run-x", trace=trace, llm=None)
+    (tmp_path / "run-x").mkdir(exist_ok=True)
+    return asyncio.run(handler(ctx))
+
+
+def _state(*, dry_run=False, verified=True):
+    debug_outputs = {
+        "root_cause": "import cycle in helper" if verified else "",
+        "fix_summary": "moved the import",
+        "verification": "pytest test_x passed" if verified else "",
+        "files_modified": ["pkg/helper.py"],
+    }
+    return {
+        "task_spec": {"kind": "pr_debug", "repo": "demo", "pr": 42},
+        "failure_groups": [{"signature": "ImportError: cycle",
+                            "jobs": ["unit"]}],
+        "outputs": {
+            "debug": debug_outputs,
+            "push": {"dry_run": True} if dry_run else {},
+        },
+    }
+
+
+def test_disabled_without_intake_dir(registry, settings, trace, tmp_path):
+    result = _run(registry, settings, trace, tmp_path, _state())
+    assert result.ok
+    assert "disabled" in result.summary
+    assert not list(tmp_path.glob("intake/*.json"))
+
+
+def test_dry_run_push_harvests_nothing(registry, settings, trace, tmp_path):
+    settings.knowledge_intake_dir = str(tmp_path / "intake")
+    result = _run(registry, settings, trace, tmp_path,
+                  _state(dry_run=True))
+    assert result.ok
+    assert "no landed fix" in result.summary
+    assert not (tmp_path / "intake").exists()
+
+
+def test_unverified_fix_is_not_harvested(registry, settings, trace, tmp_path):
+    settings.knowledge_intake_dir = str(tmp_path / "intake")
+    result = _run(registry, settings, trace, tmp_path,
+                  _state(verified=False))
+    assert result.ok
+    assert "no verified fixes" in result.summary
+
+
+def test_landed_fix_writes_one_record(registry, settings, trace, tmp_path):
+    settings.knowledge_intake_dir = str(tmp_path / "intake")
+    settings.repo_full_names = {"demo": "owner/demo"}
+    result = _run(registry, settings, trace, tmp_path, _state())
+    assert result.ok
+    path = tmp_path / "intake" / "run-x.json"
+    assert path.is_file()
+    record = json.loads(path.read_text(encoding="utf-8"))
+    assert record["run_id"] == "run-x"
+    assert record["repo"] == "owner/demo"  # full identity, not the alias
+    assert record["pr"] == 42
+    assert record["kind"] == "bugfix_run"
+    assert record["groups"][0]["signature"] == "ImportError: cycle"
+    assert record["groups"][0]["root_cause"] == "import cycle in helper"
+    assert result.outputs["fixes"] == 1
+    # No stray temp file left behind by the atomic write.
+    assert list((tmp_path / "intake").iterdir()) == [path]
+
+
+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_error" in result.outputs
+
+
+def _executor_env(settings, trace, tmp_path, calls=None):
+    """A real Executor over the real registry plus two stand-ins for the
+    debug/push steps' externals — the e2e boundary is the process edge
+    (agent LLM, git push), not the engine."""
+    from infermatrix_copilot.engine import Executor, StepResult, StepSpec
+    from infermatrix_copilot.notify import Notifier
+
+    registry = register_builtin_steps(StepRegistry())
+
+    async def debug_sim(ctx):
+        if calls is not None:
+            calls.append("debug")
+        return StepResult(
+            True, summary="fixed",
+            outputs={"root_cause": "flaky fixture reuse",
+                     "fix_summary": "isolated the fixture",
+                     "verification": "pytest -k fixture passed",
+                     "files_modified": ["tests/conftest.py"],
+                     "state_updates": {
+                         "failure_groups": [{"signature": "FixtureError",
+                                             "jobs": ["unit"]}]}})
+
+    async def push_sim(ctx):
+        return StepResult(True, summary="pushed", outputs={})
+
+    registry.register(StepSpec("test.debug_sim", "deterministic", "read",
+                               debug_sim))
+    registry.register(StepSpec("test.push_sim", "deterministic", "read",
+                               push_sim))
+    run_dir = tmp_path / "run-e2e"
+    notifier = Notifier(settings, run_dir, trace, "run-e2e")
+    executor = Executor(registry, settings, run_dir=run_dir, trace=trace,
+                        notifier=notifier)
+    return executor
+
+
+def _harvest_playbook(*, include_harvest=True):
+    from infermatrix_copilot.playbooks.store import Playbook, PlaybookStep
+
+    steps = [PlaybookStep("debug", "test.debug_sim"),
+             PlaybookStep("push", "test.push_sim")]
+    if include_harvest:
+        steps.append(PlaybookStep("harvest", "pr.harvest_debug_knowledge"))
+    return Playbook(name="e2e-debug", version=1, status="active",
+                    task_kinds=["pr_debug"], repos=[], steps=steps)
+
+
+def test_e2e_executor_runs_harvest_after_push(settings, trace, tmp_path):
+    """Whole-pipeline path: the executor's own outputs map (not hand-built
+    state) feeds the harvest step, and the drop lands with the checkpoint
+    contract intact."""
+    settings.knowledge_intake_dir = str(tmp_path / "intake")
+    settings.repo_full_names = {"demo": "owner/demo"}
+    executor = _executor_env(settings, trace, tmp_path)
+    outcome = asyncio.run(executor.run(
+        _harvest_playbook(),
+        {"task_spec": {"kind": "pr_debug", "repo": "demo", "pr": 42}},
+    ))
+    assert outcome.status == "done"
+    drops = list((tmp_path / "intake").glob("*.json"))
+    assert len(drops) == 1
+    record = json.loads(drops[0].read_text(encoding="utf-8"))
+    assert record["repo"] == "owner/demo"
+    assert record["groups"][0]["root_cause"] == "flaky fixture reuse"
+    assert record["run_id"] == "run-e2e"
+
+
+def test_e2e_resume_restores_outputs_for_harvest(settings, trace, tmp_path):
+    """Crash-before-harvest then --resume: the harvest step consumes the
+    executor-restored outputs map from progress.json, so a resumed run
+    still drops the record (invariant #2's restore path, exercised end to
+    end rather than with hand-built state)."""
+    settings.knowledge_intake_dir = str(tmp_path / "intake")
+    settings.repo_full_names = {"demo": "owner/demo"}
+    calls = []
+    executor = _executor_env(settings, trace, tmp_path, calls)
+    state = {"task_spec": {"kind": "pr_debug", "repo": "demo", "pr": 42}}
+    # Phase 1: the run completes debug+push, then "crashes" before harvest
+    # (the step simply is not in this playbook revision).
+    outcome = asyncio.run(executor.run(
+        _harvest_playbook(include_harvest=False), dict(state)))
+    assert outcome.status == "done"
+    assert calls == ["debug"]
+    assert not (tmp_path / "intake").exists()
+    # Phase 2: resume over the same run dir with FRESH state — completed
+    # steps short-circuit from the checkpoint and only harvest executes.
+    executor2 = _executor_env(settings, trace, tmp_path, calls)
+    outcome = asyncio.run(executor2.run(
+        _harvest_playbook(), dict(state)))
+    assert outcome.status == "done"
+    assert calls == ["debug"]  # the checkpoint replayed; no re-execution
+    drops = list((tmp_path / "intake").glob("*.json"))
+    assert len(drops) == 1
+    record = json.loads(drops[0].read_text(encoding="utf-8"))
+    assert record["groups"][0]["root_cause"] == "flaky fixture reuse"
+
+
+def test_playbook_wires_the_harvest_after_push(settings):
+    import yaml
+    from pathlib import Path
+
+    playbook = yaml.safe_load(
+        (Path(__file__).resolve().parents[1] / "playbooks" /
+         "pr-debug.yaml").read_text(encoding="utf-8")
+    )
+    ids = [step["id"] for step in playbook["steps"]]
+    assert ids.index("harvest") == ids.index("push") + 1
+    harvest = next(s for s in playbook["steps"] if s["id"] == "harvest")
+    assert harvest["step"] == "pr.harvest_debug_knowledge"
+    assert harvest["when"] == "not report_only"