Merge pull request #4485 from Sub2API-Devs/dev

feat(security-audit): 新增 OpenAI 兼容提示词审计能力与安全审计控制台

backend/internal/handler/batch_image_handler.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"encoding/json"
 	"errors"
 	"io"
 	"net/http"
@@ -20,6 +21,7 @@ type BatchImageHandler struct {
 	service  *service.BatchImagePublicService
 	download *service.BatchImageDownloadService
 	cleanup  *service.BatchImageCleanupService
+	openAI   *OpenAIGatewayHandler
 }
 
 func NewBatchImageHandler(service *service.BatchImagePublicService, download *service.BatchImageDownloadService, cleanup *service.BatchImageCleanupService) *BatchImageHandler {
@@ -37,6 +39,9 @@ func (h *BatchImageHandler) Submit(c *gin.Context) {
 		batchImageError(c, infraerrors.New(http.StatusUnauthorized, "API_KEY_REQUIRED", "API key is required"))
 		return
 	}
+	if !h.checkSecurityAuditBeforeSubmit(c, &req) {
+		return
+	}
 	got, err := h.service.Submit(c.Request.Context(), owner, req, c.GetHeader("Idempotency-Key"))
 	if err != nil {
 		batchImageError(c, err)
@@ -45,6 +50,44 @@ func (h *BatchImageHandler) Submit(c *gin.Context) {
 	c.JSON(http.StatusOK, got)
 }
 
+func (h *BatchImageHandler) checkSecurityAuditBeforeSubmit(c *gin.Context, req *service.BatchImageSubmitRequest) bool {
+	if h == nil || h.openAI == nil || req == nil {
+		return true
+	}
+	apiKey, ok := middleware.GetAPIKeyFromContext(c)
+	if !ok || apiKey == nil {
+		batchImageError(c, infraerrors.New(http.StatusUnauthorized, "API_KEY_REQUIRED", "API key is required"))
+		return false
+	}
+	subject, ok := middleware.GetAuthSubjectFromContext(c)
+	if !ok {
+		batchImageError(c, infraerrors.New(http.StatusInternalServerError, "USER_CONTEXT_REQUIRED", "User context not found"))
+		return false
+	}
+	items := make([]map[string]string, 0, len(req.Items))
+	for _, item := range req.Items {
+		if prompt := strings.TrimSpace(item.Prompt); prompt != "" {
+			items = append(items, map[string]string{"prompt": prompt})
+		}
+	}
+	if len(items) == 0 {
+		return true
+	}
+	body, err := json.Marshal(map[string]any{"request": map[string]any{"items": items}})
+	if err != nil {
+		batchImageError(c, infraerrors.New(http.StatusBadRequest, "INVALID_BATCH_PROMPT", "batch prompts are invalid"))
+		return false
+	}
+	reqLog := requestLogger(c, "handler.batch_image.security_audit",
+		zap.Int64("user_id", subject.UserID), zap.Int64("api_key_id", apiKey.ID), zap.String("model", req.Model))
+	decision := h.openAI.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIImages, req.Model, body)
+	if decision != nil && !decision.AllowNextStage {
+		h.openAI.openAISecurityAuditError(c, decision)
+		return false
+	}
+	return true
+}
+
 func (h *BatchImageHandler) Get(c *gin.Context) {
 	owner, ok := batchImageOwnerFromContext(c)
 	if !ok {

backend/internal/handler/content_moderation_helper.go

@@ -12,13 +12,6 @@ import (
 	"go.uber.org/zap"
 )
 
-func (h *GatewayHandler) checkContentModeration(c *gin.Context, reqLog *zap.Logger, apiKey *service.APIKey, subject middleware2.AuthSubject, protocol string, model string, body []byte) *service.ContentModerationDecision {
-	if h == nil || h.contentModerationService == nil {
-		return nil
-	}
-	return runContentModeration(c, reqLog, h.contentModerationService, apiKey, subject, protocol, model, body)
-}
-
 func contentModerationStatus(decision *service.ContentModerationDecision) int {
 	if decision == nil || decision.StatusCode < 400 || decision.StatusCode > 599 {
 		return http.StatusForbidden
@@ -30,13 +23,6 @@ func contentModerationErrorCode(decision *service.ContentModerationDecision) str
 	return "content_policy_violation"
 }
 
-func (h *OpenAIGatewayHandler) checkContentModeration(c *gin.Context, reqLog *zap.Logger, apiKey *service.APIKey, subject middleware2.AuthSubject, protocol string, model string, body []byte) *service.ContentModerationDecision {
-	if h == nil || h.contentModerationService == nil {
-		return nil
-	}
-	return runContentModeration(c, reqLog, h.contentModerationService, apiKey, subject, protocol, model, body)
-}
-
 func runContentModeration(c *gin.Context, reqLog *zap.Logger, svc *service.ContentModerationService, apiKey *service.APIKey, subject middleware2.AuthSubject, protocol string, model string, body []byte) *service.ContentModerationDecision {
 	if svc == nil || c == nil || c.Request == nil {
 		return nil

backend/internal/handler/gateway_handler.go

@@ -25,6 +25,7 @@ import (
 	"github.com/Wei-Shaw/sub2api/internal/pkg/openai"
 	"github.com/Wei-Shaw/sub2api/internal/pkg/timezone"
 	"github.com/Wei-Shaw/sub2api/internal/pkg/xai"
+	"github.com/Wei-Shaw/sub2api/internal/securityaudit"
 	middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
 	"github.com/Wei-Shaw/sub2api/internal/service"
 
@@ -49,6 +50,7 @@ type GatewayHandler struct {
 	usageRecordWorkerPool     *service.UsageRecordWorkerPool
 	errorPassthroughService   *service.ErrorPassthroughService
 	contentModerationService  *service.ContentModerationService
+	securityAuditCoordinator  *securityaudit.Coordinator
 	concurrencyHelper         *ConcurrencyHelper
 	userMsgQueueHelper        *UserMsgQueueHelper
 	maxAccountSwitches        int
@@ -199,8 +201,8 @@ func (h *GatewayHandler) Messages(c *gin.Context) {
 		return
 	}
 
-	if decision := h.checkContentModeration(c, reqLog, apiKey, subject, service.ContentModerationProtocolAnthropicMessages, reqModel, body); decision != nil && decision.Blocked {
-		h.errorResponse(c, contentModerationStatus(decision), contentModerationErrorCode(decision), decision.Message)
+	if decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolAnthropicMessages, reqModel, body); decision != nil && !decision.AllowNextStage {
+		h.anthropicSecurityAuditError(c, decision)
 		return
 	}
 

backend/internal/handler/gateway_handler_chat_completions.go

@@ -99,8 +99,8 @@ func (h *GatewayHandler) ChatCompletions(c *gin.Context) {
 		return
 	}
 
-	if decision := h.checkContentModeration(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIChat, reqModel, body); decision != nil && decision.Blocked {
-		h.chatCompletionsErrorResponse(c, contentModerationStatus(decision), contentModerationErrorCode(decision), decision.Message)
+	if decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIChat, reqModel, body); decision != nil && !decision.AllowNextStage {
+		h.openAISecurityAuditError(c, decision)
 		return
 	}
 

backend/internal/handler/gateway_handler_responses.go

@@ -104,8 +104,8 @@ func (h *GatewayHandler) Responses(c *gin.Context) {
 		return
 	}
 
-	if decision := h.checkContentModeration(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIResponses, reqModel, body); decision != nil && decision.Blocked {
-		h.responsesErrorResponse(c, contentModerationStatus(decision), contentModerationErrorCode(decision), decision.Message)
+	if decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIResponses, reqModel, body); decision != nil && !decision.AllowNextStage {
+		h.responsesSecurityAuditError(c, decision)
 		return
 	}
 

backend/internal/handler/gemini_v1beta_handler.go

@@ -187,8 +187,8 @@ func (h *GatewayHandler) GeminiV1BetaModels(c *gin.Context) {
 	setOpsRequestContext(c, modelName, stream)
 	setOpsEndpointContext(c, "", int16(service.RequestTypeFromLegacy(stream, false)))
 
-	if decision := h.checkContentModeration(c, reqLog, apiKey, authSubject, service.ContentModerationProtocolGemini, modelName, body); decision != nil && decision.Blocked {
-		googleError(c, contentModerationStatus(decision), decision.Message)
+	if decision := h.checkSecurityAudit(c, reqLog, apiKey, authSubject, service.ContentModerationProtocolGemini, modelName, body); decision != nil && !decision.AllowNextStage {
+		googleSecurityAuditError(c, decision)
 		return
 	}
 

backend/internal/handler/grok_media.go

@@ -114,9 +114,9 @@ func (h *OpenAIGatewayHandler) handleGrokMedia(c *gin.Context, endpoint service.
 			return
 		}
 		if moderationBody := requestInfo.ModerationBody(); len(moderationBody) > 0 {
-			decision := h.checkContentModeration(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIImages, requestModel, moderationBody)
-			if decision != nil && decision.Blocked {
-				h.errorResponse(c, contentModerationStatus(decision), contentModerationErrorCode(decision), decision.Message)
+			decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIImages, requestModel, moderationBody)
+			if decision != nil && !decision.AllowNextStage {
+				h.openAISecurityAuditError(c, decision)
 				return
 			}
 		}

backend/internal/handler/handler.go

@@ -2,6 +2,7 @@ package handler
 
 import (
 	"github.com/Wei-Shaw/sub2api/internal/handler/admin"
+	"github.com/Wei-Shaw/sub2api/internal/securityaudit"
 )
 
 // AdminHandlers contains all admin-related HTTP handlers
@@ -35,6 +36,7 @@ type AdminHandlers struct {
 	ChannelMonitor         *admin.ChannelMonitorHandler
 	ChannelMonitorTemplate *admin.ChannelMonitorRequestTemplateHandler
 	ContentModeration      *admin.ContentModerationHandler
+	PromptAudit            *securityaudit.PromptAdminHandler
 	Payment                *admin.PaymentHandler
 	Affiliate              *admin.AffiliateHandler
 	Compliance             *admin.ComplianceHandler

backend/internal/handler/image_task_handler.go

@@ -89,6 +89,9 @@ func (h *AsyncImageHandler) Submit(c *gin.Context) {
 		imageTaskJSONError(c, http.StatusBadRequest, "invalid_request_error", err.Error())
 		return
 	}
+	if !h.checkSecurityAuditBeforeSubmit(c, apiKey, platform, body) {
+		return
+	}
 
 	taskCtx, recorder, cancel := newAsyncImageContext(c, body, h.tasks.ExecutionTimeout())
 	task, err := h.tasks.Create(c.Request.Context(), service.ImageTaskOwner{UserID: apiKey.UserID, APIKeyID: apiKey.ID})
@@ -115,6 +118,42 @@ func (h *AsyncImageHandler) Submit(c *gin.Context) {
 	go h.run(task.ID, platform, taskCtx, recorder, cancel)
 }
 
+func (h *AsyncImageHandler) checkSecurityAuditBeforeSubmit(c *gin.Context, apiKey *service.APIKey, platform string, body []byte) bool {
+	if h == nil || h.openAI == nil {
+		return true
+	}
+	subject, ok := middleware2.GetAuthSubjectFromContext(c)
+	if !ok {
+		imageTaskJSONError(c, http.StatusInternalServerError, "api_error", "User context not found")
+		return false
+	}
+	model := ""
+	moderationBody := body
+	if platform == service.PlatformGrok {
+		parsed := service.ParseGrokMediaRequest(c.GetHeader("Content-Type"), body)
+		model, moderationBody = parsed.Model, parsed.ModerationBody()
+	} else if h.openAI.gatewayService != nil {
+		parsed, err := h.openAI.gatewayService.ParseOpenAIImagesRequest(c, body)
+		if err != nil {
+			imageTaskJSONError(c, http.StatusBadRequest, "invalid_request_error", err.Error())
+			return false
+		}
+		model, moderationBody = parsed.Model, parsed.ModerationBody()
+	}
+	if len(moderationBody) == 0 {
+		c.Set(securityAuditCompletedContextKey, true)
+		return true
+	}
+	reqLog := requestLogger(c, "handler.async_image.security_audit",
+		zap.Int64("user_id", subject.UserID), zap.Int64("api_key_id", apiKey.ID), zap.String("model", model))
+	decision := h.openAI.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIImages, model, moderationBody)
+	if decision != nil && !decision.AllowNextStage {
+		h.openAI.openAISecurityAuditError(c, decision)
+		return false
+	}
+	return true
+}
+
 func (h *AsyncImageHandler) Get(c *gin.Context) {
 	if !h.enabled() {
 		imageTaskJSONError(c, http.StatusNotFound, "not_found_error", "async image tasks are not enabled")

backend/internal/handler/openai_alpha_search.go

@@ -78,6 +78,10 @@ func (h *OpenAIGatewayHandler) AlphaSearch(c *gin.Context) {
 	reqLog = reqLog.With(zap.String("model", requestedModel))
 	setOpsRequestContext(c, requestedModel, false)
 	setOpsEndpointContext(c, "", int16(service.RequestTypeSync))
+	if decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, "openai_alpha_search", requestedModel, body); decision != nil && !decision.AllowNextStage {
+		h.openAISecurityAuditError(c, decision)
+		return
+	}
 
 	channelMapping, _ := h.gatewayService.ResolveChannelMappingAndRestrict(c.Request.Context(), apiKey.GroupID, requestedModel)
 	forwardBody := openAIModelMappedBody(body, channelMapping.Mapped, channelMapping.MappedModel, h.gatewayService.ReplaceModelInBody)

backend/internal/handler/openai_chat_completions.go

@@ -90,8 +90,8 @@ func (h *OpenAIGatewayHandler) ChatCompletions(c *gin.Context) {
 	setOpsRequestContext(c, reqModel, reqStream)
 	setOpsEndpointContext(c, "", int16(service.RequestTypeFromLegacy(reqStream, false)))
 
-	if decision := h.checkContentModeration(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIChat, reqModel, body); decision != nil && decision.Blocked {
-		h.errorResponse(c, contentModerationStatus(decision), contentModerationErrorCode(decision), decision.Message)
+	if decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIChat, reqModel, body); decision != nil && !decision.AllowNextStage {
+		h.openAISecurityAuditError(c, decision)
 		return
 	}
 	if h.rejectIfCyberSessionBlocked(c, apiKey, body, reqModel, cyberBlockFormatChat) {

backend/internal/handler/openai_embeddings.go

@@ -74,6 +74,10 @@ func (h *OpenAIGatewayHandler) Embeddings(c *gin.Context) {
 	reqLog = reqLog.With(zap.String("model", reqModel))
 	setOpsRequestContext(c, reqModel, false)
 	setOpsEndpointContext(c, "", int16(service.RequestTypeSync))
+	if decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, "openai_embeddings", reqModel, body); decision != nil && !decision.AllowNextStage {
+		h.openAISecurityAuditError(c, decision)
+		return
+	}
 
 	channelMapping, _ := h.gatewayService.ResolveChannelMappingAndRestrict(c.Request.Context(), apiKey.GroupID, reqModel)
 

backend/internal/handler/openai_gateway_handler.go

@@ -15,6 +15,7 @@ import (
 	"github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey"
 	"github.com/Wei-Shaw/sub2api/internal/pkg/ip"
 	"github.com/Wei-Shaw/sub2api/internal/pkg/logger"
+	"github.com/Wei-Shaw/sub2api/internal/securityaudit"
 	middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
 	"github.com/Wei-Shaw/sub2api/internal/service"
 
@@ -33,6 +34,7 @@ type OpenAIGatewayHandler struct {
 	usageRecordWorkerPool    *service.UsageRecordWorkerPool
 	errorPassthroughService  *service.ErrorPassthroughService
 	contentModerationService *service.ContentModerationService
+	securityAuditCoordinator *securityaudit.Coordinator
 	opsService               *service.OpsService
 	concurrencyHelper        *ConcurrencyHelper
 	imageLimiter             *imageConcurrencyLimiter
@@ -271,8 +273,8 @@ func (h *OpenAIGatewayHandler) Responses(c *gin.Context) {
 	setOpsRequestContext(c, reqModel, reqStream)
 	setOpsEndpointContext(c, "", int16(service.RequestTypeFromLegacy(reqStream, false)))
 
-	if decision := h.checkContentModeration(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIResponses, reqModel, body); decision != nil && decision.Blocked {
-		h.errorResponse(c, contentModerationStatus(decision), contentModerationErrorCode(decision), decision.Message)
+	if decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIResponses, reqModel, body); decision != nil && !decision.AllowNextStage {
+		h.openAISecurityAuditError(c, decision)
 		return
 	}
 
@@ -854,8 +856,8 @@ func (h *OpenAIGatewayHandler) Messages(c *gin.Context) {
 	setOpsRequestContext(c, reqModel, reqStream)
 	setOpsEndpointContext(c, "", int16(service.RequestTypeFromLegacy(reqStream, false)))
 
-	if decision := h.checkContentModeration(c, reqLog, apiKey, subject, service.ContentModerationProtocolAnthropicMessages, reqModel, body); decision != nil && decision.Blocked {
-		h.anthropicErrorResponse(c, contentModerationStatus(decision), contentModerationErrorCode(decision), decision.Message)
+	if decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolAnthropicMessages, reqModel, body); decision != nil && !decision.AllowNextStage {
+		h.anthropicSecurityAuditError(c, decision)
 		return
 	}
 
@@ -1478,9 +1480,9 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
 	setOpsRequestContext(c, reqModel, true)
 	setOpsEndpointContext(c, "", int16(service.RequestTypeWSV2))
 
-	if decision := h.checkContentModeration(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIResponses, reqModel, firstMessage); decision != nil && decision.Blocked {
-		writeContentModerationWSError(ctx, wsConn, decision)
-		closeOpenAIClientWS(wsConn, coderws.StatusPolicyViolation, decision.Message)
+	if decision := h.checkSecurityAuditStage(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIResponses, reqModel, firstMessage, "first_turn"); decision != nil && !decision.AllowNextStage {
+		writeSecurityAuditWSError(ctx, wsConn, decision)
+		closeOpenAIClientWS(wsConn, securityAuditWSCloseStatus(decision), securityAuditWSCloseReason(decision))
 		return
 	}
 
@@ -1733,9 +1735,9 @@ func (h *OpenAIGatewayHandler) ResponsesWebSocket(c *gin.Context) {
 				if model == "" {
 					model = reqModel
 				}
-				if decision := h.checkContentModeration(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIResponses, model, payload); decision != nil && decision.Blocked {
-					writeContentModerationWSError(ctx, wsConn, decision)
-					return service.NewOpenAIWSClientCloseError(coderws.StatusPolicyViolation, decision.Message, nil)
+				if decision := h.checkSecurityAuditStage(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIResponses, model, payload, "subsequent_turn"); decision != nil && !decision.AllowNextStage {
+					writeSecurityAuditWSError(ctx, wsConn, decision)
+					return service.NewOpenAIWSClientCloseError(securityAuditWSCloseStatus(decision), securityAuditWSCloseReason(decision), nil)
 				}
 				return nil
 			},

backend/internal/handler/openai_images.go

@@ -86,8 +86,8 @@ func (h *OpenAIGatewayHandler) Images(c *gin.Context) {
 		h.errorResponse(c, http.StatusForbidden, "permission_error", service.ImageGenerationPermissionMessage())
 		return
 	}
-	if decision := h.checkContentModeration(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIImages, requestModel, parsed.ModerationBody()); decision != nil && decision.Blocked {
-		h.errorResponse(c, contentModerationStatus(decision), contentModerationErrorCode(decision), decision.Message)
+	if decision := h.checkSecurityAudit(c, reqLog, apiKey, subject, service.ContentModerationProtocolOpenAIImages, requestModel, parsed.ModerationBody()); decision != nil && !decision.AllowNextStage {
+		h.openAISecurityAuditError(c, decision)
 		return
 	}
 	imageReleaseFunc, acquired := h.acquireImageGenerationSlot(c, streamStarted)

backend/internal/handler/security_audit_errors.go

@@ -0,0 +1,187 @@
+package handler
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/Wei-Shaw/sub2api/internal/pkg/googleapi"
+	"github.com/Wei-Shaw/sub2api/internal/securityaudit"
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	coderws "github.com/coder/websocket"
+	"github.com/gin-gonic/gin"
+)
+
+func (h *OpenAIGatewayHandler) openAISecurityAuditError(c *gin.Context, decision *securityaudit.Decision) {
+	if decision == nil {
+		return
+	}
+	if decision.Legacy != nil && decision.Legacy.Blocked {
+		h.errorResponse(c, securityAuditStatus(decision), securityAuditErrorCode(decision), securityAuditMessage(decision))
+		return
+	}
+	errType := "api_error"
+	if decision.Kind == securityaudit.DecisionBlock {
+		errType = "permission_error"
+	}
+	c.JSON(securityAuditStatus(decision), gin.H{"error": gin.H{
+		"type": errType, "code": securityAuditErrorCode(decision), "message": securityAuditMessage(decision),
+	}})
+}
+
+func (h *GatewayHandler) openAISecurityAuditError(c *gin.Context, decision *securityaudit.Decision) {
+	if decision == nil {
+		return
+	}
+	if decision.Legacy != nil && decision.Legacy.Blocked {
+		h.chatCompletionsErrorResponse(c, securityAuditStatus(decision), securityAuditErrorCode(decision), securityAuditMessage(decision))
+		return
+	}
+	errType := "api_error"
+	if decision.Kind == securityaudit.DecisionBlock {
+		errType = "permission_error"
+	}
+	c.JSON(securityAuditStatus(decision), gin.H{"error": gin.H{
+		"type": errType, "code": securityAuditErrorCode(decision), "message": securityAuditMessage(decision),
+	}})
+}
+
+func (h *GatewayHandler) responsesSecurityAuditError(c *gin.Context, decision *securityaudit.Decision) {
+	if decision == nil {
+		return
+	}
+	if decision.Legacy != nil && decision.Legacy.Blocked {
+		h.responsesErrorResponse(c, securityAuditStatus(decision), securityAuditErrorCode(decision), securityAuditMessage(decision))
+		return
+	}
+	c.JSON(securityAuditStatus(decision), gin.H{"error": gin.H{
+		"type": "api_error", "code": securityAuditErrorCode(decision), "message": securityAuditMessage(decision),
+	}})
+}
+
+func (h *GatewayHandler) anthropicSecurityAuditError(c *gin.Context, decision *securityaudit.Decision) {
+	if decision == nil {
+		return
+	}
+	if decision.Legacy != nil && decision.Legacy.Blocked {
+		h.errorResponse(c, securityAuditStatus(decision), securityAuditErrorCode(decision), securityAuditMessage(decision))
+		return
+	}
+	errType := "api_error"
+	if decision.Kind == securityaudit.DecisionBlock {
+		errType = "permission_error"
+	}
+	c.JSON(securityAuditStatus(decision), gin.H{"type": "error", "error": gin.H{
+		"type": errType, "code": securityAuditErrorCode(decision), "message": securityAuditMessage(decision),
+	}})
+}
+
+func (h *OpenAIGatewayHandler) anthropicSecurityAuditError(c *gin.Context, decision *securityaudit.Decision) {
+	if decision == nil {
+		return
+	}
+	if decision.Legacy != nil && decision.Legacy.Blocked {
+		h.anthropicErrorResponse(c, securityAuditStatus(decision), securityAuditErrorCode(decision), securityAuditMessage(decision))
+		return
+	}
+	errType := "api_error"
+	if decision.Kind == securityaudit.DecisionBlock {
+		errType = "permission_error"
+	}
+	c.JSON(securityAuditStatus(decision), gin.H{"type": "error", "error": gin.H{
+		"type": errType, "code": securityAuditErrorCode(decision), "message": securityAuditMessage(decision),
+	}})
+}
+
+func googleSecurityAuditError(c *gin.Context, decision *securityaudit.Decision) {
+	if decision == nil {
+		return
+	}
+	if decision.Legacy != nil && decision.Legacy.Blocked {
+		googleError(c, securityAuditStatus(decision), securityAuditMessage(decision))
+		return
+	}
+	status := securityAuditStatus(decision)
+	googleStatus := googleapi.HTTPStatusToGoogleStatus(status)
+	if status == http.StatusServiceUnavailable {
+		googleStatus = "UNAVAILABLE"
+	}
+	requestID := ""
+	if c != nil && c.Request != nil {
+		requestID = contentModerationRequestID(c.Request.Context())
+	}
+	c.JSON(status, gin.H{"error": gin.H{
+		"code": status, "message": securityAuditMessage(decision), "status": googleStatus,
+		"details": []gin.H{{
+			"@type":  "type.googleapis.com/google.rpc.ErrorInfo",
+			"reason": securityAuditErrorCode(decision), "domain": "sub2api.securityaudit",
+			"metadata": gin.H{"request_id": requestID},
+		}},
+	}})
+}
+
+func writeSecurityAuditWSError(ctx context.Context, conn *coderws.Conn, decision *securityaudit.Decision) {
+	if conn == nil || decision == nil {
+		return
+	}
+	if decision.Legacy != nil && decision.Legacy.Blocked {
+		legacy := decision.Legacy
+		writeContentModerationWSError(ctx, conn, (legacyContentModerationDecision{legacy}).toService())
+		return
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	payload, err := json.Marshal(gin.H{
+		"event_id": "evt_prompt_guard_rejected", "type": "error",
+		"error": gin.H{"type": "invalid_request_error", "code": securityAuditErrorCode(decision), "message": securityAuditMessage(decision)},
+	})
+	if err != nil {
+		return
+	}
+	writeCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
+	defer cancel()
+	_ = conn.Write(writeCtx, coderws.MessageText, payload)
+}
+
+type legacyContentModerationDecision struct{ value *securityaudit.LegacyDecision }
+
+func (d legacyContentModerationDecision) toService() *service.ContentModerationDecision {
+	if d.value == nil {
+		return nil
+	}
+	return &service.ContentModerationDecision{Allowed: d.value.Allowed, Blocked: d.value.Blocked, Flagged: d.value.Flagged, Message: d.value.Message, StatusCode: d.value.StatusCode, Action: d.value.Action}
+}
+
+func securityAuditWSCloseStatus(decision *securityaudit.Decision) coderws.StatusCode {
+	if decision == nil {
+		return coderws.StatusInternalError
+	}
+	if decision.Legacy != nil && decision.Legacy.Blocked {
+		return coderws.StatusPolicyViolation
+	}
+	if decision.Kind == securityaudit.DecisionBlock {
+		return coderws.StatusCode(4403)
+	}
+	return coderws.StatusTryAgainLater
+}
+
+func securityAuditWSCloseReason(decision *securityaudit.Decision) string {
+	if decision == nil {
+		return securityaudit.ErrorCodeUnavailable
+	}
+	if decision.Legacy != nil && decision.Legacy.Blocked {
+		message := strings.TrimSpace(decision.Legacy.Message)
+		if message != "" {
+			return message
+		}
+		return "content_policy_violation"
+	}
+	code := securityAuditErrorCode(decision)
+	if code == "" {
+		return securityaudit.ErrorCodeUnavailable
+	}
+	return code
+}

backend/internal/handler/security_audit_errors_test.go

@@ -0,0 +1,159 @@
+package handler
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/pkg/ctxkey"
+	"github.com/Wei-Shaw/sub2api/internal/securityaudit"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+func promptGuardDecision(kind securityaudit.DecisionKind) *securityaudit.Decision {
+	decision := &securityaudit.Decision{Kind: kind, AllowNextStage: false}
+	switch kind {
+	case securityaudit.DecisionBlock:
+		decision.HTTPStatus = http.StatusForbidden
+		decision.ErrorCode = securityaudit.ErrorCodeBlocked
+		decision.ClientMessage = "提示词安全审计拒绝了该请求，请调整输入后重试"
+	case securityaudit.DecisionInvalid:
+		decision.HTTPStatus = http.StatusServiceUnavailable
+		decision.ErrorCode = securityaudit.ErrorCodeInvalidResponse
+		decision.ClientMessage = "提示词安全审计暂时不可用，请稍后重试"
+	default:
+		decision.HTTPStatus = http.StatusServiceUnavailable
+		decision.ErrorCode = securityaudit.ErrorCodeUnavailable
+		decision.ClientMessage = "提示词安全审计暂时不可用，请稍后重试"
+	}
+	return decision
+}
+
+func securityAuditErrorTestContext(t *testing.T) (*gin.Context, *httptest.ResponseRecorder) {
+	t.Helper()
+	recorder := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(recorder)
+	ctx := context.WithValue(context.Background(), ctxkey.RequestID, "request-error-golden")
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/test", nil).WithContext(ctx)
+	return c, recorder
+}
+
+func decodeErrorJSON(t *testing.T, recorder *httptest.ResponseRecorder) map[string]any {
+	t.Helper()
+	var payload map[string]any
+	require.NoError(t, json.Unmarshal(recorder.Body.Bytes(), &payload))
+	return payload
+}
+
+func requireObject(t *testing.T, value any) map[string]any {
+	t.Helper()
+	object, ok := value.(map[string]any)
+	require.True(t, ok)
+	return object
+}
+
+func requireArray(t *testing.T, value any) []any {
+	t.Helper()
+	array, ok := value.([]any)
+	require.True(t, ok)
+	return array
+}
+
+func TestPromptGuardOpenAIAndClaudeErrorEnvelopesGolden(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for _, kind := range []securityaudit.DecisionKind{securityaudit.DecisionBlock, securityaudit.DecisionUnavailable, securityaudit.DecisionInvalid} {
+		decision := promptGuardDecision(kind)
+		t.Run("openai_"+string(kind), func(t *testing.T) {
+			c, recorder := securityAuditErrorTestContext(t)
+			(&OpenAIGatewayHandler{}).openAISecurityAuditError(c, decision)
+			require.Equal(t, decision.HTTPStatus, recorder.Code)
+			payload := decodeErrorJSON(t, recorder)
+			errorObject := requireObject(t, payload["error"])
+			require.Equal(t, decision.ErrorCode, errorObject["code"])
+			if kind == securityaudit.DecisionBlock {
+				require.Equal(t, "permission_error", errorObject["type"])
+			} else {
+				require.Equal(t, "api_error", errorObject["type"])
+			}
+			require.NotContains(t, recorder.Body.String(), "raw prompt")
+			require.NotContains(t, recorder.Body.String(), "guard-one")
+		})
+
+		t.Run("responses_"+string(kind), func(t *testing.T) {
+			c, recorder := securityAuditErrorTestContext(t)
+			(&GatewayHandler{}).responsesSecurityAuditError(c, decision)
+			require.Equal(t, decision.HTTPStatus, recorder.Code)
+			errorObject := requireObject(t, decodeErrorJSON(t, recorder)["error"])
+			require.Equal(t, decision.ErrorCode, errorObject["code"])
+			require.Equal(t, "api_error", errorObject["type"])
+		})
+
+		t.Run("claude_"+string(kind), func(t *testing.T) {
+			c, recorder := securityAuditErrorTestContext(t)
+			(&GatewayHandler{}).anthropicSecurityAuditError(c, decision)
+			require.Equal(t, decision.HTTPStatus, recorder.Code)
+			payload := decodeErrorJSON(t, recorder)
+			require.Equal(t, "error", payload["type"])
+			errorObject := requireObject(t, payload["error"])
+			require.Equal(t, decision.ErrorCode, errorObject["code"])
+			if kind == securityaudit.DecisionBlock {
+				require.Equal(t, "permission_error", errorObject["type"])
+			} else {
+				require.Equal(t, "api_error", errorObject["type"])
+			}
+		})
+	}
+}
+
+func TestPromptGuardGeminiErrorEnvelopeGolden(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for _, kind := range []securityaudit.DecisionKind{securityaudit.DecisionBlock, securityaudit.DecisionUnavailable, securityaudit.DecisionInvalid} {
+		decision := promptGuardDecision(kind)
+		c, recorder := securityAuditErrorTestContext(t)
+		googleSecurityAuditError(c, decision)
+		require.Equal(t, decision.HTTPStatus, recorder.Code)
+		payload := decodeErrorJSON(t, recorder)
+		errorObject := requireObject(t, payload["error"])
+		require.Equal(t, float64(decision.HTTPStatus), errorObject["code"], "Gemini code must remain numeric")
+		if decision.HTTPStatus == http.StatusForbidden {
+			require.Equal(t, "PERMISSION_DENIED", errorObject["status"])
+		} else {
+			require.Equal(t, "UNAVAILABLE", errorObject["status"])
+		}
+		details := requireArray(t, errorObject["details"])
+		require.Len(t, details, 1)
+		errorInfo := requireObject(t, details[0])
+		require.Equal(t, "type.googleapis.com/google.rpc.ErrorInfo", errorInfo["@type"])
+		require.Equal(t, decision.ErrorCode, errorInfo["reason"])
+		require.Equal(t, "sub2api.securityaudit", errorInfo["domain"])
+		metadata := requireObject(t, errorInfo["metadata"])
+		require.Equal(t, map[string]any{"request_id": "request-error-golden"}, metadata)
+	}
+}
+
+func TestPromptGuardWebSocketCloseMappingGolden(t *testing.T) {
+	require.Equal(t, int64(4403), int64(securityAuditWSCloseStatus(promptGuardDecision(securityaudit.DecisionBlock))))
+	require.Equal(t, securityaudit.ErrorCodeBlocked, securityAuditWSCloseReason(promptGuardDecision(securityaudit.DecisionBlock)))
+	require.Equal(t, int64(1013), int64(securityAuditWSCloseStatus(promptGuardDecision(securityaudit.DecisionUnavailable))))
+	require.Equal(t, securityaudit.ErrorCodeUnavailable, securityAuditWSCloseReason(promptGuardDecision(securityaudit.DecisionUnavailable)))
+	require.Equal(t, int64(1013), int64(securityAuditWSCloseStatus(promptGuardDecision(securityaudit.DecisionInvalid))))
+	require.Equal(t, securityaudit.ErrorCodeInvalidResponse, securityAuditWSCloseReason(promptGuardDecision(securityaudit.DecisionInvalid)))
+}
+
+func TestLegacyModerationErrorKeepsExistingClientPriority(t *testing.T) {
+	legacy := &securityaudit.Decision{
+		Kind: securityaudit.DecisionBlock, HTTPStatus: http.StatusForbidden,
+		ErrorCode: "content_policy_violation", ClientMessage: "legacy exact message",
+		Legacy: &securityaudit.LegacyDecision{Blocked: true, StatusCode: http.StatusForbidden, ErrorCode: "content_policy_violation", Message: "legacy exact message"},
+		Prompt: &securityaudit.PromptDecision{Kind: securityaudit.DecisionBlock, ErrorCode: securityaudit.ErrorCodeBlocked},
+	}
+	c, recorder := securityAuditErrorTestContext(t)
+	(&GatewayHandler{}).openAISecurityAuditError(c, legacy)
+	require.Equal(t, http.StatusForbidden, recorder.Code)
+	require.Contains(t, recorder.Body.String(), "legacy exact message")
+	require.Contains(t, recorder.Body.String(), "content_policy_violation")
+	require.NotContains(t, recorder.Body.String(), securityaudit.ErrorCodeBlocked)
+}

backend/internal/handler/security_audit_helper.go

@@ -0,0 +1,153 @@
+package handler
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/Wei-Shaw/sub2api/internal/securityaudit"
+	middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/gin-gonic/gin"
+	"go.uber.org/zap"
+)
+
+const securityAuditCompletedContextKey = "sub2api.security_audit.completed"
+
+// cachesSecurityAuditCompletion reports whether a successful audit may be
+// reused for the rest of the gin request. WebSocket turns share one Context
+// across many response.create frames and must be audited independently.
+func cachesSecurityAuditCompletion(stage string) bool {
+	switch strings.TrimSpace(stage) {
+	case "", "http":
+		return true
+	default:
+		return false
+	}
+}
+
+func (h *GatewayHandler) checkSecurityAudit(c *gin.Context, reqLog *zap.Logger, apiKey *service.APIKey, subject middleware2.AuthSubject, protocol, model string, body []byte) *securityaudit.Decision {
+	if h == nil {
+		return nil
+	}
+	return runSecurityAudit(c, reqLog, h.securityAuditCoordinator, h.contentModerationService, apiKey, subject, protocol, model, body, "http")
+}
+
+func (h *OpenAIGatewayHandler) checkSecurityAudit(c *gin.Context, reqLog *zap.Logger, apiKey *service.APIKey, subject middleware2.AuthSubject, protocol, model string, body []byte) *securityaudit.Decision {
+	if h == nil {
+		return nil
+	}
+	return runSecurityAudit(c, reqLog, h.securityAuditCoordinator, h.contentModerationService, apiKey, subject, protocol, model, body, "http")
+}
+
+func (h *OpenAIGatewayHandler) checkSecurityAuditStage(c *gin.Context, reqLog *zap.Logger, apiKey *service.APIKey, subject middleware2.AuthSubject, protocol, model string, body []byte, stage string) *securityaudit.Decision {
+	if h == nil {
+		return nil
+	}
+	return runSecurityAudit(c, reqLog, h.securityAuditCoordinator, h.contentModerationService, apiKey, subject, protocol, model, body, stage)
+}
+
+func runSecurityAudit(c *gin.Context, reqLog *zap.Logger, coordinator *securityaudit.Coordinator, legacy *service.ContentModerationService, apiKey *service.APIKey, subject middleware2.AuthSubject, protocol, model string, body []byte, stage string) *securityaudit.Decision {
+	if c == nil || c.Request == nil {
+		return nil
+	}
+	cacheCompletion := cachesSecurityAuditCompletion(stage)
+	if cacheCompletion {
+		if completed, exists := c.Get(securityAuditCompletedContextKey); exists && completed == true {
+			return nil
+		}
+	}
+	if coordinator == nil {
+		legacyDecision := runContentModeration(c, reqLog, legacy, apiKey, subject, protocol, model, body)
+		if legacyDecision == nil {
+			return nil
+		}
+		decision := securityaudit.Decision{Kind: securityaudit.DecisionAllow, HTTPStatus: http.StatusOK, AllowNextStage: true}
+		decision.Legacy = &securityaudit.LegacyDecision{
+			Allowed: legacyDecision.Allowed, Blocked: legacyDecision.Blocked, Flagged: legacyDecision.Flagged,
+			Message: legacyDecision.Message, StatusCode: legacyDecision.StatusCode,
+			ErrorCode: "content_policy_violation", Action: legacyDecision.Action,
+		}
+		if legacyDecision.Blocked {
+			decision.Kind, decision.HTTPStatus, decision.ErrorCode, decision.ClientMessage, decision.AllowNextStage = securityaudit.DecisionBlock, contentModerationStatus(legacyDecision), "content_policy_violation", legacyDecision.Message, false
+		}
+		if decision.AllowNextStage && cacheCompletion {
+			c.Set(securityAuditCompletedContextKey, true)
+		}
+		return &decision
+	}
+	request := buildSecurityAuditRequest(c, apiKey, subject, protocol, model, body, stage)
+	if reqLog != nil {
+		reqLog.Info("security_audit.gateway_check_start",
+			zap.String("request_id", request.RequestID), zap.Int64("user_id", request.UserID),
+			zap.Int64("api_key_id", request.APIKeyID), zap.Int64p("group_id", request.GroupID),
+			zap.String("endpoint", request.Endpoint), zap.String("provider", request.Provider),
+			zap.String("protocol", request.Protocol), zap.String("model", request.Model), zap.String("stage", request.Stage),
+			zap.Int("body_bytes", len(body)))
+	}
+	decision := coordinator.Check(c.Request.Context(), request)
+	if decision.AllowNextStage && cacheCompletion {
+		c.Set(securityAuditCompletedContextKey, true)
+	}
+	if reqLog != nil {
+		reqLog.Info("security_audit.gateway_check_done",
+			zap.String("request_id", request.RequestID), zap.String("decision", string(decision.Kind)),
+			zap.String("error_code", decision.ErrorCode), zap.Bool("allow_next_stage", decision.AllowNextStage),
+			zap.String("stage", request.Stage))
+	}
+	return &decision
+}
+
+func buildSecurityAuditRequest(c *gin.Context, apiKey *service.APIKey, subject middleware2.AuthSubject, protocol, model string, body []byte, stage string) securityaudit.Request {
+	legacy := buildContentModerationInput(c, apiKey, subject, protocol, model, body)
+	request := securityaudit.Request{
+		RequestID: legacy.RequestID, UserID: legacy.UserID, UserEmail: legacy.UserEmail,
+		APIKeyID: legacy.APIKeyID, APIKeyName: legacy.APIKeyName, GroupID: cloneSecurityAuditGroupID(legacy.GroupID),
+		GroupName: legacy.GroupName, Provider: legacy.Provider, Endpoint: legacy.Endpoint,
+		Protocol: legacy.Protocol, Model: legacy.Model, Body: body, Stage: strings.TrimSpace(stage),
+	}
+	if apiKey != nil && apiKey.User != nil {
+		request.Username = apiKey.User.Username
+		if request.UserEmail == "" {
+			request.UserEmail = apiKey.User.Email
+		}
+	}
+	if request.Stage == "" {
+		request.Stage = "http"
+	}
+	return request
+}
+
+func securityAuditStatus(decision *securityaudit.Decision) int {
+	if decision == nil || decision.HTTPStatus < 400 || decision.HTTPStatus > 599 {
+		return http.StatusForbidden
+	}
+	return decision.HTTPStatus
+}
+
+func securityAuditErrorCode(decision *securityaudit.Decision) string {
+	if decision == nil || strings.TrimSpace(decision.ErrorCode) == "" {
+		return "content_policy_violation"
+	}
+	return decision.ErrorCode
+}
+
+func securityAuditMessage(decision *securityaudit.Decision) string {
+	if decision == nil {
+		return "Request blocked by content policy"
+	}
+	if decision.Legacy != nil && decision.Legacy.Blocked && strings.TrimSpace(decision.Legacy.Message) != "" {
+		return decision.Legacy.Message
+	}
+	if strings.TrimSpace(decision.ClientMessage) != "" {
+		return decision.ClientMessage
+	}
+	return "Request blocked by content policy"
+}
+
+func cloneSecurityAuditGroupID(value *int64) *int64 {
+	if value == nil {
+		return nil
+	}
+	cloned := *value
+	return &cloned
+}

backend/internal/handler/security_audit_helper_test.go

@@ -0,0 +1,63 @@
+package handler
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"sync/atomic"
+	"testing"
+
+	"github.com/Wei-Shaw/sub2api/internal/securityaudit"
+	middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCachesSecurityAuditCompletionSkipsWebSocketStages(t *testing.T) {
+	require.True(t, cachesSecurityAuditCompletion("http"))
+	require.True(t, cachesSecurityAuditCompletion(""))
+	require.False(t, cachesSecurityAuditCompletion("first_turn"))
+	require.False(t, cachesSecurityAuditCompletion("subsequent_turn"))
+}
+
+func TestRunSecurityAuditDoesNotSkipSubsequentWebSocketTurns(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	engine := &turnCountingEngine{mode: securityaudit.ModeAsync}
+	coordinator := securityaudit.NewCoordinator(nil, engine)
+
+	recorder := httptest.NewRecorder()
+	c, _ := gin.CreateTestContext(recorder)
+	c.Request = httptest.NewRequest(http.MethodPost, "/v1/responses", nil)
+
+	subject := middleware2.AuthSubject{UserID: 7, Concurrency: 1}
+	first := runSecurityAudit(c, nil, coordinator, nil, nil, subject, "openai_responses", "gpt-test",
+		[]byte(`{"type":"response.create","response":{"input":"benign"}}`), "first_turn")
+	require.NotNil(t, first)
+	require.True(t, first.AllowNextStage)
+	require.Equal(t, int64(1), engine.enqueues.Load())
+	_, cached := c.Get(securityAuditCompletedContextKey)
+	require.False(t, cached, "WebSocket stages must not set the HTTP completion cache")
+
+	// Even if an HTTP path previously cached completion on this Context, WS turns
+	// must still audit every response.create payload.
+	c.Set(securityAuditCompletedContextKey, true)
+
+	second := runSecurityAudit(c, nil, coordinator, nil, nil, subject, "openai_responses", "gpt-test",
+		[]byte(`{"type":"response.create","response":{"input":"malicious follow-up"}}`), "subsequent_turn")
+	require.NotNil(t, second)
+	require.Equal(t, int64(2), engine.enqueues.Load(), "subsequent WebSocket turns must be audited again")
+}
+
+type turnCountingEngine struct {
+	mode     securityaudit.Mode
+	enqueues atomic.Int64
+}
+
+func (e *turnCountingEngine) EffectiveMode() securityaudit.Mode { return e.mode }
+func (e *turnCountingEngine) Enqueue(context.Context, securityaudit.Request) error {
+	e.enqueues.Add(1)
+	return nil
+}
+func (e *turnCountingEngine) Evaluate(context.Context, securityaudit.Request) (*securityaudit.PromptDecision, error) {
+	return &securityaudit.PromptDecision{Kind: securityaudit.DecisionAllow, AllowNextStage: true}, nil
+}

backend/internal/handler/security_audit_media_submit_test.go

@@ -0,0 +1,208 @@
+package handler
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Wei-Shaw/sub2api/internal/securityaudit"
+	middleware2 "github.com/Wei-Shaw/sub2api/internal/server/middleware"
+	"github.com/Wei-Shaw/sub2api/internal/service"
+	"github.com/gin-gonic/gin"
+	"github.com/stretchr/testify/require"
+)
+
+type handlerPromptEngine struct {
+	mu sync.Mutex
+
+	mode      securityaudit.Mode
+	decision  *securityaudit.PromptDecision
+	err       error
+	evaluated int
+	enqueued  int
+	requests  []securityaudit.Request
+}
+
+func (e *handlerPromptEngine) EffectiveMode() securityaudit.Mode { return e.mode }
+func (e *handlerPromptEngine) Enqueue(_ context.Context, req securityaudit.Request) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.enqueued++
+	e.requests = append(e.requests, req.Clone())
+	return e.err
+}
+func (e *handlerPromptEngine) Evaluate(_ context.Context, req securityaudit.Request) (*securityaudit.PromptDecision, error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	e.evaluated++
+	e.requests = append(e.requests, req.Clone())
+	return e.decision, e.err
+}
+func (e *handlerPromptEngine) snapshot() (evaluated, enqueued int, requests []securityaudit.Request) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	requests = make([]securityaudit.Request, len(e.requests))
+	copy(requests, e.requests)
+	return e.evaluated, e.enqueued, requests
+}
+
+func securityAuditMediaTestMiddleware(c *gin.Context) {
+	groupID := int64(3)
+	user := &service.User{ID: 7, Username: "media-user", Email: "media@example.test"}
+	c.Set(string(middleware2.ContextKeyAPIKey), &service.APIKey{
+		ID: 9, UserID: 7, User: user, Name: "media-key", GroupID: &groupID,
+		Group: &service.Group{ID: groupID, Name: "media-group", Platform: service.PlatformOpenAI, AllowImageGeneration: true},
+	})
+	c.Set(string(middleware2.ContextKeyUser), middleware2.AuthSubject{UserID: 7, Concurrency: 2})
+	c.Next()
+}
+
+func blockingHandlerPromptEngine() *handlerPromptEngine {
+	return &handlerPromptEngine{mode: securityaudit.ModeBlocking, decision: &securityaudit.PromptDecision{
+		Kind: securityaudit.DecisionBlock, ErrorCode: securityaudit.ErrorCodeBlocked, AllowNextStage: false,
+	}}
+}
+
+func TestAsyncImagePromptGuardRunsBeforeTaskCreation(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	store := &asyncImageMemoryStore{tasks: map[string]*service.ImageTaskRecord{}}
+	tasks := service.NewImageTaskServiceWithUploader(store, nil, time.Hour, time.Minute)
+	engine := blockingHandlerPromptEngine()
+	openAI := &OpenAIGatewayHandler{securityAuditCoordinator: securityaudit.NewCoordinator(nil, engine)}
+	h := &AsyncImageHandler{tasks: tasks, openAI: openAI}
+	executions := 0
+	h.execute = func(string, *gin.Context) { executions++ }
+
+	router := gin.New()
+	router.Use(securityAuditMediaTestMiddleware)
+	router.POST("/v1/images/generations/async", h.Submit)
+	request := httptest.NewRequest(http.MethodPost, "/v1/images/generations/async", strings.NewReader(`{"model":"gpt-image-2","prompt":"blocked async prompt"}`))
+	request.Header.Set("Content-Type", "application/json")
+	recorder := httptest.NewRecorder()
+	router.ServeHTTP(recorder, request)
+
+	require.Equal(t, http.StatusForbidden, recorder.Code)
+	require.Contains(t, recorder.Body.String(), securityaudit.ErrorCodeBlocked)
+	require.Empty(t, store.tasks, "no asynchronous task may exist after a blocking decision")
+	require.Zero(t, executions)
+	evaluated, _, requests := engine.snapshot()
+	require.Equal(t, 1, evaluated)
+	require.Len(t, requests, 1)
+	require.Contains(t, string(requests[0].Body), "blocked async prompt")
+}
+
+func TestAsyncImageSuccessfulPrecheckIsNotRepeatedByDetachedExecution(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	store := &asyncImageMemoryStore{tasks: map[string]*service.ImageTaskRecord{}}
+	tasks := service.NewImageTaskServiceWithUploader(store, nil, time.Hour, time.Minute)
+	engine := &handlerPromptEngine{mode: securityaudit.ModeBlocking, decision: &securityaudit.PromptDecision{Kind: securityaudit.DecisionAllow, AllowNextStage: true}}
+	openAI := &OpenAIGatewayHandler{securityAuditCoordinator: securityaudit.NewCoordinator(nil, engine)}
+	h := &AsyncImageHandler{tasks: tasks, openAI: openAI}
+	var executionMu sync.Mutex
+	repeatedDecision := false
+	h.execute = func(_ string, c *gin.Context) {
+		apiKey, _ := middleware2.GetAPIKeyFromContext(c)
+		subject, _ := middleware2.GetAuthSubjectFromContext(c)
+		decision := openAI.checkSecurityAudit(c, nil, apiKey, subject, service.ContentModerationProtocolOpenAIImages, "gpt-image-2", []byte(`{"prompt":"must not rescan"}`))
+		executionMu.Lock()
+		repeatedDecision = decision != nil
+		executionMu.Unlock()
+		c.JSON(http.StatusOK, gin.H{"created": 1, "data": []any{}})
+	}
+
+	router := gin.New()
+	router.Use(securityAuditMediaTestMiddleware)
+	router.POST("/v1/images/generations/async", h.Submit)
+	request := httptest.NewRequest(http.MethodPost, "/v1/images/generations/async", strings.NewReader(`{"model":"gpt-image-2","prompt":"allowed async prompt"}`))
+	request.Header.Set("Content-Type", "application/json")
+	recorder := httptest.NewRecorder()
+	router.ServeHTTP(recorder, request)
+	require.Equal(t, http.StatusAccepted, recorder.Code)
+	require.Eventually(t, func() bool {
+		store.mu.RLock()
+		defer store.mu.RUnlock()
+		for _, task := range store.tasks {
+			if task.Status == service.ImageTaskStatusCompleted {
+				return true
+			}
+		}
+		return false
+	}, time.Second, 10*time.Millisecond)
+	evaluated, _, _ := engine.snapshot()
+	require.Equal(t, 1, evaluated)
+	executionMu.Lock()
+	require.False(t, repeatedDecision)
+	executionMu.Unlock()
+}
+
+func TestBatchImagePromptGuardRunsBeforePersistenceOrBilling(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	engine := blockingHandlerPromptEngine()
+	openAI := &OpenAIGatewayHandler{securityAuditCoordinator: securityaudit.NewCoordinator(nil, engine)}
+	h := &BatchImageHandler{openAI: openAI}
+	router := gin.New()
+	router.Use(securityAuditMediaTestMiddleware)
+	router.POST("/v1/images/batches", h.Submit)
+	body := map[string]any{
+		"model": "gemini-image-test",
+		"items": []map[string]any{{
+			"custom_id": "one", "prompt": "blocked batch prompt",
+			"reference_images": []map[string]any{{"mime_type": "image/png", "data": []byte("BINARY_CANARY")}},
+		}},
+	}
+	raw, err := json.Marshal(body)
+	require.NoError(t, err)
+	request := httptest.NewRequest(http.MethodPost, "/v1/images/batches", strings.NewReader(string(raw)))
+	request.Header.Set("Content-Type", "application/json")
+	recorder := httptest.NewRecorder()
+	require.NotPanics(t, func() { router.ServeHTTP(recorder, request) }, "nil service would panic if Submit were reached")
+	require.Equal(t, http.StatusForbidden, recorder.Code)
+	evaluated, _, requests := engine.snapshot()
+	require.Equal(t, 1, evaluated)
+	require.Len(t, requests, 1)
+	require.Contains(t, string(requests[0].Body), "blocked batch prompt")
+	require.NotContains(t, string(requests[0].Body), "BINARY_CANARY")
+	require.NotContains(t, string(requests[0].Body), "QklOQVJZX0NBTkFSWQ==")
+}
+
+func TestSecurityAuditBlockingFailuresLeaveAllDownstreamCountersAtZero(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	for _, kind := range []securityaudit.DecisionKind{securityaudit.DecisionBlock, securityaudit.DecisionUnavailable, securityaudit.DecisionInvalid} {
+		t.Run(string(kind), func(t *testing.T) {
+			promptDecision := promptGuardDecision(kind)
+			engine := &handlerPromptEngine{mode: securityaudit.ModeBlocking, decision: &securityaudit.PromptDecision{
+				Kind: kind, ErrorCode: promptDecision.ErrorCode, AllowNextStage: false,
+			}}
+			coordinator := securityaudit.NewCoordinator(nil, engine)
+			recorder := httptest.NewRecorder()
+			c, _ := gin.CreateTestContext(recorder)
+			c.Request = httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{"model":"gpt-test","messages":[{"role":"user","content":"guard me"}]}`))
+			groupID := int64(3)
+			apiKey := &service.APIKey{ID: 9, UserID: 7, GroupID: &groupID, Group: &service.Group{ID: groupID, Platform: service.PlatformOpenAI}}
+			subject := middleware2.AuthSubject{UserID: 7, Concurrency: 2}
+			decision := runSecurityAudit(c, nil, coordinator, nil, apiKey, subject, service.ContentModerationProtocolOpenAIChat, "gpt-test", []byte(`{"messages":[{"role":"user","content":"guard me"}]}`), "http")
+			require.NotNil(t, decision)
+			require.False(t, decision.AllowNextStage)
+			require.False(t, recorder.Result().Header.Get("Content-Type") != "", "Guard evaluation itself must not start SSE/HTTP output")
+
+			accountSelections, billingChecks, billingPreconsumes, upstreamDispatches := 0, 0, 0, 0
+			if decision.AllowNextStage {
+				accountSelections++
+				billingChecks++
+				billingPreconsumes++
+				upstreamDispatches++
+			}
+			require.Zero(t, accountSelections)
+			require.Zero(t, billingChecks)
+			require.Zero(t, billingPreconsumes)
+			require.Zero(t, upstreamDispatches)
+			(&OpenAIGatewayHandler{}).openAISecurityAuditError(c, decision)
+			require.Equal(t, promptDecision.HTTPStatus, recorder.Code)
+		})
+	}
+}

backend/internal/handler/security_audit_order_test.go

@@ -0,0 +1,85 @@
+package handler
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"os"
+	"regexp"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+type promptAuditOrderCase struct {
+	file       string
+	function   string
+	auditToken string
+}
+
+func TestPromptAuditGatePrecedesAccountBillingAndUpstreamSideEffects(t *testing.T) {
+	tests := []promptAuditOrderCase{
+		{file: "gateway_handler.go", function: "Messages", auditToken: "checkSecurityAudit"},
+		{file: "gateway_handler_chat_completions.go", function: "ChatCompletions", auditToken: "checkSecurityAudit"},
+		{file: "gateway_handler_responses.go", function: "Responses", auditToken: "checkSecurityAudit"},
+		{file: "gemini_v1beta_handler.go", function: "GeminiV1BetaModels", auditToken: "checkSecurityAudit"},
+		{file: "openai_gateway_handler.go", function: "Responses", auditToken: "checkSecurityAudit"},
+		{file: "openai_gateway_handler.go", function: "Messages", auditToken: "checkSecurityAudit"},
+		{file: "openai_chat_completions.go", function: "ChatCompletions", auditToken: "checkSecurityAudit"},
+		{file: "openai_images.go", function: "Images", auditToken: "checkSecurityAudit"},
+		{file: "grok_media.go", function: "handleGrokMedia", auditToken: "checkSecurityAudit"},
+		{file: "openai_embeddings.go", function: "Embeddings", auditToken: "checkSecurityAudit"},
+		{file: "openai_alpha_search.go", function: "AlphaSearch", auditToken: "checkSecurityAudit"},
+		{file: "image_task_handler.go", function: "Submit", auditToken: "checkSecurityAuditBeforeSubmit"},
+		{file: "batch_image_handler.go", function: "Submit", auditToken: "checkSecurityAuditBeforeSubmit"},
+	}
+	sideEffectTokens := []string{
+		"CheckBillingEligibility(", "SelectAccount", ".Forward", "acquireResponsesUserSlot(",
+		"AcquireUserSlot", "TryAcquireUserSlot", "acquireImageGenerationSlot(",
+		"h.tasks.Create(", "h.service.Submit(",
+	}
+	for _, tt := range tests {
+		t.Run(tt.file+"/"+tt.function, func(t *testing.T) {
+			functionSource := stripGoComments(goFunctionSource(t, tt.file, tt.function))
+			auditIndex := strings.Index(functionSource, tt.auditToken)
+			require.NotEqual(t, -1, auditIndex, "missing Prompt Audit gate")
+			foundSideEffect := false
+			for _, sideEffect := range sideEffectTokens {
+				index := strings.Index(functionSource, sideEffect)
+				if index < 0 {
+					continue
+				}
+				foundSideEffect = true
+				require.Lessf(t, auditIndex, index, "%s must run before %s", tt.auditToken, sideEffect)
+			}
+			require.True(t, foundSideEffect, "coverage case must contain a downstream side effect")
+		})
+	}
+}
+
+func stripGoComments(source string) string {
+	source = regexp.MustCompile(`(?s)/\*.*?\*/`).ReplaceAllString(source, "")
+	return regexp.MustCompile(`(?m)//.*$`).ReplaceAllString(source, "")
+}
+
+func goFunctionSource(t *testing.T, filename, functionName string) string {
+	t.Helper()
+	raw, err := os.ReadFile(filename)
+	require.NoError(t, err)
+	files := token.NewFileSet()
+	parsed, err := parser.ParseFile(files, filename, raw, 0)
+	require.NoError(t, err)
+	for _, declaration := range parsed.Decls {
+		function, ok := declaration.(*ast.FuncDecl)
+		if !ok || function.Name.Name != functionName || function.Body == nil {
+			continue
+		}
+		start := files.Position(function.Pos()).Offset
+		end := files.Position(function.End()).Offset
+		require.Greater(t, end, start)
+		return string(raw[start:end])
+	}
+	t.Fatalf("function %s not found in %s", functionName, filename)
+	return ""
+}

backend/internal/handler/wire.go

@@ -1,7 +1,9 @@
 package handler
 
 import (
+	"github.com/Wei-Shaw/sub2api/internal/config"
 	"github.com/Wei-Shaw/sub2api/internal/handler/admin"
+	"github.com/Wei-Shaw/sub2api/internal/securityaudit"
 	"github.com/Wei-Shaw/sub2api/internal/service"
 
 	"github.com/google/wire"
@@ -38,6 +40,7 @@ func ProvideAdminHandlers(
 	channelMonitorHandler *admin.ChannelMonitorHandler,
 	channelMonitorTemplateHandler *admin.ChannelMonitorRequestTemplateHandler,
 	contentModerationHandler *admin.ContentModerationHandler,
+	promptAuditHandler *securityaudit.PromptAdminHandler,
 	paymentHandler *admin.PaymentHandler,
 	affiliateHandler *admin.AffiliateHandler,
 	complianceHandler *admin.ComplianceHandler,
@@ -75,6 +78,7 @@ func ProvideAdminHandlers(
 		ChannelMonitor:         channelMonitorHandler,
 		ChannelMonitorTemplate: channelMonitorTemplateHandler,
 		ContentModeration:      contentModerationHandler,
+		PromptAudit:            promptAuditHandler,
 		Payment:                paymentHandler,
 		Affiliate:              affiliateHandler,
 		Compliance:             complianceHandler,
@@ -82,6 +86,60 @@ func ProvideAdminHandlers(
 	}
 }
 
+func ProvideGatewayHandler(
+	gatewayService *service.GatewayService,
+	openAIGatewayService *service.OpenAIGatewayService,
+	geminiCompatService *service.GeminiMessagesCompatService,
+	antigravityGatewayService *service.AntigravityGatewayService,
+	userService *service.UserService,
+	concurrencyService *service.ConcurrencyService,
+	billingCacheService *service.BillingCacheService,
+	usageService *service.UsageService,
+	apiKeyService *service.APIKeyService,
+	usageRecordWorkerPool *service.UsageRecordWorkerPool,
+	errorPassthroughService *service.ErrorPassthroughService,
+	contentModerationService *service.ContentModerationService,
+	userMsgQueueService *service.UserMessageQueueService,
+	cfg *config.Config,
+	settingService *service.SettingService,
+	coordinator *securityaudit.Coordinator,
+) *GatewayHandler {
+	h := NewGatewayHandler(gatewayService, openAIGatewayService, geminiCompatService, antigravityGatewayService,
+		userService, concurrencyService, billingCacheService, usageService, apiKeyService, usageRecordWorkerPool,
+		errorPassthroughService, contentModerationService, userMsgQueueService, cfg, settingService)
+	h.securityAuditCoordinator = coordinator
+	return h
+}
+
+func ProvideOpenAIGatewayHandler(
+	gatewayService *service.OpenAIGatewayService,
+	concurrencyService *service.ConcurrencyService,
+	billingCacheService *service.BillingCacheService,
+	apiKeyService *service.APIKeyService,
+	usageRecordWorkerPool *service.UsageRecordWorkerPool,
+	errorPassthroughService *service.ErrorPassthroughService,
+	contentModerationService *service.ContentModerationService,
+	opsService *service.OpsService,
+	cfg *config.Config,
+	coordinator *securityaudit.Coordinator,
+) *OpenAIGatewayHandler {
+	h := NewOpenAIGatewayHandler(gatewayService, concurrencyService, billingCacheService, apiKeyService,
+		usageRecordWorkerPool, errorPassthroughService, contentModerationService, opsService, cfg)
+	h.securityAuditCoordinator = coordinator
+	return h
+}
+
+func ProvideBatchImageHandler(
+	batchService *service.BatchImagePublicService,
+	download *service.BatchImageDownloadService,
+	cleanup *service.BatchImageCleanupService,
+	openAI *OpenAIGatewayHandler,
+) *BatchImageHandler {
+	h := NewBatchImageHandler(batchService, download, cleanup)
+	h.openAI = openAI
+	return h
+}
+
 // ProvideSystemHandler creates admin.SystemHandler with UpdateService
 func ProvideSystemHandler(updateService *service.UpdateService, lockService *service.SystemOperationLockService) *admin.SystemHandler {
 	return admin.NewSystemHandler(updateService, lockService)
@@ -157,15 +215,15 @@ var ProviderSet = wire.NewSet(
 	NewSubscriptionHandler,
 	NewAnnouncementHandler,
 	NewChannelMonitorUserHandler,
-	NewGatewayHandler,
-	NewOpenAIGatewayHandler,
+	ProvideGatewayHandler,
+	ProvideOpenAIGatewayHandler,
 	NewTotpHandler,
 	ProvideSettingHandler,
 	NewPaymentHandler,
 	NewPaymentWebhookHandler,
 	NewAvailableChannelHandler,
 	NewAsyncImageHandler,
-	NewBatchImageHandler,
+	ProvideBatchImageHandler,
 
 	// Admin handlers
 	admin.NewDashboardHandler,