From 290abffe42e0017d4e6bc4932b8d740d49627168 Mon Sep 17 00:00:00 2001 From: mulfyx Date: Tue, 14 Jul 2026 08:29:34 +0500 Subject: [PATCH] preserve Codex encrypted reasoning across turns --- src/providers/codex/count_tokens.rs | 20 ++ src/providers/codex/request_summary.rs | 2 + src/providers/codex/translate/accumulate.rs | 53 +++- src/providers/codex/translate/live_stream.rs | 161 +++++++++++- src/providers/codex/translate/mod.rs | 1 + .../codex/translate/reasoning_signature.rs | 135 ++++++++++ src/providers/codex/translate/reducer.rs | 236 ++++++++++++++++-- src/providers/codex/translate/request.rs | 63 +++++ src/providers/codex/translate/stream.rs | 56 +++++ src/providers/kimi/translate/request.rs | 2 +- src/providers/translate_shared.rs | 10 +- 11 files changed, 704 insertions(+), 35 deletions(-) create mode 100644 src/providers/codex/translate/reasoning_signature.rs diff --git a/src/providers/codex/count_tokens.rs b/src/providers/codex/count_tokens.rs index a0f546c..4ef09a7 100644 --- a/src/providers/codex/count_tokens.rs +++ b/src/providers/codex/count_tokens.rs @@ -50,9 +50,22 @@ fn count_input_item_tokens(item: &ResponsesInputItem) -> u64 { name, arguments, .. } => approx_token_count(name) + approx_token_count(arguments), ResponsesInputItem::FunctionCallOutput { output, .. } => approx_token_count(output), + ResponsesInputItem::Reasoning { + encrypted_content, .. + } => approx_reasoning_token_count(encrypted_content), } } +fn approx_reasoning_token_count(encoded_content: &str) -> u64 { + let model_visible_bytes = encoded_content + .len() + .saturating_mul(3) + .checked_div(4) + .unwrap_or(0) + .saturating_sub(650); + u64::try_from(model_visible_bytes.saturating_add(3) / 4).unwrap_or(u64::MAX) +} + fn count_content_part_tokens(part: &ResponsesContentPart) -> u64 { match part { ResponsesContentPart::InputText { text } => approx_token_count(text), @@ -163,4 +176,11 @@ mod tests { .unwrap(); assert!(count_translated_tokens(&long) >= count_translated_tokens(&short)); } + + #[test] + fn encrypted_reasoning_uses_codex_model_visible_size_estimate() { + let encoded_content = "A".repeat(4000); + assert_eq!(approx_reasoning_token_count(&encoded_content), 588); + assert_eq!(approx_reasoning_token_count("short"), 0); + } } diff --git a/src/providers/codex/request_summary.rs b/src/providers/codex/request_summary.rs index cd04ac4..6099133 100644 --- a/src/providers/codex/request_summary.rs +++ b/src/providers/codex/request_summary.rs @@ -83,6 +83,7 @@ pub fn summarize_codex_request_size(body: &ResponsesRequest) -> CodexRequestSize ResponsesInputItem::Message { .. } => Some("message".to_string()), ResponsesInputItem::FunctionCall { .. } => Some("function_call".to_string()), ResponsesInputItem::FunctionCallOutput { .. } => Some("function_call_output".to_string()), + ResponsesInputItem::Reasoning { .. } => Some("reasoning".to_string()), }); let role_counts = count_items_by(&body.input, |item| match item { @@ -108,6 +109,7 @@ pub fn summarize_codex_request_size(body: &ResponsesRequest) -> CodexRequestSize ResponsesInputItem::FunctionCallOutput { .. } => { ("function_call_output".to_string(), None) } + ResponsesInputItem::Reasoning { .. } => ("reasoning".to_string(), None), }; let json_bytes_val = json_bytes(Some(&serde_json::to_value(item).unwrap_or_default())); diff --git a/src/providers/codex/translate/accumulate.rs b/src/providers/codex/translate/accumulate.rs index 3df5c5a..6656c12 100644 --- a/src/providers/codex/translate/accumulate.rs +++ b/src/providers/codex/translate/accumulate.rs @@ -43,6 +43,7 @@ pub fn accumulate_response_with_traffic( enum BlockKind { Thinking { text: String, + signature: String, }, Text { text: String, @@ -69,16 +70,24 @@ pub fn accumulate_response_with_traffic( index: *index, kind: BlockKind::Thinking { text: String::new(), + signature: String::new(), }, }); } ReducerEvent::ThinkingDelta { index, text } => { if let Some(block) = blocks.iter_mut().rev().find(|b| b.index == *index) - && let BlockKind::Thinking { text: t } = &mut block.kind + && let BlockKind::Thinking { text: t, .. } = &mut block.kind { t.push_str(text); } } + ReducerEvent::ThinkingSignature { index, signature } => { + if let Some(block) = blocks.iter_mut().rev().find(|b| b.index == *index) + && let BlockKind::Thinking { signature: s, .. } = &mut block.kind + { + *s = signature.clone(); + } + } ReducerEvent::TextStart { index } => { blocks.push(AccumulatedBlock { index: *index, @@ -177,14 +186,14 @@ pub fn accumulate_response_with_traffic( for block in &blocks { match &block.kind { - BlockKind::Thinking { text } => { - if !text.is_empty() { + BlockKind::Thinking { text, signature } => { + if !text.is_empty() || !signature.is_empty() { indexed_content.push(( block.index, serde_json::json!({ "type": "thinking", "thinking": text, - "signature": "", + "signature": signature, }), )); } @@ -535,4 +544,40 @@ mod tests { let result = accumulate_response(upstream.as_bytes(), "msg_e", "model"); assert!(result.is_err()); } + + #[test] + fn accumulate_preserves_signature_without_visible_summary() { + let upstream = format!( + "{}{}{}", + sse_event( + "response.output_item.added", + json!({ + "output_index":0, + "item":{"type":"reasoning","id":"rs_1","encrypted_content":"opaque"} + }) + ), + sse_event( + "response.output_item.done", + json!({ + "output_index":0, + "item":{"type":"reasoning","id":"rs_1"} + }) + ), + sse_event( + "response.completed", + json!({"response":{"id":"resp_1","usage":{}}}) + ), + ); + let response = accumulate_response(upstream.as_bytes(), "msg_1", "gpt-5.5").unwrap(); + let content = response["content"].as_array().unwrap(); + assert_eq!(content.len(), 1); + assert_eq!(content[0]["type"], "thinking"); + assert_eq!(content[0]["thinking"], ""); + assert!( + content[0]["signature"] + .as_str() + .unwrap() + .starts_with("ccp:codex:v1:") + ); + } } diff --git a/src/providers/codex/translate/live_stream.rs b/src/providers/codex/translate/live_stream.rs index fe78e5b..cfc4844 100644 --- a/src/providers/codex/translate/live_stream.rs +++ b/src/providers/codex/translate/live_stream.rs @@ -4,6 +4,7 @@ use crate::anthropic::sse::encode_sse_event; use crate::traffic::TrafficCapture; use super::read_rewrite::sanitize_read_args; +use super::reasoning_signature::{PendingReasoning, encode_reasoning_signature}; use super::reducer::{ CodexUsage, STOP_END_TURN, STOP_MAX_TOKENS, STOP_TOOL_USE, map_codex_usage_to_anthropic, }; @@ -41,6 +42,12 @@ struct LiveWebSearchResult { url: String, } +#[derive(Clone, Copy)] +struct LiveThinking { + output_index: usize, + anthropic_index: usize, +} + pub struct LiveStreamTranslator { message_id: String, model: String, @@ -48,7 +55,8 @@ pub struct LiveStreamTranslator { blocks_by_output_index: HashMap, item_id_to_output_index: HashMap, anthropic_index: usize, - thinking_index: Option, + thinking: Option, + reasoning_by_output_index: HashMap, saw_tool_use: bool, web_search_requests: usize, web_searches: Vec, @@ -66,7 +74,8 @@ impl LiveStreamTranslator { blocks_by_output_index: HashMap::new(), item_id_to_output_index: HashMap::new(), anthropic_index: 0, - thinking_index: None, + thinking: None, + reasoning_by_output_index: HashMap::new(), saw_tool_use: false, web_search_requests: 0, web_searches: Vec::new(), @@ -110,14 +119,18 @@ impl LiveStreamTranslator { self.output_item_added(payload, traffic, &mut out); } "response.reasoning_summary_part.added" => { - if let Some(index) = self.thinking_index { + let output_index = output_index(payload); + if let Some(thinking) = self + .thinking + .filter(|thinking| thinking.output_index == output_index) + { self.emit( traffic, &mut out, "content_block_delta", &serde_json::json!({ "type": "content_block_delta", - "index": index, + "index": thinking.anthropic_index, "delta": {"type": "thinking_delta", "thinking": "\n\n"} }), ); @@ -256,6 +269,12 @@ impl LiveStreamTranslator { let item_type = item.get("type").and_then(|v| v.as_str()).unwrap_or(""); match item_type { + "reasoning" => { + self.reasoning_by_output_index + .entry(output_index) + .or_default() + .capture(item); + } "message" => { self.close_thinking(traffic, out); let index = self.anthropic_index; @@ -344,14 +363,19 @@ impl LiveStreamTranslator { traffic: Option<&TrafficCapture>, out: &mut Vec, ) { + let output_index = output_index(payload); let delta = payload.get("delta").and_then(|v| v.as_str()).unwrap_or(""); if delta.is_empty() { return; } - if self.thinking_index.is_none() { + if self.thinking.map(|thinking| thinking.output_index) != Some(output_index) { + self.close_thinking(traffic, out); let index = self.anthropic_index; self.anthropic_index += 1; - self.thinking_index = Some(index); + self.thinking = Some(LiveThinking { + output_index, + anthropic_index: index, + }); self.ensure_message_start(traffic, out); self.emit( traffic, @@ -364,7 +388,10 @@ impl LiveStreamTranslator { }), ); } - let index = self.thinking_index.unwrap(); + let index = self + .thinking + .expect("thinking block was started") + .anthropic_index; self.emit( traffic, out, @@ -575,13 +602,24 @@ impl LiveStreamTranslator { out: &mut Vec, ) { let output_index = output_index(payload); - if payload + if let Some(item) = payload .get("item") .and_then(|item| item.get("type")) .and_then(|v| v.as_str()) - == Some("reasoning") + .filter(|item_type| *item_type == "reasoning") + .and_then(|_| payload.get("item")) { + self.reasoning_by_output_index + .entry(output_index) + .or_default() + .capture(item); + let had_active_summary = self + .thinking + .is_some_and(|thinking| thinking.output_index == output_index); self.close_thinking(traffic, out); + if !had_active_summary { + self.emit_signature_only_reasoning(output_index, traffic, out); + } return; } @@ -896,10 +934,45 @@ impl LiveStreamTranslator { } } - fn close_thinking(&mut self, traffic: Option<&TrafficCapture>, out: &mut Vec) { - let Some(index) = self.thinking_index.take() else { + fn emit_signature_only_reasoning( + &mut self, + output_index: usize, + traffic: Option<&TrafficCapture>, + out: &mut Vec, + ) { + let Some(replay) = self + .reasoning_by_output_index + .remove(&output_index) + .and_then(|pending| pending.replay()) + else { return; }; + let Some(signature) = encode_reasoning_signature(&replay) else { + return; + }; + let index = self.anthropic_index; + self.anthropic_index += 1; + self.ensure_message_start(traffic, out); + self.emit( + traffic, + out, + "content_block_start", + &serde_json::json!({ + "type": "content_block_start", + "index": index, + "content_block": {"type": "thinking", "thinking": "", "signature": ""} + }), + ); + self.emit( + traffic, + out, + "content_block_delta", + &serde_json::json!({ + "type": "content_block_delta", + "index": index, + "delta": {"type": "signature_delta", "signature": signature} + }), + ); self.emit( traffic, out, @@ -910,6 +983,38 @@ impl LiveStreamTranslator { }), ); } + + fn close_thinking(&mut self, traffic: Option<&TrafficCapture>, out: &mut Vec) { + let Some(thinking) = self.thinking.take() else { + return; + }; + if let Some(signature) = self + .reasoning_by_output_index + .remove(&thinking.output_index) + .and_then(|pending| pending.replay()) + .and_then(|replay| encode_reasoning_signature(&replay)) + { + self.emit( + traffic, + out, + "content_block_delta", + &serde_json::json!({ + "type": "content_block_delta", + "index": thinking.anthropic_index, + "delta": {"type": "signature_delta", "signature": signature} + }), + ); + } + self.emit( + traffic, + out, + "content_block_stop", + &serde_json::json!({ + "type": "content_block_stop", + "index": thinking.anthropic_index, + }), + ); + } } fn web_search_query(item: &serde_json::Value) -> String { @@ -1302,4 +1407,38 @@ mod tests { .unwrap_err(); assert_eq!(err, "rate limit reached"); } + + #[test] + fn live_stream_emits_signature_delta_before_thinking_stop() { + let out = render(vec![ + json!({ + "type":"response.output_item.added", + "output_index":0, + "item":{"type":"reasoning","id":"rs_1","encrypted_content":"opaque"} + }), + json!({ + "type":"response.reasoning_summary_text.delta", + "output_index":0, + "delta":"plan" + }), + json!({ + "type":"response.output_item.done", + "output_index":0, + "item":{"type":"reasoning","id":"rs_1"} + }), + json!({ + "type":"response.completed", + "response":{"id":"resp_1","usage":{}} + }), + ]); + let thinking_delta = out.find(r#""type":"thinking_delta""#).unwrap(); + let signature_delta = out.find(r#""type":"signature_delta""#).unwrap(); + let thinking_stop = out[signature_delta..] + .find("event: content_block_stop") + .map(|offset| signature_delta + offset) + .unwrap(); + assert!(thinking_delta < signature_delta); + assert!(signature_delta < thinking_stop); + assert!(out.contains("ccp:codex:v1:")); + } } diff --git a/src/providers/codex/translate/mod.rs b/src/providers/codex/translate/mod.rs index 74fd510..cdf6f42 100644 --- a/src/providers/codex/translate/mod.rs +++ b/src/providers/codex/translate/mod.rs @@ -2,6 +2,7 @@ pub mod accumulate; pub mod live_stream; pub mod model_allowlist; pub mod read_rewrite; +pub mod reasoning_signature; pub mod reducer; pub mod request; pub mod stream; diff --git a/src/providers/codex/translate/reasoning_signature.rs b/src/providers/codex/translate/reasoning_signature.rs new file mode 100644 index 0000000..78a88bf --- /dev/null +++ b/src/providers/codex/translate/reasoning_signature.rs @@ -0,0 +1,135 @@ +use base64::Engine; +use serde_json::Value; + +const PREFIX: &str = "ccp:codex:v1:"; +const MAX_ID_BYTES: usize = 4 * 1024; +const MAX_ENCRYPTED_CONTENT_BYTES: usize = 8 * 1024 * 1024; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ReasoningReplay { + pub id: String, + pub encrypted_content: String, +} + +#[derive(Debug, Clone, Default)] +pub struct PendingReasoning { + id: Option, + encrypted_content: Option, +} + +impl PendingReasoning { + pub fn capture(&mut self, item: &Value) { + if let Some(id) = non_empty_string(item.get("id")) { + self.id = Some(id.to_string()); + } + if let Some(encrypted_content) = non_empty_string(item.get("encrypted_content")) { + self.encrypted_content = Some(encrypted_content.to_string()); + } + } + + pub fn replay(&self) -> Option { + Some(ReasoningReplay { + id: self.id.clone()?, + encrypted_content: self.encrypted_content.clone()?, + }) + } +} + +pub fn encode_reasoning_signature(replay: &ReasoningReplay) -> Option { + if replay.id.is_empty() + || replay.id.len() > MAX_ID_BYTES + || replay.encrypted_content.is_empty() + || replay.encrypted_content.len() > MAX_ENCRYPTED_CONTENT_BYTES + { + return None; + } + let encoded_id = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(replay.id.as_bytes()); + Some(format!("{PREFIX}{encoded_id}:{}", replay.encrypted_content)) +} + +pub fn decode_reasoning_signature(signature: &str) -> Option { + let payload = signature.strip_prefix(PREFIX)?; + if payload.is_empty() || payload.len() > max_payload_len() { + return None; + } + let (encoded_id, encrypted_content) = payload.split_once(':')?; + if encoded_id.is_empty() + || encoded_id.len() > encoded_id_len_limit() + || encrypted_content.is_empty() + || encrypted_content.len() > MAX_ENCRYPTED_CONTENT_BYTES + { + return None; + } + let id = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(encoded_id) + .ok()?; + if id.is_empty() || id.len() > MAX_ID_BYTES { + return None; + } + Some(ReasoningReplay { + id: String::from_utf8(id).ok()?, + encrypted_content: encrypted_content.to_string(), + }) +} + +fn encoded_id_len_limit() -> usize { + (MAX_ID_BYTES + 2) / 3 * 4 +} + +fn max_payload_len() -> usize { + encoded_id_len_limit() + 1 + MAX_ENCRYPTED_CONTENT_BYTES +} + +fn non_empty_string(value: Option<&Value>) -> Option<&str> { + value?.as_str().filter(|value| !value.is_empty()) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn signature_round_trip_preserves_reasoning_identity() { + let replay = ReasoningReplay { + id: "rs_1".to_string(), + encrypted_content: "gAAAAopaque".to_string(), + }; + let signature = encode_reasoning_signature(&replay).unwrap(); + assert!(signature.starts_with(PREFIX)); + assert!(signature.ends_with(":gAAAAopaque")); + assert_eq!(decode_reasoning_signature(&signature), Some(replay)); + } + + #[test] + fn foreign_and_malformed_signatures_are_ignored() { + assert_eq!(decode_reasoning_signature("anthropic-signature"), None); + assert_eq!(decode_reasoning_signature("ccp:codex:v1:not-base64"), None); + } + + #[test] + fn pending_reasoning_keeps_early_metadata_when_done_omits_it() { + let mut pending = PendingReasoning::default(); + pending.capture(&json!({ + "id": "rs_1", + "encrypted_content": "early" + })); + pending.capture(&json!({"id": "rs_1"})); + assert_eq!( + pending.replay(), + Some(ReasoningReplay { + id: "rs_1".to_string(), + encrypted_content: "early".to_string(), + }) + ); + } + + #[test] + fn oversized_signature_is_ignored_without_decoding() { + let signature = format!( + "{PREFIX}cnNfMQ:{}", + "A".repeat(MAX_ENCRYPTED_CONTENT_BYTES + 1) + ); + assert_eq!(decode_reasoning_signature(&signature), None); + } +} diff --git a/src/providers/codex/translate/reducer.rs b/src/providers/codex/translate/reducer.rs index 295c4ed..79a9635 100644 --- a/src/providers/codex/translate/reducer.rs +++ b/src/providers/codex/translate/reducer.rs @@ -1,6 +1,7 @@ use crate::anthropic::sse::parse_sse_events; use super::read_rewrite::sanitize_read_args; +use super::reasoning_signature::{PendingReasoning, ReasoningReplay, encode_reasoning_signature}; use super::request::ResponsesInputItem; #[derive(Debug, Clone)] @@ -72,6 +73,10 @@ pub enum ReducerEvent { index: usize, text: String, }, + ThinkingSignature { + index: usize, + signature: String, + }, ThinkingStop { index: usize, }, @@ -143,6 +148,82 @@ enum BlockState { }, } +#[derive(Clone, Copy)] +struct ActiveThinking { + output_index: usize, + anthropic_index: usize, +} + +fn reasoning_input_item(replay: ReasoningReplay) -> ResponsesInputItem { + ResponsesInputItem::Reasoning { + id: replay.id, + summary: Vec::new(), + encrypted_content: replay.encrypted_content, + } +} + +fn finalize_active_thinking( + active: ActiveThinking, + out: &mut Vec, + reasoning_by_output_index: &mut std::collections::HashMap, + output_items_by_index: &mut std::collections::BTreeMap, +) { + if let Some(replay) = reasoning_by_output_index + .remove(&active.output_index) + .and_then(|pending| pending.replay()) + && let Some(signature) = encode_reasoning_signature(&replay) + { + out.push(ReducerEvent::ThinkingSignature { + index: active.anthropic_index, + signature, + }); + output_items_by_index.insert(active.output_index, reasoning_input_item(replay)); + } + out.push(ReducerEvent::ThinkingStop { + index: active.anthropic_index, + }); +} + +fn close_thinking( + out: &mut Vec, + active_thinking: &mut Option, + reasoning_by_output_index: &mut std::collections::HashMap, + output_items_by_index: &mut std::collections::BTreeMap, +) { + if let Some(active) = active_thinking.take() { + finalize_active_thinking( + active, + out, + reasoning_by_output_index, + output_items_by_index, + ); + } +} + +fn emit_signature_only_reasoning( + output_index: usize, + anthropic_index: &mut usize, + out: &mut Vec, + reasoning_by_output_index: &mut std::collections::HashMap, + output_items_by_index: &mut std::collections::BTreeMap, +) { + let Some(replay) = reasoning_by_output_index + .remove(&output_index) + .and_then(|pending| pending.replay()) + else { + return; + }; + let Some(signature) = encode_reasoning_signature(&replay) else { + return; + }; + let index = *anthropic_index; + *anthropic_index += 1; + out.push(ReducerEvent::ThinkingStart { index }); + out.push(ReducerEvent::ThinkingSignature { index, signature }); + out.push(ReducerEvent::ThinkingStop { index }); + output_items_by_index.insert(output_index, reasoning_input_item(replay)); +} + pub fn finish_metadata_from_upstream( input: &[u8], ) -> Result, UpstreamStreamError> { @@ -172,8 +253,10 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream std::collections::BTreeMap::new(); let mut item_id_to_output_index: std::collections::HashMap = std::collections::HashMap::new(); + let mut reasoning_by_output_index: std::collections::HashMap = + std::collections::HashMap::new(); let mut anthropic_index = 0usize; - let mut thinking_index: Option = None; + let mut active_thinking: Option = None; let mut saw_tool_use = false; let mut final_usage: Option = None; let mut response_id: Option = None; @@ -226,12 +309,6 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream } } - fn close_thinking(out: &mut Vec, thinking_index: &mut Option) { - if let Some(index) = thinking_index.take() { - out.push(ReducerEvent::ThinkingStop { index }); - } - } - for evt in &sse_events { let data = evt.data.trim(); if data.is_empty() { @@ -318,6 +395,10 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream let item_type = item.get("type").and_then(|v| v.as_str()).unwrap_or(""); if item_type == "reasoning" { + reasoning_by_output_index + .entry(output_index) + .or_default() + .capture(item); continue; } if item_type == "web_search_call" { @@ -326,7 +407,12 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream } if item_type == "message" { - close_thinking(&mut out, &mut thinking_index); + close_thinking( + &mut out, + &mut active_thinking, + &mut reasoning_by_output_index, + &mut output_items_by_index, + ); let idx = anthropic_index; anthropic_index += 1; if let Some(id) = item.get("id").and_then(|v| v.as_str()) { @@ -344,7 +430,12 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream } if item_type == "function_call" { - close_thinking(&mut out, &mut thinking_index); + close_thinking( + &mut out, + &mut active_thinking, + &mut reasoning_by_output_index, + &mut output_items_by_index, + ); saw_tool_use = true; let idx = anthropic_index; anthropic_index += 1; @@ -384,9 +475,12 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream } if t == "response.reasoning_summary_part.added" { - if let Some(index) = thinking_index { + let output_index = p.get("output_index").and_then(|v| v.as_u64()).unwrap_or(0) as usize; + if let Some(active) = + active_thinking.filter(|active| active.output_index == output_index) + { out.push(ReducerEvent::ThinkingDelta { - index, + index: active.anthropic_index, text: "\n\n".to_string(), }); } @@ -394,25 +488,42 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream } if t == "response.reasoning_summary_text.delta" { + let output_index = p.get("output_index").and_then(|v| v.as_u64()).unwrap_or(0) as usize; let delta = p.get("delta").and_then(|v| v.as_str()).unwrap_or(""); if delta.is_empty() { continue; } - if thinking_index.is_none() { + if active_thinking.map(|active| active.output_index) != Some(output_index) { + close_thinking( + &mut out, + &mut active_thinking, + &mut reasoning_by_output_index, + &mut output_items_by_index, + ); let index = anthropic_index; anthropic_index += 1; - thinking_index = Some(index); + active_thinking = Some(ActiveThinking { + output_index, + anthropic_index: index, + }); out.push(ReducerEvent::ThinkingStart { index }); } out.push(ReducerEvent::ThinkingDelta { - index: thinking_index.unwrap(), + index: active_thinking + .expect("thinking block was started") + .anthropic_index, text: delta.to_string(), }); continue; } if t == "response.output_text.delta" { - close_thinking(&mut out, &mut thinking_index); + close_thinking( + &mut out, + &mut active_thinking, + &mut reasoning_by_output_index, + &mut output_items_by_index, + ); let output_index = p .get("output_index") .and_then(|v| v.as_u64()) @@ -550,14 +661,39 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream if let Some(item_val) = item && item_val.get("type").and_then(|v| v.as_str()) == Some("reasoning") { - close_thinking(&mut out, &mut thinking_index); + reasoning_by_output_index + .entry(output_index) + .or_default() + .capture(item_val); + let had_active_summary = + active_thinking.is_some_and(|active| active.output_index == output_index); + close_thinking( + &mut out, + &mut active_thinking, + &mut reasoning_by_output_index, + &mut output_items_by_index, + ); + if !had_active_summary { + emit_signature_only_reasoning( + output_index, + &mut anthropic_index, + &mut out, + &mut reasoning_by_output_index, + &mut output_items_by_index, + ); + } continue; } if let Some(item_val) = item && item_val.get("type").and_then(|v| v.as_str()) == Some("web_search_call") { - close_thinking(&mut out, &mut thinking_index); + close_thinking( + &mut out, + &mut active_thinking, + &mut reasoning_by_output_index, + &mut output_items_by_index, + ); let idx = anthropic_index; anthropic_index += 1; let result_index = anthropic_index; @@ -673,7 +809,12 @@ pub fn reduce_upstream_bytes(input: &[u8]) -> Result, Upstream }); } - close_thinking(&mut out, &mut thinking_index); + close_thinking( + &mut out, + &mut active_thinking, + &mut reasoning_by_output_index, + &mut output_items_by_index, + ); let stop_reason: StopReason = if incomplete { STOP_MAX_TOKENS @@ -1576,4 +1717,63 @@ mod tests { .collect(); assert_eq!(deltas, vec!["first", "second"]); } + + #[test] + fn reasoning_signature_precedes_stop_and_enters_continuation_transcript() { + let upstream = format!( + "{}{}{}{}", + sse( + "response.output_item.added", + json!({ + "output_index":0, + "item":{"type":"reasoning","id":"rs_1","summary":[],"encrypted_content":"opaque"} + }) + ), + sse( + "response.reasoning_summary_text.delta", + json!({"output_index":0,"summary_index":0,"delta":"plan"}) + ), + sse( + "response.output_item.done", + json!({ + "output_index":0, + "item":{"type":"reasoning","id":"rs_1","summary":[]} + }) + ), + sse( + "response.completed", + json!({"response":{"id":"resp_1","usage":{}}}) + ), + ); + let events = reduce_upstream_bytes(upstream.as_bytes()).unwrap(); + let signature_index = events + .iter() + .position(|event| matches!(event, ReducerEvent::ThinkingSignature { .. })) + .unwrap(); + let stop_index = events + .iter() + .position(|event| matches!(event, ReducerEvent::ThinkingStop { .. })) + .unwrap(); + assert!(signature_index < stop_index); + + let ReducerEvent::ThinkingSignature { signature, .. } = &events[signature_index] else { + unreachable!(); + }; + let replay = + super::super::reasoning_signature::decode_reasoning_signature(signature).unwrap(); + assert_eq!(replay.id, "rs_1"); + assert_eq!(replay.encrypted_content, "opaque"); + + let ReducerEvent::Finish { output_items, .. } = events.last().unwrap() else { + panic!("expected Finish"); + }; + assert!(matches!( + output_items.as_slice(), + [ResponsesInputItem::Reasoning { + id, + encrypted_content, + .. + }] if id == "rs_1" && encrypted_content == "opaque" + )); + } } diff --git a/src/providers/codex/translate/request.rs b/src/providers/codex/translate/request.rs index 05f24c1..6c8674e 100644 --- a/src/providers/codex/translate/request.rs +++ b/src/providers/codex/translate/request.rs @@ -10,6 +10,7 @@ use crate::providers::translate_shared::{ }; use super::read_rewrite::{ReadOffsetRewrite, read_offset_rewrite}; +use super::reasoning_signature::decode_reasoning_signature; // --------------------------------------------------------------------------- // Types @@ -152,6 +153,12 @@ pub enum ResponsesInputItem { call_id: String, output: String, }, + #[serde(rename = "reasoning")] + Reasoning { + id: String, + summary: Vec, + encrypted_content: String, + }, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -742,6 +749,19 @@ fn build_input(req: &MessagesRequest) -> Vec { arguments: args, }); } + ContentBlock::Thinking { signature, .. } => { + let Some(replay) = + signature.as_deref().and_then(decode_reasoning_signature) + else { + continue; + }; + flush_text(&mut out, &mut text_parts); + out.push(ResponsesInputItem::Reasoning { + id: replay.id, + summary: Vec::new(), + encrypted_content: replay.encrypted_content, + }); + } _ => {} } } @@ -1666,4 +1686,47 @@ mod tests { assert!(keys.contains(*key), "missing key: {key}"); } } + + #[test] + fn assistant_thinking_signature_replays_codex_reasoning_item() { + let replay = super::super::reasoning_signature::ReasoningReplay { + id: "rs_1".to_string(), + encrypted_content: "opaque".to_string(), + }; + let signature = + super::super::reasoning_signature::encode_reasoning_signature(&replay).unwrap(); + let req: MessagesRequest = serde_json::from_value(json!({ + "model": "gpt-5.5", + "messages": [ + {"role":"user","content":"start"}, + {"role":"assistant","content":[ + {"type":"thinking","thinking":"visible summary","signature":signature}, + {"type":"text","text":"done"} + ]}, + {"role":"user","content":"continue"} + ] + })) + .unwrap(); + let out = translate_request(&req, opts()).unwrap(); + let reasoning_index = out + .input + .iter() + .position(|item| matches!(item, ResponsesInputItem::Reasoning { .. })) + .unwrap(); + let ResponsesInputItem::Reasoning { + id, + summary, + encrypted_content, + } = &out.input[reasoning_index] + else { + unreachable!(); + }; + assert_eq!(id, "rs_1"); + assert!(summary.is_empty()); + assert_eq!(encrypted_content, "opaque"); + assert!(matches!( + out.input.get(reasoning_index + 1), + Some(ResponsesInputItem::Message { role, .. }) if role == "assistant" + )); + } } diff --git a/src/providers/codex/translate/stream.rs b/src/providers/codex/translate/stream.rs index e12a219..c3f3076 100644 --- a/src/providers/codex/translate/stream.rs +++ b/src/providers/codex/translate/stream.rs @@ -100,6 +100,19 @@ fn emit_content_event( ); true } + ReducerEvent::ThinkingSignature { index, signature } => { + emit( + out, + traffic, + "content_block_delta", + &serde_json::json!({ + "type": "content_block_delta", + "index": index, + "delta": {"type": "signature_delta", "signature": signature} + }), + ); + true + } ReducerEvent::ThinkingStop { index } => { open_blocks.remove(index); emit( @@ -221,6 +234,7 @@ fn is_content_event(event: &ReducerEvent) -> bool { event, ReducerEvent::ThinkingStart { .. } | ReducerEvent::ThinkingDelta { .. } + | ReducerEvent::ThinkingSignature { .. } | ReducerEvent::ThinkingStop { .. } | ReducerEvent::TextStart { .. } | ReducerEvent::TextDelta { .. } @@ -687,4 +701,46 @@ mod tests { assert!(!out.contains("\"type\":\"thinking_delta\"")); assert!(out.contains("event: message_stop")); } + + #[test] + fn stream_emits_signature_delta_before_thinking_stop() { + let upstream = format!( + "{}{}{}{}", + sse_event( + "response.output_item.added", + serde_json::json!({ + "output_index":0, + "item":{"type":"reasoning","id":"rs_1","encrypted_content":"opaque"} + }) + ), + sse_event( + "response.reasoning_summary_text.delta", + serde_json::json!({"output_index":0,"delta":"plan"}) + ), + sse_event( + "response.output_item.done", + serde_json::json!({ + "output_index":0, + "item":{"type":"reasoning","id":"rs_1"} + }) + ), + sse_event( + "response.completed", + serde_json::json!({"response":{"id":"resp_1","usage":{}}}) + ), + ); + let out = String::from_utf8( + translate_stream_bytes(upstream.as_bytes(), "msg_1", "gpt-5.5").unwrap(), + ) + .unwrap(); + let thinking_delta = out.find(r#""type":"thinking_delta""#).unwrap(); + let signature_delta = out.find(r#""type":"signature_delta""#).unwrap(); + let thinking_stop = out[signature_delta..] + .find("event: content_block_stop") + .map(|offset| signature_delta + offset) + .unwrap(); + assert!(thinking_delta < signature_delta); + assert!(signature_delta < thinking_stop); + assert!(out.contains("ccp:codex:v1:")); + } } diff --git a/src/providers/kimi/translate/request.rs b/src/providers/kimi/translate/request.rs index 21ded8e..7509b03 100644 --- a/src/providers/kimi/translate/request.rs +++ b/src/providers/kimi/translate/request.rs @@ -433,7 +433,7 @@ fn push_assistant_message(out: &mut Vec, blocks: &[ContentBlock]) { text_parts.push(text.clone()); } } - ContentBlock::Thinking { thinking } => { + ContentBlock::Thinking { thinking, .. } => { if !thinking.is_empty() { thinking_parts.push(thinking.clone()); } diff --git a/src/providers/translate_shared.rs b/src/providers/translate_shared.rs index baa4d9e..97201f5 100644 --- a/src/providers/translate_shared.rs +++ b/src/providers/translate_shared.rs @@ -22,6 +22,7 @@ pub enum ContentBlock { }, Thinking { thinking: String, + signature: Option, }, } @@ -202,7 +203,14 @@ fn parse_content_block(value: &Value, missing_tool_input: Value) -> Option None, }