feat(auth): add create bank validation hook

Add a precise operation-validator hook for bank creation, with a
no-op default so deployments without custom validators keep existing
behavior.

Route lazy bank creation through the hook from retain, imports, MCP
create_bank, and the default get_bank_profile auto-create path. This
keeps create-bank authorization separate from bank-scoped write
validation, which often assumes the target bank already exists.

Add regression coverage for rejected creation, existing-bank skips,
HTTP create/import paths, async retain, profile auto-create, and MCP
create_bank.

Author: Sanderhoff-alt

hindsight-api-slim/hindsight_api/api/http.py

@@ -5588,8 +5588,12 @@ async def api_create_or_update_bank(
     ):
         """Create or update an agent with disposition and mission."""
         try:
-            # Ensure bank exists by getting profile (auto-creates with defaults)
-            await app.state.memory.get_bank_profile(bank_id, request_context=request_context)
+            # Ensure bank exists, validating create_bank only when this call
+            # actually creates a missing bank.
+            await app.state.memory._ensure_bank_exists(
+                bank_id,
+                request_context,
+            )
 
             # Update name if provided (stored in DB for display only, deprecated)
             if request.name is not None:
@@ -5772,8 +5776,12 @@ async def api_import_bank_template(
                     dry_run=True,
                 )
 
