fix(translation): harden Responses compatibility

Signed-off-by: Alex Steiner <asteiner@nvidia.com>

Author: pst2154

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,

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,63 @@ impl StreamCodec for OpenAiResponsesStreamCodec {
         state: &mut StreamTranslationState,
         event: LlmResponseChunk,
     ) -> Vec<Value> {
-        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<LlmResponseChunk>,
+    ) {
+        let replayed_terminal = normalized
+            .iter()
+            .any(|chunk| matches!(chunk, LlmResponseChunk::MessageStop { .. }));
+        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<Value> {
-        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<u64>,
+    error: Option<Value>,
+    incomplete_details: Option<Value>,
+    instructions: Option<Value>,
+    metadata: Option<Value>,
+    model: String,
+    output: Vec<Value>,
+    parallel_tool_calls: bool,
+    frequency_penalty: Option<f64>,
+    presence_penalty: Option<f64>,
+    status: &'static str,
+    temperature: Option<f64>,
+    tool_choice: &'static str,
+    tools: Vec<Value>,
+    top_p: Option<f64>,
+    usage: Value,
+}
+
 // Decodes one OpenAI Responses event into neutral streaming events.
 fn decode_responses_stream(
     state: &mut StreamTranslationState,
@@ -224,6 +274,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec<Value> {
             "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 +316,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec<Value> {
             output_index,
             json!({
                 "type": "message",
+                "id": format!("msg_{output_index}"),
                 "role": "assistant",
                 "status": status,
                 "content": [{"type": "output_text", "text": state.response_text}],
@@ -306,15 +358,7 @@ fn finish_responses_stream(state: &mut StreamTranslationState) -> Vec<Value> {
 
     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 +437,54 @@ fn ensure_responses_created(state: &mut StreamTranslationState) -> Vec<Value> {
     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<Value>,
+    output: Vec<Value>,
+) -> 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<Value>) -> Vec<Value> {
+    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<Value> {
     let mut out = ensure_responses_created(state);

crates/switchyard-translation/src/codecs/stream.rs

@@ -58,6 +58,7 @@ pub struct StreamTranslationState {
     pub(crate) response_reasoning_output_index: Option<usize>,
     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<usize>,
     pub(crate) reasoning_block_started: bool,

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 {

crates/switchyard-translation/tests/stream_translation.rs

@@ -986,6 +986,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 {