Skip to content

Commit dc385ee

Browse files
AlemTuzlakAlemTuzlak
committed
fix(ai-openrouter, ai): emit single STEP_FINISHED per reasoning block, remove [DONE] sentinel
STEP_FINISHED was emitted on every reasoning delta (N events for N deltas) but only one STEP_STARTED was emitted, causing verifiers to report orphan STEP_FINISHED events. Move the single STEP_FINISHED to the point where reasoning closes (before text starts or at stream end) so every STEP_STARTED has exactly one matching STEP_FINISHED. Remove the `data: [DONE]\n\n` sentinel from toServerSentEventsStream. The AG-UI protocol already uses RUN_FINISHED as the terminal event, so the [DONE] marker is redundant and forces every client to special-case non-JSON data in the SSE stream. Client-side parsers still tolerate [DONE] for backward compatibility with external servers.
1 parent 1f677e9 commit dc385ee

4 files changed

Lines changed: 164 additions & 61 deletions

File tree

‎packages/typescript/ai-openrouter/src/adapters/text.ts‎

Lines changed: 24 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -377,17 +377,6 @@ export class OpenRouterTextAdapter<
377377
model: meta.model,
378378
timestamp: meta.timestamp,
379379
})
380-
381-
// Legacy STEP event
382-
yield asChunk({
383-
type: 'STEP_FINISHED',
384-
stepName: aguiState.stepId!,
385-
stepId: aguiState.stepId!,
386-
model: meta.model,
387-
timestamp: meta.timestamp,
388-
delta: text,
389-
content: accumulated.reasoning,
390-
})
391380
continue
392381
}
393382
if (detail.type === 'reasoning.summary') {
@@ -436,17 +425,6 @@ export class OpenRouterTextAdapter<
436425
model: meta.model,
437426
timestamp: meta.timestamp,
438427
})
439-
440-
// Legacy STEP event
441-
yield asChunk({
442-
type: 'STEP_FINISHED',
443-
stepName: aguiState.stepId!,
444-
stepId: aguiState.stepId!,
445-
model: meta.model,
446-
timestamp: meta.timestamp,
447-
delta: text,
448-
content: accumulated.reasoning,
449-
})
450428
continue
451429
}
452430
}
@@ -468,6 +446,18 @@ export class OpenRouterTextAdapter<
468446
model: meta.model,
469447
timestamp: meta.timestamp,
470448
})
449+
450+
// Legacy: single STEP_FINISHED to close the STEP_STARTED
451+
if (aguiState.stepId) {
452+
yield asChunk({
453+
type: 'STEP_FINISHED',
454+
stepName: aguiState.stepId,
455+
stepId: aguiState.stepId,
456+
model: meta.model,
457+
timestamp: meta.timestamp,
458+
content: accumulated.reasoning,
459+
})
460+
}
471461
}
472462

473463
// Emit TEXT_MESSAGE_START on first text content
@@ -624,6 +614,18 @@ export class OpenRouterTextAdapter<
624614
model: meta.model,
625615
timestamp: meta.timestamp,
626616
})
617+
618+
// Legacy: single STEP_FINISHED to close the STEP_STARTED
619+
if (aguiState.stepId) {
620+
yield asChunk({
621+
type: 'STEP_FINISHED',
622+
stepName: aguiState.stepId,
623+
stepId: aguiState.stepId,
624+
model: meta.model,
625+
timestamp: meta.timestamp,
626+
content: accumulated.reasoning,
627+
})
628+
}
627629
}
628630

629631
// Emit TEXT_MESSAGE_END if we had text content

‎packages/typescript/ai-openrouter/tests/openrouter-adapter.test.ts‎

