test(websocket): add robust unit tests for tool call handling and reconciliation

- Implemented tests for scenarios involving incomplete, conflicting, and reconciled tool calls in response payloads.
- Enhanced coverage for `restoreResponsesWebsocketCompletionOutput` and `responseCompletedOutputFromPayload` logic.
- Refactored `isCompleteResponsesWebsocketToolCall` to validate string fields in tool calls.
- Improved caching behavior to skip incomplete tool calls and ensure reconciliation logic aligns with expected outputs.

Closes: #4447

sdk/api/handlers/openai/openai_responses_websocket.go

@@ -1771,7 +1771,15 @@ func collectResponsesWebsocketOutputItem(payload []byte, outputItemsByIndex map[
 func restoreResponsesWebsocketCompletionOutput(payload []byte, outputItemsByIndex map[int64][]byte, outputItemsFallback [][]byte) []byte {
 	output := gjson.GetBytes(payload, "response.output")
 	if output.Exists() && output.IsArray() && len(output.Array()) > 0 {
-		return payload
+		reconciledOutput, changed := reconcileResponsesWebsocketCompletionToolCalls(output, outputItemsByIndex, outputItemsFallback)
+		if !changed {
+			return payload
+		}
+		restored, errSet := sjson.SetRawBytes(payload, "response.output", reconciledOutput)
+		if errSet != nil {
+			return payload
+		}
+		return restored
 	}
 	if len(outputItemsByIndex) == 0 && len(outputItemsFallback) == 0 {
 		return payload
@@ -1784,6 +1792,81 @@ func restoreResponsesWebsocketCompletionOutput(payload []byte, outputItemsByInde
 	return restored
 }
 
+func reconcileResponsesWebsocketCompletionToolCalls(output gjson.Result, outputItemsByIndex map[int64][]byte, outputItemsFallback [][]byte) ([]byte, bool) {
+	collectedToolCalls := make(map[string]json.RawMessage)
+	recordCollectedToolCall := func(raw []byte) {
+		item := gjson.ParseBytes(raw)
+		if !isCompleteResponsesWebsocketToolCall(item) {
+			return
+		}
+		callID := strings.TrimSpace(item.Get("call_id").String())
+		collectedToolCalls[callID] = append(json.RawMessage(nil), raw...)
+	}
+
+	indexes := make([]int64, 0, len(outputItemsByIndex))
+	for index := range outputItemsByIndex {
+		indexes = append(indexes, index)
+	}
+	sort.Slice(indexes, func(i, j int) bool {
+		return indexes[i] < indexes[j]
+	})
+	for _, index := range indexes {
+		recordCollectedToolCall(outputItemsByIndex[index])
+	}
+	for _, item := range outputItemsFallback {
+		recordCollectedToolCall(item)
+	}
+	if len(collectedToolCalls) == 0 {
+		return nil, false
+	}
+
+	items := output.Array()
+	reconciled := make([]json.RawMessage, 0, len(items))
+	changed := false
+	for _, item := range items {
+		raw := json.RawMessage(item.Raw)
+		if isResponsesToolCallType(item.Get("type").String()) {
+			callID := strings.TrimSpace(item.Get("call_id").String())
+			if collected, ok := collectedToolCalls[callID]; ok && !bytes.Equal(raw, collected) {
+				raw = collected
+				changed = true
+			}
+		}
+		reconciled = append(reconciled, raw)
+	}
+	if !changed {
+		return nil, false
+	}
+
+	marshaledOutput, errMarshal := json.Marshal(reconciled)
+	if errMarshal != nil {
+		return nil, false
+	}
+	return marshaledOutput, true
+}
+
+func isCompleteResponsesWebsocketToolCall(item gjson.Result) bool {
+	if !item.Exists() || !item.IsObject() {
+		return false
+	}
+	callID := item.Get("call_id")
+	name := item.Get("name")
+	if callID.Type != gjson.String || strings.TrimSpace(callID.String()) == "" || name.Type != gjson.String || strings.TrimSpace(name.String()) == "" {
+		return false
+	}
+
+	switch strings.TrimSpace(item.Get("type").String()) {
+	case "function_call":
+		arguments := item.Get("arguments")
+		return arguments.Exists() && arguments.Type == gjson.String
+	case "custom_tool_call":
+		input := item.Get("input")
+		return input.Exists() && input.Type == gjson.String
+	default:
+		return false
+	}
+}
+
 func responseCompletedOutputFromPayload(payload []byte, outputItemsByIndex map[int64][]byte, outputItemsFallback [][]byte) []byte {
 	output := gjson.GetBytes(payload, "response.output")
 	if output.Exists() && output.IsArray() && len(output.Array()) > 0 {
@@ -1802,11 +1885,18 @@ func responseCompletedOutputFromPayload(payload []byte, outputItemsByIndex map[i
 	})
 
 	items := make([]json.RawMessage, 0, len(outputItemsByIndex)+len(outputItemsFallback))
+	appendCollectedItem := func(raw []byte) {
+		item := gjson.ParseBytes(raw)
+		if isResponsesToolCallType(item.Get("type").String()) && !isCompleteResponsesWebsocketToolCall(item) {
+			return
+		}
+		items = append(items, append(json.RawMessage(nil), raw...))
+	}
 	for _, index := range indexes {
-		items = append(items, json.RawMessage(outputItemsByIndex[index]))
+		appendCollectedItem(outputItemsByIndex[index])
 	}
 	for _, item := range outputItemsFallback {
-		items = append(items, json.RawMessage(item))
+		appendCollectedItem(item)
 	}
 
 	marshaledOutput, errMarshal := json.Marshal(items)
@@ -1839,10 +1929,11 @@ func updatePendingToolCallIDsFromItem(pending map[string]struct{}, item gjson.Re
 	}
 	switch strings.TrimSpace(item.Get("type").String()) {
 	case "function_call", "custom_tool_call":
-		callID := strings.TrimSpace(item.Get("call_id").String())
-		if callID != "" {
-			pending[callID] = struct{}{}
+		if !isCompleteResponsesWebsocketToolCall(item) {
+			return
 		}
+		callID := strings.TrimSpace(item.Get("call_id").String())
+		pending[callID] = struct{}{}
 	case "function_call_output", "custom_tool_call_output":
 		callID := strings.TrimSpace(item.Get("call_id").String())
 		if callID != "" {

sdk/api/handlers/openai/openai_responses_websocket_test.go

@@ -1225,6 +1225,27 @@ func TestResponseCompletedOutputFromPayload(t *testing.T) {
 	}
 }
 
+func TestResponseCompletedOutputFromPayloadDropsIncompleteCollectedToolCalls(t *testing.T) {
+	payload := []byte(`{"type":"response.completed","response":{"id":"resp-1","output":[]}}`)
+	collector := map[int64][]byte{
+		0: []byte(`{"type":"message","id":"msg-1"}`),
+		1: []byte(`{"type":"function_call","call_id":"call-1","name":"exec"}`),
+		2: []byte(`{"type":"custom_tool_call","call_id":"call-2","name":"exec","input":"pwd"}`),
+	}
+
+	output := responseCompletedOutputFromPayload(payload, collector, nil)
+	items := gjson.ParseBytes(output).Array()
+	if len(items) != 2 {
+		t.Fatalf("output len = %d, want 2: %s", len(items), output)
+	}
+	if items[0].Get("type").String() != "message" || items[0].Get("id").String() != "msg-1" {
+		t.Fatalf("unexpected first output item: %s", items[0].Raw)
+	}
+	if items[1].Get("type").String() != "custom_tool_call" || items[1].Get("call_id").String() != "call-2" {
+		t.Fatalf("unexpected second output item: %s", items[1].Raw)
+	}
+}
+
 func TestRestoreResponsesWebsocketCompletionOutputPreservesNonEmptyOutput(t *testing.T) {
 	payload := []byte(`{"type":"response.completed","response":{"id":"resp-1","output":[{"type":"message","id":"out-1"}]}}`)
 	collector := map[int64][]byte{0: []byte(`{"type":"function_call","id":"call-1","call_id":"call-1"}`)}
@@ -1235,6 +1256,100 @@ func TestRestoreResponsesWebsocketCompletionOutputPreservesNonEmptyOutput(t *tes
 	}
 }
 
+func TestRestoreResponsesWebsocketCompletionOutputReconcilesConflictingToolCall(t *testing.T) {
+	payload := []byte(`{"type":"response.completed","response":{"id":"resp-1","output":[{"type":"message","id":"msg-1"},{"type":"function_call","call_id":"call-1","name":"exec"}]}}`)
+	collector := map[int64][]byte{0: []byte(`{"type":"custom_tool_call","id":"ctc-1","call_id":"call-1","name":"exec","input":"pwd","status":"completed"}`)}
+
+	restored := restoreResponsesWebsocketCompletionOutput(payload, collector, nil)
+	output := gjson.GetBytes(restored, "response.output").Array()
+	if len(output) != 2 {
+		t.Fatalf("restored output len = %d, want 2: %s", len(output), restored)
+	}
+	if output[0].Get("type").String() != "message" || output[0].Get("id").String() != "msg-1" {
+		t.Fatalf("unrelated completion item changed: %s", output[0].Raw)
+	}
+	if output[1].Get("type").String() != "custom_tool_call" || output[1].Get("call_id").String() != "call-1" {
+		t.Fatalf("conflicting tool call was not reconciled: %s", output[1].Raw)
+	}
+	if input := output[1].Get("input"); input.Type != gjson.String || input.String() != "pwd" {
+		t.Fatalf("reconciled custom tool input = %s, want string pwd", input.Raw)
+	}
+
+	lastRequest := []byte(`{"model":"gpt-test","stream":true,"input":[{"type":"message","id":"user-1","role":"user","content":"run pwd"}]}`)
+	nextRequest := []byte(`{"type":"response.create","previous_response_id":"resp-1","input":[{"type":"custom_tool_call_output","call_id":"call-1","output":"ok"}]}`)
+	completedOutput := []byte(gjson.GetBytes(restored, "response.output").Raw)
+	normalized, _, errMsg := normalizeResponsesWebsocketRequestWithIncrementalState(
+		nextRequest,
+		lastRequest,
+		completedOutput,
+		"resp-1",
+		[]string{"call-1"},
+		false,
+		false,
+	)
+	if errMsg != nil {
+		t.Fatalf("normalize next request: %v", errMsg.Error)
+	}
+	if gjson.GetBytes(normalized, "previous_response_id").Exists() {
+		t.Fatalf("previous_response_id must not be forwarded to HTTP/SSE upstream: %s", normalized)
+	}
+	input := gjson.GetBytes(normalized, "input").Array()
+	if len(input) != 4 {
+		t.Fatalf("replayed input len = %d, want 4: %s", len(input), normalized)
+	}
+	if input[2].Get("type").String() != "custom_tool_call" || input[2].Get("input").String() != "pwd" {
+		t.Fatalf("replayed tool call is invalid: %s", input[2].Raw)
+	}
+	if input[3].Get("type").String() != "custom_tool_call_output" || input[3].Get("call_id").String() != "call-1" {
+		t.Fatalf("replayed tool output is invalid: %s", input[3].Raw)
+	}
+
+	cache := newWebsocketToolOutputCache(time.Minute, 10)
+	donePayload := []byte(`{"type":"response.output_item.done","item":{"type":"custom_tool_call","id":"ctc-1","call_id":"call-1","name":"exec","input":"pwd","status":"completed"}}`)
+	recordResponsesWebsocketToolCallsFromPayloadWithCache(cache, "session-1", donePayload)
+	recordResponsesWebsocketToolCallsFromPayloadWithCache(cache, "session-1", restored)
+	cached, ok := cache.get("session-1", "call-1")
+	if !ok {
+		t.Fatalf("reconciled custom tool call was not cached")
+	}
+	if gjson.GetBytes(cached, "type").String() != "custom_tool_call" || gjson.GetBytes(cached, "input").String() != "pwd" {
+		t.Fatalf("cached tool call is invalid: %s", cached)
+	}
+}
+
+func TestRestoreResponsesWebsocketCompletionOutputIgnoresIncompleteCollectedToolCall(t *testing.T) {
+	payload := []byte(`{"type":"response.completed","response":{"id":"resp-1","output":[{"type":"function_call","call_id":"call-1","name":"exec"}]}}`)
+	collector := map[int64][]byte{0: []byte(`{"type":"custom_tool_call","call_id":"call-1","name":"exec"}`)}
+
+	restored := restoreResponsesWebsocketCompletionOutput(payload, collector, nil)
+	if string(restored) != string(payload) {
+		t.Fatalf("incomplete collected tool call overwrote completion output: %s", restored)
+	}
+}
+
+func TestIsCompleteResponsesWebsocketToolCallRequiresStringFields(t *testing.T) {
+	tests := []struct {
+		name string
+		item string
+		want bool
+	}{
+		{name: "numeric call id", item: `{"type":"function_call","call_id":123,"name":"exec","arguments":"{}"}`},
+		{name: "boolean name", item: `{"type":"function_call","call_id":"call-1","name":true,"arguments":"{}"}`},
+		{name: "numeric arguments", item: `{"type":"function_call","call_id":"call-1","name":"exec","arguments":123}`},
+		{name: "object custom input", item: `{"type":"custom_tool_call","call_id":"call-1","name":"exec","input":{}}`},
+		{name: "valid function call", item: `{"type":"function_call","call_id":"call-1","name":"exec","arguments":""}`, want: true},
+		{name: "valid custom tool call", item: `{"type":"custom_tool_call","call_id":"call-1","name":"exec","input":""}`, want: true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isCompleteResponsesWebsocketToolCall(gjson.Parse(tt.item)); got != tt.want {
+				t.Fatalf("isCompleteResponsesWebsocketToolCall() = %t, want %t", got, tt.want)
+			}
+		})
+	}
+}
+
 func TestAppendWebsocketEvent(t *testing.T) {
 	var builder strings.Builder
 
@@ -1567,6 +1682,22 @@ func TestRepairResponsesWebsocketToolCallsDropsOrphanCustomToolOutputWhenCallMis
 	}
 }
 
+func TestRecordResponsesWebsocketToolCallsIgnoresIncompleteCall(t *testing.T) {
+	cache := newWebsocketToolOutputCache(time.Minute, 10)
+	pending := make(map[string]struct{})
+	payload := []byte(`{"type":"response.output_item.done","item":{"type":"function_call","call_id":"call-1","name":"exec"}}`)
+
+	recordResponsesWebsocketToolCallsFromPayloadWithCache(cache, "session-1", payload)
+	recordPendingToolCallIDsFromPayload(pending, payload)
+
+	if cached, ok := cache.get("session-1", "call-1"); ok {
+		t.Fatalf("incomplete tool call was cached: %s", cached)
+	}
+	if len(pending) != 0 {
+		t.Fatalf("incomplete tool call was recorded as pending: %v", pending)
+	}
+}
+
 func TestRecordResponsesWebsocketToolCallsFromPayloadWithCache(t *testing.T) {
 	cache := newWebsocketToolOutputCache(time.Minute, 10)
 	sessionKey := "session-1"
@@ -3765,7 +3896,7 @@ func TestResponsesWebsocketOutputCollectorRestoresCompletedOutput(t *testing.T)
 	for _, payload := range [][]byte{
 		[]byte(`{"type":"response.output_item.done","output_index":1,"item":{"type":"message","id":"reply-1","role":"assistant"}}`),
 		[]byte(`{"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","id":"summary-1","summary":[]}}`),
-		[]byte(`{"type":"response.output_item.done","item":{"type":"function_call","id":"call-1","call_id":"call-1"}}`),
+		[]byte(`{"type":"response.output_item.done","item":{"type":"function_call","id":"call-1","call_id":"call-1","name":"exec","arguments":"{}"}}`),
 	} {
 		collectResponsesWebsocketOutputItem(payload, outputItemsByIndex, &outputItemsFallback)
 	}

sdk/api/handlers/openai/openai_responses_websocket_toolcall_repair.go

@@ -384,27 +384,18 @@ func recordResponsesWebsocketToolCallsFromPayloadWithCache(cache *websocketToolO
 			return
 		}
 		for _, item := range output.Array() {
-			if !isResponsesToolCallType(item.Get("type").String()) {
+			if !isCompleteResponsesWebsocketToolCall(item) {
 				continue
 			}
 			callID := strings.TrimSpace(item.Get("call_id").String())
-			if callID == "" {
-				continue
-			}
 			cache.record(sessionKey, callID, json.RawMessage(item.Raw))
 		}
 	case "response.output_item.added", "response.output_item.done":
 		item := gjson.GetBytes(payload, "item")
-		if !item.Exists() || !item.IsObject() {
-			return
-		}
-		if !isResponsesToolCallType(item.Get("type").String()) {
+		if !isCompleteResponsesWebsocketToolCall(item) {
 			return
 		}
 		callID := strings.TrimSpace(item.Get("call_id").String())
-		if callID == "" {
-			return
-		}
 		cache.record(sessionKey, callID, json.RawMessage(item.Raw))
 	}
 }