feat(notifications): Phase 2.2 inhibitor and grouper

Adds inline pipeline stages between the bus and the severity router:

- Inhibitor: suppresses lower-or-equal severity events when a key
  in event.inhibit_keys is already active. TTL configurable, default
  30 minutes.
- Grouper: coalesces events sharing group_key within a time window
  (default 60s). Most recent event wins. tick() and flush_all()
  drive emission; no background timers in Phase 2.2.
- build_pipeline now wires: bus -> inhibitor -> grouper -> severity_router

Phase 2.1 dispatcher tests continue to pass unchanged.

meshai/notifications/pipeline/__init__.py

@@ -1,10 +1,12 @@
 """Notification pipeline package.
 
-Phase 2.1 skeleton:
-  - EventBus: pub/sub for adapter ingress
-  - SeverityRouter: forks immediate vs digest paths
-  - Dispatcher: routes immediate events to channels via existing rules
-  - StubDigestQueue: placeholder for Phase 2.3 aggregator
+Phase 2.1 + 2.2:
+  - EventBus: pub/sub ingress
+  - Inhibitor: suppresses redundant events by inhibit_keys
+  - Grouper: coalesces events sharing group_key within a window
+  - SeverityRouter: forks immediate vs digest
+  - Dispatcher: routes immediate via channels (existing rules schema)
+  - StubDigestQueue: placeholder for Phase 2.3
 
 Usage:
     from meshai.notifications.pipeline import build_pipeline
@@ -19,14 +21,15 @@ from meshai.notifications.pipeline.severity_router import (
     StubDigestQueue,
 )
 from meshai.notifications.pipeline.dispatcher import Dispatcher
+from meshai.notifications.pipeline.inhibitor import Inhibitor
+from meshai.notifications.pipeline.grouper import Grouper
 
 
 def build_pipeline(config) -> EventBus:
     """Build the pipeline and return the EventBus.
 
-    Adapters emit events to this bus and they flow through the
-    severity router to either the dispatcher (immediate) or the
-    digest stub (priority/routine).
+    Wiring:
+      bus -> inhibitor -> grouper -> severity_router -> (dispatcher | digest_stub)
     """
     bus = EventBus()
     dispatcher = Dispatcher(config, create_channel)
@@ -35,14 +38,16 @@ def build_pipeline(config) -> EventBus:
         immediate_handler=dispatcher.dispatch,
         digest_handler=digest.enqueue,
     )
-    bus.subscribe(severity_router.handle)
+    grouper = Grouper(next_handler=severity_router.handle)
+    inhibitor = Inhibitor(next_handler=grouper.handle)
+    bus.subscribe(inhibitor.handle)
     return bus
 
 
 def build_pipeline_components(config) -> tuple:
     """Like build_pipeline, but returns all components for test inspection.
 
