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


def _frame_thread_id(frame: dict[str, Any]) -> str | None:
    """A frame's lane: its ``thread_id``, normalized (absent/blank -> None)."""
    tid = frame.get("thread_id")
    if not isinstance(tid, str) or not tid.strip():
        return None
    return tid


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
            # Exact lane match: a flat-lane history (thread_id=None) must NOT
            # include frames that belong to a thread, and vice versa. (The old
            # loose filter let auto-threaded messages leak into the flat lane
            # on restart, where a delete sent with thread_id=None then matched
            # nothing in the outbox and the messages "resurrected" later.)
            if _frame_thread_id(frame) != 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).

        The lane is matched exactly first (a flat-lane lookup, ``thread_id
        = None``, sees only frames with no ``thread_id``); when that finds
        nothing the lookup falls back to the ``message_id`` alone across all
        lanes (it is a unique uuid4), so a stale/missing ``thread_id`` on the
        request still resolves the row.
        """
        if not message_id:
            return None
        with self._lock:
            rows = self._conn.execute(
                "SELECT frame FROM outbox WHERE chat_id = ?", (chat_id,)
            ).fetchall()

        def scan(lane: str | None, exact: bool) -> dict[str, Any] | None:
            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 exact and _frame_thread_id(frame) != lane:
                    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

        return scan(thread_id, exact=True) or scan(None, exact=False)

    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 first: a flat-lane delete (``thread_id=None``) matches only
        frames with no ``thread_id``, and a thread delete matches only that
        thread's frames. When the exact lane matches nothing, the delete falls
        back to the ``message_id`` alone (it is a unique uuid4, so it cannot
        hit the wrong message) — this keeps deletes working when the request's
        lane is stale or missing (e.g. a message the app cached in the flat
        lane that the gateway auto-threaded). 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()

            def cursors_for(lane: str | None, exact: bool) -> list[int]:
                out: list[int] = []
                for r in rows:
                    try:
                        frame = json.loads(r["frame"])
                    except (json.JSONDecodeError, TypeError):
                        continue
                    if not isinstance(frame, dict):
                        continue
                    if exact and _frame_thread_id(frame) != lane:
                        continue
                    payload = frame.get("payload")
                    if isinstance(payload, dict) and payload.get("message_id") == message_id:
                        out.append(int(r["cursor"]))
                return out

            # Exact lane first (a flat-lane delete must not reach into
            # threads); fall back to the message_id across all lanes only
            # when the exact lane matches nothing (stale/missing thread_id).
            cursors = cursors_for(thread_id, exact=True) or cursors_for(thread_id, exact=False)
            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()