From aeb86a69a35db8afca288a616e8622e866e03c52 Mon Sep 17 00:00:00 2001 From: BFlat Date: Tue, 3 Nov 2026 20:41:06 +0700 Subject: [PATCH] Agent setup and streaming fixes: Aider, Antigravity on Windows, chat-only models, SSE limits and retries (#653) Replays #616, #617, #619, #620 and #621. Aider isn't offered until it has an adapter, Antigravity is found under a redirected LOCALAPPDATA, servers with only specialised models aren't set up for chat, an SSE event's data is capped, and a retry keeps the original request when the agent's session can't be restored. Program lookup adds lower-case extensions on Windows. Co-authored-by: Afloat <132809511+Afloat16@users.noreply.github.com> --- crates/ai/src/cli_agents/acp.rs | 137 +++++++++++++++++++++++++++++--- crates/ai/src/cli_agents/mod.rs | 136 +++++++++++++++++++++++-------- crates/ai/src/detect.rs | 49 ++++++++++-- crates/ai/src/http.rs | 37 +++++++++ crates/ai/src/lib.rs | 4 +- crates/ai/src/tests.rs | 4 + crates/platform/src/which.rs | 56 ++++++++++--- docs/ROADMAP.md | 14 +++- 8 files changed, 369 insertions(+), 68 deletions(-) diff --git a/crates/ai/src/cli_agents/acp.rs b/crates/ai/src/cli_agents/acp.rs index 7a379fbf..6c36c862 100644 --- a/crates/ai/src/cli_agents/acp.rs +++ b/crates/ai/src/cli_agents/acp.rs @@ -110,6 +110,11 @@ enum Pending { Prompt, } +struct QueuedPrompt { + text: String, + continuation: Option, +} + /// The ACP protocol for one conversation. pub struct AcpCore { backend: String, @@ -121,7 +126,9 @@ pub struct AcpCore { pending: HashMap, session: Option, ready: bool, - queued: VecDeque, + queued: VecDeque, + /// Only a loaded session or one already sent a prompt can continue an interrupted turn. + session_has_history: bool, /// Permission requests waiting for an answer: our call id → (JSON-RPC id, options). asks: HashMap)>, /// Tool call titles by id, for the finished event. @@ -144,6 +151,7 @@ impl AcpCore { session: None, ready: false, queued: VecDeque::new(), + session_has_history: false, asks: HashMap::new(), tools: HashMap::new(), decides: false, @@ -181,8 +189,12 @@ impl AcpCore { /// A message from the person: sent now, or once the session is ready. pub fn prompt(&mut self, text: &str) -> Vec { + self.queue_prompt(text, None) + } + + fn queue_prompt(&mut self, text: &str, continuation: Option) -> Vec { self.failure = None; - self.queued.push_back(text.to_string()); + self.queued.push_back(QueuedPrompt { text: text.to_string(), continuation }); // A start that failed earlier (an `initialize` error) left nothing under way: start again, // or the message would wait for a session that never comes (#271). if !self.ready && self.pending.is_empty() { @@ -194,7 +206,9 @@ impl AcpCore { fn flush(&mut self) -> Vec { let busy = self.pending.values().any(|p| *p == Pending::Prompt); let Some(sid) = self.session.clone().filter(|_| self.ready && !busy) else { return Vec::new() }; - let Some(text) = self.queued.pop_front() else { return Vec::new() }; + let Some(prompt) = self.queued.pop_front() else { return Vec::new() }; + let text = if self.session_has_history { prompt.continuation.unwrap_or(prompt.text) } else { prompt.text }; + self.session_has_history = true; let params = json!({"sessionId": sid, "prompt": [{"type": "text", "text": text}]}); vec![self.request("session/prompt", params, Pending::Prompt)] } @@ -234,6 +248,7 @@ impl AcpCore { // A session that can't be continued starts afresh rather than failing the message. if kind == Pending::LoadSession { self.resume = None; + self.session = None; let params = json!({"cwd": self.cwd, "mcpServers": self.mcp_servers()}); return vec![self.request("session/new", params, Pending::NewSession)]; } @@ -270,6 +285,7 @@ impl AcpCore { return self.fail(format!("{} started no session", self.agent_name), None); } self.ready = true; + self.session_has_history = kind == Pending::LoadSession; self.flush() } Pending::Prompt => { @@ -573,14 +589,7 @@ impl AcpConversation { s.apply(outs, &self.shared, &self.tx, &self.busy); } } -} - -impl Conversation for AcpConversation { - fn last_failure(&self) -> Option { - self.failure.or_else(|| self.lock().as_ref().and_then(|s| s.core.failure)) - } - - fn send(&mut self, text: &str) { + fn send_prompt(&mut self, text: &str, retry: bool) { if self.busy.swap(true, Ordering::SeqCst) { let _ = self.tx.send(AgentEvent::Error { message: "Still working on the last message.".into(), @@ -608,7 +617,22 @@ impl Conversation for AcpConversation { } } let text = format!("{}{text}", self.preamble); - self.with_core(|c| c.prompt(&text)); + let continuation = retry.then(|| format!("{}{}", self.preamble, crate::failure::CONTINUE_TURN)); + self.with_core(|c| c.queue_prompt(&text, continuation)); + } +} + +impl Conversation for AcpConversation { + fn last_failure(&self) -> Option { + self.failure.or_else(|| self.lock().as_ref().and_then(|s| s.core.failure)) + } + + fn send(&mut self, text: &str) { + self.send_prompt(text, false); + } + + fn retry(&mut self, text: &str) { + self.send_prompt(text, true); } fn session_id(&self) -> Option { @@ -739,6 +763,95 @@ mod tests { assert_eq!(sent(&c.line(r#"{"jsonrpc":"2.0","id":1,"result":{}}"#))[0]["method"], "session/new"); } + #[cfg(unix)] + fn prompt_after_interrupted_acp_turn(recovery: &str) -> String { + use std::os::unix::fs::PermissionsExt; + let dir = tempfile::tempdir().unwrap(); + let program = dir.path().join("acp-agent"); + let script = r#"#!/bin/sh +d=$(dirname "$0") +if [ ! -f "$d/started" ]; then + : > "$d/started" + IFS= read -r init + echo '{"jsonrpc":"2.0","id":1,"result":{"agentCapabilities":{"loadSession":true}}}' + IFS= read -r new + echo '{"jsonrpc":"2.0","id":2,"result":{"sessionId":"saved"}}' + IFS= read -r prompt + printf '%s\n' "$prompt" > "$d/first-prompt" + exit 7 +fi +IFS= read -r init +if [ "$1" = unsupported ]; then + echo '{"jsonrpc":"2.0","id":1,"result":{}}' + IFS= read -r new + echo '{"jsonrpc":"2.0","id":2,"result":{"sessionId":"fresh"}}' + prompt_id=3 +else + echo '{"jsonrpc":"2.0","id":1,"result":{"agentCapabilities":{"loadSession":true}}}' + IFS= read -r load + printf '%s\n' "$load" > "$d/load" + if [ "$1" = loaded ]; then + echo '{"jsonrpc":"2.0","id":2,"result":{}}' + prompt_id=3 + else + echo '{"jsonrpc":"2.0","id":2,"error":{"code":-32603,"message":"session unavailable"}}' + IFS= read -r new + echo '{"jsonrpc":"2.0","id":3,"result":{"sessionId":"fresh"}}' + prompt_id=4 + fi +fi +IFS= read -r prompt +printf '%s\n' "$prompt" > "$d/retry-prompt" +printf '{"jsonrpc":"2.0","id":%s,"result":{"stopReason":"end_turn"}}\n' "$prompt_id" +"#; + std::fs::write(&program, script).unwrap(); + std::fs::set_permissions(&program, std::fs::Permissions::from_mode(0o755)).unwrap(); + let mcp = McpCommand { command: "craft-anything-cli".into(), args: vec!["mcp".into()], notes: None }; + let mut c = AcpConversation::new("Scripted ACP", "gemini-cli", program.display().to_string(), vec![recovery.into()], dir.path().into(), mcp, None) + .with_preamble("P:".into()); + let finish = |c: &mut AcpConversation| { + let deadline = std::time::Instant::now() + Duration::from_secs(10); + let mut all = Vec::new(); + while !all.iter().any(|e| matches!(e, AgentEvent::TurnDone { .. })) { + assert!(std::time::Instant::now() < deadline, "no TurnDone: {all:?}"); + all.extend(c.poll()); + std::thread::sleep(Duration::from_millis(10)); + } + all + }; + let original = "Build the requested blue header."; + c.send(original); + assert!(finish(&mut c).iter().any(|e| matches!(e, AgentEvent::Error { .. }))); + assert_eq!(c.last_failure(), Some(TurnFailure::new(FailureKind::Transient))); + assert_eq!(c.session_id().as_deref(), Some("saved")); + c.retry(original); + let events = finish(&mut c); + assert!(!events.iter().any(|e| matches!(e, AgentEvent::Error { .. })), "{events:?}"); + let read = |name: &str| -> Value { serde_json::from_str(&std::fs::read_to_string(dir.path().join(name)).unwrap()).unwrap() }; + assert_eq!(read("first-prompt")["params"]["prompt"][0]["text"], format!("P:{original}")); + if recovery != "unsupported" { + assert_eq!(read("load")["params"]["sessionId"], "saved"); + } + let retry = read("retry-prompt"); + assert_eq!(retry["method"], "session/prompt"); + assert_eq!(retry["params"]["sessionId"], if recovery == "loaded" { "saved" } else { "fresh" }); + retry["params"]["prompt"][0]["text"].as_str().unwrap().to_string() + } + + #[cfg(unix)] + #[test] + fn acp_retry_preserves_the_request_when_recovery_starts_a_new_session() { + for recovery in ["missing", "unsupported"] { + assert_eq!(prompt_after_interrupted_acp_turn(recovery), "P:Build the requested blue header.", "{recovery}"); + } + } + + #[cfg(unix)] + #[test] + fn acp_retry_continues_a_successfully_loaded_session() { + assert_eq!(prompt_after_interrupted_acp_turn("loaded"), format!("P:{}", crate::failure::CONTINUE_TURN)); + } + fn update(c: &mut AcpCore, value: Value) -> Vec { c.line(&json!({"jsonrpc": "2.0", "method": "session/update", "params": {"sessionId": "old", "update": value}}).to_string()) } diff --git a/crates/ai/src/cli_agents/mod.rs b/crates/ai/src/cli_agents/mod.rs index e4c46363..9be49c42 100644 --- a/crates/ai/src/cli_agents/mod.rs +++ b/crates/ai/src/cli_agents/mod.rs @@ -45,7 +45,6 @@ pub const KNOWN: &[(&str, &str, &str)] = &[ ("qwen-code", "qwen", "Qwen Code"), ("goose", "goose", "Goose"), ("maki", "maki", "Maki"), - ("aider", "aider", "Aider"), ]; pub fn program_for(agent: &str) -> String { @@ -516,6 +515,38 @@ impl CliConversation { pub fn session_id(&self) -> Option { self.session.lock().unwrap_or_else(PoisonError::into_inner).clone() } + + fn send_prompt(&mut self, text: &str, retry: bool) { + if self.busy.swap(true, Ordering::SeqCst) { + let _ = self.tx.send(AgentEvent::Error { + message: "Still working on the last message.".into(), + remediation: Some("Wait for it to finish or press stop.".into()), + log: None, + }); + return; + } + self.stop.reset(); + self.lost.store(false, Ordering::SeqCst); + *self.failure.lock().unwrap_or_else(PoisonError::into_inner) = None; + let (spec, tx, stop, busy) = (self.spec.clone(), self.tx.clone(), self.stop.clone(), self.busy.clone()); + let state = RunState { session: self.session.clone(), verified: self.verified.clone(), lost: self.lost.clone(), failure: self.failure.clone() }; + let text = text.to_string(); + let spawned = std::thread::Builder::new().name("ai-cli".into()).spawn(move || { + let done = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| run_cli(&spec, &text, retry, &tx, &stop, &state))).unwrap_or_else(|_| { + let _ = tx.send(AgentEvent::Error { message: "The agent runner hit a bug and stopped safely.".into(), remediation: None, log: None }); + nothing_done(&spec.config.name) + }); + end_turn(&busy, done, |e| { + let _ = tx.send(e); + }); + }); + if spawned.is_err() { + let _ = self.tx.send(AgentEvent::Error { message: "Couldn't start a worker thread.".into(), remediation: None, log: None }); + end_turn(&self.busy, nothing_done(&self.spec.config.name), |e| { + let _ = self.tx.send(e); + }); + } + } } /// The environment an agent program runs with: Claude Code's limits raised for a long build, and @@ -574,7 +605,7 @@ fn nothing_done(backend: &str) -> AgentEvent { /// send once the conversation can take the next message. The session it reports is kept for the /// next message; one it was asked to continue but doesn't have is forgotten and flagged in /// `state.lost` before the error that says so is sent. -fn run_cli(spec: &CliSpec, text: &str, tx: &Sender, stop: &Stop, state: &RunState) -> AgentEvent { +fn run_cli(spec: &CliSpec, text: &str, retry: bool, tx: &Sender, stop: &Stop, state: &RunState) -> AgentEvent { let RunState { session, verified, lost, failure } = state; let name = spec.config.name.clone(); let agent = spec.config.model.clone(); @@ -608,6 +639,9 @@ fn run_cli(spec: &CliSpec, text: &str, tx: &Sender, stop: &Stop, sta text: format!("{program}'s help doesn't list {} here, so it runs with the plain adapter (prompt in, text out).", needed.join(" / ")), }); adapter = Arc::new(adapters::Plain); + // A saved session cannot be continued through the adapter that will actually run. + // Forget it before retry chooses between the original request and a continuation. + *session.lock().unwrap_or_else(PoisonError::into_inner) = None; } else if !ok { let _ = tx.send(AgentEvent::Status { text: format!("Couldn't confirm that {program} has {}; trying it anyway.", needed.join(" / ")) }); } @@ -615,6 +649,7 @@ fn run_cli(spec: &CliSpec, text: &str, tx: &Sender, stop: &Stop, sta let preamble = spec.preamble.clone().unwrap_or_else(|| crate::prompt::cli_preamble(spec.beginner)); let mcp = mcp_for(spec, remote.is_some()); let resume = session.lock().unwrap_or_else(PoisonError::into_inner).clone(); + let text = if retry && adapter.resumes() && resume.is_some() { crate::failure::CONTINUE_TURN } else { text }; let builder = spec.builder.is_some(); // A question goes only to an agent that can be held to reading (a fallback adapter can't). if spec.question && !adapter.answers_questions() { @@ -863,8 +898,7 @@ impl FailureContext { impl Conversation for CliConversation { fn retry(&mut self, text: &str) { - let resumable = self.spec.adapters.for_agent(&self.spec.config.model).resumes() && self.session_id().is_some(); - self.send(if resumable { crate::failure::CONTINUE_TURN } else { text }); + self.send_prompt(text, true); } fn last_failure(&self) -> Option { @@ -885,35 +919,7 @@ impl Conversation for CliConversation { /// and the conversation takes the next message before that's sent: whoever reads it may send /// the next message at once (a session that wasn't there is followed by a fresh start). fn send(&mut self, text: &str) { - if self.busy.swap(true, Ordering::SeqCst) { - let _ = self.tx.send(AgentEvent::Error { - message: "Still working on the last message.".into(), - remediation: Some("Wait for it to finish or press stop.".into()), - log: None, - }); - return; - } - self.stop.reset(); - self.lost.store(false, Ordering::SeqCst); - *self.failure.lock().unwrap_or_else(PoisonError::into_inner) = None; - let (spec, tx, stop, busy) = (self.spec.clone(), self.tx.clone(), self.stop.clone(), self.busy.clone()); - let state = RunState { session: self.session.clone(), verified: self.verified.clone(), lost: self.lost.clone(), failure: self.failure.clone() }; - let text = text.to_string(); - let spawned = std::thread::Builder::new().name("ai-cli".into()).spawn(move || { - let done = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| run_cli(&spec, &text, &tx, &stop, &state))).unwrap_or_else(|_| { - let _ = tx.send(AgentEvent::Error { message: "The agent runner hit a bug and stopped safely.".into(), remediation: None, log: None }); - nothing_done(&spec.config.name) - }); - end_turn(&busy, done, |e| { - let _ = tx.send(e); - }); - }); - if spawned.is_err() { - let _ = self.tx.send(AgentEvent::Error { message: "Couldn't start a worker thread.".into(), remediation: None, log: None }); - end_turn(&self.busy, nothing_done(&self.spec.config.name), |e| { - let _ = self.tx.send(e); - }); - } + self.send_prompt(text, false); } fn poll(&mut self) -> Vec { @@ -1642,6 +1648,70 @@ mod tests { assert!(text.contains("Request"), "preamble is passed too"); } + #[cfg(unix)] + #[test] + fn a_retry_uses_the_adapter_selected_after_checking_required_flags() { + retry_after_checking_flags(true); + } + + #[cfg(unix)] + #[test] + fn a_first_retry_waits_for_the_actual_adapter_selection() { + retry_after_checking_flags(false); + } + + #[cfg(unix)] + fn retry_after_checking_flags(send_first: bool) { + use std::os::unix::fs::PermissionsExt; + for supported in [false, true] { + let dir = tempfile::tempdir().unwrap(); + let program = dir.path().join("agy"); + let help = if supported { "--output-format stream-json" } else { "old command options" }; + let script = format!( + "#!/bin/sh\nd=$(dirname \"$0\")\nif [ \"$1\" = --help ]; then echo '{help}'; exit 0; fi\nprintf '%s\\n' \"$@\" > \"$d/argv\"\necho run >> \"$d/runs\"\nexit 7\n" + ); + std::fs::write(&program, script).unwrap(); + std::fs::set_permissions(&program, std::fs::Permissions::from_mode(0o755)).unwrap(); + let config = BackendConfig { + name: "Antigravity".into(), + kind: BackendKind::AgentCli, + model: "antigravity".into(), + base_url: None, + ssh_host: None, + enabled: true, + }; + let options = BackendOptions { command: Some(program.display().to_string()), ..Default::default() }; + let mut spec = CliSpec::new(config, options, false, None); + spec.preamble = Some("P:".into()); + spec.resume = Some("saved".into()); + let mut c = CliConversation::new(spec); + let finish = |c: &mut CliConversation| { + let deadline = std::time::Instant::now() + Duration::from_secs(10); + let mut all = Vec::new(); + while !all.iter().any(|e| matches!(e, AgentEvent::TurnDone { .. })) { + assert!(std::time::Instant::now() < deadline, "no TurnDone: {all:?}"); + all.extend(c.poll()); + std::thread::sleep(Duration::from_millis(10)); + } + all + }; + let original = "Build the requested blue header."; + if send_first { + c.send(original); + let first = finish(&mut c); + assert!(first.iter().any(|e| matches!(e, AgentEvent::Error { .. })), "{first:?}"); + } + c.retry(original); + let retried = finish(&mut c); + assert_eq!(retried.iter().any(|e| matches!(e, AgentEvent::Status { text } if text.contains("plain adapter"))), !supported); + let argv = std::fs::read_to_string(dir.path().join("argv")).unwrap(); + let expected = format!("P:{}", if supported { crate::failure::CONTINUE_TURN } else { original }); + assert_eq!(argv.lines().last(), Some(expected.as_str()), "supported={supported}: {argv}"); + assert_eq!(argv.contains("--conversation\nsaved\n"), supported, "{argv}"); + assert_eq!(std::fs::read_to_string(dir.path().join("runs")).unwrap(), if send_first { "run\nrun\n" } else { "run\n" }); + } + } + /// #402: a 34,641-character message in argv passed Windows' command-line limit (os error 206); /// Claude Code reads it from stdin instead. #[cfg(unix)] diff --git a/crates/ai/src/detect.rs b/crates/ai/src/detect.rs index 4906d813..7b8c5776 100644 --- a/crates/ai/src/detect.rs +++ b/crates/ai/src/detect.rs @@ -55,14 +55,17 @@ pub fn port_open(base_url: &str, timeout: Duration) -> bool { addrs.iter().any(|a| TcpStream::connect_timeout(a, timeout).is_ok()) } -/// Words in the names of models that turn text into numbers, rank or transcribe, and can't chat: -/// an embedding model listed first was used for every message and each one failed (#127). -const NOT_CHAT: &[&str] = &["embed", "minilm", "bge-", "rerank", "whisper", "tts"]; +/// Task tokens and known model families that turn text into numbers, rank or transcribe. +/// Match the model name, not its publisher: `matts/llama-3-instruct` can chat (#127). +const NOT_CHAT: &[&str] = &["embed", "embedding", "embeddings", "embeddinggemma", "minilm", "bge", "rerank", "reranker", "whisper", "tts", "xtts"]; -/// The first model a server lists that can chat (else its first, so something is chosen). +/// The first model a server lists that can chat. A specialised model isn't a chat fallback. pub(crate) fn chat_model(models: &[String]) -> Option { - let chats = |m: &&String| !NOT_CHAT.iter().any(|w| m.to_ascii_lowercase().contains(w)); - models.iter().find(chats).or(models.first()).cloned() + let chats = |m: &&String| { + let name = m.rsplit('/').next().unwrap_or(m).to_ascii_lowercase(); + !name.split(|c: char| !c.is_ascii_alphanumeric()).any(|token| NOT_CHAT.contains(&token)) + }; + models.iter().find(chats).cloned() } /// One local server: connect, then list models. @@ -78,7 +81,11 @@ pub fn probe_local(label: &str, base: &str) -> Option { // No answer in time: it may still be loading a model. Err(_) => Vec::new(), }; - let model = chat_model(&models).unwrap_or_default(); + let model = match chat_model(&models) { + Some(model) => model, + None if models.is_empty() => String::new(), + None => return None, + }; let reason = match models.len() { 0 => format!("Found {label} running at {base}"), 1 => format!("Found {label} running with 1 model"), @@ -287,12 +294,38 @@ mod tests { let names = |l: &[&str]| l.iter().map(|s| s.to_string()).collect::>(); assert_eq!(chat_model(&names(&["nomic-embed-text:latest", "llama3.2:3b"])).as_deref(), Some("llama3.2:3b")); assert_eq!(chat_model(&names(&["mxbai-embed-large", "all-MiniLM-L6-v2", "qwen3:8b"])).as_deref(), Some("qwen3:8b")); - assert_eq!(chat_model(&names(&["nomic-embed-text"])).as_deref(), Some("nomic-embed-text"), "something is chosen"); + assert_eq!(chat_model(&names(&["nomic-embed-text"])), None, "an embedding model cannot answer a chat request"); assert_eq!(chat_model(&[]), None); let url = crate::tests::serve("200 OK", r#"{"object":"list","data":[{"id":"nomic-embed-text:latest"},{"id":"gemma3:4b"}]}"#); assert_eq!(probe_local("Ollama", &url).unwrap().suggestion.model, "gemma3:4b"); } + #[test] + fn specialised_only_servers_are_not_offered_for_chat() { + for body in [ + r#"{"object":"list","data":[{"id":"nomic-embed-text"}]}"#, + r#"{"object":"list","data":[{"id":"bge-reranker-v2"}]}"#, + r#"{"object":"list","data":[{"id":"whisper-large"}]}"#, + r#"{"object":"list","data":[{"id":"tts-1"}]}"#, + r#"{"object":"list","data":[{"id":"all-MiniLM-L6-v2"}]}"#, + ] { + let url = crate::tests::serve("200 OK", body); + assert!(probe_local("Local server", &url).is_none(), "{body} has no chat model"); + } + } + + #[test] + fn task_markers_do_not_match_publishers_or_unrelated_words() { + for model in ["matts/llama-3-instruct", "embedding-author/qwen-2.5-instruct", "embedded-chat", "whisper/gemma-3"] { + assert_eq!(chat_model(&[model.into()]).as_deref(), Some(model)); + } + let url = crate::tests::serve("200 OK", r#"{"object":"list","data":[{"id":"matts/llama-3-instruct"}]}"#); + assert_eq!(probe_local("Local", &url).unwrap().suggestion.model, "matts/llama-3-instruct"); + for model in ["vendor/text-embedding-3-small", "nomic-embed-text:latest", "BAAI/bge-m3", "vendor/embeddinggemma-300m", "XTTS-v2"] { + assert!(chat_model(&[model.into()]).is_none(), "{model}"); + } + } + /// #83: a server set up in OpenCode (here LM Studio on another port) is offered; hosted /// services and broken settings aren't. #[test] diff --git a/crates/ai/src/http.rs b/crates/ai/src/http.rs index 87caafc9..537e9fe7 100644 --- a/crates/ai/src/http.rs +++ b/crates/ai/src/http.rs @@ -152,6 +152,8 @@ pub struct SseEvent { /// Longest single SSE line we accept (protects memory against a hostile server). const MAX_SSE_LINE: usize = 8 * 1024 * 1024; +/// Most data retained for one event, including newlines between its `data:` fields. +const MAX_SSE_EVENT_DATA: usize = 8 * 1024 * 1024; /// Parse an SSE stream, calling `f` for each event. `f` returns false to stop early. /// Handles CRLF, multi-line `data:`, comments (`:`), and a final event without a blank line. @@ -194,6 +196,10 @@ pub fn parse_sse(r: R, mut f: impl FnMut(SseEvent) -> Result cur.event = Some(value.to_string()), "data" => { + let size = cur.data.len().saturating_add(usize::from(has_data)).saturating_add(value.len()); + if size > MAX_SSE_EVENT_DATA { + return Err(ToolError::new(ErrorCode::OverLimit, "The stream sent an event whose data was too large.")); + } if has_data { cur.data.push('\n'); } @@ -243,6 +249,37 @@ mod tests { assert_eq!(n, 1); } + #[test] + fn sse_event_data_is_bounded_across_lines() { + let first = "é".repeat(2 * 1024 * 1024); + let full = format!("data:{first}\ndata:{}\n", "x".repeat(4 * 1024 * 1024 - 1)); + for stream in [format!("data:{first}\ndata:{first}\n\n"), format!("data:{first}\ndata:{first}"), format!("{full}data:\n\n")] { + let mut emitted = 0; + let error = parse_sse(stream.as_bytes(), |_| { + emitted += 1; + Ok(true) + }) + .unwrap_err(); + assert_eq!(error.code, ErrorCode::OverLimit); + assert_eq!(emitted, 0, "an oversized event must not reach the provider parser"); + assert!(!error.message.contains('é'), "the error must not echo response data"); + } + } + + #[test] + fn sse_event_data_limit_preserves_boundaries_and_eof() { + let first = "é".repeat(2 * 1024 * 1024); + let full = format!("data:{first}\r\ndata:{}\r\n", "x".repeat(4 * 1024 * 1024 - 1)); + let stream = format!("{full}\r\ndata:\r\n\r\n{full}"); + let mut lengths = Vec::new(); + parse_sse(stream.as_bytes(), |event| { + lengths.push(event.data.len()); + Ok(true) + }) + .unwrap(); + assert_eq!(lengths, [8 * 1024 * 1024, 0, 8 * 1024 * 1024]); + } + #[test] fn provider_errors_include_the_nested_reason() { let reason = "The selected model does not accept this request parameter."; diff --git a/crates/ai/src/lib.rs b/crates/ai/src/lib.rs index 68941c18..7dca3f29 100644 --- a/crates/ai/src/lib.rs +++ b/crates/ai/src/lib.rs @@ -417,7 +417,9 @@ impl AiHub { let model = match model.map(str::trim).filter(|m| !m.is_empty()) { Some(m) => m.to_string(), None => detect::chat_model(&models).ok_or_else(|| { - ToolError::new(ErrorCode::NotFound, format!("{address} has no models yet.")).with_remediation("Add a model on the server first, or name one.") + let reason = if models.is_empty() { "has no models yet" } else { "doesn't list a model that can chat" }; + ToolError::new(ErrorCode::NotFound, format!("{address} {reason}.")) + .with_remediation("Load a chat model on the server first, or enter the name of a chat model you know is available.") })?, }; let config = BackendConfig { diff --git a/crates/ai/src/tests.rs b/crates/ai/src/tests.rs index 4e161801..328e9bd7 100644 --- a/crates/ai/src/tests.rs +++ b/crates/ai/src/tests.rs @@ -349,6 +349,10 @@ fn a_server_is_added_by_its_address() { assert_eq!(h.add_server("Named", &url, Some("llama3"), None).unwrap(), "llama3", "a named model is kept"); assert!(h.add_server("Web page", &serve("404 Not Found", "nope"), None, None).is_err(), "not a model server"); assert!(h.add_server("Empty", &serve("200 OK", r#"{"object":"list","data":[]}"#), None, None).is_err(), "no models to use"); + let embeddings = serve("200 OK", r#"{"object":"list","data":[{"id":"nomic-embed-text"}]}"#); + let error = h.add_server("Embeddings", &embeddings, None, None).unwrap_err(); + assert_eq!(error.code, ErrorCode::NotFound); + assert!(error.message.contains("model that can chat")); assert!(h.add_server("No scheme", "192.168.1.20:11434/v1", None, None).is_err()); assert!(h.add_server(" ", &url, None, None).is_err()); assert_eq!(h.backends().len(), 2, "only the ones that answered"); diff --git a/crates/platform/src/which.rs b/crates/platform/src/which.rs index 578cfdd3..eb051805 100644 --- a/crates/platform/src/which.rs +++ b/crates/platform/src/which.rs @@ -40,7 +40,8 @@ const HOME_VAR: &str = if cfg!(windows) { "USERPROFILE" } else { "HOME" }; /// shell's, and would miss a coding agent installed in `~/.local/bin`; on Windows an installer /// may not have added its folder to `PATH` at all. pub fn search_path() -> Option { - with_usual_dirs(std::env::var_os("PATH"), std::env::var_os(HOME_VAR).map(PathBuf::from)) + let local_data = cfg!(windows).then(|| std::env::var_os("LOCALAPPDATA")).flatten().map(PathBuf::from); + with_usual_dirs(std::env::var_os("PATH"), std::env::var_os(HOME_VAR).map(PathBuf::from), local_data) } /// How to start `program`, and the arguments that go before the caller's. On Windows only `.exe` @@ -145,11 +146,15 @@ fn npmrc_prefix(home: &Path) -> Option { }) } -fn with_usual_dirs(path: Option, home: Option) -> Option { +fn with_usual_dirs(path: Option, home: Option, local_data: Option) -> Option { let mut dirs: Vec = path.as_ref().map(|p| std::env::split_paths(p).collect()).unwrap_or_default(); let before = dirs.len(); let in_home = home.iter().flat_map(|h| USUAL_HOME_DIRS.iter().map(move |d| h.join(d)).chain(nvm_bins(h)).chain(npm_global_bin(h))); - for d in in_home.chain(USUAL_DIRS.iter().map(PathBuf::from)) { + // Antigravity's per-user installer may use a redirected LOCALAPPDATA directory. + let local_data = + local_data.filter(|p| !p.as_os_str().is_empty()).or_else(|| home.as_ref().filter(|_| cfg!(windows)).map(|h| h.join("AppData").join("Local"))); + let antigravity = local_data.map(|p| p.join("agy").join("bin")); + for d in in_home.chain(antigravity).chain(USUAL_DIRS.iter().map(PathBuf::from)) { if !dirs.contains(&d) { dirs.push(d); } @@ -231,7 +236,8 @@ fn candidates(base: &Path) -> Vec { let exts = std::env::var("PATHEXT").unwrap_or_else(|_| ".COM;.EXE;.BAT;.CMD".into()); for e in exts.split(';').filter(|e| !e.is_empty()) { let mut s = base.as_os_str().to_os_string(); - s.push(e); + // Lower case, as installers name their files; PATHEXT spells them `.EXE`. + s.push(e.to_ascii_lowercase()); v.push(PathBuf::from(s)); } } @@ -272,14 +278,14 @@ mod tests { fn usual_dirs_come_after_path() { let home = PathBuf::from("/home/u"); let path = std::env::join_paths([PathBuf::from("/usr/bin"), home.join(".local/bin")]).unwrap(); - let dirs: Vec = std::env::split_paths(&with_usual_dirs(Some(path), Some(home.clone())).unwrap()).collect(); + let dirs: Vec = std::env::split_paths(&with_usual_dirs(Some(path), Some(home.clone()), None).unwrap()).collect(); assert_eq!(dirs[..3], [PathBuf::from("/usr/bin"), home.join(".local/bin"), home.join(".claude/local")]); // (Plus npm's prefix where the environment sets one.) let npm = usize::from(npm_global_bin(&home).is_some()); assert_eq!(dirs.len(), 1 + USUAL_HOME_DIRS.len() + USUAL_DIRS.len() + npm, "nothing twice"); assert!(dirs.contains(&PathBuf::from("/opt/homebrew/bin"))); // No PATH and no home folder: the system-wide folders still. - assert_eq!(with_usual_dirs(None, None), std::env::join_paths(USUAL_DIRS).ok()); + assert_eq!(with_usual_dirs(None, None, None), std::env::join_paths(USUAL_DIRS).ok()); } /// #112: agents npm installed under nvm, newest Node first, after the usual folders. @@ -293,7 +299,7 @@ mod tests { let bins = nvm_bins(home.path()); let node = home.path().join(".nvm/versions/node"); assert_eq!(bins, [node.join("v20.10.0/bin"), node.join("v18.19.1/bin"), node.join("v9.11.2/bin")]); - let dirs: Vec = std::env::split_paths(&with_usual_dirs(None, Some(home.path().to_path_buf())).unwrap()).collect(); + let dirs: Vec = std::env::split_paths(&with_usual_dirs(None, Some(home.path().to_path_buf()), None).unwrap()).collect(); assert!(dirs.contains(&home.path().join(".opencode/bin"))); let at = |p: &Path| dirs.iter().position(|d| d == p).unwrap(); assert!(at(&home.path().join(".cargo/bin")) < at(&bins[0]) && at(&bins[2]) < at(Path::new("/usr/local/bin"))); @@ -310,7 +316,7 @@ mod tests { let bin = home.path().join(r"AppData\Local\hermes\bin"); std::fs::create_dir_all(&bin).unwrap(); std::fs::write(bin.join("hermes.exe"), "").unwrap(); - let path = with_usual_dirs(None, Some(home.path().to_path_buf())); + let path = with_usual_dirs(None, Some(home.path().to_path_buf()), None); assert_eq!(which_in("hermes", path), [bin.join("hermes.exe")]); } @@ -330,21 +336,49 @@ mod tests { if std::env::var_os("NPM_CONFIG_PREFIX").is_none() && std::env::var_os("npm_config_prefix").is_none() { std::fs::write(&rc, "prefix=~/.npm-packages\n").unwrap(); let bin = if cfg!(windows) { home.path().join(".npm-packages") } else { home.path().join(".npm-packages").join("bin") }; - let dirs: Vec = std::env::split_paths(&with_usual_dirs(None, Some(home.path().to_path_buf())).unwrap()).collect(); + let dirs: Vec = std::env::split_paths(&with_usual_dirs(None, Some(home.path().to_path_buf()), None).unwrap()).collect(); assert!(dirs.contains(&bin), "{dirs:?}"); } } + #[test] + fn antigravity_is_found_in_redirected_local_app_data() { + let dir = tempfile::tempdir().unwrap(); + let home = dir.path().join("profile"); + let local_data = dir.path().join("redirected-local-data"); + let bin = local_data.join("agy").join("bin"); + std::fs::create_dir_all(&bin).unwrap(); + let program = bin.join(if cfg!(windows) { "agy.exe" } else { "agy" }); + std::fs::write(&program, b"fixture").unwrap(); + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions(&program, std::fs::Permissions::from_mode(0o755)).unwrap(); + } + let path = with_usual_dirs(None, Some(home.clone()), Some(local_data.clone())); + assert_eq!(which_in("agy", path.clone()).as_slice(), std::slice::from_ref(&program)); + let dirs: Vec<_> = std::env::split_paths(path.as_ref().unwrap()).collect(); + assert!(!dirs.contains(&home.join("AppData").join("Local").join("agy").join("bin"))); + // A folder already in PATH retains its position and is not added again. + let explicit = std::env::join_paths([&bin]).unwrap(); + let path = with_usual_dirs(Some(explicit), Some(home), Some(local_data)).unwrap(); + let dirs: Vec<_> = std::env::split_paths(&path).collect(); + assert_eq!(dirs.first(), Some(&bin)); + assert_eq!(dirs.iter().filter(|d| *d == &bin).count(), 1); + assert_eq!(which_in("agy", Some(path)), [program]); + } + #[cfg(windows)] #[test] fn usual_windows_dirs_come_after_path() { let home = PathBuf::from(r"C:\Users\u"); let path = OsString::from(r"C:\Windows;C:\Users\u\.local\bin"); - let dirs: Vec = std::env::split_paths(&with_usual_dirs(Some(path), Some(home.clone())).unwrap()).collect(); + let dirs: Vec = std::env::split_paths(&with_usual_dirs(Some(path), Some(home.clone()), None).unwrap()).collect(); assert_eq!(dirs[..3], [PathBuf::from(r"C:\Windows"), home.join(r".local\bin"), home.join(r"AppData\Roaming\npm")]); // (Plus npm's prefix where the environment sets one.) let npm = usize::from(npm_global_bin(&home).is_some()); - assert_eq!(dirs.len(), 1 + USUAL_HOME_DIRS.len() + USUAL_DIRS.len() + npm, "nothing twice"); + assert_eq!(dirs.len(), 2 + USUAL_HOME_DIRS.len() + USUAL_DIRS.len() + npm, "nothing twice"); + assert!(dirs.contains(&home.join("AppData").join("Local").join("agy").join("bin"))); // Each way Claude Code installs on Windows: native, npm, WinGet (per user and machine-wide). assert!(dirs.contains(&home.join(r"AppData\Local\Microsoft\WinGet\Links"))); assert!(dirs.contains(&PathBuf::from(r"C:\Program Files\WinGet\Links"))); diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 1e034ea0..3bb5e578 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -170,6 +170,9 @@ only when it's green and a maintainer has fully reviewed it, code as well as wor ### R5.3 A builder that can't make progress stops Owner: open · Status: todo Every round spends the person's money or quota. One that can't do anything shouldn't start. +- [x] Preserve the interrupted request when an agent cannot restore its session: ACP starts a new + session, or a registered CLI adapter falls back to Plain (#552 review). + — claimed: Afloat16, branch bugfix/r5.3-interrupted-request-recovery-20261103 - [x] A build stops and says why after two builder turns in a row that changed nothing in `app/`, on every backend (#44, #110, #417, #481). The count is saved in the project, so it holds across runs: a run ends at 7 checks (`max_rounds`) but check numbers carry on, so #44's check @@ -214,9 +217,13 @@ been run live. services whose real window differs from 150k (#374) — implementation: `6bd94b84828ec89b805cb8a246f557d0a778c037` - [ ] Find LM Studio on another port (#73) - [ ] See why Bionic or LM Studio can't be picked (#140) +- [x] Do not select an embedding, ranking or transcription model as a chat backend when a local server has no chat models — claimed: Afloat16, branch bugfix/r5.4-nonchat-model-detection-20261103 ### R5.5 API-key builds fail in words people can use Owner: open · Status: todo +- [x] Bound accumulated data across all lines of one server-sent event, so short lines cannot + grow a streamed API response without limit + — claimed: Afloat16, branch bugfix/r5.5-bounded-sse-events-20261103 - [x] Show the provider's own reason: #369 bought $10 of credit and got only "OpenRouter answered 400: Provider returned error". Find the reason in the response body and show it — PR #542 - [ ] A live API-key build (needs a key on the test machine) @@ -349,8 +356,9 @@ Owner: open · Status: in progress - [ ] OpenCode's plain adapter never passes our tool server, so its builder has none of our tools and hunts for a setting that doesn't exist (#481). Hand it the server, or build over ACP by default once that passes -- [ ] Aider has no adapter, so the plain one hands it the prompt as a file name, and it can open a - browser sign-in nobody agreed to (#72). Give it an adapter, or stop offering it +- [x] Aider has no adapter, so the plain one hands it the prompt as a file name, and it can open a + browser sign-in nobody agreed to (#72). Give it an adapter, or stop offering it — claimed: Afloat16, branch bugfix/r2.2-aider-recommendations-20261103 +- [ ] Add a non-interactive Aider adapter before recommending it again; detection currently leaves it out. - [ ] Codex (#499): in a chat the tool server has no active target, tools turned on with `enable_toolset` can't be called, The reporter attached a patch for chat continuity; look at that first @@ -418,7 +426,7 @@ Owner: open · Status: in progress (#259, a guess at its cause) - [ ] macOS: `/Applications/Antigravity.app` ended up as the agent's command (#467); find what writes it (no code for it was found; maybe the file picker) -- [ ] Windows: look in `%LOCALAPPDATA%\agy\bin` too (#62) +- [x] Windows: look in `%LOCALAPPDATA%\agy\bin` too (#62) — claimed: Afloat16, branch bugfix/r2.2-antigravity-search-20261103 - [ ] "Claude Code on " is offered for any SSH host, whether or not it has Claude Code (#62) - [ ] Gemini CLI's sign-in reportedly refuses individuals now (#400): check, and stop recommending