Module sse 

Server-Sent Events stream handling for GET /event.

opencode publishes all server activity — incremental message parts, tool calls, permission requests, session lifecycle — on a single long-lived SSE stream at GET /event. This module wraps that stream, decoding each data: frame into the generated Event union and exposing the result as an async Stream (and an inherent EventStream::next for consumers that would rather not pull in futures/tokio-stream).

Endpoints

• EVENT_PATH (/event) — events scoped to the server’s working directory. Accepts optional directory and workspace query parameters; attach them with EventStream::from_request if you need them.
• GLOBAL_EVENT_PATH (/global/event) — events from every directory the opencode process manages, each wrapped in a GlobalEvent envelope that carries directory/workspace context. That envelope is a different wire shape than the bare Event this stream decodes, so global frames arrive as StreamEvent::Unknown here; decode them yourself with serde_json::from_value::<GlobalEvent> off the raw serde_json::Value.

Reconnection

The underlying reqwest_eventsource::EventSource reconnects on its own with a capped exponential backoff (configurable via RetryConfig) and replays the Last-Event-ID header so the server can resume. Each successful (re)open surfaces as StreamEvent::Connected. Transient failures that the backoff will retry are swallowed; the stream yields an Err only when the retry budget is exhausted (RetryConfig::max_retries), after which it ends.

Because EventStream::connect performs no network I/O — it only prepares the request — it returns Ok even against an unreachable or permanently-dead server. The initial connection failure (connection refused, DNS error) is treated as just another transient error: with the default RetryConfig::max_retries of None it is retried forever and never surfaces. A naive while let Some(ev) = stream.next().await loop against a server that never comes up therefore blocks indefinitely, yielding neither an event nor an Err. Guard against this by giving RetryConfig::max_retries a bound (so the stream ends with an Err once the budget is spent) and/or by keeping the periodic-poll safety net shown below, which stays live regardless of the stream’s connection state.

Reconcile, don’t trust

SSE is best-effort: frames can be dropped across a reconnect, and the window between your prompt landing and your subscription opening is not covered by the stream. Treat GET /session/{sessionID}/message as the source of truth and the stream as a low-latency hint. The idiomatic shape is to select! the event stream against a periodic poll, and to force a poll every time StreamEvent::Connected arrives (a reconnect means you may have missed frames):

use opencode_codes::sse::{EventStream, StreamEvent};

let mut events = EventStream::connect("http://127.0.0.1:41999")?;
let mut poll = tokio::time::interval(std::time::Duration::from_secs(2));

loop {
    tokio::select! {
        item = events.next() => match item {
            Some(Ok(StreamEvent::Connected)) => {
                // (Re)connected — the stream may have gaps; reconcile now.
                reconcile(&client, &session_id).await?;
            }
            Some(Ok(StreamEvent::Event(ev))) => handle(ev),
            Some(Ok(StreamEvent::Unknown(raw))) => log_unknown(raw),
            Some(Err(e)) => return Err(e),   // retry budget exhausted
            None => break,                   // stream closed
        },
        _ = poll.tick() => {
            // Periodic safety net independent of the stream.
            reconcile(&client, &session_id).await?;
        }
    }
}

Structs

EventStream

A decoded, self-reconnecting subscription to an opencode event stream.

RetryConfig

Reconnection backoff for the SSE stream.

Enums

StreamEvent

A single item yielded by an EventStream.

Constants

EVENT_PATH

Path of the directory-scoped event stream (GET /event).

GLOBAL_EVENT_PATH

Path of the cross-directory event stream (GET /global/event).