feat(lead): gate promotion and record discarded runs

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"],
+}

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)

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")

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