From 2aa5002a45efc245a94c761f62c96513abebe02a Mon Sep 17 00:00:00 2001 From: tbphp Date: Sun, 6 Sep 2026 09:29:35 +0800 Subject: [PATCH] =?UTF-8?q?fix(subscription):=20=E6=8C=89=E6=9D=A5?= =?UTF-8?q?=E6=BA=90=E5=92=8C=E5=91=A8=E6=9C=9F=E5=8C=B9=E9=85=8D=20Codex?= =?UTF-8?q?=20=E8=A2=AB=E5=8A=A8=E9=A2=9D=E5=BA=A6=E7=AA=97=E5=8F=A3=20(#5?= =?UTF-8?q?74)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(subscription): 拒绝跨周期的被动额度覆盖 被动额度合并前校验已有窗口与样本的周期；同 ID 但周期冲突时丢弃整个窗口补丁，避免 7d 被改成 5h，同时污染用量、重置时间及健康摘要。 保留兼容窗口与缺少周期字段的增量更新行为；补充合并与持久化回归测试，不改变限流、额度池或路由调度。 * fix(subscription): 按来源和周期匹配 Codex 被动额度 主动同步保留 metered_feature，响应头按命名空间生成相同来源标识；不依赖模型名称或 Limit-Name。 按同一账号内的来源与实际周期唯一匹配已有窗口，允许 primary/secondary 槽位变化，但保留卡片 ID 与展示元数据。缺少来源/周期或存在歧义时不覆盖；其他渠道继续使用原有 ID 匹配。 补充解析、跨槽位合并、来源隔离、控制面序列化及持久化健康投影测试。无 429、路由或调度改动；旧快照需主动同步一次补齐来源。 * docs(subscription): 修正被动额度快照合并的注释说明 明确无变化时保持原始字节，仅在窗口数据变化时重新编码；不修改任何运行逻辑。 --- internal/control/credential_observations.go | 3 +- .../control/credential_quota_source_test.go | 40 +++++ internal/subscription/passive_quota_flush.go | 71 +++++--- .../subscription/passive_quota_period_test.go | 166 ++++++++++++++++++ .../passive_quota_source_flush_test.go | 66 +++++++ .../subscription/passive_quota_source_test.go | 117 ++++++++++++ .../providers/codex/observation.go | 104 +++++------ .../providers/codex/observation_test.go | 17 +- .../providers/codex/quota_source_test.go | 73 ++++++++ .../providers/observation/snapshot.go | 2 + 10 files changed, 569 insertions(+), 90 deletions(-) create mode 100644 internal/control/credential_quota_source_test.go create mode 100644 internal/subscription/passive_quota_period_test.go create mode 100644 internal/subscription/passive_quota_source_flush_test.go create mode 100644 internal/subscription/passive_quota_source_test.go create mode 100644 internal/subscription/providers/codex/quota_source_test.go diff --git a/internal/control/credential_observations.go b/internal/control/credential_observations.go index 90abdc11..8938c69b 100644 --- a/internal/control/credential_observations.go +++ b/internal/control/credential_observations.go @@ -44,6 +44,7 @@ type ObservationAccountSummary struct { } type ObservationQuotaWindow struct { + SourceID string `json:"source_id,omitempty"` ID string `json:"id"` Label string `json:"label"` LabelKey string `json:"label_key,omitempty"` @@ -674,7 +675,7 @@ func providerQuotaWindows(windows []ObservationQuotaWindow) []providerobservatio for _, window := range windows { result = append(result, providerobservation.QuotaWindow{ ID: window.ID, Label: window.Label, LabelKey: window.LabelKey, - Scope: window.Scope, Unit: window.Unit, + Scope: window.Scope, Unit: window.Unit, SourceID: window.SourceID, Used: window.Used, Limit: window.Limit, Remaining: window.Remaining, Utilization: window.Utilization, ResetAtMS: window.ResetAtMS, WindowSeconds: window.WindowSeconds, ModelIDs: window.ModelIDs, State: window.State, IsPrimary: window.IsPrimary, diff --git a/internal/control/credential_quota_source_test.go b/internal/control/credential_quota_source_test.go new file mode 100644 index 00000000..4ac52fca --- /dev/null +++ b/internal/control/credential_quota_source_test.go @@ -0,0 +1,40 @@ +package control + +import ( + "encoding/json" + "reflect" + "testing" + + "gpt-load/internal/subscription/providers/codex" + providerobservation "gpt-load/internal/subscription/providers/observation" +) + +func TestCredentialQuotaSourceSurvivesSnapshotRoundTrip(t *testing.T) { + raw, err := codex.NormalizeQuota([]byte(`{ + "rate_limit":{"primary_window":{"used_percent":80,"limit_window_seconds":604800}}, + "additional_rate_limits":[{"metered_feature":"codex_bengalfox","limit_name":"Spark", + "rate_limit":{"primary_window":{"used_percent":10,"limit_window_seconds":18000}}}] + }`), nil) + if err != nil { + t.Fatal(err) + } + var snapshot CredentialObservationSnapshot + if err := json.Unmarshal(raw, &snapshot); err != nil { + t.Fatal(err) + } + encoded, err := json.Marshal(snapshot) + if err != nil { + t.Fatal(err) + } + var stored providerobservation.Snapshot + if err := json.Unmarshal(encoded, &stored); err != nil { + t.Fatal(err) + } + if len(stored.QuotaWindows) != 2 || stored.QuotaWindows[0].SourceID != "codex" || + stored.QuotaWindows[1].SourceID != "codex_bengalfox" { + t.Fatalf("control-plane serialization lost quota identity: %s", encoded) + } + if projected := providerQuotaWindows(snapshot.QuotaWindows); !reflect.DeepEqual(projected, stored.QuotaWindows) { + t.Fatalf("quota projection lost fields: %#v", projected) + } +} diff --git a/internal/subscription/passive_quota_flush.go b/internal/subscription/passive_quota_flush.go index 0368bccf..bc4abea3 100644 --- a/internal/subscription/passive_quota_flush.go +++ b/internal/subscription/passive_quota_flush.go @@ -25,8 +25,8 @@ const passiveQuotaFlushBatchSize = 20 // only ever touches snapshot_json, observed_at_ms, and updated_at_ms. State, // plan/account summaries, reset credits, and the most recent active // attempt/error metadata are always left untouched. A credential with no -// existing observation row, a stale identity generation, or no window ID -// the row already tracks is discarded without writing anything, since manual +// existing observation row, a stale identity generation, or no matching window +// in the row is discarded without writing anything, since manual // or automatic active observation remains the sole owner of window creation. func (manager *CredentialManager) FlushPassiveQuotaObservations(ctx context.Context) (bool, error) { if manager == nil || manager.passiveQuota == nil || manager.db == nil { @@ -68,7 +68,7 @@ func (manager *CredentialManager) flushOnePassiveQuotaObservation( // was persisted while this sample sat pending. Writing it now would // rewind observed_at_ms and overwrite newer quota values with older // ones. The tie is included on purpose: a passive header captured - // just before an active refresh that finished within the same + // just before an active refresh that completed within the same // millisecond truncates to the same value, and the active result is // the authoritative one. The CAS below only catches writes that land // after this read. @@ -83,9 +83,8 @@ func (manager *CredentialManager) flushOnePassiveQuotaObservation( return nil } if !merge.Matched { - // The response named no window this snapshot tracks, so it is no - // observation of this credential's quota at all. Creating windows - // is the active observation's job. + // 未能唯一匹配已有窗口的样本不能推进同步时间。 + // 窗口创建和周期变更仍由主动观测负责。 manager.passiveQuota.ack(observation.CredentialID, observation.Version) return nil } @@ -139,12 +138,9 @@ type passiveQuotaMerge struct { Changed bool } -// mergePassiveQuotaSnapshot overlays patches onto the quota_windows array -// inside a stored snapshot, carrying every other field through unchanged by -// value. The snapshot is decoded and re-encoded, so key order and formatting -// are not preserved byte-for-byte. Only window IDs the snapshot already -// tracks are updated; an unmatched patch ID is silently ignored so a passive -// signal never creates a window. +// mergePassiveQuotaSnapshot 仅更新已有且唯一匹配的窗口，不创建或改换窗口。 +// 无变化时直接返回原始 raw，保持字节不变；仅在窗口数据变化时重新编码， +// 此时保留其他字段的值，但不保证原始键顺序或格式。 func mergePassiveQuotaSnapshot( raw []byte, patches []providerobservation.QuotaWindow, @@ -162,19 +158,32 @@ func mergePassiveQuotaSnapshot( return passiveQuotaMerge{}, fmt.Errorf("decode credential observation quota windows: %w", err) } merged := append([]providerobservation.QuotaWindow(nil), existing...) - index := make(map[string]int, len(merged)) - for position, window := range merged { - index[window.ID] = position + positions := make([]int, len(patches)) + matches := make(map[int]int, len(patches)) + for index, patch := range patches { + position := matchPassiveQuotaWindow(existing, patch) + positions[index] = position + if position >= 0 { + matches[position]++ + } } outcome := passiveQuotaMerge{Encoded: raw, Windows: existing} - for _, patch := range patches { - position, exists := index[patch.ID] - if !exists { + for index, patch := range patches { + position := positions[index] + // 多个响应窗口指向同一目标也是歧义，不能取最后一个值覆盖。 + if position < 0 || matches[position] != 1 { + continue + } + previous := merged[position] + // primary/secondary 只是上游槽位；周期冲突时连用量、重置时间和状态也不能合并。 + if previous.WindowSeconds != nil && patch.WindowSeconds != nil && + *previous.WindowSeconds != *patch.WindowSeconds { continue } outcome.Matched = true - next := providerobservation.MergeQuotaWindow(merged[position], patch) - if !reflect.DeepEqual(next, merged[position]) { + patch.ID = previous.ID // 响应槽位可以变化，卡片窗口身份不变。 + next := providerobservation.MergeQuotaWindow(previous, patch) + if !reflect.DeepEqual(next, previous) { outcome.Changed = true } merged[position] = next @@ -194,3 +203,25 @@ func mergePassiveQuotaSnapshot( outcome.Encoded, outcome.Windows = encoded, merged return outcome, nil } + +// matchPassiveQuotaWindow 按来源和实际周期对齐主动/被动数据；槽位不参与推断。 +// 其他未提供 SourceID 的渠道保留 ID 匹配，但不能覆盖带来源标识的窗口。 +func matchPassiveQuotaWindow(windows []providerobservation.QuotaWindow, patch providerobservation.QuotaWindow) int { + matched := -1 + for index, window := range windows { + if patch.SourceID != "" { + if window.SourceID != patch.SourceID || patch.WindowSeconds == nil || + *patch.WindowSeconds <= 0 || window.WindowSeconds == nil || + *window.WindowSeconds != *patch.WindowSeconds { + continue + } + } else if window.SourceID != "" || window.ID != patch.ID { + continue + } + if matched >= 0 { + return -1 + } + matched = index + } + return matched +} diff --git a/internal/subscription/passive_quota_period_test.go b/internal/subscription/passive_quota_period_test.go new file mode 100644 index 00000000..cc15003a --- /dev/null +++ b/internal/subscription/passive_quota_period_test.go @@ -0,0 +1,166 @@ +package subscription + +import ( + "bytes" + "encoding/json" + "reflect" + "testing" + "time" + + "gpt-load/internal/storage/models" + "gpt-load/internal/subscription/providers/codex" + providerobservation "gpt-load/internal/subscription/providers/observation" +) + +func TestMergePassiveQuotaSnapshotChecksWindowPeriod(t *testing.T) { + weekly, fiveHour := int64(7*24*60*60), int64(5*60*60) + for _, test := range []struct { + name string + id string + storedPeriod *int64 + patchPeriod *int64 + wantMatched bool + }{ + {"weekly to five hour", "primary", &weekly, &fiveHour, false}, + {"five hour to weekly", "primary", &fiveHour, &weekly, false}, + {"secondary conflict", "secondary", &weekly, &fiveHour, false}, + {"additional window conflict", "spark-primary", &weekly, &fiveHour, false}, + {"same period", "primary", &weekly, &weekly, true}, + {"partial header omits period", "primary", &weekly, nil, true}, + {"fills unknown period", "primary", nil, &weekly, true}, + {"both periods unknown", "primary", nil, nil, true}, + } { + t.Run(test.name, func(t *testing.T) { + used, limit, remaining, utilization := 100.0, 100.0, 0.0, 1.0 + resetAt := int64(1800000000000) + previous := providerobservation.QuotaWindow{ + ID: test.id, Label: "Active window", Scope: "account", Unit: "percent", + Used: &used, Limit: &limit, Remaining: &remaining, Utilization: &utilization, + ResetAtMS: &resetAt, WindowSeconds: test.storedPeriod, + State: "exhausted", IsPrimary: true, + } + raw, err := json.Marshal(providerobservation.Snapshot{ + Plan: providerobservation.PlanSummary{Name: "Pro"}, + QuotaWindows: []providerobservation.QuotaWindow{previous}, + }) + if err != nil { + t.Fatal(err) + } + patchUsed, patchRemaining, patchUtilization := 10.0, 90.0, 0.1 + patchResetAt := resetAt + 60000 + patch := providerobservation.QuotaWindow{ + ID: test.id, Used: &patchUsed, Limit: &limit, + Remaining: &patchRemaining, Utilization: &patchUtilization, + ResetAtMS: &patchResetAt, WindowSeconds: test.patchPeriod, State: "available", + } + + merged, err := mergePassiveQuotaSnapshot(raw, []providerobservation.QuotaWindow{patch}) + if err != nil { + t.Fatal(err) + } + if merged.Matched != test.wantMatched || merged.Changed != test.wantMatched { + t.Fatalf("Matched=%v Changed=%v, want both %v", merged.Matched, merged.Changed, test.wantMatched) + } + want := previous + if test.wantMatched { + want.Used, want.Remaining, want.Utilization = &patchUsed, &patchRemaining, &patchUtilization + want.ResetAtMS, want.State = &patchResetAt, "available" + if test.patchPeriod != nil { + want.WindowSeconds = test.patchPeriod + } + } else if !bytes.Equal(merged.Encoded, raw) { + t.Fatalf("conflicting sample changed the snapshot: %s", merged.Encoded) + } + if !reflect.DeepEqual(merged.Windows, []providerobservation.QuotaWindow{want}) { + t.Fatalf("merged windows = %#v, want %#v", merged.Windows, want) + } + }) + } +} + +const passiveQuotaWeeklyAndSparkSnapshot = `{"plan_summary":{"name":"Pro"},"quota_windows":[{"id":"primary","label":"7d","label_key":"weekly","scope":"account","unit":"percent","used":100,"limit":100,"remaining":0,"utilization":1,"reset_at_ms":1800000000000,"window_seconds":604800,"state":"exhausted","is_primary":true},{"id":"gpt-5-3-codex-spark-primary","label":"GPT 5.3 Codex Spark · 5h","scope":"GPT 5.3 Codex Spark","unit":"percent","used":20,"limit":100,"remaining":80,"utilization":0.2,"reset_at_ms":1799900000000,"window_seconds":18000,"state":"available"}],"reset_credits_available":3}` + +func TestMergePassiveQuotaSnapshotKeepsCompatibleWindowAlongsideConflict(t *testing.T) { + var previous providerobservation.Snapshot + if err := json.Unmarshal([]byte(passiveQuotaWeeklyAndSparkSnapshot), &previous); err != nil { + t.Fatal(err) + } + fiveHour := int64(5 * 60 * 60) + used, limit, remaining, utilization := 30.0, 100.0, 70.0, 0.3 + patch := providerobservation.QuotaWindow{ + ID: "primary", WindowSeconds: &fiveHour, State: "available", + Used: &used, Limit: &limit, Remaining: &remaining, Utilization: &utilization, + } + sparkPatch := patch + sparkPatch.ID = "gpt-5-3-codex-spark-primary" + + merged, err := mergePassiveQuotaSnapshot([]byte(passiveQuotaWeeklyAndSparkSnapshot), + []providerobservation.QuotaWindow{patch, sparkPatch}) + if err != nil { + t.Fatal(err) + } + if !merged.Matched || !merged.Changed || len(merged.Windows) != 2 { + t.Fatalf("merge = %#v, want only the compatible window updated", merged) + } + want := append([]providerobservation.QuotaWindow(nil), previous.QuotaWindows...) + want[1].Used, want[1].Remaining, want[1].Utilization = &used, &remaining, &utilization + if !reflect.DeepEqual(merged.Windows, want) { + t.Fatalf("weekly or Spark window corrupted: %s", merged.Encoded) + } + var stored providerobservation.Snapshot + if err := json.Unmarshal(merged.Encoded, &stored); err != nil { + t.Fatal(err) + } + previous.QuotaWindows = want + if !reflect.DeepEqual(stored, previous) { + t.Fatalf("encoded snapshot differs from expected windows and metadata: %s", merged.Encoded) + } +} + +func TestFlushPassiveQuotaObservationsDiscardsConflictingPeriodWithoutAdvancingFreshness(t *testing.T) { + manager, db, registry, _, row := newCredentialManagerFixture(t, + credentialJSON("access", "refresh", time.Now().Add(time.Hour))) + existing := newFlushableCredentialObservation(t, manager, row.ID, + models.CredentialObservationFresh, passiveQuotaWeeklyAndSparkSnapshot) + var snapshot providerobservation.Snapshot + if err := json.Unmarshal(existing.SnapshotJSON, &snapshot); err != nil { + t.Fatal(err) + } + if !registry.ApplyQuotaWindows(row.ID, snapshot.QuotaWindows) { + t.Fatal("failed to initialize quota health projection") + } + before := registry.Snapshot() + ref, ok := registry.CredentialRef(row.ID) + if !ok { + t.Fatal("credential ref is unavailable") + } + // 普通 7d 与 Spark 5h 都可能占用 primary；冲突样本不能刷新任何账号级状态。 + observedAt := time.UnixMilli(2000) + windows := codex.NormalizePassiveQuotaWindows(map[string]string{ + "X-Codex-Primary-Used-Percent": "10", + "X-Codex-Primary-Window-Minutes": "300", + "X-Codex-Primary-Reset-At": "1799900000", + }, observedAt) + if len(windows) != 1 || windows[0].ID != "primary" { + t.Fatalf("unexpected passive windows: %#v", windows) + } + manager.RecordPassiveQuotaObservation(row.ID, ref.IdentityGeneration, observedAt.UnixMilli(), windows) + + pending, err := manager.FlushPassiveQuotaObservations(t.Context()) + if err != nil { + t.Fatal(err) + } + if pending || len(manager.DirtyPassiveQuotaObservations(1)) != 0 { + t.Fatal("conflicting sample must be acknowledged, not retried forever") + } + var stored models.CredentialObservation + if err := db.Take(&stored, "credential_id = ?", row.ID).Error; err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(stored, existing) { + t.Fatalf("conflicting sample changed observation data or freshness: %#v", stored) + } + if after := registry.Snapshot(); !reflect.DeepEqual(after, before) { + t.Fatalf("conflicting sample changed quota health projection: %#v", after) + } +} diff --git a/internal/subscription/passive_quota_source_flush_test.go b/internal/subscription/passive_quota_source_flush_test.go new file mode 100644 index 00000000..6ca7c6b0 --- /dev/null +++ b/internal/subscription/passive_quota_source_flush_test.go @@ -0,0 +1,66 @@ +package subscription + +import ( + "encoding/json" + "testing" + "time" + + "gpt-load/internal/storage/models" + "gpt-load/internal/subscription/providers/codex" + providerobservation "gpt-load/internal/subscription/providers/observation" +) + +func TestFlushPassiveQuotaSourcesPreservesCardWindowsAndHealth(t *testing.T) { + manager, db, registry, _, row := newCredentialManagerFixture(t, + credentialJSON("access", "refresh", time.Now().Add(time.Hour))) + raw, err := codex.NormalizeQuota([]byte(codexSourceQuotaPayload), nil) + if err != nil { + t.Fatal(err) + } + initial := newFlushableCredentialObservation(t, manager, row.ID, models.CredentialObservationFresh, string(raw)) + ref, ok := registry.CredentialRef(row.ID) + if !ok { + t.Fatal("credential ref is unavailable") + } + observedAt := time.UnixMilli(2000) + patches := codex.NormalizePassiveQuotaWindows(map[string]string{ + "X-Codex-Primary-Used-Percent": "0", + "X-Codex-Primary-Window-Minutes": "300", + "X-Codex-Secondary-Used-Percent": "100", + "X-Codex-Secondary-Window-Minutes": "10080", + "X-Codex-Secondary-Reset-At": "1800000000", + "X-Codex-Bengalfox-Primary-Used-Percent": "30", + "X-Codex-Bengalfox-Primary-Window-Minutes": "300", + }, observedAt) + manager.RecordPassiveQuotaObservation(row.ID, ref.IdentityGeneration, observedAt.UnixMilli(), patches) + pending, err := manager.FlushPassiveQuotaObservations(t.Context()) + if err != nil || pending || len(manager.DirtyPassiveQuotaObservations(1)) != 0 { + t.Fatalf("flush pending=%v error=%v", pending, err) + } + var stored models.CredentialObservation + if err := db.Take(&stored, "credential_id = ?", row.ID).Error; err != nil { + t.Fatal(err) + } + if stored.ObservedAtMS == nil || *stored.ObservedAtMS != 2000 || + stored.ObservationVersion != initial.ObservationVersion || stored.State != initial.State { + t.Fatalf("unexpected observation metadata: %#v", stored) + } + var snapshot providerobservation.Snapshot + if err := json.Unmarshal(stored.SnapshotJSON, &snapshot); err != nil { + t.Fatal(err) + } + if len(snapshot.QuotaWindows) != 2 { + t.Fatalf("passive update changed the card's window set: %s", stored.SnapshotJSON) + } + weekly, spark := snapshot.QuotaWindows[0], snapshot.QuotaWindows[1] + if weekly.ID != "primary" || weekly.SourceID != "codex" || + weekly.WindowSeconds == nil || *weekly.WindowSeconds != 604800 || + weekly.Used == nil || *weekly.Used != 100 || weekly.State != "exhausted" || + spark.SourceID != "codex_bengalfox" || spark.Used == nil || *spark.Used != 30 { + t.Fatalf("quota sources were not refreshed independently: %s", stored.SnapshotJSON) + } + views := registry.Snapshot() + if len(views) != 1 || views[0].ObservedQuotaRemaining() == nil || *views[0].ObservedQuotaRemaining() != 0 { + t.Fatalf("Spark or the untracked 5h window polluted account health: %#v", views) + } +} diff --git a/internal/subscription/passive_quota_source_test.go b/internal/subscription/passive_quota_source_test.go new file mode 100644 index 00000000..ffddea97 --- /dev/null +++ b/internal/subscription/passive_quota_source_test.go @@ -0,0 +1,117 @@ +package subscription + +import ( + "bytes" + "encoding/json" + "reflect" + "testing" + "time" + + "gpt-load/internal/subscription/providers/codex" + providerobservation "gpt-load/internal/subscription/providers/observation" +) + +const codexSourceQuotaPayload = `{"plan_type":"pro","rate_limit":{"primary_window":{"used_percent":80,"limit_window_seconds":604800,"reset_at":1800000000}},"additional_rate_limits":[{"metered_feature":"codex_bengalfox","limit_name":"GPT-5.3-Codex-Spark","rate_limit":{"primary_window":{"used_percent":10,"limit_window_seconds":18000,"reset_at":1799900000}}}]}` + +func TestPassiveQuotaMatchesSourceAndPeriodAcrossSlots(t *testing.T) { + raw, err := codex.NormalizeQuota([]byte(codexSourceQuotaPayload), nil) + if err != nil { + t.Fatal(err) + } + var active providerobservation.Snapshot + if err := json.Unmarshal(raw, &active); err != nil { + t.Fatal(err) + } + patches := codex.NormalizePassiveQuotaWindows(map[string]string{ + "X-Codex-Primary-Used-Percent": "5", + "X-Codex-Primary-Window-Minutes": "300", + "X-Codex-Secondary-Used-Percent": "100", + "X-Codex-Secondary-Window-Minutes": "10080", + "X-Codex-Secondary-Reset-At": "1800000600", + "X-Codex-Bengalfox-Secondary-Used-Percent": "30", + "X-Codex-Bengalfox-Secondary-Window-Minutes": "300", + "X-Codex-Bengalfox-Secondary-Reset-At": "1799900600", + }, time.Unix(1799890000, 0)) + merged, err := mergePassiveQuotaSnapshot(raw, patches) + if err != nil { + t.Fatal(err) + } + if !merged.Matched || !merged.Changed || len(merged.Windows) != 2 { + t.Fatalf("matched=%v changed=%v windows=%s, want both existing windows refreshed", merged.Matched, merged.Changed, merged.Encoded) + } + for index, expected := range []struct { + used float64 + reset int64 + state string + }{{100, 1800000600000, "exhausted"}, {30, 1799900600000, "available"}} { + got, previous := merged.Windows[index], active.QuotaWindows[index] + if got.ID != previous.ID || got.SourceID != previous.SourceID || + got.Label != previous.Label || got.Scope != previous.Scope || + !reflect.DeepEqual(got.WindowSeconds, previous.WindowSeconds) || + got.Used == nil || *got.Used != expected.used || + got.ResetAtMS == nil || *got.ResetAtMS != expected.reset || got.State != expected.state { + t.Fatalf("wrong target or metadata overwritten: %s", merged.Encoded) + } + } + var encoded providerobservation.Snapshot + if err := json.Unmarshal(merged.Encoded, &encoded); err != nil || + !reflect.DeepEqual(encoded.QuotaWindows, merged.Windows) { + t.Fatalf("encoded quota windows differ: %s, %v", merged.Encoded, err) + } +} + +func TestPassiveQuotaSourceMatchingRejectsUncertainTargets(t *testing.T) { + period, otherPeriod := int64(18000), int64(604800) + used, utilization := 50.0, 0.5 + patch := providerobservation.QuotaWindow{ + ID: "primary", SourceID: "codex", WindowSeconds: &period, + Used: &used, Utilization: &utilization, State: "exhausted", + } + window := providerobservation.QuotaWindow{ + ID: "primary", SourceID: "codex", WindowSeconds: &period, State: "available", + } + for _, name := range []string{ + "different source same ID and period", "unknown source", "missing patch period", + "missing stored period", "legacy snapshot", "unidentified patch", + "ambiguous stored windows", "duplicate response windows", "period conflict", + } { + t.Run(name, func(t *testing.T) { + windows, patches := []providerobservation.QuotaWindow{window}, []providerobservation.QuotaWindow{patch} + switch name { + case "different source same ID and period": + patches[0].SourceID = "codex_bengalfox" + case "unknown source": + patches[0].SourceID = "codex_unknown" + case "missing patch period": + patches[0].WindowSeconds = nil + case "missing stored period": + windows[0].WindowSeconds = nil + case "legacy snapshot": + windows[0].SourceID = "" + case "unidentified patch": + patches[0].SourceID = "" + case "ambiguous stored windows": + second := window + second.ID = "secondary" + windows = append(windows, second) + case "duplicate response windows": + second := patch + second.ID = "secondary" + patches = append(patches, second) + case "period conflict": + patches[0].WindowSeconds = &otherPeriod + } + raw, err := json.Marshal(providerobservation.Snapshot{QuotaWindows: windows}) + if err != nil { + t.Fatal(err) + } + merged, err := mergePassiveQuotaSnapshot(raw, patches) + if err != nil { + t.Fatal(err) + } + if merged.Matched || merged.Changed || !bytes.Equal(merged.Encoded, raw) { + t.Fatalf("uncertain sample changed data or freshness: matched=%v changed=%v snapshot=%s", merged.Matched, merged.Changed, merged.Encoded) + } + }) + } +} diff --git a/internal/subscription/providers/codex/observation.go b/internal/subscription/providers/codex/observation.go index d796e657..28357ab7 100644 --- a/internal/subscription/providers/codex/observation.go +++ b/internal/subscription/providers/codex/observation.go @@ -36,7 +36,7 @@ func NormalizeQuota(primary, details []byte) ([]byte, error) { } } if rate, ok := object(firstValue(payload, "rate_limit", "rateLimit")); ok { - result.QuotaWindows = append(result.QuotaWindows, normalizeRateWindows(rate, "", "account")...) + result.QuotaWindows = append(result.QuotaWindows, normalizeRateWindows(rate, "", "account", "codex")...) } if additional, ok := firstValue(payload, "additional_rate_limits", "additionalRateLimits").([]any); ok { for index, rawEntry := range additional { @@ -52,7 +52,8 @@ func NormalizeQuota(primary, details []byte) ([]byte, error) { if !ok { continue } - windows := normalizeRateWindows(rate, providerobservation.SafeID(name)+"-", name) + sourceID := cleanString(firstValue(entry, "metered_feature", "meteredFeature")) + windows := normalizeRateWindows(rate, providerobservation.SafeID(name)+"-", name, sourceID) modelIDs := stringsFrom(firstValue(entry, "model_ids", "modelIds")) for windowIndex := range windows { windows[windowIndex].ModelIDs = append([]string(nil), modelIDs...) @@ -112,27 +113,15 @@ var passiveQuotaNamespaceFields = []string{"limit-reached", "limit-name", "allow // passiveQuotaNamespace holds one Codex limit namespace parsed out of the // response headers. The empty namespace is the credential's own rate limit; -// any other namespace is an additional limit identified by its Limit-Name. +// any other namespace identifies an additional metered feature. type passiveQuotaNamespace struct { rateLevel map[string]string windows map[string]map[string]string } -// NormalizePassiveQuotaWindows converts one response's passive quota headers -// into the same windows NormalizeQuota produces from the active JSON payload, -// reusing normalizeRateWindows so Scope, Unit, and State follow one rule -// regardless of source. -// -// Namespaces are discovered from the headers rather than enumerated, so a -// limit Codex introduces later is picked up without a code change as long as -// it follows the same X-Codex--- shape and carries -// an X-Codex--Limit-Name. An additional namespace without that -// name is skipped: its ID could not be made to match the active JSON path. -// -// Label and LabelKey are deliberately left empty. They are derived from the -// window period, which a partial response may omit, and this output only ever -// updates windows an active observation already created -- that observation -// stays the owner of the display metadata. +// NormalizePassiveQuotaWindows 提取响应头中的来源、周期和额度数据。 +// 来源按上游 limit_id 规则与主动查询的 metered_feature 对齐，不依赖 Limit-Name。 +// 展示元数据仍由主动观测负责；缺少周期的样本由合并层拒绝，不能退回槽位匹配。 func NormalizePassiveQuotaWindows(signals map[string]string, observedAt time.Time) []quotaWindow { if len(signals) == 0 { return nil @@ -141,11 +130,12 @@ func NormalizePassiveQuotaWindows(signals map[string]string, observedAt time.Tim namespaces := parsePassiveQuotaNamespaces(signals) for _, key := range sortedNamespaceKeys(namespaces) { namespace := namespaces[key] - prefix, scope := "", "account" + prefix, scope, sourceID := "", "account", "codex" if key != "" { + sourceID += "-" + key limitName := strings.TrimSpace(namespace.rateLevel["limit-name"]) - if limitName == "" || providerobservation.SafeID(limitName) == "" { - continue + if limitName == "" { + limitName = sourceID } scope = limitName prefix = providerobservation.SafeID(limitName) + "-" @@ -154,7 +144,7 @@ func NormalizePassiveQuotaWindows(signals map[string]string, observedAt time.Tim if rate == nil { continue } - for _, window := range normalizeRateWindows(rate, prefix, scope) { + for _, window := range normalizeRateWindows(rate, prefix, scope, sourceID) { // Scope and Unit only served to build the ID and pick the label // rules above; they are cleared, like Label, so the merge updates // nothing but the quota numbers and state. @@ -168,11 +158,8 @@ func NormalizePassiveQuotaWindows(signals map[string]string, observedAt time.Tim return result } -// parsePassiveQuotaNamespaces groups X-Codex-* headers by limit namespace. -// The first path segment equal to "primary" or "secondary" separates the -// namespace from the field name, so a field that merely mentions a window -// (X-Codex-Primary-Over-Secondary-Limit-Percent) stays attached to the window -// that owns it. +// parsePassiveQuotaNamespaces 从已知字段后缀定位槽位，避免将来源名称中的 +// primary/secondary 误认成窗口，或把 over-secondary-limit-percent 当成额度值。 func parsePassiveQuotaNamespaces(signals map[string]string) map[string]*passiveQuotaNamespace { namespaces := make(map[string]*passiveQuotaNamespace) ensure := func(key string) *passiveQuotaNamespace { @@ -192,41 +179,35 @@ func parsePassiveQuotaNamespaces(signals map[string]string) map[string]*passiveQ if !found || suffix == "" { continue } - segments := strings.Split(suffix, "-") - windowIndex := -1 - for index, segment := range segments { - if segment == "primary" || segment == "secondary" { - windowIndex = index + for _, field := range passiveQuotaNamespaceFields { + if suffix == field { + ensure("").rateLevel[field] = value + break + } + if owner, ok := strings.CutSuffix(suffix, "-"+field); ok && owner != "" { + ensure(owner).rateLevel[field] = value break } } - if windowIndex < 0 { - // No window segment: this is either a namespace-wide field or a - // header this normalizer has no meaning for (plan type, credits). - for _, field := range passiveQuotaNamespaceFields { - if suffix == field { - ensure("").rateLevel[field] = value - break - } - if owner, trimmed := strings.CutSuffix(suffix, "-"+field); trimmed && owner != "" { - ensure(owner).rateLevel[field] = value - break - } + for _, field := range []string{"used-percent", "window-minutes", "reset-at", "reset-after-seconds"} { + owner, ok := strings.CutSuffix(suffix, "-"+field) + if !ok { + continue } - continue + namespaceKey, windowName := "", owner + if index := strings.LastIndexByte(owner, '-'); index >= 0 { + namespaceKey, windowName = owner[:index], owner[index+1:] + } + if windowName != "primary" && windowName != "secondary" { + continue + } + namespace := ensure(namespaceKey) + if namespace.windows[windowName] == nil { + namespace.windows[windowName] = map[string]string{} + } + namespace.windows[windowName][field] = value + break } - namespaceKey := strings.Join(segments[:windowIndex], "-") - field := strings.Join(segments[windowIndex+1:], "-") - if field == "" { - continue - } - namespace := ensure(namespaceKey) - window := namespace.windows[segments[windowIndex]] - if window == nil { - window = map[string]string{} - namespace.windows[segments[windowIndex]] = window - } - window[field] = value } return namespaces } @@ -341,7 +322,12 @@ func passiveQuotaBool(value string) (bool, bool) { return parsed, true } -func normalizeRateWindows(rate map[string]any, prefix, scope string) []quotaWindow { +// 与 Codex 的 normalize_limit_id 一致；名称仅用于显示，不能作为来源标识。 +func normalizeQuotaSourceID(value string) string { + return strings.ReplaceAll(strings.ToLower(strings.TrimSpace(value)), "-", "_") +} + +func normalizeRateWindows(rate map[string]any, prefix, scope, sourceID string) []quotaWindow { result := make([]quotaWindow, 0, 2) for _, name := range []string{"primary", "secondary"} { window, ok := object(firstValue(rate, name+"_window", name+"Window")) @@ -350,7 +336,7 @@ func normalizeRateWindows(rate map[string]any, prefix, scope string) []quotaWind } item := quotaWindow{ ID: prefix + name, Label: windowLabel(window, name, scope), LabelKey: codexWindowLabelKey(name, scope), - Scope: scope, Unit: "percent", State: "unknown", + Scope: scope, Unit: "percent", State: "unknown", SourceID: normalizeQuotaSourceID(sourceID), } if used, ok := number(firstValue(window, "used_percent", "usedPercent")); ok { used = math.Max(0, math.Min(100, used)) diff --git a/internal/subscription/providers/codex/observation_test.go b/internal/subscription/providers/codex/observation_test.go index ba902e20..14250b82 100644 --- a/internal/subscription/providers/codex/observation_test.go +++ b/internal/subscription/providers/codex/observation_test.go @@ -71,8 +71,7 @@ func TestNormalizePassiveQuotaWindowsMapsAdditionalLimitNamespaces(t *testing.T) if _, ok := byID["primary"]; !ok { t.Fatalf("windows = %#v, want the account-scope primary window", windows) } - // The Limit-Name only has to produce the matching window ID; Scope, like - // the other display metadata, stays owned by the active observation. + // 保留原有 ID 格式；实际匹配使用 SourceID 和周期，不使用展示名称。 spark, ok := byID["gpt-5-3-codex-spark-primary"] if !ok || spark.Used == nil || *spark.Used != 33 { t.Fatalf("additional primary window = %#v (all=%#v)", spark, windows) @@ -84,8 +83,7 @@ func TestNormalizePassiveQuotaWindowsMapsAdditionalLimitNamespaces(t *testing.T) } func TestNormalizePassiveQuotaWindowsAdditionalIDMatchesActiveJSON(t *testing.T) { - // The passive namespace ID must equal the ID the active JSON path builds - // for the same limit_name, otherwise the merge silently never applies. + // 保留现有窗口 ID 的兼容性；来源与周期的匹配另有回归测试。 raw, err := NormalizeQuota([]byte(`{ "plan_type":"pro", "additional_rate_limits":[{ @@ -115,20 +113,19 @@ func TestNormalizePassiveQuotaWindowsAdditionalIDMatchesActiveJSON(t *testing.T) } } -func TestNormalizePassiveQuotaWindowsSkipsNamespaceWithoutLimitName(t *testing.T) { - // Without a Limit-Name there is no way to build the same stable ID the - // active path uses, so the namespace must be ignored entirely. +func TestNormalizePassiveQuotaWindowsKeepsSourceWithoutLimitName(t *testing.T) { + // 命名空间足以标识来源；缺少展示名称不应丢弃额度信息。 windows := NormalizePassiveQuotaWindows(map[string]string{ "X-Codex-Mysteryns-Primary-Used-Percent": "33", }, time.Now()) - if len(windows) != 0 { - t.Fatalf("windows = %#v, want none without a Limit-Name", windows) + if len(windows) != 1 || windows[0].SourceID != "codex_mysteryns" { + t.Fatalf("windows = %#v, want source identified without a Limit-Name", windows) } } func TestNormalizePassiveQuotaWindowsOmitsAllDisplayMetadata(t *testing.T) { // A passive response only ever updates quota numbers and state on windows - // an active observation already created. Every identity/display field is + // an active observation already created. Every display field is // left empty so the merge cannot overwrite what that observation owns -- // Label in particular is derived from the window period, which a partial // response may not carry. diff --git a/internal/subscription/providers/codex/quota_source_test.go b/internal/subscription/providers/codex/quota_source_test.go new file mode 100644 index 00000000..ee1c747f --- /dev/null +++ b/internal/subscription/providers/codex/quota_source_test.go @@ -0,0 +1,73 @@ +package codex + +import ( + "encoding/json" + "testing" + "time" +) + +func TestQuotaSourcesMatchWithoutDependingOnDisplayName(t *testing.T) { + raw, err := NormalizeQuota([]byte(`{ + "rate_limit":{"primary_window":{"used_percent":80,"limit_window_seconds":604800}}, + "additional_rate_limits":[{ + "metered_feature":" CODEX-FUTURE-TIER ","limit_name":"Display name", + "rate_limit":{"primary_window":{"used_percent":10,"limit_window_seconds":18000}} + }] + }`), nil) + if err != nil { + t.Fatal(err) + } + var active quotaSnapshot + if err := json.Unmarshal(raw, &active); err != nil { + t.Fatal(err) + } + if len(active.QuotaWindows) != 2 || active.QuotaWindows[0].SourceID != "codex" || + active.QuotaWindows[1].SourceID != "codex_future_tier" { + t.Fatalf("active sources = %#v", active.QuotaWindows) + } + // 名称变化不影响来源；Active-Limit 不应把普通窗口改绑到当前受限来源。 + // 周窗口从主动 primary 变为响应头 secondary。 + passive := NormalizePassiveQuotaWindows(map[string]string{ + "X-Codex-Active-Limit": "codex_future_tier", + "X-Codex-Future-Tier-Limit-Name": "Renamed display", + "X-Codex-Secondary-Used-Percent": "90", + "X-Codex-Secondary-Window-Minutes": "10080", + "X-Codex-Future-Tier-Primary-Used-Percent": "20", + "X-Codex-Future-Tier-Primary-Window-Minutes": "300", + }, time.Now()) + if len(passive) != 2 || passive[0].SourceID != active.QuotaWindows[0].SourceID || + passive[1].SourceID != active.QuotaWindows[1].SourceID { + t.Fatalf("passive sources = %#v", passive) + } +} + +func TestQuotaSourceIsNotInferredFromLimitName(t *testing.T) { + raw, err := NormalizeQuota([]byte(`{"additional_rate_limits":[{ + "limit_name":"codex_bengalfox", + "rate_limit":{"primary_window":{"used_percent":10,"limit_window_seconds":18000}} + }]}`), nil) + if err != nil { + t.Fatal(err) + } + var active quotaSnapshot + if err := json.Unmarshal(raw, &active); err != nil { + t.Fatal(err) + } + if len(active.QuotaWindows) != 1 || active.QuotaWindows[0].SourceID != "" { + t.Fatalf("missing metered_feature must remain display-only: %#v", active) + } +} + +func TestPassiveQuotaSourceMayContainWindowNames(t *testing.T) { + windows := NormalizePassiveQuotaWindows(map[string]string{ + "X-Codex-Secondary-Primary-Used-Percent": "25", + "X-Codex-Secondary-Primary-Window-Minutes": "90", + "X-Codex-Secondary-Limit-Name": "Other limit", + "X-Codex-Primary-Over-Secondary-Limit-Percent": "95", + }, time.Now()) + if len(windows) != 1 || windows[0].SourceID != "codex_secondary" || + windows[0].Used == nil || *windows[0].Used != 25 || + windows[0].WindowSeconds == nil || *windows[0].WindowSeconds != 5400 { + t.Fatalf("window name in source was parsed as a slot: %#v", windows) + } +} diff --git a/internal/subscription/providers/observation/snapshot.go b/internal/subscription/providers/observation/snapshot.go index d3b1eddd..7f2aabd3 100644 --- a/internal/subscription/providers/observation/snapshot.go +++ b/internal/subscription/providers/observation/snapshot.go @@ -41,6 +41,8 @@ type AccountSummary struct { } type QuotaWindow struct { + // SourceID 是提供方的额度来源标识，不是展示名称或请求模型。 + SourceID string `json:"source_id,omitempty"` ID string `json:"id"` Label string `json:"label"` LabelKey string `json:"label_key,omitempty"`