phase 5: evaluation harness (SROIE)

Two-phase harness so inference runs exactly once and tuning is free:
- predict phase (eval/predict.py): runs process_document over a dataset slice,
saves PIL images to temp files, caches gold + predicted Document + confidence
+ validation report to eval/cache/ keyed by id. The only phase that hits the
model; idempotent (cached ids skipped) so re-runs never re-bill.
- score phase (eval/score.py + eval/metrics.py): loads the cache and computes
per-field and per-critical-field precision/recall/F1, the auto-accept critical
precision, and a threshold sweep (0.50-0.99) that replays the real route() over
cached (confidence, report), honoring the hard-failure override. Fully offline.

Datasets: SROIE (jsdnrs/ICDAR2019-SROIE test) wired end-to-end; CORD and an
invoice-JSON set scaffolded as adapters but not wired (load() raises, absent from
WIRED_DATASETS). Only dataset-labeled fields are scored.

Comparison (eval/normalize.py) reuses the schema's number/date coercion, then
compares money cent-exact (NOT the reconciliation relative epsilon, which would
score a materially-wrong total as correct and inflate critical precision), dates
on ISO equality, text case/whitespace-insensitive.

18 offline unit tests (no model, no downloads) pin the comparator and the score
phase on synthetic cached entries, incl. a regression test for the money
tolerance and the hard-failure override. Harness adversarially reviewed before
running; eval/cache/ is git-ignored (no benchmark data committed).

.gitignore

@@ -1,6 +1,9 @@
 # Runtime data (inbox/processed/review/exports) and the local SQLite DB
 data/
 
+# Evaluation prediction cache (no benchmark data committed to git; build-plan 5.1)
+eval/cache/
+
 # Secrets and local environment
 .env
 

eval/__init__.py

@@ -0,0 +1,15 @@
+"""Evaluation harness for the document-extraction agent (build-plan phase 5).
+
+The harness is deliberately split into two phases so that model inference
+happens exactly once and threshold tuning is free:
+
+- **predict** (``eval.predict``) runs ``core.process_document`` over a dataset
+  slice and caches each result (gold labels, predicted document, confidence,
+  validation report) to ``eval/cache/`` keyed by example id. This is the only
+  phase that calls a model backend and the only phase that spends API quota.
+- **score** (``eval.score``) loads the cache and computes every metric plus the
+  threshold sweep purely offline. Re-tuning never re-runs inference.
+
+See ``docs/03_data_and_extraction_spec.md`` section 6 for the evaluation
+methodology and ``docs/05_build_plan.md`` phase 5 for the task definition.
+"""

eval/cache.py

@@ -0,0 +1,116 @@
+"""Read/write the per-example prediction cache and reconstruct reports.
+
+The predict phase writes one JSON file per example under
+``eval/cache/<dataset>/<id>.json``; the score phase reads them back. Keeping the
+model output on disk is what makes tuning free: the threshold sweep replays the
+pure ``route`` function over the cached ``(confidence, validation)`` pairs and
+never touches a model.
+
+A cached entry has this shape::
+
+    {
+      "id": "X00016469670",
+      "dataset": "sroie",
+      "gold": {"vendor_name": ..., "total": ..., ...},
+      "labeled_fields": ["vendor_name", "vendor_address", "document_date", "total"],
+      "predicted": { ...Document.model_dump(mode="json")... },
+      "confidence": 0.5,
+      "decision": "review",         # decision at the predict-run threshold (informational)
+      "modality": "image",
+      "backend": "gemini",
+      "validation": { "hard_failed": bool, "results": [...], ... },
+      "error": null
+    }
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+from typing import Any
+
+from doc_agent.validation.rules import RuleResult, ValidationReport
+
+# Default location for the cache; git-ignored (no evaluation data in the repo).
+DEFAULT_CACHE_BASE = Path("eval/cache")
+
+_UNSAFE_ID = re.compile(r"[^A-Za-z0-9._-]")
+
+
+def _safe_filename(example_id: str) -> str:
+    """Turn an example id into a filesystem-safe file stem."""
+    return _UNSAFE_ID.sub("_", example_id)
+
+
+def dataset_dir(cache_base: Path, dataset: str) -> Path:
+    """Return the cache directory for a dataset (not created)."""
+    return Path(cache_base) / dataset
+
+
+def write_entry(cache_base: Path, dataset: str, entry: dict[str, Any]) -> Path:
+    """Write one cache entry to ``<cache_base>/<dataset>/<id>.json``.
+
+    Args:
+        cache_base: Root cache directory.
+        dataset: Dataset name (subdirectory).
+        entry: The entry dict; must contain an ``"id"`` key.
+
+    Returns:
+        The path the entry was written to.
+    """
+    directory = dataset_dir(cache_base, dataset)
+    directory.mkdir(parents=True, exist_ok=True)
+    path = directory / f"{_safe_filename(str(entry['id']))}.json"
+    path.write_text(json.dumps(entry, indent=2, default=str), encoding="utf-8")
+    return path
+
+
+def read_entries(cache_base: Path, dataset: str) -> list[dict[str, Any]]:
+    """Load all cached entries for a dataset, sorted by filename.
+
+    Args:
+        cache_base: Root cache directory.
+        dataset: Dataset name (subdirectory).
+
+    Returns:
+        A list of entry dicts (empty if the directory does not exist).
+    """
+    directory = dataset_dir(cache_base, dataset)
+    if not directory.exists():
+        return []
+    return [
+        json.loads(path.read_text(encoding="utf-8"))
+        for path in sorted(directory.glob("*.json"))
+    ]
+
+
+def existing_ids(cache_base: Path, dataset: str) -> set[str]:
+    """Return the set of example ids already cached for a dataset."""
+    return {str(entry["id"]) for entry in read_entries(cache_base, dataset)}
+
+
+def report_from_dict(validation: dict[str, Any]) -> ValidationReport:
+    """Reconstruct a :class:`ValidationReport` from its cached dict form.
+
+    This lets the score phase replay the real ``route`` function over cached
+    results -- in particular ``report.hard_failed`` is recomputed from the
+    per-rule results, so the hard-failure override is honored during the sweep.
+
+    Args:
+        validation: The ``validation`` sub-dict of a cache entry (as produced by
+            ``ValidationReport.to_dict``).
+
+    Returns:
+        A ``ValidationReport`` whose ``results`` mirror the cached rule outcomes.
+    """
+    results = tuple(
+        RuleResult(
+            code=item["code"],
+            severity=item["severity"],
+            status=item["status"],
+            message=item.get("message", ""),
+        )
+        for item in validation.get("results", [])
+    )
+    return ValidationReport(results=results)

eval/datasets/__init__.py

@@ -0,0 +1,61 @@
+"""Dataset adapters mapping public benchmark labels onto the ``Document`` schema.
+
+Each adapter yields :class:`~eval.datasets.base.GoldExample` records: an id, the
+input (a PIL image or a file path), and a gold dict keyed by ``Document`` field
+names. Only the fields a dataset actually labels appear in ``labeled_fields``;
+the scorer restricts every metric to that set (an unlabeled field is neither a
+false positive nor a miss -- there is simply no ground truth for it).
+
+SROIE is wired end-to-end first (T10). CORD and the invoice-JSON set are
+scaffolded as adapters with their intended field mappings documented, but are
+intentionally not wired yet -- calling ``load`` on them raises.
+"""
+
+from __future__ import annotations
+
+from eval.datasets.base import DatasetAdapter, GoldExample
+from eval.datasets.cord import CordAdapter
+from eval.datasets.invoice_json import InvoiceJsonAdapter
+from eval.datasets.sroie import SroieAdapter
+
+# Registry of all known adapters, keyed by stable name.
+ADAPTERS: dict[str, type[DatasetAdapter]] = {
+    SroieAdapter.name: SroieAdapter,
+    CordAdapter.name: CordAdapter,
+    InvoiceJsonAdapter.name: InvoiceJsonAdapter,
+}
+
+# Adapters proven end-to-end and safe to run the predict phase against. The
+# others are scaffolds; ``get_adapter`` still returns them (so their metadata is
+# inspectable) but ``eval.predict`` refuses to run an unwired dataset.
+WIRED_DATASETS: frozenset[str] = frozenset({SroieAdapter.name})
+
+
+def get_adapter(name: str) -> DatasetAdapter:
+    """Instantiate a dataset adapter by name.
+
+    Args:
+        name: The adapter's stable name (e.g. "sroie").
+
+    Returns:
+        A new adapter instance.
+
+    Raises:
+        KeyError: If no adapter is registered under ``name``.
+    """
+    if name not in ADAPTERS:
+        available = ", ".join(sorted(ADAPTERS))
+        raise KeyError(f"Unknown dataset {name!r}; available: {available}")
+    return ADAPTERS[name]()
+
+
+__all__ = [
+    "ADAPTERS",
+    "WIRED_DATASETS",
+    "DatasetAdapter",
+    "GoldExample",
+    "SroieAdapter",
+    "CordAdapter",
+    "InvoiceJsonAdapter",
+    "get_adapter",
+]

eval/datasets/base.py

@@ -0,0 +1,67 @@
+"""Common types for dataset adapters.
+
+A :class:`GoldExample` is the unit the predict phase consumes: an id, the input
+(a PIL image for image datasets, or a file path for file-based ones), and the
+gold labels mapped onto ``Document`` field names. :class:`DatasetAdapter` is the
+protocol every concrete adapter satisfies.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from dataclasses import dataclass, field
+from typing import Any, Protocol, runtime_checkable
+
+
+@dataclass(frozen=True)
+class GoldExample:
+    """One evaluation example with its ground-truth labels.
+
+    Attributes:
+        id: Stable, unique identifier for the example (used as the cache key).
+        gold: Ground-truth values keyed by ``Document`` field name. A value is
+            ``None`` (or absent) when the dataset does not label that field for
+            this example. Values are raw (strings/numbers as the dataset stores
+            them); the scorer normalizes them before comparison.
+        image: The input image as a PIL ``Image`` for image datasets, or
+            ``None`` for file-based datasets.
+        source_path: Path to an input file for file-based datasets, or ``None``
+            when the input is an in-memory ``image``.
+        suffix: File extension to use when writing ``image`` to a temp file so
+            modality detection sees the right type (e.g. ".png").
+    """
+
+    id: str
+    gold: dict[str, Any]
+    image: Any = None
+    source_path: str | None = None
+    suffix: str = ".png"
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@runtime_checkable
+class DatasetAdapter(Protocol):
+    """Interface every dataset adapter implements.
+
+    Attributes:
+        name: Stable identifier for the dataset (e.g. "sroie").
+        hf_id: The Hugging Face dataset id it loads from.
+        labeled_fields: The ``Document`` field names this dataset provides gold
+            labels for; the scorer computes metrics only for these.
+    """
+
+    name: str
+    hf_id: str
+    labeled_fields: tuple[str, ...]
+
+    def load(self, limit: int | None = None) -> Iterator[GoldExample]:
+        """Yield gold examples, at most ``limit`` of them.
+
+        Args:
+            limit: Maximum number of examples to yield; ``None`` for all.
+
+        Yields:
+            :class:`GoldExample` records in a fixed, deterministic order so a
+            given ``limit`` always selects the same held-out slice.
+        """
+        ...

