feat(env): native adapters broadcast only newly-RECEIVED items (no backlog)

Replace the native path's "scan accumulated state + suppress what we've
already broadcast" model with "broadcast only what newly arrived from the
API this poll." Storage is unchanged (self._events + firms_pixels etc. are
populated for EVERY received item, so the LLM/get_active backlog is intact);
only the BROADCAST decision changes.

- env/store.py: per-adapter in-memory seen-set (_seen) + _seeded. First
  data-bearing poll for an adapter seeds keys and emits NOTHING (that batch
  is pre-existing backlog); later polls emit only keys not seen before.
  Restart => empty sets => next poll re-seeds silently. Structurally
  impossible to broadcast backlog on cold start / restart / re-enable.
  Key = external_id -> event_id -> content hash, namespaced per adapter.
  self._events[key]=evt still runs unconditionally (storage preserved).
- Fixes the ~175 (roads511) / ~782 (wzdx) cold-start bursts AND the latent
  quake/nws version (they only looked safe because Central pre-populated
  their broadcast tables).
- env/satpass.py: broadcast on AOS IMMINENCE (now < aos <= now+lead,
  broadcast_lead_seconds default 3600), future-only; window_hours still
  governs prediction depth. Strict norad_ids post-filter + fixed
  _parse_norad_ids char-iteration bug (cause of GOES/METEOR leak).
- Central path + broadcast-state tables untouched (native-only gate).

13 new tests; full suite at 10-failure baseline (1697 passed).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/config.py

@@ -513,6 +513,11 @@ class SatpassConfig(_SourcedFeed):
     # Predictor pass filters (used by the next task):
     min_elevation_deg: float = 10.0
     window_hours: int = 24
+    # BROADCAST imminence lead (seconds): a predicted pass is broadcast only
+    # once its AOS is within this near-term window (and still in the future).
+    # `window_hours` governs how far ahead we PREDICT; this governs when a
+    # predicted pass is actually announced. Default 60 min.
+    broadcast_lead_seconds: int = 3600
 
 
 @dataclass

work/meshai/env/satpass.py

@@ -60,6 +60,18 @@ logger = logging.getLogger(__name__)
 # pointless). Overridable via SatpassConfig.pass_refresh_seconds if present.
 DEFAULT_POLL_SECONDS = 900
 
+# Broadcast imminence lead. Satpass is NOT an API-delta feed — it computes the
+# SAME future passes every poll, so a pure "new since last poll" gate would
+# seed them all as backlog and then never broadcast. A pass's "newly received"
+# analog is IMMINENCE: it becomes broadcast-worthy only once its AOS first
+# enters this near-term lead window (and is still in the future). Combined with
+# the store's received-delta seen-set (keyed on the pass canonical id), each
+# pass then emits exactly once as it crosses into imminence, and the
+# first-poll-seed suppresses any startup burst. `window_hours` still governs
+# how far ahead we PREDICT (the schedule); this only governs the BROADCAST
+# trigger. Overridable via SatpassConfig.broadcast_lead_seconds.
+DEFAULT_BROADCAST_LEAD_SECONDS = 3600
+
 
 class SatpassAdapter:
     """Native SGP4 pass predictor — consolidates in-memory, gates synchronously."""
@@ -71,6 +83,9 @@ class SatpassAdapter:
             or DEFAULT_POLL_SECONDS)
         self._window_h = int(getattr(config, "window_hours", 24) or 24)
         self._min_el = float(getattr(config, "min_elevation_deg", 10.0))
+        self._lead_s = int(
+            getattr(config, "broadcast_lead_seconds", DEFAULT_BROADCAST_LEAD_SECONDS)
+            or DEFAULT_BROADCAST_LEAD_SECONDS)
         self._norad_ids = self._parse_norad_ids(
             getattr(config, "norad_ids", None) or [])
 
@@ -84,21 +99,39 @@ class SatpassAdapter:
 
     @staticmethod
     def _parse_norad_ids(raw) -> list[int]:
