The harness

"""Load config/lab.toml: model registry, run settings, paths. The registry maps a model alias to an OpenAI-compatible endpoint. A modelwith base_url "mock:" is served by the built-in mock client (used by thesmoke test so the pipeline runs without live inference). API keys are neverstored in config — `api_key_env` names an environment variable.""" from __future__ import annotations import osimport tomllibfrom dataclasses import dataclass, fieldfrom pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parentDEFAULT_CONFIG = REPO_ROOT / "config" / "lab.toml"  @dataclass(frozen=True)class Model:    alias: str    base_url: str    model: str    temperature: float = 0.0    max_tokens: int = 8192    api_key_env: str | None = None     @property    def is_mock(self) -> bool:        return self.base_url.startswith("mock:")     def api_key(self) -> str | None:        if not self.api_key_env:            return None        key = os.environ.get(self.api_key_env)        if not key:            raise RuntimeError(                f"model {self.alias!r} needs the {self.api_key_env} environment variable"            )        return key  @dataclass(frozen=True)class RunSettings:    samples: int = 1    modes: tuple[str, ...] = ("oneshot", "agentic")    max_agent_turns: int = 24    test_timeout_secs: int = 120    request_retries: int = 3  @dataclass(frozen=True)class Config:    models: dict[str, Model] = field(default_factory=dict)    run: RunSettings = field(default_factory=RunSettings)    raw: dict = field(default_factory=dict)  def load(path: Path | str = DEFAULT_CONFIG) -> Config:    raw = tomllib.loads(Path(path).read_text())    models = {}    for alias, entry in raw.get("models", {}).items():        if "base_url" not in entry:            raise ValueError(                f"[models.{alias}] has no base_url — if the alias contains a dot, "                f'quote it: [models."{alias}"]'            )        models[alias] = Model(            alias=alias,            base_url=entry["base_url"],            model=entry.get("model", alias),            temperature=entry.get("temperature", 0.0),            max_tokens=entry.get("max_tokens", 8192),            api_key_env=entry.get("api_key_env"),        )    run_raw = raw.get("run", {})    run = RunSettings(        samples=run_raw.get("samples", 1),        modes=tuple(run_raw.get("modes", ["oneshot", "agentic"])),        max_agent_turns=run_raw.get("max_agent_turns", 24),        test_timeout_secs=run_raw.get("test_timeout_secs", 120),        request_retries=run_raw.get("request_retries", 3),    )    return Config(models=models, run=run, raw=raw)

"""OpenAI-compatible chat client (urllib, no deps) with degenerate-output handling. Lesson from llmcensor: empty or truncated model output must be a retryableerror, never a silent result. `chat()` retries transport errors anddegenerate completions, then raises DegenerateOutput so the runner recordsan errored result instead of a fake score. Mock models (base_url "mock:") return canned deterministic responses via ahandler registered by the caller — the smoke test registers one that solvesthe sample challenge.""" from __future__ import annotations import jsonimport timeimport urllib.errorimport urllib.requestfrom dataclasses import dataclassfrom typing import Callable from .config import Model  class DegenerateOutput(RuntimeError):    """Model returned empty/whitespace output or was truncated, after retries."""  class TransportError(RuntimeError):    """Endpoint unreachable or returned a non-JSON / error response, after retries."""  @dataclassclass Completion:    content: str    finish_reason: str    prompt_tokens: int    completion_tokens: int  MockHandler = Callable[[Model, list[dict]], str]_mock_handlers: dict[str, MockHandler] = {}  def register_mock(name: str, handler: MockHandler) -> None:    _mock_handlers[name] = handler  def chat(model: Model, messages: list[dict], retries: int = 3, timeout: int = 300) -> Completion:    last_err: Exception | None = None    for attempt in range(retries):        if attempt:            time.sleep(min(2 ** attempt, 30))        try:            completion = _request(model, messages, timeout)        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, OSError) as err:            last_err = TransportError(f"{model.alias}: {err}")            continue        if not completion.content.strip():            last_err = DegenerateOutput(                f"{model.alias}: empty completion (finish_reason={completion.finish_reason})"            )            continue        if completion.finish_reason == "length":            last_err = DegenerateOutput(                f"{model.alias}: truncated at max_tokens={model.max_tokens}"            )            continue        return completion    raise last_err if last_err is not None else TransportError(f"{model.alias}: no attempts made")  def _request(model: Model, messages: list[dict], timeout: int) -> Completion:    if model.is_mock:        name = model.base_url.removeprefix("mock:")        handler = _mock_handlers.get(name)        if handler is None:            raise KeyError(f"no mock handler registered for {name!r}")        return Completion(handler(model, messages), "stop", 0, 0)     payload = {        "model": model.model,        "messages": messages,        "temperature": model.temperature,        "max_tokens": model.max_tokens,    }    headers = {"Content-Type": "application/json"}    key = model.api_key()    if key:        headers["Authorization"] = f"Bearer {key}"    req = urllib.request.Request(        model.base_url.rstrip("/") + "/chat/completions",        data=json.dumps(payload).encode(),        headers=headers,    )    with urllib.request.urlopen(req, timeout=timeout) as resp:        body = json.loads(resp.read().decode())    choice = body["choices"][0]    usage = body.get("usage", {})    return Completion(        content=choice["message"].get("content") or "",        finish_reason=choice.get("finish_reason") or "stop",        prompt_tokens=usage.get("prompt_tokens", 0),        completion_tokens=usage.get("completion_tokens", 0),    )

