From 32722e6ef3ba7c5e7d88a32b02633b93e159718a Mon Sep 17 00:00:00 2001 From: Kalvin C Date: Wed, 10 Jun 2026 15:40:27 -0700 Subject: [PATCH] feat: surface Anthropic stream refusals as visible errors (#9724) Signed-off-by: Kalvin Chau --- crates/goose-providers/src/errors.rs | 15 +++ crates/goose/src/agents/agent.rs | 124 ++++++++++++++++-- crates/goose/src/providers/anthropic.rs | 2 +- crates/goose/src/providers/databricks_v2.rs | 2 +- .../goose/src/providers/formats/anthropic.rs | 124 ++++++++++++++++-- crates/goose/src/providers/gcpvertexai.rs | 3 +- crates/goose/src/providers/kimicode.rs | 4 +- 7 files changed, 246 insertions(+), 28 deletions(-) diff --git a/crates/goose-providers/src/errors.rs b/crates/goose-providers/src/errors.rs index 46a2aa4ca..c5f60dd78 100644 --- a/crates/goose-providers/src/errors.rs +++ b/crates/goose-providers/src/errors.rs @@ -42,6 +42,12 @@ pub enum ProviderError { details: String, top_up_url: Option, }, + + #[error("Provider refused request: {details}")] + Refusal { + details: String, + category: Option, + }, } impl ProviderError { @@ -58,12 +64,21 @@ impl ProviderError { ProviderError::NotImplemented(_) => "not_implemented", ProviderError::EndpointNotFound(_) => "endpoint_not_found", ProviderError::CreditsExhausted { .. } => "credits_exhausted", + ProviderError::Refusal { .. } => "refusal", } } pub fn is_endpoint_not_found(&self) -> bool { matches!(self, ProviderError::EndpointNotFound(_)) } + + /// Recover a typed `ProviderError` from a streaming decode error, falling + /// back to `RequestFailed` for errors that did not originate as one. + pub fn from_stream_error(error: anyhow::Error) -> Self { + error + .downcast() + .unwrap_or_else(|e| ProviderError::RequestFailed(format!("Stream decode error: {e}"))) + } } fn is_network_error(err: &reqwest::Error) -> bool { diff --git a/crates/goose/src/agents/agent.rs b/crates/goose/src/agents/agent.rs index 07d34995b..540fe9edb 100644 --- a/crates/goose/src/agents/agent.rs +++ b/crates/goose/src/agents/agent.rs @@ -2244,6 +2244,21 @@ impl Agent { ); break; } + Err(ref provider_err @ ProviderError::Refusal { ref details, ref category }) => { + #[cfg(feature = "telemetry")] + crate::posthog::emit_error(provider_err.telemetry_type(), &provider_err.to_string()); + error!("Error: {}", provider_err); + + let category = category.as_deref().map(|c| format!("\n\nCategory: {c}")).unwrap_or_default(); + yield AgentEvent::Message(Message::assistant().with_text(format!( + "The provider refused this request.\n\n{details}{category}\n\nPlease start a new session to continue — resending this conversation is likely to be refused again." + ))); + // A refusal is terminal: skip goal/grind nudges and + // recipe retry_config, which would resend the same + // refused conversation. + exit_chat = true; + break; + } Err(ref provider_err @ ProviderError::NetworkError(_)) => { #[cfg(feature = "telemetry")] crate::posthog::emit_error(provider_err.telemetry_type(), &provider_err.to_string()); @@ -2287,7 +2302,7 @@ impl Agent { } } - if no_tools_called { + if no_tools_called && !exit_chat { // Lock, extract state, drop guard before branching — handle_retry_logic // also locks final_output_tool and tokio::sync::Mutex is not reentrant. let final_output = { @@ -3279,12 +3294,86 @@ exit 0 } } - async fn create_stop_hook_test_agent( - env: &StopHookTestEnv, - stop_hook_block_cap: u32, - ) -> Result<(Agent, String, Arc)> { - let session_manager = Arc::new(SessionManager::new(env.data_dir())); - let permission_manager = Arc::new(PermissionManager::new(env.data_dir())); + struct RefusingProvider { + call_count: AtomicUsize, + } + + #[async_trait::async_trait] + impl crate::providers::base::Provider for RefusingProvider { + async fn stream( + &self, + _model_config: &crate::model::ModelConfig, + _session_id: &str, + _system_prompt: &str, + _messages: &[Message], + _tools: &[Tool], + ) -> Result { + self.call_count.fetch_add(1, Ordering::SeqCst); + Ok(Box::pin(futures::stream::once(async { + Err(ProviderError::Refusal { + details: "This request was declined.".to_string(), + category: Some("cyber".to_string()), + }) + }))) + } + + fn get_model_config(&self) -> crate::model::ModelConfig { + crate::model::ModelConfig::new("mock-model").unwrap() + } + + fn get_name(&self) -> &str { + "refusing" + } + } + + #[tokio::test] + async fn refusal_exits_turn_without_recipe_retry() -> Result<()> { + let temp_dir = tempfile::tempdir()?; + let provider = Arc::new(RefusingProvider { + call_count: AtomicUsize::new(0), + }); + let hook_manager = crate::hooks::HookManager::from_plugins_for_test(vec![]); + let (agent, session_id) = + create_test_agent(temp_dir.path().join("data"), hook_manager, provider.clone()).await?; + + let session_config = SessionConfig { + id: session_id, + schedule_id: None, + max_turns: Some(10), + retry_config: Some(crate::agents::types::RetryConfig { + max_retries: 3, + checks: vec![crate::agents::types::SuccessCheck::Shell { + command: "false".to_string(), + }], + on_failure: None, + timeout_seconds: None, + on_failure_timeout_seconds: None, + }), + }; + + let reply_stream = agent + .reply(Message::user().with_text("hi"), session_config, None) + .await?; + tokio::pin!(reply_stream); + while let Some(event) = reply_stream.next().await { + event?; + } + + assert_eq!( + provider.call_count.load(Ordering::SeqCst), + 1, + "a refused request must not be resent" + ); + Ok(()) + } + + async fn create_test_agent( + data_dir: PathBuf, + hook_manager: crate::hooks::HookManager, + provider: Arc, + ) -> Result<(Agent, String)> { + let session_manager = Arc::new(SessionManager::new(data_dir.clone())); + let permission_manager = Arc::new(PermissionManager::new(data_dir)); let config = AgentConfig::new( session_manager.clone(), permission_manager, @@ -3294,19 +3383,28 @@ exit 0 GoosePlatform::GooseCli, ); let mut agent = Agent::with_config(config); - agent.set_hook_manager_for_test(env.hook_manager()); - agent.set_stop_hook_block_cap_for_test(stop_hook_block_cap); - let provider = Arc::new(CountingTextProvider::new()); + agent.set_hook_manager_for_test(hook_manager); let session = session_manager .create_session( PathBuf::default(), - "stop-hook-test".to_string(), + "test".to_string(), SessionType::Hidden, GooseMode::Auto, ) .await?; - agent.update_provider(provider.clone(), &session.id).await?; - Ok((agent, session.id, provider)) + agent.update_provider(provider, &session.id).await?; + Ok((agent, session.id)) + } + + async fn create_stop_hook_test_agent( + env: &StopHookTestEnv, + stop_hook_block_cap: u32, + ) -> Result<(Agent, String, Arc)> { + let provider = Arc::new(CountingTextProvider::new()); + let (mut agent, session_id) = + create_test_agent(env.data_dir(), env.hook_manager(), provider.clone()).await?; + agent.set_stop_hook_block_cap_for_test(stop_hook_block_cap); + Ok((agent, session_id, provider)) } async fn run_stop_hook_test_turn( diff --git a/crates/goose/src/providers/anthropic.rs b/crates/goose/src/providers/anthropic.rs index 37f129057..0a659fee3 100644 --- a/crates/goose/src/providers/anthropic.rs +++ b/crates/goose/src/providers/anthropic.rs @@ -379,7 +379,7 @@ impl Provider for AnthropicProvider { let message_stream = response_to_streaming_message(framed); pin!(message_stream); while let Some(message) = futures::StreamExt::next(&mut message_stream).await { - let (message, usage) = message.map_err(|e| ProviderError::RequestFailed(format!("Stream decode error: {}", e)))?; + let (message, usage) = message.map_err(ProviderError::from_stream_error)?; log.write(&message, usage.as_ref().map(|f| f.usage).as_ref())?; yield (message, usage); } diff --git a/crates/goose/src/providers/databricks_v2.rs b/crates/goose/src/providers/databricks_v2.rs index 15b715ef4..56a20e24b 100644 --- a/crates/goose/src/providers/databricks_v2.rs +++ b/crates/goose/src/providers/databricks_v2.rs @@ -328,7 +328,7 @@ impl DatabricksV2Provider { let message_stream = anthropic::response_to_streaming_message(framed); pin!(message_stream); while let Some(message) = futures::StreamExt::next(&mut message_stream).await { - let (message, usage) = message.map_err(|e| ProviderError::RequestFailed(format!("Stream decode error: {e}")))?; + let (message, usage) = message.map_err(ProviderError::from_stream_error)?; log.write(&message, usage.as_ref().map(|f| f.usage).as_ref())?; yield (message, usage); } diff --git a/crates/goose/src/providers/formats/anthropic.rs b/crates/goose/src/providers/formats/anthropic.rs index ac7b50675..b71788cf7 100644 --- a/crates/goose/src/providers/formats/anthropic.rs +++ b/crates/goose/src/providers/formats/anthropic.rs @@ -118,6 +118,8 @@ const EVENT_MESSAGE_STOP: &str = "message_stop"; const EVENT_CONTENT_BLOCK_START: &str = "content_block_start"; const EVENT_CONTENT_BLOCK_DELTA: &str = "content_block_delta"; const EVENT_CONTENT_BLOCK_STOP: &str = "content_block_stop"; +const STOP_REASON_REFUSAL: &str = "refusal"; +const REFUSAL_FALLBACK_DETAILS: &str = "No additional details were provided."; /// Coerce a tool call's optional arguments into the JSON value Anthropic /// expects for the `input` field of a `tool_use` content block. @@ -890,6 +892,33 @@ where final_usage = Some(ProviderUsage::new(model, delta_usage)); } } + if let Some(delta) = event.data.get("delta") { + let stop_details = delta.get("stop_details").filter(|d| !d.is_null()); + if delta.get("stop_reason").and_then(|v| v.as_str()) == Some(STOP_REASON_REFUSAL) { + let str_field = |key: &str| stop_details + .and_then(|d| d.get(key)) + .and_then(|v| v.as_str()) + .map(str::to_string); + let details = str_field("explanation") + .or_else(|| stop_details.map(|d| d.to_string())) + .unwrap_or_else(|| REFUSAL_FALLBACK_DETAILS.to_string()); + let category = str_field("category"); + // The refusal delta carries the request's usage; + // flush it so refused turns are still accounted. + if let Some(usage) = final_usage.take() { + yield (None, Some(usage)); + } + Err(ProviderError::Refusal { details, category })?; + } else if let Some(details) = stop_details { + // No specific handling for these stop details yet — + // forward them rather than silently dropping the turn. + let mut message = Message::assistant().with_text(format!( + "The provider ended the response with: {details}" + )); + message.id = message_id.clone(); + yield (Some(message), None); + } + } continue; } EVENT_MESSAGE_STOP => { @@ -1576,16 +1605,10 @@ mod tests { } async fn collect_stream(events: &str) -> StreamedParts { - use futures::StreamExt; - - let lines: Vec> = - events.lines().map(|l| Ok(l.to_string())).collect(); - let stream = Box::pin(futures::stream::iter(lines)); - let mut msg_stream = std::pin::pin!(response_to_streaming_message(stream)); let mut parts = StreamedParts::default(); - while let Some(Ok((message, _usage))) = msg_stream.next().await { - if let Some(msg) = message { + for result in collect_stream_results(events).await { + if let Ok((Some(msg), _usage)) = result { for c in &msg.content { match c { MessageContent::Thinking(t) => { @@ -1734,4 +1757,89 @@ mod tests { assert_eq!(parts.text, vec!["Let me search for that."]); assert_eq!(parts.tool_calls, vec!["search"]); } + + async fn collect_stream_results( + events: &str, + ) -> Vec, Option)>> { + use futures::StreamExt; + + let lines: Vec> = + events.lines().map(|l| Ok(l.to_string())).collect(); + let stream = Box::pin(futures::stream::iter(lines)); + response_to_streaming_message(stream).collect().await + } + + fn expect_refusal( + results: Vec, Option)>>, + ) -> (String, Option) { + let err = results + .into_iter() + .find_map(|r| r.err()) + .expect("refusal should surface as a stream error"); + match err.downcast_ref::() { + Some(ProviderError::Refusal { details, category }) => { + (details.clone(), category.clone()) + } + other => panic!("expected ProviderError::Refusal, got {:?}", other), + } + } + + #[tokio::test] + async fn test_streaming_refusal() { + let events = concat!( + r#"data: {"type":"message_start","message":{"id":"msg_1","role":"assistant","content":[],"model":"claude-opus-4-6","usage":{"input_tokens":10,"output_tokens":0}}}"#, + "\n", + r#"data: {"type":"message_delta","delta":{"stop_reason":"refusal","stop_details":{"explanation":"This request violates the usage policy.","category":"cyber"}},"usage":{"output_tokens":5}}"#, + "\n", + r#"data: {"type":"message_stop"}"#, + ); + + let results = collect_stream_results(events).await; + let usage = results + .iter() + .filter_map(|r| r.as_ref().ok()) + .find_map(|(_, usage)| usage.clone()) + .expect("a refused request should still yield its usage"); + assert_eq!(usage.usage.input_tokens, Some(10)); + assert_eq!(usage.usage.output_tokens, Some(5)); + + let (details, category) = expect_refusal(results); + assert_eq!(details, "This request violates the usage policy."); + assert_eq!(category.as_deref(), Some("cyber")); + } + + #[tokio::test] + async fn test_streaming_refusal_forwards_unrecognized_stop_details() { + let events = concat!( + r#"data: {"type":"message_start","message":{"id":"msg_1","role":"assistant","content":[],"model":"claude-opus-4-6","usage":{"input_tokens":10,"output_tokens":0}}}"#, + "\n", + r#"data: {"type":"message_delta","delta":{"stop_reason":"refusal","stop_details":{"code":42}},"usage":{"output_tokens":5}}"#, + "\n", + r#"data: {"type":"message_stop"}"#, + ); + + let (details, category) = expect_refusal(collect_stream_results(events).await); + assert!(details.contains("\"code\":42"), "details: {details}"); + assert_eq!(category, None); + } + + #[tokio::test] + async fn test_streaming_forwards_unhandled_stop_details() { + let events = concat!( + r#"data: {"type":"message_start","message":{"id":"msg_1","role":"assistant","content":[],"model":"claude-opus-4-6","usage":{"input_tokens":10,"output_tokens":0}}}"#, + "\n", + r#"data: {"type":"message_delta","delta":{"stop_reason":"model_context_window_exceeded","stop_details":{"reason":"context_window"}},"usage":{"output_tokens":5}}"#, + "\n", + r#"data: {"type":"message_stop"}"#, + ); + + let parts = collect_stream(events).await; + assert_eq!(parts.text.len(), 1); + assert!( + parts.text[0].starts_with("The provider ended the response with:"), + "text: {}", + parts.text[0] + ); + assert!(parts.text[0].contains("context_window")); + } } diff --git a/crates/goose/src/providers/gcpvertexai.rs b/crates/goose/src/providers/gcpvertexai.rs index 8f215b925..7aba01fa3 100644 --- a/crates/goose/src/providers/gcpvertexai.rs +++ b/crates/goose/src/providers/gcpvertexai.rs @@ -618,8 +618,7 @@ impl Provider for GcpVertexAIProvider { let mut message_stream = response_to_streaming_message(framed, &context_clone); while let Some(message) = message_stream.next().await { - let (message, usage) = message - .map_err(|e| ProviderError::RequestFailed(format!("Stream decode error: {}", e)))?; + let (message, usage) = message.map_err(ProviderError::from_stream_error)?; log.write(&message, usage.as_ref().map(|u| &u.usage))?; yield (message, usage); } diff --git a/crates/goose/src/providers/kimicode.rs b/crates/goose/src/providers/kimicode.rs index 5cb6e4275..5f00521ce 100644 --- a/crates/goose/src/providers/kimicode.rs +++ b/crates/goose/src/providers/kimicode.rs @@ -426,9 +426,7 @@ impl Provider for KimiCodeProvider { let message_stream = response_to_streaming_message(framed); pin!(message_stream); while let Some(message) = futures::StreamExt::next(&mut message_stream).await { - let (message, usage) = message.map_err(|e| { - ProviderError::RequestFailed(format!("Stream decode error: {}", e)) - })?; + let (message, usage) = message.map_err(ProviderError::from_stream_error)?; log.write(&message, usage.as_ref().map(|f| f.usage).as_ref())?; yield (message, usage); }