yao-meta-skill/scripts/provider_output_eval_runner.py

#!/usr/bin/env python3
import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import Request, urlopen


ROOT = Path(__file__).resolve().parent.parent
DEFAULT_BASE_URL = "https://api.openai.com/v1/responses"
DEFAULT_HOST = urlparse(DEFAULT_BASE_URL).hostname or "api.openai.com"
ALLOWED_PATH_PREFIX = "/v1/responses"
LOCAL_HOSTS = {"127.0.0.1", "localhost", "::1"}


def fail(message: str) -> None:
    print(message, file=sys.stderr)
    raise SystemExit(2)


def validate_base_url(base_url: str, allow_insecure_localhost: bool, allow_custom_base_url: bool) -> None:
    parsed = urlparse(base_url)
    host = parsed.hostname or ""
    if not parsed.path.startswith(ALLOWED_PATH_PREFIX):
        fail(f"provider endpoint path must start with {ALLOWED_PATH_PREFIX}")
    if parsed.scheme == "https" and (host == DEFAULT_HOST or allow_custom_base_url):
        return
    if parsed.scheme == "https":
        fail("custom provider host requires --allow-custom-base-url")
    if parsed.scheme == "http" and allow_insecure_localhost and (parsed.hostname or "") in LOCAL_HOSTS:
        return
    fail("provider runner requires HTTPS; use --allow-insecure-localhost only for local test servers")


def load_request() -> dict[str, Any]:
    raw = sys.stdin.read()
    if not raw.strip():
        fail("provider runner requires a JSON request on stdin")
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError as exc:
        fail(f"invalid JSON request: {exc}")
    if not isinstance(payload, dict):
        fail("runner request must be a JSON object")
    return payload


def safe_relative(path_value: str) -> Path | None:
    path = Path(path_value)
    if path.is_absolute() or ".." in path.parts:
        return None
    return path


def read_input_files(paths: Any, input_root: Path, max_chars: int) -> list[dict[str, str]]:
    if not isinstance(paths, list):
        return []
    files: list[dict[str, str]] = []
    for item in paths:
        rel = safe_relative(str(item))
        if rel is None:
            files.append({"path": str(item), "status": "skipped-unsafe-path", "content": ""})
            continue
        path = input_root / rel
        if not path.exists() or not path.is_file():
            files.append({"path": str(item), "status": "missing", "content": ""})
            continue
        text = path.read_text(encoding="utf-8", errors="replace")
        files.append({"path": str(item), "status": "loaded", "content": text[:max_chars]})
    return files


def read_skill_instructions(path: Path, max_chars: int) -> str:
    if not path.exists() or not path.is_file():
        return ""
    return path.read_text(encoding="utf-8", errors="replace")[:max_chars]


def build_provider_input(request: dict[str, Any], skill_text: str, input_files: list[dict[str, str]]) -> str:
    variant = str(request.get("variant", ""))
    lines = [
        "You are producing one output for a Yao Meta Skill output-eval case.",
        f"Case id: {request.get('case_id', '')}",
        f"Variant: {variant}",
        "",
        "User task:",
        str(request.get("prompt", "")),
        "",
    ]
    if variant == "with_skill":
        lines.extend(
            [
                "Use the skill instructions below as the operating guidance. Preserve concrete evidence paths and boundaries.",
                "",
                "Skill instructions:",
                skill_text or "(Skill instructions were not available.)",
                "",
            ]
        )
    else:
        lines.extend(
            [
                "Produce a direct baseline answer without using the Yao Meta Skill guidance.",
                "Do not invent files, reports, governance evidence, or hidden review artifacts.",
                "",
            ]
        )
    if input_files:
        lines.append("Input files:")
        for item in input_files:
            lines.append(f"--- {item['path']} ({item['status']}) ---")
            if item["content"]:
                lines.append(item["content"])
        lines.append("")
    lines.extend(
        [
            "Return only the final user-facing answer for this variant.",
            "Do not mention or copy any fixture output from the eval case.",
        ]
    )
    return "\n".join(lines)