"""Challenge loading, validation, and challenge-set hashing. A challenge is one JSON file in challenges/:     {      "id": "kebab-case-id",      "title": "...",      "spec": "prompt text given to the model",      "starter_files": {"path": "content", ...},        # optional      "tests": {"test_x.py": "pytest content", ...},      "expected_files": ["solution.py", ...],           # deliverables      "content_checks": {"path": ["substring", ...]},   # optional      "reference_solution": {"path": "content", ...}    # must pass tests    } Results are only comparable within one challenge-set version, so a sha256over the sorted ids + specs + tests is stamped onto every run.""" from __future__ import annotations import hashlibimport jsonfrom dataclasses import dataclass, fieldfrom pathlib import Path from .config import REPO_ROOT CHALLENGES_DIR = REPO_ROOT / "challenges"REQUIRED_KEYS = {"id", "title", "spec", "tests", "expected_files", "reference_solution"}  @dataclass(frozen=True)class Challenge:    id: str    title: str    spec: str    tests: dict[str, str]    expected_files: list[str]    starter_files: dict[str, str] = field(default_factory=dict)    content_checks: dict[str, list[str]] = field(default_factory=dict)    reference_solution: dict[str, str] = field(default_factory=dict)  def load_challenge(path: Path) -> Challenge:    raw = json.loads(path.read_text())    missing = REQUIRED_KEYS - raw.keys()    if missing:        raise ValueError(f"{path.name}: missing keys {sorted(missing)}")    if raw["id"] != path.stem:        raise ValueError(f"{path.name}: id {raw['id']!r} must match filename")    if not raw["tests"]:        raise ValueError(f"{path.name}: challenge has no tests")    return Challenge(        id=raw["id"],        title=raw["title"],        spec=raw["spec"],        tests=raw["tests"],        expected_files=raw["expected_files"],        starter_files=raw.get("starter_files", {}),        content_checks=raw.get("content_checks", {}),        reference_solution=raw["reference_solution"],    )  def load_all(directory: Path = CHALLENGES_DIR) -> list[Challenge]:    challenges = [load_challenge(p) for p in sorted(directory.glob("*.json"))]    if not challenges:        raise ValueError(f"no challenges found in {directory}")    return challenges  def set_hash(challenges: list[Challenge]) -> str:    h = hashlib.sha256()    for c in sorted(challenges, key=lambda c: c.id):        h.update(c.id.encode())        h.update(c.spec.encode())        h.update(json.dumps(c.tests, sort_keys=True).encode())    return h.hexdigest()[:16]