-        """Coerce a config norad_ids list (ints or GUI strings) to int list."""
+        """Coerce a config norad_ids value to a sorted unique int list.
+
+        Accepts a real list (ints or GUI strings) OR a single comma/space
+        separated string (as some GUI paths persist it). The string case
+        MUST be split first — iterating a bare string would walk its
+        CHARACTERS, turning "25544,33591" into garbage single-digit ids and
+        silently defeating the norad filter (a cause of non-listed sats
+        leaking into predictions).
+        """
+        if isinstance(raw, str):
+            raw = raw.replace(",", " ").split()
         return sorted({int(x) for x in raw if str(x).strip().isdigit()})
 
     def _resolve_tles(self) -> list[dict]:
         """Resolve the target TLE set from `sat_tles` (fresh only).
 
         `norad_ids` configured -> exactly those; empty -> all fresh.
+
+        When `norad_ids` is configured the result is ALSO post-filtered to
+        that set: this is belt-and-suspenders so that only configured NORADs
+        are ever predicted, regardless of which lookup path ran. Without a
+        configured list, the fetcher's `tle_groups` (e.g. "weather") stock
+        sat_tles with GOES/METEOR/FENGYUN etc.; the strict post-filter
+        guarantees those never leak into predictions once a list is set.
         """
         from meshai.central.tle_handler import get_fresh_tles, get_tle_by_norad
 
         if self._norad_ids:
+            allowed = set(self._norad_ids)
             out: list[dict] = []
             for nid in self._norad_ids:
                 tle = get_tle_by_norad(nid)
-                if tle is not None:
+                if tle is not None and int(tle["norad_id"]) in allowed:
                     out.append(tle)
             return out
         return get_fresh_tles()
@@ -186,6 +219,18 @@ class SatpassAdapter:
         staged: list[dict] = []
         for cid, recs in groups.items():
             consolidated = self._consolidate(cid, recs)
+            # IMMINENCE gate — applied on the CONSOLIDATED pass (earliest AOS
+            # across observers), so multi-observer consolidation is never
+            # fragmented. A pass is broadcast-worthy only once its AOS is both
+            # in the FUTURE and within the near-term lead window. Far-future
+            # passes are predicted (on the schedule) but not staged until they
+            # become imminent; a past AOS is never staged. This is satpass's
+            # "just received" signal — the store's received-delta seen-set then
+            # emits each pass exactly once as it crosses into the window, and
+            # the first-poll-seed prevents any startup burst.
+            aos = consolidated["aos_epoch"]
+            if aos <= now_epoch or aos > now_epoch + self._lead_s:
+                continue
             try:
                 result = sh.gate_consolidated_pass(consolidated, now=now_epoch)
             except Exception as e:

work/meshai/env/store.py

@@ -1,5 +1,7 @@
 """Environmental data store with tick-based adapter polling."""
 
+import hashlib
+import json
 import logging
 import time
 from typing import TYPE_CHECKING, Optional
@@ -29,6 +31,21 @@ class EnvironmentalStore:
         self._mesh_zones = config.nws_zones or []
         self._region_anchors = region_anchors or []
 
+        # ── Received-delta gate (NATIVE-only) ────────────────────────────
+        # Per-adapter set of item keys seen in PRIOR polls, plus the set of
+        # adapters whose first poll has completed. The model the operator
+        # demanded: a native adapter broadcasts an item ONLY when it was newly
+        # RECEIVED from the API this poll — never by scanning an accumulated
+        # backlog. So the FIRST poll for an adapter records every current key
+        # as "seen" and emits NOTHING (that batch is pre-existing backlog, not
+        # ours to announce); every later poll emits only keys not already
+        # seen. In-memory only (per process) — a restart empties the sets, so
+        # the next poll is a fresh "first poll" that re-seeds silently. This
+        # makes broadcasting a backlog item structurally impossible on cold
+        # start, restart, or after the adapter was disabled for days.
+        self._seen: dict[str, set] = {}   # adapter name -> set of item keys
+        self._seeded: set[str] = set()    # adapters past their first poll
+
         # Create adapter instances with error isolation
         self._register_adapter("nws", config.nws, ".nws", "NWSAlertsAdapter",
             lambda cfg: (cfg,))
