diff --git a/src/lead/gates.py b/src/lead/gates.py index ebc307f..0747173 100644 --- a/src/lead/gates.py +++ b/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") diff --git a/src/lead/manifest.py b/src/lead/manifest.py index d4bd03d..fb30149 100644 --- a/src/lead/manifest.py +++ b/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 "", diff --git a/src/lead/tools.py b/src/lead/tools.py index 05ecad3..6d8de62 100644 --- a/src/lead/tools.py +++ b/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", diff --git a/tests/lead/test_gates.py b/tests/lead/test_gates.py index 261a97c..ed91caf 100644 --- a/tests/lead/test_gates.py +++ b/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",)) diff --git a/tests/lead/test_manifest.py b/tests/lead/test_manifest.py index 900482d..e3fea0a 100644 --- a/tests/lead/test_manifest.py +++ b/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") diff --git a/tests/lead/test_tools.py b/tests/lead/test_tools.py index 2dd2e0b..f69aacf 100644 --- a/tests/lead/test_tools.py +++ b/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] = []