From f9fa64f39de7fc2621eea70ab4303decde7c16b5 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Wed, 29 Jul 2026 03:51:44 +0300 Subject: [PATCH 1/7] fix(kimi): normalize nested moonshot tool schemas --- .../runtime/executor/helps/moonshot_schema.go | 379 ++++++++++++++++++ .../executor/helps/moonshot_schema_test.go | 82 ++++ internal/runtime/executor/kimi_executor.go | 2 + .../runtime/executor/kimi_executor_test.go | 65 +++ 4 files changed, 528 insertions(+) create mode 100644 internal/runtime/executor/helps/moonshot_schema.go create mode 100644 internal/runtime/executor/helps/moonshot_schema_test.go diff --git a/internal/runtime/executor/helps/moonshot_schema.go b/internal/runtime/executor/helps/moonshot_schema.go new file mode 100644 index 00000000..efbac473 --- /dev/null +++ b/internal/runtime/executor/helps/moonshot_schema.go @@ -0,0 +1,379 @@ +package helps + +import ( + "bytes" + "encoding/json" +) + +// NormalizeMoonshotToolSchemas adds explicit types to schema nodes that +// Moonshot rejects when they appear in tool parameter unions. +func NormalizeMoonshotToolSchemas(payload []byte) []byte { + var root map[string]json.RawMessage + if len(payload) == 0 || json.Unmarshal(payload, &root) != nil || root == nil { + return payload + } + + var tools []json.RawMessage + if json.Unmarshal(root["tools"], &tools) != nil { + return payload + } + + changed := false + for index, toolRaw := range tools { + tool, ok := moonshotJSONObject(toolRaw) + if !ok { + continue + } + toolChanged := false + + if function, ok := moonshotJSONObject(tool["function"]); ok { + if normalized, changedParameters := normalizeMoonshotToolParameters(function); changedParameters { + function["parameters"] = normalized + updatedFunction, errMarshal := json.Marshal(function) + if errMarshal != nil { + continue + } + tool["function"] = updatedFunction + toolChanged = true + } + } + + if normalized, changedParameters := normalizeMoonshotToolParameters(tool); changedParameters { + tool["parameters"] = normalized + toolChanged = true + } + if !toolChanged { + continue + } + + updatedTool, errMarshal := json.Marshal(tool) + if errMarshal != nil { + continue + } + tools[index] = updatedTool + changed = true + } + if !changed { + return payload + } + + updatedTools, errMarshalTools := json.Marshal(tools) + if errMarshalTools != nil { + return payload + } + root["tools"] = updatedTools + updatedPayload, errMarshalPayload := json.Marshal(root) + if errMarshalPayload != nil { + return payload + } + return updatedPayload +} + +func normalizeMoonshotToolParameters(tool map[string]json.RawMessage) (json.RawMessage, bool) { + rawParameters, ok := tool["parameters"] + if !ok { + return nil, false + } + normalized, changed := normalizeMoonshotSchema(rawParameters, "", false) + return normalized, changed +} + +func normalizeMoonshotSchema(raw json.RawMessage, inheritedType string, requireType bool) (json.RawMessage, bool) { + schema, ok := moonshotJSONObject(raw) + if !ok { + return raw, false + } + + changed := false + currentType, hasType := moonshotSchemaType(schema["type"]) + if !hasType { + if inferredType, okInfer := inferMoonshotSchemaType(schema); okInfer { + schema["type"] = inferredType + currentType, hasType = moonshotSchemaType(inferredType) + changed = true + } else if requireType && inheritedType != "" { + schema["type"] = json.RawMessage(`"` + inheritedType + `"`) + currentType = inheritedType + hasType = true + changed = true + } else if requireType { + schema["type"] = json.RawMessage(`"object"`) + currentType = "object" + hasType = true + changed = true + } + } + + if normalized, childChanged := normalizeMoonshotSchemaMap(schema, "properties", currentType); childChanged { + schema["properties"] = normalized + changed = true + } + if normalized, childChanged := normalizeMoonshotSchemaMap(schema, "patternProperties", currentType); childChanged { + schema["patternProperties"] = normalized + changed = true + } + if normalized, childChanged := normalizeMoonshotSchemaMap(schema, "dependentSchemas", currentType); childChanged { + schema["dependentSchemas"] = normalized + changed = true + } + if normalized, childChanged := normalizeMoonshotSchemaMap(schema, "$defs", currentType); childChanged { + schema["$defs"] = normalized + changed = true + } + if normalized, childChanged := normalizeMoonshotSchemaMap(schema, "definitions", currentType); childChanged { + schema["definitions"] = normalized + changed = true + } + + for _, field := range []string{"additionalProperties", "propertyNames", "contains", "if", "then", "else", "not", "unevaluatedProperties", "unevaluatedItems", "items"} { + rawChild, exists := schema[field] + if !exists || string(rawChild) == "false" || string(rawChild) == "true" { + continue + } + normalized, childChanged := normalizeMoonshotSchema(rawChild, currentType, false) + if childChanged { + schema[field] = normalized + changed = true + } + } + + if rawPrefixItems, exists := schema["prefixItems"]; exists { + var items []json.RawMessage + if json.Unmarshal(rawPrefixItems, &items) == nil { + for index, item := range items { + normalized, itemChanged := normalizeMoonshotSchema(item, "", false) + if itemChanged { + items[index] = normalized + changed = true + } + } + if changed { + if normalizedItems, errMarshal := json.Marshal(items); errMarshal == nil { + schema["prefixItems"] = normalizedItems + } + } + } + } + + for _, unionName := range []string{"anyOf", "oneOf", "allOf"} { + rawUnion, exists := schema[unionName] + if !exists { + continue + } + var branches []json.RawMessage + if json.Unmarshal(rawUnion, &branches) != nil { + continue + } + unionChanged := false + for index, branch := range branches { + normalized, branchChanged := normalizeMoonshotSchema(branch, currentType, true) + if branchChanged { + branches[index] = normalized + unionChanged = true + } + } + if unionChanged { + if normalizedUnion, errMarshal := json.Marshal(branches); errMarshal == nil { + schema[unionName] = normalizedUnion + changed = true + } + } + } + + if !changed { + return raw, false + } + normalized, errMarshal := json.Marshal(schema) + if errMarshal != nil { + return raw, false + } + return normalized, true +} + +func normalizeMoonshotSchemaMap(schema map[string]json.RawMessage, field, inheritedType string) (json.RawMessage, bool) { + rawMap, exists := schema[field] + if !exists { + return nil, false + } + children, ok := moonshotJSONObject(rawMap) + if !ok { + return rawMap, false + } + changed := false + for name, child := range children { + normalized, childChanged := normalizeMoonshotSchema(child, inheritedType, false) + if childChanged { + children[name] = normalized + changed = true + } + } + if !changed { + return rawMap, false + } + normalized, errMarshal := json.Marshal(children) + if errMarshal != nil { + return rawMap, false + } + return normalized, true +} + +func inferMoonshotSchemaType(schema map[string]json.RawMessage) (json.RawMessage, bool) { + if rawEnum, exists := schema["enum"]; exists { + if inferred, ok := inferMoonshotEnumType(rawEnum); ok { + return inferred, true + } + } + if rawConst, exists := schema["const"]; exists { + if inferred, ok := moonshotJSONValueType(rawConst); ok { + return inferred, true + } + } + for _, field := range []string{"properties", "patternProperties", "required", "additionalProperties", "propertyNames", "dependentRequired", "dependentSchemas", "minProperties", "maxProperties"} { + if _, exists := schema[field]; exists { + return json.RawMessage(`"object"`), true + } + } + for _, field := range []string{"items", "prefixItems", "contains", "minItems", "maxItems", "uniqueItems"} { + if _, exists := schema[field]; exists { + return json.RawMessage(`"array"`), true + } + } + for _, field := range []string{"minLength", "maxLength", "pattern", "format", "contentEncoding", "contentMediaType"} { + if _, exists := schema[field]; exists { + return json.RawMessage(`"string"`), true + } + } + for _, field := range []string{"minimum", "maximum", "exclusiveMinimum", "exclusiveMaximum", "multipleOf"} { + if _, exists := schema[field]; exists { + return json.RawMessage(`"number"`), true + } + } + for _, unionName := range []string{"anyOf", "oneOf", "allOf"} { + rawUnion, exists := schema[unionName] + if !exists { + continue + } + var branches []json.RawMessage + if json.Unmarshal(rawUnion, &branches) != nil || len(branches) == 0 { + continue + } + types := make([]string, 0, len(branches)) + seen := make(map[string]struct{}, len(branches)) + for _, branch := range branches { + branchSchema, ok := moonshotJSONObject(branch) + if !ok { + types = nil + break + } + branchType, hasType := moonshotSchemaType(branchSchema["type"]) + if !hasType { + inferredType, okInfer := inferMoonshotSchemaType(branchSchema) + if !okInfer || json.Unmarshal(inferredType, &branchType) != nil { + types = nil + break + } + } + if branchType == "" { + types = nil + break + } + if _, exists := seen[branchType]; exists { + continue + } + seen[branchType] = struct{}{} + types = append(types, branchType) + } + if len(types) == 1 { + return json.RawMessage(`"` + types[0] + `"`), true + } + if len(types) > 1 { + inferredType, errMarshal := json.Marshal(types) + if errMarshal == nil { + return inferredType, true + } + } + } + return nil, false +} + +func inferMoonshotEnumType(raw json.RawMessage) (json.RawMessage, bool) { + var values []json.RawMessage + if json.Unmarshal(raw, &values) != nil || len(values) == 0 { + return nil, false + } + types := make([]string, 0, len(values)) + seen := make(map[string]struct{}, len(values)) + for _, value := range values { + inferred, ok := moonshotJSONValueType(value) + if !ok { + return nil, false + } + var typeName string + if json.Unmarshal(inferred, &typeName) != nil { + return nil, false + } + if _, exists := seen[typeName]; exists { + continue + } + seen[typeName] = struct{}{} + types = append(types, typeName) + } + if len(types) == 1 { + return json.RawMessage(`"` + types[0] + `"`), true + } + typeArray, errMarshal := json.Marshal(types) + if errMarshal != nil { + return nil, false + } + return typeArray, true +} + +func moonshotJSONValueType(raw json.RawMessage) (json.RawMessage, bool) { + var value any + decoder := json.NewDecoder(bytes.NewReader(raw)) + decoder.UseNumber() + if decoder.Decode(&value) != nil { + return nil, false + } + var typeName string + switch value.(type) { + case nil: + typeName = "null" + case string: + typeName = "string" + case bool: + typeName = "boolean" + case json.Number: + typeName = "number" + case []any: + typeName = "array" + case map[string]any: + typeName = "object" + default: + return nil, false + } + return json.RawMessage(`"` + typeName + `"`), true +} + +func moonshotSchemaType(raw json.RawMessage) (string, bool) { + var typeName string + if json.Unmarshal(raw, &typeName) == nil && typeName != "" { + return typeName, true + } + var typeNames []string + if json.Unmarshal(raw, &typeNames) == nil && len(typeNames) > 0 { + if len(typeNames) == 1 { + return typeNames[0], true + } + return "", true + } + return "", false +} + +func moonshotJSONObject(raw json.RawMessage) (map[string]json.RawMessage, bool) { + var object map[string]json.RawMessage + if len(raw) == 0 || json.Unmarshal(raw, &object) != nil || object == nil { + return nil, false + } + return object, true +} diff --git a/internal/runtime/executor/helps/moonshot_schema_test.go b/internal/runtime/executor/helps/moonshot_schema_test.go new file mode 100644 index 00000000..1db58b09 --- /dev/null +++ b/internal/runtime/executor/helps/moonshot_schema_test.go @@ -0,0 +1,82 @@ +package helps + +import ( + "fmt" + "testing" + + "github.com/tidwall/gjson" +) + +func TestNormalizeMoonshotToolSchemasAddsNestedUnionTypes(t *testing.T) { + payload := []byte(`{ + "tools":[{ + "type":"function", + "function":{ + "name":"lookup", + "parameters":{ + "type":"object", + "properties":{ + "details":{ + "type":"object", + "properties":{ + "icon":{ + "type":"object", + "properties":{ + "color":{"anyOf":[{"enum":["red","blue"]},{"const":"auto"}]} + } + } + } + } + } + } + } + }] + }`) + + out := NormalizeMoonshotToolSchemas(payload) + if got := gjson.GetBytes(out, "tools.0.function.parameters.properties.details.properties.icon.properties.color.type").String(); got != "string" { + t.Fatalf("color.type = %q, want string; payload=%s", got, out) + } + for index := range []int{0, 1} { + got := gjson.GetBytes(out, fmt.Sprintf("tools.0.function.parameters.properties.details.properties.icon.properties.color.anyOf.%d.type", index)).String() + if got != "string" { + t.Fatalf("color.anyOf.%d.type = %q, want string; payload=%s", index, got, out) + } + } +} + +func TestNormalizeMoonshotToolSchemasInfersObjectBranchesAndPreservesExistingTypes(t *testing.T) { + payload := []byte(`{"tools":[ + {"type":"function","function":{"parameters":{ + "type":"object", + "anyOf":[{"required":["query"]},{"properties":{"id":{"type":"integer"}}}], + "properties":{"mode":{"anyOf":[{"type":"string","enum":["a"]},{"enum":["b"]}]}} + }}}, + {"type":"function","function":{"parameters":{"type":"object","properties":{"value":{"anyOf":[{"type":"number"},{"type":"null"}]}}}}} + ]}`) + + out := NormalizeMoonshotToolSchemas(payload) + for index := range []int{0, 1} { + if got := gjson.GetBytes(out, fmt.Sprintf("tools.0.function.parameters.anyOf.%d.type", index)).String(); got != "object" { + t.Fatalf("root anyOf.%d.type = %q, want object; payload=%s", index, got, out) + } + } + if got := gjson.GetBytes(out, "tools.0.function.parameters.properties.mode.anyOf.1.type").String(); got != "string" { + t.Fatalf("mode.anyOf.1.type = %q, want string; payload=%s", got, out) + } + if got := gjson.GetBytes(out, "tools.1.function.parameters.properties.value.anyOf.0.type").String(); got != "number" { + t.Fatalf("existing value.anyOf.0.type = %q, want number; payload=%s", got, out) + } +} + +func TestNormalizeMoonshotToolSchemasLeavesUnrelatedPayloadsUnchanged(t *testing.T) { + for _, payload := range [][]byte{ + []byte(`{"messages":[]}`), + []byte(`{"tools":[]}`), + []byte(`not-json`), + } { + if got := NormalizeMoonshotToolSchemas(payload); string(got) != string(payload) { + t.Fatalf("payload changed: got %q, want %q", got, payload) + } + } +} diff --git a/internal/runtime/executor/kimi_executor.go b/internal/runtime/executor/kimi_executor.go index ec270705..a707af91 100644 --- a/internal/runtime/executor/kimi_executor.go +++ b/internal/runtime/executor/kimi_executor.go @@ -125,6 +125,7 @@ func (e *KimiExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req if err != nil { return resp, err } + body = helps.NormalizeMoonshotToolSchemas(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) url := kimiauth.KimiAPIBaseURL + "/v1/chat/completions" @@ -238,6 +239,7 @@ func (e *KimiExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Aut if err != nil { return nil, err } + body = helps.NormalizeMoonshotToolSchemas(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) url := kimiauth.KimiAPIBaseURL + "/v1/chat/completions" diff --git a/internal/runtime/executor/kimi_executor_test.go b/internal/runtime/executor/kimi_executor_test.go index c9e79d92..ffe5000d 100644 --- a/internal/runtime/executor/kimi_executor_test.go +++ b/internal/runtime/executor/kimi_executor_test.go @@ -241,6 +241,71 @@ func TestKimiExecutorClaudeStreamForwardsAnthropicBetaAndLogsUpstream(t *testing } } +func TestKimiExecutorNormalizesMoonshotToolSchemasForBothExecutionModes(t *testing.T) { + var upstreamBodies [][]byte + ctx := context.WithValue(context.Background(), "cliproxy.roundtripper", kimiRoundTripperFunc(func(req *http.Request) (*http.Response, error) { + body, errRead := io.ReadAll(req.Body) + if errRead != nil { + return nil, errRead + } + upstreamBodies = append(upstreamBodies, body) + if gjson.GetBytes(body, "stream").Bool() { + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader( + "data: {\"id\":\"chatcmpl-test\",\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n" + + "data: [DONE]\n\n", + )), + }, nil + } + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader( + `{"id":"chatcmpl-test","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}`, + )), + }, nil + })) + + executor := NewKimiExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{ + Attributes: map[string]string{}, + Metadata: map[string]any{"access_token": "test-token"}, + } + toolSchema := `{"type":"object","properties":{"details":{"type":"object","properties":{"icon":{"type":"object","properties":{"color":{"anyOf":[{"enum":["red","blue"]},{"const":"auto"}]}}}}}}}` + requestPayload := `{"model":"kimi-k3","messages":[{"role":"user","content":"lookup"}],"tools":[{"type":"function","function":{"name":"lookup","parameters":` + toolSchema + `}}]}` + if _, err := executor.Execute(ctx, auth, cliproxyexecutor.Request{Model: "kimi-k3", Payload: []byte(requestPayload)}, cliproxyexecutor.Options{SourceFormat: sdktranslator.FormatOpenAI}); err != nil { + t.Fatalf("Execute() error = %v", err) + } + + streamPayload := strings.Replace(requestPayload, `{"model":"kimi-k3"`, `{"model":"kimi-k3","stream":true`, 1) + result, err := executor.ExecuteStream(ctx, auth, cliproxyexecutor.Request{Model: "kimi-k3", Payload: []byte(streamPayload)}, cliproxyexecutor.Options{Stream: true, SourceFormat: sdktranslator.FormatOpenAI}) + if err != nil { + t.Fatalf("ExecuteStream() error = %v", err) + } + for chunk := range result.Chunks { + if chunk.Err != nil { + t.Fatalf("stream chunk error = %v", chunk.Err) + } + } + + if len(upstreamBodies) != 2 { + t.Fatalf("upstream request count = %d, want 2", len(upstreamBodies)) + } + for index, body := range upstreamBodies { + for _, path := range []string{ + "tools.0.function.parameters.properties.details.properties.icon.properties.color.type", + "tools.0.function.parameters.properties.details.properties.icon.properties.color.anyOf.0.type", + "tools.0.function.parameters.properties.details.properties.icon.properties.color.anyOf.1.type", + } { + if got := gjson.GetBytes(body, path).String(); got != "string" { + t.Fatalf("request %d %s = %q, want string; body=%s", index, path, got, body) + } + } + } +} + type kimiRoundTripperFunc func(*http.Request) (*http.Response, error) func (f kimiRoundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { -- 2.51.2 From ff108acac74f5866d330848a373a4dcdcc487fdb Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Wed, 29 Jul 2026 04:09:51 +0300 Subject: [PATCH 2/7] fix(codex): strip namespaces from spark tool history --- .../executor/codex_executor_execute.go | 1 + .../runtime/executor/codex_executor_stream.go | 1 + .../runtime/executor/helps/codex_namespace.go | 29 +++++++++++++++++++ .../executor/helps/codex_namespace_test.go | 21 ++++++++++++++ 4 files changed, 52 insertions(+) create mode 100644 internal/runtime/executor/helps/codex_namespace.go create mode 100644 internal/runtime/executor/helps/codex_namespace_test.go diff --git a/internal/runtime/executor/codex_executor_execute.go b/internal/runtime/executor/codex_executor_execute.go index 87279814..ba88d94e 100644 --- a/internal/runtime/executor/codex_executor_execute.go +++ b/internal/runtime/executor/codex_executor_execute.go @@ -61,6 +61,7 @@ func (e *CodexExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, re body, _ = sjson.DeleteBytes(body, "safety_identifier") body, _ = sjson.DeleteBytes(body, "stream_options") body = normalizeCodexInstructions(body) + body = helps.SanitizeCodexResponseInputNamespaces(body) if e.cfg == nil || e.cfg.DisableImageGeneration == config.DisableImageGenerationOff { body = ensureImageGenerationTool(body, baseModel, auth, opts.Headers) } diff --git a/internal/runtime/executor/codex_executor_stream.go b/internal/runtime/executor/codex_executor_stream.go index 8d5c8993..c56580db 100644 --- a/internal/runtime/executor/codex_executor_stream.go +++ b/internal/runtime/executor/codex_executor_stream.go @@ -61,6 +61,7 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au body, _ = sjson.DeleteBytes(body, "stream_options") body = helps.SetStringIfDifferent(body, "model", baseModel) body = normalizeCodexInstructions(body) + body = helps.SanitizeCodexResponseInputNamespaces(body) if e.cfg == nil || e.cfg.DisableImageGeneration == config.DisableImageGenerationOff { body = ensureImageGenerationTool(body, baseModel, auth, opts.Headers) } diff --git a/internal/runtime/executor/helps/codex_namespace.go b/internal/runtime/executor/helps/codex_namespace.go new file mode 100644 index 00000000..6617a447 --- /dev/null +++ b/internal/runtime/executor/helps/codex_namespace.go @@ -0,0 +1,29 @@ +package helps + +import ( + "fmt" + + "github.com/tidwall/gjson" + "github.com/tidwall/sjson" +) + +// SanitizeCodexResponseInputNamespaces removes namespaces from tool-call +// history items that the Codex Responses endpoint rejects. +func SanitizeCodexResponseInputNamespaces(body []byte) []byte { + items := gjson.GetBytes(body, "input") + if !items.IsArray() { + return body + } + + for index, item := range items.Array() { + itemType := item.Get("type").String() + if itemType != "function_call" && itemType != "custom_tool_call" { + continue + } + updated, errDelete := sjson.DeleteBytes(body, fmt.Sprintf("input.%d.namespace", index)) + if errDelete == nil { + body = updated + } + } + return body +} diff --git a/internal/runtime/executor/helps/codex_namespace_test.go b/internal/runtime/executor/helps/codex_namespace_test.go new file mode 100644 index 00000000..a9f7d3a6 --- /dev/null +++ b/internal/runtime/executor/helps/codex_namespace_test.go @@ -0,0 +1,21 @@ +package helps + +import ( + "testing" + + "github.com/tidwall/gjson" +) + +func TestSanitizeCodexResponseInputNamespaces(t *testing.T) { + body := []byte(`{"input":[{"type":"function_call","namespace":"exec","name":"run"},{"type":"custom_tool_call","namespace":"tools","name":"lookup"},{"type":"message","namespace":"keep"}]}`) + + got := SanitizeCodexResponseInputNamespaces(body) + for _, path := range []string{"input.0.namespace", "input.1.namespace"} { + if gjson.GetBytes(got, path).Exists() { + t.Fatalf("%s was not removed: %s", path, got) + } + } + if namespace := gjson.GetBytes(got, "input.2.namespace").String(); namespace != "keep" { + t.Fatalf("message namespace = %q, want keep", namespace) + } +} -- 2.51.2 From 4dedd7ec0818673e22ef21ac651562f42a80df03 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Wed, 29 Jul 2026 05:03:17 +0300 Subject: [PATCH 3/7] fix(kimi): keep moonshot union types on branches --- .../runtime/executor/helps/moonshot_schema.go | 59 ++++--------------- .../executor/helps/moonshot_schema_test.go | 21 ++++++- .../runtime/executor/kimi_executor_test.go | 4 +- 3 files changed, 34 insertions(+), 50 deletions(-) diff --git a/internal/runtime/executor/helps/moonshot_schema.go b/internal/runtime/executor/helps/moonshot_schema.go index efbac473..57d7c6c8 100644 --- a/internal/runtime/executor/helps/moonshot_schema.go +++ b/internal/runtime/executor/helps/moonshot_schema.go @@ -85,8 +85,20 @@ func normalizeMoonshotSchema(raw json.RawMessage, inheritedType string, requireT } changed := false + _, hasAnyOf := schema["anyOf"] + _, hasOneOf := schema["oneOf"] currentType, hasType := moonshotSchemaType(schema["type"]) - if !hasType { + if hasAnyOf || hasOneOf { + // Moonshot requires union branch types and rejects a parent type next to + // anyOf/oneOf. The branches below carry the useful type information. + if _, exists := schema["type"]; exists { + delete(schema, "type") + changed = true + } + currentType = "" + hasType = false + } + if !hasType && !hasAnyOf && !hasOneOf { if inferredType, okInfer := inferMoonshotSchemaType(schema); okInfer { schema["type"] = inferredType currentType, hasType = moonshotSchemaType(inferredType) @@ -248,51 +260,6 @@ func inferMoonshotSchemaType(schema map[string]json.RawMessage) (json.RawMessage return json.RawMessage(`"number"`), true } } - for _, unionName := range []string{"anyOf", "oneOf", "allOf"} { - rawUnion, exists := schema[unionName] - if !exists { - continue - } - var branches []json.RawMessage - if json.Unmarshal(rawUnion, &branches) != nil || len(branches) == 0 { - continue - } - types := make([]string, 0, len(branches)) - seen := make(map[string]struct{}, len(branches)) - for _, branch := range branches { - branchSchema, ok := moonshotJSONObject(branch) - if !ok { - types = nil - break - } - branchType, hasType := moonshotSchemaType(branchSchema["type"]) - if !hasType { - inferredType, okInfer := inferMoonshotSchemaType(branchSchema) - if !okInfer || json.Unmarshal(inferredType, &branchType) != nil { - types = nil - break - } - } - if branchType == "" { - types = nil - break - } - if _, exists := seen[branchType]; exists { - continue - } - seen[branchType] = struct{}{} - types = append(types, branchType) - } - if len(types) == 1 { - return json.RawMessage(`"` + types[0] + `"`), true - } - if len(types) > 1 { - inferredType, errMarshal := json.Marshal(types) - if errMarshal == nil { - return inferredType, true - } - } - } return nil, false } diff --git a/internal/runtime/executor/helps/moonshot_schema_test.go b/internal/runtime/executor/helps/moonshot_schema_test.go index 1db58b09..9e61a6af 100644 --- a/internal/runtime/executor/helps/moonshot_schema_test.go +++ b/internal/runtime/executor/helps/moonshot_schema_test.go @@ -22,7 +22,8 @@ func TestNormalizeMoonshotToolSchemasAddsNestedUnionTypes(t *testing.T) { "icon":{ "type":"object", "properties":{ - "color":{"anyOf":[{"enum":["red","blue"]},{"const":"auto"}]} + "color":{"anyOf":[{"enum":["red","blue"]},{"const":"auto"}]}, + "surface":{"type":"string","anyOf":[{"enum":["matte"]},{"const":"glossy"}]} } } } @@ -34,8 +35,13 @@ func TestNormalizeMoonshotToolSchemasAddsNestedUnionTypes(t *testing.T) { }`) out := NormalizeMoonshotToolSchemas(payload) - if got := gjson.GetBytes(out, "tools.0.function.parameters.properties.details.properties.icon.properties.color.type").String(); got != "string" { - t.Fatalf("color.type = %q, want string; payload=%s", got, out) + for _, path := range []string{ + "tools.0.function.parameters.properties.details.properties.icon.properties.color.type", + "tools.0.function.parameters.properties.details.properties.icon.properties.surface.type", + } { + if gjson.GetBytes(out, path).Exists() { + t.Fatalf("union parent type should be absent at %s; payload=%s", path, out) + } } for index := range []int{0, 1} { got := gjson.GetBytes(out, fmt.Sprintf("tools.0.function.parameters.properties.details.properties.icon.properties.color.anyOf.%d.type", index)).String() @@ -43,6 +49,12 @@ func TestNormalizeMoonshotToolSchemasAddsNestedUnionTypes(t *testing.T) { t.Fatalf("color.anyOf.%d.type = %q, want string; payload=%s", index, got, out) } } + for index := range []int{0, 1} { + got := gjson.GetBytes(out, fmt.Sprintf("tools.0.function.parameters.properties.details.properties.icon.properties.surface.anyOf.%d.type", index)).String() + if got != "string" { + t.Fatalf("surface.anyOf.%d.type = %q, want string; payload=%s", index, got, out) + } + } } func TestNormalizeMoonshotToolSchemasInfersObjectBranchesAndPreservesExistingTypes(t *testing.T) { @@ -56,6 +68,9 @@ func TestNormalizeMoonshotToolSchemasInfersObjectBranchesAndPreservesExistingTyp ]}`) out := NormalizeMoonshotToolSchemas(payload) + if gjson.GetBytes(out, "tools.0.function.parameters.type").Exists() { + t.Fatalf("root union parent type should be absent; payload=%s", out) + } for index := range []int{0, 1} { if got := gjson.GetBytes(out, fmt.Sprintf("tools.0.function.parameters.anyOf.%d.type", index)).String(); got != "object" { t.Fatalf("root anyOf.%d.type = %q, want object; payload=%s", index, got, out) diff --git a/internal/runtime/executor/kimi_executor_test.go b/internal/runtime/executor/kimi_executor_test.go index ffe5000d..3179f468 100644 --- a/internal/runtime/executor/kimi_executor_test.go +++ b/internal/runtime/executor/kimi_executor_test.go @@ -294,8 +294,10 @@ func TestKimiExecutorNormalizesMoonshotToolSchemasForBothExecutionModes(t *testi t.Fatalf("upstream request count = %d, want 2", len(upstreamBodies)) } for index, body := range upstreamBodies { + if gjson.GetBytes(body, "tools.0.function.parameters.properties.details.properties.icon.properties.color.type").Exists() { + t.Fatalf("request %d has a forbidden union parent type; body=%s", index, body) + } for _, path := range []string{ - "tools.0.function.parameters.properties.details.properties.icon.properties.color.type", "tools.0.function.parameters.properties.details.properties.icon.properties.color.anyOf.0.type", "tools.0.function.parameters.properties.details.properties.icon.properties.color.anyOf.1.type", } { -- 2.51.2 From e035861516bf662dbeca1936a3adcd77aa940667 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Wed, 29 Jul 2026 05:43:22 +0300 Subject: [PATCH 4/7] fix(kimi): preserve root tool parameter object type --- internal/runtime/executor/helps/moonshot_schema.go | 14 ++++++++++++++ .../runtime/executor/helps/moonshot_schema_test.go | 4 ++-- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/internal/runtime/executor/helps/moonshot_schema.go b/internal/runtime/executor/helps/moonshot_schema.go index 57d7c6c8..21e89313 100644 --- a/internal/runtime/executor/helps/moonshot_schema.go +++ b/internal/runtime/executor/helps/moonshot_schema.go @@ -75,6 +75,20 @@ func normalizeMoonshotToolParameters(tool map[string]json.RawMessage) (json.RawM return nil, false } normalized, changed := normalizeMoonshotSchema(rawParameters, "", false) + parameters, ok := moonshotJSONObject(normalized) + if !ok { + return normalized, changed + } + // Moonshot requires every function's top-level parameters schema to be an + // object, even when that schema also contains a union. Nested union parents + // must omit type; this boundary is the exception. + parameterType, hasType := moonshotSchemaType(parameters["type"]) + if !hasType || parameterType != "object" { + parameters["type"] = json.RawMessage(`"object"`) + if updated, errMarshal := json.Marshal(parameters); errMarshal == nil { + return updated, true + } + } return normalized, changed } diff --git a/internal/runtime/executor/helps/moonshot_schema_test.go b/internal/runtime/executor/helps/moonshot_schema_test.go index 9e61a6af..8f12eb49 100644 --- a/internal/runtime/executor/helps/moonshot_schema_test.go +++ b/internal/runtime/executor/helps/moonshot_schema_test.go @@ -68,8 +68,8 @@ func TestNormalizeMoonshotToolSchemasInfersObjectBranchesAndPreservesExistingTyp ]}`) out := NormalizeMoonshotToolSchemas(payload) - if gjson.GetBytes(out, "tools.0.function.parameters.type").Exists() { - t.Fatalf("root union parent type should be absent; payload=%s", out) + if got := gjson.GetBytes(out, "tools.0.function.parameters.type").String(); got != "object" { + t.Fatalf("root union parameters.type = %q, want object; payload=%s", got, out) } for index := range []int{0, 1} { if got := gjson.GetBytes(out, fmt.Sprintf("tools.0.function.parameters.anyOf.%d.type", index)).String(); got != "object" { -- 2.51.2 From 33ab7734b1d0c73d2cbecb0b97263c22c7dbe0fe Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Wed, 29 Jul 2026 15:21:16 +0300 Subject: [PATCH 5/7] fix(kimi): preserve images from tool outputs --- internal/runtime/executor/kimi_executor.go | 162 ++++++++++++++++++ .../runtime/executor/kimi_executor_test.go | 116 +++++++++++++ 2 files changed, 278 insertions(+) diff --git a/internal/runtime/executor/kimi_executor.go b/internal/runtime/executor/kimi_executor.go index a707af91..b05d00fa 100644 --- a/internal/runtime/executor/kimi_executor.go +++ b/internal/runtime/executor/kimi_executor.go @@ -125,6 +125,12 @@ func (e *KimiExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req if err != nil { return resp, err } + if from == sdktranslator.FormatOpenAIResponse { + body, err = moveKimiToolOutputImagesToUserMessages(body) + if err != nil { + return resp, err + } + } body = helps.NormalizeMoonshotToolSchemas(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) @@ -239,6 +245,12 @@ func (e *KimiExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Aut if err != nil { return nil, err } + if from == sdktranslator.FormatOpenAIResponse { + body, err = moveKimiToolOutputImagesToUserMessages(body) + if err != nil { + return nil, err + } + } body = helps.NormalizeMoonshotToolSchemas(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) @@ -342,6 +354,156 @@ func (e *KimiExecutor) CountTokens(ctx context.Context, auth *cliproxyauth.Auth, return e.ClaudeExecutor.countTokensUpstream(ctx, auth, req, opts) } +type kimiToolOutputImage struct { + toolCallID string + url string + detail string +} + +// moveKimiToolOutputImagesToUserMessages keeps tool results as strings while +// forwarding Responses API image parts through Kimi's native user image input. +func moveKimiToolOutputImagesToUserMessages(body []byte) ([]byte, error) { + if len(body) == 0 || !gjson.ValidBytes(body) { + return body, nil + } + + messages := util.GetGJSONBytesNoCopy(body, "messages") + if !messages.Exists() || !messages.IsArray() { + return body, nil + } + + messageItems := make([]string, 0, len(messages.Array())) + pendingImages := make([]kimiToolOutputImage, 0) + movedImages := 0 + + flushImages := func() error { + if len(pendingImages) == 0 { + return nil + } + + contentItems := make([]string, 0, len(pendingImages)*2) + for _, image := range pendingImages { + label := "Image output from the preceding tool call:" + if image.toolCallID != "" { + label = fmt.Sprintf("Image output from tool call %s:", image.toolCallID) + } + textPart := []byte(`{"type":"text","text":""}`) + textPart, _ = sjson.SetBytes(textPart, "text", label) + contentItems = append(contentItems, string(textPart)) + + imagePart := []byte(`{"type":"image_url","image_url":{"url":""}}`) + imagePart, _ = sjson.SetBytes(imagePart, "image_url.url", image.url) + if image.detail != "" { + imagePart, _ = sjson.SetBytes(imagePart, "image_url.detail", image.detail) + } + contentItems = append(contentItems, string(imagePart)) + } + + userMessage := []byte(`{"role":"user","content":[]}`) + userMessage, errSet := sjson.SetRawBytes(userMessage, "content", helps.JoinRawJSONStrings(contentItems)) + if errSet != nil { + return fmt.Errorf("kimi executor: failed to attach tool output images: %w", errSet) + } + messageItems = append(messageItems, string(userMessage)) + pendingImages = pendingImages[:0] + return nil + } + + for _, message := range messages.Array() { + if strings.TrimSpace(message.Get("role").String()) != "tool" { + if errFlush := flushImages(); errFlush != nil { + return body, errFlush + } + messageItems = append(messageItems, message.Raw) + continue + } + + text, images, ok := splitKimiStructuredToolOutput(message.Get("content")) + if !ok { + messageItems = append(messageItems, message.Raw) + continue + } + + toolCallID := strings.TrimSpace(message.Get("tool_call_id").String()) + for imageIndex := range images { + images[imageIndex].toolCallID = toolCallID + } + pendingImages = append(pendingImages, images...) + movedImages += len(images) + + updatedMessage, errSet := sjson.SetBytes([]byte(message.Raw), "content", text) + if errSet != nil { + return body, fmt.Errorf("kimi executor: failed to replace structured tool image output: %w", errSet) + } + messageItems = append(messageItems, string(updatedMessage)) + } + if errFlush := flushImages(); errFlush != nil { + return body, errFlush + } + + if movedImages == 0 { + return body, nil + } + out, errSet := sjson.SetRawBytes(body, "messages", helps.JoinRawJSONStrings(messageItems)) + if errSet != nil { + return body, fmt.Errorf("kimi executor: failed to rebuild messages with tool output images: %w", errSet) + } + log.WithField("moved_tool_output_images", movedImages).Debug("kimi executor: moved tool output images to native user content") + return out, nil +} + +func splitKimiStructuredToolOutput(content gjson.Result) (string, []kimiToolOutputImage, bool) { + if content.Type != gjson.String { + return "", nil, false + } + + structured := gjson.Parse(content.String()) + var parts []gjson.Result + switch { + case structured.IsArray(): + parts = structured.Array() + case structured.IsObject(): + parts = []gjson.Result{structured} + default: + return "", nil, false + } + + textParts := make([]string, 0, len(parts)) + images := make([]kimiToolOutputImage, 0) + for _, part := range parts { + switch strings.TrimSpace(part.Get("type").String()) { + case "input_text", "output_text", "text": + if text := part.Get("text").String(); text != "" { + textParts = append(textParts, text) + } + case "input_image", "image_url", "image": + imageURL := part.Get("image_url") + url := imageURL.String() + if imageURL.IsObject() { + url = imageURL.Get("url").String() + } + if url == "" { + url = part.Get("url").String() + } + if url == "" { + textParts = append(textParts, part.Raw) + continue + } + images = append(images, kimiToolOutputImage{ + url: url, + detail: part.Get("detail").String(), + }) + textParts = append(textParts, "[image output attached in the following user message]") + default: + textParts = append(textParts, part.Raw) + } + } + if len(images) == 0 { + return "", nil, false + } + return strings.Join(textParts, "\n"), images, true +} + func normalizeKimiToolMessageLinks(body []byte) ([]byte, error) { if len(body) == 0 || !gjson.ValidBytes(body) { return body, nil diff --git a/internal/runtime/executor/kimi_executor_test.go b/internal/runtime/executor/kimi_executor_test.go index 3179f468..d2dd7e97 100644 --- a/internal/runtime/executor/kimi_executor_test.go +++ b/internal/runtime/executor/kimi_executor_test.go @@ -308,6 +308,122 @@ func TestKimiExecutorNormalizesMoonshotToolSchemasForBothExecutionModes(t *testi } } +func TestKimiExecutorMovesResponsesToolImagesToNativeUserContentForBothExecutionModes(t *testing.T) { + var upstreamBodies [][]byte + ctx := context.WithValue(context.Background(), "cliproxy.roundtripper", kimiRoundTripperFunc(func(req *http.Request) (*http.Response, error) { + body, errRead := io.ReadAll(req.Body) + if errRead != nil { + return nil, errRead + } + upstreamBodies = append(upstreamBodies, body) + if gjson.GetBytes(body, "stream").Bool() { + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader( + "data: {\"id\":\"chatcmpl-test\",\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n" + + "data: [DONE]\n\n", + )), + }, nil + } + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader( + `{"id":"chatcmpl-test","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}`, + )), + }, nil + })) + + executor := NewKimiExecutor(&config.Config{}) + auth := &cliproxyauth.Auth{ + Attributes: map[string]string{}, + Metadata: map[string]any{"access_token": "test-token"}, + } + const imageURL = "data:image/png;base64,aGVsbG8=" + payload := []byte(`{ + "model":"kimi-k3", + "input":[ + {"type":"function_call","call_id":"call_image","name":"view_image","arguments":"{\"path\":\"/tmp/screenshot.png\"}"}, + {"type":"function_call","call_id":"call_text","name":"read_file","arguments":"{\"path\":\"/tmp/note.txt\"}"}, + {"type":"function_call_output","call_id":"call_image","output":[ + {"type":"input_text","text":"screenshot captured"}, + {"type":"input_image","image_url":"` + imageURL + `","detail":"original"} + ]}, + {"type":"function_call_output","call_id":"call_text","output":"plain result"}, + {"type":"message","role":"user","content":[{"type":"input_text","text":"next"}]} + ] + }`) + + if _, err := executor.Execute(ctx, auth, cliproxyexecutor.Request{Model: "kimi-k3", Payload: payload}, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FormatOpenAIResponse, + OriginalRequest: payload, + }); err != nil { + t.Fatalf("Execute() error = %v", err) + } + + streamPayload := []byte(strings.Replace(string(payload), `"model":"kimi-k3"`, `"model":"kimi-k3","stream":true`, 1)) + result, err := executor.ExecuteStream(ctx, auth, cliproxyexecutor.Request{Model: "kimi-k3", Payload: streamPayload}, cliproxyexecutor.Options{ + Stream: true, + SourceFormat: sdktranslator.FormatOpenAIResponse, + OriginalRequest: streamPayload, + }) + if err != nil { + t.Fatalf("ExecuteStream() error = %v", err) + } + for chunk := range result.Chunks { + if chunk.Err != nil { + t.Fatalf("stream chunk error = %v", chunk.Err) + } + } + + if len(upstreamBodies) != 2 { + t.Fatalf("upstream request count = %d, want 2", len(upstreamBodies)) + } + for requestIndex, body := range upstreamBodies { + messages := gjson.GetBytes(body, "messages").Array() + if len(messages) != 5 { + t.Fatalf("request %d messages length = %d, want 5; body=%s", requestIndex, len(messages), body) + } + if got := messages[0].Get("role").String(); got != "assistant" { + t.Fatalf("request %d messages.0.role = %q, want assistant", requestIndex, got) + } + if got := len(messages[0].Get("tool_calls").Array()); got != 2 { + t.Fatalf("request %d messages.0.tool_calls length = %d, want 2", requestIndex, got) + } + for messageIndex, callID := range []string{"call_image", "call_text"} { + message := messages[messageIndex+1] + if got := message.Get("role").String(); got != "tool" { + t.Fatalf("request %d messages.%d.role = %q, want tool", requestIndex, messageIndex+1, got) + } + if got := message.Get("tool_call_id").String(); got != callID { + t.Fatalf("request %d messages.%d.tool_call_id = %q, want %q", requestIndex, messageIndex+1, got, callID) + } + } + if content := messages[1].Get("content").String(); strings.Contains(content, imageURL) || !strings.Contains(content, "screenshot captured") || !strings.Contains(content, "attached") { + t.Fatalf("request %d image tool content = %q, want text and marker without data URL", requestIndex, content) + } + if got := messages[2].Get("content").String(); got != "plain result" { + t.Fatalf("request %d text tool content = %q, want plain result", requestIndex, got) + } + if got := messages[3].Get("role").String(); got != "user" { + t.Fatalf("request %d messages.3.role = %q, want user", requestIndex, got) + } + if got := messages[3].Get("content.1.type").String(); got != "image_url" { + t.Fatalf("request %d native image type = %q, want image_url", requestIndex, got) + } + if got := messages[3].Get("content.1.image_url.url").String(); got != imageURL { + t.Fatalf("request %d native image URL = %q, want %q", requestIndex, got, imageURL) + } + if got := messages[3].Get("content.1.image_url.detail").String(); got != "original" { + t.Fatalf("request %d native image detail = %q, want original", requestIndex, got) + } + if got := messages[4].Get("content.0.text").String(); got != "next" { + t.Fatalf("request %d final user text = %q, want next", requestIndex, got) + } + } +} + type kimiRoundTripperFunc func(*http.Request) (*http.Response, error) func (f kimiRoundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { -- 2.51.2 From 10d60930982cbe52da50bb5d0d93694dd0e3b331 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Wed, 29 Jul 2026 16:12:33 +0300 Subject: [PATCH 6/7] fix(kimi): retry transient engine overloads --- internal/runtime/executor/kimi_executor.go | 198 ++++++++++-------- .../runtime/executor/kimi_executor_test.go | 137 ++++++++++++ 2 files changed, 252 insertions(+), 83 deletions(-) diff --git a/internal/runtime/executor/kimi_executor.go b/internal/runtime/executor/kimi_executor.go index b05d00fa..1ae52dc3 100644 --- a/internal/runtime/executor/kimi_executor.go +++ b/internal/runtime/executor/kimi_executor.go @@ -30,9 +30,12 @@ import ( // KimiExecutor is a stateless executor for Kimi API using OpenAI-compatible chat completions. type KimiExecutor struct { ClaudeExecutor - cfg *config.Config + cfg *config.Config + retryWait func(context.Context, time.Duration) error } +const kimiEngineOverloadedMaxAttempts = 4 + // NewKimiExecutor creates a new Kimi executor. func NewKimiExecutor(cfg *config.Config) *KimiExecutor { return &KimiExecutor{ @@ -41,7 +44,8 @@ func NewKimiExecutor(cfg *config.Config) *KimiExecutor { requestLogProvider: "kimi", upstreamModelNormalizer: normalizeKimiUpstreamModel, }, - cfg: cfg, + cfg: cfg, + retryWait: waitForKimiRetry, } } @@ -81,6 +85,113 @@ func (e *KimiExecutor) HttpRequest(ctx context.Context, auth *cliproxyauth.Auth, return httpClient.Do(httpReq) } +func (e *KimiExecutor) executeKimiChatCompletionsRequest(ctx context.Context, auth *cliproxyauth.Auth, body []byte, token string, stream bool, httpClient *http.Client) (*http.Response, error) { + url := kimiauth.KimiAPIBaseURL + "/v1/chat/completions" + var authID, authLabel, authType, authValue string + if auth != nil { + authID = auth.ID + authLabel = auth.Label + authType, authValue = auth.AccountInfo() + } + + for attempt := 0; attempt < kimiEngineOverloadedMaxAttempts; attempt++ { + httpReq, errRequest := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if errRequest != nil { + return nil, errRequest + } + applyKimiHeadersWithAuth(httpReq, token, stream, auth) + var attrs map[string]string + if auth != nil { + attrs = auth.Attributes + } + util.ApplyCustomHeadersFromAttrs(httpReq, attrs) + helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ + URL: url, + Method: http.MethodPost, + Headers: httpReq.Header.Clone(), + Body: body, + Provider: e.Identifier(), + AuthID: authID, + AuthLabel: authLabel, + AuthType: authType, + AuthValue: authValue, + }) + + httpResp, errDo := httpClient.Do(httpReq) + if errDo != nil { + helps.RecordAPIResponseError(ctx, e.cfg, errDo) + return nil, errDo + } + helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) + if httpResp.StatusCode >= http.StatusOK && httpResp.StatusCode < http.StatusMultipleChoices { + return httpResp, nil + } + + responseBody, errRead := io.ReadAll(httpResp.Body) + if errClose := httpResp.Body.Close(); errClose != nil { + log.Errorf("kimi executor: close error response body error: %v", errClose) + } + if errRead != nil { + helps.RecordAPIResponseError(ctx, e.cfg, errRead) + return nil, errRead + } + helps.AppendAPIResponseChunk(ctx, e.cfg, responseBody) + helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), responseBody)) + + if !isKimiEngineOverloadedError(httpResp.StatusCode, responseBody) || attempt+1 >= kimiEngineOverloadedMaxAttempts { + return nil, statusErr{code: httpResp.StatusCode, msg: string(responseBody)} + } + + delay := kimiEngineOverloadedRetryDelay(attempt) + log.WithFields(log.Fields{ + "attempt": attempt + 1, + "max_attempts": kimiEngineOverloadedMaxAttempts, + "retry_delay": delay, + "upstream_model": gjson.GetBytes(body, "model").String(), + }).Warn("kimi executor: upstream engine overloaded, retrying") + wait := e.retryWait + if wait == nil { + wait = waitForKimiRetry + } + if errWait := wait(ctx, delay); errWait != nil { + return nil, errWait + } + } + + return nil, statusErr{code: http.StatusTooManyRequests, msg: "kimi engine remained overloaded after retries"} +} + +func isKimiEngineOverloadedError(statusCode int, body []byte) bool { + if statusCode != http.StatusTooManyRequests || len(body) == 0 { + return false + } + return strings.EqualFold(strings.TrimSpace(gjson.GetBytes(body, "error.type").String()), "engine_overloaded_error") +} + +func kimiEngineOverloadedRetryDelay(attempt int) time.Duration { + if attempt < 0 { + attempt = 0 + } + if attempt > 3 { + attempt = 3 + } + return time.Second << attempt +} + +func waitForKimiRetry(ctx context.Context, delay time.Duration) error { + if delay <= 0 { + return nil + } + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + // Execute performs a non-streaming chat completion request to Kimi. func (e *KimiExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { from := opts.SourceFormat @@ -134,40 +245,10 @@ func (e *KimiExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req body = helps.NormalizeMoonshotToolSchemas(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) - url := kimiauth.KimiAPIBaseURL + "/v1/chat/completions" - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) - if err != nil { - return resp, err - } - applyKimiHeadersWithAuth(httpReq, token, false, auth) - var attrs map[string]string - if auth != nil { - attrs = auth.Attributes - } - util.ApplyCustomHeadersFromAttrs(httpReq, attrs) - var authID, authLabel, authType, authValue string - if auth != nil { - authID = auth.ID - authLabel = auth.Label - authType, authValue = auth.AccountInfo() - } - helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ - URL: url, - Method: http.MethodPost, - Headers: httpReq.Header.Clone(), - Body: body, - Provider: e.Identifier(), - AuthID: authID, - AuthLabel: authLabel, - AuthType: authType, - AuthValue: authValue, - }) - httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpClient = reporter.TrackHTTPClient(httpClient) - httpResp, err := httpClient.Do(httpReq) + httpResp, err := e.executeKimiChatCompletionsRequest(ctx, auth, body, token, false, httpClient) if err != nil { - helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } defer func() { @@ -175,14 +256,6 @@ func (e *KimiExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req log.Errorf("kimi executor: close response body error: %v", errClose) } }() - helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) - if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { - b, _ := io.ReadAll(httpResp.Body) - helps.AppendAPIResponseChunk(ctx, e.cfg, b) - helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b)) - err = statusErr{code: httpResp.StatusCode, msg: string(b)} - return resp, err - } data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) @@ -254,51 +327,10 @@ func (e *KimiExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Aut body = helps.NormalizeMoonshotToolSchemas(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) - url := kimiauth.KimiAPIBaseURL + "/v1/chat/completions" - httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) - if err != nil { - return nil, err - } - applyKimiHeadersWithAuth(httpReq, token, true, auth) - var attrs map[string]string - if auth != nil { - attrs = auth.Attributes - } - util.ApplyCustomHeadersFromAttrs(httpReq, attrs) - var authID, authLabel, authType, authValue string - if auth != nil { - authID = auth.ID - authLabel = auth.Label - authType, authValue = auth.AccountInfo() - } - helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{ - URL: url, - Method: http.MethodPost, - Headers: httpReq.Header.Clone(), - Body: body, - Provider: e.Identifier(), - AuthID: authID, - AuthLabel: authLabel, - AuthType: authType, - AuthValue: authValue, - }) - httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpClient = reporter.TrackHTTPClient(httpClient) - httpResp, err := httpClient.Do(httpReq) + httpResp, err := e.executeKimiChatCompletionsRequest(ctx, auth, body, token, true, httpClient) if err != nil { - helps.RecordAPIResponseError(ctx, e.cfg, err) - return nil, err - } - helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) - if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 { - b, _ := io.ReadAll(httpResp.Body) - helps.AppendAPIResponseChunk(ctx, e.cfg, b) - helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b)) - if errClose := httpResp.Body.Close(); errClose != nil { - log.Errorf("kimi executor: close response body error: %v", errClose) - } - err = statusErr{code: httpResp.StatusCode, msg: string(b)} return nil, err } out := make(chan cliproxyexecutor.StreamChunk) diff --git a/internal/runtime/executor/kimi_executor_test.go b/internal/runtime/executor/kimi_executor_test.go index d2dd7e97..125c8812 100644 --- a/internal/runtime/executor/kimi_executor_test.go +++ b/internal/runtime/executor/kimi_executor_test.go @@ -5,8 +5,10 @@ import ( "io" "net/http" "net/http/httptest" + "slices" "strings" "testing" + "time" "github.com/gin-gonic/gin" "github.com/router-for-me/CLIProxyAPI/v7/internal/config" @@ -424,6 +426,141 @@ func TestKimiExecutorMovesResponsesToolImagesToNativeUserContentForBothExecution } } +func TestKimiExecutorRetriesEngineOverloadForNonStreamingRequest(t *testing.T) { + attempts := 0 + ctx := context.WithValue(context.Background(), "cliproxy.roundtripper", kimiRoundTripperFunc(func(req *http.Request) (*http.Response, error) { + attempts++ + if attempts < kimiEngineOverloadedMaxAttempts { + return kimiEngineOverloadedResponse(), nil + } + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader( + `{"id":"chatcmpl-test","object":"chat.completion","choices":[{"index":0,"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}`, + )), + }, nil + })) + + executor := NewKimiExecutor(&config.Config{}) + var delays []time.Duration + executor.retryWait = func(_ context.Context, delay time.Duration) error { + delays = append(delays, delay) + return nil + } + auth := &cliproxyauth.Auth{Attributes: map[string]string{}, Metadata: map[string]any{"access_token": "test-token"}} + payload := []byte(`{"model":"kimi-k3","messages":[{"role":"user","content":"hello"}]}`) + + if _, err := executor.Execute(ctx, auth, cliproxyexecutor.Request{Model: "kimi-k3", Payload: payload}, cliproxyexecutor.Options{SourceFormat: sdktranslator.FormatOpenAI}); err != nil { + t.Fatalf("Execute() error = %v", err) + } + if attempts != kimiEngineOverloadedMaxAttempts { + t.Fatalf("upstream attempts = %d, want %d", attempts, kimiEngineOverloadedMaxAttempts) + } + wantDelays := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second} + if !slices.Equal(delays, wantDelays) { + t.Fatalf("retry delays = %v, want %v", delays, wantDelays) + } +} + +func TestKimiExecutorRetriesEngineOverloadForStreamingRequest(t *testing.T) { + attempts := 0 + ctx := context.WithValue(context.Background(), "cliproxy.roundtripper", kimiRoundTripperFunc(func(req *http.Request) (*http.Response, error) { + attempts++ + if attempts == 1 { + return kimiEngineOverloadedResponse(), nil + } + return &http.Response{ + StatusCode: http.StatusOK, + Header: http.Header{"Content-Type": []string{"text/event-stream"}}, + Body: io.NopCloser(strings.NewReader( + "data: {\"id\":\"chatcmpl-test\",\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n" + + "data: [DONE]\n\n", + )), + }, nil + })) + + executor := NewKimiExecutor(&config.Config{}) + var delays []time.Duration + executor.retryWait = func(_ context.Context, delay time.Duration) error { + delays = append(delays, delay) + return nil + } + auth := &cliproxyauth.Auth{Attributes: map[string]string{}, Metadata: map[string]any{"access_token": "test-token"}} + payload := []byte(`{"model":"kimi-k3","stream":true,"messages":[{"role":"user","content":"hello"}]}`) + + result, err := executor.ExecuteStream(ctx, auth, cliproxyexecutor.Request{Model: "kimi-k3", Payload: payload}, cliproxyexecutor.Options{Stream: true, SourceFormat: sdktranslator.FormatOpenAI}) + if err != nil { + t.Fatalf("ExecuteStream() error = %v", err) + } + for chunk := range result.Chunks { + if chunk.Err != nil { + t.Fatalf("stream chunk error = %v", chunk.Err) + } + } + if attempts != 2 { + t.Fatalf("upstream attempts = %d, want 2", attempts) + } + if !slices.Equal(delays, []time.Duration{time.Second}) { + t.Fatalf("retry delays = %v, want [1s]", delays) + } +} + +func TestKimiExecutorDoesNotRetryQuota429(t *testing.T) { + attempts := 0 + ctx := context.WithValue(context.Background(), "cliproxy.roundtripper", kimiRoundTripperFunc(func(req *http.Request) (*http.Response, error) { + attempts++ + return &http.Response{ + StatusCode: http.StatusTooManyRequests, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader(`{"error":{"type":"rate_limit_error","message":"quota exhausted"}}`)), + }, nil + })) + + executor := NewKimiExecutor(&config.Config{}) + executor.retryWait = func(_ context.Context, _ time.Duration) error { + t.Fatal("quota 429 should not wait for a retry") + return nil + } + auth := &cliproxyauth.Auth{Attributes: map[string]string{}, Metadata: map[string]any{"access_token": "test-token"}} + payload := []byte(`{"model":"kimi-k3","messages":[{"role":"user","content":"hello"}]}`) + + _, err := executor.Execute(ctx, auth, cliproxyexecutor.Request{Model: "kimi-k3", Payload: payload}, cliproxyexecutor.Options{SourceFormat: sdktranslator.FormatOpenAI}) + assertStatusErr(t, err, http.StatusTooManyRequests) + if attempts != 1 { + t.Fatalf("upstream attempts = %d, want 1", attempts) + } +} + +func TestKimiExecutorBoundsPersistentEngineOverloadRetries(t *testing.T) { + attempts := 0 + ctx := context.WithValue(context.Background(), "cliproxy.roundtripper", kimiRoundTripperFunc(func(req *http.Request) (*http.Response, error) { + attempts++ + return kimiEngineOverloadedResponse(), nil + })) + + executor := NewKimiExecutor(&config.Config{}) + executor.retryWait = func(_ context.Context, _ time.Duration) error { return nil } + auth := &cliproxyauth.Auth{Attributes: map[string]string{}, Metadata: map[string]any{"access_token": "test-token"}} + payload := []byte(`{"model":"kimi-k3","messages":[{"role":"user","content":"hello"}]}`) + + _, err := executor.Execute(ctx, auth, cliproxyexecutor.Request{Model: "kimi-k3", Payload: payload}, cliproxyexecutor.Options{SourceFormat: sdktranslator.FormatOpenAI}) + assertStatusErr(t, err, http.StatusTooManyRequests) + if attempts != kimiEngineOverloadedMaxAttempts { + t.Fatalf("upstream attempts = %d, want %d", attempts, kimiEngineOverloadedMaxAttempts) + } +} + +func kimiEngineOverloadedResponse() *http.Response { + return &http.Response{ + StatusCode: http.StatusTooManyRequests, + Header: http.Header{"Content-Type": []string{"application/json"}}, + Body: io.NopCloser(strings.NewReader( + `{"error":{"message":"The engine is currently overloaded, please try again later","type":"engine_overloaded_error"}}`, + )), + } +} + type kimiRoundTripperFunc func(*http.Request) (*http.Response, error) func (f kimiRoundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { -- 2.51.2 From 35c67c5147a1fa6f3c23aef173cf9621c94f2618 Mon Sep 17 00:00:00 2001 From: "nekomimi.pet" Date: Wed, 29 Jul 2026 19:23:00 -0400 Subject: [PATCH 7/7] recover collapsed claude thinking replays --- .../executor/claude_executor_execute.go | 1 + .../executor/claude_executor_request.go | 78 +++++++++++++++++++ .../executor/claude_executor_stream.go | 1 + .../runtime/executor/claude_executor_test.go | 37 +++++++++ 4 files changed, 117 insertions(+) diff --git a/internal/runtime/executor/claude_executor_execute.go b/internal/runtime/executor/claude_executor_execute.go index da100536..16c93508 100644 --- a/internal/runtime/executor/claude_executor_execute.go +++ b/internal/runtime/executor/claude_executor_execute.go @@ -62,6 +62,7 @@ func (e *ClaudeExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, r requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", body, originalTranslated, requestedModel, requestPath, opts.Headers) body = ensureModelMaxTokens(body, baseModel) + body = sanitizeCollapsedClaudeThinkingReplay(body) // Disable thinking if tool_choice forces tool use (Anthropic API constraint) body = disableThinkingIfToolChoiceForced(body) diff --git a/internal/runtime/executor/claude_executor_request.go b/internal/runtime/executor/claude_executor_request.go index 847132e9..c14cbc71 100644 --- a/internal/runtime/executor/claude_executor_request.go +++ b/internal/runtime/executor/claude_executor_request.go @@ -64,6 +64,84 @@ func disableThinkingIfToolChoiceForced(body []byte) []byte { return body } +// sanitizeCollapsedClaudeThinkingReplay recovers Responses history where +// unsupported Claude server-tool blocks were dropped between thinking blocks. +// Anthropic rejects the collapsed assistant message because signed thinking +// blocks must retain their original positions. In that narrow case, omit all +// historical thinking and disable thinking for the recovery request. +func sanitizeCollapsedClaudeThinkingReplay(body []byte) []byte { + messages := gjson.GetBytes(body, "messages") + if !messages.Exists() || !messages.IsArray() { + return body + } + + collapsed := false + messages.ForEach(func(_, message gjson.Result) bool { + if message.Get("role").String() != "assistant" { + return true + } + content := message.Get("content") + if !content.IsArray() { + return true + } + previousThinking := false + content.ForEach(func(_, block gjson.Result) bool { + blockType := block.Get("type").String() + thinkingBlock := blockType == "thinking" || blockType == "redacted_thinking" + if thinkingBlock && previousThinking { + collapsed = true + return false + } + previousThinking = thinkingBlock + return true + }) + return !collapsed + }) + if !collapsed { + return body + } + + keptMessages := make([]string, 0, len(messages.Array())) + messages.ForEach(func(_, message gjson.Result) bool { + content := message.Get("content") + if message.Get("role").String() != "assistant" || !content.IsArray() { + keptMessages = append(keptMessages, message.Raw) + return true + } + + keptBlocks := make([]string, 0, len(content.Array())) + content.ForEach(func(_, block gjson.Result) bool { + blockType := block.Get("type").String() + if blockType != "thinking" && blockType != "redacted_thinking" { + keptBlocks = append(keptBlocks, block.Raw) + } + return true + }) + if len(keptBlocks) == 0 { + return true + } + updatedMessage, errSetContent := sjson.SetRaw(message.Raw, "content", string(rawJSONArray(keptBlocks))) + if errSetContent == nil { + keptMessages = append(keptMessages, updatedMessage) + } else { + keptMessages = append(keptMessages, message.Raw) + } + return true + }) + + updated, errSetMessages := sjson.SetRawBytes(body, "messages", rawJSONArray(keptMessages)) + if errSetMessages != nil { + return body + } + body = updated + body, _ = sjson.DeleteBytes(body, "thinking") + body, _ = sjson.DeleteBytes(body, "output_config.effort") + if outputConfig := gjson.GetBytes(body, "output_config"); outputConfig.Exists() && outputConfig.IsObject() && len(outputConfig.Map()) == 0 { + body, _ = sjson.DeleteBytes(body, "output_config") + } + return body +} + // normalizeClaudeSamplingForUpstream keeps Anthropic message requests valid. func normalizeClaudeSamplingForUpstream(body []byte) []byte { body, _ = sjson.DeleteBytes(body, "temperature") diff --git a/internal/runtime/executor/claude_executor_stream.go b/internal/runtime/executor/claude_executor_stream.go index 7f549933..4d4579f4 100644 --- a/internal/runtime/executor/claude_executor_stream.go +++ b/internal/runtime/executor/claude_executor_stream.go @@ -63,6 +63,7 @@ func (e *ClaudeExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.A requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", body, originalTranslated, requestedModel, requestPath, opts.Headers) body = ensureModelMaxTokens(body, baseModel) + body = sanitizeCollapsedClaudeThinkingReplay(body) // Disable thinking if tool_choice forces tool use (Anthropic API constraint) body = disableThinkingIfToolChoiceForced(body) diff --git a/internal/runtime/executor/claude_executor_test.go b/internal/runtime/executor/claude_executor_test.go index 9f6e9013..0c723193 100644 --- a/internal/runtime/executor/claude_executor_test.go +++ b/internal/runtime/executor/claude_executor_test.go @@ -3050,6 +3050,43 @@ func TestEnsureClaudeThinkingDisplay_SkipsWhenThinkingMissing(t *testing.T) { } } +func TestSanitizeCollapsedClaudeThinkingReplay(t *testing.T) { + payload := []byte(`{ + "thinking":{"type":"adaptive"}, + "output_config":{"effort":"high"}, + "messages":[ + {"role":"assistant","content":[{"type":"thinking","thinking":"old","signature":"sig-old"},{"type":"tool_use","id":"tool-1","name":"Read","input":{}}]}, + {"role":"user","content":[{"type":"tool_result","tool_use_id":"tool-1","content":"ok"}]}, + {"role":"assistant","content":[{"type":"thinking","thinking":"a","signature":"sig-a"},{"type":"redacted_thinking","data":"opaque"},{"type":"tool_use","id":"tool-2","name":"Bash","input":{}}]} + ] + }`) + + out := sanitizeCollapsedClaudeThinkingReplay(payload) + if gjson.GetBytes(out, "thinking").Exists() { + t.Fatalf("thinking config exists after collapsed replay recovery: %s", out) + } + if gjson.GetBytes(out, "output_config.effort").Exists() { + t.Fatalf("thinking effort exists after collapsed replay recovery: %s", out) + } + if strings.Contains(string(out), `"type":"thinking"`) || strings.Contains(string(out), `"type":"redacted_thinking"`) { + t.Fatalf("thinking blocks remain after collapsed replay recovery: %s", out) + } + if got := gjson.GetBytes(out, "messages.0.content.0.id").String(); got != "tool-1" { + t.Fatalf("first tool id = %q, want tool-1: %s", got, out) + } + if got := gjson.GetBytes(out, "messages.2.content.0.id").String(); got != "tool-2" { + t.Fatalf("second tool id = %q, want tool-2: %s", got, out) + } +} + +func TestSanitizeCollapsedClaudeThinkingReplayLeavesValidHistoryUnchanged(t *testing.T) { + payload := []byte(`{"thinking":{"type":"adaptive"},"messages":[{"role":"assistant","content":[{"type":"thinking","thinking":"a","signature":"sig-a"},{"type":"tool_use","id":"tool-1","name":"Read","input":{}}]}]}`) + out := sanitizeCollapsedClaudeThinkingReplay(payload) + if !bytes.Equal(out, payload) { + t.Fatalf("valid replay changed: got %s want %s", out, payload) + } +} + func TestPrependToFirstUserMessage_KeepsToolResultBlocksFirst(t *testing.T) { // A conversation that opens on an assistant tool_use makes the first user // message a tool_result carrier. Anthropic requires those blocks to stay at -- 2.51.2