diff --git a/README.md b/README.md
index 2a7a01c4..9a65e2b7 100644
--- a/README.md
+++ b/README.md
@@ -179,6 +179,8 @@ for file_path in sorted(p for p in documents_root.rglob("*") if p.is_file()):
 
 Already keeping your corpus in a bucket? `discover_documents()` and `extract_text()` name each document by
 its full S3 object key, so that is what `correct_answer_document_keys` must reference — prefix included.
+Discovery can span several locations in one bucket (`prefixes=["manuals/", "reports/"]`) and fails with a
+`BenchmarkKeyError` if a benchmark key matches none of the documents it found.
 
 
 ### Prepare `benchmark_data.json`
diff --git a/ai4rag/assets_generator/notebook_templates/maas_indexing_template.ipynb b/ai4rag/assets_generator/notebook_templates/maas_indexing_template.ipynb
index 7c682a9d..b9e6fc55 100644
--- a/ai4rag/assets_generator/notebook_templates/maas_indexing_template.ipynb
+++ b/ai4rag/assets_generator/notebook_templates/maas_indexing_template.ipynb
@@ -206,7 +206,7 @@
   {
    "cell_type": "markdown",
    "metadata": {},
-   "source": "The data processing pipeline prepares documents for the RAG system in multiple steps. Each step produces outputs stored under `step_outputs/`. \n\n| Step | Function | Purpose |\n|------|----------|---------| \n| 1 | **`discover_documents`** | List documents in the bucket, prioritize benchmark-referenced docs, apply a size cap, and write a JSON manifest (no content download). |\n| 2 | **`extract_text`** | Download the listed documents from S3 and extract text to DoclingDocument JSON files using Docling. |"
+   "source": "The data processing pipeline prepares documents for the RAG system in multiple steps. Each step produces outputs stored under `step_outputs/`. \n\n| Step | Function | Purpose |\n|------|----------|---------| \n| 1 | **`discover_documents`** | List documents under every selected prefix, prioritize benchmark-referenced docs, apply a size cap, and write a JSON manifest (no content download). |\n| 2 | **`extract_text`** | Download the listed documents from S3 and extract text to DoclingDocument JSON files using Docling. |"
   },
   {
    "cell_type": "code",
@@ -219,7 +219,7 @@
     "\n",
     "step_output_dir = Path(\"./step_outputs\")\n",
     "input_data_bucket_name = os.environ[\"AWS_S3_BUCKET\"]\n",
-    "input_data_key = \"{INPUT_DATA_KEY}\"\n",
+    "input_data_keys = {INPUT_DATA_KEYS}\n",
     "step_output_dir.mkdir(parents=True, exist_ok=True)"
    ]
   },
