Reverse Websocket Channel #

The framework’s reverse channel lets a control plane (e.g. INFINI Console) call HTTP APIs on agents that sit behind NAT or firewalls — without opening any inbound port on the agent side. The connection direction is inverted: the agent dials out and holds a long-lived websocket; the control plane pushes requests down that connection and the agent executes them locally.

┌────────────┐  ① agent dials out (outbound-only)   ┌─────────────────┐
│   Agent    │ ──────────────────────────────────▶  │  Control Plane   │
│ (NAT/fire- │      long-lived websocket            │  (Console)       │
│  walled)   │                                      │                  │
│            │  ② control plane sends request        │                  │
│            │ ◀──────────────────────────────────  │  ProxyRequest()  │
│  executes  │      RequestMessage payload          │                  │
│  locally   │                                      │                  │
│            │  ③ agent streams response back        │                  │
│            │ ──────────────────────────────────▶  │  (chunked)       │
└────────────┘      ResponseMessage payloads         └─────────────────┘

The package lives in core/api/websocket/reverse and provides two layers:

• Protocol (protocol.go) — the wire format: message structs, framing helpers, and payload (de)serialization.
• Session manager (manager.go) — the control-plane state machine: session registration/activation, request multiplexing, timeouts, disconnect handling, and reconnect windows.

When to Use It #

Architecture #

Roles #

• Peer — the agent-side endpoint. Identified by a peerID (typically the instance ID). Holds the outbound websocket connection.
• Control plane — the server-side consumer of this package. Owns a SessionManager, calls ProxyRequest for each logical request, and feeds inbound websocket payloads into the manager via the Handle*Payload methods.
• Session — one established websocket connection between a peer and the control plane. Identified by a sessionID issued by the control plane (RegisterPendingSession) and confirmed by the agent’s Hello.

Connection Lifecycle #

Control Plane                                Agent (peer)
     │                                            │
     │ (out-of-band: hands sessionID+peerID       │
     │  to the agent, e.g. via install script     │
     │  or managed config)                        │
     │                                            │
     │ ◀──── websocket connect ────────────────── │ ① agent dials out
     │                                            │
     │ ◀──── HELLO {sessionID, peerID} ────────── │ ② handshake
     │  ActivateSession()                         │
     │      (replaces any stale session           │
     │       for the same peer)                   │
     │                                            │
     │ ──── REQUEST {id, method, path, body} ───▶ │ ③ proxied call
     │      (ProxyRequest blocks)                 │    agent executes
     │                                            │    locally
     │ ◀──── RESPONSE chunks {id, chunk} ──────── │ ④ streamed back
     │ ◀──── RESPONSE {id, done, status} ───────── │    (base64)
     │                                            │
     │ ◀──── (ws close) ────────────────────────── │ disconnect
     │  OnDisconnect():                            │
     │   - drops session state                    │
     │   - fails in-flight requests               │
     │      (ErrDisconnected)                     │

Wire Protocol #

All messages travel as websocket text payloads produced by the framing helpers in protocol.go. A frame is a command prefix followed by a JSON body.

Message Types #

// HelloMessage — sent by the agent right after connecting.
type HelloMessage struct {
    SessionID string
    PeerID    string
}

// RequestMessage — control plane → agent: one proxied HTTP request.
type RequestMessage struct {
    RequestID   string
    PeerID      string
    Method      string
    Path        string
    Headers     http.Header
    AccessToken string      // extracted from the caller's Bearer header
    // body carried via SetBody/BodyBytes (base64)
}

// ResponseMessage — agent → control plane: chunked response stream.
type ResponseMessage struct {
    RequestID string
    PeerID    string
    Chunk     string        // base64-encoded body chunk
    Done      bool          // final frame
    Status    int           // HTTP status (on Done; 0 is not valid —
                            // a Done frame without Status is only legal with Error)
    Error     string        // execution failure (on Done): the agent could not
                            // produce an HTTP response at all; mutually exclusive
                            // with Status
}

Framing Helpers #

Constants #

HelloCommand     // HELLO frame prefix
RequestCommand   // REQUEST frame prefix
ResponseCommand  // RESPONSE frame prefix
HeaderPeerID     // header key carrying the peer ID on the dial-out request

Session Manager API #

The control plane creates one manager and shares it across connections:

