use bytes::Bytes;
use futures::stream::{Stream, StreamExt};
use serde::{Deserialize, Serialize};
use serde_json::json;
use std::collections::{HashMap, HashSet};

use crate::proxy::response::StreamCompletion;

#[derive(Debug, Deserialize)]
struct OpenAIStreamChunk {
    id: String,
    model: String,
    choices: Vec<StreamChoice>,
    #[serde(default)]
    usage: Option<Usage>,
}

#[derive(Debug, Deserialize)]
struct StreamChoice {
    delta: Delta,
    #[serde(default)]
    finish_reason: Option<String>,
}

#[derive(Debug, Deserialize)]
struct Delta {
    #[serde(default)]
    content: Option<String>,
    #[serde(default, alias = "reasoning_content")]
    reasoning: Option<String>,
    #[serde(default)]
    tool_calls: Option<Vec<DeltaToolCall>>,
    #[serde(default)]
    function_call: Option<DeltaFunction>,
}

#[derive(Debug, Deserialize, Serialize)]
struct DeltaToolCall {
    index: usize,
    #[serde(default)]
    id: Option<String>,
    #[serde(rename = "type", default)]
    call_type: Option<String>,
    #[serde(default)]
    function: Option<DeltaFunction>,
}

#[derive(Debug, Deserialize, Serialize)]
struct DeltaFunction {
    #[serde(default)]
    name: Option<String>,
    #[serde(default)]
    arguments: Option<String>,
}

#[derive(Debug, Deserialize)]
struct Usage {
    #[serde(default)]
    prompt_tokens: u32,
    #[serde(default)]
    completion_tokens: u32,
    #[serde(default)]
    prompt_tokens_details: Option<PromptTokensDetails>,
    #[serde(default)]
    cache_read_input_tokens: Option<u32>,
    #[serde(default)]
    cache_creation_input_tokens: Option<u32>,
}

#[derive(Debug, Deserialize)]
struct PromptTokensDetails {
    #[serde(default)]
    cached_tokens: u32,
}

#[derive(Debug, Clone)]
struct ToolBlockState {
    anthropic_index: u32,
    id: String,
    name: String,
    started: bool,
    pending_args: String,
}

