feat(context): durable SQLite-backed retention for MeshContext observations (#114)

Mesh-context buffer (recap/summary source) was pure in-memory
(deque(maxlen=50000)), wiped on every container restart. Adds
mesh_observations table (migration v28) + persistence/mesh_observations.py
accessors following the existing get_db()/observer_locations.py pattern.

MeshContext.observe() now write-throughs each observation to SQLite
(fail-safe -- DB errors are logged and swallowed, never block or drop the
in-memory path). __init__ loads recent rows (within max_age, up to the
hard cap) back into the deque on startup so recap works immediately after
a restart. prune() now also deletes SQLite rows older than max_age on the
same hourly cadence as the existing in-memory prune.

Verified: wrote one observation through the real observe() path, confirmed
1 row in mesh_observations; docker restart meshai; log confirms "Loaded 1
mesh observations from durable store" and a fresh MeshContext instance
recovers the observation via get_context_block().

Investigated a suspected recap-grounding bug (LLM ignoring the mesh-context
block and replying with a disclaimer) per the same task brief, but could
not reproduce it against gemini-3.1-flash-lite across 8 query phrasings
(minimal and full production-fidelity system prompts, including the
configured static prompt clause that explicitly permits the disclaimer
when no traffic is shown) -- the model correctly grounds its replies in
the observed traffic block in every case tested. No grounding-clause
change made; premise did not reproduce.

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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:

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

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);