Method call_llm

core/src/agent/llm_turn.rs:348–430

Call the LLM, handling streaming vs non-streaming internally. Streaming events (`TextDelta`, `ToolStart`) are forwarded to `event_tx` as they arrive. Non-streaming mode simply awaits the complete response. Tool definitions are selected per turn by the centralized tool selector. Returns `Err` on any LLM API failure. The circuit breaker in `execute_loop` wraps this call with retry logic for non-s

(
        &self,
        messages: &[Message],
        system: Option<&str>,
        event_tx: &Option<mpsc::Sender<AgentEvent>>,
        cancel_token: &tokio_util::sync::CancellationToken,
    )

    /// Returns `Err` on any LLM API failure. The circuit breaker in
    /// `execute_loop` wraps this call with retry logic for non-streaming mode.
    async fn call_llm(
        &self,
        messages: &[Message],
        system: Option<&str>,
        event_tx: &Option<mpsc::Sender<AgentEvent>>,
        cancel_token: &tokio_util::sync::CancellationToken,
    ) -> anyhow::Result<LlmResponse> {
        let tools = crate::tools::select_tools_for_messages(&self.config.tools, messages);

        if event_tx.is_some() {
            let mut stream_rx = match self
                .llm_client
                .complete_streaming(messages, system, &tools, cancel_token.clone())
                .await
            {
                Ok(rx) => rx,
                Err(stream_error) => {
                    // Do not fall back to non-streaming if cancelled — propagate cancellation
                    if cancel_token.is_cancelled() {
                        anyhow::bail!("Operation cancelled by user");
                    }
                    tracing::warn!(
                        error = %stream_error,
                        "LLM streaming setup failed; falling back to non-streaming completion"
                    );
                    return self
                        .llm_client
                        .complete(messages, system, &tools)
                        .await
                        .with_context(|| {
                            format!(
                                "LLM streaming call failed ({stream_error}); non-streaming fallback also failed"
                            )
                        });
                }
            };

            let mut final_response: Option<LlmResponse> = None;
            loop {
                tokio::select! {
                    _ = cancel_token.cancelled() => {
                        tracing::info!("🛑 LLM streaming cancelled by CancellationToken");
                        anyhow::bail!("Operation cancelled by user");
                    }
                    event = stream_rx.recv() => {
                        match event {
                            Some(crate::llm::StreamEvent::TextDelta(text)) => {
                                if let Some(tx) = event_tx {
                                    tx.send(AgentEvent::TextDelta { text }).await.ok();
                                }
                            }
                            Some(crate::llm::StreamEvent::ReasoningDelta(text)) => {
                                if let Some(tx) = event_tx {
                                    tx.send(AgentEvent::ReasoningDelta { text }).await.ok();
                                }
                            }
                            Some(crate::llm::StreamEvent::ToolUseStart { id, name }) => {
                                if let Some(tx) = event_tx {