fix: stop-detection budget preserves stops found before a dense segment

bhulan/analytics/insights.py

@@ -482,15 +482,17 @@ def compute_insights(request: InsightsRequest) -> InsightsReport:
             max_scan_work=_MAX_STOP_SCAN_WORK,
             progressive_drift_fraction=opts.progressive_drift_fraction,
         )
-    except StopScanBudgetExceeded:
-        # The input is too dense/degenerate for stop detection within budget.
-        # Skip stops (and therefore trips) but still return every other insight
-        # — distance, speed, bbox, hotspots — rather than hang or 500.
-        raw_stops = []
+    except StopScanBudgetExceeded as exc:
+        # A dense/degenerate segment (very slow drift or same-timestamp mass)
+        # blew the scan budget partway through. Keep the stops found *before*
+        # it — discarding them turned one bad segment into "zero stops" for an
+        # otherwise ordinary track — and only note that detection past that
+        # point was truncated. Everything else (distance, speed, bbox,
+        # hotspots) is still returned.
+        raw_stops = exc.partial_stops
         quality.issues.append(
-            "Stop detection skipped: the track is too dense or degenerate "
-            "(e.g. a very slow drift or many samples at one timestamp) to "
-            "analyze within the allotted budget."
+            "Stop detection was truncated after a very dense or slowly-drifting "
+            "segment; any stops beyond that point may be missing."
         )
     stops = merge_nearby_stops(
         raw_stops,

bhulan/analytics/stops.py

@@ -59,10 +59,20 @@
 
 
 class StopScanBudgetExceeded(Exception):  # noqa: N818 — "Exceeded" reads clearly; renaming would churn 5 call sites
-    """detect_stops' scan work exceeded its budget — the input is a pathological
-    single giant cluster (a slow drift or same-timestamp mass), not a real
-    track. Callers degrade gracefully (report no stops + a quality note) rather
-    than hang."""
+    """detect_stops' scan work exceeded its budget partway through a track.
+
+    The trigger is a pathological *segment* — a very slow drift or a
+    same-timestamp mass — not a wholly bad track: the samples before it are
+    usually a normal sequence of real stops. ``partial_stops`` carries every
+    stop already found when the budget was hit, so callers keep those and only
+    lose detection past the offending point, rather than discarding the entire
+    track's results (which turned one dense segment into "zero stops" for an
+    otherwise ordinary file). See ADR 0016.
+    """
+
+    def __init__(self, message: str, partial_stops: Optional[List["Stop"]] = None):
+        super().__init__(message)
+        self.partial_stops: List[Stop] = list(partial_stops) if partial_stops else []
 
 
 @dataclass(frozen=True)
@@ -225,10 +235,12 @@ def detect_stops(
             approach of :func:`bhulan.analytics.trips._trip_bounds`.
         max_scan_work: Optional cap on the total cluster-scan work (summed
             grown-cluster sizes). When exceeded, :class:`StopScanBudgetExceeded`
-            is raised — the input is a pathological single giant cluster, not a
-            real track. ``None`` (the default) leaves the scan uncapped; the
-            public API passes ``_MAX_STOP_SCAN_WORK_PER_POINT * n`` so a crafted
-            input can't tie up a worker.
+            is raised carrying ``partial_stops`` — every stop found before the
+            budget was hit — so the caller keeps those and only truncates
+            detection past the pathological point. ``None`` (the default) leaves
+            the scan uncapped; the public API passes the fixed absolute
+            ``_MAX_STOP_SCAN_WORK`` (not scaled by ``n``) so a crafted input
+            can't tie up a worker.
     """
     ts_points: List[TrackSample] = [p for p in points if p.ts_utc is not None]
     n = len(ts_points)
@@ -252,8 +264,13 @@ def detect_stops(
         if max_scan_work is not None:
             scan_work += work
             if scan_work > max_scan_work:
+                # Stop scanning at the pathological segment, but hand back every
+                # stop found *before* it — discarding the whole list turned one
+                # dense/slow-drift segment into "zero stops" for an otherwise
+                # ordinary track (a real dwell followed by a long slow walk).
                 raise StopScanBudgetExceeded(
-                    f"stop scan exceeded {max_scan_work} work units at sample {i}"
+                    f"stop scan exceeded {max_scan_work} work units at sample {i}",
+                    partial_stops=stops,
                 )
         if end > i:
             duration = (

tests/adversary/test_detect_stops_scan_work_dos.py

@@ -100,7 +100,9 @@ def test_insights_endpoint_bounds_a_pathological_body(client: TestClient):
     assert r.status_code == 200
     assert elapsed < 20.0, f"a 100k-point drift took {elapsed:.1f}s — the scan-work budget is not bounding it"
     body = r.json()
+    # A pure drift has no real dwell before the budget hit, so there are no
+    # partial stops to keep — the truncation note stands in for "no stops".
     assert body["stops"] == []
-    assert any("too dense or degenerate" in issue for issue in body["quality"]["issues"])
+    assert any("truncated" in issue for issue in body["quality"]["issues"])
     # the rest of the pipeline still ran
     assert body["summary"]["total_distance_km"] > 0

tests/adversary/test_stop_budget_preserves_earlier_stops.py

@@ -0,0 +1,81 @@
+"""
+Defect (silently-wrong-answer): when ``detect_stops`` hits its scan-work budget
+partway through a track, it raised ``StopScanBudgetExceeded`` and
+``compute_insights`` caught it by setting ``raw_stops = []`` — discarding *every*
+stop, including real dwells already detected before the offending segment.
+
+So an ordinary track — a real 10-minute dwell followed by a long slow walk
+(~9k samples, entirely realistic for a hike/commute file) — trips the absolute
+12M-work cap on the walk segment and returns **zero stops**, silently dropping
+the real dwell. The budget was meant to prevent a DoS from a pathological single
+giant cluster; instead it turned one dense segment into "no stops" for a whole
+legitimate file.
+
+Fix: the budget still bounds work (DoS protection intact), but on exceeding it
+``detect_stops`` hands back ``partial_stops`` — the stops found before the
+budget was hit — so the caller keeps them and only truncates detection past that
+point (plus a quality note). See ADR 0016.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi.testclient import TestClient
+
+import bhulan.storage.mongo_repo as mongo_repo  # noqa: F401  (fixture parity)
+from bhulan.analytics.mobility import TrackSample
+from bhulan.analytics.stops import StopScanBudgetExceeded, detect_stops
+
+_T0 = datetime(2024, 1, 1, tzinfo=timezone.utc)
+
+
+def _real_dwell(n: int, start_s: int):
+    # n samples at one fixed spot, one per minute -> a genuine multi-minute stop.
+    return [
+        TrackSample(lat=40.0, lon=-73.0, ts_utc=_T0 + timedelta(seconds=start_s + i * 60))
+        for i in range(n)
+    ]
+
+
+def _slow_drift(n: int, start_s: int):
+    # A slow directional walk within one radius: forces repeated exact-spread
+    # recomputes, i.e. the O(n*cluster_size) work the budget is there to cap.
+    return [
+        TrackSample(lat=41.0 + i * 7e-8, lon=-73.0, ts_utc=_T0 + timedelta(seconds=start_s + i))
+        for i in range(n)
+    ]
+
+
+def test_detect_stops_returns_partial_stops_on_budget_exceeded():
+    """Unit level: the real dwell is on the exception's partial_stops."""
+    track = _real_dwell(11, 0) + _slow_drift(4000, 3600)
+    # A low budget forces the cap deterministically once the drift segment is
+    # scanned (the drift alone charges ~1 work unit per sample).
+    with pytest.raises(StopScanBudgetExceeded) as ei:
+        detect_stops(track, radius_m=50.0, min_duration_s=300.0, max_scan_work=1000)
+    partial = ei.value.partial_stops
+    assert len(partial) == 1, f"expected the real dwell preserved, got {partial}"
+    assert partial[0].sample_count == 11
+    # And the whole track under the default (very high) cap still finds the dwell.
+    assert len(detect_stops(track, radius_m=50.0, min_duration_s=300.0)) >= 1
+
+
+def test_insights_keeps_real_stop_before_a_budget_tripping_walk(client: TestClient):
+    """End to end: a real dwell + a ~9k-point slow walk must not return 0 stops."""
+    t0 = datetime(2025, 1, 1, tzinfo=timezone.utc)
+
+    def iso(s):
+        return (t0 + timedelta(seconds=s)).isoformat().replace("+00:00", "Z")
+
+    dwell = [{"lat": 40.0, "lon": -73.0, "ts_utc": iso(i * 60)} for i in range(11)]
+    walk = [{"lat": 41.0 + i * 7e-8, "lon": -73.0, "ts_utc": iso(3600 + i)} for i in range(9000)]
+
+    r = client.post("/v1/insights", json={"points": dwell + walk})
+    assert r.status_code == 200, r.text[:200]
+    body = r.json()
+    assert len(body["stops"]) >= 1, (
+        "the real 10-minute dwell must survive a later budget-tripping walk, "
+        f"got {len(body['stops'])} stops"
+    )
+    # The surviving stop is the real dwell, not an artefact of the walk.
+    assert any(s["sample_count"] == 11 for s in body["stops"])