refactor(phase0): source-agnostic formatter/gating scaffold + harness (inert)

Foundation for making all hazard formatting+gating source-agnostic. ZERO
behavior change — the formatter/decider registries are empty (get_formatter/
get_decider return None → existing precomposed/Mode-B path preserved), and the
shadow comparator is off unless MESHAI_SHADOW_CATEGORIES is set.

- notifications/formatters/ (registry+dispatch with family fallback), gating/
  (GateResult + deferred-commit contract), both empty registries.
- notifications/clock.py determinism seam; route wfigs/quake/nws gating time
  reads through it (identical values) so goldens can freeze time.
- formatters/_budget.py = copy of central/budget.py; central/budget.py is now a
  re-export shim (import-smoke test guards it).
- compose_mesh_message consults the registry first (verbatim, no Mode-B re-cap),
  falls back to legacy; _resolve_budget injects per-category budget.
- notifications/shadow.py + two DRY-RUN hooks (consumer._normalize, dispatcher
  render): compute the new result and diff-log SHADOW_MISMATCH JSONL, but NEVER
  commit/emit/write tables and always broadcast the OLD result. Inert by default.
- tests/harness (pinned_time/pinned_tz, byte-golden, gate-sequence) +
  scripts/capture_fixtures.py (ephemeral read-only NATS capture); tzdata pinned.

Tests: +60 (18 scaffold + 42 harness/shadow); 0 new failures (34 baseline).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/central/budget.py

@@ -1,32 +1,9 @@
-"""Shared per-adapter mesh packet budget helpers.
+"""Re-export shim — implementation has moved to meshai.notifications.formatters._budget.
 
-Every broadcast handler fits its final wire string to the live mesh
+Every existing ``from meshai.central.budget import budget_for, fit_to_budget``
-transport's single-packet character budget. main.py injects the active
+import continues to work unchanged.  This shim is the sole consumer of the
-transport's `max_chars` (140 for the current LoRa configs) into
+canonical implementation; update the implementation there, not here.
-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.notifications.formatters._budget import budget_for, fit_to_budget
 
-from meshai.adapter_config import adapter_config
+__all__ = ["budget_for", "fit_to_budget"]
-
-
-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/consumer.py

@@ -590,6 +590,22 @@ class CentralConsumer:
         # Scheduled broadcasters (band_conditions) bypass _normalize()
         # entirely -- they enter via Dispatcher.dispatch_scheduled_broadcast()
         # and are unaffected by this gate.
