fix(desktop): track one unscoped stream pin per session

apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts

@@ -214,7 +214,9 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
     upsertToolCall
   } = deps
 
-  const unscopedStreamSessionIdRef = useRef<string | null>(null)
+  // One pin per concurrent unscoped stream, not a single shared slot: two chats
+  // streaming at once used to clobber each other's pin (#46194 / #62823).
+  const unscopedStreamSessionIdsRef = useRef<readonly string[]>([])
 
   // session.info arrives in bursts (agent build ready + turn end + title /
   // MCP / compress edges within the same second). Each used to fire its own
@@ -260,10 +262,10 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
         activeSessionId: activeSessionIdRef.current,
         eventType: event.type,
         explicitSessionId: explicitSid,
-        unscopedStreamSessionId: unscopedStreamSessionIdRef.current
+        unscopedStreamSessionIds: unscopedStreamSessionIdsRef.current
       })
 
-      unscopedStreamSessionIdRef.current = route.nextUnscopedStreamSessionId
+      unscopedStreamSessionIdsRef.current = route.nextUnscopedStreamSessionIds
 
       if (route.drop) {
         return

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

@@ -31,38 +31,38 @@ describe('gateway event routing', () => {
       activeSessionId: 'session-a',
       eventType: 'message.start',
       explicitSessionId: '',
-      unscopedStreamSessionId: null
+      unscopedStreamSessionIds: []
     })
 
     expect(started).toEqual({
       drop: false,
-      nextUnscopedStreamSessionId: 'session-a',
+      nextUnscopedStreamSessionIds: ['session-a'],
       sessionId: 'session-a'
     })
 
     const delta = resolveGatewayEventSessionId({
       activeSessionId: 'session-b',
       eventType: 'message.delta',
       explicitSessionId: '',
-      unscopedStreamSessionId: started.nextUnscopedStreamSessionId
+      unscopedStreamSessionIds: started.nextUnscopedStreamSessionIds
     })
 
     expect(delta).toEqual({
       drop: false,
-      nextUnscopedStreamSessionId: 'session-a',
+      nextUnscopedStreamSessionIds: ['session-a'],
       sessionId: 'session-a'
     })
 
     const completed = resolveGatewayEventSessionId({
       activeSessionId: 'session-b',
       eventType: 'message.complete',
       explicitSessionId: '',
-      unscopedStreamSessionId: delta.nextUnscopedStreamSessionId
+      unscopedStreamSessionIds: delta.nextUnscopedStreamSessionIds
     })
 
     expect(completed).toEqual({
       drop: false,
-      nextUnscopedStreamSessionId: null,
+      nextUnscopedStreamSessionIds: [],
       sessionId: 'session-a'
     })
   })
@@ -72,12 +72,14 @@ describe('gateway event routing', () => {
       activeSessionId: 'session-b',
       eventType: 'message.start',
       explicitSessionId: '',
-      unscopedStreamSessionId: 'session-a'
+      unscopedStreamSessionIds: ['session-a']
     })
 
+    // Session B owns its own start, but A's stream is still running and keeps
+    // its pin — the second start adds, it does not take over.
     expect(routed).toEqual({
       drop: false,
-      nextUnscopedStreamSessionId: 'session-b',
+      nextUnscopedStreamSessionIds: ['session-a', 'session-b'],
       sessionId: 'session-b'
     })
   })
@@ -87,13 +89,151 @@ describe('gateway event routing', () => {
       activeSessionId: 'session-b',
       eventType: 'message.complete',
       explicitSessionId: 'session-a',
-      unscopedStreamSessionId: 'session-a'
+      unscopedStreamSessionIds: ['session-a']
     })
 
     expect(routed).toEqual({
       drop: false,
-      nextUnscopedStreamSessionId: null,
+      nextUnscopedStreamSessionIds: [],
       sessionId: 'session-a'
     })
   })
