diff --git a/src/cli/lead.py b/src/cli/lead.py index d3769c7..fb65e11 100644 --- a/src/cli/lead.py +++ b/src/cli/lead.py @@ -4,9 +4,13 @@ from __future__ import annotations import typer +from lead.ledger.store import open_ledger +from lead.ledger.store import status as ledger_status from lead.targets import TARGETS app = typer.Typer(no_args_is_help=True) +ledger_app = typer.Typer(no_args_is_help=True) +app.add_typer(ledger_app, name="ledger", help="Evidence ledger.") @app.command(name="targets") @@ -14,3 +18,16 @@ def targets() -> None: """Print every target ID, family, and name.""" for target in TARGETS: typer.echo(f"{target.id}\t{target.family}\t{target.name}") + + +@ledger_app.command(name="status") +def ledger_status_cmd( + db: str = typer.Option(..., help="Path to the DuckDB ledger."), +) -> None: + """Print one status row per target.""" + con = open_ledger(db) + try: + for row in ledger_status(con): + typer.echo(row.target_id) + finally: + con.close() diff --git a/src/lead/ledger/__init__.py b/src/lead/ledger/__init__.py new file mode 100644 index 0000000..d9ce127 --- /dev/null +++ b/src/lead/ledger/__init__.py @@ -0,0 +1,12 @@ +"""Append-only LEAD evidence ledger.""" + +from lead.ledger.schema import schema_sql +from lead.ledger.store import TargetStatus, append, open_ledger, status + +__all__ = [ + "TargetStatus", + "append", + "open_ledger", + "schema_sql", + "status", +] diff --git a/src/lead/ledger/schema.py b/src/lead/ledger/schema.py new file mode 100644 index 0000000..cb5f2bb --- /dev/null +++ b/src/lead/ledger/schema.py @@ -0,0 +1,117 @@ +"""DDL for the lead_evidence ledger. + +DuckDB applies this in :func:`lead.ledger.store.open_ledger`. The Delta +string is the same tables with no catalog, so the bundle target's catalog +wraps it. +""" + +from __future__ import annotations + +_COMMON = ("id", "created_at", "actor", "content_hash") + +COLUMNS: dict[str, tuple[str, ...]] = { + "source": _COMMON + + ( + "zotero_key", + "doi", + "pmid", + "url", + "database", + "file_sha256", + "strands", + "project", + ), + "search": _COMMON + + ( + "project", + "database", + "interface", + "searched_on", + "query", + "filters", + "hit_count", + "export_sha256", + ), + "decision": _COMMON + + ( + "project", + "zotero_key", + "stage", + "decision", + "reason", + "model", + "prompt_hash", + "criteria_hash", + "human_audited", + ), + "extraction": _COMMON + + ("project", "zotero_key", "extraction_version", "body", "target_id"), + "claim": _COMMON + + ( + "project", + "kind", + "subject", + "target_id", + "direction", + "effect_measure", + "effect_value", + "effect_ci_low", + "effect_ci_high", + "lead_time", + "population", + "setting", + "grain", + "design", + "risk_of_bias", + "certainty", + "extraction_ids", + "tiers", + ), + "feature": _COMMON + + ( + "target_id", + "claim_ids", + "expression", + "grain", + "code_lists", + "data_source_ids", + ), + "run": _COMMON + + ( + "target_id", + "mlflow_run_id", + "where_ran", + "finished_on", + "snapshot_hashes", + "feature_ids", + "model", + "metrics", + "gate_results", + ), + "finding": _COMMON + ("claim_id", "relation", "grain", "numbers"), + "exogenous": _COMMON + + ("project", "target_id", "series_or_event", "event_on", "citation"), +} + +_TYPES = { + "hit_count": "BIGINT", + "effect_value": "DOUBLE", + "effect_ci_low": "DOUBLE", + "effect_ci_high": "DOUBLE", + "human_audited": "BOOLEAN", +} + + +def schema_sql(dialect: str) -> str: + """Return DDL for ``duckdb`` or ``delta``. + + Raises: + ValueError: ``dialect`` is neither of those. + """ + if dialect not in {"duckdb", "delta"}: + raise ValueError(dialect) + statements = ["CREATE SCHEMA IF NOT EXISTS lead_evidence"] + for table, columns in COLUMNS.items(): + body = ", ".join(f"{name} {_TYPES.get(name, 'VARCHAR')}" for name in columns) + statements.append(f"CREATE TABLE IF NOT EXISTS lead_evidence.{table} ({body})") + return ";\n".join(statements) + ";" diff --git a/src/lead/ledger/store.py b/src/lead/ledger/store.py new file mode 100644 index 0000000..395e35a --- /dev/null +++ b/src/lead/ledger/store.py @@ -0,0 +1,131 @@ +"""Append-only writes and per-target status for the evidence ledger.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import datetime, timezone + +import duckdb + +from lead.ledger.schema import COLUMNS, schema_sql +from lead.targets import TARGETS, by_id + +TIERS = ("rule", "aggregate", "literature", "synthetic") + + +@dataclass(frozen=True) +class TargetStatus: + """How much the ledger knows about one target.""" + + target_id: str + claims_by_tier: dict[str, int] + gaps: int + last_run: str | None + open_contradictions: int + + +def open_ledger(path: str) -> duckdb.DuckDBPyConnection: + """Open a DuckDB file and apply the ledger schema.""" + con = duckdb.connect(path) + for statement in schema_sql("duckdb").split(";"): + text = statement.strip() + if text: + con.execute(text) + return con + + +def append( + con: duckdb.DuckDBPyConnection, table: str, row: Mapping[str, object] +) -> bool: + """Insert ``row`` into ``table``. + + The caller supplies ``content_hash``. A hash already stored in that + table inserts nothing and returns ``False``. ``created_at`` is set + when the row omits it. A claim's ``id`` defaults to its content hash + so a finding can point at it. A claim whose ``target_id`` is unknown + raises ``KeyError`` and writes nothing. + """ + if table not in COLUMNS: + raise ValueError(table) + data = dict(row) + if table == "claim": + by_id(str(data["target_id"])) + content_hash = data.get("content_hash") + if not isinstance(content_hash, str) or not content_hash: + raise ValueError("content_hash is required") + unknown = set(data) - set(COLUMNS[table]) + if unknown: + raise ValueError(f"unknown columns: {sorted(unknown)}") + existing = con.execute( + f"SELECT 1 FROM lead_evidence.{table} WHERE content_hash = ?", + [content_hash], + ).fetchone() + if existing: + return False + if not data.get("created_at"): + data["created_at"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + if not data.get("id"): + data["id"] = content_hash + columns = [ + name for name in COLUMNS[table] if name in data and data[name] is not None + ] + placeholders = ", ".join("?" for _ in columns) + names = ", ".join(columns) + con.execute( + f"INSERT INTO lead_evidence.{table} ({names}) VALUES ({placeholders})", + [data[name] for name in columns], + ) + return True + + +def status(con: duckdb.DuckDBPyConnection) -> list[TargetStatus]: + """One row per registry target, including targets with no claims.""" + claims = con.execute( + "SELECT id, target_id, tiers FROM lead_evidence.claim" + ).fetchall() + tier_counts: dict[str, dict[str, int]] = {} + claimed: set[str] = set() + for _claim_id, target_id, tiers in claims: + claimed.add(target_id) + bucket = tier_counts.setdefault(target_id, {}) + for name in str(tiers or "").split(","): + tier = name.strip() + if tier in TIERS: + bucket[tier] = bucket.get(tier, 0) + 1 + contradictions = { + target_id: int(count) + for target_id, count in con.execute( + """ + SELECT c.target_id, count(*) + FROM lead_evidence.finding AS f + JOIN lead_evidence.claim AS c ON c.id = f.claim_id + WHERE f.relation = 'contradicts' + GROUP BY c.target_id + """ + ).fetchall() + } + last_runs = { + target_id: finished_on + for target_id, finished_on in con.execute( + """ + SELECT target_id, max(finished_on) + FROM lead_evidence.run + WHERE target_id IS NOT NULL AND finished_on IS NOT NULL + GROUP BY target_id + """ + ).fetchall() + } + rows: list[TargetStatus] = [] + for target in TARGETS: + counts = {tier: tier_counts.get(target.id, {}).get(tier, 0) for tier in TIERS} + rows.append( + TargetStatus( + target_id=target.id, + claims_by_tier=counts, + gaps=0 if target.id in claimed else 1, + last_run=last_runs.get(target.id), + open_contradictions=contradictions.get(target.id, 0), + ) + ) + return rows diff --git a/tests/lead/test_ledger_store.py b/tests/lead/test_ledger_store.py new file mode 100644 index 0000000..3878333 --- /dev/null +++ b/tests/lead/test_ledger_store.py @@ -0,0 +1,221 @@ +"""Append-only evidence ledger: schema, idempotent insert, per-target status.""" + +from __future__ import annotations + +import pytest +from typer.testing import CliRunner + +from cli import app +from lead.ledger.schema import schema_sql +from lead.ledger.store import append, open_ledger, status +from lead.targets import TARGETS + +runner = CliRunner() + +TIERS = ("rule", "aggregate", "literature", "synthetic") + + +def claim_row(**overrides: object) -> dict[str, object]: + row: dict[str, object] = { + "kind": "precursor", + "subject": "readmission", + "target_id": "Q.ACR", + "tiers": "literature", + "content_hash": "abc", + "actor": "tester", + } + row.update(overrides) + return row + + +def test_append_is_idempotent_and_status_covers_every_target(tmp_path): + con = open_ledger(str(tmp_path / "ledger.duckdb")) + try: + row = claim_row(target_id="Q.ACR", tiers="literature", content_hash="abc") + assert append(con, "claim", row) is True + assert append(con, "claim", row) is False + assert ( + con.execute("select count(*) from lead_evidence.claim").fetchone()[0] == 1 + ) + created_at = con.execute( + "select created_at from lead_evidence.claim" + ).fetchone()[0] + assert created_at + found = {s.target_id: s for s in status(con)} + assert set(found) == {t.id for t in TARGETS} + assert found["Q.ACR"].claims_by_tier["literature"] == 1 + assert found["Q.ACR"].gaps == 0 + assert found["HN.1"].gaps == 1 + assert found["HN.1"].claims_by_tier == dict.fromkeys(TIERS, 0) + assert found["HN.1"].open_contradictions == 0 + assert found["HN.1"].last_run is None + finally: + con.close() + + +def test_unknown_claim_target_raises(tmp_path): + con = open_ledger(str(tmp_path / "ledger.duckdb")) + try: + with pytest.raises(KeyError): + append(con, "claim", claim_row(target_id="nope", content_hash="missing")) + finally: + con.close() + + +def test_contradicting_finding_bumps_open_contradictions(tmp_path): + con = open_ledger(str(tmp_path / "ledger.duckdb")) + try: + assert append(con, "claim", claim_row(content_hash="abc", id="abc")) + assert append( + con, + "finding", + { + "claim_id": "abc", + "relation": "confirms", + "grain": "beneficiary", + "content_hash": "f-ok", + "actor": "tester", + }, + ) + found = {s.target_id: s for s in status(con)} + assert found["Q.ACR"].open_contradictions == 0 + assert append( + con, + "finding", + { + "claim_id": "abc", + "relation": "contradicts", + "grain": "beneficiary", + "content_hash": "f-no", + "actor": "tester", + }, + ) + found = {s.target_id: s for s in status(con)} + assert found["Q.ACR"].open_contradictions == 1 + assert found["HN.1"].open_contradictions == 0 + finally: + con.close() + + +def test_last_run_is_the_latest_finished_on_for_that_target(tmp_path): + con = open_ledger(str(tmp_path / "ledger.duckdb")) + try: + assert append( + con, + "run", + { + "target_id": "Q.ACR", + "where_ran": "local", + "finished_on": "2026-10-30", + "content_hash": "run-old", + "actor": "tester", + }, + ) + assert append( + con, + "run", + { + "target_id": "Q.ACR", + "where_ran": "databricks", + "finished_on": "2026-10-31", + "content_hash": "run-new", + "actor": "tester", + }, + ) + found = {s.target_id: s for s in status(con)} + assert found["Q.ACR"].last_run == "2026-10-31" + assert found["HN.1"].last_run is None + finally: + con.close() + + +def test_a_tier_list_counts_each_named_tier(tmp_path): + con = open_ledger(str(tmp_path / "ledger.duckdb")) + try: + assert append( + con, "claim", claim_row(content_hash="both", tiers="rule, synthetic") + ) + found = {s.target_id: s for s in status(con)} + assert found["Q.ACR"].claims_by_tier["rule"] == 1 + assert found["Q.ACR"].claims_by_tier["synthetic"] == 1 + assert found["Q.ACR"].claims_by_tier["literature"] == 0 + finally: + con.close() + + +def test_delta_sql_names_the_schema_and_not_a_catalog(): + sql = schema_sql("delta") + assert "lead_evidence" in sql + assert "aco" not in sql + with pytest.raises(ValueError): + schema_sql("nope") + + +def test_search_exogenous_and_extraction_columns_roundtrip(tmp_path): + con = open_ledger(str(tmp_path / "ledger.duckdb")) + try: + assert append( + con, + "search", + { + "project": "lead-precursors", + "database": "pubmed", + "interface": "eutils", + "searched_on": "2026-10-31", + "query": "frailty", + "hit_count": 2, + "export_sha256": "deadbeef", + "content_hash": "s1", + "actor": "tester", + }, + ) + assert append( + con, + "exogenous", + { + "project": "lead-external", + "target_id": "Q.ACR", + "content_hash": "e1", + "actor": "tester", + }, + ) + assert append( + con, + "extraction", + { + "project": "lead-precursors", + "zotero_key": "ABCD1234", + "extraction_version": "v1", + "body": "target_id: Q.ACR", + "content_hash": "x1", + "actor": "tester", + }, + ) + assert append( + con, + "source", + { + "zotero_key": "ABCD1234", + "doi": "10.1/x", + "database": "pubmed", + "content_hash": "src1", + "actor": "tester", + }, + ) + searched = con.execute( + "select project, searched_on from lead_evidence.search" + ).fetchone() + assert searched == ("lead-precursors", "2026-10-31") + exogenous = con.execute( + "select target_id from lead_evidence.exogenous" + ).fetchone() + assert exogenous == ("Q.ACR",) + finally: + con.close() + + +def test_status_command_prints_every_target(tmp_path): + path = tmp_path / "ledger.duckdb" + result = runner.invoke(app, ["lead", "ledger", "status", "--db", str(path)]) + assert result.exit_code == 0 + assert "Q.ACR" in result.stdout