iris_x_hermes/gateway-plugin/protocol.py

"""Frame schemas -- the single source of truth for the wire protocol.

Every frame the plugin sends/receives is modelled here as a dataclass +
constants. ``docs/protocol/frames.schema.json`` is generated/mirrored from
this module, and the Kotlin side mirrors these shapes (see
``docs/04-wire-protocol.md``).

Milestone M1: hello/hello.ack, message, message.send, error, ping/pong,
typing (typing is pulled forward from M2 so the app gets a live
"working…" indicator during the first milestone).
Milestone M2: message.start/update/stop, reasoning (on message /
message.stop), tool.start/progress/end, commentary.
Milestone M3: channel.*, search, sync.
Milestone M4: media.*.
Milestone M5: notification, fcm.register, read.receipt, status.
"""

import json
from dataclasses import dataclass, field
from typing import Any, Optional

PROTOCOL_VERSION = 1

# ---------------------------------------------------------------------------
# Frame type constants (the ``type`` field of every frame)
# ---------------------------------------------------------------------------

# Pairing / lifecycle
TYPE_HELLO_ACK = "hello.ack"
TYPE_ERROR = "error"

# Chat
TYPE_MESSAGE = "message"
TYPE_MESSAGE_SEND = "message.send"
TYPE_TYPING = "typing"

# Streaming (M2)
TYPE_MESSAGE_START = "message.start"
TYPE_MESSAGE_UPDATE = "message.update"
TYPE_MESSAGE_STOP = "message.stop"

# Message deletion (app requests; broadcast to all devices)
TYPE_MESSAGE_DELETE = "message.delete"
TYPE_MESSAGE_DELETED = "message.deleted"

# Tool activity (M2)
TYPE_TOOL_START = "tool.start"
TYPE_TOOL_PROGRESS = "tool.progress"
TYPE_TOOL_END = "tool.end"

# Intermediate assistant beat (M2)
TYPE_COMMENTARY = "commentary"

# Channels / threads (M3)
TYPE_CHANNEL_CREATE = "channel.create"
TYPE_CHANNEL_RENAME = "channel.rename"
TYPE_CHANNEL_SET_DEFAULT = "channel.set_default"
TYPE_CHANNEL_FAVORITE = "channel.favorite"
TYPE_CHANNEL_ICON = "channel.icon"
TYPE_CHANNEL_SET_AUTOMATION = "channel.set_automation"
TYPE_CHANNEL_DELETE = "channel.delete"
TYPE_CHANNEL_CREATED = "channel.created"
TYPE_CHANNEL_RENAMED = "channel.renamed"
TYPE_CHANNEL_DELETED = "channel.deleted"
TYPE_CHANNEL_LIST = "channel.list"

# Search (M3)
TYPE_SEARCH = "search"
TYPE_SEARCH_RESULTS = "search.results"

# Slash-command catalog (app's "/" drawer)
TYPE_COMMANDS_CATALOG = "commands.catalog"

# Interactive pickers (slash-command choice menus, e.g. /reasoning, /fast)
TYPE_PICKER_CHOICE = "picker.choice"
TYPE_PICKER_SELECT = "picker.select"

# Reconnect catch-up (M3 outbox; extended by M5 push)
TYPE_SYNC = "sync"
TYPE_SYNC_DONE = "sync.done"

# Full message history (initial channel open / scroll-up pagination)
TYPE_HISTORY = "history"

# Media (M4)
TYPE_MEDIA_UPLOAD_ACK = "media.upload.ack"
TYPE_MEDIA_OFFER = "media.offer"

# Push / notifications (M5)
TYPE_NOTIFICATION = "notification"
TYPE_FCM_REGISTER = "fcm.register"
TYPE_READ_RECEIPT = "read.receipt"

# Gateway health (M5)
TYPE_STATUS = "status"

# ---------------------------------------------------------------------------
# Error codes (``error`` frame payload.code)
# ---------------------------------------------------------------------------

ERR_AUTH = "auth"
ERR_NOT_FOUND = "not_found"
ERR_RATE_LIMITED = "rate_limited"
ERR_MEDIA_TOO_LARGE = "media_too_large"
ERR_UNSUPPORTED = "unsupported"
ERR_INTERNAL = "internal"

