rho-coding-agent 2.30.0

use std::{
    collections::{BTreeMap, HashMap},
    path::{Path, PathBuf},
};

use agent_client_protocol::schema::v1::{
    ContentBlock, ContentChunk, Diff, MessageId, Notice, NoticeSeverity, SessionId,
    SessionNotification, SessionUpdate, StopReason, Terminal, ToolCall, ToolCallContent,
    ToolCallId, ToolCallLocation, ToolCallStatus, ToolCallUpdate, ToolCallUpdateFields, ToolKind,
};
use rho_sdk::{
    tool::{ToolMetadata, ToolOutput},
    RunEvent, ToolCompletion,
};
use rho_tools::tool_card::ToolHeader;
use serde_json::Value;

use super::{
    compaction_card::{self, CompactionCard},
    HostFeatures,
};
use crate::{
    app::interactive_presenter::call_header,
    display_transcript::{DisplayRow, DisplayTranscript},
    presentation::NotificationTone,
    tools::mcp::exported_name::McpToolNames,
};

const PROVIDER_STREAM_RESET_NOTICE: &str = "[provider response discarded; retrying]";
/// Title for a tool call whose name the mapper never saw proposed.
const UNNAMED_TOOL_TITLE: &str = "tool";

#[path = "events_replay.rs"]
mod replay;
#[path = "events_tools.rs"]
mod tools;

/// Maps one prompt's SDK events onto ACP session updates.
///
/// Proposed tool arguments are held until [`RunEvent::ToolStarted`] so the first
/// ACP `ToolCall` can carry name, kind, locations, and raw input together.
///
/// Hosts only accept a `tool_call_update` for an id they saw in a `tool_call`
/// (Zed shows anything else as a failed "Tool call not found" card), so every
/// call opens with a full `tool_call`, even one the SDK finishes without
/// starting, such as a call to a tool that does not exist. When the run ends,
/// calls still open are closed as failed, since the SDK does not finish calls
/// that a cancel interrupts and ACP has no cancelled tool status.
///
/// Hosts that advertise compaction see each compaction the run performs as
/// a card; one still open when the run ends is closed the same way.
///
/// Message and thought chunks carry a `messageId`. Consecutive chunks of one
/// kind share it; a chunk of the other kind, a new model step, or any update
/// this mapper sends in between, such as a tool call, starts a new message.
pub(super) struct EventMapper {
    features: HostFeatures,
    /// Session working directory; titles show paths relative to it.
    cwd: PathBuf,
    /// Identifies MCP calls, whose capped names cannot be decoded.
    mcp_tools: McpToolNames,
    proposed: HashMap<String, rho_sdk::model::ToolCall>,
    announced: BTreeMap<String, CallState>,
    /// Host terminals calls run in, by call id. Every later update of such a
    /// call shows its terminal, since ACP content replaces the card's.
    terminals: HashMap<String, String>,
    compaction: Option<CompactionCard>,
    /// The message the latest chunk belongs to, while later chunks of its
    /// kind still continue it.
    message: Option<OpenMessage>,
}

/// Which ACP message stream a chunk belongs to.
#[derive(Clone, Copy, PartialEq)]
enum ChunkKind {
    Message,
    Thought,
}

struct OpenMessage {
    kind: ChunkKind,
    id: MessageId,
}

/// Whether the host may still be waiting for a call's final status.
#[derive(Clone, Copy, PartialEq)]
enum CallState {
    Open,
    Finished,
}

/// How a run ended, for closing what it left open.
#[derive(Clone, Copy, Debug)]
pub(super) enum RunEnding<'a> {
    Cancelled,
    Failed(&'a str),
    /// It completed, or its event stream ended without a terminal event.
    WithoutResult,
}

impl EventMapper {
    pub(super) fn new(features: HostFeatures, cwd: PathBuf, mcp_tools: McpToolNames) -> Self {
        Self {
            features,
            cwd,
            mcp_tools,
            proposed: HashMap::new(),
            announced: BTreeMap::new(),
            terminals: HashMap::new(),
            compaction: None,
            message: None,
        }
    }

