gatekeep/scripts/resolve-providers.py

#!/usr/bin/env python3
"""
Resolve Gatekeep providers from YAML (or legacy JSON).

Each provider entry needs:
  key:      FREE-FORM secret/env var name (NO required gatekeep_ prefix)
  protocol: openai | openai-completions | openai-responses | anthropic
Optional:
  base_url, models (defaults applied for known provider ids)

Discovers model lists from the provider API; validates requested names; writes
provider-plan.json for mixed parallel review.
"""
from __future__ import annotations

import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path

PROTOCOLS = ("openai", "openai-completions", "openai-responses", "anthropic")
# Wire aliases used by call-model.sh
PROTOCOL_WIRE = {
    "openai": "openai-completions",
    "openai-completions": "openai-completions",
    "openai-compatible": "openai-completions",  # legacy
    "openai-responses": "openai-responses",
    "anthropic": "anthropic",
}

KNOWN_DEFAULTS = {
    "openrouter": {
        "base_url": "https://openrouter.ai/api/v1",
        "protocol": "openai",
    },
    "anthropic": {
        "base_url": "https://api.anthropic.com",
        "protocol": "anthropic",
    },
    "openai": {
        "base_url": "https://api.openai.com/v1",
        "protocol": "openai",
    },
}


def die(msg: str, code: int = 1) -> None:
    print(f"gatekeep: {msg}", file=sys.stderr)
    sys.exit(code)


def load_raw(path: Path) -> dict:
    text = path.read_text()
    if path.suffix.lower() in (".yaml", ".yml"):
        try:
            import yaml  # type: ignore
        except ImportError:
            die("PyYAML required to load providers.yaml (pip install pyyaml)")
        raw = yaml.safe_load(text) or {}
    else:
        raw = json.loads(text)

    # Legacy models.json → single openrouter provider
    if "providers" not in raw and "models" in raw:
        base = raw.get("openrouter_base_url") or KNOWN_DEFAULTS["openrouter"]["base_url"]
        raw = {
            "agreement_min": raw.get("agreement_min", 2),
            "temperature": raw.get("temperature", 0),
            "max_tokens": raw.get("max_tokens", 4096),
            "providers": {
                "openrouter": {
                    "key": "OPENROUTER_API_KEY",
                    "protocol": "openai",
                    "base_url": base,
                    "models": raw["models"],
                }
            },
        }

    # Legacy JSON providers without `key`
    providers = raw.get("providers") or {}
    for pid, pconf in list(providers.items()):
        if not isinstance(pconf, dict):
            continue
        if "key" not in pconf:
            # migrate old shape: invent free-form env name suggestion from provider id
            # (not forcing gatekeep_ prefix)
            pconf = {**pconf, "key": f"{pid.upper()}_API_KEY"}
            if pconf.get("protocol") == "openai-compatible":
                pconf["protocol"] = "openai"
            providers[pid] = pconf
    raw["providers"] = providers

    if not providers:
        die(f"no providers in {path}")
    return raw


def secret_value(key_name: str) -> str:
    """Look up FREE-FORM secret name in the process environment. No prefix rules."""
    name = (key_name or "").strip()
    if not name:
        return ""
    val = os.environ.get(name, "")
    if val and val.strip():
        return val.strip()
    return ""


def http_json(url: str, headers: dict) -> dict | list:
    req = urllib.request.Request(url, headers=headers, method="GET")
    with urllib.request.urlopen(req, timeout=60) as resp:
        return json.loads(resp.read().decode("utf-8"))


def normalize_protocol(p: str) -> str:
    p = (p or "").strip().lower()
    if p not in PROTOCOL_WIRE:
        die(f"unknown protocol {p!r} (want {' | '.join(PROTOCOLS)})")
    return p


def wire_protocol(p: str) -> str:
    return PROTOCOL_WIRE[normalize_protocol(p)]