Lines changed: 82 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -801,13 +801,13 @@ describe('OpenRouter AG-UI event emission', () => {
801801
expect(stepStartedChunk.stepType).toBe('thinking')
802802
}
803803

804-
// Check for STEP_FINISHED event
804+
// Check for STEP_FINISHED event — emitted once when reasoning closes
805805
const stepFinishedChunks = chunks.filter((c) => c.type === 'STEP_FINISHED')
806-
expect(stepFinishedChunks.length).toBeGreaterThan(0)
806+
expect(stepFinishedChunks).toHaveLength(1)
807807
const stepFinishedChunk = stepFinishedChunks[0]
808808
if (stepFinishedChunk?.type === 'STEP_FINISHED') {
809809
expect(stepFinishedChunk.stepId).toBeDefined()
810-
expect(stepFinishedChunk.delta).toBe('Let me think about this...')
810+
expect(stepFinishedChunk.content).toBe('Let me think about this...')
811811
}
812812
})
813813
})
@@ -1162,4 +1162,83 @@ describe('OpenRouter STEP event consistency', () => {
11621162
expect(hasMatchingStart).toBe(true)
11631163
}
11641164
})
1165+
1166+
it('emits exactly one STEP_STARTED and one STEP_FINISHED for multi-delta reasoning', async () => {
1167+
// When multiple reasoning deltas arrive, the adapter should emit a
1168+
// single STEP_STARTED/STEP_FINISHED pair — not one STEP_FINISHED per
1169+
// delta. A 1:N ratio causes verifiers to report orphan STEP_FINISHED.
1170+
const streamChunks = [
1171+
{
1172+
id: 'chatcmpl-multi',
1173+
model: 'openai/o1-preview',
1174+
choices: [
1175+
{
1176+
delta: {
1177+
reasoningDetails: [{ type: 'reasoning.text', text: 'Let me ' }],
1178+
},
1179+
finishReason: null,
1180+
},
1181+
],
1182+
},
1183+
{
1184+
id: 'chatcmpl-multi',
1185+
model: 'openai/o1-preview',
1186+
choices: [
1187+
{
1188+
delta: {
1189+
reasoningDetails: [
1190+
{ type: 'reasoning.text', text: 'think about ' },
1191+
],
1192+
},
1193+
finishReason: null,
1194+
},
1195+
],
1196+
},
1197+
{
1198+
id: 'chatcmpl-multi',
1199+
model: 'openai/o1-preview',
1200+
choices: [
1201+
{
1202+
delta: {
1203+
reasoningDetails: [{ type: 'reasoning.text', text: 'this...' }],
1204+
},
1205+
finishReason: null,
1206+
},
1207+
],
1208+
},
1209+
{
1210+
id: 'chatcmpl-multi',
1211+
model: 'openai/o1-preview',
1212+
choices: [
1213+
{
1214+
delta: { content: 'The answer is 42.' },
1215+
finishReason: null,
1216+
},
1217+
],
1218+
},
1219+
{
1220+
id: 'chatcmpl-multi',
1221+
model: 'openai/o1-preview',
1222+
choices: [{ delta: {}, finishReason: 'stop' }],
1223+
usage: { promptTokens: 20, completionTokens: 10, totalTokens: 30 },
1224+
},
1225+
]
1226+
1227+
setupMockSdkClient(streamChunks)
1228+
const adapter = createAdapter()
1229+
const chunks: Array<StreamChunk> = []
1230+
1231+
for await (const chunk of adapter.chatStream({
1232+
model: 'openai/o1-preview',
1233+
messages: [{ role: 'user', content: 'What is the meaning of life?' }],
1234+
})) {
1235+
chunks.push(chunk)
1236+
}
1237+
1238+
const stepStarted = chunks.filter((c) => c.type === 'STEP_STARTED')
1239+
const stepFinished = chunks.filter((c) => c.type === 'STEP_FINISHED')
1240+
1241+
expect(stepStarted).toHaveLength(1)
1242+
expect(stepFinished).toHaveLength(1)
1243+
})
11651244
})

‎packages/typescript/ai/src/stream-to-response.ts‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ export async function streamToText(
4040
* This creates a ReadableStream that emits chunks in SSE format:
4141
* - Each chunk is prefixed with "data: "
4242
* - Each chunk is followed by "\n\n"
43-
* - Stream ends with "data: [DONE]\n\n"
43+
* - Stream ends when the underlying iterable is exhausted (RUN_FINISHED is the terminal event)
4444
*
4545
* @param stream - AsyncIterable of StreamChunks from chat()
4646
* @param abortController - Optional AbortController to abort when stream is cancelled
@@ -67,8 +67,6 @@ export function toServerSentEventsStream(
6767
)
6868
}
6969

