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