diff --git a/README.md b/README.md
index 29d1585..c8b6c5c 100644
--- a/README.md
+++ b/README.md
@@ -84,12 +84,18 @@ Targeted allocation is experimental and remains opt-in because it conditions lat
 To avoid wasting judge calls on uninformative pairs, the judge skips comparisons where **both** outputs are shorter than `--min-chars` (default 20 — neither model produced meaningful text), and scores **identical** outputs as an automatic tie without calling the judge. Pass `--min-chars 0` to disable the length filter.
 
 > [!IMPORTANT]
-> If an existing results repository was judged with an older `ocr-bench`
-> checkout from output configs containing multiple appended `inference_info`
-> entries, run the first upgraded evaluation with `--full-rejudge`. Older
-> versions could pair the latest model identity with an earlier OCR text
-> column, and their incremental skip data does not record enough provenance to
-> distinguish those verdicts safely.
+> Results written before per-comparison resume provenance require a one-time
+> `--full-rejudge` or a new results repository. Older versions could pair the
+> latest model identity with an earlier OCR text column when configs contained
+> appended `inference_info` entries; legacy skip data cannot establish which
+> inputs produced a verdict.
+>
+> New comparisons and checkpoints record a fingerprint of the selected source
+> rows/images, OCR contents, resolved model and original output column, judge
+> configuration, and text/image preprocessing settings. Resuming with changed
+> inputs stops before reusing verdicts and requires `--full-rejudge` or a new
+> results repository. Unchanged pairs can still resume when adding a new model,
+> including after an interrupted run that published only a comparisons checkpoint.
 
 **`ocr-bench view`** serves a local web viewer with a leaderboard, comparison browser, and human validation. Vote on comparisons to cross-check the automated judge with human judgement.
 
diff --git a/src/ocr_bench/cli.py b/src/ocr_bench/cli.py
index 41de068..8a271d3 100644
--- a/src/ocr_bench/cli.py
+++ b/src/ocr_bench/cli.py
@@ -32,6 +32,7 @@
     DatasetError,
     discover_configs,
     discover_pr_configs,
+    evaluation_input_fingerprints,
     load_config_dataset,
     load_flat_dataset,
 )
@@ -58,6 +59,8 @@
 )
 from ocr_bench.publish import (
     EvalMetadata,
+    evaluation_provenance_hash,
+    hash_judge_specs,
     load_existing_comparisons,
     load_existing_metadata,
     publish_checkpoint,
@@ -590,7 +593,10 @@ def _trim_to_budget(
 
 
 def _convert_results(
-    comparisons: list[Comparison], aggregated: list[dict]
+    comparisons: list[Comparison],
+    aggregated: list[dict],
+    *,
+    provenance_hashes: dict[tuple[str, str], str] | None = None,
 ) -> list[ComparisonResult]:
     """Convert judged comparisons + aggregated outputs into ComparisonResult list."""
     results: list[ComparisonResult] = []
@@ -614,6 +620,9 @@ def _convert_results(
                 col_b=comp.col_b,
                 truncated_a=comp.truncated_a,
                 truncated_b=comp.truncated_b,
+                provenance_hash=(provenance_hashes or {}).get(
+                    tuple(sorted((comp.col_a, comp.col_b))), ""
+                ),
             )
         )
     return results
@@ -872,10 +881,18 @@ def cmd_judge(args: argparse.Namespace) -> None:
         console.print(f"Results will be published to [bold]{results_repo}[/bold]")
 
     # --- Load dataset (cascading auto-detection) ---
+    source_configs: list[str] = []
+    source_output_columns: dict[str, str] = {}
     if args.configs:
         # Explicit configs — use them directly
         config_names = args.configs
-        ds, ocr_columns = load_config_dataset(args.dataset, config_names, split=args.split)
+        source_configs = list(config_names)
+        ds, ocr_columns = load_config_dataset(
+            args.dataset,
+            config_names,
+            split=args.split,
+            source_output_columns=source_output_columns,
+        )
     elif args.columns:
         # Explicit columns — flat loading
         ds, ocr_columns = load_flat_dataset(args.dataset, split=args.split, columns=args.columns)
@@ -885,12 +902,14 @@ def cmd_judge(args: argparse.Namespace) -> None:
         if not config_names:
             raise DatasetError("No configs found in open PRs")
         from_prs = True
+        source_configs = list(config_names)
         console.print(f"Discovered {len(config_names)} configs from PRs: {config_names}")
         ds, ocr_columns = load_config_dataset(
             args.dataset,
             config_names,
             split=args.split,
             pr_revisions=pr_revisions if not merge else None,
+            source_output_columns=source_output_columns,
         )
     else:
         # Auto-detect: PRs + main branch configs combined, fall back to flat
@@ -904,6 +923,7 @@ def cmd_judge(args: argparse.Namespace) -> None:
                 config_names.append(mc)
 
         if config_names:
+            source_configs = list(config_names)
             if pr_configs:
                 from_prs = True
                 console.print(f"Auto-detected {len(pr_configs)} configs from PRs: {pr_configs}")
@@ -916,11 +936,15 @@ def cmd_judge(args: argparse.Namespace) -> None:
                 config_names,
                 split=args.split,
                 pr_revisions=pr_revisions if pr_configs else None,
+                source_output_columns=source_output_columns,
             )
         else:
             # No configs anywhere — fall back to flat loading
             ds, ocr_columns = load_flat_dataset(args.dataset, split=args.split)
 
+    if not source_configs:
+        source_output_columns = {column: column for column in ocr_columns}
+
     console.print(f"Loaded {len(ds)} samples with {len(ocr_columns)} models:")
     for col, model in ocr_columns.items():
         console.print(f"  {col} → {model}")
@@ -961,6 +985,49 @@ def cmd_judge(args: argparse.Namespace) -> None:
     model_names = sorted(set(ocr_columns.values()) - set(failed_models))
     current_models = set(model_names)
 
+    model_specs = args.models or [DEFAULT_JUDGE]
+    judges = [
+        parse_judge_spec(spec, max_tokens=args.max_tokens, concurrency=args.concurrency)
+        for spec in model_specs
+    ]
+    is_jury = len(judges) > 1
+    judge_names = [judge.name for judge in judges]
+    judge_spec_hashes = hash_judge_specs(model_specs)
+    requested_max_samples = args.max_samples or len(ds)
+    source_fingerprint = ""
+    source_column_fingerprints: dict[str, str] = {}
+    pair_provenance_hashes: dict[tuple[str, str], str] = {}
+    if results_repo:
+        evaluated_indices = sample_indices(len(ds), args.max_samples, args.seed)
+        source_fingerprint, source_column_fingerprints = evaluation_input_fingerprints(
+            ds, ocr_columns, evaluated_indices
+        )
+        from itertools import combinations as _provenance_pairs
+
+        for col_a, col_b in _provenance_pairs(ocr_columns, 2):
+            pair = (min(col_a, col_b), max(col_a, col_b))
+            pair_provenance_hashes[pair] = evaluation_provenance_hash(
+                source_dataset=args.dataset,
+                source_split=args.split,
+                source_fingerprint=source_fingerprint,
+                source_column_fingerprints={
+                    column: source_column_fingerprints[column] for column in pair
+                },
+                source_columns={column: ocr_columns[column] for column in pair},
+                source_output_columns={
+                    column: source_output_columns[column] for column in pair
+                },
+                judge_specs=model_specs,
+                seed=args.seed,
+                max_samples=requested_max_samples,
+                prompt_hash=criteria_prompt_hash,
+                judge_text_mode=args.judge_text_mode,
+                max_ocr_text_len=args.max_ocr_text_len,
+                judge_image_dim=args.judge_image_dim,
+                max_tokens=args.max_tokens,
+                min_chars=args.min_chars,
+            )
+
     # --- Incremental: load existing comparisons ---
     existing_results: list[ComparisonResult] = []
     # Comparisons outside this invocation's rankable model set (subset/retired
