Merge feat/custom-announcements into main

Extends the existing ScheduledBroadcasts page with an Announcements
section (list + create/edit form) over the announcement_routes.py API
from c0572b04. No new page/route/nav entry.

- AnnouncementsPanel: list (plain-English schedule summary, channel
  summary, enabled toggle, last sent, delete-with-confirm) + a
  create/edit form (message textarea with a live char count against
  the 140-char budget, schedule-kind-specific controls, Preview button
  wired to POST /announcements/{id}/preview -- never sends).
- AnnouncementChannelPicker: polls GET /api/channels (Meshtastic) and
  GET /api/meshcore/channels/detail (MeshCore) and renders both as
  checkbox groups, mixing transports freely. Degrades per radio: shows
  a plain "unreachable" note and still renders any already-selected
  channel by its stored name so an existing announcement stays
  editable even when a radio is down. Blocks save on zero channels.
- New announcements always save disabled (enabled is never sent from
  the create/edit form); the list-row toggle is the only place this
  form family arms one, matching the backend contract where PUT is the
  sole place `enabled` can change.
- lib/api.ts: fetchMeshtasticChannels + the Announcement CRUD/preview
  client functions, following the existing fetchJson/detail-surfacing
  error conventions.

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

work/dashboard-frontend/src/components/AnnouncementChannelPicker.tsx

@@ -0,0 +1,163 @@
+// Checkbox channel picker for custom announcements. Polls both radios for
+// their CURRENT channels and lets the owner check any number of them,
+// mixing Meshtastic and MeshCore freely. This is deliberately NOT the
+// single-transport ChannelPicker.tsx (that one is Meshtastic-only and
+// stores plain channel indices) -- announcement channels are a mixed-
+// transport list of {transport, channel, name}.
+import { useState, useEffect, useCallback } from 'react'
+import { Check, AlertTriangle } from 'lucide-react'
+import {
+  fetchMeshtasticChannels, getMeshcoreChannelsDetail,
+  type MeshtasticChannel, type MeshcoreChannelsDetail, type AnnouncementChannelRef,
+} from '@/lib/api'
+
+interface Props {
+  value: AnnouncementChannelRef[]
+  onChange: (value: AnnouncementChannelRef[]) => void
+}
+
+export default function AnnouncementChannelPicker({ value, onChange }: Props) {
+  const [mtChannels, setMtChannels] = useState<MeshtasticChannel[] | null>(null)
+  const [mcDetail, setMcDetail] = useState<MeshcoreChannelsDetail | null>(null)
+  const [loading, setLoading] = useState(true)
+
+  const load = useCallback(() => {
+    setLoading(true)
+    Promise.allSettled([fetchMeshtasticChannels(), getMeshcoreChannelsDetail()]).then(
+      ([mt, mc]) => {
+        setMtChannels(mt.status === 'fulfilled' ? mt.value : [])
+        setMcDetail(mc.status === 'fulfilled' ? mc.value : { active: false, channels: [] })
+        setLoading(false)
+      }
+    )
+  }, [])
+
+  useEffect(() => { load() }, [load])
+
+  const isSelected = (transport: 'meshtastic' | 'meshcore', channel: number | string) =>
+    value.some((v) => v.transport === transport && v.channel === channel)
+
+  const toggle = (transport: 'meshtastic' | 'meshcore', channel: number | string, name: string) => {
+    if (isSelected(transport, channel)) {
+      onChange(value.filter((v) => !(v.transport === transport && v.channel === channel)))
+    } else {
+      onChange([...value, { transport, channel, name }])
+    }
+  }
+
+  if (loading) {
+    return <div className="text-sm text-slate-500">Polling radios for current channels...</div>
+  }
+
+  // Meshtastic: only channels the radio currently reports as enabled are
+  // offerable. Empty list (radio unreachable or nothing enabled) still
+  // needs to surface any channel this announcement already had selected,
+  // by its stored name, so an existing announcement stays editable.
+  const mtLive = (mtChannels ?? []).filter((c) => c.enabled)
+  const mtStaleSelections = value.filter(
+    (v) => v.transport === 'meshtastic' && !mtLive.some((c) => c.index === v.channel)
+  )
+
+  const mcActive = mcDetail?.active ?? false
+  const mcLive = mcDetail?.channels ?? []
+  const mcStaleSelections = value.filter(
+    (v) => v.transport === 'meshcore' && !mcLive.some((c) => c.name === v.channel)
+  )
+
+  return (
+    <div className="space-y-4">
+      {/* Meshtastic */}
+      <div className="space-y-1">
+        <label className="block text-xs text-slate-500 uppercase tracking-wide">Meshtastic channels</label>
+        {mtLive.length === 0 && (
+          <div className="flex items-center gap-2 text-xs text-amber-400 mb-1">
+            <AlertTriangle size={13} />
+            Meshtastic radio unreachable, or it isn&apos;t reporting any enabled channels right now.
+          </div>
+        )}
+        <div className="border border-[#1e2a3a] p-2 space-y-1">
+          {mtLive.map((ch) => (
+            <label
+              key={ch.index}
+              onClick={() => toggle('meshtastic', ch.index, ch.name || `Channel ${ch.index}`)}
+              className="flex items-center gap-2 p-2 rounded hover:bg-[#0a0e17] cursor-pointer"
+            >
+              <CheckBox checked={isSelected('meshtastic', ch.index)} />
+              <span className="text-sm text-slate-200">
+                {ch.index}: {ch.name || `Channel ${ch.index}`}
+                {ch.role === 'PRIMARY' && <span className="text-slate-500"> (Primary)</span>}
+              </span>
+            </label>
+          ))}
+          {mtStaleSelections.map((v) => (
+            <label
+              key={`stale-mt-${v.channel}`}
+              onClick={() => toggle('meshtastic', v.channel, v.name || `Channel ${v.channel}`)}
+              className="flex items-center gap-2 p-2 rounded hover:bg-[#0a0e17] cursor-pointer"
+            >
+              <CheckBox checked />
+              <span className="text-sm text-slate-400">
+                {v.name || `Channel ${v.channel}`} <span className="text-slate-600">(not currently reported by the radio)</span>
+              </span>
+            </label>
+          ))}
+          {mtLive.length === 0 && mtStaleSelections.length === 0 && (
+            <div className="text-sm text-slate-500 p-2">No Meshtastic channels available.</div>
+          )}
+        </div>
+      </div>
+
+      {/* MeshCore */}
+      <div className="space-y-1">
+        <label className="block text-xs text-slate-500 uppercase tracking-wide">MeshCore channels</label>
+        {!mcActive && (
+          <div className="flex items-center gap-2 text-xs text-amber-400 mb-1">
+            <AlertTriangle size={13} />
+            MeshCore companion unreachable right now.
+          </div>
+        )}
+        <div className="border border-[#1e2a3a] p-2 space-y-1">
+          {mcLive.map((ch) => (
+            <label
+              key={ch.name}
+              onClick={() => toggle('meshcore', ch.name, ch.name)}
+              className="flex items-center gap-2 p-2 rounded hover:bg-[#0a0e17] cursor-pointer"
+            >
+              <CheckBox checked={isSelected('meshcore', ch.name)} />
+              <span className="text-sm text-slate-200">{ch.name}</span>
+            </label>
+          ))}
+          {mcStaleSelections.map((v) => (
+            <label
+              key={`stale-mc-${v.channel}`}
+              onClick={() => toggle('meshcore', v.channel, v.name || String(v.channel))}
+              className="flex items-center gap-2 p-2 rounded hover:bg-[#0a0e17] cursor-pointer"
+            >
+              <CheckBox checked />
+              <span className="text-sm text-slate-400">
+                {v.name || v.channel} <span className="text-slate-600">(not currently reported by the companion)</span>
+              </span>
+            </label>
+          ))}
+          {mcLive.length === 0 && mcStaleSelections.length === 0 && (
+            <div className="text-sm text-slate-500 p-2">No MeshCore channels available.</div>
+          )}
+        </div>
+      </div>
+
+      {value.length === 0 && (
+        <p className="text-xs text-red-400">Select at least one channel -- an announcement with nowhere to send can&apos;t be saved.</p>
+      )}
+    </div>
+  )
+}
+
+function CheckBox({ checked }: { checked: boolean }) {
+  return (
+    <div className={`w-4 h-4 rounded border flex items-center justify-center flex-shrink-0 ${
+      checked ? 'bg-accent border-accent' : 'border-slate-600'
+    }`}>
+      {checked && <Check size={12} className="text-white" />}
+    </div>
+  )
+}

work/dashboard-frontend/src/components/AnnouncementsPanel.tsx

