submit: reject byte-identical re-submits via SHA-256 dedup

Before: clicking Submit twice (or re-uploading the same zip after
a transient confusion) created two rows pointing at byte-identical
content, ran the eval twice for the same input, and bloated the
submissions/reports blobs.

After: the cheap-sync handler computes sha256 of the uploaded zip
right after validation passes, queries results.jsonl for any row
with the same submission_sha256, and rejects with a pointer to
the existing submission_id if a match is found. Hash also lands
on the new row's submission_sha256 field (schema bump in the
companion commit on the submissions dataset).

Dedup-fetch failures fall back to letting the submit through
rather than rejecting on a Hub blip (logged). Race window: two
near-simultaneous identical submits whose dedup checks both
return None before either appends will land two rows; recoverable
by manual cleanup if it ever happens, not worth tightening before
auth + login throttling lands at pre-launch.

Drive-by: _build_pending_row's unused fixture_names arg (already
noqa'd as "kept for future") gets dropped. Per code-review pass.

submit.py

@@ -57,6 +57,7 @@ Background worker, per submission:
 """
 from __future__ import annotations
 
+import hashlib
 import json
 import logging
 import os
@@ -98,6 +99,7 @@ FAILURE_REASON_MAX_CHARS = 200
 EVAL_TIMEOUT_SECONDS = 15 * 60
 REPORT_TIMEOUT_SECONDS = 2 * 60
 EVAL_WORKER_COUNT = "8"
+SHA256_BLOCK_SIZE = 64 * 1024
 
 # One HfApi client per process. HF_TOKEN is picked up from the env at
 # construction time and reused for every call.
@@ -154,13 +156,25 @@ def handle_submit(zip_file) -> str:
         except _ValidationError as e:
             return f"**Submission rejected.** {e}"
 
+        # Dedup gate: hash the raw zip bytes and reject if an existing
+        # row carries the same hash. Runs after validation so a clearly
+        # malformed upload still gets the specific validation error.
+        zip_sha256 = _compute_sha256(zip_path)
+        existing_id = _find_existing_submission_by_sha256(zip_sha256)
+        if existing_id is not None:
+            return (
+                f"**Submission rejected.** This zip's contents are identical "
+                f"to an existing submission (`{existing_id}`). Resubmit only "
+                f"after changing at least one byte of the upload."
+            )
+
         submission_id = _mint_submission_id(
             meta["submitter_name"], meta["submission_name"]
         )
         try:
             blob_url = _upload_submission_zip(submission_id, zip_path)
             row = _build_pending_row(
-                submission_id, meta, fixture_names, blob_url
+                submission_id, meta, blob_url, zip_sha256
             )
             _append_pending_row(row)
         except _HubWriteError as e:
@@ -370,14 +384,15 @@ def _upload_submission_zip(submission_id: str, zip_path: Path) -> str:
 def _build_pending_row(
     submission_id: str,
     meta: dict[str, Any],
-    fixture_names: set[str],  # noqa: ARG001 - kept for future per-fixture pre-fill
     blob_url: str,
+    submission_sha256: str,
 ) -> dict[str, Any]:
     """Construct the JSON row written for a freshly-queued submission.
 
     Mirrors the pending regime in ``cadgenbench-submissions/schema.md``:
-    metadata + ``status: pending`` + ``submission_blob_url``; every
-    score-shaped field is ``null`` until the worker flips the row.
+    metadata + ``status: pending`` + ``submission_blob_url`` +
+    ``submission_sha256``; every score-shaped field is ``null`` until
+    the worker flips the row.
     """
     return {
         "submission_id": submission_id,
@@ -397,9 +412,47 @@ def _build_pending_row(
         "per_fixture_scores": None,
         "per_fixture_breakdown": None,
         "submission_blob_url": blob_url,
+        "submission_sha256": submission_sha256,
     }
 
 
+def _compute_sha256(path: Path) -> str:
+    """Hex-encoded SHA-256 of the file at *path*. Streams in 64 KiB chunks."""
+    h = hashlib.sha256()
+    with path.open("rb") as f:
+        for chunk in iter(lambda: f.read(SHA256_BLOCK_SIZE), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def _find_existing_submission_by_sha256(zip_sha256: str) -> str | None:
+    """Return the ``submission_id`` of an existing row with the same hash, else None.
+
+    Reads the current ``results.jsonl`` once (no lock; a worst-case
+    race lands a duplicate row, which is recoverable by a cleanup pass
+    if it ever happens). Hub-fetch failures are non-fatal: the caller
+    just doesn't get the dedup gate this submit (logged).
+    """
+    try:
+        body = _download_results_jsonl()
+    except Exception as e:  # noqa: BLE001 - Hub API surface is broad
+        logger.warning(
+            "Dedup check skipped, Hub fetch failed (%s: %s)",
+            type(e).__name__, e,
+        )
+        return None
+    for line in body.splitlines():
+        if not line.strip():
+            continue
+        try:
+            row = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if row.get("submission_sha256") == zip_sha256:
+            return row.get("submission_id")
+    return None
+
+
 def _append_pending_row(row: dict[str, Any]) -> None:
     """Append a pending row to ``results.jsonl`` on the Hub under the lock."""
     submission_id = row["submission_id"]