From 24b82ac695abc04a7f62404ed12ffcdc754a3f79 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Sat, 5 Sep 2026 11:38:18 -0700 Subject: [PATCH 01/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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/11] 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 8d13ce1bb9d13c3727ea54fcbe7f90c426326fe2 Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Wed, 9 Sep 2026 10:41:44 -0700 Subject: [PATCH 11/11] chore(translation): changelog entries and review tidy-ups for the Responses codec fixes Adds the changelog entries for the Responses reasoning, item-id, and raw trace changes, removes a comment left behind when a helper moved to the shared codec module, and lists the new tracing dependency with the rest of the crate's dependencies. Signed-off-by: Lin Jia --- CHANGELOG.md | 14 ++++++++++++++ crates/switchyard-translation/Cargo.toml | 2 +- .../src/codecs/responses/buffered.rs | 2 -- 3 files changed, 15 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 76884e283..025506956 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,9 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ### Added +- **Raw Responses stream trace** — an opt-in trace of every upstream Responses + event as received, under `RUST_LOG=switchyard_translation::responses::raw=trace`, + for diagnosing provider-specific event shapes. (#646) - **NeMo Relay native plugin** — a dynamically loaded integration that loads Switchyard's standard TOML deployment and executes its `switchyard-runner`- supported configured routes in process. Managed calls require NeMo Relay @@ -99,6 +102,17 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ### Fixed +- **Responses reasoning through transforming routes** — reasoning that a route + buffers or re-encodes now reaches the client in the standard `summary_text` + shape with `reasoning_summary_*` events, encrypted-only and done-only + reasoning items are decoded from every carrier a provider uses, and encrypted + payloads are re-emitted under the provider's item id so the client's replay + verifies upstream. Responses with several reasoning items keep all of them. + (#646) +- **Unique, bounded Responses item ids** — synthesized output-item ids carry a + per-response discriminator so replayed history no longer repeats `rs_0` and + `fc_1` across turns, and upstream response ids longer than 40 characters are + digested to stay within OpenAI's 64-character item-id limit. (#646) - **Reasoning order in mixed stream chunks** — the OpenAI Chat stream decoder emits reasoning deltas before content deltas from the same chunk, so interleaved reasoning is no longer reordered. (#387) diff --git a/crates/switchyard-translation/Cargo.toml b/crates/switchyard-translation/Cargo.toml index 3ba53a4f0..bb121d931 100644 --- a/crates/switchyard-translation/Cargo.toml +++ b/crates/switchyard-translation/Cargo.toml @@ -17,7 +17,6 @@ keywords = ["llm", "translation", "openai", "anthropic"] publish = ["crates-io"] [dependencies] -tracing.workspace = true base64.workspace = true serde.workspace = true serde_json.workspace = true @@ -25,6 +24,7 @@ switchyard-protocol.workspace = true thiserror.workspace = true futures.workspace = true async-stream.workspace = true +tracing.workspace = true [dev-dependencies] pretty_assertions = "1" diff --git a/crates/switchyard-translation/src/codecs/responses/buffered.rs b/crates/switchyard-translation/src/codecs/responses/buffered.rs index 97321c21e..10fab3684 100644 --- a/crates/switchyard-translation/src/codecs/responses/buffered.rs +++ b/crates/switchyard-translation/src/codecs/responses/buffered.rs @@ -668,8 +668,6 @@ fn decode_responses_reasoning_item(item: &Map) -> Vec Vec { match value {