eval/datasets/cord.py

@@ -0,0 +1,51 @@
+"""CORD (v2) dataset adapter -- scaffold, not yet wired.
+
+CORD is ~11,000 Indonesian receipts captured in the wild with rich key-value
+and line-item labels. The ``naver-clova-ix/cord-v2`` build stores ground truth
+as a JSON string in ``ground_truth`` (a ``gt_parse`` object with ``menu`` line
+items and a ``sub_total`` / ``total`` block).
+
+Intended field mapping onto the ``Document`` schema (to implement when wired):
+
+- ``gt_parse.total.total_price``       -> ``total``    (critical)
+- ``gt_parse.sub_total.subtotal_price``-> ``subtotal``
+- ``gt_parse.sub_total.tax_price``     -> ``tax``      (critical)
+- ``gt_parse.menu[*]``                 -> ``line_items`` (name/cnt/price)
+- store name (where present)           -> ``vendor_name``
+
+This adapter is intentionally left unwired for T10 (SROIE-first). ``load`` raises
+so an accidental predict run against CORD fails fast rather than spending quota
+on an unvalidated mapping.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+
+from eval.datasets.base import GoldExample
+
+
+class CordAdapter:
+    """Scaffold adapter for the CORD v2 receipt benchmark (not wired)."""
+
+    name: str = "cord"
+    hf_id: str = "naver-clova-ix/cord-v2"
+    split: str = "test"
+    # Fields CORD would provide once the gt_parse mapping is implemented.
+    labeled_fields: tuple[str, ...] = ("vendor_name", "subtotal", "tax", "total")
+
+    def load(self, limit: int | None = None) -> Iterator[GoldExample]:
+        """Not implemented -- CORD is scaffolded but not wired for T10.
+
+        Args:
+            limit: Unused.
+
+        Raises:
+            NotImplementedError: Always; wire the ``gt_parse`` mapping first.
+        """
+        raise NotImplementedError(
+            "CORD adapter is scaffolded but not wired (T10 is SROIE-first). "
+            "Implement the gt_parse -> Document mapping and add 'cord' to "
+            "WIRED_DATASETS before running the predict phase against it."
+        )
+        yield  # pragma: no cover -- makes this a generator for the Protocol.

eval/datasets/invoice_json.py

@@ -0,0 +1,59 @@
+"""Invoice-JSON dataset adapter -- scaffold, not yet wired.
+
+The ``GokulRajaR/invoice-ocr-json`` set pairs native invoice images with
+structured JSON ground truth (key-values covering invoice number, dates, and
+monetary totals) -- convenient for the eval harness because the labels are
+already close to the ``Document`` schema.
+
+Intended field mapping onto the ``Document`` schema (to implement when wired):
+
+- invoice number field  -> ``invoice_number``  (critical)
+- invoice/issue date    -> ``document_date``
+- due date              -> ``due_date``
+- tax/VAT amount        -> ``tax``              (critical)
+- grand total           -> ``total``            (critical)
+- vendor/seller name    -> ``vendor_name``
+
+This is the dataset that would exercise ``invoice_number`` and ``tax`` -- the two
+critical fields SROIE does not label. It is intentionally left unwired for T10;
+``load`` raises so a predict run cannot spend quota on an unvalidated mapping.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+
+from eval.datasets.base import GoldExample
+
+
+class InvoiceJsonAdapter:
+    """Scaffold adapter for the invoice-OCR-JSON benchmark (not wired)."""
+
+    name: str = "invoice_json"
+    hf_id: str = "GokulRajaR/invoice-ocr-json"
+    split: str = "train"
+    labeled_fields: tuple[str, ...] = (
+        "vendor_name",
+        "invoice_number",
+        "document_date",
+        "due_date",
+        "tax",
+        "total",
+    )
+
+    def load(self, limit: int | None = None) -> Iterator[GoldExample]:
+        """Not implemented -- this set is scaffolded but not wired for T10.
+
+        Args:
+            limit: Unused.
+
+        Raises:
+            NotImplementedError: Always; implement the JSON->Document mapping
+                first.
+        """
+        raise NotImplementedError(
+            "invoice_json adapter is scaffolded but not wired (T10 is "
+            "SROIE-first). Implement the JSON -> Document mapping and add "
+            "'invoice_json' to WIRED_DATASETS before running predict against it."
+        )
+        yield  # pragma: no cover -- makes this a generator for the Protocol.

eval/datasets/sroie.py

