fix(env): leak-proof received-delta — durable persistent baseline + non-empty seed guard (#44)

Two live backlog-broadcast leaks traced to the in-memory first-poll seed:
(1) incremental-fetch adapters (wzdx: registry tick [0 events] then feeds
tick [many]) got marked _seeded on the EMPTY first tick, so the real batch
next tick all looked "new" and broadcast; (2) in-memory seed lost on restart.

Fix — durable baseline + guard:
- _seed_from_persistent() at store init: pre-load already-received item keys
  from the persistent hazard tables into self._seen, so nothing ever received
  can re-broadcast (immune to fetch staging + restart). Only sources whose
  native emit key PROVABLY equals a persistent key are durably seeded:
  wzdx (traffic_events.external_id) + usgs_quake (quake_events.event_id).
  Resilient (per-table try/except; missing table -> skip).
- _seen_key() now namespaces by evt["source"] (matches persistent tables),
  via shared _key_ext/_key_eid helpers used by both seed and live emit so
  they can't drift.
- non-empty-seed guard: _ingest marks only sources that carried >=1 event
  this poll as _seeded -> an empty first tick can never seed-then-leak. This
  is the root-cause fix; covers all adapters (roads511/traffic fetch
  atomically per tick, so the guard fully protects them).
- storage untouched (self._events populated for every event); Central
  path/deciders untouched.

Live-DB verified: seed pre-loads 784 wzdx + 8 quake keys -> a live wzdx poll
of 784 known zones broadcasts 0. +6 tests (incremental staging, restart,
persistent-preseed, fresh-DB fallback); suite at 10-failure baseline.

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/env/store.py

@@ -12,6 +12,23 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+# ── Received-delta key format ────────────────────────────────────────────────
+# A single, shared key format used by BOTH the live emit path (_seen_key) and
+# the durable startup pre-seed (_seed_from_persistent). Keeping the two on one
+# helper guarantees they can never drift apart — a drift would silently defeat
+# the pre-seed (keys wouldn't match), which is exactly the leak we are fixing.
+_SEP = "\x1e"
+
+
+def _key_ext(source: str, external_id) -> str:
+    """Seen-key for an item identified by its upstream external_id."""
+    return f"{source}{_SEP}ext:{external_id}"
+
+
+def _key_eid(source: str, event_id) -> str:
+    """Seen-key for an item identified by the adapter's stable event_id."""
+    return f"{source}{_SEP}eid:{event_id}"
+
 
 class EnvironmentalStore:
     """Cache and tick-driver for all environmental feed adapters."""
@@ -32,19 +49,31 @@ class EnvironmentalStore:
         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
+        # The model the operator demanded: a native adapter broadcasts an item
-        # adapters whose first poll has completed. The model the operator
+        # ONLY when it was newly RECEIVED from the API this poll — never by
-        # demanded: a native adapter broadcasts an item ONLY when it was newly
+        # scanning an accumulated backlog. Two layers enforce this:
-        # RECEIVED from the API this poll — never by scanning an accumulated
+        #
-        # backlog. So the FIRST poll for an adapter records every current key
+        #   1. DURABLE pre-seed (_seed_from_persistent, below). meshai already
-        # as "seen" and emits NOTHING (that batch is pre-existing backlog, not
+        #      has a durable record of everything it has ever received: the
-        # ours to announce); every later poll emits only keys not already
+        #      persistent hazard tables. At startup we load the identifying
-        # seen. In-memory only (per process) — a restart empties the sets, so
+        #      keys of every already-received item into `self._seen`, so nothing
-        # the next poll is a fresh "first poll" that re-seeds silently. This
+        #      ever received can re-broadcast — immune to fetch staging AND to
-        # makes broadcasting a backlog item structurally impossible on cold
+        #      restarts. Keyed by the event's `source` (NOT adapter name) so the
-        # start, restart, or after the adapter was disabled for days.
+        #      seed and the live-emit key line up on the same value.
-        self._seen: dict[str, set] = {}   # adapter name -> set of item keys
+        #   2. IN-MEMORY first-poll seed. The FIRST *non-empty* poll for a
-        self._seeded: set[str] = set()    # adapters past their first poll
+        #      source records every current key as "seen" and emits NOTHING
+        #      (that batch is pre-existing backlog); later polls emit only keys
+        #      not already seen. Belt-and-suspenders: a source is marked
+        #      "seeded" ONLY after a non-empty ingest, so an empty first poll
+        #      (e.g. wzdx's registry-only tick) can never mark it seeded and
+        #      then leak the real batch on the next tick.
+        #
+        # `self._seen` is keyed by SOURCE (evt["source"]), matching the durable
+        # pre-seed. Durable keys added at startup are never removed, so they
+        # suppress a backlog item even AFTER the source is marked seeded (this
+        # is what closes cross-tick, non-empty staging leaks).
+        self._seen: dict[str, set] = {}   # source -> set of item keys
+        self._seeded: set[str] = set()    # sources past their first non-empty poll
 
         # Create adapter instances with error isolation
         self._register_adapter("nws", config.nws, ".nws", "NWSAlertsAdapter",
@@ -97,6 +126,11 @@ class EnvironmentalStore:
             logger.warning("Failed adapters: %s", list(self._failed_adapters.keys()))
         logger.info(f"EnvironmentalStore initialized with {len(self._adapters)} adapters")
 
+        # Durable pre-seed: load every already-received item key from the
+        # persistent hazard tables into `self._seen` so nothing ever received
+        # can re-broadcast, regardless of fetch staging or process restarts.
+        self._seed_from_persistent()
+
 
     def _register_adapter(self, name: str, cfg, module_path: str, class_name: str, args_fn):
         """Register a single adapter with error isolation."""
@@ -129,7 +163,7 @@ class EnvironmentalStore:
         self._purge_expired()
         return changed
 
-    def _seen_key(self, name: str, raw_evt: dict) -> str:
+    def _seen_key(self, 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
@@ -144,16 +178,23 @@ class EnvironmentalStore:
              ``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.
+
+        Namespaced by the event's ``source`` (NOT the adapter name). The
+        persistent hazard tables key on ``source`` (e.g. ``wzdx``), so keying
+        the live-emit path on the same value is what lets the durable startup
+        pre-seed line up with what the adapter emits. Every native raw event
+        carries a stable ``source``; if one somehow does not, fall back to a
+        neutral namespace so two feeds still never cross-contaminate.
         """
+        source = raw_evt.get("source") or "?"
         ext = raw_evt.get("external_id")
         if ext:
-            return f"{name}\x1eext:{ext}"
+            return _key_ext(source, ext)
         eid = raw_evt.get("event_id")
         if eid:
-            return f"{name}\x1eeid:{eid}"
+            return _key_eid(source, eid)
         blob = json.dumps(raw_evt, sort_keys=True, default=str)
-        return f"{name}\x1ehash:" + hashlib.sha1(blob.encode()).hexdigest()[:16]
+        return f"{source}{_SEP}hash:" + 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.
@@ -172,15 +213,16 @@ class EnvironmentalStore:
         content filtering (severity/threshold/impact, decider gates) still runs
         downstream in ``_emit_event``.
         """
-        seen = self._seen.setdefault(name, set())
+        source = raw_evt.get("source") or name
-        key = self._seen_key(name, raw_evt)
+        seen = self._seen.setdefault(source, set())
-        first_poll = name not in self._seeded
+        key = self._seen_key(raw_evt)
+        first_poll = source 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
+            return                   # already received (prior poll OR durable pre-seed)
         seen.add(key)
 
         if self._event_bus is not None and hasattr(adapter, "to_event"):
@@ -193,19 +235,30 @@ class EnvironmentalStore:
         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.
+
+        Belt-and-suspenders: a source is marked "seeded" (past its first poll)
+        ONLY after a NON-EMPTY ingest for that source. An empty poll — e.g.
+        wzdx's registry-only tick that yields 0 events but returns ``tick()``
+        True — must never mark a source seeded, or the next tick's real batch
+        would be treated as "new" and leak. We collect the sources that
+        actually carried ≥1 event this ingest and mark only those.
         """
+        touched: set[str] = set()
+
         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"])
                 self._events[key] = evt
+                touched.add(evt.get("source") or name)
                 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"])
                 self._events[key] = evt
+                touched.add(evt.get("source") or name)
                 self._delta_emit(name, adapter, evt)
         elif name == "avalanche":
             # Avalanche: re-emit on danger_level rise (Update:) not just new
@@ -219,16 +272,118 @@ class EnvironmentalStore:
                 level_rose = (prior is not None) and (
                     evt.get("danger_level", -1) > prior_level)
                 evt["_is_update"] = level_rose   # signal to to_event()
+                touched.add(evt.get("source") or name)
                 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"])
                 self._events[key] = evt
+                touched.add(evt.get("source") or name)
                 self._delta_emit(name, adapter, evt)
 
-        # First poll for this adapter is now complete: later polls may emit.
+        # First (non-empty) poll for these sources is now complete: later polls
-        self._seeded.add(name)
+        # may emit. Empty ingests touch nothing and so never mark a source
+        # seeded — the leak-proof invariant behind the wzdx staging fix.
+        self._seeded.update(touched)
+
+    def _seed_from_persistent(self) -> None:
+        """Pre-seed ``self._seen`` from the durable hazard tables at startup.
+
+        A row exists in these tables iff meshai has ALREADY RECEIVED that item,
+        so loading their identifying keys guarantees a native adapter can never
+        re-broadcast a previously-received item — immune to fetch staging and
+        to process restarts. Durable keys are added to ``self._seen`` and,
+        when ≥1 key is loaded for a source, that source is marked ``_seeded``
+        so the durable set becomes its baseline: the very next poll emits ONLY
+        items NOT already received (the operator's "newly received → send it
+        now") and suppresses everything in the table — no silent first-poll
+        needed, and no dependency on how the fetch is staged. Because durable
+        keys are never removed, they keep suppressing backlog across every
+        later poll, which is what closes cross-tick, non-empty staging leaks.
+        (When a source has 0 durable rows we do NOT mark it seeded, so it falls
+        back to the leak-proof in-memory silent first-poll seed — a fresh DB
+        can never cause the first real batch to be treated as new.)
+
+        Only sources whose native emit key PROVABLY equals the persistent key
+        are seeded here (verified against the live schema + adapter code):
+
+          * ``wzdx``       — the native WZDx adapter carries a stable
+            ``external_id`` on its raw event and the incident decider persists
+            it as ``traffic_events(source='wzdx', external_id)``. Same value on
+            both sides → ``_key_ext('wzdx', external_id)`` matches exactly.
+          * ``usgs_quake`` — the native adapter's raw ``event_id`` is the bare
+            USGS id (e.g. ``us6000t9bn``) and the quake decider persists it as
+            ``quake_events.event_id`` verbatim → ``_key_eid('usgs_quake', id)``
+            matches exactly.
+
+        DELIBERATELY NOT seeded here (native emit key ≠ any persistent key —
+        see the report; these stay protected by the in-memory first-poll +
+        non-empty-seed guard, which is sufficient because each fetches
+        atomically per tick rather than across ticks):
+          roads511 / traffic (persistent rows are Central-keyed:
+          itd_511/tomtom_incidents with idaho_511:event:* external_ids, but the
+          native adapters emit source '511'/'traffic' with derived event_ids),
+          fires (native ``nifc_<name>_<state>`` vs persistent IRWIN GUID),
+          firms (native ``firms_<lat>_<lon>_<date>_<time>`` — no matching PK),
+          satpass (native ``<norad>:<bucket>`` vs persistent
+          ``<norad>:<observer>:<bucket>``), nws (native CAP url vs persistent
+          bare urn:oid), swpc (native ``swpc_<scale><level>`` vs timestamped
+          persistent ids), usgs hydro (time-series, no per-item PK).
+
+        Resilient by construction: each table is loaded in its own try/except
+        so a missing/renamed table (or an unavailable DB) logs and continues —
+        startup never crashes. The SELECTs are cheap key-only scans of bounded
+        tables.
+        """
+        try:
+            from meshai.persistence import get_db
+            conn = get_db()
+        except Exception as e:
+            logger.warning("received-delta pre-seed skipped (DB unavailable): %s", e)
+            return
+
+        # (source, description, SQL, row->key) — one spec per verified source.
+        specs = [
+            (
+                "wzdx",
+                "traffic_events(source='wzdx')",
+                "SELECT external_id FROM traffic_events "
+                "WHERE source='wzdx' AND external_id IS NOT NULL",
+                lambda row: _key_ext("wzdx", row[0]),
+            ),
+            (
+                "usgs_quake",
+                "quake_events",
+                "SELECT event_id FROM quake_events WHERE event_id IS NOT NULL",
+                lambda row: _key_eid("usgs_quake", row[0]),
+            ),
+        ]
+
+        total = 0
+        for source, desc, sql, key_fn in specs:
+            try:
+                seen = self._seen.setdefault(source, set())
+                n = 0
+                for row in conn.execute(sql).fetchall():
+                    seen.add(key_fn(row))
+                    n += 1
+                total += n
+                if n:
+                    # Durable baseline established → skip the silent first-poll
+                    # seed; the next poll emits only genuinely-new items.
+                    self._seeded.add(source)
+                logger.info(
+                    "received-delta pre-seed: %d key(s) from %s -> source %r%s",
+                    n, desc, source, " (marked seeded)" if n else "",
+                )
+            except Exception as e:
+                # Missing/renamed table or bad row — never fatal at startup.
+                logger.warning(
+                    "received-delta pre-seed: %s failed (continuing): %s",
+                    desc, e,
+                )
+        logger.info("received-delta pre-seed complete: %d durable key(s) loaded", total)
 
     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

@@ -335,8 +335,12 @@ class TestAdapterTickFusion:
         # 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.
+        # The gate keys on the event's ``source`` (not the adapter name), and
+        # this adapter emits BOTH "firms" (raw hotspots) and "firms_fusion"
+        # (consolidated growth) — mark both seeded, exactly as a non-empty
+        # ingest would (store._ingest marks every source it touched).
         store._seen = {}
-        store._seeded = {"firms"}
+        store._seeded = {"firms", "firms_fusion"}
         assert a.tick() is True
         store._ingest("firms", a)
 

work/tests/test_store_received_delta.py

@@ -154,3 +154,238 @@ def test_disabled_for_days_then_backlog_is_not_broadcast():
     adapter.set_batch(backlog + ["fresh"])
     store.refresh()
     assert _emitted_ids(captured) == ["fresh"]
+
+
+# ═══════════════════════════════════════════════════════════════════════════
+# Durable persistent pre-seed (the robust anti-leak fix)
+#
+# meshai already has a durable record of everything it has ever received: the
+# persistent hazard tables. At startup the store loads their identifying keys
+# into self._seen so nothing ever received can re-broadcast, immune to how the
+# fetch is staged and immune to restarts. These tests use the REAL persistent
+# DB (conftest points MESHAI_DB_PATH at a fresh migrated tmp file per test).
+# ═══════════════════════════════════════════════════════════════════════════
+from meshai.persistence import get_db
+
+
+class _FakeWZDx:
+    """Native-WZDx stand-in. Its raw events carry a stable ``external_id``
+    (like the real adapter), so the seen-key is ``wzdx\x1eext:<id>`` — exactly
+    what the durable pre-seed loads from traffic_events(source='wzdx')."""
+
+    def __init__(self):
+        self._batch: list[dict] = []
+
+    def set_batch(self, ext_ids: list[str]) -> None:
+        self._batch = [
+            {"source": "wzdx", "event_id": f"wzdx_{x}",
+             "external_id": x, "fetched_at": 0}
+            for x in ext_ids
+        ]
+
+    def tick(self) -> bool:
+        # Real wzdx.tick() returns True even on a 0-event (registry-only) tick.
+        return True
+
+    def get_events(self) -> list:
+        return list(self._batch)
+
+    def to_event(self, raw_evt: dict):
+        eid = raw_evt["external_id"]
+        return make_event(source="wzdx", category="test_delta",
+                          severity="routine", title=eid, summary=eid,
+                          group_key=eid)
+
+
+class _FakeQuake:
+    """Native usgs_quake stand-in. Raw events have NO external_id, so the
+    seen-key falls to ``usgs_quake\x1eeid:<event_id>`` — exactly what the
+    durable pre-seed loads from quake_events.event_id."""
+
+    def __init__(self):
+        self._batch: list[dict] = []
+
+    def set_batch(self, event_ids: list[str]) -> None:
+        self._batch = [
+            {"source": "usgs_quake", "event_id": e, "fetched_at": 0}
+            for e in event_ids
+        ]
+
+    def tick(self) -> bool:
+        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="usgs_quake", category="test_delta",
+                          severity="routine", title=eid, summary=eid,
+                          group_key=eid)
+
+
+def _insert_traffic(external_ids: list[str], source: str = "wzdx") -> None:
+    conn = get_db()
+    for x in external_ids:
+        conn.execute(
+            "INSERT OR IGNORE INTO traffic_events"
+            "(source, external_id, first_seen_at, last_seen_at) "
+            "VALUES (?,?,?,?)",
+            (source, x, 0, 0),
+        )
+
+
+def _insert_quake(event_ids: list[str]) -> None:
+    conn = get_db()
+    for e in event_ids:
+        conn.execute(
+            "INSERT OR IGNORE INTO quake_events(event_id, first_seen_at) "
+            "VALUES (?,?)",
+            (e, 0),
+        )
+
+
+def _build_store(adapter_name: str, adapter):
+    """Construct a store (runs the durable pre-seed against the current DB),
+    then inject a fake adapter. Insert persistent rows BEFORE calling this."""
+    bus = EventBus()
+    captured: list = []
+    bus.subscribe(lambda e: captured.append(e))
+    store = EnvironmentalStore(EnvironmentalConfig(), event_bus=bus)
+    store._adapters[adapter_name] = adapter
+    return store, captured
+
+
+def test_persistent_preseed_known_suppressed_new_emitted():
+    # N durable rows in traffic_events(source=wzdx). A fresh store must treat
+    # them as already-received: a poll of those same N broadcasts NOTHING; a
+    # poll adding one external_id NOT in the table broadcasts only that one.
+    known = [f"z{i}" for i in range(5)]
+    _insert_traffic(known)
+
+    adapter = _FakeWZDx()
+    store, captured = _build_store("wzdx", adapter)
+
+    # wzdx was pre-seeded from the durable table AND marked seeded (baseline).
+    assert "wzdx" in store._seeded
+    assert len(store._seen["wzdx"]) == 5
+
+    adapter.set_batch(known)
+    store.refresh()
+    assert captured == [], "all 5 are durably-known → zero broadcast"
+
+    adapter.set_batch(known + ["z_new"])
+    store.refresh()
+    assert _emitted_ids(captured) == ["z_new"], "only the not-in-table id broadcasts"
+
+
+def test_persistent_preseed_cross_tick_staging_no_leak():
+    # The scenario the DURABLE seed uniquely fixes: a source is marked seeded
+    # on a partial tick, then a LATER tick brings a backlog item that was never
+    # seen in-process. Without the durable record it would leak.
+    _insert_traffic(["A", "B"])          # both already RECEIVED (durable)
+    adapter = _FakeWZDx()
+    store, captured = _build_store("wzdx", adapter)
+
+    adapter.set_batch(["A"])
+    store.refresh()                       # tick 1: only A present
+    adapter.set_batch(["A", "B"])
+    store.refresh()                       # tick 2: B appears (backlog)
+    assert captured == [], "B is durably-known — must NOT leak on a later tick"
+
+    # CONTROL: identical staging but NO durable rows → B leaks (proves the
+    # durable seed is what prevents it; in-memory alone cannot).
+    ctrl = _FakeWZDx()
+    bus = EventBus(); cap2: list = []
+    bus.subscribe(lambda e: cap2.append(e))
+    # A separate source name so its 0-row durable seed doesn't mark it seeded.
+    ctrl._batch = []
+    store2 = EnvironmentalStore(EnvironmentalConfig(), event_bus=bus)
+    store2._adapters["wzdx_ctrl"] = ctrl
+    ctrl.set_batch(["A"])
+    # Re-point ctrl events to a fresh source with no durable rows.
+    for e in ctrl._batch:
+        e["source"] = "wzdx_ctrl"
+    store2.refresh()
+    ctrl.set_batch(["A", "B"])
+    for e in ctrl._batch:
+        e["source"] = "wzdx_ctrl"
+    store2.refresh()
+    assert [e.title for e in cap2] == ["B"], "without a durable record, B leaks"
+
+
+def test_incremental_empty_first_tick_then_only_new_broadcasts():
+    # Task's incremental case: tick 1 yields [] (e.g. wzdx registry-only tick),
+    # tick 2 yields [A,B,C] where A,B are durably-known and C is new.
+    # The empty tick must not mark-and-leak; the durable seed catches A,B; only
+    # C — genuinely never received — broadcasts.
+    _insert_traffic(["A", "B"])
+    adapter = _FakeWZDx()
+    store, captured = _build_store("wzdx", adapter)
+
+    adapter.set_batch([])                 # empty first tick
+    store.refresh()
+    assert captured == [], "empty tick emits nothing"
+
+    adapter.set_batch(["A", "B", "C"])    # backlog A,B + new C
+    store.refresh()
+    assert _emitted_ids(captured) == ["C"], "only the never-received C broadcasts"
+
+
+def test_restart_against_same_persistent_db_never_rebroadcasts():
+    # A full received backlog is durable. Process 1 broadcasts nothing for it.
+    # After a RESTART (fresh store, same DB) the backlog still never broadcasts.
+    backlog = ["A", "B", "C", "D"]
+    _insert_traffic(backlog)
+
+    a1 = _FakeWZDx()
+    store1, cap1 = _build_store("wzdx", a1)
+    a1.set_batch(backlog)
+    store1.refresh()
+    assert cap1 == [], "process 1: durable backlog is silent"
+
+    # RESTART: brand-new store, same persistent DB → pre-seed reloads.
+    a2 = _FakeWZDx()
+    store2, cap2 = _build_store("wzdx", a2)
+    a2.set_batch(backlog)
+    store2.refresh()
+    assert cap2 == [], "restart must NEVER re-broadcast the durable backlog"
+
+    a2.set_batch(backlog + ["E"])
+    store2.refresh()
+    assert _emitted_ids(cap2) == ["E"], "a genuinely-new item still broadcasts once"
+
+
+def test_persistent_preseed_quake_by_event_id():
+    # Durable seed for the event_id-keyed path (no external_id): quake_events.
+    _insert_quake(["us1000aaaa", "us1000bbbb"])
+    adapter = _FakeQuake()
+    store, captured = _build_store("usgs_quake", adapter)
+
+    assert "usgs_quake" in store._seeded
+    assert len(store._seen["usgs_quake"]) == 2
+
+    adapter.set_batch(["us1000aaaa", "us1000bbbb"])
+    store.refresh()
+    assert captured == [], "both quakes already received → zero broadcast"
+
+    adapter.set_batch(["us1000aaaa", "us1000bbbb", "us1000cccc"])
+    store.refresh()
+    assert _emitted_ids(captured) == ["us1000cccc"], "only the new quake broadcasts"
+
+
+def test_no_durable_rows_falls_back_to_silent_first_poll():
+    # Fresh DB (0 durable rows): the source must NOT be pre-marked seeded, so
+    # the first real batch is silently seeded (never leaked) — the fresh-start
+    # safety property.
+    adapter = _FakeWZDx()
+    store, captured = _build_store("wzdx", adapter)
+    assert "wzdx" not in store._seeded, "0 durable rows → not pre-marked seeded"
+
+    adapter.set_batch(["A", "B"])
+    store.refresh()
+    assert captured == [], "first non-empty poll on a fresh DB is silent"
+
+    adapter.set_batch(["A", "B", "C"])
+    store.refresh()
+    assert _emitted_ids(captured) == ["C"]