diff --git a/internal/runtime/executor/xai_executor.go b/internal/runtime/executor/xai_executor.go index 45ab5422b..1309b22ed 100644 --- a/internal/runtime/executor/xai_executor.go +++ b/internal/runtime/executor/xai_executor.go @@ -45,6 +45,7 @@ const ( xaiNamespaceToolType = "namespace" xaiToolSearchType = "tool_search" xaiWebSearchToolType = "web_search" + xaiXSearchToolType = "x_search" // Codex Desktop injects codex_app.automation_update with a large oneOf+$ref // schema. xAI's free/build Responses path accepts the HTTP request but never // emits SSE when that schema is present, so Desktop hangs on "thinking". @@ -183,12 +184,17 @@ func (e *XAIExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req outputItemsByIndex := make(map[int64][]byte) var outputItemsFallback [][]byte + responseFilter := newXAIInternalXSearchResponseFilter(prepared.filterInternalXSearch) for _, line := range bytes.Split(data, []byte("\n")) { if !bytes.HasPrefix(line, xaiDataTag) { continue } eventData := xaiNormalizeReasoningSummaryData(bytes.TrimSpace(line[len(xaiDataTag):])) eventData = restoreXAINamespaceToolCalls(eventData, prepared.namespaceTools) + eventData = responseFilter.apply(eventData) + if len(eventData) == 0 { + continue + } switch gjson.GetBytes(eventData, "type").String() { case "response.output_item.done": xaiCollectOutputItemDone(eventData, outputItemsByIndex, &outputItemsFallback) @@ -645,6 +651,7 @@ func (e *XAIExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth var param any outputItemsByIndex := make(map[int64][]byte) var outputItemsFallback [][]byte + responseFilter := newXAIInternalXSearchResponseFilter(prepared.filterInternalXSearch) var pendingEventLine []byte emitTranslatedLine := func(translatedLine []byte) bool { chunks := sdktranslator.TranslateStream(ctx, prepared.to, prepared.responseFormat, req.Model, prepared.originalPayload, prepared.body, translatedLine, ¶m) @@ -674,6 +681,13 @@ func (e *XAIExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth hasPendingEventLine := pendingEventLine != nil for i, eventData := range eventDataList { eventData = restoreXAINamespaceToolCalls(eventData, prepared.namespaceTools) + eventData = responseFilter.apply(eventData) + if len(eventData) == 0 { + if hasPendingEventLine && i == 0 { + pendingEventLine = nil + } + continue + } normalizedEventName := gjson.GetBytes(eventData, "type").String() switch normalizedEventName { case "response.output_item.done": @@ -813,15 +827,16 @@ func (e *XAIExecutor) Refresh(ctx context.Context, auth *cliproxyauth.Auth) (*cl } type xaiPreparedRequest struct { - baseModel string - from sdktranslator.Format - responseFormat sdktranslator.Format - to sdktranslator.Format - originalPayload []byte - body []byte - namespaceTools map[string]xaiNamespaceToolRef - sessionID string - replayScope xaiReasoningReplayScope + baseModel string + from sdktranslator.Format + responseFormat sdktranslator.Format + to sdktranslator.Format + originalPayload []byte + body []byte + namespaceTools map[string]xaiNamespaceToolRef + sessionID string + replayScope xaiReasoningReplayScope + filterInternalXSearch bool } type xaiNamespaceToolRef struct { @@ -869,6 +884,7 @@ func (e *XAIExecutor) prepareResponsesRequestTo(ctx context.Context, req cliprox if err != nil { return nil, err } + body = normalizeXAIInputCustomToolCalls(body) body = normalizeXAIInputNamespaceToolCalls(body) body = normalizeXAIInputReasoningItems(body) body = sanitizeXAIInputEncryptedContent(body) @@ -884,15 +900,16 @@ func (e *XAIExecutor) prepareResponsesRequestTo(ctx context.Context, req cliprox } return &xaiPreparedRequest{ - baseModel: baseModel, - from: from, - responseFormat: responseFormat, - to: to, - originalPayload: originalPayload, - body: body, - namespaceTools: namespaceTools, - sessionID: sessionID, - replayScope: replayScope, + baseModel: baseModel, + from: from, + responseFormat: responseFormat, + to: to, + originalPayload: originalPayload, + body: body, + namespaceTools: namespaceTools, + sessionID: sessionID, + replayScope: replayScope, + filterInternalXSearch: xaiRequestHasNativeXSearch(body), }, nil } @@ -1455,6 +1472,252 @@ func collectXAINamespaceToolRefs(body []byte) map[string]xaiNamespaceToolRef { return refs } +func normalizeXAIInputCustomToolCalls(body []byte) []byte { + input := gjson.GetBytes(body, "input") + if !input.Exists() || !input.IsArray() { + return body + } + + changed := false + inputArray := input.Array() + items := make([]json.RawMessage, 0, len(inputArray)) + for _, item := range inputArray { + var normalized []byte + switch item.Get("type").String() { + case "custom_tool_call": + callID := strings.TrimSpace(item.Get("call_id").String()) + name := strings.TrimSpace(item.Get("name").String()) + if callID == "" || name == "" { + changed = true + continue + } + normalized = []byte(`{"type":"function_call"}`) + normalized, _ = sjson.SetBytes(normalized, "call_id", callID) + normalized, _ = sjson.SetBytes(normalized, "name", name) + normalized, _ = sjson.SetBytes(normalized, "arguments", xaiCustomToolCallArguments(item.Get("input"))) + case "custom_tool_call_output": + callID := strings.TrimSpace(item.Get("call_id").String()) + if callID == "" { + changed = true + continue + } + normalized = []byte(`{"type":"function_call_output"}`) + normalized, _ = sjson.SetBytes(normalized, "call_id", callID) + normalized, _ = sjson.SetBytes(normalized, "output", xaiCustomToolCallOutput(item.Get("output"))) + default: + items = append(items, json.RawMessage(item.Raw)) + continue + } + items = append(items, json.RawMessage(normalized)) + changed = true + } + if !changed { + return body + } + + rawInput, errMarshal := json.Marshal(items) + if errMarshal != nil { + return body + } + updated, errSet := sjson.SetRawBytes(body, "input", rawInput) + if errSet != nil { + return body + } + return updated +} + +func xaiCustomToolCallArguments(input gjson.Result) string { + if !input.Exists() { + return "{}" + } + if input.Type == gjson.String { + text := input.String() + trimmed := strings.TrimSpace(text) + if gjson.Valid(trimmed) { + parsed := gjson.Parse(trimmed) + if parsed.IsObject() { + return parsed.Raw + } + } + encoded, errMarshal := json.Marshal(text) + if errMarshal != nil { + return "{}" + } + return `{"input":` + string(encoded) + `}` + } + if input.IsObject() { + return input.Raw + } + if input.Raw != "" { + return `{"input":` + input.Raw + `}` + } + return "{}" +} + +func xaiCustomToolCallOutput(output gjson.Result) string { + if !output.Exists() { + return "" + } + if output.Type == gjson.String { + return output.String() + } + return output.Raw +} + +// xAI executes these x_search subtools server-side but exposes their trace as +// client-style tool calls. Hide the trace so Responses clients do not execute it again. +type xaiInternalXSearchResponseFilter struct { + enabled bool + droppedOutputIndexes map[int64]struct{} + droppedItemIDs map[string]struct{} +} + +func newXAIInternalXSearchResponseFilter(enabled bool) *xaiInternalXSearchResponseFilter { + filter := &xaiInternalXSearchResponseFilter{enabled: enabled} + if enabled { + filter.droppedOutputIndexes = make(map[int64]struct{}) + filter.droppedItemIDs = make(map[string]struct{}) + } + return filter +} + +func xaiRequestHasNativeXSearch(body []byte) bool { + hasXSearch := func(tools gjson.Result) bool { + if !tools.IsArray() { + return false + } + for _, tool := range tools.Array() { + if strings.TrimSpace(tool.Get("type").String()) == xaiXSearchToolType { + return true + } + } + return false + } + if hasXSearch(gjson.GetBytes(body, "tools")) { + return true + } + input := gjson.GetBytes(body, "input") + if !input.IsArray() { + return false + } + for _, item := range input.Array() { + if item.Get("type").String() == "additional_tools" && hasXSearch(item.Get("tools")) { + return true + } + } + return false +} + +func xaiIsInternalXSearchToolName(name string) bool { + switch strings.TrimSpace(name) { + case "x_user_search", "x_semantic_search", "x_keyword_search", "x_thread_fetch": + return true + default: + return false + } +} + +func xaiIsInternalXSearchCall(item gjson.Result) bool { + itemType := strings.TrimSpace(item.Get("type").String()) + if itemType != "custom_tool_call" && itemType != "function_call" { + return false + } + return xaiIsInternalXSearchToolName(item.Get("name").String()) +} + +func (f *xaiInternalXSearchResponseFilter) apply(eventData []byte) []byte { + if f == nil || !f.enabled || len(eventData) == 0 || !gjson.ValidBytes(eventData) { + return eventData + } + + if item := gjson.GetBytes(eventData, "item"); xaiIsInternalXSearchCall(item) { + f.recordDroppedItem(eventData, item) + return nil + } + + eventData = filterXAIInternalXSearchCompletedOutput(eventData) + if f.referencesDroppedItem(eventData) { + return nil + } + return f.compactOutputIndex(eventData) +} + +func (f *xaiInternalXSearchResponseFilter) recordDroppedItem(eventData []byte, item gjson.Result) { + if outputIndex := gjson.GetBytes(eventData, "output_index"); outputIndex.Exists() { + f.droppedOutputIndexes[outputIndex.Int()] = struct{}{} + } + for _, path := range []string{"id", "call_id"} { + if id := strings.TrimSpace(item.Get(path).String()); id != "" { + f.droppedItemIDs[id] = struct{}{} + } + } +} + +func (f *xaiInternalXSearchResponseFilter) referencesDroppedItem(eventData []byte) bool { + if outputIndex := gjson.GetBytes(eventData, "output_index"); outputIndex.Exists() { + if _, dropped := f.droppedOutputIndexes[outputIndex.Int()]; dropped { + return true + } + } + for _, path := range []string{"item_id", "call_id"} { + id := strings.TrimSpace(gjson.GetBytes(eventData, path).String()) + if _, dropped := f.droppedItemIDs[id]; id != "" && dropped { + return true + } + } + return false +} + +func (f *xaiInternalXSearchResponseFilter) compactOutputIndex(eventData []byte) []byte { + outputIndex := gjson.GetBytes(eventData, "output_index") + if !outputIndex.Exists() { + return eventData + } + original := outputIndex.Int() + removedBefore := int64(0) + for dropped := range f.droppedOutputIndexes { + if dropped < original { + removedBefore++ + } + } + if removedBefore == 0 { + return eventData + } + updated, errSet := sjson.SetBytes(eventData, "output_index", original-removedBefore) + if errSet != nil { + return eventData + } + return updated +} + +func filterXAIInternalXSearchCompletedOutput(eventData []byte) []byte { + output := gjson.GetBytes(eventData, "response.output") + if !output.IsArray() { + return eventData + } + items := make([]json.RawMessage, 0, len(output.Array())) + changed := false + for _, item := range output.Array() { + if xaiIsInternalXSearchCall(item) { + changed = true + continue + } + items = append(items, json.RawMessage(item.Raw)) + } + if !changed { + return eventData + } + rawOutput, errMarshal := json.Marshal(items) + if errMarshal != nil { + return eventData + } + updated, errSet := sjson.SetRawBytes(eventData, "response.output", rawOutput) + if errSet != nil { + return eventData + } + return updated +} + func normalizeXAIInputNamespaceToolCalls(body []byte) []byte { if !gjson.ValidBytes(body) { return body diff --git a/internal/runtime/executor/xai_executor_test.go b/internal/runtime/executor/xai_executor_test.go index 929e58c9d..f52a38476 100644 --- a/internal/runtime/executor/xai_executor_test.go +++ b/internal/runtime/executor/xai_executor_test.go @@ -6,6 +6,7 @@ import ( "crypto/sha256" "encoding/base64" "errors" + "fmt" "io" "net/http" "net/http/httptest" @@ -249,6 +250,235 @@ func TestXAIExecutorExecuteRestoresAdditionalToolsNamespaceCalls(t *testing.T) { } } +func TestXAIExecutorExecuteNormalizesCustomToolCallHistory(t *testing.T) { + var gotBody []byte + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var errRead error + gotBody, errRead = io.ReadAll(r.Body) + if errRead != nil { + t.Fatalf("read body: %v", errRead) + } + for _, item := range gjson.GetBytes(gotBody, "input").Array() { + if strings.HasPrefix(item.Get("type").String(), "custom_tool_call") { + http.Error(w, `{"error":"data did not match any variant of untagged enum ModelInput"}`, http.StatusUnprocessableEntity) + return + } + } + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"object\":\"response\",\"created_at\":0,\"status\":\"completed\",\"model\":\"grok-4.5\",\"output\":[{\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"ok\"}]}]}}\n\n")) + })) + defer server.Close() + + exec := NewXAIExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{ + Provider: "xai", + Attributes: map[string]string{ + "base_url": server.URL, + "auth_kind": "oauth", + }, + Metadata: map[string]any{"access_token": "xai-token"}, + } + payload := []byte(`{ + "model":"grok-4.5", + "input":[ + {"type":"message","role":"user","content":[{"type":"input_text","text":"search"}]}, + {"type":"custom_tool_call","name":"missing_call_id","input":"invalid"}, + {"type":"custom_tool_call_output","output":"missing call id"}, + {"type":"custom_tool_call","status":"completed","call_id":"xs_call-1","name":"x_semantic_search","input":"{\"query\":\"US stocks\",\"limit\":\"10\"}","internal_chat_message_metadata_passthrough":{"turn_id":"turn-1"}}, + {"type":"custom_tool_call_output","call_id":"xs_call-1","output":"unsupported custom tool call: x_semantic_search","internal_chat_message_metadata_passthrough":{"turn_id":"turn-1"}}, + {"type":"custom_tool_call","call_id":"call-2","name":"apply_patch","input":"*** Begin Patch"}, + {"type":"custom_tool_call_output","call_id":"call-2","output":[{"type":"input_text","text":"done"}]} + ], + "tools":[{"type":"x_search"}], + "tool_choice":"auto" + }`) + + _, err := exec.Execute(context.Background(), auth, cliproxyexecutor.Request{ + Model: "grok-4.5", + Payload: payload, + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FormatOpenAIResponse, + Stream: false, + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + + input := gjson.GetBytes(gotBody, "input").Array() + if len(input) != 5 { + t.Fatalf("input length = %d, want 5; body=%s", len(input), gotBody) + } + if got := input[1].Get("type").String(); got != "function_call" { + t.Fatalf("input.1.type = %q, want function_call; body=%s", got, gotBody) + } + if got := gjson.Get(input[1].Get("arguments").String(), "query").String(); got != "US stocks" { + t.Fatalf("input.1 arguments query = %q, want US stocks; body=%s", got, gotBody) + } + if input[1].Get("input").Exists() || input[1].Get("internal_chat_message_metadata_passthrough").Exists() { + t.Fatalf("input.1 contains unsupported custom fields: %s", input[1].Raw) + } + if got := input[2].Get("type").String(); got != "function_call_output" { + t.Fatalf("input.2.type = %q, want function_call_output; body=%s", got, gotBody) + } + if got := input[2].Get("output").String(); got != "unsupported custom tool call: x_semantic_search" { + t.Fatalf("input.2.output = %q; body=%s", got, gotBody) + } + if got := gjson.Get(input[3].Get("arguments").String(), "input").String(); got != "*** Begin Patch" { + t.Fatalf("input.3 freeform arguments = %q, want patch input; body=%s", got, gotBody) + } + if got := input[4].Get("output").String(); got != `[{"type":"input_text","text":"done"}]` { + t.Fatalf("input.4 output = %q, want flattened JSON string; body=%s", got, gotBody) + } + if got := gjson.GetBytes(gotBody, "tools.0.type").String(); got != "x_search" { + t.Fatalf("tools.0.type = %q, want x_search; body=%s", got, gotBody) + } +} + +func TestXAIExecutorExecuteStreamFiltersInternalXSearchCalls(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + names := []string{"x_user_search", "x_semantic_search", "x_keyword_search", "x_thread_fetch"} + completed := []byte(`{"type":"response.completed","response":{"id":"resp_1","object":"response","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}`) + for i, name := range names { + itemID := fmt.Sprintf("ctc_%d", i) + callID := fmt.Sprintf("xs_call-%d", i) + _, _ = fmt.Fprintf(w, "event: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"output_index\":%d,\"item\":{\"id\":%q,\"type\":\"custom_tool_call\",\"call_id\":%q,\"name\":%q,\"input\":\"\",\"status\":\"in_progress\"}}\n\n", i, itemID, callID, name) + _, _ = fmt.Fprintf(w, "event: response.custom_tool_call_input.done\ndata: {\"type\":\"response.custom_tool_call_input.done\",\"output_index\":%d,\"item_id\":%q,\"input\":\"{}\"}\n\n", i, itemID) + _, _ = fmt.Fprintf(w, "event: response.output_item.done\ndata: {\"type\":\"response.output_item.done\",\"output_index\":%d,\"item\":{\"id\":%q,\"type\":\"custom_tool_call\",\"call_id\":%q,\"name\":%q,\"input\":\"{}\",\"status\":\"completed\"}}\n\n", i, itemID, callID, name) + item := []byte(`{"id":"","type":"custom_tool_call","call_id":"","name":"","input":"{}","status":"completed"}`) + item, _ = sjson.SetBytes(item, "id", itemID) + item, _ = sjson.SetBytes(item, "call_id", callID) + item, _ = sjson.SetBytes(item, "name", name) + completed, _ = sjson.SetRawBytes(completed, "response.output.-1", item) + } + + messageIndex := len(names) + _, _ = fmt.Fprintf(w, "event: response.output_item.added\ndata: {\"type\":\"response.output_item.added\",\"output_index\":%d,\"item\":{\"id\":\"msg_1\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[],\"status\":\"in_progress\"}}\n\n", messageIndex) + _, _ = fmt.Fprintf(w, "event: response.output_text.delta\ndata: {\"type\":\"response.output_text.delta\",\"output_index\":%d,\"item_id\":\"msg_1\",\"content_index\":0,\"delta\":\"answer\"}\n\n", messageIndex) + _, _ = fmt.Fprintf(w, "event: response.output_item.done\ndata: {\"type\":\"response.output_item.done\",\"output_index\":%d,\"item\":{\"id\":\"msg_1\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"answer\"}],\"status\":\"completed\"}}\n\n", messageIndex) + message := []byte(`{"id":"msg_1","type":"message","role":"assistant","content":[{"type":"output_text","text":"answer"}],"status":"completed"}`) + completed, _ = sjson.SetRawBytes(completed, "response.output.-1", message) + _, _ = fmt.Fprintf(w, "event: response.completed\ndata: %s\n\n", completed) + })) + 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":"search X","tools":[{"type":"x_search"}]}`), + }, 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') + } + streamText := stream.String() + for _, name := range []string{"x_user_search", "x_semantic_search", "x_keyword_search", "x_thread_fetch"} { + if strings.Contains(streamText, name) { + t.Fatalf("internal x_search call %q leaked downstream: %s", name, streamText) + } + } + if strings.Contains(streamText, "response.custom_tool_call_input") { + t.Fatalf("custom tool input event leaked downstream: %s", streamText) + } + + var completed gjson.Result + messageIndexChecks := 0 + for _, line := range strings.Split(streamText, "\n") { + line = strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if !gjson.Valid(line) { + continue + } + event := gjson.Parse(line) + if event.Get("item.id").String() == "msg_1" || event.Get("item_id").String() == "msg_1" { + messageIndexChecks++ + if got := event.Get("output_index").Int(); got != 0 { + t.Fatalf("message output_index = %d, want 0; event=%s", got, line) + } + } + if event.Get("type").String() == "response.completed" { + completed = event + } + } + if messageIndexChecks == 0 { + t.Fatal("no message events found") + } + if got := completed.Get("response.output.#").Int(); got != 1 { + t.Fatalf("completed output length = %d, want 1; completed=%s", got, completed.Raw) + } + if got := completed.Get("response.output.0.type").String(); got != "message" { + t.Fatalf("completed output type = %q, want message; completed=%s", got, completed.Raw) + } +} + +func TestXAIExecutorExecuteFiltersInternalXSearchCalls(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\":\"ctc_1\",\"type\":\"custom_tool_call\",\"call_id\":\"xs_call-1\",\"name\":\"x_user_search\",\"input\":\"{}\",\"status\":\"completed\"}}\n\n")) + _, _ = w.Write([]byte("data: {\"type\":\"response.output_item.done\",\"output_index\":1,\"item\":{\"id\":\"msg_1\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"answer\"}],\"status\":\"completed\"}}\n\n")) + _, _ = w.Write([]byte("data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"object\":\"response\",\"status\":\"completed\",\"output\":[{\"id\":\"ctc_1\",\"type\":\"custom_tool_call\",\"call_id\":\"xs_call-1\",\"name\":\"x_user_search\",\"input\":\"{}\"},{\"id\":\"msg_1\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"answer\"}]}]}}\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":"search X","tools":[{"type":"x_search"}]}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FormatOpenAIResponse, + Stream: false, + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + if strings.Contains(string(resp.Payload), "x_user_search") || strings.Contains(string(resp.Payload), "custom_tool_call") { + t.Fatalf("internal X search call leaked into response: %s", resp.Payload) + } + if got := gjson.GetBytes(resp.Payload, "output.#").Int(); got != 1 { + t.Fatalf("response output length = %d, want 1; payload=%s", got, resp.Payload) + } + if got := gjson.GetBytes(resp.Payload, "output.0.content.0.text").String(); got != "answer" { + t.Fatalf("response text = %q, want answer; payload=%s", got, resp.Payload) + } +} + +func TestXAIInternalXSearchResponseFilterRequiresNativeTool(t *testing.T) { + if xaiRequestHasNativeXSearch([]byte(`{"tools":[{"type":"web_search"}]}`)) { + t.Fatal("web_search must not enable internal X search filtering") + } + if !xaiRequestHasNativeXSearch([]byte(`{"tools":[{"type":"x_search"}]}`)) { + t.Fatal("x_search should enable internal X search filtering") + } + + event := []byte(`{"type":"response.output_item.done","output_index":0,"item":{"id":"ctc_1","type":"custom_tool_call","name":"x_keyword_search"}}`) + if got := newXAIInternalXSearchResponseFilter(false).apply(event); !bytes.Equal(got, event) { + t.Fatalf("disabled filter changed event: %s", got) + } + if got := newXAIInternalXSearchResponseFilter(true).apply(event); got != nil { + t.Fatalf("enabled filter retained internal call: %s", got) + } +} + func TestXAIExecutorComposerSessionIsolation(t *testing.T) { exec := NewXAIExecutor(&config.Config{}) auth := &cliproxyauth.Auth{ diff --git a/internal/runtime/executor/xai_websockets_executor.go b/internal/runtime/executor/xai_websockets_executor.go index 0a55f1c91..87d5a779f 100644 --- a/internal/runtime/executor/xai_websockets_executor.go +++ b/internal/runtime/executor/xai_websockets_executor.go @@ -578,6 +578,7 @@ func (e *XAIWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *cliprox var param any outputItemsByIndex := make(map[int64][]byte) var outputItemsFallback [][]byte + responseFilter := newXAIInternalXSearchResponseFilter(prepared.filterInternalXSearch) recordedTranscript := false for { if ctx != nil && ctx.Err() != nil { @@ -638,6 +639,10 @@ func (e *XAIWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *cliprox for _, payload := range xaiNormalizeReasoningSummaryDataEvents(payload) { payload = restoreXAINamespaceToolCalls(payload, prepared.namespaceTools) + payload = responseFilter.apply(payload) + if len(payload) == 0 { + continue + } eventType := gjson.GetBytes(payload, "type").String() isTerminalEvent := eventType == "response.completed" || eventType == "response.done" || eventType == "error" warmupCompletedPayload := []byte(nil)