@@ -0,0 +1,75 @@
+"""SROIE (ICDAR 2019) dataset adapter -- wired end-to-end.
+
+SROIE is ~1,000 real scanned receipts with four labeled key fields. Using the
+``jsdnrs/ICDAR2019-SROIE`` mirror, the "test" split is the held-out evaluation
+slice (it is never tuned against). Each example exposes:
+
+- ``key``       -> the example id
+- ``image``     -> a PIL image of the receipt (scan modality)
+- ``entities``  -> {"company", "date", "address", "total"}
+
+Field mapping onto the ``Document`` schema (data spec section 2):
+
+- ``entities["company"]`` -> ``vendor_name``
+- ``entities["address"]`` -> ``vendor_address``
+- ``entities["date"]``    -> ``document_date``  (day-first D/M/Y, e.g. 15/01/2019)
+- ``entities["total"]``   -> ``total``
+
+SROIE does **not** label ``tax`` or ``invoice_number``, so of the three critical
+fields only ``total`` is scored here. The dataset is loaded in streaming mode so
+a small slice does not download the full split.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+
+from eval.datasets.base import GoldExample
+
+
+class SroieAdapter:
+    """Adapter for the SROIE scanned-receipt benchmark."""
+
+    name: str = "sroie"
+    hf_id: str = "jsdnrs/ICDAR2019-SROIE"
+    split: str = "test"
+    labeled_fields: tuple[str, ...] = (
+        "vendor_name",
+        "vendor_address",
+        "document_date",
+        "total",
+    )
+
+    def load(self, limit: int | None = None) -> Iterator[GoldExample]:
+        """Yield the first ``limit`` SROIE test examples as gold examples.
+
+        Streaming keeps a small slice cheap (no full-split download). The first
+        ``limit`` examples form a fixed, reproducible slice.
+
+        Args:
+            limit: Maximum number of examples to yield; ``None`` for the whole
+                split.
+
+        Yields:
+            :class:`GoldExample` records with the receipt image and mapped gold.
+        """
+        from datasets import load_dataset
+
+        dataset = load_dataset(self.hf_id, split=self.split, streaming=True)
+        for index, example in enumerate(dataset):
+            if limit is not None and index >= limit:
+                break
+            entities = example.get("entities") or {}
+            gold = {
+                "vendor_name": entities.get("company"),
+                "vendor_address": entities.get("address"),
+                "document_date": entities.get("date"),
+                "total": entities.get("total"),
+            }
+            yield GoldExample(
+                id=str(example["key"]),
+                gold=gold,
+                image=example["image"],
+                suffix=".png",
+                metadata={"dataset": self.name, "index": index},
+            )

eval/metrics.py

@@ -0,0 +1,265 @@
+"""Metric aggregation and the threshold sweep (pure functions over cache entries).
+
+Everything here operates on already-cached entries (see ``eval.cache``) and never
+calls a model, so it is fully unit-testable on synthetic data and re-runnable for
+free. Two computations implement the evaluation methodology (data spec section 6):
+
+- :func:`compute_field_metrics` -- per-field precision / recall / F1 over the
+  whole slice, restricted to the fields a dataset labels.
+- :func:`sweep_thresholds` -- for each candidate threshold, replay the real
+  ``route`` over the cached ``(confidence, validation)`` pairs (honoring the
+  hard-failure override) and report auto-accept volume and the precision/recall
+  of the auto-accepted critical fields -- the precision/recall trade-off curve.
+
+Definitions (field level, against ground truth):
+
+- A predicted value is *present* if it normalizes to a non-absent value; a match
+  requires the gold to be present too and the normalized values to agree.
+- *precision* = matches / predicted-present. *recall* = matches / gold-present.
+- *F1* = harmonic mean.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Any
+
+from doc_agent.routing.score import route
+
+from eval.cache import report_from_dict
+from eval.normalize import is_present, values_match
+
+# Critical, precision-prioritised fields (data spec section 2 / CLAUDE.md).
+CRITICAL_FIELDS: tuple[str, ...] = ("total", "tax", "invoice_number")
+
+# Threshold grid for the sweep: 0.50 -> 0.99 inclusive at 0.01 steps.
+THRESHOLDS: tuple[float, ...] = tuple(round(0.50 + 0.01 * i, 2) for i in range(50))
+
+
+def _f1(precision: float | None, recall: float | None) -> float | None:
+    """Harmonic mean of precision and recall (``None`` if both are undefined)."""
+    if precision is None and recall is None:
+        return None
+    if not precision or not recall:  # covers None or 0.0 on either side
+        return 0.0
+    return 2 * precision * recall / (precision + recall)
+
+
+@dataclass(frozen=True)
+class FieldMetric:
+    """Precision / recall / F1 for one field over a slice.
+
+    Attributes:
+        field: The ``Document`` field name.
+        n_pred: Number of examples where the pipeline produced a value.
+        n_gold: Number of examples where the gold labels a value.
+        n_match: Number of examples where prediction and gold agree.
+    """
+
+    field: str
+    n_pred: int
+    n_gold: int
+    n_match: int
+
+    @property
+    def precision(self) -> float | None:
+        """matches / predicted-present, or ``None`` if nothing was predicted."""
+        return self.n_match / self.n_pred if self.n_pred else None
+
+    @property
+    def recall(self) -> float | None:
+        """matches / gold-present, or ``None`` if there is no gold."""
+        return self.n_match / self.n_gold if self.n_gold else None
+
+    @property
+    def f1(self) -> float | None:
+        """Harmonic mean of precision and recall."""
+        return _f1(self.precision, self.recall)
+
+
+@dataclass(frozen=True)
+class SweepRow:
+    """One threshold's auto-accept volume and critical-field trade-off.
+
+    Attributes:
+        threshold: The candidate auto-accept threshold.
+        n_total: Total examples in the slice.
+        n_accepted: How many examples ``route`` auto-accepts at this threshold.
+        crit_pred: Predicted-present critical values among accepted examples
+            (the denominator of auto-accept precision).
+        crit_match: Correct critical values among accepted examples.
+        crit_gold_total: Gold-present critical values across the whole slice
+            (the denominator of critical recall).
+    """
+
+    threshold: float
+    n_total: int
+    n_accepted: int
+    crit_pred: int
+    crit_match: int
+    crit_gold_total: int
+
+    @property
+    def accept_rate(self) -> float:
+        """Fraction of the slice auto-accepted at this threshold."""
+        return self.n_accepted / self.n_total if self.n_total else 0.0
+
+    @property
+    def crit_precision(self) -> float | None:
+        """Precision on critical fields over the auto-accepted subset.
+
+        This is the metric the operating point targets (>= 0.98). ``None`` when
+        no critical value was auto-accepted (precision undefined).
+        """
+        return self.crit_match / self.crit_pred if self.crit_pred else None
+
+    @property
+    def crit_recall(self) -> float | None:
+        """Correctly auto-accepted critical values / all gold critical values.
+
+        The recall "kept" at this threshold; the rest is review-queue volume.
+        """
+        return self.crit_match / self.crit_gold_total if self.crit_gold_total else None
+
+
+def compute_field_metrics(
+    entries: Sequence[dict[str, Any]],
+    fields: Sequence[str],
+) -> list[FieldMetric]:
+    """Compute per-field precision/recall/F1 over the slice.
+
+    Args:
+        entries: Cached prediction entries (each with ``predicted`` and ``gold``).
+        fields: The ``Document`` field names to score (a dataset's labeled set).
+
+    Returns:
+        One :class:`FieldMetric` per field, in the order of ``fields``.
+    """
+    metrics: list[FieldMetric] = []
+    for field in fields:
+        n_pred = n_gold = n_match = 0
+        for entry in entries:
+            predicted = entry.get("predicted", {}).get(field)
+            gold = entry.get("gold", {}).get(field)
+            if is_present(field, predicted):
+                n_pred += 1
+            if is_present(field, gold):
+                n_gold += 1
+            if values_match(field, predicted, gold):
+                n_match += 1
+        metrics.append(FieldMetric(field, n_pred, n_gold, n_match))
+    return metrics
+
+
+def sweep_thresholds(
+    entries: Sequence[dict[str, Any]],
+    critical_fields: Sequence[str],
+    thresholds: Sequence[float] = THRESHOLDS,
+) -> list[SweepRow]:
+    """Replay ``route`` across thresholds and measure the critical-field trade-off.
+
+    For each threshold the real ``route`` is applied to every entry's cached
+    ``(confidence, validation)`` pair -- so a hard-failure entry is forced to
+    review at *every* threshold, exactly as in production -- and the auto-accepted
+    subset's critical-field precision and recall are computed. No inference runs.
+
+    Args:
+        entries: Cached prediction entries.
+        critical_fields: The critical fields the dataset labels (the subset of
+            ``total``/``tax``/``invoice_number`` with gold present).
+        thresholds: The candidate thresholds to sweep. Defaults to
+            :data:`THRESHOLDS` (0.50->0.99).
+
+    Returns:
+        One :class:`SweepRow` per threshold, in ``thresholds`` order.
+    """
+    reports = {entry["id"]: report_from_dict(entry.get("validation", {})) for entry in entries}
+    n_total = len(entries)
+
+    # Denominator for critical recall: gold-present critical values across all.
+    crit_gold_total = sum(
+        1
+        for entry in entries
+        for field in critical_fields
+        if is_present(field, entry.get("gold", {}).get(field))
+    )
+
+    rows: list[SweepRow] = []
+    for threshold in thresholds:
+        crit_pred = crit_match = n_accepted = 0
+        for entry in entries:
+            report = reports[entry["id"]]
+            decision = route(entry.get("confidence", 0.0), report, threshold=threshold)
+            if decision != "accept":
+                continue
+            n_accepted += 1
+            for field in critical_fields:
+                predicted = entry.get("predicted", {}).get(field)
+                gold = entry.get("gold", {}).get(field)
+                if is_present(field, predicted):
+                    crit_pred += 1
+                    if values_match(field, predicted, gold):
+                        crit_match += 1
+        rows.append(
+            SweepRow(
+                threshold=threshold,
+                n_total=n_total,
+                n_accepted=n_accepted,
+                crit_pred=crit_pred,
+                crit_match=crit_match,
+                crit_gold_total=crit_gold_total,
+            )
+        )
+    return rows
+
+
+def confidence_histogram(
+    entries: Sequence[dict[str, Any]],
+    ndigits: int = 2,
+) -> dict[float, int]:
+    """Count cached confidence scores, rounded, for distribution reporting.
+
+    Surfaces why the sweep looks the way it does: when a backend exposes no
+    per-field confidence the scorer starts from a neutral 0.5, capping scores at
+    0.5, so almost nothing clears a threshold above 0.5.
+
+    Args:
+        entries: Cached prediction entries.
+        ndigits: Rounding precision for bucketing confidences.
+
+    Returns:
+        A dict mapping rounded confidence to count, ascending by confidence.
+    """
+    counts: dict[float, int] = {}
+    for entry in entries:
+        bucket = round(float(entry.get("confidence", 0.0)), ndigits)
+        counts[bucket] = counts.get(bucket, 0) + 1
+    return dict(sorted(counts.items()))
+
+
+def smallest_threshold_meeting(
+    rows: Sequence[SweepRow],
+    target_precision: float,
+) -> SweepRow | None:
+    """Return the lowest-threshold row whose critical precision meets a target.
+
+    Reported as analysis only -- the operator chooses the actual threshold.
+
+    Args:
+        rows: Sweep rows (assumed ascending by threshold).
+        target_precision: The critical auto-accept precision to meet (e.g. 0.98).
+
+    Returns:
+        The first row (lowest threshold) with a defined critical precision at or
+        above ``target_precision`` and at least one auto-accepted example, or
+        ``None`` if no threshold achieves it.
+    """
+    for row in rows:
+        if (
+            row.n_accepted > 0
+            and row.crit_precision is not None
+            and row.crit_precision >= target_precision
+        ):
+            return row
+    return None

