From 4c7270db8dfd182f46f0451451a624cdb36b7970 Mon Sep 17 00:00:00 2001 From: Ryan Ward Date: Tue, 4 Aug 2026 02:15:18 -0500 Subject: [PATCH] feat: introduce Rig agent runtime migration --- Cargo.lock | 131 +++++- Cargo.toml | 3 + app/Cargo.toml | 2 + app/src/ai/agent/api/impl.rs | 16 +- .../blocklist/controller/response_stream.rs | 43 +- app/src/ai/crosscheck/reviewer.rs | 2 + app/src/ai/llms.rs | 6 + app/src/ai/mod.rs | 1 + app/src/ai/openai/client.rs | 2 + app/src/ai/openai/response_translator.rs | 21 +- app/src/ai/openai/translator.rs | 97 ++-- app/src/ai/provider/types.rs | 137 +----- app/src/ai/runtime/mod.rs | 5 + app/src/ai/runtime/provider.rs | 27 ++ app/src/ai/runtime/rig.rs | 319 +++++++++++++ app/src/ai/runtime/rig_tests.rs | 59 +++ app/src/bin/oss.rs | 2 +- app/src/settings/ai.rs | 49 +- app/src/settings/ai_tests.rs | 31 ++ crates/galaxy_agent_core/Cargo.toml | 13 + crates/galaxy_agent_core/src/lib.rs | 11 + crates/galaxy_agent_core/src/runtime.rs | 144 ++++++ crates/galaxy_agent_core/src/runtime_tests.rs | 120 +++++ crates/galaxy_agent_core/src/types.rs | 227 +++++++++ crates/galaxy_agent_core/src/types_tests.rs | 38 ++ crates/galaxy_agent_rig/Cargo.toml | 20 + crates/galaxy_agent_rig/src/lib.rs | 5 + .../galaxy_agent_rig/src/openai_compatible.rs | 432 ++++++++++++++++++ .../src/openai_compatible_tests.rs | 201 ++++++++ .../galaxy_core/src/channel/channel_tests.rs | 10 + crates/galaxy_core/src/channel/config.rs | 21 + .../galaxy_core/src/channel/config_tests.rs | 12 + crates/galaxy_core/src/channel/mod.rs | 16 + crates/galaxy_core/src/channel/state.rs | 2 +- .../galaxy_server_client/src/auth/session.rs | 8 + crates/warp_tui/src/bin/oss.rs | 2 +- plans/galaxy-local-first-rig.md | 393 ++++++++++++++++ plans/galaxy-refactor.md | 5 + plans/ui-flow-inventory.md | 129 ++++++ 39 files changed, 2551 insertions(+), 211 deletions(-) create mode 100644 app/src/ai/runtime/mod.rs create mode 100644 app/src/ai/runtime/provider.rs create mode 100644 app/src/ai/runtime/rig.rs create mode 100644 app/src/ai/runtime/rig_tests.rs create mode 100644 crates/galaxy_agent_core/Cargo.toml create mode 100644 crates/galaxy_agent_core/src/lib.rs create mode 100644 crates/galaxy_agent_core/src/runtime.rs create mode 100644 crates/galaxy_agent_core/src/runtime_tests.rs create mode 100644 crates/galaxy_agent_core/src/types.rs create mode 100644 crates/galaxy_agent_core/src/types_tests.rs create mode 100644 crates/galaxy_agent_rig/Cargo.toml create mode 100644 crates/galaxy_agent_rig/src/lib.rs create mode 100644 crates/galaxy_agent_rig/src/openai_compatible.rs create mode 100644 crates/galaxy_agent_rig/src/openai_compatible_tests.rs create mode 100644 crates/galaxy_core/src/channel/config_tests.rs create mode 100644 plans/galaxy-local-first-rig.md create mode 100644 plans/ui-flow-inventory.md diff --git a/Cargo.lock b/Cargo.lock index ea8fc5f7..931bb13e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1025,6 +1025,12 @@ version = "0.7.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" +[[package]] +name = "as-any" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0f477b951e452a0b6b4a10b53ccd569042d1d01729b519e02074a9c0958a063" + [[package]] name = "as-raw-xcb-connection" version = "1.0.1" @@ -1327,7 +1333,7 @@ dependencies = [ "rustls-pki-types", "tokio", "tokio-rustls", - "tungstenite", + "tungstenite 0.24.0", ] [[package]] @@ -3373,6 +3379,15 @@ dependencies = [ "unicode-segmentation", ] +[[package]] +name = "convert_case" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "affbf0190ed2caf063e3def54ff444b449371d55c58e513a95ab98eca50adb49" +dependencies = [ + "unicode-segmentation", +] + [[package]] name = "core-foundation" version = "0.9.4" @@ -5653,6 +5668,8 @@ dependencies = [ "futures-util", "fuzzy_match", "galaxy_acp", + "galaxy_agent_core", + "galaxy_agent_rig", "galaxy_cli", "galaxy_completer", "galaxy_core", @@ -5863,6 +5880,32 @@ dependencies = [ "thiserror 2.0.19", ] +[[package]] +name = "galaxy_agent_core" +version = "0.1.0" +dependencies = [ + "async-channel", + "async-trait", + "futures", + "serde", + "serde_json", +] + +[[package]] +name = "galaxy_agent_rig" +version = "0.1.0" +dependencies = [ + "async-stream", + "async-trait", + "bytes", + "futures", + "galaxy_agent_core", + "rig-core", + "serde_json", + "tokio", + "uuid", +] + [[package]] name = "galaxy_cli" version = "0.0.0" @@ -7161,7 +7204,7 @@ dependencies = [ "serde", "serde_json", "thiserror 1.0.69", - "tungstenite", + "tungstenite 0.24.0", "ws_stream_wasm", ] @@ -12735,6 +12778,55 @@ dependencies = [ "bytemuck", ] +[[package]] +name = "rig-core" +version = "0.40.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d8731dd5532b3a12ce1613af73073fb2051ef750f50c504778c21d55ae933cac" +dependencies = [ + "as-any", + "async-stream", + "base64 0.22.1", + "bytes", + "eventsource-stream", + "fastrand 2.5.0", + "futures", + "futures-timer", + "glob", + "http 1.5.0", + "indexmap 2.14.0", + "mime", + "mime_guess", + "ordered-float 5.3.0", + "pin-project-lite", + "reqwest 0.13.4", + "rig-derive", + "schemars 1.2.2", + "serde", + "serde_json", + "thiserror 2.0.19", + "tokio", + "tokio-tungstenite", + "tracing", + "tracing-futures", + "url", +] + +[[package]] +name = "rig-derive" +version = "0.40.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3e98dde7a4e59e083e7396126ee4c83498c5bff605d126654e67815fa230a78" +dependencies = [ + "convert_case 0.11.0", + "indoc", + "proc-macro-crate 3.5.0", + "proc-macro2", + "quote", + "serde_json", + "syn 2.0.119", +] + [[package]] name = "ring" version = "0.17.14" @@ -15220,6 +15312,22 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d25a406cddcc431a75d3d9afc6a7c0f7428d4891dd973e4d54c56b46127bf857" +dependencies = [ + "futures-util", + "log", + "rustls", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tungstenite 0.28.0", + "webpki-roots 0.26.11", +] + [[package]] name = "tokio-util" version = "0.7.19" @@ -15583,6 +15691,25 @@ dependencies = [ "utf-8", ] +[[package]] +name = "tungstenite" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8628dcc84e5a09eb3d8423d6cb682965dea9133204e8fb3efee74c2a0c259442" +dependencies = [ + "bytes", + "data-encoding", + "http 1.5.0", + "httparse", + "log", + "rand 0.9.5", + "rustls", + "rustls-pki-types", + "sha1", + "thiserror 2.0.19", + "utf-8", +] + [[package]] name = "twox-hash" version = "2.1.3" diff --git a/Cargo.toml b/Cargo.toml index 039317de..e719cc72 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,6 +29,8 @@ publish = false [workspace.dependencies] # Local workspace crates. This lets us reference them in other crates without specifying a path. galaxy_acp = { path = "crates/acp" } +galaxy_agent_core = { path = "crates/galaxy_agent_core" } +galaxy_agent_rig = { path = "crates/galaxy_agent_rig" } ai = { path = "crates/ai" } app-installation-detection = { path = "crates/app-installation-detection" } asset_cache = { path = "crates/asset_cache" } @@ -257,6 +259,7 @@ reqwest = { version = "0.13", features = [ "stream", ] } reqwest-eventsource = { package = "aha-reqwest-eventsource", version = "0.1" } +rig-core = "=0.40.0" resvg = "0.47.0" rust-embed = { version = "8.7.0", features = ["include-exclude"] } rustc-hash = "2.1.1" diff --git a/app/Cargo.toml b/app/Cargo.toml index e6014c80..91091d99 100644 --- a/app/Cargo.toml +++ b/app/Cargo.toml @@ -234,6 +234,8 @@ warp_assets.workspace = true warp_channel_config.workspace = true galaxy_completer.workspace = true galaxy_core.workspace = true +galaxy_agent_core.workspace = true +galaxy_agent_rig.workspace = true galaxy_editor.workspace = true galaxy_graphql.workspace = true galaxy_js = { workspace = true, optional = true } diff --git a/app/src/ai/agent/api/impl.rs b/app/src/ai/agent/api/impl.rs index c0a8ef8e..3cee997b 100644 --- a/app/src/ai/agent/api/impl.rs +++ b/app/src/ai/agent/api/impl.rs @@ -25,7 +25,7 @@ pub async fn generate_multi_agent_output( let supported_cli_agent_tools = supported_tools_override.unwrap_or_else(|| get_supported_cli_agent_tools(¶ms)); let mut logging_metadata = HashMap::new(); - if let Some(metadata) = params.metadata { + if let Some(ref metadata) = params.metadata { logging_metadata.insert( "is_autodetected_user_query".to_owned(), prost_types::Value { @@ -56,6 +56,12 @@ pub async fn generate_multi_agent_output( redaction::redact_inputs(&mut params.input); } + let rig_params = matches!( + &provider_config, + ProviderConfig::OpenAI(config) if config.use_rig + ) + .then(|| params.clone()); + let mut request = api::Request { task_context: Some(api::request::TaskContext { tasks: params.tasks, @@ -138,6 +144,14 @@ pub async fn generate_multi_agent_output( }; match provider_config { + ProviderConfig::OpenAI(config) if config.use_rig => { + Ok(crate::ai::runtime::rig_openai_response_stream( + config, + rig_params.expect("Rig request parameters should be retained for a Rig model"), + &mut request, + cancellation_rx, + )) + } ProviderConfig::OpenAI(config) => { let translator_request = openai_translator::TranslatorRequest { config, diff --git a/app/src/ai/blocklist/controller/response_stream.rs b/app/src/ai/blocklist/controller/response_stream.rs index 0c0cfe8f..7b2b8c5e 100644 --- a/app/src/ai/blocklist/controller/response_stream.rs +++ b/app/src/ai/blocklist/controller/response_stream.rs @@ -24,7 +24,7 @@ use crate::ai::acp::{ resolve_acp_permissions, validate_acp_dispatch, validate_acp_launch_identity, AcpRuntimeModel, AcpSessionHandleSlot, AcpSessionMetadata, AcpSteeringRequest, GalaxyMcpTarget, }; -use crate::ai::agent::api::{self, generate_multi_agent_output, ConvertToAPITypeError}; +use crate::ai::agent::api::{self, ConvertToAPITypeError}; use crate::ai::agent::conversation::AIConversationId; #[cfg(not(target_family = "wasm"))] use crate::ai::agent::AIAgentInput; @@ -35,6 +35,7 @@ use crate::ai::blocklist::BlocklistAIPermissions; use crate::ai::llms::{LLMId, LLMPreferences}; use crate::ai::openai::client::OpenAIClientConfig; use crate::ai::provider::ProviderConfig; +use crate::ai::runtime::ProviderRuntime; use crate::network::NetworkStatus; #[cfg(not(target_family = "wasm"))] use crate::pane_group::PaneGroup; @@ -233,6 +234,8 @@ impl ResponseStream { model: Some(model_id.to_string()), max_input_tokens: client_config.max_input_tokens, max_output_tokens: client_config.max_output_tokens, + use_rig: client_config.use_rig, + supports_system_messages: client_config.supports_system_messages, }); } } @@ -400,15 +403,16 @@ impl ResponseStream { cancellation_rx: oneshot::Receiver<()>, ctx: &mut ModelContext, ) { - let _ = - ctx.spawn( - async move { - generate_multi_agent_output(provider_config, params, cancellation_rx).await - }, - move |me, stream, ctx| { - me.handle_response_stream_result(request_id, stream, ctx); - }, - ); + let _ = ctx.spawn( + async move { + ProviderRuntime::new(provider_config) + .start_turn(params, cancellation_rx) + .await + }, + move |me, stream, ctx| { + me.handle_response_stream_result(request_id, stream, ctx); + }, + ); } pub fn new( @@ -604,15 +608,16 @@ impl ResponseStream { self.current_request_id = Some(request_id); let params = self.params.clone(); let provider_config = Self::resolve_provider_config(params.model.as_str(), ctx); - let _ = - ctx.spawn( - async move { - generate_multi_agent_output(provider_config, params, cancellation_rx).await - }, - move |me, stream, ctx| { - me.handle_response_stream_result(request_id, stream, ctx); - }, - ); + let _ = ctx.spawn( + async move { + ProviderRuntime::new(provider_config) + .start_turn(params, cancellation_rx) + .await + }, + move |me, stream, ctx| { + me.handle_response_stream_result(request_id, stream, ctx); + }, + ); } fn should_fallback_to_coding_model( diff --git a/app/src/ai/crosscheck/reviewer.rs b/app/src/ai/crosscheck/reviewer.rs index 52a2318f..27845d69 100644 --- a/app/src/ai/crosscheck/reviewer.rs +++ b/app/src/ai/crosscheck/reviewer.rs @@ -168,6 +168,8 @@ impl CrosscheckReviewer { model: Some(model_id.to_string()), max_input_tokens: client_config.max_input_tokens, max_output_tokens: Some(REVIEWER_MAX_OUTPUT_TOKENS), + use_rig: client_config.use_rig, + supports_system_messages: client_config.supports_system_messages, }); } } diff --git a/app/src/ai/llms.rs b/app/src/ai/llms.rs index 305bc004..950eff9e 100644 --- a/app/src/ai/llms.rs +++ b/app/src/ai/llms.rs @@ -999,6 +999,8 @@ impl LLMPreferences { model: None, // filled per-request from model_id max_input_tokens: Some(openai_model_context_size(model)), max_output_tokens: model.max_output_tokens, + use_rig: model.use_rig, + supports_system_messages: model.supports_system_messages(), }; self.openai_provider_routing .insert(model.model_id.clone(), client_config); @@ -2115,6 +2117,8 @@ async fn fetch_from_litellm_model_info( max_input_tokens, max_output_tokens, provider, + use_rig: false, + supports_system_messages: model_info["supports_system_messages"].as_bool(), }) }) .collect(); @@ -2236,6 +2240,8 @@ async fn fetch_from_openai_models( max_input_tokens, max_output_tokens, provider, + use_rig: false, + supports_system_messages: m["supports_system_messages"].as_bool(), }) }) .collect(); diff --git a/app/src/ai/mod.rs b/app/src/ai/mod.rs index 8307cfa3..93c68252 100644 --- a/app/src/ai/mod.rs +++ b/app/src/ai/mod.rs @@ -55,6 +55,7 @@ pub(crate) mod remote_agent_context; pub(crate) mod remote_context_files; pub mod request_usage_model; pub(crate) mod restored_conversations; +pub(crate) mod runtime; pub(crate) mod skills; pub(crate) mod voice; pub use agent_tips::*; diff --git a/app/src/ai/openai/client.rs b/app/src/ai/openai/client.rs index 125d39b9..2828faa1 100644 --- a/app/src/ai/openai/client.rs +++ b/app/src/ai/openai/client.rs @@ -11,6 +11,8 @@ pub struct OpenAIClientConfig { pub model: Option, pub max_input_tokens: Option, pub max_output_tokens: Option, + pub use_rig: bool, + pub supports_system_messages: bool, } pub struct OpenAIClient { diff --git a/app/src/ai/openai/response_translator.rs b/app/src/ai/openai/response_translator.rs index b5b01f36..4f110276 100644 --- a/app/src/ai/openai/response_translator.rs +++ b/app/src/ai/openai/response_translator.rs @@ -33,14 +33,14 @@ pub struct OpenAIStreamContext { pub tool_result_archive: Vec, } -struct StreamUsage { - input_tokens: i32, - output_tokens: i32, - cache_read_tokens: i32, - cache_write_tokens: i32, - cost_in_cents: f32, - model_id: String, - max_context_tokens: Option, +pub(crate) struct StreamUsage { + pub(crate) input_tokens: i32, + pub(crate) output_tokens: i32, + pub(crate) cache_read_tokens: i32, + pub(crate) cache_write_tokens: i32, + pub(crate) cost_in_cents: f32, + pub(crate) model_id: String, + pub(crate) max_context_tokens: Option, } pub fn openai_stream_to_response_events( @@ -533,7 +533,10 @@ fn build_tool_call_message( ) } -fn build_stream_finished(reason: stream_finished::Reason, usage: StreamUsage) -> ResponseEvent { +pub(crate) fn build_stream_finished( + reason: stream_finished::Reason, + usage: StreamUsage, +) -> ResponseEvent { let StreamUsage { input_tokens, output_tokens, diff --git a/app/src/ai/openai/translator.rs b/app/src/ai/openai/translator.rs index 638f023c..03bcbe9a 100644 --- a/app/src/ai/openai/translator.rs +++ b/app/src/ai/openai/translator.rs @@ -24,27 +24,32 @@ pub struct TranslatorRequest { pub global_rules: Vec<(String, String)>, } -pub async fn execute( - params: TranslatorRequest, - request: &mut api::Request, -) -> Result { - let client = OpenAIClient::from_config(params.config.clone()); +pub(crate) struct PreparedTurn { + pub(crate) task_id: String, + pub(crate) needs_create_task: bool, + pub(crate) user_query: Option, + pub(crate) messages: Vec, + pub(crate) system_prompt: Option, + pub(crate) tools: Vec, + pub(crate) model_id: String, + pub(crate) persistent_message_count: usize, +} - let task_id = params.root_task_id.unwrap_or_else(|| { +pub(crate) fn prepare_turn(params: &TranslatorRequest, request: &mut api::Request) -> PreparedTurn { + let task_id = params.root_task_id.clone().unwrap_or_else(|| { request .task_context .as_ref() .and_then(|tc| tc.tasks.first()) - .map(|t| t.id.clone()) + .map(|task| task.id.clone()) .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()) }); let needs_create_task = request .task_context .as_ref() - .map(|tc| tc.tasks.is_empty()) + .map(|task_context| task_context.tasks.is_empty()) .unwrap_or(true); - let model_id = if params.model_id.is_empty() || params.model_id == "auto" { params .config @@ -52,7 +57,6 @@ pub async fn execute( .clone() .unwrap_or_else(|| "anthropic/claude-sonnet-4-6".to_string()) } else { - // If a model override is configured in settings, use it params .config .model @@ -60,25 +64,17 @@ pub async fn execute( .unwrap_or_else(|| params.model_id.clone()) }; - log::info!( - "[openai] Translator: model={model_id}, task_id={task_id}, needs_create_task={needs_create_task}" - ); - request_translator::inject_input_messages_into_task(request); - let new_input_messages = request_translator::extract_new_input_messages(request); - let new_input_count = new_input_messages.len(); - + let persistent_message_count = params.message_history.len() + new_input_messages.len(); let mut messages = Vec::new(); - // Prepend progressive summary as first message pair if present - if let Some(ref summary) = params.progressive_summary { + if let Some(summary) = ¶ms.progressive_summary { messages.push(ConversationMessage { role: MessageRole::User, content: MessageContent::Text(format!( - "\n{}\n\n\n\ - The above summarizes earlier conversation history. The detailed messages below are the most recent exchanges.", - summary + "\n{summary}\n\n\n\ + The above summarizes earlier conversation history. The detailed messages below are the most recent exchanges." )), }); messages.push(ConversationMessage { @@ -90,26 +86,44 @@ pub async fn execute( }); } - let history_len = params.message_history.len(); - messages.extend(params.message_history); - - if !new_input_messages.is_empty() { - log::info!( - "[openai] Appending {} new input messages to history of {}", - new_input_messages.len(), - history_len - ); - messages.extend(new_input_messages); - } - + messages.extend(params.message_history.clone()); + messages.extend(new_input_messages); for message in &mut messages { message.truncate_tool_results_for_provider_request(); } - sanitize_messages_for_openai(&mut messages); - let system_prompt = request_translator::extract_system_prompt(request, ¶ms.global_rules); - let tools = request_translator::extract_tools(request); + PreparedTurn { + task_id, + needs_create_task, + user_query: request_translator::extract_user_query_text(request), + messages, + system_prompt: request_translator::extract_system_prompt(request, ¶ms.global_rules), + tools: request_translator::extract_tools(request), + model_id, + persistent_message_count, + } +} + +pub async fn execute( + params: TranslatorRequest, + request: &mut api::Request, +) -> Result { + let client = OpenAIClient::from_config(params.config.clone()); + let PreparedTurn { + task_id, + needs_create_task, + user_query, + mut messages, + system_prompt, + tools, + model_id, + persistent_message_count, + } = prepare_turn(¶ms, request); + + log::info!( + "[openai] Translator: model={model_id}, task_id={task_id}, needs_create_task={needs_create_task}" + ); log::info!( "[openai] Sending {} messages, system_prompt={}, tools={}", @@ -118,8 +132,6 @@ pub async fn execute( tools.len() ); - let user_query_text = request_translator::extract_user_query_text(request); - let max_output_tokens = params .config .max_output_tokens @@ -139,9 +151,8 @@ pub async fn execute( // Store the message history for the controller if let Ok(mut sent) = params.messages_sent.lock() { - let persistent_count = history_len + new_input_count; - if persistent_count > 0 && messages.len() >= persistent_count { - *sent = messages.split_off(messages.len() - persistent_count); + if persistent_message_count > 0 && messages.len() >= persistent_message_count { + *sent = messages.split_off(messages.len() - persistent_message_count); } else { *sent = messages; } @@ -152,7 +163,7 @@ pub async fn execute( OpenAIStreamContext { task_id, needs_create_task, - user_query: user_query_text, + user_query, messages_sent: params.messages_sent.clone(), model_id, max_context_tokens: params.config.max_input_tokens, diff --git a/app/src/ai/provider/types.rs b/app/src/ai/provider/types.rs index dbb74b5c..69df765d 100644 --- a/app/src/ai/provider/types.rs +++ b/app/src/ai/provider/types.rs @@ -1,131 +1,6 @@ -use serde_json::Value as JsonValue; - -pub const MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST: usize = 64_000; - -#[derive(Clone, Debug)] -pub struct ConversationMessage { - pub role: MessageRole, - pub content: MessageContent, -} - -impl ConversationMessage { - pub fn truncate_tool_results_for_provider_request(&mut self) { - truncate_tool_results_in_content(&mut self.content); - } -} - -#[derive(Clone, Debug, PartialEq)] -pub enum MessageRole { - User, - Assistant, -} - -#[derive(Clone, Debug)] -pub enum MessageContent { - Text(String), - ToolUse { - tool_use_id: String, - name: String, - input: JsonValue, - }, - ToolResult { - tool_use_id: String, - content: String, - is_error: bool, - }, - MultiPart(Vec), -} - -#[derive(Clone, Debug)] -pub enum ContentPart { - Text(String), - Image { - data: Vec, - mime_type: String, - }, - ToolUse { - tool_use_id: String, - name: String, - input: JsonValue, - }, - ToolResult { - tool_use_id: String, - content: String, - is_error: bool, - }, -} - -#[derive(Clone, Debug)] -pub struct ToolDefinition { - pub name: String, - pub description: String, - pub input_schema: JsonValue, -} - -fn truncate_tool_results_in_content(content: &mut MessageContent) { - match content { - MessageContent::Text(_) | MessageContent::ToolUse { .. } => {} - MessageContent::ToolResult { content, .. } => truncate_tool_result_text(content), - MessageContent::MultiPart(parts) => { - for part in parts { - if let ContentPart::ToolResult { content, .. } = part { - truncate_tool_result_text(content); - } - } - } - } -} - -fn truncate_tool_result_text(content: &mut String) { - let char_count = content.chars().count(); - if char_count <= MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST { - return; - } - - let omitted_chars = char_count.saturating_sub(MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST); - let marker = format!("\n... [tool result truncated; omitted {omitted_chars} chars] ...\n"); - let marker_chars = marker.chars().count(); - let retained_chars = MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST.saturating_sub(marker_chars); - let head_chars = retained_chars / 2; - let tail_chars = retained_chars.saturating_sub(head_chars); - let head: String = content.chars().take(head_chars).collect(); - let tail: String = content - .chars() - .rev() - .take(tail_chars) - .collect::() - .chars() - .rev() - .collect(); - *content = format!("{head}{marker}{tail}"); -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn truncates_large_tool_results_for_provider_request() { - let prefix = "start:"; - let suffix = ":end"; - let middle = "x".repeat(MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST + 1_000); - let mut message = ConversationMessage { - role: MessageRole::User, - content: MessageContent::ToolResult { - tool_use_id: "toolu_1".to_string(), - content: format!("{prefix}{middle}{suffix}"), - is_error: false, - }, - }; - - message.truncate_tool_results_for_provider_request(); - - let MessageContent::ToolResult { content, .. } = message.content else { - panic!("expected tool result"); - }; - assert!(content.len() <= MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST + 128); - assert!(content.starts_with(prefix)); - assert!(content.ends_with(suffix)); - assert!(content.contains("tool result truncated")); - } -} +// Keep this module as a compatibility import path while provider-neutral message +// types move out of the application crate. +pub use galaxy_agent_core::{ + ContentPart, ConversationMessage, MessageContent, MessageRole, ToolDefinition, + MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST, +}; diff --git a/app/src/ai/runtime/mod.rs b/app/src/ai/runtime/mod.rs new file mode 100644 index 00000000..ebac1ef1 --- /dev/null +++ b/app/src/ai/runtime/mod.rs @@ -0,0 +1,5 @@ +mod provider; +mod rig; + +pub(crate) use provider::ProviderRuntime; +pub(crate) use rig::rig_openai_response_stream; diff --git a/app/src/ai/runtime/provider.rs b/app/src/ai/runtime/provider.rs new file mode 100644 index 00000000..77e569a6 --- /dev/null +++ b/app/src/ai/runtime/provider.rs @@ -0,0 +1,27 @@ +use futures::channel::oneshot; + +use crate::ai::agent::api::{self, ConvertToAPITypeError}; +use crate::ai::provider::ProviderConfig; + +/// Application-facing provider runtime dispatcher. +/// +/// OpenAI-compatible models can opt into the provider-neutral Rig runtime; +/// other models continue through their current translators while migration is +/// in progress. Both paths preserve the existing UI response stream contract. +pub(crate) struct ProviderRuntime { + provider_config: ProviderConfig, +} + +impl ProviderRuntime { + pub(crate) fn new(provider_config: ProviderConfig) -> Self { + Self { provider_config } + } + + pub(crate) async fn start_turn( + self, + params: api::RequestParams, + cancellation_rx: oneshot::Receiver<()>, + ) -> Result { + api::generate_multi_agent_output(self.provider_config, params, cancellation_rx).await + } +} diff --git a/app/src/ai/runtime/rig.rs b/app/src/ai/runtime/rig.rs new file mode 100644 index 00000000..ba0d21f4 --- /dev/null +++ b/app/src/ai/runtime/rig.rs @@ -0,0 +1,319 @@ +use std::sync::Arc; + +use futures::channel::oneshot; +use futures::{FutureExt, StreamExt}; +use galaxy_agent_core::{ + turn_control, AgentError, AgentEvent, AgentRuntime, MessageContent, MessageRole, StopReason, + TurnCommand, TurnRequest, Usage, +}; +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}; + +use crate::ai::agent::api::{Event, RequestParams, ResponseStream}; +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::client::OpenAIClientConfig; +use crate::ai::openai::response_translator::{build_stream_finished, StreamUsage}; +use crate::ai::openai::translator::{prepare_turn, PreparedTurn, TranslatorRequest}; +use crate::ai::provider::types::ConversationMessage; +use crate::server::server_api::AIApiError; + +pub(crate) fn rig_openai_response_stream( + config: OpenAIClientConfig, + params: RequestParams, + request: &mut api::Request, + cancellation_rx: oneshot::Receiver<()>, +) -> ResponseStream { + let translator_request = TranslatorRequest { + config: config.clone(), + model_id: params.model.as_str().to_string(), + root_task_id: params.root_task_id, + message_history: params.bedrock_message_history, + tool_result_archive: params.bedrock_tool_result_archive, + progressive_summary: params.bedrock_progressive_summary, + messages_sent: params.bedrock_messages_sent, + global_rules: params.global_rules, + }; + let PreparedTurn { + task_id, + needs_create_task, + user_query, + messages, + system_prompt, + tools: _, + model_id, + persistent_message_count, + } = prepare_turn(&translator_request, request); + + store_messages_sent( + &translator_request.messages_sent, + &messages, + persistent_message_count, + ); + + let conversation_id = request + .metadata + .as_ref() + .map(|metadata| metadata.conversation_id.clone()) + .filter(|id| !id.is_empty()); + let mut turn_request = TurnRequest::new(model_id.clone(), messages); + turn_request.conversation_id = conversation_id.clone(); + turn_request.system_prompt = system_prompt; + // Phase 2 deliberately validates the model streaming seam. Galaxy tool + // execution moves behind AgentRuntime in Phase 3; exposing the legacy tool + // list here would split ownership across both systems. + turn_request.tools = Vec::new(); + turn_request.max_output_tokens = config.max_output_tokens.map(u64::from); + + let runtime = OpenAICompatibleRuntime::new(OpenAICompatibleRuntimeConfig { + base_url: config.base_url, + api_key: config.api_key, + model: model_id.clone(), + max_output_tokens: config.max_output_tokens.map(u64::from), + supports_system_messages: config.supports_system_messages, + }); + let messages_sent = translator_request.messages_sent; + let max_context_tokens = config.max_input_tokens; + let stream = async_stream::stream! { + let (control_sender, control) = turn_control(); + let start_future = runtime.start_turn(turn_request, control).fuse(); + let cancel_future = cancellation_rx.fuse(); + futures::pin_mut!(start_future, cancel_future); + + let mut agent_events = futures::select_biased! { + _ = cancel_future => { + let _ = control_sender.try_send(TurnCommand::Cancel); + match start_future.await { + Ok(stream) => stream, + Err(error) => { + yield Err(agent_error(error)); + return; + } + } + } + result = start_future => match result { + Ok(stream) => stream, + Err(error) => { + yield Err(agent_error(error)); + return; + } + }, + }; + + 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 full_text = String::new(); + let mut usage = Usage::default(); + + loop { + let next_event = agent_events.next().fuse(); + futures::pin_mut!(next_event); + futures::select_biased! { + _ = cancel_future => { + let _ = control_sender.try_send(TurnCommand::Cancel); + } + event = next_event => { + let Some(event) = event else { + yield Err(Arc::new(AIApiError::UnexpectedEof)); + return; + }; + let event = match event { + Ok(event) => event, + Err(error) => { + yield Err(agent_error(error)); + return; + } + }; + + match event { + AgentEvent::TurnStarted { .. } => { + initialized = true; + yield Ok(build_stream_init(&request_id, &conversation_id)); + if needs_create_task { + yield Ok(build_create_task(&task_id)); + } + if let Some(user_query) = &user_query { + yield Ok(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(build_append_text(&task_id, message_id, &text)); + } else { + let message_id = Uuid::new_v4().to_string(); + yield Ok(build_add_agent_output_message(&task_id, &message_id, &text)); + current_text_message_id = Some(message_id); + } + } + AgentEvent::ReasoningDelta { text } => { + if let Some(message_id) = ¤t_reasoning_message_id { + yield Ok(build_append_reasoning(&task_id, message_id, &text)); + } else { + let message_id = Uuid::new_v4().to_string(); + yield Ok(build_add_reasoning(&task_id, &message_id, &text)); + current_reasoning_message_id = Some(message_id); + } + } + AgentEvent::UsageUpdated { usage: updated } => usage = updated, + AgentEvent::TurnStopped { reason } => { + if !initialized { + yield Ok(build_stream_init(&request_id, &conversation_id)); + } + store_assistant_text(&messages_sent, full_text); + yield Ok(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, + }, + )); + return; + } + AgentEvent::ToolProposed { .. } + | AgentEvent::PermissionRequested { .. } + | AgentEvent::ToolStarted { .. } + | AgentEvent::ToolCompleted { .. } => { + yield Err(agent_error(AgentError::new( + galaxy_agent_core::AgentErrorKind::Protocol, + "the Phase 2 Rig runtime emitted a tool event while tools are disabled", + ))); + return; + } + } + } + } + } + }; + + Box::pin(stream) +} + +fn store_messages_sent( + messages_sent: &std::sync::Arc>>, + messages: &[ConversationMessage], + persistent_message_count: usize, +) { + let Ok(mut sent) = messages_sent.lock() else { + return; + }; + if persistent_message_count > 0 && messages.len() >= persistent_message_count { + *sent = messages[messages.len() - persistent_message_count..].to_vec(); + } else { + *sent = messages.to_vec(); + } +} + +fn store_assistant_text( + messages_sent: &std::sync::Arc>>, + text: String, +) { + if text.is_empty() { + return; + } + if let Ok(mut sent) = messages_sent.lock() { + sent.push(ConversationMessage { + role: MessageRole::Assistant, + content: MessageContent::Text(text), + }); + } +} + +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) -> Arc { + Arc::new( + AIApiError::Stream { + stream_type: "rig_openai_compatible", + source: anyhow::anyhow!(error), + } + .into_quota_limit_if_provider_budget_exhausted(), + ) +} + +#[cfg(test)] +#[path = "rig_tests.rs"] +mod tests; diff --git a/app/src/ai/runtime/rig_tests.rs b/app/src/ai/runtime/rig_tests.rs new file mode 100644 index 00000000..fc97d5eb --- /dev/null +++ b/app/src/ai/runtime/rig_tests.rs @@ -0,0 +1,59 @@ +use galaxy_agent_core::StopReason; +use warp_multi_agent_api::response_event::stream_finished; + +use super::{build_add_reasoning, build_append_reasoning, map_stop_reason, saturating_i32}; + +#[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"] + ); +} diff --git a/app/src/bin/oss.rs b/app/src/bin/oss.rs index 9aa7156e..932b4077 100644 --- a/app/src/bin/oss.rs +++ b/app/src/bin/oss.rs @@ -14,7 +14,7 @@ fn main() -> Result<()> { ChannelConfig { app_id: AppId::new("com", "samsung", "Galaxy"), logfile_name: "galaxy.log".into(), - server_config: WarpServerConfig::production(), + server_config: WarpServerConfig::disabled(), oz_config: OzConfig::production(), telemetry_config: None, autoupdate_config: None, diff --git a/app/src/settings/ai.rs b/app/src/settings/ai.rs index d2bd88aa..127664d4 100644 --- a/app/src/settings/ai.rs +++ b/app/src/settings/ai.rs @@ -874,10 +874,27 @@ pub struct OpenAIModelConfig { description = "Optional provider hint (e.g. anthropic, openai, google) for icon display." )] pub provider: Option, + #[serde(default)] + #[schemars( + description = "Route this model through Galaxy's Rig runtime. This is an opt-in migration path." + )] + pub use_rig: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schemars( + description = "Whether this endpoint accepts system-role messages. Set false for ChatGPT-backed LiteLLM models that reject them." + )] + pub supports_system_messages: Option, } impl settings_value::SettingsValue for OpenAIModelConfig {} +impl OpenAIModelConfig { + pub fn supports_system_messages(&self) -> bool { + self.supports_system_messages + .unwrap_or_else(|| !self.model_id.starts_with("codex-gpt-")) + } +} + /// Configuration for a single OpenAI-compatible provider endpoint. /// /// Multiple providers can be configured simultaneously (e.g. LiteLLM for cloud models, @@ -901,6 +918,30 @@ pub struct OpenAIProviderConfig { impl settings_value::SettingsValue for OpenAIProviderConfig {} +const INITIAL_LITELLM_BASE_URL: &str = "https://ai.ryserve.net/v1"; +const INITIAL_RIG_MODEL_ID: &str = "codex-gpt-5.6-sol-xhigh"; + +fn default_openai_providers() -> Vec { + vec![OpenAIProviderConfig { + name: "LiteLLM (ai.ryserve.net)".to_string(), + base_url: INITIAL_LITELLM_BASE_URL.to_string(), + // Credentials are deliberately never committed. Set this locally in + // ~/.galaxy/settings.toml before sending a request. + api_key: None, + models: vec![OpenAIModelConfig { + model_id: INITIAL_RIG_MODEL_ID.to_string(), + display_name: "Codex GPT-5.6 SOL (xhigh)".to_string(), + vision_supported: false, + context_size: default_context_size(), + max_input_tokens: None, + max_output_tokens: None, + provider: Some("openai".to_string()), + use_rig: true, + supports_system_messages: Some(false), + }], + }] +} + /// Cached metadata and runtime session options for an ACP agent. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, schemars::JsonSchema)] pub struct AcpAgentSettings { @@ -1447,7 +1488,7 @@ define_settings_group!(AISettings, settings: [ // Whether the OpenAI-compatible (LiteLLM) provider is enabled. openai_enabled: OpenAIEnabled { type: bool, - default: false, + default: true, supported_platforms: SupportedPlatforms::DESKTOP, sync_to_cloud: SyncToCloud::Globally(RespectUserSyncSetting::Yes), private: false, @@ -1498,9 +1539,11 @@ define_settings_group!(AISettings, settings: [ // Each provider has its own name, base_url, api_key, and model list. openai_providers: OpenAIProviders { type: Vec, - default: Vec::new(), + default: default_openai_providers(), supported_platforms: SupportedPlatforms::DESKTOP, - sync_to_cloud: SyncToCloud::Globally(RespectUserSyncSetting::Yes), + // Provider entries may contain API keys, so the complete setting must + // remain local even when preference sync is enabled. + sync_to_cloud: SyncToCloud::Never, private: false, toml_path: "ai.providers", description: "Multiple OpenAI-compatible provider endpoints (e.g. LiteLLM, Ollama, local models).", diff --git a/app/src/settings/ai_tests.rs b/app/src/settings/ai_tests.rs index 8a7621be..74a07fff 100644 --- a/app/src/settings/ai_tests.rs +++ b/app/src/settings/ai_tests.rs @@ -345,6 +345,37 @@ fn test_toolbar_command_map_roundtrip() { assert_eq!(original, restored); } +#[test] +fn initial_litellm_provider_maps_codex_model_to_rig_without_a_committed_key() { + let providers = default_openai_providers(); + + assert_eq!(providers.len(), 1); + let provider = &providers[0]; + assert_eq!(provider.base_url, INITIAL_LITELLM_BASE_URL); + assert_eq!(provider.api_key, None); + assert_eq!(provider.models.len(), 1); + let model = &provider.models[0]; + assert_eq!(model.model_id, INITIAL_RIG_MODEL_ID); + assert_eq!(model.use_rig, true); + assert_eq!(model.supports_system_messages, Some(false)); + assert_eq!(model.supports_system_messages(), false); +} + +#[test] +fn codex_litellm_model_infers_missing_system_message_capability() { + let mut model = default_openai_providers().remove(0).models.remove(0); + model.supports_system_messages = None; + + assert_eq!(model.supports_system_messages(), false); + + model.model_id = "gpt-4o".to_string(); + assert_eq!(model.supports_system_messages(), true); + + model.model_id = INITIAL_RIG_MODEL_ID.to_string(); + model.supports_system_messages = Some(true); + assert_eq!(model.supports_system_messages(), true); +} + #[test] fn test_toolbar_command_map_matched_agent() { App::test((), |mut app| async move { diff --git a/crates/galaxy_agent_core/Cargo.toml b/crates/galaxy_agent_core/Cargo.toml new file mode 100644 index 00000000..230e8e0c --- /dev/null +++ b/crates/galaxy_agent_core/Cargo.toml @@ -0,0 +1,13 @@ +[package] +name = "galaxy_agent_core" +version = "0.1.0" +edition = "2024" +publish.workspace = true +license.workspace = true + +[dependencies] +async-channel.workspace = true +async-trait.workspace = true +futures.workspace = true +serde.workspace = true +serde_json.workspace = true diff --git a/crates/galaxy_agent_core/src/lib.rs b/crates/galaxy_agent_core/src/lib.rs new file mode 100644 index 00000000..4bfd6dc5 --- /dev/null +++ b/crates/galaxy_agent_core/src/lib.rs @@ -0,0 +1,11 @@ +//! Provider- and UI-independent contracts for Galaxy agent runtimes. +//! +//! This crate is the stable boundary between Galaxy application services and +//! concrete runtimes such as Rig-backed providers or ACP agents. It must not +//! depend on GalaxyUI, provider SDKs, persistence, or Warp wire protocols. + +mod runtime; +mod types; + +pub use runtime::*; +pub use types::*; diff --git a/crates/galaxy_agent_core/src/runtime.rs b/crates/galaxy_agent_core/src/runtime.rs new file mode 100644 index 00000000..7d7792bb --- /dev/null +++ b/crates/galaxy_agent_core/src/runtime.rs @@ -0,0 +1,144 @@ +use std::error::Error; +use std::fmt; +use std::pin::Pin; + +use async_channel::{Receiver, Sender, TrySendError}; +use async_trait::async_trait; +use futures::Stream; +use serde::{Deserialize, Serialize}; + +use crate::{AgentEvent, TurnRequest}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum RuntimeKind { + Provider, + Acp, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct RuntimeCapabilities { + pub model_selection: bool, + pub session_resume: bool, + pub steering: bool, + pub tool_permissions: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct RuntimeDescriptor { + pub id: String, + pub display_name: String, + pub kind: RuntimeKind, + pub capabilities: RuntimeCapabilities, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum TurnCommand { + Cancel, + Steer { text: String }, +} + +#[derive(Clone, Debug)] +pub struct TurnCommandSender(Sender); + +impl TurnCommandSender { + pub async fn send(&self, command: TurnCommand) -> Result<(), TurnControlClosed> { + self.0.send(command).await.map_err(|_| TurnControlClosed) + } + + pub fn try_send(&self, command: TurnCommand) -> Result<(), TrySendError> { + self.0.try_send(command) + } +} + +#[derive(Clone, Debug)] +pub struct TurnControl(Receiver); + +impl TurnControl { + pub async fn receive(&self) -> Result { + self.0.recv().await.map_err(|_| TurnControlClosed) + } + + pub fn try_receive(&self) -> Result { + self.0.try_recv() + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct TurnControlClosed; + +impl fmt::Display for TurnControlClosed { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("turn control channel is closed") + } +} + +impl Error for TurnControlClosed {} + +pub fn turn_control() -> (TurnCommandSender, TurnControl) { + let (sender, receiver) = async_channel::unbounded(); + (TurnCommandSender(sender), TurnControl(receiver)) +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum AgentErrorKind { + Configuration, + Authentication, + RateLimited, + ContextWindowExceeded, + InvalidRequest, + Transport, + Provider, + Protocol, + Tool, + Cancelled, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct AgentError { + pub kind: AgentErrorKind, + pub message: String, + pub user_message: Option, + pub recoverable: bool, +} + +impl AgentError { + pub fn new(kind: AgentErrorKind, message: impl Into) -> Self { + Self { + kind, + message: message.into(), + user_message: None, + recoverable: false, + } + } +} + +impl fmt::Display for AgentError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(&self.message) + } +} + +impl Error for AgentError {} + +#[cfg(not(target_family = "wasm"))] +pub type AgentEventStream = + Pin> + Send + 'static>>; + +#[cfg(target_family = "wasm")] +pub type AgentEventStream = Pin> + 'static>>; + +#[cfg_attr(target_family = "wasm", async_trait(?Send))] +#[cfg_attr(not(target_family = "wasm"), async_trait)] +pub trait AgentRuntime: Send + Sync { + fn descriptor(&self) -> &RuntimeDescriptor; + + async fn start_turn( + &self, + request: TurnRequest, + control: TurnControl, + ) -> Result; +} + +#[cfg(test)] +#[path = "runtime_tests.rs"] +mod tests; diff --git a/crates/galaxy_agent_core/src/runtime_tests.rs b/crates/galaxy_agent_core/src/runtime_tests.rs new file mode 100644 index 00000000..04938a0c --- /dev/null +++ b/crates/galaxy_agent_core/src/runtime_tests.rs @@ -0,0 +1,120 @@ +use std::sync::Mutex; + +use futures::{StreamExt, stream}; + +use super::*; +use crate::{ + AgentEvent, ConversationMessage, MessageContent, MessageRole, ModelId, StopReason, Usage, +}; + +struct FakeRuntime { + descriptor: RuntimeDescriptor, + requests: Mutex>, + events: Vec, +} + +impl FakeRuntime { + fn new(events: Vec) -> Self { + Self { + descriptor: RuntimeDescriptor { + id: "fake".to_string(), + display_name: "Deterministic fake".to_string(), + kind: RuntimeKind::Provider, + capabilities: RuntimeCapabilities { + model_selection: true, + ..RuntimeCapabilities::default() + }, + }, + requests: Mutex::new(Vec::new()), + events, + } + } +} + +#[async_trait] +impl AgentRuntime for FakeRuntime { + fn descriptor(&self) -> &RuntimeDescriptor { + &self.descriptor + } + + async fn start_turn( + &self, + request: TurnRequest, + _control: TurnControl, + ) -> Result { + self.requests.lock().unwrap().push(request); + let events = self.events.clone().into_iter().map(Ok); + Ok(Box::pin(stream::iter(events))) + } +} + +#[test] +fn fake_runtime_preserves_request_and_event_contract() { + futures::executor::block_on(async { + let expected_events = vec![ + AgentEvent::TurnStarted { + runtime_request_id: "request-1".to_string(), + }, + AgentEvent::TextDelta { + text: "hello".to_string(), + }, + AgentEvent::UsageUpdated { + usage: Usage { + input_tokens: 4, + output_tokens: 1, + ..Usage::default() + }, + }, + AgentEvent::TurnStopped { + reason: StopReason::Completed, + }, + ]; + let fake_runtime = FakeRuntime::new(expected_events.clone()); + let runtime: &dyn AgentRuntime = &fake_runtime; + let request = TurnRequest::new( + ModelId::new("fake-model"), + vec![ConversationMessage { + role: MessageRole::User, + content: MessageContent::Text("Say hello".to_string()), + }], + ); + let (_sender, control) = turn_control(); + + let actual_events = runtime + .start_turn(request.clone(), control) + .await + .unwrap() + .collect::>() + .await + .into_iter() + .collect::, _>>() + .unwrap(); + + assert_eq!(runtime.descriptor().id, "fake"); + assert_eq!(*fake_runtime.requests.lock().unwrap(), vec![request]); + assert_eq!(actual_events, expected_events); + }); +} + +#[test] +fn turn_control_delivers_cancel_and_steering_in_order() { + futures::executor::block_on(async { + let (sender, control) = turn_control(); + + sender + .send(TurnCommand::Steer { + text: "focus on tests".to_string(), + }) + .await + .unwrap(); + sender.send(TurnCommand::Cancel).await.unwrap(); + + assert_eq!( + control.receive().await.unwrap(), + TurnCommand::Steer { + text: "focus on tests".to_string(), + } + ); + assert_eq!(control.receive().await.unwrap(), TurnCommand::Cancel); + }); +} diff --git a/crates/galaxy_agent_core/src/types.rs b/crates/galaxy_agent_core/src/types.rs new file mode 100644 index 00000000..f47f32d8 --- /dev/null +++ b/crates/galaxy_agent_core/src/types.rs @@ -0,0 +1,227 @@ +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; +use serde_json::Value as JsonValue; + +pub const MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST: usize = 64_000; + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct ConversationMessage { + pub role: MessageRole, + pub content: MessageContent, +} + +impl ConversationMessage { + pub fn truncate_tool_results_for_provider_request(&mut self) { + truncate_tool_results_in_content(&mut self.content); + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum MessageRole { + User, + Assistant, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub enum MessageContent { + Text(String), + ToolUse { + tool_use_id: String, + name: String, + input: JsonValue, + }, + ToolResult { + tool_use_id: String, + content: String, + is_error: bool, + }, + MultiPart(Vec), +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub enum ContentPart { + Text(String), + Image { + data: Vec, + mime_type: String, + }, + ToolUse { + tool_use_id: String, + name: String, + input: JsonValue, + }, + ToolResult { + tool_use_id: String, + content: String, + is_error: bool, + }, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct ToolDefinition { + pub name: String, + pub description: String, + pub input_schema: JsonValue, +} + +#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct ModelId(String); + +impl ModelId { + pub fn new(value: impl Into) -> Self { + Self(value.into()) + } + + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl From for ModelId { + fn from(value: String) -> Self { + Self(value) + } +} + +impl From<&str> for ModelId { + fn from(value: &str) -> Self { + Self(value.to_string()) + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct TurnRequest { + pub conversation_id: Option, + pub model: ModelId, + pub system_prompt: Option, + pub messages: Vec, + pub tools: Vec, + pub max_output_tokens: Option, + pub metadata: BTreeMap, +} + +impl TurnRequest { + pub fn new(model: impl Into, messages: Vec) -> Self { + Self { + conversation_id: None, + model: model.into(), + system_prompt: None, + messages, + tools: Vec::new(), + max_output_tokens: None, + metadata: BTreeMap::new(), + } + } +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct ToolCall { + pub id: String, + pub name: String, + pub arguments: JsonValue, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct ToolResult { + pub call_id: String, + pub content: String, + pub is_error: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum PermissionKind { + Read, + Write, + Execute, + Network, + ExternalTool, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct PermissionRequest { + pub id: String, + pub tool_call: ToolCall, + pub kind: PermissionKind, + pub reason: Option, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct Usage { + pub input_tokens: u64, + pub output_tokens: u64, + pub cached_input_tokens: u64, + pub cache_creation_input_tokens: u64, +} + +impl Usage { + pub fn total_tokens(&self) -> u64 { + self.input_tokens.saturating_add(self.output_tokens) + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum StopReason { + Completed, + Cancelled, + MaxTokens, + ContextWindowExceeded, + Refusal, + ToolLoopLimit, + Other(String), +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub enum AgentEvent { + TurnStarted { runtime_request_id: String }, + TextDelta { text: String }, + ReasoningDelta { text: String }, + ToolProposed { call: ToolCall }, + PermissionRequested { request: PermissionRequest }, + ToolStarted { call: ToolCall }, + ToolCompleted { result: ToolResult }, + UsageUpdated { usage: Usage }, + TurnStopped { reason: StopReason }, +} + +fn truncate_tool_results_in_content(content: &mut MessageContent) { + match content { + MessageContent::Text(_) | MessageContent::ToolUse { .. } => {} + MessageContent::ToolResult { content, .. } => truncate_tool_result_text(content), + MessageContent::MultiPart(parts) => { + for part in parts { + if let ContentPart::ToolResult { content, .. } = part { + truncate_tool_result_text(content); + } + } + } + } +} + +fn truncate_tool_result_text(content: &mut String) { + let char_count = content.chars().count(); + if char_count <= MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST { + return; + } + + let omitted_chars = char_count.saturating_sub(MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST); + let marker = format!("\n... [tool result truncated; omitted {omitted_chars} chars] ...\n"); + let marker_chars = marker.chars().count(); + let retained_chars = MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST.saturating_sub(marker_chars); + let head_chars = retained_chars / 2; + let tail_chars = retained_chars.saturating_sub(head_chars); + let head: String = content.chars().take(head_chars).collect(); + let tail: String = content + .chars() + .rev() + .take(tail_chars) + .collect::() + .chars() + .rev() + .collect(); + *content = format!("{head}{marker}{tail}"); +} + +#[cfg(test)] +#[path = "types_tests.rs"] +mod tests; diff --git a/crates/galaxy_agent_core/src/types_tests.rs b/crates/galaxy_agent_core/src/types_tests.rs new file mode 100644 index 00000000..4fdebce2 --- /dev/null +++ b/crates/galaxy_agent_core/src/types_tests.rs @@ -0,0 +1,38 @@ +use super::*; + +#[test] +fn truncates_large_tool_results_for_provider_request() { + let prefix = "start:"; + let suffix = ":end"; + let middle = "x".repeat(MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST + 1_000); + let mut message = ConversationMessage { + role: MessageRole::User, + content: MessageContent::ToolResult { + tool_use_id: "toolu_1".to_string(), + content: format!("{prefix}{middle}{suffix}"), + is_error: false, + }, + }; + + message.truncate_tool_results_for_provider_request(); + + let MessageContent::ToolResult { content, .. } = message.content else { + panic!("expected tool result"); + }; + assert!(content.len() <= MAX_TOOL_RESULT_CHARS_FOR_PROVIDER_REQUEST + 128); + assert!(content.starts_with(prefix)); + assert!(content.ends_with(suffix)); + assert!(content.contains("tool result truncated")); +} + +#[test] +fn usage_total_excludes_cached_breakdown_to_avoid_double_counting() { + let usage = Usage { + input_tokens: 100, + output_tokens: 25, + cached_input_tokens: 80, + cache_creation_input_tokens: 10, + }; + + assert_eq!(usage.total_tokens(), 125); +} diff --git a/crates/galaxy_agent_rig/Cargo.toml b/crates/galaxy_agent_rig/Cargo.toml new file mode 100644 index 00000000..90bd9358 --- /dev/null +++ b/crates/galaxy_agent_rig/Cargo.toml @@ -0,0 +1,20 @@ +[package] +name = "galaxy_agent_rig" +version = "0.1.0" +edition = "2024" +publish.workspace = true +license.workspace = true + +[dependencies] +async-stream.workspace = true +async-trait.workspace = true +futures.workspace = true +galaxy_agent_core.workspace = true +rig-core.workspace = true +serde_json.workspace = true +uuid.workspace = true + +[dev-dependencies] +bytes.workspace = true +rig-core = { workspace = true, features = ["test-utils"] } +tokio = { workspace = true, features = ["macros", "rt"] } diff --git a/crates/galaxy_agent_rig/src/lib.rs b/crates/galaxy_agent_rig/src/lib.rs new file mode 100644 index 00000000..b00371d5 --- /dev/null +++ b/crates/galaxy_agent_rig/src/lib.rs @@ -0,0 +1,5 @@ +//! Rig-backed implementations of Galaxy's provider-neutral agent runtime. + +mod openai_compatible; + +pub use openai_compatible::*; diff --git a/crates/galaxy_agent_rig/src/openai_compatible.rs b/crates/galaxy_agent_rig/src/openai_compatible.rs new file mode 100644 index 00000000..ef52131b --- /dev/null +++ b/crates/galaxy_agent_rig/src/openai_compatible.rs @@ -0,0 +1,432 @@ +use async_trait::async_trait; +use futures::{FutureExt, StreamExt}; +use galaxy_agent_core::{ + AgentError, AgentErrorKind, AgentEvent, AgentEventStream, AgentRuntime, ContentPart, + ConversationMessage, MessageContent, MessageRole, RuntimeCapabilities, RuntimeDescriptor, + RuntimeKind, StopReason, ToolCall, TurnCommand, TurnControl, TurnRequest, Usage, +}; +use rig_core::OneOrMany; +use rig_core::client::CompletionClient; +use rig_core::completion::{ + AssistantContent, CompletionError, CompletionModel, CompletionRequest, GetTokenUsage, Message, + ToolDefinition, +}; +use rig_core::message::{ + DocumentSourceKind, Image, ImageMediaType, MimeType, ToolResultContent, UserContent, +}; +use rig_core::providers::openai; +use rig_core::streaming::StreamedAssistantContent; +use uuid::Uuid; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct OpenAICompatibleRuntimeConfig { + pub base_url: String, + pub api_key: Option, + pub model: String, + pub max_output_tokens: Option, + pub supports_system_messages: bool, +} + +#[derive(Clone, Debug)] +pub struct OpenAICompatibleRuntime { + config: OpenAICompatibleRuntimeConfig, + descriptor: RuntimeDescriptor, +} + +impl OpenAICompatibleRuntime { + pub fn new(config: OpenAICompatibleRuntimeConfig) -> Self { + let descriptor = RuntimeDescriptor { + 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, + }, + }; + Self { config, descriptor } + } +} + +#[async_trait] +impl AgentRuntime for OpenAICompatibleRuntime { + fn descriptor(&self) -> &RuntimeDescriptor { + &self.descriptor + } + + async fn start_turn( + &self, + request: TurnRequest, + control: TurnControl, + ) -> Result { + let client = openai::CompletionsClient::builder() + // Rig 0.40 requires an API-key builder value. An empty key preserves + // compatibility with unauthenticated local OpenAI-compatible servers. + .api_key(self.config.api_key.as_deref().unwrap_or_default()) + .base_url(&self.config.base_url) + .build() + .map_err(|error| AgentError::new(AgentErrorKind::Configuration, error.to_string()))?; + let model = client.completion_model(&self.config.model); + start_model_turn( + model, + request, + control, + self.config.max_output_tokens, + self.config.supports_system_messages, + ) + .await + } +} + +async fn start_model_turn( + model: M, + request: TurnRequest, + control: TurnControl, + configured_max_output_tokens: Option, + supports_system_messages: bool, +) -> Result +where + M: CompletionModel + Send + Sync + 'static, + M::StreamingResponse: Send + Sync + 'static, +{ + let runtime_request_id = Uuid::new_v4().to_string(); + let max_output_tokens = request.max_output_tokens.or(configured_max_output_tokens); + let completion_request = build_completion_request( + request, + configured_max_output_tokens, + supports_system_messages, + )?; + let stream_future = model.stream(completion_request).fuse(); + let initial_control = control.clone(); + let control_future = initial_control.receive().fuse(); + futures::pin_mut!(stream_future, control_future); + + let mut rig_stream = futures::select_biased! { + command = control_future => match command { + Ok(TurnCommand::Cancel) => { + return Ok(stopped_before_stream(runtime_request_id)); + } + Ok(TurnCommand::Steer { .. }) | Err(_) => { + stream_future.await.map_err(map_completion_error)? + } + }, + result = stream_future => result.map_err(map_completion_error)?, + }; + + let events = async_stream::stream! { + yield Ok(AgentEvent::TurnStarted { + runtime_request_id, + }); + + let mut control_open = true; + let mut last_output_tokens = 0; + loop { + let next_item = rig_stream.next().fuse(); + let next_command = if control_open { + futures::future::Either::Left(control.receive()) + } else { + futures::future::Either::Right(futures::future::pending()) + } + .fuse(); + futures::pin_mut!(next_item, next_command); + + futures::select_biased! { + command = next_command => { + match command { + Ok(TurnCommand::Cancel) => { + rig_stream.cancel(); + yield Ok(AgentEvent::TurnStopped { + reason: StopReason::Cancelled, + }); + return; + } + Ok(TurnCommand::Steer { .. }) => { + // Steering is not advertised by this runtime yet. + } + Err(_) => control_open = false, + } + } + item = next_item => { + let Some(item) = item else { + yield Ok(AgentEvent::TurnStopped { + reason: if max_output_tokens.is_some_and(|max| { + last_output_tokens >= max + }) { + StopReason::MaxTokens + } else { + StopReason::Completed + }, + }); + return; + }; + + match item { + Ok(StreamedAssistantContent::Text(text)) => { + if !text.text.is_empty() { + yield Ok(AgentEvent::TextDelta { text: text.text }); + } + } + Ok(StreamedAssistantContent::Reasoning(reasoning)) => { + let text = reasoning.display_text(); + if !text.is_empty() { + yield Ok(AgentEvent::ReasoningDelta { text }); + } + } + Ok(StreamedAssistantContent::ReasoningDelta { reasoning, .. }) => { + if !reasoning.is_empty() { + yield Ok(AgentEvent::ReasoningDelta { text: reasoning }); + } + } + Ok(StreamedAssistantContent::ToolCall { tool_call, .. }) => { + yield Ok(AgentEvent::ToolProposed { + call: ToolCall { + id: tool_call.id, + name: tool_call.function.name, + arguments: tool_call.function.arguments, + }, + }); + } + Ok(StreamedAssistantContent::ToolCallDelta { .. }) => { + // Rig emits a complete ToolCall after its deltas, which + // is the canonical event Galaxy consumes. + } + Ok(StreamedAssistantContent::Final(response)) => { + let mapped_usage = map_usage(response.token_usage()); + last_output_tokens = mapped_usage.output_tokens; + yield Ok(AgentEvent::UsageUpdated { + usage: mapped_usage, + }); + } + Ok(StreamedAssistantContent::Unknown(value)) => { + yield Err(AgentError::new( + AgentErrorKind::Protocol, + format!("Rig returned an unsupported provider event: {value}"), + )); + return; + } + Err(error) => { + yield Err(map_completion_error(error)); + return; + } + } + } + } + } + }; + + Ok(Box::pin(events)) +} + +fn stopped_before_stream(runtime_request_id: String) -> AgentEventStream { + Box::pin(futures::stream::iter([ + Ok(AgentEvent::TurnStarted { runtime_request_id }), + Ok(AgentEvent::TurnStopped { + reason: StopReason::Cancelled, + }), + ])) +} + +fn build_completion_request( + request: TurnRequest, + configured_max_output_tokens: Option, + supports_system_messages: bool, +) -> Result { + let mut messages = Vec::new(); + if let Some(system_prompt) = request.system_prompt { + if supports_system_messages { + messages.push(Message::System { + content: system_prompt, + }); + } else { + messages.push(Message::User { + content: OneOrMany::one(UserContent::text(system_prompt)), + }); + } + } + for message in request.messages { + messages.push(convert_message(message)?); + } + + let chat_history = OneOrMany::many(messages).map_err(|_| { + AgentError::new( + AgentErrorKind::InvalidRequest, + "a Rig turn requires at least one conversation message", + ) + })?; + + Ok(CompletionRequest { + model: Some(request.model.as_str().to_string()), + preamble: None, + chat_history, + documents: Vec::new(), + tools: request + .tools + .into_iter() + .map(|tool| ToolDefinition { + name: tool.name, + description: tool.description, + parameters: tool.input_schema, + }) + .collect(), + temperature: None, + max_tokens: request.max_output_tokens.or(configured_max_output_tokens), + tool_choice: None, + additional_params: Some(serde_json::json!({ + "stream_options": { "include_usage": true } + })), + output_schema: None, + }) +} + +fn convert_message(message: ConversationMessage) -> Result { + match message.role { + MessageRole::User => Ok(Message::User { + content: user_content(message.content)?, + }), + MessageRole::Assistant => Ok(Message::Assistant { + id: None, + content: assistant_content(message.content)?, + }), + } +} + +fn user_content(content: MessageContent) -> Result, AgentError> { + let parts = match content { + MessageContent::Text(text) => vec![UserContent::text(text)], + MessageContent::ToolResult { + tool_use_id, + content, + .. + } => vec![UserContent::tool_result( + tool_use_id, + OneOrMany::one(ToolResultContent::text(content)), + )], + MessageContent::MultiPart(parts) => parts + .into_iter() + .map(convert_user_part) + .collect::, _>>()?, + MessageContent::ToolUse { .. } => { + return Err(invalid_role("tool use", "user")); + } + }; + one_or_many(parts, "user") +} + +fn assistant_content(content: MessageContent) -> Result, AgentError> { + let parts = match content { + MessageContent::Text(text) => vec![AssistantContent::text(text)], + MessageContent::ToolUse { + tool_use_id, + name, + input, + } => vec![AssistantContent::tool_call(tool_use_id, name, input)], + MessageContent::MultiPart(parts) => parts + .into_iter() + .map(convert_assistant_part) + .collect::, _>>()?, + MessageContent::ToolResult { .. } => { + return Err(invalid_role("tool result", "assistant")); + } + }; + one_or_many(parts, "assistant") +} + +fn convert_user_part(part: ContentPart) -> Result { + match part { + ContentPart::Text(text) => Ok(UserContent::text(text)), + ContentPart::Image { data, mime_type } => Ok(UserContent::image_raw( + data, + ImageMediaType::from_mime_type(&mime_type), + None, + )), + ContentPart::ToolResult { + tool_use_id, + content, + .. + } => Ok(UserContent::tool_result( + tool_use_id, + OneOrMany::one(ToolResultContent::text(content)), + )), + ContentPart::ToolUse { .. } => Err(invalid_role("tool use", "user")), + } +} + +fn convert_assistant_part(part: ContentPart) -> Result { + match part { + ContentPart::Text(text) => Ok(AssistantContent::text(text)), + ContentPart::Image { data, mime_type } => Ok(AssistantContent::Image(Image { + data: DocumentSourceKind::Raw(data), + media_type: ImageMediaType::from_mime_type(&mime_type), + detail: None, + additional_params: None, + })), + ContentPart::ToolUse { + tool_use_id, + name, + input, + } => Ok(AssistantContent::tool_call(tool_use_id, name, input)), + ContentPart::ToolResult { .. } => Err(invalid_role("tool result", "assistant")), + } +} + +fn one_or_many(parts: Vec, role: &str) -> Result, AgentError> { + OneOrMany::many(parts).map_err(|_| { + AgentError::new( + AgentErrorKind::InvalidRequest, + format!("{role} message has no content"), + ) + }) +} + +fn invalid_role(content: &str, role: &str) -> AgentError { + AgentError::new( + AgentErrorKind::InvalidRequest, + format!("{content} content cannot appear in a {role} message"), + ) +} + +fn map_usage(usage: rig_core::completion::Usage) -> Usage { + Usage { + input_tokens: usage.input_tokens, + output_tokens: usage.output_tokens, + cached_input_tokens: usage.cached_input_tokens, + cache_creation_input_tokens: usage.cache_creation_input_tokens, + } +} + +fn map_completion_error(error: CompletionError) -> AgentError { + let status = error + .provider_response_status() + .map(|status| status.as_u16()); + let kind = match status { + Some(401 | 403) => AgentErrorKind::Authentication, + Some(429) => AgentErrorKind::RateLimited, + Some(400 | 404 | 413 | 422) => AgentErrorKind::InvalidRequest, + Some(500..=599) => AgentErrorKind::Provider, + Some(_) => AgentErrorKind::Provider, + None => match &error { + CompletionError::HttpError(_) + | CompletionError::UrlError(_) + | CompletionError::RequestError(_) => AgentErrorKind::Transport, + CompletionError::JsonError(_) | CompletionError::ResponseError(_) => { + AgentErrorKind::Protocol + } + CompletionError::ProviderError(_) | CompletionError::ProviderResponse(_) => { + AgentErrorKind::Provider + } + _ => AgentErrorKind::Provider, + }, + }; + let mut mapped = AgentError::new(kind, error.to_string()); + mapped.recoverable = matches!( + kind, + AgentErrorKind::RateLimited | AgentErrorKind::Transport + ); + mapped +} + +#[cfg(test)] +#[path = "openai_compatible_tests.rs"] +mod tests; diff --git a/crates/galaxy_agent_rig/src/openai_compatible_tests.rs b/crates/galaxy_agent_rig/src/openai_compatible_tests.rs new file mode 100644 index 00000000..234dbee4 --- /dev/null +++ b/crates/galaxy_agent_rig/src/openai_compatible_tests.rs @@ -0,0 +1,201 @@ +use futures::StreamExt; +use galaxy_agent_core::{ + AgentEvent, AgentRuntime, ConversationMessage, MessageContent, MessageRole, +}; +use rig_core::client::CompletionClient; +use rig_core::providers::openai; +use rig_core::test_utils::MockStreamingClient; + +use super::*; + +fn text_request() -> TurnRequest { + TurnRequest::new( + "test-model", + vec![ConversationMessage { + role: MessageRole::User, + content: MessageContent::Text("Hello".to_string()), + }], + ) +} + +fn sse(lines: &[&str]) -> bytes::Bytes { + lines + .iter() + .map(|line| format!("data: {line}\n\n")) + .collect::() + .into() +} + +#[tokio::test] +async fn rig_stream_maps_reasoning_text_usage_and_stop() { + let http_client = MockStreamingClient { + sse_bytes: sse(&[ + r#"{"id":"cmpl-1","model":"test-model","choices":[{"delta":{"reasoning_content":"thinking ","tool_calls":[]},"finish_reason":null}],"usage":null}"#, + r#"{"id":"cmpl-1","model":"test-model","choices":[{"delta":{"content":"Hello ","tool_calls":[]},"finish_reason":null}],"usage":null}"#, + r#"{"id":"cmpl-1","model":"test-model","choices":[{"delta":{"content":"world","tool_calls":[]},"finish_reason":"stop"}],"usage":null}"#, + r#"{"choices":[],"usage":{"prompt_tokens":4,"completion_tokens":6,"total_tokens":10,"prompt_tokens_details":{"cached_tokens":2}}}"#, + "[DONE]", + ]), + }; + let client = openai::CompletionsClient::builder() + .api_key("test-key") + .base_url("http://localhost/v1") + .http_client(http_client) + .build() + .unwrap(); + let model = client.completion_model("test-model"); + let (_, control) = galaxy_agent_core::turn_control(); + + let events = start_model_turn(model, text_request(), control, None, true) + .await + .unwrap() + .collect::>() + .await + .into_iter() + .collect::, _>>() + .unwrap(); + + assert!(matches!(events[0], AgentEvent::TurnStarted { .. })); + assert_eq!( + events[1..], + [ + AgentEvent::ReasoningDelta { + text: "thinking ".to_string(), + }, + AgentEvent::TextDelta { + text: "Hello ".to_string(), + }, + AgentEvent::TextDelta { + text: "world".to_string(), + }, + AgentEvent::UsageUpdated { + usage: Usage { + input_tokens: 4, + output_tokens: 6, + cached_input_tokens: 2, + cache_creation_input_tokens: 0, + }, + }, + AgentEvent::TurnStopped { + reason: StopReason::Completed, + }, + ] + ); +} + +#[tokio::test] +async fn cancellation_before_stream_start_is_a_normal_stop() { + let runtime = OpenAICompatibleRuntime::new(OpenAICompatibleRuntimeConfig { + base_url: "http://localhost/v1".to_string(), + api_key: None, + model: "test-model".to_string(), + max_output_tokens: None, + supports_system_messages: true, + }); + let (sender, control) = galaxy_agent_core::turn_control(); + sender.send(TurnCommand::Cancel).await.unwrap(); + + let events = runtime + .start_turn(text_request(), control) + .await + .unwrap() + .collect::>() + .await + .into_iter() + .collect::, _>>() + .unwrap(); + + assert!(matches!(events[0], AgentEvent::TurnStarted { .. })); + assert_eq!( + events[1], + AgentEvent::TurnStopped { + reason: StopReason::Cancelled, + } + ); +} + +#[tokio::test] +async fn usage_at_the_requested_limit_maps_to_max_tokens() { + let http_client = MockStreamingClient { + sse_bytes: sse(&[ + r#"{"choices":[{"delta":{"content":"cut off","tool_calls":[]},"finish_reason":"length"}],"usage":null}"#, + r#"{"choices":[],"usage":{"prompt_tokens":2,"completion_tokens":6,"total_tokens":8}}"#, + "[DONE]", + ]), + }; + let client = openai::CompletionsClient::builder() + .api_key("test-key") + .base_url("http://localhost/v1") + .http_client(http_client) + .build() + .unwrap(); + let model = client.completion_model("test-model"); + let (sender, control) = galaxy_agent_core::turn_control(); + let mut request = text_request(); + request.max_output_tokens = Some(6); + + let events = start_model_turn(model, request, control, None, true) + .await + .unwrap() + .collect::>() + .await + .into_iter() + .collect::, _>>() + .unwrap(); + drop(sender); + + assert_eq!( + events.last(), + Some(&AgentEvent::TurnStopped { + reason: StopReason::MaxTokens, + }) + ); +} + +#[test] +fn request_conversion_preserves_history_tools_and_limits() { + let mut request = text_request(); + request.system_prompt = Some("Be useful".to_string()); + request.max_output_tokens = Some(123); + request.tools.push(galaxy_agent_core::ToolDefinition { + name: "shell".to_string(), + description: "Run a command".to_string(), + input_schema: serde_json::json!({"type": "object"}), + }); + + let converted = build_completion_request(request, Some(999), true).unwrap(); + + assert_eq!(converted.max_tokens, Some(123)); + assert_eq!(converted.tools.len(), 1); + assert_eq!(converted.tools[0].name, "shell"); + assert_eq!(converted.chat_history.len(), 2); + assert!(matches!( + converted.chat_history.iter().next(), + Some(Message::System { content }) if content == "Be useful" + )); +} + +#[test] +fn request_conversion_places_system_prompt_in_user_message_when_system_role_is_unsupported() { + let mut request = text_request(); + request.system_prompt = Some("Be useful".to_string()); + + let converted = build_completion_request(request, None, false).unwrap(); + let messages = converted.chat_history.iter().collect::>(); + + assert_eq!(messages.len(), 2); + let Message::User { content } = messages[0] else { + panic!("expected the system prompt to use the user role"); + }; + let Some(UserContent::Text(text)) = content.iter().next() else { + panic!("expected text instructions"); + }; + assert_eq!(text.text, "Be useful"); + assert_eq!( + messages + .iter() + .filter(|message| matches!(message, Message::System { .. })) + .count(), + 0 + ); +} diff --git a/crates/galaxy_core/src/channel/channel_tests.rs b/crates/galaxy_core/src/channel/channel_tests.rs index 56ec14bc..b5089e69 100644 --- a/crates/galaxy_core/src/channel/channel_tests.rs +++ b/crates/galaxy_core/src/channel/channel_tests.rs @@ -12,3 +12,13 @@ fn local_control_channel_names_do_not_expose_legacy_branding() { ); assert_eq!(Channel::Oss.local_control_channel_name(), "oss"); } + +#[test] +fn only_oss_disables_warp_service_egress() { + assert!(Channel::Stable.allows_warp_service_egress()); + assert!(Channel::Preview.allows_warp_service_egress()); + assert!(Channel::Dev.allows_warp_service_egress()); + assert!(Channel::Local.allows_warp_service_egress()); + assert!(Channel::Integration.allows_warp_service_egress()); + assert!(!Channel::Oss.allows_warp_service_egress()); +} diff --git a/crates/galaxy_core/src/channel/config.rs b/crates/galaxy_core/src/channel/config.rs index 9e14b251..6a08547f 100644 --- a/crates/galaxy_core/src/channel/config.rs +++ b/crates/galaxy_core/src/channel/config.rs @@ -52,6 +52,23 @@ pub struct WarpServerConfig { } impl WarpServerConfig { + /// Returns a loopback-only configuration for builds that must not communicate + /// with Warp-operated services. + /// + /// Loopback URLs keep legacy URL construction code valid while ensuring any + /// accidentally reachable request remains on the user's machine. Callers + /// must still fail closed before attempting authentication because Firebase + /// token exchange uses provider-owned URLs rather than `server_root_url`. + pub fn disabled() -> Self { + Self { + server_root_url: "http://127.0.0.1:9".into(), + rtc_server_url: "ws://127.0.0.1:9/graphql/v2".into(), + session_sharing_server_url: None, + firebase_auth_api_key: "".into(), + iap_config: None, + } + } + pub fn production() -> Self { Self { server_root_url: "https://app.warp.dev".into(), @@ -63,6 +80,10 @@ impl WarpServerConfig { } } +#[cfg(test)] +#[path = "config_tests.rs"] +mod tests; + #[derive(Debug, Deserialize, Serialize)] pub struct OzConfig { /// Root URL for the Oz (ambient agent management) dashboard. diff --git a/crates/galaxy_core/src/channel/config_tests.rs b/crates/galaxy_core/src/channel/config_tests.rs new file mode 100644 index 00000000..76653c99 --- /dev/null +++ b/crates/galaxy_core/src/channel/config_tests.rs @@ -0,0 +1,12 @@ +use super::WarpServerConfig; + +#[test] +fn disabled_warp_services_are_loopback_only() { + let config = WarpServerConfig::disabled(); + + assert_eq!(config.server_root_url, "http://127.0.0.1:9"); + assert_eq!(config.rtc_server_url, "ws://127.0.0.1:9/graphql/v2"); + assert!(config.session_sharing_server_url.is_none()); + assert!(config.firebase_auth_api_key.is_empty()); + assert!(config.iap_config.is_none()); +} diff --git a/crates/galaxy_core/src/channel/mod.rs b/crates/galaxy_core/src/channel/mod.rs index e2621d4b..c0735d42 100644 --- a/crates/galaxy_core/src/channel/mod.rs +++ b/crates/galaxy_core/src/channel/mod.rs @@ -47,6 +47,22 @@ impl Channel { } } + /// Whether this channel may communicate with Warp-operated services. + /// + /// The OSS product is local-first. Provider endpoints explicitly configured + /// by the user are outside this policy, but inherited Warp authentication, + /// cloud sync, RTC, and session-sharing services must remain unavailable. + pub fn allows_warp_service_egress(&self) -> bool { + match self { + Channel::Stable + | Channel::Preview + | Channel::Dev + | Channel::Local + | Channel::Integration => true, + Channel::Oss => false, + } + } + /// Returns the CLI command name corresponding to this channel. pub fn cli_command_name(&self) -> &'static str { match self { diff --git a/crates/galaxy_core/src/channel/state.rs b/crates/galaxy_core/src/channel/state.rs index 9e90d149..e03f91bc 100644 --- a/crates/galaxy_core/src/channel/state.rs +++ b/crates/galaxy_core/src/channel/state.rs @@ -44,7 +44,7 @@ impl ChannelState { config: ChannelConfig { app_id, logfile_name: "".into(), - server_config: WarpServerConfig::production(), + server_config: WarpServerConfig::disabled(), oz_config: OzConfig::production(), telemetry_config: None, autoupdate_config: None, diff --git a/crates/galaxy_server_client/src/auth/session.rs b/crates/galaxy_server_client/src/auth/session.rs index 6e9751c6..ca177785 100644 --- a/crates/galaxy_server_client/src/auth/session.rs +++ b/crates/galaxy_server_client/src/auth/session.rs @@ -214,6 +214,14 @@ impl AuthSession { &self, token: FirebaseToken, ) -> BoxFuture<'static, StdResult> { + if !ChannelState::channel().allows_warp_service_egress() { + return Box::pin(async { + Err(UserAuthenticationError::Unexpected(anyhow::anyhow!( + "Warp authentication is disabled in this local-only build" + ))) + }); + } + let client = self.client.clone(); Box::pin(async move { let firebase_api_key = ChannelState::firebase_api_key(); diff --git a/crates/warp_tui/src/bin/oss.rs b/crates/warp_tui/src/bin/oss.rs index f4b3b10c..e04fc75e 100644 --- a/crates/warp_tui/src/bin/oss.rs +++ b/crates/warp_tui/src/bin/oss.rs @@ -16,7 +16,7 @@ fn main() -> Result<()> { ChannelConfig { app_id: AppId::new("dev", "warp", "WarpTui"), logfile_name: "warp-tui.log".into(), - server_config: WarpServerConfig::production(), + server_config: WarpServerConfig::disabled(), oz_config: OzConfig::production(), telemetry_config: None, autoupdate_config: None, diff --git a/plans/galaxy-local-first-rig.md b/plans/galaxy-local-first-rig.md new file mode 100644 index 00000000..e5b820c8 --- /dev/null +++ b/plans/galaxy-local-first-rig.md @@ -0,0 +1,393 @@ +# Galaxy Local-First Recovery and Rig Migration + +> **Started:** 2026-08-04 +> **Status:** Active architecture recovery +> **UI ledger:** [`ui-flow-inventory.md`](ui-flow-inventory.md) +> **Supersedes:** [`galaxy-refactor.md`](galaxy-refactor.md) + +## Product contract + +Galaxy is a local-first developer terminal with Warp-quality interaction design. It may communicate +with a model provider, remote machine, or tool only when the user or an administrator has explicitly +configured that boundary. It must not depend on Warp authentication, cloud storage, billing, +telemetry, remote logging, remote feature control, session sharing, or Oz. + +The non-negotiable properties are: + +1. A fresh install works without an account. +2. Terminal, editor, conversation, rules, profiles, notebooks, workflows, and history data are local. +3. No inherited Warp endpoint can address an external host in the OSS build. +4. Model traffic goes only to the provider selected for the active model. +5. ACP agents are explicit, trusted local subprocesses with a visible permission boundary. +6. Network-capable tools are off by default and visible when enabled or invoked. +7. The UI consumes Galaxy-owned domain types, not a provider SDK or Warp wire protocol. +8. Provider and agent implementations are replaceable without changing conversation UI code. + +## Current baseline + +The code is not merely untidy; it has conflicting architectural centers. + +- `app/src` contains roughly 1.08 million lines of Rust across product and test code. +- `app/src/ai` alone contains roughly 270,000 lines in 555 Rust files. +- Seventy-six app files reference `warp_multi_agent_api`. +- The provider and ACP implementation inspected for this plan spans more than 22,000 lines. +- Large presentation/coordinator files include `workspace/view.rs` (about 29,000 lines), + `terminal/view.rs` (about 29,000), `terminal/input.rs` (about 16,000), and + `settings_view/ai_page.rs` (about 8,500). +- The OSS binary configured Warp production HTTP, RTC, session-sharing, and Firebase values even + though telemetry sending had already been stubbed out. The first safety patch replaces those + values with loopback-only disabled configuration and rejects Warp/Firebase auth exchange. + +The provider-backed prompt path currently resembles: + +```text +Galaxy UI/controller + -> RequestParams (already contains provider-specific Bedrock history fields) + -> warp_multi_agent_api::Request protobuf + -> Bedrock or OpenAI request translator + -> provider SDK / JSON / SSE + -> provider response translator + -> warp_multi_agent_api::ResponseEvent protobuf + -> Galaxy controller/history/UI +``` + +ACP takes another branch inside the same `ResponseStream` model and translates ACP events into the +same legacy Warp response events. Provider choice, provider credentials, ACP session state, retry +policy, network recovery, cancellation, telemetry remnants, and UI event emission therefore meet in +one coordinator. + +The problem is not that translations exist. Every integration needs one boundary translation. The +problem is that Warp's former server protocol is acting as Galaxy's domain model, so every new +provider needs translations on both sides of a protocol Galaxy does not own. + +## Target architecture + +```text +GalaxyUI views and models + | + v +Galaxy application services + conversation / permissions / local persistence / provider registry + | + v +galaxy_agent_core + TurnRequest, Message, Content, ToolSpec, AgentEvent, Usage, StopReason, AgentError + AgentRuntime trait -> AgentEventStream + | + +-------------------------+ + | | + v v +galaxy_agent_rig galaxy_agent_acp + OpenAI-compatible ACP subprocess/session + LiteLLM/Ollama/LM Studio ACP event adapter + AWS Bedrock Galaxy tool bridge + Rig/MCP tool bridge + | | + +------------+------------+ + v + explicit egress policy +``` + +### `galaxy_agent_core` + +This crate is the dependency rule that makes the refactor possible. It owns only stable Galaxy +concepts: + +- ordered conversation messages with text, images, reasoning, tool calls, and tool results; +- model/provider identifiers that do not encode a particular SDK type; +- dynamic tool descriptions and JSON schemas; +- turn events such as text delta, reasoning delta, tool proposed, permission requested, tool + started, tool completed, usage updated, turn stopped, and failure; +- cancellation and live steering control; +- structured stop and error classification; +- the `AgentRuntime` interface. + +It must not depend on GalaxyUI, `warp_multi_agent_api`, Rig, an AWS SDK, ACP, GraphQL, or app +persistence. + +### `galaxy_agent_runtime` + +This application-service layer owns: + +- resolving a conversation's backend once per conversation; +- resolving a model to a configured provider endpoint; +- building system and project context; +- conversation history and summarization policy; +- tool registration and permission policy; +- retry, cancellation, steering, and recovery semantics; +- mapping runtime events to local persistence and UI-facing models. + +The current UI can initially be kept alive with a temporary adapter from `AgentEvent` to legacy +`warp_multi_agent_api::ResponseEvent`. That adapter is a migration device, not the final boundary. + +### `galaxy_agent_rig` + +Rig becomes the implementation for provider-backed conversations. The version evaluated for this +plan is Rig 0.40.0. When introduced, it must be pinned exactly until its documented breaking-change +cadence settles for Galaxy. + +Rig is a good fit for the provider side because it already defines a canonical completion request, +provider implementations, streaming content/tool events, model history, typed tools, hooks, MCP via +`rmcp`, and a multi-turn agent runner. The integration should use those abstractions rather than +copying Rig's internal provider request structs into Galaxy types. + +Provider coverage for the first migration: + +| Galaxy provider | Rig implementation | Notes | +|---|---|---| +| LiteLLM / generic OpenAI-compatible | Rig OpenAI-compatible client | Custom base URL and key; preserve per-model endpoint routing. | +| Ollama / LM Studio | OpenAI-compatible or Rig provider adapter | Treat as explicit local/LAN endpoints. | +| AWS Bedrock | `rig-bedrock` through the Rig facade | Preserve profile, static credential, SSO, region, and inference-profile behavior through a focused compatibility audit. | +| MCP tools | Rig `rmcp` tool server/client support | Reuse existing Galaxy MCP lifecycle where it is stronger; bridge tools at one boundary. | + +Rig's documented integrations cover model providers and MCP, not Agent Client Protocol. ACP should +not be forced through Rig. It is a peer implementation of `AgentRuntime`. + +### Tool execution and permissions + +Galaxy must continue to own the user-facing tool lifecycle. A model framework may drive the loop, +but it must not silently bypass Galaxy's permission cards or execute a shell/file operation before +the UI can authorize it. + +The Rig adapter will therefore: + +1. register thin Rig tools that delegate into Galaxy's tool executor; +2. attach a Rig agent hook to observe and fail closed on tool calls; +3. emit a Galaxy `ToolProposed` or `PermissionRequested` event before execution; +4. await a permission decision when required; +5. execute through the existing Galaxy tool implementation; +6. return the result to Rig and emit correlated start/result events using a stable Galaxy call ID. + +Rig 0.40's streamed model-tool-call, tool-execution-start, tool-result, hooks, request patching, and +fail-closed flow semantics are useful here, but contract tests must prove the exact ordering Galaxy's +UI expects. + +### `galaxy_agent_acp` + +The existing `crates/acp` runtime has useful protocol/session work and should be retained initially. +Its application adapter should move out of `ResponseStream` and emit `AgentEvent` directly. + +ACP-specific capabilities remain visible in backend metadata: + +- session load/new-session support; +- agent authentication methods; +- configuration discovery; +- filesystem and terminal capability negotiation; +- permission requests; +- prompt steering and cancellation. + +Provider model controls should not appear for ACP-owned conversations because the external agent +owns its model and authentication. + +## Local data architecture + +"Galaxy Drive" becomes a local content library, not a renamed cloud sync client. Existing UI for +rules, profiles, notebooks, workflows, environment-variable collections, and MCP configurations can +be preserved while its storage service is replaced. + +The target repository interface is local and revisioned: + +```text +LocalObjectRepository + list(kind, scope) + get(id) + create(object) + update(id, expected_revision, object) + delete(id) + watch(kind/scope) +``` + +SQLite remains the default store. Filesystem import/export can be layered on later. The UI should +not know whether an object used to be a `CloudObject`; it should receive local object IDs and local +repository events. + +Migration must preserve existing local rows before removing cloud-shaped schemas. A temporary +compatibility repository can read the current tables without starting `SyncQueue`, `UpdateManager`, +GraphQL, or RTC listeners. + +## Network and trust model + +Every runtime network path belongs to one of these classes: + +| Class | Default | Examples | +|---|---|---| +| Inherited product service | Forbidden | Warp auth, GraphQL, RTC, session sharing, Oz, telemetry, remote logs, remote flags. | +| Configured model provider | Allowed only when selected | Bedrock, LiteLLM, OpenAI-compatible endpoint, Ollama on another host. | +| User-initiated remote development | Allowed with visible intent | SSH, Git fetch/push, remote MCP, provider/model discovery. | +| Agent network tool | Disabled until enabled by policy | Web fetch/search, HTTP MCP tools, computer-use browser actions. | +| Product maintenance | Separate explicit policy | Update checks, release download, optional LSP/runtime downloads. | + +The OSS binary must never contain a usable inherited Warp endpoint. UI hiding and feature flags do +not satisfy this requirement by themselves. + +## Migration sequence + +### Phase 0 — Freeze, inventory, and close implicit egress + +- Maintain the UI ledger and classify every registered action/menu/settings route. +- Disable inherited Warp service endpoints in OSS and fail closed on auth exchange. +- Rotate and revoke the signing credential currently tracked in migration documentation, remove it + from the working tree, and purge it from repository history in a coordinated security change. +- Add an automated forbidden-domain test for shipped configuration and runtime network fixtures. +- Mark old Bedrock-only architecture documents as historical. +- Stop porting upstream cloud, billing, telemetry, or Oz features during this migration. + +Exit condition: a clean OSS launch and normal local terminal use cannot address a Warp-operated +runtime endpoint, even if a stale UI action is triggered. + +### Phase 1 — Introduce the Galaxy agent domain seam + +- [x] Add `galaxy_agent_core` with requests, messages, events, errors, turn control, and the + `AgentRuntime` trait. +- [x] Add contract tests using a deterministic fake runtime. +- [x] Move the provider-neutral conversation message and tool-definition types out of the app + crate, retaining only a temporary compatibility re-export. +- [x] Route provider request startup through a named `ProviderRuntime` boundary so the controller + no longer calls the provider generator directly. +- [ ] Map legacy response events to `AgentEvent` and make the compatibility runtime implement + `AgentRuntime`; keep the inverse UI adapter until consumers migrate. +- [ ] Remove provider-specific fields such as `bedrock_message_history` from UI-level + `RequestParams`. + +The compatibility path still produces a Warp protobuf stream for the UI, but Rig-backed providers +implement `AgentRuntime` and cross that protocol boundary only in the app-owned UI adapter. Legacy +providers remain behind `ProviderRuntime` while their migrations continue. + +Exit condition: the conversation controller selects an `AgentRuntime` and does not match directly on +Bedrock/OpenAI/ACP configuration. + +### Phase 2 — First Rig vertical slice: OpenAI-compatible streaming + +- [x] Pin `rig-core` 0.40.0 and implement one explicit OpenAI-compatible provider. +- [x] Support text, reasoning where available, cancellation, stop reason, usage, and persisted + history. +- [x] Route any model entry with `use_rig = true` through Rig while leaving unmarked models on the + compatibility path. +- [x] Test Rig's real Chat Completions SSE parser against normalized events, plus the UI stop/usage + compatibility mappings. + +Initial opt-in example: + +```toml +[ai.openai] +enabled = true + +[[ai.providers]] +name = "LiteLLM (ai.ryserve.net)" +base_url = "https://ai.ryserve.net/v1" +api_key = "REPLACE_WITH_LOCAL_KEY" + +[[ai.providers.models]] +model_id = "codex-gpt-5.6-sol-xhigh" +display_name = "Codex GPT-5.6 SOL (xhigh)" +context_size = 200000 +provider = "openai" +use_rig = true +supports_system_messages = false +``` + +Phase 2 intentionally does not expose Galaxy's legacy tool list to Rig. That ownership moves as a +unit in Phase 3; until then, the opt-in slice validates text conversation streaming without two +competing tool executors. + +Exit condition: a LiteLLM or local OpenAI-compatible conversation streams through Rig without +`warp_multi_agent_api::Request` on the provider side. + +### Phase 3 — Tools, permissions, MCP, and multi-turn behavior + +- Bridge the core Galaxy tools into Rig. +- Preserve permission cards, denial, cancellation, parallel-call ordering, and error visibility. +- Bridge current MCP tools through Rig's `rmcp` support or a single Galaxy tool-server adapter. +- Port loop prevention and unknown-tool handling to domain-level policies. + +Exit condition: representative read, edit, shell, MCP, denial, and failure flows pass integration +tests without provider-specific UI code. + +### Phase 4 — Bedrock through Rig + +- Implement Bedrock client construction and model resolution through `rig-bedrock`. +- Compare request behavior for system prompts, images, tool schemas, cache controls, reasoning, + inference profiles, token usage, and context limits. +- Keep a short-lived compatibility fallback for unsupported Bedrock behavior, measured by tests. +- Delete custom Bedrock translation code only after parity is proven. + +Exit condition: supported Bedrock models use the same `AgentRuntime` event contract as +OpenAI-compatible models. + +### Phase 5 — ACP convergence + +- Move ACP launch/session/transport control behind `galaxy_agent_acp`. +- Translate ACP events directly to `AgentEvent`. +- Remove ACP branching from the UI response stream model. +- Keep ACP-specific settings and capability disclosure, but share transcript and permission UI. + +Exit condition: the controller cannot distinguish ACP from Rig except through backend capability +metadata. + +### Phase 6 — Local Galaxy Drive and identity removal + +- Introduce `LocalObjectRepository` over existing SQLite data. +- Move rules, profiles, notebooks, workflows, env collections, and MCP configs to the local service. +- Replace account/workspace ownership with local scopes. +- Remove auth, teams, billing, referral, cloud sync, GraphQL, RTC, sharing, and remote-control UI. + +Exit condition: none of the kept content flows require `AuthState`, `CloudModel`, `UpdateManager`, +`SyncQueue`, or a server ID. + +### Phase 7 — UI untangling + +- Split coordinator files along the flow boundaries in the UI ledger. +- Views render state and emit intent; application services perform persistence and runtime work. +- Reuse existing shared button themes and theme tokens. +- Remove unreachable modals/actions instead of continuing to hide them behind flags. + +Exit condition: every kept flow has an owner, a state model, a service boundary, and automated +coverage for success, failure, cancellation, and restore where applicable. + +### Phase 8 — Delete the legacy protocol center + +- Remove `warp_multi_agent_api` from UI/controller and persistence code. +- Delete the custom OpenAI/Bedrock request and response translators replaced by Rig. +- Delete no-op telemetry schemas/macros after call sites no longer depend on them. +- Remove Warp server, GraphQL, Firebase, cloud-object, Oz, billing, and referral crates from default + and then workspace builds when no retained feature needs them. + +Exit condition: `rg` finds no runtime dependency from the shipped app to Warp service code or Warp's +multi-agent wire protocol. + +## Verification gates + +Every phase must keep these checks green: + +- formatting and Clippy for changed crates; +- unit tests for the new domain/runtime layer; +- deterministic transcript contract tests; +- integration coverage for terminal and agent flows touched by the phase; +- a local-only egress test using request interception or a denied-network test environment; +- restart/restore tests for conversations and local content; +- no secret or prompt contents in logs unless a user explicitly enables a diagnostic mode. + +Provider parity tests should compare semantic events, not provider JSON snapshots alone. The stable +contract is what the UI and persistence observe. + +## Decisions + +| Decision | Choice | +|---|---| +| Provider abstraction | Rig behind a Galaxy-owned runtime interface. | +| ACP relationship | Peer runtime, not a Rig provider. | +| UI compatibility during migration | Temporary `AgentEvent` to legacy response-event adapter. | +| Long-term UI model | Galaxy domain events only. | +| Galaxy Drive | Local SQLite-backed content library. | +| Login/account | Remove from OSS product flows. | +| Telemetry/remote logs/remote flags | Remove, not merely default-off. | +| SSH and remote Git | Keep as explicit user-initiated remote development boundaries. | +| Web/network agent tools | Disabled by default and permission-visible. | +| Rig dependency | Exact version pin with upgrade contract tests. | + +## Immediate next vertical slice + +After the Phase 0 egress guard and UI ledger are verified, the next implementation change is a small +`galaxy_agent_core` crate plus a legacy adapter. It should move only provider-neutral message/event +types and runtime selection. Adding Rig before this seam would couple the UI to a new framework and +repeat the current mistake with a different name. diff --git a/plans/galaxy-refactor.md b/plans/galaxy-refactor.md index f331d800..1ac6f0b6 100644 --- a/plans/galaxy-refactor.md +++ b/plans/galaxy-refactor.md @@ -1,5 +1,10 @@ # Galaxy Refactor — Implementation Plan +> **Superseded:** This Bedrock-only plan no longer represents the product direction. +> Use [`galaxy-local-first-rig.md`](galaxy-local-first-rig.md) and +> [`ui-flow-inventory.md`](ui-flow-inventory.md). This file remains as historical +> context so completed work and earlier decisions are not silently lost. + > **Created:** 2026-05-07 > **Status:** In Progress > **Current Phase:** Phase 1 — Crate Renaming diff --git a/plans/ui-flow-inventory.md b/plans/ui-flow-inventory.md new file mode 100644 index 00000000..7b391734 --- /dev/null +++ b/plans/ui-flow-inventory.md @@ -0,0 +1,129 @@ +# Galaxy UI Flow Inventory + +> **Started:** 2026-08-04 +> **Status:** First-pass surface classification; action-level trace audit in progress +> **Architecture:** [`galaxy-local-first-rig.md`](galaxy-local-first-rig.md) + +## How this ledger is used + +This is the source of truth for deciding what Galaxy keeps, rebuilds, or removes. A directory name is +not a product decision. Each user intent is traced from every entry point through state, persistence, +runtime/network dependencies, and rendered outcomes. + +Audit sources include: + +- root/onboarding states in `app/src/root_view.rs`; +- registered workspace actions in `app/src/workspace/action.rs`; +- app menus, command palette, keybindings, context menus, URI handlers, and toolbar buttons; +- settings navigation and widgets under `app/src/settings_view`; +- left/right panels and terminal/agent input modes; +- existing integration-test modules under `app/src/integration_testing` and `crates/integration`; +- feature flags that make otherwise hidden flows reachable in OSS/dogfood builds. + +For each kept or rebuilt flow, completion means checking: + +- [ ] every mouse, keyboard, command-palette, menu, URI, startup, and programmatic entry point; +- [ ] empty, loading, success, partial-stream, denied, cancelled, offline, error, and retry states; +- [ ] close/reopen, restart, and session-restore behavior where state persists; +- [ ] focus, hover, accessibility, and context-flag behavior; +- [ ] local writes and migration behavior; +- [ ] every network destination and the user intent that authorizes it; +- [ ] unit and integration coverage; +- [ ] removal of obsolete actions, flags, settings, assets, and service code after migration. + +Status values: + +- **Keep/local:** core behavior remains and must require no service. +- **Keep/explicit:** remote behavior remains only behind explicit user/admin configuration or action. +- **Rebuild:** preserve the intent/UI value but replace its backing service or state model. +- **Remove:** the intent belongs to Warp's hosted product and should disappear completely. +- **Audit:** disposition or reachability still needs code/runtime validation. + +## Flow ledger + +| ID | Surface and user intent | Current coupling observed | Target disposition | Status | +|---|---|---|---|---| +| BOOT-01 | Launch app and reach a usable workspace | Root auth/onboarding state, server API provider, auth manager, cloud/update models | Launch directly into local workspace; provider setup is optional and non-blocking | Rebuild | +| BOOT-02 | First-run education and appearance setup | Agent onboarding, login slide, server `is_onboarded` state | Local onboarding focused on terminal mode, privacy boundary, and provider/ACP choices | Rebuild | +| BOOT-03 | Restore windows, tabs, panes, CWDs, and agent conversations | SQLite plus cloud-shaped conversation/object state | Local SQLite restore only | Keep/local | +| BOOT-04 | Sign in, sign out, reauth, SSO, anonymous user | Firebase/Warp auth and account UI | No account in OSS | Remove | +| WS-01 | Create, close, reorder, rename, pin, group, and color tabs | Workspace action/controller mega-file | Preserve behavior; split state ownership later | Keep/local | +| WS-02 | Split, close, focus, rename, maximize, and navigate panes | PaneGroup, Workspace, terminal model | Preserve | Keep/local | +| WS-03 | Save/launch tab configurations and worktrees | Local TOML/repo plus some telemetry/cloud vocabulary | Preserve as local templates | Keep/local | +| WS-04 | Open settings, resource center, logs, and diagnostic panes | Mixed local and server/account actions | Preserve local pages; remove hosted links/actions | Rebuild | +| TERM-01 | Run shell commands and view structured blocks | Terminal/UI core | Preserve | Keep/local | +| TERM-02 | Search command history, blocks, commands, files, and palettes | SQLite/local index plus cloud object sources | Preserve local sources; remove hosted sources | Rebuild | +| TERM-03 | Use SSH, remote shells, and Wormhole/warpification | Remote host and remote-server components | Keep only explicit remote-host behavior; audit branding and hidden service calls | Keep/explicit | +| TERM-04 | Share a terminal/session by URL or QR code | Warp session-sharing service | No hosted replacement in local-first scope | Remove | +| TERM-05 | Sync terminal input across panes/tabs | Local workspace state | Preserve | Keep/local | +| AGENT-01 | Start an agent conversation in a tab/pane | Blocklist controller, Warp proto request, provider/ACP branch | Route through `AgentRuntime` | Rebuild | +| AGENT-02 | Select provider, model, profile, and context limits | LLM preferences, Bedrock/OpenAI settings, ACP special cases | Unified provider registry; capability-aware controls | Rebuild | +| AGENT-03 | Compose prompts with files, selections, images, rules, and project context | Context chips, cloud-shaped rules, provider-specific request fields | Galaxy domain content/context builder | Rebuild | +| AGENT-04 | Watch text, reasoning, status, usage, and stop state stream | Provider translators emit Warp response events | Render `AgentEvent` stream | Rebuild | +| AGENT-05 | Review/approve/deny shell, file, MCP, and other tool calls | Blocklist action model and permissions; ACP has a parallel policy | One Galaxy tool/permission lifecycle shared by Rig and ACP | Rebuild | +| AGENT-06 | Cancel, interrupt, queue, send-now, or steer a running turn | ResponseStream/PendingResponseStreams and ACP steering | Provider-neutral turn control | Rebuild | +| AGENT-07 | Rename, pin, resume, fork, summarize, rewind, or delete conversations | SQLite plus server/cloud conversation vocabulary | Preserve meaningful local operations; remove cloud handoff/link actions | Rebuild | +| AGENT-08 | Inspect context usage, costs, and progressive summary | Bedrock-specific history fields and usage mapping | Provider-neutral usage; cost shown only when pricing is known/configured | Rebuild | +| AGENT-09 | Spawn and inspect child agents/orchestration | Warp MAA task schema, blocklist orchestration, some cloud assumptions | Defer until single-agent Rig tools are stable; local-only implementation | Audit | +| AGENT-10 | Start/restore an ACP-backed conversation | ACP runtime + separate ResponseStream branch | `galaxy_agent_acp` peer runtime with shared transcript and permissions | Rebuild | +| AGENT-11 | Detect/manage CLI agents and notifications | Agent SDK, Codex/OpenCode/Claude/Gemini harness/plugin code | Keep only ACP configuration and explicitly requested local integrations; remove Warp plugin cruft | Audit | +| AGENT-12 | Run Oz/cloud/ambient/scheduled agents and hand off local/cloud work | Agent SDK, cloud environments, Warp APIs, RTC | Hosted intent is out of scope | Remove | +| AGENT-13 | Configure/use MCP servers and resources | Local files, OAuth, managed/server MCP, tool execution | Keep local/explicit remote MCP; remove managed Warp gallery/secrets dependencies | Rebuild | +| AGENT-14 | Create/use global and project rules and skills | CloudModel AIFacts plus local rule/skill files | Local repository/filesystem only | Rebuild | +| AGENT-15 | Use voice input/transcription | Local capture plus Warp transcription endpoint or provider assumptions | Keep only with an explicit local/configured transcription backend | Audit | +| CODE-01 | Browse project files and global search | Local filesystem/index plus remote indexing branches | Preserve local; remote only for explicit SSH session | Keep/local | +| CODE-02 | Edit files with LSP completion, diagnostics, actions, rename, and signature help | Local filesystem/LSP/runtime downloads | Preserve; downloads are explicit product-maintenance egress | Keep/local | +| CODE-03 | Review local Git diffs, comments, stage/revert, commit | Local Git plus optional remote/GitHub models | Preserve local Git review | Keep/local | +| CODE-04 | Fetch PR metadata, push, or authenticate GitHub | Git/GitHub/server integration paths | Keep ordinary explicit Git operations; remove Warp-mediated GitHub auth | Rebuild | +| DRIVE-01 | Open Galaxy Drive/content library and navigate folders | Drive UI backed by CloudModel/UpdateManager/GraphQL | Local content library over SQLite | Rebuild | +| DRIVE-02 | Create/edit/import/export notebooks | Cloud object ownership/sync around useful local editors | Preserve editor; replace repository | Rebuild | +| DRIVE-03 | Create/edit/run/import/export workflows | Cloud object ownership/sync around useful local runner/UI | Preserve runner/editor; replace repository | Rebuild | +| DRIVE-04 | Manage environment-variable collections and external secrets | Cloud objects, server-managed secrets, local execution | Local encrypted/OS-keychain-backed storage; never cloud sync | Rebuild | +| DRIVE-05 | Manage profiles, rules, prompts, and MCP objects | Cloud object polymorphism | Local typed repositories | Rebuild | +| DRIVE-06 | Share objects, team folders, team roles, and sync conflicts | Warp cloud/team services | No hosted replacement in current scope | Remove | +| SET-01 | Change appearance, fonts, themes, terminal behavior, keyboard shortcuts | Local settings plus some cloud preference sync | Local settings only | Keep/local | +| SET-02 | Configure AI providers, models, profiles, ACP, MCP, rules, and experiments | One 8,500-line page with provider/hosted modes interleaved | Split by intent and capability; remove hosted modes | Rebuild | +| SET-03 | Configure privacy, telemetry, crash reporting, and cloud storage | No-op telemetry plus hosted-setting vocabulary | Replace with a read-only local-first network/privacy status page | Rebuild | +| SET-04 | Teams, billing, usage plans, referrals, upgrades | Warp account/services | Remove | Remove | +| SET-05 | About, update check, release notes, diagnostics | Local info plus remote release/service URLs | Keep; network operations separately disclosed/configured | Keep/explicit | +| NET-01 | Emit telemetry, analytics, remote logs, or crash reports | Most send macros are no-op, but schemas and hooks remain | Delete runtime path and eventually schemas/call sites | Remove | +| NET-02 | Discover models and call inference | Bedrock SDK, OpenAI client, provider routing map | Rig provider registry; selected provider only | Rebuild | +| NET-03 | Open web links, web fetch/search, browser/computer use | External URLs and agent tools | Explicit user action/policy with visible destination class | Keep/explicit | +| NET-04 | Check/download updates, fonts, LSPs, runtimes, or plugins | Several independent download paths, including inherited server-root usage | Audit each destination; allow only signed/pinned, explicit maintenance paths | Audit | +| UI-01 | Use command palette, menus, keybindings, context menus, toolbar, and URI routes | Hundreds of action variants include both local and hosted intents | Retain as entry-point layer; remove every obsolete registered action | Audit | +| UI-02 | Receive notifications, toasts, modals, and banners | Local status mixed with billing/login/Oz/agent marketing | Preserve local status; remove hosted/marketing state machines | Rebuild | +| UI-03 | Accessibility, focus, mouse/hover, themes, and responsive panels | GalaxyUI view state | Preserve and cover while splitting views | Keep/local | + +## First reachability findings + +1. The OSS binary enabled dogfood flags, including ACP and multiple experimental local/remote UI + features. Audit cannot assume a `DOGFOOD_FLAGS` item is unreachable in OSS. +2. Telemetry send macros and collectors are no-ops, but thousands of telemetry event definitions and + call-site dependencies remain architectural glue. +3. The settings sidebar exposes Agents, Code, Appearance, Features, Keyboard shortcuts, Wormhole, + Galaxy Drive, Privacy, About, and optionally Galaxy Control. The Agents page combines Galaxy + Agent, Profiles, MCP servers, Knowledge, third-party CLI agents, Bedrock, OpenAI/LiteLLM, and + Experiments. +4. The left panel combines Project Explorer, Global Search, Galaxy Drive, and Conversation List. + Code Review is a separate right panel. This is a useful UI shell, but both panels currently import + cloud/telemetry vocabulary. +5. `WorkspaceAction` still registers login, upgrade, sharing, team Drive creation, cloud handoff, + cloud-agent setup, Oz install/launch, ambient agents, and other hosted actions alongside core tab, + pane, terminal, editor, and local-agent actions. + +## Audit order + +The action-level audit proceeds in this order because each later surface depends on the earlier +state boundary: + +1. boot/onboarding and network initialization; +2. workspace/tabs/panes and session restoration; +3. terminal input, blocks, history, and search; +4. provider-backed agent conversation happy path; +5. tool permissions, errors, cancellation, queueing, and restore; +6. ACP parity; +7. local content library and settings; +8. editor/code review/remote development; +9. removal sweep across menus, palette, URI routes, banners, modals, flags, and tests. + +The ledger is complete only when every user-visible action variant has a flow ID or has been deleted.