Fix DLS ingestion mapping client

Author: phact

src/models/processors.py

@@ -312,9 +312,13 @@ async def process_document_standard(
 
         dimensions = len(embeddings[0])
 
+        # Mapping updates are index-admin operations and fail under document-level
+        # security when attempted with a user-scoped OpenSearch client.
+        mapping_client = clients.opensearch or opensearch_client
+
         # Ensure the embedding field exists for this model
         embedding_field_name = await ensure_embedding_field_exists(
-            opensearch_client, embedding_model, get_index_name(), dimensions
+            mapping_client, embedding_model, get_index_name(), dimensions
         )
 
         # Index each chunk as a separate document

tests/integration/core/test_api_endpoints.py

@@ -165,8 +165,11 @@ async def test_upload_and_search_endpoint(tmp_path: Path, disable_langflow_inges
         "services.startup_orchestrator",
         "app.routes.internal",
         "app.routes",
+        "app.container",
         "app.factory",
         "app.lifespan",
+        "dependencies",
+        "utils.opensearch_init",
     ]:
         sys.modules.pop(mod, None)
     from config.settings import clients, get_index_name
@@ -396,8 +399,11 @@ async def test_langflow_chat_and_nudges_endpoints():
         "services.startup_orchestrator",
         "app.routes.internal",
         "app.routes",
+        "app.container",
         "app.factory",
         "services.chat_service",
+        "dependencies",
+        "utils.opensearch_init",
     ]:
         sys.modules.pop(mod, None)
 
@@ -503,8 +509,11 @@ async def test_search_multi_embedding_models(tmp_path: Path):
         "services.startup_orchestrator",
         "app.routes.internal",
         "app.routes",
+        "app.container",
         "app.factory",
         "app.lifespan",
+        "dependencies",
+        "utils.opensearch_init",
     ]:
         sys.modules.pop(mod, None)
 
@@ -647,8 +656,11 @@ async def test_router_upload_ingest_traditional(tmp_path: Path, disable_langflow
         "services.startup_orchestrator",
         "app.routes.internal",
         "app.routes",
+        "app.container",
         "app.factory",
         "app.lifespan",
+        "dependencies",
+        "utils.opensearch_init",
     ]:
         sys.modules.pop(mod, None)
     from config.settings import clients, get_index_name

tests/integration/core/test_onboarding_sample_docs.py

@@ -37,6 +37,7 @@
     "services.default_docs_service",
     "services.search_service",
     "services.startup_orchestrator",
+    "utils.opensearch_init",
 ]
 
 

tests/integration/core/test_startup_ingest.py

@@ -53,9 +53,21 @@ async def test_startup_ingest_creates_task(disable_langflow_ingest: bool):
     for mod in [
         "api.router",
         "api.connector_router",
+        "api",
+        "app.container",
+        "app.factory",
+        "app.lifespan",
+        "app.routes",
+        "app.routes.internal",
         "config.settings",
+        "dependencies",
         "auth_middleware",
         "main",
+        "services",
+        "services.default_docs_service",
+        "services.search_service",
+        "services.startup_orchestrator",
+        "utils.opensearch_init",
     ]:
         sys.modules.pop(mod, None)
 

tests/unit/test_processor_mapping_client.py

@@ -0,0 +1,97 @@
+from types import SimpleNamespace
+
+import pytest
+
+from models.processors import TaskProcessor
+
+
+@pytest.mark.asyncio
+async def test_standard_processor_uses_admin_client_for_embedding_mapping(
+    tmp_path,
+    monkeypatch,
+):
+    user_client = SimpleNamespace(
+        exists_calls=[],
+        index_calls=[],
+    )
+    admin_client = object()
+    mapping_clients = []
+
+    async def exists(*, index, id):
+        user_client.exists_calls.append({"index": index, "id": id})
+        return False
+
+    async def index(**kwargs):
+        user_client.index_calls.append(kwargs)
+
+    user_client.exists = exists
+    user_client.index = index
+
+    class SessionManager:
+        def get_user_opensearch_client(self, user_id, jwt_token):
+            assert user_id == "user-1"
+            assert jwt_token == "Bearer user-token"
+            return user_client
+
+    class ModelsService:
+        async def get_litellm_model_name(self, embedding_model):
+            return embedding_model
+
+    class EmbeddingClient:
+        class Embeddings:
+            async def create(self, model, input):
+                return SimpleNamespace(
+                    data=[
+                        SimpleNamespace(embedding=[0.1, 0.2, 0.3])
+                        for _ in input
+                    ]
+                )
+
+        embeddings = Embeddings()
+
+    async def ensure_embedding_field_exists(client, model_name, index_name, dimensions):
+        mapping_clients.append(client)
+        assert model_name == "text-embedding-3-small"
+        assert index_name == "documents"
+        assert dimensions == 3
+        return "chunk_embedding_text_embedding_3_small"
+
+    monkeypatch.setattr(
+        "config.settings.clients",
+        SimpleNamespace(
+            opensearch=admin_client,
+            patched_embedding_client=EmbeddingClient(),
+        ),
+    )
+    monkeypatch.setattr("config.settings.get_index_name", lambda: "documents")
+    monkeypatch.setattr(
+        "config.settings.get_openrag_config",
+        lambda: SimpleNamespace(knowledge=SimpleNamespace(embedding_model="")),
+    )
+    monkeypatch.setattr(
+        "utils.embedding_fields.ensure_embedding_field_exists",
+        ensure_embedding_field_exists,
+    )
+
+    file_path = tmp_path / "doc.md"
+    file_path.write_text("# Test\n\nhello world", encoding="utf-8")
+    document_service = SimpleNamespace(session_manager=SessionManager())
+    processor = TaskProcessor(
+        document_service=document_service,
+        models_service=ModelsService(),
+        docling_service=None,
+    )
+
+    result = await processor.process_document_standard(
+        file_path=str(file_path),
+        file_hash="file-1",
+        owner_user_id="user-1",
+        original_filename="doc.md",
+        jwt_token="Bearer user-token",
+        embedding_model="text-embedding-3-small",
+    )
+
+    assert result == {"status": "indexed", "id": "file-1"}
+    assert mapping_clients == [admin_client]
+    assert user_client.exists_calls == [{"index": "documents", "id": "file-1"}]
+    assert user_client.index_calls