feat: add parameter preprocessing audit trail

2026-05-12 13:54:51 +08:00
parent 9ea83be718
commit b9c9f457e9
17 changed files with 2323 additions and 28 deletions
@@ -96,11 +96,24 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 		}
 		return Result{Task: failed, Output: failed.Result}, err
 	}
+	firstCandidateBody := body
+	normalizedModelType := modelType
+	var firstPreprocessing parameterPreprocessingLog
 	if len(candidates) > 0 {
-		estimatedBillings := s.estimatedBillings(ctx, user, task.Kind, body, candidates[0])
+		preprocessing := preprocessRequestWithLog(task.Kind, body, candidates[0])
+		firstCandidateBody = preprocessing.Body
+		firstPreprocessing = preprocessing.Log
+		normalizedModelType = candidates[0].ModelType
+		if err := s.store.MarkTaskRunning(ctx, task.ID, candidates[0].ModelType, firstCandidateBody); err != nil {
+			return Result{}, err
+		}
+		estimatedBillings := s.estimatedBillings(ctx, user, task.Kind, firstCandidateBody, candidates[0])
 		if err := s.ensureWalletBalance(ctx, user, estimatedBillings); err != nil {
 			if errors.Is(err, store.ErrInsufficientWalletBalance) {
-				failed, finishErr := s.failTask(ctx, task.ID, "insufficient_balance", err.Error(), task.RunMode == "simulation", err)
+				if logErr := s.recordTaskParameterPreprocessing(ctx, task.ID, "", 0, candidates[0], firstPreprocessing); logErr != nil {
+					return Result{}, logErr
+				}
+				failed, finishErr := s.failTask(ctx, task.ID, "insufficient_balance", err.Error(), task.RunMode == "simulation", err, parameterPreprocessingMetrics(firstPreprocessing))
 				if finishErr != nil {
 					return Result{}, finishErr
 				}
@@ -109,7 +122,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 			return Result{}, err
 		}
 	}
-	if err := s.emit(ctx, task.ID, "task.progress", "running", "normalizing", 0.15, "request normalized", map[string]any{"modelType": modelType}, task.RunMode == "simulation"); err != nil {
+	if err := s.emit(ctx, task.ID, "task.progress", "running", "normalizing", 0.15, "request normalized", map[string]any{"modelType": normalizedModelType}, task.RunMode == "simulation"); err != nil {
 		return Result{}, err
 	}

@@ -122,6 +135,7 @@ func (s *Service) execute(ctx context.Context, task store.GatewayTask, user *aut
 	attemptNo := task.AttemptCount
 	var lastErr error
 	var lastCandidate store.RuntimeModelCandidate
+	var lastPreprocessing *parameterPreprocessingLog
 candidatesLoop:
 	for index, candidate := range candidates {
 		if index >= maxPlatforms {
@@ -132,11 +146,16 @@ candidatesLoop:
 		var candidateErr error
 		for clientAttempt := 1; clientAttempt <= clientAttempts; clientAttempt++ {
 			nextAttemptNo := attemptNo + 1
-			response, err := s.runCandidate(ctx, task, user, body, candidate, nextAttemptNo, onDelta)
+			preprocessing := preprocessRequestWithLog(task.Kind, body, candidate)
+			preprocessingLog := preprocessing.Log
+			lastPreprocessing = &preprocessingLog
+			candidateBody := preprocessing.Body
+			response, err := s.runCandidate(ctx, task, user, candidateBody, preprocessing.Log, candidate, nextAttemptNo, onDelta)
 			if err == nil {
 				attemptNo = nextAttemptNo
-				billings := s.billings(ctx, user, task.Kind, body, candidate, response, isSimulation(task, candidate))
-				record := buildSuccessRecord(task, user, body, candidate, response, billings, isSimulation(task, candidate))
+				billings := s.billings(ctx, user, task.Kind, candidateBody, candidate, response, isSimulation(task, candidate))
+				record := buildSuccessRecord(task, user, candidateBody, candidate, response, billings, isSimulation(task, candidate))
+				record.Metrics = mergeMetrics(record.Metrics, parameterPreprocessingMetrics(preprocessing.Log))
 				record.Metrics = s.withAttemptHistory(ctx, task.ID, record.Metrics)
 				finished, finishErr := s.store.FinishTaskSuccess(ctx, store.FinishTaskSuccessInput{
 					TaskID:             task.ID,
@@ -305,15 +324,20 @@ candidatesLoop:
 		}
 		return Result{Task: queued, Output: queued.Result}, &TaskQueuedError{Delay: delay}
 	}
-	failed, err := s.failTask(ctx, task.ID, code, message, task.RunMode == "simulation", lastErr)
+	extraMetrics := []map[string]any{}
+	if lastPreprocessing != nil {
+		extraMetrics = append(extraMetrics, parameterPreprocessingMetrics(*lastPreprocessing))
+	}
+	failed, err := s.failTask(ctx, task.ID, code, message, task.RunMode == "simulation", lastErr, extraMetrics...)
 	if err != nil {
 		return Result{}, err
 	}
 	return Result{Task: failed, Output: failed.Result}, lastErr
 }

-func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user *auth.User, body map[string]any, candidate store.RuntimeModelCandidate, attemptNo int, onDelta clients.StreamDelta) (clients.Response, error) {
+func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user *auth.User, body map[string]any, preprocessing parameterPreprocessingLog, candidate store.RuntimeModelCandidate, attemptNo int, onDelta clients.StreamDelta) (clients.Response, error) {
 	simulated := isSimulation(task, candidate)
+	baseAttemptMetrics := mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), parameterPreprocessingMetrics(preprocessing))
 	reservations := s.rateLimitReservations(ctx, user, candidate, body)
 	limitResult, err := s.store.ReserveRateLimits(ctx, task.ID, "", reservations)
 	if err != nil {
@@ -339,18 +363,30 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 		Status:          "running",
 		Simulated:       simulated,
 		RequestSnapshot: body,
-		Metrics:         attemptMetrics(candidate, attemptNo, simulated),
+		Metrics:         baseAttemptMetrics,
 	})
 	if err != nil {
 		return clients.Response{}, fmt.Errorf("create task attempt: %w", err)
 	}
+	if err := s.recordTaskParameterPreprocessing(ctx, task.ID, attemptID, attemptNo, candidate, preprocessing); err != nil {
+		clientErr := &clients.ClientError{Code: "runtime_error", Message: err.Error(), Retryable: false}
+		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
+			AttemptID:    attemptID,
+			Status:       "failed",
+			Retryable:    false,
+			Metrics:      mergeMetrics(baseAttemptMetrics, map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(clientErr, false)}}),
+			ErrorCode:    clients.ErrorCode(clientErr),
+			ErrorMessage: err.Error(),
+		})
+		return clients.Response{}, fmt.Errorf("record parameter preprocessing: %w", err)
+	}
 	if err := s.store.AttachRateLimitResultToAttempt(ctx, attemptID, limitResult); err != nil {
 		clientErr := &clients.ClientError{Code: "runtime_error", Message: err.Error(), Retryable: false}
 		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
 			AttemptID:    attemptID,
 			Status:       "failed",
 			Retryable:    false,
-			Metrics:      mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(clientErr, false)}}),
+			Metrics:      mergeMetrics(baseAttemptMetrics, map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(clientErr, false)}}),
 			ErrorCode:    clients.ErrorCode(clientErr),
 			ErrorMessage: err.Error(),
 		})
