From 0cb56b12c67704809eaf31fb1eaeed5c2f5e45ff Mon Sep 17 00:00:00 2001 From: Anthony Ronning <101225832+AnthonyRonning@users.noreply.github.com> Date: Tue, 27 Oct 2026 01:54:24 +0000 Subject: [PATCH 1/3] Add managed local API controls to Maple Agent --- apps/maple-agent/Cargo.lock | 3 + apps/maple-agent/README.md | 12 +- apps/maple-agent/app/Cargo.toml | 5 +- apps/maple-agent/app/src/backend.rs | 80 +- apps/maple-agent/app/src/backend/local.rs | 21 +- apps/maple-agent/app/src/backend/local_api.rs | 160 +++ .../app/src/backend/local_api_tests.rs | 68 + apps/maple-agent/app/src/desktop.rs | 103 +- .../app/src/desktop/local_api_tests.rs | 374 ++++++ apps/maple-agent/app/src/host.rs | 109 +- .../app/src/host/local_api_tests.rs | 517 ++++++++ apps/maple-agent/app/src/host/protocol.rs | 4 + apps/maple-agent/app/src/local_api.rs | 1148 +++++++++++++++++ apps/maple-agent/app/src/main.rs | 32 +- apps/maple-agent/app/src/remote/control.rs | 8 +- apps/maple-agent/app/src/ui/chat/mod.rs | 3 +- apps/maple-agent/app/src/ui/settings.rs | 30 +- .../app/src/ui/settings/local_api.rs | 715 ++++++++++ .../app/src/ui/settings/navigation.rs | 16 + .../crates/maple-agent/src/agent/types.rs | 3 + apps/maple-agent/docs/local-api.md | 89 ++ 21 files changed, 3460 insertions(+), 40 deletions(-) create mode 100644 apps/maple-agent/app/src/backend/local_api.rs create mode 100644 apps/maple-agent/app/src/backend/local_api_tests.rs create mode 100644 apps/maple-agent/app/src/desktop/local_api_tests.rs create mode 100644 apps/maple-agent/app/src/host/local_api_tests.rs create mode 100644 apps/maple-agent/app/src/local_api.rs create mode 100644 apps/maple-agent/app/src/ui/settings/local_api.rs create mode 100644 apps/maple-agent/docs/local-api.md diff --git a/apps/maple-agent/Cargo.lock b/apps/maple-agent/Cargo.lock index 54670da25..854c4af72 100644 --- a/apps/maple-agent/Cargo.lock +++ b/apps/maple-agent/Cargo.lock @@ -5829,10 +5829,13 @@ dependencies = [ "dirs 6.0.0", "env_logger", "futures-lite", + "futures-util", "gpui", "gpui_platform", "hostname", "httparse", + "hyper", + "hyper-util", "image", "iroh", "libc", diff --git a/apps/maple-agent/README.md b/apps/maple-agent/README.md index d887e7bea..4153aec75 100644 --- a/apps/maple-agent/README.md +++ b/apps/maple-agent/README.md @@ -504,6 +504,16 @@ compatibility](docs/local-host.md) before configuring an ACP launcher. ### `maple-agent proxy` +The desktop also offers **Settings → Local API**. It starts this computer's +loopback endpoint in the shared local host, including while a remote computer +is selected. Create a key under API keys, enable Local API, then configure your +client with the displayed `/v1` base URL and that key. See +[Local API ownership and lifecycle](docs/local-api.md). + +The command below remains an independent foreground server. It cannot share a +port with an enabled desktop listener; an occupied port is reported without +stopping or adopting the other server. + ``` --host HOST bind address (default 127.0.0.1, env MAPLE_PROXY_HOST) --port PORT bind port (default 8080, env MAPLE_PORT) @@ -527,7 +537,7 @@ window and its display libraries: | --- | --- | | `desktop` | The gpui window. Without it the binary is headless. | | `acp` | `maple-agent acp` and `maple_agent::acp`. | -| `proxy` | `maple-agent proxy`. | +| `proxy` | `maple-agent proxy` and the local host's optional API service. | ```sh cargo build --release -p maple-agent-app --no-default-features --features acp diff --git a/apps/maple-agent/app/Cargo.toml b/apps/maple-agent/app/Cargo.toml index dc5ff35a1..2cf384673 100644 --- a/apps/maple-agent/app/Cargo.toml +++ b/apps/maple-agent/app/Cargo.toml @@ -35,7 +35,7 @@ desktop = [ # `maple-agent acp`: the Agent Client Protocol server on stdio. acp = ["maple-agent/acp"] # `maple-agent proxy`: the OpenAI-compatible HTTP endpoint. -proxy = ["dep:maple-proxy", "dep:axum", "dep:tower-http"] +proxy = ["dep:maple-proxy", "dep:axum", "dep:tower-http", "dep:hyper", "dep:hyper-util", "dep:futures-util"] [dependencies] dirs = "6" @@ -86,6 +86,9 @@ pake-cpace = { git = "https://github.com/1Password/cpace", rev = "7e2e42c7d35996 zeroize = { workspace = true } axum = { version = "0.8", optional = true } tower-http = { version = "0.6", features = ["cors"], optional = true } +hyper = { version = "1", features = ["http1", "server"], optional = true } +hyper-util = { version = "0.1", features = ["tokio", "service"], optional = true } +futures-util = { workspace = true, optional = true } # Used only to ask the compositor whether it draws window decorations; see # `ui::decorations`. gpui cannot answer that question. diff --git a/apps/maple-agent/app/src/backend.rs b/apps/maple-agent/app/src/backend.rs index 10c6c78d7..4e51325fc 100644 --- a/apps/maple-agent/app/src/backend.rs +++ b/apps/maple-agent/app/src/backend.rs @@ -14,6 +14,8 @@ pub use attachments::AttachmentUploadRequest; pub use attachments::ImportAttachmentRequest; #[cfg(feature = "desktop")] pub use paging::*; +#[cfg(all(test, feature = "proxy"))] +mod local_api_tests; mod mcp; use crate::host::{RemoteClient, protocol::Request}; @@ -60,6 +62,40 @@ enum BackendClient { } impl AgentBackend { + #[cfg(feature = "proxy")] + fn require_local_api(&self) -> Result<(), String> { + self.ensure_active()?; + if self.remote_host.is_some() { + return Err("Local API settings require this computer's local host".into()); + } + Ok(()) + } + + #[cfg(feature = "proxy")] + pub async fn get_local_api_status(&self) -> Result { + self.require_local_api()?; + self.call("get_local_api_status", ()).await + } + + #[cfg(feature = "proxy")] + // Recovery clients can stop without reading encrypted preferences. The + // normal settings switch uses the durable configure operation instead. + #[allow(dead_code)] + pub async fn stop_local_api(&self) -> Result { + self.require_local_api()?; + self.call("stop_local_api", ()).await + } + + #[cfg(feature = "proxy")] + pub async fn set_local_api_config( + &self, + user_id: &str, + config: crate::local_api::LocalApiConfig, + ) -> Result { + self.require_local_api()?; + self.call("set_local_api_config", (user_id, config)).await + } + #[cfg(feature = "acp")] pub async fn get_acp_config( &self, @@ -608,6 +644,19 @@ async fn dispatch_special( "restore_auth" => { serde_json::to_value(backend.validate_persisted_auth().await).map_err(|e| e.to_string()) } + #[cfg(feature = "proxy")] + "get_local_api_status" => serde_json::to_value(backend.get_local_api_status().await?) + .map_err(|_| "Could not encode Local API status".into()), + #[cfg(feature = "proxy")] + "stop_local_api" => serde_json::to_value(backend.stop_local_api().await?) + .map_err(|_| "Could not encode Local API status".into()), + #[cfg(feature = "proxy")] + "set_local_api_config" => { + let (user_id, config): (String, crate::local_api::LocalApiConfig) = + serde_json::from_value(request.params).map_err(invalid)?; + serde_json::to_value(backend.set_local_api_config(&user_id, config).await?) + .map_err(|_| "Could not encode Local API status".into()) + } "cancel_oauth" => { backend.cancel_oauth(); Ok(serde_json::Value::Null) @@ -672,6 +721,13 @@ pub(crate) async fn dispatch( backend: &LocalBackend, request: Request, ) -> Result { + #[cfg(feature = "proxy")] + if matches!( + request.method.as_str(), + "get_local_api_status" | "stop_local_api" + ) { + return dispatch_inner(backend, request).await; + } // Account switches cannot interleave with an operation admitted for the // previous account. Restoring that same account is a reader: its network // validation must not prevent local history, configuration, or stop calls. @@ -691,11 +747,31 @@ pub(crate) async fn dispatch( | "oauth_complete" | "confirm_account_deletion" ) { + #[cfg(feature = "proxy")] + backend.local_api.stop().await; let _account = backend.account_gate.write().await; - dispatch_inner(backend, request).await + #[cfg(feature = "proxy")] + backend.local_api.stop().await; + #[cfg(feature = "proxy")] + let generation = backend.local_api.generation(); + let result = dispatch_inner(backend, request).await; + #[cfg(feature = "proxy")] + if result.is_ok() { + let _ = backend.resume_local_api(generation).await; + } + result } else { let _account = backend.account_gate.read().await; - dispatch_inner(backend, request).await + #[cfg(feature = "proxy")] + let resume = (request.method == "restore_auth").then(|| backend.local_api.generation()); + let result = dispatch_inner(backend, request).await; + #[cfg(feature = "proxy")] + if result.is_ok() + && let Some(generation) = resume + { + let _ = backend.resume_local_api(generation).await; + } + result } } diff --git a/apps/maple-agent/app/src/backend/local.rs b/apps/maple-agent/app/src/backend/local.rs index 422053b52..03a5946ca 100644 --- a/apps/maple-agent/app/src/backend/local.rs +++ b/apps/maple-agent/app/src/backend/local.rs @@ -35,6 +35,9 @@ use crate::hosted_oauth::{ATTEMPT_TIMEOUT, HandoffListener}; mod deletion; pub use deletion::AccountDeletionStatus; +#[cfg(feature = "proxy")] +#[path = "local_api.rs"] +mod local_api; #[path = "project_folders.rs"] mod project_folders; #[allow(unused_imports)] @@ -107,6 +110,12 @@ pub struct DictionaryPage { } pub struct LocalBackend { + #[cfg(feature = "proxy")] + pub(crate) local_api: crate::local_api::LocalApiManager, + #[cfg(feature = "proxy")] + local_api_settings_gate: tokio::sync::Mutex<()>, + #[cfg(feature = "proxy")] + local_api_control: std::sync::Mutex, store: Option>, pub(crate) remote_updates: Arc, row_reader: Option>, @@ -1053,6 +1062,8 @@ impl LocalBackend { } pub(crate) async fn forget_store_session(&self) -> Result<(), String> { + #[cfg(feature = "proxy")] + self.local_api.stop().await; self.service.finish_store_reset().await?; self.service.invalidate_store_session(); if let Some(user) = self.persisted_auth.forget_owner() { @@ -1138,6 +1149,12 @@ impl LocalBackend { .as_ref() .map(|store| Arc::new(maple_agent::store::RowReader::new(store.clone()))); Ok(Self { + #[cfg(feature = "proxy")] + local_api: crate::local_api::LocalApiManager::new(api_url.clone())?, + #[cfg(feature = "proxy")] + local_api_settings_gate: tokio::sync::Mutex::new(()), + #[cfg(feature = "proxy")] + local_api_control: std::sync::Mutex::new(local_api::Control::default()), remote_updates: Arc::new(crate::remote_updates::Broker::new(store.clone())), store, row_reader, @@ -1438,7 +1455,7 @@ impl LocalBackend { /// Validate the saved credentials with the server and install the /// session on success. Definitive rejections compare-clear the saved record. - pub(super) async fn validate_persisted_auth(&self) -> RestoreOutcome { + pub(crate) async fn validate_persisted_auth(&self) -> RestoreOutcome { let _restore = self.restore_lock.lock().await; if let Some(user_id) = self.saved_user_id() && self.auth.session_for(&user_id).await.is_ok() @@ -1521,6 +1538,8 @@ impl LocalBackend { /// Drop the live session and the persisted record. `revoke` also sends /// `POST /logout`; a deleted account has no session left to report. async fn clear_session(&self, user_id: &str, revoke: bool) -> Result<(), String> { + #[cfg(feature = "proxy")] + self.local_api.stop().await; let _account = self.restore_lock.lock().await; let pending = self.pending_deletion_status()?; if let Some(pending) = pending.as_ref() { diff --git a/apps/maple-agent/app/src/backend/local_api.rs b/apps/maple-agent/app/src/backend/local_api.rs new file mode 100644 index 000000000..f4689bee4 --- /dev/null +++ b/apps/maple-agent/app/src/backend/local_api.rs @@ -0,0 +1,160 @@ +//! Account-owned preferences; the independent HTTP service never borrows the +//! desktop SDK session or keeps account/store guards during inference streams. +use super::LocalBackend; +use crate::local_api::{LocalApiConfig, LocalApiStatus}; +use std::future::Future; + +const SETTING: &str = "local_api.config"; + +#[derive(Default)] +pub(super) struct Control { + epoch: u64, + suspended: bool, +} + +impl LocalBackend { + #[cfg(test)] + pub(crate) fn local_api_suspended_for_test(&self) -> bool { + self.local_api_control.lock().unwrap().suspended + } + pub(crate) fn local_api_control_epoch(&self) -> u64 { + self.local_api_control.lock().unwrap().epoch + } + pub async fn get_local_api_status(&self) -> Result { + Ok(self.local_api.status()) + } + + /// Runtime stop remains available even when the encrypted preferences cannot + /// be read. The settings switch uses set_local_api_config to persist Off. + pub async fn stop_local_api(&self) -> Result { + { + let mut control = self.local_api_control.lock().unwrap(); + control.epoch = control.epoch.wrapping_add(1); + control.suspended = true; + } + Ok(self.local_api.stop().await) + } + + pub(crate) async fn local_api_while_current( + &self, + generation: u64, + operation: impl Future>, + ) -> Result { + let mut changes = self.local_api.subscribe(); + tokio::pin!(operation); + loop { + if self.local_api.generation() != generation { + return Err("Local API operation was cancelled. Try again.".into()); + } + tokio::select! { + biased; + changed = changes.changed() => { + if changed.is_err() { return Err("Local API is unavailable".into()); } + } + result = &mut operation => return result, + } + } + } + + pub async fn set_local_api_config( + &self, + user_id: &str, + config: LocalApiConfig, + ) -> Result { + config.validate()?; + // Serialize durable settings edits, without preventing an independent + // Stop/logout from cancelling an authentication or disk wait. + let control_epoch = self.local_api_control_epoch(); + let generation = self.local_api.generation(); + let _settings = self.local_api_settings_gate.lock().await; + if self.local_api.generation() != generation { + return Err("Local API operation was cancelled. Try again.".into()); + } + let (store, account) = self.presentation_account(user_id)?; + // Only a deliberate On clears the pause. Off must stay latched even + // if its write fails and the previous durable setting still says On. + // Apply this before awaits so a later Stop cannot be overwritten. + { + let mut control = self.local_api_control.lock().unwrap(); + if control.epoch != control_epoch { + return Err("Local API operation was cancelled. Try again.".into()); + } + control.epoch = control.epoch.wrapping_add(1); + control.suspended = !config.enabled; + } + if config.enabled { + self.local_api_while_current(generation, async { + self.session_for(user_id).await.map(|_| ()) + }) + .await?; + } else { + // Turn off network access even if the subsequent durable write fails. + self.local_api.stop().await; + } + let generation = if config.enabled { + generation + } else { + self.local_api.generation() + }; + if self.local_api.generation() != generation { + return Err("Local API operation was cancelled. Try again.".into()); + } + let saved = config.clone(); + tokio::task::spawn_blocking(move || { + store + .set_account_setting( + account, + SETTING, + &serde_json::to_value(saved).map_err(|_| "Invalid Local API settings")?, + crate::settings::timestamp_ms(), + ) + .map_err(|_| "Could not save Local API settings".to_string()) + }) + .await + .map_err(|_| "Could not save Local API settings".to_string())??; + self.authorize_account(user_id)?; + self.local_api + .configure_if_current(config, user_id.to_owned(), generation) + .await + } + + /// Called only under the host's short admission/account guard after a + /// verified restore or sign-in. It never performs account validation itself. + pub(crate) async fn resume_local_api(&self, generation: u64) -> Result<(), String> { + if self.local_api.generation() != generation + || self.local_api_control.lock().unwrap().suspended + { + return Ok(()); + } + let Some(user_id) = self.saved_user_id() else { + return Ok(()); + }; + if self + .local_api_while_current(generation, async { + self.auth.session_for(&user_id).await.map(|_| ()) + }) + .await + .is_err() + { + return Ok(()); + } + let (store, account) = self.presentation_account(&user_id)?; + let config = tokio::task::spawn_blocking(move || { + let config = store + .account_setting(account, SETTING) + .map_err(|_| "Could not read Local API settings")?; + match config { + Some(value) => serde_json::from_value::(value) + .map_err(|_| "Saved Local API settings are invalid".to_string()), + None => Ok(LocalApiConfig::default()), + } + }) + .await + .map_err(|_| "Could not read Local API settings".to_string())??; + self.authorize_account(&user_id)?; + self.local_api + .configure_if_current(config, user_id, generation) + .await?; + Ok(()) + } +} diff --git a/apps/maple-agent/app/src/backend/local_api_tests.rs b/apps/maple-agent/app/src/backend/local_api_tests.rs new file mode 100644 index 000000000..1ae58c790 --- /dev/null +++ b/apps/maple-agent/app/src/backend/local_api_tests.rs @@ -0,0 +1,68 @@ +//! Local API controls must never become selected-computer operations. + +use super::*; +use crate::local_api::LocalApiConfig; + +fn fixture() -> Arc { + Arc::new(AgentBackend::for_test("http://127.0.0.1:1".into(), String::new()).unwrap()) +} + +#[test] +fn selected_remote_cannot_read_configure_or_stop_a_local_api() { + let controller = fixture(); + let runtime = controller.runtime_handle(); + runtime.block_on(async { + let remote = controller + .with_remote_connection("fixture-peer".into(), || async { + let mut backend = LocalBackend::new(controller.api_url.clone(), String::new())?; + backend.ui_fixture = true; + Ok(BackendClient::InProcess(Box::new(backend))) + }) + .await + .unwrap(); + + for result in [ + remote.get_local_api_status().await, + remote + .set_local_api_config( + "fixture-user", + LocalApiConfig { + enabled: true, + port: 0, + }, + ) + .await, + remote + .set_local_api_config( + "fixture-user", + LocalApiConfig { + enabled: false, + port: 0, + }, + ) + .await, + remote.stop_local_api().await, + ] { + let error = result.expect_err("a selected remote must not receive API controls"); + assert!( + error.to_ascii_lowercase().contains("local"), + "the facade must reject before validating a port or account: {error}" + ); + } + }); +} + +#[test] +fn installation_retirement_rejects_local_api_enable_before_account_or_port_validation() { + let controller = fixture(); + let runtime = controller.runtime_handle(); + controller.retire_after_update(); + let result = runtime.block_on(controller.set_local_api_config( + "fixture-user", + LocalApiConfig { + enabled: true, + port: 0, + }, + )); + assert_eq!(result.err(), Some(RETIRED_AFTER_UPDATE.into())); +} diff --git a/apps/maple-agent/app/src/desktop.rs b/apps/maple-agent/app/src/desktop.rs index 1e5fce1f4..758f21a1e 100644 --- a/apps/maple-agent/app/src/desktop.rs +++ b/apps/maple-agent/app/src/desktop.rs @@ -137,6 +137,14 @@ struct MapleApp { password_settings_refreshing: bool, persistence_status: maple_agent::store::PersistenceStatus, persistence_retrying: bool, + #[cfg(feature = "proxy")] + local_api_revision: u64, + #[cfg(feature = "proxy")] + local_api_host_epoch: u64, + #[cfg(feature = "proxy")] + local_api_refreshing: bool, + #[cfg(feature = "proxy")] + local_api_refresh_again: bool, pending_store_attention: std::collections::VecDeque, store_attention_open: bool, restore_window_geometry: bool, @@ -620,6 +628,14 @@ impl MapleApp { ) { let mut batch = Vec::new(); for event in events { + #[cfg(feature = "proxy")] + if matches!( + event, + maple_agent::agent::AgentServiceEvent::LocalApiChanged + ) { + self.refresh_local_api(cx); + continue; + } if let maple_agent::agent::AgentServiceEvent::PersistenceFailureChanged { status } = &event { @@ -633,6 +649,17 @@ impl MapleApp { ) { // A restarted host has a new in-memory revision baseline. self.persistence_status.revision = 0; + #[cfg(feature = "proxy")] + { + self.local_api_revision = 0; + self.local_api_host_epoch = self.local_api_host_epoch.wrapping_add(1); + self.local_api_refreshing = false; + self.local_api_refresh_again = false; + if let Screen::Settings(settings) = &self.screen { + settings.update(cx, |settings, cx| settings.local_api_host_reconnected(cx)); + } + self.refresh_local_api(cx); + } self.verify_current_store_state(cx); batch.push(event); } else if let maple_agent::agent::AgentServiceEvent::StoreStateChanged { state } = @@ -913,6 +940,50 @@ impl MapleApp { cx.notify(); } + #[cfg(feature = "proxy")] + fn refresh_local_api(&mut self, cx: &mut Context) { + if self.local_api_refreshing { + self.local_api_refresh_again = true; + return; + } + self.local_api_refreshing = true; + let backend = self.backend.clone(); + let epoch = self.store_load_epoch; + let host_epoch = self.local_api_host_epoch; + let bridge = ui::task::call( + &self.backend, + async move { + let status = backend.get_local_api_status().await?; + let store = backend + .local_control::("store_status", ()) + .await?; + Ok((status, store)) + }, + cx, + move |app, result, cx| { + if app.local_api_host_epoch != host_epoch { + return; + } + app.local_api_refreshing = false; + if app.store_load_epoch == epoch + && let Ok((status, store)) = result + && status.revision >= app.local_api_revision + { + app.local_api_revision = status.revision; + app.set_store_resident(store.keeps_resident(), cx); + if let Screen::Settings(settings) = &app.screen { + settings + .update(cx, |settings, cx| settings.set_local_api_status(status, cx)); + } + } + if std::mem::take(&mut app.local_api_refresh_again) { + app.refresh_local_api(cx); + } + }, + ); + ui::task::retain(&self.remote_bridges, bridge); + } + fn update_tray(&self) { #[cfg(any(target_os = "linux", target_os = "windows"))] if let Some(tray) = &self.tray { @@ -1935,6 +2006,8 @@ impl Render for MapleApp { } } +#[cfg(all(test, feature = "proxy"))] +mod local_api_tests; #[cfg(target_os = "macos")] mod native_quit; #[cfg(target_os = "macos")] @@ -1963,6 +2036,8 @@ struct DesktopStoreStatus { mode_change_notice: Option, #[serde(default)] persistence_status: maple_agent::store::PersistenceStatus, + #[serde(default)] + local_api_running: bool, } #[derive(Debug, Clone, serde::Deserialize)] @@ -2065,7 +2140,8 @@ impl DesktopStoreStatus { } fn keeps_resident(&self) -> bool { - (self.mode == "password" && self.state == maple_agent::store::StoreState::Open) + self.local_api_running + || (self.mode == "password" && self.state == maple_agent::store::StoreState::Open) || self.state == maple_agent::store::StoreState::Maintenance || self.setup_operation_id.is_some() } @@ -2078,6 +2154,10 @@ struct HostActivity { remote_controllers: usize, #[serde(default)] background_tasks: usize, + #[serde(default)] + local_api_running: bool, + #[serde(default)] + local_api_requests: usize, } impl HostActivity { fn needs_confirmation(&self) -> bool { @@ -2085,6 +2165,8 @@ impl HostActivity { || self.running_tasks > 0 || self.remote_controllers > 0 || self.background_tasks > 0 + || self.local_api_running + || self.local_api_requests > 0 } } @@ -2392,7 +2474,7 @@ impl Render for QuitPrompt { let (title, message, retry) = match &self.kind { QuitPromptKind::Confirm(activity) => ( "Stop this computer's Maple host?", - format!("Quitting stops {} running tasks, {} attached editors, {} remote controllers and {} background jobs. Maple waits for them to drain before closing.", activity.running_tasks, activity.acp_clients, activity.remote_controllers, activity.background_tasks), + format!("Quitting stops {} running tasks, {} attached editors, {} remote controllers and {} background jobs. It also stops this computer's local API and its {} active requests. Maple waits for them to drain before closing.", activity.running_tasks, activity.acp_clients, activity.remote_controllers, activity.background_tasks, activity.local_api_requests), false, ), QuitPromptKind::Failed(message) => ("Maple is still open", message.clone(), true), @@ -3178,6 +3260,14 @@ pub fn run() { password_settings_refreshing: false, persistence_status: maple_agent::store::PersistenceStatus::default(), persistence_retrying: false, + #[cfg(feature = "proxy")] + local_api_revision: 0, + #[cfg(feature = "proxy")] + local_api_host_epoch: 0, + #[cfg(feature = "proxy")] + local_api_refreshing: false, + #[cfg(feature = "proxy")] + local_api_refresh_again: false, pending_store_attention: Default::default(), store_attention_open: false, restore_window_geometry: true, @@ -3568,6 +3658,7 @@ mod tests { completion_notice: None, mode_change_notice: None, persistence_status: maple_agent::store::PersistenceStatus::default(), + local_api_running: false, }; assert!(status.keeps_resident()); status.state = StoreState::Locked; @@ -3905,6 +3996,14 @@ mod tests { password_settings_refreshing: false, persistence_status: Default::default(), persistence_retrying: false, + #[cfg(feature = "proxy")] + local_api_revision: 0, + #[cfg(feature = "proxy")] + local_api_host_epoch: 0, + #[cfg(feature = "proxy")] + local_api_refreshing: false, + #[cfg(feature = "proxy")] + local_api_refresh_again: false, pending_store_attention: Default::default(), store_attention_open: false, restore_window_geometry: false, diff --git a/apps/maple-agent/app/src/desktop/local_api_tests.rs b/apps/maple-agent/app/src/desktop/local_api_tests.rs new file mode 100644 index 000000000..1bc6a18a8 --- /dev/null +++ b/apps/maple-agent/app/src/desktop/local_api_tests.rs @@ -0,0 +1,374 @@ +//! Actual Settings clicks and private IPC across three disposable profiles. +//! Account validation, persistence and unrelated presentation APIs are fixture effects; +//! every profile uses the production local HTTP manager and router. + +use super::*; +use crate::host::{RemoteClient, ipc, protocol}; +use crate::local_api::{LocalApiConfig, LocalApiManager, LocalApiState}; +use maple_agent::agent::AgentServiceEvent; +use maple_agent::store::StoreState; +use serde_json::{Value, json}; +use std::{collections::VecDeque, sync::Mutex, time::Duration}; + +struct FixtureHost { + backend: Arc, + manager: LocalApiManager, + requests: Arc>>, + status_holds: Arc>>>, + runtime: Option, + _root: tempfile::TempDir, +} + +impl FixtureHost { + fn new(profile: &str, port: u16) -> Self { + #[cfg(unix)] + let root = tempfile::Builder::new() + .prefix("maple-api-ui-") + .tempdir_in("/tmp") + .unwrap(); + #[cfg(windows)] + let root = tempfile::tempdir().unwrap(); + let mut endpoint = ipc::Endpoint::for_test(root.path().to_path_buf()); + endpoint.profile = format!("local-api-ui-{profile}"); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .enable_all() + .build() + .unwrap(); + let requests = Arc::new(Mutex::new(Vec::new())); + let recorded = requests.clone(); + let status_holds = Arc::new(Mutex::new( + VecDeque::>::new(), + )); + let pending_holds = status_holds.clone(); + let (backend, manager) = runtime.block_on(async { + let manager = LocalApiManager::new("http://127.0.0.1:1".into()).unwrap(); + manager + .configure( + LocalApiConfig { + enabled: false, + port, + }, + "user".into(), + ) + .await + .unwrap(); + let service = manager.clone(); + let mut listener = endpoint.bind().unwrap(); + let (connection, accepted) = tokio::join!(endpoint.connect(), listener.accept()); + let connection = connection.unwrap(); + let accepted = accepted.unwrap(); + let welcome = protocol::Welcome { + version: protocol::VERSION, + pid: std::process::id(), + build: protocol::BUILD.into(), + user_id: Some("user".into()), + store: StoreState::Open, + }; + tokio::spawn(async move { + let (mut input, output) = tokio::io::split(accepted); + let output = Arc::new(tokio::sync::Mutex::new(output)); + while let Ok(request) = protocol::read::(&mut input).await { + recorded.lock().unwrap().push(request.method.clone()); + if request.method == "get_local_api_status" { + let hold = pending_holds.lock().unwrap().pop_front(); + if let Some(hold) = hold { + let response = protocol::Message::Response { + id: request.id, + result: Ok(json!(service.status())), + }; + let output = output.clone(); + tokio::spawn(async move { + let _ = hold.await; + let _ = protocol::write(&mut *output.lock().await, &response).await; + }); + continue; + } + } + let result = match request.method.as_str() { + "get_local_api_status" => Ok(json!(service.status())), + "store_status" => Ok(json!({ + "state":StoreState::Open, + "mode":"plain", + "store_id":"fixture", + "setup_operation_id":null, + "connection_rebind_required":false, + "persistence_status":maple_agent::store::PersistenceStatus::default(), + "local_api_running":service.stays_running(), + })), + "set_local_api_config" => { + // This boundary deliberately supplies successful + // account authorization without an external service. + let (user, config): (String, LocalApiConfig) = + serde_json::from_value(request.params).unwrap(); + assert_eq!(user, "user"); + service + .configure(config, user) + .await + .map(|status| json!(status)) + } + "stop_local_api" => Ok(json!(service.stop().await)), + "cancel_oauth" | "set_ui_settings" | "flush_ui_settings" | "detach" => { + Ok(Value::Null) + } + _ => Err("Optional presentation fixture operation unavailable".into()), + }; + if protocol::write( + &mut *output.lock().await, + &protocol::Message::Response { + id: request.id, + result, + }, + ) + .await + .is_err() + { + break; + } + } + service.stop().await; + }); + ( + Arc::new(AgentBackend::from_host_client_for_test( + RemoteClient::attach_for_test(endpoint, connection, welcome), + )), + manager, + ) + }); + Self { + backend, + manager, + requests, + status_holds, + runtime: Some(runtime), + _root: root, + } + } + + fn count(&self, method: &str) -> usize { + self.requests + .lock() + .unwrap() + .iter() + .filter(|value| value.as_str() == method) + .count() + } + + fn api_calls(&self) -> usize { + self.requests + .lock() + .unwrap() + .iter() + .filter(|value| value.contains("local_api")) + .count() + } + + fn hold_next_status(&self) -> tokio::sync::oneshot::Sender<()> { + let (release, hold) = tokio::sync::oneshot::channel(); + self.status_holds.lock().unwrap().push_back(hold); + release + } +} + +impl Drop for FixtureHost { + fn drop(&mut self) { + if let Some(runtime) = self.runtime.take() { + runtime.block_on(self.manager.stop()); + runtime.shutdown_timeout(Duration::from_secs(3)); + } + } +} + +struct AppHost(Entity); + +impl Render for AppHost { + fn render(&mut self, _: &mut Window, _: &mut Context) -> impl IntoElement { + div().w(px(1100.)).h(px(900.)).child(self.0.clone()) + } +} + +fn wait( + cx: &mut gpui::VisualTestContext, + what: &str, + mut condition: impl FnMut(&mut gpui::VisualTestContext) -> bool, +) { + for _ in 0..500 { + cx.run_until_parked(); + if condition(cx) { + return; + } + std::thread::sleep(Duration::from_millis(10)); + } + panic!("Timed out waiting for {what}"); +} + +fn select_and_show_settings(app: &Entity, host: &str, cx: &mut gpui::VisualTestContext) { + app.update(cx, |app, cx| { + if let Screen::Settings(settings) = &app.screen { + let snapshot = settings.read(cx).settings_snapshot(); + app.close_settings(snapshot, cx); + } + let chat = app.remote_computers[host].chat.clone().unwrap(); + app.selected_host = Some(host.into()); + app.screen = Screen::Chat(chat); + app.show_settings(Section::LocalApi, cx); + }); + wait(cx, "current controller API status", |cx| { + app.read_with(cx, |app, cx| match &app.screen { + Screen::Settings(settings) => settings.read(cx).local_api_status_for_test().is_some(), + _ => false, + }) + }); +} + +fn toggle(cx: &mut gpui::VisualTestContext) { + let bounds = cx + .debug_bounds("local-api-enabled") + .expect("local API switch"); + cx.simulate_click(bounds.center(), gpui::Modifiers::default()); +} + +fn assert_socket(port: u16, open: bool) { + let result = std::net::TcpStream::connect_timeout( + &std::net::SocketAddr::from((std::net::Ipv4Addr::LOCALHOST, port)), + Duration::from_secs(1), + ); + assert_eq!(result.is_ok(), open, "port {port}"); +} + +#[gpui::test] +fn selected_execution_computers_cannot_move_settings_listener_off_the_controller( + cx: &mut gpui::TestAppContext, +) { + let settings_lock = crate::settings::SETTINGS_IO_LOCK.lock(); + cx.executor().allow_parking(); + // Reserve all three before releasing any, so a port cannot be reused by + // another fixture merely because the OS selected the same ephemeral port. + let reservations: Vec<_> = (0..3) + .map(|_| std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap()) + .collect(); + let ports: Vec<_> = reservations + .iter() + .map(|socket| socket.local_addr().unwrap().port()) + .collect(); + drop(reservations); + let local = FixtureHost::new("controller", ports[0]); + let east = FixtureHost::new("east", ports[1]); + let west = FixtureHost::new("west", ports[2]); + let app = super::tests::project_shell_with_backend(cx, local.backend.clone(), &settings_lock); + app.update(cx, |app, cx| { + for (host, backend) in [("east", &east.backend), ("west", &west.backend)] { + let chat = + cx.new(|cx| ChatScreen::new_without_start(backend.clone(), "user".into(), cx)); + app.remote_computers.get_mut(host).unwrap().chat = Some(chat); + } + }); + let (_, cx) = cx.add_window_view(|window, cx| { + window.activate_window(); + crate::desktop::register_key_bindings(cx); + AppHost(app.clone()) + }); + cx.simulate_resize(size(px(1100.), px(900.))); + + select_and_show_settings(&app, "east", cx); + toggle(cx); + wait(cx, "controller's real listening socket", |_| { + local.manager.status().state == LocalApiState::Running + && local.count("set_local_api_config") == 1 + }); + assert_socket(ports[0], true); + assert_socket(ports[1], false); + assert_socket(ports[2], false); + assert_eq!(east.api_calls(), 0); + assert_eq!(west.api_calls(), 0); + + select_and_show_settings(&app, "west", cx); + toggle(cx); + wait(cx, "controller's Off action", |_| { + local.manager.status().state == LocalApiState::Stopped + && local.count("set_local_api_config") == 2 + }); + assert!(!local.manager.status().config.enabled); + assert_socket(ports[0], false); + assert_socket(ports[1], false); + assert_socket(ports[2], false); + assert!(local.count("get_local_api_status") >= 2); + assert_eq!(east.api_calls(), 0); + assert_eq!(west.api_calls(), 0); +} + +#[gpui::test] +fn desktop_coalesces_api_pushes_and_retires_old_host_callbacks(cx: &mut gpui::TestAppContext) { + let settings_lock = crate::settings::SETTINGS_IO_LOCK.lock(); + cx.executor().allow_parking(); + let reservation = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap(); + let port = reservation.local_addr().unwrap().port(); + drop(reservation); + let local = FixtureHost::new("refresh-controller", port); + let app = super::tests::project_shell_with_backend(cx, local.backend.clone(), &settings_lock); + let (_, cx) = cx.add_window_view(|window, _cx| { + window.activate_window(); + AppHost(app.clone()) + }); + cx.simulate_resize(size(px(1100.), px(900.))); + + let first = local.hold_next_status(); + app.update(cx, |app, cx| { + app.handle_service_events(vec![AgentServiceEvent::LocalApiChanged; 100], cx); + }); + wait(cx, "one admitted status query", |_| { + local.count("get_local_api_status") == 1 + }); + app.read_with(cx, |app, _| { + assert!(app.local_api_refreshing); + assert!(app.local_api_refresh_again); + }); + first.send(()).unwrap(); + wait(cx, "one coalesced status query", |cx| { + local.count("get_local_api_status") == 2 + && app.read_with(cx, |app, _| !app.local_api_refreshing) + }); + assert_eq!(local.count("get_local_api_status"), 2); + + let old_reply = local.hold_next_status(); + let new_reply = local.hold_next_status(); + let old_bridge = app.update(cx, |app, cx| { + let index = app.remote_bridges.borrow().len(); + app.local_api_revision = 100; + app.refresh_local_api(cx); + index + }); + wait(cx, "old host's held query", |_| { + local.count("get_local_api_status") == 3 + }); + let previous_epoch = app.read_with(cx, |app, _| app.local_api_host_epoch); + app.update(cx, |app, cx| { + app.handle_service_events(vec![AgentServiceEvent::HostReconnected], cx); + }); + wait(cx, "replacement host's held query", |_| { + local.count("get_local_api_status") == 4 + }); + old_reply.send(()).unwrap(); + wait(cx, "retired old-host callback", |cx| { + app.read_with(cx, |app, _| { + app.remote_bridges.borrow()[old_bridge].is_ready() + }) + }); + app.read_with(cx, |app, _| { + assert_eq!(app.local_api_host_epoch, previous_epoch + 1); + assert_eq!(app.local_api_revision, 0); + assert!( + app.local_api_refreshing, + "an old callback cannot clear the replacement host's in-flight query" + ); + assert!(!app.local_api_refresh_again); + }); + assert_eq!(local.count("get_local_api_status"), 4); + new_reply.send(()).unwrap(); + wait(cx, "replacement host status", |cx| { + app.read_with(cx, |app, _| { + !app.local_api_refreshing && app.local_api_revision == local.manager.status().revision + }) + }); + assert_eq!(local.count("get_local_api_status"), 4); +} diff --git a/apps/maple-agent/app/src/host.rs b/apps/maple-agent/app/src/host.rs index 147954967..79b6d3f65 100644 --- a/apps/maple-agent/app/src/host.rs +++ b/apps/maple-agent/app/src/host.rs @@ -5,6 +5,8 @@ pub(crate) mod hardening; pub(crate) mod ipc; #[cfg(test)] mod lifecycle_tests; +#[cfg(all(test, feature = "proxy"))] +mod local_api_tests; mod passwords; pub mod protocol; pub(crate) mod receipts; @@ -884,6 +886,8 @@ fn gate(state: &StoreState, method: &str) -> Result<(), String> { "store_status" | "host_status" | "detach" | "cancel_oauth" ) || (method == "erase_local_data" && !matches!(state, StoreState::Refused { .. })); + #[cfg(feature = "proxy")] + let control = control || matches!(method, "get_local_api_status" | "stop_local_api"); let allowed = control || match state { StoreState::Locked => method == "unlock", @@ -1020,6 +1024,10 @@ impl OwnerContext { } fn keeps_host(&self, backend: &LocalBackend) -> bool { + #[cfg(feature = "proxy")] + if backend.local_api.stays_running() { + return true; + } let state = self.state(backend); self.retained_failure.load(Ordering::Acquire) || self.busy.load(Ordering::Acquire) @@ -1047,6 +1055,10 @@ impl OwnerContext { // Release exposure remains gated on native G3 evidence. Local debug // builds permit exact-app validation without shipping this capability. value["password_settings_available"] = serde_json::json!(cfg!(debug_assertions)); + #[cfg(feature = "proxy")] + { + value["local_api_running"] = serde_json::json!(backend.local_api.stays_running()); + } let persistence = serde_json::to_value(store.persistence_status()) .map_err(|_| "Could not encode persistence status")?; value["persistence_failure"] = persistence @@ -1086,6 +1098,26 @@ impl OwnerContext { running_tasks, remote_controllers, background_tasks, + local_api_running: { + #[cfg(feature = "proxy")] + { + backend.local_api.stays_running() + } + #[cfg(not(feature = "proxy"))] + { + false + } + }, + local_api_requests: { + #[cfg(feature = "proxy")] + { + backend.local_api.status().active_requests + } + #[cfg(not(feature = "proxy"))] + { + 0 + } + }, }) .map_err(|_| "Could not encode host status".into()) } @@ -1105,6 +1137,10 @@ impl OwnerContext { return Err(error.to_owned()); } gate(&self.state(backend), method)?; // Must precede remote_* routing. + #[cfg(feature = "proxy")] + if matches!(method, "get_local_api_status" | "stop_local_api") { + return backend::dispatch(backend, request.into_request()).await; + } if method == "store_status" { let mut status = self.status(backend)?; status["connection_rebind_required"] = @@ -1114,6 +1150,15 @@ impl OwnerContext { if method == "host_status" { return self.host_status(backend, remote).await; } + #[cfg(feature = "proxy")] + if matches!( + method, + "login" | "logout_and_clear" | "oauth_wait_for_handoff" | "oauth_complete" + ) { + // Wake a pending proxy restore before waiting for admission. Its + // account validation must not hold up a local account transition. + backend.local_api.stop().await; + } let store = backend .store() .ok_or("Local storage is unavailable")? @@ -1201,9 +1246,13 @@ impl OwnerContext { .lock() .map_err(|_| "Store notice is unavailable")? = None; } + #[cfg(feature = "proxy")] + let local_api_control_epoch = backend.local_api_control_epoch(); let (_exclusive_admission, _shared_admission) = if drain_operation { backend.begin_store_drain(); self.drain_acp(); + #[cfg(feature = "proxy")] + backend.local_api.stop().await; remote.shutdown().await; (Some(self.exclusive_admission(backend).await), None) } else { @@ -1231,6 +1280,8 @@ impl OwnerContext { } else { (None, Some(backend.account_gate.read().await)) }; + #[cfg(feature = "proxy")] + if drain_operation { backend.local_api.stop().await; } generation.check(backend, method)?; gate(&store.state(), method)?; if drain_operation && !destructive { @@ -1363,6 +1414,19 @@ impl OwnerContext { remote.shutdown().await; if backend.saved_user_id().is_some() && store.state() == StoreState::Open { remote.restore().await; + #[cfg(feature = "proxy")] + { + let generation = backend.local_api.generation(); + let _account = backend.account_gate.read().await; + if backend.local_api_control_epoch() == local_api_control_epoch { + let _ = backend + .local_api_while_current(generation, async { + Ok(backend.validate_persisted_auth().await) + }) + .await; + let _ = backend.resume_local_api(generation).await; + } + } } } return result; @@ -1553,6 +1617,8 @@ impl OwnerContext { // Administrative local Stop intentionally cancels local work while // waiting for its exclusive operation/admission guards. backend.cancel_oauth(); + #[cfg(feature = "proxy")] + backend.local_api.stop().await; let operation = self.operation.lock(); tokio::pin!(operation); loop { @@ -1619,12 +1685,16 @@ impl OwnerContext { let restore_remote = remote.stays_running(); backend.begin_store_drain(); self.drain_acp(); + #[cfg(feature = "proxy")] + backend.local_api.stop().await; remote.shutdown().await; let _admission = match remote_admission { Some(admission) => admission, None => self.exclusive_admission(backend).await, }; let _account = backend.account_gate.write().await; + #[cfg(feature = "proxy")] + backend.local_api.stop().await; // An already-admitted account operation may have restored its remote // node while Stop waited for admission. Close that late node as well; // no account operation can publish another under these exclusive gates. @@ -1844,6 +1914,38 @@ async fn serve_runtime( }); } let owner = Arc::new(OwnerContext::new(backend.store())); + #[cfg(feature = "proxy")] + let local_api_pump = { + let mut updates = backend.local_api.subscribe(); + let events = events.clone(); + tokio::spawn(async move { + while updates.changed().await.is_ok() { + let _ = events.send(Arc::new(AgentServiceEvent::LocalApiChanged)); + } + }) + }; + #[cfg(feature = "proxy")] + let local_api_restore = { + let backend = backend.clone(); + let owner = owner.clone(); + let generation = backend.local_api.generation(); + tokio::spawn(async move { + let _admission = owner.admission.read().await; + let _account = backend.account_gate.read().await; + if owner.state(&backend) == StoreState::Open + && matches!( + backend + .local_api_while_current(generation, async { + Ok(backend.validate_persisted_auth().await) + }) + .await, + Ok(backend::RestoreOutcome::Valid(_)) + ) + { + let _ = backend.resume_local_api(generation).await; + } + }) + }; let sink = events.clone(); let account_remote = remote.clone(); let pump = tokio::spawn(async move { @@ -1932,6 +2034,11 @@ async fn serve_runtime( clients.abort_all(); while clients.join_next().await.is_some() {} pump.abort(); + #[cfg(feature = "proxy")] + { + local_api_pump.abort(); + local_api_restore.abort(); + } Ok(()) } @@ -2166,7 +2273,7 @@ async fn serve_client_inner( continue; } let readable = (!stale && owner.state(&backend) == StoreState::Open) - || matches!(&*event, AgentServiceEvent::StoreStateChanged { .. } | AgentServiceEvent::AccountChanged(None) | AgentServiceEvent::PersistenceFailureChanged { .. }); + || matches!(&*event, AgentServiceEvent::StoreStateChanged { .. } | AgentServiceEvent::AccountChanged(None) | AgentServiceEvent::PersistenceFailureChanged { .. } | AgentServiceEvent::LocalApiChanged); if readable && task_subscription.allows(&event) { write_event(&mut output, &event).await?; } diff --git a/apps/maple-agent/app/src/host/local_api_tests.rs b/apps/maple-agent/app/src/host/local_api_tests.rs new file mode 100644 index 000000000..8c64eb993 --- /dev/null +++ b/apps/maple-agent/app/src/host/local_api_tests.rs @@ -0,0 +1,517 @@ +//! The local listener's safety controls remain available during store recovery. + +use super::*; +use crate::local_api::{LocalApiConfig, LocalApiState, LocalApiStatus}; +use maple_agent::store::records::{CredentialTokens, SecretString}; +use maple_agent::store::{DamageReason, RefusalReason, StorePaths}; +use serde_json::json; + +#[test] +fn unavailable_stores_keep_api_status_and_emergency_stop_but_reject_configuration() { + for state in [ + StoreState::Locked, + StoreState::NeedsSetup, + StoreState::Maintenance, + StoreState::Damaged { + reason: DamageReason::Generation, + }, + StoreState::Refused { + reason: RefusalReason::NewerVersion, + }, + ] { + for method in ["get_local_api_status", "stop_local_api"] { + assert!( + gate(&state, method).is_ok(), + "{method} must stay available while the store is {state:?}" + ); + } + assert!(gate(&state, "set_local_api_config").is_err()); + } +} + +fn directory() -> tempfile::TempDir { + #[cfg(unix)] + { + tempfile::Builder::new() + .prefix("maple-api-") + .tempdir_in("/tmp") + .unwrap() + } + #[cfg(windows)] + { + tempfile::tempdir().unwrap() + } +} + +const USER: &str = "local-api-lifecycle@example.test"; + +fn credentials(backend: &LocalBackend) { + let store = backend.store().unwrap(); + let account = store.upsert_account(USER, 1).unwrap(); + store + .replace_credentials( + None, + account, + backend.remote_api_url(), + "local-api-lifecycle-session", + 1, + &CredentialTokens { + access_token: SecretString::new("synthetic-api-lifecycle-token".into()), + refresh_token: None, + }, + 1, + ) + .unwrap(); + backend.reload_open_store().unwrap(); +} + +struct HostFixture { + client: RemoteClient, + backend: Arc, + store: Arc, + owner: Arc, + remote: Arc, + events: broadcast::Sender>, + server: tokio::task::JoinHandle>, +} + +impl HostFixture { + async fn new(root: &std::path::Path, backend: Arc) -> Self { + let store = backend.store().unwrap().clone(); + let receipts = Arc::new(receipts::Receipts::new(store.clone())); + let (events, _) = broadcast::channel(32); + let remote = + crate::remote::manager::Manager::new(backend.clone(), events.clone(), receipts.clone()); + let owner = Arc::new(OwnerContext::new(Some(&store))); + let endpoint = ipc::Endpoint::for_test(root.to_path_buf()); + let mut listener = endpoint.bind().unwrap(); + let (stop, _stopped) = mpsc::channel(1); + let server = { + let endpoint = endpoint.clone(); + let backend = backend.clone(); + let remote = remote.clone(); + let owner = owner.clone(); + let events = events.clone(); + tokio::spawn(async move { + let connection = listener.accept().await.unwrap(); + serve_client( + connection, endpoint, backend, events, stop, receipts, remote, owner, + ) + .await + }) + }; + let (connection, welcome) = connect(&endpoint, ConnectionMode::Client, false) + .await + .unwrap(); + let client = RemoteClient::attach(endpoint, connection, welcome); + Self { + client, + backend, + store, + owner, + remote, + events, + server, + } + } + + async fn start_listener(&self) -> u16 { + let reservation = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap(); + let port = reservation.local_addr().unwrap().port(); + drop(reservation); + // Only upstream account validation is supplied by the test. This is + // the production listener, router, request boundary and cancellation. + self.backend + .local_api + .start( + LocalApiConfig { + enabled: true, + port, + }, + USER.to_string(), + ) + .await + .unwrap(); + let http = reqwest::Client::builder().no_proxy().build().unwrap(); + assert_eq!( + http.get(format!("http://127.0.0.1:{port}/health")) + .send() + .await + .unwrap() + .status(), + reqwest::StatusCode::OK + ); + assert_eq!( + http.get(format!("http://127.0.0.1:{port}/v1/models")) + .send() + .await + .unwrap() + .status(), + reqwest::StatusCode::UNAUTHORIZED + ); + port + } + + async fn finish(self) { + self.backend.local_api.stop().await; + self.client.detach().await; + assert!(self.server.await.unwrap().unwrap()); + self.remote.shutdown().await; + if self.store.state() == StoreState::Open { + self.store.close_for_shutdown().unwrap(); + } + } +} + +async fn assert_closed(port: u16) { + assert!( + tokio::net::TcpStream::connect((std::net::Ipv4Addr::LOCALHOST, port)) + .await + .is_err(), + "the listening socket must be gone when cleanup acknowledges success" + ); +} + +#[tokio::test] +async fn emergency_stop_over_ipc_closes_the_real_listener_and_preserves_requested_configuration() { + let root = directory(); + let backend = Arc::new(LocalBackend::isolated_for_test(root.path())); + credentials(&backend); + let fixture = HostFixture::new(root.path(), backend).await; + let port = fixture.start_listener().await; + let status: LocalApiStatus = fixture + .client + .call("get_local_api_status", ()) + .await + .unwrap(); + assert_eq!(status.state, LocalApiState::Running); + assert!(status.config.enabled); + assert_eq!(status.config.port, port); + + let stopped: LocalApiStatus = fixture.client.call("stop_local_api", ()).await.unwrap(); + assert_eq!(stopped.state, LocalApiState::Stopped); + assert!(stopped.config.enabled); + assert_eq!(stopped.config.port, port); + assert_closed(port).await; + + // The UI switch has a different contract from emergency Stop: Off is + // durable and must not require a network request just to stop spending. + let disabled: LocalApiStatus = fixture + .client + .call( + "set_local_api_config", + ( + USER, + LocalApiConfig { + enabled: false, + port, + }, + ), + ) + .await + .unwrap(); + assert_eq!(disabled.state, LocalApiState::Stopped); + assert!(!disabled.config.enabled); + let account = fixture.store.account_id(USER).unwrap().unwrap(); + assert_eq!( + fixture + .store + .account_setting(account, "local_api.config") + .unwrap(), + Some(json!({"enabled":false,"port":port})) + ); + fixture.finish().await; +} + +#[tokio::test] +async fn saved_account_and_enabled_preference_do_not_start_before_authentication_is_validated() { + let root = directory(); + let backend = LocalBackend::isolated_for_test(root.path()); + credentials(&backend); + let store = backend.store().unwrap(); + let account = store.account_id(USER).unwrap().unwrap(); + let reservation = std::net::TcpListener::bind((std::net::Ipv4Addr::LOCALHOST, 0)).unwrap(); + let port = reservation.local_addr().unwrap().port(); + drop(reservation); + store + .set_account_setting( + account, + "local_api.config", + &json!({"enabled":true,"port":port}), + 1, + ) + .unwrap(); + assert_eq!(backend.saved_user_id().as_deref(), Some(USER)); + + tokio::time::timeout( + Duration::from_secs(3), + backend.resume_local_api(backend.local_api.generation()), + ) + .await + .expect("resume must not try to validate a cached account implicitly") + .unwrap(); + assert_eq!(backend.local_api.status().state, LocalApiState::Stopped); + assert_closed(port).await; + assert_eq!( + store.account_setting(account, "local_api.config").unwrap(), + Some(json!({"enabled":true,"port":port})), + "an unvalidated restore must retain the account's desired preference" + ); + store.close_for_shutdown().unwrap(); +} + +#[test] +fn failed_off_write_keeps_automatic_resume_paused_with_durable_on_retained() { + use std::{future::Future, task::Poll}; + + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .max_blocking_threads(1) + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let root = directory(); + let backend = LocalBackend::isolated_for_test(root.path()); + credentials(&backend); + let store = backend.store().unwrap(); + let account = store.account_id(USER).unwrap().unwrap(); + let old = json!({"enabled":true,"port":8123}); + store + .set_account_setting(account, "local_api.config", &old, 1) + .unwrap(); + + // Hold this runtime's only blocking worker. The actual setter can + // admit Off and stop the listener, but its durable write cannot run + // until the public store lifecycle has closed the fixture database. + let (entered, wait_entered) = oneshot::channel(); + let (release, blocked) = std::sync::mpsc::channel(); + let blocker = tokio::task::spawn_blocking(move || { + let _ = entered.send(()); + let _ = blocked.recv(); + }); + wait_entered.await.unwrap(); + { + let off = backend.set_local_api_config( + USER, + LocalApiConfig { + enabled: false, + port: 8123, + }, + ); + tokio::pin!(off); + std::future::poll_fn(|cx| { + assert!(off.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + assert!(backend.local_api_suspended_for_test()); + store.close().unwrap(); + release.send(()).unwrap(); + blocker.await.unwrap(); + assert_eq!(off.await.unwrap_err(), "Could not save Local API settings"); + } + store.initialize_plain().unwrap(); + assert_eq!( + store.account_setting(account, "local_api.config").unwrap(), + Some(old), + "the refused write must leave the old durable On preference intact" + ); + assert!(backend.local_api_suspended_for_test()); + backend + .resume_local_api(backend.local_api.generation()) + .await + .unwrap(); + assert!(backend.local_api_suspended_for_test()); + assert_eq!(backend.local_api.status().state, LocalApiState::Stopped); + store.close_for_shutdown().unwrap(); + }); + runtime.shutdown_timeout(Duration::from_secs(3)); +} + +#[tokio::test] +async fn logout_closes_the_listener_without_relying_on_the_account_event_pump() { + let root = directory(); + let backend = Arc::new(LocalBackend::isolated_for_test(root.path())); + credentials(&backend); + let fixture = HostFixture::new(root.path(), backend).await; + let port = fixture.start_listener().await; + fixture + .client + .call::<()>("logout_and_clear", (USER,)) + .await + .unwrap(); + assert!(fixture.backend.saved_user_id().is_none()); + assert_closed(port).await; + let stopped: LocalApiStatus = fixture + .client + .call("get_local_api_status", ()) + .await + .unwrap(); + assert_eq!(stopped.state, LocalApiState::Stopped); + fixture.finish().await; +} + +#[tokio::test] +async fn emergency_stop_during_confirmed_erasure_survives_a_precommit_cleanup_refusal() { + let root = directory(); + let backend = Arc::new(LocalBackend::isolated_for_test(root.path())); + credentials(&backend); + let fixture = HostFixture::new(root.path(), backend).await; + let port = fixture.start_listener().await; + let generation = fixture.store.store_id().unwrap(); + let epoch = fixture.backend.local_api_control_epoch(); + let spills = maple_agent::store::spill::SpillRoot::new(fixture.store.paths()).unwrap(); + spills + .task_dir("00000000-0000-4000-8000-000000000003") + .unwrap(); + std::fs::remove_dir_all(spills.path()).unwrap(); + // An unsafe owned-spill type deterministically refuses before replacement + // verification. No native checker, fake crypto or bypass is needed. + std::fs::write(spills.path(), b"unsafe owned-spill type").unwrap(); + let admission = fixture.owner.admission.read().await; + { + let erasure = fixture + .client + .call::("erase_local_data", json!({"confirmation":"ERASE"})); + tokio::pin!(erasure); + tokio::time::timeout(Duration::from_secs(3), async { + tokio::select! { + result = &mut erasure => panic!("erasure passed held admission: {result:?}"), + _ = async { + while !fixture.owner.maintenance.load(Ordering::Acquire) + || fixture.backend.local_api.status().state != LocalApiState::Stopped + { + tokio::task::yield_now().await; + } + } => {}, + } + }) + .await + .expect("maintenance must suspend the listener before waiting for readers"); + assert_eq!(fixture.backend.local_api_control_epoch(), epoch); + assert_closed(port).await; + let stopped: LocalApiStatus = tokio::time::timeout( + Duration::from_secs(3), + fixture.client.call("stop_local_api", ()), + ) + .await + .expect("emergency Stop must bypass held maintenance admission") + .unwrap(); + assert_eq!(stopped.state, LocalApiState::Stopped); + assert_eq!(fixture.backend.local_api_control_epoch(), epoch + 1); + assert!(fixture.backend.local_api_suspended_for_test()); + drop(admission); + assert!(erasure.await.is_err()); + } + assert_eq!(fixture.backend.local_api_control_epoch(), epoch + 1); + assert_eq!(fixture.store.store_id().unwrap(), generation); + assert_eq!(fixture.store.state(), StoreState::Open); + assert_closed(port).await; + let stopped: LocalApiStatus = fixture + .client + .call("get_local_api_status", ()) + .await + .unwrap(); + assert_eq!(stopped.state, LocalApiState::Stopped); + assert!(stopped.config.enabled); + fixture + .backend + .resume_local_api(fixture.backend.local_api.generation()) + .await + .unwrap(); + assert!(fixture.backend.local_api_suspended_for_test()); + assert_eq!( + fixture.backend.local_api.status().state, + LocalApiState::Stopped + ); + assert_closed(port).await; + std::fs::remove_file(spills.path()).unwrap(); + fixture.finish().await; +} + +#[tokio::test] +async fn owner_shutdown_closes_the_listener_before_releasing_local_data() { + let root = directory(); + let backend = Arc::new(LocalBackend::isolated_for_test(root.path())); + let fixture = HostFixture::new(root.path(), backend).await; + let port = fixture.start_listener().await; + tokio::time::timeout( + Duration::from_secs(3), + fixture + .owner + .shutdown(&fixture.backend, &fixture.remote, &fixture.events), + ) + .await + .expect("a local listener must not hold an account reader through shutdown") + .unwrap(); + assert_ne!(fixture.store.state(), StoreState::Open); + assert_closed(port).await; + fixture.finish().await; +} + +#[tokio::test] +async fn locked_profile_answers_status_and_stop_over_real_private_ipc() { + let root = directory(); + let store = Arc::new( + Store::unavailable( + StorePaths::new(root.path().join("locked-store")), + StoreState::Locked, + ) + .unwrap(), + ); + let backend = Arc::new( + LocalBackend::new_with_store("http://127.0.0.1:1".into(), String::new(), store.clone()) + .unwrap(), + ); + let receipts = Arc::new(receipts::Receipts::new(store.clone())); + let (events, _) = broadcast::channel(32); + let remote = + crate::remote::manager::Manager::new(backend.clone(), events.clone(), receipts.clone()); + let owner = Arc::new(OwnerContext::new(Some(&store))); + let endpoint = ipc::Endpoint::for_test(root.path().to_path_buf()); + let mut listener = endpoint.bind().unwrap(); + let (stop, _stopped) = mpsc::channel(1); + let server = { + let endpoint = endpoint.clone(); + let backend = backend.clone(); + let remote = remote.clone(); + tokio::spawn(async move { + let connection = listener.accept().await.unwrap(); + serve_client( + connection, endpoint, backend, events, stop, receipts, remote, owner, + ) + .await + }) + }; + let (connection, welcome) = connect(&endpoint, ConnectionMode::Client, false) + .await + .unwrap(); + assert_eq!(welcome.store, StoreState::Locked); + let client = RemoteClient::attach(endpoint, connection, welcome); + + for method in ["get_local_api_status", "stop_local_api"] { + let status = tokio::time::timeout( + Duration::from_secs(3), + client.call::(method, ()), + ) + .await + .expect("locked controls must not wait for a store or account gate") + .unwrap(); + assert_eq!(status.state, LocalApiState::Stopped); + assert_eq!(status.active_requests, 0); + } + let error = client + .call::( + "set_local_api_config", + json!(["unavailable-account", {"enabled":true,"port":8080}]), + ) + .await + .unwrap_err(); + assert!(error.contains("locked")); + assert_eq!(store.state(), StoreState::Locked); + + client.detach().await; + assert!(server.await.unwrap().unwrap()); + remote.shutdown().await; +} diff --git a/apps/maple-agent/app/src/host/protocol.rs b/apps/maple-agent/app/src/host/protocol.rs index 09e237a45..165b8a303 100644 --- a/apps/maple-agent/app/src/host/protocol.rs +++ b/apps/maple-agent/app/src/host/protocol.rs @@ -322,6 +322,10 @@ pub struct HostStatus { pub running_tasks: usize, pub remote_controllers: usize, pub background_tasks: usize, + #[serde(default)] + pub local_api_running: bool, + #[serde(default)] + pub local_api_requests: usize, } pub struct SensitiveMessage(pub Message); diff --git a/apps/maple-agent/app/src/local_api.rs b/apps/maple-agent/app/src/local_api.rs new file mode 100644 index 000000000..dea822711 --- /dev/null +++ b/apps/maple-agent/app/src/local_api.rs @@ -0,0 +1,1148 @@ +//! This computer's optional HTTP API. The local host owns every listener, +//! connection and request; selecting a remote computer never moves this service. + +use axum::{ + Router, + body::Body, + extract::Request, + http::{Method, StatusCode, header}, + middleware::{self, Next}, + response::{IntoResponse, Response}, +}; +use futures_util::StreamExt; +use hyper_util::{ + rt::{TokioIo, TokioTimer}, + service::TowerToHyperService, +}; +use serde::{Deserialize, Serialize}; +use std::{ + io, + net::{Ipv4Addr, SocketAddr}, + sync::{Arc, Mutex, Weak}, + time::Duration, +}; +use tokio::{ + net::TcpListener, + sync::{Mutex as AsyncMutex, OwnedSemaphorePermit, Semaphore, watch}, + task::{JoinHandle, JoinSet}, +}; +use tokio_util::sync::CancellationToken; +use tower_http::cors::{AllowHeaders, Any, CorsLayer}; + +const MAX_CONNECTIONS: usize = 32; +const MAX_REQUESTS: usize = 8; +const HEADER_TIMEOUT: Duration = Duration::from_secs(10); +const UPLOAD_TIMEOUT: Duration = Duration::from_secs(30); +const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2); + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LocalApiConfig { + pub enabled: bool, + pub port: u16, +} + +impl Default for LocalApiConfig { + fn default() -> Self { + Self { + enabled: false, + port: crate::env::env_string("MAPLE_PORT") + .and_then(|value| value.parse().ok()) + .filter(|port| *port != 0) + .unwrap_or(8080), + } + } +} + +impl LocalApiConfig { + pub fn validate(&self) -> Result<(), String> { + if self.port == 0 { + return Err("Choose a local API port between 1 and 65535.".into()); + } + Ok(()) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum LocalApiState { + Stopped, + Starting, + Running, + Stopping, + Failed, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct LocalApiStatus { + pub config: LocalApiConfig, + pub state: LocalApiState, + pub active_requests: usize, + pub error: Option, + pub owner: Option, + pub revision: u64, +} + +impl Default for LocalApiStatus { + fn default() -> Self { + Self { + config: LocalApiConfig::default(), + state: LocalApiState::Stopped, + active_requests: 0, + error: None, + owner: None, + revision: 0, + } + } +} + +/// The CLI retains its opt-in browser mode. The host-owned service always calls +/// this with `false`, and never configures a server-held API key. +pub fn apply_browser_policy(app: Router, cors: bool) -> Router { + if cors { + app.layer( + CorsLayer::new() + .allow_origin(Any) + .allow_methods([Method::GET, Method::POST, Method::OPTIONS]) + .allow_headers(AllowHeaders::mirror_request()), + ) + } else { + app.layer(middleware::from_fn( + |request: Request, next: Next| async move { + if request.headers().contains_key(header::ORIGIN) + || request.headers().contains_key("sec-fetch-site") + { + return StatusCode::FORBIDDEN.into_response(); + } + next.run(request).await + }, + )) + } +} + +#[derive(Clone)] +pub struct LocalApiManager { + inner: Arc, +} + +struct ManagerInner { + api_url: String, + pcr0: maple_proxy::Pcr0Environment, + state: Mutex, + running: AsyncMutex>, + updates: watch::Sender, +} + +struct ManagerState { + generation: u64, + cancel: CancellationToken, + status: LocalApiStatus, +} + +struct Running { + cancel: CancellationToken, + task: JoinHandle<()>, +} + +impl Drop for ManagerInner { + fn drop(&mut self) { + self.state.get_mut().unwrap().cancel.cancel(); + if let Some(running) = self.running.get_mut().take() { + running.cancel.cancel(); + // The accept task owns a JoinSet: aborting it also aborts every + // connection, including peers that have not sent request headers. + running.task.abort(); + } + } +} + +impl ManagerInner { + fn change(&self, change: impl FnOnce(&mut ManagerState)) -> LocalApiStatus { + let mut state = self.state.lock().unwrap(); + change(&mut state); + state.status.revision = state.status.revision.saturating_add(1); + let status = state.status.clone(); + self.updates.send_replace(status.clone()); + status + } + + fn current(&self, generation: u64) -> bool { + let state = self.state.lock().unwrap(); + state.generation == generation && !state.cancel.is_cancelled() + } + + fn finish(&self, generation: u64, phase: LocalApiState, error: Option) { + let mut state = self.state.lock().unwrap(); + if state.generation != generation { + return; + } + state.status.state = phase; + state.status.error = error; + if phase == LocalApiState::Stopped { + state.status.owner = None; + } + state.status.revision = state.status.revision.saturating_add(1); + self.updates.send_replace(state.status.clone()); + } +} + +impl LocalApiManager { + pub fn new(api_url: String) -> Result { + let api_url = maple_agent::maple_api::validate_api_url(&api_url)?; + let pcr0 = maple_agent::open_secret_config::configured_pcr0_environment()?; + let status = LocalApiStatus::default(); + let (updates, _) = watch::channel(status.clone()); + Ok(Self { + inner: Arc::new(ManagerInner { + api_url, + pcr0, + state: Mutex::new(ManagerState { + generation: 0, + cancel: CancellationToken::new(), + status, + }), + running: AsyncMutex::new(None), + updates, + }), + }) + } + + pub fn status(&self) -> LocalApiStatus { + self.inner.state.lock().unwrap().status.clone() + } + + pub fn subscribe(&self) -> watch::Receiver { + self.inner.updates.subscribe() + } + + pub fn generation(&self) -> u64 { + self.inner.state.lock().unwrap().generation + } + + pub fn stays_running(&self) -> bool { + matches!( + self.status().state, + LocalApiState::Starting | LocalApiState::Running | LocalApiState::Stopping + ) + } + + #[cfg(test)] + pub async fn configure( + &self, + config: LocalApiConfig, + owner: String, + ) -> Result { + self.configure_if_current(config, owner, self.generation()) + .await + } + + /// Fence async account/store validation performed by the caller. Stop, + /// logout or another configuration invalidates the captured generation. + pub async fn configure_if_current( + &self, + config: LocalApiConfig, + owner: String, + expected: u64, + ) -> Result { + config.validate()?; + if config.enabled { + self.start_router(config, owner, Some(expected), None).await + } else { + Ok(self.stop_with_config(Some(config), Some(expected)).await) + } + } + + #[cfg(test)] + pub async fn start( + &self, + config: LocalApiConfig, + owner: String, + ) -> Result { + self.start_router(config, owner, None, None).await + } + + async fn start_router( + &self, + config: LocalApiConfig, + owner: String, + expected: Option, + router: Option, + ) -> Result { + config.validate()?; + if !config.enabled { + return Ok(self.stop_with_config(Some(config), expected).await); + } + if owner.is_empty() { + return Err("Sign in on this computer before enabling its local API.".into()); + } + // Invalidate immediately, before waiting for the lifecycle lock. A + // delayed bind/start from the previous owner cannot publish afterward. + let Some((generation, cancel)) = self.begin(Some(config.clone()), Some(owner), expected) + else { + return Ok(self.status()); + }; + let Some(mut running) = self.running_if_current(generation).await else { + return Ok(self.status()); + }; + stop_running(&mut running).await; + if !self.inner.current(generation) { + return Ok(self.status()); + } + let addr = SocketAddr::from((Ipv4Addr::LOCALHOST, config.port)); + let listener = tokio::select! { + biased; + _ = cancel.cancelled() => return Ok(self.status()), + listener = TcpListener::bind(addr) => listener, + }; + let listener = match listener { + Ok(listener) => listener, + Err(error) => { + let message = if error.kind() == io::ErrorKind::AddrInUse { + "That local API port is already in use. Choose another port." + } else { + "The local API could not bind its loopback port. Check the port and try again." + }; + self.inner + .finish(generation, LocalApiState::Failed, Some(message.into())); + return Err(message.into()); + } + }; + let router = router.unwrap_or_else(|| { + maple_proxy::create_app( + maple_proxy::Config::new( + "127.0.0.1".into(), + config.port, + self.inner.api_url.clone(), + ) + .with_pcr0_environment(self.inner.pcr0) + .with_debug(false) + .with_cors(false), + ) + }); + let requests = Arc::new(Semaphore::new(MAX_REQUESTS)); + let request_state = RequestState { + manager: Arc::downgrade(&self.inner), + cancel: cancel.clone(), + requests, + }; + let router = apply_browser_policy( + router.layer(middleware::from_fn_with_state(request_state, admit_request)), + false, + ); + let mut state = self.inner.state.lock().unwrap(); + if state.generation != generation || cancel.is_cancelled() { + return Ok(state.status.clone()); + } + let task = tokio::spawn(serve( + listener, + router, + cancel.clone(), + Arc::downgrade(&self.inner), + generation, + )); + *running = Some(Running { cancel, task }); + state.status.state = LocalApiState::Running; + state.status.revision = state.status.revision.saturating_add(1); + self.inner.updates.send_replace(state.status.clone()); + Ok(state.status.clone()) + } + + /// Suspend the listener while retaining the user's desired configuration. + pub async fn stop(&self) -> LocalApiStatus { + self.stop_with_config(None, None).await + } + + async fn running_if_current( + &self, + generation: u64, + ) -> Option>> { + let running = self.inner.running.lock().await; + // A newer operation may have acquired this mutex before us after + // publishing its intent. Never tear down that newer listener. + self.inner.current(generation).then_some(running) + } + + fn begin( + &self, + config: Option, + owner: Option, + expected: Option, + ) -> Option<(u64, CancellationToken)> { + let mut state = self.inner.state.lock().unwrap(); + if expected.is_some_and(|expected| expected != state.generation) { + return None; + } + // Restoring the same account from another attached client must not + // restart its already-running listener or interrupt inference streams. + // Compare under the same lock as intent publication so a concurrent + // Stop/configuration cannot be overwritten by this redundant restore. + if expected.is_some() + && state.status.state == LocalApiState::Running + && !state.cancel.is_cancelled() + && owner.is_some() + && owner == state.status.owner + && config.as_ref() == Some(&state.status.config) + { + return None; + } + state.cancel.cancel(); + state.generation = state.generation.wrapping_add(1); + state.cancel = CancellationToken::new(); + if let Some(config) = config { + state.status.config = config; + } + state.status.state = if owner.is_some() { + LocalApiState::Starting + } else { + LocalApiState::Stopping + }; + state.status.owner = owner; + state.status.error = None; + state.status.revision = state.status.revision.saturating_add(1); + self.inner.updates.send_replace(state.status.clone()); + Some((state.generation, state.cancel.clone())) + } + + async fn stop_with_config( + &self, + config: Option, + expected: Option, + ) -> LocalApiStatus { + let Some((generation, _)) = self.begin(config, None, expected) else { + return self.status(); + }; + let Some(mut running) = self.running_if_current(generation).await else { + return self.status(); + }; + stop_running(&mut running).await; + self.inner.finish(generation, LocalApiState::Stopped, None); + self.status() + } +} + +async fn stop_running(running: &mut Option) { + if let Some(mut running) = running.take() { + running.cancel.cancel(); + if tokio::time::timeout(SHUTDOWN_TIMEOUT, &mut running.task) + .await + .is_err() + { + running.task.abort(); + // Poll the aborted parent once more so its JoinSet is dropped. + let _ = running.task.await; + } + } +} + +async fn serve( + listener: TcpListener, + router: Router, + cancel: CancellationToken, + manager: Weak, + generation: u64, +) { + let connection_slots = Arc::new(Semaphore::new(MAX_CONNECTIONS)); + let mut connections = JoinSet::new(); + loop { + tokio::select! { + biased; + _ = cancel.cancelled() => break, + _ = connections.join_next(), if !connections.is_empty() => {}, + incoming = listener.accept() => { + let (stream, _) = match incoming { + Ok(incoming) => incoming, + Err(_) => { + if let Some(manager) = manager.upgrade() { + manager.finish(generation, LocalApiState::Failed, Some("The local API listener stopped unexpectedly. Turn it off and on to retry.".into())); + } + break; + } + }; + let Ok(slot) = connection_slots.clone().try_acquire_owned() else { + drop(stream); + continue; + }; + let cancel = cancel.clone(); + let service = TowerToHyperService::new(router.clone()); + connections.spawn(async move { + let _slot = slot; + let mut http = hyper::server::conn::http1::Builder::new(); + http.timer(TokioTimer::new()).header_read_timeout(HEADER_TIMEOUT).max_headers(64); + tokio::select! { + biased; + _ = cancel.cancelled() => {}, + _ = http.serve_connection(TokioIo::new(stream), service) => {}, + } + }); + } + } + } + // Cancel both request setup and response polling; abort also closes sockets + // whose peers are stalled while reading, writing or sending HTTP headers. + cancel.cancel(); + drop(listener); + connections.abort_all(); + while connections.join_next().await.is_some() {} +} + +#[derive(Clone)] +struct RequestState { + manager: Weak, + cancel: CancellationToken, + requests: Arc, +} + +struct RequestGuard { + manager: Weak, + _permit: OwnedSemaphorePermit, +} + +impl Drop for RequestGuard { + fn drop(&mut self) { + if let Some(manager) = self.manager.upgrade() { + manager.change(|state| { + state.status.active_requests = state.status.active_requests.saturating_sub(1); + }); + } + } +} + +async fn admit_request( + axum::extract::State(state): axum::extract::State, + request: Request, + next: Next, +) -> Response { + if state.cancel.is_cancelled() { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + } + if request.uri().path().starts_with("/v1/") + && !request + .headers() + .get(header::AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")) + .is_some_and(|key| !key.trim().is_empty()) + { + return ( + StatusCode::UNAUTHORIZED, + "Send a Maple API key as Authorization: Bearer .", + ) + .into_response(); + } + // Admission precedes the shared router's Bytes extractor, so rejected + // uploads cannot occupy its 50 MiB body buffer while waiting for a slot. + let Ok(permit) = state.requests.clone().try_acquire_owned() else { + return ( + StatusCode::SERVICE_UNAVAILABLE, + "The local API is busy. Try again after an active request finishes.", + ) + .into_response(); + }; + if let Some(manager) = state.manager.upgrade() { + manager.change(|state| state.status.active_requests += 1); + } + let guard = RequestGuard { + manager: state.manager, + _permit: permit, + }; + let (parts, body) = request.into_parts(); + let upload_cancel = state.cancel.clone(); + let upload_deadline = tokio::time::Instant::now() + UPLOAD_TIMEOUT; + let upload = futures_util::stream::unfold( + (body.into_data_stream(), upload_cancel), + move |(mut body, cancel)| async move { + let chunk = tokio::select! { + biased; + _ = cancel.cancelled() => return None, + result = tokio::time::timeout_at(upload_deadline, body.next()) => result, + }; + match chunk { + Ok(Some(chunk)) => Some((chunk.map_err(io::Error::other), (body, cancel))), + Ok(None) => None, + Err(_) => Some(( + Err(io::Error::new( + io::ErrorKind::TimedOut, + "Local API upload timed out", + )), + (body, cancel), + )), + } + }, + ); + let request = Request::from_parts(parts, Body::from_stream(upload)); + let response = tokio::select! { + biased; + _ = state.cancel.cancelled() => return StatusCode::SERVICE_UNAVAILABLE.into_response(), + response = next.run(request) => response, + }; + let (parts, body) = response.into_parts(); + // Keep the permit and counter through streaming completion or disconnect, + // rather than releasing them as soon as response headers are available. + let response = futures_util::stream::unfold( + (body.into_data_stream(), guard, state.cancel), + |(mut body, guard, cancel)| async move { + let chunk = tokio::select! { + biased; + _ = cancel.cancelled() => return None, + chunk = body.next() => chunk, + }; + chunk.map(|chunk| (chunk, (body, guard, cancel))) + }, + ); + Response::from_parts(parts, Body::from_stream(response)) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::{ + body::Bytes, + routing::{get, post}, + }; + use std::sync::atomic::{AtomicBool, Ordering}; + use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpStream, + sync::Notify, + }; + + fn config() -> LocalApiConfig { + let socket = std::net::TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + LocalApiConfig { + enabled: true, + port: socket.local_addr().unwrap().port(), + } + } + + fn manager() -> LocalApiManager { + LocalApiManager::new("http://127.0.0.1:3000".into()).unwrap() + } + + fn client() -> reqwest::Client { + reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(3)) + .build() + .unwrap() + } + + fn endpoint(config: &LocalApiConfig, path: &str) -> String { + format!("http://127.0.0.1:{}{path}", config.port) + } + + async fn start( + manager: &LocalApiManager, + config: LocalApiConfig, + router: Router, + ) -> LocalApiStatus { + manager + .start_router(config, "local-api-test-owner".into(), None, Some(router)) + .await + .unwrap() + } + + async fn wait_for(manager: &LocalApiManager, predicate: impl Fn(&LocalApiStatus) -> bool) { + let mut updates = manager.subscribe(); + tokio::time::timeout(Duration::from_secs(3), async { + loop { + if predicate(&updates.borrow().clone()) { + break; + } + updates.changed().await.unwrap(); + } + }) + .await + .unwrap(); + } + + async fn assert_closed(mut stream: TcpStream) { + let mut data = [0; 1024]; + tokio::time::timeout(Duration::from_secs(1), async move { + loop { + match stream.read(&mut data).await { + Ok(0) | Err(_) => break, + Ok(_) => {} + } + } + }) + .await + .unwrap(); + } + + struct DropFlag(Arc); + + impl Drop for DropFlag { + fn drop(&mut self) { + self.0.store(true, Ordering::SeqCst); + } + } + + #[tokio::test] + async fn binds_loopback_reports_busy_port_and_retains_requested_configuration() { + let manager = manager(); + let socket = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); + let config = LocalApiConfig { + enabled: true, + port: socket.local_addr().unwrap().port(), + }; + let error = manager + .start(config.clone(), "owner".into()) + .await + .unwrap_err(); + assert!(error.contains("already in use")); + let status = manager.status(); + assert_eq!(status.config, config); + assert_eq!(status.state, LocalApiState::Failed); + assert_eq!(status.owner.as_deref(), Some("owner")); + assert_eq!(status.active_requests, 0); + drop(socket); + let running = start( + &manager, + config.clone(), + Router::new().route("/health", get(|| async { "ok" })), + ) + .await; + assert_eq!(running.state, LocalApiState::Running); + assert!(manager.stays_running()); + assert!(running.revision > status.revision); + assert_eq!( + client() + .get(endpoint(&config, "/health")) + .send() + .await + .unwrap() + .status(), + StatusCode::OK + ); + let stopped = manager.stop().await; + assert_eq!(stopped.config, config); + assert_eq!(stopped.owner, None); + assert_eq!(stopped.state, LocalApiState::Stopped); + assert!(!manager.stays_running()); + TcpListener::bind((Ipv4Addr::LOCALHOST, config.port)) + .await + .unwrap(); + } + + #[tokio::test] + async fn browser_and_missing_bearer_are_rejected_before_request_body_is_read() { + let manager = manager(); + let config = config(); + let called = Arc::new(AtomicBool::new(false)); + let router = Router::new().route( + "/v1/chat/completions", + post({ + let called = called.clone(); + move |_: Bytes| { + let called = called.clone(); + async move { + called.store(true, Ordering::SeqCst); + "ok" + } + } + }), + ); + start(&manager, config.clone(), router).await; + for browser in [ + Some("Origin: https://untrusted.example\r\n"), + Some("Sec-Fetch-Site: cross-site\r\n"), + None, + ] { + let mut stream = TcpStream::connect((Ipv4Addr::LOCALHOST, config.port)) + .await + .unwrap(); + let bearer = if browser.is_some() { + "Authorization: Bearer fixture-key\r\n" + } else { + "" + }; + stream.write_all(format!("POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1:{}\r\n{bearer}{}Content-Length: 50000000\r\n\r\n", config.port, browser.unwrap_or("")).as_bytes()).await.unwrap(); + let mut response = [0; 1024]; + let read = tokio::time::timeout(Duration::from_secs(1), stream.read(&mut response)) + .await + .unwrap() + .unwrap(); + let response = String::from_utf8_lossy(&response[..read]); + assert!( + response.contains(if browser.is_some() { "403" } else { "401" }), + "{response}" + ); + } + assert!(!called.load(Ordering::SeqCst)); + assert_eq!(manager.status().active_requests, 0); + manager.stop().await; + } + + #[tokio::test] + async fn admission_limit_precedes_buffering_and_stop_cancels_request_setup() { + let manager = manager(); + let config = config(); + let entered = Arc::new(Notify::new()); + let dropped = Arc::new(AtomicBool::new(false)); + let router = Router::new().route( + "/v1/chat/completions", + post({ + let entered = entered.clone(); + let dropped = dropped.clone(); + move || { + let entered = entered.clone(); + let dropped = dropped.clone(); + async move { + let _guard = DropFlag(dropped); + entered.notify_waiters(); + std::future::pending::<&'static str>().await + } + } + }), + ); + start(&manager, config.clone(), router).await; + let mut callers = JoinSet::new(); + for _ in 0..MAX_REQUESTS { + let url = endpoint(&config, "/v1/chat/completions"); + callers + .spawn(async move { client().post(url).bearer_auth("fixture-key").send().await }); + } + wait_for(&manager, |status| status.active_requests == MAX_REQUESTS).await; + let mut excess = TcpStream::connect((Ipv4Addr::LOCALHOST, config.port)) + .await + .unwrap(); + excess.write_all(format!("POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1:{}\r\nAuthorization: Bearer fixture-key\r\nContent-Length: 50000000\r\n\r\n", config.port).as_bytes()).await.unwrap(); + let mut response = [0; 1024]; + let read = tokio::time::timeout(Duration::from_secs(1), excess.read(&mut response)) + .await + .unwrap() + .unwrap(); + assert!(String::from_utf8_lossy(&response[..read]).contains("503")); + let stopped = tokio::time::timeout(Duration::from_secs(3), manager.stop()) + .await + .unwrap(); + assert_eq!(stopped.active_requests, 0); + assert!(dropped.load(Ordering::SeqCst)); + while let Some(result) = callers.join_next().await { + assert!(result.unwrap().is_err()); + } + TcpListener::bind((Ipv4Addr::LOCALHOST, config.port)) + .await + .unwrap(); + } + + #[tokio::test] + async fn streaming_permit_lives_until_disconnect_and_stop_cancels_slow_readers() { + let manager = manager(); + let config = config(); + let dropped = Arc::new(AtomicBool::new(false)); + let router = Router::new().route( + "/v1/chat/completions", + post({ + let dropped = dropped.clone(); + move || { + let guard = DropFlag(dropped.clone()); + async move { + Body::from_stream(futures_util::stream::unfold( + (0_u64, guard), + |(count, guard)| async move { + tokio::task::yield_now().await; + Some(( + Ok::<_, io::Error>(Bytes::from(vec![b'x'; 64 * 1024])), + (count + 1, guard), + )) + }, + )) + } + } + }), + ); + start(&manager, config.clone(), router).await; + let response = client() + .post(endpoint(&config, "/v1/chat/completions")) + .bearer_auth("fixture-key") + .send() + .await + .unwrap(); + wait_for(&manager, |status| status.active_requests == 1).await; + assert!(!dropped.load(Ordering::SeqCst)); + // Do not consume the stream. Connection cancellation must interrupt a + // peer whose receive window fills, not wait for another body poll. + let stopped = tokio::time::timeout(Duration::from_secs(3), manager.stop()) + .await + .unwrap(); + assert_eq!(stopped.active_requests, 0); + assert!(dropped.load(Ordering::SeqCst)); + drop(response); + TcpListener::bind((Ipv4Addr::LOCALHOST, config.port)) + .await + .unwrap(); + } + + #[tokio::test] + async fn ordinary_stream_disconnect_releases_upstream_and_admission_without_stop() { + let manager = manager(); + let config = config(); + let dropped = Arc::new(AtomicBool::new(false)); + let router = Router::new().route( + "/v1/chat/completions", + post({ + let dropped = dropped.clone(); + move || { + let guard = DropFlag(dropped.clone()); + async move { + Body::from_stream(futures_util::stream::unfold( + (true, guard), + |(first, guard)| async move { + if first { + Some(( + Ok::<_, io::Error>(Bytes::from_static(b"first")), + (false, guard), + )) + } else { + std::future::pending().await + } + }, + )) + } + } + }), + ); + start(&manager, config.clone(), router).await; + let mut response = client() + .post(endpoint(&config, "/v1/chat/completions")) + .bearer_auth("fixture-key") + .send() + .await + .unwrap(); + assert_eq!(response.chunk().await.unwrap().unwrap(), "first"); + assert_eq!(manager.status().active_requests, 1); + drop(response); + wait_for(&manager, |status| status.active_requests == 0).await; + assert!(dropped.load(Ordering::SeqCst)); + assert_eq!(manager.status().state, LocalApiState::Running); + manager.stop().await; + } + + #[tokio::test] + async fn identical_account_restore_preserves_live_sse_and_its_admission_permit() { + let manager = manager(); + let config = config(); + let resume = Arc::new(Notify::new()); + let dropped = Arc::new(AtomicBool::new(false)); + let router = Router::new().route( + "/v1/chat/completions", + post({ + let resume = resume.clone(); + let dropped = dropped.clone(); + move || { + let guard = DropFlag(dropped.clone()); + let resume = resume.clone(); + async move { + let body = Body::from_stream(futures_util::stream::unfold( + (0, guard, resume), + |(index, guard, resume)| async move { + let chunk = match index { + 0 => Bytes::from_static(b"data: first\n\n"), + 1 => { + resume.notified().await; + Bytes::from_static(b"data: second\n\n") + } + _ => std::future::pending().await, + }; + Some((Ok::<_, io::Error>(chunk), (index + 1, guard, resume))) + }, + )); + ([(header::CONTENT_TYPE, "text/event-stream")], body) + } + } + }), + ); + start(&manager, config.clone(), router).await; + let mut response = client() + .post(endpoint(&config, "/v1/chat/completions")) + .bearer_auth("fixture-key") + .send() + .await + .unwrap(); + assert_eq!(response.chunk().await.unwrap().unwrap(), "data: first\n\n"); + let before = manager.status(); + let generation = manager.generation(); + assert_eq!(before.active_requests, 1); + let restored = manager + .configure_if_current(config, "local-api-test-owner".into(), generation) + .await + .unwrap(); + assert_eq!(restored, before); + assert_eq!(manager.generation(), generation); + assert!(!dropped.load(Ordering::SeqCst)); + resume.notify_one(); + assert_eq!(response.chunk().await.unwrap().unwrap(), "data: second\n\n"); + assert_eq!(manager.status().active_requests, 1); + let stopped = manager.stop().await; + assert_eq!(stopped.active_requests, 0); + assert!(dropped.load(Ordering::SeqCst)); + } + + #[tokio::test] + async fn stop_closes_clients_stalled_before_headers_and_during_upload() { + let manager = manager(); + let config = config(); + start( + &manager, + config.clone(), + Router::new().route("/v1/chat/completions", post(|_: Bytes| async { "ok" })), + ) + .await; + let mut headers = TcpStream::connect((Ipv4Addr::LOCALHOST, config.port)) + .await + .unwrap(); + headers + .write_all(b"POST /v1/chat/completions HTTP/1.1\r\nHost: ") + .await + .unwrap(); + let mut upload = TcpStream::connect((Ipv4Addr::LOCALHOST, config.port)) + .await + .unwrap(); + upload.write_all(format!("POST /v1/chat/completions HTTP/1.1\r\nHost: 127.0.0.1:{}\r\nAuthorization: Bearer fixture-key\r\nContent-Length: 1000\r\n\r\nx", config.port).as_bytes()).await.unwrap(); + wait_for(&manager, |status| status.active_requests == 1).await; + assert_eq!(manager.stop().await.active_requests, 0); + assert_closed(headers).await; + assert_closed(upload).await; + } + + #[tokio::test] + async fn stop_invalidates_queued_start_and_a_previously_validated_configuration() { + let manager = manager(); + let config = config(); + let lock = manager.inner.running.lock().await; + let starting = { + let manager = manager.clone(); + let config = config.clone(); + tokio::spawn(async move { manager.start(config, "old-owner".into()).await }) + }; + wait_for(&manager, |status| status.state == LocalApiState::Starting).await; + let old_generation = manager.generation(); + let stopping = { + let manager = manager.clone(); + tokio::spawn(async move { manager.stop().await }) + }; + wait_for(&manager, |status| status.state == LocalApiState::Stopping).await; + drop(lock); + starting.await.unwrap().unwrap(); + let stopped = stopping.await.unwrap(); + assert_eq!(stopped.state, LocalApiState::Stopped); + assert_eq!(stopped.owner, None); + let delayed = manager + .configure_if_current(config.clone(), "old-owner".into(), old_generation) + .await + .unwrap(); + assert_eq!(delayed.state, LocalApiState::Stopped); + assert_eq!(delayed.owner, None); + TcpListener::bind((Ipv4Addr::LOCALHOST, config.port)) + .await + .unwrap(); + let disabled = LocalApiConfig { + enabled: false, + ..config + }; + assert_eq!( + manager + .configure(disabled.clone(), String::new()) + .await + .unwrap() + .config, + disabled + ); + } + + #[tokio::test] + async fn dropping_last_manager_closes_owned_listener_and_connections() { + let manager = manager(); + let config = config(); + start(&manager, config.clone(), Router::new()).await; + let mut stream = TcpStream::connect((Ipv4Addr::LOCALHOST, config.port)) + .await + .unwrap(); + stream.write_all(b"GET / HTTP/1.1\r\n").await.unwrap(); + drop(manager); + assert_closed(stream).await; + TcpListener::bind((Ipv4Addr::LOCALHOST, config.port)) + .await + .unwrap(); + } + + #[tokio::test] + async fn stale_published_intent_cannot_acquire_a_newer_runtime_for_teardown() { + let manager = manager(); + let config = config(); + for owner in [Some("stale-start".to_owned()), None] { + // Pause immediately after an old start/stop publishes intent, then + // let a newer start publish and acquire the runtime mutex first. + let (stale_generation, _) = manager.begin(None, owner, None).unwrap(); + start( + &manager, + config.clone(), + Router::new().route("/health", get(|| async { "ok" })), + ) + .await; + assert!(manager.running_if_current(stale_generation).await.is_none()); + assert_eq!( + client() + .get(endpoint(&config, "/health")) + .send() + .await + .unwrap() + .status(), + StatusCode::OK + ); + assert_eq!(manager.status().state, LocalApiState::Running); + } + manager.stop().await; + } + + #[tokio::test] + async fn shared_cli_browser_policy_retains_opt_in_preflight_and_bearer_headers() { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let cancel = CancellationToken::new(); + let router = apply_browser_policy( + Router::new().route("/v1/models", get(|| async { "ok" })), + true, + ); + let task = tokio::spawn(serve(listener, router, cancel.clone(), Weak::new(), 0)); + let url = format!("http://127.0.0.1:{port}/v1/models"); + let preflight = client() + .request(Method::OPTIONS, &url) + .header(header::ORIGIN, "https://caller.example") + .header(header::ACCESS_CONTROL_REQUEST_METHOD, "GET") + .header( + header::ACCESS_CONTROL_REQUEST_HEADERS, + "authorization,x-client-test", + ) + .send() + .await + .unwrap(); + assert!(preflight.status().is_success()); + assert_eq!( + preflight + .headers() + .get(header::ACCESS_CONTROL_ALLOW_ORIGIN) + .unwrap(), + "*" + ); + let allowed = preflight + .headers() + .get(header::ACCESS_CONTROL_ALLOW_HEADERS) + .unwrap() + .to_str() + .unwrap(); + assert!(allowed.contains("authorization")); + assert!(allowed.contains("x-client-test")); + let response = client() + .get(&url) + .header(header::ORIGIN, "https://caller.example") + .bearer_auth("fixture-key") + .send() + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + cancel.cancel(); + task.await.unwrap(); + } +} diff --git a/apps/maple-agent/app/src/main.rs b/apps/maple-agent/app/src/main.rs index 8653045d3..abc4ccc39 100644 --- a/apps/maple-agent/app/src/main.rs +++ b/apps/maple-agent/app/src/main.rs @@ -16,6 +16,8 @@ mod host; mod hosted_oauth; #[cfg(feature = "desktop")] mod keymap; +#[cfg(feature = "proxy")] +mod local_api; #[cfg(feature = "desktop")] mod notify; #[cfg(feature = "desktop")] @@ -381,10 +383,6 @@ fn host_error_summary(error: &str) -> &'static str { /// enclave, for tools that speak the OpenAI API. Runs until killed. #[cfg(feature = "proxy")] fn run_proxy(args: ProxyArgs) -> Result<(), String> { - use axum::http::{Method, StatusCode, header::ORIGIN}; - use axum::response::IntoResponse; - use tower_http::cors::{AllowHeaders, Any, CorsLayer}; - let backend_url = maple_agent::maple_api::validate_api_url(&configured_api_url())?; let pcr0 = maple_agent::open_secret_config::configured_pcr0_environment()?; let mut config = maple_proxy::Config::new(args.host.clone(), args.port, backend_url) @@ -400,31 +398,7 @@ fn run_proxy(args: ProxyArgs) -> Result<(), String> { let addr = config .socket_addr() .map_err(|error| format!("Invalid proxy address: {error}"))?; - let app = maple_proxy::create_app(config); - let app = if args.cors { - app.layer( - CorsLayer::new() - .allow_origin(Any) - .allow_methods([Method::GET, Method::POST, Method::OPTIONS]) - // Authorization is not covered by `*` under Fetch; mirror - // the preflight list so bearer keys and SDK headers pass. - .allow_headers(AllowHeaders::mirror_request()), - ) - } else { - // Disabling CORS alone only hides responses. A no-cors browser POST - // can still reach loopback and spend the saved key, so fail closed - // on browser-only headers before the body is read. - app.layer(axum::middleware::from_fn( - |request: axum::extract::Request, next: axum::middleware::Next| async move { - if request.headers().contains_key(ORIGIN) - || request.headers().contains_key("sec-fetch-site") - { - return StatusCode::FORBIDDEN.into_response(); - } - next.run(request).await - }, - )) - }; + let app = local_api::apply_browser_policy(maple_proxy::create_app(config), args.cors); let runtime = tokio::runtime::Builder::new_multi_thread() .enable_all() diff --git a/apps/maple-agent/app/src/remote/control.rs b/apps/maple-agent/app/src/remote/control.rs index 5ec5efb08..fecf58f07 100644 --- a/apps/maple-agent/app/src/remote/control.rs +++ b/apps/maple-agent/app/src/remote/control.rs @@ -520,7 +520,8 @@ pub(super) fn remote_event(event: &AgentServiceEvent) -> bool { | AgentServiceEvent::SideQuestion { .. } => true, AgentServiceEvent::HistoryReplaced { .. } | AgentServiceEvent::PersistenceFailureChanged { .. } => true, - AgentServiceEvent::StoreStateChanged { .. } + AgentServiceEvent::LocalApiChanged + | AgentServiceEvent::StoreStateChanged { .. } | AgentServiceEvent::AccountChanged(_) | AgentServiceEvent::RemoteAccessEnded(_) | AgentServiceEvent::HostReconnecting @@ -781,6 +782,11 @@ mod tests { use super::*; use tokio::io::AsyncWriteExt; + #[test] + fn local_api_status_never_enters_remote_event_streams() { + assert!(!remote_event(&AgentServiceEvent::LocalApiChanged)); + } + #[test] fn peer_info_journal_failure_keeps_baked_identity_and_blocks_installation() { let private_error = "journal-canary-account/private-path"; diff --git a/apps/maple-agent/app/src/ui/chat/mod.rs b/apps/maple-agent/app/src/ui/chat/mod.rs index 0aa45dcb5..d4146cbc7 100644 --- a/apps/maple-agent/app/src/ui/chat/mod.rs +++ b/apps/maple-agent/app/src/ui/chat/mod.rs @@ -6097,7 +6097,8 @@ impl ChatScreen { self.loading_timeline.insert(item.id.clone(), item.clone()); } match event { - AgentServiceEvent::AccountChanged(_) + AgentServiceEvent::LocalApiChanged + | AgentServiceEvent::AccountChanged(_) | AgentServiceEvent::StoreStateChanged { .. } | AgentServiceEvent::PersistenceFailureChanged { .. } => { return false; diff --git a/apps/maple-agent/app/src/ui/settings.rs b/apps/maple-agent/app/src/ui/settings.rs index f651f3dd9..1e4c96069 100644 --- a/apps/maple-agent/app/src/ui/settings.rs +++ b/apps/maple-agent/app/src/ui/settings.rs @@ -36,11 +36,15 @@ mod api_credits; mod api_keys; mod billing; mod licenses; +#[cfg(feature = "proxy")] +mod local_api; mod mcp; mod navigation; use self::account::AccountState; use self::api_keys::ApiKeysState; use self::billing::BillingState; +#[cfg(feature = "proxy")] +use self::local_api::LocalApiUiState; use self::mcp::{McpSecretAction, McpSecretField, definition_edit, name_collides}; use self::navigation::{GeneralTarget, SettingsApplicationVimState, SettingsTarget}; @@ -60,6 +64,8 @@ pub enum Section { Account, Billing, ApiKeys, + #[cfg(feature = "proxy")] + LocalApi, Shortcuts, Prompt, Integrations, @@ -76,6 +82,8 @@ impl Section { Self::Account => "Account", Self::Billing => "Billing", Self::ApiKeys => "API keys", + #[cfg(feature = "proxy")] + Self::LocalApi => "Local API", Self::Shortcuts => "Keyboard Shortcuts", Self::Prompt => "System prompt", Self::Integrations => "Integrations", @@ -86,11 +94,13 @@ impl Section { } } - const ALL: [Self; 11] = [ + const ALL: &[Self] = &[ Self::General, Self::Account, Self::Billing, Self::ApiKeys, + #[cfg(feature = "proxy")] + Self::LocalApi, Self::Shortcuts, Self::Prompt, Self::Integrations, @@ -148,6 +158,8 @@ pub struct SettingsScreen { account: AccountState, billing: BillingState, api_keys: ApiKeysState, + #[cfg(feature = "proxy")] + local_api: LocalApiUiState, /// Plan usage meter, same source as the sidebar card. plan: Option, /// Account MCP servers; None until loaded. @@ -425,6 +437,8 @@ impl SettingsScreen { account: AccountState::new(application_vim_enabled, application_focus.clone(), cx), billing: BillingState::new(), api_keys: ApiKeysState::new(application_vim_enabled, application_focus.clone(), cx), + #[cfg(feature = "proxy")] + local_api: LocalApiUiState::new(application_vim_enabled, application_focus.clone(), cx), plan: None, mcp_servers: None, mcp_editor: None, @@ -460,6 +474,12 @@ impl SettingsScreen { this.load_mcp_servers(cx); this.load_integrations(cx); this.load_advanced_status(cx); + #[cfg(feature = "proxy")] + let this = { + let mut this = this; + this.refresh_local_api_status(cx); + this + }; this } @@ -1504,6 +1524,10 @@ impl SettingsScreen { // A dropdown belongs to the pane that opened it. self.close_setting_menu(cx); self.section = section; + #[cfg(feature = "proxy")] + if section == Section::LocalApi { + self.refresh_local_api_status(cx); + } self.remote_screen.update(cx, |screen, cx| { screen.set_active(section == Section::Remote, cx) }); @@ -1881,6 +1905,10 @@ impl SettingsScreen { Section::ApiKeys => { pane = pane.child(self.render_api_keys_pane(cx)); } + #[cfg(feature = "proxy")] + Section::LocalApi => { + pane = pane.child(self.render_local_api_pane(cx)); + } Section::Shortcuts => { pane = pane.child(self.render_shortcuts_pane(cx)); } diff --git a/apps/maple-agent/app/src/ui/settings/local_api.rs b/apps/maple-agent/app/src/ui/settings/local_api.rs new file mode 100644 index 000000000..3fd43b795 --- /dev/null +++ b/apps/maple-agent/app/src/ui/settings/local_api.rs @@ -0,0 +1,715 @@ +//! This computer's optional HTTP inference listener. Account keys are created +//! in the existing API keys pane and supplied by each calling application. + +use gpui::{Context, Div, Entity, Focusable, SharedString, Window, div, prelude::*}; + +use super::{Section, SettingsScreen, SettingsTarget, section_title, setting_copy}; +use crate::local_api::{LocalApiConfig, LocalApiState, LocalApiStatus}; +use crate::ui::{text_input::TextInput, theme, widgets}; + +#[derive(Clone, Debug, Eq, PartialEq)] +pub(super) enum LocalApiTarget { + Enabled, + Retry, + Port, + SavePort, + CopyUrl, + ApiKeys, + Refresh, +} + +pub(super) struct LocalApiUiState { + status: Option, + pub(super) port: Entity, + base_url: SharedString, + request_count: SharedString, + saving: bool, + refreshing: bool, + host_epoch: u64, + refresh_generation: u64, + notice: Option<&'static str>, +} + +impl LocalApiUiState { + pub(super) fn new( + application_vim_enabled: bool, + application_focus: gpui::FocusHandle, + cx: &mut Context, + ) -> Self { + Self { + status: None, + port: cx.new(move |cx| { + TextInput::new("Port (1–65535)", cx) + .with_tab_index(19) + .application_vim(application_vim_enabled) + .on_application_escape(move |window, cx| window.focus(&application_focus, cx)) + }), + base_url: SharedString::default(), + request_count: SharedString::default(), + saving: false, + refreshing: false, + host_epoch: 0, + refresh_generation: 0, + notice: None, + } + } + + fn can_configure(&self) -> bool { + self.status.is_some() && !self.saving + } + + fn can_edit_port(&self) -> bool { + self.can_configure() + && self.status.as_ref().is_some_and(|status| { + matches!(status.state, LocalApiState::Stopped | LocalApiState::Failed) + }) + } + + fn can_retry(&self) -> bool { + self.can_configure() + && self.status.as_ref().is_some_and(|status| { + status.config.enabled + && matches!(status.state, LocalApiState::Stopped | LocalApiState::Failed) + }) + } +} + +impl SettingsScreen { + #[cfg(test)] + pub(crate) fn local_api_status_for_test(&self) -> Option { + self.local_api.status.clone() + } + + /// Called only with this computer's host status. A selected remote agent + /// never changes the backend used by this pane. + pub(crate) fn set_local_api_status(&mut self, status: LocalApiStatus, cx: &mut Context) { + if self + .local_api + .status + .as_ref() + .is_some_and(|current| status.revision < current.revision) + { + return; + } + let port_changed = self + .local_api + .status + .as_ref() + .is_none_or(|current| current.config.port != status.config.port); + self.local_api.base_url = format!("http://127.0.0.1:{}/v1", status.config.port).into(); + self.local_api.request_count = format!("{} active requests", status.active_requests).into(); + self.local_api.status = Some(status); + if port_changed { + let port = self + .local_api + .status + .as_ref() + .unwrap() + .config + .port + .to_string(); + self.local_api + .port + .update(cx, |input, cx| input.set_text(&port, cx)); + } + self.reconcile_application_vim_target(); + cx.notify(); + } + + /// Refresh on pane entry and after reconnecting to the local host. + pub(crate) fn refresh_local_api_status(&mut self, cx: &mut Context) { + self.local_api.refreshing = true; + self.local_api.notice = None; + self.local_api.refresh_generation = self.local_api.refresh_generation.wrapping_add(1); + let generation = self.local_api.refresh_generation; + let host_epoch = self.local_api.host_epoch; + let backend = self.backend.clone(); + self.call( + async move { backend.get_local_api_status().await }, + cx, + move |this, result, cx| { + if host_epoch != this.local_api.host_epoch + || generation != this.local_api.refresh_generation + { + return; + } + this.local_api.refreshing = false; + match result { + Ok(status) => { + this.set_local_api_status(status, cx); + } + Err(_) => { + this.local_api.notice = Some( + "Could not read the local API status. Reconnect to this computer and try again.", + ); + } + } + this.reconcile_application_vim_target(); + cx.notify(); + }, + ); + self.reconcile_application_vim_target(); + cx.notify(); + } + + /// A replacement host starts a new revision sequence. Old requests must + /// neither install a snapshot nor clear a newer host's pending operation. + pub(crate) fn local_api_host_reconnected(&mut self, cx: &mut Context) { + self.local_api.host_epoch = self.local_api.host_epoch.wrapping_add(1); + self.local_api.status = None; + self.local_api.saving = false; + self.refresh_local_api_status(cx); + } + + fn toggle_local_api(&mut self, cx: &mut Context) { + if !self.local_api.can_configure() { + return; + } + let Some(status) = &self.local_api.status else { + return; + }; + let enabled = !status.config.enabled; + let port = if enabled { + match parse_port(self.local_api.port.read(cx).text_ref()) { + Ok(port) => port, + Err(message) => { + self.local_api.notice = Some(message); + cx.notify(); + return; + } + } + } else { + status.config.port + }; + self.save_local_api_config(LocalApiConfig { enabled, port }, cx); + } + + fn save_local_api_port(&mut self, cx: &mut Context) { + if !self.local_api.can_edit_port() { + return; + } + let port = match parse_port(self.local_api.port.read(cx).text_ref()) { + Ok(port) => port, + Err(message) => { + self.local_api.notice = Some(message); + cx.notify(); + return; + } + }; + let enabled = self.local_api.status.as_ref().unwrap().config.enabled; + self.save_local_api_config(LocalApiConfig { enabled, port }, cx); + } + + fn retry_local_api(&mut self, cx: &mut Context) { + if self.local_api.can_retry() { + let config = self.local_api.status.as_ref().unwrap().config.clone(); + self.save_local_api_config(config, cx); + } + } + + fn save_local_api_config(&mut self, config: LocalApiConfig, cx: &mut Context) { + self.local_api.saving = true; + self.local_api.notice = None; + let backend = self.backend.clone(); + let user_id = self.user_id.clone(); + let host_epoch = self.local_api.host_epoch; + self.call( + async move { backend.set_local_api_config(&user_id, config).await }, + cx, + move |this, result, cx| { + if host_epoch != this.local_api.host_epoch { + return; + } + this.local_api.saving = false; + match result { + Ok(status) => this.set_local_api_status(status, cx), + Err(_) => { + this.refresh_local_api_status(cx); + this.local_api.notice = Some( + "Could not update the local API. Refresh its status and try again.", + ); + } + } + this.reconcile_application_vim_target(); + cx.notify(); + }, + ); + self.reconcile_application_vim_target(); + cx.notify(); + } + + pub(super) fn local_api_targets(&self) -> Vec { + let target = SettingsTarget::LocalApi; + let mut targets = Vec::new(); + if self.local_api.can_configure() { + targets.push(target(LocalApiTarget::Enabled)); + } + if self.local_api.can_retry() { + targets.push(target(LocalApiTarget::Retry)); + } + if self.local_api.can_edit_port() { + targets.extend([ + target(LocalApiTarget::Port), + target(LocalApiTarget::SavePort), + ]); + } + if self.local_api.status.is_some() { + targets.push(target(LocalApiTarget::CopyUrl)); + } + targets.push(target(LocalApiTarget::ApiKeys)); + if !self.local_api.refreshing && !self.local_api.saving { + targets.push(target(LocalApiTarget::Refresh)); + } + targets + } + + pub(super) fn activate_local_api_target( + &mut self, + target: LocalApiTarget, + window: &mut Window, + cx: &mut Context, + ) { + match target { + LocalApiTarget::Enabled => self.toggle_local_api(cx), + LocalApiTarget::Retry => self.retry_local_api(cx), + LocalApiTarget::Port if self.local_api.can_edit_port() => { + let focus = self.local_api.port.read(cx).focus_handle(cx); + window.focus(&focus, cx); + } + LocalApiTarget::Port => {} + LocalApiTarget::SavePort => self.save_local_api_port(cx), + LocalApiTarget::CopyUrl => self.copy_local_api_url(cx), + LocalApiTarget::ApiKeys => self.select_section(Section::ApiKeys, cx), + LocalApiTarget::Refresh => self.refresh_local_api_status(cx), + } + } + + fn copy_local_api_url(&self, cx: &mut Context) { + if self.local_api.status.is_some() { + cx.write_to_clipboard(gpui::ClipboardItem::new_string( + self.local_api.base_url.to_string(), + )); + } + } + + pub(super) fn render_local_api_pane(&self, cx: &mut Context) -> Div { + let state = &self.local_api; + let status = state.status.as_ref(); + let enabled = status.is_some_and(|status| status.config.enabled); + let label = if state.refreshing && status.is_none() { + "Checking status…" + } else { + status.map_or("Status unavailable", state_label) + }; + let mut pane = div() + .flex() + .flex_col() + .gap_4() + .child(section_title("Local API")) + .child(setting_copy( + "Connect AI applications on this computer", + "Expose Maple's inference APIs at a local address. Requests use Maple's secure backend and your API credits.", + )) + .child( + widgets::card_row() + .flex() + .items_center() + .justify_between() + .gap_4() + .child(setting_copy( + "Enable local API", + "Applies to this computer, including while you work with a remote agent.", + )) + .child(self.application_target( + || SettingsTarget::LocalApi(LocalApiTarget::Enabled), + widgets::switch("local-api-enabled", enabled) + .debug_selector(|| "local-api-enabled".to_string()) + .aria_label("Enable local API on this computer") + .when(!state.can_configure(), |switch| { + switch.opacity(0.5).aria_description( + "Unavailable while local API settings are loading or being updated.", + ) + }) + .when(state.can_configure(), |switch| { + switch + .tab_index(18) + .on_click(cx.listener(|this, _, _, cx| this.toggle_local_api(cx))) + .on_key_down(cx.listener(|this, event: &gpui::KeyDownEvent, _, cx| { + if activates(event) { + cx.stop_propagation(); + this.toggle_local_api(cx); + } + })) + }), + )), + ) + .child( + div() + .id("local-api-status") + .role(gpui::Role::Status) + .flex() + .flex_col() + .gap_1() + .text_sm() + .text_color(gpui::rgb(theme::text_secondary())) + .child(label) + .when(state.saving, |row| row.child("Updating local API settings…")) + .when(status.is_some_and(|status| status.state == LocalApiState::Running), |row| { + row.child(state.request_count.clone()) + }) + .when(status.is_some_and(|status| status.state == LocalApiState::Stopped && status.config.enabled), |row| { + row.child("The local API is enabled but paused. Select Resume to accept requests on this computer.") + }), + ); + if let Some(message) = state.notice { + pane = pane.child(widgets::banner(theme::status_error()).child(message)); + } + if let Some(error) = status.and_then(|status| status.error.as_ref()) { + pane = pane.child(widgets::banner(theme::status_error()).child(error.clone())); + } + if state.can_retry() { + pane = pane.child( + self.application_target( + || SettingsTarget::LocalApi(LocalApiTarget::Retry), + widgets::secondary_button("local-api-retry") + .tab_index(19) + .on_click(cx.listener(|this, _, _, cx| this.retry_local_api(cx))) + .on_key_down(cx.listener(|this, event: &gpui::KeyDownEvent, _, cx| { + if activates(event) { + cx.stop_propagation(); + this.retry_local_api(cx); + } + })) + .child(retry_label(status.expect("retry requires a status"))), + ), + ); + } + if let Some(status) = status { + let port_control = if state.can_edit_port() { + self.application_target( + || SettingsTarget::LocalApi(LocalApiTarget::Port), + widgets::input_frame() + .w(gpui::px(120.)) + .child(state.port.clone()), + ) + } else { + div() + .text_sm() + .text_color(gpui::rgb(theme::text_muted())) + .child(status.config.port.to_string()) + .into_any_element() + }; + pane = pane + .child( + widgets::card_row() + .flex() + .items_center() + .justify_between() + .gap_4() + .child(setting_copy("Port", "The local address always uses 127.0.0.1. Stop the API to change its port.")) + .child(div().flex().items_center().gap_2().child(port_control).when(state.can_edit_port(), |row| { + row.child(self.application_target( + || SettingsTarget::LocalApi(LocalApiTarget::SavePort), + widgets::secondary_button("local-api-save-port") + .tab_index(20) + .on_click(cx.listener(|this, _, _, cx| this.save_local_api_port(cx))) + .on_key_down(cx.listener(|this, event: &gpui::KeyDownEvent, _, cx| { + if activates(event) { + cx.stop_propagation(); + this.save_local_api_port(cx); + } + })) + .child(if !enabled { + "Save port" + } else if status.state == LocalApiState::Stopped { + "Save and resume" + } else { + "Save and retry" + }), + )) + })), + ) + .child( + widgets::card_row() + .flex() + .items_center() + .justify_between() + .gap_4() + .child( + div().flex().flex_col().gap_1() + .child(setting_copy("OpenAI-compatible base URL", "Paste this address into your AI application's provider settings.")) + .child(div().font_family(crate::assets::FONT_MONO).text_sm().child(state.base_url.clone())), + ) + .child(self.application_target( + || SettingsTarget::LocalApi(LocalApiTarget::CopyUrl), + widgets::copy_button("local-api-copy-url", state.base_url.clone(), None, Some(cx.entity_id())) + .aria_label("Copy local API base URL") + .tab_index(21) + .on_key_down(cx.listener(|this, event: &gpui::KeyDownEvent, _, cx| { + if activates(event) { + cx.stop_propagation(); + this.copy_local_api_url(cx); + } + })), + )), + ); + } + pane.child( + div().flex().flex_col().gap_2().text_sm().text_color(gpui::rgb(theme::text_secondary())) + .child("When enabled, the API resumes when this computer's Maple host starts and its store is unlocked. Signing out, locking, or quitting the host stops the listener.") + .child("Create a dedicated Maple API key, copy it once, and enter it in your AI application's API key field. Every request must supply that key as Authorization: Bearer .") + .child(self.application_target( + || SettingsTarget::LocalApi(LocalApiTarget::ApiKeys), + widgets::secondary_button("local-api-open-keys") + .tab_index(22) + .on_click(cx.listener(|this, _, _, cx| this.select_section(Section::ApiKeys, cx))) + .on_key_down(cx.listener(|this, event: &gpui::KeyDownEvent, _, cx| { + if activates(event) { + cx.stop_propagation(); + this.select_section(Section::ApiKeys, cx); + } + })) + .child("Manage API keys"), + )), + ).child( + widgets::card_row().flex().flex_col().gap_2() + .child(section_title("Available endpoints")) + .children([ + "GET /v1/models", + "POST /v1/chat/completions", + "POST /v1/embeddings", + "POST /v1/systemone", + ].into_iter().map(|route| div().font_family(crate::assets::FONT_MONO).text_xs().child(route))) + .child(div().text_xs().text_color(gpui::rgb(theme::text_muted())).child("Browser-origin requests are blocked. Use a desktop application, CLI, or server client.")), + ).when(!state.refreshing && !state.saving, |pane| { + pane.child(self.application_target( + || SettingsTarget::LocalApi(LocalApiTarget::Refresh), + widgets::ghost_button("local-api-refresh") + .tab_index(23) + .on_click(cx.listener(|this, _, _, cx| this.refresh_local_api_status(cx))) + .on_key_down(cx.listener(|this, event: &gpui::KeyDownEvent, _, cx| { + if activates(event) { + cx.stop_propagation(); + this.refresh_local_api_status(cx); + } + })) + .child("Refresh status"), + )) + }) + } +} + +fn parse_port(value: &str) -> Result { + value + .trim() + .parse::() + .ok() + .filter(|port| *port > 0) + .ok_or("Enter a port between 1 and 65535.") +} + +fn state_label(status: &LocalApiStatus) -> &'static str { + match status.state { + LocalApiState::Stopped if status.config.enabled => "Paused", + LocalApiState::Stopped => "Stopped", + LocalApiState::Starting => "Starting…", + LocalApiState::Running => "Running on this computer", + LocalApiState::Stopping => "Stopping…", + LocalApiState::Failed => "Could not start", + } +} + +fn retry_label(status: &LocalApiStatus) -> &'static str { + if status.state == LocalApiState::Stopped { + "Resume" + } else { + "Retry start" + } +} + +fn activates(event: &gpui::KeyDownEvent) -> bool { + !event.keystroke.modifiers.modified() + && matches!(event.keystroke.key.as_str(), "enter" | "space" | " ") +} + +#[cfg(test)] +mod tests { + use super::*; + use gpui::{AppContext, TestAppContext}; + + fn screen(cx: &mut TestAppContext) -> Entity { + cx.executor().allow_parking(); + let backend = std::sync::Arc::new( + crate::backend::AgentBackend::for_test("http://127.0.0.1:9".into(), String::new()) + .expect("test backend"), + ); + cx.new(|cx| { + SettingsScreen::new( + backend.clone(), + backend, + "user".into(), + crate::settings::AppSettings { + application_vim_enabled: true, + ..Default::default() + }, + crate::shortcuts::ShortcutSnapshot { + generation: 1, + rows: Vec::new(), + last_error: None, + compatibility_warning: None, + }, + Section::LocalApi, + cx, + ) + }) + } + + fn status(revision: u64, state: LocalApiState, enabled: bool, port: u16) -> LocalApiStatus { + LocalApiStatus { + config: LocalApiConfig { enabled, port }, + state, + owner: Some("user".into()), + revision, + ..Default::default() + } + } + + #[test] + fn invalid_ports_cannot_enable_an_ephemeral_or_wrapped_listener() { + for value in ["", "0", "65536", "-1", "1.5", "localhost", "8080/tcp"] { + assert!(parse_port(value).is_err(), "{value:?}"); + } + assert_eq!(parse_port("1"), Ok(1)); + assert_eq!(parse_port(" 65535 "), Ok(65535)); + } + + #[gpui::test] + fn pushed_status_rejects_stale_replies_and_preserves_unsaved_port(cx: &mut TestAppContext) { + let settings = screen(cx); + settings.update(cx, |this, cx| { + this.local_api.refreshing = false; + this.set_local_api_status(status(100, LocalApiState::Stopped, false, 8123), cx); + this.local_api + .port + .update(cx, |input, cx| input.set_text("9123", cx)); + this.set_local_api_status(status(101, LocalApiState::Running, true, 8123), cx); + assert_eq!(this.local_api.port.read(cx).text_ref(), "9123"); + assert_eq!(this.local_api.base_url.as_ref(), "http://127.0.0.1:8123/v1"); + + // A late stopped snapshot must not resurrect old configuration or + // re-enable port editing after the listener started. + this.set_local_api_status(status(99, LocalApiState::Stopped, false, 7123), cx); + assert_eq!( + this.local_api.status.as_ref().unwrap().state, + LocalApiState::Running + ); + assert!(!this.local_api.can_edit_port()); + assert_eq!(this.local_api.base_url.as_ref(), "http://127.0.0.1:8123/v1"); + + this.set_local_api_status(status(102, LocalApiState::Stopped, false, 9123), cx); + assert_eq!(this.local_api.port.read(cx).text_ref(), "9123"); + assert_eq!(this.local_api.base_url.as_ref(), "http://127.0.0.1:9123/v1"); + assert!(this.local_api.can_edit_port()); + }); + } + + #[gpui::test] + fn keyboard_targets_follow_listener_state_and_retry_availability(cx: &mut TestAppContext) { + let settings = screen(cx); + settings.update(cx, |this, cx| { + this.local_api.refreshing = false; + this.set_local_api_status(status(100, LocalApiState::Stopped, false, 8123), cx); + let target = SettingsTarget::LocalApi; + assert!( + this.local_api_targets() + .contains(&target(LocalApiTarget::Port)) + ); + assert!( + !this + .local_api_targets() + .contains(&target(LocalApiTarget::Retry)) + ); + for (revision, state) in [ + LocalApiState::Starting, + LocalApiState::Running, + LocalApiState::Stopping, + ] + .into_iter() + .enumerate() + { + this.set_local_api_status(status(101 + revision as u64, state, true, 8123), cx); + assert!( + !this + .local_api_targets() + .contains(&target(LocalApiTarget::Port)) + ); + assert!( + !this + .local_api_targets() + .contains(&target(LocalApiTarget::SavePort)) + ); + } + this.set_local_api_status(status(104, LocalApiState::Failed, true, 8123), cx); + assert!( + this.local_api_targets() + .contains(&target(LocalApiTarget::Retry)) + ); + this.local_api.saving = true; + assert_eq!( + this.local_api_targets(), + vec![ + target(LocalApiTarget::CopyUrl), + target(LocalApiTarget::ApiKeys), + ] + ); + }); + } + + #[gpui::test] + fn invalid_typed_port_is_rejected_before_changing_enabled_state(cx: &mut TestAppContext) { + let settings = screen(cx); + settings.update(cx, |this, cx| { + this.local_api.refreshing = false; + this.set_local_api_status(status(100, LocalApiState::Stopped, false, 8123), cx); + this.local_api + .port + .update(cx, |input, cx| input.set_text("0", cx)); + this.toggle_local_api(cx); + assert!(!this.local_api.saving); + assert!(!this.local_api.status.as_ref().unwrap().config.enabled); + assert_eq!( + this.local_api.notice, + Some("Enter a port between 1 and 65535.") + ); + }); + } + + #[gpui::test] + fn enabled_but_stopped_api_offers_explicit_resume(cx: &mut TestAppContext) { + let settings = screen(cx); + settings.update(cx, |this, cx| { + this.local_api.refreshing = false; + this.set_local_api_status(status(100, LocalApiState::Stopped, true, 8123), cx); + let snapshot = this.local_api.status.as_ref().unwrap(); + assert_eq!(state_label(snapshot), "Paused"); + assert_eq!(retry_label(snapshot), "Resume"); + assert!( + this.local_api_targets() + .contains(&SettingsTarget::LocalApi(LocalApiTarget::Retry)) + ); + + this.set_local_api_status(status(101, LocalApiState::Failed, true, 8123), cx); + assert_eq!( + retry_label(this.local_api.status.as_ref().unwrap()), + "Retry start" + ); + + this.set_local_api_status(status(102, LocalApiState::Stopped, false, 8123), cx); + assert_eq!( + state_label(this.local_api.status.as_ref().unwrap()), + "Stopped" + ); + assert!(!this.local_api.can_retry()); + }); + } +} diff --git a/apps/maple-agent/app/src/ui/settings/navigation.rs b/apps/maple-agent/app/src/ui/settings/navigation.rs index 3c1e1a2ea..c6ca61782 100644 --- a/apps/maple-agent/app/src/ui/settings/navigation.rs +++ b/apps/maple-agent/app/src/ui/settings/navigation.rs @@ -10,6 +10,8 @@ use gpui::{Context, Focusable, IntoElement, StatefulInteractiveElement, Window, use super::account::AccountTarget; use super::api_keys::ApiKeysTarget; use super::billing::BillingTarget; +#[cfg(feature = "proxy")] +use super::local_api::LocalApiTarget; use super::{ Section, SettingMenu, SettingsScreen, integration_can_setup, integration_can_toggle, integration_is_visible, @@ -45,6 +47,8 @@ pub(super) enum SettingsTarget { Account(AccountTarget), Billing(BillingTarget), ApiKeys(ApiKeysTarget), + #[cfg(feature = "proxy")] + LocalApi(LocalApiTarget), Shortcut(String), PromptEditor, PromptSave, @@ -87,6 +91,8 @@ impl SettingsScreen { let mut inputs = vec![self.prompt_editor.clone(), self.shortcut_search.clone()]; inputs.extend(self.account.inputs()); inputs.push(self.api_keys.name.clone()); + #[cfg(feature = "proxy")] + inputs.push(self.local_api.port.clone()); if let Some(editor) = &self.mcp_editor { inputs.extend([ editor.name.clone(), @@ -145,6 +151,8 @@ impl SettingsScreen { Section::Account => self.account_targets(), Section::Billing => self.billing_targets(), Section::ApiKeys => self.api_keys_targets(), + #[cfg(feature = "proxy")] + Section::LocalApi => self.local_api_targets(), Section::Shortcuts => self .shortcut_list_cache .visible_indices @@ -355,6 +363,10 @@ impl SettingsScreen { Some(SettingsTarget::ApiKeys(target)) => { self.activate_api_keys_target(target, window, cx) } + #[cfg(feature = "proxy")] + Some(SettingsTarget::LocalApi(target)) => { + self.activate_local_api_target(target, window, cx) + } Some(SettingsTarget::Shortcut(slot_id)) => self.begin_shortcut_recording(slot_id, cx), Some(SettingsTarget::PromptEditor) => { let handle = self.prompt_editor.read(cx).focus_handle(cx); @@ -425,6 +437,10 @@ impl SettingsScreen { fn application_text_input_focused(&self, window: &Window, cx: &gpui::App) -> bool { let focused = window.focused(cx); let mut inputs = vec![&self.prompt_editor, &self.shortcut_search]; + #[cfg(feature = "proxy")] + if self.section == Section::LocalApi { + inputs.push(&self.local_api.port); + } if let Some(editor) = &self.mcp_editor { inputs.extend([ &editor.name, diff --git a/apps/maple-agent/crates/maple-agent/src/agent/types.rs b/apps/maple-agent/crates/maple-agent/src/agent/types.rs index 8bfdf377d..d60ef2eda 100644 --- a/apps/maple-agent/crates/maple-agent/src/agent/types.rs +++ b/apps/maple-agent/crates/maple-agent/src/agent/types.rs @@ -656,6 +656,9 @@ pub enum AgentRunEvent { #[derive(Debug, Clone, Serialize, Deserialize)] pub enum AgentServiceEvent { + /// Content-free invalidation for this computer's optional API service. + /// The application fetches its local host snapshot; never sent remotely. + LocalApiChanged, /// Local store initialization, locking or maintenance changed. Never remote. StoreStateChanged { state: crate::store::StoreState, diff --git a/apps/maple-agent/docs/local-api.md b/apps/maple-agent/docs/local-api.md new file mode 100644 index 000000000..c308bb44a --- /dev/null +++ b/apps/maple-agent/docs/local-api.md @@ -0,0 +1,89 @@ +# Local API + +Settings → Local API controls an optional HTTP/1 endpoint on this computer. +The default is Off, bound only to `127.0.0.1`, on port 8080 (or the managed +workspace's `MAPLE_PORT` reservation). Change the port while stopped. The base +URL is `http://127.0.0.1:/v1`. + +Create a named Maple API key in Settings → API keys and put it in your AI +application's API-key field. Every inference request must supply +`Authorization: Bearer `. The service saves no fallback key. The key's +account determines authorization and usage; it need not be the desktop's +signed-in account. OpenSecret remains authoritative for key validity, plans, +models and quotas. Local refers to the endpoint: inference uses Maple's +attested encrypted SDK transport to the configured backend. + +The existing routes are GET `/v1/models`, POST `/v1/chat/completions`, +POST `/v1/embeddings`, POST `/v1/systemone`, and GET `/health` (also `/`). +Responses, audio and legacy `/v1/completions` are not exposed. Browser access +is disabled; requests carrying `Origin` or `Sec-Fetch-Site` are rejected. +Running means the socket is listening. It does not attest backend availability +or successful inference, and `/health` checks only local liveness. + +## Ownership + +The profile's local host owns the listener, all connections and active response +bodies. GPUI is a controller; it does not start a second runtime or an external +proxy process. Settings uses its local backend even when a task executes on a +paired remote computer. Proxy RPCs reject remote facades and are excluded from +the remote method allowlist. Local status invalidations never enter paired +remote event streams. No remote protocol change is involved. + +The open local store persists `local_api.config` under the selected account. +It contains only the desired enabled state and port. Actual state, safe bind +errors and active-request counts are host snapshots with monotonically ordered +revisions; reconnect starts a new UI epoch. These preferences and API keys are +separate. The proxy's SDK transport never borrows the desktop login session. + +Enabling requires a validated local account. A saved account ID alone cannot +autostart the endpoint. Enabled preferences resume after the profile's host +starts and its restore/unlock succeeds. This does not install an OS login +service. Reattaching a frontend to an identical running service leaves its +streams intact. Bind conflicts are errors, without automatic port changes or +process takeover. + +## Lifetime and limits + +An enabled listener retains the host after the window closes. Existing tray +and no-tray behavior applies; explicit Quit explains and stops the local API. +Logout/account transitions, destructive reset, draining store operations and +updates stop network admission before releasing account/store ownership. +Key-only password rewraps retain the existing non-draining lane. + +Stop and status remain available through private local IPC while storage is +locked or maintaining. The settings switch persists Off. Emergency +`stop_local_api` suspends the current service without accessing locked +preferences; an intervening emergency Stop prevents a maintenance operation +from resuming it. Automatic account restore keeps it paused for this host's +lifetime; use Resume or restart the daemon to resume persisted On. + +Generations fence pending authentication, configuration and bind work. Stop +invalidates pending starts before waiting for teardown. Requests never hold +account/store guards during model streams. Connection tasks are tracked and +cancelled, including incomplete headers, uploads, response setup and response +bodies. Shutdown bounds draining and closes remaining sockets. Local +cancellation cannot promise cancellation of provider work already accepted +upstream or reversal of charges. + +There are at most 32 accepted HTTP connections and 8 active requests. Request +permits are acquired before body buffering and retained until body completion +or drop. Excess requests receive 503; excess connections close. The shared +engine retains its 50 MiB body limit. Headers and uploads have bounded waits; +the engine owns inference setup and SSE idle timeouts. Persistent provider-cache +identity is intentionally separate; the existing ephemeral behavior is used. + +The existing `maple-agent proxy` command remains an independent foreground +mode with its documented host/CORS/key options. Only one runtime may own a +given port. Both modes use the same browser-policy helper. Builds without the +`proxy` feature omit the managed service and its UI; proxy/headless builds do +not depend on GPUI. + +## Verification + +Run the component's pinned `just ci`. Focused `cargo test -p maple-agent-app +local_api` covers real loopback connections, permit lifetimes, blocked clients, +stale starts, redundant restore, locked controls, account/store teardown and +rendered Settings ownership across distinct execution profiles. The UI fixtures +exercise real private IPC and listeners with synthetic account effects; they +are not a real Iroh pairing or live backend inference check. Validate those +separately using an isolated linked Local workspace and the exact built app. From b51a3006c01e11b7d94c8a96d1c45923f173a08e Mon Sep 17 00:00:00 2001 From: Anthony Ronning <101225832+AnthonyRonning@users.noreply.github.com> Date: Tue, 27 Oct 2026 04:54:27 +0000 Subject: [PATCH 2/3] Link Local API settings to proxy documentation --- .../app/src/ui/settings/local_api.rs | 27 ++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/apps/maple-agent/app/src/ui/settings/local_api.rs b/apps/maple-agent/app/src/ui/settings/local_api.rs index 3fd43b795..030a247f1 100644 --- a/apps/maple-agent/app/src/ui/settings/local_api.rs +++ b/apps/maple-agent/app/src/ui/settings/local_api.rs @@ -7,6 +7,8 @@ use super::{Section, SettingsScreen, SettingsTarget, section_title, setting_copy use crate::local_api::{LocalApiConfig, LocalApiState, LocalApiStatus}; use crate::ui::{text_input::TextInput, theme, widgets}; +const PROXY_DOCS_URL: &str = "https://maple.ai/docs/proxy"; + #[derive(Clone, Debug, Eq, PartialEq)] pub(super) enum LocalApiTarget { Enabled, @@ -15,6 +17,7 @@ pub(super) enum LocalApiTarget { SavePort, CopyUrl, ApiKeys, + LearnMore, Refresh, } @@ -257,6 +260,7 @@ impl SettingsScreen { targets.push(target(LocalApiTarget::CopyUrl)); } targets.push(target(LocalApiTarget::ApiKeys)); + targets.push(target(LocalApiTarget::LearnMore)); if !self.local_api.refreshing && !self.local_api.saving { targets.push(target(LocalApiTarget::Refresh)); } @@ -280,6 +284,7 @@ impl SettingsScreen { LocalApiTarget::SavePort => self.save_local_api_port(cx), LocalApiTarget::CopyUrl => self.copy_local_api_url(cx), LocalApiTarget::ApiKeys => self.select_section(Section::ApiKeys, cx), + LocalApiTarget::LearnMore => crate::ui::rich_text::open_link(PROXY_DOCS_URL), LocalApiTarget::Refresh => self.refresh_local_api_status(cx), } } @@ -470,6 +475,25 @@ impl SettingsScreen { } })) .child("Manage API keys"), + )) + .child(self.application_target( + || SettingsTarget::LocalApi(LocalApiTarget::LearnMore), + div() + .id("local-api-learn-more") + .role(gpui::Role::Link) + .aria_label("Learn more about the local API") + .text_color(gpui::rgb(theme::link())) + .focus_visible(widgets::focus_ring) + .hover(|style| style.underline().cursor_pointer()) + .tab_index(23) + .on_click(|_, _, _| crate::ui::rich_text::open_link(PROXY_DOCS_URL)) + .on_key_down(cx.listener(|_, event: &gpui::KeyDownEvent, _, cx| { + if activates(event) { + cx.stop_propagation(); + crate::ui::rich_text::open_link(PROXY_DOCS_URL); + } + })) + .child("Learn more"), )), ).child( widgets::card_row().flex().flex_col().gap_2() @@ -485,7 +509,7 @@ impl SettingsScreen { pane.child(self.application_target( || SettingsTarget::LocalApi(LocalApiTarget::Refresh), widgets::ghost_button("local-api-refresh") - .tab_index(23) + .tab_index(24) .on_click(cx.listener(|this, _, _, cx| this.refresh_local_api_status(cx))) .on_key_down(cx.listener(|this, event: &gpui::KeyDownEvent, _, cx| { if activates(event) { @@ -660,6 +684,7 @@ mod tests { vec![ target(LocalApiTarget::CopyUrl), target(LocalApiTarget::ApiKeys), + target(LocalApiTarget::LearnMore), ] ); }); From 4b9332b0c5f035665a46f984e58c2655c6faf344 Mon Sep 17 00:00:00 2001 From: Anthony Ronning <101225832+AnthonyRonning@users.noreply.github.com> Date: Tue, 27 Oct 2026 05:34:33 +0000 Subject: [PATCH 3/3] Move Local API documentation link into the introduction --- .../app/src/ui/settings/local_api.rs | 55 ++++++++++--------- 1 file changed, 29 insertions(+), 26 deletions(-) diff --git a/apps/maple-agent/app/src/ui/settings/local_api.rs b/apps/maple-agent/app/src/ui/settings/local_api.rs index 030a247f1..7c668012f 100644 --- a/apps/maple-agent/app/src/ui/settings/local_api.rs +++ b/apps/maple-agent/app/src/ui/settings/local_api.rs @@ -243,7 +243,7 @@ impl SettingsScreen { pub(super) fn local_api_targets(&self) -> Vec { let target = SettingsTarget::LocalApi; - let mut targets = Vec::new(); + let mut targets = vec![target(LocalApiTarget::LearnMore)]; if self.local_api.can_configure() { targets.push(target(LocalApiTarget::Enabled)); } @@ -260,7 +260,6 @@ impl SettingsScreen { targets.push(target(LocalApiTarget::CopyUrl)); } targets.push(target(LocalApiTarget::ApiKeys)); - targets.push(target(LocalApiTarget::LearnMore)); if !self.local_api.refreshing && !self.local_api.saving { targets.push(target(LocalApiTarget::Refresh)); } @@ -311,10 +310,33 @@ impl SettingsScreen { .flex_col() .gap_4() .child(section_title("Local API")) - .child(setting_copy( - "Connect AI applications on this computer", - "Expose Maple's inference APIs at a local address. Requests use Maple's secure backend and your API credits.", - )) + .child( + div().flex().flex_col().gap_2() + .child(setting_copy( + "Connect AI applications on this computer", + "Expose Maple's inference APIs at a local address. Requests use Maple's secure backend and your API credits.", + )) + .child(self.application_target( + || SettingsTarget::LocalApi(LocalApiTarget::LearnMore), + div() + .id("local-api-learn-more") + .role(gpui::Role::Link) + .aria_label("Learn more about the local API") + .text_sm() + .text_color(gpui::rgb(theme::link())) + .focus_visible(widgets::focus_ring) + .hover(|style| style.underline().cursor_pointer()) + .tab_index(17) + .on_click(|_, _, _| crate::ui::rich_text::open_link(PROXY_DOCS_URL)) + .on_key_down(cx.listener(|_, event: &gpui::KeyDownEvent, _, cx| { + if activates(event) { + cx.stop_propagation(); + crate::ui::rich_text::open_link(PROXY_DOCS_URL); + } + })) + .child("Learn more"), + )), + ) .child( widgets::card_row() .flex() @@ -475,25 +497,6 @@ impl SettingsScreen { } })) .child("Manage API keys"), - )) - .child(self.application_target( - || SettingsTarget::LocalApi(LocalApiTarget::LearnMore), - div() - .id("local-api-learn-more") - .role(gpui::Role::Link) - .aria_label("Learn more about the local API") - .text_color(gpui::rgb(theme::link())) - .focus_visible(widgets::focus_ring) - .hover(|style| style.underline().cursor_pointer()) - .tab_index(23) - .on_click(|_, _, _| crate::ui::rich_text::open_link(PROXY_DOCS_URL)) - .on_key_down(cx.listener(|_, event: &gpui::KeyDownEvent, _, cx| { - if activates(event) { - cx.stop_propagation(); - crate::ui::rich_text::open_link(PROXY_DOCS_URL); - } - })) - .child("Learn more"), )), ).child( widgets::card_row().flex().flex_col().gap_2() @@ -682,9 +685,9 @@ mod tests { assert_eq!( this.local_api_targets(), vec![ + target(LocalApiTarget::LearnMore), target(LocalApiTarget::CopyUrl), target(LocalApiTarget::ApiKeys), - target(LocalApiTarget::LearnMore), ] ); });