Merge pull request #4264 from lyon-le/perf/content-moderation-keyword-hot-path

perf(risk): 优化内容审核关键词热路径（优化性能）

backend/internal/service/content_moderation.go

@@ -92,6 +92,9 @@ const (
 	contentModerationCleanupInterval = 24 * time.Hour
 	contentModerationCleanupTimeout  = 30 * time.Minute
 	contentModerationCleanupDelay    = 5 * time.Minute
+
+	contentModerationRuntimeCacheTTL       = time.Second
+	contentModerationRuntimeRefreshTimeout = 5 * time.Second
 )
 
 var contentModerationCategoryOrder = []string{
@@ -512,10 +515,22 @@ type ContentModerationService struct {
 	lastCleanupUnix          atomic.Int64
 	lastCleanupDeletedHit    atomic.Int64
 	lastCleanupDeletedNonHit atomic.Int64
+	runtimeSnapshot          atomic.Pointer[contentModerationRuntimeSnapshot]
+	runtimeRefreshMu         sync.Mutex
+	runtimeCacheTTL          time.Duration
+	runtimeRefreshRetryAt    atomic.Int64
 	keyHealthMu              sync.Mutex
 	keyHealth                map[string]*contentModerationKeyHealth
 }
 
+type contentModerationRuntimeSnapshot struct {
+	riskControlEnabled bool
+	config             *ContentModerationConfig
+	keywordMatcher     *contentModerationKeywordMatcher
+	configDigest       [sha256.Size]byte
+	loadedAt           time.Time
+}
+
 type contentModerationTask struct {
 	input            ContentModerationCheckInput
 	content          ContentModerationInput
@@ -700,6 +715,7 @@ func (s *ContentModerationService) UpdateConfig(ctx context.Context, input Updat
 	if err := s.settingRepo.Set(ctx, SettingKeyContentModerationConfig, string(raw)); err != nil {
 		return nil, fmt.Errorf("save content moderation config: %w", err)
 	}
+	s.replaceRuntimeConfig(cfg, raw)
 	return s.configView(cfg), nil
 }
 
@@ -776,16 +792,7 @@ func (s *ContentModerationService) Check(ctx context.Context, input ContentModer
 			"protocol", input.Protocol)
 		return allow, nil
 	}
-	if !s.isRiskControlEnabled(ctx) {
-		slog.Info("content_moderation.skip_feature_disabled",
-			"user_id", input.UserID,
-			"api_key_id", input.APIKeyID,
-			"group_id", contentModerationLogGroupID(input.GroupID),
-			"endpoint", input.Endpoint,
-			"protocol", input.Protocol)
-		return allow, nil
-	}
-	cfg, err := s.loadConfig(ctx)
+	runtimeSnapshot, err := s.loadRuntimeSnapshot(ctx)
 	if err != nil {
 		slog.Warn("content_moderation.skip_config_load_failed",
 			"user_id", input.UserID,
@@ -796,6 +803,16 @@ func (s *ContentModerationService) Check(ctx context.Context, input ContentModer
 			"error", err)
 		return allow, nil
 	}
+	if !runtimeSnapshot.riskControlEnabled {
+		slog.Info("content_moderation.skip_feature_disabled",
+			"user_id", input.UserID,
+			"api_key_id", input.APIKeyID,
+			"group_id", contentModerationLogGroupID(input.GroupID),
+			"endpoint", input.Endpoint,
+			"protocol", input.Protocol)
+		return allow, nil
+	}
+	cfg := runtimeSnapshot.config
 	inGroupScope := cfg.includesGroup(input.GroupID)
 	inModelScope := cfg.includesModel(input.Model)
 	slog.Info("content_moderation.config_loaded",
@@ -885,7 +902,7 @@ func (s *ContentModerationService) Check(ctx context.Context, input ContentModer
 	hashText := content.Hash()
 	if cfg.Mode == ContentModerationModePreBlock {
 		if cfg.KeywordBlockingMode != ContentModerationKeywordModeAPIOnly && len(cfg.BlockedKeywords) > 0 {
-			if keyword, hit := matchBlockedKeyword(content.Text, cfg.BlockedKeywords); hit {
+			if keyword, hit := runtimeSnapshot.matchBlockedKeyword(content.Text); hit {
 				s.recordPreBlockSyncMetric(0, ContentModerationActionKeywordBlock)
 				slog.Info("content_moderation.keyword_block",
 					"user_id", input.UserID,
@@ -1178,12 +1195,13 @@ func (s *ContentModerationService) enqueueRecord(input ContentModerationCheckInp
 func (s *ContentModerationService) worker(id int) {
 	for {
 		ctx, cancel := context.WithTimeout(context.Background(), maxContentModerationTimeoutMS*time.Millisecond+10*time.Second)
-		cfg, err := s.loadConfig(ctx)
-		if err != nil || id >= cfg.WorkerCount {
+		runtimeSnapshot, err := s.loadRuntimeSnapshot(ctx)
+		if err != nil || runtimeSnapshot == nil || runtimeSnapshot.config == nil || id >= runtimeSnapshot.config.WorkerCount {
 			cancel()
 			time.Sleep(time.Second)
 			continue
 		}
+		cfg := runtimeSnapshot.config
 		task, ok := s.dequeueAsyncTask(ctx, time.Second)
 		if !ok {
 			cancel()
@@ -1438,15 +1456,18 @@ func (s *ContentModerationService) runCleanupOnce() {
 }
 
 func (s *ContentModerationService) loadConfig(ctx context.Context) (*ContentModerationConfig, error) {
-	cfg := defaultContentModerationConfig()
 	raw, err := s.settingRepo.GetValue(ctx, SettingKeyContentModerationConfig)
 	if err != nil {
 		if errors.Is(err, ErrSettingNotFound) {
-			cfg.normalize()
-			return cfg, nil
+			return parseContentModerationConfig("")
 		}
 		return nil, fmt.Errorf("get content moderation config: %w", err)
 	}
+	return parseContentModerationConfig(raw)
+}
+
+func parseContentModerationConfig(raw string) (*ContentModerationConfig, error) {
+	cfg := defaultContentModerationConfig()
 	if strings.TrimSpace(raw) == "" {
 		cfg.normalize()
 		return cfg, nil
@@ -1458,6 +1479,137 @@ func (s *ContentModerationService) loadConfig(ctx context.Context) (*ContentMode
 	return cfg, nil
 }
 
+func (s *ContentModerationService) loadRuntimeSnapshot(ctx context.Context) (*contentModerationRuntimeSnapshot, error) {
+	if s == nil || s.settingRepo == nil {
+		return nil, errors.New("content moderation setting repository unavailable")
+	}
+	now := time.Now()
+	if snapshot := s.runtimeSnapshot.Load(); snapshot != nil {
+		if now.Sub(snapshot.loadedAt) < s.runtimeSnapshotTTL() {
+			return snapshot, nil
+		}
+		s.triggerRuntimeSnapshotRefresh()
+		return snapshot, nil
+	}
+
+	s.runtimeRefreshMu.Lock()
+	defer s.runtimeRefreshMu.Unlock()
+	if snapshot := s.runtimeSnapshot.Load(); snapshot != nil {
+		return snapshot, nil
+	}
+	return s.refreshRuntimeSnapshot(ctx)
+}
+
+func (s *ContentModerationService) runtimeSnapshotTTL() time.Duration {
+	if s != nil && s.runtimeCacheTTL > 0 {
+		return s.runtimeCacheTTL
+	}
+	return contentModerationRuntimeCacheTTL
+}
+
+func (s *ContentModerationService) triggerRuntimeSnapshotRefresh() {
+	if s == nil || s.runtimeRefreshDeferred() || !s.runtimeRefreshMu.TryLock() {
+		return
+	}
+	if s.runtimeRefreshDeferred() {
+		s.runtimeRefreshMu.Unlock()
+		return
+	}
+	go func() {
+		defer s.runtimeRefreshMu.Unlock()
+		ctx, cancel := context.WithTimeout(context.Background(), contentModerationRuntimeRefreshTimeout)
+		defer cancel()
+		if _, err := s.refreshRuntimeSnapshot(ctx); err != nil {
+			s.runtimeRefreshRetryAt.Store(time.Now().Add(s.runtimeSnapshotTTL()).UnixNano())
+			slog.Warn("content_moderation.runtime_snapshot_refresh_failed", "error", err)
+		}
+	}()
+}
+
+func (s *ContentModerationService) runtimeRefreshDeferred() bool {
+	if s == nil {
+		return false
+	}
+	return time.Now().UnixNano() < s.runtimeRefreshRetryAt.Load()
+}
+
+func (s *ContentModerationService) refreshRuntimeSnapshot(ctx context.Context) (*contentModerationRuntimeSnapshot, error) {
+	values, err := s.settingRepo.GetMultiple(ctx, []string{
+		SettingKeyRiskControlEnabled,
+		SettingKeyContentModerationConfig,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("get content moderation runtime settings: %w", err)
+	}
+	rawConfig := values[SettingKeyContentModerationConfig]
+	configDigest := sha256.Sum256([]byte(rawConfig))
+	if current := s.runtimeSnapshot.Load(); current != nil && current.configDigest == configDigest {
+		snapshot := &contentModerationRuntimeSnapshot{
+			riskControlEnabled: values[SettingKeyRiskControlEnabled] == "true",
+			config:             current.config,
+			keywordMatcher:     current.keywordMatcher,
+			configDigest:       configDigest,
+			loadedAt:           time.Now(),
+		}
+		s.runtimeSnapshot.Store(snapshot)
+		s.runtimeRefreshRetryAt.Store(0)
+		return snapshot, nil
+	}
+	cfg, err := parseContentModerationConfig(rawConfig)
+	if err != nil {
+		return nil, err
+	}
+	snapshot := &contentModerationRuntimeSnapshot{
+		riskControlEnabled: values[SettingKeyRiskControlEnabled] == "true",
+		config:             cfg,
+		keywordMatcher:     newContentModerationKeywordMatcher(cfg.BlockedKeywords),
+		configDigest:       configDigest,
+		loadedAt:           time.Now(),
+	}
+	s.runtimeSnapshot.Store(snapshot)
+	s.runtimeRefreshRetryAt.Store(0)
+	return snapshot, nil
+}
+
+func (s *ContentModerationService) replaceRuntimeConfig(cfg *ContentModerationConfig, raw []byte) {
+	if s == nil || cfg == nil {
+		return
+	}
+	s.runtimeRefreshMu.Lock()
+	hasSnapshot := s.runtimeSnapshot.Load() != nil
+	s.runtimeRefreshMu.Unlock()
+	if !hasSnapshot {
+		return
+	}
+	config := cloneContentModerationConfig(cfg)
+	keywordMatcher := newContentModerationKeywordMatcher(cfg.BlockedKeywords)
+	configDigest := sha256.Sum256(raw)
+
+	s.runtimeRefreshMu.Lock()
+	defer s.runtimeRefreshMu.Unlock()
+	current := s.runtimeSnapshot.Load()
+	if current == nil {
+		return
+	}
+	s.runtimeSnapshot.Store(&contentModerationRuntimeSnapshot{
+		riskControlEnabled: current.riskControlEnabled,
+		config:             config,
+		keywordMatcher:     keywordMatcher,
+		configDigest:       configDigest,
+		loadedAt:           time.Now(),
+	})
+}
+
+func (s *contentModerationRuntimeSnapshot) matchBlockedKeyword(text string) (string, bool) {
+	if s == nil || s.config == nil {
+		return "", false
+	}
+	if s.keywordMatcher != nil {
+		return s.keywordMatcher.Match(text)
+	}
+	return matchBlockedKeyword(text, s.config.BlockedKeywords)
+}
+
 func (s *ContentModerationService) isRiskControlEnabled(ctx context.Context) bool {
 	raw, err := s.settingRepo.GetValue(ctx, SettingKeyRiskControlEnabled)
 	if err != nil {

backend/internal/service/content_moderation_keyword_matcher.go

@@ -0,0 +1,222 @@
+package service
+
+import (
+	"strings"
+)
+
+type contentModerationKeywordMatcher struct {
+	nodes           []contentModerationKeywordNode
+	edges           []contentModerationKeywordEdge
+	rootTransitions [256]int32
+	keywords        []string
+}
+
+type contentModerationKeywordNode struct {
+	failure     int32
+	bestKeyword int32
+	edgeStart   uint32
+	edgeCount   uint16
+}
+
+type contentModerationKeywordEdge struct {
+	target int32
+	label  byte
+}
+
+type contentModerationKeywordBuildEdge struct {
+	target      int32
+	nextSibling int32
+	label       byte
+}
+
+func newContentModerationKeywordMatcher(keywords []string) *contentModerationKeywordMatcher {
+	if len(keywords) == 0 {
+		return nil
+	}
+
+	buildNodes := []contentModerationKeywordNode{newContentModerationKeywordNode()}
+	buildEdges := make([]contentModerationKeywordBuildEdge, 0)
+	originalKeywords := append([]string(nil), keywords...)
+
+	for keywordIndex, keyword := range keywords {
+		if keyword == "" {
+			continue
+		}
+		state := int32(0)
+		for _, label := range []byte(strings.ToLower(keyword)) {
+			next := contentModerationKeywordBuildTransition(buildNodes, buildEdges, state, label)
+			if next < 0 {
+				next = int32(len(buildNodes))
+				buildNodes = append(buildNodes, newContentModerationKeywordNode())
+				buildEdges = append(buildEdges, contentModerationKeywordBuildEdge{
+					target:      next,
+					nextSibling: contentModerationKeywordBuildFirstEdge(buildNodes[state]),
+					label:       label,
+				})
+				buildNodes[state].edgeStart = uint32(len(buildEdges))
+			}
+			state = next
+		}
+		if current := buildNodes[state].bestKeyword; current < 0 || int32(keywordIndex) < current {
+			buildNodes[state].bestKeyword = int32(keywordIndex)
+		}
+	}
+
+	if len(buildNodes) == 1 {
+		return nil
+	}
+
+	queue := make([]int32, 0, len(buildNodes)-1)
+	var rootTransitions [256]int32
+	for edgeIndex := contentModerationKeywordBuildFirstEdge(buildNodes[0]); edgeIndex >= 0; edgeIndex = buildEdges[edgeIndex].nextSibling {
+		edge := buildEdges[edgeIndex]
+		rootTransitions[edge.label] = edge.target
+		queue = append(queue, edge.target)
+	}
+
+	for queueIndex := 0; queueIndex < len(queue); queueIndex++ {
+		state := queue[queueIndex]
+		for edgeIndex := contentModerationKeywordBuildFirstEdge(buildNodes[state]); edgeIndex >= 0; edgeIndex = buildEdges[edgeIndex].nextSibling {
+			edge := buildEdges[edgeIndex]
+			failure := buildNodes[state].failure
+			fallback := contentModerationKeywordBuildTransition(buildNodes, buildEdges, failure, edge.label)
+			for fallback < 0 && failure != 0 {
+				failure = buildNodes[failure].failure
+				fallback = contentModerationKeywordBuildTransition(buildNodes, buildEdges, failure, edge.label)
+			}
+			if fallback >= 0 {
+				buildNodes[edge.target].failure = fallback
+			}
+			buildNodes[edge.target].bestKeyword = minKeywordIndex(
+				buildNodes[edge.target].bestKeyword,
+				buildNodes[buildNodes[edge.target].failure].bestKeyword,
+			)
+			queue = append(queue, edge.target)
+		}
+	}
+
+	edges := make([]contentModerationKeywordEdge, 0, len(buildEdges))
+	var outgoing [256]contentModerationKeywordEdge
+	for nodeIndex := range buildNodes {
+		count := 0
+		for edgeIndex := contentModerationKeywordBuildFirstEdge(buildNodes[nodeIndex]); edgeIndex >= 0; edgeIndex = buildEdges[edgeIndex].nextSibling {
+			edge := buildEdges[edgeIndex]
+			outgoing[count] = contentModerationKeywordEdge{target: edge.target, label: edge.label}
+			count++
+		}
+		for index := 1; index < count; index++ {
+			current := outgoing[index]
+			insertAt := index
+			for insertAt > 0 && current.label < outgoing[insertAt-1].label {
+				outgoing[insertAt] = outgoing[insertAt-1]
+				insertAt--
+			}
+			outgoing[insertAt] = current
+		}
+		buildNodes[nodeIndex].edgeStart = uint32(len(edges))
+		buildNodes[nodeIndex].edgeCount = uint16(count)
+		edges = append(edges, outgoing[:count]...)
+	}
+
+	return &contentModerationKeywordMatcher{
+		nodes:           buildNodes,
+		edges:           edges,
+		rootTransitions: rootTransitions,
+		keywords:        originalKeywords,
+	}
+}
+
+func newContentModerationKeywordNode() contentModerationKeywordNode {
+	return contentModerationKeywordNode{bestKeyword: -1}
+}
+
+func contentModerationKeywordBuildFirstEdge(node contentModerationKeywordNode) int32 {
+	if node.edgeStart == 0 {
+		return -1
+	}
+	return int32(node.edgeStart - 1)
+}
+
+func contentModerationKeywordBuildTransition(
+	nodes []contentModerationKeywordNode,
+	edges []contentModerationKeywordBuildEdge,
+	state int32,
+	label byte,
+) int32 {
+	if state < 0 || int(state) >= len(nodes) {
+		return -1
+	}
+	for edgeIndex := contentModerationKeywordBuildFirstEdge(nodes[state]); edgeIndex >= 0; edgeIndex = edges[edgeIndex].nextSibling {
+		if edges[edgeIndex].label == label {
+			return edges[edgeIndex].target
+		}
+	}
+	return -1
+}
+
+func minKeywordIndex(left, right int32) int32 {
+	if left < 0 {
+		return right
+	}
+	if right < 0 || left < right {
+		return left
+	}
+	return right
+}
+
+func (m *contentModerationKeywordMatcher) Match(text string) (string, bool) {
+	if m == nil || text == "" || len(m.nodes) == 0 || len(m.keywords) == 0 {
+		return "", false
+	}
+	lower := strings.ToLower(text)
+	state := int32(0)
+	bestKeyword := int32(-1)
+	for index := 0; index < len(lower); index++ {
+		label := lower[index]
+		for {
+			next := m.next(state, label)
+			if next != 0 {
+				state = next
+				break
+			}
+			if state == 0 {
+				break
+			}
+			state = m.nodes[state].failure
+		}
+		bestKeyword = minKeywordIndex(bestKeyword, m.nodes[state].bestKeyword)
+		if bestKeyword == 0 {
+			return m.keywords[0], true
+		}
+	}
+	if bestKeyword < 0 || int(bestKeyword) >= len(m.keywords) {
+		return "", false
+	}
+	return m.keywords[bestKeyword], true
+}
+
+func (m *contentModerationKeywordMatcher) next(state int32, label byte) int32 {
+	if state == 0 {
+		return m.rootTransitions[label]
+	}
+	if state < 0 || int(state) >= len(m.nodes) {
+		return 0
+	}
+	node := m.nodes[state]
+	left := int(node.edgeStart)
+	right := left + int(node.edgeCount)
+	for left < right {
+		middle := left + (right-left)/2
+		edge := m.edges[middle]
+		if edge.label < label {
+			left = middle + 1
+			continue
+		}
+		right = middle
+	}
+	end := int(node.edgeStart) + int(node.edgeCount)
+	if left < end && m.edges[left].label == label {
+		return m.edges[left].target
+	}
+	return 0
+}

backend/internal/service/content_moderation_keyword_matcher_test.go

@@ -0,0 +1,59 @@
+package service
+
+import (
+	"math/rand"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestContentModerationKeywordMatcherMatchesLegacyBehavior(t *testing.T) {
+	tests := []struct {
+		name     string
+		text     string
+		keywords []string
+	}{
+		{name: "miss", text: "clean prompt", keywords: []string{"blocked", "secret"}},
+		{name: "case insensitive", text: "contains SECRET value", keywords: []string{"secret"}},
+		{name: "configured order wins", text: "early appears before later", keywords: []string{"later", "early"}},
+		{name: "overlap uses configured order", text: "abc", keywords: []string{"bc", "abc"}},
+		{name: "unicode", text: "这里包含敏感词和世界", keywords: []string{"世界", "敏感词"}},
+		{name: "duplicates", text: "duplicate", keywords: []string{"duplicate", "DUPLICATE"}},
+		{name: "empty entries", text: "blocked", keywords: []string{"", "blocked"}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			wantKeyword, wantHit := matchBlockedKeyword(tt.text, tt.keywords)
+			gotKeyword, gotHit := newContentModerationKeywordMatcher(tt.keywords).Match(tt.text)
+			require.Equal(t, wantHit, gotHit)
+			require.Equal(t, wantKeyword, gotKeyword)
+		})
+	}
+}
+
+func TestContentModerationKeywordMatcherRandomizedParity(t *testing.T) {
+	rng := rand.New(rand.NewSource(20260714))
+	const alphabet = "abcXYZ"
+	for iteration := 0; iteration < 1000; iteration++ {
+		keywords := make([]string, 1+rng.Intn(30))
+		for index := range keywords {
+			length := 1 + rng.Intn(8)
+			var value strings.Builder
+			for range length {
+				_ = value.WriteByte(alphabet[rng.Intn(len(alphabet))])
+			}
+			keywords[index] = value.String()
+		}
+		var text strings.Builder
+		for range 20 + rng.Intn(100) {
+			_ = text.WriteByte(alphabet[rng.Intn(len(alphabet))])
+		}
+
+		wantKeyword, wantHit := matchBlockedKeyword(text.String(), keywords)
+		gotKeyword, gotHit := newContentModerationKeywordMatcher(keywords).Match(text.String())
+		require.Equal(t, wantHit, gotHit, "iteration %d", iteration)
+		require.Equal(t, wantKeyword, gotKeyword, "iteration %d", iteration)
+	}
+}

backend/internal/service/content_moderation_runtime_cache_test.go

@@ -0,0 +1,451 @@
+package service
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+type contentModerationRuntimeSettingRepo struct {
+	mu               sync.Mutex
+	values           map[string]string
+	getValueCalls    int
+	getMultipleCalls int
+	getMultipleErr   error
+	getMultipleStart chan<- struct{}
+	getMultipleWait  <-chan struct{}
+}
+
+func (r *contentModerationRuntimeSettingRepo) Get(_ context.Context, key string) (*Setting, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	value, ok := r.values[key]
+	if !ok {
+		return nil, ErrSettingNotFound
+	}
+	return &Setting{Key: key, Value: value}, nil
+}
+
+func (r *contentModerationRuntimeSettingRepo) GetValue(_ context.Context, key string) (string, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.getValueCalls++
+	value, ok := r.values[key]
+	if !ok {
+		return "", ErrSettingNotFound
+	}
+	return value, nil
+}
+
+func (r *contentModerationRuntimeSettingRepo) Set(_ context.Context, key, value string) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.values == nil {
+		r.values = make(map[string]string)
+	}
+	r.values[key] = value
+	return nil
+}
+
+func (r *contentModerationRuntimeSettingRepo) GetMultiple(_ context.Context, keys []string) (map[string]string, error) {
+	r.mu.Lock()
+	r.getMultipleCalls++
+	if err := r.getMultipleErr; err != nil {
+		r.mu.Unlock()
+		return nil, err
+	}
+	out := make(map[string]string, len(keys))
+	for _, key := range keys {
+		if value, ok := r.values[key]; ok {
+			out[key] = value
+		}
+	}
+	start := r.getMultipleStart
+	wait := r.getMultipleWait
+	r.getMultipleStart = nil
+	r.getMultipleWait = nil
+	r.mu.Unlock()
+	if start != nil {
+		start <- struct{}{}
+	}
+	if wait != nil {
+		<-wait
+	}
+	return out, nil
+}
+
+func (r *contentModerationRuntimeSettingRepo) SetMultiple(_ context.Context, values map[string]string) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.values == nil {
+		r.values = make(map[string]string)
+	}
+	for key, value := range values {
+		r.values[key] = value
+	}
+	return nil
+}
+
+func (r *contentModerationRuntimeSettingRepo) GetAll(_ context.Context) (map[string]string, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	out := make(map[string]string, len(r.values))
+	for key, value := range r.values {
+		out[key] = value
+	}
+	return out, nil
+}
+
+func (r *contentModerationRuntimeSettingRepo) Delete(_ context.Context, key string) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	delete(r.values, key)
+	return nil
+}
+
+func (r *contentModerationRuntimeSettingRepo) calls() (getValue, getMultiple int) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.getValueCalls, r.getMultipleCalls
+}
+
+func (r *contentModerationRuntimeSettingRepo) failMultiple(err error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.getMultipleErr = err
+}
+
+func (r *contentModerationRuntimeSettingRepo) blockNextMultiple(start chan<- struct{}, wait <-chan struct{}) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.getMultipleStart = start
+	r.getMultipleWait = wait
+}
+
+func runtimeCacheTestConfig(t *testing.T, keywords ...string) string {
+	t.Helper()
+	cfg := defaultContentModerationConfig()
+	cfg.Enabled = true
+	cfg.Mode = ContentModerationModePreBlock
+	cfg.KeywordBlockingMode = ContentModerationKeywordModeKeywordOnly
+	cfg.BlockedKeywords = keywords
+	raw, err := json.Marshal(cfg)
+	require.NoError(t, err)
+	return string(raw)
+}
+
+func runtimeCacheTestService(repo *contentModerationRuntimeSettingRepo, ttl time.Duration) *ContentModerationService {
+	return &ContentModerationService{
+		settingRepo:     repo,
+		repo:            &contentModerationTestRepo{},
+		runtimeCacheTTL: ttl,
+	}
+}
+
+func runtimeCacheTestInput(text string) ContentModerationCheckInput {
+	return ContentModerationCheckInput{
+		Protocol: ContentModerationProtocolOpenAIChat,
+		Model:    "risk-cache-test",
+		Body:     []byte(`{"messages":[{"role":"user","content":"` + text + `"}]}`),
+	}
+}
+
+func TestContentModerationRuntimeSnapshotCachesSettings(t *testing.T) {
+	repo := &contentModerationRuntimeSettingRepo{values: map[string]string{
+		SettingKeyRiskControlEnabled:      "true",
+		SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "blocked"),
+	}}
+	svc := runtimeCacheTestService(repo, time.Hour)
+
+	for range 20 {
+		decision, err := svc.Check(context.Background(), runtimeCacheTestInput("clean prompt"))
+		require.NoError(t, err)
+		require.True(t, decision.Allowed)
+	}
+
+	getValue, getMultiple := repo.calls()
+	require.Zero(t, getValue)
+	require.Equal(t, 1, getMultiple)
+}
+
+func TestContentModerationRuntimeSnapshotUpdateConfigIsImmediate(t *testing.T) {
+	repo := &contentModerationRuntimeSettingRepo{values: map[string]string{
+		SettingKeyRiskControlEnabled:      "true",
+		SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "old-keyword"),
+	}}
+	svc := runtimeCacheTestService(repo, time.Hour)
+
+	decision, err := svc.Check(context.Background(), runtimeCacheTestInput("new-keyword"))
+	require.NoError(t, err)
+	require.True(t, decision.Allowed)
+
+	keywords := []string{"new-keyword"}
+	_, err = svc.UpdateConfig(context.Background(), UpdateContentModerationConfigInput{
+		BlockedKeywords: &keywords,
+	})
+	require.NoError(t, err)
+
+	decision, err = svc.Check(context.Background(), runtimeCacheTestInput("new-keyword"))
+	require.NoError(t, err)
+	require.True(t, decision.Blocked)
+
+	_, getMultiple := repo.calls()
+	require.Equal(t, 1, getMultiple)
+}
+
+func TestContentModerationRuntimeSnapshotUpdateWinsOverInitialLoad(t *testing.T) {
+	repo := &contentModerationRuntimeSettingRepo{values: map[string]string{
+		SettingKeyRiskControlEnabled:      "true",
+		SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "old-keyword"),
+	}}
+	svc := runtimeCacheTestService(repo, time.Hour)
+
+	refreshStarted := make(chan struct{}, 1)
+	releaseRefresh := make(chan struct{})
+	released := false
+	defer func() {
+		if !released {
+			close(releaseRefresh)
+		}
+	}()
+	repo.blockNextMultiple(refreshStarted, releaseRefresh)
+
+	initialCheckDone := make(chan error, 1)
+	go func() {
+		decision, err := svc.Check(context.Background(), runtimeCacheTestInput("clean prompt"))
+		if err == nil && (decision == nil || !decision.Allowed) {
+			err = errors.New("unexpected initial moderation decision")
+		}
+		initialCheckDone <- err
+	}()
+	require.Eventually(t, func() bool {
+		select {
+		case <-refreshStarted:
+			return true
+		default:
+			return false
+		}
+	}, time.Second, time.Millisecond)
+
+	updateDone := make(chan error, 1)
+	go func() {
+		keywords := []string{"new-keyword"}
+		_, updateErr := svc.UpdateConfig(context.Background(), UpdateContentModerationConfigInput{
+			BlockedKeywords: &keywords,
+		})
+		updateDone <- updateErr
+	}()
+	select {
+	case updateErr := <-updateDone:
+		require.NoError(t, updateErr)
+		t.Fatal("configuration update completed before the initial load released its lock")
+	case <-time.After(10 * time.Millisecond):
+	}
+
+	close(releaseRefresh)
+	released = true
+	require.NoError(t, <-initialCheckDone)
+	require.NoError(t, <-updateDone)
+
+	decision, err := svc.Check(context.Background(), runtimeCacheTestInput("new-keyword"))
+	require.NoError(t, err)
+	require.True(t, decision.Blocked)
+	decision, err = svc.Check(context.Background(), runtimeCacheTestInput("old-keyword"))
+	require.NoError(t, err)
+	require.True(t, decision.Allowed)
+}
+
+func TestContentModerationRuntimeSnapshotRefreshFailureKeepsStaleConfig(t *testing.T) {
+	repo := &contentModerationRuntimeSettingRepo{values: map[string]string{
+		SettingKeyRiskControlEnabled:      "true",
+		SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "blocked"),
+	}}
+	svc := runtimeCacheTestService(repo, time.Nanosecond)
+	input := runtimeCacheTestInput("blocked")
+
+	decision, err := svc.Check(context.Background(), input)
+	require.NoError(t, err)
+	require.True(t, decision.Blocked)
+
+	repo.failMultiple(errors.New("database unavailable"))
+	decision, err = svc.Check(context.Background(), input)
+	require.NoError(t, err)
+	require.True(t, decision.Blocked)
+	require.Eventually(t, func() bool {
+		_, calls := repo.calls()
+		return calls >= 2
+	}, time.Second, time.Millisecond)
+}
+
+func TestContentModerationRuntimeSnapshotRefreshFailureBacksOff(t *testing.T) {
+	repo := &contentModerationRuntimeSettingRepo{values: map[string]string{
+		SettingKeyRiskControlEnabled:      "true",
+		SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "blocked"),
+	}}
+	svc := runtimeCacheTestService(repo, time.Minute)
+	input := runtimeCacheTestInput("blocked")
+
+	decision, err := svc.Check(context.Background(), input)
+	require.NoError(t, err)
+	require.True(t, decision.Blocked)
+
+	current := svc.runtimeSnapshot.Load()
+	require.NotNil(t, current)
+	expired := *current
+	expired.loadedAt = time.Now().Add(-2 * time.Minute)
+	svc.runtimeSnapshot.Store(&expired)
+	repo.failMultiple(errors.New("database unavailable"))
+
+	decision, err = svc.Check(context.Background(), input)
+	require.NoError(t, err)
+	require.True(t, decision.Blocked)
+	require.Eventually(t, func() bool {
+		_, calls := repo.calls()
+		return calls == 2
+	}, time.Second, time.Millisecond)
+
+	for range 100 {
+		decision, err = svc.Check(context.Background(), input)
+		require.NoError(t, err)
+		require.True(t, decision.Blocked)
+	}
+	_, calls := repo.calls()
+	require.Equal(t, 2, calls)
+}
+
+func TestContentModerationRuntimeSnapshotRefreshReusesUnchangedMatcher(t *testing.T) {
+	repo := &contentModerationRuntimeSettingRepo{values: map[string]string{
+		SettingKeyRiskControlEnabled:      "true",
+		SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "blocked"),
+	}}
+	svc := runtimeCacheTestService(repo, time.Minute)
+	input := runtimeCacheTestInput("blocked")
+
+	decision, err := svc.Check(context.Background(), input)
+	require.NoError(t, err)
+	require.True(t, decision.Blocked)
+
+	current := svc.runtimeSnapshot.Load()
+	require.NotNil(t, current)
+	expired := *current
+	expired.loadedAt = time.Now().Add(-2 * time.Minute)
+	svc.runtimeSnapshot.Store(&expired)
+
+	decision, err = svc.Check(context.Background(), input)
+	require.NoError(t, err)
+	require.True(t, decision.Blocked)
+	require.Eventually(t, func() bool {
+		refreshed := svc.runtimeSnapshot.Load()
+		return refreshed != nil && refreshed.loadedAt.After(expired.loadedAt)
+	}, time.Second, time.Millisecond)
+
+	refreshed := svc.runtimeSnapshot.Load()
+	require.Same(t, current.config, refreshed.config)
+	require.Same(t, current.keywordMatcher, refreshed.keywordMatcher)
+	_, calls := repo.calls()
+	require.Equal(t, 2, calls)
+}
+
+func TestContentModerationRuntimeSnapshotUpdateWinsOverInFlightRefresh(t *testing.T) {
+	repo := &contentModerationRuntimeSettingRepo{values: map[string]string{
+		SettingKeyRiskControlEnabled:      "true",
+		SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "old-keyword"),
+	}}
+	svc := runtimeCacheTestService(repo, time.Minute)
+
+	decision, err := svc.Check(context.Background(), runtimeCacheTestInput("old-keyword"))
+	require.NoError(t, err)
+	require.True(t, decision.Blocked)
+
+	current := svc.runtimeSnapshot.Load()
+	require.NotNil(t, current)
+	expired := *current
+	expired.loadedAt = time.Now().Add(-2 * time.Minute)
+	svc.runtimeSnapshot.Store(&expired)
+
+	refreshStarted := make(chan struct{}, 1)
+	releaseRefresh := make(chan struct{})
+	repo.blockNextMultiple(refreshStarted, releaseRefresh)
+	decision, err = svc.Check(context.Background(), runtimeCacheTestInput("clean prompt"))
+	require.NoError(t, err)
+	require.True(t, decision.Allowed)
+	require.Eventually(t, func() bool {
+		select {
+		case <-refreshStarted:
+			return true
+		default:
+			return false
+		}
+	}, time.Second, time.Millisecond)
+
+	updateDone := make(chan error, 1)
+	go func() {
+		keywords := []string{"new-keyword"}
+		_, updateErr := svc.UpdateConfig(context.Background(), UpdateContentModerationConfigInput{
+			BlockedKeywords: &keywords,
+		})
+		updateDone <- updateErr
+	}()
+	select {
+	case updateErr := <-updateDone:
+		require.NoError(t, updateErr)
+		t.Fatal("configuration update completed before the in-flight refresh released its lock")
+	case <-time.After(10 * time.Millisecond):
+	}
+
+	close(releaseRefresh)
+	require.NoError(t, <-updateDone)
+	decision, err = svc.Check(context.Background(), runtimeCacheTestInput("new-keyword"))
+	require.NoError(t, err)
+	require.True(t, decision.Blocked)
+	decision, err = svc.Check(context.Background(), runtimeCacheTestInput("old-keyword"))
+	require.NoError(t, err)
+	require.True(t, decision.Allowed)
+}
+
+func TestContentModerationRuntimeSnapshotConcurrentReadAndReplace(t *testing.T) {
+	repo := &contentModerationRuntimeSettingRepo{values: map[string]string{
+		SettingKeyRiskControlEnabled:      "true",
+		SettingKeyContentModerationConfig: runtimeCacheTestConfig(t, "blocked-0"),
+	}}
+	svc := runtimeCacheTestService(repo, time.Hour)
+	_, err := svc.Check(context.Background(), runtimeCacheTestInput("clean prompt"))
+	require.NoError(t, err)
+
+	var wg sync.WaitGroup
+	errs := make(chan error, 8)
+	for range 8 {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for range 100 {
+				decision, checkErr := svc.Check(context.Background(), runtimeCacheTestInput("clean prompt"))
+				if checkErr != nil {
+					errs <- checkErr
+					return
+				}
+				if decision == nil || !decision.Allowed {
+					errs <- errors.New("unexpected moderation decision")
+					return
+				}
+			}
+		}()
+	}
+	for i := 1; i <= 20; i++ {
+		keywords := []string{"blocked-" + time.Duration(i).String()}
+		_, err := svc.UpdateConfig(context.Background(), UpdateContentModerationConfigInput{
+			BlockedKeywords: &keywords,
+		})
+		require.NoError(t, err)
+	}
+	wg.Wait()
+	close(errs)
+	for err := range errs {
+		require.NoError(t, err)
+	}
+}