openai-interface 0.14.0

//! Streaming event types for the Responses API.
//!
//! When `stream: true`, the server returns a sequence of semantic SSE
//! events instead of a single response object. Every event carries an
//! incrementing `sequence_number`; the stream ends with
//! `response.completed`, `response.incomplete`, or `response.failed`
//! (there is **no** `data: [DONE]` sentinel).
//!
//! The event set follows the OpenAI Responses streaming events together
//! with the subsets documented by
//! [DeepSeek](https://api-docs.deepseek.com/guides/responses_api#streaming)
//! and
//! [Qwen](https://www.alibabacloud.com/help/zh/model-studio/qwen-api-via-openai-responses).
//! Events not modeled explicitly deserialize to
//! [`ResponseStreamEvent::Other`].

use serde::Deserialize;

use crate::responses::{Response, ResponseOutputItem};

/// One event of a streamed Responses API call.
///
/// Use [`Self::final_response`] to extract the finished [`Response`] from
/// the terminal `response.completed` / `response.incomplete` /
/// `response.failed` events.
#[derive(Debug, Deserialize, Clone)]
#[serde(tag = "type")]
pub enum ResponseStreamEvent {
    /// First event: the response was created, status `in_progress`.
    #[serde(rename = "response.created")]
    ResponseCreated {
        #[serde(default)]
        sequence_number: Option<u64>,
        response: Response,
    },

    /// The response is being generated.
    #[serde(rename = "response.in_progress")]
    ResponseInProgress {
        #[serde(default)]
        sequence_number: Option<u64>,
        response: Response,
    },

    /// The response completed normally; final event with the full
    /// response object including `usage`.
    #[serde(rename = "response.completed")]
    ResponseCompleted {
        #[serde(default)]
        sequence_number: Option<u64>,
        response: Response,
    },

    /// The response was truncated (e.g. `max_output_tokens` reached);
    /// final event with the full response object.
    #[serde(rename = "response.incomplete")]
    ResponseIncomplete {
        #[serde(default)]
        sequence_number: Option<u64>,
        response: Response,
    },

    /// The response failed; final event with the full response object
    /// including the `error` details.
    #[serde(rename = "response.failed")]
    ResponseFailed {
        #[serde(default)]
        sequence_number: Option<u64>,
        response: Response,
    },

    /// An error event emitted mid-stream.
    #[serde(rename = "error")]
    Error {
        #[serde(default)]
        sequence_number: Option<u64>,
        /// A human-readable description of the error.
        #[serde(default)]
        message: Option<String>,
        /// The machine-readable error code.
        #[serde(default)]
        code: Option<String>,
        /// The request parameter that caused the error, if any.
        #[serde(default)]
        param: Option<String>,
    },

    /// An output item (reasoning, message, tool call, ...) started.
    #[serde(rename = "response.output_item.added")]
    OutputItemAdded {
        #[serde(default)]
        sequence_number: Option<u64>,
        /// The index of the item in the `output` array.
        #[serde(default)]
        output_index: Option<u32>,
        /// The item that was added.
        item: ResponseOutputItem,
    },

    /// An output item completed.
    #[serde(rename = "response.output_item.done")]
    OutputItemDone {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        output_index: Option<u32>,
        /// The completed item.
        item: ResponseOutputItem,
    },

    /// A content part started within an output item.
    #[serde(rename = "response.content_part.added")]
    ContentPartAdded {
        #[serde(default)]
        sequence_number: Option<u64>,
        /// The ID of the item the content part belongs to.
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        /// The index of the content part within the item.
        #[serde(default)]
        content_index: Option<u32>,
        /// The content part that was added.
        part: EventContentPart,
    },

    /// A content part completed.
    #[serde(rename = "response.content_part.done")]
    ContentPartDone {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        #[serde(default)]
        content_index: Option<u32>,
        /// The completed content part.
        part: EventContentPart,
    },

