scripts/meta3_run_packets.py

#!/usr/bin/env python3
"""Run or dry-run Meta3 workpackets against an OpenAI-compatible API."""

from __future__ import annotations

import argparse
import json
import os
import socket
import time
import urllib.error
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timezone
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


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 write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8") as handle:
        for row in rows:
            handle.write(json.dumps(row, ensure_ascii=False, sort_keys=True) + "\n")


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


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


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


def stop_requested(run_dir: Path, mode: str) -> str | None:
    for path in stop_paths(run_dir, mode):
        if path.exists():
            return str(path)
    return None


def acquire_runner_lock(run_dir: Path, mode: str, selected_count: int) -> Path:
    path = runner_lock_path(run_dir, mode)
    payload = {
        "schema": "meta3.runner_lock.v1",
        "run_id": run_dir.name,
        "mode": mode,
        "pid": os.getpid(),
        "host": socket.gethostname(),
        "created_at": utc_now(),
        "selected_packet_count": selected_count,
    }
    path.parent.mkdir(parents=True, exist_ok=True)
    try:
        fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
    except FileExistsError as exc:
        raise SystemExit(f"Runner lock exists, refusing double start: {path}") from exc
    with os.fdopen(fd, "w", encoding="utf-8") as handle:
        handle.write(json.dumps(payload, ensure_ascii=False, indent=2) + "\n")
    return path


def release_runner_lock(path: Path) -> None:
    try:
        path.unlink()
    except FileNotFoundError:
        pass


def write_heartbeat(
    *,
    run_dir: Path,
    mode: str,
    status: str,
    selected_count: int,
    completed: int,
    failed: int,
    in_flight: int,
    next_index: int,
    network_calls_made: int,
    stop_reason: str | None = None,
) -> None:
    write_json(
        runner_heartbeat_path(run_dir, mode),
        {
            "schema": "meta3.runner_heartbeat.v1",
            "run_id": run_dir.name,
            "mode": mode,
            "updated_at": utc_now(),
            "status": status,
            "selected_packet_count": selected_count,
            "completed": completed,
            "failed": failed,
            "in_flight": in_flight,
            "next_index": next_index,
            "network_calls_made": network_calls_made,
            "stop_reason": stop_reason,
        },
    )


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_keys_from_env(value: str | None) -> list[str]:
    if not value:
        return []
    keys = [part.strip() for part in value.replace("\n", ",").split(",")]
    return [key for key in keys if key]


def resolve_api_settings(manifest: dict[str, Any]) -> dict[str, Any]:
    profile = manifest["profile"]
    api_profile = profile.get("api_profile", {})
    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", [])))
    model_env = api_profile.get("model_env")
    model = os.environ.get(model_env) if model_env else None
    return {
        "base_url": (base_value or api_profile.get("default_base_url") or "").rstrip("/"),
        "base_url_env": base_env,
        "api_key_env": key_env,
        "api_keys": api_keys_from_env(key_value),
        "model": model or api_profile.get("default_model"),
        "model_env": model_env if model else None,
    }


def require_network_start_gates(
    *,
    run_dir: Path,
    mode: str,
    selected_count: int,
    accepted_call_budget: int,
    limits: dict[str, Any],
) -> None:
    failures: list[str] = []
    requirements_lock = ROOT / "requirements/LOCKED.md"
    if not requirements_lock.exists():
        failures.append("requirements/LOCKED.md is missing")
    else:
        lock_text = requirements_lock.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 lock_text]
        if missing:
            failures.append("requirements/LOCKED.md is incomplete: " + ", ".join(missing))
        if "TODO" in lock_text or "TBD" in lock_text:
            failures.append("requirements/LOCKED.md still contains TODO/TBD placeholders")

    readiness_path = run_dir / "reports/run_readiness_report.json"
    if not readiness_path.exists():
        failures.append("run_readiness_report.json is missing")
    else:
        readiness = load_json(readiness_path)
        if readiness.get("status") != "pass":
            failures.append(f"run readiness status is {readiness.get('status')!r}")

    preflight_path = run_dir / "reports/network_preflight.json"
    if not preflight_path.exists():
        failures.append("network_preflight.json is missing")
    else:
        preflight = load_json(preflight_path)
        if preflight.get("status") != "pass":
            failures.append(f"network preflight status is {preflight.get('status')!r}")

    dry_run_report_path = run_dir / "logs" / f"{mode}_runner_dry_run.json"
    if limits.get("dry_run_required", True) and not dry_run_report_path.is_file():
        failures.append(f"runner dry-run report is missing: {dry_run_report_path}")

    max_calls = int(limits.get("max_api_calls") or 0)
    if max_calls and selected_count > max_calls:
        failures.append(f"selected packets {selected_count} exceed max_api_calls {max_calls}")
    if accepted_call_budget < selected_count:
        failures.append(
            f"accepted-call-budget={accepted_call_budget} is below selected packets={selected_count}"
        )

    if failures:
        raise SystemExit("Network start refused:\n- " + "\n- ".join(failures))