"""Isolated per-attempt workspaces and sandboxed pytest execution.""" from __future__ import annotations import reimport subprocessimport sysfrom dataclasses import dataclassfrom pathlib import Path from .challenges import Challenge  @dataclassclass TestOutcome:    __test__ = False  # keep pytest from collecting this as a test class    passed: int    failed: int    errored: bool    output: str     @property    def total(self) -> int:        return self.passed + self.failed     @property    def green(self) -> bool:        return not self.errored and self.failed == 0 and self.passed > 0  def setup(root: Path, challenge: Challenge, files: dict[str, str] | None = None) -> Path:    """Create a workspace seeded with starter files + tests (+ optional files)."""    root.mkdir(parents=True, exist_ok=True)    for rel, content in {**challenge.starter_files, **challenge.tests, **(files or {})}.items():        write_file(root, rel, content)    return root  def write_file(root: Path, rel: str, content: str) -> Path:    dest = (root / rel).resolve()    if not dest.is_relative_to(root.resolve()):        raise ValueError(f"path escapes workspace: {rel}")    dest.parent.mkdir(parents=True, exist_ok=True)    dest.write_text(content)    return dest  def run_tests(root: Path, timeout: int = 120) -> TestOutcome:    try:        proc = subprocess.run(            [sys.executable, "-m", "pytest", "-q", "--tb=short", "-p", "no:cacheprovider"],            cwd=root,            capture_output=True,            text=True,            timeout=timeout,        )    except subprocess.TimeoutExpired as err:        return TestOutcome(0, 0, True, f"pytest timed out after {timeout}s\n{err.stdout or ''}")    output = proc.stdout + proc.stderr    passed = _summary_count(output, "passed")    failed = _summary_count(output, "failed") + _summary_count(output, "error")    # Exit codes: 0 ok, 1 failures; anything else (2 usage, 3 internal, 4 no tests) is an error.    errored = proc.returncode not in (0, 1)    return TestOutcome(passed, failed, errored, output[-8000:])  def _summary_count(output: str, word: str) -> int:    match = re.search(rf"(\d+) {word}", output)    return int(match.group(1)) if match else 0

"""Single-shot mode: one prompt, model returns files, we grade what it wrote. The model answers with one JSON object {"files": {"path": "content", ...}}(optionally inside a ```json fence). No feedback loop, no test access.""" from __future__ import annotations import jsonimport refrom pathlib import Path from . import client, workspacefrom .challenges import Challengefrom .config import Modelfrom .session import Session PROMPT = """You are completing a coding task in one shot. Write complete files that satisfy the spec below. Tests (pytest) will be run against your files; you cannot run them yourself. Reply with ONE JSON object and nothing else:{{"files": {{"<relative path>": "<complete file content>", ...}}}} Required deliverable files: {expected} SPEC:{spec}{starter}"""  def run(model: Model, challenge: Challenge, root: Path, session: Session, retries: int = 3) -> None:    starter = ""    if challenge.starter_files:        listing = "\n".join(            f"--- {rel} ---\n{content}" for rel, content in challenge.starter_files.items()        )        starter = f"\nSTARTER FILES (already in the workspace):\n{listing}"    prompt = PROMPT.format(        expected=", ".join(challenge.expected_files),        spec=challenge.spec,        starter=starter,    )    session.log("prompt", mode="oneshot", content=prompt)    completion = client.chat(model, [{"role": "user", "content": prompt}], retries=retries)    session.log("completion", content=completion.content,                tokens=completion.completion_tokens,                tokens_in=completion.prompt_tokens)     files = _parse_files(completion.content)    for rel, content in files.items():        workspace.write_file(root, rel, content)        session.log("write_file", path=rel, bytes=len(content))  def _parse_files(content: str) -> dict[str, str]:    fenced = re.search(r"```(?:json)?\s*(\{.*)```", content, re.DOTALL)    text = fenced.group(1) if fenced else content    start = text.find("{")    if start < 0:        raise client.DegenerateOutput("one-shot reply contains no JSON object")    try:        obj, _ = json.JSONDecoder().raw_decode(text[start:])    except json.JSONDecodeError as err:        raise client.DegenerateOutput(f"one-shot reply is not valid JSON: {err}") from err    files = obj.get("files")    if not isinstance(files, dict) or not files:        raise client.DegenerateOutput("one-shot reply JSON has no files")    return {str(k): str(v) for k, v in files.items()}