def discover_model_ids(base_url: str, protocol: str, api_key: str) -> set[str]:
    base = base_url.rstrip("/")
    wire = wire_protocol(protocol)
    ids: set[str] = set()
    try:
        if wire in ("openai-completions", "openai-responses"):
            data = http_json(
                f"{base}/models",
                {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
            )
            for m in data.get("data") or []:
                mid = m.get("id")
                if mid:
                    ids.add(mid)
        elif wire == "anthropic":
            data = http_json(
                f"{base}/v1/models",
                {
                    "x-api-key": api_key,
                    "anthropic-version": "2023-06-01",
                    "Content-Type": "application/json",
                },
            )
            for m in data.get("data") or []:
                mid = m.get("id")
                if mid:
                    ids.add(mid)
        else:
            die(f"unhandled protocol wire {wire!r}")
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")[:400]
        die(f"model discovery failed for {base}: HTTP {e.code} {body}")
    except Exception as e:
        die(f"model discovery failed for {base}: {e}")
    return ids


def enrich_provider(pid: str, pconf: dict) -> dict:
    known = KNOWN_DEFAULTS.get(pid, {})
    out = {**pconf}
    if not out.get("base_url"):
        if known.get("base_url"):
            out["base_url"] = known["base_url"]
        else:
            die(f"provider {pid}: base_url required (no default for this id)")
    if not out.get("protocol"):
        out["protocol"] = known.get("protocol") or "openai"
    out["protocol"] = normalize_protocol(out["protocol"])
    if not out.get("key"):
        die(f"provider {pid}: `key` (free-form secret/env name) is required")
    if not isinstance(out.get("models"), list) or not out["models"]:
        die(f"provider {pid}: `models` list required (names validated against provider API)")
    return out


def main() -> None:
    if len(sys.argv) < 3:
        die("usage: resolve-providers.py <providers.yaml|json> <out-plan.json> [override-models-csv]")
    cfg_path = Path(sys.argv[1])
    out_path = Path(sys.argv[2])
    override_csv = sys.argv[3] if len(sys.argv) > 3 else ""

    cfg = load_raw(cfg_path)
    providers_in = cfg["providers"]
    providers: dict[str, dict] = {
        pid: enrich_provider(pid, pconf) for pid, pconf in providers_in.items()
    }

    override_items: list[tuple[str | None, str]] = []
    if override_csv.strip():
        for part in override_csv.split(","):
            part = part.strip()
            if not part:
                continue
            head, _, rest = part.partition(":")
            if rest and head in providers:
                override_items.append((head, rest))
            else:
                override_items.append((None, part))

    requested: list[dict] = []
    if override_items:
        catalog: dict[str, list[str]] = {}
        for pid, pconf in providers.items():
            for m in pconf["models"]:
                catalog.setdefault(m, []).append(pid)
        for prov, model in override_items:
            if prov:
                requested.append({"provider": prov, "model": model})
            else:
                hits = catalog.get(model) or []
                if len(hits) == 1:
                    requested.append({"provider": hits[0], "model": model})
                elif len(hits) > 1:
                    die(f"model {model!r} under multiple providers {hits}; use provider:model")
                elif len(providers) == 1:
                    requested.append({"provider": next(iter(providers)), "model": model})
                else:
                    die(f"model {model!r} not in config; use provider:model")
    else:
        for pid, pconf in providers.items():
            for m in pconf["models"]:
                requested.append({"provider": pid, "model": m})

    seen: set[tuple[str, str]] = set()
    uniq: list[dict] = []
    for r in requested:
        key = (r["provider"], r["model"])
        if key not in seen:
            seen.add(key)
            uniq.append(r)
    requested = uniq

    if len(requested) < 2:
        die(f"need 2–3 models for parallel review; got {len(requested)}")
    if len(requested) > 3:
        print(f"gatekeep: truncating to first 3 of {len(requested)} models", file=sys.stderr)
        requested = requested[:3]

    provider_meta: dict[str, dict] = {}
    for pid in {r["provider"] for r in requested}:
        pconf = providers[pid]
        key_name = str(pconf["key"])
        api_key = secret_value(key_name)
        if not api_key:
            die(
                f"missing secret/env for provider {pid!r}: expected env var named {key_name!r} "
                f"(from providers.*.key — free-form; set it on the workflow step env from secrets)."
            )
        available = discover_model_ids(pconf["base_url"], pconf["protocol"], api_key)
        if not available:
            die(f"provider {pid}: model list empty — check base_url/protocol/key")
        provider_meta[pid] = {
            "base_url": pconf["base_url"].rstrip("/"),
            "protocol": pconf["protocol"],
            "wire_protocol": wire_protocol(pconf["protocol"]),
            "key_name": key_name,
            "api_key": api_key,
            "available_count": len(available),
            "_available": available,
        }

    plan_models = []
    for r in requested:
        pid, model = r["provider"], r["model"]
        avail = provider_meta[pid]["_available"]
        if model not in avail:
            sample = sorted(avail)[:8]
            die(
                f"model {model!r} not available on provider {pid} "
                f"({provider_meta[pid]['available_count']} listed). Examples: {sample}"
            )
        plan_models.append(
            {
                "provider": pid,
                "model": model,
                "base_url": provider_meta[pid]["base_url"],
                "protocol": provider_meta[pid]["protocol"],
                "wire_protocol": provider_meta[pid]["wire_protocol"],
                "key_name": provider_meta[pid]["key_name"],
            }
        )

    secrets_path = out_path.with_suffix(".secrets.json")
    secrets_path.write_text(
        json.dumps({pid: provider_meta[pid]["api_key"] for pid in provider_meta}, indent=2)
    )
    os.chmod(secrets_path, 0o600)

    plan = {
        "agreement_min": int(cfg.get("agreement_min", 2)),
        "temperature": cfg.get("temperature", 0),
        "max_tokens": int(cfg.get("max_tokens", 4096)),
        "models": plan_models,
        "secrets_file": str(secrets_path),
        "providers_used": sorted(provider_meta.keys()),
    }
    out_path.write_text(json.dumps(plan, indent=2))
    print(
        json.dumps(
            {
                "providers": plan["providers_used"],
                "models": [f"{m['provider']}:{m['model']}" for m in plan_models],
                "keys": {m["provider"]: m["key_name"] for m in plan_models},
                "agreement_min": plan["agreement_min"],
            }
        ),
        flush=True,
    )


if __name__ == "__main__":
    main()