fix(dispatcher): commit guards only after successful delivery (B13) + dedup suffix for WFIGS lifecycle updates

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

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.
 

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