feat: experimental SSE transport for the kernel connection (MARIMO_SERVER_TRANSPORT)

[mscolnick]

Adds an experimental server-sent-events alternative to the /ws kernel
websocket, for deployments behind proxies or services that do not
support WebSockets. Opt in with:

MARIMO_SERVER_TRANSPORT=sse

The kernel channel is already unidirectional (client->kernel traffic
goes over HTTP POST keyed by the Marimo-Session-Id header), so SSE only
replaces the outbound stream. Terminal, LSP, and /ws_sync are untouched;
RTC is auto-disabled under SSE.

Server:

• Extract a transport-agnostic SessionHandler base (session lifecycle,
  kernel-ready, reconnect/resume/replay, TTL close) shared by
  WebSocketHandler and the new SSESessionHandler.
• New GET /sse endpoint streams the same {"op", "data"} wire payloads as
  SSE events, mirrors WebSocket close frames with an in-band close
  event, sends keep-alive comments, and connects to the session lazily
  on first iteration so an unconsumed response never attaches a
  consumer. Pending messages are flushed before a server close.
• Shared SSE framing + HTTP-disconnect helpers in marimo/_server/sse.py,
  reused by the scratchpad /execute stream.

Frontend:

• New SseTransport (fetch + incremental SSE parser) behind the existing
  IConnectionTransport interface, with partysocket-parity retries: the
  budget counts consecutive attempts without a delivered message, gives
  up with TRANSPORT_EXHAUSTED, and never stacks a retry on top of a
  consumer-driven reconnect.
• Auth travels in the Authorization header (like the REST API) instead
  of the URL; classifyCloseEvent now treats MARIMO_UNAUTHORIZED and
  MARIMO_KIOSK_NOT_ALLOWED as terminal.

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
marimo-docs	Ready	Preview, Comment	Jul 9, 2026 6:15pm

[Copilot]

Pull request overview

Adds an experimental Server-Sent Events (SSE) transport for the kernel→frontend stream as an alternative to the existing /ws WebSocket transport, intended for deployments where WebSockets are unavailable or unreliable.

Changes:

• Backend: adds /sse streaming endpoint + shared transport-agnostic session lifecycle (SessionHandler) and shared SSE framing/disconnect helpers.
• Frontend: adds SseTransport implementing the existing connection-transport interface, selectable via server.transport = "sse", and updates close-reason classification for SSE in-band close events.
• Docs/tests/openapi: documents the new config option and adds extensive test coverage for SSE transport behavior and auth.

Reviewed changes

Copilot reviewed 33 out of 33 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
tests/_server/api/endpoints/ws/test_ws_formatter.py	Renames formatter helper to be transport-agnostic (*_for_wire).
tests/_server/api/endpoints/ws/test_ws_connection_validator.py	Adds unit tests for transport-agnostic connection param parsing.
tests/_server/api/endpoints/test_ws.py	Adds regression test for rejecting invalid WS tokens.
tests/_server/api/endpoints/test_ws_message_loop.py	Updates tests to use shared wire-prep helper and validates wire format.
tests/_server/api/endpoints/test_sse.py	Adds end-to-end and unit tests for /sse streaming, auth, reconnect, heartbeat, and framing.
tests/_server/api/endpoints/test_requires_decorator.py	Exempts /sse from generic auth decorator to allow in-band close events.
tests/_server/api/endpoints/test_auto_instantiate.py	Updates auto-instantiate test to use generic HTTPConnection, not WebSocket.
packages/openapi/src/api.ts	Exposes server.transport in generated OpenAPI TS types.
packages/openapi/api.yaml	Documents/enumerates server.transport in OpenAPI schema.
marimo/_smoke_tests/inline_sse_transport.py	Adds a manual smoke-test notebook configured for SSE transport.
marimo/_server/sse.py	Introduces shared SSE framing + HTTP disconnect helper.
marimo/_server/scratchpad.py	Reuses shared SSE formatting utility for scratchpad streaming.
marimo/_server/api/endpoints/ws/ws_session_connector.py	Generalizes connector to accept HTTPConnection + transport-agnostic handler base.
marimo/_server/api/endpoints/ws/ws_message_loop.py	Extracts shared kiosk filtering + wire serialization (prepare_wire_message).
marimo/_server/api/endpoints/ws/ws_formatter.py	Renames/clarifies formatter as wire-format shared by WS + SSE.
marimo/_server/api/endpoints/ws/ws_connection_validator.py	Extracts transport-agnostic parse_connection_params and rejection type.
marimo/_server/api/endpoints/ws/sse_handler.py	Implements SSE session handler streaming kernel messages and close events.
marimo/_server/api/endpoints/ws/session_handler.py	New shared session lifecycle base for WS + SSE handlers.
marimo/_server/api/endpoints/ws_endpoint.py	Adds /sse endpoint and refactors WS handler to subclass SessionHandler.
marimo/_server/api/endpoints/execution.py	Reuses shared HTTP-disconnect helper for scratchpad execute SSE stream.
marimo/_config/config.py	Adds server.transport config option to server config type/docs.
frontend/src/core/websocket/useWebSocket.tsx	Allows selecting transport type and providing headers for SSE fetch.
frontend/src/core/websocket/useMarimoKernelConnection.tsx	Chooses WS vs SSE URL, supplies SSE auth headers, and classifies more terminal close reasons.
frontend/src/core/websocket/transports/sse.ts	Adds fetch-based SSE transport with retry semantics matching existing WS transport.
frontend/src/core/websocket/transports/tests/sse.test.ts	Adds comprehensive unit tests for SSE transport parsing, retries, and close handling.
frontend/src/core/websocket/tests/useWebSocket.test.ts	Tests transport selection (WsTransport vs SseTransport vs BasicTransport).
frontend/src/core/websocket/tests/useMarimoKernelConnection.test.ts	Tests terminal close classification for SSE-delivered close reasons.
frontend/src/core/runtime/runtime.ts	Adds getSseURL() and refactors session search-param construction.
frontend/src/core/runtime/tests/runtime.test.ts	Tests getSseURL() behavior, including not leaking auth token.
frontend/src/core/config/config.ts	Adds atom deriving selected transport type from config.
frontend/src/core/config/config-schema.ts	Adds server.transport to user config schema.
frontend/src/core/config/tests/config-schema.test.ts	Tests default and overridden transport selection atom behavior.
docs/guides/deploying/index.md	Documents experimental SSE transport option and deployment considerations.

