fix(console): emit complete Responses-API SSE lifecycle for converted streams

packages/console/app/src/routes/zen/util/provider/openai-compatible.ts

@@ -1,4 +1,4 @@
-import { ProviderHelper, CommonRequest, CommonResponse, CommonChunk } from "./provider"
+import type { ProviderHelper, CommonRequest, CommonResponse, CommonChunk } from "./provider"
 
 type Usage = {
   prompt_tokens?: number

packages/console/app/src/routes/zen/util/provider/openai.ts

@@ -1,4 +1,4 @@
-import { ProviderHelper, CommonRequest, CommonResponse, CommonChunk } from "./provider"
+import type { ProviderHelper, CommonRequest, CommonResponse, CommonChunk } from "./provider"
 
 type Usage = {
   input_tokens?: number
@@ -556,75 +556,160 @@ export function fromOpenaiChunk(chunk: string): CommonChunk | string {
   return out
 }
 
-export function toOpenaiChunk(chunk: CommonChunk): string {
-  if (!chunk.choices || !Array.isArray(chunk.choices) || chunk.choices.length === 0) {
-    return ""
-  }
+export function createToOpenaiChunk() {
+  let responseId = ""
+  let itemId = ""
+  let started = false
+  let text = ""
+  let itemAdded = false
+  let partAdded = false
+  let itemDone = false
+  let toolCallId = ""
+  let toolName = ""
+  let toolArguments = ""
+
+  const messageItem = (status: string) => ({
+    id: itemId,
+    type: "message",
+    status,
+    role: "assistant",
+    content: partAdded ? [{ type: "output_text", text, annotations: [] }] : [],
+  })
+
+  const toolCallItem = () => ({
+    id: toolCallId,
+    type: "function_call",
+    call_id: toolCallId,
+    name: toolName,
+    arguments: toolArguments,
+  })
+
+  return (chunk: CommonChunk): string => {
+    if (!chunk.choices || !Array.isArray(chunk.choices) || chunk.choices.length === 0) {
+      return ""
+    }
 
-  const choice = chunk.choices[0]
-  const d = choice.delta
-  if (!d) return ""
+    const choice = chunk.choices[0]
+    const d = choice.delta
+    if (!d) return ""
 
-  const id = chunk.id
-  const model = chunk.model
+    const events: string[] = []
+    const emit = (event: string, data: Record<string, unknown>) => {
+      events.push(`event: ${event}\ndata: ${JSON.stringify(data)}`)
+    }
 
-  if (d.content) {
-    const data = {
-      id,
-      type: "response.output_text.delta",
-      delta: d.content,
-      response: { id, model },
+    if (!started) {
+      started = true
+      responseId = chunk.id || `resp_${Math.random().toString(36).slice(2)}`
+      itemId = `msg_${Math.random().toString(36).slice(2)}`
+      const response = { id: responseId, object: "response", status: "in_progress", model: chunk.model, output: [] }
+      emit("response.created", { type: "response.created", response })
+      emit("response.in_progress", { type: "response.in_progress", response })
     }
-    return `event: response.output_text.delta\ndata: ${JSON.stringify(data)}`
-  }
 
-  if (d.tool_calls) {
-    for (const tc of d.tool_calls) {
-      if (tc.function?.name) {
-        const data = {
+    if (d.content) {
+      if (!itemAdded) {
+        itemAdded = true
+        emit("response.output_item.added", {
           type: "response.output_item.added",
           output_index: 0,
-          item: {
-            id: tc.id,
-            type: "function_call",
-            name: tc.function.name,
-            call_id: tc.id,
-            arguments: "",
-          },
-        }
-        return `event: response.output_item.added\ndata: ${JSON.stringify(data)}`
+          item: messageItem("in_progress"),
+        })
       }
-      if (tc.function?.arguments) {
-        const data = {
-          type: "response.function_call_arguments.delta",
+      if (!partAdded) {
+        partAdded = true
+        emit("response.content_part.added", {
+          type: "response.content_part.added",
           output_index: 0,
-          delta: tc.function.arguments,
+          item_id: itemId,
+          part: { type: "output_text", text: "", annotations: [] },
+        })
+      }
+      text += d.content
+      emit("response.output_text.delta", {
+        id: responseId,
+        type: "response.output_text.delta",
+        delta: d.content,
+        response: { id: responseId, model: chunk.model },
+      })
+    }
+
+    if (d.tool_calls) {
+      for (const tc of d.tool_calls) {
+        if (!tc || tc.type !== "function") continue
+        if (tc.function?.name) {
+          toolCallId = tc.id || `call_${Math.random().toString(36).slice(2)}`
+          toolName = tc.function.name
+          itemAdded = true
+          emit("response.output_item.added", {
+            type: "response.output_item.added",
+            output_index: 0,
+            item: toolCallItem(),
+          })
+        }
+        if (tc.function?.arguments) {
+          toolArguments += tc.function.arguments
+          emit("response.function_call_arguments.delta", {
+            type: "response.function_call_arguments.delta",
+            output_index: 0,
+            item_id: toolCallId,
+            delta: tc.function.arguments,
+          })
         }
-        return `event: response.function_call_arguments.delta\ndata: ${JSON.stringify(data)}`
       }
     }
-  }
 
-  if (choice.finish_reason) {
-    const u = chunk.usage
-    const usage = u
-      ? {
-          input_tokens: u.prompt_tokens,
-          output_tokens: u.completion_tokens,
-          total_tokens: u.total_tokens,
-          ...(u.prompt_tokens_details?.cached_tokens
-            ? { input_tokens_details: { cached_tokens: u.prompt_tokens_details.cached_tokens } }
-            : {}),
+    if (choice.finish_reason) {
+      if (itemAdded && !itemDone) {
+        itemDone = true
+        if (toolCallId) {
+          emit("response.function_call_arguments.done", {
+            type: "response.function_call_arguments.done",
+            output_index: 0,
+            item_id: toolCallId,
+            arguments: toolArguments,
+          })
+          emit("response.output_item.done", {
+            type: "response.output_item.done",
+            output_index: 0,
+            item: toolCallItem(),
+          })
+        } else {
+          emit("response.output_item.done", {
+            type: "response.output_item.done",
+            output_index: 0,
+            item: messageItem("completed"),
+          })
         }
-      : undefined
+      }
 
-    const data: any = {
-      id,
-      type: "response.completed",
-      response: { id, model, ...(usage ? { usage } : {}) },
+      const u = chunk.usage
+      const usage = u
+        ? {
+            input_tokens: u.prompt_tokens,
+            output_tokens: u.completion_tokens,
+            total_tokens: u.total_tokens,
+            ...(u.prompt_tokens_details?.cached_tokens
+              ? { input_tokens_details: { cached_tokens: u.prompt_tokens_details.cached_tokens } }
+              : {}),
+          }
+        : undefined
+
+      const stop_reason = (() => {
+        const r = choice.finish_reason
+        if (r === "stop") return "stop"
+        if (r === "tool_calls") return "tool_call"
+        if (r === "length") return "max_output_tokens"
+        if (r === "content_filter") return "content_filter"
+        return null
+      })()
+
+      const output = toolCallId ? [toolCallItem()] : itemAdded ? [messageItem("completed")] : []
+      const response: any = { id: responseId, model: chunk.model, output, stop_reason }
+      if (usage) response.usage = usage
+      emit("response.completed", { id: responseId, type: "response.completed", response })
     }
-    return `event: response.completed\ndata: ${JSON.stringify(data)}`
-  }
 
-  return ""
+    return events.join("\n\n")
+  }
 }

packages/console/app/src/routes/zen/util/provider/provider.ts

@@ -8,10 +8,10 @@ import {
   toAnthropicResponse,
 } from "./anthropic"
 import {
+  createToOpenaiChunk,
   fromOpenaiChunk,
   fromOpenaiRequest,
   fromOpenaiResponse,
-  toOpenaiChunk,
   toOpenaiRequest,
   toOpenaiResponse,
 } from "./openai"
@@ -195,6 +195,7 @@ export function createBodyConverter(from: ZenData.Format, to: ZenData.Format) {
 }
 
 export function createStreamPartConverter(from: ZenData.Format, to: ZenData.Format) {
+  const toOpenaiChunk = to === "openai" ? createToOpenaiChunk() : undefined
   return (part: any): any => {
     if (from === to) return part
 
@@ -207,7 +208,7 @@ export function createStreamPartConverter(from: ZenData.Format, to: ZenData.Form
     if (typeof raw === "string") return raw
 
     if (to === "anthropic") return toAnthropicChunk(raw)
-    if (to === "openai") return toOpenaiChunk(raw)
+    if (to === "openai") return toOpenaiChunk!(raw)
     if (to === "oa-compat") return toOaCompatibleChunk(raw)
   }
 }

packages/console/app/test/openaiResponsesStream.test.ts

@@ -0,0 +1,105 @@
+import { describe, expect, test } from "bun:test"
+import { createToOpenaiChunk } from "../src/routes/zen/util/provider/openai"
+import { fromOaCompatibleChunk } from "../src/routes/zen/util/provider/openai-compatible"
+
+function convert(chunks: string[]) {
+  const toOpenai = createToOpenaiChunk()
+  return chunks
+    .map((part) => {
+      const raw = fromOaCompatibleChunk(part)
+      return typeof raw === "string" ? raw : toOpenai(raw)
+    })
+    .filter((part) => part.length > 0)
+    .join("\n\n")
+}
+
+const chunk = (delta: string, finishReason?: string, usage?: Record<string, unknown>) =>
+  JSON.stringify({
+    id: "chatcmpl-1",
+    object: "chat.completion.chunk",
+    created: 1,
+    model: "deepseek-v4-flash",
+    choices: [{ index: 0, delta: JSON.parse(delta), finish_reason: finishReason ?? null }],
+    ...(usage ? { usage } : {}),
+  })
+
+const eventOrder = (stream: string) => [...stream.matchAll(/event: ([a-z_.]+)/g)].map((m) => m[1])
+
+const eventData = (stream: string, event: string) => {
+  const match = stream.match(new RegExp(`event: ${event}\\ndata: (\\{.*?\\})(?:\\n\\n|$)`))
+  if (!match) throw new Error(`missing ${event}`)
+  return JSON.parse(match[1])
+}
+
+describe("createToOpenaiChunk", () => {
+  test("text stream emits the full Responses-API lifecycle in order", () => {
+    const stream = convert([
+      `data: ${chunk('{"role":"assistant","content":""}')}`,
+      `data: ${chunk('{"content":"ok"}')}`,
+      `data: ${chunk('{"content":"!"}')}`,
+      `data: ${chunk("{}", "stop", { prompt_tokens: 10, completion_tokens: 3, total_tokens: 13 })}`,
+      "data: [DONE]",
+    ])
+
+    expect(eventOrder(stream)).toEqual([
+      "response.created",
+      "response.in_progress",
+      "response.output_item.added",
+      "response.content_part.added",
+      "response.output_text.delta",
+      "response.output_text.delta",
+      "response.output_item.done",
+      "response.completed",
+    ])
+    expect((stream.match(/event: response\.created/g) ?? []).length).toBe(1)
+    expect(stream).toContain("data: [DONE]")
+
+    const item = eventData(stream, "response.output_item.done").item
+    expect(item).toMatchObject({ type: "message", status: "completed", role: "assistant" })
+    expect(item.content[0].text).toBe("ok!")
+
+    const completed = eventData(stream, "response.completed").response
+    expect(completed.output[0].content[0].text).toBe("ok!")
+    expect(completed.usage).toEqual({
+      input_tokens: 10,
+      output_tokens: 3,
+      total_tokens: 13,
+    })
+  })
+
+  test("[DONE] passes through unchanged", () => {
+    expect(convert(["data: [DONE]"])).toBe("data: [DONE]")
+  })
+
+  test("tool call stream emits the function-call lifecycle", () => {
+    const stream = convert([
+      `data: ${chunk('{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"get_weather","arguments":""}}]}')}`,
+      `data: ${chunk('{"tool_calls":[{"index":0,"function":{"arguments":"{\\"city\\":"}}]}')}`,
+      `data: ${chunk('{"tool_calls":[{"index":0,"function":{"arguments":"\\"Sydney\\""}}]}')}`,
+      `data: ${chunk('{"tool_calls":[{"index":0,"function":{"arguments":"}"}}]}')}`,
+      `data: ${chunk("{}", "tool_calls")}`,
+    ])
+
+    expect(eventOrder(stream)).toEqual([
+      "response.created",
+      "response.in_progress",
+      "response.output_item.added",
+      "response.function_call_arguments.delta",
+      "response.function_call_arguments.delta",
+      "response.function_call_arguments.delta",
+      "response.function_call_arguments.done",
+      "response.output_item.done",
+      "response.completed",
+    ])
+
+    const item = eventData(stream, "response.output_item.done").item
+    expect(item).toMatchObject({ type: "function_call", name: "get_weather" })
+    expect(item.arguments).toBe('{"city":"Sydney"}')
+  })
+
+  test("stream with no visible text still opens and closes the response", () => {
+    const stream = convert([`data: ${chunk("{}", "stop")}`])
+    expect(eventOrder(stream)).toEqual(["response.created", "response.in_progress", "response.completed"])
+    expect(eventData(stream, "response.completed").response.output).toEqual([])
+  })
+})