feat(vector-io): enforce metadata_store when access control is enabled

Gate vector store providers to require metadata_store (AuthorizedSqlStore)
when access control policies are active, ensuring tenant isolation cannot
be silently bypassed. Auto-inject metadata_store from server-level
storage.stores.vector_stores into all vector_io providers via the
resolver, so operators configure it once instead of per-provider. Fix
batch cleanup to use unfiltered SQL access for internal bookkeeping.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Signed-off-by: Sébastien Han <seb@redhat.com>

Author: leseb

.github/workflows/integration-responses-conversations-auth-tests.yml

@@ -167,13 +167,18 @@ jobs:
               prompts:
                 table_name: prompts
                 backend: sql_default
+              vector_stores:
+                table_name: vector_store_metadata
+                backend: sql_default
           models:
           - model_id: openai/gpt-4o
             model_type: llm
             provider_id: openai
           - model_id: sentence-transformers/nomic-ai/nomic-embed-text-v1.5
             model_type: embedding
             provider_id: sentence-transformers
+            metadata:
+              embedding_dimension: 768
           server:
             port: 8321
             auth:

src/ogx/core/resolver.py

@@ -400,10 +400,29 @@ async def instantiate_provider(
 
     logger.debug("Instantiating provider", provider_id=provider.provider_id, module=provider_spec.module)
     module = importlib.import_module(provider_spec.module)
+
+    def _inject_config_defaults(config_type: type[Any], provider_config: dict[str, Any]) -> dict[str, Any]:
+        fields = getattr(config_type, "__fields__", None)
+        if fields is None:
+            return provider_config
+
+        # Inject vector_stores_config for providers that need it (introspection-based).
+        # Only inject if vector_stores is provided, otherwise let default_factory handle it.
+        if "vector_stores_config" in fields and run_config.vector_stores is not None:
+            provider_config["vector_stores_config"] = run_config.vector_stores
+
+        # Inject metadata_store from server stores config when not explicitly configured.
+        if "metadata_store" in fields:
+            if provider_config.get("metadata_store") is None and run_config.storage.stores.vector_stores is not None:
+                provider_config["metadata_store"] = run_config.storage.stores.vector_stores.model_dump()
+
+        return provider_config
+
     args = []
     if isinstance(provider_spec, RemoteProviderSpec):
         config_type = instantiate_class_type(provider_spec.config_class)
-        config = config_type(**provider.config)
+        provider_config = _inject_config_defaults(config_type, provider.config.copy())
+        config = config_type(**provider_config)
 
         method = "get_adapter_impl"
         args = [config, deps]
@@ -423,15 +442,8 @@ async def instantiate_provider(
         args = [provider_spec.api, inner_impls, deps, dist_registry, policy]
     else:
         method = "get_provider_impl"
-        provider_config = provider.config.copy()
-
-        # Inject vector_stores_config for providers that need it (introspection-based)
         config_type = instantiate_class_type(provider_spec.config_class)
-        if hasattr(config_type, "__fields__") and "vector_stores_config" in config_type.__fields__:
-            # Only inject if vector_stores is provided, otherwise let default_factory handle it
-            if run_config.vector_stores is not None:
-                provider_config["vector_stores_config"] = run_config.vector_stores
-
+        provider_config = _inject_config_defaults(config_type, provider.config.copy())
         config = config_type(**provider_config)
         args = [config, deps]
         if "policy" in inspect.signature(getattr(module, method)).parameters:

src/ogx/providers/utils/memory/openai_vector_store_mixin.py

@@ -599,15 +599,18 @@ async def _delete_openai_vector_store_file_batch(self, batch_id: str) -> None:
     async def _cleanup_expired_file_batches(self) -> None:
         """Clean up expired file batches from persistent storage."""
         if self.metadata_store:
-            results = await self.metadata_store.fetch_all(table=TABLE_VECTOR_STORE_FILE_BATCHES)
+            rows = await self._fetch_all_metadata_rows_unfiltered(table=TABLE_VECTOR_STORE_FILE_BATCHES)
             current_time = int(time.time())
             expired_count = 0
-            for row in results.data:
+            for row in rows:
                 info = row["batch_data"]
                 expires_at = info.get("expires_at")
                 if expires_at and current_time > expires_at:
                     logger.info("Cleaning up expired file batch", id=info["id"])
-                    await self.metadata_store.delete(table=TABLE_VECTOR_STORE_FILE_BATCHES, where={"id": info["id"]})
+                    await self.metadata_store.sql_store.delete(
+                        table=TABLE_VECTOR_STORE_FILE_BATCHES,
+                        where={"id": info["id"]},
+                    )
                     self.openai_file_batches.pop(info["id"], None)
                     expired_count += 1
             if expired_count > 0:
@@ -723,6 +726,12 @@ async def initialize_openai_vector_stores(self) -> None:
                 "Files API is not available. File attachment operations on vector stores will fail. "
                 "Ensure a 'files' provider is configured if file operations are needed."
             )
+        policy = getattr(self, "_policy", [])
+        if policy and not self.metadata_store:
+            raise ValueError(
+                "Failed to initialize vector store provider: metadata_store is required when access control "
+                "policies are configured. Configure storage.stores.vector_stores in your server config."
+            )
         if self.metadata_store:
             await self._create_metadata_tables()
             if self.kvstore:

tests/unit/core/test_resolver_metadata_store_injection.py

@@ -0,0 +1,123 @@
+# Copyright (c) The OGX Contributors.
+# All rights reserved.
+#
+# This source code is licensed under the terms described in the LICENSE file in
+# the root directory of this source tree.
+
+from types import SimpleNamespace
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+from ogx.core.datatypes import StackConfig
+from ogx.core.resolver import instantiate_provider
+from ogx.core.storage.datatypes import ServerStoresConfig, SqlStoreReference, StorageConfig
+from ogx_api import Api, RemoteProviderSpec
+
+
+class DummyRemoteConfig(BaseModel):
+    metadata_store: SqlStoreReference | None = Field(default=None)
+
+
+def _make_run_config(vector_store_table: str = "vector_store_metadata") -> StackConfig:
+    return StackConfig(
+        distro_name="test",
+        providers={},
+        storage=StorageConfig(
+            stores=ServerStoresConfig(
+                vector_stores=SqlStoreReference(
+                    backend="sql_default",
+                    table_name=vector_store_table,
+                )
+            )
+        ),
+    )
+
+
+class TestResolverMetadataStoreInjection:
+    async def test_injects_metadata_store_for_remote_provider_when_missing(self, monkeypatch):
+        captured: dict[str, Any] = {}
+
+        async def get_adapter_impl(config, deps, policy=None):
+            captured["config"] = config
+            return SimpleNamespace()
+
+        monkeypatch.setattr("ogx.core.resolver.instantiate_class_type", lambda _: DummyRemoteConfig)
+        monkeypatch.setattr(
+            "ogx.core.resolver.importlib.import_module",
+            lambda _: SimpleNamespace(get_adapter_impl=get_adapter_impl),
+        )
+        monkeypatch.setattr("ogx.core.resolver.check_protocol_compliance", lambda *_args, **_kwargs: None)
+
+        provider = SimpleNamespace(
+            provider_id="dummy-remote",
+            provider_type="remote::dummy",
+            config={},
+            spec=RemoteProviderSpec(
+                api=Api.vector_io,
+                provider_type="remote::dummy",
+                config_class="dummy.Config",
+                module="dummy.remote.module",
+                adapter_type="dummy-adapter",
+            ),
+        )
+
+        await instantiate_provider(
+            provider=provider,
+            deps={},
+            inner_impls={},
+            dist_registry=SimpleNamespace(),
+            run_config=_make_run_config(),
+            policy=[],
+        )
+
+        config = captured["config"]
+        assert config.metadata_store is not None
+        assert config.metadata_store.backend == "sql_default"
+        assert config.metadata_store.table_name == "vector_store_metadata"
+
+    async def test_preserves_explicit_remote_metadata_store_config(self, monkeypatch):
+        captured: dict[str, Any] = {}
+
+        async def get_adapter_impl(config, deps, policy=None):
+            captured["config"] = config
+            return SimpleNamespace()
+
+        monkeypatch.setattr("ogx.core.resolver.instantiate_class_type", lambda _: DummyRemoteConfig)
+        monkeypatch.setattr(
+            "ogx.core.resolver.importlib.import_module",
+            lambda _: SimpleNamespace(get_adapter_impl=get_adapter_impl),
+        )
+        monkeypatch.setattr("ogx.core.resolver.check_protocol_compliance", lambda *_args, **_kwargs: None)
+
+        provider = SimpleNamespace(
+            provider_id="dummy-remote",
+            provider_type="remote::dummy",
+            config={
+                "metadata_store": {
+                    "backend": "sql_default",
+                    "table_name": "custom_vector_store_table",
+                }
+            },
+            spec=RemoteProviderSpec(
+                api=Api.vector_io,
+                provider_type="remote::dummy",
+                config_class="dummy.Config",
+                module="dummy.remote.module",
+                adapter_type="dummy-adapter",
+            ),
+        )
+
+        await instantiate_provider(
+            provider=provider,
+            deps={},
+            inner_impls={},
+            dist_registry=SimpleNamespace(),
+            run_config=_make_run_config(vector_store_table="default_vector_store_table"),
+            policy=[],
+        )
+
+        config = captured["config"]
+        assert config.metadata_store is not None
+        assert config.metadata_store.backend == "sql_default"
+        assert config.metadata_store.table_name == "custom_vector_store_table"

tests/unit/providers/utils/memory/test_openai_vector_store_mixin.py

@@ -327,3 +327,92 @@ async def test_migration_copies_batches(self):
             conflict_columns=["id"],
             update_columns=["store_id", "batch_data", "expires_at"],
         )
