|
- use crate::runner::{
- HttpDebugRequest, HttpDebugResponse, ModelRequest, ModelResponse, request_headers_for_debug,
- response_headers_for_debug,
- };
- use anyhow::{Context, Result, bail};
- use futures::StreamExt;
- use reqwest::Client;
- use serde::Deserialize;
- use serde_json::{Value, json};
- use std::time::Instant;
-
- pub async fn send(client: &Client, request: &ModelRequest) -> Result<ModelResponse> {
- let url = super::endpoint_url(&request.base_url, "/v1/messages")?;
- let debug_url = url.to_string();
- let request_body = request_body(request, false);
- let response = client
- .post(url)
- .header("x-api-key", &request.api_token)
- .header("anthropic-version", "2023-06-01")
- .json(&request_body)
- .send()
- .await
- .map_err(|error| {
- write_request_error_debug_blocking(
- request,
- &debug_url,
- request_body.clone(),
- format!("failed to send Anthropic messages request: {error}"),
- "anthropic-request-error",
- );
- error
- })
- .context("failed to send Anthropic messages request")?;
-
- let status = response.status();
- let status_code = status.as_u16();
- let response_headers = response.headers().clone();
- let body = response
- .text()
- .await
- .map_err(|error| {
- write_request_error_debug_blocking(
- request,
- &debug_url,
- request_body.clone(),
- format!("failed to read Anthropic response body: {error}"),
- "anthropic-request-error",
- );
- error
- })
- .context("failed to read Anthropic response body")?;
- if status.is_success()
- && let Some(raw_debug) = &request.raw_debug
- && raw_debug.write_success_raw()
- {
- raw_debug
- .write_response("anthropic-json", &body)
- .await
- .context("failed to write Anthropic raw debug response")?;
- }
-
- if !status.is_success() {
- write_error_debug(
- request,
- &debug_url,
- request_body,
- status_code,
- &response_headers,
- &body,
- "anthropic-error-http",
- )
- .await?;
- bail!(
- "{}",
- super::upstream_error_message("Anthropic", status_code, &body)
- );
- }
-
- let parsed: AnthropicResponse = match serde_json::from_str(&body) {
- Ok(parsed) => parsed,
- Err(error) => {
- write_response_request_error_debug(
- request,
- &debug_url,
- request_body,
- status_code,
- &response_headers,
- &body,
- format!("failed to parse Anthropic response JSON: {error}"),
- )
- .await?;
- return Err(error).context("failed to parse Anthropic response JSON");
- }
- };
- let Some(text) = parsed.content.into_iter().find_map(|block| match block {
- AnthropicContentBlock::Text { text } if !text.is_empty() => Some(text),
- _ => None,
- }) else {
- write_response_request_error_debug(
- request,
- &debug_url,
- request_body,
- status_code,
- &response_headers,
- &body,
- "Anthropic response missing first text content block".to_string(),
- )
- .await?;
- bail!("Anthropic response missing first text content block");
- };
-
- Ok(ModelResponse {
- text,
- status: status_code,
- elapsed_ms: 0,
- first_token_ms: None,
- chunk_elapsed_ms: Vec::new(),
- })
- }
-
- pub async fn send_stream(client: &Client, request: &ModelRequest) -> Result<ModelResponse> {
- let url = super::endpoint_url(&request.base_url, "/v1/messages")?;
- let debug_url = url.to_string();
- let started = Instant::now();
- let request_body = request_body(request, true);
- let response = client
- .post(url)
- .header("x-api-key", &request.api_token)
- .header("anthropic-version", "2023-06-01")
- .json(&request_body)
- .send()
- .await
- .map_err(|error| {
- write_request_error_debug_blocking(
- request,
- &debug_url,
- request_body.clone(),
- format!("failed to send Anthropic streaming request: {error}"),
- "anthropic-request-error",
- );
- error
- })
- .context("failed to send Anthropic streaming request")?;
-
- let status = response.status();
- let status_code = status.as_u16();
- let response_headers = response.headers().clone();
-
- if !status.is_success() {
- let body = response
- .text()
- .await
- .map_err(|error| {
- write_request_error_debug_blocking(
- request,
- &debug_url,
- request_body.clone(),
- format!("failed to read Anthropic error response body: {error}"),
- "anthropic-request-error",
- );
- error
- })
- .context("failed to read Anthropic error response body")?;
- if let Some(raw_debug) = &request.raw_debug {
- raw_debug
- .write_http_error(
- "anthropic-error-http",
- debug_request(request, &debug_url, request_body),
- HttpDebugResponse {
- status: Some(status_code),
- headers: response_headers_for_debug(&response_headers),
- body: body.clone(),
- error_kind: None,
- error: None,
- },
- )
- .await
- .context("failed to write Anthropic raw debug error response")?;
- }
- bail!(
- "{}",
- super::upstream_error_message("Anthropic", status_code, &body)
- );
- }
-
- let mut stream = response.bytes_stream();
- let mut buffer = super::SseLineBuffer::new();
- let mut text = String::new();
- let mut raw_stream = String::new();
- let mut first_token_ms: Option<u128> = None;
- let mut chunk_elapsed_ms = Vec::new();
- let mut current_event = String::new();
- let mut done = false;
-
- while let Some(chunk) = stream.next().await {
- let chunk = match chunk {
- Ok(chunk) => chunk,
- Err(error) => {
- write_response_request_error_debug(
- request,
- &debug_url,
- request_body.clone(),
- status_code,
- &response_headers,
- &raw_stream,
- format!("Anthropic stream interrupted: {error}"),
- )
- .await?;
- return Err(error).context("Anthropic stream interrupted");
- }
- };
- chunk_elapsed_ms.push(started.elapsed().as_millis());
- for line in buffer.feed(&chunk) {
- raw_stream.push_str(&line);
- raw_stream.push('\n');
- if let Some(event_type) = line.strip_prefix("event: ") {
- current_event = event_type.to_string();
- continue;
- }
- if let Some(data) = line.strip_prefix("data: ") {
- if current_event == "content_block_delta" {
- if let Some(content) = serde_json::from_str::<AnthropicStreamDelta>(data)
- .ok()
- .and_then(|p| p.delta.text)
- .filter(|c| !c.is_empty())
- {
- if first_token_ms.is_none() {
- first_token_ms = Some(started.elapsed().as_millis());
- }
- text.push_str(&content);
- }
- } else if current_event == "message_stop" {
- done = true;
- break;
- }
- }
- if line.is_empty() {
- current_event.clear();
- }
- }
- if done {
- break;
- }
- }
-
- if let Some(raw_debug) = &request.raw_debug
- && raw_debug.write_success_raw()
- {
- raw_debug
- .write_response("anthropic-sse", &raw_stream)
- .await
- .context("failed to write Anthropic raw debug stream")?;
- }
-
- if text.is_empty() {
- write_response_request_error_debug(
- request,
- &debug_url,
- request_body,
- status_code,
- &response_headers,
- &raw_stream,
- "Anthropic stream completed without producing any content".to_string(),
- )
- .await?;
- bail!("Anthropic stream completed without producing any content");
- }
-
- Ok(ModelResponse {
- text,
- status: status_code,
- elapsed_ms: 0,
- first_token_ms,
- chunk_elapsed_ms,
- })
- }
-
- #[derive(Debug, Deserialize)]
- struct AnthropicResponse {
- content: Vec<AnthropicContentBlock>,
- }
-
- #[derive(Debug, Deserialize)]
- #[serde(tag = "type")]
- enum AnthropicContentBlock {
- #[serde(rename = "text")]
- Text { text: String },
- #[serde(other)]
- Other,
- }
-
- #[derive(Debug, Deserialize)]
- struct AnthropicStreamDelta {
- delta: AnthropicDeltaContent,
- }
-
- #[derive(Debug, Deserialize)]
- struct AnthropicDeltaContent {
- text: Option<String>,
- }
-
- fn request_body(request: &ModelRequest, stream: bool) -> Value {
- let mut body = json!({
- "model": request.model,
- "messages": [{"role": "user", "content": request.prompt}],
- "temperature": request.temperature,
- "max_tokens": request.max_tokens
- });
- let object = body
- .as_object_mut()
- .expect("Anthropic request body is an object");
- if stream {
- object.insert("stream".to_string(), json!(true));
- }
- if let Some(thinking) = &request.thinking
- && thinking.enabled
- {
- object.remove("temperature");
- let mut thinking_body = serde_json::Map::new();
- let thinking_type = thinking.kind.as_deref().unwrap_or("enabled");
- thinking_body.insert("type".to_string(), json!(thinking_type));
- if thinking_type == "adaptive" {
- if let Some(effort) = &thinking.effort {
- thinking_body.insert("effort".to_string(), json!(effort));
- }
- } else if let Some(budget_tokens) = thinking.budget_tokens {
- thinking_body.insert("budget_tokens".to_string(), json!(budget_tokens));
- }
- if let Some(display) = &thinking.display {
- thinking_body.insert("display".to_string(), json!(display));
- }
- object.insert("thinking".to_string(), Value::Object(thinking_body));
- }
- body
- }
-
- async fn write_error_debug(
- request: &ModelRequest,
- url: &str,
- request_body: Value,
- status: u16,
- response_headers: &reqwest::header::HeaderMap,
- response_body: &str,
- response_kind: &str,
- ) -> Result<()> {
- if let Some(raw_debug) = &request.raw_debug {
- raw_debug
- .write_http_error(
- response_kind,
- debug_request(request, url, request_body),
- HttpDebugResponse {
- status: Some(status),
- headers: response_headers_for_debug(response_headers),
- body: response_body.to_string(),
- error_kind: None,
- error: None,
- },
- )
- .await
- .context("failed to write Anthropic raw debug error response")?;
- }
- Ok(())
- }
-
- async fn write_response_request_error_debug(
- request: &ModelRequest,
- url: &str,
- request_body: Value,
- status: u16,
- response_headers: &reqwest::header::HeaderMap,
- response_body: &str,
- error: String,
- ) -> Result<()> {
- if let Some(raw_debug) = &request.raw_debug {
- raw_debug
- .write_http_error(
- "anthropic-request-error",
- debug_request(request, url, request_body),
- HttpDebugResponse {
- status: Some(status),
- headers: response_headers_for_debug(response_headers),
- body: response_body.to_string(),
- error_kind: Some("request_error".to_string()),
- error: Some(error),
- },
- )
- .await
- .context("failed to write Anthropic raw debug request error response")?;
- }
- Ok(())
- }
-
- fn debug_request(request: &ModelRequest, url: &str, body: Value) -> HttpDebugRequest {
- HttpDebugRequest {
- method: "POST".to_string(),
- url: url.to_string(),
- headers: request_headers_for_debug(&[
- ("x-api-key", request.api_token.clone()),
- ("anthropic-version", "2023-06-01".to_string()),
- ]),
- body,
- }
- }
-
- fn write_request_error_debug_blocking(
- request: &ModelRequest,
- url: &str,
- request_body: Value,
- error: String,
- response_kind: &str,
- ) {
- if let Some(raw_debug) = &request.raw_debug {
- let _ = futures::executor::block_on(raw_debug.write_request_error(
- response_kind,
- debug_request(request, url, request_body),
- error,
- ));
- }
- }
-
- #[cfg(test)]
- mod tests {
- use crate::runner::{ModelRequest, RawDebugConfig, ThinkingConfig};
- use reqwest::Client;
- use std::path::{Path, PathBuf};
- use wiremock::matchers::{body_json, header, method, path};
- use wiremock::{Mock, MockServer, ResponseTemplate};
-
- fn read_single_debug_file(root: &Path) -> String {
- let mut files = Vec::new();
- collect_debug_files(root, &mut files);
- assert_eq!(files.len(), 1);
- std::fs::read_to_string(&files[0]).expect("read raw debug file")
- }
-
- fn collect_debug_files(dir: &Path, files: &mut Vec<PathBuf>) {
- for entry in std::fs::read_dir(dir)
- .expect("read debug dir")
- .collect::<Result<Vec<_>, _>>()
- .expect("debug entries")
- {
- let path = entry.path();
- if path.is_dir() {
- collect_debug_files(&path, files);
- } else {
- files.push(path);
- }
- }
- }
-
- #[tokio::test]
- async fn extracts_first_text_block() {
- let server = MockServer::start().await;
- Mock::given(method("POST"))
- .and(path("/v1/messages"))
- .and(header("x-api-key", "test-token"))
- .and(header("anthropic-version", "2023-06-01"))
- .and(body_json(serde_json::json!({
- "model": "claude-test",
- "messages": [{"role": "user", "content": "hello"}],
- "temperature": 0.0,
- "max_tokens": 1024
- })))
- .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
- "content": [
- {"type": "text", "text": "hi from claude"}
- ]
- })))
- .mount(&server)
- .await;
-
- let request = ModelRequest {
- base_url: server.uri(),
- api_token: "test-token".to_string(),
- model: "claude-test".to_string(),
- prompt: "hello".to_string(),
- temperature: 0.0,
- max_tokens: 1024,
- stream: false,
- raw_debug: None,
- thinking: None,
- };
-
- let response = super::send(&Client::new(), &request)
- .await
- .expect("response");
-
- assert_eq!(response.status, 200);
- assert_eq!(response.text, "hi from claude");
- }
-
- #[tokio::test]
- async fn sends_thinking_and_omits_temperature_when_enabled() {
- let server = MockServer::start().await;
- Mock::given(method("POST"))
- .and(path("/v1/messages"))
- .and(body_json(serde_json::json!({
- "model": "claude-test",
- "messages": [{"role": "user", "content": "hello"}],
- "max_tokens": 1024,
- "thinking": {
- "type": "enabled",
- "budget_tokens": 10000,
- "display": "omitted"
- }
- })))
- .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
- "content": [
- {"type": "text", "text": "hi from claude"}
- ]
- })))
- .mount(&server)
- .await;
-
- let mut request = anthropic_request(server.uri());
- request.thinking = Some(ThinkingConfig {
- enabled: true,
- kind: Some("enabled".to_string()),
- budget_tokens: Some(10000),
- effort: None,
- display: Some("omitted".to_string()),
- reasoning_effort: None,
- reasoning_summary: None,
- });
-
- let response = super::send(&Client::new(), &request)
- .await
- .expect("response");
-
- assert_eq!(response.text, "hi from claude");
- }
-
- #[tokio::test]
- async fn sends_adaptive_thinking_effort_without_budget_tokens() {
- let server = MockServer::start().await;
- Mock::given(method("POST"))
- .and(path("/v1/messages"))
- .and(body_json(serde_json::json!({
- "model": "claude-test",
- "messages": [{"role": "user", "content": "hello"}],
- "max_tokens": 1024,
- "thinking": {
- "type": "adaptive",
- "effort": "high",
- "display": "omitted"
- }
- })))
- .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
- "content": [
- {"type": "text", "text": "hi from claude"}
- ]
- })))
- .mount(&server)
- .await;
-
- let mut request = anthropic_request(server.uri());
- request.thinking = Some(ThinkingConfig {
- enabled: true,
- kind: Some("adaptive".to_string()),
- budget_tokens: Some(10000),
- effort: Some("high".to_string()),
- display: Some("omitted".to_string()),
- reasoning_effort: None,
- reasoning_summary: None,
- });
-
- let response = super::send(&Client::new(), &request)
- .await
- .expect("response");
-
- assert_eq!(response.text, "hi from claude");
- }
-
- #[tokio::test]
- async fn errors_on_non_success_status() {
- let server = MockServer::start().await;
- Mock::given(method("POST"))
- .and(path("/v1/messages"))
- .respond_with(ResponseTemplate::new(401).set_body_json(serde_json::json!({
- "error": {
- "message": "unauthorized"
- }
- })))
- .mount(&server)
- .await;
-
- let request = ModelRequest {
- base_url: server.uri(),
- api_token: "test-token".to_string(),
- model: "claude-test".to_string(),
- prompt: "hello".to_string(),
- temperature: 0.0,
- max_tokens: 1024,
- stream: false,
- raw_debug: None,
- thinking: None,
- };
-
- let error = super::send(&Client::new(), &request)
- .await
- .expect_err("non-success should fail");
-
- assert!(error.to_string().contains("401"));
- assert!(error.to_string().contains("unauthorized"));
- }
-
- #[tokio::test]
- async fn non_success_error_does_not_leak_sensitive_response_body() {
- let server = MockServer::start().await;
- Mock::given(method("POST"))
- .and(path("/v1/messages"))
- .respond_with(ResponseTemplate::new(429).set_body_json(serde_json::json!({
- "error": {
- "message": "rate limited secret prompt sk-leaked-token"
- }
- })))
- .mount(&server)
- .await;
-
- let request = ModelRequest {
- base_url: server.uri(),
- api_token: "test-token".to_string(),
- model: "claude-test".to_string(),
- prompt: "hello".to_string(),
- temperature: 0.0,
- max_tokens: 1024,
- stream: false,
- raw_debug: None,
- thinking: None,
- };
-
- let message = super::send(&Client::new(), &request)
- .await
- .expect_err("non-success should fail")
- .to_string();
-
- assert!(message.contains("Anthropic"));
- assert!(message.contains("429"));
- assert!(message.contains("rate limited"));
- assert!(!message.contains("secret prompt"));
- assert!(!message.contains("sk-leaked-token"));
- }
-
- #[tokio::test]
- async fn non_success_error_debug_records_request_and_response_without_token() {
- let server = MockServer::start().await;
- let temp_dir = tempfile::tempdir().expect("create temp dir");
- Mock::given(method("POST"))
- .and(path("/v1/messages"))
- .respond_with(
- ResponseTemplate::new(504)
- .insert_header("x-request-id", "anthropic-req")
- .set_body_json(serde_json::json!({
- "error": {
- "message": "gateway timeout"
- }
- })),
- )
- .mount(&server)
- .await;
-
- let request = ModelRequest {
- base_url: server.uri(),
- api_token: "anthropic-secret-token".to_string(),
- model: "claude-test".to_string(),
- prompt: "full anthropic prompt".to_string(),
- temperature: 0.0,
- max_tokens: 1024,
- stream: false,
- raw_debug: Some(RawDebugConfig::new(
- temp_dir.path().to_path_buf(),
- "anthropic-claude-test".to_string(),
- )),
- thinking: None,
- };
-
- let _ = super::send(&Client::new(), &request)
- .await
- .expect_err("non-success should fail");
-
- let raw = read_single_debug_file(temp_dir.path());
- let debug: serde_json::Value = serde_json::from_str(&raw).expect("debug json");
-
- assert_eq!(debug["request"]["headers"]["x-api-key"], "[REDACTED]");
- assert_eq!(
- debug["request"]["body"]["messages"][0]["content"],
- "full anthropic prompt"
- );
- assert_eq!(debug["response"]["status"], 504);
- assert_eq!(
- debug["response"]["headers"]["x-request-id"],
- "anthropic-req"
- );
- assert!(
- debug["response"]["body"]
- .as_str()
- .expect("response body")
- .contains("gateway timeout")
- );
- assert!(!raw.contains("anthropic-secret-token"));
- }
-
- #[tokio::test]
- async fn request_error_debug_records_request_and_local_error_without_token() {
- let temp_dir = tempfile::tempdir().expect("create temp dir");
- let request = ModelRequest {
- base_url: "http://127.0.0.1:9".to_string(),
- api_token: "anthropic-secret-token".to_string(),
- model: "claude-test".to_string(),
- prompt: "prompt before anthropic connect error".to_string(),
- temperature: 0.0,
- max_tokens: 1024,
- stream: false,
- raw_debug: Some(RawDebugConfig::new(
- temp_dir.path().to_path_buf(),
- "anthropic-claude-test".to_string(),
- )),
- thinking: None,
- };
-
- let _ = super::send(&Client::new(), &request)
- .await
- .expect_err("connection should fail");
-
- let raw = read_single_debug_file(temp_dir.path());
- let debug: serde_json::Value = serde_json::from_str(&raw).expect("debug json");
-
- assert_eq!(debug["request"]["method"], "POST");
- assert!(
- debug["request"]["url"]
- .as_str()
- .expect("request url")
- .ends_with("/v1/messages")
- );
- assert_eq!(debug["request"]["headers"]["x-api-key"], "[REDACTED]");
- assert_eq!(
- debug["request"]["body"]["messages"][0]["content"],
- "prompt before anthropic connect error"
- );
- assert_eq!(debug["response"]["status"], serde_json::Value::Null);
- assert_eq!(debug["response"]["error_kind"], "request_error");
- assert!(
- debug["response"]["error"]
- .as_str()
- .expect("local error")
- .contains("failed to send Anthropic messages request")
- );
- assert!(!raw.contains("anthropic-secret-token"));
- }
-
- #[tokio::test]
- async fn base_url_with_v1_prefix_does_not_duplicate_messages_path() {
- let server = MockServer::start().await;
- Mock::given(method("POST"))
- .and(path("/v1/messages"))
- .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
- "content": [
- {"type": "text", "text": "prefixed"}
- ]
- })))
- .mount(&server)
- .await;
-
- let request = ModelRequest {
- base_url: format!("{}/v1", server.uri()),
- api_token: "test-token".to_string(),
- model: "claude-test".to_string(),
- prompt: "hello".to_string(),
- temperature: 0.0,
- max_tokens: 1024,
- stream: false,
- raw_debug: None,
- thinking: None,
- };
-
- let response = super::send(&Client::new(), &request)
- .await
- .expect("response");
-
- assert_eq!(response.text, "prefixed");
- }
-
- #[tokio::test]
- async fn extracts_streaming_messages_text_and_ttft() {
- let server = MockServer::start().await;
- let temp_dir = tempfile::tempdir().expect("create temp dir");
- Mock::given(method("POST"))
- .and(path("/v1/messages"))
- .and(header("x-api-key", "test-token"))
- .and(header("anthropic-version", "2023-06-01"))
- .and(body_json(serde_json::json!({
- "model": "claude-test",
- "messages": [{"role": "user", "content": "hello"}],
- "temperature": 0.0,
- "max_tokens": 1024,
- "stream": true
- })))
- .respond_with(
- ResponseTemplate::new(200)
- .insert_header("content-type", "text/event-stream")
- .set_body_string(
- "event: content_block_delta\n\
- data: {\"type\":\"content_block_delta\",\"delta\":{\"type\":\"text_delta\",\"text\":\"hi \"}}\n\n\
- event: content_block_delta\n\
- data: {\"type\":\"content_block_delta\",\"delta\":{\"type\":\"text_delta\",\"text\":\"there\"}}\n\n\
- event: message_stop\n\
- data: {\"type\":\"message_stop\"}\n\n",
- ),
- )
- .mount(&server)
- .await;
-
- let request = ModelRequest {
- base_url: server.uri(),
- api_token: "test-token".to_string(),
- model: "claude-test".to_string(),
- prompt: "hello".to_string(),
- temperature: 0.0,
- max_tokens: 1024,
- stream: true,
- raw_debug: Some(RawDebugConfig::new(
- temp_dir.path().to_path_buf(),
- "anthropic-claude-test".to_string(),
- )),
- thinking: None,
- };
-
- let response = super::send_stream(&Client::new(), &request)
- .await
- .expect("stream response");
-
- assert_eq!(response.status, 200);
- assert_eq!(response.text, "hi there");
- assert!(response.first_token_ms.is_some());
- assert_eq!(response.chunk_elapsed_ms.len(), 1);
-
- let raw = read_single_debug_file(temp_dir.path());
- assert!(raw.contains("event: content_block_delta"));
- assert!(raw.contains("\"text\":\"hi \""));
- }
-
- #[tokio::test]
- async fn empty_stream_request_error_debug_records_request_and_raw_stream() {
- let server = MockServer::start().await;
- let temp_dir = tempfile::tempdir().expect("create temp dir");
- Mock::given(method("POST"))
- .and(path("/v1/messages"))
- .respond_with(
- ResponseTemplate::new(200)
- .insert_header("content-type", "text/event-stream")
- .set_body_string(
- "event: message_start\n\
- data: {\"type\":\"message_start\"}\n\n\
- event: message_stop\n\
- data: {\"type\":\"message_stop\"}\n\n",
- ),
- )
- .mount(&server)
- .await;
-
- let request = ModelRequest {
- base_url: server.uri(),
- api_token: "anthropic-secret-token".to_string(),
- model: "claude-test".to_string(),
- prompt: "prompt before empty stream".to_string(),
- temperature: 0.0,
- max_tokens: 1024,
- stream: true,
- raw_debug: Some(
- RawDebugConfig::new(
- temp_dir.path().to_path_buf(),
- "anthropic-claude-test".to_string(),
- )
- .with_success_raw(false),
- ),
- thinking: None,
- };
-
- let error = super::send_stream(&Client::new(), &request)
- .await
- .expect_err("empty stream should fail");
- assert!(
- error
- .to_string()
- .contains("completed without producing any content")
- );
-
- let raw = read_single_debug_file(temp_dir.path());
- let debug: serde_json::Value = serde_json::from_str(&raw).expect("debug json");
-
- assert_eq!(
- debug["request"]["body"]["messages"][0]["content"],
- "prompt before empty stream"
- );
- assert_eq!(debug["response"]["status"], 200);
- assert_eq!(debug["response"]["error_kind"], "request_error");
- assert!(
- debug["response"]["body"]
- .as_str()
- .expect("response body")
- .contains("event: message_stop")
- );
- assert!(!raw.contains("anthropic-secret-token"));
- }
-
- #[tokio::test]
- async fn interrupted_stream_debug_records_status_headers_and_partial_stream() {
- use tokio::io::{AsyncReadExt, AsyncWriteExt};
- use tokio::net::TcpListener;
-
- let listener = TcpListener::bind("127.0.0.1:0")
- .await
- .expect("bind listener");
- let addr = listener.local_addr().expect("listener addr");
- let server = tokio::spawn(async move {
- let (mut socket, _) = listener.accept().await.expect("accept request");
- let mut buffer = [0_u8; 4096];
- let _ = socket.read(&mut buffer).await.expect("read request");
- let chunk = "event: ping\ndata: {\"type\":\"ping\"}\n\n";
- let response = format!(
- "HTTP/1.1 200 OK\r\n\
- content-type: text/event-stream\r\n\
- transfer-encoding: chunked\r\n\
- x-request-id: broken-stream\r\n\
- \r\n\
- {:x}\r\n{}\r\n\
- zz\r\nbroken\r\n",
- chunk.len(),
- chunk
- );
- socket
- .write_all(response.as_bytes())
- .await
- .expect("write malformed response");
- });
-
- let temp_dir = tempfile::tempdir().expect("create temp dir");
- let request = ModelRequest {
- base_url: format!("http://{addr}"),
- api_token: "anthropic-secret-token".to_string(),
- model: "claude-test".to_string(),
- prompt: "prompt before broken stream".to_string(),
- temperature: 0.0,
- max_tokens: 1024,
- stream: true,
- raw_debug: Some(
- RawDebugConfig::new(
- temp_dir.path().to_path_buf(),
- "anthropic-claude-test".to_string(),
- )
- .with_success_raw(false),
- ),
- thinking: None,
- };
-
- let error = super::send_stream(&Client::new(), &request)
- .await
- .expect_err("malformed chunked stream should fail");
- server.await.expect("server task");
- assert!(error.to_string().contains("stream interrupted"));
-
- let raw = read_single_debug_file(temp_dir.path());
- let debug: serde_json::Value = serde_json::from_str(&raw).expect("debug json");
-
- assert_eq!(debug["response"]["status"], 200);
- assert_eq!(
- debug["response"]["headers"]["x-request-id"],
- "broken-stream"
- );
- assert_eq!(debug["response"]["error_kind"], "request_error");
- assert!(
- debug["response"]["body"]
- .as_str()
- .expect("partial stream")
- .contains("event: ping")
- );
- assert!(
- debug["response"]["error"]
- .as_str()
- .expect("stream error")
- .contains("Anthropic stream interrupted")
- );
- assert!(!raw.contains("anthropic-secret-token"));
- }
-
- fn anthropic_request(base_url: String) -> ModelRequest {
- ModelRequest {
- base_url,
- api_token: "test-token".to_string(),
- model: "claude-test".to_string(),
- prompt: "hello".to_string(),
- temperature: 0.0,
- max_tokens: 1024,
- stream: false,
- raw_debug: None,
- thinking: None,
- }
- }
- }
|