# ---------------------------------------------------------------------------
# Message roles (``message`` frame payload.role)
# ---------------------------------------------------------------------------

ROLE_USER = "user"
ROLE_ASSISTANT = "assistant"
ROLE_SYSTEM = "system"
ROLE_CRON = "cron"

# ---------------------------------------------------------------------------
# Notification kinds (``notification`` frame payload.kind, M5)
# ---------------------------------------------------------------------------

NOTIF_CHANNEL_CREATED = "channel_created"
NOTIF_CHANNEL_RENAMED = "channel_renamed"
NOTIF_CHANNEL_DELETED = "channel_deleted"
NOTIF_CRON = "cron"
NOTIF_APPROVAL = "approval"
NOTIF_CLARIFY = "clarify"
NOTIF_GENERIC = "generic"

# Kinds that push even when a device is live (the app may be backgrounded;
# it decides whether to also show an in-app banner).
HIGH_PRIORITY_NOTIF_KINDS = frozenset({NOTIF_APPROVAL, NOTIF_CLARIFY, NOTIF_CRON})

# ---------------------------------------------------------------------------
# Gateway health states (``status`` frame payload.state)
# ---------------------------------------------------------------------------

STATUS_ONLINE = "online"
STATUS_RESTARTING = "restarting"
STATUS_DEGRADED = "degraded"


# ---------------------------------------------------------------------------
# Envelope
# ---------------------------------------------------------------------------


@dataclass
class Frame:
    """One wire frame.

    ``v`` is always serialised; ``id``/``chat_id``/``thread_id``/``cursor``
    are omitted when ``None`` (events carry no ``id``; chat-scoped frames
    carry ``chat_id``/``thread_id`` at the top level for convenience).

    ``cursor`` is set only on frames replayed by ``sync``: the outbox cursor
    the frame was parked under. The app uses it to skip re-notifying frames
    that already woke the device via push (docs/08 §8.7).
    """

    type: str
    payload: dict[str, Any] = field(default_factory=dict)
    id: int | None = None
    chat_id: str | None = None
    thread_id: str | None = None
    cursor: int | None = None
    v: int = PROTOCOL_VERSION

    def to_dict(self) -> dict[str, Any]:
        d: dict[str, Any] = {"v": self.v, "type": self.type}
        if self.id is not None:
            d["id"] = self.id
        if self.chat_id is not None:
            d["chat_id"] = self.chat_id
        if self.thread_id is not None:
            d["thread_id"] = self.thread_id
        if self.cursor is not None:
            d["cursor"] = self.cursor
        d["payload"] = self.payload
        return d

    def to_json(self) -> str:
        return json.dumps(self.to_dict(), separators=(",", ":"), ensure_ascii=False)

    @classmethod
    def from_json(cls, raw: "str | bytes") -> Optional["Frame"]:
        """Parse a text frame. Returns ``None`` for anything not a valid
        frame (bad JSON, non-object, missing/invalid ``type``) so callers
        can ignore malformed input (forward-compat)."""
        try:
            data = json.loads(raw)
        except (json.JSONDecodeError, TypeError, UnicodeDecodeError, ValueError):
            return None
        if not isinstance(data, dict):
            return None
        ftype = data.get("type")
        if not isinstance(ftype, str) or not ftype:
            return None
        payload = data.get("payload")
        if not isinstance(payload, dict):
            payload = {}
        fid = data.get("id")
        if not isinstance(fid, int) or isinstance(fid, bool):
            fid = None
        chat_id = data.get("chat_id")
        if not isinstance(chat_id, str):
            chat_id = None
        thread_id = data.get("thread_id")
        if not isinstance(thread_id, str):
            thread_id = None
        cursor = data.get("cursor")
        if not isinstance(cursor, int) or isinstance(cursor, bool):
            cursor = None
        return cls(
            type=ftype,
            payload=payload,
            id=fid,
            chat_id=chat_id,
            thread_id=thread_id,
            cursor=cursor,
        )


# ---------------------------------------------------------------------------
# Frame constructors (server -> app)
# ---------------------------------------------------------------------------


