repro_fullstack-agent/scripts/repro_experiment.py

#!/usr/bin/env python3
"""
Toy reproduction of FullStack-Agent (ICML 2026, OpenReview xblX2nQrj7).

Tests the paper's core mechanisms on a stratified 5-instruction sample of
FullStack-Bench using the REAL models (paper's 480B + 30B via HF Inference
router) and the REAL FullStack-Learn-LM-30B (self-hosted on the Job GPU),
with the REAL FullStack-Dev planning prompt and the REAL FullStack-Bench
DB/Backend judge prompts.

Scope is explicitly `toy`: 5/101 instructions, no live Docker/Chrome/Postgres
website execution (the full pipeline needs Docker-in-Docker + Chrome +
Postgres + 400 tool calls/instruction across 101 instructions + 5 baselines
on a 480B model — genuinely infeasible in this environment). Instead we verify
the mechanisms that the claims rest on:
  - Claim 1/5: multi-agent planning quality (FullStack-Dev) vs single-agent
    ablation, with the 480B.
  - Claim 2: FullStack-Learn-LM-30B (released) vs base Qwen3-Coder-30B on
    full-stack planning + code generation.
  - Claim 4: determinism / self-agreement of the exact automated judges
    (DB judge + Backend judge) that the paper uses, with the exact backbone
    the paper uses for BE/DB judging (Qwen3-Coder-480B), at temperature 0.
"""
from __future__ import annotations

import argparse
import json
import os
import re
import sys
import time
import traceback
from pathlib import Path

import requests

HERE = Path(__file__).resolve().parent
ROOT = HERE.parent
sys.path.insert(0, str(HERE))  # for planning_prompt


def load_bench(path: Path) -> list[dict]:
    return [json.loads(l) for l in path.read_text().splitlines() if l.strip()]


def stratified_sample(records: list[dict], n_per_cat: int = 2) -> list[dict]:
    by_cat: dict[str, list[dict]] = {}
    for r in records:
        by_cat.setdefault(r["Category"]["primary_category"], []).append(r)
    order = ["Content Presentation", "User Interaction", "Data Management"]
    out: list[dict] = []
    for cat in order:
        out.extend(by_cat.get(cat, [])[:n_per_cat])
    return out[:5]


# --------------------------------------------------------------------------- #
#  LLM clients
# --------------------------------------------------------------------------- #

ROUTER = "https://router.huggingface.co/v1/chat/completions"
_vllm_base: str | None = None


def _token() -> str:
    tok = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN") or ""
    if not tok:
        p = Path("/root/.cache/huggingface/token")
        if p.is_file():
            tok = p.read_text().strip()
    return tok


def _vllm_chat_base() -> str:
    """Base URL of a self-hosted vLLM server (if VLLM_HOST/PORT set)."""
    global _vllm_base
    if _vllm_base is None:
        host = os.environ.get("VLLM_HOST", "127.0.0.1")
        port = os.environ.get("VLLM_PORT", "8000")
        _vllm_base = f"http://{host}:{port}/v1"
    return _vllm_base


def llm_chat(
    model: str,
    messages: list[dict],
    *,
    max_tokens: int = 2048,
    temperature: float = 0.0,
    timeout: int = 180,
    self_hosted: bool = False,
) -> str:
    """Chat completion. If self_hosted=True, call local vLLM; else HF router."""
    headers = {"Authorization": f"Bearer {_token()}"} if not self_hosted else {}
    url = _vllm_chat_base() + "/chat/completions" if self_hosted else ROUTER
    payload = {
        "model": model,
        "messages": messages,
        "max_tokens": max_tokens,
        "temperature": temperature,
    }
    last_err = None
    for attempt in range(3):
        try:
            t = time.time()
            r = requests.post(url, headers=headers, json=payload, timeout=timeout)
            dt = time.time() - t
            if r.status_code != 200:
                last_err = f"HTTP {r.status_code}: {r.text[:200]}"
                time.sleep(2 ** attempt)
                continue
            data = r.json()
            content = data["choices"][0]["message"]["content"]
            usage = data.get("usage", {})
            print(
                f"    [llm] {model} temp={temperature} "
                f"tok={usage.get('total_tokens','?')} t={dt:.1f}s",
                file=sys.stderr,
            )
            return content
        except Exception as e:
            last_err = repr(e)
            time.sleep(2 ** attempt)
    raise RuntimeError(f"LLM call failed for {model}: {last_err}")