pub fn create_anthropic_sse_stream(
    stream: impl Stream<Item = Result<Bytes, std::io::Error>> + Send + 'static,
    stream_completion: StreamCompletion,
) -> impl Stream<Item = Result<Bytes, std::io::Error>> + Send {
    async_stream::stream! {
        let mut buffer = String::new();
        let mut message_id = None;
        let mut current_model = None;
        let mut next_content_index: u32 = 0;
        let mut has_sent_message_start = false;
        let mut current_non_tool_block_type: Option<&'static str> = None;
        let mut current_non_tool_block_index: Option<u32> = None;
        let mut tool_blocks_by_index: HashMap<usize, ToolBlockState> = HashMap::new();
        let mut open_tool_block_indices: HashSet<u32> = HashSet::new();
        let mut legacy_function_name: Option<String> = None;
        let mut legacy_function_block_index: Option<u32> = None;

        tokio::pin!(stream);

        while let Some(chunk) = stream.next().await {
            match chunk {
                Ok(bytes) => {
                    buffer.push_str(&String::from_utf8_lossy(&bytes));

                    while let Some(pos) = buffer.find("\n\n") {
                        let line = buffer[..pos].to_string();
                        buffer = buffer[pos + 2..].to_string();

                        if line.trim().is_empty() {
                            continue;
                        }

                        for raw_line in line.lines() {
                            let Some(data) = raw_line.strip_prefix("data:").map(str::trim_start) else {
                                continue;
                            };

                            if data.trim() == "[DONE]" {
                                let event = json!({"type": "message_stop"});
                                let sse_data = format!(
                                    "event: message_stop\ndata: {}\n\n",
                                    serde_json::to_string(&event).unwrap_or_default()
                                );
                                stream_completion.record_success();
                                yield Ok(Bytes::from(sse_data));
                                return;
                            }

                            let Ok(chunk) = serde_json::from_str::<OpenAIStreamChunk>(data) else {
                                continue;
                            };

                            if message_id.is_none() {
                                message_id = Some(chunk.id.clone());
                            }
                            if current_model.is_none() {
                                current_model = Some(chunk.model.clone());
                            }

                            let Some(choice) = chunk.choices.first() else {
                                continue;
                            };

                            if !has_sent_message_start {
                                let mut start_usage = json!({
                                    "input_tokens": 0,
                                    "output_tokens": 0
                                });
                                if let Some(usage) = &chunk.usage {
                                    start_usage["input_tokens"] = json!(usage.prompt_tokens);
                                    if let Some(cached) = extract_cache_read_tokens(usage) {
                                        start_usage["cache_read_input_tokens"] = json!(cached);
                                    }
                                    if let Some(created) = usage.cache_creation_input_tokens {
                                        start_usage["cache_creation_input_tokens"] = json!(created);
                                    }
                                }

                                let event = json!({
                                    "type": "message_start",
                                    "message": {
                                        "id": message_id.clone().unwrap_or_default(),
                                        "type": "message",
                                        "role": "assistant",
                                        "model": current_model.clone().unwrap_or_default(),
                                        "usage": start_usage
                                    }
                                });
                                let sse_data = format!(
                                    "event: message_start\ndata: {}\n\n",
                                    serde_json::to_string(&event).unwrap_or_default()
                                );
                                yield Ok(Bytes::from(sse_data));
                                has_sent_message_start = true;
                            }

                            if let Some(reasoning) = &choice.delta.reasoning {
                                if current_non_tool_block_type != Some("thinking") {
                                    if let Some(index) = current_non_tool_block_index.take() {
                                        let event = json!({
                                            "type": "content_block_stop",
                                            "index": index
                                        });
                                        let sse_data = format!(
                                            "event: content_block_stop\ndata: {}\n\n",
                                            serde_json::to_string(&event).unwrap_or_default()
                                        );
                                        yield Ok(Bytes::from(sse_data));
                                    }
                                    let index = next_content_index;
                                    next_content_index += 1;
                                    let event = json!({
                                        "type": "content_block_start",
                                        "index": index,
                                        "content_block": {
                                            "type": "thinking",
                                            "thinking": ""
                                        }
                                    });
                                    let sse_data = format!(
                                        "event: content_block_start\ndata: {}\n\n",
                                        serde_json::to_string(&event).unwrap_or_default()
                                    );
                                    yield Ok(Bytes::from(sse_data));
                                    current_non_tool_block_type = Some("thinking");
                                    current_non_tool_block_index = Some(index);
                                }

                                if let Some(index) = current_non_tool_block_index {
                                    let event = json!({
                                        "type": "content_block_delta",
                                        "index": index,
                                        "delta": {
                                            "type": "thinking_delta",
                                            "thinking": reasoning
                                        }
                                    });
                                    let sse_data = format!(
                                        "event: content_block_delta\ndata: {}\n\n",
                                        serde_json::to_string(&event).unwrap_or_default()
                                    );
                                    yield Ok(Bytes::from(sse_data));
                                }
                            }

                            if let Some(content) = &choice.delta.content {
                                if !content.is_empty() {
                                    if current_non_tool_block_type != Some("text") {
                                        if let Some(index) = current_non_tool_block_index.take() {
                                            let event = json!({
                                                "type": "content_block_stop",
                                                "index": index
                                            });
                                            let sse_data = format!(
                                                "event: content_block_stop\ndata: {}\n\n",
                                                serde_json::to_string(&event).unwrap_or_default()
                                            );
                                            yield Ok(Bytes::from(sse_data));
                                        }

                                        let index = next_content_index;
                                        next_content_index += 1;
                                        let event = json!({
                                            "type": "content_block_start",
                                            "index": index,
                                            "content_block": {
                                                "type": "text",
                                                "text": ""
                                            }
                                        });
                                        let sse_data = format!(
                                            "event: content_block_start\ndata: {}\n\n",
                                            serde_json::to_string(&event).unwrap_or_default()
                                        );
                                        yield Ok(Bytes::from(sse_data));
                                        current_non_tool_block_type = Some("text");
                                        current_non_tool_block_index = Some(index);
                                    }

                                    if let Some(index) = current_non_tool_block_index {
                                        let event = json!({
                                            "type": "content_block_delta",
                                            "index": index,
                                            "delta": {
                                                "type": "text_delta",
                                                "text": content
                                            }
                                        });
                                        let sse_data = format!(
                                            "event: content_block_delta\ndata: {}\n\n",
                                            serde_json::to_string(&event).unwrap_or_default()
                                        );
                                        yield Ok(Bytes::from(sse_data));
                                    }
                                }
                            }

                            if let Some(tool_calls) = &choice.delta.tool_calls {
                                if let Some(index) = current_non_tool_block_index.take() {
                                    let event = json!({
                                        "type": "content_block_stop",
                                        "index": index
                                    });
                                    let sse_data = format!(
                                        "event: content_block_stop\ndata: {}\n\n",
                                        serde_json::to_string(&event).unwrap_or_default()
                                    );
                                    yield Ok(Bytes::from(sse_data));
                                }
                                current_non_tool_block_type = None;

                                for tool_call in tool_calls {
                                    let (
                                        anthropic_index,
                                        id,
                                        name,
                                        should_start,
                                        pending_after_start,
                                        immediate_delta,
                                    ) = {
                                        let state = tool_blocks_by_index
                                            .entry(tool_call.index)
                                            .or_insert_with(|| {
                                                let index = next_content_index;
                                                next_content_index += 1;
                                                ToolBlockState {
                                                    anthropic_index: index,
                                                    id: String::new(),
                                                    name: String::new(),
                                                    started: false,
                                                    pending_args: String::new(),
                                                }
                                            });

                                        if let Some(id) = &tool_call.id {
                                            state.id = id.clone();
                                        }
                                        if let Some(function) = &tool_call.function {
                                            if let Some(name) = &function.name {
                                                state.name = name.clone();
                                            }
                                        }

                                        let should_start = !state.started
                                            && !state.id.is_empty()
                                            && !state.name.is_empty();
                                        if should_start {
                                            state.started = true;
                                        }
                                        let pending_after_start = if should_start
                                            && !state.pending_args.is_empty()
                                        {
                                            Some(std::mem::take(&mut state.pending_args))
                                        } else {
                                            None
                                        };
                                        let args_delta = tool_call
                                            .function
                                            .as_ref()
                                            .and_then(|f| f.arguments.clone());
                                        let immediate_delta = if let Some(args) = args_delta {
                                            if state.started {
                                                Some(args)
                                            } else {
                                                state.pending_args.push_str(&args);
                                                None
                                            }
                                        } else {
                                            None
                                        };

                                        (
                                            state.anthropic_index,
                                            state.id.clone(),
                                            state.name.clone(),
                                            should_start,
                                            pending_after_start,
                                            immediate_delta,
                                        )
                                    };

                                    if should_start {
                                        let event = json!({
                                            "type": "content_block_start",
                                            "index": anthropic_index,
                                            "content_block": {
                                                "type": "tool_use",
                                                "id": id,
                                                "name": name,
                                                "input": {}
                                            }
                                        });
                                        let sse_data = format!(
                                            "event: content_block_start\ndata: {}\n\n",
                                            serde_json::to_string(&event).unwrap_or_default()
                                        );
                                        yield Ok(Bytes::from(sse_data));
                                        open_tool_block_indices.insert(anthropic_index);
                                    }

                                    if let Some(args) = pending_after_start {
                                        let event = json!({
                                            "type": "content_block_delta",
                                            "index": anthropic_index,
                                            "delta": {
                                                "type": "input_json_delta",
                                                "partial_json": args
                                            }
                                        });
                                        let sse_data = format!(
                                            "event: content_block_delta\ndata: {}\n\n",
                                            serde_json::to_string(&event).unwrap_or_default()
                                        );
                                        yield Ok(Bytes::from(sse_data));
                                    }

                                    if let Some(args) = immediate_delta {
                                        let event = json!({
                                            "type": "content_block_delta",
                                            "index": anthropic_index,
                                            "delta": {
                                                "type": "input_json_delta",
                                                "partial_json": args
                                            }
                                        });
                                        let sse_data = format!(
                                            "event: content_block_delta\ndata: {}\n\n",
                                            serde_json::to_string(&event).unwrap_or_default()
                                        );
                                        yield Ok(Bytes::from(sse_data));
                                    }
                                }
                            }

                            if let Some(function_call) = &choice.delta.function_call {
                                if let Some(name) = &function_call.name {
                                    legacy_function_name = Some(name.clone());
                                }

                                if function_call.name.is_some() || function_call.arguments.is_some() {
                                    if let Some(index) = current_non_tool_block_index.take() {
                                        let event = json!({
                                            "type": "content_block_stop",
                                            "index": index
                                        });
                                        let sse_data = format!(
                                            "event: content_block_stop\ndata: {}\n\n",
                                            serde_json::to_string(&event).unwrap_or_default()
                                        );
                                        yield Ok(Bytes::from(sse_data));
                                    }
                                    current_non_tool_block_type = None;

                                    if legacy_function_block_index.is_none() {
                                        let index = next_content_index;
                                        next_content_index += 1;
                                        legacy_function_block_index = Some(index);
                                        let event = json!({
                                            "type": "content_block_start",
                                            "index": index,
                                            "content_block": {
                                                "type": "tool_use",
                                                "id": "",
                                                "name": legacy_function_name.clone().unwrap_or_default(),
                                                "input": {}
                                            }
                                        });
                                        let sse_data = format!(
                                            "event: content_block_start\ndata: {}\n\n",
                                            serde_json::to_string(&event).unwrap_or_default()
                                        );
                                        yield Ok(Bytes::from(sse_data));
                                    }

                                    if let Some(arguments) = &function_call.arguments {
                                        if let Some(index) = legacy_function_block_index {
                                            let event = json!({
                                                "type": "content_block_delta",
                                                "index": index,
                                                "delta": {
                                                    "type": "input_json_delta",
                                                    "partial_json": arguments
                                                }
                                            });
                                            let sse_data = format!(
                                                "event: content_block_delta\ndata: {}\n\n",
                                                serde_json::to_string(&event).unwrap_or_default()
                                            );
                                            yield Ok(Bytes::from(sse_data));
                                        }
                                    }
                                }
                            }

                            if let Some(finish_reason) = &choice.finish_reason {
                                if let Some(index) = current_non_tool_block_index.take() {
                                    let event = json!({
                                        "type": "content_block_stop",
                                        "index": index
                                    });
                                    let sse_data = format!(
                                        "event: content_block_stop\ndata: {}\n\n",
                                        serde_json::to_string(&event).unwrap_or_default()
                                    );
                                    yield Ok(Bytes::from(sse_data));
                                }
                                current_non_tool_block_type = None;

                                let mut late_tool_starts: Vec<(u32, String, String, String)> =
                                    Vec::new();
                                for (tool_idx, state) in tool_blocks_by_index.iter_mut() {
                                    if state.started {
                                        continue;
                                    }
                                    let has_payload = !state.pending_args.is_empty()
                                        || !state.id.is_empty()
                                        || !state.name.is_empty();
                                    if !has_payload {
                                        continue;
                                    }
                                    let fallback_id = if state.id.is_empty() {
                                        format!("tool_call_{tool_idx}")
                                    } else {
                                        state.id.clone()
                                    };
                                    let fallback_name = if state.name.is_empty() {
                                        "unknown_tool".to_string()
                                    } else {
                                        state.name.clone()
                                    };
                                    state.started = true;
                                    let pending = std::mem::take(&mut state.pending_args);
                                    late_tool_starts.push((
                                        state.anthropic_index,
                                        fallback_id,
                                        fallback_name,
                                        pending,
                                    ));
                                }
                                late_tool_starts.sort_unstable_by_key(|(index, _, _, _)| *index);
                                for (index, id, name, pending) in late_tool_starts {
                                    let event = json!({
                                        "type": "content_block_start",
                                        "index": index,
                                        "content_block": {
                                            "type": "tool_use",
                                            "id": id,
                                            "name": name,
                                            "input": {}
                                        }
                                    });
                                    let sse_data = format!(
                                        "event: content_block_start\ndata: {}\n\n",
                                        serde_json::to_string(&event).unwrap_or_default()
                                    );
                                    yield Ok(Bytes::from(sse_data));
                                    open_tool_block_indices.insert(index);
                                    if !pending.is_empty() {
                                        let delta_event = json!({
                                            "type": "content_block_delta",
                                            "index": index,
                                            "delta": {
                                                "type": "input_json_delta",
                                                "partial_json": pending
                                            }
                                        });
                                        let delta_sse = format!(
                                            "event: content_block_delta\ndata: {}\n\n",
                                            serde_json::to_string(&delta_event).unwrap_or_default()
                                        );
                                        yield Ok(Bytes::from(delta_sse));
                                    }
                                }

                                if let Some(index) = legacy_function_block_index.take() {
                                    let event = json!({
                                        "type": "content_block_stop",
                                        "index": index
                                    });
                                    let sse_data = format!(
                                        "event: content_block_stop\ndata: {}\n\n",
                                        serde_json::to_string(&event).unwrap_or_default()
                                    );
                                    yield Ok(Bytes::from(sse_data));
                                }

                                if !open_tool_block_indices.is_empty() {
                                    let mut tool_indices: Vec<u32> =
                                        open_tool_block_indices.iter().copied().collect();
                                    tool_indices.sort_unstable();
                                    for index in tool_indices {
                                        let event = json!({
                                            "type": "content_block_stop",
                                            "index": index
                                        });
                                        let sse_data = format!(
                                            "event: content_block_stop\ndata: {}\n\n",
                                            serde_json::to_string(&event).unwrap_or_default()
                                        );
                                        yield Ok(Bytes::from(sse_data));
                                    }
                                    open_tool_block_indices.clear();
                                }

                                let usage_json = if let Some(usage) = chunk.usage.as_ref() {
                                    let mut usage_json = json!({
                                        "input_tokens": usage.prompt_tokens,
                                        "output_tokens": usage.completion_tokens
                                    });
                                    if let Some(cached) = extract_cache_read_tokens(usage) {
                                        usage_json["cache_read_input_tokens"] = json!(cached);
                                    }
                                    if let Some(created) = usage.cache_creation_input_tokens {
                                        usage_json["cache_creation_input_tokens"] = json!(created);
                                    }
                                    usage_json
                                } else {
                                    json!({
                                        "output_tokens": 0
                                    })
                                };
                                let event = json!({
                                    "type": "message_delta",
                                    "delta": {
                                        "stop_reason": map_stop_reason(Some(finish_reason)),
                                        "stop_sequence": null
                                    },
                                    "usage": usage_json
                                });
                                let sse_data = format!(
                                    "event: message_delta\ndata: {}\n\n",
                                    serde_json::to_string(&event).unwrap_or_default()
                                );
                                yield Ok(Bytes::from(sse_data));
                            }
                        }
                    }
                }
                Err(error) => {
                    stream_completion.record_error(error.to_string());
                    let error_event = json!({
                        "type": "error",
                        "error": {
                            "type": "stream_error",
                            "message": format!("Stream error: {error}")
                        }
                    });
                    let sse_data = format!(
                        "event: error\ndata: {}\n\n",
                        serde_json::to_string(&error_event).unwrap_or_default()
                    );
                    yield Ok(Bytes::from(sse_data));
                    return;
                }
            }
        }

        stream_completion.record_success();
    }
}

