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 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()/"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: 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("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: 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
                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,
        }

    # ── message deletion ──────────────────────────────────────────────────

    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
            placeholders = ",".join("?" * len(cursors))
            sql = f"DELETE FROM outbox WHERE cursor IN ({placeholders})"
            # Safe: ``placeholders`` is only ``?`` markers; every cursor value is
            # bound as a parameter (no user data in the SQL text).
            # pi-lens-ignore: python-sql-injection
            self._conn.execute(sql, cursors)
            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("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, contextlib.suppress(Exception):
            # Best-effort: a close failure on shutdown is not actionable.
            self._conn.close()