Add CERC plan validation

Author: Your Name

causetrace/cli.py

@@ -25,6 +25,7 @@
     compile_subsets,
     ingest_feedback,
     plan_experiments,
+    validate_experiment_plan,
     reprioritize_experiments,
     update_gaps,
 )
@@ -365,6 +366,10 @@ def cli(argv: list[str] | None = None) -> None:
     p_cr_reprioritize.add_argument("input", help="Feedback report JSON path")
     p_cr_reprioritize.add_argument("--output-dir", help="Output directory (default: docs/research/dataset_design/feedback)")
     p_cr_reprioritize.add_argument("--json", action="store_true", help="Print reprioritized plan as JSON")
+    p_cr_validate = p_cr_sub.add_parser("validate-plan", help="Validate a CERC experiment plan without executing it")
+    p_cr_validate.add_argument("plan_dir", help="Experiment plan directory")
+    p_cr_validate.add_argument("--output-dir", help="Output directory (default: docs/research/dataset_design/plan_validation)")
+    p_cr_validate.add_argument("--json", action="store_true", help="Print full validation report as JSON")
 
     p_cmp = sub.add_parser("compare", help="Compare two sessions side by side")
     p_cmp.add_argument("session_a", help="First session ID")
@@ -1876,6 +1881,27 @@ def _handle_corpus(store, args) -> None:
         print(f"  Top priority: {report['priorities'][0]['subset_id'] if report['priorities'] else 'none'}")
         return
 
+    if args.corpus_command == "validate-plan":
+        report = validate_experiment_plan(
+            store,
+            plan_dir=args.plan_dir,
+            output_dir=args.output_dir,
+            write=True,
+        )
+        if args.json:
+            json.dump(report, sys.stdout, indent=2)
+            print()
+            return
+        print(f"Plan validation: {report['output_dir']}")
+        print(f"  Plan dir: {report['plan_dir']}")
+        print(f"  Target subset: {report['target_subset']}")
+        print(f"  Required sessions: {report['required_sessions']}")
+        print(f"  Missing sessions: {report['necessity']['missing_sessions']}")
+        print(f"  Duplicate plans: {len(report['duplicate_plans'])}")
+        print(f"  Validation ok: {report['validation']['ok']}")
+        print(f"  Status: {report['validation']['status']}")
+        return
+
     if args.corpus_command == "verify":
         result = verify_snapshot(args.snapshot_dir)
         print(f"Snapshot: {result['snapshot_dir']}")

causetrace/crdd/__init__.py

@@ -8,6 +8,7 @@
 from .feedback import ingest_feedback, reprioritize_experiments, update_gaps
 from .experiment_planner import plan_experiments
 from .gap_analyzer import analyze_gaps
+from .plan_validation import validate_experiment_plan
 from .subset_builder import build_subset, compile_subsets
 from .subset_registry import SUBSET_DEFINITIONS, get_subset_definition
 
@@ -20,6 +21,7 @@
     "ingest_feedback",
     "get_subset_definition",
     "plan_experiments",
+    "validate_experiment_plan",
     "reprioritize_experiments",
     "update_gaps",
 ]

causetrace/crdd/plan_validation.py