fn map_stop_reason(finish_reason: Option<&str>) -> Option<String> {
    finish_reason.map(|reason| {
        match reason {
            "tool_calls" | "function_call" => "tool_use",
            "stop" => "end_turn",
            "length" => "max_tokens",
            "content_filter" => "end_turn",
            other => {
                log::warn!("[Claude/OpenAI] Unknown finish_reason in streaming: {other}");
                "end_turn"
            }
        }
        .to_string()
    })
}

fn extract_cache_read_tokens(usage: &Usage) -> Option<u32> {
    if let Some(value) = usage.cache_read_input_tokens {
        return Some(value);
    }

    usage
        .prompt_tokens_details
        .as_ref()
        .map(|details| details.cached_tokens)
}

#[cfg(test)]
mod tests {
    use super::*;
    use futures::{stream, StreamExt};
    use serde_json::Value;
    use std::collections::HashMap;

    async fn collect_events(input: &str) -> Vec<Value> {
        let upstream = stream::iter(vec![Ok(Bytes::from(input.as_bytes().to_vec()))]);
        let converted = create_anthropic_sse_stream(upstream, StreamCompletion::default());
        let chunks: Vec<_> = converted.collect().await;

        chunks
            .into_iter()
            .map(|chunk| String::from_utf8_lossy(chunk.unwrap().as_ref()).to_string())
            .collect::<String>()
            .split("\n\n")
            .filter_map(|block| {
                let data = block.lines().find_map(|line| line.strip_prefix("data: "))?;
                serde_json::from_str::<Value>(data).ok()
            })
            .collect()
    }

