rmon/docs/protocol.md

# RMon Wire Protocol

## Transport

- **TCP** over an SSH local port-forward tunnel
- **Port**: configurable, default `7891`
- One persistent TCP connection per UI client
- Multiple clients can connect to the same agent simultaneously

## Framing

Every message (in both directions) uses the same length-prefix frame:

```
┌─────────────────────┬────────────────────────────┐
│  length : u32 LE    │  payload : [u8; length]    │
└─────────────────────┴────────────────────────────┘
```

- `length` is the byte count of the serialized payload (excludes the 4-byte length field itself)
- `payload` is a `bincode`-serialized `ClientMessage` (client→agent) or `AgentMessage` (agent→client)
- `bincode` configuration: little-endian, fixed-size integers, no trailing bytes

Maximum frame size: **16 MiB** (`length` values above this are a protocol error; close connection).

## Versioning

The `Handshake` / `HandshakeAck` exchange negotiates compatibility. If the agent does not support the client's version it sends `HandshakeAck { accepted: false }` and closes the connection. Both sides must be built from the same `rmon-common` version to guarantee `bincode` schema compatibility.

## Client → Agent Messages

```rust
enum ClientMessage {
    /// Must be the first message sent. No other messages are valid before HandshakeAck.
    Handshake {
        protocol_version: u16,   // current: 1
        client_version:   String,
    },

    /// Request the full list of signals the agent knows about.
    /// Agent replies with one SignalList message.
    ListSignals,

    /// Subscribe to live data for the given signals.
    /// Agent begins sending DataBatch messages for these signals.
    /// `from_ns`: if Some, also request buffered/stored data from that timestamp forward
    ///            before live data begins.
    Subscribe {
        ids:     Vec<SignalId>,
        from_ns: Option<i64>,
    },

    /// Stop live delivery for these signals. Does not affect storage.
    Unsubscribe {
        ids: Vec<SignalId>,
    },

    /// Request stored historical data for a signal.
    /// Agent replies with one or more HistoryChunk messages (done=true on last chunk).
    /// `max_points`: agent may downsample to fit; 0 = no limit.
    QueryHistory {
        id:         SignalId,
        from_ns:    i64,
        to_ns:      i64,
        max_points: u32,
    },

    /// Start recording the given signals. Mode overrides any existing mode for these signals.
    StartRecording {
        ids:  Vec<SignalId>,
        mode: StorageMode,
    },

    /// Stop recording (data already written is kept).
    StopRecording {
        ids: Vec<SignalId>,
    },

    /// Request current agent status (version, uptime, source states).
    GetStatus,
}
```

## Agent → Client Messages

```rust
enum AgentMessage {
    /// Response to Handshake.
    HandshakeAck {
        protocol_version: u16,
        agent_version:    String,
        accepted:         bool,
        reject_reason:    Option<String>,
    },

    /// Full signal list, sent in response to ListSignals.
    /// Sent unsolicited again if the signal list changes (source added/removed).
    SignalList {
        signals: Vec<SignalInfo>,
    },

    /// Live data batch for subscribed signals.
    /// The agent batches samples collected since the last send (target: ~50ms intervals).
    /// Multiple signals may be grouped into one message to reduce framing overhead.
    DataBatch {
        batches: Vec<SignalBatch>,
    },

    /// One chunk of a historical query response.
    HistoryChunk {
        id:      SignalId,
        samples: Vec<Sample>,
        done:    bool,   // true on the final chunk for this query
    },

    /// Acknowledgement of StartRecording / StopRecording.
    RecordingStatus {
        id:        SignalId,
        recording: bool,
        mode:      Option<StorageMode>,
    },

    /// Response to GetStatus.
    StatusReport {
        agent_version: String,
        hostname:      String,
        uptime_secs:   u64,
        sources:       Vec<SourceStatus>,
    },

    /// Unrecoverable error for a specific operation. Connection remains open.
    Error {
        context: String,   // e.g. "QueryHistory" or "Subscribe"
        message: String,
    },
}

struct SignalBatch {
    id:      SignalId,
    samples: Vec<Sample>,
}

struct SourceStatus {
    name:      String,
    source_type: SourceType,
    connected: bool,
    last_sample_ns: Option<i64>,
    error:     Option<String>,
}
```

## Session Lifecycle

```
Client                                  Agent
  │                                       │
  │──── Handshake ────────────────────►   │
  │                                       │
  │◄─── HandshakeAck (accepted=true) ───  │
  │                                       │
  │──── ListSignals ───────────────────►  │
  │◄─── SignalList ─────────────────────  │
  │                                       │
  │──── Subscribe {ids, from_ns} ──────►  │  ←── live data starts flowing
  │◄═══ DataBatch (repeated) ═══════════  │
  │                                       │
  │──── QueryHistory ──────────────────►  │
  │◄─── HistoryChunk (done=false) ──────  │
  │◄─── HistoryChunk (done=false) ──────  │
  │◄─── HistoryChunk (done=true) ───────  │
  │                                       │
  │──── StartRecording ────────────────►  │
  │◄─── RecordingStatus ────────────────  │
  │                                       │
  │  [TCP close / SSH tunnel drops]       │
  │                                       │  ←── agent continues running,
                                               data continues to be stored
```

## Batching Policy

The agent collects samples from all sources in a shared `tokio::broadcast` channel. A per-session task drains the channel and sends `DataBatch` messages at a target interval of **50 ms** (20 Hz UI update rate). If a batch would exceed 1 MiB, it is split and sent immediately.

High-frequency signals (e.g., 10 kHz UDP) are automatically decimated to 10× the batch rate (200 Hz) for live display; full-resolution data is still stored.

## Error Handling

- `Error` messages are non-fatal; the session continues
- If the agent cannot fulfill a `QueryHistory` (signal not found, time range not stored), it sends `Error` then continues
- If the TCP connection drops, the agent discards the session's subscriptions; no other sessions are affected
- The UI reconnects automatically (with exponential backoff) if the tunnel drops