rho-coding-agent 1.18.0

//! Live provider stream and transcript event handling for the interactive TUI.
//!
//! This module owns App methods that switch and drain assistant/reasoning
//! streams, schedule stream previews, record usage and cost from view-model
//! events, drive tool-call lifecycle display state, and merge finished text
//! into the transcript. Expand/collapse of truncated tool output lives in
//! `tool_output_ui`. Stream finalization that must happen before recording a
//! lifecycle event is classified exhaustively via
//! [`should_finish_streams_before_recording`].

use std::time::Instant;

use ratatui::{backend::Backend, DefaultTerminal, Terminal};

use super::{
    activity::ActivityPhase,
    event_adapter::ViewModelEvent,
    markdown::update_code_block_state,
    render::padded_content_width,
    stream::StreamFragment,
    usage_cost::{
        add_optional, merge_usage, usage_difference, usage_with_estimated_cost, CostSource,
    },
    App, Entry, FinalAnswerDelta, LiveStreamPreview, ReasoningEntry, StreamKind, ToolEntry,
};

pub(super) fn final_answer_delta<'a>(emitted_text: &str, answer: &'a str) -> FinalAnswerDelta<'a> {
    match answer.strip_prefix(emitted_text) {
        Some("") => FinalAnswerDelta::None,
        Some(suffix) => FinalAnswerDelta::Append(suffix),
        None => FinalAnswerDelta::Mismatch,
    }
}

fn should_finish_streams_before_recording(event: &ViewModelEvent) -> bool {
    match event {
        ViewModelEvent::StepStarted(_)
        | ViewModelEvent::ToolCallUpdated { .. }
        | ViewModelEvent::ToolCallProposed { .. }
        | ViewModelEvent::ToolStarted { .. }
        | ViewModelEvent::ToolFinished { .. } => true,
        ViewModelEvent::RunStarted
        | ViewModelEvent::SteeringApplied(_)
        | ViewModelEvent::ProviderStreamReset
        | ViewModelEvent::ProviderRetry
        | ViewModelEvent::OutputDelta(_)
        | ViewModelEvent::ReasoningDelta(_)
        | ViewModelEvent::ContextUsage(_)
        | ViewModelEvent::Usage(_)
        | ViewModelEvent::ToolUpdated { .. } => false,
    }
}

impl App {
    pub(super) fn reset_streams(&mut self) {
        self.streams.reset();
        // Discard an unfinished reasoning phase. Callers that should keep a
        // summary must finalize before reset (for example `finish_streams`).
        self.turn.reasoning_phase_mut().reset();
    }

    pub(super) fn handle_agent_event<B: Backend>(
        &mut self,
        event: ViewModelEvent,
        terminal: &mut Terminal<B>,
    ) -> Result<bool, B::Error> {
        if let Some(phase) = event.activity_phase() {
            self.turn.set_activity_phase(phase);
        }
        match event {
            ViewModelEvent::ProviderStreamReset => {
                self.reset_provider_attempt_stream();
                Ok(true)
            }
            ViewModelEvent::OutputDelta(text) => {
                let switched = self.switch_stream_kind(StreamKind::Assistant);
                let drained = self.receive_stream_delta(terminal, StreamKind::Assistant, &text)?;
                Ok(switched || drained)
            }
            ViewModelEvent::ReasoningDelta(text) => {
                let show_reasoning = self.info.runtime.show_reasoning_output;
                self.turn
                    .reasoning_phase_mut()
                    .on_reasoning_delta(show_reasoning);
                if !show_reasoning {
                    return Ok(true);
                }
                let switched = self.switch_stream_kind(StreamKind::Reasoning);
                let drained = self.receive_stream_delta(terminal, StreamKind::Reasoning, &text)?;
                Ok(switched || drained)
            }
            other => {
                if should_finish_streams_before_recording(&other) {
                    self.finish_streams();
                }
                if let Some(entry) = self.record_agent_event(other) {
                    self.insert_entry(&entry);
                }
                self.drain_streams(terminal)?;
                Ok(true)
            }
        }
    }

