fix: cos share all

.env.example

@@ -163,6 +163,26 @@ MICROSOFT_GRAPH_OAUTH_CLIENT_SECRET=
 AWS_ACCESS_KEY_ID=
 AWS_SECRET_ACCESS_KEY=
 
+# --- IBM Cloud Object Storage connector (Enterprise/SaaS) ------------------
+# Enterprise/SaaS-only bucket connector, gated by IBM_AUTH_ENABLED (like the
+# AWS S3 / Azure Blob connectors). Credentials are normally entered per-connection
+# in the UI; these env vars are OPTIONAL fallbacks / defaults. Two auth modes are
+# supported: IAM (api_key + service_instance_id, default) or HMAC (access key +
+# secret, S3-compatible -- works against MinIO for local testing).
+IBM_COS_ENDPOINT=
+IBM_COS_API_KEY=
+IBM_COS_SERVICE_INSTANCE_ID=
+IBM_COS_HMAC_ACCESS_KEY_ID=
+IBM_COS_HMAC_SECRET_ACCESS_KEY=
+#
+# Local dev / MinIO testing: set OPENRAG_DEV_IBM_COS=true to enable the
+# connector without IBM_AUTH_ENABLED. NEVER use in production.
+# OPENRAG_DEV_IBM_COS=false
+# Frontend counterpart: also set NEXT_PUBLIC_DEV_IBM_COS=true so the Connectors
+# tab and admin Connectors Permission page show IBM COS locally (requires a
+# frontend restart to take effect -- Next.js reads NEXT_PUBLIC_* at startup).
+# NEXT_PUBLIC_DEV_IBM_COS=false
+
 # --- Azure Blob Storage connector (Enterprise/SaaS) ------------------------
 # Enterprise/SaaS-only bucket connector, gated by IBM_AUTH_ENABLED (like the
 # AWS S3 / IBM COS connectors). Credentials are normally entered per-connection

enhancements/connectors/ibm_cos/connector.py

@@ -1,12 +1,16 @@
-"""IBM Cloud Object Storage connector for OpenRAG."""
+"""IBM Cloud Object Storage connector for OpenRAG.
+
+Enterprise/SaaS-only ``bucket``-kind connector, gated by ``IBM_AUTH_ENABLED``
+(or ``OPENRAG_DEV_IBM_COS=true`` for local dev/MinIO testing).
+"""
 
 import mimetypes
 import os
 from datetime import UTC, datetime
 from posixpath import basename
 from typing import Any
 
-from config.settings import IBM_AUTH_ENABLED
+from config.settings import IBM_AUTH_ENABLED, is_dev_ibm_cos_enabled
 from connectors.base import BaseConnector, ConnectorDocument, DocumentACL
 from utils.logging_config import get_logger
 
@@ -64,7 +68,10 @@ class IBMCOSConnector(BaseConnector):
 
     @classmethod
     def is_available(cls, manager, user_id=None) -> bool:
-        return IBM_AUTH_ENABLED
+        # Enterprise/SaaS gate is IBM_AUTH_ENABLED, like the other bucket
+        # connectors (aws_s3, azure_blob). OPENRAG_DEV_IBM_COS=true bypasses it
+        # for local dev (e.g. against MinIO in HMAC mode; never in production).
+        return IBM_AUTH_ENABLED or is_dev_ibm_cos_enabled()
 
     @classmethod
     def register_routes(cls, app) -> None:

frontend/lib/brand.ts

@@ -4,6 +4,14 @@ export type Brand = "oss" | "ibm";
 
 export const IBM_THEME_DEV = process.env.NEXT_PUBLIC_IBM_THEME_DEV === "true";
 
+/**
+ * Local dev: show the IBM COS connector in the Connectors tab / permission
+ * admin without IBM_AUTH_ENABLED. Mirrors the backend `OPENRAG_DEV_IBM_COS`
+ * bypass (see `enhancements/connectors/ibm_cos/connector.py`). Never enable
+ * in production.
+ */
+export const DEV_IBM_COS = process.env.NEXT_PUBLIC_DEV_IBM_COS === "true";
+
 /** Default when no localStorage — must match `BrandProvider` initial state. */
 export const DEFAULT_BRAND: Brand = IBM_THEME_DEV ? "ibm" : "oss";
 
@@ -82,7 +90,8 @@ export function isConnectorTypeVisible(
     isIbmAuthMode,
   }: { isCloudBrand: boolean; isIbmAuthMode: boolean },
 ): boolean {
-  if (type === "ibm_cos" || type === "aws_s3") return isIbmAuthMode;
+  if (type === "ibm_cos") return isIbmAuthMode || DEV_IBM_COS;
+  if (type === "aws_s3") return isIbmAuthMode;
   if (cloudBrand && type === "onedrive") return false;
   return true;
 }

src/api/connectors.py