def planned_packets(
    packets: list[dict[str, Any]],
    ledger_rows: list[dict[str, Any]],
    *,
    retry_failed: bool,
) -> list[dict[str, Any]]:
    ledger_by_packet = {str(row.get("packet_id") or ""): row for row in ledger_rows}
    selected: list[dict[str, Any]] = []
    for packet in packets:
        packet_id = str(packet.get("packet_id") or "")
        ledger = ledger_by_packet.get(packet_id, {})
        status = ledger.get("status", "planned")
        if status == "planned" or (retry_failed and status == "failed"):
            selected.append(packet)
    return selected


def request_payload(packet: dict[str, Any], model: str) -> dict[str, Any]:
    return {
        "model": model,
        "messages": [
            {
                "role": "system",
                "content": "You are a careful JSON-only evaluator. Use only provided evidence.",
            },
            {
                "role": "user",
                "content": packet["prompt"],
            },
        ],
        "temperature": 0.2,
    }


def call_openai_compatible(
    *,
    endpoint: str,
    api_key: str,
    model: str,
    packet: dict[str, Any],
    timeout_seconds: int,
) -> dict[str, Any]:
    payload = request_payload(packet, model)
    request = urllib.request.Request(
        endpoint,
        data=json.dumps(payload).encode("utf-8"),
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    started = datetime.now(timezone.utc).isoformat()
    with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
        raw = response.read().decode("utf-8")
        status_code = response.status
    parsed = json.loads(raw)
    return {
        "packet_id": packet["packet_id"],
        "status_code": status_code,
        "started_at": started,
        "finished_at": datetime.now(timezone.utc).isoformat(),
        "request_model": model,
        "response": parsed,
    }


def run_one_packet(
    *,
    packet: dict[str, Any],
    packet_index: int,
    api_keys: list[str],
    endpoint: str,
    model: str,
    timeout_seconds: int,
    max_retries: int,
) -> tuple[str, dict[str, Any] | None, str | None]:
    api_key = api_keys[packet_index % len(api_keys)]
    last_error: str | None = None
    for attempt in range(max_retries + 1):
        try:
            result = call_openai_compatible(
                endpoint=endpoint,
                api_key=api_key,
                model=model,
                packet=packet,
                timeout_seconds=timeout_seconds,
            )
            result["attempt"] = attempt + 1
            return packet["packet_id"], result, None
        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, ValueError) as exc:
            last_error = f"{type(exc).__name__}: {exc}"
            if attempt < max_retries:
                time.sleep(min(30, 2**attempt))
    return packet["packet_id"], None, last_error


