diff --git a/.github/scripts/fork-check.sh b/.github/scripts/fork-check.sh index fbb5cb9a826..6405a016f3b 100755 --- a/.github/scripts/fork-check.sh +++ b/.github/scripts/fork-check.sh @@ -13,6 +13,7 @@ fork_packages=( ./internal/api/... ./internal/config/... ./internal/registry/... + ./internal/runtime/executor/... ./internal/watcher/... ./sdk/api/handlers/... ./sdk/cliproxy/... @@ -53,6 +54,11 @@ require_hook sdk/cliproxy/auth/conductor_selection.go 'availableAuthsForPoolSele require_hook sdk/cliproxy/auth/conductor_selection.go 'HasAccountPools(ctx)' 2 require_hook sdk/api/handlers/handlers_execution.go 'h.accountPoolExecutionContext(ctx)' 2 require_hook sdk/api/handlers/handlers_stream.go 'h.accountPoolExecutionContext(ctx)' +require_hook internal/runtime/executor/meta_executor_stream.go 'helps.ObserveMetaSubscriptionUsage(ctx, eventData)' +require_hook internal/runtime/executor/meta_executor_execute.go 'helps.ObserveMetaSubscriptionUsageSSE(ctx, data)' +require_hook sdk/cliproxy/auth/quota_signals.go 'metaQuotaSignalHeaders[name]' +require_hook sdk/cliproxy/auth/conductor_lifecycle.go 'auth.Quota = mergeQuotaObservation(auth.Quota, existing.Quota)' +require_hook sdk/api/handlers/openai/openai_responses_handlers.go 'responsesSSETrailingFrame(frame)' echo "== fork tests" go test -count=1 -run '^TestFork' "${fork_packages[@]}" diff --git a/FORK.md b/FORK.md index 2caf785faa7..c3eef6091c5 100644 --- a/FORK.md +++ b/FORK.md @@ -21,6 +21,7 @@ Everything upstream documents still applies. This page covers only what the fork | `client.native-model-lists` | `false` | Claude Code is only shown Claude models; Codex is only shown Codex models. | | `client.key-scopes` | none | Limit a client API key to the credentials of the providers it lists. | | (no setting) | on | Serve the Muse Code client endpoints (`/muse-code/models` and friends) so the Muse CLI can use the proxy. | +| (no setting) | on | Record Meta's subscription usage on the credential and pass its event through to Responses clients. | `docker/fork/config.default.yaml`, the configuration the image writes on first start, has every one of them turned on. `config.example.yaml` is left exactly as upstream ships it, so it does not @@ -185,6 +186,30 @@ to an endpoint named there, not to one passed with `--base-url`) and give it a c export META_API_KEY="sk-muse-77b0..." ``` +### Meta subscription usage + +Meta has no usage endpoint. It reports the account's subscription windows with every Responses +stream, in a `response.subscription_usage` event after `response.completed`: + +```json +{"type":"response.subscription_usage","subscription":{"tier":"…", + "window":{"used_percent":12,"window_duration_mins":300,"resets_at":1791465557}, + "weekly":{"used_percent":5,"resets_at":1791763200}}} +``` + +The fork records that event on the Meta credential as quota signals, the same way Claude and Codex +rate-limit headers are recorded: `X-Meta-Window-Used-Percent`, `X-Meta-Window-Minutes`, +`X-Meta-Window-Reset-At`, `X-Meta-Weekly-Used-Percent`, `X-Meta-Weekly-Reset-At` and `X-Meta-Tier` +(percentages verbatim, reset times in unix seconds). They appear under `quota` in the management +`auth-files` list, which is what T3 Code's usage view reads, and the soonest-reset order uses +them. A credential reports nothing until something has used it through the proxy. + +Responses clients also receive the event, although it arrives after the terminal event; the Muse +CLI reads it to show its quota. Every other frame after a terminal event is still dropped. + +Reloading a credential file from disk keeps its last observation, since the file does not store +it. This applies to every provider. + ## Container image `ghcr.io/ntindle/cliproxyapi:latest` is built by `.github/workflows/fork-image.yml` from @@ -255,5 +280,9 @@ The fork changes these upstream files, each by a few lines, so these are where c | `sdk/api/handlers/handlers_execution.go`, `sdk/api/handlers/handlers_stream.go` | Apply key scopes to the providers resolved for a request. | | `sdk/api/handlers/handlers_interceptors.go` | Applies key scopes to every model list. | | `internal/api/server_routes.go` | Registers the Muse Code endpoints. | +| `internal/runtime/executor/meta_executor_stream.go`, `internal/runtime/executor/meta_executor_execute.go` | Record Meta's subscription usage event. | +| `sdk/cliproxy/auth/quota_signals.go` | Accepts the `X-Meta-*` quota signals. | +| `sdk/cliproxy/auth/conductor_lifecycle.go` | Keeps the quota observation when a credential file is reloaded. | +| `sdk/api/handlers/openai/openai_responses_handlers.go` | Passes the subscription usage event after the terminal event. | Everything else the fork adds lives in files upstream does not have. diff --git a/internal/runtime/executor/helps/meta_quota.go b/internal/runtime/executor/helps/meta_quota.go new file mode 100644 index 00000000000..78c3b0a9d2f --- /dev/null +++ b/internal/runtime/executor/helps/meta_quota.go @@ -0,0 +1,113 @@ +package helps + +import ( + "bytes" + "context" + "net/http" + "strconv" + "strings" + + "github.com/router-for-me/CLIProxyAPI/v8/internal/logging" + "github.com/tidwall/gjson" +) + +// MetaSubscriptionUsageEventType is the Responses stream event Meta sends after +// response.completed with the account's subscription usage. +const MetaSubscriptionUsageEventType = "response.subscription_usage" + +// Quota signal names for Meta subscription usage. They follow the X-Codex-* +// pattern so the passive quota snapshot and the management API treat both alike. +const ( + MetaQuotaTierHeader = "X-Meta-Tier" + MetaQuotaWindowUsedHeader = "X-Meta-Window-Used-Percent" + MetaQuotaWindowMinutesHeader = "X-Meta-Window-Minutes" + MetaQuotaWindowResetHeader = "X-Meta-Window-Reset-At" + MetaQuotaWeeklyUsedHeader = "X-Meta-Weekly-Used-Percent" + MetaQuotaWeeklyResetHeader = "X-Meta-Weekly-Reset-At" +) + +// ParseMetaSubscriptionUsageHeaders converts one Meta response.subscription_usage +// event into the bounded header representation used for passive quota +// observations. Meta sends it once per response, after response.completed: +// +// {"type":"response.subscription_usage","subscription":{"tier":"…", +// "window":{"used_percent":0,"window_duration_mins":300,"resets_at":1791465557}, +// "weekly":{"used_percent":5,"resets_at":1791763200}}} +// +// window is the five-hour-class block and weekly the rolling week. Percentages +// pass through verbatim (above 100 means over quota) and reset times stay unix +// seconds. A window without a percentage or reset time is dropped, and an event +// with no usable window returns nil. +func ParseMetaSubscriptionUsageHeaders(payload []byte) http.Header { + if len(payload) == 0 || gjson.GetBytes(payload, "type").String() != MetaSubscriptionUsageEventType { + return nil + } + return metaSubscriptionUsageHeaders(gjson.GetBytes(payload, "subscription")) +} + +// ParseMetaSubscriptionUsageSSE finds the last response.subscription_usage +// event in a buffered Responses event stream. +func ParseMetaSubscriptionUsageSSE(body []byte) http.Header { + var headers http.Header + for _, line := range bytes.Split(body, []byte("\n")) { + line = bytes.TrimSpace(line) + if !bytes.HasPrefix(line, []byte("data:")) { + continue + } + if parsed := ParseMetaSubscriptionUsageHeaders(bytes.TrimSpace(line[len("data:"):])); parsed != nil { + headers = parsed + } + } + return headers +} + +// ObserveMetaSubscriptionUsage records a subscription usage event on the request +// context. The conductor stores it on the credential when the request finishes. +func ObserveMetaSubscriptionUsage(ctx context.Context, payload []byte) { + logging.MergeResponseHeaders(ctx, ParseMetaSubscriptionUsageHeaders(payload)) +} + +// ObserveMetaSubscriptionUsageSSE records the subscription usage carried by a +// buffered Responses event stream. +func ObserveMetaSubscriptionUsageSSE(ctx context.Context, body []byte) { + logging.MergeResponseHeaders(ctx, ParseMetaSubscriptionUsageSSE(body)) +} + +func metaSubscriptionUsageHeaders(subscription gjson.Result) http.Header { + if !subscription.IsObject() { + return nil + } + headers := make(http.Header) + window := subscription.Get("window") + minutes := window.Get("window_duration_mins") + if used, reset, ok := metaQuotaWindow(window); ok && minutes.Type == gjson.Number && minutes.Int() > 0 { + headers.Set(MetaQuotaWindowUsedHeader, used) + headers.Set(MetaQuotaWindowMinutesHeader, strconv.FormatInt(minutes.Int(), 10)) + headers.Set(MetaQuotaWindowResetHeader, reset) + } + if used, reset, ok := metaQuotaWindow(subscription.Get("weekly")); ok { + headers.Set(MetaQuotaWeeklyUsedHeader, used) + headers.Set(MetaQuotaWeeklyResetHeader, reset) + } + if len(headers) == 0 { + return nil + } + // The tier is an opaque plan id. It only labels the snapshot, so a value that + // could forge a request-log line is dropped rather than failing the event. + if tier := strings.TrimSpace(subscription.Get("tier").String()); tier != "" && len(tier) <= 128 && !strings.ContainsAny(tier, "\r\n") { + headers.Set(MetaQuotaTierHeader, tier) + } + return headers +} + +func metaQuotaWindow(window gjson.Result) (used, reset string, ok bool) { + if !window.IsObject() { + return "", "", false + } + usedPercent := window.Get("used_percent") + resetsAt := window.Get("resets_at") + if usedPercent.Type != gjson.Number || usedPercent.Float() < 0 || resetsAt.Type != gjson.Number || resetsAt.Int() <= 0 { + return "", "", false + } + return strings.TrimSpace(usedPercent.Raw), strconv.FormatInt(resetsAt.Int(), 10), true +} diff --git a/internal/runtime/executor/helps/meta_quota_test.go b/internal/runtime/executor/helps/meta_quota_test.go new file mode 100644 index 00000000000..147f896f995 --- /dev/null +++ b/internal/runtime/executor/helps/meta_quota_test.go @@ -0,0 +1,93 @@ +package helps + +import ( + "context" + "net/http" + "testing" + + "github.com/router-for-me/CLIProxyAPI/v8/internal/logging" +) + +// Recorded from api.meta.ai on 2026-10-08; only the tier id is replaced. +const metaSubscriptionUsageEvent = `{"subscription":{"tier":"tier-1","weekly":{"resets_at":1791763200,"used_percent":5},"window":{"resets_at":1791465557,"used_percent":0,"window_duration_mins":300}},"type":"response.subscription_usage"}` + +func TestForkParseMetaSubscriptionUsageHeaders(t *testing.T) { + headers := ParseMetaSubscriptionUsageHeaders([]byte(metaSubscriptionUsageEvent)) + want := map[string]string{ + MetaQuotaTierHeader: "tier-1", + MetaQuotaWindowUsedHeader: "0", + MetaQuotaWindowMinutesHeader: "300", + MetaQuotaWindowResetHeader: "1791465557", + MetaQuotaWeeklyUsedHeader: "5", + MetaQuotaWeeklyResetHeader: "1791763200", + } + if len(headers) != len(want) { + t.Fatalf("headers = %#v, want %d entries", headers, len(want)) + } + for name, value := range want { + if got := headers.Get(name); got != value { + t.Fatalf("%s = %q, want %q", name, got, value) + } + } +} + +func TestForkParseMetaSubscriptionUsageKeepsOverQuotaAndDropsInvalidWindows(t *testing.T) { + headers := ParseMetaSubscriptionUsageHeaders([]byte(`{"type":"response.subscription_usage","subscription":{` + + `"window":{"used_percent":104.5,"window_duration_mins":300,"resets_at":1791465557},` + + `"weekly":{"used_percent":-1,"resets_at":1791763200}}}`)) + if got := headers.Get(MetaQuotaWindowUsedHeader); got != "104.5" { + t.Fatalf("window used = %q, want the over-quota value verbatim", got) + } + if headers.Get(MetaQuotaWeeklyUsedHeader) != "" || headers.Get(MetaQuotaWeeklyResetHeader) != "" { + t.Fatalf("negative weekly percentage was kept: %#v", headers) + } + if headers.Get(MetaQuotaTierHeader) != "" { + t.Fatalf("missing tier produced a header: %#v", headers) + } + + for _, payload := range []string{ + ``, + `{"type":"response.completed","subscription":{"window":{"used_percent":1,"window_duration_mins":300,"resets_at":1}}}`, + `{"type":"response.subscription_usage"}`, + `{"type":"response.subscription_usage","subscription":{"window":{"used_percent":1,"window_duration_mins":0,"resets_at":1791465557}}}`, + `{"type":"response.subscription_usage","subscription":{"weekly":{"used_percent":"5","resets_at":1791763200}}}`, + `{"type":"response.subscription_usage","subscription":{"tier":"t","weekly":{"used_percent":5}}}`, + } { + if headers := ParseMetaSubscriptionUsageHeaders([]byte(payload)); headers != nil { + t.Fatalf("payload %s produced %#v, want nil", payload, headers) + } + } +} + +func TestForkParseMetaSubscriptionUsageDropsUnsafeTier(t *testing.T) { + headers := ParseMetaSubscriptionUsageHeaders([]byte(`{"type":"response.subscription_usage","subscription":{"tier":"a\nb","weekly":{"used_percent":5,"resets_at":1791763200}}}`)) + if headers.Get(MetaQuotaWeeklyUsedHeader) != "5" || headers.Get(MetaQuotaTierHeader) != "" { + t.Fatalf("headers = %#v, want the weekly window without the tier", headers) + } +} + +func TestForkParseMetaSubscriptionUsageSSEUsesTheLastEvent(t *testing.T) { + body := "event: response.completed\n" + + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\"}}\n\n" + + "event: response.subscription_usage\n" + + "data: {\"type\":\"response.subscription_usage\",\"subscription\":{\"weekly\":{\"used_percent\":5,\"resets_at\":1791763200}}}\n\n" + + "event: response.subscription_usage\n" + + "data: " + metaSubscriptionUsageEvent + "\n\n" + headers := ParseMetaSubscriptionUsageSSE([]byte(body)) + if headers.Get(MetaQuotaWindowMinutesHeader) != "300" || headers.Get(MetaQuotaTierHeader) != "tier-1" { + t.Fatalf("headers = %#v, want the last event", headers) + } + if ParseMetaSubscriptionUsageSSE([]byte("data: {\"type\":\"response.completed\"}\n\n")) != nil { + t.Fatal("a stream without subscription usage produced headers") + } +} + +func TestForkObserveMetaSubscriptionUsageMergesIntoResponseHeaders(t *testing.T) { + ctx := logging.WithResponseHeadersHolder(context.Background()) + logging.SetResponseHeaders(ctx, http.Header{"X-Request-Id": []string{"req-1"}}) + ObserveMetaSubscriptionUsage(ctx, []byte(metaSubscriptionUsageEvent)) + headers := logging.GetResponseHeaders(ctx) + if headers.Get("X-Request-Id") != "req-1" || headers.Get(MetaQuotaWeeklyUsedHeader) != "5" { + t.Fatalf("response headers = %#v", headers) + } +} diff --git a/internal/runtime/executor/meta_executor_execute.go b/internal/runtime/executor/meta_executor_execute.go index 516ce7e4cc8..e45ce7d0cda 100644 --- a/internal/runtime/executor/meta_executor_execute.go +++ b/internal/runtime/executor/meta_executor_execute.go @@ -142,6 +142,9 @@ func (e *MetaExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), data)) return resp, wrapMetaUpstreamError(httpResp.StatusCode, data) } + // Meta sends subscription usage after response.completed, which is where the + // translation below stops reading. + helps.ObserveMetaSubscriptionUsageSSE(ctx, data) out, errCompleted := e.translateMetaCompleted(ctx, req, prepared, data) if errCompleted != nil { diff --git a/internal/runtime/executor/meta_executor_quota_test.go b/internal/runtime/executor/meta_executor_quota_test.go new file mode 100644 index 00000000000..60cefed746e --- /dev/null +++ b/internal/runtime/executor/meta_executor_quota_test.go @@ -0,0 +1,94 @@ +package executor + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/router-for-me/CLIProxyAPI/v8/internal/config" + "github.com/router-for-me/CLIProxyAPI/v8/internal/logging" + "github.com/router-for-me/CLIProxyAPI/v8/internal/runtime/executor/helps" + cliproxyauth "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/auth" + cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/executor" + sdktranslator "github.com/router-for-me/CLIProxyAPI/v8/sdk/translator" +) + +// Meta sends the subscription usage after response.completed (recorded 2026-10-08). +const metaSubscriptionUsageSSE = "event: response.output_item.done\n" + + "data: {\"type\":\"response.output_item.done\",\"output_index\":0,\"item\":{\"id\":\"item_0\",\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"OK\"}]}}\n\n" + + "event: response.completed\n" + + "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_1\",\"status\":\"completed\",\"model\":\"muse-spark-1.3\",\"usage\":{\"input_tokens\":3,\"output_tokens\":1,\"total_tokens\":4}}}\n\n" + + "event: response.subscription_usage\n" + + "data: {\"subscription\":{\"tier\":\"tier-1\",\"weekly\":{\"resets_at\":1791763200,\"used_percent\":5},\"window\":{\"resets_at\":1791465557,\"used_percent\":12,\"window_duration_mins\":300}},\"type\":\"response.subscription_usage\"}\n\n" + +func newMetaSubscriptionUsageServer(t *testing.T) *cliproxyauth.Auth { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(metaSubscriptionUsageSSE)) + })) + t.Cleanup(server.Close) + return &cliproxyauth.Auth{ + Provider: "meta", + Attributes: map[string]string{"api_key": "meta-token", "base_url": server.URL}, + } +} + +func assertMetaSubscriptionUsageObserved(t *testing.T, ctx context.Context) { + t.Helper() + headers := logging.GetResponseHeaders(ctx) + if headers.Get(helps.MetaQuotaWindowUsedHeader) != "12" || + headers.Get(helps.MetaQuotaWindowMinutesHeader) != "300" || + headers.Get(helps.MetaQuotaWeeklyResetHeader) != "1791763200" || + headers.Get(helps.MetaQuotaTierHeader) != "tier-1" { + t.Fatalf("observed response headers = %#v, want the Meta subscription usage", headers) + } +} + +func TestForkMetaExecutorStreamRecordsAndForwardsSubscriptionUsage(t *testing.T) { + auth := newMetaSubscriptionUsageServer(t) + ctx := logging.WithResponseHeadersHolder(context.Background()) + result, err := NewMetaExecutor(&config.Config{}).ExecuteStream(ctx, auth, cliproxyexecutor.Request{ + Model: "muse-spark-1.3", + Payload: []byte(`{"model":"muse-spark-1.3","input":"hello","stream":true}`), + }, cliproxyexecutor.Options{ + Stream: true, + SourceFormat: sdktranslator.FromString("openai-response"), + }) + if err != nil { + t.Fatalf("ExecuteStream() error = %v", err) + } + var forwarded strings.Builder + for chunk := range result.Chunks { + if chunk.Err != nil { + t.Fatalf("stream chunk error = %v", chunk.Err) + } + forwarded.Write(chunk.Payload) + forwarded.WriteByte('\n') + } + // The Muse CLI reads the event after response.completed, so the client must still get it. + if !strings.Contains(forwarded.String(), `"type":"response.subscription_usage"`) { + t.Fatalf("forwarded stream lacks the subscription usage event:\n%s", forwarded.String()) + } + assertMetaSubscriptionUsageObserved(t, ctx) +} + +func TestForkMetaExecutorExecuteRecordsSubscriptionUsage(t *testing.T) { + auth := newMetaSubscriptionUsageServer(t) + ctx := logging.WithResponseHeadersHolder(context.Background()) + resp, err := NewMetaExecutor(&config.Config{}).Execute(ctx, auth, cliproxyexecutor.Request{ + Model: "muse-spark-1.3", + Payload: []byte(`{"model":"muse-spark-1.3","input":"hello"}`), + }, cliproxyexecutor.Options{ + SourceFormat: sdktranslator.FromString("openai-response"), + }) + if err != nil { + t.Fatalf("Execute() error = %v", err) + } + if !strings.Contains(string(resp.Payload), "resp_1") { + t.Fatalf("Execute() payload = %s, want the completed response", resp.Payload) + } + assertMetaSubscriptionUsageObserved(t, ctx) +} diff --git a/internal/runtime/executor/meta_executor_stream.go b/internal/runtime/executor/meta_executor_stream.go index 8d5f54626da..6efc7a4cb82 100644 --- a/internal/runtime/executor/meta_executor_stream.go +++ b/internal/runtime/executor/meta_executor_stream.go @@ -146,6 +146,8 @@ func (e *MetaExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Aut streamUsage.Observe(detail, true) } eventData = patchCodexCompletedOutput(eventData, outputItemsByIndex, outputItemsFallback) + case helps.MetaSubscriptionUsageEventType: + helps.ObserveMetaSubscriptionUsage(ctx, eventData) } if !emitTranslatedLine(append([]byte("data: "), eventData...)) { return diff --git a/sdk/api/handlers/openai/openai_responses_handlers.go b/sdk/api/handlers/openai/openai_responses_handlers.go index 977aa6cf8c0..0240da6cf1f 100644 --- a/sdk/api/handlers/openai/openai_responses_handlers.go +++ b/sdk/api/handlers/openai/openai_responses_handlers.go @@ -62,15 +62,14 @@ type responsesSSEFramer struct { } func (f *responsesSSEFramer) WriteChunk(w io.Writer, chunk []byte) { - if len(chunk) == 0 || f.terminalEvent != "" { + // Frames after an error terminal are dropped outright. After any other + // terminal they are still framed so writeFrame can pass trailing events. + if len(chunk) == 0 || f.terminalError != nil { return } if responsesSSEStartsNewDataFrame(f.pending, chunk) { f.writeFrame(w, f.pending) f.pending = f.pending[:0] - if f.terminalEvent != "" { - return - } } if responsesSSENeedsLineBreak(f.pending, chunk) { f.pending = append(f.pending, '\n') @@ -84,7 +83,7 @@ func (f *responsesSSEFramer) WriteChunk(w io.Writer, chunk []byte) { f.writeFrame(w, f.pending[:frameLen]) copy(f.pending, f.pending[frameLen:]) f.pending = f.pending[:len(f.pending)-frameLen] - if f.terminalEvent != "" { + if f.terminalError != nil { f.pending = f.pending[:0] return } @@ -101,7 +100,7 @@ func (f *responsesSSEFramer) WriteChunk(w io.Writer, chunk []byte) { } func (f *responsesSSEFramer) Flush(w io.Writer) { - if len(f.pending) == 0 || f.terminalEvent != "" { + if len(f.pending) == 0 || f.terminalError != nil { return } if len(bytes.TrimSpace(f.pending)) == 0 { @@ -117,9 +116,30 @@ func (f *responsesSSEFramer) Flush(w io.Writer) { } func (f *responsesSSEFramer) writeFrame(w io.Writer, frame []byte) { + if f.terminalEvent != "" { + // The response is over for the client; only trailing events still belong to it. + if responsesSSETrailingFrame(frame) { + writeResponsesSSEChunk(w, frame) + } + return + } writeResponsesSSEChunk(w, f.repairFrame(frame)) } +// responsesSubscriptionUsageEvent is the event Meta sends after response.completed +// with the account's subscription usage. The Muse CLI reads it to show its quota. +const responsesSubscriptionUsageEvent = "response.subscription_usage" + +// responsesSSETrailingFrame reports whether a frame that arrives after the +// terminal event is still forwarded. +func responsesSSETrailingFrame(frame []byte) bool { + if responsesSSEEventName(frame) == responsesSubscriptionUsageEvent { + return true + } + payload, ok := responsesSSEDataPayload(frame) + return ok && json.Valid(payload) && gjson.GetBytes(payload, "type").String() == responsesSubscriptionUsageEvent +} + func (f *responsesSSEFramer) shouldFilterPrivateEvent(streamEvent, payloadType string) bool { check := func(name string) bool { name = strings.TrimSpace(name) diff --git a/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go b/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go index 58d1349a59a..0f879e031e1 100644 --- a/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go +++ b/sdk/api/handlers/openai/openai_responses_handlers_stream_test.go @@ -68,6 +68,62 @@ func TestResponsesSSEFramerFlushesMultilineDataWithoutDelimiter(t *testing.T) { } } +func TestForkResponsesSSEFramerForwardsSubscriptionUsageAfterCompleted(t *testing.T) { + var output bytes.Buffer + framer := &responsesSSEFramer{} + // The Meta executor forwards the stream one line at a time. + for _, line := range []string{ + "event: response.completed", + `data: {"type":"response.completed","response":{"id":"resp-1","status":"completed"}}`, + "", + "event: response.output_text.delta", + `data: {"type":"response.output_text.delta","delta":"late"}`, + "", + "event: response.subscription_usage", + `data: {"type":"response.subscription_usage","subscription":{"weekly":{"used_percent":5,"resets_at":1791763200}}}`, + "", + } { + framer.WriteChunk(&output, []byte(line+"\n")) + } + framer.Flush(&output) + + got := output.String() + if framer.terminalEvent != "response.completed" { + t.Fatalf("terminal event = %q, want response.completed", framer.terminalEvent) + } + if strings.Count(got, "event: response.subscription_usage") != 1 || !strings.Contains(got, `"used_percent":5`) { + t.Fatalf("subscription usage after response.completed was not forwarded once: %q", got) + } + if strings.Contains(got, "late") { + t.Fatalf("an ordinary event after response.completed was forwarded: %q", got) + } + if strings.Index(got, "response.completed") > strings.Index(got, "response.subscription_usage") { + t.Fatalf("subscription usage was written before the terminal event: %q", got) + } +} + +func TestForkResponsesSSEFramerForwardsUndelimitedSubscriptionUsageOnFlush(t *testing.T) { + var output bytes.Buffer + framer := &responsesSSEFramer{} + framer.WriteChunk(&output, []byte("event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp-1\"}}\n\n")) + framer.WriteChunk(&output, []byte("data: {\"type\":\"response.subscription_usage\",\"subscription\":{}}")) + framer.Flush(&output) + if !strings.Contains(output.String(), "response.subscription_usage") { + t.Fatalf("undelimited subscription usage frame was dropped on flush: %q", output.String()) + } +} + +func TestForkResponsesSSEFramerDropsSubscriptionUsageAfterErrorTerminal(t *testing.T) { + var output bytes.Buffer + framer := &responsesSSEFramer{} + framer.WriteChunk(&output, []byte("event: error\ndata: {\"type\":\"error\",\"message\":\"failed\"}\n\n")) + framer.WriteChunk(&output, []byte("event: response.subscription_usage\ndata: {\"type\":\"response.subscription_usage\",\"subscription\":{}}\n\n")) + framer.Flush(&output) + if strings.Contains(output.String(), "response.subscription_usage") { + t.Fatalf("frame after an error terminal was forwarded: %q", output.String()) + } +} + func TestResponsesSSEFramerUsesPayloadErrorOverCompletedEvent(t *testing.T) { var output bytes.Buffer framer := &responsesSSEFramer{failureEvent: "response.failed"} diff --git a/sdk/cliproxy/auth/conductor_lifecycle.go b/sdk/cliproxy/auth/conductor_lifecycle.go index 6ce03dc3d67..50a8e2ba390 100644 --- a/sdk/cliproxy/auth/conductor_lifecycle.go +++ b/sdk/cliproxy/auth/conductor_lifecycle.go @@ -230,6 +230,9 @@ func (m *Manager) updateInternal(ctx context.Context, base, auth *Auth, mode upd if len(auth.ModelStates) == 0 && len(existing.ModelStates) > 0 { auth.ModelStates = existing.ModelStates } + // Credential files do not store passive quota observations, so reloading one + // must not erase the live snapshot. The newest observation wins. + auth.Quota = mergeQuotaObservation(auth.Quota, existing.Quota) credChanged := CredentialsChanged(existing, auth) if credChanged { if hasUnauthorizedAuthFailure(existing) || (auth.LastError != nil && isUnauthorizedError(auth.LastError)) { diff --git a/sdk/cliproxy/auth/quota_signals.go b/sdk/cliproxy/auth/quota_signals.go index fea368e5319..c7ad9f76b8c 100644 --- a/sdk/cliproxy/auth/quota_signals.go +++ b/sdk/cliproxy/auth/quota_signals.go @@ -12,11 +12,21 @@ const ( maxQuotaSignalValue = 512 ) +// metaQuotaSignalHeaders lists the lower-cased Meta subscription usage signals. +var metaQuotaSignalHeaders = map[string]bool{ + "x-meta-tier": true, + "x-meta-window-used-percent": true, + "x-meta-window-minutes": true, + "x-meta-window-reset-at": true, + "x-meta-weekly-used-percent": true, + "x-meta-weekly-reset-at": true, +} + // ProviderSupportsQuotaObservation reports whether the named provider emits a // passive credential-level quota snapshot understood by collectQuotaSignals. func ProviderSupportsQuotaObservation(provider string) bool { switch strings.ToLower(strings.TrimSpace(provider)) { - case "claude", "codex", "devin": + case "claude", "codex", "devin", "meta": return true default: return false @@ -176,6 +186,11 @@ func isQuotaSignalHeaderForProvider(provider, name string) bool { if strings.HasPrefix(name, "anthropic-ratelimit-unified-") { return provider == "claude" } + if strings.HasPrefix(name, "x-meta-") { + // Meta reports usage in a response.subscription_usage stream event, which + // the executor converts to these names (helps.ParseMetaSubscriptionUsageHeaders). + return provider == "meta" && metaQuotaSignalHeaders[name] + } if strings.HasPrefix(name, "x-ratelimit-") { // Observed Codex responses do not carry x-ratelimit-* headers; the only // upstream seen emitting them is Grok, which is excluded from quota diff --git a/sdk/cliproxy/auth/quota_signals_test.go b/sdk/cliproxy/auth/quota_signals_test.go index 2a2b08dd23b..b3b191c822a 100644 --- a/sdk/cliproxy/auth/quota_signals_test.go +++ b/sdk/cliproxy/auth/quota_signals_test.go @@ -150,6 +150,35 @@ func TestQuotaStateObserveResponseHeadersDropsKimiGrokAndAntigravitySignals(t *t } } +func TestForkQuotaSignalsKeepMetaSubscriptionUsageForMetaOnly(t *testing.T) { + headers := http.Header{ + "X-Meta-Tier": []string{"tier-1"}, + "X-Meta-Window-Used-Percent": []string{"12"}, + "X-Meta-Window-Minutes": []string{"300"}, + "X-Meta-Window-Reset-At": []string{"1791465557"}, + "X-Meta-Weekly-Used-Percent": []string{"5"}, + "X-Meta-Weekly-Reset-At": []string{"1791763200"}, + "X-Meta-Ai-Gateway-Error": []string{"not a quota signal"}, + "X-Request-Id": []string{"req-1"}, + "X-Codex-Primary-Used-Percent": []string{"50"}, + } + var meta QuotaState + if !meta.ObserveResponseHeadersForProvider("meta", headers, time.Unix(1791447762, 0)) { + t.Fatal("meta subscription usage was not observed") + } + if len(meta.Signals) != 6 || meta.Signals["X-Meta-Weekly-Used-Percent"] != "5" || meta.Signals["X-Meta-Tier"] != "tier-1" { + t.Fatalf("meta signals = %#v, want only the six subscription usage signals", meta.Signals) + } + + var codex QuotaState + codex.ObserveResponseHeadersForProvider("codex", headers, time.Now()) + for key := range codex.Signals { + if strings.HasPrefix(strings.ToLower(key), "x-meta-") { + t.Fatalf("codex kept Meta signal %s: %#v", key, codex.Signals) + } + } +} + func TestCooldownEqualityIgnoresObservationSignals(t *testing.T) { base := QuotaState{ Exceeded: true, @@ -260,6 +289,27 @@ func TestResetModelStatePreservesObservationSignals(t *testing.T) { } } +func TestForkUpdateFromCredentialFileKeepsObservation(t *testing.T) { + manager := NewManager(nil, nil, nil) + if _, errRegister := manager.Register(context.Background(), &Auth{ID: "meta-reload", Provider: "meta"}); errRegister != nil { + t.Fatal(errRegister) + } + if !manager.ObserveQuotaSignals("meta-reload", "meta", http.Header{ + "X-Meta-Weekly-Used-Percent": []string{"5"}, + "X-Meta-Weekly-Reset-At": []string{"1791763200"}, + }, time.Unix(1791447762, 0)) { + t.Fatal("observation was not recorded") + } + // The file watcher reloads a saved credential without any observation in it. + if _, errUpdate := manager.Update(context.Background(), &Auth{ID: "meta-reload", Provider: "meta"}); errUpdate != nil { + t.Fatal(errUpdate) + } + reloaded, ok := manager.GetByID("meta-reload") + if !ok || reloaded.Quota.Signals["X-Meta-Weekly-Used-Percent"] != "5" || !reloaded.Quota.ObservedAt.Equal(time.Unix(1791447762, 0)) { + t.Fatalf("reloading the credential erased its observation: %#v", reloaded.Quota) + } +} + func TestMergeModelStateKeepsNewestObservationSnapshot(t *testing.T) { target := &ModelState{UpdatedAt: time.Unix(20, 0), Quota: QuotaState{ ObservedAt: time.Unix(20, 0), Signals: map[string]string{"X-Codex-Plan-Type": "pro"}, diff --git a/sdk/cliproxy/auth/quota_window.go b/sdk/cliproxy/auth/quota_window.go index dca739db372..4a441b9e3ef 100644 --- a/sdk/cliproxy/auth/quota_window.go +++ b/sdk/cliproxy/auth/quota_window.go @@ -44,6 +44,8 @@ func QuotaWindows(provider string, quota QuotaState) []QuotaWindow { windows = claudeQuotaWindows(quota.Signals) case "codex": windows = codexQuotaWindows(quota.Signals, quota.ObservedAt) + case "meta": + windows = metaQuotaWindows(quota.Signals) default: return nil } @@ -135,6 +137,39 @@ func codexQuotaWindows(signals map[string]string, observedAt time.Time) []QuotaW return windows } +// metaQuotaWindows reads the Meta subscription windows recorded from the +// response.subscription_usage stream event. Meta does not report the weekly +// window's length, so it is the rolling week the event names it after. +func metaQuotaWindows(signals map[string]string) []QuotaWindow { + lower := make(map[string]string, len(signals)) + for key, raw := range signals { + lower[strings.ToLower(strings.TrimSpace(key))] = strings.TrimSpace(raw) + } + windows := make([]QuotaWindow, 0, 2) + for _, name := range []string{"window", "weekly"} { + prefix := "x-meta-" + name + "-" + window := QuotaWindow{Name: name, Used: -1} + if name == "weekly" { + window.Duration = 7 * 24 * time.Hour + } else if minutes, errMinutes := strconv.ParseFloat(lower["x-meta-window-minutes"], 64); errMinutes == nil && minutes > 0 && !math.IsInf(minutes, 0) { + window.Duration = time.Duration(minutes * float64(time.Minute)) + } + found := false + if percent, ok := parseQuotaFraction(lower[prefix+"used-percent"]); ok { + window.Used = percent / 100 + found = true + } + if resetAt, ok := parseQuotaResetInstant(lower[prefix+"reset-at"]); ok { + window.ResetAt = resetAt + found = true + } + if found { + windows = append(windows, window) + } + } + return windows +} + // quotaWindowLabelDuration reads the duration from a credential-wide window // label such as "5h" or "7d". Scoped labels such as "7d_oi" are rejected. func quotaWindowLabelDuration(label string) (time.Duration, bool) { diff --git a/sdk/cliproxy/auth/selector_soonest_reset_test.go b/sdk/cliproxy/auth/selector_soonest_reset_test.go index c5889c4b22b..31e7a820d06 100644 --- a/sdk/cliproxy/auth/selector_soonest_reset_test.go +++ b/sdk/cliproxy/auth/selector_soonest_reset_test.go @@ -108,6 +108,32 @@ func TestForkQuotaWindowsCodexUsesAbsoluteResetBeforeRelative(t *testing.T) { } } +func TestForkQuotaWindowsMetaReadsSubscriptionWindows(t *testing.T) { + windowReset := time.Unix(1_791_465_557, 0) + weeklyReset := time.Unix(1_791_763_200, 0) + windows := QuotaWindows("meta", QuotaState{ + ObservedAt: time.Unix(1_791_447_762, 0), + Signals: map[string]string{ + "X-Meta-Tier": "tier-1", + "X-Meta-Window-Used-Percent": "104", + "X-Meta-Window-Minutes": "300", + "X-Meta-Window-Reset-At": strconv.FormatInt(windowReset.Unix(), 10), + "X-Meta-Weekly-Used-Percent": "5", + "X-Meta-Weekly-Reset-At": strconv.FormatInt(weeklyReset.Unix(), 10), + }, + }) + if len(windows) != 2 { + t.Fatalf("windows = %+v, want window and weekly", windows) + } + short, long := windows[0], windows[1] + if short.Name != "window" || short.Duration != 5*time.Hour || short.Used != 1.04 || !short.ResetAt.Equal(windowReset) { + t.Fatalf("short window = %+v", short) + } + if long.Name != "weekly" || long.Duration != 7*24*time.Hour || long.Used != 0.05 || !long.ResetAt.Equal(weeklyReset) { + t.Fatalf("long window = %+v", long) + } +} + func TestForkQuotaWindowsUnknownProviderOrEmptySignals(t *testing.T) { if got := QuotaWindows("claude", QuotaState{}); got != nil { t.Fatalf("empty signals = %+v, want nil", got)