magi-code 0.63.0

use super::openai::{
    CODEX_MODEL_CATALOG_CLIENT_VERSION, extract_chatgpt_account_id,
    parse_codex_model_catalog_response,
};
use super::*;
use crate::agent::cancellation::{AgentCancellation, AgentCancellationHandle, is_run_canceled};
use base64::Engine;
use serde_json::{Value, json};
use std::{
    collections::VecDeque,
    io::{Read, Write},
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, AtomicUsize, Ordering},
    },
    time::{Duration, Instant},
};

#[derive(Default)]
struct CapturingTransport {
    chunks: Vec<String>,
}

impl HttpTransport for CapturingTransport {
    fn stream_json(
        &self,
        request: HttpRequest,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        // ponytail: test transport delegates cancellable path; cancellation ceiling is chunk-boundary polling.
        self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
    }

    fn stream_json_cancellable(
        &self,
        _request: HttpRequest,
        cancellation: &AgentCancellation,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        for chunk in &self.chunks {
            cancellation.check()?;
            on_chunk(chunk)?;
        }
        Ok(())
    }
    fn stream_json_cancellable_with_semantic_deadline(
        &self,
        request: HttpRequest,
        cancellation: &AgentCancellation,
        _semantic_deadline: &std::sync::atomic::AtomicU64,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        self.stream_json_cancellable(request, cancellation, on_chunk)
    }
}

enum ScriptStep {
    Chunks(Vec<&'static str>),
    ChunksThenError(Vec<&'static str>, anyhow::Error),
    Error(anyhow::Error),
}

struct ScriptedTransport {
    steps: Mutex<VecDeque<ScriptStep>>,
    attempts: Arc<Mutex<usize>>,
    requests: Arc<Mutex<Vec<HttpRequest>>>,
}

struct SlowChunkTransport {
    chunks: Vec<&'static str>,
    delay: Duration,
    attempts: Arc<Mutex<usize>>,
}

impl SlowChunkTransport {
    fn new(chunks: Vec<&'static str>, delay: Duration) -> Self {
        Self {
            chunks,
            delay,
            attempts: Arc::new(Mutex::new(0)),
        }
    }

    fn attempts_handle(&self) -> Arc<Mutex<usize>> {
        Arc::clone(&self.attempts)
    }
}

impl HttpTransport for SlowChunkTransport {
    fn stream_json(
        &self,
        request: HttpRequest,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        // ponytail: test transport delegates cancellable path; cancellation ceiling is chunk-boundary polling after sleep.
        self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
    }

    fn stream_json_cancellable(
        &self,
        _request: HttpRequest,
        cancellation: &AgentCancellation,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        *self.attempts.lock().unwrap() += 1;
        for chunk in &self.chunks {
            cancellation.check()?;
            std::thread::sleep(self.delay);
            cancellation.check()?;
            on_chunk(chunk)?;
        }
        Ok(())
    }
    fn stream_json_cancellable_with_semantic_deadline(
        &self,
        request: HttpRequest,
        cancellation: &AgentCancellation,
        _semantic_deadline: &std::sync::atomic::AtomicU64,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        self.stream_json_cancellable(request, cancellation, on_chunk)
    }
}

struct CancelingChunkTransport {
    cancel: Arc<AtomicBool>,
    chunks_seen: Arc<Mutex<usize>>,
}

impl CancelingChunkTransport {
    fn new(cancel: Arc<AtomicBool>) -> Self {
        Self {
            cancel,
            chunks_seen: Arc::new(Mutex::new(0)),
        }
    }

    fn chunks_seen_handle(&self) -> Arc<Mutex<usize>> {
        Arc::clone(&self.chunks_seen)
    }
}

impl HttpTransport for CancelingChunkTransport {
    fn stream_json(
        &self,
        request: HttpRequest,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        // ponytail: test transport delegates cancellable path; cancellation ceiling is intentional self-cancel fixture.
        self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
    }

    fn stream_json_cancellable(
        &self,
        _request: HttpRequest,
        cancellation: &AgentCancellation,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        cancellation.check()?;
        on_chunk(": keepalive\n\n")?;
        *self.chunks_seen.lock().unwrap() += 1;
        self.cancel.store(true, Ordering::SeqCst);
        cancellation.check()?;
        on_chunk(": keepalive\n\n")?;
        *self.chunks_seen.lock().unwrap() += 1;
        Ok(())
    }
    fn stream_json_cancellable_with_semantic_deadline(
        &self,
        request: HttpRequest,
        cancellation: &AgentCancellation,
        _semantic_deadline: &std::sync::atomic::AtomicU64,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        self.stream_json_cancellable(request, cancellation, on_chunk)
    }
}

impl ScriptedTransport {
    fn new(steps: Vec<ScriptStep>) -> Self {
        Self {
            steps: Mutex::new(steps.into()),
            attempts: Arc::new(Mutex::new(0)),
            requests: Arc::new(Mutex::new(Vec::new())),
        }
    }

    fn attempts_handle(&self) -> Arc<Mutex<usize>> {
        Arc::clone(&self.attempts)
    }

    fn requests_handle(&self) -> Arc<Mutex<Vec<HttpRequest>>> {
        Arc::clone(&self.requests)
    }
}

impl HttpTransport for ScriptedTransport {
    fn stream_json(
        &self,
        request: HttpRequest,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        // ponytail: test transport delegates cancellable path; cancellation ceiling is chunk-boundary polling.
        self.stream_json_cancellable(request, &AgentCancellation::default(), on_chunk)
    }

    fn stream_json_cancellable(
        &self,
        request: HttpRequest,
        cancellation: &AgentCancellation,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        cancellation.check()?;
        *self.attempts.lock().unwrap() += 1;
        self.requests.lock().unwrap().push(request);
        match self.steps.lock().unwrap().pop_front().unwrap() {
            ScriptStep::Chunks(chunks) => {
                for chunk in chunks {
                    cancellation.check()?;
                    on_chunk(chunk)?;
                }
                Ok(())
            }
            ScriptStep::ChunksThenError(chunks, error) => {
                for chunk in chunks {
                    cancellation.check()?;
                    on_chunk(chunk)?;
                }
                Err(error)
            }
            ScriptStep::Error(error) => Err(error),
        }
    }
    fn stream_json_cancellable_with_semantic_deadline(
        &self,
        request: HttpRequest,
        cancellation: &AgentCancellation,
        _semantic_deadline: &std::sync::atomic::AtomicU64,
        on_chunk: &mut dyn FnMut(&str) -> anyhow::Result<()>,
    ) -> anyhow::Result<()> {
        self.stream_json_cancellable(request, cancellation, on_chunk)
    }
}

fn provider_with_scripted_transport<T>(transport: T) -> OpenAiCompatibleProvider<T> {
    OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        transport,
    )
}

fn retryable_503_error() -> anyhow::Error {
    super::error::ProviderError::http_status(
            503,
            "provider request failed for https://provider.test/v1/chat/completions with status 503 Service Unavailable: upstream connect error or disconnect/reset before headers. reset reason: connection termination",
        )
        .into()
}

fn non_retryable_401_error() -> anyhow::Error {
    super::error::ProviderError::http_status(
            401,
            "provider request failed for https://provider.test/v1/chat/completions with status 401 Unauthorized: invalid credentials",
        )
        .into()
}

fn codex_session_expired_401_error() -> anyhow::Error {
    super::error::ProviderError::http_status(
            401,
            format!(
                "provider request failed for {CODEX_RESPONSES_URL} with status 401 Unauthorized: {{\"detail\":{{\"message\":\"Your ChatGPT session expired before this request finished.\"}},\"error\":{{\"message\":\"Your ChatGPT session expired before this request finished.\"}}}}"
            ),
        )
        .into()
}

fn cancel_after(delay: Duration, handle: AgentCancellationHandle) -> std::thread::JoinHandle<()> {
    std::thread::spawn(move || {
        std::thread::sleep(delay);
        handle.cancel();
    })
}

#[test]
fn provider_cancellation_during_header_wait_returns_promptly() {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let base_url = format!(
        "http://127.0.0.1:{}/v1",
        listener.local_addr().unwrap().port()
    );
    let release_server = Arc::new(AtomicBool::new(false));
    let server_release = Arc::clone(&release_server);
    let server = std::thread::spawn(move || {
        let (mut stream, _) = listener.accept().unwrap();
        stream
            .set_read_timeout(Some(Duration::from_secs(1)))
            .unwrap();
        let mut request_bytes = [0_u8; 2048];
        let _ = stream.read(&mut request_bytes);
        while !server_release.load(Ordering::SeqCst) {
            std::thread::sleep(Duration::from_millis(10));
        }
    });
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        None,
        base_url,
        false,
        ReqwestHttpTransport,
    );
    let (cancellation, handle): (AgentCancellation, AgentCancellationHandle) =
        AgentCancellation::default().child_token();
    let canceler = cancel_after(Duration::from_millis(50), handle);
    let started = Instant::now();