eval/normalize.py

@@ -0,0 +1,135 @@
+"""Value normalization and per-field comparison for evaluation (pure, no I/O).
+
+Before predicted and gold values are compared they must be normalized so that
+cosmetic differences (casing, whitespace, currency symbols, date formats, the
+JSON round-trip through the cache) do not count as errors -- exactly the
+normalization the evaluation methodology calls for (data spec section 6, step 2).
+
+The normalization reuses the pipeline's own coercion helpers so eval judges
+values the same way the pipeline produces them:
+
+- **money** fields go through the schema's number coercion and are compared for
+  **exact equality at cent precision** (``round(x, 2)``). This deliberately does
+  *not* reuse the validation module's ``money_close``: that check carries a 0.5%
+  relative tolerance whose purpose is to absorb accumulated line-item rounding in
+  the H2/H3 arithmetic cross-checks (data spec section 3). Applied to a single
+  gold-vs-prediction comparison it would count a materially-wrong total as
+  correct (e.g. 502.00 vs a gold of 500.00 -> within 0.5% of 500), which would
+  overstate exactly the ``total``/``tax`` auto-accept precision this harness
+  exists to measure against the >= 0.98 target. Section 6 asks only for
+  normalization then comparison; cent-exact equality is that comparison, with the
+  cent rounding absorbing sub-cent floating-point representation noise.
+- **date** fields go through the schema's date coercion (day-first for ambiguous
+  D/M/Y, matching SROIE) and are compared for exact ISO-date equality.
+- **text** fields are lower-cased and whitespace-collapsed, then compared for
+  exact equality.
+
+A value that normalizes to ``None`` (absent, blank, or unparseable) is treated as
+"not present": it can never match, so predicting a value where the gold is absent
+counts against precision, and missing a gold value counts against recall.
+"""
+
+from __future__ import annotations
+
+import re
+from datetime import date
+from typing import Any
+
+from doc_agent.schema.models import _coerce_date, _coerce_number
+
+# How each schema field is compared. Fields not listed default to "text".
+FIELD_KIND: dict[str, str] = {
+    "doc_type": "text",
+    "vendor_name": "text",
+    "vendor_address": "text",
+    "invoice_number": "text",
+    "currency": "text",
+    "document_date": "date",
+    "due_date": "date",
+    "subtotal": "money",
+    "tax": "money",
+    "total": "money",
+}
+
+
+def _normalize_text(value: Any) -> str | None:
+    """Lower-case and whitespace-collapse a text value; blanks become ``None``."""
+    if value is None:
+        return None
+    collapsed = re.sub(r"\s+", " ", str(value).strip().lower())
+    return collapsed or None
+
+
+def _normalize_money(value: Any) -> float | None:
+    """Coerce a monetary value to ``float``; unparseable/absent becomes ``None``."""
+    try:
+        return _coerce_number(value)
+    except (ValueError, TypeError):
+        return None
+
+
+def _normalize_date(value: Any) -> date | None:
+    """Coerce a date value to ``datetime.date``; unparseable/absent becomes ``None``."""
+    try:
+        return _coerce_date(value)
+    except (ValueError, TypeError):
+        return None
+
+
+def normalize(field: str, value: Any) -> Any:
+    """Normalize a single value according to its field's comparison kind.
+
+    Args:
+        field: The ``Document`` field name.
+        value: The raw predicted or gold value.
+
+    Returns:
+        A normalized comparable value (``float`` for money, ``date`` for dates,
+        lower-cased string for text), or ``None`` when the value is absent,
+        blank, or unparseable.
+    """
+    kind = FIELD_KIND.get(field, "text")
+    if kind == "money":
+        return _normalize_money(value)
+    if kind == "date":
+        return _normalize_date(value)
+    return _normalize_text(value)
+
+
+def is_present(field: str, value: Any) -> bool:
+    """Whether ``value`` normalizes to a real (non-absent) value for ``field``.
+
+    Args:
+        field: The ``Document`` field name.
+        value: The raw value to test.
+
+    Returns:
+        ``True`` if the value normalizes to something other than ``None``.
+    """
+    return normalize(field, value) is not None
+
+
+def values_match(field: str, predicted: Any, gold: Any) -> bool:
+    """Whether a predicted value matches the gold value for ``field``.
+
+    Both sides are normalized first. A match requires both to be present;
+    monetary fields match within the rounding tolerance, dates and text match on
+    exact normalized equality.
+
+    Args:
+        field: The ``Document`` field name being compared.
+        predicted: The pipeline's predicted value (possibly a JSON-cached form).
+        gold: The dataset's ground-truth value.
+
+    Returns:
+        ``True`` if the values are considered equal after normalization.
+    """
+    left = normalize(field, predicted)
+    right = normalize(field, gold)
+    if left is None or right is None:
+        return False
+    if FIELD_KIND.get(field, "text") == "money":
+        # Cent-exact: no relative tolerance, so a materially-wrong total is never
+        # scored correct. round() absorbs sub-cent float representation noise.
+        return round(left, 2) == round(right, 2)
+    return left == right