@@ -371,7 +407,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 			AttemptID:    attemptID,
 			Status:       "failed",
 			Retryable:    false,
-			Metrics:      mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(err, false)}}),
+			Metrics:      mergeMetrics(baseAttemptMetrics, map[string]any{"error": err.Error(), "retryable": false, "trace": []any{failureTraceEntry(err, false)}}),
 			ErrorCode:    clients.ErrorCode(err),
 			ErrorMessage: err.Error(),
 		})
@@ -425,7 +461,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 				responseDurationMS = 0
 			}
 		}
-		metrics = mergeMetrics(attemptMetrics(candidate, attemptNo, simulated), metrics)
+		metrics = mergeMetrics(baseAttemptMetrics, metrics)
 		_ = s.store.FinishTaskAttempt(ctx, store.FinishTaskAttemptInput{
 			AttemptID:          attemptID,
 			Status:             "failed",
@@ -444,7 +480,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 	}
 	uploadedResult, err := s.uploadGeneratedAssets(ctx, response.Result)
 	if err != nil {
-		metrics := mergeMetrics(taskMetrics(task, user, body, candidate, response, simulated), map[string]any{
+		metrics := mergeMetrics(taskMetrics(task, user, body, candidate, response, simulated), parameterPreprocessingMetrics(preprocessing), map[string]any{
 			"error":     err.Error(),
 			"retryable": clients.IsRetryable(err),
 			"trace":     []any{failureTraceEntry(err, clients.IsRetryable(err))},
@@ -480,7 +516,7 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 		Status:             "succeeded",
 		RequestID:          response.RequestID,
 		Usage:              usageToMap(response.Usage),
-		Metrics:            taskMetrics(task, user, body, candidate, response, simulated),
+		Metrics:            mergeMetrics(taskMetrics(task, user, body, candidate, response, simulated), parameterPreprocessingMetrics(preprocessing)),
 		ResponseSnapshot:   response.Result,
 		ResponseStartedAt:  response.ResponseStartedAt,
 		ResponseFinishedAt: response.ResponseFinishedAt,
@@ -491,6 +527,25 @@ func (s *Service) runCandidate(ctx context.Context, task store.GatewayTask, user
 	return response, nil
 }

+func (s *Service) recordTaskParameterPreprocessing(ctx context.Context, taskID string, attemptID string, attemptNo int, candidate store.RuntimeModelCandidate, log parameterPreprocessingLog) error {
+	_, err := s.store.CreateTaskParamPreprocessingLog(ctx, store.CreateTaskParamPreprocessingLogInput{
+		TaskID:          taskID,
+		AttemptID:       attemptID,
+		AttemptNo:       attemptNo,
+		ModelType:       log.ModelType,
+		PlatformID:      candidate.PlatformID,
+		PlatformModelID: candidate.PlatformModelID,
+		ClientID:        candidate.ClientID,
+		Changed:         log.Changed,
+		ChangeCount:     len(log.Changes),
+		ActualInput:     log.Input,
+		ConvertedOutput: log.Output,
+		Changes:         log.Changes,
+		ModelSnapshot:   log.Model,
+	})
+	return err
+}
+
 func (s *Service) clientFor(candidate store.RuntimeModelCandidate, simulated bool) clients.Client {
 	if simulated {
 		return s.clients["simulation"]
@@ -505,8 +560,12 @@ func (s *Service) clientFor(candidate store.RuntimeModelCandidate, simulated boo
 	return s.clients["openai"]
 }

-func (s *Service) failTask(ctx context.Context, taskID string, code string, message string, simulated bool, cause error) (store.GatewayTask, error) {
+func (s *Service) failTask(ctx context.Context, taskID string, code string, message string, simulated bool, cause error, extraMetrics ...map[string]any) (store.GatewayTask, error) {
 	requestID, metrics, responseStartedAt, responseFinishedAt, responseDurationMS := failureMetrics(cause, simulated)
+	if len(extraMetrics) > 0 {
+		values := append([]map[string]any{metrics}, extraMetrics...)
+		metrics = mergeMetrics(values...)
+	}
 	metrics = s.withAttemptHistory(ctx, taskID, metrics)
 	failed, err := s.store.FinishTaskFailure(ctx, store.FinishTaskFailureInput{
 		TaskID:             taskID,
@@ -589,6 +648,9 @@ func (s *Service) emit(ctx context.Context, taskID string, eventType string, sta
 }

 func modelTypeFromKind(kind string, body map[string]any) string {
+	if requested := requestedModelTypeFromBody(body); requested != "" {
+		return requested
+	}
 	switch kind {
 	case "chat.completions", "responses":
 		return "text_generate"
@@ -598,6 +660,9 @@ func modelTypeFromKind(kind string, body map[string]any) string {
 		}
 		return "image_generate"
 	case "videos.generations":
+		if videoRequestHasVideoOrAudioReference(body) {
+			return "omni_video"
+		}
 		if videoRequestHasReferenceImage(body) {
 			return "image_to_video"
 		}
@@ -607,6 +672,25 @@ func modelTypeFromKind(kind string, body map[string]any) string {
 	}
 }

+func requestedModelTypeFromBody(body map[string]any) string {
+	for _, key := range []string{"modelType", "model_type", "capability", "capabilityType"} {
+		value := strings.TrimSpace(stringFromMap(body, key))
+		if isKnownModelType(value) {
+			return value
+		}
+	}
+	return ""
+}
+
+func isKnownModelType(value string) bool {
+	switch value {
+	case "text_generate", "image_generate", "image_edit", "video_generate", "image_to_video", "text_to_video", "video_edit", "omni_video", "omni":
+		return true
+	default:
+		return false
+	}
+}
+
 func videoRequestHasReferenceImage(body map[string]any) bool {
 	if body == nil {
 		return false
@@ -622,6 +706,23 @@ func videoRequestHasReferenceImage(body map[string]any) bool {
 	return false
 }

+func videoRequestHasVideoOrAudioReference(body map[string]any) bool {
+	if body == nil {
+		return false
+	}
+	for _, key := range []string{"video", "video_url", "videoUrl", "reference_video", "referenceVideo", "audio_url", "audioUrl", "reference_audio", "referenceAudio"} {
+		if hasAnyString(body, key) {
+			return true
+		}
+	}
+	for _, item := range contentItems(body["content"]) {
+		if isVideoContent(item) || isAudioContent(item) {
+			return true
+		}
+	}
+	return false
+}
+
 func isTextGenerationKind(kind string) bool {
 	return kind == "chat.completions" || kind == "responses"
 }
@@ -692,10 +793,7 @@ func failoverTimeBudgetExceeded(start time.Time, maxDuration time.Duration) bool
 }

 func normalizeRequest(kind string, body map[string]any) map[string]any {
-	out := map[string]any{}
-	for key, value := range body {
-		out[key] = value
-	}
+	out := cloneMap(body)
 	if kind == "responses" && out["messages"] == nil && out["input"] != nil {
 		out["messages"] = []any{map[string]any{"role": "user", "content": out["input"]}}
 	}