fix(openai-ws): remember rejected encrypted content across turns

invalid_encrypted_content 恢复此前只清理当次请求体：同一失效密文随客户
端维护的会话历史在后续每一轮重新出现，反复触发"整包被拒→剥离→重试/
重连"（#6382 生产观测 2h 内 251 次 recovery、273 次 WS 重连），既是
QoS 放大器也是内存压力放大器。

按会话记录被上游拒绝过的 encrypted_content SHA-256 摘要（lineage）：
- OpenAIWSStateStore 新增会话级摘要绑定，TTL 复用 session sticky，
  单会话 512 摘要为存储自保护，超出仅退化为既有单次 recovery
- 写点覆盖 forward WSv2 recovery、forward 非 WS 400 重试、ingress
  error 事件、bridge 400/流内 error；读点在 forward 进场与 ingress/
  bridge turn 循环入口按摘要预剥离，新密文摘要不同不受影响
- 只收剥离端可处理的项类型（reasoning/compaction/compaction_summary），
  避免记入永远剥不掉的摘要后每轮空转解码
- replay 历史序列同步剥离，避免与已剥离的当前 input 项错位导致
  prefix 复用失配

含端到端集成测试：turn1 上游拒绝即记录，turn2 上游收到的请求不再携带
失效密文且 reasoning 骨架保留。

backend/internal/service/openai_encrypted_content_lineage.go

