mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 09:21:33 +00:00
Compare commits
1 commit
main
...
feat/meshc
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
118c83c69e |
3 changed files with 172 additions and 0 deletions
|
|
@ -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:
|
||||
|
|
|
|||
78
work/meshai/persistence/mesh_observations.py
Normal file
78
work/meshai/persistence/mesh_observations.py
Normal 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
|
||||
27
work/meshai/persistence/migrations/v28.sql
Normal file
27
work/meshai/persistence/migrations/v28.sql
Normal 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);
|
||||
Loading…
Add table
Add a link
Reference in a new issue