tkmind/tkmind_go
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

feat: surface Anthropic stream refusals as visible errors (#9724)

Browse Source 
Signed-off-by: Kalvin Chau <kalvin@block.xyz>
This commit is contained in:
Kalvin C
2026-06-10 15:40:27 -07:00
committed by GitHub
parent 58185fe4ae
commit 32722e6ef3
7 changed files with 246 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/goose-providers/src/errors.rs
+15
View File
@@ -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
+111 -13
View File
@@ -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<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,
@@ -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<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
+1 -1
View File
@@ -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
+1 -1
View File
@@ -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
+116 -8
View File
@@ -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
+1 -2
View File
@@ -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
+1 -3
View File
@@ -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);
}