@@ -0,0 +1,353 @@
+package service
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/tidwall/gjson"
+)
+
+// invalid_encrypted_content 失效密文 lineage。
+//
+// 上游判定某轮请求中的加密 reasoning/compaction 项不可解后，同一失效密文会随
+// 客户端维护的会话历史在后续每一轮重新出现，重复触发"整包被拒→剥离→重试/
+// 重连"。这里按会话记录已被上游拒绝过的 encrypted_content 摘要
+// （OpenAIWSStateStore，带 TTL 与容量自保护）；后续请求进场时仅剥离摘要命中
+// 的项，新生成的密文摘要不同，不会被误删。
+
+const openAIWSFallbackReasonInvalidEncryptedContent = "invalid_encrypted_content"
+
+// openAIWSIngressSessionHashContextKey 在 gin context 中携带 ingress 会话哈希，
+// 供 HTTP bridge turn 内的 lineage 记录复用同一会话键。
+const openAIWSIngressSessionHashContextKey = "openai_ws_ingress_session_hash"
+
+func openAIEncryptedContentDigest(encrypted string) string {
+	sum := sha256.Sum256([]byte(encrypted))
+	return hex.EncodeToString(sum[:])
+}
+
+// openAIEncryptedLineageItemType 限定 lineage 覆盖的项类型，与剥离端
+// sanitizeEncryptedReasoningInputItem 能处理的类型对称；其他类型即使携带
+// encrypted_content 也不收摘要，避免记入永远剥不掉的摘要后每轮空转解码。
+func openAIEncryptedLineageItemType(itemType string) bool {
+	switch strings.TrimSpace(itemType) {
+	case "reasoning", "compaction", "compaction_summary":
+		return true
+	default:
+		return false
+	}
+}
+
+// collectOpenAIEncryptedContentDigestsRaw 收集 input 中 lineage 覆盖类型所携
+// 密文的摘要。payload 须遵守 replay 所有权不变式（零拷贝视图解析）。
+func collectOpenAIEncryptedContentDigestsRaw(payload []byte) []string {
+	if len(payload) == 0 {
+		return nil
+	}
+	input := gjson.Get(openAIWSPayloadStringView(payload), "input")
+	if !input.Exists() {
+		return nil
+	}
+	var digests []string
+	appendItem := func(item gjson.Result) {
+		if !openAIEncryptedLineageItemType(item.Get("type").String()) {
+			return
+		}
+		encrypted := item.Get("encrypted_content")
+		if encrypted.Type == gjson.String && encrypted.String() != "" {
+			digests = append(digests, openAIEncryptedContentDigest(encrypted.String()))
+		}
+	}
+	if input.IsArray() {
+		input.ForEach(func(_, item gjson.Result) bool {
+			appendItem(item)
+			return true
+		})
+		return digests
+	}
+	if input.IsObject() {
+		appendItem(input)
+	}
+	return digests
+}
+
+// stripOpenAIInvalidEncryptedContentItems 剥离 input 中摘要命中 invalid 的密文
+// 项，剥离语义与 trimOpenAIEncryptedReasoningItems 对齐：reasoning 仅剥
+// encrypted_content（连带 content:null 清理与空骨架删除），compaction/
+// compaction_summary 整项删除。返回被改写的项数。
+func stripOpenAIInvalidEncryptedContentItems(reqBody map[string]any, invalid map[string]struct{}) int {
+	if len(reqBody) == 0 || len(invalid) == 0 {
+		return 0
+	}
+	inputValue, has := reqBody["input"]
+	if !has {
+		return 0
+	}
+	stripped := 0
+	stripItem := func(item any) (next any, changed bool, keep bool) {
+		inputItem, ok := item.(map[string]any)
+		if !ok {
+			return item, false, true
+		}
+		encrypted, ok := inputItem["encrypted_content"].(string)
+		if !ok || encrypted == "" {
+			return item, false, true
+		}
+		if _, hit := invalid[openAIEncryptedContentDigest(encrypted)]; !hit {
+			return item, false, true
+		}
+		return sanitizeEncryptedReasoningInputItem(item)
+	}
+	switch input := inputValue.(type) {
+	case []any:
+		filtered := input[:0]
+		for _, item := range input {
+			nextItem, changed, keep := stripItem(item)
+			if changed {
+				stripped++
+			}
+			if !keep {
+				continue
+			}
+			filtered = append(filtered, nextItem)
+		}
+		if stripped == 0 {
+			return 0
+		}
+		if len(filtered) == 0 {
+			delete(reqBody, "input")
+			return stripped
+		}
+		reqBody["input"] = filtered
+		return stripped
+	case map[string]any:
+		nextItem, changed, keep := stripItem(input)
+		if !changed {
+			return 0
+		}
+		if !keep {
+			delete(reqBody, "input")
+			return 1
+		}
+		nextMap, ok := nextItem.(map[string]any)
+		if !ok {
+			return 0
+		}
+		reqBody["input"] = nextMap
+		return 1
+	default:
+		return 0
+	}
+}
+
+// openAIRawPayloadHasInvalidEncryptedContent 探测 raw payload 是否携带命中
+// invalid 摘要的密文项；零命中的常态路径不做任何改写。payload 须遵守 replay
+// 所有权不变式（零拷贝视图解析）。
+func openAIRawPayloadHasInvalidEncryptedContent(payload []byte, invalid map[string]struct{}) bool {
+	if len(payload) == 0 || len(invalid) == 0 {
+		return false
+	}
+	input := gjson.Get(openAIWSPayloadStringView(payload), "input")
+	if !input.Exists() {
+		return false
+	}
+	hit := false
+	checkItem := func(item gjson.Result) bool {
+		encrypted := item.Get("encrypted_content")
+		if encrypted.Type != gjson.String || encrypted.String() == "" {
+			return false
+		}
+		_, matched := invalid[openAIEncryptedContentDigest(encrypted.String())]
+		return matched
+	}
+	if input.IsArray() {
+		input.ForEach(func(_, item gjson.Result) bool {
+			if checkItem(item) {
+				hit = true
+				return false
+			}
+			return true
+		})
+		return hit
+	}
+	if input.IsObject() {
+		return checkItem(input)
+	}
+	return false
+}
+
+// stripOpenAIInvalidEncryptedContentFromReplayItems 对 replay 历史序列做同款
+// 剥离。有改动时返回新头数组，命中项以重写后的新正文整体替换（遵守 replay
+// 所有权不变式，原正文不被修改）；未命中时原样返回。
+func stripOpenAIInvalidEncryptedContentFromReplayItems(items []json.RawMessage, invalid map[string]struct{}) ([]json.RawMessage, int) {
+	if len(items) == 0 || len(invalid) == 0 {
+		return items, 0
+	}
+	hit := false
+	for _, item := range items {
+		encrypted := gjson.Get(openAIWSPayloadStringView(item), "encrypted_content")
+		if encrypted.Type != gjson.String || encrypted.String() == "" {
+			continue
+		}
+		if _, matched := invalid[openAIEncryptedContentDigest(encrypted.String())]; matched {
+			hit = true
+			break
+		}
+	}
+	if !hit {
+		return items, 0
+	}
+	stripped := 0
+	next := make([]json.RawMessage, 0, len(items))
+	for _, item := range items {
+		var decoded map[string]any
+		if err := decodeOpenAIJSONUseNumber(item, &decoded); err != nil {
+			next = append(next, item)
+			continue
+		}
+		encrypted, ok := decoded["encrypted_content"].(string)
+		if !ok || encrypted == "" {
+			next = append(next, item)
+			continue
+		}
+		if _, matched := invalid[openAIEncryptedContentDigest(encrypted)]; !matched {
+			next = append(next, item)
+			continue
+		}
+		nextItem, changed, keep := sanitizeEncryptedReasoningInputItem(decoded)
+		if !changed {
+			next = append(next, item)
+			continue
+		}
+		stripped++
+		if !keep {
+			continue
+		}
+		rebuilt, err := marshalOpenAIUpstreamJSON(nextItem)
+		if err != nil {
+			next = append(next, item)
+			stripped--
+			continue
+		}
+		next = append(next, json.RawMessage(rebuilt))
+	}
+	if stripped == 0 {
+		return items, 0
+	}
+	return next, stripped
+}
+
+// stripOpenAIInvalidEncryptedContentRaw 是 raw JSON 版剥离：先零成本探测，命中
+// 才解码改写重编码。返回改写后的 payload 与剥离项数；未命中时原样返回。
+func stripOpenAIInvalidEncryptedContentRaw(payload []byte, invalid map[string]struct{}) ([]byte, int, error) {
+	if !openAIRawPayloadHasInvalidEncryptedContent(payload, invalid) {
+		return payload, 0, nil
+	}
+	var decoded map[string]any
+	if err := decodeOpenAIJSONUseNumber(payload, &decoded); err != nil {
+		return payload, 0, err
+	}
+	stripped := stripOpenAIInvalidEncryptedContentItems(decoded, invalid)
+	if stripped == 0 {
+		return payload, 0, nil
+	}
+	rebuilt, err := marshalOpenAIUpstreamJSON(decoded)
+	if err != nil {
+		return payload, 0, err
+	}
+	return rebuilt, stripped, nil
+}
+
+// markOpenAIWSInvalidEncryptedContentLineage 把本次被上游拒绝的密文摘要写入
+// 会话 lineage。digests 须在剥离前收集。
+func (s *OpenAIGatewayService) markOpenAIWSInvalidEncryptedContentLineage(groupID int64, sessionHash string, digests []string) {
+	if s == nil || len(digests) == 0 || strings.TrimSpace(sessionHash) == "" {
+		return
+	}
+	stateStore := s.getOpenAIWSStateStore()
+	if stateStore == nil {
+		return
+	}
+	stateStore.MarkSessionInvalidEncryptedContent(groupID, sessionHash, digests, s.openAIWSSessionStickyTTL())
+}
+
+// sessionInvalidEncryptedContentDigests 返回会话已知失效密文摘要；全局无记录
+// 时（常态）零成本返回 nil。
+func (s *OpenAIGatewayService) sessionInvalidEncryptedContentDigests(groupID int64, sessionHash string) map[string]struct{} {
+	if s == nil || strings.TrimSpace(sessionHash) == "" {
+		return nil
+	}
+	stateStore := s.getOpenAIWSStateStore()
+	if stateStore == nil || !stateStore.HasAnySessionInvalidEncryptedContent() {
+		return nil
+	}
+	return stateStore.GetSessionInvalidEncryptedContentDigests(groupID, sessionHash)
+}
+
+// openAIWSLineageSessionHashFromContext 取 lineage 会话键：优先 ingress 循环
+// 写入的会话哈希（与读取侧同键），否则按请求体派生。
+func (s *OpenAIGatewayService) openAIWSLineageSessionHashFromContext(c *gin.Context, body []byte) string {
+	if c != nil {
+		if fromCtx := strings.TrimSpace(c.GetString(openAIWSIngressSessionHashContextKey)); fromCtx != "" {
+			return fromCtx
+		}
+	}
+	return s.GenerateSessionHash(c, body)
+}
+
+// markOpenAIWSInvalidEncryptedContentLineageFromPayload 在上游以
+// invalid_encrypted_content 拒绝 payload 时记录其密文摘要并输出观测日志。
+func (s *OpenAIGatewayService) markOpenAIWSInvalidEncryptedContentLineageFromPayload(
+	c *gin.Context,
+	payload []byte,
+	logKey string,
+	accountID int64,
+	turn int,
+) {
+	digests := collectOpenAIEncryptedContentDigestsRaw(payload)
+	if len(digests) == 0 {
+		return
+	}
+	s.markOpenAIWSInvalidEncryptedContentLineage(
+		getOpenAIGroupIDFromContext(c),
+		s.openAIWSLineageSessionHashFromContext(c, payload),
+		digests,
+	)
+	logOpenAIWSModeInfo("%s account_id=%d turn=%d digests=%d", logKey, accountID, turn, len(digests))
+}
+
+// stripSessionInvalidEncryptedContentLogged 对 payload 执行会话失效密文剥离并
+// 输出观测日志（logKey / logKey+"_skip"），返回（可能已替换的）payload 与剥离
+// 项数；未命中或剥离失败时原样返回。
+func (s *OpenAIGatewayService) stripSessionInvalidEncryptedContentLogged(
+	payload []byte,
+	invalid map[string]struct{},
+	logKey string,
+	accountID int64,
+	turn int,
+) ([]byte, int) {
+	strippedPayload, strippedCount, stripErr := stripOpenAIInvalidEncryptedContentRaw(payload, invalid)
+	if stripErr != nil {
+		logOpenAIWSModeInfo(
+			"%s_skip account_id=%d turn=%d reason=strip_error cause=%s",
+			logKey,
+			accountID,
+			turn,
+			truncateOpenAIWSLogValue(stripErr.Error(), openAIWSLogValueMaxLen),
+		)
+		return payload, 0
+	}
+	if strippedCount > 0 {
+		logOpenAIWSModeInfo(
+			"%s account_id=%d turn=%d stripped_items=%d",
+			logKey,
+			accountID,
+			turn,
+			strippedCount,
+		)
+	}
+	return strippedPayload, strippedCount
+}

