feat: make embedding_vec the sole store and hard-fail pgvector-on-SQLite

Drop the legacy CSV embeddings column in c7d8e9f0a1b2 after backfill, port
readers/writers (including Module C) to embedding_vec, and SystemExit when
pgvector similarity is requested without Postgres.

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: northdpole

application/cmd/cre_main.py

@@ -1115,14 +1115,12 @@ def run_librarian(
 
     backend = RetrieverBackend(cfg.retriever_backend)
     if backend is RetrieverBackend.pgvector:
-        # Readiness must include the embedding_vec column (not dialect alone):
-        # Postgres pre-migration would otherwise fail on every retrieve().
+        # Hard-fail: do not silently fall back to in_memory / sklearn CSV paths.
+        from application.database.pgvector_utils import fail_pgvector_unavailable
+
         if not database.can_use_pgvector_similarity():
-            logger.warning(
-                "CRE_LIBRARIAN_RETRIEVER_BACKEND=pgvector selected, but "
-                "Postgres is not ready for pgvector similarity (need Alembic "
-                "c7d8e9f0a1b2 / #977 embedding_vec). Set the backend to "
-                "in_memory until then."
+            fail_pgvector_unavailable(
+                context="CRE_LIBRARIAN_RETRIEVER_BACKEND=pgvector"
             )
     # The CRE ids present in the hub are exactly the known ids the explicit
     # resolver may auto-link to (W2 seeded this from the golden set; here it is
@@ -1136,13 +1134,16 @@ def run_librarian(
         if backend is RetrieverBackend.in_memory
         else None
     )
+    pg_connection = None
+    if backend is RetrieverBackend.pgvector:
+        pg_connection = database.session.connection()
     retriever = build_retriever(
         backend,
         embed_fn=ph.get_text_embeddings,
         top_k=cfg.top_k_retrieval,
         threshold=cfg.link_threshold,
         pool=pool,
-        connection=database.session.connection(),
+        connection=pg_connection,
     )
 
     # C.2 reranker: reads each (section, candidate-CRE) pair together and

application/database/db.py

@@ -174,7 +174,6 @@ class Embeddings(BaseModel):  # type: ignore
     __tablename__ = "embeddings"
 
     id = sqla.Column(sqla.String, primary_key=True, default=generate_uuid)
-    embeddings = sqla.Column(sqla.String, nullable=False)
     doc_type = sqla.Column(sqla.String, nullable=False)
     cre_id = sqla.Column(
         sqla.String,
@@ -191,9 +190,10 @@ class Embeddings(BaseModel):  # type: ignore
     embeddings_content = sqla.Column(sqla.String, nullable=True, default=None)
     embedding_model_id = sqla.Column(sqla.String, nullable=True, default=None)
     embedding_dim = sqla.Column(sqla.Integer, nullable=True, default=None)
-    # Postgres: real ``vector(N)`` via Alembic (c7d8e9f0a1b2). SQLite tests map
-    # this as Text storing a pgvector literal for dual-write coverage.
-    embedding_vec = sqla.Column(sqla.Text, nullable=True, default=None)
+    # Postgres: real ``vector(N)`` via Alembic (c7d8e9f0a1b2) — sole store for
+    # vectors (legacy CSV ``embeddings`` column is dropped in that migration).
+    # SQLite/tests: Text holding a pgvector literal ``[1.0,2.0,...]``.
+    embedding_vec = sqla.Column(sqla.Text, nullable=False)
 
 
 class GapAnalysisResults(BaseModel):
@@ -2367,16 +2367,27 @@ def health_check(self) -> Dict[str, Any]:
         }
 
     def get_embeddings_by_doc_type(self, doc_type: str) -> Dict[str, List[float]]:
+        from application.database.pgvector_utils import (
+            parse_stored_embedding_vec,
+            require_embedding_vec_store,
+        )
+
+        require_embedding_vec_store(
+            self.session.get_bind(), context="get_embeddings_by_doc_type"
+        )
         res = {}
         embeddings = (
             self.session.query(Embeddings).filter(Embeddings.doc_type == doc_type).all()
         )
         if embeddings:
             for entry in embeddings:
+                vec = parse_stored_embedding_vec(entry.embedding_vec)
+                if not vec:
+                    continue
                 if doc_type == cre_defs.Credoctypes.CRE.value:
-                    res[entry.cre_id] = [float(e) for e in entry.embeddings.split(",")]
+                    res[entry.cre_id] = vec
                 else:
-                    res[entry.node_id] = [float(e) for e in entry.embeddings.split(",")]
+                    res[entry.node_id] = vec
         return res
 
     def get_embedding_contents_by_doc_type(self, doc_type: str) -> Dict[str, str]:
@@ -2403,6 +2414,14 @@ def get_embedding_contents_by_doc_type(self, doc_type: str) -> Dict[str, str]:
     def get_embeddings_by_doc_type_paginated(
         self, doc_type: str, page: int = 1, per_page: int = 100
     ) -> Tuple[Dict[str, List[float]], int, int]:
+        from application.database.pgvector_utils import (
+            parse_stored_embedding_vec,
+            require_embedding_vec_store,
+        )
+
+        require_embedding_vec_store(
+            self.session.get_bind(), context="get_embeddings_by_doc_type_paginated"
+        )
         res = {}
         embeddings = (
             self.session.query(Embeddings)
@@ -2412,13 +2431,21 @@ def get_embeddings_by_doc_type_paginated(
         total_pages = embeddings.pages
         if embeddings.items:
             for entry in embeddings.items:
+                vec = parse_stored_embedding_vec(entry.embedding_vec)
+                if not vec:
+                    continue
                 if doc_type == cre_defs.Credoctypes.CRE.value:
-                    res[entry.cre_id] = [float(e) for e in entry.embeddings.split(",")]
+                    res[entry.cre_id] = vec
                 else:
-                    res[entry.node_id] = [float(e) for e in entry.embeddings.split(",")]
+                    res[entry.node_id] = vec
         return res, total_pages, page
 
     def get_embeddings_for_doc(self, doc: cre_defs.Node | cre_defs.CRE) -> Embeddings:
+        from application.database.pgvector_utils import require_embedding_vec_store
+
+        require_embedding_vec_store(
+            self.session.get_bind(), context="get_embeddings_for_doc"
+        )
         if doc.doctype == cre_defs.Credoctypes.CRE:
             obj = self.session.query(CRE).filter(CRE.external_id == doc.id).first()
             return (
@@ -2487,13 +2514,15 @@ def add_embedding(
                     f"embedding dimension mismatch for {db_object.id}: "
                     f"expected {expected_dim}, got {len(embeddings)}"
                 )
-        existing = self.get_embedding(db_object.id)
-        embeddings_str = ",".join([str(e) for e in embeddings])
-        # Dual-write CSV + pgvector literal. On Postgres after migration the
-        # column is ``vector(N)``; on SQLite tests it is Text. Never LLM
-        # re-embed here — see application.database.pgvector_utils.
-        from application.database.pgvector_utils import to_pgvector_literal
+        from application.database.pgvector_utils import (
+            require_embedding_vec_store,
+            to_pgvector_literal,
+        )
 
+        require_embedding_vec_store(self.session.get_bind(), context="add_embedding")
+        existing = self.get_embedding(db_object.id)
+        # Sole store is ``embedding_vec`` (pgvector on Postgres; Text literal
+        # on SQLite). Never LLM re-embed here — see pgvector_utils.
         embedding_vec_literal = to_pgvector_literal(embeddings)
         resolved_node_url: Optional[str] = None
         if doctype != cre_defs.Credoctypes.CRE:
@@ -2505,7 +2534,6 @@ def add_embedding(
             emb = None
             if doctype == cre_defs.Credoctypes.CRE:
                 emb = Embeddings(
-                    embeddings=embeddings_str,
                     cre_id=db_object.id,
                     doc_type=cre_defs.Credoctypes.CRE.value,
                     embeddings_content=embedding_text,
@@ -2515,7 +2543,6 @@ def add_embedding(
                 )
             else:
                 emb = Embeddings(
-                    embeddings=embeddings_str,
                     node_id=db_object.id,
                     doc_type=db_object.ntype,
                     embeddings_content=embedding_text,
@@ -2529,7 +2556,6 @@ def add_embedding(
             return emb
         else:
             logger.debug(f"knew of embedding for object {db_object.id} ,updating")
-            existing[0].embeddings = embeddings_str
             existing[0].embeddings_content = embedding_text
             existing[0].embedding_model_id = embedding_model_id
             existing[0].embedding_dim = embedding_dim
@@ -2571,14 +2597,20 @@ def find_most_similar_embedding_id(
         """Top-1 cosine match via pgvector ``<=>`` (Postgres only).
 
         Returns ``(object_id, score)`` or ``(None, None)`` below threshold /
-        when no rows match.
+        when no rows match. Refuses SQLite / missing ``embedding_vec`` with
+        ``SystemExit`` — callers that need a soft fallback must gate on
+        ``can_use_pgvector_similarity()`` first and use sklearn instead.
         """
         from application.database.pgvector_utils import (
+            fail_pgvector_unavailable,
             most_similar_id_sql,
             to_pgvector_literal,
         )
         from sqlalchemy import text as sql_text
 
+        if not self.can_use_pgvector_similarity():
+            fail_pgvector_unavailable(context="find_most_similar_embedding_id")
+
         sql = most_similar_id_sql(id_column)
         bind = self.session.get_bind()
         params = {
@@ -2599,6 +2631,8 @@ def find_most_similar_embedding_id(
             if score < similarity_threshold:
                 return None, None
             return str(row.object_id), score
+        except SystemExit:
+            raise
         except Exception as exc:
             # Match prior sklearn-path resilience: chat/import should degrade
             # to "no match" rather than 500 on a transient driver/query error.

application/database/pgvector_utils.py

@@ -3,8 +3,8 @@
 Do **not** re-embed the full corpus on Heroku/production dynos — likely
 OOM-killed. Re-embedding (fix missing / wrong-dim rows via the LLM) must run
 only where there is enough RAM (local or a high-resource worker), then
-sync/backfill vectors. Alembic on prod only: enable extension, add column,
-copy CSV→vector for matching dims.
+sync/backfill vectors. Alembic on prod: enable extension, add column, copy
+CSV→vector for matching dims, then DROP the legacy CSV ``embeddings`` column.
 
 HNSW/IVFFlat indexes are deferred (Essential-1 capacity); see migration
 ``c7d8e9f0a1b2`` docstring for the follow-up plan.
@@ -21,14 +21,123 @@
 HEROKU_REEMBED_WARNING = (
     "Do not re-embed the full corpus on Heroku/production dynos — likely "
     "OOM-killed. Re-embedding must run only on a machine with enough RAM, "
-    "then sync/backfill vectors. Alembic on prod only copies CSV→vector."
+    "then sync/backfill vectors. Alembic on prod only copies CSV→vector "
+    "then drops the CSV column."
+)
+
+PGVECTOR_UNAVAILABLE_EXIT_MSG = (
+    "pgvector embeddings are required but unavailable on this database. "
+    "Need Postgres with embeddings.embedding_vec (Alembic c7d8e9f0a1b2). "
+    "SQLite cannot serve pgvector similarity (``<=>``). For Module C / "
+    "Librarian on SQLite or CI, set CRE_LIBRARIAN_RETRIEVER_BACKEND=in_memory "
+    "(reads ``embedding_vec`` Text literals into an in-RAM hub). For "
+    "Postgres-side similarity, use local Postgres (make docker-postgres) or "
+    "a remote Postgres URL with the vector extension."
+)
+
+EMBEDDING_VEC_REQUIRED_EXIT_MSG = (
+    "embeddings.embedding_vec is required; the legacy CSV ``embeddings`` "
+    "column is no longer a supported store. Postgres: run Alembic upgrade "
+    "to c7d8e9f0a1b2. SQLite caches: rewrite with "
+    "`python scripts/rewrite_sqlite_embeddings_to_vec.py --db PATH` or "
+    "re-export from Postgres after the migration."
 )
 
 
 class MultiDimEmbeddingError(RuntimeError):
     """Raised when more than one embedding dimension is present in the DB."""
 
 
+class PgVectorUnavailableError(RuntimeError):
+    """Raised when a pgvector operation is requested without Postgres+vector."""
+
+
+def _exit_with_message(msg: str) -> None:
+    import logging
+
+    logging.getLogger(__name__).error(msg)
+    raise SystemExit(msg)
+
+
+def fail_pgvector_unavailable(*, context: str = "") -> None:
+    """Log and exit — callers must not silently fall back to CSV/sklearn.
+
+    Intentionally raises ``SystemExit`` so CLI/import paths stop with a clear
+    message when ``CRE_LIBRARIAN_RETRIEVER_BACKEND=pgvector`` (or other
+    pgvector-only code) runs against SQLite / pre-migration Postgres.
+    """
+    detail = f" ({context})" if context else ""
+    _exit_with_message(PGVECTOR_UNAVAILABLE_EXIT_MSG + detail)
+
+
+def fail_embedding_vec_required(*, context: str = "") -> None:
+    """Exit when the vector store column is missing (legacy CSV-only schema)."""
+    detail = f" ({context})" if context else ""
+    _exit_with_message(EMBEDDING_VEC_REQUIRED_EXIT_MSG + detail)
+
+
+def connection_dialect_name(connection: Any) -> str:
+    """Best-effort dialect name from a SQLAlchemy Connection/Engine/Session bind."""
+    dialect = getattr(connection, "dialect", None)
+    if dialect is not None:
+        return getattr(dialect, "name", "") or ""
+    bind = getattr(connection, "get_bind", None)
+    if callable(bind):
+        return connection_dialect_name(bind())
+    engine = getattr(connection, "engine", None)
+    if engine is not None:
+        return connection_dialect_name(engine)
+    return ""
+
+
+def require_pgvector_connection(connection: Any, *, context: str = "") -> None:
+    """Refuse SQLite (and unknown non-Postgres) when loading pgvector embeddings."""
+    dialect = connection_dialect_name(connection)
+    if dialect == "sqlite":
+        fail_pgvector_unavailable(context=context or "sqlite connection")
+    # Hermetic test fakes often omit dialect — allow those through.
+    if dialect and dialect != "postgresql":
+        fail_pgvector_unavailable(context=context or f"unsupported dialect {dialect!r}")
+
+
+def _embeddings_table_columns(connection: Any) -> Optional[Set[str]]:
+    """Return column names for ``embeddings``, or None when the table is absent."""
+    dialect = connection_dialect_name(connection)
+    if dialect == "sqlite":
+        rows = _execute(connection, text("PRAGMA table_info(embeddings)")).fetchall()
+        if not rows:
+            return None
+        return {str(r[1]) for r in rows}
+    if dialect != "postgresql":
+        return None
+    rows = _execute(
+        connection,
+        text(
+            "SELECT column_name FROM information_schema.columns "
+            "WHERE table_name = 'embeddings'"
+        ),
+    ).fetchall()
+    if not rows:
+        return None
+    return {str(r[0]) for r in rows}
+
+
+def require_embedding_vec_store(connection: Any, *, context: str = "") -> None:
+    """Refuse legacy CSV-only schemas; require ``embedding_vec``.
+
+    No-ops for hermetic fakes (empty dialect) and when the ``embeddings``
+    table does not exist yet (caller will hit a normal SQLAlchemy error).
+    """
+    dialect = connection_dialect_name(connection)
+    if not dialect:
+        return
+    cols = _embeddings_table_columns(connection)
+    if cols is None:
+        return
+    if "embedding_vec" not in cols:
+        fail_embedding_vec_required(context=context or f"{dialect} embeddings schema")
+
+
 def _execute(connection: Any, statement: Any, params: Optional[dict] = None):
     """Run ``statement`` on a Connection/Session, or open one from an Engine.
 
@@ -53,6 +162,23 @@ def csv_embeddings_to_literal(csv: str) -> str:
     return f"[{(csv or '').strip()}]"
 
 
+def parse_stored_embedding_vec(value: Any) -> list[float]:
+    """Parse a stored ``embedding_vec`` (pgvector / Text literal) to floats.
+
+    Accepts ``[1.0, 2.0]``, ``1.0,2.0``, or a sequence of numbers.
+    """
+    if value is None:
+        return []
+    if isinstance(value, (list, tuple)):
+        return [float(x) for x in value]
+    s = str(value).strip()
+    if not s:
+        return []
+    if s.startswith("[") and s.endswith("]"):
+        s = s[1:-1]
+    return [float(part.strip()) for part in s.split(",") if part.strip() != ""]
+
+
 def parse_csv_embedding_dim(csv: str) -> int:
     """Return the number of floats implied by a CSV embeddings string."""
     parts = [p for p in (csv or "").split(",") if p.strip() != ""]

application/tests/db_test.py

@@ -2390,8 +2390,8 @@ def test_add_embedding_persists_embedding_contract_metadata(self):
         finally:
             os.environ.pop("CRE_EMBED_MODEL", None)
 
-    def test_add_embedding_dual_writes_embedding_vec_literal(self):
-        """CSV + embedding_vec literal are persisted together (#977)."""
+    def test_add_embedding_writes_embedding_vec_literal(self):
+        """Vectors are stored only in embedding_vec (#977 option C)."""
         from application.database.pgvector_utils import to_pgvector_literal
 
         dbsa = db.Node(
@@ -2408,13 +2408,17 @@ def test_add_embedding_dual_writes_embedding_vec_literal(self):
             db_object=dbsa,
             doctype=defs.Credoctypes.Standard.value,
             embeddings=vec,
-            embedding_text="dual-write",
+            embedding_text="vec-only",
         )
         row = self.collection.get_embedding(dbsa.id)[0]
-        self.assertEqual(row.embeddings, "0.1,0.2,0.3")
-        self.assertEqual(row.embeddings_content, "dual-write")
+        self.assertNotIn("embeddings", type(row).__table__.c)
+        self.assertEqual(row.embeddings_content, "vec-only")
         self.assertEqual(row.embedding_vec, to_pgvector_literal(vec))
         self.assertEqual(row.embedding_dim, 3)
+        by_type = self.collection.get_embeddings_by_doc_type(
+            defs.Credoctypes.Standard.value
+        )
+        self.assertEqual(by_type[dbsa.id], vec)
 
     def test_assert_embedding_contract_fails_on_mixed_dimensions(self):
         n1 = db.Node(