//! 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; 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` is held in [`crate::state::AppState`], /// the twin of [`crate::pty::PtyBridge`]. pub struct ChatBridge { inner: ReplayOutputBridge, } 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) -> 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 { self.inner.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::TextDelta { text } => text.len(), ReplyChunk::ToolActivity { label } => label.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 { match event { ReplyEvent::TextDelta { text } => Some(ReplyChunk::TextDelta { text }), ReplyEvent::ToolActivity { label } => Some(ReplyChunk::ToolActivity { label }), 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 { .. } => None, ReplyEvent::Heartbeat => None, ReplyEvent::RateLimited { .. } => None, } }