Deduplicate call-log prompt text into a hash-keyed prompts table

src/xrouter_llm/migrations/versions/0004_dedupe_prompt_text.py

@@ -0,0 +1,66 @@
+"""Deduplicate prompt text into a prompts table
+
+Revision ID: 0004
+Revises: 0003
+Create Date: 2026-07-08
+"""
+from __future__ import annotations
+
+import hashlib
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "0004"
+down_revision: str = "0003"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "prompts",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("sha256", sa.String(64), nullable=False),
+        sa.Column("text", sa.Text(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("sha256", name="uq_prompts_sha256"),
+    )
+
+    conn = op.get_bind()
+    prompts = conn.execute(sa.text("SELECT DISTINCT prompt FROM calls")).fetchall()
+    for (prompt,) in prompts:
+        conn.execute(
+            sa.text("INSERT INTO prompts (sha256, text) VALUES (:sha, :text)"),
+            {"sha": hashlib.sha256(prompt.encode("utf-8")).hexdigest(), "text": prompt},
+        )
+
+    op.add_column("calls", sa.Column("prompt_id", sa.Integer(), nullable=True))
+    conn.execute(sa.text(
+        "UPDATE calls SET prompt_id = "
+        "(SELECT p.id FROM prompts p WHERE p.text = calls.prompt)"
+    ))
+
+    # batch mode: SQLite cannot ALTER to NOT NULL / add FK in place
+    with op.batch_alter_table("calls") as batch:
+        batch.alter_column("prompt_id", existing_type=sa.Integer(), nullable=False)
+        batch.create_foreign_key(
+            "fk_calls_prompt_id_prompts", "prompts", ["prompt_id"], ["id"]
+        )
+        batch.drop_column("prompt")
+    op.create_index("ix_calls_prompt_id", "calls", ["prompt_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_calls_prompt_id", table_name="calls")
+    op.add_column("calls", sa.Column("prompt", sa.Text(), nullable=True))
+    conn = op.get_bind()
+    conn.execute(sa.text(
+        "UPDATE calls SET prompt = "
+        "(SELECT p.text FROM prompts p WHERE p.id = calls.prompt_id)"
+    ))
+    with op.batch_alter_table("calls") as batch:
+        batch.alter_column("prompt", existing_type=sa.Text(), nullable=False)
+        batch.drop_constraint("fk_calls_prompt_id_prompts", type_="foreignkey")
+        batch.drop_column("prompt_id")
+    op.drop_table("prompts")

src/xrouter_llm/store.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import hashlib
 from pathlib import Path
 from typing import Any
 
@@ -10,7 +11,15 @@
 from alembic.config import Config as AlembicConfig
 from sqlalchemy import create_engine
 from sqlalchemy.engine import Engine
-from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import (
+    DeclarativeBase,
+    Mapped,
+    Session,
+    mapped_column,
+    relationship,
+    sessionmaker,
+)
 from sqlalchemy.pool import StaticPool
 
 _MIGRATIONS_DIR = Path(__file__).parent / "migrations"
@@ -20,13 +29,25 @@ class Base(DeclarativeBase):
     pass
 
 
+class PromptRecord(Base):
+    """Prompt text stored once per distinct prompt, keyed by content hash."""
+
+    __tablename__ = "prompts"
+
+    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    sha256: Mapped[str] = mapped_column(sa.String(64), nullable=False, unique=True)
+    text: Mapped[str] = mapped_column(sa.Text, nullable=False)
+
+
 class CallRecord(Base):
     __tablename__ = "calls"
 
     id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
     ts: Mapped[float] = mapped_column(sa.Float, nullable=False)
     config: Mapped[str] = mapped_column(sa.String(255), nullable=False)
-    prompt: Mapped[str] = mapped_column(sa.Text, nullable=False)
+    prompt_id: Mapped[int] = mapped_column(
+        sa.ForeignKey("prompts.id"), nullable=False
+    )
     task: Mapped[str | None] = mapped_column(sa.String(255), nullable=True)
     selected: Mapped[Any] = mapped_column(sa.JSON, nullable=False)
     candidates: Mapped[Any] = mapped_column(sa.JSON, nullable=False)
@@ -36,8 +57,11 @@ class CallRecord(Base):
     feedback: Mapped[Any] = mapped_column(sa.JSON, nullable=True)
     user_id: Mapped[str | None] = mapped_column(sa.String(255), nullable=True)
 
+    prompt_rec: Mapped[PromptRecord] = relationship(lazy="joined")
+
     __table_args__ = (
         sa.Index("ix_calls_user_id_id", "user_id", "id"),
+        sa.Index("ix_calls_prompt_id", "prompt_id"),
     )
 
 
@@ -47,9 +71,14 @@ class CallRecord(Base):
 _SCHEMA_CHECKPOINTS: list[tuple[str, str]] = [
     ("feedback", "0002"),
     ("user_id", "0003"),
+    ("prompt_id", "0004"),
 ]
 
 
+def prompt_sha256(prompt: str) -> str:
+    return hashlib.sha256(prompt.encode("utf-8")).hexdigest()
+
+
 def run_migrations(engine: Engine) -> None:
     """Run Alembic migrations using an already-created engine.
 
@@ -153,7 +182,7 @@ def record(
             row = CallRecord(
                 ts=ts,
                 config=config,
-                prompt=prompt,
+                prompt_id=_get_or_create_prompt(session, prompt),
                 task=task,
                 selected=selected,
                 candidates=candidates,
@@ -196,7 +225,20 @@ def delete(self, call_id: int, *, owner_user_id: str | None = None) -> bool:
             row = session.scalars(stmt).first()
             if row is None:
                 return False
+            prompt_id = row.prompt_id
             session.delete(row)
+            session.flush()
+            # GC the prompt text once no call references it (the log holds
+            # user prompts; orphaned text must not outlive its last call).
+            still_referenced = session.scalar(
+                sa.select(sa.func.count(CallRecord.id)).where(
+                    CallRecord.prompt_id == prompt_id
+                )
+            )
+            if not still_referenced:
+                prompt_row = session.get(PromptRecord, prompt_id)
+                if prompt_row is not None:
+                    session.delete(prompt_row)
             session.commit()
             return True
 
@@ -232,12 +274,36 @@ def model_counts(self) -> dict[str, int]:
         return counts
 
 
+def _get_or_create_prompt(session: Session, prompt: str) -> int:
+    """Return the id of the deduplicated prompt row, inserting if new.
+
+    Insert-then-recover (savepoint) rather than check-then-insert alone, so a
+    concurrent writer inserting the same hash cannot fail this transaction.
+    """
+    sha = prompt_sha256(prompt)
+    prompt_id = session.scalar(
+        sa.select(PromptRecord.id).where(PromptRecord.sha256 == sha)
+    )
+    if prompt_id is not None:
+        return prompt_id
+    try:
+        with session.begin_nested():
+            row = PromptRecord(sha256=sha, text=prompt)
+            session.add(row)
+            session.flush()
+            return row.id
+    except IntegrityError:
+        return session.scalar(
+            sa.select(PromptRecord.id).where(PromptRecord.sha256 == sha)
+        )
+
+
 def _row_to_dict(r: CallRecord) -> dict[str, Any]:
     return {
         "id": r.id,
         "ts": r.ts,
         "config": r.config,
-        "prompt": r.prompt,
+        "prompt": r.prompt_rec.text,
         "task": r.task,
         "selected": r.selected,
         "candidates": r.candidates,

tests/conftest.py

@@ -6,7 +6,13 @@
 import pytest
 import sqlalchemy as sa
 
-from xrouter_llm.store import CallRecord, CallStore, make_engine, normalize_db_url
+from xrouter_llm.store import (
+    CallRecord,
+    CallStore,
+    PromptRecord,
+    make_engine,
+    normalize_db_url,
+)
 
 _SAMPLE_RECORD = dict(
     ts=1_000_000.0,
@@ -52,3 +58,4 @@ def store(db_url):
         engine = make_engine(normalize_db_url(db_url))
         with engine.begin() as conn:
             conn.execute(sa.delete(CallRecord))
+            conn.execute(sa.delete(PromptRecord))

tests/test_store.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import sqlalchemy as sa
-from xrouter_llm.store import Base, CallStore, make_engine
+from xrouter_llm.store import Base, CallStore, PromptRecord, make_engine
 
 
 def test_record_and_recent(store) -> None:
@@ -74,6 +74,44 @@ def test_model_counts(store) -> None:
     assert counts["strong"] == 1
 
 
+def _record(store, **overrides):
+    kwargs = dict(
+        ts=1.0, config="all", prompt="p", task=None,
+        selected=["m"], candidates=[], expected_quality=0.8,
+        cost=0.0, latency=0.0,
+    )
+    kwargs.update(overrides)
+    return store.record(**kwargs)
+
+
+def _prompt_count(store) -> int:
+    with store._Session() as session:
+        return session.execute(
+            sa.select(sa.func.count(PromptRecord.id))
+        ).scalar_one()
+
+
+def test_prompt_text_is_deduplicated(store) -> None:
+    _record(store, ts=1.0, prompt="same prompt")
+    _record(store, ts=2.0, prompt="same prompt")
+    _record(store, ts=3.0, prompt="other prompt")
+    assert store.count() == 3
+    assert _prompt_count(store) == 2
+    rows = store.recent()
+    assert [r["prompt"] for r in rows] == ["other prompt", "same prompt", "same prompt"]
+
+
+def test_delete_gcs_orphaned_prompt(store) -> None:
+    id1 = _record(store, ts=1.0, prompt="shared")
+    id2 = _record(store, ts=2.0, prompt="shared")
+    assert store.delete(id1) is True
+    # still referenced by the second call
+    assert _prompt_count(store) == 1
+    assert store.recent()[0]["prompt"] == "shared"
+    assert store.delete(id2) is True
+    assert _prompt_count(store) == 0
+
+
 def _legacy_db(tmp_path):
     """Return a sqlite:// URL for a pre-0002 DB: calls table without feedback column, no alembic_version.
 
@@ -145,6 +183,28 @@ def test_legacy_db_no_alembic_version(tmp_path) -> None:
     assert row["user_id"] is None
 
 
+def test_legacy_db_backfill_dedupes_prompts(tmp_path) -> None:
+    """Migration 0004 moves existing prompt text into the prompts table, deduplicated."""
+    url = _legacy_db(tmp_path)
+    engine = make_engine(url)
+    with engine.begin() as conn:
+        for i, prompt in enumerate(["dup", "dup", "unique"]):
+            conn.execute(
+                sa.text(
+                    "INSERT INTO calls (ts, config, prompt, selected, candidates) "
+                    "VALUES (:ts, 'all', :prompt, '[\"m\"]', '[]')"
+                ),
+                {"ts": float(i), "prompt": prompt},
+            )
+    engine.dispose()
+
+    store = CallStore(url)
+    assert store.count() == 3
+    assert _prompt_count(store) == 2
+    rows = store.recent()
+    assert [r["prompt"] for r in rows] == ["unique", "dup", "dup"]
+
+
 def test_legacy_db_empty_alembic_version(tmp_path) -> None:
     """CallStore recovers from a DB where alembic_version exists but is empty."""
     url = _legacy_db_empty_version(tmp_path)