v0.14.2: exempt global satellite telemetry from bbox filter (#108)

Bypass the v0.14.0 monitoring-area bbox filter for global-by-design satellite telemetry (sat_positions, sat_orbits). Recovers 462k+/87k+ dropped events; null-geom celestrak_tle + observer-anchored satpass_predict/n2yo_visualpasses untouched. Drift-detection test keeps the archive static set in sync with the adapter class attr.

src/central/adapter.py

@@ -57,6 +57,15 @@ class SourceAdapter(ABC):
     gauges) that would drown discrete-event signal; shown on /telemetry instead.
     GUI-only — does not affect publishing or the events.json contract."""
 
+    bypass_bbox_filter: bool = False
+    """Set True for adapters whose events are global-by-design (satellite
+    telemetry, space weather) and should never be filtered by a geographic
+    monitoring area. The supervisor publishes such events unconditionally,
+    skipping the publish-time ``classify_geom_areas`` check (v0.14.2). The
+    archive consumer mirrors this via the module-level ``_BYPASS_BBOX_ADAPTERS``
+    set in ``central.archive`` -- the two MUST stay in sync (enforced by
+    ``tests/test_bypass_bbox_consistency.py``)."""
+
     @abstractmethod
     async def poll(self) -> AsyncIterator[Event]:
         """

src/central/adapters/sat_orbits.py

@@ -121,6 +121,11 @@ class SatOrbitsAdapter(SourceAdapter):
     default_cadence_s = 300  # 5 min
     data_class = "telemetry"
     enrichment_locations = []
+    # Global-by-design: forward-orbit LineStrings (mostly polar) span the globe,
+    # not a region. A geographic monitoring area would drop nearly all of them.
+    # Skip the publish-time/archive bbox filter (v0.14.2). Keep in sync with
+    # archive._BYPASS_BBOX_ADAPTERS.
+    bypass_bbox_filter = True
 
     def __init__(
         self,

src/central/adapters/sat_positions.py

@@ -130,6 +130,12 @@ class SatPositionsAdapter(SourceAdapter):
     default_cadence_s = 60
     data_class = "telemetry"
     enrichment_locations = []
+    # Global-by-design: one event per satellite per poll with the sub-satellite
+    # point as centroid -- a worldwide firehose. A geographic monitoring area
+    # (e.g. Idaho) would drop ~99.7% of it, defeating "where is the ISS" global
+    # queries. Skip the publish-time/archive bbox filter (v0.14.2). Keep in sync
+    # with archive._BYPASS_BBOX_ADAPTERS.
+    bypass_bbox_filter = True
 
     def __init__(
         self,

src/central/archive.py

@@ -36,6 +36,15 @@ BATCH_SIZE = 100
 FETCH_TIMEOUT = 5.0
 ACK_WAIT = 30
 
+# v0.14.2: adapters whose events are global-by-design (satellite telemetry) and
+# must bypass the geographic monitoring-area bbox filter. The archive consumer
+# only has the adapter NAME at runtime (it reads off the wire, not the adapter
+# class), so it can't reach SourceAdapter.bypass_bbox_filter directly -- this
+# static set is the consumer-side mirror. It MUST stay in sync with the adapter
+# classes that set bypass_bbox_filter = True; tests/test_bypass_bbox_consistency.py
+# fails CI if the two drift.
+_BYPASS_BBOX_ADAPTERS = {"sat_positions", "sat_orbits"}
+
 
 def consumer_name_for(stream: str) -> str:
     """Generate consumer name for a stream."""
@@ -89,6 +98,9 @@ class ArchiveConsumer:
         self._shutdown_event = asyncio.Event()
         self._monitoring_areas: list[MonitoringArea] = []
         self._dropped: dict[str, int] = {}
+        # v0.14.2: events archived unconditionally because their adapter is in
+        # _BYPASS_BBOX_ADAPTERS (global-by-design satellite telemetry).
+        self._bypassed: dict[str, int] = {}
 
     @property
     def _monitoring_area(self) -> MonitoringArea | None:
@@ -249,20 +261,29 @@ class ArchiveConsumer:
 
         geom_json = build_geom_json(geo_data)
 
-        verdict = classify_geom_areas(geom_json, self._monitoring_areas)
-        if verdict == "out-of-bounds":
-            self._dropped[adapter] = self._dropped.get(adapter, 0) + 1
-            logger.debug(
-                "Dropped out-of-bounds event (archive bbox filter)",
-                extra={"id": event_id, "adapter": adapter, "category": category},
-            )
-            await msg.ack()
-            return
-        if verdict == "invalid-geom":
-            logger.warning(
-                "Geom could not be evaluated for bbox filter; archiving",
-                extra={"id": event_id, "adapter": adapter},
-            )
+        # v0.14.2: global-by-design adapters (satellite telemetry) bypass the
+        # geographic monitoring-area filter entirely -- mirrors the supervisor's
+        # SourceAdapter.bypass_bbox_filter carve-out, but keyed on adapter NAME
+        # because the consumer reads events off the wire, not the adapter class.
+        # _BYPASS_BBOX_ADAPTERS must stay in sync with the adapter class attrs
+        # (tests/test_bypass_bbox_consistency.py enforces it).
+        if adapter in _BYPASS_BBOX_ADAPTERS:
+            self._bypassed[adapter] = self._bypassed.get(adapter, 0) + 1
+        else:
+            verdict = classify_geom_areas(geom_json, self._monitoring_areas)
+            if verdict == "out-of-bounds":
+                self._dropped[adapter] = self._dropped.get(adapter, 0) + 1
+                logger.debug(
+                    "Dropped out-of-bounds event (archive bbox filter)",
+                    extra={"id": event_id, "adapter": adapter, "category": category},
+                )
+                await msg.ack()
+                return
+            if verdict == "invalid-geom":
+                logger.warning(
+                    "Geom could not be evaluated for bbox filter; archiving",
+                    extra={"id": event_id, "adapter": adapter},
+                )
 
         try:
             if geom_json:
@@ -377,6 +398,8 @@ class ArchiveConsumer:
                 ],
             },
         )
