diff --git a/CHANGELOG.md b/CHANGELOG.md
index 209804d..eb6b242 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,8 +13,36 @@ Refactors, CI, and formatting land in the git history, not here.
 
 ## [Unreleased]
 
+### Changed
+
+- **Breaking (beta): `POST /v3/cohort/counts` and `POST /v3/licenses` now take the filter under
+  `filters`**, like every other filter endpoint: `{"filters": {"terms": …}}`. The old bare body
+  (`{"terms": …}`) returns `422` naming the fix. The shapes used to differ per endpoint, which is
+  what made the bug below possible.
+- **Breaking (beta): the series-enumerating surfaces require at least one filter predicate** —
+  `POST /v3/cohort/manifest`, `POST /v3/cohort/manifest.txt`, and the MCP `build_cohort` /
+  `get_cohort_urls`. Unfiltered they returned a download payload for the whole archive; they now
+  return `400`. The aggregate surfaces (`cohort/counts`, `licenses`) still answer an unfiltered
+  filter, with a warning.
+
+### Added
+
+- Responses built from a cohort filter now carry `filters_applied` (the predicates that reached
+  SQL) and `warnings` (predicates dropped, whether nothing was filtered, and — when a cohort
+  matches nothing only because of letter case — the casing that does exist): `POST /v3/cohort/counts`,
+  `POST /v3/licenses`, `POST /v3/citations`, the `counts` object in `POST /v3/cohort/manifest`,
+  and the MCP `build_cohort` / `get_licenses` / `get_citations` results.
+- A "Limits" section in the user guide: the per-request caps (SQL timeout and row ceiling,
+  manifest cap, page size, memory) and the fact that there is no per-caller rate limit or `429`.
+
 ### Fixed
 