    pub(super) fn switch_stream_kind(&mut self, kind: StreamKind) -> bool {
        let inserted = if self
            .streams
            .current_stream_kind
            .is_some_and(|current| current != kind)
        {
            self.finish_current_stream()
        } else {
            false
        };
        // Closing into assistant ends the reasoning phase so the thought
        // footer lands after any finished reasoning text.
        let thought = if kind == StreamKind::Assistant
            && self.streams.current_stream_kind != Some(StreamKind::Assistant)
        {
            self.close_reasoning_phase()
        } else {
            false
        };
        self.streams.current_stream_kind = Some(kind);
        self.streams.schedule_tick(kind, Instant::now());
        inserted || thought
    }

    pub(super) fn drain_streams<B: Backend>(
        &mut self,
        terminal: &mut Terminal<B>,
    ) -> Result<bool, B::Error> {
        let reasoning_drained = self.drain_stream(terminal, StreamKind::Reasoning)?;
        let assistant_drained = self.drain_stream(terminal, StreamKind::Assistant)?;
        Ok(reasoning_drained || assistant_drained)
    }

    /// Appends one provider delta through the hold/pacer and commits renderable lines.
    fn receive_stream_delta<B: Backend>(
        &mut self,
        terminal: &mut Terminal<B>,
        kind: StreamKind,
        text: &str,
    ) -> Result<bool, B::Error> {
        self.streams.push_delta(kind, text, Instant::now());
        self.drain_stream(terminal, kind)
    }

    /// Plays out everything the pacer is holding.
    ///
    /// Stands in for the frame ticks a live terminal supplies, so tests can
    /// assert on streamed text without depending on wall-clock timing.
    #[cfg(test)]
    pub(super) fn play_out_streams<B: Backend>(
        &mut self,
        terminal: &mut Terminal<B>,
    ) -> Result<bool, B::Error> {
        self.streams.play_out();
        self.drain_streams(terminal)
    }

    pub(super) fn drain_stream<B: Backend>(
        &mut self,
        terminal: &mut Terminal<B>,
        kind: StreamKind,
    ) -> Result<bool, B::Error> {
        let width = terminal.size()?.width as usize;
        let inner_width = padded_content_width(width);
        let in_code_block = self.streams.code_fence(kind).is_open();
        let fragment = self
            .streams
            .stream_mut(kind)
            .drain_renderable_markdown(inner_width, in_code_block);
        if let Some(fragment) = fragment {
            self.streams.live_stream_preview = None;
            self.insert_stream_fragment(fragment, kind);
            Ok(true)
        } else {
            Ok(false)
        }
    }

    pub(super) fn finish_current_stream(&mut self) -> bool {
        self.streams
            .current_stream_kind
            .is_some_and(|kind| self.finish_stream(kind))
    }

    /// Releases held text and refreshes the partial-line preview on the shared
    /// stream UI cadence.
    pub(super) fn drain_stream_tick(
        &mut self,
        terminal: &mut DefaultTerminal,
    ) -> std::io::Result<bool> {
        let now = Instant::now();
        if self
            .streams
            .stream_tick_deadline
            .is_none_or(|deadline| now < deadline)
        {
            return Ok(false);
        }
        let released = self.streams.on_tick(now);
        let Some(kind) = self.streams.current_stream_kind else {
            return Ok(false);
        };
        let drained = if released {
            self.drain_stream(terminal, kind)?
        } else {
            false
        };
        let preview_changed = self.refresh_stream_preview(terminal, kind)?;
        Ok(drained || preview_changed)
    }

    fn refresh_stream_preview(
        &mut self,
        terminal: &mut DefaultTerminal,
        kind: StreamKind,
    ) -> std::io::Result<bool> {
        let width = terminal.size()?.width as usize;
        let inner_width = padded_content_width(width);
        let in_code_block = self.streams.code_fence(kind).is_open();
        let preview = self
            .streams
            .stream(kind)
            .drain_preview_markdown(inner_width, in_code_block);
        if let Some(preview) = preview {
            self.streams.live_stream_preview = Some(LiveStreamPreview {
                kind,
                text: preview.render_text().to_string(),
                include_leading_blank: preview.include_leading_blank(),
            });
            Ok(true)
        } else if self.streams.live_stream_preview.is_some() {
            self.streams.live_stream_preview = None;
            Ok(true)
        } else {
            Ok(false)
        }
    }

