#!/usr/bin/env python3 """Validate a prepared Meta3 run folder. No network calls.""" from __future__ import annotations import argparse import json from datetime import datetime, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] def load_json(path: Path) -> dict[str, Any]: with path.open("r", encoding="utf-8") as handle: data = json.load(handle) if not isinstance(data, dict): raise ValueError(f"Expected JSON object: {path}") return data def load_jsonl(path: Path) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] with path.open("r", encoding="utf-8") as handle: for line_no, line in enumerate(handle, start=1): if not line.strip(): continue data = json.loads(line) if not isinstance(data, dict): raise ValueError(f"Expected JSON object in {path}:{line_no}") rows.append(data) return rows def write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def latest_run_dir() -> Path: runs_root = ROOT / "runs" candidates = sorted(p for p in runs_root.iterdir() if p.is_dir()) if not candidates: raise FileNotFoundError("No run dirs found. Run scripts/meta3_prepare.py first.") return candidates[-1] def run_dir_for(run_id: str | None) -> Path: return ROOT / "runs" / run_id if run_id else latest_run_dir() def check_file(path: Path, errors: list[str], label: str) -> bool: if not path.is_file(): errors.append(f"missing {label}: {path}") return False return True def validate_workpacket_mode( *, run_dir: Path, run_id: str, mode: str, errors: list[str], warnings: list[str], expected_digest_path: str | None, ) -> dict[str, Any] | None: manifest_path = run_dir / f"outputs/workpackets/{mode}_packet_manifest.json" if not check_file(manifest_path, errors, f"{mode} packet manifest"): return None manifest = load_json(manifest_path) if manifest.get("schema") != "meta3.workpacket_manifest.v1": errors.append(f"{mode} packet manifest schema invalid") if manifest.get("run_id") != run_id: errors.append(f"{mode} packet manifest run_id mismatch") if manifest.get("network_calls_made") != 0: errors.append(f"{mode} packet manifest records network calls") if expected_digest_path and manifest.get("source_digest_path") != expected_digest_path: errors.append( f"{mode} packet manifest source mismatch: {manifest.get('source_digest_path')} != {expected_digest_path}" ) if manifest.get("source_drift_allowed") is True: errors.append(f"{mode} packet manifest used --allow-source-drift") packet_path = Path(str(manifest.get("packet_path") or "")) ledger_path = Path(str(manifest.get("call_ledger_path") or "")) if not check_file(packet_path, errors, f"{mode} packet file"): return manifest if not check_file(ledger_path, errors, f"{mode} call ledger"): return manifest packets = load_jsonl(packet_path) ledger = load_jsonl(ledger_path) if len(packets) != int(manifest.get("packet_count") or -1): errors.append(f"{mode} packet_count mismatch: manifest={manifest.get('packet_count')} actual={len(packets)}") if len(ledger) != int(manifest.get("planned_call_count") or -1): errors.append( f"{mode} planned_call_count mismatch: manifest={manifest.get('planned_call_count')} actual={len(ledger)}" ) packet_ids = [str(packet.get("packet_id") or "") for packet in packets] if len(packet_ids) != len(set(packet_ids)): errors.append(f"{mode} duplicate packet ids") ledger_packet_ids = {str(row.get("packet_id") or "") for row in ledger} if set(packet_ids) != ledger_packet_ids: errors.append(f"{mode} packet ids do not match call ledger") required_response_fields = { "question_id", "mode", "canonical_thesis", "claims", "counterevidence_status", "uncertainty", "owner_next_step", } for packet in packets: if packet.get("schema") != "meta3.workpacket.v1": errors.append(f"{mode} packet schema invalid: {packet.get('packet_id')}") if packet.get("run_id") != run_id: errors.append(f"{mode} packet run_id mismatch: {packet.get('packet_id')}") if packet.get("mode") != mode: errors.append(f"{mode} packet mode mismatch: {packet.get('packet_id')}") if not packet.get("prompt"): errors.append(f"{mode} packet missing prompt: {packet.get('packet_id')}") contract = packet.get("expected_output_contract", {}) fields = set(contract.get("required_top_level_fields", [])) if isinstance(contract, dict) else set() missing = required_response_fields - fields if missing: errors.append(f"{mode} packet contract missing fields {sorted(missing)}: {packet.get('packet_id')}") for row in ledger: if row.get("status") != "planned": warnings.append(f"{mode} call ledger has non-planned status: {row.get('packet_id')} -> {row.get('status')}") if row.get("network_calls_made") != 0: errors.append(f"{mode} call ledger records network calls: {row.get('packet_id')}") if mode == "claim_ledger": available = int(manifest.get("available_digest_rows") or 0) unique_refs = int(manifest.get("unique_digest_refs_in_packets") or 0) if available and unique_refs < available: errors.append(f"claim_ledger does not cover all available digests: {unique_refs}/{available}") source_mentions = manifest.get("source_family_mentions", {}) if isinstance(source_mentions, dict) and not source_mentions: warnings.append(f"{mode} has no source family mentions") return manifest def markdown_report(report: dict[str, Any]) -> str: lines = [ "# Meta3 Run Readiness Report", "", f"Run: `{report['run_id']}`", f"Created: `{report['created_at']}`", f"Status: `{report['status']}`", "", "## Counts", "", f"- Errors: {len(report['errors'])}", f"- Warnings: {len(report['warnings'])}", ] for mode, summary in sorted(report["workpacket_modes"].items()): lines.extend( [ f"- {mode}: {summary.get('packet_count')} packets, {summary.get('planned_call_count')} planned calls, " f"{summary.get('unique_digest_refs_in_packets')} unique digest refs", ] ) if report["errors"]: lines.extend(["", "## Errors", ""]) lines.extend(f"- {error}" for error in report["errors"]) if report["warnings"]: lines.extend(["", "## Warnings", ""]) lines.extend(f"- {warning}" for warning in report["warnings"]) lines.extend(["", "## Network", "", "- network_calls_made: 0"]) return "\n".join(lines) + "\n" def main() -> int: parser = argparse.ArgumentParser(description="Validate Meta3 run artifacts without API calls.") parser.add_argument("--run-id", default=None, help="Run id under runs/. Defaults to latest run.") parser.add_argument( "--mode", action="append", choices=["calibration", "claim_ledger"], help="Workpacket mode to require. Can be repeated. Defaults to both.", ) args = parser.parse_args() run_dir = run_dir_for(args.run_id) run_id = run_dir.name required_modes = args.mode or ["calibration", "claim_ledger"] errors: list[str] = [] warnings: list[str] = [] manifest_path = run_dir / "manifests/run_manifest.json" batch_plan_path = run_dir / "manifests/batch_plan.json" baseline_path = run_dir / "manifests/baseline_snapshot.json" for path, label in [ (manifest_path, "run manifest"), (batch_plan_path, "batch plan"), (baseline_path, "baseline snapshot"), ]: check_file(path, errors, label) manifest = load_json(manifest_path) if manifest_path.exists() else {} batch_plan = load_json(batch_plan_path) if batch_plan_path.exists() else {} baseline = load_json(baseline_path) if baseline_path.exists() else {} if manifest.get("schema") != "meta3.run_manifest.v1": errors.append("run manifest schema invalid") if batch_plan.get("schema") != "meta3.batch_plan.v1": errors.append("batch plan schema invalid") if baseline.get("schema") != "meta3.baseline_snapshot.v1": errors.append("baseline snapshot schema invalid") if manifest.get("network_calls_made") != 0: errors.append("run manifest records network calls") if batch_plan.get("network_calls_made") != 0: errors.append("batch plan records network calls") if baseline.get("network_calls_made") != 0: errors.append("baseline snapshot records network calls") source_snapshot = baseline.get("source_snapshot", {}) if isinstance(baseline, dict) else {} audit_unique = int(source_snapshot.get("layer1_audit_unique_digest_paths") or 0) raw_rows = int(source_snapshot.get("layer1_raw_digest_file_rows_counted") or 0) source_snapshot_manifest_path = run_dir / "manifests/source_snapshot_manifest.json" source_snapshot_manifest = load_json(source_snapshot_manifest_path) if source_snapshot_manifest_path.exists() else None expected_digest_path = None if source_snapshot_manifest: if source_snapshot_manifest.get("schema") != "meta3.source_snapshot_manifest.v1": errors.append("source snapshot manifest schema invalid") frozen_path = Path(str(source_snapshot_manifest.get("frozen_digest_path") or "")) if not frozen_path.is_file(): errors.append(f"missing frozen digest file: {frozen_path}") else: expected_digest_path = str(frozen_path) frozen_rows = len(load_jsonl(frozen_path)) summary = source_snapshot_manifest.get("summary", {}) manifest_frozen_rows = int(summary.get("frozen_rows") or -1) if isinstance(summary, dict) else -1 if frozen_rows != manifest_frozen_rows: errors.append(f"frozen digest row mismatch: manifest={manifest_frozen_rows} actual={frozen_rows}") if source_snapshot_manifest.get("network_calls_made") != 0: errors.append("source snapshot manifest records network calls") if audit_unique != raw_rows and not source_snapshot_manifest: errors.append(f"layer1 audit/raw digest mismatch: audit_unique={audit_unique} raw_rows={raw_rows}") workpacket_modes: dict[str, Any] = {} for mode in required_modes: mode_manifest = validate_workpacket_mode( run_dir=run_dir, run_id=run_id, mode=mode, errors=errors, warnings=warnings, expected_digest_path=expected_digest_path, ) if mode_manifest: workpacket_modes[mode] = { "packet_count": mode_manifest.get("packet_count"), "planned_call_count": mode_manifest.get("planned_call_count"), "unique_digest_refs_in_packets": mode_manifest.get("unique_digest_refs_in_packets"), "source_family_mentions": mode_manifest.get("source_family_mentions", {}), } report = { "schema": "meta3.run_readiness_report.v1", "run_id": run_id, "created_at": datetime.now(timezone.utc).isoformat(), "status": "pass" if not errors else "fail", "required_modes": required_modes, "workpacket_modes": workpacket_modes, "errors": errors, "warnings": warnings, "network_calls_made": 0, } report_path = run_dir / "reports/run_readiness_report.json" report_md_path = run_dir / "reports/RUN_READINESS_REPORT.md" write_json(report_path, report) report_md_path.write_text(markdown_report(report), encoding="utf-8") print(f"readiness_report={report_path}") print(f"readiness_report_md={report_md_path}") print(f"status={report['status']}") print(f"errors={len(errors)}") print(f"warnings={len(warnings)}") print("network_calls_made=0") return 0 if not errors else 1 if __name__ == "__main__": raise SystemExit(main())