@@ -1124,6 +1124,7 @@ async def connector_sync(
                             new_ids,
                             jwt_token=jwt_token,
                             ingest_settings=body.settings,
+                            shared=body.shared,
                         )
                     )
                 if changed_ids:
@@ -1135,6 +1136,7 @@ async def connector_sync(
                             jwt_token=jwt_token,
                             ingest_settings=body.settings,
                             replace_duplicates=True,
+                            shared=body.shared,
                         )
                     )
             else:

src/config/settings.py

@@ -258,6 +258,17 @@ def is_dev_azure_blob_enabled() -> bool:
     return raw in ("true", "1", "yes", "on")
 
 
+def is_dev_ibm_cos_enabled() -> bool:
+    """Local dev: enable the IBM COS connector without IBM_AUTH_ENABLED.
+
+    Allows testing the IBM COS connector (e.g. against MinIO in HMAC mode) in a
+    local environment where IBM auth is not configured. Never enable in
+    production. Requires ``OPENRAG_DEV_IBM_COS=true``.
+    """
+    raw = os.getenv("OPENRAG_DEV_IBM_COS", "false").strip().lower()
+    return raw in ("true", "1", "yes", "on")
+
+
 def is_azure_blob_enabled() -> bool:
     """Feature kill switch for the Azure Blob connector (default: enabled).
 

src/models/processors.py

@@ -21,7 +21,7 @@
 )
 from utils.hash_utils import hash_id
 from utils.logging_config import get_logger
-from utils.opensearch_queries import build_filename_search_body
+from utils.opensearch_queries import build_filename_search_body, build_replace_filename_query
 
 from .tasks import FileTask, TaskStatus, UploadTask
 
@@ -816,6 +816,58 @@ def __init__(
         self.connector_type = connector_type
         self.shared = shared
 
+    async def _reconcile_shared_owner(self, filename: str) -> None:
+        """Update owner fields on already-indexed chunks for `filename` to match
+        the connector's current `shared` setting.
+
+        Called on the duplicate/unchanged skip paths below, where a file's
+        content and name haven't changed since a prior sync but the connector's
+        "Make documents available to all users" setting may have been toggled
+        since then. Without this, those chunks would keep whatever owner they
+        got on their original ingest forever, since a byte-identical re-sync
+        never reaches resolve_shared_owner_fields(). Scoped to chunks owned by
+        this user or already ownerless (matching the same boundary
+        delete_document_by_filename uses), so it can't touch another user's
+        private document that happens to share this filename.
+        """
+        write_client = clients.opensearch
+        if write_client is None:
+            return
+        owner, owner_name, owner_email = resolve_shared_owner_fields(
+            self.user_id, self.owner_name, self.owner_email, self.shared
+        )
+        for candidate in get_filename_aliases(filename):
+            try:
+                await write_client.update_by_query(
+                    index=get_index_name(),
+                    body={
+                        "query": build_replace_filename_query(candidate, self.user_id),
+                        "script": {
+                            "source": """
+                                if (params.shared) {
+                                    ctx._source.remove('owner');
+                                } else {
+                                    ctx._source.owner = params.owner;
+                                }
+                                ctx._source.owner_name = params.owner_name;
+                                ctx._source.owner_email = params.owner_email;
+                            """,
+                            "params": {
+                                "shared": self.shared,
+                                "owner": owner,
+                                "owner_name": owner_name,
+                                "owner_email": owner_email,
+                            },
+                        },
+                    },
+                )
+            except Exception as e:
+                logger.warning(
+                    "Failed to reconcile owner fields for skipped duplicate",
+                    filename=candidate,
+                    error=str(e),
+                )
+
     async def process_item(self, upload_task: UploadTask, item: str, file_task: FileTask) -> None:
         """Process a connector file using unified methods"""
         file_task.status = TaskStatus.RUNNING
@@ -964,6 +1016,7 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
 
             if await self.check_filename_exists(file_task.filename, opensearch_client):
                 if not self.replace_duplicates:
+                    await self._reconcile_shared_owner(file_task.filename)
                     file_task.status = TaskStatus.SKIPPED
                     file_task.error = None
                     file_task.result = {
@@ -994,6 +1047,7 @@ async def process_item(self, upload_task: UploadTask, item: str, file_task: File
                 file_hash = hash_id(tmp_path)
 
                 if not renamed and await self.check_document_exists(file_hash, opensearch_client):
+                    await self._reconcile_shared_owner(file_task.filename)
                     file_task.status = TaskStatus.COMPLETED
                     file_task.result = {"status": "unchanged", "id": file_hash}
                     file_task.updated_at = time.time()

tests/unit/connectors/test_ibm_cos_connector.py

@@ -0,0 +1,41 @@
+"""Unit tests for the IBM COS connector's `is_available` gating.
+
+Covers IBM_AUTH_ENABLED / OPENRAG_DEV_IBM_COS, mirroring the equivalent
+Azure Blob coverage (tests/unit/connectors/test_azure_blob_connector.py).
+"""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+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.ibm_cos import connector as ibm_cos_connector  # noqa: E402
+from enhancements.connectors.ibm_cos.connector import IBMCOSConnector  # noqa: E402
+
+
+def test_is_available_gated_on_ibm_auth_enabled(monkeypatch):
+    monkeypatch.setattr(ibm_cos_connector, "IBM_AUTH_ENABLED", True)
+    monkeypatch.setattr(ibm_cos_connector, "is_dev_ibm_cos_enabled", lambda: False)
+    assert IBMCOSConnector.is_available(MagicMock()) is True
+    monkeypatch.setattr(ibm_cos_connector, "IBM_AUTH_ENABLED", False)
+    assert IBMCOSConnector.is_available(MagicMock()) is False
+
+
+def test_is_available_dev_flag_bypasses_ibm_auth(monkeypatch):
+    monkeypatch.setattr(ibm_cos_connector, "IBM_AUTH_ENABLED", False)
+    monkeypatch.setattr(ibm_cos_connector, "is_dev_ibm_cos_enabled", lambda: True)
+    assert IBMCOSConnector.is_available(MagicMock()) is True
+
+
+def test_is_dev_ibm_cos_enabled_defaults_false_when_unset(monkeypatch):
+    """Regression: must default to disabled, or IBM COS becomes available in
+    every deployment (including production) whenever OPENRAG_DEV_IBM_COS is
+    simply unset, defeating the point of the dev-only bypass."""
+    from config.settings import is_dev_ibm_cos_enabled
+
+    monkeypatch.delenv("OPENRAG_DEV_IBM_COS", raising=False)
+    assert is_dev_ibm_cos_enabled() is False

tests/unit/test_connector_sync_bucket_filter_shared.py

@@ -0,0 +1,88 @@
+"""Regression test: connector_sync's bucket_filter branch must forward
+`shared` to both sync_specific_files batches (new + changed files).
+
+This is the code path the COS connector UI's "Ingest N Buckets" button
+always hits (it sends bucket_filter, never selected_files/sync_all), so a
+dropped `shared` here means the "Make documents available to all users"
+toggle silently has no effect on real syncs.
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from api.connectors import ConnectorSyncBody, connector_sync
+
+
+def _make_connection():
+    connection = MagicMock()
+    connection.connection_id = "conn-1"
+    connection.is_active = True
+    return connection
+
+
+def _make_connector():
+    connector = MagicMock()
+    connector.authenticate = AsyncMock(return_value=True)
+    connector.bucket_names = []
+    connector.list_files = AsyncMock(
+        return_value={
+            "files": [
+                {"id": "new-file-1", "modified_time": None},
+                {"id": "changed-file-1", "modified_time": None},
+            ],
+            "next_page_token": None,
+        }
+    )
+    return connector
+
+
+@pytest.mark.asyncio
+async def test_bucket_filter_sync_forwards_shared_to_new_and_changed_batches():
+    body = ConnectorSyncBody(bucket_filter=["bucket-a"], shared=True)
+
+    connector_service = MagicMock()
+    connection = _make_connection()
+    connector_service.connection_manager.list_connections = AsyncMock(return_value=[connection])
+    connector = _make_connector()
+    connector_service.get_connector = AsyncMock(return_value=connector)
+    connector_service.sync_specific_files = AsyncMock(side_effect=["task-new", "task-changed"])
+
+    session_manager = MagicMock()
+    user = MagicMock()
+    user.jwt_token = "token"
+    user.user_id = "user-1"
+
+    with (
+        patch(
+            "api.connectors.get_synced_file_ids_for_connector",
+            new=AsyncMock(return_value=(["changed-file-1"], [], "connector_file_id")),
+        ),
+        patch(
+            "api.connectors.get_synced_id_to_modified_time_map",
+            new=AsyncMock(return_value={"changed-file-1": 0.0}),
+        ),
+        patch(
+            "api.connectors.classify_remote_file_change",
+            side_effect=lambda fid, *_a, **_k: "new" if fid == "new-file-1" else "changed",
+        ),
+    ):
+        response = await connector_sync(
+            connector_type="ibm_cos",
+            body=body,
+            request=MagicMock(),
+            connector_service=connector_service,
+            session_manager=session_manager,
+            user=user,
+            session=AsyncMock(),
+            rbac=MagicMock(),
+        )
+
+    assert response.status_code == 201
+    assert connector_service.sync_specific_files.await_count == 2
+
+    new_call, changed_call = connector_service.sync_specific_files.await_args_list
+    assert new_call.args[2] == ["new-file-1"]
+    assert new_call.kwargs["shared"] is True
+    assert changed_call.args[2] == ["changed-file-1"]
+    assert changed_call.kwargs["shared"] is True