480 lines
19 KiB
Rust
480 lines
19 KiB
Rust
//! 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::collections::HashMap;
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use backend::stream::ReplayOutputBridge;
|
|
use tauri::ipc::Channel;
|
|
|
|
use domain::ids::{AgentId, ProjectId, SessionId};
|
|
use domain::ports::{
|
|
ReplyEvent, ReplyProgress, ReplyProgressKind, ReplyProgressSource, ReplyProgressStage,
|
|
};
|
|
|
|
use crate::dto::{ReplyChunk, ReplyProgressDto, ReplyProgressKindDto, ReplyProgressStageDto};
|
|
|
|
/// Visible terminal message when a structured stream ends without the authoritative
|
|
/// `ReplyEvent::Final`.
|
|
pub const MODEL_RESPONSE_NOT_FINALIZED_MESSAGE: &str =
|
|
"La réponse du modèle n'a pas pu être finalisée.";
|
|
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>,
|
|
successors: Mutex<HashMap<SessionId, SessionId>>,
|
|
owners: Mutex<HashMap<SessionId, (ProjectId, AgentId)>>,
|
|
}
|
|
|
|
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,
|
|
),
|
|
successors: Mutex::new(HashMap::new()),
|
|
owners: Mutex::new(HashMap::new()),
|
|
}
|
|
}
|
|
|
|
fn canonical_session(&self, session: &SessionId) -> SessionId {
|
|
let Ok(successors) = self.successors.lock() else {
|
|
return *session;
|
|
};
|
|
let mut current = *session;
|
|
for _ in 0..successors.len() {
|
|
let Some(next) = successors.get(¤t).copied() else {
|
|
break;
|
|
};
|
|
if next == current {
|
|
break;
|
|
}
|
|
current = next;
|
|
}
|
|
current
|
|
}
|
|
|
|
/// Returns the latest transport session id for a logical chat that crossed
|
|
/// one or more structured-runtime restarts.
|
|
#[must_use]
|
|
pub fn successor(&self, session: &SessionId) -> SessionId {
|
|
self.canonical_session(session)
|
|
}
|
|
|
|
/// Associates a transport session with the stable runtime owner used to
|
|
/// recover the current session after an out-of-band restart.
|
|
pub fn bind_owner(&self, session: SessionId, project_id: ProjectId, agent_id: AgentId) {
|
|
let session = self.canonical_session(&session);
|
|
if let Ok(mut owners) = self.owners.lock() {
|
|
owners.insert(session, (project_id, agent_id));
|
|
}
|
|
}
|
|
|
|
/// Returns the stable owner previously bound to this logical chat.
|
|
#[must_use]
|
|
pub fn owner(&self, session: &SessionId) -> Option<(ProjectId, AgentId)> {
|
|
let canonical = self.canonical_session(session);
|
|
self.owners
|
|
.lock()
|
|
.ok()
|
|
.and_then(|owners| owners.get(&canonical).copied())
|
|
}
|
|
|
|
/// Moves a logical chat to the replacement runtime session while preserving
|
|
/// its sink and retained transcript. Future writes using either the old or
|
|
/// new id resolve to the replacement id.
|
|
pub fn rebind_session(&self, previous: SessionId, replacement: SessionId) {
|
|
let previous = self.canonical_session(&previous);
|
|
if previous == replacement {
|
|
return;
|
|
}
|
|
if let Ok(mut successors) = self.successors.lock() {
|
|
for successor in successors.values_mut() {
|
|
if *successor == previous {
|
|
*successor = replacement;
|
|
}
|
|
}
|
|
successors.insert(previous, replacement);
|
|
}
|
|
if let Ok(mut owners) = self.owners.lock() {
|
|
if let Some(owner) = owners.remove(&previous) {
|
|
owners.entry(replacement).or_insert(owner);
|
|
}
|
|
}
|
|
self.inner.rekey(&previous, replacement);
|
|
}
|
|
|
|
/// 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 {
|
|
let session = self.canonical_session(&session);
|
|
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(&self.canonical_session(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(&self.canonical_session(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) {
|
|
let canonical = self.canonical_session(session);
|
|
self.inner.unregister(&canonical);
|
|
if let Ok(mut successors) = self.successors.lock() {
|
|
successors.retain(|from, to| *from != canonical && *to != canonical);
|
|
}
|
|
if let Ok(mut owners) = self.owners.lock() {
|
|
owners.remove(&canonical);
|
|
}
|
|
}
|
|
|
|
/// Detaches the current output channel while preserving the live session's
|
|
/// retained scrollback. This is the view-unmount path; it is intentionally
|
|
/// distinct from [`unregister`](Self::unregister), which is the real close.
|
|
pub fn detach(&self, session: &SessionId) {
|
|
self.inner.detach(&self.canonical_session(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(&self.canonical_session(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(&self.canonical_session(session), chunk)
|
|
}
|
|
|
|
/// Sends `chunk`, first completing any previous running progress entity when
|
|
/// the next chunk proves that another process/tool/MCP has taken over.
|
|
pub fn send_output_settled(
|
|
&self,
|
|
session: &SessionId,
|
|
settler: &mut ProgressSettler,
|
|
chunk: ReplyChunk,
|
|
) -> bool {
|
|
let mut delivered = true;
|
|
if let Some(completed) = settler.before(&chunk) {
|
|
delivered &= self.send_output(session, completed);
|
|
}
|
|
let current_delivered = self.send_output(session, chunk);
|
|
delivered && current_delivered
|
|
}
|
|
|
|
/// Number of currently-tracked sessions (handy for tests/diagnostics).
|
|
#[must_use]
|
|
pub fn active_sessions(&self) -> usize {
|
|
self.inner.active_sessions()
|
|
}
|
|
}
|
|
|
|
/// Tracks one running progress entity for a chat stream and emits a synthetic
|
|
/// `completed` update when subsequent activity makes the previous spinner stale.
|
|
#[derive(Default)]
|
|
pub struct ProgressSettler {
|
|
active: Option<ReplyProgressDto>,
|
|
}
|
|
|
|
impl ProgressSettler {
|
|
/// Completes the currently active progress entity, if any.
|
|
pub fn finish(&mut self) -> Option<ReplyChunk> {
|
|
self.complete_active()
|
|
}
|
|
|
|
fn before(&mut self, chunk: &ReplyChunk) -> Option<ReplyChunk> {
|
|
match chunk {
|
|
ReplyChunk::Progress { progress } => self.before_progress(progress),
|
|
ReplyChunk::TextDelta { .. }
|
|
| ReplyChunk::ToolActivity { .. }
|
|
| ReplyChunk::Final { .. }
|
|
| ReplyChunk::Error { .. }
|
|
| ReplyChunk::UserPrompt { .. } => self.complete_active(),
|
|
}
|
|
}
|
|
|
|
fn before_progress(&mut self, progress: &ReplyProgressDto) -> Option<ReplyChunk> {
|
|
if !is_settleable_progress(progress) {
|
|
return self.complete_active();
|
|
}
|
|
match progress.stage {
|
|
ReplyProgressStageDto::Started | ReplyProgressStageDto::Delta => {
|
|
let next_key = progress_entity_key(progress);
|
|
let previous = self.active.as_ref().and_then(progress_entity_key);
|
|
if previous.as_deref() == next_key.as_deref() {
|
|
self.active = Some(progress.clone());
|
|
None
|
|
} else {
|
|
let completed = self.complete_active();
|
|
self.active = Some(progress.clone());
|
|
completed
|
|
}
|
|
}
|
|
ReplyProgressStageDto::Completed => {
|
|
let next_key = progress_entity_key(progress);
|
|
let previous = self.active.as_ref().and_then(progress_entity_key);
|
|
if previous.as_deref() == next_key.as_deref() {
|
|
self.active = None;
|
|
}
|
|
None
|
|
}
|
|
ReplyProgressStageDto::Info => self.complete_active(),
|
|
}
|
|
}
|
|
|
|
fn complete_active(&mut self) -> Option<ReplyChunk> {
|
|
let mut progress = self.active.take()?;
|
|
progress.stage = ReplyProgressStageDto::Completed;
|
|
Some(ReplyChunk::Progress { progress })
|
|
}
|
|
}
|
|
|
|
fn is_settleable_progress(progress: &ReplyProgressDto) -> bool {
|
|
matches!(
|
|
progress.kind,
|
|
ReplyProgressKindDto::Tool | ReplyProgressKindDto::Mcp | ReplyProgressKindDto::Message
|
|
)
|
|
}
|
|
|
|
fn progress_entity_key(progress: &ReplyProgressDto) -> Option<String> {
|
|
if !is_settleable_progress(progress) {
|
|
return None;
|
|
}
|
|
if let Some(progress_id) = progress
|
|
.progress_id
|
|
.as_deref()
|
|
.map(str::trim)
|
|
.filter(|id| !id.is_empty())
|
|
{
|
|
return Some(progress_id.to_owned());
|
|
}
|
|
let name = progress
|
|
.tool_name
|
|
.as_ref()
|
|
.filter(|name| !name.trim().is_empty())
|
|
.unwrap_or(&progress.label);
|
|
let name = name.trim().to_ascii_lowercase();
|
|
(!name.is_empty()).then(|| format!("{:?}:{name}", progress.kind))
|
|
}
|
|
|
|
fn reply_chunk_bytes(chunk: &ReplyChunk) -> usize {
|
|
match chunk {
|
|
ReplyChunk::UserPrompt {
|
|
at_ms: _,
|
|
text,
|
|
attachments,
|
|
attachment_refs,
|
|
} => {
|
|
text.len()
|
|
+ attachments.iter().map(String::len).sum::<usize>()
|
|
+ attachment_refs
|
|
.iter()
|
|
.map(|attachment| {
|
|
attachment.reference.len()
|
|
+ attachment.filename.len()
|
|
+ attachment.mime.len()
|
|
+ attachment.readable_path.len()
|
|
})
|
|
.sum::<usize>()
|
|
}
|
|
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> {
|
|
chunk_from_event_at(event, None)
|
|
}
|
|
|
|
/// Returns whether one provider event closes the concrete structured turn.
|
|
///
|
|
/// This is the control-plane boundary used by the `agent_send` pump. In
|
|
/// particular, the provider is not required to close its iterator after a
|
|
/// terminal event: keeping the iterator alive must not keep the agent Busy.
|
|
#[must_use]
|
|
pub fn reply_event_ends_turn(event: &ReplyEvent) -> bool {
|
|
matches!(event, ReplyEvent::Final { .. } | ReplyEvent::Error { .. })
|
|
}
|
|
|
|
/// Maps a domain reply event while preserving the backend acceptance time.
|
|
#[must_use]
|
|
pub fn chunk_from_event_at(event: ReplyEvent, at_ms: Option<u64>) -> Option<ReplyChunk> {
|
|
match event {
|
|
ReplyEvent::Progress { progress } => {
|
|
let mut progress: ReplyProgressDto = progress.into();
|
|
progress.occurred_at_ms = at_ms;
|
|
Some(ReplyChunk::Progress { progress })
|
|
}
|
|
ReplyEvent::TextDelta { text } => Some(ReplyChunk::TextDelta { text, at_ms }),
|
|
ReplyEvent::ToolActivity { label } => Some(ReplyChunk::Progress {
|
|
progress: {
|
|
let mut progress: ReplyProgressDto = ReplyProgress::new(
|
|
ReplyProgressSource::ProviderNative,
|
|
ReplyProgressKind::Tool,
|
|
ReplyProgressStage::Info,
|
|
label.clone(),
|
|
)
|
|
.with_tool_name(label)
|
|
.into();
|
|
progress.occurred_at_ms = at_ms;
|
|
progress
|
|
},
|
|
}),
|
|
ReplyEvent::Error { message } => Some(ReplyChunk::Error { message, at_ms }),
|
|
ReplyEvent::Final { content } => {
|
|
if content.trim().is_empty() {
|
|
Some(ReplyChunk::Error {
|
|
message: "Réponse vide du modèle.".to_owned(),
|
|
at_ms,
|
|
})
|
|
} else {
|
|
Some(ReplyChunk::Final { content, at_ms })
|
|
}
|
|
}
|
|
ReplyEvent::Announcement { text } => Some(ReplyChunk::Progress {
|
|
progress: {
|
|
let mut progress: ReplyProgressDto = ReplyProgress::new(
|
|
ReplyProgressSource::ProviderNative,
|
|
ReplyProgressKind::Message,
|
|
ReplyProgressStage::Delta,
|
|
"message intermédiaire",
|
|
)
|
|
.with_text(text)
|
|
.into();
|
|
progress.occurred_at_ms = at_ms;
|
|
progress
|
|
},
|
|
}),
|
|
ReplyEvent::Heartbeat => None,
|
|
ReplyEvent::RateLimited { .. } => None,
|
|
}
|
|
}
|
|
|
|
/// Terminal wire chunk for an exhausted structured stream that never produced the
|
|
/// authoritative `ReplyEvent::Final`.
|
|
#[must_use]
|
|
pub fn unfinished_turn_chunk() -> ReplyChunk {
|
|
ReplyChunk::Error {
|
|
message: MODEL_RESPONSE_NOT_FINALIZED_MESSAGE.to_owned(),
|
|
at_ms: None,
|
|
}
|
|
}
|
|
|
|
/// Assigns an acceptance timestamp to a transport-created chunk.
|
|
#[must_use]
|
|
pub fn with_chunk_timestamp(mut chunk: ReplyChunk, at_ms: Option<u64>) -> ReplyChunk {
|
|
match &mut chunk {
|
|
ReplyChunk::UserPrompt { at_ms: value, .. }
|
|
| ReplyChunk::TextDelta { at_ms: value, .. }
|
|
| ReplyChunk::ToolActivity { at_ms: value, .. }
|
|
| ReplyChunk::Final { at_ms: value, .. }
|
|
| ReplyChunk::Error { at_ms: value, .. } => *value = at_ms,
|
|
ReplyChunk::Progress { progress } => progress.occurred_at_ms = at_ms,
|
|
}
|
|
chunk
|
|
}
|