feat(lead): append-only evidence ledger in DuckDB
This commit is contained in:
kert
@@ -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()
|
||||
|
||||
12
src/lead/ledger/__init__.py
Normal file
12
src/lead/ledger/__init__.py
Normal file
@@ -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",
|
||||
]
|
||||
117
src/lead/ledger/schema.py
Normal file
117
src/lead/ledger/schema.py
Normal file
@@ -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) + ";"
|
||||
131
src/lead/ledger/store.py
Normal file
131
src/lead/ledger/store.py
Normal file
@@ -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
|
||||
221
tests/lead/test_ledger_store.py
Normal file
221
tests/lead/test_ledger_store.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user