backend/internal/service/openai_encrypted_content_lineage_test.go

@@ -0,0 +1,216 @@
+package service
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestCollectOpenAIEncryptedContentDigestsRaw(t *testing.T) {
+	t.Parallel()
+
+	digests := collectOpenAIEncryptedContentDigestsRaw([]byte(`{"input":[
+		{"type":"reasoning","encrypted_content":"cipher-a"},
+		{"type":"compaction","encrypted_content":"cipher-b"},
+		{"type":"input_text","text":"hello"},
+		{"type":"reasoning"},
+		{"type":"message","encrypted_content":"cipher-alien"}
+	]}`))
+	// 剥离端处理不了的类型（message）不收摘要，否则该摘要永远命中却剥不掉。
+	require.Equal(t, []string{
+		openAIEncryptedContentDigest("cipher-a"),
+		openAIEncryptedContentDigest("cipher-b"),
+	}, digests)
+
+	require.Nil(t, collectOpenAIEncryptedContentDigestsRaw([]byte(`{"model":"gpt-5"}`)))
+}
+
+func TestStripOpenAIInvalidEncryptedContentItems(t *testing.T) {
+	t.Parallel()
+
+	invalid := map[string]struct{}{
+		openAIEncryptedContentDigest("stale-cipher"): {},
+	}
+
+	t.Run("only_strips_matching_items", func(t *testing.T) {
+		t.Parallel()
+		reqBody := map[string]any{
+			"input": []any{
+				map[string]any{"type": "reasoning", "id": "rs_1", "encrypted_content": "stale-cipher", "summary": []any{}},
+				map[string]any{"type": "reasoning", "id": "rs_2", "encrypted_content": "fresh-cipher"},
+				map[string]any{"type": "input_text", "text": "hello"},
+			},
+		}
+		stripped := stripOpenAIInvalidEncryptedContentItems(reqBody, invalid)
+		require.Equal(t, 1, stripped)
+		input, ok := reqBody["input"].([]any)
+		require.True(t, ok)
+		require.Len(t, input, 3)
+		first, ok := input[0].(map[string]any)
+		require.True(t, ok)
+		_, hasEncrypted := first["encrypted_content"]
+		require.False(t, hasEncrypted, "命中项应剥离 encrypted_content")
+		require.Equal(t, "rs_1", first["id"], "reasoning 骨架应保留")
+		second, ok := input[1].(map[string]any)
+		require.True(t, ok)
+		require.Equal(t, "fresh-cipher", second["encrypted_content"], "新密文不得误删")
+	})
+
+	t.Run("compaction_removed_entirely", func(t *testing.T) {
+		t.Parallel()
+		reqBody := map[string]any{
+			"input": []any{
+				map[string]any{"type": "compaction", "encrypted_content": "stale-cipher"},
+				map[string]any{"type": "input_text", "text": "hello"},
+			},
+		}
+		stripped := stripOpenAIInvalidEncryptedContentItems(reqBody, invalid)
+		require.Equal(t, 1, stripped)
+		input, ok := reqBody["input"].([]any)
+		require.True(t, ok)
+		require.Len(t, input, 1)
+		survivor, ok := input[0].(map[string]any)
+		require.True(t, ok)
+		require.Equal(t, "input_text", survivor["type"])
+	})
+
+	t.Run("no_hit_no_change", func(t *testing.T) {
+		t.Parallel()
+		reqBody := map[string]any{
+			"input": []any{
+				map[string]any{"type": "reasoning", "encrypted_content": "fresh-cipher"},
+			},
+		}
+		require.Zero(t, stripOpenAIInvalidEncryptedContentItems(reqBody, invalid))
+	})
+}
+
+func TestStripOpenAIInvalidEncryptedContentFromReplayItems(t *testing.T) {
+	t.Parallel()
+
+	invalid := map[string]struct{}{
+		openAIEncryptedContentDigest("stale-cipher"): {},
+	}
+
+	t.Run("rewrites_hit_items_and_shares_the_rest", func(t *testing.T) {
+		t.Parallel()
+		original := json.RawMessage(`{"type":"reasoning","id":"rs_1","encrypted_content":"stale-cipher"}`)
+		clean := json.RawMessage(`{"type":"input_text","text":"hello"}`)
+		compaction := json.RawMessage(`{"type":"compaction","encrypted_content":"stale-cipher"}`)
+		items := []json.RawMessage{original, clean, compaction}
+
+		next, stripped := stripOpenAIInvalidEncryptedContentFromReplayItems(items, invalid)
+		require.Equal(t, 2, stripped)
+		require.Len(t, next, 2, "compaction 命中项应整项删除")
+		var rewritten map[string]any
+		require.NoError(t, json.Unmarshal(next[0], &rewritten))
+		require.Equal(t, "rs_1", rewritten["id"])
+		_, hasEncrypted := rewritten["encrypted_content"]
+		require.False(t, hasEncrypted)
+		// 未命中项正文共享；原命中项正文不被修改（整体替换语义）。
+		require.Same(t, &clean[0], &next[1][0])
+		require.Contains(t, string(original), "stale-cipher")
+	})
+
+	t.Run("miss_returns_original_header", func(t *testing.T) {
+		t.Parallel()
+		items := []json.RawMessage{json.RawMessage(`{"type":"reasoning","encrypted_content":"fresh-cipher"}`)}
+		next, stripped := stripOpenAIInvalidEncryptedContentFromReplayItems(items, invalid)
+		require.Zero(t, stripped)
+		require.Same(t, &items[0][0], &next[0][0])
+	})
+}
+
+func TestStripOpenAIInvalidEncryptedContentRaw(t *testing.T) {
+	t.Parallel()
+
+	invalid := map[string]struct{}{
+		openAIEncryptedContentDigest("stale-cipher"): {},
+	}
+	payload := []byte(`{"model":"gpt-5","input":[
+		{"type":"reasoning","id":"rs_1","encrypted_content":"stale-cipher"},
+		{"type":"reasoning","id":"rs_2","encrypted_content":"fresh-cipher"},
+		{"type":"input_text","text":"hello"}
+	],"stream":true}`)
+
+	stripped, count, err := stripOpenAIInvalidEncryptedContentRaw(payload, invalid)
+	require.NoError(t, err)
+	require.Equal(t, 1, count)
+	var decoded map[string]any
+	require.NoError(t, json.Unmarshal(stripped, &decoded))
+	require.Equal(t, "gpt-5", decoded["model"])
+	input, ok := decoded["input"].([]any)
+	require.True(t, ok)
+	require.Len(t, input, 3)
+	first, ok := input[0].(map[string]any)
+	require.True(t, ok)
+	_, hasEncrypted := first["encrypted_content"]
+	require.False(t, hasEncrypted)
+	second, ok := input[1].(map[string]any)
+	require.True(t, ok)
+	require.Equal(t, "fresh-cipher", second["encrypted_content"])
+
+	t.Run("miss_fast_path_returns_original", func(t *testing.T) {
+		t.Parallel()
+		cleanPayload := []byte(`{"input":[{"type":"reasoning","encrypted_content":"fresh-cipher"}]}`)
+		unchanged, count, err := stripOpenAIInvalidEncryptedContentRaw(cleanPayload, invalid)
+		require.NoError(t, err)
+		require.Zero(t, count)
+		require.Same(t, &cleanPayload[0], &unchanged[0], "未命中时应原样返回，不重编码")
+	})
+}
+
+func TestOpenAIWSStateStoreInvalidEncryptedContentLineage(t *testing.T) {
+	t.Parallel()
+
+	t.Run("mark_get_roundtrip_with_merge", func(t *testing.T) {
+		t.Parallel()
+		store := NewOpenAIWSStateStore(nil)
+		require.False(t, store.HasAnySessionInvalidEncryptedContent())
+		require.Nil(t, store.GetSessionInvalidEncryptedContentDigests(1, "session-a"))
+
+		store.MarkSessionInvalidEncryptedContent(1, "session-a", []string{"d1", "d2"}, time.Minute)
+		store.MarkSessionInvalidEncryptedContent(1, "session-a", []string{"d2", "d3"}, time.Minute)
+		require.True(t, store.HasAnySessionInvalidEncryptedContent())
+
+		digests := store.GetSessionInvalidEncryptedContentDigests(1, "session-a")
+		require.Len(t, digests, 3)
+		for _, digest := range []string{"d1", "d2", "d3"} {
+			require.Contains(t, digests, digest)
+		}
+		// 组隔离：另一组同名会话不可见。
+		require.Nil(t, store.GetSessionInvalidEncryptedContentDigests(2, "session-a"))
+		// 返回值是拷贝，调用方修改不得影响存储。
+		digests["d4"] = struct{}{}
+		require.Len(t, store.GetSessionInvalidEncryptedContentDigests(1, "session-a"), 3)
+	})
+
+	t.Run("expired_binding_not_returned", func(t *testing.T) {
+		t.Parallel()
+		raw := NewOpenAIWSStateStore(nil)
+		raw.MarkSessionInvalidEncryptedContent(1, "session-b", []string{"d1"}, time.Minute)
+		store, ok := raw.(*defaultOpenAIWSStateStore)
+		require.True(t, ok)
+		store.sessionInvalidEncryptedMu.Lock()
+		binding := store.sessionInvalidEncrypted["1:session-b"]
+		binding.expiresAt = time.Now().Add(-time.Second)
+		store.sessionInvalidEncrypted["1:session-b"] = binding
+		store.sessionInvalidEncryptedMu.Unlock()
+		require.Nil(t, store.GetSessionInvalidEncryptedContentDigests(1, "session-b"))
+	})
+
+	t.Run("per_session_capacity_degrades_gracefully", func(t *testing.T) {
+		t.Parallel()
+		store := NewOpenAIWSStateStore(nil)
+		oversized := make([]string, 0, openAIWSInvalidEncryptedDigestsPerSession+10)
+		for i := range openAIWSInvalidEncryptedDigestsPerSession + 10 {
+			oversized = append(oversized, openAIEncryptedContentDigest(fmt.Sprintf("cipher-%d", i)))
+		}
+		store.MarkSessionInvalidEncryptedContent(1, "session-c", oversized, time.Minute)
+		digests := store.GetSessionInvalidEncryptedContentDigests(1, "session-c")
+		require.Len(t, digests, openAIWSInvalidEncryptedDigestsPerSession)
+	})
+}