# --------------------------------------------------------------------------- #
#  Real FullStack-Dev planning prompt (imported from the official repo)
# --------------------------------------------------------------------------- #

from planning_prompt import get_planning_prompt  # noqa: E402  (verbatim from official repo)


def run_planning_agent(instruction: str, model: str, *, self_hosted: bool) -> dict:
    """FullStack-Dev Planning Agent: 480B/30B produces a JSON full-stack plan."""
    prompt = get_planning_prompt(instruction, is_pure_frontend=False)
    messages = [
        {"role": "system", "content": "You are a senior full-stack software architect."},
        {"role": "user", "content": prompt},
    ]
    raw = llm_chat(
        model, messages, max_tokens=4096, temperature=0.0, self_hosted=self_hosted
    )
    plan = _extract_json(raw)
    return {"raw": raw, "plan": plan, "valid_json": bool(plan)}


def run_single_agent_ablation(instruction: str, model: str, *, self_hosted: bool) -> dict:
    """Ablation: single-agent (no multi-agent planning). One-shot full-stack spec."""
    prompt = (
        "You are a single full-stack developer. Design and specify the complete "
        "full-stack website for the following instruction in ONE response. Include "
        "frontend pages, backend API endpoints, and database entities as a single "
        "JSON object with keys: pages, apiEndpoints, entities. Place the JSON in a "
        "```json``` block.\n\nInstruction: " + instruction
    )
    messages = [{"role": "user", "content": prompt}]
    raw = llm_chat(
        model, messages, max_tokens=4096, temperature=0.0, self_hosted=self_hosted
    )
    spec = _extract_json(raw)
    return {"raw": raw, "spec": spec, "valid_json": bool(spec)}


# --------------------------------------------------------------------------- #
#  Real FullStack-Bench judge prompts (imported / inlined from official repo)
# --------------------------------------------------------------------------- #

GRADE_PROMPT_TEMPLATE = """You are a meticulous software engineer specialised in databases.

Below is the implemented database schema (tables → columns with a few rows) dumped at runtime.

IMPLEMENTED DATABASE SCHEMA:
```json
{implemented_schema}
```

Answer the following question:
Does the database contain {required_content}?

Consider whether the implemented database contain a table (or a combination of tables) that can provide the content in the question.
Answer "Yes" if:
  - a single table matches, OR
  - several tables together cover the purpose, OR
  - a superset of a table matches.
Otherwise answer "No".

Steps for you:
1. Think step-by-step.  Show your reasoning for each question.
2. After finishing all reasoning, output ONLY one valid JSON object
   fenced in ```json … ``` with the key "answer".
   - If the answer is yes, output: {{"answer": "Yes"}}
   - Otherwise, output: {{"answer": "No"}}"""


def db_judge(required_content: str, implemented_schema: dict, model: str) -> str:
    """Exact FullStack-Bench DB judge (eval_db.py:115). Returns 'Yes'/'No'."""
    schema = json.loads(json.dumps(implemented_schema))
    for k in schema:
        if "columns" in schema[k]:
            schema[k].pop("columns")
    prompt = GRADE_PROMPT_TEMPLATE.format(
        required_content=required_content,
        implemented_schema=json.dumps(schema, indent=2, ensure_ascii=False),
    )
    raw = llm_chat(
        model,
        [{"role": "user", "content": prompt}],
        max_tokens=1024,
        temperature=0.0,
    )
    m = re.search(r'"answer"\s*:\s*"(Yes|No)"', raw, flags=re.IGNORECASE)
    return m.group(1).capitalize() if m else "No"


