diff --git a/src/crates/interfaces/acp/AGENTS.md b/src/crates/interfaces/acp/AGENTS.md index 7881bcb45c..38fa9ab570 100644 --- a/src/crates/interfaces/acp/AGENTS.md +++ b/src/crates/interfaces/acp/AGENTS.md @@ -41,4 +41,11 @@ Keep these role features additive and do not replace either closure with cargo check -p openbitfun-acp --no-default-features --features client cargo check -p openbitfun-acp --no-default-features --features server cargo test -p openbitfun-acp +cargo test -p openbitfun-acp --no-default-features --features client,openbitfun-core/git --lib client::prompt::tests ``` + +The focused client prompt tests cover protocol errors, retry, cancellation, +partial output, and transport termination with in-memory agent streams. These +fixtures do not require a live provider or a device connection. The explicit +Core `git` feature satisfies the worktree tool dependency in the current client +closure without enabling the server role or `product-full`. diff --git a/src/crates/interfaces/acp/src/client/manager.rs b/src/crates/interfaces/acp/src/client/manager.rs index f674a255b8..92cea0edd4 100644 --- a/src/crates/interfaces/acp/src/client/manager.rs +++ b/src/crates/interfaces/acp/src/client/manager.rs @@ -44,6 +44,7 @@ use super::config::{ AcpClientRequirementProbe, AcpClientStatus, RemoteAcpClientRequirementSnapshot, }; use super::dsh_profile::{ensure_bundled_profile, ensure_bundled_profile_remote}; +use super::prompt::AcpPrompt; use super::remote_capability_store::RemoteAcpCapabilityStore; use super::remote_session::{preferred_resume_strategies, AcpRemoteSessionStrategy}; use super::remote_shell::{remote_user_shell_command, render_remote_env_assignments, shell_escape}; @@ -63,6 +64,7 @@ use super::stream::{ AcpToolCallTracker, }; use super::tool::AcpAgentTool; +use super::transport::{handle_transport_closed, AcpTransport}; const CONFIG_PATH: &str = "acp_clients"; const CLIENT_STARTUP_TIMEOUT_SECS: u64 = 60; @@ -641,6 +643,7 @@ impl AcpClientService { return Err(error); } }; + let transport = AcpTransport::new(transport); *connection.child.lock().await = child; let service = self.clone(); let connection_for_task = connection.clone(); @@ -652,6 +655,10 @@ impl AcpClientService { let result = Client .builder() .name("openbitfun-acp-client") + .on_receive_notification( + handle_transport_closed, + agent_client_protocol::on_receive_notification!(), + ) .on_receive_request( { let service = service.clone(); @@ -1184,8 +1191,8 @@ impl AcpClientService { .active .as_mut() .ok_or_else(|| OpenBitFunError::service("ACP session was not initialized"))?; - active.send_prompt(prompt).map_err(protocol_error)?; - read_turn_to_string(&mut session).await + let mut prompt = AcpPrompt::start(active, prompt); + read_turn_to_string(&mut session, &mut prompt).await }; if let Some(seconds) = timeout_seconds.filter(|seconds| *seconds > 0) { @@ -1235,13 +1242,13 @@ impl AcpClientService { .await?; discard_pending_session_updates_if_needed(&mut session).await; - { + let mut prompt = { let active = session .active .as_mut() .ok_or_else(|| OpenBitFunError::service("ACP session was not initialized"))?; - active.send_prompt(prompt).map_err(protocol_error)?; - } + AcpPrompt::start(active, prompt) + }; let mut round_tracker = AcpStreamRoundTracker::new(); let mut tool_call_tracker = AcpToolCallTracker::new(); @@ -1250,7 +1257,7 @@ impl AcpClientService { let active = session.active.as_mut().ok_or_else(|| { OpenBitFunError::service("ACP session was not initialized") })?; - active.read_update().await.map_err(protocol_error)? + prompt.read_update(active).await.map_err(protocol_error)? }; match message { @@ -2343,7 +2350,10 @@ where Ok(()) } -async fn read_turn_to_string(session: &mut AcpRemoteSession) -> OpenBitFunResult { +async fn read_turn_to_string( + session: &mut AcpRemoteSession, + prompt: &mut AcpPrompt, +) -> OpenBitFunResult { let mut output = String::new(); let mut tool_call_tracker = AcpToolCallTracker::new(); loop { @@ -2352,7 +2362,7 @@ async fn read_turn_to_string(session: &mut AcpRemoteSession) -> OpenBitFunResult .active .as_mut() .ok_or_else(|| OpenBitFunError::service("ACP session was not initialized"))?; - active.read_update().await.map_err(protocol_error)? + prompt.read_update(active).await.map_err(protocol_error)? }; match message { diff --git a/src/crates/interfaces/acp/src/client/mod.rs b/src/crates/interfaces/acp/src/client/mod.rs index 086cb3b845..87fa28db65 100644 --- a/src/crates/interfaces/acp/src/client/mod.rs +++ b/src/crates/interfaces/acp/src/client/mod.rs @@ -2,6 +2,7 @@ mod builtin_clients; mod config; mod dsh_profile; mod manager; +mod prompt; mod remote_capability_store; mod remote_session; mod remote_shell; @@ -11,6 +12,7 @@ mod session_persistence; mod stream; mod tool; mod tool_card_bridge; +mod transport; pub use config::{ AcpClientConfig, AcpClientConfigFile, AcpClientInfo, AcpClientPermissionMode, diff --git a/src/crates/interfaces/acp/src/client/prompt.rs b/src/crates/interfaces/acp/src/client/prompt.rs new file mode 100644 index 0000000000..45bc2667d6 --- /dev/null +++ b/src/crates/interfaces/acp/src/client/prompt.rs @@ -0,0 +1,42 @@ +use std::future::Future; +use std::pin::Pin; + +use agent_client_protocol::schema::{PromptRequest, PromptResponse}; +use agent_client_protocol::{ActiveSession, Agent, Error, SessionMessage}; + +/// Own the prompt response separately from the session notification queue. +/// ACP 0.12's ActiveSession::send_prompt only queues successful stop reasons; +/// its error callback terminates the connection without waking read_update. +pub(super) struct AcpPrompt { + response: Pin> + Send>>, +} + +impl AcpPrompt { + pub(super) fn start(active: &ActiveSession<'_, Agent>, prompt: String) -> Self { + let request = PromptRequest::new(active.session_id().clone(), vec![prompt.into()]); + let response = active.connection().send_request(request).block_task(); + Self { + response: Box::pin(response), + } + } + + pub(super) async fn read_update( + &mut self, + active: &mut ActiveSession<'_, Agent>, + ) -> Result { + tokio::select! { + // Preserve notifications queued before the response, including any + // partial output preceding an error. The response also wakes us + // when the connection drops its pending request sender. + biased; + update = active.read_update() => update, + response = &mut self.response => { + response.map(|response| SessionMessage::StopReason(response.stop_reason)) + } + } + } +} + +#[cfg(test)] +#[path = "prompt/tests.rs"] +mod tests; diff --git a/src/crates/interfaces/acp/src/client/prompt/tests.rs b/src/crates/interfaces/acp/src/client/prompt/tests.rs new file mode 100644 index 0000000000..559b4e2748 --- /dev/null +++ b/src/crates/interfaces/acp/src/client/prompt/tests.rs @@ -0,0 +1,240 @@ +use std::time::Duration; + +use agent_client_protocol::schema::{CancelNotification, NewSessionResponse, StopReason}; +use agent_client_protocol::{ActiveSession, Agent, ByteStreams, Client, Error, SessionMessage}; +use serde_json::{json, Value}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream, ReadHalf, WriteHalf}; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; +use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; + +use super::super::transport::{handle_transport_closed, AcpTransport}; +use super::AcpPrompt; + +const DEADLINE: Duration = Duration::from_secs(2); + +struct Harness { + active: ActiveSession<'static, Agent>, + reader: BufReader>, + writer: WriteHalf, + connection_task: JoinHandle>, +} + +impl Harness { + async fn new() -> Self { + let (client, agent) = tokio::io::duplex(16 * 1024); + let (client_reader, client_writer) = tokio::io::split(client); + let (agent_reader, agent_writer) = tokio::io::split(agent); + let transport = AcpTransport::new(ByteStreams::new( + client_writer.compat_write(), + client_reader.compat(), + )); + let (ready_tx, ready_rx) = oneshot::channel(); + let connection_task = tokio::spawn(async move { + Client + .builder() + .on_receive_notification( + handle_transport_closed, + agent_client_protocol::on_receive_notification!(), + ) + .connect_with(transport, async move |cx| { + let active = + cx.attach_session(NewSessionResponse::new("test-session"), vec![])?; + ready_tx.send(active).map_err(|_| Error::internal_error())?; + std::future::pending::>().await + }) + .await + }); + let active = tokio::time::timeout(DEADLINE, ready_rx) + .await + .expect("connection should start") + .expect("connection should publish a session"); + Self { + active, + reader: BufReader::new(agent_reader), + writer: agent_writer, + connection_task, + } + } + + async fn request(&mut self) -> Value { + let mut line = String::new(); + tokio::time::timeout(DEADLINE, self.reader.read_line(&mut line)) + .await + .expect("request should reach agent") + .expect("agent input should remain open"); + serde_json::from_str(&line).expect("valid ACP JSON-RPC request") + } + + async fn send(&mut self, message: Value) { + let mut bytes = serde_json::to_vec(&message).unwrap(); + bytes.push(b'\n'); + self.writer.write_all(&bytes).await.unwrap(); + } + + async fn next(&mut self, prompt: &mut AcpPrompt) -> Result { + tokio::time::timeout(DEADLINE, prompt.read_update(&mut self.active)) + .await + .expect("prompt must settle without relying on a user-supplied timeout") + } +} + +impl Drop for Harness { + fn drop(&mut self) { + self.connection_task.abort(); + } +} + +#[tokio::test] +async fn prompt_error_is_reported_and_the_same_session_can_retry() { + let mut harness = Harness::new().await; + let mut prompt = AcpPrompt::start(&harness.active, "offline request".into()); + let request = harness.request().await; + assert_eq!(request["method"], "session/prompt"); + assert_eq!(request["params"]["sessionId"], "test-session"); + assert_eq!(request["params"]["prompt"][0]["text"], "offline request"); + harness + .send(json!({ + "jsonrpc": "2.0", + "id": request["id"], + "error": { + "code": -32603, + "message": "Network connection failed", + "data": { "code": "ENETUNREACH" } + } + })) + .await; + let error = harness.next(&mut prompt).await.unwrap_err(); + assert_eq!(error.message, "Network connection failed"); + assert_eq!(error.data.unwrap()["code"], "ENETUNREACH"); + assert!(!harness.connection_task.is_finished()); + + let mut retry = AcpPrompt::start(&harness.active, "retry after reconnect".into()); + let request = harness.request().await; + harness + .send(json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": { "stopReason": "end_turn" } + })) + .await; + assert!(matches!( + harness.next(&mut retry).await.unwrap(), + SessionMessage::StopReason(StopReason::EndTurn) + )); +} + +#[tokio::test] +async fn partial_output_is_preserved_before_a_prompt_error() { + let mut harness = Harness::new().await; + let mut prompt = AcpPrompt::start(&harness.active, "request".into()); + let request = harness.request().await; + harness + .send(json!({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": "test-session", + "update": { + "sessionUpdate": "agent_message_chunk", + "content": { "type": "text", "text": "partial output" } + } + } + })) + .await; + harness + .send(json!({ + "jsonrpc": "2.0", + "id": request["id"], + "error": { "code": -32603, "message": "Connection reset" } + })) + .await; + assert!(matches!( + harness.next(&mut prompt).await.unwrap(), + SessionMessage::SessionMessage(_) + )); + assert_eq!( + harness.next(&mut prompt).await.unwrap_err().message, + "Connection reset" + ); +} + +#[tokio::test] +async fn cancellation_still_returns_the_agent_stop_reason() { + let mut harness = Harness::new().await; + let mut prompt = AcpPrompt::start(&harness.active, "request".into()); + let request = harness.request().await; + harness + .active + .connection() + .send_notification(CancelNotification::new(harness.active.session_id().clone())) + .unwrap(); + assert_eq!(harness.request().await["method"], "session/cancel"); + harness + .send(json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": { "stopReason": "cancelled" } + })) + .await; + assert!(matches!( + harness.next(&mut prompt).await.unwrap(), + SessionMessage::StopReason(StopReason::Cancelled) + )); +} + +#[tokio::test] +async fn agent_output_eof_fails_a_pending_prompt_even_if_its_input_is_open() { + let mut harness = Harness::new().await; + let mut prompt = AcpPrompt::start(&harness.active, "request".into()); + let _ = harness.request().await; + harness.writer.shutdown().await.unwrap(); + assert!(harness.next(&mut prompt).await.is_err()); +} + +#[tokio::test] +async fn stopping_the_connection_wakes_a_pending_prompt() { + let mut harness = Harness::new().await; + let mut prompt = AcpPrompt::start(&harness.active, "request".into()); + let _ = harness.request().await; + harness.connection_task.abort(); + assert!(harness.next(&mut prompt).await.is_err()); +} + +#[tokio::test] +async fn a_final_response_is_preserved_when_the_agent_immediately_closes_output() { + let mut harness = Harness::new().await; + let mut prompt = AcpPrompt::start(&harness.active, "request".into()); + let request = harness.request().await; + harness + .send(json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": { "stopReason": "end_turn" } + })) + .await; + harness.writer.shutdown().await.unwrap(); + assert!(matches!( + harness.next(&mut prompt).await.unwrap(), + SessionMessage::StopReason(StopReason::EndTurn) + )); +} + +#[tokio::test] +async fn an_error_response_is_preserved_when_the_agent_immediately_closes_output() { + let mut harness = Harness::new().await; + let mut prompt = AcpPrompt::start(&harness.active, "request".into()); + let request = harness.request().await; + harness + .send(json!({ + "jsonrpc": "2.0", + "id": request["id"], + "error": { "code": -32603, "message": "Network connection failed" } + })) + .await; + harness.writer.shutdown().await.unwrap(); + assert_eq!( + harness.next(&mut prompt).await.unwrap_err().message, + "Network connection failed" + ); +} diff --git a/src/crates/interfaces/acp/src/client/transport.rs b/src/crates/interfaces/acp/src/client/transport.rs new file mode 100644 index 0000000000..c77ea809ab --- /dev/null +++ b/src/crates/interfaces/acp/src/client/transport.rs @@ -0,0 +1,94 @@ +use agent_client_protocol::{ + Agent, Channel, ConnectTo, ConnectionTo, Error, JsonRpcNotification, Role, +}; +use futures::channel::{mpsc, oneshot}; +use futures::future::BoxFuture; +use futures::{FutureExt, StreamExt}; +use serde::{Deserialize, Serialize}; + +// This marker is injected only into the client's in-memory incoming queue. +// It is never sent to the agent and does not extend the ACP wire contract. +#[derive(Debug, Clone, Serialize, Deserialize, JsonRpcNotification)] +#[notification(method = "_openbitfun/client_transport_closed")] +pub(super) struct AcpTransportClosed { + error: Error, +} + +pub(super) async fn handle_transport_closed( + notification: AcpTransportClosed, + connection: ConnectionTo, +) -> Result<(), Error> { + // The dispatcher has consumed all frames before the marker. Failing a + // connection task now wakes unanswered requests without losing final replies. + connection.spawn(async move { Err(notification.error) }) +} + +/// Make transport termination observable without discarding already-read frames. +/// ACP 0.12 otherwise keeps waiting on other channel senders after input EOF. +pub(super) struct AcpTransport { + inner: T, +} + +impl AcpTransport { + pub(super) fn new(inner: T) -> Self { + Self { inner } + } +} + +impl> ConnectTo for AcpTransport { + async fn connect_to(self, client: impl ConnectTo) -> Result<(), Error> { + let (channel, transport) = >::into_channel_and_future(self); + futures::try_join!(client.connect_to(channel), transport)?; + Ok(()) + } + + fn into_channel_and_future(self) -> (Channel, BoxFuture<'static, Result<(), Error>>) { + let (Channel { mut rx, tx }, transport) = self.inner.into_channel_and_future(); + let (incoming_tx, incoming_rx) = mpsc::unbounded(); + let (termination_tx, mut termination_rx) = oneshot::channel(); + let run_transport = async move { + // Deliver I/O failures through the incoming queue too. Returning an + // error here could terminate dispatch before the last reply is read. + let _ = termination_tx.send(transport.await); + Ok::<(), Error>(()) + }; + let forward_incoming = async move { + while let Some(message) = rx.next().await { + incoming_tx + .unbounded_send(message) + .map_err(Error::into_internal_error)?; + } + + let error = termination_rx + .try_recv() + .ok() + .flatten() + .and_then(Result::err) + .unwrap_or_else(|| Error::internal_error().data("ACP agent output stream closed")); + // SDK channel errors are treated as recoverable parse errors. + // Use a local dispatcher marker to terminate only after every frame, + // including a prompt response immediately followed by EOF. + let marker = serde_json::from_value(serde_json::json!({ + "jsonrpc": "2.0", + "method": "_openbitfun/client_transport_closed", + "params": AcpTransportClosed { error }, + })) + .map_err(Error::into_internal_error)?; + incoming_tx + .unbounded_send(Ok(marker)) + .map_err(Error::into_internal_error)?; + Ok::<(), Error>(()) + }; + let task = async move { + futures::try_join!(run_transport, forward_incoming)?; + Ok(()) + }; + ( + Channel { + rx: incoming_rx, + tx, + }, + task.boxed(), + ) + } +} diff --git a/src/web-ui/src/app/scenes/session/AuxPane.tsx b/src/web-ui/src/app/scenes/session/AuxPane.tsx index c076caee11..fea29e2cd3 100644 --- a/src/web-ui/src/app/scenes/session/AuxPane.tsx +++ b/src/web-ui/src/app/scenes/session/AuxPane.tsx @@ -101,8 +101,7 @@ const AuxPane = forwardRef( const prevWorkspaceIdRef = useRef(undefined); - useEffect(() => { - const next = workspaceId; + const syncAgentCanvasWorkspace = useCallback((next: string | undefined) => { const prev = prevWorkspaceIdRef.current; if (prev === next) return; @@ -112,16 +111,29 @@ const AuxPane = forwardRef( }); switchAgentCanvasWorkspace(prev ?? null, next ?? null); prevWorkspaceIdRef.current = next; - }, [workspaceId]); + }, []); + + useEffect(() => { + syncAgentCanvasWorkspace(workspaceId); + }, [syncAgentCanvasWorkspace, workspaceId]); useEffect(() => { const removeListener = workspaceManager.addEventListener((event) => { + if ( + event.type === 'workspace:switched' + || event.type === 'workspace:active-changed' + ) { + // WorkspaceManager emits these events synchronously while activation is + // still in progress. Swap the canvas before callers can open the target + // session's review tab; the context effect above remains a fallback. + syncAgentCanvasWorkspace(event.workspace?.id); + } if (event.type === 'workspace:closed') { removeAgentCanvasSnapshot(event.workspaceId); } }); return () => removeListener(); - }, []); + }, [syncAgentCanvasWorkspace]); const handleInteraction = useCallback(async (itemId: string, userInput: string) => { log.debug('Panel interaction', { itemId, userInput }); diff --git a/src/web-ui/src/app/scenes/session/AuxPane.workspace-switch.test.tsx b/src/web-ui/src/app/scenes/session/AuxPane.workspace-switch.test.tsx new file mode 100644 index 0000000000..1c05732e9a --- /dev/null +++ b/src/web-ui/src/app/scenes/session/AuxPane.workspace-switch.test.tsx @@ -0,0 +1,123 @@ +/** + * @vitest-environment jsdom + */ + +import React, { act } from 'react'; +import { createRoot, type Root } from 'react-dom/client'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +const mocks = vi.hoisted(() => ({ + workspaceId: 'workspace-a' as string | undefined, + workspaceEventListener: null as ((event: any) => void) | null, + removeWorkspaceEventListener: vi.fn(), + switchAgentCanvasWorkspace: vi.fn(), + removeAgentCanvasSnapshot: vi.fn(), + canvasStore: { + addTab: vi.fn(), + switchToTab: vi.fn(), + findTabByMetadata: vi.fn(() => null), + updateTabContent: vi.fn(), + closeAllTabs: vi.fn(), + primaryGroup: { tabs: [] }, + secondaryGroup: { tabs: [] }, + }, +})); + +vi.mock('../../components/panels/content-canvas', () => ({ + ContentCanvas: () =>
, + useCanvasStore: (selector: (state: typeof mocks.canvasStore) => unknown) => ( + selector(mocks.canvasStore) + ), +})); + +vi.mock('../../components/panels/content-canvas/stores', () => ({ + switchAgentCanvasWorkspace: mocks.switchAgentCanvasWorkspace, + removeAgentCanvasSnapshot: mocks.removeAgentCanvasSnapshot, +})); + +vi.mock('@/infrastructure/contexts/WorkspaceContext', () => ({ + useCurrentWorkspace: () => ({ + workspace: mocks.workspaceId ? { id: mocks.workspaceId } : null, + }), +})); + +vi.mock('@/infrastructure/services/business/workspaceManager', () => ({ + workspaceManager: { + addEventListener: (listener: (event: any) => void) => { + mocks.workspaceEventListener = listener; + return mocks.removeWorkspaceEventListener; + }, + }, +})); + +vi.mock('@/infrastructure/i18n', () => ({ + useI18n: () => ({ t: (key: string) => key }), +})); + +vi.mock('@/shared/utils/logger', () => ({ + createLogger: () => ({ + debug: vi.fn(), + }), +})); + +import AuxPane from './AuxPane'; + +globalThis.IS_REACT_ACT_ENVIRONMENT = true; + +describe('AuxPane workspace canvas switching', () => { + let container: HTMLDivElement; + let root: Root; + + beforeEach(() => { + mocks.workspaceId = 'workspace-a'; + mocks.workspaceEventListener = null; + mocks.removeWorkspaceEventListener.mockReset(); + mocks.switchAgentCanvasWorkspace.mockReset(); + mocks.removeAgentCanvasSnapshot.mockReset(); + container = document.createElement('div'); + document.body.appendChild(container); + root = createRoot(container); + }); + + afterEach(() => { + act(() => root.unmount()); + container.remove(); + }); + + it('swaps canvas snapshots synchronously before the target review tab can open', () => { + act(() => { + root.render(); + }); + + expect(mocks.switchAgentCanvasWorkspace).toHaveBeenCalledWith(null, 'workspace-a'); + mocks.switchAgentCanvasWorkspace.mockClear(); + + const order: string[] = []; + mocks.switchAgentCanvasWorkspace.mockImplementation(() => { + order.push('canvas-swapped'); + }); + + act(() => { + mocks.workspaceEventListener?.({ + type: 'workspace:switched', + workspace: { id: 'workspace-b' }, + }); + order.push('open-target-review'); + }); + + expect(mocks.switchAgentCanvasWorkspace).toHaveBeenCalledWith( + 'workspace-a', + 'workspace-b', + ); + expect(order).toEqual(['canvas-swapped', 'open-target-review']); + + act(() => { + mocks.workspaceEventListener?.({ + type: 'workspace:active-changed', + workspace: { id: 'workspace-b' }, + }); + }); + + expect(mocks.switchAgentCanvasWorkspace).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/web-ui/src/flow_chat/components/btw/BtwSessionPanel.review-action.test.tsx b/src/web-ui/src/flow_chat/components/btw/BtwSessionPanel.review-action.test.tsx index 46f57bfbe3..a0b954a1cb 100644 --- a/src/web-ui/src/flow_chat/components/btw/BtwSessionPanel.review-action.test.tsx +++ b/src/web-ui/src/flow_chat/components/btw/BtwSessionPanel.review-action.test.tsx @@ -24,6 +24,7 @@ const panelMocks = vi.hoisted(() => ({ respondPermission: vi.fn(() => Promise.resolve()), respondPermissionBatch: vi.fn(() => Promise.resolve()), virtualItems: [] as unknown[], + flowChatSubscriber: null as ((state: FlowChatState) => void) | null, })); let flowChatState: FlowChatState; @@ -184,8 +185,15 @@ vi.mock('../../store/FlowChatStore', () => ({ }, flowChatStore: { getState: () => flowChatState, - subscribe: () => () => {}, subscribeSelector: () => () => {}, + subscribe: (listener: (state: FlowChatState) => void) => { + panelMocks.flowChatSubscriber = listener; + return () => { + if (panelMocks.flowChatSubscriber === listener) { + panelMocks.flowChatSubscriber = null; + } + }; + }, }, })); @@ -535,6 +543,7 @@ describe('BtwSessionPanel review action bar integration', () => { panelMocks.respondPermissionBatch.mockReset(); panelMocks.respondPermissionBatch.mockResolvedValue(undefined); panelMocks.virtualItems = []; + panelMocks.flowChatSubscriber = null; container = document.createElement('div'); document.body.appendChild(container); root = createRoot(container); @@ -1607,6 +1616,68 @@ describe('BtwSessionPanel review action bar integration', () => { }); }); + it('does not reapply stale persisted minimize state after a local restore during streaming', async () => { + vi.mocked(loadPersistedReviewState).mockClear(); + vi.mocked(loadPersistedReviewState).mockResolvedValueOnce({ + version: 1, + phase: 'review_running', + completedRemediationIds: [], + minimized: true, + customInstructions: '', + persistedAt: 2, + }); + const runningSession = createRunningDeepReviewSession(); + flowChatState = { + ...flowChatState, + sessions: new Map([ + ['deep-review-child', runningSession], + ['parent-session', flowChatState.sessions.get('parent-session')!], + ]), + } as FlowChatState; + + await act(async () => { + root.render( + , + ); + await Promise.resolve(); + }); + + const restoreButton = container.querySelector( + '.btw-session-panel__minimized-button', + ); + expect(restoreButton).toBeTruthy(); + await act(async () => { + restoreButton?.click(); + }); + expect(useReviewActionBarStore.getState().getSessionState('deep-review-child')?.minimized) + .toBe(false); + + const streamedSession = { + ...runningSession, + lastActiveAt: runningSession.lastActiveAt + 1, + }; + flowChatState = { + ...flowChatState, + sessions: new Map([ + ['deep-review-child', streamedSession], + ['parent-session', flowChatState.sessions.get('parent-session')!], + ]), + } as FlowChatState; + + await act(async () => { + panelMocks.flowChatSubscriber?.(flowChatState); + await Promise.resolve(); + }); + + expect(loadPersistedReviewState).toHaveBeenCalledTimes(1); + expect(useReviewActionBarStore.getState().getSessionState('deep-review-child')?.minimized) + .toBe(false); + }); + it('restores persisted follow-up and review scope only when the child still exists', async () => { vi.mocked(loadPersistedReviewState).mockResolvedValueOnce({ version: 1, diff --git a/src/web-ui/src/flow_chat/components/btw/BtwSessionPanel.tsx b/src/web-ui/src/flow_chat/components/btw/BtwSessionPanel.tsx index 2c89c7db43..f238a745fa 100644 --- a/src/web-ui/src/flow_chat/components/btw/BtwSessionPanel.tsx +++ b/src/web-ui/src/flow_chat/components/btw/BtwSessionPanel.tsx @@ -177,6 +177,8 @@ export const BtwSessionPanel: React.FC = ({ const childSession = childSessionId ? flowChatState.sessions.get(childSessionId) : undefined; const parentSession = parentSessionId ? flowChatState.sessions.get(parentSessionId) : undefined; + const childSessionRef = useRef(childSession); + childSessionRef.current = childSession; const childRelationship = resolveSessionRelationship(childSession); const childKind = childRelationship.kind === 'review' || childRelationship.kind === 'deep_review' || @@ -772,9 +774,15 @@ export const BtwSessionPanel: React.FC = ({ t, ]); - // Restore persisted review action state on mount + const persistedReviewWorkspacePath = childSession + ? sessionProjectWorkspacePath(childSession) + : undefined; + const persistedReviewRemoteConnectionId = childSession?.remoteConnectionId; + const persistedReviewRemoteSshHost = childSession?.remoteSshHost; + + // Restore persisted review action state once for each stable session location. useEffect(() => { - if (!isReviewSession || !childSessionId || !childSession) return; + if (!isReviewSession || !childSessionId || !persistedReviewWorkspacePath) return; const store = useReviewActionBarStore.getState(); const currentActionState = store.getSessionState(childSessionId); @@ -787,20 +795,23 @@ export const BtwSessionPanel: React.FC = ({ // action state is more specific for fix/review recovery than that projection. if (!canReplaceDerivedReviewState && currentActionState && currentActionState.phase !== 'idle') return; - const workspacePath = sessionProjectWorkspacePath(childSession); - if (!workspacePath) return; - let cancelled = false; loadPersistedReviewState( childSessionId, - workspacePath, - childSession.remoteConnectionId, - childSession.remoteSshHost, + persistedReviewWorkspacePath, + persistedReviewRemoteConnectionId, + persistedReviewRemoteSshHost, ).then((persisted: ReviewActionPersistedState | null) => { - if (cancelled || !persisted) return; + const latestChildSession = childSessionRef.current; + if (cancelled || !persisted || !latestChildSession) return; + if ( + sessionProjectWorkspacePath(latestChildSession) !== persistedReviewWorkspacePath + || latestChildSession.remoteConnectionId !== persistedReviewRemoteConnectionId + || latestChildSession.remoteSshHost !== persistedReviewRemoteSshHost + ) return; - const latestReviewData = findLatestCodeReviewResult(childSession) as DeepReviewActionData | null; + const latestReviewData = findLatestCodeReviewResult(latestChildSession) as DeepReviewActionData | null; const reviewMode: ReviewActionMode = isDeepReview ? 'deep' : 'standard'; // Detect fix interruption @@ -812,7 +823,7 @@ export const BtwSessionPanel: React.FC = ({ persisted.remediationScopeRequiresWorkspaceFallback ?? false; if (persisted.phase === 'fix_running') { - const lastTurn = childSession.dialogTurns[childSession.dialogTurns.length - 1]; + const lastTurn = latestChildSession.dialogTurns[latestChildSession.dialogTurns.length - 1]; const isStillRunning = isActiveReviewTurnStatus(lastTurn?.status); if (!isStillRunning) { @@ -829,16 +840,16 @@ export const BtwSessionPanel: React.FC = ({ ...new Set([ ...remediationModifiedFilePaths, ...collectModifiedFilePathsFromTurns( - childSession.dialogTurns, + latestChildSession.dialogTurns, fixingBaselineTurnId, - childSession.workspacePath, + latestChildSession.workspacePath, ), ]), ]; remediationScopeRequiresWorkspaceFallback = remediationScopeRequiresWorkspaceFallback || hasOpaqueWorkspaceMutationRisk( - childSession.dialogTurns, + latestChildSession.dialogTurns, fixingBaselineTurnId, ); } @@ -904,7 +915,15 @@ export const BtwSessionPanel: React.FC = ({ return () => { cancelled = true; }; - }, [childSession, childSessionId, parentSessionId, isReviewSession, isDeepReview]); + }, [ + childSessionId, + parentSessionId, + isReviewSession, + isDeepReview, + persistedReviewWorkspacePath, + persistedReviewRemoteConnectionId, + persistedReviewRemoteSshHost, + ]); // Observe action bar height to adjust body padding dynamically useEffect(() => { diff --git a/src/web-ui/src/flow_chat/components/btw/DeepReviewActionBar.test.tsx b/src/web-ui/src/flow_chat/components/btw/DeepReviewActionBar.test.tsx index 31bfab13ce..81627816ce 100644 --- a/src/web-ui/src/flow_chat/components/btw/DeepReviewActionBar.test.tsx +++ b/src/web-ui/src/flow_chat/components/btw/DeepReviewActionBar.test.tsx @@ -1166,6 +1166,43 @@ describeWithJsdom('DeepReviewActionBar', () => { expect(prompt).not.toContain('Recommended option 2: Staged path'); }); + it('toggles the Other group from partial selection and excludes completed items', async () => { + useReviewActionBarStore.getState().showActionBar({ + childSessionId: 'child-session', + parentSessionId: 'parent-session', + reviewData: { + summary: { recommended_action: 'request_changes' }, + remediation_plan: ['Completed fix', 'Fix issue 1', 'Fix issue 2'], + }, + phase: 'review_completed', + completedRemediationIds: new Set(['remediation-0']), + }); + useReviewActionBarStore.getState().setSelectedRemediationIds(new Set(['remediation-1'])); + + await act(async () => { + root.render(); + }); + + const groupCheckbox = container.querySelector( + '.deep-review-action-bar__remediation-group-header input[type="checkbox"]', + )!; + expect(groupCheckbox.getAttribute('aria-checked')).toBe('mixed'); + + await act(async () => { groupCheckbox.click(); }); + expect([...useReviewActionBarStore.getState().selectedRemediationIds].sort()) + .toEqual(['remediation-1', 'remediation-2']); + expect(groupCheckbox.checked).toBe(true); + + await act(async () => { groupCheckbox.click(); }); + expect(useReviewActionBarStore.getState().selectedRemediationIds.size).toBe(0); + expect(groupCheckbox.checked).toBe(false); + + await act(async () => { groupCheckbox.click(); }); + expect([...useReviewActionBarStore.getState().selectedRemediationIds].sort()) + .toEqual(['remediation-1', 'remediation-2']); + expect(groupCheckbox.checked).toBe(true); + }); + it('marks completed remediation items when fix completes', async () => { const store = useReviewActionBarStore.getState(); store.showActionBar({ diff --git a/src/web-ui/src/flow_chat/deep-review/action-bar/DeepReviewActionBar.tsx b/src/web-ui/src/flow_chat/deep-review/action-bar/DeepReviewActionBar.tsx index 0903dedf8b..5d37a18d74 100644 --- a/src/web-ui/src/flow_chat/deep-review/action-bar/DeepReviewActionBar.tsx +++ b/src/web-ui/src/flow_chat/deep-review/action-bar/DeepReviewActionBar.tsx @@ -443,9 +443,8 @@ export const ReviewActionBar: React.FC = ({ childSessionId store.toggleAllRemediation(childSessionId ?? undefined); }, [childSessionId, store]); - const handleToggleGroup = useCallback((groupId: string) => { - if (groupId === 'ungrouped') return; - store.toggleGroupRemediation(groupId as RemediationGroupId, childSessionId ?? undefined); + const handleToggleGroup = useCallback((groupId: RemediationGroupId | 'ungrouped') => { + store.toggleGroupRemediation(groupId, childSessionId ?? undefined); }, [childSessionId, store]); const handleToggleDecisionExpansion = useCallback((id: string) => { diff --git a/src/web-ui/src/flow_chat/deep-review/action-bar/RemediationSelectionPanel.tsx b/src/web-ui/src/flow_chat/deep-review/action-bar/RemediationSelectionPanel.tsx index 17362ae652..b2fc7bb909 100644 --- a/src/web-ui/src/flow_chat/deep-review/action-bar/RemediationSelectionPanel.tsx +++ b/src/web-ui/src/flow_chat/deep-review/action-bar/RemediationSelectionPanel.tsx @@ -19,7 +19,7 @@ interface RemediationSelectionPanelProps { selectionDisabled?: boolean; onToggleRemediation: (id: string) => void; onToggleAll: () => void; - onToggleGroup: (groupId: string) => void; + onToggleGroup: (groupId: RemediationGroupId | 'ungrouped') => void; onToggleList: () => void; onToggleDecisionExpansion: (id: string) => void; onSetDecisionSelection: (id: string, optionIndex: number) => void; @@ -78,7 +78,7 @@ export const RemediationSelectionPanel: React.FC }, [remediationItems]); const groupOrder = useMemo(() => { - const ordered: string[] = []; + const ordered: Array = []; for (const gid of REMEDIATION_GROUP_ORDER) { if (groupedItems[gid]?.length) ordered.push(gid); } diff --git a/src/web-ui/src/flow_chat/store/deepReviewActionBarStore.ts b/src/web-ui/src/flow_chat/store/deepReviewActionBarStore.ts index 0aaab880dd..ac219d4d84 100644 --- a/src/web-ui/src/flow_chat/store/deepReviewActionBarStore.ts +++ b/src/web-ui/src/flow_chat/store/deepReviewActionBarStore.ts @@ -191,7 +191,7 @@ export interface ReviewActionBarState extends ReviewActionBarData { updatePhase: (phase: ReviewActionPhase, errorMessage?: string | null, childSessionId?: string) => void; toggleRemediation: (id: string, childSessionId?: string) => void; toggleAllRemediation: (childSessionId?: string) => void; - toggleGroupRemediation: (groupId: RemediationGroupId, childSessionId?: string) => void; + toggleGroupRemediation: (groupId: RemediationGroupId | 'ungrouped', childSessionId?: string) => void; setActiveAction: ( action: 'fix' | 'fix-review' | 'review' | 'resume' | 'retry' | null, options?: { baselineTurnId?: string | null },