test(openai): cover terminal tool-call ownership

Co-authored-by: Peter Steinberger <steipete@users.noreply.github.com>

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",
     );