    let error = provider
        .stream_cancellable(
            ProviderRequest::new_without_tools("codex-test", vec![ChatMessage::user("hello")]),
            &cancellation,
            &mut |_| Ok(()),
        )
        .unwrap_err();

    release_server.store(true, Ordering::SeqCst);
    canceler.join().unwrap();
    server.join().unwrap();
    assert!(is_run_canceled(&error));
    assert!(started.elapsed() < Duration::from_secs(1));
}

#[test]
fn provider_cancellation_during_idle_read_returns_promptly() {
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let base_url = format!(
        "http://127.0.0.1:{}/v1",
        listener.local_addr().unwrap().port()
    );
    let release_server = Arc::new(AtomicBool::new(false));
    let server_release = Arc::clone(&release_server);
    let server = std::thread::spawn(move || {
        let (mut stream, _) = listener.accept().unwrap();
        stream
            .set_read_timeout(Some(Duration::from_secs(1)))
            .unwrap();
        let mut request_bytes = [0_u8; 2048];
        let _ = stream.read(&mut request_bytes);
        let headers = "HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ntransfer-encoding: chunked\r\nconnection: keep-alive\r\n\r\n";
        stream.write_all(headers.as_bytes()).unwrap();
        while !server_release.load(Ordering::SeqCst) {
            std::thread::sleep(Duration::from_millis(10));
        }
    });
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        None,
        base_url,
        false,
        ReqwestHttpTransport,
    );
    let (cancellation, handle): (AgentCancellation, AgentCancellationHandle) =
        AgentCancellation::default().child_token();
    let canceler = cancel_after(Duration::from_millis(50), handle);
    let started = Instant::now();

    let error = provider
        .stream_cancellable(
            ProviderRequest::new_without_tools("codex-test", vec![ChatMessage::user("hello")]),
            &cancellation,
            &mut |_| Ok(()),
        )
        .unwrap_err();

    release_server.store(true, Ordering::SeqCst);
    canceler.join().unwrap();
    server.join().unwrap();
    assert!(is_run_canceled(&error));
    assert!(started.elapsed() < Duration::from_secs(1));
}

#[test]
fn provider_retry_backoff_observes_cancellation() {
    let transport = ScriptedTransport::new(vec![
        ScriptStep::Error(retryable_503_error()),
        ScriptStep::Chunks(vec!["data: [DONE]\n\n"]),
    ]);
    let attempts = transport.attempts_handle();
    let provider = provider_with_scripted_transport(transport);
    let (cancellation, handle): (AgentCancellation, AgentCancellationHandle) =
        AgentCancellation::default().child_token();
    let canceler = cancel_after(Duration::from_millis(50), handle);
    let started = Instant::now();
    let mut events = Vec::new();

    let error = provider
        .stream_cancellable(
            ProviderRequest::new("codex-test", vec![]),
            &cancellation,
            &mut |event| {
                events.push(event);
                Ok(())
            },
        )
        .unwrap_err();

    canceler.join().unwrap();
    assert!(is_run_canceled(&error));
    assert_eq!(*attempts.lock().unwrap(), 1);
    assert!(events.is_empty());
    assert!(started.elapsed() < Duration::from_secs(1));
}

#[test]
fn provider_retries_retryable_http_status_before_events() {
    let transport = ScriptedTransport::new(vec![
        ScriptStep::Error(retryable_503_error()),
        ScriptStep::Error(retryable_503_error()),
        ScriptStep::Chunks(vec!["data: [DONE]\n\n"]),
    ]);
    let attempts = transport.attempts_handle();
    let provider = provider_with_scripted_transport(transport);
    let mut events = Vec::new();

    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();

    assert_eq!(*attempts.lock().unwrap(), 3);
    assert_eq!(events, vec![ProviderEvent::Done]);
}

#[test]
fn provider_does_not_retry_non_retryable_http_status() {
    let transport = ScriptedTransport::new(vec![
        ScriptStep::Error(non_retryable_401_error()),
        ScriptStep::Chunks(vec!["data: [DONE]\n\n"]),
    ]);
    let attempts = transport.attempts_handle();
    let provider = provider_with_scripted_transport(transport);
    let mut events = Vec::new();

    let error = provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap_err()
        .to_string();

    assert_eq!(*attempts.lock().unwrap(), 1);
    assert!(error.contains("401"), "{error}");
    assert!(events.is_empty());
}

