| 1 | //! The ACP client side, and the agent it serves in-process. |
| 2 | //! |
| 3 | //! [`AgentState::connection`] starts both halves on first use: |
| 4 | //! |
| 5 | //! - `ed_acp::serve(profile)` on one end of a [`Channel::duplex`], exactly what the product's |
| 6 | //! agent runs over stdio; |
| 7 | //! - an ACP client on the other end, which forwards `session/update` notifications and |
| 8 | //! `session/request_permission` requests to the webview, serves `fs/read_text_file` and |
| 9 | //! `fs/write_text_file` inside the session's roots, and opens the page of a URL-mode |
| 10 | //! elicitation in the system browser. |
| 11 | |
| 12 | use std::collections::HashMap; |
| 13 | use std::path::{Path, PathBuf}; |
| 14 | use std::sync::atomic::{AtomicBool, Ordering}; |
| 15 | use std::sync::{Arc, Mutex}; |
| 16 | |
| 17 | use agent_client_protocol::schema::v1::{ |
| 18 | AuthMethodId, AuthenticateRequest, CancelNotification, ClientCapabilities, |
| 19 | CompleteElicitationNotification, ContentBlock, CreateElicitationRequest, |
| 20 | CreateElicitationResponse, ElicitationAcceptAction, ElicitationAction, ElicitationCapabilities, |
| 21 | ElicitationMode, ElicitationUrlCapabilities, ErrorCode, FileSystemCapabilities, Implementation, |
| 22 | InitializeRequest, ListSessionsRequest, LoadSessionRequest, LogoutRequest, NewSessionRequest, |
| 23 | PromptRequest, ReadTextFileRequest, ReadTextFileResponse, RequestPermissionOutcome, |
| 24 | RequestPermissionRequest, RequestPermissionResponse, SelectedPermissionOutcome, |
| 25 | SessionConfigOption, SessionConfigOptionValue, SessionId, SessionNotification, |
| 26 | SetSessionConfigOptionRequest, TextContent, WriteTextFileRequest, WriteTextFileResponse, |
| 27 | }; |
| 28 | use agent_client_protocol::schema::ProtocolVersion; |
| 29 | use agent_client_protocol::{Agent, Channel, Client, ConnectionTo}; |
| 30 | use ed_acp::llm::LlmClient; |
| 31 | use ed_acp::{cli, LlmConfig, ServeOptions}; |
| 32 | use log::{error, info, warn}; |
| 33 | use serde_json::json; |
| 34 | use tauri::{AppHandle, Emitter}; |
| 35 | use tauri_plugin_opener::OpenerExt; |
| 36 | use tokio::sync::oneshot; |
| 37 | |
| 38 | use crate::{ |
| 39 | AgentSessionOpened, AgentSessionSummary, AgentStatus, AgentTurnEndedPayload, HostConfig, Roots, |
| 40 | AUTH_REQUIRED, |
| 41 | }; |
| 42 | |
| 43 | type Pending = Arc<Mutex<HashMap<String, oneshot::Sender<RequestPermissionOutcome>>>>; |
| 44 | /// The roots each open session may use, by session id. |
| 45 | type SessionRoots = Arc<Mutex<HashMap<String, Roots>>>; |
| 46 | |
| 47 | /// Managed Tauri state: the live connection and the permission requests waiting on the user. |
| 48 | pub struct AgentState { |
| 49 | config: HostConfig, |
| 50 | connection: tokio::sync::Mutex<Option<Live>>, |
| 51 | permissions: Pending, |
| 52 | roots: SessionRoots, |
| 53 | } |
| 54 | |
| 55 | struct Live { |
| 56 | conn: ConnectionTo<Agent>, |
| 57 | /// The ids of the auth methods the agent advertised in `initialize`. |
| 58 | auth_methods: Vec<String>, |
| 59 | /// Dropped (or fired) to hang up, which stops the agent too. |
| 60 | _hang_up: oneshot::Sender<()>, |
| 61 | /// Set when the client side has ended, so the next call reconnects. |
| 62 | ended: Arc<AtomicBool>, |
| 63 | } |
| 64 | |
| 65 | /// The `cwd` to ask `session/list` for, or `None` for every stored session. |
| 66 | /// |
| 67 | /// On a desktop the agent's data directory can be shared with the command-line agent, so the |
| 68 | /// app asks only for the conversations in the folder it opened. On iOS that directory is |
| 69 | /// inside the app's container, which nothing else writes to and whose path changes with every |
| 70 | /// app update. Sessions stored under the old path would never match, and the user's earlier |
| 71 | /// conversations would drop out of the list. Loading one rebinds it to the current path. |
| 72 | fn list_filter(cwd: &Path, private_data_dir: bool) -> Option<PathBuf> { |
| 73 | (!private_data_dir).then(|| cwd.to_path_buf()) |
| 74 | } |
| 75 | |
| 76 | fn list_request(roots: &Roots) -> ListSessionsRequest { |
| 77 | ListSessionsRequest::new().cwd(list_filter(&roots.cwd, cfg!(target_os = "ios"))) |
| 78 | } |
| 79 | |
| 80 | /// What the agent will use, without starting it: the profile's own session when it has an |
| 81 | /// auth provider, else the environment and the stored `env` file. |
| 82 | pub fn status(host: &HostConfig) -> AgentStatus { |
| 83 | let profile = (host.profile)(); |
| 84 | let config = match profile.auth() { |
| 85 | Some(auth) => auth.llm_config(), |
| 86 | None => Some(LlmConfig::from_env(&profile.info().env)), |
| 87 | }; |
| 88 | match config { |
| 89 | Some(config) => AgentStatus { |
| 90 | available: true, |
| 91 | has_key: config.api_key.is_some(), |
| 92 | model: config.model, |
| 93 | }, |
| 94 | None => AgentStatus { |
| 95 | available: true, |
| 96 | has_key: false, |
| 97 | model: profile.info().env.default_model.to_string(), |
| 98 | }, |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | /// What `<agent> --setup` does for Onde Inference, minus the terminal: check the key with Onde |
| 103 | /// Cloud, then write it to the agent's `env` file (mode 0600). With a `model` it is written as |
| 104 | /// `<PREFIX>_MODEL` beside the key and used for the check, so the agent doesn't fall back to a |
| 105 | /// default tier the key isn't allowed to call. In the App Store sandbox that file lives in |
| 106 | /// the app's container, so a key stored by the command-line agent isn't shared. |
| 107 | pub async fn set_api_key(host: &HostConfig, key: &str, model: Option<&str>) -> Result<(), String> { |
| 108 | let info = host.info(); |
| 109 | let key = key.trim(); |
| 110 | if !cli::valid_onde_key(key) { |
| 111 | return Err("An Onde key looks like \"app-id:app-secret\".".to_string()); |
| 112 | } |
| 113 | let provider_var = info.env.var("PROVIDER"); |
| 114 | let mut vars = vec![ |
| 115 | (provider_var, "onde".to_string()), |
| 116 | ("ONDE_API_KEY".to_string(), key.to_string()), |
| 117 | ]; |
| 118 | if let Some(model) = model { |
| 119 | vars.push((info.env.var("MODEL"), model.to_string())); |
| 120 | } |
| 121 | let config = LlmConfig::from_lookup(&info.env, |name| { |
| 122 | vars.iter() |
| 123 | .find(|(k, _)| k == name) |
| 124 | .map(|(_, v)| v.clone()) |
| 125 | .or_else(|| std::env::var(name).ok()) |
| 126 | }); |
| 127 | LlmClient::new(config) |
| 128 | .check_auth() |
| 129 | .await |
| 130 | .map_err(|e| format!("Onde Cloud rejected the key: {e:#}"))?; |
| 131 | |
| 132 | let dir = info |
| 133 | .env |
| 134 | .config_dir() |
| 135 | .ok_or("No config directory available.")?; |
| 136 | std::fs::create_dir_all(&dir).map_err(|e| format!("{}: {e}", dir.display()))?; |
| 137 | let path = dir.join("env"); |
| 138 | let existing = std::fs::read_to_string(&path).unwrap_or_default(); |
| 139 | std::fs::write(&path, cli::with_vars(&existing, &vars)) |
| 140 | .map_err(|e| format!("{}: {e}", path.display()))?; |
| 141 | #[cfg(unix)] |
| 142 | { |
| 143 | use std::os::unix::fs::PermissionsExt; |
| 144 | std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)) |
| 145 | .map_err(|e| format!("{}: {e}", path.display()))?; |
| 146 | } |
| 147 | info!("Stored the agent's Onde key in {}", path.display()); |
| 148 | Ok(()) |
| 149 | } |
| 150 | |
| 151 | /// Remove the agent's Onde key, and the model that came with it, from its env file. Nothing to |
| 152 | /// do when there is no file or no key in it. |
| 153 | pub fn clear_api_key(host: &HostConfig) -> Result<(), String> { |
| 154 | let info = host.info(); |
| 155 | let Some(dir) = info.env.config_dir() else { |
| 156 | return Ok(()); |
| 157 | }; |
| 158 | let path = dir.join("env"); |
| 159 | let existing = match std::fs::read_to_string(&path) { |
| 160 | Ok(existing) => existing, |
| 161 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()), |
| 162 | Err(e) => return Err(format!("{}: {e}", path.display())), |
| 163 | }; |
| 164 | let (kept, removed) = without_api_key(&existing, &info.env.var("MODEL")); |
| 165 | if removed { |
| 166 | std::fs::write(&path, kept).map_err(|e| format!("{}: {e}", path.display()))?; |
| 167 | info!("Removed the agent's Onde key from {}", path.display()); |
| 168 | } |
| 169 | Ok(()) |
| 170 | } |
| 171 | |
| 172 | /// `env` file content without any `ONDE_API_KEY` or `model_var` line, and whether a key was |
| 173 | /// there. |
| 174 | fn without_api_key(existing: &str, model_var: &str) -> (String, bool) { |
| 175 | let model_prefix = format!("{model_var}="); |
| 176 | let mut removed = false; |
| 177 | let kept: Vec<&str> = existing |
| 178 | .lines() |
| 179 | .filter(|line| { |
| 180 | let line = line.trim(); |
| 181 | let line = line.strip_prefix("export ").unwrap_or(line); |
| 182 | let is_key = line.starts_with("ONDE_API_KEY="); |
| 183 | removed |= is_key; |
| 184 | !is_key && !line.starts_with(&model_prefix) |
| 185 | }) |
| 186 | .collect(); |
| 187 | let mut out = kept.join("\n"); |
| 188 | if !out.is_empty() { |
| 189 | out.push('\n'); |
| 190 | } |
| 191 | (out, removed) |
| 192 | } |
| 193 | |
| 194 | impl AgentState { |
| 195 | pub fn new(config: HostConfig) -> Self { |
| 196 | Self { |
| 197 | config, |
| 198 | connection: tokio::sync::Mutex::new(None), |
| 199 | permissions: Pending::default(), |
| 200 | roots: SessionRoots::default(), |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | /// The live connection, starting the agent and the client on first use or after the |
| 205 | /// previous connection ended. |
| 206 | async fn connection( |
| 207 | &self, |
| 208 | app: &AppHandle, |
| 209 | ) -> Result<(ConnectionTo<Agent>, Vec<String>), String> { |
| 210 | let mut slot = self.connection.lock().await; |
| 211 | if let Some(live) = slot.as_ref() { |
| 212 | if !live.ended.load(Ordering::SeqCst) { |
| 213 | return Ok((live.conn.clone(), live.auth_methods.clone())); |
| 214 | } |
| 215 | } |
| 216 | let live = self.connect(app).await?; |
| 217 | let out = (live.conn.clone(), live.auth_methods.clone()); |
| 218 | *slot = Some(live); |
| 219 | Ok(out) |
| 220 | } |
| 221 | |
| 222 | async fn connect(&self, app: &AppHandle) -> Result<Live, String> { |
| 223 | let (agent_end, client_end) = Channel::duplex(); |
| 224 | |
| 225 | let opts = ServeOptions { |
| 226 | yolo: false, |
| 227 | surface: self.config.surface, |
| 228 | }; |
| 229 | let profile = (self.config.profile)(); |
| 230 | let agent_name = profile.info().display_name; |
| 231 | tauri::async_runtime::spawn(async move { |
| 232 | if let Err(e) = ed_acp::serve(profile, opts, agent_end).await { |
| 233 | error!("{agent_name} agent stopped: {e}"); |
| 234 | } |
| 235 | }); |
| 236 | |
| 237 | let (ready_tx, ready_rx) = |
| 238 | oneshot::channel::<Result<(ConnectionTo<Agent>, Vec<String>), String>>(); |
| 239 | let (hang_up, hung_up) = oneshot::channel::<()>(); |
| 240 | let ended = Arc::new(AtomicBool::new(false)); |
| 241 | |
| 242 | let client = Client |
| 243 | .builder() |
| 244 | .name(self.config.client_name) |
| 245 | .on_receive_notification( |
| 246 | { |
| 247 | let app = app.clone(); |
| 248 | let event = self.config.events.session_update; |
| 249 | async move |n: SessionNotification, _cx| { |
| 250 | if let Err(e) = app.emit(event, &n) { |
| 251 | error!("Failed to emit agent session update: {e:?}"); |
| 252 | } |
| 253 | Ok(()) |
| 254 | } |
| 255 | }, |
| 256 | agent_client_protocol::on_receive_notification!(), |
| 257 | ) |
| 258 | .on_receive_notification( |
| 259 | async move |n: CompleteElicitationNotification, _cx| { |
| 260 | info!("Sign-in page completed ({})", n.elicitation_id); |
| 261 | Ok(()) |
| 262 | }, |
| 263 | agent_client_protocol::on_receive_notification!(), |
| 264 | ) |
| 265 | .on_receive_request( |
| 266 | { |
| 267 | let app = app.clone(); |
| 268 | async move |req: CreateElicitationRequest, responder, _cx| { |
| 269 | responder.respond(CreateElicitationResponse::new(open_url(&app, &req))) |
| 270 | } |
| 271 | }, |
| 272 | agent_client_protocol::on_receive_request!(), |
| 273 | ) |
| 274 | .on_receive_request( |
| 275 | { |
| 276 | let app = app.clone(); |
| 277 | let pending = self.permissions.clone(); |
| 278 | let event = self.config.events.permission_requested; |
| 279 | async move |req: RequestPermissionRequest, responder, cx| { |
| 280 | // Wait for the user off the dispatch loop so updates keep flowing. |
| 281 | let request_id = uuid::Uuid::new_v4().to_string(); |
| 282 | let (answer_tx, answer_rx) = oneshot::channel(); |
| 283 | pending |
| 284 | .lock() |
| 285 | .unwrap() |
| 286 | .insert(request_id.clone(), answer_tx); |
| 287 | let payload = json!({ "requestId": request_id, "request": req }); |
| 288 | if let Err(e) = app.emit(event, &payload) { |
| 289 | error!("Failed to emit agent permission request: {e:?}"); |
| 290 | pending.lock().unwrap().remove(&request_id); |
| 291 | } |
| 292 | cx.spawn(async move { |
| 293 | let outcome = answer_rx |
| 294 | .await |
| 295 | .unwrap_or(RequestPermissionOutcome::Cancelled); |
| 296 | responder.respond(RequestPermissionResponse::new(outcome)) |
| 297 | }) |
| 298 | } |
| 299 | }, |
| 300 | agent_client_protocol::on_receive_request!(), |
| 301 | ) |
| 302 | .on_receive_request( |
| 303 | { |
| 304 | let roots = self.roots.clone(); |
| 305 | async move |req: ReadTextFileRequest, responder, _cx| { |
| 306 | let roots = roots.clone(); |
| 307 | let result = |
| 308 | tokio::task::spawn_blocking(move || read_text_file(&roots, &req)).await; |
| 309 | match result { |
| 310 | Ok(Ok(r)) => responder.respond(r), |
| 311 | Ok(Err(e)) => responder.respond_with_error(e), |
| 312 | Err(e) => responder.respond_with_internal_error(e.to_string()), |
| 313 | } |
| 314 | } |
| 315 | }, |
| 316 | agent_client_protocol::on_receive_request!(), |
| 317 | ) |
| 318 | .on_receive_request( |
| 319 | { |
| 320 | let roots = self.roots.clone(); |
| 321 | async move |req: WriteTextFileRequest, responder, _cx| { |
| 322 | let roots = roots.clone(); |
| 323 | let result = |
| 324 | tokio::task::spawn_blocking(move || write_text_file(&roots, &req)) |
| 325 | .await; |
| 326 | match result { |
| 327 | Ok(Ok(r)) => responder.respond(r), |
| 328 | Ok(Err(e)) => responder.respond_with_error(e), |
| 329 | Err(e) => responder.respond_with_internal_error(e.to_string()), |
| 330 | } |
| 331 | } |
| 332 | }, |
| 333 | agent_client_protocol::on_receive_request!(), |
| 334 | ); |
| 335 | |
| 336 | let ended_flag = ended.clone(); |
| 337 | let pending = self.permissions.clone(); |
| 338 | let client_name = self.config.client_name; |
| 339 | let client_version = self.config.client_version; |
| 340 | tauri::async_runtime::spawn(async move { |
| 341 | let result = client |
| 342 | .connect_with(client_end, |conn: ConnectionTo<Agent>| async move { |
| 343 | let init = conn |
| 344 | .send_request( |
| 345 | InitializeRequest::new(ProtocolVersion::V1) |
| 346 | .client_capabilities( |
| 347 | ClientCapabilities::new() |
| 348 | .fs(FileSystemCapabilities::new() |
| 349 | .read_text_file(true) |
| 350 | .write_text_file(true)) |
| 351 | .elicitation( |
| 352 | ElicitationCapabilities::new() |
| 353 | .url(ElicitationUrlCapabilities::new()), |
| 354 | ), |
| 355 | ) |
| 356 | .client_info(Implementation::new(client_name, client_version)), |
| 357 | ) |
| 358 | .block_task() |
| 359 | .await; |
| 360 | match init { |
| 361 | Ok(init) => { |
| 362 | let methods = init |
| 363 | .auth_methods |
| 364 | .iter() |
| 365 | .map(|m| m.id().to_string()) |
| 366 | .collect(); |
| 367 | let _ = ready_tx.send(Ok((conn, methods))); |
| 368 | // Stay connected until the state is dropped or replaced. |
| 369 | let _ = hung_up.await; |
| 370 | Ok(()) |
| 371 | } |
| 372 | Err(e) => { |
| 373 | let _ = ready_tx.send(Err(format!("Agent failed to start: {e}"))); |
| 374 | Err(e) |
| 375 | } |
| 376 | } |
| 377 | }) |
| 378 | .await; |
| 379 | if let Err(e) = result { |
| 380 | warn!("Agent connection ended: {e}"); |
| 381 | } |
| 382 | ended_flag.store(true, Ordering::SeqCst); |
| 383 | // Nobody can answer these any more. |
| 384 | pending.lock().unwrap().clear(); |
| 385 | }); |
| 386 | |
| 387 | let (conn, auth_methods) = ready_rx |
| 388 | .await |
| 389 | .map_err(|_| "Agent failed to start.".to_string())??; |
| 390 | info!("Agent connected in-process"); |
| 391 | Ok(Live { |
| 392 | conn, |
| 393 | auth_methods, |
| 394 | _hang_up: hang_up, |
| 395 | ended, |
| 396 | }) |
| 397 | } |
| 398 | |
| 399 | fn roots_for(&self, app: &AppHandle, cwd: Option<&Path>) -> Result<Roots, String> { |
| 400 | (self.config.roots)(app, cwd) |
| 401 | } |
| 402 | |
| 403 | /// The agent's conversations in `cwd` (the host's default roots when `None`), newest |
| 404 | /// first. |
| 405 | pub async fn list_sessions( |
| 406 | &self, |
| 407 | app: &AppHandle, |
| 408 | cwd: Option<PathBuf>, |
| 409 | ) -> Result<Vec<AgentSessionSummary>, String> { |
| 410 | let roots = self.roots_for(app, cwd.as_deref())?; |
| 411 | let (conn, _) = self.connection(app).await?; |
| 412 | let listed = conn |
| 413 | .send_request(list_request(&roots)) |
| 414 | .block_task() |
| 415 | .await |
| 416 | .map_err(describe)?; |
| 417 | Ok(listed |
| 418 | .sessions |
| 419 | .into_iter() |
| 420 | .map(|session| AgentSessionSummary { |
| 421 | session_id: session.session_id.to_string(), |
| 422 | title: session.title, |
| 423 | updated_at: session.updated_at, |
| 424 | }) |
| 425 | .collect()) |
| 426 | } |
| 427 | |
| 428 | /// Open a conversation in `cwd`: the one named by `session_id`, else (unless `fresh`) the |
| 429 | /// latest in that folder, else a new one. |
| 430 | pub async fn open_session( |
| 431 | &self, |
| 432 | app: &AppHandle, |
| 433 | cwd: Option<PathBuf>, |
| 434 | fresh: bool, |
| 435 | access_token: Option<String>, |
| 436 | session_id: Option<String>, |
| 437 | ) -> Result<AgentSessionOpened, String> { |
| 438 | let roots = self.roots_for(app, cwd.as_deref())?; |
| 439 | let (conn, _) = self.connection(app).await?; |
| 440 | let mcp_servers = (self.config.mcp_servers)(app, access_token); |
| 441 | // The one the user picked, else the latest. |
| 442 | let picked = session_id.is_some(); |
| 443 | let target = match session_id { |
| 444 | Some(id) => Some(SessionId::new(id)), |
| 445 | None if !fresh => conn |
| 446 | .send_request(list_request(&roots)) |
| 447 | .block_task() |
| 448 | .await |
| 449 | .map_err(describe)? |
| 450 | .sessions |
| 451 | .into_iter() |
| 452 | // Newest first. |
| 453 | .next() |
| 454 | .map(|latest| latest.session_id), |
| 455 | None => None, |
| 456 | }; |
| 457 | if let Some(id) = target { |
| 458 | // The agent may read files while it replays, so the roots come first. |
| 459 | self.roots |
| 460 | .lock() |
| 461 | .unwrap() |
| 462 | .insert(id.to_string(), roots.clone()); |
| 463 | let load = LoadSessionRequest::new(id.clone(), roots.cwd.clone()) |
| 464 | .additional_directories(roots.extra.clone()) |
| 465 | .mcp_servers(mcp_servers.clone()); |
| 466 | match conn.send_request(load).block_task().await { |
| 467 | Ok(loaded) => { |
| 468 | return Ok(AgentSessionOpened { |
| 469 | session_id: id.to_string(), |
| 470 | restored: true, |
| 471 | config_options: option_values(loaded.config_options), |
| 472 | }); |
| 473 | } |
| 474 | Err(e) if e.code == ErrorCode::AuthRequired => { |
| 475 | self.roots.lock().unwrap().remove(&id.to_string()); |
| 476 | return Err(AUTH_REQUIRED.to_string()); |
| 477 | } |
| 478 | Err(e) => { |
| 479 | self.roots.lock().unwrap().remove(&id.to_string()); |
| 480 | // One the user picked and can't have is an error to show. |
| 481 | if picked { |
| 482 | return Err(describe(e)); |
| 483 | } |
| 484 | // The latest no longer loading shouldn't block a new one. |
| 485 | warn!("Could not load agent session {id}: {e}"); |
| 486 | } |
| 487 | } |
| 488 | } |
| 489 | let new = NewSessionRequest::new(roots.cwd.clone()) |
| 490 | .additional_directories(roots.extra.clone()) |
| 491 | .mcp_servers(mcp_servers); |
| 492 | let session = conn |
| 493 | .send_request(new) |
| 494 | .block_task() |
| 495 | .await |
| 496 | .map_err(describe)?; |
| 497 | self.roots |
| 498 | .lock() |
| 499 | .unwrap() |
| 500 | .insert(session.session_id.to_string(), roots); |
| 501 | Ok(AgentSessionOpened { |
| 502 | session_id: session.session_id.to_string(), |
| 503 | restored: false, |
| 504 | config_options: option_values(session.config_options), |
| 505 | }) |
| 506 | } |
| 507 | |
| 508 | pub async fn prompt( |
| 509 | &self, |
| 510 | app: &AppHandle, |
| 511 | session_id: String, |
| 512 | text: String, |
| 513 | ) -> Result<(), String> { |
| 514 | let (conn, _) = self.connection(app).await?; |
| 515 | let request = PromptRequest::new( |
| 516 | SessionId::new(session_id.clone()), |
| 517 | vec![ContentBlock::Text(TextContent::new(text))], |
| 518 | ); |
| 519 | let app = app.clone(); |
| 520 | let event = self.config.events.turn_ended; |
| 521 | // Returned before the turn ends: a long turn would otherwise hold the invoke |
| 522 | // callback open for minutes. |
| 523 | tauri::async_runtime::spawn(async move { |
| 524 | let result = conn.send_request(request).block_task().await; |
| 525 | let payload = match result { |
| 526 | Ok(r) => AgentTurnEndedPayload { |
| 527 | session_id, |
| 528 | stop_reason: serde_json::to_value(r.stop_reason) |
| 529 | .ok() |
| 530 | .and_then(|v| v.as_str().map(String::from)), |
| 531 | error: None, |
| 532 | }, |
| 533 | Err(e) => AgentTurnEndedPayload { |
| 534 | session_id, |
| 535 | stop_reason: None, |
| 536 | error: Some(describe(e)), |
| 537 | }, |
| 538 | }; |
| 539 | if let Err(e) = app.emit(event, &payload) { |
| 540 | error!("Failed to emit agent turn end: {e:?}"); |
| 541 | } |
| 542 | }); |
| 543 | Ok(()) |
| 544 | } |
| 545 | |
| 546 | pub async fn cancel(&self, session_id: String) -> Result<(), String> { |
| 547 | // ACP: a client cancelling a turn answers its open permission requests with |
| 548 | // `cancelled`. |
| 549 | for (_, answer) in self.permissions.lock().unwrap().drain() { |
| 550 | let _ = answer.send(RequestPermissionOutcome::Cancelled); |
| 551 | } |
| 552 | let slot = self.connection.lock().await; |
| 553 | let Some(live) = slot.as_ref() else { |
| 554 | return Ok(()); |
| 555 | }; |
| 556 | live.conn |
| 557 | .send_notification(CancelNotification::new(SessionId::new(session_id))) |
| 558 | .map_err(|e| e.to_string()) |
| 559 | } |
| 560 | |
| 561 | pub fn respond_permission( |
| 562 | &self, |
| 563 | request_id: &str, |
| 564 | option_id: Option<String>, |
| 565 | ) -> Result<(), String> { |
| 566 | let answer = self |
| 567 | .permissions |
| 568 | .lock() |
| 569 | .unwrap() |
| 570 | .remove(request_id) |
| 571 | .ok_or("That request is no longer waiting for an answer.")?; |
| 572 | let outcome = match option_id { |
| 573 | Some(id) => RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(id)), |
| 574 | None => RequestPermissionOutcome::Cancelled, |
| 575 | }; |
| 576 | let _ = answer.send(outcome); |
| 577 | Ok(()) |
| 578 | } |
| 579 | |
| 580 | /// Change a config option (the model, auto-approve). `value` is a select value id as a |
| 581 | /// string, or a boolean. Returns the session's options as they are afterwards. |
| 582 | pub async fn set_config_option( |
| 583 | &self, |
| 584 | app: &AppHandle, |
| 585 | session_id: String, |
| 586 | config_id: String, |
| 587 | value: serde_json::Value, |
| 588 | ) -> Result<Vec<serde_json::Value>, String> { |
| 589 | let value = match value { |
| 590 | serde_json::Value::Bool(value) => SessionConfigOptionValue::from(value), |
| 591 | serde_json::Value::String(value) => SessionConfigOptionValue::from(value.as_str()), |
| 592 | other => return Err(format!("Not a config option value: {other}")), |
| 593 | }; |
| 594 | let (conn, _) = self.connection(app).await?; |
| 595 | let response = conn |
| 596 | .send_request(SetSessionConfigOptionRequest::new( |
| 597 | SessionId::new(session_id), |
| 598 | config_id, |
| 599 | value, |
| 600 | )) |
| 601 | .block_task() |
| 602 | .await |
| 603 | .map_err(describe)?; |
| 604 | Ok(option_values(Some(response.config_options))) |
| 605 | } |
| 606 | |
| 607 | /// Sign in with one of the agent's auth methods: `method_id`, else the first it |
| 608 | /// advertised. For a product with its own sign-in this opens the browser through the |
| 609 | /// client's URL elicitation and returns once the page is finished. |
| 610 | pub async fn authenticate( |
| 611 | &self, |
| 612 | app: &AppHandle, |
| 613 | method_id: Option<String>, |
| 614 | ) -> Result<(), String> { |
| 615 | let (conn, methods) = self.connection(app).await?; |
| 616 | let method = method_id |
| 617 | .or_else(|| methods.first().cloned()) |
| 618 | .ok_or("The agent offers no way to sign in.")?; |
| 619 | conn.send_request(AuthenticateRequest::new(AuthMethodId::new(method))) |
| 620 | .block_task() |
| 621 | .await |
| 622 | .map_err(|e| e.to_string())?; |
| 623 | Ok(()) |
| 624 | } |
| 625 | |
| 626 | /// Sign out: the agent forgets its stored session. |
| 627 | pub async fn logout(&self, app: &AppHandle) -> Result<(), String> { |
| 628 | let (conn, _) = self.connection(app).await?; |
| 629 | conn.send_request(LogoutRequest::new()) |
| 630 | .block_task() |
| 631 | .await |
| 632 | .map_err(|e| e.to_string())?; |
| 633 | Ok(()) |
| 634 | } |
| 635 | } |
| 636 | |
| 637 | /// Answer a URL-mode elicitation by opening its page in the system browser. Only `http` and |
| 638 | /// `https` are opened; the user "declines" anything else, or a browser that won't launch. |
| 639 | fn open_url(app: &AppHandle, req: &CreateElicitationRequest) -> ElicitationAction { |
| 640 | let ElicitationMode::Url(mode) = &req.mode else { |
| 641 | // Form mode was never advertised. |
| 642 | return ElicitationAction::Decline; |
| 643 | }; |
| 644 | if !(mode.url.starts_with("https://") || mode.url.starts_with("http://")) { |
| 645 | warn!("Refusing to open a non-web sign-in URL"); |
| 646 | return ElicitationAction::Decline; |
| 647 | } |
| 648 | match app.opener().open_url(&mode.url, None::<&str>) { |
| 649 | Ok(()) => ElicitationAction::Accept(ElicitationAcceptAction::new()), |
| 650 | Err(e) => { |
| 651 | warn!("Could not open the sign-in page: {e}"); |
| 652 | ElicitationAction::Decline |
| 653 | } |
| 654 | } |
| 655 | } |
| 656 | |
| 657 | fn option_values(options: Option<Vec<SessionConfigOption>>) -> Vec<serde_json::Value> { |
| 658 | options |
| 659 | .unwrap_or_default() |
| 660 | .into_iter() |
| 661 | .filter_map(|option| serde_json::to_value(option).ok()) |
| 662 | .collect() |
| 663 | } |
| 664 | |
| 665 | /// [`AUTH_REQUIRED`] for a missing key or session, so the page can ask for sign-in; the |
| 666 | /// agent's message otherwise. |
| 667 | fn describe(e: agent_client_protocol::Error) -> String { |
| 668 | if e.code == ErrorCode::AuthRequired { |
| 669 | AUTH_REQUIRED.to_string() |
| 670 | } else { |
| 671 | e.to_string() |
| 672 | } |
| 673 | } |
| 674 | |
| 675 | fn outside(path: &Path) -> agent_client_protocol::Error { |
| 676 | agent_client_protocol::Error::invalid_params().data(format!( |
| 677 | "{} is outside the session's workspace", |
| 678 | path.display() |
| 679 | )) |
| 680 | } |
| 681 | |
| 682 | fn roots_of( |
| 683 | roots: &SessionRoots, |
| 684 | session: &SessionId, |
| 685 | path: &Path, |
| 686 | ) -> agent_client_protocol::Result<Roots> { |
| 687 | roots |
| 688 | .lock() |
| 689 | .unwrap() |
| 690 | .get(&session.to_string()) |
| 691 | .cloned() |
| 692 | .filter(|roots| roots.contains(path)) |
| 693 | .ok_or_else(|| outside(path)) |
| 694 | } |
| 695 | |
| 696 | /// `fs/read_text_file`, honouring the optional 1-based `line` and `limit`. |
| 697 | fn read_text_file( |
| 698 | roots: &SessionRoots, |
| 699 | req: &ReadTextFileRequest, |
| 700 | ) -> agent_client_protocol::Result<ReadTextFileResponse> { |
| 701 | roots_of(roots, &req.session_id, &req.path)?; |
| 702 | let content = std::fs::read_to_string(&req.path).map_err(|e| { |
| 703 | if e.kind() == std::io::ErrorKind::NotFound { |
| 704 | agent_client_protocol::Error::resource_not_found(Some(req.path.display().to_string())) |
| 705 | } else { |
| 706 | agent_client_protocol::Error::internal_error().data(e.to_string()) |
| 707 | } |
| 708 | })?; |
| 709 | let skip = req.line.map_or(0, |l| l.saturating_sub(1) as usize); |
| 710 | let take = req.limit.map_or(usize::MAX, |l| l as usize); |
| 711 | let content = content |
| 712 | .lines() |
| 713 | .skip(skip) |
| 714 | .take(take) |
| 715 | .collect::<Vec<_>>() |
| 716 | .join("\n"); |
| 717 | Ok(ReadTextFileResponse::new(content)) |
| 718 | } |
| 719 | |
| 720 | /// `fs/write_text_file`: the client creates the file and its parent directories when they |
| 721 | /// don't exist. |
| 722 | fn write_text_file( |
| 723 | roots: &SessionRoots, |
| 724 | req: &WriteTextFileRequest, |
| 725 | ) -> agent_client_protocol::Result<WriteTextFileResponse> { |
| 726 | roots_of(roots, &req.session_id, &req.path)?; |
| 727 | let write = || { |
| 728 | if let Some(parent) = req.path.parent() { |
| 729 | std::fs::create_dir_all(parent)?; |
| 730 | } |
| 731 | std::fs::write(&req.path, &req.content) |
| 732 | }; |
| 733 | write().map_err(|e| agent_client_protocol::Error::internal_error().data(e.to_string()))?; |
| 734 | Ok(WriteTextFileResponse::new()) |
| 735 | } |
| 736 | |
| 737 | #[cfg(test)] |
| 738 | mod tests { |
| 739 | use super::*; |
| 740 | |
| 741 | #[test] |
| 742 | fn sessions_are_listed_by_directory_unless_the_data_dir_is_private() { |
| 743 | let cwd = Path::new("/data/agent"); |
| 744 | assert_eq!(list_filter(cwd, false), Some(cwd.to_path_buf())); |
| 745 | // iOS: the container path changes with every update, so no filter. |
| 746 | assert_eq!(list_filter(cwd, true), None); |
| 747 | } |
| 748 | |
| 749 | #[test] |
| 750 | fn removing_the_key_keeps_every_other_line() { |
| 751 | let (kept, removed) = without_api_key( |
| 752 | "AGENT_PROVIDER=onde\nONDE_API_KEY=app:secret\nAGENT_MODEL=x\nAGENT_SURFACE=app\n", |
| 753 | "AGENT_MODEL", |
| 754 | ); |
| 755 | assert!(removed); |
| 756 | assert_eq!(kept, "AGENT_PROVIDER=onde\nAGENT_SURFACE=app\n"); |
| 757 | } |
| 758 | |
| 759 | #[test] |
| 760 | fn an_exported_key_goes_too_and_a_file_without_one_is_untouched() { |
| 761 | assert_eq!( |
| 762 | without_api_key("export ONDE_API_KEY=a:b\n", "AGENT_MODEL"), |
| 763 | (String::new(), true) |
| 764 | ); |
| 765 | assert_eq!( |
| 766 | without_api_key("AGENT_PROVIDER=onde\n", "AGENT_MODEL"), |
| 767 | ("AGENT_PROVIDER=onde\n".to_string(), false) |
| 768 | ); |
| 769 | assert_eq!(without_api_key("", "AGENT_MODEL"), (String::new(), false)); |
| 770 | } |
| 771 | |
| 772 | #[test] |
| 773 | fn a_similar_variable_is_not_the_key() { |
| 774 | let (kept, removed) = |
| 775 | without_api_key("ONDE_API_KEY_BACKUP=x\nAGENT_MODEL_X=y\n", "AGENT_MODEL"); |
| 776 | assert!(!removed); |
| 777 | assert_eq!(kept, "ONDE_API_KEY_BACKUP=x\nAGENT_MODEL_X=y\n"); |
| 778 | } |
| 779 | |
| 780 | #[test] |
| 781 | fn a_session_reaches_only_its_own_roots() { |
| 782 | let roots = SessionRoots::default(); |
| 783 | roots.lock().unwrap().insert( |
| 784 | "a".into(), |
| 785 | Roots { |
| 786 | cwd: PathBuf::from("/work/a"), |
| 787 | extra: Vec::new(), |
| 788 | }, |
| 789 | ); |
| 790 | let a = SessionId::new("a"); |
| 791 | let b = SessionId::new("b"); |
| 792 | assert!(roots_of(&roots, &a, Path::new("/work/a/src/main.rs")).is_ok()); |
| 793 | assert!(roots_of(&roots, &a, Path::new("/work/b/src/main.rs")).is_err()); |
| 794 | // An id nobody opened has no roots at all. |
| 795 | assert!(roots_of(&roots, &b, Path::new("/work/a/src/main.rs")).is_err()); |
| 796 | } |
| 797 | } |