@@ -0,0 +1,463 @@
+// Custom scheduled announcements (owner-authored free-text broadcasts on a
+// clock-slot schedule). CRUD over /api/announcements -- see
+// work/meshai/dashboard/api/announcement_routes.py for the backend
+// contract this follows exactly (new rows always start disabled; PUT is
+// the only place `enabled` can change; preview never sends).
+import { useState, useEffect, useCallback } from 'react'
+import { Plus, Trash2, Check, X, Loader2, Eye, Megaphone } from 'lucide-react'
+import {
+  fetchAnnouncements, createAnnouncement, updateAnnouncement, deleteAnnouncement,
+  previewAnnouncement,
+  type Announcement, type AnnouncementDraft, type AnnouncementChannelRef, type AnnouncementPreview,
+} from '@/lib/api'
+import { TextInput, NumberInput, TimeInput } from '@/pages/Notifications'
+import { SelectInput } from '@/pages/Config'
+import AnnouncementChannelPicker from '@/components/AnnouncementChannelPicker'
+
+const DOW_LABELS = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun']
+const MESSAGE_BUDGET_HINT = 140 // typical mesh packet budget; Preview after saving confirms the exact figure
+
+function ordinal(n: number): string {
+  const rem100 = n % 100
+  if (rem100 >= 11 && rem100 <= 13) return `${n}th`
+  switch (n % 10) {
+    case 1: return `${n}st`
+    case 2: return `${n}nd`
+    case 3: return `${n}rd`
+    default: return `${n}th`
+  }
+}
+
+function describeSchedule(a: {
+  schedule_kind: string
+  time_of_day: string
+  interval_days: number | null
+  dow_mask: boolean[] | null
+  day_of_month: number | null
+}): string {
+  const t = a.time_of_day
+  switch (a.schedule_kind) {
+    case 'daily':
+      return `Every day at ${t}`
+    case 'interval_days': {
+      const n = a.interval_days ?? 1
+      return `Every ${n} day${n === 1 ? '' : 's'} at ${t}`
+    }
+    case 'weekly': {
+      const days = (a.dow_mask ?? []).map((on, i) => (on ? DOW_LABELS[i] : null)).filter(Boolean)
+      return days.length ? `${days.join('/')} at ${t}` : `No days selected, at ${t}`
+    }
+    case 'monthly':
+      return `The ${ordinal(a.day_of_month ?? 1)} of each month at ${t}`
+    default:
+      return t
+  }
+}
+
+function channelSummary(channels: AnnouncementChannelRef[]): string {
+  if (channels.length === 0) return 'No channels'
+  const names = channels.map((c) => c.name || String(c.channel))
+  return `${channels.length} channel${channels.length === 1 ? '' : 's'}: ${names.join(', ')}`
+}
+
+function formatLastSent(v: number | null): string {
+  if (!v) return 'Never'
+  return new Date(v * 1000).toLocaleString()
+}
+
+const EMPTY_DRAFT: AnnouncementDraft = {
+  name: '',
+  message: '',
+  schedule_kind: 'daily',
+  time_of_day: '08:00',
+  interval_days: 1,
+  dow_mask: [false, false, false, false, false, false, false],
+  day_of_month: 1,
+  timezone: 'America/Boise',
+  channels: [],
+}
+
+export default function AnnouncementsPanel() {
+  const [rows, setRows] = useState<Announcement[]>([])
+  const [loading, setLoading] = useState(true)
+  const [loadError, setLoadError] = useState<string | null>(null)
+
+  const [formOpen, setFormOpen] = useState(false)
+  const [editingId, setEditingId] = useState<number | null>(null)
+  const [draft, setDraft] = useState<AnnouncementDraft>(EMPTY_DRAFT)
+  const [savedMessage, setSavedMessage] = useState<string>('') // the last-persisted message, for the preview staleness note
+  const [saving, setSaving] = useState(false)
+  const [saveError, setSaveError] = useState<string | null>(null)
+  const [notice, setNotice] = useState<string | null>(null)
+
+  const [preview, setPreview] = useState<AnnouncementPreview | null>(null)
+  const [previewLoading, setPreviewLoading] = useState(false)
+  const [previewError, setPreviewError] = useState<string | null>(null)
+
+  const [busyRowId, setBusyRowId] = useState<number | null>(null)
+
+  const refresh = useCallback(async () => {
+    setLoading(true)
+    setLoadError(null)
+    try {
+      setRows(await fetchAnnouncements())
+    } catch (e) {
+      setLoadError(e instanceof Error ? e.message : String(e))
+    } finally {
+      setLoading(false)
+    }
+  }, [])
+
+  useEffect(() => { refresh() }, [refresh])
+
+  const openNew = () => {
+    setEditingId(null)
+    setDraft({ ...EMPTY_DRAFT, dow_mask: [...EMPTY_DRAFT.dow_mask!] })
+    setSavedMessage('')
+    setPreview(null)
+    setPreviewError(null)
+    setSaveError(null)
+    setFormOpen(true)
+  }
+
+  const openEdit = (a: Announcement) => {
+    setEditingId(a.announcement_id)
+    setDraft({
+      name: a.name,
+      message: a.message,
+      schedule_kind: a.schedule_kind,
+      time_of_day: a.time_of_day,
+      interval_days: a.interval_days ?? 1,
+      dow_mask: a.dow_mask ?? [false, false, false, false, false, false, false],
+      day_of_month: a.day_of_month ?? 1,
+      timezone: a.timezone,
+      channels: a.channels,
+    })
+    setSavedMessage(a.message)
+    setPreview(null)
+    setPreviewError(null)
+    setSaveError(null)
+    setFormOpen(true)
+  }
+
+  const closeForm = () => {
+    setFormOpen(false)
+    setEditingId(null)
+    setPreview(null)
+    setPreviewError(null)
+    setSaveError(null)
+  }
+
+  const save = async () => {
+    setSaveError(null)
+    if (!draft.name.trim()) { setSaveError('Name is required.'); return }
+    if (!draft.message.trim()) { setSaveError('Message is required.'); return }
+    if (draft.channels.length === 0) { setSaveError('Select at least one channel.'); return }
+    if (draft.schedule_kind === 'interval_days' && (!draft.interval_days || draft.interval_days < 1)) {
+      setSaveError('Interval must be at least 1 day.'); return
+    }
+    if (draft.schedule_kind === 'weekly' && !(draft.dow_mask ?? []).some(Boolean)) {
+      setSaveError('Select at least one day of the week.'); return
+    }
+    if (draft.schedule_kind === 'monthly' && (!draft.day_of_month || draft.day_of_month < 1 || draft.day_of_month > 31)) {
+      setSaveError('Day of month must be between 1 and 31.'); return
+    }
+
+    setSaving(true)
+    try {
+      if (editingId === null) {
+        await createAnnouncement(draft)
+        setNotice('Saved. Enable it when you’re ready — new announcements always start disabled.')
+      } else {
+        // Never send `enabled` from this form -- the list view's toggle is
+        // the only thing allowed to arm/disarm an announcement.
+        await updateAnnouncement(editingId, draft)
+        setNotice('Saved.')
+      }
+      setFormOpen(false)
+      setEditingId(null)
+      await refresh()
+      setTimeout(() => setNotice(null), 5000)
+    } catch (e) {
+      setSaveError(e instanceof Error ? e.message : String(e))
+    } finally {
+      setSaving(false)
+    }
+  }
+
+  const runPreview = async () => {
+    if (editingId === null) return
+    setPreviewLoading(true)
+    setPreviewError(null)
+    try {
+      setPreview(await previewAnnouncement(editingId))
+    } catch (e) {
+      setPreviewError(e instanceof Error ? e.message : String(e))
+    } finally {
+      setPreviewLoading(false)
+    }
+  }
+
+  const toggleEnabled = async (a: Announcement) => {
+    setBusyRowId(a.announcement_id)
+    try {
+      await updateAnnouncement(a.announcement_id, { enabled: !a.enabled })
+      await refresh()
+    } catch (e) {
+      alert(`Could not change enabled state: ${e instanceof Error ? e.message : String(e)}`)
+    } finally {
+      setBusyRowId(null)
+    }
+  }
+
+  const remove = async (a: Announcement) => {
+    if (!confirm(`Delete announcement "${a.name}"? This cannot be undone.`)) return
+    setBusyRowId(a.announcement_id)
+    try {
+      await deleteAnnouncement(a.announcement_id)
+      await refresh()
+    } catch (e) {
+      alert(`Delete failed: ${e instanceof Error ? e.message : String(e)}`)
+      setBusyRowId(null)
+    }
+  }
+
+  const messageLen = draft.message.length
+  const overBudget = messageLen > MESSAGE_BUDGET_HINT
+  const messageDirty = editingId !== null && draft.message !== savedMessage
+
+  return (
+    <div className="bg-bg-card border border-border p-6 space-y-4">
+      <div className="flex items-center gap-2">
+        <Megaphone size={16} className="text-accent" />
+        <label className="text-xs text-slate-500 uppercase tracking-wide">Announcements</label>
+        <span className="text-xs text-slate-600">{rows.length} configured</span>
+        {!formOpen && (
+          <button
+            onClick={openNew}
+            className="ml-auto flex items-center gap-1 px-3 py-1.5 bg-accent hover:bg-accent/80 text-white text-sm transition-colors"
+          >
+            <Plus size={14} /> New announcement
+          </button>
+        )}
+      </div>
+      <p className="text-xs text-slate-600">
+        Free-text broadcasts you write and schedule yourself, separate from the automatic hazard/weather
+        notifications above. New announcements always save disabled &mdash; nothing transmits until you
+        switch one on below.
+      </p>
+
+      {notice && (
+        <div className="p-3 text-sm bg-green-500/10 text-green-400 border border-green-500/20">
+          <Check size={14} className="inline mr-2" />{notice}
+        </div>
+      )}
+
+      {/* ---- Create / edit form ---- */}
+      {formOpen && (
+        <div className="border border-[#2a3a4a] p-4 space-y-4 bg-[#0d1420]">
+          <div className="flex items-center justify-between">
+            <h3 className="text-sm font-medium text-slate-200">{editingId === null ? 'New announcement' : 'Edit announcement'}</h3>
+            <button onClick={closeForm} className="text-slate-500 hover:text-slate-300"><X size={16} /></button>
+          </div>
+
+          {saveError && (
+            <div className="p-2 text-sm bg-red-500/10 text-red-400 border border-red-500/20">{saveError}</div>
+          )}
+
+          <TextInput
+            label="Name"
+            value={draft.name}
+            onChange={(v) => setDraft({ ...draft, name: v })}
+            placeholder="Short label, e.g. Range closure notice"
+            helper="For your own reference in the list below -- not sent on the air."
+          />
+
+          <div className="space-y-1">
+            <label className="block text-xs text-slate-500 uppercase tracking-wide">Message</label>
+            <textarea
+              value={draft.message}
+              onChange={(e) => setDraft({ ...draft, message: e.target.value })}
+              rows={4}
+              placeholder="The exact text to broadcast"
+              className="w-full px-3 py-2 bg-[#0a0e17] border border-[#1e2a3a] rounded text-sm text-slate-200 font-mono focus:outline-none focus:border-accent placeholder-slate-600"
+            />
+            <p className={`text-xs ${overBudget ? 'text-amber-400' : 'text-slate-600'}`}>
+              {messageLen} / {MESSAGE_BUDGET_HINT} characters
+              {overBudget && ' — this will be truncated to fit the packet budget. Save, then use Preview below to see exactly what gets sent.'}
+            </p>
+          </div>
+
+          <SelectInput
+            label="Schedule"
+            value={draft.schedule_kind}
+            onChange={(v) => setDraft({ ...draft, schedule_kind: v as AnnouncementDraft['schedule_kind'] })}
+            options={[
+              { value: 'daily', label: 'Every day' },
+              { value: 'interval_days', label: 'Every N days' },
+              { value: 'weekly', label: 'Specific days of the week' },
+              { value: 'monthly', label: 'Specific day of the month' },
+            ]}
+          />
+
+          {draft.schedule_kind === 'interval_days' && (
+            <NumberInput
+              label="Interval (days)"
+              value={draft.interval_days ?? 1}
+              onChange={(v) => setDraft({ ...draft, interval_days: v })}
+              min={1}
+              helper="Sends every N days, counted from the first day it fires."
+            />
+          )}
+
+          {draft.schedule_kind === 'weekly' && (
+            <div className="space-y-1">
+              <label className="block text-xs text-slate-500 uppercase tracking-wide">Days of the week</label>
+              <div className="flex flex-wrap gap-2">
+                {DOW_LABELS.map((label, i) => {
+                  const checked = (draft.dow_mask ?? [])[i] ?? false
+                  return (
+                    <label
+                      key={label}
+                      onClick={() => {
+                        const mask = [...(draft.dow_mask ?? [false, false, false, false, false, false, false])]
+                        mask[i] = !mask[i]
+                        setDraft({ ...draft, dow_mask: mask })
+                      }}
+                      className={`px-3 py-1.5 text-sm cursor-pointer border ${
+                        checked ? 'bg-accent border-accent text-white' : 'border-[#1e2a3a] text-slate-300 hover:bg-[#0a0e17]'
+                      }`}
+                    >
+                      {label}
+                    </label>
+                  )
+                })}
+              </div>
+            </div>
+          )}
+
+          {draft.schedule_kind === 'monthly' && (
+            <NumberInput
+              label="Day of month"
+              value={draft.day_of_month ?? 1}
+              onChange={(v) => setDraft({ ...draft, day_of_month: v })}
+              min={1}
+              max={31}
+              helper="If the month is shorter than this day (e.g. day 31 in April), it fires on that month's last day instead."
+            />
+          )}
+
+          <div className="grid grid-cols-2 gap-3">
+            <TimeInput
+              label="Time of day"
+              value={draft.time_of_day}
+              onChange={(v) => setDraft({ ...draft, time_of_day: v })}
+            />
+            <TextInput
+              label="Timezone"
+              value={draft.timezone}
+              onChange={(v) => setDraft({ ...draft, timezone: v })}
+              helper="IANA timezone name, e.g. America/Boise."
+            />
+          </div>
+
+          <AnnouncementChannelPicker
+            value={draft.channels}
+            onChange={(v) => setDraft({ ...draft, channels: v })}
+          />
+
+          {/* Preview -- requires a saved row, since the backend reads the
+              stored message by id. Never sends. */}
+          <div className="space-y-2 pt-2 border-t border-[#1e2a3a]">
+            <div className="flex items-center gap-2">
+              <button
+                onClick={runPreview}
+                disabled={editingId === null || previewLoading}
+                className="flex items-center gap-1 px-3 py-1.5 border border-[#1e2a3a] text-slate-300 hover:bg-[#0a0e17] disabled:opacity-40 disabled:cursor-not-allowed text-sm transition-colors"
+              >
+                <Eye size={14} /> {previewLoading ? 'Loading preview...' : 'Preview'}
+              </button>
+              <span className="text-xs text-slate-600">
+                {editingId === null
+                  ? 'Save the announcement first, then Preview shows the exact wire text. This never sends anything.'
+                  : 'Shows the exact wire text and size. This never sends anything.'}
+              </span>
+            </div>
+            {messageDirty && editingId !== null && (
+              <p className="text-xs text-amber-400">You&apos;ve edited the message since it was last saved -- Preview still shows the saved version until you save again.</p>
+            )}
+            {previewError && <div className="p-2 text-sm bg-red-500/10 text-red-400 border border-red-500/20">{previewError}</div>}
+            {preview && (
+              <div className="p-3 bg-[#0a0e17] border border-[#1e2a3a] space-y-1">
+                <p className="text-sm text-slate-200 font-mono whitespace-pre-wrap break-words">{preview.wire_text}</p>
+                <p className={`text-xs ${preview.truncated ? 'text-amber-400' : 'text-slate-500'}`}>
+                  {preview.char_count} characters / {preview.byte_count} bytes (budget {preview.budget})
+                  {preview.truncated ? ' — truncated to fit the budget' : ' — fits, not truncated'}
+                </p>
+              </div>
+            )}
+          </div>
+
+          <div className="flex items-center justify-end gap-2 pt-2">
+            <button onClick={closeForm} className="px-3 py-1.5 text-slate-400 hover:text-slate-200 text-sm">Cancel</button>
+            <button
+              onClick={save}
+              disabled={saving}
+              className="flex items-center gap-1 px-4 py-1.5 bg-accent hover:bg-accent/80 disabled:bg-slate-700 disabled:cursor-not-allowed text-white text-sm transition-colors"
+            >
+              {saving ? 'Saving...' : 'Save'}
+            </button>
+          </div>
+        </div>
+      )}
+
+      {/* ---- List ---- */}
+      {loading ? (
+        <div className="text-sm text-slate-500 flex items-center gap-2"><Loader2 size={14} className="animate-spin" /> Loading announcements...</div>
+      ) : loadError ? (
+        <div className="text-sm text-red-400">Failed to load announcements: {loadError}</div>
+      ) : rows.length === 0 ? (
+        <div className="text-sm text-slate-500">No announcements yet.</div>
+      ) : (
+        <div className="border border-border divide-y divide-border">
+          {rows.map((a) => (
+            <div key={a.announcement_id} className="p-3 flex items-start gap-3">
+              <button
+                onClick={() => toggleEnabled(a)}
+                disabled={busyRowId === a.announcement_id}
+                title={a.enabled ? 'Enabled -- click to disable' : 'Disabled -- click to enable'}
+                className={`relative w-11 h-6 rounded-full transition-colors flex-shrink-0 mt-0.5 disabled:opacity-50 ${
+                  a.enabled ? 'bg-accent' : 'bg-[#1e2a3a]'
+                }`}
+              >
+                <span className={`absolute top-1 left-1 w-4 h-4 rounded-full bg-white transition-transform ${a.enabled ? 'translate-x-5' : ''}`} />
+              </button>
+              <div className="flex-1 min-w-0 space-y-0.5">
+                <div className="flex items-center gap-2">
+                  <span className="text-sm text-slate-100 font-medium">{a.name}</span>
+                  <span className={`text-xs ${a.enabled ? 'text-green-400' : 'text-slate-500'}`}>
+                    {a.enabled ? 'Enabled' : 'Disabled'}
+                  </span>
+                </div>
+                <p className="text-xs text-slate-400">{describeSchedule(a)}</p>
+                <p className="text-xs text-slate-500 truncate">{channelSummary(a.channels)}</p>
+                <p className="text-xs text-slate-600">Last sent: {formatLastSent(a.last_sent_at)}</p>
+              </div>
+              <div className="flex items-center gap-3 flex-shrink-0">
+                <button onClick={() => openEdit(a)} className="text-accent hover:text-accent text-xs">Edit</button>
+                <button
+                  onClick={() => remove(a)}
+                  disabled={busyRowId === a.announcement_id}
+                  className="text-red-400 hover:text-red-300 disabled:opacity-50"
+                  title="Delete"
+                >
+                  <Trash2 size={14} />
+                </button>
+              </div>
+            </div>
+          ))}
+        </div>
+      )}
+    </div>
+  )
+}

work/dashboard-frontend/src/lib/api.ts

@@ -925,3 +925,120 @@ export async function sendTestMessage(body: {
   }
   return response.json()
 }