#[test]
fn codex_session_expired_401_refreshes_and_retries_once() {
    let transport = ScriptedTransport::new(vec![
        ScriptStep::Error(codex_session_expired_401_error()),
        ScriptStep::Chunks(vec!["data: [DONE]\n\n"]),
    ]);
    let attempts = transport.attempts_handle();
    let requests = transport.requests_handle();
    let refreshes = Arc::new(AtomicUsize::new(0));
    let refreshes_for_callback = Arc::clone(&refreshes);
    let provider = OpenAiCodexProvider::new(
        "codex-test",
        "old-token",
        Some("old-acct".to_string()),
        transport,
    )
    .with_auth_refresh(move |_| {
        refreshes_for_callback.fetch_add(1, Ordering::SeqCst);
        Ok(CodexRefreshedAuth {
            access_token: "new-token".to_string(),
            account_id: Some("new-acct".to_string()),
        })
    });
    let request = ProviderRequest::new("codex-test", vec![ChatMessage::user("hello")]);
    let mut events = Vec::new();

    provider
        .stream(request, &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();

    assert_eq!(*attempts.lock().unwrap(), 2);
    assert_eq!(refreshes.load(Ordering::SeqCst), 1);
    let identities = events
        .iter()
        .filter_map(|event| match event {
            ProviderEvent::ResponseIdentity(identity) => Some(identity.attempt),
            _ => None,
        })
        .collect::<Vec<_>>();
    assert_eq!(identities, vec![1, 2]);
    events.retain(|event| !matches!(event, ProviderEvent::ResponseIdentity(_)));
    assert_eq!(events, vec![ProviderEvent::Done]);
    let requests = requests.lock().unwrap();
    assert_eq!(requests.len(), 2);
    assert_eq!(requests[0].body, requests[1].body);
    assert_eq!(requests[0].headers["authorization"], "Bearer old-token");
    assert_eq!(requests[0].headers["chatgpt-account-id"], "old-acct");
    assert_eq!(requests[1].headers["authorization"], "Bearer new-token");
    assert_eq!(requests[1].headers["chatgpt-account-id"], "new-acct");
}

#[test]
fn codex_session_expired_401_refresh_failure_returns_actionable_login_error() {
    let transport = ScriptedTransport::new(vec![
        ScriptStep::Error(codex_session_expired_401_error()),
        ScriptStep::Chunks(vec!["data: [DONE]\n\n"]),
    ]);
    let attempts = transport.attempts_handle();
    let refreshes = Arc::new(AtomicUsize::new(0));
    let refreshes_for_callback = Arc::clone(&refreshes);
    let provider = OpenAiCodexProvider::new(
        "codex-test",
        "old-token",
        Some("old-acct".to_string()),
        transport,
    )
    .with_auth_refresh(move |_| {
        refreshes_for_callback.fetch_add(1, Ordering::SeqCst);
        anyhow::bail!("refresh token rejected")
    });

    let error = provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |_| Ok(()))
        .unwrap_err()
        .to_string();

    assert_eq!(*attempts.lock().unwrap(), 1);
    assert_eq!(refreshes.load(Ordering::SeqCst), 1);
    assert!(
        error.contains("openai-codex OAuth credentials need re-login"),
        "{error}"
    );
    assert!(error.contains("/login openai-codex"), "{error}");
    assert!(error.contains("refresh token rejected"), "{error}");
}

#[test]
fn codex_session_expired_401_refresh_callback_preserves_cancellation_error() {
    let transport = ScriptedTransport::new(vec![
        ScriptStep::Error(codex_session_expired_401_error()),
        ScriptStep::Chunks(vec!["data: [DONE]\n\n"]),
    ]);
    let attempts = transport.attempts_handle();
    let refreshes = Arc::new(AtomicUsize::new(0));
    let refreshes_for_callback = Arc::clone(&refreshes);
    let (cancellation, handle) = AgentCancellation::default().child_token();
    let provider = OpenAiCodexProvider::new(
        "codex-test",
        "old-token",
        Some("old-acct".to_string()),
        transport,
    )
    .with_auth_refresh(move |cancellation| {
        refreshes_for_callback.fetch_add(1, Ordering::SeqCst);
        handle.cancel();
        cancellation.check()?;
        Ok(CodexRefreshedAuth {
            access_token: "new-token".to_string(),
            account_id: Some("new-acct".to_string()),
        })
    });

    let error = provider
        .stream_cancellable(
            ProviderRequest::new("codex-test", vec![]),
            &cancellation,
            &mut |_| Ok(()),
        )
        .unwrap_err();

    assert!(is_run_canceled(&error));
    assert_eq!(*attempts.lock().unwrap(), 1);
    assert_eq!(refreshes.load(Ordering::SeqCst), 1);
}

#[test]
fn codex_session_expired_401_after_semantic_progress_does_not_refresh_or_retry() {
    let transport = ScriptedTransport::new(vec![
        ScriptStep::ChunksThenError(
            vec!["data: {\"delta\":\"partial\"}\n\n"],
            codex_session_expired_401_error(),
        ),
        ScriptStep::Chunks(vec!["data: [DONE]\n\n"]),
    ]);
    let attempts = transport.attempts_handle();
    let refreshes = Arc::new(AtomicUsize::new(0));
    let refreshes_for_callback = Arc::clone(&refreshes);
    let provider = OpenAiCodexProvider::new(
        "codex-test",
        "old-token",
        Some("old-acct".to_string()),
        transport,
    )
    .with_auth_refresh(move |_| {
        refreshes_for_callback.fetch_add(1, Ordering::SeqCst);
        Ok(CodexRefreshedAuth {
            access_token: "new-token".to_string(),
            account_id: Some("new-acct".to_string()),
        })
    });
    let mut events = Vec::new();

    let error = provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap_err();
    assert!(
        error
            .downcast_ref::<super::error::ProviderError>()
            .is_some_and(super::error::ProviderError::is_codex_session_expired_401),
        "{error}"
    );
    let error = error.to_string();

    assert_eq!(*attempts.lock().unwrap(), 1);
    assert_eq!(refreshes.load(Ordering::SeqCst), 0);
    assert!(error.contains("Your ChatGPT session expired"), "{error}");
    events.retain(|event| !matches!(event, ProviderEvent::ResponseIdentity(_)));
    assert_eq!(
        events,
        vec![ProviderEvent::TextDelta("partial".to_string())]
    );
}

#[test]
fn provider_retries_failed_stream_response_before_events() {
    let transport = ScriptedTransport::new(vec![
        ScriptStep::Chunks(vec!["data: {\"type\":\"response.failed\"}\n\n"]),
        ScriptStep::Chunks(vec!["data: [DONE]\n\n"]),
    ]);
    let attempts = transport.attempts_handle();
    let provider = provider_with_scripted_transport(transport);
    let mut events = Vec::new();

    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();

    assert_eq!(*attempts.lock().unwrap(), 2);
    events.retain(|event| !matches!(event, ProviderEvent::ResponseIdentity(_)));
    assert_eq!(events, vec![ProviderEvent::Done]);
}

