Match connector_file_id in ACL/metadata updates (#1704)

Ensure ACL and metadata updates target both Langflow and non-Langflow chunks by matching document IDs against multiple fields. Introduce _build_id_query and id_fields parameters to should_update_acl, update_document_acl, and batch_update_acls so updates can match document_id and connector_file_id. Stop overwriting chunk owner during ACL syncs (only allowed_users/allowed_groups are updated) and set owner at ingest to the syncing user in TaskProcessor. Adjust tests to expect the combined bool query and make minor typing/import/formatting cleanups.

Author: edwinjosechittilappilly

src/connectors/service.py

@@ -191,11 +191,15 @@ async def _update_connector_metadata(
             owner_user_id, jwt_token
         )
 
-        # Update ACL if changed (hash-based skip optimization)
+        # 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.
         acl_result = await update_document_acl(
             document_id=document.id,
             acl=document.acl,
             opensearch_client=opensearch_client,
+            id_fields=("document_id", "connector_file_id"),
         )
 
         # Log ACL update result
@@ -214,7 +218,18 @@ async def _update_connector_metadata(
             await opensearch_client.update_by_query(
                 index=self.index_name,
                 body={
-                    "query": {"term": {"document_id": document.id}},
+                    # 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.
+                    "query": {
+                        "bool": {
+                            "should": [
+                                {"term": {"document_id": document.id}},
+                                {"term": {"connector_file_id": document.id}},
+                            ],
+                            "minimum_should_match": 1,
+                        }
+                    },
                     "script": {
                         "source": """
                             ctx._source.source_url = params.source_url;

src/models/processors.py

@@ -368,15 +368,19 @@ async def process_document_standard(
             if connector_file_id:
                 chunk_doc["connector_file_id"] = connector_file_id
 
-            # Set owner and ACL fields
+            # Set owner and ACL fields.
+            # owner is always the syncing/uploading user (matching the Langflow
+            # pipeline, which indexes owner from the OWNER global var = user id).
+            # acl.owner (e.g. the SharePoint file's author) is intentionally NOT
+            # used here — read access comes from allowed_users/allowed_groups + DLS,
+            # while owner gates deletion and "my documents" ownership.
+            chunk_doc["owner"] = owner_user_id
             if acl:
-                # Use ACL data if provided (from connector)
-                chunk_doc["owner"] = acl.owner if acl.owner else owner_user_id
+                # Use ACL access lists if provided (from connector)
                 chunk_doc["allowed_users"] = acl.allowed_users
                 chunk_doc["allowed_groups"] = acl.allowed_groups
             else:
-                # Fallback to owner_user_id if no ACL (local uploads)
-                chunk_doc["owner"] = owner_user_id
+                # No ACL provided
                 chunk_doc["allowed_users"] = []
                 chunk_doc["allowed_groups"] = []
 

src/utils/acl_utils.py

@@ -5,10 +5,10 @@
 to minimize write amplification when ACLs change.
 """
 
+import asyncio
 import hashlib
 import json
-import asyncio
-from typing import Dict, List, Tuple, Optional
+
 from src.connectors.base import DocumentACL
 from utils.logging_config import get_logger
 
@@ -30,15 +30,32 @@ def compute_acl_hash(acl: DocumentACL) -> str:
         "allowed_users": sorted(acl.allowed_users),
         "allowed_groups": sorted(acl.allowed_groups),
     }
-    return hashlib.sha256(
-        json.dumps(acl_data, sort_keys=True).encode()
-    ).hexdigest()
+    return hashlib.sha256(json.dumps(acl_data, sort_keys=True).encode()).hexdigest()
+
+
+def _build_id_query(document_id: str, id_fields: tuple[str, ...]) -> dict:
+    """Build an OpenSearch query matching ``document_id`` against one or more fields.
+
+    Non-Langflow connector chunks store the connector source id in
+    ``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.
+    """
+    if len(id_fields) == 1:
+        return {"term": {id_fields[0]: document_id}}
+    return {
+        "bool": {
+            "should": [{"term": {field: document_id}} for field in id_fields],
+            "minimum_should_match": 1,
+        }
+    }
 
 
 async def should_update_acl(
     document_id: str,
     new_acl: DocumentACL,
-    opensearch_client
+    opensearch_client,
+    id_fields: tuple[str, ...] = ("document_id",),
 ) -> bool:
     """
     Check if ACL has changed by querying one chunk and comparing hashes.
@@ -59,14 +76,14 @@ async def should_update_acl(
         response = await opensearch_client.search(
             index="documents",
             body={
-                "query": {"term": {"document_id": document_id}},
+                "query": _build_id_query(document_id, id_fields),
                 "size": 1,
                 "_source": [
                     "owner",
                     "allowed_users",
                     "allowed_groups",
-                ]
-            }
+                ],
+            },
         )
 
         if not response["hits"]["hits"]:
@@ -96,24 +113,34 @@ async def should_update_acl(
 async def update_document_acl(
     document_id: str,
     acl: DocumentACL,
-    opensearch_client
-) -> Dict[str, any]:
+    opensearch_client,
+    id_fields: tuple[str, ...] = ("document_id",),
+) -> dict[str, any]:
     """
     Update ACL for all chunks of a document.
 
     Uses hash-based skip optimization: queries one chunk to check if ACL changed,
     only updates if changed. When updating, uses bulk update_by_query for efficiency.
 
+    Only the access lists (``allowed_users``/``allowed_groups``) are updated — the
+    ``owner`` field is intentionally left untouched so an ACL re-sync never
+    reassigns document ownership (owner is set once at ingest to the syncing user).
+
     Args:
         document_id: Document identifier
         acl: New ACL to apply
         opensearch_client: OpenSearch client instance
+        id_fields: Chunk fields to match ``document_id`` against. Defaults to
+            ``("document_id",)``; pass ``("document_id", "connector_file_id")`` for
+            connector chunks where the connector id lives in ``connector_file_id``.
 
     Returns:
         Dict with status ("unchanged" or "updated") and chunks_updated count
     """
     # Check if ACL changed (queries one chunk)
-    should_update = await should_update_acl(document_id, acl, opensearch_client)
+    should_update = await should_update_acl(
+        document_id, acl, opensearch_client, id_fields=id_fields
+    )
 
     if not should_update:
         return {"status": "unchanged", "chunks_updated": 0}
@@ -123,36 +150,32 @@ async def update_document_acl(
         response = await opensearch_client.update_by_query(
             index="documents",
             body={
-                "query": {"term": {"document_id": document_id}},
+                "query": _build_id_query(document_id, id_fields),
                 "script": {
                     "source": """
-                        ctx._source.owner = params.owner;
                         ctx._source.allowed_users = params.allowed_users;
                         ctx._source.allowed_groups = params.allowed_groups;
                     """,
                     "params": {
-                        "owner": acl.owner,
                         "allowed_users": acl.allowed_users,
                         "allowed_groups": acl.allowed_groups,
-                    }
-                }
-            }
+                    },
+                },
+            },
         )
 
-        return {
-            "status": "updated",
-            "chunks_updated": response.get("updated", 0)
-        }
+        return {"status": "updated", "chunks_updated": response.get("updated", 0)}
 
     except Exception as e:
         logger.error("[OPENSEARCH] ACL update failed", document_id=document_id, error=str(e))
         return {"status": "error", "chunks_updated": 0, "error": str(e)}
 
 
 async def batch_update_acls(
-    acl_updates: List[Tuple[str, DocumentACL]],
-    opensearch_client
-) -> Dict[str, any]:
+    acl_updates: list[tuple[str, DocumentACL]],
+    opensearch_client,
+    id_fields: tuple[str, ...] = ("document_id",),
+) -> dict[str, any]:
     """
     Batch update ACLs for multiple documents.
 
@@ -161,9 +184,14 @@ async def batch_update_acls(
     - Skip unchanged ACLs (95%+ of webhook notifications)
     - Parallel bulk updates for changed ACLs
 
+    Only ``allowed_users``/``allowed_groups`` are updated; ``owner`` is left
+    untouched (see :func:`update_document_acl`).
+
     Args:
         acl_updates: List of (document_id, acl) tuples
         opensearch_client: OpenSearch client instance
+        id_fields: Chunk fields to match the id against (see
+            :func:`update_document_acl`).
 
     Returns:
         Dict with status, documents_updated count, and chunks_updated count
@@ -173,15 +201,15 @@ async def batch_update_acls(
 
     # Filter to only changed ACLs (parallel chunk queries)
     check_tasks = [
-        should_update_acl(doc_id, acl, opensearch_client)
+        should_update_acl(doc_id, acl, opensearch_client, id_fields=id_fields)
         for doc_id, acl in acl_updates
     ]
     should_update_flags = await asyncio.gather(*check_tasks)
 
     # Filter to documents with changed ACLs
     changed = [
         (doc_id, acl)
-        for (doc_id, acl), should_update in zip(acl_updates, should_update_flags)
+        for (doc_id, acl), should_update in zip(acl_updates, should_update_flags, strict=True)
         if should_update
     ]
 
@@ -190,28 +218,26 @@ async def batch_update_acls(
             "status": "no_changes",
             "documents_updated": 0,
             "chunks_updated": 0,
-            "skipped": len(acl_updates)
+            "skipped": len(acl_updates),
         }
 
     # Bulk update chunks for each document (parallelized)
     update_tasks = [
         opensearch_client.update_by_query(
             index="documents",
             body={
-                "query": {"term": {"document_id": doc_id}},
+                "query": _build_id_query(doc_id, id_fields),
                 "script": {
                     "source": """
-                        ctx._source.owner = params.owner;
                         ctx._source.allowed_users = params.allowed_users;
                         ctx._source.allowed_groups = params.allowed_groups;
                     """,
                     "params": {
-                        "owner": acl.owner,
                         "allowed_users": acl.allowed_users,
                         "allowed_groups": acl.allowed_groups,
-                    }
-                }
-            }
+                    },
+                },
+            },
         )
         for doc_id, acl in changed
     ]
@@ -222,7 +248,7 @@ async def batch_update_acls(
         # Count successful updates
         total_chunks_updated = 0
         errors = []
-        for i, result in enumerate(results):
+        for result in results:
             if isinstance(result, Exception):
                 errors.append(str(result))
             else:
@@ -233,13 +259,8 @@ async def batch_update_acls(
             "documents_updated": len(changed) - len(errors),
             "chunks_updated": total_chunks_updated,
             "skipped": len(acl_updates) - len(changed),
-            "errors": errors if errors else None
+            "errors": errors if errors else None,
         }
 
     except Exception as e:
-        return {
-            "status": "error",
-            "documents_updated": 0,
-            "chunks_updated": 0,
-            "error": str(e)
-        }
+        return {"status": "error", "documents_updated": 0, "chunks_updated": 0, "error": str(e)}

tests/unit/connectors/test_update_connector_metadata_filename.py

@@ -79,8 +79,19 @@ async def _noop_acl(**_kwargs):
         "the new name written through to indexed chunks"
     )
     assert params["filename"] == "Report-FY26.docx"
-    # Sanity: scope is by document_id (the stable connector ID).
-    assert body["query"] == {"term": {"document_id": "graph-item-id-stable"}}
+    # Scope matches the stable connector ID against BOTH document_id (Langflow
+    # chunks) and connector_file_id (non-Langflow chunks, whose document_id is
+    # the content hash). Without the connector_file_id clause, non-Langflow
+    # chunks never get source_url/filename/timestamps re-applied.
+    assert body["query"] == {
+        "bool": {
+            "should": [
+                {"term": {"document_id": "graph-item-id-stable"}},
+                {"term": {"connector_file_id": "graph-item-id-stable"}},
+            ],
+            "minimum_should_match": 1,
+        }
+    }
 
 
 @pytest.mark.asyncio