#!/usr/bin/env python3 """Create a local API-call estimate and batch plan for a prepared Meta3 run.""" from __future__ import annotations import argparse import json import math 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 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 load_manifest(run_id: str | None) -> tuple[Path, dict[str, Any]]: run_dir = ROOT / "runs" / run_id if run_id else latest_run_dir() manifest_path = run_dir / "manifests/run_manifest.json" return run_dir, load_json(manifest_path) def count_jsonl(path: Path) -> int: if not path.exists(): return 0 count = 0 with path.open("r", encoding="utf-8") as handle: for _ in handle: count += 1 return count def load_process_phases(manifest: dict[str, Any]) -> list[dict[str, str]]: process_path = Path(manifest.get("process_model_path", "")) if not process_path.exists(): return [] process_doc = load_json(process_path) phases = process_doc.get("phases", []) if not isinstance(phases, list): return [] planned = [] for phase in phases: if not isinstance(phase, dict): continue planned.append( { "id": str(phase.get("id") or ""), "kind": str(phase.get("kind") or ""), "purpose": str(phase.get("purpose") or ""), "exit_gate": str(phase.get("exit_gate") or ""), } ) return planned def build_process_phase_plan( process_phases: list[dict[str, str]], *, inventory_total: int, missing_layer1: int, layer1_calls: int, answer_calls: int, synthesis_calls: int, coverage_pct: float, min_coverage: float, ) -> list[dict[str, Any]]: phase_plan: list[dict[str, Any]] = [] for phase in process_phases: phase_id = phase["id"] planned: dict[str, Any] = { "id": phase_id, "kind": phase["kind"], "status": "planned", "purpose": phase["purpose"], "exit_gate": phase["exit_gate"], } if phase_id == "p00_requirements_lock": planned["status"] = "open" planned["blocking_reason"] = "requirements/PENDING.md has not been locked." elif phase_id == "p01_baseline_audit": planned["status"] = "complete" if inventory_total else "blocked" planned["inventory_total"] = inventory_total planned["coverage_pct"] = coverage_pct elif phase_id == "p02_calibration_slice": planned["status"] = "blocked_until_requirements_lock" planned["estimated_api_calls"] = 0 elif phase_id == "p03_stratified_backfill": planned["status"] = "planned" planned["remaining_sessions"] = missing_layer1 planned["estimated_api_calls"] = layer1_calls planned["recommended_command"] = ( "cd /home/smlflg/Projekte/Meta2.0 && " "python3 scripts/meta2_layer1_digest.py --limit 120 --batch-size 3 " "--concurrency 2 --order source-balanced" ) elif phase_id == "p04_claim_ledger_generation": planned["estimated_api_calls"] = answer_calls if coverage_pct < min_coverage: planned["status"] = "blocked_until_requirements_calibration_and_coverage" else: planned["status"] = "blocked_until_requirements_and_calibration" elif phase_id == "p05_counterevidence_pass": planned["status"] = "blocked_until_claim_ledger" planned["estimated_api_calls"] = 0 elif phase_id == "p06_validity_review": planned["status"] = "blocked_until_claim_and_counterevidence_ledgers" planned["estimated_api_calls"] = 0 elif phase_id == "p07_owner_synthesis": planned["status"] = "blocked_until_validity_review" planned["estimated_api_calls"] = synthesis_calls elif phase_id == "p08_morning_closeout": planned["status"] = "planned_after_run_end" planned["estimated_api_calls"] = 0 phase_plan.append(planned) return phase_plan def main() -> int: parser = argparse.ArgumentParser(description="Plan Meta3 overnight batches without API calls.") parser.add_argument("--run-id", default=None, help="Run id under runs/. Defaults to latest run.") args = parser.parse_args() run_dir, manifest = load_manifest(args.run_id) profile = manifest["profile"] planning = profile.get("planning", {}) limits = profile.get("limits", {}) sources = profile.get("source_artifacts", {}) audit_path = Path(sources.get("layer1_audit", "")) audit = load_json(audit_path) if audit_path.exists() else {} inventory_total = int(audit.get("inventory_total") or 0) processed = int(audit.get("checkpoint_processed") or 0) coverage_pct = float(audit.get("coverage_pct") or 0.0) audit_digest_rows = int(audit.get("digest_rows") or 0) audit_unique_digest_paths = int(audit.get("unique_digest_paths") or 0) missing_layer1 = max(0, inventory_total - processed) digest_path = Path(sources.get("layer1_digests", "")) raw_digest_rows = count_jsonl(digest_path) batch_size = int(limits.get("safe_batch_size") or 1) concurrency = int(limits.get("safe_concurrency") or 1) question_count = int(manifest.get("question_count") or 0) answer_passes = int(planning.get("answer_passes_per_question") or 1) synthesis_passes = int(planning.get("synthesis_passes") or 0) layer1_calls = math.ceil(missing_layer1 / batch_size) if planning.get("layer1_backfill_enabled", True) else 0 answer_calls = question_count * answer_passes synthesis_calls = synthesis_passes estimated_total_calls = layer1_calls + answer_calls + synthesis_calls max_calls = int(limits.get("max_api_calls") or 0) min_coverage = float(planning.get("minimum_layer1_coverage_pct_before_final_answers") or 0) process_phases = load_process_phases(manifest) process_phase_plan = build_process_phase_plan( process_phases, inventory_total=inventory_total, missing_layer1=missing_layer1, layer1_calls=layer1_calls, answer_calls=answer_calls, synthesis_calls=synthesis_calls, coverage_pct=coverage_pct, min_coverage=min_coverage, ) gates = [] if raw_digest_rows != audit_unique_digest_paths: gates.append( { "gate": "layer1_audit_raw_digest_mismatch", "status": "blocked", "audit_unique_digest_paths": audit_unique_digest_paths, "raw_digest_file_rows": raw_digest_rows, "meaning": "Meta2.0 digest file and audit disagree; do not build final workpackets until Meta2 audit is refreshed.", } ) if coverage_pct < min_coverage: gates.append( { "gate": "layer1_coverage_below_final_answer_threshold", "status": "open", "coverage_pct": coverage_pct, "required_pct": min_coverage, "meaning": "Final answers should wait; layer1 backfill can still run.", } ) if limits.get("dry_run_required", True): gates.append( { "gate": "dry_run_required_before_network_calls", "status": "open", "meaning": "This scaffold did not run a network dry-run.", } ) if limits.get("network_calls_enabled_by_default") is not True: gates.append( { "gate": "network_calls_disabled_by_default", "status": "open", "meaning": "A future runner must require an explicit start flag.", } ) if max_calls and estimated_total_calls > max_calls: gates.append( { "gate": "estimated_calls_exceed_cap", "status": "blocked", "estimated_total_calls": estimated_total_calls, "max_api_calls": max_calls, } ) source_snapshot = { "inventory_total": inventory_total, "layer1_processed": processed, "layer1_coverage_pct": coverage_pct, "layer1_audit_digest_rows": audit_digest_rows, "layer1_audit_unique_digest_paths": audit_unique_digest_paths, "layer1_raw_digest_file_rows_counted": raw_digest_rows, "unresolved_failures": int(audit.get("unresolved_failures") or 0), "raw_failure_rows": int(audit.get("raw_failure_rows") or 0), "by_source": audit.get("by_source", {}), "confidence": audit.get("confidence", {}), "strategic_relevance": audit.get("strategic_relevance", {}), } baseline_snapshot = { "schema": "meta3.baseline_snapshot.v1", "run_id": manifest["run_id"], "created_at": datetime.now(timezone.utc).isoformat(), "baseline_project": sources.get("meta2_root"), "source_artifacts": sources, "source_snapshot": source_snapshot, "network_calls_made": 0, } baseline_snapshot_path = run_dir / "manifests/baseline_snapshot.json" write_json(baseline_snapshot_path, baseline_snapshot) batch_plan = { "schema": "meta3.batch_plan.v1", "run_id": manifest["run_id"], "created_at": datetime.now(timezone.utc).isoformat(), "network_calls_made": 0, "source_snapshot": source_snapshot, "baseline_snapshot_path": str(baseline_snapshot_path), "settings": { "safe_batch_size": batch_size, "safe_concurrency": concurrency, "question_count": question_count, "answer_passes_per_question": answer_passes, "synthesis_passes": synthesis_passes, "max_api_calls": max_calls, }, "process_model": { "phase_count": len(process_phases), "source": manifest.get("process_model_path"), }, "estimated_calls": { "layer1_backfill": layer1_calls, "claim_ledger_generation": answer_calls, "owner_synthesis": synthesis_calls, "total": estimated_total_calls, }, "phases": process_phase_plan, "gates": gates, } out = run_dir / "manifests/batch_plan.json" write_json(out, batch_plan) print(f"batch_plan={out}") print(f"inventory_total={inventory_total}") print(f"layer1_processed={processed}") print(f"estimated_total_calls={estimated_total_calls}") print(f"open_gates={len(gates)}") return 0 if __name__ == "__main__": raise SystemExit(main())