#[test]
fn provider_does_not_retry_retryable_stream_failure_after_text_delta() {
    let transport = ScriptedTransport::new(vec![
        ScriptStep::Chunks(vec![
            "data: {\"delta\":\"partial\"}\n\n",
            "data: {\"type\":\"response.failed\"}\n\n",
        ]),
        ScriptStep::Chunks(vec!["data: [DONE]\n\n"]),
    ]);
    let attempts = transport.attempts_handle();
    let provider = provider_with_scripted_transport(transport);
    let mut events = Vec::new();

    let error = provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap_err()
        .to_string();

    assert_eq!(*attempts.lock().unwrap(), 1);
    assert!(error.contains("failed or incomplete response"), "{error}");
    assert_eq!(
        events,
        vec![ProviderEvent::TextDelta("partial".to_string())]
    );
}

#[test]
fn provider_no_progress_timeout_ignores_sse_comments_and_empty_events() {
    let transport =
        SlowChunkTransport::new(vec![": keepalive\n\n", "\n\n"], Duration::from_millis(12));
    let attempts = transport.attempts_handle();
    let provider = provider_with_scripted_transport(transport);
    let mut events = Vec::new();

    let error = provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap_err()
        .to_string();

    assert_eq!(*attempts.lock().unwrap(), 3);
    events.retain(|event| !matches!(event, ProviderEvent::ResponseIdentity(_)));
    assert!(error.contains("no semantic progress"), "{error}");
    assert!(events.is_empty());
}

#[test]
fn provider_no_progress_timeout_can_be_extended_per_request() {
    let transport = SlowChunkTransport::new(
        vec![
            ": keepalive\n\n",
            "\n\n",
            "data: {\"delta\":\"ok\"}\n\n",
            "data: [DONE]\n\n",
        ],
        Duration::from_millis(12),
    );
    let attempts = transport.attempts_handle();
    let provider = provider_with_scripted_transport(transport);
    let mut events = Vec::new();

    provider
        .stream(
            ProviderRequest::new("codex-test", vec![])
                .with_semantic_progress_timeout(Duration::from_millis(100)),
            &mut |event| {
                events.push(event);
                Ok(())
            },
        )
        .unwrap();

    assert_eq!(*attempts.lock().unwrap(), 1);
    events.retain(|event| !matches!(event, ProviderEvent::ResponseIdentity(_)));
    assert_eq!(events[0], ProviderEvent::TextDelta("ok".to_string()));
    assert!(events.contains(&ProviderEvent::Done));
}

#[test]
fn provider_no_progress_timeout_resets_on_pending_tool_argument_delta() {
    let transport = SlowChunkTransport::new(
        vec![
            "data: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"function_call\",\"id\":\"fc_1\",\"call_id\":\"call_1\",\"name\":\"read\"}}\n\n",
            "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"delta\":\"{\\\"path\\\"\"}\n\n",
            "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"delta\":\":\\\"file.txt\\\"}\"}\n\n",
            "data: {\"type\":\"response.function_call_arguments.done\",\"item\":{\"type\":\"function_call\",\"id\":\"fc_1\",\"call_id\":\"call_1\",\"name\":\"read\",\"arguments\":\"{\\\"path\\\":\\\"file.txt\\\"}\"}}\n\n",
            "data: [DONE]\n\n",
        ],
        Duration::from_millis(5),
    );
    let attempts = transport.attempts_handle();
    let provider = provider_with_scripted_transport(transport);
    let mut events = Vec::new();

    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();

    assert_eq!(*attempts.lock().unwrap(), 1);
    assert!(events.iter().any(|event| matches!(
            event,
            ProviderEvent::ToolCall(call)
                if call.id == "call_1" && call.name == "read" && call.arguments == json!({"path":"file.txt"})
        )));
    assert!(events.contains(&ProviderEvent::Done));
}

#[test]
fn provider_cancellation_stops_raw_chunk_stream_without_parsed_events() {
    let cancel = Arc::new(AtomicBool::new(false));
    let transport = CancelingChunkTransport::new(Arc::clone(&cancel));
    let chunks_seen = transport.chunks_seen_handle();
    let provider = provider_with_scripted_transport(transport);
    let mut events = Vec::new();

    let error = provider
        .stream_cancellable(
            ProviderRequest::new("codex-test", vec![]),
            &AgentCancellation::new(cancel),
            &mut |event| {
                events.push(event);
                Ok(())
            },
        )
        .unwrap_err();

    assert!(is_run_canceled(&error));
    assert_eq!(*chunks_seen.lock().unwrap(), 1);
    assert!(events.is_empty());
}

#[test]
fn provider_does_not_retry_malformed_stream_json() {
    let transport = ScriptedTransport::new(vec![
        ScriptStep::Chunks(vec!["data: {bad}\n\n"]),
        ScriptStep::Chunks(vec!["data: [DONE]\n\n"]),
    ]);
    let attempts = transport.attempts_handle();
    let provider = provider_with_scripted_transport(transport);
    let mut events = Vec::new();

    let error = provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap_err()
        .to_string();

    assert_eq!(*attempts.lock().unwrap(), 1);
    assert!(
        error.contains("malformed provider SSE data JSON"),
        "{error}"
    );
    assert!(events.is_empty());
}

#[test]
fn utf8_dispatch_buffers_split_multibyte_sequences() {
    let text = "data: {\"delta\":\"snowman ☃\"}\n\n";
    let split = text.find('☃').unwrap() + 1;
    let bytes = text.as_bytes();
    let mut pending = Vec::new();
    let mut chunks = Vec::new();

    dispatch_utf8_bytes(&mut pending, &bytes[..split], &mut |chunk| {
        chunks.push(chunk.to_string());
        Ok(())
    })
    .unwrap();
    dispatch_utf8_bytes(&mut pending, &bytes[split..], &mut |chunk| {
        chunks.push(chunk.to_string());
        Ok(())
    })
    .unwrap();

    assert_eq!(chunks.concat(), text);
    assert!(pending.is_empty());
}

#[test]
fn request_construction_targets_custom_chat_completions_streaming() {
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        CapturingTransport::default(),
    );
    let request = ProviderRequest::new(
        "ignored",
        vec![ChatMessage::system("system"), ChatMessage::user("hello")],
    );
    let http = provider.build_http_request(&request);
    assert_eq!(http.method, "POST");
    assert_eq!(http.url, "https://provider.test/v1/chat/completions");
    assert_eq!(http.headers["authorization"], "Bearer secret");
    assert_eq!(http.body["model"], "codex-test");
    assert_eq!(http.body["stream"], true);
    assert_eq!(http.body["messages"][0]["role"], "system");
    assert_eq!(http.body["messages"][1]["content"], "hello");
    assert_eq!(http.body["tools"][0]["type"], "function");
    assert_eq!(http.body["tools"][0]["function"]["name"], "read");
}