backend/internal/service/openai_gateway_forward.go

@@ -705,6 +705,26 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 		requestView = newOpenAIRequestView(body)
 		reqBody = nil
 	}
+	// 剥离本会话已被上游判定失效的加密项（invalid_encrypted_content lineage），
+	// 阻断同一失效密文随客户端历史在每一轮重复触发"被拒→剥离→重试/重连"。
+	// lineage 会话键统一按进场形态的 body 派生：后续重试可能改写 body，
+	// 延迟计算会与下一请求的进场键漂移。
+	lineageGroupID := getOpenAIGroupIDFromContext(c)
+	lineageEntryBody := body
+	lineageSessionHash := ""
+	if stateStore := s.getOpenAIWSStateStore(); stateStore != nil && stateStore.HasAnySessionInvalidEncryptedContent() {
+		lineageSessionHash = s.GenerateSessionHash(c, body)
+		if invalidDigests := stateStore.GetSessionInvalidEncryptedContentDigests(lineageGroupID, lineageSessionHash); len(invalidDigests) > 0 {
+			strippedBody, strippedCount := s.stripSessionInvalidEncryptedContentLogged(
+				body, invalidDigests, "invalid_encrypted_lineage_strip", account.ID, 0,
+			)
+			if strippedCount > 0 {
+				body = strippedBody
+				requestView = newOpenAIRequestView(body)
+				reqBody = nil
+			}
+		}
+	}
 	imageBillingModel := ""
 	imageSizeTier := ""
 	imageInputSize := ""
