diff --git a/work/meshai/context.py b/work/meshai/context.py index 138dac4..8a02a48 100644 --- a/work/meshai/context.py +++ b/work/meshai/context.py @@ -32,6 +32,14 @@ class MeshContext: Passively observes all mesh messages (channels, DMs, BBS notifications) and makes them available as context when generating LLM responses. Observations older than max_age are pruned periodically. + + Durability: every observe() writes through to the mesh_observations + SQLite table (meshai.sqlite) in addition to the in-memory deque, and + __init__ loads recent rows back from SQLite so recap context survives + a container restart. The SQLite path is entirely fail-safe -- any DB + error is logged and swallowed so it never blocks or drops the + in-memory observation path (which is what response generation reads + from on the hot path). """ def __init__( @@ -52,6 +60,37 @@ class MeshContext: self._ignore_nodes = set(ignore_nodes) if ignore_nodes else set() self._max_age = max_age + self._load_from_db() + + def _load_from_db(self) -> None: + """Load recent observations from SQLite into the deque on startup. + + Fail-safe: any DB error is logged and swallowed -- an empty deque + (today's pre-fix behavior) is an acceptable fallback, never a + startup failure. + """ + try: + from .persistence.mesh_observations import load_recent, prune_older_than + + prune_older_than(self._max_age) + rows = load_recent(self._max_age, _HARD_CAP) + for row in rows: + self._buffer.append( + MeshObservation( + timestamp=row["ts"], + sender_name=row["sender_name"] or "", + sender_id=row["sender_id"] or "", + channel=row["channel"] if row["channel"] is not None else 0, + is_dm=bool(row["is_dm"]), + text=row["text"] or "", + transport=row["transport"] or "meshtastic", + ) + ) + if rows: + logger.info(f"Loaded {len(rows)} mesh observations from durable store") + except Exception: + logger.exception("Failed to load mesh observations from durable store") + def observe( self, sender_name: str, @@ -95,6 +134,24 @@ class MeshContext: self._buffer.append(obs) logger.debug(f"Observed: ch{channel} {sender_name} [{transport}]: {text[:40]}...") + # Write-through to durable store. Fail-safe: never let a DB error + # break ingestion -- the in-memory buffer above already has the + # observation regardless of what happens here. + try: + from .persistence.mesh_observations import insert_observation + + insert_observation( + ts=obs.timestamp, + transport=obs.transport, + channel=obs.channel, + is_dm=obs.is_dm, + sender_name=obs.sender_name, + sender_id=obs.sender_id, + text=obs.text, + ) + except Exception: + logger.exception("Failed to write mesh observation to durable store") + def prune(self) -> int: """Remove observations older than max_age. @@ -113,6 +170,16 @@ class MeshContext: pruned = before - len(self._buffer) if pruned > 0: logger.info(f"Pruned {pruned} expired mesh observations ({len(self._buffer)} remaining)") + + # Durable-store prune rides the same cadence as the in-memory prune + # (called hourly from main.py's periodic cleanup). Fail-safe. + try: + from .persistence.mesh_observations import prune_older_than + + prune_older_than(self._max_age) + except Exception: + logger.exception("Failed to prune durable mesh observations store") + return pruned def get_context_block(self, max_items: int = 20, transport: Optional[str] = None) -> str: diff --git a/work/meshai/persistence/mesh_observations.py b/work/meshai/persistence/mesh_observations.py new file mode 100644 index 0000000..1f9d630 --- /dev/null +++ b/work/meshai/persistence/mesh_observations.py @@ -0,0 +1,78 @@ +"""mesh_observations accessors (v28). + +Durable backing store for MeshContext's passive mesh-traffic buffer. +Follows the same pattern as persistence/observer_locations.py: a migration +creates the table, MeshContext writes through to it on every observe() call +and loads recent rows back into its in-memory deque on startup so recap +context survives a container restart. + +Fail-safe by design: every function here is expected to be called from +MeshContext, which wraps each call in try/except so a DB hiccup never +drops the in-memory (deque) observation path. +""" +from __future__ import annotations + +import logging +import sqlite3 +import time +from typing import Optional + +logger = logging.getLogger(__name__) + + +def insert_observation( + ts: float, + transport: str, + channel: Optional[int], + is_dm: bool, + sender_name: str, + sender_id: str, + text: str, + conn: Optional[sqlite3.Connection] = None, +) -> None: + """Write one observation through to SQLite.""" + if conn is None: + from meshai.persistence import get_db + conn = get_db() + conn.execute( + "INSERT INTO mesh_observations" + "(ts, transport, channel, is_dm, sender_name, sender_id, text) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", + (ts, transport, channel, 1 if is_dm else 0, sender_name, sender_id, text), + ) + + +def load_recent( + max_age: int, + limit: int, + conn: Optional[sqlite3.Connection] = None, +) -> list[dict]: + """Return up to `limit` observations newer than `max_age` seconds ago, + oldest first (chronological -- ready to feed straight into the deque). + """ + if conn is None: + from meshai.persistence import get_db + conn = get_db() + cutoff = time.time() - max_age + rows = conn.execute( + "SELECT ts, transport, channel, is_dm, sender_name, sender_id, text " + "FROM mesh_observations WHERE ts >= ? " + "ORDER BY ts DESC LIMIT ?", + (cutoff, limit), + ).fetchall() + out = [dict(r) for r in rows] + out.reverse() # newest-first -> chronological + return out + + +def prune_older_than( + max_age: int, + conn: Optional[sqlite3.Connection] = None, +) -> int: + """Delete rows older than max_age seconds. Returns rows deleted.""" + if conn is None: + from meshai.persistence import get_db + conn = get_db() + cutoff = time.time() - max_age + cur = conn.execute("DELETE FROM mesh_observations WHERE ts < ?", (cutoff,)) + return cur.rowcount if cur.rowcount is not None else 0 diff --git a/work/meshai/persistence/migrations/v28.sql b/work/meshai/persistence/migrations/v28.sql new file mode 100644 index 0000000..a6a40d0 --- /dev/null +++ b/work/meshai/persistence/migrations/v28.sql @@ -0,0 +1,27 @@ +-- v28 durable retention for the passive mesh-context buffer (MeshContext). +-- +-- MeshContext._buffer is an in-memory deque(maxlen=50000) that observes all +-- mesh traffic (channel broadcasts, DMs, BBS notifications) for LLM recap +-- context -- but it was wiped on every container restart, so recap requests +-- right after a restart got an empty buffer even though max_age keeps +-- observations "live" for up to config.context.max_age seconds (14d +-- default). This table is the write-through / load-on-boot backing store so +-- observations survive restarts, following the same durable-store pattern +-- as generic_events (v26) and satpass_events. +-- +-- One row per observed message. transport distinguishes meshtastic vs +-- meshcore (MeshContext.get_context_block filters by it). is_dm stored as +-- 0/1 (SQLite has no native bool). Indexed on ts for the age-cutoff prune +-- query and the load-on-boot "most recent N within max_age" query. + +CREATE TABLE IF NOT EXISTS mesh_observations ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + ts REAL NOT NULL, -- epoch seconds (MeshObservation.timestamp) + transport TEXT NOT NULL, -- "meshtastic" | "meshcore" + channel INTEGER, -- channel index + is_dm INTEGER NOT NULL DEFAULT 0, + sender_name TEXT, + sender_id TEXT, + text TEXT +); +CREATE INDEX IF NOT EXISTS idx_mesh_observations_ts ON mesh_observations(ts);