+        # v0.14.2: surface the global-by-design carve-out for operators.
+        logger.info("Bypass-bbox adapters: %s", sorted(_BYPASS_BBOX_ADAPTERS))
 
     async def run(self) -> None:
         """Run consume loops for all streams until shutdown."""

src/central/supervisor.py

@@ -245,6 +245,10 @@ class Supervisor:
         # every MONITORING_AREA_REFRESH_S from config.system.
         self._monitoring_areas: list[MonitoringArea] = []
         self._dropped_publish: dict[str, int] = {}
+        # v0.14.2: events published unconditionally because their adapter sets
+        # bypass_bbox_filter (global-by-design satellite telemetry). Counted
+        # separately so operators can confirm the carve-out is doing work.
+        self._bypassed_publish: dict[str, int] = {}
 
     @property
     def _monitoring_area(self) -> MonitoringArea | None:
@@ -286,6 +290,13 @@ class Supervisor:
                 ],
             },
         )
+        # v0.14.2: surface the global-by-design carve-out so operators can see
+        # which adapters skip the bbox filter.
+        bypass_adapters = sorted(
+            name for name, cls in self._adapters.items()
+            if getattr(cls, "bypass_bbox_filter", False)
+        )
+        logger.info("Bypass-bbox adapters: %s", bypass_adapters)
 
     async def disconnect(self) -> None:
         """Disconnect from NATS."""
@@ -334,10 +345,13 @@ class Supervisor:
                         "Could not refresh monitoring area; keeping previous value",
                         extra={"error": str(e)},
                     )
-                if self._dropped_publish:
+                if self._dropped_publish or self._bypassed_publish:
                     logger.info(
                         "publish bbox filter drop summary (cumulative)",
-                        extra={"dropped_by_adapter": dict(self._dropped_publish)},
+                        extra={
+                            "dropped_by_adapter": dict(self._dropped_publish),
+                            "bypassed_by_adapter": dict(self._bypassed_publish),
+                        },
                     )
 
     def _create_adapter(self, config: AdapterConfig) -> SourceAdapter:
@@ -425,35 +439,45 @@ class Supervisor:
 
                     subject = state.adapter.subject_for(event)
 