70-
// Send completion marker
71-
controller.enqueue(encoder.encode('data: [DONE]\n\n'))
7270
controller.close()
7371
} catch (error: any) {
7472
// Don't send error if aborted
@@ -109,7 +107,7 @@ export function toServerSentEventsStream(
109107
* This creates a Response that emits chunks in SSE format:
110108
* - Each chunk is prefixed with "data: "
111109
* - Each chunk is followed by "\n\n"
112-
* - Stream ends with "data: [DONE]\n\n"
110+
* - Stream ends when the underlying iterable is exhausted (RUN_FINISHED is the terminal event)
113111
*
114112
* @param stream - AsyncIterable of StreamChunks from chat()
115113
* @param init - Optional Response initialization options (including `abortController`)

‎packages/typescript/ai/tests/stream-to-response.test.ts‎

Lines changed: 56 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,7 @@ describe('toServerSentEventsStream', () => {
6161
expect(output).toContain('data: ')
6262
expect(output).toContain('"type":"TEXT_MESSAGE_CONTENT"')
6363
expect(output).toContain('\n\n')
64-
expect(output).toContain('data: [DONE]\n\n')
64+
expect(output).not.toContain('[DONE]')
6565
})
6666

6767
it('should format each chunk with data: prefix', async () => {
@@ -82,30 +82,7 @@ describe('toServerSentEventsStream', () => {
8282

8383
const lines = output.split('\n\n').filter((line) => line.trim())
8484
expect(lines[0]).toMatch(/^data: /)
85-
expect(lines[lines.length - 1]).toBe('data: [DONE]')
86-
})
87-
88-
it('should end with [DONE] marker', async () => {
89-
const chunks: Array<Record<string, unknown>> = [
90-
{
91-
type: 'TEXT_MESSAGE_CONTENT',
92-
messageId: 'msg-1',
93-
model: 'test',
94-
timestamp: Date.now(),
95-
delta: 'Test',
96-
content: 'Test',
97-
},
98-
]
99-
100-
const stream = createMockStream(chunks)
101-
const sseStream = toServerSentEventsStream(stream)
102-
const output = await readStream(sseStream)
103-
104-
// Should end with [DONE] marker followed by newlines
105-
expect(output).toContain('data: [DONE]')
106-
const doneIndex = output.lastIndexOf('data: [DONE]')
107-
const afterDone = output.slice(doneIndex)
108-
expect(afterDone).toBe('data: [DONE]\n\n')
85+
expect(lines[lines.length - 1]).toMatch(/^data: \{/)
10986
})
11087

11188
it('should handle tool call events', async () => {
@@ -126,7 +103,7 @@ describe('toServerSentEventsStream', () => {
126103

127104
expect(output).toContain('"type":"TOOL_CALL_START"')
128105
expect(output).toContain('"toolName":"getWeather"')
129-
expect(output).toContain('data: [DONE]\n\n')
106+
expect(output).not.toContain('[DONE]')
130107
})
131108

132109
it('should handle RUN_FINISHED events', async () => {
@@ -146,7 +123,7 @@ describe('toServerSentEventsStream', () => {
146123

147124
expect(output).toContain('"type":"RUN_FINISHED"')
148125
expect(output).toContain('"finishReason":"stop"')
149-
expect(output).toContain('data: [DONE]\n\n')
126+
expect(output).not.toContain('[DONE]')
150127
})
151128

152129
it('should handle RUN_ERROR events', async () => {
@@ -165,15 +142,62 @@ describe('toServerSentEventsStream', () => {
165142
const output = await readStream(sseStream)
166143

167144
expect(output).toContain('"type":"RUN_ERROR"')
168-
expect(output).toContain('data: [DONE]\n\n')
145+
expect(output).not.toContain('[DONE]')
169146
})
170147

171148
it('should handle empty stream', async () => {
172149
const stream = createMockStream([])
173150
const sseStream = toServerSentEventsStream(stream)
174151
const output = await readStream(sseStream)
175152

176-
expect(output).toBe('data: [DONE]\n\n')
153+
expect(output).toBe('')
154+
})
155+
156+
it('should not emit [DONE] sentinel — RUN_FINISHED is the stream terminator', async () => {
157+
const chunks: Array<Record<string, unknown>> = [
158+
{
159+
type: 'RUN_STARTED',
160+
runId: 'run-1',
161+
model: 'test',
162+
timestamp: Date.now(),
163+
},
164+
{
165+
type: 'TEXT_MESSAGE_START',
166+
messageId: 'msg-1',
167+
model: 'test',
168+
timestamp: Date.now(),
169+
role: 'assistant',
170+
},
171+
{
172+
type: 'TEXT_MESSAGE_CONTENT',
173+
messageId: 'msg-1',
174+
model: 'test',
175+
timestamp: Date.now(),
176+
delta: 'Hello',
177+
content: 'Hello',
178+
},
179+
{
180+
type: 'TEXT_MESSAGE_END',
181+
messageId: 'msg-1',
182+
model: 'test',
183+
timestamp: Date.now(),
184+
},
185+
{
186+
type: 'RUN_FINISHED',
187+
runId: 'run-1',
188+
model: 'test',
189+
timestamp: Date.now(),
190+
finishReason: 'stop',
191+
},
192+
]
193+
194+
const stream = createMockStream(chunks)
195+
const sseStream = toServerSentEventsStream(stream)
196+
const output = await readStream(sseStream)
197+
198+
expect(output).not.toContain('[DONE]')
199+
// Stream should end with the RUN_FINISHED event
200+
expect(output).toContain('"type":"RUN_FINISHED"')
177201
})
178202

179203
it('should abort when abortController signals abort', async () => {
@@ -294,8 +318,8 @@ describe('toServerSentEventsStream', () => {
294318
const dataLines = output
295319
.split('\n\n')
296320
.filter((line) => line.startsWith('data: '))
297-
expect(dataLines.length).toBeGreaterThanOrEqual(3) // At least 3 chunks + [DONE]
298-
expect(output).toContain('data: [DONE]\n\n')
321+
expect(dataLines.length).toBeGreaterThanOrEqual(3) // At least 3 chunks
322+
expect(output).not.toContain('[DONE]')
299323
})
300324
})
301325

@@ -423,7 +447,7 @@ describe('toServerSentEventsResponse', () => {
423447
expect(output).toContain('"type":"TEXT_MESSAGE_CONTENT"')
424448
expect(output).toContain('"delta":"Hello"')
425449
expect(output).toContain('"delta":" world"')
426-
expect(output).toContain('data: [DONE]\n\n')
450+
expect(output).not.toContain('[DONE]')
427451
})
428452

429453
it('should handle undefined init parameter', async () => {

0 commit comments

Comments
 (0)