feat: 完善模型请求适配与输出限制

apps/api/internal/runner/output_token_limit.go

@@ -0,0 +1,232 @@
+package runner
+
+import (
+	"fmt"
+	"math"
+	"strings"
+
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
+)
+
+const (
+	volcesOutputTokenThreshold       = 10240
+	volcesOutputCapabilityErrorCode  = "model_capability_configuration_error"
+	volcesOutputInvalidParameterCode = "invalid_parameter"
+)
+
+type outputTokenLimitProcessor struct{}
+
+func (outputTokenLimitProcessor) Name() string { return "OutputTokenLimitProcessor" }
+
+func (outputTokenLimitProcessor) ShouldProcess(_ map[string]any, modelType string, context *paramProcessContext) bool {
+	return context != nil && isOpenAITextGenerationKind(context.kind) && isTextOutputModelType(modelType) && isVolcesCandidate(context.candidate)
+}
+
+func (outputTokenLimitProcessor) Process(params map[string]any, modelType string, context *paramProcessContext) bool {
+	modelMax, sourceType, capabilityValue, ok := candidateMaxOutputTokens(context.candidate, modelType)
+	path := capabilityPath(sourceType, "max_output_tokens")
+	if !ok {
+		return context.reject(
+			"OutputTokenLimitProcessor",
+			outputTokenParameter(context.kind),
+			nil,
+			"火山引擎文本候选未配置正整数 max_output_tokens，已禁止执行该候选。",
+			path,
+			capabilityValue,
+		)
+	}
+
+	if context.kind == "chat.completions" {
+		if value, explicit := nonNullParameter(params, "max_tokens"); explicit {
+			if parsed, valid := positiveInteger(value); !valid || parsed > modelMax {
+				return context.reject("OutputTokenLimitProcessor", "max_tokens", value, fmt.Sprintf("max_tokens must be a positive integer no greater than the selected Volcengine model limit (%d)", modelMax), path, capabilityValue)
+			}
+			return true
+		}
+		if _, explicit := nonNullParameter(params, "max_completion_tokens"); explicit {
+			return true
+		}
+		value := defaultVolcesOutputTokens(modelMax)
+		params["max_tokens"] = value
+		context.recordChange(
+			"OutputTokenLimitProcessor", "set", "max_tokens", nil, value,
+			fmt.Sprintf("火山候选未显式设置输出上限：floor(%d/3)=%d，阈值=%d，注入候选级默认值。", modelMax, modelMax/3, volcesOutputTokenThreshold),
+			path, capabilityValue,
+		)
+		return true
+	}
+
+	if value, explicit := nonNullParameter(params, "max_output_tokens"); explicit {
+		if parsed, valid := positiveInteger(value); !valid || parsed > modelMax {
+			return context.reject("OutputTokenLimitProcessor", "max_output_tokens", value, fmt.Sprintf("max_output_tokens must be a positive integer no greater than the selected Volcengine model limit (%d)", modelMax), path, capabilityValue)
+		}
+		return true
+	}
+	value := defaultVolcesOutputTokens(modelMax)
+	params["max_output_tokens"] = value
+	context.recordChange(
+		"OutputTokenLimitProcessor", "set", "max_output_tokens", nil, value,
+		fmt.Sprintf("火山候选未显式设置输出上限：floor(%d/3)=%d，阈值=%d，注入候选级默认值。", modelMax, modelMax/3, volcesOutputTokenThreshold),
+		path, capabilityValue,
+	)
+	return true
+}
+
+func filterRuntimeCandidatesByOutputTokens(kind string, requestedModel string, modelType string, body map[string]any, candidates []store.RuntimeModelCandidate) ([]store.RuntimeModelCandidate, map[string]any, error) {
+	if !isOpenAITextGenerationKind(kind) || !isTextOutputModelType(modelType) || len(candidates) == 0 {
+		return candidates, nil, nil
+	}
+	filtered := make([]store.RuntimeModelCandidate, 0, len(candidates))
+	rejected := make([]map[string]any, 0)
+	invalidExplicit := false
+	for _, candidate := range candidates {
+		if !isVolcesCandidate(candidate) {
+			filtered = append(filtered, candidate)
+			continue
+		}
+		modelMax, sourceType, raw, configured := candidateMaxOutputTokens(candidate, modelType)
+		detail := map[string]any{
+			"platformId": candidate.PlatformID, "platformKey": candidate.PlatformKey, "provider": candidate.Provider,
+			"platformModelId": candidate.PlatformModelID, "providerModelName": candidate.ProviderModelName,
+			"modelType": modelType, "capabilityPath": capabilityPath(sourceType, "max_output_tokens"), "capabilityValue": raw,
+		}
+		if !configured {
+			detail["reason"] = "max_output_tokens_missing"
+			rejected = append(rejected, detail)
+			continue
+		}
+		if parameter, value, explicit := explicitOutputTokenParameter(kind, body); explicit {
+			parsed, valid := positiveInteger(value)
+			if !valid || (parameter != "max_completion_tokens" && parsed > modelMax) {
+				detail["reason"] = "requested_output_tokens_exceed_capability"
+				detail["parameter"] = parameter
+				detail["requestedValue"] = value
+				invalidExplicit = true
+				rejected = append(rejected, detail)
+				continue
+			}
+		}
+		filtered = append(filtered, candidate)
+	}
+	if len(rejected) == 0 {
+		return filtered, nil, nil
+	}
+	summary := map[string]any{
+		"filter": "volces_output_token_limit", "kind": kind, "requestedModel": requestedModel, "modelType": modelType,
+		"candidateCount": len(candidates), "supportedCandidateCount": len(filtered), "filteredCandidateCount": len(rejected),
+		"threshold": volcesOutputTokenThreshold, "formula": "third=floor(modelMaxOutputTokens/3); default=third>=10240?third:modelMaxOutputTokens",
+		"rejectedCandidates": rejected,
+	}
+	if len(filtered) > 0 {
+		return filtered, summary, nil
+	}
+	code := volcesOutputCapabilityErrorCode
+	message := "所有火山引擎文本候选都缺少有效的 max_output_tokens 能力配置"
+	if invalidExplicit {
+		code = volcesOutputInvalidParameterCode
+		message = "请求的输出 token 上限超过所有可用火山引擎候选的模型能力"
+	}
+	summary["code"] = code
+	return nil, summary, &store.ModelCandidateUnavailableError{Code: code, Message: message, Details: summary}
+}
+
+func defaultVolcesOutputTokens(modelMax int) int {
+	third := modelMax / 3
+	if third >= volcesOutputTokenThreshold {
+		return third
+	}
+	return modelMax
+}
+
+func isVolcesCandidate(candidate store.RuntimeModelCandidate) bool {
+	provider := strings.ToLower(strings.TrimSpace(candidate.Provider))
+	baseURL := strings.ToLower(strings.TrimSpace(candidate.BaseURL))
+	return provider == "volces-openai" || strings.Contains(baseURL, "volces.com") || strings.Contains(baseURL, "byteplus.com")
+}
+
+func candidateMaxOutputTokens(candidate store.RuntimeModelCandidate, modelType string) (int, string, any, bool) {
+	capabilities := effectiveModelCapability(candidate)
+	seen := map[string]struct{}{}
+	for _, candidateType := range []string{candidate.ModelType, modelType, "text_generate"} {
+		candidateType = strings.TrimSpace(candidateType)
+		if candidateType == "" {
+			continue
+		}
+		if _, ok := seen[candidateType]; ok {
+			continue
+		}
+		seen[candidateType] = struct{}{}
+		capability := capabilityForType(capabilities, candidateType)
+		if capability == nil {
+			continue
+		}
+		raw, exists := capability["max_output_tokens"]
+		if !exists {
+			continue
+		}
+		value, ok := positiveInteger(raw)
+		return value, candidateType, raw, ok
+	}
+	return 0, firstNonEmptyString(candidate.ModelType, modelType, "text_generate"), nil, false
+}
+
+func positiveInteger(value any) (int, bool) {
+	number := floatFromAny(value)
+	if number <= 0 || math.Trunc(number) != number || number > float64(math.MaxInt) {
+		return 0, false
+	}
+	return int(number), true
+}
+
+func nonNullParameter(body map[string]any, key string) (any, bool) {
+	value, ok := body[key]
+	return value, ok && value != nil
+}
+
+func explicitOutputTokenParameter(kind string, body map[string]any) (string, any, bool) {
+	if kind == "responses" {
+		value, ok := nonNullParameter(body, "max_output_tokens")
+		return "max_output_tokens", value, ok
+	}
+	if value, ok := nonNullParameter(body, "max_tokens"); ok {
+		return "max_tokens", value, true
+	}
+	value, ok := nonNullParameter(body, "max_completion_tokens")
+	return "max_completion_tokens", value, ok
+}
+
+func outputTokenParameter(kind string) string {
+	if kind == "responses" {
+		return "max_output_tokens"
+	}
+	return "max_tokens"
+}
+
+func isOpenAITextGenerationKind(kind string) bool {
+	return kind == "chat.completions" || kind == "responses"
+}
+
+func isTextOutputModelType(modelType string) bool {
+	switch strings.TrimSpace(modelType) {
+	case "", "text_generate", "chat", "responses", "text":
+		return true
+	default:
+		return false
+	}
+}
+
+func mergeCandidateFilterSummaries(summaries ...map[string]any) map[string]any {
+	nonEmpty := make([]any, 0, len(summaries))
+	for _, summary := range summaries {
+		if len(summary) > 0 {
+			nonEmpty = append(nonEmpty, summary)
+		}
+	}
+	if len(nonEmpty) == 0 {
+		return nil
+	}
+	if len(nonEmpty) == 1 {
+		return nonEmpty[0].(map[string]any)
+	}
+	return map[string]any{"filters": nonEmpty}
+}

