diff --git a/src/conversion/chat_to_responses.rs b/src/conversion/chat_to_responses.rs index a236a6a..8f0f37b 100644 --- a/src/conversion/chat_to_responses.rs +++ b/src/conversion/chat_to_responses.rs @@ -3,7 +3,10 @@ use serde_json::{json, Value}; use uuid::Uuid; use super::responses_to_chat::repair_history; -use super::text::{arguments_text, as_text, canonicalize_json_string_if_parseable, reasoning_text}; +use super::text::{ + arguments_text, as_text, canonicalize_json_string_if_parseable, reasoning_text, + split_leading_think_block, +}; use super::tool_context::{ToolContext, ToolKind}; #[allow(clippy::too_many_arguments)] // All params map 1:1 to distinct JSON fields. @@ -26,8 +29,14 @@ where .cloned() .unwrap_or_else(|| json!({})); let message = choice.get("message").cloned().unwrap_or_else(|| json!({})); - let content = message.get("content").map(as_text).unwrap_or_default(); - let reasoning = reasoning_text(&message).unwrap_or_default(); + let mut content = message.get("content").map(as_text).unwrap_or_default(); + let mut reasoning = reasoning_text(&message).unwrap_or_default(); + if reasoning.is_empty() { + if let Some((extracted_reasoning, answer)) = split_leading_think_block(&content) { + reasoning = extracted_reasoning; + content = answer; + } + } let response_id = format!("resp_{}", Uuid::new_v4().simple()); let created_at = now_ts(); diff --git a/src/conversion/stream_chat_to_responses.rs b/src/conversion/stream_chat_to_responses.rs index 48501e2..7573818 100644 --- a/src/conversion/stream_chat_to_responses.rs +++ b/src/conversion/stream_chat_to_responses.rs @@ -5,7 +5,11 @@ use uuid::Uuid; use super::chat_to_responses::{completion_status, message_item, reasoning_item, response_shell}; use super::responses_to_chat::repair_history; -use super::text::{arguments_text, as_text, canonicalize_json_string_if_parseable, reasoning_text}; +use super::text::{ + arguments_text, as_text, canonicalize_json_string_if_parseable, is_leading_think_prefix, + reasoning_text, split_at_think_close, split_incomplete_think_close_suffix, + split_leading_think_block, strip_leading_think_open_tag, +}; use super::tool_context::{ToolContext, ToolKind}; pub type EmitFn = Box anyhow::Result<()> + Send>; @@ -35,6 +39,13 @@ impl StreamingToolCall { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ThinkBlockState { + Detecting, + InThink, + Done, +} + pub struct StreamAssembler { body: Value, model_alias: String, @@ -59,6 +70,8 @@ pub struct StreamAssembler { text_done: bool, reasoning_done: bool, terminal_emitted: bool, + think_state: ThinkBlockState, + think_buffer: String, } impl StreamAssembler { @@ -95,6 +108,8 @@ impl StreamAssembler { text_done: false, reasoning_done: false, terminal_emitted: false, + think_state: ThinkBlockState::Detecting, + think_buffer: String::new(), } } @@ -139,15 +154,17 @@ impl StreamAssembler { for choice in choices { let delta = choice.get("delta").unwrap_or(&Value::Null); if let Some(text) = reasoning_text(delta) { + self.disable_leading_think_detection()?; self.push_reasoning_delta(&text)?; } if let Some(content) = delta.get("content") { let text = as_text(content); if !text.is_empty() { - self.push_text_delta(&text)?; + self.accept_content_delta(&text)?; } } if let Some(calls) = delta.get("tool_calls").and_then(Value::as_array) { + self.flush_pending_think_buffer()?; if !self.reasoning.is_empty() && !self.reasoning_done { self.finish_reasoning_item()?; } @@ -172,6 +189,7 @@ impl StreamAssembler { if self.terminal_emitted { return Ok(json!({})); } + self.flush_pending_think_buffer()?; if !self.reasoning.is_empty() { self.finish_reasoning_item()?; } @@ -357,6 +375,94 @@ impl StreamAssembler { ) } + fn disable_leading_think_detection(&mut self) -> anyhow::Result<()> { + if self.think_state != ThinkBlockState::Done { + self.flush_pending_think_buffer()?; + self.think_state = ThinkBlockState::Done; + } + Ok(()) + } + + fn accept_content_delta(&mut self, text: &str) -> anyhow::Result<()> { + match self.think_state { + ThinkBlockState::Done => self.push_text_delta(text), + ThinkBlockState::Detecting => { + self.think_buffer.push_str(text); + if let Some((reasoning, answer)) = split_leading_think_block(&self.think_buffer) { + if !reasoning.is_empty() { + self.push_reasoning_delta(&reasoning)?; + } + self.think_buffer.clear(); + self.think_state = ThinkBlockState::Done; + if !answer.is_empty() { + self.push_text_delta(&answer)?; + } + return Ok(()); + } + if let Some(reasoning_start) = strip_leading_think_open_tag(&self.think_buffer) { + self.think_buffer.clear(); + self.think_state = ThinkBlockState::InThink; + if !reasoning_start.is_empty() { + self.accept_in_think_content(&reasoning_start)?; + } + return Ok(()); + } + if is_leading_think_prefix(&self.think_buffer) { + return Ok(()); + } + let text = std::mem::take(&mut self.think_buffer); + self.think_state = ThinkBlockState::Done; + self.push_text_delta(&text) + } + ThinkBlockState::InThink => self.accept_in_think_content(text), + } + } + + fn accept_in_think_content(&mut self, text: &str) -> anyhow::Result<()> { + self.think_buffer.push_str(text); + if let Some((reasoning, answer)) = split_at_think_close(&self.think_buffer) { + if !reasoning.is_empty() { + self.push_reasoning_delta(&reasoning)?; + } + self.think_buffer.clear(); + self.think_state = ThinkBlockState::Done; + if !answer.is_empty() { + self.push_text_delta(&answer)?; + } + return Ok(()); + } + + let (emit, keep) = split_incomplete_think_close_suffix(&self.think_buffer); + let emit = emit.to_string(); + let keep = keep.to_string(); + if !emit.is_empty() { + self.push_reasoning_delta(&emit)?; + } + self.think_buffer = keep; + Ok(()) + } + + fn flush_pending_think_buffer(&mut self) -> anyhow::Result<()> { + match self.think_state { + ThinkBlockState::Detecting => { + if !self.think_buffer.is_empty() { + let text = std::mem::take(&mut self.think_buffer); + self.think_state = ThinkBlockState::Done; + self.push_text_delta(&text)?; + } + } + ThinkBlockState::InThink => { + if !self.think_buffer.is_empty() { + let reasoning = std::mem::take(&mut self.think_buffer); + self.push_reasoning_delta(&reasoning)?; + } + self.think_state = ThinkBlockState::Done; + } + ThinkBlockState::Done => {} + } + Ok(()) + } + fn push_reasoning_delta(&mut self, delta: &str) -> anyhow::Result<()> { if self.reasoning_output_index.is_none() { let index = self.allocate_output_index(); @@ -570,6 +676,7 @@ impl StreamAssembler { pub fn has_substantive_output(&self) -> bool { !self.content.trim().is_empty() || !self.reasoning.trim().is_empty() + || !self.think_buffer.trim().is_empty() || self.tool_calls.values().any(|call| { call.added || !call.arguments.trim().is_empty() || !call.name.trim().is_empty() }) diff --git a/src/conversion/text.rs b/src/conversion/text.rs index cfd3597..28a1b8f 100644 --- a/src/conversion/text.rs +++ b/src/conversion/text.rs @@ -1,5 +1,8 @@ use serde_json::Value; +const THINK_OPEN_TAG: &str = concat!("<", "think", ">"); +const THINK_CLOSE_TAG: &str = concat!(""); + pub fn compact_json(value: &Value) -> String { serde_json::to_string(value).unwrap_or_else(|_| "null".to_string()) } @@ -148,6 +151,61 @@ pub fn canonicalize_json_string_if_parseable(value: &str) -> String { } } +pub fn split_leading_think_block(text: &str) -> Option<(String, String)> { + let leading_ws_len = text.len() - text.trim_start().len(); + let after_ws = &text[leading_ws_len..]; + if !after_ws.starts_with(THINK_OPEN_TAG) { + return None; + } + + let body_start = leading_ws_len + THINK_OPEN_TAG.len(); + let close_relative = text[body_start..].find(THINK_CLOSE_TAG)?; + let close_start = body_start + close_relative; + let answer_start = close_start + THINK_CLOSE_TAG.len(); + + Some(( + text[body_start..close_start].trim().to_string(), + strip_think_answer_separator(&text[answer_start..]).to_string(), + )) +} + +pub fn strip_leading_think_open_tag(text: &str) -> Option { + let leading_ws_len = text.len() - text.trim_start().len(); + let after_ws = &text[leading_ws_len..]; + after_ws + .strip_prefix(THINK_OPEN_TAG) + .map(|value| value.trim().to_string()) +} + +pub fn is_leading_think_prefix(text: &str) -> bool { + let leading_ws_len = text.len() - text.trim_start().len(); + let after_ws = &text[leading_ws_len..]; + after_ws.is_empty() || THINK_OPEN_TAG.starts_with(after_ws) +} + +pub fn split_at_think_close(text: &str) -> Option<(String, String)> { + let close_start = text.find(THINK_CLOSE_TAG)?; + let answer_start = close_start + THINK_CLOSE_TAG.len(); + Some(( + text[..close_start].to_string(), + strip_think_answer_separator(&text[answer_start..]).to_string(), + )) +} + +pub fn split_incomplete_think_close_suffix(text: &str) -> (&str, &str) { + for len in (1..THINK_CLOSE_TAG.len()).rev() { + if text.ends_with(&THINK_CLOSE_TAG[..len]) { + let split = text.len() - len; + return (&text[..split], &text[split..]); + } + } + (text, "") +} + +fn strip_think_answer_separator(text: &str) -> &str { + text.trim_start_matches(['\r', '\n', '\t', ' ']) +} + #[cfg(test)] mod tests { use super::*; @@ -246,4 +304,31 @@ mod tests { let msg = json!({"content": "just a normal message"}); assert!(reasoning_text(&msg).is_none()); } + + #[test] + fn split_leading_think_block_extracts_reasoning_and_answer() { + let text = format!(" {THINK_OPEN_TAG} hidden reasoning {THINK_CLOSE_TAG}\n\nfinal answer"); + let (reasoning, answer) = split_leading_think_block(&text).unwrap(); + assert_eq!(reasoning, "hidden reasoning"); + assert_eq!(answer, "final answer"); + } + + #[test] + fn split_leading_think_block_ignores_non_leading_block() { + let text = format!("answer first {THINK_OPEN_TAG}hidden{THINK_CLOSE_TAG}"); + assert!(split_leading_think_block(&text).is_none()); + } + + #[test] + fn leading_think_prefix_detects_partial_open_tag() { + assert!(is_leading_think_prefix(" "); +const CLOSE_TAG: &str = concat!(""); + +fn tagged_content(reasoning: &str, answer: &str) -> String { + format!("{OPEN_TAG}{reasoning}{CLOSE_TAG}\n{answer}") +} + +#[test] +fn nonstream_leading_reasoning_block_becomes_reasoning_item() { + let body = json!({ + "model": "opencode-go/test-model", + "input": "answer with reasoning", + "stream": false + }); + let (_payload, messages, _reverse, tool_ctx) = + build_chat_payload(&body, "test-model", None, json!({})).unwrap(); + let upstream = json!({ + "id": "chatcmpl_reasoning_block", + "choices": [{ + "message": { + "role": "assistant", + "content": tagged_content("hidden chain", "visible answer") + }, + "finish_reason": "stop" + }], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2} + }); + let mut stored = Vec::::new(); + + let response = build_response( + &body, + &upstream, + "opencode-go/test-model", + "test-model", + &messages, + &tool_ctx, + |item| { + stored.push(item); + Ok(()) + }, + ) + .unwrap(); + + let output = response["output"].as_array().unwrap(); + let reasoning = output + .iter() + .find(|item| item["type"] == "reasoning") + .unwrap(); + assert_eq!(reasoning["summary"][0]["text"], "hidden chain"); + + let message = output + .iter() + .find(|item| item["type"] == "message") + .unwrap(); + assert_eq!(message["content"][0]["text"], "visible answer"); + assert!(!message["content"][0]["text"] + .as_str() + .unwrap() + .contains("hidden chain")); + assert_eq!(stored[0].messages.last().unwrap()["content"], "visible answer"); + assert_eq!( + stored[0].messages.last().unwrap()["reasoning_content"], + "hidden chain" + ); +} + +#[test] +fn stream_leading_reasoning_block_is_not_emitted_as_output_text() { + let body = json!({ + "model": "opencode-go/test-model", + "input": "answer with reasoning", + "stream": true + }); + let (_payload, messages, _reverse, tool_ctx) = + build_chat_payload(&body, "test-model", None, json!({})).unwrap(); + let events = Arc::new(Mutex::new(Vec::<(String, Value)>::new())); + let stored = Arc::new(Mutex::new(Vec::::new())); + let events_for_emit = Arc::clone(&events); + let stored_for_put = Arc::clone(&stored); + let mut assembler = StreamAssembler::new( + body, + "opencode-go/test-model".to_string(), + "test-model".to_string(), + messages, + tool_ctx, + Box::new(move |item| { + stored_for_put.lock().unwrap().push(item); + Ok(()) + }), + Box::new(move |event, data| { + events_for_emit + .lock() + .unwrap() + .push((event.to_string(), data)); + Ok(()) + }), + ); + + assembler.start().unwrap(); + assembler + .accept(&json!({"choices":[{"delta":{"content":"hidden chain\nvisible answer"},"finish_reason":"stop"}]})) + .unwrap(); + let response = assembler.finalize().unwrap(); + + let events = events.lock().unwrap(); + let reasoning_text = events + .iter() + .filter(|(name, _)| name == "response.reasoning_summary_text.delta") + .filter_map(|(_, data)| data["delta"].as_str()) + .collect::(); + assert_eq!(reasoning_text, "hidden chain"); + + let output_text = events + .iter() + .filter(|(name, _)| name == "response.output_text.delta") + .filter_map(|(_, data)| data["delta"].as_str()) + .collect::(); + assert_eq!(output_text, "visible answer"); + assert!(!output_text.contains("hidden chain")); + assert!(!output_text.contains("think")); + + let output = response["output"].as_array().unwrap(); + assert!(output.iter().any(|item| item["type"] == "reasoning")); + let message = output + .iter() + .find(|item| item["type"] == "message") + .unwrap(); + assert_eq!(message["content"][0]["text"], "visible answer"); +}