feat(v2): Phase 3 — LanceDB index

docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md

@@ -374,3 +374,54 @@ retirements; job-queue offload of index sync; RAG/chat rewrite.
   gains it when a use case (and an embedding target) exists.
 - **`metadata` filtering in `:search`.** Stored as a JSON string column in Lance for
   now; promote to typed/filterable columns if a filtering use case appears.
+
+## 16. Resolved during Phase 3 implementation (2026-07-07)
+
+Phase 3 built the LanceDB index engine per §15 with no design reversals. As-built
+outcomes and the deviations added during implementation/review (roborev jobs 80–98):
+
+- **As-built package layout.** `index/` = `mapping.py` (pure Arrow/row helpers, no
+  LanceDB/DB import), `base.py` (`IndexEngine` ABC + `IndexFilter`/`IndexSearchHit`/
+  `IndexSearchResult`, deliberately *not* the v1 `SearchEngine` ABC), `lancedb_engine.py`
+  (`LanceIndexEngine` on the async LanceDB API), `__init__.py` (`get_index_engine` DI
+  provider mirroring `get_search_engine`). Sync glue lives in
+  `contexts/v2/index_sync.py`; reindex CLI in `cli/index/`. v1 `search_engine/` untouched
+  (verified: branch diff is empty over that path).
+- **Best-effort semantics as-built.** `sync_schema_table` (publish), `sync_upserted_records`
+  (bulk-upsert), `sync_deleted_records` (delete) each `try/except Exception` → log WARNING
+  with schema + record ids → swallow; the API request still 200s/204s (tested by forcing
+  the real engine to raise and asserting 200). Only `:rebuild-index`, `rebuild_schema_index`,
+  and the reindex CLI surface errors.
+- **Search hydration as-built.** `POST /schemas/{id}/records:search` takes
+  `{text, filters, offset, limit}`, consults Lance for `record_id`s + scores, then fetches
+  `V2Record` rows from Postgres by `id IN (...)` scoped to the schema and **re-orders to
+  Lance's hit order**, skipping ids missing from PG (stale-index tolerance). `total` is the
+  engine's match total (FTS totals materialize the match set, saturating at a 10k ceiling,
+  since `count_rows` cannot evaluate an FTS predicate).
+- **SQL-safety hardening (beyond the plan draft).** Filter column identifiers are validated
+  against an allow-list built from the live Lance schema ∪ system fields (Datafusion has no
+  parameterized identifiers); values go through a typed literal builder with `'`→`''`
+  escaping. `eq` + `None` → `IS NULL`; `op="in"` with a scalar is rejected twice —
+  a `RecordFilter` pydantic validator → **422** at the API boundary, and a `TypeError`
+  guard in the engine for direct (CLI/SDK) callers.
+- **Column-union determinism.** `table_columns` orders versions `version ASC` so
+  `union_columns` first-wins keeps the earliest version's dtype; rebuild pagination adds a
+  unique `id ASC` tiebreaker (equal `inserted_at` rows would otherwise skip/dup across
+  OFFSET pages). `arrow_schema_for` raises `ValueError` if a user column name collides with
+  a reserved system field.
+- **Rebuild efficiency.** `LanceIndexEngine.upsert` gained a keyword-only `optimize=True`;
+  the ABC adds a concrete no-op `optimize_table(schema_id)`. Write-time sync upserts
+  eagerly optimize (search freshness); `rebuild_schema_index` upserts every batch with
+  `optimize=False` and calls `optimize_table` once at the end — O(1) FTS rebuilds instead
+  of O(batches). `_list_tables` paginates the LanceDB `list_tables()` response to
+  exhaustion via `page_token`.
+- **LanceDB API notes (installed `lancedb>=0.34.0`).** Async API throughout
+  (`connect_async`, `merge_insert(...).when_matched_update_all().when_not_matched_insert_all()`,
+  `create_index("text", config=FTS())`, `add_columns`, `optimize`, `count_rows`). The
+  deprecated `Connection.table_names()` was replaced with `list_tables()`. FTS relevance is
+  the `_score` column.
+- **Review outcome.** Per-task spec+quality reviews and a whole-branch review passed;
+  roborev branch review (job 98) returned **no issues**. Deferred as follow-ups (non-blocking):
+  no test asserts the deferred-optimize call count; `ge`/`le` filters with a `None` value
+  produce an always-false `>= NULL` (no 422 guard on ordered ops); `lancedb_uri: str | None`
+  is always `str` at runtime and treats `""` as unset.

