stop paging when the provider stops making progress

An ArcGIS layer without supportsPagination accepts resultOffset and ignores it,
answering every page with the first one. The VHA profile reader advanced its
offset by the page length and stopped at the announced count, so for a count
that is a multiple of the page size it collected N copies of page one — and its
completeness check, len(features) == candidate_count, passed. Four announced
records became four stored records, two of them duplicates, filed under an
official provenance. That is the substitution bounded acquisition exists to
prevent, arriving through the front door.

The reader now refuses a record it already collected. It fails rather than
silently dropping the duplicate: a provider that cannot page is a provider
whose count proves nothing, so a smaller-but-clean result would still be
unverifiable. Its watercourse-name loop was worse — a bare `while True` that
ended only when the provider stopped setting exceededTransferLimit, with names
deduplicated by code so a stuck provider produced no visible change while the
requests continued. It now refuses a repeated page body, and both loops have
the page budget the sibling readers already had.

Those siblings turned out to be fine. GRB and official vector already refuse a
repeated page URL, bound the page count, and deduplicate on feature identity —
but none of it had a test, so none of it was known to work. Exercised now,
including the case where distinct URLs defeat the loop check and the budget is
the only backstop. A duplicate across two genuinely different pages is kept
once rather than failing, because a cursor over a changing table produces that
legitimately.

Also: _bash_path fell back to the raw path whenever wslpath failed, except on
timeout, which propagated and reddened the suite when starting WSL took more
than ten seconds under load.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/app/services/bathymetry_profile_acquisition_service.py

@@ -276,6 +276,45 @@ class BathymetryProfileAcquisitionService:
             "spatialRel": "esriSpatialRelIntersects",
         }
 
+    @staticmethod
+    def _unseen_records(
+        page_features: list[Any],
+        seen_object_ids: set[str],
+    ) -> list[dict[str, Any]]:
+        """Every page must bring records the earlier pages did not.
+
+        An ArcGIS layer without ``supportsPagination`` accepts ``resultOffset``
+        and ignores it, answering every page with the first one. Advancing the
+        offset by the page length still reaches the announced count, so the
+        completeness check below passed while the dataset held N copies of page
+        one — a silent substitution of the source data, which is the one thing
+        bounded acquisition exists to prevent.
+        """
+
+        fresh: list[dict[str, Any]] = []
+        for item in page_features:
+            if not isinstance(item, dict):
+                continue
+            attributes = item.get("attributes")
+            object_id = attributes.get("OBJECTID") if isinstance(attributes, dict) else None
+            if object_id is None:
+                raise AppError(
+                    code="BATHYMETRY_PROVIDER_INVALID_RESPONSE",
+                    message="VHA profile record has no OBJECTID, so pagination cannot be verified",
+                    status_code=502,
+                )
+            key = str(object_id)
+            if key in seen_object_ids:
+                raise AppError(
+                    code="BATHYMETRY_PROVIDER_UNSTABLE_PAGINATION",
+                    message="VHA profile pagination repeated a record; the layer is not honouring resultOffset",
+                    details={"object_id": key},
+                    status_code=502,
+                )
+            seen_object_ids.add(key)
+            fresh.append(item)
+        return fresh
+
     @staticmethod
     def _fetch_profiles(
         bbox_values: tuple[float, float, float, float],
@@ -307,8 +346,16 @@ class BathymetryProfileAcquisitionService:
         features: list[dict[str, Any]] = []
         response_hashes: list[str] = []
         request_urls: list[str] = [count_url]
+        seen_object_ids: set[str] = set()
         offset = 0
         while offset < candidate_count:
+            if len(request_urls) > settings.bathymetry_profiles_max_pages:
+                raise AppError(
+                    code="BATHYMETRY_SCOPE_TOO_LARGE",
+                    message="VHA profile pagination exceeded the configured page limit; acquire smaller area partitions",
+                    details={"max_pages": settings.bathymetry_profiles_max_pages},
+                    status_code=422,
+                )
             page_url = BathymetryProfileAcquisitionService._query_url(
                 base,
                 {
@@ -328,11 +375,13 @@ class BathymetryProfileAcquisitionService:
                     message="VHA profile response does not contain a feature list",
                     status_code=502,
                 )
-            features.extend(item for item in page_features if isinstance(item, dict))
             response_hashes.append(page_sha)
             request_urls.append(page_url)
             if not page_features:
                 break
+            features.extend(
+                BathymetryProfileAcquisitionService._unseen_records(page_features, seen_object_ids)
+            )
             offset += len(page_features)
         if len(features) != candidate_count:
             raise AppError(
@@ -363,7 +412,15 @@ class BathymetryProfileAcquisitionService:
         for start in range(0, len(ordered_codes), 100):
             chunk = ordered_codes[start : start + 100]
             offset = 0
+            seen_page_hashes: set[str] = set()
             while True:
+                if len(seen_page_hashes) >= settings.bathymetry_profiles_max_pages:
+                    raise AppError(
+                        code="BATHYMETRY_SCOPE_TOO_LARGE",
+                        message="VHA watercourse pagination exceeded the configured page limit",
+                        details={"max_pages": settings.bathymetry_profiles_max_pages},
+                        status_code=422,
+                    )
                 url = BathymetryProfileAcquisitionService._query_url(
                     base,
                     {
@@ -386,6 +443,16 @@ class BathymetryProfileAcquisitionService:
                         message="VHA watercourse response does not contain a feature list",
                         status_code=502,
                     )
+                if response_sha in seen_page_hashes:
+                    # The names themselves deduplicate by code, so a stuck
+                    # provider produced no visible change while the loop, which
+                    # ended only on exceededTransferLimit, kept requesting.
+                    raise AppError(
+                        code="BATHYMETRY_PROVIDER_UNSTABLE_PAGINATION",
+                        message="VHA watercourse pagination returned the same page again",
+                        status_code=502,
+                    )
+                seen_page_hashes.add(response_sha)
                 for feature in page_features:
                     attributes = feature.get("attributes") if isinstance(feature, dict) else None
                     if not isinstance(attributes, dict):