+
+// ---- Meshtastic channel list (radio-polled, for channel pickers) ------
+
+export interface MeshtasticChannel {
+  index: number
+  name: string
+  role: 'DISABLED' | 'PRIMARY' | 'SECONDARY' | 'UNKNOWN' | string
+  enabled: boolean
+}
+
+// Returns [] when the radio is unreachable or reports nothing -- there is
+// no separate "active" flag on this endpoint (unlike the MeshCore one).
+export async function fetchMeshtasticChannels(): Promise<MeshtasticChannel[]> {
+  return fetchJson<MeshtasticChannel[]>('/api/channels')
+}
+
+// ---- Custom scheduled announcements (owner-authored broadcasts) -------
+//
+// Backend: work/meshai/dashboard/api/announcement_routes.py. New rows
+// always start disabled (create never accepts `enabled`); PUT is the only
+// place that can arm one. There is deliberately no send-now endpoint.
+
+// A single delivery target on an announcement. `channel` is a Meshtastic
+// channel index (int) or a MeshCore channel name (str); `name` is a
+// display label captured at save time so the target stays readable even
+// if the radio is later unreachable or the channel is renamed/removed.
+export interface AnnouncementChannelRef {
+  transport: 'meshtastic' | 'meshcore'
+  channel: number | string
+  name?: string
+}
+
+export interface Announcement {
+  announcement_id: number
+  name: string
+  message: string
+  schedule_kind: 'daily' | 'interval_days' | 'weekly' | 'monthly'
+  time_of_day: string
+  interval_days: number | null
+  dow_mask: boolean[] | null
+  day_of_month: number | null
+  timezone: string
+  channels: AnnouncementChannelRef[]
+  enabled: boolean
+  last_sent_at: number | null
+  created_at: number
+  updated_at: number
+}
+
+// Fields the create/edit form fills in. Shared by POST (as-is) and PUT
+// (merged as a partial update by the backend).
+export interface AnnouncementDraft {
+  name: string
+  message: string
+  schedule_kind: 'daily' | 'interval_days' | 'weekly' | 'monthly'
+  time_of_day: string
+  interval_days?: number | null
+  dow_mask?: boolean[] | null
+  day_of_month?: number | null
+  timezone: string
+  channels: AnnouncementChannelRef[]
+}
+
+export interface AnnouncementPreview {
+  wire_text: string
+  char_count: number
+  byte_count: number
+  budget: number
+  truncated: boolean
+}
+
+async function announcementError(response: Response): Promise<never> {
+  const body = await response.json().catch(() => null)
+  throw new Error(body?.detail || `API error: ${response.status} ${response.statusText}`)
+}
+
+export async function fetchAnnouncements(): Promise<Announcement[]> {
+  return fetchJson<Announcement[]>('/api/announcements')
+}
+
+// Always creates disabled -- there is no `enabled` field in the request body.
+export async function createAnnouncement(draft: AnnouncementDraft): Promise<Announcement> {
+  const response = await fetch('/api/announcements', {
+    method: 'POST',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify(draft),
+  })
+  if (!response.ok) return announcementError(response)
+  return response.json()
+}
+
+// Partial update. Pass only the fields that changed; include `enabled` to
+// arm/disarm -- this is the only endpoint that can flip it.
+export async function updateAnnouncement(
+  id: number,
+  changes: Partial<AnnouncementDraft> & { enabled?: boolean }
+): Promise<Announcement> {
+  const response = await fetch(`/api/announcements/${id}`, {
+    method: 'PUT',
+    headers: { 'Content-Type': 'application/json' },
+    body: JSON.stringify(changes),
+  })
+  if (!response.ok) return announcementError(response)
+  return response.json()
+}
+
+export async function deleteAnnouncement(id: number): Promise<void> {
+  const response = await fetch(`/api/announcements/${id}`, { method: 'DELETE' })
+  if (!response.ok) return announcementError(response)
+}
+
+// Returns the exact wire text plus size-vs-budget info. Never sends.
+export async function previewAnnouncement(id: number): Promise<AnnouncementPreview> {
+  const response = await fetch(`/api/announcements/${id}/preview`, { method: 'POST' })
+  if (!response.ok) return announcementError(response)
+  return response.json()
+}

work/dashboard-frontend/src/pages/ScheduledBroadcasts.tsx

@@ -7,6 +7,7 @@ import {
   Toggle, NumberInput, TimeInput,
   type NotificationsConfig,
 } from '@/pages/Notifications'
+import AnnouncementsPanel from '@/components/AnnouncementsPanel'
 
 interface Props {
   family?: 'meshtastic' | 'meshcore'
@@ -214,6 +215,10 @@ export default function ScheduledBroadcasts({ family = 'meshtastic' }: Props) {
         )}
         <p className="text-xs text-slate-600">All times are Mountain Time (America/Boise). DST handled automatically.</p>
       </div>
+
+      {/* Announcements -- owner-authored free-text broadcasts, separate from
+          the automatic band-conditions/hazard notifications above. */}
+      <AnnouncementsPanel />
     </div>
   )
 }

work/meshai/dashboard/api/announcement_routes.py

@@ -0,0 +1,372 @@
+"""Custom scheduled-announcement REST API (v30 custom_announcements table).
+
+New resource type (owner-authored free-text broadcasts on a clock-slot
+schedule -- see notifications/scheduled/custom_announcements.py), so this is
+a NEW router file rather than an extension of notification_routes.py:
+notification_routes.py's CRUD-shaped endpoints (region-routing, rules) are
+all config-file-backed (save_section/load_config over YAML), while
+announcements are individual SQLite rows with their own id, closer in shape
+to adapter_config_routes.py's raw get_db()/HTTPException style -- which this
+file follows (grouped list, per-row GET/PUT/DELETE, explicit 400s with a
+clear message per bad field, no ORM).
+
+Endpoints:
+    GET    /api/announcements                 -- list all, newest first
+    POST   /api/announcements                  -- create (ALWAYS starts disabled)
+    PUT    /api/announcements/{id}             -- partial update (validated as a whole)
+    DELETE /api/announcements/{id}             -- delete
+    POST   /api/announcements/{id}/preview     -- exact wire text + char/byte
+                                                   count against budget.
+                                                   NEVER sends.
+
+There is deliberately NO "send now" / test-send route anywhere in this file.
+"""
+from __future__ import annotations
+
+import json
+import logging
+import time
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, HTTPException, Request
+from pydantic import BaseModel
+
+from meshai.notifications.formatters._budget import budget_for, fit_to_budget
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(tags=["announcements"])
+
+_VALID_KINDS = {"daily", "interval_days", "weekly", "monthly"}
+_VALID_TRANSPORTS = {"meshtastic", "meshcore"}
+
+
+# ============================================================================
+# request bodies
+# ============================================================================
+
+
+class AnnouncementCreateBody(BaseModel):
+    name: str
+    message: str
+    schedule_kind: str
+    time_of_day: str
+    interval_days: Optional[int] = None
+    dow_mask: Optional[List[Any]] = None
+    day_of_month: Optional[int] = None
+    timezone: str = "America/Boise"
+    channels: List[Dict[str, Any]]
+
+
+class AnnouncementUpdateBody(BaseModel):
+    """Partial update -- only fields present in the request change.
+
+    The MERGED result (existing row + these overrides) is re-validated as a
+    whole, so e.g. switching schedule_kind to "weekly" without also
+    supplying dow_mask is rejected even though dow_mask itself wasn't
+    touched by this request.
+    """
+    name: Optional[str] = None
+    message: Optional[str] = None
+    schedule_kind: Optional[str] = None
+    time_of_day: Optional[str] = None
+    interval_days: Optional[int] = None
+    dow_mask: Optional[List[Any]] = None
+    day_of_month: Optional[int] = None
+    timezone: Optional[str] = None
+    channels: Optional[List[Dict[str, Any]]] = None
+    enabled: Optional[bool] = None
+
+
+# ============================================================================
+# helpers
+# ============================================================================
+
+
+def _get_conn():
+    from meshai.persistence import get_db
+    return get_db()
+
+
+def _row_to_dict(r) -> dict:
+    channels_raw = r["channels"]
+    try:
+        channels = json.loads(channels_raw) if channels_raw else []
+    except Exception:
+        channels = []
+    dow_mask_raw = r["dow_mask"]
+    try:
+        dow_mask = json.loads(dow_mask_raw) if dow_mask_raw else None
+    except Exception:
+        dow_mask = None
+    return {
+        "announcement_id": r["announcement_id"],
+        "name": r["name"],
+        "message": r["message"],
+        "schedule_kind": r["schedule_kind"],
+        "time_of_day": r["time_of_day"],
+        "interval_days": r["interval_days"],
+        "dow_mask": dow_mask,
+        "day_of_month": r["day_of_month"],
+        "timezone": r["timezone"],
+        "channels": channels,
+        "enabled": bool(r["enabled"]),
+        "last_sent_at": r["last_sent_at"],
+        "created_at": r["created_at"],
+        "updated_at": r["updated_at"],
+    }
+
+
+def _validate_time_of_day(value: Any) -> str:
+    if not isinstance(value, str) or ":" not in value:
+        raise HTTPException(400, f"time_of_day must be 'HH:MM', got {value!r}")
+    parts = value.split(":")
+    if len(parts) != 2:
+        raise HTTPException(400, f"time_of_day must be 'HH:MM', got {value!r}")
+    try:
+        hh, mm = int(parts[0]), int(parts[1])
+    except ValueError:
+        raise HTTPException(400, f"time_of_day must be 'HH:MM' with numeric parts, got {value!r}")
+    if not (0 <= hh <= 23 and 0 <= mm <= 59):
+        raise HTTPException(400, f"time_of_day out of range, got {value!r}")
+    return f"{hh:02d}:{mm:02d}"
+
+
+def _validate_timezone(value: Any) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise HTTPException(400, "timezone must be a non-empty string")
+    try:
+        from zoneinfo import ZoneInfo
+        ZoneInfo(value)
+    except Exception:
+        raise HTTPException(400, f"unknown timezone {value!r}")
+    return value
+
+
+def _validate_channels(value: Any) -> List[Dict[str, Any]]:
+    """SHAPE validation only. Rejects an empty list (nowhere to send) but
+    places NO upper bound on length -- any number of mixed meshtastic/
+    meshcore targets is valid. Does NOT check whether a channel is
+    currently configured on a live radio (that is a send-time concern --
+    dispatch_scheduled_custom_broadcast logs a warning and skips a target
+    that no longer exists rather than failing the whole announcement)."""
+    if not isinstance(value, list) or len(value) == 0:
+        raise HTTPException(400, "channels must be a non-empty list")
+    out = []
+    for i, item in enumerate(value):
+        if not isinstance(item, dict):
+            raise HTTPException(400, f"channels[{i}] must be an object")
+        transport = item.get("transport")
+        if transport not in _VALID_TRANSPORTS:
+            raise HTTPException(
+                400,
+                f"channels[{i}].transport must be one of {sorted(_VALID_TRANSPORTS)}, "
+                f"got {transport!r}",
+            )
+        channel = item.get("channel")
+        if transport == "meshtastic":
+            if isinstance(channel, bool) or not isinstance(channel, int):
+                raise HTTPException(
+                    400, f"channels[{i}].channel must be an int meshtastic "
+                         f"channel index, got {channel!r}")
+        else:  # meshcore
+            if not isinstance(channel, str) or not channel.strip():
+                raise HTTPException(
+                    400, f"channels[{i}].channel must be a non-empty "
+                         f"meshcore channel name, got {channel!r}")
+        entry = {"transport": transport, "channel": channel}
+        name = item.get("name")
+        if name is not None:
+            entry["name"] = name
+        out.append(entry)
+    return out
+
+
+def _validate_dow_mask(value: Any) -> List[bool]:
+    if not isinstance(value, list) or len(value) != 7 or not all(
+        isinstance(b, bool) for b in value
+    ):
+        raise HTTPException(
+            400, f"dow_mask must be a list of exactly 7 booleans, got {value!r}")
+    return value
+
+
+def _validate_full(merged: dict) -> dict:
+    """Validate a full (post-merge) announcement record. Returns a cleaned
+    dict ready to persist. Raises HTTPException(400) on the first bad field."""
+    name = merged.get("name")
+    if not isinstance(name, str) or not name.strip():
+        raise HTTPException(400, "name must be a non-empty string")
+
+    message = merged.get("message")
+    if not isinstance(message, str) or not message.strip():
+        raise HTTPException(400, "message must be a non-empty string")
+
+    kind = merged.get("schedule_kind")
+    if kind not in _VALID_KINDS:
+        raise HTTPException(
+            400, f"schedule_kind must be one of {sorted(_VALID_KINDS)}, got {kind!r}")
+
+    time_of_day = _validate_time_of_day(merged.get("time_of_day"))
+    tz_name = _validate_timezone(merged.get("timezone") or "America/Boise")
+    channels = _validate_channels(merged.get("channels"))
+
+    interval_days = merged.get("interval_days")
+    dow_mask = merged.get("dow_mask")
+    day_of_month = merged.get("day_of_month")
+
+    if kind == "interval_days":
+        if not isinstance(interval_days, int) or isinstance(interval_days, bool) or interval_days < 1:
+            raise HTTPException(
+                400, f"interval_days must be an int >= 1 for schedule_kind="
+                     f"'interval_days', got {interval_days!r}")
+        dow_mask = None
+        day_of_month = None
+    elif kind == "weekly":
+        dow_mask = _validate_dow_mask(dow_mask)
+        interval_days = None
+        day_of_month = None
+    elif kind == "monthly":
+        if (not isinstance(day_of_month, int) or isinstance(day_of_month, bool)
+                or not (1 <= day_of_month <= 31)):
+            raise HTTPException(
+                400, f"day_of_month must be an int 1-31 for schedule_kind="
+                     f"'monthly', got {day_of_month!r}")
+        interval_days = None
+        dow_mask = None
+    else:  # daily
+        interval_days = None
+        dow_mask = None
+        day_of_month = None
+
+    return {
+        "name": name.strip(),
+        "message": message,
+        "schedule_kind": kind,
+        "time_of_day": time_of_day,
+        "interval_days": interval_days,
+        "dow_mask": json.dumps(dow_mask) if dow_mask is not None else None,
+        "day_of_month": day_of_month,
+        "timezone": tz_name,
+        "channels": json.dumps(channels),
+    }
+
+
+# ============================================================================
+# endpoints
+# ============================================================================
+
+
+@router.get("/announcements")
+async def list_announcements(request: Request) -> list[dict]:
+    conn = _get_conn()
+    rows = conn.execute(
+        "SELECT * FROM custom_announcements ORDER BY announcement_id DESC"
+    ).fetchall()
+    return [_row_to_dict(r) for r in rows]
+
+
+@router.post("/announcements")
+async def create_announcement(request: Request, body: AnnouncementCreateBody) -> dict:
+    """Create a new announcement. ALWAYS starts disabled (enabled=0) --
+    the owner arms it explicitly via PUT after reviewing a preview. This
+    endpoint never sends anything."""
+    clean = _validate_full(body.model_dump())
+    now = time.time()
+    conn = _get_conn()
+    cur = conn.execute(
+        "INSERT INTO custom_announcements "
+        "(name, message, schedule_kind, time_of_day, interval_days, "
+        "dow_mask, day_of_month, timezone, channels, enabled, "
+        "last_sent_at, created_at, updated_at) "
+        "VALUES (?,?,?,?,?,?,?,?,?,0,NULL,?,?)",
+        (clean["name"], clean["message"], clean["schedule_kind"],
+         clean["time_of_day"], clean["interval_days"], clean["dow_mask"],
+         clean["day_of_month"], clean["timezone"], clean["channels"],
+         now, now),
+    )
+    new_id = cur.lastrowid
+    logger.info("announcement created id=%s name=%r (disabled)", new_id, clean["name"])
+    r = conn.execute(
+        "SELECT * FROM custom_announcements WHERE announcement_id=?", (new_id,)
+    ).fetchone()
+    return _row_to_dict(r)
+
+
+@router.put("/announcements/{announcement_id}")
+async def update_announcement(
+    announcement_id: int, request: Request, body: AnnouncementUpdateBody
+) -> dict:
+    conn = _get_conn()
+    existing = conn.execute(
+        "SELECT * FROM custom_announcements WHERE announcement_id=?",
+        (announcement_id,),
+    ).fetchone()
+    if existing is None:
+        raise HTTPException(404, f"announcement {announcement_id} not found")
+
+    current = _row_to_dict(existing)
+    overrides = {k: v for k, v in body.model_dump().items() if v is not None}
+    enabled_override = overrides.pop("enabled", None)
+    merged = {**current, **overrides}
+    clean = _validate_full(merged)
+
+    new_enabled = current["enabled"] if enabled_override is None else bool(enabled_override)
+
+    now = time.time()
+    conn.execute(
+        "UPDATE custom_announcements SET name=?, message=?, schedule_kind=?, "
+        "time_of_day=?, interval_days=?, dow_mask=?, day_of_month=?, "
+        "timezone=?, channels=?, enabled=?, updated_at=? "
+        "WHERE announcement_id=?",
+        (clean["name"], clean["message"], clean["schedule_kind"],
+         clean["time_of_day"], clean["interval_days"], clean["dow_mask"],
+         clean["day_of_month"], clean["timezone"], clean["channels"],
+         1 if new_enabled else 0, now, announcement_id),
+    )
+    logger.info("announcement updated id=%s enabled=%s", announcement_id, new_enabled)
+    r = conn.execute(
+        "SELECT * FROM custom_announcements WHERE announcement_id=?",
+        (announcement_id,),
+    ).fetchone()
+    return _row_to_dict(r)
+
+
+@router.delete("/announcements/{announcement_id}")
+async def delete_announcement(announcement_id: int, request: Request) -> dict:
+    conn = _get_conn()
+    existing = conn.execute(
+        "SELECT announcement_id FROM custom_announcements WHERE announcement_id=?",
+        (announcement_id,),
+    ).fetchone()
+    if existing is None:
+        raise HTTPException(404, f"announcement {announcement_id} not found")
+    conn.execute(
+        "DELETE FROM custom_announcements WHERE announcement_id=?",
+        (announcement_id,),
+    )
+    logger.info("announcement deleted id=%s", announcement_id)
+    return {"ok": True, "deleted": announcement_id}
+
+
+@router.post("/announcements/{announcement_id}/preview")
+async def preview_announcement(announcement_id: int, request: Request) -> dict:
+    """Return the exact wire text and its size against the mesh packet
+    budget. Does NOT send -- there is no send-now endpoint in this API."""
+    conn = _get_conn()
+    r = conn.execute(
+        "SELECT message FROM custom_announcements WHERE announcement_id=?",
+        (announcement_id,),
+    ).fetchone()
+    if r is None:
+        raise HTTPException(404, f"announcement {announcement_id} not found")
+
+    budget = budget_for("custom_announcements")
+    wire = fit_to_budget(r["message"] or "", budget)
+    return {
+        "wire_text": wire,
+        "char_count": len(wire),
+        "byte_count": len(wire.encode("utf-8")),
+        "budget": budget,
+        "truncated": wire != (r["message"] or ""),
+    }

