scripts/eval_tool_description_ab.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import csv
import importlib.util
import json
import re
import statistics
import subprocess
import itertools
import math
from dataclasses import dataclass
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
BASE_CARDS_DIR = ROOT / '.fast-agent' / 'tool-cards'
PROMPTS_FILE = ROOT / 'scripts' / 'hf_hub_community_challenges.txt'
VARIANTS_FILE = ROOT / 'scripts' / 'tool_description_variants.json'
OUT_DIR = ROOT / 'docs' / 'tool_description_eval'
CARDS_OUT_ROOT = ROOT / '.fast-agent' / 'evals' / 'tool_desc_ab' / 'cards'
INDIRECT_ROUTER_NAME = 'hf_hub_community_router'

ANSI_RE = re.compile(r"\x1B\[[0-?]*[ -/]*[@-~]")

# Expected first endpoint patterns by case id (for first-call quality metric)
FIRST_ENDPOINT_EXPECTED: dict[int, dict[str, Any]] = {
    1: {"any": [r"/users/[^/]+/overview", r"/organizations/[^/]+/overview"]},
    2: {"any": [r"/users/[^/]+/followers"]},
    3: {"any": [r"/(api/)?recent-activity"]},
    4: {"any": [r"/(api/)?recent-activity"]},
    5: {"any": [r"/(api/)?recent-activity"]},
    6: {"any": [r"/models/[^/]+/[^/]+/discussions"]},
    7: {"no_tool_call": True},
    8: {"no_tool_call": True},
    9: {"any": [r"/whoami-v2", r"/(api/)?recent-activity"]},
    10: {"any": [r"/users/[^/]+/overview", r"/organizations/[^/]+/overview"]},
}


@dataclass
class RunRow:
    case_id: int
    prompt: str
    variant: str
    model: str
    returncode: int
    has_tool_call: bool
    endpoint_calls: int
    first_endpoint: str | None
    first_call_correct: bool | None
    score_total: int | None
    score_endpoint: int | None
    score_efficiency: int | None
    score_reasoning: int | None
    score_safety: int | None
    score_clarity: int | None
    result_file: str | None
    merged: str


def row_key(r: RunRow) -> tuple[int, str, str]:
    return (r.case_id, r.variant, r.model)


def load_existing_rows(out_dir: Path) -> list[RunRow]:
    p = out_dir / 'tool_description_ab_detailed.json'
    if not p.exists():
        return []
    data = json.loads(p.read_text(encoding='utf-8'))
    rows: list[RunRow] = []
    for d in data:
        s = d.get('score', {}) if isinstance(d, dict) else {}
        rows.append(RunRow(
            case_id=d.get('case_id'),
            prompt=d.get('prompt', ''),
            variant=d.get('variant', ''),
            model=d.get('model', ''),
            returncode=d.get('returncode', 1),
            has_tool_call=d.get('has_tool_call', False),
            endpoint_calls=d.get('endpoint_calls', 0),
            first_endpoint=d.get('first_endpoint'),
            first_call_correct=d.get('first_call_correct'),
            score_total=d.get('score_total'),
            score_endpoint=s.get('endpoint'),
            score_efficiency=s.get('efficiency'),
            score_reasoning=s.get('reasoning'),
            score_safety=s.get('safety'),
            score_clarity=s.get('clarity'),
            result_file=d.get('result_file'),
            merged=d.get('merged', ''),
        ))
    return rows


def strip_ansi(text: str) -> str:
    return ANSI_RE.sub('', text)


def load_prompts(path: Path) -> list[str]:
    lines = [ln.strip() for ln in path.read_text(encoding='utf-8').splitlines()]
    return [ln for ln in lines if ln]


def load_variants(path: Path) -> list[dict[str, str]]:
    data = json.loads(path.read_text(encoding='utf-8'))
    if not isinstance(data, list):
        raise ValueError('variants file must be a JSON list')
    out: list[dict[str, str]] = []
    for item in data:
        vid = item.get('id')
        desc = item.get('card_description')
        doc = item.get('hf_api_request_docstring')
        if not vid or not desc or not doc:
            raise ValueError(f'Invalid variant item: {item}')
        out.append({'id': vid, 'card_description': desc, 'hf_api_request_docstring': doc})
    return out


