feat(extensions): declare + provision extension-owned bank-scoped tables (#2903)

* feat(extensions): let extensions declare bank-scoped tables for backup + teardown

An extension can provision its own bank-scoped tables in the tenant schema
(audit receipts, per-bank policy state, ...), but core knows nothing about
them, so they silently fall out of the per-tenant data-lifecycle operations it
owns:

- admin backup/restore copies a fixed core table set and TRUNCATEs it CASCADE
  on restore; an extension table absent from that set is dropped from the
  backup and — if it FKs banks — wiped by the cascade with no way back;
- delete_bank clears a bank via core deletes + the banks FK cascade; an
  extension table scoping by bank_id without a cascading FK leaks orphaned rows.

Add a BankScopedTable descriptor and TenantExtension.extra_bank_tables() so an
extension declares its tables; core consults them in:

- admin backup/restore (_effective_backup_tables appends declared tables after
  the core set so restore's forward COPY / reversed TRUNCATE keep FK order);
- MemoryEngine.delete_bank (sweeps declared tables by bank_id on full delete,
  with a PG-only to_regclass guard so a declared-but-unprovisioned table can't
  abort the delete).

The extension still owns the DDL; this only tells core which tables to sweep.
Default behaviour is unchanged — the base method returns no tables, so the OSS
default path is a no-op. Descriptor names are validated to a safe SQL
identifier shape since they're interpolated into SQL.

Covered by descriptor-validation + effective-list unit tests, a delete_bank
sweep test, and a backup/restore round-trip that proves a declared extension
table survives truncate+restore.

* feat(extensions): provision extension bank tables on the migration path

Adds the creation half of the bank-scoped-table lifecycle. Previously an
extension's tables were created only by its own imperative DDL run lazily on
first request (e.g. Cloud's provision_schema off authenticate), so:
  - hindsight-admin run-db-migration migrated core schema across all tenants
    but never touched extension tables, and
  - a provisioning failure was swallowed, surfacing later as a runtime error.

Add TenantExtension.provision_bank_tables(conn, schema) — idempotent DDL the
extension owns — and invoke it right after core migrations from both migration
entry points:
  - ExtensionContext.run_migration (every tenant-schema provision), and
  - the run-db-migration sweep (_provision_extra_bank_tables, per schema),
    where a failure now aborts the command and names the schema instead of
    being swallowed.

So extension schema evolves on the same lifecycle as core schema. Default is a
no-op, so the OSS default path is unchanged. Pairs with extra_bank_tables()
(declares for backup/teardown) — one creates, the other declares.

Covered by a default-no-op test plus provisioning through both the CLI sweep
helper and ExtensionContext.run_migration against real Postgres.

* chore: ruff format after rebase (cli.py, memory_engine.py)

hindsight-api-slim/hindsight_api/admin/cli.py

@@ -131,6 +131,27 @@ async def _validate_restore_schema(
     return restore_columns
 
 
+def _effective_backup_tables() -> list[str]:
+    """Core backup tables plus any bank-scoped tables a loaded extension declares.
+
+    ``BACKUP_TABLES`` covers only the tables core owns. An extension that
+    provisions its own bank-scoped tables (via ``TenantExtension``) declares
+    them through ``extra_bank_tables()`` so they aren't dropped on restore.
+    Extension tables are appended *after* the core set so restore's forward
+    COPY inserts them after their FK parents (e.g. ``banks``) and the reversed
+    TRUNCATE clears them before those parents.
+    """
+    tables = list(BACKUP_TABLES)
+    tenant_extension = load_extension("TENANT", TenantExtension)
+    if tenant_extension is not None:
+        seen = set(tables)
+        for spec in tenant_extension.extra_bank_tables():
+            if spec.include_in_backup and spec.name not in seen:
+                tables.append(spec.name)
+                seen.add(spec.name)
+    return tables
+
+
 async def _admin_connect(db_url: str) -> asyncpg.Connection:
     """Open a raw asyncpg connection to an admin DB URL.
 
@@ -149,8 +170,18 @@ async def _admin_connect(db_url: str) -> asyncpg.Connection:
     return conn
 
 
-async def _backup(database_url: str, output_path: Path, schema: str = "public") -> dict[str, Any]:
-    """Backup all tables to a zip file using binary COPY protocol."""
+async def _backup(
+    database_url: str,
+    output_path: Path,
+    schema: str = "public",
+    backup_tables: list[str] | None = None,
+) -> dict[str, Any]:
+    """Backup all tables to a zip file using binary COPY protocol.
+
+    ``backup_tables`` defaults to the core ``BACKUP_TABLES``; callers pass the
+    extension-augmented list from ``_effective_backup_tables()``.
+    """
+    backup_tables = backup_tables if backup_tables is not None else BACKUP_TABLES
     conn = await asyncpg.connect(database_url)
     try:
         tables: dict[str, Any] = {}
@@ -167,8 +198,8 @@ async def _backup(database_url: str, output_path: Path, schema: str = "public")
         # entities table was backed up.
         async with conn.transaction(isolation="repeatable_read"):
             with zipfile.ZipFile(output_path, "w", zipfile.ZIP_DEFLATED) as zf:
-                for i, table in enumerate(BACKUP_TABLES, 1):
-                    typer.echo(f"  [{i}/{len(BACKUP_TABLES)}] Backing up {table}...", nl=False)
+                for i, table in enumerate(backup_tables, 1):
+                    typer.echo(f"  [{i}/{len(backup_tables)}] Backing up {table}...", nl=False)
 
                     buffer = io.BytesIO()
 
@@ -207,8 +238,20 @@ async def _backup(database_url: str, output_path: Path, schema: str = "public")
         await conn.close()
 
 
-async def _restore(database_url: str, input_path: Path, schema: str = "public") -> dict[str, Any]:
-    """Restore all tables from a zip file using binary COPY protocol."""
+async def _restore(
+    database_url: str,
+    input_path: Path,
+    schema: str = "public",
+    backup_tables: list[str] | None = None,
+) -> dict[str, Any]:
+    """Restore all tables from a zip file using binary COPY protocol.
+
+    ``backup_tables`` defaults to the core ``BACKUP_TABLES``; callers pass the
+    extension-augmented list from ``_effective_backup_tables()``. Tables named
+    here but absent from the archive are truncated then skipped for restore, so
+    a stale extension registration never leaves pre-restore rows behind.
+    """
+    backup_tables = backup_tables if backup_tables is not None else BACKUP_TABLES
     conn = await asyncpg.connect(database_url)
     try:
         with zipfile.ZipFile(input_path, "r") as zf:
@@ -227,19 +270,19 @@ async def _restore(database_url: str, input_path: Path, schema: str = "public")
             async with conn.transaction():
                 typer.echo("  Clearing existing data...")
                 # Truncate tables in reverse order (respects FK constraints)
-                for table in reversed(BACKUP_TABLES):
+                for table in reversed(backup_tables):
                     qualified_table = _fq_table(table, schema)
                     await conn.execute(f"TRUNCATE TABLE {qualified_table} CASCADE")
 
                 # Restore tables in forward order
-                for i, table in enumerate(BACKUP_TABLES, 1):
+                for i, table in enumerate(backup_tables, 1):
                     filename = f"{table}.bin"
                     if filename not in zf.namelist():
-                        typer.echo(f"  [{i}/{len(BACKUP_TABLES)}] {table}: skipped (not in backup)")
+                        typer.echo(f"  [{i}/{len(backup_tables)}] {table}: skipped (not in backup)")
                         continue
 
                     expected_rows = manifest["tables"].get(table, {}).get("rows", "?")
-                    typer.echo(f"  [{i}/{len(BACKUP_TABLES)}] Restoring {table}... {expected_rows} rows")
+                    typer.echo(f"  [{i}/{len(backup_tables)}] Restoring {table}... {expected_rows} rows")
 
                     data = zf.read(filename)
                     buffer = io.BytesIO(data)
@@ -268,7 +311,7 @@ async def _run_backup(db_url: str, output: Path, schema: str = "public") -> dict
     if is_pg0:
         typer.echo(f"Starting embedded PostgreSQL (instance: {instance_name})...")
     resolved_url = await resolve_database_url(db_url)
-    return await _backup(resolved_url, output, schema)
+    return await _backup(resolved_url, output, schema, backup_tables=_effective_backup_tables())
 
 
 async def _run_restore(db_url: str, input_file: Path, schema: str = "public") -> dict[str, Any]:
@@ -278,7 +321,7 @@ async def _run_restore(db_url: str, input_file: Path, schema: str = "public") ->
     if is_pg0:
         typer.echo(f"Starting embedded PostgreSQL (instance: {instance_name})...")
     resolved_url = await resolve_database_url(db_url)
-    return await _restore(resolved_url, input_file, schema)
+    return await _restore(resolved_url, input_file, schema, backup_tables=_effective_backup_tables())
 
 
 @app.command()
@@ -302,7 +345,7 @@ def backup(
     manifest = asyncio.run(_run_backup(config.database_url, output, schema))
 
     total_rows = sum(t["rows"] for t in manifest["tables"].values())
-    typer.echo(f"Backed up {total_rows} rows across {len(BACKUP_TABLES)} tables")
+    typer.echo(f"Backed up {total_rows} rows across {len(manifest['tables'])} tables")
     typer.echo(f"Backup saved to {output}")
 
 
@@ -335,7 +378,7 @@ def restore(
     manifest = asyncio.run(_run_restore(config.database_url, input_file, schema))
 
     total_rows = sum(t["rows"] for t in manifest["tables"].values())
-    typer.echo(f"Restored {total_rows} rows across {len(BACKUP_TABLES)} tables")
+    typer.echo(f"Restored {total_rows} rows across {len(manifest['tables'])} tables")
     typer.echo("Restore complete")
 
 
@@ -356,11 +399,10 @@ async def _run_migration(
     resolved_url = await resolve_database_url(db_url)
 
     config = HindsightConfig.from_env()
+    tenant_extension = load_extension("TENANT", TenantExtension)
     if schema:
         schemas = [schema]
     else:
-        tenant_extension = load_extension("TENANT", TenantExtension)
-
         schemas = [base_schema or DEFAULT_DATABASE_SCHEMA]
         if tenant_extension:
             tenants = await tenant_extension.list_tenants()
@@ -385,9 +427,36 @@ async def _run_migration(
         ensure_extensions=ensure_extensions,
     )
 
+    # After core migrations, provision any extension-owned bank-scoped tables
+    # per schema so extension schema evolves on the same lifecycle as core
+    # schema (rather than via a lazy first-request path).
+    if tenant_extension is not None:
+        await _provision_extra_bank_tables(resolved_url, schemas, tenant_extension)
+
     return schemas
 
 
+async def _provision_extra_bank_tables(
+    resolved_url: str, schemas: list[str], tenant_extension: TenantExtension
+) -> None:
+    """Run the tenant extension's table provisioner for each migrated schema.
+
+    Fires after core migrations complete so extension-owned bank tables are
+    created/evolved on the same lifecycle as core schema. A failure aborts the
+    migration command (and names the offending schema) rather than being
+    swallowed — provisioning is idempotent, so the operator can fix and re-run.
+    """
+    for schema in schemas:
+        conn = await asyncpg.connect(resolved_url)
+        try:
+            await tenant_extension.provision_bank_tables(conn, schema)
+        except Exception as e:
+            typer.echo(f"  Failed to provision extension tables for schema '{schema}': {e}", err=True)
+            raise
+        finally:
+            await conn.close()
+
+
 @app.command(name="run-db-migration")
 def run_db_migration(
     schema: str | None = typer.Option(

hindsight-api-slim/hindsight_api/engine/memory_engine.py

@@ -6376,6 +6376,28 @@ class MemoryEngine(MemoryEngineInterface):
                         # Delete entities (cascades to unit_entities, entity_cooccurrences, memory_links with entity_id)
                         await conn.execute(f"DELETE FROM {fq_table('entities')} WHERE bank_id = $1", bank_id)
 
+                        # Sweep extension-owned bank-scoped tables (audit receipts,
+                        # per-bank policy state, ...). These scope by bank_id without
+                        # a cascading FK to banks, so deleting the bank row below
+                        # would otherwise leave them as orphaned rows.
+                        extra_tables = self._tenant_extension.extra_bank_tables() if self._tenant_extension else []
+                        if extra_tables:
+                            from .schema import _is_oracle  # noqa: PLC0415
+
+                            for spec in extra_tables:
+                                if not spec.delete_with_bank:
+                                    continue
+                                qualified = fq_table(spec.name)
+                                # PG-only existence guard: a declared-but-unprovisioned
+                                # table must not abort the whole bank delete. (to_regclass
+                                # is PG syntax; extension bank tables are a PG feature.)
+                                if (
+                                    not _is_oracle()
+                                    and await conn.fetchval("SELECT to_regclass($1)", qualified) is None
+                                ):
+                                    continue
+                                await conn.execute(f"DELETE FROM {qualified} WHERE {spec.bank_id_column} = $1", bank_id)
+
                         result = {
                             "memory_units_deleted": units_count,
                             "entities_deleted": entities_count,

hindsight-api-slim/hindsight_api/extensions/__init__.py

@@ -15,6 +15,7 @@ Extensions receive an ExtensionContext that provides a controlled API for intera
 with the system (e.g., running migrations for tenant schemas).
 """
 
+from hindsight_api.extensions.bank_tables import BankScopedTable
 from hindsight_api.extensions.base import Extension
 from hindsight_api.extensions.builtin import (
     ApiKeyTenantExtension,
@@ -78,6 +79,7 @@ from hindsight_api.worker.exceptions import DeferOperation
 __all__ = [
     # Base
     "Extension",
+    "BankScopedTable",
     "load_extension",
     # Context
     "ExtensionContext",

hindsight-api-slim/hindsight_api/extensions/bank_tables.py

@@ -0,0 +1,60 @@
+"""Extension-declared, bank-scoped tables.
+
+An extension may provision its own tables in the tenant schema (e.g. audit
+receipts, per-bank policy state). Those tables are invisible to core, so they
+silently fall out of the per-tenant data-lifecycle operations core owns:
+
+* **Backup / restore** — ``hindsight-admin backup``/``restore`` copies a fixed
+  set of core tables and ``TRUNCATE ... CASCADE``\\ s them on restore. An
+  extension table absent from that set is dropped from the backup *and* — if it
+  carries a FK to ``banks`` — wiped by the cascade with no way to restore it.
+* **Bank teardown** — :meth:`MemoryEngine.delete_bank` clears a bank by
+  deleting the core rows and letting ``banks``' FK cascade handle the rest. An
+  extension table that scopes by ``bank_id`` without a cascading FK leaks
+  orphaned rows when the bank is deleted.
+
+An extension declares its bank-scoped tables via
+:meth:`TenantExtension.extra_bank_tables`; core consults that list in the
+operations above. The extension still owns the DDL (creation lives in its
+provisioning path) — this descriptor only tells core which tables to sweep.
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+
+# Unquoted SQL identifiers only. ``name`` and ``bank_id_column`` are
+# interpolated into SQL (schema-qualified via ``fq_table``), so they must be
+# validated to a safe identifier shape rather than trusted verbatim.
+_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
+
+
+@dataclass(frozen=True)
+class BankScopedTable:
+    """A bank-scoped table an extension owns and core should sweep.
+
+    Args:
+        name: Unqualified table name. Schema-qualified at use via ``fq_table``.
+        bank_id_column: Column holding the bank id, used to scope a per-bank
+            delete. Defaults to ``"bank_id"``.
+        include_in_backup: Include the table in ``hindsight-admin``
+            backup/restore. Defaults to ``True`` — a bank-scoped table almost
+            always wants restore coverage; opt out only for regenerable or
+            transient state.
+        delete_with_bank: Delete the table's rows for a bank during a full
+            :meth:`MemoryEngine.delete_bank`. Defaults to ``True``. Set
+            ``False`` to retain rows that should outlive the bank (e.g. audit
+            receipts a compliance regime requires kept).
+    """
+
+    name: str
+    bank_id_column: str = "bank_id"
+    include_in_backup: bool = True
+    delete_with_bank: bool = True
+
+    def __post_init__(self) -> None:
+        if not _IDENTIFIER_RE.match(self.name):
+            raise ValueError(f"BankScopedTable.name {self.name!r} is not a valid SQL identifier")
+        if not _IDENTIFIER_RE.match(self.bank_id_column):
+            raise ValueError(f"BankScopedTable.bank_id_column {self.bank_id_column!r} is not a valid SQL identifier")

hindsight-api-slim/hindsight_api/extensions/context.py

@@ -160,6 +160,19 @@ class DefaultExtensionContext(ExtensionContext):
             schema=schema,
         )
 
+        # Provision any extension-owned bank-scoped tables for this schema,
+        # right after core migrations, so extension schema evolves on the same
+        # lifecycle as core schema (instead of via a lazy per-request path).
+        # No-op unless a tenant extension declares a provisioner; errors
+        # propagate so a failed provision surfaces here, not at request time.
+        engine = self._memory_engine
+        get_pool = getattr(engine, "_get_pool", None)
+        tenant_extension = getattr(engine, "tenant_extension", None)
+        if get_pool is not None and tenant_extension is not None:
+            pool = await get_pool()
+            async with pool.acquire() as conn:
+                await tenant_extension.provision_bank_tables(conn, schema)
+
     def get_memory_engine(self) -> "MemoryEngineInterface":
         """Get the memory engine interface."""
         if self._memory_engine is None:

hindsight-api-slim/hindsight_api/extensions/tenant.py

@@ -2,11 +2,15 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
+from hindsight_api.extensions.bank_tables import BankScopedTable
 from hindsight_api.extensions.base import Extension
 from hindsight_api.models import RequestContext
 
+if TYPE_CHECKING:
+    import asyncpg
+
 
 class AuthenticationError(Exception):
     """Raised when authentication fails."""
@@ -143,6 +147,50 @@ class TenantExtension(Extension, ABC):
         """
         return None
 
+    def extra_bank_tables(self) -> list[BankScopedTable]:
+        """Bank-scoped tables this extension provisions in the tenant schema.
+
+        Core consults this list so extension-owned tables participate in the
+        per-tenant data-lifecycle operations it manages — admin backup/restore
+        and :meth:`MemoryEngine.delete_bank` teardown — instead of silently
+        falling out of them (dropped on restore, or leaked as orphaned rows on
+        bank deletion). See :class:`BankScopedTable`.
+
+        The extension still owns the DDL; this only declares which tables exist.
+        The default is no extra tables.
+
+        Returns:
+            The extension's bank-scoped tables. Empty by default.
+        """
+        return []
+
+    async def provision_bank_tables(self, conn: "asyncpg.Connection", schema: str) -> None:
+        """Create/evolve this extension's tables in ``schema`` (idempotent DDL).
+
+        Called from the **migration path** for every schema — both when a
+        tenant schema is provisioned (via ``ExtensionContext.run_migration``)
+        and by the ``hindsight-admin run-db-migration`` sweep across all
+        existing schemas — right after core migrations complete. This is the
+        counterpart to :meth:`extra_bank_tables`: that one *declares* the
+        tables for backup/teardown, this one *creates* them, so extension
+        schema evolves on the same lifecycle as core schema instead of via a
+        lazy per-request path.
+
+        Implementations MUST be idempotent (``CREATE TABLE IF NOT EXISTS`` /
+        ``ADD COLUMN IF NOT EXISTS``) — it runs on every provision and every
+        migration sweep — and MUST schema-qualify every statement with
+        ``schema`` (the connection's ``search_path`` is not set for you).
+        Exceptions propagate so a failed provision surfaces at migration time
+        rather than as a runtime error on a later request.
+
+        Args:
+            conn: An open connection to the target database.
+            schema: The schema to provision tables into.
+
+        The default does nothing.
+        """
+        return None
+
     async def authenticate_mcp(self, context: RequestContext) -> TenantContext:
         """
         Authenticate MCP requests.

hindsight-api-slim/tests/test_admin_backup_restore.py

@@ -10,6 +10,7 @@ import tempfile
 import uuid
 import zipfile
 from pathlib import Path
+from unittest.mock import AsyncMock
 
 import asyncpg
 import pytest
@@ -474,6 +475,69 @@ async def test_restore_succeeds_when_target_has_additional_nullable_column(backu
             backup_path.unlink()
 
 
+@pytest.mark.asyncio
+async def test_backup_restore_includes_extension_table(backup_test_schema):
+    """An extension-declared bank-scoped table rides along backup + restore.
+
+    Simulates a table an extension provisions in the tenant schema (core knows
+    nothing about it). Passing the augmented ``backup_tables`` list — as
+    ``_effective_backup_tables()`` builds from ``TenantExtension.extra_bank_tables``
+    — must back it up AND restore it, so restore's ``TRUNCATE ... CASCADE`` can't
+    silently drop it.
+    """
+    db_url, schema_name, _fq, _embeddings = backup_test_schema
+    extra = "ext_audit_receipts"
+    effective = [*BACKUP_TABLES, extra]
+
+    conn = await asyncpg.connect(db_url)
+    try:
+        await conn.execute(f"CREATE TABLE {_fq(extra)} (id uuid PRIMARY KEY, bank_id text NOT NULL, payload text)")
+        kept_id = uuid.uuid4()
+        await conn.execute(
+            f"INSERT INTO {_fq(extra)} (id, bank_id, payload) VALUES ($1, $2, $3)",
+            kept_id,
+            "bank-x",
+            "original receipt",
+        )
+    finally:
+        await conn.close()
+
+    with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as f:
+        backup_path = Path(f.name)
+
+    try:
+        manifest = await _backup(db_url, backup_path, schema=schema_name, backup_tables=effective)
+        assert manifest["tables"][extra]["rows"] == 1
+
+        # Mutate after backup: a row that must NOT survive restore.
+        conn = await asyncpg.connect(db_url)
+        try:
+            await conn.execute(
+                f"INSERT INTO {_fq(extra)} (id, bank_id, payload) VALUES ($1, $2, $3)",
+                uuid.uuid4(),
+                "bank-x",
+                "post-backup row",
+            )
+        finally:
+            await conn.close()
+
+        await _restore(db_url, backup_path, schema=schema_name, backup_tables=effective)
+
+        conn = await asyncpg.connect(db_url)
+        try:
+            rows = await conn.fetch(f"SELECT id, payload FROM {_fq(extra)}")
+        finally:
+            await conn.close()
+
+        # Restore reset the table to exactly its backed-up contents.
+        assert len(rows) == 1
+        assert rows[0]["id"] == kept_id
+        assert rows[0]["payload"] == "original receipt"
+    finally:
+        if backup_path.exists():
+            backup_path.unlink()
+
+
 @pytest.mark.asyncio
 async def test_run_migration_without_schema_discovers_and_deduplicates_schemas(monkeypatch):
     """run-db-migration without --schema should include the base schema and deduplicate tenant schemas."""
@@ -515,6 +579,8 @@ async def test_run_migration_without_schema_discovers_and_deduplicates_schemas(m
     monkeypatch.setenv("HINDSIGHT_API_DATABASE_URL", "postgresql://test")
     monkeypatch.setattr(admin_cli, "load_extension", lambda *args, **kwargs: MockTenantExtension())
     monkeypatch.setattr(admin_cli, "resolve_database_url", fake_resolve_database_url)
+    # Extension-table provisioning does a real connect; these tests mock the DB, so stub it.
+    monkeypatch.setattr(admin_cli, "_provision_extra_bank_tables", AsyncMock())
 
     from hindsight_api import migrations as migrations_module
 
@@ -585,6 +651,8 @@ async def test_run_migration_without_schema_runs_optional_post_migration_hooks(m
 
     monkeypatch.setattr(admin_cli, "load_extension", lambda *args, **kwargs: MockTenantExtension())
     monkeypatch.setattr(admin_cli, "resolve_database_url", fake_resolve_database_url)
+    # Extension-table provisioning does a real connect; these tests mock the DB, so stub it.
+    monkeypatch.setattr(admin_cli, "_provision_extra_bank_tables", AsyncMock())
 
     from hindsight_api import migrations as migrations_module
 
@@ -655,6 +723,8 @@ async def test_run_migration_with_schema_only_runs_requested_schema(monkeypatch)
 
     monkeypatch.setattr(admin_cli, "load_extension", lambda *args, **kwargs: MockTenantExtension())
     monkeypatch.setattr(admin_cli, "resolve_database_url", fake_resolve_database_url)
+    # Extension-table provisioning does a real connect; these tests mock the DB, so stub it.
+    monkeypatch.setattr(admin_cli, "_provision_extra_bank_tables", AsyncMock())
 
     from hindsight_api import migrations as migrations_module
 
@@ -693,6 +763,8 @@ async def test_run_migration_threads_ensure_extensions_flag(monkeypatch, ensure_
 
     monkeypatch.setattr(admin_cli, "load_extension", lambda *args, **kwargs: None)
     monkeypatch.setattr(admin_cli, "resolve_database_url", fake_resolve_database_url)
+    # Extension-table provisioning does a real connect; these tests mock the DB, so stub it.
+    monkeypatch.setattr(admin_cli, "_provision_extra_bank_tables", AsyncMock())
 
     from hindsight_api import migrations as migrations_module
 

hindsight-api-slim/tests/test_extension_bank_tables.py

@@ -0,0 +1,212 @@
+"""Tests for the extension bank-scoped-table registry seam.
+
+Extensions declare bank-scoped tables via ``TenantExtension.extra_bank_tables``
+so those tables participate in core's per-tenant data lifecycle:
+
+* admin backup/restore includes them (``_effective_backup_tables``);
+* ``MemoryEngine.delete_bank`` sweeps them so no orphaned rows survive.
+
+The descriptor validation and the backup-list computation are pure-Python; the
+delete_bank sweep runs against a real Postgres via the ``memory`` fixture.
+"""
+
+import uuid
+
+import asyncpg
+import pytest
+
+from hindsight_api import RequestContext
+from hindsight_api.admin import cli as admin_cli
+from hindsight_api.admin.cli import BACKUP_TABLES, _effective_backup_tables, _provision_extra_bank_tables
+from hindsight_api.engine.memory_engine import MemoryEngine
+from hindsight_api.extensions.bank_tables import BankScopedTable
+from hindsight_api.extensions.tenant import Tenant, TenantContext, TenantExtension
+
+
+class _StubTenant(TenantExtension):
+    """Minimal TenantExtension that only declares extra bank tables."""
+
+    def __init__(self, specs: list[BankScopedTable]):
+        super().__init__(config={})
+        self._specs = specs
+
+    async def authenticate(self, context: RequestContext) -> TenantContext:  # pragma: no cover - unused
+        raise NotImplementedError
+
+    async def list_tenants(self) -> list[Tenant]:  # pragma: no cover - unused
+        return []
+
+    def extra_bank_tables(self) -> list[BankScopedTable]:
+        return self._specs
+
+
+class _ProvisioningTenant(_StubTenant):
+    """Stub whose provisioner creates a marker table so we can assert it ran."""
+
+    def __init__(self, marker_table: str):
+        super().__init__([])
+        self._marker_table = marker_table
+
+    async def provision_bank_tables(self, conn: asyncpg.Connection, schema: str) -> None:
+        await conn.execute(f'CREATE TABLE IF NOT EXISTS "{schema}".{self._marker_table} (id int)')
+
+
+# ---------------------------------------------------------------------------
+# Descriptor validation
+# ---------------------------------------------------------------------------
+
+
+def test_defaults_and_valid_identifier():
+    spec = BankScopedTable(name="privacy_events")
+    assert spec.bank_id_column == "bank_id"
+    assert spec.include_in_backup is True
+    assert spec.delete_with_bank is True
+
+
+@pytest.mark.parametrize("bad", ["has-dash", "with space", "semi;colon", "", "1leading", "a.b"])
+def test_invalid_table_name_rejected(bad):
+    with pytest.raises(ValueError, match="not a valid SQL identifier"):
+        BankScopedTable(name=bad)
+
+
+def test_invalid_bank_id_column_rejected():
+    with pytest.raises(ValueError, match="not a valid SQL identifier"):
+        BankScopedTable(name="ok_table", bank_id_column="bank id")
+
+
+# ---------------------------------------------------------------------------
+# Effective backup-table computation
+# ---------------------------------------------------------------------------
+
+
+def test_effective_backup_tables_no_extension(monkeypatch):
+    monkeypatch.setattr(admin_cli, "load_extension", lambda *a, **k: None)
+    assert _effective_backup_tables() == BACKUP_TABLES
+
+
+def test_effective_backup_tables_appends_after_core(monkeypatch):
+    ext = _StubTenant(
+        [
+            BankScopedTable(name="privacy_events"),
+            BankScopedTable(name="privacy_exports", include_in_backup=False),  # excluded
+            BankScopedTable(name="banks"),  # dup of a core table — deduped
+        ]
+    )
+    monkeypatch.setattr(admin_cli, "load_extension", lambda *a, **k: ext)
+
+    result = _effective_backup_tables()
+
+    # Core list preserved in order and comes first.
+    assert result[: len(BACKUP_TABLES)] == BACKUP_TABLES
+    # Only backup-participating, non-duplicate extension tables appended.
+    assert result[len(BACKUP_TABLES) :] == ["privacy_events"]
+    assert "privacy_exports" not in result
+    assert result.count("banks") == 1
+
+
+# ---------------------------------------------------------------------------
+# delete_bank sweep (real DB)
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_delete_bank_sweeps_extension_tables(memory: MemoryEngine, request_context: RequestContext, monkeypatch):
+    swept = "ext_receipts_swept"
+    kept = "ext_receipts_kept"
+    pool = await memory._get_pool()
+    async with pool.acquire() as conn:
+        for tbl in (swept, kept):
+            await conn.execute(f"CREATE TABLE IF NOT EXISTS {tbl} (id uuid PRIMARY KEY, bank_id text NOT NULL)")
+            await conn.execute(f"TRUNCATE {tbl}")
+
+    bank_a = f"test-ext-a-{uuid.uuid4().hex[:8]}"
+    bank_b = f"test-ext-b-{uuid.uuid4().hex[:8]}"
+    await memory.get_bank_profile(bank_id=bank_a, request_context=request_context)
+    await memory.get_bank_profile(bank_id=bank_b, request_context=request_context)
+
+    async with pool.acquire() as conn:
+        for tbl in (swept, kept):
+            for bank in (bank_a, bank_b):
+                await conn.execute(f"INSERT INTO {tbl} (id, bank_id) VALUES ($1, $2)", uuid.uuid4(), bank)
+
+    # Declare the tables on the live tenant extension; `kept` opts out of
+    # teardown. Patch only extra_bank_tables so authentication still works.
+    specs = [
+        BankScopedTable(name=swept),
+        BankScopedTable(name=kept, delete_with_bank=False),
+        BankScopedTable(name="does_not_exist_table"),  # unprovisioned → skipped, must not error
+    ]
+    monkeypatch.setattr(memory._tenant_extension, "extra_bank_tables", lambda: specs)
+
+    await memory.delete_bank(bank_a, request_context=request_context)
+
+    async with pool.acquire() as conn:
+        # `swept` lost bank_a's rows, kept bank_b's.
+        assert await conn.fetchval(f"SELECT count(*) FROM {swept} WHERE bank_id = $1", bank_a) == 0
+        assert await conn.fetchval(f"SELECT count(*) FROM {swept} WHERE bank_id = $1", bank_b) == 1
+        # `kept` opted out — both banks' rows survive.
+        assert await conn.fetchval(f"SELECT count(*) FROM {kept} WHERE bank_id = $1", bank_a) == 1
+
+        await conn.execute(f"DROP TABLE IF EXISTS {swept}")
+        await conn.execute(f"DROP TABLE IF EXISTS {kept}")
+
+    await memory.delete_bank(bank_b, request_context=request_context)
+
+
+# ---------------------------------------------------------------------------
+# Creation hook: provision_bank_tables on the migration path
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_provision_bank_tables_default_is_noop():
+    """The base provisioner does nothing (and never touches the connection)."""
+    ext = _StubTenant([])
+    # A no-op must not touch conn — passing None proves it.
+    await ext.provision_bank_tables(None, "public")  # type: ignore[arg-type]
+
+
+@pytest.mark.asyncio
+async def test_cli_migration_sweep_provisions_extension_tables(pg0_db_url):
+    """The run-db-migration sweep helper runs the extension provisioner per schema."""
+    schema = f"prov_cli_{uuid.uuid4().hex[:8]}"
+    conn = await asyncpg.connect(pg0_db_url)
+    try:
+        await conn.execute(f'CREATE SCHEMA "{schema}"')
+    finally:
+        await conn.close()
+
+    ext = _ProvisioningTenant("cli_marker")
+    try:
+        await _provision_extra_bank_tables(pg0_db_url, [schema], ext)
+
+        conn = await asyncpg.connect(pg0_db_url)
+        try:
+            assert await conn.fetchval("SELECT to_regclass($1)", f"{schema}.cli_marker") is not None
+        finally:
+            await conn.close()
+    finally:
+        conn = await asyncpg.connect(pg0_db_url)
+        try:
+            await conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')
+        finally:
+            await conn.close()
+
+
+@pytest.mark.asyncio
+async def test_run_migration_provisions_extension_tables(memory: MemoryEngine, monkeypatch):
+    """ExtensionContext.run_migration provisions extension tables right after core
+    migrations, so tenant provisioning creates them (not a lazy request path)."""
+    schema = f"prov_ctx_{uuid.uuid4().hex[:8]}"
+    monkeypatch.setattr(
+        memory._tenant_extension, "provision_bank_tables", _ProvisioningTenant("ctx_marker").provision_bank_tables
+    )
+
+    pool = await memory._get_pool()
+    try:
+        await memory._ext_ctx.run_migration(schema)
+        async with pool.acquire() as conn:
+            assert await conn.fetchval("SELECT to_regclass($1)", f"{schema}.ctx_marker") is not None
+    finally:
+        async with pool.acquire() as conn:
+            await conn.execute(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE')