fix(lancedb): detect schema type drift and add cascade rebuild recovery

docs/cascade_runbook.md

@@ -82,6 +82,39 @@ The CLI builds the same `CascadeOrchestrator` as the daemon but only
 calls `sync_once` / `drain_once` — no watcher / scanner background
 task. So it's safe to run in parallel with a live `everos server`.
 
+## Rebuild the index: `everos cascade rebuild`
+
+The safe recovery from a drifted or corrupt LanceDB index. It rebuilds
+the whole index from markdown (the source of truth) in one shot:
+
+```bash
+everos cascade rebuild          # prompts for confirmation
+everos cascade rebuild --yes    # non-interactive
+```
+
+What it does, in order:
+
+1. **Drops** every business LanceDB table (`drop_business_tables`) and
+   evicts them from the connection cache.
+2. **Recreates** them empty from the current schema + FTS indexes
+   (`ensure_business_indexes`).
+3. **Clears** the cascade queue (`md_change_state.reset_all`) so every
+   md file re-enqueues as `added` on the next scan.
+4. **Re-scans + drains** (`sync_once`): re-embeds and re-inserts every
+   md entry.
+
+It deliberately **skips `verify_business_schemas`** — the drift it
+recovers from would otherwise trip that guard on startup before the
+rebuild could run (chicken-and-egg).
+
+Why not a bare `rm`:
+
+| Recovery | Re-populates `done` entries | Preserves `unprocessed_buffer` |
+|---|---|---|
+| `rm -rf .index/lancedb` | ❌ scanner skips `done` rows → empty index | ✅ |
+| `rm -rf .index` | ✅ | ❌ deletes un-extracted messages |
+| `everos cascade rebuild` | ✅ | ✅ |
+
 ## Recovery paths
 
 ### LanceDB schema drift on startup
@@ -91,15 +124,28 @@ an on-disk table has columns the current Pydantic schema does not
 declare (or vice versa), the boot fails with:
 
 ```
-LanceDB table 'episode' schema drift: missing=[...], extra=[...].
-The index is rebuildable from md — recover with
-`rm -rf ~/.everos/.index/lancedb` and restart.
+LanceDB table 'episode' schema drift: missing=[...], extra=[...],
+type_drift=[...]. The index is rebuildable from md — recover with
+`everos cascade rebuild`.
 ```
 
