diff --git a/docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md b/docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md
index 0198fc0a7..1d209c82a 100644
--- a/docs/superpowers/specs/2026-06-27-schema-centric-data-model-design.md
+++ b/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.
diff --git a/extralit-server/pyproject.toml b/extralit-server/pyproject.toml
index a8a8d765b..349b21f90 100644
--- a/extralit-server/pyproject.toml
+++ b/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]
diff --git a/extralit-server/src/extralit_server/api/schemas/v2/search.py b/extralit-server/src/extralit_server/api/schemas/v2/search.py
new file mode 100644
index 000000000..bd8a93d7e
--- /dev/null
+++ b/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)
diff --git a/extralit-server/src/extralit_server/api/v2/records.py b/extralit-server/src/extralit_server/api/v2/records.py
index 87e573600..f603499b7 100644
--- a/extralit-server/src/extralit_server/api/v2/records.py
+++ b/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
diff --git a/extralit-server/src/extralit_server/api/v2/schemas.py b/extralit-server/src/extralit_server/api/v2/schemas.py
index a2a291484..9fa8a7607 100644
--- a/extralit-server/src/extralit_server/api/v2/schemas.py
+++ b/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])
diff --git a/extralit-server/src/extralit_server/cli/__init__.py b/extralit-server/src/extralit_server/cli/__init__.py
index 1fe56523b..54ecaa73a 100644
--- a/extralit-server/src/extralit_server/cli/__init__.py
+++ b/extralit-server/src/extralit_server/cli/__init__.py
@@ -1,6 +1,7 @@
 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
@@ -8,6 +9,7 @@
 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)