+- A mis-shaped filter body no longer returns all of IDC at HTTP 200. Unrecognized keys in a
+  filter body (`{"term": …}`, a range bound misspelled `{"min": …}`) and malformed MCP filter
+  arguments are now errors rather than silently dropped predicates.
+- `POST /v3/citations` / `get_citations` resolve DOIs in batches via DataCite instead of one
+  request each — a cohort spanning every DOI in IDC took 237 serial round-trips, now 5. Any DOI
+  the batch doesn't cover still falls back to a per-DOI resolve, so citations stay complete.
 - The `/v3/viewer-url` OpenAPI examples (the values Swagger UI's "Try it out" pre-fills) used a StudyInstanceUID and SeriesInstanceUID that are not present in IDC, so running the example returned a `not_found` error instead of a viewer link. Both now use resolvable UIDs.
 
 ## [3.0.0b2] — 2026-07-14
diff --git a/docs/user-guide.md b/docs/user-guide.md
index 8fbca5a..d7e7637 100644
--- a/docs/user-guide.md
+++ b/docs/user-guide.md
@@ -114,6 +114,16 @@ from S3/GCS; see [§4](#4-getting-the-data)) — and **be a good citizen** — c
 > default and flags truncation, so a peek stays cheap; large unfiltered results mostly just waste
 > the agent's context.
 >
+> **Knowing your filter was applied.** Every filtered response echoes `filters_applied` — the
+> predicates the server actually used — alongside a `warnings` list. Empty `filters_applied`
+> means *nothing* was filtered and the counts cover the entire archive; `warnings` says so in
+> words. A mis-shaped filter body is a `422`, never a silently unfiltered `200`; a predicate that
+> can't constrain anything (an empty value list, a range with neither bound) is dropped and named
+> in `warnings`; and the surfaces that *enumerate* series (`cohort/manifest`,
+> `cohort/manifest.txt`, `build_cohort`, `get_cohort_urls`) refuse an unfiltered request outright.
+> Values match **case-sensitively**, and a cohort that matches nothing tells you when case is the
+> only reason — `Modality: ["mr"]` comes back with zero counts *and* a warning that `MR` exists.
+>
 > **Knowing you got it all.** Size-capped responses include a `truncated` boolean:
 > `truncated: false` means the result is complete; `true` means raise the limit and re-check (or
 > narrow/aggregate). `run_sql`'s `max_rows` is clamped to a server ceiling (`SQL_MAX_ROWS_CAP`),
@@ -239,6 +249,27 @@ uv run idc-api          # http://127.0.0.1:8000  — Swagger UI at /v3/docs
 | `POST /v3/citations` | Citations for a cohort |
 | `POST /v3/licenses` | License breakdown for a cohort |
 
+### Filter bodies: one shape
+
+**Every** filter-taking endpoint — `cohort/counts`, `cohort/manifest`, `cohort/manifest.txt`,
+`citations`, `licenses` — takes the filter object under `filters`, alongside that endpoint's own
+options:
+
+```json
+{"filters": {"terms": {"collection_id": ["nlst"]}, "ranges": {"instanceCount": {"gte": 100}}}}
+```
+
+Sending the filter bare (`{"terms": …}` at the top level) is a `422` naming the fix, and so is any
+unrecognized key inside a filter body (`{"term": …}`, a range bound misspelled `{"min": …}`) —
+an ignored key is a dropped predicate, and a dropped predicate silently widens the selection to
+the whole archive. Each response reports what was actually applied; see *Knowing your filter was
+applied* in [§1](#recommended-workflow).
+
+An empty filter is answered by the aggregate endpoints (`cohort/counts`, `licenses`) with a
+warning — "how big is IDC" is a legitimate question. The endpoints that *enumerate series*
+(`cohort/manifest`, `cohort/manifest.txt`) refuse it with a `400`: an unfiltered manifest is a
+download payload for 100+ TB.
+
 ### Worked examples
 
 Every endpoint below is also documented interactively at `/v3/docs` (Swagger UI), with a filled-in
@@ -266,15 +297,15 @@ curl -s localhost:8000/v3/clinical/tables/nlst_canc                     # clinic
 curl -s 'localhost:8000/v3/clinical/tables/nlst_canc/rows?max_rows=100' # clinical rows (capped)
 ```
 
-**Cheap size check** — the `counts` body is the filter object directly:
+**Cheap size check** — the filter goes under `filters`, as it does on every filter endpoint:
 
 ```bash
 curl -s localhost:8000/v3/cohort/counts \
   -H 'content-type: application/json' \
-  -d '{"terms": {"Modality": ["MR"], "BodyPartExamined": ["BREAST"]}}'
+  -d '{"filters": {"terms": {"Modality": ["MR"], "BodyPartExamined": ["BREAST"]}}}'
 ```
 
-**Build a cohort** — `manifest` wraps the filter in a request with paging:
+**Build a cohort** — `manifest` adds paging to the same filter body:
 
 ```bash
 curl -s localhost:8000/v3/cohort/manifest \
@@ -302,15 +333,15 @@ curl -s localhost:8000/v3/sql \
   -d '{"sql": "SELECT Modality, count(*) n FROM index GROUP BY 1 ORDER BY n DESC", "max_rows": 20}'
 ```
 
-**License check** — like `counts`, the body is the filter object directly:
+**License check:**
 
 ```bash
 curl -s localhost:8000/v3/licenses \
   -H 'content-type: application/json' \
-  -d '{"terms": {"collection_id": ["nlst"]}}'
+  -d '{"filters": {"terms": {"collection_id": ["nlst"]}}}'
 ```
 
-**Citations for a cohort** — body wraps the filter, like `manifest`:
+**Citations for a cohort:**
 
 ```bash
 curl -s localhost:8000/v3/citations \
@@ -447,6 +478,28 @@ Values interpolated into curated (non-SQL) queries are always passed as bound pa
 ([OWASP](https://cheatsheetseries.owasp.org/cheatsheets/SQL_Injection_Prevention_Cheat_Sheet.html)).
 See [`dev/api_v3_plan.md`](../dev/api_v3_plan.md) for the full threat model.
 
+### Limits
+
+The public deployment is unauthenticated and needs no API key, and there is **no per-caller rate
+limit or quota** — nothing to budget against, and no `429`. What *is* bounded is each individual
+request:
+
+| Limit | Default | Applies to |
+|---|---|---|
+| SQL statement timeout | 30 s | `POST /v3/sql`, `run_sql` |
+| SQL rows returned | 5 000, hard ceiling 10 000 | `POST /v3/sql`, `run_sql` (`truncated` flags it) |
+| Manifest series enumerated | 100 000 | `cohort/manifest*`, `get_cohort_urls` |
+| Page size | 5 000 | `cohort/manifest` |
+| Query memory | 4 GB | every query |
+| At least one filter predicate | required | the series-enumerating endpoints (see above) |
+
+All of these are configurable per deployment ([§7](#7-configuration)). Beyond them, a burst is
+absorbed by Cloud Run autoscaling and surfaces as slower responses or a `503` — back off and
+retry rather than treating it as a permanent failure. Please keep automated use reasonable; if
+you need sustained heavy access, query the [`idc-index`](https://github.com/ImagingDataCommons/idc-index)
+Parquet or IDC's BigQuery tables directly instead of driving this API hard, and note that data
+transfer never goes through this server anyway ([§4](#4-getting-the-data)).
+
 ---
 
 ## 6. Licenses & citations
diff --git a/src/idc_api/core/filters.py b/src/idc_api/core/filters.py
index 4209bd5..49c9650 100644
--- a/src/idc_api/core/filters.py
+++ b/src/idc_api/core/filters.py
@@ -3,21 +3,49 @@
 Attribute *names* are validated against a fixed allow-list (they can't be parameterized, so
 we whitelist + double-quote them); attribute *values* are always passed as bound parameters
 — the OWASP-recommended primary defense for the inputs we control.
+
+Compilation also reports what it *dropped*. A filter that compiles to no predicates selects
+every series in IDC, which is a plausible-looking answer rather than an error, so the compiler
+returns the predicates it actually applied plus caller-facing warnings and every response built
+from a filter echoes them (see ``CohortCounts.filters_applied`` / ``.warnings``).
 """
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, NamedTuple
 
 from . import schema
 from .errors import InvalidQueryError
-from .models import CohortFilters
+from .models import CohortFilters, NumericRange
+
+UNFILTERED_WARNING = (
+    "No filter predicates were applied, so this result describes the ENTIRE IDC archive, not a "
+    "cohort. If you meant to filter, the filter did not arrive in a usable shape — compare "
+    "`filters_applied` with what you sent, then re-send it."
+)
+
+
+class CompiledFilters(NamedTuple):
+    """The compiled filter, plus what the caller needs to see to trust it.
 
+    ``where`` is ``TRUE`` when nothing survived compilation — hence ``applied`` and ``warnings``,
+    which let a caller tell "no data matched" apart from "my filter was silently dropped".
+    """
 
-def compile_filters(filters: CohortFilters) -> tuple[str, list[Any]]:
-    """Return ``(where_sql, params)``. ``where_sql`` is ``TRUE`` when no filters are given."""
+    where: str
+    params: list[Any]
+    applied: CohortFilters  # only the predicates that made it into `where`
+    warnings: list[str]
+
+
+def compile_filters(filters: CohortFilters) -> CompiledFilters:
+    """Compile ``filters`` into a WHERE clause with bound params, the effective filter, and
+    warnings for anything dropped along the way."""
     clauses: list[str] = []
     params: list[Any] = []
+    applied_terms: dict[str, list[str]] = {}
+    applied_ranges: dict[str, NumericRange] = {}
+    warnings: list[str] = []
 
     for attr, values in (filters.terms or {}).items():
         if attr not in schema.TERM_ATTRIBUTES:
@@ -27,10 +55,15 @@ def compile_filters(filters: CohortFilters) -> tuple[str, list[Any]]:
             )
         values = [v for v in (values or []) if v is not None]
         if not values:
+            warnings.append(
+                f"Term filter {attr!r} was ignored because its value list is empty; it "
+                "constrains nothing."
+            )
             continue
         placeholders = ", ".join(["?"] * len(values))
         clauses.append(f'"{attr}" IN ({placeholders})')
         params.extend(values)
+        applied_terms[attr] = values
 
     for attr, rng in (filters.ranges or {}).items():
         if attr not in schema.RANGE_ATTRIBUTES:
@@ -38,12 +71,45 @@ def compile_filters(filters: CohortFilters) -> tuple[str, list[Any]]:
                 f"Unknown or non-range filter attribute: {attr!r}. "
                 "Use list_attributes to see valid attributes."
             )
+        if rng.gte is None and rng.lte is None:
+            warnings.append(
+                f"Range filter {attr!r} was ignored because neither 'gte' nor 'lte' was set; it "
+                "constrains nothing."
+            )
+            continue
         if rng.gte is not None:
             clauses.append(f'"{attr}" >= ?')
             params.append(rng.gte)
         if rng.lte is not None:
             clauses.append(f'"{attr}" <= ?')
             params.append(rng.lte)
+        applied_ranges[attr] = rng
+
+    if not clauses:
+        warnings.append(UNFILTERED_WARNING)
+
+    return CompiledFilters(
+        where=" AND ".join(clauses) if clauses else "TRUE",
+        params=params,
+        applied=CohortFilters(terms=applied_terms, ranges=applied_ranges),
+        warnings=warnings,
+    )
+
+
+def require_filter(compiled: CompiledFilters, action: str) -> None:
+    """Refuse ``action`` when no predicate survived compilation.
 
-    where = " AND ".join(clauses) if clauses else "TRUE"
-    return where, params
+    The aggregate surfaces (counts, licenses) answer an unfiltered filter honestly — "how big is
+    IDC" is a real question, and it costs one query. The surfaces that enumerate *per series* do
+    not: an unfiltered manifest is a download payload for the whole archive, which no caller
+    means to ask for. Those raise instead, with the dropped-predicate warnings attached so the
+    caller can see *why* their filter came out empty.
+    """
+    if compiled.applied.terms or compiled.applied.ranges:
+        return
+    dropped = [w for w in compiled.warnings if w != UNFILTERED_WARNING]
+    raise InvalidQueryError(
+        f"At least one filter predicate is required to {action}: unfiltered, that is every series "
+        "in IDC (100+ TB). Use the stats surface for archive-wide totals, or cohort counts to "
+        "size a filter first." + ("".join(f" {w}" for w in dropped))
+    )
diff --git a/src/idc_api/core/models.py b/src/idc_api/core/models.py
index 3717337..c7afba8 100644
--- a/src/idc_api/core/models.py
+++ b/src/idc_api/core/models.py
@@ -148,6 +148,10 @@ class ClinicalTableList(BaseModel):
 
 
 class NumericRange(BaseModel):
+    # Unknown keys are rejected rather than ignored: {"min": 5} would otherwise compile to a
+    # range that constrains nothing (see CohortFilters).
+    model_config = ConfigDict(extra="forbid")
+
     gte: float | str | None = None
     lte: float | str | None = None
 
@@ -158,6 +162,11 @@ class CohortFilters(BaseModel):
     names with ``list_attributes`` and valid values with ``get_attribute_values``."""
 
     model_config = ConfigDict(
+        # An unrecognized key is a hard error, never silently ignored. Ignoring one meant a
+        # mis-shaped body dropped every predicate and returned all of IDC at HTTP 200 — a wrong
+        # answer that looks like a right one. Failing loud beats forward-compatibility for a
+        # request object this small and this consequential.
+        extra="forbid",
         json_schema_extra={
             "examples": [
                 {
@@ -165,7 +174,7 @@ class CohortFilters(BaseModel):
                     "ranges": {},
                 }
             ]
-        }
+        },
     )
 
     terms: dict[str, list[str]] = Field(
@@ -175,12 +184,26 @@ class CohortFilters(BaseModel):
     ranges: dict[str, NumericRange] = Field(default_factory=dict)
 
 
+# Shared by every response built from a filter, so the same two field names mean the same thing
+# everywhere a cohort is described.
+_APPLIED_DESC = (
+    "The filter predicates the server actually applied. Compare it with what you sent: if it is "
+    "empty, nothing was filtered and these numbers cover all of IDC."
+)
+_WARNINGS_DESC = (
+    "Non-fatal problems with the request — above all, filter predicates that were dropped. "
+    "Empty when the request was honored in full."
+)
+
+
 class CohortCounts(BaseModel):
     patients: int
     studies: int
     series: int
     instances: int
     size_TB: float
+    filters_applied: CohortFilters = Field(default_factory=CohortFilters, description=_APPLIED_DESC)
+    warnings: list[str] = Field(default_factory=list, description=_WARNINGS_DESC)
 
 
 class SeriesManifestRow(BaseModel):
@@ -252,7 +275,11 @@ class CitationsResult(BaseModel):
         "In addition to the per-dataset citations, always acknowledge IDC itself by citing "
         "Fedorov et al., https://doi.org/10.1148/rg.230180 (see idc_acknowledgment)."
     )
+    filters_applied: CohortFilters = Field(default_factory=CohortFilters, description=_APPLIED_DESC)
+    warnings: list[str] = Field(default_factory=list, description=_WARNINGS_DESC)
 
 
 class LicensesResult(BaseModel):
     licenses: list[LicenseItem]
+    filters_applied: CohortFilters = Field(default_factory=CohortFilters, description=_APPLIED_DESC)
+    warnings: list[str] = Field(default_factory=list, description=_WARNINGS_DESC)
diff --git a/src/idc_api/core/services/citations.py b/src/idc_api/core/services/citations.py
index 22aced3..13e8d67 100644
--- a/src/idc_api/core/services/citations.py
+++ b/src/idc_api/core/services/citations.py
@@ -25,6 +25,18 @@
 
 _MAIN_IDC_DOI = "10.1148/rg.230180"
 
+# Batch resolution. Every IDC *dataset* DOI is DataCite-registered (TCIA 10.7937, Zenodo
+# 10.5281), and DataCite's list endpoint honours the same content negotiation as a single-DOI
+# resolve — so one request returns N formatted citations instead of N round-trips. This matters:
+# an unfiltered cohort spans every DOI in the archive (237 at IDC v24), which one-at-a-time meant
+# 237 serial network calls holding a worker for minutes.
+_DATACITE_DOIS_URL = "https://api.datacite.org/dois"
+# All 237 DOIs in a single OR-query URL gets an HTTP 414 from DataCite; 50 keeps the URL near
+# 2 KB and collapses the whole archive into five requests.
+_BATCH_CHUNK = 50
+# Turtle is excluded on purpose: concatenated RDF can't be split back into per-DOI entries.
+_BATCHABLE_FORMATS = {"apa", "bibtex", "csl-json"}
+
 
 class CitationsService:
     def __init__(self, backend: QueryBackend):
@@ -41,23 +53,99 @@ def get_citations(
             )
         accept = CITATION_FORMATS[fmt]
 
-        where, params = compile_filters(filters)
+        f = compile_filters(filters)
         dataset_dois = [
             r["source_DOI"]
             for r in self.backend.query(
-                # `where` is compile_filters output: allow-listed columns, values bound below.
-                f"SELECT DISTINCT source_DOI FROM index WHERE {where} "  # nosec B608
+                # `f.where` is compile_filters output: allow-listed columns, values bound below.
+                f"SELECT DISTINCT source_DOI FROM index WHERE {f.where} "  # nosec B608
                 f"AND source_DOI IS NOT NULL AND source_DOI <> ''",
-                params,
+                params=f.params,
             ).rows
         ]
 
-        citations = [c for doi in dataset_dois if (c := self._fetch(doi, accept, fmt, timeout))]
+        citations = self._resolve(dataset_dois, accept, fmt, timeout)
         # The IDC paper is kept separate from the dataset citations so callers can surface it as
         # the acknowledgment for IDC itself, alongside the recommendation on CitationsResult.
         idc_ack = self._fetch(_MAIN_IDC_DOI, accept, fmt, timeout)
 
-        return CitationsResult(format=fmt, citations=citations, idc_acknowledgment=idc_ack)
+        return CitationsResult(
+            format=fmt,
+            citations=citations,
+            idc_acknowledgment=idc_ack,
+            filters_applied=f.applied,
+            warnings=f.warnings,
+        )
+
+    def _resolve(self, dois: list[str], accept: str, fmt: str, timeout: float) -> list:
+        """Formatted citations for ``dois``, batched where possible.
+
+        Anything the batch didn't cover — a non-DataCite DOI, a chunk that failed — falls back to
+        the per-DOI resolve, so batching can only make this faster, never less complete.
+        """
+        batched = (
+            self._batch_fetch(dois, accept, fmt, timeout)
+            if fmt in _BATCHABLE_FORMATS and len(dois) > 1
+            else {}
+        )
+        out = []
+        for doi in dois:  # index order, so the same cohort always cites in the same order
+            hit = batched.get(doi.lower()) or self._fetch(doi, accept, fmt, timeout)
+            if hit:
+                out.append(hit)
+        return out
+
+    @classmethod
+    def _batch_fetch(cls, dois: list[str], accept: str, fmt: str, timeout: float) -> dict:
+        """Resolve ``dois`` in chunks against DataCite's list endpoint -> {lowercased doi: entry}.
+
+        Best-effort by construction: a chunk that errors or comes back unparseable is simply
+        absent from the result, and ``_resolve`` fills the gap one DOI at a time.
+        """
+        found: dict[str, object] = {}
+        for start in range(0, len(dois), _BATCH_CHUNK):
+            chunk = dois[start : start + _BATCH_CHUNK]
+            query = "doi:(" + " OR ".join(f'"{d}"' for d in chunk) + ")"
+            try:
+                resp = requests.get(
+                    _DATACITE_DOIS_URL,
+                    params={"query": query, "page[size]": len(chunk)},
+                    headers={"accept": accept},
+                    timeout=timeout,
+                )
+            except requests.RequestException:
+                continue
+            if resp.status_code == 200:
+                found.update(cls._split_batch(resp, chunk, fmt))
+        return found
+
+    @staticmethod
+    def _split_batch(resp, chunk: list[str], fmt: str) -> dict:
+        """Map a batch response back onto the DOIs that asked for it.
+
+        DataCite doesn't preserve request order, so entries are matched by the DOI each one
+        carries: a `DOI` field for csl-json, and the doi.org URL (APA) or `@misc` key (BibTeX)
+        embedded in the text formats, which arrive as one blank-line-separated blob.
+        """
+        if fmt == "csl-json":
+            try:
+                items = resp.json()
+            except ValueError:
+                return {}
+            if not isinstance(items, list):
+                return {}
+            return {str(i["DOI"]).lower(): i for i in items if isinstance(i, dict) and i.get("DOI")}
+
+        found: dict[str, object] = {}
+        for entry in (e.strip() for e in resp.text.split("\n\n")):
+            if not entry:
+                continue
+            lowered = entry.lower()
+            for doi in chunk:
+                if doi.lower() in lowered:
+                    found[doi.lower()] = entry
+                    break
+        return found
 
     @staticmethod
     def _fetch(doi: str, accept: str, fmt: str, timeout: float):
diff --git a/src/idc_api/core/services/cohort.py b/src/idc_api/core/services/cohort.py
index df4be14..396f6a3 100644
--- a/src/idc_api/core/services/cohort.py
+++ b/src/idc_api/core/services/cohort.py
@@ -4,7 +4,7 @@
 from __future__ import annotations
 
 from ..backend.base import QueryBackend
-from ..filters import compile_filters
+from ..filters import compile_filters, require_filter
 from ..models import (
     CohortCounts,
     CohortFilters,
@@ -37,24 +37,56 @@ def __init__(self, backend: QueryBackend, settings):
         self.manifest = ManifestService(backend, settings)
 
     def counts(self, filters: CohortFilters) -> CohortCounts:
-        where, params = compile_filters(filters)
-        # `where` is compile_filters output: allow-listed columns, values bound below.
+        f = compile_filters(filters)
+        # `f.where` is compile_filters output: allow-listed columns, values bound below.
         row = self.backend.query(
             f"SELECT count(DISTINCT PatientID) patients, "  # nosec B608
             f"count(DISTINCT StudyInstanceUID) studies, "
             f"count(DISTINCT SeriesInstanceUID) series, "
             f"COALESCE(sum(instanceCount),0) instances, "
-            f"COALESCE(sum(series_size_MB),0) size_mb FROM index WHERE {where}",
-            params,
+            f"COALESCE(sum(series_size_MB),0) size_mb FROM index WHERE {f.where}",
+            params=f.params,
         ).rows[0]
+        warnings = list(f.warnings)
+        if row["series"] == 0:
+            warnings.extend(self._casing_hints(f.applied))
         return CohortCounts(
             patients=row["patients"],
             studies=row["studies"],
             series=row["series"],
             instances=int(row["instances"]),
             size_TB=round(row["size_mb"] / _MB_PER_TB, 3),
+            # Echo the effective filter so a caller can tell an empty cohort apart from a
+            # dropped filter without guessing from the magnitude of the numbers.
+            filters_applied=f.applied,
+            warnings=warnings,
         )
 
+    def _casing_hints(self, applied: CohortFilters) -> list[str]:
+        """Explain a zero-row cohort when the only thing wrong was letter case.
+
+        Values are matched exactly, so `Modality=['mr']` counts zero the same way a genuinely
+        empty cohort does. Only reached when nothing matched, so the extra probe — one query per
+        term attribute — never costs anything on a query that worked.
+        """
+        hints: list[str] = []
+        for attr, values in applied.terms.items():
+            # `attr` came through compile_filters, so it is already allow-listed; quote it the
+            # same way and bind the values.
+            placeholders = ", ".join(["?"] * len(values))
+            rows = self.backend.query(
+                f'SELECT DISTINCT "{attr}" v FROM index '  # nosec B608
+                f'WHERE lower(CAST("{attr}" AS VARCHAR)) IN ({placeholders}) LIMIT 5',
+                params=[v.lower() for v in values],
+            ).rows
+            actual = sorted({r["v"] for r in rows if r["v"] is not None and r["v"] not in values})
+            if actual:
+                hints.append(
+                    f"No series matched {attr}={values}, but {actual} exists — values are matched "
+                    "case-sensitively. Use the attribute-values surface to get the exact casing."
+                )
+        return hints
+
     def build_manifest(
         self,
         filters: CohortFilters,
@@ -66,19 +98,22 @@ def build_manifest(
         page_size = page_size if page_size is not None else self.settings.default_page_size
         page_size = max(1, min(int(page_size), self.settings.max_page_size))
 
+        f = compile_filters(filters)
+        # Checked before the counting scan, not after: a manifest of the whole archive is never
+        # what a caller meant, so refuse it rather than pay for it.
+        require_filter(f, "build a manifest")
         counts = self.counts(filters)
-        where, params = compile_filters(filters)
 
         series: list[SeriesManifestRow] = []
         if include_rows:
             cols = ", ".join(f'"{c}"' for c in _ROW_COLUMNS)
-            # `cols` is a fixed constant list (_ROW_COLUMNS); `where` is compile_filters output
+            # `cols` is a fixed constant list (_ROW_COLUMNS); `f.where` is compile_filters output
             # (allow-listed columns, values bound below); page/page_size are clamped ints.
             rows = self.backend.query(
-                f"SELECT {cols} FROM index WHERE {where} "  # nosec B608
+                f"SELECT {cols} FROM index WHERE {f.where} "  # nosec B608
                 f"ORDER BY collection_id, PatientID, StudyInstanceUID, SeriesInstanceUID "
                 f"LIMIT {page_size} OFFSET {page * page_size}",
-                params,
+                params=f.params,
             ).rows
             series = [SeriesManifestRow(**r) for r in rows]
 
diff --git a/src/idc_api/core/services/licenses.py b/src/idc_api/core/services/licenses.py
index e874d10..c6515eb 100644
--- a/src/idc_api/core/services/licenses.py
+++ b/src/idc_api/core/services/licenses.py
@@ -14,13 +14,13 @@ def __init__(self, backend: QueryBackend):
         self.backend = backend
 
     def get_licenses(self, filters: CohortFilters) -> LicensesResult:
-        where, params = compile_filters(filters)
-        # `where` is compile_filters output: allow-listed columns, values bound below.
+        f = compile_filters(filters)
+        # `f.where` is compile_filters output: allow-listed columns, values bound below.
         rows = self.backend.query(
             f"SELECT license_short_name, count(DISTINCT SeriesInstanceUID) series, "  # nosec B608
-            f"COALESCE(sum(series_size_MB),0) size_mb FROM index WHERE {where} "
+            f"COALESCE(sum(series_size_MB),0) size_mb FROM index WHERE {f.where} "
             f"GROUP BY 1 ORDER BY series DESC",
-            params,
+            params=f.params,
         ).rows
         return LicensesResult(
             licenses=[
@@ -30,5 +30,7 @@ def get_licenses(self, filters: CohortFilters) -> LicensesResult:
                     size_TB=round(r["size_mb"] / _MB_PER_TB, 3),
                 )
                 for r in rows
-            ]
+            ],
+            filters_applied=f.applied,
+            warnings=f.warnings,
         )
diff --git a/src/idc_api/core/services/manifest.py b/src/idc_api/core/services/manifest.py
index 61fd587..afaa96e 100644
--- a/src/idc_api/core/services/manifest.py
+++ b/src/idc_api/core/services/manifest.py
@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 from ..backend.base import QueryBackend
-from ..filters import compile_filters
+from ..filters import compile_filters, require_filter
 from ..models import CohortFilters, DownloadInfo
 
 # GCS is reached via its S3-compatible interop endpoint, so URLs keep the s3:// scheme even
@@ -52,8 +52,9 @@ def manifest_lines(
         if source not in ("aws", "gcs"):
             raise ValueError("source must be 'aws' or 'gcs'")
         limit = limit if limit is not None else self.settings.manifest_hard_cap
-        where, params = compile_filters(filters)
-        urls, truncated = self._series_urls(where, params, limit)
+        f = compile_filters(filters)
+        require_filter(f, "list series URLs")
+        urls, truncated = self._series_urls(f.where, f.params, limit)
         if source == "gcs":
             urls = [remap_bucket_for_gcs(u) for u in urls]
         return urls, truncated
@@ -67,8 +68,9 @@ def manifest_text(
     def download_info(
         self, filters: CohortFilters, total_series: int, size_TB: float
     ) -> DownloadInfo:
-        where, params = compile_filters(filters)
-        preview, _ = self._series_urls(where, params, 5)
+        f = compile_filters(filters)
+        require_filter(f, "build a download payload")
+        preview, _ = self._series_urls(f.where, f.params, 5)
         truncated = total_series > self.settings.manifest_hard_cap
 
         commands: list[str] = []
diff --git a/src/idc_api/mcp/server.py b/src/idc_api/mcp/server.py
index e5cffde..a5db696 100644
--- a/src/idc_api/mcp/server.py
+++ b/src/idc_api/mcp/server.py
@@ -28,13 +28,14 @@
 from mcp.server.fastmcp import FastMCP
 from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.transport_security import TransportSecuritySettings
+from pydantic import ValidationError
 from starlette.applications import Starlette
 from starlette.routing import Route
 
 from ..core import schema as core_schema
 from ..core import version as core_version
 from ..core.context import AppContext
-from ..core.errors import IDCAPIError
+from ..core.errors import IDCAPIError, InvalidQueryError
 from ..core.models import CohortFilters, NumericRange
 from ..http_headers import HSTSMiddleware
 from ..settings import get_settings
@@ -61,6 +62,8 @@
 3. IDC is large (100+ TB) — always report counts/size_TB and warn before any download. To get
    data, use get_cohort_urls / the returned `idc` commands — direct S3/GCS transfer from public
    buckets, no server involved.
+4. Trust `filters_applied`, not your intent: surface any `warnings` a result carries, and treat
+   an empty `filters_applied` as the whole archive rather than a cohort.
 Cite with get_citations (per-dataset citations plus the IDC paper to acknowledge IDC itself);
 respect get_licenses (CC-BY vs CC-BY-NC). See `idc://guide` for the data model, the full tool
 list, and join examples."""
@@ -204,10 +207,30 @@ def wrapper(*args, **kwargs):
 
 
 def _filters(terms: dict | None, ranges: dict | None) -> CohortFilters:
-    return CohortFilters(
-        terms=terms or {},
-        ranges={k: NumericRange(**v) for k, v in (ranges or {}).items()},
-    )
+    """Build the core filter object from loose tool arguments.
+
+    The models reject unknown keys (a silently-dropped predicate would return all of IDC as if
+    it were a cohort), so a mis-shaped argument raises here. Convert it to an InvalidQueryError
+    — `guard` turns that into a ToolError stating the expected shape, rather than the opaque
+    "Internal error" any other exception would produce.
+    """
+    try:
+        return CohortFilters(
+            terms=terms or {},
+            ranges={k: NumericRange(**v) for k, v in (ranges or {}).items()},
+        )
+    except ValidationError as exc:
+        raise InvalidQueryError(
+            "Malformed filter arguments: terms is {attribute: [values]} (e.g. "
+            '{"Modality": ["MR"]}) and ranges is {attribute: {"gte": x, "lte": y}}. '
+            f"Details: {exc.error_count()} validation error(s): "
+            + "; ".join(f"{'.'.join(str(p) for p in e['loc'])}: {e['msg']}" for e in exc.errors())
+        ) from None
+    except TypeError as exc:  # e.g. ranges={"instanceCount": 5} — not a mapping to unpack
+        raise InvalidQueryError(
+            'Malformed filter arguments: ranges is {attribute: {"gte": x, "lte": y}} and terms '
+            f"is {{attribute: [values]}}. Details: {exc}"
+        ) from None
 
 
 # --- discovery ----------------------------------------------------------------------------
@@ -357,7 +380,11 @@ def build_cohort(
     `terms` is {attribute: [values]} for equality/IN (e.g. {"Modality": ["MR"],
     "BodyPartExamined": ["BREAST"]}). `ranges` is {attribute: {"gte": x, "lte": y}} for
     numeric/date ranges. Discover valid attributes with list_attributes and valid values with
-    get_attribute_values. For anything these structured filters can't express, use run_sql."""
+    get_attribute_values. For anything these structured filters can't express, use run_sql.
+
+    At least one filter predicate is required — an unfiltered cohort is the whole 100+ TB archive;
+    use get_stats for archive-wide totals. The result echoes `counts.filters_applied` and
+    `counts.warnings`: check them rather than assuming your filter landed."""
     f = _filters(terms, ranges)
     return ctx.cohort.build_manifest(f, page=page, page_size=page_size).model_dump(mode="json")
 
@@ -403,7 +430,8 @@ def get_cohort_urls(
     ever expects s3:// lines). Returns up to `limit` URLs (increase for full manifests). These
     are anonymous public URLs — easiest is the `idc` CLI (handles either cloud); driving it
     yourself, `s5cmd --no-sign-request` works directly for source=aws, and for source=gcs add
-    `--endpoint-url https://storage.googleapis.com`."""
+    `--endpoint-url https://storage.googleapis.com`. At least one filter predicate is required:
+    unfiltered, this would enumerate every series in IDC."""
     f = _filters(terms, ranges)
     urls, truncated = ctx.manifest.manifest_lines(f, source=source, limit=limit)
     return {
@@ -496,6 +524,14 @@ def get_licenses(terms: dict | None = None, ranges: dict | None = None) -> dict:
    property you need is not there (e.g. what anatomy a segmentation contains), it likely lives
    in a specialized index — see *Tables for run_sql* below.
 3. *Build:* `build_cohort(terms={...}, ranges={...})` → counts, sample series, download payload.
+   *How to tell your filter was applied:* every filtered result echoes `filters_applied` (the
+   predicates actually used) plus a `warnings` list. Empty `filters_applied` means nothing was
+   filtered and the counts describe the ENTIRE archive — never report that as a cohort. A
+   predicate that constrains nothing (empty value list, range with neither bound) is dropped and
+   named in `warnings`; a malformed filter argument is an error, never an empty filter; and
+   `build_cohort` / `get_cohort_urls` refuse an unfiltered request outright (use `get_stats` for
+   archive-wide totals). Values are case-sensitive, and a zero-count cohort says so when case is
+   the only reason — `Modality=['mr']` returns zeros *plus* a warning that 'MR' exists.
    For complex queries: `list_tables` → `get_table_schema('index')` → `run_sql('SELECT ...')`.
    *Explore narrow, then widen:* keep result sizes small while you're still figuring out the
    query (small `max_rows` / `limit` / `page_size`, or COUNT/GROUP BY instead of raw rows), and
diff --git a/src/idc_api/rest/app.py b/src/idc_api/rest/app.py
index ef59a23..5a41e4c 100644
--- a/src/idc_api/rest/app.py
+++ b/src/idc_api/rest/app.py
@@ -11,11 +11,12 @@
 import logging
 import time
 from contextlib import asynccontextmanager
+from typing import Any
 
 from fastapi import FastAPI, Path, Query, Request
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse, PlainTextResponse, RedirectResponse
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, model_validator
 
 from ..core.context import AppContext, get_context
 from ..core.errors import IDCAPIError
@@ -60,7 +61,133 @@ def _format_sql(sql: str, settings) -> str:
 # --- request bodies (response models are the shared core models) --------------------------
 
 
-class ManifestRequest(BaseModel):
+class _FilterRequest(BaseModel):
+    """Base for every body that carries a cohort filter.
+
+    **One shape for all of them**: the filter object always lives under ``filters``, so nothing
+    depends on remembering which endpoint takes it bare. It used to vary — counts/licenses bare,
+    manifest/citations wrapped — and sending one shape where the other was expected validated
+    cleanly, dropped every predicate, and answered with all of IDC at HTTP 200. Now the shape is
+    uniform and both mistakes are loud: ``extra="forbid"`` rejects stray keys generically, and
+    the check below names the fix for the one callers actually make.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    filters: CohortFilters = Field(default_factory=CohortFilters)
+
+    @model_validator(mode="before")
+    @classmethod
+    def _reject_bare_filter(cls, data: Any) -> Any:
+        if isinstance(data, dict):
+            stray = [k for k in ("terms", "ranges") if k in data]
+            if stray:
+                raise ValueError(
+                    "the filter object goes under `filters`, e.g. "
+                    '{"filters": {"terms": {"collection_id": ["nlst"]}}} — got top-level '
+                    f"{', '.join(repr(k) for k in stray)}"
+                )
+        return data
+
+
+def _filter_errors(*, requires_filter: bool = False) -> dict:
+    """OpenAPI error documentation shared by the filter-taking endpoints.
+
+    How a filter *fails* is part of this API's contract — it is refused, never silently widened
+    to the whole archive — so Swagger UI and any agent reading the schema see the failures spelled
+    out, not just the happy path.
+    """
+    bad_request = {
+        "unknown attribute": {
+            "summary": "Attribute is not filterable (check GET /v3/attributes)",
+            "value": {
+                "error": {
+                    "code": "invalid_query",
+                    "message": "Unknown or non-term filter attribute: 'Modaliti'. Use "
+                    "list_attributes to see valid attributes.",
+                }
+            },
+        }
+    }
+    if requires_filter:
+        bad_request["no filter predicate"] = {
+            "summary": "Refused: unfiltered, this would enumerate every series in IDC",
+            "value": {
+                "error": {
+                    "code": "invalid_query",
+                    "message": "At least one filter predicate is required to build a manifest: "
+                    "unfiltered, that is every series in IDC (100+ TB). Use the stats surface for "
+                    "archive-wide totals, or cohort counts to size a filter first.",
+                }
+            },
+        }
+    return {
+        400: {
+            "description": "Filter could not be applied as given."
+            + (" Includes an unfiltered request, which is refused." if requires_filter else ""),
+            "content": {"application/json": {"examples": bad_request}},
+        },
+        422: {
+            "description": "Malformed body. Unrecognized keys are rejected rather than ignored, "
+            "because an ignored key is a dropped predicate — which would silently widen the "
+            "selection to all of IDC.",
+            "content": {
+                "application/json": {
+                    "examples": {
+                        "filter not under `filters`": {
+                            "summary": "The filter object goes under `filters`",
+                            "value": {
+                                "detail": [
+                                    {
+                                        "type": "value_error",
+                                        "loc": ["body"],
+                                        "msg": "Value error, the filter object goes under "
+                                        '`filters`, e.g. {"filters": {"terms": {"collection_id": '
+                                        "[\"nlst\"]}}} — got top-level 'terms'",
+                                    }
+                                ]
+                            },
+                        },
+                        "misspelled key": {
+                            "summary": "`min` is not a range bound (`gte` / `lte` are)",
+                            "value": {
+                                "detail": [
+                                    {
+                                        "type": "extra_forbidden",
+                                        "loc": [
+                                            "body",
+                                            "filters",
+                                            "ranges",
+                                            "instanceCount",
+                                            "min",
+                                        ],
+                                        "msg": "Extra inputs are not permitted",
+                                    }
+                                ]
+                            },
+                        },
+                    }
+                }
+            },
+        },
+    }
+
+
+class CountsRequest(_FilterRequest):
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [{"filters": {"terms": {"collection_id": ["nlst"], "Modality": ["CT"]}}}]
+        }
+    )
+
+
+class LicensesRequest(_FilterRequest):
+    model_config = ConfigDict(
+        json_schema_extra={"examples": [{"filters": {"terms": {"collection_id": ["nlst"]}}}]}
+    )
+
+
+class ManifestRequest(_FilterRequest):
     model_config = ConfigDict(
         json_schema_extra={
             "examples": [
@@ -74,13 +201,12 @@ class ManifestRequest(BaseModel):
         }
     )
 
-    filters: CohortFilters = Field(default_factory=CohortFilters)
     page: int = 0
     page_size: int | None = None
     include_rows: bool = True
 
 
-class ManifestTextRequest(BaseModel):
+class ManifestTextRequest(_FilterRequest):
     model_config = ConfigDict(
         json_schema_extra={
             "examples": [
@@ -89,7 +215,6 @@ class ManifestTextRequest(BaseModel):
         }
     )
 
-    filters: CohortFilters = Field(default_factory=CohortFilters)
     source: str = "aws"
     limit: int | None = None
 
@@ -110,7 +235,7 @@ class SqlRequest(BaseModel):
     max_rows: int | None = None
 
 
-class CitationsRequest(BaseModel):
+class CitationsRequest(_FilterRequest):
     model_config = ConfigDict(
         json_schema_extra={
             "examples": [
@@ -119,7 +244,6 @@ class CitationsRequest(BaseModel):
         }
     )
 
-    filters: CohortFilters = Field(default_factory=CohortFilters)
     citation_format: str = "apa"
 
 
@@ -447,20 +571,27 @@ def clinical_table_rows(
         response_model=CohortCounts,
         tags=["cohort"],
         summary="Cohort counts",
+        responses=_filter_errors(),
     )
-    def cohort_counts(filters: CohortFilters):
+    def cohort_counts(req: CountsRequest):
         """Return distinct counts for a filtered cohort — patients, studies, series, instances,
         and total `size_TB` — without the sample rows or download payload. Use it as a fast size
-        check before building a full manifest or downloading. `terms` is `{attribute: [values]}`
-        for equality/IN; `ranges` is `{attribute: {"gte": x, "lte": y}}` for numeric or date
-        ranges."""
-        return C().cohort.counts(filters)
+        check before building a full manifest or downloading. `filters.terms` is
+        `{attribute: [values]}` for equality/IN; `filters.ranges` is
+        `{attribute: {"gte": x, "lte": y}}` for numeric or date ranges.
+
+        The response echoes `filters_applied` (the predicates actually used) and `warnings`.
+        **Check them rather than assuming your filter landed**: an empty `filters_applied` means
+        nothing was filtered and these counts describe all of IDC. An empty filter is allowed
+        here — it is how you ask how big the archive is — and reported in `warnings`."""
+        return C().cohort.counts(req.filters)
 
     @app.post(
         f"{API_PREFIX}/cohort/manifest",
         response_model=ManifestResponse,
         tags=["cohort"],
         summary="Build cohort manifest",
+        responses=_filter_errors(requires_filter=True),
     )
     def cohort_manifest(req: ManifestRequest):
         """Build a cohort from structured filters and get back distinct counts (patients,
@@ -469,7 +600,11 @@ def cohort_manifest(req: ManifestRequest):
         equality/IN (e.g. `{"Modality": ["MR"]}`); `filters.ranges` is
         `{attribute: {"gte": x, "lte": y}}` for numeric or date ranges. Discover valid attributes
         via `/attributes` and valid values via `/attributes/{attribute}/values`. For anything
-        these structured filters can't express, use `/sql`."""
+        these structured filters can't express, use `/sql`.
+
+        **At least one filter predicate is required** — unfiltered, this is a download payload for
+        the entire archive; use `/stats` for archive-wide totals. `counts.filters_applied` echoes
+        the predicates actually used."""
         return C().cohort.build_manifest(
             req.filters, page=req.page, page_size=req.page_size, include_rows=req.include_rows
         )
@@ -478,6 +613,7 @@ def cohort_manifest(req: ManifestRequest):
         f"{API_PREFIX}/cohort/manifest.txt",
         tags=["cohort"],
         summary="Cohort manifest (plain text)",
+        responses=_filter_errors(requires_filter=True),
     )
     def cohort_manifest_text(req: ManifestTextRequest):
         """Return a plain-text manifest of public download URLs (one `s3://` per series,
@@ -486,7 +622,11 @@ def cohort_manifest_text(req: ManifestTextRequest):
         reached via its S3-compatible endpoint, matching idc-index). These are anonymous public
         URLs — feed the file to the `idc` CLI, or `s5cmd --no-sign-request` directly for
         `source=aws` (add `--endpoint-url https://storage.googleapis.com` for `source=gcs`). The
-        response is `text/plain`, one URL per line."""
+        response is `text/plain`, one URL per line.
+
+        **At least one filter predicate is required**: unfiltered, this enumerates every series in
+        IDC. Being plain text, this response carries no `filters_applied` echo — use
+        `/cohort/counts` first to confirm the filter and the size."""
         text = C().manifest.manifest_text(req.filters, source=req.source, limit=req.limit)
         return PlainTextResponse(text)
 
@@ -544,13 +684,16 @@ def viewer_url(
         response_model=CitationsResult,
         tags=["tools"],
         summary="Cohort citations",
+        responses=_filter_errors(),
     )
     def citations(req: CitationsRequest):
         """Return the publications to cite for a cohort: per-dataset citations (from the cohort's
         source DOIs) in `citations`, plus the IDC paper in `idc_acknowledgment`.
         `citation_format` is one of `apa`, `bibtex`, `csl-json`, `turtle`. When publishing
         results that use IDC data, include the per-dataset citations and acknowledge IDC itself
-        (see the `recommendation` field)."""
+        (see the `recommendation` field). `filters_applied` echoes the cohort these citations are
+        for. Broad cohorts span many DOIs and take proportionally longer to resolve, so filter to
+        the data you actually used."""
         return C().citations.get_citations(req.filters, citation_format=req.citation_format)
 
     @app.post(
@@ -558,12 +701,14 @@ def citations(req: CitationsRequest):
         response_model=LicensesResult,
         tags=["tools"],
         summary="Cohort license breakdown",
+        responses=_filter_errors(),
     )
-    def licenses(filters: CohortFilters):
+    def licenses(req: LicensesRequest):
         """Return the license breakdown (series count and size per license) for a cohort. Use it
         to check whether the data is commercial-friendly (CC BY) or non-commercial only
-        (CC BY-NC) before reuse."""
-        return C().licenses.get_licenses(filters)
+        (CC BY-NC) before reuse. `filters_applied` echoes the predicates actually used — an empty
+        one means this is the license breakdown of all of IDC, not of your cohort."""
+        return C().licenses.get_licenses(req.filters)
 
     return app
 
diff --git a/tests/test_citations.py b/tests/test_citations.py
new file mode 100644
index 0000000..0609ee1
--- /dev/null
+++ b/tests/test_citations.py
@@ -0,0 +1,125 @@
+"""Citation resolution: batched where possible, complete regardless.
+
+An unfiltered cohort spans every DOI in IDC (237 at v24). One content-negotiation request each
+meant 237 serial round-trips holding a worker for minutes, so DataCite's list endpoint — which
+honours the same content negotiation and covers every IDC dataset DOI (TCIA 10.7937, Zenodo
+10.5281) — resolves them in chunks instead.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+import idc_api.core.services.citations as cite_mod
+from idc_api.core.models import CohortFilters
+from idc_api.core.services.citations import CitationsService
+
+_DOIS = [f"10.7937/fake-{i}" for i in range(3)]
+
+
+class _Recorder:
+    """Stand in for `requests.get`, recording batch vs per-DOI calls."""
+
+    def __init__(self, *, batch_text=None, batch_status=200, per_doi_text="PER-DOI"):
+        self.batch_text = batch_text
+        self.batch_status = batch_status
+        self.per_doi_text = per_doi_text
+        self.batch_calls: list[list[str]] = []
+        self.per_doi_calls: list[str] = []
+
+    def __call__(self, url, headers=None, timeout=None, params=None):
+        recorder = self
+
+        class _Resp:
+            def __init__(self, status, text):
+                self.status_code = status
+                self.text = text
+
+            def json(self):
+                return json.loads(self.text)
+
+        if params and "query" in params:  # the DataCite batch endpoint
+            recorder.batch_calls.append(params["query"])
+            return _Resp(recorder.batch_status, recorder.batch_text or "")
+        recorder.per_doi_calls.append(url)
+        return _Resp(200, recorder.per_doi_text)
+
+
+@pytest.fixture
+def svc(monkeypatch):
+    def _make(recorder):
+        monkeypatch.setattr(cite_mod, "requests", type("R", (), {"get": staticmethod(recorder)}))
+        monkeypatch.setattr(cite_mod.requests, "RequestException", Exception, raising=False)
+        return CitationsService(backend=None)
+
+    return _make
+
+
+def test_many_dois_resolve_in_one_request(svc):
+    """Three DOIs, one call. Entries come back out of order, so each is matched by the DOI it
+    carries rather than by position."""
+    blob = "\n\n".join(
+        f"Author, A. (2024). Title {i}. https://doi.org/{d.upper()}"
+        for i, d in reversed(list(enumerate(_DOIS)))
+    )
+    rec = _Recorder(batch_text=blob)
+    out = CitationsService._resolve(svc(rec), _DOIS, "text/x-bibliography", "apa", 30.0)
+
+    assert len(rec.batch_calls) == 1
+    assert rec.per_doi_calls == []  # nothing fell back
+    # Returned in index order, not DataCite's order.
+    assert [c.split("Title ")[1][0] for c in out] == ["0", "1", "2"]
+
+
+def test_dois_the_batch_missed_fall_back_individually(svc):
+    """Batching may only make this faster, never less complete — a DOI DataCite doesn't know
+    (e.g. a Crossref one) still gets resolved."""
+    blob = f"Author, A. (2024). Title. https://doi.org/{_DOIS[0]}"
+    rec = _Recorder(batch_text=blob)
+    out = CitationsService._resolve(svc(rec), _DOIS, "text/x-bibliography", "apa", 30.0)
+
+    assert len(rec.batch_calls) == 1
+    assert len(rec.per_doi_calls) == 2  # only the two the batch didn't cover
+    assert len(out) == 3
+
+
+def test_batch_failure_degrades_to_per_doi(svc):
+    rec = _Recorder(batch_status=503)
+    out = CitationsService._resolve(svc(rec), _DOIS, "text/x-bibliography", "apa", 30.0)
+    assert len(rec.per_doi_calls) == 3 and out == ["PER-DOI"] * 3
+
+
+def test_csl_json_batch_maps_by_doi_field(svc):
+    rec = _Recorder(batch_text=json.dumps([{"DOI": d.upper(), "title": d} for d in _DOIS]))
+    out = CitationsService._resolve(svc(rec), _DOIS, "application/json", "csl-json", 30.0)
+    assert rec.per_doi_calls == []
+    assert [c["title"] for c in out] == _DOIS
+
+
+def test_turtle_is_not_batched(svc):
+    """Concatenated RDF can't be split back into per-DOI entries, so it keeps the safe path."""
+    rec = _Recorder(batch_text="ignored")
+    CitationsService._resolve(svc(rec), _DOIS, "text/turtle", "turtle", 30.0)
+    assert rec.batch_calls == [] and len(rec.per_doi_calls) == 3
+
+
+def test_chunking_keeps_the_url_short(svc):
+    """237 DOIs in a single OR-query URL gets an HTTP 414 from DataCite, so chunk them."""
+    many = [f"10.7937/fake-{i}" for i in range(120)]
+    rec = _Recorder(batch_text="")
+    CitationsService._resolve(svc(rec), many, "text/x-bibliography", "apa", 30.0)
+    assert len(rec.batch_calls) == 3  # 120 / 50, rounded up
+    assert all(len(q) < 2500 for q in rec.batch_calls)
+
+
+def test_citations_echo_the_filter_applied(ctx, monkeypatch):
+    """The filter echo reaches citations too, so a caller can see which cohort was cited."""
+    monkeypatch.setattr(
+        cite_mod, "requests", type("R", (), {"get": staticmethod(_Recorder(batch_text=""))})
+    )
+    monkeypatch.setattr(cite_mod.requests, "RequestException", Exception, raising=False)
+    terms = {"collection_id": ["rider_pilot"]}
+    out = ctx.citations.get_citations(CohortFilters(terms=terms))
+    assert out.filters_applied.terms == terms and out.warnings == []
diff --git a/tests/test_filter_shape.py b/tests/test_filter_shape.py
new file mode 100644
index 0000000..116f7ec
--- /dev/null
+++ b/tests/test_filter_shape.py
@@ -0,0 +1,203 @@
+"""A dropped filter must never look like an answer.
+
+Filter bodies used to come in two shapes — bare for counts/licenses, wrapped for
+manifest/citations — and sending one where the other was expected validated cleanly, dropped
+every predicate, and returned the whole archive at HTTP 200, indistinguishable from a
+legitimately huge cohort. These tests pin the three parts of the fix: one uniform shape whose
+violations are hard errors, a filter that compiles to nothing reported as such, and a flat
+refusal to enumerate series without a predicate.
+"""
+
+from __future__ import annotations
+
+import pytest
+from mcp.server.fastmcp.exceptions import ToolError
+
+from idc_api.core.filters import compile_filters, require_filter
+from idc_api.core.models import CohortFilters
+from idc_api.mcp.server import mcp
+
+_TERMS = {"collection_id": ["rider_pilot"]}
+# Below the real archive (>1M series) but far above any single test collection: the number a
+# dropped filter would produce.
+_ALL_OF_IDC = 1_000_000
+
+# Every filter-taking body, in the one shape they all now share.
+_WRAPPED = ("/v3/cohort/counts", "/v3/licenses", "/v3/cohort/manifest", "/v3/citations")
+
+
+# --- one shape, and violations are hard errors ----------------------------------------------
+
+
+def test_every_filter_endpoint_takes_the_same_shape(client):
+    for path in _WRAPPED:
+        r = client.post(path, json={"filters": {"terms": _TERMS}})
+        assert r.status_code == 200, (path, r.text[:200])
+
+
+def test_bare_filter_body_is_rejected_everywhere(client):
+    """The mistake that used to return all of IDC at 200 is now a 422 naming the fix."""
+    for path in (*_WRAPPED, "/v3/cohort/manifest.txt"):
+        r = client.post(path, json={"terms": _TERMS})
+        assert r.status_code == 422, (path, r.text[:200])
+        assert "goes under `filters`" in r.text, path
+
+
+def test_misspelled_filter_keys_are_rejected(client):
+    """An unrecognized key is refused rather than ignored — each of these would otherwise have
+    compiled to no predicate at all."""
+    for body in (
+        {"filters": {"term": _TERMS}},
+        {"filters": {"terms": _TERMS}, "junk": 1},
+        {"filters": {"terms": _TERMS, "ranges": {}, "extra": 1}},
+        {"filters": {"ranges": {"instanceCount": {"min": 5}}}},
+    ):
+        assert client.post("/v3/cohort/counts", json=body).status_code == 422, body
+
+
+# --- what survived compilation is always reported ------------------------------------------
+
+
+def test_counts_echo_the_filter_they_applied(client):
+    body = {"filters": {"terms": _TERMS, "ranges": {"instanceCount": {"gte": 2}}}}
+    r = client.post("/v3/cohort/counts", json=body).json()
+    assert r["series"] > 0
+    assert r["filters_applied"]["terms"] == _TERMS
+    assert r["filters_applied"]["ranges"] == {"instanceCount": {"gte": 2.0, "lte": None}}
+    assert r["warnings"] == []
+
+
+def test_empty_filter_says_it_covers_everything(client):
+    """Aggregates still answer an empty filter — "how big is IDC" is a real question — but it can
+    no longer be mistaken for a cohort."""
+    r = client.post("/v3/cohort/counts", json={"filters": {}}).json()
+    assert r["series"] > _ALL_OF_IDC
+    assert r["filters_applied"] == {"terms": {}, "ranges": {}}
+    assert any("ENTIRE IDC archive" in w for w in r["warnings"])
+
+
+def test_predicates_that_constrain_nothing_are_reported(client):
+    """An empty value list is dropped by the compiler; unreported, it is the same footgun with a
+    correctly-shaped body."""
+    r = client.post("/v3/cohort/counts", json={"filters": {"terms": {"collection_id": []}}}).json()
+    assert r["series"] > _ALL_OF_IDC
+    assert r["filters_applied"]["terms"] == {}
+    assert any("'collection_id' was ignored" in w for w in r["warnings"])
+    assert any("ENTIRE IDC archive" in w for w in r["warnings"])
+
+
+def test_licenses_echo_the_filter_they_applied(client):
+    r = client.post("/v3/licenses", json={"filters": {"terms": _TERMS}}).json()
+    assert r["licenses"] and r["filters_applied"]["terms"] == _TERMS
+    assert r["warnings"] == []
+
+
+def test_wrong_case_value_is_explained_not_just_zeroed(client):
+    """`Modality: ['mr']` counts zero exactly like a genuinely empty cohort. Saying so is the
+    difference between "no such data" and "wrong casing"."""
+    r = client.post("/v3/cohort/counts", json={"filters": {"terms": {"Modality": ["mr"]}}}).json()
+    assert r["series"] == 0
+    assert any("case-sensitively" in w and "MR" in w for w in r["warnings"]), r["warnings"]
+
+    # A genuinely empty cohort stays quiet — the hint must not fire on correct casing.
+    empty = client.post(
+        "/v3/cohort/counts",
+        json={"filters": {"terms": {"collection_id": ["rider_pilot"], "Modality": ["MR"]}}},
+    ).json()
+    assert empty["series"] == 0 and empty["warnings"] == []
+
+
+# --- enumerating series requires a predicate ------------------------------------------------
+
+
+def test_unfiltered_enumeration_is_refused(client):
+    """Counts describe the archive; a manifest *enumerates* it. No caller means to ask for a
+    download payload covering 100+ TB, so it is a 400 rather than a warning."""
+    for path in ("/v3/cohort/manifest", "/v3/cohort/manifest.txt"):
+        r = client.post(path, json={"filters": {}})
+        assert r.status_code == 400, (path, r.text[:200])
+        assert "At least one filter predicate is required" in r.text, path
+
+    # ... and the refusal explains a filter that *looks* present but compiled to nothing.
+    r = client.post("/v3/cohort/manifest", json={"filters": {"terms": {"collection_id": []}}})
+    assert r.status_code == 400
+    assert "'collection_id' was ignored" in r.json()["error"]["message"]
+
+
+def test_filtered_manifest_still_works(client):
+    m = client.post(
+        "/v3/cohort/manifest", json={"filters": {"terms": _TERMS}, "page_size": 3}
+    ).json()
+    assert 0 < m["returned"] <= 3
+    assert m["counts"]["filters_applied"]["terms"] == _TERMS
+    assert m["counts"]["warnings"] == []
+
+
+# --- core + MCP see the same guarantees ----------------------------------------------------
+
+
+def test_compiler_reports_what_it_dropped():
+    ok = compile_filters(CohortFilters(terms=_TERMS))
+    assert ok.where == '"collection_id" IN (?)' and ok.params == ["rider_pilot"]
+    assert ok.applied.terms == _TERMS and ok.warnings == []
+
+    dropped = compile_filters(
+        CohortFilters(terms={"collection_id": []}, ranges={"instanceCount": {}})
+    )
+    assert dropped.where == "TRUE"  # selects everything, so it must never pass require_filter
+    assert dropped.applied.terms == {} and dropped.applied.ranges == {}
+    assert len(dropped.warnings) == 3  # both dropped predicates, plus "no filter applied"
+
+    with pytest.raises(Exception, match="At least one filter predicate"):
+        require_filter(dropped, "do the dangerous thing")
+    require_filter(ok, "do the dangerous thing")  # a real predicate passes
+
+
+async def test_mcp_refuses_to_enumerate_the_whole_archive(parse_mcp):
+    for tool in ("build_cohort", "get_cohort_urls"):
+        with pytest.raises(ToolError, match="At least one filter predicate"):
+            await mcp.call_tool(tool, {})
+    # Aggregates stay available unfiltered, with the warning attached.
+    lic = parse_mcp(await mcp.call_tool("get_licenses", {}))
+    assert any("ENTIRE IDC archive" in w for w in lic["warnings"])
+
+
+def test_openapi_documents_the_filter_contract(client):
+    """Swagger UI and the post-deploy example smoke test read this spec, and an agent may be
+    working from it alone — so the uniform shape and the refusals have to be *in* it, not just in
+    the implementation."""
+    spec = client.get("/v3/openapi.json").json()
+    schemas = spec["components"]["schemas"]
+
+    for name in ("CountsRequest", "LicensesRequest", "ManifestRequest", "CitationsRequest"):
+        assert list(schemas[name]["properties"])[0] == "filters", name
+        assert schemas[name]["additionalProperties"] is False, name
+        # The declared example is what Swagger pre-fills and what the smoke test fires.
+        assert "filters" in schemas[name]["examples"][0], name
+    for name in ("CohortFilters", "NumericRange"):
+        assert schemas[name]["additionalProperties"] is False, name
+
+    def error_examples(path, code):
+        responses = spec["paths"][path]["post"]["responses"]
+        assert {"400", "422"} <= set(responses), path
+        return responses[code]["content"]["application/json"]["examples"]
+
+    for path in (*_WRAPPED, "/v3/cohort/manifest.txt"):
+        assert "filter not under `filters`" in error_examples(path, "422"), path
+    # Only the enumerating endpoints document the unfiltered refusal.
+    assert "no filter predicate" in error_examples("/v3/cohort/manifest", "400")
+    assert "no filter predicate" in error_examples("/v3/cohort/manifest.txt", "400")
+    assert "no filter predicate" not in error_examples("/v3/cohort/counts", "400")
+
+    for name in ("CohortCounts", "LicensesResult", "CitationsResult"):
+        props = schemas[name]["properties"]
+        assert props["filters_applied"]["description"] and props["warnings"]["description"], name
+
+
+async def test_mcp_malformed_filter_argument_is_actionable():
+    """A rejected filter must reach the model as the expected shape — `guard`'s generic
+    "Internal error" would tell it nothing to act on."""
+    with pytest.raises(ToolError) as exc:
+        await mcp.call_tool("build_cohort", {"ranges": {"instanceCount": {"min": 5}}})
+    assert "Malformed filter arguments" in str(exc.value)
+    assert "gte" in str(exc.value)
diff --git a/tests/test_parity.py b/tests/test_parity.py
index e7c832c..1281a87 100644
--- a/tests/test_parity.py
+++ b/tests/test_parity.py
@@ -30,15 +30,20 @@ async def test_version_parity(ctx, client, parse_mcp):
 async def test_counts_parity(ctx, client, parse_mcp):
     core_series = ctx.cohort.counts(CohortFilters(terms=_TERMS)).series
 
-    rest_series = client.post("/v3/cohort/counts", json={"terms": _TERMS}).json()["series"]
+    rest_series = client.post("/v3/cohort/counts", json={"filters": {"terms": _TERMS}}).json()[
+        "series"
+    ]
 
     mcp_series = parse_mcp(await mcp.call_tool("build_cohort", {"terms": _TERMS}))["total_series"]
 
     assert core_series == rest_series == mcp_series > 0
 
 
-def _fake_doi_get(url, headers=None, timeout=None):
-    """Stub DOI content negotiation so citation tests don't touch the network."""
+def _fake_doi_get(url, headers=None, timeout=None, params=None):
+    """Stub DOI resolution so citation tests don't touch the network. Accepts `params` because
+    the service tries DataCite's batch endpoint first; the reply carries no DOI, so every entry
+    falls back to the per-DOI path — which is what keeps this a parity test and not a batch one
+    (see tests/test_citations.py)."""
 
     class _Resp:
         status_code = 200