diff --git a/meshai/central/consumer.py b/meshai/central/consumer.py index 24b8ee2..892b179 100644 --- a/meshai/central/consumer.py +++ b/meshai/central/consumer.py @@ -11,6 +11,7 @@ Wire format (see Central CONSUMER-INTEGRATION guide, confirmed in v0.4 Phase A): -> Event["data"] (upstream payload, verbatim, incl `_enriched`) """ +import asyncio import json import logging import re @@ -332,6 +333,16 @@ class CentralConsumer: self._nc = None self._js = None self._subs: list = [] + # Drain mode: suppress bus.emit() during backlog catch-up. + # After all pending messages are consumed, _drain_complete() runs + # a decision pass over accumulated fire IrwinIDs and emits at most + # one event per fire through the pacer. + self._draining: bool = False + self._drain_irwin_ids: set = set() + self._drain_start: float = 0.0 # monotonic time drain started + self._drain_timeout: float = 30.0 # seconds before auto-exit + self._drain_msg_count: int = 0 # messages processed during drain + self._pacer = None # FirePacer, injected from main.py # ---- subject derivation ---- def _region(self) -> str: @@ -630,6 +641,11 @@ class CentralConsumer: owned: set of meshai source names this subscription may emit (sub-adapter routing for shared subjects); None = no filtering. + + During drain mode (_draining=True), bus.emit() is suppressed. Fire + IrwinIDs are tracked in _drain_irwin_ids for the post-drain decision + pass. All handler DB writes still happen inside _normalize() before + this point. """ try: envelope = json.loads(raw) @@ -643,24 +659,50 @@ class CentralConsumer: logger.debug("CentralConsumer: dropping %s source=%s -- not owned by " "subscription %s", subject, event.source, sorted(owned)) return None - if self._bus is not None: - self._bus.emit(event) + + if self._draining: + # Track fire IrwinIDs touched during drain for decision pass + irwin_id = (event.data or {}).get("_cooldown_suffix", "") + if irwin_id and event.source in ("fires", "wfigs"): + self._drain_irwin_ids.add(irwin_id) + elif self._bus is not None: + # Normal mode: route fire events through pacer, others direct + if (self._pacer is not None + and event.source in ("fires", "wfigs") + and (event.data or {}).get("_severity_override") == "priority"): + self._pacer.enqueue(event) + else: + self._bus.emit(event) return event async def _on_message(self, msg, owned=None) -> None: - """JetStream callback: normalize + emit, then ack.""" + """JetStream callback: normalize + emit, then ack. + + During drain mode, checks msg.metadata.num_pending after each + message. When pending hits 0, the backlog is consumed and + _drain_complete() runs the fire decision pass. + """ try: self._handle(msg.subject, msg.data, owned) except Exception: logger.exception("CentralConsumer: handler failed on %s", getattr(msg, "subject", "?")) - finally: + # Check drain completion BEFORE ack so the decision pass runs + # while we still hold the message (prevents interleaving). + if self._draining: + self._drain_msg_count += 1 + try: + meta = msg.metadata + if meta is not None and getattr(meta, "num_pending", None) == 0: + self._drain_complete() + except Exception: + logger.exception("drain: metadata check failed") + try: ack = getattr(msg, "ack", None) if ack is not None: - try: - await ack() - except Exception: - pass + await ack() + except Exception: + pass # ---- lifecycle ---- async def start(self) -> None: @@ -675,6 +717,18 @@ class CentralConsumer: sorted(subject_owned)) return + # Enter drain mode: suppress bus.emit() until the backlog from + # LAST_PER_SUBJECT delivery is fully consumed. The first _on_message + # callback processes through drain mode; when num_pending hits 0, + # _drain_complete() runs the fire decision pass. A timeout auto-exits + # drain if no messages arrive (empty backlog scenario). + self._draining = True + self._drain_irwin_ids.clear() + self._drain_msg_count = 0 + self._drain_start = time.monotonic() + logger.info("CentralConsumer: entering drain mode (timeout=%.0fs)", + self._drain_timeout) + region = self._region() logger.info("CentralConsumer: connecting region=%r subjects=%s", region or "(bare wildcards)", sorted(subject_owned)) @@ -690,7 +744,127 @@ class CentralConsumer: subj, durable=durable, cb=self._make_cb(owned), config=consumer_config()) self._subs.append(sub) logger.info("CentralConsumer subscribed %s owned-sources=%s", subj, sorted(owned)) - logger.info("CentralConsumer started; %d subjects subscribed", len(subject_owned)) + logger.info("CentralConsumer started; %d subjects subscribed (drain mode active)", + len(subject_owned)) + + # Schedule drain timeout: if no messages trigger drain completion + # within the window (e.g. empty backlog), auto-exit drain mode. + asyncio.get_event_loop().call_later( + self._drain_timeout, self._drain_timeout_check) + + # ---- drain mode ---- + + def _drain_timeout_check(self) -> None: + """Called by call_later after drain_timeout seconds. If still draining, + auto-exit. This handles the empty-backlog case where no messages arrive + to trigger num_pending == 0.""" + if not self._draining: + return + logger.info("drain: timeout after %.0fs (%d msgs processed) โ€” auto-completing", + time.monotonic() - self._drain_start, self._drain_msg_count) + self._drain_complete() + + def _drain_complete(self) -> None: + """Post-drain decision pass: one broadcast per fire, based on final DB state. + + Runs synchronously (no awaits) to prevent _on_message interleaving. + For each IrwinID touched during drain, reads the fires row and + decides: NEW, UPDATE, CLOSURE, or SILENCE. Events route through + the pacer (<=1/min) instead of direct bus.emit(). + """ + self._draining = False + if not self._drain_irwin_ids: + logger.info("drain complete: 0 fires touched") + return + + from meshai.persistence import get_db + from meshai.central.wfigs_handler import ( + _render, _location_anchor, _attach_commit_handles, + ) + from meshai.notifications.events import make_event + + conn = get_db() + emitted = 0 + silenced = 0 + + for irwin_id in self._drain_irwin_ids: + row = conn.execute( + "SELECT irwin_id, incident_name, incident_type, " + "current_acres, current_contained_pct, " + "lat, lon, county, state, landclass, " + "declared_at, tombstoned_at, last_broadcast_at, " + "last_broadcast_acres, last_broadcast_contained, " + "fire_cause, unique_fire_id, geocoder_city " + "FROM fires WHERE irwin_id = ?", (irwin_id,) + ).fetchone() + + if row is None: + continue + + tombstoned = row["tombstoned_at"] is not None + announced = row["last_broadcast_at"] is not None + + # Decision table (meshai-fire-fix-plan.md ยง3): + # Case 3: Never announced + already closed -> SILENCE + if not announced and tombstoned: + silenced += 1 + continue + + wire = None + category = "wildfire_incident" + + if announced and tombstoned: + # Case 4: Announced before + closed during gap -> CLOSURE + wire = _build_closure_wire(row) + category = "wildfire_closed" + elif not announced and not tombstoned: + # Case 2: Never announced + still active -> NEW + wire = _render(_row_to_normalized(row), prefix="New") + category = "wildfire_declared" + else: + # Case 1: Announced before + grew during gap -> UPDATE + # Check if anything actually changed + if (row["current_acres"] == row["last_broadcast_acres"] + and row["current_contained_pct"] == row["last_broadcast_contained"]): + silenced += 1 + continue + wire = _render( + _row_to_normalized(row), prefix="Update", + last_bcast_acres=row["last_broadcast_acres"], + last_bcast_contained=row["last_broadcast_contained"], + ) + + if wire is None: + silenced += 1 + continue + + # Build Event + data = {"_meshai_precomposed": True, "_severity_override": "priority"} + _attach_commit_handles( + data, irwin_id=irwin_id, + acres=row["current_acres"], + contained_pct=row["current_contained_pct"], + ) + data["_cooldown_suffix"] = irwin_id + data["_dedup_suffix"] = ( + f"{row['current_acres']}|{row['current_contained_pct']}|drain" + ) + + event = make_event( + source="fires", category=category, severity="priority", + title=wire, lat=row["lat"], lon=row["lon"], + group_key=irwin_id, inhibit_keys=[irwin_id], data=data, + ) + + # Route through pacer (<=1/min), fall back to direct emit + if self._pacer is not None: + self._pacer.enqueue(event) + elif self._bus is not None: + self._bus.emit(event) + emitted += 1 + + self._drain_irwin_ids.clear() + logger.info("drain complete: %d fires emitted, %d silenced", emitted, silenced) async def stop(self) -> None: if self._nc is not None: @@ -705,3 +879,56 @@ class CentralConsumer: self._nc = None self._js = None self._subs = [] + + +# ---------- drain-mode helpers (module-level) ----------------------------- + + +def _row_to_normalized(row) -> dict: + """Map a fires DB row (sqlite3.Row) to the normalized dict _render() expects. + + sqlite3.Row supports bracket access and .keys() but not .get() on + Python < 3.13. Use _safe_get() for optional columns. + """ + keys = set(row.keys()) + return { + "incident_name": row["incident_name"], + "acres": row["current_acres"], + "contained_pct": row["current_contained_pct"], + "fire_cause": row["fire_cause"] if "fire_cause" in keys else None, + "unique_fire_id": row["unique_fire_id"] if "unique_fire_id" in keys else None, + "declared_at_epoch": row["declared_at"], + "lat": row["lat"], + "lon": row["lon"], + "county": row["county"], + "state": row["state"], + "landclass": row["landclass"] if "landclass" in keys else None, + "geocoder_city": row["geocoder_city"] if "geocoder_city" in keys else None, + } + + +def _build_closure_wire(row) -> str: + """Build a closure wire string from a fires DB row. + + Replicates the tombstone wire format from wfigs_handler.py. + """ + from meshai.central.wfigs_handler import _location_anchor + + name = row["incident_name"] or "(unnamed fire)" + parts = [] + if row["current_acres"] is not None: + parts.append(f"{int(row['current_acres']):,} ac") + if row["current_contained_pct"] is not None: + parts.append(f"{int(row['current_contained_pct'])}% contained") + # Location anchor from row fields + loc_dict = { + "lat": row["lat"], "lon": row["lon"], + "county": row["county"], "state": row["state"], + } + anchor = _location_anchor(loc_dict) + if anchor and anchor != "(location unknown)": + parts.append(anchor) + lines = [f"\u2705 {name} \u2014 contained & closed"] + if parts: + lines.append(" | ".join(parts)) + return "\n".join(lines) diff --git a/meshai/central/wfigs_handler.py b/meshai/central/wfigs_handler.py index 1cbe072..6e4d16e 100644 --- a/meshai/central/wfigs_handler.py +++ b/meshai/central/wfigs_handler.py @@ -148,7 +148,7 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, wire = "\n".join(lines) if isinstance(data, dict): data["category"] = "wildfire_closed" - data["_severity_override"] = "immediate" + data["_severity_override"] = "priority" _attach_commit_handles( data, irwin_id=irwin_id, acres=fire_row["current_acres"], @@ -207,9 +207,10 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, # from acres/containment updates (wildfire_incident). if isinstance(data, dict): data["category"] = "wildfire_declared" - # v0.6-3c: severity override for fire broadcasts + # v0.6-3c: severity override for fire broadcasts (downgraded from + # immediate to priority to prevent cooldown/grouper bypass) if isinstance(data, dict): - data["_severity_override"] = "immediate" + data["_severity_override"] = "priority" _attach_commit_handles(data, irwin_id=irwin_id, acres=acres, contained_pct=contained_pct, event_log_row_id=log_id) @@ -230,9 +231,10 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, # handler call ran but no actual broadcast went out. if isinstance(data, dict): data["category"] = "wildfire_declared" - # v0.6-3c: severity override for fire broadcasts + # v0.6-3c: severity override for fire broadcasts (downgraded from + # immediate to priority to prevent cooldown/grouper bypass) if isinstance(data, dict): - data["_severity_override"] = "immediate" + data["_severity_override"] = "priority" _attach_commit_handles(data, irwin_id=irwin_id, acres=acres, contained_pct=contained_pct, event_log_row_id=log_id) @@ -274,9 +276,10 @@ def handle_wfigs(normalized: dict, envelope: dict, subject: str, wire = _render(normalized, prefix="Update", last_bcast_acres=last_bcast_acres, last_bcast_contained=last_bcast_contained) - # v0.6-3c: severity override for fire updates + # v0.6-3c: severity override for fire updates (downgraded from + # immediate to priority to prevent cooldown/grouper bypass) if isinstance(data, dict): - data["_severity_override"] = "immediate" + data["_severity_override"] = "priority" _attach_commit_handles(data, irwin_id=irwin_id, acres=acres, contained_pct=contained_pct, event_log_row_id=log_id) diff --git a/meshai/main.py b/meshai/main.py index 7aff133..45ba167 100644 --- a/meshai/main.py +++ b/meshai/main.py @@ -51,6 +51,7 @@ class MeshAI: self._pipeline_scheduler = None # DigestScheduler from start_pipeline() self.env_store = None # Environmental feeds store self._central_consumer = None # Central NATS consumer (v0.4) + self._fire_pacer = None # FirePacer for rate-limited fire broadcasts self._last_sub_check: float = 0.0 self.router: Optional[MessageRouter] = None self.responder: Optional[Responder] = None @@ -101,8 +102,15 @@ class MeshAI: self._pipeline_scheduler = await start_pipeline(self.event_bus, self.config) logger.info("Notification pipeline started") + # Fire pacer: rate-limits fire broadcasts to <=1/min during both + # drain catch-up and normal operation. + from .notifications.pipeline.pacer import FirePacer + self._fire_pacer = FirePacer(bus=self.event_bus, interval_seconds=60.0) + await self._fire_pacer.start() + from .central.consumer import CentralConsumer self._central_consumer = CentralConsumer(self.config.environmental, self.event_bus) + self._central_consumer._pacer = self._fire_pacer await self._central_consumer.start() logger.info("MeshAI started successfully") @@ -215,6 +223,9 @@ class MeshAI: if self._central_consumer is not None: await self._central_consumer.stop() + if self._fire_pacer is not None: + await self._fire_pacer.stop() + if self.connector: self.connector.disconnect() diff --git a/meshai/notifications/pipeline/pacer.py b/meshai/notifications/pipeline/pacer.py new file mode 100644 index 0000000..16039b2 --- /dev/null +++ b/meshai/notifications/pipeline/pacer.py @@ -0,0 +1,87 @@ +"""FIFO output pacer for fire broadcasts. + +Rate-limits fire events to at most one broadcast per interval (default 60s). +Used both during drain catch-up (post-reconnect decision pass) and during +normal operation to prevent burst-delivery of multiple fire events in rapid +succession. + +The queue is in-memory only โ€” no persistence. On restart, the drain mode in +consumer.py re-evaluates from DB state, so queued events lost on shutdown +are re-derived naturally. +""" + +from __future__ import annotations + +import asyncio +import logging +from typing import Optional + +logger = logging.getLogger(__name__) + + +class FirePacer: + """Unbounded FIFO queue that drains events to the bus at a fixed rate.""" + + def __init__(self, bus, interval_seconds: float = 60.0): + """Args: + bus: the EventBus whose .emit() method delivers events downstream + interval_seconds: minimum seconds between consecutive deliveries + """ + self._bus = bus + self._interval = interval_seconds + self._queue: asyncio.Queue = asyncio.Queue() + self._task: Optional[asyncio.Task] = None + + def enqueue(self, event) -> None: + """Non-blocking enqueue. Safe to call from sync code in the same loop.""" + self._queue.put_nowait(event) + logger.debug("pacer: enqueued event source=%s category=%s (pending=%d)", + event.source, event.category, self._queue.qsize()) + + async def _drain_loop(self) -> None: + """Pop one event, emit it, sleep interval, repeat.""" + while True: + try: + event = await self._queue.get() + except asyncio.CancelledError: + return + try: + self._bus.emit(event) + logger.info("pacer: emitted event source=%s category=%s (remaining=%d)", + event.source, event.category, self._queue.qsize()) + except Exception: + logger.exception("pacer: bus.emit() failed for event source=%s", + event.source) + if self._queue.empty(): + # No point sleeping when nothing is queued โ€” wait for next put + continue + try: + await asyncio.sleep(self._interval) + except asyncio.CancelledError: + return + + async def start(self) -> None: + """Spawn the drain loop as a background task.""" + if self._task is not None: + return + self._task = asyncio.create_task(self._drain_loop()) + logger.info("pacer: started (interval=%.0fs)", self._interval) + + async def stop(self) -> None: + """Cancel the drain loop. Queued events are discarded.""" + if self._task is not None: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + remaining = self._queue.qsize() + if remaining: + logger.warning("pacer: stopped with %d events still queued", remaining) + else: + logger.info("pacer: stopped (queue empty)") + + def pending_count(self) -> int: + """Number of events waiting in the queue.""" + return self._queue.qsize()