    pub(super) fn record_agent_event(&mut self, event: ViewModelEvent) -> Option<Entry> {
        match event {
            ViewModelEvent::RunStarted => {
                self.usage.usage_cost_tracker.run_started();
                self.usage.usage_before_current_run = self.usage.cumulative_usage.clone();
                self.usage.run_usage.clear();
                None
            }
            ViewModelEvent::StepStarted(step) => {
                self.usage.usage_cost_tracker.step_started();
                self.usage.run_usage.step_started();
                self.reset_streams();
                self.turn.provider_attempt_mut().begin(self.history.len());
                self.turn
                    .reasoning_phase_mut()
                    .begin_step(self.info.runtime.show_reasoning_output);
                self.begin_provider_turn_ui();
                self.turn.clear_tool_calls();
                self.turn.start_loading_if_needed();
                self.status = format!("running step {step}");
                None
            }
            ViewModelEvent::SteeringApplied(ids) => {
                self.mark_steering_applied(&ids);
                None
            }
            ViewModelEvent::ToolStarted { call_id, card } => {
                self.turn.tool_started(call_id, card);
                None
            }
            ViewModelEvent::ToolUpdated { call_id, card } => {
                self.turn.tool_updated(call_id, card);
                None
            }
            ViewModelEvent::ToolCallUpdated {
                index,
                call_id,
                card,
            } => {
                self.turn.tool_call_preview(index, call_id, card);
                None
            }
            ViewModelEvent::ToolCallProposed { call_id, card } => {
                self.turn.tool_call_proposed(call_id, card);
                None
            }
            ViewModelEvent::ProviderStreamReset | ViewModelEvent::ProviderRetry => {
                self.usage.usage_cost_tracker.attempt_restarted();
                self.usage.run_usage.attempt_reset();
                None
            }
            ViewModelEvent::OutputDelta(_) | ViewModelEvent::ReasoningDelta(_) => None,
            ViewModelEvent::ContextUsage(usage) => {
                self.info.services.diagnostics.record_context(usage.clone());
                self.usage.current_context = Some(usage);
                None
            }
            ViewModelEvent::Usage(usage) => {
                let current_cost_source = self.usage.usage_cost_tracker.record_usage(&usage);
                let model_metadata = self.model_metadata.as_ref();
                let mut current_run_usage =
                    self.usage.run_usage.apply_snapshot(usage, |snapshot| {
                        usage_with_estimated_cost(snapshot, model_metadata)
                    });
                let step_baseline = self
                    .usage
                    .run_usage
                    .before_step()
                    .cloned()
                    .map(|usage| usage_with_estimated_cost(usage, model_metadata));
                let mut latest_usage = usage_difference(&current_run_usage, step_baseline.as_ref());
                latest_usage = usage_with_estimated_cost(latest_usage, model_metadata);
                if current_cost_source == CostSource::Estimated {
                    current_run_usage.cost_usd_micros = add_optional(
                        step_baseline
                            .as_ref()
                            .and_then(|usage| usage.cost_usd_micros),
                        latest_usage.cost_usd_micros,
                    );
                    if let Some(current) = self.usage.run_usage.current_mut() {
                        current.cost_usd_micros = current_run_usage.cost_usd_micros;
                    }
                }
                self.usage.latest_usage = Some(latest_usage);
                self.usage
                    .cumulative_usage
                    .clone_from(&self.usage.usage_before_current_run);
                merge_usage(&mut self.usage.cumulative_usage, current_run_usage);
                None
            }
            ViewModelEvent::ToolFinished {
                call_id,
                mut card,
                image_asset,
            } => {
                self.statusline.refresh_git_branch();
                let expanded = self.turn.tool_finished(&call_id);
                self.turn
                    .set_activity_phase(if self.turn.tool_calls().is_running() {
                        ActivityPhase::RunningTool
                    } else {
                        ActivityPhase::Starting
                    });
                let image =
                    image_asset
                        .as_ref()
                        .and_then(|asset| match self.load_feed_image(asset) {
                            Ok(image) => image,
                            Err(error) => {
                                card.push_fact(rho_tools::tool_card::ToolFact::Error {
                                    text: format!("image preview unavailable: {error}"),
                                });
                                None
                            }
                        });
                Some(Entry::Tool(ToolEntry {
                    card,
                    expanded,
                    image,
                }))
            }
        }
    }

