From 97816d60afc065ea5dd13dd7b8363759da359cc7 Mon Sep 17 00:00:00 2001 From: kert Date: Sat, 31 Oct 2026 22:06:25 -0400 Subject: [PATCH] feat(lead): gate promotion and record discarded runs --- src/aco/lake/lineage.py | 16 ++++ src/lead/gates.py | 140 ++++++++++++++++++++++++++++ tests/lead/test_gates.py | 148 ++++++++++++++++++++++++++++++ tests/lead/test_ledger_lineage.py | 16 ++++ 4 files changed, 320 insertions(+) create mode 100644 src/lead/gates.py create mode 100644 tests/lead/test_gates.py create mode 100644 tests/lead/test_ledger_lineage.py diff --git a/src/aco/lake/lineage.py b/src/aco/lake/lineage.py index e10eb9a..33f5631 100644 --- a/src/aco/lake/lineage.py +++ b/src/aco/lake/lineage.py @@ -33,6 +33,12 @@ class LineageGraph: """Return direct input tables for a given output table.""" return self.table_edges.get(table_ref, []) + def table_outputs(self, table_ref: str) -> list[str]: + """Return tables that list ``table_ref`` as an input.""" + return sorted( + output for output, inputs in self.table_edges.items() if table_ref in inputs + ) + def all_tables(self) -> list[str]: """Return all tables in the lineage graph.""" tables = set(self.table_edges.keys()) @@ -98,4 +104,14 @@ def build_lineage() -> LineageGraph: if col_sources: graph.column_sources[output_name] = col_sources + for output, inputs in _LEDGER_EDGES.items(): + graph.table_edges[output] = list(inputs) return graph + + +_LEDGER_EDGES = { + "lead_evidence.claim": ["lead_evidence.extraction", "lead_evidence.source"], + "lead_evidence.feature": ["lead_evidence.claim"], + "lead_evidence.run": ["lead_evidence.feature", "lead_evidence.claim"], + "lead_evidence.finding": ["lead_evidence.claim", "lead_evidence.run"], +} diff --git a/src/lead/gates.py b/src/lead/gates.py new file mode 100644 index 0000000..ebc307f --- /dev/null +++ b/src/lead/gates.py @@ -0,0 +1,140 @@ +"""Promotion gates and the record of runs that did not pass them. + +A candidate is promoted only when every gate holds. A failed or discarded +run still appends one run row and one finding row. The actor is not part +of either hash. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Mapping, Sequence +from dataclasses import dataclass + +from lead.ledger.store import append +from lead.targets import by_id + +_RELATIONS = frozenset({"failed", "discarded", "promoted"}) + + +@dataclass(frozen=True) +class GateInput: + """What one candidate showed, out of time, against named claims.""" + + beats_automl: bool + beats_naive: bool + train_through: int + test_year: int + calibration_error: float + calibration_tolerance: float + conformal_coverage: float + coverage_target: float + subgroup_pass: bool + feature_claim_ids: tuple[str, ...] + known_claim_ids: frozenset[str] + + +@dataclass(frozen=True) +class GateResult: + """Empty reasons means the candidate is promoted.""" + + promoted: bool + reasons: tuple[str, ...] + + +def promotion_gates(item: GateInput) -> GateResult: + """Check the Part 3 gates. Reasons stay in a fixed order.""" + reasons: list[str] = [] + if not item.beats_automl or not item.beats_naive: + reasons.append("baseline") + if item.test_year != item.train_through + 1: + reasons.append("out_of_time") + if item.calibration_error > item.calibration_tolerance: + reasons.append("calibration") + if item.conformal_coverage < item.coverage_target: + reasons.append("coverage") + if not item.subgroup_pass: + reasons.append("subgroup") + if not item.feature_claim_ids or any( + claim_id not in item.known_claim_ids for claim_id in item.feature_claim_ids + ): + reasons.append("claim") + return GateResult(promoted=not reasons, reasons=tuple(reasons)) + + +def _hash(*parts: str) -> str: + return hashlib.sha256("\n".join(parts).encode()).hexdigest() + + +def record_outcome( + con, + *, + actor: str, + target_id: str, + model: str, + where_ran: str, + finished_on: str, + metrics: str, + gate: GateResult, + relation: str, + claim_id: str = "", +) -> tuple[bool, bool]: + """Append a run and a finding. Return whether each insert was new. + + Raises: + ValueError: ``relation`` is not failed, discarded, or promoted. + KeyError: ``target_id`` is not in the registry. + """ + if relation not in _RELATIONS: + raise ValueError(relation) + by_id(target_id) + reasons = ",".join(gate.reasons) + run_hash = _hash( + target_id, model, where_ran, finished_on, metrics, reasons, relation + ) + finding_hash = _hash(run_hash, relation, claim_id) + run_wrote = append( + con, + "run", + { + "id": run_hash, + "target_id": target_id, + "where_ran": where_ran, + "finished_on": finished_on, + "model": model, + "metrics": metrics, + "gate_results": reasons, + "content_hash": run_hash, + "actor": actor, + }, + ) + finding: dict[str, object] = { + "id": finding_hash, + "relation": relation, + "numbers": reasons, + "content_hash": finding_hash, + "actor": actor, + } + if claim_id: + finding["claim_id"] = claim_id + finding_wrote = append(con, "finding", finding) + return run_wrote, finding_wrote + + +def events_in_window( + *, + window_start: str, + alarm_on: str, + events: Sequence[Mapping[str, str]], +) -> tuple[str, ...]: + """Citations whose day falls inside the window, including both ends.""" + start = window_start[:10] + end = alarm_on[:10] + found: list[str] = [] + for event in events: + day = str(event.get("event_on") or "")[:10] + if start <= day <= end: + found.append( + str(event.get("citation") or event.get("series_or_event") or "") + ) + return tuple(found) diff --git a/tests/lead/test_gates.py b/tests/lead/test_gates.py new file mode 100644 index 0000000..261a97c --- /dev/null +++ b/tests/lead/test_gates.py @@ -0,0 +1,148 @@ +"""Promotion gates, the failure record, and drift windows.""" + +from __future__ import annotations + +import pytest + +from lead.gates import ( + GateInput, + GateResult, + events_in_window, + promotion_gates, + record_outcome, +) +from lead.ledger.store import open_ledger + + +def _passing(**overrides: object) -> GateInput: + base: dict[str, object] = { + "beats_automl": True, + "beats_naive": True, + "train_through": 2026, + "test_year": 2027, + "calibration_error": 0.01, + "calibration_tolerance": 0.05, + "conformal_coverage": 0.91, + "coverage_target": 0.90, + "subgroup_pass": True, + "feature_claim_ids": ("claim-1",), + "known_claim_ids": frozenset({"claim-1"}), + } + base.update(overrides) + return GateInput(**base) # type: ignore[arg-type] + + +def test_a_candidate_that_clears_every_gate_is_promoted(): + result = promotion_gates(_passing()) + assert result == GateResult(promoted=True, reasons=()) + + +@pytest.mark.parametrize( + ("overrides", "reason"), + [ + ({"beats_automl": False}, "baseline"), + ({"beats_naive": False}, "baseline"), + ({"test_year": 2026}, "out_of_time"), + ({"test_year": 2028}, "out_of_time"), + ({"calibration_error": 0.06}, "calibration"), + ({"conformal_coverage": 0.89}, "coverage"), + ({"subgroup_pass": False}, "subgroup"), + ({"feature_claim_ids": ()}, "claim"), + ({"feature_claim_ids": ("missing",)}, "claim"), + ], +) +def test_each_failed_gate_is_one_reason(overrides, reason): + result = promotion_gates(_passing(**overrides)) + assert result.promoted is False + assert result.reasons == (reason,) + + +def test_a_discarded_run_is_recorded_once(tmp_path): + con = open_ledger(str(tmp_path / "ledger.duckdb")) + gate = GateResult(promoted=False, reasons=("baseline",)) + try: + assert record_outcome( + con, + actor="tester", + target_id="Q.ACR", + model="glm", + where_ran="local", + finished_on="2026-10-31", + metrics="mae=1", + gate=gate, + relation="discarded", + ) == (True, True) + assert record_outcome( + con, + actor="other", + target_id="Q.ACR", + model="glm", + where_ran="local", + finished_on="2026-10-31", + metrics="mae=1", + gate=gate, + relation="discarded", + ) == (False, False) + runs = con.execute("SELECT count(*) FROM lead_evidence.run").fetchone()[0] + findings = con.execute("SELECT count(*) FROM lead_evidence.finding").fetchone()[ + 0 + ] + assert runs == 1 + assert findings == 1 + finally: + con.close() + + +def test_a_promoted_run_is_recorded(tmp_path): + con = open_ledger(str(tmp_path / "ledger.duckdb")) + try: + wrote = record_outcome( + con, + actor="tester", + target_id="Q.ACR", + model="glm", + where_ran="local", + finished_on="2026-10-31", + metrics="mae=1", + gate=promotion_gates(_passing()), + relation="promoted", + claim_id="claim-1", + ) + assert wrote == (True, True) + finally: + con.close() + + +def test_an_unknown_relation_writes_nothing(tmp_path): + con = open_ledger(str(tmp_path / "ledger.duckdb")) + try: + with pytest.raises(ValueError): + record_outcome( + con, + actor="tester", + target_id="Q.ACR", + model="glm", + where_ran="local", + finished_on="2026-10-31", + metrics="mae=1", + gate=GateResult(promoted=False, reasons=("baseline",)), + relation="other", + ) + assert con.execute("SELECT count(*) FROM lead_evidence.run").fetchone()[0] == 0 + assert ( + con.execute("SELECT count(*) FROM lead_evidence.finding").fetchone()[0] == 0 + ) + finally: + con.close() + + +def test_drift_window_includes_both_ends(): + events = ( + {"event_on": "2026-10-01", "citation": "start"}, + {"event_on": "2026-10-31", "citation": "alarm"}, + {"event_on": "2026-09-30", "citation": "before"}, + {"event_on": "2026-11-01", "citation": "after"}, + ) + assert events_in_window( + window_start="2026-10-01", alarm_on="2026-10-31", events=events + ) == ("start", "alarm") diff --git a/tests/lead/test_ledger_lineage.py b/tests/lead/test_ledger_lineage.py new file mode 100644 index 0000000..8abdb90 --- /dev/null +++ b/tests/lead/test_ledger_lineage.py @@ -0,0 +1,16 @@ +"""Ledger tables walk backward to sources and forward to runs.""" + +from __future__ import annotations + +from aco.lake.lineage import build_lineage + + +def test_a_claim_lists_its_inputs_and_its_outputs(): + graph = build_lineage() + deps = graph.table_deps("lead_evidence.claim") + assert "lead_evidence.extraction" in deps + assert "lead_evidence.source" in deps + outputs = graph.table_outputs("lead_evidence.claim") + assert "lead_evidence.feature" in outputs + assert "lead_evidence.run" in outputs + assert "lead_evidence.finding" in outputs