From 846fb116ae6e982f507b009a3279bd84c99ffb05 Mon Sep 17 00:00:00 2001 From: Elyas Mehtabuddin Date: Mon, 12 Oct 2026 14:35:47 -0700 Subject: [PATCH 1/5] fix(translation): finish Codex tool calls and compaction through a LiteLLM gateway Signed-off-by: Elyas Mehtabuddin --- crates/libsy-llm-client/src/client.rs | 114 ++- crates/switchyard-server/src/sse.rs | 4 + .../tests/litellm_gateway.rs | 776 ++++++++++++++++++ .../src/codecs/responses/stream.rs | 35 +- .../src/codecs/stream.rs | 5 + .../tests/stream_translation.rs | 79 +- 6 files changed, 994 insertions(+), 19 deletions(-) create mode 100644 crates/switchyard-server/tests/litellm_gateway.rs diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index e1976de291..650f8d591f 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -4,7 +4,7 @@ //! [`TranslatingLlmClient`]: encode a neutral //! request, call the configured backend over HTTP, decode the neutral response. -use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::future::ready; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, SystemTime}; @@ -277,6 +277,9 @@ impl TranslatingLlmClient { if matches!(backend, Backend::OpenAiResponses(_)) { sanitize_openai_responses_provider_body(&mut body); } + if matches!(endpoint, UpstreamEndpoint::Completion) { + define_tools_for_replayed_calls(&mut body, wire_format); + } // Strip before `merge_extra_body` so a target can reinstate either field // deliberately via `extra_body`. if matches!(backend, Backend::Anthropic(_)) { @@ -1074,6 +1077,115 @@ fn ensure_responses_function_tool_description(object: &mut Map) { } } +/// Description of a tool definition that Switchyard adds only so that earlier calls to the +/// tool stay valid. The same request sets `tool_choice` to `none`. +const REPLAYED_TOOL_DESCRIPTION: &str = + "Earlier calls to this tool appear in the conversation. It is not available in this request."; + +// Adds a definition for each tool that the conversation history calls when the request +// defines no tools, and sets `tool_choice` to `none`. +// +// Codex compacts a long conversation by sending the whole history, including its tool calls +// and results, with an empty `tools` list. OpenAI accepts that. LiteLLM in front of Bedrock +// Claude rejects it on both OpenAI APIs ("Bedrock doesn't support tool calling without +// `tools=` param specified"), so compaction could never finish on such a target. A +// same-format request is replayed verbatim, so the fix has to apply to the outbound body. +// +// Bedrock has no `tool_choice` of `none`, and LiteLLM drops it. A Bedrock model can therefore +// still call one of these tools if the prompt asks it to. +fn define_tools_for_replayed_calls(body: &mut Value, wire_format: WireFormat) { + let Value::Object(object) = body else { + return; + }; + if object + .get("tools") + .and_then(Value::as_array) + .is_some_and(|tools| !tools.is_empty()) + { + return; + } + let tools = match wire_format { + WireFormat::OpenAiResponses => replayed_responses_tools(object.get("input")), + WireFormat::OpenAiChat => replayed_chat_tools(object.get("messages")), + WireFormat::AnthropicMessages => return, + }; + if tools.is_empty() { + return; + } + object.insert("tools".to_string(), Value::Array(tools)); + object.insert("tool_choice".to_string(), json!("none")); +} + +// One definition per tool name called in Responses `input`, in first-call order. A freeform +// `custom_tool_call` gets a `custom` definition and a `function_call` gets a `function` one. +fn replayed_responses_tools(input: Option<&Value>) -> Vec { + let Some(Value::Array(items)) = input else { + return Vec::new(); + }; + // A Responses-lite request defines its tools inside `input`. + if items + .iter() + .any(|item| item.get("type").and_then(Value::as_str) == Some("additional_tools")) + { + return Vec::new(); + } + let mut names = HashSet::new(); + let mut tools = Vec::new(); + for item in items { + let kind = match item.get("type").and_then(Value::as_str) { + Some("function_call") => "function", + Some("custom_tool_call") => "custom", + _ => continue, + }; + let Some(name) = item.get("name").and_then(Value::as_str) else { + continue; + }; + if !names.insert(name) { + continue; + } + tools.push(if kind == "function" { + json!({ + "type": "function", + "name": name, + "description": REPLAYED_TOOL_DESCRIPTION, + "parameters": {"type": "object"}, + }) + } else { + json!({"type": "custom", "name": name, "description": REPLAYED_TOOL_DESCRIPTION}) + }); + } + tools +} + +// One function definition per tool name called by assistant messages, in first-call order. +fn replayed_chat_tools(messages: Option<&Value>) -> Vec { + let Some(Value::Array(messages)) = messages else { + return Vec::new(); + }; + let mut names = HashSet::new(); + let mut tools = Vec::new(); + let calls = messages + .iter() + .filter_map(|message| message.get("tool_calls").and_then(Value::as_array)) + .flatten(); + for call in calls { + let Some(name) = call.pointer("/function/name").and_then(Value::as_str) else { + continue; + }; + if names.insert(name) { + tools.push(json!({ + "type": "function", + "function": { + "name": name, + "description": REPLAYED_TOOL_DESCRIPTION, + "parameters": {"type": "object"}, + }, + })); + } + } + tools +} + // Drops fields accepted by OpenAI-like APIs but rejected by Anthropic Messages. // // A router can serve earlier turns of a session from an OpenAI-format target and diff --git a/crates/switchyard-server/src/sse.rs b/crates/switchyard-server/src/sse.rs index dd2cb1595f..c6549eff16 100644 --- a/crates/switchyard-server/src/sse.rs +++ b/crates/switchyard-server/src/sse.rs @@ -39,6 +39,10 @@ pub(crate) fn frame_stream( // Preserve the upstream error's fields, apart from credential redaction, // rather than replacing it with a synthesized error. Err(LlmStreamError::Upstream(value)) => { + // The error message can quote request content, so the log records only + // the event type. + let event_type = value.get("type").and_then(Value::as_str).unwrap_or("error"); + tracing::warn!(event_type, "stream ended with an in-band error event"); failed = true; frame_event(target_format, value.clone(), &redactor).unwrap_or_else(|error| { tracing::warn!(error = %error, "in-band error event could not be framed"); diff --git a/crates/switchyard-server/tests/litellm_gateway.rs b/crates/switchyard-server/tests/litellm_gateway.rs new file mode 100644 index 0000000000..d32cf6c541 --- /dev/null +++ b/crates/switchyard-server/tests/litellm_gateway.rs @@ -0,0 +1,776 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Regression tests for Codex traffic sent through a LiteLLM gateway to Bedrock Claude. +//! +//! The stub gateway reproduces three behaviors of the real gateway that broke Codex: +//! it streams a freeform tool call as function-style argument deltas whose JSON differs from +//! the finished call, it stops at 4,096 output tokens when a request sets no limit, and it +//! rejects tool-call history in a request that defines no tools. Each test runs the +//! production server, config loader, router, and HTTP client against that stub. + +use std::convert::Infallible; +use std::error::Error; +use std::io::Write; +use std::sync::Arc; + +use axum::body::Body; +use axum::extract::State; +use axum::http::{Request, StatusCode, Uri}; +use axum::response::sse::{Event, Sse}; +use axum::response::{IntoResponse, Response}; +use axum::routing::post; +use axum::{Json, Router}; +use http_body_util::BodyExt; +use serde_json::{Value, json}; +use switchyard_server::{build_switchyard_router, config::load_server_state}; +use tokio::net::TcpListener; +use tokio::sync::Mutex; +use tokio::task::JoinHandle; +use tower::ServiceExt; + +type TestResult = Result>; + +/// Output tokens the gateway allows Bedrock Claude when a request sets no limit. +const GATEWAY_DEFAULT_OUTPUT_CAP: u64 = 4096; +/// Output limit that the `claude` target adds through `extra_body`. +const CONFIGURED_OUTPUT_LIMIT: u64 = 64000; +const LITELLM_TOOLS_ERROR: &str = "litellm.UnsupportedParamsError: Bedrock doesn't support tool calling without `tools=` param specified."; +const CONFLICT_MESSAGE: &str = "Responses snapshot conflicts with streamed content"; +/// Both route types send the requests in these tests to the `claude` target. +const CLAUDE_ROUTES: [&str; 2] = ["gateway/passthrough", "gateway/stage"]; + +// `Gateway::drop` stops the server task even when an assertion fails. +struct Gateway { + root: String, + calls: Arc>>, + task: JoinHandle>, +} + +impl Gateway { + async fn start() -> TestResult { + let calls = Arc::new(Mutex::new(Vec::new())); + let app = Router::new() + .route("/v1/responses", post(gateway)) + .route("/v1/chat/completions", post(gateway)) + .with_state(Arc::clone(&calls)); + let listener = TcpListener::bind("127.0.0.1:0").await?; + let root = format!("http://{}", listener.local_addr()?); + let task = tokio::spawn(async move { axum::serve(listener, app).await }); + Ok(Self { root, calls, task }) + } + + /// The single request the gateway received since the last call. + async fn only_call(&self) -> TestResult { + let mut calls = self.calls.lock().await; + let [call] = std::mem::take(&mut *calls) + .try_into() + .map_err(|calls: Vec| { + format!("expected one gateway call, got {}", calls.len()) + })?; + Ok(call) + } +} + +impl Drop for Gateway { + fn drop(&mut self) { + self.task.abort(); + } +} + +/// Builds the server from a config file, as `switchyard-server --config` does. +fn app(gateway: &Gateway) -> TestResult { + let mut config = tempfile::Builder::new().suffix(".toml").tempfile()?; + write!( + config, + r#" +schema_version = 1 + +[llm_clients.responses] +format = "openai_responses" +base_url = "{root}/v1" +max_retries = 0 + +[llm_clients.chat] +format = "openai_chat" +base_url = "{root}/v1" +max_retries = 0 + +[targets.claude] +id = "model/claude" +llm_client = "responses" +extra_body = {{ max_output_tokens = {CONFIGURED_OUTPUT_LIMIT} }} + +[targets.claude_uncapped] +id = "model/claude-uncapped" +llm_client = "responses" + +[targets.gpt] +id = "model/gpt" +llm_client = "responses" + +[targets.claude_chat] +id = "model/claude-chat" +llm_client = "chat" + +[routes.passthrough] +id = "gateway/passthrough" +type = "passthrough" +target = "claude" + +[routes.stage] +id = "gateway/stage" +type = "stage_router" +capable_target = "gpt" +efficient_target = "claude" +picker = "efficient_first" +confidence_threshold = 0.5 + +[routes.uncapped] +id = "gateway/uncapped" +type = "passthrough" +target = "claude_uncapped" + +[routes.chat] +id = "gateway/chat" +type = "passthrough" +target = "claude_chat" +"#, + root = gateway.root, + )?; + Ok(build_switchyard_router(load_server_state(config.path())?)) +} + +async fn post_json(app: &Router, path: &str, body: &Value) -> TestResult<(StatusCode, String)> { + let request = Request::builder() + .method("POST") + .uri(path) + .header("content-type", "application/json") + .body(Body::from(serde_json::to_vec(body)?))?; + let response = app.clone().oneshot(request).await?; + let status = response.status(); + let bytes = response.into_body().collect().await?.to_bytes(); + Ok((status, String::from_utf8(bytes.to_vec())?)) +} + +/// Every `data:` frame of an SSE body, parsed as JSON. +fn sse_events(body: &str) -> Vec { + body.lines() + .filter_map(|line| line.strip_prefix("data: ")) + .filter(|data| *data != "[DONE]") + .filter_map(|data| serde_json::from_str(data).ok()) + .collect() +} + +fn event_types(events: &[Value]) -> Vec<&str> { + events + .iter() + .map(|event| event["type"].as_str().unwrap_or_default()) + .collect() +} + +/// A Codex turn: the tools Codex offers and one user message naming the gateway scenario. +fn codex_turn(route: &str, scenario: &str, stream: bool) -> Value { + json!({ + "model": route, + "stream": stream, + "tools": [ + {"type": "custom", "name": "exec", "description": "Run JavaScript", "format": {"type": "text"}}, + {"type": "function", "name": "shell", "description": "Run a command", + "parameters": {"type": "object", "properties": {"cmd": {"type": "string"}}}} + ], + "tool_choice": "auto", + "parallel_tool_calls": true, + "input": [user_text(&format!("scenario:{scenario}"))] + }) +} + +fn user_text(text: &str) -> Value { + json!({"type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}) +} + +/// Answers like LiteLLM in front of Bedrock Claude. The `scenario:` text in the request +/// selects the response stream; see [`gateway_output`]. +async fn gateway( + State(calls): State>>>, + uri: Uri, + Json(body): Json, +) -> Response { + calls.lock().await.push(body.clone()); + if calls_a_tool(&body) && !defines_tools(&body) { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": {"message": LITELLM_TOOLS_ERROR, "type": "invalid_request_error", "code": "400"}})), + ) + .into_response(); + } + let model = body["model"].as_str().unwrap_or_default(); + if uri.path().ends_with("/chat/completions") { + return Json(json!({ + "id": "chatcmpl-gateway", "object": "chat.completion", "model": model, + "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "ok"}}], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2} + })) + .into_response(); + } + let (mut events, terminal) = gateway_output(&body); + if body["stream"] != true { + return Json(terminal["response"].clone()).into_response(); + } + let mut created = terminal["response"].clone(); + created["status"] = json!("in_progress"); + created["output"] = json!([]); + events.insert(0, json!({"type": "response.created", "response": created})); + events.push(terminal); + // The real gateway sends data-only frames without `event:` names. + let frames = events + .into_iter() + .map(|event| Ok::(Event::default().data(event.to_string()))); + Sse::new(futures_util::stream::iter(frames)).into_response() +} + +fn calls_a_tool(body: &Value) -> bool { + let responses_call = body["input"].as_array().into_iter().flatten().any(|item| { + matches!( + item["type"].as_str(), + Some("function_call" | "custom_tool_call") + ) + }); + let chat_call = body["messages"] + .as_array() + .into_iter() + .flatten() + .any(|message| message.get("tool_calls").is_some()); + responses_call || chat_call +} + +fn defines_tools(body: &Value) -> bool { + body["tools"] + .as_array() + .is_some_and(|tools| !tools.is_empty()) + || body["input"] + .as_array() + .into_iter() + .flatten() + .any(|item| item["type"] == "additional_tools") +} + +/// The streamed events between `response.created` and the terminal event, plus the terminal +/// event itself, for the scenario named in the request. +fn gateway_output(body: &Value) -> (Vec, Value) { + let text = body.to_string(); + let scenario: String = text + .split("scenario:") + .nth(1) + .unwrap_or_default() + .chars() + .take_while(|c| c.is_ascii_lowercase() || *c == '-') + .collect(); + let mut items = Vec::new(); + match scenario.as_str() { + "custom" => items.push(litellm_custom_call(1, "call_1", "ls")), + "two-custom" => { + items.push(litellm_custom_call(1, "call_1", "ls")); + items.push(litellm_custom_call(2, "call_2", "pwd")); + } + "custom-and-function" => { + items.push(litellm_custom_call(1, "call_1", "ls")); + let arguments = r#"{"cmd": "pwd"}"#; + items.push(function_call(2, "call_2", arguments, arguments)); + } + "native-custom" => items.push(openai_custom_call(0, "call_1", "ls\n")), + "conflicting-arguments" => { + items.push(function_call( + 0, + "call_1", + r#"{"cmd": "ls"}"#, + r#"{"cmd": "pwd"}"#, + )); + } + "conflicting-text" => items.push(text_message(0, "Hello", "Goodbye")), + "long-answer" if body.get("max_output_tokens").is_none() => { + let (events, item) = text_message(0, "1\n2\n3\n", "1\n2\n3\n"); + let mut response = + response_body(body, "incomplete", vec![item], GATEWAY_DEFAULT_OUTPUT_CAP); + response["incomplete_details"] = json!({"reason": "max_output_tokens"}); + let terminal = json!({"type": "response.incomplete", "response": response}); + return (events, terminal); + } + _ => items.push(text_message(0, "ok", "ok")), + } + // After tool calls, the gateway closes an empty text message at output index 0 that it + // never announced. + if items + .iter() + .any(|(_, item)| item["type"] == "custom_tool_call") + && scenario != "native-custom" + { + items.push(litellm_empty_message()); + } + let events = items + .iter() + .flat_map(|(events, _)| events.clone()) + .collect(); + let mut output: Vec<(u64, Value)> = items + .into_iter() + .map(|(events, item)| (output_index(&events), item)) + .collect(); + output.sort_by_key(|(index, _)| *index); + let output = output.into_iter().map(|(_, item)| item).collect(); + let terminal = json!({"type": "response.completed", "response": response_body(body, "completed", output, 1)}); + (events, terminal) +} + +fn output_index(events: &[Value]) -> u64 { + events + .first() + .and_then(|event| event["output_index"].as_u64()) + .unwrap_or_default() +} + +fn response_body(body: &Value, status: &str, output: Vec, output_tokens: u64) -> Value { + json!({ + "id": "resp_gateway", "object": "response", "model": body["model"], "status": status, + "output": output, + "usage": {"input_tokens": 1, "output_tokens": output_tokens, "total_tokens": 1 + output_tokens} + }) +} + +/// A freeform call in the gateway's shape: function-style argument deltas that spell +/// `{"content": ...}`, then a finished `custom_tool_call` whose `input` is the bare text. +fn litellm_custom_call(index: u64, call_id: &str, input: &str) -> (Vec, Value) { + let id = format!("ctc_{call_id}"); + let call = json!({"type": "custom_tool_call", "id": id, "call_id": call_id, "name": "exec"}); + let mut added = call.clone(); + added["status"] = json!("in_progress"); + added["input"] = json!(""); + let mut done = call; + done["status"] = json!("completed"); + done["input"] = json!(input); + let arguments = json!({"content": input}).to_string(); + let (first, second) = arguments.split_at(arguments.len() / 2); + let events = vec![ + json!({"type": "response.output_item.added", "output_index": index, "item": added}), + json!({"type": "response.function_call_arguments.delta", "item_id": id, "output_index": index, "delta": first}), + json!({"type": "response.function_call_arguments.delta", "item_id": id, "output_index": index, "delta": second}), + json!({"type": "response.function_call_arguments.done", "item_id": id, "output_index": index, "arguments": arguments}), + json!({"type": "response.output_item.done", "output_index": index, "item": done}), + ]; + (events, done) +} + +/// A freeform call in OpenAI's own shape, which streams the input with custom input deltas. +fn openai_custom_call(index: u64, call_id: &str, input: &str) -> (Vec, Value) { + let id = format!("ctc_{call_id}"); + let call = json!({"type": "custom_tool_call", "id": id, "call_id": call_id, "name": "exec"}); + let mut added = call.clone(); + added["status"] = json!("in_progress"); + added["input"] = json!(""); + let mut done = call; + done["status"] = json!("completed"); + done["input"] = json!(input); + let events = vec![ + json!({"type": "response.output_item.added", "output_index": index, "item": added}), + json!({"type": "response.custom_tool_call_input.delta", "item_id": id, "output_index": index, "delta": input}), + json!({"type": "response.custom_tool_call_input.done", "item_id": id, "output_index": index, "input": input}), + json!({"type": "response.output_item.done", "output_index": index, "item": done}), + ]; + (events, done) +} + +/// A function call whose streamed arguments are `streamed` and whose finished item says `done`. +fn function_call( + index: u64, + call_id: &str, + streamed: &str, + done_arguments: &str, +) -> (Vec, Value) { + let id = format!("fc_{call_id}"); + let call = json!({"type": "function_call", "id": id, "call_id": call_id, "name": "shell"}); + let mut added = call.clone(); + added["status"] = json!("in_progress"); + added["arguments"] = json!(""); + let mut done = call; + done["status"] = json!("completed"); + done["arguments"] = json!(done_arguments); + let events = vec![ + json!({"type": "response.output_item.added", "output_index": index, "item": added}), + json!({"type": "response.function_call_arguments.delta", "item_id": id, "output_index": index, "delta": streamed}), + json!({"type": "response.function_call_arguments.done", "item_id": id, "output_index": index, "arguments": streamed}), + json!({"type": "response.output_item.done", "output_index": index, "item": done}), + ]; + (events, done) +} + +/// A text message whose streamed text is `streamed` and whose finished item says `done`. +fn text_message(index: u64, streamed: &str, done_text: &str) -> (Vec, Value) { + let added = json!({"type": "message", "id": "msg_1", "role": "assistant", "status": "in_progress", "content": []}); + let done = json!({"type": "message", "id": "msg_1", "role": "assistant", "status": "completed", + "content": [{"type": "output_text", "text": done_text, "annotations": []}]}); + let events = vec![ + json!({"type": "response.output_item.added", "output_index": index, "item": added}), + json!({"type": "response.content_part.added", "item_id": "msg_1", "output_index": index, "content_index": 0, + "part": {"type": "output_text", "text": "", "annotations": []}}), + json!({"type": "response.output_text.delta", "item_id": "msg_1", "output_index": index, "content_index": 0, "delta": streamed}), + json!({"type": "response.output_text.done", "item_id": "msg_1", "output_index": index, "content_index": 0, "text": done_text}), + json!({"type": "response.output_item.done", "output_index": index, "item": done}), + ]; + (events, done) +} + +/// The empty text message the gateway closes at output index 0 after a tool call. +fn litellm_empty_message() -> (Vec, Value) { + let done = json!({"type": "message", "id": "msg_empty", "role": "assistant", "status": "completed", + "content": [{"type": "output_text", "text": "", "annotations": []}]}); + let events = vec![ + json!({"type": "response.output_text.done", "item_id": "msg_empty", "output_index": 0, "content_index": 0, "text": ""}), + json!({"type": "response.content_part.done", "item_id": "msg_empty", "output_index": 0, "content_index": 0, + "part": {"type": "output_text", "text": "", "annotations": []}}), + json!({"type": "response.output_item.done", "output_index": 0, "item": done}), + ]; + (events, done) +} + +/// The tool calls a stream finished, as `(type, name, input or arguments)`. +fn finished_tool_calls(events: &[Value]) -> Vec<(String, String, String)> { + events + .iter() + .filter(|event| event["type"] == "response.output_item.done") + .map(|event| &event["item"]) + .filter(|item| { + matches!( + item["type"].as_str(), + Some("custom_tool_call" | "function_call") + ) + }) + .map(|item| { + let value = item["input"].as_str().or(item["arguments"].as_str()); + ( + item["type"].as_str().unwrap_or_default().to_string(), + item["name"].as_str().unwrap_or_default().to_string(), + value.unwrap_or_default().to_string(), + ) + }) + .collect() +} + +/// An expected finished tool call: `(type, name, input or arguments)`. +type ToolCall<'a> = (&'a str, &'a str, &'a str); + +/// Codex showed "Reconnecting... (stream closed before response.completed)" after every tool +/// call because the server ended the stream after the gateway's freeform call. +#[tokio::test] +async fn codex_tool_call_streams_reach_response_completed() -> TestResult { + let gateway = Gateway::start().await?; + let app = app(&gateway)?; + let cases: [(&str, &[ToolCall]); 4] = [ + ("custom", &[("custom_tool_call", "exec", "ls")]), + ( + "two-custom", + &[ + ("custom_tool_call", "exec", "ls"), + ("custom_tool_call", "exec", "pwd"), + ], + ), + ( + "custom-and-function", + &[ + ("custom_tool_call", "exec", "ls"), + ("function_call", "shell", r#"{"cmd": "pwd"}"#), + ], + ), + // OpenAI streams freeform input with its own events; it must keep working. + ("native-custom", &[("custom_tool_call", "exec", "ls\n")]), + ]; + for route in CLAUDE_ROUTES { + for (scenario, expected) in cases { + let case = format!("{route} {scenario}"); + let (status, body) = + post_json(&app, "/v1/responses", &codex_turn(route, scenario, true)).await?; + assert_eq!(status, StatusCode::OK, "{case}: {body}"); + let events = sse_events(&body); + let types = event_types(&events); + assert_eq!( + types.last(), + Some(&"response.completed"), + "{case}: {types:?}" + ); + assert!(!types.contains(&"error"), "{case}: {types:?}"); + let expected: Vec<_> = expected + .iter() + .map(|(kind, name, value)| (kind.to_string(), name.to_string(), value.to_string())) + .collect(); + assert_eq!(finished_tool_calls(&events), expected, "{case}"); + assert_eq!( + gateway.only_call().await?["model"], + "model/claude", + "{case}" + ); + } + } + Ok(()) +} + +/// A finished item that contradicts content that was already streamed ends the stream with an +/// error event in its place. The contradicting event must not be forwarded as if it were the +/// error, which ended the stream with no error event and no log. +#[tokio::test] +async fn contradicting_snapshots_end_the_stream_with_an_error_event() -> TestResult { + let gateway = Gateway::start().await?; + let app = app(&gateway)?; + for route in CLAUDE_ROUTES { + for scenario in ["conflicting-arguments", "conflicting-text"] { + let case = format!("{route} {scenario}"); + let (status, body) = + post_json(&app, "/v1/responses", &codex_turn(route, scenario, true)).await?; + assert_eq!(status, StatusCode::OK, "{case}: {body}"); + let events = sse_events(&body); + let types = event_types(&events); + let last = events.last().ok_or("the stream must not be empty")?; + assert_eq!(last["type"], "error", "{case}: {types:?}"); + assert_eq!(last["message"], CONFLICT_MESSAGE, "{case}"); + // The contradicting item is replaced by the error, and nothing follows it. + assert!( + !types.contains(&"response.output_item.done"), + "{case}: {types:?}" + ); + assert!(!types.contains(&"response.completed"), "{case}: {types:?}"); + gateway.only_call().await?; + } + } + Ok(()) +} + +/// Codex sends no `max_output_tokens`, so the gateway stopped Sonnet at 4,096 tokens and Codex +/// ended long turns with an empty reply. A target's `extra_body` limit lifts the cap without +/// overriding a limit the caller sets. +#[tokio::test] +async fn target_output_limit_lifts_the_gateway_default_cap() -> TestResult { + let gateway = Gateway::start().await?; + let app = app(&gateway)?; + + // Without a configured limit the gateway cuts the answer off, and the client receives + // `response.incomplete`. + let (status, body) = post_json( + &app, + "/v1/responses", + &codex_turn("gateway/uncapped", "long-answer", true), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{body}"); + let events = sse_events(&body); + assert_eq!( + event_types(&events).last(), + Some(&"response.incomplete"), + "{body}" + ); + assert!( + gateway + .only_call() + .await? + .get("max_output_tokens") + .is_none() + ); + + for route in CLAUDE_ROUTES { + for stream in [true, false] { + let case = format!("{route} stream={stream}"); + let (status, body) = post_json( + &app, + "/v1/responses", + &codex_turn(route, "long-answer", stream), + ) + .await?; + assert_eq!(status, StatusCode::OK, "{case}: {body}"); + if stream { + let events = sse_events(&body); + assert_eq!( + event_types(&events).last(), + Some(&"response.completed"), + "{case}: {body}" + ); + } else { + assert_eq!( + serde_json::from_str::(&body)?["status"], + "completed", + "{case}" + ); + } + let call = gateway.only_call().await?; + assert_eq!(call["max_output_tokens"], CONFIGURED_OUTPUT_LIMIT, "{case}"); + } + + let mut limited = codex_turn(route, "long-answer", true); + limited["max_output_tokens"] = json!(1000); + let (status, body) = post_json(&app, "/v1/responses", &limited).await?; + assert_eq!(status, StatusCode::OK, "{route}: {body}"); + assert_eq!( + gateway.only_call().await?["max_output_tokens"], + 1000, + "{route}" + ); + } + Ok(()) +} + +/// The `(type, name)` of each tool definition in a gateway request, for either OpenAI API. +fn defined_tools(call: &Value) -> Vec<(String, String)> { + call["tools"] + .as_array() + .into_iter() + .flatten() + .map(|tool| { + let name = tool["name"].as_str().or(tool["function"]["name"].as_str()); + ( + tool["type"].as_str().unwrap_or_default().to_string(), + name.unwrap_or_default().to_string(), + ) + }) + .collect() +} + +fn codex_compaction(route: &str, stream: bool) -> Value { + json!({ + "model": route, + "stream": stream, + "tools": [], + "tool_choice": "auto", + "parallel_tool_calls": false, + "input": [ + user_text("List files"), + {"type": "custom_tool_call", "call_id": "call_1", "name": "exec", "input": "ls"}, + {"type": "custom_tool_call_output", "call_id": "call_1", "output": "README.md"}, + {"type": "function_call", "call_id": "call_2", "name": "shell", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_2", "output": "ok"}, + {"type": "function_call", "call_id": "call_3", "name": "shell", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_3", "output": "ok"}, + {"type": "function_call", "call_id": "call_4", "namespace": "collaboration", + "name": "spawn_agent", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_4", "output": "ok"}, + user_text("Summarize the conversation.") + ] + }) +} + +/// Codex compacts a long thread by sending its tool-call history with an empty `tools` list. +/// The gateway rejected those requests with HTTP 400, so compaction never finished. +#[tokio::test] +async fn tool_history_without_tools_gets_definitions_the_gateway_accepts() -> TestResult { + let gateway = Gateway::start().await?; + let app = app(&gateway)?; + let codex_tools = [ + ("custom".to_string(), "exec".to_string()), + ("function".to_string(), "shell".to_string()), + ( + "function".to_string(), + "collaboration__spawn_agent".to_string(), + ), + ]; + for route in CLAUDE_ROUTES { + for stream in [true, false] { + let case = format!("{route} compaction stream={stream}"); + let (status, body) = + post_json(&app, "/v1/responses", &codex_compaction(route, stream)).await?; + assert_eq!(status, StatusCode::OK, "{case}: {body}"); + if stream { + let types = event_types(&sse_events(&body)).join(","); + assert!(types.ends_with("response.completed"), "{case}: {types}"); + } + let call = gateway.only_call().await?; + assert_eq!(call["model"], "model/claude", "{case}"); + assert_eq!(defined_tools(&call), codex_tools, "{case}"); + assert_eq!(call["tool_choice"], "none", "{case}"); + } + + // Anthropic Messages clients such as Claude Code reach the same target through + // translation. + let messages = json!({ + "model": route, + "max_tokens": 300, + "messages": [ + {"role": "user", "content": "List files"}, + {"role": "assistant", "content": [{"type": "tool_use", "id": "toolu_1", "name": "Bash", "input": {"command": "ls"}}]}, + {"role": "user", "content": [ + {"type": "tool_result", "tool_use_id": "toolu_1", "content": "README.md"}, + {"type": "text", "text": "Summarize the conversation."} + ]} + ] + }); + let (status, body) = post_json(&app, "/v1/messages", &messages).await?; + assert_eq!(status, StatusCode::OK, "{route} messages: {body}"); + let call = gateway.only_call().await?; + assert_eq!( + defined_tools(&call), + [("function".to_string(), "Bash".to_string())], + "{route} messages" + ); + assert_eq!(call["tool_choice"], "none", "{route} messages"); + } + + let chat = json!({ + "model": "gateway/chat", + "messages": [ + {"role": "user", "content": "List files"}, + {"role": "assistant", "content": null, "tool_calls": [ + {"id": "call_1", "type": "function", "function": {"name": "shell", "arguments": "{}"}}, + {"id": "call_2", "type": "function", "function": {"name": "shell", "arguments": "{}"}} + ]}, + {"role": "tool", "tool_call_id": "call_1", "content": "README.md"}, + {"role": "tool", "tool_call_id": "call_2", "content": "README.md"}, + {"role": "user", "content": "Summarize the conversation."} + ] + }); + let (status, body) = post_json(&app, "/v1/chat/completions", &chat).await?; + assert_eq!(status, StatusCode::OK, "chat: {body}"); + let call = gateway.only_call().await?; + assert_eq!( + defined_tools(&call), + [("function".to_string(), "shell".to_string())] + ); + assert_eq!(call["tool_choice"], "none"); + Ok(()) +} + +/// Requests that define their own tools, or that call no tools, reach the gateway unchanged. +#[tokio::test] +async fn requests_with_their_own_tools_or_no_tool_history_are_unchanged() -> TestResult { + let gateway = Gateway::start().await?; + let app = app(&gateway)?; + for route in CLAUDE_ROUTES { + let mut defined = codex_compaction(route, true); + defined["tools"] = json!([{"type": "function", "name": "shell", "description": "Run a command", + "parameters": {"type": "object"}}]); + let (status, body) = post_json(&app, "/v1/responses", &defined).await?; + assert_eq!(status, StatusCode::OK, "{route} defined: {body}"); + let call = gateway.only_call().await?; + assert_eq!(call["tools"], defined["tools"], "{route} defined"); + assert_eq!(call["tool_choice"], "auto", "{route} defined"); + + // Codex's Responses-lite requests carry their tools inside `input`. + let mut lite = codex_compaction(route, true); + let lite_tools = json!({"type": "additional_tools", "role": "developer", + "tools": [{"type": "function", "name": "shell", "parameters": {"type": "object"}}]}); + lite.as_object_mut() + .ok_or("request must be an object")? + .remove("tools"); + lite["input"] + .as_array_mut() + .ok_or("input must be an array")? + .insert(0, lite_tools.clone()); + let (status, body) = post_json(&app, "/v1/responses", &lite).await?; + assert_eq!(status, StatusCode::OK, "{route} lite: {body}"); + let call = gateway.only_call().await?; + assert!(call.get("tools").is_none(), "{route} lite: {call}"); + assert_eq!(call["input"][0], lite_tools, "{route} lite"); + assert_eq!(call["tool_choice"], "auto", "{route} lite"); + + let mut plain = codex_turn(route, "ok", true); + plain["tools"] = json!([]); + let (status, body) = post_json(&app, "/v1/responses", &plain).await?; + assert_eq!(status, StatusCode::OK, "{route} plain: {body}"); + let call = gateway.only_call().await?; + assert_eq!(defined_tools(&call), [], "{route} plain"); + assert_eq!(call["tool_choice"], "auto", "{route} plain"); + } + Ok(()) +} diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index 6d30a60e2d..d64ed93669 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -188,22 +188,23 @@ fn decode_responses_stream( event .get("delta") .and_then(Value::as_str) - .map(|delta| { + .and_then(|delta| { state.decoded_tool_call = true; + let tool = state.tool_states.entry(output_index as usize).or_default(); + // The completed item carries a custom call's input. The decoder skips + // argument deltas for that call because their JSON can differ from it. + if tool.decoded_custom_call { + return None; + } // 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 { + tool.decoded_arguments.push_str(delta); + Some(vec![LlmResponseChunk::ToolCallDelta { index: output_index as usize, id: None, name: None, arguments_delta: Some(delta.to_string()), - }] + }]) }) .unwrap_or_default() } @@ -269,7 +270,7 @@ fn decode_responses_stream( for (position, item) in items.iter().enumerate() { if let Some(item) = item.as_object() { out.extend(decode_responses_completed_item(item, position, state)); - if matches!(out.last(), Some(LlmResponseChunk::StreamError { .. })) { + if matches!(out.last(), Some(LlmResponseChunk::DecodeError { .. })) { return out; } } @@ -670,12 +671,10 @@ fn decode_responses_output_item_added( .or_else(|| item.get("id")) .and_then(Value::as_str); let name = item.get("name").and_then(Value::as_str); - state - .tool_states - .entry(index) - .or_default() - .has_decoded_identity |= + let tool = state.tool_states.entry(index).or_default(); + tool.has_decoded_identity |= id.is_some_and(|id| !id.is_empty()) && name.is_some_and(|name| !name.is_empty()); + tool.decoded_custom_call |= item_type == Some("custom_tool_call"); // 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") { @@ -789,13 +788,15 @@ fn decode_responses_completed_item( Vec::new() } -// A snapshot may extend streamed content, but cannot retract content already sent. +// A snapshot may extend streamed content, but cannot retract content already sent. A +// conflict means the provider sent inconsistent data, so it decodes to `DecodeError`: the +// server then replaces the event with an error frame instead of replaying it. fn snapshot_suffix( decoded: &mut String, snapshot: &str, ) -> Result, LlmResponseChunk> { let Some(suffix) = snapshot.strip_prefix(decoded.as_str()) else { - return Err(LlmResponseChunk::StreamError { + return Err(LlmResponseChunk::DecodeError { message: "Responses snapshot conflicts with streamed content".to_string(), }); }; diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index db83015a7c..04f3b0f299 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -131,6 +131,11 @@ pub(crate) struct StreamToolState { pub(crate) decoded_arguments: String, #[serde(default)] pub(crate) has_decoded_identity: bool, + /// True when the source announced a `custom_tool_call`. The decoder then reads the call's + /// input only from the completed item, because some providers also stream argument deltas + /// whose JSON does not match that input. + #[serde(default)] + pub(crate) decoded_custom_call: bool, 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 26d65adae9..fdd093c06e 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -3419,7 +3419,7 @@ fn responses_terminal_snapshots_recover_missing_output_once() -> TestResult { )?; assert!(matches!( decoded.normalized().last(), - Some(LlmResponseChunk::StreamError { .. }) + Some(LlmResponseChunk::DecodeError { .. }) )); assert!( !decoded @@ -3430,3 +3430,80 @@ fn responses_terminal_snapshots_recover_missing_output_once() -> TestResult { } Ok(()) } + +// LiteLLM in front of Bedrock streams a freeform call as function-style argument deltas that +// spell `{"content": ...}`. OpenAI streams it with custom input deltas. Both must decode to +// one call whose only arguments are the finished input, and the terminal snapshot that +// repeats the call must neither conflict with it nor repeat it. +#[test] +fn responses_custom_tool_calls_decode_from_the_finished_item() -> TestResult { + let engine = TranslationEngine::default(); + let source = WireFormat::OpenAiResponses; + let call = + json!({"type": "custom_tool_call", "id": "ctc_1", "call_id": "call_1", "name": "exec"}); + let mut added = call.clone(); + added["input"] = json!(""); + let mut done = call; + done["input"] = json!("ls"); + let added = json!({"type": "response.output_item.added", "output_index": 1, "item": added}); + let item_done = json!({"type": "response.output_item.done", "output_index": 1, "item": done}); + let completed = json!({"type": "response.completed", "response": {"id": "resp_1", "status": "completed", + "output": [{"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": ""}]}, done]}}); + let litellm = vec![ + added.clone(), + json!({"type": "response.function_call_arguments.delta", "output_index": 1, "delta": "{\"content\": "}), + json!({"type": "response.function_call_arguments.delta", "output_index": 1, "delta": "\"ls\"}"}), + json!({"type": "response.function_call_arguments.done", "output_index": 1, "arguments": "{\"content\": \"ls\"}"}), + item_done.clone(), + completed.clone(), + ]; + let openai = vec![ + added, + json!({"type": "response.custom_tool_call_input.delta", "output_index": 1, "delta": "ls"}), + json!({"type": "response.custom_tool_call_input.done", "output_index": 1, "input": "ls"}), + item_done, + completed, + ]; + for (shape, events) in [("litellm", litellm), ("openai", openai)] { + let mut state = StreamTranslationState::new(source, WireFormat::OpenAiChat); + let mut chunks = Vec::new(); + for event in events { + chunks.extend( + engine + .decode_stream_event(&mut state, source, event)? + .normalized() + .to_vec(), + ); + } + assert!( + !chunks.iter().any(|chunk| matches!( + chunk, + LlmResponseChunk::DecodeError { .. } | LlmResponseChunk::StreamError { .. } + )), + "{shape}: {chunks:?}" + ); + let arguments: String = chunks + .iter() + .filter_map(|chunk| match chunk { + LlmResponseChunk::ToolCallDelta { + arguments_delta: Some(arguments), + .. + } => Some(arguments.as_str()), + _ => None, + }) + .collect(); + assert_eq!(arguments, r#"{"input":"ls"}"#, "{shape}"); + assert!( + chunks.iter().any(|chunk| matches!(chunk, + LlmResponseChunk::ToolCallDelta { id: Some(id), name: Some(name), .. } + if id == "call_1" && name == "exec")), + "{shape}: {chunks:?}" + ); + assert!( + chunks.iter().any(|chunk| matches!(chunk, + LlmResponseChunk::MessageStop { reason: Some(reason) } if reason == "tool_use")), + "{shape}: {chunks:?}" + ); + } + Ok(()) +} From 1273bd3a6d9d812e968799862526fde6fc323a4a Mon Sep 17 00:00:00 2001 From: Elyas Mehtabuddin Date: Tue, 13 Oct 2026 09:14:12 -0700 Subject: [PATCH 2/5] test(server): shrink the LiteLLM gateway regression tests Signed-off-by: Elyas Mehtabuddin --- crates/libsy-llm-client/src/client.rs | 63 ++ .../tests/litellm_gateway.rs | 724 ++++-------------- 2 files changed, 227 insertions(+), 560 deletions(-) diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 650f8d591f..291a8cb80f 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -1775,6 +1775,69 @@ mod tests { ); } + // A gateway in front of Bedrock rejects tool-call history in a request with no tools, so + // the client defines each called tool. Requests with their own tool definitions, with no + // tool calls, or bound for an Anthropic backend keep their bodies. + #[test] + fn replayed_tool_calls_get_definitions_only_when_the_request_has_none() { + let custom = + json!({"type": "custom_tool_call", "call_id": "c1", "name": "exec", "input": "ls"}); + let function = + json!({"type": "function_call", "call_id": "c2", "name": "shell", "arguments": "{}"}); + let mut responses = + json!({"tools": [], "tool_choice": "auto", "input": [custom, function, function]}); + define_tools_for_replayed_calls(&mut responses, WireFormat::OpenAiResponses); + let defined: Vec<_> = responses["tools"] + .as_array() + .into_iter() + .flatten() + .map(|tool| (&tool["type"], &tool["name"])) + .collect(); + assert_eq!( + defined, + [ + (&json!("custom"), &json!("exec")), + (&json!("function"), &json!("shell")) + ] + ); + assert_eq!(responses["tool_choice"], "none"); + + let shell = json!({"type": "function", "function": {"name": "shell", "arguments": "{}"}}); + let mut chat = json!({"messages": [{"role": "assistant", "tool_calls": [shell, shell]}]}); + define_tools_for_replayed_calls(&mut chat, WireFormat::OpenAiChat); + assert_eq!(chat["tools"].as_array().map(Vec::len), Some(1)); + assert_eq!(chat["tools"][0]["function"]["name"], "shell"); + assert_eq!(chat["tool_choice"], "none"); + + let anthropic_history = json!([ + {"role": "assistant", "content": [{"type": "tool_use", "id": "t1", "name": "Bash", "input": {}}]}, + {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1", "content": "ok"}]} + ]); + for (body, format) in [ + ( + json!({"tools": [{"type": "function", "name": "own"}], "tool_choice": "auto", "input": [function]}), + WireFormat::OpenAiResponses, + ), + // A Responses-lite request defines its tools inside `input`. + ( + json!({"input": [{"type": "additional_tools", "tools": []}, function]}), + WireFormat::OpenAiResponses, + ), + ( + json!({"tools": [], "input": [{"type": "message", "role": "user", "content": "hi"}]}), + WireFormat::OpenAiResponses, + ), + ( + json!({"messages": anthropic_history}), + WireFormat::AnthropicMessages, + ), + ] { + let mut edited = body.clone(); + define_tools_for_replayed_calls(&mut edited, format); + assert_eq!(edited, body, "{format}"); + } + } + // A request that pins `format` in its metadata, so the client resolves that // wire format instead of the model's default backend. fn request_with_wire_format(model: &str, format: WireFormat) -> Request { diff --git a/crates/switchyard-server/tests/litellm_gateway.rs b/crates/switchyard-server/tests/litellm_gateway.rs index d32cf6c541..ced8e109ed 100644 --- a/crates/switchyard-server/tests/litellm_gateway.rs +++ b/crates/switchyard-server/tests/litellm_gateway.rs @@ -3,11 +3,10 @@ //! Regression tests for Codex traffic sent through a LiteLLM gateway to Bedrock Claude. //! -//! The stub gateway reproduces three behaviors of the real gateway that broke Codex: -//! it streams a freeform tool call as function-style argument deltas whose JSON differs from -//! the finished call, it stops at 4,096 output tokens when a request sets no limit, and it -//! rejects tool-call history in a request that defines no tools. Each test runs the -//! production server, config loader, router, and HTTP client against that stub. +//! The stub gateway reproduces two behaviors of the real gateway that broke Codex: it streams +//! a freeform tool call as function-style argument deltas whose JSON differs from the +//! finished call, and it rejects tool-call history in a request that defines no tools. Each +//! test runs the production server, config loader, router, and HTTP client against that stub. use std::convert::Infallible; use std::error::Error; @@ -16,7 +15,7 @@ use std::sync::Arc; use axum::body::Body; use axum::extract::State; -use axum::http::{Request, StatusCode, Uri}; +use axum::http::{Request, StatusCode}; use axum::response::sse::{Event, Sse}; use axum::response::{IntoResponse, Response}; use axum::routing::post; @@ -31,12 +30,6 @@ use tower::ServiceExt; type TestResult = Result>; -/// Output tokens the gateway allows Bedrock Claude when a request sets no limit. -const GATEWAY_DEFAULT_OUTPUT_CAP: u64 = 4096; -/// Output limit that the `claude` target adds through `extra_body`. -const CONFIGURED_OUTPUT_LIMIT: u64 = 64000; -const LITELLM_TOOLS_ERROR: &str = "litellm.UnsupportedParamsError: Bedrock doesn't support tool calling without `tools=` param specified."; -const CONFLICT_MESSAGE: &str = "Responses snapshot conflicts with streamed content"; /// Both route types send the requests in these tests to the `claude` target. const CLAUDE_ROUTES: [&str; 2] = ["gateway/passthrough", "gateway/stage"]; @@ -52,7 +45,6 @@ impl Gateway { let calls = Arc::new(Mutex::new(Vec::new())); let app = Router::new() .route("/v1/responses", post(gateway)) - .route("/v1/chat/completions", post(gateway)) .with_state(Arc::clone(&calls)); let listener = TcpListener::bind("127.0.0.1:0").await?; let root = format!("http://{}", listener.local_addr()?); @@ -62,12 +54,9 @@ impl Gateway { /// The single request the gateway received since the last call. async fn only_call(&self) -> TestResult { - let mut calls = self.calls.lock().await; - let [call] = std::mem::take(&mut *calls) - .try_into() - .map_err(|calls: Vec| { - format!("expected one gateway call, got {}", calls.len()) - })?; + let calls = std::mem::take(&mut *self.calls.lock().await); + let [call] = <[Value; 1]>::try_from(calls) + .map_err(|calls| format!("expected one gateway call, got {}", calls.len()))?; Ok(call) } } @@ -86,32 +75,18 @@ fn app(gateway: &Gateway) -> TestResult { r#" schema_version = 1 -[llm_clients.responses] +[llm_clients.gateway] format = "openai_responses" base_url = "{root}/v1" max_retries = 0 -[llm_clients.chat] -format = "openai_chat" -base_url = "{root}/v1" -max_retries = 0 - [targets.claude] id = "model/claude" -llm_client = "responses" -extra_body = {{ max_output_tokens = {CONFIGURED_OUTPUT_LIMIT} }} - -[targets.claude_uncapped] -id = "model/claude-uncapped" -llm_client = "responses" +llm_client = "gateway" [targets.gpt] id = "model/gpt" -llm_client = "responses" - -[targets.claude_chat] -id = "model/claude-chat" -llm_client = "chat" +llm_client = "gateway" [routes.passthrough] id = "gateway/passthrough" @@ -125,26 +100,16 @@ capable_target = "gpt" efficient_target = "claude" picker = "efficient_first" confidence_threshold = 0.5 - -[routes.uncapped] -id = "gateway/uncapped" -type = "passthrough" -target = "claude_uncapped" - -[routes.chat] -id = "gateway/chat" -type = "passthrough" -target = "claude_chat" "#, root = gateway.root, )?; Ok(build_switchyard_router(load_server_state(config.path())?)) } -async fn post_json(app: &Router, path: &str, body: &Value) -> TestResult<(StatusCode, String)> { +async fn post_responses(app: &Router, body: &Value) -> TestResult<(StatusCode, String)> { let request = Request::builder() .method("POST") - .uri(path) + .uri("/v1/responses") .header("content-type", "application/json") .body(Body::from(serde_json::to_vec(body)?))?; let response = app.clone().oneshot(request).await?; @@ -157,7 +122,6 @@ async fn post_json(app: &Router, path: &str, body: &Value) -> TestResult<(Status fn sse_events(body: &str) -> Vec { body.lines() .filter_map(|line| line.strip_prefix("data: ")) - .filter(|data| *data != "[DONE]") .filter_map(|data| serde_json::from_str(data).ok()) .collect() } @@ -169,59 +133,76 @@ fn event_types(events: &[Value]) -> Vec<&str> { .collect() } -/// A Codex turn: the tools Codex offers and one user message naming the gateway scenario. -fn codex_turn(route: &str, scenario: &str, stream: bool) -> Value { +fn user_text(text: &str) -> Value { + json!({"type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}) +} + +/// A streamed Codex turn whose user message names the gateway scenario. +fn codex_turn(route: &str, scenario: &str) -> Value { json!({ "model": route, - "stream": stream, + "stream": true, "tools": [ {"type": "custom", "name": "exec", "description": "Run JavaScript", "format": {"type": "text"}}, {"type": "function", "name": "shell", "description": "Run a command", "parameters": {"type": "object", "properties": {"cmd": {"type": "string"}}}} ], "tool_choice": "auto", - "parallel_tool_calls": true, - "input": [user_text(&format!("scenario:{scenario}"))] + "input": [user_text(scenario)] }) } -fn user_text(text: &str) -> Value { - json!({"type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}) -} - -/// Answers like LiteLLM in front of Bedrock Claude. The `scenario:` text in the request -/// selects the response stream; see [`gateway_output`]. -async fn gateway( - State(calls): State>>>, - uri: Uri, - Json(body): Json, -) -> Response { +/// Answers like LiteLLM in front of Bedrock Claude: it rejects tool-call history without tool +/// definitions, and otherwise replies with the scenario that the request's text names. +async fn gateway(State(calls): State>>>, Json(body): Json) -> Response { calls.lock().await.push(body.clone()); - if calls_a_tool(&body) && !defines_tools(&body) { + let input = body["input"].as_array().cloned().unwrap_or_default(); + let calls_a_tool = input.iter().any(|item| { + matches!( + item["type"].as_str(), + Some("function_call" | "custom_tool_call") + ) + }); + if calls_a_tool && body["tools"].as_array().is_none_or(Vec::is_empty) { + let message = "litellm.UnsupportedParamsError: Bedrock doesn't support tool calling without `tools=` param specified."; return ( StatusCode::BAD_REQUEST, - Json(json!({"error": {"message": LITELLM_TOOLS_ERROR, "type": "invalid_request_error", "code": "400"}})), + Json(json!({"error": {"message": message, "type": "invalid_request_error"}})), ) .into_response(); } - let model = body["model"].as_str().unwrap_or_default(); - if uri.path().ends_with("/chat/completions") { - return Json(json!({ - "id": "chatcmpl-gateway", "object": "chat.completion", "model": model, - "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": "ok"}}], - "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2} - })) - .into_response(); - } - let (mut events, terminal) = gateway_output(&body); + let (mut events, output) = match body.to_string() { + text if text.contains("custom-and-function") => { + // The gateway also closes an empty text message at output index 0 that it never + // announced. + let (custom_events, custom) = litellm_custom_call(1, "ls"); + let arguments = r#"{"cmd": "pwd"}"#; + let (function_events, function) = function_call(2, arguments); + let (empty_events, empty) = text_message(0, None, ""); + let events = [custom_events, function_events, empty_events].concat(); + (events, vec![empty, custom, function]) + } + text if text.contains("conflicting-text") => { + let (events, message) = text_message(0, Some("Hello"), "Goodbye"); + (events, vec![message]) + } + _ => { + let (events, message) = text_message(0, Some("ok"), "ok"); + (events, vec![message]) + } + }; + let response = json!({ + "id": "resp_gateway", "object": "response", "model": body["model"], "status": "completed", + "output": output, "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2} + }); if body["stream"] != true { - return Json(terminal["response"].clone()).into_response(); + return Json(response).into_response(); } - let mut created = terminal["response"].clone(); + let mut created = response.clone(); created["status"] = json!("in_progress"); created["output"] = json!([]); events.insert(0, json!({"type": "response.created", "response": created})); - events.push(terminal); + events.push(json!({"type": "response.completed", "response": response})); // The real gateway sends data-only frames without `event:` names. let frames = events .into_iter() @@ -229,548 +210,171 @@ async fn gateway( Sse::new(futures_util::stream::iter(frames)).into_response() } -fn calls_a_tool(body: &Value) -> bool { - let responses_call = body["input"].as_array().into_iter().flatten().any(|item| { - matches!( - item["type"].as_str(), - Some("function_call" | "custom_tool_call") - ) - }); - let chat_call = body["messages"] - .as_array() - .into_iter() - .flatten() - .any(|message| message.get("tool_calls").is_some()); - responses_call || chat_call -} - -fn defines_tools(body: &Value) -> bool { - body["tools"] - .as_array() - .is_some_and(|tools| !tools.is_empty()) - || body["input"] - .as_array() - .into_iter() - .flatten() - .any(|item| item["type"] == "additional_tools") -} - -/// The streamed events between `response.created` and the terminal event, plus the terminal -/// event itself, for the scenario named in the request. -fn gateway_output(body: &Value) -> (Vec, Value) { - let text = body.to_string(); - let scenario: String = text - .split("scenario:") - .nth(1) - .unwrap_or_default() - .chars() - .take_while(|c| c.is_ascii_lowercase() || *c == '-') - .collect(); - let mut items = Vec::new(); - match scenario.as_str() { - "custom" => items.push(litellm_custom_call(1, "call_1", "ls")), - "two-custom" => { - items.push(litellm_custom_call(1, "call_1", "ls")); - items.push(litellm_custom_call(2, "call_2", "pwd")); - } - "custom-and-function" => { - items.push(litellm_custom_call(1, "call_1", "ls")); - let arguments = r#"{"cmd": "pwd"}"#; - items.push(function_call(2, "call_2", arguments, arguments)); - } - "native-custom" => items.push(openai_custom_call(0, "call_1", "ls\n")), - "conflicting-arguments" => { - items.push(function_call( - 0, - "call_1", - r#"{"cmd": "ls"}"#, - r#"{"cmd": "pwd"}"#, - )); - } - "conflicting-text" => items.push(text_message(0, "Hello", "Goodbye")), - "long-answer" if body.get("max_output_tokens").is_none() => { - let (events, item) = text_message(0, "1\n2\n3\n", "1\n2\n3\n"); - let mut response = - response_body(body, "incomplete", vec![item], GATEWAY_DEFAULT_OUTPUT_CAP); - response["incomplete_details"] = json!({"reason": "max_output_tokens"}); - let terminal = json!({"type": "response.incomplete", "response": response}); - return (events, terminal); - } - _ => items.push(text_message(0, "ok", "ok")), - } - // After tool calls, the gateway closes an empty text message at output index 0 that it - // never announced. - if items - .iter() - .any(|(_, item)| item["type"] == "custom_tool_call") - && scenario != "native-custom" - { - items.push(litellm_empty_message()); - } - let events = items - .iter() - .flat_map(|(events, _)| events.clone()) - .collect(); - let mut output: Vec<(u64, Value)> = items - .into_iter() - .map(|(events, item)| (output_index(&events), item)) - .collect(); - output.sort_by_key(|(index, _)| *index); - let output = output.into_iter().map(|(_, item)| item).collect(); - let terminal = json!({"type": "response.completed", "response": response_body(body, "completed", output, 1)}); - (events, terminal) -} - -fn output_index(events: &[Value]) -> u64 { - events - .first() - .and_then(|event| event["output_index"].as_u64()) - .unwrap_or_default() -} - -fn response_body(body: &Value, status: &str, output: Vec, output_tokens: u64) -> Value { - json!({ - "id": "resp_gateway", "object": "response", "model": body["model"], "status": status, - "output": output, - "usage": {"input_tokens": 1, "output_tokens": output_tokens, "total_tokens": 1 + output_tokens} - }) -} - /// A freeform call in the gateway's shape: function-style argument deltas that spell /// `{"content": ...}`, then a finished `custom_tool_call` whose `input` is the bare text. -fn litellm_custom_call(index: u64, call_id: &str, input: &str) -> (Vec, Value) { - let id = format!("ctc_{call_id}"); - let call = json!({"type": "custom_tool_call", "id": id, "call_id": call_id, "name": "exec"}); - let mut added = call.clone(); - added["status"] = json!("in_progress"); - added["input"] = json!(""); - let mut done = call; - done["status"] = json!("completed"); +fn litellm_custom_call(index: u64, input: &str) -> (Vec, Value) { + let added = json!({"type": "custom_tool_call", "id": "ctc_1", "call_id": "call_1", "name": "exec", "input": ""}); + let mut done = added.clone(); done["input"] = json!(input); let arguments = json!({"content": input}).to_string(); let (first, second) = arguments.split_at(arguments.len() / 2); let events = vec![ json!({"type": "response.output_item.added", "output_index": index, "item": added}), - json!({"type": "response.function_call_arguments.delta", "item_id": id, "output_index": index, "delta": first}), - json!({"type": "response.function_call_arguments.delta", "item_id": id, "output_index": index, "delta": second}), - json!({"type": "response.function_call_arguments.done", "item_id": id, "output_index": index, "arguments": arguments}), + json!({"type": "response.function_call_arguments.delta", "item_id": "ctc_1", "output_index": index, "delta": first}), + json!({"type": "response.function_call_arguments.delta", "item_id": "ctc_1", "output_index": index, "delta": second}), + json!({"type": "response.function_call_arguments.done", "item_id": "ctc_1", "output_index": index, "arguments": arguments}), json!({"type": "response.output_item.done", "output_index": index, "item": done}), ]; (events, done) } -/// A freeform call in OpenAI's own shape, which streams the input with custom input deltas. -fn openai_custom_call(index: u64, call_id: &str, input: &str) -> (Vec, Value) { - let id = format!("ctc_{call_id}"); - let call = json!({"type": "custom_tool_call", "id": id, "call_id": call_id, "name": "exec"}); - let mut added = call.clone(); - added["status"] = json!("in_progress"); - added["input"] = json!(""); - let mut done = call; - done["status"] = json!("completed"); - done["input"] = json!(input); - let events = vec![ - json!({"type": "response.output_item.added", "output_index": index, "item": added}), - json!({"type": "response.custom_tool_call_input.delta", "item_id": id, "output_index": index, "delta": input}), - json!({"type": "response.custom_tool_call_input.done", "item_id": id, "output_index": index, "input": input}), - json!({"type": "response.output_item.done", "output_index": index, "item": done}), - ]; - (events, done) -} - -/// A function call whose streamed arguments are `streamed` and whose finished item says `done`. -fn function_call( - index: u64, - call_id: &str, - streamed: &str, - done_arguments: &str, -) -> (Vec, Value) { - let id = format!("fc_{call_id}"); - let call = json!({"type": "function_call", "id": id, "call_id": call_id, "name": "shell"}); - let mut added = call.clone(); - added["status"] = json!("in_progress"); - added["arguments"] = json!(""); - let mut done = call; - done["status"] = json!("completed"); - done["arguments"] = json!(done_arguments); +fn function_call(index: u64, arguments: &str) -> (Vec, Value) { + let added = json!({"type": "function_call", "id": "fc_2", "call_id": "call_2", "name": "shell", "arguments": ""}); + let mut done = added.clone(); + done["arguments"] = json!(arguments); let events = vec![ json!({"type": "response.output_item.added", "output_index": index, "item": added}), - json!({"type": "response.function_call_arguments.delta", "item_id": id, "output_index": index, "delta": streamed}), - json!({"type": "response.function_call_arguments.done", "item_id": id, "output_index": index, "arguments": streamed}), + json!({"type": "response.function_call_arguments.delta", "item_id": "fc_2", "output_index": index, "delta": arguments}), json!({"type": "response.output_item.done", "output_index": index, "item": done}), ]; (events, done) } -/// A text message whose streamed text is `streamed` and whose finished item says `done`. -fn text_message(index: u64, streamed: &str, done_text: &str) -> (Vec, Value) { - let added = json!({"type": "message", "id": "msg_1", "role": "assistant", "status": "in_progress", "content": []}); +/// A text message whose streamed text is `streamed` and whose finished item says `done_text`. +/// Without `streamed`, the message is only closed, as the gateway does after a tool call. +fn text_message(index: u64, streamed: Option<&str>, done_text: &str) -> (Vec, Value) { let done = json!({"type": "message", "id": "msg_1", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": done_text, "annotations": []}]}); - let events = vec![ - json!({"type": "response.output_item.added", "output_index": index, "item": added}), - json!({"type": "response.content_part.added", "item_id": "msg_1", "output_index": index, "content_index": 0, - "part": {"type": "output_text", "text": "", "annotations": []}}), - json!({"type": "response.output_text.delta", "item_id": "msg_1", "output_index": index, "content_index": 0, "delta": streamed}), + let mut events = Vec::new(); + if let Some(streamed) = streamed { + events.extend([ + json!({"type": "response.output_item.added", "output_index": index, + "item": {"type": "message", "id": "msg_1", "role": "assistant", "content": []}}), + json!({"type": "response.output_text.delta", "item_id": "msg_1", "output_index": index, "content_index": 0, "delta": streamed}), + ]); + } + events.extend([ json!({"type": "response.output_text.done", "item_id": "msg_1", "output_index": index, "content_index": 0, "text": done_text}), json!({"type": "response.output_item.done", "output_index": index, "item": done}), - ]; + ]); (events, done) } -/// The empty text message the gateway closes at output index 0 after a tool call. -fn litellm_empty_message() -> (Vec, Value) { - let done = json!({"type": "message", "id": "msg_empty", "role": "assistant", "status": "completed", - "content": [{"type": "output_text", "text": "", "annotations": []}]}); - let events = vec![ - json!({"type": "response.output_text.done", "item_id": "msg_empty", "output_index": 0, "content_index": 0, "text": ""}), - json!({"type": "response.content_part.done", "item_id": "msg_empty", "output_index": 0, "content_index": 0, - "part": {"type": "output_text", "text": "", "annotations": []}}), - json!({"type": "response.output_item.done", "output_index": 0, "item": done}), - ]; - (events, done) -} - -/// The tool calls a stream finished, as `(type, name, input or arguments)`. -fn finished_tool_calls(events: &[Value]) -> Vec<(String, String, String)> { - events - .iter() - .filter(|event| event["type"] == "response.output_item.done") - .map(|event| &event["item"]) - .filter(|item| { - matches!( - item["type"].as_str(), - Some("custom_tool_call" | "function_call") - ) - }) - .map(|item| { - let value = item["input"].as_str().or(item["arguments"].as_str()); - ( - item["type"].as_str().unwrap_or_default().to_string(), - item["name"].as_str().unwrap_or_default().to_string(), - value.unwrap_or_default().to_string(), - ) - }) - .collect() -} - -/// An expected finished tool call: `(type, name, input or arguments)`. -type ToolCall<'a> = (&'a str, &'a str, &'a str); - /// Codex showed "Reconnecting... (stream closed before response.completed)" after every tool /// call because the server ended the stream after the gateway's freeform call. #[tokio::test] async fn codex_tool_call_streams_reach_response_completed() -> TestResult { let gateway = Gateway::start().await?; let app = app(&gateway)?; - let cases: [(&str, &[ToolCall]); 4] = [ - ("custom", &[("custom_tool_call", "exec", "ls")]), - ( - "two-custom", - &[ - ("custom_tool_call", "exec", "ls"), - ("custom_tool_call", "exec", "pwd"), - ], - ), - ( - "custom-and-function", - &[ - ("custom_tool_call", "exec", "ls"), - ("function_call", "shell", r#"{"cmd": "pwd"}"#), - ], - ), - // OpenAI streams freeform input with its own events; it must keep working. - ("native-custom", &[("custom_tool_call", "exec", "ls\n")]), - ]; for route in CLAUDE_ROUTES { - for (scenario, expected) in cases { - let case = format!("{route} {scenario}"); - let (status, body) = - post_json(&app, "/v1/responses", &codex_turn(route, scenario, true)).await?; - assert_eq!(status, StatusCode::OK, "{case}: {body}"); - let events = sse_events(&body); - let types = event_types(&events); - assert_eq!( - types.last(), - Some(&"response.completed"), - "{case}: {types:?}" - ); - assert!(!types.contains(&"error"), "{case}: {types:?}"); - let expected: Vec<_> = expected - .iter() - .map(|(kind, name, value)| (kind.to_string(), name.to_string(), value.to_string())) - .collect(); - assert_eq!(finished_tool_calls(&events), expected, "{case}"); - assert_eq!( - gateway.only_call().await?["model"], - "model/claude", - "{case}" - ); - } + let (status, body) = + post_responses(&app, &codex_turn(route, "custom-and-function")).await?; + assert_eq!(status, StatusCode::OK, "{route}: {body}"); + let events = sse_events(&body); + let types = event_types(&events); + assert_eq!( + types.last(), + Some(&"response.completed"), + "{route}: {types:?}" + ); + let calls: Vec<_> = events + .iter() + .filter(|event| event["type"] == "response.output_item.done") + .map(|event| &event["item"]) + .filter(|item| item["type"] != "message") + .map(|item| { + ( + &item["name"], + item.get("input").unwrap_or(&item["arguments"]), + ) + }) + .collect(); + assert_eq!( + calls, + [ + (&json!("exec"), &json!("ls")), + (&json!("shell"), &json!(r#"{"cmd": "pwd"}"#)) + ], + "{route}" + ); + assert_eq!( + gateway.only_call().await?["model"], + "model/claude", + "{route}" + ); } Ok(()) } -/// A finished item that contradicts content that was already streamed ends the stream with an -/// error event in its place. The contradicting event must not be forwarded as if it were the -/// error, which ended the stream with no error event and no log. +/// A finished item that contradicts text that was already streamed ends the stream with an +/// error event in its place. Before the fix the stream ended with no error event. #[tokio::test] async fn contradicting_snapshots_end_the_stream_with_an_error_event() -> TestResult { let gateway = Gateway::start().await?; let app = app(&gateway)?; for route in CLAUDE_ROUTES { - for scenario in ["conflicting-arguments", "conflicting-text"] { - let case = format!("{route} {scenario}"); - let (status, body) = - post_json(&app, "/v1/responses", &codex_turn(route, scenario, true)).await?; - assert_eq!(status, StatusCode::OK, "{case}: {body}"); - let events = sse_events(&body); - let types = event_types(&events); - let last = events.last().ok_or("the stream must not be empty")?; - assert_eq!(last["type"], "error", "{case}: {types:?}"); - assert_eq!(last["message"], CONFLICT_MESSAGE, "{case}"); - // The contradicting item is replaced by the error, and nothing follows it. - assert!( - !types.contains(&"response.output_item.done"), - "{case}: {types:?}" - ); - assert!(!types.contains(&"response.completed"), "{case}: {types:?}"); - gateway.only_call().await?; - } - } - Ok(()) -} - -/// Codex sends no `max_output_tokens`, so the gateway stopped Sonnet at 4,096 tokens and Codex -/// ended long turns with an empty reply. A target's `extra_body` limit lifts the cap without -/// overriding a limit the caller sets. -#[tokio::test] -async fn target_output_limit_lifts_the_gateway_default_cap() -> TestResult { - let gateway = Gateway::start().await?; - let app = app(&gateway)?; - - // Without a configured limit the gateway cuts the answer off, and the client receives - // `response.incomplete`. - let (status, body) = post_json( - &app, - "/v1/responses", - &codex_turn("gateway/uncapped", "long-answer", true), - ) - .await?; - assert_eq!(status, StatusCode::OK, "{body}"); - let events = sse_events(&body); - assert_eq!( - event_types(&events).last(), - Some(&"response.incomplete"), - "{body}" - ); - assert!( - gateway - .only_call() - .await? - .get("max_output_tokens") - .is_none() - ); - - for route in CLAUDE_ROUTES { - for stream in [true, false] { - let case = format!("{route} stream={stream}"); - let (status, body) = post_json( - &app, - "/v1/responses", - &codex_turn(route, "long-answer", stream), - ) - .await?; - assert_eq!(status, StatusCode::OK, "{case}: {body}"); - if stream { - let events = sse_events(&body); - assert_eq!( - event_types(&events).last(), - Some(&"response.completed"), - "{case}: {body}" - ); - } else { - assert_eq!( - serde_json::from_str::(&body)?["status"], - "completed", - "{case}" - ); - } - let call = gateway.only_call().await?; - assert_eq!(call["max_output_tokens"], CONFIGURED_OUTPUT_LIMIT, "{case}"); - } - - let mut limited = codex_turn(route, "long-answer", true); - limited["max_output_tokens"] = json!(1000); - let (status, body) = post_json(&app, "/v1/responses", &limited).await?; + let (status, body) = post_responses(&app, &codex_turn(route, "conflicting-text")).await?; assert_eq!(status, StatusCode::OK, "{route}: {body}"); + let events = sse_events(&body); + let types = event_types(&events); + assert_eq!(types.last(), Some(&"error"), "{route}: {types:?}"); assert_eq!( - gateway.only_call().await?["max_output_tokens"], - 1000, + events[events.len() - 1]["message"], + "Responses snapshot conflicts with streamed content", "{route}" ); + assert!( + !types.contains(&"response.output_item.done"), + "{route}: {types:?}" + ); + gateway.only_call().await?; } Ok(()) } -/// The `(type, name)` of each tool definition in a gateway request, for either OpenAI API. -fn defined_tools(call: &Value) -> Vec<(String, String)> { - call["tools"] - .as_array() - .into_iter() - .flatten() - .map(|tool| { - let name = tool["name"].as_str().or(tool["function"]["name"].as_str()); - ( - tool["type"].as_str().unwrap_or_default().to_string(), - name.unwrap_or_default().to_string(), - ) - }) - .collect() -} - -fn codex_compaction(route: &str, stream: bool) -> Value { - json!({ - "model": route, - "stream": stream, - "tools": [], - "tool_choice": "auto", - "parallel_tool_calls": false, - "input": [ - user_text("List files"), - {"type": "custom_tool_call", "call_id": "call_1", "name": "exec", "input": "ls"}, - {"type": "custom_tool_call_output", "call_id": "call_1", "output": "README.md"}, - {"type": "function_call", "call_id": "call_2", "name": "shell", "arguments": "{}"}, - {"type": "function_call_output", "call_id": "call_2", "output": "ok"}, - {"type": "function_call", "call_id": "call_3", "name": "shell", "arguments": "{}"}, - {"type": "function_call_output", "call_id": "call_3", "output": "ok"}, - {"type": "function_call", "call_id": "call_4", "namespace": "collaboration", - "name": "spawn_agent", "arguments": "{}"}, - {"type": "function_call_output", "call_id": "call_4", "output": "ok"}, - user_text("Summarize the conversation.") - ] - }) -} - /// Codex compacts a long thread by sending its tool-call history with an empty `tools` list. /// The gateway rejected those requests with HTTP 400, so compaction never finished. #[tokio::test] async fn tool_history_without_tools_gets_definitions_the_gateway_accepts() -> TestResult { let gateway = Gateway::start().await?; let app = app(&gateway)?; - let codex_tools = [ - ("custom".to_string(), "exec".to_string()), - ("function".to_string(), "shell".to_string()), - ( - "function".to_string(), - "collaboration__spawn_agent".to_string(), - ), - ]; for route in CLAUDE_ROUTES { for stream in [true, false] { - let case = format!("{route} compaction stream={stream}"); - let (status, body) = - post_json(&app, "/v1/responses", &codex_compaction(route, stream)).await?; + let case = format!("{route} stream={stream}"); + let compaction = json!({ + "model": route, + "stream": stream, + "tools": [], + "tool_choice": "auto", + "input": [ + user_text("List files"), + {"type": "custom_tool_call", "call_id": "call_1", "name": "exec", "input": "ls"}, + {"type": "custom_tool_call_output", "call_id": "call_1", "output": "README.md"}, + {"type": "function_call", "call_id": "call_2", "namespace": "collaboration", + "name": "spawn_agent", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_2", "output": "ok"}, + user_text("Summarize the conversation.") + ] + }); + let (status, body) = post_responses(&app, &compaction).await?; assert_eq!(status, StatusCode::OK, "{case}: {body}"); - if stream { - let types = event_types(&sse_events(&body)).join(","); - assert!(types.ends_with("response.completed"), "{case}: {types}"); - } let call = gateway.only_call().await?; - assert_eq!(call["model"], "model/claude", "{case}"); - assert_eq!(defined_tools(&call), codex_tools, "{case}"); + let tools: Vec<_> = call["tools"] + .as_array() + .into_iter() + .flatten() + .map(|tool| (&tool["type"], &tool["name"])) + .collect(); + assert_eq!( + tools, + [ + (&json!("custom"), &json!("exec")), + (&json!("function"), &json!("collaboration__spawn_agent")) + ], + "{case}" + ); assert_eq!(call["tool_choice"], "none", "{case}"); } - - // Anthropic Messages clients such as Claude Code reach the same target through - // translation. - let messages = json!({ - "model": route, - "max_tokens": 300, - "messages": [ - {"role": "user", "content": "List files"}, - {"role": "assistant", "content": [{"type": "tool_use", "id": "toolu_1", "name": "Bash", "input": {"command": "ls"}}]}, - {"role": "user", "content": [ - {"type": "tool_result", "tool_use_id": "toolu_1", "content": "README.md"}, - {"type": "text", "text": "Summarize the conversation."} - ]} - ] - }); - let (status, body) = post_json(&app, "/v1/messages", &messages).await?; - assert_eq!(status, StatusCode::OK, "{route} messages: {body}"); - let call = gateway.only_call().await?; - assert_eq!( - defined_tools(&call), - [("function".to_string(), "Bash".to_string())], - "{route} messages" - ); - assert_eq!(call["tool_choice"], "none", "{route} messages"); - } - - let chat = json!({ - "model": "gateway/chat", - "messages": [ - {"role": "user", "content": "List files"}, - {"role": "assistant", "content": null, "tool_calls": [ - {"id": "call_1", "type": "function", "function": {"name": "shell", "arguments": "{}"}}, - {"id": "call_2", "type": "function", "function": {"name": "shell", "arguments": "{}"}} - ]}, - {"role": "tool", "tool_call_id": "call_1", "content": "README.md"}, - {"role": "tool", "tool_call_id": "call_2", "content": "README.md"}, - {"role": "user", "content": "Summarize the conversation."} - ] - }); - let (status, body) = post_json(&app, "/v1/chat/completions", &chat).await?; - assert_eq!(status, StatusCode::OK, "chat: {body}"); - let call = gateway.only_call().await?; - assert_eq!( - defined_tools(&call), - [("function".to_string(), "shell".to_string())] - ); - assert_eq!(call["tool_choice"], "none"); - Ok(()) -} - -/// Requests that define their own tools, or that call no tools, reach the gateway unchanged. -#[tokio::test] -async fn requests_with_their_own_tools_or_no_tool_history_are_unchanged() -> TestResult { - let gateway = Gateway::start().await?; - let app = app(&gateway)?; - for route in CLAUDE_ROUTES { - let mut defined = codex_compaction(route, true); - defined["tools"] = json!([{"type": "function", "name": "shell", "description": "Run a command", - "parameters": {"type": "object"}}]); - let (status, body) = post_json(&app, "/v1/responses", &defined).await?; - assert_eq!(status, StatusCode::OK, "{route} defined: {body}"); - let call = gateway.only_call().await?; - assert_eq!(call["tools"], defined["tools"], "{route} defined"); - assert_eq!(call["tool_choice"], "auto", "{route} defined"); - - // Codex's Responses-lite requests carry their tools inside `input`. - let mut lite = codex_compaction(route, true); - let lite_tools = json!({"type": "additional_tools", "role": "developer", - "tools": [{"type": "function", "name": "shell", "parameters": {"type": "object"}}]}); - lite.as_object_mut() - .ok_or("request must be an object")? - .remove("tools"); - lite["input"] - .as_array_mut() - .ok_or("input must be an array")? - .insert(0, lite_tools.clone()); - let (status, body) = post_json(&app, "/v1/responses", &lite).await?; - assert_eq!(status, StatusCode::OK, "{route} lite: {body}"); - let call = gateway.only_call().await?; - assert!(call.get("tools").is_none(), "{route} lite: {call}"); - assert_eq!(call["input"][0], lite_tools, "{route} lite"); - assert_eq!(call["tool_choice"], "auto", "{route} lite"); - - let mut plain = codex_turn(route, "ok", true); - plain["tools"] = json!([]); - let (status, body) = post_json(&app, "/v1/responses", &plain).await?; - assert_eq!(status, StatusCode::OK, "{route} plain: {body}"); - let call = gateway.only_call().await?; - assert_eq!(defined_tools(&call), [], "{route} plain"); - assert_eq!(call["tool_choice"], "auto", "{route} plain"); } Ok(()) } From 4982c4e063e54113b66680e5837c2a674ab9d24f Mon Sep 17 00:00:00 2001 From: Elyas Mehtabuddin Date: Tue, 13 Oct 2026 11:18:32 -0700 Subject: [PATCH 3/5] refactor: simplify replayed tool definitions, custom-call decoding, and gateway tests Signed-off-by: Elyas Mehtabuddin --- crates/libsy-llm-client/src/client.rs | 146 +++++++----------- crates/switchyard-server/src/sse.rs | 3 +- .../tests/litellm_gateway.rs | 113 ++++++-------- .../src/codecs/responses/stream.rs | 41 ++--- .../tests/stream_translation.rs | 29 +--- 5 files changed, 136 insertions(+), 196 deletions(-) diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 291a8cb80f..9718636175 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -1078,21 +1078,16 @@ fn ensure_responses_function_tool_description(object: &mut Map) { } /// Description of a tool definition that Switchyard adds only so that earlier calls to the -/// tool stay valid. The same request sets `tool_choice` to `none`. +/// tool stay valid. const REPLAYED_TOOL_DESCRIPTION: &str = "Earlier calls to this tool appear in the conversation. It is not available in this request."; -// Adds a definition for each tool that the conversation history calls when the request -// defines no tools, and sets `tool_choice` to `none`. -// -// Codex compacts a long conversation by sending the whole history, including its tool calls -// and results, with an empty `tools` list. OpenAI accepts that. LiteLLM in front of Bedrock -// Claude rejects it on both OpenAI APIs ("Bedrock doesn't support tool calling without -// `tools=` param specified"), so compaction could never finish on such a target. A -// same-format request is replayed verbatim, so the fix has to apply to the outbound body. -// -// Bedrock has no `tool_choice` of `none`, and LiteLLM drops it. A Bedrock model can therefore -// still call one of these tools if the prompt asks it to. +// When a request defines no tools, this adds a definition for each tool that its history +// calls and sets `tool_choice` to `none`. Codex compacts a thread by resending its tool-call +// history with an empty `tools` list, and LiteLLM in front of Bedrock Claude rejects that on +// both OpenAI APIs. This edits the outbound body because a same-format request is otherwise +// sent verbatim. Bedrock has no `tool_choice` of `none` and LiteLLM drops it, so a Bedrock +// model can still call one of these tools if the prompt asks it to. fn define_tools_for_replayed_calls(body: &mut Value, wire_format: WireFormat) { let Value::Object(object) = body else { return; @@ -1104,86 +1099,61 @@ fn define_tools_for_replayed_calls(body: &mut Value, wire_format: WireFormat) { { return; } - let tools = match wire_format { - WireFormat::OpenAiResponses => replayed_responses_tools(object.get("input")), - WireFormat::OpenAiChat => replayed_chat_tools(object.get("messages")), - WireFormat::AnthropicMessages => return, - }; - if tools.is_empty() { - return; - } - object.insert("tools".to_string(), Value::Array(tools)); - object.insert("tool_choice".to_string(), json!("none")); -} - -// One definition per tool name called in Responses `input`, in first-call order. A freeform -// `custom_tool_call` gets a `custom` definition and a `function_call` gets a `function` one. -fn replayed_responses_tools(input: Option<&Value>) -> Vec { - let Some(Value::Array(items)) = input else { - return Vec::new(); - }; - // A Responses-lite request defines its tools inside `input`. - if items - .iter() - .any(|item| item.get("type").and_then(Value::as_str) == Some("additional_tools")) - { - return Vec::new(); - } - let mut names = HashSet::new(); - let mut tools = Vec::new(); - for item in items { - let kind = match item.get("type").and_then(Value::as_str) { - Some("function_call") => "function", - Some("custom_tool_call") => "custom", - _ => continue, - }; - let Some(name) = item.get("name").and_then(Value::as_str) else { - continue; - }; - if !names.insert(name) { - continue; + let mut seen = HashSet::new(); + let tools: Vec = match wire_format { + WireFormat::OpenAiResponses => { + let items = object + .get("input") + .and_then(Value::as_array) + .map(Vec::as_slice) + .unwrap_or_default(); + // A Responses-lite request defines its tools inside `input`. + if items.iter().any(|item| item["type"] == "additional_tools") { + return; + } + items + .iter() + .filter_map(|item| Some((item.get("type")?.as_str()?, item.get("name")?.as_str()?))) + .filter(|(kind, name)| { + matches!(*kind, "function_call" | "custom_tool_call") && seen.insert(*name) + }) + .map(|(kind, name)| match kind { + "function_call" => json!({ + "type": "function", + "name": name, + "description": REPLAYED_TOOL_DESCRIPTION, + "parameters": {"type": "object"}, + }), + _ => json!({"type": "custom", "name": name, "description": REPLAYED_TOOL_DESCRIPTION}), + }) + .collect() } - tools.push(if kind == "function" { - json!({ - "type": "function", - "name": name, - "description": REPLAYED_TOOL_DESCRIPTION, - "parameters": {"type": "object"}, + WireFormat::OpenAiChat => object + .get("messages") + .and_then(Value::as_array) + .into_iter() + .flatten() + .filter_map(|message| message.get("tool_calls")?.as_array()) + .flatten() + .filter_map(|call| call.pointer("/function/name")?.as_str()) + .filter(|name| seen.insert(*name)) + .map(|name| { + json!({ + "type": "function", + "function": { + "name": name, + "description": REPLAYED_TOOL_DESCRIPTION, + "parameters": {"type": "object"}, + }, + }) }) - } else { - json!({"type": "custom", "name": name, "description": REPLAYED_TOOL_DESCRIPTION}) - }); - } - tools -} - -// One function definition per tool name called by assistant messages, in first-call order. -fn replayed_chat_tools(messages: Option<&Value>) -> Vec { - let Some(Value::Array(messages)) = messages else { - return Vec::new(); + .collect(), + WireFormat::AnthropicMessages => return, }; - let mut names = HashSet::new(); - let mut tools = Vec::new(); - let calls = messages - .iter() - .filter_map(|message| message.get("tool_calls").and_then(Value::as_array)) - .flatten(); - for call in calls { - let Some(name) = call.pointer("/function/name").and_then(Value::as_str) else { - continue; - }; - if names.insert(name) { - tools.push(json!({ - "type": "function", - "function": { - "name": name, - "description": REPLAYED_TOOL_DESCRIPTION, - "parameters": {"type": "object"}, - }, - })); - } + if !tools.is_empty() { + object.insert("tools".to_string(), Value::Array(tools)); + object.insert("tool_choice".to_string(), json!("none")); } - tools } // Drops fields accepted by OpenAI-like APIs but rejected by Anthropic Messages. diff --git a/crates/switchyard-server/src/sse.rs b/crates/switchyard-server/src/sse.rs index c6549eff16..b7794309e1 100644 --- a/crates/switchyard-server/src/sse.rs +++ b/crates/switchyard-server/src/sse.rs @@ -39,8 +39,7 @@ pub(crate) fn frame_stream( // Preserve the upstream error's fields, apart from credential redaction, // rather than replacing it with a synthesized error. Err(LlmStreamError::Upstream(value)) => { - // The error message can quote request content, so the log records only - // the event type. + // Log only the event type; the error message can quote request content. let event_type = value.get("type").and_then(Value::as_str).unwrap_or("error"); tracing::warn!(event_type, "stream ended with an in-band error event"); failed = true; diff --git a/crates/switchyard-server/tests/litellm_gateway.rs b/crates/switchyard-server/tests/litellm_gateway.rs index ced8e109ed..30b971f701 100644 --- a/crates/switchyard-server/tests/litellm_gateway.rs +++ b/crates/switchyard-server/tests/litellm_gateway.rs @@ -173,21 +173,19 @@ async fn gateway(State(calls): State>>>, Json(body): Json { - // The gateway also closes an empty text message at output index 0 that it never - // announced. - let (custom_events, custom) = litellm_custom_call(1, "ls"); - let arguments = r#"{"cmd": "pwd"}"#; - let (function_events, function) = function_call(2, arguments); - let (empty_events, empty) = text_message(0, None, ""); - let events = [custom_events, function_events, empty_events].concat(); - (events, vec![empty, custom, function]) + let (custom_events, custom) = litellm_custom_call(0, "ls"); + let (function_events, function) = function_call(1, r#"{"cmd": "pwd"}"#); + ( + [custom_events, function_events].concat(), + vec![custom, function], + ) } text if text.contains("conflicting-text") => { - let (events, message) = text_message(0, Some("Hello"), "Goodbye"); + let (events, message) = text_message(0, "Hello", "Goodbye"); (events, vec![message]) } _ => { - let (events, message) = text_message(0, Some("ok"), "ok"); + let (events, message) = text_message(0, "ok", "ok"); (events, vec![message]) } }; @@ -195,9 +193,6 @@ async fn gateway(State(calls): State>>>, Json(body): Json (Vec, Value) { } /// A text message whose streamed text is `streamed` and whose finished item says `done_text`. -/// Without `streamed`, the message is only closed, as the gateway does after a tool call. -fn text_message(index: u64, streamed: Option<&str>, done_text: &str) -> (Vec, Value) { +fn text_message(index: u64, streamed: &str, done_text: &str) -> (Vec, Value) { let done = json!({"type": "message", "id": "msg_1", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": done_text, "annotations": []}]}); - let mut events = Vec::new(); - if let Some(streamed) = streamed { - events.extend([ - json!({"type": "response.output_item.added", "output_index": index, - "item": {"type": "message", "id": "msg_1", "role": "assistant", "content": []}}), - json!({"type": "response.output_text.delta", "item_id": "msg_1", "output_index": index, "content_index": 0, "delta": streamed}), - ]); - } - events.extend([ + let events = vec![ + json!({"type": "response.output_item.added", "output_index": index, + "item": {"type": "message", "id": "msg_1", "role": "assistant", "content": []}}), + json!({"type": "response.output_text.delta", "item_id": "msg_1", "output_index": index, "content_index": 0, "delta": streamed}), json!({"type": "response.output_text.done", "item_id": "msg_1", "output_index": index, "content_index": 0, "text": done_text}), json!({"type": "response.output_item.done", "output_index": index, "item": done}), - ]); + ]; (events, done) } @@ -280,9 +269,8 @@ async fn codex_tool_call_streams_reach_response_completed() -> TestResult { let calls: Vec<_> = events .iter() .filter(|event| event["type"] == "response.output_item.done") - .map(|event| &event["item"]) - .filter(|item| item["type"] != "message") - .map(|item| { + .map(|event| { + let item = &event["item"]; ( &item["name"], item.get("input").unwrap_or(&item["arguments"]), @@ -339,42 +327,39 @@ async fn tool_history_without_tools_gets_definitions_the_gateway_accepts() -> Te let gateway = Gateway::start().await?; let app = app(&gateway)?; for route in CLAUDE_ROUTES { - for stream in [true, false] { - let case = format!("{route} stream={stream}"); - let compaction = json!({ - "model": route, - "stream": stream, - "tools": [], - "tool_choice": "auto", - "input": [ - user_text("List files"), - {"type": "custom_tool_call", "call_id": "call_1", "name": "exec", "input": "ls"}, - {"type": "custom_tool_call_output", "call_id": "call_1", "output": "README.md"}, - {"type": "function_call", "call_id": "call_2", "namespace": "collaboration", - "name": "spawn_agent", "arguments": "{}"}, - {"type": "function_call_output", "call_id": "call_2", "output": "ok"}, - user_text("Summarize the conversation.") - ] - }); - let (status, body) = post_responses(&app, &compaction).await?; - assert_eq!(status, StatusCode::OK, "{case}: {body}"); - let call = gateway.only_call().await?; - let tools: Vec<_> = call["tools"] - .as_array() - .into_iter() - .flatten() - .map(|tool| (&tool["type"], &tool["name"])) - .collect(); - assert_eq!( - tools, - [ - (&json!("custom"), &json!("exec")), - (&json!("function"), &json!("collaboration__spawn_agent")) - ], - "{case}" - ); - assert_eq!(call["tool_choice"], "none", "{case}"); - } + let compaction = json!({ + "model": route, + "stream": true, + "tools": [], + "tool_choice": "auto", + "input": [ + user_text("List files"), + {"type": "custom_tool_call", "call_id": "call_1", "name": "exec", "input": "ls"}, + {"type": "custom_tool_call_output", "call_id": "call_1", "output": "README.md"}, + {"type": "function_call", "call_id": "call_2", "namespace": "collaboration", + "name": "spawn_agent", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_2", "output": "ok"}, + user_text("Summarize the conversation.") + ] + }); + let (status, body) = post_responses(&app, &compaction).await?; + assert_eq!(status, StatusCode::OK, "{route}: {body}"); + let call = gateway.only_call().await?; + let tools: Vec<_> = call["tools"] + .as_array() + .into_iter() + .flatten() + .map(|tool| (&tool["type"], &tool["name"])) + .collect(); + assert_eq!( + tools, + [ + (&json!("custom"), &json!("exec")), + (&json!("function"), &json!("collaboration__spawn_agent")) + ], + "{route}" + ); + assert_eq!(call["tool_choice"], "none", "{route}"); } Ok(()) } diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index d64ed93669..35f3a1411a 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -188,23 +188,29 @@ fn decode_responses_stream( event .get("delta") .and_then(Value::as_str) - .and_then(|delta| { + // The decoder reads a custom call's input only from its completed item. + .filter(|_| { + !state + .tool_states + .get(&(output_index as usize)) + .is_some_and(|tool| tool.decoded_custom_call) + }) + .map(|delta| { state.decoded_tool_call = true; - let tool = state.tool_states.entry(output_index as usize).or_default(); - // The completed item carries a custom call's input. The decoder skips - // argument deltas for that call because their JSON can differ from it. - if tool.decoded_custom_call { - return None; - } // Recorded so `response.output_item.done`, which repeats // the complete arguments, can tell it is a repeat. - tool.decoded_arguments.push_str(delta); - Some(vec![LlmResponseChunk::ToolCallDelta { + 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, name: None, arguments_delta: Some(delta.to_string()), - }]) + }] }) .unwrap_or_default() } @@ -674,10 +680,11 @@ fn decode_responses_output_item_added( let tool = state.tool_states.entry(index).or_default(); tool.has_decoded_identity |= id.is_some_and(|id| !id.is_empty()) && name.is_some_and(|name| !name.is_empty()); - tool.decoded_custom_call |= item_type == Some("custom_tool_call"); + let custom = item_type == Some("custom_tool_call"); + tool.decoded_custom_call |= custom; // 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") { + let arguments_delta = if custom { None } else { item.get("arguments") @@ -686,12 +693,7 @@ fn decode_responses_output_item_added( .map(ToOwned::to_owned) }; if let Some(arguments) = arguments_delta.as_deref() { - state - .tool_states - .entry(index) - .or_default() - .decoded_arguments - .push_str(arguments); + tool.decoded_arguments.push_str(arguments); } vec![LlmResponseChunk::ToolCallDelta { index, @@ -789,8 +791,7 @@ fn decode_responses_completed_item( } // A snapshot may extend streamed content, but cannot retract content already sent. A -// conflict means the provider sent inconsistent data, so it decodes to `DecodeError`: the -// server then replaces the event with an error frame instead of replaying it. +// conflict is invalid provider data, so it is a `DecodeError`. fn snapshot_suffix( decoded: &mut String, snapshot: &str, diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index fdd093c06e..55fb434173 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -3432,21 +3432,17 @@ fn responses_terminal_snapshots_recover_missing_output_once() -> TestResult { } // LiteLLM in front of Bedrock streams a freeform call as function-style argument deltas that -// spell `{"content": ...}`. OpenAI streams it with custom input deltas. Both must decode to -// one call whose only arguments are the finished input, and the terminal snapshot that -// repeats the call must neither conflict with it nor repeat it. +// spell `{"content": ...}`. OpenAI streams it with custom input deltas. Both must decode, +// without a conflict, to one call whose only arguments are the finished input. #[test] fn responses_custom_tool_calls_decode_from_the_finished_item() -> TestResult { let engine = TranslationEngine::default(); let source = WireFormat::OpenAiResponses; - let call = - json!({"type": "custom_tool_call", "id": "ctc_1", "call_id": "call_1", "name": "exec"}); - let mut added = call.clone(); - added["input"] = json!(""); - let mut done = call; - done["input"] = json!("ls"); - let added = json!({"type": "response.output_item.added", "output_index": 1, "item": added}); - let item_done = json!({"type": "response.output_item.done", "output_index": 1, "item": done}); + let done = json!({"type": "custom_tool_call", "id": "ctc_1", "call_id": "call_1", "name": "exec", "input": "ls"}); + let added = json!({"type": "response.output_item.added", "output_index": 1, + "item": {"type": "custom_tool_call", "id": "ctc_1", "call_id": "call_1", "name": "exec", "input": ""}}); + let item_done = + json!({"type": "response.output_item.done", "output_index": 1, "item": done.clone()}); let completed = json!({"type": "response.completed", "response": {"id": "resp_1", "status": "completed", "output": [{"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": ""}]}, done]}}); let litellm = vec![ @@ -3493,17 +3489,6 @@ fn responses_custom_tool_calls_decode_from_the_finished_item() -> TestResult { }) .collect(); assert_eq!(arguments, r#"{"input":"ls"}"#, "{shape}"); - assert!( - chunks.iter().any(|chunk| matches!(chunk, - LlmResponseChunk::ToolCallDelta { id: Some(id), name: Some(name), .. } - if id == "call_1" && name == "exec")), - "{shape}: {chunks:?}" - ); - assert!( - chunks.iter().any(|chunk| matches!(chunk, - LlmResponseChunk::MessageStop { reason: Some(reason) } if reason == "tool_use")), - "{shape}: {chunks:?}" - ); } Ok(()) } From 12f2c30488b1a0d004e4453a521403511e75a6f1 Mon Sep 17 00:00:00 2001 From: Elyas Mehtabuddin Date: Tue, 13 Oct 2026 11:23:57 -0700 Subject: [PATCH 4/5] test: move the snapshot-conflict check to the translation tests Signed-off-by: Elyas Mehtabuddin --- crates/libsy-llm-client/src/client.rs | 4 +- .../tests/litellm_gateway.rs | 47 +++---------------- .../tests/stream_translation.rs | 33 +++++++++++++ 3 files changed, 41 insertions(+), 43 deletions(-) diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 9718636175..97f30c3419 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -4,7 +4,7 @@ //! [`TranslatingLlmClient`]: encode a neutral //! request, call the configured backend over HTTP, decode the neutral response. -use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap}; use std::future::ready; use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, SystemTime}; @@ -1099,7 +1099,7 @@ fn define_tools_for_replayed_calls(body: &mut Value, wire_format: WireFormat) { { return; } - let mut seen = HashSet::new(); + let mut seen = BTreeSet::new(); let tools: Vec = match wire_format { WireFormat::OpenAiResponses => { let items = object diff --git a/crates/switchyard-server/tests/litellm_gateway.rs b/crates/switchyard-server/tests/litellm_gateway.rs index 30b971f701..552b28401d 100644 --- a/crates/switchyard-server/tests/litellm_gateway.rs +++ b/crates/switchyard-server/tests/litellm_gateway.rs @@ -180,12 +180,8 @@ async fn gateway(State(calls): State>>>, Json(body): Json { - let (events, message) = text_message(0, "Hello", "Goodbye"); - (events, vec![message]) - } _ => { - let (events, message) = text_message(0, "ok", "ok"); + let (events, message) = text_message(0, "ok"); (events, vec![message]) } }; @@ -235,15 +231,15 @@ fn function_call(index: u64, arguments: &str) -> (Vec, Value) { (events, done) } -/// A text message whose streamed text is `streamed` and whose finished item says `done_text`. -fn text_message(index: u64, streamed: &str, done_text: &str) -> (Vec, Value) { +/// A text message that streams `text` in one delta. +fn text_message(index: u64, text: &str) -> (Vec, Value) { let done = json!({"type": "message", "id": "msg_1", "role": "assistant", "status": "completed", - "content": [{"type": "output_text", "text": done_text, "annotations": []}]}); + "content": [{"type": "output_text", "text": text, "annotations": []}]}); let events = vec![ json!({"type": "response.output_item.added", "output_index": index, "item": {"type": "message", "id": "msg_1", "role": "assistant", "content": []}}), - json!({"type": "response.output_text.delta", "item_id": "msg_1", "output_index": index, "content_index": 0, "delta": streamed}), - json!({"type": "response.output_text.done", "item_id": "msg_1", "output_index": index, "content_index": 0, "text": done_text}), + json!({"type": "response.output_text.delta", "item_id": "msg_1", "output_index": index, "content_index": 0, "delta": text}), + json!({"type": "response.output_text.done", "item_id": "msg_1", "output_index": index, "content_index": 0, "text": text}), json!({"type": "response.output_item.done", "output_index": index, "item": done}), ]; (events, done) @@ -285,37 +281,6 @@ async fn codex_tool_call_streams_reach_response_completed() -> TestResult { ], "{route}" ); - assert_eq!( - gateway.only_call().await?["model"], - "model/claude", - "{route}" - ); - } - Ok(()) -} - -/// A finished item that contradicts text that was already streamed ends the stream with an -/// error event in its place. Before the fix the stream ended with no error event. -#[tokio::test] -async fn contradicting_snapshots_end_the_stream_with_an_error_event() -> TestResult { - let gateway = Gateway::start().await?; - let app = app(&gateway)?; - for route in CLAUDE_ROUTES { - let (status, body) = post_responses(&app, &codex_turn(route, "conflicting-text")).await?; - assert_eq!(status, StatusCode::OK, "{route}: {body}"); - let events = sse_events(&body); - let types = event_types(&events); - assert_eq!(types.last(), Some(&"error"), "{route}: {types:?}"); - assert_eq!( - events[events.len() - 1]["message"], - "Responses snapshot conflicts with streamed content", - "{route}" - ); - assert!( - !types.contains(&"response.output_item.done"), - "{route}: {types:?}" - ); - gateway.only_call().await?; } Ok(()) } diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 55fb434173..e4dd05f87b 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -203,6 +203,39 @@ fn preserved_same_format_replay_stops_after_an_error() -> TestResult { Ok(()) } +// A finished item that contradicts text that was already streamed reaches a same-format client +// as an error event in its place. Before, the contradicting event was replayed and the stream +// ended with no error event. +#[test] +fn preserved_same_format_snapshot_conflict_replays_as_an_error() -> TestResult { + let format = WireFormat::OpenAiResponses; + let events = [ + json!({"type": "response.output_text.delta", "output_index": 0, "content_index": 0, "delta": "Hello"}), + json!({"type": "response.output_item.done", "output_index": 0, "item": {"type": "message", + "role": "assistant", "content": [{"type": "output_text", "text": "Goodbye"}]}}), + ]; + let engine = TranslationEngine::default(); + let mut decode_state = StreamTranslationState::new(format, format); + let mut encode_state = StreamTranslationState::new(format, format); + let mut replayed = Vec::new(); + for event in events { + let preserved = engine.decode_stream_event(&mut decode_state, format, event)?; + replayed.extend(engine.encode_stream_event(&mut encode_state, format, preserved)?); + } + + let types: Vec<_> = replayed.iter().map(|event| &event["type"]).collect(); + assert_eq!( + types, + [&json!("response.output_text.delta"), &json!("error")], + "{replayed:?}" + ); + assert_eq!( + replayed[1]["message"], + "Responses snapshot conflicts with streamed content" + ); + Ok(()) +} + #[test] fn anthropic_errors_include_type_and_terminate_stream() -> TestResult { let engine = TranslationEngine::default(); From 4f46860fc361d30ab56194b9829d5625f3bfc43c Mon Sep 17 00:00:00 2001 From: Elyas Mehtabuddin Date: Wed, 14 Oct 2026 17:38:24 -0700 Subject: [PATCH 5/5] fix: stop adding tool definitions to requests that replay tool calls Signed-off-by: Elyas Mehtabuddin --- crates/libsy-llm-client/src/client.rs | 145 ------------------ .../tests/litellm_gateway.rs | 141 ++--------------- 2 files changed, 17 insertions(+), 269 deletions(-) diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 97f30c3419..e1976de291 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -277,9 +277,6 @@ impl TranslatingLlmClient { if matches!(backend, Backend::OpenAiResponses(_)) { sanitize_openai_responses_provider_body(&mut body); } - if matches!(endpoint, UpstreamEndpoint::Completion) { - define_tools_for_replayed_calls(&mut body, wire_format); - } // Strip before `merge_extra_body` so a target can reinstate either field // deliberately via `extra_body`. if matches!(backend, Backend::Anthropic(_)) { @@ -1077,85 +1074,6 @@ fn ensure_responses_function_tool_description(object: &mut Map) { } } -/// Description of a tool definition that Switchyard adds only so that earlier calls to the -/// tool stay valid. -const REPLAYED_TOOL_DESCRIPTION: &str = - "Earlier calls to this tool appear in the conversation. It is not available in this request."; - -// When a request defines no tools, this adds a definition for each tool that its history -// calls and sets `tool_choice` to `none`. Codex compacts a thread by resending its tool-call -// history with an empty `tools` list, and LiteLLM in front of Bedrock Claude rejects that on -// both OpenAI APIs. This edits the outbound body because a same-format request is otherwise -// sent verbatim. Bedrock has no `tool_choice` of `none` and LiteLLM drops it, so a Bedrock -// model can still call one of these tools if the prompt asks it to. -fn define_tools_for_replayed_calls(body: &mut Value, wire_format: WireFormat) { - let Value::Object(object) = body else { - return; - }; - if object - .get("tools") - .and_then(Value::as_array) - .is_some_and(|tools| !tools.is_empty()) - { - return; - } - let mut seen = BTreeSet::new(); - let tools: Vec = match wire_format { - WireFormat::OpenAiResponses => { - let items = object - .get("input") - .and_then(Value::as_array) - .map(Vec::as_slice) - .unwrap_or_default(); - // A Responses-lite request defines its tools inside `input`. - if items.iter().any(|item| item["type"] == "additional_tools") { - return; - } - items - .iter() - .filter_map(|item| Some((item.get("type")?.as_str()?, item.get("name")?.as_str()?))) - .filter(|(kind, name)| { - matches!(*kind, "function_call" | "custom_tool_call") && seen.insert(*name) - }) - .map(|(kind, name)| match kind { - "function_call" => json!({ - "type": "function", - "name": name, - "description": REPLAYED_TOOL_DESCRIPTION, - "parameters": {"type": "object"}, - }), - _ => json!({"type": "custom", "name": name, "description": REPLAYED_TOOL_DESCRIPTION}), - }) - .collect() - } - WireFormat::OpenAiChat => object - .get("messages") - .and_then(Value::as_array) - .into_iter() - .flatten() - .filter_map(|message| message.get("tool_calls")?.as_array()) - .flatten() - .filter_map(|call| call.pointer("/function/name")?.as_str()) - .filter(|name| seen.insert(*name)) - .map(|name| { - json!({ - "type": "function", - "function": { - "name": name, - "description": REPLAYED_TOOL_DESCRIPTION, - "parameters": {"type": "object"}, - }, - }) - }) - .collect(), - WireFormat::AnthropicMessages => return, - }; - if !tools.is_empty() { - object.insert("tools".to_string(), Value::Array(tools)); - object.insert("tool_choice".to_string(), json!("none")); - } -} - // Drops fields accepted by OpenAI-like APIs but rejected by Anthropic Messages. // // A router can serve earlier turns of a session from an OpenAI-format target and @@ -1745,69 +1663,6 @@ mod tests { ); } - // A gateway in front of Bedrock rejects tool-call history in a request with no tools, so - // the client defines each called tool. Requests with their own tool definitions, with no - // tool calls, or bound for an Anthropic backend keep their bodies. - #[test] - fn replayed_tool_calls_get_definitions_only_when_the_request_has_none() { - let custom = - json!({"type": "custom_tool_call", "call_id": "c1", "name": "exec", "input": "ls"}); - let function = - json!({"type": "function_call", "call_id": "c2", "name": "shell", "arguments": "{}"}); - let mut responses = - json!({"tools": [], "tool_choice": "auto", "input": [custom, function, function]}); - define_tools_for_replayed_calls(&mut responses, WireFormat::OpenAiResponses); - let defined: Vec<_> = responses["tools"] - .as_array() - .into_iter() - .flatten() - .map(|tool| (&tool["type"], &tool["name"])) - .collect(); - assert_eq!( - defined, - [ - (&json!("custom"), &json!("exec")), - (&json!("function"), &json!("shell")) - ] - ); - assert_eq!(responses["tool_choice"], "none"); - - let shell = json!({"type": "function", "function": {"name": "shell", "arguments": "{}"}}); - let mut chat = json!({"messages": [{"role": "assistant", "tool_calls": [shell, shell]}]}); - define_tools_for_replayed_calls(&mut chat, WireFormat::OpenAiChat); - assert_eq!(chat["tools"].as_array().map(Vec::len), Some(1)); - assert_eq!(chat["tools"][0]["function"]["name"], "shell"); - assert_eq!(chat["tool_choice"], "none"); - - let anthropic_history = json!([ - {"role": "assistant", "content": [{"type": "tool_use", "id": "t1", "name": "Bash", "input": {}}]}, - {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1", "content": "ok"}]} - ]); - for (body, format) in [ - ( - json!({"tools": [{"type": "function", "name": "own"}], "tool_choice": "auto", "input": [function]}), - WireFormat::OpenAiResponses, - ), - // A Responses-lite request defines its tools inside `input`. - ( - json!({"input": [{"type": "additional_tools", "tools": []}, function]}), - WireFormat::OpenAiResponses, - ), - ( - json!({"tools": [], "input": [{"type": "message", "role": "user", "content": "hi"}]}), - WireFormat::OpenAiResponses, - ), - ( - json!({"messages": anthropic_history}), - WireFormat::AnthropicMessages, - ), - ] { - let mut edited = body.clone(); - define_tools_for_replayed_calls(&mut edited, format); - assert_eq!(edited, body, "{format}"); - } - } - // A request that pins `format` in its metadata, so the client resolves that // wire format instead of the model's default backend. fn request_with_wire_format(model: &str, format: WireFormat) -> Request { diff --git a/crates/switchyard-server/tests/litellm_gateway.rs b/crates/switchyard-server/tests/litellm_gateway.rs index 552b28401d..a83d00998f 100644 --- a/crates/switchyard-server/tests/litellm_gateway.rs +++ b/crates/switchyard-server/tests/litellm_gateway.rs @@ -1,20 +1,17 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! Regression tests for Codex traffic sent through a LiteLLM gateway to Bedrock Claude. +//! Regression test for Codex traffic sent through a LiteLLM gateway to Bedrock Claude. //! -//! The stub gateway reproduces two behaviors of the real gateway that broke Codex: it streams -//! a freeform tool call as function-style argument deltas whose JSON differs from the -//! finished call, and it rejects tool-call history in a request that defines no tools. Each -//! test runs the production server, config loader, router, and HTTP client against that stub. +//! The stub gateway streams a freeform tool call the way the real gateway does: as +//! function-style argument deltas whose JSON differs from the finished call. The test runs the +//! production server, config loader, router, and HTTP client against that stub. use std::convert::Infallible; use std::error::Error; use std::io::Write; -use std::sync::Arc; use axum::body::Body; -use axum::extract::State; use axum::http::{Request, StatusCode}; use axum::response::sse::{Event, Sse}; use axum::response::{IntoResponse, Response}; @@ -24,40 +21,27 @@ use http_body_util::BodyExt; use serde_json::{Value, json}; use switchyard_server::{build_switchyard_router, config::load_server_state}; use tokio::net::TcpListener; -use tokio::sync::Mutex; use tokio::task::JoinHandle; use tower::ServiceExt; type TestResult = Result>; -/// Both route types send the requests in these tests to the `claude` target. +/// Both route types send the test's requests to the `claude` target. const CLAUDE_ROUTES: [&str; 2] = ["gateway/passthrough", "gateway/stage"]; // `Gateway::drop` stops the server task even when an assertion fails. struct Gateway { root: String, - calls: Arc>>, task: JoinHandle>, } impl Gateway { async fn start() -> TestResult { - let calls = Arc::new(Mutex::new(Vec::new())); - let app = Router::new() - .route("/v1/responses", post(gateway)) - .with_state(Arc::clone(&calls)); + let app = Router::new().route("/v1/responses", post(gateway)); let listener = TcpListener::bind("127.0.0.1:0").await?; let root = format!("http://{}", listener.local_addr()?); let task = tokio::spawn(async move { axum::serve(listener, app).await }); - Ok(Self { root, calls, task }) - } - - /// The single request the gateway received since the last call. - async fn only_call(&self) -> TestResult { - let calls = std::mem::take(&mut *self.calls.lock().await); - let [call] = <[Value; 1]>::try_from(calls) - .map_err(|calls| format!("expected one gateway call, got {}", calls.len()))?; - Ok(call) + Ok(Self { root, task }) } } @@ -133,12 +117,8 @@ fn event_types(events: &[Value]) -> Vec<&str> { .collect() } -fn user_text(text: &str) -> Value { - json!({"type": "message", "role": "user", "content": [{"type": "input_text", "text": text}]}) -} - -/// A streamed Codex turn whose user message names the gateway scenario. -fn codex_turn(route: &str, scenario: &str) -> Value { +/// A streamed Codex turn that offers a freeform tool and a function tool. +fn codex_turn(route: &str) -> Value { json!({ "model": route, "stream": true, @@ -148,46 +128,18 @@ fn codex_turn(route: &str, scenario: &str) -> Value { "parameters": {"type": "object", "properties": {"cmd": {"type": "string"}}}} ], "tool_choice": "auto", - "input": [user_text(scenario)] + "input": [{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "List files"}]}] }) } -/// Answers like LiteLLM in front of Bedrock Claude: it rejects tool-call history without tool -/// definitions, and otherwise replies with the scenario that the request's text names. -async fn gateway(State(calls): State>>>, Json(body): Json) -> Response { - calls.lock().await.push(body.clone()); - let input = body["input"].as_array().cloned().unwrap_or_default(); - let calls_a_tool = input.iter().any(|item| { - matches!( - item["type"].as_str(), - Some("function_call" | "custom_tool_call") - ) - }); - if calls_a_tool && body["tools"].as_array().is_none_or(Vec::is_empty) { - let message = "litellm.UnsupportedParamsError: Bedrock doesn't support tool calling without `tools=` param specified."; - return ( - StatusCode::BAD_REQUEST, - Json(json!({"error": {"message": message, "type": "invalid_request_error"}})), - ) - .into_response(); - } - let (mut events, output) = match body.to_string() { - text if text.contains("custom-and-function") => { - let (custom_events, custom) = litellm_custom_call(0, "ls"); - let (function_events, function) = function_call(1, r#"{"cmd": "pwd"}"#); - ( - [custom_events, function_events].concat(), - vec![custom, function], - ) - } - _ => { - let (events, message) = text_message(0, "ok"); - (events, vec![message]) - } - }; +/// Answers like LiteLLM in front of Bedrock Claude: a freeform call, then a function call. +async fn gateway(Json(body): Json) -> Response { + let (custom_events, custom) = litellm_custom_call(0, "ls"); + let (function_events, function) = function_call(1, r#"{"cmd": "pwd"}"#); + let mut events = [custom_events, function_events].concat(); let response = json!({ "id": "resp_gateway", "object": "response", "model": body["model"], "status": "completed", - "output": output, "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2} + "output": [custom, function], "usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2} }); let mut created = response.clone(); created["status"] = json!("in_progress"); @@ -231,20 +183,6 @@ fn function_call(index: u64, arguments: &str) -> (Vec, Value) { (events, done) } -/// A text message that streams `text` in one delta. -fn text_message(index: u64, text: &str) -> (Vec, Value) { - let done = json!({"type": "message", "id": "msg_1", "role": "assistant", "status": "completed", - "content": [{"type": "output_text", "text": text, "annotations": []}]}); - let events = vec![ - json!({"type": "response.output_item.added", "output_index": index, - "item": {"type": "message", "id": "msg_1", "role": "assistant", "content": []}}), - json!({"type": "response.output_text.delta", "item_id": "msg_1", "output_index": index, "content_index": 0, "delta": text}), - json!({"type": "response.output_text.done", "item_id": "msg_1", "output_index": index, "content_index": 0, "text": text}), - json!({"type": "response.output_item.done", "output_index": index, "item": done}), - ]; - (events, done) -} - /// Codex showed "Reconnecting... (stream closed before response.completed)" after every tool /// call because the server ended the stream after the gateway's freeform call. #[tokio::test] @@ -252,8 +190,7 @@ async fn codex_tool_call_streams_reach_response_completed() -> TestResult { let gateway = Gateway::start().await?; let app = app(&gateway)?; for route in CLAUDE_ROUTES { - let (status, body) = - post_responses(&app, &codex_turn(route, "custom-and-function")).await?; + let (status, body) = post_responses(&app, &codex_turn(route)).await?; assert_eq!(status, StatusCode::OK, "{route}: {body}"); let events = sse_events(&body); let types = event_types(&events); @@ -284,47 +221,3 @@ async fn codex_tool_call_streams_reach_response_completed() -> TestResult { } Ok(()) } - -/// Codex compacts a long thread by sending its tool-call history with an empty `tools` list. -/// The gateway rejected those requests with HTTP 400, so compaction never finished. -#[tokio::test] -async fn tool_history_without_tools_gets_definitions_the_gateway_accepts() -> TestResult { - let gateway = Gateway::start().await?; - let app = app(&gateway)?; - for route in CLAUDE_ROUTES { - let compaction = json!({ - "model": route, - "stream": true, - "tools": [], - "tool_choice": "auto", - "input": [ - user_text("List files"), - {"type": "custom_tool_call", "call_id": "call_1", "name": "exec", "input": "ls"}, - {"type": "custom_tool_call_output", "call_id": "call_1", "output": "README.md"}, - {"type": "function_call", "call_id": "call_2", "namespace": "collaboration", - "name": "spawn_agent", "arguments": "{}"}, - {"type": "function_call_output", "call_id": "call_2", "output": "ok"}, - user_text("Summarize the conversation.") - ] - }); - let (status, body) = post_responses(&app, &compaction).await?; - assert_eq!(status, StatusCode::OK, "{route}: {body}"); - let call = gateway.only_call().await?; - let tools: Vec<_> = call["tools"] - .as_array() - .into_iter() - .flatten() - .map(|tool| (&tool["type"], &tool["name"])) - .collect(); - assert_eq!( - tools, - [ - (&json!("custom"), &json!("exec")), - (&json!("function"), &json!("collaboration__spawn_agent")) - ], - "{route}" - ); - assert_eq!(call["tool_choice"], "none", "{route}"); - } - Ok(()) -}