From 510e5000fa81b356237d049dc7886d5850a8bf77 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nicol=C3=B2=20Boschi?= Date: Tue, 21 Jul 2026 15:48:03 +0200 Subject: [PATCH] fix(vector-index): repair per-bank index coverage after restore/upgrade (#2645) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per-(bank, fact_type) partial vector indexes are created only at fresh-bank creation. A bank populated outside that path (logical restore, cross-version upgrade, extension switch) never gets them, so its recall silently falls back to the global index + post-filter — slower and under-returning (~0.63-0.72 recall@10 measured by the reporter). Two fixes: - import-bank: create the per-bank indexes explicitly after restoring the banks row. The prior get_or_create_bank_profile call was a no-op here (the row already exists, so it takes the SELECT branch), leaving every restored bank uncovered. - hindsight-admin repair-bank (--bank ID | --all): re-runnable operator escape hatch for the out-of-app routes (raw pg_dump restore, extension switch) that a one-time migration can't cover (a restore carries alembic_version at head, so the migration is already stamped). Detects missing OR invalid coverage (INVALID leftovers / drifted access method count as missing, unlike a name-only check) and rebuilds with CREATE INDEX CONCURRENTLY off any txn. Idempotent; concurrency handled by idempotency, not advisory locks. Deliberately excludes the boot/periodic background reconcile and retain-path self-heal: a bank restored and only ever read stays degraded until an operator runs repair-bank. That background layer can be a follow-up. --- hindsight-api-slim/hindsight_api/admin/cli.py | 130 +++++++ .../hindsight_api/engine/transfer/importer.py | 13 +- .../engine/vector_index_health.py | 216 ++++++++++++ .../tests/test_repair_bank_vector_indexes.py | 316 ++++++++++++++++++ hindsight-docs/docs/developer/admin-cli.md | 38 +++ .../references/developer/admin-cli.md | 38 +++ .../references/developer/configuration.md | 1 + 7 files changed, 749 insertions(+), 3 deletions(-) create mode 100644 hindsight-api-slim/hindsight_api/engine/vector_index_health.py create mode 100644 hindsight-api-slim/tests/test_repair_bank_vector_indexes.py diff --git a/hindsight-api-slim/hindsight_api/admin/cli.py b/hindsight-api-slim/hindsight_api/admin/cli.py index 69c0712068..ee269fdb98 100644 --- a/hindsight-api-slim/hindsight_api/admin/cli.py +++ b/hindsight-api-slim/hindsight_api/admin/cli.py @@ -18,8 +18,10 @@ from ..config import DEFAULT_DATABASE_SCHEMA, HindsightConfig from ..engine.memory_engine import _current_schema +from ..engine.retain.bank_utils import _vector_index_clause from ..engine.schema import fq_table_explicit as _fq_table from ..engine.transfer import export_bank +from ..engine.vector_index_health import SchemaVectorIndexResult, repair_vector_indexes from ..extensions import TenantExtension, load_extension from ..pg0 import parse_pg0_url, resolve_database_url @@ -357,6 +359,134 @@ def run_db_migration( typer.echo(f"Database migrations completed successfully for {len(schemas)} schema(s)") +async def _resolve_schemas(base_schema: str | None) -> list[str]: + """Base schema plus every discovered tenant schema, de-duplicated in order.""" + schemas = [base_schema or DEFAULT_DATABASE_SCHEMA] + tenant_extension = load_extension("TENANT", TenantExtension) + if tenant_extension: + tenants = await tenant_extension.list_tenants() + schemas.extend(tenant.schema for tenant in tenants if tenant.schema) + return list(dict.fromkeys(schemas)) + + +async def _run_repair_bank( + db_url: str, + *, + base_schema: str, + schema: str | None, + bank_id: str | None, + dry_run: bool, +) -> list[SchemaVectorIndexResult]: + """Reconcile per-(bank, fact_type) vector index coverage over a raw connection. + + A single autocommit connection is used because ``CREATE INDEX CONCURRENTLY`` + (used by ``repair_vector_indexes``) cannot run inside a transaction block. + """ + schemas = [schema] if schema else await _resolve_schemas(base_schema) + index_clause = _vector_index_clause() + # Guarded by the command, but assert so this helper is never called for a + # backend without per-bank indexes. + assert index_clause is not None + + conn = await _admin_connect(db_url) + try: + results = await repair_vector_indexes(conn, schemas, index_clause, dry_run=dry_run, bank_id=bank_id) + for result in results: + typer.echo( + f" schema '{result.schema}': {result.banks_scanned} bank(s) scanned, " + f"{result.already_present} present, {result.created} created, " + f"{result.skipped} to-create (dry-run), {result.failed} failed" + ) + return results + finally: + await conn.close() + + +@app.command(name="repair-bank") +def repair_bank( + bank_id: str | None = typer.Option( + None, + "--bank", + "-b", + help="Bank id to repair. Mutually exclusive with --all.", + ), + all_banks: bool = typer.Option( + False, + "--all", + help="Repair every bank in the base schema and all discovered tenant schemas.", + ), + schema: str | None = typer.Option( + None, + "--schema", + "-s", + help="Limit to a single schema. Defaults to the base schema plus discovered tenant schemas.", + ), + dry_run: bool = typer.Option( + False, + "--dry-run", + help="Report what would be repaired without creating or dropping any index.", + ), +): + """Verify and repair a bank's per-(bank, fact_type) vector index coverage. + + Per-bank partial vector indexes are created when a bank is first created + (instant on an empty bank). Banks that arrive populated — via logical + restore, a cross-version upgrade, or a vector-extension switch — never hit + that path, so their recall silently falls back to a global index + + post-filter (slower, under-returning). This command detects missing OR + invalid coverage (an INVALID leftover or an index whose access method + drifted counts as missing) and rebuilds it with CREATE INDEX CONCURRENTLY, + so it never blocks the live fleet. Idempotent and safe to re-run — the + escape hatch after a restore, upgrade, or backend switch. + """ + if bool(bank_id) == all_banks: + typer.echo("Error: pass exactly one of --bank or --all.", err=True) + raise typer.Exit(2) + + config = HindsightConfig.from_env() + if not config.database_url: + typer.echo("Error: Database URL not configured.", err=True) + typer.echo("Set HINDSIGHT_API_DATABASE_URL environment variable.", err=True) + raise typer.Exit(1) + + # Backend guard: backends with a single global vector index (AlloyDB ScaNN, + # Oracle) have no per-bank indexes to repair. + if _vector_index_clause() is None: + typer.echo("Configured vector backend does not use per-bank vector indexes — nothing to repair.") + return + + target = f"bank '{bank_id}'" if bank_id else "all banks" + scope = f"schema '{schema}'" if schema else "base schema and all discovered tenant schemas" + typer.echo(f"Repairing per-bank vector indexes for {target} across {scope}...") + if dry_run: + typer.echo("Dry run: no indexes will be created or dropped.") + + results = asyncio.run( + _run_repair_bank( + config.database_url, + base_schema=config.database_schema, + schema=schema, + bank_id=bank_id, + dry_run=dry_run, + ) + ) + + total_banks = sum(r.banks_scanned for r in results) + total_present = sum(r.already_present for r in results) + total_created = sum(r.created for r in results) + total_skipped = sum(r.skipped for r in results) + total_failed = sum(r.failed for r in results) + typer.echo( + f"Done: {len(results)} schema(s), {total_banks} bank(s) scanned, " + f"{total_present} already present, {total_created} created, " + f"{total_skipped} to-create (dry-run), {total_failed} failed" + ) + if total_failed: + failed_names = [name for r in results for name in r.failed_indexes] + typer.echo(f"Failed indexes (dropped, retry with a re-run): {', '.join(failed_names)}", err=True) + raise typer.Exit(1) + + async def _run_export_bank(db_url: str, bank_id: str, output: Path, schema: str, include_history: bool) -> int: """Export a whole bank to a ZIP archive.""" conn = await _admin_connect(db_url) diff --git a/hindsight-api-slim/hindsight_api/engine/transfer/importer.py b/hindsight-api-slim/hindsight_api/engine/transfer/importer.py index 33cca38df8..202c45d297 100644 --- a/hindsight-api-slim/hindsight_api/engine/transfer/importer.py +++ b/hindsight-api-slim/hindsight_api/engine/transfer/importer.py @@ -405,9 +405,16 @@ async def import_bank( parsed.bank_rows.get("banks", []), bank_rows_json_encoding=bank_rows_json_encoding, ) - # Ensure the bank's per-bank vector indexes exist (no-op for global-index - # extensions); idempotent and keeps the restored banks row (ON CONFLICT DO NOTHING). - await bank_utils.get_or_create_bank_profile(backend, bank_id) + # The restored banks row bypasses the fresh-INSERT gate that normally + # creates per-bank vector indexes, so create them explicitly here while + # the bank is still empty (facts are imported below, so the build is + # instant). get_or_create_bank_profile would NOT do this: the row now + # exists, so it takes the SELECT branch and skips index creation — + # leaving the restored bank falling back to the global index + + # post-filter (slower, under-returning recall). See #2645. + internal_id = await conn.fetchval(f"SELECT internal_id FROM {fq_table('banks')} WHERE bank_id = $1", bank_id) + if internal_id is not None: + await bank_utils.create_bank_vector_indexes(conn, bank_id, str(internal_id), ops=ops) doc_result = await import_documents( backend=backend, diff --git a/hindsight-api-slim/hindsight_api/engine/vector_index_health.py b/hindsight-api-slim/hindsight_api/engine/vector_index_health.py new file mode 100644 index 0000000000..78c35b1d0c --- /dev/null +++ b/hindsight-api-slim/hindsight_api/engine/vector_index_health.py @@ -0,0 +1,216 @@ +"""Per-bank vector index coverage checks and repair. + +Per-(bank, fact_type) partial vector indexes are created only when a bank is +first created (instant on an empty bank). A bank that becomes *populated* +outside that fresh-INSERT path — via a logical restore, a cross-version upgrade, +or a vector-extension switch (e.g. ScaNN→pgvector) — never gets them, so its +bank-scoped recall silently falls back to the global index + post-filter, which +is both slower and under-returns results. See issue #2645. + +This module is the shared engine for detecting and repairing that gap. It is +driven by the ``repair-bank`` admin command; the build always uses +``CREATE INDEX CONCURRENTLY`` on a raw autocommit connection so it never takes +``ACCESS EXCLUSIVE`` on the shared ``memory_units`` table. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import Any + +from .retain.bank_utils import _BANK_INDEX_FACT_TYPES, _bank_index_name + +logger = logging.getLogger(__name__) + +# Postgres renders the partial predicate of an indexdef with parenthesized +# comparison operands and an explicit ::text cast, e.g. +# `... WHERE ((fact_type = 'world'::text) AND (bank_id = 'b1'::text))`. +# fact_type is emitted first (it is written first in the CREATE INDEX). Match +# that exact rendering so a mere name collision never counts as healthy. +_BANK_INDEX_PARTIAL_SUFFIX = " WHERE ((fact_type = " + +# Access methods that legitimately back a per-(bank, fact_type) partial index. +# An index whose access method drifted after a backend switch does not match, +# so the health check treats it as unhealthy (rebuild). +_SUPPORTED_INDEX_AM: tuple[str, ...] = ( + "btree", + "gin", + "gist", + "hnsw", + "ivfflat", + "diskann", + "vchordrq", +) + + +@dataclass +class SchemaVectorIndexResult: + """Per-schema outcome of a vector-index repair pass.""" + + schema: str + banks_scanned: int = 0 + already_present: int = 0 + created: int = 0 + skipped: int = 0 # would-create, reported under --dry-run + failed: int = 0 + failed_indexes: list[str] = field(default_factory=list) + + +def _quote_identifier(value: str) -> str: + return '"' + value.replace('"', '""') + '"' + + +async def _index_health(conn: Any, schema: str, index_names: list[str]) -> dict[str, bool]: + """Return valid-and-usable state for each requested index in one query. + + Health requires the index to be valid AND ready, defined over the expected + ``memory_units`` table, to use a supported access method, and to carry our + partial predicate. A name-only match is *not* enough: an INVALID leftover + (from an interrupted concurrent build) or an index whose access method + drifted after a backend switch must count as unhealthy so it is rebuilt — + ``pg_indexes``/``IF NOT EXISTS`` alone would silently treat those as present. + """ + if not index_names: + return {} + rows = await conn.fetch( + """ + SELECT c.relname AS index_name, + (i.indisvalid AND i.indisready + AND t.relname = 'memory_units' + AND am.amname = ANY($3::text[]) + AND pg_get_indexdef(i.indexrelid) LIKE $4 + ) AS healthy + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + JOIN pg_index i ON i.indexrelid = c.oid + JOIN pg_class t ON t.oid = i.indrelid + JOIN pg_am am ON am.oid = c.relam + WHERE n.nspname = $1 AND c.relname = ANY($2::text[]) + """, + schema, + index_names, + list(_SUPPORTED_INDEX_AM), + "%" + _BANK_INDEX_PARTIAL_SUFFIX + "%", + ) + return {row["index_name"]: bool(row["healthy"]) for row in rows} + + +async def _repair_schema( + conn: Any, + schema: str, + index_clause: str, + *, + dry_run: bool, + bank_id: str | None, +) -> SchemaVectorIndexResult: + result = SchemaVectorIndexResult(schema=schema) + qschema = _quote_identifier(schema) + + if bank_id is not None: + banks = await conn.fetch( + f"SELECT bank_id, internal_id FROM {qschema}.banks WHERE bank_id = $1", # noqa: S608 — schema is a quoted identifier + bank_id, + ) + else: + banks = await conn.fetch(f"SELECT bank_id, internal_id FROM {qschema}.banks ORDER BY bank_id") # noqa: S608 + result.banks_scanned = len(banks) + + # Resolve expected index names for every bank, then check them all in one + # catalog query rather than one round-trip per index. + expected_by_bank: list[tuple[str, dict[str, str]]] = [] + all_index_names: list[str] = [] + for bank in banks: + expected = {ft: _bank_index_name(ft, str(bank["internal_id"])) for ft in _BANK_INDEX_FACT_TYPES} + expected_by_bank.append((bank["bank_id"], expected)) + all_index_names.extend(expected.values()) + health = await _index_health(conn, schema, all_index_names) + + for bid, expected in expected_by_bank: + # Render the bank_id literal server-side so escaping does not depend on + # standard_conforming_strings (the predicate is inlined into the DDL). + bank_id_literal = await conn.fetchval("SELECT quote_literal($1::text)", bid) + for ft in _BANK_INDEX_FACT_TYPES: + index_name = expected[ft] + healthy = health.get(index_name) + if healthy is True: + result.already_present += 1 + continue + if dry_run: + result.skipped += 1 + continue + + qindex = _quote_identifier(index_name) + qualified = f"{qschema}.{qindex}" + try: + # An unhealthy-but-present index (INVALID leftover, wrong access + # method) must be dropped first — IF NOT EXISTS cannot repair it. + if healthy is False: + await conn.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {qualified}") + await conn.execute( + f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {qindex} " + f"ON {qschema}.memory_units {index_clause} " + f"WHERE fact_type = '{ft}' AND bank_id = {bank_id_literal}" + ) + result.created += 1 + except Exception as exc: # noqa: BLE001 — one failed index must not abort the rest + result.failed += 1 + result.failed_indexes.append(qualified) + logger.warning( + "Failed to repair vector index %s (bank=%s, fact_type=%s): %s — " + "dropping the invalid leftover so a re-run can retry.", + qualified, + bid, + ft, + exc, + ) + # A failed concurrent build leaves an INVALID index behind that + # would shadow the good one; drop it so a re-run retries cleanly. + try: + await conn.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {qualified}") + except Exception as cleanup_exc: # noqa: BLE001 + logger.warning("Cleanup DROP INDEX for %s also failed: %s", qualified, cleanup_exc) + + return result + + +async def _safe_repair_schema( + conn: Any, + schema: str, + index_clause: str, + *, + dry_run: bool, + bank_id: str | None, +) -> SchemaVectorIndexResult: + try: + return await _repair_schema(conn, schema, index_clause, dry_run=dry_run, bank_id=bank_id) + except Exception as exc: # noqa: BLE001 — one bad schema must not abort the whole sweep + logger.warning("Vector index repair aborted for schema %s: %s", schema, exc) + return SchemaVectorIndexResult(schema=schema, failed=1, failed_indexes=[f"{schema}."]) + + +async def repair_vector_indexes( + conn: Any, + schemas: list[str], + index_clause: str, + *, + dry_run: bool = False, + bank_id: str | None = None, +) -> list[SchemaVectorIndexResult]: + """Rebuild missing or invalid per-bank vector indexes across ``schemas``. + + ``conn`` must be a raw autocommit PostgreSQL connection: ``CREATE INDEX + CONCURRENTLY`` cannot run inside a transaction block. When ``bank_id`` is + given, only that bank is reconciled (in each schema); otherwise every bank + is scanned. + + Concurrency is handled by idempotency, not a lock (project rule: no advisory + locks — they are unreliable behind connection poolers). Every build is + ``CREATE INDEX CONCURRENTLY IF NOT EXISTS`` guarded by a valid/ready health + check, so a second concurrent run is a no-op on already-built indexes; if two + runs race the *same* missing index, Postgres rejects one build and the + per-index handler drops the leftover so a re-run converges cleanly. + """ + return [ + await _safe_repair_schema(conn, schema, index_clause, dry_run=dry_run, bank_id=bank_id) for schema in schemas + ] diff --git a/hindsight-api-slim/tests/test_repair_bank_vector_indexes.py b/hindsight-api-slim/tests/test_repair_bank_vector_indexes.py new file mode 100644 index 0000000000..950a48ae6b --- /dev/null +++ b/hindsight-api-slim/tests/test_repair_bank_vector_indexes.py @@ -0,0 +1,316 @@ +"""Regression tests for per-bank vector index coverage repair (issue #2645). + +Per-(bank, fact_type) partial vector indexes are only created at fresh-bank +creation time. Banks that arrive already populated — via logical restore, a +cross-version upgrade, or a vector-extension switch — never hit that path, so +their recall silently falls back to a global index + post-filter (slower, +~30% recall@10 miss). + +Two things are proven here: + +* the `import-bank` leak is plugged — a restored bank gets its per-bank indexes; +* the `repair-bank` command is the re-runnable escape hatch — it rebuilds + missing OR invalid coverage (a name-colliding index that lacks the partial + predicate counts as invalid and is rebuilt, unlike a name-only check), is a + no-op on non-per-bank backends, and validates its target flags. + +Everything asserted is deterministic (index presence/shape via the catalog) — +no LLM is needed, so memory_units are inserted directly. +""" + +import uuid + +import pytest + +from hindsight_api import RequestContext +from hindsight_api.admin import cli +from hindsight_api.admin.cli import _run_repair_bank +from hindsight_api.engine.db_utils import acquire_with_retry +from hindsight_api.engine.memory_engine import MemoryEngine +from hindsight_api.engine.retain.bank_utils import _BANK_INDEX_FACT_TYPES, _bank_index_name +from hindsight_api.engine.transfer import export_bank + +_TEST_SCHEMA = "public" + + +async def _bank_internal_id(conn, bank_id: str) -> str: + row = await conn.fetchrow("SELECT internal_id FROM banks WHERE bank_id = $1", bank_id) + assert row is not None, f"bank {bank_id} not found" + return str(row["internal_id"]) + + +async def _index_exists(conn, idx_name: str) -> bool: + return bool( + await conn.fetchval( + "SELECT 1 FROM pg_indexes WHERE schemaname = $1 AND indexname = $2", + _TEST_SCHEMA, + idx_name, + ) + ) + + +async def _index_is_partial_vector(conn, idx_name: str) -> bool: + """True only if the index carries our per-(bank, fact_type) partial predicate.""" + indexdef = await conn.fetchval( + "SELECT pg_get_indexdef(c.oid) FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace " + "WHERE n.nspname = $1 AND c.relname = $2", + _TEST_SCHEMA, + idx_name, + ) + return bool(indexdef) and "WHERE ((fact_type = " in indexdef + + +async def _expected_index_names(conn, bank_id: str) -> list[str]: + internal_id = await _bank_internal_id(conn, bank_id) + return [_bank_index_name(ft, internal_id) for ft in _BANK_INDEX_FACT_TYPES] + + +async def _seed_bank(memory: MemoryEngine, request_context: RequestContext) -> str: + """Create a bank (auto-creates internal_id + per-bank indexes) and populate it.""" + bank_id = f"test-repair-{uuid.uuid4().hex[:8]}" + await memory.get_bank_profile(bank_id=bank_id, request_context=request_context) + + backend = await memory._get_backend() + async with acquire_with_retry(backend) as conn: + for ft in _BANK_INDEX_FACT_TYPES: + await conn.execute( + """ + INSERT INTO memory_units (id, bank_id, text, fact_type, event_date, created_at, updated_at) + VALUES ($1, $2, $3, $4, NOW(), NOW(), NOW()) + """, + uuid.uuid4(), + bank_id, + f"seed {ft} fact", + ft, + ) + return bank_id + + +async def _drop_bank_indexes(conn, bank_id: str) -> list[str]: + """Drop every per-(bank, fact_type) index to simulate the restore/upgrade gap.""" + names = await _expected_index_names(conn, bank_id) + for name in names: + await conn.execute(f"DROP INDEX IF EXISTS {_TEST_SCHEMA}.{name}") + return names + + +class TestRepairBankCommand: + @pytest.mark.asyncio + async def test_repair_single_bank_recreates_dropped_indexes( + self, memory: MemoryEngine, request_context: RequestContext, pg0_db_url: str + ): + bank_id = await _seed_bank(memory, request_context) + backend = await memory._get_backend() + try: + async with acquire_with_retry(backend) as conn: + names = await _drop_bank_indexes(conn, bank_id) + for name in names: + assert not await _index_exists(conn, name), f"{name} should be dropped" + + results = await _run_repair_bank( + pg0_db_url, base_schema=_TEST_SCHEMA, schema=_TEST_SCHEMA, bank_id=bank_id, dry_run=False + ) + + assert len(results) == 1 + result = results[0] + assert result.failed == 0 + assert result.created >= len(_BANK_INDEX_FACT_TYPES) + # Only the targeted bank was scanned. + assert result.banks_scanned == 1 + + async with acquire_with_retry(backend) as conn: + for name in names: + assert await _index_is_partial_vector(conn, name), f"{name} should be rebuilt as a partial index" + finally: + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_repair_all_scans_every_bank( + self, memory: MemoryEngine, request_context: RequestContext, pg0_db_url: str + ): + bank_id = await _seed_bank(memory, request_context) + backend = await memory._get_backend() + try: + async with acquire_with_retry(backend) as conn: + names = await _drop_bank_indexes(conn, bank_id) + + results = await _run_repair_bank( + pg0_db_url, base_schema=_TEST_SCHEMA, schema=_TEST_SCHEMA, bank_id=None, dry_run=False + ) + + result = results[0] + # --all scans more than just our bank (other banks may exist in the shared db). + assert result.banks_scanned >= 1 + assert result.created >= len(_BANK_INDEX_FACT_TYPES) + async with acquire_with_retry(backend) as conn: + for name in names: + assert await _index_exists(conn, name), f"{name} should be rebuilt" + finally: + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_invalid_shape_index_is_rebuilt( + self, memory: MemoryEngine, request_context: RequestContext, pg0_db_url: str + ): + """A name-colliding index that lacks the partial predicate is unhealthy → rebuilt. + + This is the differentiator over a name-only existence check (which would + treat the collision — or a stale INVALID leftover — as 'already present' + and never repair it). + """ + bank_id = await _seed_bank(memory, request_context) + backend = await memory._get_backend() + try: + async with acquire_with_retry(backend) as conn: + names = await _drop_bank_indexes(conn, bank_id) + # Recreate the FIRST expected index name with the WRONG definition: + # a plain btree with no partial predicate. Name matches, shape does not. + bogus = names[0] + await conn.execute(f"CREATE INDEX {bogus} ON memory_units (bank_id)") + assert await _index_exists(conn, bogus) + assert not await _index_is_partial_vector(conn, bogus) + + results = await _run_repair_bank( + pg0_db_url, base_schema=_TEST_SCHEMA, schema=_TEST_SCHEMA, bank_id=bank_id, dry_run=False + ) + assert results[0].failed == 0 + + async with acquire_with_retry(backend) as conn: + for name in names: + assert await _index_is_partial_vector(conn, name), f"{name} should now be the partial vector index" + finally: + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_dry_run_creates_nothing( + self, memory: MemoryEngine, request_context: RequestContext, pg0_db_url: str + ): + bank_id = await _seed_bank(memory, request_context) + backend = await memory._get_backend() + try: + async with acquire_with_retry(backend) as conn: + names = await _drop_bank_indexes(conn, bank_id) + + results = await _run_repair_bank( + pg0_db_url, base_schema=_TEST_SCHEMA, schema=_TEST_SCHEMA, bank_id=bank_id, dry_run=True + ) + assert results[0].created == 0 + assert results[0].skipped >= len(_BANK_INDEX_FACT_TYPES) + + async with acquire_with_retry(backend) as conn: + for name in names: + assert not await _index_exists(conn, name), f"{name} must NOT exist after dry-run" + finally: + await memory.delete_bank(bank_id, request_context=request_context) + + @pytest.mark.asyncio + async def test_rerun_is_idempotent(self, memory: MemoryEngine, request_context: RequestContext, pg0_db_url: str): + bank_id = await _seed_bank(memory, request_context) + backend = await memory._get_backend() + try: + async with acquire_with_retry(backend) as conn: + names = await _drop_bank_indexes(conn, bank_id) + + first = ( + await _run_repair_bank( + pg0_db_url, base_schema=_TEST_SCHEMA, schema=_TEST_SCHEMA, bank_id=bank_id, dry_run=False + ) + )[0] + assert first.created >= len(_BANK_INDEX_FACT_TYPES) + + second = ( + await _run_repair_bank( + pg0_db_url, base_schema=_TEST_SCHEMA, schema=_TEST_SCHEMA, bank_id=bank_id, dry_run=False + ) + )[0] + assert second.created == 0 + assert second.failed == 0 + assert second.already_present >= len(_BANK_INDEX_FACT_TYPES) + + async with acquire_with_retry(backend) as conn: + for name in names: + count = await conn.fetchval( + "SELECT count(*) FROM pg_indexes WHERE schemaname = $1 AND indexname = $2", + _TEST_SCHEMA, + name, + ) + assert count == 1, f"{name} should exist exactly once" + finally: + await memory.delete_bank(bank_id, request_context=request_context) + + def test_requires_exactly_one_target(self): + """Neither / both of --bank and --all is a usage error (exit 2).""" + from typer.testing import CliRunner + + runner = CliRunner() + neither = runner.invoke(cli.app, ["repair-bank"]) + assert neither.exit_code == 2, neither.output + both = runner.invoke(cli.app, ["repair-bank", "--bank", "b1", "--all"]) + assert both.exit_code == 2, both.output + + @pytest.mark.asyncio + async def test_backend_without_per_bank_indexes_is_noop( + self, memory: MemoryEngine, request_context: RequestContext, pg0_db_url: str, monkeypatch + ): + bank_id = await _seed_bank(memory, request_context) + backend = await memory._get_backend() + try: + async with acquire_with_retry(backend) as conn: + names = await _drop_bank_indexes(conn, bank_id) + + # Simulate a backend (AlloyDB ScaNN / Oracle) with a single global index. + monkeypatch.setattr(cli, "_vector_index_clause", lambda: None) + + from typer.testing import CliRunner + + runner = CliRunner() + result = runner.invoke(cli.app, ["repair-bank", "--all"]) + assert result.exit_code == 0, result.output + assert "does not use per-bank vector indexes" in result.output + + async with acquire_with_retry(backend) as conn: + for name in names: + assert not await _index_exists(conn, name), f"{name} must NOT exist for no-op backend" + finally: + await memory.delete_bank(bank_id, request_context=request_context) + + +class TestImportBankCreatesIndexes: + @pytest.mark.asyncio + async def test_import_bank_creates_per_bank_indexes(self, memory: MemoryEngine, request_context: RequestContext): + """The import-bank leak (#2645): a restored bank must get its per-bank indexes. + + export → delete → import round-trip, then assert the per-bank partial + indexes exist for the restored bank. Before the fix, import took the + SELECT branch of get_or_create_bank_profile and skipped index creation. + """ + bank_id = f"test-import-{uuid.uuid4().hex[:8]}" + backend = await memory._get_backend() + try: + # Populate directly (deterministic; no LLM) so the bank has real rows. + await memory.get_bank_profile(bank_id=bank_id, request_context=request_context) + async with acquire_with_retry(backend) as conn: + for ft in _BANK_INDEX_FACT_TYPES: + await conn.execute( + """ + INSERT INTO memory_units (id, bank_id, text, fact_type, event_date, created_at, updated_at) + VALUES ($1, $2, $3, $4, NOW(), NOW(), NOW()) + """, + uuid.uuid4(), + bank_id, + f"seed {ft} fact", + ft, + ) + archive = await export_bank(conn, bank_id) + + await memory.delete_bank(bank_id, request_context=request_context) + result = await memory.import_bank_async(archive, request_context) + assert result.bank_id == bank_id + + async with acquire_with_retry(backend) as conn: + for name in await _expected_index_names(conn, bank_id): + assert await _index_is_partial_vector(conn, name), ( + f"{name} should exist after import-bank (the restore leak, #2645)" + ) + finally: + await memory.delete_bank(bank_id, request_context=request_context) diff --git a/hindsight-docs/docs/developer/admin-cli.md b/hindsight-docs/docs/developer/admin-cli.md index 1f19486f63..20aee59415 100644 --- a/hindsight-docs/docs/developer/admin-cli.md +++ b/hindsight-docs/docs/developer/admin-cli.md @@ -68,6 +68,44 @@ To disable automatic migrations on API startup, set `HINDSIGHT_API_RUN_MIGRATION --- +### repair-bank + +Verify and repair a bank's per-`(bank, fact_type)` vector index coverage. + +These partial indexes are normally created when a bank is first created (instant on an empty bank), and PostgreSQL maintains them incrementally as the bank grows. A bank that becomes **populated outside that create-time path** — via a logical restore, a cross-version upgrade, or a vector-extension switch — never gets them, so its bank-scoped recall silently falls back to a global index + post-filter. That fallback is both **slower** and can **under-return** results (the approximate nearest-neighbour search draws its candidates from every bank, then filters to yours afterward). + +Run this after any of those events to restore full coverage. It detects **missing or invalid** coverage (an INVALID leftover from an interrupted build, or an index whose type drifted after a backend switch, both count as missing) and rebuilds it with `CREATE INDEX CONCURRENTLY`, so it never blocks live retain/recall/consolidation. It is idempotent and safe to re-run. + +```bash +hindsight-admin repair-bank (--bank BANK_ID | --all) [OPTIONS] +``` + +**Options:** + +| Option | Description | Default | +|--------|-------------|---------| +| `--bank`, `-b` | Bank id to repair. Mutually exclusive with `--all`. | — | +| `--all` | Repair every bank in the base schema and all discovered tenant schemas. | — | +| `--schema`, `-s` | Limit to a single schema. | All schemas | +| `--dry-run` | Report what would be repaired without creating or dropping any index. | Off | + +Exactly one of `--bank` or `--all` is required. No-op for backends that use a single global vector index (AlloyDB ScaNN, Oracle). It is idempotent — safe to re-run and safe to run while the API is serving traffic. + +**Examples:** + +```bash +# See which banks are missing coverage without changing anything +hindsight-admin repair-bank --all --dry-run + +# Repair every bank across all schemas (run once after a restore/upgrade) +hindsight-admin repair-bank --all + +# Repair a single bank +hindsight-admin repair-bank --bank acme-prod +``` + +--- + ### backup Create a backup of all Hindsight data to a zip file. diff --git a/skills/hindsight-docs/references/developer/admin-cli.md b/skills/hindsight-docs/references/developer/admin-cli.md index 1f19486f63..20aee59415 100644 --- a/skills/hindsight-docs/references/developer/admin-cli.md +++ b/skills/hindsight-docs/references/developer/admin-cli.md @@ -68,6 +68,44 @@ To disable automatic migrations on API startup, set `HINDSIGHT_API_RUN_MIGRATION --- +### repair-bank + +Verify and repair a bank's per-`(bank, fact_type)` vector index coverage. + +These partial indexes are normally created when a bank is first created (instant on an empty bank), and PostgreSQL maintains them incrementally as the bank grows. A bank that becomes **populated outside that create-time path** — via a logical restore, a cross-version upgrade, or a vector-extension switch — never gets them, so its bank-scoped recall silently falls back to a global index + post-filter. That fallback is both **slower** and can **under-return** results (the approximate nearest-neighbour search draws its candidates from every bank, then filters to yours afterward). + +Run this after any of those events to restore full coverage. It detects **missing or invalid** coverage (an INVALID leftover from an interrupted build, or an index whose type drifted after a backend switch, both count as missing) and rebuilds it with `CREATE INDEX CONCURRENTLY`, so it never blocks live retain/recall/consolidation. It is idempotent and safe to re-run. + +```bash +hindsight-admin repair-bank (--bank BANK_ID | --all) [OPTIONS] +``` + +**Options:** + +| Option | Description | Default | +|--------|-------------|---------| +| `--bank`, `-b` | Bank id to repair. Mutually exclusive with `--all`. | — | +| `--all` | Repair every bank in the base schema and all discovered tenant schemas. | — | +| `--schema`, `-s` | Limit to a single schema. | All schemas | +| `--dry-run` | Report what would be repaired without creating or dropping any index. | Off | + +Exactly one of `--bank` or `--all` is required. No-op for backends that use a single global vector index (AlloyDB ScaNN, Oracle). It is idempotent — safe to re-run and safe to run while the API is serving traffic. + +**Examples:** + +```bash +# See which banks are missing coverage without changing anything +hindsight-admin repair-bank --all --dry-run + +# Repair every bank across all schemas (run once after a restore/upgrade) +hindsight-admin repair-bank --all + +# Repair a single bank +hindsight-admin repair-bank --bank acme-prod +``` + +--- + ### backup Create a backup of all Hindsight data to a zip file. diff --git a/skills/hindsight-docs/references/developer/configuration.md b/skills/hindsight-docs/references/developer/configuration.md index 3a5c27cece..6d2f259163 100644 --- a/skills/hindsight-docs/references/developer/configuration.md +++ b/skills/hindsight-docs/references/developer/configuration.md @@ -198,6 +198,7 @@ For non-English banks (especially CJK) and the language/extraction-language trad | `HINDSIGHT_API_LLM_GEMINI_SAFETY_SETTINGS` | JSON-encoded list of `{category, threshold}` dicts for Gemini/VertexAI content safety filtering | `null` | | `HINDSIGHT_API_LLM_PROMPT_CACHE_ENABLED` | Reuse the fixed system prefix via the provider's explicit prompt cache, billed at the cached-input rate (Gemini/Vertex `CachedContent`). The cached prefix is shared across all banks and soft-fails to an uncached call. Set to `false` to disable. See [Models](./models#provider-capabilities). | `true` | | `HINDSIGHT_API_REFLECT_PROMPT_CACHE_ENABLED` | For reflect specifically, roll a step-by-step context cache forward through the agent's tool loop so each turn reuses the whole prior conversation (system + tools + all prior tool results) at the cached-input rate instead of only the static prefix. Requires `HINDSIGHT_API_LLM_PROMPT_CACHE_ENABLED`. The per-reflect caches are ephemeral and deleted when the reflect ends. Set to `false` to run reflect uncached while leaving prompt caching on elsewhere. | `true` | +| `HINDSIGHT_API_LLM_DEBUG_DUMP_4XX` | Diagnostic: when enabled, on any LLM `4xx` the provider logs `[LLM_4XX_DUMP]` with the request as actually assembled — the serialized request config (response schema + generation params, message bodies stripped) and length-capped per-message previews — so an otherwise-unreproducible rejected request can be inspected. Wired into all remote providers (Gemini/Vertex, OpenAI-compatible incl. Fireworks/Nous, Anthropic, LiteLLM incl. Router, Codex). Off by default; leave off in normal operation. | `false` | When `HINDSIGHT_API_LLM_PROVIDER=ollama`, Hindsight no longer sends the previous native API default `num_ctx=16384` unless you set it explicitly. To keep the old request behavior, set `HINDSIGHT_API_LLM_OLLAMA_NUM_CTX=16384`; otherwise Ollama uses the model Modelfile or server default.