def hello_ack(
    server_caps: dict[str, Any],
    sync_cursor: int = 0,
    channels: list | None = None,
    last_pushed_cursor: int = 0,
) -> Frame:
    return Frame(
        type=TYPE_HELLO_ACK,
        payload={
            "server_caps": server_caps,
            "sync_cursor": sync_cursor,
            "channels": channels or [],
            # M5: highest outbox cursor already delivered to THIS device via
            # the push backend (0 = never). The app skips system
            # notifications for sync-replayed frames at/below it (dedupe,
            # docs/08 §8.7).
            "last_pushed_cursor": last_pushed_cursor,
        },
    )


# ---------------------------------------------------------------------------
# Runtime-metadata footer (app-controlled display)
#
# The gateway ALWAYS attaches a structured ``runtime`` object to final
# assistant messages so the app can render a Telegram-style footer (model,
# context %, cwd, latency, cost). Whether/what is shown is a per-app setting,
# NOT a hermes config — the gateway sends the data unconditionally and the
# app decides. Mirrors hermes' ``gateway/runtime_footer.py`` fields but as
# structured data (the app formats + picks fields).
#
# Recognised keys (all optional; absent when the data is unavailable):
#   model        — bare model id, vendor prefix dropped (``gpt-5.4``)
#   context_pct  — last-call context occupancy, 0-100 (int)
#   cwd          — home-relative working dir (``~``)
#   latency      — wall-clock turn duration, seconds (float)
#   cost         — turn cost, USD (float); absent for local/free models
# ---------------------------------------------------------------------------

RUNTIME_FIELDS: tuple[str, ...] = ("model", "context_pct", "cwd", "latency", "cost")


def runtime_footer(
    *,
    model: str | None = None,
    context_pct: int | None = None,
    cwd: str | None = None,
    latency: float | None = None,
    cost: float | None = None,
) -> dict[str, Any]:
    """Build the structured ``runtime`` footer object.

    Only fields with data are included (a partially-populated footer is
    better than empty slots). Returns ``{}`` when nothing is available.
    """
    d: dict[str, Any] = {}
    if model:
        d["model"] = model
    if context_pct is not None:
        d["context_pct"] = max(0, min(100, int(context_pct)))
    if cwd:
        d["cwd"] = cwd
    if latency is not None and latency >= 0:
        d["latency"] = round(latency, 3)
    if cost is not None and cost > 0:
        d["cost"] = round(cost, 6)
    return d


# Frame builder mirrors the wire schema (docs/04); the many fields are the
# message's full shape, so the arg count is intentional.
def message(  # noqa: PLR0913
    chat_id: str,
    message_id: str,
    role: str,
    text: str,
    *,
    thread_id: str | None = None,
    reasoning: str | None = None,
    media: list | None = None,
    reply_to: str | None = None,
    model: str | None = None,
    tokens: int | None = None,
    runtime: dict[str, Any] | None = None,
    ts: int | None = None,
) -> Frame:
    payload: dict[str, Any] = {
        "message_id": message_id,
        "role": role,
        "text": text,
    }
    if reasoning:
        payload["reasoning"] = reasoning
    if media:
        payload["media"] = media
    if reply_to:
        payload["reply_to"] = reply_to
    if model:
        payload["model"] = model
    if tokens is not None:
        payload["tokens"] = tokens
    if runtime:
        payload["runtime"] = runtime
    if ts is not None:
        payload["ts"] = ts
    return Frame(type=TYPE_MESSAGE, chat_id=chat_id, thread_id=thread_id, payload=payload)


def typing(chat_id: str, on: bool = True, *, thread_id: str | None = None) -> Frame:
    return Frame(
        type=TYPE_TYPING,
        chat_id=chat_id,
        thread_id=thread_id,
        payload={"on": on},
    )


# ---------------------------------------------------------------------------
# Streaming frames (M2)
# ---------------------------------------------------------------------------


def message_start(
    chat_id: str,
    message_id: str,
    role: str = ROLE_ASSISTANT,
    *,
    thread_id: str | None = None,
) -> Frame:
    """Open a streaming bubble."""
    return Frame(
        type=TYPE_MESSAGE_START,
        chat_id=chat_id,
        thread_id=thread_id,
        payload={"message_id": message_id, "role": role},
    )


def message_update(
    chat_id: str,
    message_id: str,
    text: str,
    *,
    thread_id: str | None = None,
) -> Frame:
    """Replace the live bubble text (full snapshot)."""
    return Frame(
        type=TYPE_MESSAGE_UPDATE,
        chat_id=chat_id,
        thread_id=thread_id,
        payload={"message_id": message_id, "text": text},
    )


