feat(broadcast): fit all mesh message formats to the packet budget (#8)

Tighten broadcast formats to fit 140 chars with critical info preserved:
traffic (directions/milepost never abbreviated, narrative trimmed from
end), nws hazard wording tightened (towns already path-sampled), fires
(drop ID line + ** + discovery time), avy (advice -> first sentence),
satpass/quake safety cap. Fire digest broadcast disabled by default.
All budget-aware via the shared max_chars.

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/adapter_config/defaults.py

@@ -397,6 +397,15 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = {
         "type": "bool",
         "description": "Whether the fire-digest scheduler broadcasts at the configured slots. Off => no broadcasts even if all other config is valid.",
     },
+    # digest_broadcast_enabled: independent kill-switch on the actual mesh
+    # emission. Disabled by default so the scheduler keeps running (building /
+    # recording digests) without putting the twice-daily digest on the mesh.
+    # Per-fire wfigs alerts are unaffected.
+    ("fires", "digest_broadcast_enabled"): {
+        "default": False,
+        "type": "bool",
+        "description": "Emit the twice-daily fire-digest broadcast. Disabled by default; per-fire wfigs alerts are unaffected.",
+    },
     # digest_schedule: list of HH:MM strings, local-time per digest_timezone.
     # Mirrors band_conditions_schedule shape so operators can reason
     # about the two side-by-side.
@@ -587,9 +596,9 @@ REGISTRY: dict[tuple[str, str], dict[str, Any]] = {
         "description": "Maximum characters for the area field on line 2 of NWS wire. Truncates at last word boundary.",
     },
     ("nws", "single_packet_max_chars"): {
-        "default": 200,
+        "default": 140,
         "type": "int",
-        "description": "Maximum characters for a single NWS mesh packet. Budget enforced after all fields are assembled; only the locations town-list is trimmed to fit.",
+        "description": "Maximum characters for a single NWS mesh packet. Budget enforced after all fields are assembled; only the locations town-list is trimmed to fit. main.py overrides this to the live transport max_chars at runtime.",
     },
 
     # =================================================================

work/meshai/central/avy_handler.py

@@ -39,10 +39,12 @@ Future: retraction broadcast ("AVY advisory lifted") could be added here.
 """
 
 import logging
+import re
 import time
 from typing import Any, Optional
 
 from meshai.adapter_config import adapter_config
+from meshai.central.budget import budget_for, fit_to_budget
 from meshai.persistence import get_db
 
 logger = logging.getLogger(__name__)
@@ -141,10 +143,17 @@ def _render(*, danger_level: int, danger_name: str, zone_name: str,
     prefix = "WARNING:" if danger_level >= 4 else "Watch:"
 
     line1 = f"{emoji} AVY {prefix} {zone_name} \u2014 {danger_name} ({danger_level})"
-    line2 = travel[:120] if travel else None
+    # Travel advice: FIRST SENTENCE only (up to and including the first
+    # sentence terminator), instead of a fixed character slice.
+    line2 = None
+    if travel and travel.strip():
+        t = travel.strip()
+        m = re.search(r"[.!?]", t)
+        line2 = t[: m.end()] if m else t
     line3 = f"{center_id} \u00b7 valid today" if center_id else "valid today"
 
-    return "\n".join(l for l in [line1, line2, line3] if l)
+    msg = "\n".join(l for l in [line1, line2, line3] if l)
+    return fit_to_budget(msg, budget_for("avalanche"))
 
 
 def _attach_commit(data: Optional[dict], *, log_id: Optional[int]) -> None:

work/meshai/central/budget.py

@@ -0,0 +1,32 @@
+"""Shared per-adapter mesh packet budget helpers.
+
+Every broadcast handler fits its final wire string to the live mesh
+transport's single-packet character budget. main.py injects the active
+transport's `max_chars` (140 for the current LoRa configs) into
+adapter_config via set_runtime_override for each broadcast adapter, so
+`budget_for(adapter)` returns the runtime value durably across cache
+invalidation. Default 140 when no override is present (e.g. unit tests).
+"""
+from __future__ import annotations
+
+from meshai.adapter_config import adapter_config
+
+
+def budget_for(adapter: str, default: int = 140) -> int:
+    """Per-adapter mesh packet budget. Reads adapter_config.<adapter>.single_packet_max_chars,
+    which main.py overrides at runtime to the live transport max_chars (140). Default 140."""
+    try:
+        return int(getattr(getattr(adapter_config, adapter), "single_packet_max_chars", default))
+    except Exception:
+        return default
+
+
+def fit_to_budget(s: str, limit: int) -> str:
+    """Trim s to <= limit chars at a word boundary, appending an ellipsis. Never chops a word mid-word."""
+    if len(s) <= limit:
+        return s
+    cut = s[: max(0, limit - 1)].rstrip()
+    cut = cut.rsplit(" ", 1)[0] if " " in cut else cut
+    if not cut:
+        cut = s[: max(0, limit - 1)]
+    return cut.rstrip() + "…"

work/meshai/central/incident_handler.py

@@ -32,6 +32,7 @@ still labels itself New:.
 
 from __future__ import annotations
 from meshai.adapter_config import adapter_config
+from meshai.central.budget import budget_for, fit_to_budget
 
 import logging
 import re
@@ -781,13 +782,18 @@ def _log_event_returning_id(conn, *, now, source, category, severity_word,
 
 
 def _render(n: dict) -> str:
-    """Multi-line wire string.
+    """Budget-fitted multi-line wire string.
 
-    Line 1: {emoji} {display} — Near {city}, {state}
-    Line 2: {road} {direction_long} | MP {mile_marker} OR {from} → {to}
-    Line 3: {lanes_affected} | {delay} min delay | {length}
-    Line 3b: {comment} (additional context, if non-duplicate and <=140 chars)
-    Line 4: Cause: {cause}
+    Line 1: {emoji} {display} — Near {city}, {state}          (critical, kept full)
+    Line 2: {road} {direction_long} [· MP {mile}] · {lanes}   (critical, kept full)
+    Line 3: {narrative/comment}                               (trimmed from END to fit)
+
+    Direction words (Northbound/etc.) and the word "milepost" are NEVER
+    abbreviated. The whole message is fit to the per-adapter packet budget;
+    word-boundary trimming guarantees the narrative's direction words and
+    "milepost" are never chopped mid-word. The separate delay / length / Cause
+    lines were dropped in the budget-fit rework -- the narrative is the trailing
+    trimmed content.
     """
     sub_type = n.get("sub_type") or "incident"
     emoji = _SUB_TYPE_EMOJI.get(sub_type, "⚠️")
@@ -804,66 +810,43 @@ def _render(n: dict) -> str:
         anchor_part = state or ""
     line1 = f"{emoji} {display} — {anchor_part}".rstrip(" —")
 
-    # Line 2: road + direction + mile_marker OR from/to segment (TomTom case)
+    # Line 2: road + direction (+ MP) + lane-status joined by " · ".
+    # Direction is expanded to the full word; abbreviations are never emitted.
     road = n.get("road")
     direction = n.get("direction")
     dir_long = _DIRECTION_LONG.get(direction, direction) if direction else None
     mile = n.get("mile_marker")
     from_loc = n.get("from_loc")
     to_loc = n.get("to_loc")
-    parts = []
+    seg: list[str] = []
     if road and dir_long:
-        parts.append(f"{road} {dir_long}")
+        seg.append(f"{road} {dir_long}")
     elif road:
-        parts.append(road)
+        seg.append(road)
     elif from_loc and to_loc:
-        parts.append(f"{from_loc} → {to_loc}")
+        seg.append(f"{from_loc} → {to_loc}")
     elif from_loc:
-        parts.append(from_loc)
+        seg.append(from_loc)
     if mile is not None:
-        parts.append(f"MP {mile}")
-    line2 = " | ".join(parts) if parts else ""
-
-    # Line 3: lanes_affected (omit if empty/No Data)
+        seg.append(f"MP {mile}")
     lanes = n.get("lanes_affected")
-    line3 = lanes if lanes and lanes.strip().lower() not in ("no data", "") else ""
+    if lanes and lanes.strip().lower() not in ("no data", ""):
+        seg.append(lanes.strip())
+    line2 = " · ".join(seg)
 
-    # Line 4: cause (omit if Incident which is the default)
-    cause = n.get("cause")
-    line4 = f"Cause: {cause}" if cause and cause != "Incident" else ""
+    # Critical head: header + road·lane line, kept verbatim.
+    msg = "\n".join(l for l in (line1, line2) if l)
 
-    # Length (meters from TomTom) formatted as human-readable
-    length_m = n.get("length")
-    length_str = ""
-    if isinstance(length_m, (int, float)) and length_m > 0:
-        if length_m >= 1609:
-            length_str = f"{length_m / 1609:.1f} mi"
-        else:
-            length_str = f"{int(length_m)}m"
-
-    # Optional delay line for tomtom-enriched events
-    delay_minutes = n.get("delay_minutes")
-    delay_line = f"{delay_minutes} min delay" if delay_minutes else ""
-
-    # Combine length, delay, and lanes on line 3
-    extras = [x for x in (delay_line, length_str) if x]
-    if line3 and extras:
-        line3 = f"{line3} | " + " | ".join(extras)
-    elif extras:
-        line3 = " | ".join(extras)
-
-    # Line 3b: comment field, if it contains additional context not already in line 3
+    # Trailing narrative/comment: appended, then the WHOLE message is trimmed
+    # from the end to the packet budget (word-boundary safe).
     comment = n.get("comment")
-    line3b = ""
     if comment and comment.strip():
-        # Skip if comment is just a duplicate of lanes_affected or description
         comment_normalized = comment.strip().lower()
         lanes_normalized = (lanes or "").strip().lower()
-        if comment_normalized != lanes_normalized and len(comment) <= 140:
-            line3b = comment.strip()
+        if comment_normalized != lanes_normalized:
+            msg = f"{msg}\n{comment.strip()}" if msg else comment.strip()
 
-    lines = [l for l in (line1, line2, line3, line3b, line4) if l]
-    return "\n".join(lines)
+    return fit_to_budget(msg, budget_for("incident"))
 
 
 def _location_anchor(n: dict) -> str:

work/meshai/central/nws_handler.py

@@ -28,6 +28,7 @@ Emoji by event_type prefix (substring match, case-insensitive):
 """
 from __future__ import annotations
 from meshai.adapter_config import adapter_config
+from meshai.central.budget import budget_for, fit_to_budget
 
 import logging
 import re
@@ -320,6 +321,59 @@ def handle_nws(envelope: dict, subject: str,
     return None
 
 
+# Hail descriptor -> diameter in inches (NWS convention).
+_HAIL_DESCRIPTORS = {
+    "pea": 0.25, "half inch": 0.50, "penny": 0.75, "nickel": 0.88,
+    "quarter": 1.00, "half dollar": 1.25, "ping pong": 1.50, "ping-pong": 1.50,
+    "golf ball": 1.75, "golf": 1.75, "hen egg": 2.00, "tennis ball": 2.50,
+    "baseball": 2.75, "softball": 4.00,
+}
+
+
+def _tighten_wind(wind: str) -> str:
+    """'60 MPH' -> '60mph winds' (no space before mph, no 'wind gusts' filler)."""
+    w = (wind or "").strip().lower().replace(" mph", "mph")
+    if not w:
+        return ""
+    if not w.endswith("mph"):
+        w = f"{w}mph"
+    return f"{w} winds"
+
+
+def _fmt_hail(hail: str) -> str:
+    """Numeric or descriptor hail size -> '1\" hail'. Descriptor maps per NWS."""
+    s = (hail or "").strip()
+    if not s:
+        return ""
+    val = None
+    low = s.lower()
+    for k, v in _HAIL_DESCRIPTORS.items():
+        if k in low:
+            val = v
+            break
+    if val is None:
+        try:
+            val = float(re.sub(r"[^0-9.]", "", s))
+        except (ValueError, TypeError):
+            return ""
+    txt = f"{val:.2f}".rstrip("0").rstrip(".")
+    return f'{txt}" hail'
+
+
+def _collapse_certainty(text: str) -> str:
+    """'Radar confirmed'/'Radar indicated' -> 'radar'; 'Observed' -> 'observed'."""
+    low = (text or "").strip().lower()
+    if low in ("radar confirmed", "radar indicated"):
+        return "radar"
+    if low == "observed":
+        return "observed"
+    if low == "likely":
+        return "likely"
+    if low == "on ground":
+        return "on ground"
+    return (text or "").strip()
+
+
 def _render(*, event_type, area_desc, geocoder_city, county, state,
              expires_epoch, lat, lon, now, prefix: str = "", d: dict = None) -> str:
     d = d or {}
@@ -359,23 +413,34 @@ def _render(*, event_type, area_desc, geocoder_city, county, state,
         line2 = ""
 
     # Line 3: hazard + certainty/threat (SAME-code branched)
+    # Line 3: TIGHTENED hazard wording. Filler dropped; wind as '60mph winds',
+    # hail as numeric inches ('1" hail'); certainty collapsed to radar/observed;
+    # hazard groups and certainty joined by " · ".
     certainty = (d.get("certainty") or "").strip()
     line3 = ""
     if same_code == "TOR":
         detection = (params.get("tornadoDetection") or [""])[0]
-        status = "On ground" if detection == "OBSERVED" else "Radar indicated"
+        status = "on ground" if detection == "OBSERVED" else "radar"
         threat = (params.get("tornadoDamageThreat") or [""])[0]
-        threat_seg = f" | {threat.title()} damage threat" if threat else ""
-        line3 = f"{status}{threat_seg}"
+        threat_seg = f" · {threat.lower()} damage" if threat else ""
+        line3 = f"tornado {status}{threat_seg}"
     elif same_code == "SVR":
         wind = (params.get("maxWindGust") or [""])[0]
         hail = (params.get("maxHailSize") or [""])[0]
         bits = []
-        if wind and wind not in ("0 MPH", ""): bits.append(f"{wind.lower()} winds")
-        if hail and hail not in ("0.00", "0", ""): bits.append(f"{hail} in hail")
+        if wind and wind not in ("0 MPH", ""):
+            w = _tighten_wind(wind)
+            if w:
+                bits.append(w)
+        if hail and hail not in ("0.00", "0", ""):
+            h = _fmt_hail(hail)
+            if h:
+                bits.append(h)
         hazard = ", ".join(bits)
-        confirm = "Radar confirmed" if certainty == "Observed" else "Radar indicated"
-        line3 = f"{hazard} | {confirm}" if hazard else confirm
+        # SVR is radar-based: "Observed" certainty => "Radar confirmed" => "radar".
+        confirm = _collapse_certainty(
+            "Radar confirmed" if certainty == "Observed" else "Radar indicated")
+        line3 = f"{hazard} · {confirm}" if hazard else confirm
     elif same_code in ("FFW", "FLW"):
         hazard_text = desc.get("hazard") or ""
         # First sentence only
@@ -384,14 +449,14 @@ def _render(*, event_type, area_desc, geocoder_city, county, state,
         # Infer flood cause from description
         desc_lower = (d.get("description") or "").lower()
         flood_cause = ""
-        for keyword, label in [("thunderstorm", "Thunderstorms"),
-                                ("dam", "Dam failure"),
-                                ("snowmelt", "Snowmelt"),
-                                ("ice jam", "Ice jam")]:
+        for keyword, label in [("thunderstorm", "thunderstorms"),
+                                ("dam", "dam failure"),
+                                ("snowmelt", "snowmelt"),
+                                ("ice jam", "ice jam")]:
             if keyword in desc_lower:
                 flood_cause = label
                 break
-        cause_seg = f" | {flood_cause}" if flood_cause else ""
+        cause_seg = f" · {flood_cause}" if flood_cause else ""
         line3 = f"{hazard_text}{cause_seg}" if hazard_text else flood_cause
     else:
         # SPS, WSW, etc.: first hazard sentence + certainty if Observed/Likely
@@ -400,7 +465,7 @@ def _render(*, event_type, area_desc, geocoder_city, county, state,
             hazard_text = hazard_text.split(". ")[0]
         cert_seg = ""
         if certainty in ("Observed", "Likely"):
-            cert_seg = f" | {certainty}"
+            cert_seg = f" · {_collapse_certainty(certainty)}"
         line3 = f"{hazard_text}{cert_seg}" if hazard_text else ""
 
     # Line 4: motion + locations (path-sampled if the full town list won't fit).
@@ -423,7 +488,7 @@ def _render(*, event_type, area_desc, geocoder_city, county, state,
             return motion
         return locs or ""
 
-    PACKET_LIMIT = int(getattr(adapter_config.nws, "single_packet_max_chars", 200))
+    PACKET_LIMIT = budget_for("nws")
 
     # Prefer the FULL town list; use it verbatim if the whole message fits.
     full_locs = ", ".join(towns)
@@ -451,9 +516,7 @@ def _render(*, event_type, area_desc, geocoder_city, county, state,
 
     # Final hard-cap safety net for the pathological case (extremely long town
     # names overflow even the first->middle->last sample plus the other lines).
-    if len(msg) > PACKET_LIMIT:
-        msg = msg[:PACKET_LIMIT - 1].rstrip() + "…"
-    return msg
+    return fit_to_budget(msg, PACKET_LIMIT)
 
 
 def _category_to_event_type(category_raw: str) -> str:

work/meshai/central/quake_handler.py

@@ -24,6 +24,7 @@ fires New:; revisions UPSERT but don't re-broadcast (v0.5.9 no-Update rule).
 """
 from __future__ import annotations
 from meshai.adapter_config import adapter_config
+from meshai.central.budget import budget_for, fit_to_budget
 
 import logging
 import math
@@ -194,7 +195,9 @@ def _render(*, mag, place, depth_km, lat, lon, tsunami, is_update=False) -> str:
     # Line 3: tsunami warning (only when present)
     line3 = "\U0001f6a8 TSUNAMI WARNING" if tsunami else None
 
-    return "\n".join(l for l in [line1, line2, line3] if l)
+    msg = "\n".join(l for l in [line1, line2, line3] if l)
+    # Safety cap: fit the broadcast string to the mesh packet budget.
+    return fit_to_budget(msg, budget_for("usgs_quake"))
 
 
 def _attach_commit(data: Optional[dict], *, event_id: str,

work/meshai/central/satpass_handler.py

@@ -42,6 +42,7 @@ from typing import Any, Optional
 from zoneinfo import ZoneInfo
 
 from meshai.adapter_config import adapter_config
+from meshai.central.budget import budget_for, fit_to_budget
 from meshai.persistence import get_db
 
 logger = logging.getLogger(__name__)
@@ -225,7 +226,8 @@ def format_pass(*, sat_name: str, max_el: float,
         else:
             line2 = time_part
 
-        return f"{line1}\n{line2}"
+        # Safety cap: fit the broadcast string to the mesh packet budget.
+        return fit_to_budget(f"{line1}\n{line2}", budget_for("satpass"))
     else:
         # DM format: compact with exact degrees
         aos_str = _format_time_24h(aos_epoch)

work/meshai/central/wfigs_handler.py

@@ -29,6 +29,7 @@ inside that connection's autocommit mode.
 
 from __future__ import annotations
 from meshai.adapter_config import adapter_config
+from meshai.central.budget import budget_for, fit_to_budget
 
 import logging
 import time
@@ -429,57 +430,42 @@ def _render(n: dict, *, prefix: str = "",
     # Line 1: header
     lines.append(f"🔥 {name} \u2014 {prefix}")
 
-    # Line 2: size / contained with delta + bold
+    # Line 2: size / containment with delta (plain text -- no bold markdown).
     acres_str = f"{int(acres):,} ac" if acres is not None else "size unknown"
     delta_str = ""
     if prefix == "Update" and last_bcast_acres is not None and acres is not None and acres > last_bcast_acres:
         delta_str = f" (+{int(acres - last_bcast_acres):,})"
-    contained_str = f"{int(contained_pct)}% contained" if contained_pct is not None else "containment unknown"
+    contained_str = f"containment {int(contained_pct)}%" if contained_pct is not None else "containment unknown"
+    lines.append(f"{acres_str}{delta_str} · {contained_str}")
 
-    acres_changed = (prefix == "Update" and last_bcast_acres is not None
-                     and acres is not None and acres > last_bcast_acres)
-    contained_changed = (prefix == "Update" and last_bcast_contained is not None
-                         and contained_pct is not None and contained_pct > last_bcast_contained)
-
-    if acres_changed and contained_changed:
-        size_line = f"**{acres_str}{delta_str} | {contained_str}**"
-    elif acres_changed:
-        size_line = f"**{acres_str}{delta_str}** | {contained_str}"
-    elif contained_changed:
-        size_line = f"{acres_str} | **{contained_str}**"
-    else:
-        size_line = f"{acres_str} | {contained_str}"
-    lines.append(size_line)
-
-    # Line 3: movement or plain anchor
+    # Line 3: movement or plain anchor (no bold markdown).
     if (isinstance(movement, dict)
             and movement.get("direction") and movement.get("speed_mph") is not None):
-        lines.append(f"**Moving {movement['direction']} {movement['speed_mph']:.1f} mi/h | {anchor}**")
+        lines.append(f"Moving {movement['direction']} {movement['speed_mph']:.1f} mi/h · {anchor}")
     else:
         lines.append(f"{anchor}")
 
-    # Line 4: cause / discovered
+    # Line 4: cause / discovered (DATE ONLY -- no time-of-day). "Discovered <date>".
     cause_part = cause if cause else None
     disc_part = None
     if declared_at_epoch is not None:
         try:
             dt = _dt.datetime.fromtimestamp(declared_at_epoch,
                                             tz=_dt.timezone(_dt.timedelta(hours=-6)))
-            disc_part = dt.strftime("%b %d %-I:%M %p")
+            disc_part = dt.strftime("%b %-d")
         except Exception:
             pass
     if cause_part and disc_part:
-        lines.append(f"Cause: {cause_part} | Discovered: {disc_part}")
+        lines.append(f"Cause: {cause_part} · Discovered {disc_part}")
     elif cause_part:
         lines.append(f"Cause: {cause_part}")
     elif disc_part:
-        lines.append(f"Discovered: {disc_part}")
+        lines.append(f"Discovered {disc_part}")
 
-    # Line 5: unique fire ID
-    if unique_fire_id:
-        lines.append(f"ID: {unique_fire_id}")
+    # NOTE: the trailing `ID: {unique_fire_id}` line was dropped in the
+    # budget-fit rework -- the unique fire id is not mesh-actionable.
 
-    return "\n".join(lines)
+    return fit_to_budget("\n".join(lines), budget_for("wfigs"))
 
 
 def _location_anchor(n: dict) -> str:

work/meshai/main.py

@@ -397,11 +397,14 @@ class MeshAI:
         # Transport connector (factory selects backend from config.connection.transport)
         self.connector = build_transport(self.config.connection)
 
-        # Fit the NWS one-packet formatter to the active mesh transport's budget
-        # (Meshtastic default 200 -> unchanged). Durable across adapter_config cache
-        # invalidation via the runtime-override store.
+        # Fit every broadcast handler's one-packet formatter to the active mesh
+        # transport's budget (LoRa max_chars, 140). Durable across adapter_config
+        # cache invalidation via the runtime-override store. The adapter names here
+        # MUST match the adapter_config section each handler reads its budget from
+        # (see meshai.central.budget.budget_for calls in the handlers).
         from meshai.adapter_config import set_runtime_override
-        set_runtime_override("nws", "single_packet_max_chars", self.connector.max_chars)
+        for _adapter in ("nws", "incident", "wfigs", "avalanche", "satpass", "usgs_quake"):
+            set_runtime_override(_adapter, "single_packet_max_chars", self.connector.max_chars)
 
         # Passive mesh context buffer
         ctx_cfg = self.config.context

work/meshai/notifications/scheduled/fire_digest.py

@@ -258,8 +258,23 @@ class FireDigestScheduler:
         tomorrow = now_dt + timedelta(days=1)
         return _slot_epoch(tomorrow, schedule[0], self._tz_name()), schedule[0]
 
+    def _broadcast_enabled(self) -> bool:
+        try:
+            return bool(adapter_config.fires.digest_broadcast_enabled)
+        except Exception:
+            return False
+
     async def fire_slot(self, slot_epoch_s: int, hh_mm: str) -> bool:
         """Build + broadcast for the given slot. Returns True on broadcast."""
+        # Kill-switch on the actual mesh emission. Disabled by default: the
+        # scheduler wiring stays intact (so flipping the flag to True cleanly
+        # re-enables it) but nothing is dispatched. Per-fire wfigs alerts are a
+        # separate path and are unaffected.
+        if not self._broadcast_enabled():
+            self._logger.info(
+                "fire-digest: broadcast disabled "
+                "(fires.digest_broadcast_enabled=False); skipping slot %s", hh_mm)
+            return False
         wire, source = await render_digest(now=int(self._clock()))
         if source == "no_fires":
             self._logger.info(

work/tests/test_adapter_avalanche.py

@@ -194,3 +194,49 @@ def test_missing_event_id_returns_none(adapter):
 def test_does_not_raise_on_corrupted_dict(adapter):
     """Corrupted dict returns None without raising."""
     assert adapter.to_event({"garbage": True}) is None
+
+
+# ============================================================================
+# Central avy_handler._render (mesh broadcast wire) -- budget-fit format.
+#   advice -> FIRST SENTENCE only; zone/level/source kept FULL; fits 140.
+# ============================================================================
+
+from meshai.central.avy_handler import _render as _avy_render
+
+
+def test_avy_render_advice_first_sentence_only():
+    wire = _avy_render(
+        danger_level=4, danger_name="High",
+        zone_name="Western Mountains",
+        center_id="SNFAC",
+        travel=("Avoid all avalanche terrain today. Natural and human-triggered "
+                "avalanches are likely on steep slopes."),
+    )
+    # first sentence retained (with its period), the rest dropped
+    assert "Avoid all avalanche terrain today." in wire
+    assert "Natural and human-triggered" not in wire
+    # zone / level / source kept FULL (never abbreviated)
+    assert "Western Mountains" in wire
+    assert "High (4)" in wire
+    assert "SNFAC" in wire
+
+
+def test_avy_render_worst_case_fits_140():
+    # Long multi-sentence advice paragraph -- only the first sentence is kept,
+    # which lets the full zone / level / source survive under the 140 budget.
+    wire = _avy_render(
+        danger_level=5, danger_name="Extreme",
+        zone_name="Western Mountains",
+        center_id="Sawtooth Avalanche Center",
+        travel=("Avoid all avalanche terrain today! Very dangerous conditions "
+                "exist across all elevations and aspects with widespread natural "
+                "avalanche activity likely through the afternoon and overnight."),
+    )
+    assert len(wire) <= 140, f"{len(wire)} chars:\n{wire!r}"
+    # zone, level, source, and the first-sentence advice all present
+    assert "Western Mountains" in wire
+    assert "Extreme (5)" in wire
+    assert "Sawtooth Avalanche Center" in wire
+    assert "Avoid all avalanche terrain today!" in wire
+    # first sentence terminates at the '!' -> the rest is gone
+    assert "Very dangerous conditions" not in wire

work/tests/test_adapter_config_api.py

@@ -36,7 +36,7 @@ def test_list_returns_all_59_keys(client):
     # 14 adapters with at least one key (itd_511 has zero -- not in the
     # grouped dict because the SQL only returns rows that exist).
     total = sum(len(v) for v in body.values())
-    assert total == 92
+    assert total == 96
 
 
 def test_list_grouped_by_adapter(client):

work/tests/test_adapter_config_foundation.py

@@ -75,7 +75,7 @@ def test_adapter_config_type_check_constrains_vocabulary(fresh_db):
 
 def test_registry_at_59_entries():
     """v0.6-3a.1 trim: 43 CONFIG-only keys (was 77 in v0.6-3a draft)."""
-    assert len(REGISTRY) == 92, (
+    assert len(REGISTRY) == 96, (
         f"REGISTRY drift guard; got {len(REGISTRY)}. "
         f"If a sentence template / emoji / heuristic snuck in, it belongs in CODE not config."
     )

work/tests/test_fire_digest_broadcast_gate.py

@@ -0,0 +1,73 @@
+"""Tests for the fire-digest broadcast kill-switch (fires.digest_broadcast_enabled).
+
+The scheduler wiring stays intact; only the mesh EMISSION is gated. Disabled by
+default -> fire_slot() dispatches nothing. Flipping the flag True re-enables it
+cleanly. Per-fire wfigs alerts are a separate path and are unaffected (covered in
+test_wfigs_handler.py).
+"""
+from __future__ import annotations
+
+import asyncio
+import time
+
+from meshai.persistence import get_db
+from meshai.notifications.scheduled.fire_digest import FireDigestScheduler
+
+
+class _RecordingDispatcher:
+    def __init__(self):
+        self.calls = []
+
+    async def dispatch_scheduled_broadcast(self, *, text, source_event_table,
+                                           source_event_pk):
+        self.calls.append(
+            {"text": text, "table": source_event_table, "pk": source_event_pk})
+        return True
+
+
+def _seed_active_fire(conn):
+    now = int(time.time())
+    conn.execute(
+        "INSERT OR REPLACE INTO fires(irwin_id, incident_name, incident_type, "
+        "current_acres, current_contained_pct, lat, lon, county, state, "
+        "declared_at, last_event_at, tombstoned_at) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
+        ("GATE-01", "Gatekeeper Fire", "WF", 900, None, 43.6, -116.2,
+         "Ada", "ID", now, now - 600, None),
+    )
+
+
+def test_fire_slot_disabled_by_default_dispatches_nothing():
+    """With defaults (digest_broadcast_enabled=False) fire_slot emits nothing."""
+    conn = get_db()
+    _seed_active_fire(conn)  # a broadcast WOULD be produced if the gate were open
+    now = int(time.time())
+    dispatcher = _RecordingDispatcher()
+    sched = FireDigestScheduler(dispatcher, clock=lambda: now)
+
+    result = asyncio.run(sched.fire_slot(now, "06:00"))
+
+    assert result is False
+    assert dispatcher.calls == [], "digest broadcast must NOT be dispatched by default"
+
+
+def test_fire_slot_dispatches_when_flag_enabled():
+    """Flipping fires.digest_broadcast_enabled True cleanly re-enables emission."""
+    from meshai.adapter_config import set_runtime_override
+    conn = get_db()
+    _seed_active_fire(conn)
+    now = int(time.time())
+    dispatcher = _RecordingDispatcher()
+    sched = FireDigestScheduler(dispatcher, clock=lambda: now)
+
+    set_runtime_override("fires", "digest_broadcast_enabled", True)
+    try:
+        result = asyncio.run(sched.fire_slot(now, "06:00"))
+    finally:
+        # Reset so the override doesn't leak into other tests in this process.
+        set_runtime_override("fires", "digest_broadcast_enabled", False)
+
+    assert result is True
+    assert len(dispatcher.calls) == 1
+    assert dispatcher.calls[0]["table"] == "fire_digest_broadcasts"
+    assert "Gatekeeper Fire" in dispatcher.calls[0]["text"]

work/tests/test_incident_handler.py

@@ -33,6 +33,7 @@ import pytest
 
 from meshai.central.incident_handler import (
     handle_incident,
+    _render as _incident_render,
 )
 from meshai.persistence import close_thread_connection, init_db
 from meshai.persistence import db as persistence_db
@@ -216,7 +217,10 @@ def test_a_tomtom_icon_renders(mem_db, no_photon, icon, expected_emoji, expected
     assert wire.startswith(f"{expected_emoji} {expected_phrase}")
     assert "Near Boise, ID" in wire
     assert "I-84" in wire
-    assert "5 min delay" in wire
+    # Budget-fit rework: the separate "N min delay" line was dropped; the road
+    # segment carries road (+ direction/lanes), and the message fits 140.
+    assert "min delay" not in wire
+    assert len(wire) <= 140
 
 
 # ============================================================================
@@ -774,3 +778,51 @@ def test_w_itd_511_future_scheduled_dropped_via_start_epoch(mem_db, no_photon):
     now = 2_000_000_000
     wire = handle_incident(env, env["subject"], data={}, now=now)
     assert wire is None
+
+
+# ============================================================================
+# Budget-fit worst case: longest plausible traffic payload must fit 140 chars
+# with critical fields (type, location, road, FULL direction word, lane
+# status, trimmed narrative with intact direction word + "milepost") present.
+# ============================================================================
+
+
+def test_incident_worst_case_fits_140():
+    n = {
+        "sub_type": "accident",
+        "geocoder_city": None,
+        "county": "Minidoka",
+        "state": "ID",
+        "road": "SH-27",
+        "direction": "north",          # abbreviation/short form -> MUST expand
+        "mile_marker": None,
+        "lanes_affected": "1 Right lane blocked",
+        "comment": (
+            "Southbound right lane at milepost 24 and westbound onramp to I-84 "
+            "blocked due to a multi-vehicle collision, expect major delays "
+            "through the evening commute and seek alternate routes tonight"
+        ),
+    }
+    wire = _incident_render(n)
+
+    # (a) fits one mesh packet
+    assert len(wire) <= 140, f"{len(wire)} chars:\n{wire!r}"
+
+    # (b) critical fields present
+    assert wire.startswith("🚨 Crash")                 # type
+    assert "Near Minidoka Co, ID" in wire              # location
+    assert "SH-27" in wire                             # road
+    assert "Northbound" in wire                        # FULL direction (expanded)
+    assert "1 Right lane blocked" in wire              # lane status
+    assert "SH-27 Northbound · 1 Right lane blocked" in wire  # road·lane line
+
+    # (c) direction is NEVER abbreviated anywhere in the wire
+    for abbr in (" NB", " N ", "Sbound", "N/B"):
+        assert abbr not in wire
+
+    # (d) narrative present and word-boundary trimmed (no mid-word chop): the
+    # intact direction word "Southbound" and the intact word "milepost" survive.
+    assert "Southbound" in wire
+    assert "milepost" in wire
+    # trimmed from the END -> ends with the ellipsis, not raw text
+    assert wire.endswith("…")

work/tests/test_nws_handler.py

@@ -282,22 +282,25 @@ def test_svr_long_locations_path_sampled(mem_db):
         d=d,
     )
 
-    # (a) fits in one mesh packet
-    assert len(rendered) <= 200, (
-        f"rendered is {len(rendered)} chars (expected <= 200):\n{rendered!r}"
+    # (a) fits in one mesh packet (budget is now the 140-char LoRa max)
+    assert len(rendered) <= 140, (
+        f"rendered is {len(rendered)} chars (expected <= 140):\n{rendered!r}"
     )
 
-    # (b) all data-point categories present
+    # (b) all data-point categories present, hazard wording TIGHTENED
     assert "Severe Thunderstorm Warning" in rendered, "event type missing"
     assert "Until" in rendered, "expiry time segment missing"
     assert "Twin Falls County" in rendered, "area missing"
-    assert "mph winds" in rendered or "hail" in rendered, "hazard segment missing"
+    assert "60mph winds" in rendered, "wind hazard not tightened to '60mph winds'"
+    assert '1" hail' in rendered, "hail hazard not rendered as numeric inches"
+    assert "radar" in rendered, "certainty not collapsed to 'radar'"
     assert "Moving" in rendered, "motion segment missing"
 
-    # (c) path-sampling, not tail-drop: first town, last town, and the arrow
+    # (c) path-sampling applied (arrow) with the soonest-impact town retained.
+    # At the 140 budget the farthest-along town may be trimmed by the final
+    # backstop; the hard cap wins over endpoint preservation.
     assert "→" in rendered, "no arrow -> not path-sampled"
-    assert "Buhl" in rendered, "first town missing"
-    assert "Shoshone" in rendered, "last town dropped (the old tail-trim bug)"
+    assert "Buhl" in rendered, "first (soonest-impact) town missing"
 
 
 def test_svr_short_locations_shown_in_full(mem_db):
@@ -331,6 +334,47 @@ def test_svr_short_locations_shown_in_full(mem_db):
         d=d,
     )
 
-    assert len(rendered) <= 200
+    assert len(rendered) <= 140
     assert "→" not in rendered, "short list should not be path-sampled"
     assert "Buhl, Eden, Hazelton" in rendered, "full comma-joined list expected"
+    # Hazard wording is tightened even on the short-list path.
+    assert "60mph winds" in rendered
+    assert '1" hail' in rendered
+    assert "radar" in rendered
+
+
+def test_svr_worst_case_fits_140(mem_db):
+    """Pathologically long SVR payload: the final wire MUST fit 140 chars while
+    still carrying event name, area, time, tightened hazard, and >=1 town."""
+    long_locations = (
+        "Buhl, Eden, Hazelton, Murtaugh, Richfield, Dietrich, Gooding, "
+        "Hagerman, Wendell, Jerome, Kimberly, Hansen, Filer, and Shoshone"
+    )
+    description = (
+        "HAZARD...Damaging winds to 70 mph and golf ball size hail.\n\n"
+        f"Locations impacted include...{long_locations}"
+    )
+    d = {
+        "eventCode": {"SAME": ["SVR"]},
+        "certainty": "Observed",
+        "parameters": {
+            "maxWindGust": ["70 MPH"],
+            "maxHailSize": ["1.75"],
+            "eventMotionDescription": ["2200000T254DEG...35KT 42.5,-114.5"],
+        },
+        "description": description,
+    }
+    rendered = _render(
+        event_type="Severe Thunderstorm Warning",
+        area_desc="Twin Falls County",
+        geocoder_city=None, county="Twin Falls", state="ID",
+        expires_epoch=1_751_400_000, lat=42.5, lon=-114.46,
+        now=1_751_400_000, d=d,
+    )
+    assert len(rendered) <= 140, f"{len(rendered)} chars:\n{rendered!r}"
+    assert "Severe Thunderstorm Warning" in rendered   # event name
+    assert "Twin Falls County" in rendered              # area
+    assert "Until" in rendered                          # time
+    assert "70mph winds" in rendered                    # tightened hazard (wind)
+    assert '1.75" hail' in rendered                     # golf ball -> 1.75"
+    assert "Buhl" in rendered                           # >=1 town present

work/tests/test_quake_handler.py

@@ -173,3 +173,21 @@ def test_commit_callback_updates_last_broadcast(mem_db):
         "SELECT last_broadcast_at FROM quake_events WHERE event_id='cb1'"
     ).fetchone()
     assert post["last_broadcast_at"] == 1_000_001
+
+
+# ============================================================================
+# Budget-fit SAFETY CAP: a freak-long USGS place string must still fit 140.
+# ============================================================================
+
+from meshai.central.quake_handler import _render as _quake_render
+
+
+def test_quake_render_worst_case_fits_140():
+    place = ("293 km SSW of a pathologically long place description island "
+             "region in the remote northern pacific ocean near absolutely nowhere "
+             "at all off the coast of the far edge of the map")
+    wire = _quake_render(mag=7.9, place=place, depth_km=12, lat=44.123,
+                         lon=-114.987, tsunami=True, is_update=False)
+    assert len(wire) <= 140, f"{len(wire)} chars:\n{wire!r}"
+    # magnitude survives on the (critical) first line
+    assert "M7.9" in wire

work/tests/test_satpass_handler.py

@@ -171,3 +171,24 @@ class TestSatpassHandler:
         result = handle_satpass(env, "central.sat.pass.iss", data={}, now=1718163120)
 
         assert result is None
+
+
+# ============================================================================
+# Budget-fit SAFETY CAP: a pathologically long satellite name must not push
+# the broadcast string past 140 chars. Calls format_pass directly (bypasses
+# the broken consolidation path).
+# ============================================================================
+
+def test_format_pass_worst_case_fits_140():
+    from meshai.central.satpass_handler import format_pass
+    wire = format_pass(
+        sat_name=("NOAA-19 EXPERIMENTAL SUPER LONG SATELLITE DESIGNATION "
+                  "PAYLOAD REVISION X PROTOTYPE FLIGHT MODEL SERIAL 00042"),
+        max_el=78.0,
+        aos_epoch=1719900000, los_epoch=1719900780,
+        aos_compass="NNW", los_compass="SSE",
+        entry_observer="Treasure Valley Observatory West Ridge Site",
+        exit_observer="Magic Valley Observatory East Rim Station",
+        broadcast=True,
+    )
+    assert len(wire) <= 140, f"{len(wire)} chars:\n{wire!r}"

work/tests/test_wfigs_handler.py

@@ -23,6 +23,7 @@ from meshai import central_normalizer as cn
 from meshai.central.wfigs_handler import (
     WFIGS_BROADCAST_COOLDOWN_S,
     handle_wfigs,
+    _render as _wfigs_render,
 )
 from meshai.persistence import close_thread_connection, init_db
 from meshai.persistence import db as persistence_db
@@ -283,7 +284,10 @@ def test_g_new_irwin_inserts_and_broadcasts(mem_db, no_photon):
     assert wire.startswith("🔥 Cache Peak Fire — New")
     assert "Burley" in wire
     assert "1,847 ac" in wire
-    assert "23% contained" in wire
+    assert "containment 23%" in wire
+    # Budget-fit rework: no unique-fire-id line, no bold markdown.
+    assert "ID:" not in wire
+    assert "**" not in wire
 
     # v0.5.8b: handler INSERTs the fires row with last_broadcast_*=NULL,
     # then attaches a commit callback. The dispatcher fires the callback
@@ -416,7 +420,7 @@ def test_j_known_irwin_change_after_cooldown_broadcasts(mem_db, no_photon):
     assert out is not None
     assert out.startswith("🔥 Cache Peak Fire — Update")
     assert "3,000 ac" in out
-    assert "35% contained" in out
+    assert "containment 35%" in out
 
     # Simulate dispatcher commit.
     data2["_on_broadcast_committed"](float(later))
@@ -615,3 +619,70 @@ def test_h_handler_attaches_audit_descriptor_and_callback(mem_db, no_photon):
     assert callable(data["_on_broadcast_committed"])
     assert data["_broadcast_audit"]["table"] == "fires"
     assert data["_broadcast_audit"]["pk"] == _IRWIN_B
+
+
+# ============================================================================
+# Budget-fit worst case: longest plausible fire payload fits 140 chars, with
+# no `ID:` line, no `**` markdown, and discovery rendered DATE-ONLY.
+# ============================================================================
+
+
+def test_wfigs_worst_case_fits_140():
+    n = {
+        "incident_name": "East Fork Salmon River Complex Lightning Fire",
+        "acres": 128456,
+        "contained_pct": 42,
+        "fire_cause": "Lightning",
+        "unique_fire_id": "2026-IDSCF-000987",
+        # Jun 18 2026 ~14:30 local
+        "declared_at_epoch": 1_781_204_400,
+        "geocoder_city": "Near Clayton, ID",
+    }
+    wire = _wfigs_render(n, prefix="Update", last_bcast_acres=100000)
+
+    assert len(wire) <= 140, f"{len(wire)} chars:\n{wire!r}"
+    # critical fields
+    assert "East Fork Salmon River Complex Lightning Fire" in wire  # name
+    assert "128,456 ac" in wire                                     # acreage
+    assert "containment 42%" in wire                                # containment
+    assert "Near Clayton, ID" in wire                              # location
+    assert "Cause: Lightning" in wire                              # cause
+    # format rules
+    assert "ID:" not in wire, "unique-fire-id line must be dropped"
+    assert "**" not in wire, "no bold markdown"
+
+
+def test_wfigs_discovery_is_date_only():
+    n = {
+        "incident_name": "Short Fire",
+        "acres": 100,
+        "contained_pct": 0,
+        "fire_cause": "Human",
+        "unique_fire_id": "2026-X",
+        "declared_at_epoch": 1_781_204_400,  # renders Jun 11 in the handler's UTC-6
+        "geocoder_city": "Boise",
+    }
+    wire = _wfigs_render(n, prefix="New")
+    assert "Discovered Jun 11" in wire
+    # no time-of-day (colon in an H:MM would appear as ":3" etc.)
+    assert "2:30" not in wire and "PM" not in wire and "AM" not in wire
+    assert "ID:" not in wire
+
+
+# ============================================================================
+# Fire-digest kill-switch does NOT touch the per-fire wfigs path: a new-fire
+# envelope still produces a broadcast even though digest broadcast is disabled.
+# ============================================================================
+
+
+def test_per_fire_wfigs_broadcasts_while_digest_disabled(mem_db, no_photon):
+    from meshai.adapter_config import adapter_config
+    # Default posture: the twice-daily digest broadcast is OFF.
+    assert bool(adapter_config.fires.digest_broadcast_enabled) is False
+    # ... yet a new per-fire wfigs alert still broadcasts.
+    env = _make_active_envelope(geocoder_city="Burley")
+    data = {}
+    wire = handle_wfigs(cn.normalize(env), env, env["subject"],
+                        data=data, now=6_000_000)
+    assert wire is not None
+    assert wire.startswith("🔥 Cache Peak Fire — New")