def backend_judge(task: str, expected_result: str, model: str) -> str:
    """Exact FullStack-Bench backend judging prompt (system_prompt.py:295)."""
    prompt = (
        f"You were performing a backend testing process based on the following "
        f"backend testing task and its expected result:\n\n"
        f"Task: {task}\n\nExpected Result: {expected_result}\n\n"
        "You have reached the maximum number of interactions allowed for testing. "
        "Now, you need to provide a final judgement on whether the testing task was "
        "successful or not.\n\nYou need to follow these steps:\n"
        "1. Analyze the previous testing process and results to determine if the "
        "expected result was fully achieved.\n"
        "2. Provide a clear judgement on whether the test passed or failed. "
        'You should output:\n  - YES: if the expected result was fully achieved.\n'
        "  - NO: if the expected result could not be achieved.\n\n"
        'Note: Output the final judgement exactly in the format: '
        '"Final Judgement: [YES|NO]".'
    )
    raw = llm_chat(
        model,
        [{"role": "user", "content": prompt}],
        max_tokens=512,
        temperature=0.0,
    )
    m = re.search(r"Final\s*Judgement:\s*(YES|NO)\b", raw, flags=re.IGNORECASE)
    return m.group(1).upper() if m else "NO"


# --------------------------------------------------------------------------- #
#  Plan-quality rubric judge (used to score FullStack-Dev vs ablation,
#  and FullStack-Learn-LM-30B vs base 30B). Rubric is grounded in the
#  FullStack-Bench test-case structure (entities, API endpoints, pages,
#  data flows, data structures) that the plans are meant to support.
# --------------------------------------------------------------------------- #

RUBRIC_PROMPT = """You are a senior full-stack architect grading a website development plan.

INSTRUCTION:
{instruction}

GROUND-TRUTH FullStack-Bench SPEC for this instruction (the entities / API endpoints / \
pages / data structures the website must support):
{ground_truth}

CANDIDATE PLAN (JSON):
{plan}

Grade the candidate plan on 5 binary criteria (1 = pass, 0 = fail):
1. backend_entities: Does it define backend entities/tables matching the ground-truth data structures?
2. api_endpoints: Does it define API endpoints that cover the ground-truth backend test cases?
3. frontend_pages: Does it define frontend pages/components for the ground-truth frontend test cases?
4. data_flows: Does it wire frontend data flows to the backend API endpoints?
5. json_valid: Is the plan valid JSON with the required structure?

Output ONLY a JSON object in a ```json``` block with keys: backend_entities, \
api_endpoints, frontend_pages, data_flows, json_valid (each 0 or 1), and \
"score" (sum 0-5)."""


def grade_plan(instruction: str, ground_truth: dict, plan_json: dict | str, judge_model: str) -> dict:
    if isinstance(plan_json, (dict, list)):
        plan_str = json.dumps(plan_json, indent=2, ensure_ascii=False)[:6000]
    else:
        plan_str = str(plan_json)[:6000]
    gt = {
        "data_structures": ground_truth.get("data_structures", []),
        "backend_test_cases": [
            {"instruction": t.get("instruction"), "expected": t.get("expected_result")}
            for t in ground_truth.get("backend_test_cases", [])[:8]
        ],
        "frontend_test_cases": [
            {"task": t.get("task"), "expected": t.get("expected_result")}
            for t in ground_truth.get("ui_instruct", [])[:8]
        ],
    }
    prompt = RUBRIC_PROMPT.format(
        instruction=instruction,
        ground_truth=json.dumps(gt, indent=2, ensure_ascii=False),
        plan=plan_str,
    )
    raw = llm_chat(
        judge_model,
        [{"role": "user", "content": prompt}],
        max_tokens=1024,
        temperature=0.0,
    )
    j = _extract_json(raw) or {}
    score = sum(1 for k in ("backend_entities", "api_endpoints", "frontend_pages",
                            "data_flows", "json_valid") if j.get(k) == 1)
    j["score"] = score
    j["raw"] = raw[:500]
    return j