"""Agentic mode: a bounded tool loop tuned for small models. One JSON action per turn (pclip's protocol, rebuilt fresh):     {"action": "list_files"}    {"action": "read_file",  "path": "..."}    {"action": "write_file", "path": "...", "content": "..."}    {"action": "run_tests"}    {"action": "done"} `done` is gated on a green test run; the gate result is fed back so themodel can keep repairing until the turn budget runs out.""" from __future__ import annotations import jsonimport refrom pathlib import Path from . import client, workspacefrom .challenges import Challengefrom .config import Modelfrom .session import Session SYSTEM = """You are a coding agent working in a workspace directory. Complete the SPEC by writing files and running the pytest suite until it passes. Each turn, reply with ONE JSON object and nothing else:  {{"action": "list_files"}}  {{"action": "read_file", "path": "<relative path>"}}  {{"action": "write_file", "path": "<relative path>", "content": "<complete file content>"}}  {{"action": "run_tests"}}  {{"action": "done"}} Rules: write COMPLETE file contents (no diffs, no placeholders). Use run_tests to check your work. Only reply "done" after run_tests passes. Required deliverable files: {expected} SPEC:{spec}"""  def run(model: Model, challenge: Challenge, root: Path, session: Session,        max_turns: int = 24, test_timeout: int = 120, retries: int = 3) -> None:    messages = [        {"role": "system", "content": SYSTEM.format(            expected=", ".join(challenge.expected_files), spec=challenge.spec)},        {"role": "user", "content": "Workspace ready. Begin."},    ]    for turn in range(1, max_turns + 1):        completion = client.chat(model, messages, retries=retries)        session.log("completion", turn=turn, tokens=completion.completion_tokens,                    tokens_in=completion.prompt_tokens)        messages.append({"role": "assistant", "content": completion.content})        try:            action = _parse_action(completion.content)        except ValueError as err:            session.log("bad_action", turn=turn, error=str(err), content=completion.content[:2000])            messages.append({"role": "user", "content": f"Invalid action: {err}. Reply with one JSON action object."})            continue         session.log("action", turn=turn, **{k: v for k, v in action.items() if k != "content"},                    content_bytes=len(action.get("content", "")))        kind = action["action"]         if kind == "done":            outcome = workspace.run_tests(root, timeout=test_timeout)            session.log("done_gate", green=outcome.green, passed=outcome.passed, failed=outcome.failed)            if outcome.green:                return            messages.append({"role": "user", "content":                             f"Not done — tests are not green:\n{outcome.output[-3000:]}\nKeep working."})            continue         result = _execute(kind, action, root, challenge, test_timeout, session)        messages.append({"role": "user", "content": result})     session.log("turn_budget_exhausted", max_turns=max_turns)  def _execute(kind: str, action: dict, root: Path, challenge: Challenge,             test_timeout: int, session: Session) -> str:    if kind == "list_files":        files = sorted(str(p.relative_to(root)) for p in root.rglob("*")                       if p.is_file() and "__pycache__" not in p.parts)        return "Files:\n" + "\n".join(files)    if kind == "read_file":        path = root / action["path"]        if not path.is_file():            return f"No such file: {action['path']}"        return path.read_text()[:20000]    if kind == "write_file":        try:            workspace.write_file(root, action["path"], action["content"])        except ValueError as err:            return str(err)        return f"Wrote {action['path']} ({len(action['content'])} bytes)."    if kind == "run_tests":        outcome = workspace.run_tests(root, timeout=test_timeout)        session.log("run_tests", passed=outcome.passed, failed=outcome.failed, errored=outcome.errored)        return f"pytest: {outcome.passed} passed, {outcome.failed} failed\n{outcome.output[-3000:]}"    return f"Unknown action {kind!r}."  def _parse_action(content: str) -> dict:    fenced = re.search(r"```(?:json)?\s*(\{.*?)```", content, re.DOTALL)    text = fenced.group(1) if fenced else content    start = text.find("{")    if start < 0:        raise ValueError("no JSON object found")    try:        obj, _ = json.JSONDecoder().raw_decode(text[start:])    except json.JSONDecodeError as err:        raise ValueError(f"invalid JSON: {err}") from err    kind = obj.get("action")    if kind not in {"list_files", "read_file", "write_file", "run_tests", "done"}:        raise ValueError(f"unknown action {kind!r}")    if kind in {"read_file", "write_file"} and not obj.get("path"):        raise ValueError(f"{kind} requires a path")    if kind == "write_file" and "content" not in obj:        raise ValueError("write_file requires content")    return obj