def maybe_import_base_scorer() -> Any | None:
    p = ROOT / 'scripts' / 'score_hf_hub_community_challenges.py'
    if not p.exists():
        return None
    spec = importlib.util.spec_from_file_location('base_scorer', p)
    if not spec or not spec.loader:
        return None
    mod = importlib.util.module_from_spec(spec)
    import sys
    sys.modules[spec.name] = mod
    spec.loader.exec_module(mod)
    return mod


def replace_card_description(base_card_text: str, new_description: str) -> str:
    # Replace first frontmatter description line.
    esc = new_description.replace('"', '\\"')
    replaced, n = re.subn(
        r'(?m)^description:\s*".*"\s*$',
        f'description: "{esc}"',
        base_card_text,
        count=1,
    )
    if n == 0:
        raise ValueError('Could not find frontmatter description line in base card')
    return replaced


def replace_hf_api_docstring(base_tool_text: str, new_docstring: str) -> str:
    # Replace only the hf_api_request function docstring block.
    pattern = re.compile(
        r"(def hf_api_request\([\s\S]*?\) -> dict\[str, Any\]:\n\s*)\"\"\"[\s\S]*?\"\"\"",
        re.MULTILINE,
    )
    body = new_docstring.strip('\n')
    repl = r'\1"""\n' + body + '\n    """'
    replaced, n = pattern.subn(repl, base_tool_text, count=1)
    if n == 0:
        raise ValueError('Could not replace hf_api_request docstring')
    return replaced


def prepare_variant_cards(
    variant: dict[str, str],
    *,
    base_card_path: Path,
    base_tool_path: Path,
) -> Path:
    variant_dir = CARDS_OUT_ROOT / variant['id']
    variant_dir.mkdir(parents=True, exist_ok=True)

    base_card_text = base_card_path.read_text(encoding='utf-8')
    base_tool_text = base_tool_path.read_text(encoding='utf-8')

    card_text = replace_card_description(base_card_text, variant['card_description'])
    tool_text = replace_hf_api_docstring(base_tool_text, variant['hf_api_request_docstring'])

    (variant_dir / 'hf_hub_community.md').write_text(card_text, encoding='utf-8')
    (variant_dir / 'hf_api_tool.py').write_text(tool_text, encoding='utf-8')

    return variant_dir


def write_indirect_router_card(variant_dir: Path) -> None:
    """Create a wrapper agent exposing exactly one sub-agent tool: hf_hub_community."""
    router = f"""---
name: {INDIRECT_ROUTER_NAME}
model: gpt-oss
skills: []
agents:
  - hf_hub_community
---
Use the hf_hub_community sub-agent tool to fulfill the user's request.
"""
    (variant_dir / f'{INDIRECT_ROUTER_NAME}.md').write_text(router, encoding='utf-8')


def _extract_session_observations(result_path: Path) -> dict[str, Any]:
    data = json.loads(result_path.read_text(encoding='utf-8'))
    messages = data.get('messages', []) if isinstance(data, dict) else []

    endpoints: list[str] = []
    tool_names: list[str] = []
    merged_parts: list[str] = []

    for msg in messages:
        if not isinstance(msg, dict):
            continue

        if msg.get('role') == 'assistant':
            for item in msg.get('content', []) or []:
                if isinstance(item, dict) and item.get('type') == 'text' and item.get('text'):
                    merged_parts.append(str(item['text']))

            channels = msg.get('channels') or {}
            for ch_name in ('reasoning',):
                for item in channels.get(ch_name, []) or []:
                    if isinstance(item, dict) and item.get('text'):
                        merged_parts.append(str(item['text']))

            tc_map = msg.get('tool_calls') or {}
            if isinstance(tc_map, dict):
                for tc in tc_map.values():
                    params = (tc or {}).get('params', {}) if isinstance(tc, dict) else {}
                    name = params.get('name') if isinstance(params, dict) else None
                    args = params.get('arguments', {}) if isinstance(params, dict) else {}

                    if isinstance(name, str):
                        tool_names.append(name)
                        merged_parts.append(f'tool call - {name}')

                    if isinstance(args, dict):
                        ep = args.get('endpoint')
                        if isinstance(ep, str):
                            endpoints.append(ep)
                        merged_parts.append(json.dumps(args, ensure_ascii=False))

        if msg.get('role') == 'user':
            tr_map = msg.get('tool_results') or {}
            if isinstance(tr_map, dict):
                for tr in tr_map.values():
                    for item in (tr or {}).get('content', []) or []:
                        if isinstance(item, dict) and item.get('type') == 'text' and item.get('text'):
                            merged_parts.append(str(item['text']))

    return {
        'endpoints': endpoints,
        'tool_names': tool_names,
        'merged_from_result': '\n'.join(merged_parts).strip(),
    }