+
+
+class TestMetadataStoreEnforcement:
+    """Tests for mandatory metadata_store when access control policies are active."""
+
+    async def test_initialize_raises_when_policy_set_but_no_metadata_store(self):
+        mixin = MockVectorStoreMixin(
+            inference_api=AsyncMock(),
+            files_api=AsyncMock(),
+            kvstore=AsyncMock(),
+        )
+        mixin._policy = [MagicMock()]
+
+        with pytest.raises(ValueError, match="metadata_store is required"):
+            await mixin.initialize_openai_vector_stores()
+
+    async def test_initialize_succeeds_when_no_policy(self):
+        kvstore = AsyncMock()
+        kvstore.values_in_range = AsyncMock(return_value=[])
+
+        mixin = MockVectorStoreMixin(
+            inference_api=AsyncMock(),
+            files_api=AsyncMock(),
+            kvstore=kvstore,
+        )
+        mixin._policy = []
+
+        await mixin.initialize_openai_vector_stores()
+
+    async def test_initialize_succeeds_when_policy_and_metadata_store_set(self):
+        metadata_store = MagicMock()
+        metadata_store.create_table = AsyncMock()
+        metadata_store.sql_store = AsyncMock()
+        metadata_store.sql_store.fetch_all = AsyncMock(return_value=MagicMock(data=[]))
+        metadata_store.fetch_all = AsyncMock(return_value=MagicMock(data=[]))
+
+        kvstore = AsyncMock()
+        kvstore.get = AsyncMock(return_value="1")
+        kvstore.values_in_range = AsyncMock(return_value=[])
+
+        mixin = MockVectorStoreMixin(
+            inference_api=AsyncMock(),
+            files_api=AsyncMock(),
+            kvstore=kvstore,
+            metadata_store=metadata_store,
+        )
+        mixin._policy = [MagicMock()]
+
+        await mixin.initialize_openai_vector_stores()
+
+
+class TestFileBatchCleanup:
+    """Tests for metadata-store file batch cleanup behavior."""
+
+    async def test_cleanup_uses_unfiltered_sql_store_access(self):
+        sql_store = AsyncMock()
+        sql_store.fetch_all = AsyncMock(
+            return_value=MagicMock(
+                data=[
+                    {
+                        "id": "batch_1",
+                        "batch_data": {"id": "batch_1", "expires_at": 1},
+                    }
+                ]
+            )
+        )
+        sql_store.delete = AsyncMock()
+
+        metadata_store = MagicMock()
+        metadata_store.sql_store = sql_store
+        metadata_store.fetch_all = AsyncMock(side_effect=AssertionError("filtered fetch_all should not be used"))
+
+        mixin = MockVectorStoreMixin(
+            inference_api=AsyncMock(),
+            files_api=AsyncMock(),
+            kvstore=AsyncMock(),
+            metadata_store=metadata_store,
+        )
+        mixin.openai_file_batches = {"batch_1": {"id": "batch_1"}}
+
+        await mixin._cleanup_expired_file_batches()
+
+        metadata_store.fetch_all.assert_not_called()
+        sql_store.fetch_all.assert_called_once_with(table="vector_store_file_batches")
+        sql_store.delete.assert_called_once_with(
+            table="vector_store_file_batches",
+            where={"id": "batch_1"},
+        )
+        assert "batch_1" not in mixin.openai_file_batches