fix(chat): le transcript s'affiche à nouveau après redémarrage automatique d'un agent — #295 (app-tauri ChatBridge: cartes successors/owners, rebind_session déplace sink+scrolloback du logical chat vers la session de remplacement et canonical_session résout les écritures avec l'ancien id sur toutes les opérations register/scrollback/detach/send_output; backend stream: ReplayOutputBridge::rekey prépare le scrollback source devant le destination et invalide les générations pré-rekey des deux pumps; commands: agent_send binde l'owner stable (projectId,agentId) au tour humain et rebind avant le flux de continuation au handoff restart; reattach_agent_chat résout le successeur du bridge puis fallback identité stable projectId/agentId explicite ou owner bindé, renvoie sessionId effectif + conversationId, NOT_FOUND sinon; web-server chat.attach même fallback stale→live par project/agent; DTO ReattachChatDto/ReattachResultDto documentent l'id effectif post-restart; frontend: ReattachAgentChatOptions, conversationId dans ChatAttachedPayload, CustomAgentChatView réattache sur agentRestartCompleted (dédupe par restartId, scrollback appliqué, tours restart préservés), retry identité stable sur NOT_FOUND dans openOrAttach avant fallback launch frais, publication sessionId effectif + conversationId, état vide «Conversation liée» quand une conversation/run/workstate lié existe; QA verte reconfirmée par Git: git diff --check, cargo fmt --all --check, app-tauri chat_bridge 34/34 dont 3 nouveaux restart_rebind + dto_chat 24/24, cargo check -p app-tauri, backend 91+7 ignored permissions socket, web-server websocket_chat_attach 3/3, npm test 77/77 CustomAgentChatView+agentGateway, npm run build; réserves non bloquantes: warnings Vite esbuild/oxc et chunk >500kB connus)

This commit is contained in:
2026-09-09 18:44:01 +02:00
parent b1c173f7d0
commit 1b1476de52
14 changed files with 666 additions and 37 deletions

View File

