fix(webhooks): route webhook endpoints through tenant-aware engine methods

Webhook create/list/get/update/delete and list-deliveries endpoints in
the HTTP layer were calling pool.fetchrow/pool.fetch directly with
fq_table("webhooks"), bypassing the async-local schema context that
fq_table reads via get_current_schema(). Under deployments that set a
per-request target schema (multi-tenant routing), this caused webhooks
to be written to and read from the default schema while every other
operation on the same bank correctly resolved to the per-target
schema. Webhooks would land in the wrong schema; the fire path
(which uses the bank's resolved schema) would not see them and never
enqueued webhook_delivery operations -- silent failure, no errors.

Move the SQL into MemoryEngine methods that call _authenticate_tenant
first (matching the pattern used by retain/consolidate/mental-models),
so fq_table sees the same schema as the rest of the bank's data.

Add schema-isolation tests covering create/list/get/update/delete and
deliveries.

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

@@ -5408,29 +5408,17 @@ def _register_routes(app: FastAPI):
     ):
         """Register a webhook for a bank."""
         try:
-            backend = await app.state.memory._get_backend()
-            from hindsight_api.engine.db_utils import acquire_with_retry
-            from hindsight_api.engine.memory_engine import fq_table
-            from hindsight_api.engine.retain import bank_utils
-
-            # Ensure the bank row exists before inserting into webhooks (FK constraint).
-            _, created = await bank_utils.get_or_create_bank_profile(backend, bank_id)
-            if created:
-                await app.state.memory._apply_default_bank_template(bank_id, request_context)
-
             webhook_id = uuid.uuid4()
-            async with acquire_with_retry(backend) as conn:
-                row = await backend.ops.create_webhook(
-                    conn,
-                    fq_table("webhooks"),
-                    webhook_id,
-                    bank_id,
-                    request.url,
-                    request.secret,
-                    request.event_types,
-                    request.enabled,
-                    request.http_config.model_dump_json(),
-                )
+            row = await app.state.memory.create_webhook(
+                bank_id,
+                webhook_id=webhook_id,
+                url=request.url,
+                secret=request.secret,
+                event_types=request.event_types,
+                enabled=request.enabled,
+                http_config_json=request.http_config.model_dump_json(),
+                request_context=request_context,
+            )
 
             event_types_val = row["event_types"] if row else []
             if isinstance(event_types_val, str):
@@ -5479,16 +5467,10 @@ def _register_routes(app: FastAPI):
     ):
         """List webhooks for a bank."""
         try:
-            backend = await app.state.memory._get_backend()
-            from hindsight_api.engine.db_utils import acquire_with_retry
-            from hindsight_api.engine.memory_engine import fq_table
-
-            async with acquire_with_retry(backend) as conn:
-                rows = await backend.ops.list_webhooks_for_bank(
-                    conn,
-                    fq_table("webhooks"),
-                    bank_id,
-                )
+            rows = await app.state.memory.list_webhooks(
+                bank_id,
+                request_context=request_context,
+            )
 
             def _parse_webhook_row(row):
                 event_types_val = row["event_types"]
@@ -5541,17 +5523,11 @@ def _register_routes(app: FastAPI):
     ):
         """Delete a webhook."""
         try:
-            backend = await app.state.memory._get_backend()
-            from hindsight_api.engine.db_utils import acquire_with_retry
-            from hindsight_api.engine.memory_engine import fq_table
-
-            async with acquire_with_retry(backend) as conn:
-                deleted = await backend.ops.delete_webhook(
-                    conn,
-                    fq_table("webhooks"),
-                    uuid.UUID(webhook_id),
-                    bank_id,
-                )
+            deleted = await app.state.memory.delete_webhook(
+                bank_id,
+                uuid.UUID(webhook_id),
+                request_context=request_context,
+            )
             if not deleted:
                 raise HTTPException(status_code=404, detail="Webhook not found")
             return DeleteResponse(success=True)
@@ -5581,10 +5557,6 @@ def _register_routes(app: FastAPI):
     ):
         """Update a webhook's fields (PATCH semantics — only sent fields are updated)."""
         try:
-            backend = await app.state.memory._get_backend()
-            from hindsight_api.engine.db_utils import acquire_with_retry
-            from hindsight_api.engine.memory_engine import fq_table
-
             set_clauses: list[str] = []
             params: list = [uuid.UUID(webhook_id), bank_id]
 
