From 9d6b5cdd163ba06f24713e8082e96599400559d6 Mon Sep 17 00:00:00 2001 From: Yoga Sakti Date: Thu, 20 Aug 2026 23:52:09 +0700 Subject: [PATCH] fix(xai): treat response.incomplete as terminal success (#5113) The xAI executor only switched on response.completed, so any turn that ended with the spec-correct response.incomplete terminal event fell out of the loop. Non-streaming requests were reported to the client as a 408 ("stream disconnected before response.completed") even though the upstream request succeeded, and because 408 is retryable it burned credential rotations against a healthy pool. Streaming requests forwarded the terminal event without patching the collected output items or publishing usage. Accept response.incomplete alongside response.completed in both paths, mirroring the Codex executor, and keep the reasoning replay cache gated on response.completed since a truncated turn has no replayable state. --- .../runtime/executor/xai_executor_execute.go | 13 ++- .../runtime/executor/xai_executor_stream.go | 8 +- .../runtime/executor/xai_executor_test.go | 90 +++++++++++++++++++ 3 files changed, 105 insertions(+), 6 deletions(-) diff --git a/internal/runtime/executor/xai_executor_execute.go b/internal/runtime/executor/xai_executor_execute.go index 72f769729..48a6aca9d 100644 --- a/internal/runtime/executor/xai_executor_execute.go +++ b/internal/runtime/executor/xai_executor_execute.go @@ -94,16 +94,21 @@ func (e *XAIExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req if len(eventData) == 0 { continue } - switch gjson.GetBytes(eventData, "type").String() { + eventType := gjson.GetBytes(eventData, "type").String() + switch eventType { case "response.output_item.done": xaiCollectOutputItemDone(eventData, outputItemsByIndex, &outputItemsFallback) - case "response.completed": + case "response.completed", "response.incomplete": if detail, ok := helps.ParseCodexUsage(eventData); ok { reporter.Publish(ctx, detail) } completedData := xaiPatchCompletedOutput(eventData, outputItemsByIndex, outputItemsFallback) completedData = xaiNormalizeReasoningSummaryData(completedData) - cacheXAIReasoningReplayFromCompleted(ctx, prepared.replayScope, completedData) + if eventType == "response.completed" { + // A truncated turn carries no replayable terminal state, so only a + // completed response may refresh the reasoning replay cache. + cacheXAIReasoningReplayFromCompleted(ctx, prepared.replayScope, completedData) + } var param any out := sdktranslator.TranslateNonStream(ctx, prepared.to, prepared.responseFormat, req.Model, prepared.originalPayload, prepared.body, completedData, ¶m) if prepared.responseFormat == sdktranslator.FormatOpenAIResponse { @@ -113,7 +118,7 @@ func (e *XAIExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req } } - return resp, statusErr{code: http.StatusRequestTimeout, msg: "xai stream error: stream disconnected before response.completed"} + return resp, statusErr{code: http.StatusRequestTimeout, msg: "xai stream error: stream disconnected before response.completed or response.incomplete"} } func (e *XAIExecutor) executeCompact(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { diff --git a/internal/runtime/executor/xai_executor_stream.go b/internal/runtime/executor/xai_executor_stream.go index 4dad2c51f..2b5f3e86f 100644 --- a/internal/runtime/executor/xai_executor_stream.go +++ b/internal/runtime/executor/xai_executor_stream.go @@ -121,13 +121,17 @@ func (e *XAIExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth switch normalizedEventName { case "response.output_item.done": xaiCollectOutputItemDone(eventData, outputItemsByIndex, &outputItemsFallback) - case "response.completed": + case "response.completed", "response.incomplete": if detail, ok := helps.ParseCodexUsage(eventData); ok { reporter.Publish(ctx, detail) } eventData = xaiPatchCompletedOutput(eventData, outputItemsByIndex, outputItemsFallback) eventData = xaiNormalizeReasoningSummaryData(eventData) - cacheXAIReasoningReplayFromCompleted(ctx, prepared.replayScope, eventData) + if normalizedEventName == "response.completed" { + // A truncated turn carries no replayable terminal state, so only a + // completed response may refresh the reasoning replay cache. + cacheXAIReasoningReplayFromCompleted(ctx, prepared.replayScope, eventData) + } normalizedEventName = gjson.GetBytes(eventData, "type").String() } diff --git a/internal/runtime/executor/xai_executor_test.go b/internal/runtime/executor/xai_executor_test.go index 15de2a67a..8be5a3d54 100644 --- a/internal/runtime/executor/xai_executor_test.go +++ b/internal/runtime/executor/xai_executor_test.go @@ -656,6 +656,96 @@ func TestXAIExecutorExecuteFiltersInternalXSearchCalls(t *testing.T) { } } +func TestXAIExecutorExecuteAcceptsResponseIncomplete(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"id\":\"rs_1\",\"type\":\"reasoning\",\"summary\":[]}}\n\n")) + _, _ = w.Write([]byte("data: {\"type\":\"response.incomplete\",\"response\":{\"id\":\"resp_1\",\"object\":\"response\",\"status\":\"incomplete\",\"incomplete_details\":{\"reason\":\"max_output_tokens\"},\"output\":[],\"usage\":{\"input_tokens\":8,\"output_tokens\":1,\"total_tokens\":9}}}\n\n")) + })) + defer server.Close() + + exec := NewXAIExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{ + Provider: "xai", + Attributes: map[string]string{"base_url": server.URL}, + Metadata: map[string]any{"access_token": "xai-token"}, + } + resp, err := exec.Execute(context.Background(), auth, cliproxyexecutor.Request{ + Model: "grok-4.5", + Payload: []byte(`{"model":"grok-4.5","input":"hi","max_output_tokens":1}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FormatOpenAIResponse, + Stream: false, + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + if got := gjson.GetBytes(resp.Payload, "status").String(); got != "incomplete" { + t.Fatalf("status = %q, want incomplete; payload=%s", got, resp.Payload) + } + if got := gjson.GetBytes(resp.Payload, "incomplete_details.reason").String(); got != "max_output_tokens" { + t.Fatalf("incomplete reason = %q, want max_output_tokens; payload=%s", got, resp.Payload) + } + if got := gjson.GetBytes(resp.Payload, "output.#").Int(); got != 1 { + t.Fatalf("output length = %d, want 1; payload=%s", got, resp.Payload) + } +} + +func TestXAIExecutorExecuteStreamAcceptsResponseIncomplete(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = fmt.Fprint(w, "event: response.output_item.done\ndata: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"id\":\"rs_1\",\"type\":\"reasoning\",\"summary\":[]}}\n\n") + _, _ = fmt.Fprint(w, "event: response.incomplete\ndata: {\"type\":\"response.incomplete\",\"response\":{\"id\":\"resp_1\",\"object\":\"response\",\"status\":\"incomplete\",\"incomplete_details\":{\"reason\":\"max_output_tokens\"},\"output\":[],\"usage\":{\"input_tokens\":8,\"output_tokens\":1,\"total_tokens\":9}}}\n\n") + })) + defer server.Close() + + exec := NewXAIExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{ + Provider: "xai", + Attributes: map[string]string{"base_url": server.URL}, + Metadata: map[string]any{"access_token": "xai-token"}, + } + result, err := exec.ExecuteStream(context.Background(), auth, cliproxyexecutor.Request{ + Model: "grok-4.5", + Payload: []byte(`{"model":"grok-4.5","input":"hi","max_output_tokens":1}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FormatOpenAIResponse, + Stream: true, + }) + if err != nil { + t.Fatalf("ExecuteStream() error = %v", err) + } + + var stream bytes.Buffer + for chunk := range result.Chunks { + if chunk.Err != nil { + t.Fatalf("stream chunk error = %v", chunk.Err) + } + stream.Write(chunk.Payload) + stream.WriteByte('\n') + } + + var incomplete gjson.Result + for _, line := range strings.Split(stream.String(), "\n") { + line = strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if !gjson.Valid(line) { + continue + } + if event := gjson.Parse(line); event.Get("type").String() == "response.incomplete" { + incomplete = event + } + } + if !incomplete.Exists() { + t.Fatalf("no response.incomplete chunk forwarded: %s", stream.String()) + } + if got := incomplete.Get("response.output.#").Int(); got != 1 { + t.Fatalf("incomplete output length = %d, want 1; event=%s", got, incomplete.Raw) + } + if got := incomplete.Get("response.usage.total_tokens").Int(); got != 9 { + t.Fatalf("incomplete usage total_tokens = %d, want 9; event=%s", got, incomplete.Raw) + } +} + func TestXAIExecutorPrepareHonorsInjectXSearchConfig(t *testing.T) { t.Parallel()