apps/api/internal/runner/output_token_limit_test.go

@@ -0,0 +1,113 @@
+package runner
+
+import (
+	"testing"
+
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
+)
+
+func volcTextCandidate(maxOutput any) store.RuntimeModelCandidate {
+	capability := map[string]any{}
+	if maxOutput != nil {
+		capability["max_output_tokens"] = maxOutput
+	}
+	return store.RuntimeModelCandidate{
+		Provider: "volces-openai", BaseURL: "https://ark.cn-beijing.volces.com/api/v3",
+		ModelType: "text_generate", ProviderModelName: "demo",
+		Capabilities: map[string]any{"text_generate": capability},
+	}
+}
+
+func TestDefaultVolcesOutputTokens(t *testing.T) {
+	tests := []struct {
+		name string
+		max  int
+		want int
+	}{
+		{name: "128K", max: 131072, want: 43690},
+		{name: "32K", max: 32768, want: 10922},
+		{name: "24K below threshold", max: 24576, want: 24576},
+		{name: "threshold exact", max: 30720, want: 10240},
+		{name: "threshold minus one", max: 30719, want: 30719},
+		{name: "floor", max: 131071, want: 43690},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if got := defaultVolcesOutputTokens(test.max); got != test.want {
+				t.Fatalf("defaultVolcesOutputTokens(%d)=%d, want %d", test.max, got, test.want)
+			}
+		})
+	}
+}
+
+func TestVolcesOutputProcessorInjectsCandidateSpecificDefaults(t *testing.T) {
+	chat := preprocessRequestWithLog("chat.completions", map[string]any{"messages": []any{}, "max_tokens": nil}, volcTextCandidate(131072))
+	if chat.Err != nil || chat.Body["max_tokens"] != 43690 {
+		t.Fatalf("unexpected Chat preprocessing: body=%+v err=%v", chat.Body, chat.Err)
+	}
+	if len(chat.Log.Changes) != 1 || chat.Log.Changes[0].CapabilityPath != "capabilities.text_generate.max_output_tokens" {
+		t.Fatalf("expected auditable capability source, got %+v", chat.Log.Changes)
+	}
+
+	responses := preprocessRequestWithLog("responses", map[string]any{"input": "hello", "max_output_tokens": nil}, volcTextCandidate(24576))
+	if responses.Err != nil || responses.Body["max_output_tokens"] != 24576 {
+		t.Fatalf("unexpected Responses preprocessing: body=%+v err=%v", responses.Body, responses.Err)
+	}
+}
+
+func TestVolcesOutputProcessorPreservesExplicitLimits(t *testing.T) {
+	for _, body := range []map[string]any{
+		{"messages": []any{}, "max_tokens": 1234},
+		{"messages": []any{}, "max_completion_tokens": 2345},
+		{"messages": []any{}, "max_tokens": 1234, "max_completion_tokens": 2345},
+	} {
+		result := preprocessRequestWithLog("chat.completions", body, volcTextCandidate(32768))
+		if result.Err != nil || len(result.Log.Changes) != 0 {
+			t.Fatalf("explicit Chat limit should remain unchanged: body=%+v err=%v changes=%+v", result.Body, result.Err, result.Log.Changes)
+		}
+	}
+	result := preprocessRequestWithLog("responses", map[string]any{"input": "hello", "max_output_tokens": 3456}, volcTextCandidate(32768))
+	if result.Err != nil || result.Body["max_output_tokens"] != 3456 || len(result.Log.Changes) != 0 {
+		t.Fatalf("explicit Responses limit should remain unchanged: %+v", result)
+	}
+}
+
+func TestVolcesOutputCandidateFilterSupportsFailover(t *testing.T) {
+	missing := volcTextCandidate(nil)
+	missing.PlatformID = "missing"
+	valid := volcTextCandidate(32768)
+	valid.PlatformID = "valid"
+	filtered, summary, err := filterRuntimeCandidatesByOutputTokens("chat.completions", "demo", "text_generate", map[string]any{"messages": []any{}}, []store.RuntimeModelCandidate{missing, valid})
+	if err != nil || len(filtered) != 1 || filtered[0].PlatformID != "valid" {
+		t.Fatalf("expected missing capability candidate to be skipped: filtered=%+v summary=%+v err=%v", filtered, summary, err)
+	}
+	result := preprocessRequestWithLog("chat.completions", map[string]any{"messages": []any{}}, filtered[0])
+	if result.Body["max_tokens"] != 10922 {
+		t.Fatalf("failover candidate must recalculate its own default, got %+v", result.Body)
+	}
+}
+
+func TestVolcesOutputCandidateFilterRejectsMissingAndExceededCapabilities(t *testing.T) {
+	_, _, err := filterRuntimeCandidatesByOutputTokens("responses", "demo", "text_generate", map[string]any{"input": "hello"}, []store.RuntimeModelCandidate{volcTextCandidate(nil)})
+	if store.ModelCandidateErrorCode(err) != volcesOutputCapabilityErrorCode {
+		t.Fatalf("expected capability configuration error, got %v", err)
+	}
+	_, _, err = filterRuntimeCandidatesByOutputTokens("chat.completions", "demo", "text_generate", map[string]any{"messages": []any{}, "max_tokens": 32769}, []store.RuntimeModelCandidate{volcTextCandidate(32768)})
+	if store.ModelCandidateErrorCode(err) != volcesOutputInvalidParameterCode {
+		t.Fatalf("expected invalid_parameter for explicit limit, got %v", err)
+	}
+}
+
+func TestNonVolcesOutputProcessorDoesNotInject(t *testing.T) {
+	candidate := volcTextCandidate(131072)
+	candidate.Provider = "openai"
+	candidate.BaseURL = "https://api.openai.com/v1"
+	chat := preprocessRequestWithLog("chat.completions", map[string]any{"messages": []any{}}, candidate)
+	if _, ok := chat.Body["max_tokens"]; ok {
+		t.Fatalf("non-Volcengine Chat candidate must remain unchanged: %+v", chat.Body)
+	}
+	responses := preprocessRequestWithLog("responses", map[string]any{"input": "hello"}, candidate)
+	if _, ok := responses.Body["max_output_tokens"]; ok {
+		t.Fatalf("non-Volcengine Responses candidate must remain unchanged: %+v", responses.Body)
+	}
+}