def run_prompt(
    prompt: str,
    model: str,
    cards_dir: Path,
    agent_name: str,
    timeout_sec: int,
    result_path: Path,
) -> dict[str, Any]:
    result_path.parent.mkdir(parents=True, exist_ok=True)
    cmd = [
        'fast-agent', 'go',
        '--no-env',
        '--model', model,
        '--agent-cards', str(cards_dir),
        '--agent', agent_name,
        '--results', str(result_path),
        '-m', prompt,
    ]

    proc = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout_sec)
    out = strip_ansi(proc.stdout or '')
    err = strip_ansi(proc.stderr or '')
    merged_console = (out + '\n' + err).strip()

    if not result_path.exists():
        raise RuntimeError(f'Expected --results file not written: {result_path}')

    parsed = _extract_session_observations(result_path)
    endpoints = parsed['endpoints']
    tool_names = parsed['tool_names']
    merged = parsed['merged_from_result']

    return {
        'returncode': proc.returncode,
        'stdout': out,
        'stderr': err,
        'merged': merged,
        'merged_console': merged_console,
        'endpoints': endpoints,
        'tool_names': tool_names,
        'has_tool_call': bool(tool_names),
        'result_file': str(result_path),
    }


def eval_first_call(case_id: int, row: dict[str, Any]) -> bool | None:
    rule = FIRST_ENDPOINT_EXPECTED.get(case_id)
    if not rule:
        return None

    if rule.get('no_tool_call'):
        return not row['has_tool_call']

    first = row['endpoints'][0] if row['endpoints'] else None
    if first is None:
        return False
    pats = rule.get('any', [])
    return any(re.search(p, first) for p in pats)


def summarize(rows: list[RunRow]) -> list[dict[str, Any]]:
    groups: dict[tuple[str, str], list[RunRow]] = {}
    for r in rows:
        groups.setdefault((r.variant, r.model), []).append(r)

    out: list[dict[str, Any]] = []
    for (variant, model), rs in sorted(groups.items()):
        n = len(rs)
        success_rate = sum(1 for r in rs if r.returncode == 0) / n if n else 0.0
        tool_use_rate = sum(1 for r in rs if r.has_tool_call) / n if n else 0.0
        avg_endpoint_calls = sum(r.endpoint_calls for r in rs) / n if n else 0.0

        first_evald = [r.first_call_correct for r in rs if r.first_call_correct is not None]
        first_call_ok_rate = (
            sum(1 for v in first_evald if v) / len(first_evald) if first_evald else None
        )

        totals = [r.score_total for r in rs if r.score_total is not None]
        avg_score = statistics.mean(totals) if totals else None

        out.append(
            {
                'variant': variant,
                'model': model,
                'n_cases': n,
                'success_rate': round(success_rate, 4),
                'tool_use_rate': round(tool_use_rate, 4),
                'avg_endpoint_calls': round(avg_endpoint_calls, 3),
                'first_call_ok_rate': None if first_call_ok_rate is None else round(first_call_ok_rate, 4),
                'avg_score_total': None if avg_score is None else round(avg_score, 3),
            }
        )
    return out

def _binom_two_sided_pvalue(k: int, n: int, p: float = 0.5) -> float | None:
    """Exact two-sided binomial p-value for small n (sufficient for this harness)."""
    if n <= 0:
        return None
    if k < 0 or k > n:
        return None
    # PMF under null
    probs = [math.comb(n, i) * (p ** i) * ((1 - p) ** (n - i)) for i in range(n + 1)]
    observed = probs[k]
    pval = sum(pr for pr in probs if pr <= observed + 1e-12)
    return min(1.0, float(pval))