# --------------------------------------------------------------------------- #
#  Code-generation quality (FullStack-Learn-LM-30B vs base 30B):
#  generate a Next.js+NestJS+Postgres skeleton from the plan and check
#  it is valid TS/JS with a real backend file (not frontend-only).
# --------------------------------------------------------------------------- #

CODEGEN_PROMPT = """Generate a minimal but real full-stack Next.js + NestJS + PostgreSQL \
codebase for the following plan. Output ONLY two files in fenced code blocks, \
each preceded by a line `// FILE: <path>`:

1. backend/src/app.controller.ts — a NestJS controller exposing the first API endpoint.
2. frontend/src/app/page.tsx — a Next.js page that fetches from that endpoint.

Plan:
{plan}

Requirement: the backend file MUST contain a real NestJS controller with a \
@PostMapping or @GetMapping decorator and a real SQL/TypeORM interaction — \
NOT mock data. The frontend MUST call the backend via fetch/axios."""


def codegen_skeleton(plan_json: dict, model: str, *, self_hosted: bool) -> dict:
    plan_str = json.dumps(plan_json, indent=2, ensure_ascii=False)[:3000] if plan_json else "{}"
    raw = llm_chat(
        model,
        [{"role": "user", "content": CODEGEN_PROMPT.format(plan=plan_str)}],
        max_tokens=4096,
        temperature=0.0,
        self_hosted=self_hosted,
    )
    has_backend = bool(re.search(r"@(Post|Get|Put|Delete)Mapping", raw))
    has_sql = bool(re.search(r"Repository|@Entity|INSERT|SELECT|typeorm|prisma", raw, re.I))
    has_fetch = bool(re.search(r"fetch\(|axios", raw))
    has_mock = bool(re.search(r"mock|Mock|MOCK|hardcode|placeholder data", raw))
    files = re.findall(r"// FILE:\s*(\S+)", raw)
    return {
        "raw": raw[:4000],
        "files": files,
        "has_backend_controller": has_backend,
        "has_real_sql": has_sql,
        "has_frontend_fetch": has_fetch,
        "uses_mock_data": has_mock,
        "fullstack_score": sum([has_backend, has_sql, has_fetch]) - (1 if has_mock else 0),
    }


# --------------------------------------------------------------------------- #
#  Helpers
# --------------------------------------------------------------------------- #

def _extract_json(text: str) -> dict | None:
    # Try fenced ```json ... ``` first
    m = re.search(r"```json\s*(.*?)```", text, flags=re.DOTALL)
    if m:
        try:
            return json.loads(m.group(1))
        except json.JSONDecodeError:
            pass
    # Try any fenced block
    m = re.search(r"```\s*(.*?)```", text, flags=re.DOTALL)
    if m:
        try:
            return json.loads(m.group(1))
        except json.JSONDecodeError:
            pass
    # Try to find a top-level { ... } or { ... } (greedy-ish)
    start = text.find("{")
    if start != -1:
        depth = 0
        for i in range(start, len(text)):
            if text[i] == "{":
                depth += 1
            elif text[i] == "}":
                depth -= 1
                if depth == 0:
                    try:
                        return json.loads(text[start : i + 1])
                    except json.JSONDecodeError:
                        break
    return None


def _safe(fn, *a, **kw):
    try:
        return fn(*a, **kw), None
    except Exception as e:
        return None, f"{type(e).__name__}: {e}\n{traceback.format_exc()[:800]}"


# --------------------------------------------------------------------------- #
#  Main experiment
# --------------------------------------------------------------------------- #