    /// Maps one SDK event onto ACP notifications. Events that only move mapper
    /// state, or that ACP has no update for, map to nothing; a run's terminal
    /// event maps to the closing updates for calls still open.
    pub(super) fn map_event(
        &mut self,
        session_id: &SessionId,
        event: &rho_sdk::RunEvent,
    ) -> Vec<SessionNotification> {
        let update = match event {
            RunEvent::AssistantTextDelta { text } => Some(self.chunk(ChunkKind::Message, text)),
            RunEvent::ReasoningDelta { text } | RunEvent::ReasoningSummaryDelta { text } => {
                Some(self.chunk(ChunkKind::Thought, text))
            }
            RunEvent::StepStarted { .. } => {
                self.message = None;
                None
            }
            RunEvent::ToolProposed { call } => {
                if !call.id.is_empty() {
                    self.proposed.insert(call.id.clone(), call.clone());
                }
                None
            }
            RunEvent::ToolStarted {
                call_id,
                name,
                metadata,
            } => {
                let proposed = self
                    .proposed
                    .remove(call_id.as_str())
                    .map(|call| call.arguments);
                self.announced
                    .insert(call_id.as_str().to_owned(), CallState::Open);
                Some(SessionUpdate::ToolCall(started_tool_call(
                    call_id,
                    name,
                    metadata,
                    proposed,
                    &self.cwd,
                    &self.mcp_tools,
                )))
            }
            RunEvent::ToolUpdated { call_id, progress } => {
                if let Some(terminal) = progress.presentation().host_terminal_id() {
                    self.terminals
                        .insert(call_id.as_str().to_owned(), terminal.to_owned());
                }
                let content = match self.terminals.get(call_id.as_str()) {
                    Some(terminal) => vec![terminal_content(terminal)],
                    None => vec![ToolCallContent::from(progress.text())],
                };
                self.tool_update(
                    call_id,
                    CallState::Open,
                    ToolCallStatus::InProgress,
                    content,
                    /*locations*/ None,
                )
            }
            RunEvent::ToolFinished { call_id, result } => {
                // Progress names the terminal as soon as it exists; the
                // output's metadata covers a call whose progress was dropped.
                let terminal = self.terminals.remove(call_id.as_str()).or_else(|| {
                    result
                        .output()
                        .and_then(|output| output.presentation().host_terminal_id())
                        .map(str::to_owned)
                });
                let (status, content) = match terminal {
                    Some(terminal) => terminal_finished_content(result, &terminal),
                    None => finished_content(result),
                };
                let locations = result
                    .output()
                    .and_then(|output| tools::finished_locations(output, &self.cwd));
                self.tool_update(call_id, CallState::Finished, status, content, locations)
            }
            RunEvent::ProviderStreamReset { .. } => {
                self.proposed.clear();
                // The notice is a message of its own, and the retried
                // response starts another.
                self.message = None;
                Some(SessionUpdate::AgentThoughtChunk(text_chunk(
                    PROVIDER_STREAM_RESET_NOTICE,
                    fresh_message_id(),
                )))
            }
            RunEvent::CompactionStarted { .. } if self.features.compaction => {
                let card = CompactionCard::new();
                let update = card.in_progress();
                self.compaction = Some(card);
                Some(update)
            }
            RunEvent::CompactionCompleted { .. } if self.features.compaction => {
                // A first update with a terminal status still creates the card.
                let card = self.compaction.take().unwrap_or_else(CompactionCard::new);
                Some(card.completed())
            }
            RunEvent::Cancelled { .. } => {
                return self.close_unfinished(session_id, RunEnding::Cancelled);
            }
            RunEvent::Failed { message, .. } => {
                return self.close_unfinished(session_id, RunEnding::Failed(message));
            }
            RunEvent::Completed { .. } => {
                return self.close_unfinished(session_id, RunEnding::WithoutResult);
            }
            RunEvent::Started { .. }
            | RunEvent::ToolCallUpdated { .. }
            | RunEvent::UsageUpdated { .. }
            | RunEvent::ProviderContextUpdated { .. }
            | RunEvent::HostInputRequested { .. }
            | RunEvent::CompactionStarted { .. }
            | RunEvent::CompactionCompleted { .. }
            | RunEvent::SteeringApplied { .. }
            | RunEvent::SteeringDelivered { .. }
            | RunEvent::ProviderDiagnostic { .. }
            | RunEvent::ToolHostInputRequested { .. }
            | RunEvent::WebSearch { .. }
            | RunEvent::ProviderRequestRetry
            | RunEvent::ModelCallCompleted { .. }
            | RunEvent::HostedToolActivity { .. }
            | RunEvent::ProviderServiceTierFallback { .. } => None,
            // RunEvent is non_exhaustive; unknown future variants have no ACP update.
            _ => None,
        };
        if update.as_ref().is_some_and(|update| {
            !matches!(
                update,
                SessionUpdate::AgentMessageChunk(_) | SessionUpdate::AgentThoughtChunk(_)
            )
        }) {
            self.message = None;
        }
        update
            .map(|update| notify(session_id, update))
            .into_iter()
            .collect()
    }

