feat: enhance useStreamingChat with background job handling and response event processing

Author: nickytonline

src/hooks/useStreamingChat.test.tsx

@@ -1140,6 +1140,123 @@ describe('useStreamingChat', () => {
     })
   })
 
+  describe('response.created events', () => {
+    it('should handle response.created with background: true option', async () => {
+      const { result } = renderHook(() => useStreamingChat())
+
+      const responseCreatedEvent = JSON.stringify({
+        type: 'response.created',
+        response: {
+          id: 'bg-job-123',
+        },
+      })
+
+      const mockReader = {
+        read: vi
+          .fn()
+          .mockResolvedValueOnce({
+            done: false,
+            value: new TextEncoder().encode(`t:${responseCreatedEvent}\n`),
+          })
+          .mockResolvedValueOnce({ done: true, value: undefined }),
+      }
+
+      const mockResponse = createMockResponse({
+        clone: vi.fn().mockReturnValue({
+          body: { getReader: vi.fn().mockReturnValue(mockReader) },
+        }),
+      })
+
+      act(() => {
+        result.current.handleResponse(mockResponse, {
+          background: true,
+          title: 'Test Background Job',
+        })
+      })
+
+      await waitFor(() => {
+        expect(result.current.streamBuffer).toHaveLength(1)
+      })
+
+      expect(result.current.streamBuffer[0]).toEqual({
+        type: 'assistant',
+        id: 'bg-job-123',
+        content:
+          "Background job started. Streaming the response in, but you can view it in the 'Background Jobs' if you leave.",
+      })
+    })
+
+    it('should ignore response.created with background: false option', async () => {
+      const { result } = renderHook(() => useStreamingChat())
+
+      const responseCreatedEvent = JSON.stringify({
+        type: 'response.created',
+        response: {
+          id: 'regular-job-123',
+        },
+      })
+
+      const mockReader = {
+        read: vi
+          .fn()
+          .mockResolvedValueOnce({
+            done: false,
+            value: new TextEncoder().encode(`t:${responseCreatedEvent}\n`),
+          })
+          .mockResolvedValueOnce({ done: true, value: undefined }),
+      }
+
+      const mockResponse = createMockResponse({
+        clone: vi.fn().mockReturnValue({
+          body: { getReader: vi.fn().mockReturnValue(mockReader) },
+        }),
+      })
+
+      act(() => {
+        result.current.handleResponse(mockResponse, { background: false })
+      })
+
+      await waitFor(() => {
+        expect(result.current.streamBuffer).toHaveLength(0)
+      })
+    })
+
+    it('should ignore response.created with no options', async () => {
+      const { result } = renderHook(() => useStreamingChat())
+
+      const responseCreatedEvent = JSON.stringify({
+        type: 'response.created',
+        response: {
+          id: 'no-options-job-123',
+        },
+      })
+
+      const mockReader = {
+        read: vi
+          .fn()
+          .mockResolvedValueOnce({
+            done: false,
+            value: new TextEncoder().encode(`t:${responseCreatedEvent}\n`),
+          })
+          .mockResolvedValueOnce({ done: true, value: undefined }),
+      }
+
+      const mockResponse = createMockResponse({
+        clone: vi.fn().mockReturnValue({
+          body: { getReader: vi.fn().mockReturnValue(mockReader) },
+        }),
+      })
+
+      act(() => {
+        result.current.handleResponse(mockResponse)
+      })
+
+      await waitFor(() => {
+        expect(result.current.streamBuffer).toHaveLength(0)
+      })
+    })
+  })
+
   describe('cleanup', () => {
     it('should cleanup timeouts on unmount', async () => {
       const clearTimeoutSpy = vi.spyOn(global, 'clearTimeout')

src/hooks/useStreamingChat.ts

@@ -3,6 +3,7 @@ import { generateMessageId } from '../mcp/client'
 import type { AnnotatedFile } from '@/lib/utils/code-interpreter'
 import { stopStreamProcessing } from '@/lib/utils/streaming'
 import { getTimestamp } from '@/lib/utils/date'
+import { useBackgroundJobs } from './useBackgroundJobs'
 
 export type AssistantStreamEvent = {
   type: 'assistant'
@@ -139,7 +140,10 @@ interface UseStreamingChatReturn {
   streaming: boolean
   timedOut: boolean
   requestId: string | null
-  handleResponse: (response: Response) => void
+  handleResponse: (
+    response: Response,
+    options?: { background: true; title: string } | { background: false },
+  ) => void
   handleError: (error: Error) => void
   addUserMessage: (content: string) => void
   cancelStream: () => void
@@ -151,6 +155,7 @@ export function useStreamingChat(): UseStreamingChatReturn {
   const [streaming, setStreaming] = useState(false)
   const [timedOut, setTimedOut] = useState(false)
   const [requestId, setRequestId] = useState<string | null>(null)
+  const { addJob: addBackgroundJob } = useBackgroundJobs()
 
   const streamUpdateTimeoutRef = useRef<NodeJS.Timeout | null>(null)
   const textBufferRef = useRef<string>('')
@@ -263,7 +268,10 @@ export function useStreamingChat(): UseStreamingChatReturn {
   }, [])
 
   const handleResponse = useCallback(
-    (response: Response) => {
+    (
+      response: Response,
+      options?: { background: true; title: string } | { background: false },
+    ) => {
       const xRequestId = response.headers.get('x-request-id')
       setRequestId(xRequestId)
 
@@ -366,6 +374,33 @@ export function useStreamingChat(): UseStreamingChatReturn {
               return
             }
 
+            // Currently only used for background jobs
+            if (toolState.type === 'response.created') {
+              if (options?.background) {
+                const requestId = toolState.response.id
+
+                const backgroundJob = {
+                  id: requestId,
+                  status: 'running' as const,
+                  createdAt: getTimestamp(),
+                  title: options.title,
+                }
+
+                addBackgroundJob(backgroundJob)
+
+                setStreamBuffer((prev: Array<StreamEvent>) => [
+                  ...prev,
+                  {
+                    type: 'assistant',
+                    id: backgroundJob.id,
+                    content:
+                      "Background job started. Streaming the response in, but you can view it in the 'Background Jobs' if you leave.",
+                  },
+                ])
+              }
+              return
+            }
+
             if (toolState.type === 'reasoning_summary_delta') {
               setStreamBuffer((prev: Array<StreamEvent>) => {
                 const last = prev[prev.length - 1]

src/lib/__snapshots__/streaming.test.ts.snap

@@ -0,0 +1,37 @@
+// Vitest Snapshot v1, https://vitest.dev/guide/snapshot.html
+
+exports[`streamText > emits stream_done when an error occurs 1`] = `
+"f:{"messageId":"msg-2"}
+e:{"type":"error","message":"An unexpected error occurred."}
+t:{"type":"stream_done"}
+"
+`;
+
+exports[`streamText > emits stream_done when the stream ends 1`] = `
+"f:{"messageId":"msg-1"}
+0:"I'm glad you asked!"
+0:" Here are a few universally nice things that"
+0:" could have happened today:\\n\\n"
+0:"- Someone smiled at a stranger, brightening"
+0:" their day\\n"
+0:"- A teacher helped a student understand a"
+0:" difficult concept\\n"
+0:"- A kind person paid for someone’s coffee"
+0:" in line\\n"
+0:"- A pet reunited with its owner at a local"
+0:" animal shelter\\n"
+0:"- A friend reached out just to say hello\\n\\n"
+0:"Would you like to hear some real, uplifting"
+0:" news from today?"
+0:" Just let me know!"
+t:{"type":"tool_call_completed","response":{}}
+t:{"type":"stream_done"}
+"
+`;
+
+exports[`streamText > passes through response.created events 1`] = `
+"f:{"messageId":"msg-3"}
+t:{"type":"response.created","response":{"id":"resp_123","model":"gpt-4","created":1234567890}}
+t:{"type":"stream_done"}
+"
+`;

src/lib/streaming.test.ts

@@ -1,7 +1,15 @@
-import { describe, expect, it } from 'vitest'
-import { stopStreamProcessing } from './utils/streaming'
+import { describe, expect, it, beforeEach, afterEach } from 'vitest'
+import { stopStreamProcessing, getMessageId } from './utils/streaming'
 import { streamText } from './streaming'
 
+vi.mock('./utils/streaming', async () => {
+  const actual = await vi.importActual('./utils/streaming')
+  return {
+    ...actual,
+    getMessageId: vi.fn(),
+  }
+})
+
 function iterableFromArray<T>(arr: Array<T>): AsyncIterable<T> {
   return {
     [Symbol.asyncIterator]() {
@@ -17,6 +25,19 @@ function iterableFromArray<T>(arr: Array<T>): AsyncIterable<T> {
   }
 }
 
+async function readStreamToString(
+  reader: ReadableStreamDefaultReader<Uint8Array>,
+): Promise<string> {
+  let result = ''
+  let done = false
+  while (!done) {
+    const { value, done: d } = await reader.read()
+    if (value) result += new TextDecoder().decode(value)
+    done = d
+  }
+  return result
+}
+
 describe('stopStreamProcessing', () => {
   it('replaces response body with an empty closed stream', async () => {
     const originalStream = new ReadableStream({
@@ -70,7 +91,31 @@ describe('stopStreamProcessing', () => {
   })
 })
 
+describe('getMessageId', () => {
+  it('returns a unique message ID with msg prefix', async () => {
+    const actual = await vi.importActual('./utils/streaming') as { getMessageId: () => string }
+    const id1 = actual.getMessageId()
+    const id2 = actual.getMessageId()
+    
+    expect(id1).toMatch(/^msg-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/)
+    expect(id2).toMatch(/^msg-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/)
+    expect(id1).not.toBe(id2)
+  })
+})
+
 describe('streamText', () => {
+  let messageCounter = 0
+
+  beforeEach(() => {
+    vi.mocked(getMessageId).mockImplementation(() => {
+      messageCounter++
+      return `msg-${messageCounter}`
+    })
+  })
+
+  afterEach(() => {
+    vi.restoreAllMocks()
+  })
   it('emits stream_done when the stream ends', async () => {
     const chunks = [
       { type: 'response.output_text.delta', delta: "I'm glad you asked!" },
@@ -127,14 +172,8 @@ describe('streamText', () => {
     ]
     const response = streamText(iterableFromArray(chunks))
     const reader = response.body!.getReader()
-    let result = ''
-    let done = false
-    while (!done) {
-      const { value, done: d } = await reader.read()
-      if (value) result += new TextDecoder().decode(value)
-      done = d
-    }
-    expect(result).toMatch(/t:{"type":"stream_done"}/)
+    const result = await readStreamToString(reader)
+    expect(result).toMatchSnapshot()
   })
 
   it('emits stream_done when an error occurs', async () => {
@@ -153,16 +192,9 @@ describe('streamText', () => {
     }
     const response = streamText(errorIterable)
     const reader = response.body!.getReader()
-    let result = ''
-    let done = false
-    while (!done) {
-      const { value, done: d } = await reader.read()
-      if (value) result += new TextDecoder().decode(value)
-      done = d
-    }
+    const result = await readStreamToString(reader)
 
-    expect(result).toMatch(/t:{"type":"stream_done"}/)
-    expect(result).toMatch(/e:{"type":"error".*}/)
+    expect(result).toMatchSnapshot()
     expect(consoleErrorSpy).toHaveBeenCalledWith(
       'Error during streamed response:',
       expect.any(Error),
@@ -173,4 +205,22 @@ describe('streamText', () => {
 
     consoleErrorSpy.mockRestore()
   })
+
+  it('passes through response.created events', async () => {
+    const chunks = [
+      {
+        type: 'response.created',
+        response: {
+          id: 'resp_123',
+          model: 'gpt-4',
+          created: 1234567890,
+        },
+      },
+    ]
+    const response = streamText(iterableFromArray(chunks))
+    const reader = response.body!.getReader()
+    const result = await readStreamToString(reader)
+
+    expect(result).toMatchSnapshot()
+  })
 })

src/lib/streaming.ts

@@ -1,4 +1,5 @@
 import { APIError } from 'openai'
+import { getMessageId } from './utils/streaming'
 
 // Event chunk for stream completion
 const STREAM_DONE_CHUNK = new TextEncoder().encode(
@@ -10,7 +11,7 @@ export function streamText(
   onMessageId?: (messageId: string) => void,
 ): Response {
   const encoder = new TextEncoder()
-  const messageId = `msg-${Math.random().toString(36).slice(2)}`
+  const messageId = getMessageId()
 
   const stream = new ReadableStream({
     async start(controller) {
@@ -336,8 +337,14 @@ export function streamText(
                 ),
               )
               break
-            // Web search tool events
+
+            case 'response.created':
+              // Pass through response.created events so background jobs can extract response ID
+              controller.enqueue(encoder.encode(`t:${JSON.stringify(chunk)}\n`))
+              break
+
             default:
+              // Web search tool events
               if (
                 chunk.type &&
                 chunk.type.startsWith('response.web_search_call.')