diff --git a/apps/desktop/src/app/desktop-controller.tsx b/apps/desktop/src/app/desktop-controller.tsx index 033cae9b5fa8..9f06facef266 100644 --- a/apps/desktop/src/app/desktop-controller.tsx +++ b/apps/desktop/src/app/desktop-controller.tsx @@ -239,6 +239,7 @@ export function DesktopController() { const { activeSessionIdRef, ensureSessionState, + resetViewSync, runtimeIdByStoredSessionIdRef, selectedStoredSessionIdRef, sessionStateByRuntimeIdRef, @@ -619,6 +620,7 @@ export function DesktopController() { getRouteToken, navigate, requestGateway, + resetViewSync, runtimeIdByStoredSessionIdRef, selectedStoredSessionId, selectedStoredSessionIdRef, diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts index 189ac0c88d6a..77104ba6295d 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts @@ -1,5 +1,5 @@ import type { QueryClient } from '@tanstack/react-query' -import { type MutableRefObject, useCallback } from 'react' +import { type MutableRefObject, useCallback, useRef } from 'react' import { writeAgentTerminalChunk } from '@/app/right-sidebar/terminal/agent-terminal-stream' import { readActiveTerminal } from '@/app/right-sidebar/terminal/buffer' @@ -9,7 +9,7 @@ import { translateNow } from '@/i18n' import { type GatewayEventPayload, textPart } from '@/lib/chat-messages' import { coerceGatewayText, coerceThinkingText, normalizePersonalityValue } from '@/lib/chat-runtime' import { playCompletionSound } from '@/lib/completion-sound' -import { gatewayEventRequiresSessionId } from '@/lib/gateway-events' +import { resolveGatewayEventSessionId } from '@/lib/gateway-events' import { triggerHaptic } from '@/lib/haptics' import { isProviderSetupErrorMessage } from '@/lib/provider-setup-errors' import { reconcileApprovalModeForProfile } from '@/store/approval-mode' @@ -95,16 +95,27 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { upsertToolCall } = deps + const unscopedStreamSessionIdRef = useRef(null) + return useCallback( (event: RpcEvent) => { const payload = event.payload as GatewayEventPayload | undefined const explicitSid = event.session_id || '' - if (!explicitSid && gatewayEventRequiresSessionId(event.type)) { + const route = resolveGatewayEventSessionId({ + activeSessionId: activeSessionIdRef.current, + eventType: event.type, + explicitSessionId: explicitSid, + unscopedStreamSessionId: unscopedStreamSessionIdRef.current + }) + + unscopedStreamSessionIdRef.current = route.nextUnscopedStreamSessionId + + if (route.drop) { return } - const sessionId = explicitSid || activeSessionIdRef.current + const sessionId = route.sessionId const isActiveEvent = !!sessionId && sessionId === activeSessionIdRef.current if (event.type === 'gateway.ready') { diff --git a/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx b/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx index 3b681966cb84..45bab0c79c1d 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx +++ b/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx @@ -74,6 +74,7 @@ function Harness({ getRouteToken: () => 'token', navigate: vi.fn() as never, requestGateway, + resetViewSync: vi.fn(), runtimeIdByStoredSessionIdRef: ref(new Map()), selectedStoredSessionId: null, selectedStoredSessionIdRef: ref(null), @@ -229,6 +230,7 @@ function ResumeHarness({ getRouteToken: () => 'token', navigate: vi.fn() as never, requestGateway, + resetViewSync: vi.fn(), runtimeIdByStoredSessionIdRef: runtimeIdByStoredSessionIdRef ?? ref(new Map()), selectedStoredSessionId: null, selectedStoredSessionIdRef: ref(null), @@ -476,6 +478,7 @@ function BranchHarness({ getRouteToken: () => 'token', navigate: vi.fn() as never, requestGateway, + resetViewSync: vi.fn(), runtimeIdByStoredSessionIdRef: ref(new Map()), selectedStoredSessionId: null, selectedStoredSessionIdRef: ref(null), diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts index eec8db608117..831fba0fe257 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts @@ -75,6 +75,7 @@ interface SessionActionsOptions { getRouteToken: () => string navigate: NavigateFunction requestGateway: (method: string, params?: Record) => Promise + resetViewSync: () => void runtimeIdByStoredSessionIdRef: MutableRefObject> selectedStoredSessionId: string | null selectedStoredSessionIdRef: MutableRefObject @@ -105,6 +106,7 @@ export function useSessionActions({ getRouteToken, navigate, requestGateway, + resetViewSync, runtimeIdByStoredSessionIdRef, selectedStoredSessionId, selectedStoredSessionIdRef, @@ -128,6 +130,7 @@ export function useSessionActions({ ? normalizeNewChatWorkspaceTarget(draftOptions.workspaceTarget) : undefined + resetViewSync() busyRef.current = false setBusy(false) setAwaitingResponse(false) @@ -171,7 +174,7 @@ export function useSessionActions({ // Never clear the composer here — ChatBar's per-thread draft swap owns it. setFreshDraftReady(true) }, - [activeSessionIdRef, busyRef, navigate, selectedStoredSessionIdRef] + [activeSessionIdRef, busyRef, navigate, resetViewSync, selectedStoredSessionIdRef] ) const createBackendSessionForSend = useCallback( @@ -237,6 +240,7 @@ export function useSessionActions({ return null } + resetViewSync() activeSessionIdRef.current = created.session_id selectedStoredSessionIdRef.current = stored ensureSessionState(created.session_id, stored) @@ -285,6 +289,7 @@ export function useSessionActions({ getRouteToken, navigate, requestGateway, + resetViewSync, selectedStoredSessionIdRef, updateSessionState ] @@ -337,6 +342,7 @@ export function useSessionActions({ // resume entry"). setFreshDraftReady(false) clearNotifications() + resetViewSync() setSelectedStoredSessionId(storedSessionId) selectedStoredSessionIdRef.current = storedSessionId // Optimistically clear any prior resume-failure latch for this session: @@ -667,6 +673,7 @@ export function useSessionActions({ busyRef, copy, requestGateway, + resetViewSync, runtimeIdByStoredSessionIdRef, selectedStoredSessionIdRef, sessionStateByRuntimeIdRef, diff --git a/apps/desktop/src/app/session/hooks/use-session-state-cache.ts b/apps/desktop/src/app/session/hooks/use-session-state-cache.ts index 3f8e02c8ca8a..22ce8796f74c 100644 --- a/apps/desktop/src/app/session/hooks/use-session-state-cache.ts +++ b/apps/desktop/src/app/session/hooks/use-session-state-cache.ts @@ -130,6 +130,18 @@ export function useSessionStateCache({ return created }, []) + const resetViewSync = useCallback(() => { + // Drop any RAF-pending transcript stage so a backgrounded turn cannot + // repaint over the chat the user just switched to (#47709 / #47743). + pendingViewStateRef.current = null + viewSessionIdRef.current = null + + if (viewSyncRafRef.current !== null && typeof window !== 'undefined') { + window.cancelAnimationFrame(viewSyncRafRef.current) + viewSyncRafRef.current = null + } + }, []) + const flushPendingViewState = useCallback(() => { const pending = pendingViewStateRef.current pendingViewStateRef.current = null @@ -306,6 +318,7 @@ export function useSessionStateCache({ return { activeSessionIdRef, ensureSessionState, + resetViewSync, runtimeIdByStoredSessionIdRef, selectedStoredSessionIdRef, sessionStateByRuntimeIdRef, diff --git a/apps/desktop/src/lib/gateway-events.test.ts b/apps/desktop/src/lib/gateway-events.test.ts index d51a943611f0..7435d22d6ee4 100644 --- a/apps/desktop/src/lib/gateway-events.test.ts +++ b/apps/desktop/src/lib/gateway-events.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' -import { gatewayEventRequiresSessionId } from './gateway-events' +import { gatewayEventRequiresSessionId, resolveGatewayEventSessionId } from './gateway-events' describe('gateway event routing', () => { it('drops only unscoped subagent events (genuinely background work)', () => { @@ -24,4 +24,75 @@ describe('gateway event routing', () => { expect(gatewayEventRequiresSessionId('session.info')).toBe(false) expect(gatewayEventRequiresSessionId(undefined)).toBe(false) }) + + it('keeps unscoped stream events pinned to the session that started them', () => { + const started = resolveGatewayEventSessionId({ + activeSessionId: 'session-a', + eventType: 'message.start', + explicitSessionId: '', + unscopedStreamSessionId: null + }) + + expect(started).toEqual({ + drop: false, + nextUnscopedStreamSessionId: 'session-a', + sessionId: 'session-a' + }) + + const delta = resolveGatewayEventSessionId({ + activeSessionId: 'session-b', + eventType: 'message.delta', + explicitSessionId: '', + unscopedStreamSessionId: started.nextUnscopedStreamSessionId + }) + + expect(delta).toEqual({ + drop: false, + nextUnscopedStreamSessionId: 'session-a', + sessionId: 'session-a' + }) + + const completed = resolveGatewayEventSessionId({ + activeSessionId: 'session-b', + eventType: 'message.complete', + explicitSessionId: '', + unscopedStreamSessionId: delta.nextUnscopedStreamSessionId + }) + + expect(completed).toEqual({ + drop: false, + nextUnscopedStreamSessionId: null, + sessionId: 'session-a' + }) + }) + + it('routes a new unscoped stream start to the currently active session', () => { + const routed = resolveGatewayEventSessionId({ + activeSessionId: 'session-b', + eventType: 'message.start', + explicitSessionId: '', + unscopedStreamSessionId: 'session-a' + }) + + expect(routed).toEqual({ + drop: false, + nextUnscopedStreamSessionId: 'session-b', + sessionId: 'session-b' + }) + }) + + it('keeps explicit events scoped and clears a matching pinned stream on completion', () => { + const routed = resolveGatewayEventSessionId({ + activeSessionId: 'session-b', + eventType: 'message.complete', + explicitSessionId: 'session-a', + unscopedStreamSessionId: 'session-a' + }) + + expect(routed).toEqual({ + drop: false, + nextUnscopedStreamSessionId: null, + sessionId: 'session-a' + }) + }) }) diff --git a/apps/desktop/src/lib/gateway-events.ts b/apps/desktop/src/lib/gateway-events.ts index 673d1df8c6d6..7871848c091f 100644 --- a/apps/desktop/src/lib/gateway-events.ts +++ b/apps/desktop/src/lib/gateway-events.ts @@ -11,6 +11,34 @@ function asRecord(payload: unknown): Record { return payload && typeof payload === 'object' ? (payload as Record) : {} } +/** + * Unscoped stream events that must stay pinned to the session that received + * ``message.start`` after the user switches chats mid-turn (#47709 / #48281). + * Without this, ``explicitSid || activeSessionId`` reattributes live deltas to + * the newly focused chat. + */ +const UNSCOPED_STREAM_EVENT_TYPES = new Set([ + 'approval.request', + 'browser.progress', + 'clarify.request', + 'error', + 'message.complete', + 'message.delta', + 'message.start', + 'reasoning.available', + 'reasoning.delta', + 'secret.request', + 'status.update', + 'sudo.request', + 'thinking.delta', + 'tool.complete', + 'tool.generating', + 'tool.progress', + 'tool.start' +]) + +const UNSCOPED_STREAM_END_EVENT_TYPES = new Set(['error', 'message.complete']) + /** * Whether an unscoped event (no `session_id`) must be dropped rather than * attributed to the focused chat. @@ -27,6 +55,71 @@ export function gatewayEventRequiresSessionId(eventType: string | undefined): bo return eventType?.startsWith('subagent.') ?? false } +export interface GatewayEventSessionRouteInput { + activeSessionId: null | string + eventType: string | undefined + explicitSessionId: string + unscopedStreamSessionId: null | string +} + +export interface GatewayEventSessionRoute { + drop: boolean + nextUnscopedStreamSessionId: null | string + sessionId: null | string +} + +/** + * Resolve which runtime session owns a gateway event. + * + * Explicit ``session_id`` always wins. Unscoped stream events pin to the + * session that received ``message.start`` so a mid-turn chat switch cannot + * steal live deltas / tool events onto the newly focused transcript. + */ +export function resolveGatewayEventSessionId({ + activeSessionId, + eventType, + explicitSessionId, + unscopedStreamSessionId +}: GatewayEventSessionRouteInput): GatewayEventSessionRoute { + if (explicitSessionId) { + const nextUnscopedStreamSessionId = + eventType && UNSCOPED_STREAM_END_EVENT_TYPES.has(eventType) && explicitSessionId === unscopedStreamSessionId + ? null + : unscopedStreamSessionId + + return { + drop: false, + nextUnscopedStreamSessionId, + sessionId: explicitSessionId + } + } + + if (gatewayEventRequiresSessionId(eventType)) { + return { + drop: true, + nextUnscopedStreamSessionId: unscopedStreamSessionId, + sessionId: null + } + } + + const streamEvent = eventType ? UNSCOPED_STREAM_EVENT_TYPES.has(eventType) : false + const sessionId = + eventType === 'message.start' ? activeSessionId : streamEvent ? unscopedStreamSessionId || activeSessionId : activeSessionId + let nextUnscopedStreamSessionId = unscopedStreamSessionId + + if (eventType === 'message.start' && activeSessionId) { + nextUnscopedStreamSessionId = activeSessionId + } else if (eventType && UNSCOPED_STREAM_END_EVENT_TYPES.has(eventType)) { + nextUnscopedStreamSessionId = null + } + + return { + drop: false, + nextUnscopedStreamSessionId, + sessionId + } +} + export function gatewayEventCompletedFileDiff(event: RpcEventLike): boolean { if (event.type !== 'tool.complete') { return false