fix(codex): 修复 WebSocket 普通账号额度停滞 (#656)

* fix(codex): 修复 WebSocket 普通账号额度停滞

* fix(codex): 支持 WS 额度跨周期重锚

* fix(codex): 改用当前 WS 事件匹配额度来源

* fix(codex): 隔离解析 WS 额度窗口

* fix(codex): 统一额度比较时基并恢复事件预检查

internal/execution/cpa/websocket_quota_test.go

@@ -65,7 +65,7 @@ func TestWebsocketQuotaEventsRefreshBeforeTurnCompletesWithoutHeaders(t *testing
 				t.Fatal("quota observation changed the forwarded event")
 			}
 			dirty := manager.DirtyPassiveQuotaObservations(1)
-			if len(dirty) != 1 || len(dirty[0].Windows) != 2 {
+			if len(dirty) != 1 || len(dirty[0].Windows) != 3 {
 				t.Fatalf("quota event was not recorded before downstream delivery: %#v", dirty)
 			}
 			if dirty[0].ObservedAtMS < startedAt || dirty[0].Version <= lastVersion {
@@ -80,6 +80,87 @@ func TestWebsocketQuotaEventsRefreshBeforeTurnCompletesWithoutHeaders(t *testing
 	}
 }
 
+func TestWebsocketQuotaReusedSessionRefreshesAccountPercentageWithoutHeaders(t *testing.T) {
+	adapter, db, _, crypt, credential := newAdapterFixture(t, credentialJSON("access", "refresh", time.Now().Add(time.Hour)))
+	manager := adapter.credentials.(*subscription.CredentialManager)
+	active, err := os.ReadFile("../../subscription/providers/codex/testdata/quota-active.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw, err := codex.NormalizeQuota(active, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var snapshot providerobservation.Snapshot
+	if err := json.Unmarshal(raw, &snapshot); err != nil {
+		t.Fatal(err)
+	}
+	for index := range snapshot.QuotaWindows {
+		window := &snapshot.QuotaWindows[index]
+		if window.SourceID == "codex" {
+			*window.Used, *window.Remaining, *window.Utilization = 93, 7, .93
+		}
+	}
+	raw, err = json.Marshal(snapshot)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.AutoMigrate(&models.CredentialObservation{}); err != nil {
+		t.Fatal(err)
+	}
+	storedAt := int64(1000)
+	row := models.CredentialObservation{CredentialID: credential.ID, State: models.CredentialObservationFresh,
+		SnapshotJSON: raw, ObservedAtMS: &storedAt, ObservationVersion: 1, UpdatedAtMS: storedAt}
+	if err := db.Create(&row).Error; err != nil {
+		t.Fatal(err)
+	}
+	payload, err := os.ReadFile("../../subscription/providers/codex/testdata/quota-ws-account.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	inner := &quotaEventSession{}
+	session := &observedWebsocketSession{WebsocketSession: inner, adapter: adapter, spec: validSpec(t, credential, crypt)}
+	for _, used := range []int{94, 95} {
+		// 沿用实测事件结构，仅改变用量，模拟同一连接连续收到的新额度。
+		inner.event = bytes.Replace(payload, []byte(`"used_percent": 6`), fmt.Appendf(nil, `"used_percent": %d`, used), 1)
+		result := session.ExecuteTurn(t.Context(), session.spec.Body, func(_ context.Context, forwarded []byte) error {
+			if !bytes.Equal(forwarded, inner.event) {
+				t.Fatal("quota observation changed the forwarded event")
+			}
+			if pending, err := manager.FlushPassiveQuotaObservations(t.Context()); err != nil || pending {
+				t.Fatalf("flush: pending=%t error=%v", pending, err)
+			}
+			if err := db.Take(&row, "credential_id = ?", credential.ID).Error; err != nil {
+				t.Fatal(err)
+			}
+			if err := json.Unmarshal(row.SnapshotJSON, &snapshot); err != nil {
+				t.Fatal(err)
+			}
+			if row.ObservedAtMS == nil || *row.ObservedAtMS <= storedAt || len(snapshot.QuotaWindows) != 3 {
+				t.Fatalf("quota observation was not refreshed: %+v", row)
+			}
+			for _, window := range snapshot.QuotaWindows {
+				want := float64(used)
+				if window.SourceID == "codex_bengalfox" {
+					want = 0
+				}
+				if window.Used == nil || *window.Used != want || window.Remaining == nil || *window.Remaining != 100-want {
+					t.Fatalf("source=%s want used=%.0f remaining=%.0f: %s", window.SourceID, want, 100-want, row.SnapshotJSON)
+				}
+			}
+			storedAt = *row.ObservedAtMS
+			return nil
+		})
+		if result.Error != nil {
+			t.Fatalf("turn failed: %+v", result.Error)
+		}
+		// 生产观测按毫秒排序，下一轮须具有更晚的信号时间。
+		for time.Now().UnixMilli() <= storedAt {
+			time.Sleep(time.Millisecond)
+		}
+	}
+}
+
 func TestWebsocketQuotaKeepsHandshakeWhenEventOnlyUpdatesSpark(t *testing.T) {
 	for _, flushDuringTurn := range []bool{false, true} {
 		t.Run(fmt.Sprintf("flush_during_turn=%t", flushDuringTurn), func(t *testing.T) {
@@ -102,7 +183,7 @@ func TestWebsocketQuotaKeepsHandshakeWhenEventOnlyUpdatesSpark(t *testing.T) {
 			if err := db.Create(&row).Error; err != nil {
 				t.Fatal(err)
 			}
-			event, err := os.ReadFile("../../subscription/providers/codex/testdata/quota-ws-account.json")
+			event, err := os.ReadFile("../../subscription/providers/codex/testdata/quota-ws-spark.json")
 			if err != nil {
 				t.Fatal(err)
 			}

internal/subscription/passive_quota_flush.go

@@ -97,7 +97,7 @@ func (manager *CredentialManager) flushOnePassiveQuotaObservationLocked(
 		}
 		if !merge.Matched {
 			// 未能唯一匹配已有窗口的样本不能推进同步时间。
-			// 窗口创建和周期变更仍由主动观测负责。
+			// 窗口创建和身份变更仍由主动观测负责。
 			manager.passiveQuota.ack(observation.CredentialID, observation.Version)
 			return nil
 		}

internal/subscription/passive_quota_websocket_test.go

@@ -3,6 +3,7 @@ package subscription
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"os"
 	"testing"
@@ -13,6 +14,51 @@ import (
 	providerobservation "gpt-load/internal/subscription/providers/observation"
 )
 
+func TestMergeDelayedWebsocketQuotaDoesNotOverwriteAccountWithSpark(t *testing.T) {
+	raw, err := codex.NormalizeQuota([]byte(`{
+		"rate_limit":{"primary_window":{"used_percent":93,"limit_window_seconds":604800,"reset_at":1800010000}},
+		"additional_rate_limits":[{"limit_name":"Spark","metered_feature":"codex_bengalfox",
+			"rate_limit":{"secondary_window":{"used_percent":7,"limit_window_seconds":604800,"reset_at":1800000000}}}]
+	}`), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, swap := range []bool{false, true} {
+		t.Run(fmt.Sprintf("swap=%t", swap), func(t *testing.T) {
+			top, named := `"reset_at":1800000000`, `"reset_after_seconds":100`
+			if swap {
+				top, named = named, top
+			}
+			payload := []byte(fmt.Sprintf(`{
+				"type":"codex.rate_limits",
+				"rate_limits":{"primary":{"used_percent":5,"window_minutes":10080,%s}},
+				"additional_rate_limits":{"Spark":{"secondary":{"used_percent":5,"window_minutes":10080,%s}}}
+			}`, top, named))
+			patches := codex.NormalizeWebsocketQuotaWindows(payload, time.Unix(1799999905, 0))
+			merged, err := mergePassiveQuotaSnapshot(raw, patches)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if len(merged.Windows) != 2 || !merged.Changed {
+				t.Fatalf("valid Spark update was lost: %s", merged.Encoded)
+			}
+			for _, window := range merged.Windows {
+				wantUsed, wantRemaining, wantReset := 5.0, 95.0, int64(1800000005000)
+				if swap {
+					wantReset = 1800000000000
+				}
+				if window.SourceID == "codex" {
+					wantUsed, wantRemaining, wantReset = 93, 7, 1800010000000
+				}
+				if window.Used == nil || *window.Used != wantUsed || window.Remaining == nil ||
+					*window.Remaining != wantRemaining || window.ResetAtMS == nil || *window.ResetAtMS != wantReset {
+					t.Fatalf("source=%s quota=%s", window.SourceID, merged.Encoded)
+				}
+			}
+		})
+	}
+}
+
 func TestFlushWebsocketQuotaCapturedEventsKeepAccountAndSparkSeparate(t *testing.T) {
 	for _, sample := range []string{"quota-ws-account.json", "quota-ws-spark.json"} {
 		t.Run(sample, func(t *testing.T) {
@@ -73,7 +119,7 @@ func TestFlushWebsocketQuotaCapturedEventsKeepAccountAndSparkSeparate(t *testing
 	}
 }
 
-func TestCapturedHTTPAndWebsocketNamedQuotasAgreeAfterMatching(t *testing.T) {
+func TestCapturedHTTPAndWebsocketQuotasAgreeAfterMatching(t *testing.T) {
 	active, err := os.ReadFile("providers/codex/testdata/quota-active.json")
 	if err != nil {
 		t.Fatal(err)
@@ -82,7 +128,7 @@ func TestCapturedHTTPAndWebsocketNamedQuotasAgreeAfterMatching(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	// 故意留下旧的普通额度，验证 WS 不会靠数值推断无来源窗口。
+	// 故意留下旧的普通额度，验证 WS 能匹配普通窗口，又不会被 Spark 覆盖。
 	var baseline providerobservation.Snapshot
 	if err := json.Unmarshal(raw, &baseline); err != nil {
 		t.Fatal(err)
@@ -126,12 +172,6 @@ func TestCapturedHTTPAndWebsocketNamedQuotasAgreeAfterMatching(t *testing.T) {
 			}
 			for index, httpWindow := range httpMerged.Windows {
 				wsWindow := wsMerged.Windows[index]
-				if wsWindow.SourceID == "codex" {
-					if *wsWindow.Used != 90 {
-						t.Fatal("WS event without source changed account quota")
-					}
-					continue
-				}
 				if httpWindow.SourceID != wsWindow.SourceID || *httpWindow.WindowSeconds != *wsWindow.WindowSeconds ||
 					*httpWindow.Used != *wsWindow.Used || *httpWindow.Remaining != *wsWindow.Remaining || httpWindow.State != wsWindow.State {
 					t.Fatalf("HTTP and WS disagree for source=%s period=%d", httpWindow.SourceID, *httpWindow.WindowSeconds)
@@ -141,6 +181,38 @@ func TestCapturedHTTPAndWebsocketNamedQuotasAgreeAfterMatching(t *testing.T) {
 	}
 }
 
+func TestMergeWebsocketQuotaUsesCurrentEventAcrossResetCycles(t *testing.T) {
+	raw, err := codex.NormalizeQuota([]byte(`{
+		"rate_limit":{"primary_window":{"used_percent":93,"limit_window_seconds":604800,"reset_at":1800000000}},
+		"additional_rate_limits":[{"limit_name":"Spark","metered_feature":"codex_bengalfox",
+			"rate_limit":{"secondary_window":{"used_percent":7,"limit_window_seconds":604800,"reset_at":1800010000}}}]
+	}`), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	patches := codex.NormalizeWebsocketQuotaWindows([]byte(`{
+		"type":"codex.rate_limits",
+		"rate_limits":{"secondary":{"used_percent":94,"window_minutes":10080,"reset_at":1800604800}},
+		"additional_rate_limits":{"Spark":{"primary":{"used_percent":5,"window_minutes":10080,"reset_at":1800610000}}}
+	}`), time.Unix(1800000001, 0))
+	merged, err := mergePassiveQuotaSnapshot(raw, patches)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, window := range merged.Windows {
+		wantUsed, wantReset := 5.0, int64(1800610000000)
+		if window.SourceID == "codex" {
+			wantUsed, wantReset = 94, 1800604800000
+		}
+		if window.Used == nil || *window.Used != wantUsed || window.ResetAtMS == nil || *window.ResetAtMS != wantReset {
+			t.Fatalf("source=%s was not refreshed from the current event: %s", window.SourceID, merged.Encoded)
+		}
+		if window.SourceName != "" {
+			t.Fatal("transient source hint reached the stored snapshot")
+		}
+	}
+}
+
 func TestPassiveQuotaSourceNameRequiresUniqueOriginalNameAndPeriod(t *testing.T) {
 	period := int64(18000)
 	window := providerobservation.QuotaWindow{SourceID: "codex_bengalfox", ID: "spark-primary", Scope: "GPT-5.3-Codex-Spark", WindowSeconds: &period}

internal/subscription/providers/codex/websocket_quota.go

@@ -1,28 +1,28 @@
 package codex
 
 import (
+	"bytes"
 	"encoding/json"
+	"io"
 	"math"
 	"sort"
-	"strconv"
 	"strings"
 	"time"
 
 	"github.com/buger/jsonparser"
 )
 
-type websocketQuotaWindow struct {
-	UsedPercent       *float64 `json:"used_percent"`
-	WindowMinutes     *int64   `json:"window_minutes"`
-	ResetAt           *int64   `json:"reset_at"`
-	ResetAfterSeconds *int64   `json:"reset_after_seconds"`
+// 原始绝对时间与同一事件中的倒计时分别比较，不能混入本地接收时间。
+type websocketQuotaWindowAnchor struct {
+	windowSeconds     int64
+	resetAtSeconds    int64
+	resetAfterSeconds int64
 }
 
-type websocketQuotaRate struct {
-	Allowed      *bool                 `json:"allowed"`
-	LimitReached *bool                 `json:"limit_reached"`
-	Primary      *websocketQuotaWindow `json:"primary"`
-	Secondary    *websocketQuotaWindow `json:"secondary"`
+type websocketQuotaRateObservation struct {
+	windows              []quotaWindow
+	comparisonWindows    map[string]websocketQuotaWindowAnchor
+	comparisonIncomplete bool
 }
 
 // NormalizeWebsocketQuotaWindows 读取 Codex 原生额度事件，不改动发给客户端的消息。
@@ -32,92 +32,192 @@ func NormalizeWebsocketQuotaWindows(payload []byte, observedAt time.Time) []quot
 	if err != nil || kind != "codex.rate_limits" {
 		return nil
 	}
-	var event struct {
-		RateLimits           *websocketQuotaRate            `json:"rate_limits"`
-		MeteredLimitName     string                         `json:"metered_limit_name"`
-		LimitName            string                         `json:"limit_name"`
-		AdditionalRateLimits map[string]*websocketQuotaRate `json:"additional_rate_limits"`
-	}
-	if json.Unmarshal(payload, &event) != nil || len(event.AdditionalRateLimits) > 8 {
+	event, ok := decodeWebsocketQuotaEvent(payload)
+	if !ok || cleanString(event["type"]) != "codex.rate_limits" {
 		return nil
 	}
-	additional := make([]quotaWindow, 0, 2*len(event.AdditionalRateLimits))
-	names := make([]string, 0, len(event.AdditionalRateLimits))
-	for name := range event.AdditionalRateLimits {
+
+	var additionalRates map[string]any
+	additionalValid := true
+	if rawAdditional, present := event["additional_rate_limits"]; present && rawAdditional != nil {
+		additionalRates, additionalValid = object(rawAdditional)
+	}
+	if additionalValid && len(additionalRates) > 8 {
+		return nil
+	}
+	additional := make([]quotaWindow, 0, 2*len(additionalRates))
+	comparisonWindows := make([]websocketQuotaWindowAnchor, 0, 2*len(additionalRates))
+	comparisonIncomplete := !additionalValid
+	names := make([]string, 0, len(additionalRates))
+	for name := range additionalRates {
 		names = append(names, name)
 	}
 	sort.Strings(names)
-	for _, name := range names {
-		windows := normalizeWebsocketQuotaRate(event.AdditionalRateLimits[name], observedAt)
-		name = strings.TrimSpace(name)
+	for _, rawName := range names {
+		observation := normalizeWebsocketQuotaRate(additionalRates[rawName], observedAt)
+		for _, anchor := range observation.comparisonWindows {
+			comparisonWindows = append(comparisonWindows, anchor)
+		}
+		comparisonIncomplete = comparisonIncomplete || observation.comparisonIncomplete
+		name := strings.TrimSpace(rawName)
 		if name == "" {
 			continue
 		}
-		for _, window := range windows {
+		for _, window := range observation.windows {
 			window.SourceName = name
 			additional = append(additional, window)
 		}
 	}
-	sourceID := normalizeQuotaSourceID(event.MeteredLimitName)
+
+	sourceID := normalizeQuotaSourceID(cleanString(event["metered_limit_name"]))
 	if sourceID == "" {
-		sourceID = normalizeQuotaSourceID(event.LimitName)
+		sourceID = normalizeQuotaSourceID(cleanString(event["limit_name"]))
 	}
-	if sourceID == "" && len(event.AdditionalRateLimits) > 0 {
-		// 与 HTTP 相同：有附加额度但缺少来源时不能采用顶层数据。
-		// 数值相等或不同均不能证明它属于普通账号或某个附加来源。
-		return additional
-	}
-	if sourceID == "" || sourceID == codexAccountActiveLimit {
+	if sourceID == codexAccountActiveLimit {
 		sourceID = codexAccountQuotaSourceID
 	}
-	primary := normalizeWebsocketQuotaRate(event.RateLimits, observedAt)
-	for index := range primary {
-		primary[index].SourceID = sourceID
+
+	primary := normalizeWebsocketQuotaRate(event["rate_limits"], observedAt)
+	result := make([]quotaWindow, 0, len(primary.windows)+len(additional))
+	for _, window := range primary.windows {
+		if sourceID == "" {
+			if !websocketQuotaTopLevelIsAccount(primary.comparisonWindows[window.ID], comparisonWindows, comparisonIncomplete) {
+				continue
+			}
+			window.SourceID = codexAccountQuotaSourceID
+		} else {
+			window.SourceID = sourceID
+		}
+		result = append(result, window)
 	}
-	// 具名来源尚未解析，不能在这里按数值去重；合并层按实际目标窗口处理副本。
-	return append(primary, additional...)
+	// 具名来源仍由已有快照解析 SourceID；顶层副本已在本事件内完成去重。
+	return append(result, additional...)
 }
 
-func normalizeWebsocketQuotaRate(rate *websocketQuotaRate, observedAt time.Time) []quotaWindow {
-	if rate == nil {
+func decodeWebsocketQuotaEvent(payload []byte) (map[string]any, bool) {
+	decoder := json.NewDecoder(bytes.NewReader(payload))
+	decoder.UseNumber()
+	var event map[string]any
+	if decoder.Decode(&event) != nil || event == nil {
+		return nil, false
+	}
+	var trailing any
+	if decoder.Decode(&trailing) != io.EOF {
+		return nil, false
+	}
+	return event, true
+}
+
+// websocketQuotaTopLevelIsAccount 用同一事件内的附加窗口排除顶层副本。
+// 槽位和用量不代表身份；同周期窗口需要在共同时间基准下排除副本。
+func websocketQuotaTopLevelIsAccount(window websocketQuotaWindowAnchor, additional []websocketQuotaWindowAnchor, comparisonIncomplete bool) bool {
+	// 存在无法读取周期的附加窗口时，不能证明顶层窗口不属于该来源。
+	if comparisonIncomplete {
+		return false
+	}
+	for _, candidate := range additional {
+		if window.windowSeconds != candidate.windowSeconds {
+			continue
+		}
+		compared := false
+		for _, pair := range [][2]int64{
+			{window.resetAtSeconds, candidate.resetAtSeconds},
+			{window.resetAfterSeconds, candidate.resetAfterSeconds},
+		} {
+			if pair[0] == 0 || pair[1] == 0 {
+				continue
+			}
+			compared = true
+			delta := pair[0] - pair[1]
+			// 任一共同基准指向副本就跳过；两种基准的判断冲突也不会写入账号。
+			if delta >= -1 && delta <= 1 {
+				return false
+			}
+		}
+		if !compared {
+			return false
+		}
+	}
+	return true
+}
+
+// normalizeWebsocketQuotaRate 分开保留来源比较所需的窗口锚点和可写入的额度值。
+// 单个窗口的用量损坏不会抹掉其周期身份，也不会阻断同一事件里的其他有效窗口。
+func normalizeWebsocketQuotaRate(raw any, observedAt time.Time) websocketQuotaRateObservation {
+	var result websocketQuotaRateObservation
+	if raw == nil {
+		return result
+	}
+	rate, valid := object(raw)
+	if !valid {
+		result.comparisonIncomplete = true
+		return result
+	}
+	allowed, hasAllowed := rate["allowed"].(bool)
+	limitReached, hasLimitReached := rate["limit_reached"].(bool)
+	result.comparisonWindows = make(map[string]websocketQuotaWindowAnchor, 2)
+	for _, slot := range []string{"primary", "secondary"} {
+		rawWindow, exists := rate[slot]
+		if !exists || rawWindow == nil {
+			continue
+		}
+		window, valid := object(rawWindow)
+		if !valid {
+			result.comparisonIncomplete = true
+			continue
+		}
+		minutes, ok := integer(window["window_minutes"])
+		if !ok || minutes <= 0 || minutes > passiveQuotaMaxResetAtSeconds/60 {
+			result.comparisonIncomplete = true
+			continue
+		}
+		seconds := minutes * 60
+		anchor := websocketQuotaWindowAnchor{windowSeconds: seconds}
+		if absolute, ok := integer(window["reset_at"]); ok && absolute > 0 && absolute <= passiveQuotaMaxResetAtSeconds {
+			anchor.resetAtSeconds = absolute
+		}
+		if relative, ok := integer(window["reset_after_seconds"]); ok && relative > 0 && relative <= passiveQuotaMaxResetAtSeconds {
+			anchor.resetAfterSeconds = relative
+		}
+		result.comparisonWindows[slot] = anchor
+		resetAtMS := websocketQuotaResetAtMS(anchor, observedAt)
+
+		used, ok := number(window["used_percent"])
+		if !ok || math.IsNaN(used) || math.IsInf(used, 0) || used < 0 || used > 100 {
+			continue
+		}
+		limit, remaining, utilization := 100.0, 100-used, used/100
+		state := "available"
+		if used >= 100 || (hasAllowed && !allowed) || (hasLimitReached && limitReached) {
+			state = "exhausted"
+		}
+		result.windows = append(result.windows, quotaWindow{
+			ID:            slot,
+			Used:          &used,
+			Limit:         &limit,
+			Remaining:     &remaining,
+			Utilization:   &utilization,
+			ResetAtMS:     resetAtMS,
+			WindowSeconds: &seconds,
+			State:         state,
+		})
+	}
+	return result
+}
+
+func websocketQuotaResetAtMS(anchor websocketQuotaWindowAnchor, observedAt time.Time) *int64 {
+	if anchor.resetAtSeconds > 0 {
+		value := anchor.resetAtSeconds * 1000
+		return &value
+	}
+	relative := anchor.resetAfterSeconds
+	if relative == 0 {
 		return nil
 	}
-	fields := map[string]any{}
-	if rate.Allowed != nil {
-		fields["allowed"] = *rate.Allowed
+	base := observedAt.Unix()
+	if base < 0 || relative > passiveQuotaMaxResetAtSeconds-base {
+		return nil
 	}
-	if rate.LimitReached != nil {
-		fields["limit_reached"] = *rate.LimitReached
-	}
-	for _, slot := range []struct {
-		name   string
-		window *websocketQuotaWindow
-	}{{"primary", rate.Primary}, {"secondary", rate.Secondary}} {
-		window := slot.window
-		if window == nil {
-			continue
-		}
-		if window.UsedPercent == nil || math.IsNaN(*window.UsedPercent) || math.IsInf(*window.UsedPercent, 0) ||
-			*window.UsedPercent < 0 || *window.UsedPercent > 100 || window.WindowMinutes == nil ||
-			*window.WindowMinutes <= 0 || *window.WindowMinutes > passiveQuotaMaxResetAtSeconds/60 {
-			continue
-		}
-		values := map[string]string{
-			"used-percent":   strconv.FormatFloat(*window.UsedPercent, 'f', -1, 64),
-			"window-minutes": strconv.FormatInt(*window.WindowMinutes, 10),
-		}
-		if window.ResetAt != nil {
-			values["reset-at"] = strconv.FormatInt(*window.ResetAt, 10)
-		}
-		if window.ResetAfterSeconds != nil {
-			values["reset-after-seconds"] = strconv.FormatInt(*window.ResetAfterSeconds, 10)
-		}
-		fields[slot.name+"_window"] = passiveQuotaWindowFields(values, observedAt)
-	}
-	windows := normalizeRateWindows(fields, "", "", "")
-	for index := range windows {
-		windows[index].Label, windows[index].LabelKey = "", ""
-		windows[index].Scope, windows[index].Unit = "", ""
-	}
-	return windows
+	value := (base + relative) * 1000
+	return &value
 }

internal/subscription/providers/codex/websocket_quota_test.go

@@ -2,22 +2,105 @@ package codex
 
 import (
 	"encoding/json"
+	"fmt"
 	"os"
+	"strings"
 	"testing"
 	"time"
 )
 
-func TestWebsocketQuotaDoesNotGuessAccountSourceWhenMeteredValuesDiffer(t *testing.T) {
+func TestWebsocketQuotaComparisonUsesSharedClock(t *testing.T) {
+	for _, test := range []struct {
+		name         string
+		topReset     string
+		otherReset   string
+		accountCount int
+	}{
+		{"absolute copy", `"reset_at":1800000000`, `"reset_at":1800000001`, 0},
+		{"absolute distinct", `"reset_at":1800000000`, `"reset_at":1800000003`, 1},
+		{"relative copy", `"reset_after_seconds":100`, `"reset_after_seconds":101`, 0},
+		{"relative distinct", `"reset_after_seconds":100`, `"reset_after_seconds":103`, 1},
+		{"mixed clocks", `"reset_at":1800000000`, `"reset_after_seconds":100`, 0},
+		{"shared relative clock", `"reset_at":1800000000,"reset_after_seconds":100`, `"reset_after_seconds":100`, 0},
+		{"shared absolute clock", `"reset_at":1800000000,"reset_after_seconds":100`, `"reset_at":1800000000`, 0},
+		{"conflicting duplicate evidence", `"reset_at":1800000000,"reset_after_seconds":100`, `"reset_at":1800000003,"reset_after_seconds":100`, 0},
+		{"missing clock", `"reset_at":1800000000`, `"reset_at":null`, 0},
+		{"invalid absolute with relative copy", `"reset_at":"bad","reset_after_seconds":100`, `"reset_after_seconds":100`, 0},
+	} {
+		for _, swap := range []bool{false, true} {
+			for _, delay := range []int64{0, 5} {
+				t.Run(fmt.Sprintf("%s/swap=%t/delay=%d", test.name, swap, delay), func(t *testing.T) {
+					top, other := test.topReset, test.otherReset
+					if swap {
+						top, other = other, top
+					}
+					payload := []byte(fmt.Sprintf(`{
+						"type":"codex.rate_limits",
+						"rate_limits":{"primary":{"used_percent":5,"window_minutes":10080,%s}},
+						"additional_rate_limits":{"Spark":{"secondary":{"used_percent":20,"window_minutes":10080,%s}}}
+					}`, top, other))
+					windows := NormalizeWebsocketQuotaWindows(payload, time.Unix(1799999900+delay, 0))
+					account, named := 0, 0
+					for _, window := range windows {
+						if window.SourceID == "codex" {
+							account++
+						} else if window.SourceName == "Spark" && window.Used != nil && *window.Used == 20 {
+							named++
+						}
+					}
+					if account != test.accountCount || named != 1 {
+						t.Fatalf("account=%d named=%d, want account=%d named=1", account, named, test.accountCount)
+					}
+				})
+			}
+		}
+	}
+}
+
+func TestWebsocketQuotaNonQuotaEventsHaveBoundedAllocations(t *testing.T) {
+	text := strings.Repeat("x", 1<<20)
+	for _, payload := range [][]byte{
+		[]byte(`{"type":"response.output_text.delta","delta":"` + text + `"}`),
+		[]byte(`{"delta":"` + text + `","type":"response.output_text.delta"}`),
+	} {
+		allocs := testing.AllocsPerRun(5, func() {
+			if len(NormalizeWebsocketQuotaWindows(payload, time.Time{})) != 0 {
+				t.Fatal("non-quota event produced quota")
+			}
+		})
+		if allocs > 4 {
+			t.Fatalf("non-quota event allocated %.0f objects, want a lightweight type check (at most 4)", allocs)
+		}
+	}
+}
+
+func BenchmarkWebsocketQuotaNonQuotaEvent(b *testing.B) {
+	text := strings.Repeat("x", 1<<20)
+	for name, payload := range map[string][]byte{
+		"type_first": []byte(`{"type":"response.output_text.delta","delta":"` + text + `"}`),
+		"type_last":  []byte(`{"delta":"` + text + `","type":"response.output_text.delta"}`),
+	} {
+		b.Run(name, func(b *testing.B) {
+			b.ReportAllocs()
+			for b.Loop() {
+				NormalizeWebsocketQuotaWindows(payload, time.Time{})
+			}
+		})
+	}
+}
+
+func TestWebsocketQuotaDropsSameEventCopyWithinResetPrecision(t *testing.T) {
 	payload, err := os.ReadFile("testdata/quota-ws-spark.json")
 	if err != nil {
 		t.Fatal(err)
 	}
-	// 同一事件的顶层与具名额度存在数值差异，也不能把无来源顶层归属普通账号。
+	// 同一事件里的副本允许 1 秒 reset_at 精度差；用量值不参与窗口身份判断。
 	var event map[string]any
 	if err := json.Unmarshal(payload, &event); err != nil {
 		t.Fatal(err)
 	}
 	weekly := event["rate_limits"].(map[string]any)["secondary"].(map[string]any)
+	weekly["used_percent"] = float64(42)
 	weekly["reset_at"] = weekly["reset_at"].(float64) + 1
 	payload, err = json.Marshal(event)
 	if err != nil {
@@ -25,7 +108,7 @@ func TestWebsocketQuotaDoesNotGuessAccountSourceWhenMeteredValuesDiffer(t *testi
 	}
 	windows := NormalizeWebsocketQuotaWindows(payload, time.Unix(1000, 0))
 	if len(windows) != 2 {
-		t.Fatalf("windows = %#v, want both Spark windows", windows)
+		t.Fatalf("windows = %#v, want only named Spark windows", windows)
 	}
 	for _, window := range windows {
 		if window.SourceID != "" || window.SourceName != "GPT-5.3-Codex-Spark" {
@@ -40,7 +123,7 @@ func TestWebsocketQuotaCapturedAccountAndSparkEvents(t *testing.T) {
 		file         string
 		accountCount int
 	}{
-		{"quota-ws-account.json", 0},
+		{"quota-ws-account.json", 1},
 		{"quota-ws-spark.json", 0},
 	} {
 		t.Run(test.file, func(t *testing.T) {
@@ -56,7 +139,7 @@ func TestWebsocketQuotaCapturedAccountAndSparkEvents(t *testing.T) {
 			for _, window := range windows {
 				if window.SourceID == "codex" {
 					account++
-					if *window.Used != 6 || *window.Remaining != 94 || *window.WindowSeconds != 604800 {
+					if test.file == "quota-ws-account.json" && (*window.Used != 6 || *window.Remaining != 94 || *window.WindowSeconds != 604800) {
 						t.Fatalf("account quota=%#v", window)
 					}
 				} else if window.SourceName == "GPT-5.3-Codex-Spark" && window.SourceID == "" {
@@ -97,7 +180,6 @@ func TestWebsocketQuotaRejectsUnusableOrAmbiguousSignals(t *testing.T) {
 		`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":101,"window_minutes":300}}}`,
 		`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":1,"window_minutes":0}}}`,
 		`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":1}}}`,
-		`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":1,"window_minutes":300}},"additional_rate_limits":{"Unknown":null}}`,
 		`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":1,"window_minutes":300}}} {}`,
 	} {
 		if got := NormalizeWebsocketQuotaWindows([]byte(payload), time.Unix(1000, 0)); len(got) != 0 {
@@ -117,6 +199,140 @@ func TestWebsocketQuotaDropsMeteredCopyEvenWhenSlotsDiffer(t *testing.T) {
 	}
 }
 
+func TestWebsocketQuotaTreatsMissingOrEmptyAdditionalLimitsAsAccount(t *testing.T) {
+	for _, additional := range []string{"", `,"additional_rate_limits":null`, `,"additional_rate_limits":{}`, `,"additional_rate_limits":{"Unknown":null}`} {
+		payload := []byte(`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":1,"window_minutes":300,"reset_at":1800000000}}` + additional + `}`)
+		windows := NormalizeWebsocketQuotaWindows(payload, time.Unix(1000, 0))
+		if len(windows) != 1 || windows[0].SourceID != "codex" {
+			t.Fatalf("additional=%q windows=%#v, want one account window", additional, windows)
+		}
+	}
+	windows := NormalizeWebsocketQuotaWindows([]byte(`{
+		"type":"codex.rate_limits",
+		"rate_limits":{"primary":{"used_percent":1,"window_minutes":300}}
+	}`), time.Unix(1000, 0))
+	if len(windows) != 1 || windows[0].SourceID != "codex" || windows[0].ResetAtMS != nil {
+		t.Fatalf("top-level account window without reset was rejected: %#v", windows)
+	}
+}
+
+func TestWebsocketQuotaComparesEveryAdditionalWindow(t *testing.T) {
+	for _, test := range []struct {
+		name         string
+		topReset     int64
+		accountCount int
+	}{
+		{"matches second source", 1800000000, 0},
+		{"differs from every source", 1800000003, 1},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			windows := NormalizeWebsocketQuotaWindows([]byte(fmt.Sprintf(`{
+				"type":"codex.rate_limits",
+				"rate_limits":{"primary":{"used_percent":40,"window_minutes":10080,"reset_at":%d}},
+				"additional_rate_limits":{
+					"FiveHour":{"primary":{"used_percent":10,"window_minutes":300,"reset_at":1700000000}},
+					"Weekly":{"secondary":{"used_percent":20,"window_minutes":10080,"reset_at":1800000000}}
+				}
+			}`, test.topReset)), time.Unix(1000, 0))
+			account := 0
+			for _, window := range windows {
+				if window.SourceID == "codex" {
+					account++
+				}
+			}
+			if len(windows) != 2+test.accountCount || account != test.accountCount {
+				t.Fatalf("windows=%#v, want account count %d", windows, test.accountCount)
+			}
+		})
+	}
+}
+
+func TestWebsocketQuotaSkipsTopLevelWhenSamePeriodAdditionalCannotBeCompared(t *testing.T) {
+	for _, payload := range []string{
+		`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":40,"window_minutes":10080,"reset_at":1800000000}},"additional_rate_limits":{"Unknown":{"secondary":{"used_percent":20,"window_minutes":10080}}}}`,
+		`{"type":"codex.rate_limits","rate_limits":{"primary":{"used_percent":40,"window_minutes":10080}},"additional_rate_limits":{"Unknown":{"secondary":{"used_percent":20,"window_minutes":10080,"reset_at":1800000000}}}}`,
+	} {
+		windows := NormalizeWebsocketQuotaWindows([]byte(payload), time.Unix(1000, 0))
+		if len(windows) != 1 || windows[0].SourceName != "Unknown" || windows[0].SourceID != "" {
+			t.Fatalf("ambiguous top-level window was retained: %#v", windows)
+		}
+	}
+}
+
+func TestWebsocketQuotaMalformedAdditionalWindowDoesNotMakeTopLevelAccount(t *testing.T) {
+	for _, test := range []struct {
+		name   string
+		window string
+	}{
+		{"missing period", `{"used_percent":20,"reset_at":1800000000}`},
+		{"missing usage", `{"window_minutes":10080,"reset_at":1800000000}`},
+		{"null usage", `{"used_percent":null,"window_minutes":10080,"reset_at":1800000000}`},
+		{"out of range usage", `{"used_percent":101,"window_minutes":10080,"reset_at":1800000000}`},
+		{"invalid period", `{"used_percent":20,"window_minutes":0,"reset_at":1800000000}`},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			payload := []byte(fmt.Sprintf(`{
+				"type":"codex.rate_limits",
+				"rate_limits":{"primary":{"used_percent":5,"window_minutes":10080,"reset_at":1800000000}},
+				"additional_rate_limits":{"Spark":{"primary":%s}}
+			}`, test.window))
+			windows := NormalizeWebsocketQuotaWindows(payload, time.Unix(1000, 0))
+			if len(windows) != 0 {
+				t.Fatalf("malformed additional window produced an update: %#v", windows)
+			}
+		})
+	}
+}
+
+func TestWebsocketQuotaIsolatesMalformedWindowFields(t *testing.T) {
+	windows := NormalizeWebsocketQuotaWindows([]byte(`{
+		"type":"codex.rate_limits","metered_limit_name":"premium",
+		"rate_limits":{
+			"primary":{"used_percent":5,"window_minutes":10080,"reset_at":1800000000},
+			"secondary":{"used_percent":"bad","window_minutes":300,"reset_at":1700000000}
+		},
+		"additional_rate_limits":{
+			"Bad":{"primary":{"used_percent":"bad","window_minutes":10080,"reset_at":1800000000}},
+			"Good":{"allowed":"bad","primary":{"used_percent":20,"window_minutes":300,"reset_at":1700000000}}
+		}
+	}`), time.Unix(1000, 0))
+	if len(windows) != 2 {
+		t.Fatalf("windows=%#v, want valid top-level and named windows", windows)
+	}
+	if windows[0].SourceID != codexAccountQuotaSourceID || windows[0].Used == nil || *windows[0].Used != 5 {
+		t.Fatalf("valid top-level window was lost: %#v", windows)
+	}
+	if windows[1].SourceName != "Good" || windows[1].SourceID != "" || windows[1].Used == nil || *windows[1].Used != 20 {
+		t.Fatalf("valid additional window was lost: %#v", windows)
+	}
+}
+
+func TestWebsocketQuotaDeduplicatesTopLevelPerWindow(t *testing.T) {
+	windows := NormalizeWebsocketQuotaWindows([]byte(`{
+		"type":"codex.rate_limits",
+		"rate_limits":{
+			"primary":{"used_percent":40,"window_minutes":300,"reset_at":1700000000},
+			"secondary":{"used_percent":50,"window_minutes":10080,"reset_at":1800000000}
+		},
+		"additional_rate_limits":{"Spark":{
+			"primary":{"used_percent":10,"window_minutes":300,"reset_at":1700000000},
+			"secondary":{"used_percent":20,"window_minutes":10080,"reset_at":1800000100}
+		}}
+	}`), time.Unix(1000, 0))
+	account := 0
+	for _, window := range windows {
+		if window.SourceID == "codex" {
+			account++
+			if window.WindowSeconds == nil || *window.WindowSeconds != 604800 {
+				t.Fatalf("wrong top-level window retained: %#v", window)
+			}
+		}
+	}
+	if len(windows) != 3 || account != 1 {
+		t.Fatalf("windows=%#v, want one account window and two named windows", windows)
+	}
+}
+
 func TestWebsocketQuotaKeepsExplicitDifferentSourcesWithEqualValues(t *testing.T) {
 	windows := NormalizeWebsocketQuotaWindows([]byte(`{
 		"type":"codex.rate_limits","metered_limit_name":"premium",