fn assert_tool_schema_contract(tools: &Value) {
    let tools = tools.as_array().unwrap();
    assert_eq!(tools.len(), 14);
    assert_eq!(
        tools
            .iter()
            .map(|tool| tool["name"].as_str().unwrap())
            .collect::<Vec<_>>(),
        vec![
            "read",
            "view_image",
            "bash",
            "browser",
            "hash_edit",
            "write",
            "grep",
            "find",
            "list_files",
            "repo_map",
            "subagents",
            "web_search",
            "code_search",
            "ast_grep"
        ]
    );
    for tool in tools {
        assert_eq!(tool["type"], "function");
        assert!(tool["description"].as_str().unwrap().len() > 20);
        assert_eq!(tool["parameters"]["type"], "object");
        assert_eq!(tool["parameters"]["additionalProperties"], false);
    }
    assert_eq!(tools[0]["parameters"]["required"], json!(["paths"]));
    for keyword in ["oneOf", "anyOf", "allOf", "enum", "not"] {
        assert!(tools[0]["parameters"].get(keyword).is_none());
    }
    assert!(tools[0]["parameters"]["properties"].get("path").is_none());
    assert_eq!(tools[0]["parameters"]["properties"]["paths"]["minItems"], 1);
    assert_eq!(tools[0]["parameters"]["properties"]["paths"]["maxItems"], 8);
    assert_eq!(tools[0]["parameters"]["x-maxFiles"], 8);
    assert_eq!(tools[0]["parameters"]["properties"]["offset"]["minimum"], 1);
    assert_eq!(
        tools[0]["parameters"]["properties"]["limit"]["maximum"],
        2000
    );
    assert_eq!(tools[0]["parameters"]["x-maxBytes"], 1024 * 1024);
    assert_eq!(
        tools[1]["parameters"]["required"],
        json!(["path", "prompt"])
    );
    assert_eq!(tools[1]["parameters"]["x-maxImageBytes"], 5 * 1024 * 1024);
    assert_eq!(tools[1]["parameters"]["x-outputMaxBytes"], 64 * 1024);
    assert_eq!(tools[2]["parameters"]["required"], json!(["command"]));
    assert_eq!(
        tools[2]["parameters"]["properties"]["timeout"]["maximum"],
        300
    );
    assert_eq!(tools[2]["parameters"]["x-notSandboxed"], true);
    assert_eq!(tools[2]["parameters"]["x-stdoutMaxBytes"], 64 * 1024);
    assert_eq!(tools[2]["parameters"]["x-stderrMaxBytes"], 16 * 1024);
    assert_eq!(tools[3]["parameters"]["required"], json!(["command"]));
    assert_eq!(
        tools[3]["parameters"]["properties"]["mode"]["enum"],
        json!(["headless", "headed", "connect"])
    );
    assert_eq!(tools[3]["parameters"]["x-timeoutSeconds"], 60);
    assert_eq!(tools[4]["parameters"]["required"], json!(["input"]));
    assert_eq!(
        tools[4]["parameters"]["properties"]["input"]["type"],
        "string"
    );
    assert_eq!(tools[4]["parameters"]["x-editTextMaxBytes"], 256 * 1024);
    assert_eq!(
        tools[5]["parameters"]["required"],
        json!(["path", "content"])
    );
    assert_eq!(
        tools[5]["parameters"]["properties"]["content"]["maxLength"],
        1024 * 1024
    );
    assert_eq!(tools[6]["parameters"]["required"], json!(["pattern"]));
    assert_eq!(
        tools[6]["parameters"]["properties"]["context"]["minimum"],
        0
    );
    assert_eq!(tools[7]["parameters"]["required"], json!(["query"]));
    assert_eq!(tools[7]["parameters"]["x-defaultLimit"], 50);
    assert_eq!(
        tools[7]["parameters"]["properties"]["limit"]["maximum"],
        200
    );
    assert_eq!(tools[8]["parameters"]["required"], json!(["path"]));
    assert_eq!(tools[8]["parameters"]["properties"]["path"]["minLength"], 1);
    assert_eq!(tools[9]["parameters"]["additionalProperties"], false);
    assert_eq!(
        tools[9]["parameters"]["properties"]["query"]["maxLength"],
        512
    );
    assert_eq!(
        tools[9]["parameters"]["properties"]["maxTokens"]["minimum"],
        512
    );
    assert_eq!(
        tools[9]["parameters"]["properties"]["maxTokens"]["maximum"],
        20_000
    );
    assert_eq!(tools[9]["parameters"]["x-defaultMaxTokens"], 4096);
    assert_eq!(tools[10]["parameters"]["required"], json!(["tasks"]));
    assert_eq!(
        tools[10]["parameters"]["properties"]["tasks"]["maxItems"],
        10
    );
    assert_eq!(
        tools[10]["parameters"]["properties"]["concurrency"]["maximum"],
        4
    );
    assert_eq!(
        tools[11]["parameters"]["properties"]["queries"]["maxItems"],
        5
    );
    assert_eq!(
        tools[11]["parameters"]["properties"]["numResults"]["maximum"],
        10
    );
    assert_eq!(tools[12]["parameters"]["required"], json!(["query"]));
    assert_eq!(
        tools[12]["parameters"]["properties"]["maxTokens"]["maximum"],
        10_000
    );
}

