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