-                    # v0.10.2 publish-time monitoring-area filter. Mirrors
-                    # archive's classify->ACK pattern but here we just `continue`
-                    # without mark_published -- if the area widens later the
-                    # next poll re-yields the same id and we'll publish it
-                    # naturally. Marking published on drop would be a forward-
-                    # only blackhole.
-                    geom_json = build_geom_json(
-                        event.geo.model_dump() if event.geo else None
-                    )
-                    verdict = classify_geom_areas(geom_json, self._monitoring_areas)
-                    if verdict == "out-of-bounds":
-                        self._dropped_publish[state.name] = (
-                            self._dropped_publish.get(state.name, 0) + 1
+                    # v0.14.2: global-by-design adapters (satellite telemetry)
+                    # bypass the geographic monitoring-area filter entirely --
+                    # their events are worldwide and a regional bbox would drop
+                    # nearly all of them. Count separately for observability, then
+                    # fall through to the shared publish path below.
+                    if state.adapter.bypass_bbox_filter:
+                        self._bypassed_publish[state.name] = (
+                            self._bypassed_publish.get(state.name, 0) + 1
                         )
-                        logger.debug(
-                            "Dropped out-of-bounds event at publish (monitoring-area filter)",
-                            extra={
-                                "id": event.id,
-                                "adapter": state.name,
-                                "category": event.category,
-                                "subject": subject,
-                            },
-                        )
-                        continue
-                    if verdict == "invalid-geom":
-                        logger.warning(
-                            "Geom could not be evaluated for publish-time bbox filter; publishing",
-                            extra={"id": event.id, "adapter": state.name},
+                    else:
+                        # v0.10.2 publish-time monitoring-area filter. Mirrors
+                        # archive's classify->ACK pattern but here we just
+                        # `continue` without mark_published -- if the area widens
+                        # later the next poll re-yields the same id and we'll
+                        # publish it naturally. Marking published on drop would
+                        # be a forward-only blackhole.
+                        geom_json = build_geom_json(
+                            event.geo.model_dump() if event.geo else None
                         )
+                        verdict = classify_geom_areas(geom_json, self._monitoring_areas)
+                        if verdict == "out-of-bounds":
+                            self._dropped_publish[state.name] = (
+                                self._dropped_publish.get(state.name, 0) + 1
+                            )
+                            logger.debug(
+                                "Dropped out-of-bounds event at publish (monitoring-area filter)",
+                                extra={
+                                    "id": event.id,
+                                    "adapter": state.name,
+                                    "category": event.category,
+                                    "subject": subject,
+                                },
+                            )
+                            continue
+                        if verdict == "invalid-geom":
+                            logger.warning(
+                                "Geom could not be evaluated for publish-time bbox filter; publishing",
+                                extra={"id": event.id, "adapter": state.name},
+                            )
 
                     # Publish
                     await self._publish_event(subject, envelope, msg_id)

tests/test_archive_bbox_filter.py

@@ -117,3 +117,30 @@ class TestProcessMessageMultiArea:
         await c._process_message(_make_msg(_envelope("wzdx", -74.0, 40.7)), conn)
         conn.execute.assert_awaited_once()
         assert c._dropped == {}
+
+
+class TestBypassBboxAdapters:
+    """v0.14.2: _BYPASS_BBOX_ADAPTERS skip the bbox filter at archive time."""
+
+    @pytest.mark.asyncio
+    async def test_sat_positions_archived_when_out_of_bounds(self):
+        # ISS sub-sat point over NYC, only IDAHO configured -> normally dropped,
+        # but sat_positions is global-by-design so it bypasses and inserts.
+        c = ArchiveConsumer("nats://x", "postgresql://x")
+        c._monitoring_area = IDAHO
+        conn = AsyncMock()
+        await c._process_message(_make_msg(_envelope("sat_positions", -74.0, 40.7)), conn)
+        conn.execute.assert_awaited_once()
+        assert c._dropped == {}
+        assert c._bypassed == {"sat_positions": 1}
+
+    @pytest.mark.asyncio
+    async def test_tomtom_flow_still_dropped(self):
+        # Regression guard: a non-bypass adapter is still filtered out-of-bounds.
+        c = ArchiveConsumer("nats://x", "postgresql://x")
+        c._monitoring_area = IDAHO
+        conn = AsyncMock()
+        await c._process_message(_make_msg(_envelope("tomtom_flow", -74.0, 40.7)), conn)
+        conn.execute.assert_not_called()
+        assert c._dropped == {"tomtom_flow": 1}
+        assert c._bypassed == {}

tests/test_bypass_bbox_consistency.py

@@ -0,0 +1,23 @@
+"""v0.14.2 drift guard: the archive consumer mirrors the per-adapter
+``bypass_bbox_filter`` flag as a static module-level set
+(``archive._BYPASS_BBOX_ADAPTERS``) because the consumer reads events off the
+wire and has no access to the adapter class at runtime. The two truth sources
+must never diverge -- if someone adds a global-by-design adapter and sets
+``bypass_bbox_filter = True`` but forgets the archive set (or vice versa), this
+test fails at CI time.
+"""
+from central.adapter_discovery import discover_adapters
+from central.archive import _BYPASS_BBOX_ADAPTERS
+
+
+def test_archive_static_set_matches_adapter_class_attr():
+    bypass_from_classes = {
+        name
+        for name, cls in discover_adapters().items()
+        if getattr(cls, "bypass_bbox_filter", False)
+    }
+    assert bypass_from_classes == _BYPASS_BBOX_ADAPTERS, (
+        "archive._BYPASS_BBOX_ADAPTERS is out of sync with the adapter classes "
+        "that set bypass_bbox_filter=True. Update the set in central.archive to "
+        f"match: {sorted(bypass_from_classes)}"
+    )

tests/test_supervisor_publish_filter.py

@@ -32,6 +32,9 @@ def _ev(eid: str, geo: Geo) -> Event:
 class _MockAdapter:
     requires_api_key = None
     enrichment_locations = ()
+    bypass_bbox_filter = False  # v0.14.2: real adapters inherit this from
+    # SourceAdapter; the mock declares it so _run_adapter_loop's direct attr
+    # access works. Bypass tests use _BypassMockAdapter (overrides True).
 
     def __init__(self, config, *_args) -> None:
         self.config = config
@@ -90,8 +93,8 @@ def sup_factory():
     return _build
 
 
-async def _drive(sup, events):
-    adapter = _MockAdapter(MagicMock(cadence_s=3600))
+async def _drive(sup, events, adapter_cls=_MockAdapter):
+    adapter = adapter_cls(MagicMock(cadence_s=3600))
     adapter.events = events
     config = AdapterConfig(
         name="mock", enabled=True, cadence_s=3600, settings={},
@@ -187,3 +190,50 @@ async def test_refresh_loop_reloads_area_and_logs_summary(
     assert any(
         "publish bbox filter drop summary" in r.message for r in caplog.records
     )
+
+
+# --- v0.14.2: global-by-design satellite telemetry bypasses the bbox filter ---
+
+class _BypassMockAdapter(_MockAdapter):
+    """Stand-in for a global-by-design adapter (sat_positions/sat_orbits)."""
+    bypass_bbox_filter = True
+
+
+class TestBypassBboxAdapters:
+    @pytest.mark.asyncio
+    async def test_sat_positions_publishes_when_out_of_bounds(self, sup_factory):
+        """Centroid over NYC (outside Idaho) still publishes; not dropped."""
+        sup = sup_factory(IDAHO)
+        a = await _drive(
+            sup, [_ev("iss", Geo(centroid=NYC))], adapter_cls=_BypassMockAdapter
+        )
+        assert sup._publish_event.await_count == 1
+        assert a.published == {"iss"}
+        assert sup._dropped_publish == {}
+        assert sup._bypassed_publish == {"mock": 1}
+
+    @pytest.mark.asyncio
+    async def test_sat_orbits_publishes_when_linestring_outside_idaho(self, sup_factory):
+        """An equatorial forward-track LineString never intersects Idaho, yet publishes."""
+        sup = sup_factory(IDAHO)
+        equator = Geo(geometry={
+            "type": "LineString",
+            "coordinates": [[0.0, 0.0], [10.0, 0.0], [20.0, 1.0]],
+        })
+        a = await _drive(
+            sup, [_ev("orbit1", equator)], adapter_cls=_BypassMockAdapter
+        )
+        assert sup._publish_event.await_count == 1
+        assert a.published == {"orbit1"}
+        assert sup._dropped_publish == {}
+        assert sup._bypassed_publish == {"mock": 1}
+
+    @pytest.mark.asyncio
+    async def test_tomtom_flow_still_drops_when_out_of_bounds(self, sup_factory):
+        """Regression guard: a non-bypass adapter is still filtered (NYC drops)."""
+        sup = sup_factory(IDAHO)
+        a = await _drive(sup, [_ev("flow1", Geo(centroid=NYC))])  # _MockAdapter: bypass False
+        sup._publish_event.assert_not_called()
+        assert a.published == set()
+        assert sup._dropped_publish == {"mock": 1}
+        assert sup._bypassed_publish == {}