extralit-server/pyproject.toml

@@ -77,6 +77,7 @@ dependencies = [
     "pdf2image>=1.17.0",
     "opencv-python-headless>=4.11.0.86",
     "pandera[io]>=0.20",
+    "lancedb>=0.34.0",
 ]
 
 [project.optional-dependencies]

extralit-server/src/extralit_server/api/schemas/v2/search.py

@@ -0,0 +1,27 @@
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, model_validator
+
+from extralit_server.api.schemas.v2.records import LIST_RECORDS_LIMIT_DEFAULT, LIST_RECORDS_LIMIT_LE
+
+
+class RecordFilter(BaseModel):
+    column: str
+    op: Literal["eq", "in", "ge", "le"]
+    value: Any
+
+    @model_validator(mode="after")
+    def _validate_in_value(self) -> "RecordFilter":
+        if self.op == "in" and (isinstance(self.value, (str, bytes)) or not hasattr(self.value, "__iter__")):
+            raise ValueError(
+                f"Filter op='in' requires a list of values, got {type(self.value).__name__!r}."
+                ' Pass a JSON array, e.g. {"op": "in", "value": [1, 2, 3]}.'
+            )
+        return self
+
+
+class RecordSearchQuery(BaseModel):
+    text: str | None = None
+    filters: list[RecordFilter] = Field(default_factory=list)
+    offset: int = Field(default=0, ge=0)
+    limit: int = Field(default=LIST_RECORDS_LIMIT_DEFAULT, ge=1, le=LIST_RECORDS_LIMIT_LE)

extralit-server/src/extralit_server/api/v2/records.py

@@ -2,6 +2,7 @@
 from uuid import UUID
 
 from fastapi import APIRouter, Depends, Query, Security, status
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from extralit_server.api.policies.v1 import SchemaPolicy, authorize
@@ -15,14 +16,18 @@
     ReferenceGroup,
     ReferenceView,
 )
+from extralit_server.api.schemas.v2.search import RecordSearchQuery
 from extralit_server.contexts import files as files_ctx
+from extralit_server.contexts.v2 import index_sync
 from extralit_server.contexts.v2 import records as records_ctx
 from extralit_server.contexts.v2 import schemas as schemas_ctx
 from extralit_server.database import get_async_db
 from extralit_server.enums import V2RecordStatus
 from extralit_server.errors.future import NotFoundError, UnprocessableEntityError
+from extralit_server.index import get_index_engine
+from extralit_server.index.base import IndexEngine, IndexFilter
 from extralit_server.models import User, Workspace
-from extralit_server.models.v2 import Schema
+from extralit_server.models.v2 import Schema, V2Record
 from extralit_server.security import auth
 from extralit_server.utils import parse_uuids
 
@@ -44,15 +49,73 @@ async def bulk_upsert_schema_records(
     payload: RecordsBulkUpsert,
     db: Annotated[AsyncSession, Depends(get_async_db)],
     s3_client=Depends(files_ctx.get_s3_client),
+    index_engine: Annotated[IndexEngine, Depends(get_index_engine)],
     current_user: Annotated[User, Security(auth.get_current_user)],
 ):
     schema = await _get_schema_or_404(db, schema_id)
     await authorize(current_user, SchemaPolicy.upsert_records(schema))
     workspace = await Workspace.get_or_raise(db, schema.workspace_id)
     records = await records_ctx.bulk_upsert_records(db, s3_client, schema, items=payload.items, bucket=workspace.name)
+    await index_sync.sync_upserted_records(index_engine, db, schema, records)
     return Records(items=records, total=len(records))
 
 
