From ed52c6147cdffdf18a9fe0cea106616a83113412 Mon Sep 17 00:00:00 2001 From: Luis Pater Date: Fri, 12 Jun 2026 04:25:44 +0800 Subject: [PATCH] test(websocket, api): add unit tests for response ID injection and handling of pending tool calls - Introduced test scenarios to validate `previous_response_id` injection during incremental and non-incremental requests. - Verified behavior for pending tool calls, including proper inclusion or exclusion in websocket requests. - Updated websocket handling logic to track `lastResponseID` and `pendingToolCallIDs`. - Added utility functions for pending tool call validation and cleanup. --- .../openai/openai_responses_websocket.go | 141 +++++++++- .../openai/openai_responses_websocket_test.go | 248 +++++++++++++++++- 2 files changed, 373 insertions(+), 16 deletions(-) diff --git a/sdk/api/handlers/openai/openai_responses_websocket.go b/sdk/api/handlers/openai/openai_responses_websocket.go index 0e6cfce4..3537f5ed 100644 --- a/sdk/api/handlers/openai/openai_responses_websocket.go +++ b/sdk/api/handlers/openai/openai_responses_websocket.go @@ -7,6 +7,7 @@ import ( "fmt" "io" "net/http" + "sort" "strconv" "strings" "time" @@ -267,6 +268,8 @@ func (h *OpenAIResponsesAPIHandler) ResponsesWebsocket(c *gin.Context) { var lastRequest []byte lastResponseOutput := []byte("[]") + lastResponseID := "" + var lastResponsePendingToolCallIDs []string pinnedAuthID := "" sessionAuthByID := func(authID string) (*coreauth.Auth, bool) { if h == nil || h.AuthManager == nil { @@ -335,10 +338,12 @@ func (h *OpenAIResponsesAPIHandler) ResponsesWebsocket(c *gin.Context) { var requestJSON []byte var updatedLastRequest []byte var errMsg *interfaces.ErrorMessage - requestJSON, updatedLastRequest, errMsg = normalizeResponsesWebsocketRequestWithMode( + requestJSON, updatedLastRequest, errMsg = normalizeResponsesWebsocketRequestWithIncrementalState( payload, lastRequest, lastResponseOutput, + lastResponseID, + lastResponsePendingToolCallIDs, allowIncrementalInputWithPreviousResponseID, allowCompactionReplayBypass, ) @@ -373,6 +378,8 @@ func (h *OpenAIResponsesAPIHandler) ResponsesWebsocket(c *gin.Context) { } lastRequest = updatedLastRequest lastResponseOutput = []byte("[]") + lastResponseID = "" + lastResponsePendingToolCallIDs = nil if errWrite := writeResponsesWebsocketSyntheticPrewarm(c, conn, requestJSON, wsTimelineLog, passthroughSessionID); errWrite != nil { wsTerminateErr = errWrite return @@ -385,6 +392,8 @@ func (h *OpenAIResponsesAPIHandler) ResponsesWebsocket(c *gin.Context) { updatedLastRequest = bytes.Clone(requestJSON) previousLastRequest := bytes.Clone(lastRequest) previousLastResponseOutput := bytes.Clone(lastResponseOutput) + previousLastResponseID := lastResponseID + previousLastResponsePendingToolCallIDs := append([]string(nil), lastResponsePendingToolCallIDs...) forcedTranscriptReplay := forceTranscriptReplayNextRequest lastRequest = updatedLastRequest if forcedTranscriptReplay { @@ -414,7 +423,7 @@ func (h *OpenAIResponsesAPIHandler) ResponsesWebsocket(c *gin.Context) { } dataChan, _, errChan := h.ExecuteStreamWithAuthManager(cliCtx, h.HandlerType(), modelName, requestJSON, "") - completedOutput, forwardErrMsg, errForward := h.forwardResponsesWebsocket(c, conn, cliCancel, dataChan, errChan, wsTimelineLog, passthroughSessionID) + completedOutput, completedResponseID, completedPendingToolCallIDs, forwardErrMsg, errForward := h.forwardResponsesWebsocket(c, conn, cliCancel, dataChan, errChan, wsTimelineLog, passthroughSessionID) if errForward != nil { wsTerminateErr = errForward log.Warnf("responses websocket: forward failed id=%s error=%v", passthroughSessionID, errForward) @@ -425,9 +434,13 @@ func (h *OpenAIResponsesAPIHandler) ResponsesWebsocket(c *gin.Context) { forceTranscriptReplayNextRequest = true lastRequest = previousLastRequest lastResponseOutput = previousLastResponseOutput + lastResponseID = previousLastResponseID + lastResponsePendingToolCallIDs = previousLastResponsePendingToolCallIDs continue } lastResponseOutput = completedOutput + lastResponseID = strings.TrimSpace(completedResponseID) + lastResponsePendingToolCallIDs = append([]string(nil), completedPendingToolCallIDs...) } } @@ -457,6 +470,14 @@ func normalizeResponsesWebsocketRequest(rawJSON []byte, lastRequest []byte, last } func normalizeResponsesWebsocketRequestWithMode(rawJSON []byte, lastRequest []byte, lastResponseOutput []byte, allowIncrementalInputWithPreviousResponseID bool, allowCompactionReplayBypass bool) ([]byte, []byte, *interfaces.ErrorMessage) { + return normalizeResponsesWebsocketRequestWithLastResponseID(rawJSON, lastRequest, lastResponseOutput, "", allowIncrementalInputWithPreviousResponseID, allowCompactionReplayBypass) +} + +func normalizeResponsesWebsocketRequestWithLastResponseID(rawJSON []byte, lastRequest []byte, lastResponseOutput []byte, lastResponseID string, allowIncrementalInputWithPreviousResponseID bool, allowCompactionReplayBypass bool) ([]byte, []byte, *interfaces.ErrorMessage) { + return normalizeResponsesWebsocketRequestWithIncrementalState(rawJSON, lastRequest, lastResponseOutput, lastResponseID, nil, allowIncrementalInputWithPreviousResponseID, allowCompactionReplayBypass) +} + +func normalizeResponsesWebsocketRequestWithIncrementalState(rawJSON []byte, lastRequest []byte, lastResponseOutput []byte, lastResponseID string, lastResponsePendingToolCallIDs []string, allowIncrementalInputWithPreviousResponseID bool, allowCompactionReplayBypass bool) ([]byte, []byte, *interfaces.ErrorMessage) { requestType := strings.TrimSpace(gjson.GetBytes(rawJSON, "type").String()) switch requestType { case wsRequestTypeCreate: @@ -464,10 +485,10 @@ func normalizeResponsesWebsocketRequestWithMode(rawJSON []byte, lastRequest []by if len(lastRequest) == 0 { return normalizeResponseCreateRequest(rawJSON) } - return normalizeResponseSubsequentRequest(rawJSON, lastRequest, lastResponseOutput, allowIncrementalInputWithPreviousResponseID, allowCompactionReplayBypass) + return normalizeResponseSubsequentRequest(rawJSON, lastRequest, lastResponseOutput, lastResponseID, lastResponsePendingToolCallIDs, allowIncrementalInputWithPreviousResponseID, allowCompactionReplayBypass) case wsRequestTypeAppend: // log.Infof("responses websocket: response.append request") - return normalizeResponseSubsequentRequest(rawJSON, lastRequest, lastResponseOutput, allowIncrementalInputWithPreviousResponseID, allowCompactionReplayBypass) + return normalizeResponseSubsequentRequest(rawJSON, lastRequest, lastResponseOutput, lastResponseID, lastResponsePendingToolCallIDs, allowIncrementalInputWithPreviousResponseID, allowCompactionReplayBypass) default: return nil, lastRequest, &interfaces.ErrorMessage{ StatusCode: http.StatusBadRequest, @@ -496,7 +517,7 @@ func normalizeResponseCreateRequest(rawJSON []byte) ([]byte, []byte, *interfaces return normalized, bytes.Clone(normalized), nil } -func normalizeResponseSubsequentRequest(rawJSON []byte, lastRequest []byte, lastResponseOutput []byte, allowIncrementalInputWithPreviousResponseID bool, allowCompactionReplayBypass bool) ([]byte, []byte, *interfaces.ErrorMessage) { +func normalizeResponseSubsequentRequest(rawJSON []byte, lastRequest []byte, lastResponseOutput []byte, lastResponseID string, lastResponsePendingToolCallIDs []string, allowIncrementalInputWithPreviousResponseID bool, allowCompactionReplayBypass bool) ([]byte, []byte, *interfaces.ErrorMessage) { if len(lastRequest) == 0 { return nil, lastRequest, &interfaces.ErrorMessage{ StatusCode: http.StatusBadRequest, @@ -524,11 +545,20 @@ func normalizeResponseSubsequentRequest(rawJSON []byte, lastRequest []byte, last // Websocket v2 mode uses response.create with previous_response_id + incremental input. // Do not expand it into a full input transcript; upstream expects the incremental payload. if allowIncrementalInputWithPreviousResponseID { - if prev := strings.TrimSpace(gjson.GetBytes(rawJSON, "previous_response_id").String()); prev != "" { + prev := strings.TrimSpace(gjson.GetBytes(rawJSON, "previous_response_id").String()) + if prev == "" { + if !inputSatisfiesPendingToolCalls(nextInput, lastResponsePendingToolCallIDs) { + normalized := normalizeResponseTranscriptReplacement(rawJSON, lastRequest) + return normalized, bytes.Clone(normalized), nil + } + prev = strings.TrimSpace(lastResponseID) + } + if prev != "" { normalized, errDelete := sjson.DeleteBytes(rawJSON, "type") if errDelete != nil { normalized = bytes.Clone(rawJSON) } + normalized, _ = sjson.SetBytes(normalized, "previous_response_id", prev) if !gjson.GetBytes(normalized, "model").Exists() { modelName := strings.TrimSpace(gjson.GetBytes(lastRequest, "model").String()) if modelName != "" { @@ -644,6 +674,35 @@ func shouldReplaceWebsocketTranscript(rawJSON []byte, nextInput gjson.Result) bo return false } +func inputSatisfiesPendingToolCalls(input gjson.Result, pendingCallIDs []string) bool { + if len(pendingCallIDs) == 0 { + return true + } + if !input.IsArray() { + return false + } + outputs := make(map[string]struct{}, len(pendingCallIDs)) + for _, item := range input.Array() { + switch strings.TrimSpace(item.Get("type").String()) { + case "function_call_output", "custom_tool_call_output": + callID := strings.TrimSpace(item.Get("call_id").String()) + if callID != "" { + outputs[callID] = struct{}{} + } + } + } + for _, callID := range pendingCallIDs { + callID = strings.TrimSpace(callID) + if callID == "" { + continue + } + if _, ok := outputs[callID]; !ok { + return false + } + } + return true +} + func normalizeResponseTranscriptReplacement(rawJSON []byte, lastRequest []byte) []byte { normalized, errDelete := sjson.DeleteBytes(rawJSON, "type") if errDelete != nil { @@ -1138,9 +1197,11 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesWebsocket( errs <-chan *interfaces.ErrorMessage, wsTimelineLog websocketTimelineAppender, sessionID string, -) ([]byte, *interfaces.ErrorMessage, error) { +) ([]byte, string, []string, *interfaces.ErrorMessage, error) { completed := false completedOutput := []byte("[]") + completedResponseID := "" + pendingToolCallIDs := make(map[string]struct{}) downstreamSessionKey := "" if c != nil && c.Request != nil { downstreamSessionKey = websocketDownstreamSessionKey(c.Request) @@ -1150,7 +1211,7 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesWebsocket( select { case <-c.Request.Context().Done(): cancel(c.Request.Context().Err()) - return completedOutput, nil, c.Request.Context().Err() + return completedOutput, completedResponseID, sortedStringSet(pendingToolCallIDs), nil, c.Request.Context().Err() case errMsg, ok := <-errs: if !ok { errs = nil @@ -1175,7 +1236,7 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesWebsocket( // errWrite, // ) cancel(errMsg.Error) - return completedOutput, errMsg, errWrite + return completedOutput, completedResponseID, sortedStringSet(pendingToolCallIDs), errMsg, errWrite } } if errMsg != nil { @@ -1183,7 +1244,7 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesWebsocket( } else { cancel(nil) } - return completedOutput, errMsg, nil + return completedOutput, completedResponseID, sortedStringSet(pendingToolCallIDs), errMsg, nil case chunk, ok := <-data: if !ok { if !completed { @@ -1209,22 +1270,24 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesWebsocket( errWrite, ) cancel(errMsg.Error) - return completedOutput, errMsg, errWrite + return completedOutput, completedResponseID, sortedStringSet(pendingToolCallIDs), errMsg, errWrite } cancel(errMsg.Error) - return completedOutput, errMsg, nil + return completedOutput, completedResponseID, sortedStringSet(pendingToolCallIDs), errMsg, nil } cancel(nil) - return completedOutput, nil, nil + return completedOutput, completedResponseID, sortedStringSet(pendingToolCallIDs), nil, nil } payloads := websocketJSONPayloadsFromChunk(chunk) for i := range payloads { recordResponsesWebsocketToolCallsFromPayload(downstreamSessionKey, payloads[i]) + recordPendingToolCallIDsFromPayload(pendingToolCallIDs, payloads[i]) eventType := gjson.GetBytes(payloads[i], "type").String() if eventType == wsEventTypeCompleted { completed = true completedOutput = responseCompletedOutputFromPayload(payloads[i]) + completedResponseID = responseCompletedIDFromPayload(payloads[i]) } markAPIResponseTimestamp(c) // log.Infof( @@ -1242,7 +1305,7 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesWebsocket( errWrite, ) cancel(errWrite) - return completedOutput, nil, errWrite + return completedOutput, completedResponseID, sortedStringSet(pendingToolCallIDs), nil, errWrite } } } @@ -1275,6 +1338,56 @@ func responseCompletedOutputFromPayload(payload []byte) []byte { return []byte("[]") } +func responseCompletedIDFromPayload(payload []byte) string { + return strings.TrimSpace(gjson.GetBytes(payload, "response.id").String()) +} + +func recordPendingToolCallIDsFromPayload(pending map[string]struct{}, payload []byte) { + if pending == nil || len(payload) == 0 { + return + } + updatePendingToolCallIDsFromItem(pending, gjson.GetBytes(payload, "item")) + output := gjson.GetBytes(payload, "response.output") + if output.IsArray() { + for _, item := range output.Array() { + updatePendingToolCallIDsFromItem(pending, item) + } + } +} + +func updatePendingToolCallIDsFromItem(pending map[string]struct{}, item gjson.Result) { + if pending == nil || !item.Exists() { + return + } + 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{}{} + } + case "function_call_output", "custom_tool_call_output": + callID := strings.TrimSpace(item.Get("call_id").String()) + if callID != "" { + delete(pending, callID) + } + } +} + +func sortedStringSet(values map[string]struct{}) []string { + if len(values) == 0 { + return nil + } + out := make([]string, 0, len(values)) + for value := range values { + value = strings.TrimSpace(value) + if value != "" { + out = append(out, value) + } + } + sort.Strings(out) + return out +} + func websocketJSONPayloadsFromChunk(chunk []byte) [][]byte { payloads := make([][]byte, 0, 2) lines := bytes.Split(chunk, []byte("\n")) diff --git a/sdk/api/handlers/openai/openai_responses_websocket_test.go b/sdk/api/handlers/openai/openai_responses_websocket_test.go index 6796023e..cefffcc9 100644 --- a/sdk/api/handlers/openai/openai_responses_websocket_test.go +++ b/sdk/api/handlers/openai/openai_responses_websocket_test.go @@ -500,6 +500,83 @@ func TestNormalizeResponsesWebsocketRequestWithPreviousResponseIDIncremental(t * } } +func TestNormalizeResponsesWebsocketRequestInjectsPreviousResponseIDForIncremental(t *testing.T) { + lastRequest := []byte(`{"model":"test-model","stream":true,"instructions":"be helpful","input":[{"type":"message","id":"msg-1"}]}`) + lastResponseOutput := []byte(`[ + {"type":"function_call","id":"fc-1","call_id":"call-1"}, + {"type":"message","id":"assistant-1"} + ]`) + raw := []byte(`{"type":"response.create","input":[{"type":"function_call_output","call_id":"call-1","id":"tool-out-1"}]}`) + + normalized, next, errMsg := normalizeResponsesWebsocketRequestWithLastResponseID(raw, lastRequest, lastResponseOutput, "resp-1", true, false) + if errMsg != nil { + t.Fatalf("unexpected error: %v", errMsg.Error) + } + if got := gjson.GetBytes(normalized, "previous_response_id").String(); got != "resp-1" { + t.Fatalf("previous_response_id = %q, want resp-1", got) + } + input := gjson.GetBytes(normalized, "input").Array() + if len(input) != 1 { + t.Fatalf("incremental input len = %d, want 1: %s", len(input), normalized) + } + if input[0].Get("id").String() != "tool-out-1" { + t.Fatalf("unexpected incremental input item id: %s", input[0].Get("id").String()) + } + if gjson.GetBytes(normalized, "model").String() != "test-model" { + t.Fatalf("unexpected model: %s", gjson.GetBytes(normalized, "model").String()) + } + if gjson.GetBytes(normalized, "instructions").String() != "be helpful" { + t.Fatalf("unexpected instructions: %s", gjson.GetBytes(normalized, "instructions").String()) + } + if !bytes.Equal(next, normalized) { + t.Fatalf("next request snapshot should match normalized request") + } +} + +func TestNormalizeResponsesWebsocketRequestInjectsPreviousResponseIDWhenPendingOutputIsPresent(t *testing.T) { + lastRequest := []byte(`{"model":"test-model","stream":true,"instructions":"be helpful","input":[{"type":"message","id":"msg-1"}]}`) + lastResponseOutput := []byte(`[]`) + raw := []byte(`{"type":"response.create","input":[{"type":"function_call_output","call_id":"call-1","id":"tool-out-1"}]}`) + + normalized, _, errMsg := normalizeResponsesWebsocketRequestWithIncrementalState(raw, lastRequest, lastResponseOutput, "resp-1", []string{"call-1"}, true, false) + if errMsg != nil { + t.Fatalf("unexpected error: %v", errMsg.Error) + } + if got := gjson.GetBytes(normalized, "previous_response_id").String(); got != "resp-1" { + t.Fatalf("previous_response_id = %q, want resp-1", got) + } + input := gjson.GetBytes(normalized, "input").Array() + if len(input) != 1 || input[0].Get("id").String() != "tool-out-1" { + t.Fatalf("unexpected incremental input: %s", normalized) + } +} + +func TestNormalizeResponsesWebsocketRequestSkipsPreviousResponseIDWhenPendingOutputIsMissing(t *testing.T) { + lastRequest := []byte(`{"model":"test-model","stream":true,"instructions":"be helpful","input":[{"type":"message","id":"msg-1"}]}`) + lastResponseOutput := []byte(`[ + {"type":"function_call","id":"fc-1","call_id":"call-1"} + ]`) + raw := []byte(`{"type":"response.create","input":[{"type":"message","role":"user","id":"summary-1","content":"compacted summary"}]}`) + + normalized, next, errMsg := normalizeResponsesWebsocketRequestWithIncrementalState(raw, lastRequest, lastResponseOutput, "resp-1", []string{"call-1"}, true, false) + if errMsg != nil { + t.Fatalf("unexpected error: %v", errMsg.Error) + } + if gjson.GetBytes(normalized, "previous_response_id").Exists() { + t.Fatalf("previous_response_id must not be injected when pending tool output is missing: %s", normalized) + } + input := gjson.GetBytes(normalized, "input").Array() + if len(input) != 1 { + t.Fatalf("replacement input len = %d, want 1: %s", len(input), normalized) + } + if input[0].Get("id").String() != "summary-1" { + t.Fatalf("unexpected replacement input: %s", normalized) + } + if !bytes.Equal(next, normalized) { + t.Fatalf("next request snapshot should match normalized request") + } +} + func TestNormalizeResponsesWebsocketRequestWithPreviousResponseIDMergedWhenIncrementalDisabled(t *testing.T) { lastRequest := []byte(`{"model":"test-model","stream":true,"input":[{"type":"message","id":"msg-1"}]}`) lastResponseOutput := []byte(`[ @@ -1014,7 +1091,7 @@ func TestForwardResponsesWebsocketPreservesCompletedEvent(t *testing.T) { close(errCh) timelineLog := newInMemoryWebsocketTimelineLog() - completedOutput, errMsg, err := (*OpenAIResponsesAPIHandler)(nil).forwardResponsesWebsocket( + completedOutput, completedResponseID, pendingToolCallIDs, errMsg, err := (*OpenAIResponsesAPIHandler)(nil).forwardResponsesWebsocket( ctx, conn, func(...interface{}) {}, @@ -1035,6 +1112,14 @@ func TestForwardResponsesWebsocketPreservesCompletedEvent(t *testing.T) { serverErrCh <- errors.New("completed output not captured") return } + if completedResponseID != "resp-1" { + serverErrCh <- fmt.Errorf("completed response id = %q, want resp-1", completedResponseID) + return + } + if len(pendingToolCallIDs) != 0 { + serverErrCh <- fmt.Errorf("pending tool call ids = %v, want empty", pendingToolCallIDs) + return + } if !strings.Contains(timelineLog.String(), "Event: websocket.response") { serverErrCh <- errors.New("websocket timeline did not capture downstream response") return @@ -1071,6 +1156,17 @@ func TestForwardResponsesWebsocketPreservesCompletedEvent(t *testing.T) { } } +func TestRecordPendingToolCallIDsFromPayloadDropsSatisfiedCalls(t *testing.T) { + pending := map[string]struct{}{} + payload := []byte(`{"type":"response.completed","response":{"output":[{"type":"function_call","call_id":"call-1","id":"fc-1"},{"type":"function_call_output","call_id":"call-1","id":"out-1"},{"type":"custom_tool_call","call_id":"call-2","id":"ctc-1"},{"type":"custom_tool_call_output","call_id":"call-2","id":"custom-out-1"}]}}`) + + recordPendingToolCallIDsFromPayload(pending, payload) + + if len(pending) != 0 { + t.Fatalf("pending tool call ids = %v, want empty", sortedStringSet(pending)) + } +} + func TestForwardResponsesWebsocketLogsAttemptedResponseOnWriteFailure(t *testing.T) { gin.SetMode(gin.TestMode) @@ -1097,7 +1193,7 @@ func TestForwardResponsesWebsocketLogsAttemptedResponseOnWriteFailure(t *testing return } - _, _, err = (*OpenAIResponsesAPIHandler)(nil).forwardResponsesWebsocket( + _, _, _, _, err = (*OpenAIResponsesAPIHandler)(nil).forwardResponsesWebsocket( ctx, conn, func(...interface{}) {}, @@ -1410,6 +1506,154 @@ func TestResponsesWebsocketPrewarmHandledLocallyForSSEUpstream(t *testing.T) { } } +func TestResponsesWebsocketInjectsPreviousResponseIDForWebsocketUpstream(t *testing.T) { + gin.SetMode(gin.TestMode) + + executor := &websocketCaptureExecutor{} + manager := coreauth.NewManager(nil, nil, nil) + manager.RegisterExecutor(executor) + auth := &coreauth.Auth{ + ID: "auth-ws", + Provider: executor.Identifier(), + Status: coreauth.StatusActive, + Attributes: map[string]string{"websockets": "true"}, + } + if _, err := manager.Register(context.Background(), auth); err != nil { + t.Fatalf("Register auth: %v", err) + } + registry.GetGlobalRegistry().RegisterClient(auth.ID, auth.Provider, []*registry.ModelInfo{{ID: "test-model"}}) + t.Cleanup(func() { + registry.GetGlobalRegistry().UnregisterClient(auth.ID) + }) + + base := handlers.NewBaseAPIHandlers(&sdkconfig.SDKConfig{}, manager) + h := NewOpenAIResponsesAPIHandler(base) + router := gin.New() + router.GET("/v1/responses/ws", h.ResponsesWebsocket) + + server := httptest.NewServer(router) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/v1/responses/ws" + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("dial websocket: %v", err) + } + defer func() { + if errClose := conn.Close(); errClose != nil { + t.Fatalf("close websocket: %v", errClose) + } + }() + + requests := []string{ + `{"type":"response.create","model":"test-model","input":[{"type":"message","id":"msg-1"}]}`, + `{"type":"response.create","input":[{"type":"message","id":"msg-2"}]}`, + } + for i := range requests { + if errWrite := conn.WriteMessage(websocket.TextMessage, []byte(requests[i])); errWrite != nil { + t.Fatalf("write websocket message %d: %v", i+1, errWrite) + } + _, payload, errReadMessage := conn.ReadMessage() + if errReadMessage != nil { + t.Fatalf("read websocket message %d: %v", i+1, errReadMessage) + } + if got := gjson.GetBytes(payload, "type").String(); got != wsEventTypeCompleted { + t.Fatalf("message %d payload type = %s, want %s", i+1, got, wsEventTypeCompleted) + } + } + + if len(executor.payloads) != 2 { + t.Fatalf("upstream payload count = %d, want 2", len(executor.payloads)) + } + secondPayload := executor.payloads[1] + if got := gjson.GetBytes(secondPayload, "previous_response_id").String(); got != "resp-upstream" { + t.Fatalf("previous_response_id = %q, want resp-upstream: %s", got, secondPayload) + } + input := gjson.GetBytes(secondPayload, "input").Array() + if len(input) != 1 { + t.Fatalf("second upstream input len = %d, want 1: %s", len(input), secondPayload) + } + if input[0].Get("id").String() != "msg-2" { + t.Fatalf("second upstream input item id = %s, want msg-2", input[0].Get("id").String()) + } +} + +func TestResponsesWebsocketDoesNotInjectPreviousResponseIDWhenPendingToolOutputMissing(t *testing.T) { + gin.SetMode(gin.TestMode) + + executor := &websocketCompactionCaptureExecutor{} + manager := coreauth.NewManager(nil, nil, nil) + manager.RegisterExecutor(executor) + auth := &coreauth.Auth{ + ID: "auth-ws", + Provider: executor.Identifier(), + Status: coreauth.StatusActive, + Attributes: map[string]string{"websockets": "true"}, + } + if _, err := manager.Register(context.Background(), auth); err != nil { + t.Fatalf("Register auth: %v", err) + } + registry.GetGlobalRegistry().RegisterClient(auth.ID, auth.Provider, []*registry.ModelInfo{{ID: "test-model"}}) + t.Cleanup(func() { + registry.GetGlobalRegistry().UnregisterClient(auth.ID) + }) + + base := handlers.NewBaseAPIHandlers(&sdkconfig.SDKConfig{}, manager) + h := NewOpenAIResponsesAPIHandler(base) + router := gin.New() + router.GET("/v1/responses/ws", h.ResponsesWebsocket) + + server := httptest.NewServer(router) + defer server.Close() + + wsURL := "ws" + strings.TrimPrefix(server.URL, "http") + "/v1/responses/ws" + conn, _, err := websocket.DefaultDialer.Dial(wsURL, nil) + if err != nil { + t.Fatalf("dial websocket: %v", err) + } + defer func() { + if errClose := conn.Close(); errClose != nil { + t.Fatalf("close websocket: %v", errClose) + } + }() + + requests := []string{ + `{"type":"response.create","model":"test-model","input":[{"type":"message","id":"msg-1"}]}`, + `{"type":"response.create","input":[{"type":"message","role":"user","id":"summary-1","content":"compacted summary"}]}`, + } + for i := range requests { + if errWrite := conn.WriteMessage(websocket.TextMessage, []byte(requests[i])); errWrite != nil { + t.Fatalf("write websocket message %d: %v", i+1, errWrite) + } + _, payload, errReadMessage := conn.ReadMessage() + if errReadMessage != nil { + t.Fatalf("read websocket message %d: %v", i+1, errReadMessage) + } + if got := gjson.GetBytes(payload, "type").String(); got != wsEventTypeCompleted { + t.Fatalf("message %d payload type = %s, want %s", i+1, got, wsEventTypeCompleted) + } + } + + executor.mu.Lock() + payloads := append([][]byte(nil), executor.streamPayloads...) + executor.mu.Unlock() + + if len(payloads) != 2 { + t.Fatalf("upstream payload count = %d, want 2", len(payloads)) + } + secondPayload := payloads[1] + if gjson.GetBytes(secondPayload, "previous_response_id").Exists() { + t.Fatalf("previous_response_id must not be injected when pending tool output is missing: %s", secondPayload) + } + input := gjson.GetBytes(secondPayload, "input").Array() + if len(input) != 1 { + t.Fatalf("second upstream input len = %d, want 1: %s", len(input), secondPayload) + } + if input[0].Get("id").String() != "summary-1" { + t.Fatalf("second upstream input item id = %s, want summary-1", input[0].Get("id").String()) + } +} + func TestResponsesWebsocketStripsGenerateWhenWebsocketAttemptFallsBackToHTTP(t *testing.T) { gin.SetMode(gin.TestMode)