+        # Phase-0b shadow gate hook — inert by default.
+        # Called RIGHT BEFORE the default-deny return so the shadow observes the
+        # same broadcast decision the legacy handler made (synthesized is not None).
+        # The OLD result is kept unchanged below; the shadow only observes.
+        try:
+            from meshai.notifications.shadow import shadow_gate as _shadow_gate
+            _shadow_gate(
+                category,
+                data,
+                source=inner.get("adapter") or "central",
+                now=time.time(),
+                old_broadcast=synthesized is not None,
+            )
+        except Exception:  # noqa: BLE001 — shadow must never affect production
+            pass
+
         if synthesized is None:
             logger.debug(
                 "consumer: default-deny -- no handler synthesized for "

work/meshai/central/nws_handler.py

@@ -37,6 +37,8 @@ import zoneinfo
 from datetime import datetime, timezone
 from typing import Any, Optional
 
+from meshai.notifications import clock
+
 from meshai.persistence import get_db
 
 logger = logging.getLogger(__name__)
@@ -130,7 +132,7 @@ def _parse_motion(params: dict) -> tuple:
     return compass, mph
 
 
-def _now() -> int: return int(time.time())
+def _now() -> int: return int(clock.now())
 
 
 def _is_update(conn, d: dict) -> bool:

work/meshai/central/quake_handler.py

@@ -31,6 +31,8 @@ import math
 import time
 from typing import Any, Optional
 
+from meshai.notifications import clock
+
 from meshai.persistence import get_db
 
 logger = logging.getLogger(__name__)
@@ -41,7 +43,7 @@ logger = logging.getLogger(__name__)
 # GUI edits take effect on the next envelope without restart.
 
 
-def _now() -> int: return int(time.time())
+def _now() -> int: return int(clock.now())
 
 
 def _haversine_mi(lat1, lon1, lat2, lon2) -> float:

work/meshai/central/wfigs_handler.py

@@ -35,6 +35,8 @@ import logging
 import time
 from typing import Any, Optional
 
+from meshai.notifications import clock
+
 from meshai.persistence import get_db
 
 logger = logging.getLogger(__name__)
@@ -62,7 +64,7 @@ def _cleanup_stale_fires(conn) -> None:
 
 
 def _now() -> int:
-    return int(time.time())
+    return int(clock.now())
 
 
 def _fire_too_old_to_announce(declared_at_epoch, now) -> bool:

work/meshai/notifications/clock.py

@@ -0,0 +1,23 @@
+"""Determinism seam for time reads across the notification pipeline.
+
+All handlers and formatters should call clock.now() / clock.now_dt() instead
+of time.time() / datetime.now() directly.  A single monkeypatch point makes
+golden-file tests and time-freeze fixtures trivial:
+
+    monkeypatch.setattr("meshai.notifications.clock.now", lambda: 1_700_000_000.0)
+
+clock.now() is semantically identical to time.time() at runtime.
+"""
+
+import time
+from datetime import datetime
+
+
+def now() -> float:
+    """Return the current Unix timestamp (float). Monkeypatchable seam."""
+    return time.time()
+
+
+def now_dt(tz=None):
+    """Return the current datetime (optionally timezone-aware). Monkeypatchable seam."""
+    return datetime.now(tz)

work/meshai/notifications/formatters/__init__.py

@@ -0,0 +1,55 @@
+"""Formatter registry for the Phase-1+ mesh-message dispatch path.
+
+FORMATTERS is intentionally empty at this phase (Phase 0 scaffold).
+No category is migrated yet — all events fall through to the legacy
+compose_mesh_message Mode-B path.  Zero behavior change.
+
+Usage (future phases):
+    from meshai.notifications.formatters import register
+
+    @register("earthquake_event")
+    def _fmt_quake(event, *, now: float, budget: int) -> str:
+        ...
+"""
+
+from typing import Callable, Optional
+
+# Empty registry — populated by per-category formatter modules (Phase 1+).
+FORMATTERS: dict[str, Callable] = {}
+
+
+def register(category: str, fn: Callable) -> Callable:
+    """Register a formatter callable for a category (or family toggle name).
+
+    Decorates or calls directly:
+        register("earthquake_event", my_fn)
+        @register("earthquake_event")
+        def my_fn(...): ...
+    """
+    FORMATTERS[category] = fn
+    return fn
+
+
+def get_formatter(category: str) -> Optional[Callable]:
+    """Return the formatter for *category*, or None if none is registered.
+
+    Resolution order:
+      1. Direct category match in FORMATTERS.
+      2. Family/toggle fallback: look up the category's toggle name via
+         get_toggle(), then check FORMATTERS for that toggle key.
+      3. None — caller falls through to legacy Mode-B composition.
+    """
+    fn = FORMATTERS.get(category)
+    if fn is not None:
+        return fn
+    # Family fallback — mirrors _category_label()'s toggle lookup in composer.py.
+    try:
+        from meshai.notifications.categories import get_toggle
+        tog = get_toggle(category)
+        if tog:
+            fn = FORMATTERS.get(tog)
+            if fn is not None:
+                return fn
+    except Exception:
+        pass
+    return None

work/meshai/notifications/formatters/_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/notifications/gating/__init__.py

@@ -0,0 +1,50 @@
+"""Gating/budgeting decision registry for the Phase-1+ dispatch path.
+
+DECIDERS is intentionally empty at this phase (Phase 0 scaffold).
+No category is migrated yet — all gating logic remains in the existing
+handler modules (wfigs_handler, nws_handler, etc.).  Zero behavior change.
+
+Usage (future phases):
+    from meshai.notifications.gating import register
+    from meshai.notifications.gating.base import GateResult
+
+    @register("earthquake_event")
+    def _gate_quake(event, *, now: float) -> GateResult:
+        ...
+"""
+
+from typing import Callable, Optional
+
+# Empty registry — populated by per-category gating modules (Phase 1+).
+DECIDERS: dict = {}
+
+
+def register(category: str, fn: Callable) -> Callable:
+    """Register a gating callable for a category (or family toggle name)."""
+    DECIDERS[category] = fn
+    return fn
+
+
+def get_decider(category: str) -> Optional[Callable]:
+    """Return the gating decider for *category*, or None if none is registered.
+
+    Resolution order:
+      1. Direct category match in DECIDERS.
+      2. Family/toggle fallback: look up the category's toggle name via
+         get_toggle(), then check DECIDERS for that toggle key.
+      3. None — caller falls through to legacy per-handler gating logic.
+    """
+    fn = DECIDERS.get(category)
+    if fn is not None:
+        return fn
+    # Family fallback — mirrors the toggle lookup pattern in composer.py.
+    try:
+        from meshai.notifications.categories import get_toggle
+        tog = get_toggle(category)
+        if tog:
+            fn = DECIDERS.get(tog)
+            if fn is not None:
+                return fn
+    except Exception:
+        pass
+    return None

work/meshai/notifications/gating/base.py

@@ -0,0 +1,49 @@
+"""GateResult — the return type for all gating/budgeting decision functions.
+
+A decider registered in meshai.notifications.gating.DECIDERS receives an
+Event and returns a GateResult.  The dispatcher inspects `broadcast` to
+decide whether to send, merges `data_patch` into event.data, and calls
+`commit(now)` exactly once per channel on confirmed delivery.
+
+Deferred-commit contract
+------------------------
+`commit(now: float)` is called ONCE, ONLY after the message has been
+successfully delivered to a mesh channel.  Rules:
+
+  * Idempotent: the callback MUST use UPSERTs (INSERT OR REPLACE / ON
+    CONFLICT DO UPDATE), never bare INSERTs, so a retry on failure does
+    not create duplicate rows.
+
+  * One call per channel: if the same event is broadcast on N mesh
+    channels, commit is called N times (once per successful deliver()).
+    The UPSERT requirement makes that safe.
+
+  * No exceptions may escape commit() — wrap all DB writes in try/except
+    and log failures; a crashing commit would abort the send loop.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Callable, Optional
+
+
+@dataclass
+class GateResult:
+    """Decision from a gating function.
+
+    Attributes:
+        broadcast: True → send the event; False → suppress.
+        lifecycle: Short label for logging/audit (e.g. "new", "update",
+                   "cooldown", "suppress").
+        reason:    Human-readable explanation used in debug logs.
+        data_patch: Dict merged into event.data before the message is
+                    rendered and sent (use for attaching callbacks, etc.).
+        commit:    Optional callable invoked once per channel on confirmed
+                   delivery.  Signature: commit(now: float) -> None.
+                   See module docstring for the full contract.
+    """
+    broadcast: bool
+    lifecycle: str = ""
+    reason: str = ""
+    data_patch: dict = field(default_factory=dict)
+    commit: Optional[Callable[[float], None]] = None

work/meshai/notifications/pipeline/dispatcher.py

@@ -443,6 +443,16 @@ class Dispatcher:
             self._logger.exception("mesh composer crashed; falling back to legacy message")
             friendly = None
 
+        # Phase-0b shadow render hook — inert by default.
+        # Called AFTER compose_mesh_message so old_wire is the real produced
+        # string.  The real path continues to use *friendly* unchanged below.
+        if friendly is not None:
+            try:
+                from meshai.notifications.shadow import shadow_render as _shadow_render
+                _shadow_render(event.category, event, old_wire=friendly)
+            except Exception:  # noqa: BLE001 — shadow must never affect production
+                pass
+
         delivered_any = False
         for ch_type in ch_types:
             rule = None

work/meshai/notifications/renderers/composer.py

@@ -20,12 +20,15 @@ exceed the budget, the primary identifier is shrunk by codepoints and
 suffixed with `…` so the byte budget always holds.
 """
 
+import logging
 import re
 from html import unescape
 from typing import Optional
 
 from meshai.notifications.events import Event
 
+logger = logging.getLogger(__name__)
+
 
 # Hard byte budget for a single mesh broadcast line (Matt-approved cap).
 _BYTE_BUDGET = 150
@@ -294,6 +297,25 @@ def _context_segment(event: Event) -> Optional[str]:
     return ", ".join(bits) if bits else None
 
 
+def _resolve_budget(event: Event) -> int:
+    """Pick a per-adapter packet budget for the new formatter path.
+
+    Mirrors how the existing precomposed handlers select their budget:
+    wfigs uses budget_for("wfigs"), quake uses budget_for("usgs_quake"),
+    etc.  Here we use event.source (the adapter name) as the budget key.
+    Falls back to 140 (the universal LoRa default) when source is absent
+    or budget_for raises.
+    """
+    src = (getattr(event, "source", "") or "").strip()
+    if src:
+        try:
+            from meshai.central.budget import budget_for
+            return budget_for(src)
+        except Exception:
+            pass
+    return 140
+
+
 def compose_mesh_message(event: Event) -> str:
     """Compose a friendly mesh-broadcast string with 150-byte UTF-8 hard cap.
 
@@ -306,6 +328,22 @@ def compose_mesh_message(event: Event) -> str:
     Return it verbatim -- no family-label prefix, no region tail, no
     severity word append.
     """
+    # Phase-1+ formatter dispatch — EMPTY REGISTRY at Phase 0, so this
+    # no-ops on every call.  When a formatter is registered for the event's
+    # category (or its toggle family), it is called here and its output
+    # returned verbatim (newlines preserved, no Mode-B re-entry).
+    from meshai.notifications.formatters import get_formatter
+    from meshai.notifications import clock
+    fmt = get_formatter(event.category)
+    if fmt is not None:
+        try:
+            return fmt(event, now=clock.now(), budget=_resolve_budget(event))
+        except Exception:
+            logger.exception(
+                "formatter failed for %s; falling back to legacy", event.category
+            )
+    # ---- existing passthrough + Mode-B unchanged below ----
+
     if event.data and event.data.get("_meshai_precomposed") and event.title:
         return event.title
 

work/meshai/notifications/shadow.py

@@ -0,0 +1,221 @@
+"""Phase-0b dry-run shadow comparator.
+
+Gated by env var MESHAI_SHADOW_CATEGORIES (comma-separated category names).
+Empty/unset  → fully OFF (zero overhead, zero side-effects on the real path).
+
+Safety contract — this module MUST uphold all of the following:
+  * NEVER calls GateResult.commit()
+  * NEVER calls bus.emit()
+  * NEVER writes to any DB table
+  * NEVER returns a value that the caller uses for a routing decision
+  * All exceptions swallowed + logged at DEBUG (shadow must never affect production)
+  * Filesystem writes only to _SHADOW_DIR (append-only JSONL, never read-back)
+
+Both public entry points — shadow_gate and shadow_render — return None always.
+"""
+from __future__ import annotations
+
+import functools
+import json
+import logging
+import os
+import time
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from meshai.notifications.events import Event
+
+logger = logging.getLogger("meshai.notifications.shadow")
+
+# JSONL output directory (inside the container data volume).
+_SHADOW_DIR = "/app/data/shadow"
+
+
+# ---------------------------------------------------------------------------
+# Env-var gate
+# ---------------------------------------------------------------------------
+
+@functools.lru_cache(maxsize=1)
+def _load_shadow_categories() -> frozenset:
+    """Parse MESHAI_SHADOW_CATEGORIES into a frozenset.  Cached after first call.
+
+    Call _clear_enabled_cache() (e.g. from tests) to force re-parse after
+    mutating the env var.
+    """
+    val = os.environ.get("MESHAI_SHADOW_CATEGORIES", "")
+    if not val.strip():
+        return frozenset()
+    return frozenset(c.strip() for c in val.split(",") if c.strip())
+
+
+def _clear_enabled_cache() -> None:
+    """Clear the lru_cache so tests can mutate MESHAI_SHADOW_CATEGORIES."""
+    _load_shadow_categories.cache_clear()
+
+
+def enabled_for(category: str) -> bool:
+    """Return True iff shadow comparison is enabled for *category*.
+
+    Fast path: returns False immediately when the env var is unset.
+    """
+    return category in _load_shadow_categories()
+
+
+# ---------------------------------------------------------------------------
+# Internal JSONL writer — never raises, never blocks the caller
+# ---------------------------------------------------------------------------
+
+def _append_jsonl(category: str, record: dict) -> None:
+    """Append one JSON line to /app/data/shadow/<category>.jsonl.
+
+    Any I/O failure is logged at DEBUG and silently discarded.
+    """
+    try:
+        os.makedirs(_SHADOW_DIR, exist_ok=True)
+        path = os.path.join(_SHADOW_DIR, f"{category}.jsonl")
+        line = json.dumps(record, default=str) + "\n"
+        with open(path, "a", encoding="utf-8") as fh:
+            fh.write(line)
+    except Exception:
+        logger.debug(
+            "shadow: JSONL write failed for category=%s", category, exc_info=True
+        )
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def shadow_gate(
+    category: str,
+    data: dict,
+    *,
+    source: str,
+    now: float,
+    old_broadcast: bool = False,
+) -> None:
+    """Dry-run the new gating decider for *category* and log mismatches.
+
+    Called from consumer._normalize, BEFORE the default-deny return.
+    The old_broadcast flag is synthesized is not None (True = would broadcast).
+    The old suffixes (_severity_override, _dedup_suffix, _cooldown_suffix) are
+    read from *data* — they are the values the legacy handler stamped.
+
+    Contract:
+      * NEVER calls GateResult.commit()
+      * NEVER emits any event
+      * NEVER writes any DB table
+      * All exceptions swallowed; returns None always
+    """
+    if not enabled_for(category):
+        return
+    try:
+        from meshai.notifications.gating import get_decider
+        decider = get_decider(category)
+        if decider is None:
+            # No decider registered yet (Phase 0 — all DECIDERS are empty).
+            # Nothing to compare; exit silently.
+            return
+
+        # ---- Phase 1+: a real decider is registered ----
+        # Extract old-path suffixes from the data dict the handler populated.
+        old_sev = (data or {}).get("_severity_override")
+        old_dedup = (data or {}).get("_dedup_suffix")
+        old_cool = (data or {}).get("_cooldown_suffix")
+
+        # Call the new decider.  data serves as a proxy for the full Event here
+        # (both carry _dedup_suffix, _cooldown_suffix, _severity_override).
+        # Phase 1 deciders that require a full Event object will need Hook 1
+        # to be relocated to post-normalize; Phase 0 never reaches here.
+        try:
+            new_result = decider(data, now=now)
+        except Exception:
+            logger.debug(
+                "shadow_gate: decider raised for category=%s", category, exc_info=True
+            )
+            return
+
+        # SAFETY CHECKPOINT: we inspect new_result but NEVER call .commit().
+        new_broadcast = new_result.broadcast if new_result is not None else False
+        new_patch = (new_result.data_patch or {}) if new_result is not None else {}
+
+        if old_broadcast != new_broadcast:
+            record = {
+                "tag": "SHADOW_MISMATCH",
+                "type": "gate",
+                "category": category,
+                "source": source,
+                "at": now,
+                "old_broadcast": old_broadcast,
+                "new_broadcast": new_broadcast,
+                "old_severity_override": old_sev,
+                "old_dedup_suffix": old_dedup,
+                "old_cooldown_suffix": old_cool,
+                "new_severity_override": new_patch.get("_severity_override"),
+                "new_dedup_suffix": new_patch.get("_dedup_suffix"),
+                "new_cooldown_suffix": new_patch.get("_cooldown_suffix"),
+                "lifecycle": getattr(new_result, "lifecycle", ""),
+                "reason": getattr(new_result, "reason", ""),
+            }
+            _append_jsonl(category, record)
+            logger.debug(
+                "shadow_gate MISMATCH cat=%s old=%s new=%s reason=%s",
+                category, old_broadcast, new_broadcast,
+                getattr(new_result, "reason", ""),
+            )
+    except Exception:
+        # Belt-and-suspenders: any unexpected error is silently absorbed.
+        logger.debug(
+            "shadow_gate: unhandled error for category=%s", category, exc_info=True
+        )
+
+
+def shadow_render(category: str, event: "Event", *, old_wire: str) -> None:
+    """Dry-run the new formatter for *category* and log wire-string mismatches.
+
+    Called from dispatcher._dispatch_toggle_path, after compose_mesh_message
+    has already produced *old_wire*.  The real path keeps using old_wire
+    unchanged; this function only observes.
+
+    Contract:
+      * NEVER modifies old_wire or event
+      * NEVER emits, commits, or writes any DB table
+      * All exceptions swallowed; returns None always
+    """
+    if not enabled_for(category):
+        return
+    try:
+        from meshai.notifications.formatters import get_formatter
+        formatter = get_formatter(category)
+        if formatter is None:
+            # No formatter registered yet (Phase 0 — all FORMATTERS are empty).
+            return
+
+        # ---- Phase 1+: a real formatter is registered ----
+        from meshai.notifications.formatters._budget import budget_for
+        budget = budget_for(getattr(event, "source", "") or "")
+        try:
+            new_wire = formatter(event, now=time.time(), budget=budget)
+        except Exception:
+            logger.debug(
+                "shadow_render: formatter raised for category=%s",
+                category, exc_info=True,
+            )
+            return
+
+        if new_wire != old_wire:
+            record = {
+                "tag": "SHADOW_MISMATCH",
+                "type": "render",
+                "category": category,
+                "event_id": getattr(event, "id", None),
+                "at": time.time(),
+                "old_wire": old_wire,
+                "new_wire": new_wire,
+            }
+            _append_jsonl(category, record)
+            logger.debug("shadow_render MISMATCH cat=%s", category)
+    except Exception:
+        logger.debug(
+            "shadow_render: unhandled error for category=%s", category, exc_info=True
+        )

work/requirements.txt

@@ -15,3 +15,4 @@ fastapi>=0.110.0
 uvicorn[standard]>=0.27.0
 aiomqtt>=2.0.0
 sgp4>=2.22
+tzdata

work/scripts/capture_fixtures.py

@@ -0,0 +1,244 @@
+"""Read-only ephemeral fixture capture from NATS JetStream.
+
+Captures real Central CloudEvents envelopes WITHOUT disturbing the live
+durable consumers by using an ephemeral pull consumer (no durable name,
+AckPolicy.none, short inactive_threshold for auto-deletion).
+
+Run from inside the meshai container::
+
+    docker exec meshai python /app/scripts/capture_fixtures.py \\
+        --hazard earthquake_event \\
+        --subject "central.usgs_quake.>" \\
+        --mode all --max 20
+
+    # Dry-run (count only, no file writes):
+    docker exec meshai python /app/scripts/capture_fixtures.py \\
+        --hazard earthquake_event \\
+        --subject "central.usgs_quake.>" \\
+        --mode all --max 20 --dry-run
+
+    # Last-per-subject snapshot:
+    docker exec meshai python /app/scripts/capture_fixtures.py \\
+        --hazard nws \\
+        --subject "central.nws.>" \\
+        --mode last
+
+Modes
+-----
+--mode last   DeliverPolicy.LAST_PER_SUBJECT — one message per subject key.
+              Useful for a current-state snapshot.
+--mode all    DeliverPolicy.ALL — bounded history.  REQUIRED: --max N cap
+              to avoid pulling 330k+ traffic messages.
+
+Output
+------
+Each captured envelope is written as::
+
+    tests/fixtures/<hazard>/<n>.json
+    {
+        "envelope":       { ... },     # raw Central CloudEvents payload
+        "subject":        "central.usgs_quake.us7000xyz",
+        "captured_epoch": 1750000000
+    }
+
+Safety
+------
+The ephemeral consumer is created with AckPolicy.none and a 30-second
+inactive_threshold.  It is never assigned a durable name, so it never
+advances the live durable consumers' sequence pointers and is automatically
+cleaned up by the NATS server after inactivity.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import os
+import pathlib
+import sys
+import time
+
+# --------------------------------------------------------------------------
+# All network + config access is deferred to main() so this module is safely
+# importable in unit-test environments without a running NATS server.
+# --------------------------------------------------------------------------
+
+
+def _output_dir(hazard: str) -> pathlib.Path:
+    """Resolve tests/fixtures/<hazard>/ relative to the repo root."""
+    # Script lives at <repo>/scripts/capture_fixtures.py;
+    # fixtures live at <repo>/tests/fixtures/<hazard>/.
+    repo_root = pathlib.Path(__file__).parent.parent
+    return repo_root / "tests" / "fixtures" / hazard
+
+
+async def _run(
+    *,
+    nats_url: str,
+    stream: str,
+    subject: str,
+    hazard: str,
+    mode: str,
+    max_msgs: int,
+    dry_run: bool,
+) -> int:
+    """Connect, create ephemeral consumer, pull messages, write fixtures.
+
+    Returns the count of messages captured (or counted, for --dry-run).
+    """
+    import nats
+    from nats.js.api import AckPolicy, ConsumerConfig, DeliverPolicy
+
+    nc = await nats.connect(nats_url)
+    try:
+        js = nc.jetstream()
+
+        # Build an ephemeral consumer config (no durable_name = ephemeral).
+        # AckPolicy.none avoids needing to ack — purely read-only.
+        # inactive_threshold of 30 s ensures the NATS server auto-deletes it.
+        deliver_policy = (
+            DeliverPolicy.LAST_PER_SUBJECT
+            if mode == "last"
+            else DeliverPolicy.ALL
+        )
+        cfg = ConsumerConfig(
+            # durable_name intentionally omitted → ephemeral consumer
+            filter_subject=subject,
+            deliver_policy=deliver_policy,
+            ack_policy=AckPolicy.NONE,
+            inactive_threshold=30.0,  # seconds → server auto-deletes after idle
+        )
+
+        # Create ephemeral pull consumer (server-side, no local binding name).
+        consumer_info = await js.add_consumer(stream, cfg)
+        consumer_name = consumer_info.name
+
+        out_dir = _output_dir(hazard)
+        if not dry_run:
+            out_dir.mkdir(parents=True, exist_ok=True)
+
+        captured = 0
+        fetch_batch = min(max_msgs, 50)  # pull in bounded batches
+
+        while captured < max_msgs:
+            batch = min(fetch_batch, max_msgs - captured)
+            try:
+                msgs = await js.pull_subscribe_bind(
+                    stream, consumer_name
+                ).fetch(batch, timeout=5.0)
+            except nats.errors.TimeoutError:
+                break  # no more messages within timeout
+
+            if not msgs:
+                break
+
+            for msg in msgs:
+                try:
+                    envelope = json.loads(msg.data)
+                except Exception:
+                    continue  # skip unparseable messages
+
+                if dry_run:
+                    captured += 1
+                    print(
+                        f"  [dry-run] #{captured} subject={msg.subject!r}",
+                        file=sys.stderr,
+                    )
+                else:
+                    record = {
+                        "envelope": envelope,
+                        "subject": msg.subject,
+                        "captured_epoch": int(time.time()),
+                    }
+                    out_path = out_dir / f"{captured:04d}.json"
+                    out_path.write_text(
+                        json.dumps(record, indent=2, ensure_ascii=False),
+                        encoding="utf-8",
+                    )
+                    captured += 1
+                    print(
+                        f"  wrote {out_path.relative_to(pathlib.Path.cwd())} "
+                        f"subject={msg.subject!r}",
+                        file=sys.stderr,
+                    )
+
+                if captured >= max_msgs:
+                    break
+
+            # For last-per-subject: a single fetch is sufficient.
+            if mode == "last":
+                break
+
+        # Delete the ephemeral consumer explicitly (belt-and-suspenders).
+        try:
+            await js.delete_consumer(stream, consumer_name)
+        except Exception:
+            pass  # server already cleaned up, or error is non-fatal
+
+        return captured
+
+    finally:
+        await nc.drain()
+        await nc.close()
+
+
+def _load_nats_url() -> str:
+    """Read the NATS URL from meshai config or env override."""
+    # Allow an explicit env override for CI / ad-hoc use.
+    if "MESHAI_NATS_URL" in os.environ:
+        return os.environ["MESHAI_NATS_URL"]
+    try:
+        from meshai.config_loader import load_config
+        cfg = load_config()
+        return cfg.environmental.central.url  # type: ignore[attr-defined]
+    except Exception:
+        return "nats://localhost:4222"
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(
+        description="Capture Central NATS envelopes as fixture files (read-only)."
+    )
+    parser.add_argument("--hazard", required=True,
+                        help="Hazard category label (used as fixture sub-dir).")
+    parser.add_argument("--subject", required=True,
+                        help="NATS subject filter, e.g. 'central.usgs_quake.>'.")
+    parser.add_argument("--stream", default="CENTRAL",
+                        help="JetStream stream name (default: CENTRAL).")
+    parser.add_argument("--mode", choices=["last", "all"], default="all",
+                        help="DeliverPolicy: last=LAST_PER_SUBJECT, all=ALL (default: all).")
+    parser.add_argument("--max", type=int, default=50, dest="max_msgs",
+                        help="Maximum messages to capture (required cap; default: 50).")
+    parser.add_argument("--nats-url", default=None,
+                        help="Override the NATS URL (default: read from meshai config).")
+    parser.add_argument("--dry-run", action="store_true",
+                        help="Count messages only; do not write fixture files.")
+    args = parser.parse_args(argv)
+
+    nats_url = args.nats_url or _load_nats_url()
+    print(
+        f"capture_fixtures: url={nats_url!r} stream={args.stream!r} "
+        f"subject={args.subject!r} hazard={args.hazard!r} "
+        f"mode={args.mode!r} max={args.max_msgs} dry_run={args.dry_run}",
+        file=sys.stderr,
+    )
+
+    count = asyncio.run(
+        _run(
+            nats_url=nats_url,
+            stream=args.stream,
+            subject=args.subject,
+            hazard=args.hazard,
+            mode=args.mode,
+            max_msgs=args.max_msgs,
+            dry_run=args.dry_run,
+        )
+    )
+
+    verb = "counted" if args.dry_run else "captured"
+    print(f"{verb} {count} envelope(s) for hazard={args.hazard!r}", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

work/tests/conftest.py

@@ -12,13 +12,47 @@ and clears the persistence-layer threading.local caches around each test.
 Existing tests that don't reference any fixture get isolation for free;
 tests that explicitly use a `db_path` (or similar) fixture can still
 override the env var inside their own fixture body -- last setenv wins.
+
+Phase-0 addition: a session-scoped TZ fixture validates that
+America/Boise resolves via zoneinfo, exercising the tzdata guard needed
+in CI environments.  The fixture saves and restores the original TZ so
+that existing tests running with naive datetimes are unaffected.
 """
+import os
+import time
+import zoneinfo
+
 import pytest
 
 from meshai.persistence import close_thread_connection
 from meshai.persistence import db as _persistence_db
 
 
+@pytest.fixture(scope="session", autouse=True)
+def _tz_boise():
+    """Set TZ=America/Boise, call tzset(), assert zoneinfo resolves, then restore.
+
+    This validates that the tzdata package (or system zoneinfo) is available
+    in CI before any formatter/renderer test runs.  The original TZ is restored
+    so that existing tests using naive datetimes are not affected by the
+    Mountain Time offset.
+    """
+    original_tz = os.environ.get("TZ")
+    os.environ["TZ"] = "America/Boise"
+    time.tzset()
+    # Assert that zoneinfo can resolve the timezone (requires tzdata package
+    # or /usr/share/zoneinfo/America/Boise on the system).
+    zoneinfo.ZoneInfo("America/Boise")
+    # Restore the original TZ so that tests using naive datetime / time.localtime()
+    # behave the same as they did before this session fixture was added.
+    if original_tz is None:
+        os.environ.pop("TZ", None)
+    else:
+        os.environ["TZ"] = original_tz
+    time.tzset()
+    yield
+
+
 @pytest.fixture(autouse=True)
 def _isolate_meshai_db(tmp_path, monkeypatch):
     """Point MESHAI_DB_PATH at a tmp file per test + run init_db so the

work/tests/harness/__init__.py

@@ -0,0 +1,32 @@
+"""Phase-0b test harness — golden + gate-sequence helpers.
+
+Import surface:
+
+    from tests.harness import (
+        pinned_time,
+        pinned_tz,
+        render_golden,
+        assert_byte_identical,
+        load_fixtures,
+        run_gate_sequence,
+    )
+
+All helpers are pure-Python, stdlib-only, and safe to import at collection time.
+"""
+from tests.harness.goldens import (
+    assert_byte_identical,
+    load_fixtures,
+    pinned_time,
+    pinned_tz,
+    render_golden,
+    run_gate_sequence,
+)
+
+__all__ = [
+    "assert_byte_identical",
+    "load_fixtures",
+    "pinned_time",
+    "pinned_tz",
+    "render_golden",
+    "run_gate_sequence",
+]

work/tests/harness/goldens.py

@@ -0,0 +1,284 @@
+"""Phase-0b golden + gate-sequence test helpers.
+
+pinned_time(epoch)
+    Context manager: monkeypatches meshai.notifications.clock.now / now_dt to
+    a fixed epoch so golden files are byte-stable.  SCOPED — does not touch
+    the process TZ.  Restores the originals on exit (exception-safe).
+
+pinned_tz(name)
+    Separate opt-in context manager: sets os.environ['TZ'] and calls
+    time.tzset(), restores on exit.  Only golden tests that format %Z need
+    this; enabling it globally would break naive-datetime tests.
+
+render_golden(handler_render_fn, envelope, *, at)
+    Run handler_render_fn(envelope) under pinned_time(at) and return the wire
+    string.  Used to CAPTURE the baseline golden (not to assert — compare
+    against assert_byte_identical later).
+
+assert_byte_identical(new, golden)
+    Compare two strings by their UTF-8 byte sequences (len + content).  On
+    mismatch emit a readable unified diff and raise AssertionError.
+
+load_fixtures(hazard)
+    Read every tests/fixtures/<hazard>/*.json file (sorted) and return a list
+    of dicts, each with keys {envelope, subject, captured_epoch}.
+
+run_gate_sequence(old_handler, new_decider, ordered_fixtures, *, timeline)
+    Stub comparator: exercise both the old and new gating path across an
+    explicit now timeline and return a diff report.
+
+    old_handler  : callable(fixture: dict, *, now: float) -> bool
+                   True = old path would broadcast, False = suppress.
+    new_decider  : callable(fixture: dict, *, now: float) -> GateResult | None
+                   May return None (no decision) → treated as suppress.
+    ordered_fixtures : list[dict]  — fixtures in replay order
+    timeline         : list[float] — one epoch per fixture (len must match)
+
+    Returns list[dict] with keys:
+        fixture_n   : int  — 0-based index into ordered_fixtures
+        now         : float
+        old_broadcast : bool
+        new_broadcast : bool
+        match         : bool  — True when both paths agree
+        diffs         : dict  — empty when match; keys describe divergence
+"""
+from __future__ import annotations
+
+import difflib
+import json
+import os
+import pathlib
+import time
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from typing import Callable, Optional
+
+
+# ---------------------------------------------------------------------------
+# pinned_time — deterministic clock seam for golden tests
+# ---------------------------------------------------------------------------
+
+@contextmanager
+def pinned_time(epoch: float):
+    """Monkeypatch meshai.notifications.clock.now / now_dt to *epoch*.
+
+    Restores the originals on exit even if the body raises.
+    Does NOT touch TZ — use pinned_tz for %Z-sensitive golden tests.
+
+    Usage::
+
+        with pinned_time(1_700_000_000.0):
+            result = some_handler(envelope)
+        # result is deterministic for golden comparison
+    """
+    import meshai.notifications.clock as _clock_mod
+    _orig_now = _clock_mod.now
+    _orig_now_dt = _clock_mod.now_dt
+    _clock_mod.now = lambda: float(epoch)
+    _clock_mod.now_dt = lambda tz=None: datetime.fromtimestamp(
+        epoch, tz=(tz if tz is not None else timezone.utc)
+    )
+    try:
+        yield epoch
+    finally:
+        _clock_mod.now = _orig_now
+        _clock_mod.now_dt = _orig_now_dt
+
+
+# ---------------------------------------------------------------------------
+# pinned_tz — opt-in TZ override (only for %Z-sensitive golden tests)
+# ---------------------------------------------------------------------------
+
+@contextmanager
+def pinned_tz(name: str = "America/Boise"):
+    """Set TZ=*name*, call time.tzset(), restore on exit.
+
+    Opt-in only: enabling TZ globally breaks existing tests that format
+    naive datetimes or rely on UTC.  Golden tests that include a %Z token
+    in their expected output should wrap only that call in pinned_tz.
+
+    Usage::
+
+        with pinned_tz("America/Boise"):
+            with pinned_time(1_700_000_000.0):
+                result = handler(envelope)
+    """
+    orig = os.environ.get("TZ")
+    os.environ["TZ"] = name
+    time.tzset()
+    try:
+        yield name
+    finally:
+        if orig is None:
+            os.environ.pop("TZ", None)
+        else:
+            os.environ["TZ"] = orig
+        time.tzset()
+
+
+# ---------------------------------------------------------------------------
+# render_golden — capture a wire string under frozen time
+# ---------------------------------------------------------------------------
+
+def render_golden(
+    handler_render_fn: Callable,
+    envelope: dict,
+    *,
+    at: float,
+) -> str:
+    """Return the wire string from handler_render_fn(envelope) with time frozen.
+
+    handler_render_fn : callable(envelope: dict) -> str
+        Typically a lambda wrapper around a handler function, e.g.::
+
+            render_golden(
+                lambda env: handle_nws(env, subject, data={}),
+                envelope,
+                at=1_700_000_000.0,
+            )
+
+    The result is used as the *golden* baseline.  Store it (e.g. in a .txt
+    fixture) and compare future renders with assert_byte_identical.
+    """
+    with pinned_time(at):
+        return handler_render_fn(envelope)
+
+
+# ---------------------------------------------------------------------------
+# assert_byte_identical — byte-level golden comparison
+# ---------------------------------------------------------------------------
+
+def assert_byte_identical(new: str, golden: str) -> None:
+    """Assert that *new* and *golden* are identical as UTF-8 byte sequences.
+
+    Checks both len(x.encode('utf-8')) and content.  On mismatch emits a
+    unified diff and raises AssertionError with a readable message.
+    """
+    new_bytes = new.encode("utf-8")
+    golden_bytes = golden.encode("utf-8")
+    if new_bytes == golden_bytes:
+        return
+    # Build a readable unified diff (character-level, single-line strings).
+    diff_lines = list(
+        difflib.unified_diff(
+            [golden],
+            [new],
+            fromfile="golden",
+            tofile="new",
+            lineterm="",
+        )
+    )
+    diff_str = "\n".join(diff_lines) if diff_lines else "<no text diff — byte sequences differ>"
+    raise AssertionError(
+        f"assert_byte_identical failed:\n"
+        f"  golden: {len(golden_bytes)} bytes\n"
+        f"  new:    {len(new_bytes)} bytes\n"
+        f"{diff_str}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# load_fixtures — read fixture JSON files for a hazard category
+# ---------------------------------------------------------------------------
+
+def load_fixtures(hazard: str) -> list[dict]:
+    """Read all tests/fixtures/<hazard>/*.json files (sorted) and return as list.
+
+    Each JSON file is expected to contain::
+
+        {
+            "envelope":       {...},   # raw Central CloudEvents envelope
+            "subject":        "...",   # NATS subject
+            "captured_epoch": 1234567  # wall-clock at capture time
+        }
+
+    Returns an empty list if the directory does not exist (no fixtures yet).
+    """
+    fixtures_dir = (
+        pathlib.Path(__file__).parent.parent / "fixtures" / hazard
+    )
+    if not fixtures_dir.is_dir():
+        return []
+    result = []
+    for p in sorted(fixtures_dir.glob("*.json")):
+        with open(p, encoding="utf-8") as fh:
+            result.append(json.load(fh))
+    return result
+
+
+# ---------------------------------------------------------------------------
+# run_gate_sequence — old-vs-new broadcast decision comparator
+# ---------------------------------------------------------------------------
+
+def run_gate_sequence(
+    old_handler: Callable,
+    new_decider: Callable,
+    ordered_fixtures: list,
+    *,
+    timeline: list,
+) -> list[dict]:
+    """Compare broadcast/suppress decisions of old_handler vs new_decider.
+
+    Parameters
+    ----------
+    old_handler:
+        callable(fixture: dict, *, now: float) -> bool
+        True  = old path would broadcast
+        False = old path would suppress
+    new_decider:
+        callable(fixture: dict, *, now: float) -> GateResult | None
+        None → treated as suppress (False).
+        The returned GateResult.commit is NEVER called here.
+    ordered_fixtures:
+        list of fixture dicts in replay order.
+    timeline:
+        list[float] — one epoch per fixture.  Must be same length as
+        ordered_fixtures; ValueError raised otherwise.
+
+    Returns
+    -------
+    list[dict] — one entry per fixture:
+        fixture_n     : int   — 0-based index
+        now           : float — epoch used for this step
+        old_broadcast : bool
+        new_broadcast : bool
+        match         : bool  — True when old_broadcast == new_broadcast
+        diffs         : dict  — empty when match; populated fields on divergence:
+            "broadcast": {"old": bool, "new": bool}
+    """
+    if len(ordered_fixtures) != len(timeline):
+        raise ValueError(
+            f"run_gate_sequence: ordered_fixtures has {len(ordered_fixtures)} entries "
+            f"but timeline has {len(timeline)} — they must match."
+        )
+
+    results = []
+    for i, (fixture, now_ts) in enumerate(zip(ordered_fixtures, timeline)):
+        # --- old path ---
+        old_broadcast = bool(old_handler(fixture, now=now_ts))
+
+        # --- new path (never commit) ---
+        new_result = None
+        try:
+            new_result = new_decider(fixture, now=now_ts)
+        except Exception:
+            pass  # treat a crashing decider as suppress
+        new_broadcast = (new_result.broadcast if new_result is not None else False)
+
+        # --- diff ---
+        diffs: dict = {}
+        if old_broadcast != new_broadcast:
+            diffs["broadcast"] = {"old": old_broadcast, "new": new_broadcast}
+
+        results.append(
+            {
+                "fixture_n": i,
+                "now": now_ts,
+                "old_broadcast": old_broadcast,
+                "new_broadcast": new_broadcast,
+                "match": not diffs,
+                "diffs": diffs,
+            }
+        )
+
+    return results

work/tests/test_budget_shim.py

@@ -0,0 +1,35 @@
+"""Phase-0: verify the budget.py re-export shim is identity-equal to the impl.
+
+Both import paths must resolve to the SAME function objects so any runtime
+patching (e.g. in existing tests) affects both paths simultaneously.
+"""
+
+import meshai.central.budget as _shim
+import meshai.notifications.formatters._budget as _impl
+
+
+def test_budget_for_is_same_object():
+    assert _shim.budget_for is _impl.budget_for, (
+        "meshai.central.budget.budget_for must be the same object as "
+        "meshai.notifications.formatters._budget.budget_for"
+    )
+
+
+def test_fit_to_budget_is_same_object():
+    assert _shim.fit_to_budget is _impl.fit_to_budget, (
+        "meshai.central.budget.fit_to_budget must be the same object as "
+        "meshai.notifications.formatters._budget.fit_to_budget"
+    )
+
+
+def test_shim_imports_work():
+    """Smoke: the public names are importable from the legacy path."""
+    from meshai.central.budget import budget_for, fit_to_budget  # noqa: F401
+    assert callable(budget_for)
+    assert callable(fit_to_budget)
+
+
+def test_budget_for_returns_default_without_adapter_config(monkeypatch):
+    """budget_for falls back to 140 when the adapter key is absent."""
+    val = _shim.budget_for("__no_such_adapter__")
+    assert val == 140

work/tests/test_capture_script_importable.py

@@ -0,0 +1,64 @@
+"""Phase-0b: verify capture_fixtures.py is importable without a NATS connection.
+
+This test guards against syntax errors, bad top-level imports, or any
+network access at import time.  It does NOT execute main() or connect to NATS.
+"""
+from __future__ import annotations
+
+import importlib.util
+import pathlib
+import sys
+import types
+
+
+_SCRIPT_PATH = (
+    pathlib.Path(__file__).parent.parent / "scripts" / "capture_fixtures.py"
+)
+
+
+class TestCaptureScriptImportable:
+    def test_script_file_exists(self):
+        assert _SCRIPT_PATH.is_file(), (
+            f"Expected capture_fixtures.py at {_SCRIPT_PATH} but file not found"
+        )
+
+    def test_script_importable_without_network(self):
+        """Load the script as a module; must succeed without NATS connection."""
+        spec = importlib.util.spec_from_file_location(
+            "capture_fixtures_script", _SCRIPT_PATH
+        )
+        module = importlib.util.module_from_spec(spec)
+        # Execute the module body — must not connect to anything.
+        spec.loader.exec_module(module)  # type: ignore[union-attr]
+
+    def test_main_function_is_callable(self):
+        """The script must expose a main() callable."""
+        spec = importlib.util.spec_from_file_location(
+            "capture_fixtures_script_main", _SCRIPT_PATH
+        )
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)  # type: ignore[union-attr]
+        assert callable(getattr(module, "main", None)), (
+            "capture_fixtures.py must expose a main() callable"
+        )
+
+    def test_output_dir_helper_is_callable(self):
+        """_output_dir is a helper for resolving fixture paths; must be importable."""
+        spec = importlib.util.spec_from_file_location(
+            "capture_fixtures_script_dir", _SCRIPT_PATH
+        )
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)  # type: ignore[union-attr]
+        assert callable(getattr(module, "_output_dir", None))
+
+    def test_help_flag_is_parseable(self):
+        """main(['--help']) must raise SystemExit(0), not a real error."""
+        spec = importlib.util.spec_from_file_location(
+            "capture_fixtures_script_help", _SCRIPT_PATH
+        )
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)  # type: ignore[union-attr]
+        import pytest
+        with pytest.raises(SystemExit) as exc_info:
+            module.main(["--help"])
+        assert exc_info.value.code == 0

work/tests/test_clock_seam.py

@@ -0,0 +1,59 @@
+"""Phase-0: verify the clock seam is monkeypatchable and handlers use it.
+
+Tests:
+  1. clock.now() returns a float close to time.time() at runtime.
+  2. Monkeypatching clock.now propagates into the three refactored handlers
+     (quake_handler._now, nws_handler._now, wfigs_handler._now).
+"""
+
+import time
+
+import pytest
+
+import meshai.notifications.clock as clock_mod
+import meshai.central.quake_handler as quake_handler
+import meshai.central.nws_handler as nws_handler
+import meshai.central.wfigs_handler as wfigs_handler
+
+
+_FROZEN_TS = 1_700_000_000.0
+
+
+def test_clock_now_returns_float_close_to_time():
+    before = time.time()
+    result = clock_mod.now()
+    after = time.time()
+    assert isinstance(result, float), "clock.now() must return float"
+    assert before <= result <= after + 0.1, "clock.now() must be current time"
+
+
+def test_clock_now_is_monkeypatchable(monkeypatch):
+    monkeypatch.setattr(clock_mod, "now", lambda: _FROZEN_TS)
+    assert clock_mod.now() == _FROZEN_TS
+
+
+def test_quake_handler_now_uses_clock_seam(monkeypatch):
+    """quake_handler._now() must reflect a monkeypatched clock.now."""
+    monkeypatch.setattr(clock_mod, "now", lambda: _FROZEN_TS)
+    result = quake_handler._now()
+    assert result == int(_FROZEN_TS), (
+        f"quake_handler._now() returned {result!r}, expected {int(_FROZEN_TS)}"
+    )
+
+
+def test_nws_handler_now_uses_clock_seam(monkeypatch):
+    """nws_handler._now() must reflect a monkeypatched clock.now."""
+    monkeypatch.setattr(clock_mod, "now", lambda: _FROZEN_TS)
+    result = nws_handler._now()
+    assert result == int(_FROZEN_TS), (
+        f"nws_handler._now() returned {result!r}, expected {int(_FROZEN_TS)}"
+    )
+
+
+def test_wfigs_handler_now_uses_clock_seam(monkeypatch):
+    """wfigs_handler._now() must reflect a monkeypatched clock.now."""
+    monkeypatch.setattr(clock_mod, "now", lambda: _FROZEN_TS)
+    result = wfigs_handler._now()
+    assert result == int(_FROZEN_TS), (
+        f"wfigs_handler._now() returned {result!r}, expected {int(_FROZEN_TS)}"
+    )

work/tests/test_formatter_scaffold.py

@@ -0,0 +1,66 @@
+"""Phase-0 scaffold tests: formatter registry empty + dispatch correctness.
+
+(a) With an empty registry, get_formatter returns None for real categories.
+(b) A registered dummy formatter is called verbatim — multi-line output is
+    returned as-is (no Mode-B single-line cap applied).
+"""
+
+import pytest
+
+from meshai.notifications.formatters import FORMATTERS, get_formatter, register
+from meshai.notifications.events import make_event
+from meshai.notifications.renderers.composer import compose_mesh_message
+
+
+# ── (a) empty registry returns None for known categories ──────────────────────
+
+@pytest.mark.parametrize("category", [
+    "weather_warning",
+    "earthquake_event",
+    "wildfire_incident",
+    "road_closure",
+    "battery_critical",
+])
+def test_get_formatter_returns_none_while_registry_empty(category):
+    """FORMATTERS is empty; get_formatter must return None for any real category."""
+    # Guarantee the registry is empty for these categories (it starts empty at
+    # module level; tests run in isolation from each other's registrations).
+    assert category not in FORMATTERS, (
+        f"Category {category!r} should not be in FORMATTERS at Phase 0"
+    )
+    assert get_formatter(category) is None
+
+
+# ── (b) registered dummy formatter is dispatched verbatim ────────────────────
+
+_SYNTHETIC_CATEGORY = "_test_scaffold_dummy_category_phase0"
+_MULTILINE_OUTPUT = "Line one\nLine two\nLine three"
+
+
+def _dummy_formatter(event, *, now: float, budget: int) -> str:
+    """Returns a fixed multi-line string to prove verbatim passthrough."""
+    return _MULTILINE_OUTPUT
+
+
+def test_registered_formatter_returns_verbatim_multiline(monkeypatch):
+    """Register a dummy for a synthetic category; compose_mesh_message must
+    return the dummy's multi-line output verbatim — newlines preserved — not
+    re-processed through Mode-B's single-line budget loop.
+    """
+    # Register the dummy (clean up afterwards to avoid cross-test pollution).
+    register(_SYNTHETIC_CATEGORY, _dummy_formatter)
+    try:
+        event = make_event(
+            source="test",
+            category=_SYNTHETIC_CATEGORY,
+            severity="routine",
+            title="First line\nSecond line",  # multi-line title
+        )
+        result = compose_mesh_message(event)
+        assert result == _MULTILINE_OUTPUT, (
+            f"Expected verbatim multi-line output, got: {result!r}"
+        )
+        # Verify newlines are preserved (Mode-B would strip them).
+        assert "\n" in result, "Newlines must survive the formatter dispatch path"
+    finally:
+        FORMATTERS.pop(_SYNTHETIC_CATEGORY, None)

work/tests/test_formatters_no_clock.py

@@ -0,0 +1,46 @@
+"""Phase-0 purity guard: formatter modules must not call time directly.
+
+Every file under meshai/notifications/formatters/ must route all time reads
+through meshai.notifications.clock (the determinism seam).  Direct calls to
+datetime.now or time.time( inside a formatter break the monkeypatch contract.
+
+This test is intentionally static (source-text check) so it catches new
+formatters that forget the contract even before any test calls them.
+"""
+
+import glob
+import os
+
+
+def _formatter_sources():
+    """Return (path, source) pairs for all .py files in the formatters package."""
+    pattern = os.path.join(
+        os.path.dirname(__file__),
+        "..", "meshai", "notifications", "formatters", "*.py",
+    )
+    paths = sorted(glob.glob(pattern))
+    return [(p, open(p).read()) for p in paths]
+
+
+def test_no_datetime_now_in_formatters():
+    """No formatter file may call datetime.now directly."""
+    violations = []
+    for path, src in _formatter_sources():
+        if "datetime.now" in src:
+            violations.append(os.path.basename(path))
+    assert not violations, (
+        "The following formatter files call datetime.now directly "
+        "(use clock.now_dt() instead):\n" + "\n".join(violations)
+    )
+
+
+def test_no_time_time_in_formatters():
+    """No formatter file may call time.time( directly."""
+    violations = []
+    for path, src in _formatter_sources():
+        if "time.time(" in src:
+            violations.append(os.path.basename(path))
+    assert not violations, (
+        "The following formatter files call time.time( directly "
+        "(use clock.now() instead):\n" + "\n".join(violations)
+    )

work/tests/test_harness_selftest.py

@@ -0,0 +1,298 @@
+"""Phase-0b: self-test the harness helpers.
+
+Verifies:
+  1. pinned_time freezes clock.now() to the given epoch.
+  2. pinned_time restores the original after the block.
+  3. pinned_tz sets and restores TZ (opt-in; tested in isolation).
+  4. assert_byte_identical passes on equal strings, fails on differing ones.
+  5. run_gate_sequence diffs correctly with two simple fake deciders.
+  6. load_fixtures returns empty list for a non-existent hazard.
+"""
+from __future__ import annotations
+
+import os
+import time
+
+import pytest
+
+import meshai.notifications.clock as clock_mod
+from tests.harness import (
+    assert_byte_identical,
+    load_fixtures,
+    pinned_time,
+    pinned_tz,
+    run_gate_sequence,
+)
+
+
+# ---------------------------------------------------------------------------
+# GateResult import for fake deciders
+# ---------------------------------------------------------------------------
+
+from meshai.notifications.gating.base import GateResult
+
+
+# ---------------------------------------------------------------------------
+# 1-2. pinned_time
+# ---------------------------------------------------------------------------
+
+_FROZEN = 1_700_000_000.0
+
+
+class TestPinnedTime:
+    def test_clock_now_is_frozen_inside_block(self):
+        with pinned_time(_FROZEN):
+            result = clock_mod.now()
+        assert result == _FROZEN
+
+    def test_clock_now_dt_is_frozen_inside_block(self):
+        from datetime import timezone
+        with pinned_time(_FROZEN):
+            dt = clock_mod.now_dt(tz=timezone.utc)
+        assert dt.timestamp() == pytest.approx(_FROZEN)
+
+    def test_clock_now_restored_after_block(self):
+        before = clock_mod.now
+        with pinned_time(_FROZEN):
+            pass
+        # The function object is restored to the original callable.
+        assert clock_mod.now is before
+
+    def test_clock_now_restored_on_exception(self):
+        original_now = clock_mod.now
+        try:
+            with pinned_time(_FROZEN):
+                raise RuntimeError("test exception")
+        except RuntimeError:
+            pass
+        # Must be restored even after exception.
+        assert clock_mod.now is original_now
+
+    def test_nested_pinned_time_restores_correctly(self):
+        outer_epoch = 1_600_000_000.0
+        inner_epoch = 1_700_000_000.0
+        with pinned_time(outer_epoch):
+            assert clock_mod.now() == outer_epoch
+            with pinned_time(inner_epoch):
+                assert clock_mod.now() == inner_epoch
+            # outer is restored
+            assert clock_mod.now() == outer_epoch
+
+    def test_pinned_time_yields_epoch(self):
+        with pinned_time(_FROZEN) as yielded:
+            assert yielded == _FROZEN
+
+
+# ---------------------------------------------------------------------------
+# 3. pinned_tz
+# ---------------------------------------------------------------------------
+
+class TestPinnedTz:
+    def test_sets_tz_inside_block(self):
+        with pinned_tz("UTC"):
+            assert os.environ.get("TZ") == "UTC"
+
+    def test_restores_tz_after_block(self):
+        orig = os.environ.get("TZ")
+        with pinned_tz("UTC"):
+            pass
+        restored = os.environ.get("TZ")
+        assert restored == orig  # both None, or both the original string
+
+    def test_restores_tz_on_exception(self):
+        orig = os.environ.get("TZ")
+        try:
+            with pinned_tz("UTC"):
+                raise RuntimeError("test")
+        except RuntimeError:
+            pass
+        assert os.environ.get("TZ") == orig
+
+    def test_yields_name(self):
+        with pinned_tz("America/Boise") as name:
+            assert name == "America/Boise"
+
+
+# ---------------------------------------------------------------------------
+# 4. assert_byte_identical
+# ---------------------------------------------------------------------------
+
+class TestAssertByteIdentical:
+    def test_passes_on_equal_strings(self):
+        assert_byte_identical("hello world", "hello world")
+
+    def test_passes_on_emoji_equal(self):
+        s = "🔥 FIRE: Test Creek NE 1500ac 35% routine"
+        assert_byte_identical(s, s)
+
+    def test_fails_on_differing_content(self):
+        with pytest.raises(AssertionError) as exc_info:
+            assert_byte_identical("hello world!", "hello world")
+        msg = str(exc_info.value)
+        assert "golden" in msg or "new" in msg  # diff headers present
+
+    def test_fails_on_byte_length_difference(self):
+        # Same visible chars but different byte lengths via Unicode.
+        a = "café"    # café (NFC, 2-byte é)
+        b = "café"   # cafe + combining accent (decomposed, also visually café)
+        # These have different UTF-8 byte lengths.
+        if a.encode("utf-8") != b.encode("utf-8"):
+            with pytest.raises(AssertionError):
+                assert_byte_identical(a, b)
+        else:
+            # If the runtime normalized them, both forms happen to be equal.
+            assert_byte_identical(a, b)
+
+    def test_diff_message_is_readable(self):
+        with pytest.raises(AssertionError) as exc_info:
+            assert_byte_identical("new string", "golden string")
+        msg = str(exc_info.value)
+        # Should mention byte counts.
+        assert "bytes" in msg
+
+    def test_passes_on_empty_strings(self):
+        assert_byte_identical("", "")
+
+
+# ---------------------------------------------------------------------------
+# 5. run_gate_sequence with fake deciders
+# ---------------------------------------------------------------------------
+
+class TestRunGateSequence:
+    """Use two simple fake deciders to verify the diff logic."""
+
+    # Fake old handler: broadcasts if the fixture has "broadcast": true.
+    @staticmethod
+    def _old_handler(fixture: dict, *, now: float) -> bool:
+        return bool(fixture.get("broadcast", True))
+
+    # Fake new decider: broadcasts if "new_broadcast" key is present and true.
+    @staticmethod
+    def _new_decider_agree(fixture: dict, *, now: float) -> GateResult:
+        # Mirrors the old handler.
+        broadcast = bool(fixture.get("broadcast", True))
+        return GateResult(broadcast=broadcast, lifecycle="test", reason="agree")
+
+    @staticmethod
+    def _new_decider_disagree(fixture: dict, *, now: float) -> GateResult:
+        # Always suppresses — will disagree when old handler broadcasts.
+        return GateResult(broadcast=False, lifecycle="test", reason="disagree")
+
+    def _make_fixtures(self, specs):
+        """Build minimal fixture dicts."""
+        return [{"broadcast": b} for b in specs]
+
+    def _make_timeline(self, n):
+        return [1_700_000_000.0 + i * 300 for i in range(n)]
+
+    def test_all_agree(self):
+        specs = [True, False, True]
+        fixtures = self._make_fixtures(specs)
+        timeline = self._make_timeline(len(fixtures))
+        results = run_gate_sequence(
+            self._old_handler,
+            self._new_decider_agree,
+            fixtures,
+            timeline=timeline,
+        )
+        assert len(results) == 3
+        for r in results:
+            assert r["match"] is True, f"Expected match at fixture {r['fixture_n']}: {r}"
+            assert r["diffs"] == {}
+
+    def test_disagree_is_flagged(self):
+        fixtures = self._make_fixtures([True, True, False])
+        timeline = self._make_timeline(len(fixtures))
+        results = run_gate_sequence(
+            self._old_handler,
+            self._new_decider_disagree,
+            fixtures,
+            timeline=timeline,
+        )
+        # fixture 0 and 1: old=True, new=False → mismatch
+        assert results[0]["match"] is False
+        assert results[0]["diffs"]["broadcast"] == {"old": True, "new": False}
+        # fixture 2: old=False, new=False → match
+        assert results[2]["match"] is True
+
+    def test_result_keys_present(self):
+        fixtures = self._make_fixtures([True])
+        results = run_gate_sequence(
+            self._old_handler,
+            self._new_decider_agree,
+            fixtures,
+            timeline=self._make_timeline(1),
+        )
+        r = results[0]
+        assert "fixture_n" in r
+        assert "now" in r
+        assert "old_broadcast" in r
+        assert "new_broadcast" in r
+        assert "match" in r
+        assert "diffs" in r
+
+    def test_length_mismatch_raises(self):
+        with pytest.raises(ValueError, match="timeline"):
+            run_gate_sequence(
+                self._old_handler,
+                self._new_decider_agree,
+                [{"broadcast": True}, {"broadcast": False}],
+                timeline=[1_700_000_000.0],  # length mismatch
+            )
+
+    def test_decider_exception_treated_as_suppress(self):
+        def _crashing_decider(fixture, *, now):
+            raise RuntimeError("boom")
+
+        fixtures = self._make_fixtures([True])
+        results = run_gate_sequence(
+            self._old_handler,
+            _crashing_decider,
+            fixtures,
+            timeline=self._make_timeline(1),
+        )
+        # old=True, new=False (crashing decider → suppress) → mismatch
+        assert results[0]["old_broadcast"] is True
+        assert results[0]["new_broadcast"] is False
+        assert results[0]["match"] is False
+
+    def test_now_is_passed_to_handlers(self):
+        received = []
+
+        def _handler_capturing_now(fixture, *, now):
+            received.append(("old", now))
+            return True
+
+        def _decider_capturing_now(fixture, *, now):
+            received.append(("new", now))
+            return GateResult(broadcast=True)
+
+        timeline = [1_700_000_000.0, 1_700_000_300.0]
+        run_gate_sequence(
+            _handler_capturing_now,
+            _decider_capturing_now,
+            [{"x": 1}, {"x": 2}],
+            timeline=timeline,
+        )
+        assert ("old", 1_700_000_000.0) in received
+        assert ("old", 1_700_000_300.0) in received
+        assert ("new", 1_700_000_000.0) in received
+
+    def test_empty_fixtures(self):
+        results = run_gate_sequence(
+            self._old_handler,
+            self._new_decider_agree,
+            [],
+            timeline=[],
+        )
+        assert results == []
+
+
+# ---------------------------------------------------------------------------
+# 6. load_fixtures for non-existent hazard
+# ---------------------------------------------------------------------------
+
+class TestLoadFixtures:
+    def test_missing_hazard_returns_empty(self):
+        result = load_fixtures("__nonexistent_hazard_xyzzy__")
+        assert result == []

work/tests/test_import_smoke.py

@@ -0,0 +1,38 @@
+"""Phase-0 import smoke test: every meshai/central/*_handler.py module
+must be importable without error after the budget shim refactor.
+
+This guards against a broken import chain (e.g. circular imports or a bad
+re-export in the shim) that would silently break all handlers.
+"""
+
+import glob
+import importlib
+import os
+
+
+def _handler_modules():
+    """Collect meshai.central.*_handler module names by globbing the source tree."""
+    pattern = os.path.join(
+        os.path.dirname(__file__),
+        "..", "meshai", "central", "*_handler.py",
+    )
+    paths = sorted(glob.glob(pattern))
+    assert paths, "No *_handler.py files found — check the glob path"
+    modules = []
+    for p in paths:
+        name = os.path.basename(p)[:-3]  # strip .py
+        modules.append(f"meshai.central.{name}")
+    return modules
+
+
+def test_all_central_handlers_importable():
+    """Each *_handler module must import cleanly (no ImportError / circular deps)."""
+    failed = []
+    for mod_name in _handler_modules():
+        try:
+            importlib.import_module(mod_name)
+        except ImportError as exc:
+            failed.append(f"{mod_name}: {exc}")
+    assert not failed, (
+        "The following handler modules raised ImportError:\n" + "\n".join(failed)
+    )

work/tests/test_shadow_inert.py

@@ -0,0 +1,199 @@
+"""Phase-0b: verify shadow hooks are inert by default and never mutate state.
+
+Test cases:
+  1. MESHAI_SHADOW_CATEGORIES unset → shadow_gate / shadow_render are pure no-ops
+     (no filesystem, no DB, no exception).
+  2. MESHAI_SHADOW_CATEGORIES set to a category with no decider registered →
+     still no-op (get_decider returns None, early return).
+  3. MESHAI_SHADOW_CATEGORIES set to a category with no formatter registered →
+     still no-op (get_formatter returns None, early return).
+  4. shadow_gate / shadow_render never raise regardless of input.
+"""
+from __future__ import annotations
+
+import os
+import pytest
+
+import meshai.notifications.shadow as shadow_mod
+
+
+def _reset_shadow_cache():
+    """Clear lru_cache so env-var changes take effect."""
+    shadow_mod._clear_enabled_cache()
+
+
+# ---------------------------------------------------------------------------
+# Helper: a minimal fake Event-like object for shadow_render
+# ---------------------------------------------------------------------------
+
+class _FakeEvent:
+    id = "test-event-001"
+    category = "earthquake_event"
+    source = "usgs_quake"
+    data: dict = {}
+
+
+# ---------------------------------------------------------------------------
+# Case 1: env var unset — must be completely off
+# ---------------------------------------------------------------------------
+
+class TestShadowInertWhenEnvUnset:
+    """With MESHAI_SHADOW_CATEGORIES unset, all shadow functions are no-ops."""
+
+    def setup_method(self):
+        os.environ.pop("MESHAI_SHADOW_CATEGORIES", None)
+        _reset_shadow_cache()
+
+    def test_enabled_for_returns_false(self):
+        assert shadow_mod.enabled_for("earthquake_event") is False
+        assert shadow_mod.enabled_for("nws") is False
+        assert shadow_mod.enabled_for("") is False
+
+    def test_shadow_gate_no_filesystem(self, tmp_path, monkeypatch):
+        """shadow_gate must not touch the filesystem when the env var is unset."""
+        monkeypatch.setattr(shadow_mod, "_SHADOW_DIR", str(tmp_path / "shadow"))
+        shadow_mod.shadow_gate(
+            "earthquake_event",
+            {"_severity_override": "immediate"},
+            source="usgs_quake",
+            now=1_700_000_000.0,
+            old_broadcast=True,
+        )
+        # No shadow dir should have been created.
+        assert not (tmp_path / "shadow").exists()
+
+    def test_shadow_render_no_filesystem(self, tmp_path, monkeypatch):
+        """shadow_render must not touch the filesystem when the env var is unset."""
+        monkeypatch.setattr(shadow_mod, "_SHADOW_DIR", str(tmp_path / "shadow"))
+        shadow_mod.shadow_render(
+            "earthquake_event",
+            _FakeEvent(),
+            old_wire="🌍 EQ M4.2: Test, ID 30mi NE routine",
+        )
+        assert not (tmp_path / "shadow").exists()
+
+    def test_shadow_gate_returns_none(self):
+        result = shadow_mod.shadow_gate(
+            "earthquake_event", {}, source="usgs", now=0.0, old_broadcast=False
+        )
+        assert result is None
+
+    def test_shadow_render_returns_none(self):
+        result = shadow_mod.shadow_render(
+            "earthquake_event", _FakeEvent(), old_wire="something"
+        )
+        assert result is None
+
+
+# ---------------------------------------------------------------------------
+# Case 2: env var set but no decider registered for that category
+# ---------------------------------------------------------------------------
+
+class TestShadowInertWhenNoDecider:
+    """MESHAI_SHADOW_CATEGORIES set but DECIDERS empty → still no-op."""
+
+    def setup_method(self):
+        os.environ["MESHAI_SHADOW_CATEGORIES"] = "earthquake_event"
+        _reset_shadow_cache()
+
+    def teardown_method(self):
+        os.environ.pop("MESHAI_SHADOW_CATEGORIES", None)
+        _reset_shadow_cache()
+
+    def test_enabled_for_returns_true(self):
+        assert shadow_mod.enabled_for("earthquake_event") is True
+
+    def test_shadow_gate_no_filesystem_when_no_decider(self, tmp_path, monkeypatch):
+        """get_decider returns None → shadow_gate exits before any file write."""
+        monkeypatch.setattr(shadow_mod, "_SHADOW_DIR", str(tmp_path / "shadow"))
+        # DECIDERS is empty (Phase 0 scaffold), so get_decider("earthquake_event")
+        # returns None and shadow_gate returns early without writing anything.
+        shadow_mod.shadow_gate(
+            "earthquake_event",
+            {"_dedup_suffix": "M4.2"},
+            source="usgs_quake",
+            now=1_700_000_000.0,
+            old_broadcast=True,
+        )
+        assert not (tmp_path / "shadow").exists()
+
+    def test_shadow_gate_does_not_raise(self):
+        """shadow_gate must not propagate any exception."""
+        try:
+            shadow_mod.shadow_gate(
+                "earthquake_event",
+                None,  # intentionally bad input — must not raise
+                source="usgs_quake",
+                now=1_700_000_000.0,
+                old_broadcast=False,
+            )
+        except Exception as exc:
+            pytest.fail(f"shadow_gate raised unexpectedly: {exc!r}")
+
+
+# ---------------------------------------------------------------------------
+# Case 3: env var set but no formatter registered for that category
+# ---------------------------------------------------------------------------
+
+class TestShadowRenderInertWhenNoFormatter:
+    """MESHAI_SHADOW_CATEGORIES set but FORMATTERS empty → still no-op."""
+
+    def setup_method(self):
+        os.environ["MESHAI_SHADOW_CATEGORIES"] = "earthquake_event"
+        _reset_shadow_cache()
+
+    def teardown_method(self):
+        os.environ.pop("MESHAI_SHADOW_CATEGORIES", None)
+        _reset_shadow_cache()
+
+    def test_shadow_render_no_filesystem_when_no_formatter(self, tmp_path, monkeypatch):
+        """get_formatter returns None → shadow_render exits before any file write."""
+        monkeypatch.setattr(shadow_mod, "_SHADOW_DIR", str(tmp_path / "shadow"))
+        shadow_mod.shadow_render(
+            "earthquake_event",
+            _FakeEvent(),
+            old_wire="old wire string",
+        )
+        assert not (tmp_path / "shadow").exists()
+
+    def test_shadow_render_does_not_raise(self):
+        """shadow_render must not propagate any exception."""
+        try:
+            shadow_mod.shadow_render(
+                "earthquake_event",
+                None,  # intentionally bad input — must not raise
+                old_wire="anything",
+            )
+        except Exception as exc:
+            pytest.fail(f"shadow_render raised unexpectedly: {exc!r}")
+
+
+# ---------------------------------------------------------------------------
+# Case 4: multiple categories, partial enable
+# ---------------------------------------------------------------------------
+
+class TestShadowPartialEnable:
+    """Only listed categories are enabled; others stay off."""
+
+    def setup_method(self):
+        os.environ["MESHAI_SHADOW_CATEGORIES"] = "nws,earthquake_event"
+        _reset_shadow_cache()
+
+    def teardown_method(self):
+        os.environ.pop("MESHAI_SHADOW_CATEGORIES", None)
+        _reset_shadow_cache()
+
+    def test_listed_category_is_enabled(self):
+        assert shadow_mod.enabled_for("nws") is True
+        assert shadow_mod.enabled_for("earthquake_event") is True
+
+    def test_unlisted_category_is_disabled(self):
+        assert shadow_mod.enabled_for("fire") is False
+        assert shadow_mod.enabled_for("geomagnetic_storm") is False
+
+    def test_shadow_gate_off_for_unlisted(self, tmp_path, monkeypatch):
+        monkeypatch.setattr(shadow_mod, "_SHADOW_DIR", str(tmp_path / "shadow"))
+        shadow_mod.shadow_gate(
+            "fire", {}, source="wfigs", now=0.0, old_broadcast=True
+        )
+        assert not (tmp_path / "shadow").exists()