Distillation P0: candidate contract + verify/accept ledger (okc distill)

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 外壳测试保证管线确定性可测，真实蒸馏按需花预算。

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())

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)),
     )
 
 

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)