//! Generic **structured reply ↔ Tauri Channel** bridge infrastructure. //! //! Twin of [`crate::pty::PtyBridge`] (ARCHITECTURE §17.7). Where `PtyBridge` //! routes raw PTY byte chunks to a per-session [`tauri::ipc::Channel`], 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 backend 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::collections::HashMap; use std::sync::Mutex; use tauri::ipc::Channel; use domain::ids::SessionId; use domain::ports::ReplyEvent; use crate::dto::ReplyChunk; /// Per-session transport state: the current attach generation, its output /// channel, and the conversation scrollback recorded so far. struct ChatEntry { /// Monotonic generation of the current (re-)attach (anti double-pump). generation: u64, /// The output channel of the current attach, if one is registered. `None` /// when the session has been recorded (scrollback kept) but no view is /// currently attached — the pump then only appends to the scrollback. channel: Option>, /// Every chunk routed for this session, in order — the conversation /// scrollback replayed on re-attach. Bounded only by the conversation length /// (a turn count), like the PTY ring buffer is bounded by its byte capacity. scrollback: Vec, } /// Registry mapping live structured (chat) sessions to their reply [`Channel`] /// plus a retained conversation scrollback. /// /// Thread-safe; a cloned `Arc` is held in [`crate::state::AppState`], /// the twin of [`crate::pty::PtyBridge`]. #[derive(Default)] pub struct ChatBridge { entries: Mutex>, } impl ChatBridge { /// Creates an empty bridge. #[must_use] pub fn new() -> Self { Self { entries: Mutex::new(HashMap::new()), } } /// 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 { if let Ok(mut map) = self.entries.lock() { match map.get_mut(&session) { Some(entry) => { entry.generation = entry.generation.wrapping_add(1); entry.channel = Some(channel); entry.generation } None => { map.insert( session, ChatEntry { generation: 0, channel: Some(channel), scrollback: Vec::new(), }, ); 0 } } } else { 0 } } /// 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.entries .lock() .ok() .and_then(|m| m.get(session).map(|e| e.scrollback.clone())) .unwrap_or_default() } /// 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) { if let Ok(mut map) = self.entries.lock() { map.remove(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) { if let Ok(mut map) = self.entries.lock() { if let Some(entry) = map.get_mut(session) { if entry.generation == gen { entry.channel = None; } } } } /// 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 { let Ok(mut map) = self.entries.lock() else { return false; }; let Some(entry) = map.get_mut(session) else { return false; }; entry.scrollback.push(chunk.clone()); match &entry.channel { Some(channel) => channel.send(chunk).is_ok(), None => false, } } /// Number of currently-tracked sessions (handy for tests/diagnostics). #[must_use] pub fn active_sessions(&self) -> usize { self.entries.lock().map(|m| m.len()).unwrap_or(0) } } /// 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::Final { content } => Some(ReplyChunk::Final { content }), ReplyEvent::Announcement { .. } => None, ReplyEvent::Heartbeat => None, ReplyEvent::RateLimited { .. } => None, } }