def message_stop(
    chat_id: str,
    message_id: str,
    final_text: str,
    *,
    thread_id: str | None = None,
    reasoning: str | None = None,
    model: str | None = None,
    tokens: int | None = None,
    runtime: dict[str, Any] | None = None,
    ts: int | None = None,
) -> Frame:
    """Finalize a streaming bubble."""
    payload: dict[str, Any] = {
        "message_id": message_id,
        "final_text": final_text,
    }
    if reasoning:
        payload["reasoning"] = reasoning
    if model:
        payload["model"] = model
    if tokens is not None:
        payload["tokens"] = tokens
    if runtime:
        payload["runtime"] = runtime
    if ts is not None:
        payload["ts"] = ts
    return Frame(
        type=TYPE_MESSAGE_STOP,
        chat_id=chat_id,
        thread_id=thread_id,
        payload=payload,
    )


# ---------------------------------------------------------------------------
# Tool activity frames (M2)
# ---------------------------------------------------------------------------


def tool_start(
    chat_id: str,
    index: int,
    name: str,
    *,
    thread_id: str | None = None,
    preview: str | None = None,
    args: dict[str, Any] | None = None,
    emoji: str | None = None,
) -> Frame:
    payload: dict[str, Any] = {"index": index, "name": name}
    if preview:
        payload["preview"] = preview
    if args:
        payload["args"] = args
    if emoji:
        payload["emoji"] = emoji
    return Frame(
        type=TYPE_TOOL_START,
        chat_id=chat_id,
        thread_id=thread_id,
        payload=payload,
    )


def tool_progress(
    chat_id: str,
    index: int,
    name: str,
    *,
    thread_id: str | None = None,
    note: str | None = None,
) -> Frame:
    payload: dict[str, Any] = {"index": index, "name": name}
    if note:
        payload["note"] = note
    return Frame(
        type=TYPE_TOOL_PROGRESS,
        chat_id=chat_id,
        thread_id=thread_id,
        payload=payload,
    )


def tool_end(
    chat_id: str,
    index: int,
    name: str,
    *,
    thread_id: str | None = None,
    ok: bool = True,
    duration: float | None = None,
    output_preview: str | None = None,
) -> Frame:
    payload: dict[str, Any] = {"index": index, "name": name, "ok": ok}
    if duration is not None:
        payload["duration"] = duration
    if output_preview:
        payload["output_preview"] = output_preview
    return Frame(
        type=TYPE_TOOL_END,
        chat_id=chat_id,
        thread_id=thread_id,
        payload=payload,
    )


# ---------------------------------------------------------------------------
# Commentary frame (M2)
# ---------------------------------------------------------------------------


def commentary(
    chat_id: str,
    message_id: str,
    text: str,
    *,
    thread_id: str | None = None,
) -> Frame:
    """An intermediate assistant beat (between tool iterations)."""
    return Frame(
        type=TYPE_COMMENTARY,
        chat_id=chat_id,
        thread_id=thread_id,
        payload={"message_id": message_id, "text": text},
    )


# ---------------------------------------------------------------------------
# Channel directory frames (M3)
# ---------------------------------------------------------------------------


def _channel_payload(entry: dict[str, Any]) -> dict[str, Any]:
    """Project a directory entry onto the wire shape."""
    payload: dict[str, Any] = {
        "chat_id": entry.get("chat_id"),
        "name": entry.get("name"),
        "kind": entry.get("kind", "channel"),
    }
    if entry.get("parent_chat_id") is not None:
        payload["parent_chat_id"] = entry["parent_chat_id"]
    if entry.get("is_default"):
        payload["is_default"] = True
    if entry.get("archived"):
        payload["archived"] = True
    if entry.get("favorite"):
        payload["favorite"] = True
    if entry.get("icon"):
        payload["icon"] = entry["icon"]
    if entry.get("color"):
        payload["color"] = entry["color"]
    if entry.get("automation"):
        payload["automation"] = True
    return payload


