feat(prisma): nightly and triggered search due check

src/cli/prisma.py

@@ -437,6 +437,25 @@ def search_protocols() -> None:
             typer.echo(f"{name}\t{query}")
 
 
+@search_app.command("due")
+def search_due(
+    project: str = typer.Option(..., "--project", help="Zotero project name."),
+    db: str = typer.Option(..., "--db", help="Path to the DuckDB ledger."),
+    now: str = typer.Option(..., "--now", help="Today as YYYY-MM-DD."),
+) -> None:
+    """Print why a search is due. Prints nothing when it is current."""
+    from lead.ledger.store import open_ledger
+    from prisma.search.due import due
+
+    con = open_ledger(db)
+    try:
+        reasons = due(con, project=project, now=now)
+    finally:
+        con.close()
+    for reason in reasons:
+        typer.echo(reason)
+
+
 @app.command("calibrate")
 def calibrate_cmd() -> None:
     """Calibration gate: sensitivity ≥ 0.95 on 200 title/abstract and 50 full text."""

src/prisma/search/due.py

@@ -0,0 +1,93 @@
+"""Whether a project's searches are due. This does not run a search.
+
+Dates are ``YYYY-MM-DD`` strings, compared as text. Nightly means the
+latest ``searched_on`` for the project is missing or earlier than today.
+The other reasons are a contradiction of moderate or high certainty, a
+strand-5 event on a target this project has a claim about, and a CMS
+source dated after that latest search.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    import duckdb
+
+
+def _latest(con: duckdb.DuckDBPyConnection, project: str) -> str | None:
+    row = con.execute(
+        """
+        SELECT max(searched_on)
+        FROM lead_evidence.search
+        WHERE project = ?
+        """,
+        [project],
+    ).fetchone()
+    if row is None or row[0] is None:
+        return None
+    return str(row[0])
+
+
+def _contradiction(con: duckdb.DuckDBPyConnection, project: str) -> bool:
+    row = con.execute(
+        """
+        SELECT 1
+        FROM lead_evidence.finding AS f
+        JOIN lead_evidence.claim AS c ON c.id = f.claim_id
+        WHERE c.project = ?
+          AND f.relation = 'contradicts'
+          AND c.certainty IN ('moderate', 'high')
+        LIMIT 1
+        """,
+        [project],
+    ).fetchone()
+    return row is not None
+
+
+def _exogenous(con: duckdb.DuckDBPyConnection, project: str) -> bool:
+    row = con.execute(
+        """
+        SELECT 1
+        FROM lead_evidence.exogenous AS e
+        WHERE EXISTS (
+            SELECT 1
+            FROM lead_evidence.claim AS c
+            WHERE c.project = ?
+              AND c.target_id = e.target_id
+        )
+        LIMIT 1
+        """,
+        [project],
+    ).fetchone()
+    return row is not None
+
+
+def _cms(con: duckdb.DuckDBPyConnection, project: str, latest: str) -> bool:
+    row = con.execute(
+        """
+        SELECT 1
+        FROM lead_evidence.source
+        WHERE project = ?
+          AND database = 'cms'
+          AND created_at > ?
+        LIMIT 1
+        """,
+        [project, latest],
+    ).fetchone()
+    return row is not None
+
+
+def due(con: duckdb.DuckDBPyConnection, *, project: str, now: str) -> tuple[str, ...]:
+    """Reasons a search should run, in a fixed order."""
+    reasons: list[str] = []
+    latest = _latest(con, project)
+    if latest is None or latest < now:
+        reasons.append("nightly")
+    if _contradiction(con, project):
+        reasons.append("contradiction")
+    if _exogenous(con, project):
+        reasons.append("exogenous")
+    if latest is not None and _cms(con, project, latest):
+        reasons.append("cms")
+    return tuple(reasons)

tests/prisma/test_search_due.py

@@ -0,0 +1,226 @@
+"""Search is due nightly, and also when a trigger lands. No search is run."""
+
+from __future__ import annotations
+
+import pytest
+from typer.testing import CliRunner
+
+from cli.prisma import app
+from lead.ledger.store import append, open_ledger
+from prisma.search.due import due
+
+runner = CliRunner()
+PROJECT = "lead-precursors"
+NOW = "2026-10-31"
+YESTERDAY = "2026-10-30"
+
+
+def _ledger(tmp_path):
+    return open_ledger(str(tmp_path / "ledger.duckdb"))
+
+
+def _search(
+    con, *, searched_on: str, project: str = PROJECT, suffix: str = "s"
+) -> None:
+    assert append(
+        con,
+        "search",
+        {
+            "project": project,
+            "database": "pubmed",
+            "interface": "eutils",
+            "searched_on": searched_on,
+            "query": "lead",
+            "filters": "",
+            "hit_count": 1,
+            "export_sha256": suffix,
+            "actor": "tester",
+            "content_hash": f"search-{suffix}",
+        },
+    )
+
+
+def _claim(con, *, certainty: str, suffix: str, target_id: str = "Q.ACR") -> None:
+    assert append(
+        con,
+        "claim",
+        {
+            "id": suffix,
+            "project": PROJECT,
+            "kind": "precursor",
+            "target_id": target_id,
+            "certainty": certainty,
+            "actor": "tester",
+            "content_hash": f"claim-{suffix}",
+        },
+    )
+
+
+def _finding(con, *, claim_id: str, suffix: str) -> None:
+    assert append(
+        con,
+        "finding",
+        {
+            "claim_id": claim_id,
+            "relation": "contradicts",
+            "grain": "beneficiary",
+            "actor": "tester",
+            "content_hash": f"finding-{suffix}",
+        },
+    )
+
+
+def test_missing_or_stale_search_is_nightly(tmp_path):
+    con = _ledger(tmp_path)
+    try:
+        assert due(con, project=PROJECT, now=NOW) == ("nightly",)
+        _search(con, searched_on=NOW, suffix="today")
+        assert due(con, project=PROJECT, now=NOW) == ()
+        _search(con, searched_on=YESTERDAY, suffix="yesterday", project="lead-history")
+        assert due(con, project=PROJECT, now=NOW) == ()
+    finally:
+        con.close()
+
+    con = open_ledger(str(tmp_path / "stale.duckdb"))
+    try:
+        _search(con, searched_on=YESTERDAY, suffix="old")
+        assert due(con, project=PROJECT, now=NOW) == ("nightly",)
+    finally:
+        con.close()
+
+
+def test_another_project_does_not_clear_nightly(tmp_path):
+    con = _ledger(tmp_path)
+    try:
+        _search(con, searched_on=NOW, project="lead-history", suffix="other")
+        assert due(con, project=PROJECT, now=NOW) == ("nightly",)
+    finally:
+        con.close()
+
+
+@pytest.mark.parametrize(
+    ("certainty", "expected"),
+    [("low", ()), ("moderate", ("contradiction",)), ("high", ("contradiction",))],
+)
+def test_contradiction_respects_certainty(tmp_path, certainty, expected):
+    con = _ledger(tmp_path)
+    try:
+        _search(con, searched_on=NOW)
+        _claim(con, certainty=certainty, suffix=certainty)
+        _finding(con, claim_id=certainty, suffix=certainty)
+        assert due(con, project=PROJECT, now=NOW) == expected
+    finally:
+        con.close()
+
+
+def test_exogenous_event_on_a_claimed_target(tmp_path):
+    con = _ledger(tmp_path)
+    try:
+        _search(con, searched_on=NOW)
+        assert append(
+            con,
+            "exogenous",
+            {
+                "project": PROJECT,
+                "target_id": "Q.ACR",
+                "series_or_event": "cms guidance",
+                "event_on": NOW,
+                "citation": "cms",
+                "actor": "tester",
+                "content_hash": "exo-1",
+            },
+        )
+        assert due(con, project=PROJECT, now=NOW) == ()
+        _claim(con, certainty="low", suffix="qacr")
+        assert due(con, project=PROJECT, now=NOW) == ("exogenous",)
+    finally:
+        con.close()
+
+
+def test_cms_source_newer_than_the_search(tmp_path):
+    con = _ledger(tmp_path)
+    try:
+        _search(con, searched_on=NOW)
+        assert append(
+            con,
+            "source",
+            {
+                "project": PROJECT,
+                "database": "cms",
+                "zotero_key": "ABCD2345",
+                "created_at": NOW,
+                "actor": "tester",
+                "content_hash": "src-same-day",
+            },
+        )
+        assert due(con, project=PROJECT, now=NOW) == ()
+        assert append(
+            con,
+            "source",
+            {
+                "project": PROJECT,
+                "database": "cms",
+                "zotero_key": "EFGH5678",
+                "created_at": "2026-11-01",
+                "actor": "tester",
+                "content_hash": "src-later",
+            },
+        )
+        assert due(con, project=PROJECT, now=NOW) == ("cms",)
+    finally:
+        con.close()
+
+
+def test_reasons_stay_in_order(tmp_path):
+    con = _ledger(tmp_path)
+    try:
+        _search(con, searched_on=YESTERDAY, suffix="old")
+        _claim(con, certainty="high", suffix="high")
+        _finding(con, claim_id="high", suffix="high")
+        assert append(
+            con,
+            "exogenous",
+            {
+                "target_id": "Q.ACR",
+                "series_or_event": "cms guidance",
+                "event_on": NOW,
+                "actor": "tester",
+                "content_hash": "exo-order",
+            },
+        )
+        assert append(
+            con,
+            "source",
+            {
+                "project": PROJECT,
+                "database": "cms",
+                "created_at": "2026-11-01",
+                "actor": "tester",
+                "content_hash": "src-order",
+            },
+        )
+        assert due(con, project=PROJECT, now=NOW) == (
+            "nightly",
+            "contradiction",
+            "exogenous",
+            "cms",
+        )
+    finally:
+        con.close()
+
+
+def test_due_command_is_registered_and_quiet_when_current(tmp_path):
+    result = runner.invoke(app, ["search", "due", "--help"])
+    assert result.exit_code == 0
+    path = tmp_path / "ledger.duckdb"
+    con = open_ledger(str(path))
+    try:
+        _search(con, searched_on=NOW, suffix="cli")
+    finally:
+        con.close()
+    result = runner.invoke(
+        app,
+        ["search", "due", "--db", str(path), "--project", PROJECT, "--now", NOW],
+    )
+    assert result.exit_code == 0
+    assert result.stdout.strip() == ""