@ondeinference / ed / commits / 1084b41

ed-acp: the agent and the Profile API; add the crate to the workspace

Last part of the ed-acp crate: sessions, the turn loop and the Profile a product implements. The crate builds from here on. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Co-Authored-By: siGit Code v1.6.3 <noreply@sigit.si>

Seto Elkahfi committed Oct 9, 2026 at 00:12 UTC 1084b4151ca064bee771f11790c2c55842c68736
6 files changed +1559
Cargo.lock
+30
@@ -1770,6 +1770,15 @@ dependencies = [
1770 "crypto-common",
1771 ]
1772
1773 +[[package]]
1774 +name = "directories"
1775 +version = "6.0.0"
1776 +source = "registry+https://github.com/rust-lang/crates.io-index"
1777 +checksum = "16f5094c54661b38d03bd7e50df373292118db60b585c08a411c6d840017fe7d"
1778 +dependencies = [
1779 + "dirs-sys",
1780 +]
1781 +
1782 [[package]]
1783 name = "dirs"
1784 version = "6.0.0"
@@ -1934,6 +1943,27 @@ version = "0.1.3"
1943 source = "registry+https://github.com/rust-lang/crates.io-index"
1944 checksum = "e1d926b4d407d372f141f93bb444696142c29d32962ccbd3531117cf3aa0bfa9"
1945
1946 +[[package]]
1947 +name = "ed-acp"
1948 +version = "1.0.1"
1949 +dependencies = [
1950 + "agent-client-protocol",
1951 + "anyhow",
1952 + "async-trait",
1953 + "base64 0.23.1",
1954 + "directories",
1955 + "ed-mcp",
1956 + "futures",
1957 + "reqwest 0.13.5",
1958 + "serde",
1959 + "serde_json",
1960 + "tokio",
1961 + "tokio-util",
1962 + "tracing",
1963 + "tracing-subscriber",
1964 + "uuid 1.27.0",
1965 +]
1966 +
1967 [[package]]
1968 name = "ed-agent"
1969 version = "1.0.1"
Cargo.toml
+2
@@ -5,6 +5,7 @@ members = [
5 "crates/ed-agent-ffi",
6 "crates/ed-agent-tauri",
7 "crates/ed-mcp",
8 + "crates/ed-acp",
9 "uniffi-bindgen",
10 ]
11
@@ -56,6 +57,7 @@ base64 = "0.23"
57 # Internal
58 ed-agent = { version = "1.0.1", path = "crates/ed-agent" }
59 ed-mcp = { version = "1.0.1", path = "crates/ed-mcp" }
60 +ed-acp = { version = "1.0.1", path = "crates/ed-acp" }
61
62 # gemm-common (via candle, via onde) depends on sysctl for every
63 # `target_vendor = "apple"`, but sysctl 0.6.0 only has a `sys` module behind
crates/ed-acp/Cargo.toml new
+32
@@ -0,0 +1,32 @@
1 +[package]
2 +name = "ed-acp"
3 +description = "Agent Client Protocol (ACP) v1 server for Onde agents: sessions, auth, the turn loop on Onde Cloud, tool approval, workspace tools and MCP. A product supplies a Profile."
4 +keywords = ["acp", "agent", "llm", "onde", "mcp"]
5 +categories = ["asynchronous", "development-tools"]
6 +version.workspace = true
7 +edition = "2024"
8 +rust-version.workspace = true
9 +license.workspace = true
10 +authors.workspace = true
11 +repository.workspace = true
12 +homepage.workspace = true
13 +
14 +[dependencies]
15 +ed-mcp = { workspace = true, features = ["acp"] }
16 +agent-client-protocol.workspace = true
17 +async-trait.workspace = true
18 +tokio = { workspace = true, features = ["full"] }
19 +tokio-util.workspace = true
20 +futures.workspace = true
21 +reqwest.workspace = true
22 +serde.workspace = true
23 +serde_json.workspace = true
24 +anyhow.workspace = true
25 +tracing.workspace = true
26 +tracing-subscriber.workspace = true
27 +uuid.workspace = true
28 +directories.workspace = true
29 +base64.workspace = true
30 +
31 +[dev-dependencies]
32 +tokio = { version = "1", features = ["full"] }
crates/ed-acp/README.md new
+17
@@ -0,0 +1,17 @@
1 +# ed-acp
2 +
3 +An [Agent Client Protocol](https://agentclientprotocol.com) v1 server for Onde agents.
4 +
5 +It is the part of an ACP agent that is the same for every product: `initialize` and version
6 +negotiation, auth, durable sessions, the model picker, prompt content, the turn loop against Onde
7 +Cloud with streaming, tool approval, workspace tools routed through the client, and MCP servers
8 +(stdio and streamable HTTP) through [`ed-mcp`](https://crates.io/crates/ed-mcp). A product
9 +supplies a `Profile`: its name, prompt, tools, slash commands and built-in MCP servers.
10 +
11 +Test an agent built on it with [`ed-acp-testkit`](https://crates.io/crates/ed-acp-testkit), and
12 +drive it from a terminal with [`ed-acp-tui`](https://crates.io/crates/ed-acp-tui).
13 +
14 +## Copyright
15 +
16 +© 2026 [Splitfire AB](https://5mb.app) ([Onde Inference](https://ondeinference.com)).
17 +Licensed under MIT or Apache-2.0.
crates/ed-acp/src/agent.rs new
+1342
@@ -0,0 +1,1342 @@
1 +//! [`AcpServer`]: the ACP v1 agent behind every Onde product. Owns sessions (in memory, backed
2 +//! by [`SessionStore`]), the turn loop, auth, and the model and auto-approve config options.
3 +//! Everything product-specific comes from the [`Profile`].
4 +
5 +use std::collections::{HashMap, HashSet};
6 +use std::path::{Path, PathBuf};
7 +use std::sync::atomic::{AtomicBool, Ordering};
8 +use std::sync::{Arc, Mutex};
9 +use std::time::{SystemTime, UNIX_EPOCH};
10 +
11 +use agent_client_protocol::schema::ProtocolVersion;
12 +use agent_client_protocol::schema::v1::{
13 + AuthMethod, AuthMethodTerminal, AuthenticateRequest, AuthenticateResponse,
14 + AvailableCommandsUpdate, ClientCapabilities, CloseSessionRequest, CloseSessionResponse,
15 + ConfigOptionUpdate, ContentBlock, ContentChunk, DeleteSessionRequest, DeleteSessionResponse,
16 + ListSessionsRequest, ListSessionsResponse, LoadSessionRequest, LoadSessionResponse,
17 + LogoutResponse, McpServer, MessageId, NewSessionRequest, NewSessionResponse, PromptRequest,
18 + PromptResponse, ResumeSessionRequest, ResumeSessionResponse, SessionConfigOption,
19 + SessionConfigOptionCategory, SessionConfigSelectOption, SessionConfigSelectOptions, SessionId,
20 + SessionInfo, SessionInfoUpdate, SessionNotification, SessionUpdate,
21 + SetSessionConfigOptionRequest, SetSessionConfigOptionResponse, StopReason, ToolCall,
22 + ToolCallContent, ToolCallLocation, ToolCallStatus, ToolCallUpdateFields, UsageUpdate,
23 +};
24 +use agent_client_protocol::{Client, ConnectionTo, Responder};
25 +use serde_json::{Value, json};
26 +use tokio_util::sync::CancellationToken;
27 +
28 +use ed_mcp::{ClientInfo, McpToolset, ToolsetConfig};
29 +
30 +use crate::content as prompt;
31 +use crate::llm::{Delta, LlmClient, LlmConfig, LlmEnv};
32 +use crate::store::{SessionStore, StoredSession};
33 +use crate::tools::{self, ToolCtx, ToolOutcome, Toolset, absolutize, normalize};
34 +use crate::{Profile, PromptCtx, SessionCtx};
35 +
36 +const MAX_TURNS: usize = 50;
37 +/// Id of the session config option that selects the model.
38 +pub const MODEL_CONFIG_ID: &str = "model";
39 +/// Id of the boolean session config option that skips permission prompts.
40 +pub const AUTO_APPROVE_CONFIG_ID: &str = "auto_approve";
41 +/// Context window reported in `usage_update` when `<PREFIX>_CONTEXT_WINDOW` is unset.
42 +const DEFAULT_CONTEXT_WINDOW: u64 = 128_000;
43 +/// Id of the terminal auth method advertised on initialize.
44 +pub const AUTH_METHOD_ID: &str = "terminal-setup";
45 +/// Cap on the sessions returned by `session/list`.
46 +const MAX_LISTED: usize = 200;
47 +
48 +/// The terminal auth method: clients run `<agent> --setup`.
49 +pub fn auth_methods() -> Vec<AuthMethod> {
50 + vec![AuthMethod::Terminal(
51 + AuthMethodTerminal::new(AUTH_METHOD_ID, "Run in terminal")
52 + .description("Interactive setup: store your Onde Inference API key")
53 + .args(vec!["--setup".into()]),
54 + )]
55 +}
56 +
57 +/// Whether the client can run terminal auth. ACP v1 has the `auth.terminal` capability; clients
58 +/// that predate it (including the ACP registry's validator, which probes every listed agent)
59 +/// say so with `_meta["terminal-auth"]: true` instead. Offering no method to such a client
60 +/// fails registry validation, so either signal counts.
61 +pub fn supports_terminal_auth(caps: &ClientCapabilities) -> bool {
62 + caps.auth.terminal
63 + || caps
64 + .meta
65 + .as_ref()
66 + .and_then(|m| m.get("terminal-auth"))
67 + .and_then(Value::as_bool)
68 + .unwrap_or(false)
69 +}
70 +
71 +pub fn auth_required_error(agent: &str) -> agent_client_protocol::Error {
72 + agent_client_protocol::Error::auth_required().data(format!(
73 + "No Onde API key configured. Authenticate with the terminal method (`{agent} --setup`)."
74 + ))
75 +}
76 +
77 +/// The only protocol version this agent speaks is v1. A client proposing a newer version gets
78 +/// v1 back (it can then decide whether to continue); an older (v0) client is told v1 too, as
79 +/// the spec requires the latest supported version when the requested one isn't supported.
80 +pub fn negotiate_version(client: ProtocolVersion) -> ProtocolVersion {
81 + let _ = client;
82 + ProtocolVersion::V1
83 +}
84 +
85 +struct Session {
86 + cwd: PathBuf,
87 + /// Additional workspace roots (from the request's `additional_directories`).
88 + roots: Vec<PathBuf>,
89 + model: String,
90 + /// Conversation history without the system prompt (rebuilt every turn).
91 + messages: Vec<Value>,
92 + cancel: CancellationToken,
93 + always_allowed: Arc<Mutex<HashSet<String>>>,
94 + always_rejected: Arc<Mutex<HashSet<String>>>,
95 + /// Approve tool calls without asking; the `auto_approve` config option. Not persisted:
96 + /// a reopened session starts from the process default (`--yolo`).
97 + auto_approve: Arc<AtomicBool>,
98 + title: Option<String>,
99 + /// Last activity, seconds since the Unix epoch.
100 + updated_at: u64,
101 + mcp: Arc<McpToolset>,
102 + /// A turn is running; its history is checked out of `messages`.
103 + busy: bool,
104 +}
105 +
106 +impl Session {
107 + fn fresh(
108 + cwd: PathBuf,
109 + roots: Vec<PathBuf>,
110 + model: String,
111 + yolo: bool,
112 + mcp: Arc<McpToolset>,
113 + ) -> Self {
114 + Self::restored(
115 + StoredSession {
116 + cwd,
117 + roots,
118 + model,
119 + title: None,
120 + updated_at: now_secs(),
121 + messages: Vec::new(),
122 + },
123 + yolo,
124 + mcp,
125 + )
126 + }
127 +
128 + /// A session brought back from the store. Permission grants start fresh; MCP servers are
129 + /// connected by the request that reopens it.
130 + fn restored(stored: StoredSession, yolo: bool, mcp: Arc<McpToolset>) -> Self {
131 + Self {
132 + cwd: stored.cwd,
133 + roots: stored.roots,
134 + model: stored.model,
135 + messages: stored.messages,
136 + cancel: CancellationToken::new(),
137 + always_allowed: Arc::default(),
138 + always_rejected: Arc::default(),
139 + auto_approve: Arc::new(AtomicBool::new(yolo)),
140 + title: stored.title,
141 + updated_at: stored.updated_at,
142 + mcp,
143 + busy: false,
144 + }
145 + }
146 +
147 + fn stored(&self) -> StoredSession {
148 + StoredSession {
149 + cwd: self.cwd.clone(),
150 + roots: self.roots.clone(),
151 + model: self.model.clone(),
152 + title: self.title.clone(),
153 + updated_at: self.updated_at,
154 + messages: self.messages.clone(),
155 + }
156 + }
157 +}
158 +
159 +fn new_message_id() -> MessageId {
160 + MessageId::new(uuid::Uuid::new_v4().to_string())
161 +}
162 +
163 +/// ACP v1 requires absolute `cwd` and `additionalDirectories`.
164 +fn check_absolute(cwd: &Path, roots: &[PathBuf]) -> agent_client_protocol::Result<()> {
165 + if !cwd.is_absolute() {
166 + return Err(invalid("cwd must be an absolute path"));
167 + }
168 + if roots.iter().any(|r| !r.is_absolute()) {
169 + return Err(invalid(
170 + "additionalDirectories entries must be absolute paths",
171 + ));
172 + }
173 + Ok(())
174 +}
175 +
176 +pub fn now_secs() -> u64 {
177 + SystemTime::now()
178 + .duration_since(UNIX_EPOCH)
179 + .map(|d| d.as_secs())
180 + .unwrap_or(0)
181 +}
182 +
183 +/// ISO 8601 UTC timestamp without sub-second precision (no chrono dependency).
184 +pub fn iso8601(secs: u64) -> String {
185 + let days = secs / 86_400;
186 + let secs_of_day = secs % 86_400;
187 + let (h, m, s) = (
188 + secs_of_day / 3600,
189 + secs_of_day % 3600 / 60,
190 + secs_of_day % 60,
191 + );
192 + // Civil-from-days algorithm (Howard Hinnant).
193 + let z = days as i64 + 719_468;
194 + let era = z.div_euclid(146_097);
195 + let doe = z.rem_euclid(146_097);
196 + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
197 + let y = yoe + era * 400;
198 + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
199 + let mp = (5 * doy + 2) / 153;
200 + let d = doy - (153 * mp + 2) / 5 + 1;
201 + let mo = if mp < 10 { mp + 3 } else { mp - 9 };
202 + let y = if mo <= 2 { y + 1 } else { y };
203 + format!("{y:04}-{mo:02}-{d:02}T{h:02}:{m:02}:{s:02}Z")
204 +}
205 +
206 +/// Derive a short session title from the first user prompt.
207 +pub fn session_title(prompt: &str) -> String {
208 + const MAX: usize = 80;
209 + let first = prompt.lines().next().unwrap_or("").trim();
210 + if first.chars().count() <= MAX {
211 + first.to_string()
212 + } else {
213 + let cut: String = first.chars().take(MAX - 1).collect();
214 + format!("{}…", cut.trim_end())
215 + }
216 +}
217 +
218 +fn invalid(msg: impl Into<String>) -> agent_client_protocol::Error {
219 + agent_client_protocol::Error::invalid_params().data(msg.into())
220 +}
221 +
222 +fn unknown_session(id: &SessionId) -> agent_client_protocol::Error {
223 + invalid(format!("unknown session {id}"))
224 +}
225 +
226 +/// A tool message; `failed` is kept in a private field that never reaches the model.
227 +fn tool_message(id: &str, content: &str, failed: bool) -> Value {
228 + let mut m = json!({ "role": "tool", "tool_call_id": id, "content": content });
229 + if failed {
230 + m["x_failed"] = json!(true);
231 + }
232 + m
233 +}
234 +
235 +fn wire_message(m: &Value) -> Value {
236 + match m.as_object() {
237 + Some(o) if o.contains_key("x_failed") => {
238 + let mut o = o.clone();
239 + o.remove("x_failed");
240 + Value::Object(o)
241 + }
242 + _ => m.clone(),
243 + }
244 +}
245 +
246 +fn locations_of(paths: &[PathBuf]) -> Vec<ToolCallLocation> {
247 + paths
248 + .iter()
249 + .map(|p| ToolCallLocation::new(absolutize(p)))
250 + .collect()
251 +}
252 +
253 +/// Text of a stored user message (`content` is a string, or an array of parts).
254 +fn user_text(msg: &Value) -> String {
255 + match msg.get("content") {
256 + Some(Value::String(s)) => s.clone(),
257 + Some(Value::Array(parts)) => parts
258 + .iter()
259 + .filter_map(|p| match p.get("type").and_then(Value::as_str) {
260 + Some("text") => p.get("text").and_then(Value::as_str).map(String::from),
261 + Some("image_url") => Some("[image]".into()),
262 + _ => None,
263 + })
264 + .collect::<Vec<_>>()
265 + .join("\n\n"),
266 + _ => String::new(),
267 + }
268 +}
269 +
270 +#[derive(Clone)]
271 +pub struct AcpServer {
272 + profile: Arc<dyn Profile>,
273 + /// The profile's toolsets, built once.
274 + toolsets: Arc<Vec<Arc<dyn Toolset>>>,
275 + env: LlmEnv,
276 + /// Replaced when credentials change: `authenticate`, `logout`, or a key stored by
277 + /// `--setup` in another process.
278 + llm: Arc<Mutex<LlmClient>>,
279 + /// Process default for each session's `auto_approve` (`--yolo` / `<PREFIX>_YOLO`).
280 + yolo: bool,
281 + /// How the agent was launched: `tui` (interactive terminal UI) or `acp` (editor).
282 + surface: &'static str,
283 + client_caps: Arc<Mutex<ClientCapabilities>>,
284 + sessions: Arc<Mutex<HashMap<SessionId, Session>>>,
285 + /// Models offered for selection, fetched from `GET /models` on first use and dropped
286 + /// when credentials change.
287 + models: Arc<tokio::sync::Mutex<Option<Vec<String>>>>,
288 + store: SessionStore,
289 + data_dir: PathBuf,
290 +}
291 +
292 +impl AcpServer {
293 + pub fn new(profile: Arc<dyn Profile>, yolo: bool, surface: &'static str) -> Self {
294 + let env = profile.info().env;
295 + let data_dir = env.data_dir();
296 + Self {
297 + toolsets: Arc::new(profile.toolsets()),
298 + profile,
299 + env,
300 + llm: Arc::new(Mutex::new(LlmClient::new(LlmConfig::from_env(&env)))),
301 + yolo,
302 + surface,
303 + client_caps: Arc::default(),
304 + sessions: Arc::default(),
305 + models: Arc::default(),
306 + store: SessionStore::new(&data_dir),
307 + data_dir,
308 + }
309 + }
310 +
311 + pub fn set_client_caps(&self, caps: ClientCapabilities) {
312 + *self.client_caps.lock().unwrap() = caps;
313 + }
314 +
315 + fn llm(&self) -> LlmClient {
316 + self.llm.lock().unwrap().clone()
317 + }
318 +
319 + pub fn has_api_key(&self) -> bool {
320 + self.llm().has_api_key()
321 + }
322 +
323 + pub fn model(&self) -> String {
324 + self.llm().model().to_string()
325 + }
326 +
327 + /// Re-read credentials (environment + stored `env` file) and report whether a key is
328 + /// configured. Editors run terminal auth (`<agent> --setup`) in a separate process
329 + /// and then retry `session/new` or `session/load` on this one without calling
330 + /// `authenticate`, so the stored key has to be picked up here. The model list is only
331 + /// dropped when the configuration actually changed.
332 + pub async fn refresh_credentials(&self) -> bool {
333 + let fresh = LlmConfig::from_env(&self.env);
334 + let changed = fresh != *self.llm().config();
335 + if changed {
336 + tracing::info!("credentials changed; reloading the Onde client");
337 + *self.llm.lock().unwrap() = LlmClient::new(fresh);
338 + *self.models.lock().await = None;
339 + }
340 + self.has_api_key()
341 + }
342 +
343 + /// `authenticate`: the client has run `<agent> --setup`; pick up the stored key.
344 + pub async fn authenticate(
345 + &self,
346 + req: AuthenticateRequest,
347 + ) -> agent_client_protocol::Result<AuthenticateResponse> {
348 + if &*req.method_id.0 != AUTH_METHOD_ID {
349 + return Err(invalid(format!("unknown auth method {}", req.method_id.0)));
350 + }
351 + if self.refresh_credentials().await {
352 + Ok(AuthenticateResponse::new())
353 + } else {
354 + Err(self.auth_required())
355 + }
356 + }
357 +
358 + /// `logout`: remove the stored key and forget it. A key exported in the agent's own
359 + /// environment can't be removed from here and keeps the agent signed in.
360 + pub async fn logout(&self) -> agent_client_protocol::Result<LogoutResponse> {
361 + let removed = match self.env.config_file() {
362 + Some(path) => match std::fs::remove_file(&path) {
363 + Err(e) if e.kind() != std::io::ErrorKind::NotFound => {
364 + Err(format!("removing {}: {e}", path.display()))
365 + }
366 + _ => Ok(()),
367 + },
368 + None => Ok(()),
369 + };
370 + self.refresh_credentials().await;
371 + removed.map_err(|e| agent_client_protocol::Error::internal_error().data(e))?;
372 + Ok(LogoutResponse::new())
373 + }
374 +
375 + fn scratch_dir(&self, id: &SessionId) -> Option<PathBuf> {
376 + let id = crate::store::sanitize_id(&id.0).ok()?;
377 + Some(self.data_dir.join("scratch").join(id))
378 + }
379 +
380 + fn remove_scratch(&self, id: &SessionId) {
381 + if let Some(dir) = self.scratch_dir(id) {
382 + let _ = std::fs::remove_dir_all(dir);
383 + }
384 + }
385 +
386 + /// `AUTH_REQUIRED`, naming this agent's setup command.
387 + pub fn auth_required(&self) -> agent_client_protocol::Error {
388 + auth_required_error(self.profile.info().name)
389 + }
390 +
391 + /// How a session's MCP servers are connected.
392 + fn mcp_config(&self) -> ToolsetConfig {
393 + let info = self.profile.info();
394 + ToolsetConfig::new(ClientInfo::new(info.name, info.version))
395 + }
396 +
397 + async fn connect_mcp(
398 + &self,
399 + servers: &[agent_client_protocol::schema::v1::McpServer],
400 + ) -> Arc<McpToolset> {
401 + Arc::new(
402 + McpToolset::connect(
403 + self.mcp_config(),
404 + servers,
405 + &self.profile.builtin_mcp_servers(),
406 + )
407 + .await,
408 + )
409 + }
410 +
411 + fn system_prompt(&self, cwd: &Path, roots: &[PathBuf], mcp: &McpToolset) -> String {
412 + self.profile.system_prompt(&PromptCtx {
413 + surface: self.surface,
414 + cwd,
415 + roots,
416 + mcp,
417 + })
418 + }
419 +
420 + /// The selectable models. Falls back to just the configured model if the endpoint
421 + /// can't list them; the configured model is always offered.
422 + async fn models(&self) -> Vec<String> {
423 + let mut cache = self.models.lock().await;
424 + if let Some(ids) = &*cache {
425 + return ids.clone();
426 + }
427 + let llm = self.llm();
428 + let default = llm.model().to_string();
429 + // An explicit list wins; editors show this list as the model picker.
430 + let configured: Vec<String> = std::env::var(self.env.var("MODELS"))
431 + .unwrap_or_default()
432 + .split(',')
433 + .map(|m| m.trim().to_string())
434 + .filter(|m| !m.is_empty())
435 + .collect();
436 + let mut ids = if !configured.is_empty() {
437 + configured
438 + } else {
439 + match llm.models().await {
440 + Ok(models) => models.into_iter().map(|m| m.id).collect(),
441 + Err(e) => {
442 + tracing::warn!("listing models failed, offering only {default}: {e:#}");
443 + Vec::new()
444 + }
445 + }
446 + };
447 + if !ids.contains(&default) {
448 + ids.insert(0, default);
449 + }
450 + *cache = Some(ids.clone());
451 + ids
452 + }
453 +
454 + /// Whether the client advertised `session.configOptions.boolean`, which ACP requires
455 + /// before an agent may offer `type: "boolean"` options.
456 + fn boolean_options_supported(&self) -> bool {
457 + self.client_caps
458 + .lock()
459 + .unwrap()
460 + .session
461 + .as_ref()
462 + .and_then(|s| s.config_options.as_ref())
463 + .is_some_and(|c| c.boolean.is_some())
464 + }
465 +
466 + async fn config_options(&self, model: &str, auto_approve: bool) -> Vec<SessionConfigOption> {
467 + let options = self
468 + .models()
469 + .await
470 + .into_iter()
471 + .map(|id| SessionConfigSelectOption::new(id.clone(), id))
472 + .collect();
473 + let mut config = vec![
474 + SessionConfigOption::select(
475 + MODEL_CONFIG_ID,
476 + "Model",
477 + model.to_string(),
478 + SessionConfigSelectOptions::Ungrouped(options),
479 + )
480 + .category(SessionConfigOptionCategory::Model),
481 + ];
482 + if self.boolean_options_supported() {
483 + config.push(
484 + SessionConfigOption::boolean(
485 + AUTO_APPROVE_CONFIG_ID,
486 + "Auto-approve actions",
487 + auto_approve,
488 + )
489 + .description("Run tools that change things without asking for permission"),
490 + );
491 + }
492 + config
493 + }
494 +
495 + /// The model and auto-approve setting of a live session.
496 + fn session_settings(&self, id: &SessionId) -> Option<(String, bool)> {
497 + self.sessions
498 + .lock()
499 + .unwrap()
500 + .get(id)
501 + .map(|s| (s.model.clone(), s.auto_approve.load(Ordering::Relaxed)))
502 + }
503 +
504 + async fn session_config_options(
505 + &self,
506 + id: &SessionId,
507 + ) -> agent_client_protocol::Result<Vec<SessionConfigOption>> {
508 + let (model, auto_approve) = self
509 + .session_settings(id)
510 + .ok_or_else(|| unknown_session(id))?;
511 + Ok(self.config_options(&model, auto_approve).await)
512 + }
513 +
514 + pub async fn set_config_option(
515 + &self,
516 + req: SetSessionConfigOptionRequest,
517 + connection: &ConnectionTo<Client>,
518 + ) -> agent_client_protocol::Result<SetSessionConfigOptionResponse> {
519 + let id = req.session_id.clone();
520 + self.restore(&id).await?;
521 + match &*req.config_id.0 {
522 + MODEL_CONFIG_ID => {
523 + let Some(model) = req.value.as_value_id().map(|v| v.0.to_string()) else {
524 + return Err(invalid("model must be a value id"));
525 + };
526 + if !self.models().await.contains(&model) {
527 + return Err(invalid(format!("unknown model {model}")));
528 + }
529 + match self.sessions.lock().unwrap().get_mut(&id) {
530 + Some(s) => s.model = model.clone(),
531 + None => return Err(unknown_session(&id)),
532 + }
533 + self.persist(&id).await;
534 + tracing::info!("session {id} now uses model {model}");
535 + }
536 + AUTO_APPROVE_CONFIG_ID if self.boolean_options_supported() => {
537 + let Some(on) = req.value.as_bool() else {
538 + return Err(invalid("auto_approve must be a boolean"));
539 + };
540 + match self.sessions.lock().unwrap().get(&id) {
541 + Some(s) => s.auto_approve.store(on, Ordering::Relaxed),
542 + None => return Err(unknown_session(&id)),
543 + }
544 + tracing::info!("session {id} auto-approve {on}");
545 + }
546 + other => return Err(invalid(format!("unknown config option {other}"))),
547 + }
548 + let options = self.session_config_options(&id).await?;
549 + // Tell every view of the session about the change, not just the requester.
550 + connection.send_notification(SessionNotification::new(
551 + id,
552 + SessionUpdate::ConfigOptionUpdate(ConfigOptionUpdate::new(options.clone())),
553 + ))?;
554 + Ok(SetSessionConfigOptionResponse::new(options))
555 + }
556 +
557 + /// Save a session that is not running a turn (a turn checks the history out; the
558 + /// end-of-turn save covers it).
559 + async fn persist(&self, id: &SessionId) {
560 + let stored = self
561 + .sessions
562 + .lock()
563 + .unwrap()
564 + .get(id)
565 + .filter(|s| !s.busy)
566 + .map(Session::stored);
567 + let Some(stored) = stored else { return };
568 + let store = self.store.clone();
569 + let sid = id.0.to_string();
570 + match tokio::task::spawn_blocking(move || store.save(&sid, &stored)).await {
571 + Ok(Ok(())) => {}
572 + Ok(Err(e)) => tracing::warn!("saving session {id} failed: {e:#}"),
573 + Err(e) => tracing::warn!("saving session {id} failed: {e}"),
574 + }
575 + }
576 +
577 + /// Bring a stored session into memory without rebinding it (a prompt or config change
578 + /// for a thread the client reopened without `session/load`). Errors if it is unknown.
579 + async fn restore(&self, id: &SessionId) -> agent_client_protocol::Result<()> {
580 + if self.sessions.lock().unwrap().contains_key(id) {
581 + return Ok(());
582 + }
583 + let stored = self
584 + .load_stored(id)
585 + .await?
586 + .ok_or_else(|| unknown_session(id))?;
587 + tracing::info!("restored session {id} from disk");
588 + self.sessions
589 + .lock()
590 + .unwrap()
591 + .entry(id.clone())
592 + .or_insert_with(|| Session::restored(stored, self.yolo, self.no_mcp()));
593 + Ok(())
594 + }
595 +
596 + async fn load_stored(
597 + &self,
598 + id: &SessionId,
599 + ) -> agent_client_protocol::Result<Option<StoredSession>> {
600 + let store = self.store.clone();
601 + let sid = id.0.to_string();
602 + tokio::task::spawn_blocking(move || store.load(&sid))
603 + .await
604 + .map_err(|e| agent_client_protocol::Error::internal_error().data(e.to_string()))?
605 + .map_err(|e| agent_client_protocol::Error::internal_error().data(format!("{e:#}")))
606 + }
607 +
608 + /// A session placeholder until its MCP servers are connected.
609 + fn no_mcp(&self) -> Arc<McpToolset> {
610 + Arc::new(McpToolset::empty(self.mcp_config()))
611 + }
612 +
613 + fn session_mcp(&self, id: &SessionId) -> Arc<McpToolset> {
614 + self.sessions
615 + .lock()
616 + .unwrap()
617 + .get(id)
618 + .map(|s| s.mcp.clone())
619 + .unwrap_or_else(|| self.no_mcp())
620 + }
621 +
622 + /// `AvailableCommandsUpdate` for a session, sent after the response that created it.
623 + pub fn send_commands(
624 + &self,
625 + connection: &ConnectionTo<Client>,
626 + id: &SessionId,
627 + ) -> agent_client_protocol::Result<()> {
628 + let mcp = self.session_mcp(id);
629 + let commands = self.profile.slash_commands(&SessionCtx { mcp: &mcp });
630 + connection.send_notification(SessionNotification::new(
631 + id.clone(),
632 + SessionUpdate::AvailableCommandsUpdate(AvailableCommandsUpdate::new(commands)),
633 + ))
634 + }
635 +
636 + pub async fn new_session(
637 + &self,
638 + req: NewSessionRequest,
639 + ) -> agent_client_protocol::Result<NewSessionResponse> {
640 + check_absolute(&req.cwd, &req.additional_directories)?;
641 + // Lexical normalization only: the client sees `cwd` again in `session/list`, so
642 + // symlinks in it are not rewritten.
643 + let cwd = normalize(&req.cwd);
644 + let roots: Vec<PathBuf> = req
645 + .additional_directories
646 + .iter()
647 + .map(|r| normalize(r))
648 + .collect();
649 + let id = SessionId::new(uuid::Uuid::new_v4().to_string());
650 + let model = self.model();
651 + let mcp = self.connect_mcp(&req.mcp_servers).await;
652 + let session = Session::fresh(cwd, roots, model.clone(), self.yolo, mcp);
653 + self.sessions.lock().unwrap().insert(id.clone(), session);
654 + // Saved right away so the thread survives a restart before its first prompt.
655 + self.persist(&id).await;
656 + let options = self.config_options(&model, self.yolo).await;
657 + Ok(NewSessionResponse::new(id).config_options(options))
658 + }
659 +
660 + /// Live and stored sessions, optionally filtered by working directory, newest first.
661 + pub fn list_sessions(
662 + &self,
663 + req: ListSessionsRequest,
664 + ) -> agent_client_protocol::Result<ListSessionsResponse> {
665 + // Every session fits on one page, so no cursor is ever handed out; any cursor is stale.
666 + if let Some(cursor) = &req.cursor {
667 + return Err(invalid(format!("invalid cursor {cursor}")));
668 + }
669 + struct Row {
670 + id: SessionId,
671 + cwd: PathBuf,
672 + roots: Vec<PathBuf>,
673 + title: Option<String>,
674 + updated_at: u64,
675 + }
676 + let mut rows: HashMap<String, Row> = self
677 + .store
678 + .list()
679 + .into_iter()
680 + .map(|(id, s)| {
681 + (
682 + id.clone(),
683 + Row {
684 + id: SessionId::new(id),
685 + cwd: s.cwd,
686 + roots: s.roots,
687 + title: s.title,
688 + updated_at: s.updated_at,
689 + },
690 + )
691 + })
692 + .collect();
693 + // Live state wins over the (possibly older) stored copy.
694 + for (id, s) in self.sessions.lock().unwrap().iter() {
695 + rows.insert(
696 + id.0.to_string(),
697 + Row {
698 + id: id.clone(),
699 + cwd: s.cwd.clone(),
700 + roots: s.roots.clone(),
701 + title: s.title.clone(),
702 + updated_at: s.updated_at,
703 + },
704 + );
705 + }
706 + // Stored cwds are normalized, so normalize the filter the same way.
707 + let wanted = req.cwd.as_deref().map(normalize);
708 + let mut rows: Vec<Row> = rows
709 + .into_values()
710 + .filter(|r| wanted.as_ref().is_none_or(|cwd| *cwd == normalize(&r.cwd)))
711 + .collect();
712 + rows.sort_by(|a, b| {
713 + b.updated_at
714 + .cmp(&a.updated_at)
715 + .then_with(|| a.id.0.cmp(&b.id.0))
716 + });
717 + rows.truncate(MAX_LISTED);
718 + Ok(ListSessionsResponse::new(
719 + rows.into_iter()
720 + .map(|r| {
721 + SessionInfo::new(r.id, r.cwd)
722 + .additional_directories(r.roots)
723 + .title(r.title)
724 + .updated_at(iso8601(r.updated_at))
725 + })
726 + .collect(),
727 + ))
728 + }
729 +
730 + pub fn cancel(&self, session_id: &SessionId) {
731 + if let Some(s) = self.sessions.lock().unwrap().get(session_id) {
732 + s.cancel.cancel();
733 + }
734 + }
735 +
736 + /// Make `id` live for `session/load` and `session/resume`: from memory, else from the
737 + /// store; rebind cwd and roots, and reconnect MCP servers from the request. A well-formed
738 + /// id with no saved history (a thread from a process that never saved it) reopens empty
739 + /// under the same id, so the editor's thread opens instead of failing to launch. Returns
740 + /// whether the history was lost.
741 + async fn reopen(
742 + &self,
743 + id: &SessionId,
744 + cwd: &Path,
745 + roots: &[PathBuf],
746 + mcp_servers: &[McpServer],
747 + ) -> agent_client_protocol::Result<bool> {
748 + check_absolute(cwd, roots)?;
749 + if crate::store::sanitize_id(&id.0).is_err() {
750 + return Err(unknown_session(id));
751 + }
752 + let live = self.sessions.lock().unwrap().contains_key(id);
753 + let stored = if live {
754 + None
755 + } else {
756 + self.load_stored(id).await?
757 + };
758 + let history_lost = !live && stored.is_none();
759 + let cwd = normalize(cwd);
760 + let roots: Vec<PathBuf> = roots.iter().map(|r| normalize(r)).collect();
761 + let mcp = self.connect_mcp(mcp_servers).await;
762 +
763 + let old_mcp = {
764 + let mut sessions = self.sessions.lock().unwrap();
765 + if !sessions.contains_key(id) {
766 + let session = match stored {
767 + Some(stored) => Session::restored(stored, self.yolo, self.no_mcp()),
768 + None => {
769 + tracing::warn!("session {id} has no saved history; reopening it empty");
770 + Session::fresh(
771 + cwd.clone(),
772 + roots.clone(),
773 + self.model(),
774 + self.yolo,
775 + self.no_mcp(),
776 + )
777 + }
778 + };
779 + sessions.insert(id.clone(), session);
780 + }
781 + let s = sessions.get_mut(id).ok_or_else(|| unknown_session(id))?;
782 + s.cwd = cwd;
783 + if !roots.is_empty() {
784 + s.roots = roots;
785 + }
786 + s.updated_at = now_secs();
787 + std::mem::replace(&mut s.mcp, mcp)
788 + };
789 + if let Some(old) = Arc::into_inner(old_mcp) {
790 + tokio::spawn(old.shutdown());
791 + }
792 + self.persist(id).await;
793 + Ok(history_lost)
794 + }
795 +
796 + /// Tell the user a reopened thread starts without its earlier history. Shown in the
797 + /// thread only; the model doesn't see it.
798 + fn notify_history_lost(
799 + &self,
800 + connection: &ConnectionTo<Client>,
801 + id: &SessionId,
802 + ) -> agent_client_protocol::Result<()> {
803 + connection.send_notification(SessionNotification::new(
804 + id.clone(),
805 + SessionUpdate::AgentMessageChunk(
806 + ContentChunk::new(
807 + format!(
808 + "{} no longer has this conversation's history, so it starts fresh \
809 + from here.",
810 + self.profile.info().display_name
811 + )
812 + .into(),
813 + )
814 + .message_id(new_message_id()),
815 + ),
816 + ))
817 + }
818 +
819 + /// Replay a session's conversation as `session/update` notifications (text, tool calls
820 + /// with their status and output), then respond with config options. `session/load`.
821 + pub async fn load_session(
822 + &self,
823 + req: LoadSessionRequest,
824 + connection: &ConnectionTo<Client>,
825 + ) -> agent_client_protocol::Result<LoadSessionResponse> {
826 + let id = req.session_id.clone();
827 + let history_lost = self
828 + .reopen(&id, &req.cwd, &req.additional_directories, &req.mcp_servers)
829 + .await?;
830 + let (messages, cwd, roots) = {
831 + let sessions = self.sessions.lock().unwrap();
832 + let s = sessions.get(&id).ok_or_else(|| unknown_session(&id))?;
833 + (s.messages.clone(), s.cwd.clone(), s.roots.clone())
834 + };
835 + let notify = |update: SessionUpdate| {
836 + connection.send_notification(SessionNotification::new(id.clone(), update))
837 + };
838 + // Tool output by call id, from the stored `tool` messages.
839 + let results: HashMap<&str, (&str, bool)> = messages
840 + .iter()
841 + .filter(|m| m.get("role").and_then(Value::as_str) == Some("tool"))
842 + .filter_map(|m| {
843 + Some((
844 + m.get("tool_call_id")?.as_str()?,
845 + (
846 + m.get("content").and_then(Value::as_str).unwrap_or(""),
847 + m.get("x_failed").and_then(Value::as_bool).unwrap_or(false),
848 + ),
849 + ))
850 + })
851 + .collect();
852 + for msg in &messages {
853 + match msg.get("role").and_then(Value::as_str).unwrap_or("") {
854 + "user" => {
855 + let text = user_text(msg);
856 + if !text.is_empty() {
857 + notify(SessionUpdate::UserMessageChunk(
858 + ContentChunk::new(text.into()).message_id(new_message_id()),
859 + ))?;
860 + }
861 + }
862 + "assistant" => {
863 + let text = msg.get("content").and_then(Value::as_str).unwrap_or("");
864 + if !text.is_empty() {
865 + notify(SessionUpdate::AgentMessageChunk(
866 + ContentChunk::new(text.to_string().into()).message_id(new_message_id()),
867 + ))?;
868 + }
869 + let calls = msg.get("tool_calls").and_then(Value::as_array);
870 + for tc in calls.into_iter().flatten() {
871 + let call_id = tc.get("id").and_then(Value::as_str).unwrap_or("unknown");
872 + let name = tc
873 + .pointer("/function/name")
874 + .and_then(Value::as_str)
875 + .unwrap_or("unknown");
876 + let args: Value = tc
877 + .pointer("/function/arguments")
878 + .and_then(Value::as_str)
879 + .and_then(|a| serde_json::from_str(a).ok())
880 + .unwrap_or_else(|| json!({}));
881 + let (title, kind, locations) =
882 + tools::describe(&self.toolsets, &cwd, &roots, name, &args);
883 + let mut call = ToolCall::new(call_id.to_string(), title)
884 + .name(name.to_string())
885 + .kind(kind)
886 + .locations(locations)
887 + .raw_input(args);
888 + match results.get(call_id) {
889 + Some((output, failed)) => {
890 + call = call
891 + .status(if *failed {
892 + ToolCallStatus::Failed
893 + } else {
894 + ToolCallStatus::Completed
895 + })
896 + .content(vec![ToolCallContent::from(output.to_string())])
897 + .raw_output(json!({ "output": output, "failed": failed }));
898 + }
899 + // The turn was interrupted before this call produced a result.
900 + None => call = call.status(ToolCallStatus::Failed),
901 + }
902 + notify(SessionUpdate::ToolCall(call))?;
903 + }
904 + }
905 + // system and tool messages are not replayed on their own.
906 + _ => {}
907 + }
908 + }
909 + if history_lost {
910 + self.notify_history_lost(connection, &id)?;
911 + }
912 + Ok(LoadSessionResponse::new().config_options(self.session_config_options(&id).await?))
913 + }
914 +
915 + /// Rebind a session to a new cwd / additional directories without replaying it.
916 + pub async fn resume_session(
917 + &self,
918 + req: ResumeSessionRequest,
919 + connection: &ConnectionTo<Client>,
920 + ) -> agent_client_protocol::Result<ResumeSessionResponse> {
921 + let id = req.session_id.clone();
922 + let history_lost = self
923 + .reopen(&id, &req.cwd, &req.additional_directories, &req.mcp_servers)
924 + .await?;
925 + if history_lost {
926 + self.notify_history_lost(connection, &id)?;
927 + }
928 + Ok(ResumeSessionResponse::new().config_options(self.session_config_options(&id).await?))
929 + }
930 +
931 + /// Cancel any in-progress work, release MCP servers and scratch audio. The stored
932 + /// conversation stays; `session/delete` removes it.
933 + pub fn close_session(
934 + &self,
935 + req: CloseSessionRequest,
936 + ) -> agent_client_protocol::Result<CloseSessionResponse> {
937 + let id = req.session_id.clone();
938 + let removed = self.sessions.lock().unwrap().remove(&id);
939 + match removed {
940 + Some(s) => self.release(&id, s),
941 + // A session from an earlier process that was never reopened has nothing to close.
942 + None if self.store.load(&id.0).ok().flatten().is_some() => {}
943 + None => return Err(unknown_session(&id)),
944 + }
945 + Ok(CloseSessionResponse::new())
946 + }
947 +
948 + /// Forget a session, in memory and on disk, with its scratch audio. Idempotent: deleting
949 + /// an unknown session succeeds.
950 + pub fn delete_session(
951 + &self,
952 + req: DeleteSessionRequest,
953 + ) -> agent_client_protocol::Result<DeleteSessionResponse> {
954 + let id = req.session_id.clone();
955 + if let Some(s) = self.sessions.lock().unwrap().remove(&id) {
956 + self.release(&id, s);
957 + }
958 + // A malformed id can't name a stored session; nothing to remove.
959 + if crate::store::sanitize_id(&id.0).is_ok() {
960 + self.store.delete(&id.0).map_err(|e| {
961 + agent_client_protocol::Error::internal_error().data(format!("{e:#}"))
962 + })?;
963 + }
964 + self.remove_scratch(&id);
965 + Ok(DeleteSessionResponse::new())
966 + }
967 +
968 + fn release(&self, id: &SessionId, s: Session) {
969 + s.cancel.cancel();
970 + self.remove_scratch(id);
971 + if let Some(mcp) = Arc::into_inner(s.mcp) {
972 + tokio::spawn(mcp.shutdown());
973 + }
974 + }
975 +
976 + /// Stop everything: used when the connection ends.
977 + pub async fn shutdown(&self) {
978 + let sessions: Vec<Session> = self
979 + .sessions
980 + .lock()
981 + .unwrap()
982 + .drain()
983 + .map(|(_, s)| s)
984 + .collect();
985 + for s in sessions {
986 + s.cancel.cancel();
987 + if let Some(mcp) = Arc::into_inner(s.mcp) {
988 + mcp.shutdown().await;
989 + }
990 + }
991 + for server in self.profile.builtin_mcp_servers() {
992 + server.shutdown().await;
993 + }
994 + }
995 +
996 + pub async fn prompt(
997 + &self,
998 + request: PromptRequest,
999 + responder: Responder<PromptResponse>,
1000 + connection: ConnectionTo<Client>,
1001 + ) -> agent_client_protocol::Result<()> {
1002 + // A client may prompt a reopened thread without loading or resuming it first.
1003 + if let Err(e) = self.restore(&request.session_id).await {
1004 + return responder.respond_with_error(e);
1005 + }
1006 + match self.run_turn(request, connection).await {
1007 + Ok(stop) => responder.respond(PromptResponse::new(stop)),
1008 + Err(e) if e.downcast_ref::<UnknownSession>().is_some() => {
1009 + responder.respond_with_error(invalid(e.to_string()))
1010 + }
1011 + Err(e) => responder.respond_with_error(
1012 + agent_client_protocol::Error::internal_error().data(format!("{e:#}")),
1013 + ),
1014 + }
1015 + }
1016 +
1017 + async fn run_turn(
1018 + &self,
1019 + request: PromptRequest,
1020 + connection: ConnectionTo<Client>,
1021 + ) -> anyhow::Result<StopReason> {
1022 + let session_id = request.session_id.clone();
1023 + let caps = self.client_caps.lock().unwrap().clone();
1024 + let blocks =
1025 + prompt::resolve_resource_links(&request.prompt, &caps, &connection, &session_id).await;
1026 + let prompt_text = prompt::prompt_to_text(&blocks, &self.profile.audio_hints());
1027 + // Check the session out for the turn.
1028 + let (
1029 + cwd,
1030 + roots,
1031 + model,
1032 + history,
1033 + cancel,
1034 + always_allowed,
1035 + always_rejected,
1036 + auto_approve,
1037 + mcp,
1038 + new_title,
1039 + ) = {
1040 + let mut sessions = self.sessions.lock().unwrap();
1041 + let s = sessions
1042 + .get_mut(&session_id)
1043 + .ok_or_else(|| UnknownSession(session_id.to_string()))?;
1044 + if s.busy {
1045 + anyhow::bail!("a prompt is already running for session {session_id}");
1046 + }
1047 + s.busy = true;
1048 + s.cancel = CancellationToken::new();
1049 + s.updated_at = now_secs();
1050 + let new_title = if s.title.is_none() && !prompt_text.trim().is_empty() {
1051 + s.title = Some(session_title(&prompt_text));
1052 + s.title.clone()
1053 + } else {
1054 + None
1055 + };
1056 + (
1057 + s.cwd.clone(),
1058 + s.roots.clone(),
1059 + s.model.clone(),
1060 + std::mem::take(&mut s.messages),
1061 + s.cancel.clone(),
1062 + s.always_allowed.clone(),
1063 + s.always_rejected.clone(),
1064 + s.auto_approve.clone(),
1065 + s.mcp.clone(),
1066 + new_title,
1067 + )
1068 + };
1069 + if let Some(title) = new_title {
1070 + let update = SessionInfoUpdate::new()
1071 + .title(title)
1072 + .updated_at(iso8601(now_secs()));
1073 + let _ = connection.send_notification(SessionNotification::new(
1074 + session_id.clone(),
1075 + SessionUpdate::SessionInfoUpdate(update),
1076 + ));
1077 + }
1078 +
1079 + let mut messages = vec![json!({
1080 + "role": "system",
1081 + "content": self.system_prompt(&cwd, &roots, &mcp),
1082 + })];
1083 + messages.extend(history);
1084 +
1085 + let ctx = ToolCtx {
1086 + connection: connection.clone(),
1087 + session_id: session_id.clone(),
1088 + cwd,
1089 + roots,
1090 + caps,
1091 + cancel,
1092 + auto_approve,
1093 + always_allowed,
1094 + always_rejected,
1095 + mcp,
1096 + };
1097 + let result = self.turn(&ctx, &model, &mut messages, &blocks).await;
1098 +
1099 + // Check the history back in, even when the turn failed.
1100 + let history = messages.split_off(1);
1101 + let stored = {
1102 + let mut sessions = self.sessions.lock().unwrap();
1103 + sessions.get_mut(&session_id).map(|s| {
1104 + s.messages = history;
1105 + s.busy = false;
1106 + s.updated_at = now_secs();
1107 + s.stored()
1108 + })
1109 + };
1110 + if let Some(stored) = stored {
1111 + let updated_at = stored.updated_at;
1112 + let store = self.store.clone();
1113 + let id = session_id.0.to_string();
1114 + let saved = tokio::task::spawn_blocking(move || store.save(&id, &stored)).await;
1115 + if let Ok(Err(e)) | Err(e) = saved.map_err(anyhow::Error::from) {
1116 + tracing::warn!("saving session {session_id} failed: {e:#}");
1117 + }
1118 + let _ = connection.send_notification(SessionNotification::new(
1119 + session_id.clone(),
1120 + SessionUpdate::SessionInfoUpdate(
1121 + SessionInfoUpdate::new().updated_at(iso8601(updated_at)),
1122 + ),
1123 + ));
1124 + }
1125 + result
1126 + }
1127 +
1128 + /// Everything inside a turn after the session is checked out.
1129 + async fn turn(
1130 + &self,
1131 + ctx: &ToolCtx,
1132 + model: &str,
1133 + messages: &mut Vec<Value>,
1134 + blocks: &[ContentBlock],
1135 + ) -> anyhow::Result<StopReason> {
1136 + let audio_paths = match self.scratch_dir(&ctx.session_id) {
1137 + Some(dir) => prompt::save_audio_blocks(blocks, &dir).await?,
1138 + None => Vec::new(),
1139 + };
1140 + let hints = self.profile.audio_hints();
1141 + let slash = self.profile.expand_slash(
1142 + &prompt::prompt_to_text(blocks, &hints),
1143 + &SessionCtx { mcp: &ctx.mcp },
1144 + );
1145 + let content = prompt::prompt_to_content(blocks, &audio_paths, slash.as_deref(), &hints);
1146 + messages.push(json!({ "role": "user", "content": content }));
1147 + self.agent_loop(ctx, model, messages).await
1148 + }
1149 +
1150 + async fn agent_loop(
1151 + &self,
1152 + ctx: &ToolCtx,
1153 + model: &str,
1154 + messages: &mut Vec<Value>,
1155 + ) -> anyhow::Result<StopReason> {
1156 + let llm = self.llm();
1157 + let tool_defs = tools::definitions(&self.toolsets, &ctx.mcp);
1158 + let notify = |update: SessionUpdate| {
1159 + ctx.connection
1160 + .send_notification(SessionNotification::new(ctx.session_id.clone(), update))
1161 + };
1162 + // `size` is the model's context window, which OpenAI-style endpoints don't report.
1163 + let context_window = std::env::var(self.env.var("CONTEXT_WINDOW"))
1164 + .ok()
1165 + .and_then(|v| v.parse().ok())
1166 + .unwrap_or(DEFAULT_CONTEXT_WINDOW);
1167 +
1168 + for _ in 0..MAX_TURNS {
1169 + // One message id per assistant message (and per thought), so clients can group
1170 + // the streamed chunks.
1171 + let message_id = new_message_id();
1172 + let thought_id = new_message_id();
1173 + let wire: Vec<Value> = messages.iter().map(wire_message).collect();
1174 + let completion = tokio::select! {
1175 + r = llm.complete(model, &wire, &tool_defs, |delta| {
1176 + let update = match delta {
1177 + Delta::Text(t) => SessionUpdate::AgentMessageChunk(
1178 + ContentChunk::new(t.to_string().into()).message_id(message_id.clone()),
1179 + ),
1180 + Delta::Reasoning(t) => SessionUpdate::AgentThoughtChunk(
1181 + ContentChunk::new(t.to_string().into()).message_id(thought_id.clone()),
1182 + ),
1183 + };
1184 + if let Err(e) = notify(update) {
1185 + tracing::warn!("failed to send update: {e}");
1186 + }
1187 + }) => r?,
1188 + () = ctx.cancel.cancelled() => return Ok(StopReason::Cancelled),
1189 + };
1190 + messages.push(completion.to_message());
1191 + if let Some(usage) = completion.usage {
1192 + let used = usage.used();
1193 + notify(SessionUpdate::UsageUpdate(UsageUpdate::new(
1194 + used,
1195 + context_window.max(used),
1196 + )))?;
1197 + }
1198 +
1199 + if completion.tool_calls.is_empty() {
1200 + return Ok(match completion.finish_reason.as_deref() {
1201 + Some("length") => StopReason::MaxTokens,
1202 + Some("content_filter") => StopReason::Refusal,
1203 + _ => StopReason::EndTurn,
1204 + });
1205 + }
1206 +
1207 + for tc in &completion.tool_calls {
1208 + // Every tool call needs a matching tool message, even after cancellation.
1209 + if ctx.cancel.is_cancelled() {
1210 + messages.push(tool_message(&tc.id, "Cancelled by user.", true));
1211 + continue;
1212 + }
1213 + let parsed: Result<Value, _> = serde_json::from_str(&tc.arguments);
1214 + let args = parsed.as_ref().cloned().unwrap_or_else(|_| json!({}));
1215 + let (title, kind, locations) =
1216 + tools::describe(&self.toolsets, &ctx.cwd, &ctx.roots, &tc.name, &args);
1217 + notify(SessionUpdate::ToolCall(
1218 + ToolCall::new(tc.id.clone(), title)
1219 + .name(tc.name.clone())
1220 + .kind(kind)
1221 + .status(ToolCallStatus::InProgress)
1222 + .locations(locations)
1223 + .raw_input(args.clone()),
1224 + ))?;
1225 +
1226 + let outcome: ToolOutcome = if parsed.is_err() {
1227 + ToolOutcome::err(format!("Invalid JSON arguments: {}", tc.arguments))
1228 + } else {
1229 + tools::execute(
1230 + ctx,
1231 + &self.toolsets,
1232 + self.profile.as_ref(),
1233 + &tc.id,
1234 + &tc.name,
1235 + args,
1236 + )
1237 + .await
1238 + };
1239 +
1240 + let mut fields = ToolCallUpdateFields::new()
1241 + .status(if outcome.failed {
1242 + ToolCallStatus::Failed
1243 + } else {
1244 + ToolCallStatus::Completed
1245 + })
1246 + .content(outcome.content)
1247 + .raw_output(json!({ "output": outcome.text, "failed": outcome.failed }));
1248 + if !outcome.locations.is_empty() {
1249 + fields = fields.locations(locations_of(&outcome.locations));
1250 + }
1251 + ctx.update(&tc.id, fields)?;
1252 + messages.push(tool_message(&tc.id, &outcome.text, outcome.failed));
1253 + }
1254 + if ctx.cancel.is_cancelled() {
1255 + return Ok(StopReason::Cancelled);
1256 + }
1257 + }
1258 + Ok(StopReason::MaxTurnRequests)
1259 + }
1260 +}
1261 +
1262 +/// Marker error: the prompt named a session that doesn't exist.
1263 +#[derive(Debug)]
1264 +struct UnknownSession(String);
1265 +
1266 +impl std::fmt::Display for UnknownSession {
1267 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1268 + write!(f, "unknown session {}", self.0)
1269 + }
1270 +}
1271 +
1272 +impl std::error::Error for UnknownSession {}
1273 +
1274 +#[cfg(test)]
1275 +mod tests {
1276 + use super::*;
1277 +
1278 + #[test]
1279 + fn iso8601_known_dates() {
1280 + assert_eq!(iso8601(0), "1970-01-01T00:00:00Z");
1281 + assert_eq!(iso8601(1_700_000_000), "2023-11-14T22:13:20Z");
1282 + assert_eq!(iso8601(951_782_400), "2000-02-29T00:00:00Z");
1283 + }
1284 +
1285 + #[test]
1286 + fn titles_are_trimmed_and_single_line() {
1287 + assert_eq!(session_title(" hello\nworld"), "hello");
1288 + let long = "x".repeat(200);
1289 + let t = session_title(&long);
1290 + assert_eq!(t.chars().count(), 80);
1291 + assert!(t.ends_with('…'));
1292 + }
1293 +
1294 + #[test]
1295 + fn terminal_auth_support_from_capability_or_legacy_meta() {
1296 + let caps = |v: Value| serde_json::from_value::<ClientCapabilities>(v).unwrap();
1297 + assert!(!supports_terminal_auth(&caps(json!({}))));
1298 + assert!(supports_terminal_auth(&caps(
1299 + json!({"auth": {"terminal": true}})
1300 + )));
1301 + // The ACP registry validator's payload.
1302 + assert!(supports_terminal_auth(&caps(
1303 + json!({"terminal": true, "_meta": {"terminal_output": true, "terminal-auth": true}})
1304 + )));
1305 + assert!(!supports_terminal_auth(&caps(
1306 + json!({"_meta": {"terminal-auth": false}})
1307 + )));
1308 + }
1309 +
1310 + #[test]
1311 + fn version_negotiation_always_v1() {
1312 + assert_eq!(negotiate_version(ProtocolVersion::V1), ProtocolVersion::V1);
1313 + assert_eq!(
1314 + negotiate_version(ProtocolVersion::from(7)),
1315 + ProtocolVersion::V1
1316 + );
1317 + assert_eq!(
1318 + negotiate_version(ProtocolVersion::from(0)),
1319 + ProtocolVersion::V1
1320 + );
1321 + }
1322 +
1323 + #[test]
1324 + fn private_failure_flag_never_reaches_the_wire() {
1325 + let m = tool_message("c1", "boom", true);
1326 + assert_eq!(m["x_failed"], true);
1327 + let w = wire_message(&m);
1328 + assert!(w.get("x_failed").is_none());
1329 + assert_eq!(w["content"], "boom");
1330 + assert!(tool_message("c1", "ok", false).get("x_failed").is_none());
1331 + }
1332 +
1333 + #[test]
1334 + fn user_text_handles_parts() {
1335 + let m = json!({"role": "user", "content": [
1336 + {"type": "text", "text": "hi"},
1337 + {"type": "image_url", "image_url": {"url": "x"}}
1338 + ]});
1339 + assert_eq!(user_text(&m), "hi\n\n[image]");
1340 + assert_eq!(user_text(&json!({"content": "plain"})), "plain");
1341 + }
1342 +}
crates/ed-acp/src/lib.rs new
+136
@@ -0,0 +1,136 @@
1 +//! An Agent Client Protocol (ACP) v1 server for Onde agents.
2 +//!
3 +//! `ed-acp` is the part of an ACP agent that is the same for every product: `initialize` and
4 +//! version negotiation, terminal auth, `authenticate` and `logout`, durable sessions (`list`,
5 +//! `load` with full replay, `resume`, `close`, `delete`), the model picker and auto-approve
6 +//! config options, slash-command advertisement, prompt content (embedded and linked files,
7 +//! images, audio), the turn loop against Onde Cloud with streaming, message ids and usage,
8 +//! tool approval, the workspace tools routed through the client's `fs/*` and `terminal/*`, and
9 +//! MCP servers through [`ed_mcp`].
10 +//!
11 +//! A product supplies a [`Profile`]: its name, prompt, tools, slash commands and built-in MCP
12 +//! servers. Everything in the Onde Agent Platform §5.5 conformance table is implemented here
13 +//! once.
14 +//!
15 +//! ```ignore
16 +//! let profile = Arc::new(MyProfile);
17 +//! ed_acp::serve_stdio(profile, ServeOptions::from_env(&info)).await?;
18 +//! ```
19 +
20 +mod agent;
21 +pub mod cli;
22 +pub mod content;
23 +pub mod llm;
24 +mod server;
25 +pub mod store;
26 +pub mod tools;
27 +mod workspace;
28 +
29 +use std::path::{Path, PathBuf};
30 +use std::sync::Arc;
31 +
32 +use agent_client_protocol::schema::v1::{AvailableCommand, PromptCapabilities};
33 +use async_trait::async_trait;
34 +use ed_mcp::{Connection, McpTool, McpToolset, SharedServer};
35 +use serde_json::Value;
36 +
37 +pub use agent_client_protocol;
38 +pub use content::AudioHints;
39 +pub use ed_mcp;
40 +pub use llm::{LlmConfig, LlmEnv};
41 +pub use server::{ServeOptions, serve_stdio};
42 +pub use tools::{
43 + DescribeCtx, ToolCtx, ToolOutcome, Toolset, absolutize, function_def, normalize, resolve_in,
44 + truncate,
45 +};
46 +pub use workspace::WorkspaceTools;
47 +
48 +/// Who the agent is.
49 +#[derive(Debug, Clone, Copy)]
50 +pub struct AgentInfo {
51 + /// Binary and protocol name, e.g. `splitfire-agent`. Reported as `agentInfo.name` and in
52 + /// `clientInfo` to MCP servers.
53 + pub name: &'static str,
54 + /// Name shown to people, e.g. `SplitFire`.
55 + pub display_name: &'static str,
56 + pub version: &'static str,
57 + /// Where settings and credentials come from.
58 + pub env: LlmEnv,
59 +}
60 +
61 +/// What the system prompt is built from, every turn.
62 +pub struct PromptCtx<'a> {
63 + /// `acp` or `tui`; see [`ServeOptions::surface`].
64 + pub surface: &'a str,
65 + pub cwd: &'a Path,
66 + pub roots: &'a [PathBuf],
67 + /// The session's MCP tools, so the prompt can mention optional capabilities only when
68 + /// they are connected.
69 + pub mcp: &'a McpToolset,
70 +}
71 +
72 +/// What slash commands depend on.
73 +pub struct SessionCtx<'a> {
74 + pub mcp: &'a McpToolset,
75 +}
76 +
77 +/// An MCP call awaiting approval.
78 +pub struct McpCall<'a> {
79 + pub tool: &'a McpTool,
80 + pub args: &'a Value,
81 + /// The server, for products that consult it (e.g. to report a download size).
82 + pub connection: Option<&'a Connection>,
83 +}
84 +
85 +/// The product-specific part of an ACP agent.
86 +#[async_trait]
87 +pub trait Profile: Send + Sync + 'static {
88 + fn info(&self) -> AgentInfo;
89 +
90 + /// The system prompt for one turn.
91 + fn system_prompt(&self, ctx: &PromptCtx<'_>) -> String;
92 +
93 + /// The tools the model may call, besides MCP tools. Include [`WorkspaceTools`] to offer
94 + /// file and shell access.
95 + fn toolsets(&self) -> Vec<Arc<dyn Toolset>>;
96 +
97 + /// Commands advertised with `available_commands_update` after a session is created or
98 + /// reopened.
99 + fn slash_commands(&self, _ctx: &SessionCtx<'_>) -> Vec<AvailableCommand> {
100 + Vec::new()
101 + }
102 +
103 + /// When `text` starts with one of this profile's commands, the instruction the model gets
104 + /// in its place.
105 + fn expand_slash(&self, _text: &str, _ctx: &SessionCtx<'_>) -> Option<String> {
106 + None
107 + }
108 +
109 + /// Process-wide MCP servers offered to every session unless the client supplies one of
110 + /// the same name.
111 + fn builtin_mcp_servers(&self) -> Vec<&'static SharedServer> {
112 + Vec::new()
113 + }
114 +
115 + /// Advertised in `initialize`. Audio is accepted only when this says so.
116 + fn prompt_capabilities(&self) -> PromptCapabilities {
117 + PromptCapabilities::new().embedded_context(true).image(true)
118 + }
119 +
120 + /// How audio links and attachments are described to the model.
121 + fn audio_hints(&self) -> AudioHints {
122 + AudioHints::default()
123 + }
124 +
125 + /// The permission prompt for a mutating MCP tool. `None` uses the default: the tool, its
126 + /// server and its arguments.
127 + async fn mcp_permission_preview(&self, _call: McpCall<'_>) -> Option<String> {
128 + None
129 + }
130 +
131 + /// Files an MCP result produced, from its `structuredContent`, shown as tool-call
132 + /// locations. Relative paths are dropped.
133 + fn mcp_result_locations(&self, _tool: &McpTool, _structured: &Value) -> Vec<PathBuf> {
134 + Vec::new()
135 + }
136 +}