feat: surface Anthropic stream refusals as visible errors

crates/goose-providers/src/errors.rs

@@ -42,6 +42,12 @@ pub enum ProviderError {
         details: String,
         top_up_url: Option<String>,
     },
+
+    #[error("Provider refused request: {details}")]
+    Refusal {
+        details: String,
+        category: Option<String>,
+    },
 }
 
 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 {

crates/goose/src/agents/agent.rs

@@ -2189,6 +2189,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());
@@ -2230,7 +2245,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 = {
@@ -3217,12 +3232,86 @@ exit 0
         }
     }
 
-    async fn create_stop_hook_test_agent(
-        env: &StopHookTestEnv,
-        stop_hook_block_cap: u32,
-    ) -> Result<(Agent, String, Arc<CountingTextProvider>)> {
-        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<MessageStream, ProviderError> {
+            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<dyn crate::providers::base::Provider>,
+    ) -> 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,
@@ -3232,19 +3321,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<CountingTextProvider>)> {
+        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(

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);
             }

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);
             }

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<Result<String, anyhow::Error>> =
-            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<anyhow::Result<(Option<Message>, Option<ProviderUsage>)>> {
+        use futures::StreamExt;
+
+        let lines: Vec<Result<String, anyhow::Error>> =
+            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<anyhow::Result<(Option<Message>, Option<ProviderUsage>)>>,
+    ) -> (String, Option<String>) {
+        let err = results
+            .into_iter()
+            .find_map(|r| r.err())
+            .expect("refusal should surface as a stream error");
+        match err.downcast_ref::<ProviderError>() {
+            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"));
+    }
 }

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);
             }

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);
             }