evaluate_objective_compiler.py

from __future__ import annotations

import argparse
import json
import sys
import urllib.error
import urllib.request
from decimal import Decimal
from pathlib import Path
from typing import Any


ALLOWED_VERIFIERS = {"command", "github_ci", "http", "schema"}


def fail(message: str) -> None:
    raise ValueError(message)


def request_plan(api_base_url: str, case: dict[str, Any]) -> dict[str, Any]:
    payload = {
        key: case[key]
        for key in (
            "objective",
            "context",
            "constraints",
            "max_tasks",
            "solver_budget_usdc",
        )
        if key in case
    }
    payload["idempotency_key"] = f"build-week-eval:{case['name']}"
    request = urllib.request.Request(
        f"{api_base_url.rstrip('/')}/v1/cloud-agent/objective-plans",
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=120) as response:
            return json.load(response)
    except urllib.error.HTTPError as error:
        body = error.read().decode("utf-8", errors="replace")
        fail(f"hosted compiler returned HTTP {error.code}: {body[:500]}")


def validate_plan(
    plan: dict[str, Any],
    case: dict[str, Any],
    expected_model_prefix: str | None = None,
) -> dict[str, Any]:
    if plan.get("schema_version") != "agent-bounties/cloud-objective-plan-v1":
        fail("wrong objective plan schema")
    if expected_model_prefix and not str(plan.get("model", "")).startswith(
        expected_model_prefix
    ):
        fail(f"plan was not generated by {expected_model_prefix}")

    tasks = plan.get("tasks")
    if not isinstance(tasks, list) or not 2 <= len(tasks) <= case["max_tasks"]:
        fail("task count is outside the requested bound")
    task_ids = [task.get("task_id") for task in tasks]
    if len(task_ids) != len(set(task_ids)) or not all(isinstance(item, str) for item in task_ids):
        fail("task IDs must be unique strings")

    dependencies: dict[str, list[str]] = {}
    for task in tasks:
        task_id = task["task_id"]
        depends_on = task.get("depends_on")
        if not isinstance(depends_on, list) or any(item not in task_ids for item in depends_on):
            fail(f"{task_id} has an unknown dependency")
        dependencies[task_id] = depends_on
        if not task.get("acceptance_criteria"):
            fail(f"{task_id} has no acceptance criteria")
        verifier = task.get("verifier", {})
        if verifier.get("kind") not in ALLOWED_VERIFIERS:
            fail(f"{task_id} has a non-deterministic verifier")
        evidence = task.get("evidence_schema", {})
        if evidence.get("type") != "object" or not evidence.get("required"):
            fail(f"{task_id} has no required evidence schema")

    visiting: set[str] = set()
    visited: set[str] = set()

    def visit(task_id: str) -> None:
        if task_id in visiting:
            fail("task graph contains a cycle")
        if task_id in visited:
            return
        visiting.add(task_id)
        for dependency in dependencies[task_id]:
            visit(dependency)
        visiting.remove(task_id)
        visited.add(task_id)

    for task_id in task_ids:
        visit(task_id)

    verification = plan.get("verification_policy", {})
    settlement = plan.get("settlement_policy", {})
    if verification.get("model_authority") != "advisory_only":
        fail("model authority is not advisory-only")
    if not verification.get("committed_before_claim"):
        fail("verification is not committed before claim")
    if settlement.get("protocol") != "autonomous-v1":
        fail("wrong settlement protocol")
    if settlement.get("network") != "base-mainnet" or settlement.get("asset") != "native USDC":
        fail("wrong settlement rail")
    if settlement.get("payout_evidence") != "confirmed canonical BountySettled":
        fail("wrong payout evidence boundary")

    if case.get("solver_budget_usdc"):
        allocated = sum(
            Decimal(task["suggested_solver_reward_usdc"])
            for task in tasks
        )
        if allocated != Decimal(case["solver_budget_usdc"]):
            fail(f"solver budget does not conserve value: {allocated}")

    combined = json.dumps(tasks, sort_keys=True).lower()
    matched = [term for term in case.get("expected_terms", []) if term.lower() in combined]
    return {
        "task_count": len(tasks),
        "keyword_coverage": len(matched) / max(1, len(case.get("expected_terms", []))),
        "matched_terms": matched,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Evaluate the hosted GPT-5.6 Objective Compiler.")
    parser.add_argument(
        "--corpus",
        default="benchmarks/openai-build-week/objective-compiler-corpus.json",
    )
    parser.add_argument("--api-base-url", default="https://api.agentbounties.app")
    parser.add_argument("--minimum-keyword-coverage", type=float, default=0.75)
    args = parser.parse_args()

    corpus = json.loads(Path(args.corpus).read_text(encoding="utf-8"))
    results = []
    failures = []
    for case in corpus["cases"]:
        try:
            plan = request_plan(args.api_base_url, case)
            metrics = validate_plan(plan, case, "gpt-5.6")
            results.append({"name": case["name"], "passed": True, **metrics})
        except (ValueError, urllib.error.URLError, TimeoutError) as error:
            failures.append({"name": case["name"], "error": str(error)})

    coverage = (
        sum(result["keyword_coverage"] for result in results) / len(results)
        if results
        else 0.0
    )
    report = {
        "schema_version": "agent-bounties/objective-compiler-eval-result-v1",
        "api_base_url": args.api_base_url,
        "cases": len(corpus["cases"]),
        "passed": len(results),
        "keyword_coverage": round(coverage, 4),
        "results": results,
        "failures": failures,
    }
    print(json.dumps(report, indent=2))
    return 0 if not failures and coverage >= args.minimum_keyword_coverage else 1


if __name__ == "__main__":
    sys.exit(main())