"""Mechanical grading: tests 60%, deliverables 20%, content checks 20%. Axes a challenge doesn't use are reweighted away, so a challenge with nocontent_checks is graded 75/25 tests/deliverables. The blend is computedhere, server-side, and stored — the site only ever displays stored numbers.""" from __future__ import annotations from dataclasses import dataclassfrom pathlib import Path from .challenges import Challengefrom .workspace import TestOutcome WEIGHTS = {"tests": 0.6, "deliverables": 0.2, "content": 0.2}  @dataclass(frozen=True)class Grade:    score_tests: float        # 0..1    score_deliverables: float    score_content: float | None  # None when the challenge has no content checks    total: float              # 0..100 blended  def grade(challenge: Challenge, root: Path, tests: TestOutcome) -> Grade:    score_tests = tests.passed / tests.total if tests.total else 0.0     present = sum(1 for rel in challenge.expected_files if (root / rel).is_file())    score_deliverables = present / len(challenge.expected_files)     score_content: float | None = None    if challenge.content_checks:        checks = [            (root / rel, needle)            for rel, needles in challenge.content_checks.items()            for needle in needles        ]        hits = sum(1 for path, needle in checks if path.is_file() and needle in path.read_text())        score_content = hits / len(checks)     axes = {"tests": score_tests, "deliverables": score_deliverables}    if score_content is not None:        axes["content"] = score_content    weight_sum = sum(WEIGHTS[name] for name in axes)    total = 100 * sum(WEIGHTS[name] * value for name, value in axes.items()) / weight_sum     return Grade(score_tests, score_deliverables, score_content, round(total, 1))

