first stream test passing

Author: amychisholm03

lib/instrumentation/openai.js

@@ -285,9 +285,9 @@ function instrumentStream({ agent, shim, request, response, segment, transaction
         err = streamErr
         throw err
       } finally {
-        if (chunk.choices) {
+        if (chunk?.choices) {
           chunk.choices[0].message = { role, content }
-        } else if (chunk.response) {
+        } else if (chunk?.response) {
           chunk = chunk.response
         }
 

test/versioned/openai/chat-completions-v5.test.js

@@ -296,4 +296,35 @@ test('responses.create', async (t) => {
       end()
     })
   })
+
+  await t.test('should create span on successful responses stream create', (t, end) => {
+    const { client, agent, host, port } = t.nr
+    helper.runInTransaction(agent, async (tx) => {
+      const content = 'Streamed response'
+      const stream = await client.responses.create({
+        stream: true,
+        input: content,
+        model: 'gpt-4'
+      })
+
+      let chunk = {}
+      for await (chunk of stream) {
+        continue
+      }
+      assert.equal(chunk.headers, undefined, 'should remove response headers from user result')
+      assert.equal(chunk.response.output[0].role, 'assistant')
+      const expectedRes = responses.get(content)
+      assert.equal(chunk.response.output[0].content[0].text, expectedRes.body.response.output[0].content[0].text)
+
+      assertSegments(
+        tx.trace,
+        tx.trace.root,
+        [OPENAI.COMPLETION, `External/${host}:${port}/responses`],
+        { exact: false }
+      )
+
+      tx.end()
+      end()
+    })
+  })
 })

test/versioned/openai/mock-responses-api-responses.js

