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 { 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, }) } pub async fn send_stream(client: &Client, request: &ModelRequest) -> Result { 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 = None; 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"); } }; 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::(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, }) } #[derive(Debug, Deserialize)] struct AnthropicResponse { content: Vec, } #[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, } 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) { for entry in std::fs::read_dir(dir) .expect("read debug dir") .collect::, _>>() .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()); 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, } } }