From 1b1476de52f15ed77f5daa3d14089a5eeacb53c8 Mon Sep 17 00:00:00 2001 From: Blomios Date: Wed, 9 Sep 2026 18:44:01 +0200 Subject: [PATCH] =?UTF-8?q?fix(chat):=20le=20transcript=20s'affiche=20?= =?UTF-8?q?=C3=A0=20nouveau=20apr=C3=A8s=20red=C3=A9marrage=20automatique?= =?UTF-8?q?=20d'un=20agent=20=E2=80=94=20#295=20(app-tauri=20ChatBridge:?= =?UTF-8?q?=20cartes=20successors/owners,=20rebind=5Fsession=20d=C3=A9plac?= =?UTF-8?q?e=20sink+scrolloback=20du=20logical=20chat=20vers=20la=20sessio?= =?UTF-8?q?n=20de=20remplacement=20et=20canonical=5Fsession=20r=C3=A9sout?= =?UTF-8?q?=20les=20=C3=A9critures=20avec=20l'ancien=20id=20sur=20toutes?= =?UTF-8?q?=20les=20op=C3=A9rations=20register/scrollback/detach/send=5Fou?= =?UTF-8?q?tput;=20backend=20stream:=20ReplayOutputBridge::rekey=20pr?= =?UTF-8?q?=C3=A9pare=20le=20scrollback=20source=20devant=20le=20destinati?= =?UTF-8?q?on=20et=20invalide=20les=20g=C3=A9n=C3=A9rations=20pr=C3=A9-rek?= =?UTF-8?q?ey=20des=20deux=20pumps;=20commands:=20agent=5Fsend=20binde=20l?= =?UTF-8?q?'owner=20stable=20(projectId,agentId)=20au=20tour=20humain=20et?= =?UTF-8?q?=20rebind=20avant=20le=20flux=20de=20continuation=20au=20handof?= =?UTF-8?q?f=20restart;=20reattach=5Fagent=5Fchat=20r=C3=A9sout=20le=20suc?= =?UTF-8?q?cesseur=20du=20bridge=20puis=20fallback=20identit=C3=A9=20stabl?= =?UTF-8?q?e=20projectId/agentId=20explicite=20ou=20owner=20bind=C3=A9,=20?= =?UTF-8?q?renvoie=20sessionId=20effectif=20+=20conversationId,=20NOT=5FFO?= =?UTF-8?q?UND=20sinon;=20web-server=20chat.attach=20m=C3=AAme=20fallback?= =?UTF-8?q?=20stale=E2=86=92live=20par=20project/agent;=20DTO=20ReattachCh?= =?UTF-8?q?atDto/ReattachResultDto=20documentent=20l'id=20effectif=20post-?= =?UTF-8?q?restart;=20frontend:=20ReattachAgentChatOptions,=20conversation?= =?UTF-8?q?Id=20dans=20ChatAttachedPayload,=20CustomAgentChatView=20r?= =?UTF-8?q?=C3=A9attache=20sur=20agentRestartCompleted=20(d=C3=A9dupe=20pa?= =?UTF-8?q?r=20restartId,=20scrollback=20appliqu=C3=A9,=20tours=20restart?= =?UTF-8?q?=20pr=C3=A9serv=C3=A9s),=20retry=20identit=C3=A9=20stable=20sur?= =?UTF-8?q?=20NOT=5FFOUND=20dans=20openOrAttach=20avant=20fallback=20launc?= =?UTF-8?q?h=20frais,=20publication=20sessionId=20effectif=20+=20conversat?= =?UTF-8?q?ionId,=20=C3=A9tat=20vide=20=C2=ABConversation=20li=C3=A9e?= =?UTF-8?q?=C2=BB=20quand=20une=20conversation/run/workstate=20li=C3=A9=20?= =?UTF-8?q?existe;=20QA=20verte=20reconfirm=C3=A9e=20par=20Git:=20git=20di?= =?UTF-8?q?ff=20--check,=20cargo=20fmt=20--all=20--check,=20app-tauri=20ch?= =?UTF-8?q?at=5Fbridge=2034/34=20dont=203=20nouveaux=20restart=5Frebind=20?= =?UTF-8?q?+=20dto=5Fchat=2024/24,=20cargo=20check=20-p=20app-tauri,=20bac?= =?UTF-8?q?kend=2091+7=20ignored=20permissions=20socket,=20web-server=20we?= =?UTF-8?q?bsocket=5Fchat=5Fattach=203/3,=20npm=20test=2077/77=20CustomAge?= =?UTF-8?q?ntChatView+agentGateway,=20npm=20run=20build;=20r=C3=A9serves?= =?UTF-8?q?=20non=20bloquantes:=20warnings=20Vite=20esbuild/oxc=20et=20chu?= =?UTF-8?q?nk=20>500kB=20connus)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/app-tauri/src/chat.rs | 98 +++++++++++- crates/app-tauri/src/commands.rs | 67 ++++++-- crates/app-tauri/tests/chat_bridge.rs | 78 ++++++++++ crates/app-tauri/tests/dto_chat.rs | 4 + crates/backend/src/dto.rs | 10 +- crates/backend/src/stream.rs | 35 +++++ crates/web-server/src/lib.rs | 137 ++++++++++++++-- frontend/src/adapters/agent.ts | 3 + .../src/adapters/http/agentGateway.test.ts | 13 +- frontend/src/adapters/http/frames.ts | 2 + frontend/src/adapters/http/streamGateways.ts | 11 +- .../agents/CustomAgentChatView.test.tsx | 147 ++++++++++++++++++ .../features/agents/CustomAgentChatView.tsx | 87 ++++++++++- frontend/src/ports/index.ts | 11 +- 14 files changed, 666 insertions(+), 37 deletions(-) diff --git a/crates/app-tauri/src/chat.rs b/crates/app-tauri/src/chat.rs index 6170e86d..f4363feb 100644 --- a/crates/app-tauri/src/chat.rs +++ b/crates/app-tauri/src/chat.rs @@ -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, + successors: Mutex>, + owners: Mutex>, } 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(¤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* @@ -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) -> 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 { - 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 diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index c080bfb4..e88b9fc5 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -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, + agent_id: Option, on_reply: Channel, state: State<'_, AppState>, ) -> Result { - 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, }) } diff --git a/crates/app-tauri/tests/chat_bridge.rs b/crates/app-tauri/tests/chat_bridge.rs index 35ee19d9..7ef65da1 100644 --- a/crates/app-tauri/tests/chat_bridge.rs +++ b/crates/app-tauri/tests/chat_bridge.rs @@ -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(); diff --git a/crates/app-tauri/tests/dto_chat.rs b/crates/app-tauri/tests/dto_chat.rs index 23bd463a..d0314d7b 100644 --- a/crates/app-tauri/tests/dto_chat.rs +++ b/crates/app-tauri/tests/dto_chat.rs @@ -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()); } // --------------------------------------------------------------------------- diff --git a/crates/backend/src/dto.rs b/crates/backend/src/dto.rs index 5b942c74..9476a164 100644 --- a/crates/backend/src/dto.rs +++ b/crates/backend/src/dto.rs @@ -1284,7 +1284,8 @@ impl From 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, @@ -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, /// 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. diff --git a/crates/backend/src/stream.rs b/crates/backend/src/stream.rs index c348ea8f..a67cb0b8 100644 --- a/crates/backend/src/stream.rs +++ b/crates/backend/src/stream.rs @@ -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); diff --git a/crates/web-server/src/lib.rs b/crates/web-server/src/lib.rs index b1efa09a..1619be79 100644 --- a/crates/web-server/src/lib.rs +++ b/crates/web-server/src/lib.rs @@ -1827,15 +1827,40 @@ async fn ws_attach_chat( tx: &mpsc::Sender, owned: &Arc>>, ) -> 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, + 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(); diff --git a/frontend/src/adapters/agent.ts b/frontend/src/adapters/agent.ts index 66c71e27..b11b4525 100644 --- a/frontend/src/adapters/agent.ts +++ b/frontend/src/adapters/agent.ts @@ -275,11 +275,14 @@ export class TauriAgentGateway implements AgentGateway { async reattachAgentChat( sessionId: string, onChunk: (chunk: ReplyChunk) => void, + options: { projectId?: string; agentId?: string } = {}, ): Promise { const channel = new Channel(); channel.onmessage = onChunk; return invoke("reattach_agent_chat", { sessionId, + ...(options.projectId ? { projectId: options.projectId } : {}), + ...(options.agentId ? { agentId: options.agentId } : {}), onReply: channel, }); } diff --git a/frontend/src/adapters/http/agentGateway.test.ts b/frontend/src/adapters/http/agentGateway.test.ts index 72a094d0..ef451447 100644 --- a/frontend/src/adapters/http/agentGateway.test.ts +++ b/frontend/src/adapters/http/agentGateway.test.ts @@ -313,21 +313,30 @@ describe("HttpAgentGateway WS round-trip (B6 frames)", () => { freshStart: false, }); - const reattached = gw.reattachAgentChat("chat-1", () => {}); + const reattached = gw.reattachAgentChat("chat-1", () => {}, { + projectId: "proj-1", + agentId: "agent-1", + }); const attach = await replyToLast(socket, 1, (id) => ({ kind: "chat.attached", replyTo: id, payload: { sessionId: "chat-1", + conversationId: "conv-replacement", scrollback: [{ kind: "final", content: "Recovered reply" }], }, })); await expect(reattached).resolves.toEqual({ sessionId: "chat-1", + conversationId: "conv-replacement", scrollback: [{ kind: "final", content: "Recovered reply" }], }); expect(attach.kind).toBe("chat.attach"); - expect(attach.payload).toEqual({ sessionId: "chat-1" }); + expect(attach.payload).toEqual({ + sessionId: "chat-1", + projectId: "proj-1", + agentId: "agent-1", + }); const chunks: unknown[] = []; const sent = gw.sendAgentChat("chat-1", "Hello", (chunk) => chunks.push(chunk), { diff --git a/frontend/src/adapters/http/frames.ts b/frontend/src/adapters/http/frames.ts index 548e3c67..53bc5918 100644 --- a/frontend/src/adapters/http/frames.ts +++ b/frontend/src/adapters/http/frames.ts @@ -152,5 +152,7 @@ export interface ChatAttachedPayload { }; cellKind?: "chat" | "pty"; assignedConversationId?: string; + conversationId?: string; + engineSessionId?: string; scrollback?: ReplyChunk[]; } diff --git a/frontend/src/adapters/http/streamGateways.ts b/frontend/src/adapters/http/streamGateways.ts index 49d7571b..97dcb98b 100644 --- a/frontend/src/adapters/http/streamGateways.ts +++ b/frontend/src/adapters/http/streamGateways.ts @@ -379,8 +379,9 @@ export class HttpAgentGateway implements AgentGateway { async reattachAgentChat( sessionId: string, onChunk: (chunk: ReplyChunk) => void, + options: { projectId?: string; agentId?: string } = {}, ): Promise { - const ack = await this.ws.send("chat.attach", { sessionId }); + const ack = await this.ws.send("chat.attach", { sessionId, ...options }); if (ack.kind !== "chat.attached") { throw { code: "CHAT_ATTACH_FAILED", @@ -392,6 +393,14 @@ export class HttpAgentGateway implements AgentGateway { this.ws.setChatSink(attachedSessionId, onChunk); return { sessionId: attachedSessionId, + ...(payload.conversationId ?? payload.assignedConversationId ?? payload.engineSessionId + ? { + conversationId: + payload.conversationId ?? + payload.assignedConversationId ?? + payload.engineSessionId, + } + : {}), scrollback: payload.scrollback ?? [], }; } diff --git a/frontend/src/features/agents/CustomAgentChatView.test.tsx b/frontend/src/features/agents/CustomAgentChatView.test.tsx index 02920ebc..630609db 100644 --- a/frontend/src/features/agents/CustomAgentChatView.test.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.test.tsx @@ -231,6 +231,143 @@ describe("CustomAgentChatView", () => { expect(screen.getByText("runtime unavailable")).not.toBeNull(); }); + it("reconciles a restarted chat to its replacement session and keeps receiving chunks", async () => { + const system = new MockSystemGateway(); + const liveReceivers: Array<(chunk: ReplyChunk) => void> = []; + const agent = { + launchAgentChat: vi.fn(), + reattachAgentChat: vi + .fn() + .mockImplementationOnce( + async (_sessionId: string, onChunk: (chunk: ReplyChunk) => void) => { + liveReceivers.push(onChunk); + return { + sessionId: "old-session", + conversationId: "old-conversation", + scrollback: [{ kind: "final", content: "Avant redémarrage" }], + }; + }, + ) + .mockImplementationOnce( + async (_sessionId: string, onChunk: (chunk: ReplyChunk) => void) => { + liveReceivers.push(onChunk); + return { + sessionId: "replacement-session", + conversationId: "replacement-conversation", + scrollback: [ + { kind: "final", content: "Avant redémarrage" }, + { kind: "userPrompt", text: "Continue" }, + { kind: "textDelta", text: "Reprise conservée" }, + ], + }; + }, + ), + sendAgentChat: vi.fn(async () => {}), + cancelAgentChat: vi.fn(async () => {}), + closeAgentChat: vi.fn(async () => {}), + }; + const onSessionId = vi.fn(); + const onConversationId = vi.fn(); + + render( + + + + + , + ); + + expect(await screen.findByText("Avant redémarrage")).toBeTruthy(); + act(() => { + system.emit({ + type: "agentRestartRequested", + agentId: "agent-1", + restartId: "restart-295", + reason: "reload plugin", + }); + system.emit({ + type: "agentRestartCompleted", + agentId: "agent-1", + restartId: "restart-295", + }); + }); + + await waitFor(() => expect(agent.reattachAgentChat).toHaveBeenCalledTimes(2)); + expect(agent.reattachAgentChat).toHaveBeenNthCalledWith( + 2, + "old-session", + expect.any(Function), + { projectId: "project-1", agentId: "agent-1" }, + ); + expect(await screen.findByText("Reprise conservée")).toBeTruthy(); + expect(onSessionId).toHaveBeenCalledWith("replacement-session"); + expect(onConversationId).toHaveBeenCalledWith("replacement-conversation"); + + act(() => { + liveReceivers[1]?.({ kind: "final", content: "Chunk après redémarrage" }); + }); + expect(await screen.findByText("Chunk après redémarrage")).toBeTruthy(); + expect(screen.queryByText(/Envoyez un message/)).toBeNull(); + }); + + it("retries a stale session attach with stable project and agent identity", async () => { + const agent = { + launchAgentChat: vi.fn(), + reattachAgentChat: vi + .fn() + .mockRejectedValueOnce({ code: "NOT_FOUND", message: "stale session" }) + .mockResolvedValueOnce({ + sessionId: "replacement-session", + conversationId: "replacement-conversation", + scrollback: [{ kind: "final", content: "Transcript restauré" }], + }), + sendAgentChat: vi.fn(async () => {}), + cancelAgentChat: vi.fn(async () => {}), + closeAgentChat: vi.fn(async () => {}), + }; + const onSessionId = vi.fn(); + + render( + + + , + ); + + expect(await screen.findByText("Transcript restauré")).toBeTruthy(); + expect(agent.reattachAgentChat).toHaveBeenNthCalledWith( + 2, + "stale-session", + expect.any(Function), + { projectId: "project-1", agentId: "agent-1" }, + ); + expect(agent.launchAgentChat).not.toHaveBeenCalled(); + expect(onSessionId).toHaveBeenCalledWith("replacement-session"); + }); + it("shows the technical open failure on mobile with retry and native TUI fallback", async () => { const openNativeTui = vi.fn(); const agent = { @@ -1344,6 +1481,7 @@ describe("CustomAgentChatView", () => { reattachAgentChat: vi .fn() .mockRejectedValueOnce({ code: "NOT_FOUND", message: "structured session gone" }) + .mockRejectedValueOnce({ code: "NOT_FOUND", message: "structured session gone" }) .mockImplementationOnce(async (sessionId: string) => ({ sessionId, scrollback: [], @@ -4829,6 +4967,9 @@ describe("CustomAgentChatView", () => { .mockRejectedValueOnce( "not found: structured session 69b57638-0ac7-4a79-8252-e0436a3265f6", ) + .mockRejectedValueOnce( + "not found: structured session 69b57638-0ac7-4a79-8252-e0436a3265f6", + ) .mockImplementationOnce(async (sessionId: string) => ({ sessionId, scrollback: [], @@ -4869,6 +5010,12 @@ describe("CustomAgentChatView", () => { ); expect(agent.reattachAgentChat).toHaveBeenNthCalledWith( 2, + "69b57638-0ac7-4a79-8252-e0436a3265f6", + expect.any(Function), + { projectId: "project-1", agentId: "agent-1" }, + ); + expect(agent.reattachAgentChat).toHaveBeenNthCalledWith( + 3, "chat-session-2", expect.any(Function), ); diff --git a/frontend/src/features/agents/CustomAgentChatView.tsx b/frontend/src/features/agents/CustomAgentChatView.tsx index c02497fc..d677225a 100644 --- a/frontend/src/features/agents/CustomAgentChatView.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.tsx @@ -1015,6 +1015,7 @@ export function CustomAgentChatView({ const mountedRef = useRef(false); const openOrAttachCountRef = useRef(0); const selfEmittedSessionIdRef = useRef(undefined); + const reconciledRestartIdsRef = useRef(new Set()); const onSessionIdRef = useRef(onSessionId); onSessionIdRef.current = onSessionId; const onConversationIdRef = useRef(onConversationId); @@ -1543,7 +1544,14 @@ export function CustomAgentChatView({ }, []); const reattachStructuredSession = useCallback( - async (sid: string, options: { applyScrollback?: boolean } = {}) => { + async ( + sid: string, + options: { + applyScrollback?: boolean; + stableIdentity?: boolean; + preserveRestartTurns?: boolean; + } = {}, + ) => { if (!agent.reattachAgentChat) throw new Error("Structured reattach unavailable"); console.debug("[ticket149] reattachStructuredSession:start", { timestamp: new Date().toISOString(), @@ -1552,7 +1560,9 @@ export function CustomAgentChatView({ }); let reattached: Awaited>>; try { - reattached = await agent.reattachAgentChat(sid, receive); + reattached = options.stableIdentity + ? await agent.reattachAgentChat(sid, receive, { projectId, agentId }) + : await agent.reattachAgentChat(sid, receive); } catch (e) { console.error("[ticket149] reattachStructuredSession:error", { timestamp: new Date().toISOString(), @@ -1569,20 +1579,62 @@ export function CustomAgentChatView({ }); if (!mountedRef.current) return reattached.sessionId; setCurrentSession(reattached.sessionId); + if (reattached.sessionId !== sid) { + publishSessionId(reattached.sessionId); + } + if (reattached.conversationId) { + onConversationIdRef.current(reattached.conversationId); + } if (options.applyScrollback) { scrollbackChunkCountRef.current = reattached.scrollback.length; - const replayed = reattached.scrollback.reduce( + let replayed = reattached.scrollback.reduce( (state, chunk) => foldChunk(state, chunk), emptyChatViewState(), ); + if (options.preserveRestartTurns) { + const restartTurns = turnsRef.current.filter((turn) => turn.role === "restart"); + if (restartTurns.length > 0) { + replayed = { ...replayed, turns: [...replayed.turns, ...restartTurns] }; + } + } applyChatViewState(replayed); setBusy(turnsIndicateBusy(replayed)); } return reattached.sessionId; }, - [agent, applyChatViewState, receive], + [agent, agentId, applyChatViewState, projectId, publishSessionId, receive], ); + useEffect(() => { + const completed = [...restartEvents] + .reverse() + .find((event) => event.type === "agentRestartCompleted"); + if (!completed || reconciledRestartIdsRef.current.has(completed.restartId)) return; + const sid = sessionRef.current ?? externalSessionId; + if (!sid) return; + + reconciledRestartIdsRef.current.add(completed.restartId); + let cancelled = false; + setOpening(true); + setError(null); + void reattachStructuredSession(sid, { + applyScrollback: true, + stableIdentity: true, + preserveRestartTurns: true, + }) + .catch((e) => { + if (!cancelled) { + setError({ message: describeOpenFailure(e), openFailure: true }); + } + }) + .finally(() => { + if (!cancelled) setOpening(false); + }); + return () => { + cancelled = true; + }; + }, [externalSessionId, reattachStructuredSession, restartEvents]); + const recoverStructuredSession = useCallback( async ( options: { @@ -1749,6 +1801,23 @@ export function CustomAgentChatView({ // the session is gone (closed/never live) and the caller falls back // to a fresh launch — any other error still surfaces to the user. if (!isNotFound(e)) throw e; + try { + const replacementSessionId = await reattachStructuredSession(externalSessionId, { + applyScrollback: true, + stableIdentity: true, + }); + if (cancelled) return; + publishSessionId(replacementSessionId); + console.debug("[ticket295] openOrAttach:stable-reattach:success", { + timestamp: new Date().toISOString(), + execution, + requestedSessionId: externalSessionId, + replacementSessionId, + }); + return; + } catch (stableError) { + if (!isNotFound(stableError)) throw stableError; + } if (!cancelled) { setCurrentSession(null); publishSessionId(null); @@ -1834,6 +1903,12 @@ export function CustomAgentChatView({ !opening, [supported, draft, attachments.length, effectiveBusy, opening], ); + const hasLinkedConversation = Boolean( + conversationId || + requesterDelegation.currentAgentBusy || + restart || + delegationActive, + ); async function openProfileDialog() { const availableProfiles = (await profileGateway.listProfiles()).filter((candidate) => @@ -2412,6 +2487,10 @@ export function CustomAgentChatView({ {externalSessionId ? "Reprise en cours..." : "Ouverture de la session structurée…"} + ) : displayTurns.length === 0 && hasLinkedConversation ? ( +

