diff --git a/crates/goose-server/src/lib.rs b/crates/goose-server/src/lib.rs index 4403ef7f85..ab35c400c2 100644 --- a/crates/goose-server/src/lib.rs +++ b/crates/goose-server/src/lib.rs @@ -3,6 +3,7 @@ pub mod configuration; pub mod error; pub mod openapi; pub mod routes; +pub mod session_event_bus; pub mod state; pub mod tls; pub mod tunnel; diff --git a/crates/goose-server/src/main.rs b/crates/goose-server/src/main.rs index 25bf34246d..66a203694b 100644 --- a/crates/goose-server/src/main.rs +++ b/crates/goose-server/src/main.rs @@ -4,6 +4,7 @@ mod error; mod logging; mod openapi; mod routes; +mod session_event_bus; mod state; mod tunnel; diff --git a/crates/goose-server/src/openapi.rs b/crates/goose-server/src/openapi.rs index 9bc09b7ebd..eec8aeefe2 100644 --- a/crates/goose-server/src/openapi.rs +++ b/crates/goose-server/src/openapi.rs @@ -435,6 +435,9 @@ derive_utoipa!(Icon as IconSchema); super::routes::agent::update_session, super::routes::action_required::confirm_tool_action, super::routes::reply::reply, + super::routes::session_events::session_events, + super::routes::session_events::session_reply, + super::routes::session_events::session_cancel, super::routes::session::list_sessions, super::routes::session::search_sessions, super::routes::session::get_session, @@ -523,6 +526,9 @@ derive_utoipa!(Icon as IconSchema); goose::prompt_template::Template, super::routes::action_required::ConfirmToolActionRequest, super::routes::reply::ChatRequest, + super::routes::session_events::SessionReplyRequest, + super::routes::session_events::SessionReplyResponse, + super::routes::session_events::CancelRequest, super::routes::session::ImportSessionRequest, super::routes::session::SessionListResponse, super::routes::session::UpdateSessionNameRequest, diff --git a/crates/goose-server/src/routes/mod.rs b/crates/goose-server/src/routes/mod.rs index 22d4e3e6af..c440a0d991 100644 --- a/crates/goose-server/src/routes/mod.rs +++ b/crates/goose-server/src/routes/mod.rs @@ -14,6 +14,7 @@ pub mod reply; pub mod sampling; pub mod schedule; pub mod session; +pub mod session_events; pub mod setup; pub mod status; pub mod telemetry; @@ -44,5 +45,6 @@ pub fn configure(state: Arc, secret_key: String) -> Rout .merge(gateway::routes(state.clone())) .merge(mcp_ui_proxy::routes(secret_key.clone())) .merge(mcp_app_proxy::routes(secret_key)) + .merge(session_events::routes(state.clone())) .merge(sampling::routes(state)) } diff --git a/crates/goose-server/src/routes/reply.rs b/crates/goose-server/src/routes/reply.rs index 70eafd3353..7df27b434c 100644 --- a/crates/goose-server/src/routes/reply.rs +++ b/crates/goose-server/src/routes/reply.rs @@ -29,7 +29,7 @@ use tokio::time::timeout; use tokio_stream::wrappers::ReceiverStream; use tokio_util::sync::CancellationToken; -fn track_tool_telemetry(content: &MessageContent, all_messages: &[Message]) { +pub fn track_tool_telemetry(content: &MessageContent, all_messages: &[Message]) { match content { MessageContent::ToolRequest(tool_request) => { if let Ok(tool_call) = &tool_request.tool_call { @@ -123,7 +123,7 @@ impl IntoResponse for SseResponse { } } -#[derive(Debug, Serialize, utoipa::ToSchema)] +#[derive(Debug, Clone, Serialize, utoipa::ToSchema)] #[serde(tag = "type")] pub enum MessageEvent { Message { @@ -149,10 +149,15 @@ pub enum MessageEvent { UpdateConversation { conversation: Conversation, }, + /// Sent at the start of an SSE stream to inform the client about + /// in-flight requests it can reattach to. + ActiveRequests { + request_ids: Vec, + }, Ping, } -async fn get_token_state(session_manager: &SessionManager, session_id: &str) -> TokenState { +pub async fn get_token_state(session_manager: &SessionManager, session_id: &str) -> TokenState { session_manager .get_session(session_id, false) .await diff --git a/crates/goose-server/src/routes/session.rs b/crates/goose-server/src/routes/session.rs index b6a61c21fc..b076dbc8bc 100644 --- a/crates/goose-server/src/routes/session.rs +++ b/crates/goose-server/src/routes/session.rs @@ -296,6 +296,13 @@ async fn delete_session( } })?; + // Cancel any in-flight replies before dropping the bus, so spawned + // agent tasks stop consuming tokens for a deleted session. + if let Some(bus) = state.get_event_bus(&session_id).await { + bus.cancel_all_requests().await; + } + state.remove_event_bus(&session_id).await; + Ok(StatusCode::OK) } diff --git a/crates/goose-server/src/routes/session_events.rs b/crates/goose-server/src/routes/session_events.rs new file mode 100644 index 0000000000..7f69aa97ed --- /dev/null +++ b/crates/goose-server/src/routes/session_events.rs @@ -0,0 +1,615 @@ +use crate::routes::errors::ErrorResponse; +use crate::routes::reply::{get_token_state, track_tool_telemetry, MessageEvent}; +use crate::state::AppState; +use axum::{ + extract::{DefaultBodyLimit, Path, State}, + http::{self, HeaderMap}, + response::IntoResponse, + routing::{get, post}, + Json, Router, +}; +use bytes::Bytes; +use futures::{stream::StreamExt, Stream}; +use goose::agents::{AgentEvent, SessionConfig}; +use goose::conversation::message::Message; +use goose::conversation::Conversation; +use serde::{Deserialize, Serialize}; +use std::{ + convert::Infallible, + pin::Pin, + sync::Arc, + task::{Context, Poll}, + time::Duration, +}; +use tokio::sync::mpsc; +use tokio::time::timeout; +use tokio_stream::wrappers::ReceiverStream; + +// ── Request / Response types ──────────────────────────────────────────── + +#[derive(Debug, Deserialize, Serialize, utoipa::ToSchema)] +pub struct SessionReplyRequest { + /// Client-generated UUIDv7 identifying this request. + pub request_id: String, + pub user_message: Message, + #[serde(default)] + pub override_conversation: Option>, + pub recipe_name: Option, + pub recipe_version: Option, +} + +#[derive(Debug, Serialize, utoipa::ToSchema)] +pub struct SessionReplyResponse { + pub request_id: String, +} + +#[derive(Debug, Deserialize, Serialize, utoipa::ToSchema)] +pub struct CancelRequest { + pub request_id: String, +} + +// ── SSE Event Stream Response ─────────────────────────────────────────── + +/// An SSE response that includes `id:` lines for Last-Event-ID reconnection. +pub struct SseEventStream { + rx: ReceiverStream, +} + +impl SseEventStream { + fn new(rx: ReceiverStream) -> Self { + Self { rx } + } +} + +impl Stream for SseEventStream { + type Item = Result; + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.rx) + .poll_next(cx) + .map(|opt| opt.map(|s| Ok(Bytes::from(s)))) + } +} + +impl IntoResponse for SseEventStream { + fn into_response(self) -> axum::response::Response { + let body = axum::body::Body::from_stream(self); + http::Response::builder() + .header("Content-Type", "text/event-stream") + .header("Cache-Control", "no-cache") + .header("Connection", "keep-alive") + .body(body) + .unwrap() + } +} + +// ── Helpers ───────────────────────────────────────────────────────────── + +fn format_sse_event(seq: u64, json: &str) -> String { + format!("id: {}\ndata: {}\n\n", seq, json) +} + +fn serialize_session_event(seq: u64, request_id: Option<&str>, event: &MessageEvent) -> String { + // Build JSON payload: { request_id?: string, ...event_fields } + // We flatten request_id into the event JSON. + let mut event_json = serde_json::to_value(event).unwrap_or_else( + |e| serde_json::json!({"type": "Error", "error": format!("Serialization error: {}", e)}), + ); + + if let Some(rid) = request_id { + if let serde_json::Value::Object(ref mut map) = event_json { + // Always insert chat_request_id for routing (the chat UUID that + // the frontend registered its listener under). + map.insert( + "chat_request_id".to_string(), + serde_json::Value::String(rid.to_string()), + ); + // Also set request_id if the event doesn't already carry one + // (e.g. Notification events have their own request_id for tool-call matching) + map.entry("request_id") + .or_insert_with(|| serde_json::Value::String(rid.to_string())); + } + } + + let json_str = serde_json::to_string(&event_json).unwrap_or_default(); + format_sse_event(seq, &json_str) +} + +// ── GET /sessions/{id}/events ─────────────────────────────────────────── + +#[utoipa::path( + get, + path = "/sessions/{id}/events", + params( + ("id" = String, Path, description = "Session ID"), + ), + responses( + (status = 200, description = "SSE event stream", + body = MessageEvent, + content_type = "text/event-stream"), + (status = 404, description = "Session not found"), + ) +)] +pub async fn session_events( + State(state): State>, + Path(session_id): Path, + headers: HeaderMap, +) -> Result { + // Validate the session exists before creating an event bus. + state + .session_manager() + .get_session(&session_id, false) + .await + .map_err(|_| axum::http::StatusCode::NOT_FOUND)?; + + let last_event_id: Option = headers + .get("Last-Event-ID") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse().ok()); + + let bus = state.get_or_create_event_bus(&session_id).await; + + let (replay, replay_max_seq, mut live_rx) = match bus.subscribe(last_event_id).await { + Ok(result) => result, + Err(_) => { + // Client's Last-Event-ID has been evicted from the replay buffer. + // Send a single error event so the client knows to reload. + let (tx, rx) = mpsc::channel::(1); + let stream = ReceiverStream::new(rx); + let seq = 0; + let error_event = MessageEvent::Error { + error: "Client too far behind — reload conversation".to_string(), + }; + let frame = serialize_session_event(seq, None, &error_event); + tokio::spawn(async move { + let _ = tx.send(frame).await; + }); + return Ok(SseEventStream::new(stream)); + } + }; + + let (tx, rx) = mpsc::channel::(256); + let stream = ReceiverStream::new(rx); + let task_bus = bus.clone(); + + tokio::spawn(async move { + let bus = task_bus; + + // Notify the client about any in-flight requests BEFORE replay + // so it can register event handlers before replayed events arrive. + // Emitted without an SSE `id:` field so it doesn't regress the + // client's Last-Event-ID cursor. + let active_ids = bus.active_request_ids().await; + if !active_ids.is_empty() { + let event = MessageEvent::ActiveRequests { + request_ids: active_ids, + }; + let json_str = serde_json::to_string(&serde_json::to_value(&event).unwrap_or_default()) + .unwrap_or_default(); + let frame = format!("data: {}\n\n", json_str); + if tx.send(frame).await.is_err() { + return; + } + } + + // Send replayed events + for event in &replay { + let frame = + serialize_session_event(event.seq, event.request_id.as_deref(), &event.event); + if tx.send(frame).await.is_err() { + return; + } + } + + // Send live events + heartbeat pings + let mut heartbeat_interval = tokio::time::interval(Duration::from_millis(500)); + // Heartbeat uses a local counter — not stored in the replay buffer + let mut heartbeat_seq = 0u64; + + loop { + tokio::select! { + _ = heartbeat_interval.tick() => { + // Send heartbeat directly without publishing to the bus, + // so pings don't evict real events from the replay buffer. + // Use a comment-style SSE id so it won't interfere with Last-Event-ID. + let frame = format!(": ping {}\n\n", heartbeat_seq); + heartbeat_seq += 1; + if tx.send(frame).await.is_err() { + return; + } + } + result = live_rx.recv() => { + match result { + Ok(event) => { + // Skip events already covered by replay to avoid duplicates + // at the replay/live handoff boundary. + if event.seq <= replay_max_seq { + continue; + } + let frame = serialize_session_event( + event.seq, + event.request_id.as_deref(), + &event.event, + ); + if tx.send(frame).await.is_err() { + return; + } + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + tracing::warn!("SSE subscriber lagged by {} events, closing stream so client reconnects with Last-Event-ID", n); + // Close the stream so the client reconnects and + // replays missed events from the buffer. + return; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => { + return; + } + } + } + } + } + }); + + Ok(SseEventStream::new(stream)) +} + +// ── POST /sessions/{id}/reply ─────────────────────────────────────────── + +#[utoipa::path( + post, + path = "/sessions/{id}/reply", + params( + ("id" = String, Path, description = "Session ID"), + ), + request_body = SessionReplyRequest, + responses( + (status = 200, description = "Request accepted", + body = SessionReplyResponse), + (status = 400, description = "Invalid request"), + (status = 404, description = "Session not found"), + (status = 424, description = "Agent not initialized"), + (status = 500, description = "Internal server error"), + ) +)] +pub async fn session_reply( + State(state): State>, + Path(session_id): Path, + Json(request): Json, +) -> Result, ErrorResponse> { + let request_id = request.request_id.clone(); + + // Validate request_id is a valid UUID + if uuid::Uuid::parse_str(&request_id).is_err() { + return Err(ErrorResponse::bad_request( + "request_id must be a valid UUID", + )); + } + + // Validate session exists before allocating a bus/registering work + state + .session_manager() + .get_session(&session_id, false) + .await + .map_err(|_| ErrorResponse::not_found(format!("Session {} not found", session_id)))?; + + let session_start = std::time::Instant::now(); + + tracing::info!( + monotonic_counter.goose.session_starts = 1, + session_type = "app", + interface = "ui", + "Session started" + ); + + if let Some(recipe_name) = request.recipe_name.clone() { + if state.mark_recipe_run_if_absent(&session_id).await { + let recipe_version = request + .recipe_version + .clone() + .unwrap_or_else(|| "unknown".to_string()); + + tracing::info!( + monotonic_counter.goose.recipe_runs = 1, + recipe_name = %recipe_name, + recipe_version = %recipe_version, + session_type = "app", + interface = "ui", + "Recipe execution started" + ); + } + } + + let bus = state.get_or_create_event_bus(&session_id).await; + let cancel_token = bus.register_request(request_id.clone()).await; + + let user_message = request.user_message; + let override_conversation = request.override_conversation; + + let task_state = state.clone(); + let task_session_id = session_id.clone(); + let task_request_id = request_id.clone(); + let task_cancel = cancel_token.clone(); + let task_bus = bus.clone(); + + drop(tokio::spawn(async move { + let publish = |rid: Option, event: MessageEvent| { + let bus = task_bus.clone(); + async move { + bus.publish(rid, event).await; + } + }; + + let agent = match task_state.get_agent(task_session_id.clone()).await { + Ok(agent) => agent, + Err(e) => { + tracing::error!("Failed to get session agent: {}", e); + publish( + Some(task_request_id.clone()), + MessageEvent::Error { + error: format!("Failed to get session agent: {}", e), + }, + ) + .await; + task_bus.cleanup_request(&task_request_id).await; + return; + } + }; + + let session = match task_state + .session_manager() + .get_session(&task_session_id, true) + .await + { + Ok(metadata) => metadata, + Err(e) => { + tracing::error!("Failed to read session for {}: {}", task_session_id, e); + publish( + Some(task_request_id.clone()), + MessageEvent::Error { + error: format!("Failed to read session: {}", e), + }, + ) + .await; + task_bus.cleanup_request(&task_request_id).await; + return; + } + }; + + let session_config = SessionConfig { + id: task_session_id.clone(), + schedule_id: session.schedule_id.clone(), + max_turns: None, + retry_config: None, + }; + + let mut all_messages = match override_conversation { + Some(history) => { + let conv = Conversation::new_unvalidated(history); + if let Err(e) = task_state + .session_manager() + .replace_conversation(&task_session_id, &conv) + .await + { + tracing::warn!( + "Failed to replace session conversation for {}: {}", + task_session_id, + e + ); + } + conv + } + None => session.conversation.unwrap_or_default(), + }; + all_messages.push(user_message.clone()); + + let mut stream = match agent + .reply( + user_message.clone(), + session_config, + Some(task_cancel.clone()), + ) + .await + { + Ok(stream) => stream, + Err(e) => { + tracing::error!("Failed to start reply stream: {:?}", e); + publish( + Some(task_request_id.clone()), + MessageEvent::Error { + error: e.to_string(), + }, + ) + .await; + task_bus.cleanup_request(&task_request_id).await; + return; + } + }; + + loop { + tokio::select! { + _ = task_cancel.cancelled() => { + tracing::info!("Agent task cancelled for request {}", task_request_id); + break; + } + response = timeout(Duration::from_millis(500), stream.next()) => { + match response { + Ok(Some(Ok(AgentEvent::Message(message)))) => { + for content in &message.content { + track_tool_telemetry(content, all_messages.messages()); + } + all_messages.push(message.clone()); + let token_state = get_token_state( + task_state.session_manager(), + &task_session_id, + ) + .await; + publish( + Some(task_request_id.clone()), + MessageEvent::Message { + message, + token_state, + }, + ) + .await; + } + Ok(Some(Ok(AgentEvent::HistoryReplaced(new_messages)))) => { + all_messages = new_messages.clone(); + publish( + Some(task_request_id.clone()), + MessageEvent::UpdateConversation { + conversation: new_messages, + }, + ) + .await; + } + Ok(Some(Ok(AgentEvent::ModelChange { model, mode }))) => { + publish( + Some(task_request_id.clone()), + MessageEvent::ModelChange { model, mode }, + ) + .await; + } + Ok(Some(Ok(AgentEvent::McpNotification((notification_request_id, n))))) => { + publish( + Some(task_request_id.clone()), + MessageEvent::Notification { + request_id: notification_request_id, + message: n, + }, + ) + .await; + } + Ok(Some(Err(e))) => { + tracing::error!("Error processing message: {}", e); + publish( + Some(task_request_id.clone()), + MessageEvent::Error { + error: e.to_string(), + }, + ) + .await; + break; + } + Ok(None) => { + break; + } + Err(_) => { + // Timeout — check if the bus still has subscribers + continue; + } + } + } + } + } + + // Telemetry + let session_duration = session_start.elapsed(); + + if let Ok(session) = task_state + .session_manager() + .get_session(&task_session_id, true) + .await + { + let total_tokens = session.total_tokens.unwrap_or(0); + tracing::info!( + monotonic_counter.goose.session_completions = 1, + session_type = "app", + interface = "ui", + exit_type = "normal", + duration_ms = session_duration.as_millis() as u64, + total_tokens = total_tokens, + message_count = session.message_count, + "Session completed" + ); + + tracing::info!( + monotonic_counter.goose.session_duration_ms = session_duration.as_millis() as u64, + session_type = "app", + interface = "ui", + "Session duration" + ); + + if total_tokens > 0 { + tracing::info!( + monotonic_counter.goose.session_tokens = total_tokens, + session_type = "app", + interface = "ui", + "Session tokens" + ); + } + } else { + tracing::info!( + monotonic_counter.goose.session_completions = 1, + session_type = "app", + interface = "ui", + exit_type = "normal", + duration_ms = session_duration.as_millis() as u64, + total_tokens = 0u64, + message_count = all_messages.len(), + "Session completed" + ); + + tracing::info!( + monotonic_counter.goose.session_duration_ms = session_duration.as_millis() as u64, + session_type = "app", + interface = "ui", + "Session duration" + ); + } + + let final_token_state = + get_token_state(task_state.session_manager(), &task_session_id).await; + + publish( + Some(task_request_id.clone()), + MessageEvent::Finish { + reason: "stop".to_string(), + token_state: final_token_state, + }, + ) + .await; + + task_bus.cleanup_request(&task_request_id).await; + })); + + Ok(Json(SessionReplyResponse { request_id })) +} + +// ── POST /sessions/{id}/cancel ────────────────────────────────────────── + +#[utoipa::path( + post, + path = "/sessions/{id}/cancel", + params( + ("id" = String, Path, description = "Session ID"), + ), + request_body = CancelRequest, + responses( + (status = 200, description = "Cancellation accepted"), + ) +)] +pub async fn session_cancel( + State(state): State>, + Path(session_id): Path, + Json(request): Json, +) -> axum::http::StatusCode { + let bus = match state.get_event_bus(&session_id).await { + Some(bus) => bus, + None => return axum::http::StatusCode::NOT_FOUND, + }; + bus.cancel_request(&request.request_id).await; + axum::http::StatusCode::OK +} + +// ── Route registration ────────────────────────────────────────────────── + +pub fn routes(state: Arc) -> Router { + Router::new() + .route("/sessions/{id}/events", get(session_events)) + .route( + "/sessions/{id}/reply", + post(session_reply).layer(DefaultBodyLimit::max(50 * 1024 * 1024)), + ) + .route("/sessions/{id}/cancel", post(session_cancel)) + .with_state(state) +} diff --git a/crates/goose-server/src/session_event_bus.rs b/crates/goose-server/src/session_event_bus.rs new file mode 100644 index 0000000000..d901eb5031 --- /dev/null +++ b/crates/goose-server/src/session_event_bus.rs @@ -0,0 +1,259 @@ +use crate::routes::reply::MessageEvent; +use std::collections::{HashMap, VecDeque}; +use std::sync::atomic::{AtomicU64, Ordering}; +use tokio::sync::{broadcast, Mutex}; +use tokio_util::sync::CancellationToken; + +const BROADCAST_CAPACITY: usize = 256; +const REPLAY_BUFFER_CAPACITY: usize = 512; + +/// Error returned by [`SessionEventBus::subscribe`]. +#[derive(Debug)] +pub enum SubscribeError { + /// The client's `Last-Event-ID` has been evicted from the replay buffer, + /// so events have been irrecoverably lost. + ClientTooFarBehind, +} + +#[derive(Clone, Debug)] +pub struct SessionEvent { + /// Monotonic sequence number, written as SSE `id:` frame (not in JSON payload). + pub seq: u64, + /// None for Ping events, Some for events associated with a specific request. + pub request_id: Option, + /// The event payload. + pub event: MessageEvent, +} + +pub struct SessionEventBus { + tx: broadcast::Sender, + buffer: Mutex>, + next_seq: AtomicU64, + active_requests: Mutex>, +} + +impl SessionEventBus { + pub fn new() -> Self { + let (tx, _) = broadcast::channel(BROADCAST_CAPACITY); + Self { + tx, + buffer: Mutex::new(VecDeque::with_capacity(REPLAY_BUFFER_CAPACITY)), + next_seq: AtomicU64::new(1), + active_requests: Mutex::new(HashMap::new()), + } + } + + /// Publish an event to the bus. Assigns a monotonic sequence number. + /// + /// The sequence ID is assigned under the buffer lock so that concurrent + /// callers cannot reorder events (i.e. seq=2 published before seq=1). + pub async fn publish(&self, request_id: Option, event: MessageEvent) -> u64 { + let session_event = { + let mut buf = self.buffer.lock().await; + let seq = self.next_seq.fetch_add(1, Ordering::Relaxed); + let session_event = SessionEvent { + seq, + request_id, + event, + }; + buf.push_back(session_event.clone()); + while buf.len() > REPLAY_BUFFER_CAPACITY { + buf.pop_front(); + } + session_event + }; + + // Send on broadcast channel (ignore error if no subscribers) + let _ = self.tx.send(session_event.clone()); + + session_event.seq + } + + /// Subscribe to live events. If `last_event_id` is provided, replay buffered + /// events with seq > last_event_id. Returns (replay_events, replay_max_seq, live_receiver). + /// + /// Returns `Err(SubscribeError::ClientTooFarBehind)` when `last_event_id` + /// refers to an event that has already been evicted from the replay buffer, + /// meaning the client has irrecoverably missed events. + /// + /// The live receiver is created *before* snapshotting the buffer so that + /// no event can fall into the gap between the two steps. The caller must + /// skip live events with `seq <= replay_max_seq` to deduplicate. + pub async fn subscribe( + &self, + last_event_id: Option, + ) -> Result<(Vec, u64, broadcast::Receiver), SubscribeError> { + // Subscribe first so that any event published while we hold the + // buffer lock is guaranteed to appear in `rx` (possibly duplicating + // a replay entry). The caller deduplicates via replay_max_seq. + let rx = self.tx.subscribe(); + + let (replay, replay_max_seq) = { + let buf = self.buffer.lock().await; + let buf_max = buf.back().map(|e| e.seq).unwrap_or(0); + let buf_min = buf.front().map(|e| e.seq).unwrap_or(0); + let last_id = last_event_id.unwrap_or(0); + + // If the client sent a Last-Event-ID that has been evicted from + // the buffer, they have irrecoverably missed events. + if last_id > 0 && buf_min > 0 && last_id < buf_min { + return Err(SubscribeError::ClientTooFarBehind); + } + + // Clamp to the actual buffer max so a stale Last-Event-ID + // (e.g. from before a server restart) doesn't suppress live events. + let events: Vec<_> = buf.iter().filter(|e| e.seq > last_id).cloned().collect(); + let max_seq = events.last().map(|e| e.seq).unwrap_or(last_id.min(buf_max)); + (events, max_seq) + }; + + Ok((replay, replay_max_seq, rx)) + } + + /// Return the IDs of all currently active (in-flight) requests. + pub async fn active_request_ids(&self) -> Vec { + let requests = self.active_requests.lock().await; + requests.keys().cloned().collect() + } + + /// Register a new request and return its cancellation token. + pub async fn register_request(&self, request_id: String) -> CancellationToken { + let token = CancellationToken::new(); + let mut requests = self.active_requests.lock().await; + requests.insert(request_id, token.clone()); + token + } + + /// Cancel a specific request by request_id. + pub async fn cancel_request(&self, request_id: &str) -> bool { + let requests = self.active_requests.lock().await; + if let Some(token) = requests.get(request_id) { + token.cancel(); + true + } else { + false + } + } + + /// Cancel all active requests (e.g. when deleting a session). + pub async fn cancel_all_requests(&self) { + let requests = self.active_requests.lock().await; + for token in requests.values() { + token.cancel(); + } + } + + /// Remove the cancellation token for a completed request. + pub async fn cleanup_request(&self, request_id: &str) { + let mut requests = self.active_requests.lock().await; + requests.remove(request_id); + } +} + +impl Default for SessionEventBus { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use goose::conversation::message::TokenState; + + #[tokio::test] + async fn test_publish_and_subscribe() { + let bus = SessionEventBus::new(); + + // Publish some events + bus.publish(Some("req-1".to_string()), MessageEvent::Ping) + .await; + bus.publish( + Some("req-1".to_string()), + MessageEvent::Finish { + reason: "stop".to_string(), + token_state: TokenState::default(), + }, + ) + .await; + + // Subscribe with replay + let (replay, replay_max_seq, _rx) = bus.subscribe(Some(0)).await.unwrap(); + assert_eq!(replay.len(), 2); + assert_eq!(replay[0].seq, 1); + assert_eq!(replay[1].seq, 2); + assert_eq!(replay_max_seq, 2); + } + + #[tokio::test] + async fn test_subscribe_with_last_event_id() { + let bus = SessionEventBus::new(); + + bus.publish(None, MessageEvent::Ping).await; + bus.publish(None, MessageEvent::Ping).await; + bus.publish(None, MessageEvent::Ping).await; + + // Only get events after seq 2 + let (replay, replay_max_seq, _rx) = bus.subscribe(Some(2)).await.unwrap(); + assert_eq!(replay.len(), 1); + assert_eq!(replay[0].seq, 3); + assert_eq!(replay_max_seq, 3); + } + + #[tokio::test] + async fn test_subscribe_without_last_event_id_replays_all() { + let bus = SessionEventBus::new(); + + bus.publish(None, MessageEvent::Ping).await; + bus.publish(None, MessageEvent::Ping).await; + + // First connect (no Last-Event-ID) should replay all buffered events + let (replay, replay_max_seq, _rx) = bus.subscribe(None).await.unwrap(); + assert_eq!(replay.len(), 2); + assert_eq!(replay_max_seq, 2); + } + + #[tokio::test] + async fn test_subscribe_with_stale_last_event_id() { + let bus = SessionEventBus::new(); + + // Buffer has seq 1..3, but client sends Last-Event-ID: 9999 + bus.publish(None, MessageEvent::Ping).await; + bus.publish(None, MessageEvent::Ping).await; + bus.publish(None, MessageEvent::Ping).await; + + let (replay, replay_max_seq, _rx) = bus.subscribe(Some(9999)).await.unwrap(); + // No replay events (all are below 9999) + assert_eq!(replay.len(), 0); + // replay_max_seq should be clamped to buf_max (3), not 9999 + assert_eq!(replay_max_seq, 3); + } + + #[tokio::test] + async fn test_cancel_request() { + let bus = SessionEventBus::new(); + + let token = bus.register_request("req-1".to_string()).await; + assert!(!token.is_cancelled()); + + let cancelled = bus.cancel_request("req-1").await; + assert!(cancelled); + assert!(token.is_cancelled()); + + // Non-existent request + let cancelled = bus.cancel_request("req-999").await; + assert!(!cancelled); + } + + #[tokio::test] + async fn test_cleanup_request() { + let bus = SessionEventBus::new(); + + bus.register_request("req-1".to_string()).await; + bus.cleanup_request("req-1").await; + + // Should return false since it was cleaned up + let cancelled = bus.cancel_request("req-1").await; + assert!(!cancelled); + } +} diff --git a/crates/goose-server/src/state.rs b/crates/goose-server/src/state.rs index da00db4a56..09d903c952 100644 --- a/crates/goose-server/src/state.rs +++ b/crates/goose-server/src/state.rs @@ -9,6 +9,7 @@ use std::sync::Arc; use tokio::sync::Mutex; use tokio::task::JoinHandle; +use crate::session_event_bus::SessionEventBus; use crate::tunnel::TunnelManager; use goose::agents::ExtensionLoadResult; use goose::gateway::manager::GatewayManager; @@ -26,6 +27,7 @@ pub struct AppState { pub gateway_manager: Arc, pub extension_loading_tasks: ExtensionLoadingTasks, pub inference_runtime: Arc, + session_buses: Arc>>>, } impl AppState { @@ -44,6 +46,7 @@ impl AppState { gateway_manager, extension_loading_tasks: Arc::new(Mutex::new(HashMap::new())), inference_runtime: InferenceRuntime::get_or_init(), + session_buses: Arc::new(Mutex::new(HashMap::new())), })) } @@ -107,6 +110,26 @@ impl AppState { } } + pub async fn get_or_create_event_bus(&self, session_id: &str) -> Arc { + let mut buses = self.session_buses.lock().await; + buses + .entry(session_id.to_string()) + .or_insert_with(|| Arc::new(SessionEventBus::new())) + .clone() + } + + /// Get an existing event bus for a session without creating one. + pub async fn get_event_bus(&self, session_id: &str) -> Option> { + let buses = self.session_buses.lock().await; + buses.get(session_id).cloned() + } + + /// Remove the event bus for a session, freeing its replay buffer. + pub async fn remove_event_bus(&self, session_id: &str) { + let mut buses = self.session_buses.lock().await; + buses.remove(session_id); + } + pub async fn get_agent(&self, session_id: String) -> anyhow::Result> { self.agent_manager.get_or_create_agent(session_id).await } diff --git a/ui/desktop/openapi.json b/ui/desktop/openapi.json index 045d9716de..2a81f2ee41 100644 --- a/ui/desktop/openapi.json +++ b/ui/desktop/openapi.json @@ -3281,6 +3281,127 @@ ] } }, + "/sessions/{id}/cancel": { + "post": { + "tags": [ + "super::routes::session_events" + ], + "operationId": "session_cancel", + "parameters": [ + { + "name": "id", + "in": "path", + "description": "Session ID", + "required": true, + "schema": { + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CancelRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Cancellation accepted" + } + } + } + }, + "/sessions/{id}/events": { + "get": { + "tags": [ + "super::routes::session_events" + ], + "operationId": "session_events", + "parameters": [ + { + "name": "id", + "in": "path", + "description": "Session ID", + "required": true, + "schema": { + "type": "string" + } + } + ], + "responses": { + "200": { + "description": "SSE event stream", + "content": { + "text/event-stream": { + "schema": { + "$ref": "#/components/schemas/MessageEvent" + } + } + } + }, + "404": { + "description": "Session not found" + } + } + } + }, + "/sessions/{id}/reply": { + "post": { + "tags": [ + "super::routes::session_events" + ], + "operationId": "session_reply", + "parameters": [ + { + "name": "id", + "in": "path", + "description": "Session ID", + "required": true, + "schema": { + "type": "string" + } + } + ], + "requestBody": { + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SessionReplyRequest" + } + } + }, + "required": true + }, + "responses": { + "200": { + "description": "Request accepted", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/SessionReplyResponse" + } + } + } + }, + "400": { + "description": "Invalid request" + }, + "404": { + "description": "Session not found" + }, + "424": { + "description": "Agent not initialized" + }, + "500": { + "description": "Internal server error" + } + } + } + }, "/sessions/{session_id}": { "get": { "tags": [ @@ -3974,6 +4095,17 @@ } } }, + "CancelRequest": { + "type": "object", + "required": [ + "request_id" + ], + "properties": { + "request_id": { + "type": "string" + } + } + }, "ChatRequest": { "type": "object", "required": [ @@ -6043,6 +6175,28 @@ } } }, + { + "type": "object", + "description": "Sent at the start of an SSE stream to inform the client about\nin-flight requests it can reattach to.", + "required": [ + "request_ids", + "type" + ], + "properties": { + "request_ids": { + "type": "array", + "items": { + "type": "string" + } + }, + "type": { + "type": "string", + "enum": [ + "ActiveRequests" + ] + } + } + }, { "type": "object", "required": [ @@ -7738,6 +7892,48 @@ } } }, + "SessionReplyRequest": { + "type": "object", + "required": [ + "request_id", + "user_message" + ], + "properties": { + "override_conversation": { + "type": "array", + "items": { + "$ref": "#/components/schemas/Message" + }, + "nullable": true + }, + "recipe_name": { + "type": "string", + "nullable": true + }, + "recipe_version": { + "type": "string", + "nullable": true + }, + "request_id": { + "type": "string", + "description": "Client-generated UUIDv7 identifying this request." + }, + "user_message": { + "$ref": "#/components/schemas/Message" + } + } + }, + "SessionReplyResponse": { + "type": "object", + "required": [ + "request_id" + ], + "properties": { + "request_id": { + "type": "string" + } + } + }, "SessionType": { "type": "string", "enum": [ diff --git a/ui/desktop/src/api/index.ts b/ui/desktop/src/api/index.ts index a7fc3801cb..dbe9fb8e9b 100644 --- a/ui/desktop/src/api/index.ts +++ b/ui/desktop/src/api/index.ts @@ -1,4 +1,4 @@ // This file is auto-generated by @hey-api/openapi-ts -export { addExtension, agentAddExtension, agentRemoveExtension, backupConfig, callTool, cancelDownload, cancelLocalModelDownload, checkProvider, cleanupProviderCache, configureProviderOauth, confirmToolAction, createCustomProvider, createRecipe, createSchedule, decodeRecipe, deleteLocalModel, deleteModel, deleteRecipe, deleteSchedule, deleteSession, detectProvider, diagnostics, downloadHfModel, downloadModel, encodeRecipe, exportApp, exportSession, forkSession, getCanonicalModelInfo, getCustomProvider, getDictationConfig, getDownloadProgress, getExtensions, getLocalModelDownloadProgress, getModelSettings, getPrompt, getPrompts, getProviderCatalog, getProviderCatalogTemplate, getProviderModels, getRepoFiles, getSession, getSessionExtensions, getSessionInsights, getSlashCommands, getTools, getTunnelStatus, importApp, importSession, initConfig, inspectRunningJob, killRunningJob, listApps, listLocalModels, listModels, listRecipes, listSchedules, listSessions, mcpUiProxy, type Options, parseRecipe, pauseSchedule, providers, readAllConfig, readConfig, readResource, recipeToYaml, recoverConfig, removeConfig, removeCustomProvider, removeExtension, reply, resetPrompt, restartAgent, resumeAgent, runNowHandler, savePrompt, saveRecipe, scanRecipe, scheduleRecipe, searchHfModels, searchSessions, sendTelemetryEvent, sessionsHandler, setConfigProvider, setRecipeSlashCommand, startAgent, startNanogptSetup, startOpenrouterSetup, startTetrateSetup, startTunnel, status, stopAgent, stopTunnel, systemInfo, transcribeDictation, unpauseSchedule, updateAgentProvider, updateCustomProvider, updateFromSession, updateModelSettings, updateSchedule, updateSession, updateSessionName, updateSessionUserRecipeValues, updateWorkingDir, upsertConfig, upsertPermissions, validateConfig } from './sdk.gen'; -export type { ActionRequired, ActionRequiredData, AddExtensionData, AddExtensionErrors, AddExtensionRequest, AddExtensionResponse, AddExtensionResponses, AgentAddExtensionData, AgentAddExtensionErrors, AgentAddExtensionResponse, AgentAddExtensionResponses, AgentRemoveExtensionData, AgentRemoveExtensionErrors, AgentRemoveExtensionResponse, AgentRemoveExtensionResponses, Annotations, Author, AuthorRequest, BackupConfigData, BackupConfigErrors, BackupConfigResponse, BackupConfigResponses, CallToolData, CallToolErrors, CallToolRequest, CallToolResponse, CallToolResponse2, CallToolResponses, CancelDownloadData, CancelDownloadErrors, CancelDownloadResponses, CancelLocalModelDownloadData, CancelLocalModelDownloadErrors, CancelLocalModelDownloadResponses, ChatRequest, CheckProviderData, CheckProviderRequest, CleanupProviderCacheData, CleanupProviderCacheErrors, CleanupProviderCacheResponse, CleanupProviderCacheResponses, ClientOptions, CommandType, ConfigKey, ConfigKeyQuery, ConfigResponse, ConfigureProviderOauthData, ConfigureProviderOauthErrors, ConfigureProviderOauthResponses, ConfirmToolActionData, ConfirmToolActionErrors, ConfirmToolActionRequest, ConfirmToolActionResponses, Content, ContentBlock, Conversation, CreateCustomProviderData, CreateCustomProviderErrors, CreateCustomProviderResponse, CreateCustomProviderResponse2, CreateCustomProviderResponses, CreateRecipeData, CreateRecipeErrors, CreateRecipeRequest, CreateRecipeResponse, CreateRecipeResponse2, CreateRecipeResponses, CreateScheduleData, CreateScheduleErrors, CreateScheduleRequest, CreateScheduleResponse, CreateScheduleResponses, CspMetadata, DeclarativeProviderConfig, DecodeRecipeData, DecodeRecipeErrors, DecodeRecipeRequest, DecodeRecipeResponse, DecodeRecipeResponse2, DecodeRecipeResponses, DeleteLocalModelData, DeleteLocalModelErrors, DeleteLocalModelResponses, DeleteModelData, DeleteModelErrors, DeleteModelResponses, DeleteRecipeData, DeleteRecipeErrors, DeleteRecipeRequest, DeleteRecipeResponse, DeleteRecipeResponses, DeleteScheduleData, DeleteScheduleErrors, DeleteScheduleResponse, DeleteScheduleResponses, DeleteSessionData, DeleteSessionErrors, DeleteSessionResponses, DetectProviderData, DetectProviderErrors, DetectProviderRequest, DetectProviderResponse, DetectProviderResponse2, DetectProviderResponses, DiagnosticsData, DiagnosticsErrors, DiagnosticsResponse, DiagnosticsResponses, DictationProvider, DictationProviderStatus, DownloadHfModelData, DownloadHfModelErrors, DownloadHfModelResponse, DownloadHfModelResponses, DownloadModelData, DownloadModelErrors, DownloadModelRequest, DownloadModelResponses, DownloadProgress, DownloadStatus, EmbeddedResource, EncodeRecipeData, EncodeRecipeErrors, EncodeRecipeRequest, EncodeRecipeResponse, EncodeRecipeResponse2, EncodeRecipeResponses, Envs, EnvVarConfig, ErrorResponse, ExportAppData, ExportAppError, ExportAppErrors, ExportAppResponse, ExportAppResponses, ExportSessionData, ExportSessionErrors, ExportSessionResponse, ExportSessionResponses, ExtensionConfig, ExtensionData, ExtensionEntry, ExtensionLoadResult, ExtensionQuery, ExtensionResponse, ForkRequest, ForkResponse, ForkSessionData, ForkSessionErrors, ForkSessionResponse, ForkSessionResponses, FrontendToolRequest, GetCanonicalModelInfoData, GetCanonicalModelInfoResponse, GetCanonicalModelInfoResponses, GetCustomProviderData, GetCustomProviderErrors, GetCustomProviderResponse, GetCustomProviderResponses, GetDictationConfigData, GetDictationConfigResponse, GetDictationConfigResponses, GetDownloadProgressData, GetDownloadProgressErrors, GetDownloadProgressResponse, GetDownloadProgressResponses, GetExtensionsData, GetExtensionsErrors, GetExtensionsResponse, GetExtensionsResponses, GetLocalModelDownloadProgressData, GetLocalModelDownloadProgressErrors, GetLocalModelDownloadProgressResponse, GetLocalModelDownloadProgressResponses, GetModelSettingsData, GetModelSettingsErrors, GetModelSettingsResponse, GetModelSettingsResponses, GetPromptData, GetPromptErrors, GetPromptResponse, GetPromptResponses, GetPromptsData, GetPromptsResponse, GetPromptsResponses, GetProviderCatalogData, GetProviderCatalogErrors, GetProviderCatalogResponse, GetProviderCatalogResponses, GetProviderCatalogTemplateData, GetProviderCatalogTemplateErrors, GetProviderCatalogTemplateResponse, GetProviderCatalogTemplateResponses, GetProviderModelsData, GetProviderModelsErrors, GetProviderModelsResponse, GetProviderModelsResponses, GetRepoFilesData, GetRepoFilesResponse, GetRepoFilesResponses, GetSessionData, GetSessionErrors, GetSessionExtensionsData, GetSessionExtensionsErrors, GetSessionExtensionsResponse, GetSessionExtensionsResponses, GetSessionInsightsData, GetSessionInsightsErrors, GetSessionInsightsResponse, GetSessionInsightsResponses, GetSessionResponse, GetSessionResponses, GetSlashCommandsData, GetSlashCommandsResponse, GetSlashCommandsResponses, GetToolsData, GetToolsErrors, GetToolsQuery, GetToolsResponse, GetToolsResponses, GetTunnelStatusData, GetTunnelStatusResponse, GetTunnelStatusResponses, GooseApp, GooseMode, HfGgufFile, HfModelInfo, HfQuantVariant, Icon, ImageContent, ImportAppData, ImportAppError, ImportAppErrors, ImportAppRequest, ImportAppResponse, ImportAppResponse2, ImportAppResponses, ImportSessionData, ImportSessionErrors, ImportSessionRequest, ImportSessionResponse, ImportSessionResponses, InitConfigData, InitConfigErrors, InitConfigResponse, InitConfigResponses, InspectJobResponse, InspectRunningJobData, InspectRunningJobErrors, InspectRunningJobResponse, InspectRunningJobResponses, JsonObject, KillJobResponse, KillRunningJobData, KillRunningJobResponses, ListAppsData, ListAppsError, ListAppsErrors, ListAppsRequest, ListAppsResponse, ListAppsResponse2, ListAppsResponses, ListLocalModelsData, ListLocalModelsResponse, ListLocalModelsResponses, ListModelsData, ListModelsResponse, ListModelsResponses, ListRecipeResponse, ListRecipesData, ListRecipesErrors, ListRecipesResponse, ListRecipesResponses, ListSchedulesData, ListSchedulesErrors, ListSchedulesResponse, ListSchedulesResponse2, ListSchedulesResponses, ListSessionsData, ListSessionsErrors, ListSessionsResponse, ListSessionsResponses, LoadedProvider, LocalModelResponse, McpAppResource, McpUiProxyData, McpUiProxyErrors, McpUiProxyResponses, Message, MessageContent, MessageEvent, MessageMetadata, ModelCapabilities, ModelConfig, ModelDownloadStatus, ModelInfo, ModelInfoData, ModelInfoQuery, ModelInfoResponse, ModelSettings, ModelTemplate, ParseRecipeData, ParseRecipeError, ParseRecipeErrors, ParseRecipeRequest, ParseRecipeResponse, ParseRecipeResponse2, ParseRecipeResponses, PauseScheduleData, PauseScheduleErrors, PauseScheduleResponse, PauseScheduleResponses, Permission, PermissionLevel, PermissionsMetadata, PrincipalType, PromptContentResponse, PromptsListResponse, ProviderCatalogEntry, ProviderDetails, ProviderEngine, ProviderMetadata, ProvidersData, ProvidersResponse, ProvidersResponse2, ProvidersResponses, ProviderTemplate, ProviderType, RawAudioContent, RawEmbeddedResource, RawImageContent, RawResource, RawTextContent, ReadAllConfigData, ReadAllConfigResponse, ReadAllConfigResponses, ReadConfigData, ReadConfigErrors, ReadConfigResponses, ReadResourceData, ReadResourceErrors, ReadResourceRequest, ReadResourceResponse, ReadResourceResponse2, ReadResourceResponses, Recipe, RecipeManifest, RecipeParameter, RecipeParameterInputType, RecipeParameterRequirement, RecipeToYamlData, RecipeToYamlError, RecipeToYamlErrors, RecipeToYamlRequest, RecipeToYamlResponse, RecipeToYamlResponse2, RecipeToYamlResponses, RecoverConfigData, RecoverConfigErrors, RecoverConfigResponse, RecoverConfigResponses, RedactedThinkingContent, RemoveConfigData, RemoveConfigErrors, RemoveConfigResponse, RemoveConfigResponses, RemoveCustomProviderData, RemoveCustomProviderErrors, RemoveCustomProviderResponse, RemoveCustomProviderResponses, RemoveExtensionData, RemoveExtensionErrors, RemoveExtensionRequest, RemoveExtensionResponse, RemoveExtensionResponses, ReplyData, ReplyErrors, ReplyResponse, ReplyResponses, RepoVariantsResponse, ResetPromptData, ResetPromptErrors, ResetPromptResponse, ResetPromptResponses, ResourceContents, ResourceMetadata, Response, RestartAgentData, RestartAgentErrors, RestartAgentRequest, RestartAgentResponse, RestartAgentResponse2, RestartAgentResponses, ResumeAgentData, ResumeAgentErrors, ResumeAgentRequest, ResumeAgentResponse, ResumeAgentResponse2, ResumeAgentResponses, RetryConfig, Role, RunNowHandlerData, RunNowHandlerErrors, RunNowHandlerResponse, RunNowHandlerResponses, RunNowResponse, SamplingConfig, SavePromptData, SavePromptErrors, SavePromptRequest, SavePromptResponse, SavePromptResponses, SaveRecipeData, SaveRecipeError, SaveRecipeErrors, SaveRecipeRequest, SaveRecipeResponse, SaveRecipeResponse2, SaveRecipeResponses, ScanRecipeData, ScanRecipeRequest, ScanRecipeResponse, ScanRecipeResponse2, ScanRecipeResponses, ScheduledJob, ScheduleRecipeData, ScheduleRecipeErrors, ScheduleRecipeRequest, ScheduleRecipeResponses, SearchHfModelsData, SearchHfModelsErrors, SearchHfModelsResponse, SearchHfModelsResponses, SearchSessionsData, SearchSessionsErrors, SearchSessionsResponse, SearchSessionsResponses, SendTelemetryEventData, SendTelemetryEventResponses, Session, SessionDisplayInfo, SessionExtensionsResponse, SessionInsights, SessionListResponse, SessionsHandlerData, SessionsHandlerErrors, SessionsHandlerResponse, SessionsHandlerResponses, SessionsQuery, SessionType, SetConfigProviderData, SetProviderRequest, SetRecipeSlashCommandData, SetRecipeSlashCommandErrors, SetRecipeSlashCommandResponses, SetSlashCommandRequest, Settings, SetupResponse, SlashCommand, SlashCommandsResponse, StartAgentData, StartAgentError, StartAgentErrors, StartAgentRequest, StartAgentResponse, StartAgentResponses, StartNanogptSetupData, StartNanogptSetupResponse, StartNanogptSetupResponses, StartOpenrouterSetupData, StartOpenrouterSetupResponse, StartOpenrouterSetupResponses, StartTetrateSetupData, StartTetrateSetupResponse, StartTetrateSetupResponses, StartTunnelData, StartTunnelError, StartTunnelErrors, StartTunnelResponse, StartTunnelResponses, StatusData, StatusResponse, StatusResponses, StopAgentData, StopAgentErrors, StopAgentRequest, StopAgentResponse, StopAgentResponses, StopTunnelData, StopTunnelError, StopTunnelErrors, StopTunnelResponses, SubRecipe, SuccessCheck, SystemInfo, SystemInfoData, SystemInfoResponse, SystemInfoResponses, SystemNotificationContent, SystemNotificationType, TaskSupport, TelemetryEventRequest, Template, TextContent, ThinkingContent, TokenState, Tool, ToolAnnotations, ToolConfirmationRequest, ToolExecution, ToolInfo, ToolPermission, ToolRequest, ToolResponse, TranscribeDictationData, TranscribeDictationErrors, TranscribeDictationResponse, TranscribeDictationResponses, TranscribeRequest, TranscribeResponse, TunnelInfo, TunnelState, UiMetadata, UnpauseScheduleData, UnpauseScheduleErrors, UnpauseScheduleResponse, UnpauseScheduleResponses, UpdateAgentProviderData, UpdateAgentProviderErrors, UpdateAgentProviderResponses, UpdateCustomProviderData, UpdateCustomProviderErrors, UpdateCustomProviderRequest, UpdateCustomProviderResponse, UpdateCustomProviderResponses, UpdateFromSessionData, UpdateFromSessionErrors, UpdateFromSessionRequest, UpdateFromSessionResponses, UpdateModelSettingsData, UpdateModelSettingsErrors, UpdateModelSettingsResponse, UpdateModelSettingsResponses, UpdateProviderRequest, UpdateScheduleData, UpdateScheduleErrors, UpdateScheduleRequest, UpdateScheduleResponse, UpdateScheduleResponses, UpdateSessionData, UpdateSessionErrors, UpdateSessionNameData, UpdateSessionNameErrors, UpdateSessionNameRequest, UpdateSessionNameResponses, UpdateSessionRequest, UpdateSessionResponses, UpdateSessionUserRecipeValuesData, UpdateSessionUserRecipeValuesError, UpdateSessionUserRecipeValuesErrors, UpdateSessionUserRecipeValuesRequest, UpdateSessionUserRecipeValuesResponse, UpdateSessionUserRecipeValuesResponse2, UpdateSessionUserRecipeValuesResponses, UpdateWorkingDirData, UpdateWorkingDirErrors, UpdateWorkingDirRequest, UpdateWorkingDirResponses, UpsertConfigData, UpsertConfigErrors, UpsertConfigQuery, UpsertConfigResponse, UpsertConfigResponses, UpsertPermissionsData, UpsertPermissionsErrors, UpsertPermissionsQuery, UpsertPermissionsResponse, UpsertPermissionsResponses, ValidateConfigData, ValidateConfigErrors, ValidateConfigResponse, ValidateConfigResponses, WhisperModelResponse, WindowProps } from './types.gen'; +export { addExtension, agentAddExtension, agentRemoveExtension, backupConfig, callTool, cancelDownload, cancelLocalModelDownload, checkProvider, cleanupProviderCache, configureProviderOauth, confirmToolAction, createCustomProvider, createRecipe, createSchedule, decodeRecipe, deleteLocalModel, deleteModel, deleteRecipe, deleteSchedule, deleteSession, detectProvider, diagnostics, downloadHfModel, downloadModel, encodeRecipe, exportApp, exportSession, forkSession, getCanonicalModelInfo, getCustomProvider, getDictationConfig, getDownloadProgress, getExtensions, getLocalModelDownloadProgress, getModelSettings, getPrompt, getPrompts, getProviderCatalog, getProviderCatalogTemplate, getProviderModels, getRepoFiles, getSession, getSessionExtensions, getSessionInsights, getSlashCommands, getTools, getTunnelStatus, importApp, importSession, initConfig, inspectRunningJob, killRunningJob, listApps, listLocalModels, listModels, listRecipes, listSchedules, listSessions, mcpUiProxy, type Options, parseRecipe, pauseSchedule, providers, readAllConfig, readConfig, readResource, recipeToYaml, recoverConfig, removeConfig, removeCustomProvider, removeExtension, reply, resetPrompt, restartAgent, resumeAgent, runNowHandler, savePrompt, saveRecipe, scanRecipe, scheduleRecipe, searchHfModels, searchSessions, sendTelemetryEvent, sessionCancel, sessionEvents, sessionReply, sessionsHandler, setConfigProvider, setRecipeSlashCommand, startAgent, startNanogptSetup, startOpenrouterSetup, startTetrateSetup, startTunnel, status, stopAgent, stopTunnel, systemInfo, transcribeDictation, unpauseSchedule, updateAgentProvider, updateCustomProvider, updateFromSession, updateModelSettings, updateSchedule, updateSession, updateSessionName, updateSessionUserRecipeValues, updateWorkingDir, upsertConfig, upsertPermissions, validateConfig } from './sdk.gen'; +export type { ActionRequired, ActionRequiredData, AddExtensionData, AddExtensionErrors, AddExtensionRequest, AddExtensionResponse, AddExtensionResponses, AgentAddExtensionData, AgentAddExtensionErrors, AgentAddExtensionResponse, AgentAddExtensionResponses, AgentRemoveExtensionData, AgentRemoveExtensionErrors, AgentRemoveExtensionResponse, AgentRemoveExtensionResponses, Annotations, Author, AuthorRequest, BackupConfigData, BackupConfigErrors, BackupConfigResponse, BackupConfigResponses, CallToolData, CallToolErrors, CallToolRequest, CallToolResponse, CallToolResponse2, CallToolResponses, CancelDownloadData, CancelDownloadErrors, CancelDownloadResponses, CancelLocalModelDownloadData, CancelLocalModelDownloadErrors, CancelLocalModelDownloadResponses, CancelRequest, ChatRequest, CheckProviderData, CheckProviderRequest, CleanupProviderCacheData, CleanupProviderCacheErrors, CleanupProviderCacheResponse, CleanupProviderCacheResponses, ClientOptions, CommandType, ConfigKey, ConfigKeyQuery, ConfigResponse, ConfigureProviderOauthData, ConfigureProviderOauthErrors, ConfigureProviderOauthResponses, ConfirmToolActionData, ConfirmToolActionErrors, ConfirmToolActionRequest, ConfirmToolActionResponses, Content, ContentBlock, Conversation, CreateCustomProviderData, CreateCustomProviderErrors, CreateCustomProviderResponse, CreateCustomProviderResponse2, CreateCustomProviderResponses, CreateRecipeData, CreateRecipeErrors, CreateRecipeRequest, CreateRecipeResponse, CreateRecipeResponse2, CreateRecipeResponses, CreateScheduleData, CreateScheduleErrors, CreateScheduleRequest, CreateScheduleResponse, CreateScheduleResponses, CspMetadata, DeclarativeProviderConfig, DecodeRecipeData, DecodeRecipeErrors, DecodeRecipeRequest, DecodeRecipeResponse, DecodeRecipeResponse2, DecodeRecipeResponses, DeleteLocalModelData, DeleteLocalModelErrors, DeleteLocalModelResponses, DeleteModelData, DeleteModelErrors, DeleteModelResponses, DeleteRecipeData, DeleteRecipeErrors, DeleteRecipeRequest, DeleteRecipeResponse, DeleteRecipeResponses, DeleteScheduleData, DeleteScheduleErrors, DeleteScheduleResponse, DeleteScheduleResponses, DeleteSessionData, DeleteSessionErrors, DeleteSessionResponses, DetectProviderData, DetectProviderErrors, DetectProviderRequest, DetectProviderResponse, DetectProviderResponse2, DetectProviderResponses, DiagnosticsData, DiagnosticsErrors, DiagnosticsResponse, DiagnosticsResponses, DictationProvider, DictationProviderStatus, DownloadHfModelData, DownloadHfModelErrors, DownloadHfModelResponse, DownloadHfModelResponses, DownloadModelData, DownloadModelErrors, DownloadModelRequest, DownloadModelResponses, DownloadProgress, DownloadStatus, EmbeddedResource, EncodeRecipeData, EncodeRecipeErrors, EncodeRecipeRequest, EncodeRecipeResponse, EncodeRecipeResponse2, EncodeRecipeResponses, Envs, EnvVarConfig, ErrorResponse, ExportAppData, ExportAppError, ExportAppErrors, ExportAppResponse, ExportAppResponses, ExportSessionData, ExportSessionErrors, ExportSessionResponse, ExportSessionResponses, ExtensionConfig, ExtensionData, ExtensionEntry, ExtensionLoadResult, ExtensionQuery, ExtensionResponse, ForkRequest, ForkResponse, ForkSessionData, ForkSessionErrors, ForkSessionResponse, ForkSessionResponses, FrontendToolRequest, GetCanonicalModelInfoData, GetCanonicalModelInfoResponse, GetCanonicalModelInfoResponses, GetCustomProviderData, GetCustomProviderErrors, GetCustomProviderResponse, GetCustomProviderResponses, GetDictationConfigData, GetDictationConfigResponse, GetDictationConfigResponses, GetDownloadProgressData, GetDownloadProgressErrors, GetDownloadProgressResponse, GetDownloadProgressResponses, GetExtensionsData, GetExtensionsErrors, GetExtensionsResponse, GetExtensionsResponses, GetLocalModelDownloadProgressData, GetLocalModelDownloadProgressErrors, GetLocalModelDownloadProgressResponse, GetLocalModelDownloadProgressResponses, GetModelSettingsData, GetModelSettingsErrors, GetModelSettingsResponse, GetModelSettingsResponses, GetPromptData, GetPromptErrors, GetPromptResponse, GetPromptResponses, GetPromptsData, GetPromptsResponse, GetPromptsResponses, GetProviderCatalogData, GetProviderCatalogErrors, GetProviderCatalogResponse, GetProviderCatalogResponses, GetProviderCatalogTemplateData, GetProviderCatalogTemplateErrors, GetProviderCatalogTemplateResponse, GetProviderCatalogTemplateResponses, GetProviderModelsData, GetProviderModelsErrors, GetProviderModelsResponse, GetProviderModelsResponses, GetRepoFilesData, GetRepoFilesResponse, GetRepoFilesResponses, GetSessionData, GetSessionErrors, GetSessionExtensionsData, GetSessionExtensionsErrors, GetSessionExtensionsResponse, GetSessionExtensionsResponses, GetSessionInsightsData, GetSessionInsightsErrors, GetSessionInsightsResponse, GetSessionInsightsResponses, GetSessionResponse, GetSessionResponses, GetSlashCommandsData, GetSlashCommandsResponse, GetSlashCommandsResponses, GetToolsData, GetToolsErrors, GetToolsQuery, GetToolsResponse, GetToolsResponses, GetTunnelStatusData, GetTunnelStatusResponse, GetTunnelStatusResponses, GooseApp, GooseMode, HfGgufFile, HfModelInfo, HfQuantVariant, Icon, ImageContent, ImportAppData, ImportAppError, ImportAppErrors, ImportAppRequest, ImportAppResponse, ImportAppResponse2, ImportAppResponses, ImportSessionData, ImportSessionErrors, ImportSessionRequest, ImportSessionResponse, ImportSessionResponses, InitConfigData, InitConfigErrors, InitConfigResponse, InitConfigResponses, InspectJobResponse, InspectRunningJobData, InspectRunningJobErrors, InspectRunningJobResponse, InspectRunningJobResponses, JsonObject, KillJobResponse, KillRunningJobData, KillRunningJobResponses, ListAppsData, ListAppsError, ListAppsErrors, ListAppsRequest, ListAppsResponse, ListAppsResponse2, ListAppsResponses, ListLocalModelsData, ListLocalModelsResponse, ListLocalModelsResponses, ListModelsData, ListModelsResponse, ListModelsResponses, ListRecipeResponse, ListRecipesData, ListRecipesErrors, ListRecipesResponse, ListRecipesResponses, ListSchedulesData, ListSchedulesErrors, ListSchedulesResponse, ListSchedulesResponse2, ListSchedulesResponses, ListSessionsData, ListSessionsErrors, ListSessionsResponse, ListSessionsResponses, LoadedProvider, LocalModelResponse, McpAppResource, McpUiProxyData, McpUiProxyErrors, McpUiProxyResponses, Message, MessageContent, MessageEvent, MessageMetadata, ModelCapabilities, ModelConfig, ModelDownloadStatus, ModelInfo, ModelInfoData, ModelInfoQuery, ModelInfoResponse, ModelSettings, ModelTemplate, ParseRecipeData, ParseRecipeError, ParseRecipeErrors, ParseRecipeRequest, ParseRecipeResponse, ParseRecipeResponse2, ParseRecipeResponses, PauseScheduleData, PauseScheduleErrors, PauseScheduleResponse, PauseScheduleResponses, Permission, PermissionLevel, PermissionsMetadata, PrincipalType, PromptContentResponse, PromptsListResponse, ProviderCatalogEntry, ProviderDetails, ProviderEngine, ProviderMetadata, ProvidersData, ProvidersResponse, ProvidersResponse2, ProvidersResponses, ProviderTemplate, ProviderType, RawAudioContent, RawEmbeddedResource, RawImageContent, RawResource, RawTextContent, ReadAllConfigData, ReadAllConfigResponse, ReadAllConfigResponses, ReadConfigData, ReadConfigErrors, ReadConfigResponses, ReadResourceData, ReadResourceErrors, ReadResourceRequest, ReadResourceResponse, ReadResourceResponse2, ReadResourceResponses, Recipe, RecipeManifest, RecipeParameter, RecipeParameterInputType, RecipeParameterRequirement, RecipeToYamlData, RecipeToYamlError, RecipeToYamlErrors, RecipeToYamlRequest, RecipeToYamlResponse, RecipeToYamlResponse2, RecipeToYamlResponses, RecoverConfigData, RecoverConfigErrors, RecoverConfigResponse, RecoverConfigResponses, RedactedThinkingContent, RemoveConfigData, RemoveConfigErrors, RemoveConfigResponse, RemoveConfigResponses, RemoveCustomProviderData, RemoveCustomProviderErrors, RemoveCustomProviderResponse, RemoveCustomProviderResponses, RemoveExtensionData, RemoveExtensionErrors, RemoveExtensionRequest, RemoveExtensionResponse, RemoveExtensionResponses, ReplyData, ReplyErrors, ReplyResponse, ReplyResponses, RepoVariantsResponse, ResetPromptData, ResetPromptErrors, ResetPromptResponse, ResetPromptResponses, ResourceContents, ResourceMetadata, Response, RestartAgentData, RestartAgentErrors, RestartAgentRequest, RestartAgentResponse, RestartAgentResponse2, RestartAgentResponses, ResumeAgentData, ResumeAgentErrors, ResumeAgentRequest, ResumeAgentResponse, ResumeAgentResponse2, ResumeAgentResponses, RetryConfig, Role, RunNowHandlerData, RunNowHandlerErrors, RunNowHandlerResponse, RunNowHandlerResponses, RunNowResponse, SamplingConfig, SavePromptData, SavePromptErrors, SavePromptRequest, SavePromptResponse, SavePromptResponses, SaveRecipeData, SaveRecipeError, SaveRecipeErrors, SaveRecipeRequest, SaveRecipeResponse, SaveRecipeResponse2, SaveRecipeResponses, ScanRecipeData, ScanRecipeRequest, ScanRecipeResponse, ScanRecipeResponse2, ScanRecipeResponses, ScheduledJob, ScheduleRecipeData, ScheduleRecipeErrors, ScheduleRecipeRequest, ScheduleRecipeResponses, SearchHfModelsData, SearchHfModelsErrors, SearchHfModelsResponse, SearchHfModelsResponses, SearchSessionsData, SearchSessionsErrors, SearchSessionsResponse, SearchSessionsResponses, SendTelemetryEventData, SendTelemetryEventResponses, Session, SessionCancelData, SessionCancelResponses, SessionDisplayInfo, SessionEventsData, SessionEventsErrors, SessionEventsResponse, SessionEventsResponses, SessionExtensionsResponse, SessionInsights, SessionListResponse, SessionReplyData, SessionReplyErrors, SessionReplyRequest, SessionReplyResponse, SessionReplyResponse2, SessionReplyResponses, SessionsHandlerData, SessionsHandlerErrors, SessionsHandlerResponse, SessionsHandlerResponses, SessionsQuery, SessionType, SetConfigProviderData, SetProviderRequest, SetRecipeSlashCommandData, SetRecipeSlashCommandErrors, SetRecipeSlashCommandResponses, SetSlashCommandRequest, Settings, SetupResponse, SlashCommand, SlashCommandsResponse, StartAgentData, StartAgentError, StartAgentErrors, StartAgentRequest, StartAgentResponse, StartAgentResponses, StartNanogptSetupData, StartNanogptSetupResponse, StartNanogptSetupResponses, StartOpenrouterSetupData, StartOpenrouterSetupResponse, StartOpenrouterSetupResponses, StartTetrateSetupData, StartTetrateSetupResponse, StartTetrateSetupResponses, StartTunnelData, StartTunnelError, StartTunnelErrors, StartTunnelResponse, StartTunnelResponses, StatusData, StatusResponse, StatusResponses, StopAgentData, StopAgentErrors, StopAgentRequest, StopAgentResponse, StopAgentResponses, StopTunnelData, StopTunnelError, StopTunnelErrors, StopTunnelResponses, SubRecipe, SuccessCheck, SystemInfo, SystemInfoData, SystemInfoResponse, SystemInfoResponses, SystemNotificationContent, SystemNotificationType, TaskSupport, TelemetryEventRequest, Template, TextContent, ThinkingContent, TokenState, Tool, ToolAnnotations, ToolConfirmationRequest, ToolExecution, ToolInfo, ToolPermission, ToolRequest, ToolResponse, TranscribeDictationData, TranscribeDictationErrors, TranscribeDictationResponse, TranscribeDictationResponses, TranscribeRequest, TranscribeResponse, TunnelInfo, TunnelState, UiMetadata, UnpauseScheduleData, UnpauseScheduleErrors, UnpauseScheduleResponse, UnpauseScheduleResponses, UpdateAgentProviderData, UpdateAgentProviderErrors, UpdateAgentProviderResponses, UpdateCustomProviderData, UpdateCustomProviderErrors, UpdateCustomProviderRequest, UpdateCustomProviderResponse, UpdateCustomProviderResponses, UpdateFromSessionData, UpdateFromSessionErrors, UpdateFromSessionRequest, UpdateFromSessionResponses, UpdateModelSettingsData, UpdateModelSettingsErrors, UpdateModelSettingsResponse, UpdateModelSettingsResponses, UpdateProviderRequest, UpdateScheduleData, UpdateScheduleErrors, UpdateScheduleRequest, UpdateScheduleResponse, UpdateScheduleResponses, UpdateSessionData, UpdateSessionErrors, UpdateSessionNameData, UpdateSessionNameErrors, UpdateSessionNameRequest, UpdateSessionNameResponses, UpdateSessionRequest, UpdateSessionResponses, UpdateSessionUserRecipeValuesData, UpdateSessionUserRecipeValuesError, UpdateSessionUserRecipeValuesErrors, UpdateSessionUserRecipeValuesRequest, UpdateSessionUserRecipeValuesResponse, UpdateSessionUserRecipeValuesResponse2, UpdateSessionUserRecipeValuesResponses, UpdateWorkingDirData, UpdateWorkingDirErrors, UpdateWorkingDirRequest, UpdateWorkingDirResponses, UpsertConfigData, UpsertConfigErrors, UpsertConfigQuery, UpsertConfigResponse, UpsertConfigResponses, UpsertPermissionsData, UpsertPermissionsErrors, UpsertPermissionsQuery, UpsertPermissionsResponse, UpsertPermissionsResponses, ValidateConfigData, ValidateConfigErrors, ValidateConfigResponse, ValidateConfigResponses, WhisperModelResponse, WindowProps } from './types.gen'; diff --git a/ui/desktop/src/api/sdk.gen.ts b/ui/desktop/src/api/sdk.gen.ts index d108f8dc6c..3d12b5b574 100644 --- a/ui/desktop/src/api/sdk.gen.ts +++ b/ui/desktop/src/api/sdk.gen.ts @@ -2,7 +2,7 @@ import type { Client, Options as Options2, TDataShape } from './client'; import { client } from './client.gen'; -import type { AddExtensionData, AddExtensionErrors, AddExtensionResponses, AgentAddExtensionData, AgentAddExtensionErrors, AgentAddExtensionResponses, AgentRemoveExtensionData, AgentRemoveExtensionErrors, AgentRemoveExtensionResponses, BackupConfigData, BackupConfigErrors, BackupConfigResponses, CallToolData, CallToolErrors, CallToolResponses, CancelDownloadData, CancelDownloadErrors, CancelDownloadResponses, CancelLocalModelDownloadData, CancelLocalModelDownloadErrors, CancelLocalModelDownloadResponses, CheckProviderData, CleanupProviderCacheData, CleanupProviderCacheErrors, CleanupProviderCacheResponses, ConfigureProviderOauthData, ConfigureProviderOauthErrors, ConfigureProviderOauthResponses, ConfirmToolActionData, ConfirmToolActionErrors, ConfirmToolActionResponses, CreateCustomProviderData, CreateCustomProviderErrors, CreateCustomProviderResponses, CreateRecipeData, CreateRecipeErrors, CreateRecipeResponses, CreateScheduleData, CreateScheduleErrors, CreateScheduleResponses, DecodeRecipeData, DecodeRecipeErrors, DecodeRecipeResponses, DeleteLocalModelData, DeleteLocalModelErrors, DeleteLocalModelResponses, DeleteModelData, DeleteModelErrors, DeleteModelResponses, DeleteRecipeData, DeleteRecipeErrors, DeleteRecipeResponses, DeleteScheduleData, DeleteScheduleErrors, DeleteScheduleResponses, DeleteSessionData, DeleteSessionErrors, DeleteSessionResponses, DetectProviderData, DetectProviderErrors, DetectProviderResponses, DiagnosticsData, DiagnosticsErrors, DiagnosticsResponses, DownloadHfModelData, DownloadHfModelErrors, DownloadHfModelResponses, DownloadModelData, DownloadModelErrors, DownloadModelResponses, EncodeRecipeData, EncodeRecipeErrors, EncodeRecipeResponses, ExportAppData, ExportAppErrors, ExportAppResponses, ExportSessionData, ExportSessionErrors, ExportSessionResponses, ForkSessionData, ForkSessionErrors, ForkSessionResponses, GetCanonicalModelInfoData, GetCanonicalModelInfoResponses, GetCustomProviderData, GetCustomProviderErrors, GetCustomProviderResponses, GetDictationConfigData, GetDictationConfigResponses, GetDownloadProgressData, GetDownloadProgressErrors, GetDownloadProgressResponses, GetExtensionsData, GetExtensionsErrors, GetExtensionsResponses, GetLocalModelDownloadProgressData, GetLocalModelDownloadProgressErrors, GetLocalModelDownloadProgressResponses, GetModelSettingsData, GetModelSettingsErrors, GetModelSettingsResponses, GetPromptData, GetPromptErrors, GetPromptResponses, GetPromptsData, GetPromptsResponses, GetProviderCatalogData, GetProviderCatalogErrors, GetProviderCatalogResponses, GetProviderCatalogTemplateData, GetProviderCatalogTemplateErrors, GetProviderCatalogTemplateResponses, GetProviderModelsData, GetProviderModelsErrors, GetProviderModelsResponses, GetRepoFilesData, GetRepoFilesResponses, GetSessionData, GetSessionErrors, GetSessionExtensionsData, GetSessionExtensionsErrors, GetSessionExtensionsResponses, GetSessionInsightsData, GetSessionInsightsErrors, GetSessionInsightsResponses, GetSessionResponses, GetSlashCommandsData, GetSlashCommandsResponses, GetToolsData, GetToolsErrors, GetToolsResponses, GetTunnelStatusData, GetTunnelStatusResponses, ImportAppData, ImportAppErrors, ImportAppResponses, ImportSessionData, ImportSessionErrors, ImportSessionResponses, InitConfigData, InitConfigErrors, InitConfigResponses, InspectRunningJobData, InspectRunningJobErrors, InspectRunningJobResponses, KillRunningJobData, KillRunningJobResponses, ListAppsData, ListAppsErrors, ListAppsResponses, ListLocalModelsData, ListLocalModelsResponses, ListModelsData, ListModelsResponses, ListRecipesData, ListRecipesErrors, ListRecipesResponses, ListSchedulesData, ListSchedulesErrors, ListSchedulesResponses, ListSessionsData, ListSessionsErrors, ListSessionsResponses, McpUiProxyData, McpUiProxyErrors, McpUiProxyResponses, ParseRecipeData, ParseRecipeErrors, ParseRecipeResponses, PauseScheduleData, PauseScheduleErrors, PauseScheduleResponses, ProvidersData, ProvidersResponses, ReadAllConfigData, ReadAllConfigResponses, ReadConfigData, ReadConfigErrors, ReadConfigResponses, ReadResourceData, ReadResourceErrors, ReadResourceResponses, RecipeToYamlData, RecipeToYamlErrors, RecipeToYamlResponses, RecoverConfigData, RecoverConfigErrors, RecoverConfigResponses, RemoveConfigData, RemoveConfigErrors, RemoveConfigResponses, RemoveCustomProviderData, RemoveCustomProviderErrors, RemoveCustomProviderResponses, RemoveExtensionData, RemoveExtensionErrors, RemoveExtensionResponses, ReplyData, ReplyErrors, ReplyResponses, ResetPromptData, ResetPromptErrors, ResetPromptResponses, RestartAgentData, RestartAgentErrors, RestartAgentResponses, ResumeAgentData, ResumeAgentErrors, ResumeAgentResponses, RunNowHandlerData, RunNowHandlerErrors, RunNowHandlerResponses, SavePromptData, SavePromptErrors, SavePromptResponses, SaveRecipeData, SaveRecipeErrors, SaveRecipeResponses, ScanRecipeData, ScanRecipeResponses, ScheduleRecipeData, ScheduleRecipeErrors, ScheduleRecipeResponses, SearchHfModelsData, SearchHfModelsErrors, SearchHfModelsResponses, SearchSessionsData, SearchSessionsErrors, SearchSessionsResponses, SendTelemetryEventData, SendTelemetryEventResponses, SessionsHandlerData, SessionsHandlerErrors, SessionsHandlerResponses, SetConfigProviderData, SetRecipeSlashCommandData, SetRecipeSlashCommandErrors, SetRecipeSlashCommandResponses, StartAgentData, StartAgentErrors, StartAgentResponses, StartNanogptSetupData, StartNanogptSetupResponses, StartOpenrouterSetupData, StartOpenrouterSetupResponses, StartTetrateSetupData, StartTetrateSetupResponses, StartTunnelData, StartTunnelErrors, StartTunnelResponses, StatusData, StatusResponses, StopAgentData, StopAgentErrors, StopAgentResponses, StopTunnelData, StopTunnelErrors, StopTunnelResponses, SystemInfoData, SystemInfoResponses, TranscribeDictationData, TranscribeDictationErrors, TranscribeDictationResponses, UnpauseScheduleData, UnpauseScheduleErrors, UnpauseScheduleResponses, UpdateAgentProviderData, UpdateAgentProviderErrors, UpdateAgentProviderResponses, UpdateCustomProviderData, UpdateCustomProviderErrors, UpdateCustomProviderResponses, UpdateFromSessionData, UpdateFromSessionErrors, UpdateFromSessionResponses, UpdateModelSettingsData, UpdateModelSettingsErrors, UpdateModelSettingsResponses, UpdateScheduleData, UpdateScheduleErrors, UpdateScheduleResponses, UpdateSessionData, UpdateSessionErrors, UpdateSessionNameData, UpdateSessionNameErrors, UpdateSessionNameResponses, UpdateSessionResponses, UpdateSessionUserRecipeValuesData, UpdateSessionUserRecipeValuesErrors, UpdateSessionUserRecipeValuesResponses, UpdateWorkingDirData, UpdateWorkingDirErrors, UpdateWorkingDirResponses, UpsertConfigData, UpsertConfigErrors, UpsertConfigResponses, UpsertPermissionsData, UpsertPermissionsErrors, UpsertPermissionsResponses, ValidateConfigData, ValidateConfigErrors, ValidateConfigResponses } from './types.gen'; +import type { AddExtensionData, AddExtensionErrors, AddExtensionResponses, AgentAddExtensionData, AgentAddExtensionErrors, AgentAddExtensionResponses, AgentRemoveExtensionData, AgentRemoveExtensionErrors, AgentRemoveExtensionResponses, BackupConfigData, BackupConfigErrors, BackupConfigResponses, CallToolData, CallToolErrors, CallToolResponses, CancelDownloadData, CancelDownloadErrors, CancelDownloadResponses, CancelLocalModelDownloadData, CancelLocalModelDownloadErrors, CancelLocalModelDownloadResponses, CheckProviderData, CleanupProviderCacheData, CleanupProviderCacheErrors, CleanupProviderCacheResponses, ConfigureProviderOauthData, ConfigureProviderOauthErrors, ConfigureProviderOauthResponses, ConfirmToolActionData, ConfirmToolActionErrors, ConfirmToolActionResponses, CreateCustomProviderData, CreateCustomProviderErrors, CreateCustomProviderResponses, CreateRecipeData, CreateRecipeErrors, CreateRecipeResponses, CreateScheduleData, CreateScheduleErrors, CreateScheduleResponses, DecodeRecipeData, DecodeRecipeErrors, DecodeRecipeResponses, DeleteLocalModelData, DeleteLocalModelErrors, DeleteLocalModelResponses, DeleteModelData, DeleteModelErrors, DeleteModelResponses, DeleteRecipeData, DeleteRecipeErrors, DeleteRecipeResponses, DeleteScheduleData, DeleteScheduleErrors, DeleteScheduleResponses, DeleteSessionData, DeleteSessionErrors, DeleteSessionResponses, DetectProviderData, DetectProviderErrors, DetectProviderResponses, DiagnosticsData, DiagnosticsErrors, DiagnosticsResponses, DownloadHfModelData, DownloadHfModelErrors, DownloadHfModelResponses, DownloadModelData, DownloadModelErrors, DownloadModelResponses, EncodeRecipeData, EncodeRecipeErrors, EncodeRecipeResponses, ExportAppData, ExportAppErrors, ExportAppResponses, ExportSessionData, ExportSessionErrors, ExportSessionResponses, ForkSessionData, ForkSessionErrors, ForkSessionResponses, GetCanonicalModelInfoData, GetCanonicalModelInfoResponses, GetCustomProviderData, GetCustomProviderErrors, GetCustomProviderResponses, GetDictationConfigData, GetDictationConfigResponses, GetDownloadProgressData, GetDownloadProgressErrors, GetDownloadProgressResponses, GetExtensionsData, GetExtensionsErrors, GetExtensionsResponses, GetLocalModelDownloadProgressData, GetLocalModelDownloadProgressErrors, GetLocalModelDownloadProgressResponses, GetModelSettingsData, GetModelSettingsErrors, GetModelSettingsResponses, GetPromptData, GetPromptErrors, GetPromptResponses, GetPromptsData, GetPromptsResponses, GetProviderCatalogData, GetProviderCatalogErrors, GetProviderCatalogResponses, GetProviderCatalogTemplateData, GetProviderCatalogTemplateErrors, GetProviderCatalogTemplateResponses, GetProviderModelsData, GetProviderModelsErrors, GetProviderModelsResponses, GetRepoFilesData, GetRepoFilesResponses, GetSessionData, GetSessionErrors, GetSessionExtensionsData, GetSessionExtensionsErrors, GetSessionExtensionsResponses, GetSessionInsightsData, GetSessionInsightsErrors, GetSessionInsightsResponses, GetSessionResponses, GetSlashCommandsData, GetSlashCommandsResponses, GetToolsData, GetToolsErrors, GetToolsResponses, GetTunnelStatusData, GetTunnelStatusResponses, ImportAppData, ImportAppErrors, ImportAppResponses, ImportSessionData, ImportSessionErrors, ImportSessionResponses, InitConfigData, InitConfigErrors, InitConfigResponses, InspectRunningJobData, InspectRunningJobErrors, InspectRunningJobResponses, KillRunningJobData, KillRunningJobResponses, ListAppsData, ListAppsErrors, ListAppsResponses, ListLocalModelsData, ListLocalModelsResponses, ListModelsData, ListModelsResponses, ListRecipesData, ListRecipesErrors, ListRecipesResponses, ListSchedulesData, ListSchedulesErrors, ListSchedulesResponses, ListSessionsData, ListSessionsErrors, ListSessionsResponses, McpUiProxyData, McpUiProxyErrors, McpUiProxyResponses, ParseRecipeData, ParseRecipeErrors, ParseRecipeResponses, PauseScheduleData, PauseScheduleErrors, PauseScheduleResponses, ProvidersData, ProvidersResponses, ReadAllConfigData, ReadAllConfigResponses, ReadConfigData, ReadConfigErrors, ReadConfigResponses, ReadResourceData, ReadResourceErrors, ReadResourceResponses, RecipeToYamlData, RecipeToYamlErrors, RecipeToYamlResponses, RecoverConfigData, RecoverConfigErrors, RecoverConfigResponses, RemoveConfigData, RemoveConfigErrors, RemoveConfigResponses, RemoveCustomProviderData, RemoveCustomProviderErrors, RemoveCustomProviderResponses, RemoveExtensionData, RemoveExtensionErrors, RemoveExtensionResponses, ReplyData, ReplyErrors, ReplyResponses, ResetPromptData, ResetPromptErrors, ResetPromptResponses, RestartAgentData, RestartAgentErrors, RestartAgentResponses, ResumeAgentData, ResumeAgentErrors, ResumeAgentResponses, RunNowHandlerData, RunNowHandlerErrors, RunNowHandlerResponses, SavePromptData, SavePromptErrors, SavePromptResponses, SaveRecipeData, SaveRecipeErrors, SaveRecipeResponses, ScanRecipeData, ScanRecipeResponses, ScheduleRecipeData, ScheduleRecipeErrors, ScheduleRecipeResponses, SearchHfModelsData, SearchHfModelsErrors, SearchHfModelsResponses, SearchSessionsData, SearchSessionsErrors, SearchSessionsResponses, SendTelemetryEventData, SendTelemetryEventResponses, SessionCancelData, SessionCancelResponses, SessionEventsData, SessionEventsErrors, SessionEventsResponses, SessionReplyData, SessionReplyErrors, SessionReplyResponses, SessionsHandlerData, SessionsHandlerErrors, SessionsHandlerResponses, SetConfigProviderData, SetRecipeSlashCommandData, SetRecipeSlashCommandErrors, SetRecipeSlashCommandResponses, StartAgentData, StartAgentErrors, StartAgentResponses, StartNanogptSetupData, StartNanogptSetupResponses, StartOpenrouterSetupData, StartOpenrouterSetupResponses, StartTetrateSetupData, StartTetrateSetupResponses, StartTunnelData, StartTunnelErrors, StartTunnelResponses, StatusData, StatusResponses, StopAgentData, StopAgentErrors, StopAgentResponses, StopTunnelData, StopTunnelErrors, StopTunnelResponses, SystemInfoData, SystemInfoResponses, TranscribeDictationData, TranscribeDictationErrors, TranscribeDictationResponses, UnpauseScheduleData, UnpauseScheduleErrors, UnpauseScheduleResponses, UpdateAgentProviderData, UpdateAgentProviderErrors, UpdateAgentProviderResponses, UpdateCustomProviderData, UpdateCustomProviderErrors, UpdateCustomProviderResponses, UpdateFromSessionData, UpdateFromSessionErrors, UpdateFromSessionResponses, UpdateModelSettingsData, UpdateModelSettingsErrors, UpdateModelSettingsResponses, UpdateScheduleData, UpdateScheduleErrors, UpdateScheduleResponses, UpdateSessionData, UpdateSessionErrors, UpdateSessionNameData, UpdateSessionNameErrors, UpdateSessionNameResponses, UpdateSessionResponses, UpdateSessionUserRecipeValuesData, UpdateSessionUserRecipeValuesErrors, UpdateSessionUserRecipeValuesResponses, UpdateWorkingDirData, UpdateWorkingDirErrors, UpdateWorkingDirResponses, UpsertConfigData, UpsertConfigErrors, UpsertConfigResponses, UpsertPermissionsData, UpsertPermissionsErrors, UpsertPermissionsResponses, ValidateConfigData, ValidateConfigErrors, ValidateConfigResponses } from './types.gen'; export type Options = Options2 & { /** @@ -509,6 +509,26 @@ export const getSessionInsights = (options export const searchSessions = (options: Options) => (options.client ?? client).get({ url: '/sessions/search', ...options }); +export const sessionCancel = (options: Options) => (options.client ?? client).post({ + url: '/sessions/{id}/cancel', + ...options, + headers: { + 'Content-Type': 'application/json', + ...options.headers + } +}); + +export const sessionEvents = (options: Options) => (options.client ?? client).sse.get({ url: '/sessions/{id}/events', ...options }); + +export const sessionReply = (options: Options) => (options.client ?? client).post({ + url: '/sessions/{id}/reply', + ...options, + headers: { + 'Content-Type': 'application/json', + ...options.headers + } +}); + export const deleteSession = (options: Options) => (options.client ?? client).delete({ url: '/sessions/{session_id}', ...options }); export const getSession = (options: Options) => (options.client ?? client).get({ url: '/sessions/{session_id}', ...options }); diff --git a/ui/desktop/src/api/types.gen.ts b/ui/desktop/src/api/types.gen.ts index 79b427041f..10b7bfeacc 100644 --- a/ui/desktop/src/api/types.gen.ts +++ b/ui/desktop/src/api/types.gen.ts @@ -59,6 +59,10 @@ export type CallToolResponse = { structuredContent?: unknown; }; +export type CancelRequest = { + request_id: string; +}; + export type ChatRequest = { /** * Override the server's conversation history. Only use this when you need absolute control @@ -696,6 +700,9 @@ export type MessageEvent = { } | { conversation: Conversation; type: 'UpdateConversation'; +} | { + request_ids: Array; + type: 'ActiveRequests'; } | { type: 'Ping'; }; @@ -1255,6 +1262,21 @@ export type SessionListResponse = { sessions: Array; }; +export type SessionReplyRequest = { + override_conversation?: Array | null; + recipe_name?: string | null; + recipe_version?: string | null; + /** + * Client-generated UUIDv7 identifying this request. + */ + request_id: string; + user_message: Message; +}; + +export type SessionReplyResponse = { + request_id: string; +}; + export type SessionType = 'user' | 'scheduled' | 'sub_agent' | 'hidden' | 'terminal' | 'gateway'; export type SessionsQuery = { @@ -4129,6 +4151,93 @@ export type SearchSessionsResponses = { export type SearchSessionsResponse = SearchSessionsResponses[keyof SearchSessionsResponses]; +export type SessionCancelData = { + body: CancelRequest; + path: { + /** + * Session ID + */ + id: string; + }; + query?: never; + url: '/sessions/{id}/cancel'; +}; + +export type SessionCancelResponses = { + /** + * Cancellation accepted + */ + 200: unknown; +}; + +export type SessionEventsData = { + body?: never; + path: { + /** + * Session ID + */ + id: string; + }; + query?: never; + url: '/sessions/{id}/events'; +}; + +export type SessionEventsErrors = { + /** + * Session not found + */ + 404: unknown; +}; + +export type SessionEventsResponses = { + /** + * SSE event stream + */ + 200: MessageEvent; +}; + +export type SessionEventsResponse = SessionEventsResponses[keyof SessionEventsResponses]; + +export type SessionReplyData = { + body: SessionReplyRequest; + path: { + /** + * Session ID + */ + id: string; + }; + query?: never; + url: '/sessions/{id}/reply'; +}; + +export type SessionReplyErrors = { + /** + * Invalid request + */ + 400: unknown; + /** + * Session not found + */ + 404: unknown; + /** + * Agent not initialized + */ + 424: unknown; + /** + * Internal server error + */ + 500: unknown; +}; + +export type SessionReplyResponses = { + /** + * Request accepted + */ + 200: SessionReplyResponse; +}; + +export type SessionReplyResponse2 = SessionReplyResponses[keyof SessionReplyResponses]; + export type DeleteSessionData = { body?: never; path: { diff --git a/ui/desktop/src/hooks/useChatStream.ts b/ui/desktop/src/hooks/useChatStream.ts index ad7d4d7fc2..6a94b09338 100644 --- a/ui/desktop/src/hooks/useChatStream.ts +++ b/ui/desktop/src/hooks/useChatStream.ts @@ -1,15 +1,15 @@ import { useCallback, useEffect, useMemo, useReducer, useRef } from 'react'; +import { v7 as uuidv7 } from 'uuid'; import { AppEvents } from '../constants/events'; import { ChatState } from '../types/chatState'; -import { toastError } from '../toasts'; import { getSession, Message, - MessageEvent, - reply, resumeAgent, Session, + sessionCancel, + sessionReply, TokenState, updateFromSession, updateSessionUserRecipeValues, @@ -27,6 +27,7 @@ import { import { errorMessage } from '../utils/conversionUtils'; import { showExtensionLoadResults } from '../utils/extensionErrorUtils'; import { maybeHandlePlatformEvent } from '../utils/platform_events'; +import { useSessionEvents, type SessionEvent } from './useSessionEvents'; const resultsCache = new Map(); @@ -217,14 +218,17 @@ function prefersReducedMotion(): boolean { const REDUCED_MOTION_BATCH_INTERVAL = 1000; -async function streamFromResponse( - stream: AsyncIterable, +/** + * Creates an event processor that handles individual SSE events for a request. + * Returns an unsubscribe function and a handler to process events. + */ +function createEventProcessor( initialMessages: Message[], dispatch: React.Dispatch, onFinish: (error?: string) => void, sessionId: string, - signal?: AbortController['signal'] -): Promise { + onReloadNeeded?: () => void, +) { let currentMessages = initialMessages; const reduceMotion = prefersReducedMotion(); let latestTokenState: TokenState | null = null; @@ -266,88 +270,80 @@ async function streamFromResponse( } }; - try { - for await (const event of stream) { - switch (event.type) { - case 'Message': { - const msg = event.message; - currentMessages = pushMessage(currentMessages, msg); + // Returns true if the event is terminal (Finish or Error) + const processEvent = (event: SessionEvent): boolean => { + switch (event.type) { + case 'Message': { + const msg = (event as Record).message as Message; + const tokenState = (event as Record).token_state as TokenState; + currentMessages = pushMessage(currentMessages, msg); - const hasToolConfirmation = msg.content.some( - (content) => - content.type === 'actionRequired' && content.data.actionType === 'toolConfirmation' - ); + const hasToolConfirmation = msg.content.some( + (content) => + content.type === 'actionRequired' && content.data.actionType === 'toolConfirmation' + ); - const hasElicitation = msg.content.some( - (content) => - content.type === 'actionRequired' && content.data.actionType === 'elicitation' - ); + const hasElicitation = msg.content.some( + (content) => + content.type === 'actionRequired' && content.data.actionType === 'elicitation' + ); - if (hasToolConfirmation || hasElicitation) { - maybeUpdateUI(event.token_state, ChatState.WaitingForUserInput, true); - } else if (getCompactingMessage(msg)) { - maybeUpdateUI(event.token_state, ChatState.Compacting); - } else if (getThinkingMessage(msg)) { - maybeUpdateUI(event.token_state, ChatState.Thinking); - } else { - maybeUpdateUI(event.token_state, ChatState.Streaming); - } - break; + if (hasToolConfirmation || hasElicitation) { + maybeUpdateUI(tokenState, ChatState.WaitingForUserInput, true); + } else if (getCompactingMessage(msg)) { + maybeUpdateUI(tokenState, ChatState.Compacting); + } else if (getThinkingMessage(msg)) { + maybeUpdateUI(tokenState, ChatState.Thinking); + } else { + maybeUpdateUI(tokenState, ChatState.Streaming); } - case 'Error': { - flushBatchedUpdates(); - onFinish('Stream error: ' + event.error); - return; - } - case 'Finish': { - flushBatchedUpdates(); - onFinish(); - return; - } - case 'ModelChange': { - break; - } - case 'UpdateConversation': { - currentMessages = event.conversation; - if (!reduceMotion) { - dispatch({ type: 'SET_MESSAGES', payload: event.conversation }); - } else { - hasPendingUpdate = true; - } - break; - } - case 'Notification': { - dispatch({ type: 'ADD_NOTIFICATION', payload: event as NotificationEvent }); - maybeHandlePlatformEvent(event.message, sessionId); - break; - } - case 'Ping': - break; + return false; } + case 'Error': { + flushBatchedUpdates(); + const errorMsg = String((event as Record).error ?? ''); + if (errorMsg.includes('too far behind') && onReloadNeeded) { + // Server indicated we missed events — end streaming without setting + // an error (which would show a blocking error screen), then reload + // the full conversation so the UI reflects the actual state. + onFinish(); + onReloadNeeded(); + } else { + onFinish('Stream error: ' + errorMsg); + } + return true; + } + case 'Finish': { + flushBatchedUpdates(); + onFinish(); + return true; + } + case 'ModelChange': { + return false; + } + case 'UpdateConversation': { + const conversation = (event as Record).conversation as Message[]; + currentMessages = conversation; + if (!reduceMotion) { + dispatch({ type: 'SET_MESSAGES', payload: conversation }); + } else { + hasPendingUpdate = true; + } + return false; + } + case 'Notification': { + dispatch({ type: 'ADD_NOTIFICATION', payload: event as unknown as NotificationEvent }); + maybeHandlePlatformEvent((event as Record).message, sessionId); + return false; + } + case 'Ping': + return false; + default: + return false; } + }; - // If we reach here, the stream ended without a Finish or Error event. - // This happens when the connection drops and retries are exhausted — the - // generator exits its loop without yielding a terminal event. We call - // onFinish() without an error to keep the conversation visible (passing - // an error would trigger a full-page error screen via sessionLoadError), - // then show a toast so the user knows the response may be incomplete. - // If the signal was aborted, the user intentionally stopped streaming, - // so we skip the toast. - flushBatchedUpdates(); - onFinish(); - if (!signal?.aborted) { - toastError({ - title: 'Connection lost', - msg: 'The response may be incomplete. You can try sending your message again.', - }); - } - } catch (error) { - flushBatchedUpdates(); - if (error instanceof Error && error.name !== 'AbortError') { - onFinish('Stream error: ' + errorMessage(error)); - } - } + return processEvent; } export function useChatStream({ @@ -357,14 +353,26 @@ export function useChatStream({ }: UseChatStreamProps): UseChatStreamReturn { const [state, dispatch] = useReducer(streamReducer, initialState); - // Refs for values needed in callbacks without causing re-renders - const abortControllerRef = useRef(null); + // Long-lived SSE connection for this session + const { addListener, setActiveRequestsHandler } = useSessionEvents(sessionId); + + // Track the active request for cancellation (includes the session that started it) + const activeRequestIdRef = useRef(null); + const activeRequestSessionIdRef = useRef(null); + const activeAbortRef = useRef(null); + const activeUnsubscribeRef = useRef<(() => void) | null>(null); const lastInteractionTimeRef = useRef(Date.now()); + // When ActiveRequests fires before resumeAgent populates messages (cold mount), + // defer the reattach until the session is loaded so the event processor has + // the full conversation history. Events are buffered in the meantime. + const pendingReattachRequestIdRef = useRef(null); + const pendingReattachBufferRef = useRef([]); const namePollingRef = useRef | null>(null); // Ref to access latest state in callbacks (avoids stale closures) const stateRef = useRef(state); stateRef.current = state; + const doReattachRef = useRef<((requestId: string, messages: Message[]) => void) | null>(null); useEffect(() => { return () => { @@ -383,6 +391,10 @@ export function useChatStream({ const onFinish = useCallback( async (error?: string): Promise => { + // Note: SSE listener/ref cleanup is handled by the terminal-event + // handler in each listener closure (which guards on requestId) so + // that overlapping requests don't clobber each other's state. + if (namePollingRef.current) { clearTimeout(namePollingRef.current); namePollingRef.current = null; @@ -404,13 +416,10 @@ export function useChatStream({ } // Refresh session name after each reply for the first 3 user messages - // The backend regenerates the name after each of the first 3 user messages - // to refine it as more context becomes available if (!error && sessionId) { const currentState = stateRef.current; const userMessageCount = currentState.messages.filter((m) => m.role === 'user').length; - // Only refresh for the first 3 user messages if (userMessageCount <= 3) { try { const response = await getSession({ @@ -424,7 +433,6 @@ export function useChatStream({ ? { ...currentState.session, name: response.data.name } : undefined, }); - // Notify sidebar of the name change window.dispatchEvent( new CustomEvent(AppEvents.SESSION_RENAMED, { detail: { sessionId, newName: response.data.name }, @@ -432,7 +440,6 @@ export function useChatStream({ ); } } catch (refreshError) { - // Silently fail - this is a nice-to-have feature console.warn('Failed to refresh session name:', refreshError); } } @@ -443,6 +450,208 @@ export function useChatStream({ [onStreamFinish, sessionId] ); + // Reload the full conversation from the server, e.g. after the SSE + // stream indicates the client fell too far behind the replay buffer. + const reloadConversation = useCallback(() => { + getSession({ + path: { session_id: sessionId }, + throwOnError: true, + }).then((response) => { + const session = response.data as Session; + if (session?.conversation) { + dispatch({ type: 'SET_MESSAGES', payload: session.conversation }); + } + }).catch((e) => { + console.warn('Failed to reload conversation after buffer overflow:', e); + }); + }, [sessionId]); + + // Perform the actual reattach: wire up an event processor and listener + // for a request that is already in-flight on the server. + const doReattach = useCallback( + (requestId: string, messages: Message[]) => { + activeRequestIdRef.current = requestId; + activeRequestSessionIdRef.current = sessionId; + pendingReattachRequestIdRef.current = null; + + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Streaming }); + dispatch({ type: 'SET_SESSION_LOAD_ERROR', payload: undefined }); + + const processEvent = createEventProcessor( + messages, + dispatch, + onFinish, + sessionId, + reloadConversation, + ); + + // Replay any events that were buffered during cold-mount wait + const buffered = pendingReattachBufferRef.current; + pendingReattachBufferRef.current = []; + let finished = false; + for (const event of buffered) { + if (processEvent(event)) { + finished = true; + break; + } + } + + if (finished) { + // The reply already completed while we were waiting for session load. + // Clean up — the buffering listener will be replaced below but the + // old one captured into activeUnsubscribeRef should be removed. + if (activeUnsubscribeRef.current) { + activeUnsubscribeRef.current(); + activeUnsubscribeRef.current = null; + } + activeRequestIdRef.current = null; + activeRequestSessionIdRef.current = null; + return; + } + + // Replace the buffering listener with a real processing listener + if (activeUnsubscribeRef.current) { + activeUnsubscribeRef.current(); + } + const unsubscribe = addListener(requestId, (event) => { + const isTerminal = processEvent(event); + if (isTerminal) { + unsubscribe(); + if (activeRequestIdRef.current === requestId) { + activeUnsubscribeRef.current = null; + activeRequestIdRef.current = null; + activeRequestSessionIdRef.current = null; + } + } + }); + activeUnsubscribeRef.current = unsubscribe; + }, + [sessionId, addListener, onFinish, reloadConversation], + ); + doReattachRef.current = doReattach; + + // Reattach to in-flight replies discovered via the SSE ActiveRequests event. + // This handles the case where the chat view remounts while a reply is still + // running on the server — the new hook instance picks up the existing request + // and starts processing its events. + useEffect(() => { + setActiveRequestsHandler((requestIds: string[]) => { + // Only reattach if we don't already have an active request + if (activeRequestIdRef.current) return; + if (requestIds.length === 0) return; + + // Reattach to the first (most recent) active request. + // Multiple concurrent requests per session aren't supported in the UI. + const requestId = requestIds[0]; + const currentMessages = stateRef.current.messages; + + if (currentMessages.length === 0) { + // Cold mount: resumeAgent hasn't populated messages yet. + // Defer event processing until session load completes so the + // processor starts with the full conversation history. + // Register a buffering listener NOW so replayed events aren't + // lost while we wait. + pendingReattachRequestIdRef.current = requestId; + pendingReattachBufferRef.current = []; + activeRequestIdRef.current = requestId; + activeRequestSessionIdRef.current = sessionId; + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Streaming }); + dispatch({ type: 'SET_SESSION_LOAD_ERROR', payload: undefined }); + + const unsubscribe = addListener(requestId, (event) => { + pendingReattachBufferRef.current.push(event); + }); + activeUnsubscribeRef.current = unsubscribe; + return; + } + + doReattach(requestId, currentMessages); + }); + + return () => { + setActiveRequestsHandler(null); + }; + }, [sessionId, addListener, onFinish, reloadConversation, setActiveRequestsHandler, doReattach]); + + /** + * Submit a message via the new POST+SSE pattern. + * 1. Generate request_id + * 2. Register SSE listener BEFORE POST (no race condition) + * 3. POST to /sessions/{id}/reply + * 4. Events arrive on the long-lived SSE connection + */ + const submitToSession = useCallback( + async ( + targetSessionId: string, + userMessage: Message, + currentMessages: Message[], + overrideConversation?: Message[], + recipeName?: string, + recipeVersion?: string, + ) => { + const requestId = uuidv7(); + const abortController = new AbortController(); + activeRequestIdRef.current = requestId; + activeRequestSessionIdRef.current = targetSessionId; + activeAbortRef.current = abortController; + + // Create event processor and register listener BEFORE the POST + const processEvent = createEventProcessor( + currentMessages, + dispatch, + onFinish, + targetSessionId, + reloadConversation, + ); + + const unsubscribe = addListener(requestId, (event) => { + const isTerminal = processEvent(event); + if (isTerminal) { + unsubscribe(); + // Only clear global refs if this request is still the active one. + // A newer request may have already replaced them. + if (activeRequestIdRef.current === requestId) { + activeUnsubscribeRef.current = null; + activeRequestIdRef.current = null; + activeRequestSessionIdRef.current = null; + activeAbortRef.current = null; + } + } + }); + activeUnsubscribeRef.current = unsubscribe; + + try { + await sessionReply({ + path: { id: targetSessionId }, + body: { + request_id: requestId, + user_message: userMessage, + override_conversation: overrideConversation, + recipe_name: recipeName, + recipe_version: recipeVersion, + }, + signal: abortController.signal, + throwOnError: true, + }); + } catch (error) { + // Abort is expected when stopStreaming races with the POST + if (abortController.signal.aborted) return; + // POST failed — clean up listener and report error. + // Only clear global refs if this request is still the active one; + // a newer request may have already replaced them. + unsubscribe(); + if (activeRequestIdRef.current === requestId) { + activeUnsubscribeRef.current = null; + activeRequestIdRef.current = null; + activeRequestSessionIdRef.current = null; + activeAbortRef.current = null; + } + onFinish('Submit error: ' + errorMessage(error)); + } + }, + [addListener, onFinish, reloadConversation] + ); + // Load session on mount or sessionId change useEffect(() => { if (!sessionId) return; @@ -494,12 +703,38 @@ export function useChatStream({ showExtensionLoadResults(extensionResults); window.dispatchEvent(new CustomEvent(AppEvents.SESSION_EXTENSIONS_LOADED)); - dispatch({ - type: 'SESSION_LOADED', - payload: { - session: loadedSession!, - messages: loadedSession?.conversation || [], - tokenState: { + const pendingRequestId = pendingReattachRequestIdRef.current; + const reattachedToActiveRequest = activeRequestIdRef.current !== null; + + if (pendingRequestId) { + // Cold-mount reattach: ActiveRequests arrived before resumeAgent + // returned. Load session state first, then complete the reattach + // with the full conversation so the event processor has context. + dispatch({ + type: 'SESSION_LOADED', + payload: { + session: loadedSession!, + messages: loadedSession?.conversation || [], + tokenState: { + inputTokens: loadedSession?.input_tokens ?? 0, + outputTokens: loadedSession?.output_tokens ?? 0, + totalTokens: loadedSession?.total_tokens ?? 0, + accumulatedInputTokens: loadedSession?.accumulated_input_tokens ?? 0, + accumulatedOutputTokens: loadedSession?.accumulated_output_tokens ?? 0, + accumulatedTotalTokens: loadedSession?.accumulated_total_tokens ?? 0, + }, + }, + }); + // Now complete the deferred reattach with the loaded messages + doReattachRef.current?.(pendingRequestId, loadedSession?.conversation || []); + } else if (reattachedToActiveRequest) { + // ActiveRequests already wired up an event processor with existing + // messages — only load session metadata, don't overwrite messages + // with the stale DB snapshot. + dispatch({ type: 'SET_SESSION', payload: loadedSession }); + dispatch({ + type: 'SET_TOKEN_STATE', + payload: { inputTokens: loadedSession?.input_tokens ?? 0, outputTokens: loadedSession?.output_tokens ?? 0, totalTokens: loadedSession?.total_tokens ?? 0, @@ -507,8 +742,24 @@ export function useChatStream({ accumulatedOutputTokens: loadedSession?.accumulated_output_tokens ?? 0, accumulatedTotalTokens: loadedSession?.accumulated_total_tokens ?? 0, }, - }, - }); + }); + } else { + dispatch({ + type: 'SESSION_LOADED', + payload: { + session: loadedSession!, + messages: loadedSession?.conversation || [], + tokenState: { + inputTokens: loadedSession?.input_tokens ?? 0, + outputTokens: loadedSession?.output_tokens ?? 0, + totalTokens: loadedSession?.total_tokens ?? 0, + accumulatedInputTokens: loadedSession?.accumulated_input_tokens ?? 0, + accumulatedOutputTokens: loadedSession?.accumulated_output_tokens ?? 0, + accumulatedTotalTokens: loadedSession?.accumulated_total_tokens ?? 0, + }, + }, + }); + } listApps({ throwOnError: true, @@ -535,7 +786,6 @@ export function useChatStream({ const { msg: userMessage, images } = input; const currentState = stateRef.current; - // Guard: Don't submit if session hasn't been loaded yet if (!currentState.session || currentState.chatState === ChatState.LoadingConversation) { return; } @@ -543,7 +793,6 @@ export function useChatStream({ const hasExistingMessages = currentState.messages.length > 0; const hasNewMessage = userMessage.trim().length > 0 || images.length > 0; - // Don't submit if there's no message and no conversation to continue if (!hasNewMessage && !hasExistingMessages) { return; } @@ -554,10 +803,8 @@ export function useChatStream({ if (!hasExistingMessages && hasNewMessage) { window.dispatchEvent(new CustomEvent(AppEvents.SESSION_CREATED)); - // Start polling for session name update during streaming - // The backend generates the name in parallel with the response const pollForName = async (attempts = 0) => { - if (attempts >= 20) return; // Max 20 attempts (10 seconds) + if (attempts >= 20) return; try { const response = await getSession({ @@ -568,7 +815,6 @@ export function useChatStream({ const currentName = currentState.session?.name; const newName = response.data?.name; - // Check if name has changed from the initial name if (newName && newName !== currentName) { dispatch({ type: 'SET_SESSION', @@ -581,13 +827,12 @@ export function useChatStream({ detail: { sessionId, newName }, }) ); - return; // Stop polling once name is updated + return; } } catch { // Silently continue polling } - // Continue polling if still streaming const latestState = stateRef.current; if ( latestState.chatState === ChatState.Streaming || @@ -598,7 +843,6 @@ export function useChatStream({ } }; - // Start polling after a short delay to give backend time to start name generation namePollingRef.current = setTimeout(() => pollForName(0), 1000); } @@ -614,38 +858,10 @@ export function useChatStream({ } dispatch({ type: 'START_STREAMING' }); - abortControllerRef.current = new AbortController(); - try { - const { stream } = await reply({ - body: { - session_id: sessionId, - user_message: newMessage, - }, - throwOnError: true, - signal: abortControllerRef.current.signal, - sseMaxRetryAttempts: 0, - }); - - await streamFromResponse( - stream, - currentMessages, - dispatch, - onFinish, - sessionId, - abortControllerRef.current.signal - ); - } catch (error) { - // AbortError is expected when user stops streaming - if (error instanceof Error && error.name === 'AbortError') { - // Silently handle abort - } else { - // Unexpected error during fetch setup (streamFromResponse handles its own errors) - onFinish('Submit error: ' + errorMessage(error)); - } - } + await submitToSession(sessionId, newMessage, currentMessages); }, - [sessionId, onFinish] + [sessionId, submitToSession] ); const submitElicitationResponse = useCallback( @@ -663,36 +879,10 @@ export function useChatStream({ dispatch({ type: 'SET_MESSAGES', payload: currentMessages }); dispatch({ type: 'START_STREAMING' }); - abortControllerRef.current = new AbortController(); - try { - const { stream } = await reply({ - body: { - session_id: sessionId, - user_message: responseMessage, - }, - throwOnError: true, - signal: abortControllerRef.current.signal, - sseMaxRetryAttempts: 0, - }); - - await streamFromResponse( - stream, - currentMessages, - dispatch, - onFinish, - sessionId, - abortControllerRef.current.signal - ); - } catch (error) { - if (error instanceof Error && error.name === 'AbortError') { - // Silently handle abort - } else { - onFinish('Submit error: ' + errorMessage(error)); - } - } + await submitToSession(sessionId, responseMessage, currentMessages); }, - [sessionId, onFinish] + [sessionId, submitToSession] ); const setRecipeUserParams = useCallback( @@ -709,7 +899,6 @@ export function useChatStream({ }, throwOnError: true, }); - // TODO(Douwe): get this from the server instead of emulating it here dispatch({ type: 'SET_SESSION', payload: { @@ -728,9 +917,6 @@ export function useChatStream({ ); useEffect(() => { - // This should happen on the server when the session is loaded or changed - // use session.id to support changing of sessions rather than depending on the - // stable sessionId. if (state.session) { updateFromSession({ body: { @@ -742,7 +928,34 @@ export function useChatStream({ }, [state.session]); const stopStreaming = useCallback(() => { - abortControllerRef.current?.abort(); + const requestId = activeRequestIdRef.current; + const requestSessionId = activeRequestSessionIdRef.current; + + // Abort the in-flight POST so the reply never starts if cancel wins the race + if (activeAbortRef.current) { + activeAbortRef.current.abort(); + activeAbortRef.current = null; + } + + if (requestId && requestSessionId) { + // Cancel against the session that originally started the request, + // not the current sessionId (which may have changed if user navigated). + sessionCancel({ + path: { id: requestSessionId }, + body: { request_id: requestId }, + }).catch((e) => { + console.warn('Failed to cancel request:', e); + }); + } + + // Clean up listener + if (activeUnsubscribeRef.current) { + activeUnsubscribeRef.current(); + activeUnsubscribeRef.current = null; + } + activeRequestIdRef.current = null; + activeRequestSessionIdRef.current = null; + dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Idle }); lastInteractionTimeRef.current = Date.now(); }, []); @@ -810,34 +1023,7 @@ export function useChatStream({ dispatch({ type: 'SET_MESSAGES', payload: messagesForUI }); dispatch({ type: 'START_STREAMING' }); - abortControllerRef.current = new AbortController(); - - try { - const { stream } = await reply({ - body: { - session_id: targetSessionId, - user_message: updatedUserMessage, - }, - throwOnError: true, - signal: abortControllerRef.current.signal, - sseMaxRetryAttempts: 0, - }); - - await streamFromResponse( - stream, - messagesForUI, - dispatch, - onFinish, - targetSessionId, - abortControllerRef.current.signal - ); - } catch (error) { - if (error instanceof Error && error.name === 'AbortError') { - dispatch({ type: 'SET_CHAT_STATE', payload: ChatState.Idle }); - } else { - throw error; - } - } + await submitToSession(targetSessionId, updatedUserMessage, messagesForUI); } else { await handleSubmit({ msg: newContent, images: [] }); } @@ -853,7 +1039,7 @@ export function useChatStream({ }); } }, - [sessionId, handleSubmit, onFinish] + [sessionId, handleSubmit, submitToSession] ); const setChatState = useCallback((newState: ChatState) => { diff --git a/ui/desktop/src/hooks/useSessionEvents.ts b/ui/desktop/src/hooks/useSessionEvents.ts new file mode 100644 index 0000000000..95c74947cb --- /dev/null +++ b/ui/desktop/src/hooks/useSessionEvents.ts @@ -0,0 +1,196 @@ +import { useEffect, useRef, useState, useCallback } from 'react'; +import { sessionEvents, type MessageEvent } from '../api'; + +/** + * An SSE event with an optional request_id (added by the server at the + * SSE framing layer, not part of the generated MessageEvent type). + */ +export type SessionEvent = MessageEvent & { + request_id?: string; + /** Chat-level request UUID used for routing events to the correct handler. */ + chat_request_id?: string; +}; + +type EventHandler = (event: SessionEvent) => void; +type ActiveRequestsHandler = (requestIds: string[]) => void; + +export function useSessionEvents(sessionId: string) { + const listenersRef = useRef(new Map>()); + const activeRequestsHandlerRef = useRef(null); + const abortRef = useRef(null); + const [connected, setConnected] = useState(false); + + useEffect(() => { + if (!sessionId) return; + + const abortController = new AbortController(); + abortRef.current = abortController; + + (async () => { + let retryDelay = 500; + const MAX_RETRY_DELAY = 10_000; + const MAX_CONSECUTIVE_ERRORS = 10; + let consecutiveErrors = 0; + let lastEventId: string | undefined; + + while (!abortController.signal.aborted) { + try { + const { stream } = await sessionEvents({ + path: { id: sessionId }, + signal: abortController.signal, + headers: lastEventId ? { 'Last-Event-ID': lastEventId } : undefined, + // Disable the inner retry loop so errors surface to our outer + // loop which tracks consecutive failures and notifies listeners. + sseMaxRetryAttempts: 1, + onSseEvent: (event) => { + if (event.id) { + lastEventId = event.id; + } + }, + }); + + let receivedEvent = false; + + for await (const event of stream) { + if (abortController.signal.aborted) break; + + // Only mark as connected after the first real event arrives, + // since the HTTP request doesn't happen until iteration starts. + if (!receivedEvent) { + receivedEvent = true; + setConnected(true); + retryDelay = 500; + consecutiveErrors = 0; + } + + // The server adds chat_request_id (the chat UUID) and request_id + // to the JSON at the SSE framing layer. Route using chat_request_id + // so that Notification events (which carry their own MCP tool-call + // request_id) still reach the correct handler. + const sessionEvent = event as SessionEvent; + const routingId = sessionEvent.chat_request_id ?? sessionEvent.request_id; + + // ActiveRequests events notify the client about in-flight requests + // it can reattach to (e.g. after a remount). + if (sessionEvent.type === 'ActiveRequests') { + const ids = (sessionEvent as unknown as { request_ids: string[] }).request_ids; + activeRequestsHandlerRef.current?.(ids); + continue; + } + + // Server-level errors without a request ID (e.g. "client too far + // behind") affect all active listeners — broadcast to everyone. + if (!routingId && sessionEvent.type === 'Error') { + for (const [id, handlers] of listenersRef.current) { + for (const handler of handlers) { + handler({ ...sessionEvent, request_id: id, chat_request_id: id }); + } + } + } else if (routingId) { + const handlers = listenersRef.current.get(routingId); + if (handlers) { + for (const handler of handlers) { + handler(sessionEvent); + } + } + } + } + + // Stream ended. Reconnect unless we were intentionally aborted. + if (abortController.signal.aborted) break; + setConnected(false); + + // If the stream ended without delivering any events, the connection + // likely failed silently (e.g. 404 with sseMaxRetryAttempts: 1). + // Treat it as an error so backoff and error counting apply. + if (!receivedEvent) { + consecutiveErrors++; + console.warn( + `SSE stream ended with no events (${consecutiveErrors}/${MAX_CONSECUTIVE_ERRORS})` + ); + if (consecutiveErrors >= MAX_CONSECUTIVE_ERRORS) { + console.error('SSE reconnect limit reached, notifying active listeners'); + const errorEvent: SessionEvent = { + type: 'Error', + error: 'Lost connection to server', + } as SessionEvent; + for (const [routingId, handlers] of listenersRef.current) { + for (const handler of handlers) { + handler({ ...errorEvent, request_id: routingId, chat_request_id: routingId }); + } + } + consecutiveErrors = 0; + } + await new Promise((r) => setTimeout(r, retryDelay)); + retryDelay = Math.min(retryDelay * 2, MAX_RETRY_DELAY); + } + } catch (error) { + if (abortController.signal.aborted) break; + consecutiveErrors++; + console.warn( + `SSE connection error (${consecutiveErrors}/${MAX_CONSECUTIVE_ERRORS}), reconnecting:`, + error, + ); + setConnected(false); + + if (consecutiveErrors >= MAX_CONSECUTIVE_ERRORS) { + console.error('SSE reconnect limit reached, notifying active listeners'); + // Send an error event to all active listeners so they can + // transition out of streaming state. Reset the counter so + // the loop keeps reconnecting for future requests. + const errorEvent: SessionEvent = { + type: 'Error', + error: 'Lost connection to server', + } as SessionEvent; + for (const [routingId, handlers] of listenersRef.current) { + for (const handler of handlers) { + handler({ ...errorEvent, request_id: routingId, chat_request_id: routingId }); + } + } + consecutiveErrors = 0; + } + + // Back off before retrying + await new Promise((r) => setTimeout(r, retryDelay)); + retryDelay = Math.min(retryDelay * 2, MAX_RETRY_DELAY); + } + } + + setConnected(false); + })(); + + const listeners = listenersRef.current; + return () => { + abortController.abort(); + abortRef.current = null; + listeners.clear(); + setConnected(false); + }; + }, [sessionId]); + + const addListener = useCallback( + (requestId: string, handler: EventHandler): (() => void) => { + if (!listenersRef.current.has(requestId)) { + listenersRef.current.set(requestId, new Set()); + } + listenersRef.current.get(requestId)!.add(handler); + + return () => { + const set = listenersRef.current.get(requestId); + if (set) { + set.delete(handler); + if (set.size === 0) { + listenersRef.current.delete(requestId); + } + } + }; + }, + [] + ); + + const setActiveRequestsHandler = useCallback((handler: ActiveRequestsHandler | null) => { + activeRequestsHandlerRef.current = handler; + }, []); + + return { connected, addListener, setActiveRequestsHandler }; +}