diff --git a/backend/internal/service/openai_ws_forwarder_ingress.go b/backend/internal/service/openai_ws_forwarder_ingress.go index ca6135352b..87d733b539 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress.go +++ b/backend/internal/service/openai_ws_forwarder_ingress.go @@ -1340,6 +1340,7 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( turnPrevRecoveryTried := false lastTurnFinishedAt := time.Time{} lastTurnResponseID := "" + lastTurnWindowID := "" lastTurnPayload := []byte(nil) var lastTurnStrictState *openAIWSIngressPreviousTurnStrictState lastTurnReplayInput := []json.RawMessage(nil) @@ -1484,8 +1485,33 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( lastTurnReplayInput, _ = stripOpenAIInvalidEncryptedContentFromReplayItems(lastTurnReplayInput, invalidDigests) } } + boundaryPayload, contextWindowBoundary, boundaryErr := normalizeOpenAIWSContextWindowBoundary( + currentPayload, + lastTurnWindowID, + ) + if boundaryErr != nil { + return fmt.Errorf("normalize Codex websocket context-window boundary: %w", boundaryErr) + } + if contextWindowBoundary.Changed { + currentPayload = boundaryPayload + currentPayloadBytes = len(boundaryPayload) + logOpenAIWSModeInfo( + "ingress_ws_context_window_changed account_id=%d turn=%d conn_id=%s action=break_previous_response_chain previous_window_id=%s current_window_id=%s previous_response_id_removed=%v", + account.ID, + turn, + truncateOpenAIWSLogValue(sessionConnID, openAIWSIDValueMaxLen), + truncateOpenAIWSLogValue(lastTurnWindowID, openAIWSIDValueMaxLen), + truncateOpenAIWSLogValue(contextWindowBoundary.WindowID, openAIWSIDValueMaxLen), + contextWindowBoundary.PreviousResponseIDRemoved, + ) + } currentPreviousResponseID := openAIWSPayloadStringFromRaw(currentPayload, "previous_response_id") expectedPrev := strings.TrimSpace(lastTurnResponseID) + if contextWindowBoundary.Changed { + // A context-window rollover is a new Responses root. Do not infer a + // continuation anchor from the response produced in the old window. + expectedPrev = "" + } toolSignals := ToolContinuationSignals{ HasFunctionCallOutput: openAIWSRawPayloadHasToolCallOutput(currentPayload), } @@ -1822,6 +1848,9 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient( } responseID := strings.TrimSpace(result.RequestID) lastTurnResponseID = responseID + if contextWindowBoundary.WindowID != "" { + lastTurnWindowID = contextWindowBoundary.WindowID + } // 正文共享:currentPayload/currentTurnReplayInput 均不可变,历史直接引用; // collector 增量经 combine 合并(新头数组)。 lastTurnReplayInput = currentTurnReplayInput diff --git a/backend/internal/service/openai_ws_forwarder_ingress_test.go b/backend/internal/service/openai_ws_forwarder_ingress_test.go index b70f83659a..915a4edf24 100644 --- a/backend/internal/service/openai_ws_forwarder_ingress_test.go +++ b/backend/internal/service/openai_ws_forwarder_ingress_test.go @@ -144,6 +144,49 @@ func TestDropPreviousResponseIDFromRawPayload(t *testing.T) { }) } +func TestNormalizeOpenAIWSContextWindowBoundary(t *testing.T) { + t.Parallel() + + t.Run("same_window_keeps_previous_response_id", func(t *testing.T) { + payload := []byte("{\"type\":\"response.create\",\"previous_response_id\":\"resp_old\",\"client_metadata\":{\"x-codex-window-id\":\"window-a\"}}") + updated, boundary, err := normalizeOpenAIWSContextWindowBoundary(payload, "window-a") + require.NoError(t, err) + require.False(t, boundary.Changed) + require.False(t, boundary.PreviousResponseIDRemoved) + require.Equal(t, "window-a", boundary.WindowID) + require.Equal(t, "resp_old", gjson.GetBytes(updated, "previous_response_id").String()) + }) + + t.Run("new_window_drops_previous_response_id", func(t *testing.T) { + payload := []byte("{\"type\":\"response.create\",\"previous_response_id\":\"resp_old\",\"client_metadata\":{\"x-codex-window-id\":\"window-b\"}}") + updated, boundary, err := normalizeOpenAIWSContextWindowBoundary(payload, "window-a") + require.NoError(t, err) + require.True(t, boundary.Changed) + require.True(t, boundary.PreviousResponseIDRemoved) + require.Equal(t, "window-b", boundary.WindowID) + require.False(t, gjson.GetBytes(updated, "previous_response_id").Exists()) + }) + + t.Run("new_window_without_previous_response_id_still_marks_boundary", func(t *testing.T) { + payload := []byte("{\"type\":\"response.create\",\"client_metadata\":{\"x-codex-window-id\":\"window-b\"}}") + updated, boundary, err := normalizeOpenAIWSContextWindowBoundary(payload, "window-a") + require.NoError(t, err) + require.True(t, boundary.Changed) + require.False(t, boundary.PreviousResponseIDRemoved) + require.Equal(t, string(payload), string(updated)) + }) + + t.Run("embedded_turn_metadata_is_fallback", func(t *testing.T) { + payload := []byte("{\"type\":\"response.create\",\"previous_response_id\":\"resp_old\",\"client_metadata\":{\"x-codex-turn-metadata\":\"{\\\"window_id\\\":\\\"window-b\\\"}\"}}") + updated, boundary, err := normalizeOpenAIWSContextWindowBoundary(payload, "window-a") + require.NoError(t, err) + require.True(t, boundary.Changed) + require.True(t, boundary.PreviousResponseIDRemoved) + require.Equal(t, "window-b", boundary.WindowID) + require.False(t, gjson.GetBytes(updated, "previous_response_id").Exists()) + }) +} + func TestStripCodexSparkImageGenerationToolFromRawPayload(t *testing.T) { t.Run("strips_image_generation_for_spark", func(t *testing.T) { payload := []byte(`{"type":"response.create","model":"gpt-5.3-codex-spark","tools":[{"type":"function","name":"shell"},{"type":"image_generation","output_format":"png"}]}`) diff --git a/backend/internal/service/openai_ws_forwarder_payload.go b/backend/internal/service/openai_ws_forwarder_payload.go index bc24ebc265..16ca9c8a63 100644 --- a/backend/internal/service/openai_ws_forwarder_payload.go +++ b/backend/internal/service/openai_ws_forwarder_payload.go @@ -378,6 +378,48 @@ func setPreviousResponseIDToRawPayload(payload []byte, previousResponseID string return rebuilt, nil } +type openAIWSContextWindowBoundary struct { + WindowID string + Changed bool + PreviousResponseIDRemoved bool +} + +func openAIWSPayloadCodexWindowID(payload []byte) string { + if len(payload) == 0 { + return "" + } + if windowID := strings.TrimSpace(gjson.GetBytes(payload, "client_metadata.x-codex-window-id").String()); windowID != "" { + return windowID + } + turnMetadata := strings.TrimSpace(gjson.GetBytes(payload, "client_metadata.x-codex-turn-metadata").String()) + if turnMetadata == "" { + return "" + } + return strings.TrimSpace(gjson.Get(turnMetadata, "window_id").String()) +} + +// normalizeOpenAIWSContextWindowBoundary breaks a Responses continuation chain +// when Codex moves to a new local context window. WebSocket response.create can +// still carry the previous window's previous_response_id after new_context, +// while HTTP starts the new window without that continuation anchor. +func normalizeOpenAIWSContextWindowBoundary( + payload []byte, + previousWindowID string, +) ([]byte, openAIWSContextWindowBoundary, error) { + currentWindowID := openAIWSPayloadCodexWindowID(payload) + boundary := openAIWSContextWindowBoundary{WindowID: currentWindowID} + if previousWindowID == "" || currentWindowID == "" || currentWindowID == previousWindowID { + return payload, boundary, nil + } + boundary.Changed = true + updated, removed, err := dropPreviousResponseIDFromRawPayload(payload) + if err != nil { + return payload, boundary, err + } + boundary.PreviousResponseIDRemoved = removed + return updated, boundary, nil +} + func shouldInferIngressFunctionCallOutputPreviousResponseID( storeDisabled bool, turn int,