diff --git a/packages/flow/openadapt_flow/ensure_record.py b/packages/flow/openadapt_flow/ensure_record.py
index 457fb9d29..03481292a 100644
--- a/packages/flow/openadapt_flow/ensure_record.py
+++ b/packages/flow/openadapt_flow/ensure_record.py
@@ -58,6 +58,7 @@
 )
 
 from openadapt_flow.fhir_read import (
+    check_collection_url,
     check_search_coverage,
     check_search_outcome,
     fhir_read_headers,
@@ -1117,7 +1118,8 @@ def bundle(
     ) -> list[dict[str, Any]]:
         """Every entry of a search or history Bundle, across all pages."""
 
-        url: Optional[str] = f"{self.base_url}/{path}"
+        initial_url = f"{self.base_url}/{path}"
+        url: Optional[str] = initial_url
         query: Optional[Mapping[str, Any]] = params
         entries: list[dict[str, Any]] = []
         total: Optional[int] = None
@@ -1125,7 +1127,7 @@ def bundle(
         match_identities: set[tuple[str, str]] = set()
         for _ in range(MAX_PAGES):
             assert url is not None
-            _, body = self.fetch(url, query)
+            headers, body = self.fetch(url, query)
             if not isinstance(body, dict) or body.get("resourceType") != "Bundle":
                 raise Unreadable(f"{path}: the reply is not a Bundle")
             if self.snapshot_path is not None:
@@ -1212,6 +1214,30 @@ def bundle(
             except ValueError as exc:
                 raise Unreadable(f"{path}: {exc}") from exc
             if not nxt:
+                if body.get("type") != "searchset" and body.get("type") != "history":
+                    try:
+                        if params:
+                            raise ValueError(
+                                "collection searches require an unrestricted empty query"
+                            )
+                        check_collection_url(initial_url)
+                        served = headers.get("content-location") or headers.get(
+                            "location"
+                        )
+                        if served:
+                            check_collection_url(served)
+                        links = body.get("link")
+                        if isinstance(links, list):
+                            for link in links:
+                                if (
+                                    isinstance(link, dict)
+                                    and link.get("relation") == "self"
+                                ):
+                                    self_url = link.get("url")
+                                    if isinstance(self_url, str):
+                                        check_collection_url(self_url)
+                    except ValueError as exc:
+                        raise Unreadable(f"{path}: {exc}") from exc
                 return entries
             if _origin(nxt) != _origin(self.base_url):
                 raise Unreadable(f"{path}: next page on another origin")
diff --git a/packages/flow/openadapt_flow/fhir_read.py b/packages/flow/openadapt_flow/fhir_read.py
index 83ff94295..3e1ee0565 100644
--- a/packages/flow/openadapt_flow/fhir_read.py
+++ b/packages/flow/openadapt_flow/fhir_read.py
@@ -135,7 +135,7 @@ def check_collection_search(
                 raise ValueError("a collection search entry is not a matching resource")
 
 
-def _check_collection_url(url: str) -> None:
+def check_collection_url(url: str) -> None:
     """The same empty-query rule covers sent, effective and self-link URLs."""
     try:
         query = urlsplit(url).query
@@ -145,6 +145,9 @@ def _check_collection_url(url: str) -> None:
         raise ValueError("collection searches require an unrestricted empty query")
 
 
+_check_collection_url = check_collection_url
+
+
 def check_search_coverage(total: Optional[int], matches: int) -> None:
     """Reject observed caps/count conflicts, without certifying native coverage.
 
diff --git a/packages/flow/tests/test_ensure_record.py b/packages/flow/tests/test_ensure_record.py
index 75f129d08..f7e8bf29e 100644
--- a/packages/flow/tests/test_ensure_record.py
+++ b/packages/flow/tests/test_ensure_record.py
@@ -1587,3 +1587,64 @@ def test_version_trail_names_authors_when_the_order_is_unambiguous():
         ],
     )
     assert [v["author"] for v in version_trail(change)] == [BOT, OTHER]
+
+
+def test_bundle_refuses_parameterized_non_searchset_bundle():
+    records = [patient("1"), patient("2")]
+    routes = {
+        "Patient": bundle(*records, kind="collection"),
+        "Patient?_count=1": bundle(*records, kind="collection"),
+    }
+    reader = reader_for(routes)
+
+    # Configured query params on non-searchset bundle are refused
+    with pytest.raises(
+        Unreadable, match="collection searches require an unrestricted empty query"
+    ):
+        reader.bundle("Patient", {"_count": "1"})
+
+    # Query in path/URL is refused
+    with pytest.raises(
+        Unreadable, match="collection searches require an unrestricted empty query"
+    ):
+        reader.bundle("Patient?_count=1")
+
+    # Query in self link is refused
+    collection_with_self_query = bundle(*records, kind="collection")
+    collection_with_self_query["link"] = [
+        {"relation": "self", "url": "http://fhir.test/Patient?_count=1"}
+    ]
+    reader_self = reader_for({"Patient": collection_with_self_query})
+    with pytest.raises(
+        Unreadable, match="collection searches require an unrestricted empty query"
+    ):
+        reader_self.bundle("Patient")
+
+    # Query in effective/served URL (Content-Location) is refused
+    def get_with_served(url, params):
+        return (
+            200,
+            {"content-location": "http://fhir.test/Patient?_count=1"},
+            bundle(*records, kind="collection"),
+        )
+
+    reader_served = FhirReader(
+        contract(change_feed={"audit": {}, "sweep_types": ["Observation"]}),
+        get_with_served,
+    )
+    with pytest.raises(
+        Unreadable, match="collection searches require an unrestricted empty query"
+    ):
+        reader_served.bundle("Patient")
+
+    # Unqueried collection bundle without next link is accepted as complete
+    unqueried = reader.bundle("Patient")
+    assert len(unqueried) == 2
+
+    # A searchset bundle with parameters retains normal searchset pagination/coverage
+    searchset_routes = {
+        "Patient": bundle(*records, total=2, kind="searchset"),
+    }
+    reader_searchset = reader_for(searchset_routes)
+    searchset_entries = reader_searchset.bundle("Patient", {"_count": "2"})
+    assert len(searchset_entries) == 2