Agentswarm/benchmark/fixtures/__init__.py

"""Benchmark task fixtures with HELD-OUT acceptance tests.

A fixture pairs a task objective with authoritative tests that the swarm never sees. These tests
— not the swarm's own testing-agent output — are what produce TestPassRate (Owner ruling: avoid
self-grading; the swarm's own tests are kept as a separate signal in quality.py).

Layout per fixture:  benchmark/fixtures/<id>/fixture.json  +  <id>/tests/test_*.py

`fixture.json` schema:
    {
      "id": "add_function",
      "objective": "...",                 # what the swarm is asked to build
      "required_capabilities": [...],      # used to decide expects_code()
      "entrypoint": "calc.py",             # the module the tests import (informational)
      "target_time_seconds": 60,           # V_speed target for reward()
      "test_files": ["tests/test_add.py"]  # held-out acceptance tests (relative to <id>/)
    }
"""
from __future__ import annotations

import json
from dataclasses import dataclass, field
from pathlib import Path
from typing import List, Optional

_FIXTURE_ROOT = Path(__file__).resolve().parent

# Capabilities that imply the deliverable is runnable/testable code → TestPassRate applies.
CODE_CAPABILITIES = {"python", "code_generation", "coding", "testing", "pytest", "implementation"}


@dataclass
class FixtureTest:
    path: str
    content: str


@dataclass
class Fixture:
    id: str
    objective: str
    required_capabilities: List[str] = field(default_factory=list)
    entrypoint: Optional[str] = None
    target_time_seconds: Optional[float] = None
    test_files: List[FixtureTest] = field(default_factory=list)

    @property
    def expects_code(self) -> bool:
        return expects_code(self.required_capabilities)


def expects_code(required_capabilities) -> bool:
    """A task is code-shaped (TestPassRate applies) if it requires any coding capability."""
    return bool(CODE_CAPABILITIES & {str(c).lower() for c in (required_capabilities or [])})


def fixture_dir(fixture_id: str) -> Path:
    return _FIXTURE_ROOT / fixture_id


def available_fixtures() -> List[str]:
    return sorted(
        p.name for p in _FIXTURE_ROOT.iterdir()
        if p.is_dir() and (p / "fixture.json").exists()
    )


def load_fixture(fixture_id: str) -> Fixture:
    base = fixture_dir(fixture_id)
    meta_path = base / "fixture.json"
    if not meta_path.exists():
        raise FileNotFoundError(f"unknown fixture: {fixture_id}")
    meta = json.loads(meta_path.read_text(encoding="utf-8"))
    tests = []
    for rel in meta.get("test_files", []):
        tpath = base / rel
        if not tpath.exists():
            raise FileNotFoundError(f"fixture {fixture_id} missing test file: {rel}")
        # Flatten to a basename so the sandbox runner (which scans CWD for test_*.py) finds it.
        tests.append(FixtureTest(path=Path(rel).name, content=tpath.read_text(encoding="utf-8")))
    return Fixture(
        id=meta.get("id", fixture_id),
        objective=meta.get("objective", ""),
        required_capabilities=meta.get("required_capabilities", []),
        entrypoint=meta.get("entrypoint"),
        target_time_seconds=meta.get("target_time_seconds"),
        test_files=tests,
    )


def load_reference_solution(fixture_id: str) -> List[FixtureTest]:
    """Offline reference solution files (path/content), used ONLY by the offline backend.

    Returns [] if the fixture has no reference_solution/ dir. These files are NOT the swarm's
    work — they exist so hermetic, key-free runs can validate the benchmark pipeline.
    """
    ref_dir = fixture_dir(fixture_id) / "reference_solution"
    if not ref_dir.is_dir():
        return []
    return [
        FixtureTest(path=p.name, content=p.read_text(encoding="utf-8"))
        for p in sorted(ref_dir.glob("*.py"))
    ]