Method send_streaming

core/src/llm/openai.rs:544–1106

Execute a fully-built streaming chat-completions request (sets `stream`).

(
        &self,
        mut request: serde_json::Value,
        cancel_token: tokio_util::sync::CancellationToken,
    )

impl OpenAiClient {
    /// Execute a fully-built streaming chat-completions request (sets `stream`).
    async fn send_streaming(
        &self,
        mut request: serde_json::Value,
        cancel_token: tokio_util::sync::CancellationToken,
    ) -> Result<mpsc::Receiver<StreamEvent>> {
        {
            request["stream"] = serde_json::json!(true);
            request["stream_options"] = serde_json::json!({ "include_usage": true });
            let request_started_at = Instant::now();
            let url = format!("{}{}", self.base_url, self.chat_completions_path);
            let request_headers = self.request_headers();

            let streaming_resp = crate::retry::with_retry(&self.retry_config, |_attempt| {
                let http = &self.http;
                let url = &url;
                let request_headers = request_headers.clone();
                let request = &request;
                let cancel_token = cancel_token.clone();
                async move {
                    let headers = request_headers
                        .iter()
                        .map(|(key, value)| (key.as_str(), value.as_str()))
                        .collect::<Vec<_>>();
                    // Wrap in tokio::select! so cancellation aborts the HTTP request mid-flight
                    let resp = tokio::select! {
                        _ = cancel_token.cancelled() => {
                            return AttemptOutcome::Fatal(anyhow::anyhow!("HTTP request cancelled"));
                        }
                        result = http.post_streaming(url, headers, request, cancel_token.clone()) => {
                            match result {
                                Ok(r) => r,
                                Err(e) => {
                                    // Transient network error (timeout, reset,
                                    // mid-flight drop — common on throttled
                                    // endpoints): retry with backoff like 429/5xx
                                    // instead of failing the turn. GLM and other
                                    // OpenAI-compatible endpoints hit this most.
                                    return if crate::retry::is_transient_error(&e) {
                                        AttemptOutcome::Retryable {
                                            status: reqwest::StatusCode::SERVICE_UNAVAILABLE,
                                            body: format!("network error: {e}"),
                                            retry_after: None,
                                        }
                                    } else {
                                        AttemptOutcome::Fatal(anyhow::anyhow!(
                                            "HTTP request failed: {}",
                                            e
                                        ))
                                    };
                                }
                            }
                        }
                    };
                    let status = reqwest::StatusCode::from_u16(resp.status)
                        .unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
                    if status.is_success() {
                        AttemptOutcome::Success(resp)
                    } else {