@@ -5608,15 +5580,13 @@ def _register_routes(app: FastAPI):
             if not set_clauses:
                 raise HTTPException(status_code=422, detail="No fields provided to update")
 
-            async with acquire_with_retry(backend) as conn:
-                row = await backend.ops.update_webhook(
-                    conn,
-                    fq_table("webhooks"),
-                    uuid.UUID(webhook_id),
-                    bank_id,
-                    set_clauses,
-                    params,
-                )
+            row = await app.state.memory.update_webhook(
+                bank_id,
+                uuid.UUID(webhook_id),
+                set_clauses=set_clauses,
+                params=params,
+                request_context=request_context,
+            )
             if not row:
                 raise HTTPException(status_code=404, detail="Webhook not found")
 
@@ -5670,28 +5640,16 @@ def _register_routes(app: FastAPI):
     ):
         """List deliveries for a specific webhook, newest first. Use next_cursor for pagination."""
         try:
-            backend = await app.state.memory._get_backend()
-            from hindsight_api.engine.db_utils import acquire_with_retry
-            from hindsight_api.engine.memory_engine import fq_table
-
-            async with acquire_with_retry(backend) as conn:
-                # Verify webhook belongs to this bank
-                webhook_row = await conn.fetchrow(
-                    f"SELECT id FROM {fq_table('webhooks')} WHERE id = $1 AND bank_id = $2",
+            try:
+                rows = await app.state.memory.list_webhook_deliveries(
+                    bank_id,
                     uuid.UUID(webhook_id),
-                    bank_id,
-                )
-                if not webhook_row:
-                    raise HTTPException(status_code=404, detail="Webhook not found")
-
-                rows = await backend.ops.list_webhook_deliveries(
-                    conn,
-                    fq_table("async_operations"),
-                    webhook_id,
-                    bank_id,
-                    limit,
-                    cursor,
+                    limit=limit,
+                    cursor=cursor,
+                    request_context=request_context,
                 )
+            except LookupError:
+                raise HTTPException(status_code=404, detail="Webhook not found")
 
             has_more = len(rows) > limit
             page = rows[:limit]

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

@@ -8757,6 +8757,198 @@ class MemoryEngine(MemoryEngineInterface):
         # Return updated profile
         return await self.get_bank_profile(bank_id, request_context=request_context)
 
+    # =========================================================================
+    # Webhook configuration methods
+    #
+    # These wrap the raw backend.ops.* calls used to be invoked directly from
+    # the HTTP layer with ``fq_table("webhooks")``. Computing the fully-qualified
+    # table name from the HTTP layer evaluates ``fq_table`` before the schema
+    # contextvar is set, which means under deployments that resolve a
+    # per-request target schema (multi-target-schema routing) the webhook rows
+    # would land in the default schema while the rest of the bank's data lives
+    # in a per-target schema. The fire path uses the bank's resolved schema
+    # and would silently never see those webhook rows.
+    #
+    # Routing through engine methods that call ``_authenticate_tenant`` first
+    # ensures ``fq_table`` resolves to the same schema used by retain,
+    # consolidate, and every other bank-scoped operation.
+    # =========================================================================
+
+    async def create_webhook(
+        self,
+        bank_id: str,
+        *,
+        webhook_id: uuid.UUID,
+        url: str,
+        secret: str | None,
+        event_types: list[str],
+        enabled: bool,
+        http_config_json: str,
+        request_context: "RequestContext",
+    ) -> dict[str, Any]:
+        """Insert a webhook row in the bank's resolved schema.
+
+        Authenticates the tenant first so ``fq_table("webhooks")`` resolves to
+        the same schema as the rest of the bank's data.
+        """
+        await self._authenticate_tenant(request_context)
+        if self._operation_validator:
+            from hindsight_api.extensions import BankWriteContext
+
+            ctx = BankWriteContext(bank_id=bank_id, operation="create_webhook", request_context=request_context)
+            await self._validate_operation(self._operation_validator.validate_bank_write(ctx))
+
+        backend = await self._get_backend()
+
+        # Ensure the bank row exists before inserting into webhooks (FK constraint).
+        _, created = await bank_utils.get_or_create_bank_profile(backend, bank_id)
+        if created:
+            await self._apply_default_bank_template(bank_id, request_context)
+
+        async with acquire_with_retry(backend) as conn:
+            row = await backend.ops.create_webhook(
+                conn,
+                fq_table("webhooks"),
+                webhook_id,
+                bank_id,
+                url,
+                secret,
+                event_types,
+                enabled,
+                http_config_json,
+            )
+        return dict(row) if row is not None else None
+
+    async def list_webhooks(
+        self,
+        bank_id: str,
+        *,
+        request_context: "RequestContext",
+    ) -> list[dict[str, Any]]:
+        """List webhooks for a bank in the bank's resolved schema."""
+        await self._authenticate_tenant(request_context)
+        if self._operation_validator:
+            from hindsight_api.extensions import BankReadContext
+
+            ctx = BankReadContext(bank_id=bank_id, operation="list_webhooks", request_context=request_context)
+            await self._validate_operation(self._operation_validator.validate_bank_read(ctx))
+
+        backend = await self._get_backend()
+        async with acquire_with_retry(backend) as conn:
+            rows = await backend.ops.list_webhooks_for_bank(
+                conn,
+                fq_table("webhooks"),
+                bank_id,
+            )
+        return [dict(row) for row in rows]
+
+    async def update_webhook(
+        self,
+        bank_id: str,
+        webhook_id: uuid.UUID,
+        *,
+        set_clauses: list[str],
+        params: list[Any],
+        request_context: "RequestContext",
+    ) -> dict[str, Any] | None:
+        """Update a webhook row in the bank's resolved schema.
+
+        ``set_clauses`` and ``params`` are pre-built by the caller using PATCH
+        semantics (only sent fields are updated). The first two ``params`` are
+        ``webhook_id`` and ``bank_id``; subsequent params correspond to the
+        SET clauses.
+        """
+        await self._authenticate_tenant(request_context)
+        if self._operation_validator:
+            from hindsight_api.extensions import BankWriteContext
+
+            ctx = BankWriteContext(bank_id=bank_id, operation="update_webhook", request_context=request_context)
+            await self._validate_operation(self._operation_validator.validate_bank_write(ctx))
+
+        backend = await self._get_backend()
+        async with acquire_with_retry(backend) as conn:
+            row = await backend.ops.update_webhook(
+                conn,
+                fq_table("webhooks"),
+                webhook_id,
+                bank_id,
+                set_clauses,
+                params,
+            )
+        return dict(row) if row is not None else None
+
+    async def delete_webhook(
+        self,
+        bank_id: str,
+        webhook_id: uuid.UUID,
+        *,
+        request_context: "RequestContext",
+    ) -> bool:
+        """Delete a webhook row from the bank's resolved schema.
+
+        Returns True if a row was deleted, False if no matching row was found.
+        """
+        await self._authenticate_tenant(request_context)
+        if self._operation_validator:
+            from hindsight_api.extensions import BankWriteContext
+
+            ctx = BankWriteContext(bank_id=bank_id, operation="delete_webhook", request_context=request_context)
+            await self._validate_operation(self._operation_validator.validate_bank_write(ctx))
+
+        backend = await self._get_backend()
+        async with acquire_with_retry(backend) as conn:
+            return await backend.ops.delete_webhook(
+                conn,
+                fq_table("webhooks"),
+                webhook_id,
+                bank_id,
+            )
+
+    async def list_webhook_deliveries(
+        self,
+        bank_id: str,
+        webhook_id: uuid.UUID,
+        *,
+        limit: int,
+        cursor: str | None,
+        request_context: "RequestContext",
+    ) -> list[dict[str, Any]]:
+        """List webhook delivery rows from the bank's resolved schema.
+
+        First verifies the webhook belongs to this bank (in the same schema),
+        then reads the delivery rows from ``async_operations``. Returns up to
+        ``limit + 1`` rows so callers can determine whether more pages exist.
+
+        Raises:
+            LookupError: When the webhook does not exist in this bank.
+        """
+        await self._authenticate_tenant(request_context)
+        if self._operation_validator:
+            from hindsight_api.extensions import BankReadContext
+
+            ctx = BankReadContext(bank_id=bank_id, operation="list_webhook_deliveries", request_context=request_context)
+            await self._validate_operation(self._operation_validator.validate_bank_read(ctx))
+
+        backend = await self._get_backend()
+        async with acquire_with_retry(backend) as conn:
+            webhook_row = await conn.fetchrow(
+                f"SELECT id FROM {fq_table('webhooks')} WHERE id = $1 AND bank_id = $2",
+                webhook_id,
+                bank_id,
+            )
+            if not webhook_row:
+                raise LookupError("Webhook not found")
+
+            rows = await backend.ops.list_webhook_deliveries(
+                conn,
+                fq_table("async_operations"),
+                str(webhook_id),
+                bank_id,
+                limit,
+                cursor,
+            )
+        return [dict(row) for row in rows]
+
     async def _submit_async_operation(
         self,
         bank_id: str,

hindsight-api-slim/tests/test_webhooks.py

@@ -901,3 +901,429 @@ class TestRetainCompletedWebhook:
                     bank_id,
                 )
                 await conn.execute("DELETE FROM webhooks WHERE id = $1", webhook_id)
+
+
+# ---------------------------------------------------------------------------
+# Schema-isolation tests
+#
+# These tests verify that webhook CRUD endpoints honour the per-request schema
+# context set by the tenant extension, rather than always operating on the
+# default (public) schema. This is the regression test for the bug where the
+# HTTP handlers built ``fq_table("webhooks")`` before ``_authenticate_tenant``
+# had set the schema context, causing webhook rows to land in the wrong schema
+# under multi-target-schema deployments. The fire path correctly resolves the
+# bank's schema and would never see those rows, producing silent failures.
+# ---------------------------------------------------------------------------
+
+
+class _NonDefaultSchemaTenantExtension:
+    """Minimal tenant extension that always returns a fixed non-default schema.
+
+    Doesn't subclass ``TenantExtension`` because we only need ``authenticate``
+    for these tests; ``_authenticate_tenant`` calls just that method.
+    """
+
+    def __init__(self, schema_name: str):
+        self._schema_name = schema_name
+
+    async def authenticate(self, context):
+        from hindsight_api.extensions import TenantContext
+
+        return TenantContext(schema_name=self._schema_name)
+
+    async def list_tenants(self):
+        from hindsight_api.extensions.tenant import Tenant
+
+        return [Tenant(schema=self._schema_name)]
+
+
+@pytest_asyncio.fixture
+async def isolated_schema(memory: MemoryEngine, pg0_db_url):
+    """Provision a fresh non-default schema with the full migration tree, then
+    swap the memory engine's tenant extension so all subsequent operations
+    resolve to it. Drops the schema on teardown.
+    """
+    import asyncpg
+
+    from hindsight_api.migrations import run_migrations
+
+    schema_name = f"tenant_wh_iso_{uuid.uuid4().hex[:8]}"
+
+    # Run migrations to provision the schema with all tables (webhooks, banks,
+    # async_operations, ...). This is the same path a real multi-tenant
+    # extension would take to provision a new tenant schema.
+    run_migrations(pg0_db_url, schema=schema_name)
+
+    original_ext = memory._tenant_extension
+    memory._tenant_extension = _NonDefaultSchemaTenantExtension(schema_name)
+
+    try:
+        yield schema_name
+    finally:
+        memory._tenant_extension = original_ext
+        # Drop the test schema. Use a dedicated connection so we don't depend
+        # on the pool's state.
+        conn = await asyncpg.connect(pg0_db_url)
+        try:
+            await conn.execute(f'DROP SCHEMA IF EXISTS "{schema_name}" CASCADE')
+        finally:
+            await conn.close()
+
+
+class TestWebhookSchemaIsolation:
+    """Verify the webhook HTTP endpoints write to and read from the schema set
+    by the tenant extension, not the default (public) schema.
+    """
+
+    @pytest.mark.asyncio
+    async def test_create_webhook_lands_in_resolved_schema(
+        self, memory: MemoryEngine, api_client: httpx.AsyncClient, isolated_schema: str
+    ):
+        """POST /webhooks should insert into the resolved schema, not public."""
+        bank_id = f"http-wh-iso-{uuid.uuid4().hex[:8]}"
+
+        create_resp = await api_client.post(
+            f"/v1/default/banks/{bank_id}/webhooks",
+            json={"url": "https://example.com/iso", "event_types": ["consolidation.completed"]},
+        )
+        assert create_resp.status_code == 201, create_resp.text
+        webhook_id = create_resp.json()["id"]
+
+        # Row should exist in the resolved schema...
+        async with memory._pool.acquire() as conn:
+            row_in_target = await conn.fetchrow(
+                f'SELECT id, bank_id, url FROM "{isolated_schema}".webhooks WHERE id = $1',
+                uuid.UUID(webhook_id),
+            )
+            # ...and must NOT exist in public.
+            row_in_public = await conn.fetchrow(
+                "SELECT id FROM public.webhooks WHERE id = $1",
+                uuid.UUID(webhook_id),
+            )
+
+        assert row_in_target is not None, (
+            "Webhook row should be inserted into the resolved schema"
+        )
+        assert row_in_target["bank_id"] == bank_id
+        assert row_in_target["url"] == "https://example.com/iso"
+        assert row_in_public is None, (
+            "Webhook row must NOT be written to public when a non-default "
+            "schema is resolved by the tenant extension"
+        )
+
+    @pytest.mark.asyncio
+    async def test_list_webhooks_reads_from_resolved_schema(
+        self, memory: MemoryEngine, api_client: httpx.AsyncClient, isolated_schema: str
+    ):
+        """GET /webhooks should only return rows from the resolved schema.
+
+        We seed an unrelated row directly into public.webhooks for the same
+        bank_id and assert it does NOT appear in the list response.
+        """
+        bank_id = f"http-wh-iso-{uuid.uuid4().hex[:8]}"
+
+        # Create one webhook through the HTTP API (lands in the isolated schema)
+        create_resp = await api_client.post(
+            f"/v1/default/banks/{bank_id}/webhooks",
+            json={"url": "https://example.com/in-target", "event_types": ["consolidation.completed"]},
+        )
+        assert create_resp.status_code == 201
+        target_webhook_id = create_resp.json()["id"]
+
+        # Seed an unrelated webhook row directly into public.webhooks for the
+        # same bank — represents data that belongs to "another tenant".
+        public_webhook_id = uuid.uuid4()
+        async with memory._pool.acquire() as conn:
+            # public.banks may not have this bank; ensure the FK does not blow up.
+            await conn.execute(
+                "INSERT INTO public.banks (bank_id, name) VALUES ($1, $2) ON CONFLICT DO NOTHING",
+                bank_id,
+                bank_id,
+            )
+            await conn.execute(
+                """
+                INSERT INTO public.webhooks
+                  (id, bank_id, url, secret, event_types, enabled, created_at, updated_at)
+                VALUES ($1, $2, 'https://example.com/in-public', NULL, $3, true, NOW(), NOW())
+                """,
+                public_webhook_id,
+                bank_id,
+                ["consolidation.completed"],
+            )
+
+        try:
+            list_resp = await api_client.get(f"/v1/default/banks/{bank_id}/webhooks")
+            assert list_resp.status_code == 200
+            ids = {item["id"] for item in list_resp.json()["items"]}
+
+            assert target_webhook_id in ids, (
+                "list_webhooks should return rows from the resolved schema"
+            )
+            assert str(public_webhook_id) not in ids, (
+                "list_webhooks must NOT leak rows from public when a non-default "
+                "schema is resolved"
+            )
+        finally:
+            async with memory._pool.acquire() as conn:
+                await conn.execute(
+                    "DELETE FROM public.webhooks WHERE id = $1", public_webhook_id
+                )
+
+    @pytest.mark.asyncio
+    async def test_update_webhook_targets_resolved_schema(
+        self, memory: MemoryEngine, api_client: httpx.AsyncClient, isolated_schema: str
+    ):
+        """PATCH /webhooks/{id} should update the row in the resolved schema only."""
+        bank_id = f"http-wh-iso-{uuid.uuid4().hex[:8]}"
+
+        create_resp = await api_client.post(
+            f"/v1/default/banks/{bank_id}/webhooks",
+            json={"url": "https://example.com/before", "event_types": ["consolidation.completed"]},
+        )
+        assert create_resp.status_code == 201
+        webhook_id = create_resp.json()["id"]
+
+        # Seed a row with the SAME id in public (impossible in practice, but
+        # demonstrates that PATCH does not silently target public).
+        async with memory._pool.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO public.banks (bank_id, name) VALUES ($1, $2) ON CONFLICT DO NOTHING",
+                bank_id,
+                bank_id,
+            )
+            await conn.execute(
+                """
+                INSERT INTO public.webhooks
+                  (id, bank_id, url, secret, event_types, enabled, created_at, updated_at)
+                VALUES ($1, $2, 'https://example.com/public-stale', NULL, $3, true, NOW(), NOW())
+                """,
+                uuid.UUID(webhook_id),
+                bank_id,
+                ["consolidation.completed"],
+            )
+
+        try:
+            patch_resp = await api_client.patch(
+                f"/v1/default/banks/{bank_id}/webhooks/{webhook_id}",
+                json={"url": "https://example.com/after"},
+            )
+            assert patch_resp.status_code == 200
+            assert patch_resp.json()["url"] == "https://example.com/after"
+
+            async with memory._pool.acquire() as conn:
+                target_url = await conn.fetchval(
+                    f'SELECT url FROM "{isolated_schema}".webhooks WHERE id = $1',
+                    uuid.UUID(webhook_id),
+                )
+                public_url = await conn.fetchval(
+                    "SELECT url FROM public.webhooks WHERE id = $1",
+                    uuid.UUID(webhook_id),
+                )
+
+            assert target_url == "https://example.com/after"
+            # The public row must remain untouched - the update targeted the
+            # resolved schema, not public.
+            assert public_url == "https://example.com/public-stale"
+        finally:
+            async with memory._pool.acquire() as conn:
+                await conn.execute(
+                    "DELETE FROM public.webhooks WHERE id = $1", uuid.UUID(webhook_id)
+                )
+
+    @pytest.mark.asyncio
+    async def test_delete_webhook_targets_resolved_schema(
+        self, memory: MemoryEngine, api_client: httpx.AsyncClient, isolated_schema: str
+    ):
+        """DELETE /webhooks/{id} should remove the row from the resolved schema only."""
+        bank_id = f"http-wh-iso-{uuid.uuid4().hex[:8]}"
+
+        create_resp = await api_client.post(
+            f"/v1/default/banks/{bank_id}/webhooks",
+            json={"url": "https://example.com/del", "event_types": ["consolidation.completed"]},
+        )
+        assert create_resp.status_code == 201
+        webhook_id = create_resp.json()["id"]
+
+        # Seed a row with the same id into public to ensure DELETE doesn't
+        # accidentally target it.
+        async with memory._pool.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO public.banks (bank_id, name) VALUES ($1, $2) ON CONFLICT DO NOTHING",
+                bank_id,
+                bank_id,
+            )
+            await conn.execute(
+                """
+                INSERT INTO public.webhooks
+                  (id, bank_id, url, secret, event_types, enabled, created_at, updated_at)
+                VALUES ($1, $2, 'https://example.com/public-survivor', NULL, $3, true, NOW(), NOW())
+                """,
+                uuid.UUID(webhook_id),
+                bank_id,
+                ["consolidation.completed"],
+            )
+
+        try:
+            del_resp = await api_client.delete(
+                f"/v1/default/banks/{bank_id}/webhooks/{webhook_id}"
+            )
+            assert del_resp.status_code == 200
+            assert del_resp.json()["success"] is True
+
+            async with memory._pool.acquire() as conn:
+                target_row = await conn.fetchrow(
+                    f'SELECT id FROM "{isolated_schema}".webhooks WHERE id = $1',
+                    uuid.UUID(webhook_id),
+                )
+                public_row = await conn.fetchrow(
+                    "SELECT id FROM public.webhooks WHERE id = $1",
+                    uuid.UUID(webhook_id),
+                )
+
+            assert target_row is None, "row in resolved schema should have been deleted"
+            assert public_row is not None, (
+                "row in public must NOT be deleted when delete targets a non-default schema"
+            )
+        finally:
+            async with memory._pool.acquire() as conn:
+                await conn.execute(
+                    "DELETE FROM public.webhooks WHERE id = $1", uuid.UUID(webhook_id)
+                )
+
+    @pytest.mark.asyncio
+    async def test_list_deliveries_targets_resolved_schema(
+        self, memory: MemoryEngine, api_client: httpx.AsyncClient, isolated_schema: str
+    ):
+        """GET /webhooks/{id}/deliveries should only see deliveries in the resolved schema.
+
+        Specifically, if a webhook exists in public with the same id but NOT in
+        the resolved schema, the endpoint must return 404 — it must look up the
+        webhook in the resolved schema, not public.
+        """
+        bank_id = f"http-wh-iso-{uuid.uuid4().hex[:8]}"
+        orphan_webhook_id = uuid.uuid4()
+
+        # Seed a webhook ONLY in public (not in the resolved schema)
+        async with memory._pool.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO public.banks (bank_id, name) VALUES ($1, $2) ON CONFLICT DO NOTHING",
+                bank_id,
+                bank_id,
+            )
+            await conn.execute(
+                """
+                INSERT INTO public.webhooks
+                  (id, bank_id, url, secret, event_types, enabled, created_at, updated_at)
+                VALUES ($1, $2, 'https://example.com/orphan', NULL, $3, true, NOW(), NOW())
+                """,
+                orphan_webhook_id,
+                bank_id,
+                ["consolidation.completed"],
+            )
+
+        try:
+            resp = await api_client.get(
+                f"/v1/default/banks/{bank_id}/webhooks/{orphan_webhook_id}/deliveries"
+            )
+            # The webhook does not exist in the resolved schema, so this must 404
+            # — not silently fall through to public.
+            assert resp.status_code == 404, resp.text
+        finally:
+            async with memory._pool.acquire() as conn:
+                await conn.execute(
+                    "DELETE FROM public.webhooks WHERE id = $1", orphan_webhook_id
+                )
+
+    @pytest.mark.asyncio
+    async def test_list_deliveries_returns_rows_from_resolved_schema(
+        self, memory: MemoryEngine, api_client: httpx.AsyncClient, isolated_schema: str
+    ):
+        """GET /webhooks/{id}/deliveries should read async_operations from the resolved schema."""
+        bank_id = f"http-wh-iso-{uuid.uuid4().hex[:8]}"
+
+        create_resp = await api_client.post(
+            f"/v1/default/banks/{bank_id}/webhooks",
+            json={"url": "https://example.com/del-iso", "event_types": ["consolidation.completed"]},
+        )
+        assert create_resp.status_code == 201
+        webhook_id = create_resp.json()["id"]
+
+        # Seed a delivery row in the RESOLVED schema's async_operations table.
+        target_delivery_id = uuid.uuid4()
+        target_payload = json.dumps(
+            {
+                "type": "webhook_delivery",
+                "bank_id": bank_id,
+                "url": "https://example.com/del-iso",
+                "secret": None,
+                "event_type": "consolidation.completed",
+                "payload": '{"event":"consolidation.completed"}',
+                "webhook_id": webhook_id,
+            }
+        )
+        # Seed a confounding delivery row with the same payload->webhook_id in
+        # public.async_operations to make sure it is NOT returned.
+        public_delivery_id = uuid.uuid4()
+        public_payload = json.dumps(
+            {
+                "type": "webhook_delivery",
+                "bank_id": bank_id,
+                "url": "https://example.com/del-iso-public",
+                "secret": None,
+                "event_type": "consolidation.completed",
+                "payload": '{"event":"consolidation.completed"}',
+                "webhook_id": webhook_id,
+            }
+        )
+        now = datetime.now(timezone.utc)
+        async with memory._pool.acquire() as conn:
+            await conn.execute(
+                f"""
+                INSERT INTO "{isolated_schema}".async_operations
+                  (operation_id, bank_id, operation_type, status, retry_count,
+                   task_payload, result_metadata, created_at, updated_at)
+                VALUES ($1, $2, 'webhook_delivery', 'completed', 0,
+                        $3::jsonb, '{{}}'::jsonb, $4, $4)
+                """,
+                target_delivery_id,
+                bank_id,
+                target_payload,
+                now,
+            )
+            await conn.execute(
+                "INSERT INTO public.banks (bank_id, name) VALUES ($1, $2) ON CONFLICT DO NOTHING",
+                bank_id,
+                bank_id,
+            )
+            await conn.execute(
+                """
+                INSERT INTO public.async_operations
+                  (operation_id, bank_id, operation_type, status, retry_count,
+                   task_payload, result_metadata, created_at, updated_at)
+                VALUES ($1, $2, 'webhook_delivery', 'completed', 0,
+                        $3::jsonb, '{}'::jsonb, $4, $4)
+                """,
+                public_delivery_id,
+                bank_id,
+                public_payload,
+                now,
+            )
+
+        try:
+            resp = await api_client.get(
+                f"/v1/default/banks/{bank_id}/webhooks/{webhook_id}/deliveries"
+            )
+            assert resp.status_code == 200
+            ids = {item["id"] for item in resp.json()["items"]}
+            assert str(target_delivery_id) in ids, (
+                "deliveries from the resolved schema should be returned"
+            )
+            assert str(public_delivery_id) not in ids, (
+                "deliveries from public must NOT leak when a non-default schema is resolved"
+            )
+        finally:
+            async with memory._pool.acquire() as conn:
+                await conn.execute(
+                    "DELETE FROM public.async_operations WHERE operation_id = $1",
+                    public_delivery_id,
+                )