From fe65af22f3f61268d7d15bb6e8f3e22ae7fdef9b Mon Sep 17 00:00:00 2001 From: zhanghui Date: Fri, 24 Jul 2026 13:55:15 +0800 Subject: [PATCH 1/2] fix(lancedb): detect schema type drift and add cascade rebuild recovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit verify_business_schemas only compared column names, so a column whose on-disk Arrow type had drifted (name unchanged) slipped through and detonated later inside merge_insert as an opaque LanceError(IO) "Spill has sent an error" (#337). Now compare each shared column's Arrow type against schema.to_arrow_schema() — the exact schema get_table builds tables from, so a healthy table never false-positives. Reproduced #337 byte-identically: an episode.subject_vector column left as string or null by an older build, plus a real 1024-d vector on upsert, yields the exact crash. No lancedb version (0.13-0.34) renders Optional[Vector] as a non-vector type, so the startup guard is what should catch it — not the runtime. Add `everos cascade rebuild` as the safe recovery: it drops the business LanceDB tables and re-indexes from markdown, skipping the verify guard (which the drift would otherwise trip on startup). Unlike removing only .index/lancedb it re-populates already-done entries (reset_all clears the cascade queue); unlike removing all of .index it preserves unprocessed_buffer (messages not yet extracted). Fixes #337. --- .../entrypoints/cli/commands/cascade.py | 71 ++++++++- .../infra/persistence/lancedb/__init__.py | 73 ++++++--- .../persistence/lancedb/lancedb_manager.py | 22 +++ .../sqlite/repos/md_change_state.py | 20 ++- .../test_cascade_cli_integration.py | 112 ++++++++++++++ .../test_cli/test_cascade_command.py | 7 +- .../test_lancedb/test_verify_schemas.py | 139 ++++++++++++++++++ .../test_repos/test_md_change_state.py | 27 ++++ 8 files changed, 447 insertions(+), 24 deletions(-) create mode 100644 tests/unit/test_infra/test_lancedb/test_verify_schemas.py diff --git a/src/everos/entrypoints/cli/commands/cascade.py b/src/everos/entrypoints/cli/commands/cascade.py index 8d27e8aa0..c868d2f1b 100644 --- a/src/everos/entrypoints/cli/commands/cascade.py +++ b/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 ────────────────────────────────────────────────────────────── diff --git a/src/everos/infra/persistence/lancedb/__init__.py b/src/everos/infra/persistence/lancedb/__init__.py index 8f8b3c589..9c360e463 100644 --- a/src/everos/infra/persistence/lancedb/__init__.py +++ b/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", diff --git a/src/everos/infra/persistence/lancedb/lancedb_manager.py b/src/everos/infra/persistence/lancedb/lancedb_manager.py index 2099a6705..8c1c6154c 100644 --- a/src/everos/infra/persistence/lancedb/lancedb_manager.py +++ b/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 diff --git a/src/everos/infra/persistence/sqlite/repos/md_change_state.py b/src/everos/infra/persistence/sqlite/repos/md_change_state.py index b0082927b..418bb4669 100644 --- a/src/everos/infra/persistence/sqlite/repos/md_change_state.py +++ b/src/everos/infra/persistence/sqlite/repos/md_change_state.py @@ -29,7 +29,7 @@ import dataclasses -from sqlalchemy import func, select, update +from sqlalchemy import delete, func, select, update from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker @@ -374,6 +374,24 @@ async def reset_retryable_to_pending(self) -> int: await s.commit() return int(result.rowcount or 0) + async def reset_all(self) -> int: + """`cascade rebuild` engine: clear the entire work-queue table. + + This table is pure sync bookkeeping — a projection of (md × + index) — so deleting every row forces the next scan to treat + every md file as newly ``added`` and re-index it from scratch. + Touches **only** ``md_change_state``; other SQLite tables + (``memcell``, ``unprocessed_buffer``, …) are left intact, which + is what makes ``cascade rebuild`` non-destructive to un-extracted + buffered messages. + + Returns the number of rows deleted. + """ + async with session_scope(self._factory) as s: + result = await s.execute(delete(MdChangeState)) + await s.commit() + return int(result.rowcount or 0) + async def queue_summary(self) -> QueueSummary: """Aggregate the table for the ``cascade status`` CLI.""" async with session_scope(self._factory) as s: diff --git a/tests/integration/test_cascade_cli_integration.py b/tests/integration/test_cascade_cli_integration.py index b8307f82a..d43047411 100644 --- a/tests/integration/test_cascade_cli_integration.py +++ b/tests/integration/test_cascade_cli_integration.py @@ -187,6 +187,118 @@ async def seed() -> None: assert "pending: 1" in result.stdout +def _fake_orchestrator_factory(): # type: ignore[no-untyped-def] + from everos.component.tokenizer import build_tokenizer + from everos.core.persistence import MemoryRoot + from everos.memory.cascade import CascadeOrchestrator + + def _build() -> CascadeOrchestrator: + root = MemoryRoot.default() + root.ensure() + return CascadeOrchestrator( + memory_root=root, + embedder=_StubEmbedder(), + tokenizer=build_tokenizer(), + ) + + return _build + + +def test_rebuild_recovers_drifted_index_and_reindexes( + cli_runtime: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """``cascade rebuild`` recovers a type-drifted index. + + Proves the three properties bare ``rm`` can't offer together: + (1) it runs despite a schema-type drift that trips the normal + startup guard (``verify_business_schemas``); (2) it drops + recreates + the drifted table with the current (correct) type; (3) it re-indexes + md from scratch even for entries the queue already marked ``done``. + """ + import datetime as _dt + + import lancedb + import pyarrow as pa + + from everos.core.persistence import MemoryRoot + from everos.infra.persistence.lancedb import get_table + from everos.infra.persistence.lancedb.tables.atomic_fact import AtomicFact + from everos.infra.persistence.lancedb.tables.episode import Episode + from everos.infra.persistence.markdown import AtomicFactWriter + + root = MemoryRoot.default() + root.ensure() + owner_id = "u_rebuild" + bucket = _dt.date(2026, 5, 18) + md_path = ( + f"default_app/default_project/users/{owner_id}/.atomic_facts/" + f"atomic_fact-{bucket.isoformat()}.md" + ) + + async def _seed_md() -> None: + writer = AtomicFactWriter(root=root) + items = [ + ( + { + "owner_id": owner_id, + "session_id": f"s_{j}", + "timestamp": "2026-05-18T07:04:26+00:00", + "parent_id": f"mc_{j}", + "sender_ids": [owner_id], + }, + {"Fact": f"seed fact {j}"}, + ) + for j in range(2) + ] + await writer.append_entries(owner_id, items, date=bucket) + + async def _drift_episode_table() -> None: + conn = await lancedb.connect_async(str(root.lancedb_dir)) + drifted = pa.schema( + [ + pa.field("subject_vector", pa.string(), nullable=True) + if f.name == "subject_vector" + else f + for f in Episode.to_arrow_schema() + ] + ) + await conn.create_table("episode", schema=drifted) + conn.close() + + async def _episode_subject_vector_type(): # type: ignore[no-untyped-def] + tbl = await get_table("episode", Episode) + return (await tbl.schema()).field("subject_vector").type + + async def _atomic_fact_row_count() -> int: + tbl = await get_table(AtomicFact.TABLE_NAME, AtomicFact) + return await tbl.count_rows(filter=f"md_path = '{md_path}'") + + asyncio.run(_seed_md()) + asyncio.run(_drift_episode_table()) + asyncio.run(_dispose_all()) + + monkeypatch.setattr( + cascade_mod, "_build_orchestrator", _fake_orchestrator_factory() + ) + + # Contrast: a normal command boots via _runtime() → verify trips on the drift. + status_result = CliRunner().invoke(cascade_mod.app, ["status"]) + assert status_result.exit_code != 0 + asyncio.run(_dispose_all()) + + # rebuild skips verify, recreates the table, and re-indexes md. + result = CliRunner().invoke(cascade_mod.app, ["rebuild", "--yes"]) + assert result.exit_code == 0, result.stdout + assert "rebuild complete" in result.stdout + asyncio.run(_dispose_all()) + + # Table recreated with the correct vector type; md re-indexed. + assert asyncio.run(_episode_subject_vector_type()).equals( + Episode.to_arrow_schema().field("subject_vector").type + ) + assert asyncio.run(_atomic_fact_row_count()) == 2 + + # Reduce false negatives on date drift. def test_resolve_relative_via_command_arg(cli_runtime: Path) -> None: """An absolute path under the root works through ``cascade sync ``.""" diff --git a/tests/unit/test_entrypoints/test_cli/test_cascade_command.py b/tests/unit/test_entrypoints/test_cli/test_cascade_command.py index d27a1c457..632212841 100644 --- a/tests/unit/test_entrypoints/test_cli/test_cascade_command.py +++ b/tests/unit/test_entrypoints/test_cli/test_cascade_command.py @@ -3,7 +3,7 @@ The orchestrator paths require live sqlite + lancedb singletons; those are exercised by integration tests. Here we cover: -- subcommand registration (sync / status / fix) +- subcommand registration (sync / status / fix / rebuild) - ``--help`` exit codes - ``_resolve_relative`` (path arithmetic vs. memory root) - ``_print_failed_table`` (formatting of failed rows) @@ -21,9 +21,9 @@ from everos.entrypoints.cli.commands import cascade as cascade_mod -def test_app_registers_three_commands() -> None: +def test_app_registers_expected_commands() -> None: names = {cmd.name for cmd in cascade_mod.app.registered_commands} - assert names == {"sync", "status", "fix"} + assert names == {"sync", "status", "fix", "rebuild"} def test_help_exits_zero() -> None: @@ -32,6 +32,7 @@ def test_help_exits_zero() -> None: assert "sync" in result.stdout assert "status" in result.stdout assert "fix" in result.stdout + assert "rebuild" in result.stdout def test_resolve_relative_under_root( diff --git a/tests/unit/test_infra/test_lancedb/test_verify_schemas.py b/tests/unit/test_infra/test_lancedb/test_verify_schemas.py new file mode 100644 index 000000000..9a5c5af8c --- /dev/null +++ b/tests/unit/test_infra/test_lancedb/test_verify_schemas.py @@ -0,0 +1,139 @@ +"""``verify_business_schemas`` startup guard. + +Regression coverage for EverOS #337: the guard must catch a column +whose on-disk Arrow *type* drifted from the current schema (not only a +missing/extra column name), and must NOT false-positive on a healthy +table freshly built from the current schema. + +White-box surfaces: builds LanceDB tables directly on disk under an +isolated ``EVEROS_ROOT`` and drives ``verify_business_schemas`` / +``get_table`` against them. +""" + +from __future__ import annotations + +from pathlib import Path + +import lancedb +import pyarrow as pa +import pytest + +from everos.core.persistence import MemoryRoot +from everos.infra.persistence.lancedb import ( + LanceDBSchemaMismatchError, + drop_business_tables, + ensure_business_indexes, + get_connection, + get_table, + lancedb_manager, + verify_business_schemas, +) +from everos.infra.persistence.lancedb.tables.episode import Episode + + +@pytest.fixture(autouse=True) +async def _isolated_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + """Point the manager singleton at an isolated memory root.""" + monkeypatch.setenv("EVEROS_ROOT", str(tmp_path)) + lancedb_manager._conn = None + lancedb_manager._tables.clear() + yield + await lancedb_manager.dispose_connection() + + +async def _create_episode_table_on_disk(arrow_schema: pa.Schema) -> None: + """Create the ``episode`` table straight on disk, then drop the + connection so the manager reopens it from disk on next use.""" + root = MemoryRoot.default() + root.ensure() + conn = await lancedb.connect_async(str(root.lancedb_dir)) + await conn.create_table("episode", schema=arrow_schema) + conn.close() + lancedb_manager._conn = None + lancedb_manager._tables.clear() + + +def _episode_schema_with_subject_vector_as(dtype: pa.DataType) -> pa.Schema: + """Current episode Arrow schema, but subject_vector forced to ``dtype``.""" + return pa.schema( + [ + pa.field("subject_vector", dtype, nullable=True) + if f.name == "subject_vector" + else f + for f in Episode.to_arrow_schema() + ] + ) + + +async def test_verify_passes_on_healthy_tables() -> None: + """A table built from the current schema must NOT trip the guard. + + Guards against a type comparison that false-positives (e.g. on the + fixed_size_list item name or the datetime tz rewrite).""" + # verify_business_schemas creates every business table fresh via + # get_table, then re-reads and compares — so a clean run proves no + # false positive across ALL business schemas. + await verify_business_schemas() + # A second run over the now-existing tables must also pass. + await verify_business_schemas() + + +async def test_verify_raises_on_subject_vector_string_drift() -> None: + """#337: subject_vector left as `string` must be caught by type.""" + await _create_episode_table_on_disk( + _episode_schema_with_subject_vector_as(pa.string()) + ) + with pytest.raises(LanceDBSchemaMismatchError) as exc: + await verify_business_schemas() + msg = str(exc.value) + assert "episode" in msg + assert "type_drift" in msg + assert "subject_vector" in msg + + +async def test_verify_raises_on_subject_vector_null_drift() -> None: + """A `null`-typed subject_vector (data-inferred all-None) is also drift.""" + await _create_episode_table_on_disk( + _episode_schema_with_subject_vector_as(pa.null()) + ) + with pytest.raises(LanceDBSchemaMismatchError) as exc: + await verify_business_schemas() + assert "subject_vector" in str(exc.value) + + +async def test_verify_raises_on_missing_column() -> None: + """A table missing a current column is caught by name (unchanged).""" + reduced = pa.schema( + [f for f in Episode.to_arrow_schema() if f.name != "subject_vector"] + ) + await _create_episode_table_on_disk(reduced) + with pytest.raises(LanceDBSchemaMismatchError) as exc: + await verify_business_schemas() + assert "missing" in str(exc.value) + assert "subject_vector" in str(exc.value) + + +async def test_drop_business_tables_removes_then_recreatable() -> None: + """drop_business_tables drops existing business tables + clears the cache; + the next get_table recreates a fresh, current-schema table.""" + # Materialise a drifted episode table + the rest of the business set. + await _create_episode_table_on_disk( + _episode_schema_with_subject_vector_as(pa.string()) + ) + await ensure_business_indexes() # create the remaining business tables + conn = await get_connection() + assert "episode" in set((await conn.list_tables()).tables) + + dropped = await drop_business_tables() + + assert "episode" in dropped + conn = await get_connection() + assert "episode" not in set((await conn.list_tables()).tables) + assert "episode" not in lancedb_manager._tables # cache evicted + # Recreated fresh from the current schema → correct vector type, not string. + tbl = await get_table("episode", Episode) + assert ( + (await tbl.schema()) + .field("subject_vector") + .type.equals(Episode.to_arrow_schema().field("subject_vector").type) + ) diff --git a/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py b/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py index 6219ab1d7..c8f0c21b9 100644 --- a/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py +++ b/tests/unit/test_infra/test_sqlite/test_repos/test_md_change_state.py @@ -358,6 +358,33 @@ async def test_reset_retryable_to_pending_zero_when_none_eligible( assert await repo.reset_retryable_to_pending() == 0 +# ── reset_all ─────────────────────────────────────────────────────────── + + +async def test_reset_all_clears_every_row(repo: _MdChangeStateRepo) -> None: + """`cascade rebuild` engine: every row is deleted regardless of status.""" + await repo.upsert("a.md", kind="episode", change_type="added", mtime=0.0) + await repo.claim_one("a.md") + await repo.mark_done("a.md") # a: done + await repo.upsert("b.md", kind="episode", change_type="added", mtime=0.0) + await repo.claim_one("b.md") + await repo.mark_failed("b.md", retryable=False, error="x", new_retry_count=0) + await repo.upsert("c.md", kind="episode", change_type="added", mtime=0.0) # pending + + deleted = await repo.reset_all() + + assert deleted == 3 + assert await repo.get_by_id("a.md") is None + assert await repo.get_by_id("b.md") is None + assert await repo.get_by_id("c.md") is None + summary = await repo.queue_summary() + assert summary.pending == 0 and summary.done == 0 + + +async def test_reset_all_zero_on_empty_table(repo: _MdChangeStateRepo) -> None: + assert await repo.reset_all() == 0 + + # ── list_failed ───────────────────────────────────────────────────────── From cd947e9d83d55c5ad7fcaa89e7320804901698cf Mon Sep 17 00:00:00 2001 From: zhanghui Date: Fri, 24 Jul 2026 14:19:58 +0800 Subject: [PATCH 2/2] docs(cascade): document cascade rebuild and correct recovery guidance MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the `everos cascade rebuild` command to the runbook, CLI, and how-memory-works docs. Correct the old recovery guidance: a bare `rm -rf .index/lancedb` leaves md_change_state marked `done`, so the scanner skips those files and the index comes back empty — the runbook previously claimed a full repopulation that does not happen. `cascade rebuild` is the safe path (re-populates done entries, preserves unprocessed_buffer). Also document that verify now checks column types. --- docs/cascade_runbook.md | 64 +++++++++++++++++++++++++++++++++++----- docs/cli.md | 6 ++-- docs/how-memory-works.md | 16 ++++++---- 3 files changed, 71 insertions(+), 15 deletions(-) diff --git a/docs/cascade_runbook.md b/docs/cascade_runbook.md index c9811eeea..a2ea8837a 100644 --- a/docs/cascade_runbook.md +++ b/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. diff --git a/docs/cli.md b/docs/cli.md index e24e6e5ea..ca63b3a52 100644 --- a/docs/cli.md +++ b/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 -`/.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` diff --git a/docs/how-memory-works.md b/docs/how-memory-works.md index 1a8136478..b9c0bb19b 100644 --- a/docs/how-memory-works.md +++ b/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 `/` 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 /.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 /.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`).