diff --git a/docs/trajectory-distillation.md b/docs/trajectory-distillation.md
index 70110ec..a239081 100644
--- a/docs/trajectory-distillation.md
+++ b/docs/trajectory-distillation.md
@@ -184,8 +184,8 @@ mechanism 半的表达方式，按确定性从高到低：
 
 ## 9. 分阶段落地
 
-**P0 · 契约与手动通路**
-`eval_candidates/` 数据契约 + `okc distill --verify/--list/--accept/--reject`（ledger，确定性，零 API）。此阶段 candidate 由人手写——先让生命周期和校验跑通，蒸馏智能后置。产出即有用：手写新 held-out 任务从此有了标准校验与台账。
+**P0 · 契约与手动通路** ✅ 已落地（分支 `codex/distill-p0`，539→551 passed）
+`eval_candidates/` 数据契约 + `okc distill --verify/--list/--accept/--reject`（ledger，确定性，零 API）。此阶段 candidate 由人手写——先让生命周期和校验跑通，蒸馏智能后置。产出即有用：手写新 held-out 任务从此有了标准校验与台账。落地要点：`EvalTaskSpec` 增加可选 `source_sessions`（§3 的 lineage 随任务走）；fail-before 硬门与 runnable 检查复用 `eval.py` 的 verifier 执行路径；accept 只打印 promote 指引，代码路径不写 `tests/e2e/`。
 
 **P1 · LLM distiller**
 `okc distill --from-session <id>` / `--from-playbook <entry-id>`：spawn distiller subagent，digest → task.json + fixture + provenance。scripted-transport 外壳测试保证管线确定性可测，真实蒸馏按需花预算。