work/meshai/dashboard/server.py

@@ -61,6 +61,7 @@ def create_app() -> FastAPI:
     from .api.debug_routes import router as debug_router
     from .api.serial_ports_routes import router as serial_ports_router
     from .api.gauge_sites_import import router as gauge_sites_import_router
+    from .api.announcement_routes import router as announcement_router
 
     app.include_router(system_router, prefix="/api")
     app.include_router(serial_ports_router, prefix="/api")
@@ -77,6 +78,7 @@ def create_app() -> FastAPI:
     app.include_router(notification_router, prefix="/api")
     app.include_router(secrets_router, prefix="/api")
     app.include_router(debug_router, prefix="/api")
+    app.include_router(announcement_router, prefix="/api")
     # WebSocket router (no prefix, path is /ws/live)
     app.include_router(ws_router)
 

work/meshai/main.py

@@ -469,7 +469,8 @@ class MeshAI:
         # MUST match the adapter_config section each handler reads its budget from
         # (see meshai.central.budget.budget_for calls in the handlers).
         from meshai.adapter_config import set_runtime_override
-        for _adapter in ("nws", "incident", "wfigs", "avalanche", "satpass", "usgs_quake"):
+        for _adapter in ("nws", "incident", "wfigs", "avalanche", "satpass", "usgs_quake",
+                          "custom_announcements"):
             set_runtime_override(_adapter, "single_packet_max_chars", self.connector.max_chars)
 
         # Passive mesh context buffer

work/meshai/notifications/pipeline/__init__.py

@@ -43,6 +43,12 @@ try:
     from meshai.notifications.reminders import ReminderScheduler
 except ImportError:
     ReminderScheduler = None
+try:
+    from meshai.notifications.scheduled.custom_announcements import (
+        CustomAnnouncementScheduler,
+    )
+except ImportError:
+    CustomAnnouncementScheduler = None
 from meshai.notifications.pipeline.inhibitor import Inhibitor
 from meshai.notifications.pipeline.grouper import Grouper
 from meshai.notifications.pipeline.toggle_filter import ToggleFilter
@@ -288,6 +294,22 @@ async def start_pipeline(bus: EventBus, config) -> DigestScheduler:
             _lg.getLogger("meshai.pipeline").exception(
                 "reminder scheduler failed to start")
 
+    # Custom announcements scheduler -- runs alongside the reminder
+    # scheduler. Best-effort: failures must NOT break notifications
+    # pipeline startup.
+    if CustomAnnouncementScheduler is not None:
+        try:
+            comps = getattr(bus, "_pipeline_components", {}) or {}
+            disp = comps.get("dispatcher")
+            if disp is not None:
+                ca_sched = CustomAnnouncementScheduler(disp)
+                await ca_sched.start()
+                comps["custom_announcement_scheduler"] = ca_sched
+                bus._pipeline_components = comps
+        except Exception:
+            import logging as _lg
+            _lg.getLogger("meshai.pipeline").exception(
+                "custom announcement scheduler failed to start")
 
     # Phase 2.16.1: periodically flush the grouper so coalesced events are
     # delivered within the window even when poll cadence is sparse.

work/meshai/notifications/pipeline/dispatcher.py

@@ -1330,6 +1330,150 @@ class Dispatcher:
                     ch_type)
         return delivered_any
 
