diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index b076e3300..856057d2b 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -101,7 +101,7 @@ fn decode_responses_stream( }) .unwrap_or_default() } - Some("response.output_item.added") => decode_responses_output_item_added(event), + Some("response.output_item.added") => decode_responses_output_item_added(event, state), Some("response.function_call_arguments.delta") => { let output_index = event .get("output_index") @@ -111,6 +111,14 @@ fn decode_responses_stream( .get("delta") .and_then(Value::as_str) .map(|delta| { + // Recorded so `response.output_item.done`, which repeats + // the complete arguments, can tell it is a repeat. + state + .tool_states + .entry(output_index as usize) + .or_default() + .decoded_arguments + .push_str(delta); vec![LlmResponseChunk::ToolCallDelta { index: output_index as usize, id: None, @@ -325,7 +333,10 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { } // Converts Responses function-call item creation into a neutral tool-call delta. -fn decode_responses_output_item_added(event: &Value) -> Vec { +fn decode_responses_output_item_added( + event: &Value, + state: &mut StreamTranslationState, +) -> Vec { let Some(item) = event.get("item").and_then(Value::as_object) else { return Vec::new(); }; @@ -336,6 +347,19 @@ fn decode_responses_output_item_added(event: &Value) -> Vec { .get("output_index") .and_then(Value::as_u64) .unwrap_or(0) as usize; + let arguments_delta = item + .get("arguments") + .and_then(Value::as_str) + .filter(|arguments| !arguments.is_empty()) + .map(ToOwned::to_owned); + if let Some(arguments) = arguments_delta.as_deref() { + state + .tool_states + .entry(index) + .or_default() + .decoded_arguments + .push_str(arguments); + } vec![LlmResponseChunk::ToolCallDelta { index, id: item @@ -347,18 +371,14 @@ fn decode_responses_output_item_added(event: &Value) -> Vec { .get("name") .and_then(Value::as_str) .map(ToOwned::to_owned), - arguments_delta: item - .get("arguments") - .and_then(Value::as_str) - .filter(|arguments| !arguments.is_empty()) - .map(ToOwned::to_owned), + arguments_delta, }] } // Emits a final tool-call argument delta when Responses only supplies arguments at item end. fn decode_responses_output_item_done( event: &Value, - state: &StreamTranslationState, + state: &mut StreamTranslationState, ) -> Vec { let Some(item) = event.get("item").and_then(Value::as_object) else { return Vec::new(); @@ -372,12 +392,13 @@ fn decode_responses_output_item_done( .unwrap_or(0) as usize; let arguments = item.get("arguments").and_then(Value::as_str); if let Some(arguments) = arguments { - let existing = state - .tool_states - .get(&index) - .map(|tool| tool.arguments.as_str()) - .unwrap_or(""); - if !arguments.is_empty() && arguments != existing { + // Compared against what THIS decoder has seen. Reading the encoder's + // `arguments` instead only deduplicates when a single state performs + // both halves of the translation, and silently duplicates when a + // caller buffers the stream with its own state. + let tool = state.tool_states.entry(index).or_default(); + if !arguments.is_empty() && arguments != tool.decoded_arguments { + tool.decoded_arguments.push_str(arguments); return vec![LlmResponseChunk::ToolCallDelta { index, id: None, diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index bcd586a27..7bd2a8f87 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -69,6 +69,14 @@ pub(crate) struct StreamToolState { pub(crate) id: Option, pub(crate) name: Option, pub(crate) arguments: String, + /// Arguments observed while DECODING the source stream. + /// + /// Separate from `arguments`, which encoders accumulate. A decoder that + /// deduplicates against `arguments` only works when one state performs + /// both halves of the translation; when a caller buffers a stream with its + /// own state and encodes later, the field is empty and the duplicate is + /// emitted. + pub(crate) decoded_arguments: String, pub(crate) pending_arguments: String, pub(crate) started: bool, pub(crate) content_index: Option, diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 98cf5f1ce..3190e57bb 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1198,3 +1198,39 @@ fn responses_incomplete_event_translates_to_chat_length_finish() -> TestResult { assert_eq!(terminal["choices"][0]["finish_reason"], "length"); Ok(()) } + +// A completion event must not repeat function-call arguments from delta events. +#[test] +fn responses_decode_emits_tool_arguments_once() -> TestResult { + let engine = TranslationEngine::default(); + let mut state = StreamTranslationState::default(); + let arguments = r#"{"skill":"demo:thing","args":{}}"#; + + let upstream = [ + json!({"type": "response.output_item.added", "output_index": 0, + "item": {"type": "function_call", "call_id": "call_1", + "name": "Skill", "arguments": ""}}), + json!({"type": "response.function_call_arguments.delta", + "output_index": 0, "delta": arguments}), + json!({"type": "response.output_item.done", "output_index": 0, + "item": {"type": "function_call", "call_id": "call_1", + "name": "Skill", "arguments": arguments}}), + ]; + + let mut seen = String::new(); + for event in upstream { + let decoded = engine.decode_stream_event(&mut state, WireFormat::OpenAiResponses, event)?; + for chunk in decoded.normalized() { + if let LlmResponseChunk::ToolCallDelta { + arguments_delta: Some(delta), + .. + } = chunk + { + seen.push_str(delta); + } + } + } + + assert_eq!(seen, arguments); + Ok(()) +}