diff --git a/meshai/central/wfigs_handler.py b/meshai/central/wfigs_handler.py index 7ff9f63..2c20426 100644 --- a/meshai/central/wfigs_handler.py +++ b/meshai/central/wfigs_handler.py @@ -325,6 +325,13 @@ def _attach_commit_handles(data: Optional[dict], *, irwin_id: str, data["_on_broadcast_committed"] = _on_commit data["_broadcast_audit"] = {"table": "fires", "pk": irwin_id} data["_cooldown_suffix"] = irwin_id + # v0.6-4: WFIGS publishes the SAME envelope id (IrwinID) for every sweep + # over an incident's life, so the dispatcher's (source, id) dedup + # permanently swallowed every post-"New" lifecycle broadcast (growth / + # containment updates) this handler deliberately synthesized. Stamping + # the state that justified THIS broadcast into the dedup suffix lets + # unchanged re-deliveries dedup as before while genuine updates pass. + data["_dedup_suffix"] = f"{acres}|{contained_pct}" # ---------- helpers ------------------------------------------------------- diff --git a/meshai/notifications/pipeline/dispatcher.py b/meshai/notifications/pipeline/dispatcher.py index 4d2e177..fa3b734 100644 --- a/meshai/notifications/pipeline/dispatcher.py +++ b/meshai/notifications/pipeline/dispatcher.py @@ -357,13 +357,38 @@ class Dispatcher: ) return - # ---------- Section 2 — per-toggle cooldown ---------- + # ---------- Section 2 — region scope + severity floor + matrix ---- + # v0.6-4 (B13 fix): resolution before commitment. Region scope, the + # min_severity floor, and severity_channels matrix resolution all + # run BEFORE the cooldown/dedup guards, so an event that was never + # going to deliver (wrong region, below floor, empty matrix row) + # cannot arm a cooldown window or burn a dedup slot. Previously a + # non-deliverable event committed both, silently suppressing later + # deliverable events for up to the dedup retention window (7 days). + regions = getattr(tog, "regions", None) or [] + if regions: + ev_regions = set(filter(None, [event.region, *(event.regions or [])])) + if not (set(regions) & ev_regions): + return + event_rank = self.SEVERITY_RANK.get(event.severity, 0) + if event_rank < self.SEVERITY_RANK.get(getattr(tog, "min_severity", "routine"), 0): + return + sev_channels = getattr(tog, "severity_channels", None) or {} + ch_types = [c for c in sev_channels.get(event.severity, []) if c != "digest"] + if not ch_types: + return + + # ---------- Section 3 — per-toggle cooldown (check only) ---------- # Immediate-severity events bypass cooldown entirely — they are # already rate-controlled by source handler change detection. + # v0.6-4 (B13 fix): this section only CHECKS the cooldown. Arming + # it is deferred to Section 6 and happens only after a delivery + # actually succeeded. if getattr(event, "severity", None) == "immediate": cooldown_s = 0 else: cooldown_s = int(getattr(tog, "cooldown_seconds", 300) or 0) + ck = None if cooldown_s > 0: suffix = (event.data or {}).get("_cooldown_suffix", "") region_key = event.region or "*" @@ -380,22 +405,22 @@ class Dispatcher: self._cooldown_dropped += 1 self._persist_state() return # silent throttle — no log spam - self._toggle_cooldown[ck] = now - self._persist_cooldown(ck, now, cooldown_s) - # In-memory prune: mirror the SQLite cutoff when the map grows - # past the threshold. The SQLite prune already ran inside - # _persist_cooldown. - # v0.6-3b: prune size + multiplier from adapter_config. - _prune_size = int(adapter_config.dispatcher.cooldown_prune_size) - _prune_mult = int(adapter_config.dispatcher.cooldown_prune_multiplier) - if len(self._toggle_cooldown) > _prune_size: - cutoff = now - (_prune_mult * cooldown_s) - self._toggle_cooldown = { - k: t for k, t in self._toggle_cooldown.items() if t >= cutoff - } + # v0.6-4 (B13 fix): arming moved to Section 6 (post-delivery). - # ---------- Section 3 — (source, event.id) dedup ---------- - dk = (event.source or "", event.id or "") + # ---------- Section 4 — (source, event.id) dedup (check only) ------ + # v0.6-4: the dedup key gains an optional handler-supplied suffix + # (event.data["_dedup_suffix"]). Feeds like WFIGS publish the SAME + # upstream id (IrwinID) on every sweep for the life of an incident, + # so a bare (source, id) key permanently suppressed every later + # lifecycle broadcast (growth/containment updates) the handler had + # deliberately synthesized. Handlers that gate their own re-renders + # stamp the state that justified THIS broadcast into the suffix; + # unchanged re-deliveries still dedup, genuine updates pass. + _dd_suffix = str((event.data or {}).get("_dedup_suffix", "") or "") + _dd_id = event.id or "" + if _dd_suffix: + _dd_id = f"{_dd_id}#{_dd_suffix}" + dk = (event.source or "", _dd_id) if dk in self._dedup_lru: # Touch to keep recent. self._dedup_lru.move_to_end(dk) @@ -405,23 +430,11 @@ class Dispatcher: # evidence we're still seeing this id. self._persist_dedup(dk, time.time()) return - self._dedup_lru[dk] = True - self._persist_dedup(dk, time.time()) - # v0.6-3b: read cap from adapter_config (default 10_000). - _lru_max = int(adapter_config.dispatcher.dedup_lru_max) - while len(self._dedup_lru) > _lru_max: - self._dedup_lru.popitem(last=False) # evict oldest + # v0.6-4 (B13 fix): recording moved to Section 6 (post-delivery). + # A delivery that fails (or raises for every channel) leaves no + # dedup trace, so the next feed sweep retries it naturally. - regions = getattr(tog, "regions", None) or [] - if regions: - ev_regions = set(filter(None, [event.region, *(event.regions or [])])) - if not (set(regions) & ev_regions): - return - event_rank = self.SEVERITY_RANK.get(event.severity, 0) - if event_rank < self.SEVERITY_RANK.get(getattr(tog, "min_severity", "routine"), 0): - return - - # ---------- Section 4 — friendly composer wired in ---------- + # ---------- Section 5 — friendly composer wired in ---------- # Render once per event; reused across every channel below. Wrapped # so a renderer fault never blocks delivery — we fall back to the # legacy make_payload_from_event message (event.summary|title|category). @@ -431,10 +444,8 @@ class Dispatcher: self._logger.exception("mesh composer crashed; falling back to legacy message") friendly = None - sev_channels = getattr(tog, "severity_channels", None) or {} - for ch_type in sev_channels.get(event.severity, []): - if ch_type == "digest": - continue + delivered_any = False + for ch_type in ch_types: try: rule = self._toggle_to_rule(tog, ch_type, event) channel = self._channel_factory(rule, self._connector) @@ -444,6 +455,7 @@ class Dispatcher: payload = make_payload_from_event(event) success = await channel.deliver(payload, rule) if success: + delivered_any = True self._logger.info(f"Dispatched event {event.id} via toggle {fam}/{ch_type}") # v0.5.8b post-broadcast commit. Persistence-side # bookkeeping that should only happen when a delivery @@ -455,6 +467,35 @@ class Dispatcher: except Exception: self._logger.exception(f"Toggle channel delivery failed for {fam}/{ch_type}") + # ---------- Section 6 — guard commit (v0.6-4, B13 fix) ---------- + # Cooldown arming + dedup recording happen ONLY after at least one + # channel actually delivered. A fully-failed delivery leaves no + # guard state behind, so the next feed sweep retries naturally + # instead of being silently suppressed. + if not delivered_any: + return + commit_now = time.time() + if ck is not None and cooldown_s > 0: + self._toggle_cooldown[ck] = commit_now + self._persist_cooldown(ck, commit_now, cooldown_s) + # In-memory prune: mirror the SQLite cutoff when the map grows + # past the threshold. The SQLite prune already ran inside + # _persist_cooldown. + # v0.6-3b: prune size + multiplier from adapter_config. + _prune_size = int(adapter_config.dispatcher.cooldown_prune_size) + _prune_mult = int(adapter_config.dispatcher.cooldown_prune_multiplier) + if len(self._toggle_cooldown) > _prune_size: + cutoff = commit_now - (_prune_mult * cooldown_s) + self._toggle_cooldown = { + k: t for k, t in self._toggle_cooldown.items() if t >= cutoff + } + self._dedup_lru[dk] = True + self._persist_dedup(dk, commit_now) + # v0.6-3b: read cap from adapter_config (default 10_000). + _lru_max = int(adapter_config.dispatcher.dedup_lru_max) + while len(self._dedup_lru) > _lru_max: + self._dedup_lru.popitem(last=False) # evict oldest + def dispatch_stats(self) -> dict: """Expose v0.5.2 toggle-path guard counters for ops/health endpoints. diff --git a/tests/test_v064_guard_commit.py b/tests/test_v064_guard_commit.py new file mode 100644 index 0000000..6cafc39 --- /dev/null +++ b/tests/test_v064_guard_commit.py @@ -0,0 +1,177 @@ +"""v0.6-4 — B13 guard-commit ordering + dedup suffix. + +Regression tests for two silent-suppression defects: + +1. B13 ordering: cooldown arming and dedup recording used to happen + BEFORE the region filter / severity floor / matrix resolution and + BEFORE delivery. A non-deliverable or failed event therefore burned + guard state and suppressed later deliverable events for up to the + dedup retention window. Now both commit only after >=1 successful + delivery. + +2. Dedup suffix: feeds like WFIGS publish the SAME upstream id + (IrwinID) on every sweep, so the bare (source, id) dedup key + permanently swallowed every post-"New" lifecycle broadcast. The + handler now stamps the broadcast-justifying state into + event.data["_dedup_suffix"]; unchanged repeats dedup, updates pass. +""" + +import asyncio + +import pytest + +from meshai.config import Config +from meshai.notifications.pipeline.dispatcher import Dispatcher +from meshai.notifications.events import make_event + + +class RecChannel: + def __init__(self, rec, succeed=True): + self.rec = rec + self.succeed = succeed + + async def deliver(self, payload, rule): + self.rec.append({ + "delivery_type": rule.delivery_type, + "message": payload.message, + }) + return self.succeed + + +def _cfg(**kw): + cfg = Config() + cfg.notifications.rules = [] + cfg.notifications.cold_start_grace_seconds = 0 + t = cfg.notifications.toggles["weather"] + t.enabled = True + t.min_severity = kw.get("min_severity", "routine") + t.regions = kw.get("regions", []) + t.severity_channels = kw.get("severity_channels", { + "routine": ["mesh_broadcast"], + "priority": ["mesh_broadcast"], + "immediate": ["mesh_broadcast"], + }) + t.freshness_seconds = 600 + t.cooldown_seconds = kw.get("cooldown_seconds", 300) + t.broadcast_channel = 1 + return cfg + + +def _ev(eid="ev-1", severity="priority", **data): + ev = make_event(source="nws", category="weather.alert.severe", + severity=severity, title="t") + ev.id = eid + if data: + ev.data.update(data) + return ev + + +def _disp(cfg, succeed=True): + rec: list = [] + d = Dispatcher(cfg, lambda rule, conn: RecChannel(rec, succeed), connector=None) + return d, rec + + +# ---------------------------------------------------------------- B13 + + +def test_empty_matrix_row_burns_no_guard_state(): + """Event whose severity row is empty must leave no cooldown/dedup.""" + cfg = _cfg(severity_channels={"routine": [], "priority": [], + "immediate": ["mesh_broadcast"]}) + d, rec = _disp(cfg) + asyncio.run(d.dispatch(_ev(eid="fire-x", severity="priority"))) + assert rec == [] + assert len(d._toggle_cooldown) == 0 + assert len(d._dedup_lru) == 0 + # Operator fixes the matrix -> the SAME event id must now deliver. + cfg.notifications.toggles["weather"].severity_channels["priority"] = \ + ["mesh_broadcast"] + asyncio.run(d.dispatch(_ev(eid="fire-x", severity="priority"))) + assert len(rec) == 1 + + +def test_below_severity_floor_burns_no_guard_state(): + cfg = _cfg(min_severity="immediate") + d, rec = _disp(cfg) + asyncio.run(d.dispatch(_ev(eid="ev-floor", severity="priority"))) + assert rec == [] and not d._toggle_cooldown and not d._dedup_lru + + +def test_wrong_region_burns_no_guard_state(): + cfg = _cfg(regions=["US-ID"]) + ev = _ev(eid="ev-region", severity="priority") + ev.region = "US-MT" + d, rec = _disp(cfg) + asyncio.run(d.dispatch(ev)) + assert rec == [] and not d._toggle_cooldown and not d._dedup_lru + + +def test_failed_delivery_leaves_no_dedup_and_retries(): + """All-channels-failed delivery must not dedup-record: the next + sweep's redelivery of the same id must reach the channel again.""" + cfg = _cfg() + d, rec = _disp(cfg, succeed=False) + asyncio.run(d.dispatch(_ev(eid="retry-me", severity="priority"))) + assert len(rec) == 1 # attempted + assert not d._dedup_lru # but not recorded + assert not d._toggle_cooldown # and no cooldown armed + # Channel recovers -> same event id delivers. + d._channel_factory = lambda rule, conn: RecChannel(rec, succeed=True) + asyncio.run(d.dispatch(_ev(eid="retry-me", severity="priority"))) + assert len(rec) == 2 + assert d._dedup_lru # recorded only on success + + +def test_successful_delivery_commits_cooldown_and_dedup(): + cfg = _cfg() + d, rec = _disp(cfg) + asyncio.run(d.dispatch(_ev(eid="ok-1", severity="priority"))) + assert len(rec) == 1 + assert len(d._toggle_cooldown) == 1 + assert ("nws", "ok-1") in d._dedup_lru + # Exact repeat is throttled (cooldown fires first at this severity). + asyncio.run(d.dispatch(_ev(eid="ok-1", severity="priority"))) + assert len(rec) == 1 + assert d._cooldown_dropped == 1 + # With cooldown out of the way, the repeat is dedup-dropped. + cfg.notifications.toggles["weather"].cooldown_seconds = 0 + asyncio.run(d.dispatch(_ev(eid="ok-1", severity="priority"))) + assert len(rec) == 1 + assert d._dedup_dropped == 1 + + +# ---------------------------------------------------------------- suffix + + +def test_dedup_suffix_lets_updates_pass_and_repeats_dedup(): + """Same upstream id: identical suffix dedups, changed suffix passes — + the WFIGS lifecycle-update fix.""" + cfg = _cfg(cooldown_seconds=0) + d, rec = _disp(cfg) + # "New" broadcast: 0.1 acres. + asyncio.run(d.dispatch( + _ev(eid="{IRWIN-1}", severity="immediate", _dedup_suffix="0.1|0"))) + assert len(rec) == 1 + # Feed re-sweeps, nothing changed: same id + same suffix -> dedup. + asyncio.run(d.dispatch( + _ev(eid="{IRWIN-1}", severity="immediate", _dedup_suffix="0.1|0"))) + assert len(rec) == 1 + assert d._dedup_dropped == 1 + # The fire blows up: same id, NEW suffix -> must broadcast. + asyncio.run(d.dispatch( + _ev(eid="{IRWIN-1}", severity="immediate", _dedup_suffix="9400.0|10"))) + assert len(rec) == 2 + # And that update's repeat dedups too. + asyncio.run(d.dispatch( + _ev(eid="{IRWIN-1}", severity="immediate", _dedup_suffix="9400.0|10"))) + assert len(rec) == 2 + + +def test_wfigs_handler_stamps_dedup_suffix(): + from meshai.central.wfigs_handler import _attach_commit_handles + data = {} + _attach_commit_handles(data, irwin_id="{X}", acres=42.0, + contained_pct=15, event_log_row_id=None) + assert data["_dedup_suffix"] == "42.0|15" + assert data["_cooldown_suffix"] == "{X}"