diff --git a/extralit-server/src/extralit_server/cli/index/__init__.py b/extralit-server/src/extralit_server/cli/index/__init__.py
new file mode 100644
index 000000000..5125a2f7f
--- /dev/null
+++ b/extralit-server/src/extralit_server/cli/index/__init__.py
@@ -0,0 +1,3 @@
+from .__main__ import app
+
+__all__ = ["app"]
diff --git a/extralit-server/src/extralit_server/cli/index/__main__.py b/extralit-server/src/extralit_server/cli/index/__main__.py
new file mode 100644
index 000000000..01a40f390
--- /dev/null
+++ b/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()
diff --git a/extralit-server/src/extralit_server/cli/index/reindex.py b/extralit-server/src/extralit_server/cli/index/reindex.py
new file mode 100644
index 000000000..e831f3bdd
--- /dev/null
+++ b/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())
diff --git a/extralit-server/src/extralit_server/contexts/v2/index_sync.py b/extralit-server/src/extralit_server/contexts/v2/index_sync.py
new file mode 100644
index 000000000..387d748e9
--- /dev/null
+++ b/extralit-server/src/extralit_server/contexts/v2/index_sync.py
@@ -0,0 +1,112 @@
+"""Best-effort synchronization between Postgres (truth) and the LanceDB index.
+
+Sync hooks mirror v1's shape (ensure on publish, upsert after record commit, delete on
+delete) but never fail the caller: any engine error is logged and swallowed, and the
+`:rebuild-index` endpoint / reindex CLI is the recovery path (spec §15). Only
+`rebuild_schema_index` raises, since the caller explicitly asked to rebuild.
+"""
+
+import logging
+from collections.abc import Iterable
+from typing import Any
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from extralit_server.index.base import IndexEngine
+from extralit_server.index.mapping import record_to_row, union_columns
+from extralit_server.models.v2 import Schema, SchemaVersion, V2Record
+
+_LOGGER = logging.getLogger("extralit_server.index")
+
+
+async def table_columns(db: AsyncSession, schema: Schema) -> list[dict[str, Any]]:
+    """The Lance table's column superset: union of every version's columns_cache.
+
+    Ordered by version ascending so that the earliest-version dtype wins when
+    the same column name appears in multiple versions with different types — making
+    the result stable and consistent across successive calls.
+    """
+    caches = (
+        (
+            await db.execute(
+                select(SchemaVersion.columns_cache)
+                .where(SchemaVersion.schema_id == schema.id)
+                .order_by(SchemaVersion.version.asc())
+            )
+        )
+        .scalars()
+        .all()
+    )
+    return union_columns([c or [] for c in caches])
+
+
+async def sync_schema_table(engine: IndexEngine, db: AsyncSession, schema: Schema) -> None:
+    try:
+        columns = await table_columns(db, schema)
+        await engine.ensure_table(schema.id, columns)
+    except Exception as exc:  # best-effort; truth is in Postgres
+        _LOGGER.warning("Index ensure_table failed for schema %s: %s", schema.id, exc)
+
+
+async def sync_upserted_records(engine: IndexEngine, db: AsyncSession, schema: Schema, records: list[V2Record]) -> None:
+    if not records:
+        return
+    try:
+        columns = await table_columns(db, schema)
+        await engine.ensure_table(schema.id, columns)
+        rows = [record_to_row(record, columns) for record in records]
+        await engine.upsert(schema.id, rows, columns)
+    except Exception as exc:
+        record_ids = [str(r.id) for r in records]
+        _LOGGER.warning("Index upsert failed for schema %s records %s: %s", schema.id, record_ids, exc)
+
+
+async def sync_deleted_records(engine: IndexEngine, schema: Schema, record_ids: Iterable[UUID]) -> None:
+    ids = list(record_ids)
+    if not ids:
+        return
+    try:
+        await engine.delete(schema.id, ids)
+    except Exception as exc:
+        _LOGGER.warning("Index delete failed for schema %s records %s: %s", schema.id, ids, exc)
+
+
+async def rebuild_schema_index(engine: IndexEngine, db: AsyncSession, schema: Schema, *, batch_size: int = 500) -> int:
+    """Drop and repopulate the schema's Lance table from Postgres. Raises on failure.
+
+    Upserts are batched with FTS optimization deferred to a single call at the end, so
+    the index is not rebuilt O(batches) times during a large reindex.
+    """
+    columns = await table_columns(db, schema)
+    await engine.drop_table(schema.id)
+    await engine.ensure_table(schema.id, columns)
+
+    total = 0
+    offset = 0
+    while True:
+        records = (
+            (
+                await db.execute(
+                    select(V2Record)
+                    .where(V2Record.schema_id == schema.id)
+                    .order_by(V2Record.inserted_at.asc(), V2Record.id.asc())
+                    .offset(offset)
+                    .limit(batch_size)
+                )
+            )
+            .scalars()
+            .all()
+        )
+        if not records:
+            break
+        rows = [record_to_row(record, columns) for record in records]
+        await engine.upsert(schema.id, rows, columns, optimize=False)
+        total += len(records)
+        offset += batch_size
+
+    # Single optimize pass after all batches — avoids O(batches) FTS rebuilds.
+    if total:
+        await engine.optimize_table(schema.id)
+    return total
diff --git a/extralit-server/src/extralit_server/index/__init__.py b/extralit-server/src/extralit_server/index/__init__.py
new file mode 100644
index 000000000..2dbd584b9
--- /dev/null
+++ b/extralit-server/src/extralit_server/index/__init__.py
@@ -0,0 +1,17 @@
+from collections.abc import AsyncGenerator
+
+from extralit_server.index.base import IndexEngine
+from extralit_server.index.lancedb_engine import LanceIndexEngine
+
+
+async def get_index_engine() -> AsyncGenerator[IndexEngine, None]:
+    """FastAPI dependency: yield a v2 index engine, closing it afterwards.
+
+    Mirrors `search_engine.get_search_engine`. The engine is currently always
+    LanceIndexEngine; a registry can be added if a second backend appears.
+    """
+    engine = await LanceIndexEngine.new_instance()
+    try:
+        yield engine
+    finally:
+        await engine.close()
diff --git a/extralit-server/src/extralit_server/index/base.py b/extralit-server/src/extralit_server/index/base.py
new file mode 100644
index 000000000..f2ef9a76d
--- /dev/null
+++ b/extralit-server/src/extralit_server/index/base.py
@@ -0,0 +1,89 @@
+"""v2 index engine interface — a small, v2-shaped abstraction over the physical index.
+
+Deliberately NOT the v1 `search_engine.base.SearchEngine` ABC: that one is typed on v1
+models (Dataset, MetadataProperty, Response) and stays untouched until Phase 6. This
+engine speaks schema ids, column caches, and plain row dicts, so it never imports v1.
+"""
+
+import dataclasses
+from abc import ABC, abstractmethod
+from collections.abc import Iterable
+from typing import Any, Literal
+from uuid import UUID
+
+from pydantic import BaseModel
+
+
+@dataclasses.dataclass
+class IndexFilter:
+    """A single scalar filter clause against a schema column or system field."""
+
+    column: str
+    op: Literal["eq", "in", "ge", "le"]
+    value: Any
+
+
+class IndexSearchHit(BaseModel):
+    record_id: UUID
+    score: float | None = None
+
+
+class IndexSearchResult(BaseModel):
+    hits: list[IndexSearchHit]
+    total: int = 0
+
+
+class IndexEngine(ABC):
+    """Physical index over derived record rows. Postgres remains the source of truth."""
+
+    @classmethod
+    @abstractmethod
+    async def new_instance(cls) -> "IndexEngine": ...
+
+    @abstractmethod
+    async def close(self) -> None: ...
+
+    @abstractmethod
+    async def ensure_table(self, schema_id: UUID, columns: list[dict[str, Any]]) -> None:
+        """Create the schema's table if absent, else evolve it to the column superset."""
+
+    @abstractmethod
+    async def drop_table(self, schema_id: UUID) -> None: ...
+
+    @abstractmethod
+    async def upsert(
+        self,
+        schema_id: UUID,
+        rows: list[dict[str, Any]],
+        columns: list[dict[str, Any]],
+        *,
+        optimize: bool = True,
+    ) -> None:
+        """Merge rows into the table keyed on `record_id` (update-or-insert).
+
+        Pass ``optimize=False`` when batching many upserts (e.g. during a rebuild) and
+        call :meth:`optimize_table` once afterwards to fold all rows into the FTS index.
+        """
+
+    async def optimize_table(self, schema_id: UUID) -> None:
+        """Compact and update the FTS index after a bulk rebuild.
+
+        Default no-op — concrete engines override when the backend supports it.
+        """
+
+    @abstractmethod
+    async def delete(self, schema_id: UUID, record_ids: Iterable[UUID]) -> None: ...
+
+    @abstractmethod
+    async def search(
+        self,
+        schema_id: UUID,
+        *,
+        text: str | None = None,
+        filters: list[IndexFilter] | None = None,
+        offset: int = 0,
+        limit: int = 50,
+    ) -> IndexSearchResult: ...
+
+    @abstractmethod
+    async def table_names(self) -> list[str]: ...
diff --git a/extralit-server/src/extralit_server/index/lancedb_engine.py b/extralit-server/src/extralit_server/index/lancedb_engine.py
new file mode 100644
index 000000000..df70e69e4
--- /dev/null
+++ b/extralit-server/src/extralit_server/index/lancedb_engine.py
@@ -0,0 +1,220 @@
+"""LanceDB-backed IndexEngine: one table per schema, BM25 full-text + scalar filtering.
+
+Uses the async LanceDB API. The connection is opened lazily on first use so importing
+the module (and constructing the engine) never touches the filesystem/URI.
+"""
+
+from collections.abc import Iterable
+from typing import Any
+from uuid import UUID
+
+import lancedb
+import pyarrow as pa
+from lancedb.index import FTS
+
+from extralit_server.index.base import IndexEngine, IndexFilter, IndexSearchHit, IndexSearchResult
+from extralit_server.index.mapping import SYSTEM_FIELDS, arrow_schema_for, arrow_type_for, table_name_for
+from extralit_server.settings import settings
+
+# Arrow type -> LanceDB SQL type name, for `add_columns` cast expressions during evolution.
+# Must cover every Arrow type `mapping.arrow_type_for` can produce, or an evolved column
+# would silently get a different type than the same column at create-table time
+# (see tests/integration/index/test_lancedb_engine.py::test_sql_type_covers_every_mapped_arrow_type).
+_SQL_TYPE_BY_ARROW = {
+    pa.large_string(): "string",
+    pa.int64(): "bigint",
+    pa.int32(): "int",
+    pa.float64(): "double",
+    pa.float32(): "float",
+    pa.bool_(): "boolean",
+    pa.timestamp("ns"): "timestamp",
+}
+
+
+def _sql_type_for(dtype: str) -> str:
+    return _SQL_TYPE_BY_ARROW.get(arrow_type_for(dtype), "string")
+
+
+# Exact FTS totals are computed by materializing the match set; beyond this many matches
+# the reported total saturates at the ceiling (extraction tables are far smaller today).
+_FTS_TOTAL_CEILING = 10_000
+
+
+def _validate_column(column: str, allowed: set[str]) -> None:
+    """Reject column identifiers that are not in the known-safe set.
+
+    This prevents SQL injection via crafted column names since Datafusion does not
+    support parameterised identifiers — only values can be safely escaped.
+    """
+    if column not in allowed:
+        raise ValueError(f"Unknown or disallowed filter column {column!r}. Allowed columns: {sorted(allowed)}")
+
+
+def _sql_literal(value: Any) -> str:
+    if value is None:
+        return "NULL"
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    if isinstance(value, (int, float)):
+        return str(value)
+    escaped = str(value).replace("'", "''")
+    return f"'{escaped}'"
+
+
+def _where_clause(filters: list[IndexFilter] | None, allowed_columns: set[str] | None = None) -> str | None:
+    if not filters:
+        return None
+    clauses = []
+    for f in filters:
+        if allowed_columns is not None:
+            _validate_column(f.column, allowed_columns)
+        if f.op == "eq":
+            if f.value is None:
+                clauses.append(f"{f.column} IS NULL")
+            else:
+                clauses.append(f"{f.column} = {_sql_literal(f.value)}")
+        elif f.op == "ge":
+            clauses.append(f"{f.column} >= {_sql_literal(f.value)}")
+        elif f.op == "le":
+            clauses.append(f"{f.column} <= {_sql_literal(f.value)}")
+        elif f.op == "in":
+            if isinstance(f.value, str) or not hasattr(f.value, "__iter__"):
+                raise TypeError(f"IndexFilter op='in' requires a list/tuple of values, got {type(f.value).__name__!r}")
+            values = ", ".join(_sql_literal(v) for v in f.value)
+            clauses.append(f"{f.column} IN ({values})")
+    return " AND ".join(clauses) if clauses else None
+
+
+class LanceIndexEngine(IndexEngine):
+    def __init__(self, uri: str) -> None:
+        self._uri = uri
+        self._db: Any = None
+
+    @classmethod
+    async def new_instance(cls) -> "LanceIndexEngine":
+        return cls(uri=settings.lancedb_uri)
+
+    async def _conn(self) -> Any:
+        if self._db is None:
+            self._db = await lancedb.connect_async(self._uri)
+        return self._db
+
+    async def close(self) -> None:
+        # AsyncConnection has no explicit close in the current API; drop the reference.
+        self._db = None
+
+    async def _list_tables(self) -> list[str]:
+        """Return all table names from the LanceDB connection, paginating to exhaustion."""
+        db = await self._conn()
+        names: list[str] = []
+        page_token: str | None = None
+        while True:
+            kwargs = {"page_token": page_token} if page_token else {}
+            response = await db.list_tables(**kwargs)
+            names.extend(response.tables)
+            page_token = response.page_token or None
+            if not page_token:
+                break
+        return names
+
+    async def table_names(self) -> list[str]:
+        return await self._list_tables()
+
+    async def ensure_table(self, schema_id: UUID, columns: list[dict[str, Any]]) -> None:
+        name = table_name_for(schema_id)
+        schema = arrow_schema_for(columns)
+        if name not in await self._list_tables():
+            db = await self._conn()
+            table = await db.create_table(name, schema=schema)
+            await table.create_index("text", config=FTS())
+            return
+        # Evolve: add any columns present in `columns` but missing from the live table.
+        db = await self._conn()
+        table = await db.open_table(name)
+        existing = set((await table.schema()).names)
+        to_add = {c["name"]: f"cast(NULL as {_sql_type_for(c['dtype'])})" for c in columns if c["name"] not in existing}
+        if to_add:
+            await table.add_columns(to_add)
+
+    async def drop_table(self, schema_id: UUID) -> None:
+        name = table_name_for(schema_id)
+        if name in await self._list_tables():
+            db = await self._conn()
+            await db.drop_table(name)
+
+    async def upsert(
+        self,
+        schema_id: UUID,
+        rows: list[dict[str, Any]],
+        columns: list[dict[str, Any]],
+        *,
+        optimize: bool = True,
+    ) -> None:
+        if not rows:
+            return
+        db = await self._conn()
+        table = await db.open_table(table_name_for(schema_id))
+        data = pa.Table.from_pylist(rows, schema=arrow_schema_for(columns))
+        await table.merge_insert("record_id").when_matched_update_all().when_not_matched_insert_all().execute(data)
+        if optimize:
+            await table.optimize()  # fold new rows into the FTS index
+
+    async def optimize_table(self, schema_id: UUID) -> None:
+        """Compact the table and update the FTS index. Call once after a bulk rebuild."""
+        db = await self._conn()
+        name = table_name_for(schema_id)
+        if name in await self._list_tables():
+            table = await db.open_table(name)
+            await table.optimize()
+
+    async def delete(self, schema_id: UUID, record_ids: Iterable[UUID]) -> None:
+        ids = [str(rid) for rid in record_ids]
+        if not ids:
+            return
+        db = await self._conn()
+        table = await db.open_table(table_name_for(schema_id))
+        values = ", ".join(_sql_literal(i) for i in ids)
+        await table.delete(f"record_id IN ({values})")
+
+    async def search(
+        self,
+        schema_id: UUID,
+        *,
+        text: str | None = None,
+        filters: list[IndexFilter] | None = None,
+        offset: int = 0,
+        limit: int = 50,
+    ) -> IndexSearchResult:
+        db = await self._conn()
+        name = table_name_for(schema_id)
+        if name not in await self._list_tables():
+            return IndexSearchResult(hits=[], total=0)
+        table = await db.open_table(name)
+        live_columns = set((await table.schema()).names)
+        allowed_columns = live_columns | set(SYSTEM_FIELDS)
+        clause = _where_clause(filters, allowed_columns)
+
+        if text:
+            # `count_rows` only evaluates scalar predicates — it cannot apply the FTS
+            # match, so the total for a text query must come from the match set itself.
+            # Materialize matches up to a ceiling and page in Python; beyond the ceiling
+            # the total saturates at _FTS_TOTAL_CEILING (documented, bounded work).
+            query = await table.search(text, query_type="fts")
+            if clause:
+                query = query.where(clause, prefilter=True)
+            arrow = await query.limit(_FTS_TOTAL_CEILING).to_arrow()
+            matches = arrow.to_pylist()
+            hits = [
+                IndexSearchHit(record_id=UUID(row["record_id"]), score=row.get("_score"))
+                for row in matches[offset : offset + limit]
+            ]
+            return IndexSearchResult(hits=hits, total=len(matches))
+
+        query = table.query()
+        if clause:
+            query = query.where(clause)
+        arrow = await query.limit(offset + limit).to_arrow()
+        rows = arrow.to_pylist()[offset:]
+        hits = [IndexSearchHit(record_id=UUID(row["record_id"]), score=None) for row in rows]
+        total = await table.count_rows(clause) if clause else await table.count_rows()
+        return IndexSearchResult(hits=hits, total=total)
diff --git a/extralit-server/src/extralit_server/index/mapping.py b/extralit-server/src/extralit_server/index/mapping.py
new file mode 100644
index 000000000..987e6dc80
--- /dev/null
+++ b/extralit-server/src/extralit_server/index/mapping.py
@@ -0,0 +1,120 @@
+"""Pure, I/O-free helpers mapping v2 schema columns and records to a Lance row layout.
+
+No LanceDB, DB, or object-store access — given a `columns_cache` (from
+`SchemaVersion.columns_cache`) and a record, build the Arrow schema and row dicts the
+index engine writes. The Lance table for a schema is the union (superset) of columns
+across its versions plus system/identity columns and a derived `text` column that
+carries the BM25 full-text index.
+"""
+
+from typing import Any
+from uuid import UUID
+
+import pyarrow as pa
+
+# Identity/system columns present in every schema's Lance table, independent of the
+# user-defined columns. `text` is the concatenated string-cell blob the FTS index covers.
+SYSTEM_FIELDS = ["record_id", "reference", "schema_version_id", "status", "external_id", "text"]
+
+# Observed pandera 0.32 / pandas 3.0 `str(column.dtype)` values -> Arrow types.
+# large_string is used for text so the FTS index has no 2GiB offset ceiling.
+_ARROW_BY_DTYPE = {
+    "string[pyarrow]": pa.large_string(),
+    "string": pa.large_string(),
+    "object": pa.large_string(),
+    "int64": pa.int64(),
+    "int32": pa.int32(),
+    "float64": pa.float64(),
+    "float32": pa.float32(),
+    "bool": pa.bool_(),
+    "boolean": pa.bool_(),
+    "datetime64[ns]": pa.timestamp("ns"),
+}
+
+# Column dtypes we treat as full-text material for the `text` blob.
+_STRING_DTYPES = {"string[pyarrow]", "string", "object"}
+
+
+def table_name_for(schema_id: UUID) -> str:
+    """Lance table name for a schema. UUID hex (no dashes) is a safe identifier."""
+    return f"schema_{schema_id.hex}"
+
+
+def arrow_type_for(dtype: str) -> pa.DataType:
+    """Map a pandera/pandas dtype string to an Arrow type; unknown -> large_string."""
+    return _ARROW_BY_DTYPE.get(dtype, pa.large_string())
+
+
+def union_columns(caches: list[list[dict[str, Any]]]) -> list[dict[str, Any]]:
+    """Union column entries across versions, first occurrence wins, order preserved."""
+    seen: dict[str, dict[str, Any]] = {}
+    for cache in caches:
+        for column in cache or []:
+            name = column["name"]
+            if name not in seen:
+                seen[name] = column
+    return list(seen.values())
+
+
+def arrow_schema_for(columns: list[dict[str, Any]]) -> pa.Schema:
+    """Build the full Arrow schema: system fields + one typed field per schema column.
+
+    All fields are nullable in Lance regardless of the Pandera `nullable` flag —
+    Postgres enforces validation; the index must hold superset rows where older-version
+    records legitimately lack newer columns.
+
+    Raises ValueError if any user column name collides with a system field, since such
+    a collision would produce a duplicate Arrow field or silently overwrite system data.
+    """
+    _system = set(SYSTEM_FIELDS)
+    collisions = [c["name"] for c in columns if c["name"] in _system]
+    if collisions:
+        raise ValueError(
+            f"Schema column name(s) collide with reserved system fields: {collisions}. Reserved names: {SYSTEM_FIELDS}"
+        )
+    fields = [
+        pa.field("record_id", pa.large_string()),
+        pa.field("reference", pa.large_string()),
+        pa.field("schema_version_id", pa.large_string()),
+        pa.field("status", pa.large_string()),
+        pa.field("external_id", pa.large_string()),
+    ]
+    for column in columns:
+        fields.append(pa.field(column["name"], arrow_type_for(column["dtype"])))
+    fields.append(pa.field("text", pa.large_string()))
+    return pa.schema(fields)
+
+
+def concat_text(fields: dict[str, Any], columns: list[dict[str, Any]]) -> str:
+    """Concatenate string-typed cells into a `col: value` blob for the FTS index."""
+    lines = []
+    for column in columns:
+        if column["dtype"] not in _STRING_DTYPES:
+            continue
+        value = fields.get(column["name"])
+        if value is None:
+            continue
+        lines.append(f"{column['name']}: {value}")
+    return "\n".join(lines)
+
+
+def record_to_row(record: Any, columns: list[dict[str, Any]]) -> dict[str, Any]:
+    """Build a Lance row dict for a record against the table's column superset.
+
+    Every field in `arrow_schema_for(columns)` is present as a key; schema columns the
+    record does not carry are filled with None (older-version rows in an evolved table).
+    Permissive extra fields in `record.fields` that are not schema columns are ignored.
+    UUID/enum system values are stringified to match the Arrow schema.
+    """
+    fields = record.fields or {}
+    row: dict[str, Any] = {
+        "record_id": str(record.id),
+        "reference": record.reference,
+        "schema_version_id": str(record.schema_version_id),
+        "status": record.status.value if hasattr(record.status, "value") else str(record.status),
+        "external_id": record.external_id,
+    }
+    for column in columns:
+        row[column["name"]] = fields.get(column["name"])
+    row["text"] = concat_text(fields, columns)
+    return row
diff --git a/extralit-server/src/extralit_server/settings.py b/extralit-server/src/extralit_server/settings.py
index 8483f2096..648b2576b 100644
--- a/extralit-server/src/extralit_server/settings.py
+++ b/extralit-server/src/extralit_server/settings.py
@@ -124,6 +124,14 @@ class Settings(BaseSettings):
     s3_secret_key: str | None = Field(default=None, description="The secret key for the S3 storage")
     s3_region: str | None = Field(default=None, description="The region for the S3 storage")
 
