From 7a00484f97a7a3edbfc1743193067ac3ade1bcf0 Mon Sep 17 00:00:00 2001 From: Nicholas Tindle Date: Thu, 8 Oct 2026 04:31:42 -0500 Subject: [PATCH] feat(meta): record Meta subscription usage and pass it to Muse Meta has no usage endpoint. It reports the account's session window and weekly usage in a response.subscription_usage event after response.completed on every Responses stream. The proxy dropped that event: the Responses framer stops at the terminal event, so the Muse CLI never saw its quota through the proxy, and nothing recorded it, so the management API had no quota for Meta credentials. - The Meta executor records the event as X-Meta-* quota signals, which the management auth-files list exposes under quota and soonest-reset reads. - The Responses framer still forwards the event after the terminal event; every other late frame is dropped as before. - Reloading a credential file keeps the passive quota observation. The file never stores it, so a reload used to erase it for every provider. Co-Authored-By: Claude Opus 5.5 --- .github/scripts/fork-check.sh | 6 + FORK.md | 29 +++++ internal/runtime/executor/helps/meta_quota.go | 113 ++++++++++++++++++ .../runtime/executor/helps/meta_quota_test.go | 93 ++++++++++++++ .../runtime/executor/meta_executor_execute.go | 3 + .../executor/meta_executor_quota_test.go | 94 +++++++++++++++ .../runtime/executor/meta_executor_stream.go | 2 + .../openai/openai_responses_handlers.go | 32 ++++- .../openai_responses_handlers_stream_test.go | 56 +++++++++ sdk/cliproxy/auth/conductor_lifecycle.go | 3 + sdk/cliproxy/auth/quota_signals.go | 17 ++- sdk/cliproxy/auth/quota_signals_test.go | 50 ++++++++ sdk/cliproxy/auth/quota_window.go | 35 ++++++ .../auth/selector_soonest_reset_test.go | 26 ++++ 14 files changed, 552 insertions(+), 7 deletions(-) create mode 100644 internal/runtime/executor/helps/meta_quota.go create mode 100644 internal/runtime/executor/helps/meta_quota_test.go create mode 100644 internal/runtime/executor/meta_executor_quota_test.go 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)