"""Sequential runner: model × challenge × mode × sample, resumable. Sequential by design — one GPU, and model swaps thrash memory. Everyattempt gets an isolated workspace and a JSONL session file; every resultis committed to SQLite as it lands. Degenerate output and transportfailures are recorded as errored results, never as scores.""" from __future__ import annotations import timeimport tracebackfrom pathlib import Path from . import agent, client, db, grade as grading, oneshot, workspacefrom .challenges import Challenge, load_all, set_hashfrom .config import REPO_ROOT, Configfrom .session import Session RUNS_DIR = REPO_ROOT / ".lab" / "runs"  def run(config: Config, models: list[str] | None = None,        challenge_ids: list[str] | None = None, note: str = "",        run_id: int | None = None, db_path: Path | None = None,        runs_dir: Path | None = None) -> int:    """Execute (or resume, via run_id) a full run. Returns the run id."""    challenges = load_all()    if challenge_ids:        challenges = [c for c in challenges if c.id in challenge_ids]    roster = [config.models[m] for m in (models or sorted(config.models))]    conn = db.connect(db_path) if db_path else db.connect()    runs_dir = runs_dir or RUNS_DIR     if run_id is None:        run_id = db.create_run(conn, set_hash(challenges), config.raw, note)    print(f"run {run_id}: {len(roster)} models × {len(challenges)} challenges × "          f"{config.run.modes} × {config.run.samples} samples")     for model in roster:        for challenge in challenges:            for mode in config.run.modes:                for sample in range(config.run.samples):                    if db.has_result(conn, run_id, model.alias, mode, challenge.id, sample):                        continue                    _attempt(conn, run_id, model, challenge, mode, sample, config, runs_dir)    conn.close()    return run_id  def _attempt(conn, run_id: int, model, challenge: Challenge, mode: str,             sample: int, config: Config, runs_dir: Path) -> None:    slug = f"{model.alias}/{mode}/{challenge.id}/s{sample}"    root = runs_dir / str(run_id) / slug    session_path = root / "session.jsonl"    workspace.setup(root, challenge)    started = time.monotonic()    print(f"  {slug} ...", end="", flush=True)     with Session(session_path) as session:        session.log("start", run=run_id, model=model.alias, mode=mode,                    challenge=challenge.id, sample=sample)        try:            if mode == "oneshot":                oneshot.run(model, challenge, root, session, retries=config.run.request_retries)            elif mode == "agentic":                agent.run(model, challenge, root, session,                          max_turns=config.run.max_agent_turns,                          test_timeout=config.run.test_timeout_secs,                          retries=config.run.request_retries)            else:                raise ValueError(f"unknown mode {mode!r}")        except client.DegenerateOutput as err:            session.log("degenerate", error=str(err))            db.record_result(conn, run_id, model.alias, mode, challenge.id, sample,                             status="degenerate", error=str(err),                             wall_secs=time.monotonic() - started,                             session_path=str(session_path))            print(" DEGENERATE")            return        except Exception as err:  # transport errors, bugs — record, keep going            session.log("error", error=str(err), trace=traceback.format_exc()[-4000:])            db.record_result(conn, run_id, model.alias, mode, challenge.id, sample,                             status="error", error=str(err),                             wall_secs=time.monotonic() - started,                             session_path=str(session_path))            print(f" ERROR ({err})")            return         tests = workspace.run_tests(root, timeout=config.run.test_timeout_secs)        result = grading.grade(challenge, root, tests)        session.log("graded", total=result.total, tests_passed=tests.passed,                    tests_failed=tests.failed, output=tests.output[-6000:])        db.record_result(conn, run_id, model.alias, mode, challenge.id, sample,                         status="ok", grade=result, tests=tests,                         wall_secs=time.monotonic() - started,                         session_path=str(session_path))        print(f" {result.total}")

"""JSONL flight recorder — every model call, action, and test run is an event.""" from __future__ import annotations import datetimeimport jsonfrom pathlib import Path  class Session:    def __init__(self, path: Path):        self.path = path        path.parent.mkdir(parents=True, exist_ok=True)        self._fh = path.open("a")     def log(self, event: str, **data) -> None:        record = {            "ts": datetime.datetime.now(datetime.timezone.utc).isoformat(),            "event": event,            **data,        }        self._fh.write(json.dumps(record, ensure_ascii=False) + "\n")        self._fh.flush()     def close(self) -> None:        self._fh.close()     def __enter__(self) -> "Session":        return self     def __exit__(self, *exc) -> None:        self.close()