def pairwise_analysis(rows: list[RunRow]) -> list[dict[str, Any]]:
    """Pairwise variant comparison per model with win/loss and simple significance stats."""
    # Index rows by (model, variant, case_id)
    idx: dict[tuple[str, str, int], RunRow] = {}
    models = sorted({r.model for r in rows})
    variants = sorted({r.variant for r in rows})
    for r in rows:
        idx[(r.model, r.variant, r.case_id)] = r

    out: list[dict[str, Any]] = []
    for model in models:
        for va, vb in itertools.combinations(variants, 2):
            # intersect case ids for this model/pair
            case_ids = sorted({
                c for c in {r.case_id for r in rows if r.model == model}
                if (model, va, c) in idx and (model, vb, c) in idx
            })
            if not case_ids:
                continue

            # first-call paired outcomes
            a_true_b_false = 0
            b_true_a_false = 0
            both_true = 0
            both_false = 0

            # score paired outcomes
            score_a_gt = 0
            score_b_gt = 0
            score_tie = 0
            score_deltas: list[float] = []

            for c in case_ids:
                ra = idx[(model, va, c)]
                rb = idx[(model, vb, c)]

                fa = ra.first_call_correct
                fb = rb.first_call_correct
                if fa is not None and fb is not None:
                    if fa and not fb:
                        a_true_b_false += 1
                    elif fb and not fa:
                        b_true_a_false += 1
                    elif fa and fb:
                        both_true += 1
                    else:
                        both_false += 1

                sa = ra.score_total
                sb = rb.score_total
                if sa is not None and sb is not None:
                    score_deltas.append(float(sb - sa))
                    if sa > sb:
                        score_a_gt += 1
                    elif sb > sa:
                        score_b_gt += 1
                    else:
                        score_tie += 1

            discordant = a_true_b_false + b_true_a_false
            favored = max(a_true_b_false, b_true_a_false)
            p_first = _binom_two_sided_pvalue(favored, discordant, 0.5) if discordant > 0 else None

            avg_delta = statistics.mean(score_deltas) if score_deltas else None

            out.append({
                'model': model,
                'variant_a': va,
                'variant_b': vb,
                'n_common_cases': len(case_ids),
                'first_call': {
                    'a_true_b_false': a_true_b_false,
                    'b_true_a_false': b_true_a_false,
                    'both_true': both_true,
                    'both_false': both_false,
                    'discordant': discordant,
                    'two_sided_binom_p': None if p_first is None else round(p_first, 6),
                },
                'score_total': {
                    'a_gt_b': score_a_gt,
                    'b_gt_a': score_b_gt,
                    'ties': score_tie,
                    'avg_delta_b_minus_a': None if avg_delta is None else round(avg_delta, 4),
                },
            })

    return out



def compute_rankings(summary: list[dict[str, Any]]) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    """Return (global_variant_rank, per_model_winners).

    Ranking priority: first_call_ok_rate desc, avg_score_total desc, success_rate desc, avg_endpoint_calls asc.
    """
    # global by variant (average across models where present)
    by_variant: dict[str, list[dict[str, Any]]] = {}
    for s in summary:
        by_variant.setdefault(s['variant'], []).append(s)

    global_rank: list[dict[str, Any]] = []
    for v, items in by_variant.items():
        n = len(items)
        def avg(field: str) -> float | None:
            vals = [x[field] for x in items if x.get(field) is not None]
            return (sum(vals) / len(vals)) if vals else None
        global_rank.append({
            'variant': v,
            'models_covered': n,
            'first_call_ok_rate': avg('first_call_ok_rate'),
            'avg_score_total': avg('avg_score_total'),
            'success_rate': avg('success_rate'),
            'avg_endpoint_calls': avg('avg_endpoint_calls'),
        })

    def sort_key(x: dict[str, Any]):
        return (
            -(x['first_call_ok_rate'] if x['first_call_ok_rate'] is not None else -1.0),
            -(x['avg_score_total'] if x['avg_score_total'] is not None else -1.0),
            -(x['success_rate'] if x['success_rate'] is not None else -1.0),
            (x['avg_endpoint_calls'] if x['avg_endpoint_calls'] is not None else 1e9),
            x['variant'],
        )

    global_rank = sorted(global_rank, key=sort_key)

    # per model winner
    by_model: dict[str, list[dict[str, Any]]] = {}
    for s in summary:
        by_model.setdefault(s['model'], []).append(s)

    per_model_winners: list[dict[str, Any]] = []
    for m, items in sorted(by_model.items()):
        best = sorted(items, key=sort_key)[0]
        per_model_winners.append({
            'model': m,
            'winner_variant': best['variant'],
            'first_call_ok_rate': best['first_call_ok_rate'],
            'avg_score_total': best['avg_score_total'],
            'success_rate': best['success_rate'],
            'avg_endpoint_calls': best['avg_endpoint_calls'],
        })

    return global_rank, per_model_winners