@@ -792,6 +812,10 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 			if wsInvalidEncryptedContentRecoveryTried {
 				return false
 			}
+			// 写入 lineage 后，同一失效密文在后续 turn 进场时被预剥离，不再重复
+			// 触发上游拒绝与重连。摘要取自进场形态的 body（密文项只可能来自
+			// 客户端进场请求，重复摘要幂等）。
+			invalidDigests := collectOpenAIEncryptedContentDigestsRaw(lineageEntryBody)
 			removedReasoningItems := trimOpenAIEncryptedReasoningItems(wsReqBody)
 			if !removedReasoningItems {
 				logOpenAIWSModeInfo(
@@ -801,6 +825,12 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 				)
 				return false
 			}
+			if len(invalidDigests) > 0 {
+				if lineageSessionHash == "" {
+					lineageSessionHash = s.GenerateSessionHash(c, lineageEntryBody)
+				}
+				s.markOpenAIWSInvalidEncryptedContentLineage(lineageGroupID, lineageSessionHash, invalidDigests)
+			}
 			previousResponseID := openAIWSPayloadString(wsReqBody, "previous_response_id")
 			hasFunctionCallOutput := HasFunctionCallOutput(wsReqBody)
 			if previousResponseID != "" && !hasFunctionCallOutput {
@@ -1052,11 +1082,18 @@ func (s *OpenAIGatewayService) Forward(ctx context.Context, c *gin.Context, acco
 				if decodeErr != nil {
 					return nil, decodeErr
 				}
+				invalidDigests := collectOpenAIEncryptedContentDigestsRaw(lineageEntryBody)
 				if trimOpenAIEncryptedReasoningItems(decoded) {
 					body, err = marshalOpenAIUpstreamJSON(decoded)
 					if err != nil {
 						return nil, fmt.Errorf("serialize invalid_encrypted_content retry body: %w", err)
 					}
+					if len(invalidDigests) > 0 {
+						if lineageSessionHash == "" {
+							lineageSessionHash = s.GenerateSessionHash(c, lineageEntryBody)
+						}
+						s.markOpenAIWSInvalidEncryptedContentLineage(lineageGroupID, lineageSessionHash, invalidDigests)
+					}
 					httpInvalidEncryptedContentRetryTried = true
 					rejectedFieldRetryState.remember(body)
 					logger.LegacyPrintf("service.openai_gateway", "[OpenAI] Retrying non-WSv2 request once after invalid_encrypted_content (account: %s)", account.Name)

backend/internal/service/openai_ws_forwarder_ingress.go

@@ -581,6 +581,26 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient(
 			if turnState != "" && c != nil && c.Request != nil {
 				c.Request.Header.Set(openAIWSTurnStateHeader, turnState)
 			}
+			if c != nil && sessionHash != "" {
+				c.Set(openAIWSIngressSessionHashContextKey, sessionHash)
+			}
+			// 剥离本会话已知失效的加密项，阻断同一失效密文随历史反复触发上游拒绝。
+			// 历史序列须同步剥离，否则与已剥离的当前 input 项错位，prefix 复用失配。
+			if invalidDigests := s.sessionInvalidEncryptedContentDigests(groupID, sessionHash); len(invalidDigests) > 0 {
+				strippedPayload, strippedCount := s.stripSessionInvalidEncryptedContentLogged(
+					currentBridgePayload.payloadRaw, invalidDigests, "ingress_ws_http_bridge_invalid_encrypted_lineage_strip", account.ID, turn,
+				)
+				if strippedCount > 0 {
+					currentBridgePayload.payloadRaw = strippedPayload
+					currentBridgePayload.payloadBytes = len(strippedPayload)
+				}
+				if bridgeReplayInputExists {
+					bridgeReplayInput, _ = stripOpenAIInvalidEncryptedContentFromReplayItems(bridgeReplayInput, invalidDigests)
+				}
+				if bridgeAccountFailoverInputExists {
+					bridgeAccountFailoverInput, _ = stripOpenAIInvalidEncryptedContentFromReplayItems(bridgeAccountFailoverInput, invalidDigests)
+				}
+			}
 			bridgePayloadRaw := currentBridgePayload.payloadRaw
 			bridgePayloadBytes := currentBridgePayload.payloadBytes
 			toolOutputCoverage := AnalyzeToolCallOutputContextCoverageBytes(currentBridgePayload.payloadRaw)
@@ -1026,6 +1046,18 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient(
 				}
 				s.persistOpenAIWSRateLimitSignal(ctx, account, lease.HandshakeHeaders(), upstreamMessage, errCodeRaw, errTypeRaw, errMsgRaw, mappedModel)
 				fallbackReason, _ := classifyOpenAIWSErrorEventFromRaw(errCodeRaw, errTypeRaw, errMsgRaw)
+				if fallbackReason == openAIWSFallbackReasonInvalidEncryptedContent {
+					// 记录被上游拒绝的密文摘要；错误照旧透传，下一轮进场时按摘要预剥离。
+					if digests := collectOpenAIEncryptedContentDigestsRaw(payload); len(digests) > 0 {
+						s.markOpenAIWSInvalidEncryptedContentLineage(groupID, sessionHash, digests)
+						logOpenAIWSModeInfo(
+							"ingress_ws_invalid_encrypted_lineage_mark account_id=%d turn=%d digests=%d",
+							account.ID,
+							turn,
+							len(digests),
+						)
+					}
+				}
 				errCode, errType, errMessage := summarizeOpenAIWSErrorEventFieldsFromRaw(errCodeRaw, errTypeRaw, errMsgRaw)
 				recoverablePrevNotFound := fallbackReason == openAIWSIngressStagePreviousResponseNotFound &&
 					turnPreviousResponseID != "" &&
@@ -1411,6 +1443,20 @@ func (s *OpenAIGatewayService) ProxyResponsesWebSocketFromClient(
 			}
 		}
 		skipBeforeTurn = false
+		// 剥离本会话已知失效的加密项，阻断同一失效密文随历史反复触发上游拒绝。
+		// 历史序列须同步剥离，否则与已剥离的当前 input 项错位，prefix 复用失配。
+		if invalidDigests := s.sessionInvalidEncryptedContentDigests(groupID, sessionHash); len(invalidDigests) > 0 {
+			strippedPayload, strippedCount := s.stripSessionInvalidEncryptedContentLogged(
+				currentPayload, invalidDigests, "ingress_ws_invalid_encrypted_lineage_strip", account.ID, turn,
+			)
+			if strippedCount > 0 {
+				currentPayload = strippedPayload
+				currentPayloadBytes = len(strippedPayload)
+			}
+			if lastTurnReplayInputExists {
+				lastTurnReplayInput, _ = stripOpenAIInvalidEncryptedContentFromReplayItems(lastTurnReplayInput, invalidDigests)
+			}
+		}
 		currentPreviousResponseID := openAIWSPayloadStringFromRaw(currentPayload, "previous_response_id")
 		expectedPrev := strings.TrimSpace(lastTurnResponseID)
 		toolSignals := ToolContinuationSignals{

backend/internal/service/openai_ws_forwarder_ingress_session_test.go

@@ -4446,3 +4446,157 @@ func TestOpenAIGatewayService_ProxyResponsesWebSocketFromClient_ClientDisconnect
 		t.Fatal("未收到断连后的 turn 结果回调")
 	}
 }
+
+func TestOpenAIGatewayService_ProxyResponsesWebSocketFromClient_InvalidEncryptedContentLineageStripsNextTurn(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	cfg := &config.Config{}
+	cfg.Security.URLAllowlist.Enabled = false
+	cfg.Security.URLAllowlist.AllowInsecureHTTP = true
+	cfg.Gateway.OpenAIWS.Enabled = true
+	cfg.Gateway.OpenAIWS.OAuthEnabled = true
+	cfg.Gateway.OpenAIWS.APIKeyEnabled = true
+	cfg.Gateway.OpenAIWS.ResponsesWebsocketsV2 = true
+	cfg.Gateway.OpenAIWS.MaxConnsPerAccount = 1
+	cfg.Gateway.OpenAIWS.MinIdlePerAccount = 0
+	cfg.Gateway.OpenAIWS.MaxIdlePerAccount = 1
+	cfg.Gateway.OpenAIWS.QueueLimitPerConn = 8
+	cfg.Gateway.OpenAIWS.DialTimeoutSeconds = 3
+	cfg.Gateway.OpenAIWS.ReadTimeoutSeconds = 3
+	cfg.Gateway.OpenAIWS.WriteTimeoutSeconds = 3
+
+	upstreamConn := &openAIWSCaptureConn{
+		events: [][]byte{
+			[]byte(`{"type":"error","error":{"type":"invalid_request_error","code":"invalid_encrypted_content","message":"The encrypted content could not be verified"}}`),
+			[]byte(`{"type":"response.failed","response":{"id":"resp_enc_lineage_1","model":"gpt-5.1","error":{"code":"invalid_encrypted_content","message":"The encrypted content could not be verified"}}}`),
+			[]byte(`{"type":"response.completed","response":{"id":"resp_enc_lineage_2","model":"gpt-5.1","usage":{"input_tokens":1,"output_tokens":1}}}`),
+		},
+	}
+	dialer := &openAIWSQueueDialer{
+		conns: []openAIWSClientConn{upstreamConn},
+	}
+	pool := newOpenAIWSConnPool(cfg)
+	pool.setClientDialerForTest(dialer)
+
+	svc := &OpenAIGatewayService{
+		cfg:              cfg,
+		httpUpstream:     &httpUpstreamRecorder{},
+		cache:            &stubGatewayCache{},
+		openaiWSResolver: NewOpenAIWSProtocolResolver(cfg),
+		toolCorrector:    NewCodexToolCorrector(),
+		openaiWSPool:     pool,
+	}
+
+	account := &Account{
+		ID:          119,
+		Name:        "openai-ingress-enc-lineage",
+		Platform:    PlatformOpenAI,
+		Type:        AccountTypeAPIKey,
+		Status:      StatusActive,
+		Schedulable: true,
+		Concurrency: 1,
+		Credentials: map[string]any{
+			"api_key": "sk-test",
+		},
+		Extra: map[string]any{
+			"responses_websockets_v2_enabled": true,
+		},
+	}
+
+	serverErrCh := make(chan error, 1)
+	wsServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := coderws.Accept(w, r, &coderws.AcceptOptions{
+			CompressionMode: coderws.CompressionContextTakeover,
+		})
+		if err != nil {
+			serverErrCh <- err
+			return
+		}
+		defer func() {
+			_ = conn.CloseNow()
+		}()
+
+		rec := httptest.NewRecorder()
+		ginCtx, _ := gin.CreateTestContext(rec)
+		req := r.Clone(r.Context())
+		req.Header = req.Header.Clone()
+		req.Header.Set("User-Agent", "unit-test-agent/1.0")
+		ginCtx.Request = req
+
+		readCtx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
+		msgType, firstMessage, readErr := conn.Read(readCtx)
+		cancel()
+		if readErr != nil {
+			serverErrCh <- readErr
+			return
+		}
+		if msgType != coderws.MessageText && msgType != coderws.MessageBinary {
+			serverErrCh <- errors.New("unsupported websocket client message type")
+			return
+		}
+
+		serverErrCh <- svc.ProxyResponsesWebSocketFromClient(r.Context(), ginCtx, conn, account, "sk-test", firstMessage, nil)
+	}))
+	defer wsServer.Close()
+
+	dialCtx, cancelDial := context.WithTimeout(context.Background(), 3*time.Second)
+	clientConn, _, err := coderws.Dial(dialCtx, "ws"+strings.TrimPrefix(wsServer.URL, "http"), nil)
+	cancelDial()
+	require.NoError(t, err)
+	defer func() {
+		_ = clientConn.CloseNow()
+	}()
+
+	writeMessage := func(payload string) {
+		writeCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		defer cancel()
+		require.NoError(t, clientConn.Write(writeCtx, coderws.MessageText, []byte(payload)))
+	}
+	readMessage := func() []byte {
+		readCtx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		defer cancel()
+		msgType, message, readErr := clientConn.Read(readCtx)
+		require.NoError(t, readErr)
+		require.Equal(t, coderws.MessageText, msgType)
+		return message
+	}
+
+	// turn1：携带失效密文，上游以 invalid_encrypted_content 拒绝（error + response.failed 透传给客户端）。
+	writeMessage(`{"type":"response.create","model":"gpt-5.1","stream":false,"input":[{"type":"reasoning","id":"rs_1","encrypted_content":"stale-cipher","summary":[]},{"type":"input_text","text":"hi"}]}`)
+	firstEvent := readMessage()
+	require.Equal(t, "error", gjson.GetBytes(firstEvent, "type").String())
+	require.Equal(t, "invalid_encrypted_content", gjson.GetBytes(firstEvent, "error.code").String())
+	secondEvent := readMessage()
+	require.Equal(t, "response.failed", gjson.GetBytes(secondEvent, "type").String())
+
+	// turn2：客户端历史仍带同一失效密文，进场应被 lineage 预剥离后再发上游。
+	writeMessage(`{"type":"response.create","model":"gpt-5.1","stream":false,"input":[{"type":"reasoning","id":"rs_1","encrypted_content":"stale-cipher","summary":[]},{"type":"input_text","text":"hi"},{"type":"input_text","text":"again"}]}`)
+	thirdEvent := readMessage()
+	require.Equal(t, "response.completed", gjson.GetBytes(thirdEvent, "type").String())
+	require.Equal(t, "resp_enc_lineage_2", gjson.GetBytes(thirdEvent, "response.id").String())
+
+	require.NoError(t, clientConn.Close(coderws.StatusNormalClosure, "done"))
+	select {
+	case serverErr := <-serverErrCh:
+		require.NoError(t, serverErr)
+	case <-time.After(5 * time.Second):
+		t.Fatal("等待 ingress websocket 结束超时")
+	}
+
+	upstreamConn.mu.Lock()
+	writes := append([]map[string]any(nil), upstreamConn.writes...)
+	upstreamConn.mu.Unlock()
+	require.Len(t, writes, 2, "两轮各应发送一次上游请求")
+
+	firstUpstream := requestToJSONString(writes[0])
+	require.Equal(t, "stale-cipher", gjson.Get(firstUpstream, "input.0.encrypted_content").String(), "首轮请求原样携带密文")
+
+	secondUpstream := requestToJSONString(writes[1])
+	secondInput := gjson.Get(secondUpstream, "input").Array()
+	require.Len(t, secondInput, 3, "剥离仅移除 encrypted_content 字段，reasoning 骨架保留")
+	for _, item := range secondInput {
+		require.False(t, item.Get("encrypted_content").Exists(), "第二轮请求不得再携带已失效密文: %s", item.Raw)
+	}
+	require.Equal(t, "rs_1", gjson.Get(secondUpstream, "input.0.id").String())
+	require.Equal(t, "again", gjson.Get(secondUpstream, "input.2.text").String())
+}

