diff --git a/extensions/openai/realtime-voice-provider.test.ts b/extensions/openai/realtime-voice-provider.test.ts index 51705a4088ea..9a7fe0d4b4db 100644 --- a/extensions/openai/realtime-voice-provider.test.ts +++ b/extensions/openai/realtime-voice-provider.test.ts @@ -236,6 +236,27 @@ function emitSessionUpdated(socket: FakeWebSocketInstance): void { emitServerEvent(socket, { type: "session.updated" }); } +function emitCompletedToolCalls( + socket: FakeWebSocketInstance, + callIds: string[] = ["call_1"], +): void { + emitServerEvent(socket, { + type: "response.done", + response: { + id: "response_tools", + status: "completed", + output: callIds.map((callId, index) => ({ + id: `item_${index + 1}`, + type: "function_call", + status: "completed", + call_id: callId, + name: "lookup_weather", + arguments: "{}", + })), + }, + }); +} + async function connectReadyBridge( bridge: RealtimeVoiceBridge, socketIndex = 0, @@ -2649,84 +2670,56 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { expect(onTranscript).toHaveBeenCalledWith("assistant", "final assistant text", true); }); - it("emits tool calls from realtime conversation item done events", async () => { - const onToolCall = vi.fn(); - const onEvent = vi.fn(); - const bridge = createNativeBridge({ onToolCall, onEvent }); - const socket = await connectReadyBridge(bridge); - - socket.emit( - "message", - Buffer.from( - JSON.stringify({ - type: "conversation.item.done", - item: { - id: "item_tool_1", - type: "function_call", - name: "openclaw_agent_consult", - call_id: "call_1", - arguments: JSON.stringify({ question: "delegate this" }), - }, - }), - ), - ); - - expect(onToolCall).toHaveBeenCalledWith({ - itemId: "item_tool_1", - callId: "call_1", - name: "openclaw_agent_consult", - args: { question: "delegate this" }, - }); - expect(onEvent).toHaveBeenCalledWith({ - direction: "server", - type: "conversation.item.done", - detail: "function_call name=openclaw_agent_consult", - }); - }); - - it("deduplicates tool calls reported by arguments done and item done events", async () => { + it("executes tool calls only from successful response output", async () => { const onToolCall = vi.fn(); const bridge = createNativeBridge({ onToolCall }); const socket = await connectReadyBridge(bridge); - socket.emit( - "message", - Buffer.from( - JSON.stringify({ - type: "response.function_call_arguments.delta", - item_id: "item_tool_1", - name: "openclaw_agent_consult", - call_id: "call_1", - delta: JSON.stringify({ question: "delegate this" }), - }), - ), - ); - socket.emit( - "message", - Buffer.from( - JSON.stringify({ - type: "response.function_call_arguments.done", - item_id: "item_tool_1", - name: "openclaw_agent_consult", - call_id: "call_1", - }), - ), - ); - socket.emit( - "message", - Buffer.from( - JSON.stringify({ - type: "conversation.item.done", - item: { + emitServerEvent(socket, { + type: "response.function_call_arguments.delta", + item_id: "item_tool_1", + name: "openclaw_agent_consult", + call_id: "call_1", + delta: '{"question":"provisional', + }); + emitServerEvent(socket, { + type: "response.function_call_arguments.done", + item_id: "item_tool_1", + name: "openclaw_agent_consult", + call_id: "call_1", + arguments: '{"question":"still provisional"}', + }); + emitServerEvent(socket, { + type: "conversation.item.done", + item: { + id: "item_tool_1", + type: "function_call", + name: "openclaw_agent_consult", + call_id: "call_1", + arguments: '{"question":"not terminal"}', + }, + }); + expect(onToolCall).not.toHaveBeenCalled(); + + const completed = { + type: "response.done", + response: { + id: "response_1", + status: "completed", + output: [ + { id: "item_tool_1", type: "function_call", + status: "completed", name: "openclaw_agent_consult", call_id: "call_1", - arguments: JSON.stringify({ question: "delegate this" }), + arguments: '{"question":"delegate this"}', }, - }), - ), - ); + ], + }, + }; + emitServerEvent(socket, completed); + emitServerEvent(socket, completed); expect(onToolCall).toHaveBeenCalledTimes(1); expect(onToolCall).toHaveBeenCalledWith({ @@ -2737,56 +2730,98 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { }); }); - it.each([ - { - name: "corrected streamed arguments", - delta: '{"city":"draft"}', - finalArguments: '{"city":"Paris"}', - expectedArguments: { city: "Paris" }, - }, - { - name: "truncated streamed arguments", - delta: '{"city":', - finalArguments: '{"city":"Paris"}', - expectedArguments: { city: "Paris" }, - }, - { - name: "an explicitly empty completed payload", - delta: '{"city":"draft"}', - finalArguments: "", - expectedArguments: {}, - }, - ])( - "uses authoritative completed tool arguments for $name", - async ({ delta, finalArguments, expectedArguments }) => { + it.each(["cancelled", "failed", "incomplete"])( + "ignores function calls from a %s response", + async (status) => { const onToolCall = vi.fn(); const bridge = createNativeBridge({ onToolCall }); const socket = await connectReadyBridge(bridge); - socket.emit( - "message", - Buffer.from( - JSON.stringify({ - type: "response.function_call_arguments.delta", - item_id: "item_tool_1", + emitServerEvent(socket, { + type: "response.done", + response: { + id: "response_1", + status, + output: [ + { + id: "item_tool_1", + type: "function_call", + status: "completed", + name: "openclaw_agent_consult", + call_id: "call_1", + arguments: '{"question":"must stay inert"}', + }, + ], + }, + }); + + expect(onToolCall).not.toHaveBeenCalled(); + }, + ); + + it("ignores malformed and unfinished response output items", async () => { + const onToolCall = vi.fn(); + const bridge = createNativeBridge({ onToolCall }); + const socket = await connectReadyBridge(bridge); + + emitServerEvent(socket, { + type: "response.done", + response: { + id: "response_1", + status: "completed", + output: [ + null, + "invalid", + { + id: "item_tool_1", + type: "function_call", + status: "incomplete", + name: "openclaw_agent_consult", call_id: "call_1", - name: "lookup_weather", - delta, - }), - ), - ); - socket.emit( - "message", - Buffer.from( - JSON.stringify({ - type: "response.function_call_arguments.done", - item_id: "item_tool_1", - call_id: "call_1", - name: "lookup_weather", - arguments: finalArguments, - }), - ), - ); + arguments: '{"question":"unfinished"}', + }, + ], + }, + }); + + expect(onToolCall).not.toHaveBeenCalled(); + }); + + it.each([ + { + name: "an argument object", + finalArguments: '{"city":"Paris"}', + expectedArguments: { city: "Paris" }, + }, + { + name: "the shipped empty argument contract", + finalArguments: "", + expectedArguments: {}, + }, + ])( + "uses terminal response arguments for $name", + async ({ finalArguments, expectedArguments }) => { + const onToolCall = vi.fn(); + const bridge = createNativeBridge({ onToolCall }); + const socket = await connectReadyBridge(bridge); + + emitServerEvent(socket, { + type: "response.done", + response: { + id: "response_1", + status: "completed", + output: [ + { + id: "item_tool_1", + type: "function_call", + status: "completed", + call_id: "call_1", + name: "lookup_weather", + arguments: finalArguments, + }, + ], + }, + }); expect(onToolCall).toHaveBeenCalledWith({ itemId: "item_tool_1", @@ -2797,6 +2832,196 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { }, ); + it.each([ + { name: "malformed JSON", arguments: '{"city":', reason: "malformed-json" }, + { name: "an array", arguments: '["Paris"]', reason: "non-object-json" }, + { name: "JSON null", arguments: "null", reason: "non-object-json" }, + { name: "a number", arguments: "42", reason: "non-object-json" }, + { name: "a boolean", arguments: "true", reason: "non-object-json" }, + { name: "missing arguments", arguments: undefined, reason: "invalid-json-type" }, + { name: "non-string arguments", arguments: { city: "Paris" }, reason: "invalid-json-type" }, + ])("rejects $name per call without ending the session", async ({ arguments: args, reason }) => { + const onToolCall = vi.fn(); + const onError = vi.fn(); + const onEvent = vi.fn(); + const bridge = createNativeBridge({ onToolCall, onError, onEvent }); + const socket = await connectReadyBridge(bridge); + const completed = { + type: "response.done", + response: { + id: "response_1", + status: "completed", + output: [ + { + id: "item_tool_1", + type: "function_call", + status: "completed", + call_id: "call_1", + name: "lookup_weather", + arguments: args, + }, + ], + }, + }; + + emitServerEvent(socket, { type: "response.created", response: { id: "response_1" } }); + emitServerEvent(socket, completed); + emitServerEvent(socket, completed); + + expect(onToolCall).not.toHaveBeenCalled(); + expect(onError).not.toHaveBeenCalled(); + expect(onEvent).toHaveBeenCalledWith({ + direction: "server", + type: "tool_call.arguments.rejected", + detail: `reason=${reason}`, + itemId: "item_tool_1", + }); + expect( + parseSent(socket).filter( + (event) => + event.type === "conversation.item.create" && + (event.item as { call_id?: string } | undefined)?.call_id === "call_1", + ), + ).toHaveLength(1); + }); + + it.each([ + { + name: "accepts", + encoding: "ASCII", + argumentBytes: 256_000, + unit: "a", + repeat: 255_992, + suffix: "", + rejected: false, + }, + { + name: "rejects", + encoding: "ASCII", + argumentBytes: 256_001, + unit: "a", + repeat: 255_993, + suffix: "", + rejected: true, + }, + { + name: "accepts", + encoding: "multibyte", + argumentBytes: 256_000, + unit: "é", + repeat: 127_996, + suffix: "", + rejected: false, + }, + { + name: "rejects", + encoding: "multibyte", + argumentBytes: 256_001, + unit: "é", + repeat: 127_996, + suffix: "a", + rejected: true, + }, + ])( + "$name $argumentBytes-byte $encoding UTF-8 arguments", + async ({ argumentBytes, unit, repeat, suffix, rejected }) => { + const onToolCall = vi.fn(); + const onError = vi.fn(); + const bridge = createNativeBridge({ onToolCall, onError }); + const socket = await connectReadyBridge(bridge); + const rawArgs = `{"x":"${unit.repeat(repeat)}${suffix}"}`; + expect(Buffer.byteLength(rawArgs, "utf8")).toBe(argumentBytes); + + emitServerEvent(socket, { + type: "response.done", + response: { + id: "response_1", + status: "completed", + output: [ + { + id: "item_tool_1", + type: "function_call", + status: "completed", + call_id: "call_1", + name: "lookup_weather", + arguments: rawArgs, + }, + ], + }, + }); + + expect(onToolCall).toHaveBeenCalledTimes(rejected ? 0 : 1); + expect( + parseSent(socket).some( + (event) => + event.type === "conversation.item.create" && + (event.item as { call_id?: string } | undefined)?.call_id === "call_1", + ), + ).toBe(rejected); + expect(onError).not.toHaveBeenCalled(); + }, + ); + + it("ends an extreme session before terminal tool-call ids become unbounded", async () => { + const onToolCall = vi.fn(); + const onError = vi.fn(); + const onClose = vi.fn(); + const bridge = createNativeBridge({ onToolCall, onError, onClose }); + const socket = await connectReadyBridge(bridge); + + emitServerEvent(socket, { + type: "response.done", + response: { + id: "response_1", + status: "completed", + output: Array.from({ length: 1_025 }, (_, index) => ({ + id: `item_${index}`, + type: "function_call", + status: "completed", + call_id: `call_${index}`, + name: "lookup_weather", + arguments: "{}", + })), + }, + }); + + expect(onToolCall).toHaveBeenCalledTimes(1_024); + expect(onError).toHaveBeenCalledOnce(); + expect(onError).toHaveBeenCalledWith( + new Error("OpenAI realtime tool-call session limit exceeded (1024)"), + ); + expect(onClose).toHaveBeenCalledWith("error"); + expect(socket.closed).toBe(true); + await expect(bridge.connect()).rejects.toThrow( + "OpenAI realtime tool-call session limit exceeded (1024)", + ); + }); + + it("stops dispatching terminal output when a tool callback closes the bridge", async () => { + const onToolCall = vi.fn(); + const bridge = createNativeBridge({ onToolCall }); + onToolCall.mockImplementation(() => bridge.close()); + const socket = await connectReadyBridge(bridge); + + emitServerEvent(socket, { + type: "response.done", + response: { + id: "response_1", + status: "completed", + output: Array.from({ length: 2 }, (_, index) => ({ + id: `item_${index}`, + type: "function_call", + status: "completed", + call_id: `call_${index}`, + name: "lookup_weather", + arguments: "{}", + })), + }, + }); + + expect(onToolCall).toHaveBeenCalledTimes(1); + }); + it("creates an explicit user item and response for manual speech", async () => { const onEvent = vi.fn(); const bridge = createNativeBridge({ onEvent }); @@ -2851,15 +3076,15 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { Buffer.from(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })), ); - void bridge.submitToolResult("call_1", { text: "done" }); + bridge.sendUserMessage?.("queued manual response"); expect(parseSent(socket).slice(-1)).toEqual([ { type: "conversation.item.create", item: { - type: "function_call_output", - call_id: "call_1", - output: JSON.stringify({ text: "done" }), + type: "message", + role: "user", + content: [{ type: "input_text", text: "queued manual response" }], }, }, ]); @@ -3005,8 +3230,9 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { }); it("does not request a realtime response for continuing tool results", async () => { - const bridge = createNativeBridge(); + const bridge = createNativeBridge({ onToolCall: vi.fn() }); const socket = await connectReadyBridge(bridge); + emitCompletedToolCalls(socket); void bridge.submitToolResult("call_1", { status: "working" }, { willContinue: true }); @@ -3046,8 +3272,9 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { }); it("does not request a realtime response for suppressed tool results", async () => { - const bridge = createNativeBridge(); + const bridge = createNativeBridge({ onToolCall: vi.fn() }); const socket = await connectReadyBridge(bridge); + emitCompletedToolCalls(socket); void bridge.submitToolResult( "call_1", @@ -3068,17 +3295,80 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { expect(hasSentEventType(socket, "response.create")).toBe(false); }); + it("waits for every parallel tool result before continuing the response", async () => { + const bridge = createNativeBridge({ onToolCall: vi.fn() }); + const socket = await connectReadyBridge(bridge); + emitCompletedToolCalls(socket, ["call_1", "call_2"]); + + void bridge.submitToolResult("call_1", { text: "first" }); + + expect(parseSent(socket).filter((event) => event.type === "response.create")).toEqual([]); + + void bridge.submitToolResult("call_2", { text: "second" }); + + expect( + parseSent(socket).filter((event) => event.type === "conversation.item.create"), + ).toHaveLength(2); + expect(parseSent(socket).filter((event) => event.type === "response.create")).toHaveLength(1); + }); + + it("releases a deferred continuation when the last parallel result is suppressed", async () => { + const bridge = createNativeBridge({ onToolCall: vi.fn() }); + const socket = await connectReadyBridge(bridge); + emitCompletedToolCalls(socket, ["call_1", "call_2"]); + + void bridge.submitToolResult("call_1", { text: "first" }); + void bridge.submitToolResult( + "call_2", + { status: "already_delivered" }, + { suppressResponse: true }, + ); + + expect(parseSent(socket).filter((event) => event.type === "response.create")).toHaveLength(1); + }); + + it("rejects stale tool results and old-socket events after reconnect", async () => { + vi.useFakeTimers(); + const onToolCall = vi.fn(); + const bridge = createNativeBridge({ onToolCall }); + const socket = await connectReadyBridge(bridge); + emitCompletedToolCalls(socket, ["call_reused"]); + expect(onToolCall).toHaveBeenCalledTimes(1); + + socket.emit("close", 1006, Buffer.from("transient drop")); + await vi.advanceTimersByTimeAsync(1000); + const reconnectedSocket = requireSocket(1); + openSocket(reconnectedSocket); + emitSessionUpdated(reconnectedSocket); + + void bridge.submitToolResult("call_reused", { text: "stale" }); + emitCompletedToolCalls(socket, ["call_from_old_socket"]); + expect( + parseSent(reconnectedSocket).filter((event) => event.type === "conversation.item.create"), + ).toEqual([]); + expect(onToolCall).toHaveBeenCalledTimes(1); + + emitCompletedToolCalls(reconnectedSocket, ["call_reused"]); + void bridge.submitToolResult("call_reused", { text: "fresh" }); + + expect(onToolCall).toHaveBeenCalledTimes(2); + expect( + parseSent(reconnectedSocket).filter( + (event) => + event.type === "conversation.item.create" && + (event.item as { call_id?: string } | undefined)?.call_id === "call_reused", + ), + ).toHaveLength(1); + }); + it("does not flush deferred response.create while a tool result is still continuing", async () => { const onError = vi.fn(); - const bridge = createNativeBridge({ onError }); + const bridge = createNativeBridge({ onError, onToolCall: vi.fn() }); const socket = await connectReadyBridge(bridge); - socket.emit( - "message", - Buffer.from(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })), - ); + emitCompletedToolCalls(socket); void bridge.submitToolResult("call_1", { status: "working" }, { willContinue: true }); - emitServerEvent(socket, { type: "response.done" }); + bridge.sendUserMessage?.("queue after tool result"); expect(onError).not.toHaveBeenCalled(); expect(parseSent(socket).filter((event) => event.type === "response.create")).toEqual([]); @@ -3107,7 +3397,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { Buffer.from(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })), ); - void bridge.submitToolResult("call_1", { text: "done" }); + bridge.sendUserMessage?.("queued after cancellation"); socket.emit("message", Buffer.from(JSON.stringify({ type: "response.cancelled" }))); expect(parseSent(socket).slice(-1)).toEqual([expectedResponseCreateEvent()]); @@ -3280,7 +3570,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { Buffer.from(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })), ); - void bridge.submitToolResult("call_1", { text: "done" }); + bridge.sendUserMessage?.("queued after cancellation error"); bridge.handleBargeIn?.({ audioPlaybackActive: true }); const responseCancelEvent = parseSent(socket).findLast( (event) => event.type === "response.cancel", @@ -3333,7 +3623,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { if (!responseCancelEvent?.event_id) { throw new Error("expected response.cancel event id"); } - void bridge.submitToolResult("call_1", { text: "done" }); + bridge.sendUserMessage?.("queued newer response"); emitServerEvent(socket, { type: "response.done" }); const sessionUpdateCount = parseSent(socket).filter( (event) => event.type === "session.update", @@ -3377,7 +3667,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { "message", Buffer.from(JSON.stringify({ type: "response.created", response: { id: "resp_1" } })), ); - void bridge.submitToolResult("call_1", { text: "done" }); + bridge.sendUserMessage?.("queued before reconnect"); expect(parseSent(socket).slice(-1)[0]?.type).toBe("conversation.item.create"); @@ -3407,7 +3697,7 @@ describe("buildOpenAIRealtimeVoiceProvider", () => { const bridge = createNativeBridge({ onError }); const socket = await connectReadyBridge(bridge); - void bridge.submitToolResult("call_1", { text: "done" }); + bridge.sendUserMessage?.("trigger active-response retry"); const responseCreateEvent = parseSent(socket).findLast( (event) => event.type === "response.create", );