#!/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())