354 lines
14 KiB
Python
354 lines
14 KiB
Python
"""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,
|
|
}
|
|
|
|
# ── message deletion ──────────────────────────────────────────────────
|
|
|
|
def delete_message(
|
|
self,
|
|
chat_id: str,
|
|
message_id: str,
|
|
thread_id: Optional[str] = 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))
|
|
self._conn.execute(
|
|
f"DELETE FROM outbox WHERE cursor IN ({placeholders})", 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:
|
|
try:
|
|
self._conn.close()
|
|
except Exception:
|
|
pass |