diff --git a/internal/runtime/executor/codex_executor.go b/internal/runtime/executor/codex_executor.go index 3f0e6eb0..f3634a5f 100644 --- a/internal/runtime/executor/codex_executor.go +++ b/internal/runtime/executor/codex_executor.go @@ -960,9 +960,6 @@ func (e *CodexExecutor) executeCompact(ctx context.Context, auth *cliproxyauth.A body, _ = sjson.SetBytes(body, "model", baseModel) body, _ = sjson.DeleteBytes(body, "stream") body = normalizeCodexInstructions(body) - if e.cfg == nil || e.cfg.DisableImageGeneration == config.DisableImageGenerationOff { - body = ensureImageGenerationTool(body, baseModel, auth, opts.Headers) - } body = sanitizeOpenAIResponsesReasoningEncryptedContent(ctx, "codex executor", body) body = normalizeCodexParallelToolCallsForTools(body) reporter.SetTranslatedReasoningEffort(body, to.String()) diff --git a/internal/runtime/executor/codex_executor_compact_test.go b/internal/runtime/executor/codex_executor_compact_test.go index 549cad9e..1d929870 100644 --- a/internal/runtime/executor/codex_executor_compact_test.go +++ b/internal/runtime/executor/codex_executor_compact_test.go @@ -14,18 +14,18 @@ import ( "github.com/tidwall/gjson" ) -func TestCodexExecutorCompactAddsDefaultInstructions(t *testing.T) { +func TestCodexExecutorCompactAddsDefaultInstructionsWithoutInjectingImageTool(t *testing.T) { cases := []struct { name string payload string }{ { name: "missing instructions", - payload: `{"model":"gpt-5.4","input":"hello"}`, + payload: `{"model":"gpt-5.4","input":[{"type":"message","role":"user","content":"history"},{"type":"compaction_trigger"}]}`, }, { name: "null instructions", - payload: `{"model":"gpt-5.4","instructions":null,"input":"hello"}`, + payload: `{"model":"gpt-5.4","instructions":null,"input":[{"type":"message","role":"user","content":"history"},{"type":"compaction_trigger"}]}`, }, } @@ -62,14 +62,15 @@ func TestCodexExecutorCompactAddsDefaultInstructions(t *testing.T) { if gotPath != "/responses/compact" { t.Fatalf("path = %q, want %q", gotPath, "/responses/compact") } - if !gjson.GetBytes(gotBody, "instructions").Exists() { - t.Fatalf("expected instructions in compact request body, got %s", string(gotBody)) + if instructions := gjson.GetBytes(gotBody, "instructions"); instructions.Type != gjson.String || instructions.String() != "" { + t.Fatalf("instructions = %s, want empty string; body=%s", instructions.Raw, gotBody) } - if gjson.GetBytes(gotBody, "instructions").Type != gjson.String { - t.Fatalf("instructions type = %v, want string", gjson.GetBytes(gotBody, "instructions").Type) + if gjson.GetBytes(gotBody, "tools").Exists() { + t.Fatalf("compact request injected image_generation tool: %s", gotBody) } - if gjson.GetBytes(gotBody, "instructions").String() != "" { - t.Fatalf("instructions = %q, want empty string", gjson.GetBytes(gotBody, "instructions").String()) + input := gjson.GetBytes(gotBody, "input").Array() + if len(input) != 2 || input[1].Get("type").String() != "compaction_trigger" { + t.Fatalf("compact input order changed: %s", gotBody) } if string(resp.Payload) != `{"id":"resp_1","object":"response.compaction","usage":{"input_tokens":1,"output_tokens":2,"total_tokens":3}}` { t.Fatalf("payload = %s", string(resp.Payload)) diff --git a/sdk/api/handlers/openai/openai_responses_websocket.go b/sdk/api/handlers/openai/openai_responses_websocket.go index 354a8107..d749a5e0 100644 --- a/sdk/api/handlers/openai/openai_responses_websocket.go +++ b/sdk/api/handlers/openai/openai_responses_websocket.go @@ -696,8 +696,7 @@ func shouldReplaceWebsocketTranscript(rawJSON []byte, nextInput gjson.Result) bo case "function_call", "custom_tool_call": return true case "message": - role := strings.TrimSpace(item.Get("role").String()) - if role == "assistant" { + if strings.TrimSpace(item.Get("role").String()) == "assistant" { return true } } @@ -1310,6 +1309,8 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesWebsocket( completed := false completedOutput := []byte("[]") completedResponseID := "" + outputItemsByIndex := make(map[int64][]byte) + var outputItemsFallback [][]byte pendingToolCallIDs := make(map[string]struct{}) downstreamSessionKey := "" if c != nil && c.Request != nil { @@ -1390,9 +1391,13 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesWebsocket( payloads := websocketJSONPayloadsFromChunk(chunk) for i := range payloads { + collectResponsesWebsocketOutputItem(payloads[i], outputItemsByIndex, &outputItemsFallback) + eventType := gjson.GetBytes(payloads[i], "type").String() + if isResponsesWebsocketCompletionEvent(eventType) { + payloads[i] = restoreResponsesWebsocketCompletionOutput(payloads[i], outputItemsByIndex, outputItemsFallback) + } recordResponsesWebsocketToolCallsFromPayload(downstreamSessionKey, payloads[i]) recordPendingToolCallIDsFromPayload(pendingToolCallIDs, payloads[i]) - eventType := gjson.GetBytes(payloads[i], "type").String() var payloadErrMsg *interfaces.ErrorMessage if eventType == wsEventTypeError { payloadErrMsg = responsesWebsocketErrorMessageFromPayload(payloads[i]) @@ -1401,7 +1406,7 @@ func (h *OpenAIResponsesAPIHandler) forwardResponsesWebsocket( } } else if isResponsesWebsocketCompletionEvent(eventType) { completed = true - completedOutput = responseCompletedOutputFromPayload(payloads[i]) + completedOutput = responseCompletedOutputFromPayload(payloads[i], outputItemsByIndex, outputItemsFallback) completedResponseID = responseCompletedIDFromPayload(payloads[i]) } markAPIResponseTimestamp(c) @@ -1467,12 +1472,68 @@ func shouldReleaseResponsesWebsocketPinnedAuth(errMsg *interfaces.ErrorMessage) return false } -func responseCompletedOutputFromPayload(payload []byte) []byte { +func collectResponsesWebsocketOutputItem(payload []byte, outputItemsByIndex map[int64][]byte, outputItemsFallback *[][]byte) { + if gjson.GetBytes(payload, "type").String() != "response.output_item.done" { + return + } + item := gjson.GetBytes(payload, "item") + if !item.Exists() || !item.IsObject() { + return + } + outputIndex := gjson.GetBytes(payload, "output_index") + if outputIndex.Exists() { + outputItemsByIndex[outputIndex.Int()] = bytes.Clone([]byte(item.Raw)) + return + } + *outputItemsFallback = append(*outputItemsFallback, bytes.Clone([]byte(item.Raw))) +} + +func restoreResponsesWebsocketCompletionOutput(payload []byte, outputItemsByIndex map[int64][]byte, outputItemsFallback [][]byte) []byte { output := gjson.GetBytes(payload, "response.output") - if output.Exists() && output.IsArray() { + if output.Exists() && output.IsArray() && len(output.Array()) > 0 { + return payload + } + if len(outputItemsByIndex) == 0 && len(outputItemsFallback) == 0 { + return payload + } + + restored, errSet := sjson.SetRawBytes(payload, "response.output", responseCompletedOutputFromPayload(payload, outputItemsByIndex, outputItemsFallback)) + if errSet != nil { + return payload + } + return restored +} + +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 { return bytes.Clone([]byte(output.Raw)) } - return []byte("[]") + if len(outputItemsByIndex) == 0 && len(outputItemsFallback) == 0 { + return []byte("[]") + } + + 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] + }) + + items := make([]json.RawMessage, 0, len(outputItemsByIndex)+len(outputItemsFallback)) + for _, index := range indexes { + items = append(items, json.RawMessage(outputItemsByIndex[index])) + } + for _, item := range outputItemsFallback { + items = append(items, json.RawMessage(item)) + } + + marshaledOutput, errMarshal := json.Marshal(items) + if errMarshal != nil { + return []byte("[]") + } + return marshaledOutput } func responseCompletedIDFromPayload(payload []byte) string { diff --git a/sdk/api/handlers/openai/openai_responses_websocket_test.go b/sdk/api/handlers/openai/openai_responses_websocket_test.go index df333492..f4f16b7a 100644 --- a/sdk/api/handlers/openai/openai_responses_websocket_test.go +++ b/sdk/api/handlers/openai/openai_responses_websocket_test.go @@ -757,7 +757,7 @@ func TestWebsocketJSONPayloadsFromPlainJSONChunk(t *testing.T) { func TestResponseCompletedOutputFromPayload(t *testing.T) { payload := []byte(`{"type":"response.completed","response":{"id":"resp-1","output":[{"type":"message","id":"out-1"}]}}`) - output := responseCompletedOutputFromPayload(payload) + output := responseCompletedOutputFromPayload(payload, nil, nil) items := gjson.ParseBytes(output).Array() if len(items) != 1 { t.Fatalf("output len = %d, want 1", len(items)) @@ -767,6 +767,16 @@ func TestResponseCompletedOutputFromPayload(t *testing.T) { } } +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"}`)} + + restored := restoreResponsesWebsocketCompletionOutput(payload, collector, nil) + if string(restored) != string(payload) { + t.Fatalf("non-empty completion output was overwritten: %s", restored) + } +} + func TestAppendWebsocketEvent(t *testing.T) { var builder strings.Builder @@ -1147,7 +1157,7 @@ func TestRecordResponsesWebsocketCustomToolCallsFromOutputItemDoneWithCache(t *t } } -func TestForwardResponsesWebsocketPreservesCompletedEvent(t *testing.T) { +func TestForwardResponsesWebsocketRestoresAndForwardsCompletedOutput(t *testing.T) { gin.SetMode(gin.TestMode) serverErrCh := make(chan error, 1) @@ -1167,9 +1177,10 @@ func TestForwardResponsesWebsocketPreservesCompletedEvent(t *testing.T) { ctx, _ := gin.CreateTestContext(httptest.NewRecorder()) ctx.Request = r - data := make(chan []byte, 1) + data := make(chan []byte, 2) errCh := make(chan *interfaces.ErrorMessage) - data <- []byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"output\":[{\"type\":\"message\",\"id\":\"out-1\"}]}}\n\n") + data <- []byte(`{"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","id":"call-1","call_id":"call-1","name":"lookup","arguments":"{}"}}`) + data <- []byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\",\"output\":[]}}\n\n") close(data) close(errCh) @@ -1191,16 +1202,16 @@ func TestForwardResponsesWebsocketPreservesCompletedEvent(t *testing.T) { serverErrCh <- fmt.Errorf("unexpected websocket error message: %v", errMsg.Error) return } - if gjson.GetBytes(completedOutput, "0.id").String() != "out-1" { - serverErrCh <- errors.New("completed output not captured") + if gjson.GetBytes(completedOutput, "0.id").String() != "call-1" { + serverErrCh <- errors.New("completed output not restored") 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) + if len(pendingToolCallIDs) != 1 || pendingToolCallIDs[0] != "call-1" { + serverErrCh <- fmt.Errorf("pending tool call ids = %v, want [call-1]", pendingToolCallIDs) return } if !strings.Contains(timelineLog.String(), "Event: websocket.response") { @@ -1223,9 +1234,17 @@ func TestForwardResponsesWebsocketPreservesCompletedEvent(t *testing.T) { } }() + _, outputItemPayload, errReadMessage := conn.ReadMessage() + if errReadMessage != nil { + t.Fatalf("read output item websocket message: %v", errReadMessage) + } + if got := gjson.GetBytes(outputItemPayload, "type").String(); got != "response.output_item.done" { + t.Fatalf("output item payload type = %s, want response.output_item.done", got) + } + _, payload, errReadMessage := conn.ReadMessage() if errReadMessage != nil { - t.Fatalf("read websocket message: %v", errReadMessage) + t.Fatalf("read completion websocket message: %v", errReadMessage) } if gjson.GetBytes(payload, "type").String() != wsEventTypeCompleted { t.Fatalf("payload type = %s, want %s", gjson.GetBytes(payload, "type").String(), wsEventTypeCompleted) @@ -1233,6 +1252,9 @@ func TestForwardResponsesWebsocketPreservesCompletedEvent(t *testing.T) { if strings.Contains(string(payload), "response.done") { t.Fatalf("payload unexpectedly rewrote completed event: %s", payload) } + if got := gjson.GetBytes(payload, "response.output.0.id").String(); got != "call-1" { + t.Fatalf("downstream completion output id = %q, want call-1; payload=%s", got, payload) + } if errServer := <-serverErrCh; errServer != nil { t.Fatalf("server error: %v", errServer) @@ -3066,6 +3088,60 @@ func TestNormalizeSubsequentRequestCompactSkipsMerge(t *testing.T) { } } +func TestNormalizeSubsequentRequestReasoningContinuationWithPreviousResponseID(t *testing.T) { + lastRequest := []byte(`{"model":"gpt-5.6-terra","stream":true,"input":[{"type":"message","role":"user","id":"old-user","content":"long history"}]}`) + lastResponseOutput := []byte(`[{"type":"function_call","id":"old-call","call_id":"old-call","name":"lookup","arguments":"{}"}]`) + + for _, requestType := range []string{"response.create", "response.append"} { + t.Run(requestType, func(t *testing.T) { + raw := []byte(`{"type":"` + requestType + `","previous_response_id":"resp-1","input":[ + {"type":"reasoning","id":"reasoning-1","summary":[]}, + {"type":"function_call_output","id":"output-1","call_id":"old-call","output":"result"} + ]}`) + + normalized, _, errMsg := normalizeResponsesWebsocketRequest(raw, lastRequest, lastResponseOutput) + 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; payload=%s", got, normalized) + } + input := gjson.GetBytes(normalized, "input").Array() + if len(input) != 2 || input[0].Get("id").String() != "reasoning-1" || input[1].Get("id").String() != "output-1" { + t.Fatalf("incremental continuation was replaced or merged: %s", normalized) + } + }) + } +} + +func TestResponsesWebsocketOutputCollectorRestoresCompletedOutput(t *testing.T) { + outputItemsByIndex := make(map[int64][]byte) + var outputItemsFallback [][]byte + 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"}}`), + } { + collectResponsesWebsocketOutputItem(payload, outputItemsByIndex, &outputItemsFallback) + } + + output := responseCompletedOutputFromPayload( + []byte(`{"type":"response.completed","response":{"id":"resp-1","output":[]}}`), + outputItemsByIndex, + outputItemsFallback, + ) + items := gjson.ParseBytes(output).Array() + if len(items) != 3 { + t.Fatalf("collected output len = %d, want 3: %s", len(items), output) + } + wantIDs := []string{"summary-1", "reply-1", "call-1"} + for i, wantID := range wantIDs { + if got := items[i].Get("id").String(); got != wantID { + t.Fatalf("output[%d].id = %q, want %q: %s", i, got, wantID, output) + } + } +} + func TestNormalizeSubsequentRequestCompactMergesWhenCompactionReplayUnsupported(t *testing.T) { lastRequest := []byte(`{"model":"gpt-5.4","stream":true,"input":[ {"type":"message","role":"user","id":"msg-1","content":"original long prompt"},