def channel_created(entry: dict[str, Any], auto: bool = False) -> Frame:
    """Broadcast: a channel/thread was created.

    ``auto=True`` marks a thread the gateway minted itself for an incoming
    message (auto-threading, docs/06 §6.3): the app jumps into it and the
    name is an instant derived title, upgraded by the LLM via a follow-up
    ``channel.renamed``.
    """
    payload = _channel_payload(entry)
    if auto:
        payload["auto"] = True
    return Frame(type=TYPE_CHANNEL_CREATED, payload=payload)


def channel_renamed(entry: dict[str, Any]) -> Frame:
    """Broadcast: a channel/thread was renamed."""
    return Frame(type=TYPE_CHANNEL_RENAMED, payload=_channel_payload(entry))


def channel_deleted(chat_id: str) -> Frame:
    """Broadcast: a channel or thread was deleted (its history wiped too)."""
    return Frame(type=TYPE_CHANNEL_DELETED, payload={"chat_id": chat_id})


def channel_list(channels: list[dict[str, Any]]) -> Frame:
    """Full directory (response to a ``channel.list`` request)."""
    return Frame(
        type=TYPE_CHANNEL_LIST,
        payload={"channels": [_channel_payload(c) for c in channels]},
    )


# ---------------------------------------------------------------------------
# Search frames (M3)
# ---------------------------------------------------------------------------


def search_results(
    query: str,
    scope: str,
    hits: list[dict[str, Any]],
    *,
    id: int | None = None,
) -> Frame:
    """Response to a ``search`` request.

    Each hit: ``{message_id, chat_id, thread_id, role, snippet, ts}``.
    """
    return Frame(
        type=TYPE_SEARCH_RESULTS,
        id=id,
        payload={"query": query, "scope": scope, "hits": hits},
    )


# ---------------------------------------------------------------------------
# Slash-command catalog frames
# ---------------------------------------------------------------------------


def commands_catalog(commands: list[dict[str, Any]], *, id: int | None = None) -> Frame:
    """Response to a ``commands.catalog`` request: the gateway's slash-command
    catalog for the app's ``/`` drawer.

    Each entry: ``{name, description, args_hint, category, aliases}`` where
    ``name``/``aliases`` carry the leading slash (``"/new"``, ``["/reset"]``).
    """
    return Frame(
        type=TYPE_COMMANDS_CATALOG,
        id=id,
        payload={"commands": commands},
    )


# ---------------------------------------------------------------------------
# Picker frames (interactive slash-command choice menus)
# ---------------------------------------------------------------------------


def picker_choice(
    picker_id: str,
    title: str,
    choices: list[dict[str, Any]],
    chat_id: str,
    *,
    thread_id: str | None = None,
) -> Frame:
    """Event: an interactive choice picker (one tap → one value).

    Used by slash commands with a finite option set (``/reasoning``,
    ``/fast``, …) on platforms that support pickers. The app renders the
    title + choice buttons and answers with a ``picker.select`` frame
    carrying the same ``picker_id``. Outboxed, so a reconnecting device
    re-renders a still-pending picker.

    Each choice: ``{"value": str, "label": str, "is_current": bool}``.
    """
    return Frame(
        type=TYPE_PICKER_CHOICE,
        chat_id=chat_id,
        thread_id=thread_id,
        payload={"picker_id": picker_id, "title": title, "choices": choices},
    )


# ---------------------------------------------------------------------------
# Sync frames (M3 outbox)
# ---------------------------------------------------------------------------


def sync_done(cursor: int, *, id: int | None = None) -> Frame:
    """Terminal frame of a ``sync`` replay: the new cursor to persist."""
    return Frame(type=TYPE_SYNC_DONE, id=id, payload={"cursor": cursor})


# ---------------------------------------------------------------------------
# History frame (full message history for a chat/thread)
# ---------------------------------------------------------------------------


def history(
    chat_id: str,
    messages: list[dict[str, Any]],
    has_more: bool,
    *,
    thread_id: str | None = None,
    oldest_message_id: str | None = None,
    id: int | None = None,
) -> Frame:
    """Response to a ``history`` request: a page of final messages for a
    chat/thread, ordered oldest → newest. ``has_more`` signals older pages
    exist; ``oldest_message_id`` is the ``before_message_id`` for the next
    (older) page."""
    payload: dict[str, Any] = {
        "messages": messages,
        "has_more": has_more,
    }
    if oldest_message_id:
        payload["oldest_message_id"] = oldest_message_id
    return Frame(
        type=TYPE_HISTORY,
        id=id,
        chat_id=chat_id,
        thread_id=thread_id,
        payload=payload,
    )