    /// A chunk of the open message of `kind`, or of a new message when the
    /// previous update was anything else.
    fn chunk(&mut self, kind: ChunkKind, text: &str) -> SessionUpdate {
        let id = match &self.message {
            Some(open) if open.kind == kind => open.id.clone(),
            _ => {
                let id = fresh_message_id();
                self.message = Some(OpenMessage {
                    kind,
                    id: id.clone(),
                });
                id
            }
        };
        let chunk = text_chunk(text, id);
        match kind {
            ChunkKind::Message => SessionUpdate::AgentMessageChunk(chunk),
            ChunkKind::Thought => SessionUpdate::AgentThoughtChunk(chunk),
        }
    }

    /// Final updates for the compaction and calls the host saw start but
    /// never saw finish: calls close `failed`, the compaction with `ending`.
    /// The run's terminal event closes them; the pump calls this again when
    /// the event stream ends, for a run that ended without one.
    pub(super) fn close_unfinished(
        &mut self,
        session_id: &SessionId,
        ending: RunEnding<'_>,
    ) -> Vec<SessionNotification> {
        self.proposed.clear();
        self.message = None;
        let compaction = self.compaction.take().map(|card| {
            let update = match ending {
                RunEnding::Cancelled => card.cancelled(),
                RunEnding::Failed(message) => card.failed(message),
                RunEnding::WithoutResult => card.failed(compaction_card::UNFINISHED),
            };
            notify(session_id, update)
        });
        let calls = self
            .announced
            .iter_mut()
            .filter(|(_, state)| **state == CallState::Open)
            .map(|(call_id, state)| {
                *state = CallState::Finished;
                notify(
                    session_id,
                    SessionUpdate::ToolCallUpdate(ToolCallUpdate::new(
                        ToolCallId::new(call_id.as_str()),
                        ToolCallUpdateFields::new().status(ToolCallStatus::Failed),
                    )),
                )
            });
        compaction.into_iter().chain(calls).collect()
    }

    /// A progress or result update for a call, sent as a full `tool_call`
    /// when the host has not seen the call yet. Progress that arrives after
    /// a call finished is dropped, so it cannot reopen the host's card.
    /// `locations` replace the call's earlier ones when given.
    fn tool_update(
        &mut self,
        call_id: &rho_sdk::ToolCallId,
        next: CallState,
        status: ToolCallStatus,
        content: Vec<ToolCallContent>,
        locations: Option<Vec<ToolCallLocation>>,
    ) -> Option<SessionUpdate> {
        let id = call_id.as_str();
        if let Some(state) = self.announced.get_mut(id) {
            if *state == CallState::Finished && next == CallState::Open {
                return None;
            }
            *state = next;
            return Some(SessionUpdate::ToolCallUpdate(ToolCallUpdate::new(
                ToolCallId::new(id),
                ToolCallUpdateFields::new()
                    .status(status)
                    .content(content)
                    .locations(locations),
            )));
        }
        self.announced.insert(id.to_owned(), next);
        let proposed = self.proposed.remove(id);
        let Some(call) = proposed else {
            let tool = ToolCall::new(ToolCallId::new(id), UNNAMED_TOOL_TITLE)
                .kind(ToolKind::Other)
                .status(status)
                .content(content)
                .locations(locations.unwrap_or_default());
            return Some(SessionUpdate::ToolCall(tool));
        };
        let title = tool_title(
            &call.name,
            Some(&call.arguments),
            &ToolMetadata::default(),
            &self.cwd,
            &self.mcp_tools,
        );
        let tool = ToolCall::new(ToolCallId::new(id), title)
            .kind(ToolKind::Other)
            .status(status)
            .content(content)
            .locations(locations.unwrap_or_default())
            .name(call.name)
            .raw_input(call.arguments);
        Some(SessionUpdate::ToolCall(tool))
    }

    pub(super) fn map_stop(outcome: &rho_sdk::RunOutcome) -> StopReason {
        map_sdk_stop_reason(outcome.stop_reason())
    }

    pub(super) fn replay_history(
        session_id: &SessionId,
        messages: &[rho_sdk::model::Message],
        cwd: &Path,
        mcp_tools: &McpToolNames,
    ) -> Vec<SessionNotification> {
        replay::history(session_id, messages, cwd, mcp_tools)
    }