    /// An incremental text delta of an output message.
    #[serde(rename = "response.output_text.delta")]
    OutputTextDelta {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        #[serde(default)]
        content_index: Option<u32>,
        /// The incremental text.
        delta: String,
    },

    /// The full output text of a message content part.
    #[serde(rename = "response.output_text.done")]
    OutputTextDone {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        #[serde(default)]
        content_index: Option<u32>,
        /// The complete text.
        text: String,
    },

    /// An incremental reasoning (chain-of-thought) text delta.
    #[serde(rename = "response.reasoning_text.delta")]
    ReasoningTextDelta {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        #[serde(default)]
        content_index: Option<u32>,
        /// The incremental reasoning text.
        delta: String,
    },

    /// The full reasoning (chain-of-thought) text of a reasoning item.
    #[serde(rename = "response.reasoning_text.done")]
    ReasoningTextDone {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        #[serde(default)]
        content_index: Option<u32>,
        /// The complete reasoning text.
        text: String,
    },

    /// An incremental delta of function call arguments.
    #[serde(rename = "response.function_call_arguments.delta")]
    FunctionCallArgumentsDelta {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        /// The incremental arguments JSON fragment.
        delta: String,
    },

    /// The full arguments of a function call.
    #[serde(rename = "response.function_call_arguments.done")]
    FunctionCallArgumentsDone {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        /// The complete arguments JSON string.
        arguments: String,
    },

    /// An incremental delta of custom tool call input (e.g. the
    /// `apply_patch` tool).
    #[serde(rename = "response.custom_tool_call_input.delta")]
    CustomToolCallInputDelta {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        /// The incremental input fragment.
        delta: String,
    },

    /// The full input of a custom tool call.
    #[serde(rename = "response.custom_tool_call_input.done")]
    CustomToolCallInputDone {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        /// The complete input.
        input: String,
    },

    /// A server-side web search started.
    #[serde(rename = "response.web_search_call.in_progress")]
    WebSearchCallInProgress {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
    },

    /// A server-side web search is searching.
    #[serde(rename = "response.web_search_call.searching")]
    WebSearchCallSearching {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
    },

    /// A server-side web search completed.
    #[serde(rename = "response.web_search_call.completed")]
    WebSearchCallCompleted {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
    },

    /// Qwen: an incremental delta of MCP tool call arguments.
    #[cfg(feature = "qwen")]
    #[serde(rename = "response.mcp_call_arguments.delta")]
    McpCallArgumentsDelta {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        /// The incremental arguments JSON fragment.
        delta: String,
    },

    /// Qwen: the full arguments of an MCP tool call.
    #[cfg(feature = "qwen")]
    #[serde(rename = "response.mcp_call_arguments.done")]
    McpCallArgumentsDone {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
        /// The complete arguments JSON string.
        arguments: String,
    },

    /// Qwen: an MCP tool call completed.
    #[cfg(feature = "qwen")]
    #[serde(rename = "response.mcp_call.completed")]
    McpCallCompleted {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
    },

    /// Qwen: a knowledge-base search started.
    #[cfg(feature = "qwen")]
    #[serde(rename = "response.file_search_call.in_progress")]
    FileSearchCallInProgress {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
    },

    /// Qwen: a knowledge-base search is searching.
    #[cfg(feature = "qwen")]
    #[serde(rename = "response.file_search_call.searching")]
    FileSearchCallSearching {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
    },

    /// Qwen: a knowledge-base search completed.
    #[cfg(feature = "qwen")]
    #[serde(rename = "response.file_search_call.completed")]
    FileSearchCallCompleted {
        #[serde(default)]
        sequence_number: Option<u64>,
        #[serde(default)]
        item_id: Option<String>,
        #[serde(default)]
        output_index: Option<u32>,
    },

    /// Any event type not covered by the variants above.
    #[serde(other)]
    Other,
}

