mirror of
https://github.com/router-for-me/CLIProxyAPI.git
synced 2026-09-07 00:24:17 +08:00
Codex and Claude already emit credential-level quota watermarks on ordinary
responses. CPA used to drop them. Keep the latest watermark in memory and
return it from the management auth-file API.
Hard rule: this is observation only. It must not change scheduling, cooldown
selection, or auth-file persistence.
Snapshot, not accumulation
- QuotaState now has ObservedAt and a bounded Signals map. MarkResult fills
them from the response headers already recorded on the request.
- Signals is the current response, not a union of earlier ones. Retry-After
and "limit reached" only appear on the response that produced them; merging
across responses would keep an expired value forever.
- A response with no quota header (transport failure, 5xx, unrelated endpoint)
leaves the previous snapshot in place.
- ObservedAt is the time of the current snapshot. It advances even when the
values did not change, so a consumer can tell a fresh reading from a stale
one.
- When two model states merge, keep the newer snapshot. Do not union keys
captured at different times.
What is observed, and what is not
- One predicate, ProviderSupportsQuotaObservation, decides the provider set.
- Keep Codex and Claude. Drop Kimi, xAI/Grok, Antigravity, and the Gemini
family (gemini/vertex/aistudio): their ordinary headers are not a reliable
credential-level remaining quota.
- Count-tokens reuses the credential but is not generation traffic.
ExecuteCount sets SkipQuotaObservation so those headers cannot replace the
last generation snapshot. Cooldown and success/failure accounting still run.
Cooldown must not overwrite the last snapshot
- Observation writes only ObservedAt and Signals.
- Cooldown writes only Exceeded, Reason, NextRecoverAt, and BackoffLevel,
through applyCooldownFields. Never assign a fresh QuotaState{...} over a
live value: that would zero the snapshot on 429, Cloudflare, credential-
scope sibling updates, and cooldown clears.
- If a credential-quota cooldown is still active, MarkResult still observes
an already-present model state. It does not create scheduler state just to
record a watermark.
- .cds files persist cooldownFieldsOf(Quota) only. Restore keeps the newer
ObservedAt, so reloading cooldown cannot clobber a newer in-memory snapshot.
- cooldownQuotaEqual still ignores observation fields, so a watermark change
cannot by itself persist cooldown or move the scheduler.
- The management payload omits every cooldown field, so it cannot be mistaken
for scheduler state or wired back into scheduling.
- Manual ResetQuota still clears the full QuotaState.
Codex websocket events
- Codex WS reports quota as codex.rate_limits frames, not HTTP headers.
ParseCodexQuotaEventHeaders turns one event into the same bounded header
shape, and MergeResponseHeaders folds it into the request-scoped holder.
additional_rate_limits is accepted as an object (websocket) or an array
(/wham/usage).
- Parse only through AppendCodexAPIWebsocketResponse. The shared
AppendAPIWebsocketResponse is also used by xAI, and xAI error frames really
do carry x-ratelimit-* headers. Parsing every frame as Codex quota would
forge Codex headers into another provider's request log.
- Also capture code_review_rate_limits.
- A malformed active-limit name drops only that one header, not the window
watermarks parsed from the same event.
- The type discriminator scans a bounded frame prefix, not every byte of
every frame.
- HTTP namespaces an extra limit by short name (x-codex-bengalfox-*); WS
namespaces it by limit name (GPT-5.3-Codex-Spark). The two paths cannot
emit the same header names. The X-Codex-Additional- prefix marks the WS
origin, and snapshot replacement keeps the two spellings from piling up.
Hardening
- Reject observed values with control characters. These strings reach the
plain-text request log, and Limit-Name is upstream-controlled, so CR/LF
could forge a header line.
- When the header cap is hit, keep plan/credits/primary ahead of
additional-limit namespaces, then sort names so truncation is deterministic.
- QuotaState.Clone deep-copies Signals and is used by Auth.Clone and
ModelState.Clone.
- Token stores still serialize credential metadata only, so observation adds
no auth-file writes.
180 lines
4.5 KiB
Go
180 lines
4.5 KiB
Go
package logging
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"sync"
|
|
"sync/atomic"
|
|
)
|
|
|
|
type endpointKey struct{}
|
|
type responseStatusKey struct{}
|
|
type responseHeadersKey struct{}
|
|
type clientRequestMetadataKey struct{}
|
|
|
|
// ClientRequestMetadata stores immutable downstream request metadata for asynchronous consumers.
|
|
type ClientRequestMetadata struct {
|
|
ClientIP string
|
|
XForwardedFor string
|
|
UserAgent string
|
|
}
|
|
|
|
type responseStatusHolder struct {
|
|
status atomic.Int32
|
|
}
|
|
|
|
type responseHeadersHolder struct {
|
|
mu sync.RWMutex
|
|
headers http.Header
|
|
}
|
|
|
|
func WithEndpoint(ctx context.Context, endpoint string) context.Context {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
return context.WithValue(ctx, endpointKey{}, endpoint)
|
|
}
|
|
|
|
func GetEndpoint(ctx context.Context) string {
|
|
if ctx == nil {
|
|
return ""
|
|
}
|
|
if endpoint, ok := ctx.Value(endpointKey{}).(string); ok {
|
|
return endpoint
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// WithClientRequestMetadata stores a snapshot of downstream request metadata in ctx.
|
|
func WithClientRequestMetadata(ctx context.Context, metadata ClientRequestMetadata) context.Context {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
return context.WithValue(ctx, clientRequestMetadataKey{}, metadata)
|
|
}
|
|
|
|
// GetClientRequestMetadata returns downstream request metadata stored in ctx.
|
|
func GetClientRequestMetadata(ctx context.Context) ClientRequestMetadata {
|
|
if ctx == nil {
|
|
return ClientRequestMetadata{}
|
|
}
|
|
if metadata, ok := ctx.Value(clientRequestMetadataKey{}).(ClientRequestMetadata); ok {
|
|
return metadata
|
|
}
|
|
return ClientRequestMetadata{}
|
|
}
|
|
|
|
func WithResponseStatusHolder(ctx context.Context) context.Context {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
if holder, ok := ctx.Value(responseStatusKey{}).(*responseStatusHolder); ok && holder != nil {
|
|
return ctx
|
|
}
|
|
return context.WithValue(ctx, responseStatusKey{}, &responseStatusHolder{})
|
|
}
|
|
|
|
func WithResponseHeadersHolder(ctx context.Context) context.Context {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
if holder, ok := ctx.Value(responseHeadersKey{}).(*responseHeadersHolder); ok && holder != nil {
|
|
return ctx
|
|
}
|
|
return context.WithValue(ctx, responseHeadersKey{}, &responseHeadersHolder{})
|
|
}
|
|
|
|
// WithFreshResponseHeadersHolder starts an isolated upstream response attempt.
|
|
// Unlike WithResponseHeadersHolder, it always shadows any holder inherited from
|
|
// the parent request so a later retry cannot observe headers from an earlier
|
|
// credential or model attempt.
|
|
func WithFreshResponseHeadersHolder(ctx context.Context) context.Context {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
return context.WithValue(ctx, responseHeadersKey{}, &responseHeadersHolder{})
|
|
}
|
|
|
|
func SetResponseStatus(ctx context.Context, status int) {
|
|
if ctx == nil || status <= 0 {
|
|
return
|
|
}
|
|
holder, ok := ctx.Value(responseStatusKey{}).(*responseStatusHolder)
|
|
if !ok || holder == nil {
|
|
return
|
|
}
|
|
holder.status.Store(int32(status))
|
|
}
|
|
|
|
func SetResponseHeaders(ctx context.Context, headers http.Header) {
|
|
if ctx == nil {
|
|
return
|
|
}
|
|
holder, ok := ctx.Value(responseHeadersKey{}).(*responseHeadersHolder)
|
|
if !ok || holder == nil {
|
|
return
|
|
}
|
|
holder.mu.Lock()
|
|
defer holder.mu.Unlock()
|
|
holder.headers = cloneHTTPHeader(headers)
|
|
}
|
|
|
|
// MergeResponseHeaders adds headers observed after the initial HTTP response,
|
|
// such as quota metadata delivered in a websocket event.
|
|
func MergeResponseHeaders(ctx context.Context, headers http.Header) {
|
|
if ctx == nil || len(headers) == 0 {
|
|
return
|
|
}
|
|
holder, ok := ctx.Value(responseHeadersKey{}).(*responseHeadersHolder)
|
|
if !ok || holder == nil {
|
|
return
|
|
}
|
|
holder.mu.Lock()
|
|
defer holder.mu.Unlock()
|
|
if holder.headers == nil {
|
|
holder.headers = make(http.Header, len(headers))
|
|
}
|
|
for key, values := range headers {
|
|
canonicalKey := http.CanonicalHeaderKey(key)
|
|
if canonicalKey == "" {
|
|
continue
|
|
}
|
|
holder.headers[canonicalKey] = append([]string(nil), values...)
|
|
}
|
|
}
|
|
|
|
func GetResponseStatus(ctx context.Context) int {
|
|
if ctx == nil {
|
|
return 0
|
|
}
|
|
holder, ok := ctx.Value(responseStatusKey{}).(*responseStatusHolder)
|
|
if !ok || holder == nil {
|
|
return 0
|
|
}
|
|
return int(holder.status.Load())
|
|
}
|
|
|
|
func GetResponseHeaders(ctx context.Context) http.Header {
|
|
if ctx == nil {
|
|
return nil
|
|
}
|
|
holder, ok := ctx.Value(responseHeadersKey{}).(*responseHeadersHolder)
|
|
if !ok || holder == nil {
|
|
return nil
|
|
}
|
|
holder.mu.RLock()
|
|
defer holder.mu.RUnlock()
|
|
return cloneHTTPHeader(holder.headers)
|
|
}
|
|
|
|
func cloneHTTPHeader(src http.Header) http.Header {
|
|
if len(src) == 0 {
|
|
return nil
|
|
}
|
|
dst := make(http.Header, len(src))
|
|
for key, values := range src {
|
|
dst[key] = append([]string(nil), values...)
|
|
}
|
|
return dst
|
|
}
|