diff --git a/hindsight-api/hindsight_api/admin/cli.py b/hindsight-api/hindsight_api/admin/cli.py index dfd74115..ff7b05d5 100644 --- a/hindsight-api/hindsight_api/admin/cli.py +++ b/hindsight-api/hindsight_api/admin/cli.py @@ -214,6 +214,36 @@ def restore( typer.echo("Restore complete") +async def _run_migration(db_url: str, schema: str = "public") -> None: + """Resolve database URL and run migrations.""" + from ..migrations import run_migrations + + is_pg0, instance_name, _ = parse_pg0_url(db_url) + if is_pg0: + typer.echo(f"Starting embedded PostgreSQL (instance: {instance_name})...") + resolved_url = await resolve_database_url(db_url) + run_migrations(resolved_url, schema=schema) + + +@app.command(name="run-db-migration") +def run_db_migration( + schema: str = typer.Option("public", "--schema", "-s", help="Database schema to run migrations on"), +): + """Run database migrations to the latest version.""" + 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) + + typer.echo(f"Running database migrations (schema: {schema})...") + + asyncio.run(_run_migration(config.database_url, schema)) + + typer.echo("Database migrations completed successfully") + + def main(): app() diff --git a/hindsight-api/hindsight_api/config.py b/hindsight-api/hindsight_api/config.py index 898a3685..c3a969f5 100644 --- a/hindsight-api/hindsight_api/config.py +++ b/hindsight-api/hindsight_api/config.py @@ -54,6 +54,9 @@ ENV_RETAIN_MAX_COMPLETION_TOKENS = "HINDSIGHT_API_RETAIN_MAX_COMPLETION_TOKENS" ENV_SKIP_LLM_VERIFICATION = "HINDSIGHT_API_SKIP_LLM_VERIFICATION" ENV_LAZY_RERANKER = "HINDSIGHT_API_LAZY_RERANKER" +# Database migrations +ENV_RUN_MIGRATIONS_ON_STARTUP = "HINDSIGHT_API_RUN_MIGRATIONS_ON_STARTUP" + # Default values DEFAULT_DATABASE_URL = "pg0" DEFAULT_LLM_PROVIDER = "openai" @@ -83,6 +86,9 @@ DEFAULT_OBSERVATION_TOP_ENTITIES = 5 # Max entities to process per retain batch # Retain settings DEFAULT_RETAIN_MAX_COMPLETION_TOKENS = 64000 # Max tokens for fact extraction LLM call +# Database migrations +DEFAULT_RUN_MIGRATIONS_ON_STARTUP = True + # Default MCP tool descriptions (can be customized via env vars) DEFAULT_MCP_RETAIN_DESCRIPTION = """Store important information to long-term memory. @@ -152,6 +158,9 @@ class HindsightConfig: skip_llm_verification: bool lazy_reranker: bool + # Database migrations + run_migrations_on_startup: bool + @classmethod def from_env(cls) -> "HindsightConfig": """Create configuration from environment variables.""" @@ -192,6 +201,8 @@ class HindsightConfig: retain_max_completion_tokens=int( os.getenv(ENV_RETAIN_MAX_COMPLETION_TOKENS, str(DEFAULT_RETAIN_MAX_COMPLETION_TOKENS)) ), + # Database migrations + run_migrations_on_startup=os.getenv(ENV_RUN_MIGRATIONS_ON_STARTUP, "true").lower() == "true", ) def get_llm_base_url(self) -> str: diff --git a/hindsight-api/hindsight_api/main.py b/hindsight-api/hindsight_api/main.py index 7f973f52..56d15767 100644 --- a/hindsight-api/hindsight_api/main.py +++ b/hindsight-api/hindsight_api/main.py @@ -187,6 +187,7 @@ def main(): retain_max_completion_tokens=config.retain_max_completion_tokens, skip_llm_verification=config.skip_llm_verification, lazy_reranker=config.lazy_reranker, + run_migrations_on_startup=config.run_migrations_on_startup, ) config.configure_logging() if not args.daemon: @@ -212,7 +213,11 @@ def main(): logging.info(f"Loaded tenant extension: {tenant_extension.__class__.__name__}") # Create MemoryEngine (reads configuration from environment) - _memory = MemoryEngine(operation_validator=operation_validator, tenant_extension=tenant_extension) + _memory = MemoryEngine( + operation_validator=operation_validator, + tenant_extension=tenant_extension, + run_migrations=config.run_migrations_on_startup, + ) # Set extension context on tenant extension (needed for schema provisioning) if tenant_extension: diff --git a/hindsight-api/tests/test_admin_backup_restore.py b/hindsight-api/tests/test_admin_backup_restore.py index 37b0a4ce..32e31e0b 100644 --- a/hindsight-api/tests/test_admin_backup_restore.py +++ b/hindsight-api/tests/test_admin_backup_restore.py @@ -1,8 +1,9 @@ """ Tests for admin backup and restore functionality. -Note: These tests run sequentially (not in parallel) because they all -manipulate the same database and do full backup/restore operations. +These tests use an isolated schema to avoid interfering with other tests. +The backup/restore operations truncate tables, which would cause deadlocks +and race conditions if run against the shared public schema. """ import tempfile @@ -10,49 +11,106 @@ import uuid import zipfile from pathlib import Path +import asyncpg import pytest +import pytest_asyncio -from hindsight_api import RequestContext from hindsight_api.admin.cli import _backup, _restore, BACKUP_TABLES +from hindsight_api.migrations import run_migrations # Run these tests sequentially since they do full DB backup/restore pytestmark = pytest.mark.xdist_group(name="backup_restore") +@pytest_asyncio.fixture(scope="function") +async def backup_test_schema(pg0_db_url, embeddings): + """Create an isolated schema for backup/restore tests. + + Uses a unique schema name per test invocation to avoid conflicts with + parallel test runs or leftover state from interrupted runs. + + Returns a tuple of (db_url, schema_name, fq_helper, embeddings). + """ + # Initialize embeddings if not already done + await embeddings.initialize() + + # Use unique schema name to avoid conflicts + schema_name = f"backup_test_{uuid.uuid4().hex[:8]}" + + def _fq(table: str) -> str: + """Get fully-qualified table name in test schema.""" + return f"{schema_name}.{table}" + + conn = await asyncpg.connect(pg0_db_url) + try: + await conn.execute(f"CREATE SCHEMA {schema_name}") + finally: + await conn.close() + + # Run migrations on the isolated schema + run_migrations(pg0_db_url, schema=schema_name) + + yield pg0_db_url, schema_name, _fq, embeddings + + # Cleanup after test + conn = await asyncpg.connect(pg0_db_url) + try: + await conn.execute(f"DROP SCHEMA IF EXISTS {schema_name} CASCADE") + finally: + await conn.close() + + @pytest.mark.asyncio -async def test_backup_restore_roundtrip(memory, pg0_db_url, request_context): +async def test_backup_restore_roundtrip(backup_test_schema): """Test that backup and restore preserves all data correctly.""" - # Use unique bank ID to avoid conflicts + db_url, schema_name, _fq, embeddings = backup_test_schema bank_id = f"test-backup-{uuid.uuid4().hex[:8]}" + conn = await asyncpg.connect(db_url) - # Create some test data - await memory.retain_batch_async( - bank_id=bank_id, - contents=[ - {"content": "Alice is a software engineer who loves Python."}, - {"content": "Bob works with Alice on the backend team."}, - {"content": "The team uses PostgreSQL for their database."}, - ], - request_context=request_context, - ) + try: + # Create a bank + await conn.execute( + f"INSERT INTO {_fq('banks')} (bank_id) VALUES ($1) ON CONFLICT DO NOTHING", + bank_id, + ) - # Get counts before backup - async with memory._pool.acquire() as conn: + # Create some test memory units with embeddings + # Convert embedding list to pgvector format string + embedding_list = embeddings.encode(["Test content about Alice"])[0] + embedding_str = "[" + ",".join(str(x) for x in embedding_list) + "]" + for text in [ + "Alice is a software engineer who loves Python.", + "Bob works with Alice on the backend team.", + "The team uses PostgreSQL for their database.", + ]: + await conn.execute( + f"""INSERT INTO {_fq('memory_units')} + (bank_id, text, fact_type, embedding, event_date) + VALUES ($1, $2, 'world', $3::vector, NOW())""", + bank_id, + text, + embedding_str, + ) + + # Get counts before backup counts_before = {} for table in BACKUP_TABLES: - counts_before[table] = await conn.fetchval(f"SELECT COUNT(*) FROM {table}") + counts_before[table] = await conn.fetchval(f"SELECT COUNT(*) FROM {_fq(table)}") - # Verify we have data - assert counts_before["banks"] > 0 - assert counts_before["memory_units"] > 0 + # Verify we have data + assert counts_before["banks"] > 0 + assert counts_before["memory_units"] > 0 + + finally: + await conn.close() # Backup to a temp file with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f: backup_path = Path(f.name) try: - manifest = await _backup(pg0_db_url, backup_path) + manifest = await _backup(db_url, backup_path, schema=schema_name) # Verify backup file exists and is valid assert backup_path.exists() @@ -72,33 +130,37 @@ async def test_backup_restore_roundtrip(memory, pg0_db_url, request_context): assert f"{table}.bin" in zf.namelist() # Clear all data - async with memory._pool.acquire() as conn: + conn = await asyncpg.connect(db_url) + try: for table in reversed(BACKUP_TABLES): - await conn.execute(f"TRUNCATE TABLE {table} CASCADE") + await conn.execute(f"TRUNCATE TABLE {_fq(table)} CASCADE") - # Verify data is gone - async with memory._pool.acquire() as conn: + # Verify data is gone for table in BACKUP_TABLES: - count = await conn.fetchval(f"SELECT COUNT(*) FROM {table}") + count = await conn.fetchval(f"SELECT COUNT(*) FROM {_fq(table)}") assert count == 0, f"Table {table} should be empty after truncate" + finally: + await conn.close() # Restore from backup - await _restore(pg0_db_url, backup_path) + await _restore(db_url, backup_path, schema=schema_name) # Verify counts match original - async with memory._pool.acquire() as conn: + conn = await asyncpg.connect(db_url) + try: for table in BACKUP_TABLES: - count = await conn.fetchval(f"SELECT COUNT(*) FROM {table}") + count = await conn.fetchval(f"SELECT COUNT(*) FROM {_fq(table)}") assert count == counts_before[table], f"Table {table} count mismatch after restore" - # Verify data content is preserved - async with memory._pool.acquire() as conn: + # Verify data content is preserved texts = await conn.fetch( - "SELECT text FROM memory_units WHERE bank_id = $1", + f"SELECT text FROM {_fq('memory_units')} WHERE bank_id = $1", bank_id, ) text_content = " ".join(r["text"] for r in texts) assert "Alice" in text_content or "software" in text_content + finally: + await conn.close() finally: # Cleanup @@ -107,42 +169,60 @@ async def test_backup_restore_roundtrip(memory, pg0_db_url, request_context): @pytest.mark.asyncio -async def test_backup_restore_preserves_all_column_types(memory, pg0_db_url, request_context): +async def test_backup_restore_preserves_all_column_types(backup_test_schema): """Test that all column types are preserved: vectors, UUIDs, timestamps, JSONB.""" - # Use unique bank ID + db_url, schema_name, _fq, embeddings = backup_test_schema bank_id = f"test-types-{uuid.uuid4().hex[:8]}" + conn = await asyncpg.connect(db_url) - # Create data with meaningful content that will produce facts - await memory.retain_batch_async( - bank_id=bank_id, - contents=[ - {"content": "John Smith is a senior engineer at Acme Corp since 2020."}, - {"content": "The project deadline is December 15th 2024."}, - ], - request_context=request_context, - ) + try: + # Create a bank + await conn.execute( + f"INSERT INTO {_fq('banks')} (bank_id) VALUES ($1) ON CONFLICT DO NOTHING", + bank_id, + ) - # Get original data with all important column types - async with memory._pool.acquire() as conn: - # memory_units: UUID (id), Vector (embedding), Timestamp (event_date, created_at), JSONB (metadata) + # Create a memory unit with all column types + # Convert embedding list to pgvector format string + embedding_list = embeddings.encode(["John Smith engineer"])[0] + embedding_str = "[" + ",".join(str(x) for x in embedding_list) + "]" + await conn.execute( + f"""INSERT INTO {_fq('memory_units')} + (bank_id, text, fact_type, embedding, event_date, metadata) + VALUES ($1, $2, 'world', $3::vector, NOW(), $4)""", + bank_id, + "John Smith is a senior engineer at Acme Corp since 2020.", + embedding_str, + '{"key": "value"}', + ) + + # Create an entity + await conn.execute( + f"""INSERT INTO {_fq('entities')} + (bank_id, canonical_name, metadata) + VALUES ($1, $2, $3)""", + bank_id, + "John Smith", + '{"role": "engineer"}', + ) + + # Get original data original_unit = await conn.fetchrow( - """SELECT id, embedding, event_date, created_at, metadata, text - FROM memory_units WHERE bank_id = $1 LIMIT 1""", + f"""SELECT id, embedding, event_date, created_at, metadata, text + FROM {_fq('memory_units')} WHERE bank_id = $1 LIMIT 1""", bank_id, ) - - # entities: UUID (id), Timestamp (first_seen, last_seen), JSONB (metadata) original_entity = await conn.fetchrow( - """SELECT id, first_seen, last_seen, metadata, canonical_name - FROM entities WHERE bank_id = $1 LIMIT 1""", + f"""SELECT id, first_seen, last_seen, metadata, canonical_name + FROM {_fq('entities')} WHERE bank_id = $1 LIMIT 1""", bank_id, ) - - # banks: JSONB (personality/disposition) original_bank = await conn.fetchrow( - "SELECT bank_id, created_at, updated_at FROM banks WHERE bank_id = $1", + f"SELECT bank_id, created_at, updated_at FROM {_fq('banks')} WHERE bank_id = $1", bank_id, ) + finally: + await conn.close() assert original_unit is not None, "Should have created memory units" assert original_unit["embedding"] is not None, "Should have embedding" @@ -153,33 +233,37 @@ async def test_backup_restore_preserves_all_column_types(memory, pg0_db_url, req backup_path = Path(f.name) try: - await _backup(pg0_db_url, backup_path) + await _backup(db_url, backup_path, schema=schema_name) # Clear all data - async with memory._pool.acquire() as conn: + conn = await asyncpg.connect(db_url) + try: for table in reversed(BACKUP_TABLES): - await conn.execute(f"TRUNCATE TABLE {table} CASCADE") + await conn.execute(f"TRUNCATE TABLE {_fq(table)} CASCADE") + finally: + await conn.close() - await _restore(pg0_db_url, backup_path) + await _restore(db_url, backup_path, schema=schema_name) # Verify all column types are preserved exactly - async with memory._pool.acquire() as conn: + conn = await asyncpg.connect(db_url) + try: restored_unit = await conn.fetchrow( - """SELECT id, embedding, event_date, created_at, metadata, text - FROM memory_units WHERE bank_id = $1 LIMIT 1""", + f"""SELECT id, embedding, event_date, created_at, metadata, text + FROM {_fq('memory_units')} WHERE bank_id = $1 LIMIT 1""", bank_id, ) - restored_entity = await conn.fetchrow( - """SELECT id, first_seen, last_seen, metadata, canonical_name - FROM entities WHERE bank_id = $1 LIMIT 1""", + f"""SELECT id, first_seen, last_seen, metadata, canonical_name + FROM {_fq('entities')} WHERE bank_id = $1 LIMIT 1""", bank_id, ) - restored_bank = await conn.fetchrow( - "SELECT bank_id, created_at, updated_at FROM banks WHERE bank_id = $1", + f"SELECT bank_id, created_at, updated_at FROM {_fq('banks')} WHERE bank_id = $1", bank_id, ) + finally: + await conn.close() # Verify memory_units assert restored_unit is not None, "Should have restored memory unit" diff --git a/hindsight-docs/docs/developer/admin-cli.md b/hindsight-docs/docs/developer/admin-cli.md new file mode 100644 index 00000000..eee17675 --- /dev/null +++ b/hindsight-docs/docs/developer/admin-cli.md @@ -0,0 +1,145 @@ +# Admin CLI + +The `hindsight-admin` CLI provides administrative commands for managing your Hindsight deployment, including database migrations, backup, and restore operations. + +## Installation + +The admin CLI is included with the `hindsight-api` package: + +```bash +pip install hindsight-api +# or +uv add hindsight-api +``` + +## Commands + +### run-db-migration + +Run database migrations to the latest version. This is useful when you want to run migrations separately from API startup (e.g., in CI/CD pipelines or before deploying a new version). + +```bash +hindsight-admin run-db-migration [OPTIONS] +``` + +**Options:** + +| Option | Description | Default | +|--------|-------------|---------| +| `--schema`, `-s` | Database schema to run migrations on | `public` | + +**Examples:** + +```bash +# Run migrations on the default public schema +hindsight-admin run-db-migration + +# Run migrations on a specific tenant schema +hindsight-admin run-db-migration --schema tenant_acme +``` + +:::tip Disabling Auto-Migrations +To disable automatic migrations on API startup, set `HINDSIGHT_API_RUN_MIGRATIONS_ON_STARTUP=false`. This is useful when you want to run migrations as a separate step in your deployment pipeline. +::: + +--- + +### backup + +Create a backup of all Hindsight data to a zip file. + +```bash +hindsight-admin backup OUTPUT [OPTIONS] +``` + +**Arguments:** + +| Argument | Description | +|----------|-------------| +| `OUTPUT` | Output file path (will add `.zip` extension if not present) | + +**Options:** + +| Option | Description | Default | +|--------|-------------|---------| +| `--schema`, `-s` | Database schema to backup | `public` | + +**Examples:** + +```bash +# Backup to a file +hindsight-admin backup /backups/hindsight-2024-01-15.zip + +# Backup a specific tenant schema +hindsight-admin backup /backups/tenant-acme.zip --schema tenant_acme +``` + +The backup includes: +- Memory banks and their configuration +- Documents and chunks +- Entities and their relationships +- Memory units (facts, experiences, opinions, observations) +- Entity cooccurrences and memory links + +:::note Consistency +Backups are created within a database transaction with `REPEATABLE READ` isolation, ensuring a consistent snapshot across all tables. +::: + +--- + +### restore + +Restore data from a backup file. **Warning: This deletes all existing data in the target schema.** + +```bash +hindsight-admin restore INPUT [OPTIONS] +``` + +**Arguments:** + +| Argument | Description | +|----------|-------------| +| `INPUT` | Input backup file (.zip) | + +**Options:** + +| Option | Description | Default | +|--------|-------------|---------| +| `--schema`, `-s` | Database schema to restore to | `public` | +| `--yes`, `-y` | Skip confirmation prompt | `false` | + +**Examples:** + +```bash +# Restore with confirmation prompt +hindsight-admin restore /backups/hindsight-2024-01-15.zip + +# Restore without confirmation (for scripts) +hindsight-admin restore /backups/hindsight-2024-01-15.zip --yes + +# Restore to a specific tenant schema +hindsight-admin restore /backups/tenant-acme.zip --schema tenant_acme --yes +``` + +:::warning Data Loss +Restore will **delete all existing data** in the target schema before importing the backup. Always verify you have a recent backup before performing a restore. +::: + +--- + +## Environment Variables + +The admin CLI uses the same environment variables as the API service. The most important one is: + +| Variable | Description | Default | +|----------|-------------|---------| +| `HINDSIGHT_API_DATABASE_URL` | PostgreSQL connection string | `pg0` (embedded) | + +**Example:** + +```bash +# Use a specific database +export HINDSIGHT_API_DATABASE_URL=postgresql://user:pass@localhost:5432/hindsight +hindsight-admin backup /backups/mybackup.zip +``` + diff --git a/hindsight-docs/docs/developer/configuration.md b/hindsight-docs/docs/developer/configuration.md index 9ef9593a..43135449 100644 --- a/hindsight-docs/docs/developer/configuration.md +++ b/hindsight-docs/docs/developer/configuration.md @@ -20,9 +20,18 @@ The API service handles all memory operations (retain, recall, reflect). | Variable | Description | Default | |----------|-------------|---------| | `HINDSIGHT_API_DATABASE_URL` | PostgreSQL connection string | `pg0` (embedded) | +| `HINDSIGHT_API_RUN_MIGRATIONS_ON_STARTUP` | Run database migrations on API startup | `true` | If not provided, the server uses embedded `pg0` — convenient for development but not recommended for production. +To run migrations manually (e.g., before starting the API), use the admin CLI: + +```bash +hindsight-admin run-db-migration +# Or for a specific schema: +hindsight-admin run-db-migration --schema tenant_acme +``` + ### LLM Provider | Variable | Description | Default | diff --git a/hindsight-docs/sidebars.ts b/hindsight-docs/sidebars.ts index 011694c7..a392a314 100644 --- a/hindsight-docs/sidebars.ts +++ b/hindsight-docs/sidebars.ts @@ -111,6 +111,11 @@ const sidebars: SidebarsConfig = { id: 'developer/configuration', label: 'Configuration', }, + { + type: 'doc', + id: 'developer/admin-cli', + label: 'Admin CLI', + }, { type: 'doc', id: 'developer/extensions',