-    Returns (bus, dispatcher, digest, severity_router).
+    Returns (bus, inhibitor, grouper, severity_router, dispatcher, digest).
     """
     bus = EventBus()
     dispatcher = Dispatcher(config, create_channel)
@@ -51,8 +56,10 @@ def build_pipeline_components(config) -> tuple:
         immediate_handler=dispatcher.dispatch,
         digest_handler=digest.enqueue,
     )
-    bus.subscribe(severity_router.handle)
-    return bus, dispatcher, digest, severity_router
+    grouper = Grouper(next_handler=severity_router.handle)
+    inhibitor = Inhibitor(next_handler=grouper.handle)
+    bus.subscribe(inhibitor.handle)
+    return bus, inhibitor, grouper, severity_router, dispatcher, digest
 
 
 __all__ = [
@@ -60,6 +67,8 @@ __all__ = [
     "SeverityRouter",
     "StubDigestQueue",
     "Dispatcher",
+    "Inhibitor",
+    "Grouper",
     "build_pipeline",
     "build_pipeline_components",
     "get_bus",

meshai/notifications/pipeline/grouper.py

@@ -0,0 +1,96 @@
+"""Event grouper.
+
+Coalesces events sharing a group_key inside a time window. The most
+recent event for a group_key wins; older versions are replaced.
+Events without a group_key pass through immediately.
+
+The grouper holds events in a window. tick() flushes events whose
+window has expired. For Phase 2.2, tick() is called explicitly by
+tests; Phase 2.3+ will integrate with the digest scheduler for live
+operation.
+"""
+
+import logging
+import time
+from typing import Callable
+
+from meshai.notifications.events import Event
+
+
+class Grouper:
+    """Coalesce same-group_key events inside a window."""
+
+    def __init__(
+        self,
+        next_handler: Callable[[Event], None],
+        window_seconds: float = 60.0,
+    ):
+        """Initialize.
+
+        Args:
+            next_handler: Callable that receives events when they
+                exit the grouper (either immediately if no group_key,
+                or after the window expires).
+            window_seconds: How long to hold a group_key before
+                emitting downstream (default 60 seconds).
+        """
+        self._next = next_handler
+        self._window = window_seconds
+        # {group_key: (event, hold_until_ts)}
+        self._held: dict[str, tuple[Event, float]] = {}
+        self._logger = logging.getLogger("meshai.pipeline.grouper")
+
+    def _now(self) -> float:
+        return time.time()
+
+    def handle(self, event: Event) -> None:
+        """Process an event.
+
+        Events without group_key pass through immediately.
+        Events with group_key are held, replacing any prior held
+        event with the same group_key. The held event is emitted
+        later via tick().
+        """
+        if not event.group_key:
+            self._next(event)
+            return
+
+        now = self._now()
+        hold_until = now + self._window
+        prior = self._held.get(event.group_key)
+        if prior is not None:
+            self._logger.info(
+                f"COALESCED event {event.id} into group {event.group_key!r}, "
+                f"replacing prior event {prior[0].id}"
+            )
+        self._held[event.group_key] = (event, hold_until)
+
+    def tick(self) -> int:
+        """Flush events whose window has expired.
+
+        Returns the number of events emitted.
+        """
+        now = self._now()
+        to_emit = [
+            (gk, ev) for gk, (ev, hu) in self._held.items() if hu <= now
+        ]
+        for gk, _ in to_emit:
+            del self._held[gk]
+        for _, ev in to_emit:
+            self._next(ev)
+        return len(to_emit)
+
+    def flush_all(self) -> int:
+        """Immediately emit every held event, regardless of window.
+
+        Used at shutdown and by tests. Returns count emitted.
+        """
+        events = [ev for ev, _ in self._held.values()]
+        self._held.clear()
+        for ev in events:
+            self._next(ev)
+        return len(events)
+
+    def held_count(self) -> int:
+        """For tests: number of events currently held."""
+        return len(self._held)

meshai/notifications/pipeline/inhibitor.py

@@ -0,0 +1,92 @@
+"""Severity-based event inhibitor.
+
+Suppresses lower-severity events when a higher-severity event for the
+same logical incident (matching inhibit_keys) is already active.
+
+Inhibit keys are operator-defined strings on the Event that identify
+the underlying incident. Two events sharing an inhibit_key refer to
+the same situation. If a critical event for "battery:BLD-MTN" fired
+recently, a subsequent warning event with the same key gets suppressed.
+"""
+
+import logging
+import time
+from typing import Callable
+
+from meshai.notifications.events import Event
+
+
+class Inhibitor:
+    """Suppress lower-severity events when higher-severity is active."""
+
+    SEVERITY_RANK = {"routine": 0, "priority": 1, "immediate": 2}
+
+    def __init__(
+        self,
+        next_handler: Callable[[Event], None],
+        ttl_seconds: float = 1800.0,
+    ):
+        """Initialize.
+
+        Args:
+            next_handler: Callable that receives non-suppressed events.
+            ttl_seconds: How long an inhibit_key remains active after
+                the originating event (default 30 minutes).
+        """
+        self._next = next_handler
+        self._ttl = ttl_seconds
+        # {inhibit_key: (rank, expires_at)}
+        self._active: dict[str, tuple[int, float]] = {}
+        self._logger = logging.getLogger("meshai.pipeline.inhibitor")
+
+    def _now(self) -> float:
+        # Hookable for tests
+        return time.time()
+
+    def _prune_expired(self, now: float) -> None:
+        expired = [k for k, (_, exp) in self._active.items() if exp <= now]
+        for k in expired:
+            del self._active[k]
+
+    def handle(self, event: Event) -> None:
+        """Process an event: either suppress it or pass it on.
+
+        If any of the event's inhibit_keys is currently active at a
+        higher-or-equal rank, the event is suppressed. Otherwise, the
+        event's inhibit_keys are recorded/upgraded, and the event is
+        passed to the next handler.
+        """
+        now = self._now()
+        self._prune_expired(now)
+
+        event_rank = self.SEVERITY_RANK.get(event.severity, 0)
+
+        # Check suppression
+        for key in event.inhibit_keys:
+            entry = self._active.get(key)
+            if entry is not None:
+                active_rank, _ = entry
+                if active_rank >= event_rank:
+                    self._logger.info(
+                        f"SUPPRESSED event {event.id} ({event.severity}) "
+                        f"by active key {key!r} at rank {active_rank}"
+                    )
+                    return
+
+        # Record / upgrade entries
+        new_expires = now + self._ttl
+        for key in event.inhibit_keys:
+            existing = self._active.get(key)
+            if existing is None or existing[0] < event_rank:
+                self._active[key] = (event_rank, new_expires)
+
+        # Pass through
+        self._next(event)
+
+    def active_keys(self) -> dict[str, tuple[int, float]]:
+        """For tests: snapshot of currently-active inhibit keys."""
+        return dict(self._active)
+
+    def clear(self) -> None:
+        """For tests: reset state."""
+        self._active.clear()