From 24b82ac695abc04a7f62404ed12ffcdc754a3f79 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Sat, 5 Sep 2026 11:38:18 -0700 Subject: [PATCH 01/13] fix(translation): carry encrypted reasoning through the Responses codec Signed-off-by: Lin Jia --- .../src/codecs/common.rs | 11 +++ .../src/codecs/responses/buffered.rs | 45 ++++++--- .../src/codecs/responses/stream.rs | 86 ++++++++++++----- .../src/codecs/stream.rs | 3 + .../tests/response_translation.rs | 58 ++++++++++- .../tests/stream_translation.rs | 96 +++++++++++++++++++ 6 files changed, 264 insertions(+), 35 deletions(-) diff --git a/crates/switchyard-translation/src/codecs/common.rs b/crates/switchyard-translation/src/codecs/common.rs index 78e5a7f89..d56081af1 100644 --- a/crates/switchyard-translation/src/codecs/common.rs +++ b/crates/switchyard-translation/src/codecs/common.rs @@ -62,6 +62,17 @@ pub(crate) fn reasoning_text_from_details(details: &[Value]) -> Option { (!parts.is_empty()).then(|| parts.join("\n")) } +/// Returns the opaque payload of the first `reasoning.encrypted` detail, if any. +pub(crate) fn encrypted_reasoning_data(details: &[Value]) -> Option { + details + .iter() + .filter_map(Value::as_object) + .find(|detail| detail.get("type").and_then(Value::as_str) == Some("reasoning.encrypted")) + .and_then(|detail| detail.get("data").and_then(Value::as_str)) + .filter(|data| !data.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, diff --git a/crates/switchyard-translation/src/codecs/responses/buffered.rs b/crates/switchyard-translation/src/codecs/responses/buffered.rs index 03e11f15b..66edc3c92 100644 --- a/crates/switchyard-translation/src/codecs/responses/buffered.rs +++ b/crates/switchyard-translation/src/codecs/responses/buffered.rs @@ -8,7 +8,8 @@ use std::collections::HashSet; use serde_json::{Map, Value, json}; use crate::codecs::common::{ - is_known_role_name, provider_extensions, reasoning_text_from_blocks, text_from_blocks, + encrypted_reasoning_data, is_known_role_name, provider_extensions, reasoning_text_from_blocks, + text_from_blocks, }; use crate::codecs::openai_chat::{decode_file_source, decode_image_source}; use crate::codecs::{ @@ -653,10 +654,17 @@ fn decode_responses_reasoning_item(item: &Map) -> Vec 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, + }); + if !reasoning.is_empty() || encrypted_reasoning.is_some() { + items.push(encode_responses_reasoning_output( + &reasoning, + encrypted_reasoning.as_deref(), + )); } if !text.is_empty() || (!has_tool_calls && reasoning.is_empty()) { @@ -1376,18 +1391,24 @@ 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>) -> Value { + let mut content = Vec::new(); + if !text.is_empty() { + content.push(json!({"type": "reasoning_text", "text": text})); + } + let mut item = json!({ "type": "reasoning", "id": "rs_switchyard", "status": "completed", - "content": [{ - "type": "reasoning_text", - "text": text, - }], + "content": content, "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..bbe3b2cce 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -7,6 +7,7 @@ use serde::Serialize; use serde_json::{Value, json}; use crate::LlmResponseChunk; +use crate::codecs::common::encrypted_reasoning_data; use crate::codecs::stream::{ StreamCodec, StreamTranslationState, record_source_identity, target_message_id_or_source_message_id, target_model_or_source_model, @@ -239,10 +240,18 @@ fn encode_responses_stream( LlmResponseChunk::ReasoningDelta { text, .. } => { encode_responses_reasoning_delta(state, text) } - LlmResponseChunk::ReasoningDetailsDelta { text, .. } if !text.is_empty() => { - encode_responses_reasoning_delta(state, text) + LlmResponseChunk::ReasoningDetailsDelta { 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. + if let Some(data) = encrypted_reasoning_data(&details) { + state.response_reasoning_encrypted = Some(data); + } + let mut out = ensure_responses_reasoning_started(state); + if !text.is_empty() { + out.extend(encode_responses_reasoning_delta(state, text)); + } + out } - LlmResponseChunk::ReasoningDetailsDelta { .. } => Vec::new(), LlmResponseChunk::ToolCallDelta { index, id, @@ -309,22 +318,31 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { 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!({ + // Encrypted-only reasoning streamed no text, so it gets no text part; the item + // itself still closes so the client can replay its `encrypted_content`. + let mut content = Vec::new(); + if !state.response_reasoning_text.is_empty() { + out.push(json!({ + "type": "response.reasoning_text.done", + "output_index": output_index, + "content_index": 0, + "text": state.response_reasoning_text, + })); + content.push(json!({ + "type": "reasoning_text", + "text": state.response_reasoning_text, + })); + } + let mut item = json!({ "type": "reasoning", "id": format!("rs_{output_index}"), "status": "completed", - "content": [{ - "type": "reasoning_text", - "text": state.response_reasoning_text, - }], + "content": content, "summary": [], }); + if let Some(encrypted) = &state.response_reasoning_encrypted { + item["encrypted_content"] = Value::String(encrypted.clone()); + } out.push(json!({ "type": "response.output_item.done", "output_index": output_index, @@ -438,13 +456,31 @@ 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; + // A reasoning item may carry only `encrypted_content`, with no streamed text. Surface it + // as a `reasoning.encrypted` detail so a buffering caller still holds something the + // client can replay. Text already arrived through `reasoning_text.delta`, so none is + // repeated here. + if item.get("type").and_then(Value::as_str) == Some("reasoning") { + return item + .get("encrypted_content") + .and_then(Value::as_str) + .filter(|data| !data.is_empty()) + .map(|data| { + vec![LlmResponseChunk::ReasoningDetailsDelta { + index, + details: vec![json!({"type": "reasoning.encrypted", "data": data})], + text: String::new(), + }] + }) + .unwrap_or_default(); + } + if item.get("type").and_then(Value::as_str) != Some("function_call") { + return Vec::new(); + } let arguments = item.get("arguments").and_then(Value::as_str); if let Some(arguments) = arguments { // Compared against what THIS decoder has seen. Reading the encoder's @@ -557,11 +593,8 @@ fn encode_responses_text_delta(state: &mut StreamTranslationState, text: String) out } -// Accumulates reasoning text and emits Responses reasoning events. -fn encode_responses_reasoning_delta( - state: &mut StreamTranslationState, - text: String, -) -> Vec { +// Opens the Responses reasoning output item once, emitting its `added` events. +fn ensure_responses_reasoning_started(state: &mut StreamTranslationState) -> Vec { let mut out = ensure_responses_created(state); if !state.response_reasoning_started { state.response_reasoning_started = true; @@ -586,6 +619,15 @@ fn encode_responses_reasoning_delta( "text": "", })); } + out +} + +// Accumulates reasoning text and emits Responses reasoning events. +fn encode_responses_reasoning_delta( + state: &mut StreamTranslationState, + text: String, +) -> Vec { + let mut out = ensure_responses_reasoning_started(state); state.response_reasoning_text.push_str(&text); out.push(json!({ "type": "response.reasoning_text.delta", diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index 0ec9b7f88..72cceb4f7 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -60,6 +60,9 @@ pub struct StreamTranslationState { pub(crate) response_reasoning_started: bool, pub(crate) response_reasoning_output_index: Option, pub(crate) response_reasoning_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) response_reasoning_encrypted: Option, pub(crate) next_response_output_index: usize, pub(crate) response_sequence_number: u64, diff --git a/crates/switchyard-translation/tests/response_translation.rs b/crates/switchyard-translation/tests/response_translation.rs index f261b5050..eb3629f15 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, @@ -724,3 +726,57 @@ 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"); + Ok(()) +} diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 3c043d59a..7a1abd505 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1546,3 +1546,99 @@ 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" + })], + 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(()) +} From 6ee30158e06633b4acf8985fb298d721188e998b Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Sat, 5 Sep 2026 13:38:35 -0700 Subject: [PATCH 02/13] fix(translation): decode text-only Responses reasoning items on output_item.done Signed-off-by: Lin Jia --- .../src/codecs/common.rs | 27 +++++++++ .../src/codecs/responses/buffered.rs | 28 +-------- .../src/codecs/responses/stream.rs | 60 +++++++++++++------ .../src/codecs/stream.rs | 3 + .../tests/stream_translation.rs | 57 ++++++++++++++++++ 5 files changed, 131 insertions(+), 44 deletions(-) diff --git a/crates/switchyard-translation/src/codecs/common.rs b/crates/switchyard-translation/src/codecs/common.rs index d56081af1..d666cf9b1 100644 --- a/crates/switchyard-translation/src/codecs/common.rs +++ b/crates/switchyard-translation/src/codecs/common.rs @@ -62,6 +62,33 @@ 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 `reasoning.encrypted` detail, if any. pub(crate) fn encrypted_reasoning_data(details: &[Value]) -> Option { details diff --git a/crates/switchyard-translation/src/codecs/responses/buffered.rs b/crates/switchyard-translation/src/codecs/responses/buffered.rs index 66edc3c92..51634372c 100644 --- a/crates/switchyard-translation/src/codecs/responses/buffered.rs +++ b/crates/switchyard-translation/src/codecs/responses/buffered.rs @@ -8,8 +8,8 @@ use std::collections::HashSet; use serde_json::{Map, Value, json}; use crate::codecs::common::{ - encrypted_reasoning_data, is_known_role_name, provider_extensions, reasoning_text_from_blocks, - text_from_blocks, + collect_responses_reasoning_text, encrypted_reasoning_data, is_known_role_name, + provider_extensions, reasoning_text_from_blocks, text_from_blocks, }; use crate::codecs::openai_chat::{decode_file_source, decode_image_source}; use crate::codecs::{ @@ -669,30 +669,6 @@ 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 { diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index bbe3b2cce..14185fc15 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -7,7 +7,7 @@ use serde::Serialize; use serde_json::{Value, json}; use crate::LlmResponseChunk; -use crate::codecs::common::encrypted_reasoning_data; +use crate::codecs::common::{collect_responses_reasoning_text, encrypted_reasoning_data}; use crate::codecs::stream::{ StreamCodec, StreamTranslationState, record_source_identity, target_message_id_or_source_message_id, target_model_or_source_model, @@ -144,11 +144,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(), }] }) @@ -460,23 +468,39 @@ fn decode_responses_output_item_done( .get("output_index") .and_then(Value::as_u64) .unwrap_or(0) as usize; - // A reasoning item may carry only `encrypted_content`, with no streamed text. Surface it - // as a `reasoning.encrypted` detail so a buffering caller still holds something the - // client can replay. Text already arrived through `reasoning_text.delta`, so none is - // repeated here. + // A completed reasoning item is the only place some providers put the reasoning at all: + // as plaintext in `text`, `content`, or `summary`, with no streamed deltas; or as an + // opaque `encrypted_content`. Surface whichever is present so a buffering caller still + // holds something the client can replay. Text that already streamed through + // `reasoning_text.delta` is not repeated. if item.get("type").and_then(Value::as_str) == Some("reasoning") { - return item + let mut out = Vec::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()) - .map(|data| { - vec![LlmResponseChunk::ReasoningDetailsDelta { - index, - details: vec![json!({"type": "reasoning.encrypted", "data": data})], - text: String::new(), - }] - }) - .unwrap_or_default(); + { + out.push(LlmResponseChunk::ReasoningDetailsDelta { + index, + details: vec![json!({"type": "reasoning.encrypted", "data": data})], + text: String::new(), + }); + } + return out; } if item.get("type").and_then(Value::as_str) != Some("function_call") { return Vec::new(); diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index 72cceb4f7..97049591b 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -52,6 +52,9 @@ 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, pub(crate) response_created: bool, pub(crate) response_text_started: bool, diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 7a1abd505..727f0e308 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1642,3 +1642,60 @@ fn responses_stream_encodes_encrypted_reasoning_details_as_reasoning_item() -> T ); 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::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); + assert_eq!(second.normalized(), &[]); + Ok(()) +} From 384ce2682548cf413f7d1645ed422f61e35a9065 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Sat, 5 Sep 2026 16:12:11 -0700 Subject: [PATCH 03/13] fix(translation): decode Responses reasoning from added, done, text.done, and completed carriers Signed-off-by: Lin Jia --- .../src/codecs/responses/stream.rs | 147 +++++++++++++----- .../src/codecs/stream.rs | 2 + .../tests/stream_translation.rs | 81 ++++++++++ 3 files changed, 195 insertions(+), 35 deletions(-) diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index 14185fc15..23f7505b2 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -190,8 +190,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) @@ -414,6 +478,47 @@ 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 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); + out.push(LlmResponseChunk::ReasoningDetailsDelta { + index, + details: vec![json!({"type": "reasoning.encrypted", "data": data})], + text: String::new(), + }); + } + out +} + fn decode_responses_output_item_added( event: &Value, state: &mut StreamTranslationState, @@ -421,13 +526,16 @@ 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; + if item.get("type").and_then(Value::as_str) == Some("reasoning") { + return decode_responses_reasoning_item(item, index, state); + } + if item.get("type").and_then(Value::as_str) != Some("function_call") { + return Vec::new(); + } let arguments_delta = item .get("arguments") .and_then(Value::as_str) @@ -468,39 +576,8 @@ fn decode_responses_output_item_done( .get("output_index") .and_then(Value::as_u64) .unwrap_or(0) as usize; - // A completed reasoning item is the only place some providers put the reasoning at all: - // as plaintext in `text`, `content`, or `summary`, with no streamed deltas; or as an - // opaque `encrypted_content`. Surface whichever is present so a buffering caller still - // holds something the client can replay. Text that already streamed through - // `reasoning_text.delta` is not repeated. if item.get("type").and_then(Value::as_str) == Some("reasoning") { - let mut out = Vec::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()) - { - out.push(LlmResponseChunk::ReasoningDetailsDelta { - index, - details: vec![json!({"type": "reasoning.encrypted", "data": data})], - text: String::new(), - }); - } - return out; + return decode_responses_reasoning_item(item, index, state); } if item.get("type").and_then(Value::as_str) != Some("function_call") { return Vec::new(); diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index 97049591b..56b5252a8 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -55,6 +55,8 @@ pub struct StreamTranslationState { /// 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, diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 727f0e308..83af0148d 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1699,3 +1699,84 @@ fn responses_stream_does_not_duplicate_streamed_reasoning_on_done() -> TestResul assert_eq!(second.normalized(), &[]); 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::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::Usage(_) => "usage", + LlmResponseChunk::MessageStop { .. } => "stop", + _ => "other", + }) + .collect(); + assert_eq!(kinds, vec!["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(()) +} From a214f4c37bea2e449915cbe95d736220be76ace3 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Sat, 5 Sep 2026 16:12:12 -0700 Subject: [PATCH 04/13] feat(translation): opt-in trace of raw Responses stream events Signed-off-by: Lin Jia --- Cargo.lock | 1 + crates/switchyard-translation/Cargo.toml | 1 + crates/switchyard-translation/src/codecs/responses/stream.rs | 3 +++ 3 files changed, 5 insertions(+) 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/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/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index 23f7505b2..eeec9e36a 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -100,6 +100,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")) From 879529b959c1a9fd802efb5ffa7598a5c5464f35 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Sat, 5 Sep 2026 16:53:08 -0700 Subject: [PATCH 05/13] fix(translation): emit Responses reasoning in the standard summary_text shape Signed-off-by: Lin Jia --- .../src/codecs/responses/buffered.rs | 8 +- .../src/codecs/responses/stream.rs | 44 ++++++--- .../tests/response_translation.rs | 6 +- .../tests/stream_translation.rs | 97 +++++++++++++++++++ 4 files changed, 132 insertions(+), 23 deletions(-) diff --git a/crates/switchyard-translation/src/codecs/responses/buffered.rs b/crates/switchyard-translation/src/codecs/responses/buffered.rs index 51634372c..63a24b9ee 100644 --- a/crates/switchyard-translation/src/codecs/responses/buffered.rs +++ b/crates/switchyard-translation/src/codecs/responses/buffered.rs @@ -1370,16 +1370,16 @@ fn encode_responses_output(outputs: &[ResponseOutput]) -> Value { // 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>) -> Value { - let mut content = Vec::new(); + // Standard Responses shape: text as `summary_text` parts, which is what clients record. + let mut summary = Vec::new(); if !text.is_empty() { - content.push(json!({"type": "reasoning_text", "text": text})); + summary.push(json!({"type": "summary_text", "text": text})); } let mut item = json!({ "type": "reasoning", "id": "rs_switchyard", "status": "completed", - "content": content, - "summary": [], + "summary": summary, }); if let Some(encrypted) = encrypted { item["encrypted_content"] = Value::String(encrypted.to_string()); diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index eeec9e36a..1ff8c43e3 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -393,27 +393,35 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { if state.response_reasoning_started && let Some(output_index) = state.response_reasoning_output_index { - // Encrypted-only reasoning streamed no text, so it gets no text part; the item + // 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 mut content = Vec::new(); + let item_id = format!("rs_{output_index}"); + let mut summary = Vec::new(); if !state.response_reasoning_text.is_empty() { out.push(json!({ - "type": "response.reasoning_text.done", + "type": "response.reasoning_summary_text.done", + "item_id": item_id, "output_index": output_index, - "content_index": 0, + "summary_index": 0, "text": state.response_reasoning_text, })); - content.push(json!({ - "type": "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": state.response_reasoning_text}, + })); + summary.push(json!({ + "type": "summary_text", "text": state.response_reasoning_text, })); } let mut item = json!({ "type": "reasoning", - "id": format!("rs_{output_index}"), + "id": item_id, "status": "completed", - "content": content, - "summary": [], + "summary": summary, }); if let Some(encrypted) = &state.response_reasoning_encrypted { item["encrypted_content"] = Value::String(encrypted.clone()); @@ -705,6 +713,8 @@ fn ensure_responses_reasoning_started(state: &mut StreamTranslationState) -> Vec let output_index = state.next_response_output_index; state.next_response_output_index += 1; state.response_reasoning_output_index = Some(output_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, @@ -712,15 +722,15 @@ fn ensure_responses_reasoning_started(state: &mut StreamTranslationState) -> Vec "type": "reasoning", "id": format!("rs_{output_index}"), "status": "in_progress", - "content": [], "summary": [], }, })); out.push(json!({ - "type": "response.reasoning_text.added", + "type": "response.reasoning_summary_part.added", + "item_id": format!("rs_{output_index}"), "output_index": output_index, - "content_index": 0, - "text": "", + "summary_index": 0, + "part": {"type": "summary_text", "text": ""}, })); } out @@ -733,10 +743,12 @@ fn encode_responses_reasoning_delta( ) -> Vec { let mut out = ensure_responses_reasoning_started(state); state.response_reasoning_text.push_str(&text); + let output_index = state.response_reasoning_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": format!("rs_{output_index}"), + "output_index": output_index, + "summary_index": 0, "delta": text, })); out diff --git a/crates/switchyard-translation/tests/response_translation.rs b/crates/switchyard-translation/tests/response_translation.rs index eb3629f15..2fd6527cd 100644 --- a/crates/switchyard-translation/tests/response_translation.rs +++ b/crates/switchyard-translation/tests/response_translation.rs @@ -327,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"); @@ -414,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(()) } diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 83af0148d..146e1f370 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1780,3 +1780,100 @@ fn responses_stream_decodes_reasoning_text_from_added_done_and_completed_once() ); 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(()) +} From 370793dd7d2876ad45614f4681e0f7dc72faac61 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Sun, 6 Sep 2026 00:40:57 -0700 Subject: [PATCH 06/13] fix(translation): make synthesized Responses item ids unique across responses Signed-off-by: Lin Jia --- .../src/codecs/responses/stream.rs | 37 +++++++--- .../tests/stream_translation.rs | 67 ++++++++++++++++++- 2 files changed, 92 insertions(+), 12 deletions(-) diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index 1ff8c43e3..782a62412 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -381,7 +381,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}], @@ -395,7 +395,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { { // 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 = format!("rs_{output_index}"); + let item_id = responses_item_id(state, "rs", output_index); let mut summary = Vec::new(); if !state.response_reasoning_text.is_empty() { out.push(json!({ @@ -440,7 +440,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}], @@ -460,7 +460,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, @@ -682,7 +682,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": [], @@ -720,14 +720,14 @@ fn ensure_responses_reasoning_started(state: &mut StreamTranslationState) -> Vec "output_index": output_index, "item": { "type": "reasoning", - "id": format!("rs_{output_index}"), + "id": responses_item_id(state, "rs", output_index), "status": "in_progress", "summary": [], }, })); out.push(json!({ "type": "response.reasoning_summary_part.added", - "item_id": format!("rs_{output_index}"), + "item_id": responses_item_id(state, "rs", output_index), "output_index": output_index, "summary_index": 0, "part": {"type": "summary_text", "text": ""}, @@ -746,7 +746,7 @@ fn encode_responses_reasoning_delta( let output_index = state.response_reasoning_output_index.unwrap_or(0); out.push(json!({ "type": "response.reasoning_summary_text.delta", - "item_id": format!("rs_{output_index}"), + "item_id": responses_item_id(state, "rs", output_index), "output_index": output_index, "summary_index": 0, "delta": text, @@ -763,6 +763,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; @@ -782,14 +783,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": "", @@ -868,6 +869,22 @@ fn responses_usage_value(usage: &Usage) -> Value { }) } +// 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 { + let resp = responses_id(state); + let resp = resp.strip_prefix("resp_").unwrap_or(&resp); + format!("{prefix}_{resp}_{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); + format!("{prefix}_{resp}_{output_index}") +} + // 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/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 146e1f370..92cd072f3 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1407,12 +1407,13 @@ 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(()) } @@ -1877,3 +1878,65 @@ fn responses_stream_encodes_reasoning_as_summary_text() -> TestResult { 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(()) +} From 746202f34f2be3aed1acd63f1c57a9633af71f62 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Sun, 6 Sep 2026 21:48:39 -0700 Subject: [PATCH 07/13] fix(translation): keep synthesized Responses item ids within 64 characters Embedding the upstream response id made synthesized item ids unique across turns, but some upstreams issue response ids several hundred characters long, and OpenAI rejects replayed item ids over 64 characters. A session that started on such an upstream and later moved to an OpenAI model failed every request with a 400 on the replayed history. Long response ids are now replaced by a 64-bit FNV-1a digest, which keeps ids distinct per response while bounding their length. Signed-off-by: Lin Jia Co-Authored-By: Claude Fable 5.1 --- .../src/codecs/responses/stream.rs | 23 ++++++-- .../tests/stream_translation.rs | 52 +++++++++++++++++++ 2 files changed, 71 insertions(+), 4 deletions(-) diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index 782a62412..b78d53858 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -869,20 +869,35 @@ 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 { - let resp = responses_id(state); - let resp = resp.strip_prefix("resp_").unwrap_or(&resp); - format!("{prefix}_{resp}_{output_index}") + 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); - format!("{prefix}_{resp}_{output_index}") + 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. diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 92cd072f3..3f921eb87 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1417,6 +1417,58 @@ fn responses_completed_event_is_schema_complete_and_retains_message_id() -> Test 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(()) +} + // Verifies a streamed token-limit stop terminates with response.incomplete. #[test] fn openai_chat_length_finish_translates_to_responses_incomplete_event() -> TestResult { From cc5a4c944a22152c12c3c8e2071995c84a56d4c2 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Mon, 7 Sep 2026 22:02:14 -0700 Subject: [PATCH 08/13] fix(translation): replay encrypted reasoning under the item id it was issued with Encrypted reasoning returned by OpenAI-compatible providers is bound to the output item id it was issued under. The buffered path re-emitted such items with a synthesized id while keeping the payload, so a client that replayed the conversation got a 400 (invalid_encrypted_content: item_id did not match the target item id) on its next request and the session died. This affected any escalation route whose efficient tier returns encrypted reasoning. Both decoders now record the provider item id on the reasoning.encrypted detail, and both encoders reuse that id for the emitted reasoning item. If the id only becomes known after the item has already opened under another id, the payload is dropped with a warning instead of poisoning the replay. Signed-off-by: Lin Jia Co-Authored-By: Claude Fable 5.1 --- .../src/codecs/common.rs | 12 ++++ .../src/codecs/responses/buffered.rs | 31 +++++++-- .../src/codecs/responses/stream.rs | 46 +++++++++++-- .../src/codecs/stream.rs | 3 + .../tests/response_translation.rs | 2 + .../tests/stream_translation.rs | 67 ++++++++++++++++++- 6 files changed, 149 insertions(+), 12 deletions(-) diff --git a/crates/switchyard-translation/src/codecs/common.rs b/crates/switchyard-translation/src/codecs/common.rs index d666cf9b1..b8771c9f8 100644 --- a/crates/switchyard-translation/src/codecs/common.rs +++ b/crates/switchyard-translation/src/codecs/common.rs @@ -100,6 +100,18 @@ pub(crate) fn encrypted_reasoning_data(details: &[Value]) -> Option { .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| detail.get("type").and_then(Value::as_str) == Some("reasoning.encrypted")) + .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, diff --git a/crates/switchyard-translation/src/codecs/responses/buffered.rs b/crates/switchyard-translation/src/codecs/responses/buffered.rs index 63a24b9ee..d3346adad 100644 --- a/crates/switchyard-translation/src/codecs/responses/buffered.rs +++ b/crates/switchyard-translation/src/codecs/responses/buffered.rs @@ -8,8 +8,8 @@ use std::collections::HashSet; use serde_json::{Map, Value, json}; use crate::codecs::common::{ - collect_responses_reasoning_text, encrypted_reasoning_data, is_known_role_name, - provider_extensions, reasoning_text_from_blocks, text_from_blocks, + 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}; use crate::codecs::{ @@ -659,7 +659,18 @@ fn decode_responses_reasoning_item(item: &Map) -> Vec Value { 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(), )); } @@ -1369,15 +1385,20 @@ fn encode_responses_output(outputs: &[ResponseOutput]) -> Value { // 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>) -> Value { +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", "summary": summary, }); diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index b78d53858..f873af383 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -7,7 +7,9 @@ use serde::Serialize; use serde_json::{Value, json}; use crate::LlmResponseChunk; -use crate::codecs::common::{collect_responses_reasoning_text, encrypted_reasoning_data}; +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, @@ -321,6 +323,21 @@ fn encode_responses_stream( if let Some(data) = encrypted_reasoning_data(&details) { state.response_reasoning_encrypted = Some(data); } + if let Some(id) = encrypted_reasoning_item_id(&details) { + if state.response_reasoning_started + && state.response_reasoning_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" + ); + state.response_reasoning_encrypted = None; + } else { + state.response_reasoning_item_id = Some(id); + } + } let mut out = ensure_responses_reasoning_started(state); if !text.is_empty() { out.extend(encode_responses_reasoning_delta(state, text)); @@ -395,7 +412,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { { // 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_item_id(state, "rs", output_index); + let item_id = responses_reasoning_item_id(state, output_index); let mut summary = Vec::new(); if !state.response_reasoning_text.is_empty() { out.push(json!({ @@ -521,9 +538,18 @@ fn decode_responses_reasoning_item( && !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![json!({"type": "reasoning.encrypted", "data": data})], + details: vec![detail], text: String::new(), }); } @@ -705,6 +731,14 @@ fn encode_responses_text_delta(state: &mut StreamTranslationState, text: String) out } +// The reasoning item id: the provider's own id when encrypted reasoning binds to it, else synthesized. +fn responses_reasoning_item_id(state: &StreamTranslationState, output_index: usize) -> String { + state + .response_reasoning_item_id + .clone() + .unwrap_or_else(|| responses_item_id(state, "rs", output_index)) +} + // Opens the Responses reasoning output item once, emitting its `added` events. fn ensure_responses_reasoning_started(state: &mut StreamTranslationState) -> Vec { let mut out = ensure_responses_created(state); @@ -720,14 +754,14 @@ fn ensure_responses_reasoning_started(state: &mut StreamTranslationState) -> Vec "output_index": output_index, "item": { "type": "reasoning", - "id": responses_item_id(state, "rs", output_index), + "id": responses_reasoning_item_id(state, output_index), "status": "in_progress", "summary": [], }, })); out.push(json!({ "type": "response.reasoning_summary_part.added", - "item_id": responses_item_id(state, "rs", output_index), + "item_id": responses_reasoning_item_id(state, output_index), "output_index": output_index, "summary_index": 0, "part": {"type": "summary_text", "text": ""}, @@ -746,7 +780,7 @@ fn encode_responses_reasoning_delta( let output_index = state.response_reasoning_output_index.unwrap_or(0); out.push(json!({ "type": "response.reasoning_summary_text.delta", - "item_id": responses_item_id(state, "rs", output_index), + "item_id": responses_reasoning_item_id(state, output_index), "output_index": output_index, "summary_index": 0, "delta": text, diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index 56b5252a8..8b8cc89e1 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -68,6 +68,9 @@ pub struct StreamTranslationState { /// 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) response_reasoning_encrypted: 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) response_reasoning_item_id: Option, pub(crate) next_response_output_index: usize, pub(crate) response_sequence_number: u64, diff --git a/crates/switchyard-translation/tests/response_translation.rs b/crates/switchyard-translation/tests/response_translation.rs index 2fd6527cd..16b9efd27 100644 --- a/crates/switchyard-translation/tests/response_translation.rs +++ b/crates/switchyard-translation/tests/response_translation.rs @@ -778,5 +778,7 @@ fn responses_encrypted_reasoning_item_survives_buffered_round_trip() -> TestResu .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(()) } diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 3f921eb87..67d2e2c31 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1469,6 +1469,70 @@ fn responses_synthesized_item_ids_stay_within_the_openai_length_limit() -> TestR 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(()) +} + // Verifies a streamed token-limit stop terminates with response.incomplete. #[test] fn openai_chat_length_finish_translates_to_responses_incomplete_event() -> TestResult { @@ -1629,7 +1693,8 @@ fn responses_stream_decodes_encrypted_reasoning_item_into_details() -> TestResul index: 0, details: vec![json!({ "type": "reasoning.encrypted", - "data": "opaque-encrypted-reasoning" + "data": "opaque-encrypted-reasoning", + "id": "rs_upstream" })], text: String::new(), }] From 0de2d5537596c8e52cfcdf3bb63819f2eb2ef37b Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Tue, 8 Sep 2026 12:28:49 -0700 Subject: [PATCH 09/13] fix(translation): accept a verbatim reasoning item as an encrypted reasoning detail The buffered request decoder in #645 keeps a provider's reasoning item whole as the reasoning detail when it carries encrypted_content. The stream and buffered response encoders here read the payload and item id through the shared helpers, so those helpers now recognise that shape alongside the documented reasoning.encrypted object. This keeps the buffered-decode, re-stream path (used by any route that buffers a reply) carrying the encrypted payload under its original id regardless of which PR lands first. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Lin Jia --- .../src/codecs/common.rs | 51 +++++++++++++++++-- 1 file changed, 47 insertions(+), 4 deletions(-) diff --git a/crates/switchyard-translation/src/codecs/common.rs b/crates/switchyard-translation/src/codecs/common.rs index b8771c9f8..1f3a31c8c 100644 --- a/crates/switchyard-translation/src/codecs/common.rs +++ b/crates/switchyard-translation/src/codecs/common.rs @@ -89,13 +89,20 @@ pub(crate) fn collect_responses_reasoning_text(value: Option<&Value>, out: &mut } } -/// Returns the opaque payload of the first `reasoning.encrypted` detail, if any. +/// 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(|detail| detail.get("type").and_then(Value::as_str) == Some("reasoning.encrypted")) - .and_then(|detail| detail.get("data").and_then(Value::as_str)) + .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) } @@ -106,7 +113,12 @@ pub(crate) fn encrypted_reasoning_item_id(details: &[Value]) -> Option { details .iter() .filter_map(Value::as_object) - .find(|detail| detail.get("type").and_then(Value::as_str) == Some("reasoning.encrypted")) + .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) @@ -138,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); + } +} From 2ce404f6f9565bba1092d5cd25b3a249d614503c Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Tue, 8 Sep 2026 22:17:54 -0700 Subject: [PATCH 10/13] fix(translation): keep every reasoning item of a streamed Responses response GPT-5 models emit a reasoning item ahead of each tool call, so one response can carry several reasoning items. The Responses stream encoder kept a single reasoning slot: the second item never opened, and when its encrypted payload arrived under a different provider id the encoder dropped it with a warning (observed on 1.5 percent of GPT-5.6 turns behind Switchyard). The same slot also opened under a synthesized id whenever summary text streamed before the payload, which made the payload unverifiable and dropped it too. The encoder now tracks reasoning items per source content index and emits each as its own output item, closing them in provider order. The Responses stream decoder announces a reasoning item's provider id as soon as the added event names it, so the encoder opens the item under that id before any text or payload arrives. The response accumulator folds the announcement into the payload detail that follows it and drops announcements whose payload never came, so replayed history keeps one detail per item; the chat encoder skips announcements since a chat client cannot use them. Tests cover a two-reasoning-item response, summary text arriving before the payload, and the decoder announcing an id exactly once across added, done and completed events. Signed-off-by: Lin Jia --- crates/protocol/src/stream.rs | 34 +++- .../src/codecs/openai_chat/stream.rs | 12 ++ .../src/codecs/responses/stream.rs | 169 +++++++++++------- .../src/codecs/stream.rs | 30 +++- .../tests/stream_translation.rs | 157 ++++++++++++++-- 5 files changed, 321 insertions(+), 81 deletions(-) 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/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/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index f873af383..d3a6ea1e7 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -314,33 +314,39 @@ 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 { details, 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. - if let Some(data) = encrypted_reasoning_data(&details) { - state.response_reasoning_encrypted = Some(data); - } - if let Some(id) = encrypted_reasoning_item_id(&details) { - if state.response_reasoning_started - && state.response_reasoning_item_id.as_deref() != Some(id.as_str()) - { + 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" ); - state.response_reasoning_encrypted = None; - } else { - state.response_reasoning_item_id = Some(id); + } + id => { + if id.is_some() { + item.item_id = id; + } + if data.is_some() { + item.encrypted = data; + } } } - let mut out = ensure_responses_reasoning_started(state); + let mut out = ensure_responses_reasoning_started(state, index); if !text.is_empty() { - out.extend(encode_responses_reasoning_delta(state, text)); + out.extend(encode_responses_reasoning_delta(state, index, text)); } out } @@ -407,31 +413,37 @@ 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 - { + 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, output_index); + let item_id = responses_reasoning_item_id(state, index); + let reasoning = &state.response_reasoning[&index]; let mut summary = Vec::new(); - if !state.response_reasoning_text.is_empty() { + 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": state.response_reasoning_text, + "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": state.response_reasoning_text}, + "part": {"type": "summary_text", "text": reasoning.text}, })); summary.push(json!({ "type": "summary_text", - "text": state.response_reasoning_text, + "text": reasoning.text, })); } let mut item = json!({ @@ -440,7 +452,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { "status": "completed", "summary": summary, }); - if let Some(encrypted) = &state.response_reasoning_encrypted { + if let Some(encrypted) = &reasoning.encrypted { item["encrypted_content"] = Value::String(encrypted.clone()); } out.push(json!({ @@ -517,6 +529,28 @@ fn decode_responses_reasoning_item( 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); @@ -731,56 +765,71 @@ fn encode_responses_text_delta(state: &mut StreamTranslationState, text: String) out } -// The reasoning item id: the provider's own id when encrypted reasoning binds to it, else synthesized. -fn responses_reasoning_item_id(state: &StreamTranslationState, output_index: usize) -> String { - state - .response_reasoning_item_id - .clone() - .unwrap_or_else(|| responses_item_id(state, "rs", output_index)) +// 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 once, emitting its `added` events. -fn ensure_responses_reasoning_started(state: &mut StreamTranslationState) -> Vec { +// 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_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); - // 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": responses_reasoning_item_id(state, output_index), - "status": "in_progress", - "summary": [], - }, - })); - out.push(json!({ - "type": "response.reasoning_summary_part.added", - "item_id": responses_reasoning_item_id(state, output_index), - "output_index": output_index, - "summary_index": 0, - "part": {"type": "summary_text", "text": ""}, - })); + 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_reasoning_started(state); - state.response_reasoning_text.push_str(&text); - let output_index = state.response_reasoning_output_index.unwrap_or(0); + 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_summary_text.delta", - "item_id": responses_reasoning_item_id(state, output_index), + "item_id": responses_reasoning_item_id(state, index), "output_index": output_index, "summary_index": 0, "delta": text, diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index 8b8cc89e1..08ee74fad 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -62,15 +62,12 @@ pub struct StreamTranslationState { 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, - /// 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) response_reasoning_encrypted: 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) response_reasoning_item_id: Option, + /// 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, @@ -78,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/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 67d2e2c31..f7d886e7e 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1779,10 +1779,17 @@ fn responses_stream_decodes_text_only_reasoning_item_from_done() -> TestResult { assert_eq!( decoded.normalized(), - &[LlmResponseChunk::ReasoningDelta { - index: 0, - text: "Let me explore the repo first.".to_string(), - }] + &[ + 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(()) } @@ -1814,7 +1821,15 @@ fn responses_stream_does_not_duplicate_streamed_reasoning_on_done() -> TestResul let second = engine.decode_stream_event(&mut state, format, done)?; assert_eq!(first.normalized().len(), 1); - assert_eq!(second.normalized(), &[]); + // 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(()) } @@ -1833,10 +1848,17 @@ fn responses_stream_decodes_reasoning_text_from_added_done_and_completed_once() let decoded = engine.decode_stream_event(&mut state, format, added)?; assert_eq!( decoded.normalized(), - &[LlmResponseChunk::ReasoningDelta { - index: 0, - text: "from added".into() - }] + &[ + 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, @@ -1875,12 +1897,13 @@ fn responses_stream_decodes_reasoning_text_from_added_done_and_completed_once() assert_eq!(text, "from completed"); "reasoning" } + LlmResponseChunk::ReasoningDetailsDelta { .. } => "id", LlmResponseChunk::Usage(_) => "usage", LlmResponseChunk::MessageStop { .. } => "stop", _ => "other", }) .collect(); - assert_eq!(kinds, vec!["reasoning", "usage", "stop"]); + assert_eq!(kinds, vec!["id", "reasoning", "usage", "stop"]); // (d) completed output repeating already-streamed reasoning adds nothing let mut state = StreamTranslationState::new(format, format); @@ -2057,3 +2080,117 @@ fn responses_stream_synthesized_item_ids_are_unique_across_responses() -> TestRe } 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(()) +} From c257a68c90ebd166d6cd49dcfadfe3ed63ad58ad Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Tue, 8 Sep 2026 10:58:19 -0700 Subject: [PATCH 11/13] feat(translation): round-trip Responses freeform custom tools Codex drives GPT-5 models with freeform tools: the definition is {"type": "custom", ...} and the model answers with custom_tool_call items whose input is a raw string. The Responses codec only modelled function tools, so a Codex session against a GPT-5 model through Switchyard lost its tool definitions and its tool calls and ended after one turn. Custom tools now pass through the IR as a function with a single input argument, with the verbatim definitions kept on the request extensions. History items custom_tool_call and custom_tool_call_output decode and re-encode with their types intact, upstream custom_tool_call output items decode on both the buffered and stream paths, and when a response is encoded with the request's extensions, calls to a custom tool are rewritten back into custom_tool_call items. Argument delta events for such calls are dropped on the stream because a partial JSON delta has no freeform equivalent; clients read the completed item. Signed-off-by: Lin Jia Co-Authored-By: Claude Fable 5.1 --- .../src/codecs/responses/buffered.rs | 137 ++++++++++- .../src/codecs/responses/stream.rs | 32 ++- .../src/codex_custom_tools.rs | 222 ++++++++++++++++++ crates/switchyard-translation/src/engine.rs | 4 + crates/switchyard-translation/src/helpers.rs | 23 +- crates/switchyard-translation/src/lib.rs | 1 + .../tests/request_translation.rs | 79 +++++++ .../tests/response_translation.rs | 73 ++++++ 8 files changed, 552 insertions(+), 19 deletions(-) create mode 100644 crates/switchyard-translation/src/codex_custom_tools.rs diff --git a/crates/switchyard-translation/src/codecs/responses/buffered.rs b/crates/switchyard-translation/src/codecs/responses/buffered.rs index d3346adad..723444ce6 100644 --- a/crates/switchyard-translation/src/codecs/responses/buffered.rs +++ b/crates/switchyard-translation/src/codecs/responses/buffered.rs @@ -91,7 +91,9 @@ 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); request.tool_choice = body .get("tool_choice") .and_then(decode_responses_tool_choice); @@ -112,6 +114,7 @@ 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); Ok(DecodedRequest { request, diagnostics, @@ -157,6 +160,7 @@ 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() { @@ -165,6 +169,7 @@ impl FormatCodec for OpenAiResponsesCodec { encode_responses_tools( &request.tools, crate::codex_namespaces::tool_namespaces(&request.extensions), + crate::codex_custom_tools::custom_tools(&request.extensions), ), ); } @@ -475,7 +480,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) @@ -488,6 +493,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"), @@ -789,6 +833,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(); @@ -805,7 +850,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); @@ -821,6 +866,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) @@ -1011,7 +1076,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 @@ -1046,17 +1113,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 { @@ -1079,6 +1154,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 { @@ -1090,6 +1167,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. @@ -1112,7 +1199,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, " "), })), @@ -1208,10 +1299,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, @@ -1309,6 +1406,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), diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index d3a6ea1e7..3b2920789 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -604,14 +604,20 @@ fn decode_responses_output_item_added( if item.get("type").and_then(Value::as_str) == Some("reasoning") { return decode_responses_reasoning_item(item, index, state); } - if item.get("type").and_then(Value::as_str) != Some("function_call") { + 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 arguments_delta = item - .get("arguments") - .and_then(Value::as_str) - .filter(|arguments| !arguments.is_empty()) - .map(ToOwned::to_owned); + // 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 @@ -650,10 +656,20 @@ fn decode_responses_output_item_done( if item.get("type").and_then(Value::as_str) == Some("reasoning") { return decode_responses_reasoning_item(item, index, state); } - if item.get("type").and_then(Value::as_str) != Some("function_call") { + 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 arguments = item.get("arguments").and_then(Value::as_str); + 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 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..2ae4464b9 --- /dev/null +++ b/crates/switchyard-translation/src/codex_custom_tools.rs @@ -0,0 +1,222 @@ +// 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"; + +/// 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)); + 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 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..90ef12fb4 100644 --- a/crates/switchyard-translation/tests/request_translation.rs +++ b/crates/switchyard-translation/tests/request_translation.rs @@ -2389,3 +2389,82 @@ 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(()) +} diff --git a/crates/switchyard-translation/tests/response_translation.rs b/crates/switchyard-translation/tests/response_translation.rs index 16b9efd27..e68a1ad61 100644 --- a/crates/switchyard-translation/tests/response_translation.rs +++ b/crates/switchyard-translation/tests/response_translation.rs @@ -782,3 +782,76 @@ fn responses_encrypted_reasoning_item_survives_buffered_round_trip() -> TestResu 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(()) +} From 03b739f6247729ba7f16b71eca470d9c3c07a857 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Tue, 8 Sep 2026 12:52:18 -0700 Subject: [PATCH 12/13] feat(translation): understand Responses-lite additional_tools input items Codex sends GPT-5 requests in a 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. The Responses codec did not know the item, so a routed GPT-5 session had no tools in the IR and the item was turned into a user message carrying the tool JSON. The request decoder now reads the item's tools as the request's tool definitions (including freeform tools) and keeps the array verbatim on the request extensions; the input decoder skips the item; the request encoder re-emits it in place and leaves top-level tools absent, so a Responses upstream receives the request in the shape the client used, while a chat upstream receives ordinary function tools. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Lin Jia --- .../src/codecs/responses/buffered.rs | 42 ++++++- .../src/codex_custom_tools.rs | 25 +++++ .../tests/request_translation.rs | 104 ++++++++++++++++++ 3 files changed, 170 insertions(+), 1 deletion(-) diff --git a/crates/switchyard-translation/src/codecs/responses/buffered.rs b/crates/switchyard-translation/src/codecs/responses/buffered.rs index 723444ce6..fddfcf380 100644 --- a/crates/switchyard-translation/src/codecs/responses/buffered.rs +++ b/crates/switchyard-translation/src/codecs/responses/buffered.rs @@ -94,6 +94,26 @@ impl FormatCodec for OpenAiResponsesCodec { 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); @@ -115,6 +135,10 @@ 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, @@ -163,7 +187,20 @@ impl FormatCodec for OpenAiResponsesCodec { &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( @@ -539,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, diff --git a/crates/switchyard-translation/src/codex_custom_tools.rs b/crates/switchyard-translation/src/codex_custom_tools.rs index 2ae4464b9..ea93c2649 100644 --- a/crates/switchyard-translation/src/codex_custom_tools.rs +++ b/crates/switchyard-translation/src/codex_custom_tools.rs @@ -21,6 +21,31 @@ use switchyard_protocol::ProviderExtensions; /// 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"; diff --git a/crates/switchyard-translation/tests/request_translation.rs b/crates/switchyard-translation/tests/request_translation.rs index 90ef12fb4..35bb1d831 100644 --- a/crates/switchyard-translation/tests/request_translation.rs +++ b/crates/switchyard-translation/tests/request_translation.rs @@ -2468,3 +2468,107 @@ fn responses_request_round_trips_custom_tools_and_custom_tool_calls() -> TestRes ); 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(()) +} From 9e4a8a31e56860aa297b74528bac53ec173a2ef1 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Tue, 8 Sep 2026 13:29:08 -0700 Subject: [PATCH 13/13] fix(translation): give rewritten custom tool calls a ctc item id prefix OpenAI validates replayed item ids by prefix and rejects a custom_tool_call whose id starts with fc_ ("Expected an ID that begins with 'ctc'"). When a function_call item is rewritten into a custom_tool_call for the client, its synthesized id now takes the ctc_ prefix. Co-Authored-By: Claude Fable 5.1 Signed-off-by: Lin Jia --- .../src/codex_custom_tools.rs | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/crates/switchyard-translation/src/codex_custom_tools.rs b/crates/switchyard-translation/src/codex_custom_tools.rs index ea93c2649..7d15d56bc 100644 --- a/crates/switchyard-translation/src/codex_custom_tools.rs +++ b/crates/switchyard-translation/src/codex_custom_tools.rs @@ -127,6 +127,12 @@ fn rewrite_item(item: &mut Value, custom: &HashSet) -> bool { 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 } @@ -219,6 +225,17 @@ mod tests { 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();