diff --git a/Cargo.lock b/Cargo.lock index 4adb2d8dc..3bbb8ab96 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2513,6 +2513,7 @@ dependencies = [ "serde_json", "switchyard-protocol", "thiserror 2.0.18", + "tracing", ] [[package]] diff --git a/crates/protocol/src/stream.rs b/crates/protocol/src/stream.rs index 6cf4eb6d9..e5152b52f 100644 --- a/crates/protocol/src/stream.rs +++ b/crates/protocol/src/stream.rs @@ -182,6 +182,29 @@ impl LlmResponse { } } +/// An encrypted reasoning detail that names its provider item id but carries no payload yet. +fn is_reasoning_id_announcement(detail: &Value) -> bool { + detail.get("type").and_then(Value::as_str) == Some("reasoning.encrypted") + && detail.get("data").is_none() +} + +/// Appends a reasoning detail to an accumulated block. A Responses stream decoder announces a +/// reasoning item's provider id (`{"type": "reasoning.encrypted", "id"}`) before the payload +/// arrives; the payload detail then replaces that announcement so history holds one detail. +fn push_reasoning_detail(details: &mut Vec, detail: Value) { + if detail.get("type").and_then(Value::as_str) == Some("reasoning.encrypted") + && let Some(id) = detail.get("id").and_then(Value::as_str) + && let Some(announcement) = details.iter_mut().find(|existing| { + is_reasoning_id_announcement(existing) + && existing.get("id").and_then(Value::as_str) == Some(id) + }) + { + *announcement = detail; + return; + } + details.push(detail); +} + impl AggLlmResponse { /// Converts a fully-buffered response into a synthetic chunk stream. /// @@ -393,7 +416,9 @@ impl ResponseAccumulator { .push_str(&text); } LlmResponseChunk::ReasoningDetailsDelta { details, text, .. } => { - self.reasoning_details.extend(details); + for detail in details { + push_reasoning_detail(&mut self.reasoning_details, detail); + } if !text.is_empty() { self.reasoning .get_or_insert_with(String::new) @@ -433,7 +458,12 @@ impl ResponseAccumulator { content.push(ContentBlock::Reasoning { text: self.reasoning.unwrap_or_default(), signature: None, - details: self.reasoning_details, + // An announcement whose payload never arrived is a stream-level hint only. + details: self + .reasoning_details + .into_iter() + .filter(|detail| !is_reasoning_id_announcement(detail)) + .collect(), }); } if !self.text.is_empty() { diff --git a/crates/switchyard-translation/Cargo.toml b/crates/switchyard-translation/Cargo.toml index 1a13fffaa..3ba53a4f0 100644 --- a/crates/switchyard-translation/Cargo.toml +++ b/crates/switchyard-translation/Cargo.toml @@ -17,6 +17,7 @@ keywords = ["llm", "translation", "openai", "anthropic"] publish = ["crates-io"] [dependencies] +tracing.workspace = true base64.workspace = true serde.workspace = true serde_json.workspace = true diff --git a/crates/switchyard-translation/src/codecs/common.rs b/crates/switchyard-translation/src/codecs/common.rs index 78e5a7f89..1f3a31c8c 100644 --- a/crates/switchyard-translation/src/codecs/common.rs +++ b/crates/switchyard-translation/src/codecs/common.rs @@ -62,6 +62,68 @@ pub(crate) fn reasoning_text_from_details(details: &[Value]) -> Option { (!parts.is_empty()).then(|| parts.join("\n")) } +/// Collects reasoning text from a Responses reasoning item's `content` or `summary` +/// array, or from a bare string, into `out`. Empty strings are skipped. +pub(crate) fn collect_responses_reasoning_text(value: Option<&Value>, out: &mut Vec) { + match value { + Some(Value::String(text)) if !text.is_empty() => out.push(text.clone()), + Some(Value::Array(items)) => { + for item in items { + match item { + Value::String(text) if !text.is_empty() => out.push(text.clone()), + Value::Object(object) => { + if matches!( + object.get("type").and_then(Value::as_str), + Some("reasoning_text" | "summary_text" | "text") + ) && let Some(text) = object.get("text").and_then(Value::as_str) + && !text.is_empty() + { + out.push(text.to_string()); + } + } + _ => {} + } + } + } + _ => {} + } +} + +/// Returns the opaque payload of the first encrypted reasoning detail, if any. +/// +/// Two detail shapes are accepted: the documented `{"type": "reasoning.encrypted", "data"}` +/// object, and a verbatim Responses `reasoning` item carrying `encrypted_content` (the shape +/// the buffered request decoder stores when it keeps the provider item whole). +pub(crate) fn encrypted_reasoning_data(details: &[Value]) -> Option { + details + .iter() + .filter_map(Value::as_object) + .find_map(|detail| match detail.get("type").and_then(Value::as_str) { + Some("reasoning.encrypted") => detail.get("data").and_then(Value::as_str), + Some("reasoning") => detail.get("encrypted_content").and_then(Value::as_str), + _ => None, + }) + .filter(|data| !data.is_empty()) + .map(ToOwned::to_owned) +} + +/// Returns the provider item id recorded on the first `reasoning.encrypted` detail, if any. +/// Encrypted reasoning is bound to the item id it was issued under, so a replay must reuse it. +pub(crate) fn encrypted_reasoning_item_id(details: &[Value]) -> Option { + details + .iter() + .filter_map(Value::as_object) + .find(|detail| { + matches!( + detail.get("type").and_then(Value::as_str), + Some("reasoning.encrypted" | "reasoning") + ) + }) + .and_then(|detail| detail.get("id").and_then(Value::as_str)) + .filter(|id| !id.is_empty()) + .map(ToOwned::to_owned) +} + /// Returns the first non-empty string stored under the requested keys. pub(crate) fn first_nonempty_string<'a>( object: &'a Map, @@ -88,3 +150,34 @@ pub(crate) fn provider_extensions( } extensions } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + #[test] + fn encrypted_reasoning_helpers_accept_both_detail_shapes() { + let documented = vec![json!({"type": "reasoning.encrypted", "data": "blob", "id": "rs_1"})]; + assert_eq!( + encrypted_reasoning_data(&documented).as_deref(), + Some("blob") + ); + assert_eq!( + encrypted_reasoning_item_id(&documented).as_deref(), + Some("rs_1") + ); + let verbatim_item = vec![json!({ + "type": "reasoning", "id": "rs_2", "summary": [], "encrypted_content": "blob2" + })]; + assert_eq!( + encrypted_reasoning_data(&verbatim_item).as_deref(), + Some("blob2") + ); + assert_eq!( + encrypted_reasoning_item_id(&verbatim_item).as_deref(), + Some("rs_2") + ); + assert_eq!(encrypted_reasoning_data(&[json!({"type": "other"})]), None); + } +} diff --git a/crates/switchyard-translation/src/codecs/openai_chat/stream.rs b/crates/switchyard-translation/src/codecs/openai_chat/stream.rs index 6972850e9..e640d196b 100644 --- a/crates/switchyard-translation/src/codecs/openai_chat/stream.rs +++ b/crates/switchyard-translation/src/codecs/openai_chat/stream.rs @@ -214,6 +214,18 @@ fn encode_openai_chat_stream( )] } LlmResponseChunk::ReasoningDetailsDelta { details, text, .. } => { + // A Responses decoder announces a reasoning item's provider id ahead of its + // payload; that announcement carries nothing a chat client can use. + let details: Vec = details + .into_iter() + .filter(|detail| { + detail.get("type").and_then(Value::as_str) != Some("reasoning.encrypted") + || detail.get("data").is_some() + }) + .collect(); + if details.is_empty() && text.is_empty() { + return Vec::new(); + } let details_text = reasoning_text_from_details(&details); let mut delta = json!({"reasoning_details": details}); if !text.is_empty() && details_text.as_deref() != Some(text.as_str()) { diff --git a/crates/switchyard-translation/src/codecs/responses/buffered.rs b/crates/switchyard-translation/src/codecs/responses/buffered.rs index 03e11f15b..fddfcf380 100644 --- a/crates/switchyard-translation/src/codecs/responses/buffered.rs +++ b/crates/switchyard-translation/src/codecs/responses/buffered.rs @@ -8,6 +8,7 @@ use std::collections::HashSet; use serde_json::{Map, Value, json}; use crate::codecs::common::{ + collect_responses_reasoning_text, encrypted_reasoning_data, encrypted_reasoning_item_id, is_known_role_name, provider_extensions, reasoning_text_from_blocks, text_from_blocks, }; use crate::codecs::openai_chat::{decode_file_source, decode_image_source}; @@ -90,7 +91,29 @@ impl FormatCodec for OpenAiResponsesCodec { request.messages = messages; request.instructions.extend(instructions); let mut tool_namespaces = Map::new(); - request.tools = decode_responses_tools(body.get("tools"), &mut tool_namespaces); + let mut custom_tools = Map::new(); + request.tools = + decode_responses_tools(body.get("tools"), &mut tool_namespaces, &mut custom_tools); + // Responses-lite clients (Codex with a GPT-5 model) carry the tool definitions inside + // `input` as an `additional_tools` developer item instead of top-level `tools`. Those + // definitions are the request's tools; the item itself is kept verbatim so the request + // can be re-emitted in the shape the client used. + let mut additional_tools = Vec::new(); + if let Some(items) = body.get("input").and_then(Value::as_array) { + for item in items { + if let Some(item) = item.as_object() + && item.get("type").and_then(Value::as_str) == Some("additional_tools") + && let Some(tools) = item.get("tools").and_then(Value::as_array) + { + request.tools.extend(decode_responses_tools( + Some(&Value::Array(tools.clone())), + &mut tool_namespaces, + &mut custom_tools, + )); + additional_tools.extend(tools.iter().cloned()); + } + } + } request.tool_choice = body .get("tool_choice") .and_then(decode_responses_tool_choice); @@ -111,6 +134,11 @@ impl FormatCodec for OpenAiResponsesCodec { ], ); crate::codex_namespaces::attach_tool_namespaces(&mut request.extensions, tool_namespaces); + crate::codex_custom_tools::attach_custom_tools(&mut request.extensions, custom_tools); + crate::codex_custom_tools::attach_additional_tools( + &mut request.extensions, + additional_tools, + ); Ok(DecodedRequest { request, diagnostics, @@ -156,14 +184,29 @@ impl FormatCodec for OpenAiResponsesCodec { &mut diagnostics, _policy, crate::codex_namespaces::tool_namespaces(&request.extensions), + &crate::codex_custom_tools::custom_tool_names(&request.extensions), )?, ); - if !request.tools.is_empty() { + if let Some(additional) = crate::codex_custom_tools::additional_tools(&request.extensions) { + // A Responses-lite request carried its tools inside `input`; give them back the same + // way, verbatim, and leave top-level `tools` absent as the client did. + if let Some(Value::Array(input)) = body.get_mut("input") { + input.insert( + 0, + json!({ + "type": "additional_tools", + "role": "developer", + "tools": additional, + }), + ); + } + } else if !request.tools.is_empty() { body.insert( "tools".to_string(), encode_responses_tools( &request.tools, crate::codex_namespaces::tool_namespaces(&request.extensions), + crate::codex_custom_tools::custom_tools(&request.extensions), ), ); } @@ -474,7 +517,7 @@ fn decode_responses_input( arguments: item.get("arguments").cloned().unwrap_or_else(|| json!({})), }); } - Some("function_call_output") => { + Some("function_call_output") | Some("custom_tool_call_output") => { let tool_call_id = item .get("call_id") .and_then(Value::as_str) @@ -487,6 +530,45 @@ fn decode_responses_input( is_error: None, }); } + Some("custom_tool_call") => { + // A freeform call carries a raw `input` string; it rides through the IR + // as the single `input` argument of a function-style call. + if !pending_tool_outputs.is_empty() { + flush_responses_tool_block( + &mut messages, + &mut pending_tool_calls, + &mut pending_tool_outputs, + &mut deferred_messages, + &mut pending_reasoning, + ); + } + let id = item + .get("call_id") + .and_then(Value::as_str) + .filter(|id| !id.is_empty()) + .map(ToOwned::to_owned) + .unwrap_or_else(|| match &policy.deterministic_ids { + DeterministicIdPolicy::GenerateStable { prefix } => { + stable_id(prefix, index + 1) + } + DeterministicIdPolicy::Preserve => String::new(), + }); + let name = item + .get("name") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + let input = item + .get("input") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + pending_tool_calls.push(ToolCall { + id, + name, + arguments: json!({crate::codex_custom_tools::INPUT_ARGUMENT: input}), + }); + } None => { return Err(TranslationError::InvalidValue { path: format!("$.input[{index}].type"), @@ -494,6 +576,9 @@ fn decode_responses_input( .to_string(), }); } + // Tool definitions, not conversation; decoded separately by the request + // decoder and re-emitted in place by the request encoder. + Some("additional_tools") => {} _ => { let message = Message { role: Role::User, @@ -653,38 +738,32 @@ fn decode_responses_reasoning_item(item: &Map) -> Vec, out: &mut Vec) { - match value { - Some(Value::String(text)) if !text.is_empty() => out.push(text.clone()), - Some(Value::Array(items)) => { - for item in items { - match item { - Value::String(text) if !text.is_empty() => out.push(text.clone()), - Value::Object(object) => { - if matches!( - object.get("type").and_then(Value::as_str), - Some("reasoning_text" | "summary_text" | "text") - ) && let Some(text) = object.get("text").and_then(Value::as_str) - && !text.is_empty() - { - out.push(text.to_string()); - } - } - _ => {} - } - } - } - _ => {} - } -} // Decodes Responses content arrays or strings into normalized content blocks. fn decode_responses_content(value: &Value) -> Vec { @@ -794,6 +873,7 @@ fn request_role_from_responses(role: Option<&str>, path: &str) -> Result { fn decode_responses_tools( value: Option<&Value>, namespaces: &mut Map, + custom_tools: &mut Map, ) -> Vec { let Some(tools) = value.and_then(Value::as_array) else { return Vec::new(); @@ -810,7 +890,7 @@ fn decode_responses_tools( .get("name") .and_then(Value::as_str) .filter(|name| !name.is_empty()); - for mut child in decode_responses_tools(tool.get("tools"), namespaces) { + for mut child in decode_responses_tools(tool.get("tools"), namespaces, custom_tools) { // A nested container already qualified its own children, and the // innermost name is the one that identifies the tool. let already_qualified = namespaces.contains_key(&child.name); @@ -826,6 +906,26 @@ fn decode_responses_tools( } out.push(child); } + } else if tool.get("type").and_then(Value::as_str) == Some("custom") { + // A freeform tool takes a raw string, not JSON arguments. The IR sees it as a + // function with a single `input` argument; the verbatim definition is kept so a + // Responses upstream still receives the freeform tool. + if let Some(name) = tool + .get("name") + .and_then(Value::as_str) + .filter(|name| !name.is_empty()) + { + custom_tools.insert(name.to_string(), Value::Object(tool.clone())); + out.push(ToolDefinition { + name: name.to_string(), + description: tool + .get("description") + .and_then(Value::as_str) + .map(ToOwned::to_owned), + parameters: crate::codex_custom_tools::input_schema(), + strict: None, + }); + } } else if tool.get("type").and_then(Value::as_str) == Some("function") { if let Some(function) = tool.get("function").and_then(Value::as_object) { if let Some(name) = function.get("name").and_then(Value::as_str) @@ -1016,7 +1116,9 @@ fn encode_responses_input( diagnostics: &mut Vec, policy: &TranslationPolicy, namespaces: Option<&Map>, + custom_tools: &std::collections::HashSet, ) -> Result { + let mut custom_call_ids: std::collections::HashSet = std::collections::HashSet::new(); if messages.len() == 1 && matches!(messages[0].role, Role::User) && messages[0].content.len() == 1 @@ -1051,17 +1153,25 @@ fn encode_responses_input( ContentBlock::ToolCall(_) | ContentBlock::ToolResult(_) ) }) { - encoded.extend( - content - .iter() - .filter_map(|block| encode_responses_special_input(block, namespaces)), - ); + encoded.extend(content.iter().filter_map(|block| { + encode_responses_special_input( + block, + namespaces, + custom_tools, + &mut custom_call_ids, + ) + })); continue; } let mut visible_content = Vec::new(); let mut emitted_special = false; for block in &content { - if let Some(item) = encode_responses_special_input(block, namespaces) { + if let Some(item) = encode_responses_special_input( + block, + namespaces, + custom_tools, + &mut custom_call_ids, + ) { encoded.push(item); emitted_special = true; } else { @@ -1084,6 +1194,8 @@ fn encode_responses_input( fn encode_responses_special_input( block: &ContentBlock, namespaces: Option<&Map>, + custom_tools: &std::collections::HashSet, + custom_call_ids: &mut std::collections::HashSet, ) -> Option { match block { ContentBlock::Reasoning { @@ -1095,6 +1207,16 @@ fn encode_responses_special_input( "content": [{"type": "reasoning_text", "text": text}], "summary": [], })), + ContentBlock::ToolCall(call) if custom_tools.contains(&call.name) => { + // A freeform tool call replays as `custom_tool_call` with its raw input. + custom_call_ids.insert(call.id.clone()); + Some(json!({ + "type": "custom_tool_call", + "call_id": call.id, + "name": call.name, + "input": crate::codex_custom_tools::input_from_arguments(&call.arguments), + })) + } ContentBlock::ToolCall(call) => { // A Responses client dispatches on name plus namespace, so undo the // qualification this request applied for a flat upstream. @@ -1117,7 +1239,11 @@ fn encode_responses_special_input( Some(item) } ContentBlock::ToolResult(result) => Some(json!({ - "type": "function_call_output", + "type": if custom_call_ids.contains(&result.tool_call_id) { + "custom_tool_call_output" + } else { + "function_call_output" + }, "call_id": result.tool_call_id, "output": text_from_blocks(&result.content, " "), })), @@ -1213,10 +1339,16 @@ fn encode_responses_content( fn encode_responses_tools( tools: &[ToolDefinition], namespaces: Option<&Map>, + custom_tools: Option<&Map>, ) -> Value { let mut out: Vec = Vec::new(); let mut containers: Vec<(String, Vec)> = Vec::new(); for tool in tools { + // A freeform tool goes back out exactly as the client defined it. + if let Some(custom) = custom_tools.and_then(|custom| custom.get(&tool.name)) { + out.push(custom.clone()); + continue; + } let mut item = json!({ "type": "function", "name": tool.name, @@ -1314,6 +1446,26 @@ fn decode_responses_output_item( })], stop_reason: Some(StopReason::ToolUse), })), + Some("custom_tool_call") => Ok(Some(ResponseOutput { + role: Role::Assistant, + content: vec![ContentBlock::ToolCall(ToolCall { + id: item + .get("call_id") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + name: item + .get("name") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + arguments: json!({ + crate::codex_custom_tools::INPUT_ARGUMENT: + item.get("input").and_then(Value::as_str).unwrap_or_default() + }), + })], + stop_reason: Some(StopReason::ToolUse), + })), Some("reasoning") => Ok(Some(ResponseOutput { role: Role::Assistant, content: decode_responses_reasoning_item(item), @@ -1342,8 +1494,20 @@ fn encode_responses_output(outputs: &[ResponseOutput]) -> Value { }; let mut items = Vec::new(); - if !reasoning.is_empty() { - items.push(encode_responses_reasoning_output(&reasoning)); + let encrypted_reasoning = output.content.iter().find_map(|block| match block { + ContentBlock::Reasoning { details, .. } => encrypted_reasoning_data(details), + _ => None, + }); + let encrypted_reasoning_id = output.content.iter().find_map(|block| match block { + ContentBlock::Reasoning { details, .. } => encrypted_reasoning_item_id(details), + _ => None, + }); + if !reasoning.is_empty() || encrypted_reasoning.is_some() { + items.push(encode_responses_reasoning_output( + &reasoning, + encrypted_reasoning.as_deref(), + encrypted_reasoning_id.as_deref(), + )); } if !text.is_empty() || (!has_tool_calls && reasoning.is_empty()) { @@ -1376,18 +1540,29 @@ fn encode_responses_output(outputs: &[ResponseOutput]) -> Value { ) } -// Encodes private reasoning as a separate Responses output item. -fn encode_responses_reasoning_output(text: &str) -> Value { - json!({ +// Encodes private reasoning as a separate Responses output item. An encrypted-only item +// carries no text part but keeps `encrypted_content` so the client can replay it. +fn encode_responses_reasoning_output( + text: &str, + encrypted: Option<&str>, + item_id: Option<&str>, +) -> Value { + // Standard Responses shape: text as `summary_text` parts, which is what clients record. + let mut summary = Vec::new(); + if !text.is_empty() { + summary.push(json!({"type": "summary_text", "text": text})); + } + // Encrypted reasoning verifies only under the id it was issued with, so reuse it. + let mut item = json!({ "type": "reasoning", - "id": "rs_switchyard", + "id": item_id.unwrap_or("rs_switchyard"), "status": "completed", - "content": [{ - "type": "reasoning_text", - "text": text, - }], - "summary": [], - }) + "summary": summary, + }); + if let Some(encrypted) = encrypted { + item["encrypted_content"] = Value::String(encrypted.to_string()); + } + item } // Serializes JSON with Python-like spacing to match legacy converter behavior. diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index 4731599bc..3b2920789 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -7,6 +7,9 @@ use serde::Serialize; use serde_json::{Value, json}; use crate::LlmResponseChunk; +use crate::codecs::common::{ + collect_responses_reasoning_text, encrypted_reasoning_data, encrypted_reasoning_item_id, +}; use crate::codecs::stream::{ StreamCodec, StreamTranslationState, record_source_identity, target_message_id_or_source_message_id, target_model_or_source_model, @@ -99,6 +102,9 @@ fn decode_responses_stream( state: &mut StreamTranslationState, event: &Value, ) -> Vec { + // Opt-in raw capture of what the upstream actually sent, for diagnosing provider-specific + // event shapes: `RUST_LOG=switchyard_translation::responses::raw=trace`. + tracing::trace!(target: "switchyard_translation::responses::raw", raw = %event); let event_type = event .get("type") .or_else(|| event.get("event")) @@ -143,11 +149,19 @@ fn decode_responses_stream( .or_else(|| event.get("text")) .and_then(Value::as_str) .map(|text| { + let index = event + .get("output_index") + .and_then(Value::as_u64) + .unwrap_or(0) as usize; + // Recorded so `response.output_item.done`, which repeats the full + // text, can tell it is a repeat. + state + .decoded_reasoning + .entry(index) + .or_default() + .push_str(text); vec![LlmResponseChunk::ReasoningDelta { - index: event - .get("output_index") - .and_then(Value::as_u64) - .unwrap_or(0) as usize, + index, text: text.to_string(), }] }) @@ -181,8 +195,72 @@ fn decode_responses_stream( .unwrap_or_default() } Some("response.output_item.done") => decode_responses_output_item_done(event, state), + Some("response.reasoning_text.done") | Some("response.reasoning_summary_text.done") => { + let index = event + .get("output_index") + .and_then(Value::as_u64) + .unwrap_or(0) as usize; + event + .get("text") + .and_then(Value::as_str) + .filter(|text| !text.is_empty()) + .filter(|_| { + state + .decoded_reasoning + .get(&index) + .is_none_or(String::is_empty) + }) + .map(|text| { + state.decoded_reasoning.insert(index, text.to_string()); + vec![LlmResponseChunk::ReasoningDelta { + index, + text: text.to_string(), + }] + }) + .unwrap_or_default() + } + Some("response.reasoning_summary_part.done") => { + let index = event + .get("output_index") + .and_then(Value::as_u64) + .unwrap_or(0) as usize; + event + .get("part") + .and_then(|part| part.get("text")) + .and_then(Value::as_str) + .filter(|text| !text.is_empty()) + .filter(|_| { + state + .decoded_reasoning + .get(&index) + .is_none_or(String::is_empty) + }) + .map(|text| { + state.decoded_reasoning.insert(index, text.to_string()); + vec![LlmResponseChunk::ReasoningDelta { + index, + text: text.to_string(), + }] + }) + .unwrap_or_default() + } Some("response.completed") => { let mut out = Vec::new(); + // Some providers surface reasoning only in the final output array. Position is the + // output index; anything already decoded is skipped by the helper. + if let Some(items) = event + .get("response") + .and_then(|response| response.get("output")) + .and_then(Value::as_array) + { + for (position, item) in items.iter().enumerate() { + if let Some(item) = item.as_object() + && item.get("type").and_then(Value::as_str) == Some("reasoning") + { + out.extend(decode_responses_reasoning_item(item, position, state)); + } + } + } if let Some(usage) = event .get("response") .and_then(Value::as_object) @@ -236,13 +314,42 @@ fn encode_responses_stream( ensure_responses_created(state) } LlmResponseChunk::TextDelta { text, .. } => encode_responses_text_delta(state, text), - LlmResponseChunk::ReasoningDelta { text, .. } => { - encode_responses_reasoning_delta(state, text) + LlmResponseChunk::ReasoningDelta { index, text } => { + encode_responses_reasoning_delta(state, index, text) } - LlmResponseChunk::ReasoningDetailsDelta { text, .. } if !text.is_empty() => { - encode_responses_reasoning_delta(state, text) + LlmResponseChunk::ReasoningDetailsDelta { + index, + details, + text, + } => { + // Encrypted reasoning has no streamable text, but the item must still be emitted + // so the client can replay it on the next turn. + let data = encrypted_reasoning_data(&details); + let item = state.response_reasoning.entry(index).or_default(); + match encrypted_reasoning_item_id(&details) { + Some(id) if item.started && item.item_id.as_deref() != Some(id.as_str()) => { + // The item already opened under another id; the payload would fail + // verification under it, so drop the payload rather than poison the replay. + tracing::warn!( + item_id = %id, + "encrypted reasoning arrived after its item opened under a different id; dropping payload" + ); + } + id => { + if id.is_some() { + item.item_id = id; + } + if data.is_some() { + item.encrypted = data; + } + } + } + let mut out = ensure_responses_reasoning_started(state, index); + if !text.is_empty() { + out.extend(encode_responses_reasoning_delta(state, index, text)); + } + out } - LlmResponseChunk::ReasoningDetailsDelta { .. } => Vec::new(), LlmResponseChunk::ToolCallDelta { index, id, @@ -297,7 +404,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { "output_index": output_index, "item": { "type": "message", - "id": format!("msg_{output_index}"), + "id": responses_item_id(state, "msg", output_index), "role": "assistant", "status": status, "content": [{"type": "output_text", "text": state.response_text}], @@ -306,25 +413,48 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { } let mut final_items: Vec<(usize, Value)> = Vec::new(); - if state.response_reasoning_started - && let Some(output_index) = state.response_reasoning_output_index - { - out.push(json!({ - "type": "response.reasoning_text.done", - "output_index": output_index, - "content_index": 0, - "text": state.response_reasoning_text, - })); - let item = json!({ + let mut reasoning_items: Vec<(usize, usize)> = state + .response_reasoning + .iter() + .filter(|(_, item)| item.started) + .filter_map(|(index, item)| item.output_index.map(|output| (*index, output))) + .collect(); + reasoning_items.sort_by_key(|(_, output_index)| *output_index); + for (index, output_index) in reasoning_items { + // Encrypted-only reasoning streamed no text, so it gets no summary part; the item + // itself still closes so the client can replay its `encrypted_content`. + let item_id = responses_reasoning_item_id(state, index); + let reasoning = &state.response_reasoning[&index]; + let mut summary = Vec::new(); + if !reasoning.text.is_empty() { + out.push(json!({ + "type": "response.reasoning_summary_text.done", + "item_id": item_id, + "output_index": output_index, + "summary_index": 0, + "text": reasoning.text, + })); + out.push(json!({ + "type": "response.reasoning_summary_part.done", + "item_id": item_id, + "output_index": output_index, + "summary_index": 0, + "part": {"type": "summary_text", "text": reasoning.text}, + })); + summary.push(json!({ + "type": "summary_text", + "text": reasoning.text, + })); + } + let mut item = json!({ "type": "reasoning", - "id": format!("rs_{output_index}"), + "id": item_id, "status": "completed", - "content": [{ - "type": "reasoning_text", - "text": state.response_reasoning_text, - }], - "summary": [], + "summary": summary, }); + if let Some(encrypted) = &reasoning.encrypted { + item["encrypted_content"] = Value::String(encrypted.clone()); + } out.push(json!({ "type": "response.output_item.done", "output_index": output_index, @@ -339,7 +469,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { output_index, json!({ "type": "message", - "id": format!("msg_{output_index}"), + "id": responses_item_id(state, "msg", output_index), "role": "assistant", "status": status, "content": [{"type": "output_text", "text": state.response_text}], @@ -359,7 +489,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { })); let item = json!({ "type": "function_call", - "id": tool.response_item_id.clone().unwrap_or_else(|| format!("fc_{output_index}")), + "id": tool.response_item_id.clone().unwrap_or_else(|| responses_item_id(state, "fc", output_index)), "call_id": tool.id.clone().unwrap_or_else(|| format!("call_{output_index}")), "name": tool.name.clone().unwrap_or_default(), "arguments": tool.arguments, @@ -388,6 +518,78 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { } // Converts Responses function-call item creation into a neutral tool-call delta. +// Decodes the reasoning a provider put on a reasoning output item itself: plaintext in +// `content`, `summary`, or top-level `text`, and/or an opaque `encrypted_content`. Text that was +// already decoded for this output index (from delta events or an earlier item event) is not +// repeated, so `added`, `done`, and the final `response.completed` output can all be inspected +// safely. +fn decode_responses_reasoning_item( + item: &serde_json::Map, + index: usize, + state: &mut StreamTranslationState, +) -> Vec { + let mut out = Vec::new(); + let has_encrypted = item + .get("encrypted_content") + .and_then(Value::as_str) + .is_some_and(|data| !data.is_empty()); + // The provider's item id is known from the first `added` event, long before the encrypted + // payload arrives on `done`. Announce it so the encoder opens the item under that id; + // otherwise summary text opens it under a synthesized id the payload cannot verify against. + if let Some(id) = item + .get("id") + .and_then(Value::as_str) + .filter(|id| !id.is_empty()) + && !state.decoded_reasoning_ids.contains(&index) + { + state.decoded_reasoning_ids.insert(index); + if !has_encrypted { + out.push(LlmResponseChunk::ReasoningDetailsDelta { + index, + details: vec![json!({"type": "reasoning.encrypted", "id": id})], + text: String::new(), + }); + } + } + let mut parts = Vec::new(); + collect_responses_reasoning_text(item.get("content"), &mut parts); + collect_responses_reasoning_text(item.get("summary"), &mut parts); + collect_responses_reasoning_text(item.get("text"), &mut parts); + let text = parts.join("\n"); + let already = state + .decoded_reasoning + .get(&index) + .map(String::as_str) + .unwrap_or(""); + if !text.is_empty() && already.is_empty() { + state.decoded_reasoning.insert(index, text.clone()); + out.push(LlmResponseChunk::ReasoningDelta { index, text }); + } + if let Some(data) = item + .get("encrypted_content") + .and_then(Value::as_str) + .filter(|data| !data.is_empty()) + && !state.decoded_reasoning_encrypted.contains(&index) + { + state.decoded_reasoning_encrypted.insert(index); + // The payload only verifies under the id it was issued with, so carry that id along. + let mut detail = json!({"type": "reasoning.encrypted", "data": data}); + if let Some(id) = item + .get("id") + .and_then(Value::as_str) + .filter(|id| !id.is_empty()) + { + detail["id"] = Value::String(id.to_string()); + } + out.push(LlmResponseChunk::ReasoningDetailsDelta { + index, + details: vec![detail], + text: String::new(), + }); + } + out +} + fn decode_responses_output_item_added( event: &Value, state: &mut StreamTranslationState, @@ -395,18 +597,27 @@ fn decode_responses_output_item_added( let Some(item) = event.get("item").and_then(Value::as_object) else { return Vec::new(); }; - if item.get("type").and_then(Value::as_str) != Some("function_call") { - return Vec::new(); - } let index = event .get("output_index") .and_then(Value::as_u64) .unwrap_or(0) as usize; - let arguments_delta = item - .get("arguments") - .and_then(Value::as_str) - .filter(|arguments| !arguments.is_empty()) - .map(ToOwned::to_owned); + if item.get("type").and_then(Value::as_str) == Some("reasoning") { + return decode_responses_reasoning_item(item, index, state); + } + let item_type = item.get("type").and_then(Value::as_str); + if item_type != Some("function_call") && item_type != Some("custom_tool_call") { + return Vec::new(); + } + // A freeform call's `input` becomes the single `input` argument; it is only complete on + // the done event, so nothing is emitted for it here beyond id and name. + let arguments_delta = if item_type == Some("custom_tool_call") { + None + } else { + item.get("arguments") + .and_then(Value::as_str) + .filter(|arguments| !arguments.is_empty()) + .map(ToOwned::to_owned) + }; if let Some(arguments) = arguments_delta.as_deref() { state .tool_states @@ -438,14 +649,27 @@ fn decode_responses_output_item_done( let Some(item) = event.get("item").and_then(Value::as_object) else { return Vec::new(); }; - if item.get("type").and_then(Value::as_str) != Some("function_call") { - return Vec::new(); - } let index = event .get("output_index") .and_then(Value::as_u64) .unwrap_or(0) as usize; - let arguments = item.get("arguments").and_then(Value::as_str); + if item.get("type").and_then(Value::as_str) == Some("reasoning") { + return decode_responses_reasoning_item(item, index, state); + } + let item_type = item.get("type").and_then(Value::as_str); + if item_type != Some("function_call") && item_type != Some("custom_tool_call") { + return Vec::new(); + } + let custom_arguments = (item_type == Some("custom_tool_call")).then(|| { + json!({ + crate::codex_custom_tools::INPUT_ARGUMENT: + item.get("input").and_then(Value::as_str).unwrap_or_default() + }) + .to_string() + }); + let arguments = custom_arguments + .as_deref() + .or_else(|| item.get("arguments").and_then(Value::as_str)); if let Some(arguments) = arguments { // Compared against what THIS decoder has seen. Reading the encoder's // `arguments` instead only deduplicates when a single state performs @@ -534,7 +758,7 @@ fn encode_responses_text_delta(state: &mut StreamTranslationState, text: String) "output_index": output_index, "item": { "type": "message", - "id": format!("msg_{output_index}"), + "id": responses_item_id(state, "msg", output_index), "role": "assistant", "status": "in_progress", "content": [], @@ -557,40 +781,73 @@ fn encode_responses_text_delta(state: &mut StreamTranslationState, text: String) out } +// The id of the reasoning item encoded for a source index: the provider's own id when +// encrypted reasoning binds to it, else synthesized from the emitted output index. +fn responses_reasoning_item_id(state: &StreamTranslationState, index: usize) -> String { + let item = state.response_reasoning.get(&index); + item.and_then(|item| item.item_id.clone()) + .unwrap_or_else(|| { + let output_index = item.and_then(|item| item.output_index).unwrap_or(0); + responses_item_id(state, "rs", output_index) + }) +} + +// Opens the Responses reasoning output item for a source index once, emitting its `added` events. +fn ensure_responses_reasoning_started( + state: &mut StreamTranslationState, + index: usize, +) -> Vec { + let mut out = ensure_responses_created(state); + if state + .response_reasoning + .get(&index) + .is_some_and(|item| item.started) + { + return out; + } + let output_index = state.next_response_output_index; + state.next_response_output_index += 1; + let item = state.response_reasoning.entry(index).or_default(); + item.started = true; + item.output_index = Some(output_index); + let item_id = responses_reasoning_item_id(state, index); + // Standard Responses shape: reasoning text lives in `summary` as `summary_text` + // parts. Clients such as Codex record reasoning items only in this shape. + out.push(json!({ + "type": "response.output_item.added", + "output_index": output_index, + "item": { + "type": "reasoning", + "id": item_id, + "status": "in_progress", + "summary": [], + }, + })); + out.push(json!({ + "type": "response.reasoning_summary_part.added", + "item_id": item_id, + "output_index": output_index, + "summary_index": 0, + "part": {"type": "summary_text", "text": ""}, + })); + out +} + // Accumulates reasoning text and emits Responses reasoning events. fn encode_responses_reasoning_delta( state: &mut StreamTranslationState, + index: usize, text: String, ) -> Vec { - let mut out = ensure_responses_created(state); - if !state.response_reasoning_started { - state.response_reasoning_started = true; - let output_index = state.next_response_output_index; - state.next_response_output_index += 1; - state.response_reasoning_output_index = Some(output_index); - out.push(json!({ - "type": "response.output_item.added", - "output_index": output_index, - "item": { - "type": "reasoning", - "id": format!("rs_{output_index}"), - "status": "in_progress", - "content": [], - "summary": [], - }, - })); - out.push(json!({ - "type": "response.reasoning_text.added", - "output_index": output_index, - "content_index": 0, - "text": "", - })); - } - state.response_reasoning_text.push_str(&text); + let mut out = ensure_responses_reasoning_started(state, index); + let item = state.response_reasoning.entry(index).or_default(); + item.text.push_str(&text); + let output_index = item.output_index.unwrap_or(0); out.push(json!({ - "type": "response.reasoning_text.delta", - "output_index": state.response_reasoning_output_index.unwrap_or(0), - "content_index": 0, + "type": "response.reasoning_summary_text.delta", + "item_id": responses_reasoning_item_id(state, index), + "output_index": output_index, + "summary_index": 0, "delta": text, })); out @@ -605,6 +862,7 @@ fn encode_responses_tool_delta( arguments_delta: Option, ) -> Vec { let mut out = ensure_responses_created(state); + let resp_id = responses_id(state); let tool = state.tool_states.entry(index).or_default(); if id.is_some() { tool.id = id; @@ -624,14 +882,14 @@ fn encode_responses_tool_delta( let output_index = state.next_response_output_index; state.next_response_output_index += 1; tool.response_output_index = Some(output_index); - tool.response_item_id = Some(format!("fc_{output_index}")); + tool.response_item_id = Some(responses_item_id_from(&resp_id, "fc", output_index)); tool.started = true; out.push(json!({ "type": "response.output_item.added", "output_index": output_index, "item": { "type": "function_call", - "id": tool.response_item_id.clone().unwrap_or_else(|| format!("fc_{output_index}")), + "id": tool.response_item_id.clone().unwrap_or_else(|| responses_item_id_from(&resp_id, "fc", output_index)), "call_id": tool.id.clone().unwrap_or_else(|| format!("call_{index}")), "name": name, "arguments": "", @@ -710,6 +968,37 @@ fn responses_usage_value(usage: &Usage) -> Value { }) } +// Longest response-id discriminator embedded verbatim in a synthesized item id. OpenAI rejects +// item ids over 64 characters, and some upstreams issue response ids several hundred characters +// long, so anything longer is replaced by a fixed-width digest. +const ITEM_ID_DISCRIMINATOR_CHARS: usize = 40; + +// Builds a synthesized output-item id that is unique across responses. Clients replay the +// whole conversation, so ids must not repeat from one turn to the next; the response id is +// unique per upstream call and is used as the discriminator. +fn responses_item_id(state: &StreamTranslationState, prefix: &str, output_index: usize) -> String { + responses_item_id_from(&responses_id(state), prefix, output_index) +} + +// Same as [`responses_item_id`], for callers that already hold the response id and cannot +// borrow `state` again (for example while a `tool_states` entry is borrowed mutably). +fn responses_item_id_from(resp_id: &str, prefix: &str, output_index: usize) -> String { + let resp = resp_id.strip_prefix("resp_").unwrap_or(resp_id); + if resp.chars().count() <= ITEM_ID_DISCRIMINATOR_CHARS { + format!("{prefix}_{resp}_{output_index}") + } else { + format!("{prefix}_{:016x}_{output_index}", fnv1a_64(resp)) + } +} + +// FNV-1a over the UTF-8 bytes: a stable, dependency-free 64-bit digest. Uniqueness across the +// handful of responses in one conversation is all that is required of it. +fn fnv1a_64(text: &str) -> u64 { + text.bytes().fold(0xcbf2_9ce4_8422_2325_u64, |hash, byte| { + (hash ^ u64::from(byte)).wrapping_mul(0x0000_0100_0000_01b3) + }) +} + // Converts any upstream message ID into a Responses-looking response ID. fn responses_id(state: &StreamTranslationState) -> String { let Some(id) = target_message_id_or_source_message_id(state) else { diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index 0ec9b7f88..08ee74fad 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -52,14 +52,22 @@ pub struct StreamTranslationState { pub(crate) text_block_started: bool, pub(crate) emitted_content_block: bool, pub(crate) tool_states: BTreeMap, + /// Reasoning text observed while DECODING, per output index, so a completed item + /// that repeats already-streamed text is not decoded twice. + pub(crate) decoded_reasoning: BTreeMap, + /// Output indexes whose encrypted reasoning payload was already decoded. + pub(crate) decoded_reasoning_encrypted: std::collections::BTreeSet, pub(crate) response_created: bool, pub(crate) response_text_started: bool, pub(crate) response_text_output_index: Option, pub(crate) response_text: String, - pub(crate) response_reasoning_started: bool, - pub(crate) response_reasoning_output_index: Option, - pub(crate) response_reasoning_text: String, + /// Reasoning items being ENCODED, keyed by source content index. A response can carry + /// several reasoning items (GPT-5 emits one ahead of each tool call); each becomes its + /// own output item so none of their encrypted payloads is lost. + pub(crate) response_reasoning: BTreeMap, + /// Source indexes whose provider reasoning item id was already announced while DECODING. + pub(crate) decoded_reasoning_ids: std::collections::BTreeSet, pub(crate) next_response_output_index: usize, pub(crate) response_sequence_number: u64, @@ -67,6 +75,21 @@ pub struct StreamTranslationState { pub(crate) reasoning_block_started: bool, } +// One Responses reasoning output item under construction by the encoder. +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +pub(crate) struct ResponseReasoningState { + /// Set once the item's `added` events were emitted. + pub(crate) started: bool, + pub(crate) output_index: Option, + /// Provider item id the encrypted reasoning was issued under. Used as the emitted item id + /// so the client's replay verifies upstream; `None` falls back to a synthesized id. + pub(crate) item_id: Option, + pub(crate) text: String, + /// Opaque `encrypted_content` carried by a Responses reasoning item. Kept verbatim so + /// the emitted item stays replayable by the client even when no plaintext streamed. + pub(crate) encrypted: Option, +} + // Tracks an in-progress streamed tool call across provider-specific deltas. #[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] pub(crate) struct StreamToolState { diff --git a/crates/switchyard-translation/src/codex_custom_tools.rs b/crates/switchyard-translation/src/codex_custom_tools.rs new file mode 100644 index 000000000..7d15d56bc --- /dev/null +++ b/crates/switchyard-translation/src/codex_custom_tools.rs @@ -0,0 +1,264 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Round-trips OpenAI Responses freeform ("custom") tools through the neutral IR. +//! +//! Codex drives GPT-5 models with freeform tools: the definition is `{"type": "custom", "name", +//! "description", "format"}` and the model answers with `custom_tool_call` items whose `input` +//! is a raw string rather than JSON arguments. The IR only knows function-style tools, so a +//! custom tool is represented as a function whose single argument is `input`, and the verbatim +//! definitions are kept on the request extensions. When the response is encoded back to +//! Responses, calls to those tools are rewritten into `custom_tool_call` items again. + +use std::collections::HashSet; + +use serde_json::{Map, Value, json}; +use switchyard_protocol::ProviderExtensions; + +/// Request-extension key holding the verbatim custom tool definitions, keyed by tool name. +/// +/// Prefixed so it cannot collide with a real provider field, and so a codec that allowlists +/// provider fields never forwards it. +pub const CUSTOM_TOOLS_KEY: &str = "switchyard_codex_custom_tools"; + +/// Request-extension key holding the verbatim `tools` array of a Responses-lite +/// `additional_tools` input item, so the request can be re-emitted in the same shape. +/// +/// Codex sends GPT-5 requests in a "lite" shape: no top-level `tools`, empty `instructions`, +/// and the tool definitions inside `input[0]` as `{"type": "additional_tools", "role": +/// "developer", "tools": [...]}`. +pub const ADDITIONAL_TOOLS_KEY: &str = "switchyard_codex_additional_tools"; + +/// Stores the verbatim tools array of an `additional_tools` input item. +pub fn attach_additional_tools(extensions: &mut ProviderExtensions, tools: Vec) { + if !tools.is_empty() { + extensions + .fields + .insert(ADDITIONAL_TOOLS_KEY.to_string(), Value::Array(tools)); + } +} + +/// Reads the verbatim `additional_tools` array back off a request's extensions. +pub fn additional_tools(extensions: &ProviderExtensions) -> Option<&Vec> { + extensions + .fields + .get(ADDITIONAL_TOOLS_KEY) + .and_then(Value::as_array) +} + +/// Argument name used to carry a custom tool's freeform input through the IR. +pub const INPUT_ARGUMENT: &str = "input"; + +/// The IR parameter schema for a custom tool: one required string, `input`. +pub fn input_schema() -> Value { + json!({ + "type": "object", + "properties": {INPUT_ARGUMENT: {"type": "string"}}, + "required": [INPUT_ARGUMENT], + "additionalProperties": false, + }) +} + +/// Stores the collected definitions on a request's extensions, when there are any. +pub fn attach_custom_tools(extensions: &mut ProviderExtensions, tools: Map) { + if !tools.is_empty() { + extensions + .fields + .insert(CUSTOM_TOOLS_KEY.to_string(), Value::Object(tools)); + } +} + +/// Reads the definitions back off a request's extensions. +pub fn custom_tools(extensions: &ProviderExtensions) -> Option<&Map> { + extensions + .fields + .get(CUSTOM_TOOLS_KEY) + .and_then(Value::as_object) +} + +/// Names of the custom tools recorded on a request. +pub fn custom_tool_names(extensions: &ProviderExtensions) -> HashSet { + custom_tools(extensions) + .map(|tools| tools.keys().cloned().collect()) + .unwrap_or_default() +} + +/// Extracts the freeform input from IR tool arguments, falling back to the serialized +/// arguments when the model did not use the `input` convention. +pub fn input_from_arguments(arguments: &Value) -> String { + match arguments { + Value::Object(object) => match object.get(INPUT_ARGUMENT) { + Some(Value::String(input)) => input.clone(), + Some(other) => other.to_string(), + None => arguments.to_string(), + }, + Value::String(text) => match serde_json::from_str::(text) { + Ok(Value::Object(object)) => match object.get(INPUT_ARGUMENT) { + Some(Value::String(input)) => input.clone(), + _ => text.clone(), + }, + _ => text.clone(), + }, + other => other.to_string(), + } +} + +/// Rewrites a `function_call` output item into a `custom_tool_call` when the tool is custom. +/// Returns whether the item was rewritten. +fn rewrite_item(item: &mut Value, custom: &HashSet) -> bool { + let Some(object) = item.as_object_mut() else { + return false; + }; + if object.get("type").and_then(Value::as_str) != Some("function_call") { + return false; + } + let Some(name) = object.get("name").and_then(Value::as_str) else { + return false; + }; + if !custom.contains(name) { + return false; + } + let input = object + .remove("arguments") + .map(|arguments| input_from_arguments(&arguments)) + .unwrap_or_default(); + object.insert( + "type".to_string(), + Value::String("custom_tool_call".to_string()), + ); + object.insert("input".to_string(), Value::String(input)); + // OpenAI validates replayed item ids by prefix: a custom tool call must be `ctc_...`. + if let Some(Value::String(id)) = object.get_mut("id") + && let Some(rest) = id.strip_prefix("fc_") + { + *id = format!("ctc_{rest}"); + } + true +} + +/// Rewrites custom tool calls inside a buffered Responses body's `output` array. +pub fn restore_custom_tool_calls(body: &mut Value, custom: &HashSet) { + if custom.is_empty() { + return; + } + if let Some(items) = body.get_mut("output").and_then(Value::as_array_mut) { + for item in items { + rewrite_item(item, custom); + } + } +} + +/// Per-stream bookkeeping for [`restore_custom_tool_calls_in_event`]. +#[derive(Default)] +pub struct CustomToolCallStreamState { + /// Output indexes whose item was rewritten into a custom tool call. + custom_indexes: HashSet, +} + +/// Rewrites a streamed Responses event so a custom tool's call reaches the client in the shape +/// it expects. Item events are rewritten in place; argument delta events for a rewritten item +/// are dropped (returns `false`), because a partial JSON delta cannot be turned into a +/// freeform input delta and clients read the completed item instead. +pub fn restore_custom_tool_calls_in_event( + event: &mut Value, + custom: &HashSet, + state: &mut CustomToolCallStreamState, +) -> bool { + if custom.is_empty() { + return true; + } + let kind = event + .get("type") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(); + let index = event.get("output_index").and_then(Value::as_u64); + match kind.as_str() { + "response.output_item.added" | "response.output_item.done" => { + if let Some(item) = event.get_mut("item") + && rewrite_item(item, custom) + && let Some(index) = index + { + state.custom_indexes.insert(index); + } + true + } + "response.function_call_arguments.delta" | "response.function_call_arguments.done" => { + !index.is_some_and(|index| state.custom_indexes.contains(&index)) + } + "response.completed" | "response.incomplete" | "response.failed" => { + if let Some(response) = event.get_mut("response") { + restore_custom_tool_calls(response, custom); + } + true + } + _ => true, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn input_is_read_from_the_input_argument_or_left_verbatim() { + assert_eq!(input_from_arguments(&json!({"input": "ls -la"})), "ls -la"); + assert_eq!(input_from_arguments(&json!("{\"input\":\"pwd\"}")), "pwd"); + assert_eq!(input_from_arguments(&json!("raw text")), "raw text"); + assert_eq!( + input_from_arguments(&json!({"cmd": "x"})), + "{\"cmd\":\"x\"}" + ); + } + + #[test] + fn function_call_items_for_custom_tools_become_custom_tool_calls() { + let custom: HashSet = ["exec".to_string()].into_iter().collect(); + let mut body = json!({"output": [ + {"type": "function_call", "call_id": "c1", "name": "exec", "arguments": "{\"input\":\"ls\"}"}, + {"type": "function_call", "call_id": "c2", "name": "update_plan", "arguments": "{}"} + ]}); + restore_custom_tool_calls(&mut body, &custom); + assert_eq!(body["output"][0]["type"], "custom_tool_call"); + assert_eq!(body["output"][0]["input"], "ls"); + assert!(body["output"][0].get("arguments").is_none()); + assert_eq!(body["output"][1]["type"], "function_call"); + } + + #[test] + fn rewritten_custom_tool_calls_take_the_ctc_id_prefix() { + let custom: HashSet = ["exec".to_string()].into_iter().collect(); + let mut body = json!({"output": [ + {"type": "function_call", "id": "fc_abc_1", "call_id": "c1", "name": "exec", "arguments": "{\"input\":\"ls\"}"} + ]}); + restore_custom_tool_calls(&mut body, &custom); + assert_eq!(body["output"][0]["type"], "custom_tool_call"); + assert_eq!(body["output"][0]["id"], "ctc_abc_1"); + } + + #[test] + fn streamed_argument_deltas_for_custom_tools_are_dropped() { + let custom: HashSet = ["exec".to_string()].into_iter().collect(); + let mut state = CustomToolCallStreamState::default(); + let mut added = json!({"type": "response.output_item.added", "output_index": 1, + "item": {"type": "function_call", "call_id": "c1", "name": "exec", "arguments": ""}}); + assert!(restore_custom_tool_calls_in_event( + &mut added, &custom, &mut state + )); + assert_eq!(added["item"]["type"], "custom_tool_call"); + let mut delta = json!({"type": "response.function_call_arguments.delta", "output_index": 1, "delta": "{\"in"}); + assert!(!restore_custom_tool_calls_in_event( + &mut delta, &custom, &mut state + )); + let mut other = json!({"type": "response.function_call_arguments.delta", "output_index": 2, "delta": "{}"}); + assert!(restore_custom_tool_calls_in_event( + &mut other, &custom, &mut state + )); + let mut done = json!({"type": "response.output_item.done", "output_index": 1, + "item": {"type": "function_call", "call_id": "c1", "name": "exec", "arguments": "{\"input\":\"ls -la\"}"}}); + assert!(restore_custom_tool_calls_in_event( + &mut done, &custom, &mut state + )); + assert_eq!(done["item"]["input"], "ls -la"); + } +} diff --git a/crates/switchyard-translation/src/engine.rs b/crates/switchyard-translation/src/engine.rs index abd30955c..4bd99f8e7 100644 --- a/crates/switchyard-translation/src/engine.rs +++ b/crates/switchyard-translation/src/engine.rs @@ -217,6 +217,10 @@ impl TranslationEngine { &mut output.body, &crate::codex_namespaces::qualified_tool_origins(request_extensions), ); + crate::codex_custom_tools::restore_custom_tool_calls( + &mut output.body, + &crate::codex_custom_tools::custom_tool_names(request_extensions), + ); Ok(output) } diff --git a/crates/switchyard-translation/src/helpers.rs b/crates/switchyard-translation/src/helpers.rs index 061517e24..95ebb75e7 100644 --- a/crates/switchyard-translation/src/helpers.rs +++ b/crates/switchyard-translation/src/helpers.rs @@ -125,6 +125,8 @@ pub fn encode_stream_with_extensions( request_extensions: &switchyard_protocol::ProviderExtensions, ) -> std::result::Result { let origins = crate::codex_namespaces::qualified_tool_origins(request_extensions); + let custom_tools = crate::codex_custom_tools::custom_tool_names(request_extensions); + let mut custom_state = crate::codex_custom_tools::CustomToolCallStreamState::default(); let target_format: FormatId = target.into(); // The target is always a built-in wire format, so this lookup cannot fail; a // failure returns as an `Err` rather than a panic. @@ -152,6 +154,19 @@ pub fn encode_stream_with_extensions( stamp_streamed_response_model(value, target, served_model_for_events.as_deref()); crate::codex_namespaces::restore_qualified_tool_names(value, &origins); } + // Argument deltas for a freeform tool cannot be expressed on the wire; the + // rewritten completed item carries the input instead. + let mut encoded: Vec = encoded + .into_iter() + .filter_map(|mut value| { + crate::codex_custom_tools::restore_custom_tool_calls_in_event( + &mut value, + &custom_tools, + &mut custom_state, + ) + .then_some(value) + }) + .collect(); let terminal = encoded.pop(); for value in encoded { yield value; @@ -172,7 +187,13 @@ pub fn encode_stream_with_extensions( served_model_for_events.as_deref(), ); crate::codex_namespaces::restore_qualified_tool_names(&mut value, &origins); - yield value; + if crate::codex_custom_tools::restore_custom_tool_calls_in_event( + &mut value, + &custom_tools, + &mut custom_state, + ) { + yield value; + } } }; diff --git a/crates/switchyard-translation/src/lib.rs b/crates/switchyard-translation/src/lib.rs index 8f9bbe976..4d0e81da6 100644 --- a/crates/switchyard-translation/src/lib.rs +++ b/crates/switchyard-translation/src/lib.rs @@ -8,6 +8,7 @@ //! servers, Python objects, or FFI bindings. pub mod codecs; +pub(crate) mod codex_custom_tools; pub(crate) mod codex_namespaces; pub mod diagnostic; pub mod engine; diff --git a/crates/switchyard-translation/tests/request_translation.rs b/crates/switchyard-translation/tests/request_translation.rs index d27b21cd9..35bb1d831 100644 --- a/crates/switchyard-translation/tests/request_translation.rs +++ b/crates/switchyard-translation/tests/request_translation.rs @@ -2389,3 +2389,186 @@ fn responses_flat_file_data_survives_into_chat() -> TestResult { assert_eq!(file["file"]["filename"], "report.pdf"); Ok(()) } + +// Codex drives GPT-5 models with freeform ("custom") tools. Through a Responses upstream the +// definition must go out verbatim and the replayed history must keep `custom_tool_call` items +// with their raw `input`; through a chat upstream the tool degrades to a single-argument function. +#[test] +fn responses_request_round_trips_custom_tools_and_custom_tool_calls() -> TestResult { + let engine = TranslationEngine::default(); + let custom_tool = json!({ + "type": "custom", + "name": "exec", + "description": "Runs a shell command.", + "format": {"type": "grammar", "syntax": "lark", "definition": "start: /.*/"} + }); + let body = json!({ + "model": "gpt-5.6-luna", + "input": [ + {"type": "message", "role": "user", "content": "List files"}, + {"type": "custom_tool_call", "call_id": "call_1", "name": "exec", "input": "ls -la"}, + {"type": "custom_tool_call_output", "call_id": "call_1", "output": "README.md"} + ], + "tools": [ + custom_tool, + {"type": "function", "name": "update_plan", "description": "Plan", "parameters": {"type": "object"}} + ] + }); + let policy = TranslationPolicy { + preservation: switchyard_translation::PreservationPolicy::Disabled, + ..TranslationPolicy::default() + }; + + let same = engine + .translate_request( + WireFormat::OpenAiResponses, + WireFormat::OpenAiResponses, + &body, + &policy, + )? + .body; + let tools = same["tools"].as_array().ok_or("tools should be an array")?; + assert!( + tools.iter().any(|tool| tool == &custom_tool), + "custom tool must be re-emitted verbatim: {tools:?}" + ); + let input = same["input"].as_array().ok_or("input should be an array")?; + let call = input + .iter() + .find(|item| item["type"] == "custom_tool_call") + .ok_or("history must keep the custom_tool_call")?; + assert_eq!(call["name"], "exec"); + assert_eq!(call["call_id"], "call_1"); + assert_eq!(call["input"], "ls -la"); + assert!(call.get("arguments").is_none(), "{call}"); + let output = input + .iter() + .find(|item| item["type"] == "custom_tool_call_output") + .ok_or("history must keep the custom_tool_call_output")?; + assert_eq!(output["call_id"], "call_1"); + + let chat = engine + .translate_request( + WireFormat::OpenAiResponses, + WireFormat::OpenAiChat, + &body, + &TranslationPolicy::default(), + )? + .body; + let exec = chat["tools"] + .as_array() + .ok_or("chat tools should be an array")? + .iter() + .find(|tool| tool["function"]["name"] == "exec") + .ok_or("chat upstream should still see the tool")?; + assert_eq!( + exec["function"]["parameters"]["required"], + json!(["input"]), + "{exec}" + ); + Ok(()) +} + +// Codex sends GPT-5 requests in the Responses-lite shape: no top-level `tools`, empty +// `instructions`, the tool definitions inside `input[0]` as an `additional_tools` developer +// item, and the base instructions as a developer message. Those definitions are the request's +// tools, the item must not leak into the conversation, and a Responses upstream must receive +// the request in the same shape. +#[test] +fn responses_lite_additional_tools_item_is_the_tool_list() -> TestResult { + let engine = TranslationEngine::default(); + let tools = json!([ + {"type": "custom", "name": "exec", "description": "Run JS.", + "format": {"type": "grammar", "syntax": "lark", "definition": "start: /.*/"}}, + {"type": "function", "name": "update_plan", "description": "Plan", + "parameters": {"type": "object", "properties": {}}} + ]); + let body = json!({ + "model": "gpt-5.6-luna-switchyard", + "instructions": "", + "input": [ + {"type": "additional_tools", "role": "developer", "tools": tools}, + {"type": "message", "role": "developer", "content": [{"type": "input_text", "text": "You are Codex."}]}, + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "List files"}]}, + {"type": "custom_tool_call", "call_id": "call_1", "name": "exec", "input": "ls"}, + {"type": "custom_tool_call_output", "call_id": "call_1", "output": "README.md"} + ], + "stream": true + }); + let policy = TranslationPolicy { + preservation: switchyard_translation::PreservationPolicy::Disabled, + ..TranslationPolicy::default() + }; + + let decoded = engine.decode_request(WireFormat::OpenAiResponses, &body, &policy)?; + let names = decoded + .request + .tools + .iter() + .map(|tool| tool.name.as_str()) + .collect::>(); + assert_eq!(names, vec!["exec", "update_plan"]); + assert!( + !decoded.request.messages.iter().any(|message| { + message + .content + .iter() + .any(|block| matches!(block, switchyard_protocol::ContentBlock::Unknown { .. })) + }), + "the additional_tools item must not become a conversation message" + ); + + let same = engine + .translate_request( + WireFormat::OpenAiResponses, + WireFormat::OpenAiResponses, + &body, + &policy, + )? + .body; + assert!(same.get("tools").is_none(), "{same}"); + let input = same["input"].as_array().ok_or("input should be an array")?; + assert_eq!(input[0]["type"], "additional_tools"); + assert_eq!(input[0]["role"], "developer"); + assert_eq!(input[0]["tools"], tools); + assert!( + input + .iter() + .skip(1) + .all(|item| item["type"] != "additional_tools"), + "{same}" + ); + assert!( + input + .iter() + .any(|item| item["type"] == "custom_tool_call" && item["input"] == "ls"), + "{same}" + ); + + let chat = engine + .translate_request( + WireFormat::OpenAiResponses, + WireFormat::OpenAiChat, + &body, + &TranslationPolicy::default(), + )? + .body; + let chat_tools = chat["tools"] + .as_array() + .ok_or("chat tools should be an array")?; + let chat_names = chat_tools + .iter() + .map(|tool| tool["function"]["name"].as_str().unwrap_or_default()) + .collect::>(); + assert_eq!(chat_names, vec!["exec", "update_plan"]); + let messages = chat["messages"] + .as_array() + .ok_or("messages should be an array")?; + assert!( + !messages + .iter() + .any(|message| message["content"].to_string().contains("additional_tools")), + "{chat}" + ); + Ok(()) +} diff --git a/crates/switchyard-translation/tests/response_translation.rs b/crates/switchyard-translation/tests/response_translation.rs index f261b5050..e68a1ad61 100644 --- a/crates/switchyard-translation/tests/response_translation.rs +++ b/crates/switchyard-translation/tests/response_translation.rs @@ -7,7 +7,9 @@ pub mod common; use pretty_assertions::assert_eq; use serde_json::json; -use switchyard_translation::{TranslationEngine, TranslationPolicy, WireFormat}; +use switchyard_translation::{ + PreservationPolicy, TranslationEngine, TranslationPolicy, WireFormat, +}; use common::{ REASONING_MODEL, normalized_policy, shell_tool_call, text_and_encrypted_reasoning_details, @@ -325,8 +327,8 @@ fn openai_reasoning_response_translates_to_responses_reasoning_item() -> TestRes assert_eq!(output["output"][0]["type"], "reasoning"); assert_eq!( - output["output"][0]["content"][0], - json!({"type": "reasoning_text", "text": "private reasoning"}) + output["output"][0]["summary"][0], + json!({"type": "summary_text", "text": "private reasoning"}) ); assert_eq!(output["output"][1]["type"], "message"); assert_eq!(output["output"][1]["content"][0]["text"], "Visible answer"); @@ -412,7 +414,7 @@ fn openai_reasoning_only_response_translates_to_responses_reasoning_only() -> Te .ok_or("Responses output should be an array")?; assert_eq!(items.len(), 1); assert_eq!(items[0]["type"], "reasoning"); - assert_eq!(items[0]["content"][0]["text"], "private reasoning"); + assert_eq!(items[0]["summary"][0]["text"], "private reasoning"); Ok(()) } @@ -724,3 +726,132 @@ fn content_filter_and_refusal_translate_across_formats() -> TestResult { assert_eq!(output["stop_details"]["category"], "cyber"); Ok(()) } + +// A Responses reasoning item that carries only `encrypted_content` must survive a +// buffered decode/encode through the codec (preservation disabled so the same-format +// shortcut cannot mask a lossy codec), or a buffering caller loses the client's only +// replayable reasoning payload. +#[test] +fn responses_encrypted_reasoning_item_survives_buffered_round_trip() -> TestResult { + let engine = TranslationEngine::default(); + let body = json!({ + "id": "resp_1", + "object": "response", + "status": "completed", + "model": "kimi-k3", + "output": [ + { + "type": "reasoning", + "id": "rs_upstream", + "status": "completed", + "summary": [], + "encrypted_content": "opaque-encrypted-reasoning" + }, + { + "type": "message", + "id": "msg_1", + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": "done", "annotations": []}] + } + ], + "usage": {"input_tokens": 4, "output_tokens": 3, "total_tokens": 7} + }); + let policy = TranslationPolicy { + preservation: PreservationPolicy::Disabled, + ..TranslationPolicy::default() + }; + + let output = engine + .translate_response( + WireFormat::OpenAiResponses, + WireFormat::OpenAiResponses, + &body, + &policy, + )? + .body; + + let reasoning = output["output"] + .as_array() + .ok_or("Responses output should be an array")? + .iter() + .find(|item| item["type"] == "reasoning") + .ok_or("output should include the reasoning item")?; + assert_eq!(reasoning["encrypted_content"], "opaque-encrypted-reasoning"); + // The payload only verifies upstream under the id it was issued with. + assert_eq!(reasoning["id"], "rs_upstream"); + Ok(()) +} + +// A freeform tool call returned by the upstream must reach the client as a `custom_tool_call` +// again once the response is re-encoded with the request's extensions, and as a function-style +// call with an `input` argument when the client speaks chat. +#[test] +fn responses_custom_tool_call_output_round_trips_with_request_extensions() -> TestResult { + let engine = TranslationEngine::default(); + let request = json!({ + "model": "gpt-5.6-luna", + "input": "List files", + "tools": [{ + "type": "custom", + "name": "exec", + "description": "Runs a shell command.", + "format": {"type": "grammar", "syntax": "lark", "definition": "start: /.*/"} + }] + }); + let decoded_request = engine.decode_request( + WireFormat::OpenAiResponses, + &request, + &TranslationPolicy::default(), + )?; + let response = json!({ + "id": "resp_1", + "object": "response", + "status": "completed", + "model": "gpt-5.6-luna", + "output": [{ + "type": "custom_tool_call", + "id": "ctc_1", + "call_id": "call_1", + "name": "exec", + "input": "ls -la", + "status": "completed" + }], + "usage": {"input_tokens": 4, "output_tokens": 3, "total_tokens": 7} + }); + let policy = TranslationPolicy { + preservation: PreservationPolicy::Disabled, + ..TranslationPolicy::default() + }; + + let ir = engine + .decode_response(WireFormat::OpenAiResponses, &response, &policy)? + .response; + let encoded = engine + .encode_response_with_extensions( + WireFormat::OpenAiResponses, + &ir, + &decoded_request.request.extensions, + &policy, + )? + .body; + let item = encoded["output"] + .as_array() + .ok_or("output should be an array")? + .iter() + .find(|item| item["type"] == "custom_tool_call") + .ok_or("the call must be re-emitted as custom_tool_call")?; + assert_eq!(item["name"], "exec"); + assert_eq!(item["call_id"], "call_1"); + assert_eq!(item["input"], "ls -la"); + assert!(item.get("arguments").is_none(), "{item}"); + + // Without the request extensions (e.g. a plain chat client) the call stays function-style. + let chat = engine + .encode_response(WireFormat::OpenAiChat, &ir, &policy)? + .body; + let call = &chat["choices"][0]["message"]["tool_calls"][0]; + assert_eq!(call["function"]["name"], "exec"); + assert_eq!(call["function"]["arguments"], "{\"input\":\"ls -la\"}"); + Ok(()) +} diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 3c043d59a..f7d886e7e 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1407,12 +1407,129 @@ fn responses_completed_event_is_schema_complete_and_retains_message_id() -> Test "missing response field {field}" ); } - assert_eq!(response["output"][0]["id"], "msg_0"); + // Synthesized item ids carry the response id so they stay unique across turns. + assert_eq!(response["output"][0]["id"], "msg_chatcmpl-test_0"); let done = events .iter() .find(|event| event["type"] == "response.output_item.done") .ok_or("expected response.output_item.done")?; - assert_eq!(done["item"]["id"], "msg_0"); + assert_eq!(done["item"]["id"], "msg_chatcmpl-test_0"); + Ok(()) +} + +// Some upstreams issue response ids several hundred characters long. Synthesized item ids +// embed the response id for uniqueness, but OpenAI rejects item ids over 64 characters, so long +// discriminators must be digested while staying distinct across responses. +#[test] +fn responses_synthesized_item_ids_stay_within_the_openai_length_limit() -> TestResult { + let engine = TranslationEngine::default(); + let mut ids = Vec::new(); + for suffix in ["a", "b"] { + let long_id = format!("chatcmpl-{}{suffix}", "x".repeat(360)); + let mut state = + StreamTranslationState::new(WireFormat::OpenAiChat, WireFormat::OpenAiResponses); + let chunk = json!({ + "id": long_id, + "object": "chat.completion.chunk", + "model": "gpt-4o", + "choices": [{ + "index": 0, + "delta": {"content": "hello"}, + "finish_reason": "stop" + }] + }); + let mut events = engine.translate_event( + &mut state, + WireFormat::OpenAiChat, + WireFormat::OpenAiResponses, + &chunk, + )?; + events.extend(engine.finish_stream(&mut state, WireFormat::OpenAiResponses)?); + let done = events + .iter() + .find(|event| event["type"] == "response.output_item.done") + .ok_or("expected response.output_item.done")?; + let id = done["item"]["id"] + .as_str() + .ok_or("item id should be a string")? + .to_string(); + assert!(id.starts_with("msg_"), "{id}"); + assert!(id.ends_with("_0"), "{id}"); + assert!( + id.chars().count() <= 64, + "{id} is {} chars", + id.chars().count() + ); + ids.push(id); + } + assert_ne!( + ids[0], ids[1], + "distinct responses must yield distinct item ids" + ); + Ok(()) +} + +// Encrypted reasoning is bound to the item id the provider issued it under. When a buffered +// reply is re-streamed to the client, the emitted reasoning item must reuse that id or the +// client's replay fails verification upstream. +#[test] +fn responses_stream_reuses_provider_id_for_encrypted_reasoning() -> TestResult { + let engine = TranslationEngine::default(); + let mut state = + StreamTranslationState::new(WireFormat::OpenAiResponses, WireFormat::OpenAiResponses); + let chunks = vec![ + LlmResponseChunk::MessageStart { + id: Some("resp_upstream".to_string()), + model: Some("gpt-5.6-luna".to_string()), + }, + LlmResponseChunk::ReasoningDetailsDelta { + index: 0, + details: vec![json!({ + "type": "reasoning.encrypted", + "data": "opaque-encrypted-reasoning", + "id": "rs_provider_issued" + })], + text: String::new(), + }, + LlmResponseChunk::ReasoningDelta { + index: 0, + text: "thinking".to_string(), + }, + LlmResponseChunk::TextDelta { + index: 1, + text: "done".to_string(), + }, + LlmResponseChunk::MessageStop { reason: None }, + ]; + let mut events = Vec::new(); + for chunk in chunks { + events.extend(engine.encode_stream_event( + &mut state, + WireFormat::OpenAiResponses, + LlmResponseStreamEvent::new(vec![chunk]), + )?); + } + events.extend(engine.finish_stream(&mut state, WireFormat::OpenAiResponses)?); + + let reasoning_done = events + .iter() + .find(|event| { + event["type"] == "response.output_item.done" && event["item"]["type"] == "reasoning" + }) + .ok_or("expected a completed reasoning item")?; + assert_eq!(reasoning_done["item"]["id"], "rs_provider_issued"); + assert_eq!( + reasoning_done["item"]["encrypted_content"], + "opaque-encrypted-reasoning" + ); + // Every reasoning event references the same provider id. + for event in events.iter().filter(|event| { + event["type"] + .as_str() + .is_some_and(|kind| kind.starts_with("response.reasoning_summary")) + }) { + assert_eq!(event["item_id"], "rs_provider_issued", "{event}"); + } Ok(()) } @@ -1546,3 +1663,534 @@ fn responses_decode_emits_tool_arguments_once() -> TestResult { assert_eq!(seen, arguments); Ok(()) } + +// A Responses `reasoning` output item may carry only `encrypted_content`, with no +// plaintext. The stream decoder must surface it as a `reasoning.encrypted` detail so a +// caller that buffers the stream (the escalation router) still holds something the +// client can replay; otherwise the reasoning is dropped before it reaches the IR. +#[test] +fn responses_stream_decodes_encrypted_reasoning_item_into_details() -> TestResult { + let engine = TranslationEngine::default(); + let format = WireFormat::OpenAiResponses; + let mut state = StreamTranslationState::new(format, format); + let event = json!({ + "type": "response.output_item.done", + "output_index": 0, + "item": { + "type": "reasoning", + "id": "rs_upstream", + "status": "completed", + "summary": [], + "encrypted_content": "opaque-encrypted-reasoning" + } + }); + + let decoded = engine.decode_stream_event(&mut state, format, event)?; + + assert_eq!( + decoded.normalized(), + &[LlmResponseChunk::ReasoningDetailsDelta { + index: 0, + details: vec![json!({ + "type": "reasoning.encrypted", + "data": "opaque-encrypted-reasoning", + "id": "rs_upstream" + })], + text: String::new(), + }] + ); + Ok(()) +} + +// The synthesized encode path (no preserved provider JSON, as produced by +// `AggLlmResponse::into_stream`) must emit an encrypted-only reasoning detail as a +// Responses `reasoning` item carrying `encrypted_content`, not drop it. +#[test] +fn responses_stream_encodes_encrypted_reasoning_details_as_reasoning_item() -> TestResult { + let engine = TranslationEngine::default(); + let format = WireFormat::OpenAiResponses; + let mut state = StreamTranslationState::new(format, format); + let chunks = vec![ + LlmResponseChunk::MessageStart { + id: Some("resp_1".to_string()), + model: Some(REASONING_MODEL.to_string()), + }, + LlmResponseChunk::ReasoningDetailsDelta { + index: 0, + details: vec![json!({ + "type": "reasoning.encrypted", + "data": "opaque-encrypted-reasoning" + })], + text: String::new(), + }, + LlmResponseChunk::MessageStop { reason: None }, + ]; + + let mut events = Vec::new(); + for chunk in chunks { + events.extend(engine.encode_stream_event( + &mut state, + format, + LlmResponseStreamEvent::new(vec![chunk]), + )?); + } + events.extend(engine.finish_stream(&mut state, format)?); + + let done_item = events + .iter() + .filter(|event| event["type"] == "response.output_item.done") + .map(|event| &event["item"]) + .find(|item| item["type"] == "reasoning") + .ok_or("expected a completed reasoning output item")?; + assert_eq!(done_item["encrypted_content"], "opaque-encrypted-reasoning"); + + let completed = events + .iter() + .find(|event| event["type"] == "response.completed") + .ok_or("expected response.completed")?; + let final_reasoning = completed["response"]["output"] + .as_array() + .ok_or("output should be an array")? + .iter() + .find(|item| item["type"] == "reasoning") + .ok_or("final output should include the reasoning item")?; + assert_eq!( + final_reasoning["encrypted_content"], + "opaque-encrypted-reasoning" + ); + Ok(()) +} + +// Some providers deliver a reasoning item only in `response.output_item.done`, as +// plaintext in a top-level `text` field, with no streamed `reasoning_text.delta` events. +// The decoder must surface that text so a buffering caller can re-emit the item. +#[test] +fn responses_stream_decodes_text_only_reasoning_item_from_done() -> TestResult { + let engine = TranslationEngine::default(); + let format = WireFormat::OpenAiResponses; + let mut state = StreamTranslationState::new(format, format); + let event = json!({ + "type": "response.output_item.done", + "output_index": 0, + "item": {"type": "reasoning", "id": "rs_upstream", "text": "Let me explore the repo first."} + }); + + let decoded = engine.decode_stream_event(&mut state, format, event)?; + + assert_eq!( + decoded.normalized(), + &[ + LlmResponseChunk::ReasoningDetailsDelta { + index: 0, + details: vec![json!({"type": "reasoning.encrypted", "id": "rs_upstream"})], + text: String::new(), + }, + LlmResponseChunk::ReasoningDelta { + index: 0, + text: "Let me explore the repo first.".to_string(), + } + ] + ); + Ok(()) +} + +// When reasoning text already streamed through `reasoning_text.delta`, the completed item +// that repeats it must not be decoded a second time. +#[test] +fn responses_stream_does_not_duplicate_streamed_reasoning_on_done() -> TestResult { + let engine = TranslationEngine::default(); + let format = WireFormat::OpenAiResponses; + let mut state = StreamTranslationState::new(format, format); + let delta = json!({ + "type": "response.reasoning_text.delta", + "output_index": 0, + "content_index": 0, + "delta": "Let me explore" + }); + let done = json!({ + "type": "response.output_item.done", + "output_index": 0, + "item": { + "type": "reasoning", + "id": "rs_upstream", + "content": [{"type": "reasoning_text", "text": "Let me explore"}] + } + }); + + let first = engine.decode_stream_event(&mut state, format, delta)?; + let second = engine.decode_stream_event(&mut state, format, done)?; + + assert_eq!(first.normalized().len(), 1); + // The done item contributes only its provider id, never the text again. + assert_eq!( + second.normalized(), + &[LlmResponseChunk::ReasoningDetailsDelta { + index: 0, + details: vec![json!({"type": "reasoning.encrypted", "id": "rs_upstream"})], + text: String::new(), + }] + ); + Ok(()) +} + +// Providers differ in which event carries a reasoning item's text. Each carrier must decode +// exactly once: `output_item.added` with text, a `reasoning_text.done` with no prior deltas, +// and a reasoning item that appears only inside `response.completed`'s output array. +#[test] +fn responses_stream_decodes_reasoning_text_from_added_done_and_completed_once() -> TestResult { + let engine = TranslationEngine::default(); + let format = WireFormat::OpenAiResponses; + + // (a) only in output_item.added + let mut state = StreamTranslationState::new(format, format); + let added = json!({"type": "response.output_item.added", "output_index": 0, + "item": {"type": "reasoning", "id": "rs_a", "text": "from added"}}); + let decoded = engine.decode_stream_event(&mut state, format, added)?; + assert_eq!( + decoded.normalized(), + &[ + LlmResponseChunk::ReasoningDetailsDelta { + index: 0, + details: vec![json!({"type": "reasoning.encrypted", "id": "rs_a"})], + text: String::new(), + }, + LlmResponseChunk::ReasoningDelta { + index: 0, + text: "from added".into() + } + ] + ); + // the matching done repeats it and must be skipped + let done = json!({"type": "response.output_item.done", "output_index": 0, + "item": {"type": "reasoning", "id": "rs_a", "text": "from added"}}); + assert_eq!( + engine + .decode_stream_event(&mut state, format, done)? + .normalized(), + &[] + ); + + // (b) only in reasoning_text.done + let mut state = StreamTranslationState::new(format, format); + let text_done = json!({"type": "response.reasoning_text.done", "output_index": 1, + "content_index": 0, "text": "from text done"}); + let decoded = engine.decode_stream_event(&mut state, format, text_done)?; + assert_eq!( + decoded.normalized(), + &[LlmResponseChunk::ReasoningDelta { + index: 1, + text: "from text done".into() + }] + ); + + // (c) only inside response.completed output; emitted before the stop + let mut state = StreamTranslationState::new(format, format); + let completed = json!({"type": "response.completed", "response": {"id": "resp_1", + "output": [{"type": "reasoning", "id": "rs_c", "text": "from completed"}], + "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2}}}); + let decoded = engine.decode_stream_event(&mut state, format, completed)?; + let kinds: Vec<&str> = decoded + .normalized() + .iter() + .map(|c| match c { + LlmResponseChunk::ReasoningDelta { text, .. } => { + assert_eq!(text, "from completed"); + "reasoning" + } + LlmResponseChunk::ReasoningDetailsDelta { .. } => "id", + LlmResponseChunk::Usage(_) => "usage", + LlmResponseChunk::MessageStop { .. } => "stop", + _ => "other", + }) + .collect(); + assert_eq!(kinds, vec!["id", "reasoning", "usage", "stop"]); + + // (d) completed output repeating already-streamed reasoning adds nothing + let mut state = StreamTranslationState::new(format, format); + let delta = + json!({"type": "response.reasoning_text.delta", "output_index": 0, "delta": "streamed"}); + engine.decode_stream_event(&mut state, format, delta)?; + let completed = json!({"type": "response.completed", "response": {"id": "resp_2", + "output": [{"type": "reasoning", "id": "rs_d", "text": "streamed"}]}}); + let decoded = engine.decode_stream_event(&mut state, format, completed)?; + assert!( + decoded + .normalized() + .iter() + .all(|c| !matches!(c, LlmResponseChunk::ReasoningDelta { .. })) + ); + Ok(()) +} + +// Codex records a reasoning item only in the standard Responses shape: text lives in +// `summary: [{"type": "summary_text", ...}]` and streams as `reasoning_summary_part.added`, +// `reasoning_summary_text.delta`, `reasoning_summary_text.done`, `reasoning_summary_part.done`. +// The encoder must emit that shape, not a `content: [reasoning_text]` item, or the client +// silently drops the reasoning. +#[test] +fn responses_stream_encodes_reasoning_as_summary_text() -> TestResult { + let engine = TranslationEngine::default(); + let format = WireFormat::OpenAiResponses; + let mut state = StreamTranslationState::new(format, format); + let chunks = vec![ + LlmResponseChunk::MessageStart { + id: Some("resp_1".into()), + model: Some(REASONING_MODEL.into()), + }, + LlmResponseChunk::ReasoningDelta { + index: 0, + text: "Let me ".into(), + }, + LlmResponseChunk::ReasoningDelta { + index: 0, + text: "think.".into(), + }, + LlmResponseChunk::TextDelta { + index: 1, + text: "done".into(), + }, + LlmResponseChunk::MessageStop { reason: None }, + ]; + let mut events = Vec::new(); + for chunk in chunks { + events.extend(engine.encode_stream_event( + &mut state, + format, + LlmResponseStreamEvent::new(vec![chunk]), + )?); + } + events.extend(engine.finish_stream(&mut state, format)?); + let types: Vec<&str> = events.iter().filter_map(|e| e["type"].as_str()).collect(); + + assert!( + types.contains(&"response.reasoning_summary_part.added"), + "{types:?}" + ); + assert_eq!( + types + .iter() + .filter(|t| **t == "response.reasoning_summary_text.delta") + .count(), + 2, + "{types:?}" + ); + assert!( + types.contains(&"response.reasoning_summary_text.done"), + "{types:?}" + ); + assert!( + types.contains(&"response.reasoning_summary_part.done"), + "{types:?}" + ); + assert!( + !types + .iter() + .any(|t| t.starts_with("response.reasoning_text.")), + "legacy reasoning_text events must not be emitted: {types:?}" + ); + + let done = events + .iter() + .filter(|e| e["type"] == "response.output_item.done") + .map(|e| &e["item"]) + .find(|i| i["type"] == "reasoning") + .ok_or("reasoning output_item.done")?; + assert_eq!( + done["summary"], + json!([{"type": "summary_text", "text": "Let me think."}]) + ); + assert!( + done.get("content") + .is_none_or(|c| c.as_array().is_some_and(Vec::is_empty)), + "no content part: {done}" + ); + + let completed = events + .iter() + .find(|e| e["type"] == "response.completed") + .ok_or("completed")?; + let final_reasoning = completed["response"]["output"] + .as_array() + .ok_or("output")? + .iter() + .find(|i| i["type"] == "reasoning") + .ok_or("final reasoning item")?; + assert_eq!(final_reasoning["summary"][0]["text"], "Let me think."); + Ok(()) +} + +// Item ids the encoder synthesizes must be unique across responses. A client replays the +// whole conversation, so two turns whose reasoning items are both `rs_0` (and whose tool +// calls are both `fc_1`) hand the upstream a history with colliding ids. Passthrough carries +// the provider's unique ids; the synthesized path must not be worse. +#[test] +fn responses_stream_synthesized_item_ids_are_unique_across_responses() -> TestResult { + let engine = TranslationEngine::default(); + let format = WireFormat::OpenAiResponses; + let mut ids = Vec::new(); + for resp in ["resp_first", "resp_second"] { + let mut state = StreamTranslationState::new(format, format); + let chunks = vec![ + LlmResponseChunk::MessageStart { + id: Some(resp.into()), + model: Some(REASONING_MODEL.into()), + }, + LlmResponseChunk::ReasoningDelta { + index: 0, + text: "think".into(), + }, + LlmResponseChunk::ToolCallDelta { + index: 1, + id: Some("call_x".into()), + name: Some("exec_command".into()), + arguments_delta: Some("{}".into()), + }, + LlmResponseChunk::TextDelta { + index: 2, + text: "done".into(), + }, + LlmResponseChunk::MessageStop { reason: None }, + ]; + let mut events = Vec::new(); + for chunk in chunks { + events.extend(engine.encode_stream_event( + &mut state, + format, + LlmResponseStreamEvent::new(vec![chunk]), + )?); + } + events.extend(engine.finish_stream(&mut state, format)?); + for e in events + .iter() + .filter(|e| e["type"] == "response.output_item.done") + { + ids.push(( + e["item"]["type"].as_str().unwrap_or("").to_string(), + e["item"]["id"].as_str().unwrap_or("").to_string(), + )); + } + } + let mut seen = std::collections::HashSet::new(); + for (kind, id) in &ids { + assert!(!id.is_empty(), "{kind} item without id"); + assert!( + seen.insert(id.clone()), + "item id {id} ({kind}) repeated across responses: {ids:?}" + ); + } + Ok(()) +} + +// Decodes a provider Responses stream and re-encodes it the way the buffered/escalation path +// does: from normalized chunks, with no preserved provider JSON. +fn reencode_responses_events( + events: Vec, +) -> Result, Box> { + let engine = TranslationEngine::default(); + let format = WireFormat::OpenAiResponses; + let mut decode_state = StreamTranslationState::new(format, format); + let mut encode_state = StreamTranslationState::new(format, format); + let mut out = Vec::new(); + for event in events { + let decoded = engine.decode_stream_event(&mut decode_state, format, event)?; + for chunk in decoded.normalized() { + out.extend(engine.encode_stream_event( + &mut encode_state, + format, + LlmResponseStreamEvent::new(vec![chunk.clone()]), + )?); + } + } + out.extend(engine.finish_stream(&mut encode_state, format)?); + Ok(out) +} + +// GPT-5 emits a reasoning item ahead of each tool call. Every one of them must reach the +// client under its provider id with its encrypted payload, in the provider's order. +#[test] +fn responses_stream_keeps_every_reasoning_item_of_a_response() -> TestResult { + let events = vec![ + json!({"type": "response.created", "response": {"id": "resp_up", "model": "gpt-5.6-luna"}}), + json!({"type": "response.output_item.added", "output_index": 0, + "item": {"type": "reasoning", "id": "rs_a", "summary": []}}), + json!({"type": "response.output_item.done", "output_index": 0, + "item": {"type": "reasoning", "id": "rs_a", "summary": [], "encrypted_content": "enc-a"}}), + json!({"type": "response.output_item.added", "output_index": 1, + "item": {"type": "function_call", "id": "fc_1", "call_id": "call_1", "name": "exec", "arguments": ""}}), + json!({"type": "response.output_item.done", "output_index": 1, + "item": {"type": "function_call", "id": "fc_1", "call_id": "call_1", "name": "exec", "arguments": "{\"cmd\":\"ls\"}"}}), + json!({"type": "response.output_item.added", "output_index": 2, + "item": {"type": "reasoning", "id": "rs_b", "summary": []}}), + json!({"type": "response.output_item.done", "output_index": 2, + "item": {"type": "reasoning", "id": "rs_b", "summary": [], "encrypted_content": "enc-b"}}), + json!({"type": "response.output_item.added", "output_index": 3, + "item": {"type": "function_call", "id": "fc_3", "call_id": "call_3", "name": "exec", "arguments": ""}}), + json!({"type": "response.output_item.done", "output_index": 3, + "item": {"type": "function_call", "id": "fc_3", "call_id": "call_3", "name": "exec", "arguments": "{\"cmd\":\"pwd\"}"}}), + ]; + let events = reencode_responses_events(events)?; + + let reasoning_done: Vec<&Value> = events + .iter() + .filter(|event| { + event["type"] == "response.output_item.done" && event["item"]["type"] == "reasoning" + }) + .collect(); + assert_eq!(reasoning_done.len(), 2, "both reasoning items must close"); + assert_eq!(reasoning_done[0]["item"]["id"], "rs_a"); + assert_eq!(reasoning_done[0]["item"]["encrypted_content"], "enc-a"); + assert_eq!(reasoning_done[1]["item"]["id"], "rs_b"); + assert_eq!(reasoning_done[1]["item"]["encrypted_content"], "enc-b"); + + let completed = events + .iter() + .find(|event| event["type"] == "response.completed") + .ok_or("expected response.completed")?; + let kinds: Vec<&str> = completed["response"]["output"] + .as_array() + .ok_or("output array")? + .iter() + .filter_map(|item| item["type"].as_str()) + .collect(); + assert_eq!( + kinds, + ["reasoning", "function_call", "reasoning", "function_call"] + ); + Ok(()) +} + +// Summary text streams before the encrypted payload arrives on `done`. The item must open +// under the provider id announced by `added`, not a synthesized one, so the payload is kept. +#[test] +fn responses_stream_opens_reasoning_under_provider_id_before_payload_arrives() -> TestResult { + let events = vec![ + json!({"type": "response.created", "response": {"id": "resp_up", "model": "gpt-5.6-luna"}}), + json!({"type": "response.output_item.added", "output_index": 0, + "item": {"type": "reasoning", "id": "rs_a", "summary": []}}), + json!({"type": "response.reasoning_summary_text.delta", "output_index": 0, "item_id": "rs_a", + "summary_index": 0, "delta": "plan"}), + json!({"type": "response.output_item.done", "output_index": 0, + "item": {"type": "reasoning", "id": "rs_a", "summary": [{"type": "summary_text", "text": "plan"}], + "encrypted_content": "enc-a"}}), + ]; + let events = reencode_responses_events(events)?; + + let added: Vec<&Value> = events + .iter() + .filter(|event| { + event["type"] == "response.output_item.added" && event["item"]["type"] == "reasoning" + }) + .collect(); + assert_eq!(added.len(), 1, "one reasoning item opens"); + assert_eq!(added[0]["item"]["id"], "rs_a"); + let done = events + .iter() + .find(|event| { + event["type"] == "response.output_item.done" && event["item"]["type"] == "reasoning" + }) + .ok_or("expected a completed reasoning item")?; + assert_eq!(done["item"]["id"], "rs_a"); + assert_eq!(done["item"]["encrypted_content"], "enc-a"); + assert_eq!(done["item"]["summary"][0]["text"], "plan"); + Ok(()) +}