From e3af4fce458bcce8632611e22fa8f3419dddb6f8 Mon Sep 17 00:00:00 2001 From: AmPlace <81105770+AmPlace@users.noreply.github.com> Date: Tue, 28 Jul 2026 09:46:05 +0800 Subject: [PATCH 1/2] fix(responses): normalize integer tool arguments --- .../infra/provider/cli/responses_arguments.go | 169 ++++++++++++++++++ .../provider/cli/responses_arguments_test.go | 128 +++++++++++++ .../infra/provider/cli/responses_response.go | 46 ++++- .../cli/responses_tool_declarations.go | 5 +- .../provider/cli/responses_tool_state.go | 2 + 5 files changed, 346 insertions(+), 4 deletions(-) create mode 100644 backend/internal/infra/provider/cli/responses_arguments.go create mode 100644 backend/internal/infra/provider/cli/responses_arguments_test.go diff --git a/backend/internal/infra/provider/cli/responses_arguments.go b/backend/internal/infra/provider/cli/responses_arguments.go new file mode 100644 index 000000000..fbfc2bf96 --- /dev/null +++ b/backend/internal/infra/provider/cli/responses_arguments.go @@ -0,0 +1,169 @@ +package cli + +import ( + "encoding/json" + "io" + "math" + "strconv" + "strings" +) + +const maxExactJSONInteger = float64(1<<53 - 1) + +// normalizeFunctionArguments repairs semantically integral JSON numbers that strict +// downstream decoders reject for integer fields. Grok Build can emit 60000.0 where +// clients such as Codex require the integer spelling 60000. +func normalizeFunctionArguments(arguments string, schema any) (string, bool) { + if strings.TrimSpace(arguments) == "" { + return arguments, false + } + decoder := json.NewDecoder(strings.NewReader(arguments)) + decoder.UseNumber() + var value any + if err := decoder.Decode(&value); err != nil { + return arguments, false + } + if err := decoder.Decode(new(any)); err != io.EOF { + return arguments, false + } + root, ok := schema.(map[string]any) + if !ok { + return arguments, false + } + normalized, changed := normalizeArgumentValue(value, root, root, 0) + if !changed { + return arguments, false + } + encoded, err := json.Marshal(normalized) + if err != nil { + return arguments, false + } + return string(encoded), true +} + +func normalizeArgumentValue(value any, schema, root map[string]any, depth int) (any, bool) { + if depth > 64 { + return value, false + } + changed := false + if ref, ok := schema["$ref"].(string); ok { + if resolved, ok := resolveLocalSchemaRef(root, ref); ok { + var current bool + value, current = normalizeArgumentValue(value, resolved, root, depth+1) + changed = changed || current + } + } + for _, keyword := range []string{"allOf", "anyOf", "oneOf"} { + branches, _ := schema[keyword].([]any) + for _, rawBranch := range branches { + branch, ok := rawBranch.(map[string]any) + if !ok { + continue + } + var current bool + value, current = normalizeArgumentValue(value, branch, root, depth+1) + changed = changed || current + } + } + if number, ok := value.(json.Number); ok && schemaRequiresInteger(schema) { + if normalized, ok := normalizeIntegralNumber(number); ok { + return normalized, true + } + return value, changed + } + switch typed := value.(type) { + case map[string]any: + properties, _ := schema["properties"].(map[string]any) + additional, _ := schema["additionalProperties"].(map[string]any) + for key, item := range typed { + property, ok := properties[key].(map[string]any) + if !ok { + property = additional + } + if property == nil { + continue + } + normalized, current := normalizeArgumentValue(item, property, root, depth+1) + if current { + typed[key] = normalized + changed = true + } + } + case []any: + prefixItems, _ := schema["prefixItems"].([]any) + items, _ := schema["items"].(map[string]any) + for index, item := range typed { + itemSchema := items + if index < len(prefixItems) { + if prefixSchema, ok := prefixItems[index].(map[string]any); ok { + itemSchema = prefixSchema + } + } + if itemSchema == nil { + continue + } + normalized, current := normalizeArgumentValue(item, itemSchema, root, depth+1) + if current { + typed[index] = normalized + changed = true + } + } + } + return value, changed +} + +func schemaRequiresInteger(schema map[string]any) bool { + switch value := schema["type"].(type) { + case string: + return value == "integer" + case []any: + integer := false + for _, item := range value { + kind, _ := item.(string) + if kind == "number" { + return false + } + integer = integer || kind == "integer" + } + return integer + default: + return false + } +} + +func normalizeIntegralNumber(number json.Number) (json.Number, bool) { + raw := number.String() + if !strings.ContainsAny(raw, ".eE") { + return number, false + } + value, err := strconv.ParseFloat(raw, 64) + if err != nil || math.IsInf(value, 0) || math.IsNaN(value) || math.Trunc(value) != value || math.Abs(value) > maxExactJSONInteger { + return number, false + } + normalized := strconv.FormatFloat(value, 'f', -1, 64) + if normalized == "-0" { + normalized = "0" + } + return json.Number(normalized), normalized != raw +} + +func schemaContainsInteger(value any) bool { + switch typed := value.(type) { + case map[string]any: + if schemaRequiresInteger(typed) { + return true + } + for _, child := range typed { + if schemaContainsInteger(child) { + return true + } + } + case []any: + for _, child := range typed { + if schemaContainsInteger(child) { + return true + } + } + } + return false +} diff --git a/backend/internal/infra/provider/cli/responses_arguments_test.go b/backend/internal/infra/provider/cli/responses_arguments_test.go new file mode 100644 index 000000000..99092fece --- /dev/null +++ b/backend/internal/infra/provider/cli/responses_arguments_test.go @@ -0,0 +1,128 @@ +package cli + +import ( + "encoding/json" + "io" + "strings" + "testing" +) + +func TestNormalizeFunctionArgumentsCoercesSchemaIntegers(t *testing.T) { + schema := map[string]any{ + "type": "object", + "properties": map[string]any{ + "timeout_ms": map[string]any{"type": "integer"}, + "ratio": map[string]any{"type": "number"}, + "items": map[string]any{ + "type": "array", + "items": map[string]any{"anyOf": []any{map[string]any{"type": "integer"}, map[string]any{"type": "null"}}}, + }, + }, + } + arguments := `{"timeout_ms":60000.0,"ratio":2.0,"items":[1.0,null,2.5]}` + normalized, changed := normalizeFunctionArguments(arguments, schema) + if !changed { + t.Fatal("expected integer arguments to be normalized") + } + if normalized != `{"items":[1,null,2.5],"ratio":2.0,"timeout_ms":60000}` { + t.Fatalf("normalized arguments = %s", normalized) + } +} + +func TestNormalizeFunctionArgumentsPreservesUnsafeValues(t *testing.T) { + schema := map[string]any{"type": "object", "properties": map[string]any{ + "fraction": map[string]any{"type": "integer"}, + "large": map[string]any{"type": "integer"}, + }} + arguments := `{"fraction":1.5,"large":9007199254740992.0}` + if normalized, changed := normalizeFunctionArguments(arguments, schema); changed || normalized != arguments { + t.Fatalf("unsafe arguments changed: changed=%v value=%s", changed, normalized) + } +} + +func TestNormalizeFunctionArgumentsFollowsLocalRefs(t *testing.T) { + schema := map[string]any{ + "$ref": "#/$defs/arguments", + "$defs": map[string]any{"arguments": map[string]any{ + "type": "object", + "properties": map[string]any{"timeout_ms": map[string]any{"type": "integer"}}, + }}, + } + normalized, changed := normalizeFunctionArguments(`{"timeout_ms":6e4}`, schema) + if !changed || normalized != `{"timeout_ms":60000}` { + t.Fatalf("referenced integer was not normalized: changed=%v value=%s", changed, normalized) + } +} + +func TestResponsesIntegerArgumentsNormalizedInJSONResponse(t *testing.T) { + request := []byte(`{ + "model":"public", + "tools":[{"type":"function","name":"wait_agent","parameters":{"type":"object","properties":{"timeout_ms":{"type":"integer"}}}}] + }`) + _, compatibility, err := normalizeResponsesRequest(request, "grok-4.5") + if err != nil { + t.Fatal(err) + } + if compatibility == nil { + t.Fatal("integer schema did not enable response compatibility") + } + response, err := compatibility.normalizeResponseJSON([]byte(`{ + "id":"resp_1","object":"response", + "output":[{"type":"function_call","call_id":"call_1","name":"wait_agent","arguments":"{\"timeout_ms\":60000.0}"}] + }`)) + if err != nil { + t.Fatal(err) + } + var payload map[string]any + if err := json.Unmarshal(response, &payload); err != nil { + t.Fatal(err) + } + call := payload["output"].([]any)[0].(map[string]any) + if call["arguments"] != `{"timeout_ms":60000}` { + t.Fatalf("function arguments = %q", call["arguments"]) + } +} + +func TestResponsesIntegerArgumentsNormalizedInStream(t *testing.T) { + request := []byte(`{ + "model":"public", + "stream":true, + "tools":[{"type":"function","name":"wait_agent","parameters":{"type":"object","properties":{"timeout_ms":{"type":"integer"}}}}] + }`) + _, compatibility, err := normalizeResponsesRequest(request, "grok-4.5") + if err != nil { + t.Fatal(err) + } + stream := strings.Join([]string{ + `event: response.output_item.added`, + `data: {"type":"response.output_item.added","item":{"id":"fc_1","type":"function_call","call_id":"call_1","name":"wait_agent","arguments":""}}`, + ``, + `event: response.function_call_arguments.delta`, + `data: {"type":"response.function_call_arguments.delta","item_id":"fc_1","delta":"{\"timeout_ms\":60000"}`, + ``, + `event: response.function_call_arguments.delta`, + `data: {"type":"response.function_call_arguments.delta","item_id":"fc_1","delta":".0}"}`, + ``, + `event: response.function_call_arguments.done`, + `data: {"type":"response.function_call_arguments.done","item_id":"fc_1","arguments":"{\"timeout_ms\":60000.0}"}`, + ``, + `event: response.output_item.done`, + `data: {"type":"response.output_item.done","item":{"id":"fc_1","type":"function_call","call_id":"call_1","name":"wait_agent","arguments":"{\"timeout_ms\":60000.0}"}}`, + ``, + }, "\n") + body, err := io.ReadAll(compatibility.normalizeResponseStream(io.NopCloser(strings.NewReader(stream)))) + if err != nil { + t.Fatal(err) + } + text := string(body) + if strings.Contains(text, `60000.0`) { + t.Fatalf("stream retained floating integer: %s", text) + } + if strings.Count(text, `response.function_call_arguments.delta`) != 2 { + // One occurrence is the event line and one is the JSON type field. + t.Fatalf("unexpected normalized delta count: %s", text) + } + if !strings.Contains(text, `{\"timeout_ms\":60000}`) { + t.Fatalf("normalized arguments missing: %s", text) + } +} diff --git a/backend/internal/infra/provider/cli/responses_response.go b/backend/internal/infra/provider/cli/responses_response.go index cded75c6c..a63f908c4 100644 --- a/backend/internal/infra/provider/cli/responses_response.go +++ b/backend/internal/infra/provider/cli/responses_response.go @@ -80,6 +80,7 @@ type responsesStreamOutput struct { type responsesStreamCall struct { identity responsesToolIdentity + schema any arguments strings.Builder lastDelta map[string]any addedPayload map[string]any @@ -117,6 +118,11 @@ func (c *responsesToolCompatibility) rewriteStreamData(event string, data []byte } if kind == "response.function_call_arguments.delta" { identity, state, found := c.streamIdentity(payload) + if found && identity.Kind == responsesFunctionTool && state.schema != nil { + state.arguments.WriteString(stringField(payload, "delta")) + state.lastDelta = cloneJSONObject(payload) + return nil, nil + } if found && (identity.Kind == responsesToolSearch || identity.Kind == responsesCustomTool || identity.Kind == responsesApplyPatchTool) { state.arguments.WriteString(stringField(payload, "delta")) if identity.Kind == responsesCustomTool { @@ -127,6 +133,31 @@ func (c *responsesToolCompatibility) rewriteStreamData(event string, data []byte } if kind == "response.function_call_arguments.done" { identity, state, found := c.streamIdentity(payload) + if found && identity.Kind == responsesFunctionTool && state.schema != nil { + arguments := stringField(payload, "arguments") + if arguments == "" { + arguments = state.arguments.String() + } + normalized, _ := normalizeFunctionArguments(arguments, state.schema) + outputs := make([]responsesStreamOutput, 0, 2) + if state.lastDelta != nil { + delta := cloneJSONObject(state.lastDelta) + delta["delta"] = normalized + encoded, err := json.Marshal(delta) + if err != nil { + return nil, fmt.Errorf("编码 function arguments delta: %w", err) + } + outputs = append(outputs, responsesStreamOutput{Event: "response.function_call_arguments.delta", Data: encoded}) + } + done := cloneJSONObject(payload) + done["arguments"] = normalized + encoded, err := json.Marshal(done) + if err != nil { + return nil, fmt.Errorf("编码 function arguments done: %w", err) + } + outputs = append(outputs, responsesStreamOutput{Event: "response.function_call_arguments.done", Data: encoded}) + return outputs, nil + } if found && (identity.Kind == responsesToolSearch || identity.Kind == responsesApplyPatchTool) { // Tool Search 的 arguments 是结构化对象；等 output_item.done 带齐参数后再对下游可见。 return nil, nil @@ -184,7 +215,7 @@ func (c *responsesToolCompatibility) rememberStreamCall(item map[string]any) *re if !exists { return nil } - state := &responsesStreamCall{identity: identity} + state := &responsesStreamCall{identity: identity, schema: c.functionSchemas[stringField(item, "name")]} for _, key := range []string{stringField(item, "id"), stringField(item, "call_id")} { if key != "" { c.streamCalls[key] = state @@ -203,7 +234,7 @@ func (c *responsesToolCompatibility) streamIdentity(payload map[string]any) (res if !exists { return responsesToolIdentity{}, nil, false } - state := &responsesStreamCall{identity: identity} + state := &responsesStreamCall{identity: identity, schema: c.functionSchemas[stringField(payload, "name")]} for _, key := range []string{stringField(payload, "item_id"), stringField(payload, "call_id")} { if key != "" { c.streamCalls[key] = state @@ -241,12 +272,21 @@ func (c *responsesToolCompatibility) rewriteResponseValue(value any) error { } func (c *responsesToolCompatibility) rewriteFunctionCall(call map[string]any) error { - identity, exists := c.aliases[stringField(call, "name")] + alias := stringField(call, "name") + identity, exists := c.aliases[alias] if !exists { return nil } switch identity.Kind { case responsesFunctionTool: + if schema := c.functionSchemas[alias]; schema != nil { + if arguments, ok := call["arguments"].(string); ok { + normalized, changed := normalizeFunctionArguments(arguments, schema) + if changed { + call["arguments"] = normalized + } + } + } call["name"] = identity.Name if identity.Namespace != "" { call["namespace"] = identity.Namespace diff --git a/backend/internal/infra/provider/cli/responses_tool_declarations.go b/backend/internal/infra/provider/cli/responses_tool_declarations.go index 685748734..d6b9ef75e 100644 --- a/backend/internal/infra/provider/cli/responses_tool_declarations.go +++ b/backend/internal/infra/provider/cli/responses_tool_declarations.go @@ -69,7 +69,7 @@ func normalizeResponsesTools(payload map[string]json.RawMessage) (*responsesTool if err := compatibility.normalizeToolChoice(payload, normalizedTools); err != nil { return nil, err } - if !compatibility.changed { + if !compatibility.changed && len(compatibility.functionSchemas) == 0 { return nil, nil } return compatibility, nil @@ -184,6 +184,9 @@ func (c *responsesToolCompatibility) normalizeTool(raw any, namespace string, cl } identity := responsesToolIdentity{Kind: responsesFunctionTool, Namespace: namespace, Name: name} alias := c.alias(identity) + if parameters, exists := tool["parameters"]; exists && schemaContainsInteger(parameters) { + c.functionSchemas[alias] = cloneJSONValue(parameters) + } converted["name"] = alias if namespace != "" || alias != name { c.changed = true diff --git a/backend/internal/infra/provider/cli/responses_tool_state.go b/backend/internal/infra/provider/cli/responses_tool_state.go index bfdbfc76b..726fad3b1 100644 --- a/backend/internal/infra/provider/cli/responses_tool_state.go +++ b/backend/internal/infra/provider/cli/responses_tool_state.go @@ -36,6 +36,7 @@ func (i responsesToolIdentity) key() string { type responsesToolCompatibility struct { aliases map[string]responsesToolIdentity identityAliases map[string]string + functionSchemas map[string]any visibleTools []any deferredSurfaces []string clientSearchTool map[string]any @@ -64,6 +65,7 @@ func newResponsesToolCompatibility() *responsesToolCompatibility { return &responsesToolCompatibility{ aliases: make(map[string]responsesToolIdentity), identityAliases: make(map[string]string), + functionSchemas: make(map[string]any), streamCalls: make(map[string]*responsesStreamCall), warningSet: make(map[string]struct{}), } From 8b5c1ed6376504e80979e0ed22b02833f77a39e1 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Tue, 28 Jul 2026 15:28:19 +0800 Subject: [PATCH 2/2] fix: safely normalize streamed integer tool arguments --- .../infra/provider/cli/responses_arguments.go | 131 +++++++++++++-- .../provider/cli/responses_arguments_test.go | 155 +++++++++++++++++- .../infra/provider/cli/responses_response.go | 134 ++++++++++----- .../provider/cli/responses_tool_state.go | 3 + 4 files changed, 360 insertions(+), 63 deletions(-) diff --git a/backend/internal/infra/provider/cli/responses_arguments.go b/backend/internal/infra/provider/cli/responses_arguments.go index fbfc2bf96..69608951f 100644 --- a/backend/internal/infra/provider/cli/responses_arguments.go +++ b/backend/internal/infra/provider/cli/responses_arguments.go @@ -3,12 +3,15 @@ package cli import ( "encoding/json" "io" - "math" "strconv" "strings" ) -const maxExactJSONInteger = float64(1<<53 - 1) +const ( + maxExactJSONInteger int64 = 1<<53 - 1 + maxNormalizedNumberBytes = 256 + maxExactJSONIntegerText = "9007199254740991" +) // normalizeFunctionArguments repairs semantically integral JSON numbers that strict // downstream decoders reject for integer fields. Grok Build can emit 60000.0 where @@ -133,37 +136,129 @@ func schemaRequiresInteger(schema map[string]any) bool { func normalizeIntegralNumber(number json.Number) (json.Number, bool) { raw := number.String() - if !strings.ContainsAny(raw, ".eE") { + if len(raw) > maxNormalizedNumberBytes || !strings.ContainsAny(raw, ".eE") { + return number, false + } + mantissa := raw + exponentText := "" + if index := strings.IndexAny(mantissa, "eE"); index >= 0 { + exponentText = mantissa[index+1:] + mantissa = mantissa[:index] + } + negative := strings.HasPrefix(mantissa, "-") + if negative { + mantissa = strings.TrimPrefix(mantissa, "-") + } + whole, fraction, hasFraction := strings.Cut(mantissa, ".") + if !hasFraction { + fraction = "" + } + digits := strings.TrimLeft(whole+fraction, "0") + if digits == "" { + return json.Number("0"), raw != "0" + } + exponent, ok := parseBoundedDecimalExponent(exponentText) + if !ok { return number, false } - value, err := strconv.ParseFloat(raw, 64) - if err != nil || math.IsInf(value, 0) || math.IsNaN(value) || math.Trunc(value) != value || math.Abs(value) > maxExactJSONInteger { + decimalShift := exponent - len(fraction) + if decimalShift < 0 { + fractionalDigits := -decimalShift + if fractionalDigits > len(digits) || strings.Trim(digits[len(digits)-fractionalDigits:], "0") != "" { + return number, false + } + digits = strings.TrimLeft(digits[:len(digits)-fractionalDigits], "0") + if digits == "" { + return json.Number("0"), true + } + } else if decimalShift > 0 { + if decimalShift > len(maxExactJSONIntegerText)-len(digits) { + return number, false + } + digits += strings.Repeat("0", decimalShift) + } + if len(digits) > len(maxExactJSONIntegerText) || len(digits) == len(maxExactJSONIntegerText) && digits > maxExactJSONIntegerText { return number, false } - normalized := strconv.FormatFloat(value, 'f', -1, 64) - if normalized == "-0" { - normalized = "0" + normalized := digits + if negative { + normalized = "-" + normalized } return json.Number(normalized), normalized != raw } -func schemaContainsInteger(value any) bool { - switch typed := value.(type) { - case map[string]any: - if schemaRequiresInteger(typed) { - return true +func parseBoundedDecimalExponent(raw string) (int, bool) { + if raw == "" { + return 0, true + } + sign := 1 + if raw[0] == '+' || raw[0] == '-' { + if raw[0] == '-' { + sign = -1 + } + raw = raw[1:] + if raw == "" { + return 0, false } - for _, child := range typed { - if schemaContainsInteger(child) { + } + raw = strings.TrimLeft(raw, "0") + if raw == "" { + return 0, true + } + if len(raw) > 9 { + return 0, false + } + exponent, err := strconv.Atoi(raw) + if err != nil { + return 0, false + } + return sign * exponent, true +} + +func schemaContainsInteger(value any) bool { + root, ok := value.(map[string]any) + if !ok { + return false + } + return schemaContainsReachableInteger(root, root, make(map[string]struct{}), 0) +} + +func schemaContainsReachableInteger(schema, root map[string]any, visitedRefs map[string]struct{}, depth int) bool { + if depth > 64 || schema == nil { + return false + } + if schemaRequiresInteger(schema) { + return true + } + if ref, ok := schema["$ref"].(string); ok { + if _, visited := visitedRefs[ref]; !visited { + visitedRefs[ref] = struct{}{} + if resolved, resolvedOK := resolveLocalSchemaRef(root, ref); resolvedOK && schemaContainsReachableInteger(resolved, root, visitedRefs, depth+1) { return true } } - case []any: - for _, child := range typed { - if schemaContainsInteger(child) { + } + for _, keyword := range []string{"allOf", "anyOf", "oneOf", "prefixItems"} { + branches, _ := schema[keyword].([]any) + for _, rawBranch := range branches { + branch, ok := rawBranch.(map[string]any) + if ok && schemaContainsReachableInteger(branch, root, visitedRefs, depth+1) { return true } } } + for _, keyword := range []string{"items", "additionalProperties"} { + child, _ := schema[keyword].(map[string]any) + if schemaContainsReachableInteger(child, root, visitedRefs, depth+1) { + return true + } + } + properties, _ := schema["properties"].(map[string]any) + for _, rawProperty := range properties { + property, ok := rawProperty.(map[string]any) + if ok && schemaContainsReachableInteger(property, root, visitedRefs, depth+1) { + return true + } + } return false } diff --git a/backend/internal/infra/provider/cli/responses_arguments_test.go b/backend/internal/infra/provider/cli/responses_arguments_test.go index 99092fece..9d85c5281 100644 --- a/backend/internal/infra/provider/cli/responses_arguments_test.go +++ b/backend/internal/infra/provider/cli/responses_arguments_test.go @@ -29,17 +29,68 @@ func TestNormalizeFunctionArgumentsCoercesSchemaIntegers(t *testing.T) { } } +func TestNormalizeIntegralNumberUsesExactBoundedDecimalArithmetic(t *testing.T) { + tests := []struct { + name string + input string + expected string + shouldChange bool + }{ + {name: "decimal", input: "60000.0", expected: "60000", shouldChange: true}, + {name: "exponent", input: "6e4", expected: "60000", shouldChange: true}, + {name: "negative exponent integer", input: "1000e-3", expected: "1", shouldChange: true}, + {name: "fraction with exponent", input: "1.2300e2", expected: "123", shouldChange: true}, + {name: "negative zero", input: "-0.0", expected: "0", shouldChange: true}, + {name: "positive exact limit", input: "9007199254740991.0", expected: "9007199254740991", shouldChange: true}, + {name: "negative exact limit", input: "-9007199254740991.0", expected: "-9007199254740991", shouldChange: true}, + {name: "fraction", input: "1e-1", expected: "1e-1", shouldChange: false}, + {name: "rounded fraction", input: "9007199254740990.5", expected: "9007199254740990.5", shouldChange: false}, + {name: "outside exact limit", input: "9007199254740992.0", expected: "9007199254740992.0", shouldChange: false}, + {name: "huge exponent", input: "1e1000000000", expected: "1e1000000000", shouldChange: false}, + {name: "zero huge exponent", input: "0e1000000000", expected: "0", shouldChange: true}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + normalized, changed := normalizeIntegralNumber(json.Number(test.input)) + if changed != test.shouldChange || normalized.String() != test.expected { + t.Fatalf("normalizeIntegralNumber(%q) = (%q, %v), want (%q, %v)", test.input, normalized, changed, test.expected, test.shouldChange) + } + }) + } +} + func TestNormalizeFunctionArgumentsPreservesUnsafeValues(t *testing.T) { schema := map[string]any{"type": "object", "properties": map[string]any{ - "fraction": map[string]any{"type": "integer"}, - "large": map[string]any{"type": "integer"}, + "fraction": map[string]any{"type": "integer"}, + "roundedFraction": map[string]any{"type": "integer"}, + "large": map[string]any{"type": "integer"}, + "hugeExponent": map[string]any{"type": "integer"}, }} - arguments := `{"fraction":1.5,"large":9007199254740992.0}` + arguments := `{"fraction":1.5,"roundedFraction":9007199254740990.5,"large":9007199254740992.0,"hugeExponent":1e1000000000}` if normalized, changed := normalizeFunctionArguments(arguments, schema); changed || normalized != arguments { t.Fatalf("unsafe arguments changed: changed=%v value=%s", changed, normalized) } } +func TestSchemaContainsIntegerOnlyFollowsReachableConstraints(t *testing.T) { + unreferenced := map[string]any{ + "type": "object", + "properties": map[string]any{"value": map[string]any{"type": "string"}}, + "$defs": map[string]any{"unused": map[string]any{"type": "integer"}}, + } + if schemaContainsInteger(unreferenced) { + t.Fatal("unreferenced integer definition enabled argument normalization") + } + referenced := map[string]any{ + "type": "object", + "properties": map[string]any{"value": map[string]any{"$ref": "#/$defs/count"}}, + "$defs": map[string]any{"count": map[string]any{"type": "integer"}}, + } + if !schemaContainsInteger(referenced) { + t.Fatal("referenced integer definition was not detected") + } +} + func TestNormalizeFunctionArgumentsFollowsLocalRefs(t *testing.T) { schema := map[string]any{ "$ref": "#/$defs/arguments", @@ -126,3 +177,101 @@ func TestResponsesIntegerArgumentsNormalizedInStream(t *testing.T) { t.Fatalf("normalized arguments missing: %s", text) } } + +func TestResponsesIntegerArgumentsParallelStreamSequenceIsMonotonic(t *testing.T) { + request := []byte(`{ + "model":"public", + "stream":true, + "parallel_tool_calls":true, + "tools":[{"type":"function","name":"wait_agent","parameters":{"type":"object","properties":{"timeout_ms":{"type":"integer"}}}}] + }`) + _, compatibility, err := normalizeResponsesRequest(request, "grok-4.5") + if err != nil { + t.Fatal(err) + } + stream := strings.Join([]string{ + `event: response.output_item.added`, + `data: {"type":"response.output_item.added","sequence_number":0,"item":{"id":"fc_1","type":"function_call","call_id":"call_1","name":"wait_agent","arguments":""}}`, + ``, + `event: response.output_item.added`, + `data: {"type":"response.output_item.added","sequence_number":1,"item":{"id":"fc_2","type":"function_call","call_id":"call_2","name":"wait_agent","arguments":""}}`, + ``, + `event: response.function_call_arguments.delta`, + `data: {"type":"response.function_call_arguments.delta","sequence_number":2,"item_id":"fc_1","delta":"{\"timeout_ms\":1.0}"}`, + ``, + `event: response.function_call_arguments.delta`, + `data: {"type":"response.function_call_arguments.delta","sequence_number":3,"item_id":"fc_2","delta":"{\"timeout_ms\":2.0}"}`, + ``, + `event: response.function_call_arguments.done`, + `data: {"type":"response.function_call_arguments.done","sequence_number":4,"item_id":"fc_1","arguments":"{\"timeout_ms\":1.0}"}`, + ``, + `event: response.function_call_arguments.done`, + `data: {"type":"response.function_call_arguments.done","sequence_number":5,"item_id":"fc_2","arguments":"{\"timeout_ms\":2.0}"}`, + ``, + }, "\n") + body, err := io.ReadAll(compatibility.normalizeResponseStream(io.NopCloser(strings.NewReader(stream)))) + if err != nil { + t.Fatal(err) + } + lastSequence := int64(-1) + for _, line := range strings.Split(string(body), "\n") { + if !strings.HasPrefix(line, "data: ") { + continue + } + var payload map[string]any + if err := json.Unmarshal([]byte(strings.TrimPrefix(line, "data: ")), &payload); err != nil { + t.Fatal(err) + } + sequence, ok := payload["sequence_number"].(float64) + if !ok { + continue + } + current := int64(sequence) + if current <= lastSequence { + t.Fatalf("sequence_number is not strictly increasing: previous=%d current=%d\n%s", lastSequence, current, body) + } + lastSequence = current + } +} + +func TestResponsesIntegerArgumentsBufferOverflowFallsBackToStreaming(t *testing.T) { + request := []byte(`{ + "model":"public", + "stream":true, + "tools":[{"type":"function","name":"write","parameters":{"type":"object","properties":{"content":{"type":"string"},"mode":{"type":"integer"}}}}] + }`) + _, compatibility, err := normalizeResponsesRequest(request, "grok-4.5") + if err != nil { + t.Fatal(err) + } + added := []byte(`{"type":"response.output_item.added","sequence_number":0,"item":{"id":"fc_1","type":"function_call","call_id":"call_1","name":"write","arguments":""}}`) + if _, err := compatibility.rewriteStreamData("response.output_item.added", added); err != nil { + t.Fatal(err) + } + first := strings.Repeat("a", maxBufferedFunctionArgumentsBytes) + firstPayload, _ := json.Marshal(map[string]any{"type": "response.function_call_arguments.delta", "sequence_number": 1, "item_id": "fc_1", "delta": first}) + outputs, err := compatibility.rewriteStreamData("response.function_call_arguments.delta", firstPayload) + if err != nil || len(outputs) != 0 { + t.Fatalf("first buffered delta: outputs=%d err=%v", len(outputs), err) + } + overflowPayload := []byte(`{"type":"response.function_call_arguments.delta","sequence_number":2,"item_id":"fc_1","delta":"b"}`) + outputs, err = compatibility.rewriteStreamData("response.function_call_arguments.delta", overflowPayload) + if err != nil { + t.Fatal(err) + } + if len(outputs) != 2 { + t.Fatalf("overflow outputs = %d", len(outputs)) + } + flushed := outputs[0].Payload + if delta, _ := flushed["delta"].(string); len(delta) != maxBufferedFunctionArgumentsBytes { + t.Fatalf("flushed delta length = %d", len(delta)) + } + current := outputs[1].Payload + if current["delta"] != "b" { + t.Fatalf("current delta = %q", current["delta"]) + } + state := compatibility.streamCalls["fc_1"] + if state == nil || !state.passthrough || state.arguments.Len() != 0 || compatibility.streamArgumentBytes != 0 { + t.Fatal("overflow did not release the buffered arguments") + } +} diff --git a/backend/internal/infra/provider/cli/responses_response.go b/backend/internal/infra/provider/cli/responses_response.go index a63f908c4..db8790097 100644 --- a/backend/internal/infra/provider/cli/responses_response.go +++ b/backend/internal/infra/provider/cli/responses_response.go @@ -10,8 +10,10 @@ import ( ) const ( - maxCompatibleResponseBytes = 128 << 20 - maxCompatibleSSEEventBytes = 8 << 20 + maxCompatibleResponseBytes = 128 << 20 + maxCompatibleSSEEventBytes = 8 << 20 + maxBufferedFunctionArgumentsBytes = 1 << 20 + maxTotalBufferedFunctionArgsBytes = 4 << 20 ) // normalizeResponseJSON 将上游普通函数别名恢复为下游 namespace 和 Tool Search 输出项。 @@ -51,6 +53,15 @@ func (c *responsesToolCompatibility) normalizeResponseStream(source io.ReadClose return rewriteErr } for index, output := range outputs { + outputData := output.Data + if output.Payload != nil { + c.resequenceStreamPayload(output.Payload) + encoded, encodeErr := json.Marshal(output.Payload) + if encodeErr != nil { + return fmt.Errorf("编码兼容 Responses SSE: %w", encodeErr) + } + outputData = encoded + } current := event if output.Event != "" { current.Event = output.Event @@ -61,7 +72,7 @@ func (c *responsesToolCompatibility) normalizeResponseStream(source io.ReadClose current.Comments = nil current.Other = nil } - current.SetData(output.Data) + current.SetData(outputData) if err := current.writeTo(writer); err != nil { return err } @@ -74,14 +85,16 @@ func (c *responsesToolCompatibility) normalizeResponseStream(source io.ReadClose } type responsesStreamOutput struct { - Event string - Data []byte + Event string + Data []byte + Payload map[string]any } type responsesStreamCall struct { identity responsesToolIdentity schema any arguments strings.Builder + passthrough bool lastDelta map[string]any addedPayload map[string]any } @@ -119,9 +132,28 @@ func (c *responsesToolCompatibility) rewriteStreamData(event string, data []byte if kind == "response.function_call_arguments.delta" { identity, state, found := c.streamIdentity(payload) if found && identity.Kind == responsesFunctionTool && state.schema != nil { - state.arguments.WriteString(stringField(payload, "delta")) - state.lastDelta = cloneJSONObject(payload) - return nil, nil + delta := stringField(payload, "delta") + callLimitExceeded := state.arguments.Len()+len(delta) > maxBufferedFunctionArgumentsBytes + totalLimitExceeded := c.streamArgumentBytes+len(delta) > maxTotalBufferedFunctionArgsBytes + if !state.passthrough && (callLimitExceeded || totalLimitExceeded) { + buffered := state.arguments.String() + c.releaseBufferedFunctionArguments(state) + state.passthrough = true + outputs := make([]responsesStreamOutput, 0, 2) + if buffered != "" { + flushed := cloneJSONObject(payload) + flushed["delta"] = buffered + outputs = append(outputs, responsesStreamOutput{Event: "response.function_call_arguments.delta", Payload: flushed}) + } + outputs = append(outputs, responsesStreamOutput{Event: "response.function_call_arguments.delta", Payload: payload}) + return outputs, nil + } + if !state.passthrough { + state.arguments.WriteString(delta) + c.streamArgumentBytes += len(delta) + state.lastDelta = cloneJSONObject(payload) + return nil, nil + } } if found && (identity.Kind == responsesToolSearch || identity.Kind == responsesCustomTool || identity.Kind == responsesApplyPatchTool) { state.arguments.WriteString(stringField(payload, "delta")) @@ -139,23 +171,21 @@ func (c *responsesToolCompatibility) rewriteStreamData(event string, data []byte arguments = state.arguments.String() } normalized, _ := normalizeFunctionArguments(arguments, state.schema) + if state.passthrough { + done := cloneJSONObject(payload) + done["arguments"] = normalized + return []responsesStreamOutput{{Event: "response.function_call_arguments.done", Payload: done}}, nil + } outputs := make([]responsesStreamOutput, 0, 2) if state.lastDelta != nil { delta := cloneJSONObject(state.lastDelta) delta["delta"] = normalized - encoded, err := json.Marshal(delta) - if err != nil { - return nil, fmt.Errorf("编码 function arguments delta: %w", err) - } - outputs = append(outputs, responsesStreamOutput{Event: "response.function_call_arguments.delta", Data: encoded}) + outputs = append(outputs, responsesStreamOutput{Event: "response.function_call_arguments.delta", Payload: delta}) } done := cloneJSONObject(payload) done["arguments"] = normalized - encoded, err := json.Marshal(done) - if err != nil { - return nil, fmt.Errorf("编码 function arguments done: %w", err) - } - outputs = append(outputs, responsesStreamOutput{Event: "response.function_call_arguments.done", Data: encoded}) + outputs = append(outputs, responsesStreamOutput{Event: "response.function_call_arguments.done", Payload: done}) + c.releaseBufferedFunctionArguments(state) return outputs, nil } if found && (identity.Kind == responsesToolSearch || identity.Kind == responsesApplyPatchTool) { @@ -171,18 +201,10 @@ func (c *responsesToolCompatibility) rewriteStreamData(event string, data []byte outputs := make([]responsesStreamOutput, 0, 2) if state.lastDelta != nil { delta := customToolStreamPayload(state.lastDelta, "response.custom_tool_call_input.delta", "delta", input) - encoded, err := json.Marshal(delta) - if err != nil { - return nil, fmt.Errorf("编码 custom tool delta: %w", err) - } - outputs = append(outputs, responsesStreamOutput{Event: "response.custom_tool_call_input.delta", Data: encoded}) + outputs = append(outputs, responsesStreamOutput{Event: "response.custom_tool_call_input.delta", Payload: delta}) } done := customToolStreamPayload(payload, "response.custom_tool_call_input.done", "input", input) - encoded, err := json.Marshal(done) - if err != nil { - return nil, fmt.Errorf("编码 custom tool done: %w", err) - } - outputs = append(outputs, responsesStreamOutput{Event: "response.custom_tool_call_input.done", Data: encoded}) + outputs = append(outputs, responsesStreamOutput{Event: "response.custom_tool_call_input.done", Payload: done}) return outputs, nil } } @@ -200,11 +222,47 @@ func (c *responsesToolCompatibility) rewriteStreamData(event string, data []byte if response, ok := payload["response"].(map[string]any); ok { c.restoreVisibleTools(response) } - converted, err := json.Marshal(payload) - if err != nil { - return nil, fmt.Errorf("编码兼容 Responses SSE: %w", err) + return []responsesStreamOutput{{Payload: payload}}, nil +} + +func (c *responsesToolCompatibility) releaseBufferedFunctionArguments(state *responsesStreamCall) { + if c == nil || state == nil { + return } - return []responsesStreamOutput{{Data: converted}}, nil + c.streamArgumentBytes -= state.arguments.Len() + if c.streamArgumentBytes < 0 { + c.streamArgumentBytes = 0 + } + state.arguments.Reset() + state.lastDelta = nil +} + +func (c *responsesToolCompatibility) resequenceStreamPayload(payload map[string]any) { + if c == nil || payload == nil { + return + } + rawSequence, exists := payload["sequence_number"] + if !exists { + return + } + if !c.streamSequenceSet { + sequence, ok := exactJSONInt64(rawSequence) + if !ok { + return + } + c.streamSequenceNext = sequence + c.streamSequenceSet = true + } + payload["sequence_number"] = c.streamSequenceNext + c.streamSequenceNext++ +} + +func exactJSONInt64(value any) (int64, bool) { + number, ok := value.(float64) + if !ok || number < 0 || number > float64(maxExactJSONInteger) || number != float64(int64(number)) { + return 0, false + } + return int64(number), true } func (c *responsesToolCompatibility) rememberStreamCall(item map[string]any) *responsesStreamCall { @@ -355,17 +413,9 @@ func (c *responsesToolCompatibility) rewriteApplyPatchDoneEvent(payload, item ma addedItem := cloneJSONObject(doneItem) addedItem["status"] = "in_progress" added["item"] = addedItem - addedData, err := json.Marshal(added) - if err != nil { - return nil, fmt.Errorf("编码 apply_patch added event: %w", err) - } - doneData, err := json.Marshal(done) - if err != nil { - return nil, fmt.Errorf("编码 apply_patch done event: %w", err) - } return []responsesStreamOutput{ - {Event: "response.output_item.added", Data: addedData}, - {Event: "response.output_item.done", Data: doneData}, + {Event: "response.output_item.added", Payload: added}, + {Event: "response.output_item.done", Payload: done}, }, nil } diff --git a/backend/internal/infra/provider/cli/responses_tool_state.go b/backend/internal/infra/provider/cli/responses_tool_state.go index 726fad3b1..ad0a94dc7 100644 --- a/backend/internal/infra/provider/cli/responses_tool_state.go +++ b/backend/internal/infra/provider/cli/responses_tool_state.go @@ -43,6 +43,9 @@ type responsesToolCompatibility struct { clientSearchParam string serverSearchEager bool streamCalls map[string]*responsesStreamCall + streamArgumentBytes int + streamSequenceNext int64 + streamSequenceSet bool legacyLocalShell bool nativeShell bool webSearchDisabled bool