@@ -118,24 +118,22 @@ responses.set('Streamed response', {
   code: 200,
   body: {
     type: 'response.completed',
-    sequence_number: 100,
+    sequence_number: 9,
     response: {
-      id: 'resp_68420d9a5d4481a1bff5b86663299e3403b76731ee674f61',
+      id: 'resp_684886977be881928c9db234e14ae7d80f8976796514dff9',
       object: 'response',
       created_at: 1749221320,
       model: 'gpt-4-0613',
       output: [{
         content: [{
-          text: "A streamed response is a way of transmitting data from a server to a client (e.g. from a website to a user's computer or mobile device) in a continuous flow or stream, rather than all at one time. This means the client can start to process the data before all of it has been received, which can improve performance for large amounts of data or slow connections. Streaming is often used for real-time or near-real-time applications like video or audio playback.",
+          text: 'Test stream',
         }],
         role: 'assistant',
         status: 'completed',
-        id: 'msg_6843007469bc8192af5e145250c297db0374f342293105d9',
+        id: 'msg_68488698f6088192a505b70393c560bc0f8976796514dff9',
       }]
     }
-  },
-  // For testing purposes only
-  streamData: "A streamed response is a way of transmitting data from a server to a client (e.g. from a website to a user's computer or mobile device) in a continuous flow or stream, rather than all at one time. This means the client can start to process the data before all of it has been received, which can improve performance for large amounts of data or slow connections. Streaming is often used for real-time or near-real-time applications like video or audio playback.",
+  }
 })
 
 responses.set('bad stream', {

test/versioned/openai/mock-server-v5.js

@@ -9,6 +9,8 @@ module.exports = openaiMockServer
 
 const http = require('node:http')
 const RESPONSES = require('./mock-responses-api-responses')
+const STREAM_CHUNKS = require('./stream-chunks-v5')
+const { Readable } = require('node:stream')
 
 /**
  * Build a mock server that listens on a 127.0.0.1 and a random port that
@@ -64,23 +66,76 @@ function handler(req, res) {
       return
     }
 
-    const { headers, code, body } = RESPONSES.get(prompt)
+    const { headers, code, body, streamData } = RESPONSES.get(prompt)
     for (const [key, value] of Object.entries(headers)) {
       res.setHeader(key, value)
     }
     res.statusCode = code
 
     if (payload.stream === true) {
-      res.statusCode = 500
-      res.write('Streaming is not yet supported in this mock server.')
-      res.end()
+      let outStream
+      if (streamData !== 'do random') {
+        outStream = finiteStream({ ...body })
+      } else {
+        outStream = randomStream({ ...body })
+        let streamChunkCount = 0
+        outStream.on('data', () => {
+          if (streamChunkCount >= 100) {
+            outStream.destroy()
+            res.destroy()
+          }
+          streamChunkCount += 1
+        })
+      }
+
+      outStream.pipe(res)
     } else {
       res.write(JSON.stringify(body))
       res.end()
     }
   })
 }
 
+/**
+ * Uses `STREAM_CHUNKS` into chunks to returns a stream that
+ * sends those chunks as OpenAI v5 data stream messages. This stream
+ * has a finite number of messages that will be sent.
+ *
+ * @param body
+ * @returns {Readable} A paused stream.
+ */
+function finiteStream(body) {
+  return new Readable({
+    read() {
+      // This is how the data is streamed from openai
+      for (let i = 0; i < STREAM_CHUNKS.length; i++) {
+        const chunkString = JSON.stringify(STREAM_CHUNKS[i])
+        this.push(`data: ${chunkString}\n\n`)
+      }
+      this.push('data: [DONE]\n\n')
+      this.push(null)
+    }
+  }).pause()
+}
+
+/**
+ * Creates a stream that will stream an infinite number of OpenAI stream data
+ * chunks.
+ *
+ * @param {object} chunkTemplate An object that is shaped like an OpenAI stream
+ * data object.
+ * @returns {Readable} A paused stream.
+ */
+function randomStream(chunkTemplate) {
+  return new Readable({
+    read(size = 16) {
+      const data = crypto.randomBytes(size)
+      chunkTemplate.choices[0].delta.content = data.toString('base64')
+      this.push('data: ' + JSON.stringify(chunkTemplate) + '\n\n')
+    }
+  }).pause()
+}
+
 function getShortenedPrompt(reqBody) {
   const prompt = reqBody.input?.[0]?.content || reqBody.input?.badContent || reqBody.input
 

test/versioned/openai/stream-chunks-v5.js

@@ -0,0 +1,137 @@
+/*
+ * Copyright 2025 New Relic Corporation. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+'use strict'
+
+const chunks = []
+module.exports = chunks
+
+// Setup chunks
+chunks.push({
+  response: {
+    id: 'resp_684886977be881928c9db234e14ae7d80f8976796514dff9',
+    model: 'gpt-4-0613',
+    object: 'response',
+    status: 'in_progress',
+    output: []
+  },
+  type: 'response.created',
+  sequence_number: 0
+})
+
+chunks.push({
+  response: {
+    id: 'resp_684886977be881928c9db234e14ae7d80f8976796514dff9',
+    model: 'gpt-4-0613',
+    object: 'response',
+    status: 'in_progress',
+    output: []
+  },
+  type: 'response.in_progress',
+  sequence_number: 1
+})
+
+chunks.push({
+  item: {
+    id: 'msg_68488698f6088192a505b70393c560bc0f8976796514dff9',
+    role: 'assistant',
+    status: 'in_progress',
+    type: 'message'
+  },
+  output_index: 0,
+  type: 'response.output_item.added',
+  sequence_number: 2
+})
+
+chunks.push({
+  part: {
+    type: 'output_text',
+    text: ''
+  },
+  content_index: 0,
+  output_index: 0,
+  item_id: 'msg_68488698f6088192a505b70393c560bc0f8976796514dff9',
+  type: 'response.content_part.added',
+  sequence_number: 3
+})
+
+// Delta chunks for the actual text
+chunks.push({
+  content_index: 0,
+  delta: 'Test',
+  item_id: 'msg_68488698f6088192a505b70393c560bc0f8976796514dff9',
+  output_index: 0,
+  sequence_number: 4,
+  type: 'response.output_text.delta',
+})
+
+chunks.push({
+  content_index: 0,
+  delta: 'stream',
+  item_id: 'msg_68488698f6088192a505b70393c560bc0f8976796514dff9',
+  output_index: 0,
+  sequence_number: 5,
+  type: 'response.output_text.delta',
+})
+
+// Finishing up - summing deltas together
+chunks.push({
+  content_index: 0,
+  item_id: 'msg_68488698f6088192a505b70393c560bc0f8976796514dff9',
+  output_index: 0,
+  sequence_number: 6,
+  text: 'Test stream',
+  type: 'response.output_text.done',
+})
+
+chunks.push({
+  content_index: 0,
+  item_id: 'msg_68488698f6088192a505b70393c560bc0f8976796514dff9',
+  output_index: 0,
+  part: {
+    type: 'output_text',
+    text: 'Test stream'
+  },
+  sequence_number: 7,
+  type: 'response.content_part.done',
+})
+
+chunks.push({
+  item: {
+    content: [{
+      text: 'Test stream',
+      type: 'output_text'
+    }],
+    id: 'msg_68488698f6088192a505b70393c560bc0f8976796514dff9',
+    role: 'assistant',
+    status: 'completed',
+    type: 'message'
+  },
+  output_index: 0,
+  sequence_number: 8,
+  type: 'response.output_item.done',
+})
+
+chunks.push({
+  response: {
+    id: 'resp_684886977be881928c9db234e14ae7d80f8976796514dff9',
+    object: 'response',
+    output: [{
+      content: [{ text: 'Test stream' }],
+      id: 'msg_68488698f6088192a505b70393c560bc0f8976796514dff9',
+      role: 'assistant',
+      status: 'completed',
+      type: 'message'
+    }],
+    status: 'completed',
+    usage: {
+      input_tokens: 13,
+      output_tokens: 4,
+      total_tokens: 17
+    }
+  },
+  sequence_number: 9,
+  type: 'response_completed',
+})