    pub(super) fn push_transcript_entry(&mut self, entry: Entry) {
        match entry {
            Entry::Assistant(text) => {
                let index = if matches!(self.history.last(), Some(Entry::Assistant(_))) {
                    self.history.len().saturating_sub(1)
                } else {
                    self.history.len()
                };
                match self.history.last_mut() {
                    Some(Entry::Assistant(previous)) => {
                        previous.push_str(&text);
                        self.history.lines_mut().assistant_appended(index);
                    }
                    _ => {
                        self.history.lines_mut().invalidate_from(index);
                        self.history.push(Entry::Assistant(text));
                    }
                }
                self.mark_markdown_images_dirty_from(index);
            }
            Entry::Reasoning(reasoning) => match self.history.last_mut() {
                Some(Entry::Reasoning(previous)) if previous.thought_for.is_none() => {
                    previous.text.push_str(&reasoning.text);
                    if reasoning.thought_for.is_some() {
                        previous.thought_for = reasoning.thought_for;
                    }
                    let index = self.history.len().saturating_sub(1);
                    self.history.lines_mut().invalidate_from(index);
                }
                _ => {
                    let index = self.history.len();
                    self.history.lines_mut().invalidate_from(index);
                    self.history.push(Entry::Reasoning(reasoning));
                }
            },
            other => {
                self.history.set_last_status_notice(match &other {
                    Entry::Notice(text) => Some(text.clone()),
                    _ => None,
                });
                let index = self.history.len();
                self.history.lines_mut().invalidate_from(index);
                self.history.push(other);
            }
        }
    }
    pub(super) fn finish_streams(&mut self) -> bool {
        let reasoning_finished = self.finish_stream(StreamKind::Reasoning);
        let assistant_finished = self.finish_stream(StreamKind::Assistant);
        self.streams.current_stream_kind = None;
        self.streams.clear_tick_deadline();
        self.streams.live_stream_preview = None;
        let thought = self.close_reasoning_phase();
        reasoning_finished || assistant_finished || thought
    }

    /// Ends the current reasoning stretch, attaching or inserting a thought duration.
    pub(super) fn close_reasoning_phase(&mut self) -> bool {
        let Some(elapsed) = self.turn.reasoning_phase_mut().finalize() else {
            return false;
        };
        match self.history.last_mut() {
            Some(Entry::Reasoning(reasoning)) if reasoning.thought_for.is_none() => {
                reasoning.thought_for = Some(elapsed);
                let index = self.history.len().saturating_sub(1);
                self.history.lines_mut().invalidate_from(index);
                true
            }
            _ => {
                self.insert_entry(&Entry::Reasoning(ReasoningEntry::summary_only(elapsed)));
                true
            }
        }
    }

    pub(super) fn finish_stream(&mut self, kind: StreamKind) -> bool {
        if self.streams.current_stream_kind == Some(kind) {
            self.streams.flush_hold(kind);
        }
        let fragment = self.streams.stream_mut(kind).finish();
        self.streams.clear_tick_deadline();
        if let Some(fragment) = fragment {
            self.streams.live_stream_preview = None;
            self.insert_stream_fragment(fragment, kind);
            true
        } else {
            false
        }
    }

    pub(super) fn insert_final_answer_suffix(&mut self, answer: &str) {
        match final_answer_delta(self.streams.assistant_stream.emitted_text(), answer) {
            FinalAnswerDelta::None => {}
            FinalAnswerDelta::Append(suffix) => {
                self.streams.assistant_stream.push_delta(suffix);
                if let Some(fragment) = self.streams.assistant_stream.finish() {
                    self.insert_stream_fragment(fragment, StreamKind::Assistant);
                }
            }
            FinalAnswerDelta::Mismatch => {
                self.replace_current_turn_assistant_transcript(answer);
            }
        }
    }

