fix(roads511): stable external_id -> durably pre-seedable (belt + suspenders) (#51)

roads511 emitted external_id=None, so it couldn't be durably pre-seeded from
the persistent tables (only wzdx/usgs_quake were) — it relied solely on the
in-memory first-poll seed. Thread a stable external_id="511_{itd_id}" through
consistently so it joins the durable layer:

- env/roads511.py: _parse_event raw event + to_event both carry
  external_id="511_{id}" (== event_id). Flips _seen_key to the ext: branch and
  makes the incident decider persist traffic_events(source='511', external_id)
  — which ALSO restores the decider's own dedup (external_id=None was the
  original roads511 leak cause).
- env/store.py _seed_from_persistent: add a "511" spec (seed from
  traffic_events where source='511', by external_id) mirroring wzdx; shared
  _key_ext helper so keys can't drift.
- consistency proven byte-identical (raw _seen_key == pre-seed key ==
  511\x1eext:511_{id}); durable-preseed + regression tests added.

Live DB: 0 source='511' rows yet (flip recent) -> durability engages as native
rows accumulate; layer-2 in-memory seed covers the interim (atomic fetch).
Central-era itd_511 rows use a different keyspace, intentionally not covered.
Suite at 10-failure baseline (1716 passed).

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

work/meshai/env/roads511.py