impl ResponseStreamEvent {
    /// Returns the final response object if this is one of the terminal
    /// events (`response.completed`, `response.incomplete`,
    /// `response.failed`).
    pub fn final_response(&self) -> Option<&Response> {
        match self {
            Self::ResponseCompleted { response, .. }
            | Self::ResponseIncomplete { response, .. }
            | Self::ResponseFailed { response, .. } => Some(response),
            _ => None,
        }
    }
}

/// A content part carried by `response.content_part.added` /
/// `response.content_part.done` events.
#[derive(Debug, Deserialize, Clone)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum EventContentPart {
    /// A text output part.
    OutputText {
        /// The text content.
        #[serde(default)]
        text: Option<String>,
    },
    /// A reasoning text part.
    ReasoningText {
        /// The reasoning text.
        #[serde(default)]
        text: Option<String>,
    },
    /// A refusal part.
    Refusal {
        /// The refusal message.
        #[serde(default)]
        refusal: Option<String>,
    },
    /// Any content part type not covered above.
    #[serde(other)]
    Other,
}

crate::impl_from_str!(ResponseStreamEvent);

#[cfg(test)]
mod tests {
    //! Offline deserialization tests using event payloads copied from the
    //! DeepSeek streaming documentation example.

    use std::str::FromStr;

    use super::*;
    use crate::responses::{ResponseObject, ResponseOutputItem};

    /// From the streaming example at
    /// <https://api-docs.deepseek.com/api/create-response>.
    #[test]
    fn parses_documented_stream_events() {
        let created = ResponseStreamEvent::from_str(
            r#"{"type": "response.created", "sequence_number": 0, "response": {"id": "resp_1", "object": "response", "created_at": 0, "status": "in_progress", "model": "deepseek-v4-flash", "output": []}}"#,
        ).unwrap();
        assert!(matches!(
            created,
            ResponseStreamEvent::ResponseCreated { .. }
        ));

        let delta = ResponseStreamEvent::from_str(
            r#"{"type": "response.reasoning_text.delta", "sequence_number": 4, "item_id": "rs_1", "output_index": 0, "content_index": 0, "delta": "The user"}"#,
        ).unwrap();
        match delta {
            ResponseStreamEvent::ReasoningTextDelta { delta, item_id, .. } => {
                assert_eq!(delta, "The user");
                assert_eq!(item_id.as_deref(), Some("rs_1"));
            }
            other => panic!("unexpected event: {other:?}"),
        }

        let text_delta = ResponseStreamEvent::from_str(
            r#"{"type": "response.output_text.delta", "sequence_number": 11, "item_id": "msg_1", "output_index": 1, "content_index": 0, "delta": "Hello"}"#,
        ).unwrap();
        assert!(matches!(
            text_delta,
            ResponseStreamEvent::OutputTextDelta { .. }
        ));
    }

    #[test]
    fn parses_completed_event_and_extracts_response() {
        let event = ResponseStreamEvent::from_str(
            r#"{"type": "response.completed", "sequence_number": 20, "response": {"id": "resp_1", "object": "response", "created_at": 0, "status": "completed", "model": "deepseek-v4-flash", "output": [{"type": "message", "id": "msg_1", "status": "completed", "role": "assistant", "content": [{"type": "output_text", "text": "Hi!", "annotations": []}]}], "usage": {"input_tokens": 1, "output_tokens": 2, "total_tokens": 3}}}"#,
        ).unwrap();

        let response = event.final_response().expect("terminal event");
        assert_eq!(response.object, ResponseObject::Response);
        assert!(response.is_completed());
        assert_eq!(response.output_text(), "Hi!");
        assert!(matches!(
            response.output.first(),
            Some(ResponseOutputItem::Message(_))
        ));
    }

    #[test]
    fn unknown_event_type_is_preserved() {
        let event = ResponseStreamEvent::from_str(
            r#"{"type": "response.some_future_event", "sequence_number": 3}"#,
        )
        .unwrap();
        assert!(matches!(event, ResponseStreamEvent::Other));
    }
}