@@ -0,0 +1,157 @@
+"""CERC plan validation helpers.
+
+This layer validates experiment plans without executing them. It checks for
+queue integrity, duplicate plan signatures, and whether the requested sampling
+is still needed.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+from causetrace.core import JSONStore
+
+from .constraints import validate_execution_queue
+from .experiment_planner import DEFAULT_PLAN_OUTPUT_DIR
+from .gap_analyzer import analyze_gaps
+from .subset_registry import SUBSET_DEFINITIONS
+
+
+DEFAULT_PLAN_VALIDATION_OUTPUT_DIR = Path.home() / ".causetrace" / "plan_validation"
+
+
+def _load_json(path: Path) -> dict[str, Any]:
+    data = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError(f"{path.name} must contain a JSON object")
+    return data
+
+
+def _canonicalize_queue(queue: dict[str, Any]) -> dict[str, Any]:
+    def _clean(value: Any) -> Any:
+        if isinstance(value, dict):
+            cleaned: dict[str, Any] = {}
+            for key, item in value.items():
+                if key in {"experiment_id", "generated_at", "output_dir", "queue_hash", "validation"}:
+                    continue
+                cleaned[key] = _clean(item)
+            return cleaned
+        if isinstance(value, list):
+            return [_clean(item) for item in value]
+        return value
+
+    return _clean(queue)
+
+
+def _queue_signature(queue: dict[str, Any]) -> str:
+    canonical = _canonicalize_queue(queue)
+    encoded = json.dumps(canonical, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
+
+
+def _scan_duplicate_plans(plan_root: Path, signature: str, current_plan_dir: Path) -> list[str]:
+    duplicates: list[str] = []
+    if not plan_root.exists():
+        return duplicates
+    for queue_path in plan_root.rglob("experiment_queue.json"):
+        if queue_path.parent == current_plan_dir:
+            continue
+        try:
+            queue = _load_json(queue_path)
+        except Exception:
+            continue
+        if _queue_signature(queue) == signature:
+            duplicates.append(str(queue_path.parent))
+    return sorted(duplicates)
+
+
+def validate_experiment_plan(
+    store: JSONStore,
+    *,
+    plan_dir: str | Path,
+    output_dir: str | Path | None = None,
+    write: bool = True,
+) -> dict[str, Any]:
+    """Validate an experiment plan without executing it."""
+    plan_path = Path(plan_dir)
+    queue_path = plan_path / "experiment_queue.json"
+    gap_path = plan_path / "gap_report.json"
+    if not queue_path.exists():
+        raise FileNotFoundError(f"missing plan queue: {queue_path}")
+
+    queue = _load_json(queue_path)
+    gap_report = _load_json(gap_path) if gap_path.exists() else {}
+    target_subset = str(queue.get("target_subset") or gap_report.get("target_subset") or "unknown")
+    if target_subset in SUBSET_DEFINITIONS:
+        current_gap = analyze_gaps(store, subset_ids=[target_subset])["subset_gaps"][0]
+    else:
+        current_gap = None
+
+    constraint_check = validate_execution_queue(queue)
+    signature = _queue_signature(queue)
+    plan_root = plan_path.parent if plan_path.parent != plan_path else DEFAULT_PLAN_OUTPUT_DIR
+    duplicate_plans = _scan_duplicate_plans(plan_root, signature, plan_path)
+    required_sessions = int(queue.get("required_sessions", 0) or 0)
+    missing_sessions = int((current_gap or {}).get("missing_sessions", required_sessions))
+    needed = missing_sessions > 0 and required_sessions > 0
+    valid = constraint_check["ok"] and not duplicate_plans and needed
+
+    report: dict[str, Any] = {
+        "schema": "causetrace.cerc.plan_validation.v0.1",
+        "generated_at": datetime.now().isoformat(),
+        "plan_dir": str(plan_path),
+        "target_subset": target_subset,
+        "required_sessions": required_sessions,
+        "current_gap": current_gap,
+        "gap_report": gap_report,
+        "queue_signature": signature,
+        "duplicate_plans": duplicate_plans,
+        "constraint_check": constraint_check,
+        "necessity": {
+            "missing_sessions": missing_sessions,
+            "sampling_needed": needed,
+        },
+        "validation": {
+            "ok": valid,
+            "status": "ready" if valid else ("duplicate" if duplicate_plans else "not_needed"),
+        },
+        "constraints": {
+            "external_only": True,
+            "no_execution": True,
+            "no_evidence_inflation": True,
+            "no_phase4_grade_promotion": True,
+        },
+    }
+
+    if write:
+        root = Path(output_dir) if output_dir else DEFAULT_PLAN_VALIDATION_OUTPUT_DIR
+        run_dir = root / plan_path.name
+        run_dir.mkdir(parents=True, exist_ok=True)
+        (run_dir / "plan_validation.json").write_text(json.dumps(report, indent=2, sort_keys=True), encoding="utf-8")
+        (run_dir / "plan_validation.md").write_text(
+            "\n".join([
+                f"# Plan validation: {plan_path.name}",
+                "",
+                f"- target subset: `{target_subset}`",
+                f"- required sessions: `{required_sessions}`",
+                f"- missing sessions: `{missing_sessions}`",
+                f"- sampling needed: `{needed}`",
+                f"- duplicate plans: `{len(duplicate_plans)}`",
+                f"- validation ok: `{valid}`",
+                f"- queue signature: `{signature}`",
+                "",
+                "## Safety Boundary",
+                "",
+                "Plan validation is read-only. It does not execute runtimes, alter evidence grades, or emit commands.",
+            ])
+            + "\n",
+            encoding="utf-8",
+        )
+        report["output_dir"] = str(run_dir)
+    else:
+        report["output_dir"] = None
+
+    return report

docs/research/README.md

@@ -69,6 +69,7 @@ Phase 4 must not enter:
 - [Subset Manifest Template](dataset_design/subset_manifest_template.md): required structure for comparable and experimental subsets used in Phase 4 candidate revalidation.
 - [Causal Experiment Requirement Compiler v0.3](dataset_design/cerc_v0.3.md): experiment planning layer that turns observed subset gaps into external-only execution queues. CERC plans do not execute agents, inflate evidence, or upgrade Phase 4 grades.
 - [CERC Feedback Integration v0.4](dataset_design/feedback_v0.4.md): read-only feedback layer that ingests external execution results, updates gap projections, and reprioritizes future experiments without changing runtime authority.
+- [CERC Plan Validation v0.3.1](dataset_design/plan_validation_v0.3.1.md): read-only checker for duplicate plans, queue constraints, and whether a planned sampling gap still needs collection.
 
 ## Roadmap And Future Directions
 

docs/research/dataset_design/plan_validation_v0.3.1.md

@@ -0,0 +1,39 @@
+# CERC Plan Validation v0.3.1
+
+Plan validation checks whether a CERC experiment plan is still necessary and
+whether it remains safe under the external-only boundary. It is read-only and
+does not change corpus data or evidence grades.
+
+## Definition
+
+```text
+Input:
+  plan directory with experiment_queue.json
+  optional gap_report.json
+
+Process:
+  validate queue constraints
+  compute canonical queue signature
+  detect duplicate plans
+  compare requested sampling against current gaps
+
+Output:
+  validation report and markdown summary
+```
+
+## Commands
+
+```bash
+causetrace corpus validate-plan docs/research/dataset_design/plans/<experiment_id>
+```
+
+## Safety Boundary
+
+Plan validation never executes runtimes, never emits commands, and never
+upgrades Phase 4 evidence. It only decides whether a proposed plan is ready,
+duplicated, or no longer needed.
+
+## Relationship To CERC
+
+CERC plans missing work. Validation decides whether that work is still needed
+and whether the plan already exists elsewhere in the corpus.

tests/test_metadata_corpus_report.py

@@ -1188,3 +1188,72 @@ def test_cerc_feedback_cli_commands(monkeypatch, tmp_path):
     )
     assert reprioritize_cmd.returncode == 0
     assert "Top priority:" in reprioritize_cmd.stdout
+
+
+def test_cerc_plan_validation_detects_duplicates(monkeypatch, tmp_path):
+    import causetrace.metadata as metadata
+    from causetrace.crdd import plan_experiments, validate_experiment_plan
+
+    monkeypatch.setattr(metadata, "METADATA_DIR", str(tmp_path / "metadata"))
+    store = JSONStore(store_dir=str(tmp_path / "data"))
+    _write_session(store, "s1")
+    merge_metadata("s1", {"runtime": "codex", "task_type": "bug_fix", "task_source": "real_work", "success": False})
+
+    plan_a = plan_experiments(
+        store,
+        target_subset="failure_enriched",
+        required_sessions=5,
+        name="exp_plan_a",
+        output_dir=tmp_path / "plans",
+    )
+    plan_b = plan_experiments(
+        store,
+        target_subset="failure_enriched",
+        required_sessions=5,
+        name="exp_plan_b",
+        output_dir=tmp_path / "plans",
+    )
+
+    report = validate_experiment_plan(store, plan_dir=Path(plan_b["output_dir"]), output_dir=tmp_path / "plan-validation")
+    assert report["constraints"]["external_only"] is True
+    assert report["validation"]["status"] == "duplicate"
+    assert report["duplicate_plans"]
+    assert Path(report["output_dir"]).joinpath("plan_validation.json").exists()
+    assert Path(report["output_dir"]).joinpath("plan_validation.md").exists()
+
+
+def test_cerc_plan_validation_cli(monkeypatch, tmp_path):
+    import causetrace.metadata as metadata
+    from causetrace.crdd import plan_experiments
+
+    monkeypatch.setattr(metadata, "METADATA_DIR", str(tmp_path / "metadata"))
+    store = JSONStore(store_dir=str(tmp_path / "data"))
+    _write_session(store, "s1")
+    merge_metadata("s1", {"runtime": "codex", "task_type": "bug_fix", "task_source": "real_work", "success": False})
+
+    plan_result = plan_experiments(
+        store,
+        target_subset="failure_enriched",
+        required_sessions=5,
+        name="exp_plan_cli",
+        output_dir=tmp_path / "plans",
+    )
+
+    cmd = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "causetrace",
+            "corpus",
+            "validate-plan",
+            str(Path(plan_result["output_dir"])),
+            "--output-dir",
+            str(tmp_path / "plan-validation"),
+        ],
+        capture_output=True,
+        text=True,
+        env={**os.environ, "HOME": str(tmp_path)},
+    )
+    assert cmd.returncode == 0
+    assert "Validation ok:" in cmd.stdout
+    assert "Status:" in cmd.stdout