[eric] streaming: a session socket that connects mid-reply is handed the text so far after its resume ack, instead of a static Thinking then the whole reply at once

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01C9zwUaHucUgrdxvK8FvjYT

frontend/src/shared/state/streamSnapshot.test.ts

@@ -0,0 +1,41 @@
+import { test } from 'node:test';
+import assert from 'node:assert/strict';
+import fs from 'node:fs';
+import path from 'node:path';
+import reducer, { streamStart, streamSnapshot, streamDelta, streamEnd } from './streamingSlice';
+
+// Every delta that arrives before the per-session socket connects is lost to that client (the ring
+// replays them, the client drops pre-ack stream frames on purpose), so the transcript sat on a static
+// "Thinking..." and then got the whole reply at once. The server now sends the text so far right after
+// the resume ack; the slice seeds the bubble with it and later deltas append.
+
+test('a snapshot seeds the streaming bubble and later deltas append to it', () => {
+  let s = reducer(undefined, streamSnapshot({ sessionId: 'a', messageId: 'm1', role: 'assistant', text: 'The Eiffel' }));
+  s = reducer(s, streamDelta({ sessionId: 'a', messageId: 'm1', delta: ' Tower' }));
+  assert.equal(s.bySession.a?.content, 'The Eiffel Tower');
+  s = reducer(s, streamEnd({ sessionId: 'a', messageId: 'm1' }));
+  assert.equal(s.bySession.a, undefined);
+});
+
+test('a snapshot never shortens text the client already has for the same message', () => {
+  let s = reducer(undefined, streamStart({ sessionId: 'a', messageId: 'm1', role: 'assistant' }));
+  s = reducer(s, streamDelta({ sessionId: 'a', messageId: 'm1', delta: 'The Eiffel Tower is tall' }));
+  s = reducer(s, streamSnapshot({ sessionId: 'a', messageId: 'm1', role: 'assistant', text: 'The Eiffel' }));
+  assert.equal(s.bySession.a?.content, 'The Eiffel Tower is tall');
+});
+
+test('a snapshot for a newer message replaces a stale bubble', () => {
+  let s = reducer(undefined, streamStart({ sessionId: 'a', messageId: 'old', role: 'assistant' }));
+  s = reducer(s, streamSnapshot({ sessionId: 'a', messageId: 'new', role: 'assistant', text: 'Second reply so far' }));
+  assert.equal(s.bySession.a?.id, 'new');
+  assert.equal(s.bySession.a?.content, 'Second reply so far');
+});
+
+test('the socket handles the snapshot ABOVE the replay-skip guard that drops pre-ack stream frames', () => {
+  const src = fs.readFileSync(path.join(process.cwd(), 'src/shared/ws/WebSocketManager.ts'), 'utf8');
+  const snapshot = src.indexOf("event === 'agent:stream_snapshot'");
+  const guard = src.indexOf('if (!this.resumeAcked) break;');
+  assert.ok(snapshot > 0 && guard > 0 && snapshot < guard, 'the snapshot must be handled before the pre-ack guard');
+  const dashboardSkip = src.indexOf('if (this.skipStreamEvents) {');
+  assert.ok(snapshot < dashboardSkip, 'the snapshot handler decides skipStreamEvents itself, above the generic skip');
+});

frontend/src/shared/state/streamingSlice.ts

@@ -41,6 +41,19 @@ const streamingSlice = createSlice({
         tool_name: action.payload.toolName,
       };
     },
+    // A socket that connected mid-reply is handed the text so far; every delta before it was lost to this client, so this seeds rather than appends.
+    streamSnapshot(
+      state,
+      action: PayloadAction<{ sessionId: string; messageId: string; role: StreamingMessage['role']; text: string }>,
+    ) {
+      const entry = state.bySession[action.payload.sessionId];
+      if (entry && entry.id === action.payload.messageId && entry.content.length >= action.payload.text.length) return;
+      state.bySession[action.payload.sessionId] = {
+        id: action.payload.messageId,
+        role: action.payload.role,
+        content: action.payload.text,
+      };
+    },
     streamDelta(
       state,
       action: PayloadAction<{ sessionId: string; messageId: string; delta: string }>,
@@ -91,7 +104,7 @@ const streamingSlice = createSlice({
   },
 });
 
-export const { streamStart, streamDelta, streamEnd, clearStreamingForSession } = streamingSlice.actions;
+export const { streamStart, streamSnapshot, streamDelta, streamEnd, clearStreamingForSession } = streamingSlice.actions;
 export default streamingSlice.reducer;
 
 /** Subscribes only to one session's stream entry; null when no stream is active. */

frontend/src/shared/ws/WebSocketManager.ts

@@ -32,7 +32,7 @@ import {
   setQueued,
   clearTurnLabel,
 } from '../state/agentsSlice';
-import { streamStart, streamDelta, streamEnd, clearStreamingForSession } from '../state/streamingSlice';
+import { streamStart, streamSnapshot, streamDelta, streamEnd, clearStreamingForSession } from '../state/streamingSlice';
 import { remountAppPreview } from '../state/outputsSlice';
 import { BackgroundDeltaBuffer } from './BackgroundDeltaBuffer';
 import { interactionActive, installInteractionListeners } from '../interactionPriority';
@@ -493,6 +493,13 @@ class WebSocketManager {
       return;
     }
 
+    // Sent once per (re)connect, after the resume ack, so it must not sit behind the replay-skip guard below that drops pre-ack stream frames.
+    if (event === 'agent:stream_snapshot') {
+      if (session_id && data.message_id && typeof data.text === 'string' && !this.skipStreamEvents) {
+        store.dispatch(streamSnapshot({ sessionId: session_id, messageId: data.message_id, role: data.role, text: data.text }));
+      }
+      return;
+    }
     if (this.skipStreamEvents) {
       if (event === 'agent:stream_start' || event === 'agent:stream_delta' || event === 'agent:stream_end') {
         return;