diff --git a/.env.example b/.env.example index ef33ab247c..9c4431c2d8 100644 --- a/.env.example +++ b/.env.example @@ -107,6 +107,7 @@ HINDSIGHT_API_LOG_LEVEL=info # HINDSIGHT_API_MIGRATION_CONCURRENCY=1 # Tenant schemas to migrate concurrently (PG only, each in its own process; per-schema work stays sequential). Each worker has ~1-2s startup cost + uses ~3 DB connections, so it only pays off with many schemas (tens+) or slow migrations; keep concurrency*3 <= spare max_connections. Default: 1 (sequential). # HINDSIGHT_API_OPERATION_RETENTION_DAYS=30 # Keep terminal operation rows, payloads, and metadata for this many days; 0 disables automatic pruning. # HINDSIGHT_API_OPERATION_CLEANUP_BATCH_SIZE=1000 # Maximum expired terminal rows deleted per tenant schema in each cleanup cycle; must be positive. +# HINDSIGHT_API_VECTOR_INDEX_RECONCILE_INTERVAL_SECONDS=3600 # Rebuild missing/invalid per-bank vector indexes at startup and on this interval; 0 disables. # Vector Extension (Optional - uses pgvector by default) # Options: "pgvector" (default), "vchord", "pgvectorscale" (DiskANN) diff --git a/hindsight-api-slim/hindsight_api/admin/cli.py b/hindsight-api-slim/hindsight_api/admin/cli.py index 8e78b9dc46..aedbb78aa7 100644 --- a/hindsight-api-slim/hindsight_api/admin/cli.py +++ b/hindsight-api-slim/hindsight_api/admin/cli.py @@ -9,7 +9,6 @@ import json import logging import zipfile -from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -18,14 +17,11 @@ import typer from ..config import DEFAULT_DATABASE_SCHEMA, HindsightConfig -from ..engine.memory_engine import _current_schema, fq_table -from ..engine.retain.bank_utils import ( - _BANK_INDEX_FACT_TYPES, - _bank_index_name, - _vector_index_clause, -) +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_reconcile import SchemaVectorIndexReconcileResult, reconcile_vector_indexes from ..extensions import TenantExtension, load_extension from ..pg0 import parse_pg0_url, resolve_database_url @@ -359,108 +355,13 @@ def run_db_migration( typer.echo(f"Database migrations completed successfully for {len(schemas)} schema(s)") -@dataclass -class SchemaBackfillResult: - """Per-schema outcome of the vector-index backfill scan. - - ``created``/``skipped`` are mutually exclusive per missing index: with - ``--dry-run`` a missing index lands in ``skipped``, otherwise a successful - CONCURRENTLY build lands in ``created`` and a failed one in ``failed``. - """ - - schema: str - banks_scanned: int = 0 - already_present: int = 0 - created: int = 0 - skipped: int = 0 - failed: int = 0 - failed_indexes: list[str] = field(default_factory=list) - - -async def _backfill_schema( - conn: asyncpg.Connection, - schema: str, - index_clause: str, - *, - dry_run: bool, -) -> SchemaBackfillResult: - """Reconcile per-(bank, fact_type) partial vector indexes for one schema. - - Runs over a raw autocommit connection (``CREATE INDEX CONCURRENTLY`` cannot - run inside a transaction block). ``_current_schema`` must already be set to - ``schema`` so ``fq_table`` resolves correctly. - """ - result = SchemaBackfillResult(schema=schema) - - banks = await conn.fetch(f"SELECT bank_id, internal_id FROM {fq_table('banks')} ORDER BY bank_id") # noqa: S608 - result.banks_scanned = len(banks) - if not banks: - return result - - mu_table = fq_table("memory_units") - for bank in banks: - bank_id = bank["bank_id"] - internal_id = str(bank["internal_id"]) - escaped = bank_id.replace("'", "''") - for ft in _BANK_INDEX_FACT_TYPES: - idx_name = _bank_index_name(ft, internal_id) - - exists = await conn.fetchval( - "SELECT 1 FROM pg_indexes WHERE schemaname = $1 AND indexname = $2", - schema, - idx_name, - ) - if exists: - result.already_present += 1 - continue - - if dry_run: - result.skipped += 1 - typer.echo(f" [dry-run] would create {schema}.{idx_name} (bank={bank_id}, fact_type={ft})") - continue - - try: - await conn.execute( - f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {idx_name} " - f"ON {mu_table} {index_clause} " - f"WHERE fact_type = '{ft}' AND bank_id = '{escaped}'" - ) - result.created += 1 - typer.echo(f" created {schema}.{idx_name} (bank={bank_id}, fact_type={ft})") - except Exception as exc: # noqa: BLE001 — one failed index must not abort the run - result.failed += 1 - result.failed_indexes.append(f"{schema}.{idx_name}") - # A failed CONCURRENTLY build leaves an INVALID index behind that - # would shadow the good one; drop it so a re-run can retry cleanly. - logger.warning( - "Failed to build vector index %s.%s (bank=%s, fact_type=%s): %s — " - "dropping the invalid leftover so a re-run can retry.", - schema, - idx_name, - bank_id, - ft, - exc, - ) - try: - await conn.execute(f"DROP INDEX IF EXISTS {schema}.{idx_name}") - except Exception as drop_exc: # noqa: BLE001 - logger.warning( - "Cleanup DROP INDEX for %s.%s also failed: %s (manual cleanup may be needed).", - schema, - idx_name, - drop_exc, - ) - - return result - - async def _run_backfill_vector_indexes( db_url: str, schema: str | None = None, base_schema: str = DEFAULT_DATABASE_SCHEMA, *, dry_run: bool = False, -) -> list[SchemaBackfillResult]: +) -> list[SchemaVectorIndexReconcileResult]: """Backfill missing per-(bank, fact_type) partial vector indexes. Iterates one schema (``--schema``) or the base schema plus all discovered @@ -487,17 +388,13 @@ async def _run_backfill_vector_indexes( conn = await _admin_connect(db_url) try: - results: list[SchemaBackfillResult] = [] - for target_schema in schemas: - _current_schema.set(target_schema) - typer.echo(f"Scanning schema '{target_schema}'...") - result = await _backfill_schema(conn, target_schema, index_clause, dry_run=dry_run) + results = await reconcile_vector_indexes(conn, schemas, index_clause, dry_run=dry_run) + for result in results: typer.echo( - f" schema '{target_schema}': {result.banks_scanned} bank(s) scanned, " + 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" ) - results.append(result) return results finally: await conn.close() @@ -549,13 +446,21 @@ def backfill_vector_indexes( results = asyncio.run( _run_backfill_vector_indexes( - config.database_url, + config.migration_database_url or config.database_url, schema=schema, base_schema=config.database_schema, dry_run=dry_run, ) ) + if results and all(r.skipped_lock_busy for r in results): + typer.echo( + "Skipped: another instance holds the vector index reconciliation advisory lock; " + "wait for it to finish and re-run.", + err=True, + ) + raise typer.Exit(75) # EX_TEMPFAIL — operator-friendly retry code + 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) diff --git a/hindsight-api-slim/hindsight_api/config.py b/hindsight-api-slim/hindsight_api/config.py index c6490115f4..4629887881 100644 --- a/hindsight-api-slim/hindsight_api/config.py +++ b/hindsight-api-slim/hindsight_api/config.py @@ -674,6 +674,7 @@ def _resolve_operation_temperature(operation_env: str, default: float) -> float # Background maintenance settings ENV_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS = "HINDSIGHT_API_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS" ENV_MENTAL_MODEL_REFRESH_TICK_SECONDS = "HINDSIGHT_API_MENTAL_MODEL_REFRESH_TICK_SECONDS" +ENV_VECTOR_INDEX_RECONCILE_INTERVAL_SECONDS = "HINDSIGHT_API_VECTOR_INDEX_RECONCILE_INTERVAL_SECONDS" # Disposition settings ENV_DISPOSITION_SKEPTICISM = "HINDSIGHT_API_DISPOSITION_SKEPTICISM" @@ -1116,6 +1117,8 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]: # facts (e.g. after a consolidation operation failed terminally and left them unscheduled). # 0 disables the reconcile sweep. DEFAULT_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS = 300 +# Reconcile per-bank vector indexes at startup, then hourly. 0 disables the sweep. +DEFAULT_VECTOR_INDEX_RECONCILE_INTERVAL_SECONDS = 3600 # How often the maintenance loop checks for cron-scheduled mental models that are # due for a refresh. This is the *check* cadence; the actual schedule is the @@ -1968,15 +1971,18 @@ class HindsightConfig: # Interval for the periodic sweep that re-schedules consolidation for banks with # eligible-but-unscheduled facts. 0 = disabled. consolidation_reconcile_interval_seconds: int + # Reconcile missing per-bank vector indexes at startup and periodically. 0 = disabled. + # New keyword-only field so existing positional callers keep working unchanged. + vector_index_reconcile_interval_seconds: int = field(default=0, kw_only=True) # How often the maintenance loop checks for cron-scheduled mental models due for # refresh (the per-model schedule lives in the mental model trigger). 0 = disabled. mental_model_refresh_tick_seconds: int # Webhook configuration (static - server-level only, not per-bank) - webhook_url: str | None # Global webhook URL (None = disabled) - webhook_secret: str | None # HMAC signing secret (None = unsigned) - webhook_event_types: list[str] # Event types to deliver globally - webhook_delivery_poll_interval_seconds: int # How often the delivery worker polls + webhook_url: str | None = None + webhook_secret: str | None = None + webhook_event_types: list[str] = field(default_factory=list) + webhook_delivery_poll_interval_seconds: int = 1 # Defaulted fields (source-compatible additions — existing direct constructor callers keep working). # Keep at the end of the dataclass; Python forbids non-default fields after default fields. @@ -3064,6 +3070,12 @@ def from_env(cls) -> "HindsightConfig": str(DEFAULT_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS), ) ), + vector_index_reconcile_interval_seconds=int( + os.getenv( + ENV_VECTOR_INDEX_RECONCILE_INTERVAL_SECONDS, + str(DEFAULT_VECTOR_INDEX_RECONCILE_INTERVAL_SECONDS), + ) + ), mental_model_refresh_tick_seconds=int( os.getenv( ENV_MENTAL_MODEL_REFRESH_TICK_SECONDS, diff --git a/hindsight-api-slim/hindsight_api/engine/maintenance.py b/hindsight-api-slim/hindsight_api/engine/maintenance.py index cb6f2f0de6..fb000b7155 100644 --- a/hindsight-api-slim/hindsight_api/engine/maintenance.py +++ b/hindsight-api-slim/hindsight_api/engine/maintenance.py @@ -11,6 +11,10 @@ consolidation operation failed terminally and left them with ``consolidated_at IS NULL AND consolidation_failed_at IS NULL`` and nothing to re-trigger them. +- **Vector index reconcile** (configurable, default hourly): at startup and on + its interval, rebuild missing or invalid per-bank vector indexes with + ``CREATE INDEX CONCURRENTLY``. This repairs coverage after logical restores, + cross-version upgrades, and vector-backend switches without blocking writes. - **Scheduled mental model refresh** (configurable check cadence, default 60s): refresh mental models whose ``trigger.refresh_cron`` schedule is due, but only when the model is stale (new memories in its scope since its last refresh), so @@ -31,14 +35,18 @@ import asyncio import logging import time -from collections.abc import Coroutine +from collections.abc import AsyncIterator, Coroutine +from contextlib import asynccontextmanager from datetime import datetime, timezone from typing import TYPE_CHECKING, Any from ..config import HindsightConfig, get_config from ..models import RequestContext +from ..pg0 import resolve_database_url from .db_utils import acquire_with_retry +from .retain.bank_utils import _vector_index_clause from .schema import _is_oracle, fq_table +from .vector_index_reconcile import reconcile_vector_indexes if TYPE_CHECKING: from .memory_engine import MemoryEngine @@ -51,6 +59,44 @@ _RETENTION_INTERVAL_SECONDS = 3600 +@asynccontextmanager +async def _raw_postgres_connection(engine: "MemoryEngine") -> AsyncIterator[Any]: + """Open an autocommit connection for ``CREATE INDEX CONCURRENTLY``. + + The session-level advisory lock used to gate reconciliation only works when + every call against this connection lives in a single PostgreSQL session. + PgBouncer's transaction-mode pool forces ``UNLOCK`` onto a different + backend than the one that acquired the lock, defeating serialization. + Refuse to reconcile when ``DATABASE_URL`` looks like a transaction pooler + unless ``MIGRATION_DATABASE_URL`` is explicitly set to a direct URL. + """ + import asyncpg + + config = get_config() + db_url = config.migration_database_url or engine.db_url + if not db_url: + raise RuntimeError("Database URL is not available for vector index reconciliation") + db_url_str = db_url.lower() + if ( + ":6543" in db_url_str + or "pool_mode=transaction" in db_url_str + or "/pgbouncer" in db_url_str + or "pgbouncer-" in db_url_str + ): + if not config.migration_database_url: + raise RuntimeError( + "HINDSIGHT_API_DATABASE_URL points to a transaction pooler; " + "vector index reconciliation requires a direct PostgreSQL " + "connection. Set HINDSIGHT_API_MIGRATION_DATABASE_URL to the " + "direct libpq URL." + ) + conn = await asyncpg.connect(await resolve_database_url(db_url)) + try: + yield conn + finally: + await conn.close() + + class MaintenanceLoop: """Owns the single periodic maintenance task for a :class:`MemoryEngine`.""" @@ -58,6 +104,7 @@ def __init__(self, engine: "MemoryEngine") -> None: self._engine = engine self._task: asyncio.Task | None = None self._stop = asyncio.Event() + self._wakeup = asyncio.Event() # Monotonic timestamps of the last run per job, keyed by job name. self._last_run: dict[str, float] = {} @@ -86,6 +133,7 @@ def start(self) -> None: async def stop(self) -> None: """Stop the loop and wait for the current tick to finish.""" self._stop.set() + self._wakeup.set() if self._task and not self._task.done(): try: await self._task @@ -97,24 +145,33 @@ async def stop(self) -> None: def _any_job_enabled() -> bool: cfg = get_config() reconcile_on = cfg.consolidation_reconcile_interval_seconds > 0 + vector_index_reconcile_on = cfg.vector_index_reconcile_interval_seconds > 0 audit_on = cfg.audit_log_enabled and cfg.audit_log_retention_days > 0 llm_on = cfg.llm_trace_enabled and cfg.llm_trace_retention_days > 0 mm_refresh_on = cfg.mental_model_refresh_tick_seconds > 0 - return reconcile_on or audit_on or llm_on or mm_refresh_on + return reconcile_on or vector_index_reconcile_on or audit_on or llm_on or mm_refresh_on # ── loop ─────────────────────────────────────────────────────────────── async def _run(self) -> None: while not self._stop.is_set(): + self._wakeup.clear() try: await self._tick() except Exception: logger.exception("Maintenance tick failed") + if self._stop.is_set(): + break try: - await asyncio.wait_for(self._stop.wait(), timeout=_TICK_SECONDS) + await asyncio.wait_for(self._wakeup.wait(), timeout=_TICK_SECONDS) except asyncio.TimeoutError: pass + def request_vector_index_reconcile(self) -> None: + """Make vector index reconciliation due and wake the loop.""" + self._last_run.pop("vector_index_reconcile", None) + self._wakeup.set() + def _is_due(self, job: str, interval_seconds: int) -> bool: """True if ``job`` has never run or its interval has elapsed; marks it run now.""" now = time.monotonic() @@ -131,6 +188,9 @@ async def _tick(self) -> None: interval = cfg.consolidation_reconcile_interval_seconds if interval > 0 and self._is_due("reconcile", interval): await self._run_timed("consolidation reconcile", self._run_reconcile()) + vector_index_interval = cfg.vector_index_reconcile_interval_seconds + if vector_index_interval > 0 and self._is_due("vector_index_reconcile", vector_index_interval): + await self._run_timed("vector index reconcile", self._run_vector_index_reconcile()) mm_interval = cfg.mental_model_refresh_tick_seconds if mm_interval > 0 and self._is_due("mm_refresh", mm_interval): await self._run_timed("scheduled mental model refresh", self._run_scheduled_mm_refresh()) @@ -237,6 +297,33 @@ async def _run_reconcile(self) -> None: + (f", skipped {skipped_unknown} in unrecognized schema(s)" if skipped_unknown else "") ) + # ── vector index reconcile ────────────────────────────────────────────── + + async def _run_vector_index_reconcile(self) -> None: + """Converge missing per-bank indexes after restore, upgrade, or backend switch.""" + index_clause = _vector_index_clause() + if index_clause is None: + return + + engine = self._engine + try: + tenants = await engine._tenant_extension.list_tenants() + schemas = [get_config().database_schema] + schemas.extend(tenant.schema for tenant in tenants if tenant.schema) + schemas = list(dict.fromkeys(schemas)) + async with _raw_postgres_connection(engine) as conn: + results = await reconcile_vector_indexes(conn, schemas, index_clause) + except Exception as exc: + logger.warning(f"Vector index reconcile failed: {exc}") + return + + created = sum(result.created for result in results) + failed = sum(result.failed for result in results) + if created or failed: + logger.info( + f"Vector index reconcile: {created} index(es) created across {len(results)} schema(s), {failed} failed" + ) + # ── scheduled mental model refresh ─────────────────────────────────────── async def _run_scheduled_mm_refresh(self) -> None: diff --git a/hindsight-api-slim/hindsight_api/engine/memory_engine.py b/hindsight-api-slim/hindsight_api/engine/memory_engine.py index 854ba3c412..c7d0af27d1 100644 --- a/hindsight-api-slim/hindsight_api/engine/memory_engine.py +++ b/hindsight-api-slim/hindsight_api/engine/memory_engine.py @@ -1269,6 +1269,7 @@ def pick(field: str) -> Any: from .maintenance import MaintenanceLoop self._maintenance_loop: MaintenanceLoop | None = None + self._vector_index_check_last: dict[tuple[str, str], float] = {} # Backpressure mechanism: limit concurrent searches to prevent overwhelming the database # Configurable via HINDSIGHT_API_RECALL_MAX_CONCURRENT (default: 50) @@ -3661,6 +3662,35 @@ async def _submit_post_insert_maintenance( await self.submit_async_graph_maintenance(bank_id=bank_id, request_context=request_context) except Exception as e: logger.warning(f"Failed to submit graph maintenance task for bank {bank_id}: {e}") + await self._maybe_request_vector_index_reconcile(bank_id) + + async def _maybe_request_vector_index_reconcile(self, bank_id: str) -> None: + """Rarely check index coverage; a miss only wakes background maintenance.""" + cfg = get_config() + interval = cfg.vector_index_reconcile_interval_seconds + if interval <= 0 or self._maintenance_loop is None or self._database_backend_type != "postgresql": + return + + from .retain.bank_utils import _vector_index_clause + from .vector_index_reconcile import bank_vector_indexes_healthy + + if _vector_index_clause() is None: + return + schema = get_current_schema() + key = (schema, bank_id) + now = time.monotonic() + last = self._vector_index_check_last.get(key) + if last is not None and now - last < interval: + return + # Mark before I/O so simultaneous retains do not stampede the catalog. + self._vector_index_check_last[key] = now + try: + async with acquire_with_retry(self._backend, max_retries=1) as conn: + healthy = await bank_vector_indexes_healthy(conn, schema, bank_id) + if not healthy: + self._maintenance_loop.request_vector_index_reconcile() + except Exception as exc: + logger.warning(f"Vector index coverage check failed for bank {bank_id}: {exc}") async def _resolve_retain_chunking_config( self, diff --git a/hindsight-api-slim/hindsight_api/engine/vector_index_reconcile.py b/hindsight-api-slim/hindsight_api/engine/vector_index_reconcile.py new file mode 100644 index 0000000000..87495e26dd --- /dev/null +++ b/hindsight-api-slim/hindsight_api/engine/vector_index_reconcile.py @@ -0,0 +1,229 @@ +"""Background reconciliation for per-bank vector index coverage.""" + +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__) + +# Session-level lock serializes reconciliation across API replicas while allowing +# CREATE INDEX CONCURRENTLY to run outside a transaction on the same connection. +_VECTOR_INDEX_RECONCILE_LOCK_ID = 0x48494E4453494748 + +# Postgres emits 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))`. +# Match that exact rendering so a name-only false positive never passes. +_BANK_INDEX_PARTIAL_SUFFIX = " WHERE ((fact_type = " + + +# Postgres access methods we recognize as supporting per-(bank, fact_type) +# partial indexes. A legitimate HNSW/IVFFLAT/DiskANN/etc. index must match one +# of these; an index whose access method drifts after a backend switch does +# not, so the health check refuses to mark it healthy. +_SUPPORTED_INDEX_AM: tuple[str, ...] = ( + "btree", + "gin", + "gist", + "hnsw", + "ivfflat", + "diskann", + "vchordrq", +) + + +@dataclass +class SchemaVectorIndexReconcileResult: + """Outcome of reconciling one PostgreSQL schema.""" + + schema: str + banks_scanned: int = 0 + already_present: int = 0 + created: int = 0 + skipped: int = 0 + failed: int = 0 + failed_indexes: list[str] = field(default_factory=list) + # Set when reconciliation was skipped because another instance holds the + # advisory lock — distinguishes "nothing to do" from "another holder is + # already working". + skipped_lock_busy: bool = False + + +def _quote_identifier(value: str) -> str: + return '"' + value.replace('"', '""') + '"' + + +async def bank_vector_indexes_healthy(conn: Any, schema: str, bank_id: str) -> bool: + """Check one bank's expected index set with a single catalog query.""" + qschema = _quote_identifier(schema) + suffixes = ", ".join(f"('{suffix}')" for suffix in _BANK_INDEX_FACT_TYPES.values()) + healthy = await conn.fetchval( + f""" + WITH expected AS ( + SELECT 'idx_mu_emb_' || spec.suffix || '_' || + left(replace(b.internal_id::text, '-', ''), 16) AS index_name + FROM {qschema}.banks b + CROSS JOIN (VALUES {suffixes}) AS spec(suffix) + WHERE b.bank_id = $1 + ) + SELECT count(i.indexrelid) = $3 + AND coalesce(bool_and( + (i.indisvalid AND i.indisready) + AND am.amname = ANY($4::text[]) + AND pg_get_indexdef(i.indexrelid) LIKE $5 + ), false) + FROM expected e + LEFT JOIN pg_namespace n ON n.nspname = $2 + LEFT JOIN pg_class c ON c.relnamespace = n.oid AND c.relname = e.index_name + LEFT JOIN pg_index i ON i.indexrelid = c.oid + LEFT JOIN pg_am am ON am.oid = c.relam + """, + bank_id, + schema, + len(_BANK_INDEX_FACT_TYPES), + list(_SUPPORTED_INDEX_AM), + "%" + _BANK_INDEX_PARTIAL_SUFFIX + "%", + ) + return bool(healthy) + + +async def _index_health(conn: Any, schema: str, index_names: list[str]) -> dict[str, bool]: + """Return valid-and-ready state for the requested indexes in one query. + + Health requires the index to be valid, ready, defined over the expected + `memory_units` table, to use a supported access method, and to carry our + partial predicate. A name-only match (e.g. an unrelated index with the + same relname) is *not* enough — backend switches would otherwise silently + be marked healthy. + """ + 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 _reconcile_schema( + conn: Any, + schema: str, + index_clause: str, + *, + dry_run: bool, +) -> SchemaVectorIndexReconcileResult: + result = SchemaVectorIndexReconcileResult(schema=schema) + qschema = _quote_identifier(schema) + banks = await conn.fetch(f"SELECT bank_id, internal_id FROM {qschema}.banks ORDER BY bank_id") + result.banks_scanned = len(banks) + + bank_specs: list[tuple[Any, dict[str, str]]] = [] + all_index_names: list[str] = [] + for bank in banks: + expected = { + fact_type: _bank_index_name(fact_type, str(bank["internal_id"])) for fact_type in _BANK_INDEX_FACT_TYPES + } + bank_specs.append((bank, expected)) + all_index_names.extend(expected.values()) + health = await _index_health(conn, schema, all_index_names) if all_index_names else {} + + for bank, expected in bank_specs: + bank_id = bank["bank_id"] + bank_id_literal = await conn.fetchval("SELECT quote_literal($1::text)", bank_id) + for fact_type in _BANK_INDEX_FACT_TYPES: + index_name = expected[fact_type] + validity = health.get(index_name) + if validity is True: + result.already_present += 1 + continue + + qindex = _quote_identifier(index_name) + qualified_index = f"{qschema}.{qindex}" + if dry_run: + result.skipped += 1 + continue + try: + if validity is False: + await conn.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {qualified_index}") + await conn.execute( + f"CREATE INDEX CONCURRENTLY IF NOT EXISTS {qindex} " + f"ON {qschema}.memory_units {index_clause} " + f"WHERE fact_type = '{fact_type}' AND bank_id = {bank_id_literal}" + ) + result.created += 1 + except Exception as exc: + result.failed += 1 + result.failed_indexes.append(f"{schema}.{index_name}") + logger.warning( + "Failed to reconcile vector index %s (bank=%s, fact_type=%s): %s", + qualified_index, + bank_id, + fact_type, + exc, + ) + try: + await conn.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {qualified_index}") + except Exception as cleanup_exc: + logger.warning("Failed to clean up vector index %s: %s", qualified_index, cleanup_exc) + + return result + + +async def _safe_reconcile_schema( + conn: Any, + schema: str, + index_clause: str, + *, + dry_run: bool, +) -> SchemaVectorIndexReconcileResult: + try: + return await _reconcile_schema(conn, schema, index_clause, dry_run=dry_run) + except Exception as exc: + logger.warning("Vector index reconcile aborted for schema %s: %s", schema, exc) + return SchemaVectorIndexReconcileResult( + schema=schema, + failed=1, + failed_indexes=[f"{schema}."], + ) + + +async def reconcile_vector_indexes( + conn: Any, + schemas: list[str], + index_clause: str, + *, + dry_run: bool = False, +) -> list[SchemaVectorIndexReconcileResult]: + """Rebuild missing/invalid per-bank vector indexes without blocking writes. + + The caller must provide a raw autocommit PostgreSQL connection because + ``CREATE INDEX CONCURRENTLY`` cannot run inside a transaction block. + """ + acquired = await conn.fetchval("SELECT pg_try_advisory_lock($1)", _VECTOR_INDEX_RECONCILE_LOCK_ID) + if not acquired: + logger.info("Vector index reconcile skipped: another instance holds the advisory lock") + busy = SchemaVectorIndexReconcileResult(schema="", skipped_lock_busy=True) + return [busy] + + try: + return [await _safe_reconcile_schema(conn, schema, index_clause, dry_run=dry_run) for schema in schemas] + finally: + await conn.fetchval("SELECT pg_advisory_unlock($1)", _VECTOR_INDEX_RECONCILE_LOCK_ID) diff --git a/hindsight-api-slim/tests/conftest.py b/hindsight-api-slim/tests/conftest.py index 4c5bf05dde..c62ac78172 100644 --- a/hindsight-api-slim/tests/conftest.py +++ b/hindsight-api-slim/tests/conftest.py @@ -88,9 +88,10 @@ def _cleanup_leaked_span_recorders(): # and llm-trace retention — with audit retention already off by default — leaves # no job enabled, so the loop never starts. Tests that exercise it call # MaintenanceLoop methods (_run_reconcile / _run_scheduled_mm_refresh / -# _purge_expired) directly. +# _purge_expired / _run_vector_index_reconcile) directly. os.environ.setdefault("HINDSIGHT_API_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS", "0") os.environ.setdefault("HINDSIGHT_API_MENTAL_MODEL_REFRESH_TICK_SECONDS", "0") +os.environ.setdefault("HINDSIGHT_API_VECTOR_INDEX_RECONCILE_INTERVAL_SECONDS", "0") os.environ.setdefault("HINDSIGHT_API_LLM_TRACE_RETENTION_DAYS", "-1") diff --git a/hindsight-api-slim/tests/test_maintenance_loop.py b/hindsight-api-slim/tests/test_maintenance_loop.py index e270d586e9..692d2f75c3 100644 --- a/hindsight-api-slim/tests/test_maintenance_loop.py +++ b/hindsight-api-slim/tests/test_maintenance_loop.py @@ -3,6 +3,8 @@ import time import uuid +from contextlib import asynccontextmanager +from types import SimpleNamespace import pytest @@ -32,6 +34,92 @@ def test_is_due_runs_at_start_then_waits_interval(): assert loop._is_due("job", 3600) is True +def test_reconcile_request_marks_job_due_and_wakes_loop(): + loop = MaintenanceLoop(engine=None) + loop._last_run["vector_index_reconcile"] = time.monotonic() + + loop.request_vector_index_reconcile() + + assert "vector_index_reconcile" not in loop._last_run + assert loop._wakeup.is_set() + + +@pytest.mark.asyncio +async def test_tick_runs_vector_index_reconcile_at_start(monkeypatch): + loop = MaintenanceLoop(engine=None) + called = 0 + + async def _record(): + nonlocal called + called += 1 + + async def _noop(*args, **kwargs): + return None + + cfg = SimpleNamespace( + audit_log_enabled=False, + audit_log_retention_days=-1, + llm_trace_enabled=False, + llm_trace_retention_days=-1, + consolidation_reconcile_interval_seconds=0, + mental_model_refresh_tick_seconds=0, + vector_index_reconcile_interval_seconds=3600, + ) + monkeypatch.setattr("hindsight_api.engine.maintenance.get_config", lambda: cfg) + monkeypatch.setattr(loop, "_run_vector_index_reconcile", _record) + monkeypatch.setattr(loop, "_run_retention", _noop) + + await loop._tick() + await loop._tick() + + assert called == 1 + + +@pytest.mark.asyncio +async def test_vector_index_reconcile_uses_raw_autocommit_connection(monkeypatch): + import hindsight_api.engine.maintenance as maintenance_mod + + events: list[object] = [] + + class FakeRawConnection: + pass + + @asynccontextmanager + async def fake_raw_connection(_engine): + events.append("opened") + try: + yield FakeRawConnection() + finally: + events.append("closed") + + async def fake_reconcile(conn, schemas, index_clause): + events.append((conn.__class__.__name__, schemas, index_clause)) + return [SimpleNamespace(created=2, failed=0)] + + engine = SimpleNamespace( + _tenant_extension=SimpleNamespace( + list_tenants=lambda: _async_result([SimpleNamespace(schema="public"), SimpleNamespace(schema="tenant_a")]) + ) + ) + monkeypatch.setattr(maintenance_mod, "_raw_postgres_connection", fake_raw_connection) + monkeypatch.setattr(maintenance_mod, "_vector_index_clause", lambda: "USING hnsw (embedding vector_cosine_ops)") + monkeypatch.setattr(maintenance_mod, "reconcile_vector_indexes", fake_reconcile) + + await MaintenanceLoop(engine)._run_vector_index_reconcile() + + assert events[0] == "opened" + assert events[1] == ( + "FakeRawConnection", + ["public", "tenant_a"], + "USING hnsw (embedding vector_cosine_ops)", + ) + assert events[-1] == "closed" + + +async def _async_result(value): + return value + + async def _make_bank(memory: MemoryEngine, request_context, suffix: str, config_json: str | None = None) -> str: bank_id = f"recon-{suffix}-{uuid.uuid4().hex[:8]}" await memory.get_bank_profile(bank_id=bank_id, request_context=request_context) diff --git a/hindsight-api-slim/tests/test_vector_index_reconcile.py b/hindsight-api-slim/tests/test_vector_index_reconcile.py new file mode 100644 index 0000000000..42a1c3fa61 --- /dev/null +++ b/hindsight-api-slim/tests/test_vector_index_reconcile.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import re +import uuid + +import pytest + +from hindsight_api.engine.retain.bank_utils import _BANK_INDEX_FACT_TYPES, _bank_index_name +from hindsight_api.engine.vector_index_reconcile import bank_vector_indexes_healthy, reconcile_vector_indexes + + +class _FakeConnection: + def __init__(self, *, lock_acquired: bool = True) -> None: + self.lock_acquired = lock_acquired + self.bank_id = "restored-bank" + self.internal_id = uuid.uuid4() + self.indexes: dict[str, bool] = {} + self.statements: list[str] = [] + self.fail_once: set[str] = set() + self.drop_fail_once: set[str] = set() + self.catalog_queries = 0 + + async def fetch(self, query: str, *args): + if "pg_index" in query: + self.catalog_queries += 1 + return [ + {"index_name": name, "healthy": healthy} for name, healthy in self.indexes.items() if name in args[1] + ] + return [{"bank_id": self.bank_id, "internal_id": self.internal_id}] + + async def fetchval(self, query: str, *args): + if "pg_try_advisory_lock" in query: + return self.lock_acquired + if "pg_advisory_unlock" in query: + return True + if "quote_literal" in query: + value = args[0] + return "'" + str(value).replace("'", "''") + "'" + raise AssertionError(f"unexpected fetchval query: {query}") + + async def execute(self, query: str): + self.statements.append(query) + match = re.search( + r'(?:CREATE INDEX CONCURRENTLY IF NOT EXISTS|DROP INDEX CONCURRENTLY IF EXISTS)\s+(?:"[^"]+"\.)?"?([a-zA-Z0-9_]+)"?', + query, + ) + assert match, query + index_name = match.group(1) + if query.startswith("CREATE"): + if index_name in self.fail_once: + self.fail_once.remove(index_name) + self.indexes[index_name] = False + raise RuntimeError("simulated concurrent build failure") + self.indexes[index_name] = True + else: + if index_name in self.drop_fail_once: + self.drop_fail_once.remove(index_name) + raise RuntimeError("simulated concurrent drop failure") + self.indexes.pop(index_name, None) + + +def _expected_names(conn: _FakeConnection) -> list[str]: + return [_bank_index_name(fact_type, str(conn.internal_id)) for fact_type in _BANK_INDEX_FACT_TYPES] + + +@pytest.mark.asyncio +async def test_bank_health_checks_all_expected_indexes_in_one_catalog_query() -> None: + calls = [] + + class HealthConnection: + async def fetchval(self, query: str, *args): + calls.append((query, args)) + return True + + healthy = await bank_vector_indexes_healthy(HealthConnection(), "tenant_a", "restored-bank") + + assert healthy is True + assert len(calls) == 1 + query, args = calls[0] + assert 'FROM "tenant_a".banks' in query + assert "i.indisvalid AND i.indisready" in query + assert "am.amname = ANY($4::text[])" in query + assert args[0] == "restored-bank" + assert args[1] == "tenant_a" + assert args[2] == len(_BANK_INDEX_FACT_TYPES) + + +@pytest.mark.asyncio +async def test_dry_run_reports_missing_indexes_without_changing_catalog() -> None: + conn = _FakeConnection() + expected = _expected_names(conn) + conn.indexes[expected[0]] = False + + results = await reconcile_vector_indexes( + conn, + ["public"], + "USING hnsw (embedding vector_cosine_ops)", + dry_run=True, + ) + + assert results[0].created == 0 + assert results[0].skipped == len(_BANK_INDEX_FACT_TYPES) + assert conn.indexes == {expected[0]: False} + assert conn.statements == [] + + +@pytest.mark.asyncio +async def test_reconcile_creates_missing_and_replaces_invalid_indexes() -> None: + conn = _FakeConnection() + expected = _expected_names(conn) + conn.indexes[expected[0]] = True + conn.indexes[expected[1]] = False + + results = await reconcile_vector_indexes(conn, ["public"], "USING hnsw (embedding vector_cosine_ops)") + + assert len(results) == 1 + result = results[0] + assert result.already_present == 1 + assert result.created == 2 + assert result.failed == 0 + assert all(conn.indexes[name] for name in expected) + assert any(statement.startswith("DROP INDEX CONCURRENTLY") for statement in conn.statements) + assert all("CONCURRENTLY" in statement for statement in conn.statements) + assert conn.catalog_queries == 1 + + +@pytest.mark.asyncio +async def test_reconcile_skips_when_another_instance_holds_lock() -> None: + conn = _FakeConnection(lock_acquired=False) + + results = await reconcile_vector_indexes(conn, ["public"], "USING hnsw (embedding vector_cosine_ops)") + + assert len(results) == 1 + assert results[0].skipped_lock_busy is True + assert conn.statements == [] + + +@pytest.mark.asyncio +async def test_failed_build_is_cleaned_up_and_retried_on_next_reconcile() -> None: + conn = _FakeConnection() + failed_index = _expected_names(conn)[0] + conn.fail_once.add(failed_index) + + first = await reconcile_vector_indexes(conn, ["public"], "USING hnsw (embedding vector_cosine_ops)") + second = await reconcile_vector_indexes(conn, ["public"], "USING hnsw (embedding vector_cosine_ops)") + + assert first[0].failed == 1 + assert f"public.{failed_index}" in first[0].failed_indexes + assert second[0].created == 1 + assert conn.indexes[failed_index] is True + + +@pytest.mark.asyncio +async def test_failed_invalid_index_drop_does_not_abort_other_indexes() -> None: + conn = _FakeConnection() + expected = _expected_names(conn) + conn.indexes[expected[0]] = False + conn.drop_fail_once.add(expected[0]) + + results = await reconcile_vector_indexes(conn, ["public"], "USING hnsw (embedding vector_cosine_ops)") + + assert results[0].failed == 1 + assert f"public.{expected[0]}" in results[0].failed_indexes + assert conn.indexes[expected[1]] is True + assert conn.indexes[expected[2]] is True diff --git a/hindsight-api-slim/tests/test_vector_index_retain_safety.py b/hindsight-api-slim/tests/test_vector_index_retain_safety.py new file mode 100644 index 0000000000..16a7d7fe5c --- /dev/null +++ b/hindsight-api-slim/tests/test_vector_index_retain_safety.py @@ -0,0 +1,51 @@ +from contextlib import asynccontextmanager +from types import SimpleNamespace + +import pytest + +from hindsight_api.engine.memory_engine import MemoryEngine + + +@pytest.mark.asyncio +async def test_retain_safety_check_is_ttl_gated_and_only_wakes_maintenance(monkeypatch) -> None: + import hindsight_api.engine.memory_engine as memory_engine_mod + import hindsight_api.engine.retain.bank_utils as bank_utils + import hindsight_api.engine.vector_index_reconcile as reconcile_mod + + queries = 0 + wakeups = 0 + + @asynccontextmanager + async def fake_acquire_with_retry(_backend, **_kwargs): + yield SimpleNamespace() + + class Maintenance: + def request_vector_index_reconcile(self): + nonlocal wakeups + wakeups += 1 + + async def unhealthy(_conn, schema, bank_id): + nonlocal queries + queries += 1 + assert schema == "tenant_a" + assert bank_id == "restored-bank" + return False + + engine = SimpleNamespace( + _backend=None, + _database_backend_type="postgresql", + _maintenance_loop=Maintenance(), + _vector_index_check_last={}, + ) + cfg = SimpleNamespace(vector_index_reconcile_interval_seconds=3600) + monkeypatch.setattr(memory_engine_mod, "get_config", lambda: cfg) + monkeypatch.setattr(memory_engine_mod, "get_current_schema", lambda: "tenant_a") + monkeypatch.setattr(memory_engine_mod, "acquire_with_retry", fake_acquire_with_retry) + monkeypatch.setattr(bank_utils, "_vector_index_clause", lambda: "USING hnsw (embedding vector_cosine_ops)") + monkeypatch.setattr(reconcile_mod, "bank_vector_indexes_healthy", unhealthy) + + await MemoryEngine._maybe_request_vector_index_reconcile(engine, "restored-bank") + await MemoryEngine._maybe_request_vector_index_reconcile(engine, "restored-bank") + + assert queries == 1 + assert wakeups == 1 diff --git a/hindsight-docs/docs/developer/configuration.md b/hindsight-docs/docs/developer/configuration.md index 3e014c755d..2958d8ff28 100644 --- a/hindsight-docs/docs/developer/configuration.md +++ b/hindsight-docs/docs/developer/configuration.md @@ -1476,6 +1476,7 @@ Observations are deduplicated, evidence-grounded knowledge consolidated from mul | `HINDSIGHT_API_ENABLE_OBSERVATIONS` | Enable observation consolidation | `true` | | `HINDSIGHT_API_ENABLE_AUTO_CONSOLIDATION` | Automatically trigger consolidation after retain, delete, and update operations. When `false`, consolidation only runs when explicitly triggered via the [consolidate endpoint](/developer/api/operations#consolidation). Configurable per bank. | `true` | | `HINDSIGHT_API_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS` | Interval for the background sweep that re-schedules consolidation for banks with unconsolidated facts but no consolidation in progress — recovering facts left unscheduled when a consolidation operation failed terminally (e.g. the LLM provider was unavailable). Only applies to banks with auto-consolidation enabled. `0` disables the sweep. | `300` | +| `HINDSIGHT_API_VECTOR_INDEX_RECONCILE_INTERVAL_SECONDS` | Interval for the PostgreSQL background sweep that rebuilds missing or invalid per-bank vector indexes with `CREATE INDEX CONCURRENTLY`. It runs once at startup, then on this interval, repairing coverage after restores, upgrades, and vector-backend switches without blocking writes. `0` disables the sweep. | `3600` | | `HINDSIGHT_API_MENTAL_MODEL_REFRESH_TICK_SECONDS` | How often the background loop checks for cron-scheduled mental models that are due for a refresh. This is only the *check* cadence; the actual schedule is the per-model `trigger.refresh_cron` expression set on the mental model. A due model is refreshed only when it is stale (new memories in its scope since the last refresh). `0` disables the sweep. | `60` | | `HINDSIGHT_API_ENABLE_OBSERVATION_HISTORY` | Track history of changes to each observation (previous text/tags/dates + timestamp), stored one row per change in the `observation_history` table. Set to `false` to disable entirely — no history rows are written. **This is how you turn the feature off** (not a zero cap). | `true` | | `HINDSIGHT_API_OBSERVATION_HISTORY_MAX_ENTRIES` | Max history rows kept per observation. On each update the previous version is inserted into the `observation_history` table and the oldest rows beyond this cap are deleted, so an often-reinforced observation's history can't grow without bound. `0` or a negative value **removes the cap** (unbounded); to turn history off entirely set `HINDSIGHT_API_ENABLE_OBSERVATION_HISTORY=false` instead. | `50` | diff --git a/hindsight-docs/versioned_docs/version-0.8/developer/configuration.md b/hindsight-docs/versioned_docs/version-0.8/developer/configuration.md index b43d4a50b9..35337fe0f2 100644 --- a/hindsight-docs/versioned_docs/version-0.8/developer/configuration.md +++ b/hindsight-docs/versioned_docs/version-0.8/developer/configuration.md @@ -1472,6 +1472,7 @@ Observations are deduplicated, evidence-grounded knowledge consolidated from mul | `HINDSIGHT_API_ENABLE_OBSERVATIONS` | Enable observation consolidation | `true` | | `HINDSIGHT_API_ENABLE_AUTO_CONSOLIDATION` | Automatically trigger consolidation after retain, delete, and update operations. When `false`, consolidation only runs when explicitly triggered via the [consolidate endpoint](/developer/api/operations#consolidation). Configurable per bank. | `true` | | `HINDSIGHT_API_CONSOLIDATION_RECONCILE_INTERVAL_SECONDS` | Interval for the background sweep that re-schedules consolidation for banks with unconsolidated facts but no consolidation in progress — recovering facts left unscheduled when a consolidation operation failed terminally (e.g. the LLM provider was unavailable). Only applies to banks with auto-consolidation enabled. `0` disables the sweep. | `300` | +| `HINDSIGHT_API_VECTOR_INDEX_RECONCILE_INTERVAL_SECONDS` | Interval for the PostgreSQL background sweep that rebuilds missing or invalid per-bank vector indexes with `CREATE INDEX CONCURRENTLY`. It runs once at startup, then on this interval, repairing coverage after restores, upgrades, and vector-backend switches without blocking writes. `0` disables the sweep. | `3600` | | `HINDSIGHT_API_MENTAL_MODEL_REFRESH_TICK_SECONDS` | How often the background loop checks for cron-scheduled mental models that are due for a refresh. This is only the *check* cadence; the actual schedule is the per-model `trigger.refresh_cron` expression set on the mental model. A due model is refreshed only when it is stale (new memories in its scope since the last refresh). `0` disables the sweep. | `60` | | `HINDSIGHT_API_ENABLE_OBSERVATION_HISTORY` | Track history of changes to each observation (previous text/tags/dates + timestamp), stored one row per change in the `observation_history` table. Set to `false` to disable entirely — no history rows are written. **This is how you turn the feature off** (not a zero cap). | `true` | | `HINDSIGHT_API_OBSERVATION_HISTORY_MAX_ENTRIES` | Max history rows kept per observation. On each update the previous version is inserted into the `observation_history` table and the oldest rows beyond this cap are deleted, so an often-reinforced observation's history can't grow without bound. `0` or a negative value **removes the cap** (unbounded); to turn history off entirely set `HINDSIGHT_API_ENABLE_OBSERVATION_HISTORY=false` instead. | `50` | diff --git a/hindsight-embed/hindsight_embed/env.example b/hindsight-embed/hindsight_embed/env.example index ef33ab247c..9c4431c2d8 100644 --- a/hindsight-embed/hindsight_embed/env.example +++ b/hindsight-embed/hindsight_embed/env.example @@ -107,6 +107,7 @@ HINDSIGHT_API_LOG_LEVEL=info # HINDSIGHT_API_MIGRATION_CONCURRENCY=1 # Tenant schemas to migrate concurrently (PG only, each in its own process; per-schema work stays sequential). Each worker has ~1-2s startup cost + uses ~3 DB connections, so it only pays off with many schemas (tens+) or slow migrations; keep concurrency*3 <= spare max_connections. Default: 1 (sequential). # HINDSIGHT_API_OPERATION_RETENTION_DAYS=30 # Keep terminal operation rows, payloads, and metadata for this many days; 0 disables automatic pruning. # HINDSIGHT_API_OPERATION_CLEANUP_BATCH_SIZE=1000 # Maximum expired terminal rows deleted per tenant schema in each cleanup cycle; must be positive. +# HINDSIGHT_API_VECTOR_INDEX_RECONCILE_INTERVAL_SECONDS=3600 # Rebuild missing/invalid per-bank vector indexes at startup and on this interval; 0 disables. # Vector Extension (Optional - uses pgvector by default) # Options: "pgvector" (default), "vchord", "pgvectorscale" (DiskANN)