eval/predict.py

@@ -0,0 +1,199 @@
+"""Predict phase: run the core over a dataset slice and cache each result.
+
+This is the **only** phase that calls a model backend and the only phase that
+spends API quota. It is deliberately idempotent: an example already present in
+the cache is skipped unless ``overwrite`` is set, so a re-run after an
+interruption resumes rather than re-billing.
+
+For each example the PIL image is written to a temporary file (``core`` takes a
+path, not bytes) with an extension that makes modality detection classify it as
+an image; ``process_document`` then runs the full detect -> acquire -> extract
+-> validate -> score -> route pipeline. The gold labels, predicted document,
+confidence, and validation report are cached; nothing is persisted to the app's
+SQLite store (eval is not production ingestion).
+"""
+
+from __future__ import annotations
+
+import logging
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from doc_agent.backends.base import create_backend
+from doc_agent.config import Settings, load_config
+from doc_agent.core import process_document
+
+from eval.cache import DEFAULT_CACHE_BASE, existing_ids, write_entry
+from eval.datasets import WIRED_DATASETS, get_adapter
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class PredictStats:
+    """Summary of a predict-phase run.
+
+    Attributes:
+        dataset: The dataset name.
+        requested: The requested slice size (``limit``), or ``None`` for all.
+        processed: Examples actually sent through the pipeline this run.
+        skipped: Examples skipped because they were already cached.
+        accepted: Processed examples the pipeline auto-accepted.
+        review: Processed examples routed to review.
+        errors: Processed examples whose pipeline stage raised (routed to review).
+        failed: Examples that could not be prepared at all (e.g. the image would
+            not encode); logged and skipped so one bad input never aborts the run.
+    """
+
+    dataset: str
+    requested: int | None
+    processed: int
+    skipped: int
+    accepted: int
+    review: int
+    errors: int
+    failed: int
+
+
+def _process_example(
+    example: Any,
+    dataset: str,
+    labeled_fields: tuple[str, ...],
+    settings: Settings,
+    backend: Any,
+) -> dict[str, Any]:
+    """Save one example's image to a temp file, run the pipeline, build the entry.
+
+    The temp file is created first and unlinked in a ``finally`` so it is cleaned
+    up even if ``image.save`` or the pipeline raises. ``process_document`` itself
+    never raises (rule 6); only image encoding can, and the caller isolates that.
+
+    Args:
+        example: The :class:`~eval.datasets.base.GoldExample` to process.
+        dataset: Dataset name (recorded in the entry).
+        labeled_fields: The dataset's labeled fields (recorded in the entry).
+        settings: Validated configuration.
+        backend: The extraction backend to use.
+
+    Returns:
+        A JSON-serializable cache entry for the example.
+    """
+    with tempfile.NamedTemporaryFile(suffix=example.suffix, delete=False) as handle:
+        temp_path = Path(handle.name)
+    try:
+        example.image.save(temp_path)
+        result = process_document(temp_path, settings=settings, backend=backend)
+    finally:
+        temp_path.unlink(missing_ok=True)
+    return _build_entry(example, result, dataset, labeled_fields)
+
+
+def _build_entry(
+    example: Any,
+    result: Any,
+    dataset: str,
+    labeled_fields: tuple[str, ...],
+) -> dict[str, Any]:
+    """Assemble a JSON-serializable cache entry from an example and its result."""
+    return {
+        "id": example.id,
+        "dataset": dataset,
+        "gold": example.gold,
+        "labeled_fields": list(labeled_fields),
+        "predicted": result.document.model_dump(mode="json"),
+        "confidence": result.confidence,
+        "decision": result.decision,
+        "modality": result.modality,
+        "backend": result.backend_name,
+        "validation": result.report.to_dict(),
+        "error": result.error,
+    }
+
+
+def run_predict(
+    dataset: str,
+    limit: int | None,
+    *,
+    settings: Settings | None = None,
+    cache_base: Path = DEFAULT_CACHE_BASE,
+    overwrite: bool = False,
+) -> PredictStats:
+    """Run the pipeline over a dataset slice and cache each result.
+
+    Args:
+        dataset: Name of a wired dataset adapter (e.g. "sroie").
+        limit: Number of examples to process (the held-out slice size); ``None``
+            for the whole split.
+        settings: Validated configuration; loaded from the environment when
+            ``None`` (must select a backend that can read images).
+        cache_base: Root cache directory. Defaults to ``eval/cache``.
+        overwrite: Re-process and overwrite examples already cached. Defaults to
+            ``False`` so re-runs resume without re-billing.
+
+    Returns:
+        A :class:`PredictStats` summary of the run.
+
+    Raises:
+        ValueError: If ``dataset`` is not wired for the predict phase.
+    """
+    if dataset not in WIRED_DATASETS:
+        wired = ", ".join(sorted(WIRED_DATASETS))
+        raise ValueError(
+            f"Dataset {dataset!r} is scaffolded but not wired for prediction. "
+            f"Wired datasets: {wired}."
+        )
+
+    adapter = get_adapter(dataset)
+    settings = settings or load_config()
+    backend = create_backend(settings)
+    already = set() if overwrite else existing_ids(cache_base, dataset)
+
+    processed = skipped = accepted = review = errors = failed = 0
+    logger.info("eval-predict: dataset=%s limit=%s backend=%s", dataset, limit, backend.name)
+
+    for example in adapter.load(limit):
+        if example.id in already:
+            skipped += 1
+            logger.info("eval-predict: skip cached id=%s", example.id)
+            continue
+
+        # Isolate per-example preparation failures (e.g. an un-encodable image) so
+        # one bad input logs and is skipped rather than aborting the whole slice.
+        try:
+            entry = _process_example(example, dataset, adapter.labeled_fields, settings, backend)
+        except Exception as exc:  # noqa: BLE001 -- never let one document halt the run.
+            failed += 1
+            logger.error("eval-predict: could not prepare id=%s -- skipping: %s", example.id, exc)
+            continue
+
+        write_entry(cache_base, dataset, entry)
+
+        processed += 1
+        if entry["error"]:
+            errors += 1
+        if entry["decision"] == "accept":
+            accepted += 1
+        else:
+            review += 1
+        logger.info(
+            "eval-predict: id=%s decision=%s confidence=%.3f error=%s",
+            example.id,
+            entry["decision"],
+            entry["confidence"],
+            bool(entry["error"]),
+        )
+
+    stats = PredictStats(
+        dataset=dataset,
+        requested=limit,
+        processed=processed,
+        skipped=skipped,
+        accepted=accepted,
+        review=review,
+        errors=errors,
+        failed=failed,
+    )
+    logger.info("eval-predict: done %s", stats)
+    return stats

eval/run_eval.py