    #[tokio::test]
    async fn done_marker_ends_stream_without_waiting_for_upstream_eof() {
        let input = concat!(
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":1}}\n\n",
            "data: [DONE]\n\n"
        );
        let upstream = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from(input))])
            .chain(stream::pending::<Result<Bytes, std::io::Error>>());
        let completion = StreamCompletion::default();
        let converted = create_anthropic_sse_stream(upstream, completion.clone());

        let chunks: Vec<_> =
            tokio::time::timeout(std::time::Duration::from_millis(100), converted.collect())
                .await
                .expect("stream should end at [DONE]");
        let merged = chunks
            .into_iter()
            .map(|chunk| String::from_utf8_lossy(chunk.unwrap().as_ref()).to_string())
            .collect::<String>();

        assert!(merged.contains("event: message_stop"));
        assert_eq!(completion.outcome(), Some(Ok(())));
    }

    #[test]
    fn map_stop_reason_covers_existing_and_parity_values() {
        assert_eq!(
            map_stop_reason(Some("tool_calls")),
            Some("tool_use".to_string())
        );
        assert_eq!(
            map_stop_reason(Some("function_call")),
            Some("tool_use".to_string())
        );
        assert_eq!(map_stop_reason(Some("stop")), Some("end_turn".to_string()));
        assert_eq!(
            map_stop_reason(Some("length")),
            Some("max_tokens".to_string())
        );
        assert_eq!(
            map_stop_reason(Some("content_filter")),
            Some("end_turn".to_string())
        );
    }

    #[tokio::test]
    async fn streaming_message_usage_includes_cache_token_fields() {
        let input = concat!(
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{}}],\"usage\":{\"prompt_tokens\":12,\"completion_tokens\":0,\"prompt_tokens_details\":{\"cached_tokens\":5},\"cache_creation_input_tokens\":2}}\n\n",
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":12,\"completion_tokens\":7,\"cache_read_input_tokens\":6,\"cache_creation_input_tokens\":3}}\n\n",
            "data: [DONE]\n\n"
        );

        let events = collect_events(input).await;
        let message_start = events
            .iter()
            .find(|event| event["type"] == "message_start")
            .expect("message_start event");
        let message_delta = events
            .iter()
            .find(|event| event["type"] == "message_delta")
            .expect("message_delta event");

        assert_eq!(message_start["message"]["usage"]["input_tokens"], 12);
        assert_eq!(message_start["message"]["usage"]["output_tokens"], 0);
        assert_eq!(
            message_start["message"]["usage"]["cache_read_input_tokens"],
            5
        );
        assert_eq!(
            message_start["message"]["usage"]["cache_creation_input_tokens"],
            2
        );

        assert_eq!(message_delta["usage"]["input_tokens"], 12);
        assert_eq!(message_delta["usage"]["output_tokens"], 7);
        assert_eq!(message_delta["usage"]["cache_read_input_tokens"], 6);
        assert_eq!(message_delta["usage"]["cache_creation_input_tokens"], 3);
    }

    #[tokio::test]
    async fn streaming_usage_preserves_zero_cached_tokens() {
        let input = concat!(
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{}}],\"usage\":{\"prompt_tokens\":12,\"completion_tokens\":0,\"prompt_tokens_details\":{\"cached_tokens\":0}}}\n\n",
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":12,\"completion_tokens\":7}}\n\n",
            "data: [DONE]\n\n"
        );

        let events = collect_events(input).await;
        let message_start = events
            .iter()
            .find(|event| event["type"] == "message_start")
            .expect("message_start event");

        assert_eq!(
            message_start["message"]["usage"]["cache_read_input_tokens"],
            0
        );
    }

    #[tokio::test]
    async fn streaming_accepts_deepseek_reasoning_content_alias() {
        let input = concat!(
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"deepseek-v4-pro\",\"choices\":[{\"delta\":{\"reasoning_content\":\"think\"}}]}\n\n",
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"deepseek-v4-pro\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":1}}\n\n",
            "data: [DONE]\n\n"
        );

        let events = collect_events(input).await;
        let thinking_delta = events
            .iter()
            .find(|event| event["type"] == "content_block_delta")
            .expect("thinking delta should be emitted");

        assert_eq!(thinking_delta["delta"]["type"], "thinking_delta");
        assert_eq!(thinking_delta["delta"]["thinking"], "think");
    }

    #[tokio::test]
    async fn legacy_function_call_stream_emits_tool_use_block_and_argument_delta() {
        let input = concat!(
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"function_call\":{\"name\":\"get_weather\"}}}]}\n\n",
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"function_call\":{\"arguments\":\"{\\\"location\\\":\\\"Tokyo\\\"}\"}}}]}\n\n",
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"function_call\"}],\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":5}}\n\n",
            "data: [DONE]\n\n"
        );

        let events = collect_events(input).await;
        let tool_start = events
            .iter()
            .find(|event| {
                event["type"] == "content_block_start"
                    && event["content_block"]["type"] == "tool_use"
            })
            .expect("tool_use block start");
        let tool_delta = events
            .iter()
            .find(|event| {
                event["type"] == "content_block_delta"
                    && event["delta"]["type"] == "input_json_delta"
            })
            .expect("tool_use argument delta");
        let message_delta = events
            .iter()
            .find(|event| event["type"] == "message_delta")
            .expect("message_delta event");

        assert_eq!(tool_start["content_block"]["name"], "get_weather");
        assert_eq!(tool_start["content_block"]["input"], json!({}));
        assert_eq!(
            tool_delta["delta"]["partial_json"],
            "{\"location\":\"Tokyo\"}"
        );
        assert_eq!(message_delta["delta"]["stop_reason"], "tool_use");
    }

    #[tokio::test]
    async fn streaming_tool_calls_route_arguments_by_index() {
        let input = concat!(
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_0\",\"type\":\"function\",\"function\":{\"name\":\"first_tool\"}}]}}]}\n\n",
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":1,\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"second_tool\"}}]}}]}\n\n",
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":1,\"function\":{\"arguments\":\"{\\\"b\\\":2}\"}}]}}]}\n\n",
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"a\\\":1}\"}}]}}]}\n\n",
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":8,\"completion_tokens\":4}}\n\n",
            "data: [DONE]\n\n"
        );

        let events = collect_events(input).await;
        let mut tool_index_by_call: HashMap<String, u64> = HashMap::new();
        for event in &events {
            if event["type"] == "content_block_start"
                && event["content_block"]["type"] == "tool_use"
            {
                if let (Some(call_id), Some(index)) = (
                    event.pointer("/content_block/id").and_then(|v| v.as_str()),
                    event.get("index").and_then(|v| v.as_u64()),
                ) {
                    tool_index_by_call.insert(call_id.to_string(), index);
                }
            }
        }

        assert_eq!(tool_index_by_call.len(), 2);
        assert_ne!(
            tool_index_by_call.get("call_0"),
            tool_index_by_call.get("call_1")
        );

        let deltas: Vec<(u64, String)> = events
            .iter()
            .filter(|event| {
                event["type"] == "content_block_delta"
                    && event["delta"]["type"] == "input_json_delta"
            })
            .filter_map(|event| {
                let index = event.get("index").and_then(|v| v.as_u64())?;
                let partial_json = event
                    .pointer("/delta/partial_json")
                    .and_then(|v| v.as_str())?
                    .to_string();
                Some((index, partial_json))
            })
            .collect();

        let second_idx = deltas
            .iter()
            .find_map(|(index, payload)| (payload == "{\"b\":2}").then_some(*index))
            .expect("second tool delta index");
        let first_idx = deltas
            .iter()
            .find_map(|(index, payload)| (payload == "{\"a\":1}").then_some(*index))
            .expect("first tool delta index");

        assert_eq!(second_idx, *tool_index_by_call.get("call_1").unwrap());
        assert_eq!(first_idx, *tool_index_by_call.get("call_0").unwrap());
    }

    #[tokio::test]
    async fn streaming_tool_call_schema_is_valid_when_finish_chunk_omits_usage() {
        let input = concat!(
            "data: {\"id\":\"chatcmpl_tool\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_0\",\"type\":\"function\",\"function\":{\"name\":\"get_weather\"}}]}}]}\n\n",
            "data: {\"id\":\"chatcmpl_tool\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"location\\\":\\\"Tokyo\\\"}\"}}]}}]}\n\n",
            "data: {\"id\":\"chatcmpl_tool\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}]}\n\n",
            "data: [DONE]\n\n"
        );

        let events = collect_events(input).await;
        let tool_start = events
            .iter()
            .find(|event| {
                event["type"] == "content_block_start"
                    && event["content_block"]["type"] == "tool_use"
            })
            .expect("tool_use block start");
        let message_delta = events
            .iter()
            .find(|event| event["type"] == "message_delta")
            .expect("message_delta event");

        assert_eq!(tool_start["content_block"]["id"], "call_0");
        assert_eq!(tool_start["content_block"]["name"], "get_weather");
        assert_eq!(tool_start["content_block"]["input"], json!({}));
        assert_eq!(message_delta["delta"]["stop_reason"], "tool_use");
        assert!(message_delta["usage"].is_object());
        assert_eq!(message_delta["usage"]["output_tokens"], 0);
        assert!(events.iter().any(|event| event["type"] == "message_stop"));
    }

    #[tokio::test]
    async fn streaming_tool_calls_delay_start_until_id_and_name_ready() {
        let input = concat!(
            "data: {\"id\":\"chatcmpl_2\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"a\\\":\"}}]}}]}\n\n",
            "data: {\"id\":\"chatcmpl_2\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_0\",\"type\":\"function\",\"function\":{\"name\":\"first_tool\"}}]}}]}\n\n",
            "data: {\"id\":\"chatcmpl_2\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"1}\"}}]}}]}\n\n",
            "data: {\"id\":\"chatcmpl_2\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":6,\"completion_tokens\":2}}\n\n",
            "data: [DONE]\n\n"
        );

        let events = collect_events(input).await;
        let starts: Vec<&Value> = events
            .iter()
            .filter(|event| {
                event["type"] == "content_block_start"
                    && event["content_block"]["type"] == "tool_use"
            })
            .collect();

        assert_eq!(starts.len(), 1);
        assert_eq!(starts[0]["content_block"]["id"], "call_0");
        assert_eq!(starts[0]["content_block"]["name"], "first_tool");

        let deltas: Vec<&str> = events
            .iter()
            .filter(|event| {
                event["type"] == "content_block_delta"
                    && event["delta"]["type"] == "input_json_delta"
            })
            .filter_map(|event| {
                event
                    .pointer("/delta/partial_json")
                    .and_then(|v| v.as_str())
            })
            .collect();
        assert!(deltas.contains(&"{\"a\":"));
        assert!(deltas.contains(&"1}"));
    }

    #[tokio::test]
    async fn streaming_tool_calls_finish_reason_flushes_pending_and_closes_in_order() {
        let input = concat!(
            "data: {\"id\":\"chatcmpl_3\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"a\\\":1}\"}}]}}]}\n\n",
            "data: {\"id\":\"chatcmpl_3\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":3,\"completion_tokens\":1}}\n\n",
            "data: [DONE]\n\n"
        );

        let events = collect_events(input).await;
        let tool_start_pos = events
            .iter()
            .position(|event| {
                event["type"] == "content_block_start"
                    && event["content_block"]["type"] == "tool_use"
            })
            .expect("tool_use start emitted at finish_reason");
        let tool_delta_pos = events
            .iter()
            .position(|event| {
                event["type"] == "content_block_delta"
                    && event["delta"]["type"] == "input_json_delta"
            })
            .expect("pending args flushed");
        let tool_stop_pos = events
            .iter()
            .position(|event| event["type"] == "content_block_stop")
            .expect("tool block closed");
        let message_delta_pos = events
            .iter()
            .position(|event| event["type"] == "message_delta")
            .expect("message_delta emitted");

        assert_eq!(events[tool_start_pos]["content_block"]["id"], "tool_call_0");
        assert_eq!(events[tool_start_pos]["content_block"]["input"], json!({}));
        assert_eq!(
            events[tool_start_pos]["content_block"]["name"],
            "unknown_tool"
        );
        assert_eq!(events[tool_delta_pos]["delta"]["partial_json"], "{\"a\":1}");
        assert!(tool_start_pos < tool_delta_pos);
        assert!(tool_delta_pos < tool_stop_pos);
        assert!(tool_stop_pos < message_delta_pos);
    }

    #[tokio::test]
    async fn empty_content_delta_does_not_open_text_block() {
        let input = concat!(
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{\"content\":\"\"}}]}\n\n",
            "data: {\"id\":\"chatcmpl_1\",\"model\":\"gpt-4o\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":0}}\n\n",
            "data: [DONE]\n\n"
        );

        let events = collect_events(input).await;
        let text_block_starts: Vec<_> = events
            .iter()
            .filter(|event| {
                event["type"] == "content_block_start" && event["content_block"]["type"] == "text"
            })
            .collect();

        assert!(
            text_block_starts.is_empty(),
            "empty content deltas should not open text blocks"
        );
    }
}