@@ -229,7 +229,7 @@
    "source": [
     "### Documents discovery\n",
     "\n",
-    "Lists objects in the S3 input bucket, filters by supported extensions (e.g., `.pdf`, `.docx`, `.pptx`, `.md`, `.html`, `.txt`), and builds a document set. Documents referenced in the benchmark are prioritized, then others are added until a configurable size limit (1 GB by default) is reached. This step does not download document contents but writes a JSON manifest (`documents_descriptor.json`) containing the bucket, prefix, and list of selected object keys and sizes for the next step."
+    "Lists objects in the S3 input bucket, filters by supported extensions (e.g., `.pdf`, `.docx`, `.pptx`, `.md`, `.html`, `.txt`), and builds a document set. Documents referenced in the benchmark are prioritized, then others are added until a configurable size limit (1 GB by default) is reached. This step does not download document contents but writes a JSON manifest (`documents_descriptor.json`) containing the bucket, prefixes, and list of selected object keys and sizes for the next step."
    ]
   },
   {
@@ -237,7 +237,7 @@
    "execution_count": null,
    "metadata": {},
    "outputs": [],
-   "source": "result = discover_documents(\n    bucket_name=input_data_bucket_name,\n    prefix=input_data_key,\n    s3_client=s3_client,\n)\nresult.save(step_output_dir / \"discovered_documents\")\n\nprint(json.dumps(result.to_dict(), indent=4, ensure_ascii=False))"
+   "source": "result = discover_documents(\n    bucket_name=input_data_bucket_name,\n    prefixes=input_data_keys,\n    s3_client=s3_client,\n)\nresult.save(step_output_dir / \"discovered_documents\")\n\nprint(json.dumps(result.to_dict(), indent=4, ensure_ascii=False))"
   },
   {
    "cell_type": "markdown",
diff --git a/ai4rag/assets_generator/templates.py b/ai4rag/assets_generator/templates.py
index 5074a674..56908ae2 100644
--- a/ai4rag/assets_generator/templates.py
+++ b/ai4rag/assets_generator/templates.py
@@ -2,6 +2,7 @@
 # Copyright IBM Corp. 2026
 # SPDX-License-Identifier: Apache-2.0
 # -----------------------------------------------------------------------------
+import json
 from pathlib import Path
 from typing import Any
 
@@ -38,7 +39,7 @@ def _format_required_env_vars(provider: str) -> str:
 def create_placeholder_mapping(
     output_data: dict[str, Any],
     test_data_key: str = "",
-    input_data_key: str = "",
+    input_data_keys: list[str] | None = None,
 ) -> dict[str, Any]:
     """Create a mapping from placeholder names to their values from a pattern definition.
 
@@ -52,8 +53,9 @@ def create_placeholder_mapping(
         The parsed ``pattern.json`` data.
     test_data_key : str, default=""
         S3 key of the test data file used as input to AI4RAG.
-    input_data_key : str, default=""
-        S3 key of the documents directory used as input to AI4RAG.
+    input_data_keys : list[str] | None, default=None
+        S3 key prefixes of the document locations used as input to AI4RAG.
+        Rendered into the notebook as a Python list literal.
 
     Returns
     -------
@@ -97,7 +99,9 @@ def create_placeholder_mapping(
     mapping["CHUNK_OVERLAP"] = ch.get("chunk_overlap", 50)
 
     mapping["TEST_DATA_KEY"] = test_data_key
-    mapping["INPUT_DATA_KEY"] = input_data_key
+    # Rendered by ``str.format`` into a bare expression, so it has to carry its
+    # own quoting and brackets to become a valid Python list literal.
+    mapping["INPUT_DATA_KEYS"] = json.dumps(input_data_keys or [])
 
     return mapping
 
@@ -107,7 +111,7 @@ def generate_notebook_from_template(
     output_data: dict[str, Any],
     output_notebook_path: str | Path,
     test_data_key: str = "",
-    input_data_key: str = "",
+    input_data_keys: list[str] | None = None,
 ) -> None:
     """Generate a filled notebook from a template and pattern configuration.
 
@@ -125,13 +129,13 @@ def generate_notebook_from_template(
         Path where the generated notebook is saved.
     test_data_key : str, default=""
         S3 key of the test data file used as input to AI4RAG.
-    input_data_key : str, default=""
-        S3 key of the documents directory used as input to AI4RAG.
+    input_data_keys : list[str] | None, default=None
+        S3 key prefixes of the document locations used as input to AI4RAG.
     """
     placeholder_mapping = create_placeholder_mapping(
         output_data,
         test_data_key=test_data_key,
-        input_data_key=input_data_key,
+        input_data_keys=input_data_keys,
     )
     notebook = Notebook.load(
         notebook_name=f"{notebook_template}_template.ipynb",
diff --git a/ai4rag/utils/data/__init__.py b/ai4rag/utils/data/__init__.py
index 81f4ad2e..49138e6a 100644
--- a/ai4rag/utils/data/__init__.py
+++ b/ai4rag/utils/data/__init__.py
@@ -2,12 +2,18 @@
 # Copyright IBM Corp. 2025-2026
 # SPDX-License-Identifier: Apache-2.0
 # -----------------------------------------------------------------------------
-from ai4rag.utils.data.documents_discovery import DiscoveryResult, DocumentDescriptor, discover_documents
+from ai4rag.utils.data.documents_discovery import (
+    BenchmarkKeyError,
+    DiscoveryResult,
+    DocumentDescriptor,
+    discover_documents,
+)
 from ai4rag.utils.data.test_data_loader import TestDataLoaderError, TestDataResult, load_test_data
 from ai4rag.utils.data.text_extraction import DoclingExtractionConfig, ExtractionResult, extract_text
 
 __all__ = [
     "discover_documents",
+    "BenchmarkKeyError",
     "DiscoveryResult",
     "DocumentDescriptor",
     "extract_text",
diff --git a/ai4rag/utils/data/documents_discovery.py b/ai4rag/utils/data/documents_discovery.py
index 024e12d5..a06d3d19 100644
--- a/ai4rag/utils/data/documents_discovery.py
+++ b/ai4rag/utils/data/documents_discovery.py
@@ -21,6 +21,10 @@
 SAMPLING_MAX_SIZE_GB: float = 1
 
 
+class BenchmarkKeyError(ValueError):
+    """Benchmark data references documents that are not part of the discovered corpus."""
+
+
 @dataclass(frozen=True)
 class DocumentDescriptor:
     """Metadata for a single document discovered in an S3 bucket.
@@ -47,10 +51,12 @@ class DiscoveryResult:
     ----------
     bucket : str
         S3 bucket name.
-    prefix : str
-        S3 key prefix used during listing.
+    prefixes : list[str]
+        S3 key prefixes used during listing.  A single empty string means the
+        whole bucket was listed.
     documents : list[DocumentDescriptor]
-        Discovered (and optionally sampled) documents.
+        Discovered (and optionally sampled) documents, deduplicated by object
+        key across all prefixes.
     total_size_bytes : int
         Combined size of all discovered documents.
     count : int
@@ -58,7 +64,7 @@ class DiscoveryResult:
     """
 
     bucket: str
-    prefix: str
+    prefixes: list[str]
     documents: list[DocumentDescriptor]
     total_size_bytes: int
     count: int
@@ -67,7 +73,7 @@ def to_dict(self) -> dict:
         """Serialise the result to a JSON-compatible dictionary."""
         return {
             "bucket": self.bucket,
-            "prefix": self.prefix,
+            "prefixes": self.prefixes,
             "documents": [{"key": d.key, "size_bytes": d.size_bytes} for d in self.documents],
             "total_size_bytes": self.total_size_bytes,
             "count": self.count,
@@ -95,17 +101,20 @@ def save(self, path: str | Path, filename: str = DOCUMENTS_DESCRIPTOR_FILENAME)
 # pylint: disable=too-many-locals
 def discover_documents(
     bucket_name: str,
-    prefix: str = "",
+    prefixes: str | list[str] | None = None,
     test_data_doc_names: list[str] | None = None,
     sampling_enabled: bool = True,
     sampling_max_size_gb: float = SAMPLING_MAX_SIZE_GB,
     supported_extensions: set[str] | None = None,
+    validate_test_data_keys: bool = True,
     s3_client: Any | None = None,
 ) -> DiscoveryResult:
-    """Discover documents in an S3-compatible bucket and optionally sample them.
+    """Discover documents across one or more bucket locations and optionally sample them.
 
-    Lists objects under *bucket_name*/*prefix*, filters by file extension,
-    and applies size-based sampling when enabled.  Documents referenced by
+    Lists objects under every entry of *prefixes*, merges the results into a
+    single corpus deduplicated by object key, filters by file extension, and
+    applies size-based sampling when enabled.  The sampling budget is shared by
+    the whole union, not applied per prefix.  Documents referenced by
     ``test_data_doc_names`` are prioritized during sampling so that
     benchmark-relevant files are always included when the budget permits.
 
@@ -113,12 +122,16 @@ def discover_documents(
     ----------
     bucket_name : str
         S3-compatible bucket name.
-    prefix : str, default=""
-        Object-key prefix to narrow the listing.
+    prefixes : str | list[str] | None, default=None
+        Object-key prefixes to narrow the listing.  A bare string is treated as
+        a one-element list.  ``None``, an empty list, or a list containing an
+        empty string lists the whole bucket.  Overlapping prefixes are safe:
+        objects matched by more than one are kept once.
     test_data_doc_names : list[str] | None, default=None
-        Keys of documents referenced by the benchmark test data, matched
-        against either the full object key or the bare file name.  These are
-        sorted first so that sampling picks them before other files.
+        Keys of documents referenced by the benchmark test data.  Each is
+        matched against the full object key, or -- when the name resolves to
+        exactly one document -- against a bare file name.  Matched documents
+        are sorted first so that sampling picks them before other files.
     sampling_enabled : bool, default=True
         When ``True``, only documents up to *sampling_max_size_gb* total
         are returned.
@@ -127,6 +140,10 @@ def discover_documents(
     supported_extensions : set[str] | None, default=None
         File extensions to accept.  Defaults to
         :data:`~ai4rag.utils.data.constants.SUPPORTED_EXTENSIONS`.
+    validate_test_data_keys : bool, default=True
+        When ``True``, every entry of *test_data_doc_names* must identify
+        exactly one discovered document, otherwise :class:`BenchmarkKeyError`
+        is raised.  Set to ``False`` to downgrade the failure to a warning.
     s3_client : Any | None, default=None
         Pre-configured ``boto3`` S3 client.  When ``None``, one is created
         via :func:`ai4rag.utils.clients.s3.create_s3_client`.
@@ -139,32 +156,37 @@ def discover_documents(
     Raises
     ------
     RuntimeError
-        If no supported documents are found in the bucket.
+        If no supported documents are found under any of the prefixes.
+    BenchmarkKeyError
+        If a benchmark key matches no discovered document, or matches more
+        than one, while *validate_test_data_keys* is enabled.
     ValueError
         If sampling produces an empty selection.
     """
     if supported_extensions is None:
         supported_extensions = set(SUPPORTED_EXTENSIONS)
 
+    resolved_prefixes = _normalize_prefixes(prefixes)
     ext_tuple = tuple(supported_extensions)
     max_size_bytes = float(sampling_max_size_gb) * 1024**3 if sampling_enabled else float(inf)
 
     if s3_client is None:
-        s3_client, contents = _list_objects_with_ssl_fallback(bucket_name, prefix)
-    else:
-        contents = s3_client.list_objects_v2(Bucket=bucket_name, Prefix=prefix).get("Contents", [])
+        s3_client = _create_s3_client_with_ssl_fallback(bucket_name, resolved_prefixes[0])
+    contents = _list_objects_union(s3_client, bucket_name, resolved_prefixes)
     supported_files = [c for c in contents if c["Key"].endswith(ext_tuple)]
 
     if not supported_files:
-        raise RuntimeError("No supported documents found.")
+        raise RuntimeError(f"No supported documents found in {_location(bucket_name, resolved_prefixes)}.")
 
+    test_keys: set[str] = set()
     if test_data_doc_names:
-        test_names_set = set(test_data_doc_names)
-        # Benchmark data names documents by their full object key, but a bare
-        # file name is still a valid identifier for a flat corpus, so accept both.
-        test_keys = {
-            c["Key"] for c in supported_files if c["Key"] in test_names_set or Path(c["Key"]).name in test_names_set
-        }
+        test_keys = _resolve_test_data_keys(
+            test_data_doc_names,
+            supported_files,
+            bucket_name=bucket_name,
+            prefixes=resolved_prefixes,
+            strict=validate_test_data_keys,
+        )
         supported_files.sort(key=lambda c: c["Key"] not in test_keys)
 
     total_size = 0
@@ -178,40 +200,178 @@ def discover_documents(
 
     if not selected:
         raise ValueError(
-            "No documents to process. Check that the bucket/prefix is correct and contains supported files."
+            "No documents to process. Check that the bucket/prefixes are correct and contain supported files."
+        )
+
+    dropped = sorted(test_keys - {d.key for d in selected})
+    if dropped:
+        _logger.warning(
+            "%d benchmark-referenced document(s) exceed the %.2f GB sampling budget and were skipped: %s",
+            len(dropped),
+            sampling_max_size_gb,
+            ", ".join(dropped),
         )
 
     result = DiscoveryResult(
         bucket=bucket_name,
-        prefix=prefix,
+        prefixes=resolved_prefixes,
         documents=selected,
         total_size_bytes=total_size,
         count=len(selected),
     )
-    _logger.info("Discovered %d document(s), total size %d bytes", result.count, result.total_size_bytes)
+    _logger.info(
+        "Discovered %d document(s) across %d location(s), total size %d bytes",
+        result.count,
+        len(resolved_prefixes),
+        result.total_size_bytes,
+    )
     return result
 
 
-def _list_objects_with_ssl_fallback(bucket_name: str, prefix: str) -> tuple[Any, list[dict]]:
-    """List S3 objects, retrying with ``verify=False`` on SSL errors.
+def _normalize_prefixes(prefixes: str | list[str] | None) -> list[str]:
+    """Coerce the *prefixes* argument into a deduplicated list of key prefixes.
+
+    A bare string becomes a one-element list.  Entries are stripped of
+    surrounding whitespace and of a leading ``/`` (object keys never start with
+    one).  Order is preserved.  An empty result, or any entry that is itself
+    empty, collapses to ``[""]`` -- the whole bucket, which subsumes every
+    other prefix.
+    """
+    if prefixes is None:
+        prefixes = []
+    elif isinstance(prefixes, str):
+        prefixes = [prefixes]
+
+    normalized: list[str] = []
+    for prefix in prefixes:
+        cleaned = (prefix or "").strip().lstrip("/")
+        if cleaned not in normalized:
+            normalized.append(cleaned)
+
+    if not normalized or "" in normalized:
+        if len(normalized) > 1:
+            _logger.info("An empty prefix was given alongside others; listing the whole bucket instead.")
+        return [""]
+    return normalized
+
+
+def _location(bucket_name: str, prefixes: list[str]) -> str:
+    """Render the listed locations for log and error messages."""
+    if prefixes == [""]:
+        return f"s3://{bucket_name}"
+    return ", ".join(f"s3://{bucket_name}/{prefix}" for prefix in prefixes)
+
+
+def _list_objects(s3_client: Any, bucket_name: str, prefix: str) -> list[dict]:
+    """List every object under *prefix*, following pagination to the end."""
+    paginator = s3_client.get_paginator("list_objects_v2")
+    contents: list[dict] = []
+    for page in paginator.paginate(Bucket=bucket_name, Prefix=prefix):
+        contents.extend(page.get("Contents", []))
+    return contents
+
+
+def _list_objects_union(s3_client: Any, bucket_name: str, prefixes: list[str]) -> list[dict]:
+    """List all *prefixes* and merge them into one key-deduplicated, key-sorted list."""
+    merged: dict[str, dict] = {}
+    for prefix in prefixes:
+        objects = _list_objects(s3_client, bucket_name, prefix)
+        _logger.info("Listed %d object(s) under s3://%s/%s", len(objects), bucket_name, prefix)
+        for obj in objects:
+            merged.setdefault(obj["Key"], obj)
+    return sorted(merged.values(), key=lambda obj: obj["Key"])
+
+
+def _resolve_test_data_keys(
+    test_data_doc_names: list[str],
+    supported_files: list[dict],
+    bucket_name: str,
+    prefixes: list[str],
+    strict: bool,
+) -> set[str]:
+    """Map benchmark document names onto discovered object keys.
+
+    An exact object-key match always wins.  A bare file name is accepted only
+    when it identifies exactly one document in the corpus -- across several
+    locations the same file name can appear more than once, and guessing which
+    one the benchmark meant would silently score against the wrong document.
+
+    Raises
+    ------
+    BenchmarkKeyError
+        When *strict* and any name matches no document or more than one.
+    """
+    all_keys = {c["Key"] for c in supported_files}
+    by_basename: dict[str, list[str]] = {}
+    for key in all_keys:
+        by_basename.setdefault(Path(key).name, []).append(key)
+
+    resolved: set[str] = set()
+    missing: list[str] = []
+    ambiguous: list[tuple[str, list[str]]] = []
+
+    for name in test_data_doc_names:
+        if name in all_keys:
+            resolved.add(name)
+            continue
+        candidates = by_basename.get(name, [])
+        if len(candidates) == 1:
+            resolved.add(candidates[0])
+        elif candidates:
+            ambiguous.append((name, sorted(candidates)))
+        else:
+            missing.append(name)
+
+    if missing or ambiguous:
+        message = _format_unresolved_keys(missing, ambiguous, bucket_name, prefixes)
+        if strict:
+            raise BenchmarkKeyError(message)
+        _logger.warning(message)
+
+    return resolved
+
+
+def _format_unresolved_keys(
+    missing: list[str],
+    ambiguous: list[tuple[str, list[str]]],
+    bucket_name: str,
+    prefixes: list[str],
+) -> str:
+    """Build the error message for benchmark keys that do not identify one document."""
+    lines = ["Benchmark data references documents that are not part of the discovered corpus."]
+    if missing:
+        lines.append("Not found: " + ", ".join(f"'{name}'" for name in sorted(missing)) + ".")
+    for name, candidates in sorted(ambiguous):
+        lines.append(
+            f"'{name}' is a file name shared by {len(candidates)} objects "
+            f"({', '.join(candidates)}); use the full object key instead."
+        )
+    lines.append(
+        "Every 'correct_answer_document_keys' entry must be the full object key including its "
+        "prefix, for example 'product-manuals/xr-200-manual.pdf'."
+    )
+    lines.append(f"Searched {_location(bucket_name, prefixes)}.")
+    return " ".join(lines)
+
+
+def _create_s3_client_with_ssl_fallback(bucket_name: str, probe_prefix: str) -> Any:
+    """Create an S3 client, retrying with ``verify=False`` on SSL errors.
 
     Returns
     -------
-    tuple[Any, list[dict]]
-        The S3 client and the ``Contents`` list from ``list_objects_v2``.
+    Any
+        A client that has successfully listed against *bucket_name*.
     """
     from botocore.exceptions import SSLError
 
+    client = create_s3_client()
     try:
-        client = create_s3_client()
-        contents = client.list_objects_v2(Bucket=bucket_name, Prefix=prefix).get("Contents", [])
-        return client, contents
+        client.list_objects_v2(Bucket=bucket_name, Prefix=probe_prefix, MaxKeys=1)
+        return client
     except SSLError:
         _logger.warning(
             "SSL error when listing objects in s3://%s/%s, retrying with verify=False",
             bucket_name,
-            prefix,
+            probe_prefix,
         )
-        client = create_s3_client(verify=False)
-        contents = client.list_objects_v2(Bucket=bucket_name, Prefix=prefix).get("Contents", [])
-        return client, contents
+        return create_s3_client(verify=False)
diff --git a/docs/about/changelog.md b/docs/about/changelog.md
index 04e49210..9c5873ae 100644
--- a/docs/about/changelog.md
+++ b/docs/about/changelog.md
@@ -12,7 +12,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Removed
 - **BREAKING CHANGE: Vector store** — removed the ChromaDB vector store backend (`ChromaConfig`, `ChromaVectorStore`) and the `chromadb` dependency, due to known security vulnerabilities in the `chromadb` package. `ai4rag.rag.vector_store` no longer exports `ChromaConfig`; the `"chroma"` value for `vector_store_type` is no longer accepted (only `"milvus"`, `"milvus_lite"`, and `"pgvector"` are supported)
 
+### Changed
+- **BREAKING CHANGE: Document discovery** — `discover_documents()` now ingests **several bucket locations at once**: the `prefix: str` parameter is replaced by `prefixes: str | list[str] | None`, and `DiscoveryResult.prefix` by `DiscoveryResult.prefixes: list[str]` (`to_dict()` and `documents_descriptor.json` emit `"prefixes"` accordingly). A bare string is still accepted and coerced to a one-element list, and omitting the argument still lists the whole bucket, so single-location callers only need to rename the keyword. Every prefix is listed and merged into one corpus deduplicated by object key — overlapping selections such as `docs/` and `docs/manuals/` are safe — and the `sampling_max_size_gb` budget applies to that union rather than to each location separately
+- **Document discovery** — listings are now paginated. `list_objects_v2` returns at most 1000 keys per call, so a location holding more than that was silently truncated; with several locations sharing one budget the truncation would have starved the later prefixes entirely
+- **BREAKING CHANGE: Asset generation** — `create_placeholder_mapping()` and `generate_notebook_from_template()` take `input_data_keys: list[str]` in place of `input_data_key: str`, and the indexing notebook template's `INPUT_DATA_KEY` placeholder becomes `INPUT_DATA_KEYS`, rendered as a Python list literal. The generated `indexing.ipynb` therefore rediscovers every location the pipeline ingested, not just the first
+
 ### Added
+- **Document discovery** — benchmark keys are now validated against the discovered corpus. Each `test_data_doc_names` entry must identify exactly one document: an exact object-key match always wins, and a bare file name is accepted only when it resolves to a single document. Anything else raises the new `BenchmarkKeyError` (a `ValueError` subclass, exported from `ai4rag.utils.data`) listing the offending keys, the locations searched, and the expected key format. Previously a `correct_answer_document_keys` entry that matched no ingested object — a prefix-relative key, say — simply retrieved nothing, leaving the question ungrounded and the scores quietly wrong. Pass `validate_test_data_keys=False` to downgrade the failure to a warning. A benchmark document that is discovered but dropped by the sampling budget is reported separately, also as a warning
 - **Vector store** — Milvus Lite, the embedded, zero-server mode of the Milvus backend, is now the recommended local/zero-config replacement for the removed Chroma store: use the new `MilvusLiteConfig(db_path="./ai4rag.db")` (or `MilvusLiteConfig()` for the default path). Unlike Chroma, Milvus Lite supports hybrid (dense + BM25) search. `pymilvus[milvus-lite]` is now a core dependency, so no extra installation step is needed
 - **Vector store** — the Milvus config was split into `MilvusConfig` (remote server / Zilliz Cloud only, which now validates that `uri` is an `http(s)://` URL and raises `ValueError` otherwise) and a new `MilvusLiteConfig` (embedded, local only, configured via `db_path`, which conversely rejects `http(s)://` values). Previously, a single `MilvusConfig` selected between a remote server and embedded Milvus Lite based on whether `uri` looked like a URL or a local file path; a mistyped or unreachable `MILVUS_URI` could therefore be silently interpreted as a local path and create an unintended throwaway local database. That silent fallback is no longer possible — a misconfigured server URI now fails loudly instead. `ai4rag.rag.vector_store` now also exports `MilvusLiteConfig`, and the `vector_store_type` search-space parameter accepts `"milvus_lite"` in addition to `"milvus"` and `"pgvector"`
 
diff --git a/docs/api-reference/utils/data.md b/docs/api-reference/utils/data.md
index ba8a87a4..7658879c 100644
--- a/docs/api-reference/utils/data.md
+++ b/docs/api-reference/utils/data.md
@@ -10,6 +10,7 @@ Data processing functions for the AutoRAG pipeline.
         - discover_documents
         - DiscoveryResult
         - DocumentDescriptor
+        - BenchmarkKeyError
 
 ## Text Extraction
 
diff --git a/docs/user-guide/evaluation.md b/docs/user-guide/evaluation.md
index 7aa0f5b8..92317a5f 100644
--- a/docs/user-guide/evaluation.md
+++ b/docs/user-guide/evaluation.md
@@ -390,20 +390,19 @@ This makes evaluation more robust to phrasing variations.
 
 **3. Accurate Document Keys**
 
-Ensure `correct_answer_document_keys` match the document keys in your knowledge base (stored on each chunk as the `document_id` metadata field):
+A document key is the document's `DoclingDocument.name`. When the corpus comes from a bucket, that is
+the full S3 object key — prefix included — and discovery rejects a benchmark key that matches no
+ingested object. Set the name explicitly when you build the documents yourself:
 
 ```python
-# When loading documents
-from langchain_core.documents import Document
-
-documents = [
-    Document(
-        page_content="...",
-        metadata={"document_id": "readme.md"}  # Must match benchmark data
-    )
-]
+from docling_core.types.doc import DoclingDocument
+
+document = DoclingDocument(name="guides/readme.md")  # Must match benchmark data
 ```
 
+Chunks carry the same value internally under the `document_id` metadata field; it surfaces as
+`document_key` in `evaluation_results.json`.
+
 ---
 
 **4. Representative Coverage**
diff --git a/docs/user-guide/event-handlers.md b/docs/user-guide/event-handlers.md
index 806a84d9..02fe5551 100644
--- a/docs/user-guide/event-handlers.md
+++ b/docs/user-guide/event-handlers.md
@@ -103,7 +103,7 @@ def on_pattern_creation(
         "answer": "According to the document ...",
         "correct_answers": ["The correct answer is ..."],
         "answer_contexts": [
-            {"text": "Retrieved chunk text ...", "document_id": "doc1.pdf"},
+            {"text": "Retrieved chunk text ...", "document_key": "manuals/doc1.pdf"},
         ],
         "metrics": [
             {"name": "faithfulness", "evaluator": "unitxt", "score": 0.71},
diff --git a/docs/user-guide/pipeline-components.md b/docs/user-guide/pipeline-components.md
index c240bd59..062dc60f 100644
--- a/docs/user-guide/pipeline-components.md
+++ b/docs/user-guide/pipeline-components.md
@@ -46,14 +46,14 @@ S3 support (`boto3`), multiprocessing (`multiprocess`), and text extraction for
 
 ### Document Discovery
 
-List and sample documents from an S3-compatible bucket:
+List and sample documents from one or more locations in an S3-compatible bucket:
 
 ```python
 from ai4rag.utils.data import discover_documents
 
 result = discover_documents(
     bucket_name="my-bucket",
-    prefix="documents/",
+    prefixes=["documents/", "manuals/"],
     sampling_enabled=True,
     sampling_max_size_gb=1.0,
 )
@@ -61,6 +61,28 @@ print(f"Found {result.count} documents ({result.total_size_bytes} bytes)")
 result.save("/tmp/discovery_output")
 ```
 
+Every prefix is listed and merged into a single corpus deduplicated by object key, so overlapping
+selections (`docs/` and `docs/manuals/`) are safe. The sampling budget applies to that union, not to
+each location. Omitting `prefixes` — or passing an empty list — lists the whole bucket; a bare string
+is accepted as a single location.
+
+When `test_data_doc_names` is given, every entry must identify exactly one discovered document.
+A key that matches nothing, or a bare file name shared by documents in two locations, raises
+`BenchmarkKeyError` naming the offending keys rather than leaving the question silently ungrounded:
+
+```python
+from ai4rag.utils.data import BenchmarkKeyError, discover_documents
+
+try:
+    result = discover_documents(
+        bucket_name="my-bucket",
+        prefixes=["documents/"],
+        test_data_doc_names=["documents/report.pdf"],
+    )
+except BenchmarkKeyError as exc:
+    print(exc)  # names the keys and the locations that were searched
+```
+
 ### Text Extraction
 
 Download documents from S3 and extract text using Docling:
diff --git a/tests/integration/test_object_key_preservation.py b/tests/integration/test_object_key_preservation.py
index 8433f761..7e94ea45 100644
--- a/tests/integration/test_object_key_preservation.py
+++ b/tests/integration/test_object_key_preservation.py
@@ -28,7 +28,7 @@
 from ai4rag.core.experiment.results import EvaluationResult, ExperimentResults
 from ai4rag.evaluator.base_evaluator import EvaluationData
 from ai4rag.utils.data import text_extraction
-from ai4rag.utils.data.documents_discovery import discover_documents
+from ai4rag.utils.data.documents_discovery import BenchmarkKeyError, discover_documents
 
 # A corpus built around the collision this naming scheme exists to prevent:
 # two documents sharing a basename under different folders.
@@ -45,15 +45,30 @@
 # ---------------------------------------------------------------------------
 
 
+class _FakePaginator:
+    """Page a listing two objects at a time, the way boto3's paginator does."""
+
+    PAGE_SIZE = 2
+
+    def __init__(self, objects: dict[str, str]):
+        self._objects = objects
+
+    def paginate(self, Bucket: str, Prefix: str = "", **_kwargs):  # noqa: N803 - boto3 casing
+        del Bucket
+        matching = [{"Key": k, "Size": len(v)} for k, v in self._objects.items() if k.startswith(Prefix)]
+        for start in range(0, max(len(matching), 1), self.PAGE_SIZE):
+            yield {"Contents": matching[start : start + self.PAGE_SIZE]}
+
+
 class _FakeS3Client:
     """Minimal stand-in for the boto3 S3 client used by discovery and download."""
 
     def __init__(self, objects: dict[str, str]):
         self._objects = objects
 
-    def list_objects_v2(self, Bucket: str, Prefix: str = ""):  # noqa: N803 - boto3 casing
-        del Bucket
-        return {"Contents": [{"Key": k, "Size": len(v)} for k, v in self._objects.items() if k.startswith(Prefix)]}
+    def get_paginator(self, operation_name: str) -> _FakePaginator:
+        assert operation_name == "list_objects_v2"
+        return _FakePaginator(self._objects)
 
     def download_file(self, Bucket: str, Key: str, Filename: str):  # noqa: N803 - boto3 casing
         del Bucket
@@ -94,7 +109,7 @@ def extracted(fake_bucket, tmp_path) -> Path:
     produces the descriptor, whose ``documents`` list is handed to extraction
     exactly as ``pipelines-components`` hands it over.
     """
-    return _ingest(discover_documents(bucket_name="bucket", prefix=PREFIX, s3_client=fake_bucket), tmp_path)
+    return _ingest(discover_documents(bucket_name="bucket", prefixes=[PREFIX], s3_client=fake_bucket), tmp_path)
 
 
 def _ingest(discovery, tmp_path: Path) -> Path:
@@ -161,6 +176,20 @@ def test_names_do_not_depend_on_the_discovery_prefix(self, fake_bucket, tmp_path
 
         assert _extracted_names(out_dir) == set(CORPUS)
 
+    def test_several_locations_ingest_into_one_corpus(self, fake_bucket, tmp_path):
+        """Selecting two folders ingests both, and colliding basenames stay distinct."""
+        discovery = discover_documents(
+            bucket_name="bucket",
+            prefixes=[f"{PREFIX}/manuals/xr-200", f"{PREFIX}/manuals/xr-300"],
+            s3_client=fake_bucket,
+        )
+        out_dir = _ingest(discovery, tmp_path)
+
+        assert _extracted_names(out_dir) == {
+            f"{PREFIX}/manuals/xr-200/setup.txt",
+            f"{PREFIX}/manuals/xr-300/setup.txt",
+        }
+
     def test_key_needing_normalisation_still_names_the_document(self, monkeypatch, tmp_path):
         """A key needing normalisation must not fall back to the bare filename.
 
@@ -211,30 +240,47 @@ def _benchmark(document_keys: list[str]) -> BenchmarkData:
             )
         )
 
-    def test_object_keys_select_the_intended_document(self, extracted):
+    @staticmethod
+    def _discover(fake_bucket, benchmark: BenchmarkData):
+        """Run discovery the way the pipeline does: benchmark keys drive sampling."""
+        return discover_documents(
+            bucket_name="bucket",
+            prefixes=[PREFIX],
+            test_data_doc_names=sorted({key for keys in benchmark.document_keys for key in keys}),
+            s3_client=fake_bucket,
+        )
+
+    def test_object_keys_select_the_intended_document(self, extracted, fake_bucket):
         """A benchmark written against object keys resolves to exactly one document."""
         benchmark = self._benchmark([f"{PREFIX}/manuals/xr-300/setup.txt"])
         referenced = {key for keys in benchmark.document_keys for key in keys}
 
         assert referenced & _extracted_names(extracted) == {f"{PREFIX}/manuals/xr-300/setup.txt"}
+        assert self._discover(fake_bucket, benchmark).count == len(CORPUS)
 
-    def test_prefix_relative_keys_match_nothing(self, extracted):
+    def test_prefix_relative_keys_are_rejected(self, fake_bucket):
         """The misconfiguration to recognise: keys written without the bucket prefix.
 
-        Nothing raises here -- the documents simply never match, so the run
-        completes with no grounding for that question and the scores reflect it.
+        Such a key matches no ingested object, which used to leave the question
+        silently ungrounded.  Discovery now fails and names the key.
         """
         benchmark = self._benchmark(["manuals/xr-300/setup.txt"])
-        referenced = {key for keys in benchmark.document_keys for key in keys}
 
-        assert not referenced & _extracted_names(extracted)
+        with pytest.raises(BenchmarkKeyError, match="'manuals/xr-300/setup.txt'"):
+            self._discover(fake_bucket, benchmark)
 
-    def test_bare_filenames_are_ambiguous_across_folders(self, extracted):
+    def test_bare_filenames_are_ambiguous_across_folders(self, fake_bucket):
         """A basename cannot address a document once folders are in play."""
         benchmark = self._benchmark(["setup.txt"])
-        referenced = {key for keys in benchmark.document_keys for key in keys}
 
-        assert not referenced & _extracted_names(extracted)
+        with pytest.raises(BenchmarkKeyError, match="shared by 2 objects"):
+            self._discover(fake_bucket, benchmark)
+
+    def test_unique_basename_is_still_accepted(self, fake_bucket):
+        """A file name that resolves to one document keeps working."""
+        benchmark = self._benchmark(["overview.txt"])
+
+        assert self._discover(fake_bucket, benchmark).count == len(CORPUS)
 
 
 # ---------------------------------------------------------------------------
diff --git a/tests/unit/ai4rag/assets_generator/test_templates.py b/tests/unit/ai4rag/assets_generator/test_templates.py
index 6379ea5f..4f561913 100644
--- a/tests/unit/ai4rag/assets_generator/test_templates.py
+++ b/tests/unit/ai4rag/assets_generator/test_templates.py
@@ -57,7 +57,7 @@ def mapping(self) -> dict:
         return create_placeholder_mapping(
             _SAMPLE_PATTERN_DATA,
             test_data_key="s3://bucket/test.jsonl",
-            input_data_key="s3://bucket/docs/",
+            input_data_keys=["s3://bucket/docs/", "s3://bucket/manuals/"],
         )
 
     def test_pattern_name(self, mapping: dict):
@@ -102,7 +102,12 @@ def test_chunking_fields(self, mapping: dict):
 
     def test_s3_keys(self, mapping: dict):
         assert mapping["TEST_DATA_KEY"] == "s3://bucket/test.jsonl"
-        assert mapping["INPUT_DATA_KEY"] == "s3://bucket/docs/"
+        # Rendered as a literal so the notebook line stays valid Python.
+        assert mapping["INPUT_DATA_KEYS"] == '["s3://bucket/docs/", "s3://bucket/manuals/"]'
+
+    def test_input_data_keys_default_to_an_empty_list(self):
+        """A pattern generated without locations must still render a valid literal."""
+        assert create_placeholder_mapping({})["INPUT_DATA_KEYS"] == "[]"
 
     def test_all_expected_keys_present(self, mapping: dict):
         """All documented placeholder names must appear in the mapping."""
@@ -128,7 +133,7 @@ def test_all_expected_keys_present(self, mapping: dict):
             "CHUNK_SIZE",
             "CHUNK_OVERLAP",
             "TEST_DATA_KEY",
-            "INPUT_DATA_KEY",
+            "INPUT_DATA_KEYS",
         }
         assert expected_keys.issubset(set(mapping.keys()))
 
