fix(affinity): 统一缓存标识亲和并补齐 Codex 缓存分组 (#635)

internal/affinity/key.go

@@ -28,11 +28,20 @@ func DeriveKey(
 	clientProtocol protocol.Protocol,
 	prefix []byte,
 ) Key {
+	return deriveKey(hasher, accessKeyID, clientProtocol, keyDomain, prefix)
+}
+
+// DerivePromptCacheKey 与提示词使用不同命名空间，保持现有租户和协议隔离。
+func DerivePromptCacheKey(hasher Hasher, accessKeyID uint, clientProtocol protocol.Protocol, key string) Key {
+	return deriveKey(hasher, accessKeyID, clientProtocol, "gpt-load/affinity/prompt-cache-key/v1", []byte(key))
+}
+
+func deriveKey(hasher Hasher, accessKeyID uint, clientProtocol protocol.Protocol, domain string, prefix []byte) Key {
 	if hasher == nil || accessKeyID == 0 || !clientProtocol.Valid() || len(prefix) == 0 {
 		return ""
 	}
 	var material bytes.Buffer
-	writeKeyField(&material, []byte(keyDomain))
+	writeKeyField(&material, []byte(domain))
 	var encodedID [8]byte
 	binary.BigEndian.PutUint64(encodedID[:], uint64(accessKeyID))
 	material.Write(encodedID[:])

internal/affinity/key_test.go

@@ -78,3 +78,20 @@ func TestDeriveKeyUsesUnambiguousFieldBoundaries(t *testing.T) {
 		t.Fatalf("keys = %q / %q, want distinct non-empty values", left, right)
 	}
 }
+
+func TestPromptCacheKeyNamespacesAndIsolation(t *testing.T) {
+	base := DerivePromptCacheKey(testHasher{}, 7, protocol.OpenAICompletions, "same")
+	if !base.Valid() || base != DerivePromptCacheKey(testHasher{}, 7, protocol.OpenAICompletions, "same") {
+		t.Fatal("explicit affinity key is not stable")
+	}
+	for _, other := range []Key{
+		DeriveKey(testHasher{}, 7, protocol.OpenAICompletions, []byte("same")),
+		DerivePromptCacheKey(testHasher{}, 8, protocol.OpenAICompletions, "same"),
+		DerivePromptCacheKey(testHasher{}, 7, protocol.OpenAIResponses, "same"),
+		DerivePromptCacheKey(testHasher{}, 7, protocol.OpenAICompletions, "different"),
+	} {
+		if base == other {
+			t.Fatal("affinity namespace or tenant boundary collapsed")
+		}
+	}
+}

internal/control/request_logs.go

@@ -149,6 +149,7 @@ type requestLogItemResponse struct {
 	ErrorCode               string                       `json:"error_code"`
 	ErrorSummary            string                       `json:"error_summary"`
 	AffinityHit             bool                         `json:"affinity_hit"`
+	AffinityKind            string                       `json:"affinity_kind"`
 	GroupID                 *uint                        `json:"group_id"`
 	ChannelID               *channel.ID                  `json:"channel_id"`
 	CredentialID            *uint                        `json:"credential_id"`
@@ -329,6 +330,7 @@ func sanitizeAccessKeyRequestLog(record requestlog.Record) requestlog.Record {
 	record.ModelConsistency = telemetry.ModelConsistencyNotApplicable
 	record.AttemptCount = 0
 	record.AffinityHit = false
+	record.AffinityKind = ""
 	record.GroupID = 0
 	record.ChannelID = ""
 	record.CredentialID = 0
@@ -925,6 +927,7 @@ func mapRequestLogItemResponse(
 		ErrorCode:               record.ErrorCode,
 		ErrorSummary:            record.ErrorSummary,
 		AffinityHit:             record.AffinityHit,
+		AffinityKind:            record.AffinityKind,
 		GroupID:                 usageCost.groupID,
 		ChannelID:               usageCost.channelID,
 		CredentialID:            usageCost.credentialID,

internal/control/request_logs_test.go

@@ -385,6 +385,7 @@ func TestRequestLogEndpointReturnsOpaqueCursorAndSafeDTO(t *testing.T) {
 					StatusCode:             200,
 					DurationMs:             1234,
 					AffinityHit:            true,
+					AffinityKind:           telemetry.AffinityPromptCacheKey,
 					GroupID:                12,
 					ChannelID:              channel.OpenAI,
 					CredentialID:           99,
@@ -452,7 +453,8 @@ func TestRequestLogEndpointReturnsOpaqueCursorAndSafeDTO(t *testing.T) {
 	if _, exists := envelope.Data.Items[0]["attempts"]; exists {
 		t.Fatalf("list item unexpectedly exposes attempts: %#v", envelope.Data.Items[0])
 	}
-	if envelope.Data.Items[0]["upstream_reported_model"] != "reported-model" ||
+	if envelope.Data.Items[0]["affinity_kind"] != telemetry.AffinityPromptCacheKey ||
+		envelope.Data.Items[0]["upstream_reported_model"] != "reported-model" ||
 		envelope.Data.Items[0]["model_consistency"] != string(telemetry.ModelConsistencyMismatch) ||
 		envelope.Data.Items[0]["route_mode"] != string(channel.RouteNative) ||
 		envelope.Data.Items[0]["operation"] != string(execution.OperationChatCompletion) ||
@@ -1093,6 +1095,7 @@ func TestRequestLogEndpointsBindAccessKeyScopeAndRedactRoutingInternals(t *testi
 		DurationMs:            120,
 		AttemptCount:          2,
 		AffinityHit:           true,
+		AffinityKind:          telemetry.AffinityPromptCacheKey,
 		GroupID:               99,
 		ChannelID:             channel.OpenAI,
 		CredentialID:          101,
@@ -1201,6 +1204,7 @@ func assertAccessKeyLogRedaction(t *testing.T, body []byte, detail bool) {
 		"upstream_reported_model": "null",
 		"model_consistency":       `"not_applicable"`,
 		"affinity_hit":            "false",
+		"affinity_kind":           `""`,
 		"group_id":                "null",
 		"channel_id":              "null",
 		"credential_id":           "null",

internal/dialect/dialect.go

@@ -22,6 +22,7 @@ type RequestMetadata struct {
 	Model                    *string
 	Stream                   bool
 	AffinityPrefix           []byte
+	PromptCacheKey           string `json:"-"`
 	Operation                execution.Operation
 	RouteRequirement         execution.RouteRequirement
 	ResponsesStorePreference execution.ResponsesStorePreference

internal/dialect/openai.go

@@ -28,6 +28,7 @@ func (d *OpenAI) InspectRequest(req *ParsedRequest) (RequestMetadata, error) {
 		return RequestMetadata{}, fmt.Errorf("decode %s request: %w", d.Protocol(), err)
 	}
 	metadata.ObserveUsage = true
+	metadata.PromptCacheKey = inspectPromptCacheKey(req.Body)
 	metadata.AffinityPrefix = inspectPromptAffinityPrefix(d.Protocol(), req.Body)
 	metadata.PricingMode, metadata.UsageDiagnostics, err = openAIRequestPricing(req.Body)
 	if err != nil {

internal/dialect/openai_responses.go

@@ -43,6 +43,9 @@ func (d *OpenAIResponses) InspectRequest(req *ParsedRequest) (RequestMetadata, e
 			return RequestMetadata{}, fmt.Errorf("decode %s request: %w", d.Protocol(), err)
 		}
 		metadata = parsed
+		if req.Method == http.MethodPost && req.Path == openAIResponsesPath {
+			metadata.PromptCacheKey = inspectPromptCacheKey(req.Body)
+		}
 	}
 	if req.Method == http.MethodGet {
 		stream, present, err := inspectResponsesStreamQuery(req.RawQuery)

internal/dialect/prompt_cache.go

@@ -0,0 +1,49 @@
+package dialect
+
+import (
+	"bytes"
+	"encoding/json"
+	"strings"
+	"unicode"
+	"unicode/utf8"
+)
+
+// inspectPromptCacheKey 只提取有界、无歧义的顶层软亲和信号，不修改客户端请求。
+func inspectPromptCacheKey(body []byte) string {
+	decoder := json.NewDecoder(bytes.NewReader(body))
+	token, err := decoder.Token()
+	if err != nil || token != json.Delim('{') {
+		return ""
+	}
+	var key string
+	seen := false
+	for decoder.More() {
+		field, err := decoder.Token()
+		if err != nil {
+			return ""
+		}
+		var value json.RawMessage
+		if decoder.Decode(&value) != nil {
+			return ""
+		}
+		if field != "prompt_cache_key" {
+			continue
+		}
+		if seen {
+			return ""
+		}
+		seen = true
+		if json.Unmarshal(value, &key) != nil {
+			return ""
+		}
+	}
+	if key == "" || len(key) > 256 || !utf8.ValidString(key) || strings.TrimSpace(key) != key {
+		return ""
+	}
+	for _, r := range key {
+		if unicode.IsControl(r) {
+			return ""
+		}
+	}
+	return key
+}

internal/dialect/prompt_cache_test.go

@@ -0,0 +1,49 @@
+package dialect
+
+import (
+	"encoding/json"
+	"net/http"
+	"strings"
+	"testing"
+)
+
+func TestPromptCacheKeyValidationAndProtocolScope(t *testing.T) {
+	cases := []struct{ name, value, want string }{
+		{"valid", `"cache-a"`, "cache-a"}, {"unicode", `"缓存分组"`, "缓存分组"},
+		{"empty", `""`, ""}, {"null", `null`, ""}, {"number", `42`, ""}, {"object", `{}`, ""},
+		{"leading space", `" cache-a"`, ""}, {"trailing space", `"cache-a "`, ""}, {"control", `"cache\u0000a"`, ""},
+		{"limit", `"` + strings.Repeat("a", 256) + `"`, strings.Repeat("a", 256)}, {"too long", `"` + strings.Repeat("a", 257) + `"`, ""},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			body := []byte(`{"model":"test","prompt_cache_key":` + tc.value + `,"messages":[{"role":"user","content":"hello"}],"input":"hello"}`)
+			for _, d := range []Dialect{NewOpenAI(), NewOpenAIResponses()} {
+				path := "/v1/chat/completions"
+				if _, ok := d.(*OpenAIResponses); ok {
+					path = "/v1/responses"
+				}
+				metadata, err := d.InspectRequest(&ParsedRequest{Method: http.MethodPost, Path: path, Body: body})
+				if err != nil || metadata.PromptCacheKey != tc.want {
+					t.Fatalf("key = %q, error = %v", metadata.PromptCacheKey, err)
+				}
+				encoded, _ := json.Marshal(metadata)
+				if strings.Contains(string(encoded), "PromptCacheKey") {
+					t.Fatal("private key serialized")
+				}
+			}
+		})
+	}
+	for _, body := range []string{
+		`{"prompt_cache_key":"a","prompt_cache_key":"b"}`, `{"metadata":{"prompt_cache_key":"a"}}`, `{"Prompt_Cache_Key":"a"}`,
+	} {
+		if inspectPromptCacheKey([]byte(body)) != "" {
+			t.Fatal("ambiguous or nested key accepted")
+		}
+	}
+	for _, path := range []string{"/v1/responses/compact", "/v1/responses/resp-test"} {
+		metadata, err := NewOpenAIResponses().InspectRequest(&ParsedRequest{Method: http.MethodPost, Path: path, Body: []byte(`{"model":"test","input":"hello","prompt_cache_key":"a"}`)})
+		if err != nil || metadata.PromptCacheKey != "" {
+			t.Fatalf("resource request parsed affinity key: %v", err)
+		}
+	}
+}

internal/execution/cpa/codex_cache_test.go

@@ -0,0 +1,131 @@
+package cpa
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"gpt-load/internal/channel"
+	"gpt-load/internal/execution"
+	"gpt-load/internal/protocol"
+	subscriptionruntime "gpt-load/internal/subscription/runtime"
+)
+
+func TestCodexCacheIdentityAcrossProtocols(t *testing.T) {
+	for _, stream := range []bool{false, true} {
+		for _, tc := range []struct {
+			protocol      protocol.Protocol
+			path, payload string
+		}{
+			{protocol.OpenAICompletions, "/v1/chat/completions", `{"model":"gpt-5.6-luna","messages":[{"role":"system","content":"stable"},{"role":"user","content":"hello"}]}`},
+			{protocol.OpenAIResponses, "/v1/responses", `{"model":"gpt-5.6-luna","instructions":"stable","input":"hello"}`},
+			{protocol.Anthropic, "/v1/messages", `{"model":"gpt-5.6-luna","max_tokens":64,"system":"stable","messages":[{"role":"user","content":"hello"}]}`},
+			{protocol.Gemini, "/v1beta/models/gpt-5.6-luna:generateContent", `{"contents":[{"role":"user","parts":[{"text":"hello"}]}]}`},
+			{protocol.OpenAICompletions, "/v1/chat/completions", `{"model":"gpt-5.6-luna","prompt_cache_key":"caller-cache","messages":[{"role":"user","content":"hello"}]}`},
+			{protocol.OpenAIResponses, "/v1/responses", `{"model":"gpt-5.6-luna","prompt_cache_key":"caller-cache","input":"hello"}`},
+		} {
+			t.Run(fmt.Sprintf("%s/stream=%t", tc.protocol, stream), func(t *testing.T) {
+				canonical := credentialJSON("access", "refresh", time.Now().Add(time.Hour))
+				adapter := NewAdapter(nil, channel.NewRegistry())
+				adapter.credentials = &fakeCredentialPreparer{credential: subscriptionruntime.NewCredential(canonical, "cache-account", subscriptionruntime.Account{}, time.Time{}, false, nil)}
+				var keys, sessions []string
+				transport := roundTripperFunc(func(req *http.Request) (*http.Response, error) {
+					body, err := io.ReadAll(req.Body)
+					if err != nil {
+						return nil, err
+					}
+					var obj map[string]any
+					if err := json.Unmarshal(body, &obj); err != nil {
+						return nil, err
+					}
+					key, _ := obj["prompt_cache_key"].(string)
+					keys = append(keys, key)
+					sessions = append(sessions, req.Header.Get("Session-Id"))
+					return &http.Response{StatusCode: 200, Header: http.Header{"Content-Type": {"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(`data: {"type":"response.completed","response":{"id":"resp_cache","object":"response","status":"completed","model":"gpt-5.6-luna","output":[],"usage":{"input_tokens":8192,"output_tokens":4,"total_tokens":8196,"input_tokens_details":{"cached_tokens":4096}}}}` + "\n\n")), Request: req}, nil
+				})
+				ctx := context.WithValue(t.Context(), "cliproxy.roundtripper", http.RoundTripper(transport))
+				for _, scope := range []string{"tenant-a-prefix", "tenant-a-prefix", "tenant-b-prefix"} {
+					operation := execution.OperationChatCompletion
+					mode := execution.RouteConverted
+					if tc.protocol == protocol.OpenAIResponses {
+						operation = execution.OperationResponsesCreate
+						mode = execution.RouteNative
+					}
+					body := []byte(tc.payload)
+					spec := execution.NewAttemptSpec(execution.AttemptSpec{RequestID: "cache-test", AttemptID: "cache-attempt", Sequence: 1, ChannelID: string(channel.Codex), RouteMode: mode, ClientProtocol: tc.protocol, Operation: operation, ClientModel: "gpt-5.6-luna", UpstreamModel: "gpt-5.6-luna", Method: http.MethodPost, Path: tc.path, Body: body, ContinuityKey: scope, Credential: execution.NewCredentialSnapshot(1, 1, 1, canonical)})
+					var evidence *execution.ErrorEvidence
+					if stream {
+						evidence = adapter.ExecuteStream(ctx, spec, func(execution.StreamEvent) error { return nil }).Error
+					} else {
+						evidence = adapter.Execute(ctx, spec).Error
+					}
+					if evidence != nil {
+						t.Fatalf("execution failed: %+v", evidence)
+					}
+					if !bytes.Equal(body, spec.Body) {
+						t.Fatal("source body mutated")
+					}
+				}
+				explicit := strings.Contains(tc.payload, "caller-cache")
+				if len(keys) != 3 || keys[0] == "" || keys[0] != keys[1] || (!explicit && keys[0] == keys[2]) || (explicit && (keys[0] != "caller-cache" || keys[2] != "caller-cache")) {
+					t.Fatalf("unstable or unisolated cache keys: %v", keys)
+				}
+				for i := range keys {
+					if keys[i] != sessions[i] {
+						t.Fatal("cache key and session disagree")
+					}
+				}
+			})
+		}
+	}
+}
+
+func TestCodexCallerSessionIdentityPrecedesDerivedCacheGroup(t *testing.T) {
+	for _, stream := range []bool{false, true} {
+		for _, claude := range []bool{false, true} {
+			t.Run(fmt.Sprintf("claude=%t/stream=%t", claude, stream), func(t *testing.T) {
+				adapter, _, _, service, row := newAdapterFixture(t, credentialJSON("access", "refresh", time.Now().Add(time.Hour)))
+				var sessions []string
+				transport := roundTripperFunc(func(req *http.Request) (*http.Response, error) {
+					sessions = append(sessions, req.Header.Get("Session-Id"))
+					return &http.Response{StatusCode: 200, Header: http.Header{"Content-Type": {"text/event-stream"}}, Body: io.NopCloser(strings.NewReader(`data: {"type":"response.completed","response":{"id":"resp_identity","object":"response","status":"completed","model":"gpt-5","output":[]}}` + "\n\n")), Request: req}, nil
+				})
+				ctx := context.WithValue(t.Context(), "cliproxy.roundtripper", http.RoundTripper(transport))
+				for _, scope := range []string{"prefix-a", "prefix-b"} {
+					spec := validSpec(t, row, service)
+					spec.ContinuityKey = scope
+					spec.Header = http.Header{"Session-Id": {"caller-session"}}
+					if claude {
+						spec.ClientProtocol = protocol.Anthropic
+						spec.Operation = execution.OperationChatCompletion
+						spec.RouteMode = execution.RouteConverted
+						spec.Path = "/v1/messages"
+						spec.Body = []byte(`{"model":"gpt-5","max_tokens":64,"messages":[{"role":"user","content":"hello"}]}`)
+						spec.Header = http.Header{"X-Claude-Code-Session-Id": {"claude-session"}}
+					}
+					var evidence *execution.ErrorEvidence
+					if stream {
+						evidence = adapter.ExecuteStream(ctx, spec, func(execution.StreamEvent) error { return nil }).Error
+					} else {
+						evidence = adapter.Execute(ctx, spec).Error
+					}
+					if evidence != nil {
+						t.Fatalf("execute: %+v", evidence)
+					}
+				}
+				if len(sessions) != 2 || sessions[0] == "" || sessions[0] != sessions[1] {
+					t.Fatal("caller identity overwritten by derived cache group")
+				}
+				if !claude && sessions[0] != "caller-session" {
+					t.Fatal("caller session changed")
+				}
+			})
+		}
+	}
+}

internal/execution/cpa/codex_provider.go

@@ -305,8 +305,9 @@ func (bridge *codexProviderBridge) Execute(
 	}
 	response, err := bridge.executor.Execute(ctx, credentialID, codexCredential.value, codex.ExecuteRequest{
 		Model: request.Model, Payload: append([]byte(nil), request.Payload...), Format: request.Format,
-		RequestPath: request.RequestPath,
-		Headers:     request.Headers.Clone(), OriginalRequest: append([]byte(nil), request.OriginalRequest...),
+		ContinuityKey: request.ContinuityKey,
+		RequestPath:   request.RequestPath,
+		Headers:       request.Headers.Clone(), OriginalRequest: append([]byte(nil), request.OriginalRequest...),
 		ConfiguredHeaders: append([]string(nil), request.ConfiguredHeaders...),
 		BaseURL:           request.BaseURL, ProxyURL: request.ProxyURL, ProxyFromEnvironment: request.ProxyFromEnvironment,
 	})
@@ -333,8 +334,9 @@ func (bridge *codexProviderBridge) ExecuteStream(
 	}
 	response, err := bridge.executor.ExecuteStream(ctx, credentialID, codexCredential.value, codex.ExecuteRequest{
 		Model: request.Model, Payload: append([]byte(nil), request.Payload...), Format: request.Format,
-		RequestPath: request.RequestPath,
-		Headers:     request.Headers.Clone(), OriginalRequest: append([]byte(nil), request.OriginalRequest...),
+		ContinuityKey: request.ContinuityKey,
+		RequestPath:   request.RequestPath,
+		Headers:       request.Headers.Clone(), OriginalRequest: append([]byte(nil), request.OriginalRequest...),
 		ConfiguredHeaders: append([]string(nil), request.ConfiguredHeaders...),
 		BaseURL:           request.BaseURL, ProxyURL: request.ProxyURL, ProxyFromEnvironment: request.ProxyFromEnvironment,
 	})

internal/gateway/affinity.go

@@ -5,6 +5,7 @@ import (
 	"gpt-load/internal/protocol"
 	"gpt-load/internal/scheduler"
 	"gpt-load/internal/state"
+	"gpt-load/internal/telemetry"
 )
 
 type requestAffinity struct {
@@ -12,6 +13,7 @@ type requestAffinity struct {
 	observation           affinity.Observation
 	preferredCredentialID uint
 	continuityKey         string
+	kind                  string
 }
 
 func (handler *Handler) resolveRequestAffinity(
@@ -20,6 +22,7 @@ func (handler *Handler) resolveRequestAffinity(
 	clientProtocol protocol.Protocol,
 	prefix []byte,
 	allowedCredentialRefs map[uint]state.CredentialRef,
+	promptCacheKey string,
 ) requestAffinity {
 	if handler == nil || snapshot == nil {
 		return requestAffinity{}
@@ -30,7 +33,12 @@ func (handler *Handler) resolveRequestAffinity(
 		clientProtocol,
 		prefix,
 	)
-	result := requestAffinity{continuityKey: string(key)}
+	// 执行层私有 replay scope 仍由提示词派生，不把客户端缓存分组当作会话身份。
+	result := requestAffinity{continuityKey: string(key), kind: telemetry.AffinityPromptPrefix}
+	if promptCacheKey != "" {
+		key = affinity.DerivePromptCacheKey(handler.encryption, accessKeyID, clientProtocol, promptCacheKey)
+		result.kind = telemetry.AffinityPromptCacheKey
+	}
 	if handler.affinityCache == nil ||
 		!handler.affinityCache.Configure(
 			snapshot.Revision,

internal/gateway/affinity_test.go

@@ -255,6 +255,7 @@ func TestHandlerIgnoresAffinityAfterCredentialIdentityChanges(t *testing.T) {
 		protocol.OpenAICompletions,
 		prefix,
 		map[uint]state.CredentialRef{1: oldRef},
+		"",
 	)
 	if initial.preferredCredentialID != 0 || !initial.key.Valid() {
 		t.Fatalf("initial affinity = %#v, want valid miss", initial)
@@ -273,6 +274,7 @@ func TestHandlerIgnoresAffinityAfterCredentialIdentityChanges(t *testing.T) {
 		protocol.OpenAICompletions,
 		prefix,
 		map[uint]state.CredentialRef{1: oldRef},
+		"",
 	)
 	if hit.preferredCredentialID != 1 {
 		t.Fatalf("preferred credential = %d, want 1", hit.preferredCredentialID)
@@ -285,6 +287,7 @@ func TestHandlerIgnoresAffinityAfterCredentialIdentityChanges(t *testing.T) {
 		protocol.OpenAICompletions,
 		prefix,
 		map[uint]state.CredentialRef{1: changedRef},
+		"",
 	)
 	if stale.preferredCredentialID != 0 {
 		t.Fatalf("preferred credential after identity change = %d, want 0", stale.preferredCredentialID)
@@ -301,6 +304,7 @@ func TestHandlerDerivesPrivateContinuityWithoutReenablingDisabledAffinity(t *tes
 		protocol.OpenAICompletions,
 		[]byte(`{"v":1,"user":["hello"]}`),
 		map[uint]state.CredentialRef{1: {ID: 1, GroupID: 1, IdentityGeneration: 1}},
+		"",
 	)
 	if resolved.key.Valid() || resolved.preferredCredentialID != 0 || resolved.continuityKey == "" {
 		t.Fatalf("disabled affinity resolution = %#v", resolved)
@@ -447,3 +451,61 @@ func assertAffinityUpstreamModels(t *testing.T, inputs []ForwardInput, want []st
 		t.Fatalf("upstream models = %#v, want %#v", got, want)
 	}
 }
+
+func TestPromptCacheKeyAffinityPrecedesPromptPrefix(t *testing.T) {
+	forwarder := &scriptedForwarder{results: successfulAffinityResults(4)}
+	handler, _, _ := newHandlerForTest(t, forwarder, "sk-one", "sk-two")
+	sink := &recordingRequestLogSink{}
+	handler.requestLogSink = sink
+	engine := newAffinityTestEngine(t, handler)
+	for _, body := range []string{
+		`{"model":"gpt-4o","prompt_cache_key":"cache-a","messages":[{"role":"user","content":"first"}]}`,
+		`{"model":"gpt-4o","prompt_cache_key":"cache-a","messages":[{"role":"user","content":"changed"}]}`,
+		`{"model":"gpt-4o","prompt_cache_key":"cache-b","messages":[{"role":"user","content":"changed"}]}`,
+		`{"model":"gpt-4o","prompt_cache_key":"cache-b","messages":[{"role":"user","content":"another"}]}`,
+	} {
+		serveAffinityRequest(t, engine, body)
+	}
+	assertAffinityAttemptKeys(t, forwarder.inputs, []string{"sk-one", "sk-one", "sk-two", "sk-two"})
+	assertAffinityHits(t, sink.snapshot(), []bool{false, true, false, true})
+	for _, i := range []int{1, 3} {
+		if sink.snapshot()[i].AffinityKind != telemetry.AffinityPromptCacheKey {
+			t.Fatal("wrong cache affinity kind")
+		}
+	}
+	// 客户端缓存分组不改变 provider-private replay 的提示词隔离。
+	if forwarder.inputs[0].ContinuityKey == forwarder.inputs[1].ContinuityKey {
+		t.Fatal("cache key became replay identity")
+	}
+}
+
+func TestPromptCacheKeyAffinityRespectsGroupSwitch(t *testing.T) {
+	forwarder := &scriptedForwarder{results: successfulAffinityResults(2)}
+	handler, manager, _ := newHandlerForTest(t, forwarder, "sk-one", "sk-two")
+	group := manager.Current().Groups[1]
+	group.AffinityEnabled = false
+	manager.Current().Groups[1] = group
+	engine := newAffinityTestEngine(t, handler)
+	for range 2 {
+		serveAffinityRequest(t, engine, `{"model":"gpt-4o","prompt_cache_key":"cache-a","messages":[{"role":"user","content":"same"}]}`)
+	}
+	assertAffinityAttemptKeys(t, forwarder.inputs, []string{"sk-one", "sk-two"})
+}
+
+func TestInvalidPromptCacheKeyFallsBackWithoutMutatingRequest(t *testing.T) {
+	forwarder := &scriptedForwarder{results: successfulAffinityResults(2)}
+	handler, _, _ := newHandlerForTest(t, forwarder, "sk-one", "sk-two")
+	sink := &recordingRequestLogSink{}
+	handler.requestLogSink = sink
+	engine := newAffinityTestEngine(t, handler)
+	body := `{"model":"gpt-4o","prompt_cache_key":" invalid ","messages":[{"role":"user","content":"hello"}]}`
+	serveAffinityRequest(t, engine, body)
+	serveAffinityRequest(t, engine, body)
+	assertAffinityAttemptKeys(t, forwarder.inputs, []string{"sk-one", "sk-one"})
+	if sink.snapshot()[1].AffinityKind != telemetry.AffinityPromptPrefix {
+		t.Fatal("invalid key did not fall back to prefix")
+	}
+	if !bytes.Contains(forwarder.inputs[1].Request.Body, []byte(`"prompt_cache_key":" invalid "`)) {
+		t.Fatal("client cache parameter changed")
+	}
+}

internal/gateway/handler.go

@@ -631,7 +631,7 @@ func (handler *Handler) Handle(ginContext *gin.Context) {
 		}
 	} else {
 		requestAffinity = handler.resolveRequestAffinity(
-			snapshot, accessKey.ID, selectedRoute.Protocol, metadata.AffinityPrefix, allowedCredentialRefs,
+			snapshot, accessKey.ID, selectedRoute.Protocol, metadata.AffinityPrefix, allowedCredentialRefs, metadata.PromptCacheKey,
 		)
 		query.PreferredCredentialID = requestAffinity.preferredCredentialID
 	}
@@ -933,7 +933,11 @@ func (handler *Handler) executeAttempts(
 		attemptSequence++
 		if attemptSequence == 1 && (originalMetadata.PreviousResponseID != "" ||
 			(requestAffinity.preferredCredentialID != 0 && selection.CredentialID == requestAffinity.preferredCredentialID)) {
-			recorder.setAffinityHit(true)
+			kind := requestAffinity.kind
+			if originalMetadata.PreviousResponseID != "" {
+				kind = telemetry.AffinityResponseContinuity
+			}
+			recorder.setAffinityHit(true, kind)
 		}
 		updateDebugHeaders(ginContext.Writer.Header(), selection.Group.Name, attemptSequence)
 		if recorder != nil {
@@ -1144,7 +1148,11 @@ func (handler *Handler) executeAttempts(
 		forwardAttempts++
 		if attemptSequence == 1 && (originalMetadata.PreviousResponseID != "" ||
 			(requestAffinity.preferredCredentialID != 0 && selection.CredentialID == requestAffinity.preferredCredentialID)) {
-			recorder.setAffinityHit(true)
+			kind := requestAffinity.kind
+			if originalMetadata.PreviousResponseID != "" {
+				kind = telemetry.AffinityResponseContinuity
+			}
+			recorder.setAffinityHit(true, kind)
 		}
 		updateDebugHeaders(ginContext.Writer.Header(), selection.Group.Name, attemptSequence)
 		executionRequestID := "untracked"

internal/gateway/request_log.go

@@ -67,6 +67,7 @@ type requestRecorder struct {
 	requestedPricingMode pricing.Mode
 	usageDiagnostics     usage.Diagnostics
 	affinityHit          bool
+	affinityKind         string
 	attempts             []telemetry.Attempt
 	attemptPricing       []frozenAttemptPricing
 	pendingPricing       frozenAttemptPricing
@@ -147,6 +148,7 @@ func (recorder *requestRecorder) emit() {
 		FirstResponseMs:       recorder.firstResponseMs,
 		DurationMs:            duration.Milliseconds(),
 		AffinityHit:           recorder.affinityHit,
+		AffinityKind:          recorder.affinityKind,
 		Reasoning:             recorder.reasoning,
 		Operation:             recorder.operation,
 		Attempts:              append([]telemetry.Attempt(nil), recorder.attempts...),
@@ -179,9 +181,10 @@ func (recorder *requestRecorder) estimatedCostNanoUSD() int64 {
 	return recorder.usage.Pricing.EstimatedCostNanoUSD
 }
 
-func (recorder *requestRecorder) setAffinityHit(hit bool) {
+func (recorder *requestRecorder) setAffinityHit(hit bool, kind string) {
 	if recorder != nil && hit {
 		recorder.affinityHit = true
+		recorder.affinityKind = kind
 	}
 }
 

internal/gateway/responses_continuation_test.go

@@ -20,6 +20,7 @@ import (
 	"gpt-load/internal/execution"
 	"gpt-load/internal/parameteroverride"
 	"gpt-load/internal/state"
+	"gpt-load/internal/telemetry"
 )
 
 func TestResponsesContinuationPinsCredentialWithoutSoftAffinity(t *testing.T) {
@@ -35,11 +36,14 @@ func TestResponsesContinuationPinsCredentialWithoutSoftAffinity(t *testing.T) {
 			handler.manager.Current().Settings.AffinityEnabled = enabled
 
 			serveContinuation(t, engine, "gl-client", `{"model":"gpt-4o","input":"initial","store":true}`, http.StatusOK)
-			serveContinuation(t, engine, "gl-client", `{"model":"gpt-4o","input":"continue","previous_response_id":"first"}`, http.StatusOK)
+			serveContinuation(t, engine, "gl-client", `{"model":"gpt-4o","input":"continue","previous_response_id":"first","prompt_cache_key":"other-cache"}`, http.StatusOK)
 			serveContinuation(t, engine, "gl-client", `{"model":"gpt-4o","input":"continue"}`, http.StatusOK)
 
 			assertAffinityAttemptKeys(t, forwarder.inputs, []string{"sk-one", "sk-one", "sk-two"})
 			assertAffinityHits(t, sink.snapshot(), []bool{false, true, false})
+			if sink.snapshot()[1].AffinityKind != telemetry.AffinityResponseContinuity {
+				t.Fatal("missing continuation kind")
+			}
 			if got := handler.registry.SchedulingState().CaptureCheckpoint().Sequence; got != 3 {
 				t.Fatalf("scheduling allocations = %d, want 3", got)
 			}
@@ -387,3 +391,27 @@ func serveContinuation(t *testing.T, engine http.Handler, key, body string, stat
 	}
 	return response
 }
+
+func TestResponsesPromptCacheKeyAffinityAndHardContinuationPriority(t *testing.T) {
+	forwarder := &scriptedForwarder{results: []UpstreamResult{storedResponse("first"), storedResponse("second"), storedResponse("third"), storedResponse("fourth"), storedResponse("fifth")}}
+	handler, engine, sink := newContinuationFixture(t, forwarder)
+	group := handler.manager.Current().Groups[1]
+	group.AffinityEnabled = true
+	handler.manager.Current().Groups[1] = group
+	for _, body := range []string{
+		`{"model":"gpt-4o","input":"first-prefix","prompt_cache_key":"a"}`,
+		`{"model":"gpt-4o","input":"changed-prefix","prompt_cache_key":"a"}`,
+		`{"model":"gpt-4o","input":"changed-prefix","prompt_cache_key":"b"}`,
+		`{"model":"gpt-4o","input":"continue","previous_response_id":"first","prompt_cache_key":"b"}`,
+		`{"model":"gpt-4o","input":"another-prefix","prompt_cache_key":"b"}`,
+	} {
+		serveContinuation(t, engine, "gl-client", body, http.StatusOK)
+	}
+	assertAffinityAttemptKeys(t, forwarder.inputs, []string{"sk-one", "sk-one", "sk-two", "sk-one", "sk-two"})
+	assertAffinityHits(t, sink.snapshot(), []bool{false, true, false, true, true})
+	for i, want := range []string{"", telemetry.AffinityPromptCacheKey, "", telemetry.AffinityResponseContinuity, telemetry.AffinityPromptCacheKey} {
+		if sink.snapshot()[i].AffinityKind != want {
+			t.Fatalf("event %d kind=%q want=%q", i, sink.snapshot()[i].AffinityKind, want)
+		}
+	}
+}

internal/gateway/websocket_test.go

@@ -1069,3 +1069,53 @@ func TestWebsocketFirstEventTimeoutClosesSession(t *testing.T) {
 		t.Fatal("first-event deadline did not stop upstream")
 	}
 }
+
+func TestWebsocketCacheAffinityAndConnectionBindingKinds(t *testing.T) {
+	var ids atomic.Int64
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := (&websocket.Upgrader{}).Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer conn.Close()
+		for {
+			if _, _, err := conn.ReadMessage(); err != nil {
+				return
+			}
+			if err := conn.WriteMessage(websocket.TextMessage, websocketCompleted(fmt.Sprintf("cache-%d", ids.Add(1)), "")); err != nil {
+				return
+			}
+		}
+	}))
+	defer upstream.Close()
+	h, engine, _ := websocketTestHandler(t, upstream.URL+"/v1", channel.OpenAI)
+	sink := &recordingRequestLogSink{}
+	h.requestLogSink = sink
+	server := httptest.NewServer(engine)
+	defer server.Close()
+	send := func(conn *websocket.Conn, key, prefix string, count int) {
+		t.Helper()
+		if err := conn.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"type":"response.create","model":"public","input":%q,"prompt_cache_key":%q}`, prefix, key))); err != nil {
+			t.Fatal(err)
+		}
+		if _, _, err := conn.ReadMessage(); err != nil {
+			t.Fatal(err)
+		}
+		waitWebsocketLogs(t, sink, count)
+	}
+	first := dialGatewayWebsocket(t, server.URL)
+	send(first, "a", "initial", 1)
+	first.Close()
+	second := dialGatewayWebsocket(t, server.URL)
+	defer second.Close()
+	send(second, "a", "different", 2)
+	send(second, "b", "bound turn", 3)
+	third := dialGatewayWebsocket(t, server.URL)
+	defer third.Close()
+	send(third, "b", "another", 4)
+	events := sink.snapshot()
+	assertAffinityHits(t, events, []bool{false, true, true, false})
+	if events[1].AffinityKind != telemetry.AffinityPromptCacheKey || events[2].AffinityKind != telemetry.AffinityResponseContinuity {
+		t.Fatal("WebSocket affinity kind mismatch")
+	}
+}

internal/gateway/websocket_turn.go

@@ -23,6 +23,7 @@ import (
 	"gpt-load/internal/protocol"
 	"gpt-load/internal/scheduler"
 	"gpt-load/internal/state"
+	"gpt-load/internal/telemetry"
 )
 
 var reasonWebsocketCapability = reason{400, "websocket_capability_unavailable", "The selected upstream does not support this WebSocket capability."}
@@ -235,7 +236,7 @@ func (s *websocketConnection) executeTurn(turn websocketTurn) {
 		}
 		return
 	}
-	affinity := h.resolveRequestAffinity(snapshot, key.ID, protocol.OpenAIResponses, original.metadata.AffinityPrefix, query.AllowedCredentialRefs)
+	affinity := h.resolveRequestAffinity(snapshot, key.ID, protocol.OpenAIResponses, original.metadata.AffinityPrefix, query.AllowedCredentialRefs, original.metadata.PromptCacheKey)
 	if requiredRef == nil {
 		query.PreferredCredentialID = affinity.preferredCredentialID
 	}
@@ -357,7 +358,13 @@ func (s *websocketConnection) executeTurn(turn websocketTurn) {
 		recorder.setPricingMode(effective.metadata.PricingMode)
 		recorder.setUsageDiagnostics(effective.metadata.UsageDiagnostics)
 		recorder.freezeNextAttemptPricing(h.freezeAttemptPricing(selection, effective.metadata, true, key.PriceMultiplier))
-		recorder.setAffinityHit(requiredRef != nil || selection.CredentialID == affinity.preferredCredentialID)
+		if sequence == 1 {
+			kind := affinity.kind
+			if requiredRef != nil {
+				kind = telemetry.AffinityResponseContinuity
+			}
+			recorder.setAffinityHit(requiredRef != nil || selection.CredentialID == affinity.preferredCredentialID, kind)
+		}
 		started := recorder.beforeForward()
 		ctx, cancel := context.WithTimeout(s.ctx, selection.Group.Timeouts.Request)
 		var firstByteDeadline time.Time
@@ -429,7 +436,7 @@ func (s *websocketConnection) executeTurn(turn websocketTurn) {
 		h.applyGroupDecisionEffect(selection.Group, ref, 0, decision, result.StatusCode, h.now(), input.UpstreamModelID)
 		if result.Stream.EndReason == StreamEndCleanEOF {
 			h.recordCredentialSuccess(ref, h.now())
-			if original.previous == "" {
+			if requiredRef == nil {
 				h.recordAffinitySuccess(affinity, selection, ref)
 			}
 		}

internal/requestlog/mapper.go

@@ -128,6 +128,7 @@ func mapEvent(
 		ErrorCode:               event.ErrorCode,
 		ErrorSummary:            sanitizeSummary(redactor, event.ErrorSummary),
 		AffinityHit:             event.AffinityHit,
+		AffinityKind:            event.AffinityKind,
 		ReasoningMode:           event.Reasoning.Mode,
 		ReasoningEffort:         event.Reasoning.Effort,
 		ReasoningBudgetTokens:   event.Reasoning.BudgetTokens,

internal/requestlog/phase4_v3_test.go

@@ -23,6 +23,7 @@ func TestMapEventV3PersistsMillisecondLedgerAndQuotesUpstreamModel(t *testing.T)
 		ModelConsistency:      telemetry.ModelConsistencyMatch,
 		Status:                telemetry.RequestStatusSuccess,
 		StatusCode:            200,
+		AffinityHit:           true, AffinityKind: telemetry.AffinityPromptCacheKey,
 		Attempts: []telemetry.Attempt{{
 			Sequence: 1, GroupID: 73, ChannelID: channel.OpenAI, CredentialID: 9,
 			Operation: execution.OperationChatCompletion, RouteMode: channel.RouteNative,
@@ -43,6 +44,9 @@ func TestMapEventV3PersistsMillisecondLedgerAndQuotesUpstreamModel(t *testing.T)
 	}
 
 	row := mustMapEvent(t, redact.New(), event)
+	if !row.AffinityHit || row.AffinityKind != telemetry.AffinityPromptCacheKey {
+		t.Fatal("affinity kind lost in persistence mapping")
+	}
 
 	if row.CompletedAtMS != 1_784_896_496_789 {
 		t.Fatalf("CompletedAtMS = %d, want 1784896496789", row.CompletedAtMS)

internal/requestlog/query.go

@@ -332,6 +332,7 @@ func decodeRequestLogRows(rows []models.RequestLog) ([]Record, error) {
 			ErrorCode:             row.ErrorCode,
 			ErrorSummary:          row.ErrorSummary,
 			AffinityHit:           row.AffinityHit,
+			AffinityKind:          row.AffinityKind,
 			Reasoning: reasoning.Config{
 				Mode:         row.ReasoningMode,
 				Effort:       row.ReasoningEffort,

internal/requestlog/query_test.go

@@ -972,3 +972,23 @@ func containsJSONFragment(encoded []byte, fragment string) bool {
 	}
 	return false
 }
+
+func TestServiceListAndDetailPreserveAffinityKind(t *testing.T) {
+	db := openRequestLogQueryDB(t)
+	row := requestLogQueryRow("00000000-0000-4000-8000-000000000630", time.Now(), 71, "model", nil)
+	row.AffinityHit = true
+	row.AffinityKind = telemetry.AffinityPromptCacheKey
+	createRequestLogQueryRow(t, db, row)
+	service := newRequestLogTestService(db)
+	page, err := service.List(context.Background(), ListQuery{Limit: 50})
+	if err != nil {
+		t.Fatal(err)
+	}
+	detail, err := service.Get(context.Background(), row.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(page.Items) != 1 || page.Items[0].AffinityKind != row.AffinityKind || detail.AffinityKind != row.AffinityKind {
+		t.Fatal("affinity kind lost on query")
+	}
+}

internal/requestlog/types.go

@@ -134,6 +134,7 @@ type Record struct {
 	ErrorCode               string
 	ErrorSummary            string
 	AffinityHit             bool
+	AffinityKind            string
 	Reasoning               reasoning.Config
 	Attempts                []Attempt
 	GroupID                 uint

internal/storage/affinity_kind_migration_test.go

@@ -0,0 +1,77 @@
+package storage
+
+import (
+	"fmt"
+	"os"
+	"testing"
+
+	"gorm.io/gorm"
+)
+
+func TestAffinityKindMigrationContract(t *testing.T) {
+	testAffinityKindMigration(t, openInternalMigrationTestDatabase)
+}
+func TestExternalAffinityKindMigrationContract(t *testing.T) {
+	dsn := os.Getenv("GPT_LOAD_DATABASE_TEST_DSN")
+	if dsn == "" {
+		t.Skip("GPT_LOAD_DATABASE_TEST_DSN is not set")
+	}
+	testAffinityKindMigration(t, func(t *testing.T) *gorm.DB { return openExternalIncrementalMigrationDatabase(t, dsn) })
+}
+func testAffinityKindMigration(t *testing.T, open func(*testing.T) *gorm.DB) {
+	for _, scenario := range []string{"fresh", "upgrade", "interrupted"} {
+		t.Run(scenario, func(t *testing.T) {
+			db := open(t)
+			if scenario != "fresh" {
+				if err := applyMigrationRegistry(db, migrations[:13]); err != nil {
+					t.Fatal(err)
+				}
+				if err := db.Table("request_logs").Create(map[string]any{
+					"id": "existing", "completed_at_ms": 1, "access_key_id": 1, "protocol": "openai-responses", "client_model": "test", "upstream_model": "test", "status": "success", "status_code": 200, "duration_ms": 1, "error_summary": "", "affinity_hit": true,
+				}).Error; err != nil {
+					t.Fatal(err)
+				}
+			}
+			if scenario == "interrupted" && len(migrations) > 13 {
+				entry := migrations[13]
+				up := entry.Up
+				entry.Up = func(tx *gorm.DB) error {
+					if err := up(tx); err != nil {
+						return err
+					}
+					return fmt.Errorf("interrupt after affinity DDL")
+				}
+				registry := append(append([]migration(nil), migrations[:13]...), entry)
+				if err := applyMigrationRegistry(db, registry); err == nil {
+					t.Fatal("expected interruption")
+				}
+			}
+			if err := AutoMigrate(db); err != nil {
+				t.Fatal(err)
+			}
+			if !db.Migrator().HasColumn("request_logs", "affinity_kind") {
+				t.Fatal("affinity_kind is missing")
+			}
+			if scenario != "fresh" {
+				var row struct {
+					AffinityHit  bool
+					AffinityKind string
+				}
+				if err := db.Table("request_logs").Where("id = ?", "existing").Take(&row).Error; err != nil {
+					t.Fatal(err)
+				}
+				if !row.AffinityHit || row.AffinityKind != "" {
+					t.Fatalf("legacy affinity changed: %#v", row)
+				}
+			}
+			if err := db.Table("request_logs").Create(map[string]any{
+				"id": "new", "completed_at_ms": 1, "access_key_id": 1, "protocol": "openai-responses", "client_model": "test", "upstream_model": "test", "status": "success", "status_code": 200, "duration_ms": 1, "error_summary": "", "affinity_hit": true, "affinity_kind": "prompt_cache_key",
+			}).Error; err != nil {
+				t.Fatal(err)
+			}
+			if err := AutoMigrate(db); err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}

internal/storage/database_integration_test.go

@@ -123,7 +123,7 @@ func TestExternalDatabaseLifecycle(t *testing.T) {
 	if err := db.Table("schema_migrations").Order("id").Pluck("id", &migrationIDs).Error; err != nil {
 		t.Fatalf("read migration ledger: %v", err)
 	}
-	if len(migrationIDs) != 13 || migrationIDs[0] != "0001_initial" ||
+	if len(migrationIDs) != 14 || migrationIDs[0] != "0001_initial" ||
 		migrationIDs[1] != "0002_access_key_cost_limits" ||
 		migrationIDs[2] != "0003_remove_observation_fresh_until" ||
 		migrationIDs[3] != "0004_usage_stats_group_activity_index" ||
@@ -131,8 +131,8 @@ func TestExternalDatabaseLifecycle(t *testing.T) {
 		migrationIDs[5] != "0006_error_decision" ||
 		migrationIDs[6] != "0007_access_key_lifecycle" ||
 		migrationIDs[7] != "0008_remove_inject_usage_options" ||
-		migrationIDs[8] != "0009_price_multipliers" || migrationIDs[9] != "0010_model_cooldown" || migrationIDs[10] != "0011_custom_access_keys" || migrationIDs[11] != "0012_access_key_mask_prefix" || migrationIDs[12] != "0013_validation_protocol" {
-		t.Fatalf("migration ledger = %v, want complete 0001-0013 chain", migrationIDs)
+		migrationIDs[8] != "0009_price_multipliers" || migrationIDs[9] != "0010_model_cooldown" || migrationIDs[10] != "0011_custom_access_keys" || migrationIDs[11] != "0012_access_key_mask_prefix" || migrationIDs[12] != "0013_validation_protocol" || migrationIDs[13] != "0014_affinity_kind" {
+		t.Fatalf("migration ledger = %v, want complete 0001-0014 chain", migrationIDs)
 	}
 	if !db.Migrator().HasIndex("usage_stats", "idx_usage_stats_group_bucket") {
 		t.Fatal("usage_stats group activity index is missing")

internal/storage/db_test.go

@@ -662,6 +662,7 @@ func TestAutoMigrateCreatesUsageJournalAndMigrationLedger(t *testing.T) {
 		"0011_custom_access_keys",
 		"0012_access_key_mask_prefix",
 		"0013_validation_protocol",
+		"0014_affinity_kind",
 	}
 	if !reflect.DeepEqual(migrationIDs, wantMigrationIDs) {
 		t.Fatalf("schema_migrations IDs = %v, want %v", migrationIDs, wantMigrationIDs)

internal/storage/migration.go

@@ -104,6 +104,7 @@ var migrations = []migration{
 		Validate: migrationfiles.Validate0012, ValidateRecoverable: migrationfiles.ValidateRecoverable0012,
 	},
 	{ID: migrationfiles.ID0013, Up: migrationfiles.Up0013, Validate: migrationfiles.Validate0013, ValidateRecoverable: migrationfiles.ValidateRecoverable0013},
+	{ID: migrationfiles.ID0014, Up: migrationfiles.Up0014, Validate: migrationfiles.Validate0014, ValidateRecoverable: migrationfiles.ValidateRecoverable0014},
 }
 
 func applyMigrations(db *gorm.DB) error {

internal/storage/migration_test.go

@@ -25,6 +25,7 @@ func TestMigrationRegistryContainsOrderedMigrations(t *testing.T) {
 		migrationfiles.ID0011,
 		migrationfiles.ID0012,
 		migrationfiles.ID0013,
+		migrationfiles.ID0014,
 	}
 	if len(migrations) != len(wantIDs) {
 		t.Fatalf("migration registry length = %d, want %d", len(migrations), len(wantIDs))

internal/storage/migrations/0006_error_decision_test.go

@@ -20,7 +20,7 @@ func TestErrorDecisionMigrationPreservesAttemptsAndAddsDecisionContract(t *testi
 		UsageState: "not_applicable", CostState: "not_applicable",
 		PricingCompleteness: "not_applicable",
 	}
-	if err := db.Create(&request).Error; err != nil {
+	if err := db.Omit("AffinityKind").Create(&request).Error; err != nil {
 		t.Fatalf("create request log: %v", err)
 	}
 	attempt := models.RequestLogAttempt{

internal/storage/migrations/0014_affinity_kind.go

@@ -0,0 +1,58 @@
+package migrations
+
+import (
+	"fmt"
+	"strings"
+
+	"gorm.io/gorm"
+)
+
+const ID0014 = "0014_affinity_kind"
+
+// Up0014 原子增加命中来源，旧日志保留空值，避免推测历史亲和类型。
+func Up0014(db *gorm.DB) error {
+	if err := ValidateRecoverable0014(db); err != nil {
+		return err
+	}
+	if !db.Migrator().HasColumn("request_logs", "affinity_kind") {
+		if err := db.Exec("ALTER TABLE request_logs ADD COLUMN affinity_kind VARCHAR(32) NOT NULL DEFAULT ''").Error; err != nil {
+			return fmt.Errorf("add affinity kind: %w", err)
+		}
+	}
+	return Validate0014(db)
+}
+func ValidateRecoverable0014(db *gorm.DB) error {
+	if !db.Migrator().HasTable("request_logs") {
+		return fmt.Errorf("request_logs table is missing")
+	}
+	if db.Migrator().HasColumn("request_logs", "affinity_kind") {
+		return Validate0014(db)
+	}
+	return nil
+}
+func Validate0014(db *gorm.DB) error {
+	columns, err := db.Migrator().ColumnTypes("request_logs")
+	if err != nil {
+		return err
+	}
+	for _, column := range columns {
+		if column.Name() != "affinity_kind" {
+			continue
+		}
+		if !strings.Contains(strings.ToLower(column.DatabaseTypeName()), "char") {
+			return fmt.Errorf("affinity kind must be varchar")
+		}
+		if nullable, known := column.Nullable(); !known || nullable {
+			return fmt.Errorf("affinity kind must be non-null")
+		}
+		if length, known := column.Length(); known && length != 32 {
+			return fmt.Errorf("affinity kind length must be 32")
+		}
+		value, known := column.DefaultValue()
+		if !known || (value != "" && value != "''" && value != "''::character varying") {
+			return fmt.Errorf("affinity kind must default to empty")
+		}
+		return nil
+	}
+	return fmt.Errorf("affinity kind is missing")
+}

internal/storage/model_cooldown_migration_test.go

@@ -71,7 +71,7 @@ func testModelCooldownMigration(t *testing.T, open func(*testing.T) *gorm.DB) {
 				Protocol: "openai-completions", ClientModel: "model", UpstreamModel: "model", ModelConsistency: "not_applicable",
 				Status: "error", StatusCode: 429, DurationMs: 1, UsageState: "not_applicable", CostState: "not_applicable", PricingCompleteness: "not_applicable"}
 			if scenario != "fresh" {
-				if err := db.Create(&legacyRequest).Error; err != nil {
+				if err := db.Omit("AffinityKind").Create(&legacyRequest).Error; err != nil {
 					t.Fatal(err)
 				}
 				legacy := models.RequestLogAttempt{RequestID: legacyRequest.ID, Sequence: 1, GroupID: 1, CredentialID: 1,

internal/storage/models/request.go

@@ -23,6 +23,7 @@ type RequestLog struct {
 	ErrorCode               string              `gorm:"type:varchar(64);not null;default:''"`
 	ErrorSummary            string              `gorm:"type:text;not null"`
 	AffinityHit             bool                `gorm:"not null;default:false"`
+	AffinityKind            string              `gorm:"type:varchar(32);not null;default:''"`
 	ReasoningMode           string              `gorm:"type:varchar(64);not null;default:''"`
 	ReasoningEffort         string              `gorm:"type:varchar(64);not null;default:''"`
 	ReasoningBudgetTokens   *int64              `gorm:"column:reasoning_budget_tokens"`

internal/subscription/providers/codex/codex.go

@@ -274,6 +274,7 @@ type ExecuteRequest struct {
 	Headers              http.Header
 	ConfiguredHeaders    []string
 	OriginalRequest      []byte
+	ContinuityKey        string
 	BaseURL              string
 	ProxyURL             string
 	ProxyFromEnvironment bool
@@ -415,6 +416,7 @@ func executeRequestToBridge(value ExecuteRequest) cpaembedded.ExecuteRequest {
 		Headers:              value.Headers.Clone(),
 		ConfiguredHeaders:    append([]string(nil), value.ConfiguredHeaders...),
 		OriginalRequest:      append([]byte(nil), value.OriginalRequest...),
+		ContinuityKey:        value.ContinuityKey,
 		BaseURL:              value.BaseURL,
 		ProxyURL:             value.ProxyURL,
 		ProxyFromEnvironment: value.ProxyFromEnvironment,

internal/telemetry/requestlog.go

@@ -149,6 +149,7 @@ type RequestEvent struct {
 	FirstResponseMs       *int64
 	DurationMs            int64
 	AffinityHit           bool
+	AffinityKind          string
 	Reasoning             reasoning.Config
 	Attempts              []Attempt
 	Usage                 UsageObservation
@@ -161,3 +162,10 @@ type RequestLogSink interface {
 type NoopRequestLogSink struct{}
 
 func (NoopRequestLogSink) Emit(RequestEvent) {}
+
+// 亲和类型仅描述实际选中账号的依据，不代表上游缓存命中。
+const (
+	AffinityPromptPrefix       = "prompt_prefix"
+	AffinityPromptCacheKey     = "prompt_cache_key"
+	AffinityResponseContinuity = "response_continuity"
+)

internal/telemetry/requestlog_test.go

@@ -34,6 +34,7 @@ func TestRequestTelemetryContractUsesExactFieldAllowlist(t *testing.T) {
 			"FirstResponseMs",
 			"DurationMs",
 			"AffinityHit",
+			"AffinityKind",
 			"Reasoning",
 			"Attempts",
 			"Usage",

third_party/cpaembedded/embedded/embedded.go

@@ -404,6 +404,7 @@ func (e *CodexHTTPExecutor) ExecuteCanonical(ctx context.Context, credentialID s
 	executionCtx := e.executionContext(ctx, auth, observation, request.ProxyFromEnvironment, &request)
 	response, err := e.inner.Execute(executionCtx, authWithoutProxyURL(auth), cliproxyexecutor.Request{
 		Model: request.Model, Payload: append([]byte(nil), request.Payload...), Format: format,
+		Metadata: codexSessionMetadata(request, format),
 	}, codexExecutionOptions(request, format, false))
 	if err != nil {
 		return ExecuteResponse{
@@ -498,6 +499,7 @@ func (e *CodexHTTPExecutor) ExecuteStreamCanonical(ctx context.Context, credenti
 	executionCtx := e.executionContext(ctx, auth, observation, request.ProxyFromEnvironment, &request)
 	response, err := e.inner.ExecuteStream(executionCtx, authWithoutProxyURL(auth), cliproxyexecutor.Request{
 		Model: request.Model, Payload: append([]byte(nil), request.Payload...), Format: format,
+		Metadata: codexSessionMetadata(request, format),
 	}, codexExecutionOptions(request, format, true))
 	if err != nil {
 		return &ExecuteStreamResponse{
@@ -526,6 +528,25 @@ func (e *CodexHTTPExecutor) ExecuteStreamCanonical(ctx context.Context, credenti
 	}, nil
 }
 
+// 给 Codex 提供稳定缓存分组兜底；CPA 优先使用客户端缓存键或 Claude Code 身份。
+func codexSessionMetadata(request ExecuteRequest, format sdktranslator.Format) map[string]any {
+	switch format {
+	case sdktranslator.FormatOpenAI, sdktranslator.FormatOpenAIResponse, sdktranslator.FormatClaude, sdktranslator.FormatGemini:
+	default:
+		return nil
+	}
+	// 保留客户端已有会话头，避免兜底缓存键覆盖其上游会话。
+	if strings.TrimSpace(request.Headers.Get("Session-Id")) != "" {
+		return nil
+	}
+	scope := strings.TrimSpace(request.ContinuityKey)
+	if scope == "" {
+		return nil
+	}
+	// 这是提示词派生的缓存分组，不是严格执行会话；避免启用额外的 reasoning replay。
+	return map[string]any{cliproxyexecutor.DerivedSessionIDMetadataKey: scope}
+}
+
 func codexExecutionOptions(
 	request ExecuteRequest,
 	format sdktranslator.Format,

web/src/app/resources/request-logs.ts

@@ -185,6 +185,7 @@ export interface RequestLogItemDto {
   error_code: string
   error_summary: string
   affinity_hit: boolean
+  affinity_kind: string
   group_id: number | null
   channel_id: string | null
   credential_id: number | null
@@ -289,6 +290,7 @@ const itemFields = [
   'error_code',
   'error_summary',
   'affinity_hit',
+  'affinity_kind',
   'group_id',
   'channel_id',
   'credential_id',
@@ -633,6 +635,7 @@ function projectItemRecord(record: Record<string, unknown>): RequestLogItemDto {
     error_code: projectString(record.error_code, { allowEmpty: true }),
     error_summary: projectString(record.error_summary, { allowEmpty: true }),
     affinity_hit: projectBoolean(record.affinity_hit),
+    affinity_kind: projectString(record.affinity_kind, { allowEmpty: true }),
     group_id: record.group_id === null ? null : projectSafeInteger(record.group_id, { minimum: 1 }),
     channel_id: record.channel_id === null ? null : projectChannelID(record.channel_id),
     credential_id:

web/src/features/monitor/LogsTab.vue

@@ -75,6 +75,20 @@ const session = useAuthSession()
 const route = useRoute()
 const router = useRouter()
 const { locale, t } = useI18n()
+
+function affinityTooltip(log: RequestLogItemDto): string {
+  switch (log.affinity_kind) {
+    case 'prompt_prefix':
+      return t('monitor.logs.drawer.affinityPromptPrefix')
+    case 'prompt_cache_key':
+      return t('monitor.logs.drawer.affinityPromptCacheKey')
+    case 'response_continuity':
+      return t('monitor.logs.drawer.affinityResponseContinuity')
+    default:
+      return t('monitor.logs.drawer.affinity')
+  }
+}
+
 const logPageSizes = [20, 50, 100] as const
 const isAccessKey = computed(() => session.state.principalType === 'access_key')
 const appliedFilters = computed(() => {
@@ -869,11 +883,11 @@ function costLabel(log: RequestLogItemDto): string {
                   {{ responseLabel(log) }}
                 </StatusBadge>
               </OverflowTooltip>
-              <AppTooltip v-if="log.affinity_hit" :content="t('monitor.logs.drawer.affinity')">
+              <AppTooltip v-if="log.affinity_hit" :content="affinityTooltip(log)">
                 <span
                   class="logs-list__hint logs-list__affinity"
                   tabindex="0"
-                  :aria-label="t('monitor.logs.drawer.affinity')"
+                  :aria-label="affinityTooltip(log)"
                 >
                   <Magnet :size="13" aria-hidden="true" />
                 </span>

web/src/i18n/locales/en-US/monitor.ts

@@ -867,6 +867,9 @@ export default {
         routeIdentity: 'Route identity',
         upstreamProtocol: 'Upstream protocol',
         affinity: 'Affinity hit',
+        affinityPromptPrefix: 'Prompt prefix affinity hit',
+        affinityPromptCacheKey: 'Prompt cache key affinity hit',
+        affinityResponseContinuity: 'Response continuation hit',
         errorCode: 'Error code',
         errorSummary: 'Error message',
         expandErrorMessage: 'Show full error message',

web/src/i18n/locales/zh-CN/monitor.ts

@@ -847,6 +847,9 @@ export default {
         routeIdentity: '路由标识',
         upstreamProtocol: '上游协议',
         affinity: '亲和命中',
+        affinityPromptPrefix: '提示词亲和命中',
+        affinityPromptCacheKey: '缓存标识亲和命中',
+        affinityResponseContinuity: '响应续接命中',
         errorCode: '错误码',
         errorSummary: '错误消息',
         expandErrorMessage: '展开完整错误消息',