scripts/meta3_preflight.py

#!/usr/bin/env python3
"""Create a Meta3 preflight report for a future network run. No network calls."""

from __future__ import annotations

import argparse
import json
import os
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 run_dir_for(run_id: str | None) -> Path:
    return ROOT / "runs" / run_id if run_id else latest_run_dir()


def first_env_value(names: list[str]) -> tuple[str | None, str | None]:
    for name in names:
        value = os.environ.get(name)
        if value:
            return name, value
    return None, None


def api_key_count(value: str | None) -> int:
    if not value:
        return 0
    return len([part for part in value.replace("\n", ",").split(",") if part.strip()])


def gate(gate_id: str, status: str, meaning: str, **details: Any) -> dict[str, Any]:
    return {"id": gate_id, "status": status, "meaning": meaning, **details}


def planned_calls_for_modes(run_dir: Path, modes: list[str], limit: int) -> tuple[int, dict[str, int]]:
    total = 0
    by_mode: dict[str, int] = {}
    for mode in modes:
        manifest_path = run_dir / f"outputs/workpackets/{mode}_packet_manifest.json"
        if not manifest_path.exists():
            by_mode[mode] = 0
            continue
        manifest = load_json(manifest_path)
        planned = int(manifest.get("planned_call_count") or 0)
        if limit:
            planned = min(planned, limit)
        by_mode[mode] = planned
        total += planned
    return total, by_mode


def runner_lock_path(run_dir: Path, mode: str) -> Path:
    return run_dir / "state" / f"{mode}_runner.lock.json"


def stop_paths(run_dir: Path, mode: str) -> list[Path]:
    return [run_dir / "state" / "STOP", run_dir / "state" / f"STOP_{mode}"]


def requirements_lock_status(path: Path) -> tuple[str, str]:
    if not path.exists():
        return "open", "requirements/LOCKED.md must exist before network execution."
    text = path.read_text(encoding="utf-8")
    required_markers = [
        "Status: locked",
        "## Ziel",
        "## Fragen",
        "## Quellen",
        "## Budget",
        "## Outputs",
        "## Abbruchbedingungen",
        "## Morgen-Erfolg",
    ]
    missing = [marker for marker in required_markers if marker not in text]
    if missing:
        return "open", "requirements/LOCKED.md is missing required markers: " + ", ".join(missing)
    if "TODO" in text or "TBD" in text:
        return "open", "requirements/LOCKED.md still contains TODO/TBD placeholders."
    return "pass", "Requirements are locked in requirements/LOCKED.md."


def markdown_report(report: dict[str, Any]) -> str:
    lines = [
        "# Meta3 Network Preflight",
        "",
        f"Run: `{report['run_id']}`",
        f"Created: `{report['created_at']}`",
        f"Status: `{report['status']}`",
        "",
        "## Planned Scope",
        "",
        f"- Modes: {', '.join(report['modes'])}",
        f"- Planned calls: {report['planned_calls_total']}",
        f"- Accepted call budget: {report['accepted_call_budget']}",
        f"- Concurrency: {report['concurrency']}",
        f"- API key env: {report['api_key_env'] or 'missing'}",
        f"- API key count: {report['api_key_count']}",
        f"- Base URL configured: {report['base_url_configured']}",
        "",
        "## Gates",
        "",
    ]
    for item in report["gates"]:
        lines.append(f"- `{item['id']}`: `{item['status']}` - {item['meaning']}")
    lines.extend(["", "## Network", "", "- network_calls_made: 0"])
    return "\n".join(lines) + "\n"


