fix(lead): keep every manifest row and refuse a repeated Databricks start

src/lead/gates.py

@@ -8,6 +8,7 @@ of either hash.
 from __future__ import annotations
 
 import hashlib
+import math
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 
@@ -49,9 +50,15 @@ def promotion_gates(item: GateInput) -> GateResult:
         reasons.append("baseline")
     if item.test_year != item.train_through + 1:
         reasons.append("out_of_time")
-    if item.calibration_error > item.calibration_tolerance:
+    if (
+        math.isnan(item.calibration_error)
+        or item.calibration_error > item.calibration_tolerance
+    ):
         reasons.append("calibration")
-    if item.conformal_coverage < item.coverage_target:
+    if (
+        math.isnan(item.conformal_coverage)
+        or item.conformal_coverage < item.coverage_target
+    ):
         reasons.append("coverage")
     if not item.subgroup_pass:
         reasons.append("subgroup")

src/lead/manifest.py

@@ -80,24 +80,31 @@ def _ensure(con) -> None:
             con.execute(text)
 
 
-def _identity(row: dict[str, str]) -> tuple[str, str, str]:
+def _fields(row: dict[str, str], line_no: int) -> tuple[str, str, str, str, str]:
     source_id = (row.get("source_id") or "").strip()
     release = (row.get("release") or "").strip()
     digest = (row.get("sha256") or "").strip()
+    path = (row.get("path") or "").strip()
+    url = (row.get("url") or "").strip()
     if not source_id:
-        raise ValueError("source_id is required")
+        raise ValueError(f"row {line_no}: source_id is required")
     if not release:
-        raise ValueError("release is required")
-    if not digest:
-        raise ValueError("sha256 is required")
-    return source_id, release, digest
+        raise ValueError(f"row {line_no}: release is required ({source_id})")
+    return source_id, release, digest, path, url
 
 
 def load_manifest(con, csv_path: str, *, actor: str) -> int:
     """Insert new release rows. Return how many were new.
 
+    A row is identified by source, release, sha256, path, and URL, so two
+    paths that share a file hash both stay. A file whose every row lacks a
+    sha256 raises and inserts nothing. A blank hash beside real rows is
+    stored with an empty sha256 and layer ``gap``. The file hash is not
+    invented.
+
     Raises:
-        ValueError: a required column or a required key is missing.
+        ValueError: a required column is missing, a row has no source or
+            release, or no row has a sha256.
     """
     _ensure(con)
     with Path(csv_path).open(newline="") as handle:
@@ -106,22 +113,29 @@ def load_manifest(con, csv_path: str, *, actor: str) -> int:
         if missing:
             raise ValueError(f"missing columns: {missing}")
         raw = list(reader)
-    identities = [_identity(row) for row in raw]
+    parsed = [(i, row, _fields(row, i)) for i, row in enumerate(raw, start=1)]
+    if parsed and all(not fields[2] for _line, _row, fields in parsed):
+        named = "; ".join(
+            f"row {line_no}: sha256 is required ({fields[0]} {fields[1]})"
+            for line_no, _row, fields in parsed
+        )
+        raise ValueError(named)
     inserted = 0
-    for row, (source_id, release, digest) in zip(raw, identities, strict=True):
+    for _line, row, (source_id, release, digest, path, url) in parsed:
         content = hashlib.sha256(
-            f"{source_id}\n{release}\n{digest}\n".encode()
+            f"{source_id}\n{release}\n{digest}\n{path}\n{url}\n".encode()
         ).hexdigest()
         existing = con.execute(
             """
             SELECT 1 FROM lead_evidence.public_release
             WHERE source_id = ? AND release = ? AND sha256 = ?
+              AND path = ? AND url = ?
             """,
-            [source_id, release, digest],
+            [source_id, release, digest, path, url],
         ).fetchone()
         if existing:
             continue
