iris_x_hermes/gateway-plugin/outbox.py

"""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()/"android"/outbox.db``.

Milestone M3 (built), extended in M5 (push integration).
"""

import json
import logging
import sqlite3
import threading
import time
from pathlib import Path
from typing import Any, Dict, List, Optional

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()/"android"``.

    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: Optional[str], 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("android 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: Optional[str] = None,
        before_message_id: Optional[str] = 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
                final.append(
                    {
                        "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"),
                        "ts": payload.get("ts"),
                        "media": payload.get("media"),
                    }
                )
            elif ftype == "message.stop":
                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"),
                        "ts": payload.get("ts"),
                        "media": None,
                    }
                )
        # 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", "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,
        }

    # ── 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("android 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:
            try:
                self._conn.close()
            except Exception:
                pass