| 1 | //! OpenAI Responses API bridge for the OpenAI Codex / ChatGPT provider. |
| 2 | //! |
| 3 | //! Implements a dedicated Responses API client that maps CodeWhale's internal |
| 4 | //! message/tool types to the Responses wire format and parses streaming SSE |
| 5 | //! events back into CodeWhale's `StreamEvent` / `MessageResponse` types. |
| 6 | //! |
| 7 | //! This is intentionally separate from the Chat Completions path |
| 8 | //! (`client/chat.rs`) to avoid protocol hacks. |
| 9 | |
| 10 | use anyhow::{Context, Result}; |
| 11 | use serde_json::{Value, json}; |
| 12 | |
| 13 | use crate::config::ApiProvider; |
| 14 | use crate::llm_client::StreamEventBox; |
| 15 | use crate::logging; |
| 16 | use crate::models::{ |
| 17 | ContentBlock, ContentBlockStart, Delta, MessageDelta, MessageRequest, MessageResponse, |
| 18 | StreamEvent, Tool, Usage, |
| 19 | }; |
| 20 | use crate::tools::schema_sanitize; |
| 21 | |
| 22 | use super::{ |
| 23 | DeepSeekClient, ERROR_BODY_MAX_BYTES, bounded_error_text, from_api_tool_name, |
| 24 | system_to_instructions, to_api_tool_name, |
| 25 | }; |
| 26 | |
| 27 | /// Base URL path for the Codex Responses endpoint. |
| 28 | pub(super) const CODEX_RESPONSES_PATH: &str = "/codex/responses"; |
| 29 | |
| 30 | /// Build the Responses API request body from a `MessageRequest`. |
| 31 | #[cfg(test)] |
| 32 | pub(super) fn build_responses_body(request: &MessageRequest) -> Value { |
| 33 | build_responses_body_for_provider(request, ApiProvider::OpenaiCodex) |
| 34 | } |
| 35 | |
| 36 | /// Build a provider-aware Responses API request body. |
| 37 | /// |
| 38 | /// DeepSeek-V4-Flash-0731 implements the Responses wire shape but is stateless |
| 39 | /// and exposes plain reasoning text rather than OpenAI encrypted summaries. |
| 40 | /// Keep those exact-route differences here instead of leaking them into the |
| 41 | /// provider-neutral message model. |
| 42 | pub(super) fn build_responses_body_for_provider( |
| 43 | request: &MessageRequest, |
| 44 | provider: ApiProvider, |
| 45 | ) -> Value { |
| 46 | let is_deepseek = matches!(provider, ApiProvider::Deepseek | ApiProvider::DeepseekCN); |
| 47 | let model = &request.model; |
| 48 | let mut body = json!({ |
| 49 | "model": model, |
| 50 | "stream": true, |
| 51 | }); |
| 52 | if !is_deepseek { |
| 53 | body["store"] = json!(false); |
| 54 | } |
| 55 | if is_deepseek { |
| 56 | if request.max_tokens > 0 { |
| 57 | body["max_output_tokens"] = json!(request.max_tokens); |
| 58 | } |
| 59 | if let Some(temperature) = request.temperature { |
| 60 | body["temperature"] = json!(temperature); |
| 61 | } |
| 62 | if let Some(top_p) = request.top_p { |
| 63 | body["top_p"] = json!(top_p); |
| 64 | } |
| 65 | } |
| 66 | |
| 67 | // Instructions (system prompt). The Codex Responses backend rejects |
| 68 | // requests without instructions, so fall back to a minimal system |
| 69 | // prompt when the caller did not supply one. |
| 70 | let instructions = system_to_instructions(request.system.clone()) |
| 71 | .filter(|text| !text.trim().is_empty()) |
| 72 | .unwrap_or_else(|| "You are a helpful assistant.".to_string()); |
| 73 | body["instructions"] = json!(instructions); |
| 74 | |
| 75 | // Convert messages to Responses input items. |
| 76 | let input = convert_messages_to_responses_input(request, is_deepseek); |
| 77 | body["input"] = json!(input); |
| 78 | |
| 79 | // Convert tools to Responses function tools. |
| 80 | if let Some(tools) = request.tools.as_ref() { |
| 81 | let responses_tools: Vec<Value> = tools.iter().map(tool_to_responses_function).collect(); |
| 82 | if !responses_tools.is_empty() { |
| 83 | body["tools"] = json!(responses_tools); |
| 84 | body["tool_choice"] = json!("auto"); |
| 85 | body["parallel_tool_calls"] = json!(true); |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | // Reasoning configuration. The Codex Responses backend accepts |
| 90 | // low/medium/high/xhigh, so provider-aware callers normalize inherited |
| 91 | // DeepSeek-only values before request construction: "off" becomes |
| 92 | // "low", and CodeWhale's "auto" falls back to "medium". DeepSeek's |
| 93 | // Responses API documents `reasoning.effort: "none"` to disable |
| 94 | // thinking, so its branch sends "none" for the off tier instead of |
| 95 | // collapsing it into low (see `responses_reasoning_effort`). |
| 96 | if let Some(raw) = request.reasoning_effort.as_deref() |
| 97 | && let Some(effort) = responses_reasoning_effort(raw, is_deepseek) |
| 98 | { |
| 99 | body["reasoning"] = if is_deepseek { |
| 100 | json!({ "effort": effort }) |
| 101 | } else { |
| 102 | json!({ |
| 103 | "effort": effort, |
| 104 | "summary": "auto", |
| 105 | }) |
| 106 | }; |
| 107 | } |
| 108 | |
| 109 | // OpenAI Codex can replay encrypted reasoning. DeepSeek exposes plain |
| 110 | // `reasoning_text` and does not support `include`. |
| 111 | if !is_deepseek { |
| 112 | body["include"] = json!(["reasoning.encrypted_content"]); |
| 113 | } |
| 114 | |
| 115 | body |
| 116 | } |
| 117 | |
| 118 | impl DeepSeekClient { |
| 119 | /// Handle a streaming Responses API request for the OpenAI Codex provider. |
| 120 | pub(super) async fn handle_responses_stream( |
| 121 | &self, |
| 122 | prepared: &super::PreparedOutboundRequest, |
| 123 | ) -> Result<StreamEventBox> { |
| 124 | // Body, endpoint, and route shape all come from the shared |
| 125 | // prepared-request seam (`prepare_outbound_request`). |
| 126 | let body = &prepared.body; |
| 127 | let is_codex = prepared.endpoint.shape == super::RouteShape::CodexResponses; |
| 128 | let url = prepared.endpoint.url.clone(); |
| 129 | // The synthetic MessageStart below is emitted from inside the stream |
| 130 | // closure, which outlives `prepared`. Clone the wire model — the id |
| 131 | // actually placed on the body by the shared seam, after route |
| 132 | // remapping — rather than borrowing the request that no longer exists |
| 133 | // at this layer. |
| 134 | let wire_model = prepared.wire_model.clone(); |
| 135 | |
| 136 | // The bearer Authorization header is already installed as a default |
| 137 | // header on both the dual and the HTTP/1.1 twin client (resolved from |
| 138 | // the Codex OAuth access token), so it must not be set again here or |
| 139 | // it would be duplicated. The ChatGPT backend additionally requires |
| 140 | // the account id and the experimental Responses beta opt-in. |
| 141 | // |
| 142 | // The open itself goes through the shared stream-entry transport |
| 143 | // policy: bounded header wait, policy-selected client, and at most |
| 144 | // one HTTP/1.1 fallback retry on a classified H2 header stall. The |
| 145 | // pre-existing provider retry loop (rate limit / transient upstream) |
| 146 | // stays inside each open attempt, before any stream body exists. |
| 147 | let account_id = self.codex_account_id.clone(); |
| 148 | let request_body = |
| 149 | serde_json::to_vec(&body).context("Failed to serialize Responses API request body")?; |
| 150 | let open_req = super::stream_entry::StreamOpenRequest::new( |
| 151 | super::stream_entry::stream_open_timeout(), |
| 152 | self.stream_idle_timeout, |
| 153 | ); |
| 154 | let response = super::stream_entry::open_sse_response(&open_req, |policy| { |
| 155 | let url = url.clone(); |
| 156 | let account_id = account_id.clone(); |
| 157 | let request_body = request_body.clone(); |
| 158 | async move { |
| 159 | let client = super::stream_entry::client_for_policy( |
| 160 | &self.http_client, |
| 161 | self.http1_fallback_client(), |
| 162 | policy, |
| 163 | ); |
| 164 | self.send_with_retry(|| { |
| 165 | let mut builder = client |
| 166 | .post(&url) |
| 167 | .header("Content-Type", "application/json") |
| 168 | .header("Accept", "text/event-stream"); |
| 169 | if is_codex { |
| 170 | builder = builder |
| 171 | .header("OpenAI-Beta", "responses=experimental") |
| 172 | .header("originator", "codex_cli_rs"); |
| 173 | if let Some(account_id) = &account_id { |
| 174 | builder = builder.header("chatgpt-account-id", account_id); |
| 175 | } |
| 176 | } |
| 177 | builder.body(request_body.clone()) |
| 178 | }) |
| 179 | .await |
| 180 | .context("Responses API request failed") |
| 181 | } |
| 182 | }) |
| 183 | .await?; |
| 184 | |
| 185 | let status = response.status(); |
| 186 | crate::client::record_provider_response(self.api_provider, status.as_u16()); |
| 187 | if !status.is_success() { |
| 188 | let raw = bounded_error_text(response, ERROR_BODY_MAX_BYTES).await; |
| 189 | anyhow::bail!("Responses API error (HTTP {status}): {raw}"); |
| 190 | } |
| 191 | |
| 192 | let stream_idle_timeout = self.stream_idle_timeout; |
| 193 | let byte_stream = response.bytes_stream(); |
| 194 | |
| 195 | let stream = async_stream::stream! { |
| 196 | use futures_util::StreamExt; |
| 197 | |
| 198 | // Emit synthetic MessageStart. |
| 199 | yield Ok(StreamEvent::MessageStart { |
| 200 | message: MessageResponse { |
| 201 | id: String::new(), |
| 202 | r#type: "message".to_string(), |
| 203 | role: "assistant".to_string(), |
| 204 | content: vec![], |
| 205 | model: wire_model.clone(), |
| 206 | stop_reason: None, |
| 207 | stop_sequence: None, |
| 208 | container: None, |
| 209 | usage: Usage::default(), |
| 210 | }, |
| 211 | }); |
| 212 | |
| 213 | let mut current_block_index: Option<u32> = None; |
| 214 | // Whether reasoning text has already been emitted for the current |
| 215 | // reasoning block. Used to insert a paragraph break between |
| 216 | // consecutive summary parts, which the wire protocol delivers |
| 217 | // back-to-back with no separator. |
| 218 | let mut reasoning_text_emitted = false; |
| 219 | let mut saw_tool_call = false; |
| 220 | let mut usage_data: Option<Usage> = None; |
| 221 | // Raw byte buffer: decode only COMPLETE lines so a multi-byte |
| 222 | // UTF-8 char split across two network reads is never corrupted |
| 223 | // to U+FFFD (line boundaries are ASCII). Mirrors chat.rs. |
| 224 | let mut buffer: Vec<u8> = Vec::new(); |
| 225 | let mut done = false; |
| 226 | let mut content_block_counter: u32 = 0; |
| 227 | let stream_start = std::time::Instant::now(); |
| 228 | let mut last_chunk_at = std::time::Instant::now(); |
| 229 | let mut bytes_received: usize = 0; |
| 230 | |
| 231 | tokio::pin!(byte_stream); |
| 232 | |
| 233 | while !done { |
| 234 | let chunk = match tokio::time::timeout(stream_idle_timeout, byte_stream.next()).await { |
| 235 | Ok(Some(Ok(chunk))) => chunk, |
| 236 | Ok(Some(Err(e))) => { |
| 237 | yield Err(anyhow::anyhow!("Stream read error: {e}")); |
| 238 | return; |
| 239 | } |
| 240 | Ok(None) => break, |
| 241 | Err(_) => { |
| 242 | yield Err(anyhow::anyhow!(super::stream_entry::idle_timeout_message( |
| 243 | stream_idle_timeout, |
| 244 | bytes_received, |
| 245 | stream_start.elapsed(), |
| 246 | last_chunk_at.elapsed(), |
| 247 | ))); |
| 248 | return; |
| 249 | } |
| 250 | }; |
| 251 | |
| 252 | bytes_received += chunk.len(); |
| 253 | last_chunk_at = std::time::Instant::now(); |
| 254 | buffer.extend_from_slice(&chunk); |
| 255 | |
| 256 | // Process complete SSE lines. |
| 257 | while let Some(line) = super::take_sse_line(&mut buffer) { |
| 258 | |
| 259 | if line.is_empty() || line.starts_with(':') { |
| 260 | continue; |
| 261 | } |
| 262 | |
| 263 | if let Some(data) = super::extract_sse_data_value(&line) { |
| 264 | if data == "[DONE]" { |
| 265 | done = true; |
| 266 | break; |
| 267 | } |
| 268 | |
| 269 | let event: Value = match serde_json::from_str(data) { |
| 270 | Ok(v) => v, |
| 271 | Err(e) => { |
| 272 | logging::warn(format!( |
| 273 | "Failed to parse Responses SSE event: {e}" |
| 274 | )); |
| 275 | continue; |
| 276 | } |
| 277 | }; |
| 278 | |
| 279 | let event_type = |
| 280 | event.get("type").and_then(|t| t.as_str()).unwrap_or(""); |
| 281 | |
| 282 | match event_type { |
| 283 | "response.output_item.added" => { |
| 284 | if let Some(item) = event.get("item") { |
| 285 | let item_type = item |
| 286 | .get("type") |
| 287 | .and_then(|v| v.as_str()) |
| 288 | .unwrap_or(""); |
| 289 | |
| 290 | match item_type { |
| 291 | "message" => { |
| 292 | content_block_counter += 1; |
| 293 | yield Ok(StreamEvent::ContentBlockStart { |
| 294 | index: content_block_counter - 1, |
| 295 | content_block: ContentBlockStart::Text { |
| 296 | text: String::new(), |
| 297 | }, |
| 298 | }); |
| 299 | current_block_index = |
| 300 | Some(content_block_counter - 1); |
| 301 | } |
| 302 | "function_call" => { |
| 303 | let call_id = item |
| 304 | .get("call_id") |
| 305 | .and_then(|v| v.as_str()) |
| 306 | .unwrap_or("") |
| 307 | .to_string(); |
| 308 | let item_id = item |
| 309 | .get("id") |
| 310 | .and_then(|v| v.as_str()) |
| 311 | .unwrap_or("") |
| 312 | .to_string(); |
| 313 | let name = item |
| 314 | .get("name") |
| 315 | .and_then(|v| v.as_str()) |
| 316 | .unwrap_or("") |
| 317 | .to_string(); |
| 318 | saw_tool_call = true; |
| 319 | // call_id and item_id are folded |
| 320 | // into a composite tool-use id so |
| 321 | // the function_call_output can be |
| 322 | // routed back to the right call. |
| 323 | let composite_id = |
| 324 | format!("{call_id}|{item_id}"); |
| 325 | content_block_counter += 1; |
| 326 | yield Ok(StreamEvent::ContentBlockStart { |
| 327 | index: content_block_counter - 1, |
| 328 | content_block: |
| 329 | ContentBlockStart::ToolUse { |
| 330 | id: composite_id, |
| 331 | name: from_api_tool_name(&name), |
| 332 | input: json!({}), |
| 333 | caller: None, |
| 334 | }, |
| 335 | }); |
| 336 | current_block_index = |
| 337 | Some(content_block_counter - 1); |
| 338 | } |
| 339 | "reasoning" => { |
| 340 | reasoning_text_emitted = false; |
| 341 | content_block_counter += 1; |
| 342 | yield Ok(StreamEvent::ContentBlockStart { |
| 343 | index: content_block_counter - 1, |
| 344 | content_block: |
| 345 | ContentBlockStart::Thinking { |
| 346 | thinking: String::new(), |
| 347 | }, |
| 348 | }); |
| 349 | current_block_index = |
| 350 | Some(content_block_counter - 1); |
| 351 | } |
| 352 | // DeepSeek can run server-side web |
| 353 | // search on this route, but Codewhale |
| 354 | // does not yet replay `web_search_call` |
| 355 | // items or their citations (the |
| 356 | // offering keeps |
| 357 | // `server_side_web_search: Unknown`). |
| 358 | // Surface a visible notice instead of |
| 359 | // dropping the item silently so the |
| 360 | // user is not handed an ungrounded |
| 361 | // answer with no explanation. |
| 362 | "web_search_call" => { |
| 363 | content_block_counter += 1; |
| 364 | yield Ok(StreamEvent::ContentBlockStart { |
| 365 | index: content_block_counter - 1, |
| 366 | content_block: |
| 367 | ContentBlockStart::Text { |
| 368 | text: "[Web search ran server-side; results are not replayed on this route.]".to_string(), |
| 369 | }, |
| 370 | }); |
| 371 | current_block_index = |
| 372 | Some(content_block_counter - 1); |
| 373 | } |
| 374 | _ => {} |
| 375 | } |
| 376 | } |
| 377 | } |
| 378 | "response.output_text.delta" => { |
| 379 | if let Some(delta_text) = |
| 380 | event.get("delta").and_then(|d| d.as_str()) |
| 381 | && let Some(idx) = current_block_index |
| 382 | { |
| 383 | yield Ok(StreamEvent::ContentBlockDelta { |
| 384 | index: idx, |
| 385 | delta: Delta::TextDelta { |
| 386 | text: delta_text.to_string(), |
| 387 | }, |
| 388 | }); |
| 389 | } |
| 390 | } |
| 391 | "response.function_call_arguments.delta" => { |
| 392 | if let Some(delta_text) = |
| 393 | event.get("delta").and_then(|d| d.as_str()) |
| 394 | && let Some(idx) = current_block_index |
| 395 | { |
| 396 | yield Ok(StreamEvent::ContentBlockDelta { |
| 397 | index: idx, |
| 398 | delta: Delta::InputJsonDelta { |
| 399 | partial_json: delta_text.to_string(), |
| 400 | }, |
| 401 | }); |
| 402 | } |
| 403 | } |
| 404 | "response.reasoning_summary_text.delta" |
| 405 | | "response.reasoning_text.delta" => { |
| 406 | if let Some(delta_text) = |
| 407 | event.get("delta").and_then(|d| d.as_str()) |
| 408 | && let Some(idx) = current_block_index |
| 409 | { |
| 410 | if !delta_text.is_empty() { |
| 411 | reasoning_text_emitted = true; |
| 412 | } |
| 413 | yield Ok(StreamEvent::ContentBlockDelta { |
| 414 | index: idx, |
| 415 | delta: Delta::ThinkingDelta { |
| 416 | thinking: delta_text.to_string(), |
| 417 | }, |
| 418 | }); |
| 419 | } |
| 420 | } |
| 421 | "response.reasoning_summary_part.added" => { |
| 422 | // Consecutive summary parts arrive with no |
| 423 | // separator in the text deltas, so without a |
| 424 | // boundary they concatenate as |
| 425 | // "…done.**Next Phase**…". Insert a paragraph |
| 426 | // break before every part after the first. |
| 427 | if reasoning_text_emitted |
| 428 | && let Some(idx) = current_block_index |
| 429 | { |
| 430 | yield Ok(StreamEvent::ContentBlockDelta { |
| 431 | index: idx, |
| 432 | delta: Delta::ThinkingDelta { |
| 433 | thinking: "\n\n".to_string(), |
| 434 | }, |
| 435 | }); |
| 436 | } |
| 437 | } |
| 438 | "response.output_item.done" => { |
| 439 | if let Some(idx) = current_block_index { |
| 440 | yield Ok(StreamEvent::ContentBlockStop { index: idx }); |
| 441 | current_block_index = None; |
| 442 | } |
| 443 | } |
| 444 | "response.completed" | "response.incomplete" => { |
| 445 | if let Some(resp) = event.get("response") { |
| 446 | if let Some(usage_val) = resp.get("usage") { |
| 447 | usage_data = |
| 448 | Some(parse_responses_usage(usage_val)); |
| 449 | } |
| 450 | let status = resp |
| 451 | .get("status") |
| 452 | .and_then(|s| s.as_str()) |
| 453 | .unwrap_or("completed"); |
| 454 | let stop_reason = match status { |
| 455 | "completed" if saw_tool_call => "tool_use", |
| 456 | "completed" => "end_turn", |
| 457 | "incomplete" => "max_tokens", |
| 458 | _ => "end_turn", |
| 459 | }; |
| 460 | yield Ok(StreamEvent::MessageDelta { |
| 461 | delta: MessageDelta { |
| 462 | stop_reason: Some(stop_reason.to_string()), |
| 463 | stop_sequence: None, |
| 464 | }, |
| 465 | usage: usage_data.take(), |
| 466 | }); |
| 467 | } |
| 468 | // DeepSeek terminates semantic Responses |
| 469 | // streams with this event and deliberately does |
| 470 | // not send `data: [DONE]`. |
| 471 | done = true; |
| 472 | } |
| 473 | "error" | "response.failed" => { |
| 474 | let (code, msg) = responses_event_error_details(&event); |
| 475 | yield Err(anyhow::anyhow!( |
| 476 | "Responses API error [{code}]: {msg}" |
| 477 | )); |
| 478 | return; |
| 479 | } |
| 480 | _ => { |
| 481 | // Ignore unknown event types. |
| 482 | } |
| 483 | } |
| 484 | } |
| 485 | } |
| 486 | } |
| 487 | |
| 488 | // Emit MessageStop. |
| 489 | yield Ok(StreamEvent::MessageStop); |
| 490 | }; |
| 491 | |
| 492 | Ok(Box::pin(stream)) |
| 493 | } |
| 494 | |
| 495 | /// Non-streaming Responses request: drive the streaming handler and fold |
| 496 | /// its events into a single `MessageResponse`. |
| 497 | /// |
| 498 | /// The ChatGPT Codex backend only serves streaming responses, so the |
| 499 | /// non-streaming entry point (`create_message`, used by `exec`) reuses the |
| 500 | /// same wire path as the interactive stream rather than a second request |
| 501 | /// shape. |
| 502 | pub(super) async fn handle_responses_message( |
| 503 | &self, |
| 504 | prepared: &super::PreparedOutboundRequest, |
| 505 | ) -> Result<MessageResponse> { |
| 506 | use futures_util::StreamExt; |
| 507 | |
| 508 | let model = prepared.wire_model.clone(); |
| 509 | let mut stream = self.handle_responses_stream(prepared).await?; |
| 510 | |
| 511 | let mut response = MessageResponse { |
| 512 | id: String::new(), |
| 513 | r#type: "message".to_string(), |
| 514 | role: "assistant".to_string(), |
| 515 | content: Vec::new(), |
| 516 | model, |
| 517 | stop_reason: None, |
| 518 | stop_sequence: None, |
| 519 | container: None, |
| 520 | usage: Usage::default(), |
| 521 | }; |
| 522 | // Accumulated tool-call argument JSON, parallel to `response.content`. |
| 523 | let mut tool_args: Vec<String> = Vec::new(); |
| 524 | |
| 525 | while let Some(event) = stream.next().await { |
| 526 | match event? { |
| 527 | StreamEvent::MessageStart { message } => { |
| 528 | response.id = message.id; |
| 529 | response.usage = message.usage; |
| 530 | } |
| 531 | StreamEvent::ContentBlockStart { content_block, .. } => { |
| 532 | let block = match content_block { |
| 533 | ContentBlockStart::Text { text } => ContentBlock::Text { |
| 534 | text, |
| 535 | cache_control: None, |
| 536 | }, |
| 537 | ContentBlockStart::Thinking { thinking } => ContentBlock::Thinking { |
| 538 | thinking, |
| 539 | signature: None, |
| 540 | }, |
| 541 | ContentBlockStart::ToolUse { |
| 542 | id, |
| 543 | name, |
| 544 | input, |
| 545 | caller, |
| 546 | } => ContentBlock::ToolUse { |
| 547 | id, |
| 548 | name, |
| 549 | input, |
| 550 | caller, |
| 551 | }, |
| 552 | ContentBlockStart::ServerToolUse { id, name, input } => { |
| 553 | ContentBlock::ServerToolUse { id, name, input } |
| 554 | } |
| 555 | }; |
| 556 | response.content.push(block); |
| 557 | tool_args.push(String::new()); |
| 558 | } |
| 559 | StreamEvent::ContentBlockDelta { index, delta } => { |
| 560 | let i = index as usize; |
| 561 | match delta { |
| 562 | Delta::TextDelta { text } => { |
| 563 | if let Some(ContentBlock::Text { text: existing, .. }) = |
| 564 | response.content.get_mut(i) |
| 565 | { |
| 566 | existing.push_str(&text); |
| 567 | } |
| 568 | } |
| 569 | Delta::ThinkingDelta { thinking } => { |
| 570 | if let Some(ContentBlock::Thinking { |
| 571 | thinking: existing, .. |
| 572 | }) = response.content.get_mut(i) |
| 573 | { |
| 574 | existing.push_str(&thinking); |
| 575 | } |
| 576 | } |
| 577 | Delta::InputJsonDelta { partial_json } => { |
| 578 | if let Some(buf) = tool_args.get_mut(i) { |
| 579 | buf.push_str(&partial_json); |
| 580 | } |
| 581 | } |
| 582 | Delta::SignatureDelta { .. } => { |
| 583 | // Anthropic-native signature deltas never occur on |
| 584 | // the Responses bridge (#3014). |
| 585 | } |
| 586 | } |
| 587 | } |
| 588 | StreamEvent::ContentBlockStop { index } => { |
| 589 | let i = index as usize; |
| 590 | if let Some(buf) = tool_args.get(i) |
| 591 | && !buf.trim().is_empty() |
| 592 | && let Ok(parsed) = serde_json::from_str::<Value>(buf) |
| 593 | && let Some(ContentBlock::ToolUse { input, .. }) = |
| 594 | response.content.get_mut(i) |
| 595 | { |
| 596 | *input = parsed; |
| 597 | } |
| 598 | } |
| 599 | StreamEvent::MessageDelta { delta, usage } => { |
| 600 | if let Some(stop_reason) = delta.stop_reason { |
| 601 | response.stop_reason = Some(stop_reason); |
| 602 | } |
| 603 | if let Some(usage) = usage { |
| 604 | response.usage = usage; |
| 605 | } |
| 606 | } |
| 607 | StreamEvent::MessageStop => break, |
| 608 | _ => {} |
| 609 | } |
| 610 | } |
| 611 | |
| 612 | Ok(response) |
| 613 | } |
| 614 | } |
| 615 | |
| 616 | /// Convert CodeWhale messages to Responses API input items. |
| 617 | fn convert_messages_to_responses_input(request: &MessageRequest, is_deepseek: bool) -> Vec<Value> { |
| 618 | let mut items = Vec::new(); |
| 619 | |
| 620 | for msg in &request.messages { |
| 621 | match msg.role.as_str() { |
| 622 | "user" => { |
| 623 | let mut content_items = Vec::new(); |
| 624 | for block in &msg.content { |
| 625 | match block { |
| 626 | ContentBlock::Text { text, .. } => { |
| 627 | content_items.push(json!({ |
| 628 | "type": "input_text", |
| 629 | "text": text, |
| 630 | })); |
| 631 | } |
| 632 | ContentBlock::ImageUrl { image_url } => { |
| 633 | content_items.push(json!({ |
| 634 | "type": "input_image", |
| 635 | "image_url": image_url.url, |
| 636 | })); |
| 637 | } |
| 638 | ContentBlock::ToolResult { |
| 639 | tool_use_id, |
| 640 | content, |
| 641 | .. |
| 642 | } => { |
| 643 | if !content_items.is_empty() { |
| 644 | items.push(json!({ |
| 645 | "type": "message", |
| 646 | "role": "user", |
| 647 | "content": content_items, |
| 648 | })); |
| 649 | content_items = Vec::new(); |
| 650 | } |
| 651 | let (call_id, _item_id) = parse_tool_use_id(tool_use_id); |
| 652 | items.push(json!({ |
| 653 | "type": "function_call_output", |
| 654 | "call_id": call_id, |
| 655 | "output": content, |
| 656 | })); |
| 657 | } |
| 658 | _ => {} |
| 659 | } |
| 660 | } |
| 661 | if !content_items.is_empty() { |
| 662 | items.push(json!({ |
| 663 | "type": "message", |
| 664 | "role": "user", |
| 665 | "content": content_items, |
| 666 | })); |
| 667 | } |
| 668 | } |
| 669 | "assistant" | crate::models::INTERRUPTED_ASSISTANT_ROLE => { |
| 670 | for block in &msg.content { |
| 671 | match block { |
| 672 | ContentBlock::Text { text, .. } => { |
| 673 | let text = if msg.role == crate::models::INTERRUPTED_ASSISTANT_ROLE { |
| 674 | format!( |
| 675 | "{}{}", |
| 676 | crate::models::INTERRUPTED_ASSISTANT_CONTEXT_PREFIX, |
| 677 | text |
| 678 | ) |
| 679 | } else { |
| 680 | text.clone() |
| 681 | }; |
| 682 | items.push(json!({ |
| 683 | "type": "message", |
| 684 | "role": "assistant", |
| 685 | "content": [{ |
| 686 | "type": "output_text", |
| 687 | "text": text, |
| 688 | }], |
| 689 | })); |
| 690 | } |
| 691 | ContentBlock::ToolUse { |
| 692 | id, name, input, .. |
| 693 | } => { |
| 694 | let (call_id, _item_id) = parse_tool_use_id(id); |
| 695 | items.push(json!({ |
| 696 | "type": "function_call", |
| 697 | "call_id": call_id, |
| 698 | "name": to_api_tool_name(name), |
| 699 | "arguments": serde_json::to_string(input).unwrap_or_default(), |
| 700 | })); |
| 701 | } |
| 702 | ContentBlock::Thinking { thinking, .. } => { |
| 703 | items.push(if is_deepseek { |
| 704 | json!({ |
| 705 | "type": "reasoning", |
| 706 | "content": [{ |
| 707 | "type": "reasoning_text", |
| 708 | "text": thinking, |
| 709 | }], |
| 710 | }) |
| 711 | } else { |
| 712 | json!({ |
| 713 | "type": "reasoning", |
| 714 | "summary": [{ |
| 715 | "type": "summary_text", |
| 716 | "text": thinking, |
| 717 | }], |
| 718 | }) |
| 719 | }); |
| 720 | } |
| 721 | _ => {} |
| 722 | } |
| 723 | } |
| 724 | } |
| 725 | "tool" => { |
| 726 | for block in &msg.content { |
| 727 | if let ContentBlock::ToolResult { |
| 728 | tool_use_id, |
| 729 | content, |
| 730 | .. |
| 731 | } = block |
| 732 | { |
| 733 | let (call_id, _item_id) = parse_tool_use_id(tool_use_id); |
| 734 | items.push(json!({ |
| 735 | "type": "function_call_output", |
| 736 | "call_id": call_id, |
| 737 | "output": content, |
| 738 | })); |
| 739 | } |
| 740 | } |
| 741 | } |
| 742 | _ => {} |
| 743 | } |
| 744 | } |
| 745 | |
| 746 | items |
| 747 | } |
| 748 | |
| 749 | /// Convert a CodeWhale tool definition to a Responses API function tool. |
| 750 | fn tool_to_responses_function(tool: &Tool) -> Value { |
| 751 | let mut parameters = tool.input_schema.clone(); |
| 752 | let constraint_note = schema_sanitize::sanitize_for_responses(&mut parameters); |
| 753 | let description = match constraint_note { |
| 754 | Some(note) if tool.description.trim().is_empty() => note, |
| 755 | Some(note) => format!("{}\n\n{}", tool.description.trim(), note), |
| 756 | None => tool.description.clone(), |
| 757 | }; |
| 758 | json!({ |
| 759 | "type": "function", |
| 760 | "name": to_api_tool_name(&tool.name), |
| 761 | "description": description, |
| 762 | "parameters": parameters, |
| 763 | "strict": false, |
| 764 | }) |
| 765 | } |
| 766 | |
| 767 | fn codex_responses_reasoning_effort(raw: &str) -> Option<&'static str> { |
| 768 | match raw.trim().to_ascii_lowercase().as_str() { |
| 769 | "off" | "disabled" | "none" | "false" => Some("low"), |
| 770 | "minimal" => Some("low"), |
| 771 | "low" => Some("low"), |
| 772 | "high" => Some("high"), |
| 773 | "xhigh" | "max" | "maximum" | "ultracode" => Some("xhigh"), |
| 774 | _ => Some("medium"), |
| 775 | } |
| 776 | } |
| 777 | |
| 778 | fn responses_reasoning_effort(raw: &str, is_deepseek: bool) -> Option<&'static str> { |
| 779 | if !is_deepseek { |
| 780 | return codex_responses_reasoning_effort(raw); |
| 781 | } |
| 782 | match raw.trim().to_ascii_lowercase().as_str() { |
| 783 | // DeepSeek's Responses API documents `reasoning.effort: "none"` as the |
| 784 | // thinking-off tier; the picker offers Off for flash, so it must stay |
| 785 | // "none" on the wire instead of collapsing into low (still thinking). |
| 786 | "off" | "disabled" | "none" | "false" => Some("none"), |
| 787 | "minimal" | "low" => Some("low"), |
| 788 | "xhigh" | "max" | "maximum" | "ultracode" => Some("max"), |
| 789 | // DeepSeek maps both low and medium to its normal high tier in |
| 790 | // thinking mode. Preserve explicit low for Codex compatibility, and |
| 791 | // normalize every remaining/automatic tier to a documented value. |
| 792 | _ => Some("high"), |
| 793 | } |
| 794 | } |
| 795 | |
| 796 | fn responses_event_error_details(event: &Value) -> (String, String) { |
| 797 | let event_type = string_at(event, "/type").unwrap_or("error"); |
| 798 | let code = first_string_at( |
| 799 | event, |
| 800 | &[ |
| 801 | "/code", |
| 802 | "/error/code", |
| 803 | "/response/error/code", |
| 804 | "/response/incomplete_details/reason", |
| 805 | "/response/status", |
| 806 | ], |
| 807 | ) |
| 808 | .unwrap_or("unknown"); |
| 809 | let message = first_string_at( |
| 810 | event, |
| 811 | &[ |
| 812 | "/message", |
| 813 | "/error/message", |
| 814 | "/response/error/message", |
| 815 | "/response/incomplete_details/reason", |
| 816 | ], |
| 817 | ) |
| 818 | .map_or_else( |
| 819 | || format!("{event_type} event received"), |
| 820 | |message| { |
| 821 | if message == code && event_type == "response.incomplete" { |
| 822 | format!("response incomplete: {message}") |
| 823 | } else { |
| 824 | message.to_string() |
| 825 | } |
| 826 | }, |
| 827 | ); |
| 828 | (code.to_string(), message) |
| 829 | } |
| 830 | |
| 831 | fn first_string_at<'a>(value: &'a Value, paths: &[&str]) -> Option<&'a str> { |
| 832 | paths.iter().find_map(|path| string_at(value, path)) |
| 833 | } |
| 834 | |
| 835 | fn string_at<'a>(value: &'a Value, path: &str) -> Option<&'a str> { |
| 836 | value.pointer(path).and_then(Value::as_str).and_then(|s| { |
| 837 | let trimmed = s.trim(); |
| 838 | (!trimmed.is_empty()).then_some(trimmed) |
| 839 | }) |
| 840 | } |
| 841 | |
| 842 | /// Parse a composite tool_use_id back to (call_id, item_id). |
| 843 | /// Composite format: "call_id|item_id" |
| 844 | fn parse_tool_use_id(id: &str) -> (String, String) { |
| 845 | if let Some(pipe_pos) = id.find('|') { |
| 846 | (id[..pipe_pos].to_string(), id[pipe_pos + 1..].to_string()) |
| 847 | } else { |
| 848 | (id.to_string(), String::new()) |
| 849 | } |
| 850 | } |
| 851 | |
| 852 | /// Parse usage from a Responses API usage object. |
| 853 | fn parse_responses_usage(val: &Value) -> Usage { |
| 854 | let input = val |
| 855 | .get("input_tokens") |
| 856 | .and_then(|v| v.as_u64()) |
| 857 | .map_or(0, super::saturating_u32); |
| 858 | let output = val |
| 859 | .get("output_tokens") |
| 860 | .and_then(|v| v.as_u64()) |
| 861 | .map_or(0, super::saturating_u32); |
| 862 | // Cache telemetry arrives in two dialects. DeepSeek's Responses payload |
| 863 | // uses the same top-level `prompt_cache_hit_tokens` / |
| 864 | // `prompt_cache_miss_tokens` fields as its Chat-Completions endpoint, |
| 865 | // while OpenAI-style payloads nest `cached_tokens` under |
| 866 | // `input_tokens_details`. Prefer the top-level hit, falling back to the |
| 867 | // nested form when the payload only reports that. |
| 868 | let nested_cached_tokens = val |
| 869 | .get("input_tokens_details") |
| 870 | .and_then(|d| d.get("cached_tokens")) |
| 871 | .and_then(|v| v.as_u64()); |
| 872 | let prompt_cache_hit_tokens = val |
| 873 | .get("prompt_cache_hit_tokens") |
| 874 | .and_then(|v| v.as_u64()) |
| 875 | .or(nested_cached_tokens) |
| 876 | .map(super::saturating_u32); |
| 877 | // DeepSeek reports the miss explicitly; otherwise mirror the |
| 878 | // Chat-Completions parser: derive the miss as input minus the cached hit |
| 879 | // when the payload reported cached input tokens. Responses nests |
| 880 | // reasoning under `output_tokens_details` (not `completion_tokens_details`). |
| 881 | let prompt_cache_miss_tokens = val |
| 882 | .get("prompt_cache_miss_tokens") |
| 883 | .and_then(|v| v.as_u64()) |
| 884 | .map(super::saturating_u32) |
| 885 | .or_else(|| prompt_cache_hit_tokens.map(|hit| input.saturating_sub(hit))); |
| 886 | // Cache-creation tokens, kept as their own class so pricing can apply the |
| 887 | // write rate where the provider publishes one. DeepSeek-style payloads |
| 888 | // nest these under `input_tokens_details`; accept a top-level spelling |
| 889 | // too for providers that flatten the object. |
| 890 | let prompt_cache_write_tokens = val |
| 891 | .get("prompt_cache_write_tokens") |
| 892 | .and_then(|v| v.as_u64()) |
| 893 | .or_else(|| { |
| 894 | val.get("input_tokens_details") |
| 895 | .and_then(|d| d.get("cache_write_tokens")) |
| 896 | .and_then(|v| v.as_u64()) |
| 897 | }) |
| 898 | .map(super::saturating_u32); |
| 899 | // `output_tokens` is already the total billable completion count, with |
| 900 | // reasoning a subset of it. A payload reporting more reasoning than output |
| 901 | // violates that, so the value is rejected as invalid telemetry rather than |
| 902 | // being trusted or turned into extra billable output (#4318). |
| 903 | let reasoning_tokens = val |
| 904 | .get("output_tokens_details") |
| 905 | .and_then(|d| d.get("reasoning_tokens")) |
| 906 | .and_then(|v| v.as_u64()) |
| 907 | .map(super::saturating_u32) |
| 908 | .filter(|reasoning| *reasoning <= output); |
| 909 | // `input_tokens` stays the provider-reported *total* (cache-hit + miss + |
| 910 | // write + uncategorized): `token_usage_for_pricing` partitions it into |
| 911 | // billable classes and the context budget measures the window with it. |
| 912 | // Reducing it here to miss-only would double-subtract at those surfaces. |
| 913 | Usage { |
| 914 | input_tokens: input, |
| 915 | output_tokens: output, |
| 916 | prompt_cache_hit_tokens, |
| 917 | prompt_cache_miss_tokens, |
| 918 | prompt_cache_write_tokens, |
| 919 | reasoning_tokens, |
| 920 | reasoning_replay_tokens: None, |
| 921 | server_tool_use: None, |
| 922 | } |
| 923 | } |
| 924 | |
| 925 | #[cfg(test)] |
| 926 | mod tests; |
| 927 |