def response_text(payload: dict[str, Any]) -> str:
    if isinstance(payload.get("output_text"), str):
        return str(payload["output_text"])
    parts: list[str] = []
    output = payload.get("output")
    if isinstance(output, list):
        for item in output:
            if not isinstance(item, dict):
                continue
            content = item.get("content")
            if not isinstance(content, list):
                continue
            for block in content:
                if not isinstance(block, dict):
                    continue
                if isinstance(block.get("text"), str):
                    parts.append(str(block["text"]))
    if parts:
        return "\n".join(part for part in parts if part).strip()
    choices = payload.get("choices")
    if isinstance(choices, list) and choices:
        first = choices[0]
        if isinstance(first, dict):
            message = first.get("message", {})
            if isinstance(message, dict) and isinstance(message.get("content"), str):
                return str(message["content"])
    return ""


def observed_usage(payload: dict[str, Any]) -> dict[str, Any]:
    usage = payload.get("usage", {})
    if not isinstance(usage, dict):
        return {}
    input_tokens = usage.get("input_tokens", usage.get("prompt_tokens"))
    output_tokens = usage.get("output_tokens", usage.get("completion_tokens"))
    total_tokens = usage.get("total_tokens")
    result: dict[str, Any] = {}
    if input_tokens is not None:
        result["input_tokens"] = int(input_tokens)
    if output_tokens is not None:
        result["output_tokens"] = int(output_tokens)
    if total_tokens is not None:
        result["total_tokens"] = int(total_tokens)
    if result:
        result["estimated"] = False
    return result


def call_provider(base_url: str, api_key: str, model: str, provider_input: str, timeout_seconds: float) -> dict[str, Any]:
    body = json.dumps({"model": model, "input": provider_input}, ensure_ascii=False).encode("utf-8")
    request = Request(
        base_url,
        data=body,
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    try:
        with urlopen(request, timeout=timeout_seconds) as response:
            response_body = response.read().decode("utf-8", errors="replace")
    except HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")[:500]
        fail(f"provider request failed with HTTP {exc.code}: {detail}")
    except URLError as exc:
        fail(f"provider request failed: {exc.reason}")
    try:
        payload = json.loads(response_body)
    except json.JSONDecodeError as exc:
        fail(f"provider returned invalid JSON: {exc}")
    if not isinstance(payload, dict):
        fail("provider response must be a JSON object")
    return payload


def main() -> None:
    parser = argparse.ArgumentParser(
        description=(
            "Provider-backed output-eval runner for run_output_execution.py. "
            "Requires real model credentials and reports execution_kind=model only after an HTTP provider call."
        )
    )
    parser.add_argument("--provider", default="openai", help="Provider label to write into execution evidence.")
    parser.add_argument("--base-url", default=DEFAULT_BASE_URL, help="OpenAI Responses API compatible endpoint.")
    parser.add_argument("--model", default=os.environ.get("YAO_OUTPUT_EVAL_MODEL", ""))
    parser.add_argument("--api-key-env", default="OPENAI_API_KEY")
    parser.add_argument("--input-root", default=str(ROOT / "evals" / "output"))
    parser.add_argument("--skill-file", default=str(ROOT / "SKILL.md"))
    parser.add_argument("--timeout-seconds", type=float, default=60.0)
    parser.add_argument("--max-input-file-chars", type=int, default=6000)
    parser.add_argument("--max-skill-chars", type=int, default=8000)
    parser.add_argument("--allow-insecure-localhost", action="store_true")
    parser.add_argument("--allow-custom-base-url", action="store_true")
    args = parser.parse_args()

    validate_base_url(args.base_url, args.allow_insecure_localhost, args.allow_custom_base_url)
    if not args.model:
        fail("missing model; pass --model or set YAO_OUTPUT_EVAL_MODEL")
    api_key = os.environ.get(args.api_key_env, "")
    if not api_key:
        fail(f"missing API key env: {args.api_key_env}")

    request = load_request()
    input_files = read_input_files(request.get("input_files", []), Path(args.input_root).resolve(), args.max_input_file_chars)
    skill_text = read_skill_instructions(Path(args.skill_file).resolve(), args.max_skill_chars)
    provider_input = build_provider_input(request, skill_text, input_files)
    response = call_provider(args.base_url, api_key, args.model, provider_input, args.timeout_seconds)
    output = response_text(response)
    if not output:
        fail("provider response did not contain output text")
    result = {
        "output": output,
        "execution_kind": "model",
        "provider": args.provider,
        "model": args.model,
        "usage": observed_usage(response),
        "response_id": str(response.get("id", "")),
    }
    print(json.dumps(result, ensure_ascii=False))


if __name__ == "__main__":
    main()