    /// A live agent message outside any run, such as a notice at load.
    pub(super) fn agent_notice(session_id: &SessionId, text: &str) -> SessionNotification {
        notify(session_id, Self::agent_message(text))
    }

    /// A whole agent message sent as one chunk, outside any run.
    pub(super) fn agent_message(text: impl Into<String>) -> SessionUpdate {
        SessionUpdate::AgentMessageChunk(text_chunk(text, fresh_message_id()))
    }

    /// Information for the user that is not part of the conversation: an ACP
    /// `notice` when the host shows them, else an agent message.
    pub(super) fn info_notice(
        session_id: &SessionId,
        features: HostFeatures,
        text: &str,
    ) -> SessionNotification {
        if features.notices {
            notify(
                session_id,
                SessionUpdate::Notice(Notice::new(NoticeSeverity::Info, text)),
            )
        } else {
            Self::agent_notice(session_id, text)
        }
    }

    /// Background results the model was given, one finished tool call card
    /// per display row. Cards keep them out of the agent's message text, and
    /// hosts already render them beside the tool calls that started the work.
    pub(super) fn display_transcript(
        session_id: &SessionId,
        transcript: &DisplayTranscript,
    ) -> Vec<SessionNotification> {
        transcript
            .0
            .iter()
            .map(|row| notify(session_id, SessionUpdate::ToolCall(display_row_call(row))))
            .collect()
    }
}

fn map_sdk_stop_reason(reason: rho_sdk::StopReason) -> StopReason {
    match reason {
        rho_sdk::StopReason::EndTurn => StopReason::EndTurn,
        rho_sdk::StopReason::MaxSteps => StopReason::MaxTurnRequests,
        rho_sdk::StopReason::MaxTokens => StopReason::MaxTokens,
        rho_sdk::StopReason::Refusal => StopReason::Refusal,
        // StopReason is non_exhaustive; a new SDK reason still ends the ACP turn.
        _ => StopReason::EndTurn,
    }
}

/// Whether a turn reached the model's final answer, including one the
/// provider truncated or refused. Cancelled and step-limited turns did not.
pub(super) fn turn_answered(stop: StopReason) -> bool {
    matches!(
        stop,
        StopReason::EndTurn | StopReason::MaxTokens | StopReason::Refusal
    )
}

fn notify(session_id: &SessionId, update: SessionUpdate) -> SessionNotification {
    SessionNotification::new(session_id.clone(), update)
}

/// A text chunk of the message `id`; every chunk Rho sends names its message.
fn text_chunk(text: impl Into<String>, id: MessageId) -> ContentChunk {
    ContentChunk::new(ContentBlock::from(text.into())).message_id(id)
}

/// A live message's ID. Replayed messages derive theirs from saved history.
fn fresh_message_id() -> MessageId {
    MessageId::new(format!("rho-msg-{}", uuid::Uuid::new_v4()))
}

fn started_tool_call(
    call_id: &rho_sdk::ToolCallId,
    name: &str,
    metadata: &ToolMetadata,
    proposed: Option<Value>,
    cwd: &Path,
    mcp_tools: &McpToolNames,
) -> ToolCall {
    let title = tool_title(name, proposed.as_ref(), metadata, cwd, mcp_tools);
    let tool = ToolCall::new(ToolCallId::new(call_id.as_str()), title)
        .name(name.to_owned())
        .kind(tools::kind(name, proposed.as_ref(), metadata))
        .status(ToolCallStatus::InProgress)
        .locations(tools::started_locations(
            name,
            proposed.as_ref(),
            metadata,
            cwd,
        ));
    match proposed {
        Some(arguments) => tool.raw_input(arguments),
        None => tool,
    }
}

