allowing scope retrieval via a list of file_ids

docs/content/docs/documentation/API.mdx

@@ -716,6 +716,7 @@ OpenAI-compatible text completion endpoint.
 | `spoken_style_answer` | `bool` | `false` | Generates a succinct spoken-style conversational answer based on the retrieved documents. |
 | `use_map_reduce` | `bool` | `false` | Uses a map-reduce strategy to aggregate information from multiple documents. See [map-reduce configuration](/openrag/documentation/env_vars/#map--reduce-configuration). |
 | `llm_override` | `object` | `null` | Overrides only the downstream LLM model name while still using OpenRAG's configured LLM endpoint and credentials. Accepts: `model` (string). Endpoint URL and API key are server-side configuration and cannot be changed by a client request. |
+| `attachments` | `list[{"id": string}]` | `null` | Scopes RAG retrieval to a specific list of file IDs within the target partition, instead of searching the whole partition. Unknown or unindexed IDs are silently dropped; duplicates are deduplicated. The response's `extra.attachments` reports which IDs were actually used. |
 
 Examples:
 
@@ -802,6 +803,30 @@ curl -X 'POST' 'http://localhost:8080/v1/chat/completions' \
 }'
 ```
 
+```bash title="Scoping retrieval to specific attachments"
+curl -X 'POST' 'http://localhost:8080/v1/chat/completions' \
+  -H 'accept: application/json' \
+  -H 'Authorization: Bearer YOUR_AUTH_TOKEN' \
+  -H 'Content-Type: application/json' \
+  -d '{
+  "model": "openrag-{partition_name}",
+  "messages": [
+    {
+      "role": "user",
+      "content": "summarize these files"
+    }
+  ],
+  "stream": false,
+  "metadata": {
+    "attachments": [
+      {"id": "file-001"},
+      {"id": "file-002"},
+      {"id": "file-003"}
+    ]
+  }
+}'
+```
+
 
 ### 🔧 Tools
 

openrag/core/ports/workspace_repo.py

@@ -53,6 +53,14 @@ async def get_existing_file_ids(self, partition: str, file_ids: list[str]) -> se
         """Return the subset of ``file_ids`` that actually exist in ``partition``."""
         ...
 
+    @abstractmethod
+    async def get_existing_file_ids_any_partition(self, file_ids: list[str]) -> set[str]:
+        """Return the subset of ``file_ids`` that exist in *any* partition.
+
+        Unscoped by design — only for the ``SUPER_ADMIN_MODE`` ``"all"`` wildcard.
+        """
+        ...
+
     @abstractmethod
     async def remove_file_from_all_workspaces(self, file_id: str, partition: str) -> None:
         """Detach ``file_id`` from every workspace in ``partition``."""

openrag/core/utils/source_filtering.py

@@ -115,6 +115,7 @@ async def stream_with_source_filtering(
     citation_protocol_active: bool = True,
     all_sources: list | None = None,
     include_all_retrieved: bool = False,
+    extra_fields: dict | None = None,
 ):
     """Process an LLM SSE stream and, when active, strip source tags.
 
@@ -292,6 +293,8 @@ async def stream_with_source_filtering(
     }
     if include_all_retrieved:
         extra_payload["all_retrieved_sources"] = all_sources if all_sources is not None else sources
+    if extra_fields:
+        extra_payload.update(extra_fields)
     if not saw_done:
         extra_payload["truncated"] = True
         logger.warning(

openrag/services/orchestrators/query_service.py

@@ -40,7 +40,7 @@
 from collections.abc import AsyncIterator, Callable
 from datetime import datetime
 from enum import Enum
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, NamedTuple
 
 from core.models.preset import resolve_partition_chat_llm
 from core.models.query import Query, SearchQueries
@@ -72,6 +72,21 @@
 
 PrepareSources = Callable[[list, list], list]
 
+
+class _PrepareChatResult(NamedTuple):
+    """A named tuple stays positionally unpackable, so existing
+    ``a, b, c, ... = await self._prepare_chat(...)`` call sites still work.
+    """
+
+    payload: dict
+    docs: list
+    web_results: list
+    retrieved_docs: list
+    retrieved_web_results: list
+    citation_protocol_active: bool
+    indexed_attachment_ids: list[str]
+
+
 _MAP_SYSTEM_PROMPT = """You are an AI assistant specialized in extracting and synthesizing relevant information from text.
 
 Your task:
@@ -445,10 +460,12 @@ async def _prepare_chat(self, partition: list[str] | None, payload: dict, llm: L
         spoken_style = metadata.get("spoken_style_answer", False)
         use_websearch = metadata.get("websearch", False)
         workspace = metadata.get("workspace")
+        attachment_ids = _extract_attachment_ids(metadata)
 
         top_k = self._mr_max if use_map_reduce else None
 
         filter_params = None
+        indexed_attachment_ids: list[str] = []
         if workspace and partition:
             scope = await self._workspace.resolve_scope(workspace, partition)
             if scope is None:
@@ -459,8 +476,15 @@ async def _prepare_chat(self, partition: list[str] | None, payload: dict, llm: L
             # partition the caller also has access to (#706).
             partition = [scope.partition]
             filter_params = {"file_id": scope.file_ids}
-
-        force_retrieval = use_websearch or use_map_reduce
+        elif attachment_ids and partition:
+            # No ownership check needed: file_id is ANDed with the server-fixed
+            # partition (or, for the "all" wildcard, SUPER_ADMIN_MODE-only).
+            indexed_attachment_ids = await self._existing_file_ids(attachment_ids, partition)
+            filter_params = {"file_id": indexed_attachment_ids}
+
+        # An attachment must never be dropped just because the classifier
+        # judged the turn conversational.
+        force_retrieval = use_websearch or use_map_reduce or bool(indexed_attachment_ids)
         if not queries.query_list:
             if not queries.requires_retrieval and not force_retrieval:
                 # Resolved per request from the library (named -> default ->
@@ -477,7 +501,7 @@ async def _prepare_chat(self, partition: list[str] | None, payload: dict, llm: L
                     context="",
                     current_date=datetime.now().strftime("%A, %B %d, %Y, %H:%M:%S"),
                 )
-                return payload, [], [], [], [], True
+                return _PrepareChatResult(payload, [], [], [], [], True, indexed_attachment_ids)
             queries = SearchQueries(query_list=[Query(query=messages[-1]["content"])])
 
         web_results: list = []
@@ -493,7 +517,7 @@ async def _prepare_chat(self, partition: list[str] | None, payload: dict, llm: L
             chunks = []
 
         if not chunks and not web_results and partition is None:
-            return payload, [], [], [], [], False
+            return _PrepareChatResult(payload, [], [], [], [], False, indexed_attachment_ids)
 
         docs = [c.to_langchain() for c in chunks]
 
@@ -555,7 +579,24 @@ async def _prepare_chat(self, partition: list[str] | None, payload: dict, llm: L
             },
         )
         payload["messages"] = new_messages
-        return payload, docs, web_results, retrieved_docs, retrieved_web_results, True
+        return _PrepareChatResult(
+            payload, docs, web_results, retrieved_docs, retrieved_web_results, True, indexed_attachment_ids
+        )
+
+    async def _existing_file_ids(self, file_ids: list[str], partitions: list[str]) -> list[str]:
+        """Order-preserving, deduplicated subset of ``file_ids`` indexed in ``partitions``.
+
+        ``"all"`` (``SUPER_ADMIN_MODE`` wildcard) takes an unscoped lookup instead
+        of a per-partition one.
+        """
+        if "all" in partitions:
+            if len(partitions) > 1:
+                raise ValueError("`partitions` cannot mix the wildcard with explicit values.")
+            found = set(await self._workspace.get_existing_file_ids_any_partition(file_ids))
+        else:
+            results = await asyncio.gather(*(self._workspace.get_existing_file_ids(p, file_ids) for p in partitions))
+            found = {fid for r in results for fid in r}
+        return [fid for fid in dict.fromkeys(file_ids) if fid in found]
 
     async def _gather_rag_and_web(self, queries, partition, top_k, filter_params):
         # Fuse the doc branch through retrieve_multi so a partition's rrf_k drives
@@ -674,15 +715,16 @@ async def chat(
         citation_protocol_active = False
         if partitions is None and not metadata.get("websearch", False):
             docs, web_results, retrieved_docs, retrieved_web_results = [], [], [], []
+            attachments: list[str] = []
         else:
-            (
-                payload,
-                docs,
-                web_results,
-                retrieved_docs,
-                retrieved_web_results,
-                citation_protocol_active,
-            ) = await self._prepare_chat(partitions, payload, llm)
+            result = await self._prepare_chat(partitions, payload, llm)
+            payload = result.payload
+            docs = result.docs
+            web_results = result.web_results
+            retrieved_docs = result.retrieved_docs
+            retrieved_web_results = result.retrieved_web_results
+            citation_protocol_active = result.citation_protocol_active
+            attachments = result.indexed_attachment_ids
         sources = prepare_sources(docs, web_results)
         # `all_retrieved_sources` is debug/eval telemetry, not needed by most
         # callers — skip building it (and calling prepare_sources on the full,
@@ -702,9 +744,11 @@ async def chat(
         else:
             clean, citations = content, None
         chunk["choices"][0]["message"]["content"] = clean
-        chunk["extra"] = json.dumps(
-            _build_extra_payload(sources, citations, all_sources, include_all_retrieved=include_all_retrieved)
-        )
+        extra = _build_extra_payload(sources, citations, all_sources, include_all_retrieved=include_all_retrieved)
+        if metadata.get("attachments"):
+            # Indicate which attachments were actually searched to generate the answer.
+            extra["attachments"] = attachments
+        chunk["extra"] = json.dumps(extra)
         return chunk
 
     async def chat_stream(
@@ -722,19 +766,22 @@ async def chat_stream(
         citation_protocol_active = False
         if partitions is None and not metadata.get("websearch", False):
             docs, web_results, retrieved_docs, retrieved_web_results = [], [], [], []
+            attachments: list[str] = []
         else:
-            (
-                payload,
-                docs,
-                web_results,
-                retrieved_docs,
-                retrieved_web_results,
-                citation_protocol_active,
-            ) = await self._prepare_chat(partitions, payload, llm)
+            result = await self._prepare_chat(partitions, payload, llm)
+            payload = result.payload
+            docs = result.docs
+            web_results = result.web_results
+            retrieved_docs = result.retrieved_docs
+            retrieved_web_results = result.retrieved_web_results
+            citation_protocol_active = result.citation_protocol_active
+            attachments = result.indexed_attachment_ids
         sources = prepare_sources(docs, web_results)
         all_sources = prepare_sources(retrieved_docs, retrieved_web_results) if include_all_retrieved else None
         structured_output = _is_structured_output(payload)
 
+        extra_fields = {"attachments": attachments} if metadata.get("attachments") else None
+
         payload["messages"] = self._sanitize_messages(payload["messages"])
         llm_stream = llm.stream_chat(payload["messages"], **_sampling(payload))
         async for sse_line in stream_with_source_filtering(
@@ -744,6 +791,7 @@ async def chat_stream(
             all_sources=all_sources,
             include_all_retrieved=include_all_retrieved,
             citation_protocol_active=citation_protocol_active and not structured_output,
+            extra_fields=extra_fields,
         ):
             yield sse_line
 
@@ -800,6 +848,14 @@ def _summary_doc(chunk, summary: str):
     return chunk.__class__(page_content=summary, metadata=chunk.metadata)
 
 
+def _extract_attachment_ids(metadata: dict) -> list[str]:
+    """file_ids from ``metadata.attachments = [{"id": ...}, ...]``; malformed payloads dropped."""
+    raw = metadata.get("attachments")
+    if not isinstance(raw, list):
+        return []
+    return [a["id"] for a in raw if isinstance(a, dict) and isinstance(a.get("id"), str) and a["id"]]
+
+
 def _dedupe_web(web_lists: list[list]) -> list:
     seen: set[str] = set()
     out: list = []

openrag/services/orchestrators/workspace_service.py

@@ -82,6 +82,9 @@ async def create_workspace(
     async def get_existing_file_ids(self, partition: str, file_ids: list[str]) -> list[str]:
         return list(await self._workspace_repo.get_existing_file_ids(partition, file_ids))
 
+    async def get_existing_file_ids_any_partition(self, file_ids: list[str]) -> list[str]:
+        return list(await self._workspace_repo.get_existing_file_ids_any_partition(file_ids))
+
     async def add_files(self, workspace_id: str, file_ids: list[str]) -> list[str]:
         """Associate files; returns any file_ids that were not found."""
         return await self._workspace_repo.add_files_to_workspace(workspace_id, file_ids)

openrag/services/persistence/workspace_repo.py

@@ -250,6 +250,22 @@ async def get_existing_file_ids(
         )
         return {r["file_id"] for r in rows}
 
+    async def get_existing_file_ids_any_partition(self, file_ids: list[str]) -> set[str]:
+        """Return the subset of ``file_ids`` that exist in *any* partition.
+
+        Unscoped by design — only for the ``SUPER_ADMIN_MODE`` ``"all"`` wildcard.
+        """
+        if not file_ids:
+            return set()
+        rows = await self.pool.fetch(
+            """
+            SELECT DISTINCT file_id FROM files
+            WHERE file_id = ANY($1::text[])
+            """,
+            file_ids,
+        )
+        return {r["file_id"] for r in rows}
+
     async def remove_file_from_all_workspaces(
         self,
         file_id: str,

tests/integration/repos/test_workspace_repo.py

@@ -99,6 +99,46 @@ async def test_get_file_workspaces_is_partition_scoped(
         assert in_a == ["ws-a"]
 
 
+class TestExistingFileIds:
+    """The two catalog-existence lookups behind attachment scoping.
+
+    Exercised against real SQL rather than a fake, so the ``ANY($1::text[])``
+    predicate and the presence/absence of the partition clause are actually
+    executed — the scoping assertions below both fail if either clause is
+    changed.
+    """
+
+    async def test_partition_scoped_ignores_other_partitions(self, postgres_store: PostgresStore):
+        await _seed_partition_and_files(postgres_store, partition="a", file_ids=("f1", "f2"))
+        await _seed_partition_and_files(postgres_store, partition="b", file_ids=("f3",))
+        found = await postgres_store.workspace_repo.get_existing_file_ids("a", ["f1", "f3", "ghost"])
+        # f3 exists, but in partition "b" — a partition-scoped lookup must not see it.
+        assert found == {"f1"}
+
+    async def test_any_partition_spans_partitions(self, postgres_store: PostgresStore):
+        await _seed_partition_and_files(postgres_store, partition="a", file_ids=("f1",))
+        await _seed_partition_and_files(postgres_store, partition="b", file_ids=("f3",))
+        found = await postgres_store.workspace_repo.get_existing_file_ids_any_partition(
+            ["f1", "f3", "ghost"],
+        )
+        assert found == {"f1", "f3"}
+
+    async def test_any_partition_reports_a_shared_file_id_once(self, postgres_store: PostgresStore):
+        # The same file_id in two partitions is two ``files`` rows. Callers treat
+        # the result as a membership set, so it must come back as a single entry.
+        # (Both the query's DISTINCT and the set-building enforce this; the test
+        # pins the contract, not either mechanism.)
+        await _seed_partition_and_files(postgres_store, partition="a", file_ids=("shared",))
+        await _seed_partition_and_files(postgres_store, partition="b", file_ids=("shared",))
+        found = await postgres_store.workspace_repo.get_existing_file_ids_any_partition(["shared"])
+        assert found == {"shared"}
+
+    async def test_empty_input_short_circuits(self, postgres_store: PostgresStore):
+        repo = postgres_store.workspace_repo
+        assert await repo.get_existing_file_ids("a", []) == set()
+        assert await repo.get_existing_file_ids_any_partition([]) == set()
+
+
 class TestDeleteWorkspace:
     async def test_returns_orphan_file_ids(self, postgres_store: PostgresStore):
         await _seed_partition_and_files(postgres_store)