feat(proxy): 消费 Codex WS 配额帧，实时更新账号用量与自动暂停

.env.example

@@ -50,6 +50,13 @@ CODEX_PORT=8080
 # Codex 上游传输：默认 http；可设 ws 强制使用 WebSocket；auto 在 HTTP 入站下仍走 http
 # CODEX_UPSTREAM_TRANSPORT=http
 
+# WS 逐帧用量快照落库的节流间隔（秒），默认 30。内存快照每帧实时更新；
+# 跨阈值、5pp 变化和 100% 仍会强制落库。大号池可按数据库承压情况调大。
+# CODEX_WS_USAGE_FRAME_PERSIST_SECONDS=30
+
+# WS 逐帧用量摄入开关，默认 true；设为 false 回退到 WHAM 探针与 HTTP 响应头
+# CODEX_WS_USAGE_FRAME_INGEST=true
+
 # 旧版 WebSocket 开关（兼容保留；推荐改用 CODEX_UPSTREAM_TRANSPORT=ws）
 # USE_WEBSOCKET=true
 

CHANGELOG.md

@@ -1,5 +1,23 @@
 # Changelog
 
+## Unreleased
+
+### Features
+
+- **Codex WebSocket quota frames update account usage in real time (#800).** On the persistent WebSocket upstream, quota data arrives in per-turn `codex.rate_limits` and `codex.response.metadata` frames instead of HTTP headers, so account usage only advanced on WHAM probes and a burst could exhaust a 5h window before auto-pause saw it. Every Codex relay path now maps those frames onto the existing header sync: Responses streaming and non-streaming, Chat Completions, downstream WebSocket, and Claude Messages. Connection-level frames that arrive 1–2 seconds after a terminal are captured by a non-blocking observer instead of being dropped. Partial frames never clear a missing window, metered limits (Spark/reserve) cannot overwrite the main 5h/7d windows, a workspace hard stop is preserved because the frame has no `rate_limit_reached_type`, and a late frame from an earlier concurrent turn cannot roll usage back within the same reset window. Relay-style accounts are skipped because forwarded frames describe the relay's own upstream account. A frame without `reset_at` after the stored reset has passed starts the new window instead of inheriting the expired timestamp, and connection-level frames also apply the post-turn credits snapshot and refresh the scheduler so queued dispatches wake when an account recovers. In-memory snapshots update on every frame; database writes are throttled to 30 seconds on a per-account coalescing worker, so a slow or locked database cannot stall the response stream, with forced writes at threshold crossings, 5pp changes and 100% (`CODEX_WS_USAGE_FRAME_PERSIST_SECONDS`). `CODEX_WS_USAGE_FRAME_INGEST=false` turns frame ingestion off and falls back to the WHAM probe and HTTP headers. `/api/admin/accounts/live` now returns runtime usage, status, credits and the last applied frame, and the account page merges it in its 1-second poll. The merge keeps unchanged row identities, and missing-usage and stale-stats retries no longer restart on every poll; a hung live request is aborted after 5 seconds so the next tick can retry.
+
+### Fixes
+
+- **Account live polling refreshes the existing plan and credit badges.** The live endpoint now includes the runtime plan and credit snapshot using the account-list field contract. Missing fields preserve older-backend values, hidden balances clear stale values, and unchanged rows keep their identity; Antigravity rows retain their own projection. Persistent-WebSocket coverage checks post-terminal plan and balance changes through the authenticated live endpoint.
+
+- **Quota-frame reset timestamps require strict integers.** Fractional, exponent-form and overflowing JSON numbers are rejected instead of being truncated or wrapped into reset deadlines; valid usage fields remain usable with the existing missing-reset fallback.
+
+- **Post-terminal quota frames survive observation contention without stalling the WebSocket reader.** Active and idle frames share an ordered runtime path; a bounded per-account FIFO retains partial updates and requests an authoritative probe on overflow. Credits, plan, quota and cooldown writes are coalesced off the reader, including exhausted-window and credits-only frames. Old observations cannot roll credits or frame freshness backward, 7d-only writes retain their reset, and delayed cooldown publication reads current state so it cannot overwrite a newer authentication error or resurrect a cleared cooldown. Failed cooldown writes retry on later observations without another failure count; repeated real Responses 429s still fence older in-flight successes. A persistent-WebSocket regression exercises the production idle observer through scheduling and the authenticated live endpoint, with no subsequent business request or probe.
+
+- **Quota observations cannot shorten an active account cooldown.** The 5h and 7d usage paths keep a longer existing cooldown and its reason, including an authoritative Responses 429 or authentication error. Later resets extend the cooldown without replacing that reason or counting another failure. A real Responses 429 can upgrade a local usage cooldown to `responses_rate_limited` without shortening it, so `ignore_usage_limit_status` cannot bypass an actual rejection; a stronger authentication failure retains priority. Usage snapshots still update when the cooldown is unchanged, and repeated observations do not write the cooldown again.
+
+- **Account page copy matches the actual quota and cleanup behavior.** The auto-pause threshold hint now says blank or 0 inherits the group/global threshold and that the switch above disables that window, instead of claiming 0 disables it. The rate-limited cleanup dialog and result toast say accounts move to the recycle bin and that auto-paused accounts appear in the cleanup when their runtime status is rate-limited or `usage_exhausted`, instead of saying status flags are cleared.
+
 ## v3.1.1 - 2026-10-17
 
 ### Features

admin/account_live.go

@@ -2,11 +2,14 @@ package admin
 
 import (
 	"net/http"
+	"time"
 
 	"github.com/gin-gonic/gin"
 )
 
 type accountLiveItem struct {
+	// A nil embedded projection omits Codex credit fields for other planes.
+	*AccountLiveCredits
 	ActiveRequests   int64 `json:"active_requests"`
 	OccupiedRequests int64 `json:"occupied_requests"`
 	// DynamicConcurrencyLimit is the admission cap the scheduler currently
@@ -15,11 +18,43 @@ type accountLiveItem struct {
 	// BaseConcurrencyEffective is the configured cap (account override,
 	// then group, then the global default) before any runtime reduction.
 	BaseConcurrencyEffective int64 `json:"base_concurrency_effective"`
+
+	// Runtime usage/status projection so the account page can reflect WS frame
+	// observations without waiting for a list reload. Optional pointers keep the
+	// backward-compat contract: an older backend omits the key (frontend keeps
+	// the list value); a pointer that is nil serializes as null (frontend clears
+	// the now-invalid window).
+	Status           *string  `json:"status,omitempty"`
+	UsingCredits     *bool    `json:"using_credits,omitempty"`
+	PlanType         *string  `json:"plan_type,omitempty"`
+	UsagePercent5h   *float64 `json:"usage_percent_5h"`
+	Reset5hAt        *string  `json:"reset_5h_at"`
+	UsageUpdatedAt5h *string  `json:"codex_5h_usage_updated_at"`
+	UsagePercent7d   *float64 `json:"usage_percent_7d"`
+	Reset7dAt        *string  `json:"reset_7d_at"`
+	UsageUpdatedAt   *string  `json:"codex_usage_updated_at"`
+	// Frame freshness proves the WS quota ingestion path is working for this
+	// account; omitted until the first frame is applied.
+	UsageFrameAt   *string `json:"codex_usage_frame_at,omitempty"`
+	UsageFrameKind *string `json:"codex_usage_frame_kind,omitempty"`
+}
+
+// AccountLiveCredits uses the existing account-list credit field contract.
+type AccountLiveCredits struct {
+	Valid                bool    `json:"credits_valid"`
+	Balance              *string `json:"credits_balance"`
+	HasCredits           *bool   `json:"credits_has_credits"`
+	Unlimited            *bool   `json:"credits_unlimited"`
+	OverageLimitReached  *bool   `json:"credits_overage_limit_reached"`
+	SpendControlReached  *bool   `json:"credits_spend_control_reached"`
+	RateLimitReachedType string  `json:"credits_rate_limit_reached_type"`
 }
 
 // GetAccountLiveState returns request-local runtime counters for the visible
 // account page. Scheduler counters use in-memory atomics; concurrency caps are
-// the scheduler's maintained values and are not recomputed here.
+// the scheduler's maintained values and are not recomputed here. The usage
+// projection is a single read lock per account and never rebuilds a snapshot,
+// so it stays cheap at the one-second cadence even for a full page.
 func (h *Handler) GetAccountLiveState(c *gin.Context) {
 	ids, err := parseAccountListIDs(c.Query("ids"))
 	if err != nil {
@@ -37,15 +72,60 @@ func (h *Handler) GetAccountLiveState(c *gin.Context) {
 		if account == nil {
 			continue
 		}
-		live[id] = accountLiveItem{
+		item := accountLiveItem{
 			ActiveRequests:           account.GetActiveRequests(),
 			OccupiedRequests:         account.GetOccupiedRequests(),
 			DynamicConcurrencyLimit:  account.GetDynamicConcurrencyLimit(),
 			BaseConcurrencyEffective: account.GetBaseConcurrencyEffective(),
 		}
+		// Antigravity rows use a persisted quota snapshot; runtimeStatusLocked
+		// does not model their display status, so leave the row's fields alone.
+		if !account.IsAntigravityAPI() {
+			usage := account.GetAccountLiveUsageSnapshot()
+			item.PlanType = &usage.PlanType
+			item.AccountLiveCredits = &AccountLiveCredits{Valid: usage.CreditsValid}
+			if usage.CreditsValid {
+				item.Balance = usage.Credits.Balance
+				item.HasCredits = &usage.Credits.HasCredits
+				item.Unlimited = &usage.Credits.Unlimited
+				item.OverageLimitReached = &usage.Credits.OverageLimitReached
+				item.SpendControlReached = usage.Credits.SpendControlReached
+				item.RateLimitReachedType = usage.Credits.RateLimitReachedType
+			}
+			status := usage.Status
+			item.Status = &status
+			usingCredits := usage.UsingCredits
+			item.UsingCredits = &usingCredits
+			if usage.UsagePercent5hValid {
+				pct := usage.UsagePercent5h
+				item.UsagePercent5h = &pct
+			}
+			item.Reset5hAt = optionalLiveTime(usage.Reset5hAt)
+			item.UsageUpdatedAt5h = optionalLiveTime(usage.UsageUpdatedAt5h)
+			if usage.UsagePercent7dValid {
+				pct := usage.UsagePercent7d
+				item.UsagePercent7d = &pct
+			}
+			item.Reset7dAt = optionalLiveTime(usage.Reset7dAt)
+			item.UsageUpdatedAt = optionalLiveTime(usage.UsageUpdatedAt)
+			if !usage.UsageFrameLastAt.IsZero() {
+				item.UsageFrameAt = optionalLiveTime(usage.UsageFrameLastAt)
+				kind := usage.UsageFrameLastKind
+				item.UsageFrameKind = &kind
+			}
+		}
+		live[id] = item
 	}
 	c.JSON(http.StatusOK, gin.H{
 		"accounts":                    live,
 		"session_slot_buffer_enabled": h.store.SessionSlotBufferEnabled(),
 	})
 }
+
+func optionalLiveTime(t time.Time) *string {
+	if t.IsZero() {
+		return nil
+	}
+	formatted := t.Format(time.RFC3339)
+	return &formatted
+}

admin/account_live_test.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"testing"
+	"time"
 
 	"github.com/codex2api/auth"
 	"github.com/gin-gonic/gin"
@@ -58,3 +59,128 @@ func TestGetAccountLiveStateReturnsVisibleInflightCounts(t *testing.T) {
 		t.Fatal("missing account unexpectedly returned")
 	}
 }
+
+func TestGetAccountLiveStateReturnsRuntimeUsage(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	store := auth.NewStore(nil, nil, nil)
+	now := time.Now()
+	account := &auth.Account{
+		DBID:        7,
+		AccessToken: "token",
+		PlanType:    "plus",
+		Status:      auth.StatusReady,
+		HealthTier:  auth.HealthTierHealthy,
+	}
+	account.UsagePercent5h = 42
+	account.UsagePercent5hValid = true
+	account.Reset5hAt = now.Add(time.Hour)
+	account.UsageUpdatedAt5h = now
+	account.UsagePercent7d = 12
+	account.UsagePercent7dValid = true
+	account.Reset7dAt = now.Add(24 * time.Hour)
+	account.UsageUpdatedAt = now
+	store.AddAccount(account)
+	store.ObserveUsageFramePlan(account, "pro")
+	balance := "12.5"
+	account.ApplySparseCreditsHeaders(true, false, &balance, nil, "")
+	handler := &Handler{store: store}
+
+	recorder := httptest.NewRecorder()
+	ctx, _ := gin.CreateTestContext(recorder)
+	ctx.Request = httptest.NewRequest(http.MethodGet, "/api/admin/accounts/live?ids=7", nil)
+	handler.GetAccountLiveState(ctx)
+
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("status = %d, body=%s", recorder.Code, recorder.Body.String())
+	}
+	var response struct {
+		Accounts map[string]accountLiveItem `json:"accounts"`
+	}
+	if err := json.Unmarshal(recorder.Body.Bytes(), &response); err != nil {
+		t.Fatal(err)
+	}
+	item := response.Accounts["7"]
+	if item.UsagePercent5h == nil || *item.UsagePercent5h != 42 {
+		t.Fatalf("usage_percent_5h = %v, want 42", item.UsagePercent5h)
+	}
+	if item.UsagePercent7d == nil || *item.UsagePercent7d != 12 {
+		t.Fatalf("usage_percent_7d = %v, want 12", item.UsagePercent7d)
+	}
+	if item.Reset5hAt == nil || item.Reset7dAt == nil {
+		t.Fatalf("reset times missing: %+v", item)
+	}
+	if item.Status == nil || *item.Status == "" {
+		t.Fatalf("status missing: %+v", item)
+	}
+	if item.UsingCredits == nil {
+		t.Fatalf("using_credits missing: %+v", item)
+	}
+	var raw struct {
+		Accounts map[string]map[string]json.RawMessage `json:"accounts"`
+	}
+	if err := json.Unmarshal(recorder.Body.Bytes(), &raw); err != nil {
+		t.Fatal(err)
+	}
+	for key, want := range map[string]string{"plan_type": `"pro"`, "credits_valid": "true", "credits_balance": `"12.5"`, "credits_has_credits": "true", "credits_unlimited": "false", "credits_spend_control_reached": "null"} {
+		if got := string(raw.Accounts["7"][key]); got != want {
+			t.Errorf("live %s=%s, want %s", key, got, want)
+		}
+	}
+}
+
+func TestGetAccountLiveStateSerializesNullWindowsAndSkipsAntigravityFields(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	store := auth.NewStore(nil, nil, nil)
+	codex := &auth.Account{
+		DBID:        11,
+		AccessToken: "token",
+		Status:      auth.StatusReady,
+		HealthTier:  auth.HealthTierHealthy,
+	}
+	antigravity := &auth.Account{
+		DBID:         12,
+		AccessToken:  "token",
+		UpstreamType: auth.UpstreamAntigravity,
+	}
+	store.AddAccount(codex)
+	store.AddAccount(antigravity)
+	codex.MarkUsageFrameObserved("codex.rate_limits", time.Now())
+	handler := &Handler{store: store}
+
+	recorder := httptest.NewRecorder()
+	ctx, _ := gin.CreateTestContext(recorder)
+	ctx.Request = httptest.NewRequest(http.MethodGet, "/api/admin/accounts/live?ids=11,12", nil)
+	handler.GetAccountLiveState(ctx)
+
+	if recorder.Code != http.StatusOK {
+		t.Fatalf("status = %d, body=%s", recorder.Code, recorder.Body.String())
+	}
+	var raw struct {
+		Accounts map[string]map[string]json.RawMessage `json:"accounts"`
+	}
+	if err := json.Unmarshal(recorder.Body.Bytes(), &raw); err != nil {
+		t.Fatal(err)
+	}
+	codexItem := raw.Accounts["11"]
+	if got := string(codexItem["usage_percent_5h"]); got != "null" {
+		t.Fatalf("codex usage_percent_5h = %s, want null", got)
+	}
+	if got := string(codexItem["status"]); got == "" || got == "null" {
+		t.Fatalf("codex status missing: %s", got)
+	}
+	if got := string(codexItem["codex_usage_frame_kind"]); got != `"codex.rate_limits"` {
+		t.Fatalf("codex frame kind = %s, want codex.rate_limits", got)
+	}
+	if got := string(codexItem["codex_usage_frame_at"]); got == "" || got == "null" {
+		t.Fatalf("codex frame time missing: %s", got)
+	}
+	antigravityItem := raw.Accounts["12"]
+	for _, key := range []string{"status", "using_credits", "plan_type", "credits_valid", "credits_balance", "codex_usage_frame_at", "codex_usage_frame_kind"} {
+		if value, ok := antigravityItem[key]; ok {
+			t.Fatalf("antigravity item unexpectedly carries %s=%s", key, value)
+		}
+	}
+	if got := string(antigravityItem["usage_percent_5h"]); got != "null" {
+		t.Fatalf("antigravity usage_percent_5h = %s, want null", got)
+	}
+}

auth/oauth_refresh_lock.go

@@ -453,11 +453,8 @@ func (s *Store) finishReloadedOAuthRefresh(ctx context.Context, acc *Account) {
 		}
 	}
 	if expiredCooldown {
-		s.deleteCachedAccountCooldown(dbID)
-		if s.db != nil {
-			_ = s.db.ClearCooldown(ctx, dbID)
-		}
+		s.persistAccountCooldownState(acc, true, false)
 	} else if !activeCooldown && s.db != nil {
-		_ = s.db.ClearError(ctx, dbID)
+		s.persistAccountCooldownState(acc, true, true)
 	}
 }