Building Regression Test Harnesses for Spatial Agents

import hashlib
import json
import logging
import platform
from dataclasses import dataclass, asdict
from typing import Callable, Sequence

log = logging.getLogger("regression_harness")


@dataclass(frozen=True)
class Environment:
    case_set_version: str
    agent_version: str
    tokenizer: str
    geos_version: str
    proj_version: str
    python: str


@dataclass(frozen=True)
class Run:
    environment: Environment
    results: tuple[dict, ...]
    case_set_digest: str


def case_set_digest(cases: Sequence[dict]) -> str:
    """A stable digest of the case set, so a silent edit is detectable."""
    payload = json.dumps(
        [{k: c[k] for k in sorted(c) if k != "notes"} for c in cases],
        sort_keys=True, separators=(",", ":"), default=str)
    return hashlib.sha256(payload.encode()).hexdigest()[:16]


def capture_environment(case_set_version: str, agent_version: str,
                        tokenizer: str) -> Environment:
    """Record what could move a score, before anything runs."""
    try:
        from shapely import geos_version_string
        geos = geos_version_string
    except Exception:                                 # a missing library is itself a finding
        geos = "unavailable"
        log.warning("geometry library version could not be determined")
    try:
        import pyproj
        proj = pyproj.__proj_version__
    except Exception:
        proj = "unavailable"
    return Environment(case_set_version, agent_version, tokenizer,
                       geos, proj, platform.python_version())

FROZEN = {"case-001": 0.8734, "case-014": 0.0000, "case-102": 1.0000}


class HarnessDrift(AssertionError):
    """The harness scores differently from before — investigate it, not the agent."""


def self_check(score_case: Callable[[str], float], tolerance: float = 1e-4) -> None:
    """Score three frozen cases before the sweep. Fails loudly if the harness moved."""
    for case_id, expected in FROZEN.items():
        try:
            got = score_case(case_id)
        except Exception as exc:
            raise HarnessDrift(f"self-check case {case_id} raised: {exc}") from exc
        if abs(got - expected) > tolerance:
            raise HarnessDrift(
                f"harness drift on {case_id}: expected {expected}, got {got:.4f}. "
                "Check library versions and truth data before blaming the agent.")


def run(cases: Sequence[dict], answer: Callable[[dict], dict],
        score: Callable[[dict, dict], dict], env: Environment) -> Run:
    """Run the sweep. Individual case failures are results, never aborts."""
    self_check(lambda cid: score(_case(cases, cid), answer(_case(cases, cid)))["score"])

    results = []
    for case in sorted(cases, key=lambda c: c["case_id"]):     # deterministic order
        try:
            produced = answer(case)
        except Exception as exc:                                # a crash is a result
            log.warning("case %s raised: %s", case["case_id"], exc)
            results.append({"case_id": case["case_id"], "family": case["family"],
                            "score": 0.0, "parsed": False, "note": f"agent raised: {exc}"})
            continue
        results.append({**score(case, produced), "case_id": case["case_id"],
                        "family": case["family"], "region": case.get("region", "all")})
    return Run(env, tuple(results), case_set_digest(cases))

def test_self_check_fires_before_the_sweep():
    def drifted(_case_id):
        return 0.5                                    # nothing like the frozen expectations
    try:
        self_check(drifted)
    except HarnessDrift as exc:
        assert "before blaming the agent" in str(exc)
        return
    raise AssertionError("harness drift must stop the sweep")


def test_agent_crash_becomes_a_scored_result():
    def crashing(_case):
        raise RuntimeError("tool timeout")
    out = run(CASES, crashing, score_fn, ENV)
    assert len(out.results) == len(CASES)
    assert all(r["score"] == 0.0 and not r["parsed"] for r in out.results)


def test_case_set_digest_changes_when_truth_changes():
    before = case_set_digest(CASES)
    edited = [{**CASES[0], "truth_geometry": "POLYGON((0 0,1 0,1 1,0 0))"}, *CASES[1:]]
    assert case_set_digest(edited) != before


def test_ordering_is_deterministic():
    a = run(CASES, answer_fn, score_fn, ENV)
    b = run(list(reversed(CASES)), answer_fn, score_fn, ENV)
    assert [r["case_id"] for r in a.results] == [r["case_id"] for r in b.results]