lywsvip/CLIProxyAPI
1
0
Fork 0
mirror of https://github.com/router-for-me/CLIProxyAPI.git synced 2026-09-03 06:35:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(xai): treat response.incomplete as terminal success (#5113)

Browse Source 
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.
This commit is contained in:
Yoga Sakti
2026-08-20 23:52:09 +07:00
parent 4b9d404fb0
commit 9d6b5cdd16
3 changed files with 105 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
internal/runtime/executor/xai_executor_execute.go
View File

@@ -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, &param)
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) {

8
internal/runtime/executor/xai_executor_stream.go
View File

@@ -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()
}

90
internal/runtime/executor/xai_executor_test.go
View File

@@ -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()