From f2702326c6b6ec44fdab764e5f4c2af1e3ba8b18 Mon Sep 17 00:00:00 2001 From: guantw Date: Tue, 8 Sep 2026 15:23:43 +0800 Subject: [PATCH 1/5] fix(review): switch review canvas with the active workspace Swap agent canvas snapshots synchronously during workspace activation before opening the target review tab. Deduplicate repeated workspace notifications and cover the ordering with a regression test. (cherry picked from commit 80da126e2984b5c92f6ca9707d3ae6894f892cce) --- src/web-ui/src/app/scenes/session/AuxPane.tsx | 20 ++- .../session/AuxPane.workspace-switch.test.tsx | 123 ++++++++++++++++++ 2 files changed, 139 insertions(+), 4 deletions(-) create mode 100644 src/web-ui/src/app/scenes/session/AuxPane.workspace-switch.test.tsx 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); + }); +}); From c7d72be3f43bb355c84caa239909052988de8be7 Mon Sep 17 00:00:00 2001 From: guantw Date: Tue, 8 Sep 2026 15:23:43 +0800 Subject: [PATCH 2/5] fix(review): preserve action bar toggles during streaming Restore persisted review actions only when the session location changes, and use the latest session data when the load completes. Keep local expand actions from being overwritten by streaming updates and cover the behavior with a regression test. (cherry picked from commit 6acfe70b6f685e7e3d41a5421b7ec69663b62c23) --- .../BtwSessionPanel.review-action.test.tsx | 73 ++++++++++++++++++- .../components/btw/BtwSessionPanel.tsx | 49 +++++++++---- 2 files changed, 106 insertions(+), 16 deletions(-) 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(() => { From bf41da2389a3bc2e691a5aaed1936556cd84ba70 Mon Sep 17 00:00:00 2001 From: guantw Date: Tue, 8 Sep 2026 17:27:38 +0800 Subject: [PATCH 3/5] fix(review): enable Other group checkbox selection Allow the ungrouped remediation checkbox to reach the group selection store and include ungrouped in the selection callback types. Cover partial selection, select-all, deselection, and exclusion of completed items in the review action bar regression test. (cherry picked from commit 9b6c945e9c63b82f8f8a7e3babaf4770636e03f3) --- .../btw/DeepReviewActionBar.test.tsx | 37 +++++++++++++++++++ .../action-bar/DeepReviewActionBar.tsx | 5 +-- .../action-bar/RemediationSelectionPanel.tsx | 4 +- .../store/deepReviewActionBarStore.ts | 2 +- 4 files changed, 42 insertions(+), 6 deletions(-) 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 }, From 6d7c7cb2f391422afd00bc339420954fd68b3e30 Mon Sep 17 00:00:00 2001 From: guantw Date: Tue, 8 Sep 2026 20:16:06 +0800 Subject: [PATCH 4/5] fix(acp): propagate prompt errors and transport termination Track prompt responses separately from session updates so protocol errors settle active turns and allow retries. Surface transport termination after queued frames are dispatched, preserving partial output and final responses. Add focused client regression tests for errors, retry, cancellation, and connection closure, and document the test entry point. (cherry picked from commit 05c234b2a5aaca344f2dfdd5b43c4ed2342863f2) --- src/crates/interfaces/acp/AGENTS.md | 7 + .../interfaces/acp/src/client/manager.rs | 26 +- src/crates/interfaces/acp/src/client/mod.rs | 2 + .../interfaces/acp/src/client/prompt.rs | 42 +++ .../interfaces/acp/src/client/prompt/tests.rs | 240 ++++++++++++++++++ .../interfaces/acp/src/client/transport.rs | 94 +++++++ 6 files changed, 403 insertions(+), 8 deletions(-) create mode 100644 src/crates/interfaces/acp/src/client/prompt.rs create mode 100644 src/crates/interfaces/acp/src/client/prompt/tests.rs create mode 100644 src/crates/interfaces/acp/src/client/transport.rs 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..46fc639da8 --- /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 = "_bitfun/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": "_bitfun/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(), + ) + } +} From 8fe4da31ede3daf1c0ff21937dd47c08effba30b Mon Sep 17 00:00:00 2001 From: guantw Date: Wed, 9 Sep 2026 00:02:34 +0800 Subject: [PATCH 5/5] fix(acp): use canonical transport notification identity Rename the client-local transport closure notification to the OpenBitFun prefix in both its declaration and injected payload. Keep transport termination dispatch consistent and satisfy the product identity audit. --- src/crates/interfaces/acp/src/client/transport.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/crates/interfaces/acp/src/client/transport.rs b/src/crates/interfaces/acp/src/client/transport.rs index 46fc639da8..c77ea809ab 100644 --- a/src/crates/interfaces/acp/src/client/transport.rs +++ b/src/crates/interfaces/acp/src/client/transport.rs @@ -9,7 +9,7 @@ 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 = "_bitfun/client_transport_closed")] +#[notification(method = "_openbitfun/client_transport_closed")] pub(super) struct AcpTransportClosed { error: Error, } @@ -70,7 +70,7 @@ impl> ConnectTo for AcpTransport { // including a prompt response immediately followed by EOF. let marker = serde_json::from_value(serde_json::json!({ "jsonrpc": "2.0", - "method": "_bitfun/client_transport_closed", + "method": "_openbitfun/client_transport_closed", "params": AcpTransportClosed { error }, })) .map_err(Error::into_internal_error)?;