@@ -1030,57 +1097,92 @@ def cmd_judge(args: argparse.Namespace) -> None:
             )
 
         if existing_results or preserved_out_of_grid_results:
-            # Provenance guard: NEVER mix criteria rubrics on one board. The
-            # existing comparisons were judged under the profile recorded in the
-            # last metadata row (pre-#44 rows = the default profile, whose prompt
-            # is byte-identical to the old hardcoded one). Judging the rest — or
-            # even just refitting — under a different --criteria would merge
-            # incompatible verdicts into one ELO board AND republish the metadata
-            # mislabeled with the current run's criteria. Refuse and exit; the
-            # only safe way to change rubric is --full-rejudge (discards existing).
-            # Compare on the prompt HASH (the true rubric identity): this catches
-            # a built-in prompt edited across code versions, matches the same
-            # custom file re-run, and blocks two different custom files even if
-            # they share a basename.
-            prev_criteria, prev_hash = _existing_criteria_provenance(existing_meta_rows)
-            if prev_hash != criteria_prompt_hash:
-                if prev_criteria.startswith("custom:"):
-                    match_hint = (
-                        f"re-supply the same custom prompt file (recorded as "
-                        f"'{prev_criteria}', prompt {prev_hash})"
+            # Provenance-stamped comparison rows are authoritative. This matters
+            # after a changed --full-rejudge checkpoints but dies before it can
+            # append matching metadata: the latest metadata row still describes
+            # the preceding completed run. Apply the narrower metadata diagnostics
+            # only to legacy unstamped comparisons.
+            existing_hashes = {result.provenance_hash for result in existing_results}
+            if "" in existing_hashes and existing_meta_rows:
+                prev_criteria, prev_hash = _existing_criteria_provenance(existing_meta_rows)
+                if prev_hash != criteria_prompt_hash:
+                    if prev_criteria.startswith("custom:"):
+                        match_hint = (
+                            f"re-supply the same custom prompt file (recorded as "
+                            f"'{prev_criteria}', prompt {prev_hash})"
+                        )
+                    else:
+                        match_hint = f"re-run with [bold]--criteria {prev_criteria}[/bold]"
+                    console.print(
+                        f"[red]Error:[/red] criteria mismatch — "
+                        f"[bold]{results_repo}[/bold] was judged under criteria "
+                        f"'[bold]{prev_criteria}[/bold]' (prompt {prev_hash}), but "
+                        f"this run requested '[bold]{criteria}[/bold]' "
+                        f"(prompt {criteria_prompt_hash}). Mixing rubrics on one "
+                        f"leaderboard would produce meaningless rankings and "
+                        f"mislabeled metadata.\nTo match the existing results, "
+                        f"{match_hint}; or use [bold]--full-rejudge[/bold] to "
+                        f"discard them and re-judge everything under '{criteria}'."
                     )
-                else:
-                    match_hint = f"re-run with [bold]--criteria {prev_criteria}[/bold]"
+                    sys.exit(1)
+
+                previous_preprocessing = _existing_preprocessing_provenance(
+                    existing_meta_rows
+                )
+                requested_preprocessing = (
+                    args.judge_text_mode,
+                    args.max_ocr_text_len,
+                    args.judge_image_dim,
+                )
+                if previous_preprocessing != requested_preprocessing:
+                    prev_mode, prev_text_cap, prev_image_cap = previous_preprocessing
+                    console.print(
+                        f"[red]Error:[/red] judge preprocessing mismatch — "
+                        f"[bold]{results_repo}[/bold] was judged with "
+                        f"text_mode={prev_mode}, max_ocr_text_len={prev_text_cap}, "
+                        f"judge_image_dim={prev_image_cap}, but this run requested "
+                        f"text_mode={args.judge_text_mode}, "
+                        f"max_ocr_text_len={args.max_ocr_text_len}, "
+                        f"judge_image_dim={args.judge_image_dim}. Mixing these "
+                        "verdicts would make the leaderboard incomparable. Re-run "
+                        "with matching settings, or use "
+                        "[bold]--full-rejudge[/bold]."
+                    )
+                    sys.exit(1)
+
+            # Every reused row must have been produced under the exact same
+            # source/config/split/fingerprint, judge or jury, sample selection,
+            # prompt, and preprocessing. Checkpoint rows carry this hash too,
+            # so an interrupted first run can resume before final metadata exists.
+            if "" in existing_hashes:
                 console.print(
-                    f"[red]Error:[/red] criteria mismatch — [bold]{results_repo}[/bold] "
-                    f"was judged under criteria '[bold]{prev_criteria}[/bold]' "
-                    f"(prompt {prev_hash}), but this run requested "
-                    f"'[bold]{criteria}[/bold]' (prompt {criteria_prompt_hash}). "
-                    f"Mixing rubrics on one leaderboard would produce meaningless "
-                    f"rankings and mislabeled metadata.\n"
-                    f"To match the existing results, {match_hint}; or use "
-                    f"[bold]--full-rejudge[/bold] to discard them and re-judge "
-                    f"everything under '{criteria}'."
+                    f"[red]Error:[/red] existing comparisons in "
+                    f"[bold]{results_repo}[/bold] lack resume provenance. They were "
+                    "written by an older ocr-bench version (or an incomplete unsafe "
+                    "checkpoint), so their sample indices cannot be verified against "
+                    "the current source. Use [bold]--full-rejudge[/bold] once to "
+                    "replace them with provenance-stamped results."
                 )
                 sys.exit(1)
-
-            previous_preprocessing = _existing_preprocessing_provenance(existing_meta_rows)
-            requested_preprocessing = (
-                args.judge_text_mode,
-                args.max_ocr_text_len,
-                args.judge_image_dim,
-            )
-            if previous_preprocessing != requested_preprocessing:
-                prev_mode, prev_text_cap, prev_image_cap = previous_preprocessing
+            mismatched = []
+            for result in existing_results:
+                pair = (
+                    min(result.col_a, result.col_b),
+                    max(result.col_a, result.col_b),
+                )
+                expected = pair_provenance_hashes.get(pair)
+                if expected is None or result.provenance_hash != expected:
+                    mismatched.append(result)
+            if mismatched:
+                recorded = ", ".join(
+                    sorted({result.provenance_hash[:12] for result in mismatched})
+                )
                 console.print(
-                    f"[red]Error:[/red] judge preprocessing mismatch — "
-                    f"[bold]{results_repo}[/bold] was judged with "
-                    f"text_mode={prev_mode}, max_ocr_text_len={prev_text_cap}, "
-                    f"judge_image_dim={prev_image_cap}, but this run requested "
-                    f"text_mode={args.judge_text_mode}, "
-                    f"max_ocr_text_len={args.max_ocr_text_len}, "
-                    f"judge_image_dim={args.judge_image_dim}. Mixing these verdicts "
-                    "would make the leaderboard incomparable. Re-run with matching "
+                    f"[red]Error:[/red] resume provenance mismatch for "
+                    f"[bold]{results_repo}[/bold] — existing comparison hash(es) "
+                    f"{recorded}. The "
+                    "source repo/config/split/fingerprint, judge or jury, sampling, "
+                    "prompt, or preprocessing changed. Re-run with the original "
                     "settings, or use [bold]--full-rejudge[/bold]."
                 )
                 sys.exit(1)
@@ -1118,14 +1220,6 @@ def cmd_judge(args: argparse.Namespace) -> None:
         }
     )
 