def main() -> int:
    parser = argparse.ArgumentParser(description="Dry-run or execute Meta3 workpackets.")
    parser.add_argument("--run-id", default=None, help="Run id under runs/. Defaults to latest run.")
    parser.add_argument("--mode", choices=["calibration", "claim_ledger"], default="calibration")
    parser.add_argument("--limit", type=int, default=0, help="0 means all selected packets.")
    parser.add_argument("--concurrency", type=int, default=0, help="Defaults to profile safe_concurrency.")
    parser.add_argument("--retry-failed", action="store_true")
    parser.add_argument("--execute-network", action="store_true", help="Actually call the API. Default is dry-run.")
    parser.add_argument("--accepted-call-budget", type=int, default=0)
    parser.add_argument("--timeout-seconds", type=int, default=120)
    parser.add_argument("--max-retries", type=int, default=2)
    parser.add_argument("--max-failures", type=int, default=0, help="0 means no fixed failure-count stop.")
    parser.add_argument("--max-error-rate", type=float, default=0.2)
    args = parser.parse_args()

    run_dir = run_dir_for(args.run_id)
    manifest = load_json(run_dir / "manifests/run_manifest.json")
    packet_manifest = load_json(run_dir / f"outputs/workpackets/{args.mode}_packet_manifest.json")
    packet_path = Path(str(packet_manifest["packet_path"]))
    ledger_path = Path(str(packet_manifest["call_ledger_path"]))
    packets = load_jsonl(packet_path)
    ledger_rows = load_jsonl(ledger_path)
    selected = planned_packets(packets, ledger_rows, retry_failed=args.retry_failed)
    if args.limit:
        selected = selected[: args.limit]

    profile = manifest["profile"]
    limits = profile.get("limits", {})
    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
    if concurrency > max_concurrency:
        raise SystemExit(
            f"Refusing concurrency={concurrency}; max without new approval is {max_concurrency}."
        )

    settings = resolve_api_settings(manifest)
    dry_run_report_path = run_dir / "logs" / f"{args.mode}_runner_dry_run.json"
    dry_run_report = {
        "schema": "meta3.runner_dry_run.v1",
        "run_id": run_dir.name,
        "mode": args.mode,
        "created_at": utc_now(),
        "execute_network": args.execute_network,
        "selected_packet_count": len(selected),
        "concurrency": concurrency,
        "model": settings["model"],
        "base_url_configured": bool(settings["base_url"]),
        "base_url_env": settings["base_url_env"],
        "api_key_env": settings["api_key_env"],
        "api_key_count": len(settings["api_keys"]),
        "runner_lock_path": str(runner_lock_path(run_dir, args.mode)),
        "runner_heartbeat_path": str(runner_heartbeat_path(run_dir, args.mode)),
        "stop_paths": [str(path) for path in stop_paths(run_dir, args.mode)],
        "stop_requested": stop_requested(run_dir, args.mode),
        "network_calls_made": 0,
    }

    if not args.execute_network:
        write_json(dry_run_report_path, dry_run_report)
        write_heartbeat(
            run_dir=run_dir,
            mode=args.mode,
            status="dry_run",
            selected_count=len(selected),
            completed=0,
            failed=0,
            in_flight=0,
            next_index=0,
            network_calls_made=0,
            stop_reason=stop_requested(run_dir, args.mode),
        )
        print(f"dry_run_report={dry_run_report_path}")
        print(f"mode={args.mode}")
        print(f"selected_packets={len(selected)}")
        print(f"concurrency={concurrency}")
        print(f"api_key_env={settings['api_key_env'] or 'missing'}")
        print(f"runner_lock={runner_lock_path(run_dir, args.mode)}")
        print(f"runner_heartbeat={runner_heartbeat_path(run_dir, args.mode)}")
        print("network_calls_made=0")
        return 0

    require_network_start_gates(
        run_dir=run_dir,
        mode=args.mode,
        selected_count=len(selected),
        accepted_call_budget=args.accepted_call_budget,
        limits=limits,
    )
    if not settings["base_url"]:
        raise SystemExit("No API base URL configured.")
    if not settings["api_keys"]:
        raise SystemExit("No API key env is present.")

    endpoint = settings["base_url"] + "/chat/completions"
    outputs_dir = run_dir / "outputs/responses" / args.mode
    outputs_dir.mkdir(parents=True, exist_ok=True)
    ledger_by_packet = {str(row.get("packet_id") or ""): row for row in ledger_rows}
    errors: list[dict[str, Any]] = []
    completed = 0
    next_index = 0
    stop_reason: str | None = stop_requested(run_dir, args.mode)
    lock_path = acquire_runner_lock(run_dir, args.mode, len(selected))

    try:
        write_heartbeat(
            run_dir=run_dir,
            mode=args.mode,
            status="running",
            selected_count=len(selected),
            completed=completed,
            failed=len(errors),
            in_flight=0,
            next_index=next_index,
            network_calls_made=0,
            stop_reason=stop_reason,
        )
        with ThreadPoolExecutor(max_workers=concurrency) as pool:
            futures: dict[Any, tuple[dict[str, Any], int]] = {}
            while (next_index < len(selected) and not stop_reason) or futures:
                while next_index < len(selected) and len(futures) < concurrency and not stop_reason:
                    packet = selected[next_index]
                    future = pool.submit(
                        run_one_packet,
                        packet=packet,
                        packet_index=next_index,
                        api_keys=settings["api_keys"],
                        endpoint=endpoint,
                        model=settings["model"],
                        timeout_seconds=args.timeout_seconds,
                        max_retries=args.max_retries,
                    )
                    futures[future] = (packet, next_index)
                    next_index += 1
                    stop_reason = stop_requested(run_dir, args.mode)

                if not futures:
                    break

                for future in as_completed(list(futures), timeout=None):
                    packet, packet_index = futures.pop(future)
                    packet_id, result, error = future.result()
                    ledger = ledger_by_packet.get(packet_id, {})
                    ledger["updated_at"] = utc_now()
                    ledger["attempts"] = int(ledger.get("attempts") or 0) + 1
                    ledger["network_calls_made"] = int(ledger.get("network_calls_made") or 0) + 1
                    if result is not None:
                        output_path = outputs_dir / f"{packet_id}.json"
                        write_json(output_path, result)
                        ledger["status"] = "succeeded"
                        ledger["output_path"] = str(output_path)
                        ledger["error"] = None
                        completed += 1
                    else:
                        ledger["status"] = "failed"
                        ledger["error"] = error
                        errors.append({"packet_id": packet_id, "error": error, "packet_index": packet_index})
                    write_jsonl(ledger_path, list(ledger_by_packet.values()))

                    attempts_done = completed + len(errors)
                    if args.max_failures and len(errors) >= args.max_failures:
                        stop_reason = f"max_failures_reached:{args.max_failures}"
                    elif attempts_done and len(errors) / attempts_done > args.max_error_rate:
                        stop_reason = f"max_error_rate_exceeded:{args.max_error_rate}"
                    else:
                        stop_reason = stop_requested(run_dir, args.mode)

                    write_heartbeat(
                        run_dir=run_dir,
                        mode=args.mode,
                        status="stopping" if stop_reason else "running",
                        selected_count=len(selected),
                        completed=completed,
                        failed=len(errors),
                        in_flight=len(futures),
                        next_index=next_index,
                        network_calls_made=attempts_done,
                        stop_reason=stop_reason,
                    )
                    break
    finally:
        release_runner_lock(lock_path)

    run_report = {
        "schema": "meta3.runner_report.v1",
        "run_id": run_dir.name,
        "mode": args.mode,
        "created_at": utc_now(),
        "selected_packet_count": len(selected),
        "succeeded": completed,
        "failed": len(errors),
        "errors": errors,
        "stop_reason": stop_reason,
        "remaining_not_started": max(0, len(selected) - next_index),
        "network_calls_made": completed + len(errors),
        "concurrency": concurrency,
        "model": settings["model"],
        "api_key_env": settings["api_key_env"],
    }
    report_path = run_dir / "reports" / f"{args.mode}_runner_report.json"
    write_json(report_path, run_report)
    write_heartbeat(
        run_dir=run_dir,
        mode=args.mode,
        status="stopped" if stop_reason else "finished",
        selected_count=len(selected),
        completed=completed,
        failed=len(errors),
        in_flight=0,
        next_index=next_index,
        network_calls_made=completed + len(errors),
        stop_reason=stop_reason,
    )
    print(f"runner_report={report_path}")
    print(f"succeeded={completed}")
    print(f"failed={len(errors)}")
    print(f"stop_reason={stop_reason or 'none'}")
    print(f"network_calls_made={completed + len(errors)}")
    return 0 if not errors else 1


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