diff --git a/src/lib/responses/ResponseStream.ts b/src/lib/responses/ResponseStream.ts index 1761f4bbf..563326787 100644 --- a/src/lib/responses/ResponseStream.ts +++ b/src/lib/responses/ResponseStream.ts @@ -27,7 +27,9 @@ export type ResponseStreamByIdParams = { */ response_id: string; /** - * If provided, the stream will start after the event with the given sequence number. + * If provided, events with a sequence number less than or equal to this value + * will not be emitted. The helper still replays them internally to build a + * complete snapshot for later events and `finalResponse()`. */ starting_after?: number; /** @@ -175,6 +177,8 @@ export class ResponseStream let stream: Stream | undefined; let starting_after: number | null = null; if ('response_id' in params) { + // Keep the full replay so that `accumulateResponse()` sees `response.created` and can build + // complete snapshots before locally filtering events at `starting_after`. stream = await client.responses.retrieve( params.response_id, { stream: true }, diff --git a/tests/lib/ResponseStream.test.ts b/tests/lib/ResponseStream.test.ts index d23449b34..a656a3a30 100644 --- a/tests/lib/ResponseStream.test.ts +++ b/tests/lib/ResponseStream.test.ts @@ -1,8 +1,77 @@ import { makeStreamSnapshotRequest } from '../utils/mock-snapshots'; +import OpenAI from 'openai'; jest.setTimeout(1000 * 30); describe('.stream()', () => { + it('replays prior events when resuming by ID so snapshots stay complete', async () => { + const requests: string[] = []; + const response = { + id: 'resp_123', + object: 'response', + created_at: 0, + model: 'gpt-4o', + output: [], + error: null, + incomplete_details: null, + instructions: null, + metadata: null, + parallel_tool_calls: false, + temperature: null, + tools: [], + top_p: null, + status: 'completed', + usage: null, + }; + const events = [ + { + type: 'response.created', + sequence_number: 0, + response: { ...response, status: 'in_progress' }, + }, + { + type: 'response.output_item.added', + sequence_number: 1, + output_index: 0, + item: { + id: 'msg_1', + type: 'message', + role: 'assistant', + status: 'in_progress', + content: [], + }, + }, + { type: 'response.completed', sequence_number: 2, response }, + ]; + const openai = new OpenAI({ + apiKey: 'My API Key', + fetch: async (url) => { + const requestURL = String(url); + requests.push(requestURL); + // Match the API's cursor behavior: forwarding `starting_after` omits the prefix needed + // by the accumulator, while omitting it replays the complete event sequence. + const streamEvents = requestURL.includes('starting_after=') ? events.slice(1) : events; + const body = `${streamEvents + .map((event) => `data: ${JSON.stringify(event)}`) + .join('\n\n')}\n\ndata: [DONE]\n\n`; + return new Response(body, { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + }); + }, + }); + + const emittedEvents: number[] = []; + const stream = openai.responses + .stream({ response_id: 'resp_123', starting_after: 0 }) + .on('event', (event) => emittedEvents.push(event.sequence_number)); + const final = await stream.finalResponse(); + + expect(requests).toEqual(['https://api.openai.com/v1/responses/resp_123?stream=true']); + expect(emittedEvents).toEqual([1, 2]); + expect(final.id).toBe('resp_123'); + }); + it('standard text works', async () => { const deltas: string[] = [];