IdeA/crates/app-tauri/src/chat.rs

//! Generic **structured reply ↔ outbound stream sink** bridge infrastructure.
//!
//! Twin of [`crate::pty::PtyBridge`] (ARCHITECTURE §17.7). Where `PtyBridge`
//! routes raw PTY byte chunks to a per-session stream sink, this bridge routes
//! typed [`ReplyChunk`]s — the serialised [`domain::ports::ReplyEvent`]s of an
//! [`domain::ports::AgentSession`] turn — to the chat cell that owns the
//! session.
//!
//! Design (mirrors the PTY path so the lifecycle guarantees are identical):
//! - The frontend opens (or re-attaches) a chat cell and passes a
//!   [`tauri::ipc::Channel`] for that session. The adapter wraps it as a sink and
//!   registers it here keyed by `SessionId`, bumping a monotonic **generation**
//!   so a superseded pump can't tear down the channel of the attach that replaced
//!   it (see [`unregister_if`]).
//! - The `agent_send` pump drains the session's [`domain::ports::ReplyStream`],
//!   maps each event to a [`ReplyChunk`], and forwards it via [`send_output`].
//! - Unlike a PTY, an [`domain::ports::AgentSession`] keeps **no** scrollback (the
//!   port is a per-turn stream, not a persistent byte ring). So the bridge itself
//!   retains the rendered chunks per session — the **conversation scrollback** —
//!   so a re-attach can repaint the turns already streamed, exactly as the PTY
//!   adapter's ring buffer lets `reattach_terminal` repaint xterm. This lives on
//!   the transport side (D4 owns transport), keeping the domain port pure.
//!
//! [`unregister_if`]: ChatBridge::unregister_if
//! [`send_output`]: ChatBridge::send_output

use std::sync::Arc;

use backend::stream::ReplayOutputBridge;
use tauri::ipc::Channel;

use domain::ids::SessionId;
use domain::ports::{
    ReplyEvent, ReplyProgress, ReplyProgressKind, ReplyProgressSource, ReplyProgressStage,
};

use crate::dto::ReplyChunk;
use crate::stream::TauriChannelSink;

/// Maximum number of recent chat chunks retained for transport reattach replay.
pub const MAX_CHAT_SCROLLBACK_CHUNKS: usize = 2_000;
/// Maximum estimated bytes retained for transport reattach replay.
pub const MAX_CHAT_SCROLLBACK_BYTES: usize = 512 * 1024;

/// Registry mapping live structured (chat) sessions to their reply sink
/// plus a retained conversation scrollback.
///
/// Thread-safe; a cloned `Arc<ChatBridge>` is held in [`crate::state::AppState`],
/// the twin of [`crate::pty::PtyBridge`].
pub struct ChatBridge {
    inner: ReplayOutputBridge<SessionId, ReplyChunk>,
}

impl Default for ChatBridge {
    fn default() -> Self {
        Self::new()
    }
}

impl ChatBridge {
    /// Creates an empty bridge.
    #[must_use]
    pub fn new() -> Self {
        Self {
            inner: ReplayOutputBridge::new(
                MAX_CHAT_SCROLLBACK_CHUNKS,
                MAX_CHAT_SCROLLBACK_BYTES,
                reply_chunk_bytes,
            ),
        }
    }

    /// Registers (or replaces) the reply channel for a session and returns the
    /// **generation** of this registration. Each call for a session bumps the
    /// generation, so the caller's pump can later tear down *only its own*
    /// registration via [`unregister_if`](Self::unregister_if). The retained
    /// conversation scrollback is **preserved** across re-attaches (only the
    /// channel and generation change), mirroring how the PTY ring buffer survives
    /// a `reattach_terminal`.
    pub fn register(&self, session: SessionId, channel: Channel<ReplyChunk>) -> u64 {
        self.inner
            .register(session, Arc::new(TauriChannelSink::new(channel)))
    }

    /// Returns the conversation scrollback retained for a session (the chunks
    /// already streamed), or an empty vector if the session is unknown.
    ///
    /// Called by `reattach_agent_chat` to repaint the prior turns into the
    /// re-mounting chat view before the new stream is wired — the typed twin of
    /// `PtyPort::scrollback`.
    #[must_use]
    pub fn scrollback(&self, session: &SessionId) -> Vec<ReplyChunk> {
        self.inner.scrollback(session)
    }

    /// Clears only the retained conversation scrollback for a live session,
    /// keeping any currently attached channel in place.
    ///
    /// Returns the number of chunks removed. This powers `/clean`: the frontend
    /// clears its mounted view from the command response, and future re-attaches
    /// no longer replay the old conversation.
    pub fn clear_scrollback(&self, session: &SessionId) -> usize {
        self.inner.clear_scrollback(session)
    }

