Merge pull request #63948 from NousResearch/bb/salvage-48281-stream-pin

fix(desktop): pin unscoped streams + clear view sync on switch (supersedes #48281)

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,

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<string | null>(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') {

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<string, string>()),
     selectedStoredSessionId: null,
     selectedStoredSessionIdRef: ref<string | null>(null),
@@ -229,6 +230,7 @@ function ResumeHarness({
     getRouteToken: () => 'token',
     navigate: vi.fn() as never,
     requestGateway,
+    resetViewSync: vi.fn(),
     runtimeIdByStoredSessionIdRef: runtimeIdByStoredSessionIdRef ?? ref(new Map<string, string>()),
     selectedStoredSessionId: null,
     selectedStoredSessionIdRef: ref<string | null>(null),
@@ -476,6 +478,7 @@ function BranchHarness({
     getRouteToken: () => 'token',
     navigate: vi.fn() as never,
     requestGateway,
+    resetViewSync: vi.fn(),
     runtimeIdByStoredSessionIdRef: ref(new Map<string, string>()),
     selectedStoredSessionId: null,
     selectedStoredSessionIdRef: ref<string | null>(null),

apps/desktop/src/app/session/hooks/use-session-actions/index.ts

@@ -75,6 +75,7 @@ interface SessionActionsOptions {
   getRouteToken: () => string
   navigate: NavigateFunction
   requestGateway: <T>(method: string, params?: Record<string, unknown>) => Promise<T>
+  resetViewSync: () => void
   runtimeIdByStoredSessionIdRef: MutableRefObject<Map<string, string>>
   selectedStoredSessionId: string | null
   selectedStoredSessionIdRef: MutableRefObject<string | null>
@@ -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,

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,

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'
+    })
+  })
 })

apps/desktop/src/lib/gateway-events.ts

@@ -11,6 +11,34 @@ function asRecord(payload: unknown): Record<string, unknown> {
   return payload && typeof payload === 'object' ? (payload as Record<string, unknown>) : {}
 }
 
+/**
+ * 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