def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--bench", default=str(ROOT / "FullStack-Bench.jsonl"))
    ap.add_argument("--out", default=str(HERE / "outputs" / "results.json"))
    ap.add_argument("--judge-model", default="Qwen/Qwen3-Coder-480B-A35B-Instruct")
    ap.add_argument("--big-model", default="Qwen/Qwen3-Coder-480B-A35B-Instruct")
    ap.add_argument("--small-model", default="Qwen/Qwen3-Coder-30B-A3B-Instruct")
    ap.add_argument("--learn-model", default="local")  # self-hosted
    ap.add_argument("--learn-model-name", default="luzimu/FullStack-Learn-LM-30B-A3B")
    ap.add_argument("--n", type=int, default=5)
    ap.add_argument("--skip-learn", action="store_true")
    ap.add_argument("--skip-judge-determinism", action="store_true")
    args = ap.parse_args()

    out_path = Path(args.out)
    out_path.parent.mkdir(parents=True, exist_ok=True)

    records = load_bench(Path(args.bench))
    sample = stratified_sample(records, n_per_cat=2)[: args.n]
    print(f"[repro] sample: {len(sample)} instructions "
          f"ids={[r['id'] for r in sample]}", file=sys.stderr)

    results: dict = {
        "meta": {
            "paper": "FullStack-Agent (ICML 2026, xblX2nQrj7)",
            "arxiv": "2602.03798",
            "scope": "toy",
            "sample_ids": [r["id"] for r in sample],
            "n_instructions": len(sample),
            "n_fe": sum(len(r["ui_instruct"]) for r in sample),
            "n_be": sum(len(r["backend_test_cases"]) for r in sample),
            "n_db": sum(len(r["data_structures"]) for r in sample),
            "judge_model": args.judge_model,
            "big_model": args.big_model,
            "small_model": args.small_model,
            "learn_model": args.learn_model,
            "learn_model_name": args.learn_model_name,
            "models_via": "HF Inference router (480B, 30B) + self-hosted vLLM (FullStack-Learn-LM-30B)",
        },
        "claim1_fullstack_dev_480b": [],
        "claim1_baseline_30b": [],
        "claim2_fullstack_learn_30b": [],
        "claim2_base_30b": [],
        "claim4_judge_determinism": {"db": [], "backend": []},
        "claim5_ablation": {"multi_agent": [], "single_agent": []},
    }

    # --- Claim 1: FullStack-Dev (480B) planning on the sample ---
    print("[claim1] FullStack-Dev 480B planning ...", file=sys.stderr)
    for r in sample:
        print(f"  - {r['id']}", file=sys.stderr)
        plan, err = _safe(run_planning_agent, r["instruction"], args.big_model, self_hosted=False)
        grade = None
        if plan and plan.get("plan"):
            grade, gerr = _safe(grade_plan, r["instruction"], r, plan["plan"], args.judge_model)
            if gerr:
                grade = {"error": gerr}
        results["claim1_fullstack_dev_480b"].append({
            "id": r["id"], "category": r["Category"]["primary_category"],
            "valid_json": plan.get("valid_json") if plan else None,
            "plan_keys": list(plan["plan"].keys()) if plan and plan.get("plan") else None,
            "grade": grade, "error": err or plan.get("error"),
        })

    # Baseline 30B (no FullStack-Learn) for reference / Claim 1 30B row
    print("[claim1] Base 30B planning ...", file=sys.stderr)
    for r in sample:
        print(f"  - {r['id']}", file=sys.stderr)
        plan, err = _safe(run_planning_agent, r["instruction"], args.small_model, self_hosted=False)
        grade = None
        if plan and plan.get("plan"):
            grade, gerr = _safe(grade_plan, r["instruction"], r, plan["plan"], args.judge_model)
            if gerr:
                grade = {"error": gerr}
        results["claim1_baseline_30b"].append({
            "id": r["id"], "category": r["Category"]["primary_category"],
            "valid_json": plan.get("valid_json") if plan else None,
            "plan_keys": list(plan["plan"].keys()) if plan and plan.get("plan") else None,
            "grade": grade, "error": err or plan.get("error"),
        })

    # --- Claim 2: FullStack-Learn-LM-30B (self-hosted) vs base 30B ---
    if not args.skip_learn:
        print("[claim2] FullStack-Learn-LM-30B (self-hosted) planning+codegen ...", file=sys.stderr)
        for r in sample:
            print(f"  - {r['id']}", file=sys.stderr)
            plan, perr = _safe(run_planning_agent, r["instruction"], args.learn_model_name,
                               self_hosted=True)
            grade = None
            codegen = None
            if plan and plan.get("plan"):
                grade, gerr = _safe(grade_plan, r["instruction"], r, plan["plan"], args.judge_model)
                if gerr: grade = {"error": gerr}
                codegen, cerr = _safe(codegen_skeleton, plan["plan"], args.learn_model_name,
                                      self_hosted=True)
                if cerr: codegen = {"error": cerr}
            results["claim2_fullstack_learn_30b"].append({
                "id": r["id"], "valid_json": plan.get("valid_json") if plan else None,
                "grade": grade, "codegen": codegen, "error": perr,
            })
        print("[claim2] base 30B planning+codegen ...", file=sys.stderr)
        for r in sample:
            print(f"  - {r['id']}", file=sys.stderr)
            plan, perr = _safe(run_planning_agent, r["instruction"], args.small_model, self_hosted=False)
            grade = None
            codegen = None
            if plan and plan.get("plan"):
                grade, gerr = _safe(grade_plan, r["instruction"], r, plan["plan"], args.judge_model)
                if gerr: grade = {"error": gerr}
                codegen, cerr = _safe(codegen_skeleton, plan["plan"], args.small_model, self_hosted=False)
                if cerr: codegen = {"error": cerr}
            results["claim2_base_30b"].append({
                "id": r["id"], "valid_json": plan.get("valid_json") if plan else None,
                "grade": grade, "codegen": codegen, "error": perr,
            })

    # --- Claim 4: judge determinism (proxy for human alignment) ---
    # Run the EXACT DB judge and Backend judge twice at temp=0 on the same
    # inputs; perfect self-agreement is a necessary condition for the
    # 90.5/94.0/97.5% human alignment the paper reports. We use a couple of
    # realistic DB snapshots derived from the bench data_structures.
    if not args.skip_judge_determinism:
        print("[claim4] judge determinism (DB + Backend) x2 runs ...", file=sys.stderr)
        for r in sample[:3]:
            ds = r.get("data_structures", [])
            if not ds:
                continue
            # realistic schema: one table per data_structures entry
            schema = {
                f"table_{i}": {"columns": [c.split()[0] for c in d.split()][:4], "rows": []}
                for i, d in enumerate(ds)
            }
            for run_idx in range(2):
                ans, err = _safe(db_judge, ds[0], schema, args.judge_model)
                results["claim4_judge_determinism"]["db"].append({
                    "id": r["id"], "run": run_idx, "answer": ans, "error": err,
                })
            bc = r.get("backend_test_cases", [])
            if bc:
                for run_idx in range(2):
                    ans, err = _safe(backend_judge, bc[0].get("instruction", ""),
                                     bc[0].get("expected_result", ""), args.judge_model)
                    results["claim4_judge_determinism"]["backend"].append({
                        "id": r["id"], "run": run_idx, "answer": ans, "error": err,
                    })

    # --- Claim 5: ablation (multi-agent vs single-agent) with 480B ---
    print("[claim5] ablation: multi-agent vs single-agent (480B) ...", file=sys.stderr)
    for r in sample:
        print(f"  - {r['id']}", file=sys.stderr)
        ma, err = _safe(run_planning_agent, r["instruction"], args.big_model, self_hosted=False)
        sa, serr = _safe(run_single_agent_ablation, r["instruction"], args.big_model, self_hosted=False)
        ma_grade = None
        sa_grade = None
        if ma and ma.get("plan"):
            ma_grade, _ = _safe(grade_plan, r["instruction"], r, ma["plan"], args.judge_model)
        if sa and sa.get("spec"):
            sa_grade, _ = _safe(grade_plan, r["instruction"], r, sa["spec"], args.judge_model)
        results["claim5_ablation"]["multi_agent"].append({
            "id": r["id"], "valid_json": ma.get("valid_json") if ma else None,
            "grade": ma_grade, "error": err,
        })
        results["claim5_ablation"]["single_agent"].append({
            "id": r["id"], "valid_json": sa.get("valid_json") if sa else None,
            "grade": sa_grade, "error": serr,
        })

    # --- Summary aggregates ---
    def avg_score(rows):
        xs = [row["grade"]["score"] for row in rows if row.get("grade") and isinstance(row["grade"], dict) and "score" in row["grade"]]
        return sum(xs) / len(xs) if xs else None

    def pct_valid(rows, key="plan"):
        ok = sum(1 for row in rows if row.get("valid_json"))
        return ok / len(rows) * 100 if rows else None

    def judge_agreement(rows):
        by_id: dict[str, list] = {}
        for row in rows:
            by_id.setdefault(row["id"], []).append(row.get("answer"))
        agree = 0
        total = 0
        for sid, ans in by_id.items():
            if len(ans) >= 2 and ans[0] is not None and ans[1] is not None:
                total += 1
                if ans[0] == ans[1]:
                    agree += 1
        return agree / total * 100 if total else None

    results["summary"] = {
        "claim1_480b_plan_score_avg": avg_score(results["claim1_fullstack_dev_480b"]),
        "claim1_480b_valid_json_pct": pct_valid(results["claim1_fullstack_dev_480b"]),
        "claim1_30b_plan_score_avg": avg_score(results["claim1_baseline_30b"]),
        "claim1_30b_valid_json_pct": pct_valid(results["claim1_baseline_30b"]),
        "claim2_learn_30b_plan_score_avg": avg_score(results["claim2_fullstack_learn_30b"]),
        "claim2_learn_30b_codegen_score_avg": (
            sum(row["codegen"]["fullstack_score"] for row in results["claim2_fullstack_learn_30b"]
                if row.get("codegen") and isinstance(row["codegen"], dict) and "fullstack_score" in row["codegen"])
            / max(1, sum(1 for row in results["claim2_fullstack_learn_30b"]
                         if row.get("codegen") and isinstance(row["codegen"], dict) and "fullstack_score" in row["codegen"]))
            if results["claim2_fullstack_learn_30b"] else None
        ),
        "claim2_base_30b_plan_score_avg": avg_score(results["claim2_base_30b"]),
        "claim2_base_30b_codegen_score_avg": (
            sum(row["codegen"]["fullstack_score"] for row in results["claim2_base_30b"]
                if row.get("codegen") and isinstance(row["codegen"], dict) and "fullstack_score" in row["codegen"])
            / max(1, sum(1 for row in results["claim2_base_30b"]
                         if row.get("codegen") and isinstance(row["codegen"], dict) and "fullstack_score" in row["codegen"]))
            if results["claim2_base_30b"] else None
        ),
        "claim4_db_judge_self_agreement_pct": judge_agreement(results["claim4_judge_determinism"]["db"]),
        "claim4_backend_judge_self_agreement_pct": judge_agreement(results["claim4_judge_determinism"]["backend"]),
        "claim5_multi_agent_score_avg": avg_score(results["claim5_ablation"]["multi_agent"]),
        "claim5_single_agent_score_avg": avg_score(results["claim5_ablation"]["single_agent"]),
        "claim5_multi_agent_valid_json_pct": pct_valid(results["claim5_ablation"]["multi_agent"]),
        "claim5_single_agent_valid_json_pct": pct_valid(results["claim5_ablation"]["single_agent"]),
    }

    out_path.write_text(json.dumps(results, ensure_ascii=False, indent=2))
    print(f"[repro] wrote {out_path}", file=sys.stderr)
    print(json.dumps(results["summary"], indent=2), file=sys.stderr)
    # also print summary to stdout for hf jobs logs
    print(json.dumps({"summary": results["summary"], "meta": results["meta"]}, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())