+@router.post("/schemas/{schema_id}/records:search", response_model=Records)
+async def search_schema_records(
+    *,
+    schema_id: UUID,
+    payload: RecordSearchQuery,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    index_engine: Annotated[IndexEngine, Depends(get_index_engine)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    """Full-text (BM25) + scalar-filter search over a schema's records.
+
+    Lance supplies matching record ids and scores; payloads are hydrated from Postgres
+    (the source of truth) and returned in the engine's hit order. `total` is the engine's
+    total match count, which may exceed the returned page.
+    """
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.list_records(schema))
+
+    filters = [IndexFilter(column=f.column, op=f.op, value=f.value) for f in payload.filters]
+    result = await index_engine.search(
+        schema.id, text=payload.text, filters=filters, offset=payload.offset, limit=payload.limit
+    )
+    if not result.hits:
+        return Records(items=[], total=result.total)
+
+    hit_ids = [hit.record_id for hit in result.hits]
+    rows = (
+        (await db.execute(select(V2Record).where(V2Record.id.in_(hit_ids), V2Record.schema_id == schema.id)))
+        .scalars()
+        .all()
+    )
+    by_id = {row.id: row for row in rows}
+    ordered = [by_id[rid] for rid in hit_ids if rid in by_id]  # preserve Lance order; skip PG-missing (stale index)
+    return Records(items=[RecordRead.model_validate(r) for r in ordered], total=result.total)
+
+
+@router.post("/schemas/{schema_id}:rebuild-index", response_model=dict[str, int])
+async def rebuild_schema_index(
+    *,
+    schema_id: UUID,
+    db: Annotated[AsyncSession, Depends(get_async_db)],
+    index_engine: Annotated[IndexEngine, Depends(get_index_engine)],
+    current_user: Annotated[User, Security(auth.get_current_user)],
+):
+    """Drop and repopulate the schema's Lance table from Postgres (the recovery path).
+
+    Unlike the write-time sync hooks, this surfaces engine errors to the caller — the
+    operator explicitly asked to rebuild. For large schemas the rebuild may take tens of
+    seconds; consider running as a background job (via the CLI) if timeouts are a concern.
+    """
+    schema = await _get_schema_or_404(db, schema_id)
+    await authorize(current_user, SchemaPolicy.upsert_records(schema))
+    indexed = await index_sync.rebuild_schema_index(index_engine, db, schema)
+    return {"indexed": indexed}
+
+
 @router.get("/schemas/{schema_id}/records", response_model=Records)
 async def list_schema_records(
     *,
@@ -77,6 +140,7 @@ async def delete_schema_records(
     *,
     schema_id: UUID,
     db: Annotated[AsyncSession, Depends(get_async_db)],
+    index_engine: Annotated[IndexEngine, Depends(get_index_engine)],
     current_user: Annotated[User, Security(auth.get_current_user)],
     ids: Annotated[str, Query(description="Comma-separated record ids to delete")],
 ):
@@ -90,6 +154,7 @@ async def delete_schema_records(
     if len(record_ids) > DELETE_RECORDS_LIMIT:
         raise UnprocessableEntityError(f"Cannot delete more than {DELETE_RECORDS_LIMIT} records at once")
     await records_ctx.delete_records(db, schema, record_ids)
+    await index_sync.sync_deleted_records(index_engine, schema, record_ids)
 
 
 # `:path` converter: references are free-form join keys and DOIs contain slashes

extralit-server/src/extralit_server/api/v2/schemas.py

@@ -14,9 +14,12 @@
     SchemaVersionRead,
 )
 from extralit_server.contexts import files as files_ctx
+from extralit_server.contexts.v2 import index_sync
 from extralit_server.contexts.v2 import schemas as schemas_ctx
 from extralit_server.database import get_async_db
 from extralit_server.errors.future import NotFoundError
+from extralit_server.index import get_index_engine
+from extralit_server.index.base import IndexEngine
 from extralit_server.models import User, Workspace
 from extralit_server.models.v2 import Schema, SchemaVersion
 from extralit_server.security import auth
@@ -109,12 +112,13 @@ async def publish_schema_version(
     payload: SchemaVersionCreate,
     db: Annotated[AsyncSession, Depends(get_async_db)],
     s3_client=Depends(files_ctx.get_s3_client),
+    index_engine: Annotated[IndexEngine, Depends(get_index_engine)],
     current_user: Annotated[User, Security(auth.get_current_user)],
 ):
     schema = await _get_schema_or_404(db, schema_id)
     await authorize(current_user, SchemaPolicy.publish(schema))
     workspace = await Workspace.get_or_raise(db, schema.workspace_id)
-    return await schemas_ctx.publish_version(
+    version = await schemas_ctx.publish_version(
         db,
         s3_client,
         schema,
@@ -123,6 +127,9 @@ async def publish_schema_version(
         review_widgets=payload.review_widgets,
         created_by=current_user.id,
     )
+    # Best-effort: ensure/evolve the Lance table to the new column superset.
+    await index_sync.sync_schema_table(index_engine, db, schema)
+    return version
 
 
 @router.get("/schemas/{schema_id}/versions", response_model=list[SchemaVersionRead])

extralit-server/src/extralit_server/cli/__init__.py

@@ -1,13 +1,15 @@
 import typer
 
 from .database import app as database_app
+from .index import app as index_app
 from .search_engine import app as search_engine_app
 from .start import start
 from .worker import worker
 
 app = typer.Typer(help="Commands for Extralit server management", no_args_is_help=True)
 
 app.add_typer(database_app, name="database")
+app.add_typer(index_app, name="index")
 app.add_typer(search_engine_app, name="search-engine")
 app.command(name="worker", help="Starts rq workers")(worker)
 app.command(name="start", help="Starts the Extralit server")(start)

extralit-server/src/extralit_server/cli/index/__init__.py

@@ -0,0 +1,3 @@
+from .__main__ import app
+
+__all__ = ["app"]

extralit-server/src/extralit_server/cli/index/__main__.py

@@ -0,0 +1,11 @@
+from typer import Typer
+
+from .reindex import list_tables, reindex
+
+app = Typer(help="Commands for the Extralit v2 LanceDB index.", no_args_is_help=True)
+
+app.command(name="list", help="List existing LanceDB index tables.")(list_tables)
+app.command(name="reindex", help="Rebuild v2 schema index tables from Postgres.")(reindex)
+
+if __name__ == "__main__":
+    app()

extralit-server/src/extralit_server/cli/index/reindex.py

@@ -0,0 +1,65 @@
+"""v2 index reindex CLI — the recovery path for the derived LanceDB index.
+
+A lean twin of `cli/search_engine/reindex.py`: iterate schemas, and for each drop and
+repopulate its Lance table from Postgres via `rebuild_schema_index`.
+"""
+
+import asyncio
+from typing import Optional
+from uuid import UUID
+
+import typer
+from rich.progress import Progress
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from extralit_server.cli.rich import echo_in_panel
+from extralit_server.contexts.v2.index_sync import rebuild_schema_index
+from extralit_server.database import AsyncSessionLocal
+from extralit_server.index import get_index_engine
+from extralit_server.index.base import IndexEngine
+from extralit_server.models.v2 import Schema
+
+
+class Reindexer:
+    @classmethod
+    async def reindex_schema(cls, db: AsyncSession, engine: IndexEngine, schema_id: UUID) -> int:
+        schema = (await db.execute(select(Schema).filter_by(id=schema_id))).scalar_one()
+        return await rebuild_schema_index(engine, db, schema)
+
+    @classmethod
+    async def reindex_all(cls, db: AsyncSession, engine: IndexEngine) -> int:
+        schemas = (await db.execute(select(Schema).order_by(Schema.inserted_at.asc()))).scalars().all()
+        for schema in schemas:
+            await rebuild_schema_index(engine, db, schema)
+        return len(schemas)
+
+
+async def _reindex(schema_id: Optional[UUID] = None) -> None:
+    async with AsyncSessionLocal() as db:
+        async for engine in get_index_engine():
+            with Progress() as progress:
+                if schema_id is not None:
+                    task = progress.add_task(f"reindexing schema {schema_id}...", total=1)
+                    indexed = await Reindexer.reindex_schema(db, engine, schema_id)
+                    progress.advance(task)
+                    echo_in_panel(f"Reindexed {indexed} records.", title="Done", title_align="left")
+                else:
+                    schemas = await Reindexer.reindex_all(db, engine)
+                    echo_in_panel(f"Reindexed {schemas} schema table(s).", title="Done", title_align="left")
+
+
+async def _list_tables() -> None:
+    async for engine in get_index_engine():
+        for name in await engine.table_names():
+            typer.echo(name)
+
+
+def reindex(
+    schema_id: Optional[UUID] = typer.Option(None, help="The id of a single schema to reindex"),
+) -> None:
+    asyncio.run(_reindex(schema_id))
+
+
+def list_tables() -> None:
+    asyncio.run(_list_tables())