[cubic-dev-ai]

All reported issues were addressed across 33 files

Architecture diagram

sequenceDiagram
    participant Browser
    participant Frontend as Frontend (React)
    participant SseTransport as SseTransport (fetch+SSE)
    participant WsTransport as WsTransport (WebSocket)
    participant ConnectTransport as createConnectionTransport
    participant RuntimeManager as RuntimeManager
    participant Config as server.transport config
    participant Server as FastAPI/Starlette
    participant Auth as Auth middleware
    participant SSEHandler as SSESessionHandler
    participant WSHandler as WebSocketHandler
    participant SessionHandler as SessionHandler (base)
    participant SessionConnector as SessionConnector
    participant Session as Session/Kernel
    participant SSEHelpers as sse.py helpers
    participant MessageLoop as prepare_wire_message

    Note over Browser,MessageLoop: SSE Transport Flow (new, experimental)

    Browser->>Config: Configure server.transport = "sse"
    Config-->>Frontend: connectionTransportTypeAtom = "sse"
    
    Frontend->>RuntimeManager: getSessionSearchParams(sessionId)
    Frontend->>RuntimeManager: getSseURL(sessionId)
    RuntimeManager-->>Frontend: https://host/sse?session_id=xxx (no access_token in URL)
    
    Frontend->>ConnectTransport: createConnectionTransport({transportType: "sse"})
    ConnectTransport-->>Frontend: new SseTransport(url, headers)
    
    SseTransport->>SseTransport: reconnect() → reset retry budget, connect()
    SseTransport->>Server: fetch("/sse?session_id=xxx", {Authorization: "Bearer token", Accept: "text/event-stream"})
    
    Server->>Auth: validate_auth(request)
    alt Auth failed
        Auth-->>Server: MARIMO_UNAUTHORIZED
        Server-->>SseTransport: 200 stream with close event {code:3000, reason:"MARIMO_UNAUTHORIZED"}
        SseTransport-->>Frontend: close event → classifyCloseEvent() → terminal
    else Auth success
        Server->>SSEHandler: new SSESessionHandler(request, manager, params, mode)
        Note over SSEHandler: allow_rc=false (RTC disabled under SSE)
        
        SSEHandler->>SessionHandler: _connect_session(request)
        SessionHandler->>SessionConnector: connect()
        alt Connection rejected
            SessionConnector-->>SessionHandler: WebSocketDisconnect (kiosk not allowed, etc.)
            SessionHandler-->>Server: close event in-band
        else Kernel startup error
            SessionConnector-->>SessionHandler: KernelStartupError
            SessionHandler-->>Server: kernel-startup-error message + close event
        else Success
            SessionConnector-->>SessionHandler: (session, connection_type)
        end
        
        SSEHandler->>SSEHandler: async stream() — lazy connection on first iteration
        Note over SSEHandler: Starts heartbeat loop (20s keep-alive comments)
        Note over SSEHandler: Starts disconnect watcher via wait_for_http_disconnect()
        
        loop Stream kernel messages
            Session->>SSEHandler: kernel_message → message_queue
            SSEHandler->>MessageLoop: prepare_wire_message(msg, is_kiosk)
            MessageLoop-->>SSEHandler: {"op": "cell-op", "data": {...}}
            SSEHandler->>SSEHelpers: format_sse_event(data)
            SSEHelpers-->>SSEHandler: "data: ...\n\n"
            SSEHandler-->>Server: yield formatted SSE event
            Server-->>SseTransport: SSE text/event-stream response
        end
        
        opt Server close (shutdown, takeover)
            SessionHandler->>SSEHandler: _request_close(code, reason)
            SSEHandler->>SSEHandler: enqueue _Signal.CLOSE (after pending messages)
            SSEHandler->>SSEHelpers: format_close_event(code, reason)
            SSEHandler-->>Server: yield close event
            Server-->>SseTransport: close event {code, reason}
        end
        
        opt HTTP disconnect
            SSEHandler->>SSEHelpers: wait_for_http_disconnect()
            SSEHelpers-->>SSEHandler: http.disconnect
            SSEHandler->>SSEHandler: enqueue _Signal.DISCONNECT
            SSEHandler->>SSEHandler: finally → cancel tasks, _on_disconnect()
            Server-->>SseTransport: Stream ends without close event
        end
    end

    SseTransport->>SseTransport: readEvents(body) — incremental SSE parser
    alt Received close event
        SseTransport->>SseTransport: close event with code+reason
        SseTransport-->>Frontend: close event
        Frontend->>Frontend: classifyCloseEvent() → terminal or retry
    else Stream ends without close
        SseTransport->>SseTransport: ABNORMAL_CLOSURE (1006)
        alt Retry budget not exhausted
            SseTransport->>SseTransport: increment retryCount, scheduleRetry()
            SseTransport-->>Frontend: close event (retryable)
            SseTransport->>Server: reconnect() with backoff
        else Budget exhausted (MAX_RETRIES without a message)
            SseTransport-->>Frontend: close event with TRANSPORT_EXHAUSTED_REASON
            Note over SseTransport,Frontend: No further retries
        end
    end
    
    opt First message received
        SseTransport->>SseTransport: reset retryCount = 0 (budget restored)
        SseTransport-->>Frontend: message event with data
    end

    Note over Browser,MessageLoop: WebSocket Transport Flow (unchanged, reference)

    Browser->>Config: server.transport = "websocket" (default)
    Config-->>Frontend: connectionTransportTypeAtom = "websocket"
    
    Frontend->>RuntimeManager: getWsURL(sessionId) — token in URL
    Frontend->>ConnectTransport: new WsTransport(url)
    WsTransport->>Server: WebSocket("/ws?session_id=xxx&access_token=xxx")
    Server->>WSHandler: new WebSocketHandler(websocket, manager, params)
    WSHandler->>SessionHandler: inherited session lifecycle (same base)
    WSHandler->>MessageLoop: WebSocketMessageLoop → listen for messages
    MessageLoop-->>WSHandler: prepare_wire_message(msg, is_kiosk)
    WSHandler-->>Browser: WebSocket text frame

    Note over Browser,MessageLoop: Key Behavioral Differences

    Note over SseTransport: Auth via Authorization header (not URL)
    Note over SseTransport: Terminals: MARIMO_UNAUTHORIZED & MARIMO_KIOSK_NOT_ALLOWED
    Note over SSEHandler: RTC auto-disabled (allow_rtc=False)
    Note over SSEHandler: No send() needed (receive-only transport)
    Note over SseTransport: Consumer reconnect() → supersedes internal retry

Loading

<sub>Reply with feedback, questions, or to request a fix.

Re-trigger cubic</sub>

[cubic-dev-ai]

All reported issues were addressed across 5 files (changes from recent commits).

<sub>Reply with feedback, questions, or to request a fix.

Re-trigger cubic</sub>

[dmadisetti]

Woops, sorry thought I approved

[dmadisetti]

Worth mentioning sage maker?

[github-actions]

🚀 Development release published. You may be able to view the changes at https://marimo.app?v=0.23.14-dev56