From 3e304ef02266519199bf42c30f9094950072d97e Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Mon, 14 Sep 2026 15:22:01 -0700 Subject: [PATCH 1/2] fix(translation): report tool_use when a streamed Responses turn ends in a function call Signed-off-by: Lin Jia --- CHANGELOG.md | 9 ++ .../src/codecs/responses/stream.rs | 24 ++- .../src/codecs/stream.rs | 3 + .../tests/stream_translation.rs | 147 ++++++++++++++++++ 4 files changed, 182 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3d5be1861..ad606717f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -126,6 +126,15 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). refunded the review budget but left the conversation's stall latch set, so every later eligible turn silently bypassed the advisor. The latch now clears whenever the reserved review is refunded or the budget is already spent. +- **Streamed Responses tool calls end with a tool-use stop reason** — the + Responses stream decoder reported every `response.completed` as a plain + completion, so a streamed `function_call` reached Anthropic clients as + `stop_reason: "end_turn"` and Chat clients as `finish_reason: "stop"`. + Stop-reason-driven tool loops, including the official Anthropic TypeScript + SDK tool runner, then returned the unfinished tool-use turn without running + the tool. The decoder now reports `tool_use` when the completed output holds + a `function_call` or `custom_tool_call`, or when it already decoded tool + deltas, matching the buffered decoder. - **Encrypted-only reasoning items open no summary part** — the Responses stream encoder opened a `reasoning_summary_part` for every reasoning item and closed it only when text had streamed, so an encrypted-only item left a part diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index db5da15de..df507ecb4 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -177,6 +177,7 @@ fn decode_responses_stream( .get("delta") .and_then(Value::as_str) .map(|delta| { + state.decoded_tool_call = true; // Recorded so `response.output_item.done`, which repeats // the complete arguments, can tell it is a repeat. state @@ -272,7 +273,18 @@ fn decode_responses_stream( state.saw_backend_usage = true; out.push(LlmResponseChunk::Usage(usage)); } - out.push(LlmResponseChunk::MessageStop { reason: None }); + // A completed response that produced a tool call ended the turn to run that + // tool, not because the assistant was done. The buffered decoder reports tool + // use for such output; the stream must too, or stop-reason-driven tool loops + // (Anthropic `tool_use`, Chat `tool_calls`) stop without running the tool. + // Carries the Anthropic spelling because every encoder already maps it. + let has_tool_call = event + .get("response") + .and_then(|response| response.get("output")) + .and_then(Value::as_array) + .is_some_and(|items| items.iter().any(is_responses_tool_call_item)); + let reason = (has_tool_call || state.decoded_tool_call).then(|| "tool_use".to_string()); + out.push(LlmResponseChunk::MessageStop { reason }); out } // Carries the Anthropic spelling because every encoder already maps it. @@ -608,6 +620,7 @@ fn decode_responses_output_item_added( if item_type != Some("function_call") && item_type != Some("custom_tool_call") { return Vec::new(); } + state.decoded_tool_call = true; // A freeform call's `input` becomes the single `input` argument; it is only complete on // the done event, so nothing is emitted for it here beyond id and name. let arguments_delta = if item_type == Some("custom_tool_call") { @@ -641,6 +654,14 @@ fn decode_responses_output_item_added( }] } +// Whether a Responses output item is a client tool call the consumer must run. +fn is_responses_tool_call_item(item: &Value) -> bool { + matches!( + item.get("type").and_then(Value::as_str), + Some("function_call" | "custom_tool_call") + ) +} + // Emits a final tool-call argument delta when Responses only supplies arguments at item end. fn decode_responses_output_item_done( event: &Value, @@ -660,6 +681,7 @@ fn decode_responses_output_item_done( if item_type != Some("function_call") && item_type != Some("custom_tool_call") { return Vec::new(); } + state.decoded_tool_call = true; let custom_arguments = (item_type == Some("custom_tool_call")).then(|| { json!({ crate::codex_custom_tools::INPUT_ARGUMENT: diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index 5a579b11c..49bfdd372 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -57,6 +57,9 @@ pub struct StreamTranslationState { pub(crate) decoded_reasoning: BTreeMap, /// Output indexes whose encrypted reasoning payload was already decoded. pub(crate) decoded_reasoning_encrypted: std::collections::BTreeSet, + /// Set once a tool call was observed while DECODING, so a terminal event that names no + /// stop reason can still report tool use. + pub(crate) decoded_tool_call: bool, 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 035abd837..300fb58c5 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1285,6 +1285,153 @@ fn responses_buffered_and_streamed_outputs_match() -> TestResult { Ok(()) } +// The Responses events a provider streams for one completed function call, ending with a +// `response.completed` whose output repeats the finished call. +fn responses_function_call_stream() -> Vec { + let call = json!({ + "id": "fc_1", + "type": "function_call", + "status": "completed", + "call_id": "call_1", + "name": "get_weather", + "arguments": "{\"city\":\"Paris\"}" + }); + vec![ + json!({ + "type": "response.created", + "response": {"id": "resp_1", "model": "gpt-5.6", "status": "in_progress", "output": []} + }), + json!({ + "type": "response.output_item.added", + "output_index": 0, + "item": { + "id": "fc_1", + "type": "function_call", + "status": "in_progress", + "call_id": "call_1", + "name": "get_weather", + "arguments": "" + } + }), + json!({ + "type": "response.function_call_arguments.delta", + "item_id": "fc_1", + "output_index": 0, + "delta": "{\"city\":\"Paris\"}" + }), + json!({ + "type": "response.function_call_arguments.done", + "item_id": "fc_1", + "output_index": 0, + "name": "get_weather", + "arguments": "{\"city\":\"Paris\"}" + }), + json!({"type": "response.output_item.done", "output_index": 0, "item": call}), + json!({ + "type": "response.completed", + "response": {"id": "resp_1", "model": "gpt-5.6", "status": "completed", "output": [call]} + }), + ] +} + +// Translates a whole source stream into `target` events, including the encoder's finish. +fn translate_stream( + engine: &TranslationEngine, + source: WireFormat, + target: WireFormat, + events: &[Value], +) -> std::result::Result, Box> { + let mut state = StreamTranslationState::new(source, target); + let mut out = Vec::new(); + for event in events { + out.extend(engine.translate_event(&mut state, source, target, event)?); + } + out.extend(engine.finish_stream(&mut state, target)?); + Ok(out) +} + +// A streamed Responses function call must end the Anthropic message with `tool_use`, the same +// stop reason the buffered decoder reports. Anthropic's tool runners dispatch on it; `end_turn` +// makes them return the unfinished tool-use turn without running the tool. +#[test] +fn responses_function_call_stream_stops_with_tool_use_for_anthropic() -> TestResult { + let events = translate_stream( + &TranslationEngine::default(), + WireFormat::OpenAiResponses, + WireFormat::AnthropicMessages, + &responses_function_call_stream(), + )?; + + let stop_reasons: Vec<&Value> = events + .iter() + .filter(|event| event["type"] == "message_delta") + .map(|event| &event["delta"]["stop_reason"]) + .collect(); + assert_eq!(stop_reasons, vec![&json!("tool_use")]); + assert!(events.iter().any(|event| { + event["type"] == "content_block_start" + && event["content_block"]["type"] == "tool_use" + && event["content_block"]["name"] == "get_weather" + })); + Ok(()) +} + +// The same stream on the Chat wire must finish with `tool_calls`, not `stop`. +#[test] +fn responses_function_call_stream_finishes_with_tool_calls_for_openai_chat() -> TestResult { + let events = translate_stream( + &TranslationEngine::default(), + WireFormat::OpenAiResponses, + WireFormat::OpenAiChat, + &responses_function_call_stream(), + )?; + + let finish_reasons: Vec<&Value> = events + .iter() + .filter_map(|event| event["choices"][0].get("finish_reason")) + .filter(|reason| !reason.is_null()) + .collect(); + assert_eq!(finish_reasons, vec![&json!("tool_calls")]); + assert!(events.iter().any(|event| { + event["choices"][0]["delta"]["tool_calls"][0]["function"]["name"] == "get_weather" + })); + Ok(()) +} + +// Some providers end a stream with a bare `response.completed`. The tool deltas already +// decoded are enough to report tool use; a text-only stream still reports no reason. +#[test] +fn responses_bare_completed_reports_tool_use_only_after_tool_deltas() -> TestResult { + let bare_completed = json!({"type": "response.completed", "response": {}}); + + let mut state = + StreamTranslationState::new(WireFormat::OpenAiResponses, WireFormat::OpenAiResponses); + let mut last = Vec::new(); + for event in responses_function_call_stream() + .iter() + .take(5) + .chain(std::iter::once(&bare_completed)) + { + last = decode_stream_event(&mut state, WireFormat::OpenAiResponses, event); + } + assert_eq!( + last, + vec![LlmResponseChunk::MessageStop { + reason: Some("tool_use".to_string()) + }] + ); + + let mut state = + StreamTranslationState::new(WireFormat::OpenAiResponses, WireFormat::OpenAiResponses); + let text = json!({"type": "response.output_text.delta", "output_index": 0, "delta": "hi"}); + decode_stream_event(&mut state, WireFormat::OpenAiResponses, &text); + assert_eq!( + decode_stream_event(&mut state, WireFormat::OpenAiResponses, &bare_completed), + vec![LlmResponseChunk::MessageStop { reason: None }] + ); + Ok(()) +} + // An OpenAI-shaped error frame carries no `choices`, so it must decode to a stream error // instead of a bare message start that silently drops the upstream message. #[test] From 831ef74529436600e3427150a56e2cd386e15edb Mon Sep 17 00:00:00 2001 From: Lin Jia Date: Mon, 14 Sep 2026 16:31:30 -0700 Subject: [PATCH 2/2] test(translation): fold the streamed tool-use stop reason checks into one test Signed-off-by: Lin Jia --- .../tests/stream_translation.rs | 55 ++++++++----------- 1 file changed, 22 insertions(+), 33 deletions(-) diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 300fb58c5..709707d8b 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -1350,68 +1350,57 @@ fn translate_stream( Ok(out) } -// A streamed Responses function call must end the Anthropic message with `tool_use`, the same -// stop reason the buffered decoder reports. Anthropic's tool runners dispatch on it; `end_turn` -// makes them return the unfinished tool-use turn without running the tool. +// A streamed Responses function call must end with the tool-use stop reason the buffered decoder +// reports: Anthropic `tool_use`, Chat `tool_calls`. Anthropic's tool runners dispatch on it; +// `end_turn` makes them return the unfinished tool-use turn without running the tool. A bare +// `response.completed` with no output array falls back to the argument deltas already decoded, +// while a text-only stream still reports no reason. #[test] -fn responses_function_call_stream_stops_with_tool_use_for_anthropic() -> TestResult { - let events = translate_stream( - &TranslationEngine::default(), +fn responses_function_call_stream_ends_with_tool_use_on_every_wire() -> TestResult { + let engine = TranslationEngine::default(); + let stream = responses_function_call_stream(); + + let anthropic = translate_stream( + &engine, WireFormat::OpenAiResponses, WireFormat::AnthropicMessages, - &responses_function_call_stream(), + &stream, )?; - - let stop_reasons: Vec<&Value> = events + let stop_reasons: Vec<&Value> = anthropic .iter() .filter(|event| event["type"] == "message_delta") .map(|event| &event["delta"]["stop_reason"]) .collect(); assert_eq!(stop_reasons, vec![&json!("tool_use")]); - assert!(events.iter().any(|event| { + assert!(anthropic.iter().any(|event| { event["type"] == "content_block_start" && event["content_block"]["type"] == "tool_use" && event["content_block"]["name"] == "get_weather" })); - Ok(()) -} -// The same stream on the Chat wire must finish with `tool_calls`, not `stop`. -#[test] -fn responses_function_call_stream_finishes_with_tool_calls_for_openai_chat() -> TestResult { - let events = translate_stream( - &TranslationEngine::default(), + let chat = translate_stream( + &engine, WireFormat::OpenAiResponses, WireFormat::OpenAiChat, - &responses_function_call_stream(), + &stream, )?; - - let finish_reasons: Vec<&Value> = events + let finish_reasons: Vec<&Value> = chat .iter() .filter_map(|event| event["choices"][0].get("finish_reason")) .filter(|reason| !reason.is_null()) .collect(); assert_eq!(finish_reasons, vec![&json!("tool_calls")]); - assert!(events.iter().any(|event| { + assert!(chat.iter().any(|event| { event["choices"][0]["delta"]["tool_calls"][0]["function"]["name"] == "get_weather" })); - Ok(()) -} -// Some providers end a stream with a bare `response.completed`. The tool deltas already -// decoded are enough to report tool use; a text-only stream still reports no reason. -#[test] -fn responses_bare_completed_reports_tool_use_only_after_tool_deltas() -> TestResult { + // Delta-only fallback: `response.created`, one argument delta, then a bare completion with + // no output-item events at all. let bare_completed = json!({"type": "response.completed", "response": {}}); - let mut state = StreamTranslationState::new(WireFormat::OpenAiResponses, WireFormat::OpenAiResponses); let mut last = Vec::new(); - for event in responses_function_call_stream() - .iter() - .take(5) - .chain(std::iter::once(&bare_completed)) - { + for event in [&stream[0], &stream[2], &bare_completed] { last = decode_stream_event(&mut state, WireFormat::OpenAiResponses, event); } assert_eq!(