@@ -112,46 +129,106 @@ class EnvironmentalStore:
         self._purge_expired()
         return changed
 
+    def _seen_key(self, name: str, raw_evt: dict) -> str:
+        """Derive a STABLE per-item key for the received-delta gate.
+
+        Stability across polls is the whole point: the SAME real-world item
+        must produce the SAME key every poll, or it would look "newly
+        received" forever and re-broadcast on each tick. Preference order,
+        most→least explicit id:
+          1. external_id — the upstream feed's own stable id (e.g. WZDx)
+          2. event_id    — the adapter's stable per-item id; every native raw
+             event carries one (the store's own dedup already keys on it, and
+             the adapters build it from natural ids: USGS quake id, 511 event
+             id, NWS alert id, ``swpc_<scale><level>``, ``ducting_<tier>_<loc>``,
+             ``avy_<center>_<zone>``, etc. — all stable across polls).
+          3. content hash — last-resort fallback if an item somehow carries no
+             id at all.
+        Namespaced by adapter so two feeds never cross-contaminate.
+        """
+        ext = raw_evt.get("external_id")
+        if ext:
+            return f"{name}\x1eext:{ext}"
+        eid = raw_evt.get("event_id")
+        if eid:
+            return f"{name}\x1eeid:{eid}"
+        blob = json.dumps(raw_evt, sort_keys=True, default=str)
+        return f"{name}\x1ehash:" + hashlib.sha1(blob.encode()).hexdigest()[:16]
+
+    def _delta_emit(self, name: str, adapter, raw_evt: dict, force: bool = False):
+        """Received-delta gate — emit ONLY items newly received THIS poll.
+
+        - First poll for ``name`` (name not yet in ``self._seeded``): record
+          the item's key as seen and emit NOTHING. That batch is the
+          pre-existing backlog.
+        - Later polls: emit only when the key is not already in the seen-set
+          (it just appeared upstream = "just received"); then record it.
+        - ``force=True`` (avalanche danger-level rise) re-emits an already-seen
+          item on a legitimate CONTENT change — but is IGNORED on the first
+          poll, so a restart still can never replay the backlog.
+
+        This replaces the native path's reliance on the deciders'
+        broadcast-state tables for the "is this new" decision. Legitimate
+        content filtering (severity/threshold/impact, decider gates) still runs
+        downstream in ``_emit_event``.
+        """
+        seen = self._seen.setdefault(name, set())
+        key = self._seen_key(name, raw_evt)
+        first_poll = name not in self._seeded
+
+        if first_poll:
+            seen.add(key)            # seed silently — this is backlog
+            return
+        if key in seen and not force:
+            return                   # already received in a prior poll
+        seen.add(key)
+
+        if self._event_bus is not None and hasattr(adapter, "to_event"):
+            self._emit_event(adapter, raw_evt)
+
     def _ingest(self, name: str, adapter):
-        """Ingest data from an adapter after it ticks."""
+        """Ingest data from an adapter after it ticks.
+
+        Emission goes through the received-delta gate (``_delta_emit``): the
+        adapter's FIRST data-bearing poll seeds the seen-set and broadcasts
+        nothing; only items that newly appear on later polls are broadcast.
+        ``self._events`` is still maintained for state/queries as before.
+        """
         if name == "swpc":
             self._swpc_status = adapter.get_status()
             # Also ingest any alert events (R-scale >= 3)
             for evt in adapter.get_events():
                 key = (evt["source"], evt["event_id"])
-                is_new = key not in self._events
                 self._events[key] = evt
-                if is_new and self._event_bus and hasattr(adapter, "to_event"):
-                    self._emit_event(adapter, evt)
+                self._delta_emit(name, adapter, evt)
         elif name == "ducting":
             self._ducting_status = adapter.get_status()
             for evt in adapter.get_events():
                 key = (evt["source"], evt["event_id"])