-        layer = layer_of(source_id, row.get("lead_use") or "")
+        layer = "gap" if not digest else layer_of(source_id, row.get("lead_use") or "")
         con.execute(
             """
             INSERT INTO lead_evidence.public_release (
@@ -136,13 +150,13 @@ def load_manifest(con, csv_path: str, *, actor: str) -> int:
                 row.get("title") or "",
                 row.get("publisher") or "",
                 row.get("kind") or "",
-                row.get("url") or "",
+                url,
                 row.get("api_endpoint") or "",
                 row.get("landing_page") or "",
                 release,
                 row.get("fetched_utc") or "",
                 row.get("status") or "",
-                row.get("path") or "",
+                path,
                 digest,
                 row.get("rows") or "",
                 row.get("size_bytes") or "",

src/lead/tools.py

@@ -167,15 +167,17 @@ def runs_start(
         by_id(target_id)
         if where_ran not in {"local", "databricks"}:
             raise ValueError(where_ran)
+        content = _hash(
+            target_id, model, where_ran, finished_on, metrics, feature_ids, gate_results
+        )
+        if _row(con, "run", content) is not None:
+            return content
         if where_ran == "databricks":
             if spent >= budget:
                 raise RuntimeError("databricks budget")
             if transport is None:
                 raise RuntimeError("databricks transport is required")
             transport({"target_id": target_id, "model": model})
-        content = _hash(
-            target_id, model, where_ran, finished_on, metrics, feature_ids, gate_results
-        )
         append(
             con,
             "run",

tests/lead/test_gates.py

@@ -57,6 +57,15 @@ def test_each_failed_gate_is_one_reason(overrides, reason):
     assert result.reasons == (reason,)
 
 
+def test_a_nan_metric_does_not_promote():
+    calibrated = promotion_gates(_passing(calibration_error=float("nan")))
+    covered = promotion_gates(_passing(conformal_coverage=float("nan")))
+    assert calibrated.promoted is False
+    assert calibrated.reasons == ("calibration",)
+    assert covered.promoted is False
+    assert covered.reasons == ("coverage",)
+
+
 def test_a_discarded_run_is_recorded_once(tmp_path):
     con = open_ledger(str(tmp_path / "ledger.duckdb"))
     gate = GateResult(promoted=False, reasons=("baseline",))

tests/lead/test_manifest.py

@@ -63,7 +63,7 @@ def test_a_blank_sha256_raises_and_inserts_nothing(tmp_path):
     )
     con = open_ledger(str(tmp_path / "ledger.duckdb"))
     try:
-        with pytest.raises(ValueError):
+        with pytest.raises(ValueError, match=r"row 1:.*ahrq.*2024"):
             load_manifest(con, str(csv_path), actor="tester")
         count = con.execute(
             "SELECT count(*) FROM lead_evidence.public_release"
@@ -73,6 +73,53 @@ def test_a_blank_sha256_raises_and_inserts_nothing(tmp_path):
         con.close()
 
 
+def test_a_blank_hash_among_real_rows_is_stored_as_a_gap(tmp_path):
+    csv_path = _csv(
+        tmp_path,
+        "ahrq,Title,Pub,table,https://example.test/a,,,2024,2026-10-31,ok,"
+        "/no/such/lead-file,abc,1,2,public,benchmark\n"
+        "cms_mds,Full file,CMS,data,https://example.test/full,,,2026-04-01,"
+        "2026-10-31,over_cap:10,,,10,public,benchmark\n",
+    )
+    con = open_ledger(str(tmp_path / "ledger.duckdb"))
+    try:
+        assert load_manifest(con, str(csv_path), actor="tester") == 2
+        assert load_manifest(con, str(csv_path), actor="tester") == 0
+        gap = con.execute(
+            """
+            SELECT sha256, layer, path
+            FROM lead_evidence.public_release
+            WHERE source_id = 'cms_mds'
+            """
+        ).fetchone()
+        assert gap == ("", "gap", "")
+    finally:
+        con.close()
+
+
+def test_the_same_bytes_at_two_paths_both_stay(tmp_path):
+    csv_path = _csv(
+        tmp_path,
+        "ahrq,Office A,Pub,table,https://example.test/a,,,2024,2026-10-31,ok,"
+        "ahrq/a.csv,abc,1,2,public,benchmark\n"
+        "ahrq,Office B,Pub,table,https://example.test/b,,,2024,2026-10-31,ok,"
+        "ahrq/b.csv,abc,1,2,public,benchmark\n",
+    )
+    con = open_ledger(str(tmp_path / "ledger.duckdb"))
+    try:
+        assert load_manifest(con, str(csv_path), actor="tester") == 2
+        assert load_manifest(con, str(csv_path), actor="tester") == 0
+        paths = con.execute(
+            """
+            SELECT path FROM lead_evidence.public_release
+            WHERE source_id = 'ahrq' ORDER BY path
+            """
+        ).fetchall()
+        assert paths == [("ahrq/a.csv",), ("ahrq/b.csv",)]
+    finally:
+        con.close()
+
+
 def test_release_ddl_and_delta_schema_omit_the_catalog_token():
     assert "aco" not in RELEASE_DDL
     assert "aco" not in schema_sql("delta")

tests/lead/test_tools.py

@@ -177,6 +177,28 @@ def test_databricks_refuses_before_calling_a_transport(tmp_path):
         con.close()
 
 
+def test_a_repeated_databricks_start_does_not_call_the_transport_again(tmp_path):
+    con = open_ledger(str(tmp_path / "ledger.duckdb"))
+    called: list[dict] = []
+    kwargs = {
+        "where_ran": "databricks",
+        "target_id": "Q.ACR",
+        "actor": "tester",
+        "model": "glm",
+        "finished_on": "2026-10-31",
+        "metrics": "mae=1",
+        "spent": 0,
+        "budget": 1,
+    }
+    try:
+        first = runs_start(con, transport=called.append, **kwargs)
+        second = runs_start(con, transport=called.append, **kwargs)
+        assert first == second
+        assert len(called) == 1
+    finally:
+        con.close()
+
+
 def test_an_injected_databricks_transport_records_the_run(tmp_path):
     con = open_ledger(str(tmp_path / "ledger.duckdb"))
     called: list[dict] = []