-    # --- Judge setup (shared by both paths) ---
-    model_specs = args.models or [DEFAULT_JUDGE]
-    judges = [
-        parse_judge_spec(spec, max_tokens=args.max_tokens, concurrency=args.concurrency)
-        for spec in model_specs
-    ]
-    is_jury = len(judges) > 1
-
     console.print(f"Judge criteria: [bold]{criteria}[/bold] (prompt {criteria_prompt_hash})")
     logger.info("judge_criteria", criteria=criteria, prompt_hash=criteria_prompt_hash)
 
@@ -1146,7 +1240,11 @@ def _judge_batch(batch_comps: list[Comparison]) -> list[ComparisonResult]:
         else:
             aggregated = []
         merged = _merge_auto_ties(batch_comps, aggregated)
-        return _convert_results(batch_comps, merged)
+        return _convert_results(
+            batch_comps,
+            merged,
+            provenance_hashes=pair_provenance_hashes,
+        )
 
     # Set when the run stops because it hit --max-comparisons (vs converging or
     # running out of samples). Recorded in the metadata row and drives the
@@ -1419,7 +1517,13 @@ def _judge_batch(batch_comps: list[Comparison]) -> list[ComparisonResult]:
                 metadata = EvalMetadata(
                     source_dataset=args.dataset,
                     source_split=args.split,
-                    judge_models=[],
+                    source_fingerprint=source_fingerprint,
+                    source_column_fingerprints=source_column_fingerprints,
+                    source_configs=source_configs,
+                    source_columns=ocr_columns,
+                    source_output_columns=source_output_columns,
+                    judge_models=judge_names,
+                    judge_spec_hashes=judge_spec_hashes,
                     seed=args.seed,
                     max_samples=args.max_samples or len(ds),
                     total_comparisons=0,
@@ -1434,6 +1538,7 @@ def _judge_batch(batch_comps: list[Comparison]) -> list[ComparisonResult]:
                     max_ocr_text_len=args.max_ocr_text_len,
                     judge_image_dim=args.judge_image_dim,
                     judge_text_mode=args.judge_text_mode,
+                    min_chars=args.min_chars,
                     adaptive_strategy=adaptive_strategy,
                     size_tie_ratio=args.size_tie_ratio,
                     size_tie_min_samples=args.size_tie_min_samples,
@@ -1547,7 +1652,13 @@ def _judge_batch(batch_comps: list[Comparison]) -> list[ComparisonResult]:
         metadata = EvalMetadata(
             source_dataset=args.dataset,
             source_split=args.split,
-            judge_models=[j.name for j in judges],
+            source_fingerprint=source_fingerprint,
+            source_column_fingerprints=source_column_fingerprints,
+            source_configs=source_configs,
+            source_columns=ocr_columns,
+            source_output_columns=source_output_columns,
+            judge_models=judge_names,
+            judge_spec_hashes=judge_spec_hashes,
             seed=args.seed,
             max_samples=args.max_samples or len(ds),
             total_comparisons=total_comparisons,
@@ -1563,6 +1674,7 @@ def _judge_batch(batch_comps: list[Comparison]) -> list[ComparisonResult]:
             max_ocr_text_len=args.max_ocr_text_len,
             judge_image_dim=args.judge_image_dim,
             judge_text_mode=args.judge_text_mode,
+            min_chars=args.min_chars,
             adaptive_strategy=adaptive_strategy,
             size_tie_ratio=args.size_tie_ratio,
             size_tie_min_samples=args.size_tie_min_samples,
diff --git a/src/ocr_bench/dataset.py b/src/ocr_bench/dataset.py
index 08f76bb..b9741f0 100644
--- a/src/ocr_bench/dataset.py
+++ b/src/ocr_bench/dataset.py
@@ -2,10 +2,11 @@
 
 from __future__ import annotations
 
+import hashlib
 import json
 import math
 from dataclasses import dataclass, field
-from typing import Literal
+from typing import Any, Literal
 
 import structlog
 from datasets import Dataset, get_dataset_config_names, load_dataset
@@ -18,6 +19,75 @@ class DatasetError(Exception):
     """Raised when dataset loading or column discovery fails."""
 
 
+def _hash_value(hasher: Any, value: Any) -> None:
+    """Add one typed value to a hashlib-compatible digest deterministically."""
+    update = hasher.update
+    if isinstance(value, bytes):
+        payload = b"bytes:" + value
+    elif hasattr(value, "tobytes") and hasattr(value, "size"):
+        # PIL images: include geometry/mode as well as pixels.
+        payload = (
+            f"image:{getattr(value, 'mode', '')}:{getattr(value, 'size', '')}:".encode()
+            + value.tobytes()
+        )
+    else:
+        payload = json.dumps(
+            value,
+            sort_keys=True,
+            separators=(",", ":"),
+            default=repr,
+        ).encode()
+    update(len(payload).to_bytes(8, "big"))
+    update(payload)
+
+
+def evaluation_input_fingerprints(
+    dataset: Dataset,
+    ocr_columns: dict[str, str],
+    indices: list[int],
+) -> tuple[str, dict[str, str]]:
+    """Fingerprint evaluated pages and each OCR column over the sampled rows.
+
+    The page fingerprint includes stable passthrough identity columns when
+    present and image pixels when available. OCR columns are fingerprinted
+    independently so adding a new model does not invalidate safe comparisons
+    between unchanged existing models.
+    """
+    identity_columns = [key for key in ALIGNMENT_KEYS if key in dataset.column_names]
+    image_columns = [
+        col for col in dataset.column_names if col == "image" or "image" in col.lower()
+    ]
+    if image_columns:
+        # IDs detect reordering; image content also catches a page being replaced
+        # in place while retaining the same ID.
+        image_column = "image" if "image" in image_columns else image_columns[0]
+        identity_columns.append(image_column)
+    if not identity_columns:
+        raise DatasetError(
+            "Cannot fingerprint evaluated pages: no stable identity or image column found"
+        )
+
+    selected = dataset.select(indices)
+    page_digest = hashlib.sha256()
+    for local_idx, source_idx in enumerate(indices):
+        row = selected[local_idx]
+        _hash_value(page_digest, source_idx)
+        for column in identity_columns:
+            _hash_value(page_digest, column)
+            _hash_value(page_digest, row[column])
+
+    column_fingerprints: dict[str, str] = {}
+    for column in ocr_columns:
+        digest = hashlib.sha256()
+        values = selected[column]
+        for source_idx, value in zip(indices, values):
+            _hash_value(digest, source_idx)
+            _hash_value(digest, value)
+        column_fingerprints[column] = digest.hexdigest()
+
+    return page_digest.hexdigest(), column_fingerprints
+
+
 # ---------------------------------------------------------------------------
 # Cross-config row alignment (issue #5)
 # ---------------------------------------------------------------------------
@@ -414,11 +484,14 @@ def load_config_dataset(
     config_names: list[str],
     split: str = "train",
     pr_revisions: dict[str, str] | None = None,
+    *,
+    source_output_columns: dict[str, str] | None = None,
 ) -> tuple[Dataset, dict[str, str]]:
     """Load multiple configs from a Hub dataset and merge into one.
 
     Each config becomes a column whose name is the config name and whose value
-    is the OCR text (from the first column matching heuristics, or ``markdown``).
+    is the OCR text selected together with its model identity from inference
+    metadata, falling back to column heuristics for legacy configs.
 
     Before merging, row alignment is verified across configs on shared
     passthrough columns (see :data:`ALIGNMENT_KEYS`). A mismatch raises
@@ -430,6 +503,8 @@ def load_config_dataset(
         config_names: List of config names to load.
         split: Dataset split to load.
         pr_revisions: Optional mapping of config_name → revision for PR-based loading.
+        source_output_columns: Optional output mapping populated with each loaded
+            config's original OCR column name, before renaming it for the merge.
 
     Returns:
         Tuple of (unified Dataset, {column_name: model_id}).
@@ -497,6 +572,8 @@ def load_config_dataset(
         if text_col is None:  # filtered into `usable` above; narrows the type
             continue
         ocr_columns[config] = lc.model_id
+        if source_output_columns is not None:
+            source_output_columns[config] = text_col
 
         # Build unified dataset using Arrow-level ops (no per-row image decode).
         # Row counts are already verified equal above, so no truncation is needed.
diff --git a/src/ocr_bench/elo.py b/src/ocr_bench/elo.py
index dc05268..f3dfcd3 100644
--- a/src/ocr_bench/elo.py
+++ b/src/ocr_bench/elo.py
@@ -33,6 +33,9 @@ class ComparisonResult:
     col_b: str = ""
     truncated_a: bool = False
     truncated_b: bool = False
+    # Hash of the source/judge/sampling provenance for safe incremental resume.
+    # Historical rows lack it and must not be silently reused.
+    provenance_hash: str = ""
 
 
 @dataclass
@@ -314,6 +317,7 @@ def compute_elo(
                 # flip: truncated_a is always about model_a's output.
                 "truncated_a": r.truncated_a,
                 "truncated_b": r.truncated_b,
+                "provenance_hash": r.provenance_hash,
             }
         )
 
diff --git a/src/ocr_bench/publish.py b/src/ocr_bench/publish.py
index 52543cd..72b9d3f 100644
--- a/src/ocr_bench/publish.py
+++ b/src/ocr_bench/publish.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import datetime
+import hashlib
 import json
 from dataclasses import dataclass, field
 
@@ -19,7 +20,7 @@
     practical_preferences,
 )
 from ocr_bench.elo import ComparisonResult, Leaderboard, compute_elo
-from ocr_bench.judge import MAX_IMAGE_DIM, MAX_OCR_TEXT_LENGTH
+from ocr_bench.judge import DEFAULT_MIN_CHARS, MAX_IMAGE_DIM, MAX_OCR_TEXT_LENGTH
 from ocr_bench.run import MODEL_REGISTRY
 
 logger = structlog.get_logger()
@@ -53,6 +54,16 @@ class EvalMetadata:
     # metadata rows predate this field and are interpreted as ``train`` by the
     # viewer for backward compatibility.
     source_split: str = "train"
+    # Exact loaded-dataset identity and selection used by the resume guard.
+    source_fingerprint: str = ""
+    source_column_fingerprints: dict[str, str] = field(default_factory=dict)
+    source_configs: list[str] = field(default_factory=list)
+    source_columns: dict[str, str] = field(default_factory=dict)
+    # Unified column -> original output column selected from the source config.
+    source_output_columns: dict[str, str] = field(default_factory=dict)
+    # One-way identities of the raw CLI judge specs. Raw endpoint specs may
+    # contain credentials or private hostnames and must never be published.
+    judge_spec_hashes: list[str] = field(default_factory=list)
     auto_tied: int = 0
     # Global comparison budget for the run (--max-comparisons); None = uncapped.
     # ``budget_exhausted`` records whether the run stopped because it hit the cap
@@ -76,6 +87,7 @@ class EvalMetadata:
     # "normalized" (HTML flattened before the cap) or "raw" (capped as-is).
     # Changes verdicts, so it's provenance alongside the caps.
     judge_text_mode: str = "normalized"
+    min_chars: int = DEFAULT_MIN_CHARS
     # Comparison-allocation provenance. ``balanced`` is the historical default;
     # ``targeted`` tops up only unresolved adjacent pairs after enough balanced evidence.
     adaptive_strategy: str = "balanced"
@@ -136,6 +148,7 @@ def load_existing_comparisons(repo_id: str) -> list[ComparisonResult]:
                     col_b=row.get("col_b", ""),
                     truncated_a=row.get("truncated_a", False),
                     truncated_b=row.get("truncated_b", False),
+                    provenance_hash=row.get("provenance_hash", ""),
                 )
             )
     except Exception as exc:
@@ -171,6 +184,56 @@ def _get_model_sizes() -> dict[str, str]:
     return {cfg.model_id: cfg.size for cfg in MODEL_REGISTRY.values()}
 
 
+def evaluation_provenance_hash(
+    *,
+    source_dataset: str,
+    source_split: str,
+    source_fingerprint: str,
+    source_column_fingerprints: dict[str, str],
+    source_columns: dict[str, str],
+    source_output_columns: dict[str, str],
+    judge_specs: list[str],
+    seed: int,
+    max_samples: int,
+    prompt_hash: str,
+    judge_text_mode: str,
+    max_ocr_text_len: int,
+    judge_image_dim: int,
+    max_tokens: int,
+    min_chars: int,
+) -> str:
+    """Hash every input that determines which verdicts are safe to reuse.
+
+    The comparison config carries this hash on every row, including mid-run
+    checkpoints. That lets a killed first run resume safely before a final
+    metadata row exists, while results produced before this guard fail closed.
+    """
+    payload = {
+        "source_dataset": source_dataset,
+        "source_split": source_split,
+        "source_fingerprint": source_fingerprint,
+        "source_column_fingerprints": source_column_fingerprints,
+        "source_columns": source_columns,
+        "source_output_columns": source_output_columns,
+        "judge_specs": judge_specs,
+        "seed": seed,
+        "max_samples": max_samples,
+        "prompt_hash": prompt_hash,
+        "judge_text_mode": judge_text_mode,
+        "max_ocr_text_len": max_ocr_text_len,
+        "judge_image_dim": judge_image_dim,
+        "max_tokens": max_tokens,
+        "min_chars": min_chars,
+    }
+    canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"))
+    return hashlib.sha256(canonical.encode()).hexdigest()
+
+
+def hash_judge_specs(judge_specs: list[str]) -> list[str]:
+    """Return publish-safe identities for potentially sensitive judge specs."""
+    return [hashlib.sha256(spec.encode()).hexdigest() for spec in judge_specs]
+
+
 def build_leaderboard_rows(
     board: Leaderboard,
     failed_models: list[str] | None = None,
@@ -238,7 +301,15 @@ def build_metadata_row(metadata: EvalMetadata) -> dict:
     return {
         "source_dataset": metadata.source_dataset,
         "source_split": metadata.source_split,
+        "source_fingerprint": metadata.source_fingerprint,
+        "source_column_fingerprints": json.dumps(
+            metadata.source_column_fingerprints, sort_keys=True
+        ),
+        "source_configs": json.dumps(metadata.source_configs),
+        "source_columns": json.dumps(metadata.source_columns, sort_keys=True),
+        "source_output_columns": json.dumps(metadata.source_output_columns, sort_keys=True),
         "judge_models": json.dumps(metadata.judge_models),
+        "judge_spec_hashes": json.dumps(metadata.judge_spec_hashes),
         "seed": metadata.seed,
         "max_samples": metadata.max_samples,
         "total_comparisons": metadata.total_comparisons,
@@ -255,6 +326,7 @@ def build_metadata_row(metadata: EvalMetadata) -> dict:
         "max_ocr_text_len": metadata.max_ocr_text_len,
         "judge_image_dim": metadata.judge_image_dim,
         "judge_text_mode": metadata.judge_text_mode,
+        "min_chars": metadata.min_chars,
         "adaptive_strategy": metadata.adaptive_strategy,
         "size_tie_ratio": metadata.size_tie_ratio,
         "size_tie_min_samples": metadata.size_tie_min_samples,
diff --git a/tests/test_budget_checkpoint.py b/tests/test_budget_checkpoint.py
index a417c35..afefb5e 100644
--- a/tests/test_budget_checkpoint.py
+++ b/tests/test_budget_checkpoint.py
@@ -7,9 +7,12 @@
 
 from __future__ import annotations
 
+import json
+from contextlib import nullcontext
 from unittest.mock import patch
 
 import pytest
+from datasets import Dataset
 from PIL import Image
 
 from ocr_bench import cli
@@ -26,16 +29,26 @@ def __init__(self, n: int, columns: dict[str, list[str]]):
         self._n = n
         self._columns = columns
         self._img = Image.new("RGB", (8, 8), "white")
+        # cmd_judge requires the same stable identity real HF Datasets expose.
+        self._fingerprint = f"fake-{n}-{'-'.join(columns)}"
 
     def __len__(self) -> int:
         return self._n
 
     @property
     def column_names(self) -> list[str]:
-        return list(self._columns)
+        return [*self._columns, "image"]
+
+    def select(self, indices: list[int]):
+        return FakeDataset(
+            len(indices),
+            {column: [values[i] for i in indices] for column, values in self._columns.items()},
+        )
 
     def __getitem__(self, key):
         if isinstance(key, str):
+            if key == "image":
+                return [self._img] * self._n
             return self._columns[key]
         row = {c: self._columns[c][key] for c in self._columns}
         row["image"] = self._img
@@ -84,21 +97,32 @@ def _run_judge(
     existing: list[ComparisonResult] | None = None,
     judge: FakeJudge | None = None,
     checkpoint_side_effect=None,
+    stamp_existing_provenance: bool = True,
+    checkpoint_resume: bool = False,
+    existing_meta_override: list[dict] | None = None,
+    config_datasets: dict[str, Dataset] | None = None,
 ):
     """Run cmd_judge with dataset load, judge, and Hub calls patched out."""
     judge = judge or FakeJudge()
     argv = [
         "judge",
         "user/ds",
-        "--columns",
-        *ocr_columns.keys(),
+        "--configs" if config_datasets is not None else "--columns",
+        *(config_datasets if config_datasets is not None else ocr_columns),
         "--save-results",
         "user/results",
         *argv_extra,
     ]
     args = build_parser().parse_args(argv)
+    if existing and stamp_existing_provenance:
+        model_to_column = {model: column for column, model in ocr_columns.items()}
+        for result in existing:
+            if result.model_a in model_to_column and result.model_b in model_to_column:
+                result.col_a = result.col_a or model_to_column[result.model_a]
+                result.col_b = result.col_b or model_to_column[result.model_b]
+                result.provenance_hash = "matching-test-provenance"
     existing_meta = []
-    if existing:
+    if existing and not checkpoint_resume:
         existing_meta = [
             {
                 "criteria": "default",
@@ -108,12 +132,30 @@ def _run_judge(
                 "judge_image_dim": 1024,
             }
         ]
+    if existing_meta_override is not None:
+        existing_meta = existing_meta_override
+    provenance_patch = (
+        patch.object(
+            cli,
+            "evaluation_provenance_hash",
+            return_value="matching-test-provenance",
+        )
+        if stamp_existing_provenance
+        else nullcontext()
+    )
 
     with (
         patch.object(cli, "load_flat_dataset", return_value=(ds, ocr_columns)),
+        patch(
+            "ocr_bench.dataset.load_dataset",
+            side_effect=lambda path, name, **kwargs: config_datasets[name],
+        )
+        if config_datasets is not None
+        else nullcontext(),
         patch.object(cli, "parse_judge_spec", return_value=judge),
         patch.object(cli, "load_existing_comparisons", return_value=existing or []),
         patch.object(cli, "load_existing_metadata", return_value=existing_meta),
+        provenance_patch,
         patch.object(cli, "publish_results") as m_publish,
         patch.object(cli, "publish_checkpoint") as m_checkpoint,
     ):
@@ -206,6 +248,9 @@ def test_resume_reconstructs_target_pairs_without_filling_sparse_grid(self):
                 winner=row["winner"],
                 reason=row.get("reason", ""),
                 agreement=row.get("agreement", "1/1"),
+                col_a=row.get("col_a", ""),
+                col_b=row.get("col_b", ""),
+                provenance_hash=row.get("provenance_hash", ""),
             )
             for row in first_board.comparison_log
         ]
@@ -522,6 +567,221 @@ def test_checkpoints_fire_in_adaptive_mode(self):
 
 
 class TestResume:
+    @staticmethod
+    def _saved_results(published):
+        return [
+            ComparisonResult(
+                sample_idx=row["sample_idx"],
+                model_a=row["model_a"],
+                model_b=row["model_b"],
+                winner=row["winner"],
+                col_a=row["col_a"],
+                col_b=row["col_b"],
+                provenance_hash=row["provenance_hash"],
+            )
+            for row in published.call_args.args[1].comparison_log
+        ]
+
+    @staticmethod
+    def _configs(ds, ocr, *, output_column="old_ocr", incomplete=False):
+        configs = {}
+        for column, model in ocr.items():
+            entries = [{"column_name": output_column, "model_id": model}]
+            if incomplete:
+                entries.append({"model_id": "org/unfinished"})
+            configs[column] = Dataset.from_dict(
+                {
+                    "id": list(range(len(ds))),
+                    "image": ds["image"],
+                    # Equal contents make the original column name the only change.
+                    "old_ocr": ds[column],
+                    "new_ocr": ds[column],
+                    "inference_info": [json.dumps(entries)] * len(ds),
+                }
+            )
+        return configs
+
+    @pytest.mark.parametrize("change", ["unchanged", "output_column", "incomplete_entry"])
+    def test_config_resolution_controls_resume_provenance(self, change, capsys):
+        ds, ocr = make_ds(n=2, models=("a", "b"))
+        flags = ["--no-adaptive", "--checkpoint-every", "0"]
+        _, published, _ = _run_judge(
+            flags, ds, ocr, stamp_existing_provenance=False,
+            config_datasets=self._configs(ds, ocr),
+        )
+        assert _published_metadata(published).source_output_columns == {
+            column: "old_ocr" for column in ocr
+        }
+        configs = self._configs(
+            ds, ocr,
+            output_column="new_ocr" if change == "output_column" else "old_ocr",
+            incomplete=change == "incomplete_entry",
+        )
+        judge = FakeJudge()
+        with pytest.raises(SystemExit) if change == "output_column" else nullcontext():
+            _run_judge(
+                flags, ds, ocr, existing=self._saved_results(published), judge=judge,
+                stamp_existing_provenance=False, checkpoint_resume=True,
+                config_datasets=configs,
+            )
+        assert judge.judged == 0
+        if change == "output_column":
+            assert "resume provenance mismatch" in capsys.readouterr().out
+
+    def test_new_config_reuses_unchanged_pair_with_real_provenance(self):
+        ds, ocr = make_ds(n=2, models=("a", "b"))
+        flags = ["--no-adaptive", "--checkpoint-every", "0"]
+        _, published, _ = _run_judge(
+            flags, ds, ocr, stamp_existing_provenance=False,
+            config_datasets=self._configs(ds, ocr),
+        )
+        ds, ocr = make_ds(n=2, models=("a", "b", "c"))
+        judge, _, _ = _run_judge(
+            flags, ds, ocr, existing=self._saved_results(published),
+            stamp_existing_provenance=False, checkpoint_resume=True,
+            config_datasets=self._configs(ds, ocr),
+        )
+        assert judge.judged == 4
+        assert all("model-c" in pair for pair in judge.pairs_seen)
+
+    @pytest.mark.parametrize("change", ["ocr_text", "judge", "split"])
+    def test_changed_inputs_reject_real_checkpoint_provenance(self, change, capsys):
+        ds, ocr = make_ds(n=2, models=("a", "b"))
+        flags = ["--no-adaptive", "--checkpoint-every", "0"]
+        _, published, _ = _run_judge(flags, ds, ocr, stamp_existing_provenance=False)
+        if change == "ocr_text":
+            ds._columns["col_a"][0] = "Changed OCR transcription for the same source page"
+        elif change == "judge":
+            flags += ["--model", "different-judge"]
+        else:
+            flags += ["--split", "validation"]
+        judge = FakeJudge()
+        with pytest.raises(SystemExit):
+            _run_judge(
+                flags, ds, ocr, existing=self._saved_results(published), judge=judge,
+                stamp_existing_provenance=False, checkpoint_resume=True,
+            )
+        assert judge.judged == 0
+        assert "resume provenance mismatch" in capsys.readouterr().out
+
+    def test_comparisons_only_checkpoint_resumes_from_row_provenance(self):
+        ds, ocr = make_ds(n=4, models=("a", "b"))
+        _, first_publish, _ = _run_judge(
+            ["--no-adaptive", "--checkpoint-every", "0"],
+            ds,
+            ocr,
+            stamp_existing_provenance=False,
+        )
+        rows = first_publish.call_args.args[1].comparison_log[:2]
+        existing = [
+            ComparisonResult(
+                sample_idx=row["sample_idx"],
+                model_a=row["model_a"],
+                model_b=row["model_b"],
+                winner=row["winner"],
+                col_a=row["col_a"],
+                col_b=row["col_b"],
+                provenance_hash=row["provenance_hash"],
+            )
+            for row in rows
+        ]
+
+        judge, _, _ = _run_judge(
+            ["--no-adaptive", "--checkpoint-every", "0"],
+            ds,
+            ocr,
+            existing=existing,
+            stamp_existing_provenance=False,
+            checkpoint_resume=True,
+        )
+        assert judge.judged == 2
+
+    def test_missing_provenance_requires_full_rejudge(self, capsys):
+        ds, ocr = make_ds(n=2, models=("a", "b"))
+        existing = [
+            ComparisonResult(
+                sample_idx=0,
+                model_a="model-a",
+                model_b="model-b",
+                winner="A",
+                col_a="col_a",
+                col_b="col_b",
+            )
+        ]
+        with pytest.raises(SystemExit) as exc:
+            _run_judge(
+                ["--no-adaptive", "--checkpoint-every", "0"],
+                ds,
+                ocr,
+                existing=existing,
+                stamp_existing_provenance=False,
+            )
+        assert exc.value.code == 1
+        assert "lack resume provenance" in capsys.readouterr().out
+
+    def test_mismatched_provenance_requires_full_rejudge(self, capsys):
+        ds, ocr = make_ds(n=2, models=("a", "b"))
+        existing = [
+            ComparisonResult(
+                sample_idx=0,
+                model_a="model-a",
+                model_b="model-b",
+                winner="A",
+                col_a="col_a",
+                col_b="col_b",
+                provenance_hash="different-run",
+            )
+        ]
+        with pytest.raises(SystemExit) as exc:
+            _run_judge(
+                ["--no-adaptive", "--checkpoint-every", "0"],
+                ds,
+                ocr,
+                existing=existing,
+                stamp_existing_provenance=False,
+            )
+        assert exc.value.code == 1
+        assert "resume provenance mismatch" in capsys.readouterr().out
+
+    def test_matching_checkpoint_hash_overrides_stale_completed_metadata(self):
+        ds, ocr = make_ds(n=2, models=("a", "b"))
+        existing = [
+            ComparisonResult(
+                sample_idx=0,
+                model_a="model-a",
+                model_b="model-b",
+                winner="A",
+                col_a="col_a",
+                col_b="col_b",
+            )
+        ]
+        stale_metadata = [
+            {
+                "criteria": "default",
+                "prompt_hash": prompt_hash(CRITERIA_PROFILES["default"]),
+                "judge_text_mode": "raw",
+                "max_ocr_text_len": 1000,
+                "judge_image_dim": 512,
+            }
+        ]
+
+        judge, m_publish, _ = _run_judge(
+            [
+                "--criteria",
+                "table-fidelity",
+                "--no-adaptive",
+                "--checkpoint-every",
+                "0",
+            ],
+            ds,
+            ocr,
+            existing=existing,
+            existing_meta_override=stale_metadata,
+        )
+
+        assert judge.judged == 1
+        m_publish.assert_called_once()
+
     def test_resume_discards_sentinel_comparison_and_rejudges_sample(self, capsys):
         # Results produced before issue #46 can contain a verdict where an OCR
         # error sentinel competed as transcription text. It must be removed from
@@ -635,6 +895,9 @@ def _run(self, argv_extra, *, existing, existing_meta, legacy_preprocessing=Fals
             "--checkpoint-every", "0", *argv_extra,
         ]
         args = build_parser().parse_args(argv)
+        for result in existing:
+            result.col_a = result.col_a or "col_a"
+            result.col_b = result.col_b or "col_b"
         if not legacy_preprocessing:
             existing_meta = [
                 {
@@ -678,16 +941,16 @@ def test_legacy_raw_results_block_normalized_incremental_run(self):
         assert judge.judged == 0
         m_publish.assert_not_called()
 
-    def test_legacy_raw_results_allow_explicit_raw_resume(self):
+    def test_legacy_raw_results_still_require_one_full_rejudge(self):
         judge, m_publish, code = self._run(
             ["--judge-text-mode", "raw"],
             existing=self._existing_one_pair(),
             existing_meta=[{"criteria": "default", "prompt_hash": self._DEFAULT_HASH}],
             legacy_preprocessing=True,
         )
-        assert code is None
-        assert judge.judged > 0
-        m_publish.assert_called_once()
+        assert code == 1
+        assert judge.judged == 0
+        m_publish.assert_not_called()
 
     def test_changed_text_cap_blocks_incremental_run(self):
         judge, m_publish, code = self._run(
@@ -709,25 +972,26 @@ def test_mismatch_exits_without_judging_or_publishing(self):
         assert judge.judged == 0  # exited before any judge call
         m_publish.assert_not_called()
 
-    def test_pre_44_none_rows_treated_as_default_so_default_run_proceeds(self):
-        # Genuinely pre-#44 metadata: no criteria/prompt_hash columns → default.
-        _, m_publish, code = self._run(
+    def test_pre_44_none_rows_require_one_full_rejudge(self):
+        # Matching legacy metadata is insufficient to prove source identity.
+        judge, m_publish, code = self._run(
             [],  # no --criteria → default
             existing=self._existing_one_pair(),
             existing_meta=[{"source_dataset": "user/ds"}],
         )
-        assert code is None
-        m_publish.assert_called_once()
+        assert code == 1
+        assert judge.judged == 0
+        m_publish.assert_not_called()
 
-    def test_matching_criteria_proceeds(self):
-        _, m_publish, code = self._run(
+    def test_matching_legacy_criteria_still_require_one_full_rejudge(self):
+        judge, m_publish, code = self._run(
             ["--criteria", "table-fidelity"],
             existing=self._existing_one_pair(),
             existing_meta=[{"criteria": "table-fidelity", "prompt_hash": self._TABLE_HASH}],
         )
-        assert code is None
-        m_publish.assert_called_once()
-        assert m_publish.call_args.args[2].criteria == "table-fidelity"
+        assert code == 1
+        assert judge.judged == 0
+        m_publish.assert_not_called()
 
     def test_full_rejudge_bypasses_guard(self):
         # Metadata says default, run requests table-fidelity — normally blocked,
@@ -741,20 +1005,20 @@ def test_full_rejudge_bypasses_guard(self):
         m_publish.assert_called_once()
         assert m_publish.call_args.args[2].criteria == "table-fidelity"
 
-    def test_same_custom_file_rerun_matches(self, tmp_path):
-        # A repo judged under a custom prompt file, re-run with the same file:
-        # identical hash → proceeds (guard compares hashes, not names).
+    def test_same_custom_file_still_requires_one_full_rejudge(self, tmp_path):
+        # The matching prompt gives a useful legacy diagnostic, but it cannot
+        # establish source identity without comparison-row provenance.
         f = tmp_path / "rubric.txt"
         f.write_text("Custom rubric. A={ocr_text_a} B={ocr_text_b}")
         file_hash = prompt_hash(f.read_text())
-        _, m_publish, code = self._run(
+        judge, m_publish, code = self._run(
             ["--criteria-file", str(f)],
             existing=self._existing_one_pair(),
             existing_meta=[{"criteria": "custom:rubric.txt", "prompt_hash": file_hash}],
         )
-        assert code is None
-        m_publish.assert_called_once()
-        assert m_publish.call_args.args[2].criteria == "custom:rubric.txt"
+        assert code == 1
+        assert judge.judged == 0
+        m_publish.assert_not_called()
 
     def test_different_custom_content_blocks(self, tmp_path, capsys):
         # Same basename, DIFFERENT content → different hash → blocked, and the
diff --git a/tests/test_dataset.py b/tests/test_dataset.py
index 5bfd04e..845f0b8 100644
--- a/tests/test_dataset.py
+++ b/tests/test_dataset.py
@@ -7,6 +7,7 @@
 
 import pytest
 from datasets import Dataset
+from PIL import Image
 
 from ocr_bench.dataset import (
     DatasetError,
@@ -16,12 +17,63 @@
     discover_configs,
     discover_ocr_columns,
     discover_pr_configs,
+    evaluation_input_fingerprints,
     find_alignment_mismatch,
     load_config_dataset,
     load_flat_dataset,
     shared_alignment_keys,
 )
 
+
+class TestEvaluationInputFingerprints:
+    def _dataset(self, *, first_text: str = "a0", first_pixel: str = "white"):
+        return Dataset.from_dict(
+            {
+                "id": ["page-0", "page-1"],
+                "image": [
+                    Image.new("RGB", (2, 2), first_pixel),
+                    Image.new("RGB", (2, 2), "white"),
+                ],
+                "ocr_a": [first_text, "a1"],
+                "ocr_b": ["b0", "b1"],
+            }
+        )
+
+    def test_fingerprints_pages_and_columns_independently(self):
+        original = self._dataset()
+        page_fp, column_fps = evaluation_input_fingerprints(
+            original, {"ocr_a": "A", "ocr_b": "B"}, [0, 1]
+        )
+
+        changed_text = self._dataset(first_text="changed")
+        changed_page_fp, changed_column_fps = evaluation_input_fingerprints(
+            changed_text, {"ocr_a": "A", "ocr_b": "B"}, [0, 1]
+        )
+        assert changed_page_fp == page_fp
+        assert changed_column_fps["ocr_a"] != column_fps["ocr_a"]
+        assert changed_column_fps["ocr_b"] == column_fps["ocr_b"]
+
+        changed_image = self._dataset(first_pixel="black")
+        image_page_fp, _ = evaluation_input_fingerprints(
+            changed_image, {"ocr_a": "A", "ocr_b": "B"}, [0, 1]
+        )
+        assert image_page_fp != page_fp
+
+    def test_adding_model_does_not_change_existing_fingerprints(self):
+        original = self._dataset()
+        page_fp, column_fps = evaluation_input_fingerprints(
+            original, {"ocr_a": "A", "ocr_b": "B"}, [0, 1]
+        )
+        extended = original.add_column("ocr_c", ["c0", "c1"])
+        extended_page_fp, extended_column_fps = evaluation_input_fingerprints(
+            extended,
+            {"ocr_a": "A", "ocr_b": "B", "ocr_c": "C"},
+            [0, 1],
+        )
+        assert extended_page_fp == page_fp
+        assert extended_column_fps["ocr_a"] == column_fps["ocr_a"]
+        assert extended_column_fps["ocr_b"] == column_fps["ocr_b"]
+
 # ---------------------------------------------------------------------------
 # discover_ocr_columns
 # ---------------------------------------------------------------------------
diff --git a/tests/test_publish.py b/tests/test_publish.py
index 8d38e38..e4f9524 100644
--- a/tests/test_publish.py
+++ b/tests/test_publish.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -13,6 +14,8 @@
     _align_metadata_rows,
     build_leaderboard_rows,
     build_metadata_row,
+    evaluation_provenance_hash,
+    hash_judge_specs,
     load_existing_comparisons,
     load_existing_metadata,
     publish_checkpoint,
@@ -135,6 +138,114 @@ def test_source_split_recorded(self):
         )
         assert build_metadata_row(meta)["source_split"] == "validation"
 
+    def test_source_fingerprints_and_selection_recorded(self):
+        meta = EvalMetadata(
+            source_dataset="repo/data",
+            source_split="validation",
+            source_fingerprint="pages123",
+            source_column_fingerprints={"a": "aaa", "b": "bbb"},
+            source_configs=["a", "b"],
+            source_columns={"a": "org/A", "b": "org/B"},
+            source_output_columns={"a": "ocr_v2", "b": "markdown"},
+            judge_models=["judge-a"],
+            judge_spec_hashes=hash_judge_specs(
+                ["https://judge.example/v1/:judge-a"]
+            ),
+            seed=42,
+            max_samples=10,
+            total_comparisons=30,
+            valid_comparisons=28,
+            min_chars=8,
+        )
+        row = build_metadata_row(meta)
+        assert row["source_fingerprint"] == "pages123"
+        assert json.loads(row["source_column_fingerprints"]) == {"a": "aaa", "b": "bbb"}
+        assert json.loads(row["source_configs"]) == ["a", "b"]
+        assert json.loads(row["source_columns"]) == {"a": "org/A", "b": "org/B"}
+        assert json.loads(row["source_output_columns"]) == {"a": "ocr_v2", "b": "markdown"}
+        assert json.loads(row["judge_spec_hashes"]) == hash_judge_specs(
+            ["https://judge.example/v1/:judge-a"]
+        )
+        assert row["min_chars"] == 8
+
+    def test_judge_spec_metadata_does_not_publish_endpoint_secrets(self):
+        secret_spec = "https://user:password@private.example/v1/?api_key=secret"
+        meta = EvalMetadata(
+            source_dataset="repo/data",
+            judge_models=["private-judge"],
+            judge_spec_hashes=hash_judge_specs([secret_spec]),
+            seed=42,
+            max_samples=10,
+            total_comparisons=30,
+            valid_comparisons=28,
+        )
+
+        serialized = json.dumps(build_metadata_row(meta))
+        assert secret_spec not in serialized
+        assert "password" not in serialized
+        assert "api_key" not in serialized
+        assert json.loads(build_metadata_row(meta)["judge_spec_hashes"]) == hash_judge_specs(
+            [secret_spec]
+        )
+
+
+class TestEvaluationProvenanceHash:
+    def _kwargs(self):
+        return {
+            "source_dataset": "repo/data",
+            "source_split": "train",
+            "source_fingerprint": "pages",
+            "source_column_fingerprints": {"a": "aaa", "b": "bbb"},
+            "source_columns": {"a": "A", "b": "B"},
+            "source_output_columns": {"a": "ocr_v2", "b": "markdown"},
+            "judge_specs": ["https://judge.example/v1/:judge"],
+            "seed": 42,
+            "max_samples": 10,
+            "prompt_hash": "prompt",
+            "judge_text_mode": "normalized",
+            "max_ocr_text_len": 2500,
+            "judge_image_dim": 1024,
+            "max_tokens": 4096,
+            "min_chars": 20,
+        }
+
+    def test_is_stable_across_mapping_order(self):
+        kwargs = self._kwargs()
+        first = evaluation_provenance_hash(**kwargs)
+        kwargs["source_columns"] = {"b": "B", "a": "A"}
+        assert evaluation_provenance_hash(**kwargs) == first
+
+    @pytest.mark.parametrize(
+        ("field", "changed"),
+        [
+            ("source_split", "validation"),
+            ("source_fingerprint", "different-pages"),
+            ("source_column_fingerprints", {"a": "changed", "b": "bbb"}),
+            ("source_columns", {"a": "different-model", "b": "B"}),
+            ("source_output_columns", {"a": "ocr_v3", "b": "markdown"}),
+            ("judge_specs", ["https://other.example/v1/:judge"]),
+            ("seed", 7),
+            ("prompt_hash", "other-prompt"),
+            ("min_chars", 0),
+            ("max_samples", 5),
+            ("judge_text_mode", "raw"),
+            ("max_ocr_text_len", 1000),
+            ("judge_image_dim", 512),
+            ("max_tokens", 2048),
+        ],
+    )
+    def test_changes_when_guarded_input_changes(self, field, changed):
+        kwargs = self._kwargs()
+        original = evaluation_provenance_hash(**kwargs)
+        kwargs[field] = changed
+        assert evaluation_provenance_hash(**kwargs) != original
+
+    def test_openai_compatible_endpoint_is_part_of_identity(self):
+        kwargs = self._kwargs()
+        first = evaluation_provenance_hash(**kwargs)
+        kwargs["judge_specs"] = ["https://another.example/v1/:judge"]
+        assert evaluation_provenance_hash(**kwargs) != first
+
     def test_preserved_timestamp(self):
         meta = EvalMetadata(
             source_dataset="repo/data",
@@ -432,7 +543,13 @@ class TestPublishCheckpoint:
     @patch("ocr_bench.publish.Dataset")
     def test_pushes_only_comparisons_config(self, mock_ds_cls):
         results = [
-            ComparisonResult(sample_idx=0, model_a="a", model_b="b", winner="A"),
+            ComparisonResult(
+                sample_idx=0,
+                model_a="a",
+                model_b="b",
+                winner="A",
+                provenance_hash="safe123",
+            ),
             ComparisonResult(sample_idx=1, model_a="a", model_b="b", winner="B"),
         ]
         publish_checkpoint("user/results", results, ["a", "b"])
@@ -441,6 +558,8 @@ def test_pushes_only_comparisons_config(self, mock_ds_cls):
         # Exactly one push — no leaderboard/metadata/README churn.
         mock_ds.push_to_hub.assert_called_once()
         assert mock_ds.push_to_hub.call_args.kwargs["config_name"] == "comparisons"
+        rows = mock_ds_cls.from_list.call_args.args[0]
+        assert rows[0]["provenance_hash"] == "safe123"
 
     @patch("ocr_bench.publish.Dataset")
     def test_no_push_when_no_results(self, mock_ds_cls):
@@ -681,6 +800,23 @@ def test_reads_truncation_flags(self, mock_load):
         assert results[0].truncated_a is True
         assert results[0].truncated_b is False
 
+    @patch("ocr_bench.publish.load_dataset")
+    def test_reads_comparison_provenance_hash(self, mock_load):
+        mock_ds = MagicMock()
+        mock_ds.__iter__ = lambda self: iter(
+            [
+                {
+                    "sample_idx": 0,
+                    "model_a": "ModelA",
+                    "model_b": "ModelB",
+                    "winner": "A",
+                    "provenance_hash": "safe123",
+                }
+            ]
+        )
+        mock_load.return_value = mock_ds
+        assert load_existing_comparisons("user/results")[0].provenance_hash == "safe123"
+
     @patch("ocr_bench.publish.HfApi")
     @patch("ocr_bench.publish.load_dataset")
     def test_returns_empty_when_comparisons_config_is_absent(self, mock_load, mock_api_cls):