diff --git a/src/acp/service.rs b/src/acp/service.rs index b86d9b0..b92f4ad 100644 --- a/src/acp/service.rs +++ b/src/acp/service.rs @@ -1557,7 +1557,7 @@ impl AcpService { let _ = stream_sender.send(crate::llm::ChunkMessage::End); }); - let response = self + let turn = self .receive_prompt_stream( &session_id, &session, @@ -1569,7 +1569,11 @@ impl AcpService { base_cost, ) .await?; - if !cancellation.is_cancelled() && response.stop_reason != StopReason::Cancelled { + // Compaction would call the provider that just failed this turn. + if !cancellation.is_cancelled() + && turn.response.stop_reason != StopReason::Cancelled + && !turn.provider_failed + { let _ = self .maybe_auto_compact_session( &session_id, @@ -1581,7 +1585,7 @@ impl AcpService { ) .await?; } - Ok(response) + Ok(turn.response) } // Keep finalization outside every cancellable await, including client requests. @@ -1595,7 +1599,7 @@ impl AcpService { base_context_tokens: usize, base_prompt_tokens: usize, base_cost: f64, - ) -> Result { + ) -> Result { let mut assistant = crate::session::types::Message::incomplete(""); let message_id = assistant.id.clone(); assistant.provider = Some(session.provider.clone()); @@ -1727,7 +1731,10 @@ impl AcpService { cancelled = true; break; } - crate::llm::ChunkMessage::Failed(error) => failed = Some(error), + crate::llm::ChunkMessage::Failed(error) => { + failed = Some(error); + break; + } crate::llm::ChunkMessage::TurnStopReason(reason) => { turn_stop_reason = Some(reason) } @@ -1780,6 +1787,51 @@ impl AcpService { } } + if !cancelled && !cancellation.is_cancelled() { + if let Some(diagnostic) = failed.as_deref() { + // The raw error is an internal diagnostic (step stats, tool names, request + // body). Keep it in the log only; the transcript and the model get a summary. + crate::emit_log!( + "[ACP] provider failure session={} error={}", + session_id, + diagnostic + ); + let summary = provider_failure_summary(diagnostic); + let error = summary.as_str(); + let pending_ids = assistant + .parts + .iter() + .filter(|part| part.part_type == "tool_call") + .filter_map(|part| part.tool_id().map(str::to_string)) + .filter(|id| { + assistant.tool_result_part_data(id).is_none_or(|data| { + matches!(data.get("status").and_then(|s| s.as_str()), + None | Some("running" | "pending")) + }) + }) + .collect::>(); + assistant.mark_running_tool_parts_failed(error); + for id in pending_ids { + let payload = assistant.tool_result_part_data(&id).unwrap(); + send_tool_result( + connection, + session_id, + crate::llm::ToolCallResult { + tool_call_id: id, + role: "tool".to_string(), + name: payload["name"].as_str().unwrap_or("tool").to_string(), + content: payload.to_string(), + }, + &session.cwd, + )?; + } + let text = format!( + "\n\nThe AI request failed: {error}\nYou can retry or send another message in this session." + ); + assistant.append(&text); + send_text(connection, session_id, &message_id, text, false)?; + } + } Ok(()) } .await; @@ -1788,12 +1840,20 @@ impl AcpService { } assistant.is_complete = true; assistant.was_interrupted = cancelled || cancellation.is_cancelled(); + if assistant.was_interrupted || failed.is_some() { + assistant.mark_running_tool_parts_failed( + &failed + .as_deref() + .map(provider_failure_summary) + .unwrap_or_else(|| "Tool execution cancelled by user".to_string()), + ); + } { let mut manager = self.session_manager.lock().map_err(|_| internal_error())?; let persist_result = manager.add_message_to_session(session_id, &assistant); let status = if assistant.was_interrupted { crate::session::types::SessionStatus::Interrupted - } else if failed.is_some() { + } else if stream_result.is_err() { crate::session::types::SessionStatus::Failed } else { crate::session::types::SessionStatus::Idle @@ -1804,13 +1864,13 @@ impl AcpService { persist_result.map_err(|_| internal_error())?; } if assistant.was_interrupted { - return Ok(PromptResponse::new(StopReason::Cancelled)); + return Ok(PromptTurn::new(StopReason::Cancelled, false)); } stream_result?; - if let Some(error) = failed { - return Err(internal_error_with(&error)); + if failed.is_some() { + return Ok(PromptTurn::new(StopReason::EndTurn, true)); } - Ok(PromptResponse::new(acp_stop_reason(turn_stop_reason))) + Ok(PromptTurn::new(acp_stop_reason(turn_stop_reason), false)) } async fn compact_session( @@ -2224,6 +2284,85 @@ fn compacted_messages( Ok((compacted, stats)) } +/// A finished prompt turn. Provider failures are reported in-band as `end_turn`. +#[derive(Debug)] +struct PromptTurn { + response: PromptResponse, + provider_failed: bool, +} + +impl PromptTurn { + fn new(stop_reason: StopReason, provider_failed: bool) -> Self { + Self { + response: PromptResponse::new(stop_reason), + provider_failed, + } + } +} + +const PROVIDER_FAILURE_SUMMARY_MAX_CHARS: usize = 300; + +/// Short, one-line provider error for the transcript and the model. +/// +/// Provider failures arrive as internal diagnostics such as +/// `provider_step_error step=1 messages=2 tools=20 ... tool_names=[...] error=Provider error: +/// API error 400 Bad Request: {"error":{"message":"..."}}`. Prefer the provider's own error +/// message and drop the step stats, tool names, URLs and raw bodies. +fn provider_failure_summary(diagnostic: &str) -> String { + let mut error = diagnostic.trim(); + if error.starts_with("provider_step_error ") { + if let Some((_, rest)) = error.split_once(" error=") { + error = rest.trim(); + } + } + let summary = provider_json_error_message(error).unwrap_or_else(|| { + let mut text = error; + for prefix in [ + "Streaming failed: ", + "Retryable provider error: ", + "Provider error: ", + "Stream error: ", + ] { + text = text.strip_prefix(prefix).unwrap_or(text); + } + // Raw bodies (possibly truncated JSON) and request URLs are diagnostics. + for marker in [" url=", " body=", ": {", ": ["] { + if let Some(index) = text.find(marker) { + text = &text[..index]; + } + } + text.to_string() + }); + let summary = summary.lines().map(str::trim).find(|line| !line.is_empty()); + let Some(summary) = summary else { + return "Provider request failed".to_string(); + }; + if summary.chars().count() <= PROVIDER_FAILURE_SUMMARY_MAX_CHARS { + return summary.to_string(); + } + let mut truncated = summary + .chars() + .take(PROVIDER_FAILURE_SUMMARY_MAX_CHARS) + .collect::(); + truncated.push_str("..."); + truncated +} + +fn provider_json_error_message(error: &str) -> Option { + let start = error.find('{')?; + let json = serde_json::Deserializer::from_str(&error[start..]) + .into_iter::() + .next()? + .ok()?; + let message = json + .get("error") + .and_then(|error| error.get("message").or(Some(error))) + .or_else(|| json.get("message")) + .and_then(serde_json::Value::as_str)? + .trim(); + (!message.is_empty()).then(|| message.to_string()) +} + fn acp_stop_reason(reason: Option) -> StopReason { match reason { Some(crate::llm::TurnStopReason::MaxTokens) => StopReason::MaxTokens, @@ -4809,6 +4948,183 @@ mod tests { } } + #[tokio::test(flavor = "current_thread")] + async fn provider_failure_without_tools_keeps_acp_session_usable() { + provider_failure_keeps_session_usable(false).await; + } + + #[tokio::test(flavor = "current_thread")] + async fn provider_failure_marks_pending_tools_failed_and_keeps_acp_session_usable() { + provider_failure_keeps_session_usable(true).await; + } + + #[test] + fn provider_failure_summary_keeps_only_the_provider_message() { + let step = "provider_step_error step=1 messages=2 tools=20 message_roles[system=1,user=1,assistant=0] tool_names=[bash,read,+18] tool_schema_bytes=9000 error="; + for (diagnostic, expected) in [ + ( + format!("{step}Provider error: API error 400 Bad Request: {{\"error\":{{\"message\":\"No tool output found for function call call_119.\",\"type\":\"invalid_request_error\"}}}}"), + "No tool output found for function call call_119.", + ), + ( + format!("{step}Provider error: OpenAI API error: status=400 Bad Request url=https://api.openai.com/v1/responses body={{\"error\":{{\"message\":\"Bad input\"}}}}"), + "Bad input", + ), + ( + format!("{step}Retryable provider error: API error 429 Too Many Requests: {{\"error\":{{\"message\":\"Slow down\"}}}} retries_exhausted=3"), + "Slow down", + ), + // Truncated bodies are not JSON; the raw body and URL still stay out. + ( + format!("{step}Provider error: OpenAI API error: status=400 Bad Request url=https://api.openai.com/v1/responses body={{\"error\":{{\"mess"), + "OpenAI API error: status=400 Bad Request", + ), + ( + "Streaming failed: connection reset by peer\nat transport layer".to_string(), + "connection reset by peer", + ), + (" ".to_string(), "Provider request failed"), + ] { + assert_eq!(provider_failure_summary(&diagnostic), expected, "{diagnostic}"); + } + + let long = provider_failure_summary(&"x".repeat(1_000)); + assert_eq!(long.chars().count(), PROVIDER_FAILURE_SUMMARY_MAX_CHARS + 3); + assert!(long.ends_with("...")); + } + + async fn provider_failure_keeps_session_usable(with_tools: bool) { + use agent_client_protocol::Agent; + use std::time::Duration; + + tokio::task::LocalSet::new().run_until(async { + let workspace = tempfile::tempdir().unwrap(); + let mut manager = SessionManager::new() + .with_history_for_workspace(workspace.path()).unwrap(); + let session_id = manager.create_session(Some("ACP provider failure regression".into())); + let db_id = manager.get_db_id(&session_id).unwrap(); + let service = AcpService { + sessions: Arc::new(AsyncMutex::new(AcpSessions::default())), + session_manager: Arc::new(Mutex::new(manager)), + client_capabilities: Arc::new(Mutex::new(Default::default())), + }; + let session = session_with_config(empty_config()); + service.sessions.lock().await.attachments.insert(session_id.clone(), session.clone()); + let (observed_tx, mut observed_rx) = tokio::sync::mpsc::unbounded_channel(); + let client = Client.builder().on_receive_notification( + async move |notification: SessionNotification, _connection| { + let _ = observed_tx.send(serde_json::to_value(notification).unwrap()); + Ok(()) + }, + agent_client_protocol::on_receive_notification!(), + ); + Agent.builder().connect_with(client, async move |connection| { + let cancellation = service.sessions.lock().await.begin_prompt(&session_id).unwrap(); + let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); + // What the aisdk step loop actually reports: step stats, tool names, raw body. + let diagnostic = concat!( + "provider_step_error step=1 messages=2 tools=20 ", + "message_roles[system=1,user=1,assistant=0] message_text_bytes=4096 images=0 ", + "tool_names=[bash,read,edit,+17] tool_schema_bytes=9000 tool_description_bytes=7000 ", + "error=Provider error: API error 400 Bad Request: {\"error\":{\"message\":", + "\"No tool output found for function call call_119.\",", + "\"type\":\"invalid_request_error\",\"param\":\"input\",\"code\":null}}", + ); + let error = "No tool output found for function call call_119."; + let leaks_diagnostic = |text: &str| { + ["provider_step_error", "tool_names", "invalid_request_error"] + .iter() + .any(|marker| text.contains(marker)) + }; + if with_tools { + sender.send(crate::llm::ChunkMessage::Text("Inspecting files.".into())).unwrap(); + let calls = ["completed-tool", "call_119"].into_iter().map(|id| crate::llm::ToolCall { + id: id.into(), + call_type: "function".into(), + function: crate::llm::tool_calls::FunctionCall { + name: "read".into(), arguments: "{\"file_path\":\"src/lib.rs\"}".into(), + }, + }).collect(); + sender.send(crate::llm::ChunkMessage::ToolCalls(calls)).unwrap(); + sender.send(crate::llm::ChunkMessage::ToolResult(crate::llm::ToolCallResult { + tool_call_id: "completed-tool".into(), role: "tool".into(), name: "read".into(), + content: serde_json::json!({"status":"ok", "output_preview":"file contents"}).to_string(), + })).unwrap(); + } + sender.send(crate::llm::ChunkMessage::Failed(diagnostic.into())).unwrap(); + sender.send(crate::llm::ChunkMessage::End).unwrap(); + let turn = service.receive_prompt_stream( + &session_id, &session, &connection, &cancellation, receiver, 0, 0, 0.0, + ).await.expect("provider errors must not become ACP RPC errors"); + assert_eq!(turn.response.stop_reason, StopReason::EndTurn); + service.sessions.lock().await.finish_prompt(&session_id); + + let mut saw_failed_tool = false; + let mut saw_error_message = false; + tokio::time::timeout(Duration::from_secs(2), async { + while let Some(notification) = observed_rx.recv().await { + let update = ¬ification["update"]; + assert!(!leaks_diagnostic(&update.to_string()), "client got the diagnostic: {update}"); + if update["sessionUpdate"] == "tool_call_update" && update["toolCallId"] == "call_119" { + assert_eq!(update["status"], "failed"); + saw_failed_tool = true; + } + if update["sessionUpdate"] == "agent_message_chunk" + && update["content"]["text"].as_str().is_some_and(|text| text.contains(error)) { + assert_eq!( + update["content"]["text"], + format!("\n\nThe AI request failed: {error}\nYou can retry or send another message in this session.") + ); + saw_error_message = true; + break; + } + } + }).await.unwrap(); + assert!(saw_error_message, "client must receive the provider error in-band"); + assert_eq!(saw_failed_tool, with_tools); + { + let manager = service.session_manager.lock().unwrap(); + let stored = manager.get_session_ref(&session_id).unwrap(); + assert_eq!(stored.status, crate::session::types::SessionStatus::Idle); + let assistant = stored.messages.last().unwrap(); + assert!(assistant.is_complete); + assert!(!assistant.was_interrupted); + assert!(!assistant.has_running_tool_parts()); + assert!(assistant.content.contains(error)); + assert!(!leaks_diagnostic(&assistant.content)); + if with_tools { + assert_eq!(assistant.tool_result_part_data("completed-tool").unwrap()["status"], "ok"); + assert_eq!(assistant.tool_result_part_data("call_119").unwrap()["status"], "error"); + assert_eq!(assistant.tool_result_part_data("call_119").unwrap()["output_preview"], error); + } + } + let dao = crate::persistence::HistoryDAO::new_for_workspace(workspace.path()).unwrap(); + assert_eq!(dao.get_session(db_id).unwrap().unwrap().status, "idle"); + let messages = dao.get_messages(db_id).unwrap(); + assert_eq!(messages.len(), 1); + assert!(messages[0].parts.iter().any(|part| part.part_type == "text" + && part.data["text"].as_str().is_some_and(|text| text.contains(error)))); + // Persisted parts are what the next request replays to the model. + for part in &messages[0].parts { + assert!(!leaks_diagnostic(&part.data.to_string()), "persisted diagnostic: {}", part.data); + } + + let cancellation = service.sessions.lock().await.begin_prompt(&session_id).unwrap(); + let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); + sender.send(crate::llm::ChunkMessage::Text("The next turn works.".into())).unwrap(); + sender.send(crate::llm::ChunkMessage::End).unwrap(); + let turn = service.receive_prompt_stream( + &session_id, &session, &connection, &cancellation, receiver, 0, 0, 0.0, + ).await.unwrap(); + assert_eq!(turn.response.stop_reason, StopReason::EndTurn); + service.sessions.lock().await.finish_prompt(&session_id); + assert_eq!(service.session_manager.lock().unwrap().get_session_ref(&session_id).unwrap().messages.len(), 2); + service.session_manager.lock().unwrap().try_delete_session(&session_id).unwrap(); + Ok(()) + }).await.unwrap(); + }).await; + } + #[tokio::test(flavor = "current_thread")] async fn cancelled_stream_persists_partial_transcript_and_interrupted_status() { cancelled_stream_finalizes(false, false).await; @@ -4921,9 +5237,9 @@ mod tests { } else { cancellation.cancel(); } - let response = tokio::time::timeout(Duration::from_secs(1), &mut stream) + let turn = tokio::time::timeout(Duration::from_secs(1), &mut stream) .await.unwrap().unwrap(); - assert_eq!(response.stop_reason, StopReason::Cancelled); + assert_eq!(turn.response.stop_reason, StopReason::Cancelled); if pending_permission { assert!(matches!(response_rx.await.unwrap(), crate::tools::PermissionResponse::Deny)); } diff --git a/src/aisdk/providers/openai.rs b/src/aisdk/providers/openai.rs index afd64ad..2a509b8 100644 --- a/src/aisdk/providers/openai.rs +++ b/src/aisdk/providers/openai.rs @@ -7,7 +7,7 @@ use crate::tool::Tool; use async_trait::async_trait; use eventsource_stream::{EventStreamError, Eventsource}; use futures::{SinkExt, StreamExt}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::error::Error as StdError; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; @@ -2359,50 +2359,59 @@ fn build_openai_messages( /// reuses an id fails with "Duplicate function_call_output". Remap each /// repeated call — and its matching output — to a unique id; ids are opaque. fn dedupe_responses_call_ids(mut items: Vec) -> Vec { + let mut reserved_ids: HashSet = items + .iter() + .filter_map(|item| item.get("call_id").and_then(|value| value.as_str())) + .map(str::to_string) + .collect(); + let mut replacements: HashMap<(String, usize), String> = HashMap::new(); // Occurrence k of a call_id pairs with occurrence k of its output, so // track counts per call_id instead of assuming strict call/output order. let mut call_occurrences: HashMap = HashMap::new(); let mut output_occurrences: HashMap = HashMap::new(); for item in items.iter_mut() { - match item.get("type").and_then(|value| value.as_str()) { - Some("function_call") => { - let Some(call_id) = item - .get("call_id") - .and_then(|value| value.as_str()) - .map(str::to_string) - else { - continue; - }; - let occurrence = call_occurrences.entry(call_id.clone()).or_insert(0); - *occurrence += 1; - if *occurrence == 1 { - continue; - } - let new_id = format!("{call_id}_dup{occurrence}"); - item["call_id"] = serde_json::Value::String(new_id); - // The item id belongs to the original response; drop it so the - // rewritten pair cannot collide on item ids either. - if let Some(obj) = item.as_object_mut() { - obj.remove("id"); - } - } - Some("function_call_output") => { - let Some(call_id) = item - .get("call_id") - .and_then(|value| value.as_str()) - .map(str::to_string) - else { - continue; - }; - let occurrence = output_occurrences.entry(call_id.clone()).or_insert(0); - *occurrence += 1; - if *occurrence > 1 { - item["call_id"] = - serde_json::Value::String(format!("{call_id}_dup{occurrence}")); + let is_call = match item.get("type").and_then(|value| value.as_str()) { + Some("function_call") => true, + Some("function_call_output") => false, + _ => continue, + }; + let Some(call_id) = item + .get("call_id") + .and_then(|value| value.as_str()) + .map(str::to_string) + else { + continue; + }; + let occurrences = if is_call { + &mut call_occurrences + } else { + &mut output_occurrences + }; + let occurrence = occurrences.entry(call_id.clone()).or_insert(0); + *occurrence += 1; + if *occurrence == 1 { + continue; + } + let new_id = replacements + .entry((call_id.clone(), *occurrence)) + .or_insert_with(|| { + let mut suffix = *occurrence; + loop { + let candidate = format!("{call_id}_dup{suffix}"); + if reserved_ids.insert(candidate.clone()) { + break candidate; + } + suffix += 1; } + }); + item["call_id"] = serde_json::Value::String(new_id.clone()); + // The item id belongs to the original response; drop it so the + // rewritten pair cannot collide on item ids either. + if is_call { + if let Some(obj) = item.as_object_mut() { + obj.remove("id"); } - _ => {} } } @@ -3029,6 +3038,50 @@ mod tests { assert!(input[3].get("id").is_none()); } + #[test] + fn dedupes_responses_call_ids_without_colliding_with_original_suffixes() { + let input = build_openai_messages( + &[ + Message::tool_call("call_1", "read", "{}"), + Message::tool_output("call_1", "read", "first", false), + Message::tool_call_with_item_id("fc_repeated", "call_1", "read", "{}"), + Message::tool_call_with_item_id("fc_original", "call_1_dup2", "glob", "{}"), + Message::tool_call("call_2", "grep", "{}"), + Message::tool_output("call_2", "grep", "concurrent third", false), + Message::tool_output("call_1_dup2", "glob", "concurrent second", false), + Message::tool_output("call_1", "read", "concurrent first", false), + Message::tool_call("call_1", "read", "{}"), + Message::tool_output("call_1", "read", "last", false), + ], + false, + false, + ); + + let call_ids: Vec<&str> = input + .iter() + .map(|item| item["call_id"].as_str().expect("call_id")) + .collect(); + assert_eq!( + call_ids, + [ + "call_1", + "call_1", + "call_1_dup3", + "call_1_dup2", + "call_2", + "call_2", + "call_1_dup2", + "call_1_dup3", + "call_1_dup4", + "call_1_dup4", + ] + ); + assert!(input[2].get("id").is_none()); + assert_eq!(input[3]["id"], "fc_original"); + assert_eq!(input[6]["output"], "concurrent second"); + assert_eq!(input[7]["output"], "concurrent first"); + } + #[test] fn serializes_tool_image_output_for_responses_input() { let input = build_openai_messages( diff --git a/tests/acp_stdio.rs b/tests/acp_stdio.rs index 26a17a2..1ba4cb6 100644 --- a/tests/acp_stdio.rs +++ b/tests/acp_stdio.rs @@ -769,3 +769,182 @@ fn automatically_compacts_provider_usage_over_stdio() { process.close_and_wait(); provider.finish(); } + +/// Minimal OpenAI-compatible endpoint: the first chat request fails with the +/// Responses API tool-output mismatch, later ones stream `reply`. +struct MockProvider { + base_url: String, + requests: std::sync::Arc>>, +} + +impl MockProvider { + fn start(reply: &'static str) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind mock provider"); + let base_url = format!("http://{}/v1", listener.local_addr().unwrap()); + let requests = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let recorded = requests.clone(); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { break }; + let Some((request_line, body)) = read_http_request(&mut stream) else { + continue; + }; + let response = if !request_line.starts_with("POST ") { + http_response("404 Not Found", "application/json", "{}") + } else { + let mut requests = recorded.lock().unwrap(); + requests.push(body); + if requests.len() == 1 { + http_response( + "400 Bad Request", + "application/json", + r#"{"error":{"message":"No tool output found for function call call_119.","type":"invalid_request_error","param":"input","code":null}}"#, + ) + } else { + let delta = serde_json::json!({ + "choices": [{ "index": 0, "delta": { "content": reply }, "finish_reason": null }] + }); + let stop = serde_json::json!({ + "choices": [{ "index": 0, "delta": {}, "finish_reason": "stop" }] + }); + http_response( + "200 OK", + "text/event-stream", + &format!("data: {delta}\n\ndata: {stop}\n\ndata: [DONE]\n\n"), + ) + } + }; + let _ = stream.write_all(response.as_bytes()); + } + }); + Self { base_url, requests } + } + + fn requests(&self) -> Vec { + self.requests.lock().unwrap().clone() + } +} + +fn read_http_request(stream: &mut TcpStream) -> Option<(String, String)> { + let mut reader = BufReader::new(stream.try_clone().ok()?); + let mut request_line = String::new(); + reader.read_line(&mut request_line).ok()?; + let mut content_length = 0; + loop { + let mut header = String::new(); + reader.read_line(&mut header).ok()?; + if header.trim().is_empty() { + break; + } + if let Some((name, value)) = header.split_once(':') { + if name.eq_ignore_ascii_case("content-length") { + content_length = value.trim().parse().ok()?; + } + } + } + let mut body = vec![0; content_length]; + reader.read_exact(&mut body).ok()?; + Some((request_line, String::from_utf8_lossy(&body).into_owned())) +} + +fn http_response(status: &str, content_type: &str, body: &str) -> String { + format!( + "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", + body.len() + ) +} + +fn agent_text(notifications: &[serde_json::Value]) -> String { + notifications + .iter() + .filter(|message| message["params"]["update"]["sessionUpdate"] == "agent_message_chunk") + .filter_map(|message| message["params"]["update"]["content"]["text"].as_str()) + .collect() +} + +#[test] +fn provider_failure_reports_short_error_skips_compaction_and_recovers_over_stdio() { + let provider = MockProvider::start("Recovered: the next turn works."); + let workspace = tempfile::tempdir().expect("workspace"); + // A small context window puts the session over the auto-compaction threshold + // as soon as the first turn is stored. + std::fs::write( + workspace.path().join("crabcode.json"), + serde_json::json!({ + "model": "mock/mock-model", + "enabled_providers": ["mock"], + "provider": { + "mock": { + "npm": "@ai-sdk/openai-compatible", + "options": { "baseURL": provider.base_url, "apiKey": "test-key" }, + "models": { + "mock-model": { "tool_call": true, "limit": { "context": 4000, "output": 1000 } } + } + } + } + }) + .to_string(), + ) + .expect("workspace config"); + // Request-time provider discovery also reads the process working directory. + let mut process = AcpProcess::spawn_with_working_directory(workspace.path(), workspace.path()); + initialize(&mut process); + process.send(serde_json::json!({ + "jsonrpc": "2.0", + "id": 2, + "method": "session/new", + "params": { "cwd": workspace.path(), "mcpServers": [] } + })); + let (new_session, _) = process.recv_response(2); + let session_id = new_session["result"]["sessionId"] + .as_str() + .unwrap_or_else(|| panic!("session id in {new_session}")) + .to_string(); + + let large_prompt = "Please review this context. ".repeat(600); + process.send(serde_json::json!({ + "jsonrpc": "2.0", + "id": 3, + "method": "session/prompt", + "params": { "sessionId": session_id, "prompt": [{ "type": "text", "text": large_prompt }] } + })); + let (failed, notifications) = process.recv_response(3); + assert_eq!(failed["result"]["stopReason"], "end_turn", "{failed}"); + assert_eq!( + agent_text(¬ifications), + "\n\nThe AI request failed: No tool output found for function call call_119.\nYou can retry or send another message in this session." + ); + // Auto-compaction would have sent a summary request to the provider that just failed. + assert_eq!( + provider.requests().len(), + 1, + "no compaction after a failed turn" + ); + + process.send(serde_json::json!({ + "jsonrpc": "2.0", + "id": 4, + "method": "session/prompt", + "params": { "sessionId": session_id, "prompt": [{ "type": "text", "text": "Try again" }] } + })); + let (recovered, notifications) = process.recv_response(4); + assert_eq!(recovered["result"]["stopReason"], "end_turn", "{recovered}"); + assert!(agent_text(¬ifications).contains("Recovered: the next turn works.")); + // The next turn compacts first (the session is over the threshold), then sends the + // turn itself. The failed turn is replayed only as the short note, never the diagnostic. + let requests = provider.requests(); + assert!(requests.len() >= 2); + assert!(requests[1..].iter().any(|body| body + .contains("The AI request failed: No tool output found for function call call_119."))); + for body in &requests[1..] { + for diagnostic in [ + "provider_step_error", + "tool_names=", + "invalid_request_error", + ] { + assert!(!body.contains(diagnostic), "replayed {diagnostic}: {body}"); + } + } + + process.close_and_wait(); +}