From dcb863f8c11e96d2eeda790fb0608496baadfc56 Mon Sep 17 00:00:00 2001 From: Seth Juarez Date: Tue, 4 Aug 2026 02:45:13 -0700 Subject: [PATCH] fix(rust): harden Anthropic transport failures Honor Anthropic base URL overrides without duplicating API versions, and surface provider versus transport SSE failures with canonical outcome semantics. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- runtime/rust/prompty-anthropic/README.md | 3 + .../rust/prompty-anthropic/src/endpoint.rs | 93 +++++++++++++++++++ .../rust/prompty-anthropic/src/executor.rs | 72 ++++++++++++-- runtime/rust/prompty-anthropic/src/lib.rs | 2 + runtime/rust/prompty-anthropic/src/models.rs | 10 +- .../rust/prompty-anthropic/src/processor.rs | 85 ++++++++++++++++- 6 files changed, 249 insertions(+), 16 deletions(-) create mode 100644 runtime/rust/prompty-anthropic/src/endpoint.rs diff --git a/runtime/rust/prompty-anthropic/README.md b/runtime/rust/prompty-anthropic/README.md index 76441e346..4759875b4 100644 --- a/runtime/rust/prompty-anthropic/README.md +++ b/runtime/rust/prompty-anthropic/README.md @@ -28,6 +28,9 @@ model: apiKey: ${env:ANTHROPIC_API_KEY} ``` +`ANTHROPIC_BASE_URL` optionally overrides the default `https://api.anthropic.com` +base URL. A non-empty `model.connection.endpoint` takes precedence. + ## License MIT — see [LICENSE](LICENSE) for details. diff --git a/runtime/rust/prompty-anthropic/src/endpoint.rs b/runtime/rust/prompty-anthropic/src/endpoint.rs new file mode 100644 index 000000000..a74d36d1e --- /dev/null +++ b/runtime/rust/prompty-anthropic/src/endpoint.rs @@ -0,0 +1,93 @@ +//! Resolve Anthropic API endpoints consistently across provider operations. + +use serde_json::Value; + +const DEFAULT_BASE_URL: &str = "https://api.anthropic.com"; + +pub(crate) fn build_api_url(connection: &Value, resource: &str) -> String { + let endpoint = connection + .get("endpoint") + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .map(ToString::to_string) + .or_else(|| { + std::env::var("ANTHROPIC_BASE_URL") + .ok() + .filter(|value| !value.is_empty()) + }) + .unwrap_or_else(|| DEFAULT_BASE_URL.to_string()); + + let base = endpoint.trim_end_matches('/'); + if base.ends_with("/v1") { + format!("{base}/{resource}") + } else { + format!("{base}/v1/{resource}") + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use serial_test::serial; + + struct BaseUrlGuard(Option); + + impl BaseUrlGuard { + fn set(value: Option<&str>) -> Self { + let original = std::env::var_os("ANTHROPIC_BASE_URL"); + match value { + Some(value) => unsafe { std::env::set_var("ANTHROPIC_BASE_URL", value) }, + None => unsafe { std::env::remove_var("ANTHROPIC_BASE_URL") }, + } + Self(original) + } + } + + impl Drop for BaseUrlGuard { + fn drop(&mut self) { + match self.0.take() { + Some(value) => unsafe { std::env::set_var("ANTHROPIC_BASE_URL", value) }, + None => unsafe { std::env::remove_var("ANTHROPIC_BASE_URL") }, + } + } + } + + #[test] + #[serial] + fn default_endpoint_is_used_when_unconfigured() { + let _guard = BaseUrlGuard::set(None); + + assert_eq!( + build_api_url(&json!({}), "messages"), + "https://api.anthropic.com/v1/messages" + ); + } + + #[test] + #[serial] + fn connection_endpoint_takes_precedence_over_environment() { + let _guard = BaseUrlGuard::set(Some("https://environment.example/v1")); + let connection = json!({"endpoint": "https://connection.example/v1/"}); + + assert_eq!( + build_api_url(&connection, "messages"), + "https://connection.example/v1/messages" + ); + } + + #[test] + #[serial] + fn environment_endpoint_is_used_without_duplicate_v1() { + let _guard = BaseUrlGuard::set(Some("https://environment.example/v1/")); + + assert_eq!( + build_api_url(&json!({}), "messages"), + "https://environment.example/v1/messages" + ); + assert_eq!( + build_api_url(&json!({}), "models"), + "https://environment.example/v1/models" + ); + } +} diff --git a/runtime/rust/prompty-anthropic/src/executor.rs b/runtime/rust/prompty-anthropic/src/executor.rs index 4ffa9d36e..05e28bdf9 100644 --- a/runtime/rust/prompty-anthropic/src/executor.rs +++ b/runtime/rust/prompty-anthropic/src/executor.rs @@ -195,13 +195,7 @@ fn resolve_connection( fn build_url(agent: &Prompty) -> Result { let conn = resolve_connection(agent)?; - let endpoint = conn - .get("endpoint") - .and_then(|e| e.as_str()) - .unwrap_or("https://api.anthropic.com"); - - let base = endpoint.trim_end_matches('/'); - Ok(format!("{base}/v1/messages")) + Ok(crate::endpoint::build_api_url(&conn, "messages")) } fn get_api_key(agent: &Prompty) -> Result { @@ -372,8 +366,14 @@ impl Stream for AnthropicSseParser { return Poll::Ready(None); } Poll::Ready(None) => { + self.pending.push_back(serde_json::json!({ + "error": { + "type": "sse_transport_error", + "message": "Anthropic SSE stream ended before message_stop", + } + })); self.done = true; - return Poll::Ready(None); + return Poll::Ready(self.pending.pop_front()); } Poll::Pending => { return Poll::Pending; @@ -391,6 +391,24 @@ mod tests { use serde_json::json; use serial_test::serial; + struct RemovedBaseUrl(Option); + + impl RemovedBaseUrl { + fn new() -> Self { + let previous = std::env::var_os("ANTHROPIC_BASE_URL"); + unsafe { std::env::remove_var("ANTHROPIC_BASE_URL") }; + Self(previous) + } + } + + impl Drop for RemovedBaseUrl { + fn drop(&mut self) { + if let Some(value) = self.0.take() { + unsafe { std::env::set_var("ANTHROPIC_BASE_URL", value) }; + } + } + } + fn make_agent(model_json: Value) -> Prompty { let mut data = json!({ "name": "test", @@ -404,6 +422,7 @@ mod tests { #[test] #[serial] fn test_build_url_default() { + let _base_url = RemovedBaseUrl::new(); let agent = make_agent(json!({"id": "claude-3", "provider": "anthropic"})); let url = build_url(&agent).unwrap(); assert_eq!(url, "https://api.anthropic.com/v1/messages"); @@ -425,6 +444,43 @@ mod tests { assert_eq!(url, "https://custom.anthropic.com/v1/messages"); } + #[test] + #[serial] + fn test_build_url_custom_endpoint_with_v1() { + let agent = make_agent(json!({ + "id": "claude-3", + "provider": "anthropic", + "connection": { + "kind": "key", + "endpoint": "https://custom.anthropic.com/v1/", + "apiKey": "test-key" + } + })); + let url = build_url(&agent).unwrap(); + assert_eq!(url, "https://custom.anthropic.com/v1/messages"); + } + + #[tokio::test] + async fn test_sse_parser_reports_premature_eof() { + use futures::StreamExt; + + let inner = futures::stream::iter(vec![Ok::(Bytes::from( + "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"text_delta\",\"text\":\"partial\"}}\n\n", + ))]); + let mut stream = AnthropicSseParser::new(inner); + + assert_eq!(stream.next().await.unwrap()["type"], "content_block_delta"); + let failure = stream.next().await.unwrap(); + assert_eq!(failure["error"]["type"], "sse_transport_error"); + assert!( + failure["error"]["message"] + .as_str() + .unwrap() + .contains("before message_stop") + ); + assert!(stream.next().await.is_none()); + } + #[test] #[serial] fn test_get_api_key_from_connection() { diff --git a/runtime/rust/prompty-anthropic/src/lib.rs b/runtime/rust/prompty-anthropic/src/lib.rs index c7d1d0705..c9258c370 100644 --- a/runtime/rust/prompty-anthropic/src/lib.rs +++ b/runtime/rust/prompty-anthropic/src/lib.rs @@ -18,6 +18,8 @@ //! // Now invoke/turn will use Anthropic for agents with provider="anthropic" //! ``` +mod endpoint; + pub mod executor; pub mod models; pub mod processor; diff --git a/runtime/rust/prompty-anthropic/src/models.rs b/runtime/rust/prompty-anthropic/src/models.rs index 38f131b14..ee9f781f2 100644 --- a/runtime/rust/prompty-anthropic/src/models.rs +++ b/runtime/rust/prompty-anthropic/src/models.rs @@ -21,13 +21,7 @@ const ANTHROPIC_VERSION: &str = "2023-06-01"; /// Build the models endpoint URL from a connection JSON value. fn build_models_url(connection: &Value) -> String { - let endpoint = connection - .get("endpoint") - .and_then(|e| e.as_str()) - .unwrap_or("https://api.anthropic.com"); - - let base = endpoint.trim_end_matches('/'); - format!("{base}/v1/models") + crate::endpoint::build_api_url(connection, "models") } /// Extract the API key from the connection or fall back to `ANTHROPIC_API_KEY`. @@ -280,7 +274,9 @@ mod tests { } #[test] + #[serial] fn test_build_models_url_default() { + let _env = RemovedEnv::new("ANTHROPIC_BASE_URL"); let conn = serde_json::json!({}); let url = build_models_url(&conn); assert_eq!(url, "https://api.anthropic.com/v1/models"); diff --git a/runtime/rust/prompty-anthropic/src/processor.rs b/runtime/rust/prompty-anthropic/src/processor.rs index a4c8fcd62..0c7d6fbbd 100644 --- a/runtime/rust/prompty-anthropic/src/processor.rs +++ b/runtime/rust/prompty-anthropic/src/processor.rs @@ -216,7 +216,7 @@ use std::collections::BTreeMap; use std::pin::Pin; use std::task::{Context, Poll}; -use prompty::types::{StreamChunk, Usage}; +use prompty::types::{StreamChunk, StreamFailure, Usage}; /// Anthropic stream processor — converts SSE JSON events into `StreamChunk` items. /// @@ -270,6 +270,27 @@ impl futures::Stream for AnthropicStreamProcessor { AnthropicStreamPhase::Streaming => { match this.inner.as_mut().poll_next(cx) { Poll::Ready(Some(event)) => { + if let Some(error) = event.get("error").and_then(Value::as_object) { + this.phase = AnthropicStreamPhase::Done; + let error_type = + error.get("type").and_then(Value::as_str).unwrap_or(""); + let provider_message = error + .get("message") + .and_then(Value::as_str) + .unwrap_or("Anthropic stream failed"); + let message = if error_type.is_empty() { + provider_message.to_string() + } else { + format!("Anthropic stream error ({error_type}): {provider_message}") + }; + let failure = if error_type.starts_with("sse_") { + StreamFailure::Indeterminate(message) + } else { + StreamFailure::Determinate(message) + }; + return Poll::Ready(Some(StreamChunk::Failure(failure))); + } + let event_type = event.get("type").and_then(Value::as_str).unwrap_or(""); match event_type { @@ -493,6 +514,68 @@ mod tests { ); } + #[tokio::test] + async fn test_stream_provider_error_is_determinate() { + use futures::StreamExt; + + let chunks = vec![ + json!({ + "type": "content_block_delta", + "delta": {"type": "text_delta", "text": "partial"} + }), + json!({ + "type": "error", + "error": { + "type": "overloaded_error", + "message": "Overloaded" + } + }), + json!({"type": "message_delta", "usage": {"output_tokens": 5}}), + ]; + let mut stream = AnthropicStreamProcessor::new(futures::stream::iter(chunks)); + + assert!(matches!( + stream.next().await, + Some(StreamChunk::Text(value)) if value == "partial" + )); + assert!(matches!( + stream.next().await, + Some(StreamChunk::Failure(StreamFailure::Determinate(message))) + if message.contains("overloaded_error") && message.contains("Overloaded") + )); + assert!(stream.next().await.is_none()); + } + + #[tokio::test] + async fn test_stream_transport_error_is_indeterminate() { + use futures::StreamExt; + + let chunks = vec![ + json!({ + "type": "content_block_delta", + "delta": {"type": "text_delta", "text": "partial"} + }), + json!({ + "error": { + "type": "sse_transport_error", + "message": "connection reset" + } + }), + ]; + let mut stream = AnthropicStreamProcessor::new(futures::stream::iter(chunks)); + + assert!(matches!( + stream.next().await, + Some(StreamChunk::Text(value)) if value == "partial" + )); + assert!(matches!( + stream.next().await, + Some(StreamChunk::Failure(StreamFailure::Indeterminate(message))) + if message.contains("connection reset") + )); + assert!(stream.next().await.is_none()); + } + #[tokio::test] async fn test_process_text_response() { let agent = make_agent();