fix: japanese file names (backport of #2053)

src/connectors/service.py

@@ -135,7 +135,7 @@ async def process_connector_document(
                 docling_polling_service=self.task_service.docling_polling_service
                 if self.task_service
                 else None,
-                document_id=document.id,
+                connector_file_id=document.id,
                 source_url=document.source_url,
                 allowed_users=allowed_users,
                 allowed_groups=allowed_groups,
@@ -215,6 +215,7 @@ async def process_connector_document(
                     file_size=len(document.content) if document.content else 0,
                     connector_type=connector_type,
                     acl=document.acl,
+                    connector_file_id=document.id,
                     **standard_kwargs,
                 )
 
@@ -228,7 +229,10 @@ async def process_connector_document(
                 if result["status"] in ["indexed", "unchanged"]:
                     # Update all chunks with connector-specific metadata
                     await self._update_connector_metadata(
-                        document, owner_user_id, connector_type, jwt_token
+                        document,
+                        owner_user_id,
+                        connector_type,
+                        jwt_token,
                     )
 
                 return {
@@ -243,7 +247,6 @@ async def _update_connector_metadata(
         owner_user_id: str,
         connector_type: str,
         jwt_token: str = None,
-        id_field: str = "document_id",
         indexed_filename: str | None = None,
     ):
         """Update indexed chunks with connector-specific metadata"""
@@ -260,9 +263,11 @@ async def _update_connector_metadata(
             raise RuntimeError("Backend OpenSearch write client is unavailable")
 
         # Update ACL if changed (hash-based skip optimization).
-        # Match both document_id and connector_file_id: non-Langflow connector
-        # chunks store the connector id in connector_file_id (document_id holds the
-        # content hash), while Langflow chunks store it in document_id.
+        # Match both document_id and connector_file_id: both the Langflow and
+        # non-Langflow ingestion paths store the raw connector id in
+        # connector_file_id (document_id holds a content/id hash), except for
+        # pre-migration chunks indexed before that split existed, which only
+        # have document_id set to the raw connector id.
         acl_result = await update_document_acl(
             document_id=document.id,
             acl=document.acl,
@@ -288,14 +293,19 @@ async def _update_connector_metadata(
             await write_client.update_by_query(
                 index=self.index_name,
                 body={
-                    # Match both fields: non-Langflow chunks carry the connector id
-                    # in connector_file_id (document_id is the content hash),
-                    # Langflow chunks carry it in document_id.
+                    # Match both fields: both ingestion paths carry the raw
+                    # connector id in connector_file_id (document_id is a
+                    # content/id hash); pre-migration chunks only have it in
+                    # document_id.
                     "query": {
                         "bool": {
                             "should": [
                                 {"term": {"document_id": document.id}},
                                 {"term": {"connector_file_id": document.id}},
+                                # See check_document_exists (models/processors.py):
+                                # some indices predate the explicit keyword
+                                # mapping for this field.
+                                {"term": {"connector_file_id.keyword": document.id}},
                             ],
                             "minimum_should_match": 1,
                         }

src/models/processors.py

@@ -79,6 +79,7 @@ async def check_document_exists(
         on_error: Literal["assume_missing", "assume_exists"] = "assume_missing",
         *,
         wait_for_visibility: bool = False,
+        field: str = "document_id",
     ) -> bool:
         """
         Check if a document with the given hash already exists in OpenSearch.
@@ -102,14 +103,34 @@ async def check_document_exists(
         max_retries = 3
         retry_delay = 1.0
 
+        # Some deployments' indices predate connector_file_id's addition to the
+        # explicit mapping (config/settings.py), so OpenSearch dynamically
+        # mapped it as analyzed `text` (with a `.keyword` multi-field) instead
+        # of the intended `keyword` type. A plain term query against such a
+        # field tokenizes the query value and rarely matches the raw id, so
+        # also match its `.keyword` multi-field. document_id has always been
+        # explicitly `keyword` since index creation and never has this issue.
+        query: dict[str, Any]
+        if field == "connector_file_id":
+            query = {
+                "bool": {
+                    "should": [
+                        {"term": {field: file_hash}},
+                        {"term": {f"{field}.keyword": file_hash}},
+                    ]
+                }
+            }
+        else:
+            query = {"term": {field: file_hash}}
+
         for attempt in range(max_retries):
             try:
                 response = await opensearch_client.search(
                     index=get_index_name(),
                     body={
                         "size": 1,
                         "_source": False,
-                        "query": {"term": {"document_id": file_hash}},
+                        "query": query,
                     },
                 )
                 hits = response.get("hits", {}).get("hits", [])
@@ -342,6 +363,11 @@ async def _delete_connector_chunks(
                                 "should": [
                                     {"term": {"document_id": file_id}},
                                     {"term": {"connector_file_id": file_id}},
+                                    # Some deployments' indices predate this field's
+                                    # addition to the explicit mapping, so it was
+                                    # dynamically mapped as analyzed text with a
+                                    # `.keyword` multi-field instead of `keyword`.
+                                    {"term": {"connector_file_id.keyword": file_id}},
                                 ],
                                 "minimum_should_match": 1,
                             }
@@ -1091,10 +1117,25 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
                             delete_document_ids,
                         )
 
+                        # Match both fields: bucket-connector chunks carry the
+                        # raw connector id in connector_file_id (document_id is
+                        # a hash), while pre-migration chunks only have it in
+                        # document_id.
                         chunk_ids = await collect_visible_document_ids(
                             opensearch_client,
                             index=get_index_name(),
-                            query={"term": {"document_id": document.id}},
+                            query={
+                                "bool": {
+                                    "should": [
+                                        {"term": {"document_id": document.id}},
+                                        {"term": {"connector_file_id": document.id}},
+                                        # See check_document_exists: some indices
+                                        # predate the explicit keyword mapping for
+                                        # this field.
+                                        {"term": {"connector_file_id.keyword": document.id}},
+                                    ]
+                                }
+                            },
                         )
                         deleted_count = await delete_document_ids(
                             opensearch_client,
@@ -1173,7 +1214,7 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
                         if self.connector_service.task_service
                         else None,
                         file_task=file_task,
-                        document_id=document.id,
+                        connector_file_id=document.id,
                         source_url=document.source_url,
                         allowed_users=allowed_users,
                         allowed_groups=allowed_groups,
@@ -1193,6 +1234,7 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
                         _verification_client(opensearch_client),
                         on_error="assume_exists",
                         wait_for_visibility=True,
+                        field="connector_file_id",
                     ):
                         result = {
                             "status": "error",
@@ -1209,7 +1251,6 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
                             self.user_id,
                             connector_type,
                             self.jwt_token,
-                            id_field="document_id",
                             indexed_filename=file_task.filename,
                         )
                 else:
@@ -1257,7 +1298,6 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
                             self.user_id,
                             connector_type,
                             self.jwt_token,
-                            id_field="connector_file_id",
                             indexed_filename=file_task.filename,
                         )
 

src/services/document_index_writer.py

@@ -32,6 +32,7 @@ class DocumentIndexContext:
     file_size: int | None = None
     connector_type: str | None = None
     source_url: str | None = None
+    connector_file_id: str | None = None
     allowed_users: list[str] = field(default_factory=list)
     allowed_groups: list[str] = field(default_factory=list)
     allowed_principals: list[str] = field(default_factory=list)
@@ -244,7 +245,9 @@ def _build_chunk_document(
             doc["owner_email"] = context.owner_email
         if context.ingest_run_id:
             doc["ingest_run_id"] = context.ingest_run_id
-        if metadata.get("connector_file_id"):
+        if context.connector_file_id:
+            doc["connector_file_id"] = context.connector_file_id
+        elif metadata.get("connector_file_id"):
             doc["connector_file_id"] = metadata["connector_file_id"]
         if context.is_sample_data:
             doc["is_sample_data"] = "true"

src/services/langflow_file_service.py

@@ -168,7 +168,16 @@ def _resolve_document_id(
         self,
         file_tuples: list[tuple[str, Any, str]] | None,
         document_id: str | None,
+        connector_file_id: str | None = None,
     ) -> str:
+        # Bucket-style connectors (COS/Azure/S3) pass their raw, potentially
+        # non-ASCII and unbounded "bucket::key" as connector_file_id. Hash it
+        # into a stable ASCII document_id instead of using it verbatim — the
+        # raw value still travels downstream as connector_file_id, but
+        # document_id must stay safe for HTTP headers and OpenSearch's chunk
+        # _id (mirrors the content-hash document_id used by manual upload).
+        if connector_file_id:
+            return hash_id(io.BytesIO(connector_file_id.encode("utf-8")))
         if document_id:
             return document_id
         if file_tuples and len(file_tuples[0]) > 1:
@@ -199,6 +208,7 @@ def _configure_ingest_callback(
         parser: str | None = None,
         chunk_size: int | None = None,
         chunk_overlap: int | None = None,
+        connector_file_id: str | None = None,
     ) -> tuple[str | None, str | None]:
         if self.ingest_token_service is None:
             logger.warning(
@@ -234,6 +244,7 @@ def _configure_ingest_callback(
             parser=parser,
             chunk_size=chunk_size,
             chunk_overlap=chunk_overlap,
+            connector_file_id=connector_file_id,
         )
         token = self.ingest_token_service.create_token(context)
         logger.info(
@@ -353,6 +364,7 @@ async def run_ingestion_flow(
         owner_email: str | None = None,
         connector_type: str | None = None,
         document_id: str | None = None,
+        connector_file_id: str | None = None,
         source_url: str | None = None,
         allowed_users: list[str] | None = None,
         allowed_groups: list[str] | None = None,
@@ -419,7 +431,9 @@ async def run_ingestion_flow(
             if file_tuples and len(file_tuples) > 0 and len(file_tuples[0]) > 2
             else ""
         )
-        resolved_document_id = self._resolve_document_id(file_tuples, document_id)
+        resolved_document_id = self._resolve_document_id(
+            file_tuples, document_id, connector_file_id
+        )
 
         # Get the current embedding model and provider credentials from config
         from config.settings import get_openrag_config
@@ -480,6 +494,7 @@ async def run_ingestion_flow(
             parser="Docling Serve 1.20.0",
             chunk_size=chunk_size,
             chunk_overlap=chunk_overlap,
+            connector_file_id=connector_file_id,
         )
         headers.update(
             self._ingest_callback_global_var_headers(
@@ -892,6 +907,7 @@ async def upload_and_ingest_file(
         docling_polling_service: Any | None = None,
         file_task: Any | None = None,
         document_id: str | None = None,
+        connector_file_id: str | None = None,
         source_url: str | None = None,
         allowed_users: list[str] | None = None,
         allowed_groups: list[str] | None = None,
@@ -1036,6 +1052,7 @@ async def upload_and_ingest_file(
                 connector_type=connector_type,
                 docling_task_id=task_id,
                 document_id=document_id,
+                connector_file_id=connector_file_id,
                 source_url=source_url,
                 selected_embedding_model=selected_embedding,
                 allowed_users=allowed_users,

src/services/langflow_ingest_token_service.py

@@ -167,6 +167,7 @@ def _context_to_payload(context: DocumentIndexContext) -> dict[str, Any]:
             "parser": context.parser,
             "chunk_size": context.chunk_size,
             "chunk_overlap": context.chunk_overlap,
+            "connector_file_id": context.connector_file_id,
         }
 
     @staticmethod
@@ -198,4 +199,5 @@ def _payload_to_context(payload: dict[str, Any]) -> DocumentIndexContext:
             parser=payload.get("parser"),
             chunk_size=payload.get("chunk_size"),
             chunk_overlap=payload.get("chunk_overlap"),
+            connector_file_id=payload.get("connector_file_id"),
         )

src/utils/acl_utils.py

@@ -41,12 +41,22 @@ def _build_id_query(document_id: str, id_fields: tuple[str, ...]) -> dict:
     ``connector_file_id`` while ``document_id`` holds the content hash, whereas
     Langflow chunks and local uploads store the id in ``document_id``. Matching
     multiple fields lets a single id reliably target chunks from either pipeline.
+
+    Also matches ``connector_file_id.keyword``: some deployments' indices
+    predate that field's addition to the explicit mapping, so OpenSearch
+    dynamically mapped it as analyzed text (with a ``.keyword`` multi-field)
+    instead of the intended ``keyword`` type, and a plain term query against
+    it tokenizes the query value instead of matching the raw id. This clause
+    is a no-op on indices where the field is already ``keyword``.
     """
-    if len(id_fields) == 1:
-        return {"term": {id_fields[0]: document_id}}
+    clauses = [{"term": {field: document_id}} for field in id_fields]
+    if "connector_file_id" in id_fields:
+        clauses.append({"term": {"connector_file_id.keyword": document_id}})
+    if len(clauses) == 1:
+        return clauses[0]
     return {
         "bool": {
-            "should": [{"term": {field: document_id}} for field in id_fields],
+            "should": clauses,
             "minimum_should_match": 1,
         }
     }

tests/unit/connectors/test_bucket_connector_unicode_file_ids.py

@@ -0,0 +1,61 @@
+"""Regression coverage: bucket-style connectors (IBM COS, Azure Blob) must
+preserve non-ASCII (e.g. Japanese) object keys unchanged through their
+composite file-id round trip.
+
+`document.id` (built from `_make_file_id`) is what downstream ingestion now
+passes as `connector_file_id` — it must stay the raw, unmangled key so the
+bucket/key can still be split back out for delete/status lookups
+(`enhancements/connectors/*/api.py`), even though `document_id` itself is now
+derived from a hash of this value (see
+`tests/unit/test_resolve_document_id_connector_file_id.py`).
+"""
+
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent.parent.parent
+SRC = ROOT / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
+from enhancements.connectors.azure_blob.connector import (  # noqa: E402
+    _make_file_id as azure_make_file_id,
+)
+from enhancements.connectors.azure_blob.connector import (  # noqa: E402
+    _split_file_id as azure_split_file_id,
+)
+from enhancements.connectors.ibm_cos.connector import (  # noqa: E402
+    _make_file_id as cos_make_file_id,
+)
+from enhancements.connectors.ibm_cos.connector import (  # noqa: E402
+    _split_file_id as cos_split_file_id,
+)
+
+JAPANESE_KEY = "報告書.pdf"
+JAPANESE_NESTED_KEY = "フォルダ/報告書_最終版.pdf"
+
+
+def test_cos_file_id_round_trips_japanese_key():
+    file_id = cos_make_file_id("my-bucket", JAPANESE_KEY)
+
+    assert file_id == f"my-bucket::{JAPANESE_KEY}"
+    bucket, key = cos_split_file_id(file_id)
+    assert bucket == "my-bucket"
+    assert key == JAPANESE_KEY
+
+
+def test_cos_file_id_round_trips_nested_japanese_key():
+    file_id = cos_make_file_id("my-bucket", JAPANESE_NESTED_KEY)
+
+    bucket, key = cos_split_file_id(file_id)
+    assert bucket == "my-bucket"
+    assert key == JAPANESE_NESTED_KEY
+
+
+def test_azure_file_id_round_trips_japanese_blob_name():
+    file_id = azure_make_file_id("my-container", JAPANESE_KEY)
+
+    assert file_id == f"my-container::{JAPANESE_KEY}"
+    container, blob = azure_split_file_id(file_id)
+    assert container == "my-container"
+    assert blob == JAPANESE_KEY

tests/unit/connectors/test_update_connector_metadata_filename.py

@@ -92,6 +92,9 @@ async def _noop_acl(**_kwargs):
             "should": [
                 {"term": {"document_id": "graph-item-id-stable"}},
                 {"term": {"connector_file_id": "graph-item-id-stable"}},
+                # Some indices predate the explicit keyword mapping for this
+                # field and dynamically mapped it as analyzed text instead.
+                {"term": {"connector_file_id.keyword": "graph-item-id-stable"}},
             ],
             "minimum_should_match": 1,
         }