+    async def dispatch_scheduled_custom_broadcast(
+        self, text: str, *,
+        announcement_id: int,
+        slot_key: str,
+        channels: list,
+    ) -> bool:
+        """Scheduled broadcast for a user-authored custom announcement.
+
+        Unlike dispatch_scheduled_fire_broadcast / dispatch_scheduled_roads_
+        broadcast (which resolve channels via a toggle + region_routes
+        matrix), a custom announcement carries its OWN explicit destination
+        list -- set by the owner in the GUI, never derived from a toggle or
+        a region. Every entry in `channels` is delivered independently and
+        gets its OWN mesh_broadcasts_out audit row: an announcement with N
+        mixed Meshtastic/MeshCore targets produces N delivery attempts and N
+        audit rows. There is no cap on N and no "primary channel" concept.
+
+        channels: list of {"transport": "meshtastic"|"meshcore",
+                            "channel": <int index>|<str name>,
+                            "name": <optional display name, ignored here>}.
+        A target with an unrecognised transport is logged and skipped. A
+        target whose channel is valid in SHAPE but no longer configured on
+        the live radio is NOT a hard failure -- create_channel()/deliver()
+        already logs a warning and returns False for that case (missing
+        connector, unset meshcore_channel, no meshcore transport, etc.); this
+        method adds its own warning on top and moves on to the next target,
+        so one stale MeshCore channel never aborts the rest of the
+        announcement.
+
+        Fan-out pacing: sends are issued one target at a time, same as
+        dispatch_scheduled_fire_broadcast's `plan` loop. Each send goes
+        through create_channel() -> connector.send_message_async() ->
+        RadioSendQueue, which already applies that transport's inter-packet
+        jitter -- so multi-target announcements are naturally paced without
+        a second spacing mechanism here.
+
+        source_event_table is always "custom_announcements";
+        source_event_pk is f"{announcement_id}:{slot_key}" so the audit
+        trail carries a key unique per (announcement, fired slot) rather
+        than colliding across sends of the same announcement on different
+        days.
+
+        Cold-start grace still applies (consistent with the other scheduled
+        broadcasts). Returns True on at least one successful mesh delivery.
+        """
+        grace_s = int(getattr(self._config.notifications,
+                              "cold_start_grace_seconds", 60) or 0)
+        if grace_s > 0:
+            now_anchor = time.time()
+            if self._first_event_at is None:
+                self._first_event_at = now_anchor
+                self._persist_state()
+            if (now_anchor - self._first_event_at) < grace_s:
+                self._cold_start_dropped += 1
+                self._persist_state()
+                self._logger.info(
+                    "cold-start grace: dropping scheduled custom broadcast "
+                    "id=%s slot=%s", announcement_id, slot_key)
+                return False
+
+        if not channels:
+            self._logger.info(
+                "scheduled-custom-broadcast: no channels for id=%s; dropping",
+                announcement_id)
+            return False
+
+        from meshai.notifications.events import make_event, make_payload_from_event
+        from types import SimpleNamespace
+        ev = make_event(
+            source="custom_announcement", category="custom_announcement",
+            severity="routine", title=text,
+        )
+        ev.data["_meshai_precomposed"] = True
+        # Minimal stand-in for a toggle object -- _toggle_to_rule() only
+        # reads .name off it when mt_override/mc_override are both supplied
+        # (which they always are here, one or the other per target).
+        fake_tog = SimpleNamespace(name=f"custom_announcement:{announcement_id}")
+
+        source_event_pk = f"{announcement_id}:{slot_key}"
+        delivered_any = False
+        for target in channels:
+            target = target or {}
+            transport = target.get("transport")
+            chan = target.get("channel")
+            if transport == "meshtastic":
+                ch_type = "mesh_broadcast"
+            elif transport == "meshcore":
+                ch_type = "meshcore_broadcast"
+            else:
+                self._logger.warning(
+                    "scheduled-custom-broadcast: unknown transport %r for "
+                    "id=%s; skipping target", transport, announcement_id)
+                continue
+
+            rule = self._toggle_to_rule(
+                fake_tog, ch_type, ev,
+                mt_override=(chan if ch_type == "mesh_broadcast" else None),
+                mc_override=(chan if ch_type == "meshcore_broadcast" else None),
+            )
+            try:
+                channel = self._channel_factory(rule, self._connector)
+                payload = make_payload_from_event(ev, message=text)
+                success = await channel.deliver(payload, rule)
+            except Exception:
+                self._logger.exception(
+                    "scheduled-custom-broadcast: delivery raised for id=%s "
+                    "transport=%s channel=%r",
+                    announcement_id, transport, chan)
+                success = False
+
+            if success:
+                delivered_any = True
+                self._logger.info(
+                    "scheduled-custom-broadcast: dispatched id=%s slot=%s "
+                    "via %s ch=%r", announcement_id, slot_key, ch_type, chan)
+            else:
+                self._logger.warning(
+                    "scheduled-custom-broadcast: delivery failed/skipped "
+                    "for id=%s transport=%s channel=%r (channel may not be "
+                    "currently configured on the radio)",
+                    announcement_id, transport, chan)
+
+            # One audit row PER TARGET, mirroring the fire/roads scheduled
+            # broadcasts' per-channel audit insert.
+            try:
+                from meshai.persistence import get_db
+                conn = get_db()
+                bytes_sent = len(text.encode("utf-8")) if text else 0
+                audit_transport, channel_id, recipient = self._audit_route(rule, ch_type)
+                conn.execute(
+                    "INSERT INTO mesh_broadcasts_out(sent_at, recipient, "
+                    "channel, text, source_event_table, source_event_pk, "
+                    "bytes_sent, ack_received, transport, success) "
+                    "VALUES (?,?,?,?,?,?,?,?,?,?)",
+                    (int(time.time()), recipient, channel_id, text,
+                     "custom_announcements", source_event_pk, bytes_sent, 0,
+                     audit_transport, 1 if success else 0),
+                )
+            except Exception:
+                self._logger.exception(
+                    "scheduled-custom-broadcast: audit row insert failed "
+                    "for id=%s transport=%s", announcement_id, transport)
+        return delivered_any
+
     @staticmethod
     def _audit_route(rule, ch_type: str):
         """Resolve (transport, channel_id, recipient) for a mesh delivery.

work/meshai/notifications/reminders/__init__.py

@@ -59,6 +59,9 @@ import time
 from datetime import datetime, timezone
 from typing import Any, Optional
 
+from meshai.notifications.formatters._budget import budget_for, fit_to_budget
+from meshai.notifications.formatters.fire import _fire_anchor
+
 logger = logging.getLogger(__name__)
 
 
@@ -370,8 +373,9 @@ class ReminderScheduler:
             cont = row["current_contained_pct"]
             acres_s = "N/A" if acres is None else f"{int(acres):,} ac"
             cont_s  = "?" if cont  is None else f"{int(cont)}%"
-            anchor = " / ".join(p for p in (row["county"], row["state"]) if p) or "?"
-            return f"🔥 {prefix}: {name}, {anchor}: {acres_s}, {cont_s} contained"
+            anchor = _fire_anchor(dict(row))
+            wire = f"🔥 {prefix}: {name}, {anchor}: {acres_s}, {cont_s} contained"
+            return fit_to_budget(wire, budget_for("wfigs"))
         if adapter == "swpc":
             return f"🌌 {prefix}: ongoing space weather event ({row['event_type']})"
         if adapter == "itd_511_work_zone":

work/meshai/notifications/scheduled/custom_announcements.py

@@ -0,0 +1,353 @@
+"""CustomAnnouncementScheduler -- user-crafted scheduled announcements.
+
+The owner's words: "at xx:xx time every day / other day / specific day /
+week / month, send the following message: 'literally anything I type in
+this box in the GUI'". Free text only -- NO placeholders, NO data sources,
+NO templating.
+
+Modelled on ReminderScheduler's clock-slot tick loop
+(notifications/reminders/__init__.py) -- same 60s tick, same timezone
+localisation via `zoneinfo`, same `spacing_seconds` roll-call pacing between
+consecutive dispatches -- but reading from the `custom_announcements` table
+(persistence/migrations/v30.sql) instead of adapter_config, and dispatching
+via ``Dispatcher.dispatch_scheduled_custom_broadcast()`` (the announcement's
+OWN explicit channel list) instead of a toggle/region_routes path.
+
+Recurrence kinds (schedule_kind column)
+----------------------------------------
+    daily         -- fires every day at time_of_day.
+    interval_days -- fires every N days, counted from the LOCAL CALENDAR
+                     DATE (in the announcement's own timezone) of
+                     created_at. created_at's date is day 0; day N, 2N, 3N...
+                     are eligible. Documented anchor choice: created_at
+                     rather than "now" so the cadence is stable and doesn't
+                     drift if the scheduler is down for a while -- an
+                     announcement created on day 0 with interval_days=2
+                     always lands on days 0, 2, 4, ... relative to its own
+                     creation date, never renumbered by a restart.
+    weekly        -- fires on days where dow_mask[local_weekday] is True.
+                     dow_mask is Mon-first (index 0 = Monday .. 6 = Sunday),
+                     matching Python's datetime.weekday().
+    monthly       -- fires on day_of_month, CLAMPED to the last day of the
+                     local month (calendar.monthrange) so day_of_month=31
+                     fires on Feb 28 (or 29 in a leap year), Apr 30, etc.
+
+Dedup (restart-safe, no double-send on a slot)
+-----------------------------------------------
+Every schedule_kind fires AT MOST ONCE per local calendar day (the
+recurrence math above decides WHICH days are eligible; time_of_day is the
+only slot on an eligible day). That makes "the concrete local datetime of
+the slot" reducible to just the local CALENDAR DATE in the announcement's
+own timezone -- there is never a second slot to disambiguate within a day.
+
+The dedup key is therefore (announcement_id, local_date_str). It is checked
+by comparing the LOCAL DATE of the stored `last_sent_at` (converted into the
+announcement's timezone) against the local date of the slot under
+consideration: if they match, the slot has already fired and is skipped.
+`last_sent_at` is written through to SQLite immediately after a successful
+send (before moving to the next row), so a restart mid-minute re-reads the
+same persisted date on its next tick and will not re-fire a slot that
+already went out -- there is no in-memory-only state a crash can lose.
+
+(The remaining crash window -- a process dying in between a successful
+`dispatch_scheduled_custom_broadcast()` call returning True and the
+`UPDATE ... SET last_sent_at` write landing -- mirrors the exact same
+after-dispatch-stamp pattern already used by ReminderScheduler and the
+other scheduled dispatchers in this codebase; closing it fully would need a
+durable pre-commit intent log, which nothing else here has either.)
+
+Pacing
+------
+Reuses ReminderScheduler's `_space()` mechanism verbatim in spirit: a
+shared `_last_dispatch_at` timestamp and a minimum gap between consecutive
+SUCCESSFUL announcement dispatches, so N announcements eligible in the same
+tick go out spaced apart rather than bursting. This is roll-call-level
+pacing (announcement vs. announcement). Fan-out to the multiple channels
+WITHIN one announcement is not separately throttled here -- each channel
+send already goes through `RadioSendQueue`'s per-transport inter-packet
+jitter at the connector layer (the same mechanism dispatch_scheduled_
+fire_broadcast's multi-channel `plan` loop relies on), so a 5-target mixed
+announcement is naturally paced without inventing a second mechanism.
+"""
+from __future__ import annotations
+
+import asyncio
+import calendar
+import json
+import logging
+import time
+from datetime import datetime, timezone
+from typing import Any, Optional
+
+try:
+    from zoneinfo import ZoneInfo
+except ImportError:  # pragma: no cover (3.9+ only)
+    ZoneInfo = None
+
+from meshai.notifications.formatters._budget import budget_for, fit_to_budget
+
+logger = logging.getLogger(__name__)
+
+
+_TICK_SECONDS = 60.0
+
+# Minimum gap between consecutive SUCCESSFUL announcement dispatches when
+# more than one is eligible in the same tick. Matches the FirePacer /
+# ReminderScheduler default so every scheduled-broadcast exit paces at the
+# same rate.
+_DEFAULT_SPACING_SECONDS = 60.0
+
+_VALID_KINDS = frozenset({"daily", "interval_days", "weekly", "monthly"})
+
+
+# ============================================================================
+# Pure helpers -- no DB / dispatcher, easy to unit test directly.
+# ============================================================================
+
+
+def clamp_day_of_month(year: int, month: int, day: int) -> int:
+    """Clamp `day` (1-31) to the last real day of `year`-`month`.
+
+    31 -> 28 or 29 in February, 30 in Apr/Jun/Sep/Nov, unchanged elsewhere.
+    """
+    last = calendar.monthrange(year, month)[1]
+    return min(max(1, int(day)), last)
+
+
+def _localize(now: float, tz_name: str) -> datetime:
+    dt_utc = datetime.fromtimestamp(now, tz=timezone.utc)
+    if ZoneInfo is not None:
+        try:
+            return dt_utc.astimezone(ZoneInfo(tz_name))
+        except Exception:
+            logger.warning("custom_announcements: bad timezone %r; using UTC", tz_name)
+    return dt_utc
+
+
+def _local_date_str(epoch: float, tz_name: str) -> str:
+    return _localize(epoch, tz_name).strftime("%Y-%m-%d")
+
+
+def is_day_eligible(row: dict, local_dt: datetime) -> bool:
+    """Does `local_dt`'s local calendar date match this announcement's
+    recurrence pattern? (Ignores time-of-day -- caller checks that
+    separately.) `row` is a mapping with the custom_announcements columns.
+    """
+    kind = row["schedule_kind"]
+    if kind == "daily":
+        return True
+
+    if kind == "interval_days":
+        interval = int(row["interval_days"] or 1)
+        if interval <= 0:
+            interval = 1
+        tz_name = row["timezone"] or "America/Boise"
+        created_at = row["created_at"]
+        if created_at is None:
+            return False
+        anchor_date = _localize(float(created_at), tz_name).date()
+        delta_days = (local_dt.date() - anchor_date).days
+        return delta_days >= 0 and delta_days % interval == 0
+
+    if kind == "weekly":
+        mask = row["dow_mask"]
+        if isinstance(mask, str):
+            try:
+                mask = json.loads(mask)
+            except Exception:
+                mask = None
+        if not isinstance(mask, list) or len(mask) != 7:
+            logger.warning(
+                "custom_announcements: bad dow_mask %r for weekly "
+                "announcement; treating as never-fire", mask)
+            return False
+        return bool(mask[local_dt.weekday()])
+
+    if kind == "monthly":
+        dom = row["day_of_month"]
+        if dom is None:
+            return False
+        clamped = clamp_day_of_month(local_dt.year, local_dt.month, int(dom))
+        return local_dt.day == clamped
+
+    logger.warning("custom_announcements: unknown schedule_kind=%r", kind)
+    return False
+
+
+# ============================================================================
+# Scheduler
+# ============================================================================
+
+
+class CustomAnnouncementScheduler:
+    """Fires enabled custom_announcements rows at their configured slots."""
+
+    def __init__(self, dispatcher, *,
+                  clock=None, sleep=None,
+                  tick_seconds: float = _TICK_SECONDS,
+                  spacing_seconds: float = _DEFAULT_SPACING_SECONDS):
+        self._dispatcher = dispatcher
+        self._clock = clock or time.time
+        self._sleep = sleep or asyncio.sleep
+        self._tick = tick_seconds
+        self._spacing = spacing_seconds
+        self._task: Optional[asyncio.Task] = None
+        self._stop: Optional[asyncio.Event] = None
+        # Shared across every announcement -- the mesh is one shared medium,
+        # same design as ReminderScheduler._last_dispatch_at.
+        self._last_dispatch_at: Optional[float] = None
+
+    async def start(self) -> None:
+        if self._task is not None and not self._task.done():
+            raise RuntimeError("CustomAnnouncementScheduler already running")
+        self._stop = asyncio.Event()
+        self._task = asyncio.create_task(self._run(), name="custom-announcement-scheduler")
+        logger.info("CustomAnnouncementScheduler started; tick=%ss", self._tick)
+
+    async def stop(self) -> None:
+        if self._stop: self._stop.set()
+        if self._task:
+            try: await self._task
+            except Exception: pass
+
+    async def _run(self) -> None:
+        while not (self._stop and self._stop.is_set()):
+            try:
+                await self.tick_once()
+            except Exception:
+                logger.exception("CustomAnnouncementScheduler tick failed")
+            try:
+                await asyncio.wait_for(self._stop.wait(), timeout=self._tick)
+                return
+            except asyncio.TimeoutError:
+                pass
+
+    async def tick_once(self, now: Optional[float] = None) -> int:
+        """One pass over every enabled announcement. Returns count fired.
+
+        Public so tests can drive single ticks deterministically."""
+        now = now if now is not None else self._clock()
+        try:
+            from meshai.persistence import get_db
+            conn = get_db()
+        except Exception:
+            return 0
+
+        rows = conn.execute(
+            "SELECT * FROM custom_announcements WHERE enabled = 1"
+        ).fetchall()
+        if not rows:
+            return 0
+
+        fired = 0
+        for row in rows:
+            try:
+                if await self._maybe_fire(dict(row), now):
+                    fired += 1
+            except Exception:
+                logger.exception(
+                    "custom_announcements: tick failed for id=%s",
+                    row["announcement_id"])
+        return fired
+
+    # ---- per-row ----------------------------------------------------
+
+    async def _maybe_fire(self, row: dict, now: float) -> bool:
+        tz_name = row["timezone"] or "America/Boise"
+        local_dt = _localize(now, tz_name)
+
+        if not is_day_eligible(row, local_dt):
+            return False
+
+        hh_mm = row["time_of_day"] or ""
+        try:
+            hh, mm = hh_mm.split(":")
+            slot_min = int(hh) * 60 + int(mm)
+        except Exception:
+            logger.warning(
+                "custom_announcements: bad time_of_day=%r for id=%s",
+                hh_mm, row["announcement_id"])
+            return False
+
+        current_min = local_dt.hour * 60 + local_dt.minute
+        tick_min = max(1, int(self._tick / 60))
+        # Has the slot just passed (within the last tick)? Mirrors
+        # ReminderScheduler._tick_clock's window check.
+        if not (current_min - tick_min <= slot_min <= current_min):
+            return False
+
+        # Dedup: already sent today (this local calendar date)?
+        today_str = local_dt.strftime("%Y-%m-%d")
+        last_sent_at = row.get("last_sent_at")
+        if last_sent_at is not None:
+            if _local_date_str(float(last_sent_at), tz_name) == today_str:
+                return False
+
+        channels = row["channels"]
+        if isinstance(channels, str):
+            try:
+                channels = json.loads(channels)
+            except Exception:
+                channels = []
+        if not channels:
+            logger.warning(
+                "custom_announcements: id=%s has no channels; skipping",
+                row["announcement_id"])
+            return False
+
+        message = row["message"] or ""
+        wire = fit_to_budget(message, budget_for("custom_announcements"))
+        if not wire:
+            return False
+
+        if not await self._space():
+            return False  # stop() signalled -- abandon the roll-call
+
+        slot_key = f"{today_str}T{hh_mm}"
+        ok = False
+        try:
+            ok = bool(await self._dispatcher.dispatch_scheduled_custom_broadcast(
+                text=wire,
+                announcement_id=row["announcement_id"],
+                slot_key=slot_key,
+                channels=channels,
+            ))
+        except Exception:
+            logger.exception(
+                "custom_announcements: dispatch failed id=%s slot=%s",
+                row["announcement_id"], slot_key)
+
+        if ok:
+            self._stamp_sent(row["announcement_id"], now)
+            self._last_dispatch_at = self._clock()
+        return ok
+
+    async def _space(self) -> bool:
+        """Wait out the inter-announcement pacing gap. Returns False if
+        stop() fired while waiting (caller abandons the roll-call). Mirrors
+        ReminderScheduler._space() exactly."""
+        if self._spacing <= 0 or self._last_dispatch_at is None:
+            return True
+        remaining = self._spacing - (self._clock() - self._last_dispatch_at)
+        if remaining <= 0:
+            return True
+        if self._stop is not None:
+            try:
+                await asyncio.wait_for(self._stop.wait(), timeout=remaining)
+                return False
+            except asyncio.TimeoutError:
+                return True
+        await self._sleep(remaining)
+        return True
+
+    def _stamp_sent(self, announcement_id: int, now: float) -> None:
+        try:
+            from meshai.persistence import get_db
+            conn = get_db()
+        except Exception:
+            return
+        conn.execute(
+            "UPDATE custom_announcements SET last_sent_at=?, updated_at=? "
+            "WHERE announcement_id=?",
+            (now, now, announcement_id),
+        )

work/meshai/persistence/migrations/v30.sql

@@ -0,0 +1,56 @@
+-- v30 custom_announcements: user-crafted scheduled announcements.
+--
+-- Free-text broadcasts the owner types straight into the GUI ("at 08:00
+-- every day, send: <literally anything>") -- NO placeholders, NO templating,
+-- NO data source. Distinct from every other scheduled broadcast in this
+-- codebase (band_conditions, reminders, wzdx_summary) in that it carries its
+-- OWN explicit destination list (`channels`) instead of routing through a
+-- toggle + region_routes matrix -- see dispatch_scheduled_custom_broadcast()
+-- in notifications/pipeline/dispatcher.py.
+--
+-- schedule_kind determines which of the cadence columns is authoritative:
+--   daily         -- time_of_day only.
+--   interval_days -- time_of_day + interval_days (anchored at created_at's
+--                    local calendar date -- see CustomAnnouncementScheduler).
+--   weekly        -- time_of_day + dow_mask (JSON list of 7 booleans,
+--                    Mon-first, i.e. index 0 = Monday .. 6 = Sunday).
+--   monthly       -- time_of_day + day_of_month (1-31; clamped to the last
+--                    day of shorter months at fire time, e.g. 31 fires on
+--                    Feb 28/29).
+--
+-- channels is a JSON list of {"transport": "meshtastic"|"meshcore",
+-- "channel": <int index>|<str name>, "name": <optional display name>} --
+-- UNBOUNDED length, arbitrary mix of transports, no "primary channel"
+-- concept. Every entry gets its own delivery + its own mesh_broadcasts_out
+-- audit row (dispatch_scheduled_custom_broadcast loops the list).
+--
+-- New announcements start DISABLED (enabled=0) -- the owner arms them
+-- explicitly after reviewing via POST /api/announcements/{id}/preview.
+-- There is deliberately NO "send now" endpoint anywhere in this feature.
+--
+-- last_sent_at doubles as the scheduler's restart-safe dedup marker: it is
+-- stamped with the actual send epoch, and the scheduler compares the LOCAL
+-- CALENDAR DATE (in `timezone`) of last_sent_at against the local date of
+-- the slot under consideration -- since every schedule_kind fires at most
+-- once per local calendar day, same-date == already-sent-today, and this
+-- survives a restart because it is written through to SQLite immediately
+-- after a successful send (see CustomAnnouncementScheduler._maybe_fire).
+
+CREATE TABLE IF NOT EXISTS custom_announcements (
+    announcement_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    name            TEXT    NOT NULL,
+    message         TEXT    NOT NULL,
+    schedule_kind   TEXT    NOT NULL,   -- daily | interval_days | weekly | monthly
+    time_of_day     TEXT    NOT NULL,   -- "HH:MM"
+    interval_days   INTEGER,            -- interval_days: N (2 = every other day)
+    dow_mask        TEXT,               -- weekly: JSON list of 7 booleans, Mon-first
+    day_of_month    INTEGER,            -- monthly: 1-31, clamped at fire time
+    timezone        TEXT    NOT NULL DEFAULT 'America/Boise',
+    channels        TEXT    NOT NULL,   -- JSON list of {"transport","channel","name"?}
+    enabled         INTEGER NOT NULL DEFAULT 0,
+    last_sent_at    REAL,
+    created_at      REAL    NOT NULL,
+    updated_at      REAL    NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_custom_announcements_enabled ON custom_announcements(enabled);
+CREATE INDEX IF NOT EXISTS idx_custom_announcements_kind    ON custom_announcements(schedule_kind);

work/tests/test_announcement_routes.py

@@ -0,0 +1,324 @@
+"""API tests for /api/announcements (custom scheduled announcements).
+
+Uses FastAPI TestClient, mirroring tests/test_adapter_config_api.py.
+Covers CRUD, validation of every bad-input case named in the spec, the
+create-always-disabled invariant, and that no code path under this router
+ever sends anything (preview is preview-only; there is no send-now route).
+"""
+from __future__ import annotations
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from meshai.dashboard.api.announcement_routes import router
+
+
+@pytest.fixture
+def client():
+    app = FastAPI()
+    app.include_router(router, prefix="/api")
+    return TestClient(app)
+
+
+def _valid_body(**overrides):
+    body = {
+        "name": "Morning greeting",
+        "message": "Good morning, mesh!",
+        "schedule_kind": "daily",
+        "time_of_day": "08:00",
+        "channels": [{"transport": "meshtastic", "channel": 2}],
+    }
+    body.update(overrides)
+    return body
+
+
+# ============================================================================
+# Create -- always starts disabled
+# ============================================================================
+
+
+def test_create_returns_201_shape_and_starts_disabled(client):
+    r = client.post("/api/announcements", json=_valid_body())
+    assert r.status_code == 200
+    body = r.json()
+    assert body["enabled"] is False
+    assert body["name"] == "Morning greeting"
+    assert body["announcement_id"] is not None
+
+
+def test_create_ignores_client_supplied_enabled_true(client):
+    """There is no field in AnnouncementCreateBody for enabled at all --
+    a client cannot arm an announcement at creation time."""
+    r = client.post("/api/announcements", json={**_valid_body(), "enabled": True})
+    assert r.status_code == 200
+    assert r.json()["enabled"] is False
+
+
+# ============================================================================
+# List / get / delete
+# ============================================================================
+
+
+def test_list_returns_created_rows(client):
+    client.post("/api/announcements", json=_valid_body(name="A"))
+    client.post("/api/announcements", json=_valid_body(name="B"))
+    r = client.get("/api/announcements")
+    assert r.status_code == 200
+    names = {row["name"] for row in r.json()}
+    assert names == {"A", "B"}
+
+
+def test_delete_removes_row(client):
+    created = client.post("/api/announcements", json=_valid_body()).json()
+    aid = created["announcement_id"]
+    r = client.delete(f"/api/announcements/{aid}")
+    assert r.status_code == 200
+    assert r.json()["ok"] is True
+    assert client.get("/api/announcements").json() == []
+
+
+def test_delete_unknown_id_404s(client):
+    r = client.delete("/api/announcements/99999")
+    assert r.status_code == 404
+
+
+def test_update_unknown_id_404s(client):
+    r = client.put("/api/announcements/99999", json={"name": "x"})
+    assert r.status_code == 404
+
+
+# ============================================================================
+# Update -- can arm (enable) an announcement, partial merge re-validates whole
+# ============================================================================
+
+
+def test_update_can_enable_after_review(client):
+    created = client.post("/api/announcements", json=_valid_body()).json()
+    aid = created["announcement_id"]
+    r = client.put(f"/api/announcements/{aid}", json={"enabled": True})
+    assert r.status_code == 200
+    assert r.json()["enabled"] is True
+
+
+def test_update_switching_kind_without_required_field_is_rejected(client):
+    """Switching to weekly without supplying dow_mask must fail even though
+    dow_mask itself wasn't part of this request -- the MERGED record is
+    what gets validated."""
+    created = client.post("/api/announcements", json=_valid_body()).json()
+    aid = created["announcement_id"]
+    r = client.put(f"/api/announcements/{aid}", json={"schedule_kind": "weekly"})
+    assert r.status_code == 400
+
+
+def test_update_message_only_preserves_other_fields(client):
+    created = client.post("/api/announcements", json=_valid_body()).json()
+    aid = created["announcement_id"]
+    r = client.put(f"/api/announcements/{aid}", json={"message": "updated text"})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["message"] == "updated text"
+    assert body["schedule_kind"] == "daily"
+    assert body["channels"] == [{"transport": "meshtastic", "channel": 2}]
+
+
+# ============================================================================
+# Validation -- each bad input named in the spec
+# ============================================================================
+
+
+def test_rejects_empty_message(client):
+    r = client.post("/api/announcements", json=_valid_body(message=""))
+    assert r.status_code == 400
+
+
+def test_rejects_empty_name(client):
+    r = client.post("/api/announcements", json=_valid_body(name="  "))
+    assert r.status_code == 400
+
+
+def test_rejects_invalid_hh_mm(client):
+    for bad in ("25:00", "08:60", "not-a-time", "8"):
+        r = client.post("/api/announcements", json=_valid_body(time_of_day=bad))
+        assert r.status_code == 400, f"{bad!r} should have been rejected"
+
+
+def test_rejects_unknown_schedule_kind(client):
+    r = client.post("/api/announcements", json=_valid_body(schedule_kind="hourly"))
+    assert r.status_code == 400
+
+
+def test_rejects_empty_channel_list(client):
+    r = client.post("/api/announcements", json=_valid_body(channels=[]))
+    assert r.status_code == 400
+
+
+def test_does_not_reject_a_long_channel_list(client):
+    """No cap: a long, mixed-transport channel list is valid."""
+    channels = [{"transport": "meshtastic", "channel": i} for i in range(25)]
+    channels += [{"transport": "meshcore", "channel": f"#ch{i}"} for i in range(25)]
+    r = client.post("/api/announcements", json=_valid_body(channels=channels))
+    assert r.status_code == 200
+    assert len(r.json()["channels"]) == 50
+
+
+def test_rejects_day_of_month_out_of_range(client):
+    for bad in (0, 32, -1):
+        r = client.post("/api/announcements", json=_valid_body(
+            schedule_kind="monthly", day_of_month=bad,
+        ))
+        assert r.status_code == 400, f"day_of_month={bad} should have been rejected"
+
+
+def test_monthly_accepts_valid_day_of_month(client):
+    r = client.post("/api/announcements", json=_valid_body(
+        schedule_kind="monthly", day_of_month=31,
+    ))
+    assert r.status_code == 200
+
+
+def test_rejects_dow_mask_wrong_length(client):
+    r = client.post("/api/announcements", json=_valid_body(
+        schedule_kind="weekly", dow_mask=[True, False, True],
+    ))
+    assert r.status_code == 400
+
+
+def test_rejects_dow_mask_non_boolean_entries(client):
+    r = client.post("/api/announcements", json=_valid_body(
+        schedule_kind="weekly", dow_mask=[1, 1, 1, 1, 1, 1, 1],
+    ))
+    assert r.status_code == 400
+
+
+def test_weekly_accepts_valid_dow_mask(client):
+    r = client.post("/api/announcements", json=_valid_body(
+        schedule_kind="weekly", dow_mask=[True, False, False, False, False, False, False],
+    ))
+    assert r.status_code == 200
+
+
+def test_rejects_missing_interval_days_for_interval_kind(client):
+    r = client.post("/api/announcements", json=_valid_body(schedule_kind="interval_days"))
+    assert r.status_code == 400
+
+
+def test_interval_days_accepts_valid_value(client):
+    r = client.post("/api/announcements", json=_valid_body(
+        schedule_kind="interval_days", interval_days=2,
+    ))
+    assert r.status_code == 200
+
+
+def test_rejects_channel_shape_meshtastic_non_int_channel(client):
+    r = client.post("/api/announcements", json=_valid_body(
+        channels=[{"transport": "meshtastic", "channel": "2"}],
+    ))
+    assert r.status_code == 400
+
+
+def test_rejects_channel_shape_meshcore_empty_channel(client):
+    r = client.post("/api/announcements", json=_valid_body(
+        channels=[{"transport": "meshcore", "channel": ""}],
+    ))
+    assert r.status_code == 400
+
+
+def test_rejects_channel_shape_unknown_transport(client):
+    r = client.post("/api/announcements", json=_valid_body(
+        channels=[{"transport": "carrier_pigeon", "channel": 1}],
+    ))
+    assert r.status_code == 400
+
+
+def test_does_not_hard_reject_meshcore_channel_not_currently_on_radio(client):
+    """Shape validation only -- a syntactically valid MeshCore channel name
+    is accepted even though this test never provisions a live radio."""
+    r = client.post("/api/announcements", json=_valid_body(
+        channels=[{"transport": "meshcore", "channel": "#not-provisioned-yet"}],
+    ))
+    assert r.status_code == 200
+
+
+# ============================================================================
+# Preview -- exact wire text + never sends
+# ============================================================================
+
+
+def test_preview_returns_wire_text_and_counts_without_sending(client, monkeypatch):
+    created = client.post("/api/announcements", json=_valid_body(
+        message="Good morning, mesh!",
+    )).json()
+    aid = created["announcement_id"]
+
+    r = client.post(f"/api/announcements/{aid}/preview")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["wire_text"] == "Good morning, mesh!"
+    assert body["char_count"] == len("Good morning, mesh!")
+    assert body["byte_count"] == len("Good morning, mesh!".encode("utf-8"))
+    assert body["truncated"] is False
+    assert "budget" in body
+
+
+def test_preview_truncates_long_message_to_budget(client):
+    long_message = "y" * 500
+    created = client.post("/api/announcements", json=_valid_body(
+        message=long_message,
+    )).json()
+    aid = created["announcement_id"]
+
+    r = client.post(f"/api/announcements/{aid}/preview")
+    assert r.status_code == 200
+    body = r.json()
+    assert body["char_count"] <= 140
+    assert body["truncated"] is True
+    assert body["wire_text"] != long_message
+
+
+def test_preview_unknown_id_404s(client):
+    r = client.post("/api/announcements/99999/preview")
+    assert r.status_code == 404
+
+
+# ============================================================================
+# No send-anywhere guarantee
+# ============================================================================
+
+
+def test_no_send_endpoint_exists_anywhere_in_router():
+    paths = {(getattr(rt, "path", ""), tuple(getattr(rt, "methods", []) or []))
+             for rt in router.routes}
+    for path, methods in paths:
+        assert "send" not in path.lower(), f"unexpected send-shaped route: {path}"
+
+
+def test_create_does_not_touch_mesh_broadcasts_out(client):
+    from meshai.persistence import get_db
+    conn = get_db()
+    before = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"]
+    client.post("/api/announcements", json=_valid_body())
+    after = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"]
+    assert after == before
+
+
+def test_update_does_not_touch_mesh_broadcasts_out(client):
+    from meshai.persistence import get_db
+    conn = get_db()
+    created = client.post("/api/announcements", json=_valid_body()).json()
+    aid = created["announcement_id"]
+    before = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"]
+    client.put(f"/api/announcements/{aid}", json={"enabled": True})
+    after = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"]
+    assert after == before
+
+
+def test_preview_does_not_touch_mesh_broadcasts_out(client):
+    from meshai.persistence import get_db
+    conn = get_db()
+    created = client.post("/api/announcements", json=_valid_body()).json()
+    aid = created["announcement_id"]
+    before = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"]
+    client.post(f"/api/announcements/{aid}/preview")
+    after = conn.execute("SELECT COUNT(*) AS c FROM mesh_broadcasts_out").fetchone()["c"]
+    assert after == before

work/tests/test_custom_announcement_dispatcher.py

@@ -0,0 +1,198 @@
+"""Dispatcher.dispatch_scheduled_custom_broadcast -- own explicit channel
+list (no toggle / no region_routes matrix), unlimited mixed targets, one
+mesh_broadcasts_out audit row per target, cold-start grace.
+
+Follows the RecChannel recorder pattern from
+tests/test_wzdx_summary_region_routing.py.
+"""
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from meshai.config import Config
+from meshai.notifications.pipeline.dispatcher import Dispatcher
+from meshai.persistence import get_db
+
+
+class RecChannel:
+    """Records each delivery's transport + channel value + message."""
+
+    def __init__(self, rec: list, succeed=True):
+        self.rec = rec
+        self.succeed = succeed
+
+    async def deliver(self, payload, rule):
+        ok = self.succeed(rule) if callable(self.succeed) else self.succeed
+        self.rec.append({
+            "delivery_type": rule.delivery_type,
+            "broadcast_channel": getattr(rule, "broadcast_channel", None),
+            "meshcore_channel": getattr(rule, "meshcore_channel", None),
+            "message": payload.message if payload else None,
+            "ok": ok,
+        })
+        return ok
+
+
+def _cfg(cold_start_grace=0):
+    cfg = Config()
+    cfg.notifications.rules = []
+    cfg.notifications.cold_start_grace_seconds = cold_start_grace
+    return cfg
+
+
+def _dispatcher(cfg, succeed=True):
+    rec: list = []
+    d = Dispatcher(cfg, lambda rule, conn: RecChannel(rec, succeed), connector=None)
+    return d, rec
+
+
+def _run(coro):
+    return asyncio.run(coro)
+
+
+# ============================================================================
+# Multi-target fan-out (any number, mixed transports)
+# ============================================================================
+
+
+def test_five_plus_mixed_targets_all_delivered_with_own_audit_row():
+    """5+ mixed targets (3 meshtastic + 2 meshcore) -> every one delivered,
+    every one gets its own mesh_broadcasts_out row."""
+    cfg = _cfg()
+    d, rec = _dispatcher(cfg, succeed=True)
+    conn = get_db()
+
+    channels = [
+        {"transport": "meshtastic", "channel": 0},
+        {"transport": "meshtastic", "channel": 2},
+        {"transport": "meshtastic", "channel": 5},
+        {"transport": "meshcore", "channel": "#sw-id-aida"},
+        {"transport": "meshcore", "channel": "#sc-id-aida"},
+    ]
+
+    ok = _run(d.dispatch_scheduled_custom_broadcast(
+        text="hello everyone", announcement_id=42, slot_key="2026-08-17T08:00",
+        channels=channels,
+    ))
+    assert ok is True
+    assert len(rec) == 5
+
+    mt = [r for r in rec if r["delivery_type"] == "mesh_broadcast"]
+    mc = [r for r in rec if r["delivery_type"] == "meshcore_broadcast"]
+    assert {r["broadcast_channel"] for r in mt} == {0, 2, 5}
+    assert {r["meshcore_channel"] for r in mc} == {"#sw-id-aida", "#sc-id-aida"}
+    assert all(r["message"] == "hello everyone" for r in rec)
+
+    audit_rows = conn.execute(
+        "SELECT transport, channel, source_event_table, source_event_pk, success "
+        "FROM mesh_broadcasts_out WHERE source_event_table='custom_announcements' "
+        "ORDER BY id"
+    ).fetchall()
+    assert len(audit_rows) == 5
+    for r in audit_rows:
+        assert r["source_event_table"] == "custom_announcements"
+        assert r["source_event_pk"] == "42:2026-08-17T08:00"
+        assert r["success"] == 1
+    audit_channels_mt = {r["channel"] for r in audit_rows if r["transport"] == "meshtastic"}
+    audit_channels_mc = {r["channel"] for r in audit_rows if r["transport"] == "meshcore"}
+    assert audit_channels_mt == {0, 2, 5}
+    assert audit_channels_mc == {"#sw-id-aida", "#sc-id-aida"}
+
+
+def test_no_cap_on_number_of_targets():
+    """40 mixed targets all get delivered -- no artificial cap anywhere."""
+    cfg = _cfg()
+    d, rec = _dispatcher(cfg, succeed=True)
+    channels = [{"transport": "meshtastic", "channel": i} for i in range(20)]
+    channels += [{"transport": "meshcore", "channel": f"#ch{i}"} for i in range(20)]
+
+    ok = _run(d.dispatch_scheduled_custom_broadcast(
+        text="big fan-out", announcement_id=1, slot_key="s", channels=channels,
+    ))
+    assert ok is True
+    assert len(rec) == 40
+
+
+def test_empty_channel_list_drops_with_no_delivery():
+    cfg = _cfg()
+    d, rec = _dispatcher(cfg, succeed=True)
+    ok = _run(d.dispatch_scheduled_custom_broadcast(
+        text="nowhere to go", announcement_id=1, slot_key="s", channels=[],
+    ))
+    assert ok is False
+    assert rec == []
+
+
+# ============================================================================
+# A missing/unrecognised target must not abort the rest of the announcement
+# ============================================================================
+
+
+def test_one_failing_target_does_not_abort_the_others(caplog):
+    """One target 'fails' (e.g. channel no longer configured on the radio --
+    deliver() returns False); the other targets still go out, and the
+    failure still gets its own audit row with success=0."""
+    cfg = _cfg()
+    conn = get_db()
+
+    def succeed_unless_missing(rule):
+        return getattr(rule, "meshcore_channel", None) != "#gone"
+
+    d, rec = _dispatcher(cfg, succeed=succeed_unless_missing)
+    channels = [
+        {"transport": "meshtastic", "channel": 1},
+        {"transport": "meshcore", "channel": "#gone"},
+        {"transport": "meshcore", "channel": "#still-here"},
+    ]
+
+    ok = _run(d.dispatch_scheduled_custom_broadcast(
+        text="partial failure ok", announcement_id=7, slot_key="s",
+        channels=channels,
+    ))
+    assert ok is True   # at least one delivery succeeded
+    assert len(rec) == 3
+
+    audit_rows = conn.execute(
+        "SELECT channel, success FROM mesh_broadcasts_out "
+        "WHERE source_event_pk='7:s' ORDER BY id"
+    ).fetchall()
+    assert len(audit_rows) == 3
+    by_channel = {r["channel"]: r["success"] for r in audit_rows}
+    assert by_channel["#gone"] == 0
+    assert by_channel["#still-here"] == 1
+    assert by_channel[1] == 1
+
+
+def test_unknown_transport_is_skipped_not_fatal():
+    cfg = _cfg()
+    d, rec = _dispatcher(cfg, succeed=True)
+    channels = [
+        {"transport": "carrier_pigeon", "channel": "n/a"},
+        {"transport": "meshtastic", "channel": 3},
+    ]
+
+    ok = _run(d.dispatch_scheduled_custom_broadcast(
+        text="skip bad transport", announcement_id=1, slot_key="s",
+        channels=channels,
+    ))
+    assert ok is True
+    assert len(rec) == 1
+    assert rec[0]["broadcast_channel"] == 3
+
+
+# ============================================================================
+# Cold-start grace (consistent with the other scheduled broadcasts)
+# ============================================================================
+
+
+def test_cold_start_grace_suppresses_first_broadcast():
+    cfg = _cfg(cold_start_grace=3600)
+    d, rec = _dispatcher(cfg, succeed=True)
+    ok = _run(d.dispatch_scheduled_custom_broadcast(
+        text="too soon", announcement_id=1, slot_key="s",
+        channels=[{"transport": "meshtastic", "channel": 0}],
+    ))
+    assert ok is False
+    assert rec == []

work/tests/test_custom_announcements_scheduler.py

@@ -0,0 +1,357 @@
+"""CustomAnnouncementScheduler -- recurrence math, dedup, pacing.
+
+Mirrors the shape of tests/test_fire_reminder_pacing.py and
+tests/test_wzdx_summary_region_routing.py: a fake injected clock/sleep so
+the suite pays no wall-clock cost, and a mock dispatcher whose
+dispatch_scheduled_custom_broadcast is asserted on directly.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import time
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from meshai.notifications.scheduled.custom_announcements import (
+    CustomAnnouncementScheduler,
+    clamp_day_of_month,
+    is_day_eligible,
+)
+from meshai.persistence import get_db
+
+
+# ---------- helpers --------------------------------------------------------
+
+
+_TZ = "America/Boise"  # UTC-7 (MST, no DST edge in our fixed test dates)
+
+
+def _epoch_for_local(y, m, d, hh, mm, tz_name=_TZ) -> float:
+    from zoneinfo import ZoneInfo
+    dt = datetime(y, m, d, hh, mm, tzinfo=ZoneInfo(tz_name))
+    return dt.timestamp()
+
+
+def _insert_announcement(
+    conn, *, name="Test", message="hello mesh", schedule_kind="daily",
+    time_of_day="08:00", interval_days=None, dow_mask=None,
+    day_of_month=None, tz_name=_TZ, channels=None, enabled=1,
+    created_at=None, last_sent_at=None,
+) -> int:
+    now = created_at if created_at is not None else time.time()
+    if channels is None:
+        channels = [{"transport": "meshtastic", "channel": 2}]
+    cur = conn.execute(
+        "INSERT INTO custom_announcements "
+        "(name, message, schedule_kind, time_of_day, interval_days, "
+        "dow_mask, day_of_month, timezone, channels, enabled, "
+        "last_sent_at, created_at, updated_at) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        (name, message, schedule_kind, time_of_day, interval_days,
+         json.dumps(dow_mask) if dow_mask is not None else None,
+         day_of_month, tz_name, json.dumps(channels), enabled,
+         last_sent_at, now, now),
+    )
+    return cur.lastrowid
+
+
+@pytest.fixture
+def mock_dispatcher():
+    d = MagicMock()
+    d.dispatch_scheduled_custom_broadcast = AsyncMock(return_value=True)
+    return d
+
+
+def _sched(dispatcher, *, clock=None, sleep=None, spacing_seconds=60.0):
+    return CustomAnnouncementScheduler(
+        dispatcher, clock=clock, sleep=sleep, spacing_seconds=spacing_seconds,
+    )
+
+
+# ============================================================================
+# clamp_day_of_month
+# ============================================================================
+
+
+def test_clamp_day_of_month_february_non_leap():
+    assert clamp_day_of_month(2026, 2, 31) == 28
+
+
+def test_clamp_day_of_month_february_leap():
+    assert clamp_day_of_month(2024, 2, 31) == 29
+
+
+def test_clamp_day_of_month_thirty_day_month():
+    assert clamp_day_of_month(2026, 4, 31) == 30
+
+
+def test_clamp_day_of_month_unaffected_when_in_range():
+    assert clamp_day_of_month(2026, 1, 15) == 15
+
+
+# ============================================================================
+# Recurrence kinds -- each fires on the right day
+# ============================================================================
+
+
+def test_daily_fires_every_day(mock_dispatcher):
+    conn = get_db()
+    now = _epoch_for_local(2026, 8, 17, 8, 0)  # Monday
+    _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00")
+
+    fired = asyncio.run(_sched(mock_dispatcher, clock=lambda: now).tick_once())
+    assert fired == 1
+
+
+def test_interval_days_fires_on_multiple_of_interval_from_creation():
+    """interval_days=2, created on day 0 -> fires day 0, 2, 4... not day 1, 3."""
+    conn = get_db()
+    anchor = _epoch_for_local(2026, 8, 10, 8, 0)  # Monday, day 0
+    row_id = _insert_announcement(
+        conn, schedule_kind="interval_days", interval_days=2,
+        time_of_day="08:00", created_at=anchor,
+    )
+    row = dict(conn.execute(
+        "SELECT * FROM custom_announcements WHERE announcement_id=?", (row_id,)
+    ).fetchone())
+
+    from zoneinfo import ZoneInfo
+    day0 = datetime.fromtimestamp(anchor, tz=timezone.utc).astimezone(ZoneInfo(_TZ))
+    day1 = day0.replace(day=day0.day + 1)
+    day2 = day0.replace(day=day0.day + 2)
+
+    assert is_day_eligible(row, day0) is True
+    assert is_day_eligible(row, day1) is False
+    assert is_day_eligible(row, day2) is True
+
+
+def test_weekly_fires_only_on_masked_days():
+    """dow_mask Mon-first: only Wed (index 2) true."""
+    conn = get_db()
+    mask = [False, False, True, False, False, False, False]
+    row_id = _insert_announcement(
+        conn, schedule_kind="weekly", dow_mask=mask, time_of_day="08:00",
+    )
+    row = dict(conn.execute(
+        "SELECT * FROM custom_announcements WHERE announcement_id=?", (row_id,)
+    ).fetchone())
+
+    from zoneinfo import ZoneInfo
+    tue = datetime(2026, 8, 18, 8, 0, tzinfo=ZoneInfo(_TZ))  # Tuesday
+    wed = datetime(2026, 8, 19, 8, 0, tzinfo=ZoneInfo(_TZ))  # Wednesday
+
+    assert is_day_eligible(row, tue) is False
+    assert is_day_eligible(row, wed) is True
+
+
+def test_monthly_fires_on_day_of_month():
+    conn = get_db()
+    row_id = _insert_announcement(
+        conn, schedule_kind="monthly", day_of_month=15, time_of_day="08:00",
+    )
+    row = dict(conn.execute(
+        "SELECT * FROM custom_announcements WHERE announcement_id=?", (row_id,)
+    ).fetchone())
+
+    from zoneinfo import ZoneInfo
+    the_14th = datetime(2026, 8, 14, 8, 0, tzinfo=ZoneInfo(_TZ))
+    the_15th = datetime(2026, 8, 15, 8, 0, tzinfo=ZoneInfo(_TZ))
+
+    assert is_day_eligible(row, the_14th) is False
+    assert is_day_eligible(row, the_15th) is True
+
+
+def test_monthly_day_31_clamps_in_february_non_leap():
+    """The headline requirement: day_of_month=31 fires on Feb 28 in a non-leap year."""
+    conn = get_db()
+    row_id = _insert_announcement(
+        conn, schedule_kind="monthly", day_of_month=31, time_of_day="09:00",
+    )
+    row = dict(conn.execute(
+        "SELECT * FROM custom_announcements WHERE announcement_id=?", (row_id,)
+    ).fetchone())
+
+    from zoneinfo import ZoneInfo
+    feb28_2026 = datetime(2026, 2, 28, 9, 0, tzinfo=ZoneInfo(_TZ))  # 2026 is not leap
+    feb27_2026 = datetime(2026, 2, 27, 9, 0, tzinfo=ZoneInfo(_TZ))
+
+    assert is_day_eligible(row, feb28_2026) is True
+    assert is_day_eligible(row, feb27_2026) is False
+
+
+def test_monthly_day_31_clamps_in_february_leap_year():
+    conn = get_db()
+    row_id = _insert_announcement(
+        conn, schedule_kind="monthly", day_of_month=31, time_of_day="09:00",
+    )
+    row = dict(conn.execute(
+        "SELECT * FROM custom_announcements WHERE announcement_id=?", (row_id,)
+    ).fetchone())
+
+    from zoneinfo import ZoneInfo
+    feb29_2024 = datetime(2024, 2, 29, 9, 0, tzinfo=ZoneInfo(_TZ))  # 2024 IS leap
+    assert is_day_eligible(row, feb29_2024) is True
+
+
+def test_monthly_day_31_fires_via_full_tick_in_february():
+    """End-to-end: tick_once actually dispatches on Feb 28 for day_of_month=31."""
+    conn = get_db()
+    now = _epoch_for_local(2026, 2, 28, 9, 0)
+    _insert_announcement(
+        conn, schedule_kind="monthly", day_of_month=31, time_of_day="09:00",
+    )
+    d = MagicMock()
+    d.dispatch_scheduled_custom_broadcast = AsyncMock(return_value=True)
+
+    fired = asyncio.run(_sched(d, clock=lambda: now).tick_once())
+    assert fired == 1
+    d.dispatch_scheduled_custom_broadcast.assert_called_once()
+
+
+# ============================================================================
+# Disabled announcements never fire
+# ============================================================================
+
+
+def test_disabled_announcement_never_fires(mock_dispatcher):
+    conn = get_db()
+    now = _epoch_for_local(2026, 8, 17, 8, 0)
+    _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00", enabled=0)
+
+    fired = asyncio.run(_sched(mock_dispatcher, clock=lambda: now).tick_once())
+    assert fired == 0
+    mock_dispatcher.dispatch_scheduled_custom_broadcast.assert_not_called()
+
+
+# ============================================================================
+# Dedup: never send the same slot twice, restart-safe
+# ============================================================================
+
+
+def test_dedup_prevents_double_send_in_same_slot(mock_dispatcher):
+    """Two ticks within the same minute window must not double-fire."""
+    conn = get_db()
+    now = _epoch_for_local(2026, 8, 17, 8, 0)
+    _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00")
+
+    sched = _sched(mock_dispatcher, clock=lambda: now)
+    fired1 = asyncio.run(sched.tick_once())
+    fired2 = asyncio.run(sched.tick_once())
+
+    assert fired1 == 1
+    assert fired2 == 0
+    assert mock_dispatcher.dispatch_scheduled_custom_broadcast.call_count == 1
+
+
+def test_dedup_survives_a_simulated_restart():
+    """A brand-new scheduler instance (simulating a process restart) reads
+    the persisted last_sent_at and still refuses to double-send the slot
+    that already went out."""
+    conn = get_db()
+    now = _epoch_for_local(2026, 8, 17, 8, 0)
+    _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00")
+
+    d1 = MagicMock()
+    d1.dispatch_scheduled_custom_broadcast = AsyncMock(return_value=True)
+    fired1 = asyncio.run(_sched(d1, clock=lambda: now).tick_once())
+    assert fired1 == 1
+
+    # Simulate restart: a fresh scheduler instance, same DB, ticking again
+    # a few seconds later within the same slot window.
+    d2 = MagicMock()
+    d2.dispatch_scheduled_custom_broadcast = AsyncMock(return_value=True)
+    fired2 = asyncio.run(_sched(d2, clock=lambda: now + 5).tick_once())
+    assert fired2 == 0
+    d2.dispatch_scheduled_custom_broadcast.assert_not_called()
+
+
+def test_dedup_allows_the_next_days_slot(mock_dispatcher):
+    conn = get_db()
+    day1 = _epoch_for_local(2026, 8, 17, 8, 0)
+    day2 = _epoch_for_local(2026, 8, 18, 8, 0)
+    _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00")
+
+    fired1 = asyncio.run(_sched(mock_dispatcher, clock=lambda: day1).tick_once())
+    fired2 = asyncio.run(_sched(mock_dispatcher, clock=lambda: day2).tick_once())
+
+    assert fired1 == 1
+    assert fired2 == 1
+
+
+# ============================================================================
+# Message truncation at the budget
+# ============================================================================
+
+
+def test_message_is_truncated_to_budget(mock_dispatcher):
+    conn = get_db()
+    now = _epoch_for_local(2026, 8, 17, 8, 0)
+    long_message = "x" * 500
+    _insert_announcement(
+        conn, schedule_kind="daily", time_of_day="08:00", message=long_message,
+    )
+
+    asyncio.run(_sched(mock_dispatcher, clock=lambda: now).tick_once())
+
+    call = mock_dispatcher.dispatch_scheduled_custom_broadcast.call_args
+    sent_text = call.kwargs["text"]
+    assert len(sent_text) <= 140
+    assert sent_text != long_message
+
+
+# ============================================================================
+# Pacing between multiple announcements firing in the same tick
+# ============================================================================
+
+
+class _FakeTime:
+    def __init__(self, start):
+        self.now = start
+        self.sleeps: list[float] = []
+
+    def clock(self) -> float:
+        return self.now
+
+    async def sleep(self, seconds: float) -> None:
+        self.sleeps.append(seconds)
+        self.now += seconds
+
+
+def test_multiple_announcements_in_one_tick_are_spaced_not_burst():
+    conn = get_db()
+    ft = _FakeTime(_epoch_for_local(2026, 8, 17, 8, 0))
+    for i in range(3):
+        _insert_announcement(
+            conn, name=f"A{i}", schedule_kind="daily", time_of_day="08:00",
+        )
+    sent_at: list[float] = []
+    d = MagicMock()
+    d.dispatch_scheduled_custom_broadcast = AsyncMock(
+        side_effect=lambda **kw: sent_at.append(ft.now) or True
+    )
+
+    fired = asyncio.run(
+        _sched(d, clock=ft.clock, sleep=ft.sleep, spacing_seconds=60.0).tick_once()
+    )
+
+    assert fired == 3
+    gaps = [b - a for a, b in zip(sent_at, sent_at[1:])]
+    assert gaps == [60.0, 60.0]
+
+
+def test_single_announcement_fire_is_not_delayed():
+    conn = get_db()
+    ft = _FakeTime(_epoch_for_local(2026, 8, 17, 8, 0))
+    _insert_announcement(conn, schedule_kind="daily", time_of_day="08:00")
+    d = MagicMock()
+    d.dispatch_scheduled_custom_broadcast = AsyncMock(return_value=True)
+
+    fired = asyncio.run(
+        _sched(d, clock=ft.clock, sleep=ft.sleep, spacing_seconds=60.0).tick_once()
+    )
+
+    assert fired == 1
+    assert ft.sleeps == []

work/tests/test_reminders.py

@@ -229,3 +229,94 @@ def test_work_zone_reminder_skipped_outside_slot(mock_dispatcher):
     sch = ReminderScheduler(mock_dispatcher, clock=lambda: now, tick_seconds=60)
     fired = asyncio.run(sch.tick_once())
     assert fired == 0
+
+
+# ============================================================================
+# Anchor resolution in the wfigs "Active:" reminder wire (_fire_anchor)
+# ============================================================================
+
+
+def test_wfigs_reminder_uses_anchor_not_county(mock_dispatcher):
+    """A fire near a known curated town_anchors entry (Bliss, ID) renders the
+    anchor phrase ('mi <bearing> of Bliss'), not the bare county/state join
+    the old renderer produced."""
+    now = 1_780_000_000
+    conn = get_db()
+    _enable_wfigs_reminders()
+    # Coordinates ~4 mi NW of Bliss, ID (see _TOWN_ANCHORS_SEED in
+    # meshai/persistence/curation.py) -- county/state deliberately different
+    # from "Bliss" so the assertion can't pass by accident via the county.
+    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 (?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        ("F-ANCHOR", "Jackalope", "WF", 15, 0,
+         42.98, -114.99, "Gooding", "ID",
+         now - 9 * 3600, now, now - 9 * 3600, now - 9 * 3600),
+    )
+
+    sch = ReminderScheduler(mock_dispatcher, clock=lambda: now)
+    fired = asyncio.run(sch.tick_once())
+    assert fired == 1
+    args = mock_dispatcher.dispatch_scheduled_fire_broadcast.call_args.kwargs
+    text = args["text"]
+    assert "Bliss" in text
+    assert "of Bliss" in text
+    # Old format ("Gooding / ID") must be gone.
+    assert "Gooding / ID" not in text
+
+
+def test_wfigs_reminder_falls_back_when_no_anchor_nearby(mock_dispatcher):
+    """A fire with no lat/lon at all still renders sensibly via
+    _fire_anchor's fallback chain (county -> state -> "location unknown"),
+    and never raises."""
+    now = 1_780_000_000
+    conn = get_db()
+    _enable_wfigs_reminders()
+    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 (?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        ("F-NOANCHOR", "Remote Fire", "WF", 42, 5,
+         None, None, "Owyhee", "ID",
+         now - 9 * 3600, now, now - 9 * 3600, now - 9 * 3600),
+    )
+
+    sch = ReminderScheduler(mock_dispatcher, clock=lambda: now)
+    fired = asyncio.run(sch.tick_once())
+    assert fired == 1
+    args = mock_dispatcher.dispatch_scheduled_fire_broadcast.call_args.kwargs
+    text = args["text"]
+    # Falls through landclass (absent) to the "{county} Co {state}" tier.
+    assert "Owyhee Co ID" in text
+    assert "Active" in text
+
+
+def test_wfigs_reminder_truncates_to_budget(mock_dispatcher):
+    """A very long incident name + anchor is truncated to the wfigs budget
+    (fit_to_budget) rather than exceeding the mesh packet limit."""
+    now = 1_780_000_000
+    conn = get_db()
+    _enable_wfigs_reminders()
+    long_name = ("The Extremely Long Wildfire Incident Name That Keeps Going "
+                 "And Going And Going Well Past Any Reasonable Mesh Packet Budget")
+    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 (?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        ("F-LONG", long_name, "WF", 123456, 3,
+         42.98, -114.99, "Gooding", "ID",
+         now - 9 * 3600, now, now - 9 * 3600, now - 9 * 3600),
+    )
+
+    from meshai.notifications.formatters._budget import budget_for
+    sch = ReminderScheduler(mock_dispatcher, clock=lambda: now)
+    fired = asyncio.run(sch.tick_once())
+    assert fired == 1
+    args = mock_dispatcher.dispatch_scheduled_fire_broadcast.call_args.kwargs
+    text = args["text"]
+    assert len(text) <= budget_for("wfigs")
+    assert text.endswith("…")