From 9d38793dd0ee446607e8750f061c5a1da8d868c0 Mon Sep 17 00:00:00 2001 From: tbphp Date: Sun, 13 Sep 2026 01:25:45 +0800 Subject: [PATCH] =?UTF-8?q?fix(affinity):=20=E7=BB=9F=E4=B8=80=E7=BC=93?= =?UTF-8?q?=E5=AD=98=E6=A0=87=E8=AF=86=E4=BA=B2=E5=92=8C=E5=B9=B6=E8=A1=A5?= =?UTF-8?q?=E9=BD=90=20Codex=20=E7=BC=93=E5=AD=98=E5=88=86=E7=BB=84=20(#63?= =?UTF-8?q?5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/affinity/key.go | 11 +- internal/affinity/key_test.go | 17 +++ internal/control/request_logs.go | 3 + internal/control/request_logs_test.go | 6 +- internal/dialect/dialect.go | 1 + internal/dialect/openai.go | 1 + internal/dialect/openai_responses.go | 3 + internal/dialect/prompt_cache.go | 49 +++++++ internal/dialect/prompt_cache_test.go | 49 +++++++ internal/execution/cpa/codex_cache_test.go | 131 ++++++++++++++++++ internal/execution/cpa/codex_provider.go | 10 +- internal/gateway/affinity.go | 10 +- internal/gateway/affinity_test.go | 62 +++++++++ internal/gateway/handler.go | 14 +- internal/gateway/request_log.go | 5 +- .../gateway/responses_continuation_test.go | 30 +++- internal/gateway/websocket_test.go | 50 +++++++ internal/gateway/websocket_turn.go | 13 +- internal/requestlog/mapper.go | 1 + internal/requestlog/phase4_v3_test.go | 4 + internal/requestlog/query.go | 1 + internal/requestlog/query_test.go | 20 +++ internal/requestlog/types.go | 1 + .../storage/affinity_kind_migration_test.go | 77 ++++++++++ internal/storage/database_integration_test.go | 6 +- internal/storage/db_test.go | 1 + internal/storage/migration.go | 1 + internal/storage/migration_test.go | 1 + .../migrations/0006_error_decision_test.go | 2 +- .../storage/migrations/0014_affinity_kind.go | 58 ++++++++ .../storage/model_cooldown_migration_test.go | 2 +- internal/storage/models/request.go | 1 + .../subscription/providers/codex/codex.go | 2 + internal/telemetry/requestlog.go | 8 ++ internal/telemetry/requestlog_test.go | 1 + third_party/cpaembedded/embedded/embedded.go | 21 +++ web/src/app/resources/request-logs.ts | 3 + web/src/features/monitor/LogsTab.vue | 18 ++- web/src/i18n/locales/en-US/monitor.ts | 3 + web/src/i18n/locales/zh-CN/monitor.ts | 3 + 40 files changed, 678 insertions(+), 22 deletions(-) create mode 100644 internal/dialect/prompt_cache.go create mode 100644 internal/dialect/prompt_cache_test.go create mode 100644 internal/execution/cpa/codex_cache_test.go create mode 100644 internal/storage/affinity_kind_migration_test.go create mode 100644 internal/storage/migrations/0014_affinity_kind.go diff --git a/internal/affinity/key.go b/internal/affinity/key.go index aebdc0a3..2295c510 100644 --- a/internal/affinity/key.go +++ b/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[:]) diff --git a/internal/affinity/key_test.go b/internal/affinity/key_test.go index 88406a8a..9ac33a41 100644 --- a/internal/affinity/key_test.go +++ b/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") + } + } +} diff --git a/internal/control/request_logs.go b/internal/control/request_logs.go index 37c76168..6c54cc0e 100644 --- a/internal/control/request_logs.go +++ b/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, diff --git a/internal/control/request_logs_test.go b/internal/control/request_logs_test.go index 07821fef..11d8dddf 100644 --- a/internal/control/request_logs_test.go +++ b/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", diff --git a/internal/dialect/dialect.go b/internal/dialect/dialect.go index d99c00a4..0f1a1c0a 100644 --- a/internal/dialect/dialect.go +++ b/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 diff --git a/internal/dialect/openai.go b/internal/dialect/openai.go index ba4a26a2..d23f7cf8 100644 --- a/internal/dialect/openai.go +++ b/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 { diff --git a/internal/dialect/openai_responses.go b/internal/dialect/openai_responses.go index a3cf2df8..23e7012a 100644 --- a/internal/dialect/openai_responses.go +++ b/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) diff --git a/internal/dialect/prompt_cache.go b/internal/dialect/prompt_cache.go new file mode 100644 index 00000000..5ca18f1a --- /dev/null +++ b/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 +} diff --git a/internal/dialect/prompt_cache_test.go b/internal/dialect/prompt_cache_test.go new file mode 100644 index 00000000..fda83f68 --- /dev/null +++ b/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) + } + } +} diff --git a/internal/execution/cpa/codex_cache_test.go b/internal/execution/cpa/codex_cache_test.go new file mode 100644 index 00000000..1c28595b --- /dev/null +++ b/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") + } + }) + } + } +} diff --git a/internal/execution/cpa/codex_provider.go b/internal/execution/cpa/codex_provider.go index 01d6ad3f..24993774 100644 --- a/internal/execution/cpa/codex_provider.go +++ b/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, }) diff --git a/internal/gateway/affinity.go b/internal/gateway/affinity.go index 3f7c4795..04330751 100644 --- a/internal/gateway/affinity.go +++ b/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, diff --git a/internal/gateway/affinity_test.go b/internal/gateway/affinity_test.go index e830f4de..3a3da126 100644 --- a/internal/gateway/affinity_test.go +++ b/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") + } +} diff --git a/internal/gateway/handler.go b/internal/gateway/handler.go index 43f6da36..48903e7e 100644 --- a/internal/gateway/handler.go +++ b/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" diff --git a/internal/gateway/request_log.go b/internal/gateway/request_log.go index 046d7a75..9dbb9d8a 100644 --- a/internal/gateway/request_log.go +++ b/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 } } diff --git a/internal/gateway/responses_continuation_test.go b/internal/gateway/responses_continuation_test.go index 07f94a0f..6e964a91 100644 --- a/internal/gateway/responses_continuation_test.go +++ b/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) + } + } +} diff --git a/internal/gateway/websocket_test.go b/internal/gateway/websocket_test.go index f1d12fef..be4468d9 100644 --- a/internal/gateway/websocket_test.go +++ b/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") + } +} diff --git a/internal/gateway/websocket_turn.go b/internal/gateway/websocket_turn.go index 26ca241b..0e114af6 100644 --- a/internal/gateway/websocket_turn.go +++ b/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) } } diff --git a/internal/requestlog/mapper.go b/internal/requestlog/mapper.go index bc2de431..a80ef4e8 100644 --- a/internal/requestlog/mapper.go +++ b/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, diff --git a/internal/requestlog/phase4_v3_test.go b/internal/requestlog/phase4_v3_test.go index 8434fd5f..9c3089f1 100644 --- a/internal/requestlog/phase4_v3_test.go +++ b/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) diff --git a/internal/requestlog/query.go b/internal/requestlog/query.go index 4d7efeb5..2b6e0b2a 100644 --- a/internal/requestlog/query.go +++ b/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, diff --git a/internal/requestlog/query_test.go b/internal/requestlog/query_test.go index 2c9eb812..af412446 100644 --- a/internal/requestlog/query_test.go +++ b/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") + } +} diff --git a/internal/requestlog/types.go b/internal/requestlog/types.go index 214a6e20..2ba145c1 100644 --- a/internal/requestlog/types.go +++ b/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 diff --git a/internal/storage/affinity_kind_migration_test.go b/internal/storage/affinity_kind_migration_test.go new file mode 100644 index 00000000..339b94b2 --- /dev/null +++ b/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) + } + }) + } +} diff --git a/internal/storage/database_integration_test.go b/internal/storage/database_integration_test.go index a95a8df2..1b56af53 100644 --- a/internal/storage/database_integration_test.go +++ b/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") diff --git a/internal/storage/db_test.go b/internal/storage/db_test.go index eb99e1d4..1f749007 100644 --- a/internal/storage/db_test.go +++ b/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) diff --git a/internal/storage/migration.go b/internal/storage/migration.go index 679ff376..eabcefe6 100644 --- a/internal/storage/migration.go +++ b/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 { diff --git a/internal/storage/migration_test.go b/internal/storage/migration_test.go index e9b095b4..14557a9a 100644 --- a/internal/storage/migration_test.go +++ b/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)) diff --git a/internal/storage/migrations/0006_error_decision_test.go b/internal/storage/migrations/0006_error_decision_test.go index e2e85b76..3b109a8b 100644 --- a/internal/storage/migrations/0006_error_decision_test.go +++ b/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{ diff --git a/internal/storage/migrations/0014_affinity_kind.go b/internal/storage/migrations/0014_affinity_kind.go new file mode 100644 index 00000000..d950ccc4 --- /dev/null +++ b/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") +} diff --git a/internal/storage/model_cooldown_migration_test.go b/internal/storage/model_cooldown_migration_test.go index 4e2a7d86..108c3489 100644 --- a/internal/storage/model_cooldown_migration_test.go +++ b/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, diff --git a/internal/storage/models/request.go b/internal/storage/models/request.go index f518df6a..e3d0608b 100644 --- a/internal/storage/models/request.go +++ b/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"` diff --git a/internal/subscription/providers/codex/codex.go b/internal/subscription/providers/codex/codex.go index 226e98c0..d92e7440 100644 --- a/internal/subscription/providers/codex/codex.go +++ b/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, diff --git a/internal/telemetry/requestlog.go b/internal/telemetry/requestlog.go index 6a12c47c..8e3cef1f 100644 --- a/internal/telemetry/requestlog.go +++ b/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" +) diff --git a/internal/telemetry/requestlog_test.go b/internal/telemetry/requestlog_test.go index 1e73adfb..5dfa1946 100644 --- a/internal/telemetry/requestlog_test.go +++ b/internal/telemetry/requestlog_test.go @@ -34,6 +34,7 @@ func TestRequestTelemetryContractUsesExactFieldAllowlist(t *testing.T) { "FirstResponseMs", "DurationMs", "AffinityHit", + "AffinityKind", "Reasoning", "Attempts", "Usage", diff --git a/third_party/cpaembedded/embedded/embedded.go b/third_party/cpaembedded/embedded/embedded.go index b2812cd2..9370d7e5 100644 --- a/third_party/cpaembedded/embedded/embedded.go +++ b/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, diff --git a/web/src/app/resources/request-logs.ts b/web/src/app/resources/request-logs.ts index 9d366bbf..ab529a8e 100644 --- a/web/src/app/resources/request-logs.ts +++ b/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): 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: diff --git a/web/src/features/monitor/LogsTab.vue b/web/src/features/monitor/LogsTab.vue index 84c335c3..e2eeed10 100644 --- a/web/src/features/monitor/LogsTab.vue +++ b/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) }} - + diff --git a/web/src/i18n/locales/en-US/monitor.ts b/web/src/i18n/locales/en-US/monitor.ts index 4083ed18..90638ba4 100644 --- a/web/src/i18n/locales/en-US/monitor.ts +++ b/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', diff --git a/web/src/i18n/locales/zh-CN/monitor.ts b/web/src/i18n/locales/zh-CN/monitor.ts index 483cc2cb..71a06b69 100644 --- a/web/src/i18n/locales/zh-CN/monitor.ts +++ b/web/src/i18n/locales/zh-CN/monitor.ts @@ -847,6 +847,9 @@ export default { routeIdentity: '路由标识', upstreamProtocol: '上游协议', affinity: '亲和命中', + affinityPromptPrefix: '提示词亲和命中', + affinityPromptCacheKey: '缓存标识亲和命中', + affinityResponseContinuity: '响应续接命中', errorCode: '错误码', errorSummary: '错误消息', expandErrorMessage: '展开完整错误消息',