From f1d85dfc5a7fbb72aed736c5b496e225640536a4 Mon Sep 17 00:00:00 2001 From: Blomios Date: Sun, 23 Aug 2026 13:34:56 +0200 Subject: [PATCH] =?UTF-8?q?fix(custom-cli):=20busy/progress/tools/chatbox?= =?UTF-8?q?=20bloqu=C3=A9s=20apr=C3=A8s=20le=20premier=20appel=20=E2=80=94?= =?UTF-8?q?=20#245=20(QA=20verte)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backend : StructuredHumanTurnGuard pour partager la comptabilité busy/FIFO entre chat direct humain (agent_send, web chat.send) et asks inter-agents structurés, avec liveness mark_alive et retrait idempotent du tour. Web-server : tap d'événements sur chat.send pour le scrollback temps réel. Frontend : scoping des clés de progression par tour pour éviter la déduplication/ghosting des progress et outils entre appels. --- crates/app-tauri/src/commands.rs | 26 +- crates/backend/src/lib.rs | 85 ++++++ crates/web-server/src/lib.rs | 269 +++++++++++++++++- .../agents/CustomAgentChatView.test.tsx | 234 +++++++++++++++ .../features/agents/CustomAgentChatView.tsx | 172 +++++++---- 5 files changed, 728 insertions(+), 58 deletions(-) diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index d295f97b..486dfd52 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -2564,6 +2564,16 @@ pub async fn agent_send( .iter() .map(|attachment| attachment.filename.clone()) .collect(); + let session_meta = state.structured_sessions.meta_for_session(&sid); + let turn_guard = session_meta.as_ref().map(|(project_id, agent_id, _, _)| { + backend::StructuredHumanTurnGuard::start( + std::sync::Arc::clone(&state.turn_watch_input), + std::sync::Arc::clone(&state.input_mailbox), + *project_id, + *agent_id, + &prompt_for_model, + ) + }); // Retain the human submit in the same live scrollback as model chunks so // `reattach_agent_chat` can repaint the whole in-flight conversation. @@ -2583,9 +2593,15 @@ pub async fn agent_send( let (tap_tx, tap_rx) = std::sync::mpsc::channel(); let tap_bridge = std::sync::Arc::clone(&state.chat_bridge); let tap_sid = sid; + let tap_turn_guard = turn_guard.as_ref().map(|guard| guard.liveness_handle()); let tap_pump = std::thread::spawn(move || { let mut settler = crate::chat::ProgressSettler::default(); for event in tap_rx { + if !matches!(event, domain::ports::ReplyEvent::Final { .. }) { + if let Some(guard) = &tap_turn_guard { + guard.mark_alive(); + } + } let Some(chunk) = crate::chat::chunk_from_event(event) else { continue; }; @@ -2617,11 +2633,16 @@ pub async fn agent_send( // event then feeds the service (détecter→planifier). Without a conversation id the // service must not pretend an automatic resume is armed. let service = std::sync::Arc::clone(&state.session_limit_service); - let meta = state.structured_sessions.meta_for_session(&sid); + let meta = session_meta; std::thread::spawn(move || { let mut terminal_visible = false; let mut settler = crate::chat::ProgressSettler::default(); for event in stream { + if !matches!(event, domain::ports::ReplyEvent::Final { .. }) { + if let Some(guard) = &turn_guard { + guard.mark_alive(); + } + } // Non-terminal, sans contenu chat : un signal de limite alimente le service // (le badge UI vient du bus `AgentRateLimited`, pas du flux chat) puis on // continue à drainer comme pour un battement. @@ -2663,6 +2684,9 @@ pub async fn agent_send( let chunk = crate::chat::unfinished_turn_chunk(); let _ = bridge.send_output_settled(&sid, &mut settler, chunk); } + if let Some(guard) = &turn_guard { + guard.finish(); + } bridge.detach_if(&sid, gen); }); diff --git a/crates/backend/src/lib.rs b/crates/backend/src/lib.rs index 469f46b9..a472e8d0 100644 --- a/crates/backend/src/lib.rs +++ b/crates/backend/src/lib.rs @@ -836,6 +836,87 @@ struct AppAgentResumer { contexts: ResumeContexts, } +/// Guard for a direct human turn sent to a structured agent session. +/// +/// Direct chat adapters (`agent_send`, web `chat.send`) still drive the provider +/// through `AgentSession::send`, but they must share the same busy/FIFO accounting +/// as structured inter-agent asks. This guard starts a silent human turn and +/// retires it on terminal outcome or drop. +pub struct StructuredHumanTurnGuard { + input: Arc, + mailbox: Arc, + agent: domain::RuntimeAgentKey, + ticket_id: domain::mailbox::TicketId, + armed: Arc, + finish_on_drop: bool, +} + +impl StructuredHumanTurnGuard { + /// Starts the mediated busy turn for a direct human structured chat message. + #[must_use] + pub fn start( + input: Arc, + mailbox: Arc, + project_id: ProjectId, + agent_id: AgentId, + prompt: &str, + ) -> Self { + let agent = domain::RuntimeAgentKey::new(project_id, agent_id); + let ticket_id = domain::mailbox::TicketId::new_random(); + let owns_turn = if input.busy_state(agent).is_busy() { + false + } else { + let ticket = domain::mailbox::Ticket::new(ticket_id, "user", prompt); + let _pending = input.enqueue_silent(agent, ticket); + input.busy_state(agent).ticket() == Some(ticket_id) + }; + Self { + input, + mailbox, + agent, + ticket_id, + armed: Arc::new(std::sync::atomic::AtomicBool::new(owns_turn)), + finish_on_drop: true, + } + } + + /// Returns a non-owning handle that can keep the turn alive from relay code. + #[must_use] + pub fn liveness_handle(&self) -> Self { + Self { + input: Arc::clone(&self.input), + mailbox: Arc::clone(&self.mailbox), + agent: self.agent, + ticket_id: self.ticket_id, + armed: Arc::clone(&self.armed), + finish_on_drop: false, + } + } + + /// Records proof of liveness for the running turn. + pub fn mark_alive(&self) { + if self.armed.load(std::sync::atomic::Ordering::Acquire) { + self.input.mark_alive(self.agent); + } + } + + /// Retires the FIFO head and marks the agent idle. Idempotent across clones. + pub fn finish(&self) { + if self.armed.swap(false, std::sync::atomic::Ordering::AcqRel) { + self.mailbox.cancel_head(self.agent, self.ticket_id); + self.input.mark_idle(self.agent); + } + } +} + +impl Drop for StructuredHumanTurnGuard { + fn drop(&mut self) { + if self.finish_on_drop { + self.finish(); + } + } +} + #[async_trait::async_trait] impl AgentResumer for AppAgentResumer { async fn resume( @@ -1384,6 +1465,9 @@ pub struct BackendCore { /// Médiateur d'entrée partagé, capturé pour câbler le callback `turn_ended` du /// [`turn_watcher`](Self::turn_watcher) à l'armement. pub turn_watch_input: Arc, + /// Mailbox paired with [`Self::turn_watch_input`], used by direct structured + /// human turns to retire the synthetic busy/FIFO ticket on terminal outcome. + pub input_mailbox: Arc, /// Handles des watches de fin-de-tour vivants, par clé runtime. (Re)lancer un agent /// dans un projet **remplace** son handle (l'ancien est droppé ⇒ polling arrêté) ; /// fermer/arrêter l'agent le retire. `Mutex` car launch/stop y accèdent concurremment. @@ -3259,6 +3343,7 @@ impl BackendCore { resume_contexts, turn_watcher, turn_watch_input: Arc::clone(&input_mediator), + input_mailbox: Arc::clone(&mailbox), turn_watch_handles: Mutex::new(HashMap::new()), fs_port: Arc::clone(&fs_port), home_dir, diff --git a/crates/web-server/src/lib.rs b/crates/web-server/src/lib.rs index c0278dcb..1dafabae 100644 --- a/crates/web-server/src/lib.rs +++ b/crates/web-server/src/lib.rs @@ -1870,13 +1870,50 @@ async fn ws_send_chat( code: "NOT_FOUND".to_owned(), message: "structured chat session not found".to_owned(), })?; - let stream = session.send(&prompt).await.map_err(|err| ErrorDto { + let session_meta = state.app.structured_sessions.meta_for_session(&sid); + let turn_guard = session_meta.as_ref().map(|(project_id, agent_id, _, _)| { + backend::StructuredHumanTurnGuard::start( + Arc::clone(&state.app.turn_watch_input), + Arc::clone(&state.app.input_mailbox), + *project_id, + *agent_id, + &prompt, + ) + }); + + let (tap_tx, tap_rx) = std::sync::mpsc::channel(); + let tap_out = tx.clone(); + let tap_turn_guard = turn_guard.as_ref().map(|guard| guard.liveness_handle()); + let tap_pump = std::thread::spawn(move || { + for event in tap_rx { + if !matches!(event, ReplyEvent::Final { .. }) { + if let Some(guard) = &tap_turn_guard { + guard.mark_alive(); + } + } + let chunk = reply_event_chunk(event); + if tap_out + .blocking_send(ServerFrame::chat_output(None, sid, chunk)) + .is_err() + { + break; + } + } + }); + let stream_result = session.send_with_tap(&prompt, tap_tx).await; + let _ = tap_pump.join(); + let stream = stream_result.map_err(|err| ErrorDto { code: "CHAT_SEND_FAILED".to_owned(), message: err.to_string(), })?; let mut sent_ack = false; for event in stream { + if !matches!(event, ReplyEvent::Final { .. }) { + if let Some(guard) = &turn_guard { + guard.mark_alive(); + } + } let chunk = reply_event_chunk(event); if !sent_ack { tx.send(ServerFrame::chat_output(Some(&frame.id), sid, chunk)) @@ -1909,6 +1946,10 @@ async fn ws_send_chat( })?; } + if let Some(guard) = &turn_guard { + guard.finish(); + } + Ok(()) } @@ -6820,6 +6861,7 @@ mod tests { id: SessionId, conversation_id: Option, events: Vec, + tap_events: Vec, prompts: Mutex>, } @@ -6829,6 +6871,21 @@ mod tests { id, conversation_id: Some("engine-chat-session".to_owned()), events, + tap_events: Vec::new(), + prompts: Mutex::new(Vec::new()), + } + } + + fn with_tap_events( + id: SessionId, + events: Vec, + tap_events: Vec, + ) -> Self { + Self { + id, + conversation_id: Some("engine-chat-session".to_owned()), + events, + tap_events, prompts: Mutex::new(Vec::new()), } } @@ -6852,6 +6909,92 @@ mod tests { Ok(Box::new(self.events.clone().into_iter())) } + async fn send_with_tap( + &self, + prompt: &str, + tap: std::sync::mpsc::Sender, + ) -> Result { + for event in self.tap_events.clone() { + let _ = tap.send(event); + } + self.send(prompt).await + } + + async fn shutdown(&self) -> Result<(), AgentSessionError> { + Ok(()) + } + } + + struct GatedChatSession { + id: SessionId, + prompts: Mutex>, + started: Mutex>>, + release: Mutex>>, + } + + impl GatedChatSession { + fn new( + id: SessionId, + started: oneshot::Sender<()>, + release: oneshot::Receiver<()>, + ) -> Self { + Self { + id, + prompts: Mutex::new(Vec::new()), + started: Mutex::new(Some(started)), + release: Mutex::new(Some(release)), + } + } + } + + #[async_trait] + impl AgentSession for GatedChatSession { + fn id(&self) -> SessionId { + self.id + } + + fn conversation_id(&self) -> Option { + Some("engine-gated-chat-session".to_owned()) + } + + async fn send(&self, prompt: &str) -> Result { + self.send_with_tap(prompt, std::sync::mpsc::channel().0) + .await + } + + async fn send_with_tap( + &self, + prompt: &str, + _tap: std::sync::mpsc::Sender, + ) -> Result { + self.prompts + .lock() + .expect("prompt trace mutex poisoned") + .push(prompt.to_owned()); + if let Some(started) = self + .started + .lock() + .expect("started gate mutex poisoned") + .take() + { + let _ = started.send(()); + } + let release = self + .release + .lock() + .expect("release gate mutex poisoned") + .take(); + if let Some(release) = release { + let _ = release.await; + } + Ok(Box::new( + vec![ReplyEvent::Final { + content: "released".to_owned(), + }] + .into_iter(), + )) + } + async fn shutdown(&self) -> Result<(), AgentSessionError> { Ok(()) } @@ -7533,6 +7676,130 @@ mod tests { ); } + #[tokio::test] + async fn websocket_chat_send_marks_direct_structured_turn_busy_until_final() { + let state = state(); + let session_id = SessionId::new_random(); + let agent_id = AgentId::from_uuid(Uuid::new_v4()); + let node_id = NodeId::from_uuid(Uuid::new_v4()); + let (started_tx, started_rx) = oneshot::channel(); + let (release_tx, release_rx) = oneshot::channel(); + let session = Arc::new(GatedChatSession::new(session_id, started_tx, release_rx)); + state.app.structured_sessions.insert( + Arc::clone(&session) as Arc, + agent_id, + node_id, + ); + let runtime_key = domain::RuntimeAgentKey::new(ProjectId::from_uuid(Uuid::nil()), agent_id); + let (tx, mut rx) = mpsc::channel(16); + let owned = Arc::new(Mutex::new(Vec::new())); + let frame = ws_frame( + "chat-send-busy", + "chat.send", + json!({ "sessionId": session_id.to_string(), "prompt": "Work" }), + ); + + let send_task = tokio::spawn({ + let state = Arc::clone(&state); + let tx = tx.clone(); + let owned = Arc::clone(&owned); + async move { + handle_client_frame(frame, &state, &tx, &owned).await; + } + }); + started_rx.await.expect("gated session started"); + + assert!( + state.app.turn_watch_input.busy_state(runtime_key).is_busy(), + "direct structured human turn should be busy while send is in flight" + ); + + release_tx.send(()).expect("release gated session"); + let final_chunk = recv_server_frame(&mut rx).await; + send_task.await.expect("chat send task finished"); + + assert_eq!(final_chunk.kind, "chat.output"); + assert_eq!(final_chunk.reply_to.as_deref(), Some("chat-send-busy")); + assert_eq!( + final_chunk.payload["chunk"], + json!({ "kind": "final", "content": "released" }) + ); + assert!( + !state.app.turn_watch_input.busy_state(runtime_key).is_busy(), + "direct structured human turn should be idle after final" + ); + } + + #[tokio::test] + async fn websocket_chat_send_relays_tap_progress_live() { + let state = state(); + let session_id = SessionId::new_random(); + let agent_id = AgentId::from_uuid(Uuid::new_v4()); + let node_id = NodeId::from_uuid(Uuid::new_v4()); + let session = Arc::new(FakeChatSession::with_tap_events( + session_id, + vec![ReplyEvent::Final { + content: "done".to_owned(), + }], + vec![ReplyEvent::Progress { + progress: ReplyProgress::new( + ReplyProgressSource::IdeaLocal, + ReplyProgressKind::Mcp, + ReplyProgressStage::Started, + "idea_ask_agent", + ) + .with_tool_name("idea_ask_agent"), + }], + )); + state.app.structured_sessions.insert( + Arc::clone(&session) as Arc, + agent_id, + node_id, + ); + let (tx, mut rx) = mpsc::channel(16); + let owned = Arc::new(Mutex::new(Vec::new())); + + handle_client_frame( + ws_frame( + "chat-send-progress", + "chat.send", + json!({ "sessionId": session_id.to_string(), "prompt": "Run tool" }), + ), + &state, + &tx, + &owned, + ) + .await; + + let first = recv_server_frame(&mut rx).await; + let second = recv_server_frame(&mut rx).await; + let frames = [first, second]; + let progress = frames + .iter() + .find(|frame| frame.payload["chunk"]["kind"] == "progress") + .expect("tap progress frame was relayed"); + let final_chunk = frames + .iter() + .find(|frame| frame.payload["chunk"]["kind"] == "final") + .expect("final frame was relayed"); + + assert_eq!(progress.kind, "chat.output"); + assert!(progress.reply_to.is_none()); + assert_eq!(progress.payload["sessionId"], session_id.to_string()); + assert_eq!(progress.payload["chunk"]["progress"]["source"], "ideaLocal"); + assert_eq!(progress.payload["chunk"]["progress"]["kind"], "mcp"); + assert_eq!(progress.payload["chunk"]["progress"]["stage"], "started"); + assert_eq!( + progress.payload["chunk"]["progress"]["toolName"], + "idea_ask_agent" + ); + assert_eq!(final_chunk.reply_to.as_deref(), Some("chat-send-progress")); + assert_eq!( + final_chunk.payload["chunk"], + json!({ "kind": "final", "content": "done" }) + ); + } + #[tokio::test] async fn websocket_chat_detach_unregisters_sink_without_killing_session() { let state = state(); diff --git a/frontend/src/features/agents/CustomAgentChatView.test.tsx b/frontend/src/features/agents/CustomAgentChatView.test.tsx index 45f84f31..01acf7b6 100644 --- a/frontend/src/features/agents/CustomAgentChatView.test.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.test.tsx @@ -4,6 +4,7 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { DIProvider } from "@/app/di"; import type { AgentProfile, ReplyChunk } from "@/domain"; +import { MockSystemGateway, MockWorkStateGateway } from "@/adapters/mock"; import { PluginRuntimeProvider } from "@/features/plugins"; import { __resetProjectCliInteractionActivityForTests, @@ -912,6 +913,158 @@ describe("CustomAgentChatView", () => { ).toBe(true); }); + it("keeps the composer locked from canonical work-state busy even without local stream activity", async () => { + const system = new MockSystemGateway(); + const workState = new MockWorkStateGateway(); + workState._setProjectWorkState("project-1", { + agents: [ + { + agentId: "agent-1", + name: "Main", + profileId: "structured", + busy: { state: "busy", ticket: "turn-1", sinceMs: 1 }, + tickets: [], + }, + ], + conversations: [], + }); + const agent = { + launchAgentChat: vi.fn(), + reattachAgentChat: vi.fn(async (sessionId: string) => ({ + sessionId, + scrollback: [], + })), + sendAgentChat: vi.fn(async () => {}), + stopLiveAgent: vi.fn(async () => ({ + agentId: "agent-1", + sessionId: "chat-session-1", + kind: "structured" as const, + })), + cancelAgentChat: vi.fn(async () => {}), + closeAgentChat: vi.fn(async () => {}), + }; + + render( + + + , + ); + + const composer = await screen.findByLabelText(/message CLI custom/); + await waitFor(() => expect(composer).toHaveProperty("disabled", true)); + expect(screen.getByRole("button", { name: "Stop" })).toBeTruthy(); + + workState._setProjectWorkState("project-1", { + agents: [ + { + agentId: "agent-1", + name: "Main", + profileId: "structured", + busy: { state: "idle" }, + tickets: [], + }, + ], + conversations: [], + }); + system.emit({ type: "agentBusyChanged", agentId: "agent-1", busy: false }); + + await waitFor(() => expect(composer).toHaveProperty("disabled", false)); + expect(screen.getByRole("button", { name: "Envoyer" })).toBeTruthy(); + }); + + it("shows the requester as waiting on the delegated target and keeps input locked", async () => { + const system = new MockSystemGateway(); + const workState = new MockWorkStateGateway(); + workState._setProjectWorkState("project-1", { + agents: [ + { + agentId: "main", + name: "Main", + profileId: "structured", + busy: { state: "idle" }, + tickets: [], + }, + { + agentId: "architect", + name: "Architect", + profileId: "structured", + busy: { state: "busy", ticket: "delegated-1", sinceMs: 1 }, + tickets: [ + { + ticketId: "delegated-1", + conversationId: "conversation-2", + position: 0, + status: "inProgress", + source: { kind: "agent", agentId: "main" }, + requesterLabel: "Main", + taskPreview: "Analyse l'architecture", + taskLen: 22, + }, + ], + }, + ], + conversations: [], + }); + const agent = { + launchAgentChat: vi.fn(), + reattachAgentChat: vi.fn(async (sessionId: string) => ({ + sessionId, + scrollback: [], + })), + sendAgentChat: vi.fn(async () => {}), + stopLiveAgent: vi.fn(async () => ({ + agentId: "main", + sessionId: "chat-session-1", + kind: "structured" as const, + })), + cancelAgentChat: vi.fn(async () => {}), + closeAgentChat: vi.fn(async () => {}), + }; + + render( + + + , + ); + + expect(await screen.findByText("Main attend Architect")).toBeTruthy(); + expect(screen.getByLabelText(/message CLI custom/)).toHaveProperty("disabled", true); + }); + it("falls back to a fresh launch when reattach reports NOT_FOUND", async () => { const agent = { launchAgentChat: vi.fn(async () => ({ @@ -1142,6 +1295,87 @@ describe("CustomAgentChatView", () => { expect((await screen.findAllByText("done")).length).toBeGreaterThan(0); }); + it("renders repeated legacy tool progress as fresh activity on each turn", async () => { + const agent = { + launchAgentChat: vi.fn(), + reattachAgentChat: vi.fn(async (sessionId: string) => ({ + sessionId, + scrollback: [], + })), + sendAgentChat: vi.fn(async (_sessionId: string, prompt: string, onChunk) => { + onChunk({ + kind: "progress", + progress: { + source: "ideaLocal", + kind: "mcp", + stage: "started", + label: "idea_context_read", + toolName: "idea_context_read", + text: "Lecture du contexte", + }, + }); + onChunk({ + kind: "progress", + progress: { + source: "ideaLocal", + kind: "mcp", + stage: "completed", + label: "idea_context_read", + toolName: "idea_context_read", + text: "Lecture du contexte", + }, + }); + onChunk({ kind: "final", content: `done ${prompt}` }); + }), + cancelAgentChat: vi.fn(async () => {}), + closeAgentChat: vi.fn(async () => {}), + }; + + render( + null) }, + } as unknown as Gateways} + > + + , + ); + + await waitFor(() => + expect(agent.reattachAgentChat).toHaveBeenCalledWith( + "chat-session-1", + expect.any(Function), + ), + ); + + fireEvent.change(screen.getByLabelText(/message CLI custom/), { + target: { value: "first" }, + }); + fireEvent.click(screen.getByRole("button", { name: "Envoyer" })); + await screen.findByText("done first"); + + fireEvent.change(screen.getByLabelText(/message CLI custom/), { + target: { value: "second" }, + }); + fireEvent.click(screen.getByRole("button", { name: "Envoyer" })); + await screen.findByText("done second"); + + expect(screen.getAllByTestId("chat-tool-progress")).toHaveLength(2); + expect(screen.getAllByText("idea_context_read")).toHaveLength(2); + }); + it("renders available pre-MCP progress before the first MCP card", async () => { let emitChunk: ((chunk: unknown) => void) | null = null; const agent = { diff --git a/frontend/src/features/agents/CustomAgentChatView.tsx b/frontend/src/features/agents/CustomAgentChatView.tsx index c2905929..f09dabca 100644 --- a/frontend/src/features/agents/CustomAgentChatView.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.tsx @@ -73,6 +73,7 @@ type ChatTurn = interface CachedChatView { turns: ChatTurn[]; progressIndex: Map; + progressScope: number; chunkCount: number; sessionId?: string; } @@ -129,7 +130,8 @@ interface ChatErrorState { } interface RequesterDelegationState { - active: boolean; + currentAgentBusy: boolean; + delegationTargetName: string | null; } function describe(e: unknown): string { @@ -144,6 +146,10 @@ function describeOpenFailure(e: unknown): string { return message || "Cause technique non disponible."; } +function workAgentDisplayName(agent: { agentId: string; name: string }): string { + return agent.name.trim() || `Agent ${agent.agentId.slice(0, 8)}`; +} + function structuredSessionLabel( adapter: AgentProfile["structuredAdapter"], ): string | null { @@ -310,14 +316,14 @@ function progressKey(progress: ReplyProgress): string { ].join("\u0000"); } -function progressUpsertKey(progress: ReplyProgress): string { +function progressUpsertKey(progress: ReplyProgress, scope: number): string { const progressId = progress.progressId?.trim(); if (progressId) return progressId; - return `legacy:${progressDisplayKey(progress)}`; + return `legacy:${scope}:${progressDisplayKey(progress)}`; } function turnProgressKey(turn: Extract): string { - return turn.progressKey ?? progressUpsertKey(turn.progress); + return turn.progressKey ?? progressUpsertKey(turn.progress, 0); } function normalizedProgressToolName(progress: ReplyProgress): string | null { @@ -352,11 +358,11 @@ function progressDisplayKey(progress: ReplyProgress): string { return progressKey(progress); } -function legacyActiveProgressKey(progress: ReplyProgress): string | null { +function legacyActiveProgressKey(progress: ReplyProgress, scope: number): string | null { const entity = progressEntityKey(progress); if (!entity) return null; if (progress.stage !== "started" && progress.stage !== "delta") return null; - return `legacy:${entity}\u0000started`; + return `legacy:${scope}:${entity}\u0000started`; } function legacyCompletedDedupKey(progress: ReplyProgress): string | null { @@ -380,17 +386,19 @@ function findLastProgressIndex( function legacyCompletedOccurrenceKey( progress: ReplyProgress, entry: ProgressIndexEntry, + scope: number, ): string { - return `legacy-completed:${progressDisplayKey(progress)}\u0000${entry.firstOccurredAtMs}\u0000${entry.sequence}`; + return `legacy-completed:${scope}:${progressDisplayKey(progress)}\u0000${entry.firstOccurredAtMs}\u0000${entry.sequence}`; } interface ChatViewState { turns: ChatTurn[]; progressIndex: Map; + progressScope: number; } function emptyChatViewState(): ChatViewState { - return { turns: [], progressIndex: new Map() }; + return { turns: [], progressIndex: new Map(), progressScope: 0 }; } function conversationCacheKey( @@ -402,14 +410,14 @@ function conversationCacheKey( } function appendProgress(state: ChatViewState, progress: ReplyProgress): ChatViewState { - let key = progressUpsertKey(progress); + let key = progressUpsertKey(progress, state.progressScope); let existingIndex = state.turns.findIndex( (turn) => turn.role === "progress" && turnProgressKey(turn) === key, ); const activeKey = !progress.progressId && progress.stage === "completed" - ? legacyActiveProgressKey({ ...progress, stage: "started" }) + ? legacyActiveProgressKey({ ...progress, stage: "started" }, state.progressScope) : null; if (activeKey) { const activeIndex = state.turns.findIndex( @@ -425,7 +433,9 @@ function appendProgress(state: ChatViewState, progress: ReplyProgress): ChatView if (completedDedupKey) { existingIndex = findLastProgressIndex( state.turns, - (turn) => legacyCompletedDedupKey(turn.progress) === completedDedupKey, + (turn) => + turnProgressKey(turn).startsWith(`legacy-completed:${state.progressScope}:`) && + legacyCompletedDedupKey(turn.progress) === completedDedupKey, ); const existingTurn = existingIndex >= 0 ? state.turns[existingIndex] : null; if (existingTurn?.role === "progress") { @@ -452,7 +462,7 @@ function appendProgress(state: ChatViewState, progress: ReplyProgress): ChatView const nextEntry = { firstOccurredAtMs, sequence }; let nextKey = key; if (!progress.progressId && progress.stage === "completed" && activeKey && key === activeKey) { - nextKey = legacyCompletedOccurrenceKey(merged, nextEntry); + nextKey = legacyCompletedOccurrenceKey(merged, nextEntry, state.progressScope); nextIndex.delete(activeKey); } nextIndex.set(nextKey, nextEntry); @@ -460,7 +470,7 @@ function appendProgress(state: ChatViewState, progress: ReplyProgress): ChatView if (existingIndex >= 0) { const nextTurns = [...state.turns]; nextTurns[existingIndex] = { role: "progress", progress: merged, progressKey: nextKey }; - return { turns: nextTurns, progressIndex: nextIndex }; + return { ...state, turns: nextTurns, progressIndex: nextIndex }; } const insertAt = state.turns.findIndex((turn) => { @@ -474,9 +484,10 @@ function appendProgress(state: ChatViewState, progress: ReplyProgress): ChatView }); const nextTurn = { role: "progress" as const, progress: merged, progressKey: nextKey }; if (insertAt < 0) { - return { turns: [...state.turns, nextTurn], progressIndex: nextIndex }; + return { ...state, turns: [...state.turns, nextTurn], progressIndex: nextIndex }; } return { + ...state, turns: [ ...state.turns.slice(0, insertAt), nextTurn, @@ -549,8 +560,8 @@ function completeChatViewProgress(state: ChatViewState): ChatViewState { stage: "completed" as const, }; const nextKey = turn.progress.progressId - ? progressUpsertKey(completedProgress) - : legacyCompletedOccurrenceKey(completedProgress, entry); + ? progressUpsertKey(completedProgress, state.progressScope) + : legacyCompletedOccurrenceKey(completedProgress, entry, state.progressScope); if (previousKey !== nextKey) nextIndex.delete(previousKey); nextIndex.set(nextKey, entry); @@ -569,7 +580,7 @@ function completeChatViewProgress(state: ChatViewState): ChatViewState { }; }); - return { turns, progressIndex: nextIndex }; + return { ...state, turns, progressIndex: nextIndex }; } function isRecoverableNoResponseTurn(turn: ChatTurn | undefined): boolean { @@ -751,8 +762,8 @@ function foldChunk( const { turns } = state; if (!isReplyRecord(raw)) { return { + ...state, turns: [...turns, { role: "unknown", text: unknownChunkLabel(raw), receivedAtMs }], - progressIndex: state.progressIndex, }; } const last = turns[turns.length - 1]; @@ -777,26 +788,26 @@ function foldChunk( switch (raw.kind) { case "textDelta": return { + ...state, turns: appendAgentDelta(turns, String(raw.text ?? ""), receivedAtMs), - progressIndex: state.progressIndex, }; case "toolActivity": return { + ...state, turns: [ ...turns, { role: "tool", label: String(raw.label ?? "Activité"), receivedAtMs }, ], - progressIndex: state.progressIndex, }; case "progress": { const progress = normalizeProgress(raw.progress); if (!progress) { return { + ...state, turns: [ ...turns, { role: "unknown", text: unknownChunkLabel(raw), receivedAtMs }, ], - progressIndex: state.progressIndex, }; } return appendProgress(state, progress); @@ -812,7 +823,7 @@ function foldChunk( text: content, receivedAtMs: lastAgentReceivedAtMs(next) ?? receivedAtMs, }); - return { turns: next, progressIndex: completedState.progressIndex }; + return { ...completedState, turns: next }; } case "error": { const completedState = completeChatViewProgress(state); @@ -824,23 +835,24 @@ function foldChunk( text: String(raw.message ?? "Erreur agent"), receivedAtMs: lastAgentReceivedAtMs(next) ?? receivedAtMs, }); - return { turns: next, progressIndex: completedState.progressIndex }; + return { ...completedState, turns: next }; } case "userPrompt": case "UserPrompt": return { + ...state, turns: appendUserPrompt( turns, String(raw.text ?? raw.prompt ?? ""), attachmentLabels(raw.attachments), receivedAtMs, ), - progressIndex: state.progressIndex, + progressScope: state.progressScope + 1, }; default: return { + ...state, turns: [...turns, { role: "unknown", text: unknownChunkLabel(raw), receivedAtMs }], - progressIndex: state.progressIndex, }; } } @@ -887,6 +899,14 @@ export function CustomAgentChatView({ ? (chatConversationViewCache.get(initialConversationCacheKey)?.progressIndex ?? new Map()) : new Map(), ); + const [progressScope, setProgressScope] = useState( + () => + sessionId + ? (chatSessionViewCache.get(sessionId)?.progressScope ?? 0) + : initialConversationCacheKey + ? (chatConversationViewCache.get(initialConversationCacheKey)?.progressScope ?? 0) + : 0, + ); const [expandedProgressKeys, setExpandedProgressKeys] = useState>( () => new Set(), ); @@ -909,6 +929,7 @@ export function CustomAgentChatView({ const chatRootRef = useRef(null); const turnsRef = useRef(turns); const progressIndexRef = useRef(progressIndex); + const progressScopeRef = useRef(progressScope); const sessionRef = useRef(sessionId); sessionRef.current = currentSession; const scrollbackChunkCountRef = useRef( @@ -1001,8 +1022,10 @@ export function CustomAgentChatView({ const applyChatViewState = useCallback((next: ChatViewState) => { turnsRef.current = next.turns; progressIndexRef.current = next.progressIndex; + progressScopeRef.current = next.progressScope; setTurns(next.turns); setProgressIndex(next.progressIndex); + setProgressScope(next.progressScope); }, []); const updateChatViewState = useCallback( @@ -1011,6 +1034,7 @@ export function CustomAgentChatView({ updater({ turns: turnsRef.current, progressIndex: progressIndexRef.current, + progressScope: progressScopeRef.current, }), ); }, @@ -1037,15 +1061,32 @@ export function CustomAgentChatView({ const displayTurns = turns; const sessionLabel = structuredSessionLabel(profile.structuredAdapter); const [requesterDelegation, setRequesterDelegation] = - useState({ active: false }); + useState({ + currentAgentBusy: false, + delegationTargetName: null, + }); + const delegationActive = requesterDelegation.delegationTargetName !== null; + const effectiveBusy = + busy || requesterDelegation.currentAgentBusy || delegationActive; + const delegationFallbackText = requesterDelegation.delegationTargetName + ? `${agentName} attend ${requesterDelegation.delegationTargetName}` + : "Délégation en cours."; const refreshRequesterDelegation = useCallback(async () => { if (!workState) { - setRequesterDelegation({ active: false }); + setRequesterDelegation({ + currentAgentBusy: false, + delegationTargetName: null, + }); return; } try { const state = await workState.getProjectWorkState(projectId); + const current = state.agents.find((row) => row.agentId === agentId); + const currentAgentBusy = Boolean( + current?.busy.state === "busy" || + current?.tickets.some((ticket) => ticket.status === "inProgress"), + ); const target = state.agents.find((row) => row.tickets.some( (ticket) => @@ -1054,13 +1095,15 @@ export function CustomAgentChatView({ ticket.source.agentId === agentId, ), ); - setRequesterDelegation( - target - ? { active: true } - : { active: false }, - ); + setRequesterDelegation({ + currentAgentBusy, + delegationTargetName: target ? workAgentDisplayName(target) : null, + }); } catch { - setRequesterDelegation({ active: false }); + setRequesterDelegation({ + currentAgentBusy: false, + delegationTargetName: null, + }); } }, [agentId, projectId, workState]); @@ -1101,6 +1144,17 @@ export function CustomAgentChatView({ event.type === "agentInboxChanged" || event.type === "agentWakeChanged" ) { + if ( + event.type === "agentBusyChanged" && + event.agentId === agentId && + !event.busy + ) { + if (activeInteractionIdRef.current) { + markProjectCliInteractionFinished(projectId, activeInteractionIdRef.current); + activeInteractionIdRef.current = null; + } + setBusy(false); + } void refreshRequesterDelegation(); } }) @@ -1112,7 +1166,7 @@ export function CustomAgentChatView({ cancelled = true; unsubscribe?.(); }; - }, [refreshRequesterDelegation, system]); + }, [agentId, projectId, refreshRequesterDelegation, system]); useEffect(() => { mountedRef.current = true; @@ -1165,6 +1219,7 @@ export function CustomAgentChatView({ chatSessionViewCache.set(currentSession, { turns, progressIndex, + progressScope, chunkCount: scrollbackChunkCountRef.current, sessionId: currentSession, }); @@ -1176,12 +1231,13 @@ export function CustomAgentChatView({ chatConversationViewCache.set(currentConversationCacheKey, { turns, progressIndex, + progressScope, chunkCount: scrollbackChunkCountRef.current, ...(cachedSessionId ? { sessionId: cachedSessionId } : {}), }); } } - }, [currentConversationCacheKey, currentSession, progressIndex, turns]); + }, [currentConversationCacheKey, currentSession, progressIndex, progressScope, turns]); useEffect(() => { draftSaveSeqRef.current += 1; @@ -1261,7 +1317,7 @@ export function CustomAgentChatView({ !supported || !agent.listSlashCommands || slashPrefix === null || - busy || + effectiveBusy || opening ) { setSlashMenuOpen(false); @@ -1286,7 +1342,7 @@ export function CustomAgentChatView({ setSlashMenuOpen(false); setSlashActiveIndex(0); }); - }, [agent, busy, opening, slashPrefix, supported]); + }, [agent, effectiveBusy, opening, slashPrefix, supported]); const selectSlashCommand = useCallback((command: SlashCommand) => { if (!isSlashCommandAvailable(command)) return; @@ -1492,6 +1548,7 @@ export function CustomAgentChatView({ applyChatViewState({ turns: cached.turns, progressIndex: cached.progressIndex ?? new Map(), + progressScope: cached.progressScope ?? 0, }); } try { @@ -1526,6 +1583,7 @@ export function CustomAgentChatView({ applyChatViewState({ turns: cached.turns, progressIndex: cached.progressIndex ?? new Map(), + progressScope: cached.progressScope ?? 0, }); if (cached.sessionId) { const reattachedSessionId = await reattachStructuredSession(cached.sessionId, { @@ -1588,9 +1646,9 @@ export function CustomAgentChatView({ () => supported && (Boolean(draft.trim()) || attachments.length > 0) && - !busy && + !effectiveBusy && !opening, - [supported, draft, attachments.length, busy, opening], + [supported, draft, attachments.length, effectiveBusy, opening], ); async function openProfileDialog() { @@ -1670,6 +1728,7 @@ export function CustomAgentChatView({ }, ], progressIndex: new Map(), + progressScope: 0, }); setProfileDialog(null); } catch (e) { @@ -1706,8 +1765,8 @@ export function CustomAgentChatView({ if (result.effect.kind === "help") { const helpText = formatSlashCommandHelp(result.effect.commands); updateChatViewState((prev) => ({ + ...prev, turns: [...prev.turns, { role: "agent", text: helpText, receivedAtMs: Date.now() }], - progressIndex: prev.progressIndex, })); return; } @@ -1726,6 +1785,7 @@ export function CustomAgentChatView({ ...result.effect.arguments, ); updateChatViewState((prev) => ({ + ...prev, turns: [ ...prev.turns, { @@ -1734,7 +1794,6 @@ export function CustomAgentChatView({ receivedAtMs: Date.now(), }, ], - progressIndex: prev.progressIndex, })); return; } @@ -1742,8 +1801,8 @@ export function CustomAgentChatView({ const message = `Commande ${invocation.name} échouée: ${describe(e)}`; setError({ message, openFailure: false }); updateChatViewState((prev) => ({ + ...prev, turns: [...prev.turns, { role: "error", text: message, receivedAtMs: Date.now() }], - progressIndex: prev.progressIndex, })); } } @@ -1859,6 +1918,7 @@ export function CustomAgentChatView({ setError(null); const sentAtMs = Date.now(); updateChatViewState((prev) => ({ + ...prev, turns: [ ...prev.turns, { @@ -1868,7 +1928,7 @@ export function CustomAgentChatView({ ...(attachmentLabels.length > 0 ? { attachments: attachmentLabels } : {}), }, ], - progressIndex: prev.progressIndex, + progressScope: prev.progressScope + 1, })); const sendTurn = (sid: string) => attachmentInputs.length > 0 @@ -1900,11 +1960,11 @@ export function CustomAgentChatView({ setBusy(false); setError({ message: describe(recoveryError), openFailure: false }); updateChatViewState((prev) => ({ + ...prev, turns: [ ...prev.turns, { role: "error", text: describe(recoveryError), receivedAtMs: Date.now() }, ], - progressIndex: prev.progressIndex, })); return; } @@ -1913,8 +1973,8 @@ export function CustomAgentChatView({ setBusy(false); setError({ message: describe(e), openFailure: false }); updateChatViewState((prev) => ({ + ...prev, turns: [...prev.turns, { role: "error", text: describe(e), receivedAtMs: Date.now() }], - progressIndex: prev.progressIndex, })); } } @@ -1937,6 +1997,7 @@ export function CustomAgentChatView({ updateChatViewState((prev) => { const completed = completeChatViewProgress(prev); return { + ...completed, turns: [ ...completed.turns, { @@ -1948,7 +2009,6 @@ export function CustomAgentChatView({ receivedAtMs: Date.now(), }, ], - progressIndex: completed.progressIndex, }; }); } catch (e) { @@ -1964,17 +2024,18 @@ export function CustomAgentChatView({ updateChatViewState((prev) => { const completed = completeChatViewProgress(prev); return { + ...completed, turns: [ ...completed.turns, { role: "tool", label: "Session déjà arrêtée.", receivedAtMs: Date.now() }, ], - progressIndex: completed.progressIndex, }; }); return; } setError({ message: describe(e), openFailure: false }); updateChatViewState((prev) => ({ + ...prev, turns: [ ...prev.turns, { @@ -1983,7 +2044,6 @@ export function CustomAgentChatView({ receivedAtMs: Date.now(), }, ], - progressIndex: prev.progressIndex, })); } finally { setStopping(false); @@ -2159,8 +2219,8 @@ export function CustomAgentChatView({
{opening && displayTurns.length === 0 ? ( @@ -2286,7 +2346,7 @@ export function CustomAgentChatView({ )} rows={2} value={draft} - disabled={!supported || opening || busy} + disabled={!supported || opening || effectiveBusy} placeholder="Message à l'agent…" style={{ height: clampComposerHeight(composerHeightPx) }} onChange={(e) => editDraft(e.target.value)} @@ -2343,24 +2403,24 @@ export function CustomAgentChatView({ size="sm" variant="ghost" className="shrink-0 whitespace-nowrap" - disabled={!supported || opening || busy} + disabled={!supported || opening || effectiveBusy} onClick={() => void pickAttachment()} > Joindre