Separate SSE streaming from POST work submission

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;

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;
 

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,

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<crate::state::AppState>, 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))
 }

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 {
@@ -152,7 +152,7 @@ pub enum MessageEvent {
     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

crates/goose-server/src/routes/session.rs

@@ -296,6 +296,9 @@ async fn delete_session(
             }
         })?;
 
+    // Clean up the event bus to free its replay buffer
+    state.remove_event_bus(&session_id).await;
+
     Ok(StatusCode::OK)
 }