@ -24,12 +24,13 @@
//! [`unregister_if`]: ChatBridge::unregister_if
//! [`send_output`]: ChatBridge::send_output
use std::sync::Arc;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use backend::stream::ReplayOutputBridge;
use tauri::ipc::Channel;
use domain::ids::SessionId;
use domain::ids::{AgentId, ProjectId, SessionId};
use domain::ports::{
ReplyEvent, ReplyProgress, ReplyProgressKind, ReplyProgressSource, ReplyProgressStage,
};
@ -54,6 +55,8 @@ pub const MAX_CHAT_SCROLLBACK_BYTES: usize = 512 * 1024;
/// 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 {
@ -72,9 +75,78 @@ impl ChatBridge {
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(&current).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*
@ -83,6 +155,7 @@ 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 {
let session = self.canonical_session(&session);
self.inner
.register(session, Arc::new(TauriChannelSink::new(channel)))
}
@ -95,7 +168,7 @@ impl ChatBridge {
/// `PtyPort::scrollback`.
#[must_use]
pub fn scrollback(&self, session: &SessionId) -> Vec<ReplyChunk> {
self.inner.scrollback(session)
self.inner.scrollback(&self.canonical_session(session))
}
/// Clears only the retained conversation scrollback for a live session,
@ -105,21 +178,29 @@ impl ChatBridge {
/// 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(session)
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) {
self.inner.unregister(session);
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(session);
self.inner.detach(&self.canonical_session(session));
}
/// Detaches a session's channel **only if** `gen` is still the current
@ -133,7 +214,7 @@ impl ChatBridge {
///
/// [`unregister`]: ChatBridge::unregister
pub fn detach_if(&self, session: &SessionId, gen: u64) {
self.inner.detach_if(session, gen);
self.inner.detach_if(&self.canonical_session(session), gen);
}
/// Records a chunk in the session's scrollback and, if a view is attached at
@ -145,7 +226,8 @@ 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 {
self.inner.send_output(session, chunk)
self.inner
.send_output(&self.canonical_session(session), chunk)
}
/// Sends `chunk`, first completing any previous running progress entity when

View File

@ -2619,6 +2619,9 @@ pub async fn agent_send(
.map(|attachment| attachment.filename.clone())
.collect();
let session_meta = state.structured_sessions.meta_for_session(&sid);
if let Some((project_id, agent_id, _, _)) = &session_meta {
state.chat_bridge.bind_owner(sid, *project_id, *agent_id);
}
let turn_guard = session_meta.as_ref().map(|(project_id, agent_id, _, _)| {
backend::StructuredHumanTurnGuard::start_with_source(
std::sync::Arc::clone(&state.turn_watch_input),
@ -2830,6 +2833,11 @@ pub async fn agent_send(
);
break;
};
// The provider replacement owns a fresh SessionId, but this remains
// one logical human chat. Move the transport transcript/sink before
// opening the continuation stream; writes still made with `sid` are
// resolved by ChatBridge to the replacement id.
bridge.rebind_session(sid, replacement.id());
let resume_prompt = resume_prompt
.unwrap_or_else(|| application::SELF_RESTART_CONTINUATION_PROMPT.to_owned());
match runtime.block_on(replacement.send(&resume_prompt)) {
@ -3413,23 +3421,63 @@ pub async fn set_front_attached(
/// then receives subsequent chunks over `on_reply`.
///
/// # Errors
/// Returns an [`ErrorDto`] (`INVALID` for a malformed id, `NOT_FOUND` if no live
/// structured session owns the id — the caller then falls back to a fresh launch).
/// `project_id` + `agent_id` are an optional stable fallback for callers whose
/// stored transport id predates a restart. The response always carries the
/// effective live `sessionId`, its current scrollback and its provider
/// `conversationId` when known.
///
/// Returns an [`ErrorDto`] (`INVALID` for malformed ids, `NOT_FOUND` if neither
/// the requested/successor session nor the project-agent fallback is live).
#[tauri::command]
pub fn reattach_agent_chat(
session_id: String,
project_id: Option<String>,
agent_id: Option<String>,
on_reply: Channel<ReplyChunk>,
state: State<'_, AppState>,
) -> Result<ReattachChatDto, ErrorDto> {
let sid = parse_session_id(&session_id)?;
let requested_sid = parse_session_id(&session_id)?;
// A missing live session means the conversation is gone (closed/never live) —
// surfaced as NOT_FOUND so the caller falls back to opening a fresh cell.
// A restart replaces the provider session id. Prefer the bridge successor
// recorded by the restart-aware pump, then allow the caller to recover by
// stable project/agent identity if it only retained a dead id.
let mut sid = state.chat_bridge.successor(&requested_sid);
if state.structured_sessions.session(&sid).is_none() {
return Err(ErrorDto::from(AppError::NotFound(format!(
"structured session {sid}"
))));
let explicit_owner = match (project_id.as_deref(), agent_id.as_deref()) {
(Some(project_id), Some(agent_id)) => {
Some((parse_project_id(project_id)?, parse_agent_id(agent_id)?))
}
(None, None) => None,
_ => {
return Err(ErrorDto::from(AppError::Invalid(
"projectId and agentId must be provided together".to_owned(),
)));
}
};
sid = match explicit_owner.or_else(|| state.chat_bridge.owner(&requested_sid)) {
Some((project_id, agent_id)) => {
let replacement = state
.structured_sessions
.session_id_for_agent_in_project(project_id, &agent_id)
.ok_or_else(|| {
ErrorDto::from(AppError::NotFound(format!(
"live structured session for agent {agent_id}"
)))
})?;
state.chat_bridge.rebind_session(requested_sid, replacement);
replacement
}
None => {
return Err(ErrorDto::from(AppError::NotFound(format!(
"structured session {requested_sid}"
))));
}
};
}
let conversation_id = state
.structured_sessions
.meta_for_session(&sid)
.and_then(|(_, _, _, conversation_id)| conversation_id);
// (1) Snapshot the conversation scrollback before swapping the channel.
let scrollback = state.chat_bridge.scrollback(&sid);
@ -3440,7 +3488,8 @@ pub fn reattach_agent_chat(
let _gen = state.chat_bridge.register(sid, on_reply);
Ok(ReattachChatDto {
session_id,
session_id: sid.to_string(),
conversation_id,
scrollback,
})
}

View File

@ -550,6 +550,84 @@ fn reattach_repaints_scrollback_and_preserves_it_without_resend() {
);
}
#[test]
fn restart_rebind_preserves_transcript_and_routes_old_id_writes_to_replacement() {
let bridge = ChatBridge::new();
let previous = sid();
let replacement = sid();
let sink = Arc::new(Mutex::new(Vec::new()));
let generation = bridge.register(previous, capturing_channel(Arc::clone(&sink)));
assert!(bridge.send_output(&previous, delta("before restart")));
bridge.rebind_session(previous, replacement);
assert_eq!(bridge.successor(&previous), replacement);
assert_eq!(
bridge.scrollback(&replacement),
vec![delta("before restart")]
);
assert!(
bridge.send_output(&previous, delta("after restart")),
"the in-flight old pump must reach the rekeyed live sink"
);
assert_eq!(
bridge.scrollback(&replacement),
vec![delta("before restart"), delta("after restart")]
);
assert_eq!(sink.lock().unwrap().len(), 2);
bridge.detach_if(&previous, generation);
assert!(
!bridge.send_output(&replacement, delta("detached")),
"the old pump generation detaches the migrated sink"
);
assert_eq!(bridge.scrollback(&previous).len(), 3);
}
#[test]
fn restart_rebind_moves_stable_owner_to_the_replacement_session() {
let bridge = ChatBridge::new();
let previous = sid();
let replacement = sid();
let project = domain::ProjectId::from_uuid(Uuid::from_u128(10));
let agent = domain::AgentId::from_uuid(Uuid::from_u128(20));
bridge.bind_owner(previous, project, agent);
bridge.rebind_session(previous, replacement);
assert_eq!(bridge.owner(&previous), Some((project, agent)));
assert_eq!(bridge.owner(&replacement), Some((project, agent)));
}
#[test]
fn restart_rebind_keeps_a_newer_replacement_sink_and_merges_scrollback_in_order() {
let bridge = ChatBridge::new();
let previous = sid();
let replacement = sid();
let old_sink = Arc::new(Mutex::new(Vec::new()));
let new_sink = Arc::new(Mutex::new(Vec::new()));
let old_generation = bridge.register(previous, capturing_channel(Arc::clone(&old_sink)));
assert!(bridge.send_output(&previous, delta("old")));
bridge.register(replacement, capturing_channel(Arc::clone(&new_sink)));
assert!(bridge.send_output(&replacement, delta("new")));
bridge.rebind_session(previous, replacement);
assert_eq!(
bridge.scrollback(&replacement),
vec![delta("old"), delta("new")]
);
assert!(bridge.send_output(&previous, delta("continued")));
assert_eq!(old_sink.lock().unwrap().len(), 1);
assert_eq!(new_sink.lock().unwrap().len(), 2);
bridge.detach_if(&previous, old_generation);
assert!(
bridge.send_output(&replacement, delta("still attached")),
"the old pump must not detach a channel already attached to the replacement"
);
}
#[test]
fn detach_view_keeps_session_scrollback_and_later_reattach_replays_absent_chunks() {
let bridge = ChatBridge::new();

View File

@ -363,6 +363,7 @@ fn import_chat_attachments_request_response_use_camel_case() {
fn reattach_chat_dto_serialises_camel_case_with_typed_scrollback() {
let dto = ReattachChatDto {
session_id: "sess-1".into(),
conversation_id: Some("conv-1".into()),
scrollback: vec![
ReplyChunk::UserPrompt {
text: "Hi".into(),
@ -380,6 +381,7 @@ fn reattach_chat_dto_serialises_camel_case_with_typed_scrollback() {
v,
json!({
"sessionId": "sess-1",
"conversationId": "conv-1",
"scrollback": [
{ "kind": "userPrompt", "text": "Hi" },
{ "kind": "textDelta", "text": "Hi" },
@ -394,10 +396,12 @@ fn reattach_chat_dto_serialises_camel_case_with_typed_scrollback() {
fn reattach_chat_dto_empty_scrollback_is_empty_array() {
let dto = ReattachChatDto {
session_id: "s".into(),
conversation_id: None,
scrollback: vec![],
};
let v = serde_json::to_value(&dto).unwrap();
assert_eq!(v["scrollback"], json!([]));
assert!(v.get("conversationId").is_none());
}
// ---------------------------------------------------------------------------

View File

@ -1284,7 +1284,8 @@ impl From<OpenTerminalOutput> for TerminalSessionDto {
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ReattachResultDto {
/// The session that was re-attached (echoed back for the frontend).
/// The effective live session that was re-attached. This can differ from the
/// requested id after a structured-runtime restart.
pub session_id: String,
/// The most-recent retained output bytes (scrollback ring buffer).
pub scrollback: Vec<u8>,
@ -3678,8 +3679,13 @@ pub enum ReplyChunk {
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ReattachChatDto {
/// The session that was re-attached (echoed back for the frontend).
/// The effective live session that was re-attached. This can differ from the
/// requested id after a structured-runtime restart.
pub session_id: String,
/// Effective provider conversation id of the live replacement session, when
/// the adapter exposes one.
#[serde(skip_serializing_if = "Option::is_none")]
pub conversation_id: Option<String>,
/// The chunks already streamed for this conversation, in order. The frontend
/// replays them to rebuild the visible turns, then receives subsequent chunks
/// over the freshly-registered channel.

View File

@ -170,6 +170,41 @@ where
.unwrap_or_default()
}
/// Moves the retained stream state from `from` to `to`.
///
/// When `to` already exists, its live sink wins and the source scrollback is
/// prepended to the destination scrollback. This is used when a logical chat
/// survives a provider restart under a fresh transport session id.
pub fn rekey(&self, from: &K, to: K) {
if *from == to {
return;
}
if let Ok(mut map) = self.entries.lock() {
let Some(source) = map.remove(from) else {
return;
};
match map.get_mut(&to) {
Some(destination) => {
// Invalidate both pre-rekey pump generations. In particular,
// the dying source pump must never detach a sink that was
// already attached to the replacement key.
let mut generation = destination.generation.wrapping_add(1);
if generation == source.generation {
generation = generation.wrapping_add(1);
}
destination.generation = generation;
let mut scrollback = source.scrollback;
scrollback.append(&mut destination.scrollback);
destination.scrollback = scrollback;
trim_scrollback(destination, self.max_chunks, self.max_bytes, self.measure);
}
None => {
map.insert(to, source);
}
}
}
}
pub fn unregister(&self, key: &K) {
if let Ok(mut map) = self.entries.lock() {
map.remove(key);

View File

@ -1827,15 +1827,40 @@ async fn ws_attach_chat(
tx: &mpsc::Sender<ServerFrame>,
owned: &Arc<Mutex<Vec<WsAttachment>>>,
) -> Result<(), ErrorDto> {
let sid = parse_session_id(&payload_string(&frame.payload, "sessionId")?)?;
let session = state
.app
.structured_sessions
.session(&sid)
.ok_or_else(|| ErrorDto {
code: "NOT_FOUND".to_owned(),
message: "structured chat session not found".to_owned(),
})?;
let requested_sid = parse_session_id(&payload_string(&frame.payload, "sessionId")?)?;
let (sid, session) =
if let Some(session) = state.app.structured_sessions.session(&requested_sid) {
(requested_sid, session)
} else {
let fallback = frame
.payload
.get("projectId")
.and_then(Value::as_str)
.zip(frame.payload.get("agentId").and_then(Value::as_str));
let (project_id, agent_id) = fallback.ok_or_else(|| ErrorDto {
code: "NOT_FOUND".to_owned(),
message: "structured chat session not found".to_owned(),
})?;
let project_id = parse_project_id(project_id)?;
let agent_id = parse_agent_id(agent_id)?;
let sid = state
.app
.structured_sessions
.session_id_for_agent_in_project(project_id, &agent_id)
.ok_or_else(|| ErrorDto {
code: "NOT_FOUND".to_owned(),
message: "structured chat session not found".to_owned(),
})?;
let session = state
.app
.structured_sessions
.session(&sid)
.ok_or_else(|| ErrorDto {
code: "NOT_FOUND".to_owned(),
message: "structured chat session not found".to_owned(),
})?;
(sid, session)
};
tx.send(ServerFrame::chat_attached(
&frame.id,
&application::StructuredSessionDescriptor {
@ -7721,7 +7746,11 @@ mod tests {
ws_frame(
"chat-attach",
"chat.attach",
json!({ "sessionId": session_id.to_string() }),
json!({
"sessionId": session_id.to_string(),
"projectId": "not-parsed-for-a-live-session",
"agentId": "not-parsed-for-a-live-session",
}),
),
&state,
&tx,
@ -7742,6 +7771,94 @@ mod tests {
assert_eq!(state.ws_chat_bridge.active_sessions(), 1);
}
#[tokio::test]
async fn websocket_chat_attach_recovers_stale_session_from_live_project_agent() {
let state = state();
let stale_session_id = SessionId::new_random();
let live_session_id = SessionId::new_random();
let project_id = ProjectId::from_uuid(Uuid::new_v4());
let agent_id = AgentId::from_uuid(Uuid::new_v4());
let node_id = NodeId::from_uuid(Uuid::new_v4());
let session = Arc::new(FakeChatSession::new(
live_session_id,
vec![ReplyEvent::Final {
content: "unused".to_owned(),
}],
));
state.app.structured_sessions.insert_in_project(
project_id,
Arc::clone(&session) as Arc<dyn AgentSession>,
agent_id,
node_id,
);
let (tx, mut rx) = mpsc::channel(8);
let owned = Arc::new(Mutex::new(Vec::new()));
handle_client_frame(
ws_frame(
"chat-attach-fallback",
"chat.attach",
json!({
"sessionId": stale_session_id.to_string(),
"projectId": project_id.to_string(),
"agentId": agent_id.to_string(),
}),
),
&state,
&tx,
&owned,
)
.await;
let attached = recv_server_frame(&mut rx).await;
assert_eq!(attached.kind, "chat.attached");
assert_eq!(attached.payload["sessionId"], live_session_id.to_string());
assert_eq!(
attached.payload["session"]["sessionId"],
live_session_id.to_string()
);
assert!(state.ws_chat_bridge.send_output(
&live_session_id,
json!({ "kind": "textDelta", "text": "after restart" })
));
let output = recv_server_frame(&mut rx).await;
assert_eq!(output.kind, "chat.output");
assert_eq!(output.payload["sessionId"], live_session_id.to_string());
assert_eq!(output.payload["chunk"]["text"], "after restart");
assert!(matches!(
owned.lock().unwrap().as_slice(),
[WsAttachment::Chat { session, .. }] if *session == live_session_id
));
}
#[tokio::test]
async fn websocket_chat_attach_stale_session_without_fallback_stays_not_found() {
let state = state();
let stale_session_id = SessionId::new_random();
let (tx, mut rx) = mpsc::channel(4);
let owned = Arc::new(Mutex::new(Vec::new()));
handle_client_frame(
ws_frame(
"chat-attach-missing",
"chat.attach",
json!({ "sessionId": stale_session_id.to_string() }),
),
&state,
&tx,
&owned,
)
.await;
let error = recv_server_frame(&mut rx).await;
assert_eq!(error.kind, "error");
assert_eq!(error.reply_to.as_deref(), Some("chat-attach-missing"));
assert_eq!(error.payload["sessionId"], stale_session_id.to_string());
assert_eq!(error.payload["code"], "NOT_FOUND");
assert_eq!(state.ws_chat_bridge.active_sessions(), 0);
assert!(owned.lock().unwrap().is_empty());
}
#[tokio::test]
async fn websocket_chat_send_acknowledges_first_chunk_and_streams_rest() {
let state = state();