diff --git a/crates/app-tauri/src/commands.rs b/crates/app-tauri/src/commands.rs index b44995dd..2889268e 100644 --- a/crates/app-tauri/src/commands.rs +++ b/crates/app-tauri/src/commands.rs @@ -3323,7 +3323,17 @@ pub async fn agent_send( concrete_terminal = true; if let Some(guard) = turn_guard.as_mut() { match runtime.block_on(guard.transition_after_provider_terminal_async()) { - backend::StructuredHumanTurnTransition::Complete => {} + backend::StructuredHumanTurnTransition::Complete => { + // The terminal chunk is observable by the frontend as + // soon as `send_output_settled` runs below. Retire the + // exact mediated ticket and the structured generation + // first, so a refresh triggered by Final can never read + // the just-finished turn as Busy. Both operations are + // idempotent and fenced against a successor; the common + // epilogue remains as a drop/error backstop. + guard.finish(); + structured.finish_turn(&sid, turn_generation); + } backend::StructuredHumanTurnTransition::Resume { prompt } => { resume = true; resume_prompt = Some(prompt); diff --git a/crates/application/src/agent/structured.rs b/crates/application/src/agent/structured.rs index fcfeea09..049134a9 100644 --- a/crates/application/src/agent/structured.rs +++ b/crates/application/src/agent/structured.rs @@ -521,6 +521,48 @@ mod tests { } } + struct BusyRecordingMediator { + state: Mutex, + calls: Mutex>, + } + + impl BusyRecordingMediator { + fn new(ticket: TicketId) -> Self { + Self { + state: Mutex::new(AgentBusyState::Busy { + ticket, + since_ms: 1, + }), + calls: Mutex::new(Vec::new()), + } + } + } + + impl InputMediator for BusyRecordingMediator { + fn enqueue(&self, _agent: RuntimeAgentKey, _ticket: Ticket) -> PendingReply { + unreachable!("non utilisé par drain_with_readiness") + } + fn preempt(&self, _agent: RuntimeAgentKey) {} + fn mark_idle(&self, _agent: RuntimeAgentKey) { + *self.state.lock().unwrap() = AgentBusyState::Idle; + self.calls.lock().unwrap().push("idle"); + } + fn mark_alive(&self, _agent: RuntimeAgentKey) { + self.calls.lock().unwrap().push("alive"); + } + fn busy_state(&self, _agent: RuntimeAgentKey) -> AgentBusyState { + *self.state.lock().unwrap() + } + fn bind_handle(&self, _agent: RuntimeAgentKey, _handle: PtyHandle) {} + fn bind_handle_with_submit( + &self, + _agent: RuntimeAgentKey, + _handle: PtyHandle, + _submit: SubmitConfig, + ) { + } + } + #[derive(Default)] struct RecordingBus { events: Mutex>, @@ -562,6 +604,32 @@ mod tests { ); } + #[tokio::test] + async fn structured_final_leaves_no_busy_state_behind() { + let ticket = TicketId::from_uuid(uuid::Uuid::from_u128(369)); + let session = FakeSession { + events: vec![ + ReplyEvent::Heartbeat, + ReplyEvent::Final { + content: "terminé".into(), + }, + ], + }; + let mediator = BusyRecordingMediator::new(ticket); + + let out = drain_with_readiness(&session, "go", None, &mediator, key(369)) + .await + .expect("Final termine le tour structuré"); + + assert_eq!(out, "terminé"); + assert_eq!(mediator.busy_state(key(369)), AgentBusyState::Idle); + assert_eq!( + *mediator.calls.lock().unwrap(), + vec!["alive", "idle"], + "le Final doit publier exactement une transition terminale après l'activité" + ); + } + #[tokio::test] async fn drain_publishes_announcement_but_only_final_marks_idle_and_resolves() { let session = FakeSession { diff --git a/crates/web-server/src/lib.rs b/crates/web-server/src/lib.rs index 4aafdae3..b9a9eafb 100644 --- a/crates/web-server/src/lib.rs +++ b/crates/web-server/src/lib.rs @@ -2144,13 +2144,20 @@ async fn ws_send_chat( concrete_terminal = true; if let Some(guard) = turn_guard.as_mut() { match guard.transition_after_provider_terminal_async().await { - backend::StructuredHumanTurnTransition::Complete => {} + backend::StructuredHumanTurnTransition::Complete => { + // The websocket frame below is the client-visible terminal + // boundary. Release the exact mediated ticket first so a + // work-state read triggered by Final/Error cannot still see + // this completed turn as Busy. + guard.finish(); + } backend::StructuredHumanTurnTransition::Resume { prompt } => { resume = true; resume_prompt = Some(prompt); break; } backend::StructuredHumanTurnTransition::Failed(message) => { + guard.finish(); let chunk = reply_event_chunk( ReplyEvent::Error { message }, u64::try_from(state.app.clock.now_millis()).ok(), @@ -2200,12 +2207,13 @@ async fn ws_send_chat( if !resume && !concrete_terminal { if let Some(guard) = turn_guard.as_mut() { match guard.transition_after_provider_terminal_async().await { - backend::StructuredHumanTurnTransition::Complete => {} + backend::StructuredHumanTurnTransition::Complete => guard.finish(), backend::StructuredHumanTurnTransition::Resume { prompt } => { resume = true; resume_prompt = Some(prompt); } backend::StructuredHumanTurnTransition::Failed(message) => { + guard.finish(); let chunk = reply_event_chunk( ReplyEvent::Error { message }, u64::try_from(state.app.clock.now_millis()).ok(), @@ -2252,6 +2260,9 @@ async fn ws_send_chat( } if !sent_ack { + if let Some(guard) = &turn_guard { + guard.finish(); + } tx.send(ServerFrame::chat_output( Some(&frame.id), sid, @@ -8735,7 +8746,7 @@ mod tests { } #[tokio::test] - async fn websocket_chat_send_marks_direct_structured_turn_busy_until_final() { + async fn websocket_chat_send_releases_busy_before_visible_final() { let state = state(); let session_id = SessionId::new_random(); let agent_id = AgentId::from_uuid(Uuid::new_v4()); @@ -8749,7 +8760,17 @@ mod tests { node_id, ); let runtime_key = domain::RuntimeAgentKey::new(ProjectId::from_uuid(Uuid::nil()), agent_id); - let (tx, mut rx) = mpsc::channel(16); + // Keep the outbound channel full while the provider completes. This makes + // publication of the visible Final block deterministically: the backend must + // become Idle while the terminal frame is still unable to enter the channel. + let (tx, mut rx) = mpsc::channel(1); + tx.send(ServerFrame::chat_output( + None, + session_id, + json!({ "kind": "testBarrier" }), + )) + .await + .expect("prefill websocket output barrier"); let owned = Arc::new(Mutex::new(Vec::new())); let frame = ws_frame( "chat-send-busy", @@ -8773,8 +8794,16 @@ mod tests { ); release_tx.send(()).expect("release gated session"); + tokio::time::timeout(std::time::Duration::from_secs(1), async { + while state.app.turn_watch_input.busy_state(runtime_key).is_busy() { + tokio::task::yield_now().await; + } + }) + .await + .expect("busy/ticket accounting must be released while Final publication is still blocked"); + let barrier = recv_server_frame(&mut rx).await; + assert_eq!(barrier.payload["chunk"]["kind"], "testBarrier"); 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")); @@ -8783,8 +8812,9 @@ mod tests { assert!(final_chunk.payload["chunk"]["atMs"].is_number()); assert!( !state.app.turn_watch_input.busy_state(runtime_key).is_busy(), - "direct structured human turn should be idle after final" + "busy/ticket accounting must be released before Final becomes visible to the client" ); + send_task.await.expect("chat send task finished"); } #[tokio::test] diff --git a/frontend/src/features/agents/CustomAgentChatView.test.tsx b/frontend/src/features/agents/CustomAgentChatView.test.tsx index 9f3afc69..207603a6 100644 --- a/frontend/src/features/agents/CustomAgentChatView.test.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.test.tsx @@ -1780,6 +1780,19 @@ describe("CustomAgentChatView", () => { nativeEvent: "turn.completed", }, }); + workState._setProjectWorkState("project-1", { + agents: [ + { + agentId: "agent-1", + name: "Worker", + profileId: "structured", + busy: { state: "idle" }, + tickets: [], + }, + ], + conversations: [], + }); + system.emit({ type: "agentBusyChanged", agentId: "agent-1", busy: false }); onChunk({ kind: "final", content: "Next answer" }); }), stopLiveAgent: vi.fn(), @@ -6324,6 +6337,81 @@ describe("CustomAgentChatView", () => { ); }); + it("waits for canonical work-state idle after Final before draining the next queued message in FIFO order (#369)", async () => { + const harness = createStreamingAgentHarness(); + const system = new MockSystemGateway(); + const workState = new MockWorkStateGateway(); + const stateFor = (busy: boolean) => ({ + agents: [ + { + agentId: "agent-1", + name: "Worker", + profileId: "structured", + busy: busy + ? ({ state: "busy", ticket: "turn-369", sinceMs: 1 } as const) + : ({ state: "idle" } as const), + tickets: [], + }, + ], + conversations: [], + }); + workState._setProjectWorkState("project-1", stateFor(false)); + + render( + + + , + ); + + const composer = await screen.findByLabelText(/Message CLI personnalisée/); + await waitFor(() => expect(screen.getByRole("button", { name: "Envoyer" })).toBeTruthy()); + fireEvent.change(composer, { target: { value: "active request" } }); + fireEvent.click(screen.getByRole("button", { name: "Envoyer" })); + await waitFor(() => expect(harness.sendAgentChat).toHaveBeenCalledTimes(1)); + + fireEvent.change(composer, { target: { value: "second queued" } }); + fireEvent.click(screen.getByRole("button", { name: "Mettre en attente" })); + fireEvent.change(composer, { target: { value: "third queued" } }); + fireEvent.click(screen.getByRole("button", { name: "Mettre en attente" })); + expect(screen.getByRole("button", { name: "En attente · 2" })).toBeTruthy(); + + workState._setProjectWorkState("project-1", stateFor(true)); + act(() => system.emit({ type: "agentBusyChanged", agentId: "agent-1", busy: true })); + await waitFor(() => expect(screen.getByRole("button", { name: "Mettre en attente" })).toBeTruthy()); + + act(() => harness.emit(0, { kind: "final", content: "first final" })); + await screen.findByText("first final"); + await waitFor(() => expect(screen.getByRole("button", { name: "Mettre en attente" })).toBeTruthy()); + expect(harness.sendAgentChat).toHaveBeenCalledTimes(1); + expect(screen.getByRole("button", { name: "En attente · 2" })).toBeTruthy(); + + workState._setProjectWorkState("project-1", stateFor(false)); + act(() => system.emit({ type: "agentBusyChanged", agentId: "agent-1", busy: false })); + + await waitFor(() => expect(harness.sendAgentChat).toHaveBeenCalledTimes(2)); + expect(harness.sendAgentChat).toHaveBeenNthCalledWith( + 2, + "chat-session-1", + "second queued", + expect.any(Function), + ); + expect(screen.getByRole("button", { name: "En attente · 1" })).toBeTruthy(); + }); + it("edits, deletes, restores and clears queued messages while Stop preserves the queue", async () => { const harness = createStreamingAgentHarness(); const cancelAgentChat = vi.fn(async () => ({ diff --git a/frontend/src/features/agents/CustomAgentChatView.tsx b/frontend/src/features/agents/CustomAgentChatView.tsx index fc8b7bf5..a6944dc7 100644 --- a/frontend/src/features/agents/CustomAgentChatView.tsx +++ b/frontend/src/features/agents/CustomAgentChatView.tsx @@ -1105,6 +1105,7 @@ export function CustomAgentChatView({ } | null>(null); const [queueAnnouncement, setQueueAnnouncement] = useState(""); const [queueDrainReady, setQueueDrainReady] = useState(false); + const [terminalWorkStatePending, setTerminalWorkStatePending] = useState(false); const [slashCommands, setSlashCommands] = useState([]); const [slashMenuOpen, setSlashMenuOpen] = useState(false); const [slashActiveIndex, setSlashActiveIndex] = useState(0); @@ -1342,6 +1343,7 @@ export function CustomAgentChatView({ const effectiveBusy = busy || stopping || + terminalWorkStatePending || Boolean(restart && restart.outcome !== "failed") || (!stopTerminalOverride && (requesterDelegation.currentAgentBusy || delegationActive)); @@ -1709,17 +1711,21 @@ export function CustomAgentChatView({ markProjectCliInteractionFinished(projectId, activeInteractionIdRef.current); activeInteractionIdRef.current = null; } - stopTerminalOverrideRef.current = true; - setStopTerminalOverride(true); - setRequesterDelegation({ - currentAgentBusy: false, - delegationTargetName: null, - }); setBusy(false); - void refreshRequesterDelegation(); + if (workState) { + setTerminalWorkStatePending(true); + void refreshRequesterDelegation().finally(() => { + if (mountedRef.current) setTerminalWorkStatePending(false); + }); + } else { + setRequesterDelegation({ + currentAgentBusy: false, + delegationTargetName: null, + }); + } } }, - [projectId, refreshRequesterDelegation, updateChatViewState], + [projectId, refreshRequesterDelegation, updateChatViewState, workState], ); useEffect(() => {