backend/internal/service/openai_ws_http_bridge.go

@@ -438,6 +438,12 @@ func (s *OpenAIGatewayService) proxyOpenAIWSHTTPBridgeTurn(
 
 		respBody, _ := io.ReadAll(io.LimitReader(resp.Body, openAIWSHTTPBridgeErrorBodyLimitBytes))
 		_ = resp.Body.Close()
+		if resp.StatusCode == http.StatusBadRequest &&
+			extractUpstreamErrorCode(respBody) == openAIWSFallbackReasonInvalidEncryptedContent {
+			s.markOpenAIWSInvalidEncryptedContentLineageFromPayload(
+				c, body, "ingress_ws_http_bridge_invalid_encrypted_lineage_mark", account.ID, turn,
+			)
+		}
 		retryBody, retryReason, changed, retryErr := normalizeOpenAIResponsesRejectedFieldRetryBody(resp.StatusCode, body, respBody)
 		if retryErr != nil {
 			return nil, fmt.Errorf("normalize websocket http bridge rejected field retry: %w", retryErr)
@@ -680,6 +686,11 @@ func (s *OpenAIGatewayService) proxyOpenAIWSHTTPBridgeTurn(
 				if account.Platform == PlatformGrok {
 					statusCode = openAIWSErrorHTTPStatusFromRaw(errCodeRaw, errTypeRaw)
 				}
+				if reason, _ := classifyOpenAIWSErrorEventFromRaw(errCodeRaw, errTypeRaw, errMessage); reason == openAIWSFallbackReasonInvalidEncryptedContent {
+					s.markOpenAIWSInvalidEncryptedContentLineageFromPayload(
+						c, body, "ingress_ws_http_bridge_invalid_encrypted_lineage_mark", account.ID, turn,
+					)
+				}
 			}
 			requestScopedCapacity := isOpenAIUpstreamCapacityShedEvent(upstreamMessage)
 			if account.Platform == PlatformGrok && eventType == "error" {

backend/internal/service/openai_ws_state_store.go

@@ -47,6 +47,18 @@ type openAIWSSessionConnBinding struct {
 	expiresAt time.Time
 }
 
+// openAIWSInvalidEncryptedBinding 记录一个会话中已被上游判定失效的
+// encrypted_content 摘要（invalid_encrypted_content lineage）。
+type openAIWSInvalidEncryptedBinding struct {
+	digests   map[string]struct{}
+	expiresAt time.Time
+}
+
+// openAIWSInvalidEncryptedDigestsPerSession 是单会话摘要集合的存储自保护上限。
+// 超出后新增摘要被忽略，仅退化为"该项下次仍触发一次上游拒绝后的常规 recovery"，
+// 不影响正确性。
+const openAIWSInvalidEncryptedDigestsPerSession = 512
+
 // OpenAIWSStateStore 管理 WSv2 的粘连状态。
 // - response_id -> account_id 用于续链路由
 // - response_id -> conn_id 用于连接内上下文复用
@@ -71,6 +83,15 @@ type OpenAIWSStateStore interface {
 	BindSessionConn(groupID int64, sessionHash, connID string, ttl time.Duration)
 	GetSessionConn(groupID int64, sessionHash string) (string, bool)
 	DeleteSessionConn(groupID int64, sessionHash string)
+
+	// invalid_encrypted_content lineage：按会话记录已被上游拒绝的
+	// encrypted_content 摘要，后续 turn 进场时仅剥离命中项，避免同一失效
+	// 密文随客户端历史反复触发"整包被拒→剥离→重试/重连"。仅进程内有效。
+	MarkSessionInvalidEncryptedContent(groupID int64, sessionHash string, digests []string, ttl time.Duration)
+	GetSessionInvalidEncryptedContentDigests(groupID int64, sessionHash string) map[string]struct{}
+	// HasAnySessionInvalidEncryptedContent 是热路径快速探测：全局无记录时
+	// 调用方可跳过会话哈希计算与摘要匹配。
+	HasAnySessionInvalidEncryptedContent() bool
 }
 
 type defaultOpenAIWSStateStore struct {
@@ -87,18 +108,22 @@ type defaultOpenAIWSStateStore struct {
 	sessionToConnMu      sync.RWMutex
 	sessionToConn        map[string]openAIWSSessionConnBinding
 
+	sessionInvalidEncryptedMu sync.RWMutex
+	sessionInvalidEncrypted   map[string]openAIWSInvalidEncryptedBinding
+
 	lastCleanupUnixNano atomic.Int64
 }
 
 // NewOpenAIWSStateStore 创建默认 WS 状态存储。
 func NewOpenAIWSStateStore(cache GatewayCache) OpenAIWSStateStore {
 	store := &defaultOpenAIWSStateStore{
-		cache:              cache,
-		responseToAccount:  make(map[string]openAIWSAccountBinding, 256),
-		responseOwners:     make(map[string]openAIHTTPResponseOwnerBinding, 256),
-		responseToConn:     make(map[string]openAIWSConnBinding, 256),
-		sessionToTurnState: make(map[string]openAIWSTurnStateBinding, 256),
-		sessionToConn:      make(map[string]openAIWSSessionConnBinding, 256),
+		cache:                   cache,
+		responseToAccount:       make(map[string]openAIWSAccountBinding, 256),
+		responseOwners:          make(map[string]openAIHTTPResponseOwnerBinding, 256),
+		responseToConn:          make(map[string]openAIWSConnBinding, 256),
+		sessionToTurnState:      make(map[string]openAIWSTurnStateBinding, 256),
+		sessionToConn:           make(map[string]openAIWSSessionConnBinding, 256),
+		sessionInvalidEncrypted: make(map[string]openAIWSInvalidEncryptedBinding),
 	}
 	store.lastCleanupUnixNano.Store(time.Now().UnixNano())
 	return store
@@ -396,6 +421,76 @@ func (s *defaultOpenAIWSStateStore) DeleteSessionConn(groupID int64, sessionHash
 	s.sessionToConnMu.Unlock()
 }
 
+func (s *defaultOpenAIWSStateStore) MarkSessionInvalidEncryptedContent(groupID int64, sessionHash string, digests []string, ttl time.Duration) {
+	key := openAIWSSessionTurnStateKey(groupID, sessionHash)
+	if key == "" || len(digests) == 0 {
+		return
+	}
+	ttl = normalizeOpenAIWSTTL(ttl)
+	s.maybeCleanup()
+
+	now := time.Now()
+	overflowed := 0
+	s.sessionInvalidEncryptedMu.Lock()
+	ensureBindingCapacity(s.sessionInvalidEncrypted, key, openAIWSStateStoreMaxEntriesPerMap)
+	binding, ok := s.sessionInvalidEncrypted[key]
+	if !ok || now.After(binding.expiresAt) || binding.digests == nil {
+		binding = openAIWSInvalidEncryptedBinding{digests: make(map[string]struct{}, len(digests))}
+	}
+	for _, digest := range digests {
+		digest = strings.TrimSpace(digest)
+		if digest == "" {
+			continue
+		}
+		if len(binding.digests) >= openAIWSInvalidEncryptedDigestsPerSession {
+			if _, exists := binding.digests[digest]; !exists {
+				overflowed++
+			}
+			continue
+		}
+		binding.digests[digest] = struct{}{}
+	}
+	binding.expiresAt = now.Add(ttl)
+	s.sessionInvalidEncrypted[key] = binding
+	s.sessionInvalidEncryptedMu.Unlock()
+
+	// HasAny/Get 在每个 OpenAI 请求热路径上抢同一把锁，日志 IO 必须在锁外。
+	if overflowed > 0 {
+		logOpenAIWSModeInfo(
+			"invalid_encrypted_lineage_capacity_overflow dropped_digests=%d capacity=%d",
+			overflowed,
+			openAIWSInvalidEncryptedDigestsPerSession,
+		)
+	}
+}
+
+func (s *defaultOpenAIWSStateStore) GetSessionInvalidEncryptedContentDigests(groupID int64, sessionHash string) map[string]struct{} {
+	key := openAIWSSessionTurnStateKey(groupID, sessionHash)
+	if key == "" {
+		return nil
+	}
+	s.maybeCleanup()
+
+	now := time.Now()
+	s.sessionInvalidEncryptedMu.RLock()
+	defer s.sessionInvalidEncryptedMu.RUnlock()
+	binding, ok := s.sessionInvalidEncrypted[key]
+	if !ok || now.After(binding.expiresAt) || len(binding.digests) == 0 {
+		return nil
+	}
+	digests := make(map[string]struct{}, len(binding.digests))
+	for digest := range binding.digests {
+		digests[digest] = struct{}{}
+	}
+	return digests
+}
+
+func (s *defaultOpenAIWSStateStore) HasAnySessionInvalidEncryptedContent() bool {
+	s.sessionInvalidEncryptedMu.RLock()
+	defer s.sessionInvalidEncryptedMu.RUnlock()
+	return len(s.sessionInvalidEncrypted) > 0
+}
+
 func (s *defaultOpenAIWSStateStore) maybeCleanup() {
 	if s == nil {
 		return
@@ -429,6 +524,26 @@ func (s *defaultOpenAIWSStateStore) maybeCleanup() {
 	s.sessionToConnMu.Lock()
 	cleanupExpiredSessionConnBindings(s.sessionToConn, now, openAIWSStateStoreCleanupMaxPerMap)
 	s.sessionToConnMu.Unlock()
+
+	s.sessionInvalidEncryptedMu.Lock()
+	cleanupExpiredInvalidEncryptedBindings(s.sessionInvalidEncrypted, now, openAIWSStateStoreCleanupMaxPerMap)
+	s.sessionInvalidEncryptedMu.Unlock()
+}
+
+func cleanupExpiredInvalidEncryptedBindings(bindings map[string]openAIWSInvalidEncryptedBinding, now time.Time, maxScan int) {
+	if len(bindings) == 0 || maxScan <= 0 {
+		return
+	}
+	scanned := 0
+	for key, binding := range bindings {
+		if now.After(binding.expiresAt) {
+			delete(bindings, key)
+		}
+		scanned++
+		if scanned >= maxScan {
+			break
+		}
+	}
 }
 
 func cleanupExpiredAccountBindings(bindings map[string]openAIWSAccountBinding, now time.Time, maxScan int) {