diff --git a/src/agent_harness/distill.py b/src/agent_harness/distill.py
new file mode 100644
index 0000000..b4d50a0
--- /dev/null
+++ b/src/agent_harness/distill.py
@@ -0,0 +1,245 @@
+"""Trajectory-distillation P0: candidate contract + manual pathway.
+
+Eval-task candidates live under `.agent-harness/eval_candidates/<candidate-id>/`
+as `task.json` (a strict EvalTaskSpec document) + `fixture/` + `provenance.json`.
+This module gives them a lifecycle:
+
+    okc distill                    list candidates + derived status
+    okc distill --verify <id>      deterministic anti-task checks (below)
+    okc distill --accept <id>      human decision; refuses unverified candidates
+    okc distill --reject <id>
+
+Verification is the fail-before hard gate from docs/trajectory-distillation.md
+section 5: the verifier must be runnable AND must FAIL on the untouched fixture
+(a verifier that passes an empty fixture tests nothing). Accept prints promote
+guidance only -- there is deliberately no code path here that writes to
+`tests/e2e/`; a human copying files under git is the sole promotion channel
+(section 6.1). Candidates in P0 are hand-written; the LLM distiller is P1.
+
+Append-only ledger folded to current status, like harness_edits_apply. Zero API,
+fully deterministic, deletable extension.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import shutil
+import tempfile
+import time
+from pathlib import Path
+from typing import Any, Sequence
+
+# _run_verifier/_copy_fixture are module-private in eval.py but stable; reusing
+# them beats growing a second verifier-execution dialect (same package).
+from agent_harness.eval import EvalTaskSpec, _copy_fixture, _run_verifier, load_eval_task
+from agent_harness.knowledge_port import utc_now
+from agent_harness.workspace import Workspace
+
+
+CANDIDATES_DIR = ".agent-harness/eval_candidates"
+LEDGER = ".agent-harness/eval_candidates/ledger.jsonl"
+
+
+class DistillError(Exception):
+    """Invalid candidate operations (unknown id, accept before verify, ...)."""
+
+
+def _candidate_dir(workspace: Workspace, candidate_id: str) -> Path:
+    path = workspace.resolve_path(f"{CANDIDATES_DIR}/{candidate_id}", require_exists=False)
+    if not path.is_dir() or not (path / "task.json").exists():
+        raise DistillError(f"unknown candidate: {candidate_id}")
+    return path
+
+
+def _append_ledger(workspace: Workspace, record: dict[str, Any]) -> None:
+    path = workspace.resolve_path(LEDGER, require_exists=False)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a", encoding="utf-8", newline="\n") as handle:
+        handle.write(json.dumps(record, ensure_ascii=False, sort_keys=True) + "\n")
+
+
+def _read_ledger(workspace: Workspace) -> list[dict[str, Any]]:
+    path = workspace.resolve_path(LEDGER, require_exists=False)
+    if not path.exists():
+        return []
+    records: list[dict[str, Any]] = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        if not line.strip():
+            continue
+        try:
+            records.append(json.loads(line))
+        except json.JSONDecodeError:
+            continue
+    return records
+
+
+def _fold_status(workspace: Workspace) -> dict[str, dict[str, Any]]:
+    """Per-candidate state from the append-only ledger: the latest accept/reject
+    decision wins; otherwise `verified` iff the latest verify passed."""
+    state: dict[str, dict[str, Any]] = {}
+    for record in _read_ledger(workspace):
+        candidate_id = str(record.get("candidate_id") or "")
+        action = record.get("action")
+        if not candidate_id or action not in {"verify", "accept", "reject"}:
+            continue
+        entry = state.setdefault(candidate_id, {"decision": None, "verified": False, "verify_report": None})
+        if action == "verify":
+            report = dict(record.get("verify_report") or {})
+            entry["verified"] = bool(report.get("passed"))
+            entry["verify_report"] = report
+        else:
+            entry["decision"] = "accepted" if action == "accept" else "rejected"
+    return state
+
+
+def candidate_status(workspace: Workspace, candidate_id: str) -> str:
+    entry = _fold_status(workspace).get(candidate_id)
+    if entry is None:
+        return "candidate"
+    if entry["decision"]:
+        return entry["decision"]
+    return "verified" if entry["verified"] else "candidate"
+
+
+def list_candidates(workspace: Workspace) -> list[dict[str, Any]]:
+    root = workspace.resolve_path(CANDIDATES_DIR, require_exists=False)
+    if not root.exists():
+        return []
+    state = _fold_status(workspace)
+    items: list[dict[str, Any]] = []
+    for path in sorted(root.iterdir()):
+        if not path.is_dir() or not (path / "task.json").exists():
+            continue
+        candidate_id = path.name
+        entry = state.get(candidate_id, {})
+        status = candidate_status(workspace, candidate_id)
+        items.append(
+            {
+                "candidate_id": candidate_id,
+                "status": status,
+                "verify_report": entry.get("verify_report"),
+            }
+        )
+    return items
+
+
+async def verify_candidate(workspace: Workspace, candidate_id: str, *, timeout: float = 120.0) -> dict[str, Any]:
+    """Deterministic anti-task checks. Appends a verify event; returns the report.
+
+    - load_ok: task.json parses as an EvalTaskSpec (fixture dir must exist).
+    - verifier_runnable: the argv starts and exits within the timeout.
+    - fail_before: the verifier FAILS on the untouched fixture (the hard gate
+      against verifiers that test nothing).
+    """
+    directory = _candidate_dir(workspace, candidate_id)
+    started = time.perf_counter()
+    report: dict[str, Any] = {
+        "load_ok": False,
+        "verifier_runnable": False,
+        "fail_before": False,
+        "passed": False,
+        "reason": "",
+    }
+    task: EvalTaskSpec | None = None
+    try:
+        task = load_eval_task(directory / "task.json")
+        report["load_ok"] = True
+    except (KeyError, TypeError, ValueError, OSError, json.JSONDecodeError) as exc:
+        report["reason"] = f"task.json failed to load: {exc}"
+
+    if task is not None:
+        scratch = Path(tempfile.mkdtemp(prefix=f"okc-distill-{candidate_id}-"))
+        try:
+            fixture_copy = scratch / "fixture"
+            _copy_fixture(task.fixture_path, fixture_copy)
+            result = await _run_verifier(task.verifier, cwd=fixture_copy, timeout=timeout)
+            report["verifier_runnable"] = not result.timed_out
+            report["fail_before"] = (not result.timed_out) and result.exit_code != 0
+            if result.timed_out:
+                report["reason"] = f"verifier timed out after {timeout}s"
+            elif result.exit_code == 0:
+                report["reason"] = "verifier passed on the untouched fixture; it tests nothing"
+        except (OSError, FileNotFoundError) as exc:
+            report["reason"] = f"verifier could not run: {exc}"
+        finally:
+            shutil.rmtree(scratch, ignore_errors=True)
+
+    report["passed"] = bool(report["load_ok"] and report["verifier_runnable"] and report["fail_before"])
+    report["duration_s"] = round(time.perf_counter() - started, 3)
+    _append_ledger(
+        workspace,
+        {"action": "verify", "candidate_id": candidate_id, "verify_report": report, "created_at": utc_now()},
+    )
+    return report
+
+
+def accept_candidate(workspace: Workspace, candidate_id: str, *, note: str = "") -> dict[str, Any]:
+    """Human decision. Refuses candidates whose latest verify did not pass.
+    Returns promote guidance -- this function never writes to tests/e2e/."""
+    _candidate_dir(workspace, candidate_id)
+    entry = _fold_status(workspace).get(candidate_id)
+    if entry is None or not entry["verified"]:
+        raise DistillError(f"candidate {candidate_id} has no passing verify; run --verify first")
+    _append_ledger(workspace, {"action": "accept", "candidate_id": candidate_id, "note": note, "created_at": utc_now()})
+    relative = f"{CANDIDATES_DIR}/{candidate_id}"
+    return {
+        "candidate_id": candidate_id,
+        "status": "accepted",
+        "promote": [
+            f"cp {relative}/task.json tests/e2e/tasks/{candidate_id}.json",
+            f"cp -R {relative}/fixture tests/e2e/fixtures/{candidate_id}",
+            "edit tests/e2e/tasks/<id>.json fixture_path to ../fixtures/<id>, add the task to a suite (e.g. tests/e2e/suites/held_out.json), and commit",
+        ],
+    }
+
+
+def reject_candidate(workspace: Workspace, candidate_id: str, *, note: str = "") -> dict[str, Any]:
+    _candidate_dir(workspace, candidate_id)
+    _append_ledger(workspace, {"action": "reject", "candidate_id": candidate_id, "note": note, "created_at": utc_now()})
+    return {"candidate_id": candidate_id, "status": "rejected", "note": note}
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Manage distilled eval-task candidates (verify / accept / reject; promotion stays manual)."
+    )
+    parser.add_argument("--workspace", default=".")
+    parser.add_argument("--list", action="store_true")
+    parser.add_argument("--verify", metavar="CANDIDATE_ID", default=None)
+    parser.add_argument("--accept", metavar="CANDIDATE_ID", default=None)
+    parser.add_argument("--reject", metavar="CANDIDATE_ID", default=None)
+    parser.add_argument("--note", default="")
+    parser.add_argument("--timeout", type=float, default=120.0)
+    args = parser.parse_args(list(argv) if argv is not None else None)
+
+    actions = [bool(args.verify), bool(args.accept), bool(args.reject)]
+    if sum(actions) > 1:
+        parser.error("use only one of --verify / --accept / --reject per invocation")
+
+    workspace = Workspace(Path(args.workspace))
+    try:
+        if args.verify:
+            report = asyncio.run(verify_candidate(workspace, args.verify, timeout=args.timeout))
+            payload: dict[str, Any] = {
+                "status": "ok",
+                "candidate_id": args.verify,
+                "verify_report": report,
+                "candidate_status": candidate_status(workspace, args.verify),
+            }
+        elif args.accept:
+            payload = {"status": "ok", **accept_candidate(workspace, args.accept, note=args.note)}
+        elif args.reject:
+            payload = {"status": "ok", **reject_candidate(workspace, args.reject, note=args.note)}
+        else:  # bare `okc distill` (or --list) lists candidates
+            payload = {"status": "ok", "candidates": list_candidates(workspace)}
+    except DistillError as exc:
+        print(json.dumps({"status": "error", "error_type": "DistillError", "message": str(exc)}, ensure_ascii=False))
+        return 1
+    print(json.dumps(payload, ensure_ascii=False, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/src/agent_harness/eval.py b/src/agent_harness/eval.py
index 9c8face..982a1a2 100644
--- a/src/agent_harness/eval.py
+++ b/src/agent_harness/eval.py
@@ -47,6 +47,11 @@ class EvalTaskSpec:
     tags: tuple[str, ...] = ()
     max_model_calls: int | None = None
     verifier_hint: str | None = None
+    # Lineage: the sessions this task was distilled from (empty for hand-written
+    # tasks, so they are never lineage-excluded). Travels with task.json so the
+    # held-out hygiene filter survives promotion into tests/e2e -- see
+    # docs/trajectory-distillation.md section 3.
+    source_sessions: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True)
@@ -471,6 +476,7 @@ def load_eval_task(path: str | Path) -> EvalTaskSpec:
         tags=tuple(str(item) for item in data.get("tags", [])),
         max_model_calls=int(data["max_model_calls"]) if data.get("max_model_calls") is not None else None,
         verifier_hint=str(data["verifier_hint"]) if data.get("verifier_hint") is not None else None,
+        source_sessions=tuple(str(item) for item in data.get("source_sessions", []) if str(item)),
     )
 
 
diff --git a/src/agent_harness/run.py b/src/agent_harness/run.py
index 12371e8..17eac5f 100644
--- a/src/agent_harness/run.py
+++ b/src/agent_harness/run.py
@@ -66,6 +66,10 @@ def main(argv: Sequence[str] | None = None) -> int:
         from agent_harness.harness_edits_apply import main as edits_main
 
         return edits_main(raw_argv[1:])
+    if raw_argv and raw_argv[0] == "distill":
+        from agent_harness.distill import main as distill_main
+
+        return distill_main(raw_argv[1:])
     parser = _parser()
     normalized_argv = _normalize_argv(raw_argv)
     args = parser.parse_args(normalized_argv)
diff --git a/tests/test_distill.py b/tests/test_distill.py
new file mode 100644
index 0000000..c527726
--- /dev/null
+++ b/tests/test_distill.py
@@ -0,0 +1,216 @@
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+
+from agent_harness.distill import (
+    DistillError,
+    accept_candidate,
+    candidate_status,
+    list_candidates,
+    reject_candidate,
+    verify_candidate,
+)
+from agent_harness.eval import load_eval_task
+from agent_harness.workspace import Workspace
+
+
+def _failing_verifier() -> list[str]:
+    # Passes only once the agent has created out.txt -> fails on the bare fixture.
+    return [sys.executable, "-c", "import pathlib,sys; sys.exit(0 if pathlib.Path('out.txt').exists() else 1)"]
+
+
+def _loose_verifier() -> list[str]:
+    return [sys.executable, "-c", "import sys; sys.exit(0)"]  # passes on anything -> tests nothing
+
+
+def _seed_candidate(
+    root: Path,
+    candidate_id: str,
+    *,
+    verifier_argv: list[str] | None = None,
+    source_sessions: list[str] | None = None,
+) -> Path:
+    directory = root / ".agent-harness" / "eval_candidates" / candidate_id
+    (directory / "fixture").mkdir(parents=True)
+    (directory / "fixture" / "seed.txt").write_text("seed\n", encoding="utf-8")
+    task = {
+        "id": candidate_id,
+        "name": candidate_id,
+        "fixture_path": "fixture",
+        "prompt": "Create out.txt containing done.",
+        "verifier": {"argv": verifier_argv or _failing_verifier()},
+    }
+    if source_sessions is not None:
+        task["source_sessions"] = source_sessions
+    (directory / "task.json").write_text(json.dumps(task), encoding="utf-8")
+    (directory / "provenance.json").write_text(
+        json.dumps({"producer": "human", "source_sessions": source_sessions or []}), encoding="utf-8"
+    )
+    return directory
+
+
+# --- contract -----------------------------------------------------------------
+
+
+def test_candidate_task_json_loads_with_lineage(tmp_path) -> None:
+    directory = _seed_candidate(tmp_path, "cand-1", source_sessions=["s-orig", "s-2"])
+    task = load_eval_task(directory / "task.json")
+    assert task.source_sessions == ("s-orig", "s-2")
+    assert task.fixture_path == (directory / "fixture").resolve()
+
+
+def test_existing_handwritten_task_loads_with_empty_lineage() -> None:
+    # Hand-written tasks have no source_sessions field -> empty, never excluded.
+    task = load_eval_task("tests/e2e/tasks/bugfix_simple.json")
+    assert task.source_sessions == ()
+
+
+# --- verify: the fail-before hard gate -----------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_verify_passes_for_a_real_candidate(tmp_path) -> None:
+    _seed_candidate(tmp_path, "good")
+    ws = Workspace(tmp_path)
+
+    report = await verify_candidate(ws, "good")
+
+    assert report["load_ok"] and report["verifier_runnable"] and report["fail_before"]
+    assert report["passed"] is True
+    assert candidate_status(ws, "good") == "verified"
+
+
+@pytest.mark.asyncio
+async def test_loose_verifier_is_rejected_by_fail_before(tmp_path) -> None:
+    _seed_candidate(tmp_path, "loose", verifier_argv=_loose_verifier())
+    ws = Workspace(tmp_path)
+
+    report = await verify_candidate(ws, "loose")
+
+    assert report["fail_before"] is False
+    assert report["passed"] is False
+    assert "tests nothing" in report["reason"]
+    assert candidate_status(ws, "loose") == "candidate"  # not verified
+    with pytest.raises(DistillError, match="no passing verify"):
+        accept_candidate(ws, "loose")
+
+
+@pytest.mark.asyncio
+async def test_missing_verifier_binary_is_not_runnable(tmp_path) -> None:
+    _seed_candidate(tmp_path, "broken", verifier_argv=["/nonexistent/verifier-binary"])
+    ws = Workspace(tmp_path)
+
+    report = await verify_candidate(ws, "broken")
+
+    assert report["verifier_runnable"] is False
+    assert report["passed"] is False
+    assert report["reason"]
+
+
+@pytest.mark.asyncio
+async def test_verify_unknown_candidate_errors(tmp_path) -> None:
+    with pytest.raises(DistillError, match="unknown candidate"):
+        await verify_candidate(Workspace(tmp_path), "ghost")
+
+
+# --- ledger state machine -------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_accept_after_verify_and_promote_guidance(tmp_path) -> None:
+    _seed_candidate(tmp_path, "good")
+    ws = Workspace(tmp_path)
+    await verify_candidate(ws, "good")
+
+    result = accept_candidate(ws, "good", note="covers blocked_retry_loop")
+
+    assert result["status"] == "accepted"
+    assert any("tests/e2e/tasks/good.json" in step for step in result["promote"])
+    assert candidate_status(ws, "good") == "accepted"
+    # promote guidance is text only: nothing was written into tests/e2e/
+    assert not (tmp_path / "tests").exists()
+
+
+@pytest.mark.asyncio
+async def test_reject_works_anytime_and_latest_decision_wins(tmp_path) -> None:
+    _seed_candidate(tmp_path, "cand")
+    ws = Workspace(tmp_path)
+
+    reject_candidate(ws, "cand", note="duplicate coverage")
+    assert candidate_status(ws, "cand") == "rejected"
+
+    # A later verify + accept overturns the rejection (latest decision wins).
+    await verify_candidate(ws, "cand")
+    accept_candidate(ws, "cand")
+    assert candidate_status(ws, "cand") == "accepted"
+
+
+@pytest.mark.asyncio
+async def test_ledger_is_append_only(tmp_path) -> None:
+    _seed_candidate(tmp_path, "cand")
+    ws = Workspace(tmp_path)
+    await verify_candidate(ws, "cand")
+    await verify_candidate(ws, "cand")  # re-verify appends, never rewrites
+    accept_candidate(ws, "cand")
+
+    ledger = tmp_path / ".agent-harness" / "eval_candidates" / "ledger.jsonl"
+    lines = [json.loads(line) for line in ledger.read_text(encoding="utf-8").splitlines()]
+    assert [record["action"] for record in lines] == ["verify", "verify", "accept"]
+
+
+@pytest.mark.asyncio
+async def test_failed_reverify_demotes_until_fixed(tmp_path) -> None:
+    directory = _seed_candidate(tmp_path, "cand")
+    ws = Workspace(tmp_path)
+    await verify_candidate(ws, "cand")
+    assert candidate_status(ws, "cand") == "verified"
+
+    # Fixture edited so the verifier now passes untouched -> re-verify fails.
+    (directory / "fixture" / "out.txt").write_text("oops\n", encoding="utf-8")
+    report = await verify_candidate(ws, "cand")
+    assert report["passed"] is False
+    assert candidate_status(ws, "cand") == "candidate"
+    with pytest.raises(DistillError):
+        accept_candidate(ws, "cand")
+
+
+# --- listing + CLI ----------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_list_candidates_joins_status(tmp_path) -> None:
+    _seed_candidate(tmp_path, "a-good")
+    _seed_candidate(tmp_path, "b-loose", verifier_argv=_loose_verifier())
+    ws = Workspace(tmp_path)
+    await verify_candidate(ws, "a-good")
+
+    items = list_candidates(ws)
+
+    assert [(item["candidate_id"], item["status"]) for item in items] == [
+        ("a-good", "verified"),
+        ("b-loose", "candidate"),
+    ]
+
+
+def test_cli_full_lifecycle_via_dispatch(tmp_path, capsys) -> None:
+    from agent_harness.run import main as run_main
+
+    _seed_candidate(tmp_path, "cand")
+
+    rc = run_main(["distill", "--accept", "cand", "--workspace", str(tmp_path)])
+    assert rc == 1  # accept before verify -> structured error
+    assert json.loads(capsys.readouterr().out)["error_type"] == "DistillError"
+
+    assert run_main(["distill", "--verify", "cand", "--workspace", str(tmp_path)]) == 0
+    assert json.loads(capsys.readouterr().out)["candidate_status"] == "verified"
+
+    assert run_main(["distill", "--accept", "cand", "--workspace", str(tmp_path)]) == 0
+    capsys.readouterr()
+
+    assert run_main(["distill", "--workspace", str(tmp_path)]) == 0  # bare = list
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["candidates"][0]["status"] == "accepted"