!subscribe, 'Lists all alert categories you can subscribe to'],
- [!subscribe fire_proximity , 'Subscribe to a specific category'],
- [!subscribe all , 'Subscribe to everything'],
- [!unsubscribe fire_proximity , 'Unsubscribe from a category'],
- [!subscriptions , "Shows what you're currently subscribed to"],
- ]}
- />
-
Conversational
Bang commands are the short, predictable interface. For anything that
diff --git a/work/dashboard-frontend/src/pages/ScheduledBroadcasts.tsx b/work/dashboard-frontend/src/pages/ScheduledBroadcasts.tsx
new file mode 100644
index 0000000..b435e67
--- /dev/null
+++ b/work/dashboard-frontend/src/pages/ScheduledBroadcasts.tsx
@@ -0,0 +1,330 @@
+import { useState, useEffect, useCallback } from 'react'
+import { Save, RotateCcw, RefreshCw, Check } from 'lucide-react'
+import { fetchConfig as apiFetchConfig, updateConfig as apiUpdateConfig } from '@/lib/api'
+import { useDirty } from '@/context/DirtyContext'
+import { notifyRestartRequired } from '@/components/RestartBanner'
+import {
+ Toggle, NumberInput, TimeInput, InfoButton,
+ type NotificationsConfig,
+} from '@/pages/Notifications'
+
+// Fires adapter config shape (digest settings)
+interface FiresConfig {
+ digest_enabled: boolean
+ digest_schedule: string[]
+ digest_timezone: string
+}
+
+interface Props {
+ family?: 'meshtastic' | 'meshcore'
+}
+
+export default function ScheduledBroadcasts({ family = 'meshtastic' }: Props) {
+ const { setDirty } = useDirty()
+
+ // Notifications config state (full object — read-modify-write to preserve other fields)
+ const [notifConfig, setNotifConfig] = useState(null)
+ const [originalNotifConfig, setOriginalNotifConfig] = useState(null)
+
+ // Fires adapter config state
+ const [firesConfig, setFiresConfig] = useState({
+ digest_enabled: true,
+ digest_schedule: ['06:00', '18:00'],
+ digest_timezone: 'America/Boise',
+ })
+ const [originalFiresConfig, setOriginalFiresConfig] = useState('')
+
+ const [loading, setLoading] = useState(true)
+ const [saving, setSaving] = useState(false)
+ const [error, setError] = useState(null)
+ const [success, setSuccess] = useState(null)
+ const [hasChanges, setHasChanges] = useState(false)
+
+ const fetchData = useCallback(async () => {
+ setLoading(true)
+ setError(null)
+ try {
+ // Load notifications config (full object for read-modify-write)
+ const notif = (await apiFetchConfig('notifications')) as NotificationsConfig
+ setNotifConfig(notif)
+ setOriginalNotifConfig(JSON.parse(JSON.stringify(notif)))
+
+ // Load fires adapter config
+ try {
+ const firesRes = await fetch('/api/adapter-config/fires')
+ if (firesRes.ok) {
+ const firesData = await firesRes.json()
+ const fires: FiresConfig = {
+ digest_enabled: firesData.digest_enabled?.value ?? true,
+ digest_schedule: firesData.digest_schedule?.value ?? ['06:00', '18:00'],
+ digest_timezone: firesData.digest_timezone?.value ?? 'America/Boise',
+ }
+ setFiresConfig(fires)
+ setOriginalFiresConfig(JSON.stringify(fires))
+ }
+ } catch {
+ // adapter-config optional — proceed with defaults
+ setOriginalFiresConfig(JSON.stringify(firesConfig))
+ }
+
+ setHasChanges(false)
+ } catch (err) {
+ setError(err instanceof Error ? err.message : 'Failed to load config')
+ } finally {
+ setLoading(false)
+ }
+ }, []) // eslint-disable-line react-hooks/exhaustive-deps
+
+ useEffect(() => {
+ document.title = `Scheduled Broadcasts - MeshAI`
+ fetchData()
+ }, [fetchData])
+
+ useEffect(() => {
+ if (notifConfig && originalNotifConfig) {
+ const notifChanged = JSON.stringify(notifConfig) !== JSON.stringify(originalNotifConfig)
+ const firesChanged = JSON.stringify(firesConfig) !== originalFiresConfig
+ setHasChanges(notifChanged || firesChanged)
+ }
+ }, [notifConfig, originalNotifConfig, firesConfig, originalFiresConfig])
+
+ useEffect(() => {
+ setDirty(hasChanges)
+ return () => setDirty(false)
+ }, [hasChanges, setDirty])
+
+ const saveAdapterKey = async (adapter: string, key: string, value: unknown) => {
+ const res = await fetch(`/api/adapter-config/${adapter}/${key}`, {
+ method: 'PUT',
+ headers: { 'Content-Type': 'application/json' },
+ body: JSON.stringify({ value }),
+ })
+ if (!res.ok) {
+ const err = await res.json().catch(() => ({}))
+ throw new Error(err.detail || `Failed to save ${adapter}.${key}`)
+ }
+ }
+
+ const saveConfig = async () => {
+ if (!notifConfig) return
+ setSaving(true)
+ setError(null)
+ setSuccess(null)
+ try {
+ // Save full notifications config (read-modify-write — band/cold-start keys round-trip)
+ const result = await apiUpdateConfig('notifications', notifConfig)
+ setOriginalNotifConfig(JSON.parse(JSON.stringify(notifConfig)))
+ if (result.restart_required) notifyRestartRequired([])
+
+ // Save fires adapter config — only changed keys
+ const origFires = originalFiresConfig ? (JSON.parse(originalFiresConfig) as FiresConfig) : null
+ if (!origFires || firesConfig.digest_enabled !== origFires.digest_enabled) {
+ await saveAdapterKey('fires', 'digest_enabled', firesConfig.digest_enabled)
+ }
+ if (!origFires || JSON.stringify(firesConfig.digest_schedule) !== JSON.stringify(origFires.digest_schedule)) {
+ await saveAdapterKey('fires', 'digest_schedule', firesConfig.digest_schedule)
+ }
+ if (!origFires || firesConfig.digest_timezone !== origFires.digest_timezone) {
+ await saveAdapterKey('fires', 'digest_timezone', firesConfig.digest_timezone)
+ }
+ setOriginalFiresConfig(JSON.stringify(firesConfig))
+
+ setHasChanges(false)
+ setDirty(false)
+ setSuccess('Scheduled broadcasts saved successfully')
+ setTimeout(() => setSuccess(null), 3000)
+ } catch (err) {
+ setError(err instanceof Error ? err.message : 'Save failed')
+ } finally {
+ setSaving(false)
+ }
+ }
+
+ const discardChanges = () => {
+ if (originalNotifConfig) setNotifConfig(JSON.parse(JSON.stringify(originalNotifConfig)))
+ if (originalFiresConfig) setFiresConfig(JSON.parse(originalFiresConfig))
+ setHasChanges(false)
+ }
+
+ const subtitle = family === 'meshcore'
+ ? 'MeshCore scheduled broadcasts and band condition reports.'
+ : 'Meshtastic scheduled broadcasts and band condition reports.'
+
+ if (loading) {
+ return (
+
+
Loading scheduled broadcasts...
+
+
Failed to load config
+
+ {/* Header / save bar */}
+
+
+
+
+
+
+
+
+
+
+
+
+ {/* Status messages */}
+ {error && (
+
{error}
+ )}
+ {success && (
+
+
+ )}
+
+ {/* Cold-start grace */}
+
+
+ Cold-start grace
+
+
setNotifConfig({ ...notifConfig, cold_start_grace_seconds: v })}
+ min={0}
+ max={600}
+ helper="Suppress broadcasts for this many seconds after the first event arrives"
+ info="When meshai starts seeing events for the first time, suppress mesh broadcasts for this many seconds to absorb any JetStream backlog. Persistence rows still get written; only broadcasts are suppressed."
+ />
+
+
+ {/* Band Conditions */}
+
+
+ Band Conditions (HF propagation)
+
+
setNotifConfig({ ...notifConfig, band_conditions_enabled: v })}
+ helper="3x/day HF propagation summary (Day/Night ratings per band group). The daily fire digest (twice-daily LLM summary of active fires + the last 24h of growth/spotting) is configured separately under Adapter Config -> fires.digest_*. See Reference -> Fire Tracker (Fusion) and Reference -> Broadcast Types for the New/Update/Active prefix system."
+ info="Source priority: (1) recent SWPC readings persisted locally; (2) HamQSL.com fallback; (3) silent skip if both fail. Persistence rows are written either way for an audit trail."
+ />
+ {(notifConfig.band_conditions_enabled ?? true) && (
+
+ {
+ const s = [...(notifConfig.band_conditions_schedule ?? ['06:00', '14:00', '22:00'])]
+ s[0] = v
+ setNotifConfig({ ...notifConfig, band_conditions_schedule: s })
+ }}
+ helper="Morning (default 06:00 MT)"
+ />
+ {
+ const s = [...(notifConfig.band_conditions_schedule ?? ['06:00', '14:00', '22:00'])]
+ s[1] = v
+ setNotifConfig({ ...notifConfig, band_conditions_schedule: s })
+ }}
+ helper="Afternoon (default 14:00 MT)"
+ />
+ {
+ const s = [...(notifConfig.band_conditions_schedule ?? ['06:00', '14:00', '22:00'])]
+ s[2] = v
+ setNotifConfig({ ...notifConfig, band_conditions_schedule: s })
+ }}
+ helper="Night (default 22:00 MT)"
+ />
+
+ )}
+ All times are Mountain Time (America/Boise). DST handled automatically.
+
+
+ {/* Fire Digest */}
+
+
+
+ Fire Digest
+
+
+
setFiresConfig({ ...firesConfig, digest_enabled: v })}
+ helper="Send a twice-daily digest of active fire conditions to the mesh"
+ />
+ {firesConfig.digest_enabled && (
+
+ {
+ const s = [...(firesConfig.digest_schedule ?? ['06:00', '18:00'])]
+ s[0] = v
+ setFiresConfig({ ...firesConfig, digest_schedule: s })
+ }}
+ helper="Morning digest (default 06:00 MT)"
+ />
+ {
+ const s = [...(firesConfig.digest_schedule ?? ['06:00', '18:00'])]
+ s[1] = v
+ setFiresConfig({ ...firesConfig, digest_schedule: s })
+ }}
+ helper="Evening digest (default 18:00 MT)"
+ />
+
+ )}
+
+
Timezone
+
setFiresConfig({ ...firesConfig, digest_timezone: e.target.value })}
+ placeholder="America/Boise"
+ className="w-full px-3 py-2 bg-[#0a0e17] border border-[#1e2a3a] rounded text-sm text-slate-200 focus:outline-none focus:border-accent"
+ />
+
IANA timezone name (e.g. America/Boise). DST handled automatically.
+
+
[time] [day] [scope]"
- aliases = ["subscribe"]
-
- def __init__(
- self,
- subscription_manager: "SubscriptionManager" = None,
- mesh_reporter: "MeshReporter" = None,
- data_store: "MeshDataStore" = None,
- notification_router: "NotificationRouter" = None,
- ):
- self._sub_manager = subscription_manager
- self._reporter = mesh_reporter
- self._data_store = data_store
- self._notification_router = notification_router
-
- async def execute(self, args: str, context: CommandContext) -> str:
- """Handle subscription command."""
- parts = args.strip().split()
-
- # No args - show available alert categories
- if not parts:
- return self._show_categories()
-
- sub_type = parts[0].lower()
-
- # Check if it's a category subscription
- if self._notification_router:
- from ..notifications.categories import ALERT_CATEGORIES
- if sub_type in ALERT_CATEGORIES or sub_type == "all":
- return self._handle_category_subscription(sub_type, context)
-
- # Legacy subscription types
- if sub_type not in ("daily", "weekly", "alerts"):
- return self._show_categories()
-
- if not self._sub_manager:
- return "Subscriptions not available."
-
- try:
- if sub_type == "daily":
- return self._handle_daily(parts[1:], context)
- elif sub_type == "weekly":
- return self._handle_weekly(parts[1:], context)
- else: # alerts
- return self._handle_alerts(parts[1:], context)
- except ValueError as e:
- return f"Error: {e}"
-
- def _show_categories(self) -> str:
- """Show available alert categories."""
- try:
- from ..notifications.categories import ALERT_CATEGORIES
- except ImportError:
- return self._usage_help()
-
- lines = ["Available alert categories:"]
- for cat_id, cat_info in ALERT_CATEGORIES.items():
- lines.append(f" {cat_id} - {cat_info['description']}")
- lines.append("")
- lines.append("Usage:")
- lines.append(" !sub - subscribe to a category")
- lines.append(" !sub all - subscribe to all alerts")
- lines.append(" !sub alerts - legacy mesh-wide alerts")
-
- return "\n".join(lines)
-
- def _handle_category_subscription(self, category: str, context: CommandContext) -> str:
- """Handle category-based alert subscription."""
- node_id = self._get_user_id(context)
-
- if category == "all":
- categories = [] # Empty = all categories
- else:
- categories = [category]
-
- # Add subscription via notification router
- rule_name = self._notification_router.add_mesh_subscription(
- node_id=node_id,
- categories=categories,
- )
-
- if category == "all":
- return "Subscribed to all alert categories. Use !unsub to remove."
- else:
- from ..notifications.categories import get_category
- cat_info = get_category(category)
- return f"Subscribed to {cat_info['name']} alerts. Use !unsub {category} to remove."
-
- def _usage_help(self) -> str:
- """Return usage help."""
- return """Usage:
-!sub daily 1830 - daily mesh report at 6:30 PM
-!sub daily 1830 region SCID - daily region report
-!sub weekly 0800 sun - weekly digest Sunday 8 AM
-!sub alerts - mesh-wide alerts (legacy)
-!sub - subscribe to alert category
-!sub all - subscribe to all alerts"""
-
- def _handle_daily(self, args: list, context: CommandContext) -> str:
- """Handle daily subscription."""
- if not args:
- raise ValueError("Time required. Example: !sub daily 1830")
-
- schedule_time = args[0]
- scope_type, scope_value = self._parse_scope(args[1:])
- scope_value = self._validate_scope(scope_type, scope_value)
-
- self._sub_manager.add(
- user_id=self._get_user_id(context),
- sub_type="daily",
- schedule_time=schedule_time,
- scope_type=scope_type,
- scope_value=scope_value,
- )
-
- time_fmt = self._format_time(schedule_time)
- scope_desc = self._format_scope(scope_type, scope_value)
- return f"Subscribed: daily {scope_desc}report at {time_fmt}"
-
- def _handle_weekly(self, args: list, context: CommandContext) -> str:
- """Handle weekly subscription."""
- if len(args) < 2:
- raise ValueError("Time and day required. Example: !sub weekly 0800 sun")
-
- schedule_time = args[0]
- schedule_day = args[1].lower()
- scope_type, scope_value = self._parse_scope(args[2:])
- scope_value = self._validate_scope(scope_type, scope_value)
-
- self._sub_manager.add(
- user_id=self._get_user_id(context),
- sub_type="weekly",
- schedule_time=schedule_time,
- schedule_day=schedule_day,
- scope_type=scope_type,
- scope_value=scope_value,
- )
-
- time_fmt = self._format_time(schedule_time)
- day_fmt = schedule_day.capitalize()
- scope_desc = self._format_scope(scope_type, scope_value)
- return f"Subscribed: weekly {scope_desc}report at {time_fmt} {day_fmt}"
-
- def _handle_alerts(self, args: list, context: CommandContext) -> str:
- """Handle alerts subscription (legacy)."""
- scope_type, scope_value = self._parse_scope(args)
- scope_value = self._validate_scope(scope_type, scope_value)
-
- self._sub_manager.add(
- user_id=self._get_user_id(context),
- sub_type="alerts",
- scope_type=scope_type,
- scope_value=scope_value,
- )
-
- scope_desc = self._format_scope(scope_type, scope_value)
- return f"Subscribed: alerts for {scope_desc.strip() or 'mesh'}"
-
- def _parse_scope(self, args: list) -> tuple[str, str]:
- """Parse scope from remaining args."""
- if not args:
- return "mesh", None
-
- scope_type = "mesh"
- scope_value = None
-
- for i, arg in enumerate(args):
- arg_lower = arg.lower()
- if arg_lower == "region":
- scope_type = "region"
- scope_value = " ".join(args[i + 1:]) if i + 1 < len(args) else None
- break
- elif arg_lower == "node":
- scope_type = "node"
- scope_value = args[i + 1] if i + 1 < len(args) else None
- break
-
- return scope_type, scope_value
-
- def _validate_scope(self, scope_type: str, scope_value: str) -> str:
- """Validate and resolve scope value."""
- if scope_type == "mesh":
- return None
-
- if not scope_value:
- raise ValueError(f"Missing {scope_type} name")
-
- if scope_type == "region" and self._reporter:
- region = self._reporter._find_region(scope_value)
- if region:
- return region.name
- return scope_value
-
- if scope_type == "node" and self._reporter:
- node = self._reporter._find_node(scope_value)
- if not node:
- raise ValueError(f"Node '{scope_value}' not found")
- return node.short_name or str(node.node_num)
-
- return scope_value
-
- def _get_user_id(self, context: CommandContext) -> str:
- """Extract user ID from context."""
- sender_id = context.sender_id
- if sender_id.startswith("!"):
- return str(int(sender_id[1:], 16))
- return sender_id
-
- def _format_time(self, hhmm: str) -> str:
- """Format HHMM as readable time."""
- hours = int(hhmm[:2])
- minutes = int(hhmm[2:])
- period = "AM" if hours < 12 else "PM"
- display_hour = hours % 12 or 12
- return f"{display_hour}:{minutes:02d} {period}"
-
- def _format_scope(self, scope_type: str, scope_value: str) -> str:
- """Format scope for display."""
- if scope_type == "mesh" or not scope_value:
- return "mesh "
- return f"{scope_type} {scope_value} "
-
-
-class UnsubCommand(CommandHandler):
- """Unsubscribe from reports or alerts."""
-
- name = "unsub"
- description = "Remove subscription(s)"
- usage = "!unsub daily|weekly|alerts||all"
- aliases = ["unsubscribe"]
-
- def __init__(
- self,
- subscription_manager: "SubscriptionManager" = None,
- notification_router: "NotificationRouter" = None,
- ):
- self._sub_manager = subscription_manager
- self._notification_router = notification_router
-
- async def execute(self, args: str, context: CommandContext) -> str:
- """Handle unsubscribe command."""
- sub_type = args.strip().lower() if args else None
-
- if not sub_type:
- return "Usage: !unsub daily|weekly|alerts||all"
-
- user_id = self._get_user_id(context)
-
- # Check if it's a category unsubscription
- if self._notification_router:
- from ..notifications.categories import ALERT_CATEGORIES
- if sub_type in ALERT_CATEGORIES or sub_type == "all":
- self._notification_router.remove_mesh_subscription(user_id)
- return "Removed alert subscriptions"
-
- # Legacy subscription types
- if not self._sub_manager:
- return "Subscriptions not available."
-
- if sub_type not in ("daily", "weekly", "alerts", "all"):
- return f"Invalid type '{sub_type}'. Use: daily, weekly, alerts, , or all"
-
- removed = self._sub_manager.remove(user_id, sub_type if sub_type != "all" else None)
-
- if removed == 0:
- return "No subscriptions found to remove"
- elif sub_type == "all":
- return f"Removed all {removed} subscription(s)"
- else:
- return f"Removed {removed} {sub_type} subscription(s)"
-
- def _get_user_id(self, context: CommandContext) -> str:
- """Extract user ID from context."""
- sender_id = context.sender_id
- if sender_id.startswith("!"):
- return str(int(sender_id[1:], 16))
- return sender_id
-
-
-class MySubsCommand(CommandHandler):
- """List active subscriptions."""
-
- name = "mysubs"
- description = "List your subscriptions"
- usage = "!mysubs"
- aliases = ["subs", "subscriptions"]
-
- def __init__(
- self,
- subscription_manager: "SubscriptionManager" = None,
- notification_router: "NotificationRouter" = None,
- ):
- self._sub_manager = subscription_manager
- self._notification_router = notification_router
-
- async def execute(self, args: str, context: CommandContext) -> str:
- """List user's subscriptions."""
- user_id = self._get_user_id(context)
- lines = []
-
- # Check notification router subscriptions
- if self._notification_router:
- categories = self._notification_router.get_node_subscriptions(user_id)
- if categories:
- if categories == ["all"]:
- lines.append("Alert subscriptions: all categories")
- else:
- lines.append(f"Alert subscriptions: {', '.join(categories)}")
-
- # Check legacy subscriptions
- if self._sub_manager:
- subs = self._sub_manager.get_user_subs(user_id)
- if subs:
- if not lines:
- lines.append("Your subscriptions:")
- else:
- lines.append("\nScheduled reports:")
- for i, sub in enumerate(subs, 1):
- lines.append(f" {i}. {self._format_sub(sub)}")
-
- if not lines:
- return "No active subscriptions. Use !sub to subscribe."
-
- return "\n".join(lines)
-
- def _format_sub(self, sub: dict) -> str:
- """Format a subscription for display."""
- sub_type = sub["sub_type"]
- scope_type = sub.get("scope_type", "mesh")
- scope_value = sub.get("scope_value")
-
- scope_desc = ""
- if scope_type == "region" and scope_value:
- scope_desc = f"region {scope_value} "
- elif scope_type == "node" and scope_value:
- scope_desc = f"node {scope_value} "
-
- if sub_type == "daily":
- time_str = self._format_time(sub.get("schedule_time", "0000"))
- return f"Daily {scope_desc}report at {time_str}"
- elif sub_type == "weekly":
- time_str = self._format_time(sub.get("schedule_time", "0000"))
- day_str = (sub.get("schedule_day") or "").capitalize()
- return f"Weekly {scope_desc}report at {time_str} {day_str}"
- else:
- return f"Alerts for {scope_desc.strip() or 'mesh'}"
-
- def _format_time(self, hhmm: str) -> str:
- """Format HHMM as readable time."""
- if not hhmm or len(hhmm) != 4:
- return hhmm
- hours = int(hhmm[:2])
- minutes = int(hhmm[2:])
- period = "AM" if hours < 12 else "PM"
- display_hour = hours % 12 or 12
- return f"{display_hour}:{minutes:02d} {period}"
-
- def _get_user_id(self, context: CommandContext) -> str:
- """Extract user ID from context."""
- sender_id = context.sender_id
- if sender_id.startswith("!"):
- return str(int(sender_id[1:], 16))
- return sender_id
diff --git a/work/meshai/config.py b/work/meshai/config.py
index 846cf3d..c53f710 100644
--- a/work/meshai/config.py
+++ b/work/meshai/config.py
@@ -104,6 +104,15 @@ class ContextConfig:
max_context_items: int = 20 # Max observations injected into LLM context
+@dataclass
+class MeshCoreContextConfig:
+ """MeshCore passive-context / bot-behavior settings (MeshCore-native)."""
+ enable_passive_context: bool = True
+ observe_channels: list[str] = field(default_factory=list) # channel NAMES, empty = all
+ ignore_contacts: list[str] = field(default_factory=list) # contact names or pubkey prefixes
+ respond_to_dms: bool = True
+
+
@dataclass
class CommandsConfig:
"""Command settings."""
@@ -782,6 +791,7 @@ class Config:
history: HistoryConfig = field(default_factory=HistoryConfig)
memory: MemoryConfig = field(default_factory=MemoryConfig)
context: ContextConfig = field(default_factory=ContextConfig)
+ meshcore_context: MeshCoreContextConfig = field(default_factory=MeshCoreContextConfig)
commands: CommandsConfig = field(default_factory=CommandsConfig)
llm: LLMConfig = field(default_factory=LLMConfig)
weather: WeatherConfig = field(default_factory=WeatherConfig)
diff --git a/work/meshai/dashboard/api/alert_routes.py b/work/meshai/dashboard/api/alert_routes.py
index 77cc6d0..a377f48 100644
--- a/work/meshai/dashboard/api/alert_routes.py
+++ b/work/meshai/dashboard/api/alert_routes.py
@@ -56,31 +56,29 @@ async def get_alert_history(
}
-@router.get("/subscriptions")
-async def get_subscriptions(request: Request):
- """Get all alert subscriptions."""
- subscription_manager = getattr(request.app.state, "subscription_manager", None)
+@router.get("/activity")
+async def get_activity(
+ request: Request,
+ limit: int = Query(100, ge=1, le=500),
+):
+ """Activity Log: most recent outbound mesh broadcasts, newest first.
- if not subscription_manager:
- return []
+ Reads mesh_broadcasts_out from the persistence/migration DB (get_db) and
+ returns every column as a plain dict. Legacy rows keep NULL
+ transport/success. If the table doesn't exist yet, returns [].
+ """
+ from meshai.persistence import get_db
try:
- subs = subscription_manager.get_all_subs()
- return [
- {
- "id": sub["id"],
- "user_id": sub["user_id"],
- "sub_type": sub["sub_type"],
- "schedule_time": sub.get("schedule_time"),
- "schedule_day": sub.get("schedule_day"),
- "scope_type": sub.get("scope_type", "mesh"),
- "scope_value": sub.get("scope_value"),
- "enabled": sub.get("enabled", 1) == 1,
- }
- for sub in subs
- ]
+ conn = get_db()
+ rows = conn.execute(
+ "SELECT * FROM mesh_broadcasts_out "
+ "ORDER BY sent_at DESC, id DESC LIMIT ?",
+ (limit,),
+ ).fetchall()
except Exception:
return []
+ return [dict(r) for r in rows]
def _map_severity(alert: dict) -> str:
diff --git a/work/meshai/dashboard/api/config_routes.py b/work/meshai/dashboard/api/config_routes.py
index 2bd2797..af85c64 100644
--- a/work/meshai/dashboard/api/config_routes.py
+++ b/work/meshai/dashboard/api/config_routes.py
@@ -45,6 +45,7 @@ VALID_SECTIONS = {
"history",
"memory",
"context",
+ "meshcore_context",
"commands",
"llm",
"weather",
diff --git a/work/meshai/dashboard/server.py b/work/meshai/dashboard/server.py
index 66400f5..0f4a16c 100644
--- a/work/meshai/dashboard/server.py
+++ b/work/meshai/dashboard/server.py
@@ -119,7 +119,6 @@ async def start_dashboard(meshai_instance: "MeshAI") -> DashboardBroadcaster:
app.state.health_engine = meshai_instance.health_engine
app.state.alert_engine = getattr(meshai_instance, "alert_engine", None)
app.state.env_store = getattr(meshai_instance, "env_store", None)
- app.state.subscription_manager = meshai_instance.subscription_manager
app.state.notification_router = getattr(meshai_instance, "notification_router", None)
app.state.connector = meshai_instance.connector
app.state.bus = getattr(meshai_instance, "event_bus", None)
diff --git a/work/meshai/main.py b/work/meshai/main.py
index def9a1d..1f5d6b3 100644
--- a/work/meshai/main.py
+++ b/work/meshai/main.py
@@ -45,7 +45,6 @@ class MeshAI:
self.data_store = None # Replaces source_manager
self.health_engine = None
self.mesh_reporter = None
- self.subscription_manager = None
self.alert_engine = None
self.notification_router = None
self.event_bus = None # Notification pipeline EventBus (v0.3)
@@ -53,7 +52,6 @@ class MeshAI:
self.env_store = None # Environmental feeds store
self._central_consumer = None # Central NATS consumer (v0.4)
self._fire_pacer = None # FirePacer for rate-limited fire broadcasts
- self._last_sub_check: float = 0.0
self.router: Optional[MessageRouter] = None
self.responder: Optional[Responder] = None
self._running = False
@@ -223,12 +221,6 @@ class MeshAI:
except Exception as e:
logger.debug("Env refresh error: %s", e)
- # Check scheduled subscriptions (every 60 seconds)
- if self.subscription_manager and self.mesh_reporter:
- if time.time() - self._last_sub_check >= 60:
- await self._check_scheduled_subs()
- self._last_sub_check = time.time()
-
# Periodic cleanup
if time.time() - self._last_cleanup >= 3600:
await self.history.cleanup_expired()
@@ -326,8 +318,6 @@ class MeshAI:
if self.data_store:
await self.data_store.stop_mqtt_sources()
self.data_store.close()
- if self.subscription_manager:
- self.subscription_manager.close()
self._remove_pid()
logger.info("MeshAI stopped")
@@ -395,7 +385,10 @@ class MeshAI:
await self._load_summaries()
# Transport connector (factory derives backend from config.connection.meshcore_host)
- self.connector = build_transport(self.config.connection)
+ self.connector = build_transport(
+ self.config.connection,
+ meshcore_context=self.config.meshcore_context,
+ )
# Fit every broadcast handler's one-packet formatter to the active mesh
# transport's budget (LoRa max_chars, 140). Durable across adapter_config
@@ -494,22 +487,13 @@ class MeshAI:
else:
self.mesh_reporter = None
- # Subscription manager (uses same db as data_store)
- if self.data_store:
- from .subscriptions import SubscriptionManager
- self.subscription_manager = SubscriptionManager(db_path="/data/mesh_history.db")
- logger.info("Subscription manager enabled")
- else:
- self.subscription_manager = None
-
- # Alert engine (needs health engine, reporter, and subscription manager)
- if self.health_engine and self.mesh_reporter and self.subscription_manager:
+ # Alert engine (needs health engine and reporter)
+ if self.health_engine and self.mesh_reporter:
from .alert_engine import AlertEngine
mi = self.config.mesh_intelligence
self.alert_engine = AlertEngine(
health_engine=self.health_engine,
reporter=self.mesh_reporter,
- subscription_manager=self.subscription_manager,
config=mi,
db_path="/data/mesh_history.db",
timezone=self.config.timezone,
@@ -610,9 +594,7 @@ class MeshAI:
mesh_reporter=self.mesh_reporter,
data_store=self.data_store,
health_engine=self.health_engine,
- subscription_manager=self.subscription_manager,
env_store=self.env_store,
- notification_router=self.notification_router,
)
# Message router
@@ -796,94 +778,9 @@ class MeshAI:
except Exception as e:
logger.error(f"Failed to send channel alert: {e}")
- # Fallback: Send DMs to matching subscribers
- if self.alert_engine and self.subscription_manager:
- subscribers = self.alert_engine.get_subscribers_for_alert(alert)
- for sub in subscribers:
- user_id = sub["user_id"]
- try:
- await self._send_sub_dm(user_id, message)
- logger.info(f"Alert DM sent to {user_id}: {alert['type']}")
- except Exception as e:
- logger.error(f"Failed to send alert DM to {user_id}: {e}")
-
if self.alert_engine:
self.alert_engine.clear_pending()
- async def _check_scheduled_subs(self) -> None:
- """Check for and deliver due scheduled reports."""
- from datetime import datetime
- from zoneinfo import ZoneInfo
-
- tz = ZoneInfo(self.config.timezone)
- now = datetime.now(tz)
- current_hhmm = now.strftime("%H%M")
- current_day = now.strftime("%a").lower()
-
- due_subs = self.subscription_manager.get_due_subscriptions(current_hhmm, current_day)
-
- for sub in due_subs:
- try:
- # Generate report based on scope
- report = self._generate_sub_report(sub)
- if not report:
- continue
-
- # Send DM to subscriber
- user_id = sub["user_id"]
- await self._send_sub_dm(user_id, report)
-
- # Mark as sent
- self.subscription_manager.mark_sent(sub["id"])
- logger.info(f"Delivered {sub['sub_type']} report to {user_id}")
-
- except Exception as e:
- logger.error(f"Error delivering subscription {sub['id']}: {e}")
-
- def _generate_sub_report(self, sub: dict) -> str:
- """Generate report content for a subscription."""
- if not self.mesh_reporter:
- return None
-
- sub_type = sub["sub_type"]
- scope_type = sub.get("scope_type", "mesh")
- scope_value = sub.get("scope_value")
-
- if scope_type == "region" and scope_value:
- # Region-scoped report
- region = self.mesh_reporter._find_region(scope_value)
- if region:
- return self.mesh_reporter.build_region_compact(region.name)
- return None
- elif scope_type == "node" and scope_value:
- # Node-scoped report
- return self.mesh_reporter.build_node_compact(scope_value)
- else:
- # Mesh-wide report
- return self.mesh_reporter.build_lora_compact(scope="mesh")
-
- async def _send_sub_dm(self, node_num: str, message: str) -> None:
- """Send a subscription DM to a node."""
- if not self.connector:
- return
-
- # Convert node_num to destination format
- try:
- dest = int(node_num)
- except ValueError:
- dest = node_num
-
- # Send via responder for proper chunking
- if self.responder:
- await self.responder.send_response(
- message,
- destination=dest,
- channel=0, # DM channel
- )
- else:
- # Fallback to direct send
- self.connector.send_message(message, destination=dest)
-
def setup_logging(verbose: bool = False) -> None:
"""Configure logging."""
diff --git a/work/meshai/notifications/pipeline/dispatcher.py b/work/meshai/notifications/pipeline/dispatcher.py
index 3569087..ba28bba 100644
--- a/work/meshai/notifications/pipeline/dispatcher.py
+++ b/work/meshai/notifications/pipeline/dispatcher.py
@@ -445,6 +445,8 @@ class Dispatcher:
delivered_any = False
for ch_type in ch_types:
+ rule = None
+ payload = None
try:
rule = self._toggle_to_rule(tog, ch_type, event)
channel = self._channel_factory(rule, self._connector)
@@ -458,15 +460,20 @@ class Dispatcher:
if success:
delivered_any = True
self._logger.info(f"Dispatched event {event.id} via toggle {fam}/{ch_type}")
- # v0.5.8b post-broadcast commit. Persistence-side
- # bookkeeping that should only happen when a delivery
- # actually went out: mesh_broadcasts_out audit row +
- # handler-supplied last_broadcast_* UPDATE callback.
- self._post_broadcast_commit(event, payload, rule, ch_type)
else:
self._logger.warning(f"Toggle channel delivery returned False for {fam}/{ch_type}")
+ # v0.5.8b post-broadcast commit -> v20 per-mesh audit.
+ # Written ONCE PER MESH CHANNEL with its own transport+success,
+ # so a fan-out to both meshes yields two rows and a skip
+ # (deliver()==False) is still visible as success=0. The
+ # last_broadcast_* callback fires only when success is truthy.
+ self._post_broadcast_commit(event, payload, rule, ch_type,
+ success=bool(success))
except Exception:
self._logger.exception(f"Toggle channel delivery failed for {fam}/{ch_type}")
+ # A crashed delivery is still a failed send -> success=0 row.
+ self._post_broadcast_commit(event, payload, rule, ch_type,
+ success=False)
# ---------- Section 6 — guard commit (v0.6-4, B13 fix) ----------
# Cooldown arming + dedup recording happen ONLY after at least one
@@ -600,39 +607,75 @@ class Dispatcher:
success = await channel.deliver(payload, rule)
except Exception:
self._logger.exception(
- "scheduled-broadcast: delivery raised for %s; skipping", ch_type)
- continue
+ "scheduled-broadcast: delivery raised for %s", ch_type)
+ success = False
if success:
delivered_any = True
- # Audit row -- mirrors _post_broadcast_commit for scheduled.
- try:
- from meshai.persistence import get_db
- conn = get_db()
- bytes_sent = len(text.encode("utf-8")) if text else 0
- conn.execute(
- "INSERT INTO mesh_broadcasts_out(sent_at, recipient, "
- "channel, text, source_event_table, source_event_pk, "
- "bytes_sent, ack_received) VALUES (?,?,?,?,?,?,?,?)",
- (int(time.time()), "broadcast",
- rf.broadcast_channel, text,
- source_event_table, str(source_event_pk),
- bytes_sent, 0),
- )
- except Exception:
- self._logger.exception(
- "scheduled-broadcast: audit row insert failed for %s", ch_type)
+
+ # v20 per-mesh audit row. Written once per mesh channel with its
+ # own transport+success, so a fan-out to both meshes yields two
+ # rows and a skip (deliver()==False) is visible as success=0.
+ try:
+ from meshai.persistence import get_db
+ conn = get_db()
+ bytes_sent = len(text.encode("utf-8")) if text else 0
+ 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,
+ source_event_table, str(source_event_pk),
+ bytes_sent, 0,
+ transport, 1 if success else 0),
+ )
+ except Exception:
+ self._logger.exception(
+ "scheduled-broadcast: audit row insert failed for %s", ch_type)
return delivered_any
- def _post_broadcast_commit(self, event, payload, rule, ch_type: str) -> None:
- """Persistence side-effects of an actually-successful broadcast.
+ @staticmethod
+ def _audit_route(rule, ch_type: str):
+ """Resolve (transport, channel_id, recipient) for a mesh delivery.
- Inserts the mesh_broadcasts_out audit row when the handler signalled
- it wants one via `event.data["_broadcast_audit"]`, then invokes the
- handler-supplied `_on_broadcast_committed` callback so the handler
- can refresh its own last_broadcast_* bookkeeping. Both calls are
- wrapped: a bookkeeping failure must NOT undo the actual broadcast
- nor break dispatch for sibling toggles.
+ transport is the mesh family the row belongs to ("meshtastic" /
+ "meshcore"); channel_id is the Meshtastic channel INDEX or the
+ MeshCore channel NAME; recipient is 'broadcast' or the DM target
+ list. Mirrors create_channel()'s delivery_type routing.
+ """
+ if ch_type == "mesh_broadcast":
+ return "meshtastic", getattr(rule, "broadcast_channel", None), "broadcast"
+ if ch_type == "meshcore_broadcast":
+ return "meshcore", getattr(rule, "meshcore_channel", None), "broadcast"
+ if ch_type == "mesh_dm":
+ node_ids = list(getattr(rule, "node_ids", []) or [])
+ return "meshtastic", None, (",".join(map(str, node_ids)) or "dm")
+ if ch_type == "meshcore_dm":
+ contacts = list(getattr(rule, "meshcore_dm_contacts", []) or [])
+ return "meshcore", None, (",".join(map(str, contacts)) or "meshcore_dm")
+ # Unknown / non-mesh: leave transport NULL, fall back to legacy channel.
+ return None, getattr(rule, "broadcast_channel", None), "broadcast"
+
+ def _post_broadcast_commit(self, event, payload, rule, ch_type: str,
+ *, success: bool = True) -> None:
+ """Persistence side-effects of a per-mesh broadcast delivery.
+
+ Called ONCE PER MESH CHANNEL (one per delivery_type family), so a
+ broadcast that fans to both meshes writes TWO mesh_broadcasts_out
+ rows -- each carrying its own `transport` + `success` flag. The row
+ is written whenever the handler signalled it wants an audit trail
+ via `event.data["_broadcast_audit"]`, REGARDLESS of success, so a
+ skip/failure (e.g. MeshCore channel-not-found -> deliver()==False)
+ is still visible as success=0.
+
+ The handler-supplied `_on_broadcast_committed` callback (which
+ refreshes last_broadcast_* bookkeeping) fires ONLY when the send
+ actually landed (success is truthy). Both calls are wrapped: a
+ bookkeeping failure must NOT undo the actual broadcast nor break
+ dispatch for sibling toggles.
"""
data = getattr(event, "data", None) or {}
if not data:
@@ -646,23 +689,17 @@ class Dispatcher:
conn = get_db()
text = payload.message if payload is not None else (event.title or "")
bytes_sent = len(text.encode("utf-8")) if text else 0
- if ch_type == "mesh_dm":
- node_ids = list(getattr(rule, "node_ids", []) or [])
- recipient = ",".join(map(str, node_ids)) or "dm"
- elif ch_type == "meshcore_dm":
- contacts = list(getattr(rule, "meshcore_dm_contacts", []) or [])
- recipient = ",".join(map(str, contacts)) or "meshcore_dm"
- else:
- recipient = "broadcast"
- channel = getattr(rule, "broadcast_channel", None)
+ transport, channel, 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) VALUES (?,?,?,?,?,?,?,?)",
+ "ack_received, transport, success) "
+ "VALUES (?,?,?,?,?,?,?,?,?,?)",
(
int(committed_at), recipient, channel, text,
audit.get("table"), audit.get("pk"),
bytes_sent, 0,
+ transport, 1 if success else 0,
),
)
except Exception:
@@ -672,6 +709,11 @@ class Dispatcher:
audit.get("table"), audit.get("pk"),
)
+ if not success:
+ # A failed/skipped send is audited above but must NOT arm the
+ # handler's last_broadcast_* bookkeeping.
+ return
+
cb = data.get("_on_broadcast_committed")
if callable(cb):
try:
diff --git a/work/meshai/notifications/router.py b/work/meshai/notifications/router.py
index 507cdf3..3b163d4 100644
--- a/work/meshai/notifications/router.py
+++ b/work/meshai/notifications/router.py
@@ -735,45 +735,6 @@ class NotificationRouter:
return {"matches": False, "conditions": [], "preview": "Unknown rule type"}
- def add_mesh_subscription(self, node_id: str, categories: list[str], rule_name: Optional[str] = None) -> str:
- """Add a mesh DM subscription for a node."""
- if not rule_name:
- rule_name = "sub_%s" % node_id
-
- for rule in self._rules:
- if rule.get("name") == rule_name:
- rule["categories"] = categories if categories else []
- rule["node_ids"] = [node_id]
- return rule_name
-
- self._rules.append({
- "name": rule_name,
- "enabled": True,
- "trigger_type": "condition",
- "categories": categories if categories else [],
- "min_severity": "priority",
- "delivery_type": "mesh_dm",
- "node_ids": [node_id],
- "cooldown_minutes": 10,
- })
-
- return rule_name
-
- def remove_mesh_subscription(self, node_id: str) -> bool:
- """Remove a mesh subscription for a node."""
- rule_name = "sub_%s" % node_id
- self._rules = [r for r in self._rules if r.get("name") != rule_name]
- return True
-
- def get_node_subscriptions(self, node_id: str) -> list[str]:
- """Get categories a node is subscribed to."""
- rule_name = "sub_%s" % node_id
- for rule in self._rules:
- if rule.get("name") == rule_name:
- categories = rule.get("categories", [])
- return categories if categories else ["all"]
- return []
-
async def generate_report(self, report_type: str, env_store, health_engine) -> str:
"""Generate an LLM-summarized report from current data."""
context_parts = []
diff --git a/work/meshai/persistence/db.py b/work/meshai/persistence/db.py
index fd977ef..82c8f14 100644
--- a/work/meshai/persistence/db.py
+++ b/work/meshai/persistence/db.py
@@ -30,7 +30,7 @@ logger = logging.getLogger(__name__)
DEFAULT_DB_PATH = "/data/meshai.sqlite"
MESHAI_DB_PATH_ENV = "MESHAI_DB_PATH"
-SCHEMA_VERSION = 19
+SCHEMA_VERSION = 20
SCHEMA_META_TABLE = "schema_meta"
MIGRATIONS_DIR = Path(__file__).parent / "migrations"
diff --git a/work/meshai/persistence/migrations/v20.sql b/work/meshai/persistence/migrations/v20.sql
new file mode 100644
index 0000000..2185818
--- /dev/null
+++ b/work/meshai/persistence/migrations/v20.sql
@@ -0,0 +1,9 @@
+-- v20: per-mesh broadcast audit. transport + success columns on
+-- mesh_broadcasts_out so each SEND records which mesh it went to and
+-- whether it landed. A broadcast fanning to BOTH meshes writes one row
+-- per mesh, each with its own success flag.
+-- Nullable, no index, no backfill. Legacy rows keep NULL transport/success
+-- (Activity Log treats NULL as legacy/meshtastic-unknown).
+
+ALTER TABLE mesh_broadcasts_out ADD COLUMN transport TEXT;
+ALTER TABLE mesh_broadcasts_out ADD COLUMN success INTEGER;
diff --git a/work/meshai/subscriptions.py b/work/meshai/subscriptions.py
deleted file mode 100644
index 1695c70..0000000
--- a/work/meshai/subscriptions.py
+++ /dev/null
@@ -1,278 +0,0 @@
-"""Subscription management for scheduled reports and alerts."""
-
-import logging
-import sqlite3
-import time
-from typing import Optional
-
-logger = logging.getLogger(__name__)
-
-# Valid subscription types
-VALID_SUB_TYPES = {"daily", "weekly", "alerts"}
-VALID_DAYS = {"mon", "tue", "wed", "thu", "fri", "sat", "sun"}
-VALID_SCOPE_TYPES = {"mesh", "region", "node"}
-
-
-class SubscriptionManager:
- """Manages user subscriptions with SQLite storage."""
-
- def __init__(self, db_path: str):
- """Initialize subscription manager.
-
- Args:
- db_path: Path to SQLite database (same as mesh_history.db)
- """
- self._db_path = db_path
- self._db: Optional[sqlite3.Connection] = None
- self._init_db()
-
- def _init_db(self):
- """Initialize database connection and schema."""
- self._db = sqlite3.connect(self._db_path, check_same_thread=False)
- self._db.row_factory = sqlite3.Row
-
- self._db.executescript("""
- CREATE TABLE IF NOT EXISTS subscriptions (
- id INTEGER PRIMARY KEY AUTOINCREMENT,
- user_id TEXT NOT NULL,
- sub_type TEXT NOT NULL,
- schedule_time TEXT,
- schedule_day TEXT,
- scope_type TEXT DEFAULT 'mesh',
- scope_value TEXT,
- created_at REAL NOT NULL,
- last_sent REAL DEFAULT 0,
- enabled INTEGER DEFAULT 1
- );
- CREATE INDEX IF NOT EXISTS idx_sub_user ON subscriptions(user_id);
- CREATE INDEX IF NOT EXISTS idx_sub_type ON subscriptions(sub_type);
- """)
- self._db.commit()
- logger.info("Subscription manager initialized")
-
- def _row_to_dict(self, row: sqlite3.Row) -> dict:
- """Convert sqlite Row to dict."""
- return dict(row)
-
- def add(self, user_id: str, sub_type: str, schedule_time: str = None,
- schedule_day: str = None, scope_type: str = "mesh",
- scope_value: str = None) -> dict:
- """Add a subscription.
-
- Args:
- user_id: Subscriber node_num
- sub_type: "daily", "weekly", or "alerts"
- schedule_time: HHMM format (required for daily/weekly)
- schedule_day: mon-sun (required for weekly)
- scope_type: "mesh", "region", or "node"
- scope_value: Region name or node identifier
-
- Returns:
- Created subscription dict
-
- Raises:
- ValueError: If validation fails
- """
- # Validate sub_type
- if sub_type not in VALID_SUB_TYPES:
- raise ValueError(f"Invalid type '{sub_type}'. Use: daily, weekly, or alerts")
-
- # Validate schedule_time for daily/weekly
- if sub_type in ("daily", "weekly"):
- if not schedule_time:
- raise ValueError(f"Time required for {sub_type} subscription. Use HHMM format (e.g., 1830)")
- if not self._validate_time(schedule_time):
- raise ValueError("Invalid time format. Use HHMM (e.g., 1830 for 6:30 PM)")
-
- # Validate schedule_day for weekly
- if sub_type == "weekly":
- if not schedule_day:
- raise ValueError("Day required for weekly subscription. Use: mon, tue, wed, thu, fri, sat, sun")
- if schedule_day.lower() not in VALID_DAYS:
- raise ValueError("Invalid day. Use: mon, tue, wed, thu, fri, sat, sun")
- schedule_day = schedule_day.lower()
-
- # Validate scope_type
- if scope_type not in VALID_SCOPE_TYPES:
- raise ValueError(f"Invalid scope '{scope_type}'. Use: mesh, region, or node")
-
- # Check for duplicates
- existing = self._db.execute("""
- SELECT id FROM subscriptions
- WHERE user_id = ? AND sub_type = ? AND scope_type = ?
- AND (scope_value = ? OR (scope_value IS NULL AND ? IS NULL))
- AND enabled = 1
- """, (user_id, sub_type, scope_type, scope_value, scope_value)).fetchone()
-
- if existing:
- scope_desc = f" for {scope_type} {scope_value}" if scope_value else ""
- raise ValueError(f"Already subscribed to {sub_type}{scope_desc}")
-
- # Insert subscription
- now = time.time()
- cursor = self._db.execute("""
- INSERT INTO subscriptions (user_id, sub_type, schedule_time, schedule_day,
- scope_type, scope_value, created_at)
- VALUES (?, ?, ?, ?, ?, ?, ?)
- """, (user_id, sub_type, schedule_time, schedule_day, scope_type, scope_value, now))
- self._db.commit()
-
- sub_id = cursor.lastrowid
- return self._get_by_id(sub_id)
-
- def _validate_time(self, time_str: str) -> bool:
- """Validate HHMM time format."""
- if not time_str or len(time_str) != 4 or not time_str.isdigit():
- return False
- hours = int(time_str[:2])
- minutes = int(time_str[2:])
- return 0 <= hours <= 23 and 0 <= minutes <= 59
-
- def _get_by_id(self, sub_id: int) -> dict:
- """Get subscription by ID."""
- row = self._db.execute(
- "SELECT * FROM subscriptions WHERE id = ?", (sub_id,)
- ).fetchone()
- return self._row_to_dict(row) if row else None
-
- def remove(self, user_id: str, sub_type: str = None) -> int:
- """Remove subscription(s).
-
- Args:
- user_id: Subscriber node_num
- sub_type: "daily", "weekly", "alerts", or None for all
-
- Returns:
- Number of subscriptions removed
- """
- if sub_type and sub_type != "all":
- cursor = self._db.execute(
- "DELETE FROM subscriptions WHERE user_id = ? AND sub_type = ?",
- (user_id, sub_type)
- )
- else:
- cursor = self._db.execute(
- "DELETE FROM subscriptions WHERE user_id = ?",
- (user_id,)
- )
- self._db.commit()
- return cursor.rowcount
-
- def get_user_subs(self, user_id: str) -> list[dict]:
- """Get all subscriptions for a user."""
- rows = self._db.execute(
- "SELECT * FROM subscriptions WHERE user_id = ? AND enabled = 1 ORDER BY created_at",
- (user_id,)
- ).fetchall()
- return [self._row_to_dict(r) for r in rows]
-
- def get_due_subscriptions(self, current_time_hhmm: str, current_day: str) -> list[dict]:
- """Get subscriptions that should fire right now.
-
- Args:
- current_time_hhmm: Current time as "HHMM" (e.g., "1830")
- current_day: Current day as 3-letter lowercase (e.g., "sun")
-
- Returns:
- List of subscription dicts that are due
- """
- now = time.time()
- due = []
-
- # Get all daily/weekly subscriptions
- rows = self._db.execute("""
- SELECT * FROM subscriptions
- WHERE sub_type IN ('daily', 'weekly') AND enabled = 1
- """).fetchall()
-
- current_minutes = int(current_time_hhmm[:2]) * 60 + int(current_time_hhmm[2:])
-
- for row in rows:
- sub = self._row_to_dict(row)
- schedule_time = sub.get("schedule_time")
- if not schedule_time:
- continue
-
- schedule_minutes = int(schedule_time[:2]) * 60 + int(schedule_time[2:])
-
- # 5-minute matching window
- if abs(schedule_minutes - current_minutes) > 5:
- continue
-
- sub_type = sub["sub_type"]
- last_sent = sub.get("last_sent", 0) or 0
-
- if sub_type == "daily":
- # Don't fire if sent within last 23 hours
- if now - last_sent < 23 * 3600:
- continue
- due.append(sub)
-
- elif sub_type == "weekly":
- # Check day matches
- schedule_day = sub.get("schedule_day", "").lower()
- if schedule_day != current_day.lower():
- continue
- # Don't fire if sent within last 6 days
- if now - last_sent < 6 * 24 * 3600:
- continue
- due.append(sub)
-
- return due
-
- def get_alert_subscribers(self, scope_type: str = None, scope_value: str = None) -> list[dict]:
- """Get users subscribed to alerts matching a scope.
-
- Args:
- scope_type: "mesh", "region", or "node"
- scope_value: Region name or node identifier
-
- Returns:
- List of subscription dicts where scope matches
- """
- # Get all alert subscriptions
- rows = self._db.execute("""
- SELECT * FROM subscriptions
- WHERE sub_type = 'alerts' AND enabled = 1
- """).fetchall()
-
- matching = []
- for row in rows:
- sub = self._row_to_dict(row)
- sub_scope = sub.get("scope_type", "mesh")
- sub_value = sub.get("scope_value")
-
- # Mesh scope gets ALL alerts
- if sub_scope == "mesh":
- matching.append(sub)
- # Region scope gets alerts for that region
- elif sub_scope == "region" and scope_type == "region":
- if sub_value and scope_value and sub_value.lower() == scope_value.lower():
- matching.append(sub)
- # Node scope gets alerts for that node
- elif sub_scope == "node" and scope_type == "node":
- if sub_value and scope_value and sub_value.lower() == scope_value.lower():
- matching.append(sub)
-
- return matching
-
- def mark_sent(self, subscription_id: int):
- """Update last_sent timestamp to now."""
- self._db.execute(
- "UPDATE subscriptions SET last_sent = ? WHERE id = ?",
- (time.time(), subscription_id)
- )
- self._db.commit()
-
- def get_all_subs(self) -> list[dict]:
- """Get all subscriptions (for admin view)."""
- rows = self._db.execute(
- "SELECT * FROM subscriptions WHERE enabled = 1 ORDER BY user_id, created_at"
- ).fetchall()
- return [self._row_to_dict(r) for r in rows]
-
- def close(self):
- """Close database connection."""
- if self._db:
- self._db.close()
- self._db = None
diff --git a/work/meshai/transport/factory.py b/work/meshai/transport/factory.py
index f0adb27..b054e40 100644
--- a/work/meshai/transport/factory.py
+++ b/work/meshai/transport/factory.py
@@ -3,7 +3,7 @@
from .base import MeshTransport
-def build_transport(config) -> MeshTransport:
+def build_transport(config, meshcore_context=None) -> MeshTransport:
"""Instantiate and return the active MeshTransport derived from config.
The active transports are derived from the connection config, not a
@@ -16,6 +16,8 @@ def build_transport(config) -> MeshTransport:
Args:
config: A ConnectionConfig (or duck-compatible object).
+ meshcore_context: Optional MeshCoreContextConfig for the MeshCore
+ passive-context / bot-behavior filter (None = pass-through).
Returns:
A concrete MeshTransport instance ready to be connected.
@@ -27,6 +29,9 @@ def build_transport(config) -> MeshTransport:
if meshcore_host.strip():
from meshai.transport.meshcore_transport import MeshCoreTransport
from meshai.transport.composite_transport import CompositeTransport
- return CompositeTransport([meshtastic, MeshCoreTransport(config)], config=config)
+ return CompositeTransport(
+ [meshtastic, MeshCoreTransport(config, meshcore_context=meshcore_context)],
+ config=config,
+ )
return meshtastic
diff --git a/work/meshai/transport/meshcore_transport.py b/work/meshai/transport/meshcore_transport.py
index adcccd6..f1db82b 100644
--- a/work/meshai/transport/meshcore_transport.py
+++ b/work/meshai/transport/meshcore_transport.py
@@ -24,6 +24,30 @@ logger = logging.getLogger(__name__)
_COMMAND_TIMEOUT = 10.0
+def mc_context_allows(cfg, msg, idx_to_name):
+ """Return True if a MeshCore inbound MeshMessage should be forwarded.
+
+ cfg: MeshCoreContextConfig or None. idx_to_name: dict[int,str] channel-idx->name.
+ """
+ if cfg is None:
+ return True
+ if msg.is_dm:
+ if not cfg.respond_to_dms:
+ return False
+ # ignore_contacts matches the pubkey prefix (sender_id) OR the contact name (sender_name)
+ if msg.sender_id in cfg.ignore_contacts or msg.sender_name in cfg.ignore_contacts:
+ return False
+ return True
+ # channel (non-DM) message -> only relevant for passive context
+ if not cfg.enable_passive_context:
+ return False
+ if cfg.observe_channels:
+ name = idx_to_name.get(msg.channel)
+ if name is None or name not in cfg.observe_channels:
+ return False
+ return True
+
+
class MeshCoreTransport(MeshTransport):
"""MeshTransport implementation over a pyMC companion TCP frame server.
@@ -41,8 +65,12 @@ class MeshCoreTransport(MeshTransport):
# Name tag used by CompositeTransport for routing hints.
transport_name: str = "meshcore"
- def __init__(self, config) -> None:
+ def __init__(self, config, meshcore_context=None) -> None:
self.config = config
+ # MeshCore passive-context / bot-behavior filter (MeshCoreContextConfig
+ # or None). None = pass-through (no filtering). Injected at construction
+ # time by the factory; can also be (re)set via set_context_config().
+ self._mc_context = meshcore_context
self._mc = None # meshcore.MeshCore instance
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._loop_thread: Optional[threading.Thread] = None
@@ -142,6 +170,13 @@ class MeshCoreTransport(MeshTransport):
"""Enumerated MeshCore channel names (from _chan_name_to_idx, populated at connect)."""
return list(self._chan_name_to_idx.keys())
+ def set_context_config(self, cfg) -> None:
+ """Set (or clear) the MeshCore passive-context filter config.
+
+ cfg: MeshCoreContextConfig or None (None = pass-through).
+ """
+ self._mc_context = cfg
+
# ------------------------------------------------------------------
# Internal coroutines (run on the dedicated loop)
# ------------------------------------------------------------------
@@ -416,13 +451,21 @@ class MeshCoreTransport(MeshTransport):
# ------------------------------------------------------------------
def _on_dm_event(self, event) -> None:
- """Handle CONTACT_MSG_RECV: normalize and dispatch to meshai."""
+ """Handle CONTACT_MSG_RECV: normalize, filter, and dispatch to meshai."""
msg = self._normalize_dm_event(event)
+ if msg is None or not mc_context_allows(
+ self._mc_context, msg, {v: k for k, v in self._chan_name_to_idx.items()}
+ ):
+ return
self._dispatch_message(msg)
def _on_channel_event(self, event) -> None:
- """Handle CHANNEL_MSG_RECV: normalize and dispatch to meshai."""
+ """Handle CHANNEL_MSG_RECV: normalize, filter, and dispatch to meshai."""
msg = self._normalize_channel_event(event)
+ if msg is None or not mc_context_allows(
+ self._mc_context, msg, {v: k for k, v in self._chan_name_to_idx.items()}
+ ):
+ return
self._dispatch_message(msg)
def _dispatch_message(self, msg: Optional[MeshMessage]) -> None:
diff --git a/work/tests/test_meshcore_context_config.py b/work/tests/test_meshcore_context_config.py
new file mode 100644
index 0000000..b9c1b73
--- /dev/null
+++ b/work/tests/test_meshcore_context_config.py
@@ -0,0 +1,50 @@
+"""Config round-trip tests for the MeshCore passive-context block.
+
+Verifies that ``meshcore_context`` survives save_config -> load_config, and
+that a YAML lacking the section yields the dataclass defaults (proving the
+generic nested-dataclass loader branch handles it with no special-casing).
+"""
+
+import yaml
+
+from meshai.config import (
+ Config,
+ MeshCoreContextConfig,
+ load_config,
+ save_config,
+)
+
+
+def test_meshcore_context_round_trip(tmp_path):
+ cfg = Config()
+ cfg.meshcore_context = MeshCoreContextConfig(
+ enable_passive_context=False,
+ observe_channels=["#aida", "#general"],
+ ignore_contacts=["a1b2", "SpamNode"],
+ respond_to_dms=False,
+ )
+
+ path = tmp_path / "config.yaml"
+ save_config(cfg, path)
+ loaded = load_config(path)
+
+ mc = loaded.meshcore_context
+ assert isinstance(mc, MeshCoreContextConfig)
+ assert mc.enable_passive_context is False
+ assert mc.observe_channels == ["#aida", "#general"]
+ assert mc.ignore_contacts == ["a1b2", "SpamNode"]
+ assert mc.respond_to_dms is False
+
+
+def test_meshcore_context_defaults_when_absent(tmp_path):
+ # A minimal YAML with no meshcore_context section at all.
+ path = tmp_path / "config.yaml"
+ path.write_text(yaml.safe_dump({"timezone": "America/Boise"}))
+
+ loaded = load_config(path)
+ mc = loaded.meshcore_context
+ assert isinstance(mc, MeshCoreContextConfig)
+ assert mc.enable_passive_context is True
+ assert mc.observe_channels == []
+ assert mc.ignore_contacts == []
+ assert mc.respond_to_dms is True
diff --git a/work/tests/test_meshcore_context_filter.py b/work/tests/test_meshcore_context_filter.py
new file mode 100644
index 0000000..2b48059
--- /dev/null
+++ b/work/tests/test_meshcore_context_filter.py
@@ -0,0 +1,85 @@
+"""Unit tests for the MeshCore passive-context / bot-behavior filter.
+
+Covers the pure module-level helper ``mc_context_allows`` — no hardware,
+no meshcore lib, no event loop. MeshMessage and MeshCoreContextConfig are
+constructed directly.
+"""
+
+from meshai.config import MeshCoreContextConfig
+from meshai.connector import MeshMessage
+from meshai.transport.meshcore_transport import mc_context_allows
+
+
+def _dm(sender_id="a1b2", sender_name="Alice", text="hi"):
+ return MeshMessage(
+ sender_id=sender_id,
+ sender_name=sender_name,
+ text=text,
+ channel=0,
+ is_dm=True,
+ transport="meshcore",
+ )
+
+
+def _chan(channel=1, text="hello", sender_id=None, sender_name=None):
+ marker = f"chan:{channel}"
+ return MeshMessage(
+ sender_id=sender_id or marker,
+ sender_name=sender_name or marker,
+ text=text,
+ channel=channel,
+ is_dm=False,
+ transport="meshcore",
+ )
+
+
+def test_cfg_none_always_passes():
+ assert mc_context_allows(None, _dm(), {}) is True
+ assert mc_context_allows(None, _chan(), {}) is True
+
+
+def test_defaults_pass_through():
+ cfg = MeshCoreContextConfig() # empty lists, passive on, respond_to_dms on
+ idx_to_name = {1: "#general"}
+ assert mc_context_allows(cfg, _chan(channel=1), idx_to_name) is True
+ assert mc_context_allows(cfg, _dm(), idx_to_name) is True
+
+
+def test_observe_channels_filters_by_name():
+ cfg = MeshCoreContextConfig(observe_channels=["#aida"])
+ idx_to_name = {1: "#general", 2: "#aida"}
+ # idx 1 -> #general -> not in observe list -> DROPPED
+ assert mc_context_allows(cfg, _chan(channel=1), idx_to_name) is False
+ # idx 2 -> #aida -> in observe list -> PASSES
+ assert mc_context_allows(cfg, _chan(channel=2), idx_to_name) is True
+ # idx 3 -> no name mapping -> DROPPED
+ assert mc_context_allows(cfg, _chan(channel=3), idx_to_name) is False
+
+
+def test_ignore_contacts_matches_id_or_name():
+ cfg = MeshCoreContextConfig(ignore_contacts=["a1b2"])
+ # matched by sender_id
+ assert mc_context_allows(cfg, _dm(sender_id="a1b2", sender_name="Alice"), {}) is False
+ # matched by sender_name
+ assert mc_context_allows(cfg, _dm(sender_id="ffff", sender_name="a1b2"), {}) is False
+ # unrelated DM passes
+ assert mc_context_allows(cfg, _dm(sender_id="c3d4", sender_name="Bob"), {}) is True
+
+
+def test_respond_to_dms_false_drops_dms_only():
+ cfg = MeshCoreContextConfig(respond_to_dms=False)
+ idx_to_name = {1: "#general"}
+ # any DM dropped
+ assert mc_context_allows(cfg, _dm(), idx_to_name) is False
+ assert mc_context_allows(cfg, _dm(sender_id="zzzz", sender_name="Zed"), idx_to_name) is False
+ # channel msgs unaffected (passive still on)
+ assert mc_context_allows(cfg, _chan(channel=1), idx_to_name) is True
+
+
+def test_passive_disabled_drops_channel_but_dm_still_respected():
+ cfg = MeshCoreContextConfig(enable_passive_context=False)
+ idx_to_name = {1: "#general"}
+ # channel msg dropped
+ assert mc_context_allows(cfg, _chan(channel=1), idx_to_name) is False
+ # DM still respects respond_to_dms (True by default) -> passes
+ assert mc_context_allows(cfg, _dm(), idx_to_name) is True