feat(v0.6-phase3): reminder system + schema split + NWS dedup relaxation

Third broadcast type Active: clock-driven re-broadcasts of still-live
events at human-scale cadences. WFIGS fires 8h, itd_511 work zones daily
8 AM Mountain, SWPC G-storms 8h. NWS is NOT a clock reminder -- instead
the per-CAP-id dedup is relaxed to allow re-broadcast if >3h since last.
Schema split first_broadcast_at + last_broadcast_at on all reminder-
eligible tables. Wire prefix logic: New (first sight), Update (WFIGS
material change), Active (clock reminder). All cadences, channels, day-
of-week patterns, timezones, and termination conditions GUI-editable
from day one via the existing adapter_config editor. Termination:
tombstone OR containment_100 OR end_date_passed (no max-count). Quiet
hours not respected -- ripped out in Phase 2.

Schema (v11.sql):
  - ALTER TABLE fires|nws_alerts|traffic_events|quake_events|swpc_events|
    gauge_readings ADD COLUMN first_broadcast_at REAL
  - Backfill: UPDATE ... SET first_broadcast_at = last_broadcast_at
    WHERE last_broadcast_at IS NOT NULL
  - ALTER TABLE adapter_meta ADD COLUMN reminder_enabled INTEGER NOT NULL
    DEFAULT 0
  - UPDATE adapter_meta SET reminder_enabled=1 WHERE adapter IN
    ('wfigs', 'swpc') -- itd_511_work_zone is a new meta row seeded
    with reminder_enabled=1
  - SCHEMA_VERSION 10 -> 11

Handler commit-callbacks (wfigs/nws/quake/swpc/incident):
  - UPDATE ... SET last_broadcast_at=?, first_broadcast_at=COALESCE(
    first_broadcast_at, ?) -- first_broadcast_at stamped once, never
    overwritten

NWS handler (meshai/central/nws_handler.py):
  - _render() gains a prefix kwarg
  - After-first-broadcast branch: when (now - last_broadcast_at) >=
    adapter_config.nws.duplicate_allowed_after_seconds (default 10800
    = 3h), allow the re-broadcast with prefix=Active. Under the
    window, suppress as before. The commit callback continues to
    update last_broadcast_at.

ReminderScheduler (meshai/notifications/reminders/__init__.py):
  - Async loop, ticks every 60s
  - Each tick: SELECT adapter FROM adapter_meta WHERE reminder_enabled=1
  - Per adapter, load reminders_<adapter> config from adapter_config
    (cadence_kind, cadence_value, channels, terminate_when, dow_mask,
    timezone)
  - Interval cadence: rows where last_broadcast_at <= now - cadence_value
  - Clock cadence: localizes now to configured tz, finds slots that
    just passed in the last tick window, gated by dow_mask
  - Termination conditions checked per adapter:
      wfigs.containment_100      -> current_contained_pct >= 100
      wfigs.last_event_age_24h   -> last_event_at older than 24h
      swpc.end_date_passed       -> payload_json end_time in past
      itd_511_work_zone.end_date_passed -> traffic_events.end_at in past
  - Active: prefix on every emitted wire; dispatcher.dispatch_scheduled_
    broadcast() honors cold-start grace, bypasses toggle path
  - On success, last_broadcast_at = now; first_broadcast_at preserved

Launched from notifications/pipeline/__init__.py:start_pipeline()
alongside BandConditionsScheduler.

adapter_config registry (+15 new keys, 43 -> 58):
  - reminders_wfigs.cadence_kind/cadence_value/channels/terminate_when
  - reminders_swpc.cadence_kind/cadence_value/channels/terminate_when
  - reminders_itd_511_work_zone.cadence_kind/cadence_value/channels/
    dow_mask/timezone/terminate_when
  - nws.duplicate_allowed_after_seconds

adapter_meta (+4 rows, 15 -> 19):
  - reminders_wfigs, reminders_swpc, reminders_itd_511_work_zone
    (pseudo-adapters carrying the reminder config)
  - itd_511_work_zone (reminder target row; reminder_enabled=1)
  - reminder_enabled flag added to wfigs/swpc (existing rows updated by
    v11.sql) and to itd_511_work_zone seed.

