diff --git a/crates/goose-acp/acp-meta.json b/crates/goose-acp/acp-meta.json index f5df690d39..8542de106c 100644 --- a/crates/goose-acp/acp-meta.json +++ b/crates/goose-acp/acp-meta.json @@ -25,26 +25,11 @@ "requestType": "UpdateWorkingDirRequest", "responseType": "EmptyResponse" }, - { - "method": "session/get", - "requestType": "GetSessionRequest", - "responseType": "GetSessionResponse" - }, { "method": "session/delete", "requestType": "DeleteSessionRequest", "responseType": "EmptyResponse" }, - { - "method": "_goose/session/export", - "requestType": "ExportSessionRequest", - "responseType": "ExportSessionResponse" - }, - { - "method": "_goose/session/import", - "requestType": "ImportSessionRequest", - "responseType": "ImportSessionResponse" - }, { "method": "_goose/config/extensions", "requestType": "GetExtensionsRequest", @@ -89,6 +74,26 @@ "method": "_goose/secret/remove", "requestType": "RemoveSecretRequest", "responseType": "EmptyResponse" + }, + { + "method": "_goose/session/export", + "requestType": "ExportSessionRequest", + "responseType": "ExportSessionResponse" + }, + { + "method": "_goose/session/import", + "requestType": "ImportSessionRequest", + "responseType": "ImportSessionResponse" + }, + { + "method": "_goose/session/archive", + "requestType": "ArchiveSessionRequest", + "responseType": "EmptyResponse" + }, + { + "method": "_goose/session/unarchive", + "requestType": "UnarchiveSessionRequest", + "responseType": "EmptyResponse" } ] } diff --git a/crates/goose-acp/acp-schema.json b/crates/goose-acp/acp-schema.json index b0e2d6ba27..b261dffb06 100644 --- a/crates/goose-acp/acp-schema.json +++ b/crates/goose-acp/acp-schema.json @@ -125,36 +125,6 @@ "x-side": "agent", "x-method": "_goose/working_dir/update" }, - "GetSessionRequest": { - "type": "object", - "properties": { - "sessionId": { - "type": "string" - }, - "includeMessages": { - "type": "boolean", - "default": false - } - }, - "required": [ - "sessionId" - ], - "description": "Get a session by ID.", - "x-side": "agent", - "x-method": "session/get" - }, - "GetSessionResponse": { - "type": "object", - "properties": { - "session": { - "description": "The session object with id, name, working_dir, timestamps, tokens, etc.", - "default": null - } - }, - "description": "Get a session response.", - "x-side": "agent", - "x-method": "session/get" - }, "DeleteSessionRequest": { "type": "object", "properties": { @@ -169,60 +139,6 @@ "x-side": "agent", "x-method": "session/delete" }, - "ExportSessionRequest": { - "type": "object", - "properties": { - "sessionId": { - "type": "string" - } - }, - "required": [ - "sessionId" - ], - "description": "Export a session as a JSON string.", - "x-side": "agent", - "x-method": "_goose/session/export" - }, - "ExportSessionResponse": { - "type": "object", - "properties": { - "data": { - "type": "string" - } - }, - "required": [ - "data" - ], - "description": "Export session response.", - "x-side": "agent", - "x-method": "_goose/session/export" - }, - "ImportSessionRequest": { - "type": "object", - "properties": { - "data": { - "type": "string" - } - }, - "required": [ - "data" - ], - "description": "Import a session from a JSON string.", - "x-side": "agent", - "x-method": "_goose/session/import" - }, - "ImportSessionResponse": { - "type": "object", - "properties": { - "session": { - "description": "The imported session object.", - "default": null - } - }, - "description": "Import session response.", - "x-side": "agent", - "x-method": "_goose/session/import" - }, "GetExtensionsRequest": { "type": "object", "description": "List configured extensions and any warnings.", @@ -458,6 +374,108 @@ "x-side": "agent", "x-method": "_goose/secret/remove" }, + "ExportSessionRequest": { + "type": "object", + "properties": { + "sessionId": { + "type": "string" + } + }, + "required": [ + "sessionId" + ], + "description": "Export a session as a JSON string.", + "x-side": "agent", + "x-method": "_goose/session/export" + }, + "ExportSessionResponse": { + "type": "object", + "properties": { + "data": { + "type": "string" + } + }, + "required": [ + "data" + ], + "description": "Export session response — raw JSON of the goose session with `conversation`.", + "x-side": "agent", + "x-method": "_goose/session/export" + }, + "ImportSessionRequest": { + "type": "object", + "properties": { + "data": { + "type": "string" + } + }, + "required": [ + "data" + ], + "description": "Import a session from a JSON string.", + "x-side": "agent", + "x-method": "_goose/session/import" + }, + "ImportSessionResponse": { + "type": "object", + "properties": { + "sessionId": { + "type": "string" + }, + "title": { + "type": [ + "string", + "null" + ] + }, + "updatedAt": { + "type": [ + "string", + "null" + ] + }, + "messageCount": { + "type": "integer", + "format": "uint64", + "minimum": 0 + } + }, + "required": [ + "sessionId", + "messageCount" + ], + "description": "Import session response — metadata about the newly created session.", + "x-side": "agent", + "x-method": "_goose/session/import" + }, + "ArchiveSessionRequest": { + "type": "object", + "properties": { + "sessionId": { + "type": "string" + } + }, + "required": [ + "sessionId" + ], + "description": "Archive a session (soft delete).", + "x-side": "agent", + "x-method": "_goose/session/archive" + }, + "UnarchiveSessionRequest": { + "type": "object", + "properties": { + "sessionId": { + "type": "string" + } + }, + "required": [ + "sessionId" + ], + "description": "Unarchive a previously archived session.", + "x-side": "agent", + "x-method": "_goose/session/unarchive" + }, "ExtRequest": { "properties": { "id": { @@ -515,15 +533,6 @@ "description": "Params for _goose/working_dir/update", "title": "UpdateWorkingDirRequest" }, - { - "allOf": [ - { - "$ref": "#/$defs/GetSessionRequest" - } - ], - "description": "Params for session/get", - "title": "GetSessionRequest" - }, { "allOf": [ { @@ -533,24 +542,6 @@ "description": "Params for session/delete", "title": "DeleteSessionRequest" }, - { - "allOf": [ - { - "$ref": "#/$defs/ExportSessionRequest" - } - ], - "description": "Params for _goose/session/export", - "title": "ExportSessionRequest" - }, - { - "allOf": [ - { - "$ref": "#/$defs/ImportSessionRequest" - } - ], - "description": "Params for _goose/session/import", - "title": "ImportSessionRequest" - }, { "allOf": [ { @@ -631,6 +622,42 @@ ], "description": "Params for _goose/secret/remove", "title": "RemoveSecretRequest" + }, + { + "allOf": [ + { + "$ref": "#/$defs/ExportSessionRequest" + } + ], + "description": "Params for _goose/session/export", + "title": "ExportSessionRequest" + }, + { + "allOf": [ + { + "$ref": "#/$defs/ImportSessionRequest" + } + ], + "description": "Params for _goose/session/import", + "title": "ImportSessionRequest" + }, + { + "allOf": [ + { + "$ref": "#/$defs/ArchiveSessionRequest" + } + ], + "description": "Params for _goose/session/archive", + "title": "ArchiveSessionRequest" + }, + { + "allOf": [ + { + "$ref": "#/$defs/UnarchiveSessionRequest" + } + ], + "description": "Params for _goose/session/unarchive", + "title": "UnarchiveSessionRequest" } ] }, @@ -686,30 +713,6 @@ ], "title": "ReadResourceResponse" }, - { - "allOf": [ - { - "$ref": "#/$defs/GetSessionResponse" - } - ], - "title": "GetSessionResponse" - }, - { - "allOf": [ - { - "$ref": "#/$defs/ExportSessionResponse" - } - ], - "title": "ExportSessionResponse" - }, - { - "allOf": [ - { - "$ref": "#/$defs/ImportSessionResponse" - } - ], - "title": "ImportSessionResponse" - }, { "allOf": [ { @@ -749,6 +752,22 @@ } ], "title": "CheckSecretResponse" + }, + { + "allOf": [ + { + "$ref": "#/$defs/ExportSessionResponse" + } + ], + "title": "ExportSessionResponse" + }, + { + "allOf": [ + { + "$ref": "#/$defs/ImportSessionResponse" + } + ], + "title": "ImportSessionResponse" } ] }, diff --git a/crates/goose-acp/src/adapters.rs b/crates/goose-acp/src/adapters.rs index 83813bd0f0..13ced2ffc3 100644 --- a/crates/goose-acp/src/adapters.rs +++ b/crates/goose-acp/src/adapters.rs @@ -6,7 +6,6 @@ use std::{ task::{Context, Poll}, }; use tokio::sync::mpsc; -use tracing::error; /// Converts an mpsc::Receiver to AsyncRead /// Each message is terminated with a newline for JSON-RPC framing @@ -61,15 +60,18 @@ impl tokio::io::AsyncRead for ReceiverToAsyncRead { } } -/// Converts an mpsc::Sender to AsyncWrite -/// Splits incoming data on newlines for JSON-RPC framing +/// Converts an unbounded mpsc::Sender to AsyncWrite. +/// Splits incoming data on newlines for JSON-RPC framing. +/// +/// Uses an unbounded sender so that bursts of outgoing messages (e.g. replaying +/// a long session history) are never silently dropped due to backpressure. pub(crate) struct SenderToAsyncWrite { - tx: mpsc::Sender, + tx: mpsc::UnboundedSender, buffer: Vec, } impl SenderToAsyncWrite { - pub(crate) fn new(tx: mpsc::Sender) -> Self { + pub(crate) fn new(tx: mpsc::UnboundedSender) -> Self { Self { tx, buffer: Vec::new(), @@ -89,24 +91,11 @@ impl tokio::io::AsyncWrite for SenderToAsyncWrite { let line = String::from_utf8_lossy(&self.buffer[..pos]).to_string(); self.buffer.drain(..=pos); - if !line.is_empty() { - if let Err(e) = self.tx.try_send(line.clone()) { - match e { - mpsc::error::TrySendError::Full(_) => { - let truncated: String = line.chars().take(100).collect(); - error!( - "Channel full, dropping message (backpressure): {}", - truncated - ); - } - mpsc::error::TrySendError::Closed(_) => { - return Poll::Ready(Err(std::io::Error::new( - std::io::ErrorKind::BrokenPipe, - "Channel closed", - ))); - } - } - } + if !line.is_empty() && self.tx.send(line).is_err() { + return Poll::Ready(Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "Channel closed", + ))); } } diff --git a/crates/goose-acp/src/server.rs b/crates/goose-acp/src/server.rs index fb1f0be9d9..3031c57f78 100644 --- a/crates/goose-acp/src/server.rs +++ b/crates/goose-acp/src/server.rs @@ -16,7 +16,6 @@ use goose::config::paths::Paths; use goose::config::permission::PermissionManager; use goose::config::{Config, GooseMode}; use goose::conversation::message::{ActionRequiredData, Message, MessageContent}; -use goose::conversation::Conversation; use goose::mcp_utils::ToolResult; use goose::permission::permission_confirmation::PrincipalType; use goose::permission::{Permission, PermissionConfirmation}; @@ -29,15 +28,15 @@ use sacp::schema::{ AgentCapabilities, AuthMethod, AuthMethodAgent, AuthenticateRequest, AuthenticateResponse, BlobResourceContents, CancelNotification, CloseSessionRequest, CloseSessionResponse, ConfigOptionUpdate, Content, ContentBlock, ContentChunk, CurrentModeUpdate, EmbeddedResource, - EmbeddedResourceResource, FileSystemCapabilities, ImageContent, InitializeRequest, - InitializeResponse, ListSessionsRequest, ListSessionsResponse, LoadSessionRequest, - LoadSessionResponse, McpCapabilities, McpServer, Meta, ModelId, ModelInfo, NewSessionRequest, - NewSessionResponse, PermissionOption, PermissionOptionKind, PromptCapabilities, PromptRequest, - PromptResponse, RequestPermissionOutcome, RequestPermissionRequest, ResourceLink, - SessionCapabilities, SessionCloseCapabilities, SessionConfigOption, - SessionConfigOptionCategory, SessionConfigSelectOption, SessionId, SessionInfo, - SessionListCapabilities, SessionMode, SessionModeId, SessionModeState, SessionModelState, - SessionNotification, SessionUpdate, SetSessionConfigOptionRequest, + EmbeddedResourceResource, FileSystemCapabilities, ForkSessionRequest, ForkSessionResponse, + ImageContent, InitializeRequest, InitializeResponse, ListSessionsRequest, ListSessionsResponse, + LoadSessionRequest, LoadSessionResponse, McpCapabilities, McpServer, Meta, ModelId, ModelInfo, + NewSessionRequest, NewSessionResponse, PermissionOption, PermissionOptionKind, + PromptCapabilities, PromptRequest, PromptResponse, RequestPermissionOutcome, + RequestPermissionRequest, ResourceLink, SessionCapabilities, SessionCloseCapabilities, + SessionConfigOption, SessionConfigOptionCategory, SessionConfigSelectOption, SessionId, + SessionInfo, SessionListCapabilities, SessionMode, SessionModeId, SessionModeState, + SessionModelState, SessionNotification, SessionUpdate, SetSessionConfigOptionRequest, SetSessionConfigOptionResponse, SetSessionModeRequest, SetSessionModeResponse, SetSessionModelRequest, SetSessionModelResponse, StopReason, TextContent, TextResourceContents, ToolCall, ToolCallContent, ToolCallId, ToolCallLocation, ToolCallStatus, ToolCallUpdate, @@ -70,11 +69,49 @@ pub type AcpProviderFactory = Arc< const DEFAULT_PROVIDER_ID: &str = "goose"; const DEFAULT_PROVIDER_LABEL: &str = "Goose (Default)"; +/// In-memory state for an active ACP session. +/// +/// ## Terminology (temporary, until all clients migrate to ACP) +/// +/// The ACP protocol uses "session" to mean the conversation as the human sees it — +/// a durable, append-only exchange of messages. Internally, goose also has a concept +/// called "Session" (the `sessions` DB table) which represents the agent's working +/// state: the message list the LLM sees, compaction state, provider binding, etc. +/// +/// To bridge these two worlds without rewriting the existing Session model: +/// - **Thread** (`threads` table) = the ACP session. The `sessionId` that ACP clients +/// see is actually a thread ID. Threads own the human-visible message log. +/// - **Session** (`sessions` table) = an internal execution context. A thread may have +/// many sessions over its lifetime (e.g. when the provider or persona changes). +/// Clients never see or manage these directly. +/// +/// The `sessions` HashMap below is keyed by **thread ID** (= ACP session ID). +/// The `internal_session_id` field tracks which goose Session is currently active. struct GooseAcpSession { - agent: Arc, - messages: Conversation, + agent: AgentHandle, + internal_session_id: String, tool_requests: HashMap, cancel_token: Option, + /// Working directory set while the agent was still loading. + /// Applied once the agent becomes ready. + pending_working_dir: Option, +} + +/// The agent may still be initializing in the background (extension loading, +/// provider setup). Callers that need the live agent (e.g. `on_prompt`) await +/// the handle; callers that only need the session metadata can proceed without it. +enum AgentHandle { + Ready(Arc), + Loading(tokio::sync::watch::Receiver, String>>>), +} + +struct AgentSetupRequest { + session_id: SessionId, + goose_session: Session, + mcp_servers: Vec, + /// Pre-resolved provider name + model config (from config, no network). + /// When present the spawn skips re-deriving these from config. + resolved_provider: Option<(String, goose::model::ModelConfig)>, } pub struct GooseAcpAgent { @@ -85,6 +122,7 @@ pub struct GooseAcpAgent { client_terminal: OnceCell, config_dir: std::path::PathBuf, session_manager: Arc, + thread_manager: Arc, permission_manager: Arc, goose_mode: GooseMode, disable_session_naming: bool, @@ -421,6 +459,55 @@ fn session_provider_selection(session: &Session) -> &str { .unwrap_or(DEFAULT_PROVIDER_ID) } +/// Resolve the provider name and model config for a session from an +/// already-loaded `Config`. +async fn resolve_provider_and_model_from_config( + config: &Config, + goose_session: &Session, +) -> Result<(String, goose::model::ModelConfig), String> { + let global_provider = config.get_goose_provider().ok(); + let provider_override = goose_session + .provider_name + .as_deref() + .filter(|p| *p != DEFAULT_PROVIDER_ID); + let provider_name = provider_override + .map(ToOwned::to_owned) + .or_else(|| global_provider.clone()) + .ok_or_else(|| "Missing provider".to_string())?; + let explicitly_switched = + provider_override.is_some() && provider_override != global_provider.as_deref(); + let model_config = match &goose_session.model_config { + Some(mc) => mc.clone(), + None if explicitly_switched => { + let entry = goose::providers::get_from_registry(&provider_name) + .await + .map_err(|e| e.to_string())?; + let default_model = &entry.metadata().default_model; + goose::model::ModelConfig::new(default_model) + .map_err(|e| e.to_string())? + .with_canonical_limits(&provider_name) + } + None => { + let model_id = config.get_goose_model().map_err(|e| e.to_string())?; + goose::model::ModelConfig::new(&model_id) + .map_err(|e| e.to_string())? + .with_canonical_limits(&provider_name) + } + }; + Ok((provider_name, model_config)) +} + +/// Convenience wrapper: reads config from disk, then resolves provider + model. +/// Cheap enough to call from `on_new_session` (file + registry reads, no network). +async fn resolve_provider_and_model( + config_dir: &std::path::Path, + goose_session: &Session, +) -> Result<(String, goose::model::ModelConfig), String> { + let config = + Config::new(config_dir.join(CONFIG_YAML_NAME), "goose").map_err(|e| e.to_string())?; + resolve_provider_and_model_from_config(&config, goose_session).await +} + fn build_mode_state(current_mode: GooseMode) -> Result { let mut available = Vec::with_capacity(GooseMode::VARIANTS.len()); for &name in GooseMode::VARIANTS { @@ -438,6 +525,33 @@ fn build_mode_state(current_mode: GooseMode) -> Result, + mode_state: &SessionModeState, + goose_session: &Session, +) -> (Option, Option>) { + let Ok((ref provider_name, ref mc)) = resolved else { + return (None, None); + }; + let recommended = goose::providers::canonical::recommended_models_from_registry(provider_name); + let available: Vec = recommended + .iter() + .map(|name| ModelInfo::new(ModelId::new(&**name), &**name)) + .collect(); + let ms = SessionModelState::new(ModelId::new(mc.model_name.as_str()), available); + let provider_selection = session_provider_selection(goose_session); + let provider_options = build_provider_options(Some(provider_name.as_str())).await; + let config_options = + build_config_options(mode_state, &ms, provider_selection, provider_options); + (Some(ms), Some(config_options)) +} + fn build_config_options( mode_state: &SessionModeState, model_state: &SessionModelState, @@ -496,6 +610,9 @@ impl GooseAcpAgent { disable_session_naming: bool, ) -> Result { let session_manager = Arc::new(SessionManager::new(data_dir)); + let thread_manager = Arc::new(goose::session::ThreadManager::new( + session_manager.storage().clone(), + )); let permission_manager = Arc::new(PermissionManager::new(config_dir.clone())); Ok(Self { @@ -506,6 +623,7 @@ impl GooseAcpAgent { client_terminal: OnceCell::new(), config_dir, session_manager, + thread_manager, permission_manager, goose_mode, disable_session_naming, @@ -525,107 +643,189 @@ impl GooseAcpAgent { (self.provider_factory)(provider_name.to_string(), model_config, extensions).await } - async fn create_agent_for_session( + fn spawn_agent_setup( &self, - cx: Option<&ConnectionTo>, - session_id: Option<&SessionId>, - goose_mode: Option, - ) -> Result> { - let mode = goose_mode.unwrap_or(self.goose_mode); - let agent = Agent::with_config(AgentConfig::new( - Arc::clone(&self.session_manager), - Arc::clone(&self.permission_manager), - None, - mode, - self.disable_session_naming, - GoosePlatform::GooseCli, - )); - let agent = Arc::new(agent); + cx: &ConnectionTo, + agent_tx: tokio::sync::watch::Sender, String>>>, + req: AgentSetupRequest, + ) { + let AgentSetupRequest { + session_id, + goose_session, + mcp_servers, + resolved_provider, + } = req; - let config_path = self.config_dir.join(CONFIG_YAML_NAME); - let mut extensions = Config::new(&config_path, "goose") - .ok() - .map(|c| get_enabled_extensions_with_config(&c)) - .unwrap_or_default(); - extensions.extend(self.builtins.iter().map(|b| builtin_to_extension_config(b))); + let goose_mode = goose_session.goose_mode; + let internal_session_id = goose_session.id.clone(); + let agent_session_id = SessionId::new(internal_session_id.clone()); - let caps = self + let cx = cx.clone(); + let sessions = Arc::clone(&self.sessions); + let session_manager = Arc::clone(&self.session_manager); + let permission_manager = Arc::clone(&self.permission_manager); + let config_dir = self.config_dir.clone(); + let builtins = self.builtins.clone(); + let client_fs_capabilities = self .client_fs_capabilities .get() .cloned() .unwrap_or_default(); - let terminal = self.client_terminal.get().copied().unwrap_or(false); - let acp_developer = match (cx, session_id) { - (Some(cx), Some(sid)) - if (caps.read_text_file || caps.write_text_file || terminal) - && extensions.iter().any(|e| e.name() == "developer") => - { - let context = agent.extension_manager.get_context().clone(); - let client: Arc = Arc::new(AcpTools { - inner: Arc::new(DeveloperClient::new(context)?), - cx: cx.clone(), - session_id: sid.clone(), - fs_read: caps.read_text_file, - fs_write: caps.write_text_file, - terminal, - }); - let dev_ext = extensions.iter().find(|e| e.name() == "developer"); - let available_tools = dev_ext - .and_then(|e| match e { - ExtensionConfig::Platform { - available_tools, .. - } => Some(available_tools.clone()), - _ => None, - }) + let client_terminal = self.client_terminal.get().copied().unwrap_or(false); + let provider_factory = Arc::clone(&self.provider_factory); + let disable_session_naming = self.disable_session_naming; + + tokio::spawn(async move { + let result: Result<(), String> = async { + let agent = Arc::new(Agent::with_config(AgentConfig::new( + session_manager, + permission_manager, + None, + goose_mode, + disable_session_naming, + GoosePlatform::GooseCli, + ))); + + let config_path = config_dir.join(CONFIG_YAML_NAME); + let mut extensions = Config::new(&config_path, "goose") + .ok() + .map(|c| get_enabled_extensions_with_config(&c)) .unwrap_or_default(); - let def = &PLATFORM_EXTENSIONS["developer"]; - let config = ExtensionConfig::Platform { - name: def.name.into(), - description: def.description.into(), - display_name: Some(def.display_name.into()), - bundled: Some(true), - available_tools, + extensions.extend(builtins.iter().map(|b| builtin_to_extension_config(b))); + + let acp_developer = if (client_fs_capabilities.read_text_file + || client_fs_capabilities.write_text_file + || client_terminal) + && extensions.iter().any(|e| e.name() == "developer") + { + let context = agent.extension_manager.get_context().clone(); + match DeveloperClient::new(context) { + Ok(dev_client) => { + let client: Arc = Arc::new(AcpTools { + inner: Arc::new(dev_client), + cx: cx.clone(), + session_id: session_id.clone(), + fs_read: client_fs_capabilities.read_text_file, + fs_write: client_fs_capabilities.write_text_file, + terminal: client_terminal, + }); + let dev_ext = extensions.iter().find(|e| e.name() == "developer"); + let available_tools = dev_ext + .and_then(|e| match e { + ExtensionConfig::Platform { + available_tools, .. + } => Some(available_tools.clone()), + _ => None, + }) + .unwrap_or_default(); + let def = &PLATFORM_EXTENSIONS["developer"]; + let config = ExtensionConfig::Platform { + name: def.name.into(), + description: def.description.into(), + display_name: Some(def.display_name.into()), + bundled: Some(true), + available_tools, + }; + Some((client, config)) + } + Err(e) => { + warn!(error = %e, "Failed to create developer client"); + None + } + } + } else { + None }; - Some((client, config)) - } - _ => None, - }; - let skip_developer = acp_developer.is_some(); - let sid_str = session_id.map(|s| s.0.to_string()); - if skip_developer { - extensions.retain(|ext| ext.name() != "developer"); - } + let skip_developer = acp_developer.is_some(); + let sid_str = Some(agent_session_id.0.to_string()); - let ext_manager = &agent.extension_manager; - let extension_futures = extensions - .into_iter() - .map(|ext| { - let ext_manager = Arc::clone(ext_manager); - let sid = sid_str.clone(); - async move { - let name = ext.name().to_string(); - match ext_manager - .add_extension(ext, None, None, sid.as_deref()) - .await - { - Ok(_) => info!(extension = %name, "extension loaded"), - Err(e) => warn!(extension = %name, error = %e, "extension load failed"), + if skip_developer { + extensions.retain(|ext| ext.name() != "developer"); + } + + let ext_manager = &agent.extension_manager; + let extension_futures = extensions + .into_iter() + .map(|ext| { + let ext_manager = Arc::clone(ext_manager); + let sid = sid_str.clone(); + async move { + let name = ext.name().to_string(); + match ext_manager + .add_extension(ext, None, None, sid.as_deref()) + .await + { + Ok(_) => info!(extension = %name, "extension loaded"), + Err(e) => { + warn!(extension = %name, error = %e, "extension load failed") + } + } + } + }) + .collect::>(); + futures::future::join_all(extension_futures).await; + + if let Some((client, config)) = acp_developer { + let info = client.get_info().cloned(); + agent + .extension_manager + .add_client("developer".into(), config, client, info, None) + .await; + } + + // Init provider — reuse the pre-resolved name + model when + // available (already computed in on_new_session), otherwise + // fall back to reading config (e.g. load_session path). + let config = Config::new(config_dir.join(CONFIG_YAML_NAME), "goose") + .map_err(|e| e.to_string())?; + let (provider_name, model_config) = match resolved_provider { + Some(resolved) => resolved, + None => resolve_provider_and_model_from_config(&config, &goose_session).await?, + }; + let ext_state = EnabledExtensionsState::extensions_or_default( + Some(&goose_session.extension_data), + &config, + ); + let provider = provider_factory(provider_name.to_string(), model_config, ext_state) + .await + .map_err(|e| e.to_string())?; + agent + .update_provider(provider.clone(), &goose_session.id) + .await + .map_err(|e| e.to_string())?; + + agent + .update_goose_mode(goose_mode, &internal_session_id) + .await + .map_err(|e| e.to_string())?; + + GooseAcpAgent::add_mcp_extensions(&agent, mcp_servers, &internal_session_id) + .await + .map_err(|e| e.to_string())?; + + // Apply any working directory that was set while we were loading. + { + let mut locked = sessions.lock().await; + if let Some(session) = locked.get_mut(session_id.0.as_ref()) { + if let Some(dir) = session.pending_working_dir.take() { + agent.extension_manager.update_working_dir(&dir).await; + } + session.agent = AgentHandle::Ready(agent.clone()); } } - }) - .collect::>(); - futures::future::join_all(extension_futures).await; - if let Some((client, config)) = acp_developer { - let info = client.get_info().cloned(); - agent - .extension_manager - .add_client("developer".into(), config, client, info, None) - .await; - } + let _ = agent_tx.send(Some(Ok(agent))); - Ok(agent) + Ok(()) + } + .await; + + if let Err(e) = &result { + error!(error = %e, "Background agent setup failed"); + let _ = agent_tx.send(Some(Err(e.clone()))); + } + }); } pub async fn has_session(&self, session_id: &str) -> bool { @@ -668,6 +868,7 @@ impl GooseAcpAgent { &self, content_item: &MessageContent, session_id: &SessionId, + agent: &Arc, session: &mut GooseAcpSession, cx: &ConnectionTo, ) -> Result<(), sacp::Error> { @@ -706,7 +907,7 @@ impl GooseAcpAgent { { self.handle_tool_permission_request( cx, - &session.agent, + agent, session_id, id.clone(), tool_name.clone(), @@ -756,7 +957,10 @@ impl GooseAcpAgent { ))?; if let Ok(tool_call) = &tool_request.tool_call { - let agent = session.agent.clone(); + let agent = match &session.agent { + AgentHandle::Ready(a) => a.clone(), + AgentHandle::Loading(_) => return Ok(()), + }; let sid = session_id.clone(); let request_id = tool_request.id.clone(); let cx = cx.clone(); @@ -775,7 +979,7 @@ impl GooseAcpAgent { .unwrap_or_default(); tokio::spawn(async move { - let provider = match agent.provider().await { + let provider: Arc = match agent.provider().await { Ok(p) => p, Err(e) => { warn!("tool call summary: failed to get provider: {e}"); @@ -810,7 +1014,7 @@ impl GooseAcpAgent { let summary: String = response .content .iter() - .filter_map(|c| c.as_text()) + .filter_map(|c: &MessageContent| c.as_text()) .collect::() .trim() .to_string(); @@ -1062,9 +1266,6 @@ impl GooseAcpAgent { ) -> Result { debug!(?args, "new session request"); - // Allow the client to request a specific provider via _meta.provider, - // avoiding the double-create when the client would otherwise call - // _goose/session/provider/update immediately after. let requested_provider = args .meta .as_ref() @@ -1072,10 +1273,94 @@ impl GooseAcpAgent { .and_then(|v| v.as_str()) .map(|s| s.to_string()); + // Create the Thread — this IS the ACP session from the client's perspective. + let thread_metadata = goose::session::ThreadMetadata { + provider_id: requested_provider.clone(), + mode: Some(self.goose_mode.to_string()), + ..Default::default() + }; + let thread = self + .thread_manager + .create_thread( + None, + Some(thread_metadata), + Some(args.cwd.display().to_string()), + ) + .await + .map_err(|e| { + sacp::Error::internal_error().data(format!("Failed to create thread: {}", e)) + })?; + let thread_id = thread.id.clone(); + + // Create the first internal Session linked to this thread. + let goose_session = self + .create_internal_session( + &thread_id, + args.cwd.clone(), + requested_provider.as_deref(), + None, + ) + .await?; + + let internal_session_id = goose_session.id.clone(); + + let (agent_tx, agent_rx) = + tokio::sync::watch::channel::, String>>>(None); + + let session = GooseAcpSession { + agent: AgentHandle::Loading(agent_rx), + internal_session_id: internal_session_id.clone(), + tool_requests: HashMap::new(), + cancel_token: None, + pending_working_dir: None, + }; + self.sessions + .lock() + .await + .insert(thread_id.clone(), session); + + let mode_state = build_mode_state(self.goose_mode)?; + + // Resolve provider + model from config so we can include the current + // model in the response without waiting for the full agent setup. + let resolved = resolve_provider_and_model(&self.config_dir, &goose_session).await; + let (model_state, config_options) = + build_eager_config(&resolved, &mode_state, &goose_session).await; + + self.spawn_agent_setup( + cx, + agent_tx, + AgentSetupRequest { + session_id: SessionId::new(thread_id.clone()), + goose_session, + mcp_servers: args.mcp_servers, + resolved_provider: resolved.ok(), + }, + ); + + let mut response = NewSessionResponse::new(SessionId::new(thread_id)).modes(mode_state); + if let Some(ms) = model_state { + response = response.models(ms); + } + if let Some(co) = config_options { + response = response.config_options(co); + } + Ok(response) + } + + /// Create a new internal goose Session linked to a thread. + /// This is the agent's working state — invisible to ACP clients. + async fn create_internal_session( + &self, + thread_id: &str, + cwd: std::path::PathBuf, + provider_name: Option<&str>, + model_name: Option<&str>, + ) -> Result { let goose_session = self .session_manager .create_session( - args.cwd.clone(), + cwd, "ACP Session".to_string(), SessionType::Acp, self.goose_mode, @@ -1085,138 +1370,67 @@ impl GooseAcpAgent { sacp::Error::internal_error().data(format!("Failed to create session: {}", e)) })?; - if let Some(ref provider_name) = requested_provider { - self.session_manager - .update(&goose_session.id) - .provider_name(provider_name) - .apply() - .await - .map_err(|e| { - sacp::Error::internal_error() - .data(format!("Failed to set provider on session: {}", e)) - })?; + let mut builder = self.session_manager.update(&goose_session.id); + builder = builder.thread_id(Some(thread_id.to_string())); + if let Some(provider) = provider_name { + builder = builder.provider_name(provider); } + if let Some(model) = model_name { + if let Ok(mc) = goose::model::ModelConfig::new(model) { + builder = builder.model_config(mc); + } + } + builder.apply().await.map_err(|e| { + sacp::Error::internal_error().data(format!("Failed to link session to thread: {}", e)) + })?; - // Reload the session so init_provider sees the updated provider_name. - let goose_session = if requested_provider.is_some() { - self.session_manager - .get_session(&goose_session.id, false) - .await - .map_err(|e| { - sacp::Error::internal_error().data(format!("Failed to reload session: {}", e)) - })? - } else { - goose_session - }; - - let session_id = SessionId::new(goose_session.id.clone()); - - let agent = self - .create_agent_for_session(Some(cx), Some(&session_id), None) + self.session_manager + .get_session(&goose_session.id, false) .await .map_err(|e| { - sacp::Error::internal_error().data(format!("Failed to create agent: {}", e)) - })?; - let provider = self - .init_provider(&agent, &goose_session) - .await - .map_err(|e| { - sacp::Error::internal_error().data(format!("Failed to set provider: {}", e)) - })?; - Self::add_mcp_extensions(&agent, args.mcp_servers, &goose_session.id).await?; - - let session = GooseAcpSession { - agent, - messages: Conversation::new_unvalidated(Vec::new()), - tool_requests: HashMap::new(), - cancel_token: None, - }; - - let mut sessions = self.sessions.lock().await; - sessions.insert(goose_session.id.clone(), session); - - info!( - session_id = %goose_session.id, - session_type = "acp", - goose_mode = %self.goose_mode, - "Session started" - ); - - let model_state = build_model_state(&*provider).await?; - let mode_state = build_mode_state(self.goose_mode)?; - let provider_selection = session_provider_selection(&goose_session).to_string(); - let session_id_for_response = SessionId::new(goose_session.id); - let provider_options = build_provider_options(Some(provider.get_name())).await; - - Ok(NewSessionResponse::new(session_id_for_response) - .models(model_state.clone()) - .modes(mode_state.clone()) - .config_options(build_config_options( - &mode_state, - &model_state, - &provider_selection, - provider_options, - ))) - } - - async fn init_provider(&self, agent: &Agent, session: &Session) -> Result> { - let config = self.load_config()?; - let global_provider = config.get_goose_provider().ok(); - let provider_override = session - .provider_name - .as_deref() - .filter(|provider| *provider != DEFAULT_PROVIDER_ID); - let provider_name = provider_override - .map(ToOwned::to_owned) - .or_else(|| global_provider.clone()) - .ok_or_else(|| anyhow::anyhow!("Could not configure agent: missing provider"))?; - let explicitly_switched = - provider_override.is_some() && provider_override != global_provider.as_deref(); - let model_config = match &session.model_config { - Some(model_config) => model_config.clone(), - None if explicitly_switched => { - // The provider was set via _meta.provider (or similar) without an - // explicit model. Use the provider's own default model from the - // registry so we don't leak the global config model (which belongs - // to a different provider) into this one. - let entry = goose::providers::get_from_registry(&provider_name).await?; - let default_model = &entry.metadata().default_model; - goose::model::ModelConfig::new(default_model)?.with_canonical_limits(&provider_name) - } - None => { - let model_id = config.get_goose_model()?; - goose::model::ModelConfig::new(&model_id)?.with_canonical_limits(&provider_name) - } - }; - let extensions = - EnabledExtensionsState::extensions_or_default(Some(&session.extension_data), &config); - let provider = self - .create_provider(&provider_name, model_config, extensions) - .await?; - agent.update_provider(provider.clone(), &session.id).await?; - Ok(provider) + sacp::Error::internal_error().data(format!("Failed to reload session: {}", e)) + }) } async fn get_session_agent( &self, - session_id: &str, + thread_id: &str, cancel_token: Option, ) -> Result, sacp::Error> { - let mut sessions = self.sessions.lock().await; - let session = sessions.get_mut(session_id).ok_or_else(|| { - sacp::Error::resource_not_found(Some(session_id.to_string())) - .data(format!("Session not found: {}", session_id)) - })?; - if let Some(token) = cancel_token { - session.cancel_token = Some(token); - } - Ok(session.agent.clone()) + let mut rx = { + let mut sessions = self.sessions.lock().await; + let session = sessions.get_mut(thread_id).ok_or_else(|| { + sacp::Error::resource_not_found(Some(thread_id.to_string())) + .data(format!("Session not found: {}", thread_id)) + })?; + if let Some(token) = cancel_token { + session.cancel_token = Some(token); + } + match &session.agent { + AgentHandle::Ready(agent) => return Ok(agent.clone()), + AgentHandle::Loading(rx) => rx.clone(), + } + }; + // Drop the lock while we wait for the background setup to finish. + // spawn_agent_setup promotes the handle to Ready before signalling. + let agent = { + let guard = rx.wait_for(|v| v.is_some()).await.map_err(|_| { + sacp::Error::internal_error().data("Agent setup task was dropped".to_string()) + })?; + guard + .as_ref() + .unwrap() + .as_ref() + .map_err(|e| sacp::Error::internal_error().data(e.clone()))? + .clone() + }; + Ok(agent) } async fn add_mcp_extensions( agent: &Arc, mcp_servers: Vec, - session_id: &str, + internal_session_id: &str, ) -> Result<(), sacp::Error> { let mut configs = Vec::with_capacity(mcp_servers.len()); for mcp_server in mcp_servers { @@ -1234,7 +1448,7 @@ impl GooseAcpAgent { } let results = agent - .add_extensions_bulk(configs, session_id) + .add_extensions_bulk(configs, internal_session_id) .await .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; for result in &results { @@ -1256,64 +1470,55 @@ impl GooseAcpAgent { ) -> Result { debug!(?args, "load session request"); - let session_id = args.session_id.0.to_string(); + // The ACP session_id IS the thread ID. + let thread_id = args.session_id.0.to_string(); - let goose_session = self - .session_manager - .get_session(&session_id, true) + let thread = self + .thread_manager + .get_thread(&thread_id) .await .map_err(|_| { - sacp::Error::resource_not_found(Some(session_id.clone())) - .data(format!("Session not found: {}", session_id)) - })?; - let loaded_mode = goose_session.goose_mode; - let acp_session_id = SessionId::new(session_id.clone()); - - let agent = self - .create_agent_for_session(Some(cx), Some(&acp_session_id), Some(loaded_mode)) - .await - .map_err(|e| { - sacp::Error::internal_error().data(format!("Failed to create agent: {}", e)) - })?; - let provider = self - .init_provider(&agent, &goose_session) - .await - .map_err(|e| { - sacp::Error::internal_error().data(format!("Failed to set provider: {}", e)) + sacp::Error::resource_not_found(Some(thread_id.clone())) + .data(format!("Session not found: {}", thread_id)) })?; - agent - .update_goose_mode(loaded_mode, &session_id) - .await - .map_err(|e| { - sacp::Error::internal_error().data(format!("Failed to restore mode: {}", e)) - })?; - Self::add_mcp_extensions(&agent, args.mcp_servers, &session_id).await?; - - let provider_selection = session_provider_selection(&goose_session).to_string(); - let conversation = goose_session.conversation.ok_or_else(|| { + // Reuse the thread's current internal session so the agent retains + // conversation context (compaction state, full message history, etc.). + // The internal session is the source of truth for provider/mode. + let internal_session_id = thread.current_session_id.clone().ok_or_else(|| { sacp::Error::internal_error() - .data(format!("Session {} has no conversation data", session_id)) + .data(format!("Thread {} has no internal session", thread_id)) })?; - - self.session_manager - .update(&session_id) - .working_dir(args.cwd.clone()) - .apply() + let goose_session = self + .session_manager + .get_session(&internal_session_id, false) .await .map_err(|e| { sacp::Error::internal_error() - .data(format!("Failed to update session working directory: {}", e)) + .data(format!("Failed to load internal session: {}", e)) + })?; + let loaded_mode = goose_session.goose_mode; + + // ── REPLAY MESSAGES FIRST ── + // Stream the thread's human-visible message history back to the client + // immediately, before the slow agent/provider/extension setup. The + // replay only needs the thread_manager (SQLite reads) so the UI gets + // messages while the agent is still booting. + let thread_messages = self + .thread_manager + .list_messages(&thread_id) + .await + .map_err(|e| { + sacp::Error::internal_error().data(format!("Failed to load thread messages: {}", e)) })?; - let mut session = GooseAcpSession { - agent, - messages: conversation.clone(), - tool_requests: HashMap::new(), - cancel_token: None, - }; + // Lightweight tool_requests map for the replay loop — we only need it + // so that handle_tool_response can extract file locations from the + // matching request. No GooseAcpSession required. + let mut replay_tool_requests = + HashMap::::new(); - for message in conversation.messages() { + for message in &thread_messages { if !message.metadata.user_visible { continue; } @@ -1334,17 +1539,68 @@ impl GooseAcpAgent { ))?; } MessageContent::ToolRequest(tool_request) => { - self.handle_tool_request(tool_request, &args.session_id, &mut session, cx) - .await?; + // Replay-only: emit the ToolCall notification and + // stash the request for location extraction, but + // don't require a full GooseAcpSession. + replay_tool_requests.insert(tool_request.id.clone(), tool_request.clone()); + + let tool_name = match &tool_request.tool_call { + Ok(tool_call) => tool_call.name.to_string(), + Err(_) => "error".to_string(), + }; + + cx.send_notification(SessionNotification::new( + args.session_id.clone(), + SessionUpdate::ToolCall( + ToolCall::new( + ToolCallId::new(tool_request.id.clone()), + format_tool_name(&tool_name), + ) + .status(ToolCallStatus::Pending), + ), + ))?; } MessageContent::ToolResponse(tool_response) => { - self.handle_tool_response( - tool_response, - &args.session_id, - &mut session, - cx, - ) - .await?; + // Replay-only: emit the ToolCallUpdate notification, + // using the stashed replay_tool_requests for location + // extraction. + let status = match &tool_response.tool_result { + Ok(result) if result.is_error == Some(true) => ToolCallStatus::Failed, + Ok(_) => ToolCallStatus::Completed, + Err(_) => ToolCallStatus::Failed, + }; + + let mut fields = ToolCallUpdateFields::new().status(status); + if !tool_response + .tool_result + .as_ref() + .is_ok_and(|r| r.is_acp_aware()) + { + let content = build_tool_call_content(&tool_response.tool_result); + fields = fields.content(content); + + let locations = extract_locations_from_meta(tool_response) + .unwrap_or_else(|| { + if let Some(tool_request) = + replay_tool_requests.get(&tool_response.id) + { + extract_tool_locations(tool_request, tool_response) + } else { + Vec::new() + } + }); + if !locations.is_empty() { + fields = fields.locations(locations); + } + } + + cx.send_notification(SessionNotification::new( + args.session_id.clone(), + SessionUpdate::ToolCallUpdate(ToolCallUpdate::new( + ToolCallId::new(tool_response.id.clone()), + fields, + )), + ))?; } MessageContent::Thinking(thinking) => { cx.send_notification(SessionNotification::new( @@ -1359,31 +1615,66 @@ impl GooseAcpAgent { } } - let mut sessions = self.sessions.lock().await; - sessions.insert(session_id.clone(), session); + // ── Lightweight DB updates (fast) ── + self.session_manager + .update(&internal_session_id) + .working_dir(args.cwd.clone()) + .apply() + .await + .map_err(|e| { + sacp::Error::internal_error() + .data(format!("Failed to update session working directory: {}", e)) + })?; - let goose_mode = loaded_mode; + self.thread_manager + .update_working_dir(&thread_id, &args.cwd.display().to_string()) + .await + .map_err(|e| { + sacp::Error::internal_error() + .data(format!("Failed to update thread working directory: {}", e)) + })?; - info!( - session_id = %session_id, - session_type = "acp", - goose_mode = %goose_mode, - "Session loaded" + // ── Register the session immediately with a Loading handle ── + let (agent_tx, agent_rx) = + tokio::sync::watch::channel::, String>>>(None); + + let session = GooseAcpSession { + agent: AgentHandle::Loading(agent_rx), + internal_session_id: internal_session_id.clone(), + tool_requests: replay_tool_requests, + cancel_token: None, + pending_working_dir: None, + }; + self.sessions + .lock() + .await + .insert(thread_id.clone(), session); + + let mode_state = build_mode_state(loaded_mode)?; + + let resolved = resolve_provider_and_model(&self.config_dir, &goose_session).await; + let (model_state, config_options) = + build_eager_config(&resolved, &mode_state, &goose_session).await; + + self.spawn_agent_setup( + cx, + agent_tx, + AgentSetupRequest { + session_id: args.session_id.clone(), + goose_session, + mcp_servers: args.mcp_servers, + resolved_provider: None, + }, ); - let model_state = build_model_state(&*provider).await?; - let mode_state = build_mode_state(goose_mode)?; - let provider_options = build_provider_options(Some(provider.get_name())).await; - - Ok(LoadSessionResponse::new() - .models(model_state.clone()) - .modes(mode_state.clone()) - .config_options(build_config_options( - &mode_state, - &model_state, - &provider_selection, - provider_options, - ))) + let mut response = LoadSessionResponse::new().modes(mode_state); + if let Some(ms) = model_state { + response = response.models(ms); + } + if let Some(co) = config_options { + response = response.config_options(co); + } + Ok(response) } async fn on_prompt( @@ -1391,17 +1682,26 @@ impl GooseAcpAgent { cx: &ConnectionTo, args: PromptRequest, ) -> Result { - let session_id = args.session_id.0.to_string(); + // The ACP session_id IS the thread ID. + let thread_id = args.session_id.0.to_string(); let cancel_token = CancellationToken::new(); + let internal_session_id = self.internal_session_id(&thread_id).await?; let agent = self - .get_session_agent(&session_id, Some(cancel_token.clone())) + .get_session_agent(&thread_id, Some(cancel_token.clone())) .await?; let user_message = self.convert_acp_prompt_to_message(args.prompt); + self.thread_manager + .append_message(&thread_id, Some(&internal_session_id), &user_message) + .await + .map_err(|e| { + sacp::Error::internal_error().data(format!("Failed to persist message: {}", e)) + })?; + let session_config = SessionConfig { - id: session_id.clone(), + id: internal_session_id.clone(), schedule_id: None, max_turns: None, retry_config: None, @@ -1426,17 +1726,29 @@ impl GooseAcpAgent { match event { Ok(goose::agents::AgentEvent::Message(message)) => { + self.thread_manager + .append_message(&thread_id, Some(&internal_session_id), &message) + .await + .map_err(|e| { + sacp::Error::internal_error() + .data(format!("Failed to persist message: {}", e)) + })?; + let mut sessions = self.sessions.lock().await; - let session = sessions.get_mut(&session_id).ok_or_else(|| { + let session = sessions.get_mut(&thread_id).ok_or_else(|| { sacp::Error::invalid_params() - .data(format!("Session not found: {}", session_id)) + .data(format!("Session not found: {}", thread_id)) })?; - session.messages.push(message.clone()); - for content_item in &message.content { - self.handle_message_content(content_item, &args.session_id, session, cx) - .await?; + self.handle_message_content( + content_item, + &args.session_id, + &agent, + session, + cx, + ) + .await?; } } Ok(_) => {} @@ -1448,7 +1760,7 @@ impl GooseAcpAgent { } let mut sessions = self.sessions.lock().await; - if let Some(session) = sessions.get_mut(&session_id) { + if let Some(session) = sessions.get_mut(&thread_id) { session.cancel_token = None; } Ok(PromptResponse::new(if was_cancelled { @@ -1461,16 +1773,16 @@ impl GooseAcpAgent { async fn on_cancel(&self, args: CancelNotification) -> Result<(), sacp::Error> { debug!(?args, "cancel request"); - let session_id = args.session_id.0.to_string(); + let thread_id = args.session_id.0.to_string(); let mut sessions = self.sessions.lock().await; - if let Some(session) = sessions.get_mut(&session_id) { + if let Some(session) = sessions.get_mut(&thread_id) { if let Some(ref token) = session.cancel_token { - info!(session_id = %session_id, "prompt cancelled"); + info!(thread_id = %thread_id, "prompt cancelled"); token.cancel(); } } else { - warn!(session_id = %session_id, "cancel request for unknown session"); + warn!(thread_id = %thread_id, "cancel request for unknown session"); } Ok(()) @@ -1478,19 +1790,20 @@ impl GooseAcpAgent { async fn on_set_model( &self, - session_id: &str, + thread_id: &str, model_id: &str, ) -> Result { + let internal_id = self.internal_session_id(thread_id).await?; let config = self.load_config().map_err(|e| { sacp::Error::internal_error().data(format!("Failed to read config: {}", e)) })?; - let agent = self.get_session_agent(session_id, None).await?; + let agent = self.get_session_agent(thread_id, None).await?; let current_provider = agent.provider().await.map_err(|e| { sacp::Error::internal_error().data(format!("Failed to get provider: {}", e)) })?; let provider_name = current_provider.get_name().to_string(); let extensions = - EnabledExtensionsState::for_session(&self.session_manager, session_id, &config).await; + EnabledExtensionsState::for_session(&self.session_manager, &internal_id, &config).await; let model_config = goose::model::ModelConfig::new(model_id) .map_err(|e| { sacp::Error::invalid_params().data(format!("Invalid model config: {}", e)) @@ -1504,7 +1817,7 @@ impl GooseAcpAgent { })?; agent - .update_provider(provider, session_id) + .update_provider(provider, &internal_id) .await .map_err(|e| { sacp::Error::internal_error().data(format!("Failed to update provider: {}", e)) @@ -1512,26 +1825,56 @@ impl GooseAcpAgent { let mode = agent.goose_mode().await; agent - .update_goose_mode(mode, session_id) + .update_goose_mode(mode, &internal_id) .await .map_err(|e| { sacp::Error::internal_error().data(format!("Failed to propagate mode: {}", e)) })?; - info!(session_id = %session_id, model_id = %model_id, "Model switched"); + let model_id = model_id.to_string(); + self.update_thread_metadata(thread_id, move |meta| { + meta.model_name = Some(model_id); + }) + .await?; + Ok(SetSessionModelResponse::new()) } + async fn internal_session_id(&self, thread_id: &str) -> Result { + self.sessions + .lock() + .await + .get(thread_id) + .map(|s| s.internal_session_id.clone()) + .ok_or_else(|| { + sacp::Error::resource_not_found(Some(thread_id.to_string())) + .data(format!("Session not found: {}", thread_id)) + }) + } + + async fn update_thread_metadata( + &self, + thread_id: &str, + f: impl FnOnce(&mut goose::session::ThreadMetadata), + ) -> Result<(), sacp::Error> { + self.thread_manager + .update_metadata(thread_id, f) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + Ok(()) + } + async fn build_config_update( &self, - session_id: &SessionId, + thread_id: &SessionId, ) -> Result<(SessionNotification, Vec), sacp::Error> { + let internal_id = self.internal_session_id(&thread_id.0).await?; let session = self .session_manager - .get_session(&session_id.0, false) + .get_session(&internal_id, false) .await .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; - let agent = self.get_session_agent(&session_id.0, None).await?; + let agent = self.get_session_agent(&thread_id.0, None).await?; let provider = agent.provider().await.map_err(|e| { sacp::Error::internal_error().data(format!("Failed to get provider: {}", e)) })?; @@ -1546,7 +1889,7 @@ impl GooseAcpAgent { provider_options, ); let notification = SessionNotification::new( - session_id.clone(), + thread_id.clone(), SessionUpdate::ConfigOptionUpdate(ConfigOptionUpdate::new(config_options.clone())), ); Ok((notification, config_options)) @@ -1554,36 +1897,44 @@ impl GooseAcpAgent { async fn on_set_mode( &self, - session_id: &str, + thread_id: &str, mode_id: &str, ) -> Result { + let internal_id = self.internal_session_id(thread_id).await?; let mode = mode_id.parse::().map_err(|_| { sacp::Error::invalid_params().data(format!("Invalid mode: {}", mode_id)) })?; - let agent = self.get_session_agent(session_id, None).await?; + let agent = self.get_session_agent(thread_id, None).await?; agent - .update_goose_mode(mode, session_id) + .update_goose_mode(mode, &internal_id) .await .map_err(|e| { sacp::Error::internal_error().data(format!("Failed to update mode: {}", e)) })?; + let mode_id = mode_id.to_string(); + self.update_thread_metadata(thread_id, move |meta| { + meta.mode = Some(mode_id); + }) + .await?; + Ok(SetSessionModeResponse::new()) } async fn update_provider( &self, - session_id: &str, + thread_id: &str, provider_name: &str, model_name: Option<&str>, context_limit: Option, request_params: Option>, ) -> Result, sacp::Error> { + let internal_id = self.internal_session_id(thread_id).await?; let config = self.load_config().map_err(|e| { sacp::Error::internal_error().data(format!("Failed to read config: {}", e)) })?; - let agent = self.get_session_agent(session_id, None).await?; + let agent = self.get_session_agent(thread_id, None).await?; let current_provider = agent.provider().await.map_err(|e| { sacp::Error::internal_error().data(format!("Failed to get provider: {}", e)) })?; @@ -1626,7 +1977,7 @@ impl GooseAcpAgent { .with_context_limit(context_limit) .with_request_params(request_params); let extensions = - EnabledExtensionsState::for_session(&self.session_manager, session_id, &config).await; + EnabledExtensionsState::for_session(&self.session_manager, &internal_id, &config).await; let new_provider = self .create_provider(&resolved_provider_name, model_config, extensions) .await @@ -1635,7 +1986,7 @@ impl GooseAcpAgent { })?; agent - .update_provider(new_provider, session_id) + .update_provider(new_provider, &internal_id) .await .map_err(|e| { sacp::Error::internal_error().data(format!("Failed to update provider: {}", e)) @@ -1643,7 +1994,7 @@ impl GooseAcpAgent { let mode = agent.goose_mode().await; agent - .update_goose_mode(mode, session_id) + .update_goose_mode(mode, &internal_id) .await .map_err(|e| { sacp::Error::internal_error().data(format!("Failed to propagate mode: {}", e)) @@ -1653,10 +2004,16 @@ impl GooseAcpAgent { sacp::Error::internal_error().data(format!("Failed to get provider: {}", e)) })?; + let provider_name_owned = provider_name.to_string(); + self.update_thread_metadata(thread_id, move |meta| { + meta.provider_id = Some(provider_name_owned); + }) + .await?; + if use_default_provider { let update = self .session_manager - .update(session_id) + .update(&internal_id) .provider_name(DEFAULT_PROVIDER_ID); if has_default_overrides { let provider_model_config = provider.get_model_config(); @@ -1681,41 +2038,121 @@ impl GooseAcpAgent { } let (_, config_options) = self - .build_config_update(&SessionId::new(session_id.to_string())) + .build_config_update(&SessionId::new(thread_id.to_string())) .await?; Ok(config_options) } async fn on_list_sessions(&self) -> Result { - let sessions = self - .session_manager - .list_sessions_by_types(&[SessionType::Acp]) + // Return threads (= ACP sessions), not internal goose sessions. + let threads = self + .thread_manager + .list_threads(false) .await .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; - let session_infos: Vec = sessions + let session_infos: Vec = threads .into_iter() - .map(|s| { - SessionInfo::new(SessionId::new(s.id), s.working_dir) - .title(s.name) - .updated_at(s.updated_at.to_rfc3339()) + .map(|t| { + let cwd = t + .working_dir + .as_deref() + .map(std::path::PathBuf::from) + .unwrap_or_default(); + let mut meta = serde_json::Map::new(); + meta.insert( + "messageCount".to_string(), + serde_json::Value::Number(t.message_count.into()), + ); + SessionInfo::new(SessionId::new(t.id), cwd) + .title(t.name) + .updated_at(t.updated_at.to_rfc3339()) + .meta(meta) }) .collect(); Ok(ListSessionsResponse::new(session_infos)) } - async fn on_close_session( + async fn on_fork_session( &self, - session_id: &str, - ) -> Result { + cx: &ConnectionTo, + args: ForkSessionRequest, + ) -> Result { + let source_thread_id = &*args.session_id.0; + + // Fork the thread (copies metadata + messages). + let new_thread = self + .thread_manager + .fork_thread(source_thread_id) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + let new_thread_id = new_thread.id.clone(); + + // Create an internal session for the new thread. + let goose_session = self + .create_internal_session(&new_thread_id, args.cwd, None, None) + .await?; + + let internal_session_id = goose_session.id.clone(); + + let (agent_tx, agent_rx) = + tokio::sync::watch::channel::, String>>>(None); + + let session = GooseAcpSession { + agent: AgentHandle::Loading(agent_rx), + internal_session_id: internal_session_id.clone(), + tool_requests: HashMap::new(), + cancel_token: None, + pending_working_dir: None, + }; + self.sessions + .lock() + .await + .insert(new_thread_id.clone(), session); + + let mode_state = build_mode_state(self.goose_mode)?; + let resolved = resolve_provider_and_model(&self.config_dir, &goose_session).await; + let (model_state, config_options) = + build_eager_config(&resolved, &mode_state, &goose_session).await; + + self.spawn_agent_setup( + cx, + agent_tx, + AgentSetupRequest { + session_id: SessionId::new(new_thread_id.clone()), + goose_session, + mcp_servers: args.mcp_servers, + resolved_provider: resolved.ok(), + }, + ); + + let mut meta = serde_json::Map::new(); + meta.insert( + "messageCount".to_string(), + serde_json::Value::Number(new_thread.message_count.into()), + ); + + let mut response = ForkSessionResponse::new(SessionId::new(new_thread_id)) + .modes(mode_state) + .meta(meta); + if let Some(ms) = model_state { + response = response.models(ms); + } + if let Some(co) = config_options { + response = response.config_options(co); + } + Ok(response) + } + + async fn on_close_session(&self, thread_id: &str) -> Result { + // Tear down the in-memory agent. The thread persists for later session/load. let mut sessions = self.sessions.lock().await; - // Cancel before removing so on_prompt sees cancellation before session disappears. - if let Some(session) = sessions.get(session_id) { + if let Some(session) = sessions.get(thread_id) { if let Some(ref token) = session.cancel_token { token.cancel(); } } - sessions.remove(session_id); - info!(session_id = %session_id, "session closed"); + sessions.remove(thread_id); + info!(thread_id = %thread_id, "ACP session closed (thread preserved)"); Ok(CloseSessionResponse::new()) } } @@ -1727,11 +2164,12 @@ impl GooseAcpAgent { &self, req: AddExtensionRequest, ) -> Result { + let internal_id = self.internal_session_id(&req.session_id).await?; let config: ExtensionConfig = serde_json::from_value(req.config) .map_err(|e| sacp::Error::invalid_params().data(format!("bad config: {e}")))?; let agent = self.get_session_agent(&req.session_id, None).await?; agent - .add_extension(config, &req.session_id) + .add_extension(config, &internal_id) .await .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; Ok(EmptyResponse {}) @@ -1742,9 +2180,10 @@ impl GooseAcpAgent { &self, req: RemoveExtensionRequest, ) -> Result { + let internal_id = self.internal_session_id(&req.session_id).await?; let agent = self.get_session_agent(&req.session_id, None).await?; agent - .remove_extension(&req.name, &req.session_id) + .remove_extension(&req.name, &internal_id) .await .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; Ok(EmptyResponse {}) @@ -1752,8 +2191,9 @@ impl GooseAcpAgent { #[custom_method(GetToolsRequest)] async fn on_get_tools(&self, req: GetToolsRequest) -> Result { + let internal_id = self.internal_session_id(&req.session_id).await?; let agent = self.get_session_agent(&req.session_id, None).await?; - let tools = agent.list_tools(&req.session_id, None).await; + let tools = agent.list_tools(&internal_id, None).await; let tools_json = tools .into_iter() .map(|t| serde_json::to_value(&t)) @@ -1767,11 +2207,12 @@ impl GooseAcpAgent { &self, req: ReadResourceRequest, ) -> Result { + let internal_id = self.internal_session_id(&req.session_id).await?; let agent = self.get_session_agent(&req.session_id, None).await?; let cancel_token = CancellationToken::new(); let result = agent .extension_manager - .read_resource(&req.session_id, &req.uri, &req.extension_name, cancel_token) + .read_resource(&internal_id, &req.uri, &req.extension_name, cancel_token) .await .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; let result_json = serde_json::to_value(&result) @@ -1794,85 +2235,47 @@ impl GooseAcpAgent { if !path.exists() || !path.is_dir() { return Err(sacp::Error::invalid_params().data("invalid directory path")); } + let internal_id = self.internal_session_id(&req.session_id).await?; self.session_manager - .update(&req.session_id) + .update(&internal_id) .working_dir(path.clone()) .apply() .await .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; - // Notify MCP servers so roots stay in sync with the new working directory. - if let Some(session) = self.sessions.lock().await.get(&req.session_id) { - session - .agent - .extension_manager - .update_working_dir(&path) - .await; + self.thread_manager + .update_working_dir(&req.session_id, &working_dir) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + + if let Some(session) = self.sessions.lock().await.get_mut(&req.session_id) { + match &session.agent { + AgentHandle::Ready(agent) => { + agent.extension_manager.update_working_dir(&path).await; + } + AgentHandle::Loading(_) => { + session.pending_working_dir = Some(path); + } + } } Ok(EmptyResponse {}) } - #[custom_method(GetSessionRequest)] - async fn on_get_session( - &self, - req: GetSessionRequest, - ) -> Result { - let session = self - .session_manager - .get_session(&req.session_id, req.include_messages) - .await - .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; - let session_json = serde_json::to_value(&session) - .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; - Ok(GetSessionResponse { - session: session_json, - }) - } - #[custom_method(DeleteSessionRequest)] async fn on_delete_session( &self, req: DeleteSessionRequest, ) -> Result { - self.session_manager - .delete_session(&req.session_id) + // Delete the thread and all its internal sessions + messages. + self.thread_manager + .delete_thread(&req.session_id) .await .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; self.sessions.lock().await.remove(&req.session_id); Ok(EmptyResponse {}) } - #[custom_method(ExportSessionRequest)] - async fn on_export_session( - &self, - req: ExportSessionRequest, - ) -> Result { - let data = self - .session_manager - .export_session(&req.session_id) - .await - .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; - Ok(ExportSessionResponse { data }) - } - - #[custom_method(ImportSessionRequest)] - async fn on_import_session( - &self, - req: ImportSessionRequest, - ) -> Result { - let session = self - .session_manager - .import_session(&req.data, Some(SessionType::Acp)) - .await - .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; - let session_json = serde_json::to_value(&session) - .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; - Ok(ImportSessionResponse { - session: session_json, - }) - } - #[custom_method(GetExtensionsRequest)] async fn on_get_extensions(&self) -> Result { let extensions = goose::config::extensions::get_all_extensions(); @@ -2005,6 +2408,107 @@ impl GooseAcpAgent { .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; Ok(EmptyResponse {}) } + + #[custom_method(ExportSessionRequest)] + async fn on_export_session( + &self, + req: ExportSessionRequest, + ) -> Result { + let thread = self + .thread_manager + .get_thread(&req.session_id) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + let internal_id = thread + .current_session_id + .ok_or_else(|| sacp::Error::internal_error().data("Thread has no internal session"))?; + let data = self + .session_manager + .export_session(&internal_id) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + Ok(ExportSessionResponse { data }) + } + + #[custom_method(ImportSessionRequest)] + async fn on_import_session( + &self, + req: ImportSessionRequest, + ) -> Result { + let session = self + .session_manager + .import_session(&req.data, Some(SessionType::Acp)) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + + // Create a thread for the imported session. + let thread = self + .thread_manager + .create_thread( + Some(session.name.clone()), + None, + Some(session.working_dir.display().to_string()), + ) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + + // Link the internal session to the thread. + self.session_manager + .update(&session.id) + .thread_id(Some(thread.id.clone())) + .apply() + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + + // Copy conversation messages into thread_messages so they appear in the thread. + if let Some(ref conversation) = session.conversation { + for msg in conversation.messages() { + self.thread_manager + .append_message(&thread.id, Some(&session.id), msg) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + } + } + + // Re-fetch thread to get accurate message_count. + let thread = self + .thread_manager + .get_thread(&thread.id) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + + Ok(ImportSessionResponse { + session_id: thread.id, + title: Some(thread.name), + updated_at: Some(thread.updated_at.to_rfc3339()), + message_count: thread.message_count as u64, + }) + } + + #[custom_method(ArchiveSessionRequest)] + async fn on_archive_session( + &self, + req: ArchiveSessionRequest, + ) -> Result { + self.thread_manager + .archive_thread(&req.session_id) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + self.sessions.lock().await.remove(&req.session_id); + Ok(EmptyResponse {}) + } + + #[custom_method(UnarchiveSessionRequest)] + async fn on_unarchive_session( + &self, + req: UnarchiveSessionRequest, + ) -> Result { + self.thread_manager + .unarchive_thread(&req.session_id) + .await + .map_err(|e| sacp::Error::internal_error().data(e.to_string()))?; + Ok(EmptyResponse {}) + } } pub struct GooseAcpHandler { @@ -2047,7 +2551,20 @@ impl HandleDispatchFrom for GooseAcpHandler { .await .if_request( |req: LoadSessionRequest, responder: Responder| async { - responder.respond_with_result(agent.on_load_session(&cx, req).await) + let agent = agent.clone(); + let cx_clone = cx.clone(); + cx.spawn(async move { + match agent.on_load_session(&cx_clone, req).await { + Ok(response) => { + responder.respond(response)?; + } + Err(e) => { + responder.respond_with_error(e)?; + } + } + Ok(()) + })?; + Ok(()) }, ) .await @@ -2170,6 +2687,14 @@ impl HandleDispatchFrom for GooseAcpHandler { } }) .await + .if_request({ + let agent = agent.clone(); + let cx = cx.clone(); + |req: ForkSessionRequest, responder: Responder| async move { + responder.respond_with_result(agent.on_fork_session(&cx, req).await) + } + }) + .await .otherwise({ let agent = agent.clone(); |message: Dispatch| async move { diff --git a/crates/goose-acp/src/transport.rs b/crates/goose-acp/src/transport.rs index 130b08e922..dc6e8d0d43 100644 --- a/crates/goose-acp/src/transport.rs +++ b/crates/goose-acp/src/transport.rs @@ -26,7 +26,7 @@ pub(crate) const JSON_MIME_TYPE: &str = "application/json"; pub(crate) struct TransportSession { pub to_agent_tx: mpsc::Sender, - pub from_agent_rx: Arc>>, + pub from_agent_rx: Arc>>, pub handle: tokio::task::JoinHandle<()>, } diff --git a/crates/goose-acp/src/transport/http.rs b/crates/goose-acp/src/transport/http.rs index 84bf195a7e..608009e459 100644 --- a/crates/goose-acp/src/transport/http.rs +++ b/crates/goose-acp/src/transport/http.rs @@ -32,7 +32,7 @@ impl HttpState { async fn create_session(&self) -> Result { let (to_agent_tx, to_agent_rx) = mpsc::channel::(256); - let (from_agent_tx, from_agent_rx) = mpsc::channel::(256); + let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::(); let agent = self.server.create_agent().await.map_err(|e| { error!("Failed to create agent: {}", e); @@ -87,7 +87,7 @@ impl HttpState { async fn get_receiver( &self, acp_session_id: &str, - ) -> Result>>, StatusCode> { + ) -> Result>>, StatusCode> { let sessions = self.sessions.read().await; let session = sessions.get(acp_session_id).ok_or(StatusCode::NOT_FOUND)?; Ok(session.from_agent_rx.clone()) @@ -95,7 +95,7 @@ impl HttpState { } fn create_sse_stream( - receiver: Arc>>, + receiver: Arc>>, cleanup: Option<(Arc, String)>, ) -> Sse>> { let stream = async_stream::stream! { diff --git a/crates/goose-acp/src/transport/websocket.rs b/crates/goose-acp/src/transport/websocket.rs index 6643e17036..65a84737ad 100644 --- a/crates/goose-acp/src/transport/websocket.rs +++ b/crates/goose-acp/src/transport/websocket.rs @@ -30,7 +30,7 @@ impl WsState { async fn create_connection(&self) -> Result { let (to_agent_tx, to_agent_rx) = mpsc::channel::(256); - let (from_agent_tx, from_agent_rx) = mpsc::channel::(256); + let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::(); let agent = self.server.create_agent().await?; diff --git a/crates/goose-acp/tests/common_tests/mod.rs b/crates/goose-acp/tests/common_tests/mod.rs index e965a9260b..1edb3f4678 100644 --- a/crates/goose-acp/tests/common_tests/mod.rs +++ b/crates/goose-acp/tests/common_tests/mod.rs @@ -58,13 +58,19 @@ pub async fn run_list_sessions() { for s in &mut response.sessions { s.updated_at = None; } + let mut expected_meta = serde_json::Map::new(); + expected_meta.insert( + "messageCount".to_string(), + serde_json::Value::Number(2.into()), + ); assert_eq!( response, ListSessionsResponse::new(vec![SessionInfo::new( session.session_id().clone(), session.work_dir() ) - .title("ACP Session".to_string())]) + .title("New Chat".to_string()) + .meta(expected_meta)]) ); } diff --git a/crates/goose-acp/tests/custom_requests_test.rs b/crates/goose-acp/tests/custom_requests_test.rs index 29f983bec1..989636f8b9 100644 --- a/crates/goose-acp/tests/custom_requests_test.rs +++ b/crates/goose-acp/tests/custom_requests_test.rs @@ -3,14 +3,15 @@ mod common_tests; use common_tests::fixtures::server::AcpServerConnection; use common_tests::fixtures::{ - run_test, send_custom, Connection, Session, SessionData, TestConnectionConfig, + run_test, send_custom, Connection, PermissionDecision, Session, SessionData, + TestConnectionConfig, }; use goose::model::ModelConfig; use goose::providers::base::{MessageStream, Provider}; use goose::providers::errors::ProviderError; use goose_acp::server::AcpProviderFactory; -use goose_test_support::EnforceSessionId; -use std::sync::Arc; +use goose_test_support::{EnforceSessionId, IgnoreSessionId}; +use std::sync::{Arc, Mutex}; use common_tests::fixtures::OpenAiFixture; @@ -65,34 +66,6 @@ fn mock_provider_factory() -> AcpProviderFactory { }) } -#[test] -fn test_custom_session_get() { - run_test(async { - let openai = OpenAiFixture::new(vec![], Arc::new(EnforceSessionId::default())).await; - let mut conn = AcpServerConnection::new(TestConnectionConfig::default(), openai).await; - - let SessionData { session, .. } = conn.new_session().await.unwrap(); - let session_id = session.session_id().0.clone(); - - let result = send_custom( - conn.cx(), - "session/get", - serde_json::json!({ - "sessionId": session_id, - }), - ) - .await; - assert!(result.is_ok(), "expected ok, got: {:?}", result); - - let response = result.unwrap(); - let returned_session = response.get("session").expect("missing 'session' field"); - assert_eq!( - returned_session.get("id").and_then(|v| v.as_str()), - Some(session_id.as_ref()) - ); - }); -} - #[test] fn test_custom_get_tools() { run_test(async { @@ -237,27 +210,6 @@ fn test_provider_switching_updates_session_state() { .await .expect("provider config option should succeed"); - let response = send_custom( - conn.cx(), - "session/get", - serde_json::json!({ - "sessionId": session_id, - }), - ) - .await - .expect("session/get should succeed"); - let session_value = response.get("session").expect("missing session"); - assert_eq!( - session_value.get("provider_name"), - Some(&serde_json::json!("anthropic")) - ); - assert_eq!( - session_value - .get("model_config") - .and_then(|value| value.get("model_name")), - Some(&serde_json::json!("current")) - ); - let response = send_custom( conn.cx(), "_goose/session/provider/update", @@ -278,27 +230,6 @@ fn test_provider_switching_updates_session_state() { "expected refreshed config options" ); - let response = send_custom( - conn.cx(), - "session/get", - serde_json::json!({ - "sessionId": session_id, - }), - ) - .await - .expect("session/get after provider update should succeed"); - let session_value = response.get("session").expect("missing session"); - assert_eq!( - session_value.get("provider_name"), - Some(&serde_json::json!("openai")) - ); - assert_eq!( - session_value - .get("model_config") - .and_then(|value| value.get("model_name")), - Some(&serde_json::json!("o4-mini")) - ); - let response = send_custom( conn.cx(), "_goose/session/provider/update", @@ -319,25 +250,6 @@ fn test_provider_switching_updates_session_state() { .any(|option| option.get("id") == Some(&serde_json::json!("provider"))), "missing provider config option after reset" ); - - let response = send_custom( - conn.cx(), - "session/get", - serde_json::json!({ - "sessionId": session_id, - }), - ) - .await - .expect("session/get after provider reset should succeed"); - let session_value = response.get("session").expect("missing session"); - assert_eq!( - session_value.get("provider_name"), - Some(&serde_json::json!("goose")) - ); - assert_eq!( - session_value.get("model_config"), - Some(&serde_json::Value::Null) - ); }); } @@ -351,3 +263,54 @@ fn test_custom_unknown_method() { assert!(result.is_err(), "expected method_not_found error"); }); } + +#[test] +fn test_developer_fs_requests_use_acp_session_id() { + run_test(async { + let seen_session_id = Arc::new(Mutex::new(None::)); + let seen_session_id_clone = Arc::clone(&seen_session_id); + let openai = OpenAiFixture::new( + vec![ + ( + "Use the read tool to read /tmp/test_acp_read.txt and output only its contents." + .to_string(), + include_str!("test_data/openai_fs_read_tool_call.txt"), + ), + ( + r#""content":"test-read-content-12345""#.into(), + include_str!("test_data/openai_fs_read_tool_result.txt"), + ), + ], + Arc::new(IgnoreSessionId), + ) + .await; + let config = TestConnectionConfig { + read_text_file: Some(Arc::new(move |req| { + *seen_session_id_clone.lock().unwrap() = Some(req.session_id.0.to_string()); + Ok(sacp::schema::ReadTextFileResponse::new( + "test-read-content-12345", + )) + })), + ..Default::default() + }; + let mut conn = AcpServerConnection::new(config, openai).await; + + let SessionData { mut session, .. } = conn.new_session().await.unwrap(); + let acp_session_id = session.session_id().0.to_string(); + + let output = session + .prompt( + "Use the read tool to read /tmp/test_acp_read.txt and output only its contents.", + PermissionDecision::Cancel, + ) + .await + .expect("prompt should succeed"); + + assert_eq!(output.text, "test-read-content-12345"); + assert_eq!( + seen_session_id.lock().unwrap().as_deref(), + Some(acp_session_id.as_str()), + "ACP read request should use the ACP session/thread ID", + ); + }); +} diff --git a/crates/goose-acp/tests/fixtures/server.rs b/crates/goose-acp/tests/fixtures/server.rs index 20fb1c093b..f394a03d1c 100644 --- a/crates/goose-acp/tests/fixtures/server.rs +++ b/crates/goose-acp/tests/fixtures/server.rs @@ -4,7 +4,7 @@ use super::{ }; use async_trait::async_trait; use goose::config::PermissionManager; -use goose_test_support::{EnforceSessionId, ExpectedSessionId}; +use goose_test_support::{ExpectedSessionId, IgnoreSessionId}; use sacp::schema::{ ClientCapabilities, CloseSessionRequest, ContentBlock, CreateTerminalRequest, FileSystemCapabilities, ImageContent, InitializeRequest, KillTerminalRequest, @@ -99,7 +99,10 @@ impl Connection for AcpServerConnection { type Session = AcpServerSession; fn expected_session_id() -> Arc { - Arc::new(EnforceSessionId::default()) + // The ACP session ID returned to clients is now a thread ID, which is + // intentionally different from the internal session ID the agent sends + // to the LLM provider. Skip strict matching. + Arc::new(IgnoreSessionId) } async fn new(config: TestConnectionConfig, openai: super::OpenAiFixture) -> Self { diff --git a/crates/goose-sdk/src/custom_requests.rs b/crates/goose-sdk/src/custom_requests.rs index afa61149ed..c5b73469b6 100644 --- a/crates/goose-sdk/src/custom_requests.rs +++ b/crates/goose-sdk/src/custom_requests.rs @@ -83,24 +83,6 @@ pub struct UpdateWorkingDirRequest { pub working_dir: String, } -/// Get a session by ID. -#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcRequest)] -#[request(method = "session/get", response = GetSessionResponse)] -#[serde(rename_all = "camelCase")] -pub struct GetSessionRequest { - pub session_id: String, - #[serde(default)] - pub include_messages: bool, -} - -/// Get a session response. -#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcResponse)] -pub struct GetSessionResponse { - /// The session object with id, name, working_dir, timestamps, tokens, etc. - #[serde(default)] - pub session: serde_json::Value, -} - /// Delete a session. #[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcRequest)] #[request(method = "session/delete", response = EmptyResponse)] @@ -109,35 +91,6 @@ pub struct DeleteSessionRequest { pub session_id: String, } -/// Export a session as a JSON string. -#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcRequest)] -#[request(method = "_goose/session/export", response = ExportSessionResponse)] -#[serde(rename_all = "camelCase")] -pub struct ExportSessionRequest { - pub session_id: String, -} - -/// Export session response. -#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcResponse)] -pub struct ExportSessionResponse { - pub data: String, -} - -/// Import a session from a JSON string. -#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcRequest)] -#[request(method = "_goose/session/import", response = ImportSessionResponse)] -pub struct ImportSessionRequest { - pub data: String, -} - -/// Import session response. -#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcResponse)] -pub struct ImportSessionResponse { - /// The imported session object. - #[serde(default)] - pub session: serde_json::Value, -} - /// List configured extensions and any warnings. #[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcRequest)] #[request(method = "_goose/config/extensions", response = GetExtensionsResponse)] @@ -254,6 +207,53 @@ pub struct ListProvidersResponse { pub providers: Vec, } +/// Archive a session (soft delete). +#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcRequest)] +#[request(method = "_goose/session/archive", response = EmptyResponse)] +#[serde(rename_all = "camelCase")] +pub struct ArchiveSessionRequest { + pub session_id: String, +} + +/// Unarchive a previously archived session. +#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcRequest)] +#[request(method = "_goose/session/unarchive", response = EmptyResponse)] +#[serde(rename_all = "camelCase")] +pub struct UnarchiveSessionRequest { + pub session_id: String, +} + +/// Export a session as a JSON string. +#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcRequest)] +#[request(method = "_goose/session/export", response = ExportSessionResponse)] +#[serde(rename_all = "camelCase")] +pub struct ExportSessionRequest { + pub session_id: String, +} + +/// Export session response — raw JSON of the goose session with `conversation`. +#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcResponse)] +pub struct ExportSessionResponse { + pub data: String, +} + +/// Import a session from a JSON string. +#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcRequest)] +#[request(method = "_goose/session/import", response = ImportSessionResponse)] +pub struct ImportSessionRequest { + pub data: String, +} + +/// Import session response — metadata about the newly created session. +#[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcResponse)] +#[serde(rename_all = "camelCase")] +pub struct ImportSessionResponse { + pub session_id: String, + pub title: Option, + pub updated_at: Option, + pub message_count: u64, +} + /// Empty success response for operations that return no data. #[derive(Debug, Default, Clone, Serialize, Deserialize, JsonSchema, JsonRpcResponse)] pub struct EmptyResponse {} diff --git a/crates/goose/src/providers/canonical/mod.rs b/crates/goose/src/providers/canonical/mod.rs index 847257c756..07ea6bd9ba 100644 --- a/crates/goose/src/providers/canonical/mod.rs +++ b/crates/goose/src/providers/canonical/mod.rs @@ -3,7 +3,9 @@ mod name_builder; mod registry; pub use model::{CanonicalModel, Limit, Modalities, Modality, Pricing}; -pub use name_builder::{canonical_name, map_to_canonical_model, strip_version_suffix}; +pub use name_builder::{ + canonical_name, map_provider_name, map_to_canonical_model, strip_version_suffix, +}; pub use registry::CanonicalModelRegistry; #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] @@ -21,6 +23,47 @@ impl ModelMapping { } } +/// Return recommended model names for a provider using only the bundled canonical registry. +/// +/// This avoids network calls by looking up all known models for the provider, +/// filtering to text-input + tool-calling models, and sorting by release date. +/// The returned names are the canonical short names (e.g. "claude-3.5-sonnet"). +/// +/// TODO: This trades speed for correctness — the canonical registry may not perfectly +/// match what the provider API returns (new models not yet in the registry, deprecated +/// models still listed, or locally-installed models for providers like Ollama). Consider +/// whether to reconcile with a live API call in the background. +pub fn recommended_models_from_registry(provider: &str) -> Vec { + let registry = match CanonicalModelRegistry::bundled() { + Ok(r) => r, + Err(_) => return vec![], + }; + + let registry_provider = map_provider_name(provider); + let all = registry.get_all_models_for_provider(registry_provider); + + let mut models_with_dates: Vec<(String, Option)> = all + .iter() + .filter(|m| m.modalities.input.contains(&Modality::Text) && m.tool_call) + .filter_map(|m| { + let (_, name) = m.id.split_once('/')?; + Some((name.to_string(), m.release_date.clone())) + }) + .collect(); + + models_with_dates.sort_by(|a, b| match (&a.1, &b.1) { + (Some(date_a), Some(date_b)) => date_b.cmp(date_a), + (Some(_), None) => std::cmp::Ordering::Less, + (None, Some(_)) => std::cmp::Ordering::Greater, + (None, None) => a.0.cmp(&b.0), + }); + + models_with_dates + .into_iter() + .map(|(name, _)| name) + .collect() +} + pub fn maybe_get_canonical_model(provider: &str, model: &str) -> Option { let registry = CanonicalModelRegistry::bundled().ok()?; diff --git a/crates/goose/src/providers/canonical/name_builder.rs b/crates/goose/src/providers/canonical/name_builder.rs index f7fb000f56..7c6e8fa5ef 100644 --- a/crates/goose/src/providers/canonical/name_builder.rs +++ b/crates/goose/src/providers/canonical/name_builder.rs @@ -39,7 +39,7 @@ fn is_meta_provider(provider: &str) -> bool { matches!(provider, "databricks" | "tetrate" | "bedrock" | "azure") } -fn map_provider_name(provider: &str) -> &str { +pub fn map_provider_name(provider: &str) -> &str { match provider { // Goose provider names that differ from models.dev names "xai" => "x-ai", diff --git a/crates/goose/src/session/mod.rs b/crates/goose/src/session/mod.rs index 5e3534b661..167762cffc 100644 --- a/crates/goose/src/session/mod.rs +++ b/crates/goose/src/session/mod.rs @@ -3,6 +3,7 @@ mod diagnostics; pub mod extension_data; mod legacy; pub mod session_manager; +pub mod thread_manager; pub use diagnostics::{ config_path, generate_diagnostics, get_system_info, latest_llm_log_path, @@ -12,3 +13,4 @@ pub use extension_data::{EnabledExtensionsState, ExtensionData, ExtensionState, pub use session_manager::{ Session, SessionInsights, SessionManager, SessionType, SessionUpdateBuilder, }; +pub use thread_manager::{Thread, ThreadManager, ThreadMetadata}; diff --git a/crates/goose/src/session/session_manager.rs b/crates/goose/src/session/session_manager.rs index 3da0a0a764..bea5ec9717 100644 --- a/crates/goose/src/session/session_manager.rs +++ b/crates/goose/src/session/session_manager.rs @@ -19,7 +19,7 @@ use std::sync::{Arc, LazyLock}; use tracing::{info, warn}; use utoipa::ToSchema; -pub const CURRENT_SCHEMA_VERSION: i32 = 9; +pub const CURRENT_SCHEMA_VERSION: i32 = 10; pub const SESSIONS_FOLDER: &str = "sessions"; pub const DB_NAME: &str = "sessions.db"; @@ -81,6 +81,8 @@ pub struct Session { pub model_config: Option, #[serde(default)] pub goose_mode: GooseMode, + #[serde(default)] + pub thread_id: Option, } pub struct SessionUpdateBuilder<'a> { @@ -103,6 +105,7 @@ pub struct SessionUpdateBuilder<'a> { provider_name: Option>, model_config: Option>, goose_mode: Option, + thread_id: Option>, } #[derive(Serialize, ToSchema, Debug)] @@ -134,6 +137,7 @@ impl<'a> SessionUpdateBuilder<'a> { provider_name: None, model_config: None, goose_mode: None, + thread_id: None, } } @@ -241,6 +245,11 @@ impl<'a> SessionUpdateBuilder<'a> { self.goose_mode = Some(mode); self } + + pub fn thread_id(mut self, thread_id: Option) -> Self { + self.thread_id = Some(thread_id); + self + } } pub struct SessionManager { @@ -361,7 +370,22 @@ impl SessionManager { if user_message_count <= MSG_COUNT_FOR_SESSION_NAME_GENERATION { let name = provider.generate_session_name(id, &conversation).await?; - self.update(id).system_generated_name(name).apply().await + self.update(id) + .system_generated_name(name.clone()) + .apply() + .await?; + + // Also update the thread name so ACP clients see it via session/list. + if let Some(ref thread_id) = session.thread_id { + let thread_mgr = super::thread_manager::ThreadManager::new(self.storage.clone()); + let thread = thread_mgr.get_thread(thread_id).await?; + if !thread.user_set_name { + thread_mgr + .update_thread(thread_id, Some(name), Some(false), None) + .await?; + } + } + Ok(()) } else { Ok(()) } @@ -407,7 +431,7 @@ pub struct SessionStorage { session_dir: PathBuf, } -fn role_to_string(role: &Role) -> &'static str { +pub(crate) fn role_to_string(role: &Role) -> &'static str { match role { Role::User => "user", Role::Assistant => "assistant", @@ -439,6 +463,7 @@ impl Default for Session { provider_name: None, model_config: None, goose_mode: GooseMode::default(), + thread_id: None, } } } @@ -508,6 +533,7 @@ impl sqlx::FromRow<'_, sqlx::sqlite::SqliteRow> for Session { .ok() .and_then(|s| s.parse().ok()) .unwrap_or_default(), + thread_id: row.try_get("thread_id").ok().flatten(), }) } } @@ -537,7 +563,7 @@ impl SessionStorage { } } - async fn pool(&self) -> Result<&Pool> { + pub(crate) async fn pool(&self) -> Result<&Pool> { self.initialized .get_or_try_init(|| async { let schema_exists = sqlx::query_scalar::<_, bool>( @@ -607,7 +633,8 @@ impl SessionStorage { user_recipe_values_json TEXT, provider_name TEXT, model_config_json TEXT, - goose_mode TEXT NOT NULL DEFAULT 'auto' + goose_mode TEXT NOT NULL DEFAULT 'auto', + thread_id TEXT ) "#, ) @@ -647,6 +674,48 @@ impl SessionStorage { sqlx::query("CREATE INDEX idx_sessions_type ON sessions(session_type)") .execute(pool) .await?; + sqlx::query("CREATE INDEX IF NOT EXISTS idx_sessions_thread ON sessions(thread_id)") + .execute(pool) + .await?; + + sqlx::query( + "CREATE TABLE IF NOT EXISTS threads ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL DEFAULT 'New Chat', + user_set_name BOOLEAN DEFAULT FALSE, + working_dir TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + archived_at TIMESTAMP, + metadata_json TEXT DEFAULT '{}' + )", + ) + .execute(pool) + .await?; + + sqlx::query( + "CREATE TABLE IF NOT EXISTS thread_messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + thread_id TEXT NOT NULL REFERENCES threads(id), + session_id TEXT, + message_id TEXT, + role TEXT NOT NULL, + content_json TEXT NOT NULL, + created_timestamp INTEGER NOT NULL, + metadata_json TEXT DEFAULT '{}' + )", + ) + .execute(pool) + .await?; + + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_thread_messages_thread ON thread_messages(thread_id)", + ) + .execute(pool) + .await?; + sqlx::query("CREATE INDEX IF NOT EXISTS idx_thread_messages_message_id ON thread_messages(message_id)") + .execute(pool) + .await?; Ok(()) } @@ -938,6 +1007,59 @@ impl SessionStorage { .execute(&mut **tx) .await?; } + 10 => { + // Check if thread_id column already exists (e.g. fresh schema) + let has_thread_id = sqlx::query_scalar::<_, i32>( + "SELECT COUNT(*) FROM pragma_table_info('sessions') WHERE name = 'thread_id'", + ) + .fetch_one(&mut **tx) + .await? + > 0; + if !has_thread_id { + sqlx::query("ALTER TABLE sessions ADD COLUMN thread_id TEXT") + .execute(&mut **tx) + .await?; + } + sqlx::query( + "CREATE INDEX IF NOT EXISTS idx_sessions_thread ON sessions(thread_id)", + ) + .execute(&mut **tx) + .await?; + sqlx::query( + "CREATE TABLE IF NOT EXISTS threads ( + id TEXT PRIMARY KEY, + name TEXT NOT NULL DEFAULT 'New Chat', + user_set_name BOOLEAN DEFAULT FALSE, + working_dir TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + archived_at TIMESTAMP, + metadata_json TEXT DEFAULT '{}' + )", + ) + .execute(&mut **tx) + .await?; + sqlx::query( + "CREATE TABLE IF NOT EXISTS thread_messages ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + thread_id TEXT NOT NULL REFERENCES threads(id), + session_id TEXT, + message_id TEXT, + role TEXT NOT NULL, + content_json TEXT NOT NULL, + created_timestamp INTEGER NOT NULL, + metadata_json TEXT DEFAULT '{}' + )", + ) + .execute(&mut **tx) + .await?; + sqlx::query("CREATE INDEX IF NOT EXISTS idx_thread_messages_thread ON thread_messages(thread_id)") + .execute(&mut **tx) + .await?; + sqlx::query("CREATE INDEX IF NOT EXISTS idx_thread_messages_message_id ON thread_messages(message_id)") + .execute(&mut **tx) + .await?; + } _ => { anyhow::bail!("Unknown migration version: {}", version); } @@ -999,7 +1121,7 @@ impl SessionStorage { total_tokens, input_tokens, output_tokens, accumulated_total_tokens, accumulated_input_tokens, accumulated_output_tokens, schedule_id, recipe_json, user_recipe_values_json, - provider_name, model_config_json, goose_mode + provider_name, model_config_json, goose_mode, thread_id FROM sessions WHERE id = ? "#, @@ -1063,6 +1185,7 @@ impl SessionStorage { add_update!(builder.provider_name, "provider_name"); add_update!(builder.model_config, "model_config_json"); add_update!(builder.goose_mode, "goose_mode"); + add_update!(builder.thread_id, "thread_id"); if updates.is_empty() { return Ok(()); @@ -1131,6 +1254,9 @@ impl SessionStorage { if let Some(goose_mode) = builder.goose_mode { q = q.bind(goose_mode.to_string()); } + if let Some(thread_id) = builder.thread_id { + q = q.bind(thread_id); + } let pool = self.pool().await?; let mut tx = pool.begin_with("BEGIN IMMEDIATE").await?; @@ -1282,10 +1408,10 @@ impl SessionStorage { s.total_tokens, s.input_tokens, s.output_tokens, s.accumulated_total_tokens, s.accumulated_input_tokens, s.accumulated_output_tokens, s.schedule_id, s.recipe_json, s.user_recipe_values_json, - s.provider_name, s.model_config_json, s.goose_mode, + s.provider_name, s.model_config_json, s.goose_mode, s.thread_id, COUNT(m.id) as message_count FROM sessions s - INNER JOIN messages m ON s.id = m.session_id + LEFT JOIN messages m ON s.id = m.session_id {} GROUP BY s.id ORDER BY s.updated_at DESC diff --git a/crates/goose/src/session/thread_manager.rs b/crates/goose/src/session/thread_manager.rs new file mode 100644 index 0000000000..ff3662d584 --- /dev/null +++ b/crates/goose/src/session/thread_manager.rs @@ -0,0 +1,433 @@ +use super::session_manager::{role_to_string, SessionStorage}; +use crate::conversation::message::Message; +use anyhow::Result; +use chrono::{DateTime, Utc}; +use rmcp::model::Role; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Thread { + pub id: String, + pub name: String, + pub user_set_name: bool, + pub working_dir: Option, + pub created_at: DateTime, + pub updated_at: DateTime, + pub archived_at: Option>, + pub metadata: ThreadMetadata, + #[serde(default)] + pub current_session_id: Option, + #[serde(default)] + pub message_count: i64, +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct ThreadMetadata { + #[serde(default)] + pub persona_id: Option, + #[serde(default)] + pub project_id: Option, + #[serde(default)] + pub provider_id: Option, + #[serde(default)] + pub model_name: Option, + #[serde(default)] + pub mode: Option, + #[serde(flatten)] + pub extra: std::collections::HashMap, +} + +pub struct ThreadManager { + storage: Arc, +} + +const THREAD_SELECT: &str = "\ + SELECT t.id, t.name, t.user_set_name, t.working_dir, t.created_at, t.updated_at, \ + t.archived_at, t.metadata_json, \ + (SELECT s.id FROM sessions s WHERE s.thread_id = t.id ORDER BY s.created_at DESC LIMIT 1) as current_session_id, \ + (SELECT COUNT(*) FROM thread_messages WHERE thread_id = t.id) as message_count \ + FROM threads t"; + +type ThreadRow = ( + String, + String, + bool, + Option, + String, + String, + Option, + String, + Option, + i64, +); + +fn thread_from_row( + ( + id, + name, + user_set_name, + working_dir, + created_at, + updated_at, + archived_at_str, + metadata_json, + current_session_id, + message_count, + ): ThreadRow, +) -> Result { + let metadata: ThreadMetadata = serde_json::from_str(&metadata_json).unwrap_or_default(); + let archived_at = archived_at_str.as_deref().and_then(|s| s.parse().ok()); + Ok(Thread { + id, + name, + user_set_name, + working_dir, + created_at: created_at.parse().unwrap_or_else(|_| Utc::now()), + updated_at: updated_at.parse().unwrap_or_else(|_| Utc::now()), + archived_at, + metadata, + current_session_id, + message_count, + }) +} + +impl ThreadManager { + pub fn new(storage: Arc) -> Self { + Self { storage } + } + + pub async fn create_thread( + &self, + name: Option, + metadata: Option, + working_dir: Option, + ) -> Result { + let pool = self.storage.pool().await?; + let id = uuid::Uuid::new_v4().to_string(); + let name = name.unwrap_or_else(|| "New Chat".to_string()); + let meta = metadata.unwrap_or_default(); + let metadata_json = serde_json::to_string(&meta)?; + + sqlx::query( + "INSERT INTO threads (id, name, user_set_name, working_dir, metadata_json) VALUES (?, ?, FALSE, ?, ?)", + ) + .bind(&id) + .bind(&name) + .bind(&working_dir) + .bind(&metadata_json) + .execute(pool) + .await?; + + self.get_thread(&id).await + } + + pub async fn get_thread(&self, id: &str) -> Result { + let pool = self.storage.pool().await?; + let sql = format!("{} WHERE t.id = ?", THREAD_SELECT); + let row = sqlx::query_as::<_, ThreadRow>(&sql) + .bind(id) + .fetch_one(pool) + .await?; + + thread_from_row(row) + } + + pub async fn update_thread( + &self, + id: &str, + name: Option, + user_set_name: Option, + metadata: Option, + ) -> Result { + let pool = self.storage.pool().await?; + let mut sets = Vec::new(); + + if name.is_some() { + sets.push("name = ?"); + sets.push("user_set_name = ?"); + } + if metadata.is_some() { + sets.push("metadata_json = ?"); + } + + if !sets.is_empty() { + let sql = format!( + "UPDATE threads SET {}, updated_at = CURRENT_TIMESTAMP WHERE id = ?", + sets.join(", ") + ); + let mut q = sqlx::query(&sql); + if let Some(ref n) = name { + q = q.bind(n); + q = q.bind(user_set_name.unwrap_or(true)); + } + if let Some(ref meta) = metadata { + q = q.bind(serde_json::to_string(meta)?); + } + q = q.bind(id); + q.execute(pool).await?; + } + + self.get_thread(id).await + } + + pub async fn list_threads(&self, include_archived: bool) -> Result> { + let pool = self.storage.pool().await?; + let sql = if include_archived { + format!("{} ORDER BY t.updated_at DESC", THREAD_SELECT) + } else { + format!( + "{} WHERE t.archived_at IS NULL ORDER BY t.updated_at DESC", + THREAD_SELECT + ) + }; + let rows = sqlx::query_as::<_, ThreadRow>(&sql).fetch_all(pool).await?; + + rows.into_iter().map(thread_from_row).collect() + } + + pub async fn archive_thread(&self, id: &str) -> Result { + let pool = self.storage.pool().await?; + sqlx::query("UPDATE threads SET archived_at = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP WHERE id = ?") + .bind(id) + .execute(pool) + .await?; + self.get_thread(id).await + } + + pub async fn unarchive_thread(&self, id: &str) -> Result { + let pool = self.storage.pool().await?; + sqlx::query( + "UPDATE threads SET archived_at = NULL, updated_at = CURRENT_TIMESTAMP WHERE id = ?", + ) + .bind(id) + .execute(pool) + .await?; + self.get_thread(id).await + } + + pub async fn update_metadata( + &self, + id: &str, + f: impl FnOnce(&mut ThreadMetadata), + ) -> Result { + let thread = self.get_thread(id).await?; + let mut meta = thread.metadata; + f(&mut meta); + self.update_thread(id, None, None, Some(meta)).await + } + + pub async fn update_working_dir(&self, id: &str, working_dir: &str) -> Result<()> { + let pool = self.storage.pool().await?; + sqlx::query( + "UPDATE threads SET working_dir = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?", + ) + .bind(working_dir) + .bind(id) + .execute(pool) + .await?; + Ok(()) + } + + pub async fn delete_thread(&self, id: &str) -> Result<()> { + let pool = self.storage.pool().await?; + let mut tx = pool.begin().await?; + + sqlx::query("DELETE FROM thread_messages WHERE thread_id = ?") + .bind(id) + .execute(&mut *tx) + .await?; + sqlx::query( + "DELETE FROM messages WHERE session_id IN (SELECT id FROM sessions WHERE thread_id = ?)", + ) + .bind(id) + .execute(&mut *tx) + .await?; + sqlx::query("DELETE FROM sessions WHERE thread_id = ?") + .bind(id) + .execute(&mut *tx) + .await?; + sqlx::query("DELETE FROM threads WHERE id = ?") + .bind(id) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok(()) + } + + pub async fn append_message( + &self, + thread_id: &str, + session_id: Option<&str>, + message: &Message, + ) -> Result { + let pool = self.storage.pool().await?; + let role_str = role_to_string(&message.role); + let metadata_json = serde_json::to_string(&message.metadata)?; + + // When the incoming message is text-only, try to coalesce it with the + // last stored row if that row has the same role and is also text-only. + // This avoids storing one row per streaming token while keeping the UI + // streaming path unchanged (callers still forward every chunk). + if message.has_only_text_content() && !message.content.is_empty() { + let new_text = message.as_concat_text(); + + let maybe_last = sqlx::query_as::<_, (i64, String, String, String, String)>( + "SELECT id, message_id, role, content_json, metadata_json \ + FROM thread_messages \ + WHERE thread_id = ? \ + ORDER BY id DESC LIMIT 1", + ) + .bind(thread_id) + .fetch_optional(pool) + .await?; + + if let Some(( + row_id, + existing_msg_id, + last_role, + last_content_json, + last_metadata_json, + )) = maybe_last + { + if last_role == role_str + && last_metadata_json == metadata_json + && is_text_only_json(&last_content_json) + { + // Append text into the existing row's single text element. + let updated_json = append_text_json(&last_content_json, &new_text)?; + sqlx::query("UPDATE thread_messages SET content_json = ? WHERE id = ?") + .bind(&updated_json) + .bind(row_id) + .execute(pool) + .await?; + + sqlx::query("UPDATE threads SET updated_at = CURRENT_TIMESTAMP WHERE id = ?") + .bind(thread_id) + .execute(pool) + .await?; + + let mut stored = message.clone(); + stored.id = Some(existing_msg_id); + return Ok(stored); + } + } + } + + // Default path: insert a new row. + let content_json = serde_json::to_string(&message.content)?; + + let message_id = message + .id + .clone() + .unwrap_or_else(|| format!("tmsg_{}", uuid::Uuid::new_v4())); + + sqlx::query( + "INSERT INTO thread_messages (thread_id, session_id, message_id, role, content_json, created_timestamp, metadata_json) VALUES (?, ?, ?, ?, ?, ?, ?)", + ) + .bind(thread_id) + .bind(session_id) + .bind(&message_id) + .bind(role_str) + .bind(&content_json) + .bind(message.created) + .bind(&metadata_json) + .execute(pool) + .await?; + + sqlx::query("UPDATE threads SET updated_at = CURRENT_TIMESTAMP WHERE id = ?") + .bind(thread_id) + .execute(pool) + .await?; + + let mut stored = message.clone(); + stored.id = Some(message_id); + Ok(stored) + } + + pub async fn fork_thread(&self, source_thread_id: &str) -> Result { + let source = self.get_thread(source_thread_id).await?; + let pool = self.storage.pool().await?; + + let new_id = uuid::Uuid::new_v4().to_string(); + let name = format!("Fork of {}", source.name); + let metadata_json = serde_json::to_string(&source.metadata)?; + + sqlx::query( + "INSERT INTO threads (id, name, user_set_name, working_dir, metadata_json) VALUES (?, ?, FALSE, ?, ?)", + ) + .bind(&new_id) + .bind(&name) + .bind(&source.working_dir) + .bind(&metadata_json) + .execute(pool) + .await?; + + // Copy all thread messages + sqlx::query( + "INSERT INTO thread_messages (thread_id, session_id, message_id, role, content_json, created_timestamp, metadata_json) \ + SELECT ?, session_id, 'tmsg_' || hex(randomblob(16)), role, content_json, created_timestamp, metadata_json \ + FROM thread_messages WHERE thread_id = ? ORDER BY id ASC", + ) + .bind(&new_id) + .bind(source_thread_id) + .execute(pool) + .await?; + + self.get_thread(&new_id).await + } + + pub async fn list_messages(&self, thread_id: &str) -> Result> { + let pool = self.storage.pool().await?; + let rows = sqlx::query_as::<_, (Option, String, Option, String, i64, String)>( + "SELECT message_id, role, session_id, content_json, created_timestamp, metadata_json FROM thread_messages WHERE thread_id = ? ORDER BY id ASC", + ) + .bind(thread_id) + .fetch_all(pool) + .await?; + + let mut messages = Vec::new(); + for (message_id, role_str, _session_id, content_json, created_timestamp, metadata_json) in + rows + { + let role = match role_str.as_str() { + "user" => Role::User, + "assistant" => Role::Assistant, + _ => continue, + }; + let content = serde_json::from_str(&content_json)?; + let metadata = serde_json::from_str(&metadata_json).unwrap_or_default(); + + let mut msg = Message::new(role, created_timestamp, content); + msg.metadata = metadata; + if let Some(id) = message_id { + msg = msg.with_id(id); + } + messages.push(msg); + } + + Ok(messages) + } +} + +/// Check whether a `content_json` string represents a single text-only element. +/// Avoids a full deserialize by inspecting the JSON structure directly. +fn is_text_only_json(content_json: &str) -> bool { + let Ok(items) = serde_json::from_str::>(content_json) else { + return false; + }; + items.len() == 1 + && items[0].get("type").and_then(|v| v.as_str()) == Some("text") + && items[0].get("text").is_some() +} + +/// Append `new_text` to the single text element in a text-only `content_json` array. +fn append_text_json(content_json: &str, new_text: &str) -> anyhow::Result { + let mut items: Vec = serde_json::from_str(content_json)?; + if let Some(text_val) = items.get_mut(0).and_then(|v| v.get_mut("text")) { + let existing = text_val.as_str().unwrap_or(""); + *text_val = serde_json::Value::String(format!("{}{}", existing, new_text)); + } + Ok(serde_json::to_string(&items)?) +} diff --git a/crates/goose/tests/thread_message_coalescing_test.rs b/crates/goose/tests/thread_message_coalescing_test.rs new file mode 100644 index 0000000000..012ada5a09 --- /dev/null +++ b/crates/goose/tests/thread_message_coalescing_test.rs @@ -0,0 +1,197 @@ +use goose::conversation::message::Message; +use goose::session::session_manager::SessionStorage; +use goose::session::thread_manager::ThreadManager; +use rmcp::model::CallToolRequestParams; +use std::sync::Arc; +use tempfile::TempDir; + +async fn setup() -> (ThreadManager, TempDir) { + let tmp = TempDir::new().unwrap(); + let storage = SessionStorage::create(tmp.path()).await.unwrap(); + let tm = ThreadManager::new(Arc::new(storage)); + (tm, tmp) +} + +#[tokio::test] +async fn consecutive_text_chunks_are_coalesced() { + let (tm, _tmp) = setup().await; + let thread = tm.create_thread(None, None, None).await.unwrap(); + + // Simulate streaming: three consecutive assistant text chunks. + tm.append_message( + &thread.id, + Some("s1"), + &Message::assistant().with_text("Hello"), + ) + .await + .unwrap(); + tm.append_message( + &thread.id, + Some("s1"), + &Message::assistant().with_text(" world"), + ) + .await + .unwrap(); + tm.append_message(&thread.id, Some("s1"), &Message::assistant().with_text("!")) + .await + .unwrap(); + + let messages = tm.list_messages(&thread.id).await.unwrap(); + assert_eq!(messages.len(), 1, "should coalesce into a single row"); + assert_eq!(messages[0].as_concat_text(), "Hello world!"); +} + +#[tokio::test] +async fn role_change_prevents_coalescing() { + let (tm, _tmp) = setup().await; + let thread = tm.create_thread(None, None, None).await.unwrap(); + + tm.append_message(&thread.id, Some("s1"), &Message::user().with_text("Hi")) + .await + .unwrap(); + tm.append_message( + &thread.id, + Some("s1"), + &Message::assistant().with_text("Hey"), + ) + .await + .unwrap(); + + let messages = tm.list_messages(&thread.id).await.unwrap(); + assert_eq!(messages.len(), 2, "different roles should not coalesce"); + assert_eq!(messages[0].as_concat_text(), "Hi"); + assert_eq!(messages[1].as_concat_text(), "Hey"); +} + +#[tokio::test] +async fn non_text_content_breaks_coalescing() { + let (tm, _tmp) = setup().await; + let thread = tm.create_thread(None, None, None).await.unwrap(); + + // Text, then tool request, then more text — should be 3 rows. + tm.append_message( + &thread.id, + Some("s1"), + &Message::assistant().with_text("Let me check"), + ) + .await + .unwrap(); + + let tool_msg = Message::assistant().with_tool_request( + "call_1", + Ok(CallToolRequestParams::new("shell").with_arguments( + serde_json::json!({"command": "ls"}) + .as_object() + .unwrap() + .clone(), + )), + ); + tm.append_message(&thread.id, Some("s1"), &tool_msg) + .await + .unwrap(); + + tm.append_message( + &thread.id, + Some("s1"), + &Message::assistant().with_text("Done"), + ) + .await + .unwrap(); + + let messages = tm.list_messages(&thread.id).await.unwrap(); + assert_eq!(messages.len(), 3, "tool request should break coalescing"); + assert_eq!(messages[0].as_concat_text(), "Let me check"); + assert_eq!(messages[2].as_concat_text(), "Done"); +} + +#[tokio::test] +async fn text_after_tool_response_not_coalesced_with_tool() { + let (tm, _tmp) = setup().await; + let thread = tm.create_thread(None, None, None).await.unwrap(); + + // A tool request message (non-text) followed by text — should not coalesce. + let tool_msg = Message::assistant().with_tool_request( + "call_1", + Ok(CallToolRequestParams::new("shell").with_arguments( + serde_json::json!({"command": "ls"}) + .as_object() + .unwrap() + .clone(), + )), + ); + tm.append_message(&thread.id, Some("s1"), &tool_msg) + .await + .unwrap(); + + tm.append_message( + &thread.id, + Some("s1"), + &Message::assistant().with_text("Result"), + ) + .await + .unwrap(); + + let messages = tm.list_messages(&thread.id).await.unwrap(); + assert_eq!( + messages.len(), + 2, + "text should not coalesce with non-text predecessor" + ); +} + +#[tokio::test] +async fn empty_message_not_coalesced() { + let (tm, _tmp) = setup().await; + let thread = tm.create_thread(None, None, None).await.unwrap(); + + tm.append_message( + &thread.id, + Some("s1"), + &Message::assistant().with_text("Hello"), + ) + .await + .unwrap(); + + // An empty assistant message (no content items). + let empty = Message::assistant(); + tm.append_message(&thread.id, Some("s1"), &empty) + .await + .unwrap(); + + let messages = tm.list_messages(&thread.id).await.unwrap(); + // Empty message should be inserted as a new row (not coalesced). + assert_eq!(messages.len(), 2); + assert_eq!(messages[0].as_concat_text(), "Hello"); +} + +#[tokio::test] +async fn metadata_change_prevents_coalescing() { + let (tm, _tmp) = setup().await; + let thread = tm.create_thread(None, None, None).await.unwrap(); + + tm.append_message( + &thread.id, + Some("s1"), + &Message::assistant().with_text("Visible"), + ) + .await + .unwrap(); + tm.append_message( + &thread.id, + Some("s1"), + &Message::assistant().with_text(" hidden").agent_only(), + ) + .await + .unwrap(); + + let messages = tm.list_messages(&thread.id).await.unwrap(); + assert_eq!( + messages.len(), + 2, + "metadata boundary should break coalescing" + ); + assert!(messages[0].metadata.user_visible); + assert!(!messages[1].metadata.user_visible); + assert_eq!(messages[0].as_concat_text(), "Visible"); + assert_eq!(messages[1].as_concat_text(), " hidden"); +} diff --git a/ui/acp/src/generated/client.gen.ts b/ui/acp/src/generated/client.gen.ts index f89d853689..39314029b8 100644 --- a/ui/acp/src/generated/client.gen.ts +++ b/ui/acp/src/generated/client.gen.ts @@ -9,6 +9,7 @@ export interface ExtMethodProvider { import type { AddExtensionRequest, + ArchiveSessionRequest, CheckSecretRequest, CheckSecretResponse, DeleteSessionRequest, @@ -16,8 +17,6 @@ import type { ExportSessionResponse, GetExtensionsRequest, GetExtensionsResponse, - GetSessionRequest, - GetSessionResponse, GetToolsRequest, GetToolsResponse, ImportSessionRequest, @@ -31,6 +30,7 @@ import type { RemoveConfigRequest, RemoveExtensionRequest, RemoveSecretRequest, + UnarchiveSessionRequest, UpdateProviderRequest, UpdateProviderResponse, UpdateWorkingDirRequest, @@ -41,7 +41,6 @@ import { zCheckSecretResponse, zExportSessionResponse, zGetExtensionsResponse, - zGetSessionResponse, zGetToolsResponse, zImportSessionResponse, zListProvidersResponse, @@ -77,29 +76,10 @@ export class GooseExtClient { await this.conn.extMethod("_goose/working_dir/update", params); } - async sessionGet(params: GetSessionRequest): Promise { - const raw = await this.conn.extMethod("session/get", params); - return zGetSessionResponse.parse(raw) as GetSessionResponse; - } - async sessionDelete(params: DeleteSessionRequest): Promise { await this.conn.extMethod("session/delete", params); } - async GooseSessionExport( - params: ExportSessionRequest, - ): Promise { - const raw = await this.conn.extMethod("_goose/session/export", params); - return zExportSessionResponse.parse(raw) as ExportSessionResponse; - } - - async GooseSessionImport( - params: ImportSessionRequest, - ): Promise { - const raw = await this.conn.extMethod("_goose/session/import", params); - return zImportSessionResponse.parse(raw) as ImportSessionResponse; - } - async GooseConfigExtensions( params: GetExtensionsRequest, ): Promise { @@ -153,4 +133,26 @@ export class GooseExtClient { async GooseSecretRemove(params: RemoveSecretRequest): Promise { await this.conn.extMethod("_goose/secret/remove", params); } + + async GooseSessionExport( + params: ExportSessionRequest, + ): Promise { + const raw = await this.conn.extMethod("_goose/session/export", params); + return zExportSessionResponse.parse(raw) as ExportSessionResponse; + } + + async GooseSessionImport( + params: ImportSessionRequest, + ): Promise { + const raw = await this.conn.extMethod("_goose/session/import", params); + return zImportSessionResponse.parse(raw) as ImportSessionResponse; + } + + async GooseSessionArchive(params: ArchiveSessionRequest): Promise { + await this.conn.extMethod("_goose/session/archive", params); + } + + async GooseSessionUnarchive(params: UnarchiveSessionRequest): Promise { + await this.conn.extMethod("_goose/session/unarchive", params); + } } diff --git a/ui/acp/src/generated/index.ts b/ui/acp/src/generated/index.ts index e6c1f0d669..727f4c1d3f 100644 --- a/ui/acp/src/generated/index.ts +++ b/ui/acp/src/generated/index.ts @@ -1,6 +1,6 @@ // This file is auto-generated by @hey-api/openapi-ts -export type { AddExtensionRequest, CheckSecretRequest, CheckSecretResponse, DeleteSessionRequest, EmptyResponse, ExportSessionRequest, ExportSessionResponse, ExtRequest, ExtResponse, GetExtensionsRequest, GetExtensionsResponse, GetSessionRequest, GetSessionResponse, GetToolsRequest, GetToolsResponse, ImportSessionRequest, ImportSessionResponse, ListProvidersRequest, ListProvidersResponse, ProviderListEntry, ReadConfigRequest, ReadConfigResponse, ReadResourceRequest, ReadResourceResponse, RemoveConfigRequest, RemoveExtensionRequest, RemoveSecretRequest, UpdateProviderRequest, UpdateProviderResponse, UpdateWorkingDirRequest, UpsertConfigRequest, UpsertSecretRequest } from './types.gen.js'; +export type { AddExtensionRequest, ArchiveSessionRequest, CheckSecretRequest, CheckSecretResponse, DeleteSessionRequest, EmptyResponse, ExportSessionRequest, ExportSessionResponse, ExtRequest, ExtResponse, GetExtensionsRequest, GetExtensionsResponse, GetToolsRequest, GetToolsResponse, ImportSessionRequest, ImportSessionResponse, ListProvidersRequest, ListProvidersResponse, ProviderListEntry, ReadConfigRequest, ReadConfigResponse, ReadResourceRequest, ReadResourceResponse, RemoveConfigRequest, RemoveExtensionRequest, RemoveSecretRequest, UnarchiveSessionRequest, UpdateProviderRequest, UpdateProviderResponse, UpdateWorkingDirRequest, UpsertConfigRequest, UpsertSecretRequest } from './types.gen.js'; export const GOOSE_EXT_METHODS = [ { @@ -28,26 +28,11 @@ export const GOOSE_EXT_METHODS = [ requestType: "UpdateWorkingDirRequest", responseType: "EmptyResponse", }, - { - method: "session/get", - requestType: "GetSessionRequest", - responseType: "GetSessionResponse", - }, { method: "session/delete", requestType: "DeleteSessionRequest", responseType: "EmptyResponse", }, - { - method: "_goose/session/export", - requestType: "ExportSessionRequest", - responseType: "ExportSessionResponse", - }, - { - method: "_goose/session/import", - requestType: "ImportSessionRequest", - responseType: "ImportSessionResponse", - }, { method: "_goose/config/extensions", requestType: "GetExtensionsRequest", @@ -93,6 +78,26 @@ export const GOOSE_EXT_METHODS = [ requestType: "RemoveSecretRequest", responseType: "EmptyResponse", }, + { + method: "_goose/session/export", + requestType: "ExportSessionRequest", + responseType: "ExportSessionResponse", + }, + { + method: "_goose/session/import", + requestType: "ImportSessionRequest", + responseType: "ImportSessionResponse", + }, + { + method: "_goose/session/archive", + requestType: "ArchiveSessionRequest", + responseType: "EmptyResponse", + }, + { + method: "_goose/session/unarchive", + requestType: "UnarchiveSessionRequest", + responseType: "EmptyResponse", + }, ] as const; export type GooseExtMethod = (typeof GOOSE_EXT_METHODS)[number]; diff --git a/ui/acp/src/generated/types.gen.ts b/ui/acp/src/generated/types.gen.ts index 4d00fa20fe..a5e8b237a7 100644 --- a/ui/acp/src/generated/types.gen.ts +++ b/ui/acp/src/generated/types.gen.ts @@ -71,24 +71,6 @@ export type UpdateWorkingDirRequest = { workingDir: string; }; -/** - * Get a session by ID. - */ -export type GetSessionRequest = { - sessionId: string; - includeMessages?: boolean; -}; - -/** - * Get a session response. - */ -export type GetSessionResponse = { - /** - * The session object with id, name, working_dir, timestamps, tokens, etc. - */ - session?: unknown; -}; - /** * Delete a session. */ @@ -96,37 +78,6 @@ export type DeleteSessionRequest = { sessionId: string; }; -/** - * Export a session as a JSON string. - */ -export type ExportSessionRequest = { - sessionId: string; -}; - -/** - * Export session response. - */ -export type ExportSessionResponse = { - data: string; -}; - -/** - * Import a session from a JSON string. - */ -export type ImportSessionRequest = { - data: string; -}; - -/** - * Import session response. - */ -export type ImportSessionResponse = { - /** - * The imported session object. - */ - session?: unknown; -}; - /** * List configured extensions and any warnings. */ @@ -245,17 +196,62 @@ export type RemoveSecretRequest = { key: string; }; +/** + * Export a session as a JSON string. + */ +export type ExportSessionRequest = { + sessionId: string; +}; + +/** + * Export session response — raw JSON of the goose session with `conversation`. + */ +export type ExportSessionResponse = { + data: string; +}; + +/** + * Import a session from a JSON string. + */ +export type ImportSessionRequest = { + data: string; +}; + +/** + * Import session response — metadata about the newly created session. + */ +export type ImportSessionResponse = { + sessionId: string; + title?: string | null; + updatedAt?: string | null; + messageCount: number; +}; + +/** + * Archive a session (soft delete). + */ +export type ArchiveSessionRequest = { + sessionId: string; +}; + +/** + * Unarchive a previously archived session. + */ +export type UnarchiveSessionRequest = { + sessionId: string; +}; + export type ExtRequest = { id: string; method: string; - params?: AddExtensionRequest | RemoveExtensionRequest | GetToolsRequest | ReadResourceRequest | UpdateWorkingDirRequest | GetSessionRequest | DeleteSessionRequest | ExportSessionRequest | ImportSessionRequest | GetExtensionsRequest | UpdateProviderRequest | ListProvidersRequest | ReadConfigRequest | UpsertConfigRequest | RemoveConfigRequest | CheckSecretRequest | UpsertSecretRequest | RemoveSecretRequest | { + params?: AddExtensionRequest | RemoveExtensionRequest | GetToolsRequest | ReadResourceRequest | UpdateWorkingDirRequest | DeleteSessionRequest | GetExtensionsRequest | UpdateProviderRequest | ListProvidersRequest | ReadConfigRequest | UpsertConfigRequest | RemoveConfigRequest | CheckSecretRequest | UpsertSecretRequest | RemoveSecretRequest | ExportSessionRequest | ImportSessionRequest | ArchiveSessionRequest | UnarchiveSessionRequest | { [key: string]: unknown; } | null; }; export type ExtResponse = { id: string; - result?: EmptyResponse | GetToolsResponse | ReadResourceResponse | GetSessionResponse | ExportSessionResponse | ImportSessionResponse | GetExtensionsResponse | UpdateProviderResponse | ListProvidersResponse | ReadConfigResponse | CheckSecretResponse | unknown; + result?: EmptyResponse | GetToolsResponse | ReadResourceResponse | GetExtensionsResponse | UpdateProviderResponse | ListProvidersResponse | ReadConfigResponse | CheckSecretResponse | ExportSessionResponse | ImportSessionResponse | unknown; } | { error: { code: number; diff --git a/ui/acp/src/generated/zod.gen.ts b/ui/acp/src/generated/zod.gen.ts index e1f3768c05..f5927e5b3b 100644 --- a/ui/acp/src/generated/zod.gen.ts +++ b/ui/acp/src/generated/zod.gen.ts @@ -61,21 +61,6 @@ export const zUpdateWorkingDirRequest = z.object({ workingDir: z.string() }); -/** - * Get a session by ID. - */ -export const zGetSessionRequest = z.object({ - sessionId: z.string(), - includeMessages: z.boolean().optional().default(false) -}); - -/** - * Get a session response. - */ -export const zGetSessionResponse = z.object({ - session: z.unknown().optional().default(null) -}); - /** * Delete a session. */ @@ -83,34 +68,6 @@ export const zDeleteSessionRequest = z.object({ sessionId: z.string() }); -/** - * Export a session as a JSON string. - */ -export const zExportSessionRequest = z.object({ - sessionId: z.string() -}); - -/** - * Export session response. - */ -export const zExportSessionResponse = z.object({ - data: z.string() -}); - -/** - * Import a session from a JSON string. - */ -export const zImportSessionRequest = z.object({ - data: z.string() -}); - -/** - * Import session response. - */ -export const zImportSessionResponse = z.object({ - session: z.unknown().optional().default(null) -}); - /** * List configured extensions and any warnings. */ @@ -226,6 +183,57 @@ export const zRemoveSecretRequest = z.object({ key: z.string() }); +/** + * Export a session as a JSON string. + */ +export const zExportSessionRequest = z.object({ + sessionId: z.string() +}); + +/** + * Export session response — raw JSON of the goose session with `conversation`. + */ +export const zExportSessionResponse = z.object({ + data: z.string() +}); + +/** + * Import a session from a JSON string. + */ +export const zImportSessionRequest = z.object({ + data: z.string() +}); + +/** + * Import session response — metadata about the newly created session. + */ +export const zImportSessionResponse = z.object({ + sessionId: z.string(), + title: z.union([ + z.string(), + z.null() + ]).optional(), + updatedAt: z.union([ + z.string(), + z.null() + ]).optional(), + messageCount: z.coerce.bigint().gte(BigInt(0)).max(BigInt('18446744073709551615'), { message: 'Invalid value: Expected uint64 to be <= 18446744073709551615' }) +}); + +/** + * Archive a session (soft delete). + */ +export const zArchiveSessionRequest = z.object({ + sessionId: z.string() +}); + +/** + * Unarchive a previously archived session. + */ +export const zUnarchiveSessionRequest = z.object({ + sessionId: z.string() +}); + export const zExtRequest = z.object({ id: z.string(), method: z.string(), @@ -236,10 +244,7 @@ export const zExtRequest = z.object({ zGetToolsRequest, zReadResourceRequest, zUpdateWorkingDirRequest, - zGetSessionRequest, zDeleteSessionRequest, - zExportSessionRequest, - zImportSessionRequest, zGetExtensionsRequest, zUpdateProviderRequest, zListProvidersRequest, @@ -248,7 +253,11 @@ export const zExtRequest = z.object({ zRemoveConfigRequest, zCheckSecretRequest, zUpsertSecretRequest, - zRemoveSecretRequest + zRemoveSecretRequest, + zExportSessionRequest, + zImportSessionRequest, + zArchiveSessionRequest, + zUnarchiveSessionRequest ]), z.union([ z.record(z.unknown()), @@ -265,14 +274,13 @@ export const zExtResponse = z.union([ zEmptyResponse, zGetToolsResponse, zReadResourceResponse, - zGetSessionResponse, - zExportSessionResponse, - zImportSessionResponse, zGetExtensionsResponse, zUpdateProviderResponse, zListProvidersResponse, zReadConfigResponse, - zCheckSecretResponse + zCheckSecretResponse, + zExportSessionResponse, + zImportSessionResponse ]), z.unknown() ]).optional() diff --git a/ui/desktop/openapi.json b/ui/desktop/openapi.json index 26cb358e78..79d88ee37d 100644 --- a/ui/desktop/openapi.json +++ b/ui/desktop/openapi.json @@ -7748,6 +7748,10 @@ "session_type": { "$ref": "#/components/schemas/SessionType" }, + "thread_id": { + "type": "string", + "nullable": true + }, "total_tokens": { "type": "integer", "format": "int32", diff --git a/ui/desktop/src/api/types.gen.ts b/ui/desktop/src/api/types.gen.ts index 902e3e860a..7eb3654612 100644 --- a/ui/desktop/src/api/types.gen.ts +++ b/ui/desktop/src/api/types.gen.ts @@ -1232,6 +1232,7 @@ export type Session = { recipe?: Recipe | null; schedule_id?: string | null; session_type?: SessionType; + thread_id?: string | null; total_tokens?: number | null; updated_at: string; user_recipe_values?: {