fn fake_jwt(account_id: &str) -> String {
    let header = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(r#"{"alg":"none"}"#);
    let payload = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(format!(
        r#"{{"https://api.openai.com/auth.chatgpt_account_id":"{account_id}"}}"#
    ));
    format!("{header}.{payload}.")
}

fn pi_style_fake_jwt(account_id: &str) -> String {
    let header = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(r#"{"alg":"none"}"#);
    let payload = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(format!(
        r#"{{"https://api.openai.com/auth":{{"chatgpt_account_id":"{account_id}"}}}}"#
    ));
    format!("{header}.{payload}.")
}

#[test]
fn codex_compatible_request_construction_targets_chatgpt_backend() {
    let token = fake_jwt("acct_123");
    let provider = OpenAiCodexProvider::new(
        "codex-test",
        token.clone(),
        None,
        CapturingTransport::default(),
    );
    let request = ProviderRequest::new(
        "ignored",
        vec![ChatMessage::system("system"), ChatMessage::user("hello")],
    );
    let http = provider.build_http_request(&request).unwrap();
    assert_eq!(http.method, "POST");
    assert_eq!(http.url, CODEX_RESPONSES_URL);
    assert_eq!(http.headers["authorization"], format!("Bearer {token}"));
    assert_eq!(http.headers["chatgpt-account-id"], "acct_123");
    assert_eq!(http.headers["originator"], "codex_cli_rs");
    assert_eq!(http.headers["user-agent"], "codex_cli_rs/0.144.0");
    assert_eq!(http.headers["openai-beta"], "responses=experimental");
    assert_eq!(http.body["model"], "codex-test");
    assert_eq!(http.body["store"], false);
    assert_eq!(http.body["stream"], true);
    assert_eq!(http.body["instructions"], "system");
    assert_eq!(http.body["input"][0]["role"], "user");
    assert_eq!(http.body["include"][0], "reasoning.encrypted_content");
    assert_eq!(http.body["tool_choice"], "auto");
    assert_eq!(http.body["parallel_tool_calls"], true);
    assert_tool_schema_contract(&http.body["tools"]);
}

#[test]
fn codex_compatible_request_normalizes_gpt_5_6_short_aliases() {
    let token = fake_jwt("acct_123");
    for (alias, canonical) in [
        ("luna", "gpt-5.6-luna"),
        ("terra", "gpt-5.6-terra"),
        ("sol", "gpt-5.6-sol"),
        ("gpt-5.6", "gpt-5.6-sol"),
    ] {
        let provider =
            OpenAiCodexProvider::new(alias, token.clone(), None, CapturingTransport::default());
        let request = ProviderRequest::new("ignored", vec![ChatMessage::user("hello")]);

        let http = provider.build_http_request(&request).unwrap();

        assert_eq!(http.body["model"], canonical);
    }
}

#[test]
fn codex_model_catalog_request_uses_chatgpt_models_endpoint_and_oauth_headers() {
    let token = fake_jwt("acct_123");
    let provider = OpenAiCodexProvider::new(
        "codex-test",
        token.clone(),
        None,
        CapturingTransport::default(),
    );
    let http = provider.build_model_catalog_request().unwrap();
    assert_eq!(http.method, "GET");
    assert_eq!(
        http.url,
        format!("{CODEX_MODELS_URL}?client_version={CODEX_MODEL_CATALOG_CLIENT_VERSION}")
    );
    assert_ne!(
        CODEX_MODEL_CATALOG_CLIENT_VERSION,
        env!("CARGO_PKG_VERSION")
    );
    assert_eq!(http.headers["authorization"], format!("Bearer {token}"));
    assert_eq!(http.headers["chatgpt-account-id"], "acct_123");
    assert_eq!(http.headers["originator"], "codex_cli_rs");
    assert_eq!(http.headers["user-agent"], "codex_cli_rs/0.144.0");
    assert_eq!(http.headers["accept"], "application/json");
    assert_eq!(http.body, Value::Null);
}

#[test]
fn codex_model_catalog_parser_skips_unusable_entries_and_ignores_unknown_fields() {
    let entries = parse_codex_model_catalog_response(
            r#"{"models":[{"slug":"gpt-5.5","display_name":"GPT 5.5","description":"fast","supported_in_api":true,"context_window":123,"unknown":"ignored"},{"slug":""},{"display_name":"missing"},{"slug":"unsupported","supported_in_api":false},{"slug":"codex-auto-review","visibility":"hide"}]}"#,
        )
        .unwrap();
    assert_eq!(entries.len(), 1);
    assert_eq!(entries[0].id, "openai-codex/gpt-5.5");
    assert_eq!(entries[0].display_name.as_deref(), Some("GPT 5.5"));
    assert_eq!(entries[0].description.as_deref(), Some("fast"));
    assert_eq!(entries[0].context_window, Some(123));
}

#[test]
fn codex_model_catalog_parser_preserves_multiple_visible_api_models() {
    let entries = parse_codex_model_catalog_response(
            r#"{"models":[{"slug":"gpt-5.2","supported_in_api":true},{"slug":"gpt-5.5","supported_in_api":true,"max_context_window":456},{"slug":"o3","visibility":"show"}]}"#,
        )
        .unwrap();

    assert_eq!(
        entries
            .iter()
            .map(|entry| entry.id.as_str())
            .collect::<Vec<_>>(),
        vec![
            "openai-codex/gpt-5.2",
            "openai-codex/gpt-5.5",
            "openai-codex/o3"
        ]
    );
    assert_eq!(entries[1].context_window, Some(456));
    assert_eq!(entries[1].max_context_window, Some(456));
}

#[test]
fn codex_model_catalog_parser_filters_hidden_visibility_values_case_insensitively() {
    let entries = parse_codex_model_catalog_response(
            r#"{"models":[{"slug":"visible-fixture","visibility":"show"},{"slug":"hide-fixture","visibility":"hide"},{"slug":"hidden-fixture","visibility":"hidden"},{"slug":"case-hidden-fixture","visibility":"HiDdEn"}]}"#,
        )
        .unwrap();

    assert_eq!(
        entries
            .iter()
            .map(|entry| entry.id.as_str())
            .collect::<Vec<_>>(),
        vec!["openai-codex/visible-fixture"]
    );
}

#[test]
fn codex_token_is_not_routed_to_standard_openai_endpoint() {
    let token = fake_jwt("acct_123");
    let provider =
        OpenAiCodexProvider::new("codex-test", token, None, CapturingTransport::default());
    let http = provider
        .build_http_request(&ProviderRequest::new(
            "ignored",
            vec![ChatMessage::user("hello")],
        ))
        .unwrap();
    assert_ne!(http.url, "https://provider.test/v1/chat/completions");
    assert_eq!(http.url, CODEX_RESPONSES_URL);
}

#[test]
fn invalid_codex_token_fails_before_transport() {
    let provider = OpenAiCodexProvider::new(
        "codex-test",
        "not-a-jwt",
        None,
        CapturingTransport::default(),
    );
    let error = provider
        .build_http_request(&ProviderRequest::new("ignored", vec![]))
        .unwrap_err()
        .to_string();
    assert!(error.contains("missing ChatGPT account id"));
}

#[test]
fn account_id_extracts_from_fake_jwt() {
    assert_eq!(
        extract_chatgpt_account_id(&fake_jwt("acct_456")).unwrap(),
        "acct_456"
    );
    assert_eq!(
        extract_chatgpt_account_id(&pi_style_fake_jwt("acct_pi")).unwrap(),
        "acct_pi"
    );
}

#[test]
fn request_construction_includes_tool_result_continuation_items() {
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        CapturingTransport::default(),
    );
    let request = ProviderRequest::new(
        "ignored",
        vec![ChatMessage::system("system"), ChatMessage::user("hello")],
    )
    .with_tool_results(vec![ProviderToolResult {
        call_id: "call_1".to_string(),
        tool_name: "read".to_string(),
        success: true,
        output: "file text".to_string(),
    }]);

    let http = provider.build_http_request(&request);

    assert_eq!(http.body["messages"][2]["role"], "tool");
    assert_eq!(http.body["messages"][2]["tool_call_id"], "call_1");
    assert_eq!(http.body["messages"][2]["content"], "file text");
}

#[test]
fn request_construction_preserves_function_call_before_tool_output() {
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        CapturingTransport::default(),
    );
    let request = ProviderRequest::new(
        "ignored",
        vec![ChatMessage::system("system"), ChatMessage::user("hello")],
    )
    .with_response_items(vec![json!({
        "type": "function_call",
        "call_id": "call_1",
        "name": "read",
        "arguments": "{\"path\":\"file.txt\"}",
        "status": "completed"
    })])
    .with_tool_results(vec![ProviderToolResult {
        call_id: "call_1".to_string(),
        tool_name: "read".to_string(),
        success: true,
        output: "file text".to_string(),
    }]);

    let http = provider.build_http_request(&request);

    assert_eq!(http.body["messages"][2]["role"], "assistant");
    assert_eq!(http.body["messages"][2]["tool_calls"][0]["id"], "call_1");
    assert_eq!(
        http.body["messages"][2]["tool_calls"][0]["function"]["name"],
        "read"
    );
    assert_eq!(http.body["messages"][3]["role"], "tool");
    assert_eq!(http.body["messages"].as_array().unwrap().len(), 4);
    for message in http.body["messages"].as_array().unwrap() {
        assert!(message.get("role").is_some());
        assert!(message.get("type").is_none());
        assert!(message.get("call_id").is_none());
        assert!(message.get("arguments").is_none());
        assert!(message.get("status").is_none());
        assert!(message.get("output").is_none());
    }
}

