From 549335db1565ccbd6324eb83a58656553cf808e0 Mon Sep 17 00:00:00 2001 From: Cruz Morales <273886418+DojoGenesis@users.noreply.github.com> Date: Mon, 3 Aug 2026 10:42:34 -0500 Subject: [PATCH] fix(desktop): track one unscoped stream pin per session MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `unscopedStreamSessionId` was a single shared slot. When a second chat started a turn while the first was still streaming, its `message.start` overwrote the pin, and every later unscoped event from the first stream resolved to the second chat — grafting one conversation's deltas, tool events and reasoning onto another's transcript (#46194 / #62823). Replace the slot with one pin per concurrent stream. `message.start` adds a pin instead of replacing it; a stream end retires only its own pin, so chats still streaming keep theirs. With several streams live an unscoped event has no field naming its owner, so it is attributed to the focused chat when that chat is itself mid-stream, and dropped otherwise. Dropping is the conservative half: the store keeps the correct rows, so the transcript recovers on refetch, whereas guessing is what painted A's output onto B. Single-stream and no-stream routing are unchanged — tightening the no-pin fallback for late events is #70376's subject, not this change. Co-Authored-By: Claude Opus 5 --- .../hooks/use-message-stream/gateway-event.ts | 8 +- apps/desktop/src/lib/gateway-events.test.ts | 160 ++++++++++++++++-- apps/desktop/src/lib/gateway-events.ts | 106 +++++++++--- 3 files changed, 237 insertions(+), 37 deletions(-) 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 7493cf19e6735..0a1c4599e7200 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 @@ -214,7 +214,9 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { upsertToolCall } = deps - const unscopedStreamSessionIdRef = useRef(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([]) // 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 diff --git a/apps/desktop/src/lib/gateway-events.test.ts b/apps/desktop/src/lib/gateway-events.test.ts index 02c3f643ca6a6..4765c08c0d396 100644 --- a/apps/desktop/src/lib/gateway-events.test.ts +++ b/apps/desktop/src/lib/gateway-events.test.ts @@ -31,12 +31,12 @@ 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' }) @@ -44,12 +44,12 @@ describe('gateway event routing', () => { 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' }) @@ -57,12 +57,12 @@ describe('gateway event routing', () => { 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 + }) + }) }) diff --git a/apps/desktop/src/lib/gateway-events.ts b/apps/desktop/src/lib/gateway-events.ts index 005e79705c15b..75c27c85bee0e 100644 --- a/apps/desktop/src/lib/gateway-events.ts +++ b/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,17 +122,18 @@ 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 } } @@ -98,32 +141,47 @@ export function resolveGatewayEventSessionId({ 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 } }