From c8fef2a76a50b8eacdd4a1670ceb6895fb7edc61 Mon Sep 17 00:00:00 2001 From: Blomios Date: Wed, 15 Jul 2026 12:50:29 +0200 Subject: [PATCH] refactor(backend): abstraire le sink de stream hors de tauri::ipc::Channel (#13) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Lot B2 du chantier server/client mode : le cœur backend émet désormais ses flux via une abstraction de sink agnostique, sans dépendre directement de tauri::ipc::Channel, préalable au futur serveur web + PTY WebSocket. - crates/backend : abstraction de sink (stream.rs) câblée dans lib.rs. - crates/app-tauri : implémentation Tauri du sink (stream.rs) et adaptation des surfaces lib.rs, pty.rs, chat.rs. - Nettoyage clippy des 2 warnings B2. Validé : cargo check --workspace vert, tests backend/app-tauri verts, cœur agnostique Tauri, clippy B2 propre. Co-Authored-By: Claude Opus 4.8 --- crates/app-tauri/src/chat.rs | 124 +++++-------------- crates/app-tauri/src/lib.rs | 1 + crates/app-tauri/src/pty.rs | 65 ++++------ crates/app-tauri/src/stream.rs | 31 +++++ crates/backend/src/lib.rs | 7 +- crates/backend/src/stream.rs | 216 +++++++++++++++++++++++++++++++++ 6 files changed, 305 insertions(+), 139 deletions(-) create mode 100644 crates/app-tauri/src/stream.rs create mode 100644 crates/backend/src/stream.rs diff --git a/crates/app-tauri/src/chat.rs b/crates/app-tauri/src/chat.rs index 13d93ef..1cf82da 100644 --- a/crates/app-tauri/src/chat.rs +++ b/crates/app-tauri/src/chat.rs @@ -1,16 +1,17 @@ -//! Generic **structured reply ↔ Tauri Channel** bridge infrastructure. +//! 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 [`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. +//! 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 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`]). +//! [`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 @@ -23,44 +24,35 @@ //! [`unregister_if`]: ChatBridge::unregister_if //! [`send_output`]: ChatBridge::send_output -use std::collections::HashMap; -use std::sync::Mutex; +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; -/// 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>, - /// Recent chunks retained for reattach, bounded by chunk count and estimated - /// byte size; older chunks are dropped from the head. This is a transport - /// replay buffer, not durable conversation history. - scrollback: Vec, -} - -/// Registry mapping live structured (chat) sessions to their reply [`Channel`] +/// 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`]. -#[derive(Default)] pub struct ChatBridge { - entries: Mutex>, + inner: ReplayOutputBridge, +} + +impl Default for ChatBridge { + fn default() -> Self { + Self::new() + } } impl ChatBridge { @@ -68,7 +60,11 @@ impl ChatBridge { #[must_use] pub fn new() -> Self { Self { - entries: Mutex::new(HashMap::new()), + inner: ReplayOutputBridge::new( + MAX_CHAT_SCROLLBACK_CHUNKS, + MAX_CHAT_SCROLLBACK_BYTES, + reply_chunk_bytes, + ), } } @@ -80,28 +76,8 @@ impl ChatBridge { /// 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 - } + self.inner + .register(session, Arc::new(TauriChannelSink::new(channel))) } /// Returns the conversation scrollback retained for a session (the chunks @@ -112,20 +88,14 @@ impl ChatBridge { /// `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() + 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) { - if let Ok(mut map) = self.entries.lock() { - map.remove(session); - } + self.inner.unregister(session); } /// Detaches a session's channel **only if** `gen` is still the current @@ -139,13 +109,7 @@ impl ChatBridge { /// /// [`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; - } - } - } + self.inner.detach_if(session, gen); } /// Records a chunk in the session's scrollback and, if a view is attached at @@ -157,40 +121,16 @@ impl ChatBridge { /// 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()); - trim_scrollback(entry); - match &entry.channel { - Some(channel) => channel.send(chunk).is_ok(), - None => false, - } + self.inner.send_output(session, chunk) } /// 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) + self.inner.active_sessions() } } -fn trim_scrollback(entry: &mut ChatEntry) { - while !entry.scrollback.is_empty() - && (entry.scrollback.len() > MAX_CHAT_SCROLLBACK_CHUNKS - || scrollback_bytes(&entry.scrollback) > MAX_CHAT_SCROLLBACK_BYTES) - { - entry.scrollback.remove(0); - } -} - -fn scrollback_bytes(chunks: &[ReplyChunk]) -> usize { - chunks.iter().map(reply_chunk_bytes).sum() -} - fn reply_chunk_bytes(chunk: &ReplyChunk) -> usize { match chunk { ReplyChunk::TextDelta { text } => text.len(), diff --git a/crates/app-tauri/src/lib.rs b/crates/app-tauri/src/lib.rs index d349d82..8829fa7 100644 --- a/crates/app-tauri/src/lib.rs +++ b/crates/app-tauri/src/lib.rs @@ -22,6 +22,7 @@ pub mod mcp_endpoint; pub mod openai_tools; pub mod pty; pub mod state; +pub mod stream; pub mod tickets; use std::process::ExitCode; diff --git a/crates/app-tauri/src/pty.rs b/crates/app-tauri/src/pty.rs index fa43e8d..b5b1ac5 100644 --- a/crates/app-tauri/src/pty.rs +++ b/crates/app-tauri/src/pty.rs @@ -1,27 +1,27 @@ -//! Generic **PTY ↔ Tauri Channel** bridge infrastructure. +//! Generic **PTY ↔ outbound stream sink** bridge infrastructure. //! -//! ARCHITECTURE §2 decides that high-frequency PTY byte streams travel over -//! per-session [`tauri::ipc::Channel`]s rather than global events, for -//! throughput and isolation. This module provides the transport-side plumbing -//! that L3 will plug a real `PtyPort` into; here there is **no real PTY** yet — -//! only the registry + the abstraction that routes byte chunks to the right -//! frontend channel. +//! ARCHITECTURE §2 decides that high-frequency PTY byte streams travel over a +//! per-session stream channel rather than global events, for throughput and +//! isolation. This module provides the desktop-facing plumbing that adapts the +//! shared backend sink abstraction to Tauri IPC. //! //! Design: //! - The frontend opens a terminal and passes a [`tauri::ipc::Channel`] for that -//! session. The backend registers it in [`PtyBridge`] keyed by `SessionId`. -//! - Whatever produces output (the PTY adapter in L3) calls -//! [`PtyBridge::send_output`], which forwards the bytes on the matching -//! channel. Bytes are sent as-is; the frontend xterm wrapper consumes them. +//! session. The adapter wraps it as a sink and registers it in [`PtyBridge`]. +//! - Whatever produces output calls [`PtyBridge::send_output`], which forwards +//! the bytes through the current sink. Bytes are sent as-is; the frontend +//! xterm wrapper consumes them. //! - [`PtyBridge::unregister`] tears the channel down on terminal close. -use std::collections::HashMap; -use std::sync::Mutex; +use std::sync::Arc; +use backend::stream::OutputBridge; use tauri::ipc::Channel; use domain::ids::SessionId; +use crate::stream::TauriChannelSink; + /// A chunk of PTY output bytes destined for a specific session's channel. /// /// Sent as a raw byte vector; serde encodes it for the IPC channel. Kept as a @@ -29,16 +29,12 @@ use domain::ids::SessionId; /// backpressure handling) without touching call sites. pub type PtyChunk = Vec; -/// Registry mapping live terminal sessions to their output [`Channel`]. +/// Registry mapping live terminal sessions to their output sink. /// /// Thread-safe; cloned `Arc` is held in [`crate::state::AppState`]. #[derive(Default)] pub struct PtyBridge { - /// Per session: a monotonically-increasing **generation** plus the current - /// output channel. The generation distinguishes successive (re-)attaches so a - /// superseded pump thread can't tear down the channel of the attach that - /// replaced it (see [`PtyBridge::unregister_if`]). - channels: Mutex)>>, + inner: OutputBridge, } impl PtyBridge { @@ -46,7 +42,7 @@ impl PtyBridge { #[must_use] pub fn new() -> Self { Self { - channels: Mutex::new(HashMap::new()), + inner: OutputBridge::new(), } } @@ -55,20 +51,13 @@ impl PtyBridge { /// generation, so the caller's pump thread can later tear down *only its own* /// registration via [`unregister_if`](Self::unregister_if). pub fn register(&self, session: SessionId, channel: Channel) -> u64 { - if let Ok(mut map) = self.channels.lock() { - let gen = map.get(&session).map_or(0, |(g, _)| g.wrapping_add(1)); - map.insert(session, (gen, channel)); - gen - } else { - 0 - } + self.inner + .register(session, Arc::new(TauriChannelSink::new(channel))) } /// Removes a session's channel unconditionally (terminal explicitly closed). pub fn unregister(&self, session: &SessionId) { - if let Ok(mut map) = self.channels.lock() { - map.remove(session); - } + self.inner.unregister(session); } /// Removes a session's channel **only if** `gen` is still the current @@ -76,11 +65,7 @@ impl PtyBridge { /// session has since been re-attached (newer generation), this is a no-op, so /// the dying thread never unregisters the live channel that superseded it. pub fn unregister_if(&self, session: &SessionId, gen: u64) { - if let Ok(mut map) = self.channels.lock() { - if matches!(map.get(session), Some((g, _)) if *g == gen) { - map.remove(session); - } - } + self.inner.unregister_if(session, gen); } /// Forwards a chunk of output bytes to a session's channel. @@ -89,18 +74,12 @@ impl PtyBridge { /// registered for the session (e.g. already closed). In L3 the PTY adapter's /// output stream drives this. pub fn send_output(&self, session: &SessionId, chunk: PtyChunk) -> bool { - let Ok(map) = self.channels.lock() else { - return false; - }; - match map.get(session) { - Some((_, channel)) => channel.send(chunk).is_ok(), - None => false, - } + self.inner.send_output(session, chunk) } /// Number of currently-registered sessions (handy for tests/diagnostics). #[must_use] pub fn active_sessions(&self) -> usize { - self.channels.lock().map(|m| m.len()).unwrap_or(0) + self.inner.active_sessions() } } diff --git a/crates/app-tauri/src/stream.rs b/crates/app-tauri/src/stream.rs new file mode 100644 index 0000000..5b357c6 --- /dev/null +++ b/crates/app-tauri/src/stream.rs @@ -0,0 +1,31 @@ +//! Tauri implementation of the shared outbound stream sink contract. +//! +//! The backend core owns the transport-agnostic [`backend::stream::OutputSink`] +//! abstraction. This adapter is the only layer that turns it into a concrete +//! [`tauri::ipc::Channel`] send. + +use backend::stream::{OutputSink, OutputSinkError}; +use serde::Serialize; +use tauri::ipc::Channel; + +/// [`OutputSink`] backed by a per-attach Tauri IPC [`Channel`]. +pub struct TauriChannelSink { + channel: Channel, +} + +impl TauriChannelSink { + /// Wraps a Tauri IPC channel as a shared backend stream sink. + #[must_use] + pub fn new(channel: Channel) -> Self { + Self { channel } + } +} + +impl OutputSink for TauriChannelSink +where + T: Serialize + Send + Sync + 'static, +{ + fn send(&self, item: T) -> Result<(), OutputSinkError> { + self.channel.send(item).map_err(|_| OutputSinkError::Closed) + } +} diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index 69345a2..b8b699d 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -85,6 +85,7 @@ use infrastructure::{ pub mod mcp_endpoint; pub mod openai_tools; +pub mod stream; use crate::mcp_endpoint::{mcp_endpoint, AppMcpRuntimeProvider, McpEndpoint}; use crate::openai_tools::{AppOpenAiToolInvoker, LateBoundOpenAiToolInvoker}; @@ -114,10 +115,8 @@ mod backend_core_build_tests { #[test] fn backend_core_builds_without_desktop_runtime() { - let dir = std::env::temp_dir().join(format!( - "idea-backend-core-build-{}", - uuid::Uuid::new_v4() - )); + let dir = + std::env::temp_dir().join(format!("idea-backend-core-build-{}", uuid::Uuid::new_v4())); let core = BackendCore::build(dir.clone()); assert!(!core.home_dir.is_empty()); diff --git a/crates/backend/src/stream.rs b/crates/backend/src/stream.rs new file mode 100644 index 0000000..96a1e21 --- /dev/null +++ b/crates/backend/src/stream.rs @@ -0,0 +1,216 @@ +use std::collections::HashMap; +use std::hash::Hash; +use std::sync::{Arc, Mutex}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum OutputSinkError { + Closed, + Full, + Other(String), +} + +pub trait OutputSink: Send + Sync + 'static { + fn send(&self, item: T) -> Result<(), OutputSinkError>; +} + +type SharedOutputSink = Arc>; +type SinkRegistry = HashMap)>; + +pub struct OutputBridge +where + K: Copy + Eq + Hash, +{ + sinks: Mutex>, +} + +impl Default for OutputBridge +where + K: Copy + Eq + Hash, + T: 'static, +{ + fn default() -> Self { + Self::new() + } +} + +impl OutputBridge +where + K: Copy + Eq + Hash, + T: 'static, +{ + #[must_use] + pub fn new() -> Self { + Self { + sinks: Mutex::new(HashMap::new()), + } + } + + pub fn register(&self, key: K, sink: SharedOutputSink) -> u64 { + if let Ok(mut map) = self.sinks.lock() { + let gen = map.get(&key).map_or(0, |(g, _)| g.wrapping_add(1)); + map.insert(key, (gen, sink)); + gen + } else { + 0 + } + } + + pub fn unregister(&self, key: &K) { + if let Ok(mut map) = self.sinks.lock() { + map.remove(key); + } + } + + pub fn unregister_if(&self, key: &K, gen: u64) { + if let Ok(mut map) = self.sinks.lock() { + if matches!(map.get(key), Some((g, _)) if *g == gen) { + map.remove(key); + } + } + } + + pub fn try_send_output(&self, key: &K, item: T) -> Result<(), OutputSinkError> { + let sink = { + let Ok(map) = self.sinks.lock() else { + return Err(OutputSinkError::Closed); + }; + map.get(key) + .map(|(_, sink)| Arc::clone(sink)) + .ok_or(OutputSinkError::Closed)? + }; + sink.send(item) + } + + pub fn send_output(&self, key: &K, item: T) -> bool { + self.try_send_output(key, item).is_ok() + } + + #[must_use] + pub fn active_sessions(&self) -> usize { + self.sinks.lock().map(|m| m.len()).unwrap_or(0) + } +} + +struct ReplayEntry { + generation: u64, + sink: Option>, + scrollback: Vec, +} + +pub struct ReplayOutputBridge +where + K: Copy + Eq + Hash, +{ + entries: Mutex>>, + max_chunks: usize, + max_bytes: usize, + measure: fn(&T) -> usize, +} + +impl ReplayOutputBridge +where + K: Copy + Eq + Hash, + T: Clone + 'static, +{ + #[must_use] + pub fn new(max_chunks: usize, max_bytes: usize, measure: fn(&T) -> usize) -> Self { + Self { + entries: Mutex::new(HashMap::new()), + max_chunks, + max_bytes, + measure, + } + } + + pub fn register(&self, key: K, sink: SharedOutputSink) -> u64 { + if let Ok(mut map) = self.entries.lock() { + match map.get_mut(&key) { + Some(entry) => { + entry.generation = entry.generation.wrapping_add(1); + entry.sink = Some(sink); + entry.generation + } + None => { + map.insert( + key, + ReplayEntry { + generation: 0, + sink: Some(sink), + scrollback: Vec::new(), + }, + ); + 0 + } + } + } else { + 0 + } + } + + #[must_use] + pub fn scrollback(&self, key: &K) -> Vec { + self.entries + .lock() + .ok() + .and_then(|m| m.get(key).map(|e| e.scrollback.clone())) + .unwrap_or_default() + } + + pub fn unregister(&self, key: &K) { + if let Ok(mut map) = self.entries.lock() { + map.remove(key); + } + } + + pub fn detach_if(&self, key: &K, gen: u64) { + if let Ok(mut map) = self.entries.lock() { + if let Some(entry) = map.get_mut(key) { + if entry.generation == gen { + entry.sink = None; + } + } + } + } + + pub fn try_send_output(&self, key: &K, item: T) -> Result<(), OutputSinkError> { + let sink = { + let Ok(mut map) = self.entries.lock() else { + return Err(OutputSinkError::Closed); + }; + let Some(entry) = map.get_mut(key) else { + return Err(OutputSinkError::Closed); + }; + entry.scrollback.push(item.clone()); + trim_scrollback(entry, self.max_chunks, self.max_bytes, self.measure); + entry.sink.as_ref().map(Arc::clone) + }; + + match sink { + Some(sink) => sink.send(item), + None => Err(OutputSinkError::Closed), + } + } + + pub fn send_output(&self, key: &K, item: T) -> bool { + self.try_send_output(key, item).is_ok() + } + + #[must_use] + pub fn active_sessions(&self) -> usize { + self.entries.lock().map(|m| m.len()).unwrap_or(0) + } +} + +fn trim_scrollback( + entry: &mut ReplayEntry, + max_chunks: usize, + max_bytes: usize, + measure: fn(&T) -> usize, +) { + while !entry.scrollback.is_empty() + && (entry.scrollback.len() > max_chunks + || entry.scrollback.iter().map(measure).sum::() > max_bytes) + { + entry.scrollback.remove(0); + } +}