# ---------------------------------------------------------------------------
# Message deletion frames
# ---------------------------------------------------------------------------


def message_deleted(
    chat_id: str,
    message_ids: list[str],
    *,
    thread_id: str | None = None,
    id: int | None = None,
) -> Frame:
    """Broadcast: the given message(s) were deleted from a chat/thread.

    Carried by the response to a ``message.delete`` request (``id`` set) and
    broadcast to every device so all of them drop the message(s) from their
    cache. Also outboxed, so a device that was offline learns of the deletion
    on its next ``sync``.
    """
    return Frame(
        type=TYPE_MESSAGE_DELETED,
        id=id,
        chat_id=chat_id,
        thread_id=thread_id,
        payload={"message_ids": list(message_ids)},
    )


# ---------------------------------------------------------------------------
# Push / notification frames (M5)
# ---------------------------------------------------------------------------


def notification(
    chat_id: str,
    kind: str,
    title: str,
    body: str,
    *,
    thread_id: str | None = None,
    ts: int | None = None,
) -> Frame:
    """Event: a transient in-app banner (and a push mirror when the device is
    offline). ``kind`` is one of the ``NOTIF_*`` constants."""
    payload: dict[str, Any] = {"kind": kind, "title": title, "body": body}
    if ts is not None:
        payload["ts"] = ts
    return Frame(type=TYPE_NOTIFICATION, chat_id=chat_id, thread_id=thread_id, payload=payload)


def fcm_register(fcm_token: str | None = None, ntfy_topic: str | None = None) -> Frame:
    """Request: update the device's push tokens (FCM rotation / ntfy topic)."""
    payload: dict[str, Any] = {}
    if fcm_token:
        payload["fcm_token"] = fcm_token
    if ntfy_topic:
        payload["ntfy_topic"] = ntfy_topic
    return Frame(type=TYPE_FCM_REGISTER, payload=payload)


def read_receipt(chat_id: str, message_id: str) -> Frame:
    """Ack to the originating device: the agent received and started
    processing the user's message (the app shows ✓✓ on the user bubble)."""
    return Frame(
        type=TYPE_READ_RECEIPT,
        chat_id=chat_id,
        payload={"message_id": message_id},
    )


# ---------------------------------------------------------------------------
# Gateway health frame (M5)
# ---------------------------------------------------------------------------


def status(state: str) -> Frame:
    """Gateway health state (``state`` is one of the ``STATUS_*`` constants)."""
    return Frame(type=TYPE_STATUS, payload={"state": state})


# ---------------------------------------------------------------------------
# Media frames (M4)
# ---------------------------------------------------------------------------


def media_offer(
    media_id: str,
    kind: str,
    mime: str,
    size: int,
    filename: str,
    *,
    chat_id: str | None = None,
    thread_id: str | None = None,
    message_id: str | None = None,
) -> Frame:
    """Event: the agent produced media the app can fetch via
    ``GET /v1/media/{media_id}`` (docs/19 §19.15).

    ``message_id`` (optional) associates the offer with the assistant message
    it belongs to (the app falls back to the lane's last assistant message).
    """
    payload: dict[str, Any] = {
        "media_id": media_id,
        "kind": kind,
        "mime": mime,
        "size": size,
        "filename": filename,
    }
    if message_id:
        payload["message_id"] = message_id
    return Frame(type=TYPE_MEDIA_OFFER, chat_id=chat_id, thread_id=thread_id, payload=payload)


def media_upload_ack(ok: bool, media_ref: str, *, id: int | None = None) -> Frame:
    """Response to ``POST /v1/media``: the ref is cached and may be used in a
    ``message.send`` ``media_refs``. Failures use ``error`` frames instead."""
    return Frame(
        type=TYPE_MEDIA_UPLOAD_ACK,
        id=id,
        payload={"ok": ok, "media_ref": media_ref},
    )


def error(code: str, message: str, *, id: int | None = None) -> Frame:
    return Frame(type=TYPE_ERROR, id=id, payload={"code": code, "message": message})