feat(audit): 请求审计中新增不同账号调度详情、完整HTTP请求/响应报文、默认7天全量日志记录等完整诊断生命周期重要功能

backend/internal/app/application.go

@@ -592,6 +592,17 @@ func (a *Application) Run(ctx context.Context) error {
 		})
 		return nil
 	})
+	startBackground("audit_retention_cleanup", func(taskCtx context.Context) error {
+		a.runPeriodicTask(taskCtx, time.Hour, "audit_retention_cleanup", func(runCtx context.Context) error {
+			retentionDays := a.settings.Get().Config.Audit.RetentionDays
+			if retentionDays == 0 {
+				return nil
+			}
+			_, err := a.audits.PurgeOutdated(runCtx, retentionDays)
+			return err
+		})
+		return nil
+	})
 	startBackground("quota_recovery", func(taskCtx context.Context) error {
 		a.quotaRecovery.Run(taskCtx)
 		return nil

backend/internal/application/audit/service.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"log/slog"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -67,13 +68,20 @@ const (
 )
 
 const (
-	auditEnqueueWait        = 25 * time.Millisecond
-	auditWriteTimeout       = 2 * time.Second
-	auditWriteAttempts      = 3
-	auditWriteRetryBase     = 250 * time.Millisecond
-	auditWriteRetryMax      = 5 * time.Second
-	auditDefaultCommitDelay = 5 * time.Millisecond
-	auditSummaryTTL         = 10 * time.Second
+	auditEnqueueWait         = 25 * time.Millisecond
+	auditWriteTimeout        = 2 * time.Second
+	auditWriteAttempts       = 3
+	auditWriteRetryBase      = 250 * time.Millisecond
+	auditWriteRetryMax       = 5 * time.Second
+	auditDefaultCommitDelay  = 5 * time.Millisecond
+	auditSummaryTTL          = 10 * time.Second
+	requestMethodLimit       = 16
+	requestPathLimit         = 2048
+	requestHeaderNameLimit   = 256
+	requestHeaderValueLimit  = 2048
+	requestHeaderValuesLimit = 32
+	requestHeaderCountLimit  = 128
+	requestHeadersLimit      = 32 << 10
 )
 
 type auditWriteRequest struct {
@@ -244,7 +252,7 @@ func (s *Service) Start() {
 
 // Record 将审计写入有界队列；突发满载时短暂等待，持续拥塞才降级丢弃审计。
 func (s *Service) Record(value auditdomain.Record) bool {
-	return s.enqueueBestEffort(context.Background(), auditWriteRequest{record: value}) == nil
+	return s.enqueueBestEffort(context.Background(), auditWriteRequest{record: sanitizeRequestMetadata(value)}) == nil
 }
 
 // Create returns success only after the audit and billing transaction commits.
@@ -265,6 +273,7 @@ func (s *Service) createAcknowledged(ctx context.Context, value auditdomain.Reco
 	if value.ClientIP == "" {
 		value.ClientIP = requestmeta.ClientIP(ctx)
 	}
+	value = sanitizeRequestMetadata(value)
 	if !s.started.Load() || s.stopped.Load() {
 		return ErrWriterUnavailable
 	}
@@ -290,6 +299,70 @@ func (s *Service) createAcknowledged(ctx context.Context, value auditdomain.Reco
 	}
 }
 
+func sanitizeRequestMetadata(value auditdomain.Record) auditdomain.Record {
+	// Query strings can contain credentials or signed resource URLs. Endpoint
+	// identity only needs the path.
+	value.RequestMethod = truncateRequestMetadata(strings.ToUpper(strings.TrimSpace(value.RequestMethod)), requestMethodLimit)
+	value.RequestPath = truncateRequestMetadata(strings.SplitN(value.RequestPath, "?", 2)[0], requestPathLimit)
+	if len(value.RequestHeaders) == 0 {
+		value.RequestHeaders = nil
+		return value
+	}
+	names := make([]string, 0, len(value.RequestHeaders))
+	for name := range value.RequestHeaders {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	result := make(map[string][]string, len(value.RequestHeaders))
+	for _, originalName := range names {
+		if len(result) >= requestHeaderCountLimit {
+			break
+		}
+		name := truncateRequestMetadata(strings.TrimSpace(originalName), requestHeaderNameLimit)
+		if name == "" {
+			continue
+		}
+		values := value.RequestHeaders[originalName]
+		cloned := make([]string, 0, len(values))
+		if isSensitiveRequestHeader(name) {
+			cloned = []string{"[REDACTED]"}
+		} else {
+			for _, item := range values[:min(len(values), requestHeaderValuesLimit)] {
+				cloned = append(cloned, truncateRequestMetadata(item, requestHeaderValueLimit))
+			}
+		}
+		result[name] = cloned
+		encoded, err := json.Marshal(result)
+		if err != nil || len(encoded) > requestHeadersLimit {
+			delete(result, name)
+			break
+		}
+	}
+	value.RequestHeaders = result
+	return value
+}
+
+func truncateRequestMetadata(value string, limit int) string {
+	value = strings.ToValidUTF8(value, "�")
+	runes := []rune(value)
+	if len(runes) <= limit {
+		return value
+	}
+	return string(runes[:limit])
+}
+
+func isSensitiveRequestHeader(name string) bool {
+	name = strings.ToLower(strings.TrimSpace(name))
+	for _, marker := range []string{
+		"auth", "cookie", "credential", "dpop", "jwt", "key", "password", "secret", "session", "signature", "token",
+	} {
+		if strings.Contains(name, marker) {
+			return true
+		}
+	}
+	return false
+}
+
 func (s *Service) tryEnqueue(request auditWriteRequest) error {
 	s.lifecycleMu.RLock()
 	defer s.lifecycleMu.RUnlock()
@@ -405,6 +478,15 @@ func (s *Service) Get(ctx context.Context, id uint64) (auditdomain.Record, error
 	return s.audits.Get(ctx, id)
 }
 
+// PurgeOutdated 清理超过指定保留天数的历史审计记录。
+func (s *Service) PurgeOutdated(ctx context.Context, retentionDays int) (int64, error) {
+	if retentionDays <= 0 {
+		return 0, nil
+	}
+	cutoff := time.Now().UTC().AddDate(0, 0, -retentionDays)
+	return s.audits.PurgeOlderThan(ctx, cutoff)
+}
+
 // CursorResult 表示按递减 ID 游标读取的一页审计记录。
 type CursorResult struct {
 	Items      []auditdomain.Record

backend/internal/application/audit/service_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log/slog"
 	"path/filepath"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -70,6 +71,40 @@ func TestCreateCapturesClientIPFromRequestContext(t *testing.T) {
 	}
 }
 
+func TestCreateKeepsOnlySanitizedRequestMetadata(t *testing.T) {
+	repo := newGatedAuditRepository()
+	close(repo.release)
+	service := NewService(repo, slog.Default(), 8, 1, time.Hour)
+	service.Start()
+	t.Cleanup(func() { closeAuditService(t, service) })
+
+	record := auditdomain.Record{
+		EventID: "evt_payload_policy_0001", RequestID: "payload-policy", ClientKeyID: 1, ModelRouteID: 1, StatusCode: 200,
+		RequestMethod: "POST", RequestPath: "/v1/responses?api_key=query-secret",
+		RequestHeaders: map[string][]string{
+			"Authorization":  {"Bearer client-secret"},
+			"X-Api-Key":      {"client-key"},
+			"X-Goog-Api-Key": {"google-key"},
+			"X-Session-ID":   {"session-secret"},
+			"User-Agent":     {"audit-test"},
+			"X-Oversized":    {strings.Repeat("x", requestHeaderValueLimit+100)},
+		},
+	}
+	if err := service.Create(context.Background(), record); err != nil {
+		t.Fatal(err)
+	}
+	stored := <-repo.started
+	if len(stored) != 1 || stored[0].RequestMethod != "POST" || stored[0].RequestPath != "/v1/responses" {
+		t.Fatalf("request metadata = %#v", stored)
+	}
+	if stored[0].RequestHeaders["Authorization"][0] != "[REDACTED]" || stored[0].RequestHeaders["X-Api-Key"][0] != "[REDACTED]" || stored[0].RequestHeaders["X-Goog-Api-Key"][0] != "[REDACTED]" || stored[0].RequestHeaders["X-Session-ID"][0] != "[REDACTED]" || stored[0].RequestHeaders["User-Agent"][0] != "audit-test" {
+		t.Fatalf("sanitized request headers = %#v", stored[0].RequestHeaders)
+	}
+	if len([]rune(stored[0].RequestHeaders["X-Oversized"][0])) != requestHeaderValueLimit {
+		t.Fatalf("oversized header was not bounded: %d", len([]rune(stored[0].RequestHeaders["X-Oversized"][0])))
+	}
+}
+
 func TestAuditBatchRetriesTransientDatabaseFailure(t *testing.T) {
 	repo := &flakyAuditRepository{failures: 5}
 	service := NewService(repo, slog.Default(), 8, 4, time.Hour)

backend/internal/application/gateway/image.go

@@ -31,6 +31,9 @@ type ImageGenerationInput struct {
 	ResponseFormat string
 	Streaming      bool
 	PartialImages  int
+	Method         string
+	Path           string
+	Headers        map[string][]string
 }
 
 // ImageEditInput 表示图片编辑用例已经完成协议校验后的输入。
@@ -48,6 +51,9 @@ type ImageEditInput struct {
 	ResponseFormat string
 	Streaming      bool
 	PartialImages  int
+	Method         string
+	Path           string
+	Headers        map[string][]string
 }
 
 type imageProviderSupport func(accountdomain.Provider) bool
@@ -69,7 +75,7 @@ func (s *Service) GenerateImage(ctx context.Context, input ImageGenerationInput)
 			Size: input.Size, AspectRatio: input.AspectRatio, Resolution: input.Resolution, Quality: input.Quality,
 			ResponseFormat: input.ResponseFormat, Streaming: input.Streaming, PartialImages: input.PartialImages,
 		})
-	}, input.Streaming, input.Resolution, input.Quality, input.Count, 0)
+	}, input.Streaming, input.Resolution, input.Quality, input.Count, 0, input.Method, input.Path, input.Headers)
 }
 
 // EditImage 选择支持图片编辑的路由和账号，并返回可统一审计的上游响应。
