Method process_stream

crates/opencode-session/src/llm.rs:386–737

Process a single stream attempt.

(
        &mut self,
        input: &StreamInput,
        provider: Arc<dyn Provider>,
    )

    /// Process a single stream attempt.
    async fn process_stream(
        &mut self,
        input: &StreamInput,
        provider: Arc<dyn Provider>,
    ) -> anyhow::Result<()> {
        let mut output = LlmProcessor::stream_from_input(input, provider).await?;

        while let Some(event) = output.events.recv().await {
            if input.abort.is_cancelled() {
                return Err(anyhow::anyhow!("Stream aborted"));
            }

            match event {
                StreamEvent::Start => {
                    // Session is now busy — already set by caller
                }

                // ============================================================
                // Reasoning events
                // ============================================================
                StreamEvent::ReasoningStart { id } => {
                    if self.reasoning_map.contains_key(&id) {
                        continue;
                    }
                    let now = chrono::Utc::now().timestamp_millis();
                    let part_id =
                        opencode_core::id::create(opencode_core::id::Prefix::Part, true, None);
                    self.reasoning_map.insert(
                        id,
                        ReasoningState {
                            _part_id: part_id,
                            text: String::new(),
                            _start_time: now,
                            _metadata: None,
                        },
                    );
                }

                StreamEvent::ReasoningDelta { id, text } => {
                    if let Some(state) = self.reasoning_map.get_mut(&id) {
                        state.text.push_str(&text);
                    }
                }

                StreamEvent::ReasoningEnd { id } => {
                    if let Some(mut state) = self.reasoning_map.remove(&id) {
                        state.text = state.text.trim_end().to_string();
                    }
                }

                // ============================================================
                // Tool input streaming events
                // ============================================================
                StreamEvent::ToolInputStart { id, tool_name } => {
                    let part_id = self
                        .tool_calls
                        .get(&id)
                        .map(|tc| tc.part_id.clone())