feat(cascade): report duplicate index primary keys

docs/cascade_runbook.md

@@ -52,6 +52,11 @@ lsn:
 - `failed (retryable=FALSE)` is always user-actionable. Cascade will
   never auto-clear these — they represent malformed md the user must
   edit.
+- `status --duplicates` additionally scans every business table and
+  reports duplicate `(app_id, project_id, id)` rows. LanceDB has no
+  unique constraint, so a write path that bypasses the merge helper can
+  leave two rows for one logical key; searches then silently see only
+  one of them. The scan is O(N) per table, which is why it is opt-in.
 
 ### Machine-readable: the `cascade` block on `GET /health`
 

docs/cli.md

@@ -33,7 +33,7 @@ everos
 ├── server
 │   └── start [--host] [--port] [--root] [--reload] [--log-level]   Start the HTTP API server (uvicorn)
 └── cascade [--root PATH]           Inspect / operate the md → LanceDB sync queue
-    ├── status                      Queue / LSN summary
+    ├── status [--duplicates]       Queue / LSN summary; opt-in duplicate-key scan
     ├── sync [PATH]                 Drain the queue now (optional PATH force-enqueues)
     └── fix [--apply]               List failed rows / re-enqueue retryable ones
 ```

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

@@ -59,6 +59,7 @@
     assert_record_dim_matches_settings,
     connect,
     count_business_rows,
+    count_duplicate_keys,
     drop_business_tables,
     ensure_business_indexes,
     shutdown,
@@ -342,6 +343,15 @@ def status(
         bool | None,
         typer.Option("--verbose", "-v", help=_VERBOSE_OPTION_HELP),
     ] = None,
+    duplicates: Annotated[
+        bool,
+        typer.Option(
+            "--duplicates",
+            help=(
+                "Scan every business table for duplicate (app_id, project_id, id) rows."
+            ),
+        ),
+    ] = False,
 ) -> None:
     """Print the queue / LSN summary."""
     _apply_root_env(root)
@@ -381,6 +391,11 @@ async def _run() -> None:
             typer.echo("index rows:")
             for table, count in rows.items():
                 typer.echo(f"  {table + ':':<24} {count}")
+            if duplicates:
+                dupes = await count_duplicate_keys()
+                typer.echo("duplicate primary keys:")
+                for table, count in dupes.items():
+                    typer.echo(f"  {table + ':':<24} {count}")
             # `summary.done` would count completed deletions, so a user who
             # removed their last markdown file would be warned that their
             # index is missing and sent to rebuild it from nothing.

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

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+from collections import Counter
 from collections.abc import Iterable
 from typing import Any
 
@@ -235,6 +236,27 @@ async def count_business_rows() -> dict[str, int]:
     return {repo.table_name: await repo.count() for repo in ALL_REPOS}
 
 
+async def count_duplicate_keys() -> dict[str, int]:
+    """Duplicate ``(app_id, project_id, id)`` rows per business table.
+
+    LanceDB has no unique constraint, so a write path that bypasses the
+    merge/upsert helper can leave two rows with the same logical primary key.
+    The next cascade write may then pick one arbitrarily, and searches silently
+    lose the other. This is an O(N) scan per table, which is why only the
+    opt-in ``cascade status --duplicates`` check calls it (issue #479).
+
+    Milvus enforces its own primary key, so its counts should always be zero;
+    keeping the check backend-neutral makes the CLI output comparable.
+    """
+    out: dict[str, int] = {}
+    for repo in ALL_REPOS:
+        counts = Counter(
+            (row.app_id, row.project_id, row.id) for row in await repo.scan()
+        )
+        out[repo.table_name] = sum(n - 1 for n in counts.values() if n > 1)
+    return out
+
+
 def repo_for_schema(schema: type[Any]) -> IndexRepository[Any]:
     """Resolve a registered repository by its logical record model."""
     for repo in ALL_REPOS:
@@ -280,6 +302,7 @@ def repo_for_schema(schema: type[Any]) -> IndexRepository[Any]:
     "connect",
     "contains",
     "count_business_rows",
+    "count_duplicate_keys",
     "drop_business_tables",
     "ensure_business_indexes",
     "episode_repo",

tests/unit/test_entrypoints/test_cli/test_cascade_command.py

@@ -39,6 +39,12 @@ def test_help_exits_zero() -> None:
     assert "rebuild" in result.stdout
 
 
+def test_status_help_lists_duplicates_flag() -> None:
+    result = CliRunner().invoke(cascade_mod.app, ["status", "--help"])
+    assert result.exit_code == 0
+    assert "--duplicates" in result.stdout
+
+
 def test_resolve_relative_under_root(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/unit/test_infra/test_index_diagnostics.py

@@ -0,0 +1,42 @@
+"""Duplicate-key scan used by ``everos cascade status --duplicates``."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from everos.infra.persistence import index as index_mod
+
+
+class _Repo:
+    def __init__(self, table_name: str, rows: list[SimpleNamespace]) -> None:
+        self.table_name = table_name
+        self._rows = rows
+
+    async def scan(self) -> list[SimpleNamespace]:
+        return self._rows
+
+
+def _row(key: str) -> SimpleNamespace:
+    return SimpleNamespace(app_id="default", project_id="default", id=key)
+
+
+async def test_count_duplicate_keys_reports_extra_rows(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # A raw write can leave two rows for one logical key; the scan reports
+    # the extra row, so a table with two unique keys is not "1 duplicate".
+    rows = [_row("a"), _row("a"), _row("b")]
+    monkeypatch.setattr(index_mod, "ALL_REPOS", (_Repo("episode", rows),))
+
+    assert await index_mod.count_duplicate_keys() == {"episode": 1}
+
+
+async def test_count_duplicate_keys_reports_zero_for_unique_tables(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    rows = [_row("a"), _row("b")]
+    monkeypatch.setattr(index_mod, "ALL_REPOS", (_Repo("episode", rows),))
+
+    assert await index_mod.count_duplicate_keys() == {"episode": 0}