-                is_new = key not in self._events
                 self._events[key] = evt
-                if is_new and self._event_bus and hasattr(adapter, "to_event"):
-                    self._emit_event(adapter, evt)
+                self._delta_emit(name, adapter, evt)
         elif name == "avalanche":
-            # Avalanche: re-emit on danger_level rise (Update:) not just new events.
+            # Avalanche: re-emit on danger_level rise (Update:) not just new
+            # events. The rise is a legitimate CONTENT change, so it passes
+            # `force=True` — but the received-delta gate still suppresses it on
+            # the first poll (backlog stays silent).
             for evt in adapter.get_events():
                 key = (evt["source"], evt["event_id"])
                 prior = self._events.get(key)
-                is_new = prior is None
                 prior_level = prior.get("danger_level", -1) if prior else -1
-                level_rose = (not is_new) and (evt.get("danger_level", -1) > prior_level)
-
-                if (is_new or level_rose) and self._event_bus and hasattr(adapter, "to_event"):
-                    evt["_is_update"] = level_rose   # signal to to_event()
-                    self._emit_event(adapter, evt)
-
+                level_rose = (prior is not None) and (
+                    evt.get("danger_level", -1) > prior_level)
+                evt["_is_update"] = level_rose   # signal to to_event()
+                self._delta_emit(name, adapter, evt, force=level_rose)
                 self._events[key] = evt   # always update stored state
         else:
             for evt in adapter.get_events():
                 key = (evt["source"], evt["event_id"])
-                is_new = key not in self._events
                 self._events[key] = evt