@@ -88,7 +94,7 @@ func (s *Service) EditImage(ctx context.Context, input ImageEditInput) (*Result,
 			Resolution: input.Resolution, Quality: input.Quality, ResponseFormat: input.ResponseFormat,
 			Streaming: input.Streaming, PartialImages: input.PartialImages,
 		})
-	}, input.Streaming, input.Resolution, input.Quality, input.Count, len(input.ImageURLs))
+	}, input.Streaming, input.Resolution, input.Quality, input.Count, len(input.ImageURLs), input.Method, input.Path, input.Headers)
 }
 
 func (s *Service) executeImage(
@@ -105,6 +111,9 @@ func (s *Service) executeImage(
 	quality string,
 	requestedCount int,
 	inputImageCount int,
+	method string,
+	path string,
+	headers map[string][]string,
 ) (*Result, error) {
 	ctx, egressTrace := infraegress.WithTrace(ctx)
 	startedAt := time.Now()
@@ -130,6 +139,7 @@ func (s *Service) executeImage(
 		ClientIP:     requestmeta.ClientIP(ctx),
 		ModelRouteID: route.ID, ModelPublicID: externalModel, ModelUpstreamModel: modeldomain.DisplayUpstreamModel(route.Provider, route.UpstreamModel),
 		Provider: string(route.Provider), Operation: operation, UsageSource: audit.UsageSourceNone, Streaming: streaming,
+		RequestMethod: method, RequestPath: path, RequestHeaders: headers,
 	}
 	if operation == audit.OperationImageEdit {
 		auditBase.MediaInputImages = int64(max(0, inputImageCount))

backend/internal/application/gateway/service.go

@@ -115,6 +115,9 @@ type Input struct {
 	// GrokTurnIndex forwards only the turn supplied by a real Grok Shell client; the server never infers or increments it.
 	GrokTurnIndex string
 	Operation     audit.Operation
+	Method        string
+	Path          string
+	Headers       map[string][]string
 	// auditOperation may classify a normal protocol request differently for
 	// operator visibility without changing routing or Provider semantics.
 	auditOperation audit.Operation
@@ -953,6 +956,7 @@ func (s *Service) createResponseAt(ctx context.Context, input Input, path string
 		ModelRouteID: route.ID, ModelPublicID: publicModel, ModelUpstreamModel: modeldomain.DisplayUpstreamModel(route.Provider, route.UpstreamModel),
 		Provider: string(route.Provider), Operation: auditOperation, UsageSource: audit.UsageSourceNone, Streaming: input.Streaming,
 		MediaInputImages: mediaSummary.InputImages,
+		RequestMethod:    input.Method, RequestPath: input.Path, RequestHeaders: input.Headers,
 	}
 	if errors.Is(routeErr, clientkeyapp.ErrModelNotAllowed) {
 		record := auditBase
@@ -1115,6 +1119,25 @@ func (s *Service) createResponseAt(ctx context.Context, input Input, path string
 				record.DurationMS = time.Since(startedAt).Milliseconds()
 				record.ErrorCode = errorCode
 				attempts := failureAttempts.snapshot()
+				if !successful && len(attempts) == 0 {
+					statusCode := response.StatusCode
+					failureAttempts.append(audit.Attempt{
+						Source:             audit.AttemptSourceUpstreamHTTP,
+						Stage:              "response_stream",
+						AccountID:          auditAccountID(credential.ID),
+						AccountName:        credential.Name,
+						Method:             http.MethodPost,
+						RequestPath:        sanitizeRequestPath(path),
+						UpstreamURL:        sanitizeUpstreamURL(response.UpstreamURL),
+						StartedAt:          upstreamStartedAt.UTC(),
+						DurationMS:         time.Since(upstreamStartedAt).Milliseconds(),
+						UpstreamStatusCode: &statusCode,
+						UpstreamStatus:     response.Status,
+						ResponseHeaders:    sanitizeDiagnosticHeaders(response.Header),
+						TransportError:     errorCode,
+					})
+					attempts = failureAttempts.snapshot()
+				}
 				if !successful || len(attempts) > 0 {
 					record.Attempts = attempts
 				}

backend/internal/application/gateway/service_test.go

@@ -1234,7 +1234,7 @@ func TestUnpricedVoiceRemainsAvailableToFiniteClientKey(t *testing.T) {
 	accountService := accountapp.NewService(accountRepo, auditRepo, memory.NewDeviceSessionStore(), sticky, registry, testCipher(t), nil)
 	service := NewService(modelRepo, auditRepo, accountService, clientkeyapp.NewService(nil, nil, nil, 60, 4, nil), registry, selector, nil, 1)
 	executed := false
-	result, err := service.executeVoice(ctx, "req-voice-billing", clientkey.Key{ID: 1, BillingLimitUSDTicks: 1}, voiceModel, audit.OperationTTS, modeldomain.CapabilityTTS, true, audit.PricingResult{}, func(account.Provider) bool {
+	result, err := service.executeVoice(ctx, "req-voice-billing", clientkey.Key{ID: 1, BillingLimitUSDTicks: 1}, voiceModel, audit.OperationTTS, modeldomain.CapabilityTTS, true, audit.PricingResult{}, "", "", nil, func(account.Provider) bool {
 		return true
 	}, func(context.Context, account.Provider, account.Credential, string) (voiceExecutionResult, error) {
 		executed = true
@@ -1305,7 +1305,7 @@ func TestVoicePricingSettlesTTSAndRESTSTTUsage(t *testing.T) {
 	if !ok {
 		t.Fatal("TTS pricing unavailable")
 	}
-	ttsResult, err := service.executeVoice(ctx, "req-priced-tts", limitedKey, ttsModel, audit.OperationTTS, modeldomain.CapabilityTTS, true, ttsPricing, func(account.Provider) bool {
+	ttsResult, err := service.executeVoice(ctx, "req-priced-tts", limitedKey, ttsModel, audit.OperationTTS, modeldomain.CapabilityTTS, true, ttsPricing, "", "", nil, func(account.Provider) bool {
 		return true
 	}, func(context.Context, account.Provider, account.Credential, string) (voiceExecutionResult, error) {
 		return voiceExecutionResult{response: jsonVoiceResponse(http.StatusOK, map[string]any{"ok": true}), pricing: ttsPricing}, nil
@@ -1328,7 +1328,7 @@ func TestVoicePricingSettlesTTSAndRESTSTTUsage(t *testing.T) {
 	if !ok {
 		t.Fatal("STT pricing unavailable")
 	}
-	sttResult, err := service.executeVoice(ctx, "req-priced-stt", limitedKey, sttModel, audit.OperationSTT, modeldomain.CapabilitySTT, true, audit.PricingResult{}, func(account.Provider) bool {
+	sttResult, err := service.executeVoice(ctx, "req-priced-stt", limitedKey, sttModel, audit.OperationSTT, modeldomain.CapabilitySTT, true, audit.PricingResult{}, "", "", nil, func(account.Provider) bool {
 		return true
 	}, func(context.Context, account.Provider, account.Credential, string) (voiceExecutionResult, error) {
 		return voiceExecutionResult{response: jsonVoiceResponse(http.StatusOK, map[string]any{"text": "hello"}), pricing: sttPricing}, nil
@@ -1369,7 +1369,7 @@ func TestVoicePricingSettlesTTSAndRESTSTTUsage(t *testing.T) {
 		t.Fatal(err)
 	}
 	executed := false
-	_, err = service.executeVoice(ctx, "req-capped-tts", cappedKey, ttsModel, audit.OperationTTS, modeldomain.CapabilityTTS, true, ttsPricing, func(account.Provider) bool {
+	_, err = service.executeVoice(ctx, "req-capped-tts", cappedKey, ttsModel, audit.OperationTTS, modeldomain.CapabilityTTS, true, ttsPricing, "", "", nil, func(account.Provider) bool {
 		return true
 	}, func(context.Context, account.Provider, account.Credential, string) (voiceExecutionResult, error) {
 		executed = true

backend/internal/application/gateway/video.go

@@ -994,6 +994,7 @@ func (s *Service) recordVideoAudit(ctx context.Context, job media.Job, durationM
 		EgressNodeID: job.EgressNodeID, EgressNodeName: job.EgressNodeName, EgressScope: job.EgressScope, EgressMode: audit.EgressMode(job.EgressMode),
 		MediaInputImages: int64(job.InputImageCount),
 		DurationMS:       durationMS, AttemptCount: len(attempts), Attempts: append([]audit.Attempt(nil), attempts...), CreatedAt: createdAt,
+		RequestMethod: http.MethodPost, RequestPath: "/v1/videos/generations",
 	}
 	if job.Status == media.StatusCompleted && job.Seconds > 0 {
 		record.MediaOutputSeconds = int64(max(0, job.Seconds))