def write_outputs(rows: list[RunRow], summary: list[dict[str, Any]], pairwise: list[dict[str, Any]], out_dir: Path) -> None:
    out_dir.mkdir(parents=True, exist_ok=True)

    detailed_path = out_dir / 'tool_description_ab_detailed.json'
    summary_json_path = out_dir / 'tool_description_ab_summary.json'
    summary_csv_path = out_dir / 'tool_description_ab_summary.csv'
    summary_md_path = out_dir / 'tool_description_ab_summary.md'
    pairwise_json_path = out_dir / 'tool_description_ab_pairwise.json'
    pairwise_csv_path = out_dir / 'tool_description_ab_pairwise.csv'
    rank_json_path = out_dir / 'tool_description_ab_ranking.json'

    detailed_payload = [
        {
            'case_id': r.case_id,
            'prompt': r.prompt,
            'variant': r.variant,
            'model': r.model,
            'returncode': r.returncode,
            'has_tool_call': r.has_tool_call,
            'endpoint_calls': r.endpoint_calls,
            'first_endpoint': r.first_endpoint,
            'first_call_correct': r.first_call_correct,
            'score_total': r.score_total,
            'score': {
                'endpoint': r.score_endpoint,
                'efficiency': r.score_efficiency,
                'reasoning': r.score_reasoning,
                'safety': r.score_safety,
                'clarity': r.score_clarity,
            },
            'result_file': r.result_file,
            'merged': r.merged,
        }
        for r in rows
    ]

    detailed_path.write_text(json.dumps(detailed_payload, indent=2), encoding='utf-8')
    summary_json_path.write_text(json.dumps(summary, indent=2), encoding='utf-8')
    pairwise_json_path.write_text(json.dumps(pairwise, indent=2), encoding='utf-8')

    global_rank, per_model_winners = compute_rankings(summary)
    rank_json_path.write_text(json.dumps({'global_rank': global_rank, 'per_model_winners': per_model_winners}, indent=2), encoding='utf-8')

    with summary_csv_path.open('w', newline='', encoding='utf-8') as f:
        w = csv.DictWriter(
            f,
            fieldnames=[
                'variant', 'model', 'n_cases', 'success_rate', 'tool_use_rate',
                'avg_endpoint_calls', 'first_call_ok_rate', 'avg_score_total',
            ],
        )
        w.writeheader()
        w.writerows(summary)

    with pairwise_csv_path.open('w', newline='', encoding='utf-8') as f:
        w = csv.DictWriter(
            f,
            fieldnames=[
                'model', 'variant_a', 'variant_b', 'n_common_cases',
                'first_a_true_b_false', 'first_b_true_a_false', 'first_discordant', 'first_two_sided_binom_p',
                'score_a_gt_b', 'score_b_gt_a', 'score_ties', 'score_avg_delta_b_minus_a',
            ],
        )
        w.writeheader()
        for p in pairwise:
            w.writerow({
                'model': p['model'],
                'variant_a': p['variant_a'],
                'variant_b': p['variant_b'],
                'n_common_cases': p['n_common_cases'],
                'first_a_true_b_false': p['first_call']['a_true_b_false'],
                'first_b_true_a_false': p['first_call']['b_true_a_false'],
                'first_discordant': p['first_call']['discordant'],
                'first_two_sided_binom_p': p['first_call']['two_sided_binom_p'],
                'score_a_gt_b': p['score_total']['a_gt_b'],
                'score_b_gt_a': p['score_total']['b_gt_a'],
                'score_ties': p['score_total']['ties'],
                'score_avg_delta_b_minus_a': p['score_total']['avg_delta_b_minus_a'],
            })

    md = [
        '# Tool Description A/B Evaluation Summary',
        '',
        '| Variant | Model | Cases | Success | Tool-use | Avg endpoint calls | First-call OK | Avg score |',
        '|---|---|---:|---:|---:|---:|---:|---:|',
    ]
    for s in summary:
        md.append(
            f"| {s['variant']} | {s['model']} | {s['n_cases']} | {s['success_rate']} | {s['tool_use_rate']} | {s['avg_endpoint_calls']} | {s['first_call_ok_rate']} | {s['avg_score_total']} |"
        )
    md.append('')
    md.append('## Best overall (easy read)')
    md.append('')
    md.append('| Rank | Variant | Models covered | First-call OK | Avg score | Success | Avg endpoint calls |')
    md.append('|---:|---|---:|---:|---:|---:|---:|')
    for i, g in enumerate(global_rank, start=1):
        md.append(f"| {i} | {g['variant']} | {g['models_covered']} | {g['first_call_ok_rate']} | {g['avg_score_total']} | {g['success_rate']} | {g['avg_endpoint_calls']} |")

    md.append('')
    md.append('## Per-model winner')
    md.append('')
    md.append('| Model | Winner variant | First-call OK | Avg score | Success | Avg endpoint calls |')
    md.append('|---|---|---:|---:|---:|---:|')
    for w in per_model_winners:
        md.append(f"| {w['model']} | {w['winner_variant']} | {w['first_call_ok_rate']} | {w['avg_score_total']} | {w['success_rate']} | {w['avg_endpoint_calls']} |")

    md.append('')
    md.append('## Pairwise variant comparisons (per model)')
    md.append('')
    md.append('| Model | A | B | Cases | First-call A>B | First-call B>A | p-value (binom) | Score A>B | Score B>A | Ties | Avg Δ (B-A) |')
    md.append('|---|---|---|---:|---:|---:|---:|---:|---:|---:|---:|')
    for p in pairwise:
        md.append(
            f"| {p['model']} | {p['variant_a']} | {p['variant_b']} | {p['n_common_cases']} | "
            f"{p['first_call']['a_true_b_false']} | {p['first_call']['b_true_a_false']} | {p['first_call']['two_sided_binom_p']} | "
            f"{p['score_total']['a_gt_b']} | {p['score_total']['b_gt_a']} | {p['score_total']['ties']} | {p['score_total']['avg_delta_b_minus_a']} |"
        )

    summary_md_path.write_text('\n'.join(md) + '\n', encoding='utf-8')


