fix(errors): 区分平台限流并保留上游状态码

原因：公开错误层将平台并发限流误标为上游限流，并把多种上游 4xx 统一压成 400，影响定位和客户端处理。

影响：新增公开错误 source，平台限流使用 gateway_rate_limited，上游请求按安全分类返回对应状态；数据库与管理端继续保留原始错误码、消息和状态用于审计。

验证：Go 全量测试、pnpm test、pnpm lint、pnpm build、pnpm openapi、gofmt 和 diff 检查均通过。

apps/api/internal/httpapi/compat_protocol.go

@@ -109,7 +109,9 @@ func writeProtocolError(w http.ResponseWriter, protocol string, status int, mess
 	case clients.ProtocolVolcesContents:
 		writeVolcesPublicError(w, standard)
 	case clients.ProtocolKlingV1Omni, clients.ProtocolKlingV2Omni:
-		writeKelingCompatError(w, "", newKelingCompatError(status, kelingCompatBusinessCode(code, message), message))
+		compatErr := newKelingCompatError(status, kelingCompatBusinessCode(code, message), message)
+		compatErr.PublicError = &standard
+		writeKelingCompatError(w, "", compatErr)
 	default:
 		writeOpenAIError(w, status, message, details, code)
 	}

apps/api/internal/httpapi/compat_protocol_test.go

@@ -84,6 +84,15 @@ func TestProtocolErrorsUseCompatibleShapesWithStandardPublicErrors(t *testing.T)
 				assertNoKeys(t, errorBody, "status", "taskId", "gateway_status")
 			},
 		},
+		{
+			name: "keling", protocol: clients.ProtocolKlingV1Omni, status: http.StatusTooManyRequests,
+			assertBody: func(t *testing.T, body map[string]any) {
+				errorBody := requireObject(t, body["error"])
+				if errorBody["code"] != "gateway_rate_limited" || errorBody["source"] != "gateway" || errorBody["httpStatus"] != float64(http.StatusTooManyRequests) {
+					t.Fatalf("unexpected Kling-compatible error: %+v", body)
+				}
+			},
+		},
 	}
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
@@ -93,6 +102,8 @@ func TestProtocolErrorsUseCompatibleShapesWithStandardPublicErrors(t *testing.T)
 				code = "rate_limit"
 			} else if test.name == "volces" {
 				code = "upstream_submission_unknown"
+			} else if test.name == "keling" {
+				code = "gateway_rate_limited"
 			}
 			writeProtocolError(recorder, test.protocol, test.status, "failed", nil, code)
 			if recorder.Code != test.status {

apps/api/internal/httpapi/core_flow_integration_test.go

@@ -1233,8 +1233,8 @@ WHERE reference_type = 'gateway_task'
 		"simulationDurationMs": 5,
 		"messages":             []map[string]any{{"role": "user", "content": "second"}},
 	}, "rate-limit-second-"+suffixText, http.StatusTooManyRequests, nil, &rateLimitTaskTwo.Task)
