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 _zh() -> bool:
    lang = (os.environ.get("GATEKEEP_LANGUAGE") or "").strip().lower()
    return lang.startswith("zh")


def die(msg: str, code: int = 1, msg_zh: str | None = None) -> None:
    text = msg_zh if (_zh() and msg_zh) else msg
    print(f"gatekeep: {text}", 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)",
                msg_zh="加载 providers.yaml 需要 PyYAML（请 pip install pyyaml 或使用 Action 的 venv 步骤）",
            )
        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, timeout: float = 25) -> dict | list:
    req = urllib.request.Request(url, headers=headers, method="GET")
    with urllib.request.urlopen(req, timeout=timeout) 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)})",
            msg_zh=f"未知 protocol {p!r}（可选 {' | '.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] | None:
    """Return model id set, or None if discovery failed after retries (skip validation)."""
    import time

    base = base_url.rstrip("/")
    wire = wire_protocol(protocol)
    if wire not in ("openai-completions", "openai-responses", "anthropic"):
        die(f"unhandled protocol wire {wire!r}")

    last_err: Exception | None = None
    attempts = int(os.environ.get("GATEKEEP_DISCOVER_RETRIES", "3"))
    for attempt in range(max(1, attempts)):
        try:
            ids: set[str] = set()
            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)
            else:
                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)
            return ids
        except urllib.error.HTTPError as e:
            body = e.read().decode("utf-8", errors="replace")[:400]
            last_err = RuntimeError(f"HTTP {e.code} {body}")
            # 401/403 are configuration errors — fail closed
            if e.code in (401, 403):
                die(
                    f"model discovery failed for {base}: {last_err}",
                    msg_zh=f"拉取模型列表失败（{base}）：{last_err} — 请检查 key",
                )
        except Exception as e:
            last_err = e
        if attempt + 1 < attempts:
            sleep_s = 2**attempt
            print(
                f"gatekeep: model discovery retry {attempt + 1}/{attempts} "
                f"after {sleep_s}s ({last_err})",
                file=sys.stderr,
            )
            time.sleep(sleep_s)

    print(
        f"gatekeep: model discovery failed for {base} after {attempts} tries "
        f"({last_err}); skipping name validation and continuing",
        file=sys.stderr,
    )
    return None


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)
    # Resolve language early so die() can speak Chinese when configured
    override_lang = (os.environ.get("GATEKEEP_LANGUAGE") or "").strip()
    cfg_lang = str(cfg.get("language") or cfg.get("report_language") or "").strip()
    language = override_lang or cfg_lang or "en"
    os.environ["GATEKEEP_LANGUAGE"] = language

    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)}",
            msg_zh=f"并行审查至少需要 2–3 个模型，当前只有 {len(requested)} 个",
        )
    fallback_requested: list[dict] = []
    if len(requested) > 3:
        print(
            f"gatekeep: primary parallel set = first 3 of {len(requested)}; "
            f"rest kept as fallbacks on 429/failure",
            file=sys.stderr,
        )
        fallback_requested = requested[3:]
        requested = requested[:3]

    provider_meta: dict[str, dict] = {}
    for pid in {r["provider"] for r in requested + fallback_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).",
                msg_zh=(
                    f"缺少供应商 {pid!r} 的密钥：需要环境变量 {key_name!r}"
                    f"（来自 providers.*.key；请在 workflow step 的 env 里从 secrets 注入）。"
                ),
            )
        available = discover_model_ids(pconf["base_url"], pconf["protocol"], api_key)
        if available is not None and len(available) == 0:
            die(
                f"provider {pid}: model list empty — check base_url/protocol/key",
                msg_zh=f"供应商 {pid} 模型列表为空 — 请检查 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) if available is not None else -1,
            "_available": available,  # None ⇒ skip validation
        }

    def to_plan_entry(r: dict) -> dict:
        pid, model = r["provider"], r["model"]
        avail = provider_meta[pid]["_available"]
        if avail is not None and 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}",
                msg_zh=(
                    f"模型 {model!r} 在供应商 {pid} 上不可用"
                    f"（列表共 {provider_meta[pid]['available_count']} 个）。示例：{sample}"
                ),
            )
        return {
            "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"],
        }

    plan_models = [to_plan_entry(r) for r in requested]
    fallback_models = [to_plan_entry(r) for r in fallback_requested]

    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)),
        "language": language,
        "models": plan_models,
        "fallback_models": fallback_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],
                "fallbacks": [f"{m['provider']}:{m['model']}" for m in fallback_models],
                "keys": {m["provider"]: m["key_name"] for m in plan_models},
                "agreement_min": plan["agreement_min"],
                "language": plan["language"],
            }
        ),
        flush=True,
    )


if __name__ == "__main__":
    main()