fix(subscription): 按来源和周期匹配 Codex 被动额度窗口 (#574)

* fix(subscription): 拒绝跨周期的被动额度覆盖

被动额度合并前校验已有窗口与样本的周期；同 ID 但周期冲突时丢弃整个窗口补丁，避免 7d 被改成 5h，同时污染用量、重置时间及健康摘要。

保留兼容窗口与缺少周期字段的增量更新行为；补充合并与持久化回归测试，不改变限流、额度池或路由调度。

* fix(subscription): 按来源和周期匹配 Codex 被动额度

主动同步保留 metered_feature，响应头按命名空间生成相同来源标识；不依赖模型名称或 Limit-Name。
按同一账号内的来源与实际周期唯一匹配已有窗口，允许 primary/secondary 槽位变化，但保留卡片 ID 与展示元数据。缺少来源/周期或存在歧义时不覆盖；其他渠道继续使用原有 ID 匹配。
补充解析、跨槽位合并、来源隔离、控制面序列化及持久化健康投影测试。无 429、路由或调度改动；旧快照需主动同步一次补齐来源。

* docs(subscription): 修正被动额度快照合并的注释说明

明确无变化时保持原始字节，仅在窗口数据变化时重新编码；不修改任何运行逻辑。

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,

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)
+	}
+}

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
+}

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)
+	}
+}

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)
+	}
+}

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)
+			}
+		})
+	}
+}

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-<namespace>-<window>-<field> shape and carries
-// an X-Codex-<namespace>-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))

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.

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)
+	}
+}

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"`