-This is the documented recovery: delete the index, restart the
-server, the scanner will pick up every md file on its first sweep and
-the worker repopulates LanceDB. Markdown is the source of truth, so
-no data is lost.
+`verify_business_schemas` compares both the column **names** and their
+**Arrow types** against the current schema. Catching type drift matters:
+an `episode.subject_vector` column left as `string` (or `null`) by an
+older build, while the schema now declares a 1024-d `fixed_size_list`,
+has the same column *name* — so a name-only check would wave it through
+and it would detonate later inside `merge_insert` as an opaque
+`LanceError(IO): Spill has sent an error` (EverOS #337). The type check
+turns that into this clean startup error.
+
+Recover with **`everos cascade rebuild`** (documented above). Do **not** just
+`rm -rf ~/.everos/.index/lancedb`: that clears the vectors but leaves
+`md_change_state` marked `done`, so the scanner skips every already-
+indexed file and the index comes back **empty**. And do **not**
+`rm -rf ~/.everos/.index`: that also deletes `unprocessed_buffer`
+(messages received but not yet extracted — not rebuildable from md).
+`cascade rebuild` is correct on both counts. Markdown is the source of
+truth, so no memory content is lost.
 
 ### inotify watch-limit exhaustion (Linux)
 
@@ -262,7 +308,9 @@ is a deployment-side change with no schema work.
 
 ## What cascade does NOT do (yet)
 
-- **Schema migration**: LanceDB column changes require `rm -rf`.
+- **Schema migration**: LanceDB has no in-place column migration; a
+  schema change is recovered by rebuilding from md (`everos cascade
+  rebuild`), not an automatic `ALTER`.
 - **Parent-id back-link**: Episode rows currently carry
   `parent_id=None`; the writer doesn't preserve the source memcell id
   in the entry inline. Tracked separately.

docs/cli.md

@@ -43,8 +43,10 @@ Each subcommand lives in its own module under
 registered in `cli/main.py`. The CLI is intentionally small — hot-path
 business (`/add` `/flush` `/search` `/get`) is the **HTTP API**, not the
 CLI; the CLI covers setup (`init`), running the server, and index ops
-(`cascade`). There is no `reindex` command — rebuild by deleting
-`<root>/.index/lancedb` and restarting, or run `everos cascade sync`.
+(`cascade`). There is no `reindex` command — for an incremental
+catch-up run `everos cascade sync`; to rebuild the whole index from
+markdown (recovery from drift / corruption) run `everos cascade
+rebuild`.
 
 ## `everos server start`
 

docs/how-memory-works.md

@@ -90,7 +90,8 @@ visually distinct from a user-named one).
     table `md_change_state`) — there is no `.cascade.log` / `.manifest.json`
     file in the current implementation. The `<app>/<project>` nesting is
     real and always present (`default_app/default_project` for the default
-    scope). There is **no `everos reindex` command** (see
+    scope). There is no command literally named `reindex`, but
+    `everos cascade rebuild` rebuilds the index from markdown (see
     [Operating it](#operating-it)).
 
 The path manager is
@@ -296,12 +297,17 @@ The CLI ([cli.md](cli.md)) is intentionally small:
 | `everos cascade status` | queue / LSN summary |
 | `everos cascade sync` | drain the cascade queue now (force md → LanceDB) |
 | `everos cascade fix` | list failed rows / re-enqueue retryable ones |
+| `everos cascade rebuild` | rebuild the whole index from markdown (drift / corruption recovery) |
 
 !!! warning "There is no `everos reindex` or `everos flush`"
-    - **Reindex** = the index is rebuildable: stop the server,
-      `rm -rf <memory-root>/.index/lancedb`, restart — the cascade
-      rebuilds from markdown. For an incremental catch-up, use
-      `everos cascade sync`.
+    - **Reindex** = the index is rebuildable from markdown. To rebuild
+      the whole index, run `everos cascade rebuild` — it drops the
+      LanceDB tables and re-indexes from md, re-populating even entries
+      the queue already marked `done` and preserving un-extracted
+      buffered messages. (A bare `rm -rf <memory-root>/.index/lancedb`
+      is **not** enough: the cascade queue still shows those files
+      `done`, so the scanner skips them and the index comes back empty.)
+      For an incremental catch-up, use `everos cascade sync`.
     - **Flush** is an HTTP endpoint (`POST /api/v1/memory/flush`), not a
       CLI command — it forces *extraction* of the session buffer, which is
       a different thing from forcing *index sync* (`cascade sync`).

src/everos/entrypoints/cli/commands/cascade.py

@@ -1,6 +1,6 @@
 """``everos cascade`` subcommand group.
 
-Three one-shot operations on the cascade subsystem, all run in-process
+One-shot operations on the cascade subsystem, all run in-process
 without standing up the FastAPI app:
 
 - ``cascade sync [PATH]`` — flush the work queue. With ``PATH`` the
@@ -11,6 +11,11 @@
 - ``cascade fix`` — list every ``failed`` row. With ``--apply``, also
   reset ``retryable=TRUE`` rows back to ``pending`` and drain the
   worker once so the retry actually runs before the command returns.
+- ``cascade rebuild`` — drop every business LanceDB table and re-index
+  all md from scratch. Recovery for a drifted / corrupt index; safe
+  because md is the source of truth and un-extracted buffered messages
+  are preserved. Skips the schema-verify guard (which the drift would
+  otherwise trip on startup).
 
 CLI is in-process (12 doc §7.1 + 16 doc §9.2): it constructs the same
 :class:`CascadeOrchestrator` as the daemon but only calls
@@ -36,6 +41,7 @@
 from everos.core.persistence import MemoryRoot
 from everos.infra.persistence.lancedb import (
     dispose_connection,
+    drop_business_tables,
     ensure_business_indexes,
     get_connection,
     verify_business_schemas,
@@ -71,18 +77,24 @@ def _cascade_callback(
 
 
 @asynccontextmanager
-async def _runtime():  # type: ignore[no-untyped-def]
+async def _runtime(*, verify: bool = True):  # type: ignore[no-untyped-def]
     """Stand up sqlite + lancedb the same way the API lifespan would.
 
     The CLI piggybacks on the same singletons as the running daemon
     (lazy + process-wide), so if a server happens to be running on
     the same memory root, both share state correctly.
+
+    ``verify=False`` skips :func:`verify_business_schemas` — required by
+    ``cascade rebuild``, whose whole purpose is to recover from a table
+    whose schema *has* drifted; running the guard there would abort
+    startup before the rebuild could fix it (chicken-and-egg).
     """
     engine = get_engine()
     async with engine.begin() as conn:
         await conn.run_sync(SQLModel.metadata.create_all)
     await get_connection()
-    await verify_business_schemas()
+    if verify:
+        await verify_business_schemas()
     await ensure_business_indexes()
     try:
         yield
@@ -234,6 +246,59 @@ async def _run() -> None:
     asyncio.run(_run())
 
 
+# ── rebuild ────────────────────────────────────────────────────────────────
+
+
+@app.command("rebuild")
+def rebuild(
+    yes: Annotated[
+        bool,
+        typer.Option("--yes", "-y", help="Skip the confirmation prompt."),
+    ] = False,
+) -> None:
+    """Rebuild the LanceDB index from markdown (recover from schema drift).
+
+    Drops every business LanceDB table and re-indexes all md from
+    scratch. Markdown is the source of truth, so no memory content is
+    lost, and this is the **safe** recovery from a drifted / corrupt
+    index (e.g. the ``verify_business_schemas`` startup failure):
+
+    - unlike ``rm -rf ~/.everos/.index/lancedb``, it re-populates
+      already-indexed entries (that command leaves the cascade queue
+      marked ``done``, so nothing re-indexes and the index comes back
+      empty);
+    - unlike ``rm -rf ~/.everos/.index``, it preserves SQLite state that
+      is NOT rebuildable from md — notably ``unprocessed_buffer``
+      (messages received but not yet extracted).
+    """
+    if not yes:
+        typer.confirm(
+            "Drop all LanceDB business tables and re-index from markdown?",
+            abort=True,
+        )
+
+    async def _run() -> None:
+        # verify=False: the on-disk schema may be exactly what we're here
+        # to fix; the startup guard would abort before we could rebuild.
+        async with _runtime(verify=False):
+            dropped = await drop_business_tables()
+            typer.echo(
+                f"dropped {len(dropped)} LanceDB table(s): "
+                f"{', '.join(dropped) or '(none)'}"
+            )
+            # Recreate the tables (current schema) + FTS indexes.
+            await ensure_business_indexes()
+            # Clear the work queue so every md file re-enqueues as `added`.
+            cleared = await md_change_state_repo.reset_all()
+            typer.echo(f"reset {cleared} cascade queue row(s)")
+            # Re-scan + drain: re-embed and re-insert every md entry.
+            orchestrator = _build_orchestrator()
+            processed = await orchestrator.sync_once()
+            typer.echo(f"rebuild complete — re-indexed {processed} md file(s)")
+
+    asyncio.run(_run())
+
+
 # ── helpers ──────────────────────────────────────────────────────────────
 
 

src/everos/infra/persistence/lancedb/__init__.py

@@ -34,6 +34,7 @@
 # schema so callers can rely on the package alone to surface every schema.
 from . import tables as tables
 from .lancedb_manager import dispose_connection as dispose_connection
+from .lancedb_manager import drop_tables as _drop_tables
 from .lancedb_manager import get_connection as get_connection
 from .lancedb_manager import get_table as get_table
 from .repos import agent_case_repo as agent_case_repo
@@ -68,9 +69,10 @@ class LanceDBSchemaMismatchError(RuntimeError):
     from the corresponding Pydantic schema.
 
     Cascade re-builds LanceDB from md (the SoT), so the recovery is
-    deterministic: delete the index directory and let it reindex.
-    The lifespan surfaces the explicit ``rm -rf ~/.everos/.index/
-    lancedb`` instruction in the error message; see
+    deterministic: ``everos cascade rebuild`` drops the business tables
+    and re-indexes from md, preserving SQLite state that is *not*
+    rebuildable from md (notably ``unprocessed_buffer`` — messages not
+    yet extracted). The error message surfaces that command; see
     ``docs/cascade_runbook.md`` for the wider context.
     """
 
@@ -147,31 +149,67 @@ async def ensure_business_indexes() -> None:
 
 async def verify_business_schemas() -> None:
     """Fail loud at startup if an existing LanceDB table's columns don't
-    match its current Pydantic schema.
+    match its current Pydantic schema — in **name or type**.
 
     LanceDB doesn't migrate columns automatically; an older index dir
-    (e.g. with the pre-``content_sha256`` shape) would fail
-    unpredictably on upsert. Checking column names up-front turns that
-    into a clean startup error pointing the user at the recovery path
-    (``rm -rf ~/.everos/.index/lancedb`` — the index is rebuildable
-    from md, see ``12_cascade_design.md``).
+    would fail unpredictably on upsert. Checking the schema up-front
+    turns that into a clean startup error pointing the user at the
+    recovery path (``rm -rf ~/.everos/.index/lancedb`` — the index is
+    rebuildable from md, see ``12_cascade_design.md``).
+
+    Both dimensions are checked against ``schema.to_arrow_schema()`` —
+    the exact schema ``get_table`` builds the table from, so a healthy
+    table never false-positives:
+
+    * **Column set** — a missing / extra column (e.g. a pre-``content_sha256``
+      table) is caught by name.
+    * **Column type** — a column whose on-disk Arrow type drifted from
+      the current schema. This is the class of drift behind EverOS #337:
+      an ``episode.subject_vector`` column left as ``string`` (or ``null``)
+      by an older build, while the current schema declares a 1024-d
+      ``fixed_size_list``. The name matches, so a name-only check waves it
+      through and it detonates deep inside ``merge_insert`` as an opaque
+      ``LanceError(IO): Spill has sent an error``. Comparing types surfaces
+      it here instead.
     """
     for schema in _BUSINESS_SCHEMAS:
         table = await get_table(schema.TABLE_NAME, schema)
-        arrow_schema = await table.schema()
-        actual = set(arrow_schema.names)
-        expected = set(schema.model_fields.keys())
-        missing = expected - actual
-        extra = actual - expected
-        if missing or extra:
+        on_disk = await table.schema()
+        expected = schema.to_arrow_schema()
+        on_disk_names = set(on_disk.names)
+        expected_names = set(expected.names)
+        missing = expected_names - on_disk_names
+        extra = on_disk_names - expected_names
+        # Type drift on columns present in both, compared against the
+        # authoritative to_arrow_schema() Arrow types.
+        type_drift = [
+            f"{name}: on-disk {on_disk.field(name).type} "
+            f"!= expected {expected.field(name).type}"
+            for name in sorted(on_disk_names & expected_names)
+            if not on_disk.field(name).type.equals(expected.field(name).type)
+        ]
+        if missing or extra or type_drift:
             raise LanceDBSchemaMismatchError(
                 f"LanceDB table {schema.TABLE_NAME!r} schema drift: "
-                f"missing={sorted(missing)}, extra={sorted(extra)}. "
+                f"missing={sorted(missing)}, extra={sorted(extra)}, "
+                f"type_drift={type_drift}. "
                 "The index is rebuildable from md — recover with "
-                "`rm -rf ~/.everos/.index/lancedb` and restart."
+                "`everos cascade rebuild` (drops + re-indexes from md, "
+                "preserving un-extracted buffered messages)."
             )
 
 
+async def drop_business_tables() -> list[str]:
+    """Drop every business LanceDB table; return the names dropped.
+
+    The tables are a rebuildable projection of markdown, so dropping is
+    non-destructive to memory content — ``cascade rebuild`` recreates and
+    re-populates them from md. Evicts the dropped tables from the manager
+    cache so a later :func:`get_table` reopens the fresh table.
+    """
+    return await _drop_tables([schema.TABLE_NAME for schema in _BUSINESS_SCHEMAS])
+
+
 __all__ = [
     "AgentCase",
     "AgentSkill",
@@ -186,6 +224,7 @@ async def verify_business_schemas() -> None:
     "agent_skill_repo",
     "atomic_fact_repo",
     "dispose_connection",
+    "drop_business_tables",
     "ensure_business_indexes",
     "episode_repo",
     "foresight_repo",

src/everos/infra/persistence/lancedb/lancedb_manager.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 import asyncio
+from collections.abc import Sequence
 
 from lancedb import AsyncConnection, AsyncTable
 
@@ -53,6 +54,27 @@ async def get_table(
         return _tables[name]
 
 
+async def drop_tables(names: Sequence[str]) -> list[str]:
+    """Drop the named tables if present; return the names actually dropped.
+
+    Each dropped table is also evicted from the cache so a later
+    :func:`get_table` recreates it fresh from the current schema. Used by
+    ``cascade rebuild`` to reset a corrupt / drifted index — the tables
+    are a rebuildable projection of markdown, so dropping is safe.
+    """
+    async with _lock:
+        conn = await _ensure_connection_locked()
+        existing = set((await conn.list_tables()).tables)
+        dropped: list[str] = []
+        for name in names:
+            if name in existing:
+                await conn.drop_table(name)
+                _tables.pop(name, None)
+                dropped.append(name)
+                logger.info("lancedb_table_dropped", name=name)
+        return dropped
+
+
 async def dispose_connection() -> None:
     """Close the connection + clear table cache. Idempotent."""
     global _conn