    /// Removes a session's transport state **and** its retained scrollback
    /// unconditionally (chat cell explicitly closed). Twin of
    /// [`PtyBridge::unregister`](crate::pty::PtyBridge::unregister).
    pub fn unregister(&self, session: &SessionId) {
        self.inner.unregister(session);
    }

    /// Detaches a session's channel **only if** `gen` is still the current
    /// generation, leaving the scrollback intact. A pump calls this when its turn
    /// stream ends: if the session was re-attached meanwhile (newer generation),
    /// this is a no-op so the dying pump never detaches the live channel that
    /// superseded it. Twin of
    /// [`PtyBridge::unregister_if`](crate::pty::PtyBridge::unregister_if), but it
    /// keeps the conversation scrollback (the conversation outlives a single
    /// turn's pump — closing the cell is what discards it, via [`unregister`]).
    ///
    /// [`unregister`]: ChatBridge::unregister
    pub fn detach_if(&self, session: &SessionId, gen: u64) {
        self.inner.detach_if(session, gen);
    }

    /// Records a chunk in the session's scrollback and, if a view is attached at
    /// the current generation, forwards it to that channel.
    ///
    /// Returns `true` if the chunk was delivered to a live channel, `false` if no
    /// channel is currently attached (e.g. the view navigated away — the chunk is
    /// still retained in scrollback for the next re-attach) or the send failed.
    /// The pump keeps draining either way so the turn still completes and the
    /// scrollback stays whole.
    pub fn send_output(&self, session: &SessionId, chunk: ReplyChunk) -> bool {
        self.inner.send_output(session, chunk)
    }

    /// Number of currently-tracked sessions (handy for tests/diagnostics).
    #[must_use]
    pub fn active_sessions(&self) -> usize {
        self.inner.active_sessions()
    }
}

fn reply_chunk_bytes(chunk: &ReplyChunk) -> usize {
    match chunk {
        ReplyChunk::UserPrompt { text } => text.len(),
        ReplyChunk::TextDelta { text } => text.len(),
        ReplyChunk::ToolActivity { label } => label.len(),
        ReplyChunk::Progress { progress } => {
            progress.label.len()
                + progress.text.as_ref().map_or(0, String::len)
                + progress.provider.as_ref().map_or(0, String::len)
                + progress.native_event.as_ref().map_or(0, String::len)
                + progress.tool_name.as_ref().map_or(0, String::len)
        }
        ReplyChunk::Final { content } => content.len(),
        ReplyChunk::Error { message } => message.len(),
    }
}

/// Maps a domain [`ReplyEvent`] to its wire [`ReplyChunk`]. Pure translation, no
/// I/O — the single point where the typed turn event becomes a serialisable chunk
/// (kept here so the pump and any test share one mapping).
///
/// Returns `None` for [`ReplyEvent::Heartbeat`]: a heartbeat is a non-terminal
/// liveness proof (readiness/heartbeat lot 1) with **no chat content**, so it maps
/// to no wire chunk — the pump simply skips it. Likewise [`ReplyEvent::RateLimited`]
/// is non-terminal and content-free (ports §21.2-T4): the UI badge comes from the
/// `DomainEvent::AgentRateLimited` bus, not the chat stream, so it maps to `None`
/// too. Every content-bearing event still maps to exactly one chunk.
#[must_use]
pub fn chunk_from_event(event: ReplyEvent) -> Option<ReplyChunk> {
    match event {
        ReplyEvent::Progress { progress } => Some(ReplyChunk::Progress {
            progress: progress.into(),
        }),
        ReplyEvent::TextDelta { text } => Some(ReplyChunk::TextDelta { text }),
        ReplyEvent::ToolActivity { label } => Some(ReplyChunk::Progress {
            progress: ReplyProgress::new(
                ReplyProgressSource::ProviderNative,
                ReplyProgressKind::Tool,
                ReplyProgressStage::Info,
                label.clone(),
            )
            .with_tool_name(label)
            .into(),
        }),
        ReplyEvent::Error { message } => Some(ReplyChunk::Error { message }),
        ReplyEvent::Final { content } => {
            if content.trim().is_empty() {
                Some(ReplyChunk::Error {
                    message: "Réponse vide du modèle.".to_owned(),
                })
            } else {
                Some(ReplyChunk::Final { content })
            }
        }
        ReplyEvent::Announcement { text } => Some(ReplyChunk::Progress {
            progress: ReplyProgress::new(
                ReplyProgressSource::ProviderNative,
                ReplyProgressKind::Message,
                ReplyProgressStage::Delta,
                "message intermédiaire",
            )
            .with_text(text)
            .into(),
        }),
        ReplyEvent::Heartbeat => None,
        ReplyEvent::RateLimited { .. } => None,
    }
}