diff --git a/cmd/interceptor/main.go b/cmd/interceptor/main.go index abcb816..0571013 100644 --- a/cmd/interceptor/main.go +++ b/cmd/interceptor/main.go @@ -40,7 +40,7 @@ func main() { // R3 trusted-header registry: parse PHOEBE_TRUSTED_HEADERS (rendered // from the phoebe chart's ConfigMap) into the active set the parser's - // envelope reads resolve through. The pinned 13 already govern from + // envelope reads resolve through. The pinned 14 already govern from // package init; this engages the runtime config once, at startup, and // warns loudly if the chart render was empty/malformed. identity.LoadTrustedHeaders(log) diff --git a/internal/admission/admission.go b/internal/admission/admission.go index 97ad289..78bde8e 100644 --- a/internal/admission/admission.go +++ b/internal/admission/admission.go @@ -82,6 +82,24 @@ type Request struct { Adapter bool OrganizationLimits RateLimits OwnerLimits RateLimits + // GroupScopes is the parsed membership-aware group quota envelope + // (X-Saturn-Group-Scopes, ruled 2026-10-07): one entry per group the + // caller belongs to that carries limits. Each entry's Limits enforce the + // four per-minute rate windows as a contract scope (the same Lua/counter + // machinery as the contract_owner scope, settling through the same + // lease/CompleteUsage path). Empty when the envelope carried no group + // (the common case). + GroupScopes []GroupScope +} + +// GroupScope is one group's quota contract within a Request: the group's id +// and its four per-minute rate limits (R4 sentinels: nil unlimited, 0 zero +// cap). The envelope's fifth field is reserved for the deferred monthly spend +// cap; the proxy rejects any envelope that carries a non-empty value there, +// so a GroupScope never holds one in this build. +type GroupScope struct { + GroupID string + Limits RateLimits } // RateLimits is the authenticated customer contract. R4 sentinel semantics @@ -199,6 +217,14 @@ func (a *RedisAdmitter) Admit(ctx context.Context, req Request) (*Lease, error) if req.OwnerLimits.Any() && req.Owner == "" { return nil, fmt.Errorf("%w: missing trusted owner identity", ErrInvalidIdentity) } + for _, gs := range req.GroupScopes { + // A group scope carrying any limit must name its group: an anonymous + // quota is not a quota, it is a broken envelope (fail closed, like + // every other structural violation). + if gs.GroupID == "" && gs.Limits.Any() { + return nil, fmt.Errorf("%w: group scope carries limits without a group id", ErrInvalidIdentity) + } + } if req.PromptBytes < 0 || req.EstimatedInputTokens <= 0 || req.ReservedOutputTokens <= 0 { return nil, &Rejected{Scope: "request", Dimension: "work estimate", RetryAfter: time.Second} } @@ -385,6 +411,17 @@ func (a *RedisAdmitter) scopes(r Request) []scope { if r.OwnerLimits.Any() { out = append(out, makeContractScope("contract_owner", r.Owner, r.OwnerLimits)) } + // Group rate scopes: one contract scope per group that carries any rate + // limit, same Lua/counter machinery and wireLimit sentinels as the org/ + // owner contract scopes, settling through the same lease/CompleteUsage + // path (the lease record holds this scope like any other, and finishScript + // charges its windows). The scope NAME carries the group id so a rejection + // names the team; the scope id digests it, so groups never collide. + for _, gs := range r.GroupScopes { + if gs.Limits.Any() { + out = append(out, makeContractScope("group:"+gs.GroupID, gs.GroupID, gs.Limits)) + } + } return out } diff --git a/internal/admission/group_test.go b/internal/admission/group_test.go new file mode 100644 index 0000000..63ca909 --- /dev/null +++ b/internal/admission/group_test.go @@ -0,0 +1,227 @@ +package admission + +import ( + "context" + "errors" + "testing" + + "github.com/saturncloud/phoebe/internal/config" +) + +const testGID = "a1b2c3d4e5f60718293a4b5c6d7e8f90" + +// groupScope builds one GroupScope. rate of nil leaves every rate unset +// (unlimited); a non-nil rate installs all four dimensions to the same value. +func groupScope(gid string, rate *int64) GroupScope { + gs := GroupScope{GroupID: gid} + if rate != nil { + gs.Limits = RateLimits{Requests: rate, TotalPromptTokens: rate, + UncachedPromptTokens: rate, GeneratedTokens: rate} + } + return gs +} + +func groupRequest(org, model string, scopes ...GroupScope) Request { + r := request(org, model) + r.GroupScopes = scopes + return r +} + +// TestGroupRateLimitsEnforcedLikeContractScope: a group's per-minute windows +// are the same Lua/counter machinery as the contract scopes — a burst up to +// the cap admits, the next request is a contractual rejection (the proxy maps +// that to 429 + Retry-After). +func TestGroupRateLimitsEnforcedLikeContractScope(t *testing.T) { + a, _ := testAdmitter(t, config.AdmissionSettings{}) + // Only the requests window is capped, so the request's own estimates + // (the helper's 10-in/20-out shape) can never trip the unsatisfiable + // check: the third request is rejected by the window counter itself. + reqCap := int64(2) + scope := GroupScope{GroupID: testGID, Limits: RateLimits{Requests: &reqCap}} + + for i := 0; i < 2; i++ { + lease, err := a.Admit(context.Background(), groupRequest("org-g", "m", scope)) + if err != nil { + t.Fatalf("admit %d: %v, want admitted", i+1, err) + } + _ = lease.Complete(context.Background(), 0) + } + _, err := a.Admit(context.Background(), groupRequest("org-g", "m", scope)) + var rejected *Rejected + if !errors.As(err, &rejected) { + t.Fatalf("third admit: %v, want a contract rejection", err) + } + if !rejected.Contractual || rejected.RetryAfter <= 0 { + t.Fatalf("rejection = %+v, want Contractual with a Retry-After (429 mapping)", rejected) + } + if rejected.Scope != "group:"+testGID { + t.Fatalf("rejection scope = %q, want the group id named", rejected.Scope) + } +} + +// TestGroupTotalPromptTokensWindowEnforced: a group scope may cap only the +// total-prompt window, leaving the other rate fields nil (unlimited). Unlike +// the request-count window, token windows charge actual usage at settlement, +// so each request settles its 10-token input estimate: two requests fill a +// 20-token cap and the next reservation is a contractual rejection naming +// the group scope. +func TestGroupTotalPromptTokensWindowEnforced(t *testing.T) { + a, _ := testAdmitter(t, config.AdmissionSettings{}) + promptCap := int64(20) + r := groupRequest("org-p", "m", GroupScope{GroupID: testGID, + Limits: RateLimits{TotalPromptTokens: &promptCap}}) + + for i := 0; i < 2; i++ { + lease, err := a.Admit(context.Background(), r) + if err != nil { + t.Fatalf("admit %d: %v, want admitted", i+1, err) + } + if err := lease.CompleteUsage(context.Background(), Usage{TotalPromptTokens: 10}); err != nil { + t.Fatalf("CompleteUsage %d: %v", i+1, err) + } + } + + _, err := a.Admit(context.Background(), r) + var rejected *Rejected + if !errors.As(err, &rejected) { + t.Fatalf("third admit: %v, want a contract rejection", err) + } + if !rejected.Contractual || rejected.RetryAfter <= 0 { + t.Fatalf("rejection = %+v, want Contractual with a Retry-After (429 mapping)", rejected) + } + if rejected.Scope != "group:"+testGID { + t.Fatalf("rejection scope = %q, want the group id named", rejected.Scope) + } + if rejected.Dimension != "total_prompt_tokens" { + t.Fatalf("rejection dimension = %q, want the total-prompt window", rejected.Dimension) + } +} + +// TestGroupRateZeroCapBlocksEveryRequest: an explicit 0 is a zero cap, not +// unlimited — the same R4 sentinel as every other contract scope. +func TestGroupRateZeroCapBlocksEveryRequest(t *testing.T) { + a, _ := testAdmitter(t, config.AdmissionSettings{}) + zero := int64(0) + _, err := a.Admit(context.Background(), groupRequest("org-z", "m", + groupScope(testGID, &zero))) + var rejected *Rejected + if !errors.As(err, &rejected) || !rejected.Contractual { + t.Fatalf("zero-capped group admit: %v, want a contractual rejection", err) + } +} + +// TestGroupRateNilLimitIsUnlimited: a group scope whose rate fields are all +// nil contributes no contract scope (unlimited), so requests flow freely. +func TestGroupRateNilLimitIsUnlimited(t *testing.T) { + a, _ := testAdmitter(t, config.AdmissionSettings{}) + for i := 0; i < 3; i++ { + lease, err := a.Admit(context.Background(), groupRequest("org-u", "m", + groupScope(testGID, nil))) + if err != nil { + t.Fatalf("admit %d: %v, want admitted (nil limits are unlimited)", i+1, err) + } + _ = lease.Complete(context.Background(), 0) + } +} + +// TestGroupScopeMissingGroupIDFailsClosed: a scope carrying limits without a +// group id is a broken envelope, and broken trusted identity fails closed — +// distinct from the store-outage bypass. +func TestGroupScopeMissingGroupIDFailsClosed(t *testing.T) { + a, _ := testAdmitter(t, config.AdmissionSettings{}) + reqCap := int64(10) + _, err := a.Admit(context.Background(), groupRequest("org-i", "m", + GroupScope{Limits: RateLimits{Requests: &reqCap}})) + if !errors.Is(err, ErrInvalidIdentity) { + t.Fatalf("anonymous group scope: %v, want ErrInvalidIdentity (fail closed)", err) + } + + // A scope with nothing to enforce and no id is inert, not an error. + lease, err := a.Admit(context.Background(), groupRequest("org-i", "m", GroupScope{})) + if err != nil { + t.Fatalf("inert anonymous scope: %v, want admission", err) + } + _ = lease.Complete(context.Background(), 0) +} + +// TestGroupScopesSettleThroughCompleteUsage: the group contract scope rides +// the lease like any other scope — CompleteUsage settles it through the same +// finishScript path, charging the window the engine-reported tokens (not the +// reservation). Generated cap 50: settle 30, then 20 — both admit; a third +// reservation that would push the window past 50 is rejected. +func TestGroupScopesSettleThroughCompleteUsage(t *testing.T) { + a, _ := testAdmitter(t, config.AdmissionSettings{}) + genCap := int64(50) + r := groupRequest("org-l", "m", GroupScope{GroupID: testGID, + Limits: RateLimits{GeneratedTokens: &genCap}}) + + lease, err := a.Admit(context.Background(), r) + if err != nil { + t.Fatalf("first admit: %v", err) + } + if err := lease.Complete(context.Background(), 30); err != nil { + t.Fatalf("first Complete: %v", err) + } + + lease2, err := a.Admit(context.Background(), r) + if err != nil { + t.Fatalf("second admit (window at 30/50): %v, want admitted", err) + } + if err := lease2.Complete(context.Background(), 20); err != nil { + t.Fatalf("second Complete: %v", err) + } + + _, err = a.Admit(context.Background(), r) + var rejected *Rejected + if !errors.As(err, &rejected) || !rejected.Contractual { + t.Fatalf("third admit (window at 50/50): %v, want a contractual rejection", err) + } +} + +// TestGroupScopesSettleZeroUsageRefundsReservation: the settlement test above +// cannot tell settlement from a reservation left charged (each request +// reserves 20 output tokens, so an unsettled window would also read 20, 40, +// 60). Here every request completes with zero generated tokens: against a cap +// of 50, five requests admit only if each settlement refunds its 20-token +// reservation — without settlement the third would be rejected at 60/50. +func TestGroupScopesSettleZeroUsageRefundsReservation(t *testing.T) { + a, _ := testAdmitter(t, config.AdmissionSettings{}) + genCap := int64(50) + r := groupRequest("org-z0", "m", GroupScope{GroupID: testGID, + Limits: RateLimits{GeneratedTokens: &genCap}}) + + for i := 0; i < 5; i++ { + lease, err := a.Admit(context.Background(), r) + if err != nil { + t.Fatalf("admit %d: %v, want admitted (each zero-usage Complete refunds the reservation)", i+1, err) + } + if err := lease.Complete(context.Background(), 0); err != nil { + t.Fatalf("Complete %d: %v", i+1, err) + } + } +} + +// TestGroupScopesSettleChargesActualUsage: settling 45 generated tokens leaves +// the window at 45/50, so the next 20-token reservation is rejected — even +// though an unsettled window (holding only the first 20-token reservation) +// would have fit it at 40/50. +func TestGroupScopesSettleChargesActualUsage(t *testing.T) { + a, _ := testAdmitter(t, config.AdmissionSettings{}) + genCap := int64(50) + r := groupRequest("org-l45", "m", GroupScope{GroupID: testGID, + Limits: RateLimits{GeneratedTokens: &genCap}}) + + lease, err := a.Admit(context.Background(), r) + if err != nil { + t.Fatalf("first admit: %v", err) + } + if err := lease.Complete(context.Background(), 45); err != nil { + t.Fatalf("Complete: %v", err) + } + + _, err = a.Admit(context.Background(), r) + var rejected *Rejected + if !errors.As(err, &rejected) || !rejected.Contractual || rejected.Scope != "group:"+testGID { + t.Fatalf("admit after settling 45/50: %v, want a contractual group rejection", err) + } +} diff --git a/internal/admission/real_valkey_integration_test.go b/internal/admission/real_valkey_integration_test.go index 856af8a..8f088f3 100644 --- a/internal/admission/real_valkey_integration_test.go +++ b/internal/admission/real_valkey_integration_test.go @@ -414,3 +414,78 @@ func TestRealValkeyContractOnlyBurst(t *testing.T) { }) } } + +// TestRealValkeyGroupScopes runs the group fan-out through real Valkey's Lua: +// one request carrying 16 rate-limited group scopes (the envelope maximum) +// admits and settles, each group's window is charged the settled usage rather +// than the reservation, and a zero-capped group is a contractual rejection. +func TestRealValkeyGroupScopes(t *testing.T) { + addr := os.Getenv("PHOEBE_TEST_ADMISSION_VALKEY_ADDR") + if addr == "" { + t.Skip("PHOEBE_TEST_ADMISSION_VALKEY_ADDR not set; skipping real-Valkey group scopes") + } + ctx, cancel := context.WithTimeout(context.Background(), 75*time.Second) + defer cancel() + client := redis.NewClient(&redis.Options{Addr: addr}) + if err := client.Ping(ctx).Err(); err != nil { + t.Fatalf("ping real Valkey: %v", err) + } + cfg := config.AdmissionSettings{ + KeyPrefix: fmt.Sprintf("phoebe-admission-groups-%d", time.Now().UnixNano()), + LeaseTTL: time.Second, + Platform: limits(64), + } + a := New(client, cfg) + t.Cleanup(func() { + _ = client.Del(context.Background(), a.counters, a.leases, a.expiries, a.windowExpiries).Err() + _ = client.Close() + }) + + // The windows are fixed per minute; start clear of a boundary so the + // settle and the per-group checks below land in the same window. + if s := time.Now().Second(); s >= 50 { + time.Sleep(time.Duration(61-s) * time.Second) + } + + genCap := int64(50) + scopes := make([]GroupScope, 16) + for i := range scopes { + scopes[i] = GroupScope{GroupID: fmt.Sprintf("%032x", i+1), + Limits: RateLimits{GeneratedTokens: &genCap}} + } + r := request("org-g", "model") + r.GroupScopes = scopes + + lease, err := a.Admit(ctx, r) + if err != nil { + t.Fatalf("admit 16-group request: %v", err) + } + // Settle 45 generated tokens: every group window now reads 45/50. A window + // left holding only the 20-token reservation would read 20/50 instead. + if err := lease.Complete(ctx, 45); err != nil { + t.Fatalf("settle 16-group request: %v", err) + } + + // Each group alone: a 6-token reservation fits an unsettled 20/50 window + // but not a settled 45/50 one, so a rejection naming that group proves its + // own window settled. + for _, gs := range scopes { + probe := request("org-g", "model") + probe.ReservedOutputTokens = 6 + probe.GroupScopes = []GroupScope{gs} + _, err := a.Admit(ctx, probe) + rejected, ok := err.(*Rejected) + if !ok || !rejected.Contractual || rejected.Scope != "group:"+gs.GroupID { + t.Fatalf("probe for group %s after settling 45/50 = %T %v, want a contractual rejection naming the group", gs.GroupID, err, err) + } + } + + zero := int64(0) + zr := request("org-g", "model") + zr.GroupScopes = []GroupScope{{GroupID: fmt.Sprintf("%032x", 99), + Limits: RateLimits{Requests: &zero}}} + _, err = a.Admit(ctx, zr) + if rejected, ok := err.(*Rejected); !ok || !rejected.Contractual { + t.Fatalf("zero-capped group admit = %T %v, want a contractual rejection", err, err) + } +} diff --git a/internal/identity/group_scopes_test.go b/internal/identity/group_scopes_test.go new file mode 100644 index 0000000..75c74c1 --- /dev/null +++ b/internal/identity/group_scopes_test.go @@ -0,0 +1,27 @@ +package identity + +import ( + "net/http/httptest" + "strings" + "testing" +) + +// TestFromRequestGroupScopesUntrustedHeaderAbsent pins the R3 gate for the new +// envelope: when the header is NOT in the active trusted set, it reads as +// ABSENT — no raw value reaches the proxy parse, no matter what the client +// sends. +func TestFromRequestGroupScopesUntrustedHeaderAbsent(t *testing.T) { + const gidA = "a1b2c3d4e5f60718293a4b5c6d7e8f90" + // The active set WITHOUT X-Saturn-Group-Scopes, as a not-yet-upgraded + // chart would render it. + withTrustedHeadersEnv(t, strings.Join([]string{ + HeaderGateway, HeaderOrgID, HeaderOwnerID, HeaderServingMode, HeaderServedModel, + }, ","), true) + + r := httptest.NewRequest("POST", "/v1/chat/completions", nil) + r.Header.Set(HeaderGroupScopes, "v1;"+gidA+":,,,,") + id := FromRequest(r) + if id.GroupScopes != "" { + t.Fatalf("untrusted group scopes read as present: GroupScopes=%q, want absent", id.GroupScopes) + } +} diff --git a/internal/identity/identity.go b/internal/identity/identity.go index ef9f715..3a81e49 100644 --- a/internal/identity/identity.go +++ b/internal/identity/identity.go @@ -147,6 +147,25 @@ const ( HeaderOwnerRateLimitUncachedPromptTokens = "X-Saturn-Owner-Rate-Limit-Uncached-Prompt-Tokens" HeaderOwnerRateLimitGeneratedTokens = "X-Saturn-Owner-Rate-Limit-Generated-Tokens" + // HeaderGroupScopes carries the membership-aware GROUP quota envelope + // (ruled 2026-10-07, Q-T1(b)+Q-T2(b)): the whole group scope set rides ONE + // allowlisted header because Traefik's authResponseHeaders allowlist is a + // fixed name list and a member can belong to several groups. Grammar + // (frozen envelope, parsed strictly at the proxy/admission layer): + // + // v1;:,,,,;:... + // + // is the 32-hex group id; each rate field is empty (unlimited), 0 + // (zero cap), or a positive per-minute value (R4 sentinels). The + // field is RESERVED for the deferred monthly spend cap: it must be EMPTY + // in this build — a non-empty value is malformed and fails closed (503) + // at the proxy (the parked-activation boundary; parseTrustedGroupScopes). + // At most 16 entries, under 4 KiB. Like every scoped limit header it + // resolves through the R3 trusted-header registry: outside the active set + // it reads as ABSENT, so no group quota is ever read from a client- + // supplied value. + HeaderGroupScopes = "X-Saturn-Group-Scopes" + // HeaderUpstream carries the EXACT backend the request must be forwarded to — // `host:port` (e.g. pd-abcde-mymodel-r123.main-namespace.svc.cluster.local:8000). // @@ -228,6 +247,13 @@ type Identity struct { OwnerRateLimitTotalPromptTokens string OwnerRateLimitUncachedPromptTokens string OwnerRateLimitGeneratedTokens string + // GroupScopes is the raw, trusted X-Saturn-Group-Scopes envelope value + // (see HeaderGroupScopes), read through the R3 trusted-header registry. + // Empty when the header is absent or untrusted. The strict parse — per-group + // rate limits, bounds — happens at the proxy/admission layer + // (parseTrustedGroupScopes); Identity carries the envelope verbatim, + // exactly like the eight scoped rate-limit fields above. + GroupScopes string } // The two serving modes, spelled exactly as they are stored in billing_event and @@ -248,8 +274,9 @@ func ValidServingMode(s string) bool { // validation beyond reading the values; authorization happened at the edge. // // The R3 envelope reads (gateway mark, org, owner, serving mode, served -// model, and every rate-limit policy header — the pinned 13) resolve through -// the trusted-header registry: a header outside the active set is treated +// model, every rate-limit policy header, and the group scope envelope — +// the pinned 14) resolve through the trusted-header registry: a header +// outside the active set is treated // as ABSENT, never read for a trust decision. The remaining identity headers // are read directly (ratified edge contract, outside the R3 gate). The proxy // calls FromRequest BEFORE StripSaturnHeaders removes every X-Saturn-* header @@ -284,6 +311,7 @@ func FromRequest(r *http.Request) Identity { OwnerRateLimitTotalPromptTokens: trustedHeaderValue(r, HeaderOwnerRateLimitTotalPromptTokens), OwnerRateLimitUncachedPromptTokens: trustedHeaderValue(r, HeaderOwnerRateLimitUncachedPromptTokens), OwnerRateLimitGeneratedTokens: trustedHeaderValue(r, HeaderOwnerRateLimitGeneratedTokens), + GroupScopes: trustedHeaderValue(r, HeaderGroupScopes), } } diff --git a/internal/identity/trusted_headers.go b/internal/identity/trusted_headers.go index 9dc0df3..aa28682 100644 --- a/internal/identity/trusted_headers.go +++ b/internal/identity/trusted_headers.go @@ -9,10 +9,10 @@ package identity // PHOEBE_TRUSTED_HEADERS, comma-separated, rendered from the saturn-k8s // phoebe chart's ConfigMap); when the config is empty, unset, or malformed // (nothing left after trimming and dropping empties), a hard-coded fallback -// equal to the pinned 13 engages so phoebe fails closed, and a loud warning +// equal to the pinned 14 engages so phoebe fails closed, and a loud warning // flags the misrendered chart for the operator. // -// The set covers exactly the R3 envelope headers listed in the pinned 13. +// The set covers exactly the R3 envelope headers listed in the pinned 14. // The remaining identity headers (AuthID / UserID / GroupID / ResourceID / // ResourceType / BaseModel / Adapter / Upstream) are read under the ratified // edge contract (ForwardAuth authResponseHeaders allowlist + Atlas @@ -62,6 +62,7 @@ var pinnedTrustedHeaders = []string{ HeaderOwnerRateLimitTotalPromptTokens, HeaderOwnerRateLimitUncachedPromptTokens, HeaderOwnerRateLimitGeneratedTokens, + HeaderGroupScopes, } // requiredTrustedHeaders are the envelope headers request handling cannot work @@ -91,7 +92,7 @@ type trustedHeaderSet map[string]struct{} var activeTrustedHeaders atomic.Pointer[trustedHeaderSet] func init() { - // Before config load the pinned 13 are active: phoebe never reads an + // Before config load the pinned 14 are active: phoebe never reads an // envelope header outside the list, before OR after the runtime config // is loaded, and the pre-load default must preserve request handling. fallback := newTrustedHeaderSet(pinnedTrustedHeaders) @@ -129,7 +130,7 @@ func trustedHeaderValue(r *http.Request, name string) string { // active (entries are canonicalized; case is ignored per HTTP // convention); // - unset, empty/whitespace-only, or nothing left after splitting and -// trimming -> the pinned 13 engage AND a loud warning flags the +// trimming -> the pinned 14 engage AND a loud warning flags the // misrendered chart. Fallback, never a startup failure: phoebe must keep // serving with the pinned set rather than crash-loop behind a broken // ConfigMap render. diff --git a/internal/identity/trusted_headers_test.go b/internal/identity/trusted_headers_test.go index 9f96166..6a9180e 100644 --- a/internal/identity/trusted_headers_test.go +++ b/internal/identity/trusted_headers_test.go @@ -48,12 +48,13 @@ func pinnedSet() map[string]struct{} { return set } -// TestPinnedFallbackIsExactlyThe13 pins the fallback to the exact +// TestPinnedFallbackIsExactlyThe14 pins the fallback to the exact // pinned names, in the exact pinned order — the set the chart's ConfigMap // is supposed to render. Ruling R8 removed the five legacy single-scope // quota headers (X-Saturn-Service-Tier, X-Saturn-Rate-Limit-*), taking the -// R3 set from 18 to 13. -func TestPinnedFallbackIsExactlyThe13(t *testing.T) { +// R3 set from 18 to 13; the membership-aware group scope envelope +// (X-Saturn-Group-Scopes, ruled 2026-10-07) takes it from 13 to 14. +func TestPinnedFallbackIsExactlyThe14(t *testing.T) { want := []string{ "X-Saturn-Gateway", "X-Saturn-Org-Id", @@ -68,14 +69,15 @@ func TestPinnedFallbackIsExactlyThe13(t *testing.T) { "X-Saturn-Owner-Rate-Limit-Total-Prompt-Tokens", "X-Saturn-Owner-Rate-Limit-Uncached-Prompt-Tokens", "X-Saturn-Owner-Rate-Limit-Generated-Tokens", + "X-Saturn-Group-Scopes", } if !reflect.DeepEqual(pinnedTrustedHeaders, want) { - t.Fatalf("pinnedTrustedHeaders = %v, want the pinned 13 in order %v", pinnedTrustedHeaders, want) + t.Fatalf("pinnedTrustedHeaders = %v, want the pinned 14 in order %v", pinnedTrustedHeaders, want) } } // TestLoadTrustedHeadersFallback: unset, empty, whitespace-only, or -// all-empty-after-split config engages the pinned 13 — never an empty set. +// all-empty-after-split config engages the pinned 14 — never an empty set. func TestLoadTrustedHeadersFallback(t *testing.T) { cases := []struct { name string @@ -92,7 +94,7 @@ func TestLoadTrustedHeadersFallback(t *testing.T) { t.Run(tc.name, func(t *testing.T) { withTrustedHeadersEnv(t, tc.value, tc.set) if got := ActiveTrustedHeaders(); !reflect.DeepEqual(got, pinnedSet()) { - t.Fatalf("active set = %v, want the pinned 13 %v", got, pinnedSet()) + t.Fatalf("active set = %v, want the pinned 14 %v", got, pinnedSet()) } }) } @@ -308,7 +310,7 @@ func TestLoadTrustedHeadersErrorsWhenOwnerOrOrgAnchorMissing(t *testing.T) { } } -// TestLoadTrustedHeadersNoErrorForPinnedList: the full pinned 13, configured +// TestLoadTrustedHeadersNoErrorForPinnedList: the full pinned 14, configured // explicitly, satisfies every required header and logs nothing at ERROR. func TestLoadTrustedHeadersNoErrorForPinnedList(t *testing.T) { out := loadTrustedHeadersCapturingErrors(t, strings.Join(pinnedTrustedHeaders, ",")) @@ -434,6 +436,7 @@ var candidateSaturnHeaders = []string{ HeaderOwnerRateLimitTotalPromptTokens, HeaderOwnerRateLimitUncachedPromptTokens, HeaderOwnerRateLimitGeneratedTokens, + HeaderGroupScopes, } // packageSaturnHeaderConstants parses the package's non-test Go files and diff --git a/internal/proxy/admission.go b/internal/proxy/admission.go index 81dc316..aa1207c 100644 --- a/internal/proxy/admission.go +++ b/internal/proxy/admission.go @@ -7,7 +7,9 @@ import ( "fmt" "math" "net/http" + "regexp" "strconv" + "strings" "sync/atomic" "time" @@ -436,15 +438,18 @@ func anyScopedLimitPresent(id identity.Identity) bool { } // trustedPolicyEnvelopePresent reports whether any part of the trusted quota -// envelope — the owner-id anchor or any scoped limit header — reached this -// request. Any part of it engages admission's fail-closed checks even under -// admission.enabled=false: a missing organization identity, a malformed or -// partial envelope (a limit header without the owner-id anchor fails closed -// in parseTrustedRateLimits), and an underivable graph scope all answer 503. +// envelope — the owner-id anchor, any scoped limit header, or the group scope +// envelope — reached this request. Any part of it engages admission's +// fail-closed checks even under admission.enabled=false: a missing +// organization identity, a malformed or partial envelope (a limit header +// without the owner-id anchor fails closed in parseTrustedRateLimits; a +// malformed group scope envelope fails closed in parseTrustedGroupScopes), +// and an underivable graph scope all answer 503. // Under admission.enabled=false the admission store itself is only consulted -// when a contract limit header is present (see the Admit call in proxy.go). +// when a limit is present to enforce — a contract limit header or a group +// scope carrying a rate limit (see the Admit call in proxy.go). func trustedPolicyEnvelopePresent(id identity.Identity) bool { - return id.OwnerID != "" || anyScopedLimitPresent(id) + return id.OwnerID != "" || anyScopedLimitPresent(id) || id.GroupScopes != "" } // undeclaredOutputReservation is the output-token reservation for a request @@ -467,6 +472,143 @@ func undeclaredOutputReservation(defaultOutput int64, scopes ...admission.RateLi // for pre-R8 producers (see legacyQuotaHeadersPresent). var errNoTrustedRateLimitPolicy = errors.New("incomplete trusted shared-inference rate-limit policy") +// The group scope envelope (X-Saturn-Group-Scopes) bounds, frozen with the +// grammar (see identity.HeaderGroupScopes): at most 16 entries and the whole +// header under 4 KiB. Exceeding either, like any malformed field, is a 503 at +// the proxy with a loud log — the fail-visible default for the multi-group +// case: usage counts against EVERY group the caller belongs to that carries +// limits, and the envelope must never silently drop one. +const ( + maxGroupScopeEntries = 16 + maxGroupScopeBytes = 4096 +) + +// groupIDRe pins a group id to 32 lowercase hex chars (uuid4 hex, as Atlas +// mints). The frozen envelope pins this grammar; anything else is malformed. +var groupIDRe = regexp.MustCompile(`^[0-9a-f]{32}$`) + +// groupRateRe pins a rate field to the frozen grammar: "0" (zero cap) or a +// positive unpadded decimal. strconv.ParseInt alone also accepts "+30", +// "-0" and "0030", which the grammar forbids — an encoder bug there must +// fail closed (503), not be silently enforced. +var groupRateRe = regexp.MustCompile(`^(0|[1-9][0-9]*)$`) + +// parseTrustedGroupScopes parses the trusted group quota envelope +// (X-Saturn-Scopes grammar: v1;:,,,,;...). +// R4 sentinels exactly as parseTrustedRateLimits: an empty rate field is +// unlimited (nil), "0" is a zero cap, a positive value is the per-minute cap. +// The parse is STRICT — a malformed entry, an out-of-grammar gid, a +// duplicate gid, an uncached-above-total relation, more than +// maxGroupScopeEntries entries, or a header at/over maxGroupScopeBytes is an +// error and the request fails closed (503), like every other envelope +// violation. An absent header parses to nil scopes (no group quotas on this +// request), and "v1" alone parses to zero groups. +// +// THE SPEND FIELD IS THE PARKED-ACTIVATION BOUNDARY: the fifth field is +// reserved for the deferred monthly spend cap and MUST be empty in this +// build. A non-empty spend field is malformed and fails closed (503) with a +// loud log, whatever its shape — activating spend caps later is a parser +// change at this one site, not an envelope renegotiation. +// +// TRUST: the value is read by identity.FromRequest ONLY when the header is in +// the active trusted set, so a client can never stamp its own group quotas. +func parseTrustedGroupScopes(id identity.Identity) ([]admission.GroupScope, error) { + raw := id.GroupScopes + if raw == "" { + return nil, nil + } + if len(raw) >= maxGroupScopeBytes { + return nil, fmt.Errorf("trusted %s header exceeds the %d-byte envelope bound (%d bytes)", identity.HeaderGroupScopes, maxGroupScopeBytes, len(raw)) + } + parts := strings.Split(raw, ";") + if parts[0] != "v1" { + return nil, fmt.Errorf("trusted %s header missing the v1 prefix", identity.HeaderGroupScopes) + } + entries := parts[1:] + if len(entries) > maxGroupScopeEntries { + return nil, fmt.Errorf("trusted %s header carries %d group entries, over the %d-entry bound", identity.HeaderGroupScopes, len(entries), maxGroupScopeEntries) + } + scopes := make([]admission.GroupScope, 0, len(entries)) + seen := make(map[string]struct{}, len(entries)) + for _, entry := range entries { + fields := strings.Split(entry, ":") + if len(fields) != 2 { + return nil, fmt.Errorf("trusted %s entry %q is not :,,,,", identity.HeaderGroupScopes, entry) + } + gid := fields[0] + if !groupIDRe.MatchString(gid) { + return nil, fmt.Errorf("trusted %s entry has a malformed group id %q (want 32 lowercase hex)", identity.HeaderGroupScopes, gid) + } + if _, dup := seen[gid]; dup { + return nil, fmt.Errorf("trusted %s header lists group %s twice", identity.HeaderGroupScopes, gid) + } + seen[gid] = struct{}{} + rates := strings.Split(fields[1], ",") + if len(rates) != 5 { + return nil, fmt.Errorf("trusted %s entry for group %s has %d fields, want ,,,,", identity.HeaderGroupScopes, gid, len(rates)) + } + var scope admission.GroupScope + scope.GroupID = gid + var err error + if scope.Limits.Requests, err = parseGroupRate(gid, "requests", rates[0]); err != nil { + return nil, err + } + if scope.Limits.TotalPromptTokens, err = parseGroupRate(gid, "total prompt tokens", rates[1]); err != nil { + return nil, err + } + if scope.Limits.UncachedPromptTokens, err = parseGroupRate(gid, "uncached prompt tokens", rates[2]); err != nil { + return nil, err + } + if scope.Limits.GeneratedTokens, err = parseGroupRate(gid, "generated tokens", rates[3]); err != nil { + return nil, err + } + if scope.Limits.TotalPromptTokens != nil && scope.Limits.UncachedPromptTokens != nil && + *scope.Limits.UncachedPromptTokens > *scope.Limits.TotalPromptTokens { + return nil, fmt.Errorf("trusted %s group %s: uncached prompt limit exceeds total prompt limit", identity.HeaderGroupScopes, gid) + } + // The fifth field is reserved for the deferred monthly spend cap: it + // MUST be empty in this build. A non-empty value — whatever its shape — + // is malformed and fails closed (the parked-activation boundary; see + // the parseTrustedGroupScopes doc). + if rates[4] != "" { + return nil, fmt.Errorf("trusted %s group %s carries a non-empty spend field %q (monthly spend caps are not active in this build)", identity.HeaderGroupScopes, gid, rates[4]) + } + scopes = append(scopes, scope) + } + return scopes, nil +} + +// parseGroupRate parses one group envelope rate field: empty is unlimited +// (nil), "0" is a zero cap, a positive integer is the per-minute cap. A +// signed, zero-padded, negative, or non-integer value is malformed and fails +// closed. +func parseGroupRate(gid, dimension, value string) (*int64, error) { + if value == "" { + return nil, nil + } + if !groupRateRe.MatchString(value) { + return nil, fmt.Errorf("trusted %s group %s has a malformed %s limit %q", identity.HeaderGroupScopes, gid, dimension, value) + } + limit, err := strconv.ParseInt(value, 10, 64) + if err != nil || limit < 0 { + return nil, fmt.Errorf("trusted %s group %s has a malformed %s limit %q", identity.HeaderGroupScopes, gid, dimension, value) + } + return &limit, nil +} + +// anyGroupScopeEnforced reports whether at least one group scope carries a +// non-nil per-minute rate limit. An all-unlimited group envelope (every rate +// empty) enforces nothing, so under admission.enabled=false it must not touch +// the admission store — exactly like a request with no envelope. +func anyGroupScopeEnforced(scopes []admission.GroupScope) bool { + for _, scope := range scopes { + if scope.Limits.Any() { + return true + } + } + return false +} + // legacyQuotaHeaderNames are the five single-scope quota headers that R8 // removed from the trusted envelope. Phoebe never reads them for a trust or // limit decision; they are listed here ONLY so the proxy can log that a diff --git a/internal/proxy/group_scopes_test.go b/internal/proxy/group_scopes_test.go new file mode 100644 index 0000000..31d6299 --- /dev/null +++ b/internal/proxy/group_scopes_test.go @@ -0,0 +1,644 @@ +package proxy + +import ( + "bytes" + "io" + "log" + "net/http" + "net/http/httptest" + "net/url" + "strconv" + "strings" + "testing" + "time" + + "github.com/redis/go-redis/v9" + + "github.com/saturncloud/phoebe/internal/admission" + "github.com/saturncloud/phoebe/internal/config" + "github.com/saturncloud/phoebe/internal/identity" + "github.com/saturncloud/phoebe/internal/logging" +) + +// These tests cover the membership-aware group quota envelope +// (X-Saturn-Group-Scopes, ruled 2026-10-07) at the proxy layer: the strict +// parse (fail-visible: malformed/oversize → 503) and the end-to-end +// enforcement of the per-minute group rate limits (429 + Retry-After on +// denial). The envelope's fifth field is reserved for the deferred monthly +// spend cap: non-empty is malformed → 503 (the parked-activation boundary). + +const ( + testGroupA = "a1b2c3d4e5f60718293a4b5c6d7e8f90" + testGroupB = "00112233445566778899aabbccddeeff" + + // groupScopesHeader is a fully-unlimited two-group envelope: every rate + // empty (unlimited). + groupScopesHeader = "v1;" + testGroupA + ":,,,," + ";" + testGroupB + ":,,,," +) + +// groupScopesValue builds an envelope for one group with the given rate field +// string (verbatim between the colons) and spend field (always "" in valid +// envelopes; non-empty builds the malformed spend cases). +func groupScopesValue(gid, rates, spend string) string { + return "v1;" + gid + ":" + rates + "," + spend +} + +// TestParseTrustedGroupScopes exercises the frozen envelope grammar: +// v1;:,,,,;:... +func TestParseTrustedGroupScopes(t *testing.T) { + valid := []struct { + name string + in string + want []admission.GroupScope + }{ + { + "absent header", + "", + nil, + }, + { + "v1 alone is zero groups", + "v1", + nil, + }, + { + "single group all unlimited", + groupScopesValue(testGroupA, ",,,", ""), + []admission.GroupScope{{GroupID: testGroupA}}, + }, + { + "all four rates set", + groupScopesValue(testGroupA, "30,1000,500,2000", ""), + []admission.GroupScope{{ + GroupID: testGroupA, + Limits: admission.RateLimits{ + Requests: ptr64(30), TotalPromptTokens: ptr64(1000), + UncachedPromptTokens: ptr64(500), GeneratedTokens: ptr64(2000), + }, + }}, + }, + { + "rate zero is a zero cap", + groupScopesValue(testGroupA, "0,,,", ""), + []admission.GroupScope{{ + GroupID: testGroupA, + Limits: admission.RateLimits{Requests: ptr64(0)}, + }}, + }, + { + "multi group in envelope order", + "v1;" + testGroupA + ":1,,,," + ";" + testGroupB + ":,2,,,", + []admission.GroupScope{ + {GroupID: testGroupA, Limits: admission.RateLimits{Requests: ptr64(1)}}, + {GroupID: testGroupB, Limits: admission.RateLimits{TotalPromptTokens: ptr64(2)}}, + }, + }, + } + for _, tc := range valid { + t.Run(tc.name, func(t *testing.T) { + scopes, err := parseTrustedGroupScopes(identity.Identity{GroupScopes: tc.in}) + if err != nil { + t.Fatalf("parse(%q): %v, want success", tc.in, err) + } + if len(scopes) != len(tc.want) { + t.Fatalf("parse(%q) = %d scopes, want %d (%+v)", tc.in, len(scopes), len(tc.want), scopes) + } + for i, want := range tc.want { + got := scopes[i] + if got.GroupID != want.GroupID { + t.Fatalf("scope %d = %+v, want %+v", i, got, want) + } + if (got.Limits.Requests == nil) != (want.Limits.Requests == nil) || + (got.Limits.Requests != nil && *got.Limits.Requests != *want.Limits.Requests) { + t.Fatalf("scope %d requests = %v, want %v", i, got.Limits.Requests, want.Limits.Requests) + } + if (got.Limits.TotalPromptTokens == nil) != (want.Limits.TotalPromptTokens == nil) || + (got.Limits.TotalPromptTokens != nil && *got.Limits.TotalPromptTokens != *want.Limits.TotalPromptTokens) { + t.Fatalf("scope %d total = %v, want %v", i, got.Limits.TotalPromptTokens, want.Limits.TotalPromptTokens) + } + if (got.Limits.UncachedPromptTokens == nil) != (want.Limits.UncachedPromptTokens == nil) || + (got.Limits.UncachedPromptTokens != nil && *got.Limits.UncachedPromptTokens != *want.Limits.UncachedPromptTokens) { + t.Fatalf("scope %d uncached = %v, want %v", i, got.Limits.UncachedPromptTokens, want.Limits.UncachedPromptTokens) + } + if (got.Limits.GeneratedTokens == nil) != (want.Limits.GeneratedTokens == nil) || + (got.Limits.GeneratedTokens != nil && *got.Limits.GeneratedTokens != *want.Limits.GeneratedTokens) { + t.Fatalf("scope %d generated = %v, want %v", i, got.Limits.GeneratedTokens, want.Limits.GeneratedTokens) + } + } + }) + } + + malformed := []struct { + name string + in string + }{ + {"missing v1 prefix", testGroupA + ":,,,,"}, + {"wrong prefix version", "v2;" + testGroupA + ":,,,,"}, + {"empty entry from trailing semicolon", "v1;" + testGroupA + ":,,,,;"}, + {"empty entry in the middle", "v1;;" + testGroupA + ":,,,,"}, + {"gid uppercase hex", "v1;A1B2C3D4E5F60718293A4B5C6D7E8F90:,,,,"}, + {"gid too short", "v1;a1b2:,,,,"}, + {"gid not hex", "v1;g1b2c3d4e5f60718293a4b5c6d7e8f90:,,,,"}, + {"missing colon", "v1;" + testGroupA + ",,,,"}, + {"too few rate fields", groupScopesValue(testGroupA, ",,", "")}, + {"too many rate fields", "v1;" + testGroupA + ":1,2,3,4,5,6"}, + {"negative rate", groupScopesValue(testGroupA, "-1,,,", "")}, + {"non-numeric rate", groupScopesValue(testGroupA, "x,,,", "")}, + {"signed requests rate", groupScopesValue(testGroupA, "+5,,,", "")}, + {"negative zero total rate", groupScopesValue(testGroupA, ",-0,,", "")}, + {"zero-padded uncached rate", groupScopesValue(testGroupA, ",,05,", "")}, + {"signed generated rate", groupScopesValue(testGroupA, ",,,+30", "")}, + {"zero-padded requests rate", groupScopesValue(testGroupA, "0030,,,", "")}, + {"non-empty spend field", groupScopesValue(testGroupA, ",,,", "100")}, + {"non-empty spend field, decimal shape", groupScopesValue(testGroupA, ",,,", "100.000000000")}, + {"non-empty spend field, zero", groupScopesValue(testGroupA, ",,,", "0")}, + {"non-empty spend field, exponent shape", groupScopesValue(testGroupA, ",,,", "1e3")}, + {"uncached above total", groupScopesValue(testGroupA, ",100,101,", "")}, + {"duplicate gid", "v1;" + testGroupA + ":,,,," + ";" + testGroupA + ":1,,,"}, + } + for _, tc := range malformed { + t.Run(tc.name, func(t *testing.T) { + if _, err := parseTrustedGroupScopes(identity.Identity{GroupScopes: tc.in}); err == nil { + t.Fatalf("parse(%q) succeeded, want a fail-closed error", tc.in) + } + }) + } + + t.Run("over the entry bound", func(t *testing.T) { + entries := make([]string, 0, maxGroupScopeEntries+1) + for i := 0; i <= maxGroupScopeEntries; i++ { + entries = append(entries, groupIDFor(i)+":,,,,") + } + if _, err := parseTrustedGroupScopes(identity.Identity{GroupScopes: "v1;" + strings.Join(entries, ";")}); err == nil { + t.Fatalf("parse of %d entries succeeded, want the %d-entry bound enforced", len(entries), maxGroupScopeEntries) + } + }) + + t.Run("over the size bound", func(t *testing.T) { + // The size check fires before any grammar validation, so one entry + // whose requests rate is a long digit string — overlong for the rate + // grammar, which never gets a say — is a valid fixture: the parse + // must fail on the byte bound alone, and the error must name it. + in := groupScopesValue(testGroupA, strings.Repeat("9", maxGroupScopeBytes), "") + if len(in) < maxGroupScopeBytes { + t.Fatalf("fixture = %d bytes, want at least the %d-byte bound", len(in), maxGroupScopeBytes) + } + _, err := parseTrustedGroupScopes(identity.Identity{GroupScopes: in}) + if err == nil { + t.Fatalf("parse of a %d-byte header succeeded, want the %d-byte envelope bound enforced", len(in), maxGroupScopeBytes) + } + if !strings.Contains(err.Error(), "4096-byte envelope bound") { + t.Fatalf("parse error = %q, want it to name the 4096-byte envelope bound", err) + } + }) + + t.Run("maximal valid envelope stays under the size bound", func(t *testing.T) { + // K4B: a >4 KiB fixture built only from grammatically valid entries + // does not exist. The 16-entry cap bounds the count and every field + // is length-bounded by its own grammar — the spend field reserved- + // empty in this build — so the largest envelope the grammar admits is + // the one below: 16 entries, every rate at the 18 nines that stay + // inside int64, about 1.3 KiB, well under the 4 KiB bound. Any header + // long enough to reach 4 KiB is malformed in at least one other way + // (an over-long rate field fails groupRateRe first), so the parser's + // size check is defense-in-depth ordering, not a bound a valid + // producer can ever hit — no valid-but-oversize fixture can pin it. + // The transport-level ceiling on any single header is the HTTP + // server's MaxHeaderBytes-style limit (net/http's default header + // block, which server.go does not override). The bounds the grammar + // actually enforces are pinned here and in the siblings above: this + // maximal envelope parses, and the shapes that could approach 4 KiB + // fail on the 16-entry cap (the "over the entry bound" case) or the + // per-entry grammar (the malformed table) — remove either check and + // this test's siblings fail. The size check is pinned by the "over + // the size bound" case above: its fixture is over 4 KiB and the rate + // grammar would also reject its overlong field, so that case asserts + // the error names the byte bound — delete the size check and the + // parse fails on the rate grammar instead, the assertion misses, and + // the case goes red. + maxRate := strings.Repeat("9", 18) // largest rate that still fits int64 + entries := make([]string, 0, maxGroupScopeEntries) + for i := 0; i < maxGroupScopeEntries; i++ { + entries = append(entries, groupIDFor(i)+":"+maxRate+","+maxRate+","+maxRate+","+maxRate+",") + } + in := "v1;" + strings.Join(entries, ";") + if len(in) >= maxGroupScopeBytes { + t.Fatalf("maximal valid envelope = %d bytes, want under the %d-byte bound — the fixture must stay the largest the grammar admits", len(in), maxGroupScopeBytes) + } + if _, err := parseTrustedGroupScopes(identity.Identity{GroupScopes: in}); err != nil { + t.Fatalf("parse of the maximal valid envelope (%d bytes) failed: %v", len(in), err) + } + }) +} + +// groupIDFor returns a distinct valid 32-hex group id for i. +func groupIDFor(i int) string { + s := strconv.FormatInt(int64(i), 16) + return strings.Repeat("0", 32-len(s)) + s +} + +// groupScopesRequest is a shared request carrying the group quota envelope. +func groupScopesRequest(t *testing.T, upstream *url.URL, envelope string) *http.Request { + t.Helper() + r := sharedRequest(upstream) + if envelope != "" { + r.Header.Set(identity.HeaderGroupScopes, envelope) + } + return r +} + +// proxyWithGroupScopes builds the admission-enabled proxy over a real +// (frozen-clock) miniredis. +func proxyWithGroupScopes(t *testing.T) (*Server, *recordingEmitter) { + t.Helper() + return proxyWithGroupScopesEnabled(t, true) +} + +// proxyWithGroupScopesEnabled builds the proxy over a real (frozen-clock) +// miniredis with the given admission.enabled flag. The admitter runs with the +// effective settings — operator tiers cleared under admission.enabled=false, +// exactly as a default chart install renders them. +func proxyWithGroupScopesEnabled(t *testing.T, enabled bool) (*Server, *recordingEmitter) { + t.Helper() + mr := frozenMiniredis(t) + cfg := proxyAdmissionConfig(10) + cfg.Enabled = enabled + cfg.ValkeyAddr = mr.Addr() + eff, ok := (&config.Settings{Admission: cfg}).EffectiveAdmission() + if !ok { + t.Fatal("effective admission settings: no store configured") + } + client := redis.NewClient(&redis.Options{Addr: mr.Addr()}) + t.Cleanup(func() { _ = client.Close() }) + admitter := admission.New(client, eff) + em := &recordingEmitter{} + s := New(&config.Settings{Admission: cfg}, logging.New(logging.ERROR), em).WithAdmitter(admitter) + return s, em +} + +// TestSharedGroupRateLimitAnswers429: a group requests-per-minute cap on the +// envelope is enforced exactly like the contract scopes — the burst admits, +// the overrun answers 429 + Retry-After (the contractual mapping). +func TestSharedGroupRateLimitAnswers429(t *testing.T) { + up, hits := sharedGroupBackend(t) + s, _ := proxyWithGroupScopes(t) + + r1 := groupScopesRequest(t, up, groupScopesValue(testGroupA, "1,,,", "")) + rr1 := httptest.NewRecorder() + s.Handler().ServeHTTP(rr1, r1) + if rr1.Code != http.StatusOK { + t.Fatalf("first request status=%d, want 200", rr1.Code) + } + + r2 := groupScopesRequest(t, up, groupScopesValue(testGroupA, "1,,,", "")) + rr2 := httptest.NewRecorder() + s.Handler().ServeHTTP(rr2, r2) + if rr2.Code != http.StatusTooManyRequests { + t.Fatalf("overrun status=%d, want 429", rr2.Code) + } + if rr2.Header().Get("Retry-After") == "" { + t.Fatal("overrun missing Retry-After") + } + if *hits != 1 { + t.Fatalf("upstream hits=%d, want 1 (the overrun never forwards)", *hits) + } +} + +// TestSharedGroupRateZeroCapAnswers429: an explicit 0 blocks every request +// for the group (R4 zero cap), whatever the other scopes say. +func TestSharedGroupRateZeroCapAnswers429(t *testing.T) { + up, _ := sharedGroupBackend(t) + s, _ := proxyWithGroupScopes(t) + + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, groupScopesRequest(t, up, groupScopesValue(testGroupA, "0,,,", ""))) + if rr.Code != http.StatusTooManyRequests { + t.Fatalf("zero-cap status=%d, want 429", rr.Code) + } +} + +// TestSharedGroupScopesMalformedHeaderFailsClosed: a malformed group envelope +// is a broken trusted policy, not a quota — 503 at the proxy, before +// admission, and nothing reaches the upstream or the emitter. +func TestSharedGroupScopesMalformedHeaderFailsClosed(t *testing.T) { + up, hits := sharedGroupBackend(t) + s, em := proxyWithGroupScopes(t) + + r := groupScopesRequest(t, up, "v1;NOT-A-GROUP:,,,,") + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, r) + if rr.Code != http.StatusServiceUnavailable { + t.Fatalf("malformed envelope status=%d, want 503", rr.Code) + } + if *hits != 0 { + t.Fatalf("upstream hits=%d, want 0 (fail closed before forwarding)", *hits) + } + if evs := em.waitForEvents(1, 200*time.Millisecond); len(evs) != 0 { + t.Fatalf("emitted events=%d, want 0 (a refused request meters nothing)", len(evs)) + } +} + +// TestSharedGroupScopesNonEmptySpendFailsClosed pins the parked-activation +// boundary end to end: the envelope's fifth field is reserved for the +// deferred monthly spend cap, and a producer stamping ANY non-empty value +// there — a well-formed decimal included — has its request refused 503 with +// a loud log, before upstream, metering nothing. Rejecting here (rather than +// ignoring the field) is deliberate: silently dropping a stamped cap would +// bill group traffic the operator believed was capped. +func TestSharedGroupScopesNonEmptySpendFailsClosed(t *testing.T) { + up, hits := sharedGroupBackend(t) + cfg := proxyAdmissionConfig(10) + var errBuf bytes.Buffer + logger := &logging.Logger{Debug: log.New(io.Discard, "", 0), Info: log.New(io.Discard, "", 0), Warn: log.New(io.Discard, "", 0), Error: log.New(&errBuf, "", 0)} + em := &recordingEmitter{} + s := New(&config.Settings{Admission: cfg}, logger, em).WithAdmitter(&countingAdmitter{}) + + r := groupScopesRequest(t, up, groupScopesValue(testGroupA, ",,,", "100.000000000")) + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, r) + if rr.Code != http.StatusServiceUnavailable { + t.Fatalf("non-empty spend field status=%d, want 503 (fail closed at the parse)", rr.Code) + } + if !strings.Contains(errBuf.String(), "non-empty spend field") { + t.Fatalf("error log = %q, want the loud non-empty-spend line", errBuf.String()) + } + if *hits != 0 { + t.Fatalf("upstream hits=%d, want 0 (fail closed before forwarding)", *hits) + } + if evs := em.waitForEvents(1, 200*time.Millisecond); len(evs) != 0 { + t.Fatalf("emitted events=%d, want 0 (a refused request meters nothing)", len(evs)) + } +} + +// TestSharedGroupScopesUnlimitedHeaderAdmits: the fully-unlimited envelope +// (every field empty) parses to scopes with nothing to enforce — requests +// flow and are metered. +func TestSharedGroupScopesUnlimitedHeaderAdmits(t *testing.T) { + up, hits := sharedGroupBackend(t) + s, em := proxyWithGroupScopes(t) + + r := groupScopesRequest(t, up, groupScopesHeader) + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, r) + if rr.Code != http.StatusOK { + t.Fatalf("status=%d, want 200", rr.Code) + } + if *hits != 1 { + t.Fatalf("upstream hits=%d, want 1", *hits) + } + evs := em.waitForEvents(1, time.Second) + if len(evs) != 1 { + t.Fatalf("emitted events=%d, want 1", len(evs)) + } +} + +// sharedGroupBackend is a shared-path upstream counting its hits. +func sharedGroupBackend(t *testing.T) (*url.URL, *int) { + t.Helper() + hits := 0 + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + hits++ + _, _ = w.Write([]byte(`{"model":"model-a","usage":{"prompt_tokens":2,"completion_tokens":3}}`)) + })) + t.Cleanup(backend.Close) + up, err := url.Parse(backend.URL) + if err != nil { + t.Fatal(err) + } + return up, &hits +} + +// TestSharedGroupScopesAdmissionTriggerWithAdmissionFlagOff pins the trigger: +// under admission.enabled=false the Admit round-trip is engaged by a scope +// that carries a rate limit, not by the mere presence of the group envelope. +// Every org/owner scoped limit header is absent, so only the group scopes can +// engage the trigger. An all-unlimited group envelope performs no admission +// call; an envelope with a rate limit still performs exactly one, unchanged +// from the enabled=true path. +func TestSharedGroupScopesAdmissionTriggerWithAdmissionFlagOff(t *testing.T) { + up, _ := sharedGroupBackend(t) + cfg := proxyAdmissionConfig(10) + cfg.Enabled = false + send := func(a *countingAdmitter, envelope string) *httptest.ResponseRecorder { + s := New(&config.Settings{Admission: cfg}, logging.New(logging.ERROR), &recordingEmitter{}).WithAdmitter(a) + req := groupScopesRequest(t, up, envelope) + for _, header := range identity.ScopedRateLimitHeaders { + req.Header.Del(header) + } + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, req) + return rr + } + + // The all-unlimited envelope (every rate empty) enforces nothing: no + // Admit, no Valkey reservation, no completion round-trip. + a := &countingAdmitter{} + if rr := send(a, groupScopesHeader); rr.Code != http.StatusOK || a.calls != 0 { + t.Fatalf("all-unlimited envelope: status=%d Admit calls=%d, want 200 and no admission round-trip", rr.Code, a.calls) + } + + // A rate-limited group scope engages admission exactly as a contract + // limit header does. + a = &countingAdmitter{} + if rr := send(a, groupScopesValue(testGroupA, "30,,,", "")); rr.Code != http.StatusOK || a.calls != 1 { + t.Fatalf("rate-limited group scope: status=%d Admit calls=%d, want 200 and exactly one Admit", rr.Code, a.calls) + } +} + +// TestSharedGroupScopesEnforcedWithAdmissionFlagOff is the envelope-driven +// enforcement test for the item-13 posture: admission.enabled=false with ONLY +// X-Saturn-Group-Scopes stamped (no org/owner limit headers). A group zero +// cap answers 429 + Retry-After before upstream, and a request within the +// group limits is admitted — the group-scope mirror of the contract-limits +// enabled=false cases in contract_limits_test.go. +func TestSharedGroupScopesEnforcedWithAdmissionFlagOff(t *testing.T) { + up, hits := sharedGroupBackend(t) + s, em := proxyWithGroupScopesEnabled(t, false) + + // Only the group envelope: every org/owner scoped limit header absent. + envelopeOnly := func(envelope string) *http.Request { + req := groupScopesRequest(t, up, envelope) + for _, header := range identity.ScopedRateLimitHeaders { + req.Header.Del(header) + } + return req + } + + // A group zero cap blocks every request with the contractual 429. + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, envelopeOnly(groupScopesValue(testGroupA, "0,,,", ""))) + if rr.Code != http.StatusTooManyRequests { + t.Fatalf("zero-cap status=%d, want 429", rr.Code) + } + if rr.Header().Get("Retry-After") == "" { + t.Fatal("zero-cap response missing Retry-After") + } + + // A request within the group limits is admitted and forwarded. + rr = httptest.NewRecorder() + s.Handler().ServeHTTP(rr, envelopeOnly(groupScopesValue(testGroupA, "30,,,", ""))) + if rr.Code != http.StatusOK { + t.Fatalf("within-limits status=%d, want 200", rr.Code) + } + if *hits != 1 { + t.Fatalf("upstream hits=%d, want 1 (the zero cap never forwards)", *hits) + } + evs := em.waitForEvents(1, time.Second) + if len(evs) != 1 { + t.Fatalf("emitted events=%d, want 1", len(evs)) + } +} + +// TestSharedGroupScopesOnlyEnvelopeFailsClosedWithAdmissionFlagOff closes the +// trigger hole in the enabled=false test above: that one builds on +// sharedRequest, which always stamps X-Saturn-Owner-Id, so its owner-id +// anchor alone makes trustedPolicyEnvelopePresent true — a regression +// dropping id.GroupScopes != "" from that function would leave every case +// above green. This test sends the group envelope with NO other trusted +// policy header: the owner-id anchor and every scoped rate-limit header are +// deleted, so only the group clause of trustedPolicyEnvelopePresent can +// engage admission. +// +// What the proxy actually answers for that shape, and why it is not 429 or +// 200: parseTrustedRateLimits runs first inside the admission block and +// fails closed on the missing contract anchor (errNoTrustedRateLimitPolicy), +// so the request is refused 503 before the group scopes are even parsed — a +// group zero cap can never be the rejecting reason here, and a within-limits +// envelope is never admitted either. Group enforcement under +// admission.enabled=false is only reachable WITH the owner-id anchor (the +// tests above); this shape pins the trigger, not the enforcement. +// +// The pin: with the group envelope as the only trusted header, both a zero +// cap and a within-limits envelope answer 503 and never reach the upstream. +// Drop id.GroupScopes != "" from trustedPolicyEnvelopePresent and both are +// forwarded (200), failing this test. +func TestSharedGroupScopesOnlyEnvelopeFailsClosedWithAdmissionFlagOff(t *testing.T) { + up, hits := sharedGroupBackend(t) + s, _ := proxyWithGroupScopesEnabled(t, false) + + // Only the group envelope: the owner-id anchor and every scoped + // rate-limit header deleted. + groupOnly := func(envelope string) *http.Request { + req := groupScopesRequest(t, up, envelope) + req.Header.Del(identity.HeaderOwnerID) + for _, header := range identity.ScopedRateLimitHeaders { + req.Header.Del(header) + } + return req + } + + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, groupOnly(groupScopesValue(testGroupA, "0,,,", ""))) + if rr.Code != http.StatusServiceUnavailable { + t.Fatalf("group-only zero-cap status=%d, want 503 (the contract-anchor parse fails closed before group scopes)", rr.Code) + } + + rr = httptest.NewRecorder() + s.Handler().ServeHTTP(rr, groupOnly(groupScopesValue(testGroupA, "30,,,", ""))) + if rr.Code != http.StatusServiceUnavailable { + t.Fatalf("group-only within-limits status=%d, want 503", rr.Code) + } + + if *hits != 0 { + t.Fatalf("upstream hits=%d, want 0 — a group-only request never forwards; with the group clause dropped from trustedPolicyEnvelopePresent both requests are admitted and this fails", *hits) + } +} + +// TestSharedGroupGeneratedCapClampsUndeclaredMaxTokensReservation: the +// undeclared-max_tokens output reservation is clamped to the tightest +// positive generated-token limit across EVERY scope class on the request — +// organization, owner, and each stamped group scope. A group whose generated +// cap sits below the configured default reserves the cap, not the default, so +// its undeclared requests admit instead of reserving the default and hitting +// the unsatisfiable 400 of +// TestMaxTokensAboveGeneratedLimitIsUnsatisfiable400. Dropping the group +// limits from the undeclaredOutputReservation call at proxy.go leaves every +// case here green only if it also breaks the group cap below the default — +// the end-to-end half fails: the unclamped 20-token default reservation over +// a 10-token group cap is rejected before upstream. +func TestSharedGroupGeneratedCapClampsUndeclaredMaxTokensReservation(t *testing.T) { + up, _ := sharedGroupBackend(t) + cfg := proxyAdmissionConfig(10) // DefaultMaxOutputTokens: 20 + + // noMaxTokens swaps the shared body ({"model":"model-a","max_tokens":20}) + // for an undeclared-output one, the shape the reservation clamp exists + // for. + noMaxTokens := func(req *http.Request) { + req.Body = io.NopCloser(strings.NewReader(`{"model":"model-a"}`)) + req.ContentLength = -1 + } + noMaxTokensReq := func(req *http.Request) *http.Request { + noMaxTokens(req) + return req + } + send := func(recording *recordingAdmitter, envelope string, mutate func(*http.Request)) *httptest.ResponseRecorder { + s := New(&config.Settings{Admission: cfg}, logging.New(logging.ERROR), &recordingEmitter{}).WithAdmitter(recording) + req := groupScopesRequest(t, up, envelope) + if mutate != nil { + mutate(req) + } + rr := httptest.NewRecorder() + s.Handler().ServeHTTP(rr, req) + return rr + } + + // The group generated cap below the default drives the reservation: 10, + // not the 20-token default. + recording := &recordingAdmitter{} + rr := send(recording, groupScopesValue(testGroupA, ",,,10", ""), noMaxTokens) + if rr.Code != http.StatusOK || recording.last.ReservedOutputTokens != 10 { + t.Fatalf("group cap 10: status=%d reserved=%d, want 200 with a 10-token reservation (the group cap, not the 20-token default)", rr.Code, recording.last.ReservedOutputTokens) + } + + // A group cap above the default changes nothing: the default stands. + recording = &recordingAdmitter{} + rr = send(recording, groupScopesValue(testGroupA, ",,,30", ""), noMaxTokens) + if rr.Code != http.StatusOK || recording.last.ReservedOutputTokens != 20 { + t.Fatalf("group cap 30: status=%d reserved=%d, want 200 with the 20-token default (a cap above the default does not raise the reservation)", rr.Code, recording.last.ReservedOutputTokens) + } + + // Every scope class feeds the same clamp and the tightest wins: org 5, + // owner 8, group 10 → 5. + recording = &recordingAdmitter{} + rr = send(recording, groupScopesValue(testGroupA, ",,,10", ""), func(req *http.Request) { + req.Header.Set(identity.HeaderOrgRateLimitGeneratedTokens, "5") + req.Header.Set(identity.HeaderOwnerRateLimitGeneratedTokens, "8") + noMaxTokens(req) + }) + if rr.Code != http.StatusOK || recording.last.ReservedOutputTokens != 5 { + t.Fatalf("org 5 / owner 8 / group 10: status=%d reserved=%d, want 200 with the tightest (5)", rr.Code, recording.last.ReservedOutputTokens) + } + + // End to end against the real store: the undeclared request from a + // group capped below the default is ADMITTED. Without the group limits in + // the clamp the reservation stays 20 over the 10-token cap and this is + // the unsatisfiable 400 — this half fails on that mutation. + s, _ := proxyWithGroupScopes(t) + rr = httptest.NewRecorder() + s.Handler().ServeHTTP(rr, noMaxTokensReq(groupScopesRequest(t, up, groupScopesValue(testGroupA, ",,,10", "")))) + if rr.Code != http.StatusOK { + t.Fatalf("undeclared request, group cap 10: status=%d body=%q, want 200 admitted (the reservation clamped to the cap)", rr.Code, strings.TrimSpace(rr.Body.String())) + } + + // The declared path is unchanged: max_tokens above the group cap can + // never be admitted — 400 without Retry-After, the group-scope twin of + // TestMaxTokensAboveGeneratedLimitIsUnsatisfiable400. + hits := 0 + backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + hits++ + _, _ = w.Write([]byte(`{"model":"model-a","usage":{"prompt_tokens":2,"completion_tokens":3}}`)) + })) + t.Cleanup(backend.Close) + declUp, err := url.Parse(backend.URL) + if err != nil { + t.Fatal(err) + } + rr = httptest.NewRecorder() + s.Handler().ServeHTTP(rr, groupScopesRequest(t, declUp, groupScopesValue(testGroupA, ",,,10", ""))) + if rr.Code != http.StatusBadRequest || rr.Header().Get("Retry-After") != "" || hits != 0 { + t.Fatalf("declared max_tokens 20 over group cap 10: status=%d Retry-After=%q hits=%d, want 400 without Retry-After before upstream", rr.Code, rr.Header().Get("Retry-After"), hits) + } + if got := strings.TrimSpace(rr.Body.String()); got != "max_tokens exceeds the per-window generated-token limit" { + t.Fatalf("body=%q", got) + } +} diff --git a/internal/proxy/proxy.go b/internal/proxy/proxy.go index 9dc1dbf..137d35c 100644 --- a/internal/proxy/proxy.go +++ b/internal/proxy/proxy.go @@ -576,7 +576,9 @@ func (s *Server) handleProxy(w http.ResponseWriter, r *http.Request) { // Contract limits therefore engage whenever Atlas stamped them, with or // without operator tiers (ruling R12 clarification). Under // admission.enabled=false the store itself is consulted only when a - // contract limit header is present (the Admit call below); a limit + // contract limit header is present (the Admit call below; a group + // envelope engages it only when a group scope carries a rate limit); + // a limit // header without the owner-id anchor still fails closed in // parseTrustedRateLimits. Only a shared request with NO envelope at all, // under admission.enabled=false, skips admission entirely: the @@ -603,6 +605,7 @@ func (s *Server) handleProxy(w http.ResponseWriter, r *http.Request) { tenantIdentity = "resource:" + id.ResourceID } var organizationLimits, ownerLimits admission.RateLimits + var groupScopes []admission.GroupScope var graph string if enforceAdmission { // A structurally broken policy — limit headers without their @@ -622,6 +625,17 @@ func (s *Server) handleProxy(w http.ResponseWriter, r *http.Request) { http.Error(w, "shared inference policy unavailable", http.StatusServiceUnavailable) return } + // The group quota envelope is a separate fail-closed parse: + // malformed, oversized, or over-count is a 503 with a loud log — + // the fail-visible default (the system never silently drops one + // group of a multi-group caller). Absent header → nil scopes. + var groupErr error + groupScopes, groupErr = parseTrustedGroupScopes(id) + if groupErr != nil { + s.log.Error.Printf("admission: invalid trusted group scope policy: %v request_id=%s", groupErr, requestID) + http.Error(w, "shared inference policy unavailable", http.StatusServiceUnavailable) + return + } graph = id.GraphK8sName if graph == "" { graph = graphFromUpstreamHost(upstream.Host) @@ -661,7 +675,11 @@ func (s *Server) handleProxy(w http.ResponseWriter, r *http.Request) { // default the client did not choose. The reservation is clamped // before prepareSharedDynamoRequest so the Dynamo output hint // still equals the reservation. - defaultOutput = undeclaredOutputReservation(defaultOutput, organizationLimits, ownerLimits) + groupLimits := make([]admission.RateLimits, len(groupScopes)) + for i, gs := range groupScopes { + groupLimits[i] = gs.Limits + } + defaultOutput = undeclaredOutputReservation(defaultOutput, append([]admission.RateLimits{organizationLimits, ownerLimits}, groupLimits...)...) } estimate, ok := admissionWork(body, defaultOutput) if !ok { @@ -690,14 +708,17 @@ func (s *Server) handleProxy(w http.ResponseWriter, r *http.Request) { } // Under admission.enabled=false a request whose envelope carries no // limit is unlimited in every scope, so it does not touch the store: no - // Admit, no lease renewal, no completion. Every fail-closed check above - // has already run. - if enforceAdmission && (s.settings.Admission.Enabled || organizationLimits.Any() || ownerLimits.Any()) { + // Admit, no lease renewal, no completion. A group envelope engages the + // store only when at least one group scope carries a rate limit; an + // all-unlimited envelope enforces + // nothing. Every fail-closed check above has already run. + if enforceAdmission && (s.settings.Admission.Enabled || organizationLimits.Any() || ownerLimits.Any() || anyGroupScopeEnforced(groupScopes)) { admitted, err = s.admitter.Admit(r.Context(), admission.Request{ Graph: graph, Organization: id.OrgID, Owner: id.OwnerID, Model: estimate.Model, PromptBytes: originalPromptBytes, EstimatedInputTokens: estimate.InputTokens, ReservedOutputTokens: estimate.OutputTokens, Adapter: id.Adapter != "", OrganizationLimits: organizationLimits, OwnerLimits: ownerLimits, + GroupScopes: groupScopes, }) if err != nil { if errors.Is(err, admission.ErrInvalidIdentity) {