diff --git a/crates/switchyard-translation/src/codecs/responses/buffered.rs b/crates/switchyard-translation/src/codecs/responses/buffered.rs index 025dbe35..9905aa00 100644 --- a/crates/switchyard-translation/src/codecs/responses/buffered.rs +++ b/crates/switchyard-translation/src/codecs/responses/buffered.rs @@ -327,8 +327,22 @@ fn decode_responses_input( )?; continue; }; - match item.get("type").and_then(Value::as_str) { - Some("message") => { + let item_type = match item.get("type") { + Some(Value::String(item_type)) => Some(item_type.as_str()), + Some(_) => { + return Err(TranslationError::InvalidType { + path: format!("$.input[{index}].type"), + expected: "a string", + }); + } + None => None, + }; + let is_message = item_type == Some("message") + || (item_type.is_none() + && item.contains_key("role") + && item.contains_key("content")); + match item_type { + _ if is_message => { let role = request_role_from_responses( item.get("role").and_then(Value::as_str), &format!("$.input[{index}].role"), @@ -423,6 +437,13 @@ fn decode_responses_input( is_error: None, }); } + None => { + return Err(TranslationError::InvalidValue { + path: format!("$.input[{index}].type"), + message: "missing type discriminator on a non-message input item" + .to_string(), + }); + } _ => { let message = Message { role: Role::User, diff --git a/crates/switchyard-translation/src/codecs/responses/stream.rs b/crates/switchyard-translation/src/codecs/responses/stream.rs index 8a7d324e..565ce8da 100644 --- a/crates/switchyard-translation/src/codecs/responses/stream.rs +++ b/crates/switchyard-translation/src/codecs/responses/stream.rs @@ -3,6 +3,7 @@ //! Streaming codec for OpenAI Responses API events. +use serde::Serialize; use serde_json::{Value, json}; use crate::LlmResponseChunk; @@ -34,14 +35,65 @@ impl StreamCodec for OpenAiResponsesStreamCodec { state: &mut StreamTranslationState, event: LlmResponseChunk, ) -> Vec { - encode_responses_stream(state, event) + let events = encode_responses_stream(state, event); + add_sequence_numbers(state, events) + } + + fn observe_replayed_event( + &self, + state: &mut StreamTranslationState, + raw: &Value, + normalized: Vec, + ) { + let replayed_terminal = normalized + .iter() + .any(|chunk| matches!(chunk, LlmResponseChunk::MessageStop { .. })); + // Exact replay emits `raw` once. Normalized encodings only advance codec state; + // their generated events are discarded and must not create sequence-number gaps. + for chunk in normalized { + drop(encode_responses_stream(state, chunk)); + } + state.response_sequence_number = raw + .get("sequence_number") + .and_then(Value::as_u64) + .map_or(state.response_sequence_number.saturating_add(1), |number| { + number.saturating_add(1) + }); + if replayed_terminal { + state.finished = true; + } } fn finish(&self, state: &mut StreamTranslationState) -> Vec { - finish_responses_stream(state) + let events = finish_responses_stream(state); + add_sequence_numbers(state, events) } } +/// Required fields shared by Responses stream snapshots. +#[derive(Serialize)] +struct ResponsesStreamResponse { + id: String, + object: &'static str, + created_at: u64, + completed_at: Option, + error: Option, + incomplete_details: Option, + instructions: Option, + metadata: Option, + model: String, + output: Vec, + parallel_tool_calls: bool, + frequency_penalty: Option, + presence_penalty: Option, + status: &'static str, + temperature: Option, + tool_choice: &'static str, + tools: Vec, + top_p: Option, + usage: Value, +} + // Decodes one OpenAI Responses event into neutral streaming events. fn decode_responses_stream( state: &mut StreamTranslationState, @@ -224,6 +276,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { "output_index": output_index, "item": { "type": "message", + "id": format!("msg_{output_index}"), "role": "assistant", "status": status, "content": [{"type": "output_text", "text": state.response_text}], @@ -265,6 +318,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { output_index, json!({ "type": "message", + "id": format!("msg_{output_index}"), "role": "assistant", "status": status, "content": [{"type": "output_text", "text": state.response_text}], @@ -306,15 +360,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec { out.push(json!({ "type": event_type, - "response": { - "id": responses_id(state), - "object": "response", - "status": status, - "incomplete_details": incomplete_details, - "model": target_model_or_source_model(state), - "output": output, - "usage": responses_usage_value(&state.usage), - }, + "response": responses_stream_response(state, status, incomplete_details, output), })); state.finished = true; out @@ -393,17 +439,54 @@ fn ensure_responses_created(state: &mut StreamTranslationState) -> Vec { state.response_created = true; vec![json!({ "type": "response.created", - "response": { - "id": responses_id(state), - "object": "response", - "status": "in_progress", - "model": target_model_or_source_model(state), - "output": [], - "usage": responses_usage_value(&state.usage), - }, + "response": responses_stream_response(state, "in_progress", None, Vec::new()), })] } +// Builds a schema-complete Responses snapshot for strict generated clients. +fn responses_stream_response( + state: &StreamTranslationState, + status: &'static str, + incomplete_details: Option, + output: Vec, +) -> ResponsesStreamResponse { + ResponsesStreamResponse { + id: responses_id(state), + object: "response", + created_at: 0, + completed_at: None, + error: None, + incomplete_details, + instructions: None, + metadata: None, + model: target_model_or_source_model(state), + output, + parallel_tool_calls: true, + frequency_penalty: None, + presence_penalty: None, + status, + temperature: None, + tool_choice: "auto", + tools: Vec::new(), + top_p: None, + usage: responses_usage_value(&state.usage), + } +} + +// Assigns monotonically increasing sequence numbers to generated Responses events. +fn add_sequence_numbers(state: &mut StreamTranslationState, mut events: Vec) -> Vec { + for event in &mut events { + if let Some(object) = event.as_object_mut() { + object.insert( + "sequence_number".to_string(), + Value::from(state.response_sequence_number), + ); + state.response_sequence_number = state.response_sequence_number.saturating_add(1); + } + } + events +} + // Accumulates assistant text and emits Responses text delta events. fn encode_responses_text_delta(state: &mut StreamTranslationState, text: String) -> Vec { let mut out = ensure_responses_created(state); diff --git a/crates/switchyard-translation/src/codecs/stream.rs b/crates/switchyard-translation/src/codecs/stream.rs index bcd586a2..991b4505 100644 --- a/crates/switchyard-translation/src/codecs/stream.rs +++ b/crates/switchyard-translation/src/codecs/stream.rs @@ -58,6 +58,7 @@ pub struct StreamTranslationState { pub(crate) response_reasoning_output_index: Option, pub(crate) response_reasoning_text: String, pub(crate) next_response_output_index: usize, + pub(crate) response_sequence_number: u64, pub(crate) reasoning_block_index: Option, pub(crate) reasoning_block_started: bool, diff --git a/crates/switchyard-translation/tests/request_translation.rs b/crates/switchyard-translation/tests/request_translation.rs index e14d38e9..41d413bb 100644 --- a/crates/switchyard-translation/tests/request_translation.rs +++ b/crates/switchyard-translation/tests/request_translation.rs @@ -728,6 +728,52 @@ fn responses_unknown_input_item_is_preserved_for_openai_chat() -> TestResult { Ok(()) } +// Responses accepts message-shaped input items without an explicit discriminator. +#[test] +fn responses_input_message_without_type_translates_normally() -> TestResult { + let engine = TranslationEngine::default(); + let body = json!({ + "model": "gpt-4", + "input": [{"role": "user", "content": "hello"}] + }); + + let output = engine + .translate_request( + WireFormat::OpenAiResponses, + WireFormat::OpenAiChat, + &body, + &TranslationPolicy::default(), + )? + .body; + + assert_eq!( + output["messages"], + json!([{"role": "user", "content": "hello"}]) + ); + Ok(()) +} + +// A discriminator-less object that is not message-shaped must not silently become prompt text. +#[test] +fn responses_input_without_type_or_message_shape_is_rejected() { + let engine = TranslationEngine::default(); + let body = json!({ + "model": "gpt-4", + "input": [{"payload": "ambiguous"}] + }); + + let error = engine + .translate_request( + WireFormat::OpenAiResponses, + WireFormat::OpenAiChat, + &body, + &TranslationPolicy::default(), + ) + .expect_err("ambiguous input item should be rejected"); + + assert!(error.to_string().contains("$.input[0].type")); +} + // Verifies orphan Responses tool outputs degrade to readable user text. #[test] fn responses_orphan_function_call_output_degrades_to_user_text_for_openai_chat() -> TestResult { diff --git a/crates/switchyard-translation/tests/stream_translation.rs b/crates/switchyard-translation/tests/stream_translation.rs index 3aeed715..70fc4d11 100644 --- a/crates/switchyard-translation/tests/stream_translation.rs +++ b/crates/switchyard-translation/tests/stream_translation.rs @@ -5,7 +5,7 @@ use pretty_assertions::assert_eq; use serde_json::json; -use switchyard_protocol::{ResponseAccumulator, StopReason}; +use switchyard_protocol::{LlmResponseStreamEvent, ResponseAccumulator, StopReason}; use switchyard_translation::{ LlmResponseChunk, StreamTranslationState, TranslationEngine, WireFormat, decode_stream_event, }; @@ -182,6 +182,55 @@ fn replayed_nonterminal_event_advances_encoder_state_before_finish() -> TestResu Ok(()) } +// Exact replay advances sequencing by the one raw event actually emitted, not discarded +// synthetic events produced while advancing encoder state. +#[test] +fn responses_replay_without_sequence_advances_by_one_emitted_event() -> TestResult { + let engine = TranslationEngine::default(); + let format = WireFormat::OpenAiResponses; + let raw = json!({ + "type": "response.output_item.added", + "output_index": 0, + "item": { + "type": "function_call", + "id": "fc_0", + "call_id": "call_0", + "name": "bash", + "arguments": "" + } + }); + let replayed = LlmResponseStreamEvent::preserved( + format, + raw.clone(), + vec![LlmResponseChunk::ToolCallDelta { + index: 0, + id: Some("call_0".to_string()), + name: Some("bash".to_string()), + arguments_delta: None, + }], + ); + let mut state = StreamTranslationState::new(format, format); + + assert_eq!( + engine.encode_stream_event(&mut state, format, replayed)?, + vec![raw] + ); + let generated = engine.encode_stream_event( + &mut state, + format, + LlmResponseStreamEvent::new(vec![LlmResponseChunk::ToolCallDelta { + index: 0, + id: None, + name: None, + arguments_delta: Some("{}".to_string()), + }]), + )?; + + assert_eq!(generated.len(), 1); + assert_eq!(generated[0]["sequence_number"], 1); + Ok(()) +} + #[test] fn replayed_anthropic_terminal_delta_finishes_with_message_stop_only() -> TestResult { let engine = TranslationEngine::default(); @@ -986,6 +1035,76 @@ fn openai_chat_stream_usage_without_breakdowns_still_emits_responses_usage_detai Ok(()) } +// Responses terminal snapshots include every field required by strict generated clients. +#[test] +fn responses_completed_event_is_schema_complete_and_retains_message_id() -> TestResult { + let engine = TranslationEngine::default(); + let mut state = + StreamTranslationState::new(WireFormat::OpenAiChat, WireFormat::OpenAiResponses); + let chunk = json!({ + "id": "chatcmpl-test", + "object": "chat.completion.chunk", + "model": "gpt-4o", + "choices": [{ + "index": 0, + "delta": {"content": "hello"}, + "finish_reason": "stop" + }] + }); + + let mut events = engine.translate_event( + &mut state, + WireFormat::OpenAiChat, + WireFormat::OpenAiResponses, + &chunk, + )?; + events.extend(engine.finish_stream(&mut state, WireFormat::OpenAiResponses)?); + + for (expected, event) in events.iter().enumerate() { + assert_eq!(event["sequence_number"], expected as u64); + } + let completed = events + .iter() + .find(|event| event["type"] == "response.completed") + .ok_or("expected response.completed")?; + let response = completed["response"] + .as_object() + .ok_or("completed response should be an object")?; + for field in [ + "id", + "object", + "created_at", + "completed_at", + "error", + "incomplete_details", + "instructions", + "metadata", + "model", + "output", + "parallel_tool_calls", + "frequency_penalty", + "presence_penalty", + "status", + "temperature", + "tool_choice", + "tools", + "top_p", + "usage", + ] { + assert!( + response.contains_key(field), + "missing response field {field}" + ); + } + assert_eq!(response["output"][0]["id"], "msg_0"); + let done = events + .iter() + .find(|event| event["type"] == "response.output_item.done") + .ok_or("expected response.output_item.done")?; + assert_eq!(done["item"]["id"], "msg_0"); + Ok(()) +} + // Verifies a streamed token-limit stop terminates with response.incomplete. #[test] fn openai_chat_length_finish_translates_to_responses_incomplete_event() -> TestResult {