| 1 | //! Chat Completions API helpers for DeepSeek's OpenAI-compatible endpoint. |
| 2 | //! |
| 3 | //! This is the production code path. Streaming (`create_message_stream`), |
| 4 | //! request building (`build_chat_messages*`), and SSE parsing (`parse_sse_chunk`) |
| 5 | //! all live here. |
| 6 | |
| 7 | use std::collections::HashSet; |
| 8 | use std::pin::Pin; |
| 9 | use std::time::Duration; |
| 10 | |
| 11 | use anyhow::{Context, Result}; |
| 12 | use serde_json::{Value, json}; |
| 13 | use tokio::time::timeout as tokio_timeout; |
| 14 | |
| 15 | /// Default idle timeout for SSE stream reads (300 seconds = 5 minutes). |
| 16 | /// After this period with no data, the stream is considered stalled and |
| 17 | /// yields a recoverable error so the caller can retry. |
| 18 | const DEFAULT_STREAM_IDLE_TIMEOUT: Duration = Duration::from_secs(300); |
| 19 | |
| 20 | /// Default timeout for the initial streaming response headers. |
| 21 | /// |
| 22 | /// `doctor` uses a bounded non-streaming request, but normal TUI turns first |
| 23 | /// wait for the SSE response to open. On some Windows/proxy paths that wait can |
| 24 | /// hang before any stream chunk exists, leaving the UI stuck at "Working...". |
| 25 | const DEFAULT_STREAM_OPEN_TIMEOUT: Duration = Duration::from_secs(45); |
| 26 | |
| 27 | /// Reads `DEEPSEEK_STREAM_OPEN_TIMEOUT_SECS` as a bounded override for the |
| 28 | /// response-header wait. This is intentionally shorter than the per-chunk idle |
| 29 | /// timeout because it only covers connection setup and upstream header return, |
| 30 | /// not model thinking time after streaming has started. |
| 31 | fn stream_open_timeout() -> Duration { |
| 32 | stream_open_timeout_from_env( |
| 33 | std::env::var("DEEPSEEK_STREAM_OPEN_TIMEOUT_SECS") |
| 34 | .ok() |
| 35 | .as_deref(), |
| 36 | ) |
| 37 | } |
| 38 | |
| 39 | fn stream_open_timeout_from_env(value: Option<&str>) -> Duration { |
| 40 | let secs = value |
| 41 | .and_then(|v| v.parse::<u64>().ok()) |
| 42 | .unwrap_or(DEFAULT_STREAM_OPEN_TIMEOUT.as_secs()) |
| 43 | .clamp(5, 300); |
| 44 | Duration::from_secs(secs) |
| 45 | } |
| 46 | |
| 47 | /// Reads the `DEEPSEEK_STREAM_IDLE_TIMEOUT_SECS` env var, falling back to |
| 48 | /// the default 300s. The parsed value is clamped to [1, 3600] seconds. |
| 49 | fn stream_idle_timeout() -> Duration { |
| 50 | let secs = std::env::var("DEEPSEEK_STREAM_IDLE_TIMEOUT_SECS") |
| 51 | .ok() |
| 52 | .and_then(|v| v.parse::<u64>().ok()) |
| 53 | .unwrap_or(DEFAULT_STREAM_IDLE_TIMEOUT.as_secs()) |
| 54 | .clamp(1, 3600); |
| 55 | Duration::from_secs(secs) |
| 56 | } |
| 57 | |
| 58 | use crate::llm_client::StreamEventBox; |
| 59 | use crate::logging; |
| 60 | use crate::models::{ |
| 61 | ContentBlock, ContentBlockStart, Delta, Message, MessageDelta, MessageRequest, MessageResponse, |
| 62 | StreamEvent, SystemPrompt, Tool, ToolCaller, Usage, |
| 63 | }; |
| 64 | |
| 65 | use super::{ |
| 66 | DeepSeekClient, ERROR_BODY_MAX_BYTES, SSE_BACKPRESSURE_HIGH_WATERMARK, |
| 67 | SSE_BACKPRESSURE_SLEEP_MS, SSE_MAX_LINES_PER_CHUNK, acquire_stream_buffer, api_url, |
| 68 | apply_reasoning_effort, bounded_error_text, from_api_tool_name, parse_usage, |
| 69 | release_stream_buffer, system_to_instructions, to_api_tool_name, |
| 70 | }; |
| 71 | |
| 72 | impl DeepSeekClient { |
| 73 | pub(super) async fn create_message_chat( |
| 74 | &self, |
| 75 | request: &MessageRequest, |
| 76 | ) -> Result<MessageResponse> { |
| 77 | let messages = build_chat_messages_for_request(request); |
| 78 | let mut body = json!({ |
| 79 | "model": request.model, |
| 80 | "messages": messages, |
| 81 | "max_tokens": request.max_tokens, |
| 82 | }); |
| 83 | |
| 84 | if let Some(temperature) = request.temperature { |
| 85 | body["temperature"] = json!(temperature); |
| 86 | } |
| 87 | if let Some(top_p) = request.top_p { |
| 88 | body["top_p"] = json!(top_p); |
| 89 | } |
| 90 | if let Some(tools) = request.tools.as_ref() { |
| 91 | body["tools"] = json!(tools.iter().map(tool_to_chat).collect::<Vec<_>>()); |
| 92 | } |
| 93 | if let Some(choice) = request.tool_choice.as_ref() |
| 94 | && let Some(mapped) = map_tool_choice_for_chat(choice) |
| 95 | { |
| 96 | body["tool_choice"] = mapped; |
| 97 | } |
| 98 | apply_reasoning_effort( |
| 99 | &mut body, |
| 100 | request.reasoning_effort.as_deref(), |
| 101 | self.api_provider, |
| 102 | ); |
| 103 | |
| 104 | let url = api_url(&self.base_url, "chat/completions"); |
| 105 | let open_timeout = stream_open_timeout(); |
| 106 | let response = match tokio_timeout( |
| 107 | open_timeout, |
| 108 | self.send_with_retry(|| self.http_client.post(&url).json(&body)), |
| 109 | ) |
| 110 | .await |
| 111 | { |
| 112 | Ok(result) => result?, |
| 113 | Err(_elapsed) => { |
| 114 | anyhow::bail!( |
| 115 | "SSE stream request did not receive response headers after {}s. \ |
| 116 | `deepseek doctor` can still pass when non-streaming requests work; \ |
| 117 | on Windows or proxy networks, try `DEEPSEEK_FORCE_HTTP1=1` and rerun `deepseek`.", |
| 118 | open_timeout.as_secs() |
| 119 | ); |
| 120 | } |
| 121 | }; |
| 122 | |
| 123 | let status = response.status(); |
| 124 | if !status.is_success() { |
| 125 | let error_text = bounded_error_text(response, ERROR_BODY_MAX_BYTES).await; |
| 126 | anyhow::bail!("Failed to call DeepSeek Chat API: HTTP {status}: {error_text}"); |
| 127 | } |
| 128 | |
| 129 | let response_text = response.text().await.unwrap_or_default(); |
| 130 | let value: Value = |
| 131 | serde_json::from_str(&response_text).context("Failed to parse Chat API JSON")?; |
| 132 | parse_chat_message(&value) |
| 133 | } |
| 134 | } |
| 135 | |
| 136 | impl DeepSeekClient { |
| 137 | pub(super) async fn handle_chat_completion_stream( |
| 138 | &self, |
| 139 | request: MessageRequest, |
| 140 | ) -> Result<StreamEventBox> { |
| 141 | // Try true SSE streaming via chat completions (widely supported) |
| 142 | let messages = build_chat_messages_for_request(&request); |
| 143 | let mut body = json!({ |
| 144 | "model": request.model, |
| 145 | "messages": messages, |
| 146 | "max_tokens": request.max_tokens, |
| 147 | "stream": true, |
| 148 | "stream_options": { |
| 149 | "include_usage": true |
| 150 | }, |
| 151 | }); |
| 152 | |
| 153 | if let Some(temperature) = request.temperature { |
| 154 | body["temperature"] = json!(temperature); |
| 155 | } |
| 156 | if let Some(top_p) = request.top_p { |
| 157 | body["top_p"] = json!(top_p); |
| 158 | } |
| 159 | if let Some(tools) = request.tools.as_ref() { |
| 160 | body["tools"] = json!(tools.iter().map(tool_to_chat).collect::<Vec<_>>()); |
| 161 | } |
| 162 | if let Some(choice) = request.tool_choice.as_ref() |
| 163 | && let Some(mapped) = map_tool_choice_for_chat(choice) |
| 164 | { |
| 165 | body["tool_choice"] = mapped; |
| 166 | } |
| 167 | apply_reasoning_effort( |
| 168 | &mut body, |
| 169 | request.reasoning_effort.as_deref(), |
| 170 | self.api_provider, |
| 171 | ); |
| 172 | |
| 173 | // Bulletproof final sanitizer: walk the wire payload and force |
| 174 | // `reasoning_content` onto any assistant message that has tool_calls |
| 175 | // but no reasoning_content. DeepSeek's thinking-mode API rejects |
| 176 | // such messages with a 400. This is the last line of defense after |
| 177 | // engine-side and build-side substitution; if either upstream path |
| 178 | // misses a case (e.g. a session restored from disk, a sub-agent |
| 179 | // adding messages directly, or a cached prefix mismatch), this pass |
| 180 | // still produces a valid request. |
| 181 | let replay_input_tokens = sanitize_thinking_mode_messages( |
| 182 | &mut body, |
| 183 | &request.model, |
| 184 | request.reasoning_effort.as_deref(), |
| 185 | ); |
| 186 | |
| 187 | let url = api_url(&self.base_url, "chat/completions"); |
| 188 | let response = self |
| 189 | .send_with_retry(|| self.http_client.post(&url).json(&body)) |
| 190 | .await?; |
| 191 | |
| 192 | let status = response.status(); |
| 193 | if !status.is_success() { |
| 194 | let error_text = bounded_error_text(response, ERROR_BODY_MAX_BYTES).await; |
| 195 | // If DeepSeek rejected for missing reasoning_content despite the |
| 196 | // sanitizer, dump the offending indices so we can diagnose where |
| 197 | // they came from on the next failure. |
| 198 | if error_text.contains("reasoning_content") { |
| 199 | log_thinking_mode_violations(&body); |
| 200 | } |
| 201 | anyhow::bail!("SSE stream request failed: HTTP {status}: {error_text}"); |
| 202 | } |
| 203 | |
| 204 | let model = request.model.clone(); |
| 205 | |
| 206 | // Capture transport-shape headers before we consume `response` into |
| 207 | // `bytes_stream()`. They are surfaced in the decode-error log path so |
| 208 | // we can tell HTTP/2 RST_STREAM from chunked-encoding corruption from |
| 209 | // gzip-compressor failure when investigating #103. |
| 210 | let response_headers = format_stream_headers(response.headers()); |
| 211 | let byte_stream = response.bytes_stream(); |
| 212 | |
| 213 | let stream = async_stream::stream! { |
| 214 | use futures_util::StreamExt; |
| 215 | |
| 216 | // Emit a synthetic MessageStart |
| 217 | yield Ok(StreamEvent::MessageStart { |
| 218 | message: MessageResponse { |
| 219 | id: String::new(), |
| 220 | r#type: "message".to_string(), |
| 221 | role: "assistant".to_string(), |
| 222 | content: Vec::new(), |
| 223 | model: model.clone(), |
| 224 | stop_reason: None, |
| 225 | stop_sequence: None, |
| 226 | container: None, |
| 227 | usage: Usage { |
| 228 | input_tokens: 0, |
| 229 | output_tokens: 0, |
| 230 | ..Usage::default() |
| 231 | }, |
| 232 | }, |
| 233 | }); |
| 234 | |
| 235 | let mut line_buf = String::new(); |
| 236 | let mut byte_buf = acquire_stream_buffer(); |
| 237 | let mut content_index: u32 = 0; |
| 238 | let mut text_started = false; |
| 239 | let mut thinking_started = false; |
| 240 | let mut tool_indices: std::collections::HashMap<u32, u32> = std::collections::HashMap::new(); |
| 241 | let is_reasoning_model = requires_reasoning_content(&model); |
| 242 | |
| 243 | let mut byte_stream = std::pin::pin!(byte_stream); |
| 244 | let idle = stream_idle_timeout(); |
| 245 | |
| 246 | // Telemetry for #103 stream-decode diagnostics: bytes received |
| 247 | // since the start of this stream and last successful event time. |
| 248 | // Surfaces in the error log when reqwest yields a chunk error so |
| 249 | // we can tell HTTP/2 RST_STREAM from chunk-decode-failure from |
| 250 | // gzip-corruption when investigating a flaky session. |
| 251 | let stream_start = std::time::Instant::now(); |
| 252 | let mut last_event_at = std::time::Instant::now(); |
| 253 | let mut bytes_received: usize = 0; |
| 254 | |
| 255 | loop { |
| 256 | let chunk_result = match tokio_timeout(idle, byte_stream.next()).await { |
| 257 | Ok(Some(result)) => result, |
| 258 | Ok(None) => break, // Stream ended normally |
| 259 | Err(_elapsed) => { |
| 260 | yield Err(anyhow::anyhow!( |
| 261 | "SSE stream idle timeout after {}s — no data received", |
| 262 | idle.as_secs(), |
| 263 | )); |
| 264 | break; |
| 265 | } |
| 266 | }; |
| 267 | let chunk = match chunk_result { |
| 268 | Ok(bytes) => bytes, |
| 269 | Err(e) => { |
| 270 | // Walk the error source chain so reqwest's underlying |
| 271 | // hyper / h2 / io error is visible — without this the |
| 272 | // outer "error decoding response body" message tells |
| 273 | // us nothing about WHY the stream died. |
| 274 | let mut error_chain = format!("{e}"); |
| 275 | let mut current: Option<&(dyn std::error::Error + 'static)> = |
| 276 | std::error::Error::source(&e); |
| 277 | while let Some(source) = current { |
| 278 | error_chain.push_str(&format!(" -> {source}")); |
| 279 | current = std::error::Error::source(source); |
| 280 | } |
| 281 | crate::logging::warn(format!( |
| 282 | "Stream read error: {error_chain} \ |
| 283 | (elapsed: {}ms, bytes_received: {}, ms_since_last_event: {}, headers: {})", |
| 284 | stream_start.elapsed().as_millis(), |
| 285 | bytes_received, |
| 286 | last_event_at.elapsed().as_millis(), |
| 287 | response_headers, |
| 288 | )); |
| 289 | yield Err(anyhow::anyhow!("Stream read error: {e}")); |
| 290 | break; |
| 291 | } |
| 292 | }; |
| 293 | |
| 294 | bytes_received = bytes_received.saturating_add(chunk.len()); |
| 295 | last_event_at = std::time::Instant::now(); |
| 296 | byte_buf.extend_from_slice(&chunk); |
| 297 | |
| 298 | // Guard against unbounded buffer growth (e.g., malformed stream without newlines) |
| 299 | const MAX_SSE_BUF: usize = 10 * 1024 * 1024; // 10 MB |
| 300 | if byte_buf.len() > MAX_SSE_BUF { |
| 301 | yield Err(anyhow::anyhow!("SSE buffer exceeded {MAX_SSE_BUF} bytes — aborting stream")); |
| 302 | break; |
| 303 | } |
| 304 | |
| 305 | if byte_buf.len() > SSE_BACKPRESSURE_HIGH_WATERMARK { |
| 306 | tokio::time::sleep(Duration::from_millis(SSE_BACKPRESSURE_SLEEP_MS)).await; |
| 307 | } |
| 308 | |
| 309 | // Process complete SSE lines from the buffer |
| 310 | let mut lines_processed = 0usize; |
| 311 | while let Some(newline_pos) = byte_buf.iter().position(|&b| b == b'\n') { |
| 312 | let mut end = newline_pos; |
| 313 | if end > 0 && byte_buf[end - 1] == b'\r' { |
| 314 | end -= 1; |
| 315 | } |
| 316 | let line = String::from_utf8_lossy(&byte_buf[..end]).into_owned(); |
| 317 | byte_buf.drain(..newline_pos + 1); |
| 318 | |
| 319 | if line.is_empty() { |
| 320 | // Empty line = event boundary, process accumulated data |
| 321 | if !line_buf.is_empty() { |
| 322 | let data = std::mem::take(&mut line_buf); |
| 323 | if data.trim() == "[DONE]" { |
| 324 | // Stream complete |
| 325 | } else if let Ok(chunk_json) = serde_json::from_str::<Value>(&data) { |
| 326 | // Parse the SSE chunk into stream events |
| 327 | for mut event in parse_sse_chunk( |
| 328 | &chunk_json, |
| 329 | &mut content_index, |
| 330 | &mut text_started, |
| 331 | &mut thinking_started, |
| 332 | &mut tool_indices, |
| 333 | is_reasoning_model, |
| 334 | ) { |
| 335 | // Stamp the client-side replay-token estimate |
| 336 | // onto the final usage so the UI can surface |
| 337 | // it (#30). We compute it pre-request and |
| 338 | // overlay it on the server-reported usage at |
| 339 | // stream completion. |
| 340 | if let Some(tokens) = replay_input_tokens |
| 341 | && let StreamEvent::MessageDelta { |
| 342 | usage: Some(usage), |
| 343 | .. |
| 344 | } = &mut event |
| 345 | { |
| 346 | usage.reasoning_replay_tokens = Some(tokens); |
| 347 | } |
| 348 | yield Ok(event); |
| 349 | } |
| 350 | } |
| 351 | } |
| 352 | continue; |
| 353 | } |
| 354 | |
| 355 | if let Some(data) = line.strip_prefix("data: ") { |
| 356 | line_buf.push_str(data); |
| 357 | } |
| 358 | // Ignore other SSE fields (event:, id:, retry:) |
| 359 | |
| 360 | lines_processed = lines_processed.saturating_add(1); |
| 361 | if lines_processed >= SSE_MAX_LINES_PER_CHUNK { |
| 362 | // Yield backpressure relief to avoid starving downstream consumers. |
| 363 | break; |
| 364 | } |
| 365 | } |
| 366 | } |
| 367 | |
| 368 | // Close any open blocks |
| 369 | if thinking_started { |
| 370 | yield Ok(StreamEvent::ContentBlockStop { index: content_index.saturating_sub(1) }); |
| 371 | } |
| 372 | if text_started { |
| 373 | yield Ok(StreamEvent::ContentBlockStop { index: content_index.saturating_sub(1) }); |
| 374 | } |
| 375 | |
| 376 | release_stream_buffer(byte_buf); |
| 377 | yield Ok(StreamEvent::MessageStop); |
| 378 | }; |
| 379 | |
| 380 | Ok(Pin::from(Box::new(stream) |
| 381 | as Box< |
| 382 | dyn futures_util::Stream<Item = Result<StreamEvent>> + Send, |
| 383 | >)) |
| 384 | } |
| 385 | } |
| 386 | |
| 387 | // === Chat Completions Helpers === |
| 388 | |
| 389 | #[cfg(test)] |
| 390 | pub(super) fn build_chat_messages( |
| 391 | system: Option<&SystemPrompt>, |
| 392 | messages: &[Message], |
| 393 | model: &str, |
| 394 | ) -> Vec<Value> { |
| 395 | build_chat_messages_with_reasoning( |
| 396 | system, |
| 397 | messages, |
| 398 | model, |
| 399 | should_replay_reasoning_content(model, None), |
| 400 | ) |
| 401 | } |
| 402 | |
| 403 | pub(super) fn build_chat_messages_for_request(request: &MessageRequest) -> Vec<Value> { |
| 404 | build_chat_messages_with_reasoning( |
| 405 | request.system.as_ref(), |
| 406 | &request.messages, |
| 407 | &request.model, |
| 408 | should_replay_reasoning_content(&request.model, request.reasoning_effort.as_deref()), |
| 409 | ) |
| 410 | } |
| 411 | |
| 412 | fn build_chat_messages_with_reasoning( |
| 413 | system: Option<&SystemPrompt>, |
| 414 | messages: &[Message], |
| 415 | _model: &str, |
| 416 | include_reasoning: bool, |
| 417 | ) -> Vec<Value> { |
| 418 | let mut out = Vec::new(); |
| 419 | let mut pending_tool_calls: HashSet<String> = HashSet::new(); |
| 420 | |
| 421 | if let Some(instructions) = system_to_instructions(system.cloned()) |
| 422 | && !instructions.trim().is_empty() |
| 423 | { |
| 424 | out.push(json!({ |
| 425 | "role": "system", |
| 426 | "content": instructions, |
| 427 | })); |
| 428 | } |
| 429 | |
| 430 | for message in messages.iter() { |
| 431 | let role = message.role.as_str(); |
| 432 | let mut text_parts = Vec::new(); |
| 433 | let mut thinking_parts = Vec::new(); |
| 434 | let mut tool_calls = Vec::new(); |
| 435 | let mut tool_call_ids = Vec::new(); |
| 436 | let mut tool_results: Vec<(String, Value)> = Vec::new(); |
| 437 | |
| 438 | for block in &message.content { |
| 439 | match block { |
| 440 | ContentBlock::Text { text, .. } => text_parts.push(text.clone()), |
| 441 | ContentBlock::Thinking { thinking } => thinking_parts.push(thinking.clone()), |
| 442 | ContentBlock::ToolUse { |
| 443 | id, |
| 444 | name, |
| 445 | input, |
| 446 | caller, |
| 447 | .. |
| 448 | } => { |
| 449 | let args = serde_json::to_string(input).unwrap_or_else(|_| input.to_string()); |
| 450 | let mut call = json!({ |
| 451 | "id": id, |
| 452 | "type": "function", |
| 453 | "function": { |
| 454 | "name": to_api_tool_name(name), |
| 455 | "arguments": args, |
| 456 | } |
| 457 | }); |
| 458 | if let Some(caller) = caller { |
| 459 | call["caller"] = json!({ |
| 460 | "type": caller.caller_type, |
| 461 | "tool_id": caller.tool_id, |
| 462 | }); |
| 463 | } |
| 464 | tool_calls.push(call); |
| 465 | tool_call_ids.push(id.clone()); |
| 466 | } |
| 467 | ContentBlock::ToolResult { |
| 468 | tool_use_id, |
| 469 | content, |
| 470 | .. |
| 471 | } => { |
| 472 | tool_results.push(( |
| 473 | tool_use_id.clone(), |
| 474 | json!({ |
| 475 | "role": "tool", |
| 476 | "tool_call_id": tool_use_id, |
| 477 | "content": content, |
| 478 | }), |
| 479 | )); |
| 480 | } |
| 481 | ContentBlock::ServerToolUse { .. } |
| 482 | | ContentBlock::ToolSearchToolResult { .. } |
| 483 | | ContentBlock::CodeExecutionToolResult { .. } => {} |
| 484 | } |
| 485 | } |
| 486 | |
| 487 | if role == "assistant" { |
| 488 | let content = text_parts.join("\n"); |
| 489 | let mut reasoning_content = thinking_parts.join("\n"); |
| 490 | let has_text = !content.trim().is_empty(); |
| 491 | let has_tool_calls = !tool_calls.is_empty(); |
| 492 | // DeepSeek thinking-mode rule: every assistant message in the |
| 493 | // conversation must carry its `reasoning_content` when thinking |
| 494 | // is enabled. The docs say non-tool-call messages' reasoning is |
| 495 | // "ignored", but the API still validates presence and rejects |
| 496 | // with a 400 if any assistant message is missing it. If reasoning |
| 497 | // was lost (e.g. a session checkpoint from before this rule was |
| 498 | // enforced, or a sub-turn with no streamed reasoning text), |
| 499 | // substitute a non-empty placeholder so the API accepts the |
| 500 | // request. |
| 501 | let include_reasoning_for_turn = include_reasoning; |
| 502 | let mut has_reasoning = |
| 503 | include_reasoning_for_turn && !reasoning_content.trim().is_empty(); |
| 504 | if include_reasoning_for_turn && !has_reasoning { |
| 505 | logging::warn( |
| 506 | "Substituting placeholder reasoning_content for DeepSeek tool-call assistant message", |
| 507 | ); |
| 508 | reasoning_content = String::from("(reasoning omitted)"); |
| 509 | has_reasoning = true; |
| 510 | } |
| 511 | |
| 512 | // DeepSeek rejects assistant messages where both `content` and |
| 513 | // `tool_calls` are missing/null. Skip such entries even if they |
| 514 | // carry reasoning-only metadata unless we can send a non-null |
| 515 | // placeholder content field. |
| 516 | if !has_text && !has_tool_calls && !has_reasoning { |
| 517 | pending_tool_calls.clear(); |
| 518 | continue; |
| 519 | } |
| 520 | |
| 521 | let mut msg = json!({ |
| 522 | "role": "assistant", |
| 523 | "content": if has_text { |
| 524 | json!(content) |
| 525 | } else if has_reasoning { |
| 526 | json!("") |
| 527 | } else { |
| 528 | Value::Null |
| 529 | }, |
| 530 | }); |
| 531 | if has_reasoning { |
| 532 | msg["reasoning_content"] = json!(reasoning_content); |
| 533 | } |
| 534 | if has_tool_calls { |
| 535 | msg["tool_calls"] = json!(tool_calls); |
| 536 | pending_tool_calls = tool_call_ids.into_iter().collect(); |
| 537 | } else { |
| 538 | pending_tool_calls.clear(); |
| 539 | } |
| 540 | out.push(msg); |
| 541 | } else if role == "user" { |
| 542 | let content = text_parts.join("\n"); |
| 543 | if !content.trim().is_empty() { |
| 544 | out.push(json!({ |
| 545 | "role": "user", |
| 546 | "content": content, |
| 547 | })); |
| 548 | } |
| 549 | } |
| 550 | |
| 551 | if !tool_results.is_empty() { |
| 552 | if pending_tool_calls.is_empty() { |
| 553 | logging::warn("Dropping tool results without matching tool_calls"); |
| 554 | } else { |
| 555 | for (tool_id, tool_msg) in tool_results { |
| 556 | if pending_tool_calls.remove(&tool_id) { |
| 557 | out.push(tool_msg); |
| 558 | } else { |
| 559 | logging::warn(format!( |
| 560 | "Dropping tool result for unknown tool_call_id: {tool_id}" |
| 561 | )); |
| 562 | } |
| 563 | } |
| 564 | } |
| 565 | } else if role != "assistant" { |
| 566 | pending_tool_calls.clear(); |
| 567 | } |
| 568 | } |
| 569 | |
| 570 | // Safety net: after compaction, an assistant message may have tool_calls |
| 571 | // whose results were summarized away. The API rejects these, so strip |
| 572 | // the tool_calls (downgrading to a plain assistant message) and remove |
| 573 | // the now-orphaned tool result messages. |
| 574 | let mut i = 0; |
| 575 | while i < out.len() { |
| 576 | let is_assistant_with_tools = out[i].get("role").and_then(Value::as_str) |
| 577 | == Some("assistant") |
| 578 | && out[i].get("tool_calls").is_some(); |
| 579 | |
| 580 | if is_assistant_with_tools { |
| 581 | let expected_ids: HashSet<String> = out[i] |
| 582 | .get("tool_calls") |
| 583 | .and_then(Value::as_array) |
| 584 | .map(|calls| { |
| 585 | calls |
| 586 | .iter() |
| 587 | .filter_map(|c| c.get("id").and_then(Value::as_str).map(String::from)) |
| 588 | .collect() |
| 589 | }) |
| 590 | .unwrap_or_default(); |
| 591 | |
| 592 | // Collect tool result IDs immediately following this assistant message. |
| 593 | let mut found_ids: HashSet<String> = HashSet::new(); |
| 594 | let mut tool_result_end = i + 1; |
| 595 | while tool_result_end < out.len() { |
| 596 | if out[tool_result_end].get("role").and_then(Value::as_str) == Some("tool") { |
| 597 | if let Some(id) = out[tool_result_end] |
| 598 | .get("tool_call_id") |
| 599 | .and_then(Value::as_str) |
| 600 | { |
| 601 | found_ids.insert(id.to_string()); |
| 602 | } |
| 603 | tool_result_end += 1; |
| 604 | } else { |
| 605 | break; |
| 606 | } |
| 607 | } |
| 608 | |
| 609 | // Also scan non-contiguous tool results up to the next assistant message |
| 610 | // in case compaction left gaps. |
| 611 | let mut scan = tool_result_end; |
| 612 | while scan < out.len() { |
| 613 | if out[scan].get("role").and_then(Value::as_str) == Some("assistant") { |
| 614 | break; |
| 615 | } |
| 616 | if out[scan].get("role").and_then(Value::as_str) == Some("tool") |
| 617 | && let Some(id) = out[scan].get("tool_call_id").and_then(Value::as_str) |
| 618 | { |
| 619 | found_ids.insert(id.to_string()); |
| 620 | } |
| 621 | scan += 1; |
| 622 | } |
| 623 | |
| 624 | if !expected_ids.is_subset(&found_ids) { |
| 625 | let missing: Vec<_> = expected_ids.difference(&found_ids).collect(); |
| 626 | logging::warn(format!( |
| 627 | "Stripping orphaned tool_calls from assistant message \ |
| 628 | (expected {} tool results, found {}, missing: {:?})", |
| 629 | expected_ids.len(), |
| 630 | found_ids.len(), |
| 631 | missing |
| 632 | )); |
| 633 | if let Some(obj) = out[i].as_object_mut() { |
| 634 | obj.remove("tool_calls"); |
| 635 | } |
| 636 | // If tool_calls were the only assistant content, remove the now-invalid |
| 637 | // assistant message entirely (DeepSeek requires content or tool_calls). |
| 638 | let assistant_content_empty = out[i] |
| 639 | .get("content") |
| 640 | .is_none_or(|v| v.is_null() || v.as_str().is_some_and(str::is_empty)); |
| 641 | if assistant_content_empty { |
| 642 | // Remove orphaned tool results tied to this stripped assistant call set. |
| 643 | let mut j = out.len(); |
| 644 | while j > i + 1 { |
| 645 | j -= 1; |
| 646 | if out[j].get("role").and_then(Value::as_str) == Some("tool") |
| 647 | && let Some(id) = out[j].get("tool_call_id").and_then(Value::as_str) |
| 648 | && expected_ids.contains(id) |
| 649 | { |
| 650 | out.remove(j); |
| 651 | } |
| 652 | } |
| 653 | out.remove(i); |
| 654 | i = i.saturating_sub(1); |
| 655 | continue; |
| 656 | } |
| 657 | // Remove contiguous tool results first |
| 658 | if tool_result_end > i + 1 { |
| 659 | out.drain((i + 1)..tool_result_end); |
| 660 | } |
| 661 | // Remove any remaining non-contiguous tool results referencing expected_ids |
| 662 | // (scan backward to avoid index shifting issues) |
| 663 | let mut j = out.len(); |
| 664 | while j > i + 1 { |
| 665 | j -= 1; |
| 666 | if out[j].get("role").and_then(Value::as_str) == Some("tool") |
| 667 | && let Some(id) = out[j].get("tool_call_id").and_then(Value::as_str) |
| 668 | && expected_ids.contains(id) |
| 669 | { |
| 670 | out.remove(j); |
| 671 | } |
| 672 | } |
| 673 | } |
| 674 | } |
| 675 | i += 1; |
| 676 | } |
| 677 | |
| 678 | out |
| 679 | } |
| 680 | |
| 681 | pub(super) fn tool_to_chat(tool: &Tool) -> Value { |
| 682 | let mut value = json!({ |
| 683 | "type": "function", |
| 684 | "function": { |
| 685 | "name": to_api_tool_name(&tool.name), |
| 686 | "description": tool.description, |
| 687 | "parameters": tool.input_schema, |
| 688 | } |
| 689 | }); |
| 690 | if let Some(allowed_callers) = &tool.allowed_callers { |
| 691 | value["allowed_callers"] = json!(allowed_callers); |
| 692 | } |
| 693 | if let Some(defer_loading) = tool.defer_loading { |
| 694 | value["defer_loading"] = json!(defer_loading); |
| 695 | } |
| 696 | if let Some(input_examples) = &tool.input_examples { |
| 697 | value["input_examples"] = json!(input_examples); |
| 698 | } |
| 699 | if let Some(strict) = tool.strict |
| 700 | && let Some(function) = value.get_mut("function") |
| 701 | { |
| 702 | function["strict"] = json!(strict); |
| 703 | } |
| 704 | value |
| 705 | } |
| 706 | |
| 707 | fn map_tool_choice_for_chat(choice: &Value) -> Option<Value> { |
| 708 | if let Some(choice_str) = choice.as_str() { |
| 709 | return Some(json!(choice_str)); |
| 710 | } |
| 711 | let Some(choice_type) = choice.get("type").and_then(Value::as_str) else { |
| 712 | return Some(choice.clone()); |
| 713 | }; |
| 714 | |
| 715 | match choice_type { |
| 716 | "auto" | "none" => Some(json!(choice_type)), |
| 717 | "any" => Some(json!("auto")), |
| 718 | "tool" => choice.get("name").and_then(Value::as_str).map(|name| { |
| 719 | json!({ |
| 720 | "type": "function", |
| 721 | "function": { "name": to_api_tool_name(name) } |
| 722 | }) |
| 723 | }), |
| 724 | _ => Some(choice.clone()), |
| 725 | } |
| 726 | } |
| 727 | |
| 728 | /// Final-pass sanitizer over the outgoing chat-completions JSON payload. |
| 729 | /// Forces a non-empty `reasoning_content` onto every `assistant` message that |
| 730 | /// carries `tool_calls`, when the model + effort combination requires it. |
| 731 | /// DeepSeek's thinking-mode API rejects such messages with a 400 error; |
| 732 | /// substituting a placeholder keeps the conversation chain intact. |
| 733 | /// |
| 734 | /// Also tallies the size of all replayed `reasoning_content` and logs it, so |
| 735 | /// users on `RUST_LOG=deepseek_tui=debug` can see how much of their input |
| 736 | /// budget is being spent re-sending prior thinking traces (V4 §5.1.1 |
| 737 | /// "Interleaved Thinking" requires the full trace to be replayed across user |
| 738 | /// message boundaries in tool-calling sessions). |
| 739 | pub(super) fn sanitize_thinking_mode_messages( |
| 740 | body: &mut Value, |
| 741 | model: &str, |
| 742 | effort: Option<&str>, |
| 743 | ) -> Option<u32> { |
| 744 | if !should_replay_reasoning_content(model, effort) { |
| 745 | return None; |
| 746 | } |
| 747 | let messages = body.get_mut("messages").and_then(Value::as_array_mut)?; |
| 748 | let mut substitutions: u32 = 0; |
| 749 | let mut replay_chars: u64 = 0; |
| 750 | let mut replay_messages: u32 = 0; |
| 751 | for (idx, msg) in messages.iter_mut().enumerate() { |
| 752 | if msg.get("role").and_then(Value::as_str) != Some("assistant") { |
| 753 | continue; |
| 754 | } |
| 755 | let needs_placeholder = msg |
| 756 | .get("reasoning_content") |
| 757 | .and_then(Value::as_str) |
| 758 | .is_none_or(|s| s.trim().is_empty()); |
| 759 | if needs_placeholder { |
| 760 | msg["reasoning_content"] = json!("(reasoning omitted)"); |
| 761 | substitutions = substitutions.saturating_add(1); |
| 762 | logging::warn(format!( |
| 763 | "Final sanitizer: forced reasoning_content placeholder on assistant[{idx}]", |
| 764 | )); |
| 765 | } |
| 766 | if let Some(reasoning) = msg.get("reasoning_content").and_then(Value::as_str) { |
| 767 | let len = reasoning.len() as u64; |
| 768 | if len > 0 { |
| 769 | replay_chars = replay_chars.saturating_add(len); |
| 770 | replay_messages = replay_messages.saturating_add(1); |
| 771 | } |
| 772 | } |
| 773 | } |
| 774 | if substitutions > 0 { |
| 775 | logging::warn(format!( |
| 776 | "Final sanitizer: {substitutions} assistant message(s) needed reasoning_content placeholder", |
| 777 | )); |
| 778 | } |
| 779 | if replay_messages == 0 { |
| 780 | return None; |
| 781 | } |
| 782 | // ~4 chars/token is the standard rough estimate; DeepSeek tokens skew |
| 783 | // a touch shorter on Chinese/code but this is order-of-magnitude info. |
| 784 | let approx_tokens = (replay_chars / 4).min(u64::from(u32::MAX)) as u32; |
| 785 | logging::info(format!( |
| 786 | "Reasoning-content replay: {replay_messages} assistant message(s), ~{approx_tokens} input tokens ({replay_chars} chars) being re-sent in this request", |
| 787 | )); |
| 788 | Some(approx_tokens) |
| 789 | } |
| 790 | |
| 791 | /// Sums the byte length of `reasoning_content` across all assistant messages in |
| 792 | /// an outgoing chat-completions body. Used by tests; the production sanitizer |
| 793 | /// computes the same number inline and logs it. |
| 794 | #[cfg(test)] |
| 795 | pub(super) fn count_reasoning_replay_chars(body: &Value) -> u64 { |
| 796 | let Some(messages) = body.get("messages").and_then(Value::as_array) else { |
| 797 | return 0; |
| 798 | }; |
| 799 | messages |
| 800 | .iter() |
| 801 | .filter(|m| m.get("role").and_then(Value::as_str) == Some("assistant")) |
| 802 | .filter_map(|m| m.get("reasoning_content").and_then(Value::as_str)) |
| 803 | .map(|s| s.len() as u64) |
| 804 | .sum() |
| 805 | } |
| 806 | |
| 807 | /// Render the transport-shape headers we care about for #103 diagnostics. |
| 808 | /// Always returns SOMETHING printable so the decode-error log line is parseable |
| 809 | /// even when the server stripped a header we expected. |
| 810 | fn format_stream_headers(headers: &reqwest::header::HeaderMap) -> String { |
| 811 | const FIELDS: &[&str] = &[ |
| 812 | "content-encoding", |
| 813 | "transfer-encoding", |
| 814 | "connection", |
| 815 | "server", |
| 816 | ]; |
| 817 | let mut parts: Vec<String> = Vec::with_capacity(FIELDS.len()); |
| 818 | for field in FIELDS { |
| 819 | let rendered = headers |
| 820 | .get(*field) |
| 821 | .and_then(|v| v.to_str().ok()) |
| 822 | .unwrap_or("(absent)"); |
| 823 | parts.push(format!("{field}={rendered}")); |
| 824 | } |
| 825 | parts.join(", ") |
| 826 | } |
| 827 | |
| 828 | /// Diagnostic logger fired when DeepSeek rejects the request despite the |
| 829 | /// sanitizer. Walks the body and logs which assistant messages have tool_calls |
| 830 | /// but no `reasoning_content` — useful to track down a code path that bypasses |
| 831 | /// the sanitizer entirely. |
| 832 | fn log_thinking_mode_violations(body: &Value) { |
| 833 | let Some(messages) = body.get("messages").and_then(Value::as_array) else { |
| 834 | logging::warn("400-after-sanitizer: body has no `messages` array"); |
| 835 | return; |
| 836 | }; |
| 837 | let mut violations: Vec<String> = Vec::new(); |
| 838 | for (idx, msg) in messages.iter().enumerate() { |
| 839 | if msg.get("role").and_then(Value::as_str) != Some("assistant") { |
| 840 | continue; |
| 841 | } |
| 842 | let reasoning = msg |
| 843 | .get("reasoning_content") |
| 844 | .and_then(Value::as_str) |
| 845 | .unwrap_or(""); |
| 846 | let has_tc = msg.get("tool_calls").is_some(); |
| 847 | if reasoning.trim().is_empty() { |
| 848 | violations.push(format!( |
| 849 | "assistant[{idx}] (reasoning_content missing, tool_calls={})", |
| 850 | has_tc |
| 851 | )); |
| 852 | } |
| 853 | } |
| 854 | if violations.is_empty() { |
| 855 | logging::warn( |
| 856 | "400-after-sanitizer: all assistant messages have reasoning_content — DeepSeek rejected for a different reason", |
| 857 | ); |
| 858 | } else { |
| 859 | logging::warn(format!( |
| 860 | "400-after-sanitizer: {} assistant message(s) lack reasoning_content despite sanitizer: {}", |
| 861 | violations.len(), |
| 862 | violations.join(", ") |
| 863 | )); |
| 864 | } |
| 865 | } |
| 866 | |
| 867 | fn requires_reasoning_content(model: &str) -> bool { |
| 868 | let lower = model.to_lowercase(); |
| 869 | lower.contains("deepseek-v4") |
| 870 | || lower.contains("reasoner") |
| 871 | || lower.contains("-reasoning") |
| 872 | || lower.contains("-thinking") |
| 873 | || has_deepseek_r_series_marker(&lower) |
| 874 | } |
| 875 | |
| 876 | fn should_replay_reasoning_content(model: &str, effort: Option<&str>) -> bool { |
| 877 | if effort |
| 878 | .map(|value| { |
| 879 | matches!( |
| 880 | value.trim().to_ascii_lowercase().as_str(), |
| 881 | "off" | "disabled" | "none" | "false" |
| 882 | ) |
| 883 | }) |
| 884 | .unwrap_or(false) |
| 885 | { |
| 886 | return false; |
| 887 | } |
| 888 | |
| 889 | requires_reasoning_content(model) |
| 890 | } |
| 891 | |
| 892 | fn has_deepseek_r_series_marker(model_lower: &str) -> bool { |
| 893 | const PREFIX: &str = "deepseek-r"; |
| 894 | model_lower.match_indices(PREFIX).any(|(idx, _)| { |
| 895 | model_lower[idx + PREFIX.len()..] |
| 896 | .chars() |
| 897 | .next() |
| 898 | .is_some_and(|ch| ch.is_ascii_digit()) |
| 899 | }) |
| 900 | } |
| 901 | |
| 902 | fn reasoning_field(value: &Value) -> Option<&str> { |
| 903 | value |
| 904 | .get("reasoning_content") |
| 905 | .or_else(|| value.get("reasoning")) |
| 906 | .and_then(Value::as_str) |
| 907 | } |
| 908 | |
| 909 | pub(super) fn parse_chat_message(payload: &Value) -> Result<MessageResponse> { |
| 910 | let id = payload |
| 911 | .get("id") |
| 912 | .and_then(Value::as_str) |
| 913 | .unwrap_or("chatcmpl") |
| 914 | .to_string(); |
| 915 | let model = payload |
| 916 | .get("model") |
| 917 | .and_then(Value::as_str) |
| 918 | .unwrap_or("unknown") |
| 919 | .to_string(); |
| 920 | |
| 921 | let choices = payload |
| 922 | .get("choices") |
| 923 | .and_then(Value::as_array) |
| 924 | .context("Chat API response missing choices")?; |
| 925 | let choice = choices |
| 926 | .first() |
| 927 | .context("Chat API response missing first choice")?; |
| 928 | let message = choice |
| 929 | .get("message") |
| 930 | .context("Chat API response missing message")?; |
| 931 | |
| 932 | let mut content_blocks = Vec::new(); |
| 933 | if let Some(reasoning) = |
| 934 | reasoning_field(message).filter(|reasoning| !reasoning.trim().is_empty()) |
| 935 | { |
| 936 | content_blocks.push(ContentBlock::Thinking { |
| 937 | thinking: reasoning.to_string(), |
| 938 | }); |
| 939 | } |
| 940 | if let Some(text) = message.get("content").and_then(Value::as_str) |
| 941 | && !text.trim().is_empty() |
| 942 | { |
| 943 | content_blocks.push(ContentBlock::Text { |
| 944 | text: text.to_string(), |
| 945 | cache_control: None, |
| 946 | }); |
| 947 | } |
| 948 | |
| 949 | if let Some(tool_calls) = message.get("tool_calls").and_then(Value::as_array) { |
| 950 | for call in tool_calls { |
| 951 | let id = call |
| 952 | .get("id") |
| 953 | .and_then(Value::as_str) |
| 954 | .unwrap_or("tool_call") |
| 955 | .to_string(); |
| 956 | let function = call.get("function"); |
| 957 | let name = function |
| 958 | .and_then(|f| f.get("name")) |
| 959 | .and_then(Value::as_str) |
| 960 | .unwrap_or("tool") |
| 961 | .to_string(); |
| 962 | let arguments = function |
| 963 | .and_then(|f| f.get("arguments")) |
| 964 | .and_then(Value::as_str) |
| 965 | .map(|raw| serde_json::from_str(raw).unwrap_or(Value::String(raw.to_string()))) |
| 966 | .unwrap_or(Value::Null); |
| 967 | let caller = call.get("caller").and_then(|v| { |
| 968 | v.get("type") |
| 969 | .and_then(Value::as_str) |
| 970 | .map(|caller_type| ToolCaller { |
| 971 | caller_type: caller_type.to_string(), |
| 972 | tool_id: v |
| 973 | .get("tool_id") |
| 974 | .and_then(Value::as_str) |
| 975 | .map(std::string::ToString::to_string), |
| 976 | }) |
| 977 | }); |
| 978 | |
| 979 | content_blocks.push(ContentBlock::ToolUse { |
| 980 | id, |
| 981 | name: from_api_tool_name(&name), |
| 982 | input: arguments, |
| 983 | caller, |
| 984 | }); |
| 985 | } |
| 986 | } |
| 987 | |
| 988 | let usage = parse_usage(payload.get("usage")); |
| 989 | |
| 990 | Ok(MessageResponse { |
| 991 | id, |
| 992 | r#type: "message".to_string(), |
| 993 | role: "assistant".to_string(), |
| 994 | content: content_blocks, |
| 995 | model, |
| 996 | stop_reason: choice |
| 997 | .get("finish_reason") |
| 998 | .and_then(Value::as_str) |
| 999 | .map(str::to_string), |
| 1000 | stop_sequence: None, |
| 1001 | container: None, |
| 1002 | usage, |
| 1003 | }) |
| 1004 | } |
| 1005 | |
| 1006 | // === Streaming Helpers === |
| 1007 | |
| 1008 | /// Build synthetic stream events from a non-streaming response (used as fallback). |
| 1009 | #[allow(dead_code)] |
| 1010 | fn build_stream_events(response: &MessageResponse) -> Vec<StreamEvent> { |
| 1011 | let mut events = Vec::new(); |
| 1012 | let mut index = 0u32; |
| 1013 | |
| 1014 | events.push(StreamEvent::MessageStart { |
| 1015 | message: response.clone(), |
| 1016 | }); |
| 1017 | |
| 1018 | for block in &response.content { |
| 1019 | match block { |
| 1020 | ContentBlock::Text { text, .. } => { |
| 1021 | events.push(StreamEvent::ContentBlockStart { |
| 1022 | index, |
| 1023 | content_block: ContentBlockStart::Text { |
| 1024 | text: String::new(), |
| 1025 | }, |
| 1026 | }); |
| 1027 | if !text.is_empty() { |
| 1028 | events.push(StreamEvent::ContentBlockDelta { |
| 1029 | index, |
| 1030 | delta: Delta::TextDelta { text: text.clone() }, |
| 1031 | }); |
| 1032 | } |
| 1033 | events.push(StreamEvent::ContentBlockStop { index }); |
| 1034 | } |
| 1035 | ContentBlock::Thinking { thinking } => { |
| 1036 | events.push(StreamEvent::ContentBlockStart { |
| 1037 | index, |
| 1038 | content_block: ContentBlockStart::Thinking { |
| 1039 | thinking: String::new(), |
| 1040 | }, |
| 1041 | }); |
| 1042 | if !thinking.is_empty() { |
| 1043 | events.push(StreamEvent::ContentBlockDelta { |
| 1044 | index, |
| 1045 | delta: Delta::ThinkingDelta { |
| 1046 | thinking: thinking.clone(), |
| 1047 | }, |
| 1048 | }); |
| 1049 | } |
| 1050 | events.push(StreamEvent::ContentBlockStop { index }); |
| 1051 | } |
| 1052 | ContentBlock::ToolUse { |
| 1053 | id, name, input, .. |
| 1054 | } => { |
| 1055 | events.push(StreamEvent::ContentBlockStart { |
| 1056 | index, |
| 1057 | content_block: ContentBlockStart::ToolUse { |
| 1058 | id: id.clone(), |
| 1059 | name: name.clone(), |
| 1060 | input: input.clone(), |
| 1061 | caller: None, |
| 1062 | }, |
| 1063 | }); |
| 1064 | events.push(StreamEvent::ContentBlockStop { index }); |
| 1065 | } |
| 1066 | ContentBlock::ToolResult { .. } => {} |
| 1067 | ContentBlock::ServerToolUse { id, name, input } => { |
| 1068 | events.push(StreamEvent::ContentBlockStart { |
| 1069 | index, |
| 1070 | content_block: ContentBlockStart::ServerToolUse { |
| 1071 | id: id.clone(), |
| 1072 | name: name.clone(), |
| 1073 | input: input.clone(), |
| 1074 | }, |
| 1075 | }); |
| 1076 | events.push(StreamEvent::ContentBlockStop { index }); |
| 1077 | } |
| 1078 | ContentBlock::ToolSearchToolResult { .. } |
| 1079 | | ContentBlock::CodeExecutionToolResult { .. } => {} |
| 1080 | } |
| 1081 | index = index.saturating_add(1); |
| 1082 | } |
| 1083 | |
| 1084 | events.push(StreamEvent::MessageDelta { |
| 1085 | delta: MessageDelta { |
| 1086 | stop_reason: response.stop_reason.clone(), |
| 1087 | stop_sequence: response.stop_sequence.clone(), |
| 1088 | }, |
| 1089 | usage: Some(response.usage.clone()), |
| 1090 | }); |
| 1091 | events.push(StreamEvent::MessageStop); |
| 1092 | |
| 1093 | events |
| 1094 | } |
| 1095 | |
| 1096 | // === SSE Chunk Parser === |
| 1097 | |
| 1098 | /// Parse a single SSE chunk from the Chat Completions streaming API into |
| 1099 | /// our internal `StreamEvent` representation. |
| 1100 | pub(super) fn parse_sse_chunk( |
| 1101 | chunk: &Value, |
| 1102 | content_index: &mut u32, |
| 1103 | text_started: &mut bool, |
| 1104 | thinking_started: &mut bool, |
| 1105 | tool_indices: &mut std::collections::HashMap<u32, u32>, |
| 1106 | is_reasoning_model: bool, |
| 1107 | ) -> Vec<StreamEvent> { |
| 1108 | let mut events = Vec::new(); |
| 1109 | |
| 1110 | let Some(choices) = chunk.get("choices").and_then(Value::as_array) else { |
| 1111 | // Usage-only chunk (sent at end with stream_options) |
| 1112 | if let Some(usage_val) = chunk.get("usage") { |
| 1113 | let usage = parse_usage(Some(usage_val)); |
| 1114 | events.push(StreamEvent::MessageDelta { |
| 1115 | delta: MessageDelta { |
| 1116 | stop_reason: None, |
| 1117 | stop_sequence: None, |
| 1118 | }, |
| 1119 | usage: Some(usage), |
| 1120 | }); |
| 1121 | } |
| 1122 | return events; |
| 1123 | }; |
| 1124 | |
| 1125 | if choices.is_empty() { |
| 1126 | if let Some(usage_val) = chunk.get("usage") { |
| 1127 | let usage = parse_usage(Some(usage_val)); |
| 1128 | events.push(StreamEvent::MessageDelta { |
| 1129 | delta: MessageDelta { |
| 1130 | stop_reason: None, |
| 1131 | stop_sequence: None, |
| 1132 | }, |
| 1133 | usage: Some(usage), |
| 1134 | }); |
| 1135 | } |
| 1136 | return events; |
| 1137 | } |
| 1138 | |
| 1139 | for choice in choices { |
| 1140 | let delta = choice.get("delta"); |
| 1141 | let finish_reason = choice |
| 1142 | .get("finish_reason") |
| 1143 | .and_then(Value::as_str) |
| 1144 | .map(str::to_string); |
| 1145 | |
| 1146 | if let Some(delta) = delta { |
| 1147 | // Handle reasoning_content / reasoning thinking deltas. |
| 1148 | if is_reasoning_model |
| 1149 | && let Some(reasoning) = reasoning_field(delta) |
| 1150 | && !reasoning.is_empty() |
| 1151 | { |
| 1152 | if !*thinking_started { |
| 1153 | events.push(StreamEvent::ContentBlockStart { |
| 1154 | index: *content_index, |
| 1155 | content_block: ContentBlockStart::Thinking { |
| 1156 | thinking: String::new(), |
| 1157 | }, |
| 1158 | }); |
| 1159 | *thinking_started = true; |
| 1160 | } |
| 1161 | events.push(StreamEvent::ContentBlockDelta { |
| 1162 | index: *content_index, |
| 1163 | delta: Delta::ThinkingDelta { |
| 1164 | thinking: reasoning.to_string(), |
| 1165 | }, |
| 1166 | }); |
| 1167 | } |
| 1168 | |
| 1169 | // Handle regular content |
| 1170 | if let Some(content) = delta.get("content").and_then(Value::as_str) |
| 1171 | && !content.is_empty() |
| 1172 | { |
| 1173 | // Close thinking block if transitioning to text |
| 1174 | if *thinking_started { |
| 1175 | events.push(StreamEvent::ContentBlockStop { |
| 1176 | index: *content_index, |
| 1177 | }); |
| 1178 | *content_index += 1; |
| 1179 | *thinking_started = false; |
| 1180 | } |
| 1181 | if !*text_started { |
| 1182 | events.push(StreamEvent::ContentBlockStart { |
| 1183 | index: *content_index, |
| 1184 | content_block: ContentBlockStart::Text { |
| 1185 | text: String::new(), |
| 1186 | }, |
| 1187 | }); |
| 1188 | *text_started = true; |
| 1189 | } |
| 1190 | events.push(StreamEvent::ContentBlockDelta { |
| 1191 | index: *content_index, |
| 1192 | delta: Delta::TextDelta { |
| 1193 | text: content.to_string(), |
| 1194 | }, |
| 1195 | }); |
| 1196 | } |
| 1197 | |
| 1198 | // Handle tool calls |
| 1199 | if let Some(tool_calls) = delta.get("tool_calls").and_then(Value::as_array) { |
| 1200 | for tc in tool_calls { |
| 1201 | let tc_index = tc.get("index").and_then(Value::as_u64).unwrap_or(0) as u32; |
| 1202 | let tool_block_index = match tool_indices.entry(tc_index) { |
| 1203 | std::collections::hash_map::Entry::Occupied(entry) => *entry.get(), |
| 1204 | std::collections::hash_map::Entry::Vacant(entry) => { |
| 1205 | // Close text block if transitioning to tool use |
| 1206 | if *text_started { |
| 1207 | events.push(StreamEvent::ContentBlockStop { |
| 1208 | index: *content_index, |
| 1209 | }); |
| 1210 | *content_index += 1; |
| 1211 | *text_started = false; |
| 1212 | } |
| 1213 | if *thinking_started { |
| 1214 | events.push(StreamEvent::ContentBlockStop { |
| 1215 | index: *content_index, |
| 1216 | }); |
| 1217 | *content_index += 1; |
| 1218 | *thinking_started = false; |
| 1219 | } |
| 1220 | |
| 1221 | let block_index = *content_index; |
| 1222 | let id = tc |
| 1223 | .get("id") |
| 1224 | .and_then(Value::as_str) |
| 1225 | .map(str::to_string) |
| 1226 | // Some upstream gateways (and the responses-API |
| 1227 | // bridge) elide the `id` on the first chunk of a |
| 1228 | // tool call. Falling back to a constant string |
| 1229 | // collides when the model emits parallel tool |
| 1230 | // calls in the same delta — every call ended up |
| 1231 | // with the same id and downstream tool-result |
| 1232 | // routing matched the first one twice. Index by |
| 1233 | // the content-block position to keep the |
| 1234 | // fallback unique within the response. |
| 1235 | .unwrap_or_else(|| format!("call_{block_index}")); |
| 1236 | let name = tc |
| 1237 | .get("function") |
| 1238 | .and_then(|f| f.get("name")) |
| 1239 | .and_then(Value::as_str) |
| 1240 | .unwrap_or("") |
| 1241 | .to_string(); |
| 1242 | let caller = tc.get("caller").and_then(|v| { |
| 1243 | v.get("type").and_then(Value::as_str).map(|caller_type| { |
| 1244 | ToolCaller { |
| 1245 | caller_type: caller_type.to_string(), |
| 1246 | tool_id: v |
| 1247 | .get("tool_id") |
| 1248 | .and_then(Value::as_str) |
| 1249 | .map(std::string::ToString::to_string), |
| 1250 | } |
| 1251 | }) |
| 1252 | }); |
| 1253 | |
| 1254 | events.push(StreamEvent::ContentBlockStart { |
| 1255 | index: block_index, |
| 1256 | content_block: ContentBlockStart::ToolUse { |
| 1257 | id, |
| 1258 | name: from_api_tool_name(&name), |
| 1259 | input: json!({}), |
| 1260 | caller, |
| 1261 | }, |
| 1262 | }); |
| 1263 | *content_index = (*content_index).saturating_add(1); |
| 1264 | entry.insert(block_index); |
| 1265 | block_index |
| 1266 | } |
| 1267 | }; |
| 1268 | |
| 1269 | // Stream tool call arguments |
| 1270 | if let Some(args) = tc |
| 1271 | .get("function") |
| 1272 | .and_then(|f| f.get("arguments")) |
| 1273 | .and_then(Value::as_str) |
| 1274 | && !args.is_empty() |
| 1275 | { |
| 1276 | events.push(StreamEvent::ContentBlockDelta { |
| 1277 | index: tool_block_index, |
| 1278 | delta: Delta::InputJsonDelta { |
| 1279 | partial_json: args.to_string(), |
| 1280 | }, |
| 1281 | }); |
| 1282 | } |
| 1283 | } |
| 1284 | } |
| 1285 | } |
| 1286 | |
| 1287 | // Handle finish reason |
| 1288 | if let Some(reason) = finish_reason { |
| 1289 | // Close any open blocks |
| 1290 | if *text_started { |
| 1291 | events.push(StreamEvent::ContentBlockStop { |
| 1292 | index: *content_index, |
| 1293 | }); |
| 1294 | *text_started = false; |
| 1295 | } |
| 1296 | if *thinking_started { |
| 1297 | events.push(StreamEvent::ContentBlockStop { |
| 1298 | index: *content_index, |
| 1299 | }); |
| 1300 | *thinking_started = false; |
| 1301 | } |
| 1302 | // Close tool blocks |
| 1303 | let mut open_tool_indices: Vec<u32> = |
| 1304 | tool_indices.drain().map(|(_, idx)| idx).collect(); |
| 1305 | open_tool_indices.sort_unstable(); |
| 1306 | for tool_block_index in open_tool_indices { |
| 1307 | events.push(StreamEvent::ContentBlockStop { |
| 1308 | index: tool_block_index, |
| 1309 | }); |
| 1310 | } |
| 1311 | |
| 1312 | // Emit usage from the chunk if available |
| 1313 | let chunk_usage = chunk.get("usage").map(|u| parse_usage(Some(u))); |
| 1314 | events.push(StreamEvent::MessageDelta { |
| 1315 | delta: MessageDelta { |
| 1316 | stop_reason: Some(reason), |
| 1317 | stop_sequence: None, |
| 1318 | }, |
| 1319 | usage: chunk_usage, |
| 1320 | }); |
| 1321 | } |
| 1322 | } |
| 1323 | |
| 1324 | events |
| 1325 | } |
| 1326 | |
| 1327 | // === #103 Phase 1: stream-decode diagnostics =================================== |
| 1328 | |
| 1329 | #[cfg(test)] |
| 1330 | mod stream_diagnostics_tests { |
| 1331 | use super::*; |
| 1332 | use reqwest::header::{HeaderMap, HeaderValue}; |
| 1333 | |
| 1334 | #[test] |
| 1335 | fn stream_open_timeout_defaults_and_clamps_env_values() { |
| 1336 | assert_eq!(stream_open_timeout_from_env(None), Duration::from_secs(45)); |
| 1337 | assert_eq!( |
| 1338 | stream_open_timeout_from_env(Some("not-a-number")), |
| 1339 | Duration::from_secs(45) |
| 1340 | ); |
| 1341 | assert_eq!( |
| 1342 | stream_open_timeout_from_env(Some("1")), |
| 1343 | Duration::from_secs(5) |
| 1344 | ); |
| 1345 | assert_eq!( |
| 1346 | stream_open_timeout_from_env(Some("120")), |
| 1347 | Duration::from_secs(120) |
| 1348 | ); |
| 1349 | assert_eq!( |
| 1350 | stream_open_timeout_from_env(Some("999")), |
| 1351 | Duration::from_secs(300) |
| 1352 | ); |
| 1353 | } |
| 1354 | |
| 1355 | #[test] |
| 1356 | fn format_stream_headers_renders_all_fields_when_present() { |
| 1357 | let mut headers = HeaderMap::new(); |
| 1358 | headers.insert("content-encoding", HeaderValue::from_static("gzip")); |
| 1359 | headers.insert("transfer-encoding", HeaderValue::from_static("chunked")); |
| 1360 | headers.insert("connection", HeaderValue::from_static("keep-alive")); |
| 1361 | headers.insert("server", HeaderValue::from_static("openresty/1.25.3.1")); |
| 1362 | |
| 1363 | let rendered = format_stream_headers(&headers); |
| 1364 | // Order is fixed by FIELDS in the helper; assert each field appears. |
| 1365 | assert!( |
| 1366 | rendered.contains("content-encoding=gzip"), |
| 1367 | "got: {rendered}" |
| 1368 | ); |
| 1369 | assert!( |
| 1370 | rendered.contains("transfer-encoding=chunked"), |
| 1371 | "got: {rendered}" |
| 1372 | ); |
| 1373 | assert!( |
| 1374 | rendered.contains("connection=keep-alive"), |
| 1375 | "got: {rendered}" |
| 1376 | ); |
| 1377 | assert!( |
| 1378 | rendered.contains("server=openresty/1.25.3.1"), |
| 1379 | "got: {rendered}" |
| 1380 | ); |
| 1381 | } |
| 1382 | |
| 1383 | #[test] |
| 1384 | fn format_stream_headers_marks_missing_fields_as_absent() { |
| 1385 | // DeepSeek frequently omits content-encoding when not compressing. |
| 1386 | // The diagnostic must still produce a parseable line so log scrapers |
| 1387 | // don't lose the slot. |
| 1388 | let headers = HeaderMap::new(); |
| 1389 | let rendered = format_stream_headers(&headers); |
| 1390 | assert!( |
| 1391 | rendered.contains("content-encoding=(absent)"), |
| 1392 | "missing field must be explicitly marked; got: {rendered}" |
| 1393 | ); |
| 1394 | assert!( |
| 1395 | rendered.contains("transfer-encoding=(absent)"), |
| 1396 | "missing field must be explicitly marked; got: {rendered}" |
| 1397 | ); |
| 1398 | } |
| 1399 | |
| 1400 | #[test] |
| 1401 | fn format_stream_headers_handles_non_ascii_value_gracefully() { |
| 1402 | // If a header value isn't UTF-8, `.to_str()` fails — we must not panic |
| 1403 | // and should still produce a parseable line. |
| 1404 | let mut headers = HeaderMap::new(); |
| 1405 | // 0xFF is a valid byte but invalid UTF-8 start byte. |
| 1406 | headers.insert( |
| 1407 | "server", |
| 1408 | HeaderValue::from_bytes(b"\xff\xfemystery").expect("header value"), |
| 1409 | ); |
| 1410 | let rendered = format_stream_headers(&headers); |
| 1411 | assert!( |
| 1412 | rendered.contains("server=(absent)"), |
| 1413 | "non-UTF8 header values fall back to (absent); got: {rendered}" |
| 1414 | ); |
| 1415 | } |
| 1416 | } |
| 1417 | |
| 1418 | // === #103 Phase 4: SSE decoder behavior on canned chunk sequences ============ |
| 1419 | |
| 1420 | #[cfg(test)] |
| 1421 | mod stream_decoder_tests { |
| 1422 | //! Drive `parse_sse_chunk` (the in-place SSE event extractor) over canned |
| 1423 | //! chunk sequences. The full `handle_chat_completion_stream` path needs a |
| 1424 | //! live `reqwest::Response` so it isn't unit-testable without a mock HTTP |
| 1425 | //! harness (issue #69 tracks that). For #103 we exercise the chunk decoder |
| 1426 | //! directly to verify each "class of stream failure" the engine relies on. |
| 1427 | use super::*; |
| 1428 | use crate::models::{ContentBlockStart, Delta, StreamEvent}; |
| 1429 | |
| 1430 | /// Decode a raw SSE-data JSON chunk into our internal events, mirroring |
| 1431 | /// the per-event call shape used by `handle_chat_completion_stream`. |
| 1432 | fn decode_chunk(json_text: &str) -> Vec<StreamEvent> { |
| 1433 | let chunk: Value = serde_json::from_str(json_text).expect("valid SSE JSON"); |
| 1434 | let mut content_index = 0u32; |
| 1435 | let mut text_started = false; |
| 1436 | let mut thinking_started = false; |
| 1437 | let mut tool_indices = std::collections::HashMap::new(); |
| 1438 | parse_sse_chunk( |
| 1439 | &chunk, |
| 1440 | &mut content_index, |
| 1441 | &mut text_started, |
| 1442 | &mut thinking_started, |
| 1443 | &mut tool_indices, |
| 1444 | true, |
| 1445 | ) |
| 1446 | } |
| 1447 | |
| 1448 | #[test] |
| 1449 | fn decoder_emits_text_delta_for_content_chunk() { |
| 1450 | // The "happy" first chunk: a normal content delta. The engine treats |
| 1451 | // this as `any_content_received = true` and would NOT transparently |
| 1452 | // retry on a subsequent error. |
| 1453 | let events = decode_chunk(r#"{"choices":[{"delta":{"content":"hello"}}]}"#); |
| 1454 | assert!( |
| 1455 | matches!( |
| 1456 | events.first(), |
| 1457 | Some(StreamEvent::ContentBlockStart { |
| 1458 | content_block: ContentBlockStart::Text { .. }, |
| 1459 | .. |
| 1460 | }) |
| 1461 | ), |
| 1462 | "first event should open a text block; got {events:?}" |
| 1463 | ); |
| 1464 | assert!( |
| 1465 | events |
| 1466 | .iter() |
| 1467 | .any(|e| matches!(e, StreamEvent::ContentBlockDelta { |
| 1468 | delta: Delta::TextDelta { text }, |
| 1469 | .. |
| 1470 | } if text == "hello")), |
| 1471 | "should yield a TextDelta carrying 'hello'; got {events:?}" |
| 1472 | ); |
| 1473 | } |
| 1474 | |
| 1475 | #[test] |
| 1476 | fn decoder_emits_thinking_delta_for_reasoning_chunk() { |
| 1477 | // V4 thinking models surface reasoning_content first — the engine |
| 1478 | // also counts these as content received (so a subsequent stream error |
| 1479 | // surfaces rather than retrying transparently). |
| 1480 | let events = decode_chunk(r#"{"choices":[{"delta":{"reasoning_content":"plan..."}}]}"#); |
| 1481 | assert!( |
| 1482 | matches!( |
| 1483 | events.first(), |
| 1484 | Some(StreamEvent::ContentBlockStart { |
| 1485 | content_block: ContentBlockStart::Thinking { .. }, |
| 1486 | .. |
| 1487 | }) |
| 1488 | ), |
| 1489 | "first event should open a thinking block; got {events:?}" |
| 1490 | ); |
| 1491 | assert!( |
| 1492 | events |
| 1493 | .iter() |
| 1494 | .any(|e| matches!(e, StreamEvent::ContentBlockDelta { |
| 1495 | delta: Delta::ThinkingDelta { thinking }, |
| 1496 | .. |
| 1497 | } if thinking == "plan...")), |
| 1498 | "should yield a ThinkingDelta carrying 'plan...'; got {events:?}" |
| 1499 | ); |
| 1500 | } |
| 1501 | |
| 1502 | #[test] |
| 1503 | fn decoder_yields_no_events_for_keepalive_chunk() { |
| 1504 | // DeepSeek often sends `{"choices":[]}` keepalive chunks before |
| 1505 | // emitting real content. The engine MUST treat a stream error after |
| 1506 | // these as "no content received" and be eligible for transparent |
| 1507 | // retry — assert here that the decoder yields no payload events. |
| 1508 | let events = decode_chunk(r#"{"choices":[]}"#); |
| 1509 | assert!( |
| 1510 | events.is_empty(), |
| 1511 | "empty-choices chunk must produce no events; got {events:?}" |
| 1512 | ); |
| 1513 | } |
| 1514 | |
| 1515 | #[test] |
| 1516 | fn decoder_emits_tool_use_block_for_tool_call_delta() { |
| 1517 | // Tool-call deltas are content too — once one arrives, transparent |
| 1518 | // retry must be off (the model has committed to a tool invocation |
| 1519 | // path that DeepSeek has billed for). |
| 1520 | let events = decode_chunk( |
| 1521 | r#"{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","function":{"name":"grep_files","arguments":"{\"pattern\":\"foo\"}"}}]}}]}"#, |
| 1522 | ); |
| 1523 | assert!( |
| 1524 | events.iter().any(|e| matches!( |
| 1525 | e, |
| 1526 | StreamEvent::ContentBlockStart { |
| 1527 | content_block: ContentBlockStart::ToolUse { name, .. }, |
| 1528 | .. |
| 1529 | } if name == "grep_files" |
| 1530 | )), |
| 1531 | "should open a ToolUse block for grep_files; got {events:?}" |
| 1532 | ); |
| 1533 | assert!( |
| 1534 | events.iter().any(|e| matches!( |
| 1535 | e, |
| 1536 | StreamEvent::ContentBlockDelta { |
| 1537 | delta: Delta::InputJsonDelta { partial_json }, |
| 1538 | .. |
| 1539 | } if partial_json.contains("\"pattern\"") |
| 1540 | )), |
| 1541 | "should yield InputJsonDelta carrying the tool args; got {events:?}" |
| 1542 | ); |
| 1543 | } |
| 1544 | |
| 1545 | /// Regression for the parallel-tool-calls-without-id collision (audit |
| 1546 | /// Finding 8): when the upstream chunk omits the `id` field, the |
| 1547 | /// fallback used to be the literal string `"tool_call"` for every |
| 1548 | /// parallel call, so two tool calls in one delta ended up sharing an |
| 1549 | /// id. Downstream routing then matched the first call's tool_result |
| 1550 | /// twice and the second call hung. The fallback is now indexed by the |
| 1551 | /// content-block position, keeping each call unique within the |
| 1552 | /// response. |
| 1553 | #[test] |
| 1554 | fn decoder_assigns_unique_fallback_ids_to_parallel_tool_calls_missing_id() { |
| 1555 | let events = decode_chunk( |
| 1556 | r#"{"choices":[{"delta":{"tool_calls":[ |
| 1557 | {"index":0,"function":{"name":"grep_files","arguments":"{\"pattern\":\"a\"}"}}, |
| 1558 | {"index":1,"function":{"name":"read_file","arguments":"{\"path\":\"x\"}"}} |
| 1559 | ]}}]}"#, |
| 1560 | ); |
| 1561 | |
| 1562 | let ids: Vec<&str> = events |
| 1563 | .iter() |
| 1564 | .filter_map(|e| match e { |
| 1565 | StreamEvent::ContentBlockStart { |
| 1566 | content_block: ContentBlockStart::ToolUse { id, .. }, |
| 1567 | .. |
| 1568 | } => Some(id.as_str()), |
| 1569 | _ => None, |
| 1570 | }) |
| 1571 | .collect(); |
| 1572 | |
| 1573 | assert_eq!( |
| 1574 | ids.len(), |
| 1575 | 2, |
| 1576 | "expected two tool-use blocks for parallel tool calls; got {events:?}" |
| 1577 | ); |
| 1578 | assert_ne!( |
| 1579 | ids[0], ids[1], |
| 1580 | "parallel tool calls without upstream `id` must get distinct fallback ids; got {ids:?}" |
| 1581 | ); |
| 1582 | } |
| 1583 | |
| 1584 | #[test] |
| 1585 | fn decoder_preserves_upstream_tool_call_id_when_present() { |
| 1586 | // Counter-test to the fallback regression: when the upstream chunk |
| 1587 | // does include `id`, we forward it verbatim — we shouldn't quietly |
| 1588 | // rewrite ids the API gave us just because we have a fallback path. |
| 1589 | let events = decode_chunk( |
| 1590 | r#"{"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_xyz","function":{"name":"grep_files","arguments":"{}"}}]}}]}"#, |
| 1591 | ); |
| 1592 | let id = events |
| 1593 | .iter() |
| 1594 | .find_map(|e| match e { |
| 1595 | StreamEvent::ContentBlockStart { |
| 1596 | content_block: ContentBlockStart::ToolUse { id, .. }, |
| 1597 | .. |
| 1598 | } => Some(id.as_str()), |
| 1599 | _ => None, |
| 1600 | }) |
| 1601 | .expect("tool-use block present"); |
| 1602 | assert_eq!(id, "call_xyz"); |
| 1603 | } |
| 1604 | } |
| 1605 |