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