-	if rateLimitTaskTwo.Task.Status != "failed" || rateLimitTaskTwo.Task.ErrorCode != "rate_limit" {
-		t.Fatalf("runtime policy rate limit should fail second task with rate_limit: %+v", rateLimitTaskTwo.Task)
+	if rateLimitTaskTwo.Task.Status != "failed" || rateLimitTaskTwo.Task.ErrorCode != "gateway_rate_limited" {
+		t.Fatalf("runtime policy rate limit should fail second task with gateway_rate_limited: %+v", rateLimitTaskTwo.Task)
 	}
 	var asyncRateLimitTask struct {
 		TaskID string `json:"taskId"`

apps/api/internal/httpapi/easyai_compat.go

@@ -68,6 +68,7 @@ func writeEasyAIAsyncError(w http.ResponseWriter, status int, message string, de
 		"status":     status,
 		"code":       code,
 		"category":   standard.Category,
+		"source":     standard.Source,
 		"httpStatus": standard.HTTPStatus,
 		"retryable":  standard.Retryable,
 		"action":     standard.Action,

apps/api/internal/httpapi/gemini_compat.go

@@ -235,7 +235,7 @@ func (s *Server) geminiGenerateContent(w http.ResponseWriter, r *http.Request) {
 		}
 		status := http.StatusConflict
 		if task.Status == "failed" || task.Status == "cancelled" {
-			status = storedTaskErrorStatus(task.ErrorCode)
+			status = taskErrorHTTPStatus(task)
 		}
 		writeGeminiTaskError(status, firstNonEmpty(task.ErrorMessage, task.Error, task.Message, "task is not complete"), nil, firstNonEmpty(task.ErrorCode, task.Status))
 		return
@@ -263,7 +263,7 @@ func (s *Server) geminiGenerateContent(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	if task.Status != "succeeded" {
-		status := storedTaskErrorStatus(task.ErrorCode)
+		status := taskErrorHTTPStatus(task)
 		writeGeminiTaskError(status, firstNonEmpty(task.ErrorMessage, task.Error, task.Message, "task failed"), nil, firstNonEmpty(task.ErrorCode, task.Status))
 		return
 	}

apps/api/internal/httpapi/handlers.go

@@ -1286,7 +1286,7 @@ func (s *Server) createTask(kind string, compatible bool) http.Handler {
 				}
 				status := http.StatusConflict
 				if task.Status == "failed" || task.Status == "cancelled" {
-					status = storedTaskErrorStatus(task.ErrorCode)
+					status = taskErrorHTTPStatus(task)
 				}
 				writeTaskError(status, firstNonEmpty(task.ErrorMessage, task.Error, task.Message, "task is not complete"), nil, firstNonEmpty(task.ErrorCode, task.Status))
 				return
@@ -1706,6 +1706,9 @@ func statusFromRunError(err error) int {
 	case errors.Is(err, store.ErrInsufficientWalletBalance):
 		return http.StatusPaymentRequired
 	default:
+		if status := clients.ErrorResponseMetadata(err).StatusCode; status >= 400 && status <= 599 {
+			return status
+		}
 		return http.StatusBadGateway
 	}
 }

apps/api/internal/httpapi/public_errors.go

@@ -10,8 +10,8 @@ import (
 
 func publicGatewayTask(task store.GatewayTask) store.GatewayTask {
 	if task.Status == "failed" || task.Status == "cancelled" || task.ErrorCode != "" || task.ErrorMessage != "" {
-		value := publicerror.FromFields(task.ErrorCode, firstNonEmpty(task.ErrorMessage, task.Error), storedTaskErrorStatus(task.ErrorCode), false)
-		if task.PublicError != nil && task.PublicError.Code != "" {
+		value := publicerror.FromFields(task.ErrorCode, firstNonEmpty(task.ErrorMessage, task.Error), taskErrorHTTPStatus(task), false)
+		if task.PublicError != nil && task.PublicError.Code != "" && task.PublicError.Source != "" {
 			value = *task.PublicError
 		}
 		value = publicerror.WithIDs(value, task.RequestID, task.ID)
@@ -21,6 +21,7 @@ func publicGatewayTask(task store.GatewayTask) store.GatewayTask {
 		task.ErrorMessage = value.Message
 		task.Error = value.Message
 	}
+	task.Attempts = append([]store.TaskAttempt(nil), task.Attempts...)
 	for index := range task.Attempts {
 		attempt := &task.Attempts[index]
 		if attempt.ErrorCode == "" && attempt.ErrorMessage == "" {
@@ -31,7 +32,7 @@ func publicGatewayTask(task store.GatewayTask) store.GatewayTask {
 			status = http.StatusBadGateway
 		}
 		value := publicerror.FromFields(attempt.ErrorCode, attempt.ErrorMessage, status, attempt.Retryable)
-		if attempt.PublicError != nil && attempt.PublicError.Code != "" {
+		if attempt.PublicError != nil && attempt.PublicError.Code != "" && attempt.PublicError.Source != "" {
 			value = *attempt.PublicError
 		}
 		value = publicerror.WithIDs(value, attempt.RequestID, task.ID)
@@ -43,16 +44,48 @@ func publicGatewayTask(task store.GatewayTask) store.GatewayTask {
 }
 
 func publicTaskError(task store.GatewayTask) publicerror.Error {
-	if task.PublicError != nil && task.PublicError.Code != "" {
+	if task.PublicError != nil && task.PublicError.Code != "" && task.PublicError.Source != "" {
 		return publicerror.WithIDs(*task.PublicError, task.RequestID, task.ID)
 	}
-	status := storedTaskErrorStatus(task.ErrorCode)
+	status := taskErrorHTTPStatus(task)
 	if status <= 0 {
 		status = http.StatusBadGateway
 	}
 	return publicerror.WithIDs(publicerror.FromFields(task.ErrorCode, firstNonEmpty(task.ErrorMessage, task.Error, task.Message), status, false), task.RequestID, task.ID)
 }
 
+func taskErrorHTTPStatus(task store.GatewayTask) int {
+	if task.PublicError != nil && task.PublicError.Source != "" && task.PublicError.HTTPStatus >= 400 {
+		return task.PublicError.HTTPStatus
+	}
+	for index := len(task.Attempts) - 1; index >= 0; index-- {
+		if status := task.Attempts[index].StatusCode; status >= 400 && status <= 599 {
+			return status
+		}
+	}
+	if status := publicErrorInt(task.Metrics["statusCode"]); status >= 400 && status <= 599 {
+		return status
+	}
+	return storedTaskErrorStatus(task.ErrorCode)
+}
+
+func publicErrorInt(value any) int {
+	switch typed := value.(type) {
+	case int:
+		return typed
+	case int32:
+		return int(typed)
+	case int64:
+		return int(typed)
+	case float32:
+		return int(typed)
+	case float64:
+		return int(typed)
+	default:
+		return 0
+	}
+}
+
 func publicTaskList(items []store.GatewayTask) []store.GatewayTask {
 	out := make([]store.GatewayTask, len(items))
 	for index, task := range items {
@@ -63,7 +96,7 @@ func publicTaskList(items []store.GatewayTask) []store.GatewayTask {
 
 func publicErrorMap(value publicerror.Error) map[string]any {
 	out := map[string]any{
-		"code": value.Code, "message": value.Message, "category": value.Category,
+		"code": value.Code, "message": value.Message, "category": value.Category, "source": value.Source,
 		"httpStatus": value.HTTPStatus, "retryable": value.Retryable, "action": value.Action,
 		"version": value.Version,
 	}

apps/api/internal/httpapi/public_errors_test.go

@@ -0,0 +1,96 @@
+package httpapi
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/publicerror"
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
+)
+
+func TestPublicHTTPErrorReportsSourceAndPreservesSafeUpstreamStatus(t *testing.T) {
+	upstream := httptest.NewRecorder()
+	writeProtocolError(upstream, clients.ProtocolOpenAIImages, http.StatusNotFound, "404 page not found at private upstream route", nil, "http_404")
+	if upstream.Code != http.StatusNotFound {
+		t.Fatalf("upstream status = %d, want 404; body=%s", upstream.Code, upstream.Body.String())
+	}
+	var upstreamBody map[string]any
+	if err := json.Unmarshal(upstream.Body.Bytes(), &upstreamBody); err != nil {
+		t.Fatal(err)
+	}
+	upstreamError := requireObject(t, upstreamBody["error"])
+	upstreamDetails := requireObject(t, upstreamError["details"])
+	upstreamPublic := requireObject(t, upstreamDetails["publicError"])
+	if upstreamError["code"] != "upstream_not_found" || upstreamPublic["source"] != "upstream" {
+		t.Fatalf("unexpected upstream response: %+v", upstreamBody)
+	}
+	if strings.Contains(upstream.Body.String(), "private upstream route") {
+		t.Fatalf("upstream response leaked raw message: %s", upstream.Body.String())
+	}
+
+	gateway := httptest.NewRecorder()
+	writeError(gateway, http.StatusTooManyRequests, "concurrency limit is saturated and queueing is disabled", "gateway_rate_limited")
+	if gateway.Code != http.StatusTooManyRequests {
+		t.Fatalf("gateway status = %d, want 429; body=%s", gateway.Code, gateway.Body.String())
+	}
+	var gatewayBody map[string]any
+	if err := json.Unmarshal(gateway.Body.Bytes(), &gatewayBody); err != nil {
+		t.Fatal(err)
+	}
+	gatewayError := requireObject(t, gatewayBody["error"])
+	if gatewayError["code"] != "gateway_rate_limited" || gatewayError["source"] != "gateway" {
+		t.Fatalf("unexpected gateway response: %+v", gatewayBody)
+	}
+}
+
+func TestPublicGatewayTaskSanitizesCopyAndKeepsRawAuditFields(t *testing.T) {
+	rawMessage := `404 page not found: {"privateProject":"secret"}`
+	task := store.GatewayTask{
+		ID:           "task-1",
+		Status:       "failed",
+		ErrorCode:    "http_404",
+		ErrorMessage: rawMessage,
+		Attempts: []store.TaskAttempt{{
+			AttemptNo:    1,
+			Status:       "failed",
+			StatusCode:   http.StatusNotFound,
+			ErrorCode:    "http_404",
+			ErrorMessage: rawMessage,
+		}},
+	}
+
+	public := publicGatewayTask(task)
+	if public.ErrorCode != "upstream_not_found" || public.ErrorMessage == rawMessage || public.PublicError == nil || public.PublicError.Source != "upstream" {
+		t.Fatalf("unexpected public task error: %+v", public)
+	}
+	if strings.Contains(public.ErrorMessage, "secret") || public.Attempts[0].ErrorMessage == rawMessage {
+		t.Fatalf("public task leaked upstream details: %+v", public)
+	}
+	if task.ErrorCode != "http_404" || task.ErrorMessage != rawMessage || task.Attempts[0].ErrorCode != "http_404" || task.Attempts[0].ErrorMessage != rawMessage {
+		t.Fatalf("public conversion mutated raw audit fields: %+v", task)
+	}
+}
+
+func TestTaskErrorHTTPStatusRebuildsLegacySnapshotFromAttempt(t *testing.T) {
+	legacy := publicerror.Error{
+		Code:       "upstream_request_rejected",
+		HTTPStatus: http.StatusBadRequest,
+		Version:    "v1",
+	}
+	task := store.GatewayTask{
+		ErrorCode:   "http_422",
+		PublicError: &legacy,
+		Attempts:    []store.TaskAttempt{{StatusCode: http.StatusUnprocessableEntity}},
+	}
+	if got := taskErrorHTTPStatus(task); got != http.StatusUnprocessableEntity {
+		t.Fatalf("taskErrorHTTPStatus = %d, want %d", got, http.StatusUnprocessableEntity)
+	}
+	public := publicTaskError(task)
+	if public.Code != "upstream_unprocessable_request" || public.HTTPStatus != http.StatusUnprocessableEntity || public.Source != "upstream" {
+		t.Fatalf("legacy public snapshot was not rebuilt: %+v", public)
+	}
+}

apps/api/internal/httpapi/rate_limit_error_detail_test.go

@@ -1,12 +1,14 @@
 package httpapi
 
 import (
+	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/easyai/easyai-ai-gateway/apps/api/internal/clients"
 	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
 )
 
@@ -66,7 +68,7 @@ func TestQueueFullAndTimeoutPublicContracts(t *testing.T) {
 	}
 	recorder := httptest.NewRecorder()
 	applyRunErrorHeaders(recorder, queueErr)
-	if statusFromRunError(queueErr) != http.StatusTooManyRequests || runErrorCode(queueErr) != "rate_limit" {
+	if statusFromRunError(queueErr) != http.StatusTooManyRequests || runErrorCode(queueErr) != "gateway_rate_limited" {
 		t.Fatalf("queue full contract status=%d code=%s", statusFromRunError(queueErr), runErrorCode(queueErr))
 	}
 	if recorder.Header().Get("Retry-After") != "2" {
@@ -84,6 +86,15 @@ func TestQueueFullAndTimeoutPublicContracts(t *testing.T) {
 	}
 }
 
+func TestStatusFromRunErrorPreservesUpstreamHTTPStatus(t *testing.T) {
+	for _, status := range []int{http.StatusBadRequest, http.StatusNotFound, http.StatusMethodNotAllowed, http.StatusConflict, http.StatusRequestEntityTooLarge, http.StatusUnsupportedMediaType, http.StatusUnprocessableEntity, http.StatusTooManyRequests} {
+		err := &clients.ClientError{Code: fmt.Sprintf("http_%d", status), Message: "raw upstream error", StatusCode: status}
+		if got := statusFromRunError(err); got != status {
+			t.Fatalf("statusFromRunError(%d) = %d", status, got)
+		}
+	}
+}
+
 func TestRunErrorMessageIncludesRateLimitSummary(t *testing.T) {
 	message := runErrorMessage(&store.RateLimitExceededError{
 		ScopeType:     "user_group",

apps/api/internal/httpapi/response.go

@@ -33,6 +33,7 @@ func writeErrorWithDetails(w http.ResponseWriter, status int, message string, de
 		"status":     status,
 		"code":       standard.Code,
 		"category":   standard.Category,
+		"source":     standard.Source,
 		"httpStatus": standard.HTTPStatus,
 		"retryable":  standard.Retryable,
 		"action":     standard.Action,

apps/api/internal/httpapi/task_idempotency.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"net/http"
+	"strconv"
 	"strings"
 
 	"github.com/easyai/easyai-ai-gateway/apps/api/internal/store"
@@ -52,7 +53,7 @@ func writeIdempotentTaskReplay(w http.ResponseWriter, task store.GatewayTask, co
 		writeJSON(w, http.StatusOK, task.Result)
 		return
 	}
-	status := storedTaskErrorStatus(task.ErrorCode)
+	status := taskErrorHTTPStatus(task)
 	message := strings.TrimSpace(task.ErrorMessage)
 	if message == "" {
 		message = strings.TrimSpace(task.Error)
@@ -81,9 +82,14 @@ func storedTaskErrorStatus(code string) int {
 		return http.StatusBadRequest
 	case "no_model_candidate", "cloned_voice_not_found":
 		return http.StatusNotFound
-	case "rate_limit", "platform_cooling_down", "model_cooling_down":
+	case "gateway_rate_limited", "rate_limit", "platform_cooling_down", "model_cooling_down":
 		return http.StatusTooManyRequests
 	default:
+		if strings.HasPrefix(strings.TrimSpace(code), "http_") {
+			if status, err := strconv.Atoi(strings.TrimPrefix(strings.TrimSpace(code), "http_")); err == nil && status >= 400 && status <= 599 {
+				return status
+			}
+		}
 		return http.StatusBadGateway
 	}
 }