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

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: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Matt Johnson 2026-07-11 06:38:35 +00:00
commit 118c83c69e
3 changed files with 172 additions and 0 deletions

View file

@ -32,6 +32,14 @@ class MeshContext:
Passively observes all mesh messages (channels, DMs, BBS notifications) Passively observes all mesh messages (channels, DMs, BBS notifications)
and makes them available as context when generating LLM responses. and makes them available as context when generating LLM responses.
Observations older than max_age are pruned periodically. 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__( def __init__(
@ -52,6 +60,37 @@ class MeshContext:
self._ignore_nodes = set(ignore_nodes) if ignore_nodes else set() self._ignore_nodes = set(ignore_nodes) if ignore_nodes else set()
self._max_age = max_age 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( def observe(
self, self,
sender_name: str, sender_name: str,
@ -95,6 +134,24 @@ class MeshContext:
self._buffer.append(obs) self._buffer.append(obs)
logger.debug(f"Observed: ch{channel} {sender_name} [{transport}]: {text[:40]}...") 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: def prune(self) -> int:
"""Remove observations older than max_age. """Remove observations older than max_age.
@ -113,6 +170,16 @@ class MeshContext:
pruned = before - len(self._buffer) pruned = before - len(self._buffer)
if pruned > 0: if pruned > 0:
logger.info(f"Pruned {pruned} expired mesh observations ({len(self._buffer)} remaining)") 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 return pruned
def get_context_block(self, max_items: int = 20, transport: Optional[str] = None) -> str: def get_context_block(self, max_items: int = 20, transport: Optional[str] = None) -> str:

View file

@ -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

View file

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