+
+  it('retires only the completing stream when several run concurrently', () => {
+    const routed = resolveGatewayEventSessionId({
+      activeSessionId: 'session-b',
+      eventType: 'message.complete',
+      explicitSessionId: 'session-a',
+      unscopedStreamSessionIds: ['session-a', 'session-b']
+    })
+
+    expect(routed).toEqual({
+      drop: false,
+      nextUnscopedStreamSessionIds: ['session-b'],
+      sessionId: 'session-a'
+    })
+  })
+
+  it("does not let a second chat's stream steal the first chat's unscoped events", () => {
+    // The #46194 / #62823 race. A single shared pin was overwritten by B's
+    // message.start, so every later unscoped event from A resolved to B and
+    // painted A's output onto B's transcript.
+    const aStarted = resolveGatewayEventSessionId({
+      activeSessionId: 'session-a',
+      eventType: 'message.start',
+      explicitSessionId: '',
+      unscopedStreamSessionIds: []
+    })
+
+    const bStarted = resolveGatewayEventSessionId({
+      activeSessionId: 'session-b',
+      eventType: 'message.start',
+      explicitSessionId: '',
+      unscopedStreamSessionIds: aStarted.nextUnscopedStreamSessionIds
+    })
+
+    expect(bStarted.nextUnscopedStreamSessionIds).toEqual(['session-a', 'session-b'])
+
+    // A's stream completes while B is focused and still streaming. Before the
+    // per-stream pins this resolved to 'session-b'.
+    const aCompleted = resolveGatewayEventSessionId({
+      activeSessionId: 'session-b',
+      eventType: 'message.complete',
+      explicitSessionId: 'session-a',
+      unscopedStreamSessionIds: bStarted.nextUnscopedStreamSessionIds
+    })
+
+    expect(aCompleted.sessionId).toBe('session-a')
+    expect(aCompleted.nextUnscopedStreamSessionIds).toEqual(['session-b'])
+
+    // B's own unscoped delta still lands on B, unambiguously.
+    const bDelta = resolveGatewayEventSessionId({
+      activeSessionId: 'session-b',
+      eventType: 'message.delta',
+      explicitSessionId: '',
+      unscopedStreamSessionIds: aCompleted.nextUnscopedStreamSessionIds
+    })
+
+    expect(bDelta).toEqual({
+      drop: false,
+      nextUnscopedStreamSessionIds: ['session-b'],
+      sessionId: 'session-b'
+    })
+  })
+
+  it('attributes an ambiguous unscoped delta to the focused chat when it is streaming', () => {
+    const routed = resolveGatewayEventSessionId({
+      activeSessionId: 'session-b',
+      eventType: 'message.delta',
+      explicitSessionId: '',
+      unscopedStreamSessionIds: ['session-a', 'session-b']
+    })
+
+    expect(routed).toEqual({
+      drop: false,
+      nextUnscopedStreamSessionIds: ['session-a', 'session-b'],
+      sessionId: 'session-b'
+    })
+  })
+
+  it('drops an unscoped delta it cannot attribute to any of several live streams', () => {
+    // Two background streams, focused chat idle: nothing in the event says
+    // which stream it came from, and guessing is what grafts A's output onto B.
+    const routed = resolveGatewayEventSessionId({
+      activeSessionId: 'session-c',
+      eventType: 'message.delta',
+      explicitSessionId: '',
+      unscopedStreamSessionIds: ['session-a', 'session-b']
+    })
+
+    expect(routed).toEqual({
+      drop: true,
+      nextUnscopedStreamSessionIds: ['session-a', 'session-b'],
+      sessionId: null
+    })
+  })
+
+  it('leaves single-stream and no-stream routing unchanged', () => {
+    // #70376 owns tightening the no-pin fallback; this change must not move it.
+    const lateDelta = resolveGatewayEventSessionId({
+      activeSessionId: 'session-b',
+      eventType: 'message.delta',
+      explicitSessionId: '',
+      unscopedStreamSessionIds: []
+    })
+
+    expect(lateDelta).toEqual({
+      drop: false,
+      nextUnscopedStreamSessionIds: [],
+      sessionId: 'session-b'
+    })
+
+    const nonStreamEvent = resolveGatewayEventSessionId({
+      activeSessionId: 'session-b',
+      eventType: 'session.info',
+      explicitSessionId: '',
+      unscopedStreamSessionIds: ['session-a']
+    })
+
+    expect(nonStreamEvent).toEqual({
+      drop: false,
+      nextUnscopedStreamSessionIds: ['session-a'],
+      sessionId: 'session-b'
+    })
+  })
+
+  it('still drops unscoped subagent events without disturbing live pins', () => {
+    const routed = resolveGatewayEventSessionId({
+      activeSessionId: 'session-b',
+      eventType: 'subagent.progress',
+      explicitSessionId: '',
+      unscopedStreamSessionIds: ['session-a']
+    })
+
+    expect(routed).toEqual({
+      drop: true,
+      nextUnscopedStreamSessionIds: ['session-a'],
+      sessionId: null
+    })
+  })
 })

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

@@ -60,15 +60,57 @@ export interface GatewayEventSessionRouteInput {
   activeSessionId: null | string
   eventType: string | undefined
   explicitSessionId: string
-  unscopedStreamSessionId: null | string
+  /**
+   * Sessions with an unscoped stream in flight, in ``message.start`` order.
+   *
+   * One pin per concurrent stream. A single shared pin could not represent two
+   * chats streaming at once: the second ``message.start`` overwrote the first,
+   * and every later unscoped event from the first stream then resolved to the
+   * second chat (#46194 / #62823).
+   */
+  unscopedStreamSessionIds: readonly string[]
 }
 
 export interface GatewayEventSessionRoute {
   drop: boolean
-  nextUnscopedStreamSessionId: null | string
+  nextUnscopedStreamSessionIds: readonly string[]
   sessionId: null | string
 }
 
