"""SQLite offline outbox + monotonic sync cursor (M3). Undelivered frames (sent while no device is live) are appended with a **monotonic** cursor so a reconnecting app can ``sync {cursor}`` the delta without re-reading full history. The cursor is a separate high-water counter that only ever increases -- pruning old rows never resets it, so a late reconnect can't be handed a cursor lower than one it already saw. Retention prunes rows older than ``outbox_retention_hours`` (default 72h). A device offline longer than the window misses those frames; it recovers full context via ``history`` (M5 wires push so the device is woken to sync). Storage: ``get_hermes_home()/"iris"/outbox.db``. Milestone M3 (built), extended in M5 (push integration). """ import contextlib import json import logging import sqlite3 import threading import time from pathlib import Path from typing import Any logger = logging.getLogger(__name__) DEFAULT_RETENTION_HOURS = 72 _REPLAY_LIMIT = 1000 _PRUNE_INTERVAL_S = 3600.0 # Row cap (docs/08 §8.3): a very busy offline period must not grow the outbox # without bound; the oldest rows beyond the cap are pruned and the app is # told (generic notification, throttled in the adapter). DEFAULT_MAX_ROWS = 5000 class Outbox: """Persistent outbox under ``get_hermes_home()/"iris"``. Thread-safe (single connection + lock); operations are small and fast enough to run inline on the gateway's asyncio loop (mirrors ``DeviceRegistry`` / ``ChannelDirectory``). """ def __init__( self, db_path: Path, retention_hours: int = DEFAULT_RETENTION_HOURS, max_rows: int = DEFAULT_MAX_ROWS, ): self._db_path = Path(db_path) self._db_path.parent.mkdir(parents=True, exist_ok=True) self._retention_hours = max(1, int(retention_hours)) self._max_rows = max(1, int(max_rows)) self._lock = threading.Lock() self._last_prune = 0.0 self._overflow_pruned = 0 self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False) self._conn.row_factory = sqlite3.Row with self._lock: self._conn.execute("PRAGMA journal_mode=WAL") self._conn.execute( """ CREATE TABLE IF NOT EXISTS outbox ( cursor INTEGER PRIMARY KEY, chat_id TEXT, frame TEXT NOT NULL, created REAL NOT NULL DEFAULT 0 ) """ ) self._conn.execute("CREATE INDEX IF NOT EXISTS idx_outbox_created ON outbox (created)") self._conn.execute( """ CREATE TABLE IF NOT EXISTS counters ( name TEXT PRIMARY KEY, value INTEGER NOT NULL DEFAULT 0 ) """ ) self._conn.commit() # ── append / cursor ─────────────────────────────────────────────────── def append(self, chat_id: str | None, frame_json: str) -> int: """Append a frame; returns the (monotonic) cursor assigned to it.""" now = time.time() with self._lock: self._conn.execute( "INSERT INTO counters (name, value) VALUES ('cursor', 1) " "ON CONFLICT(name) DO UPDATE SET value = value + 1" ) row = self._conn.execute("SELECT value FROM counters WHERE name = 'cursor'").fetchone() cursor = int(row["value"]) if row else 1 self._conn.execute( "INSERT INTO outbox (cursor, chat_id, frame, created) VALUES (?, ?, ?, ?)", (cursor, chat_id, frame_json, now), ) self._enforce_row_cap() self._conn.commit() self._maybe_prune() return cursor def take_overflow_pruned(self) -> int: """Rows pruned by the row cap since the last call (and reset to 0). The adapter turns a non-zero count into a (throttled) generic notification so the app knows older frames are gone. """ with self._lock: n = self._overflow_pruned self._overflow_pruned = 0 return n def _enforce_row_cap(self) -> None: """Drop the oldest rows beyond ``max_rows`` (caller holds the lock).""" row = self._conn.execute("SELECT COUNT(*) AS n FROM outbox").fetchone() n = int(row["n"]) if row else 0 excess = n - self._max_rows if excess <= 0: return self._conn.execute( "DELETE FROM outbox WHERE cursor IN (" " SELECT cursor FROM outbox ORDER BY cursor ASC LIMIT ?)", (excess,), ) self._overflow_pruned += excess logger.info("iris outbox: row cap pruned %s oldest row(s)", excess) def latest_cursor(self) -> int: """The high-water cursor (0 when nothing has been appended).""" with self._lock: row = self._conn.execute("SELECT value FROM counters WHERE name = 'cursor'").fetchone() return int(row["value"]) if row else 0 # ── replay ──────────────────────────────────────────────────────────── def replay(self, cursor: int, limit: int = _REPLAY_LIMIT) -> list[dict[str, Any]]: """Frames with ``cursor > `cursor```, oldest first. Each entry: ``{cursor, chat_id, frame}`` where ``frame`` is the parsed frame dict (the caller re-serializes / forwards it to the device). """ cursor = max(0, int(cursor or 0)) limit = max(1, min(int(limit or _REPLAY_LIMIT), _REPLAY_LIMIT)) with self._lock: rows = self._conn.execute( "SELECT cursor, chat_id, frame FROM outbox " "WHERE cursor > ? ORDER BY cursor ASC LIMIT ?", (cursor, limit), ).fetchall() out: list[dict[str, Any]] = [] for r in rows: try: frame = json.loads(r["frame"]) except (json.JSONDecodeError, TypeError): continue if not isinstance(frame, dict): continue out.append({"cursor": int(r["cursor"]), "chat_id": r["chat_id"], "frame": frame}) return out # ── history (full message history for a chat/thread) ────────────────── def history( self, chat_id: str, thread_id: str | None = None, before_message_id: str | None = None, limit: int = 50, ) -> dict[str, Any]: """Final messages for a chat/thread, for the ``history`` frame. Reconstructs the message list from the outbox log: a final message is either a standalone ``message`` frame (user echo / non-streaming assistant) or a ``message.stop`` frame (streaming assistant final). Intermediate frames (``message.start``/``message.update``, tool, commentary, notification, …) are skipped. Messages are deduplicated by ``message_id`` and returned oldest → newest. Pagination: ``before_message_id`` returns the page of messages older than that id (the newest page when omitted). Returns ``{messages, has_more, oldest_message_id}``. """ limit = max(1, min(int(limit or 50), 200)) with self._lock: rows = self._conn.execute( "SELECT cursor, frame FROM outbox WHERE chat_id = ? ORDER BY cursor ASC", (chat_id,), ).fetchall() final: list[dict[str, Any]] = [] for r in rows: try: frame = json.loads(r["frame"]) except (json.JSONDecodeError, TypeError): continue if not isinstance(frame, dict): continue if thread_id is not None and frame.get("thread_id") != thread_id: continue ftype = frame.get("type") payload = frame.get("payload") if not isinstance(payload, dict): continue if ftype == "message": role = payload.get("role") if role not in ("user", "assistant"): continue msg = { "cursor": int(r["cursor"]), "message_id": payload.get("message_id"), "role": role, "text": payload.get("text", ""), "reasoning": payload.get("reasoning"), "model": payload.get("model"), "tokens": payload.get("tokens"), "runtime": payload.get("runtime"), "ts": payload.get("ts"), } # Omit ``media`` when absent (schema: array, not null) — a # ``"media": null`` would break the app's deserialization. if payload.get("media"): msg["media"] = payload["media"] final.append(msg) elif ftype == "message.stop": # Streaming finals carry no media (offers are separate # frames); omit the key (schema: array, not null). final.append( { "cursor": int(r["cursor"]), "message_id": payload.get("message_id"), "role": "assistant", "text": payload.get("final_text", ""), "reasoning": payload.get("reasoning"), "model": payload.get("model"), "tokens": payload.get("tokens"), "runtime": payload.get("runtime"), "ts": payload.get("ts"), } ) # Deduplicate by message_id (keep the latest occurrence), keep order. by_id: dict[str, dict[str, Any]] = {} for m in final: mid = m.get("message_id") if mid: by_id[mid] = m ordered = sorted(by_id.values(), key=lambda m: m["cursor"]) # Paginate: messages older than ``before_message_id`` (newest page when # omitted / not found — a pruned anchor falls back to the newest page). if before_message_id: idx = next( (i for i, m in enumerate(ordered) if m["message_id"] == before_message_id), None, ) pool = ordered[:idx] if idx is not None else ordered else: pool = ordered has_more = len(pool) > limit page = pool[-limit:] if has_more else list(pool) oldest_message_id = page[0]["message_id"] if page else None for m in page: m.pop("cursor", None) # Omit absent optional fields (the app's serializer treats a # missing key as its default, but a JSON ``null`` for a # non-nullable field like ``media`` would fail to parse). for key in ("reasoning", "model", "tokens", "runtime", "ts", "media"): if m.get(key) is None: m.pop(key, None) return { "messages": page, "has_more": has_more, "oldest_message_id": oldest_message_id, } # ── message deletion ────────────────────────────────────────────────── def message_info( self, chat_id: str, message_id: str, thread_id: str | None = None, ) -> dict[str, Any] | None: """Look up a message's final frame data (role / text / ts) in the outbox. Used to match a ``message.delete`` to the hermes session-store row (which is keyed by content + timestamp, not the plugin's message id). Returns ``{role, text, ts}`` for the message's final frame -- a standalone ``message`` frame when present, else the ``message.stop`` frame of a streamed reply -- or ``None`` when the message is not in the outbox (e.g. already pruned by retention). """ if not message_id: return None with self._lock: rows = self._conn.execute( "SELECT frame FROM outbox WHERE chat_id = ?", (chat_id,) ).fetchall() msg_frame: dict[str, Any] | None = None stop_frame: dict[str, Any] | None = None for r in rows: try: frame = json.loads(r["frame"]) except (json.JSONDecodeError, TypeError): continue if not isinstance(frame, dict): continue if thread_id is not None and frame.get("thread_id") != thread_id: continue payload = frame.get("payload") if not isinstance(payload, dict) or payload.get("message_id") != message_id: continue ftype = frame.get("type") if ftype == "message": msg_frame = { "role": payload.get("role"), "text": payload.get("text", ""), "ts": payload.get("ts"), } elif ftype == "message.stop": stop_frame = { "role": "assistant", "text": payload.get("final_text", ""), "ts": payload.get("ts"), } return msg_frame or stop_frame def delete_message( self, chat_id: str, message_id: str, thread_id: str | None = None, ) -> int: """Remove every outbox frame belonging to *message_id* in *chat_id*. A message can span several frames (``message`` / ``message.start`` / ``message.update`` / ``message.stop`` / ``media.offer`` / ``commentary``); all of them are removed so neither ``history`` nor a ``sync`` replay can resurrect the message. The delete is scoped to the exact lane: a flat-lane delete (``thread_id=None``) matches only frames with no ``thread_id``, and a thread delete matches only that thread's frames (a ``message_id`` is unique to one lane, so this is a safety net, not a filter that drops real frames). Returns the number of rows removed (0 when the message is not in the outbox — e.g. already pruned by retention). """ if not message_id: return 0 with self._lock: rows = self._conn.execute( "SELECT cursor, frame FROM outbox WHERE chat_id = ?", (chat_id,) ).fetchall() cursors: list[int] = [] for r in rows: try: frame = json.loads(r["frame"]) except (json.JSONDecodeError, TypeError): continue if not isinstance(frame, dict): continue if frame.get("thread_id") != thread_id: continue payload = frame.get("payload") if isinstance(payload, dict) and payload.get("message_id") == message_id: cursors.append(int(r["cursor"])) if not cursors: return 0 # One bound-parameter delete per cursor (a message spans only a few # frames); same transaction, no string-built SQL. for cursor in cursors: self._conn.execute("DELETE FROM outbox WHERE cursor = ?", (cursor,)) self._conn.commit() return len(cursors) def delete_lane(self, chat_id: str, thread_id: str | None = None) -> int: """Remove every outbox frame for a lane (channel or thread). * ``thread_id is None`` -> a **channel**: all frames whose ``chat_id`` column is *chat_id* (the flat lane plus every thread under it). * ``thread_id`` set -> a **thread**: frames for *chat_id* whose frame carries that ``thread_id``. Called on channel/thread deletion so neither ``history`` nor a ``sync`` replay can resurrect the lane's messages. Returns the number of rows removed. """ with self._lock: rows = self._conn.execute( "SELECT cursor, frame FROM outbox WHERE chat_id = ?", (chat_id,) ).fetchall() cursors: list[int] = [] for r in rows: if thread_id is None: cursors.append(int(r["cursor"])) continue try: frame = json.loads(r["frame"]) except (json.JSONDecodeError, TypeError): continue if isinstance(frame, dict) and frame.get("thread_id") == thread_id: cursors.append(int(r["cursor"])) if not cursors: return 0 for cursor in cursors: self._conn.execute("DELETE FROM outbox WHERE cursor = ?", (cursor,)) self._conn.commit() return len(cursors) # ── retention ───────────────────────────────────────────────────────── def _maybe_prune(self) -> None: now = time.time() if now - self._last_prune < _PRUNE_INTERVAL_S: return self._last_prune = now cutoff = now - self._retention_hours * 3600 with self._lock: try: self._conn.execute("DELETE FROM outbox WHERE created < ?", (cutoff,)) self._conn.commit() except sqlite3.Error as e: logger.debug("iris outbox: prune failed: %s", e) def prune(self) -> None: """Force a retention prune (ignores the interval throttle).""" self._last_prune = 0.0 self._maybe_prune() def close(self) -> None: with self._lock, contextlib.suppress(Exception): # Best-effort: a close failure on shutdown is not actionable. self._conn.close()