Tests (tests/test_reminders.py, 10 cases):
  - wfigs reminder fires past 8h cadence, stamps last_broadcast_at,
    preserves first_broadcast_at
  - reminder skipped within cadence
  - reminder skipped when containment_100, last_event_age_24h
  - swpc reminder fires (interval)
  - work_zone clock reminder fires at 08:00 Mountain on enabled DOW
  - work_zone reminder skipped when end_date_passed
  - work_zone reminder skipped outside slot window
  - reminder_enabled=0 suppresses all reminders for that adapter

tests/test_nws_dedup_relaxation.py (5 cases):
  - First sighting renders without Active: prefix
  - Re-broadcast within 3h suppressed
  - Re-broadcast after 3h allowed with Active: prefix
  - adapter_config.nws.duplicate_allowed_after_seconds override takes
    effect (1h window verified)
  - First sighting stamps first_broadcast_at=committed_at,
    last_broadcast_at=committed_at; 4h later broadcast stamps
    last_broadcast_at only, first_broadcast_at preserved

Test count: 829 -> 844 (+15 new, 0 regressions). Foundation tests
updated for new counts (REGISTRY=58, ADAPTER_META=19, schema=v11).
This commit is contained in:
Matt Johnson (via Claude) 2026-06-05 21:11:32 +00:00
commit 3a410d5087
15 changed files with 940 additions and 24 deletions

View file

@ -29,14 +29,14 @@ def client():
# ============================================================================
def test_list_returns_all_43_keys(client):
def test_list_returns_all_58_keys(client):
r = client.get("/api/adapter-config")
assert r.status_code == 200
body = r.json()
# 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 == 43
assert total == 58
def test_list_grouped_by_adapter(client):

View file