+const withStreamPin = (pins: readonly string[], sessionId: string): readonly string[] =>
+  pins.includes(sessionId) ? pins : [...pins, sessionId]
+
+const withoutStreamPin = (pins: readonly string[], sessionId: string): readonly string[] =>
+  pins.includes(sessionId) ? pins.filter(pin => pin !== sessionId) : pins
+
+/**
+ * Which in-flight stream owns an unscoped stream event.
+ *
+ * Returns `null` when ownership is genuinely ambiguous, so the caller drops the
+ * event instead of grafting one chat's output onto another.
+ */
+function resolveUnscopedStreamOwner(pins: readonly string[], activeSessionId: null | string): null | string {
+  // No concurrent streams: preserve the established single-stream behaviour —
+  // the lone pin owns it, and with no pin at all the focused chat does. The
+  // late-event case for that second branch is #70376's subject, not this one.
+  if (pins.length <= 1) {
+    return pins[0] ?? activeSessionId
+  }
+
+  // Two or more streams are live. The gateway stamps background sessions'
+  // events with their own id, so an unscoped one is the focused turn's output —
+  // but only when the focused chat is itself mid-stream.
+  if (activeSessionId && pins.includes(activeSessionId)) {
+    return activeSessionId
+  }
+
+  // The focused chat is idle, so this belongs to one of several background
+  // streams and nothing in the event says which. Guessing is what painted A's
+  // deltas onto B; drop instead. The store keeps the correct rows, so the
+  // transcript recovers on refetch.
+  return null
+}
+
 /**
  * Resolve which runtime session owns a gateway event.
  *
@@ -80,50 +122,66 @@ export function resolveGatewayEventSessionId({
   activeSessionId,
   eventType,
   explicitSessionId,
-  unscopedStreamSessionId
+  unscopedStreamSessionIds
 }: GatewayEventSessionRouteInput): GatewayEventSessionRoute {
-  if (explicitSessionId) {
-    const nextUnscopedStreamSessionId =
-      eventType && UNSCOPED_STREAM_END_EVENT_TYPES.has(eventType) && explicitSessionId === unscopedStreamSessionId
-        ? null
-        : unscopedStreamSessionId
+  const streamEnd = eventType ? UNSCOPED_STREAM_END_EVENT_TYPES.has(eventType) : false
 
+  if (explicitSessionId) {
     return {
       drop: false,
-      nextUnscopedStreamSessionId,
+      // Retire only the pin this event names. Streams still running in other
+      // chats keep theirs.
+      nextUnscopedStreamSessionIds: streamEnd
+        ? withoutStreamPin(unscopedStreamSessionIds, explicitSessionId)
+        : unscopedStreamSessionIds,
       sessionId: explicitSessionId
     }
   }
 
   if (gatewayEventRequiresSessionId(eventType)) {
     return {
       drop: true,
-      nextUnscopedStreamSessionId: unscopedStreamSessionId,
+      nextUnscopedStreamSessionIds: unscopedStreamSessionIds,
       sessionId: null
     }
   }
 
-  const streamEvent = eventType ? UNSCOPED_STREAM_EVENT_TYPES.has(eventType) : false
+  if (eventType === 'message.start') {
+    return {
+      drop: false,
+      // Add a pin rather than replace one, so a second chat starting a turn
+      // cannot take ownership of a stream that is already running elsewhere.
+      nextUnscopedStreamSessionIds: activeSessionId
+        ? withStreamPin(unscopedStreamSessionIds, activeSessionId)
+        : unscopedStreamSessionIds,
+      sessionId: activeSessionId
+    }
+  }
 
-  const sessionId =
-    eventType === 'message.start'
-      ? activeSessionId
-      : streamEvent
-        ? unscopedStreamSessionId || activeSessionId
-        : activeSessionId
+  if (!(eventType && UNSCOPED_STREAM_EVENT_TYPES.has(eventType))) {
+    return {
+      drop: false,
+      nextUnscopedStreamSessionIds: unscopedStreamSessionIds,
+      sessionId: activeSessionId
+    }
+  }
 
-  let nextUnscopedStreamSessionId = unscopedStreamSessionId
+  const owner = resolveUnscopedStreamOwner(unscopedStreamSessionIds, activeSessionId)
 
-  if (eventType === 'message.start' && activeSessionId) {
-    nextUnscopedStreamSessionId = activeSessionId
-  } else if (eventType && UNSCOPED_STREAM_END_EVENT_TYPES.has(eventType)) {
-    nextUnscopedStreamSessionId = null
+  if (!owner) {
+    return {
+      drop: true,
+      nextUnscopedStreamSessionIds: unscopedStreamSessionIds,
+      sessionId: null
+    }
   }
 
   return {
     drop: false,
-    nextUnscopedStreamSessionId,
-    sessionId
+    nextUnscopedStreamSessionIds: streamEnd
+      ? withoutStreamPin(unscopedStreamSessionIds, owner)
+      : unscopedStreamSessionIds,
+    sessionId: owner
   }
 }