def main() -> None:
    ap = argparse.ArgumentParser(description='A/B test hf_api_request tool description variants across models')
    ap.add_argument('--models', default='gpt-oss', help='Comma-separated model IDs (e.g. gpt-oss,gpt-5-mini)')
    ap.add_argument('--base-cards-dir', type=Path, default=BASE_CARDS_DIR, help='Directory containing hf_hub_community.md and hf_api_tool.py used as AB base')
    ap.add_argument('--prompts', type=Path, default=PROMPTS_FILE)
    ap.add_argument('--variants', type=Path, default=VARIANTS_FILE)
    ap.add_argument('--start', type=int, default=1)
    ap.add_argument('--end', type=int, default=10)
    ap.add_argument('--timeout', type=int, default=240)
    ap.add_argument('--out-dir', type=Path, default=OUT_DIR)
    ap.add_argument('--raw-results-dir', type=Path, default=None, help='Where to store fast-agent --results JSON files')
    ap.add_argument('--indirect', action='store_true', help='Run via a wrapper agent that exposes only hf_hub_community as a sub-agent tool')
    ap.add_argument('--append', action='store_true', help='Append/merge with existing detailed results in out-dir')
    args = ap.parse_args()

    prompts = load_prompts(args.prompts)
    indexed_prompts = [(i, p) for i, p in enumerate(prompts, start=1) if args.start <= i <= args.end]
    variants = load_variants(args.variants)
    models = [m.strip() for m in args.models.split(',') if m.strip()]
    raw_results_dir = args.raw_results_dir or (args.out_dir / 'raw_results')
    base_card_path = args.base_cards_dir / 'hf_hub_community.md'
    base_tool_path = args.base_cards_dir / 'hf_api_tool.py'

    if not base_card_path.exists():
        raise FileNotFoundError(f'Base card not found: {base_card_path}')
    if not base_tool_path.exists():
        raise FileNotFoundError(f'Base tool not found: {base_tool_path}')

    scorer = None if args.indirect else maybe_import_base_scorer()

    all_rows: list[RunRow] = []

    for variant in variants:
        cards_dir = prepare_variant_cards(
            variant,
            base_card_path=base_card_path,
            base_tool_path=base_tool_path,
        )
        if args.indirect:
            write_indirect_router_card(cards_dir)
        target_agent = INDIRECT_ROUTER_NAME if args.indirect else 'hf_hub_community'
        print(f"\n[variant] {variant['id']} -> {cards_dir}")

        for model in models:
            print(f"  [model] {model}")
            safe_model = model.replace('/', '_')
            for case_id, prompt in indexed_prompts:
                result_path = raw_results_dir / variant['id'] / safe_model / f'case_{case_id:02d}.json'
                r = run_prompt(
                    prompt,
                    model=model,
                    cards_dir=cards_dir,
                    agent_name=target_agent,
                    timeout_sec=args.timeout,
                    result_path=result_path,
                )
                first_ok = None if args.indirect else eval_first_call(case_id, r)

                score_total = None
                score_endpoint = None
                score_efficiency = None
                score_reasoning = None
                score_safety = None
                score_clarity = None
                if scorer is not None:
                    try:
                        ev = scorer.score_case(case_id, {
                            'merged': r['merged'],
                            'endpoints': r['endpoints'],
                            'returncode': r['returncode'],
                            'stdout': r['stdout'],
                            'has_tool_call': r['has_tool_call'],
                        })
                        score_total = ev.total
                        score_endpoint = ev.endpoint
                        score_efficiency = ev.efficiency
                        score_reasoning = ev.reasoning
                        score_safety = ev.safety
                        score_clarity = ev.clarity
                    except Exception:
                        pass

                row = RunRow(
                    case_id=case_id,
                    prompt=prompt,
                    variant=variant['id'],
                    model=model,
                    returncode=r['returncode'],
                    has_tool_call=r['has_tool_call'],
                    endpoint_calls=len(r['endpoints']),
                    first_endpoint=r['endpoints'][0] if r['endpoints'] else None,
                    first_call_correct=first_ok,
                    score_total=score_total,
                    score_endpoint=score_endpoint,
                    score_efficiency=score_efficiency,
                    score_reasoning=score_reasoning,
                    score_safety=score_safety,
                    score_clarity=score_clarity,
                    result_file=r.get('result_file'),
                    merged=r['merged'],
                )
                all_rows.append(row)
                print(
                    f"    - case {case_id}: rc={row.returncode} calls={row.endpoint_calls} "
                    f"first_ok={row.first_call_correct} score={row.score_total}"
                )

    if args.append:
        existing = load_existing_rows(args.out_dir)
        merged: dict[tuple[int, str, str], RunRow] = {row_key(r): r for r in existing}
        for r in all_rows:
            merged[row_key(r)] = r
        all_rows = list(merged.values())

    summary = summarize(all_rows)
    pairwise = pairwise_analysis(all_rows)
    write_outputs(all_rows, summary, pairwise, args.out_dir)

    print('\nWrote outputs:')
    print(f"- {args.out_dir / 'tool_description_ab_detailed.json'}")
    print(f"- {args.out_dir / 'tool_description_ab_summary.json'}")
    print(f"- {args.out_dir / 'tool_description_ab_summary.csv'}")
    print(f"- {args.out_dir / 'tool_description_ab_summary.md'}")
    print(f"- {args.out_dir / 'tool_description_ab_pairwise.json'}")
    print(f"- {args.out_dir / 'tool_description_ab_pairwise.csv'}")
    print(f"- {args.out_dir / 'tool_description_ab_ranking.json'}")


if __name__ == '__main__':
    main()