blomios/IdeA
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(backend): abstraire le sink de stream hors de tauri::ipc::Channel (#13)

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Blomios
2026-07-15 12:50:29 +02:00
parent 955db79e97
commit c8fef2a76a
6 changed files with 305 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
crates/app-tauri/src/chat.rs
View File

@ -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<Channel<ReplyChunk>>,
/// 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<ReplyChunk>,
}
/// 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<ChatBridge>` is held in [`crate::state::AppState`],
/// the twin of [`crate::pty::PtyBridge`].
#[derive(Default)]
pub struct ChatBridge {
entries: Mutex<HashMap<SessionId, ChatEntry>>,
inner: ReplayOutputBridge<SessionId, ReplyChunk>,
}
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<ReplyChunk>) -> 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<ReplyChunk> {
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(),

1
crates/app-tauri/src/lib.rs
View File

@ -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;

65
crates/app-tauri/src/pty.rs
View File

@ -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<u8>;
/// Registry mapping live terminal sessions to their output [`Channel`].
/// Registry mapping live terminal sessions to their output sink.
///
/// Thread-safe; cloned `Arc<PtyBridge>` 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<HashMap<SessionId, (u64, Channel<PtyChunk>)>>,
inner: OutputBridge<SessionId, PtyChunk>,
}
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<PtyChunk>) -> 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()
}
}

31
crates/app-tauri/src/stream.rs Normal file
View File

@ -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<T> {
channel: Channel<T>,
}
impl<T> TauriChannelSink<T> {
/// Wraps a Tauri IPC channel as a shared backend stream sink.
#[must_use]
pub fn new(channel: Channel<T>) -> Self {
Self { channel }
}
}
impl<T> OutputSink<T> for TauriChannelSink<T>
where
T: Serialize + Send + Sync + 'static,
{
fn send(&self, item: T) -> Result<(), OutputSinkError> {
self.channel.send(item).map_err(|_| OutputSinkError::Closed)
}
}

7
crates/backend/src/lib.rs
View File

@ -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());

216
crates/backend/src/stream.rs Normal file
View File

@ -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<T>: Send + Sync + 'static {
fn send(&self, item: T) -> Result<(), OutputSinkError>;
}
type SharedOutputSink<T> = Arc<dyn OutputSink<T>>;
type SinkRegistry<K, T> = HashMap<K, (u64, SharedOutputSink<T>)>;
pub struct OutputBridge<K, T>
where
K: Copy + Eq + Hash,
{
sinks: Mutex<SinkRegistry<K, T>>,
}
impl<K, T> Default for OutputBridge<K, T>
where
K: Copy + Eq + Hash,
T: 'static,
{
fn default() -> Self {
Self::new()
}
}
impl<K, T> OutputBridge<K, T>
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<T>) -> 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<T> {
generation: u64,
sink: Option<SharedOutputSink<T>>,
scrollback: Vec<T>,
}
pub struct ReplayOutputBridge<K, T>
where
K: Copy + Eq + Hash,
{
entries: Mutex<HashMap<K, ReplayEntry<T>>>,
max_chunks: usize,
max_bytes: usize,
measure: fn(&T) -> usize,
}
impl<K, T> ReplayOutputBridge<K, T>
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<T>) -> 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<T> {
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<T>(
entry: &mut ReplayEntry<T>,
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::<usize>() > max_bytes)
{
entry.scrollback.remove(0);
}
}