fix(gateway): cache affinity and Anthropic message rectification

changelog/unreleased/cache-affinity-and-message-rectifier.md

@@ -0,0 +1,15 @@
+### English
+
+- Recognise client conversation ids from headers and body metadata aliases (`X-Conversation-Id`, `X-Session-Id`, `X-Thread-Id`, `conversation_id`, `conversationId`, `metadata.*`, ...), so chat turns stay on one account and keep the upstream prompt cache warm instead of rotating through the pool.
+- Add opt-in `CLI2API_MSGSTAT_SHARDS` to bound one-shot Messages calls to a small, stable set of accounts per (API key, model, system prefix, tools), matching `WB_MSGSTAT_SHARDS` in workbuddy2api-hub.
+- Add opt-in `CLI2API_AFFINITY_DEBUG` diagnostics that separate route-level session affinity from real upstream cache-token hits, including prompt/cache tokens and hit rate.
+- Rectify historical Anthropic `thinking` / `redacted_thinking` blocks and stale signatures on `/v1/messages` (cc-switch-style minimal cleanup) so OpenAI-compatible upstreams no longer receive invalid signed blocks. Disable with `CLI2API_THINKING_RECTIFIER=off`.
+- Skip WAF fingerprint regexes for request bodies with no known trigger, avoiding unnecessary per-request CPU work on large histories and keeping cached prefixes stable.
+
+### 中文
+
+- 支持从请求头和 body 元数据别名识别会话 ID（`X-Conversation-Id`、`X-Session-Id`、`X-Thread-Id`、`conversation_id`、`conversationId`、`metadata.*` 等），让同一对话固定在同一账号上，避免轮转导致上游 prompt 缓存失效。
+- 新增可选 `CLI2API_MSGSTAT_SHARDS`，按（API Key、模型、system 前缀、tools）把单发 Messages 请求收敛到少量固定账号，对齐 workbuddy2api-hub 的 `WB_MSGSTAT_SHARDS`。
+- 新增可选 `CLI2API_AFFINITY_DEBUG` 诊断日志，把「路由级会话亲和」与「上游真实缓存 token 命中」分开记录，包含 prompt/cache tokens 与命中率。
+- `/v1/messages` 入站按 cc-switch 思路最小清理历史 `thinking` / `redacted_thinking` 块和失效签名，避免把无效签名转发给 OpenAI 兼容上游；可用 `CLI2API_THINKING_RECTIFIER=off` 关闭。
+- 对不含任何已知指纹的请求体跳过 WAF 正则清洗，减少长历史请求的 CPU 开销，并避免清洗改动缓存前缀。

deploy/README.md

@@ -240,3 +240,17 @@ Keep a separate database backup before upgrading.
 Keep `deploy/.env` private: Docker Desktop mode stores an updater token there.
 Do not remove `qoder-data` or run `docker compose down -v` unless you intend to
 delete the accounts and credentials.
+
+
+## Cache affinity and message rectification
+
+These switches follow the reference 2api gateways and are off by default except
+the Anthropic thinking rectifier, which is on because OpenAI-compatible
+upstreams do not validate Anthropic thinking signatures.
+
+| Variable | Default | Description |
+|---|---|---|
+| `CLI2API_MSGSTAT_SHARDS` | `0` | Bound one-shot `/v1/messages` calls to N account buckets per (API key, model, system prefix, tools). Mirrors `WB_MSGSTAT_SHARDS`; `0` keeps the historical content-seed routing. |
+| `CLI2API_AFFINITY_DEBUG` | `0` | Log session-affinity source and cache-token diagnostics per request. |
+| `CLI2API_THINKING_RECTIFIER` | `on` | Set `off` to replay historical Anthropic thinking blocks unchanged. |
+| `X-CLI2API-Session` / `X-Conversation-Id` / `X-Session-Id` / `X-Thread-Id` | — | Explicit conversation id. Body `conversation_id` / `conversationId` / `session_id` / `metadata.*` aliases are also recognised so client turns stay on one account and keep the upstream prompt cache warm. |

internal/executor/prepare.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"encoding/json"
 	"log"
+	"os"
 	"strconv"
 	"strings"
 	"time"
@@ -59,6 +60,7 @@ type PrepareInput struct {
 	Identity          auth.Identity
 	PreferAccount     string
 	SessionHeader     string
+	ConversationID    string
 	CrossProviderPool bool
 	Now               time.Time
 	NewRequestID      func() string
@@ -146,10 +148,18 @@ func (e ChatExecutor) Prepare(in PrepareInput) (PreparedRequest, error) {
 		})
 	}
 	ctx := WithAllowedProviders(WithRequestID(in.Context, requestID), in.Identity.AllowedProviders)