@ -54,11 +54,11 @@ def test_v6_tables_exist(fresh_db):
assert "adapter_meta" in tables
def test_schema_meta_at_v10(fresh_db):
def test_schema_meta_at_v11(fresh_db):
v = fresh_db.execute(
"SELECT value FROM schema_meta WHERE key='version'"
).fetchone()["value"]
assert int(v) == 10
assert int(v) == 11
def test_adapter_config_type_check_constrains_vocabulary(fresh_db):
@ -73,16 +73,16 @@ def test_adapter_config_type_check_constrains_vocabulary(fresh_db):
# ---------- registry shape -----------------------------------------------
def test_registry_at_43_entries():
def test_registry_at_58_entries():
"""v0.6-3a.1 trim: 43 CONFIG-only keys (was 77 in v0.6-3a draft)."""
assert len(REGISTRY) == 43, (
assert len(REGISTRY) == 58, (
f"REGISTRY should have 43 entries after CONFIG-vs-CODE trim; got {len(REGISTRY)}. "
f"If a sentence template / emoji / heuristic snuck in, it belongs in CODE not config."
)
def test_adapter_meta_at_15(fresh_db):
assert len(ADAPTER_META) == 15
def test_adapter_meta_at_19(fresh_db):
assert len(ADAPTER_META) == 19
# ---------- seed ----------------------------------------------------------

View file

@ -0,0 +1,140 @@
"""v0.6-phase3 NWS dedup-window relaxation tests.
The same CAP id is now re-broadcast (with Active: prefix) when more than
`nws.duplicate_allowed_after_seconds` (default 10800 = 3h) have elapsed
since the last broadcast.
"""
from __future__ import annotations
import time
import pytest
from meshai.central.nws_handler import handle_nws
from meshai.persistence import get_db
def _env(*, cap_id="urn:oid:dedup.001", event="Severe Thunderstorm Warning",
severity="Severe", area="Ada County", county="Ada", state="ID",
expires="2026-06-05T03:00:00Z", lat=43.6, lon=-116.2,
category="wx.alert.severe_thunderstorm_warning"):
return {
"id": cap_id, "subject": "central.wx.alert.us.id",
"data": {
"id": cap_id, "adapter": "nws", "category": category,
"severity": 2,
"geo": {"centroid": [lon, lat], "primary_region": "US-ID"},
"data": {
"id": cap_id, "event": event, "severity": severity,
"areaDesc": area, "msgType": "Alert",
"headline": f"{event} for {area}",
"description": "X", "expires": expires,
"_enriched": {"geocoder": {"city": None,
"county": county, "state": state}},
},
},
}
def _commit(data, ts):
cb = data["_on_broadcast_committed"]
cb(float(ts))
def test_first_broadcast_no_active_prefix():
"""A first sighting renders without Active: prefix."""
env = _env()
data = {}
wire = handle_nws(env, env["subject"], data=data, now=1_000_000)
assert wire is not None
assert "Active:" not in wire
def test_repeat_within_3h_suppressed():
env = _env(cap_id="urn:oid:rep1")
data = {}
# First broadcast at t=0.
wire1 = handle_nws(env, env["subject"], data=data, now=0)
assert wire1 is not None
_commit(data, 0)
# Same CAP id again 2h later -- inside 3h window -> suppressed.
env2 = _env(cap_id="urn:oid:rep1")
data2 = {}
wire2 = handle_nws(env2, env2["subject"], data=data2, now=2 * 3600)
assert wire2 is None
def test_repeat_after_3h_allowed_with_active_prefix():
env = _env(cap_id="urn:oid:rep2")
data = {}
wire1 = handle_nws(env, env["subject"], data=data, now=0)
assert wire1 is not None
_commit(data, 0)
# Same CAP id 4h later -> allowed, Active: prefix.
env2 = _env(cap_id="urn:oid:rep2")
data2 = {}
wire2 = handle_nws(env2, env2["subject"], data=data2, now=4 * 3600)
assert wire2 is not None
assert "Active:" in wire2
def test_dedup_window_respects_config_override():
"""Changing nws.duplicate_allowed_after_seconds via adapter_config takes effect."""
from meshai.adapter_config import invalidate_cache
conn = get_db()
conn.execute(
"UPDATE adapter_config SET value_json='3600' "
"WHERE adapter='nws' AND key='duplicate_allowed_after_seconds'"
)
invalidate_cache()
env = _env(cap_id="urn:oid:tunable")
data = {}
handle_nws(env, env["subject"], data=data, now=0)
_commit(data, 0)
# 90 min later -> still suppressed (under new 1h window).
# Wait, 90 min > 60 min so it would BE allowed. Use 30 min instead.
env2 = _env(cap_id="urn:oid:tunable")
data2 = {}
wire = handle_nws(env2, env2["subject"], data=data2, now=30 * 60)
assert wire is None # 30 min < 60 min window
# 2h later -> allowed (over 1h window now).
env3 = _env(cap_id="urn:oid:tunable")
data3 = {}
wire3 = handle_nws(env3, env3["subject"], data=data3, now=2 * 3600)
assert wire3 is not None
assert "Active:" in wire3
def test_handler_stamps_first_broadcast_at():
"""The commit callback writes first_broadcast_at via COALESCE -- only on
the first commit, never overwriting it."""
env = _env(cap_id="urn:oid:stamp")
data = {}
handle_nws(env, env["subject"], data=data, now=0)
_commit(data, 100.0)
conn = get_db()
row = conn.execute(
"SELECT first_broadcast_at, last_broadcast_at FROM nws_alerts "
"WHERE event_id='urn:oid:stamp'"
).fetchone()
assert row["first_broadcast_at"] == 100.0
assert row["last_broadcast_at"] == 100.0
# Second broadcast 4h later -> last_broadcast_at updates, first_broadcast_at preserved.
env2 = _env(cap_id="urn:oid:stamp")
data2 = {}
wire2 = handle_nws(env2, env2["subject"], data=data2, now=4 * 3600)
assert wire2 is not None
_commit(data2, 4 * 3600.0)
row2 = conn.execute(
"SELECT first_broadcast_at, last_broadcast_at FROM nws_alerts "
"WHERE event_id='urn:oid:stamp'"
).fetchone()
assert row2["first_broadcast_at"] == 100.0 # unchanged
assert row2["last_broadcast_at"] == 4 * 3600.0

206
tests/test_reminders.py Normal file
View file

@ -0,0 +1,206 @@
"""v0.6-phase3 ReminderScheduler tests."""
from __future__ import annotations
import asyncio
import time
from unittest.mock import AsyncMock, MagicMock
import pytest
from meshai.notifications.reminders import ReminderScheduler
from meshai.persistence import get_db
# ---------- helpers --------------------------------------------------------
def _seed_fire(conn, *, irwin_id, last_broadcast_at, current_contained_pct=10,
last_event_at=None, name="Test Fire"):
if last_event_at is None: last_event_at = 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, first_broadcast_at, last_broadcast_at) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
(irwin_id, name, "WF", 500, current_contained_pct,
42.5, -114.5, "Cassia", "ID",
last_broadcast_at, last_event_at, last_broadcast_at, last_broadcast_at),
)
def _seed_swpc(conn, *, event_id, last_broadcast_at, event_type="swpc_kindex"):
conn.execute(
"INSERT OR REPLACE INTO swpc_events(event_id, event_type, severity_int, "
"payload_json, occurred_at, first_seen_at, first_broadcast_at, "
"last_broadcast_at) VALUES (?,?,?,?,?,?,?,?)",
(event_id, event_type, 7, "{}", last_broadcast_at, last_broadcast_at,
last_broadcast_at, last_broadcast_at),
)
def _seed_work_zone(conn, *, external_id, last_broadcast_at, end_at=None,
sub_type="road_works"):
conn.execute(
"INSERT OR REPLACE INTO traffic_events(source, external_id, road, "
"direction, county, state, sub_type, impact, first_seen_at, "
"last_seen_at, first_broadcast_at, last_broadcast_at, end_at) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
("itd_511", external_id, "I-84", "E", "Ada", "ID", sub_type,
None, last_broadcast_at, last_broadcast_at,
last_broadcast_at, last_broadcast_at, end_at),
)
@pytest.fixture
def mock_dispatcher():
d = MagicMock()
d.dispatch_scheduled_broadcast = AsyncMock(return_value=True)
return d
# ============================================================================
# Interval cadence (wfigs)
# ============================================================================
def test_wfigs_reminder_fires_past_cadence(mock_dispatcher):
"""A fire whose last_broadcast_at is past the 8h cadence emits Active:."""
now = 1_780_000_000
conn = get_db()
_seed_fire(conn, irwin_id="F1", last_broadcast_at=now - 9 * 3600)
sch = ReminderScheduler(mock_dispatcher, clock=lambda: now)
fired = asyncio.run(sch.tick_once())
assert fired == 1
mock_dispatcher.dispatch_scheduled_broadcast.assert_called_once()
args = mock_dispatcher.dispatch_scheduled_broadcast.call_args.kwargs
assert "Active" in args["text"]
assert args["source_event_table"] == "fires"
assert args["source_event_pk"] == "F1"
def test_wfigs_reminder_skipped_within_cadence(mock_dispatcher):
"""A fire broadcast 1h ago is within the 8h cadence -> no reminder."""
now = 1_780_000_000
conn = get_db()
_seed_fire(conn, irwin_id="F1", last_broadcast_at=now - 3600)
sch = ReminderScheduler(mock_dispatcher, clock=lambda: now)
fired = asyncio.run(sch.tick_once())
assert fired == 0
mock_dispatcher.dispatch_scheduled_broadcast.assert_not_called()
def test_wfigs_reminder_skipped_when_containment_100(mock_dispatcher):
"""containment_100 termination overrides the cadence."""
now = 1_780_000_000
conn = get_db()
_seed_fire(conn, irwin_id="F1", last_broadcast_at=now - 9 * 3600,
current_contained_pct=100)
sch = ReminderScheduler(mock_dispatcher, clock=lambda: now)
fired = asyncio.run(sch.tick_once())
assert fired == 0
mock_dispatcher.dispatch_scheduled_broadcast.assert_not_called()
def test_wfigs_reminder_skipped_when_last_event_age_24h(mock_dispatcher):
"""If last_event_at > 24h ago, treat the fire as gone -> no reminder."""
now = 1_780_000_000
conn = get_db()
_seed_fire(conn, irwin_id="F1", last_broadcast_at=now - 9 * 3600,
last_event_at=now - 48 * 3600)
sch = ReminderScheduler(mock_dispatcher, clock=lambda: now)
fired = asyncio.run(sch.tick_once())
assert fired == 0
def test_wfigs_reminder_stamps_last_broadcast_at(mock_dispatcher):
now = 1_780_000_000
conn = get_db()
_seed_fire(conn, irwin_id="F1", last_broadcast_at=now - 9 * 3600)
sch = ReminderScheduler(mock_dispatcher, clock=lambda: now)
asyncio.run(sch.tick_once())
row = conn.execute("SELECT last_broadcast_at, first_broadcast_at FROM fires WHERE irwin_id='F1'").fetchone()
assert row["last_broadcast_at"] == now
# first_broadcast_at preserved
assert row["first_broadcast_at"] == now - 9 * 3600
def test_reminder_disabled_adapter_does_nothing(mock_dispatcher):
"""When adapter_meta.reminder_enabled=0 for wfigs, no reminders fire."""
now = 1_780_000_000
conn = get_db()
_seed_fire(conn, irwin_id="F1", last_broadcast_at=now - 9 * 3600)
conn.execute("UPDATE adapter_meta SET reminder_enabled=0 WHERE adapter='wfigs'")
sch = ReminderScheduler(mock_dispatcher, clock=lambda: now)
fired = asyncio.run(sch.tick_once())
assert fired == 0
# ============================================================================
# SWPC interval cadence
# ============================================================================
def test_swpc_reminder_fires(mock_dispatcher):
now = 1_780_000_000
conn = get_db()
_seed_swpc(conn, event_id="S1", last_broadcast_at=now - 10 * 3600)
sch = ReminderScheduler(mock_dispatcher, clock=lambda: now)
fired = asyncio.run(sch.tick_once())
assert fired == 1
args = mock_dispatcher.dispatch_scheduled_broadcast.call_args.kwargs
assert "Active" in args["text"]
assert args["source_event_table"] == "swpc_events"
# ============================================================================
# itd_511_work_zone clock cadence
# ============================================================================
def test_work_zone_reminder_fires_at_clock_slot(mock_dispatcher):
"""At the 08:00 Mountain slot, an active work zone broadcasts Active:."""
# 08:00 America/Boise on a Monday. Compute UTC equivalent.
from datetime import datetime
from zoneinfo import ZoneInfo
# Pick a Monday at exactly 08:00 Mountain.
local = datetime(2026, 6, 8, 8, 0, 0, tzinfo=ZoneInfo("America/Boise"))
now = int(local.timestamp())
conn = get_db()
_seed_work_zone(conn, external_id="WZ1", last_broadcast_at=now - 48 * 3600)
sch = ReminderScheduler(mock_dispatcher, clock=lambda: now, tick_seconds=60)
fired = asyncio.run(sch.tick_once())
assert fired == 1
args = mock_dispatcher.dispatch_scheduled_broadcast.call_args.kwargs
assert "Active" in args["text"]
assert args["source_event_table"] == "traffic_events"
def test_work_zone_reminder_skipped_when_end_date_passed(mock_dispatcher):
from datetime import datetime
from zoneinfo import ZoneInfo
local = datetime(2026, 6, 8, 8, 0, 0, tzinfo=ZoneInfo("America/Boise"))
now = int(local.timestamp())
conn = get_db()
_seed_work_zone(conn, external_id="WZ1",
last_broadcast_at=now - 48 * 3600,
end_at=now - 3600) # end_at in the past
sch = ReminderScheduler(mock_dispatcher, clock=lambda: now, tick_seconds=60)
fired = asyncio.run(sch.tick_once())
assert fired == 0
def test_work_zone_reminder_skipped_outside_slot(mock_dispatcher):
"""At 14:00 Mountain (no slot), no reminder fires."""
from datetime import datetime
from zoneinfo import ZoneInfo
local = datetime(2026, 6, 8, 14, 0, 0, tzinfo=ZoneInfo("America/Boise"))
now = int(local.timestamp())
conn = get_db()
_seed_work_zone(conn, external_id="WZ1", last_broadcast_at=now - 48 * 3600)
sch = ReminderScheduler(mock_dispatcher, clock=lambda: now, tick_seconds=60)
fired = asyncio.run(sch.tick_once())
assert fired == 0