#[test]
fn custom_chat_completion_requests_do_not_add_openai_reasoning_effort() {
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "gpt-5.5",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        CapturingTransport::default(),
    );
    let http = provider.build_http_request(&ProviderRequest::new(
        "ignored",
        vec![ChatMessage::user("hello")],
    ));

    assert_eq!(http.body["model"], "gpt-5.5");
    assert!(http.body.get("reasoning").is_none());
}

#[test]
fn stream_parser_emits_text_and_usage_events_without_double_done() {
    let transport = CapturingTransport {
        chunks: vec![
            concat!(
                "data: {\"delta\":\"hi\",",
                "\"usage\":{\"input_tokens\":2,\"output_tokens\":3,\"total_tokens\":5}}\n\n",
                "data: [DONE]\n"
            )
            .to_string(),
        ],
    };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        transport,
    );
    let mut events = Vec::new();
    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();
    assert!(events.contains(&ProviderEvent::TextDelta("hi".to_string())));
    assert!(events.contains(&ProviderEvent::Usage(Usage {
        input: 2,
        output: 3,
        cache_read: 0,
        cache_write: 0,
        total: 5,
        reasoning_tokens: None,
    })));
    assert_eq!(
        events
            .iter()
            .filter(|event| **event == ProviderEvent::Done)
            .count(),
        1
    );
}

#[test]
fn stream_parser_emits_chat_completion_text_delta() {
    let transport = CapturingTransport {
        chunks: vec![
            concat!(
                "data: {\"choices\":[{\"delta\":{\"content\":\"hi\"}}]}\n\n",
                "data: [DONE]\n\n"
            )
            .to_string(),
        ],
    };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        transport,
    );
    let mut events = Vec::new();
    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();

    assert_eq!(
        events,
        vec![
            ProviderEvent::TextDelta("hi".to_string()),
            ProviderEvent::Done
        ]
    );
}

#[test]
fn stream_parser_emits_chat_completion_tool_calls_and_continuation_item() {
    let chunk = format!(
        "data: {}\n\ndata: {}\n\ndata: {}\n\ndata: [DONE]\n\n",
        json!({"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"read","arguments":"{\"path\":"}}]}}]}),
        json!({"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"src/lib.rs\"}"}}]}}]}),
        json!({"choices":[{"finish_reason":"tool_calls"}]})
    );
    let transport = CapturingTransport {
        chunks: vec![chunk],
    };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        transport,
    );
    let mut events = Vec::new();
    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();

    assert!(events.contains(&ProviderEvent::ResponseItem(json!({
        "role": "assistant",
        "content": null,
        "tool_calls": [{
            "id": "call_1",
            "type": "function",
            "function": {
                "name": "read",
                "arguments": "{\"path\":\"src/lib.rs\"}",
            }
        }]
    }))));
    assert!(events.contains(&ProviderEvent::ToolCall(ToolCall {
        id: "call_1".to_string(),
        name: "read".to_string(),
        arguments: json!({"path":"src/lib.rs"}),
    })));
}

#[test]
fn responses_endpoint_provider_streams_response_text_events() {
    let transport = CapturingTransport {
            chunks: vec![concat!(
                "data: {\"type\":\"response.output_text.delta\",\"delta\":\"MC_LIVE_OK_2026\"}\n\n",
                "data: {\"type\":\"response.output_text.done\",\"text\":\"MC_LIVE_OK_2026\"}\n\n",
                "data: {\"type\":\"response.completed\",\"response\":{\"usage\":{\"input_tokens\":2,\"output_tokens\":3,\"total_tokens\":5}}}\n\n",
                "data: [DONE]\n\n"
            )
            .to_string()],
        };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        true,
        transport,
    );
    let mut text = String::new();
    let mut events = Vec::new();
    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            if let ProviderEvent::TextDelta(delta) = &event {
                text.push_str(delta);
            }
            events.push(event);
            Ok(())
        })
        .unwrap();

    assert_eq!(text, "MC_LIVE_OK_2026");
    assert_eq!(
        events
            .iter()
            .filter(|event| matches!(event, ProviderEvent::TextDelta(_)))
            .count(),
        1
    );
    assert!(events.contains(&ProviderEvent::Usage(Usage {
        input: 2,
        output: 3,
        cache_read: 0,
        cache_write: 0,
        total: 5,
        reasoning_tokens: None,
    })));
    assert_eq!(
        events
            .iter()
            .filter(|event| **event == ProviderEvent::Done)
            .count(),
        1
    );
}

#[test]
fn responses_endpoint_provider_streams_reasoning_summary_events() {
    let transport = CapturingTransport {
        chunks: vec![
            concat!(
                "data: {\"type\":\"response.output_item.done\",",
                "\"item\":{\"id\":\"rs_1\",\"type\":\"reasoning\",",
                "\"summary\":[{\"type\":\"summary_text\",\"text\":\"checked options\"}],",
                "\"encrypted_content\":\"opaque\"}}\n\n",
                "data: {\"type\":\"response.completed\"}\n\n"
            )
            .to_string(),
        ],
    };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        true,
        transport,
    );
    let mut events = Vec::new();
    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();

    assert!(events.iter().any(|event| matches!(
        event,
        ProviderEvent::ReasoningSummaryCompleteIdentified(ReasoningSummary {
            text,
            item_id: Some(item_id),
            ..
        }) if text == "checked options" && item_id == "rs_1"
    )));
    assert!(events.iter().any(|event| matches!(
        event,
        ProviderEvent::ResponseItem(item)
            if item.get("encrypted_content").and_then(Value::as_str) == Some("opaque")
    )));
    assert!(events.contains(&ProviderEvent::Done));
}

#[test]
fn responses_endpoint_provider_streams_function_call_events() {
    let transport = CapturingTransport {
            chunks: vec![concat!(
                "data: {\"type\":\"response.output_item.done\",",
                "\"item\":{\"type\":\"function_call\",\"call_id\":\"call_1\",",
                "\"name\":\"shell\",\"arguments\":\"{\\\"cmd\\\":\\\"pwd\\\"}\"}}\n\n",
                "data: {\"response\":{\"output\":[{\"type\":\"function_call\",",
                "\"call_id\":\"call_2\",\"name\":\"read\",\"arguments\":{\"path\":\"src/lib.rs\"}}]}}\n\n",
                "data: [DONE]\n\n"
            )
            .to_string()],
        };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        true,
        transport,
    );
    let mut events = Vec::new();
    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();
    assert!(events.contains(&ProviderEvent::ToolCall(ToolCall {
        id: "call_2".to_string(),
        name: "read".to_string(),
        arguments: json!({"path":"src/lib.rs"}),
    })));
}