"""SQLite results store with run stamping and idempotent result keys. Every run records: git sha, engine version, full config snapshot, and thechallenge-set content hash — results are only comparable within one hash.Results are unique on (run_id, model, mode, challenge, sample); the runnerskips tuples that already exist, which is what makes crashed or extendedruns resumable (llmcensor's append pattern).""" from __future__ import annotations import jsonimport sqlite3import subprocessfrom datetime import datetime, timezonefrom pathlib import Path from . import ENGINE_VERSIONfrom .config import REPO_ROOT DB_PATH = REPO_ROOT / ".lab" / "results.sqlite3" SCHEMA = """CREATE TABLE IF NOT EXISTS runs (    id INTEGER PRIMARY KEY,    started_at TEXT NOT NULL,    git_sha TEXT NOT NULL,    engine_version TEXT NOT NULL,    challenge_set_hash TEXT NOT NULL,    config_json TEXT NOT NULL,    note TEXT DEFAULT '');CREATE TABLE IF NOT EXISTS results (    run_id INTEGER NOT NULL REFERENCES runs(id),    model TEXT NOT NULL,    mode TEXT NOT NULL,    challenge TEXT NOT NULL,    sample INTEGER NOT NULL,    status TEXT NOT NULL,              -- ok | error | degenerate    error TEXT,    score_total REAL,    score_tests REAL,    score_deliverables REAL,    score_content REAL,    tests_passed INTEGER,    tests_failed INTEGER,    wall_secs REAL,    session_path TEXT,    finished_at TEXT NOT NULL,    UNIQUE (run_id, model, mode, challenge, sample));"""  def connect(path: Path = DB_PATH) -> sqlite3.Connection:    path.parent.mkdir(parents=True, exist_ok=True)    conn = sqlite3.connect(path)    conn.row_factory = sqlite3.Row    conn.executescript(SCHEMA)    return conn  def git_sha() -> str:    try:        return subprocess.run(            ["git", "rev-parse", "--short", "HEAD"],            cwd=REPO_ROOT, capture_output=True, text=True, check=True,        ).stdout.strip()    except (subprocess.CalledProcessError, FileNotFoundError):        return "unknown"  def create_run(conn: sqlite3.Connection, challenge_set_hash: str,               config_snapshot: dict, note: str = "") -> int:    cur = conn.execute(        "INSERT INTO runs (started_at, git_sha, engine_version, challenge_set_hash, config_json, note)"        " VALUES (?, ?, ?, ?, ?, ?)",        (_now(), git_sha(), ENGINE_VERSION, challenge_set_hash,         json.dumps(config_snapshot, sort_keys=True), note),    )    conn.commit()    return cur.lastrowid  def has_result(conn: sqlite3.Connection, run_id: int, model: str, mode: str,               challenge: str, sample: int) -> bool:    row = conn.execute(        "SELECT 1 FROM results WHERE run_id=? AND model=? AND mode=? AND challenge=? AND sample=?",        (run_id, model, mode, challenge, sample),    ).fetchone()    return row is not None  def record_result(conn: sqlite3.Connection, run_id: int, model: str, mode: str,                  challenge: str, sample: int, *, status: str, error: str | None = None,                  grade=None, tests=None, wall_secs: float = 0.0,                  session_path: str | None = None) -> None:    conn.execute(        "INSERT INTO results (run_id, model, mode, challenge, sample, status, error,"        " score_total, score_tests, score_deliverables, score_content,"        " tests_passed, tests_failed, wall_secs, session_path, finished_at)"        " VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",        (run_id, model, mode, challenge, sample, status, error,         grade.total if grade else None,         grade.score_tests if grade else None,         grade.score_deliverables if grade else None,         grade.score_content if grade else None,         tests.passed if tests else None,         tests.failed if tests else None,         wall_secs, session_path, _now()),    )    conn.commit()  # commit per result: a crash never loses prior work  def _now() -> str:    return datetime.now(timezone.utc).isoformat()

