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