@@ -329,6 +329,12 @@ class Roads511Adapter:
             event = {
                 "source": "511",
                 "event_id": f"511_{event_id}",
+                # Stable ITD id, EQUAL to event_id. Threading it as external_id
+                # makes _seen_key use the ext: branch (511\x1eext:511_{id}) and
+                # lets the incident decider persist a matching
+                # traffic_events(source='511', external_id='511_{id}') row so the
+                # store's durable startup pre-seed can suppress this item.
+                "external_id": f"511_{event_id}",
                 "event_type": event_type,
                 "headline": headline,
                 "description": description[:500] if description else "",
@@ -398,8 +404,15 @@ class Roads511Adapter:
             _roadway = props.get("roadway")
             _desc = evt.get("description", "") or ""
             _is_closure = bool(props.get("is_closure"))
+            # Stable ITD id (the raw event carries the same value as its
+            # event_id). Threading it lets the incident decider persist a
+            # traffic_events(source='511', external_id='511_{id}') row that the
+            # store's durable pre-seed matches via _key_ext('511', external_id).
+            # NOTE: Central-era rows used source 'itd_511' with 'idaho_511:event:*'
+            # ids — a different keyspace the pre-seed intentionally does not cover.
+            _external_id = evt.get("external_id") or event_id
             canonical_data = {
-                "external_id":    None,   # native adapter, no Central dedup
+                "external_id":    _external_id,   # 511_{itd_id}; enables durable pre-seed
                 "source":         "511",
                 "sub_type":       "road_closed" if _is_closure else "incident",
                 "road":           _roadway or None,

work/meshai/env/store.py

@@ -316,14 +316,26 @@ class EnvironmentalStore:
             USGS id (e.g. ``us6000t9bn``) and the quake decider persists it as
             ``quake_events.event_id`` verbatim → ``_key_eid('usgs_quake', id)``
             matches exactly.
+          * ``roads511``    — the native 511 adapter now carries a stable
+            ``external_id`` (``511_{itd_id}``, EQUAL to its ``event_id``) on its
+            raw event and the incident decider persists it as
+            ``traffic_events(source='511', external_id='511_{itd_id}')``. Same
+            value on both sides → ``_key_ext('511', external_id)`` matches
+            exactly. Central-era rows used source ``itd_511`` with
+            ``idaho_511:event:*`` ids — a DIFFERENT keyspace this pre-seed does
+            NOT cover. On the first restart after this change there may be ~0
+            ``source='511'`` rows yet; the layer-2 in-memory silent-first-poll
+            seed covers any Central-era backlog in the interim (safe because
+            roads511 fetches atomically per tick), and durability grows as
+            native ``source='511'`` rows accumulate.
 
         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:
+          traffic (persistent rows are Central-keyed: tomtom_incidents with
-          itd_511/tomtom_incidents with idaho_511:event:* external_ids, but the
+          tomtom external_ids, but the native adapter emits source 'traffic'
-          native adapters emit source '511'/'traffic' with derived event_ids),
+          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
@@ -352,6 +364,13 @@ class EnvironmentalStore:
                 "WHERE source='wzdx' AND external_id IS NOT NULL",
                 lambda row: _key_ext("wzdx", row[0]),
             ),
+            (
+                "511",
+                "traffic_events(source='511')",
+                "SELECT external_id FROM traffic_events "
+                "WHERE source='511' AND external_id IS NOT NULL",
+                lambda row: _key_ext("511", row[0]),
+            ),
             (
                 "usgs_quake",
                 "quake_events",

work/tests/test_store_received_delta.py

@@ -15,7 +15,7 @@ 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.env.store import EnvironmentalStore, _key_ext
 from meshai.config import EnvironmentalConfig
 from meshai.notifications.pipeline.bus import EventBus
 from meshai.notifications.events import make_event
@@ -389,3 +389,87 @@ def test_no_durable_rows_falls_back_to_silent_first_poll():
     adapter.set_batch(["A", "B", "C"])
     store.refresh()
     assert _emitted_ids(captured) == ["C"]
+
+
+class _FakeRoads511:
+    """Native roads511 stand-in. Raw events carry a stable external_id
+    '511_{id}' EQUAL to their event_id (like the patched adapter), so the
+    seen-key is '511\x1eext:511_{id}' — exactly what the durable pre-seed loads
+    from traffic_events(source='511')."""
+
+    def __init__(self):
+        self._batch: list[dict] = []
+
+    def set_batch(self, ext_ids: list[str]) -> None:
+        self._batch = [
+            {"source": "511", "event_id": x, "external_id": x, "fetched_at": 0}
+            for x in ext_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["external_id"]
+        return make_event(source="511", category="test_delta",
+                          severity="routine", title=eid, summary=eid,
+                          group_key=eid)
+
+
+def test_roads511_seen_key_matches_persistent_preseed_key():
+    # CONSISTENCY PROOF: the SAME roads511 item must produce a byte-identical
+    # self._seen key on both sides — the live emit path (_seen_key over the real
+    # adapter's raw event) and the durable pre-seed (_key_ext built from the
+    # persisted traffic_events row). If these drift the pre-seed silently does
+    # nothing.
+    from meshai.env.roads511 import Roads511Adapter
+
+    # Build a raw event exactly as the adapter would from an ITD item.
+    # _parse_event does not touch self, so __new__ (no network/init) is safe.
+    adapter = Roads511Adapter.__new__(Roads511Adapter)
+    raw = adapter._parse_event(
+        {"id": "11165", "Latitude": 43.6, "Longitude": -116.2,
+         "Description": "US-20 closed"},
+        now=0.0,
+    )
+    assert raw["source"] == "511"
+    assert raw["external_id"] == "511_11165"
+    assert raw["event_id"] == "511_11165"
+
+    # Live emit key (no bus needed for _seen_key).
+    store = EnvironmentalStore(EnvironmentalConfig(), event_bus=None)
+    live_key = store._seen_key(raw)
+
+    # The incident decider persists traffic_events.external_id == raw external_id.
+    persisted_external_id = raw["external_id"]
+    preseed_key = _key_ext("511", persisted_external_id)
+
+    assert live_key == preseed_key == "511\x1eext:511_11165", (
+        f"live={live_key!r} preseed={preseed_key!r} must be identical"
+    )
+
+
+def test_persistent_preseed_roads511_by_external_id():
+    # N durable rows in traffic_events(source='511'). A fresh store must treat
+    # them as already-received: polling those same N broadcasts NOTHING; adding
+    # one external_id NOT in the table broadcasts only that one.
+    known = [f"511_{i}" for i in range(4)]
+    _insert_traffic(known, source="511")
+
+    adapter = _FakeRoads511()
+    store, captured = _build_store("roads511", adapter)
+
+    # roads511's SOURCE is '511'; pre-seed keys/marks by source, not adapter name.
+    assert "511" in store._seeded
+    assert len(store._seen["511"]) == 4
+
+    adapter.set_batch(known)
+    store.refresh()
+    assert captured == [], "all 4 durably-known 511 rows → zero broadcast"
+
+    adapter.set_batch(known + ["511_99"])
+    store.refresh()
+    assert _emitted_ids(captured) == ["511_99"], "only the not-in-table id broadcasts"