-                if is_new and self._event_bus and hasattr(adapter, "to_event"):
-                    self._emit_event(adapter, evt)
+                self._delta_emit(name, adapter, evt)
+
+        # First poll for this adapter is now complete: later polls may emit.
+        self._seeded.add(name)
 
     def _emit_event(self, adapter, raw_evt: dict):
         """Convert raw event to pipeline Event and emit to bus.

work/tests/test_firms_native_fusion.py

@@ -330,6 +330,13 @@ class TestAdapterTickFusion:
         store = EnvironmentalStore.__new__(EnvironmentalStore)
         store._events = {}
         store._event_bus = bus
+        # Received-delta gate state. This test exercises STEADY-STATE fusion
+        # routing (fusion reaches the bus, raw hotspots never do), so mark firms
+        # as already past its first-poll seed — otherwise the store's
+        # received-delta gate would (correctly) seed this first batch silently.
+        # First-poll seed-silence is covered by test_store_received_delta.
+        store._seen = {}
+        store._seeded = {"firms"}
         assert a.tick() is True
         store._ingest("firms", a)
 

work/tests/test_satpass_native.py

@@ -162,8 +162,10 @@ def test_second_tick_does_not_rebroadcast_after_commit(monkeypatch):
     commit = staged[0]["data"]["_on_broadcast_committed"]
     commit(float(T0))  # marks satpass_events.last_broadcast_at
 
-    # Tick 2 (interval elapsed): same canonical pass must be suppressed.
-    assert adapter.tick(now=T0 + 2000) is False
+    # Tick 2 (interval elapsed): the pass is STILL imminent (AOS 30 min out,
+    # inside the 60-min lead) so the imminence gate would stage it — but the
+    # satpass_events commit from tick 1 must suppress the re-broadcast.
+    assert adapter.tick(now=T0 - 1800) is False
     assert adapter.get_events() == []
 
 
@@ -176,8 +178,9 @@ def test_second_tick_without_commit_is_not_deduped(monkeypatch):
 
     adapter = _adapter()
     assert adapter.tick(now=T0 - 3600) is True
-    # No commit fired -> last_broadcast_at still NULL -> re-stages next tick.
-    assert adapter.tick(now=T0 + 2000) is True
+    # No commit fired -> last_broadcast_at still NULL. The pass is still
+    # imminent at T0-1800 (AOS 30 min out) so it re-stages next tick.
+    assert adapter.tick(now=T0 - 1800) is True
 
 
 # ══════════════════════════════════════════════════════════════════════
@@ -305,3 +308,107 @@ def test_central_consolidate_feeds_shared_gate_merged(monkeypatch):
         "SELECT COUNT(*) AS n FROM satpass_pending WHERE consolidated_id=?",
         (cid,)).fetchone()
     assert left["n"] == 0
+
+
+# ══════════════════════════════════════════════════════════════════════
+# 6. IMMINENCE BROADCAST TRIGGER (satpass's "just received" analog)
+# ══════════════════════════════════════════════════════════════════════
+#
+# Satpass is NOT an API-delta feed — it recomputes the SAME future passes
+# every poll. Its "newly received" signal is IMMINENCE: a pass is staged for
+# broadcast only once its AOS is both in the FUTURE and within the near-term
+# lead window (default 60 min). A far-future pass is predicted (on schedule)
+# but not staged; a past AOS is never staged. Combined with the store's
+# received-delta seen-set, each pass then emits exactly once as it crosses in.
+
+# The fixed mock pass sits at AOS=T0 (both observers consolidate to one).
+
+def test_far_future_pass_not_staged(monkeypatch):
+    _enable_satpass_db(dry_run=False)
+    _seed_iss_tle()
+    _patch_observers(monkeypatch, [_BOISE, _TWIN])
+    _patch_predictor(monkeypatch, _two_observer_pass)
+
+    adapter = _adapter()  # default broadcast_lead_seconds = 3600 (60 min)
+    # AOS=T0 is 4 hours ahead of now -> well outside the 60-min lead window.
+    assert adapter.tick(now=T0 - 4 * 3600) is False
+    assert adapter.get_events() == []
+
+
+def test_imminent_pass_is_staged(monkeypatch):
+    _enable_satpass_db(dry_run=False)
+    _seed_iss_tle()
+    _patch_observers(monkeypatch, [_BOISE, _TWIN])
+    _patch_predictor(monkeypatch, _two_observer_pass)
+
+    adapter = _adapter()
+    # AOS=T0 is 30 min ahead of now -> inside the 60-min lead window.
+    assert adapter.tick(now=T0 - 1800) is True
+    staged = adapter.get_events()
+    assert len(staged) == 1
+    assert staged[0]["event_id"] == f"25544:{T0 // 3600}"
+
+
+def test_past_aos_never_staged(monkeypatch):
+    _enable_satpass_db(dry_run=False)
+    _seed_iss_tle()
+    _patch_observers(monkeypatch, [_BOISE, _TWIN])
+    _patch_predictor(monkeypatch, _two_observer_pass)
+
+    adapter = _adapter()
+    # now is 10 min AFTER AOS=T0 -> the pass is in the past, never staged.
+    assert adapter.tick(now=T0 + 600) is False
+    assert adapter.get_events() == []
+
+
+def test_configurable_lead_window(monkeypatch):
+    _enable_satpass_db(dry_run=False)
+    _seed_iss_tle()
+    _patch_observers(monkeypatch, [_BOISE, _TWIN])
+    _patch_predictor(monkeypatch, _two_observer_pass)
+
+    # With a 3-hour lead, an AOS 2 hours out is now imminent.
+    adapter = _adapter(broadcast_lead_seconds=3 * 3600)
+    assert adapter.tick(now=T0 - 2 * 3600) is True
+    assert len(adapter.get_events()) == 1
+
+
+# ══════════════════════════════════════════════════════════════════════
+# 7. NORAD_IDS FILTER IS STRICT (non-listed sats never predicted)
+# ══════════════════════════════════════════════════════════════════════
+
+# A non-ISS TLE (a stand-in for a GOES/METEOR/FENGYUN weather sat the fetcher
+# stocks via tle_groups). Different NORAD so a filter break would surface as a
+# second consolidated pass with a different canonical id.
+OTHER_L1 = "1 43226U 18022A   26182.50000000  .00000000  00000-0  00000-0 0  9999"
+OTHER_L2 = "2 43226   0.0300 100.0000 0001000  90.0000 270.0000  1.00270000 12345"
+
+
+def _seed_other_tle():
+    conn = get_db()
+    fresh = datetime.now(timezone.utc).isoformat()
+    upsert_tle(conn, 43226, "GOES-17", OTHER_L1, OTHER_L2, fresh)
+
+
+def test_norad_ids_filter_excludes_non_listed_sats(monkeypatch):
+    _enable_satpass_db(dry_run=False)
+    _seed_iss_tle()     # 25544 — configured
+    _seed_other_tle()   # 43226 — fresh but NOT in norad_ids
+    _patch_observers(monkeypatch, [_BOISE, _TWIN])
+    # Predictor would return a pass for ANY satellite (keyed on observer lat),
+    # so if 43226 leaked through it would produce a second staged pass.
+    _patch_predictor(monkeypatch, _two_observer_pass)
+
+    adapter = _adapter()  # norad_ids=[25544]
+    assert adapter.tick(now=T0 - 1800) is True
+    staged = adapter.get_events()
+    # Only the configured NORAD is predicted, despite 43226 being fresh.
+    assert len(staged) == 1
+    assert staged[0]["norad_id"] == 25544
+
+
+def test_parse_norad_ids_handles_comma_string():
+    # A GUI-persisted comma string must NOT be char-iterated into garbage ids.
+    assert SatpassAdapter._parse_norad_ids("25544, 33591") == [25544, 33591]
+    assert SatpassAdapter._parse_norad_ids([25544, "33591"]) == [25544, 33591]
+    assert SatpassAdapter._parse_norad_ids([]) == []

work/tests/test_store_received_delta.py

@@ -0,0 +1,156 @@
+"""Received-delta gate tests for the native EnvironmentalStore path.
+
+The operator's required model: a native adapter broadcasts an item ONLY when
+it was newly RECEIVED from the API this poll — never by scanning an accumulated
+backlog. The store enforces this with an in-memory, per-adapter seen-set:
+
+  * the FIRST data-bearing poll for an adapter records every current item key
+    as "seen" and broadcasts NOTHING (that batch is pre-existing backlog);
+  * every later poll broadcasts only items whose key is not already seen;
+  * a fresh store (process restart) has empty sets, so its next poll is again a
+    "first poll" that re-seeds silently — backlog is never re-broadcast.
+
+These tests drive the real EnvironmentalStore + EventBus with a fake adapter
+whose per-poll batch we control, and assert exactly which events reach the bus.
+"""
+from __future__ import annotations
+
+from meshai.env.store import EnvironmentalStore
+from meshai.config import EnvironmentalConfig
+from meshai.notifications.pipeline.bus import EventBus
+from meshai.notifications.events import make_event
+
+
+class _FakeAdapter:
+    """Native-adapter stand-in: returns a controllable batch of raw events.
+
+    Raw events carry the (source, event_id) shape every native adapter emits,
+    which is exactly what the store's seen-set keys on. Each poll's batch is
+    set by the test via `set_batch`.
+    """
+
+    def __init__(self, source: str = "roads511"):
+        self._source = source
+        self._batch: list[dict] = []
+
+    def set_batch(self, ids: list[str]) -> None:
+        self._batch = [
+            {"source": self._source, "event_id": eid, "fetched_at": 0}
+            for eid in ids
+        ]
+
+    def tick(self) -> bool:
+        # Data is "fetched" every poll; the store decides what is new.
+        return True
+
+    def get_events(self) -> list:
+        return list(self._batch)
+
+    def to_event(self, raw_evt: dict):
+        eid = raw_evt["event_id"]
+        return make_event(
+            source=raw_evt["source"],
+            category="test_delta",   # no decider registered -> emits directly
+            severity="routine",
+            title=eid,
+            summary=eid,
+            group_key=eid,
+        )
+
+
+def _make_store(adapter_name: str = "roads511"):
+    """Build a store with NO real adapters, then inject one fake adapter."""
+    bus = EventBus()
+    captured: list = []
+    bus.subscribe(lambda e: captured.append(e))
+
+    # All feeds default disabled -> zero native adapters register.
+    store = EnvironmentalStore(EnvironmentalConfig(), event_bus=bus)
+    adapter = _FakeAdapter(source="511")
+    store._adapters[adapter_name] = adapter
+    return store, adapter, captured
+
+
+def _emitted_ids(captured) -> list[str]:
+    return [e.title for e in captured]
+
+
+def test_first_poll_seeds_and_broadcasts_nothing():
+    store, adapter, captured = _make_store()
+    adapter.set_batch(["A", "B", "C"])
+
+    store.refresh()  # poll 1 — the backlog
+
+    assert captured == [], "first poll must broadcast NOTHING (backlog seed)"
+
+
+def test_second_poll_emits_only_newly_received():
+    store, adapter, captured = _make_store()
+
+    adapter.set_batch(["A", "B", "C"])
+    store.refresh()                       # poll 1: seed
+    assert _emitted_ids(captured) == []
+
+    adapter.set_batch(["A", "B", "C", "D"])
+    store.refresh()                       # poll 2: only D is new
+    assert _emitted_ids(captured) == ["D"]
+
+
+def test_unchanged_poll_emits_nothing():
+    store, adapter, captured = _make_store()
+
+    adapter.set_batch(["A", "B", "C"])
+    store.refresh()                       # poll 1: seed
+    adapter.set_batch(["A", "B", "C", "D"])
+    store.refresh()                       # poll 2: D
+    adapter.set_batch(["A", "B", "C", "D"])
+    store.refresh()                       # poll 3: nothing new
+
+    assert _emitted_ids(captured) == ["D"], "poll 3 has no new items"
+
+
+def test_restart_reseeds_and_never_rebroadcasts_backlog():
+    # Process 1 sees A,B,C,D and broadcasts D.
+    store1, adapter1, cap1 = _make_store()
+    adapter1.set_batch(["A", "B", "C"])
+    store1.refresh()
+    adapter1.set_batch(["A", "B", "C", "D"])
+    store1.refresh()
+    assert _emitted_ids(cap1) == ["D"]
+
+    # RESTART: a fresh store has an empty seen-set. The SAME backlog [A,B,C,D]
+    # arriving on its first poll must be re-seeded silently, not re-broadcast.
+    store2, adapter2, cap2 = _make_store()
+    adapter2.set_batch(["A", "B", "C", "D"])
+    store2.refresh()
+    assert cap2 == [], "restart must NEVER re-broadcast the existing backlog"
+
+    # And a genuinely new item after the restart still broadcasts once.
+    adapter2.set_batch(["A", "B", "C", "D", "E"])
+    store2.refresh()
+    assert _emitted_ids(cap2) == ["E"]
+
+
+def test_stable_key_prevents_reemit_when_batch_reorders():
+    # The same real-world items in a different order are NOT "newly received".
+    store, adapter, captured = _make_store()
+    adapter.set_batch(["A", "B", "C"])
+    store.refresh()                       # seed
+    adapter.set_batch(["C", "A", "B"])    # reordered, same items
+    store.refresh()
+    assert captured == [], "reordering the same items emits nothing"
+
+
+def test_disabled_for_days_then_backlog_is_not_broadcast():
+    # Simulate an adapter that was off for days: its first poll after coming
+    # back returns a large accumulated backlog. None of it may broadcast.
+    store, adapter, captured = _make_store()
+    backlog = [f"evt{i}" for i in range(200)]
+    adapter.set_batch(backlog)
+    store.refresh()                       # first poll after re-enable
+    assert captured == [], "a days-old backlog is seeded silently, never sent"
+
+    # Only a truly new arrival afterward is announced.
+    adapter.set_batch(backlog + ["fresh"])
+    store.refresh()
+    assert _emitted_ids(captured) == ["fresh"]