-	sessionKey := SessionKeyFor(in.SessionHeader, in.Identity, request)
+	sessionKey := sessionKeyFor(in.SessionHeader, in.ConversationID, in.Identity, request)
 	if sessionKey == "" && in.NativeRequest != nil {
 		sessionKey = NativeSessionKeyFor(in.NativeRequest, in.Identity)
 	}
+	// No caller-supplied conversation: hub-style sharding keeps one
+	// (key, model, system, tools) tuple on a small account set. The content
+	// seed still runs when sharding is disabled (default).
+	if strings.TrimSpace(in.SessionHeader) == "" && strings.TrimSpace(in.ConversationID) == "" {
+		if shard := ShardFallbackKey(in.Identity, request); shard != "" {
+			sessionKey = shard
+		}
+	}
 	if sessionKey != "" {
 		ctx = WithSessionKey(ctx, sessionKey)
 		ctx = WithConversation(ctx, sessionKey)
@@ -160,6 +170,18 @@ func (e ChatExecutor) Prepare(in PrepareInput) (PreparedRequest, error) {
 	if requestedReasoning != "" {
 		log.Printf("request reasoning request_id=%q model=%q level=%q", requestID, request.Model, requestedReasoning)
 	}
+	if sessionKey != "" && sessionAffinityDebugOn() {
+		source := "content-seed"
+		if strings.TrimSpace(in.SessionHeader) != "" {
+			source = "header"
+		} else if strings.TrimSpace(in.ConversationID) != "" {
+			source = "conversation"
+		} else if MessageStatShards() > 0 {
+			source = "shard"
+		}
+		log.Printf("session affinity request_id=%q source=%q key=%q model=%q tools=%t msgs=%d",
+			requestID, source, shortSessionKey(sessionKey), request.Model, len(request.Tools) > 0, len(request.Messages))
+	}
 	return PreparedRequest{
 		Context:        ctx,
 		RequestID:      requestID,
@@ -312,22 +334,7 @@ func NativeSessionKeyFor(req *translate.NativeResponsesRequest, identity auth.Id
 }
 
 func SessionKeyFor(header string, identity auth.Identity, req translate.ChatRequest) string {
-	raw := strings.TrimSpace(header)
-	kind := "content"
-	if raw != "" {
-		kind = "header"
-	} else {
-		raw = translate.ContentSessionSeed(req)
-	}
-	if raw == "" {
-		return ""
-	}
-	namespace := "console"
-	if identity.Kind == auth.KindKey && identity.KeyID != "" {
-		namespace = "key:" + identity.KeyID
-	}
-	sum := sha256.Sum256([]byte(namespace + "\x00" + kind + "\x00" + raw))
-	return hex.EncodeToString(sum[:])
+	return sessionKeyFor(header, "", identity, req)
 }
 
 // RequestedReasoningLevel surfaces the reasoning level the client asked for,
@@ -371,3 +378,14 @@ func RequestedReasoningLevel(req translate.ChatRequest) string {
 	}
 	return ""
 }
+
+// sessionAffinityDebugOn mirrors qoder2api-hub's QD_AFFINITY_DEBUG: routing
+// diagnostics are opt-in so normal request paths stay quiet.
+func sessionAffinityDebugOn() bool {
+	switch strings.ToLower(strings.TrimSpace(os.Getenv("CLI2API_AFFINITY_DEBUG"))) {
+	case "1", "true", "on", "yes":
+		return true
+	default:
+		return false
+	}
+}

internal/executor/session_key.go

@@ -0,0 +1,173 @@
+package executor
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"os"
+	"strconv"
+	"strings"
+
+	"github.com/caigee-cmd/cli2api/internal/auth"
+	"github.com/caigee-cmd/cli2api/internal/translate"
+)
+
+// sessionIDAliases mirrors the reference gateways' conversation-key lookup
+// (Sliverkiss/workbuddy2api #35, qoder2api-hub session affinity). Clients use
+// several spellings for the same conversation id; recognising only one makes
+// turns rotate accounts and silently destroys the upstream prompt cache.
+var sessionIDAliases = []string{
+	"conversation_id", "conversationId", "conversation-id",
+	"session_id", "sessionId", "session-id",
+	"thread_id", "threadId",
+	"conversation", "session",
+}
+
+// sessionKeySources carries every id a caller can use to pin one chat to one
+// account. HTTP headers win over body metadata; the gateway's own header stays
+// first for compatibility.
+type sessionKeySources struct {
+	Headers   []string
+	BodyJSON  []byte
+	SessionID string
+}
+
+// extractConversationKey returns the first non-empty, bounded conversation id
+// found in headers, then in the request body (top level and metadata), then in
+// the native Responses session seed. Unknown keys are ignored; no id is ever
+// invented. Values are capped so a hostile client cannot pin unbounded memory.
+func extractConversationKey(src sessionKeySources) string {
+	for _, value := range src.Headers {
+		if id := cleanConversationID(value); id != "" {
+			return id
+		}
+	}
+	if len(src.BodyJSON) > 0 && len(src.BodyJSON) <= 64<<20 {
+		var body map[string]json.RawMessage
+		if json.Unmarshal(src.BodyJSON, &body) == nil {
+			if id := conversationIDFromBody(body); id != "" {
+				return id
+			}
+		}
+	}
+	if id := cleanConversationID(src.SessionID); id != "" {
+		return id
+	}
+	return ""
+}
+
+func conversationIDFromBody(body map[string]json.RawMessage) string {
+	for _, key := range sessionIDAliases {
+		if id, ok := rawConversationString(body[key]); ok {
+			return id
+		}
+	}
+	for _, metaKey := range []string{"metadata", "meta", "context"} {
+		var meta map[string]json.RawMessage
+		if json.Unmarshal(body[metaKey], &meta) != nil {
+			continue
+		}
+		for _, key := range sessionIDAliases {
+			if id, ok := rawConversationString(meta[key]); ok {
+				return id
+			}
+		}
+	}
+	return ""
+}
+
+func rawConversationString(raw json.RawMessage) (string, bool) {
+	if len(raw) == 0 {
+		return "", false
+	}
+	var value string
+	if json.Unmarshal(raw, &value) == nil {
+		if id := cleanConversationID(value); id != "" {
+			return id, true
+		}
+	}
+	return "", false
+}
+
+func cleanConversationID(value string) string {
+	value = strings.TrimSpace(value)
+	if value == "" || len(value) > 512 {
+		return ""
+	}
+	return value
+}
+
+// sessionKeyFor is the shared implementation; PrepareInput carries an explicit
+// conversation id in addition to the legacy header.
+func sessionKeyFor(header, conversation string, identity auth.Identity, req translate.ChatRequest) string {
+	raw := strings.TrimSpace(header)
+	kind := "content"
+	if raw == "" {
+		raw = strings.TrimSpace(conversation)
+	}
+	if raw != "" {
+		kind = "header"
+	} else {
+		raw = translate.ContentSessionSeed(req)
+	}
+	if raw == "" {
+		return ""
+	}
+	namespace := "console"
+	if identity.Kind == auth.KindKey && identity.KeyID != "" {
+		namespace = "key:" + identity.KeyID
+	}
+	sum := sha256.Sum256([]byte(namespace + "\x00" + kind + "\x00" + raw))
+	return hex.EncodeToString(sum[:])
+}
+
+// ShardKeyFor implements workbuddy2api-hub's WB_MSGSTAT_SHARDS idea with a pure
+// function: one (key, model, tools, system-prefix) tuple maps to a small, stable
+// set of accounts instead of being scattered by the first user text. The caller
+// owns the account selection; shards is the maximum number of buckets.
+func ShardKeyFor(identity auth.Identity, req translate.ChatRequest, shards int) string {
+	if shards <= 0 {
+		return ""
+	}
+	namespace := "console"
+	if identity.Kind == auth.KindKey && identity.KeyID != "" {
+		namespace = "key:" + identity.KeyID
+	}
+	seed := translate.PrefixSessionSeed(req)
+	if seed == "" {
+		return ""
+	}
+	sum := sha256.Sum256([]byte(namespace + "\x00shard\x00" + seed))
+	bucket := int(sum[0])%shards
+	return namespace + "\x00" + hex.EncodeToString(sum[:8]) + "\x00" + string(rune('0'+bucket))
+}
+
+
+func shortSessionKey(key string) string {
+	if len(key) <= 12 {
+		return key
+	}
+	return key[:12]
+}
+
+
+// MessageStatShards mirrors workbuddy2api-hub's WB_MSGSTAT_SHARDS. Default 0
+// keeps historical routing unchanged; a positive value bounds one-shot
+// /v1/messages calls to N account buckets per (key, model, system, tools).
+func MessageStatShards() int {
+	raw := strings.TrimSpace(os.Getenv("CLI2API_MSGSTAT_SHARDS"))
+	if raw == "" {
+		return 0
+	}
+	n, err := strconv.Atoi(raw)
+	if err != nil || n < 1 || n > 64 {
+		return 0
+	}
+	return n
+}
+
+// ShardFallbackKey returns a stable, non-empty affinity key only when sharding
+// is enabled. It never overrides an explicit conversation id.
+func ShardFallbackKey(identity auth.Identity, req translate.ChatRequest) string {
+	return ShardKeyFor(identity, req, MessageStatShards())
+}