Harden provider diagnostics and tool failure handling

This commit is contained in:
Ryan Ward
2026-11-06 19:49:39 +00:00
parent 4b968f34f8
commit 38f1dcdef6
35 changed files with 1558 additions and 79 deletions
+1 -1
View File
@@ -10,6 +10,7 @@ async-stream.workspace = true
async-trait.workspace = true
aws-sdk-bedrockruntime.workspace = true
base64.workspace = true
bytes.workspace = true
futures.workspace = true
galaxy_agent_core.workspace = true
http.workspace = true
@@ -21,6 +22,5 @@ uuid.workspace = true
[dev-dependencies]
aws-smithy-http-client.workspace = true
bytes.workspace = true
rig-core = { workspace = true, features = ["test-utils"] }
tokio = { workspace = true, features = ["macros", "rt"] }
+11 -2
View File
@@ -13,6 +13,8 @@ use rig_core::providers::anthropic;
use rig_core::providers::anthropic::client::AnthropicExt;
use serde_json::json;
use crate::http_client::DiagnosticHttpClient;
pub fn is_claude_subscription_token(credential: &str) -> bool {
credential.trim().contains("sk-ant-oat")
}
@@ -64,15 +66,22 @@ impl ProviderBuilder for SubscriptionBuilder {
}
}
pub(crate) fn anthropic_client(credential: &str) -> Result<anthropic::Client, String> {
pub(crate) fn anthropic_client(
credential: &str,
) -> Result<anthropic::Client<DiagnosticHttpClient>, String> {
let credential = credential.trim();
if !is_claude_subscription_token(credential) {
return anthropic::Client::new(credential).map_err(|error| error.to_string());
return anthropic::Client::builder()
.api_key(credential)
.http_client(DiagnosticHttpClient::new())
.build()
.map_err(|error| error.to_string());
}
let headers = subscription_headers();
Client::<SubscriptionProvider>::builder()
.api_key(SubscriptionToken(credential.to_string()))
.http_headers(headers)
.http_client(DiagnosticHttpClient::new())
.build()
// Never include a credential or raw header-validation error in diagnostics.
.map_err(|_| "Could not configure Claude subscription authentication.".to_string())
+2
View File
@@ -11,6 +11,7 @@ use galaxy_agent_core::{
use rig_core::client::CompletionClient;
use rig_core::providers::chatgpt::{self, ChatGPTAuth};
use crate::http_client::DiagnosticHttpClient;
use crate::request::build_completion_request;
use crate::stream::start_model_turn;
@@ -155,6 +156,7 @@ impl AgentRuntime for ChatGPTSubscriptionRuntime {
.api_key(auth)
.allow_device_flow(false)
.originator("galaxy")
.http_client(DiagnosticHttpClient::new())
.build()
.map_err(|error| AgentError::new(AgentErrorKind::Configuration, error.to_string()))?;
let model = client.completion_model(&self.config.model);
+224
View File
@@ -0,0 +1,224 @@
//! Keep useful transport causes before Rig reduces SSE errors to their Display text.
use std::error::Error;
use std::fmt;
use bytes::Bytes;
use futures::StreamExt;
use rig_core::http_client::{
self, HttpClientExt, LazyBody, MultipartForm, Request, Response, StreamingResponse,
};
use rig_core::wasm_compat::WasmCompatSend;
#[derive(Clone, Debug, Default)]
pub(crate) struct DiagnosticHttpClient<H = http_client::ReqwestClient>(H);
impl DiagnosticHttpClient {
pub(crate) fn new() -> Self {
Self(http_client::ReqwestClient::default())
}
}
impl<H: HttpClientExt + Clone + 'static> HttpClientExt for DiagnosticHttpClient<H> {
fn send<T, U>(
&self,
request: Request<T>,
) -> impl Future<Output = http_client::Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
where
T: Into<Bytes> + WasmCompatSend,
U: From<Bytes> + WasmCompatSend + 'static,
{
let client = self.0.clone();
let request = request.map(Into::<Bytes>::into);
async move {
let response = client
.send(request)
.await
.map_err(diagnose_transport_error)?;
Ok(response.map(|body| {
Box::pin(async move { body.await.map_err(diagnose_transport_error) }) as LazyBody<U>
}))
}
}
fn send_multipart<U>(
&self,
request: Request<MultipartForm>,
) -> impl Future<Output = http_client::Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
where
U: From<Bytes> + WasmCompatSend + 'static,
{
let client = self.0.clone();
async move {
let response = client
.send_multipart(request)
.await
.map_err(diagnose_transport_error)?;
Ok(response.map(|body| {
Box::pin(async move { body.await.map_err(diagnose_transport_error) }) as LazyBody<U>
}))
}
}
fn send_streaming<T>(
&self,
request: Request<T>,
) -> impl Future<Output = http_client::Result<StreamingResponse>> + WasmCompatSend
where
T: Into<Bytes> + WasmCompatSend,
{
let client = self.0.clone();
let request = request.map(Into::<Bytes>::into);
async move {
let response = client
.send_streaming(request)
.await
.map_err(diagnose_transport_error)?;
Ok(response.map(|body| {
Box::pin(body.map(|chunk| chunk.map_err(diagnose_transport_error)))
as http_client::sse::BoxedStream
}))
}
}
}
// Only static categories and numeric OS codes leave this module. Transport errors
// can contain proxy passwords, URL userinfo/query keys, or arbitrary source text;
// neither Display nor Debug of the original error is safe to forward verbatim.
// HTTP status/body errors pass through unchanged so Rig's provider classification
// and its body-inspection helpers retain their original semantics.
pub(crate) fn diagnose_transport_error(error: http_client::Error) -> http_client::Error {
match error {
http_client::Error::Instance(source) => http_client::Error::Instance(Box::new(
TransportDiagnostic::from_source(source.as_ref()),
)),
other => other,
}
}
#[derive(Debug)]
struct TransportDiagnostic(String);
impl TransportDiagnostic {
fn from_source(source: &(dyn Error + 'static)) -> Self {
const MAX_DEPTH: usize = 12;
const MAX_CAUSES: usize = 4;
let mut causes = Vec::<String>::new();
let mut source = Some(source);
for _ in 0..MAX_DEPTH {
let Some(error) = source else { break };
let text = bounded_error_text(error)
.split_whitespace()
.filter(|word| !word.contains("://"))
.collect::<Vec<_>>()
.join(" ")
.to_ascii_lowercase();
if let Some(category) = transport_category(&text) {
let category = category.to_string();
if !causes.contains(&category) && causes.len() < MAX_CAUSES {
causes.push(category);
}
}
if let Some(error) = error.downcast_ref::<std::io::Error>() {
let detail = match error.raw_os_error() {
Some(code) => format!("I/O {:?} (OS error {code})", error.kind()),
None => format!("I/O {:?}", error.kind()),
};
if !causes.contains(&detail) && causes.len() < MAX_CAUSES {
causes.push(detail);
}
}
source = error.source();
}
if causes.is_empty() {
causes.push("underlying transport cause unavailable".to_string());
}
Self(format!(
"provider request transport failed: {}",
causes.join("; caused by: ")
))
}
}
impl fmt::Display for TransportDiagnostic {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(&self.0)
}
}
impl Error for TransportDiagnostic {}
fn bounded_error_text(error: &(dyn Error + 'static)) -> String {
struct BoundedText(String);
impl fmt::Write for BoundedText {
fn write_str(&mut self, value: &str) -> fmt::Result {
let remaining = 1024usize.saturating_sub(self.0.chars().count());
match value.char_indices().nth(remaining) {
Some((boundary, _)) => {
self.0.push_str(&value[..boundary]);
Err(fmt::Error)
}
None => {
self.0.push_str(value);
Ok(())
}
}
}
}
let mut text = BoundedText(String::new());
let _ = fmt::write(&mut text, format_args!("{error}"));
text.0
}
fn transport_category(text: &str) -> Option<&'static str> {
if text.contains("unknownissuer")
|| text.contains("unknown issuer")
|| text.contains("unknown ca")
{
Some("TLS certificate issuer is not trusted")
} else if text.contains("notvalidforname")
|| text.contains("certificate")
&& (text.contains("hostname") || text.contains("not valid for name"))
{
Some("TLS certificate hostname mismatch")
} else if text.contains("certificate")
&& (text.contains("expired")
|| text.contains("notvalidyet")
|| text.contains("not valid yet"))
{
Some("TLS certificate is expired or not yet valid")
} else if text.contains("certificate")
&& (text.contains("invalid") || text.contains("verify") || text.contains("validation"))
|| text.contains("cert verify")
{
Some("TLS certificate validation failed")
} else if text.contains("dns")
|| text.contains("failed to lookup address")
|| text.contains("name or service not known")
|| text.contains("nodename nor servname")
{
Some("DNS resolution failed")
} else if text.contains("proxy") && (text.contains("authentication") || text.contains("407")) {
Some("proxy authentication failed")
} else if text.contains("connection refused") {
Some("connection refused")
} else if text.contains("connection reset") {
Some("connection reset")
} else if text.contains("timed out") || text.contains("timeout") {
Some("network operation timed out")
} else if text.contains("unreachable") {
Some("network or destination unreachable")
} else if text.contains("tls") || text.contains("ssl") || text.contains("handshake") {
Some("TLS handshake failed")
} else if text.contains("unexpected eof")
|| text.contains("connection closed")
|| text.contains("incomplete message")
{
Some("connection closed before response completed")
} else {
None
}
}
#[cfg(test)]
#[path = "http_client_tests.rs"]
mod tests;
@@ -0,0 +1,253 @@
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use futures::executor::block_on;
use futures::{StreamExt, future, stream};
use http::StatusCode;
use super::*;
#[derive(Debug)]
struct SourceError {
message: &'static str,
source: Option<Box<SourceError>>,
}
impl fmt::Display for SourceError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(self.message)
}
}
impl Error for SourceError {
fn source(&self) -> Option<&(dyn Error + 'static)> {
self.source.as_deref().map(|source| source as &dyn Error)
}
}
fn transport_failure() -> http_client::Error {
http_client::Error::Instance(Box::new(SourceError {
message: "error sending request for url (https://user:password@example.test/v1?api_key=secret-certificate)",
source: Some(Box::new(SourceError {
message: "failed to lookup address information: Name or service not known; Authorization: Bearer secret-header",
source: None,
})),
}))
}
#[test]
fn nested_dns_cause_survives_without_url_credentials_queries_or_headers() {
let message = diagnose_transport_error(transport_failure()).to_string();
assert_eq!(
message,
"Http client error: provider request transport failed: DNS resolution failed"
);
}
#[test]
fn tls_certificate_cause_is_specific_and_contains_only_static_diagnostics() {
let error = http_client::Error::Instance(Box::new(SourceError {
message: "error sending request for url (https://example.test/?token=private)",
source: Some(Box::new(SourceError {
message: "invalid peer certificate: UnknownIssuer (private-certificate-content)",
source: None,
})),
}));
assert_eq!(
diagnose_transport_error(error).to_string(),
"Http client error: provider request transport failed: TLS certificate issuer is not trusted"
);
}
#[test]
fn unrecognized_causes_never_forward_arbitrary_source_text() {
let error = http_client::Error::Instance(Box::new(SourceError {
message: "Authorization: Bearer secret; api_key=private; password=hidden",
source: None,
}));
assert_eq!(
diagnose_transport_error(error).to_string(),
"Http client error: provider request transport failed: underlying transport cause unavailable"
);
}
#[test]
fn os_errors_retain_numeric_codes_without_the_original_message() {
let message = diagnose_transport_error(http_client::Error::Instance(Box::new(
std::io::Error::from_raw_os_error(111),
)))
.to_string();
assert_eq!(message.contains("OS error 111"), true);
assert_eq!(message.len() < 400, true);
}
#[test]
fn large_unicode_error_display_stops_before_formatting_later_chunks() {
#[derive(Debug)]
struct LargeError {
payload: String,
later_chunks: AtomicUsize,
}
impl fmt::Display for LargeError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.payload)?;
self.later_chunks.fetch_add(1, Ordering::Relaxed);
formatter.write_str("sensitive trailing data")
}
}
impl Error for LargeError {}
let error = LargeError {
payload: "🌌".repeat(100_000),
later_chunks: AtomicUsize::new(0),
};
assert_eq!(bounded_error_text(&error), "🌌".repeat(1024));
assert_eq!(error.later_chunks.load(Ordering::Relaxed), 0);
}
#[test]
fn http_status_and_provider_body_remain_available_for_classification() {
for status in [
StatusCode::UNAUTHORIZED,
StatusCode::TOO_MANY_REQUESTS,
StatusCode::SERVICE_UNAVAILABLE,
] {
let error = diagnose_transport_error(http_client::Error::InvalidStatusCodeWithMessage(
status,
"provider error body".to_owned(),
));
match error {
http_client::Error::InvalidStatusCodeWithMessage(actual_status, body) => {
assert_eq!(actual_status, status);
assert_eq!(body, "provider error body");
}
other => panic!("HTTP response changed error variant: {other:?}"),
}
}
}
#[derive(Clone, Debug)]
struct FakeHttpClient {
calls: Arc<AtomicUsize>,
fail_before_response: bool,
}
impl HttpClientExt for FakeHttpClient {
fn send<T, U>(
&self,
_request: Request<T>,
) -> impl Future<Output = http_client::Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
where
T: Into<Bytes> + WasmCompatSend,
U: From<Bytes> + WasmCompatSend + 'static,
{
self.calls.fetch_add(1, Ordering::Relaxed);
future::ready(if self.fail_before_response {
Err(transport_failure())
} else {
Ok(Response::new(
Box::pin(async { Err(transport_failure()) }) as LazyBody<U>
))
})
}
fn send_multipart<U>(
&self,
request: Request<MultipartForm>,
) -> impl Future<Output = http_client::Result<Response<LazyBody<U>>>> + WasmCompatSend + 'static
where
U: From<Bytes> + WasmCompatSend + 'static,
{
self.send(request.map(|_| Bytes::new()))
}
fn send_streaming<T>(
&self,
request: Request<T>,
) -> impl Future<Output = http_client::Result<StreamingResponse>> + WasmCompatSend
where
T: Into<Bytes> + WasmCompatSend,
{
self.calls.fetch_add(1, Ordering::Relaxed);
// The interceptor must forward the approved payload and credential header unchanged.
assert_eq!(request.headers()["authorization"], "Bearer synthetic-key");
assert_eq!(
request.into_body().into(),
Bytes::from_static(b"approved request")
);
future::ready(if self.fail_before_response {
Err(transport_failure())
} else {
let body = stream::iter([
Ok(Bytes::from_static(b"original SSE bytes")),
Err(transport_failure()),
]);
Ok(Response::builder()
.status(StatusCode::OK)
.header("x-request-id", "original-request-id")
.body(Box::pin(body) as http_client::sse::BoxedStream)
.unwrap())
})
}
}
fn request() -> Request<Bytes> {
Request::builder()
.uri("https://example.test/v1?secret=private")
.header("authorization", "Bearer synthetic-key")
.body(Bytes::from_static(b"approved request"))
.unwrap()
}
#[test]
fn streaming_adapter_preserves_response_and_chunks_without_internal_retries() {
block_on(async {
let calls = Arc::new(AtomicUsize::new(0));
let client = DiagnosticHttpClient(FakeHttpClient {
calls: calls.clone(),
fail_before_response: false,
});
let response = client.send_streaming(request()).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(response.headers()["x-request-id"], "original-request-id");
let mut body = response.into_body();
assert_eq!(
body.next().await.unwrap().unwrap(),
Bytes::from_static(b"original SSE bytes")
);
assert_eq!(
body.next()
.await
.unwrap()
.unwrap_err()
.to_string()
.contains("DNS resolution failed"),
true
);
assert_eq!(body.next().await.is_none(), true);
assert_eq!(calls.load(Ordering::Relaxed), 1);
});
}
#[test]
fn adapter_preserves_causes_before_headers_and_during_lazy_body_reads() {
block_on(async {
let calls = Arc::new(AtomicUsize::new(0));
let client = DiagnosticHttpClient(FakeHttpClient {
calls: calls.clone(),
fail_before_response: true,
});
let error = client.send_streaming(request()).await.err().unwrap();
assert_eq!(error.to_string().contains("DNS resolution failed"), true);
assert_eq!(calls.load(Ordering::Relaxed), 1);
let client = DiagnosticHttpClient(FakeHttpClient {
calls: calls.clone(),
fail_before_response: false,
});
let response = client.send::<_, Bytes>(request()).await.unwrap();
let error = response.into_body().await.unwrap_err();
assert_eq!(error.to_string().contains("DNS resolution failed"), true);
assert_eq!(calls.load(Ordering::Relaxed), 2);
});
}
+1
View File
@@ -3,6 +3,7 @@
mod anthropic_auth;
mod bedrock;
mod chatgpt;
mod http_client;
mod native;
mod openai_compatible;
mod request;
+10 -2
View File
@@ -13,6 +13,7 @@ use rig_core::http_client::{self, HttpClientExt};
use rig_core::providers::gemini;
use crate::anthropic_auth::{anthropic_client, prepare_subscription_request, restore_tool_names};
use crate::http_client::DiagnosticHttpClient;
use crate::request::build_completion_request;
use crate::stream::{start_model_completion, start_model_turn};
@@ -124,7 +125,11 @@ fn anthropic_reasoning_efforts(entry: &serde_json::Value) -> Vec<String> {
}
pub async fn discover_gemini_models(api_key: &str) -> Result<Vec<RigModelInfo>, String> {
let client = gemini::Client::new(api_key).map_err(|error| error.to_string())?;
let client = gemini::Client::builder()
.api_key(api_key)
.http_client(DiagnosticHttpClient::new())
.build()
.map_err(|error| error.to_string())?;
let models = client
.list_models()
.await
@@ -349,7 +354,10 @@ impl AgentRuntime for GeminiRuntime {
"A Gemini API key is required",
));
}
let client = gemini::Client::new(&self.config.api_key)
let client = gemini::Client::builder()
.api_key(self.config.api_key.as_str())
.http_client(DiagnosticHttpClient::new())
.build()
.map_err(|error| AgentError::new(AgentErrorKind::Configuration, error.to_string()))?;
let model = client.completion_model(&self.config.model);
let max_output_tokens = request.max_output_tokens.or(self.config.max_output_tokens);
@@ -7,6 +7,7 @@ use rig_core::client::CompletionClient;
use rig_core::completion::{CompletionModel, CompletionRequest};
use rig_core::providers::openai;
use crate::http_client::DiagnosticHttpClient;
use crate::request::build_completion_request as build_provider_completion_request;
use crate::stream::start_model_turn as start_provider_model_turn;
@@ -63,6 +64,7 @@ impl AgentRuntime for OpenAICompatibleRuntime {
// compatibility with unauthenticated local OpenAI-compatible servers.
.api_key(self.config.api_key.as_deref().unwrap_or_default())
.base_url(&self.config.base_url)
.http_client(DiagnosticHttpClient::new())
.build()
.map_err(|error| AgentError::new(AgentErrorKind::Configuration, error.to_string()))?;
let model = client.completion_model(&self.config.model);
+2 -2
View File
@@ -484,9 +484,9 @@ fn map_completion_error(error: CompletionError) -> AgentError {
}
};
let mut mapped = AgentError::new(kind, error.to_string());
if is_authentication_failure {
if kind == AgentErrorKind::Authentication {
mapped.user_message = Some(
"AWS credentials for Bedrock are missing or expired. Refresh your AWS login and try again."
"Provider authentication failed. Check the credentials and access permissions for this connection, then try again."
.to_string(),
);
}
@@ -15,6 +15,22 @@ fn flattened_sse_http_client_error_is_recoverable_transport() {
assert!(mapped.recoverable);
}
#[test]
fn safe_transport_diagnostics_survive_rig_flattening_and_remain_retryable() {
let error = crate::http_client::diagnose_transport_error(
rig_core::http_client::Error::Instance(Box::new(std::io::Error::new(
std::io::ErrorKind::ConnectionRefused,
"private-token-not-for-diagnostics",
))),
);
// Rig uses Display text for SSE failures without a provider response status.
let mapped = map_completion_error(CompletionError::ProviderError(error.to_string()));
assert_eq!(mapped.kind, AgentErrorKind::Transport);
assert_eq!(mapped.recoverable, true);
assert_eq!(mapped.message.contains("ConnectionRefused"), true);
assert_eq!(mapped.message.contains("private-token"), false);
}
#[test]
fn flattened_streamed_provider_overload_is_recoverable() {
let error = CompletionError::ProviderError(
@@ -117,6 +117,22 @@ fn flattened_bedrock_credential_errors_are_authentication_failures() {
}
}
#[test]
fn non_aws_authentication_errors_do_not_recommend_an_aws_login() {
for error in [
CompletionError::from_http_response(http::StatusCode::UNAUTHORIZED, "Invalid API key"),
CompletionError::ProviderError("AccessDenied: model access is not permitted".to_owned()),
] {
let mapped = map_completion_error(error);
assert_eq!(mapped.kind, AgentErrorKind::Authentication);
assert_eq!(mapped.recoverable, false);
let message = mapped.user_message.unwrap();
assert_eq!(message.contains("Provider authentication failed"), true);
assert_eq!(message.contains("AWS"), false);
assert_eq!(message.contains("Bedrock"), false);
}
}
#[test]
fn unrelated_provider_failures_do_not_become_authentication_failures() {
for message in [