"""Code and session metrics for the engineer reports. Python files get AST-derived metrics (functions, classes, cyclomaticcomplexity, nesting depth, imports); any file gets size metrics. Cyclomaticcomplexity is the classic approximation: 1 per function plus 1 per branchpoint (if/elif/for/while/except/assert/bool-op/comprehension-if/ternary).""" from __future__ import annotations import astfrom dataclasses import dataclass, field _BRANCHES = (ast.If, ast.For, ast.AsyncFor, ast.While, ast.ExceptHandler,             ast.Assert, ast.IfExp)  @dataclassclass FileMetrics:    path: str    loc: int = 0            # physical lines    sloc: int = 0           # non-blank, non-comment lines    chars: int = 0    is_python: bool = False    functions: int = 0    classes: int = 0    complexity: int = 0     # sum over functions (module counts as one unit)    max_complexity: int = 0    max_depth: int = 0      # deepest statement nesting    imports: list[str] = field(default_factory=list)    avg_function_len: float = 0.0    parse_error: bool = False  def file_metrics(path: str, content: str) -> FileMetrics:    lines = content.splitlines()    m = FileMetrics(        path=path,        loc=len(lines),        sloc=sum(1 for l in lines if l.strip() and not l.strip().startswith("#")),        chars=len(content),        is_python=path.endswith(".py"),    )    if not m.is_python:        return m    try:        tree = ast.parse(content)    except SyntaxError:        m.parse_error = True        return m     fn_lens: list[int] = []    for node in ast.walk(tree):        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):            m.functions += 1            fn_complexity = 1 + _branch_count(node)            m.max_complexity = max(m.max_complexity, fn_complexity)            if node.end_lineno:                fn_lens.append(node.end_lineno - node.lineno + 1)        elif isinstance(node, ast.ClassDef):            m.classes += 1        elif isinstance(node, ast.Import):            m.imports += [a.name.split(".")[0] for a in node.names]        elif isinstance(node, ast.ImportFrom) and node.module:            m.imports.append(node.module.split(".")[0])    m.imports = sorted(set(m.imports))    m.complexity = 1 + _branch_count(tree)    m.max_depth = _max_depth(tree)    m.avg_function_len = round(sum(fn_lens) / len(fn_lens), 1) if fn_lens else 0.0    return m  def _branch_count(node: ast.AST) -> int:    count = 0    for child in ast.walk(node):        if isinstance(child, _BRANCHES):            count += 1        elif isinstance(child, ast.BoolOp):            count += len(child.values) - 1        elif isinstance(child, (ast.ListComp, ast.SetComp, ast.DictComp, ast.GeneratorExp)):            count += sum(len(gen.ifs) + 1 for gen in child.generators)    return count  def _max_depth(tree: ast.AST) -> int:    deepest = 0     def walk(node: ast.AST, depth: int) -> None:        nonlocal deepest        for child in ast.iter_child_nodes(node):            if isinstance(child, (ast.If, ast.For, ast.AsyncFor, ast.While, ast.With,                                  ast.AsyncWith, ast.Try, ast.FunctionDef,                                  ast.AsyncFunctionDef, ast.ClassDef)):                deepest = max(deepest, depth + 1)                walk(child, depth + 1)            else:                walk(child, depth)     walk(tree, 0)    return deepest  def aggregate(files: list[FileMetrics]) -> dict:    py = [f for f in files if f.is_python and not f.parse_error]    return {        "files": len(files),        "loc": sum(f.loc for f in files),        "sloc": sum(f.sloc for f in files),        "chars": sum(f.chars for f in files),        "functions": sum(f.functions for f in py),        "classes": sum(f.classes for f in py),        "complexity": sum(f.complexity for f in py),        "max_complexity": max((f.max_complexity for f in py), default=0),        "max_depth": max((f.max_depth for f in py), default=0),        "imports": sorted({i for f in py for i in f.imports}),        "avg_function_len": round(            sum(f.avg_function_len * f.functions for f in py) /            max(1, sum(f.functions for f in py)), 1),        "parse_errors": sum(1 for f in files if f.parse_error),    }  def session_metrics(events: list[dict]) -> dict:    actions: dict[str, int] = {}    for e in events:        if e["event"] == "action":            kind = e.get("action", "?")            actions[kind] = actions.get(kind, 0) + 1    completions = sum(1 for e in events if e["event"] == "completion")    return {        # older sessions have no per-turn completion events; fall back to action counts        "model_calls": completions or sum(            1 for e in events if e["event"] in ("action", "bad_action")),        "actions": actions,        "bad_actions": sum(1 for e in events if e["event"] == "bad_action"),        "test_runs": sum(1 for e in events if e["event"] == "run_tests"),        "tokens_out": sum(e.get("tokens", 0) for e in events if e["event"] == "completion"),        "tokens_in": sum(e.get("tokens_in", 0) for e in events if e["event"] == "completion"),        "turns": max((e.get("turn", 0) for e in events), default=0),    }