#[test]
fn stream_parser_buffers_split_sse_data_events_across_chunks() {
    let transport = CapturingTransport {
        chunks: vec![
            "data: {\"delta\":\"hel".to_string(),
            "lo\"}\n\ndata: [DO".to_string(),
            "NE]\n\n".to_string(),
        ],
    };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        transport,
    );
    let mut events = Vec::new();
    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();
    assert_eq!(
        events,
        vec![
            ProviderEvent::TextDelta("hello".to_string()),
            ProviderEvent::Done,
        ]
    );
}

#[test]
fn responses_tool_call_sequence_with_item_id_emits_full_arguments() {
    let transport = CapturingTransport {
        chunks: vec![
            concat!(
                "data: {\"type\":\"response.output_item.added\",",
                "\"item\":{\"type\":\"function_call\",\"id\":\"fc_1\",",
                "\"call_id\":\"call_1\",\"name\":\"bash\",\"arguments\":\"\"}}\n\n",
                "data: {\"type\":\"response.function_call_arguments.delta\",",
                "\"item_id\":\"fc_1\",\"delta\":\"{\\\"command\\\":\"}\n\n",
                "data: {\"type\":\"response.function_call_arguments.done\",",
                "\"item_id\":\"fc_1\",\"arguments\":\"{\\\"command\\\":\\\"pwd\\\"}\"}\n\n",
                "data: {\"type\":\"response.completed\"}\n\n"
            )
            .to_string(),
        ],
    };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        transport,
    );
    let mut events = Vec::new();
    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();
    assert_eq!(
        events,
        vec![
            ProviderEvent::ToolCall(ToolCall {
                id: "call_1".to_string(),
                name: "bash".to_string(),
                arguments: json!({"command":"pwd"}),
            }),
            ProviderEvent::Done,
        ]
    );
}

#[test]
fn responses_tool_call_sequence_emits_once_when_arguments_complete() {
    let transport = CapturingTransport {
        chunks: vec![
            concat!(
                "data: {\"type\":\"response.output_item.added\",",
                "\"item\":{\"type\":\"function_call\",\"call_id\":\"call_1\",",
                "\"name\":\"shell\"}}\n\n",
                "data: {\"type\":\"response.function_call_arguments.delta\",",
                "\"call_id\":\"call_1\",\"delta\":\"{\\\"cmd\\\":\"}\n\n"
            )
            .to_string(),
            concat!(
                "data: {\"type\":\"response.function_call_arguments.delta\",",
                "\"call_id\":\"call_1\",\"delta\":\"\\\"pwd\\\"}\"}\n\n",
                "data: {\"type\":\"response.function_call_arguments.done\",",
                "\"call_id\":\"call_1\"}\n\n",
                "data: [DONE]\n\n"
            )
            .to_string(),
        ],
    };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        transport,
    );
    let mut events = Vec::new();
    provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap();
    assert_eq!(
        events,
        vec![
            ProviderEvent::ToolCall(ToolCall {
                id: "call_1".to_string(),
                name: "shell".to_string(),
                arguments: json!({"cmd":"pwd"}),
            }),
            ProviderEvent::Done,
        ]
    );
}

#[test]
fn stream_parser_errors_on_provider_clean_eof_after_text_delta() {
    let transport = CapturingTransport {
        chunks: vec!["data: {\"delta\":\"partial\"}\n\n".to_string()],
    };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        transport,
    );
    let mut events = Vec::new();
    let error = provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap_err()
        .to_string();

    assert!(error.contains("missing provider stream completion"));
    assert_eq!(
        events,
        vec![ProviderEvent::TextDelta("partial".to_string())]
    );
    assert!(!events.contains(&ProviderEvent::Done));
}

#[test]
fn stream_parser_does_not_accept_item_level_completion_as_provider_terminal() {
    let transport = CapturingTransport {
        chunks: vec![
            concat!(
                "data: {\"type\":\"response.output_item.done\",",
                "\"item\":{\"type\":\"function_call\",\"call_id\":\"call_1\",",
                "\"name\":\"shell\",\"arguments\":\"{\\\"cmd\\\":\\\"pwd\\\"}\",",
                "\"status\":\"completed\"}}\n\n"
            )
            .to_string(),
        ],
    };
    let provider = OpenAiCompatibleProvider::custom(
        "test",
        "codex-test",
        Some("secret".to_string()),
        "https://provider.test/v1",
        false,
        transport,
    );
    let mut events = Vec::new();
    let error = provider
        .stream(ProviderRequest::new("codex-test", vec![]), &mut |event| {
            events.push(event);
            Ok(())
        })
        .unwrap_err()
        .to_string();

    assert!(error.contains("missing provider stream completion"));
    assert!(events.contains(&ProviderEvent::ToolCall(ToolCall {
        id: "call_1".to_string(),
        name: "shell".to_string(),
        arguments: json!({"cmd":"pwd"}),
    })));
    assert!(!events.contains(&ProviderEvent::Done));
}

#[test]
fn custom_reasoning_protocols_keep_openai_stream_parser_semantics() {
    for protocol in [
        crate::config::CustomReasoningProtocol::GptLike,
        crate::config::CustomReasoningProtocol::AnthropicLike,
    ] {
        let transport = CapturingTransport {
            chunks: vec![
                "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"think\"}}]}\n\n"
                    .to_string(),
                "data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}]}\n\n".to_string(),
                "data: {\"usage\":{\"input_tokens\":2,\"output_tokens\":3,\"total_tokens\":5}}\n\n"
                    .to_string(),
                "data: [DONE]\n\n".to_string(),
            ],
        };
        let provider = OpenAiCompatibleProvider::custom(
            "custom",
            "model-a",
            None,
            "https://provider.test/v1",
            false,
            transport,
        )
        .with_reasoning_protocol(protocol, None);
        let mut events = Vec::new();
        provider
            .stream(ProviderRequest::new("model-a", vec![]), &mut |event| {
                events.push(event);
                Ok(())
            })
            .unwrap();
        assert!(events.contains(&ProviderEvent::ReasoningSummaryDelta("think".to_string())));
        assert!(events.contains(&ProviderEvent::TextDelta("hello".to_string())));
        assert!(events.contains(&ProviderEvent::Usage(Usage {
            input: 2,
            output: 3,
            cache_read: 0,
            cache_write: 0,
            total: 5,
            reasoning_tokens: None,
        })));
        assert_eq!(
            events
                .iter()
                .filter(|event| **event == ProviderEvent::Done)
                .count(),
            1
        );
    }
}