fix(rust): harden Anthropic transport failures

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.

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<std::ffi::OsString>);
+
+    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"
+        );
+    }
+}

runtime/rust/prompty-anthropic/src/executor.rs

@@ -195,13 +195,7 @@ fn resolve_connection(
 
 fn build_url(agent: &Prompty) -> Result<String, InvokerError> {
     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<String, InvokerError> {
@@ -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<std::ffi::OsString>);
+
+    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, reqwest::Error>(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() {

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;

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

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