@@ -0,0 +1,103 @@
+"""Command-line entry point for the two-phase evaluation harness.
+
+Usage::
+
+    # Phase 1 -- runs the model, spends quota; start with a small slice.
+    uv run python -m eval.run_eval predict --dataset sroie --limit 20
+
+    # Phase 2 -- offline; recompute metrics and sweep as often as you like.
+    uv run python -m eval.run_eval score --dataset sroie
+
+The predict phase caches results under ``eval/cache/<dataset>/`` and is
+idempotent (already-cached ids are skipped unless ``--overwrite``). The score
+phase reads that cache and prints the metrics tables and threshold sweep; it
+never re-runs inference.
+"""
+
+from __future__ import annotations
+
+import argparse
+import logging
+import sys
+from pathlib import Path
+
+from eval.cache import DEFAULT_CACHE_BASE
+from eval.predict import run_predict
+from eval.score import build_report, format_report
+
+
+def _add_common(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument("--dataset", default="sroie", help="Dataset adapter name (default: sroie).")
+    parser.add_argument(
+        "--cache-base",
+        type=Path,
+        default=DEFAULT_CACHE_BASE,
+        help="Root cache directory (default: eval/cache).",
+    )
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(prog="eval.run_eval", description=__doc__)
+    subparsers = parser.add_subparsers(dest="command", required=True)
+
+    predict = subparsers.add_parser("predict", help="Run the model over a slice and cache results.")
+    _add_common(predict)
+    predict.add_argument(
+        "--limit",
+        type=int,
+        default=20,
+        help="Number of examples to process (the held-out slice size; default: 20).",
+    )
+    predict.add_argument(
+        "--overwrite",
+        action="store_true",
+        help="Re-process and overwrite already-cached examples.",
+    )
+
+    score = subparsers.add_parser("score", help="Compute metrics + sweep from the cache (offline).")
+    _add_common(score)
+
+    return parser
+
+
+def main(argv: list[str] | None = None) -> int:
+    """Run the CLI.
+
+    Args:
+        argv: Argument list (defaults to ``sys.argv[1:]``).
+
+    Returns:
+        Process exit code (0 on success).
+    """
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    )
+    args = _build_parser().parse_args(argv)
+
+    if args.command == "predict":
+        stats = run_predict(
+            args.dataset,
+            args.limit,
+            cache_base=args.cache_base,
+            overwrite=args.overwrite,
+        )
+        print(
+            f"\nPredict complete for {stats.dataset}: "
+            f"processed={stats.processed} skipped={stats.skipped} "
+            f"accepted={stats.accepted} review={stats.review} errors={stats.errors} "
+            f"failed={stats.failed}\n"
+            f"Now run: uv run python -m eval.run_eval score --dataset {stats.dataset}"
+        )
+        return 0
+
+    if args.command == "score":
+        report = build_report(args.dataset, cache_base=args.cache_base)
+        print(format_report(report))
+        return 0
+
+    return 1  # pragma: no cover -- argparse enforces a valid subcommand.
+
+
+if __name__ == "__main__":
+    sys.exit(main())

eval/score.py