@@ -198,13 +203,13 @@ def test_passes_s3_keys(self, mocker, tmp_path: Path):
             output_data={},
             output_notebook_path=tmp_path / "out.ipynb",
             test_data_key="key/test",
-            input_data_key="key/input",
+            input_data_keys=["key/input"],
         )
 
         mock_create.assert_called_once_with(
             {},
             test_data_key="key/test",
-            input_data_key="key/input",
+            input_data_keys=["key/input"],
         )
 
 
@@ -290,6 +295,26 @@ def test_generated_notebook_has_no_unresolved_placeholders(self, template: str,
         assert leftover == [], f"Unresolved placeholders: {leftover}"
 
 
+def test_indexing_notebook_renders_every_input_location(tmp_path: Path):
+    """The indexing notebook must rediscover the same corpus the pipeline ingested.
+
+    The keys are rendered as a Python list literal and handed to
+    ``discover_documents`` as ``prefixes``, so a reader re-running the notebook
+    gets every location, not just the first.
+    """
+    output_path = tmp_path / "maas_indexing.ipynb"
+    generate_notebook_from_template(
+        notebook_template="maas_indexing",
+        output_data=_SAMPLE_PATTERN_DATA,
+        output_notebook_path=output_path,
+        input_data_keys=["manuals/", "reports/"],
+    )
+    text = _read_notebook_text(output_path)
+
+    assert 'input_data_keys = ["manuals/", "reports/"]' in text
+    assert "prefixes=input_data_keys" in text
+
+
 def test_inference_notebook_passes_detected_language(tmp_path: Path):
     """The inference notebook must rebuild the detected language and pass it to the foundation model.
 
diff --git a/tests/unit/ai4rag/utils/data/test_discovery.py b/tests/unit/ai4rag/utils/data/test_discovery.py
index 2dd3f439..f22fa47c 100644
--- a/tests/unit/ai4rag/utils/data/test_discovery.py
+++ b/tests/unit/ai4rag/utils/data/test_discovery.py
@@ -8,6 +8,7 @@
 
 from ai4rag.utils.data.documents_discovery import (
     DOCUMENTS_DESCRIPTOR_FILENAME,
+    BenchmarkKeyError,
     DiscoveryResult,
     DocumentDescriptor,
     discover_documents,
@@ -24,9 +25,26 @@ def _s3_object(key: str, size: int) -> dict:
 
 
 def _make_mock_s3_client(mocker, contents: list[dict]):
-    """Return a mock S3 client whose ``list_objects_v2`` yields *contents*."""
+    """Return a mock S3 client whose paginator yields the objects under a prefix.
+
+    Mirrors the real API: ``paginate`` returns only the objects whose key starts
+    with the requested prefix, in a single page.
+    """
     mock = mocker.MagicMock()
-    mock.list_objects_v2.return_value = {"Contents": contents}
+
+    def _paginate(Bucket, Prefix, **_kwargs):  # noqa: N803 - boto3 kwarg names
+        return iter([{"Contents": [c for c in contents if c["Key"].startswith(Prefix)]}])
+
+    mock.get_paginator.return_value.paginate.side_effect = _paginate
+    return mock
+
+
+def _make_paginated_mock_s3_client(mocker, pages: list[list[dict]]):
+    """Return a mock S3 client whose paginator yields *pages* verbatim."""
+    mock = mocker.MagicMock()
+    mock.get_paginator.return_value.paginate.side_effect = lambda **_kwargs: iter(
+        [{"Contents": page} for page in pages]
+    )
     return mock
 
 
@@ -68,7 +86,7 @@ def result(self) -> DiscoveryResult:
         ]
         return DiscoveryResult(
             bucket="test-bucket",
-            prefix="docs/",
+            prefixes=["docs/"],
             documents=docs,
             total_size_bytes=300,
             count=2,
@@ -78,7 +96,7 @@ def test_to_dict_structure(self, result: DiscoveryResult):
         """``to_dict`` must produce JSON-serialisable output with correct keys."""
         d = result.to_dict()
         assert d["bucket"] == "test-bucket"
-        assert d["prefix"] == "docs/"
+        assert d["prefixes"] == ["docs/"]
         assert d["total_size_bytes"] == 300
         assert d["count"] == 2
         assert len(d["documents"]) == 2
@@ -136,7 +154,7 @@ def test_happy_path_returns_all_supported(self, mocker):
 
         result = discover_documents(
             bucket_name="bucket",
-            prefix="docs/",
+            prefixes=["docs/"],
             sampling_enabled=False,
             s3_client=mock_client,
         )
@@ -144,7 +162,7 @@ def test_happy_path_returns_all_supported(self, mocker):
         assert result.count == 13
         assert result.total_size_bytes == 2000
         assert result.bucket == "bucket"
-        assert result.prefix == "docs/"
+        assert result.prefixes == ["docs/"]
         keys = [d.key for d in result.documents]
         assert "docs/report.pdf" in keys
         assert "docs/notes.md" in keys
@@ -253,7 +271,7 @@ def test_test_data_doc_names_prioritised(self, mocker):
 
         result = discover_documents(
             bucket_name="bucket",
-            prefix="docs/",
+            prefixes=["docs/"],
             test_data_doc_names=["benchmark.pdf", "important.md"],
             sampling_enabled=False,
             s3_client=mock_client,
@@ -274,7 +292,7 @@ def test_test_data_prioritised_under_sampling(self, mocker):
 
         result = discover_documents(
             bucket_name="bucket",
-            prefix="docs/",
+            prefixes=["docs/"],
             test_data_doc_names=["benchmark.pdf"],
             sampling_enabled=True,
             sampling_max_size_gb=900 / 1024**3,
@@ -341,18 +359,37 @@ def test_custom_supported_extensions(self, mocker):
         assert result.documents[0].key == "a.csv"
 
     def test_list_objects_called_correctly(self, mocker):
-        """``list_objects_v2`` must receive the correct bucket and prefix."""
+        """The paginator must receive the correct bucket and prefix."""
         contents = [_s3_object("prefix/x.pdf", 10)]
         mock_client = _make_mock_s3_client(mocker, contents)
 
         discover_documents(
             bucket_name="my-bucket",
-            prefix="prefix/",
+            prefixes=["prefix/"],
             sampling_enabled=False,
             s3_client=mock_client,
         )
 
-        mock_client.list_objects_v2.assert_called_once_with(Bucket="my-bucket", Prefix="prefix/")
+        mock_client.get_paginator.assert_called_with("list_objects_v2")
+        mock_client.get_paginator.return_value.paginate.assert_called_once_with(Bucket="my-bucket", Prefix="prefix/")
+
+    def test_listing_follows_pagination(self, mocker):
+        """Listings longer than one page must not be truncated."""
+        pages = [
+            [_s3_object(f"docs/page1-{i}.pdf", 10) for i in range(1000)],
+            [_s3_object("docs/page2-0.pdf", 10)],
+        ]
+        mock_client = _make_paginated_mock_s3_client(mocker, pages)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes=["docs/"],
+            sampling_enabled=False,
+            s3_client=mock_client,
+        )
+
+        assert result.count == 1001
+        assert "docs/page2-0.pdf" in {d.key for d in result.documents}
 
     def test_audio_extensions_discovered(self, mocker):
         """Audio files with supported extensions must be discovered."""
@@ -368,7 +405,7 @@ def test_audio_extensions_discovered(self, mocker):
 
         result = discover_documents(
             bucket_name="bucket",
-            prefix="audio/",
+            prefixes=["audio/"],
             sampling_enabled=False,
             s3_client=mock_client,
         )
@@ -396,7 +433,7 @@ def test_mixed_audio_and_document_extensions(self, mocker):
 
         result = discover_documents(
             bucket_name="bucket",
-            prefix="data/",
+            prefixes=["data/"],
             sampling_enabled=False,
             s3_client=mock_client,
         )
@@ -429,7 +466,7 @@ def test_nested_keys_are_prioritised(self, mocker):
 
         result = discover_documents(
             bucket_name="bucket",
-            prefix="docs",
+            prefixes=["docs"],
             test_data_doc_names=["docs/manuals/xr-200/setup.pdf"],
             sampling_enabled=True,
             sampling_max_size_gb=400 / 1024**3,
@@ -448,7 +485,7 @@ def test_bare_filename_benchmark_keys_still_prioritised(self, mocker):
 
         result = discover_documents(
             bucket_name="bucket",
-            prefix="docs",
+            prefixes=["docs"],
             test_data_doc_names=["benchmark.pdf"],
             sampling_enabled=True,
             sampling_max_size_gb=400 / 1024**3,
@@ -467,7 +504,7 @@ def test_same_basename_in_different_folders_stays_distinct(self, mocker):
 
         result = discover_documents(
             bucket_name="bucket",
-            prefix="docs",
+            prefixes=["docs"],
             sampling_enabled=False,
             s3_client=mock_client,
         )
@@ -475,3 +512,299 @@ def test_same_basename_in_different_folders_stays_distinct(self, mocker):
         keys = [d.key for d in result.documents]
         assert keys == ["docs/a/setup.txt", "docs/b/setup.txt"]
         assert len(set(keys)) == 2
+
+
+# ---------------------------------------------------------------------------
+# Multi-location discovery
+# ---------------------------------------------------------------------------
+
+
+class TestMultiplePrefixes:
+    """Documents from every selected location form one corpus."""
+
+    CONTENTS = [
+        _s3_object("manuals/setup.pdf", 100),
+        _s3_object("manuals/nested/spec.pdf", 100),
+        _s3_object("reports/q1.pdf", 100),
+        _s3_object("archive/old.pdf", 100),
+    ]
+
+    def test_union_covers_every_prefix(self, mocker):
+        """Every listed prefix contributes its documents."""
+        mock_client = _make_mock_s3_client(mocker, self.CONTENTS)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes=["manuals/", "reports/"],
+            sampling_enabled=False,
+            s3_client=mock_client,
+        )
+
+        assert [d.key for d in result.documents] == [
+            "manuals/nested/spec.pdf",
+            "manuals/setup.pdf",
+            "reports/q1.pdf",
+        ]
+        assert result.prefixes == ["manuals/", "reports/"]
+
+    def test_overlapping_prefixes_are_deduplicated(self, mocker):
+        """An object matched by two prefixes is kept once."""
+        mock_client = _make_mock_s3_client(mocker, self.CONTENTS)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes=["manuals/", "manuals/nested/"],
+            sampling_enabled=False,
+            s3_client=mock_client,
+        )
+
+        keys = [d.key for d in result.documents]
+        assert keys == ["manuals/nested/spec.pdf", "manuals/setup.pdf"]
+        assert result.count == 2
+
+    def test_sampling_budget_is_shared_across_prefixes(self, mocker):
+        """The size cap applies to the union, not to each location separately."""
+        contents = [
+            _s3_object("a/one.pdf", 100),
+            _s3_object("b/two.pdf", 100),
+            _s3_object("c/three.pdf", 100),
+        ]
+        mock_client = _make_mock_s3_client(mocker, contents)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes=["a/", "b/", "c/"],
+            sampling_enabled=True,
+            sampling_max_size_gb=200 / 1024**3,
+            s3_client=mock_client,
+        )
+
+        assert result.total_size_bytes == 200
+        assert result.count == 2
+
+    def test_benchmark_priority_spans_prefixes(self, mocker):
+        """A benchmark document in the last location still wins the budget."""
+        contents = [
+            _s3_object("a/filler.pdf", 100),
+            _s3_object("z/benchmark.pdf", 100),
+        ]
+        mock_client = _make_mock_s3_client(mocker, contents)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes=["a/", "z/"],
+            test_data_doc_names=["z/benchmark.pdf"],
+            sampling_enabled=True,
+            sampling_max_size_gb=100 / 1024**3,
+            s3_client=mock_client,
+        )
+
+        assert [d.key for d in result.documents] == ["z/benchmark.pdf"]
+
+    def test_bare_string_prefix_is_coerced(self, mocker):
+        """A single string is accepted for backward compatibility."""
+        mock_client = _make_mock_s3_client(mocker, self.CONTENTS)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes="reports/",
+            sampling_enabled=False,
+            s3_client=mock_client,
+        )
+
+        assert result.prefixes == ["reports/"]
+        assert [d.key for d in result.documents] == ["reports/q1.pdf"]
+
+    @pytest.mark.parametrize("prefixes", [None, [], [""], ["", "manuals/"]])
+    def test_empty_prefix_lists_whole_bucket(self, mocker, prefixes):
+        """No prefix -- or an empty one among others -- means the whole bucket."""
+        mock_client = _make_mock_s3_client(mocker, self.CONTENTS)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes=prefixes,
+            sampling_enabled=False,
+            s3_client=mock_client,
+        )
+
+        assert result.prefixes == [""]
+        assert result.count == len(self.CONTENTS)
+
+    def test_prefixes_are_normalised(self, mocker):
+        """Leading slashes are stripped and duplicates dropped, order preserved."""
+        mock_client = _make_mock_s3_client(mocker, self.CONTENTS)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes=["/reports/", "manuals/", "reports/"],
+            sampling_enabled=False,
+            s3_client=mock_client,
+        )
+
+        assert result.prefixes == ["reports/", "manuals/"]
+        assert result.count == 3
+
+    def test_no_documents_in_any_prefix_names_the_locations(self, mocker):
+        """The error tells the user which locations were searched."""
+        mock_client = _make_mock_s3_client(mocker, [_s3_object("other/x.pdf", 10)])
+
+        with pytest.raises(RuntimeError, match=r"s3://bucket/a/, s3://bucket/b/"):
+            discover_documents(
+                bucket_name="bucket",
+                prefixes=["a/", "b/"],
+                sampling_enabled=False,
+                s3_client=mock_client,
+            )
+
+
+# ---------------------------------------------------------------------------
+# Benchmark key validation
+# ---------------------------------------------------------------------------
+
+
+class TestBenchmarkKeyValidation:
+    """A benchmark key that matches no ingested object must fail loudly."""
+
+    CONTENTS = [
+        _s3_object("manuals/setup.pdf", 100),
+        _s3_object("reports/setup.pdf", 100),
+        _s3_object("reports/q1.pdf", 100),
+    ]
+
+    def test_unknown_key_raises(self, mocker):
+        """A key absent from the corpus is reported by name."""
+        mock_client = _make_mock_s3_client(mocker, self.CONTENTS)
+
+        with pytest.raises(BenchmarkKeyError, match="'manuals/missing.pdf'"):
+            discover_documents(
+                bucket_name="bucket",
+                prefixes=["manuals/", "reports/"],
+                test_data_doc_names=["manuals/setup.pdf", "manuals/missing.pdf"],
+                sampling_enabled=False,
+                s3_client=mock_client,
+            )
+
+    def test_ambiguous_basename_raises(self, mocker):
+        """A file name shared by two locations cannot identify one document."""
+        mock_client = _make_mock_s3_client(mocker, self.CONTENTS)
+
+        with pytest.raises(BenchmarkKeyError, match="shared by 2 objects"):
+            discover_documents(
+                bucket_name="bucket",
+                prefixes=["manuals/", "reports/"],
+                test_data_doc_names=["setup.pdf"],
+                sampling_enabled=False,
+                s3_client=mock_client,
+            )
+
+    def test_exact_key_wins_over_basename(self, mocker):
+        """A full object key is unambiguous even when the file name collides."""
+        mock_client = _make_mock_s3_client(mocker, self.CONTENTS)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes=["manuals/", "reports/"],
+            test_data_doc_names=["reports/setup.pdf"],
+            sampling_enabled=True,
+            sampling_max_size_gb=100 / 1024**3,
+            s3_client=mock_client,
+        )
+
+        assert [d.key for d in result.documents] == ["reports/setup.pdf"]
+
+    def test_error_message_explains_the_key_format(self, mocker):
+        """The message has to be actionable for someone editing the benchmark JSON."""
+        mock_client = _make_mock_s3_client(mocker, self.CONTENTS)
+
+        with pytest.raises(BenchmarkKeyError) as exc_info:
+            discover_documents(
+                bucket_name="bucket",
+                prefixes=["reports/"],
+                test_data_doc_names=["q1.pdf", "nope.pdf"],
+                sampling_enabled=False,
+                s3_client=mock_client,
+            )
+
+        message = str(exc_info.value)
+        assert "correct_answer_document_keys" in message
+        assert "full object key" in message
+        assert "s3://bucket/reports/" in message
+
+    def test_validation_can_be_disabled(self, mocker, caplog):
+        """With validation off an unmatched key only warns."""
+        mock_client = _make_mock_s3_client(mocker, self.CONTENTS)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes=["reports/"],
+            test_data_doc_names=["nope.pdf"],
+            sampling_enabled=False,
+            validate_test_data_keys=False,
+            s3_client=mock_client,
+        )
+
+        assert result.count == 2
+        assert "not part of the discovered corpus" in caplog.text
+
+    def test_benchmark_document_dropped_by_budget_warns(self, mocker, caplog):
+        """Benchmark docs that alone exceed the budget are flagged, not swallowed."""
+        contents = [
+            _s3_object("reports/small.pdf", 100),
+            _s3_object("reports/huge.pdf", 1000),
+        ]
+        mock_client = _make_mock_s3_client(mocker, contents)
+
+        result = discover_documents(
+            bucket_name="bucket",
+            prefixes=["reports/"],
+            test_data_doc_names=["reports/small.pdf", "reports/huge.pdf"],
+            sampling_enabled=True,
+            sampling_max_size_gb=500 / 1024**3,
+            s3_client=mock_client,
+        )
+
+        assert [d.key for d in result.documents] == ["reports/small.pdf"]
+        assert "reports/huge.pdf" in caplog.text
+        assert "sampling budget" in caplog.text
+
+
+# ---------------------------------------------------------------------------
+# S3 client creation
+# ---------------------------------------------------------------------------
+
+
+class TestS3ClientCreation:
+    """A client is built on demand, retrying without TLS verification when needed."""
+
+    CONTENTS = [_s3_object("docs/report.pdf", 100)]
+
+    def test_client_is_created_when_none_is_supplied(self, mocker):
+        """Without an explicit client, discovery builds a verified one."""
+        client = _make_mock_s3_client(mocker, self.CONTENTS)
+        create = mocker.patch(
+            "ai4rag.utils.data.documents_discovery.create_s3_client",
+            return_value=client,
+        )
+
+        result = discover_documents(bucket_name="bucket", prefixes=["docs/"], sampling_enabled=False)
+
+        create.assert_called_once_with()
+        client.list_objects_v2.assert_called_once_with(Bucket="bucket", Prefix="docs/", MaxKeys=1)
+        assert result.count == 1
+
+    def test_ssl_error_retries_without_verification(self, mocker):
+        """A self-signed endpoint falls back to an unverified client."""
+        from botocore.exceptions import SSLError
+
+        failing = mocker.MagicMock()
+        failing.list_objects_v2.side_effect = SSLError(endpoint_url="https://s3.example", error="self-signed")
+        working = _make_mock_s3_client(mocker, self.CONTENTS)
+        create = mocker.patch(
+            "ai4rag.utils.data.documents_discovery.create_s3_client",
+            side_effect=[failing, working],
+        )
+
+        result = discover_documents(bucket_name="bucket", prefixes=["docs/"], sampling_enabled=False)
+
+        assert create.call_args_list == [mocker.call(), mocker.call(verify=False)]
+        assert result.count == 1