feat: collapsible expansion, canonical formalism pages, citation enrichment

db.py

@@ -0,0 +1,591 @@
+"""
+SQLite interface for paper cache, extraction results, and match outcomes.
+
+Schema:
+  papers:        arXiv metadata + pipeline status + citation metadata
+  concepts:      extracted concepts per paper (from LLM deconstruction)
+  reductions:    match results per concept (from compositional matching engine)
+  formalism_kb:  cached KB entries for fast lookup (mirrors YAML)
+  pipeline_runs: audit log of pipeline executions
+
+Thread-safe: uses WAL mode. Single-writer by design (pipeline is sequential).
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import time
+from dataclasses import dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Optional
+
+# ---------------------------------------------------------------------------
+# Schema
+# ---------------------------------------------------------------------------
+
+SCHEMA_SQL = """
+PRAGMA journal_mode=WAL;
+PRAGMA foreign_keys=ON;
+
+CREATE TABLE IF NOT EXISTS papers (
+    arxiv_id        TEXT PRIMARY KEY,
+    title           TEXT NOT NULL,
+    abstract        TEXT NOT NULL,
+    authors         TEXT NOT NULL,          -- JSON array of strings
+    categories      TEXT NOT NULL,          -- JSON array of strings
+    published       TEXT NOT NULL,          -- ISO 8601
+    updated         TEXT NOT NULL,          -- ISO 8601
+    pdf_url         TEXT,
+    -- Pipeline status
+    status          TEXT NOT NULL DEFAULT 'ingested',  -- ingested|triaged|extracted|matched|displayed|skipped|error
+    triage_passed   INTEGER,               -- 1 = novelty claim detected, 0 = skipped, NULL = not triaged
+    triage_reason   TEXT,                   -- why it passed or was skipped
+    ingestion_ts    TEXT NOT NULL DEFAULT (datetime('now')),
+    extraction_ts   TEXT,
+    matching_ts     TEXT,
+    error_message   TEXT,
+    citation_count  INTEGER DEFAULT 0,
+    citation_fetched_ts TEXT
+);
+
+CREATE TABLE IF NOT EXISTS concepts (
+    id              INTEGER PRIMARY KEY AUTOINCREMENT,
+    paper_arxiv_id  TEXT NOT NULL REFERENCES papers(arxiv_id) ON DELETE CASCADE,
+    name            TEXT NOT NULL,          -- the term the paper uses
+    is_claimed_novel INTEGER NOT NULL DEFAULT 0,
+    claimed_novelty_text TEXT,
+    mathematical_operation TEXT NOT NULL,
+    domain          TEXT,
+    codomain        TEXT,
+    objective       TEXT,
+    constraints     TEXT,                   -- JSON array
+    canonical_analog TEXT,
+    deconstructive_move TEXT,
+    confidence      TEXT NOT NULL DEFAULT 'low',  -- high|medium|low
+    confidence_rationale TEXT,
+    flags           TEXT,                   -- JSON array
+    -- Extraction metadata
+    extraction_json TEXT NOT NULL,          -- full concept JSON from LLM
+    extracted_at    TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE TABLE IF NOT EXISTS reductions (
+    id              INTEGER PRIMARY KEY AUTOINCREMENT,
+    concept_id      INTEGER NOT NULL REFERENCES concepts(id) ON DELETE CASCADE,
+    paper_arxiv_id  TEXT NOT NULL REFERENCES papers(arxiv_id) ON DELETE CASCADE,
+    concept_name    TEXT NOT NULL,
+    result_type     TEXT NOT NULL,          -- identity|compositional|analogy|unknown|confused
+    reduction       TEXT NOT NULL,          -- e.g., "Kernel CCA ∘ neuralize ∘ predict_in_codomain"
+    canonical_analog TEXT,
+    genuine_delta   TEXT,
+    micro           TEXT,
+    meso            TEXT,
+    macro           TEXT,
+    confidence      REAL NOT NULL DEFAULT 0.0,
+    display         TEXT NOT NULL,          -- sous rature formatted string
+    notes           TEXT,                   -- JSON array
+    match_json      TEXT NOT NULL,          -- full MatchResult as JSON
+    matched_at      TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE TABLE IF NOT EXISTS formalism_kb (
+    id              TEXT PRIMARY KEY,       -- matches formalism.id in YAML
+    name            TEXT NOT NULL,
+    signature_json  TEXT NOT NULL,          -- JSON: {operation, domain, codomain, objective_family}
+    meso_type       TEXT,
+    macro_type      TEXT,
+    canonical_reference TEXT,
+    researchor_artifact_id TEXT,
+    researchor_mental_model_id TEXT,
+    status          TEXT NOT NULL DEFAULT 'seed',
+    cached_at       TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE TABLE IF NOT EXISTS pipeline_runs (
+    id              INTEGER PRIMARY KEY AUTOINCREMENT,
+    run_type        TEXT NOT NULL,          -- daily|single|retroactive|manual
+    started_at      TEXT NOT NULL DEFAULT (datetime('now')),
+    finished_at     TEXT,
+    papers_ingested INTEGER DEFAULT 0,
+    papers_triaged  INTEGER DEFAULT 0,
+    papers_extracted INTEGER DEFAULT 0,
+    papers_matched  INTEGER DEFAULT 0,
+    papers_error    INTEGER DEFAULT 0,
+    status          TEXT NOT NULL DEFAULT 'running',  -- running|completed|failed
+    error_message   TEXT,
+    config_json     TEXT                    -- snapshot of pipeline config at run time
+);
+
+CREATE INDEX IF NOT EXISTS idx_papers_status ON papers(status);
+CREATE INDEX IF NOT EXISTS idx_papers_updated ON papers(updated);
+CREATE INDEX IF NOT EXISTS idx_concepts_paper ON concepts(paper_arxiv_id);
+CREATE INDEX IF NOT EXISTS idx_reductions_paper ON reductions(paper_arxiv_id);
+CREATE INDEX IF NOT EXISTS idx_reductions_concept ON reductions(concept_id);
+CREATE INDEX IF NOT EXISTS idx_reductions_type ON reductions(result_type);
+CREATE INDEX IF NOT EXISTS idx_pipeline_runs_started ON pipeline_runs(started_at);
+
+-- Migration: add citation columns (safe to run on existing DBs)
+ALTER TABLE papers ADD COLUMN citation_count INTEGER DEFAULT 0;
+ALTER TABLE papers ADD COLUMN citation_fetched_ts TEXT;
+"""
+
+
+# ---------------------------------------------------------------------------
+# Database wrapper
+# ---------------------------------------------------------------------------
+
+@dataclass
+class Database:
+    """SQLite database interface for the Différance Engine pipeline."""
+
+    path: Path
+    _conn: sqlite3.Connection | None = field(default=None, repr=False, init=False)
+
+    def __post_init__(self):
+        self.path = Path(self.path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+
+    def connect(self):
+        """Open connection and ensure schema exists."""
+        if self._conn is not None:
+            return
+        self._conn = sqlite3.connect(str(self.path))
+        self._conn.row_factory = sqlite3.Row
+        self._conn.executescript(SCHEMA_SQL)
+        self._conn.commit()
+
+    def close(self):
+        if self._conn is not None:
+            self._conn.close()
+            self._conn = None
+
+    def __enter__(self):
+        self.connect()
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+
+    # ---- Papers ----
+
+    def paper_exists(self, arxiv_id: str) -> bool:
+        self.connect()
+        row = self._conn.execute("SELECT 1 FROM papers WHERE arxiv_id = ?", (arxiv_id,)).fetchone()
+        return row is not None
+
+    def insert_paper(self, paper: dict) -> bool:
+        """Insert a paper from arXiv API parsed data. Returns True if new."""
+        self.connect()
+        if self.paper_exists(paper["arxiv_id"]):
+            return False
+        self._conn.execute(
+            """INSERT INTO papers (arxiv_id, title, abstract, authors, categories,
+               published, updated, pdf_url, status)
+               VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'ingested')""",
+            (
+                paper["arxiv_id"],
+                paper["title"],
+                paper["abstract"],
+                json.dumps(paper.get("authors", [])),
+                json.dumps(paper.get("categories", [])),
+                paper.get("published", ""),
+                paper.get("updated", ""),
+                paper.get("pdf_url", ""),
+            ),
+        )
+        self._conn.commit()
+        return True
+
+    def update_triage(self, arxiv_id: str, passed: bool, reason: str = ""):
+        self.connect()
+        self._conn.execute(
+            """UPDATE papers SET triage_passed = ?, triage_reason = ?,
+               status = CASE WHEN ? THEN 'triaged' ELSE 'skipped' END
+               WHERE arxiv_id = ?""",
+            (1 if passed else 0, reason, passed, arxiv_id),
+        )
+        self._conn.commit()
+
+    def update_status(self, arxiv_id: str, status: str, error: str = ""):
+        self.connect()
+        ts = datetime.now(timezone.utc).isoformat()
+        field = {"extracted": "extraction_ts", "matched": "matching_ts",
+                 "displayed": "matching_ts"}.get(status, "")
+        if field:
+            self._conn.execute(
+                f"UPDATE papers SET status = ?, {field} = ?, error_message = ? WHERE arxiv_id = ?",
+                (status, ts, error, arxiv_id),
+            )
+        else:
+            self._conn.execute(
+                "UPDATE papers SET status = ?, error_message = ? WHERE arxiv_id = ?",
+                (status, error, arxiv_id),
+            )
+        self._conn.commit()
+
+    def update_citation(self, arxiv_id: str, count: int):
+        self.connect()
+        try:
+            self._conn.execute(
+                "UPDATE papers SET citation_count = ?, citation_fetched_ts = ? WHERE arxiv_id = ?",
+                (count, "now", arxiv_id),
+            )
+            self._conn.commit()
+        except sqlite3.OperationalError:
+            # Column might not exist yet
+            try:
+                self._conn.execute("ALTER TABLE papers ADD COLUMN citation_count INTEGER DEFAULT 0")
+                self._conn.execute("ALTER TABLE papers ADD COLUMN citation_fetched_ts TEXT")
+                self._conn.execute(
+                    "UPDATE papers SET citation_count = ?, citation_fetched_ts = ? WHERE arxiv_id = ?",
+                    (count, "now", arxiv_id),
+                )
+                self._conn.commit()
+            except Exception:
+                pass
+
+    def get_papers_by_status(self, status: str, limit: int = 100) -> list[dict]:
+        self.connect()
+        rows = self._conn.execute(
+            "SELECT * FROM papers WHERE status = ? ORDER BY updated DESC LIMIT ?",
+            (status, limit),
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+
+    def get_papers_needing_extraction(self, limit: int = 10) -> list[dict]:
+        self.connect()
+        rows = self._conn.execute(
+            "SELECT * FROM papers WHERE status = 'triaged' AND triage_passed = 1 ORDER BY updated DESC LIMIT ?",
+            (limit,),
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+
+    def get_paper(self, arxiv_id: str) -> dict | None:
+        self.connect()
+        row = self._conn.execute("SELECT * FROM papers WHERE arxiv_id = ?", (arxiv_id,)).fetchone()
+        return _row_to_dict(row) if row else None
+
+    def find_paper(self, arxiv_id: str) -> dict | None:
+        """Look up a paper by arXiv ID, trying version-suffix variations.
+
+        arXiv IDs can be stored with version suffixes (e.g. 2301.07093v1)
+        but users may query without them (2301.07093). This tries exact
+        match first, then strips/adds version suffixes.
+        """
+        # 1. Exact match
+        paper = self.get_paper(arxiv_id)
+        if paper:
+            return paper
+
+        # 2. User provided no version — try v1, v2, v3
+        import re
+        if not re.search(r'v\d+$', arxiv_id):
+            for v in range(1, 4):
+                paper = self.get_paper(f"{arxiv_id}v{v}")
+                if paper:
+                    return paper
+        else:
+            # 3. User provided version — try stripping it
+            base = re.sub(r'v\d+$', '', arxiv_id)
+            paper = self.get_paper(base)
+            if paper:
+                return paper
+
+        # 4. LIKE prefix match (finds any version of this paper)
+        self.connect()
+        row = self._conn.execute(
+            "SELECT * FROM papers WHERE arxiv_id LIKE ? || '%' LIMIT 1",
+            (arxiv_id,),
+        ).fetchone()
+        return _row_to_dict(row) if row else None
+
+    def count_by_status(self) -> dict[str, int]:
+        self.connect()
+        rows = self._conn.execute(
+            "SELECT status, COUNT(*) as cnt FROM papers GROUP BY status"
+        ).fetchall()
+        return {r["status"]: r["cnt"] for r in rows}
+
+    # ---- Concepts ----
+
+    def insert_concept(self, paper_arxiv_id: str, concept: dict):
+        self.connect()
+        self._conn.execute(
+            """INSERT INTO concepts (paper_arxiv_id, name, is_claimed_novel,
+               claimed_novelty_text, mathematical_operation, domain, codomain,
+               objective, constraints, canonical_analog, deconstructive_move,
+               confidence, confidence_rationale, flags, extraction_json)
+               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+            (
+                paper_arxiv_id,
+                concept.get("name", ""),
+                1 if concept.get("is_claimed_novel") else 0,
+                concept.get("claimed_novelty_text"),
+                concept.get("mathematical_operation", ""),
+                concept.get("domain"),
+                concept.get("codomain"),
+                concept.get("objective"),
+                json.dumps(concept.get("constraints", [])),
+                concept.get("canonical_analog"),
+                concept.get("deconstructive_move"),
+                concept.get("confidence", "low"),
+                concept.get("confidence_rationale"),
+                json.dumps(concept.get("flags", [])),
+                json.dumps(concept),
+            ),
+        )
+        self._conn.commit()
+        return self._conn.execute("SELECT last_insert_rowid()").fetchone()[0]
+
+    def delete_concepts_for_paper(self, arxiv_id: str):
+        self.connect()
+        self._conn.execute("DELETE FROM concepts WHERE paper_arxiv_id = ?", (arxiv_id,))
+        self._conn.commit()
+
+    # ---- Reductions ----
+
+    def insert_reduction(self, concept_id: int, paper_arxiv_id: str, match: dict):
+        self.connect()
+        self._conn.execute(
+            """INSERT INTO reductions (concept_id, paper_arxiv_id, concept_name,
+               result_type, reduction, canonical_analog, genuine_delta, micro,
+               meso, macro, confidence, display, notes, match_json)
+               VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+            (
+                concept_id, paper_arxiv_id,
+                match.get("concept_name", ""),
+                match.get("result_type", ""),
+                match.get("reduction", ""),
+                match.get("canonical_analog", ""),
+                match.get("genuine_delta", ""),
+                match.get("micro", ""),
+                match.get("meso", ""),
+                match.get("macro", ""),
+                match.get("confidence", 0.0),
+                match.get("display", ""),
+                json.dumps(match.get("notes", [])),
+                json.dumps(match),
+            ),
+        )
+        self._conn.commit()
+
+    def delete_reductions_for_paper(self, arxiv_id: str):
+        self.connect()
+        self._conn.execute("DELETE FROM reductions WHERE paper_arxiv_id = ?", (arxiv_id,))
+        self._conn.commit()
+
+    def get_reductions_for_paper(self, arxiv_id: str) -> list[dict]:
+        self.connect()
+        rows = self._conn.execute(
+            "SELECT * FROM reductions WHERE paper_arxiv_id = ? ORDER BY id",
+            (arxiv_id,),
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+
+    def get_displayable_papers(self, limit: int = 50) -> list[dict]:
+        """Papers ready for display: matched results exist."""
+        self.connect()
+        rows = self._conn.execute(
+            """SELECT p.* FROM papers p
+               WHERE p.status = 'matched'
+               ORDER BY p.updated DESC LIMIT ?""",
+            (limit,),
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+
+    # ---- Cross-reference indexing ----
+
+    def get_canonical_analogs(self) -> list[dict]:
+        """All distinct canonical analogs cited, with paper counts."""
+        self.connect()
+        rows = self._conn.execute(
+            """SELECT canonical_analog, COUNT(*) as paper_count,
+                      GROUP_CONCAT(DISTINCT paper_arxiv_id) as paper_ids
+               FROM reductions
+               WHERE canonical_analog IS NOT NULL AND canonical_analog != ''
+               GROUP BY canonical_analog
+               ORDER BY paper_count DESC"""
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+
+    def get_papers_by_analog(self, analog: str, limit: int = 20) -> list[dict]:
+        """Papers that share a canonical analog (fuzzy match)."""
+        self.connect()
+        rows = self._conn.execute(
+            """SELECT DISTINCT p.* FROM papers p
+               JOIN reductions r ON r.paper_arxiv_id = p.arxiv_id
+               WHERE r.canonical_analog LIKE ?
+               ORDER BY p.updated DESC LIMIT ?""",
+            (f"%{analog}%", limit),
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+
+    def get_papers_by_move(self, move: str, limit: int = 20) -> list[dict]:
+        """Papers whose concepts use a specific deconstructive move."""
+        self.connect()
+        rows = self._conn.execute(
+            """SELECT DISTINCT p.*, c.deconstructive_move, c.name as concept_name
+               FROM papers p
+               JOIN concepts c ON c.paper_arxiv_id = p.arxiv_id
+               WHERE c.deconstructive_move = ?
+               ORDER BY p.updated DESC LIMIT ?""",
+            (move, limit),
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+
+    def get_move_counts(self) -> list[dict]:
+        """Count of each deconstructive move across all concepts."""
+        self.connect()
+        rows = self._conn.execute(
+            """SELECT deconstructive_move, COUNT(*) as cnt
+               FROM concepts
+               WHERE deconstructive_move IS NOT NULL
+               GROUP BY deconstructive_move
+               ORDER BY cnt DESC"""
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+
+    def get_papers_with_unknown(self, limit: int = 100) -> list[dict]:
+        """Papers with at least one UNKNOWN reduction (retroactive queue)."""
+        self.connect()
+        rows = self._conn.execute(
+            """SELECT DISTINCT p.* FROM papers p
+               JOIN reductions r ON r.paper_arxiv_id = p.arxiv_id
+               WHERE r.result_type = 'unknown'
+               ORDER BY p.updated DESC LIMIT ?""",
+            (limit,),
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+
+    # ---- KB cache ----
+
+    def sync_kb_cache(self, formalisms: list[dict]):
+        """Sync the cached KB table from the current formalisms list."""
+        self.connect()
+        self._conn.execute("DELETE FROM formalism_kb")
+        for fm in formalisms:
+            self._conn.execute(
+                """INSERT OR REPLACE INTO formalism_kb
+                   (id, name, signature_json, meso_type, macro_type,
+                    canonical_reference, researchor_artifact_id,
+                    researchor_mental_model_id, status)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+                (
+                    fm["id"],
+                    fm["name"],
+                    json.dumps(fm.get("signature", {})),
+                    fm.get("meso_type"),
+                    fm.get("macro_type"),
+                    fm.get("canonical_reference"),
+                    fm.get("researchor_artifact_id"),
+                    fm.get("researchor_mental_model_id"),
+                    fm.get("status", "seed"),
+                ),
+            )
+        self._conn.commit()
+
+    # ---- Pipeline runs ----
+
+    def start_run(self, run_type: str, config: dict | None = None) -> int:
+        self.connect()
+        cur = self._conn.execute(
+            "INSERT INTO pipeline_runs (run_type, config_json) VALUES (?, ?)",
+            (run_type, json.dumps(config) if config else "{}"),
+        )
+        self._conn.commit()
+        return cur.lastrowid
+
+    def finish_run(self, run_id: int, counts: dict, error: str = ""):
+        self.connect()
+        status = "failed" if error else "completed"
+        self._conn.execute(
+            """UPDATE pipeline_runs SET finished_at = ?, papers_ingested = ?,
+               papers_triaged = ?, papers_extracted = ?, papers_matched = ?,
+               papers_error = ?, status = ?, error_message = ?
+               WHERE id = ?""",
+            (
+                datetime.now(timezone.utc).isoformat(),
+                counts.get("ingested", 0),
+                counts.get("triaged", 0),
+                counts.get("extracted", 0),
+                counts.get("matched", 0),
+                counts.get("errors", 0),
+                status,
+                error,
+                run_id,
+            ),
+        )
+        self._conn.commit()
+
+    def recent_runs(self, limit: int = 10) -> list[dict]:
+        self.connect()
+        rows = self._conn.execute(
+            "SELECT * FROM pipeline_runs ORDER BY started_at DESC LIMIT ?",
+            (limit,),
+        ).fetchall()
+        return [_row_to_dict(r) for r in rows]
+
+    # ---- Stats ----
+
+    def stats(self) -> dict:
+        self.connect()
+        counts = self.count_by_status()
+        total = sum(counts.values())
+
+        # Reduction type breakdown
+        red_rows = self._conn.execute(
+            "SELECT result_type, COUNT(*) as cnt FROM reductions GROUP BY result_type"
+        ).fetchall()
+        reduction_counts = {r["result_type"]: r["cnt"] for r in red_rows}
+
+        # Top deconstructive moves
+        move_rows = self._conn.execute(
+            "SELECT deconstructive_move, COUNT(*) as cnt FROM concepts WHERE deconstructive_move IS NOT NULL GROUP BY deconstructive_move ORDER BY cnt DESC LIMIT 5"
+        ).fetchall()
+
+        return {
+            "total_papers": total,
+            "by_status": counts,
+            "total_concepts": sum(
+                r["cnt"] for r in self._conn.execute("SELECT COUNT(*) as cnt FROM concepts").fetchall()
+            ),
+            "total_reductions": sum(reduction_counts.values()),
+            "reductions_by_type": reduction_counts,
+            "reduction_rate": (
+                (reduction_counts.get("identity", 0) + reduction_counts.get("compositional", 0))
+                / max(sum(reduction_counts.values()), 1)
+            ),
+            "top_moves": [{"move": r["deconstructive_move"], "count": r["cnt"]} for r in move_rows],
+            "recent_runs": self.recent_runs(3),
+        }
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _row_to_dict(row: sqlite3.Row | None) -> dict | None:
+    if row is None:
+        return None
+    d = dict(row)
+    # Deserialize JSON fields
+    for field in ("authors", "categories", "constraints", "flags", "notes"):
+        if field in d and isinstance(d[field], str):
+            try:
+                d[field] = json.loads(d[field])
+            except (json.JSONDecodeError, TypeError):
+                pass
+    return d
+
+
+# ---------------------------------------------------------------------------
+# Convenience
+# ---------------------------------------------------------------------------
+
+def get_db(path: Path | str | None = None) -> Database:
+    """Get a Database instance for the default data directory."""
+    if path is None:
+        path = Path(__file__).resolve().parent.parent / "data" / "papers.db"
+    return Database(Path(path))