@@ -0,0 +1,221 @@
+"""Score phase: compute all metrics and the threshold sweep from the cache.
+
+Purely offline -- it loads cached predictions (``eval.cache``), runs the pure
+aggregations (``eval.metrics``), and formats human-readable tables. Re-running
+with a different threshold grid or comparison never re-runs inference, which is
+the whole point of the two-phase split.
+
+The formatted output covers the four things the methodology asks for (data spec
+section 6): per-field precision/recall/F1, per-critical-field metrics, document
+routing stats, and the threshold sweep trade-off curve. It also prints the
+confidence distribution (so a flat sweep is explained by the backend exposing no
+per-field confidence) and, as analysis only, the lowest threshold that reaches a
+target auto-accept precision -- the operator still chooses the value.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from eval.cache import DEFAULT_CACHE_BASE, read_entries
+from eval.metrics import (
+    CRITICAL_FIELDS,
+    THRESHOLDS,
+    FieldMetric,
+    SweepRow,
+    compute_field_metrics,
+    confidence_histogram,
+    smallest_threshold_meeting,
+    sweep_thresholds,
+)
+from eval.normalize import is_present
+
+# Auto-accept precision target on critical fields (data spec section 6).
+TARGET_CRITICAL_PRECISION: float = 0.98
+
+
+@dataclass(frozen=True)
+class ScoreReport:
+    """Everything the score phase computed for one dataset slice."""
+
+    dataset: str
+    n: int
+    labeled_fields: tuple[str, ...]
+    critical_labeled: tuple[str, ...]
+    field_metrics: list[FieldMetric]
+    sweep: list[SweepRow]
+    confidence_hist: dict[float, int]
+    n_error: int
+
+
+def _labeled_fields(entries: list[dict[str, Any]]) -> tuple[str, ...]:
+    """Union of the ``labeled_fields`` recorded across cached entries."""
+    seen: list[str] = []
+    for entry in entries:
+        for field in entry.get("labeled_fields", []):
+            if field not in seen:
+                seen.append(field)
+    return tuple(seen)
+
+
+def _critical_labeled(labeled: tuple[str, ...], entries: list[dict[str, Any]]) -> tuple[str, ...]:
+    """Critical fields the dataset actually labels *and* has gold present for.
+
+    A critical field with no gold anywhere in the slice cannot be scored, so it
+    is excluded from the critical-precision denominators.
+    """
+    result: list[str] = []
+    for field in CRITICAL_FIELDS:
+        if field not in labeled:
+            continue
+        if any(is_present(field, entry.get("gold", {}).get(field)) for entry in entries):
+            result.append(field)
+    return tuple(result)
+
+
+def build_report(
+    dataset: str,
+    *,
+    cache_base: Path = DEFAULT_CACHE_BASE,
+    thresholds: tuple[float, ...] = THRESHOLDS,
+) -> ScoreReport:
+    """Load the cache for a dataset and compute the full score report.
+
+    Args:
+        dataset: Dataset name whose cache to score.
+        cache_base: Root cache directory. Defaults to ``eval/cache``.
+        thresholds: The threshold grid to sweep.
+
+    Returns:
+        A :class:`ScoreReport`.
+
+    Raises:
+        FileNotFoundError: If no cached entries exist for the dataset.
+    """
+    entries = read_entries(cache_base, dataset)
+    if not entries:
+        raise FileNotFoundError(
+            f"No cached predictions for dataset {dataset!r} under {cache_base}. "
+            "Run the predict phase first."
+        )
+
+    labeled = _labeled_fields(entries)
+    critical_labeled = _critical_labeled(labeled, entries)
+    field_metrics = compute_field_metrics(entries, labeled)
+    sweep = sweep_thresholds(entries, critical_labeled, thresholds)
+    hist = confidence_histogram(entries)
+    n_error = sum(1 for entry in entries if entry.get("error"))
+
+    return ScoreReport(
+        dataset=dataset,
+        n=len(entries),
+        labeled_fields=labeled,
+        critical_labeled=critical_labeled,
+        field_metrics=field_metrics,
+        sweep=sweep,
+        confidence_hist=hist,
+        n_error=n_error,
+    )
+
+
+# --- Formatting ----------------------------------------------------------------
+
+
+def _pct(value: float | None) -> str:
+    """Format an optional ratio as a percentage, or 'n/a' when undefined."""
+    return "  n/a" if value is None else f"{value * 100:5.1f}%"
+
+
+def _format_field_table(report: ScoreReport) -> list[str]:
+    lines = [
+        "Per-field metrics (whole slice):",
+        f"  {'field':<16} {'P':>7} {'R':>7} {'F1':>7}   {'pred':>4} {'gold':>4} {'ok':>4}",
+    ]
+    for metric in report.field_metrics:
+        marker = " *" if metric.field in report.critical_labeled else "  "
+        lines.append(
+            f"{marker}{metric.field:<16} {_pct(metric.precision)} {_pct(metric.recall)} "
+            f"{_pct(metric.f1)}   {metric.n_pred:>4} {metric.n_gold:>4} {metric.n_match:>4}"
+        )
+    lines.append("  (* = critical field)")
+    return lines
+
+
+def _format_sweep_table(report: ScoreReport, coarse_step: int = 5) -> list[str]:
+    lines = [
+        "Threshold sweep (critical fields on the auto-accepted subset):",
+        f"  {'thr':>5} {'accept':>7} {'accept%':>8} {'crit P':>8} {'crit R':>8}",
+    ]
+    for index, row in enumerate(report.sweep):
+        # Print a coarse grid plus the final threshold to keep it readable.
+        is_grid = index % coarse_step == 0 or index == len(report.sweep) - 1
+        if not is_grid:
+            continue
+        lines.append(
+            f"  {row.threshold:>5.2f} {row.n_accepted:>7} {row.accept_rate * 100:>7.1f}% "
+            f"{_pct(row.crit_precision)} {_pct(row.crit_recall)}"
+        )
+    return lines
+
+
+def _format_confidence(report: ScoreReport) -> list[str]:
+    lines = ["Confidence distribution (cached scores):"]
+    for value, count in report.confidence_hist.items():
+        bar = "#" * count
+        lines.append(f"  {value:>5.2f}  {count:>3}  {bar}")
+    return lines
+
+
+def _format_routing(report: ScoreReport) -> list[str]:
+    target = smallest_threshold_meeting(report.sweep, TARGET_CRITICAL_PRECISION)
+    lines = [
+        "Operating point analysis (you choose the threshold):",
+        f"  Target: auto-accept precision on critical fields "
+        f"{report.critical_labeled or '(none labeled)'} >= "
+        f"{TARGET_CRITICAL_PRECISION:.2f}",
+    ]
+    if not report.critical_labeled:
+        lines.append(
+            "  This dataset labels none of total/tax/invoice_number, so critical "
+            "auto-accept precision cannot be measured here."
+        )
+    elif target is None:
+        lines.append(
+            "  No threshold in the sweep reaches the target with any "
+            "auto-accepted document (see the confidence distribution above)."
+        )
+    else:
+        lines.append(
+            f"  Lowest qualifying threshold: {target.threshold:.2f} "
+            f"(accept {target.n_accepted}/{target.n_total} = "
+            f"{target.accept_rate * 100:.1f}%, crit P {_pct(target.crit_precision)}, "
+            f"crit R {_pct(target.crit_recall)})."
+        )
+    return lines
+
+
+def format_report(report: ScoreReport) -> str:
+    """Render a :class:`ScoreReport` as a plain-text report.
+
+    Args:
+        report: The computed score report.
+
+    Returns:
+        A multi-line string ready to print.
+    """
+    header = [
+        "=" * 68,
+        f"Evaluation: {report.dataset}  (n={report.n}, errors={report.n_error})",
+        f"Labeled fields: {', '.join(report.labeled_fields) or '(none)'}",
+        "=" * 68,
+    ]
+    sections = [
+        header,
+        _format_field_table(report),
+        _format_confidence(report),
+        _format_sweep_table(report),
+        _format_routing(report),
+    ]
+    return "\n\n".join("\n".join(section) for section in sections)

pyproject.toml

@@ -8,9 +8,11 @@ authors = [
 ]
 requires-python = ">=3.11"
 dependencies = [
+    "datasets[vision]>=5.0.0",
     "docling>=2.107.0",
     "google-genai>=2.10.0",
     "gradio>=6.19.0",
+    "pillow>=12.2.0",
     "pydantic>=2.13.4",
     "pydantic-settings>=2.14.2",
     "watchdog>=6.0.0",
@@ -25,6 +27,9 @@ line-length = 100
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+# Repo root on the path so tests can import the top-level `eval` package
+# (the harness lives at ./eval, outside the installed src/doc_agent package).
+pythonpath = ["."]
 
 [dependency-groups]
 dev = [

tests/test_eval.py

@@ -0,0 +1,314 @@
+"""Unit tests for the evaluation harness (build-plan phase 5).
+
+Fully offline: no model calls, no dataset downloads. The comparison function is
+tested directly, and the score-phase computation is tested on hand-built cached
+entries with known, hand-computed metrics. The threshold sweep is tested for the
+hard-failure override and the precision/recall trade-off.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from eval.cache import read_entries, report_from_dict, write_entry
+from eval.metrics import (
+    THRESHOLDS,
+    compute_field_metrics,
+    confidence_histogram,
+    smallest_threshold_meeting,
+    sweep_thresholds,
+)
+from eval.normalize import is_present, normalize, values_match
+from eval.score import build_report
+
+
+# ---------------------------------------------------------------------------
+# Comparison function (normalize / values_match)
+# ---------------------------------------------------------------------------
+
+
+def test_money_matches_cent_exact() -> None:
+    """Monetary values match only when equal at cent precision."""
+    assert values_match("total", 193.0, "193.00")
+    assert values_match("total", "1,234.56", "1234.56")
+    assert values_match("total", 100.004, 100.0)  # sub-cent noise rounds away
+    assert not values_match("total", 100.0, 100.01)  # a genuine 1-cent difference
+    assert not values_match("total", 100.0, 105.0)
+
+
+def test_money_rejects_relative_tolerance_error() -> None:
+    """Regression: a materially-wrong total within 0.5% is NOT scored correct.
+
+    The eval comparator must not reuse the reconciliation relative epsilon, or a
+    $2 error on a $500 total (and $10 on $2000) would inflate critical precision.
+    """
+    assert not values_match("total", 502.0, "500.00")   # within money_close's 0.5% band
+    assert not values_match("total", 2010.0, 2000.0)     # +/-$10 window at $2000
+    assert not values_match("tax", 9.05, 9.0)            # 5-cent tax error
+
+
+def test_money_handles_currency_symbols_and_separators() -> None:
+    """Currency symbols and thousands separators normalize away."""
+    assert normalize("total", "$1,000.00") == pytest.approx(1000.0)
+    assert values_match("total", "RM 193.00", "193.0")
+
+
+def test_date_matches_day_first_format() -> None:
+    """SROIE-style day-first dates match the ISO-cached prediction."""
+    # gold "15/01/2019" (D/M/Y) vs predicted cached ISO "2019-01-15".
+    assert values_match("document_date", "2019-01-15", "15/01/2019")
+    assert not values_match("document_date", "2019-01-16", "15/01/2019")
+
+
+def test_text_matches_case_and_whitespace_insensitive() -> None:
+    """Text matches after lower-casing and whitespace collapsing."""
+    assert values_match("vendor_name", "OJC  Marketing   SDN BHD", "ojc marketing sdn bhd")
+    assert not values_match("vendor_name", "Acme Corp", "Beta LLC")
+
+
+def test_absent_values_never_match() -> None:
+    """A present prediction against absent gold (or vice versa) is not a match."""
+    assert not values_match("total", 100.0, None)
+    assert not values_match("total", None, 100.0)
+    assert not values_match("vendor_name", "", "acme")
+    assert not is_present("vendor_name", "   ")
+    assert not is_present("total", "N/A")
+
+
+def test_unparseable_money_is_absent() -> None:
+    """An unparseable monetary string normalizes to None (absent), not a crash."""
+    assert normalize("total", "not a number") is None
+    assert not is_present("total", "abc")
+
+
+# ---------------------------------------------------------------------------
+# Synthetic cached entries with known metrics
+# ---------------------------------------------------------------------------
+
+
+def _validation(hard_failed: bool, *, hard_codes: list[str] | None = None) -> dict[str, Any]:
+    """Build a minimal validation dict as ValidationReport.to_dict would."""
+    results = []
+    for code in hard_codes or []:
+        results.append(
+            {"code": code, "severity": "hard", "status": "fail", "message": "synthetic"}
+        )
+    return {
+        "hard_failed": hard_failed,
+        "results": results,
+        "hard_failures": list(hard_codes or []),
+        "soft_failures": [],
+    }
+
+
+def _entry(
+    example_id: str,
+    *,
+    predicted: dict[str, Any],
+    gold: dict[str, Any],
+    confidence: float,
+    hard_failed: bool = False,
+    hard_codes: list[str] | None = None,
+    labeled: tuple[str, ...] = ("vendor_name", "vendor_address", "document_date", "total"),
+) -> dict[str, Any]:
+    return {
+        "id": example_id,
+        "dataset": "synthetic",
+        "gold": gold,
+        "labeled_fields": list(labeled),
+        "predicted": predicted,
+        "confidence": confidence,
+        "decision": "review",
+        "modality": "image",
+        "backend": "stub",
+        "validation": _validation(hard_failed, hard_codes=hard_codes),
+        "error": None,
+    }
+
+
+@pytest.fixture
+def synthetic_entries() -> list[dict[str, Any]]:
+    """Four entries with hand-computable per-field metrics.
+
+    total field: preds present on all 4; golds present on all 4;
+      - e1 correct, e2 correct, e3 wrong value, e4 correct  -> 3/4 match.
+    vendor_name: preds present on 3 (e4 missing); golds present on 4;
+      - e1 correct, e2 correct, e3 correct                  -> 3 match.
+      => precision 3/3 = 1.0, recall 3/4 = 0.75.
+    """
+    return [
+        _entry(
+            "e1",
+            predicted={"vendor_name": "Acme", "total": 100.0},
+            gold={"vendor_name": "acme", "total": "100.00", "vendor_address": None,
+                  "document_date": None},
+            confidence=0.90,
+        ),
+        _entry(
+            "e2",
+            predicted={"vendor_name": "Beta", "total": 50.0},
+            gold={"vendor_name": "beta", "total": "50.00", "vendor_address": None,
+                  "document_date": None},
+            confidence=0.80,
+        ),
+        _entry(
+            "e3",
+            predicted={"vendor_name": "Gamma", "total": 999.0},  # total wrong
+            gold={"vendor_name": "gamma", "total": "10.00", "vendor_address": None,
+                  "document_date": None},
+            confidence=0.70,
+        ),
+        _entry(
+            "e4",
+            predicted={"vendor_name": None, "total": 25.0},  # vendor missing
+            gold={"vendor_name": "delta", "total": "25.00", "vendor_address": None,
+                  "document_date": None},
+            confidence=0.60,
+        ),
+    ]
+
+
+def test_field_metrics_match_hand_computed(synthetic_entries: list[dict[str, Any]]) -> None:
+    """Per-field precision/recall/F1 equal the hand-computed values."""
+    metrics = {m.field: m for m in compute_field_metrics(
+        synthetic_entries, ("vendor_name", "total"))}
+
+    total = metrics["total"]
+    assert (total.n_pred, total.n_gold, total.n_match) == (4, 4, 3)
+    assert total.precision == pytest.approx(0.75)
+    assert total.recall == pytest.approx(0.75)
+    assert total.f1 == pytest.approx(0.75)
+
+    vendor = metrics["vendor_name"]
+    assert (vendor.n_pred, vendor.n_gold, vendor.n_match) == (3, 4, 3)
+    assert vendor.precision == pytest.approx(1.0)
+    assert vendor.recall == pytest.approx(0.75)
+    assert vendor.f1 == pytest.approx(2 * 1.0 * 0.75 / (1.0 + 0.75))
+
+
+def test_field_precision_none_when_no_prediction() -> None:
+    """Precision is None (undefined) when nothing was predicted for a field."""
+    entries = [
+        _entry("e1", predicted={"total": None}, gold={"total": "5.00"}, confidence=0.9),
+    ]
+    (metric,) = compute_field_metrics(entries, ("total",))
+    assert metric.n_pred == 0
+    assert metric.precision is None
+    assert metric.recall == pytest.approx(0.0)
+
+
+# ---------------------------------------------------------------------------
+# Threshold sweep
+# ---------------------------------------------------------------------------
+
+
+def test_sweep_accept_count_falls_as_threshold_rises(
+    synthetic_entries: list[dict[str, Any]],
+) -> None:
+    """Higher thresholds auto-accept no more documents than lower ones."""
+    rows = sweep_thresholds(synthetic_entries, ("total",), THRESHOLDS)
+    accept_counts = [row.n_accepted for row in rows]
+    assert accept_counts == sorted(accept_counts, reverse=True)
+    # At 0.50 every clean doc (confidence >= 0.50) is accepted; all 4 here.
+    assert rows[0].threshold == 0.50
+    assert rows[0].n_accepted == 4
+
+
+def test_sweep_hard_failure_never_accepted() -> None:
+    """A hard-failed document is forced to review at every threshold."""
+    entries = [
+        _entry(
+            "hard",
+            predicted={"total": 100.0},
+            gold={"total": "100.00"},
+            confidence=0.99,  # very confident...
+            hard_failed=True,
+            hard_codes=["H2"],  # ...but a hard rule failed.
+        ),
+    ]
+    rows = sweep_thresholds(entries, ("total",), THRESHOLDS)
+    assert all(row.n_accepted == 0 for row in rows)
+    # And the reconstructed report reports the hard failure.
+    report = report_from_dict(entries[0]["validation"])
+    assert report.hard_failed is True
+
+
+def test_sweep_critical_precision_and_recall(
+    synthetic_entries: list[dict[str, Any]],
+) -> None:
+    """At threshold 0.50 the critical (total) precision/recall match hand calc.
+
+    All 4 accepted; total correct on 3/4 => precision 0.75; gold present on 4 =>
+    recall 3/4 = 0.75.
+    """
+    rows = sweep_thresholds(synthetic_entries, ("total",), THRESHOLDS)
+    row_050 = rows[0]
+    assert row_050.crit_pred == 4
+    assert row_050.crit_match == 3
+    assert row_050.crit_precision == pytest.approx(0.75)
+    assert row_050.crit_recall == pytest.approx(0.75)
+
+
+def test_smallest_threshold_meeting_target() -> None:
+    """The lowest qualifying threshold is found when a clean high-conf doc exists."""
+    entries = [
+        _entry("ok", predicted={"total": 10.0}, gold={"total": "10.00"}, confidence=0.95),
+    ]
+    rows = sweep_thresholds(entries, ("total",), THRESHOLDS)
+    target = smallest_threshold_meeting(rows, 0.98)
+    assert target is not None
+    # confidence 0.95 accepts for thresholds <= 0.95; precision is 1.0 (perfect).
+    assert target.threshold == 0.50
+    assert target.crit_precision == pytest.approx(1.0)
+
+
+def test_smallest_threshold_meeting_none_when_unreachable() -> None:
+    """Returns None when no threshold reaches the target precision."""
+    entries = [
+        _entry("bad", predicted={"total": 999.0}, gold={"total": "10.00"}, confidence=0.95),
+    ]
+    rows = sweep_thresholds(entries, ("total",), THRESHOLDS)
+    assert smallest_threshold_meeting(rows, 0.98) is None
+
+
+def test_confidence_histogram_counts() -> None:
+    """The histogram buckets rounded confidences."""
+    entries = [
+        _entry("a", predicted={}, gold={}, confidence=0.50),
+        _entry("b", predicted={}, gold={}, confidence=0.50),
+        _entry("c", predicted={}, gold={}, confidence=0.40),
+    ]
+    hist = confidence_histogram(entries)
+    assert hist == {0.40: 1, 0.50: 2}
+
+
+# ---------------------------------------------------------------------------
+# End-to-end score phase over a written cache (still offline)
+# ---------------------------------------------------------------------------
+
+
+def test_build_report_from_written_cache(
+    tmp_path: Path, synthetic_entries: list[dict[str, Any]]
+) -> None:
+    """Writing entries then building a report round-trips and computes metrics."""
+    for entry in synthetic_entries:
+        write_entry(tmp_path, "synthetic", entry)
+
+    assert len(read_entries(tmp_path, "synthetic")) == 4
+
+    report = build_report("synthetic", cache_base=tmp_path)
+    assert report.n == 4
+    assert "total" in report.labeled_fields
+    # SROIE-like labeling: total is the only critical field labeled here.
+    assert report.critical_labeled == ("total",)
+    total = next(m for m in report.field_metrics if m.field == "total")
+    assert total.n_match == 3
+
+
+def test_build_report_raises_without_cache(tmp_path: Path) -> None:
+    """Scoring a dataset with no cache raises a clear error."""
+    with pytest.raises(FileNotFoundError):
+        build_report("missing", cache_base=tmp_path)