/// What a call does, as the interactive card header says it. ACP sends a
/// title once, so it names the call, never a live phase:
/// - a shell call is its bare command, since hosts such as Zed show an
///   execute call's title verbatim as the command it runs; the call's
///   `command_line` metadata is that command, and replay reads it from the
///   arguments instead;
/// - a status-first header keeps its identity and drops the phase;
/// - an MCP call leads with its server, which the card shows on a fact line.
///
/// Replayed history has no metadata, so a generic tool that describes
/// itself through `command_summary` is titled by name alone there.
fn tool_title(
    name: &str,
    arguments: Option<&Value>,
    metadata: &ToolMetadata,
    cwd: &Path,
    mcp_tools: &McpToolNames,
) -> String {
    if let Some(command) = metadata.command_line_text().filter(|c| !c.is_empty()) {
        return command.to_owned();
    }
    let title = match call_header(name, arguments, metadata, cwd, mcp_tools) {
        ToolHeader::Shell {
            command: Some(command),
            ..
        } => command,
        // A shell call with no command yet; `$` alone would read as one.
        ToolHeader::Shell { command: None, .. } => name.to_owned(),
        ToolHeader::StatusFirst { identity, .. } => identity,
        ToolHeader::Call {
            verb,
            primary: Some(primary),
        } if !primary.is_empty() => format!("{verb}({primary})"),
        ToolHeader::Call { verb, .. } => verb,
    };
    match mcp_tools.identify(name) {
        Some(identity) => format!("{}/{title}", identity.server),
        None => title,
    }
}

fn finished_content(result: &ToolCompletion) -> (ToolCallStatus, Vec<ToolCallContent>) {
    match result {
        ToolCompletion::CompletedFailure(output) => {
            (ToolCallStatus::Failed, success_content(output))
        }
        ToolCompletion::Success(output) => (ToolCallStatus::Completed, success_content(output)),
        ToolCompletion::Failure(failure) => (
            ToolCallStatus::Failed,
            vec![ToolCallContent::from(failure.message())],
        ),
        // The model called a tool that does not exist; say so on the card.
        ToolCompletion::Unavailable => (
            ToolCallStatus::Failed,
            vec![ToolCallContent::from("tool is unavailable")],
        ),
        // ToolCompletion is non_exhaustive; unknown results cannot be shown as success.
        _ => (ToolCallStatus::Failed, Vec::new()),
    }
}

/// A finished call that ran in a host terminal: the terminal shows its
/// output, so the card keeps the terminal and adds only the headline of an
/// error, such as a timeout; the rest of the error repeats the output.
fn terminal_finished_content(
    result: &ToolCompletion,
    terminal: &str,
) -> (ToolCallStatus, Vec<ToolCallContent>) {
    let status = if result.is_failure() {
        ToolCallStatus::Failed
    } else {
        ToolCallStatus::Completed
    };
    let mut content = vec![terminal_content(terminal)];
    if let ToolCompletion::Failure(failure) = result {
        let headline = failure.message().lines().next().unwrap_or_default();
        content.push(ToolCallContent::from(headline));
    }
    (status, content)
}

fn terminal_content(terminal: &str) -> ToolCallContent {
    ToolCallContent::Terminal(Terminal::new(terminal.to_owned()))
}

fn success_content(output: &ToolOutput) -> Vec<ToolCallContent> {
    let mut content = Vec::new();
    if !output.content().is_empty() {
        content.push(ToolCallContent::from(output.content()));
    }
    let metadata = output.presentation();
    if metadata.file_changes().is_empty() {
        // A patch is not the complete file text ACP Diff.new_text requires.
        if let Some(diff) = metadata.unified_diff() {
            content.push(ToolCallContent::from(diff));
        }
    } else {
        content.extend(metadata.file_changes().iter().map(|change| {
            ToolCallContent::Diff(
                Diff::new(change.path(), change.new_text())
                    .old_text(change.old_text().map(str::to_owned)),
            )
        }));
    }
    content
}

fn display_row_call(row: &DisplayRow) -> ToolCall {
    let status = match row {
        DisplayRow::Notification(card) => match card.tone {
            NotificationTone::Error => ToolCallStatus::Failed,
            NotificationTone::Neutral
            | NotificationTone::Accent
            | NotificationTone::Success
            | NotificationTone::Warning => ToolCallStatus::Completed,
        },
        DisplayRow::Notice(_) => ToolCallStatus::Completed,
    };
    let text = row.card_text();
    // Rows carry no stable identity of their own; a fresh id per card keeps
    // hosts from merging two deliveries about the same run.
    let id = format!("rho-notice-{}", uuid::Uuid::new_v4());
    let mut call = ToolCall::new(ToolCallId::new(id), text.title)
        .kind(ToolKind::Other)
        .status(status);
    // Content is markdown for hosts that render it. Hosts that show a tool's
    // output as is, such as T3 Code, read `rawOutput` first, so it carries
    // the plain form without code fences.
    if !text.plain.trim().is_empty() {
        call = call
            .content(vec![ToolCallContent::from(text.markdown)])
            .raw_output(Value::String(text.plain));
    }
    call
}

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