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

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: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-07-06 02:38:44 +00:00
commit 96040257c1
3 changed files with 121 additions and 5 deletions

View file

@ -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,

View file

@ -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:
itd_511/tomtom_incidents with idaho_511:event:* external_ids, but the
native adapters emit source '511'/'traffic' with derived event_ids),
traffic (persistent rows are Central-keyed: tomtom_incidents with
tomtom external_ids, but the native adapter emits source '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
@ -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",

View file

@ -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"