def main() -> int:
    parser = argparse.ArgumentParser(description="Write Meta3 network preflight report 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="Mode to include. Can be repeated. Defaults to both.",
    )
    parser.add_argument("--limit", type=int, default=0, help="Per-mode selected packet cap. 0 means all.")
    parser.add_argument("--concurrency", type=int, default=0, help="Defaults to profile safe_concurrency.")
    parser.add_argument("--accepted-call-budget", type=int, default=0)
    parser.add_argument("--strict", action="store_true", help="Exit nonzero when gates are open or blocked.")
    args = parser.parse_args()

    run_dir = run_dir_for(args.run_id)
    run_id = run_dir.name
    modes = args.mode or ["calibration", "claim_ledger"]
    manifest = load_json(run_dir / "manifests/run_manifest.json")
    profile = manifest["profile"]
    limits = profile.get("limits", {})
    api_profile = profile.get("api_profile", {})

    safe_concurrency = int(limits.get("safe_concurrency") or 1)
    max_concurrency = int(limits.get("max_concurrency_without_new_approval") or safe_concurrency)
    concurrency = args.concurrency or safe_concurrency
    max_api_calls = int(limits.get("max_api_calls") or 0)

    base_env, base_value = first_env_value(list(api_profile.get("base_url_envs", [])))
    key_env, key_value = first_env_value(list(api_profile.get("api_key_envs", [])))
    base_url = (base_value or api_profile.get("default_base_url") or "").rstrip("/")
    key_count = api_key_count(key_value)
    planned_total, planned_by_mode = planned_calls_for_modes(run_dir, modes, args.limit)

    gates: list[dict[str, Any]] = []
    requirements_lock = ROOT / "requirements/LOCKED.md"
    lock_status, lock_meaning = requirements_lock_status(requirements_lock)
    gates.append(
        gate(
            "requirements_locked",
            lock_status,
            lock_meaning,
            path=str(requirements_lock),
        )
    )

    readiness_path = run_dir / "reports/run_readiness_report.json"
    readiness_status = None
    if readiness_path.exists():
        readiness_status = load_json(readiness_path).get("status")
    gates.append(
        gate(
            "run_readiness_passed",
            "pass" if readiness_status == "pass" else "open",
            "Run artifact validator must pass.",
            path=str(readiness_path),
            readiness_status=readiness_status,
        )
    )

    for mode in modes:
        dry_run_path = run_dir / "logs" / f"{mode}_runner_dry_run.json"
        dry_run_status = "pass" if dry_run_path.exists() else "open"
        gates.append(
            gate(
                f"{mode}_runner_dry_run_exists",
                dry_run_status,
                "Runner dry-run must exist before network execution.",
                path=str(dry_run_path),
            )
        )
        lock_path = runner_lock_path(run_dir, mode)
        gates.append(
            gate(
                f"{mode}_runner_lock_absent",
                "pass" if not lock_path.exists() else "blocked",
                "No runner lock may exist before a new network execution starts.",
                path=str(lock_path),
            )
        )
        active_stop_paths = [str(path) for path in stop_paths(run_dir, mode) if path.exists()]
        gates.append(
            gate(
                f"{mode}_stop_file_absent",
                "pass" if not active_stop_paths else "blocked",
                "Stop files block network execution until intentionally removed.",
                active_stop_paths=active_stop_paths,
            )
        )

    gates.append(
        gate(
            "accepted_call_budget_covers_plan",
            "pass" if args.accepted_call_budget >= planned_total and planned_total > 0 else "open",
            "Accepted call budget must cover selected planned calls.",
            accepted_call_budget=args.accepted_call_budget,
            planned_calls_total=planned_total,
        )
    )
    gates.append(
        gate(
            "planned_calls_within_profile_cap",
            "pass" if not max_api_calls or planned_total <= max_api_calls else "blocked",
            "Planned calls must be inside profile max_api_calls.",
            max_api_calls=max_api_calls,
            planned_calls_total=planned_total,
        )
    )
    gates.append(
        gate(
            "concurrency_within_approved_limit",
            "pass" if concurrency <= max_concurrency else "blocked",
            "Concurrency must not exceed max_concurrency_without_new_approval.",
            concurrency=concurrency,
            max_concurrency_without_new_approval=max_concurrency,
        )
    )
    gates.append(
        gate(
            "api_base_url_configured",
            "pass" if bool(base_url) else "open",
            "OpenAI-compatible base URL must be configured.",
            base_url_env=base_env,
        )
    )
    gates.append(
        gate(
            "api_key_env_present",
            "pass" if key_count > 0 else "open",
            "At least one API key must be present via configured env vars.",
            api_key_env=key_env,
            api_key_count=key_count,
        )
    )

    status = "pass" if all(item["status"] == "pass" for item in gates) else "blocked"
    report = {
        "schema": "meta3.network_preflight.v1",
        "run_id": run_id,
        "created_at": datetime.now(timezone.utc).isoformat(),
        "status": status,
        "modes": modes,
        "planned_calls_total": planned_total,
        "planned_calls_by_mode": planned_by_mode,
        "accepted_call_budget": args.accepted_call_budget,
        "concurrency": concurrency,
        "api_key_env": key_env,
        "api_key_count": key_count,
        "base_url_configured": bool(base_url),
        "base_url_env": base_env,
        "gates": gates,
        "network_calls_made": 0,
    }
    report_path = run_dir / "reports/network_preflight.json"
    report_md_path = run_dir / "reports/NETWORK_PREFLIGHT.md"
    write_json(report_path, report)
    report_md_path.write_text(markdown_report(report), encoding="utf-8")

    print(f"network_preflight={report_path}")
    print(f"network_preflight_md={report_md_path}")
    print(f"status={status}")
    print(f"planned_calls_total={planned_total}")
    print(f"accepted_call_budget={args.accepted_call_budget}")
    print("network_calls_made=0")
    return 0 if status == "pass" or not args.strict else 1


if __name__ == "__main__":
    raise SystemExit(main())