+    lancedb_uri: str | None = Field(
+        default=None,
+        validate_default=True,
+        description="URI for the LanceDB index store (v2). Defaults to `{home_path}/lance`. "
+        "A local path works on the compose named volume and HF-Spaces persistent storage; "
+        "an s3:// URI is accepted by lancedb.connect but unsupported/unvalidated for now.",
+    )
+
     extralit_url: str | None = Field(default=None, description="The extralit server url for LLM serving endpoint")
 
     elasticsearch: str = "http://localhost:9200"
@@ -194,6 +202,14 @@ def set_enable_telemetry(cls, enable_telemetry: bool) -> bool:
     def set_home_path_default(cls, home_path: str):
         return home_path or os.path.join(Path.home(), ".extralit")
 
+    @field_validator("lancedb_uri", mode="before")
+    @classmethod
+    def set_lancedb_uri_default(cls, lancedb_uri: str | None, info: ValidationInfo) -> str:
+        if lancedb_uri:
+            return lancedb_uri
+        home_path = info.data.get("home_path") or os.path.join(Path.home(), ".extralit")
+        return os.path.join(home_path, "lance")
+
     @field_validator("base_url")
     @classmethod
     def normalize_base_url(cls, base_url: str):
diff --git a/extralit-server/tests/integration/api/v2/test_records.py b/extralit-server/tests/integration/api/v2/test_records.py
index d5eda531e..8c7d0d9a3 100644
--- a/extralit-server/tests/integration/api/v2/test_records.py
+++ b/extralit-server/tests/integration/api/v2/test_records.py
@@ -179,3 +179,61 @@ async def test_non_member_cannot_read_records(async_client, annotator_auth_heade
 
     resp = await async_client.get(f"/api/v2/schemas/{schema.id}/records", headers=annotator_auth_header)
     assert resp.status_code == 403, resp.text
+
+
+async def test_bulk_upsert_syncs_index(async_client, owner_auth_header, db, monkeypatch):
+    from unittest.mock import AsyncMock
+
+    sync = AsyncMock()
+    monkeypatch.setattr("extralit_server.contexts.v2.index_sync.sync_upserted_records", sync)
+    _patch_fetch(monkeypatch)
+    schema, _ = await _published_schema(db)
+
+    resp = await async_client.post(
+        f"/api/v2/schemas/{schema.id}/records:bulk-upsert",
+        headers=owner_auth_header,
+        json={"items": [{"fields": {"name": "Ada", "age": 36}, "reference": "pmid:1"}]},
+    )
+    assert resp.status_code == 200, resp.text
+    sync.assert_awaited_once()
+
+
+async def test_bulk_upsert_survives_index_failure(async_client, owner_auth_header, db, monkeypatch):
+    from unittest.mock import AsyncMock
+
+    # Real best-effort path: engine raises, request must still be 200.
+    monkeypatch.setattr(
+        "extralit_server.index.lancedb_engine.LanceIndexEngine.upsert",
+        AsyncMock(side_effect=RuntimeError("lance down")),
+    )
+    monkeypatch.setattr(
+        "extralit_server.index.lancedb_engine.LanceIndexEngine.ensure_table",
+        AsyncMock(side_effect=RuntimeError("lance down")),
+    )
+    _patch_fetch(monkeypatch)
+    schema, _ = await _published_schema(db)
+
+    resp = await async_client.post(
+        f"/api/v2/schemas/{schema.id}/records:bulk-upsert",
+        headers=owner_auth_header,
+        json={"items": [{"fields": {"name": "Ada", "age": 36}, "reference": "pmid:1"}]},
+    )
+    assert resp.status_code == 200, resp.text
+
+
+async def test_delete_syncs_index(async_client, owner_auth_header, db, monkeypatch):
+    from unittest.mock import AsyncMock
+
+    sync = AsyncMock()
+    monkeypatch.setattr("extralit_server.contexts.v2.index_sync.sync_deleted_records", sync)
+    _patch_fetch(monkeypatch)
+    schema, version = await _published_schema(db)
+    from tests.factories import V2RecordFactory
+
+    record = await V2RecordFactory.create(schema=schema, version=version, fields={"name": "X"})
+    resp = await async_client.delete(
+        f"/api/v2/schemas/{schema.id}/records?ids={record.id}",
+        headers=owner_auth_header,
+    )
+    assert resp.status_code == 204, resp.text
+    sync.assert_awaited_once()
diff --git a/extralit-server/tests/integration/api/v2/test_records_search.py b/extralit-server/tests/integration/api/v2/test_records_search.py
new file mode 100644
index 000000000..437efa4ee
--- /dev/null
+++ b/extralit-server/tests/integration/api/v2/test_records_search.py
@@ -0,0 +1,116 @@
+from unittest.mock import AsyncMock
+
+import pandera.pandas as pa
+import pytest
+
+from extralit_server.index.base import IndexSearchHit, IndexSearchResult
+from tests.factories import SchemaFactory, SchemaVersionFactory, V2RecordFactory
+
+pytestmark = pytest.mark.asyncio
+
+BODY = pa.DataFrameSchema(
+    columns={"title": pa.Column(pa.String, nullable=False), "year": pa.Column(pa.Int, nullable=True)}
+).to_json()
+
+
+async def _published(db):
+    schema = await SchemaFactory.create()
+    version = await SchemaVersionFactory.create(
+        schema=schema,
+        version=1,
+        columns_cache=[
+            {"name": "title", "dtype": "string[pyarrow]", "nullable": False, "review": None},
+            {"name": "year", "dtype": "int64", "nullable": True, "review": None},
+        ],
+    )
+    await schema.update(db, current_version_id=version.id)
+    return schema, version
+
+
+async def test_search_hydrates_from_postgres_in_hit_order(async_client, owner_auth_header, db, monkeypatch):
+    schema, version = await _published(db)
+    r1 = await V2RecordFactory.create(schema=schema, version=version, fields={"title": "Deep", "year": 2016})
+    r2 = await V2RecordFactory.create(schema=schema, version=version, fields={"title": "Shallow", "year": 1999})
+
+    # Engine returns r2 then r1; response must preserve that order and hydrate real payloads.
+    fake = IndexSearchResult(hits=[IndexSearchHit(record_id=r2.id), IndexSearchHit(record_id=r1.id)], total=2)
+    monkeypatch.setattr("extralit_server.index.lancedb_engine.LanceIndexEngine.search", AsyncMock(return_value=fake))
+
+    resp = await async_client.post(
+        f"/api/v2/schemas/{schema.id}/records:search",
+        headers=owner_auth_header,
+        json={"text": "deep"},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    assert body["total"] == 2
+    assert [item["id"] for item in body["items"]] == [str(r2.id), str(r1.id)]
+    assert body["items"][1]["fields"]["title"] == "Deep"  # real PG payload, not from Lance
+
+
+async def test_search_empty_result(async_client, owner_auth_header, db, monkeypatch):
+    schema, _ = await _published(db)
+    monkeypatch.setattr(
+        "extralit_server.index.lancedb_engine.LanceIndexEngine.search",
+        AsyncMock(return_value=IndexSearchResult(hits=[], total=0)),
+    )
+    resp = await async_client.post(
+        f"/api/v2/schemas/{schema.id}/records:search",
+        headers=owner_auth_header,
+        json={"filters": [{"column": "year", "op": "ge", "value": 3000}]},
+    )
+    assert resp.status_code == 200, resp.text
+    assert resp.json() == {"items": [], "total": 0}
+
+
+async def test_search_requires_membership(async_client, annotator_auth_header, db):
+    # `annotator_auth_header` is a non-member of the schema's workspace (repo idiom for
+    # the 403 case; see test_records.py::test_non_member_cannot_read_records).
+    schema, _ = await _published(db)
+    resp = await async_client.post(
+        f"/api/v2/schemas/{schema.id}/records:search",
+        headers=annotator_auth_header,
+        json={"text": "x"},
+    )
+    assert resp.status_code == 403, resp.text
+
+
+async def test_rebuild_index_reindexes_all_records(async_client, owner_auth_header, db, monkeypatch):
+    schema, version = await _published(db)
+    await V2RecordFactory.create(schema=schema, version=version, fields={"title": "A", "year": 2001})
+    await V2RecordFactory.create(schema=schema, version=version, fields={"title": "B", "year": 2002})
+
+    calls = {}
+
+    async def fake_rebuild(engine, db_, s, *, batch_size=500):
+        calls["schema_id"] = s.id
+        return 2
+
+    monkeypatch.setattr("extralit_server.contexts.v2.index_sync.rebuild_schema_index", fake_rebuild)
+
+    resp = await async_client.post(f"/api/v2/schemas/{schema.id}:rebuild-index", headers=owner_auth_header)
+    assert resp.status_code == 200, resp.text
+    assert resp.json() == {"indexed": 2}
+    assert calls["schema_id"] == schema.id
+
+
+async def test_rebuild_index_requires_write_access(async_client, annotator_auth_header, db):
+    # Non-member of the workspace → 403 (repo idiom; see test_records.py negative-authz tests).
+    schema, _ = await _published(db)
+    resp = await async_client.post(
+        f"/api/v2/schemas/{schema.id}:rebuild-index",
+        headers=annotator_auth_header,
+    )
+    assert resp.status_code == 403, resp.text
+
+
+async def test_search_in_filter_with_scalar_value_returns_422(async_client, owner_auth_header, db):
+    # op="in" with a scalar string (not a list) must be rejected at the schema layer → 422,
+    # not a 500 from an unmapped TypeError in the engine.
+    schema, _ = await _published(db)
+    resp = await async_client.post(
+        f"/api/v2/schemas/{schema.id}/records:search",
+        headers=owner_auth_header,
+        json={"filters": [{"column": "year", "op": "in", "value": "2016"}]},
+    )
+    assert resp.status_code == 422, resp.text
diff --git a/extralit-server/tests/integration/api/v2/test_schemas.py b/extralit-server/tests/integration/api/v2/test_schemas.py
index f85a1f253..81451fc45 100644
--- a/extralit-server/tests/integration/api/v2/test_schemas.py
+++ b/extralit-server/tests/integration/api/v2/test_schemas.py
@@ -82,3 +82,43 @@ async def test_non_member_cannot_create_or_read_schema(async_client, annotator_a
         json={"name": "secret", "kind": SchemaKind.table.value, "workspace_id": str(ws.id)},
     )
     assert resp.status_code == 403, resp.text
+
+
+async def test_publish_version_creates_index_table(async_client, owner_auth_header, db, monkeypatch):
+    from datetime import datetime
+    from unittest.mock import AsyncMock
+
+    from extralit_server.contexts.files import ObjectMetadata
+
+    monkeypatch.setattr(
+        "extralit_server.contexts.v2.schemas.files_ctx.put_object",
+        AsyncMock(
+            return_value=ObjectMetadata(
+                bucket_name="b",
+                object_name="k",
+                etag="etag-1",
+                size=1,
+                last_modified=datetime(2026, 1, 1),
+                content_type="application/json",
+                version_id="ver-1",
+                metadata={},
+            )
+        ),
+    )
+
+    ensure = AsyncMock()
+    monkeypatch.setattr("extralit_server.contexts.v2.index_sync.sync_schema_table", ensure)
+
+    from tests.factories import SchemaFactory
+
+    schema = await SchemaFactory.create()
+    import pandera.pandas as pa
+
+    body = pa.DataFrameSchema(columns={"title": pa.Column(pa.String, nullable=False)}).to_json()
+    resp = await async_client.post(
+        f"/api/v2/schemas/{schema.id}/versions",
+        headers=owner_auth_header,
+        json={"body": body},
+    )
+    assert resp.status_code in (200, 201), resp.text
+    ensure.assert_awaited_once()
diff --git a/extralit-server/tests/integration/cli/__init__.py b/extralit-server/tests/integration/cli/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/cli/test_index_reindex.py b/extralit-server/tests/integration/cli/test_index_reindex.py
new file mode 100644
index 000000000..983a51f84
--- /dev/null
+++ b/extralit-server/tests/integration/cli/test_index_reindex.py
@@ -0,0 +1,33 @@
+from unittest.mock import AsyncMock
+
+import pytest
+
+from extralit_server.cli.index.reindex import Reindexer
+from tests.factories import SchemaFactory, SchemaVersionFactory
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_reindex_schema_rebuilds(db, monkeypatch):
+    schema = await SchemaFactory.create()
+    version = await SchemaVersionFactory.create(schema=schema, version=1)
+    await schema.update(db, current_version_id=version.id)
+
+    rebuild = AsyncMock(return_value=0)
+    monkeypatch.setattr("extralit_server.cli.index.reindex.rebuild_schema_index", rebuild)
+    engine = AsyncMock()
+
+    await Reindexer.reindex_schema(db, engine, schema.id)
+    rebuild.assert_awaited_once()
+
+
+async def test_reindex_all_iterates_schemas(db, monkeypatch):
+    await SchemaFactory.create()
+    await SchemaFactory.create()
+    rebuild = AsyncMock(return_value=0)
+    monkeypatch.setattr("extralit_server.cli.index.reindex.rebuild_schema_index", rebuild)
+    engine = AsyncMock()
+
+    count = await Reindexer.reindex_all(db, engine)
+    assert count >= 2
+    assert rebuild.await_count >= 2
diff --git a/extralit-server/tests/integration/contexts/v2/test_index_sync.py b/extralit-server/tests/integration/contexts/v2/test_index_sync.py
new file mode 100644
index 000000000..0fd3ebfb8
--- /dev/null
+++ b/extralit-server/tests/integration/contexts/v2/test_index_sync.py
@@ -0,0 +1,93 @@
+from unittest.mock import AsyncMock
+
+import pytest
+
+from extralit_server.contexts.v2 import index_sync
+from tests.factories import SchemaFactory, SchemaVersionFactory, V2RecordFactory
+
+pytestmark = pytest.mark.asyncio
+
+
+async def _published(db):
+    schema = await SchemaFactory.create()
+    version = await SchemaVersionFactory.create(
+        schema=schema,
+        version=1,
+        columns_cache=[{"name": "title", "dtype": "string[pyarrow]", "nullable": False, "review": None}],
+    )
+    await schema.update(db, current_version_id=version.id)
+    return schema, version
+
+
+async def test_table_columns_unions_versions(db):
+    schema, _v1 = await _published(db)
+    await SchemaVersionFactory.create(
+        schema=schema,
+        version=2,
+        columns_cache=[
+            {"name": "title", "dtype": "string[pyarrow]", "nullable": False, "review": None},
+            {"name": "year", "dtype": "int64", "nullable": True, "review": None},
+        ],
+    )
+    columns = await index_sync.table_columns(db, schema)
+    assert {c["name"] for c in columns} == {"title", "year"}
+
+
+async def test_table_columns_dtype_first_wins(db):
+    """Earliest version's dtype must win when two versions disagree on a column's type.
+
+    v1 defines `title` as string[pyarrow]; v2 redefines it as int64 and adds `year`.
+    Because versions are ordered ASC, `title` must keep the v1 dtype (string[pyarrow]).
+    """
+    schema, _v1 = await _published(db)  # v1: title=string[pyarrow]
+    await SchemaVersionFactory.create(
+        schema=schema,
+        version=2,
+        columns_cache=[
+            {"name": "title", "dtype": "int64", "nullable": True, "review": None},  # conflicting dtype
+            {"name": "year", "dtype": "int64", "nullable": True, "review": None},
+        ],
+    )
+    columns = await index_sync.table_columns(db, schema)
+    col_by_name = {c["name"]: c for c in columns}
+    # Name union is correct.
+    assert set(col_by_name) == {"title", "year"}
+    # Earliest-version dtype wins for `title` — v1's string[pyarrow] beats v2's int64.
+    assert col_by_name["title"]["dtype"] == "string[pyarrow]", (
+        f"Expected v1 dtype 'string[pyarrow]', got {col_by_name['title']['dtype']!r}"
+    )
+
+
+async def test_sync_schema_table_calls_ensure(db):
+    schema, _ = await _published(db)
+    engine = AsyncMock()
+    await index_sync.sync_schema_table(engine, db, schema)
+    engine.ensure_table.assert_awaited_once()
+
+
+async def test_sync_upserted_records_builds_rows(db):
+    schema, version = await _published(db)
+    record = await V2RecordFactory.create(schema=schema, version=version, fields={"title": "Hi"})
+    engine = AsyncMock()
+    await index_sync.sync_upserted_records(engine, db, schema, [record])
+    engine.upsert.assert_awaited_once()
+    args, _kwargs = engine.upsert.call_args
+    rows = args[1]
+    assert rows[0]["title"] == "Hi"
+
+
+async def test_sync_swallows_engine_errors(db):
+    schema, version = await _published(db)
+    record = await V2RecordFactory.create(schema=schema, version=version, fields={"title": "Hi"})
+    engine = AsyncMock()
+    engine.upsert.side_effect = RuntimeError("lance down")
+    # Must NOT raise — best-effort.
+    await index_sync.sync_upserted_records(engine, db, schema, [record])
+
+
+async def test_rebuild_raises_on_failure(db):
+    schema, _ = await _published(db)
+    engine = AsyncMock()
+    engine.drop_table.side_effect = RuntimeError("lance down")
+    with pytest.raises(RuntimeError):
+        await index_sync.rebuild_schema_index(engine, db, schema)
diff --git a/extralit-server/tests/integration/index/__init__.py b/extralit-server/tests/integration/index/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/integration/index/test_get_index_engine.py b/extralit-server/tests/integration/index/test_get_index_engine.py
new file mode 100644
index 000000000..9aeb57c2c
--- /dev/null
+++ b/extralit-server/tests/integration/index/test_get_index_engine.py
@@ -0,0 +1,14 @@
+import pytest
+
+from extralit_server.index import get_index_engine
+from extralit_server.index.lancedb_engine import LanceIndexEngine
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_get_index_engine_yields_lance_engine(monkeypatch, tmp_path):
+    monkeypatch.setattr("extralit_server.settings.settings.lancedb_uri", str(tmp_path / "lance"))
+    seen = None
+    async for engine in get_index_engine():
+        seen = engine
+    assert isinstance(seen, LanceIndexEngine)
diff --git a/extralit-server/tests/integration/index/test_lancedb_engine.py b/extralit-server/tests/integration/index/test_lancedb_engine.py
new file mode 100644
index 000000000..0d91b7f5a
--- /dev/null
+++ b/extralit-server/tests/integration/index/test_lancedb_engine.py
@@ -0,0 +1,181 @@
+from uuid import uuid4
+
+import pytest
+
+from extralit_server.index.base import IndexFilter
+from extralit_server.index.lancedb_engine import LanceIndexEngine
+from extralit_server.index.mapping import record_to_row
+
+pytestmark = pytest.mark.asyncio
+
+COLUMNS = [
+    {"name": "title", "dtype": "string[pyarrow]", "nullable": False, "review": None},
+    {"name": "year", "dtype": "int64", "nullable": True, "review": None},
+]
+
+
+class _Rec:
+    def __init__(self, title, year, reference="pmid:1", external_id=None):
+        from extralit_server.enums import V2RecordStatus
+
+        self.id = uuid4()
+        self.reference = reference
+        self.schema_version_id = uuid4()
+        self.status = V2RecordStatus.pending
+        self.external_id = external_id
+        self.fields = {"title": title, "year": year}
+
+
+@pytest.fixture
+async def engine(tmp_path):
+    eng = LanceIndexEngine(uri=str(tmp_path / "lance"))
+    yield eng
+    await eng.close()
+
+
+async def test_ensure_upsert_and_fts_search(engine):
+    sid = uuid4()
+    await engine.ensure_table(sid, COLUMNS)
+    recs = [_Rec("Deep Learning Foundations", 2016), _Rec("Shallow Ponds", 1999)]
+    await engine.upsert(sid, [record_to_row(r, COLUMNS) for r in recs], COLUMNS)
+
+    result = await engine.search(sid, text="Deep Learning", offset=0, limit=10)
+    assert result.total >= 1
+    assert recs[0].id in [h.record_id for h in result.hits]
+
+
+async def test_scalar_filter_without_text(engine):
+    sid = uuid4()
+    await engine.ensure_table(sid, COLUMNS)
+    recs = [_Rec("A", 2016), _Rec("B", 1999)]
+    await engine.upsert(sid, [record_to_row(r, COLUMNS) for r in recs], COLUMNS)
+
+    result = await engine.search(sid, filters=[IndexFilter(column="year", op="ge", value=2000)], limit=10)
+    ids = [h.record_id for h in result.hits]
+    assert recs[0].id in ids and recs[1].id not in ids
+
+
+async def test_upsert_updates_in_place(engine):
+    sid = uuid4()
+    await engine.ensure_table(sid, COLUMNS)
+    rec = _Rec("Original", 2000)
+    await engine.upsert(sid, [record_to_row(rec, COLUMNS)], COLUMNS)
+    rec.fields["title"] = "Rewritten"
+    await engine.upsert(sid, [record_to_row(rec, COLUMNS)], COLUMNS)
+
+    result = await engine.search(sid, filters=[IndexFilter(column="year", op="eq", value=2000)], limit=10)
+    assert len([h for h in result.hits if h.record_id == rec.id]) == 1
+
+
+async def test_delete_removes_rows(engine):
+    sid = uuid4()
+    await engine.ensure_table(sid, COLUMNS)
+    rec = _Rec("Doomed", 2010)
+    await engine.upsert(sid, [record_to_row(rec, COLUMNS)], COLUMNS)
+    await engine.delete(sid, [rec.id])
+
+    result = await engine.search(sid, filters=[IndexFilter(column="year", op="eq", value=2010)], limit=10)
+    assert rec.id not in [h.record_id for h in result.hits]
+
+
+async def test_ensure_table_evolves_to_superset(engine):
+    sid = uuid4()
+    await engine.ensure_table(sid, COLUMNS)
+    evolved = [*COLUMNS, {"name": "doi", "dtype": "string[pyarrow]", "nullable": True, "review": None}]
+    await engine.ensure_table(sid, evolved)  # idempotent + adds `doi`
+
+    # Confirm the new column is present in the live schema.
+    from extralit_server.index.mapping import table_name_for
+
+    db = engine._db
+    table = await db.open_table(table_name_for(sid))
+    live_names = (await table.schema()).names
+    assert "doi" in live_names
+
+    # Confirm the evolved column is usable in a filter.
+    from extralit_server.index.base import IndexFilter
+
+    result = await engine.search(sid, filters=[IndexFilter(column="doi", op="eq", value=None)], limit=10)
+    assert isinstance(result.total, int)  # no exception; doi is a valid filter column
+
+
+async def test_fts_total_counts_matches_not_table_rows(engine):
+    # `count_rows` cannot evaluate the FTS match, so `total` must come from the match
+    # set: 2 of 3 rows match "Deep"; the page respects `limit` but `total` reports 2.
+    sid = uuid4()
+    await engine.ensure_table(sid, COLUMNS)
+    recs = [_Rec("Deep Learning Foundations", 2016), _Rec("Shallow Ponds", 1999), _Rec("Deep Sea Biology", 2005)]
+    await engine.upsert(sid, [record_to_row(r, COLUMNS) for r in recs], COLUMNS)
+
+    result = await engine.search(sid, text="Deep", offset=0, limit=1)
+    assert len(result.hits) == 1
+    assert result.total == 2
+
+
+async def test_unknown_filter_column_raises(engine):
+    sid = uuid4()
+    await engine.ensure_table(sid, COLUMNS)
+    recs = [_Rec("A", 2016)]
+    await engine.upsert(sid, [record_to_row(r, COLUMNS) for r in recs], COLUMNS)
+
+    with pytest.raises(ValueError, match="disallowed filter column"):
+        await engine.search(sid, filters=[IndexFilter(column="injected) OR (1=1", op="eq", value=1)], limit=10)
+
+
+async def test_filter_op_in_rejects_scalar_string(engine):
+    sid = uuid4()
+    await engine.ensure_table(sid, COLUMNS)
+    recs = [_Rec("A", 2016)]
+    await engine.upsert(sid, [record_to_row(r, COLUMNS) for r in recs], COLUMNS)
+
+    with pytest.raises(TypeError, match="list/tuple"):
+        await engine.search(sid, filters=[IndexFilter(column="year", op="in", value="2016")], limit=10)
+
+
+async def test_filter_eq_none_matches_null_rows(engine):
+    sid = uuid4()
+    await engine.ensure_table(sid, COLUMNS)
+    # external_id is a SYSTEM_FIELDS column; rows with external_id=None should match IS NULL
+    rec = _Rec("A", 2016, external_id=None)
+    await engine.upsert(sid, [record_to_row(rec, COLUMNS)], COLUMNS)
+
+    result = await engine.search(sid, filters=[IndexFilter(column="external_id", op="eq", value=None)], limit=10)
+    assert rec.id in [h.record_id for h in result.hits]
+
+
+async def test_sql_type_covers_every_mapped_arrow_type():
+    # Create-path (`arrow_schema_for`) and evolve-path (`add_columns` cast) must agree:
+    # every Arrow type `arrow_type_for` can produce needs a SQL type entry, or an evolved
+    # column (e.g. datetime) would silently become `string`.
+    from extralit_server.index.lancedb_engine import _SQL_TYPE_BY_ARROW
+    from extralit_server.index.mapping import _ARROW_BY_DTYPE
+
+    for arrow_type in set(_ARROW_BY_DTYPE.values()):
+        assert arrow_type in _SQL_TYPE_BY_ARROW, f"no SQL type for Arrow type {arrow_type}"
+
+
+async def test_list_tables_pagination():
+    """_list_tables must follow page_token links until page_token is exhausted."""
+    import types
+    from unittest.mock import AsyncMock, patch
+
+    engine = LanceIndexEngine(uri="/tmp/fake-lance-uri")
+
+    page1 = types.SimpleNamespace(tables=["a", "b"], page_token="tok")
+    page2 = types.SimpleNamespace(tables=["c"], page_token=None)
+
+    async def _fake_list_tables(**kwargs):
+        if kwargs.get("page_token") == "tok":
+            return page2
+        return page1
+
+    fake_db = AsyncMock()
+    fake_db.list_tables = _fake_list_tables
+
+    async def _fake_conn():
+        return fake_db
+
+    with patch.object(engine, "_conn", _fake_conn):
+        result = await engine._list_tables()
+
+    assert result == ["a", "b", "c"]
diff --git a/extralit-server/tests/unit/index/__init__.py b/extralit-server/tests/unit/index/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/extralit-server/tests/unit/index/test_base.py b/extralit-server/tests/unit/index/test_base.py
new file mode 100644
index 000000000..a4ce88391
--- /dev/null
+++ b/extralit-server/tests/unit/index/test_base.py
@@ -0,0 +1,29 @@
+import inspect
+from uuid import uuid4
+
+from extralit_server.index.base import IndexEngine, IndexFilter, IndexSearchHit, IndexSearchResult
+
+
+def test_index_engine_is_abstract():
+    import pytest
+
+    with pytest.raises(TypeError):
+        IndexEngine()  # abstract methods unimplemented
+
+
+def test_required_async_methods_present():
+    for name in ("close", "ensure_table", "drop_table", "upsert", "delete", "search", "table_names"):
+        method = getattr(IndexEngine, name)
+        assert inspect.iscoroutinefunction(method), f"{name} must be async"
+
+
+def test_result_models_roundtrip():
+    hit = IndexSearchHit(record_id=uuid4(), score=1.5)
+    result = IndexSearchResult(hits=[hit], total=1)
+    assert result.total == 1
+    assert result.hits[0].score == 1.5
+
+
+def test_index_filter_shape():
+    f = IndexFilter(column="year", op="ge", value=2000)
+    assert (f.column, f.op, f.value) == ("year", "ge", 2000)
diff --git a/extralit-server/tests/unit/index/test_mapping.py b/extralit-server/tests/unit/index/test_mapping.py
new file mode 100644
index 000000000..d125476d5
--- /dev/null
+++ b/extralit-server/tests/unit/index/test_mapping.py
@@ -0,0 +1,89 @@
+from types import SimpleNamespace
+from uuid import uuid4
+
+import pyarrow as pa
+
+from extralit_server.enums import V2RecordStatus
+from extralit_server.index import mapping
+
+COLUMNS = [
+    {"name": "title", "dtype": "string[pyarrow]", "nullable": False, "review": None},
+    {"name": "year", "dtype": "int64", "nullable": True, "review": None},
+    {"name": "score", "dtype": "float64", "nullable": True, "review": None},
+]
+
+
+def test_table_name_is_hex_prefixed():
+    sid = uuid4()
+    assert mapping.table_name_for(sid) == f"schema_{sid.hex}"
+
+
+def test_arrow_type_mapping_covers_pandera_dtypes():
+    assert mapping.arrow_type_for("string[pyarrow]") == pa.large_string()
+    assert mapping.arrow_type_for("int64") == pa.int64()
+    assert mapping.arrow_type_for("float64") == pa.float64()
+    assert mapping.arrow_type_for("bool") == pa.bool_()
+    assert pa.types.is_timestamp(mapping.arrow_type_for("datetime64[ns]"))
+
+
+def test_arrow_type_unknown_dtype_falls_back_to_string():
+    assert mapping.arrow_type_for("category") == pa.large_string()
+
+
+def test_arrow_schema_raises_on_system_field_collision():
+    import pytest
+
+    colliding = [{"name": "text", "dtype": "string[pyarrow]", "nullable": True, "review": None}]
+    with pytest.raises(ValueError, match="reserved system fields"):
+        mapping.arrow_schema_for(colliding)
+
+    colliding2 = [{"name": "record_id", "dtype": "string[pyarrow]", "nullable": True, "review": None}]
+    with pytest.raises(ValueError, match="reserved system fields"):
+        mapping.arrow_schema_for(colliding2)
+
+
+def test_arrow_schema_has_system_fields_and_typed_columns():
+    schema = mapping.arrow_schema_for(COLUMNS)
+    names = schema.names
+    for sys_field in mapping.SYSTEM_FIELDS:
+        assert sys_field in names
+    assert schema.field("year").type == pa.int64()
+    assert schema.field("text").type == pa.large_string()
+
+
+def test_union_columns_dedupes_by_name_first_wins():
+    v1 = [{"name": "a", "dtype": "int64", "nullable": True, "review": None}]
+    v2 = [
+        {"name": "a", "dtype": "int64", "nullable": True, "review": None},
+        {"name": "b", "dtype": "string[pyarrow]", "nullable": True, "review": None},
+    ]
+    union = mapping.union_columns([v1, v2])
+    assert [c["name"] for c in union] == ["a", "b"]
+
+
+def test_concat_text_joins_string_columns_only():
+    fields = {"title": "Deep Learning", "year": 2016, "score": 9.1}
+    text = mapping.concat_text(fields, COLUMNS)
+    assert "title: Deep Learning" in text
+    assert "year" not in text  # non-string dtype excluded
+
+
+def test_record_to_row_fills_missing_cells_with_none():
+    rec = SimpleNamespace(
+        id=uuid4(),
+        reference="pmid:1",
+        schema_version_id=uuid4(),
+        status=V2RecordStatus.pending,
+        external_id="x-1",
+        fields={"title": "Deep Learning"},  # `year`/`score` absent
+    )
+    row = mapping.record_to_row(rec, COLUMNS)
+    assert row["record_id"] == str(rec.id)
+    assert row["schema_version_id"] == str(rec.schema_version_id)
+    assert row["status"] == V2RecordStatus.pending.value
+    assert row["title"] == "Deep Learning"
+    assert row["year"] is None
+    assert row["score"] is None
+    assert row["text"] == "title: Deep Learning"
+    # every arrow-schema field is present as a key
+    assert set(row) == set(mapping.arrow_schema_for(COLUMNS).names)
diff --git a/extralit-server/tests/unit/index/test_settings_lancedb_uri.py b/extralit-server/tests/unit/index/test_settings_lancedb_uri.py
new file mode 100644
index 000000000..64ed5eba4
--- /dev/null
+++ b/extralit-server/tests/unit/index/test_settings_lancedb_uri.py
@@ -0,0 +1,19 @@
+import os
+
+from extralit_server.settings import Settings
+
+
+def test_lancedb_uri_defaults_under_home_path():
+    s = Settings(home_path="/tmp/extralit-home", lancedb_uri=None)
+    assert s.lancedb_uri == os.path.join("/tmp/extralit-home", "lance")
+
+
+def test_lancedb_uri_explicit_value_is_respected():
+    s = Settings(home_path="/tmp/extralit-home", lancedb_uri="s3://bucket/lance")
+    assert s.lancedb_uri == "s3://bucket/lance"
+
+
+def test_lancedb_uri_reads_env_prefix(monkeypatch):
+    monkeypatch.setenv("EXTRALIT_LANCEDB_URI", "/data/custom-lance")
+    s = Settings(home_path="/tmp/extralit-home")
+    assert s.lancedb_uri == "/data/custom-lance"
diff --git a/extralit-server/uv.lock b/extralit-server/uv.lock
index 20ee48233..3fac23957 100644
--- a/extralit-server/uv.lock
+++ b/extralit-server/uv.lock
@@ -1171,6 +1171,7 @@ dependencies = [
     { name = "httpx" },
     { name = "huggingface-hub" },
     { name = "jinja2" },
+    { name = "lancedb" },
     { name = "lazy-loader" },
     { name = "litellm" },
     { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
@@ -1246,6 +1247,7 @@ requires-dist = [
     { name = "httpx", specifier = "~=0.27.0" },
     { name = "huggingface-hub", specifier = "~=0.34.0" },
     { name = "jinja2", specifier = ">=3.1.4" },
+    { name = "lancedb", specifier = ">=0.34.0" },
     { name = "lazy-loader", specifier = ">=0.4" },
     { name = "litellm", specifier = ">=1.80.0,<=1.82.6" },
     { name = "marker-pdf", marker = "extra == 'marker'", specifier = ">=1.9.3" },
@@ -1984,6 +1986,55 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/41/45/1a4ed80516f02155c51f51e8cedb3c1902296743db0bbc66608a0db2814f/jsonschema_specifications-2025.9.1-py3-none-any.whl", hash = "sha256:98802fee3a11ee76ecaca44429fda8a41bff98b00a0f2838151b113f210cc6fe", size = 18437, upload-time = "2025-09-08T01:34:57.871Z" },
 ]
 
+[[package]]
+name = "lance-namespace"
+version = "0.9.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "lance-namespace-urllib3-client" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/00/81/4cf8d0412e1f37b2bfa70d0aeb9c7ae4ab73607534e44d60b55efb485306/lance_namespace-0.9.0.tar.gz", hash = "sha256:f738b641cc615b17323baa4eb47900f184688739ee3d2ea9fe39396b9588e53d", size = 11637, upload-time = "2026-07-01T07:42:41.78Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/e1/fe/f38747c9610ade83dd9a99a0470b9432b6f21ce4e2bb5524edbe66f626fd/lance_namespace-0.9.0-py3-none-any.whl", hash = "sha256:f785ff10927e4ce0db69986576670fedd37f8a33521e8a4630c6be22db8061b2", size = 13501, upload-time = "2026-07-01T07:42:39.372Z" },
+]
+
+[[package]]
+name = "lance-namespace-urllib3-client"
+version = "0.9.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "pydantic" },
+    { name = "python-dateutil" },
+    { name = "typing-extensions" },
+    { name = "urllib3" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/d3/c3/32d0e2618549ace857c80a457e5915ef3e1145661baff876c8a5ec27be5b/lance_namespace_urllib3_client-0.9.0.tar.gz", hash = "sha256:cf796fa5307fa4dde91fe4bec2af28b90ba79191852d4394e8fe44276538e40f", size = 235805, upload-time = "2026-07-01T07:42:42.563Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/cc/ab/c8754da0a1efc817f8480100cfe12b7e04034df834759de8ecf02beff3cc/lance_namespace_urllib3_client-0.9.0-py3-none-any.whl", hash = "sha256:be819c8cffb1e460a3a504dbf52d1ca009560a48e7202b8c4279998e4adf9fe4", size = 405586, upload-time = "2026-07-01T07:42:40.503Z" },
+]
+
+[[package]]
+name = "lancedb"
+version = "0.34.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "deprecation" },
+    { name = "lance-namespace" },
+    { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" },
+    { name = "numpy", version = "2.4.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" },
+    { name = "overrides", marker = "python_full_version < '3.12'" },
+    { name = "packaging" },
+    { name = "pyarrow" },
+    { name = "pydantic" },
+    { name = "tqdm" },
+]
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/df/f7/5262b9aa593f790757163c0165ab0da1dda054758901bea7e4f02c9cb633/lancedb-0.34.0-cp39-abi3-macosx_11_0_arm64.whl", hash = "sha256:c462f2e6f933cad659fd0179394eaab578acbc9151fe2ef41bc29b36ecca5058", size = 52654213, upload-time = "2026-07-02T17:13:31.102Z" },
+    { url = "https://files.pythonhosted.org/packages/69/99/05ea0d32229ebea695193ff20c15d6ecae25785ad82a9d4723d98832a284/lancedb-0.34.0-cp39-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:48829e88e708947d0520454ab9e4f8efa35f3e3626469eadd3a6e061b89cb223", size = 55434501, upload-time = "2026-07-02T17:13:34.81Z" },
+    { url = "https://files.pythonhosted.org/packages/cd/4e/4325c13d5afa93c466428a5a0f168ad4d96f5eb4a77bbe7c5100d39c9897/lancedb-0.34.0-cp39-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:05ba8a5b58e064edfbe5be71b1abf2e411b4eaf295d1a173dcb1a55c5bfb5285", size = 58659359, upload-time = "2026-07-02T17:13:38.424Z" },
+    { url = "https://files.pythonhosted.org/packages/d9/5d/8ca165f1386caf6c4d1c515afd52f345b66432264eecfdfb7fd33eefd9af/lancedb-0.34.0-cp39-abi3-win_amd64.whl", hash = "sha256:51cbc11808f9e3332819b9367c975b3a888541447a8e7bea09c57c852a279153", size = 63530726, upload-time = "2026-07-02T17:13:41.612Z" },
+]
+
 [[package]]
 name = "lazy-loader"
 version = "0.5"
@@ -2941,6 +2992,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/5b/08/81573904c9ee6d9aca77430bfe4f88d32f8dd596afc306e3dd6603b3a935/opensearch_py-2.0.1-py2.py3-none-any.whl", hash = "sha256:daa5eb2279b89bf15d63312a922bd5ab7f266d3c2737e48dec6ff862d7b1838a", size = 214677, upload-time = "2022-11-29T23:17:15.46Z" },
 ]
 
+[[package]]
+name = "overrides"
+version = "7.7.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/36/86/b585f53236dec60aba864e050778b25045f857e17f6e5ea0ae95fe80edd2/overrides-7.7.0.tar.gz", hash = "sha256:55158fa3d93b98cc75299b1e67078ad9003ca27945c76162c1c0766d6f91820a", size = 22812, upload-time = "2024-01-27T21:01:33.423Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/2c/ab/fc8290c6a4c722e5514d80f62b2dc4c4df1a68a41d1364e625c35990fcf3/overrides-7.7.0-py3-none-any.whl", hash = "sha256:c7ed9d062f78b8e4c1a7b70bd8796b35ead4d9f510227ef9c5dc7626c60d7e49", size = 17832, upload-time = "2024-01-27T21:01:31.393Z" },
+]
+
 [[package]]
 name = "packaging"
 version = "26.0"