From a9645132fcbb33eed4ef3f5a5bfc896f4454043a Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 6 Mar 2026 10:00:07 -0500 Subject: [PATCH] Add Last-Event-ID support to Sessions SSE endpoint Each SSE frame now includes a sequential numeric id: field. Clients can reconnect with the Last-Event-ID header to resume the stream after the last received event, skipping already-processed events. Co-Authored-By: Claude Opus 4.6 --- crates/arc-api/src/demo/mod.rs | 49 +++++++++++++++++-- docs/api-reference/arc-api.yaml | 15 +++++- .../arc-api-client/src/api/sessions-api.ts | 29 ++++++----- 3 files changed, 78 insertions(+), 15 deletions(-) diff --git a/crates/arc-api/src/demo/mod.rs b/crates/arc-api/src/demo/mod.rs index b06dcad51..49f57c36f 100644 --- a/crates/arc-api/src/demo/mod.rs +++ b/crates/arc-api/src/demo/mod.rs @@ -532,10 +532,53 @@ pub async fn send_message_stub( pub async fn session_events_stub( _auth: AuthenticatedService, State(_state): State>, - Path(_id): Path, + headers: axum::http::HeaderMap, + Path(id): Path, ) -> Response { - // Return an empty SSE-like response - ApiError::new(StatusCode::GONE, "Event stream closed.").into_response() + use axum::response::sse::{Event, Sse}; + + let session = match sessions::detail(&id) { + Some(s) => s, + None => return ApiError::not_found("Session not found.").into_response(), + }; + + let last_event_id: Option = headers + .get("Last-Event-ID") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse().ok()); + + let mut events: Vec> = Vec::new(); + let mut seq: usize = 0; + + for turn in &session.turns { + let (event_type, data) = match turn { + arc_types::SessionTurn::UserTurn(_) => continue, + arc_types::SessionTurn::AssistantTurn(t) => { + ("assistant_turn", serde_json::to_string(t).unwrap()) + } + arc_types::SessionTurn::ToolTurn(t) => { + ("tool_turn", serde_json::to_string(t).unwrap()) + } + }; + + if last_event_id.is_none() || seq > last_event_id.unwrap() { + events.push(Ok(Event::default() + .id(seq.to_string()) + .event(event_type) + .data(data))); + } + seq += 1; + } + + // Append done event + if last_event_id.is_none() || seq > last_event_id.unwrap() { + events.push(Ok(Event::default() + .id(seq.to_string()) + .event("done") + .data("{}"))); + } + + Sse::new(tokio_stream::iter(events)).into_response() } // ── Insights ─────────────────────────────────────────────────────────── diff --git a/docs/api-reference/arc-api.yaml b/docs/api-reference/arc-api.yaml index 77e027889..cca110f35 100644 --- a/docs/api-reference/arc-api.yaml +++ b/docs/api-reference/arc-api.yaml @@ -814,6 +814,9 @@ paths: description: | Opens a server-sent event (SSE) stream for real-time session updates. + Each SSE frame includes a sequential numeric `id:` field that supports + resumption via the `Last-Event-ID` request header. + The stream emits the following SSE event types: - `event: assistant_turn` — data: `AssistantTurn` JSON object @@ -821,9 +824,19 @@ paths: - `event: done` — data: `{}` (stream complete) - `event: error` — data: `{"message": "..."}` (error occurred) - Each SSE frame has an `event:` line (the event type) and a `data:` line (the JSON payload). + Each SSE frame has an `id:` line (sequential integer), an `event:` line (the event type), + and a `data:` line (the JSON payload). parameters: - $ref: "#/components/parameters/SessionId" + - name: Last-Event-ID + in: header + required: false + description: > + SSE reconnection header. When provided, the server resumes the + stream after the event with this ID. IDs are 0-based sequential + integers assigned to each emitted SSE frame. + schema: + type: string responses: "200": description: Server-sent event stream diff --git a/packages/arc-api-client/src/api/sessions-api.ts b/packages/arc-api-client/src/api/sessions-api.ts index a56e309eb..19ad9589a 100644 --- a/packages/arc-api-client/src/api/sessions-api.ts +++ b/packages/arc-api-client/src/api/sessions-api.ts @@ -217,13 +217,14 @@ export const SessionsApiAxiosParamCreator = function (configuration?: Configurat }; }, /** - * Opens a server-sent event (SSE) stream for real-time session updates. The stream emits the following SSE event types: - `event: assistant_turn` — data: `AssistantTurn` JSON object - `event: tool_turn` — data: `ToolTurn` JSON object - `event: done` — data: `{}` (stream complete) - `event: error` — data: `{\"message\": \"...\"}` (error occurred) Each SSE frame has an `event:` line (the event type) and a `data:` line (the JSON payload). + * Opens a server-sent event (SSE) stream for real-time session updates. Each SSE frame includes a sequential numeric `id:` field that supports resumption via the `Last-Event-ID` request header. The stream emits the following SSE event types: - `event: assistant_turn` — data: `AssistantTurn` JSON object - `event: tool_turn` — data: `ToolTurn` JSON object - `event: done` — data: `{}` (stream complete) - `event: error` — data: `{\"message\": \"...\"}` (error occurred) Each SSE frame has an `id:` line (sequential integer), an `event:` line (the event type), and a `data:` line (the JSON payload). * @summary Stream Session Events * @param {string} id Unique session identifier. + * @param {string} [lastEventID] SSE reconnection header. When provided, the server resumes the stream after the event with this ID. IDs are 0-based sequential integers assigned to each emitted SSE frame. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - streamSessionEvents: async (id: string, options: RawAxiosRequestConfig = {}): Promise => { + streamSessionEvents: async (id: string, lastEventID?: string, options: RawAxiosRequestConfig = {}): Promise => { // verify required parameter 'id' is not null or undefined assertParamExists('streamSessionEvents', 'id', id) const localVarPath = `/sessions/{id}/events` @@ -248,6 +249,9 @@ export const SessionsApiAxiosParamCreator = function (configuration?: Configurat localVarHeaderParameter['Accept'] = 'text/event-stream,application/json'; + if (lastEventID != null) { + localVarHeaderParameter['Last-Event-ID'] = String(lastEventID); + } setSearchParams(localVarUrlObj, localVarQueryParameter); let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {}; localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers}; @@ -321,14 +325,15 @@ export const SessionsApiFp = function(configuration?: Configuration) { return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); }, /** - * Opens a server-sent event (SSE) stream for real-time session updates. The stream emits the following SSE event types: - `event: assistant_turn` — data: `AssistantTurn` JSON object - `event: tool_turn` — data: `ToolTurn` JSON object - `event: done` — data: `{}` (stream complete) - `event: error` — data: `{\"message\": \"...\"}` (error occurred) Each SSE frame has an `event:` line (the event type) and a `data:` line (the JSON payload). + * Opens a server-sent event (SSE) stream for real-time session updates. Each SSE frame includes a sequential numeric `id:` field that supports resumption via the `Last-Event-ID` request header. The stream emits the following SSE event types: - `event: assistant_turn` — data: `AssistantTurn` JSON object - `event: tool_turn` — data: `ToolTurn` JSON object - `event: done` — data: `{}` (stream complete) - `event: error` — data: `{\"message\": \"...\"}` (error occurred) Each SSE frame has an `id:` line (sequential integer), an `event:` line (the event type), and a `data:` line (the JSON payload). * @summary Stream Session Events * @param {string} id Unique session identifier. + * @param {string} [lastEventID] SSE reconnection header. When provided, the server resumes the stream after the event with this ID. IDs are 0-based sequential integers assigned to each emitted SSE frame. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - async streamSessionEvents(id: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { - const localVarAxiosArgs = await localVarAxiosParamCreator.streamSessionEvents(id, options); + async streamSessionEvents(id: string, lastEventID?: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + const localVarAxiosArgs = await localVarAxiosParamCreator.streamSessionEvents(id, lastEventID, options); const localVarOperationServerIndex = configuration?.serverIndex ?? 0; const localVarOperationServerBasePath = operationServerMap['SessionsApi.streamSessionEvents']?.[localVarOperationServerIndex]?.url; return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); @@ -385,14 +390,15 @@ export const SessionsApiFactory = function (configuration?: Configuration, baseP return localVarFp.sendSessionMessage(id, sendMessageRequest, options).then((request) => request(axios, basePath)); }, /** - * Opens a server-sent event (SSE) stream for real-time session updates. The stream emits the following SSE event types: - `event: assistant_turn` — data: `AssistantTurn` JSON object - `event: tool_turn` — data: `ToolTurn` JSON object - `event: done` — data: `{}` (stream complete) - `event: error` — data: `{\"message\": \"...\"}` (error occurred) Each SSE frame has an `event:` line (the event type) and a `data:` line (the JSON payload). + * Opens a server-sent event (SSE) stream for real-time session updates. Each SSE frame includes a sequential numeric `id:` field that supports resumption via the `Last-Event-ID` request header. The stream emits the following SSE event types: - `event: assistant_turn` — data: `AssistantTurn` JSON object - `event: tool_turn` — data: `ToolTurn` JSON object - `event: done` — data: `{}` (stream complete) - `event: error` — data: `{\"message\": \"...\"}` (error occurred) Each SSE frame has an `id:` line (sequential integer), an `event:` line (the event type), and a `data:` line (the JSON payload). * @summary Stream Session Events * @param {string} id Unique session identifier. + * @param {string} [lastEventID] SSE reconnection header. When provided, the server resumes the stream after the event with this ID. IDs are 0-based sequential integers assigned to each emitted SSE frame. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - streamSessionEvents(id: string, options?: RawAxiosRequestConfig): AxiosPromise { - return localVarFp.streamSessionEvents(id, options).then((request) => request(axios, basePath)); + streamSessionEvents(id: string, lastEventID?: string, options?: RawAxiosRequestConfig): AxiosPromise { + return localVarFp.streamSessionEvents(id, lastEventID, options).then((request) => request(axios, basePath)); }, }; }; @@ -448,14 +454,15 @@ export class SessionsApi extends BaseAPI { } /** - * Opens a server-sent event (SSE) stream for real-time session updates. The stream emits the following SSE event types: - `event: assistant_turn` — data: `AssistantTurn` JSON object - `event: tool_turn` — data: `ToolTurn` JSON object - `event: done` — data: `{}` (stream complete) - `event: error` — data: `{\"message\": \"...\"}` (error occurred) Each SSE frame has an `event:` line (the event type) and a `data:` line (the JSON payload). + * Opens a server-sent event (SSE) stream for real-time session updates. Each SSE frame includes a sequential numeric `id:` field that supports resumption via the `Last-Event-ID` request header. The stream emits the following SSE event types: - `event: assistant_turn` — data: `AssistantTurn` JSON object - `event: tool_turn` — data: `ToolTurn` JSON object - `event: done` — data: `{}` (stream complete) - `event: error` — data: `{\"message\": \"...\"}` (error occurred) Each SSE frame has an `id:` line (sequential integer), an `event:` line (the event type), and a `data:` line (the JSON payload). * @summary Stream Session Events * @param {string} id Unique session identifier. + * @param {string} [lastEventID] SSE reconnection header. When provided, the server resumes the stream after the event with this ID. IDs are 0-based sequential integers assigned to each emitted SSE frame. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - public streamSessionEvents(id: string, options?: RawAxiosRequestConfig) { - return SessionsApiFp(this.configuration).streamSessionEvents(id, options).then((request) => request(this.axios, this.basePath)); + public streamSessionEvents(id: string, lastEventID?: string, options?: RawAxiosRequestConfig) { + return SessionsApiFp(this.configuration).streamSessionEvents(id, lastEventID, options).then((request) => request(this.axios, this.basePath)); } }