arborist/bench/batteries/base.py

"""Battery base classes.

A Battery runs a fixture file end-to-end against a specified runtime
(usually arborist's existing surface) and emits a deterministic
:class:`BatteryResult`. Determinism: same fixture digest + same code
digest → identical result (modulo wall-clock fields).
"""

from __future__ import annotations

import hashlib
import json
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
from typing import Any, Iterable, Optional, Protocol


@dataclass
class TaskResult:
    """Per-task outcome inside a battery run."""

    task_id: str
    passed: bool
    detail: dict[str, Any] = field(default_factory=dict)


@dataclass
class BatteryResult:
    """End-to-end battery result; deterministic given fixed inputs."""

    battery: str
    sub_battery: str
    fixture_path: str
    fixture_digest: str
    pass_count: int
    fail_count: int
    metrics: dict[str, float] = field(default_factory=dict)
    per_task: list[TaskResult] = field(default_factory=list)
    runtime_digest: str = ""
    timestamp: int = field(default_factory=lambda: int(time.time()))

    def to_dict(self) -> dict:
        d = asdict(self)
        return d


class Battery(Protocol):
    """One sub-battery (e.g., 5S Syntax)."""

    name: str

    def run(self, fixtures_path: Path) -> BatteryResult:
        ...


def fixture_digest(fixtures_path: Path) -> str:
    """SHA-256 over the raw fixture-file bytes.

    Used as the canonical identity of a battery run; bench results
    cite this digest so cross-run comparisons are unambiguous.
    """
    return hashlib.sha256(
        Path(fixtures_path).read_bytes()
    ).hexdigest()


def iter_tasks(fixtures_path: Path) -> Iterable[dict]:
    """Yield non-meta task dicts from a JSONL fixture file."""
    with open(fixtures_path, "r", encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            obj = json.loads(line)
            if "_meta" in obj:
                continue
            yield obj


def fixture_meta(fixtures_path: Path) -> dict:
    """Return the ``_meta`` dict if present at the top of the JSONL."""
    with open(fixtures_path, "r", encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            obj = json.loads(line)
            if "_meta" in obj:
                return obj["_meta"]
            return {}
    return {}


# Phase-1 carrier whitelist (per tickets #000023/#000024/#000025).
# Tasks lacking ``carrier`` default to "text" for backward compat with
# Phase-1a fixtures (syntax-v1, semantics-v1, transfer-v1) — those
# digests stay pinned and the carrier field is purely additive.
PHASE_1_CARRIERS = frozenset({
    "text",
    "claim_lattice",
    "prose",
    "memory_snapshot",
    "selfmodel_snapshot",
    "providence_record",
    "audit_event",
    "verifier_strategies",
    "propositional_logic",
    "relation_graph",
    "qa_answer",
    "capability_transition",
    "memory_root",
    # First non-text carrier, landed via code-py-ast@v1 graduation.
    "code",
    # Math π*'s — arithmetic@v1 + logic-kernel@v1 graduations
    # (SQD §14.1 + §14.3).
    "arithmetic",
    "logic",
    # Sensor / temporal-signal carrier — time-series-quantized@v1.
    "time_series",
})


def validate_carrier(task: dict) -> Optional[str]:
    """Validate the task's ``carrier`` field against the Phase-1 whitelist.

    Returns ``None`` if the carrier is allowed (or absent — defaults
    to ``"text"`` for backward compat). Returns a reason string if
    the carrier is set to something Phase-1 doesn't support, so the
    caller can mark the task ``failed`` with
    ``reason="unsupported_carrier"`` per the cross-modality
    discipline.
    """
    carrier = task.get("carrier")
    if carrier is None:
        # Phase-1a fixtures have no carrier field; default to "text".
        return None
    if not isinstance(carrier, str):
        return f"unsupported_carrier: non-string carrier {carrier!r}"
    if carrier in PHASE_1_CARRIERS:
        return None
    return f"unsupported_carrier: {carrier!r} not in Phase-1 whitelist"