-            # Ensure bank exists (auto-creates with defaults if needed)
-            await app.state.memory.get_bank_profile(bank_id, request_context=request_context)
+            # Ensure bank exists, validating create_bank only when this import
+            # actually creates a missing target bank.
+            await app.state.memory._ensure_bank_exists(
+                bank_id,
+                request_context,
+            )
 
             return await apply_bank_template_manifest(
                 memory=app.state.memory,

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

@@ -3223,6 +3223,8 @@ async def retain_batch_async(
             if result and result.contents is not None:
                 contents = cast(list[RetainContentDict], result.contents)
 
+        await self._ensure_bank_exists(bank_id, request_context)
+
         # Engine-owned copy: the orchestrator clears per-item "content" strings
         # after building the document's combined text (memory pressure
         # optimization, see retain/orchestrator.py). Without an internal copy
@@ -3709,6 +3711,14 @@ async def import_bank_async(
         # target bank's config before the restore.
         parsed = parse_bank_archive(archive_bytes)
         bank_id = target_bank_id or parsed.manifest.source_bank_id
+        if self._operation_validator and await bank_utils.get_bank_profile_if_exists(backend, bank_id) is None:
+            from hindsight_api.extensions import CreateBankContext
+
+            ctx = CreateBankContext(
+                bank_id=bank_id,
+                request_context=request_context,
+            )
+            await self._validate_operation(self._operation_validator.validate_create_bank(ctx))
         resolved_config = await self._config_resolver.resolve_full_config(bank_id, request_context)
         return await import_bank(
             backend=backend,
@@ -8401,16 +8411,12 @@ async def get_bank_profile(
             existing = await bank_utils.get_bank_profile_if_exists(backend, bank_id)
             if existing is None:
                 return None
-            profile, created = existing, False
+            profile = existing
         else:
-            result = await bank_utils.get_or_create_bank_profile(backend, bank_id)
-            profile, created = result.profile, result.created
-
-        # Apply HINDSIGHT_API_DEFAULT_BANK_TEMPLATE to freshly-created banks. Done
-        # before reading the resolved config below so the template's overrides
-        # (e.g. reflect_mission, dispositions) are visible on this very call.
-        if created:
-            await self._apply_default_bank_template(bank_id, request_context)
+            await self._ensure_bank_exists(bank_id, request_context)
+            profile = await bank_utils.get_bank_profile_if_exists(backend, bank_id)
+            if profile is None:
+                raise RuntimeError(f"Bank '{bank_id}' was not found after ensuring it exists")
 
         # reflect_mission and disposition in config take precedence over the legacy DB columns
         config_dict = await self._config_resolver.get_bank_config(bank_id, request_context)
@@ -8466,6 +8472,20 @@ async def _ensure_bank_exists(
             True if the bank was freshly created on this call.
         """
         backend = await self._get_backend()
+        if self._operation_validator:
+            if conn is not None:
+                exists = await conn.fetchval(f"SELECT 1 FROM {fq_table('banks')} WHERE bank_id = $1", bank_id)
+            else:
+                exists = await bank_utils.get_bank_profile_if_exists(backend, bank_id)
+            if not exists:
+                from hindsight_api.extensions import CreateBankContext
+
+                ctx = CreateBankContext(
+                    bank_id=bank_id,
+                    request_context=request_context,
+                )
+                await self._validate_operation(self._operation_validator.validate_create_bank(ctx))
+
         if conn is not None:
             result = await bank_utils.get_or_create_bank_profile_on_conn(conn, bank_id, ops=backend.ops)
             return result.created
@@ -10181,7 +10201,11 @@ async def create_mental_model(
         # outlives a mental-model insert that ultimately fails.
         async with acquire_with_retry(backend) as conn:
             async with conn.transaction():
-                created = await self._ensure_bank_exists(bank_id, request_context, conn=conn)
+                created = await self._ensure_bank_exists(
+                    bank_id,
+                    request_context,
+                    conn=conn,
+                )
                 if mental_model_id:
                     row = await conn.fetchrow(
                         f"""
@@ -11773,7 +11797,11 @@ async def create_webhook(
         # commit (or roll back) atomically.
         async with acquire_with_retry(backend) as conn:
             async with conn.transaction():
-                created = await self._ensure_bank_exists(bank_id, request_context, conn=conn)
+                created = await self._ensure_bank_exists(
+                    bank_id,
+                    request_context,
+                    conn=conn,
+                )
                 row = await backend.ops.create_webhook(
                     conn,
                     fq_table("webhooks"),
@@ -12214,7 +12242,11 @@ async def submit_async_retain(
                 # async_operations.bank_id has a FK to banks. Create the bank
                 # lazily inside this same transaction so it is atomic with the
                 # parent + child operation rows.
-                created = await self._ensure_bank_exists(bank_id, request_context, conn=conn)
+                created = await self._ensure_bank_exists(
+                    bank_id,
+                    request_context,
+                    conn=conn,
+                )
                 await conn.execute(
                     f"""
                     INSERT INTO {fq_table("async_operations")} (operation_id, bank_id, operation_type, result_metadata, status)

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

@@ -43,6 +43,7 @@
     # Consolidation operation
     ConsolidateContext,
     ConsolidateResult,
+    CreateBankContext,
     # File Conversion
     FileConvertResult,
     # Mental Model operations
@@ -99,6 +100,7 @@
     "BankListResult",
     "BankReadContext",
     "BankWriteContext",
+    "CreateBankContext",
     # Operation Validator - Consolidation
     "ConsolidateContext",
     "ConsolidateResult",

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

@@ -321,6 +321,14 @@ class BankWriteContext:
     request_context: "RequestContext"
 
 
+@dataclass
+class CreateBankContext:
+    """Context for validating creation of a new bank."""
+
+    bank_id: str
+    request_context: "RequestContext"
+
+
 @dataclass
 class BankListContext:
     """Context for filtering the bank list (post-query)."""
@@ -805,6 +813,23 @@ async def validate_bank_write(self, ctx: BankWriteContext) -> ValidationResult:
         """
         return ValidationResult.accept()
 
+    async def validate_create_bank(self, ctx: CreateBankContext) -> ValidationResult:
+        """
+        Validate creation of a new bank before the bank row is inserted.
+
+        Override to implement custom validation logic for operations that
+        explicitly or implicitly create a bank.
+
+        Args:
+            ctx: Context containing:
+                - bank_id: Bank identifier
+                - request_context: Request context with auth info
+
+        Returns:
+            ValidationResult indicating whether the bank may be created.
+        """
+        return ValidationResult.accept()
+
     async def filter_bank_list(self, ctx: BankListContext) -> BankListResult:
         """
         Filter the bank list after querying.

hindsight-api-slim/hindsight_api/mcp_tools.py

@@ -1209,7 +1209,9 @@ async def create_bank(bank_id: str, name: str | None = None, mission: str | None
         """
         try:
             request_context = _get_request_context(config)
-            # get_bank_profile auto-creates bank if it doesn't exist
+            # create_bank may auto-create the bank; validate that explicit
+            # creation permission before reading the resulting profile.
+            await memory._ensure_bank_exists(bank_id, request_context)
             profile = await memory.get_bank_profile(bank_id, request_context=request_context)
 
             # Update name/mission if provided

hindsight-api-slim/tests/test_extensions.py

@@ -9,6 +9,10 @@
 from hindsight_api.extensions import (
     ApiKeyTenantExtension,
     AuthenticationError,
+    # Consolidation operation
+    ConsolidateContext,
+    ConsolidateResult,
+    CreateBankContext,
     Extension,
     HttpExtension,
     OperationValidationError,
@@ -25,9 +29,6 @@
     TenantExtension,
     ValidationResult,
     load_extension,
-    # Consolidation operation
-    ConsolidateContext,
-    ConsolidateResult,
 )
 
 
@@ -169,6 +170,30 @@ async def on_consolidate_complete(self, result: ConsolidateResult) -> None:
         self.post_consolidate_calls.append(result)
 
 
+class CreateBankRejectingValidator(OperationValidatorExtension):
+    """Validator that records create-bank checks and can reject them."""
+
+    def __init__(self, *, reject: bool = True):
+        super().__init__({})
+        self.reject = reject
+        self.create_bank_calls: list[CreateBankContext] = []
+
+    async def validate_retain(self, ctx: RetainContext) -> ValidationResult:
+        return ValidationResult.accept()
+
+    async def validate_recall(self, ctx: RecallContext) -> ValidationResult:
+        return ValidationResult.accept()
+
+    async def validate_reflect(self, ctx: ReflectContext) -> ValidationResult:
+        return ValidationResult.accept()
+
+    async def validate_create_bank(self, ctx: CreateBankContext) -> ValidationResult:
+        self.create_bank_calls.append(ctx)
+        if self.reject:
+            return ValidationResult.reject("bank creation not allowed", status_code=402)
+        return ValidationResult.accept()
+
+
 class TestMemoryEngineValidation:
     """Tests for validation integration with MemoryEngine.
 
@@ -264,6 +289,109 @@ async def test_reflect_validation(self, memory_with_validator):
 
         assert "limit exceeded" in str(exc_info.value).lower()
 
+    @pytest.mark.asyncio
+    async def test_retain_validates_create_bank_for_missing_bank(self, memory):
+        """Retain may lazily create a bank, so missing-bank retain validates create_bank."""
+        bank_id = "test-retain-create-bank-validation"
+        ctx = RequestContext()
+        validator = CreateBankRejectingValidator()
+        memory._operation_validator = validator
+
+        with pytest.raises(OperationValidationError) as exc_info:
+            await memory.retain_batch_async(
+                bank_id=bank_id,
+                contents=[{"content": "Should not create a bank"}],
+                request_context=ctx,
+            )
+
+        assert "bank creation not allowed" in str(exc_info.value)
+        assert len(validator.create_bank_calls) == 1
+        assert validator.create_bank_calls[0].bank_id == bank_id
+        assert await memory.get_bank_profile(bank_id, request_context=ctx, create_if_missing=False) is None
+
+    @pytest.mark.asyncio
+    async def test_async_retain_validates_create_bank_for_missing_bank(self, memory):
+        """Async retain validates create_bank before creating its parent operation."""
+        bank_id = "test-async-retain-create-bank-validation"
+        ctx = RequestContext()
+        validator = CreateBankRejectingValidator()
+        memory._operation_validator = validator
+
+        with pytest.raises(OperationValidationError) as exc_info:
+            await memory.submit_async_retain(
+                bank_id=bank_id,
+                contents=[{"content": "Should not create a bank"}],
+                request_context=ctx,
+            )
+
+        assert "bank creation not allowed" in str(exc_info.value)
+        assert len(validator.create_bank_calls) == 1
+        assert validator.create_bank_calls[0].bank_id == bank_id
+        assert await memory.get_bank_profile(bank_id, request_context=ctx, create_if_missing=False) is None
+
+    @pytest.mark.asyncio
+    async def test_create_bank_validation_skips_existing_bank(self, memory):
+        """Existing banks do not require create_bank validation."""
+        bank_id = "test-create-bank-existing"
+        ctx = RequestContext()
+        await memory.get_bank_profile(bank_id, request_context=ctx)
+
+        validator = CreateBankRejectingValidator()
+        memory._operation_validator = validator
+
+        created = await memory._ensure_bank_exists(bank_id, ctx)
+
+        assert created is False
+        assert validator.create_bank_calls == []
+
+    @pytest.mark.asyncio
+    async def test_get_bank_profile_validates_create_bank_for_missing_bank(self, memory):
+        """The default auto-create profile path validates create_bank."""
+        bank_id = "test-profile-create-bank-validation"
+        ctx = RequestContext()
+        validator = CreateBankRejectingValidator()
+        memory._operation_validator = validator
+
+        with pytest.raises(OperationValidationError) as exc_info:
+            await memory.get_bank_profile(bank_id, request_context=ctx)
+
+        assert "bank creation not allowed" in str(exc_info.value)
+        assert len(validator.create_bank_calls) == 1
+        assert validator.create_bank_calls[0].bank_id == bank_id
+        assert await memory.get_bank_profile(bank_id, request_context=ctx, create_if_missing=False) is None
+
+    @pytest.mark.asyncio
+    async def test_http_create_or_update_validates_create_bank(self, api_client, memory):
+        bank_id = "test-http-put-create-bank-validation"
+        validator = CreateBankRejectingValidator()
+        memory._operation_validator = validator
+
+        resp = await api_client.put(
+            f"/v1/default/banks/{bank_id}",
+            json={"name": "Blocked Bank"},
+        )
+
+        assert resp.status_code == 402
+        assert resp.json()["detail"] == "bank creation not allowed"
+        assert len(validator.create_bank_calls) == 1
+        assert validator.create_bank_calls[0].bank_id == bank_id
+
+    @pytest.mark.asyncio
+    async def test_http_template_import_validates_create_bank(self, api_client, memory):
+        bank_id = "test-http-import-create-bank-validation"
+        validator = CreateBankRejectingValidator()
+        memory._operation_validator = validator
+
+        resp = await api_client.post(
+            f"/v1/default/banks/{bank_id}/import",
+            json={"version": "1"},
+        )
+
+        assert resp.status_code == 402
+        assert resp.json()["detail"] == "bank creation not allowed"
+        assert len(validator.create_bank_calls) == 1
+        assert validator.create_bank_calls[0].bank_id == bank_id
+
 
 @pytest.fixture
 def memory_with_validator(memory):
@@ -356,6 +484,7 @@ async def test_retain_post_hook_receives_all_parameters_and_result(self, memory_
     async def test_recall_pre_hook_receives_all_parameters(self, memory_with_tracking_validator):
         """Pre-recall hook receives all user-provided parameters."""
         from datetime import datetime, timezone
+
         from hindsight_api.engine.memory_engine import Budget
 
         memory, validator = memory_with_tracking_validator

hindsight-api-slim/tests/test_mcp_tools.py

@@ -188,6 +188,7 @@ async def _get_mental_model(**kwargs):
 
     # Tags & bank methods
     memory.list_tags = AsyncMock(return_value={"items": ["tag1", "tag2"], "total": 2})
+    memory._ensure_bank_exists = AsyncMock(return_value=True)
     memory.get_bank_profile = AsyncMock(return_value={"id": "test-bank", "name": "Test Bank", "mission": "Testing"})
     memory.get_bank_stats = AsyncMock(return_value={"nodes": 100, "links": 50})
     memory.update_bank = AsyncMock(return_value={"id": "test-bank", "name": "Updated"})
@@ -1512,6 +1513,14 @@ async def test_get_bank(self, mock_memory):
         result = await _tools(mcp)["get_bank"].fn()
         assert '"test-bank"' in result or "test-bank" in result
 
+    async def test_create_bank_validates_creation(self, mock_memory):
+        mcp = _make_mcp_server(mock_memory, {"create_bank"}, include_bank_id=True)
+        result = await _tools(mcp)["create_bank"].fn(bank_id="new-bank")
+        assert '"test-bank"' in result or "test-bank" in result
+        call = mock_memory._ensure_bank_exists.call_args
+        assert call.args[0] == "new-bank"
+        assert "operation" not in call.kwargs
+
     async def test_get_bank_stats(self, mock_memory):
         mcp = _make_mcp_server(mock_memory, {"get_bank_stats"}, include_bank_id=True)
         result = await _tools(mcp)["get_bank_stats"].fn()