apps/api/internal/runner/param_processor.go

@@ -55,6 +55,7 @@ type parameterPreprocessChange struct {
 func NewParamProcessorChain() ParamProcessorChain {
 	return ParamProcessorChain{
 		processors: []paramProcessor{
+			outputTokenLimitProcessor{},
 			resolutionNormalizeProcessor{},
 			aspectRatioProcessor{},
 			imageSizeProcessor{},

apps/api/internal/runner/pricing.go

@@ -28,6 +28,10 @@ func (s *Service) Estimate(ctx context.Context, kind string, model string, body
 	if err != nil {
 		return EstimateResult{}, err
 	}
+	candidates, _, err = filterRuntimeCandidatesByOutputTokens(kind, model, modelType, body, candidates)
+	if err != nil {
+		return EstimateResult{}, err
+	}
 	candidate := candidates[0]
 	body = preprocessRequest(kind, body, candidate)
 	items := s.estimatedBillings(ctx, user, kind, body, candidate)

apps/api/internal/runner/queue_worker.go

@@ -88,7 +88,10 @@ func (s *Service) startRiverQueue(ctx context.Context) error {
 		Queues: map[string]river.QueueConfig{
 			asyncTaskQueueName: {MaxWorkers: 32},
 		},
-		RescueStuckJobsAfter: 30 * time.Second,
+		// Provider-backed media jobs commonly poll for 10-20 minutes. River may
+		// execute a still-running job again once this window elapses, so keep the
+		// rescue horizon above the longest configured provider poll timeout.
+		RescueStuckJobsAfter: time.Hour,
 		TestOnly:             s.cfg.AppEnv == "test",
 		Workers:              workers,
 	})

apps/api/internal/runner/service.go

@@ -230,6 +230,22 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 		}
 		return Result{Task: failed, Output: failed.Result}, err
 	}
+	var outputTokenFilterSummary map[string]any
+	candidates, outputTokenFilterSummary, err = filterRuntimeCandidatesByOutputTokens(task.Kind, task.Model, modelType, body, candidates)
+	candidateFilterSummary = mergeCandidateFilterSummaries(candidateFilterSummary, outputTokenFilterSummary)
+	if err != nil {
+		candidateFilterMetrics := candidateCapabilityFilterMetrics(candidateFilterSummary)
+		s.recordFailedAttempt(ctx, failedAttemptRecord{
+			Task: task, Body: body, AttemptNo: task.AttemptCount + 1, Code: store.ModelCandidateErrorCode(err), Cause: err,
+			Simulated: task.RunMode == "simulation", Scope: "candidate_output_token_filter", Reason: store.ModelCandidateErrorCode(err),
+			ExtraMetrics: []map[string]any{candidateFilterMetrics}, ModelType: modelType,
+		})
+		failed, finishErr := s.failTask(ctx, task.ID, store.ModelCandidateErrorCode(err), err.Error(), task.RunMode == "simulation", err, candidateFilterMetrics)
+		if finishErr != nil {
+			return Result{}, finishErr
+		}
+		return Result{Task: failed, Output: failed.Result}, err
+	}
 	if task.Kind == "responses" {
 		candidates, err = prepareResponseCandidates(candidates, responseExecution)
 		if err != nil {
@@ -839,7 +855,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 }
 
 func (s *Service) recordTaskParameterPreprocessing(ctx context.Context, taskID string, attemptID string, attemptNo int, candidate store.RuntimeModelCandidate, log parameterPreprocessingLog) error {
-	if skipTaskParameterPreprocessingLog(log.ModelType) {
+	if skipTaskParameterPreprocessingLog(log.ModelType) && !log.Changed {
 		return nil
 	}
 	_, err := s.store.CreateTaskParamPreprocessingLog(ctx, store.CreateTaskParamPreprocessingLogInput{
@@ -1348,10 +1364,22 @@ func validateRequest(kind string, body map[string]any) error {
 		if err := clients.ValidateOpenAIReasoningEffort(body["reasoning_effort"]); err != nil {
 			return err
 		}
+		for _, key := range []string{"max_tokens", "max_completion_tokens"} {
+			if value, explicit := nonNullParameter(body, key); explicit {
+				if _, ok := positiveInteger(value); !ok {
+					return &clients.ClientError{Code: "invalid_parameter", Message: key + " must be a positive integer", Param: key, StatusCode: 400, Retryable: false}
+				}
+			}
+		}
 	case "responses":
 		if body["input"] == nil && body["messages"] == nil {
 			return errors.New("input or messages is required")
 		}
+		if value, explicit := nonNullParameter(body, "max_output_tokens"); explicit {
+			if _, ok := positiveInteger(value); !ok {
+				return &clients.ClientError{Code: "invalid_parameter", Message: "max_output_tokens must be a positive integer", Param: "max_output_tokens", StatusCode: 400, Retryable: false}
+			}
+		}
 	case "embeddings":
 		if body["input"] == nil {
 			return errors.New("input is required")

apps/api/internal/runner/service_test.go

@@ -16,7 +16,7 @@ func (namedClient) Run(context.Context, clients.Request) (clients.Response, erro
 }
 
 func TestValidateRequestAcceptsOpenAIReasoningEffort(t *testing.T) {
-	for _, effort := range []string{"none", "minimal", "low", "medium", "high", "xhigh"} {
+	for _, effort := range []string{"none", "minimal", "low", "medium", "high", "xhigh", "max"} {
 		t.Run(effort, func(t *testing.T) {
 			err := validateRequest("chat.completions", map[string]any{
 				"messages":         []any{map[string]any{"role": "user", "content": "ping"}},
@@ -30,7 +30,7 @@ func TestValidateRequestAcceptsOpenAIReasoningEffort(t *testing.T) {
 }
 
 func TestValidateRequestRejectsNonOpenAIReasoningEffort(t *testing.T) {
-	for _, effort := range []string{"max", "auto"} {
+	for _, effort := range []string{"auto"} {
 		t.Run(effort, func(t *testing.T) {
 			err := validateRequest("chat.completions", map[string]any{
 				"messages":         []any{map[string]any{"role": "user", "content": "ping"}},