    pub(super) fn insert_stream_fragment(&mut self, fragment: StreamFragment, kind: StreamKind) {
        let render_text = fragment.render_text();
        if !render_text.is_empty() {
            update_code_block_state(render_text, self.streams.code_fence_mut(kind));
        }
        let text = fragment.into_text();
        self.push_transcript_entry(kind.entry(text));
    }

    pub(super) fn replace_current_turn_assistant_transcript(&mut self, answer: &str) {
        let start = self.turn.current_turn_start().unwrap_or(0);
        let assistant_indices = self
            .history
            .entries()
            .iter()
            .enumerate()
            .skip(start)
            .filter_map(|(index, entry)| matches!(entry, Entry::Assistant(_)).then_some(index))
            .collect::<Vec<_>>();

        let Some((first, stale)) = assistant_indices.split_first() else {
            self.push_transcript_entry(Entry::Assistant(answer.to_string()));
            return;
        };

        if let Entry::Assistant(text) = &mut self.history.entries_mut()[*first] {
            *text = answer.to_string();
        }
        self.history.images_mut().clear();
        self.history.invalidate_from(*first);
        for index in stale.iter().rev() {
            self.history.entries_mut().remove(*index);
        }
    }

    pub(super) fn insert_entry(&mut self, entry: &Entry) {
        self.record_inserted_entry(entry.clone());
    }

    pub(super) fn notify_status(&mut self, status: impl Into<String>) {
        let status = status.into();
        self.status = status.clone();
        if self.history.last_status_notice() == Some(status.as_str()) {
            return;
        }
        self.insert_entry(&Entry::Notice(status));
    }

    pub(super) fn record_inserted_entry(&mut self, entry: Entry) {
        self.history.set_last_status_notice(match &entry {
            Entry::Notice(text) => Some(text.clone()),
            Entry::User(_)
            | Entry::Assistant(_)
            | Entry::Reasoning(_)
            | Entry::RuntimeInfo(_)
            | Entry::UsageLimits(_)
            | Entry::Tool(_)
            | Entry::Error(_) => None,
        });
        self.push_transcript_entry(entry);
    }

    /// Apply the live `show_reasoning_output` setting to in-flight turn UI.
    pub(super) fn apply_reasoning_output_visibility(&mut self) {
        if self.info.runtime.show_reasoning_output {
            self.turn
                .reasoning_phase_mut()
                .set_hidden_placeholder(false);
            return;
        }

        self.discard_live_reasoning_output();

        // Keep the Thinking... placeholder while this step is still waiting for
        // or streaming reasoning. Later phases (response, tools) stay clear.
        let hide_placeholder = self.is_ui_busy()
            && (self.turn.reasoning_phase().has_started()
                || matches!(
                    self.turn.activity_phase(),
                    ActivityPhase::Starting
                        | ActivityPhase::WaitingForProvider
                        | ActivityPhase::Thinking
                        | ActivityPhase::RetryingProvider
                ));
        self.turn
            .reasoning_phase_mut()
            .set_hidden_placeholder(hide_placeholder);
    }

    pub(super) fn discard_live_reasoning_output(&mut self) {
        let clearing_reasoning = matches!(
            self.streams.current_stream_kind,
            Some(StreamKind::Reasoning)
        ) || self
            .streams
            .live_stream_preview
            .as_ref()
            .is_some_and(|preview| preview.kind == StreamKind::Reasoning);
        if !clearing_reasoning {
            return;
        }
        if matches!(
            self.streams.current_stream_kind,
            Some(StreamKind::Reasoning)
        ) {
            self.streams.discard_hold();
            self.streams.reasoning_stream.reset();
            self.streams.reasoning_stream_code_fence = Default::default();
            self.streams.current_stream_kind = None;
        }
        self.streams.clear_tick_deadline();
        self.streams.live_stream_preview = None;
    }

    pub(super) fn reset_provider_attempt_stream(&mut self) {
        self.reset_streams();
        self.turn.clear_tool_calls();
        if let Some(start) = self
            .turn
            .provider_attempt_mut()
            .reset_output(self.history.entries_mut())
        {
            self.history.images_mut().clear();
            self.history.invalidate_from(start);
        }
        self.status = "retrying provider response".into();
    }
}

#[cfg(test)]
#[path = "transcript_events_tests.rs"]
mod tests;