diff --git a/.changeset/sly-candles-hide.md b/.changeset/sly-candles-hide.md new file mode 100644 index 0000000000..be720c0250 --- /dev/null +++ b/.changeset/sly-candles-hide.md @@ -0,0 +1,5 @@ +--- +"roo-cline": patch +--- + +Fix OpenAI Codex and OpenAI Native stream parsing for done-only and `content_part` events, including duplicate-text guards when deltas are already streamed. diff --git a/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts b/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts index 608f639ed4..0ac1e9b884 100644 --- a/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts +++ b/src/api/providers/__tests__/openai-codex-native-tool-calls.spec.ts @@ -97,4 +97,312 @@ describe("OpenAiCodexHandler native tool calls", () => { name: "attempt_completion", }) }) + + it("yields text when Codex emits assistant message only in response.output_item.done", async () => { + vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + ;(handler as any).client = { + responses: { + create: vi.fn().mockResolvedValue({ + async *[Symbol.asyncIterator]() { + yield { + type: "response.output_item.done", + item: { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "hello from spark" }], + }, + output_index: 0, + } + yield { + type: "response.completed", + response: { + id: "resp_done_only", + status: "completed", + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "hello from spark" }], + }, + ], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + } + }, + }), + }, + } + + const stream = handler.createMessage("system", [{ role: "user", content: "test" } as any], { + taskId: "t", + tools: [], + }) + + const chunks: any[] = [] + for await (const chunk of stream) { + chunks.push(chunk) + } + + const textChunks = chunks.filter((c) => c.type === "text") + expect(textChunks.length).toBeGreaterThan(0) + expect(textChunks.map((c) => c.text).join("")).toContain("hello from spark") + }) + + it("yields text when Codex emits assistant message only in response.completed output", async () => { + vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + ;(handler as any).client = { + responses: { + create: vi.fn().mockResolvedValue({ + async *[Symbol.asyncIterator]() { + yield { + type: "response.completed", + response: { + id: "resp_completed_only", + status: "completed", + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "final payload only" }], + }, + ], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + } + }, + }), + }, + } + + const stream = handler.createMessage("system", [{ role: "user", content: "test" } as any], { + taskId: "t", + tools: [], + }) + + const chunks: any[] = [] + for await (const chunk of stream) { + chunks.push(chunk) + } + + const textChunks = chunks.filter((c) => c.type === "text") + expect(textChunks.length).toBeGreaterThan(0) + expect(textChunks.map((c) => c.text).join("")).toContain("final payload only") + }) + + it("yields text when Codex emits response.output_text.done without deltas", async () => { + vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + ;(handler as any).client = { + responses: { + create: vi.fn().mockResolvedValue({ + async *[Symbol.asyncIterator]() { + yield { + type: "response.output_text.done", + text: "done-event text only", + } + yield { + type: "response.completed", + response: { + id: "resp_done_text_only", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + } + }, + }), + }, + } + + const stream = handler.createMessage("system", [{ role: "user", content: "test" } as any], { + taskId: "t", + tools: [], + }) + + const chunks: any[] = [] + for await (const chunk of stream) { + chunks.push(chunk) + } + + const textChunks = chunks.filter((c) => c.type === "text") + expect(textChunks.length).toBeGreaterThan(0) + expect(textChunks.map((c) => c.text).join("")).toContain("done-event text only") + }) + + it("yields tool_call when Codex emits function_call only in response.output_item.done", async () => { + vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + ;(handler as any).client = { + responses: { + create: vi.fn().mockResolvedValue({ + async *[Symbol.asyncIterator]() { + yield { + type: "response.output_item.done", + item: { + type: "function_call", + call_id: "call_done_only", + name: "attempt_completion", + arguments: '{"result":"ok"}', + }, + output_index: 0, + } + yield { + type: "response.completed", + response: { + id: "resp_done_tool_only", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + } + }, + }), + }, + } + + const stream = handler.createMessage("system", [{ role: "user", content: "test" } as any], { + taskId: "t", + tools: [], + }) + + const chunks: any[] = [] + for await (const chunk of stream) { + chunks.push(chunk) + } + + const toolCalls = chunks.filter((c) => c.type === "tool_call") + expect(toolCalls.length).toBeGreaterThan(0) + expect(toolCalls[0]).toMatchObject({ + type: "tool_call", + id: "call_done_only", + name: "attempt_completion", + }) + }) + + it("yields text when Codex emits response.content_part.added", async () => { + vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + ;(handler as any).client = { + responses: { + create: vi.fn().mockResolvedValue({ + async *[Symbol.asyncIterator]() { + yield { + type: "response.content_part.added", + part: { + type: "output_text", + text: "content part text", + }, + output_index: 0, + content_index: 0, + } + yield { + type: "response.completed", + response: { + id: "resp_content_part", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + } + }, + }), + }, + } + + const stream = handler.createMessage("system", [{ role: "user", content: "test" } as any], { + taskId: "t", + tools: [], + }) + + const chunks: any[] = [] + for await (const chunk of stream) { + chunks.push(chunk) + } + + const textChunks = chunks.filter((c) => c.type === "text") + expect(textChunks.length).toBeGreaterThan(0) + expect(textChunks.map((c) => c.text).join("")).toContain("content part text") + }) + + it("does not duplicate text when Codex emits delta and output_text.done", async () => { + vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + ;(handler as any).client = { + responses: { + create: vi.fn().mockResolvedValue({ + async *[Symbol.asyncIterator]() { + yield { type: "response.output_text.delta", delta: "hello " } + yield { type: "response.output_text.delta", delta: "world" } + yield { type: "response.output_text.done", text: "hello world" } + yield { + type: "response.completed", + response: { + id: "resp_delta_done", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + } + }, + }), + }, + } + + const stream = handler.createMessage("system", [{ role: "user", content: "test" } as any], { + taskId: "t", + tools: [], + }) + + const chunks: any[] = [] + for await (const chunk of stream) { + chunks.push(chunk) + } + + const textChunks = chunks.filter((c) => c.type === "text") + expect(textChunks.map((c) => c.text).join("")).toBe("hello world") + }) + + it("does not duplicate text when Codex emits delta and content_part.added", async () => { + vi.spyOn(openAiCodexOAuthManager, "getAccessToken").mockResolvedValue("test-token") + vi.spyOn(openAiCodexOAuthManager, "getAccountId").mockResolvedValue("acct_test") + ;(handler as any).client = { + responses: { + create: vi.fn().mockResolvedValue({ + async *[Symbol.asyncIterator]() { + yield { type: "response.output_text.delta", delta: "hello world" } + yield { + type: "response.content_part.added", + part: { type: "output_text", text: "hello world" }, + output_index: 0, + content_index: 0, + } + yield { + type: "response.completed", + response: { + id: "resp_delta_content_part", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + } + }, + }), + }, + } + + const stream = handler.createMessage("system", [{ role: "user", content: "test" } as any], { + taskId: "t", + tools: [], + }) + + const chunks: any[] = [] + for await (const chunk of stream) { + chunks.push(chunk) + } + + const textChunks = chunks.filter((c) => c.type === "text") + expect(textChunks.map((c) => c.text).join("")).toBe("hello world") + }) }) diff --git a/src/api/providers/__tests__/openai-native-tools.spec.ts b/src/api/providers/__tests__/openai-native-tools.spec.ts index e0746f792e..1e0e09c9a5 100644 --- a/src/api/providers/__tests__/openai-native-tools.spec.ts +++ b/src/api/providers/__tests__/openai-native-tools.spec.ts @@ -360,3 +360,212 @@ describe("OpenAiNativeHandler MCP tool schema handling", () => { }) }) }) + +describe("OpenAiNativeHandler done-event fallbacks", () => { + const createHandlerWithEvents = (events: any[]) => { + const handler = new OpenAiNativeHandler({ + openAiNativeApiKey: "test-key", + apiModelId: "gpt-4o", + } as ApiHandlerOptions) + + ;(handler as any).client = { + responses: { + create: vi.fn().mockResolvedValue({ + async *[Symbol.asyncIterator]() { + for (const event of events) { + yield event + } + }, + }), + }, + } + + return handler + } + + const collectChunksFromEvents = async (events: any[]) => { + const handler = createHandlerWithEvents(events) + const stream = handler.createMessage("system", [{ role: "user", content: "test" } as any], { + taskId: "t", + tools: [], + }) + + const chunks: any[] = [] + for await (const chunk of stream) { + chunks.push(chunk) + } + return chunks + } + + it.each([ + [ + "response.output_item.done message", + [ + { + type: "response.output_item.done", + item: { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "hello from done item" }], + }, + output_index: 0, + }, + { + type: "response.completed", + response: { + id: "resp_done_item_only", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + }, + ], + "hello from done item", + ], + [ + "response.completed output", + [ + { + type: "response.completed", + response: { + id: "resp_completed_only", + status: "completed", + output: [ + { + type: "message", + role: "assistant", + content: [{ type: "output_text", text: "final payload only" }], + }, + ], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + }, + ], + "final payload only", + ], + [ + "response.output_text.done", + [ + { + type: "response.output_text.done", + text: "done-event text only", + }, + { + type: "response.completed", + response: { + id: "resp_done_text_only", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + }, + ], + "done-event text only", + ], + [ + "response.content_part.added", + [ + { + type: "response.content_part.added", + part: { + type: "output_text", + text: "content part text", + }, + output_index: 0, + content_index: 0, + }, + { + type: "response.completed", + response: { + id: "resp_content_part", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + }, + ], + "content part text", + ], + ])("yields text when native emits %s", async (_caseName, events, expectedText) => { + const chunks = await collectChunksFromEvents(events) + const textChunks = chunks.filter((c) => c.type === "text") + expect(textChunks.length).toBeGreaterThan(0) + expect(textChunks.map((c) => c.text).join("")).toContain(expectedText) + }) + + it("yields tool_call when native emits function_call only in response.output_item.done", async () => { + const chunks = await collectChunksFromEvents([ + { + type: "response.output_item.done", + item: { + type: "function_call", + call_id: "call_done_only", + name: "attempt_completion", + arguments: '{"result":"ok"}', + }, + output_index: 0, + }, + { + type: "response.completed", + response: { + id: "resp_done_tool_only", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + }, + ]) + + const toolCalls = chunks.filter((c) => c.type === "tool_call") + expect(toolCalls.length).toBeGreaterThan(0) + expect(toolCalls[0]).toMatchObject({ + type: "tool_call", + id: "call_done_only", + name: "attempt_completion", + }) + }) + + it("does not duplicate text when delta and output_text.done are both emitted", async () => { + const chunks = await collectChunksFromEvents([ + { type: "response.output_text.delta", delta: "hello " }, + { type: "response.output_text.delta", delta: "world" }, + { type: "response.output_text.done", text: "hello world" }, + { + type: "response.completed", + response: { + id: "resp_delta_done", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + }, + ]) + + const textChunks = chunks.filter((c) => c.type === "text") + expect(textChunks.map((c) => c.text).join("")).toBe("hello world") + }) + + it("does not duplicate text when delta and content_part.added are both emitted", async () => { + const chunks = await collectChunksFromEvents([ + { type: "response.output_text.delta", delta: "hello world" }, + { + type: "response.content_part.added", + part: { type: "output_text", text: "hello world" }, + output_index: 0, + content_index: 0, + }, + { + type: "response.completed", + response: { + id: "resp_delta_content_part", + status: "completed", + output: [], + usage: { input_tokens: 1, output_tokens: 2 }, + }, + }, + ]) + + const textChunks = chunks.filter((c) => c.type === "text") + expect(textChunks.map((c) => c.text).join("")).toBe("hello world") + }) +}) diff --git a/src/api/providers/openai-codex.ts b/src/api/providers/openai-codex.ts index d64780c555..9dfb37bc72 100644 --- a/src/api/providers/openai-codex.ts +++ b/src/api/providers/openai-codex.ts @@ -64,11 +64,21 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion */ private pendingToolCallId: string | undefined private pendingToolCallName: string | undefined + // Tracks whether this response already emitted text to avoid duplicate done-event rendering. + private sawTextOutputInCurrentResponse = false + // Tracks whether text arrived through delta events so content_part events can be treated as fallback-only. + private sawTextDeltaInCurrentResponse = false + // Tracks tool call IDs emitted via streaming partial events to prevent done-event duplicates. + private streamedToolCallIds = new Set() // Event types handled by the shared event processor private readonly coreHandledEventTypes = new Set([ "response.text.delta", "response.output_text.delta", + "response.text.done", + "response.output_text.done", + "response.content_part.added", + "response.content_part.done", "response.reasoning.delta", "response.reasoning_text.delta", "response.reasoning_summary.delta", @@ -149,6 +159,9 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion this.lastResponseId = undefined this.pendingToolCallId = undefined this.pendingToolCallName = undefined + this.sawTextOutputInCurrentResponse = false + this.sawTextDeltaInCurrentResponse = false + this.streamedToolCallIds.clear() // Get access token from OAuth manager let accessToken = await openAiCodexOAuthManager.getAccessToken() @@ -378,6 +391,9 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } for await (const outChunk of this.processEvent(event, model)) { + if (outChunk.type === "text") { + this.sawTextOutputInCurrentResponse = true + } yield outChunk } } @@ -647,6 +663,9 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion for await (const outChunk of this.processEvent(parsed, model)) { if (outChunk.type === "text" || outChunk.type === "reasoning") { hasContent = true + if (outChunk.type === "text") { + this.sawTextOutputInCurrentResponse = true + } } yield outChunk } @@ -660,6 +679,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion for (const content of outputItem.content) { if (content.type === "text" && content.text) { hasContent = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: content.text } } } @@ -685,8 +705,26 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion ) { if (parsed.delta) { hasContent = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: parsed.delta } } + } else if ( + (parsed.type === "response.text.done" || parsed.type === "response.output_text.done") && + !hasContent + ) { + const doneText = + typeof parsed.text === "string" + ? parsed.text + : typeof parsed.output_text === "string" + ? parsed.output_text + : typeof parsed.delta === "string" + ? parsed.delta + : undefined + if (doneText) { + hasContent = true + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: doneText } + } } else if ( parsed.type === "response.reasoning.delta" || parsed.type === "response.reasoning_text.delta" @@ -706,12 +744,14 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } else if (parsed.type === "response.refusal.delta") { if (parsed.delta) { hasContent = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: `[Refusal] ${parsed.delta}` } } } else if (parsed.type === "response.output_item.added") { if (parsed.item) { if (parsed.item.type === "text" && parsed.item.text) { hasContent = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: parsed.item.text } } else if (parsed.item.type === "reasoning" && parsed.item.text) { hasContent = true @@ -720,6 +760,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion for (const content of parsed.item.content) { if (content.type === "text" && content.text) { hasContent = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: content.text } } } @@ -760,6 +801,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion for (const content of outputItem.content) { if (content.type === "output_text" && content.text) { hasContent = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: content.text } } } @@ -779,6 +821,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } } else if (parsed.choices?.[0]?.delta?.content) { hasContent = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: parsed.choices[0].delta.content } } else if ( parsed.item && @@ -786,6 +829,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion parsed.item.text.length > 0 ) { hasContent = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: parsed.item.text } } else if (parsed.usage) { const usageData = this.normalizeUsage(parsed.usage, model) @@ -803,6 +847,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion const parsed = JSON.parse(line) if (parsed.content || parsed.text || parsed.message) { hasContent = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: parsed.content || parsed.text || parsed.message } } } catch { @@ -836,11 +881,45 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // Handle text deltas if (event?.type === "response.text.delta" || event?.type === "response.output_text.delta") { if (event?.delta) { + this.sawTextDeltaInCurrentResponse = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: event.delta } } return } + if (event?.type === "response.text.done" || event?.type === "response.output_text.done") { + const doneText = + typeof event?.text === "string" + ? event.text + : typeof event?.output_text === "string" + ? event.output_text + : typeof event?.delta === "string" + ? event.delta + : undefined + if (!this.sawTextOutputInCurrentResponse && doneText) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: doneText } + } + return + } + + if (event?.type === "response.content_part.added" || event?.type === "response.content_part.done") { + const part = event?.part + if ( + !this.sawTextDeltaInCurrentResponse && + (part?.type === "text" || part?.type === "output_text") && + (typeof part?.text === "string" || typeof part?.text?.value === "string") + ) { + const partText = typeof part.text === "string" ? part.text : part.text.value + if (partText) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: partText } + } + } + return + } + // Handle reasoning deltas if ( event?.type === "response.reasoning.delta" || @@ -857,6 +936,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // Handle refusal deltas if (event?.type === "response.refusal.delta") { if (event?.delta) { + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: `[Refusal] ${event.delta}` } } return @@ -875,6 +955,7 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // to include a stable id/name. Avoid emitting incomplete tool_call_partial chunks because // NativeToolCallParser requires a name to start a call. if (typeof callId === "string" && callId.length > 0 && typeof name === "string" && name.length > 0) { + this.streamedToolCallIds.add(callId) yield { type: "tool_call_partial", index: event.index ?? 0, @@ -908,17 +989,64 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion } } - // For "added" events, yield text/reasoning content (streaming path) - // For "done" events, do NOT yield text/reasoning - it's already been streamed via deltas - // and would cause double-emission (A, B, C, ABC). + // For "added" events, yield text/reasoning content (streaming path). + // For "done" events, normally text was already streamed via deltas, but some models + // only provide assistant text on done events. Emit fallback text only if none was emitted yet. if (event.type === "response.output_item.added") { if (item.type === "text" && item.text) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: item.text } + } else if (item.type === "output_text" && item.text) { + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: item.text } } else if (item.type === "reasoning" && item.text) { yield { type: "reasoning", text: item.text } } else if (item.type === "message" && Array.isArray(item.content)) { for (const content of item.content) { if ((content?.type === "text" || content?.type === "output_text") && content?.text) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: content.text } + } + } + } + } else if ( + event.type === "response.output_item.done" && + (item.type === "function_call" || item.type === "tool_call") + ) { + const callId = item.call_id || item.tool_call_id || item.id + const name = item.name || item.function?.name || item.function_name + const argsRaw = item.arguments || item.function?.arguments || item.input + const args = + typeof argsRaw === "string" + ? argsRaw + : argsRaw && typeof argsRaw === "object" + ? JSON.stringify(argsRaw) + : "" + + // Fallback for models that only emit a complete function_call in output_item.done. + // If we already streamed partials for this ID, skip to avoid duplicate tool execution. + if ( + typeof callId === "string" && + callId.length > 0 && + typeof name === "string" && + name.length > 0 && + !this.streamedToolCallIds.has(callId) + ) { + yield { + type: "tool_call", + id: callId, + name, + arguments: args, + } + } + } else if (!this.sawTextOutputInCurrentResponse) { + if ((item.type === "text" || item.type === "output_text") && item.text) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: item.text } + } else if (item.type === "message" && Array.isArray(item.content)) { + for (const content of item.content) { + if ((content?.type === "text" || content?.type === "output_text") && content?.text) { + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: content.text } } } @@ -937,6 +1065,26 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // Handle completion events if (event?.type === "response.done" || event?.type === "response.completed") { + // Some Codex variants only provide assistant text in the final completed payload. + if (!this.sawTextOutputInCurrentResponse && Array.isArray(event?.response?.output)) { + for (const outputItem of event.response.output) { + if ((outputItem?.type === "text" || outputItem?.type === "output_text") && outputItem?.text) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: outputItem.text } + continue + } + + if (outputItem?.type === "message" && Array.isArray(outputItem.content)) { + for (const content of outputItem.content) { + if ((content?.type === "text" || content?.type === "output_text") && content?.text) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: content.text } + } + } + } + } + } + const usage = event?.response?.usage || event?.usage || undefined const usageData = this.normalizeUsage(usage, model) if (usageData) { @@ -947,6 +1095,8 @@ export class OpenAiCodexHandler extends BaseProvider implements SingleCompletion // Fallbacks if (event?.choices?.[0]?.delta?.content) { + this.sawTextDeltaInCurrentResponse = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: event.choices[0].delta.content } return } diff --git a/src/api/providers/openai-native.ts b/src/api/providers/openai-native.ts index d7c60c5daf..3dfad3ed35 100644 --- a/src/api/providers/openai-native.ts +++ b/src/api/providers/openai-native.ts @@ -45,6 +45,12 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio */ private pendingToolCallId: string | undefined private pendingToolCallName: string | undefined + // Tracks whether this response already emitted text to avoid duplicate done-event rendering. + private sawTextOutputInCurrentResponse = false + // Tracks whether text arrived through delta events so content_part events can be treated as fallback-only. + private sawTextDeltaInCurrentResponse = false + // Tracks tool call IDs emitted via streaming partial events to prevent done-event duplicates. + private streamedToolCallIds = new Set() // Resolved service tier from Responses API (actual tier used by OpenAI) private lastServiceTier: ServiceTier | undefined // Complete response output array (includes reasoning items with encrypted_content) @@ -58,6 +64,10 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio private readonly coreHandledEventTypes = new Set([ "response.text.delta", "response.output_text.delta", + "response.text.done", + "response.output_text.done", + "response.content_part.added", + "response.content_part.done", "response.reasoning.delta", "response.reasoning_text.delta", "response.reasoning_summary.delta", @@ -184,6 +194,9 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Reset pending tool identity for this request this.pendingToolCallId = undefined this.pendingToolCallName = undefined + this.sawTextOutputInCurrentResponse = false + this.sawTextDeltaInCurrentResponse = false + this.streamedToolCallIds.clear() // Use Responses API for ALL models const { verbosity, reasoning } = this.getModel() @@ -700,7 +713,8 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio this.lastResponseId = parsed.response.id as string } - // Delegate standard event types to the shared processor to avoid duplication + // Delegate standard event types to the shared processor to avoid duplication. + // This applies to both SDK and raw SSE fallback paths. if (parsed?.type && this.coreHandledEventTypes.has(parsed.type)) { for await (const outChunk of this.processEvent(parsed, model)) { // Track whether we've emitted any content so fallback handling can decide appropriately @@ -1051,13 +1065,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // For SSE path, usage often arrives separately; avoid double-emitting here. } // These are structural or status events, we can just log them at a lower level or ignore. - else if ( - parsed.type === "response.created" || - parsed.type === "response.in_progress" || - parsed.type === "response.output_item.done" || - parsed.type === "response.content_part.added" || - parsed.type === "response.content_part.done" - ) { + else if (parsed.type === "response.created" || parsed.type === "response.in_progress") { // Status events - no action needed } // Fallback for older formats or unexpected responses @@ -1146,14 +1154,50 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio this.lastResponseId = event.response.id as string } - // Handle known streaming text deltas + // Handle text deltas if (event?.type === "response.text.delta" || event?.type === "response.output_text.delta") { if (event?.delta) { + this.sawTextDeltaInCurrentResponse = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: event.delta } } return } + // Handle done-only text for variants that skip delta events. + if (event?.type === "response.text.done" || event?.type === "response.output_text.done") { + const doneText = + typeof event?.text === "string" + ? event.text + : typeof event?.output_text === "string" + ? event.output_text + : typeof event?.delta === "string" + ? event.delta + : undefined + if (!this.sawTextOutputInCurrentResponse && doneText) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: doneText } + } + return + } + + // Handle content-part text for structured streaming payloads. + if (event?.type === "response.content_part.added" || event?.type === "response.content_part.done") { + const part = event?.part + if ( + !this.sawTextDeltaInCurrentResponse && + (part?.type === "text" || part?.type === "output_text") && + (typeof part?.text === "string" || typeof part?.text?.value === "string") + ) { + const partText = typeof part.text === "string" ? part.text : part.text.value + if (partText) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: partText } + } + } + return + } + // Handle reasoning deltas (including summary variants) if ( event?.type === "response.reasoning.delta" || @@ -1170,6 +1214,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Handle refusal deltas if (event?.type === "response.refusal.delta") { if (event?.delta) { + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: `[Refusal] ${event.delta}` } } return @@ -1189,6 +1234,7 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Avoid emitting incomplete tool_call_partial chunks; the downstream // NativeToolCallParser needs a name to start a call. if (typeof name === "string" && name.length > 0 && typeof callId === "string" && callId.length > 0) { + this.streamedToolCallIds.add(callId) yield { type: "tool_call_partial", index: event.index ?? 0, @@ -1223,11 +1269,15 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio } } - // For "added" events, yield text/reasoning content (streaming path) - // For "done" events, do NOT yield text/reasoning - it's already been streamed via deltas - // and would cause double-emission (A, B, C, ABC). + // For "added" events, yield text/reasoning content (streaming path). + // For "done" events, normally text was already streamed via deltas, but some models + // only provide assistant text on done events. Emit fallback text only if none was emitted yet. if (event.type === "response.output_item.added") { if (item.type === "text" && item.text) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: item.text } + } else if (item.type === "output_text" && item.text) { + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: item.text } } else if (item.type === "reasoning" && item.text) { yield { type: "reasoning", text: item.text } @@ -1235,6 +1285,49 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio for (const content of item.content) { // Some implementations send 'text'; others send 'output_text' if ((content?.type === "text" || content?.type === "output_text") && content?.text) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: content.text } + } + } + } + } else if ( + event.type === "response.output_item.done" && + (item.type === "function_call" || item.type === "tool_call") + ) { + const callId = item.call_id || item.tool_call_id || item.id + const name = item.name || item.function?.name || item.function_name + const argsRaw = item.arguments || item.function?.arguments || item.input + const args = + typeof argsRaw === "string" + ? argsRaw + : argsRaw && typeof argsRaw === "object" + ? JSON.stringify(argsRaw) + : "" + + // Fallback for models that only emit a complete function_call in output_item.done. + // If we already streamed partials for this ID, skip to avoid duplicate tool execution. + if ( + typeof callId === "string" && + callId.length > 0 && + typeof name === "string" && + name.length > 0 && + !this.streamedToolCallIds.has(callId) + ) { + yield { + type: "tool_call", + id: callId, + name, + arguments: args, + } + } + } else if (!this.sawTextOutputInCurrentResponse) { + if ((item.type === "text" || item.type === "output_text") && item.text) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: item.text } + } else if (item.type === "message" && Array.isArray(item.content)) { + for (const content of item.content) { + if ((content?.type === "text" || content?.type === "output_text") && content?.text) { + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: content.text } } } @@ -1242,17 +1335,33 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio } // Note: We intentionally do NOT emit tool_call from response.output_item.done - // for function_call/tool_call items. The streaming path handles tool calls via: - // 1. tool_call_partial events during argument deltas - // 2. NativeToolCallParser.finalizeRawChunks() at stream end emitting tool_call_end - // 3. NativeToolCallParser.finalizeStreamingToolCall() creating the final ToolUse - // Emitting tool_call here would cause duplicate tool rendering. + // for function_call/tool_call items if we already saw streaming partials. } return } // Completion events that may carry usage if (event?.type === "response.done" || event?.type === "response.completed") { + // Some OpenAI variants only provide assistant text in the final completed payload. + if (!this.sawTextOutputInCurrentResponse && Array.isArray(event?.response?.output)) { + for (const outputItem of event.response.output) { + if ((outputItem?.type === "text" || outputItem?.type === "output_text") && outputItem?.text) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: outputItem.text } + continue + } + + if (outputItem?.type === "message" && Array.isArray(outputItem.content)) { + for (const content of outputItem.content) { + if ((content?.type === "text" || content?.type === "output_text") && content?.text) { + this.sawTextOutputInCurrentResponse = true + yield { type: "text", text: content.text } + } + } + } + } + } + const usage = event?.response?.usage || event?.usage || undefined const usageData = this.normalizeUsage(usage, model) if (usageData) { @@ -1263,6 +1372,8 @@ export class OpenAiNativeHandler extends BaseProvider implements SingleCompletio // Fallbacks for older formats or unexpected objects if (event?.choices?.[0]?.delta?.content) { + this.sawTextDeltaInCurrentResponse = true + this.sawTextOutputInCurrentResponse = true yield { type: "text", text: event.choices[0].delta.content } return }