diff --git a/app/src/ai/acp/mod.rs b/app/src/ai/acp/mod.rs index 9327524c..597640da 100644 --- a/app/src/ai/acp/mod.rs +++ b/app/src/ai/acp/mod.rs @@ -7,7 +7,6 @@ mod launch; mod permissions; mod prompt; -mod response_translator; mod runtime_model; mod transport; diff --git a/app/src/ai/acp/response_translator.rs b/app/src/ai/acp/response_translator.rs deleted file mode 100644 index d6aaf5ce..00000000 --- a/app/src/ai/acp/response_translator.rs +++ /dev/null @@ -1,357 +0,0 @@ -use std::collections::HashMap; - -use galaxy_agent_core::{AgentEvent, RuntimeActivity, RuntimeActivityStatus, StopReason}; -use uuid::Uuid; -use warp_multi_agent_api::response_event::stream_finished; -use warp_multi_agent_api::{self as api, ClientAction, ResponseEvent}; - -use crate::ai::agent::runtime_activity; -use crate::ai::bedrock::response_translator::{ - build_add_agent_output_message, build_append_text, build_create_task, build_stream_init, - build_user_query_message, -}; - -/// Stateful translation from ACP session updates to Galaxy's existing agent UI -/// response protocol. -pub(super) struct AcpResponseTranslator { - task_id: String, - request_id: String, - needs_create_task: bool, - user_query: Option, - model_id: String, - initialized: bool, - message_id: Option, - activity_message_ids: HashMap, - activities: HashMap, - has_visible_output: bool, - used_tokens: u64, - context_size: u64, -} - -impl AcpResponseTranslator { - pub(super) fn new( - task_id: String, - needs_create_task: bool, - user_query: Option, - model_id: String, - ) -> Self { - Self { - task_id, - request_id: Uuid::new_v4().to_string(), - needs_create_task, - user_query, - model_id, - initialized: false, - message_id: None, - activity_message_ids: HashMap::new(), - activities: HashMap::new(), - has_visible_output: false, - used_tokens: 0, - context_size: 0, - } - } - - pub(super) fn translate(&mut self, event: AgentEvent) -> Result, String> { - let mut events = Vec::new(); - match event { - AgentEvent::TurnStarted { .. } => self.initialize(&mut events), - AgentEvent::TextDelta { text } => { - self.initialize(&mut events); - self.add_or_append(&text, &mut events); - } - // The legacy transcript has no dedicated reasoning surface on this - // path yet. The shared runtime event remains available for the UI - // convergence phase instead of being flattened into answer text. - AgentEvent::ReasoningDelta { .. } | AgentEvent::ReasoningCompleted { .. } => {} - AgentEvent::RuntimeActivityUpdated { activity } => { - self.initialize(&mut events); - self.message_id = None; - self.upsert_runtime_activity(activity, &mut events)?; - } - AgentEvent::ContextUsageUpdated { - used_tokens, - context_size, - } => { - self.used_tokens = used_tokens; - self.context_size = context_size; - } - AgentEvent::UserInputAccepted { text } => { - self.initialize(&mut events); - events.push(build_user_query_message(&self.task_id, &text)); - self.message_id = None; - } - AgentEvent::RuntimeNotice { message } => { - self.initialize(&mut events); - self.message_id = None; - self.add_or_append(&message, &mut events); - self.message_id = None; - } - AgentEvent::TurnStopped { reason } => { - self.initialize(&mut events); - if !self.has_visible_output && reason != StopReason::Cancelled { - self.add_or_append( - "> ACP agent completed without a text response.", - &mut events, - ); - } - events.push(self.finished(reason)); - } - AgentEvent::Tool { .. } => { - return Err( - "ACP runtime attempted to hand agent-owned tool execution to Galaxy".to_owned(), - ); - } - AgentEvent::UsageUpdated { .. } => { - return Err("ACP runtime reported provider-style request usage".to_owned()); - } - } - Ok(events) - } - - pub(super) fn startup_error(&mut self, error: &str) -> Vec { - let mut events = Vec::new(); - self.initialize(&mut events); - self.message_id = None; - self.add_or_append( - &format!("Galaxy couldn't start the ACP agent: {error}"), - &mut events, - ); - events.push(self.finished(StopReason::Refusal)); - events - } - - fn initialize(&mut self, events: &mut Vec) { - if self.initialized { - return; - } - // The ACP session ID is persisted separately. An empty conversation ID - // keeps this synthetic Init event out of Galaxy cloud token paths. - events.push(build_stream_init(&self.request_id, "")); - if self.needs_create_task { - events.push(build_create_task(&self.task_id)); - } - if let Some(user_query) = &self.user_query { - events.push(build_user_query_message(&self.task_id, user_query)); - } - self.initialized = true; - } - - fn add_or_append(&mut self, text: &str, events: &mut Vec) { - if text.is_empty() { - return; - } - self.has_visible_output = true; - if let Some(message_id) = &self.message_id { - events.push(build_append_text(&self.task_id, message_id, text)); - } else { - let message_id = Uuid::new_v4().to_string(); - events.push(build_add_agent_output_message( - &self.task_id, - &message_id, - text, - )); - self.message_id = Some(message_id); - } - } - - fn upsert_runtime_activity( - &mut self, - activity: RuntimeActivity, - events: &mut Vec, - ) -> Result<(), String> { - let activity_id = activity.id.clone(); - let merged_activity = self - .activities - .entry(activity_id.clone()) - .or_insert_with(|| activity.clone()); - if !activity.title.trim().is_empty() { - merged_activity.title = activity.title; - } - if activity.status.is_some() { - merged_activity.status = activity.status; - } - if activity.output.is_some() { - merged_activity.output = activity.output; - } - - let server_message_data = runtime_activity::encode(merged_activity) - .map_err(|error| format!("failed to encode ACP runtime activity: {error}"))?; - let fallback_text = runtime_activity_fallback_text(merged_activity); - if let Some(message_id) = self.activity_message_ids.get(&activity_id) { - events.push(build_update_runtime_activity_message( - &self.task_id, - message_id, - &fallback_text, - &server_message_data, - )); - } else { - let message_id = Uuid::new_v4().to_string(); - events.push(build_add_runtime_activity_message( - &self.task_id, - &message_id, - &fallback_text, - &server_message_data, - )); - self.activity_message_ids.insert(activity_id, message_id); - } - self.has_visible_output = true; - Ok(()) - } - - fn finished(&self, stop_reason: StopReason) -> ResponseEvent { - let reason = match stop_reason { - StopReason::Completed | StopReason::Cancelled => { - stream_finished::Reason::Done(stream_finished::Done {}) - } - StopReason::MaxTokens => { - stream_finished::Reason::MaxTokenLimit(stream_finished::ReachedMaxTokenLimit {}) - } - StopReason::ContextWindowExceeded => stream_finished::Reason::ContextWindowExceeded( - stream_finished::ContextWindowExceeded {}, - ), - StopReason::Refusal | StopReason::ToolLoopLimit | StopReason::Other(_) => { - stream_finished::Reason::Other(stream_finished::Other {}) - } - }; - let used_tokens = u32::try_from(self.used_tokens).unwrap_or(u32::MAX); - let context_usage = if self.context_size == 0 { - 0.0 - } else { - (self.used_tokens as f32 / self.context_size as f32).clamp(0.0, 1.0) - }; - #[allow(deprecated)] - let usage_metadata = stream_finished::ConversationUsageMetadata { - context_window_usage: context_usage, - summarized: false, - credits_spent: 0.0, - platform_credits_spent: 0.0, - total_input_tokens: used_tokens, - token_usage: Vec::new(), - tool_usage_metadata: None, - warp_token_usage: HashMap::new(), - byok_token_usage: HashMap::new(), - custom_endpoint_token_usage: HashMap::new(), - context_window_segments: Vec::new(), - }; - ResponseEvent { - r#type: Some(api::response_event::Type::Finished( - api::response_event::StreamFinished { - reason: Some(reason), - token_usage: vec![stream_finished::TokenUsage { - model_id: self.model_id.clone(), - // ACP reports current context occupancy, not the input - // consumed by this individual request. Galaxy separately - // accumulates per-request token usage, so counting it - // here would grow the total again on every turn. - total_input: 0, - output: 0, - input_cache_read: 0, - input_cache_write: 0, - cost_in_cents: 0.0, - }], - should_refresh_model_config: false, - request_cost: None, - conversation_usage_metadata: Some(usage_metadata), - }, - )), - } - } -} - -fn runtime_activity_fallback_text(activity: &RuntimeActivity) -> String { - let title = &activity.title; - let status = activity.status.as_ref().map(|status| match status { - RuntimeActivityStatus::Pending => "waiting", - RuntimeActivityStatus::InProgress => "running", - RuntimeActivityStatus::Completed => "completed", - RuntimeActivityStatus::Failed => "failed", - RuntimeActivityStatus::Other(_) => "updated", - }); - let mut text = match status { - Some(status) => format!("> **{title}** — {status}"), - None => format!("> **{title}**"), - }; - if let Some(output) = &activity.output { - text.push_str("\n\n"); - for line in output.lines() { - text.push_str(" "); - text.push_str(line); - text.push('\n'); - } - } - text -} - -fn build_add_runtime_activity_message( - task_id: &str, - message_id: &str, - fallback_text: &str, - server_message_data: &str, -) -> ResponseEvent { - let message = runtime_activity_message(task_id, message_id, fallback_text, server_message_data); - runtime_activity_client_action(api::client_action::Action::AddMessagesToTask( - api::client_action::AddMessagesToTask { - task_id: task_id.to_owned(), - messages: vec![message], - }, - )) -} - -fn build_update_runtime_activity_message( - task_id: &str, - message_id: &str, - fallback_text: &str, - server_message_data: &str, -) -> ResponseEvent { - let message = runtime_activity_message(task_id, message_id, fallback_text, server_message_data); - runtime_activity_client_action(api::client_action::Action::UpdateTaskMessage( - api::client_action::UpdateTaskMessage { - task_id: task_id.to_owned(), - message: Some(message), - mask: Some(prost_types::FieldMask { - paths: vec![ - "agent_output.text".to_owned(), - "server_message_data".to_owned(), - ], - }), - }, - )) -} - -fn runtime_activity_message( - task_id: &str, - message_id: &str, - fallback_text: &str, - server_message_data: &str, -) -> api::Message { - api::Message { - id: message_id.to_owned(), - task_id: task_id.to_owned(), - request_id: String::new(), - timestamp: None, - server_message_data: server_message_data.to_owned(), - citations: Vec::new(), - fetched_memories: Vec::new(), - message: Some(api::message::Message::AgentOutput( - api::message::AgentOutput { - text: fallback_text.to_owned(), - }, - )), - } -} - -fn runtime_activity_client_action(action: api::client_action::Action) -> ResponseEvent { - ResponseEvent { - r#type: Some(api::response_event::Type::ClientActions( - api::response_event::ClientActions { - actions: vec![ClientAction { - action: Some(action), - }], - }, - )), - } -} - -#[cfg(test)] -#[path = "response_translator_tests.rs"] -mod tests; diff --git a/app/src/ai/acp/response_translator_tests.rs b/app/src/ai/acp/response_translator_tests.rs deleted file mode 100644 index 9479e820..00000000 --- a/app/src/ai/acp/response_translator_tests.rs +++ /dev/null @@ -1,411 +0,0 @@ -use galaxy_agent_core::{AgentEvent, RuntimeActivity, RuntimeActivityStatus, StopReason}; -use warp_multi_agent_api::{client_action, message, response_event}; - -use super::AcpResponseTranslator; -use crate::ai::agent::runtime_activity; - -#[test] -fn initializes_the_existing_chat_exchange_and_persists_user_text() { - let mut translator = AcpResponseTranslator::new( - "task".to_owned(), - true, - Some("hello".to_owned()), - "acp:codex".to_owned(), - ); - let events = translator - .translate(AgentEvent::TurnStarted { - runtime_request_id: "session".to_owned(), - }) - .expect("translate"); - - assert!(matches!( - events[0].r#type, - Some(response_event::Type::Init(_)) - )); - assert!(matches!( - events[1].r#type, - Some(response_event::Type::ClientActions(_)) - )); - assert!(matches!( - events[2].r#type, - Some(response_event::Type::ClientActions(_)) - )); -} - -#[test] -fn streams_agent_text_as_add_then_append() { - let mut translator = - AcpResponseTranslator::new("task".to_owned(), false, None, "acp:codex".to_owned()); - let first = translator - .translate(AgentEvent::TextDelta { - text: "one".to_owned(), - }) - .expect("first"); - let second = translator - .translate(AgentEvent::TextDelta { - text: " two".to_owned(), - }) - .expect("second"); - - let Some(response_event::Type::ClientActions(first_actions)) = &first[1].r#type else { - panic!("expected first client action"); - }; - assert!(matches!( - first_actions.actions[0].action, - Some(client_action::Action::AddMessagesToTask(_)) - )); - let Some(response_event::Type::ClientActions(second_actions)) = &second[0].r#type else { - panic!("expected append client action"); - }; - assert!(matches!( - second_actions.actions[0].action, - Some(client_action::Action::AppendToMessageContent(_)) - )); -} - -#[test] -fn renders_acp_tool_progress_as_structured_non_executable_activity() { - let mut translator = - AcpResponseTranslator::new("task".to_owned(), false, None, "acp:codex".to_owned()); - let events = translator - .translate(AgentEvent::RuntimeActivityUpdated { - activity: RuntimeActivity { - id: "tool-1".to_owned(), - title: "Read file".to_owned(), - status: Some(RuntimeActivityStatus::InProgress), - output: None, - }, - }) - .expect("tool"); - let Some(response_event::Type::ClientActions(actions)) = &events[1].r#type else { - panic!("expected client action"); - }; - let Some(client_action::Action::AddMessagesToTask(add)) = &actions.actions[0].action else { - panic!("expected display-only message"); - }; - assert!(matches!( - add.messages[0].message, - Some(message::Message::AgentOutput(_)) - )); - assert_eq!( - runtime_activity::decode(&add.messages[0].server_message_data), - Some(RuntimeActivity { - id: "tool-1".to_owned(), - title: "Read file".to_owned(), - status: Some(RuntimeActivityStatus::InProgress), - output: None, - }) - ); -} - -#[test] -fn maps_usage_and_successful_completion() { - let mut translator = - AcpResponseTranslator::new("task".to_owned(), false, None, "acp:codex".to_owned()); - translator - .translate(AgentEvent::ContextUsageUpdated { - used_tokens: 25, - context_size: 100, - }) - .expect("usage"); - let events = translator - .translate(AgentEvent::TurnStopped { - reason: StopReason::Completed, - }) - .expect("finished"); - let Some(finished) = events.iter().find_map(|event| { - let Some(response_event::Type::Finished(finished)) = &event.r#type else { - return None; - }; - Some(finished) - }) else { - panic!("expected finished"); - }; - assert_eq!(finished.token_usage[0].total_input, 0); - assert_eq!( - finished - .conversation_usage_metadata - .as_ref() - .expect("metadata") - .context_window_usage, - 0.25 - ); -} - -#[test] -fn updates_the_existing_activity_card_with_bounded_tool_output() { - let mut translator = - AcpResponseTranslator::new("task".to_owned(), false, None, "acp:codex".to_owned()); - let started = translator - .translate(AgentEvent::RuntimeActivityUpdated { - activity: RuntimeActivity { - id: "tool-1".to_owned(), - title: "Run tests".to_owned(), - status: Some(RuntimeActivityStatus::InProgress), - output: None, - }, - }) - .expect("started"); - let Some(response_event::Type::ClientActions(start_actions)) = &started[1].r#type else { - panic!("expected initial activity action"); - }; - let Some(client_action::Action::AddMessagesToTask(add)) = &start_actions.actions[0].action - else { - panic!("expected initial activity message"); - }; - let message_id = add.messages[0].id.clone(); - - let events = translator - .translate(AgentEvent::RuntimeActivityUpdated { - activity: RuntimeActivity { - id: "tool-1".to_owned(), - title: "Run tests".to_owned(), - status: Some(RuntimeActivityStatus::Completed), - output: Some("test one ... ok\ntest two ... ok".to_owned()), - }, - }) - .expect("tool"); - - let Some(response_event::Type::ClientActions(update_actions)) = &events[0].r#type else { - panic!("expected update action"); - }; - let Some(client_action::Action::UpdateTaskMessage(update)) = &update_actions.actions[0].action - else { - panic!("expected in-place activity update"); - }; - let updated_message = update.message.as_ref().expect("updated message"); - assert_eq!(updated_message.id, message_id); - assert_eq!( - runtime_activity::decode(&updated_message.server_message_data), - Some(RuntimeActivity { - id: "tool-1".to_owned(), - title: "Run tests".to_owned(), - status: Some(RuntimeActivityStatus::Completed), - output: Some("test one ... ok\ntest two ... ok".to_owned()), - }) - ); -} - -#[test] -fn activity_only_turn_does_not_claim_the_agent_returned_no_output() { - let mut translator = - AcpResponseTranslator::new("task".to_owned(), false, None, "acp:codex".to_owned()); - translator - .translate(AgentEvent::RuntimeActivityUpdated { - activity: RuntimeActivity { - id: "tool-1".to_owned(), - title: "Inspect repository".to_owned(), - status: Some(RuntimeActivityStatus::Completed), - output: Some("done".to_owned()), - }, - }) - .expect("activity"); - - let events = translator - .translate(AgentEvent::TurnStopped { - reason: StopReason::Completed, - }) - .expect("finished"); - - assert_eq!(events.len(), 1); - assert!(matches!( - events[0].r#type, - Some(response_event::Type::Finished(_)) - )); -} - -#[test] -fn successful_turn_without_agent_output_is_still_visible() { - let mut translator = - AcpResponseTranslator::new("task".to_owned(), false, None, "acp:codex".to_owned()); - let events = translator - .translate(AgentEvent::TurnStopped { - reason: StopReason::Completed, - }) - .expect("finished"); - - assert!(events.iter().any(|event| { - let Some(response_event::Type::ClientActions(actions)) = &event.r#type else { - return false; - }; - let Some(client_action::Action::AddMessagesToTask(add)) = &actions.actions[0].action else { - return false; - }; - let Some(message::Message::AgentOutput(output)) = &add.messages[0].message else { - return false; - }; - output.text.contains("completed without a text response") - })); -} - -#[test] -fn reasoning_is_not_flattened_into_the_plain_answer_transcript() { - let mut translator = - AcpResponseTranslator::new("task".to_owned(), false, None, "acp:codex".to_owned()); - - let events = translator - .translate(AgentEvent::ReasoningDelta { - text: "private chain of thought".to_owned(), - }) - .expect("translate"); - - assert!(events.is_empty()); -} - -#[test] -fn live_steering_adds_a_user_bubble_and_starts_a_new_assistant_bubble() { - let mut translator = - AcpResponseTranslator::new("task".to_owned(), false, None, "acp:codex".to_owned()); - translator - .translate(AgentEvent::TextDelta { - text: "original response".to_owned(), - }) - .expect("initial output"); - - let steered = translator - .translate(AgentEvent::UserInputAccepted { - text: "stop at 75s".to_owned(), - }) - .expect("steering"); - let Some(response_event::Type::ClientActions(user_actions)) = &steered[0].r#type else { - panic!("expected user client action"); - }; - let Some(client_action::Action::AddMessagesToTask(add_user)) = &user_actions.actions[0].action - else { - panic!("expected user message"); - }; - assert!(matches!( - add_user.messages[0].message, - Some(message::Message::UserQuery(_)) - )); - - let resumed = translator - .translate(AgentEvent::TextDelta { - text: "steered response".to_owned(), - }) - .expect("resumed output"); - let Some(response_event::Type::ClientActions(agent_actions)) = &resumed[0].r#type else { - panic!("expected agent client action"); - }; - assert!(matches!( - agent_actions.actions[0].action, - Some(client_action::Action::AddMessagesToTask(_)) - )); -} - -#[test] -fn steering_failure_surfaces_an_indeterminate_delivery_warning() { - let mut translator = - AcpResponseTranslator::new("task".to_owned(), false, None, "acp:codex".to_owned()); - translator - .translate(AgentEvent::TurnStarted { - runtime_request_id: "session".to_owned(), - }) - .expect("initialize"); - - let events = translator - .translate(AgentEvent::RuntimeNotice { - message: "Galaxy couldn't confirm that live steering message: turn is no longer active. The agent may not have received it; check the current terminal and file state before retrying.".to_owned(), - }) - .expect("notice"); - - assert_eq!(events.len(), 1); - let Some(response_event::Type::ClientActions(error_actions)) = &events[0].r#type else { - panic!("expected visible error action"); - }; - let Some(client_action::Action::AddMessagesToTask(add_error)) = - &error_actions.actions[0].action - else { - panic!("expected visible error message"); - }; - let Some(message::Message::AgentOutput(output)) = &add_error.messages[0].message else { - panic!("expected agent output"); - }; - assert!(output.text.contains("couldn't confirm")); - assert!(output.text.contains("before retrying")); -} - -#[test] -fn implicit_steering_turn_warning_does_not_recommend_a_blind_retry() { - let mut translator = - AcpResponseTranslator::new("task".to_owned(), false, None, "acp:codex".to_owned()); - - let events = translator - .translate(AgentEvent::RuntimeNotice { - message: "The ACP adapter started that steering message as a separate turn instead of injecting it into the active one. Galaxy terminated the adapter process immediately, but the turn may have begun acting; check the current terminal and file state before retrying.".to_owned(), - }) - .expect("notice"); - - let text = - events - .iter() - .filter_map(|event| match &event.r#type { - Some(response_event::Type::ClientActions(actions)) => actions - .actions - .iter() - .find_map(|action| match &action.action { - Some(client_action::Action::AddMessagesToTask(add)) => add - .messages - .iter() - .find_map(|message| match &message.message { - Some(message::Message::AgentOutput(output)) => { - Some(output.text.as_str()) - } - _ => None, - }), - _ => None, - }), - _ => None, - }) - .collect::(); - assert!(text.contains("started")); - assert!(text.contains("terminated")); - assert!(text.contains("immediately")); - assert!(text.contains("may have begun acting")); -} - -#[test] -fn startup_error_keeps_the_user_request_and_finishes_visibly() { - let mut translator = AcpResponseTranslator::new( - "task".to_owned(), - false, - Some("help me".to_owned()), - "acp:codex".to_owned(), - ); - - let events = translator.startup_error("adapter missing"); - - assert_eq!(events.len(), 4); - assert!(matches!( - events[0].r#type, - Some(response_event::Type::Init(_)) - )); - let Some(response_event::Type::ClientActions(user_actions)) = &events[1].r#type else { - panic!("expected visible user request"); - }; - let Some(client_action::Action::AddMessagesToTask(user_messages)) = - &user_actions.actions[0].action - else { - panic!("expected user message"); - }; - assert!(matches!( - user_messages.messages[0].message, - Some(message::Message::UserQuery(_)) - )); - let Some(response_event::Type::ClientActions(error_actions)) = &events[2].r#type else { - panic!("expected visible startup error"); - }; - let Some(client_action::Action::AddMessagesToTask(error_messages)) = - &error_actions.actions[0].action - else { - panic!("expected error message"); - }; - let Some(message::Message::AgentOutput(output)) = &error_messages.messages[0].message else { - panic!("expected agent output"); - }; - assert!(output.text.contains("adapter missing")); - assert!(matches!( - events[3].r#type, - Some(response_event::Type::Finished(_)) - )); -} diff --git a/app/src/ai/acp/transport.rs b/app/src/ai/acp/transport.rs index adcafc02..c4146177 100644 --- a/app/src/ai/acp/transport.rs +++ b/app/src/ai/acp/transport.rs @@ -9,14 +9,15 @@ use galaxy_acp::{ SessionId, }; use galaxy_agent_core::{ - turn_control, AgentRuntime as _, TurnCommand, TurnCommandSender, TurnRequest, + turn_control, AgentRuntime as _, RuntimeCapabilities, TurnCommand, TurnCommandSender, + TurnRequest, }; use super::launch::acp_selection_identity; use super::prompt::{prompt_content, GalaxyTerminalTools}; -use super::response_translator::AcpResponseTranslator; use crate::ai::agent::api::{self, RequestParams}; use crate::ai::agent::EntrypointType; +use crate::ai::runtime::{RuntimeResponseConfig, RuntimeResponseTranslator}; use crate::persistence::model::AcpConversationData; use crate::server::server_api::AIApiError; @@ -170,18 +171,24 @@ pub(crate) fn acp_startup_error_stream( fn response_translator( params: &RequestParams, backend: &AcpConversationData, -) -> AcpResponseTranslator { +) -> RuntimeResponseTranslator { let task_id = params .root_task_id .clone() .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()); let user_query = request_user_query(params); - AcpResponseTranslator::new( + RuntimeResponseTranslator::new(RuntimeResponseConfig { task_id, - params.tasks.is_empty(), + // ACP owns its session identifier. Keeping this empty prevents the + // compatibility Init event from entering Galaxy cloud-token paths. + conversation_id: String::new(), + needs_create_task: params.tasks.is_empty(), user_query, - acp_selection_identity(&backend.agent_id, &backend.config_values), - ) + model_id: acp_selection_identity(&backend.agent_id, &backend.config_values), + max_context_tokens: None, + capabilities: RuntimeCapabilities::session_runtime(), + empty_output_message: Some("> ACP agent completed without a text response.".to_owned()), + }) } fn request_user_query(params: &RequestParams) -> Option { @@ -250,10 +257,11 @@ fn galaxy_mcp_args( } fn translated_startup_error_stream( - mut translator: AcpResponseTranslator, + mut translator: RuntimeResponseTranslator, message: &str, ) -> api::ResponseStream { - let events = translator.startup_error(message); + let events = + translator.startup_error(&format!("Galaxy couldn't start the ACP agent: {message}")); Box::pin(futures::stream::iter( events .into_iter() diff --git a/app/src/ai/blocklist/context_model.rs b/app/src/ai/blocklist/context_model.rs index fc356e84..858da3c5 100644 --- a/app/src/ai/blocklist/context_model.rs +++ b/app/src/ai/blocklist/context_model.rs @@ -188,13 +188,37 @@ impl BlocklistAIContextModel { ); ctx.subscribe_to_model(&LLMPreferences::handle(ctx), |me, _, event, ctx| { - if let LLMPreferencesEvent::UpdatedActiveAgentModeLLM = event { + if matches!( + event, + LLMPreferencesEvent::UpdatedActiveAgentModeLLM + | LLMPreferencesEvent::UpdatedAvailableLLMs + ) { let llm_prefs = LLMPreferences::as_ref(ctx); let vision_supported = llm_prefs.vision_supported(ctx, Some(me.terminal_surface_id)); + #[cfg(not(target_family = "wasm"))] + let desired_backend = + llm_prefs.agent_backend_for_active_model(Some(me.terminal_surface_id), ctx); if !vision_supported { me.clear_pending_images(ctx); } + + // ACP and provider histories have different owners. When the + // selected model crosses that boundary, make the next prompt a + // fresh conversation instead of silently sending it through + // the backend that owned the existing conversation. + #[cfg(not(target_family = "wasm"))] + { + let selected_backend = me + .selected_conversation(ctx) + .map(|conversation| conversation.agent_backend().clone()); + if selected_backend.is_some_and(|backend| backend != desired_backend) { + me.set_pending_query_state_for_new_conversation( + AgentViewEntryOrigin::ConversationSelector, + ctx, + ); + } + } } }); diff --git a/app/src/ai/blocklist/controller.rs b/app/src/ai/blocklist/controller.rs index 65ba41db..f8c5f9dc 100644 --- a/app/src/ai/blocklist/controller.rs +++ b/app/src/ai/blocklist/controller.rs @@ -818,7 +818,7 @@ impl BlocklistAIController { if can_attempt_live_steering { if let Some((stream_id, model_id)) = self .in_flight_response_streams - .try_steer_acp_stream_for_conversation(conversation_id, query.clone(), ctx) + .try_steer_runtime_for_conversation(conversation_id, query.clone(), ctx) { ctx.emit(BlocklistAIControllerEvent::SentRequest { contains_user_query: true, @@ -3446,7 +3446,7 @@ impl BlocklistAIController { Ok(api::StreamEvent::Response(event)) => { // If this controller is part of a shared session, forward the entire response event to viewers first. if FeatureFlag::AgentSharedSessions.is_enabled() - && !response_stream.as_ref(ctx).is_acp() + && response_stream.as_ref(ctx).supports_shared_session_sync() { let mut model = self.terminal_model.lock(); if model.shared_session_status().is_sharer() { @@ -3527,7 +3527,9 @@ impl BlocklistAIController { // After the stream finishes, persist the full message // history (input + assistant response) from the Arc back // into the conversation for the next request cycle. - let new_history = (!response_stream.as_ref(ctx).is_acp()) + let new_history = response_stream + .as_ref(ctx) + .host_manages_history() .then(|| response_stream.as_ref(ctx).messages_sent().clone()) .and_then(|messages_sent| { messages_sent.lock().ok().and_then(|sent| { @@ -3632,9 +3634,10 @@ impl BlocklistAIController { const MAX_ERROR_RETRIES: usize = 2; let retry_count = self.error_retry_counts.entry(conversation_id).or_insert(0); - let should_corrective_retry = !response_stream.as_ref(ctx).is_acp() - && is_corrective_retry_candidate - && *retry_count < MAX_ERROR_RETRIES; + let should_corrective_retry = + response_stream.as_ref(ctx).allows_corrective_retries() + && is_corrective_retry_candidate + && *retry_count < MAX_ERROR_RETRIES; if should_corrective_retry { *retry_count += 1; diff --git a/app/src/ai/blocklist/controller/pending_response_streams.rs b/app/src/ai/blocklist/controller/pending_response_streams.rs index 804e6e8f..6dce2378 100644 --- a/app/src/ai/blocklist/controller/pending_response_streams.rs +++ b/app/src/ai/blocklist/controller/pending_response_streams.rs @@ -52,11 +52,11 @@ impl PendingResponseStreams { .collect() } - /// Attempts to inject a plain-text follow-up into the active ACP turn. + /// Attempts to inject a plain-text follow-up into an active steerable runtime. /// /// Returning `None` leaves the caller free to use the normal /// cancel-and-queue path without dropping the user's message. - pub fn try_steer_acp_stream_for_conversation( + pub fn try_steer_runtime_for_conversation( &self, conversation_id: AIConversationId, display_text: String, @@ -71,7 +71,7 @@ impl PendingResponseStreams { let model_id = stream.as_ref(app).llm_id().clone(); stream .as_ref(app) - .try_steer_acp(display_text) + .try_steer_runtime(display_text) .then(|| (stream_id.clone(), model_id)) } diff --git a/app/src/ai/blocklist/controller/response_stream.rs b/app/src/ai/blocklist/controller/response_stream.rs index 8aaa6680..9eb745c1 100644 --- a/app/src/ai/blocklist/controller/response_stream.rs +++ b/app/src/ai/blocklist/controller/response_stream.rs @@ -11,6 +11,7 @@ use ::local_control::remote_command::is_potential_remote_ssh_command; use anyhow::anyhow; use chrono::{DateTime, Local, TimeDelta}; use futures::channel::oneshot; +use galaxy_agent_core::RuntimeCapabilities; #[cfg(not(target_family = "wasm"))] use galaxy_agent_core::TurnCommand; #[cfg(not(target_family = "wasm"))] @@ -122,7 +123,7 @@ struct AcpRequestControl { /// received yet, ensuring we don't retry after the AI has started executing actions. pub struct ResponseStream { id: ResponseStreamId, - agent_backend: AgentBackend, + runtime_capabilities: RuntimeCapabilities, #[cfg(not(target_family = "wasm"))] acp_session_metadata: Arc>, #[cfg(not(target_family = "wasm"))] @@ -193,7 +194,7 @@ impl ResponseStream { let (cancellation_tx, _rx) = oneshot::channel(); Self { id, - agent_backend: AgentBackend::Provider, + runtime_capabilities: RuntimeCapabilities::provider(), #[cfg(not(target_family = "wasm"))] acp_session_metadata: Arc::new(Mutex::new(AcpSessionMetadata::default())), #[cfg(not(target_family = "wasm"))] @@ -439,6 +440,10 @@ impl ResponseStream { let start_time = Local::now(); let request_id = Uuid::new_v4(); + let runtime_capabilities = match &agent_backend { + AgentBackend::Provider => RuntimeCapabilities::provider(), + AgentBackend::Acp(_) => RuntimeCapabilities::session_runtime(), + }; #[cfg(not(target_family = "wasm"))] let acp_session_metadata = Arc::new(Mutex::new(AcpSessionMetadata::default())); #[cfg(not(target_family = "wasm"))] @@ -489,7 +494,7 @@ impl ResponseStream { } Self { id: ResponseStreamId(Uuid::new_v4().to_string()), - agent_backend, + runtime_capabilities, #[cfg(not(target_family = "wasm"))] acp_session_metadata, #[cfg(not(target_family = "wasm"))] @@ -516,13 +521,22 @@ impl ResponseStream { &self.id } - pub fn is_acp(&self) -> bool { - matches!(self.agent_backend, AgentBackend::Acp(_)) + pub fn supports_shared_session_sync(&self) -> bool { + self.runtime_capabilities.shared_session_sync + } + + pub fn host_manages_history(&self) -> bool { + self.runtime_capabilities.host_managed_history + } + + pub fn allows_corrective_retries(&self) -> bool { + self.runtime_capabilities.corrective_retries } #[cfg(not(target_family = "wasm"))] pub(crate) fn acp_session_metadata(&self) -> Option { - self.is_acp() + self.runtime_capabilities + .session_resume .then(|| { self.acp_session_metadata .lock() @@ -532,10 +546,10 @@ impl ResponseStream { .flatten() } - pub(super) fn try_steer_acp(&self, display_text: String) -> bool { + pub(super) fn try_steer_runtime(&self, display_text: String) -> bool { #[cfg(not(target_family = "wasm"))] { - if !self.is_acp() + if !self.runtime_capabilities.steering || self.current_request_id.is_none() || !self .acp_session_metadata() @@ -637,7 +651,10 @@ impl ResponseStream { &self, error: &Arc, ) -> bool { - if self.is_acp() || self.coding_model_fallback_attempted || self.has_received_client_actions + if !self.runtime_capabilities.model_selection + || !self.runtime_capabilities.request_retries + || self.coding_model_fallback_attempted + || self.has_received_client_actions { return false; } @@ -820,7 +837,7 @@ impl ResponseStream { let is_online = NetworkStatus::as_ref(ctx).is_online(); match recovery_action( self.has_received_client_actions, - e.is_recoverable() && !self.is_acp(), + e.is_recoverable() && self.runtime_capabilities.request_retries, self.retry_count < MAX_RETRIES, self.can_attempt_resume_on_error, is_online, @@ -893,7 +910,7 @@ impl ResponseStream { let is_online = NetworkStatus::as_ref(ctx).is_online(); match recovery_action( self.has_received_client_actions, - unexpected_eof.is_recoverable() && !self.is_acp(), + unexpected_eof.is_recoverable() && self.runtime_capabilities.request_retries, self.retry_count < MAX_RETRIES, self.can_attempt_resume_on_error, is_online, diff --git a/app/src/ai/blocklist/history_model.rs b/app/src/ai/blocklist/history_model.rs index fa85cff0..5bec2280 100644 --- a/app/src/ai/blocklist/history_model.rs +++ b/app/src/ai/blocklist/history_model.rs @@ -1183,6 +1183,7 @@ impl BlocklistAIHistoryModel { } fn configured_agent_backend( + terminal_surface_id: EntityId, is_viewing_shared_session: bool, is_cli_agent_transcript: bool, ctx: &AppContext, @@ -1200,6 +1201,11 @@ impl BlocklistAIHistoryModel { return AgentBackend::Provider; } + #[cfg(not(target_family = "wasm"))] + if let Some(llm_preferences) = ctx.try_get_singleton_model_as_ref::() { + return llm_preferences.agent_backend_for_active_model(Some(terminal_surface_id), ctx); + } + let configured_agent_id = settings.acp_agent_id.value().trim(); let agent_id = if configured_agent_id.is_empty() { "codex" @@ -1224,12 +1230,6 @@ impl BlocklistAIHistoryModel { .iter() .find(|agent| agent.id.eq_ignore_ascii_case(agent_id)) .map(|agent| { - #[cfg(not(target_family = "wasm"))] - if let Some(selection) = - LLMPreferences::as_ref(ctx).selected_acp_config_for_agent(&agent.name, ctx) - { - return selection; - } crate::ai::acp::AcpRuntimeModel::current_config_values(&agent.config_options) }) .unwrap_or_default(), @@ -1249,7 +1249,12 @@ impl BlocklistAIHistoryModel { let Some(conversation) = self.conversation(&conversation_id) else { return; }; + let Some(terminal_surface_id) = self.terminal_surface_id_for_conversation(&conversation_id) + else { + return; + }; let agent_backend = Self::configured_agent_backend( + terminal_surface_id, conversation.is_viewing_shared_session(), conversation.is_cli_agent_transcript(), ctx, @@ -1277,8 +1282,12 @@ impl BlocklistAIHistoryModel { is_cli_agent_transcript: bool, ctx: &mut ModelContext, ) -> AIConversationId { - let agent_backend = - Self::configured_agent_backend(is_viewing_shared_session, is_cli_agent_transcript, ctx); + let agent_backend = Self::configured_agent_backend( + terminal_surface_id, + is_viewing_shared_session, + is_cli_agent_transcript, + ctx, + ); let mut new_conversation = AIConversation::new_with_agent_backend( is_viewing_shared_session, is_cli_agent_transcript, diff --git a/app/src/ai/llms.rs b/app/src/ai/llms.rs index 30284623..6d27651e 100644 --- a/app/src/ai/llms.rs +++ b/app/src/ai/llms.rs @@ -16,13 +16,15 @@ use warp_multi_agent_api as api; use super::custom_model_routers::{self, CustomModelRouter, ModelConfigError}; use super::execution_profiles::profiles::AIExecutionProfilesModel; -use crate::ai::acp::acp_selection_identity; +use crate::ai::acp::{acp_launch_fingerprint, acp_selection_identity}; use crate::ai::bedrock::models::get_effective_models; use crate::auth::auth_manager::{AuthManager, AuthManagerEvent}; use crate::auth::AuthStateProvider; use crate::network::{NetworkStatus, NetworkStatusEvent, NetworkStatusKind}; +#[cfg(not(target_family = "wasm"))] +use crate::persistence::model::{AcpConversationData, AgentBackend}; use crate::server::server_api::ServerApiProvider; -use crate::settings::{AcpAgentSettings, BedrockModelConfig, OpenAIModelConfig}; +use crate::settings::{AcpConfigValueSettings, BedrockModelConfig, OpenAIModelConfig}; use crate::user_config::{WarpConfig, WarpConfigUpdateEvent}; use crate::workspaces::user_workspaces::{UserWorkspaces, UserWorkspacesEvent}; use crate::{report_error, AISettings}; @@ -112,6 +114,8 @@ pub enum LLMProvider { Bedrock, /// Models served through an OpenAI-compatible proxy (e.g. LiteLLM). LiteLLM, + /// Models selected and executed by an Agent Client Protocol runtime. + Acp, Unknown, } @@ -124,6 +128,7 @@ impl LLMProvider { LLMProvider::Google => Some(Icon::GeminiLogo), LLMProvider::Bedrock => Some(Icon::BedrockLogo), LLMProvider::LiteLLM => Some(Icon::OpenAILogo), + LLMProvider::Acp => Some(Icon::Terminal), LLMProvider::Xai => None, LLMProvider::Unknown => None, } @@ -138,6 +143,7 @@ impl LLMProvider { LLMProvider::Xai => "xAI", LLMProvider::Bedrock => "AWS Bedrock", LLMProvider::LiteLLM => "LiteLLM", + LLMProvider::Acp => "ACP", LLMProvider::Unknown => "this provider", } } @@ -578,7 +584,14 @@ pub struct LLMPreferences { #[cfg(not(target_family = "wasm"))] fetched_openai_models: Vec, #[cfg(not(target_family = "wasm"))] - acp_selections: HashMap>, + acp_selections: HashMap, +} + +#[cfg(not(target_family = "wasm"))] +#[derive(Clone, Debug, PartialEq)] +pub(crate) struct AcpModelSelection { + pub(crate) agent_id: String, + pub(crate) config_values: BTreeMap, } impl LLMPreferences { @@ -699,6 +712,7 @@ impl LLMPreferences { Self::ensure_default_models_in_settings(ctx); me.inject_bedrock_models(ctx); me.inject_openai_models(ctx); + me.ensure_default_model_present(); me.fetch_openai_models_from_endpoint(ctx); } @@ -741,19 +755,12 @@ impl LLMPreferences { #[cfg(not(target_family = "wasm"))] fn inject_bedrock_models(&mut self, ctx: &AppContext) { - // Strip both existing Bedrock models and placeholder Unknown models. - self.models_by_feature - .agent_mode - .choices - .retain(|m| m.provider != LLMProvider::Bedrock && m.provider != LLMProvider::Unknown); - self.models_by_feature - .coding - .choices - .retain(|m| m.provider != LLMProvider::Bedrock && m.provider != LLMProvider::Unknown); + // Galaxy's runtime inventory is rebuilt exclusively from enabled local + // providers. Never retain Warp-hosted or stale cached model entries. + self.models_by_feature.agent_mode.choices.clear(); + self.models_by_feature.coding.choices.clear(); if let Some(ref mut cli) = self.models_by_feature.cli_agent { - cli.choices.retain(|m| { - m.provider != LLMProvider::Bedrock && m.provider != LLMProvider::Unknown - }); + cli.choices.clear(); } let settings = AISettings::as_ref(ctx); @@ -1062,93 +1069,202 @@ impl LLMPreferences { if !*settings.acp_enabled.value() { return; } - for agent in settings.acp_agents.value() { - let model_option = agent - .config_options - .iter() - .find(|option| option.category.as_deref() == Some("model")); - let Some(model_option) = model_option else { - continue; - }; - let secondary = agent.config_options.iter().filter(|option| { - matches!( - option.category.as_deref(), - Some("mode") | Some("thought_level") - ) - }); - for value in &model_option.options { - let suffix = secondary - .clone() - .filter_map(|option| { - option - .options - .iter() - .find(|value| value.value == option.current_value) - .or_else(|| option.options.first()) - .map(|value| value.name.clone()) - }) - .collect::>(); - let display_name = if suffix.is_empty() { - value.name.clone() - } else { - format!("{} ({})", value.name, suffix.join(", ")) - }; - let mut selection = - crate::ai::acp::AcpRuntimeModel::current_config_values(&agent.config_options); - selection.insert(model_option.id.clone(), value.value.clone()); - let id = acp_selection_identity(&agent.id, &selection); - let llm_id = LLMId::from(id.as_str()); - self.acp_selections.insert(llm_id.clone(), selection); - let info = LLMInfo { - id: llm_id, - display_name, - base_model_name: value.name.clone(), - reasoning_level: None, - usage_metadata: LLMUsageMetadata { - request_multiplier: 1, - credit_multiplier: None, - }, - description: Some(agent.name.clone()), - disable_reason: None, - vision_supported: false, - spec: None, - provider: LLMProvider::Unknown, - host_configs: HashMap::new(), - discount_percentage: None, - context_window: LLMContextWindow::default(), - }; - self.models_by_feature.agent_mode.choices.push(info.clone()); - self.models_by_feature.coding.choices.push(info.clone()); - if let Some(ref mut cli) = self.models_by_feature.cli_agent { - cli.choices.push(info); + let configured_agent_id = settings.acp_agent_id.value().trim(); + let configured_agent_id = if configured_agent_id.is_empty() { + "codex" + } else { + configured_agent_id + }; + let bedrock_enabled = *settings.bedrock_enabled.value(); + let configured_agent = settings + .acp_agents + .value() + .iter() + .find(|agent| agent.id.eq_ignore_ascii_case(configured_agent_id)); + let Some(agent) = configured_agent else { + let display_name = acp_agent_display_name(configured_agent_id); + self.push_acp_model( + configured_agent_id, + &display_name, + &display_name, + BTreeMap::new(), + None, + ); + return; + }; + let model_option = agent + .config_options + .iter() + .find(|option| option.category.as_deref() == Some("model")); + let Some(model_option) = model_option else { + let selection = + crate::ai::acp::AcpRuntimeModel::current_config_values(&agent.config_options); + self.push_acp_model(&agent.id, &agent.name, &agent.name, selection, None); + return; + }; + let reasoning_option = agent + .config_options + .iter() + .find(|option| option.category.as_deref() == Some("thought_level")); + for value in model_option + .options + .iter() + .filter(|value| acp_model_is_enabled(&value.value, bedrock_enabled)) + { + let mut selection = + crate::ai::acp::AcpRuntimeModel::current_config_values(&agent.config_options); + selection.insert(model_option.id.clone(), value.value.clone()); + if let Some(reasoning_option) = + reasoning_option.filter(|option| !option.options.is_empty()) + { + for reasoning in &reasoning_option.options { + let mut selection = selection.clone(); + selection.insert(reasoning_option.id.clone(), reasoning.value.clone()); + self.push_acp_model( + &agent.id, + &value.name, + &value.name, + selection, + Some(reasoning), + ); } + } else { + self.push_acp_model(&agent.id, &value.name, &value.name, selection, None); } } } #[cfg(not(target_family = "wasm"))] - pub fn acp_selection_for_model( - &self, - model_id: &LLMId, - ) -> Option<&BTreeMap> { - self.acp_selections.get(model_id) + fn push_acp_model( + &mut self, + agent_id: &str, + display_name: &str, + base_model_name: &str, + selection: BTreeMap, + reasoning: Option<&AcpConfigValueSettings>, + ) { + let display_name = reasoning.map_or_else( + || display_name.to_owned(), + |reasoning| format!("{display_name} ({})", reasoning.name), + ); + let id = acp_selection_identity(agent_id, &selection); + let llm_id = LLMId::from(id.as_str()); + self.acp_selections.insert( + llm_id.clone(), + AcpModelSelection { + agent_id: agent_id.to_owned(), + config_values: selection, + }, + ); + let info = LLMInfo { + id: llm_id, + display_name, + base_model_name: base_model_name.to_owned(), + reasoning_level: reasoning.map(|reasoning| reasoning.name.clone()), + usage_metadata: LLMUsageMetadata { + request_multiplier: 1, + credit_multiplier: None, + }, + description: None, + disable_reason: None, + vision_supported: false, + spec: None, + provider: LLMProvider::Acp, + host_configs: HashMap::new(), + discount_percentage: None, + context_window: LLMContextWindow::default(), + }; + self.models_by_feature.agent_mode.choices.push(info.clone()); + self.models_by_feature.coding.choices.push(info.clone()); + if let Some(ref mut cli) = self.models_by_feature.cli_agent { + cli.choices.push(info); + } } #[cfg(not(target_family = "wasm"))] - pub fn selected_acp_config_for_agent( + pub(crate) fn acp_runtime_selection_for_model( &self, - agent_name: &str, + model_id: &LLMId, + ) -> Option<&AcpModelSelection> { + self.acp_selections.get(model_id) + } + + /// Resolves the runtime that owns the active model for a terminal surface. + /// + /// ACP is an execution backend, not a global lock on Agent Mode. Selecting + /// an ACP-advertised model routes the conversation to that ACP agent, while + /// selecting a Rig/provider model routes it through Galaxy's provider path. + #[cfg(not(target_family = "wasm"))] + pub(crate) fn agent_backend_for_active_model( + &self, + terminal_view_id: Option, ctx: &AppContext, - ) -> Option> { - let profile = AIExecutionProfilesModel::as_ref(ctx).active_profile(None, ctx); - let model_id = profile.data().base_model.as_ref()?; - let model = self.models_by_feature.agent_mode.info_for_id(model_id)?; - model - .description - .as_deref() - .is_some_and(|name| name.eq_ignore_ascii_case(agent_name)) - .then(|| self.acp_selections.get(model_id).cloned()) - .flatten() + ) -> AgentBackend { + if !cfg!(unix) || !FeatureFlag::AgentClientProtocol.is_enabled() { + return AgentBackend::Provider; + } + + let settings = AISettings::as_ref(ctx); + if !*settings.acp_enabled.value() { + return AgentBackend::Provider; + } + + let active_model = self.get_active_base_model(ctx, terminal_view_id); + if let Some(selection) = self.acp_runtime_selection_for_model(&active_model.id) { + return AgentBackend::Acp(AcpConversationData { + agent_id: selection.agent_id.clone(), + launch_fingerprint: acp_launch_fingerprint( + &selection.agent_id, + settings.acp_agent_command.value(), + settings.acp_agent_args.value(), + ), + session_id: None, + config_values: selection.config_values.clone(), + }); + } + + if active_model.id.as_str() != "none" { + return AgentBackend::Provider; + } + + // ACP remains a valid runtime even before discovery has returned a + // model option (and for agents that do not expose model selection at + // all). A discovered model catalog with no enabled entries must not + // fall back to its disabled current model, though. + let configured_agent_id = settings.acp_agent_id.value().trim(); + let agent_id = if configured_agent_id.is_empty() { + "codex" + } else { + configured_agent_id + }; + let configured_agent = settings + .acp_agents + .value() + .iter() + .find(|agent| agent.id.eq_ignore_ascii_case(agent_id)); + if configured_agent.is_some_and(|agent| { + agent + .config_options + .iter() + .any(|option| option.category.as_deref() == Some("model")) + }) { + return AgentBackend::Provider; + } + + AgentBackend::Acp(AcpConversationData { + agent_id: agent_id.to_owned(), + launch_fingerprint: acp_launch_fingerprint( + agent_id, + settings.acp_agent_command.value(), + settings.acp_agent_args.value(), + ), + session_id: None, + config_values: configured_agent + .map(|agent| { + crate::ai::acp::AcpRuntimeModel::current_config_values(&agent.config_options) + }) + .unwrap_or_default(), + }) } /// Ensures the default model ID in each feature's choices still points to @@ -2064,6 +2180,23 @@ impl Entity for LLMPreferences { impl SingletonEntity for LLMPreferences {} +#[cfg(not(target_family = "wasm"))] +fn acp_agent_display_name(agent_id: &str) -> String { + match agent_id.to_ascii_lowercase().as_str() { + "codex" => "Codex".to_owned(), + "opencode" => "OpenCode".to_owned(), + _ => agent_id.to_owned(), + } +} + +#[cfg(not(target_family = "wasm"))] +fn acp_model_is_enabled(value: &serde_json::Value, bedrock_enabled: bool) -> bool { + bedrock_enabled + || !value + .as_str() + .is_some_and(|model_id| model_id.starts_with("amazon-bedrock/")) +} + fn get_new_agent_mode_choices( old_config: &AvailableLLMs, new_config: &AvailableLLMs, diff --git a/app/src/ai/llms_tests.rs b/app/src/ai/llms_tests.rs index c3f19afb..9101ae41 100644 --- a/app/src/ai/llms_tests.rs +++ b/app/src/ai/llms_tests.rs @@ -155,6 +155,62 @@ fn openai_model(model_id: &str) -> OpenAIModelConfig { } } +fn acp_select_option( + id: &str, + category: &str, + current_value: &str, + values: &[(&str, &str)], +) -> AcpConfigOptionSettings { + AcpConfigOptionSettings { + id: id.to_owned(), + name: id.to_owned(), + description: None, + category: Some(category.to_owned()), + kind: "select".to_owned(), + current_value: serde_json::json!(current_value), + options: values + .iter() + .map(|(value, name)| AcpConfigValueSettings { + value: serde_json::json!(value), + name: (*name).to_owned(), + description: None, + }) + .collect(), + } +} + +fn acp_agent( + id: &str, + name: &str, + config_options: Vec, +) -> AcpAgentSettings { + AcpAgentSettings { + id: id.to_owned(), + name: name.to_owned(), + version: None, + description: None, + icon_url: None, + capabilities: Vec::new(), + config_options, + discovery_timestamp: None, + discovery_source: None, + discovery_error: None, + } +} + +fn empty_preferences() -> LLMPreferences { + LLMPreferences { + models_by_feature: ModelsByFeature::default(), + last_update: None, + base_llm_for_terminal_view: HashMap::new(), + custom_llms: Vec::new(), + custom_model_routers: Vec::new(), + openai_provider_routing: HashMap::new(), + fetched_openai_models: Vec::new(), + acp_selections: HashMap::new(), + } +} + #[test] fn provider_discovery_preserves_local_model_overrides() { let mut existing = openai_model("codex-gpt-5.6-sol-xhigh"); @@ -275,6 +331,231 @@ fn acp_models_are_injected_only_while_acp_is_enabled() { }); } +#[test] +fn acp_is_selectable_before_protocol_model_discovery_completes() { + App::test((), |mut app| async move { + initialize_settings_for_tests(&mut app); + AISettings::handle(&app).update(&mut app, |settings, ctx| { + settings + .acp_enabled + .set_value(true, ctx) + .expect("ACP setting should update"); + settings + .acp_agent_id + .set_value("opencode".to_owned(), ctx) + .expect("ACP agent should update"); + settings + .bedrock_enabled + .set_value(false, ctx) + .expect("Bedrock setting should update"); + settings + .acp_agents + .set_value(Vec::new(), ctx) + .expect("ACP discovery cache should update"); + }); + + let mut preferences = empty_preferences(); + app.read(|ctx| { + preferences.inject_bedrock_models(ctx); + preferences.inject_acp_models(ctx); + }); + + let models = &preferences.models_by_feature.agent_mode.choices; + assert_eq!(models.len(), 1); + assert_eq!(models[0].display_name, "OpenCode"); + assert_eq!(models[0].provider, LLMProvider::Acp); + assert_eq!( + preferences + .acp_selections + .get(&models[0].id) + .map(|selection| selection.agent_id.as_str()), + Some("opencode") + ); + assert!(preferences.has_any_provider_models()); + }); +} + +#[test] +fn acp_models_expand_reasoning_levels_for_only_the_configured_agent() { + App::test((), |mut app| async move { + initialize_settings_for_tests(&mut app); + AISettings::handle(&app).update(&mut app, |settings, ctx| { + settings + .acp_enabled + .set_value(true, ctx) + .expect("ACP setting should update"); + settings + .acp_agent_id + .set_value("opencode".to_owned(), ctx) + .expect("ACP agent should update"); + settings + .bedrock_enabled + .set_value(false, ctx) + .expect("Bedrock setting should update"); + settings + .acp_agents + .set_value( + vec![ + acp_agent( + "codex", + "Codex", + vec![acp_select_option( + "model", + "model", + "stale-model", + &[("stale-model", "Stale Model")], + )], + ), + acp_agent( + "opencode", + "OpenCode", + vec![ + acp_select_option( + "model", + "model", + "gpt-test", + &[("gpt-test", "GPT Test")], + ), + acp_select_option( + "mode", + "mode", + "read-only", + &[("read-only", "Read-only")], + ), + acp_select_option( + "thought_level", + "thought_level", + "xhigh", + &[("high", "High"), ("xhigh", "Xhigh")], + ), + ], + ), + ], + ctx, + ) + .expect("ACP agents should update"); + }); + + let mut preferences = empty_preferences(); + app.read(|ctx| { + preferences.inject_bedrock_models(ctx); + preferences.inject_acp_models(ctx); + }); + + let models = &preferences.models_by_feature.agent_mode.choices; + assert_eq!(models.len(), 2); + assert_eq!( + models + .iter() + .map(|model| model.display_name.as_str()) + .collect::>(), + HashSet::from(["GPT Test (High)", "GPT Test (Xhigh)"]) + ); + assert!(models.iter().all(|model| { + model.provider == LLMProvider::Acp && !model.display_name.contains("Read-only") + })); + assert!(preferences.acp_selections.values().all(|selection| { + selection.agent_id == "opencode" + && selection.config_values.get("mode") == Some(&serde_json::json!("read-only")) + })); + }); +} + +#[test] +fn acp_bedrock_models_are_hidden_while_bedrock_is_disabled() { + App::test((), |mut app| async move { + initialize_settings_for_tests(&mut app); + AISettings::handle(&app).update(&mut app, |settings, ctx| { + settings + .acp_enabled + .set_value(true, ctx) + .expect("ACP setting should update"); + settings + .acp_agent_id + .set_value("opencode".to_owned(), ctx) + .expect("ACP agent should update"); + settings + .bedrock_enabled + .set_value(false, ctx) + .expect("Bedrock setting should update"); + settings + .acp_agents + .set_value( + vec![acp_agent( + "opencode", + "OpenCode", + vec![acp_select_option( + "model", + "model", + "openai/gpt-test", + &[ + ("amazon-bedrock/claude-test", "Bedrock Claude"), + ("openai/gpt-test", "GPT Test"), + ], + )], + )], + ctx, + ) + .expect("ACP agents should update"); + }); + + let mut preferences = empty_preferences(); + app.read(|ctx| { + preferences.inject_bedrock_models(ctx); + preferences.inject_acp_models(ctx); + }); + assert_eq!(preferences.models_by_feature.agent_mode.choices.len(), 1); + assert_eq!( + preferences.models_by_feature.agent_mode.choices[0].display_name, + "GPT Test" + ); + + AISettings::handle(&app).update(&mut app, |settings, ctx| { + settings + .bedrock_enabled + .set_value(true, ctx) + .expect("Bedrock setting should update"); + }); + app.read(|ctx| preferences.inject_acp_models(ctx)); + assert_eq!(preferences.models_by_feature.agent_mode.choices.len(), 2); + }); +} + +#[test] +fn disabled_providers_do_not_leave_models_in_the_runtime_inventory() { + App::test((), |mut app| async move { + initialize_settings_for_tests(&mut app); + AISettings::handle(&app).update(&mut app, |settings, ctx| { + settings + .bedrock_enabled + .set_value(false, ctx) + .expect("Bedrock setting should update"); + settings + .openai_enabled + .set_value(false, ctx) + .expect("OpenAI setting should update"); + settings + .acp_enabled + .set_value(false, ctx) + .expect("ACP setting should update"); + }); + + let mut preferences = empty_preferences(); + app.read(|ctx| { + preferences.inject_bedrock_models(ctx); + preferences.inject_openai_models(ctx); + }); + + assert!(preferences.models_by_feature.agent_mode.choices.is_empty()); + assert!(preferences.models_by_feature.coding.choices.is_empty()); + assert!(preferences + .models_by_feature + .cli_agent + .as_ref() + .is_none_or(|models| models.choices.is_empty())); + }); +} + #[test] fn provider_discovery_enables_rig_for_new_models_and_keeps_manual_models() { let manual = openai_model("manual-model"); diff --git a/app/src/ai/runtime/event_translator.rs b/app/src/ai/runtime/event_translator.rs new file mode 100644 index 00000000..41cbc62c --- /dev/null +++ b/app/src/ai/runtime/event_translator.rs @@ -0,0 +1,484 @@ +use std::collections::HashMap; + +use galaxy_agent_core::{ + AgentEvent, RuntimeActivity, RuntimeActivityStatus, RuntimeCapabilities, StopReason, Usage, +}; +use uuid::Uuid; +use warp_multi_agent_api::response_event::stream_finished; +use warp_multi_agent_api::{self as api, ClientAction, ResponseEvent}; + +use crate::ai::agent::runtime_activity; +use crate::ai::bedrock::response_translator::{ + build_add_agent_output_message, build_append_text, build_create_task, build_stream_init, + build_user_query_message, +}; +use crate::ai::openai::response_translator::{build_stream_finished, StreamUsage}; + +pub(crate) struct RuntimeResponseConfig { + pub(crate) task_id: String, + pub(crate) conversation_id: String, + pub(crate) needs_create_task: bool, + pub(crate) user_query: Option, + pub(crate) model_id: String, + pub(crate) max_context_tokens: Option, + pub(crate) capabilities: RuntimeCapabilities, + pub(crate) empty_output_message: Option, +} + +/// Converts the provider-neutral runtime lifecycle into Galaxy's existing +/// transcript protocol. Runtime implementations never need to know about the +/// protobuf messages consumed by the UI. +pub(crate) struct RuntimeResponseTranslator { + config: RuntimeResponseConfig, + request_id: String, + initialized: bool, + text_message_id: Option, + reasoning_message_id: Option, + activity_message_ids: HashMap, + activities: HashMap, + has_visible_output: bool, + usage: Usage, + context_usage: Option<(u64, u64)>, +} + +impl RuntimeResponseTranslator { + pub(crate) fn new(config: RuntimeResponseConfig) -> Self { + Self { + config, + request_id: Uuid::new_v4().to_string(), + initialized: false, + text_message_id: None, + reasoning_message_id: None, + activity_message_ids: HashMap::new(), + activities: HashMap::new(), + has_visible_output: false, + usage: Usage::default(), + context_usage: None, + } + } + + pub(crate) fn translate(&mut self, event: AgentEvent) -> Result, String> { + let mut events = Vec::new(); + match event { + AgentEvent::TurnStarted { .. } => self.initialize(&mut events), + AgentEvent::TextDelta { text } => { + self.initialize(&mut events); + self.add_or_append_text(&text, &mut events); + } + AgentEvent::ReasoningDelta { text } => { + self.initialize(&mut events); + self.add_or_append_reasoning(&text, &mut events); + } + AgentEvent::ReasoningCompleted { text, .. } => { + self.initialize(&mut events); + if self.reasoning_message_id.is_none() && !text.is_empty() { + self.add_or_append_reasoning(&text, &mut events); + } + } + AgentEvent::RuntimeActivityUpdated { activity } => { + if self.config.capabilities.host_tool_execution { + return Err( + "a host-tool runtime emitted runtime-owned tool activity".to_owned() + ); + } + self.initialize(&mut events); + self.text_message_id = None; + self.upsert_runtime_activity(activity, &mut events)?; + } + AgentEvent::ContextUsageUpdated { + used_tokens, + context_size, + } => { + if self.config.capabilities.host_managed_history { + return Err( + "a host-history runtime reported session context occupancy".to_owned() + ); + } + self.context_usage = Some((used_tokens, context_size)); + } + AgentEvent::UserInputAccepted { text } => { + if !self.config.capabilities.steering { + return Err("a non-steerable runtime accepted live user input".to_owned()); + } + self.initialize(&mut events); + events.push(build_user_query_message(&self.config.task_id, &text)); + self.text_message_id = None; + self.reasoning_message_id = None; + } + AgentEvent::RuntimeNotice { message } => { + self.initialize(&mut events); + self.text_message_id = None; + self.add_or_append_text(&message, &mut events); + self.text_message_id = None; + } + AgentEvent::UsageUpdated { usage } => { + if !self.config.capabilities.host_managed_history { + return Err("a session runtime reported provider request usage".to_owned()); + } + self.usage = usage; + } + AgentEvent::TurnStopped { reason } => { + self.initialize(&mut events); + if !self.has_visible_output && reason != StopReason::Cancelled { + if let Some(message) = self.config.empty_output_message.clone() { + self.add_or_append_text(&message, &mut events); + } + } + events.push(self.finished(reason)); + } + AgentEvent::Tool { .. } => { + let owner = if self.config.capabilities.host_tool_execution { + "host-tool runtime emitted an unhandled tool lifecycle event" + } else { + "session runtime handed runtime-owned tool execution to Galaxy" + }; + return Err(owner.to_owned()); + } + } + Ok(events) + } + + pub(crate) fn startup_error(&mut self, message: &str) -> Vec { + let mut events = Vec::new(); + self.initialize(&mut events); + self.text_message_id = None; + self.add_or_append_text(message, &mut events); + events.push(self.finished(StopReason::Refusal)); + events + } + + fn initialize(&mut self, events: &mut Vec) { + if self.initialized { + return; + } + events.push(build_stream_init( + &self.request_id, + &self.config.conversation_id, + )); + if self.config.needs_create_task { + events.push(build_create_task(&self.config.task_id)); + } + if let Some(user_query) = &self.config.user_query { + events.push(build_user_query_message(&self.config.task_id, user_query)); + } + self.initialized = true; + } + + fn add_or_append_text(&mut self, text: &str, events: &mut Vec) { + if text.is_empty() { + return; + } + self.has_visible_output = true; + if let Some(message_id) = &self.text_message_id { + events.push(build_append_text(&self.config.task_id, message_id, text)); + } else { + let message_id = Uuid::new_v4().to_string(); + events.push(build_add_agent_output_message( + &self.config.task_id, + &message_id, + text, + )); + self.text_message_id = Some(message_id); + } + } + + fn add_or_append_reasoning(&mut self, text: &str, events: &mut Vec) { + if text.is_empty() { + return; + } + self.has_visible_output = true; + if let Some(message_id) = &self.reasoning_message_id { + events.push(build_reasoning_message( + &self.config.task_id, + message_id, + text, + true, + )); + } else { + let message_id = Uuid::new_v4().to_string(); + events.push(build_reasoning_message( + &self.config.task_id, + &message_id, + text, + false, + )); + self.reasoning_message_id = Some(message_id); + } + } + + fn upsert_runtime_activity( + &mut self, + activity: RuntimeActivity, + events: &mut Vec, + ) -> Result<(), String> { + let activity_id = activity.id.clone(); + let merged_activity = self + .activities + .entry(activity_id.clone()) + .or_insert_with(|| activity.clone()); + if !activity.title.trim().is_empty() { + merged_activity.title = activity.title; + } + if activity.status.is_some() { + merged_activity.status = activity.status; + } + if activity.output.is_some() { + merged_activity.output = activity.output; + } + + let server_message_data = runtime_activity::encode(merged_activity) + .map_err(|error| format!("failed to encode runtime activity: {error}"))?; + let fallback_text = runtime_activity_fallback_text(merged_activity); + if let Some(message_id) = self.activity_message_ids.get(&activity_id) { + events.push(build_update_runtime_activity_message( + &self.config.task_id, + message_id, + &fallback_text, + &server_message_data, + )); + } else { + let message_id = Uuid::new_v4().to_string(); + events.push(build_add_runtime_activity_message( + &self.config.task_id, + &message_id, + &fallback_text, + &server_message_data, + )); + self.activity_message_ids.insert(activity_id, message_id); + } + self.has_visible_output = true; + Ok(()) + } + + fn finished(&self, reason: StopReason) -> ResponseEvent { + let reason = map_stop_reason(reason); + if !self.config.capabilities.host_managed_history { + let (used_tokens, context_size) = self.context_usage.unwrap_or_default(); + return build_context_finished( + reason, + &self.config.model_id, + used_tokens, + context_size, + ); + } + build_stream_finished( + reason, + StreamUsage { + input_tokens: saturating_i32(self.usage.input_tokens), + output_tokens: saturating_i32(self.usage.output_tokens), + cache_read_tokens: saturating_i32(self.usage.cached_input_tokens), + cache_write_tokens: saturating_i32(self.usage.cache_creation_input_tokens), + cost_in_cents: 0.0, + model_id: self.config.model_id.clone(), + max_context_tokens: self.config.max_context_tokens, + }, + ) + } +} + +fn build_reasoning_message( + task_id: &str, + message_id: &str, + text: &str, + append: bool, +) -> ResponseEvent { + let message = api::Message { + id: message_id.to_owned(), + task_id: task_id.to_owned(), + request_id: String::new(), + timestamp: None, + server_message_data: String::new(), + citations: Vec::new(), + fetched_memories: Vec::new(), + message: Some(api::message::Message::AgentReasoning( + api::message::AgentReasoning { + reasoning: text.to_owned(), + finished_duration: None, + }, + )), + }; + let action = if append { + api::client_action::Action::AppendToMessageContent( + api::client_action::AppendToMessageContent { + task_id: task_id.to_owned(), + message: Some(message), + mask: Some(prost_types::FieldMask { + paths: vec!["agent_reasoning.reasoning".to_owned()], + }), + }, + ) + } else { + api::client_action::Action::AddMessagesToTask(api::client_action::AddMessagesToTask { + task_id: task_id.to_owned(), + messages: vec![message], + }) + }; + runtime_client_action(action) +} + +fn runtime_activity_fallback_text(activity: &RuntimeActivity) -> String { + let title = &activity.title; + let status = activity.status.as_ref().map(|status| match status { + RuntimeActivityStatus::Pending => "waiting", + RuntimeActivityStatus::InProgress => "running", + RuntimeActivityStatus::Completed => "completed", + RuntimeActivityStatus::Failed => "failed", + RuntimeActivityStatus::Other(_) => "updated", + }); + let mut text = match status { + Some(status) => format!("> **{title}** — {status}"), + None => format!("> **{title}**"), + }; + if let Some(output) = &activity.output { + text.push_str("\n\n"); + for line in output.lines() { + text.push_str(" "); + text.push_str(line); + text.push('\n'); + } + } + text +} + +fn build_add_runtime_activity_message( + task_id: &str, + message_id: &str, + fallback_text: &str, + server_message_data: &str, +) -> ResponseEvent { + let message = runtime_activity_message(task_id, message_id, fallback_text, server_message_data); + runtime_client_action(api::client_action::Action::AddMessagesToTask( + api::client_action::AddMessagesToTask { + task_id: task_id.to_owned(), + messages: vec![message], + }, + )) +} + +fn build_update_runtime_activity_message( + task_id: &str, + message_id: &str, + fallback_text: &str, + server_message_data: &str, +) -> ResponseEvent { + let message = runtime_activity_message(task_id, message_id, fallback_text, server_message_data); + runtime_client_action(api::client_action::Action::UpdateTaskMessage( + api::client_action::UpdateTaskMessage { + task_id: task_id.to_owned(), + message: Some(message), + mask: Some(prost_types::FieldMask { + paths: vec![ + "agent_output.text".to_owned(), + "server_message_data".to_owned(), + ], + }), + }, + )) +} + +fn runtime_activity_message( + task_id: &str, + message_id: &str, + fallback_text: &str, + server_message_data: &str, +) -> api::Message { + api::Message { + id: message_id.to_owned(), + task_id: task_id.to_owned(), + request_id: String::new(), + timestamp: None, + server_message_data: server_message_data.to_owned(), + citations: Vec::new(), + fetched_memories: Vec::new(), + message: Some(api::message::Message::AgentOutput( + api::message::AgentOutput { + text: fallback_text.to_owned(), + }, + )), + } +} + +fn runtime_client_action(action: api::client_action::Action) -> ResponseEvent { + ResponseEvent { + r#type: Some(api::response_event::Type::ClientActions( + api::response_event::ClientActions { + actions: vec![ClientAction { + action: Some(action), + }], + }, + )), + } +} + +fn build_context_finished( + reason: stream_finished::Reason, + model_id: &str, + used_tokens: u64, + context_size: u64, +) -> ResponseEvent { + let total_input_tokens = u32::try_from(used_tokens).unwrap_or(u32::MAX); + let context_window_usage = if context_size == 0 { + 0.0 + } else { + (used_tokens as f32 / context_size as f32).clamp(0.0, 1.0) + }; + #[allow(deprecated)] + let usage_metadata = stream_finished::ConversationUsageMetadata { + context_window_usage, + summarized: false, + credits_spent: 0.0, + platform_credits_spent: 0.0, + total_input_tokens, + token_usage: Vec::new(), + tool_usage_metadata: None, + warp_token_usage: HashMap::new(), + byok_token_usage: HashMap::new(), + custom_endpoint_token_usage: HashMap::new(), + context_window_segments: Vec::new(), + }; + ResponseEvent { + r#type: Some(api::response_event::Type::Finished( + api::response_event::StreamFinished { + reason: Some(reason), + token_usage: vec![stream_finished::TokenUsage { + model_id: model_id.to_owned(), + // Session runtimes report current occupancy, not tokens + // consumed by this individual request. + total_input: 0, + output: 0, + input_cache_read: 0, + input_cache_write: 0, + cost_in_cents: 0.0, + }], + should_refresh_model_config: false, + request_cost: None, + conversation_usage_metadata: Some(usage_metadata), + }, + )), + } +} + +pub(crate) fn map_stop_reason(reason: StopReason) -> stream_finished::Reason { + match reason { + StopReason::Completed | StopReason::Cancelled => { + stream_finished::Reason::Done(stream_finished::Done {}) + } + StopReason::MaxTokens => { + stream_finished::Reason::MaxTokenLimit(stream_finished::ReachedMaxTokenLimit {}) + } + StopReason::ContextWindowExceeded => stream_finished::Reason::ContextWindowExceeded( + stream_finished::ContextWindowExceeded {}, + ), + StopReason::Refusal | StopReason::ToolLoopLimit | StopReason::Other(_) => { + stream_finished::Reason::Other(stream_finished::Other {}) + } + } +} + +pub(crate) fn saturating_i32(value: u64) -> i32 { + i32::try_from(value).unwrap_or(i32::MAX) +} + +#[cfg(test)] +#[path = "event_translator_tests.rs"] +mod tests; diff --git a/app/src/ai/runtime/event_translator_tests.rs b/app/src/ai/runtime/event_translator_tests.rs new file mode 100644 index 00000000..c2c9e260 --- /dev/null +++ b/app/src/ai/runtime/event_translator_tests.rs @@ -0,0 +1,210 @@ +use galaxy_agent_core::{ + AgentEvent, RuntimeActivity, RuntimeActivityStatus, RuntimeCapabilities, StopReason, Usage, +}; +use warp_multi_agent_api::{client_action, message, response_event}; + +use super::{RuntimeResponseConfig, RuntimeResponseTranslator}; +use crate::ai::agent::runtime_activity; + +fn provider_translator() -> RuntimeResponseTranslator { + RuntimeResponseTranslator::new(RuntimeResponseConfig { + task_id: "task".to_owned(), + conversation_id: "conversation".to_owned(), + needs_create_task: false, + user_query: None, + model_id: "model".to_owned(), + max_context_tokens: Some(1_000), + capabilities: RuntimeCapabilities::provider(), + empty_output_message: None, + }) +} + +fn session_translator() -> RuntimeResponseTranslator { + RuntimeResponseTranslator::new(RuntimeResponseConfig { + task_id: "task".to_owned(), + conversation_id: String::new(), + needs_create_task: false, + user_query: None, + model_id: "session-runtime".to_owned(), + max_context_tokens: None, + capabilities: RuntimeCapabilities::session_runtime(), + empty_output_message: Some("> runtime completed without text".to_owned()), + }) +} + +#[test] +fn provider_and_session_runtimes_share_text_translation() { + for mut translator in [provider_translator(), session_translator()] { + let first = translator + .translate(AgentEvent::TextDelta { + text: "one".to_owned(), + }) + .expect("first delta"); + let second = translator + .translate(AgentEvent::TextDelta { + text: " two".to_owned(), + }) + .expect("second delta"); + + let Some(response_event::Type::ClientActions(first_actions)) = &first[1].r#type else { + panic!("expected first client action"); + }; + assert!(matches!( + first_actions.actions[0].action, + Some(client_action::Action::AddMessagesToTask(_)) + )); + let Some(response_event::Type::ClientActions(second_actions)) = &second[0].r#type else { + panic!("expected append client action"); + }; + assert!(matches!( + second_actions.actions[0].action, + Some(client_action::Action::AppendToMessageContent(_)) + )); + } +} + +#[test] +fn reasoning_uses_the_native_reasoning_message_contract() { + let mut translator = provider_translator(); + let first = translator + .translate(AgentEvent::ReasoningDelta { + text: "think".to_owned(), + }) + .expect("reasoning"); + let second = translator + .translate(AgentEvent::ReasoningDelta { + text: " more".to_owned(), + }) + .expect("reasoning append"); + + let Some(response_event::Type::ClientActions(actions)) = &first[1].r#type else { + panic!("expected reasoning action"); + }; + let Some(client_action::Action::AddMessagesToTask(add)) = &actions.actions[0].action else { + panic!("expected reasoning message"); + }; + assert!(matches!( + add.messages[0].message, + Some(message::Message::AgentReasoning(_)) + )); + let Some(response_event::Type::ClientActions(actions)) = &second[0].r#type else { + panic!("expected reasoning append"); + }; + assert!(matches!( + actions.actions[0].action, + Some(client_action::Action::AppendToMessageContent(_)) + )); +} + +#[test] +fn session_activity_updates_the_same_structured_message() { + let mut translator = session_translator(); + let started = translator + .translate(AgentEvent::RuntimeActivityUpdated { + activity: RuntimeActivity { + id: "tool-1".to_owned(), + title: "Run tests".to_owned(), + status: Some(RuntimeActivityStatus::InProgress), + output: None, + }, + }) + .expect("started"); + let Some(response_event::Type::ClientActions(actions)) = &started[1].r#type else { + panic!("expected activity action"); + }; + let Some(client_action::Action::AddMessagesToTask(add)) = &actions.actions[0].action else { + panic!("expected activity message"); + }; + let message_id = add.messages[0].id.clone(); + + let completed = translator + .translate(AgentEvent::RuntimeActivityUpdated { + activity: RuntimeActivity { + id: "tool-1".to_owned(), + title: String::new(), + status: Some(RuntimeActivityStatus::Completed), + output: Some("ok".to_owned()), + }, + }) + .expect("completed"); + let Some(response_event::Type::ClientActions(actions)) = &completed[0].r#type else { + panic!("expected activity update"); + }; + let Some(client_action::Action::UpdateTaskMessage(update)) = &actions.actions[0].action else { + panic!("expected in-place update"); + }; + let message = update.message.as_ref().expect("updated message"); + assert_eq!(message.id, message_id); + assert_eq!( + runtime_activity::decode(&message.server_message_data), + Some(RuntimeActivity { + id: "tool-1".to_owned(), + title: "Run tests".to_owned(), + status: Some(RuntimeActivityStatus::Completed), + output: Some("ok".to_owned()), + }) + ); +} + +#[test] +fn usage_shape_follows_history_ownership_capability() { + let mut provider = provider_translator(); + provider + .translate(AgentEvent::UsageUpdated { + usage: Usage { + input_tokens: 250, + output_tokens: 10, + ..Usage::default() + }, + }) + .expect("provider usage"); + let provider_finished = provider + .translate(AgentEvent::TurnStopped { + reason: StopReason::Completed, + }) + .expect("provider finished"); + let Some(response_event::Type::Finished(finished)) = &provider_finished[1].r#type else { + panic!("expected provider finish"); + }; + assert_eq!(finished.token_usage[0].total_input, 250); + + let mut session = session_translator(); + session + .translate(AgentEvent::ContextUsageUpdated { + used_tokens: 25, + context_size: 100, + }) + .expect("context usage"); + let session_finished = session + .translate(AgentEvent::TurnStopped { + reason: StopReason::Completed, + }) + .expect("session finished"); + let Some(response_event::Type::Finished(finished)) = &session_finished[2].r#type else { + panic!("expected session finish"); + }; + assert_eq!(finished.token_usage[0].total_input, 0); + assert_eq!( + finished + .conversation_usage_metadata + .as_ref() + .expect("context metadata") + .context_window_usage, + 0.25 + ); +} + +#[test] +fn capabilities_reject_events_owned_by_the_other_runtime_shape() { + assert!(provider_translator() + .translate(AgentEvent::ContextUsageUpdated { + used_tokens: 1, + context_size: 2, + }) + .is_err()); + assert!(session_translator() + .translate(AgentEvent::UsageUpdated { + usage: Usage::default(), + }) + .is_err()); +} diff --git a/app/src/ai/runtime/mod.rs b/app/src/ai/runtime/mod.rs index 333c9c25..7ddf3487 100644 --- a/app/src/ai/runtime/mod.rs +++ b/app/src/ai/runtime/mod.rs @@ -1,7 +1,9 @@ +mod event_translator; mod provider; mod rig; mod rig_request; mod rig_tool; +pub(crate) use event_translator::{RuntimeResponseConfig, RuntimeResponseTranslator}; pub(crate) use provider::ProviderRuntime; pub(crate) use rig::{rig_bedrock_response_stream, rig_openai_response_stream}; diff --git a/app/src/ai/runtime/rig.rs b/app/src/ai/runtime/rig.rs index 34d4c333..5e205c7f 100644 --- a/app/src/ai/runtime/rig.rs +++ b/app/src/ai/runtime/rig.rs @@ -3,13 +3,12 @@ use std::sync::Arc; use futures::channel::oneshot; use futures::{FutureExt, StreamExt}; use galaxy_agent_core::{ - turn_control, AgentError, AgentEvent, AgentRuntime, MessageContent, MessageRole, StopReason, - ToolCall, ToolCallDecision, ToolEvent, ToolPolicy, ToolResult, TurnCommand, Usage, + turn_control, AgentError, AgentEvent, AgentRuntime, MessageContent, MessageRole, ToolCall, + ToolCallDecision, ToolEvent, ToolPolicy, ToolResult, TurnCommand, }; use galaxy_agent_rig::{OpenAICompatibleRuntime, OpenAICompatibleRuntimeConfig}; use uuid::Uuid; -use warp_multi_agent_api::response_event::stream_finished; -use warp_multi_agent_api::{self as api, ClientAction, ResponseEvent, ToolType}; +use warp_multi_agent_api::ToolType; use super::rig_request::{prepare_bedrock_rig_turn, prepare_rig_turn, PreparedRigTurn}; use super::rig_tool::action_from_tool_call; @@ -17,13 +16,10 @@ use crate::ai::agent::api::{Event, RequestParams, ResponseStream, StreamEvent}; use crate::ai::agent::AIAgentAction; use crate::ai::bedrock::client::{BedrockClient, BedrockClientConfig}; use crate::ai::bedrock::external_config::ExternalBedrockConfig; -use crate::ai::bedrock::response_translator::{ - build_add_agent_output_message, build_append_text, build_create_task, build_stream_init, - build_user_query_message, -}; +use crate::ai::bedrock::response_translator::build_add_agent_output_message; use crate::ai::openai::client::OpenAIClientConfig; -use crate::ai::openai::response_translator::{build_stream_finished, StreamUsage}; use crate::ai::provider::types::{ContentPart, ConversationMessage}; +use crate::ai::runtime::{RuntimeResponseConfig, RuntimeResponseTranslator}; use crate::server::server_api::AIApiError; pub(crate) fn rig_openai_response_stream( @@ -103,6 +99,7 @@ fn rig_response_stream( where R: AgentRuntime + Send + Sync + 'static, { + let runtime_capabilities = runtime.descriptor().capabilities.clone(); let PreparedRigTurn { task_id, needs_create_task, @@ -143,17 +140,22 @@ where }, }; - let request_id = Uuid::new_v4().to_string(); let conversation_id = conversation_id.unwrap_or_else(|| Uuid::new_v4().to_string()); - let mut initialized = false; - let mut current_text_message_id: Option = None; - let mut current_reasoning_message_id: Option = None; + let mut translator = RuntimeResponseTranslator::new(RuntimeResponseConfig { + task_id: task_id.clone(), + conversation_id, + needs_create_task, + user_query, + model_id, + max_context_tokens, + capabilities: runtime_capabilities, + empty_output_message: None, + }); let mut full_text = String::new(); let mut full_reasoning = String::new(); let mut reasoning_signature = None; let mut proposed_tools = Vec::new(); let mut assistant_history_index = None; - let mut usage = Usage::default(); loop { let next_event = agent_events.next().fuse(); @@ -176,48 +178,6 @@ where }; match event { - AgentEvent::TurnStarted { .. } => { - initialized = true; - yield Ok(StreamEvent::Response(build_stream_init(&request_id, &conversation_id))); - if needs_create_task { - yield Ok(StreamEvent::Response(build_create_task(&task_id))); - } - if let Some(user_query) = &user_query { - yield Ok(StreamEvent::Response(build_user_query_message(&task_id, user_query))); - } - } - AgentEvent::TextDelta { text } => { - full_text.push_str(&text); - if let Some(message_id) = ¤t_text_message_id { - yield Ok(StreamEvent::Response(build_append_text(&task_id, message_id, &text))); - } else { - let message_id = Uuid::new_v4().to_string(); - yield Ok(StreamEvent::Response(build_add_agent_output_message(&task_id, &message_id, &text))); - current_text_message_id = Some(message_id); - } - } - AgentEvent::ReasoningDelta { text } => { - full_reasoning.push_str(&text); - if let Some(message_id) = ¤t_reasoning_message_id { - yield Ok(StreamEvent::Response(build_append_reasoning(&task_id, message_id, &text))); - } else { - let message_id = Uuid::new_v4().to_string(); - yield Ok(StreamEvent::Response(build_add_reasoning(&task_id, &message_id, &text))); - current_reasoning_message_id = Some(message_id); - } - } - AgentEvent::ReasoningCompleted { text, signature } => { - if current_reasoning_message_id.is_none() && !text.is_empty() { - let message_id = Uuid::new_v4().to_string(); - yield Ok(StreamEvent::Response(build_add_reasoning(&task_id, &message_id, &text))); - current_reasoning_message_id = Some(message_id); - } - if !text.is_empty() { - full_reasoning = text; - } - reasoning_signature = signature; - } - AgentEvent::UsageUpdated { usage: updated } => usage = updated, AgentEvent::Tool { event: ToolEvent::Proposed { call }, } => { @@ -271,9 +231,6 @@ where } } AgentEvent::TurnStopped { reason } => { - if !initialized { - yield Ok(StreamEvent::Response(build_stream_init(&request_id, &conversation_id))); - } sync_assistant_turn( &messages_sent, &full_reasoning, @@ -282,38 +239,57 @@ where &proposed_tools, &mut assistant_history_index, ); - yield Ok(StreamEvent::Response(build_stream_finished( - map_stop_reason(reason), - StreamUsage { - input_tokens: saturating_i32(usage.input_tokens), - output_tokens: saturating_i32(usage.output_tokens), - cache_read_tokens: saturating_i32(usage.cached_input_tokens), - cache_write_tokens: saturating_i32( - usage.cache_creation_input_tokens, - ), - cost_in_cents: 0.0, - model_id, - max_context_tokens, - }, - ))); + let response_events = match translator + .translate(AgentEvent::TurnStopped { reason }) + { + Ok(response_events) => response_events, + Err(message) => { + yield Err(agent_error(AgentError::new( + galaxy_agent_core::AgentErrorKind::Protocol, + message, + ), stream_type)); + return; + } + }; + for response_event in response_events { + yield Ok(StreamEvent::Response(response_event)); + } return; } - AgentEvent::Tool { .. } => { - yield Err(agent_error(AgentError::new( - galaxy_agent_core::AgentErrorKind::Protocol, - "the provider runtime attempted to execute a tool outside Galaxy's permission boundary", - ), stream_type)); - return; - } - AgentEvent::RuntimeActivityUpdated { .. } - | AgentEvent::ContextUsageUpdated { .. } - | AgentEvent::UserInputAccepted { .. } - | AgentEvent::RuntimeNotice { .. } => { - yield Err(agent_error(AgentError::new( - galaxy_agent_core::AgentErrorKind::Protocol, - "the provider runtime emitted a session-runtime event", - ), stream_type)); - return; + event => { + match &event { + AgentEvent::TextDelta { text } => full_text.push_str(text), + AgentEvent::ReasoningDelta { text } => { + full_reasoning.push_str(text); + } + AgentEvent::ReasoningCompleted { text, signature } => { + if !text.is_empty() { + full_reasoning.clone_from(text); + } + reasoning_signature.clone_from(signature); + } + AgentEvent::TurnStarted { .. } + | AgentEvent::Tool { .. } + | AgentEvent::UsageUpdated { .. } + | AgentEvent::RuntimeActivityUpdated { .. } + | AgentEvent::ContextUsageUpdated { .. } + | AgentEvent::UserInputAccepted { .. } + | AgentEvent::RuntimeNotice { .. } + | AgentEvent::TurnStopped { .. } => {} + } + let response_events = match translator.translate(event) { + Ok(response_events) => response_events, + Err(message) => { + yield Err(agent_error(AgentError::new( + galaxy_agent_core::AgentErrorKind::Protocol, + message, + ), stream_type)); + return; + } + }; + for response_event in response_events { + yield Ok(StreamEvent::Response(response_event)); + } } } } @@ -426,77 +402,6 @@ fn build_tool_proposed( action_from_tool_call(task_id, call, skill_path_origin) } -fn build_add_reasoning(task_id: &str, message_id: &str, text: &str) -> ResponseEvent { - reasoning_action(task_id, message_id, text, false) -} - -fn build_append_reasoning(task_id: &str, message_id: &str, text: &str) -> ResponseEvent { - reasoning_action(task_id, message_id, text, true) -} - -fn reasoning_action(task_id: &str, message_id: &str, text: &str, append: bool) -> ResponseEvent { - let message = api::Message { - id: message_id.to_string(), - task_id: task_id.to_string(), - request_id: String::new(), - timestamp: None, - server_message_data: String::new(), - citations: Vec::new(), - fetched_memories: Vec::new(), - message: Some(api::message::Message::AgentReasoning( - api::message::AgentReasoning { - reasoning: text.to_string(), - finished_duration: None, - }, - )), - }; - let action = if append { - api::client_action::Action::AppendToMessageContent( - api::client_action::AppendToMessageContent { - task_id: task_id.to_string(), - message: Some(message), - mask: Some(prost_types::FieldMask { - paths: vec!["agent_reasoning.reasoning".to_string()], - }), - }, - ) - } else { - api::client_action::Action::AddMessagesToTask(api::client_action::AddMessagesToTask { - task_id: task_id.to_string(), - messages: vec![message], - }) - }; - ResponseEvent { - r#type: Some(api::response_event::Type::ClientActions( - api::response_event::ClientActions { - actions: vec![ClientAction { - action: Some(action), - }], - }, - )), - } -} - -fn map_stop_reason(reason: StopReason) -> stream_finished::Reason { - match reason { - StopReason::Completed => stream_finished::Reason::Done(stream_finished::Done {}), - StopReason::MaxTokens => { - stream_finished::Reason::MaxTokenLimit(stream_finished::ReachedMaxTokenLimit {}) - } - StopReason::ContextWindowExceeded => stream_finished::Reason::ContextWindowExceeded( - stream_finished::ContextWindowExceeded {}, - ), - StopReason::Cancelled - | StopReason::Refusal - | StopReason::ToolLoopLimit - | StopReason::Other(_) => stream_finished::Reason::Other(stream_finished::Other {}), - } -} - -fn saturating_i32(value: u64) -> i32 { - i32::try_from(value).unwrap_or(i32::MAX) -} - fn agent_error(error: AgentError, stream_type: &'static str) -> Arc { Arc::new( AIApiError::Stream { diff --git a/app/src/ai/runtime/rig_tests.rs b/app/src/ai/runtime/rig_tests.rs index b18f7089..981aeb75 100644 --- a/app/src/ai/runtime/rig_tests.rs +++ b/app/src/ai/runtime/rig_tests.rs @@ -2,69 +2,10 @@ use std::sync::{Arc, Mutex}; use ai::skills::SkillPathOrigin; use galaxy_agent_core::{ - ContentPart, MessageContent, MessageRole, StopReason, ToolCall, ToolResult, ToolResultStatus, -}; -use warp_multi_agent_api::response_event::stream_finished; - -use super::{ - append_tool_result, build_add_reasoning, build_append_reasoning, build_tool_proposed, - map_stop_reason, saturating_i32, sync_assistant_turn, + ContentPart, MessageContent, MessageRole, ToolCall, ToolResult, ToolResultStatus, }; -#[test] -fn stop_reasons_map_to_the_existing_ui_contract() { - assert!(matches!( - map_stop_reason(StopReason::Completed), - stream_finished::Reason::Done(_) - )); - assert!(matches!( - map_stop_reason(StopReason::MaxTokens), - stream_finished::Reason::MaxTokenLimit(_) - )); - assert!(matches!( - map_stop_reason(StopReason::Cancelled), - stream_finished::Reason::Other(_) - )); -} - -#[test] -fn token_counts_saturate_at_the_proto_limit() { - assert_eq!(saturating_i32(u64::MAX), i32::MAX); -} - -#[test] -fn reasoning_events_match_the_existing_ui_message_contract() { - let add = build_add_reasoning("task", "message", "think"); - let append = build_append_reasoning("task", "message", " more"); - - let Some(warp_multi_agent_api::response_event::Type::ClientActions(add)) = add.r#type else { - panic!("expected client actions"); - }; - let Some(warp_multi_agent_api::client_action::Action::AddMessagesToTask(add)) = - &add.actions[0].action - else { - panic!("expected add-message action"); - }; - assert!(matches!( - add.messages[0].message.as_ref(), - Some(warp_multi_agent_api::message::Message::AgentReasoning(reasoning)) - if reasoning.reasoning == "think" - )); - - let Some(warp_multi_agent_api::response_event::Type::ClientActions(append)) = append.r#type - else { - panic!("expected client actions"); - }; - let Some(warp_multi_agent_api::client_action::Action::AppendToMessageContent(append)) = - &append.actions[0].action - else { - panic!("expected append-message action"); - }; - assert_eq!( - append.mask.as_ref().unwrap().paths, - ["agent_reasoning.reasoning"] - ); -} +use super::{append_tool_result, build_tool_proposed, sync_assistant_turn}; #[test] fn tool_proposal_matches_the_domain_permission_contract() { diff --git a/app/src/terminal/profile_model_selector.rs b/app/src/terminal/profile_model_selector.rs index f36e3211..d0f5fd40 100644 --- a/app/src/terminal/profile_model_selector.rs +++ b/app/src/terminal/profile_model_selector.rs @@ -20,8 +20,6 @@ use instant::{Duration, Instant}; use parking_lot::FairMutex; use pathfinder_color::ColorU; use pathfinder_geometry::vector::vec2f; -use settings::Setting; - const SIDECAR_POSITION_ID: &str = "model_sidecar_panel"; use galaxy_cli::agent::Harness; @@ -55,8 +53,6 @@ use crate::cloud_object::model::generic_string_model::StringModel; use crate::context_chips::display_chip::{udi_font_size, udi_icon_size}; use crate::context_chips::spacing; use crate::menu::{Event as MenuEvent, Menu, MenuItem, MenuItemFields}; -use crate::persistence::model::AgentBackend; -use crate::settings::AISettings; use crate::settings_view::SettingsSection; use crate::terminal::input::{MenuPositioning, MenuPositioningProvider}; use crate::terminal::view::ambient_agent::AmbientAgentViewModel; @@ -677,24 +673,6 @@ impl ProfileModelSelector { self.is_locked_for_cloud_followup(app) || self.is_locked_for_non_oz_run(app) } - fn is_acp_model_managed(&self, app: &AppContext) -> bool { - if self.ambient_agent_view_model.is_some() { - return false; - } - let history = BlocklistAIHistoryModel::as_ref(app); - if let Some(conversation_id) = history.active_conversation_id(self.terminal_view_id) { - return history - .conversation(&conversation_id) - .is_some_and(|conversation| { - matches!(conversation.agent_backend(), AgentBackend::Acp(_)) - }); - } - - cfg!(unix) - && FeatureFlag::AgentClientProtocol.is_enabled() - && *AISettings::as_ref(app).acp_enabled.value() - } - /// True when a non-Oz harness is selected. fn is_third_party_harness(&self, app: &AppContext) -> bool { self.ambient_agent_view_model.as_ref().is_some_and(|m| { @@ -1635,7 +1613,6 @@ impl ProfileModelSelector { let appearance = Appearance::as_ref(app); let theme = appearance.theme(); let llm_preferences = LLMPreferences::as_ref(app); - let is_acp_model_managed = self.is_acp_model_managed(app); // Allow editing if composing an ambient agent query, or if the user has edit access // in a shared session (i.e., not a viewer, or is an executor). @@ -1658,9 +1635,7 @@ impl ProfileModelSelector { .is_agent_in_control_or_tagged_in(); drop(terminal_model); - let model_display_name = if is_acp_model_managed { - "Managed by ACP".to_owned() - } else if self.is_third_party_harness(app) { + let model_display_name = if self.is_third_party_harness(app) { self.harness_model_display_name(app) } else if is_lrc { llm_preferences @@ -1717,8 +1692,7 @@ impl ProfileModelSelector { // Only show chevron icon if the user can click to open the menu (i.e. has edit access) // and the InlineMenuHeaders feature flag is not enabled // (when enabled, clicking opens the inline model selector instead of a dropdown). - if has_edit_access && !is_acp_model_managed && !FeatureFlag::InlineMenuHeaders.is_enabled() - { + if has_edit_access && !FeatureFlag::InlineMenuHeaders.is_enabled() { let chevron_icon = Icon::ChevronDown .to_galaxyui_icon(Fill::Solid(text_color)) .finish(); @@ -1746,7 +1720,7 @@ impl ProfileModelSelector { let is_locked_for_followup = self.is_locked_for_cloud_followup(app); let is_locked_for_non_oz = self.is_locked_for_non_oz_run(app); let is_locked = is_locked_for_followup || is_locked_for_non_oz; - let can_interact = has_edit_access && !is_locked && !is_acp_model_managed; + let can_interact = has_edit_access && !is_locked; let hoverable = Hoverable::new(self.model_mouse_state.clone(), move |state| { if state.is_hovered() && can_interact { @@ -1774,9 +1748,7 @@ impl ProfileModelSelector { stack.finish() } else if state.is_hovered() { // Non-Oz runs lock silently — skip the tooltip entirely. - let tooltip_text: Option<&str> = if is_acp_model_managed { - Some("Model selection is managed by the ACP agent") - } else if is_locked_for_followup { + let tooltip_text: Option<&str> = if is_locked_for_followup { Some(MODEL_LOCKED_FOR_FOLLOWUP_TOOLTIP) } else if is_locked_for_non_oz { None @@ -1849,19 +1821,6 @@ impl TypedActionView for ProfileModelSelector { type Action = ProfileModelSelectorAction; fn handle_action(&mut self, action: &Self::Action, ctx: &mut ViewContext) { - let is_model_action = matches!( - action, - ProfileModelSelectorAction::SelectModel(_) - | ProfileModelSelectorAction::SelectAutoModel - | ProfileModelSelectorAction::SelectReasoningModel(_) - | ProfileModelSelectorAction::SelectHarnessModel { .. } - | ProfileModelSelectorAction::ToggleModelMenu - ); - if is_model_action && self.is_acp_model_managed(ctx) { - self.set_model_menu_visibility(false, ctx); - return; - } - match action { ProfileModelSelectorAction::SelectProfile(profile_id) => { AIExecutionProfilesModel::handle(ctx).update(ctx, |profiles_model, ctx| { @@ -1950,7 +1909,6 @@ impl View for ProfileModelSelector { let theme = appearance.theme(); let profiles_model = AIExecutionProfilesModel::as_ref(app); let has_multiple_profiles = profiles_model.has_multiple_profiles(); - let is_acp_model_managed = self.is_acp_model_managed(app); // Check if user is a viewer in a shared session let is_viewer = self @@ -1974,14 +1932,12 @@ impl View for ProfileModelSelector { compact_row.add_child(profile_button_with_save_position); } - if !is_acp_model_managed { - let model_button_with_save_position = SavePosition::new( - ChildView::new(&self.model_compact_button).finish(), - "profile_model_selector_model_compact_button", - ) - .finish(); - compact_row.add_child(model_button_with_save_position); - } + let model_button_with_save_position = SavePosition::new( + ChildView::new(&self.model_compact_button).finish(), + "profile_model_selector_model_compact_button", + ) + .finish(); + compact_row.add_child(model_button_with_save_position); let compact_layout = compact_row.finish(); @@ -2027,7 +1983,7 @@ impl View for ProfileModelSelector { stack.add_positioned_overlay_child(profile_menu, positioning); } - if self.is_model_menu_open && !is_acp_model_managed { + if self.is_model_menu_open { let model_menu = ChildView::new(&self.model_dropdown).finish(); let positioning = self.get_menu_positioning(app, false); stack.add_positioned_overlay_child(model_menu, positioning); @@ -2039,8 +1995,7 @@ impl View for ProfileModelSelector { // The popup overflows the viewport on wasm mobile. let is_wasm_mobile = warpui::platform::is_mobile_device(); - if !is_acp_model_managed - && !is_wasm_mobile + if !is_wasm_mobile && (is_udi_enabled || self .input_model diff --git a/crates/acp/src/agent_runtime.rs b/crates/acp/src/agent_runtime.rs index 614b2d8a..b1fbacc5 100644 --- a/crates/acp/src/agent_runtime.rs +++ b/crates/acp/src/agent_runtime.rs @@ -74,18 +74,18 @@ pub struct AcpAgentRuntime { } impl AcpAgentRuntime { + #[must_use] + pub const fn capabilities() -> RuntimeCapabilities { + RuntimeCapabilities::session_runtime() + } + #[must_use] pub fn new(manager: AcpSessionManager, config: AcpAgentRuntimeConfig) -> Self { let descriptor = RuntimeDescriptor { id: config.runtime_id.clone(), display_name: config.display_name.clone(), kind: RuntimeKind::Acp, - capabilities: RuntimeCapabilities { - model_selection: false, - session_resume: true, - steering: true, - tool_permissions: true, - }, + capabilities: Self::capabilities(), }; Self { manager, diff --git a/crates/galaxy_agent_core/src/runtime.rs b/crates/galaxy_agent_core/src/runtime.rs index 823e6cc4..eb4977e6 100644 --- a/crates/galaxy_agent_core/src/runtime.rs +++ b/crates/galaxy_agent_core/src/runtime.rs @@ -21,6 +21,48 @@ pub struct RuntimeCapabilities { pub session_resume: bool, pub steering: bool, pub tool_permissions: bool, + /// Galaxy owns and persists the message history supplied to each turn. + pub host_managed_history: bool, + /// Tool proposals cross the runtime boundary for Galaxy to approve and execute. + pub host_tool_execution: bool, + /// A failed turn can be safely replayed from the same request payload. + pub request_retries: bool, + /// Galaxy can append corrective instructions and start a follow-up turn. + pub corrective_retries: bool, + /// Transcript events can be forwarded through Galaxy shared sessions. + pub shared_session_sync: bool, +} + +impl RuntimeCapabilities { + #[must_use] + pub const fn provider() -> Self { + Self { + model_selection: true, + session_resume: false, + steering: false, + tool_permissions: false, + host_managed_history: true, + host_tool_execution: true, + request_retries: true, + corrective_retries: true, + shared_session_sync: true, + } + } + + #[must_use] + pub const fn session_runtime() -> Self { + Self { + model_selection: false, + session_resume: true, + steering: true, + tool_permissions: true, + host_managed_history: false, + host_tool_execution: false, + request_retries: false, + corrective_retries: false, + shared_session_sync: false, + } + } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/galaxy_agent_core/src/runtime_tests.rs b/crates/galaxy_agent_core/src/runtime_tests.rs index 3d4ef187..efc82813 100644 --- a/crates/galaxy_agent_core/src/runtime_tests.rs +++ b/crates/galaxy_agent_core/src/runtime_tests.rs @@ -120,3 +120,22 @@ fn turn_control_delivers_cancel_and_steering_in_order() { assert_eq!(control.receive().await.unwrap(), TurnCommand::Cancel); }); } + +#[test] +fn runtime_shapes_publish_controller_policy_as_capabilities() { + let provider = RuntimeCapabilities::provider(); + assert!(provider.host_managed_history); + assert!(provider.host_tool_execution); + assert!(provider.request_retries); + assert!(provider.corrective_retries); + assert!(provider.shared_session_sync); + assert!(!provider.steering); + + let session = RuntimeCapabilities::session_runtime(); + assert!(!session.host_managed_history); + assert!(!session.host_tool_execution); + assert!(!session.request_retries); + assert!(!session.corrective_retries); + assert!(!session.shared_session_sync); + assert!(session.steering); +} diff --git a/crates/galaxy_agent_rig/src/bedrock.rs b/crates/galaxy_agent_rig/src/bedrock.rs index 0ecb59df..96504fba 100644 --- a/crates/galaxy_agent_rig/src/bedrock.rs +++ b/crates/galaxy_agent_rig/src/bedrock.rs @@ -47,12 +47,7 @@ impl BedrockRuntime { id: format!("rig-bedrock:{resolved_model}"), display_name: format!("Rig / Bedrock / {resolved_model}"), kind: RuntimeKind::Provider, - capabilities: RuntimeCapabilities { - model_selection: true, - session_resume: false, - steering: false, - tool_permissions: false, - }, + capabilities: RuntimeCapabilities::provider(), }; Ok(Self { diff --git a/crates/galaxy_agent_rig/src/openai_compatible.rs b/crates/galaxy_agent_rig/src/openai_compatible.rs index 963f6d63..7f06854d 100644 --- a/crates/galaxy_agent_rig/src/openai_compatible.rs +++ b/crates/galaxy_agent_rig/src/openai_compatible.rs @@ -31,12 +31,7 @@ impl OpenAICompatibleRuntime { id: format!("rig-openai-compatible:{}", config.model), display_name: format!("Rig / {}", config.model), kind: RuntimeKind::Provider, - capabilities: RuntimeCapabilities { - model_selection: true, - session_resume: false, - steering: false, - tool_permissions: false, - }, + capabilities: RuntimeCapabilities::provider(), }; Self { config, descriptor } } diff --git a/crates/galaxyui_core/src/core/app.rs b/crates/galaxyui_core/src/core/app.rs index 792c6844..f3f03498 100644 --- a/crates/galaxyui_core/src/core/app.rs +++ b/crates/galaxyui_core/src/core/app.rs @@ -4909,6 +4909,17 @@ impl GetSingletonModelHandle for AppContext { } impl AppContext { + /// Returns a singleton model when it has already been registered. + /// + /// This is useful for infrastructure models that can also be constructed + /// in isolation by tests before the full application singleton graph is + /// available. + pub fn try_get_singleton_model_as_ref(&self) -> Option<&T> { + self.singleton_models + .get(&std::any::TypeId::of::())? + .downcast_ref(self) + } + pub(super) fn get_singleton_model_as_ref(&self) -> &T { match self.singleton_models.get(&std::any::TypeId::of::()) { Some(model_handle) => model_handle