+ Conversation liée — récupération du transcript en cours. +

) : displayTurns.length === 0 ? (

Envoyez un message pour démarrer la conversation structurée. diff --git a/frontend/src/ports/index.ts b/frontend/src/ports/index.ts index cddc4b47..492d333a 100644 --- a/frontend/src/ports/index.ts +++ b/frontend/src/ports/index.ts @@ -204,12 +204,20 @@ export interface AgentChatHandle { /** Result of re-attaching a custom chat view to a live structured session. */ export interface ReattachAgentChatResult { - /** Stable structured session id. */ + /** Effective live structured session id (may replace the requested id after restart). */ sessionId: string; + /** Effective provider conversation id, when known by the structured adapter. */ + conversationId?: string; /** Retained chunks already streamed for the session. */ scrollback: ReplyChunk[]; } +/** Stable owner identity used to reconcile a stale transport id after restart. */ +export interface ReattachAgentChatOptions { + projectId?: string; + agentId?: string; +} + /** Agents: create, list, read/update context, delete, launch (L6). */ export interface AgentGateway { /** Lists all agents belonging to the given project. */ @@ -324,6 +332,7 @@ export interface AgentGateway { reattachAgentChat?( sessionId: string, onChunk: (chunk: ReplyChunk) => void, + options?: ReattachAgentChatOptions, ): Promise; /** Detaches the current custom chat view transport while keeping the session alive. */ detachAgentChat?(sessionId: string): Promise;