codex/discover.py

@@ -1,22 +1,32 @@
-"""Discovery queries over the citation graph stored in PostgreSQL."""
+"""Discovery queries over the citation graph stored in PostgreSQL.
+
+Every query resolves ``citations.cited_id`` to a canonical ``papers.id`` through
+the shared :data:`codex.graph.RESOLVED_CITATIONS_SQL`, so this module and the
+citation-graph layer agree on what counts as "ingested". Previously this module
+compared the raw OpenAlex ``cited_id`` against ``papers.id`` (a DOI/arXiv id),
+which mis-reported already-ingested papers as discovery leads (audit C-1).
+"""
 
 from __future__ import annotations
 
 from codex.db import get_conn
+from codex.graph import RESOLVED_CITATIONS_SQL
 
 
 def discovery_leads(limit: int = 20) -> list[dict[str, int | str]]:
     """Return papers referenced by already-ingested papers but not yet collected.
 
     Returns list of dicts with keys:
-        cited_id (str)   — arXiv ID, DOI, or OpenAlex W-ID of the target
+        cited_id (str)   — canonical id of the referenced (not-yet-ingested) work
         pull     (int)   — how many already-ingested papers cite this target
 
-    Ordered by pull DESC, limited to `limit` rows.
+    Ordered by pull DESC, limited to `limit` rows. References that resolve to a
+    paper already in the corpus are excluded — they are not leads.
     """
-    sql = """
+    sql = f"""
+        WITH resolved AS ({RESOLVED_CITATIONS_SQL})
         SELECT cited_id, count(*) AS pull
-        FROM   citations
+        FROM   resolved
         WHERE  cited_id NOT IN (SELECT id FROM papers)
         GROUP  BY cited_id
         ORDER  BY pull DESC
@@ -29,7 +39,10 @@ def discovery_leads(limit: int = 20) -> list[dict[str, int | str]]:
 
 def citing_papers(paper_id: str) -> list[str]:
     """Return IDs of all papers that cite `paper_id` (in-graph reverse citations)."""
-    sql = "SELECT citing_id FROM citations WHERE cited_id = %(paper_id)s"
+    sql = f"""
+        WITH resolved AS ({RESOLVED_CITATIONS_SQL})
+        SELECT citing_id FROM resolved WHERE cited_id = %(paper_id)s
+    """
     with get_conn() as conn:
         rows = conn.execute(sql, {"paper_id": paper_id}).fetchall()
     return [row["citing_id"] for row in rows]
@@ -37,7 +50,10 @@ def citing_papers(paper_id: str) -> list[str]:
 
 def cited_by(paper_id: str) -> list[str]:
     """Return IDs of all papers that `paper_id` cites (forward citations)."""
-    sql = "SELECT cited_id FROM citations WHERE citing_id = %(paper_id)s"
+    sql = f"""
+        WITH resolved AS ({RESOLVED_CITATIONS_SQL})
+        SELECT cited_id FROM resolved WHERE citing_id = %(paper_id)s
+    """
     with get_conn() as conn:
         rows = conn.execute(sql, {"paper_id": paper_id}).fetchall()
     return [row["cited_id"] for row in rows]
@@ -49,10 +65,11 @@ def cocited_papers(paper_id: str, limit: int = 10) -> list[dict[str, int | str]]
     A paper X is co-cited with `paper_id` if some paper cites both.
     Returns list of dicts: {paper_id: str, co_citations: int}, ordered DESC.
     """
-    sql = """
+    sql = f"""
+        WITH resolved AS ({RESOLVED_CITATIONS_SQL})
         SELECT c2.cited_id AS paper_id, count(*) AS co_citations
-        FROM   citations c1
+        FROM   resolved c1
-        JOIN   citations c2 ON c1.citing_id = c2.citing_id
+        JOIN   resolved c2 ON c1.citing_id = c2.citing_id
         WHERE  c1.cited_id = %(paper_id)s
           AND  c2.cited_id != %(paper_id)s
         GROUP  BY c2.cited_id

codex/graph.py

@@ -23,6 +23,23 @@ logger = logging.getLogger(__name__)
 
 _MIN_PAGERANK_NODES = 5
 
+# Single source of truth for resolving ``citations.cited_id`` to a canonical
+# paper id. ``cited_id`` stores OpenAlex IDs while ``papers.id`` stores DOIs /
+# arXiv ids, so an in-KB reference cited by its OpenAlex id does not match
+# ``papers.id`` directly. This resolves it to the canonical ``papers.id`` (via
+# ``papers.openalex_id``, then the ``paper_identifiers`` alias table); dangling
+# references keep their raw id. Both :func:`build_citation_graph` and
+# :mod:`codex.discover` query through this so the "what counts as ingested" rule
+# cannot drift between them again (audit C-1). Trusted constant — no user input.
+RESOLVED_CITATIONS_SQL = """
+    SELECT c.citing_id,
+           COALESCE(p.id, pi.paper_id, c.cited_id) AS cited_id
+    FROM citations c
+    LEFT JOIN papers p ON p.openalex_id = c.cited_id
+    LEFT JOIN paper_identifiers pi
+           ON pi.openalex_id = c.cited_id AND p.id IS NULL
+"""
+
 
 def build_citation_graph(conn: Any) -> nx.DiGraph:
     """Load the ``citations`` table into a directed graph.
@@ -45,15 +62,7 @@ def build_citation_graph(conn: Any) -> nx.DiGraph:
     -------
     nx.DiGraph with every (citing_id, cited_id) pair as an edge.
     """
-    rows = conn.execute(
+    rows = conn.execute(RESOLVED_CITATIONS_SQL).fetchall()
-        """
-        SELECT c.citing_id, COALESCE(p.id, pi.paper_id, c.cited_id) AS cited_id
-        FROM citations c
-        LEFT JOIN papers p ON p.openalex_id = c.cited_id
-        LEFT JOIN paper_identifiers pi
-               ON pi.openalex_id = c.cited_id AND p.id IS NULL
-        """
-    ).fetchall()
     g: nx.DiGraph = nx.DiGraph()
     for row in rows:
         g.add_edge(row["citing_id"], row["cited_id"])