manager := reverse.NewSessionManager(reverse.ManagerOptions{
    DefaultTimeout:   30 * time.Second,   // per logical request
    MaxResponseBytes: 8 * 1024 * 1024,    // 8 MiB response cap
    ReconnectWait:    6 * time.Second,    // retry window after disconnect
    ReconnectPoll:    200 * time.Millisecond,
})

Session Bookkeeping #

Proxying Requests #

res, err := manager.ProxyRequest(
    peerID,                    // target agent
    &util.Request{             // logical request
        Method: http.MethodGet,
        Path:   "/_node/stats",
        Context: ctx,
    },
    headers,                   // headers to forward
    send,                      // func(sessionID, payload string) error
    nil,                       // optional: unmarshal JSON body into this
)

ProxyRequest is the core call:

1. Resolves the peer’s active session; ErrNotConnected if none.
2. Builds a RequestMessage (the caller’s Authorization: Bearer ... header is lifted into AccessToken), registers a pending response, and pushes the frame via the send callback (your websocket write path).
3. Blocks until the response completes, the context/deadline expires, or the connection drops.
4. Reconnect retry: on a recoverable error (ErrDisconnected/ErrNotConnected) it waits up to ReconnectWait (polling every ReconnectPoll) for the agent to re-establish its session, then retries once. This makes brief agent restarts transparent to callers.
5. Returns *util.Result{Body, StatusCode}; a non-200 status yields both the result and an error.

While a request is in flight, the agent streams ResponseMessage chunks; feed each one into the manager from your websocket read loop:

manager.HandleResponsePayload(frameBody) // assembles chunks, enforces the cap

Error Values #

A Done frame only means “the response stream ends here” — success or failure is expressed exclusively by Status (a real HTTP status from the locally executed request) or Error (no HTTP response exists). The manager rejects a bare Done with neither field rather than defaulting to 200, so agent-side bugs surface immediately instead of masquerading as successful empty responses. Agent implementations should always terminate via WriteChunkedResponse (which validates the status) or WriteFailureResponse.

Security Model #

The reverse channel is a transport, not an authenticator. Compose it with the framework’s existing security layers:

• Control plane → agent authorization: the agent’s own API authentication applies to the proxied request. The AccessToken field carries the caller’s bearer token end-to-end, so the agent’s API layer can validate it exactly as if the call were direct.
• Agent → control plane: the outbound websocket rides the agent’s existing authenticated channel to the control plane (e.g. Console’s managed websocket with instance credentials).
• Session binding: the handshake check (RegisterPendingSession → HandleHelloPayload) prevents an agent from claiming another peer’s session slot.

Do not use the channel to move secrets in bulk — AccessToken is the only credential field with dedicated handling; everything else is treated as opaque request data.

Integration Recipe (Control Plane) #

// 1. One manager per process; zero-value options take the defaults
//    documented in the Tuning section below.
reverseManager := reverse.NewSessionManager(reverse.ManagerOptions{})

// 2. Issue session credentials when provisioning an agent.
sessionID := util.GetUUID()
reverseManager.RegisterPendingSession(sessionID, agentInstanceID)
// hand {sessionID, peerID: agentInstanceID} to the agent
// (install script, managed config, etc.)

// 3. Wire your websocket server: on each connection, read frames and
//    dispatch by prefix.
api.HandleWebSocketCommand(reverse.HelloCommand, "agent hello", func(...) {
    _ = reverseManager.HandleHelloPayload(body)
})
api.HandleWebSocketCommand(reverse.ResponseCommand, "agent response", func(...) {
    _ = reverseManager.HandleResponsePayload(body)
})
// track the connection's sessionID to call OnDisconnect on close.

// 4. Call agents.
res, err := reverseManager.ProxyRequest(
    agentInstanceID, req, headers,
    func(sessionID, payload string) error {
        return websocketHub.SendPrivateMessage(sessionID, payload)
    }, nil,
)
if reverse.IsRecoverableError(err) {
    // optionally surface "agent temporarily unreachable" to the UI
}

Tuning #

Relationship to Other Framework Facilities #

• Managed config sync (modules/configs) — separate channel; agents pull configs and heartbeat over HTTP. The reverse channel complements it for on-demand RPC.
• OTLP transport (otlp_export / gateway intake) — the data path for log shipping. The reverse channel is the control path.
• Console’s agent management — the reference consumer: stats fetching, log tailing, and UI websocket proxying all ride ProxyRequest / the session manager.