diff --git a/CHANGELOG.md b/CHANGELOG.md index 030be4b..cb7864a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,8 @@ ## Unreleased +- Select scoped inference grants in the admission transaction using fenced, secretless authority projections; keep credential-owner materialization authoritative for revocation and retain legacy projection fallback. + - Collapse proxy-key admission into one canonical authority read and preserve atomic grant selection, migration fences and revocation checks; reuse compiled discovery metadata and instrument local authority RPC counts. - Add a dependency-free load generator and synthetic local subscription-pool stress harness with latency, disconnect and accounting diagnostics; avoid deep-copying prompts and tools during top-level request-field renames. diff --git a/docs/architecture.md b/docs/architecture.md index 21c933a..4af8cf4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -19,7 +19,7 @@ and result ordering instead of duplicating those decisions in client-side maps. Assignment rules, scoped upstream-grant secrets, and provider health remain outside that global authority because they have separate lifecycle and consistency needs. `GRANT_CREDENTIALS` is the canonical raw-secret owner, -sharded one Durable Object per grant. KV holds its redacted routing projection; +sharded one Durable Object per grant. The authority and KV hold its redacted routing projection; a legacy secret-bearing KV grant is imported and scrubbed on first use. Public HTTP Responses affinity uses authorization-scoped `ACCESS_CONTROL` @@ -147,11 +147,16 @@ usage event and delegates independent settlement and delivery to `accounting.ts` 3. Reserve the conservative budget before provider work. 4. Retain eligible LLM request content in R2 when policy requires it. Storage failure is fail-closed and prevents the upstream call. -5. Filter grants by the policy's provider allowlist, explicit grant eligibility, +5. For eligible proxy-key HTTP creates, admission filters and selects within one + synchronous authority transaction using owner-published routing projections. + Legacy projection rows retain the Worker KV/read-and-select path. + Filter grants by the policy's provider allowlist, explicit grant eligibility, cooldown, stale-state rule, and lowest active priority tier. The access authority atomically applies priority, round-robin, least-used, quota-aware, or weighted selection and records only counters and privacy-safe sticky input. -6. Materialize the selected credential from its per-grant owner. Expiring OAuth +6. Materialize the selected credential from its per-grant owner on every request. + Secret material is not cached: owner revocation must take effect even when + publication to the global authority fails. Expiring OAuth tokens refresh through that same serialized owner, which commits each rotated access/refresh pair and generation together. 7. Sign and forward the provider request, then normalize manifest-declared quota diff --git a/docs/hot-path-performance.md b/docs/hot-path-performance.md index a9aa375..5f48a21 100644 --- a/docs/hot-path-performance.md +++ b/docs/hot-path-performance.md @@ -461,3 +461,152 @@ unrelated compilation and system activity. The configured fake upstream is only isolate a response-path cause. It retained exactly two foreground authority RPCs per request and reconciled all 400 selections and usage events. No live provider benchmark or deployment was performed. + +## Admit-and-select follow-up (2026-10-11) + +Implemented work-order parts 1–2 on `perf/single-hop-inference`. Ordinary +proxy-key HTTP inference passes the prepared endpoint, transport, upstream model, +catalog model, capability and hashed session into admission. After any cold +imports, the authority rereads authentication, policy, connection and pool state, +filters candidates and updates selection counters in one synchronous SQLite +transaction. No network wait occurs in that transaction. The Worker consumes +that result once, only for the matching operation. Failed-provider exclusions +use a fresh combined admission, including current policy and connection checks. + +`worker/grant-eligibility.ts` is the shared owner of candidate filtering, routing +normalization, priority, credential usability and revision matching. Both paths +use the existing provider transport/configuration functions and the same +stateful authority selector. Provider logic and selection strategies are not +copied into a second implementation. The immutable bundled provider catalog +resolves the endpoint facts; no request content enters the authority. + +The additive `upstream_grant_projections` table stores an allowlisted, secretless +routing projection. The credential owner's existing attachment publication +writes it in the same transaction as membership/version changes, before the +existing KV publication. Generation and attachment revision accompany the +selected grant. An overlapping old owner publication without a projection marks +the row legacy, preventing a stale projection from surviving a version update. +This revision is publication provenance, **not** permission to cache credentials. + +A missing row triggers at most one bounded bulk read for the two pools and two +default keys (66 maximum). Each inserted row is its import marker, including +known-absent defaults. Import rereads the marker and attachment revision inside +a synchronous transaction, so concurrent owner publication, rotation or revoke +wins over delayed KV. A version mismatch or raw legacy secret marks the row +legacy and retains the old Worker KV + select path until owner publication +repairs it. Failed KV reads and missing metadata for an attached/versioned owner create no +marker and may retry; only an unowned default-key miss becomes a durable absent +marker. Old writers that +only change KV after the marker are not supported; use the credential owner. +The upgrade is additive; do not remove markers or restore old projections over +new owner versions. + +### Why materialization remains + +The credential cache escape hatch was taken. Revocation currently commits a +secretless tombstone even if the global authority is unavailable. Pause and +permanent OAuth-refresh rejection likewise commit owner state before repairing +its derived index. During a publication outage, the authority may still return +the old active generation. A cached token keyed by that generation would pass +the proposed check and bypass revocation. Blocking secret deletion on a global +index write would violate the existing owner safety contract. + +No secret cache was added. Every selected grant still reaches its credential +owner, which rejects revoked/disabled/reauthorization-required state and handles +near-expiry refresh. A successful rotation is visible in the next authority +projection; a failed publication is repaired or rejected at materialization, +never served from cached secret material. Policy and proxy-key disable are +checked afresh at admission. Requests admitted before a concurrent mutation +retain the existing in-flight semantics; already dispatched requests cannot be +retroactively revoked. + +Completing part 3 requires a separately designed pre-mutation invalidation +protocol, including index-unavailable revocation, recovery and overlapping old +writers. A TTL or matching generation alone does not solve this boundary. + +| Foreground path (excluding budgets and upstream) | Before | After | +| --- | --- | --- | +| Warm proxy-key HTTP inference | 2 authority + 1 KV bulk read + 1 credential owner | 1 authority + 1 credential owner | +| Cold authority projection, current owner metadata in KV | 2 authority + 1 Worker KV + 1 owner | 1 authority (internal bounded KV import) + 1 owner | +| Legacy/unimportable projection | 2 authority + 1 Worker KV + 1 owner | Same fallback; first read may attempt bounded import | +| Alternate after 401/403/429, projections ready | 2 authority + 1 KV + 1 owner | 1 fresh authority + 1 owner | +| Environment fallback, projections ready | admission, optional select, KV | 1 authority; no credential owner | +| Discovery, Access policy selection, pinned continuations, Fusion preflight, WebSocket creates | Existing paths | Existing paths | + +The full target of one authority call **plus upstream only is not claimed**. +No authorization, policy, pool, runtime, selection or credential secret is cached +between requests. + +The regression suite compares old and combined outcomes/counters across all six +strategies, three stickiness modes, priority tiers, reserve, model-scoped +exhaustion, cooldown, stale revisions/state, eligibility, exclusions and OpenAI +HTTP/WebSocket transport requirements. It checks warm hop counts, fresh failover, +legacy migration, bounded once-only backfill, concurrent revoke/rotation/tenant +changes, generation publication and secret exclusion. The publication-outage +regression explicitly observes an old active authority generation after owner +revocation, then proves materialization rejects it. + +Validation on the final implementation: + +```text +pnpm check exit 0 +Worker: tests 1301, pass 1301, fail 0 +Admin: tests 224, pass 224, fail 0 +Scripts: tests 361, pass 343, fail 0, skipped 18 +focused selection/continuation: tests 91, pass 91, fail 0 + +pnpm worker:e2e exit 0 +local Worker smoke passed +Responses WebSocket workerd fixture passed +contentCleanup: ok + +autoreview (P0–P2): scoped-clean +``` + +The review caught and fixed two stale-input races on legacy failover (tenant +scope and sticky input), plus a transient KV-miss import that could hide an +attached grant. Regression tests cover each. Existing cancellation/deadline +fault-injection hooks now target combined admission instead of the removed +standalone select call. An early full run was stopped after those old hooks +left a continuation test waiting forever; the corrected 43-test continuation +file and final full check pass. E2e's intentional accounting and cleanup failure +diagnostics remain expected. No live provider call or deployment was performed. + +The final synthetic load pass ran separately from tests: + +```sh +pnpm loadtest:local --requests 100 --concurrency 10,30 --max-tokens 16 \ + --out docs/performance/single-authority-selection-local.json \ + --label single-authority-selection +``` + +All 1,000 requests completed with **zero unexpected errors**. The 800 inference +requests had exactly 800 selections and 800 persisted usage events, including +104 intentional aborts. Each cell recorded exactly one foreground admission +RPC per request, with no standalone pool resolve/select RPC. Quota feedback is +reported separately (one background observation per inference). All seven +grants received quota updates; bounded recent-event checks found no session +using multiple grants. Clean credential materialization is still one separate +owner invocation, proved by the HTTP unit test; refresh/dirty-owner repair can +add provider-token and owner-to-authority publication work. + +| Scenario | Concurrency | TTFB p50 ms | Total p50 ms | Requests/s | +| --- | ---: | ---: | ---: | ---: | +| catalog | 10 | 30.7 | 31.2 | 316.5 | +| catalog | 30 | 75.8 | 76.3 | 342.6 | +| count | 10 | 45.5 | 46.3 | 82.6 | +| count | 30 | 74.3 | 76.6 | 196.0 | +| messages | 10 | 64.5 | 68.0 | 113.4 | +| messages | 30 | 314.6 | 315.1 | 63.4 | +| stream | 10 | 52.2 | 102.7 | 71.1 | +| stream | 30 | 146.1 | 221.6 | 124.3 | +| abort | 10 | 102.4 | 99.0 | 58.4 | +| abort | 30 | 87.1 | 92.7 | 237.6 | + +TTFB for streaming means first content delta. Abort latency samples include +different completion subsets. The fake provider waits 20 ms before content; +the same workerd process hosts the router and fixture. Host load averages near +the final pass were 40.64 / 48.28 / 44.55. These are indicative local results, +not an idle-host A/B or a production speedup claim. The coordinator should +compare the saved [raw results](performance/single-authority-selection-local.json) +against main on an idle host. diff --git a/docs/performance/single-authority-selection-local.json b/docs/performance/single-authority-selection-local.json new file mode 100644 index 0000000..d69d266 --- /dev/null +++ b/docs/performance/single-authority-selection-local.json @@ -0,0 +1,831 @@ +{ + "label": "single-authority-selection", + "environment": { + "node": "v26.11.1", + "platform": "darwin", + "arch": "arm64", + "cpus": 32 + }, + "delayMs": 20, + "maxTokens": 16, + "cells": [ + { + "scenario": "catalog", + "concurrency": 10, + "completed": 100, + "seconds": 0.31596325000000003, + "throughput": 316.49250347943945, + "dropped": 0, + "errors": 0, + "intentionalAborts": 0, + "totalMs": { + "samples": 100, + "p50": 31.174332999999933, + "p90": 34.9861249999999, + "p95": 35.4447090000001, + "p99": 36.353624999999965, + "max": 42.81787499999996 + }, + "ttfbMs": { + "samples": 100, + "p50": 30.674584000000095, + "p90": 34.81224999999995, + "p95": 35.306542000000036, + "p99": 36.238124999999854, + "max": 42.51741700000002 + }, + "firstByteMs": { + "samples": 100, + "p50": 30.674584000000095, + "p90": 34.81224999999995, + "p95": 35.306542000000036, + "p99": 36.238124999999854, + "max": 42.51741700000002 + }, + "statuses": { + "200": 100 + }, + "errorsByCode": { + "none": 100 + }, + "lanes": { + "none": 100 + }, + "failovers": 0, + "stickiness": { + "note": "No grant identity header exposed; consistency cannot be inferred from lane alone." + }, + "authorityRpc": { + "/admission/resolve": 100 + }, + "routerAddedP50Ms": null, + "bookkeeping": { + "expected": 0, + "selected": 0, + "usage": 0, + "initialBacklog": 0, + "drainMs": 1.715708000000177, + "upstream": { + "requests": 0, + "streams": 0, + "count": 0, + "messages": 0, + "canceled": 0, + "completed": 0 + }, + "recentEventsChecked": 0, + "recentEventStatuses": {}, + "recentTokenTotals": { + "input": 0, + "output": 0 + }, + "sessionsChecked": 0, + "sessionsWithMultipleGrants": 0, + "quotaUpdatedGrants": 0 + }, + "workerd": { + "peakRssMiB": 138.375, + "peakCpuPercent": 76.7 + } + }, + { + "scenario": "catalog", + "concurrency": 30, + "completed": 100, + "seconds": 0.29187716600000024, + "throughput": 342.60987719745066, + "dropped": 0, + "errors": 0, + "intentionalAborts": 0, + "totalMs": { + "samples": 100, + "p50": 76.27125000000024, + "p90": 110.33283299999994, + "p95": 140.14195800000016, + "p99": 145.73616700000002, + "max": 145.78491699999995 + }, + "ttfbMs": { + "samples": 100, + "p50": 75.78333299999986, + "p90": 109.9491250000001, + "p95": 139.16449999999986, + "p99": 145.20249999999987, + "max": 145.34283300000016 + }, + "firstByteMs": { + "samples": 100, + "p50": 75.78333299999986, + "p90": 109.9491250000001, + "p95": 139.16449999999986, + "p99": 145.20249999999987, + "max": 145.34283300000016 + }, + "statuses": { + "200": 100 + }, + "errorsByCode": { + "none": 100 + }, + "lanes": { + "none": 100 + }, + "failovers": 0, + "stickiness": { + "note": "No grant identity header exposed; consistency cannot be inferred from lane alone." + }, + "authorityRpc": { + "/admission/resolve": 100 + }, + "routerAddedP50Ms": null, + "bookkeeping": { + "expected": 0, + "selected": 0, + "usage": 0, + "initialBacklog": 0, + "drainMs": 2.088792000000012, + "upstream": { + "requests": 0, + "streams": 0, + "count": 0, + "messages": 0, + "canceled": 0, + "completed": 0 + }, + "recentEventsChecked": 0, + "recentEventStatuses": {}, + "recentTokenTotals": { + "input": 0, + "output": 0 + }, + "sessionsChecked": 0, + "sessionsWithMultipleGrants": 0, + "quotaUpdatedGrants": 0 + }, + "workerd": { + "peakRssMiB": 188.34375, + "peakCpuPercent": 74.9 + } + }, + { + "scenario": "count", + "concurrency": 10, + "completed": 100, + "seconds": 1.2112012080000003, + "throughput": 82.56266534370891, + "dropped": 0, + "errors": 0, + "intentionalAborts": 0, + "totalMs": { + "samples": 100, + "p50": 46.26495799999975, + "p90": 164.64800000000014, + "p95": 650.6310830000002, + "p99": 651.809541999999, + "max": 652.3845829999991 + }, + "ttfbMs": { + "samples": 100, + "p50": 45.47479199999907, + "p90": 163.8361670000004, + "p95": 650.1191250000011, + "p99": 650.2120840000007, + "max": 650.3073330000007 + }, + "firstByteMs": { + "samples": 100, + "p50": 45.47479199999907, + "p90": 163.8361670000004, + "p95": 650.1191250000011, + "p99": 650.2120840000007, + "max": 650.3073330000007 + }, + "statuses": { + "200": 100 + }, + "errorsByCode": { + "none": 100 + }, + "lanes": { + "subscription": 100 + }, + "failovers": 0, + "stickiness": { + "note": "No grant identity header exposed; consistency cannot be inferred from lane alone." + }, + "authorityRpc": { + "/admission/resolve": 100, + "/grant-pools/feedback": 100 + }, + "routerAddedP50Ms": 25.47479199999907, + "bookkeeping": { + "expected": 100, + "selected": 100, + "usage": 100, + "initialBacklog": 0, + "drainMs": 3.746624999999767, + "upstream": { + "requests": 100, + "streams": 0, + "count": 100, + "messages": 0, + "canceled": 0, + "completed": 100 + }, + "recentEventsChecked": 100, + "recentEventStatuses": { + "success": 100 + }, + "recentTokenTotals": { + "input": 0, + "output": 0 + }, + "sessionsChecked": 70, + "sessionsWithMultipleGrants": 0, + "quotaUpdatedGrants": 7 + }, + "workerd": { + "peakRssMiB": 226.515625, + "peakCpuPercent": 22.1 + } + }, + { + "scenario": "count", + "concurrency": 30, + "completed": 100, + "seconds": 0.5102750830000005, + "throughput": 195.97272790997695, + "dropped": 0, + "errors": 0, + "intentionalAborts": 0, + "totalMs": { + "samples": 100, + "p50": 76.55262500000026, + "p90": 315.93154200000026, + "p95": 316.4247919999998, + "p99": 316.6394170000003, + "max": 316.7752079999991 + }, + "ttfbMs": { + "samples": 100, + "p50": 74.34637499999917, + "p90": 315.5674589999999, + "p95": 316.04712500000096, + "p99": 316.3105420000011, + "max": 316.39145800000006 + }, + "firstByteMs": { + "samples": 100, + "p50": 74.34637499999917, + "p90": 315.5674589999999, + "p95": 316.04712500000096, + "p99": 316.3105420000011, + "max": 316.39145800000006 + }, + "statuses": { + "200": 100 + }, + "errorsByCode": { + "none": 100 + }, + "lanes": { + "subscription": 100 + }, + "failovers": 0, + "stickiness": { + "note": "No grant identity header exposed; consistency cannot be inferred from lane alone." + }, + "authorityRpc": { + "/admission/resolve": 100, + "/grant-pools/feedback": 100 + }, + "routerAddedP50Ms": 54.34637499999917, + "bookkeeping": { + "expected": 100, + "selected": 100, + "usage": 100, + "initialBacklog": 0, + "drainMs": 4.07445799999914, + "upstream": { + "requests": 100, + "streams": 0, + "count": 100, + "messages": 0, + "canceled": 0, + "completed": 100 + }, + "recentEventsChecked": 100, + "recentEventStatuses": { + "success": 100 + }, + "recentTokenTotals": { + "input": 0, + "output": 0 + }, + "sessionsChecked": 70, + "sessionsWithMultipleGrants": 0, + "quotaUpdatedGrants": 7 + }, + "workerd": { + "peakRssMiB": 258.28125, + "peakCpuPercent": 32.1 + } + }, + { + "scenario": "messages", + "concurrency": 10, + "completed": 100, + "seconds": 0.8820086669999982, + "throughput": 113.37757069908724, + "dropped": 0, + "errors": 0, + "intentionalAborts": 0, + "totalMs": { + "samples": 100, + "p50": 67.95233299999745, + "p90": 99.11549999999988, + "p95": 252.71045799999774, + "p99": 266.1582920000001, + "max": 266.30375000000276 + }, + "ttfbMs": { + "samples": 100, + "p50": 64.54766699999891, + "p90": 98.69329199999993, + "p95": 251.59741700000086, + "p99": 265.18987500000003, + "max": 265.3599170000016 + }, + "firstByteMs": { + "samples": 100, + "p50": 64.54766699999891, + "p90": 98.69329199999993, + "p95": 251.59741700000086, + "p99": 265.18987500000003, + "max": 265.3599170000016 + }, + "statuses": { + "200": 100 + }, + "errorsByCode": { + "none": 100 + }, + "lanes": { + "subscription": 100 + }, + "failovers": 0, + "stickiness": { + "note": "No grant identity header exposed; consistency cannot be inferred from lane alone." + }, + "authorityRpc": { + "/admission/resolve": 100, + "/grant-pools/feedback": 100 + }, + "routerAddedP50Ms": 44.54766699999891, + "bookkeeping": { + "expected": 100, + "selected": 100, + "usage": 100, + "initialBacklog": 0, + "drainMs": 14.511834000000817, + "upstream": { + "requests": 100, + "streams": 0, + "count": 0, + "messages": 100, + "canceled": 0, + "completed": 100 + }, + "recentEventsChecked": 100, + "recentEventStatuses": { + "success": 100 + }, + "recentTokenTotals": { + "input": 1200, + "output": 400 + }, + "sessionsChecked": 70, + "sessionsWithMultipleGrants": 0, + "quotaUpdatedGrants": 7 + }, + "workerd": { + "peakRssMiB": 288.21875, + "peakCpuPercent": 58.1 + } + }, + { + "scenario": "messages", + "concurrency": 30, + "completed": 100, + "seconds": 1.578517291, + "throughput": 63.35058891667218, + "dropped": 0, + "errors": 0, + "intentionalAborts": 0, + "totalMs": { + "samples": 100, + "p50": 315.1411670000016, + "p90": 869.1427500000027, + "p95": 869.895875000002, + "p99": 870.7925829999986, + "max": 870.9056659999987 + }, + "ttfbMs": { + "samples": 100, + "p50": 314.5932079999984, + "p90": 867.6343749999978, + "p95": 868.2468749999971, + "p99": 868.6752919999999, + "max": 869.0369160000009 + }, + "firstByteMs": { + "samples": 100, + "p50": 314.5932079999984, + "p90": 867.6343749999978, + "p95": 868.2468749999971, + "p99": 868.6752919999999, + "max": 869.0369160000009 + }, + "statuses": { + "200": 100 + }, + "errorsByCode": { + "none": 100 + }, + "lanes": { + "subscription": 100 + }, + "failovers": 0, + "stickiness": { + "note": "No grant identity header exposed; consistency cannot be inferred from lane alone." + }, + "authorityRpc": { + "/admission/resolve": 100, + "/grant-pools/feedback": 100 + }, + "routerAddedP50Ms": 294.5932079999984, + "bookkeeping": { + "expected": 100, + "selected": 100, + "usage": 100, + "initialBacklog": 0, + "drainMs": 5.404499999996915, + "upstream": { + "requests": 100, + "streams": 0, + "count": 0, + "messages": 100, + "canceled": 0, + "completed": 100 + }, + "recentEventsChecked": 100, + "recentEventStatuses": { + "success": 100 + }, + "recentTokenTotals": { + "input": 1200, + "output": 400 + }, + "sessionsChecked": 70, + "sessionsWithMultipleGrants": 0, + "quotaUpdatedGrants": 7 + }, + "workerd": { + "peakRssMiB": 335.296875, + "peakCpuPercent": 44.5 + } + }, + { + "scenario": "stream", + "concurrency": 10, + "completed": 100, + "seconds": 1.4058468750000028, + "throughput": 71.13150214172492, + "dropped": 0, + "errors": 0, + "intentionalAborts": 0, + "totalMs": { + "samples": 100, + "p50": 102.7443329999951, + "p90": 236.88650000000052, + "p95": 237.90645799999766, + "p99": 354.6710000000021, + "max": 376.16370800000004 + }, + "ttfbMs": { + "samples": 100, + "p50": 52.24304200000188, + "p90": 193.2467919999981, + "p95": 222.96658300000126, + "p99": 224.1194170000017, + "max": 224.8517920000013 + }, + "firstByteMs": { + "samples": 100, + "p50": 27.023583000001963, + "p90": 117.2546250000014, + "p95": 166.7119160000002, + "p99": 172.93054099999426, + "max": 188.7820419999989 + }, + "statuses": { + "200": 100 + }, + "errorsByCode": { + "none": 100 + }, + "lanes": { + "subscription": 100 + }, + "failovers": 0, + "stickiness": { + "note": "No grant identity header exposed; consistency cannot be inferred from lane alone." + }, + "authorityRpc": { + "/admission/resolve": 100, + "/grant-pools/feedback": 100 + }, + "routerAddedP50Ms": 32.24304200000188, + "bookkeeping": { + "expected": 100, + "selected": 100, + "usage": 100, + "initialBacklog": 0, + "drainMs": 6.108542000001762, + "upstream": { + "requests": 100, + "streams": 100, + "count": 0, + "messages": 0, + "canceled": 0, + "completed": 100 + }, + "recentEventsChecked": 100, + "recentEventStatuses": { + "success": 100 + }, + "recentTokenTotals": { + "input": 1200, + "output": 400 + }, + "sessionsChecked": 70, + "sessionsWithMultipleGrants": 0, + "quotaUpdatedGrants": 7 + }, + "workerd": { + "peakRssMiB": 364.015625, + "peakCpuPercent": 49.3 + } + }, + { + "scenario": "stream", + "concurrency": 30, + "completed": 100, + "seconds": 0.804565208, + "throughput": 124.2907336853174, + "dropped": 0, + "errors": 0, + "intentionalAborts": 0, + "totalMs": { + "samples": 100, + "p50": 221.57337499999994, + "p90": 279.7315830000007, + "p95": 280.1678749999992, + "p99": 280.6806250000009, + "max": 280.7899170000019 + }, + "ttfbMs": { + "samples": 100, + "p50": 146.07687499999884, + "p90": 174.90695799999958, + "p95": 175.01070900000195, + "p99": 175.29733300000225, + "max": 175.31104200000118 + }, + "firstByteMs": { + "samples": 100, + "p50": 120.64470799999981, + "p90": 171.328042000001, + "p95": 171.64495799999713, + "p99": 171.91120900000533, + "max": 171.9135000000024 + }, + "statuses": { + "200": 100 + }, + "errorsByCode": { + "none": 100 + }, + "lanes": { + "subscription": 100 + }, + "failovers": 0, + "stickiness": { + "note": "No grant identity header exposed; consistency cannot be inferred from lane alone." + }, + "authorityRpc": { + "/admission/resolve": 100, + "/grant-pools/feedback": 100 + }, + "routerAddedP50Ms": 126.07687499999884, + "bookkeeping": { + "expected": 100, + "selected": 100, + "usage": 100, + "initialBacklog": 0, + "drainMs": 7.570749999998952, + "upstream": { + "requests": 100, + "streams": 100, + "count": 0, + "messages": 0, + "canceled": 0, + "completed": 100 + }, + "recentEventsChecked": 100, + "recentEventStatuses": { + "success": 100 + }, + "recentTokenTotals": { + "input": 1200, + "output": 400 + }, + "sessionsChecked": 70, + "sessionsWithMultipleGrants": 0, + "quotaUpdatedGrants": 7 + }, + "workerd": { + "peakRssMiB": 375.1875, + "peakCpuPercent": 60.9 + } + }, + { + "scenario": "abort", + "concurrency": 10, + "completed": 100, + "seconds": 1.7120564159999994, + "throughput": 58.40929017610132, + "dropped": 0, + "errors": 0, + "intentionalAborts": 53, + "totalMs": { + "samples": 100, + "p50": 99.02416700000322, + "p90": 320.28391700000066, + "p95": 702.8609170000054, + "p99": 730.3696249999994, + "max": 737.7832079999935 + }, + "ttfbMs": { + "samples": 47, + "p50": 102.35995800000092, + "p90": 610.4199579999986, + "p95": 618.6156249999985, + "p99": 627.7165409999943, + "max": 627.7165409999943 + }, + "firstByteMs": { + "samples": 100, + "p50": 58.54374999999709, + "p90": 126.03679199999897, + "p95": 226.86499999999796, + "p99": 646.5820000000022, + "max": 648.1213329999955 + }, + "statuses": { + "200": 100 + }, + "errorsByCode": { + "intentional_abort": 53, + "none": 47 + }, + "lanes": { + "subscription": 100 + }, + "failovers": 0, + "stickiness": { + "note": "No grant identity header exposed; consistency cannot be inferred from lane alone." + }, + "authorityRpc": { + "/admission/resolve": 100, + "/grant-pools/feedback": 100 + }, + "routerAddedP50Ms": null, + "bookkeeping": { + "expected": 100, + "selected": 100, + "usage": 100, + "initialBacklog": 0, + "drainMs": 13.687374999994063, + "upstream": { + "requests": 100, + "streams": 100, + "count": 0, + "messages": 0, + "canceled": 0, + "completed": 62 + }, + "recentEventsChecked": 100, + "recentEventStatuses": { + "success": 47, + "client_error": 53 + }, + "recentTokenTotals": { + "input": 564, + "output": 188 + }, + "sessionsChecked": 70, + "sessionsWithMultipleGrants": 0, + "quotaUpdatedGrants": 7 + }, + "workerd": { + "peakRssMiB": 380.171875, + "peakCpuPercent": 45.5 + } + }, + { + "scenario": "abort", + "concurrency": 30, + "completed": 100, + "seconds": 0.42087750000000235, + "throughput": 237.59882626179694, + "dropped": 0, + "errors": 0, + "intentionalAborts": 51, + "totalMs": { + "samples": 100, + "p50": 92.72737499999494, + "p90": 173.564750000005, + "p95": 175.17399999999907, + "p99": 175.484583999998, + "max": 194.8707499999946 + }, + "ttfbMs": { + "samples": 49, + "p50": 87.13700000000244, + "p90": 122.00850000000355, + "p95": 122.04537500000151, + "p99": 127.76037499999802, + "max": 127.76037499999802 + }, + "firstByteMs": { + "samples": 100, + "p50": 59.41049999999814, + "p90": 82.8202919999967, + "p95": 83.60695900000428, + "p99": 92.53066699999908, + "max": 96.5315830000036 + }, + "statuses": { + "200": 100 + }, + "errorsByCode": { + "intentional_abort": 51, + "none": 49 + }, + "lanes": { + "subscription": 100 + }, + "failovers": 0, + "stickiness": { + "note": "No grant identity header exposed; consistency cannot be inferred from lane alone." + }, + "authorityRpc": { + "/admission/resolve": 100, + "/grant-pools/feedback": 100 + }, + "routerAddedP50Ms": null, + "bookkeeping": { + "expected": 100, + "selected": 100, + "usage": 100, + "initialBacklog": 0, + "drainMs": 7.975500000000466, + "upstream": { + "requests": 100, + "streams": 100, + "count": 0, + "messages": 0, + "canceled": 0, + "completed": 49 + }, + "recentEventsChecked": 100, + "recentEventStatuses": { + "success": 49, + "client_error": 51 + }, + "recentTokenTotals": { + "input": 588, + "output": 196 + }, + "sessionsChecked": 70, + "sessionsWithMultipleGrants": 0, + "quotaUpdatedGrants": 7 + }, + "workerd": { + "peakRssMiB": 386.90625, + "peakCpuPercent": 69.4 + } + } + ] +} diff --git a/worker/authority-admission.ts b/worker/authority-admission.ts index d344de3..9521607 100644 --- a/worker/authority-admission.ts +++ b/worker/authority-admission.ts @@ -1,10 +1,17 @@ import type { AccessControlUser, AccessPolicyEntry, AccessUserRecord, Env, GrantRuntimeState, ProviderConnection, ProxyCredentialEntry } from "./types"; +import type { GrantSelectionResult } from "./grant-eligibility.ts"; import { normalizeEmail, safeEqual } from "./utils.ts"; export interface AdmissionProvider { id: string; tokenRef: string } -export interface AdmissionRequest { credentialId: string; secretSha256?: string; providers?: AdmissionProvider[]; allConnections?: boolean } +export interface AdmissionSelection { + providerId: string; endpointId: string; mode: "http" | "websocket"; + model?: string | null; catalogModelId?: string | null; capability: string; + sessionHash?: string; excludedKeys?: string[]; +} +export interface AdmissionRequest { selection?: AdmissionSelection; credentialId: string; secretSha256?: string; providers?: AdmissionProvider[]; allConnections?: boolean } export interface AdmissionPool { keys: string[]; states: Record; hasAttachment: boolean; unavailable?: true } export interface AdmissionResult { + selection?: GrantSelectionResult; credential: ProxyCredentialEntry | null; policy: AccessPolicyEntry | null; owner: AccessControlUser | null; @@ -16,6 +23,7 @@ export interface AdmissionResult { } interface AdmissionStore { + finishSelection?(input: AdmissionRequest, result: AdmissionResult): Promise; getCredential(id: string): ProxyCredentialEntry | null; getPolicy(id: string): AccessPolicyEntry | null; getUser(email: string): AccessControlUser | null; @@ -97,7 +105,7 @@ export async function resolveAdmission(env: Env, input: AdmissionRequest, store: try { result.pools[provider.id] = store.resolveGrantPool({ providerId: provider.id, policyId, tenantId: policy.policy.tenantId ?? "default", defaultKeys: [`oauth/${policyId}/${provider.tokenRef}`, `oauth/tenants/${policy.policy.tenantId ?? "default"}/${provider.tokenRef}`] }); } catch { result.pools[provider.id] = { keys: [], states: {}, hasAttachment: false, unavailable: true }; } } - return result; + return input.selection && store.finishSelection ? store.finishSelection(input, result) : result; } export async function genuineLegacyKey(env: Env, id: string): Promise | null> { diff --git a/worker/authority-grant-selection.ts b/worker/authority-grant-selection.ts new file mode 100644 index 0000000..3a3b8c5 --- /dev/null +++ b/worker/authority-grant-selection.ts @@ -0,0 +1,84 @@ +import snapshotJson from "./generated/provider-snapshot.json" with { type: "json" }; +import type { AccessControlUser, AccessPolicyEntry, Env, ProviderConnection, ProviderSnapshot, ProxyCredentialEntry, UpstreamGrant } from "./types"; +import type { AdmissionPool, AdmissionRequest, AdmissionResult } from "./authority-admission.ts"; +import type { GrantAttachmentSnapshot } from "./authority.ts"; +import { routingProjection } from "./grant-projection.ts"; +import { activeOperationCandidates, filterGrantCandidates, currentGrantRuntime, grantRoutingPolicy, grantWeight } from "./grant-eligibility.ts"; +import { grantQuotaRatio } from "./grant-quota.ts"; +import { assertPolicyModel } from "./policy-models.ts"; +import { normalizeEmail, safeEqual, sha256Hex } from "./utils.ts"; + +interface SelectionStore { + getCredential(id: string): ProxyCredentialEntry | null; + getPolicy(id: string): AccessPolicyEntry | null; + getUser(email: string): AccessControlUser | null; + resolveConnections(ids: string[]): { connections: ProviderConnection[] }; + resolveGrantPool(input: { providerId: string; policyId: string; tenantId: string; defaultKeys: string[] }): AdmissionPool; + grantAttachment(key: string): GrantAttachmentSnapshot; + selectGrant(input: { poolKey: string; strategy: string; stickyHash: string | null; switchAtUsedPercent: number; hysteresisPercent: number; candidates: Array<{ key: string; weight: number; remainingRatio: number | null }> }): { selectedKey: string }; +} +function rows(cursor: Iterable): T[] { return Array.from(cursor); } + +export async function finishAdmissionSelection(env: Env, storage: DurableObjectStorage, input: AdmissionRequest, result: AdmissionResult, store: SelectionStore): Promise { + const sql = storage.sql; + const request = input.selection!; + const provider = (snapshotJson as unknown as ProviderSnapshot).providers.find(item => item.id === request.providerId); + const endpoint = provider?.endpoints.find(item => item.id === request.endpointId); + if (!provider || !endpoint || !result.policy || result.connectionsUnavailable || result.pools[provider.id]?.unavailable) return result; + const poolInput = { providerId: provider.id, policyId: result.policy.policyId, tenantId: result.policy.policy.tenantId ?? "default", defaultKeys: [`oauth/${result.policy.policyId}/${provider.auth.schemes.find(scheme => scheme.type === "oauth")?.tokenRef ?? provider.id}`, `oauth/tenants/${result.policy.policy.tenantId ?? "default"}/${provider.auth.schemes.find(scheme => scheme.type === "oauth")?.tokenRef ?? provider.id}`] }; + const keys = store.resolveGrantPool(poolInput).keys; + // One bounded import attempt per key. Recheck both the row marker and the + // attachment revision after KV yields; an owner publication always wins. + const missing = keys.filter(key => !rows(sql.exec("SELECT 1 FROM upstream_grant_projections WHERE grant_key = ?", key)).length); + const versions = new Map(missing.map(key => [key, store.grantAttachment(key).revision])); + let backfillFailed = false; + if (missing.length) { + try { + const grants = await env.POLICY_KV.get(missing, "json"); + storage.transactionSync(() => { + for (const key of missing) { + if (rows(sql.exec("SELECT 1 FROM upstream_grant_projections WHERE grant_key = ?", key)).length || store.grantAttachment(key).revision !== versions.get(key)) continue; + const grant = grants.get(key); + const version = store.grantAttachment(key); + if (!grant && (version.attached || version.pending || version.generation > 0)) continue; + const legacy = !!grant && (grant.credentialStore !== "durable_object" || grant.credentialGeneration !== version.generation || version.pending); + sql.exec("INSERT INTO upstream_grant_projections (grant_key, projection_json, legacy) VALUES (?, ?, ?)", key, grant && !legacy ? JSON.stringify(routingProjection(grant)) : null, legacy ? 1 : 0); + } + }); + } catch { backfillFailed = true; } // Migration failures retain the old read path. + } + const identity = result.credential!.credential.principalId ?? result.credential!.credentialId; + const [identityHash, fallbackSessionHash] = await Promise.all([sha256Hex(`identity:${identity}`), sha256Hex(`session:${identity}`)]); + return storage.transactionSync(() => { + // Imports/hashing yielded. Reread authorization and the entire pool before + // filtering and updating counters in this same synchronous transaction. + const credential = store.getCredential(input.credentialId); + const policy = credential ? store.getPolicy(credential.credential.policyId) : null; + const owner = credential?.credential.principalId ? store.getUser(normalizeEmail(credential.credential.principalId) ?? credential.credential.principalId) : null; + const fresh: AdmissionResult = { ...result, credential, policy, owner }; + if (!credential || !policy || !credential.credential.enabled || !policy.policy.enabled || credential.credential.policyGeneration !== policy.policy.generation || owner?.record.enabled === false || !input.secretSha256 || !safeEqual(input.secretSha256, credential.credential.secretSha256)) return fresh; + if (credential.credential.principalId !== result.credential!.credential.principalId || policy.policyId !== poolInput.policyId || (policy.policy.tenantId ?? "default") !== poolInput.tenantId) return { ...fresh, pools: {} }; + fresh.connections = store.resolveConnections((input.providers ?? []).map(item => item.id)).connections; + if (policy.policy.providers.length && !policy.policy.providers.includes(provider.id) || fresh.connections.find(item => item.providerId === provider.id)?.enabled === false) return fresh; + try { assertPolicyModel(policy.policy, { model: request.catalogModelId ? { id: request.catalogModelId } : null, endpoint, capability: request.capability }); } catch { return fresh; } + const pool = store.resolveGrantPool(poolInput); + fresh.pools = { ...fresh.pools, [provider.id]: pool }; + if (backfillFailed) return fresh; + const candidates = []; + for (const key of pool.keys) { + const row = rows<{ projection_json: string | null; legacy: number }>(sql.exec("SELECT projection_json, legacy FROM upstream_grant_projections WHERE grant_key = ?", key))[0]; + if (!row || row.legacy) return fresh; + const grant = row.projection_json ? JSON.parse(row.projection_json) as UpstreamGrant : null; + candidates.push({ key, grant, runtimeState: grant ? currentGrantRuntime(grant, pool.states[key]) : null }); + } + const routing = grantRoutingPolicy(policy.policy.grantRouting), now = Date.now(); + const requirement = { provider, endpoint, mode: request.mode, model: request.model }; + const { available, hasConfiguredGrant } = filterGrantCandidates(provider.id, candidates, new Set(request.excludedKeys ?? []), routing, undefined, requirement, now); + const active = activeOperationCandidates(available, env, requirement); + const stickyHash = routing.stickiness === "identity" ? identityHash : routing.stickiness === "session" ? request.sessionHash ?? fallbackSessionHash : null; + const choice = active.length ? store.selectGrant({ poolKey: `${encodeURIComponent(provider.id)}/${encodeURIComponent(policy.policyId)}/${encodeURIComponent(poolInput.tenantId)}`, strategy: routing.strategy, stickyHash, switchAtUsedPercent: routing.switchAtUsedPercent, hysteresisPercent: routing.hysteresisPercent, candidates: active.map(item => ({ key: item.key, weight: grantWeight(item.grant), remainingRatio: grantQuotaRatio(item.runtimeState, now, routing.staleAfterSeconds * 1000, request.model) })) }) : null; + const selected = active.find(item => item.key === choice?.selectedKey) ?? null; + return { ...fresh, selection: { selected: selected ? { ...selected, authorityRevision: store.grantAttachment(selected.key).revision } : null, hasConfiguredGrant } }; + }); +} + diff --git a/worker/authority.ts b/worker/authority.ts index 634c4bb..851651d 100644 --- a/worker/authority.ts +++ b/worker/authority.ts @@ -1,7 +1,8 @@ +import { finishAdmissionSelection } from "./authority-grant-selection.ts"; import { mergeGrantRuntimeState, normalizeGrantRuntimeState } from "./grant-runtime.ts"; import type { AccessControlUser, AccessPolicyEntry, AccessUserRecord, Env, OAuthState, PolicyBinding, - AssignmentState, GrantRoutingPolicy, GrantRuntimeState, ProviderConnection, ProxyCredentialEntry, + UpstreamGrant, AssignmentState, GrantRoutingPolicy, GrantRuntimeState, ProviderConnection, ProxyCredentialEntry, } from "./types"; import { evaluateUserAssignments, withLegacyAssignmentState, type AssignmentEvidence, type AssignmentRuleEntry } from "./assignment-evaluator.ts"; import type { AuthorizationSnapshot, AuthorizationSnapshotRequest, CredentialMutation, CredentialMutationResult } from "./authority-contracts.ts"; @@ -9,6 +10,7 @@ import { contentRetentionDefault } from "./content-retention.ts"; import { normalizeConnectionMutation, type ProviderConnectionMutation } from "./provider-connections.ts"; import type { ResponsesScopeStore } from "./responses-scope.ts"; import { boundedPercent, selectThresholdGrantKey, stickyScore, weightedRandom } from "./grant-selection-strategies.ts"; +import { publishRoutingProjection } from "./grant-projection.ts"; import { errorResponse, HttpError, json, normalizeEmail, readJson, safeEqual } from "./utils.ts"; type Principal = { principalType: "user" | "group"; principalId: string }; @@ -44,7 +46,11 @@ export class PolicyBindingIndexObject implements DurableObject { seedCredential: (id, value) => this.initializeCredentials([{ credentialId: id, credential: normalizeCredentialRecord(value, id) }]), seedPolicy: (id, value) => this.initializePolicies([{ policyId: id, policy: normalizePolicyRecord(value, contentRetentionDefault(this.env)) }]), initializeUsers: users => this.initializeUsers(users), initializeConnections: connections => this.initializeConnections(connections), - resolveConnections: ids => this.resolveConnections(ids), putMeta: key => this.putMeta(key), resolveGrantPool: input => this.resolveGrantPool(input), + resolveConnections: ids => this.resolveConnections(ids), putMeta: key => this.putMeta(key), resolveGrantPool: input => this.resolveGrantPool(input), finishSelection: (input, result) => finishAdmissionSelection(this.env, this.storage, input, result, { + getCredential: id => this.getCredential(id), getPolicy: id => this.getPolicy(id), getUser: email => this.getUser(email), + resolveConnections: ids => this.resolveConnections(ids), resolveGrantPool: request => this.resolveGrantPool(request), + grantAttachment: key => this.grantAttachment(key), selectGrant: request => this.selectGrant(request), + }), })); if (path === "/resolve") return json({ initialized: this.hasMeta("bindings_global_initialized"), ...this.resolveBindings((await readJson<{ principals: Principal[] }>(request)).principals) }); if (path === "/initialize") { this.initializeBindings(await readJson(request)); return new Response("initialized"); } @@ -121,6 +127,7 @@ export class PolicyBindingIndexObject implements DurableObject { // Detach retains this fence, so delayed publication cannot resurrect it. // Revision also fences admission/repair while owner generation is unchanged. this.sql.exec("CREATE TABLE IF NOT EXISTS upstream_grant_pool_versions (grant_key TEXT PRIMARY KEY, generation INTEGER NOT NULL, revision INTEGER NOT NULL)"); + this.sql.exec("CREATE TABLE IF NOT EXISTS upstream_grant_projections (grant_key TEXT PRIMARY KEY, projection_json TEXT, legacy INTEGER NOT NULL DEFAULT 0)"); this.sql.exec("CREATE TABLE IF NOT EXISTS upstream_grant_runtime (grant_key TEXT PRIMARY KEY, state_json TEXT NOT NULL)"); this.sql.exec("CREATE TABLE IF NOT EXISTS upstream_grant_selection (pool_key TEXT NOT NULL, grant_key TEXT NOT NULL, selected_count INTEGER NOT NULL, last_selected_ms INTEGER NOT NULL, PRIMARY KEY (pool_key, grant_key))"); this.sql.exec("CREATE TABLE IF NOT EXISTS upstream_grant_pool_cursors (pool_key TEXT PRIMARY KEY, grant_key TEXT NOT NULL)"); @@ -430,6 +437,7 @@ export class PolicyBindingIndexObject implements DurableObject { if (status ? !member || member.status === status && !others : !prepared.attached) { // Retire the consumed reservation even when owner state is unchanged. if (restored) this.storeAttachmentVersion(input.key, input.generation, current.revision + 1); + publishRoutingProjection(this.sql, input); return { ...this.grantAttachment(input.key), outcome }; } throw new HttpError(409, "grant_attachment_changed", "same-generation attachment publication differs from its committed state"); @@ -442,6 +450,7 @@ export class PolicyBindingIndexObject implements DurableObject { } else this.sql.exec("DELETE FROM upstream_grant_pool_members WHERE scope = ? AND scope_id = ? AND token_ref = ? AND status IN ('pending', 'pending_inactive')", ...identity); if (status !== "active") this.sql.exec("DELETE FROM upstream_grant_runtime WHERE grant_key = ?", input.key); this.storeAttachmentVersion(input.key, input.generation, current.revision + 1); + publishRoutingProjection(this.sql, input); return { ...this.grantAttachment(input.key), outcome }; }); } @@ -631,7 +640,7 @@ interface GrantPoolResolveRequest { policyId: string; tenantId: string; provider export type GrantAttachmentStatus = "active" | "paused" | "reauth_required"; export interface GrantAttachmentSnapshot { generation: number; revision: number; pending: boolean; attached: boolean } export interface GrantAttachmentResult extends GrantAttachmentSnapshot { outcome: "attached" | "detached" | "unattached" | "pending_cancelled" | "unresolved" } -interface GrantAttachmentMutation { key: string; generation: number; revision: number; admissionRevision?: number; provider: string | null; status: GrantAttachmentStatus | null } +interface GrantAttachmentMutation { key: string; generation: number; revision: number; admissionRevision?: number; provider: string | null; status: GrantAttachmentStatus | null; projection?: UpstreamGrant } interface GrantRuntimeFeedbackRequest { key: string; state: GrantRuntimeState } interface GrantRuntimeStatesRequest { keys: string[] } interface GrantPoolSelectRequest { diff --git a/worker/grant-credentials.ts b/worker/grant-credentials.ts index 7175584..184c118 100644 --- a/worker/grant-credentials.ts +++ b/worker/grant-credentials.ts @@ -1,3 +1,4 @@ +import { routingProjection } from "./grant-projection.ts"; import snapshotJson from "./generated/provider-snapshot.json" with { type: "json" }; import { authorityCall, type GrantAttachmentResult, type GrantAttachmentSnapshot, type GrantAttachmentStatus } from "./authority.ts"; import { grantCoolingDown, grantQuotaExhausted, grantQuotaRatio, observeGrantQuota, observeGrantQuotaProbe, quotaRuntime } from "./grant-quota.ts"; @@ -464,7 +465,7 @@ export class GrantCredentialObject implements DurableObject { const record = await this.state.storage.get("credential"); const indexed = await authorityCall(this.env, "/grant-pools/attachment", { key }); const result = record - ? await authorityCall(this.env, "/grant-pools/publish", { key, generation: record.generation, revision: indexed.revision, admissionRevision: record.poolAdmissionRevision, provider: record.providerId ?? null, status: attachmentStatus(record) }) + ? await authorityCall(this.env, "/grant-pools/publish", { key, generation: record.generation, revision: indexed.revision, admissionRevision: record.poolAdmissionRevision, provider: record.providerId ?? null, status: attachmentStatus(record), projection: routingProjection(metadataGrant(record)) }) : await authorityCall(this.env, "/grant-pools/cancel-pending", { key, revision: indexed.revision }); if (record?.poolSyncPending) { record.poolSyncPending = false; diff --git a/worker/grant-eligibility.ts b/worker/grant-eligibility.ts new file mode 100644 index 0000000..a088349 --- /dev/null +++ b/worker/grant-eligibility.ts @@ -0,0 +1,131 @@ +import { HttpError } from "./utils.ts"; +import { assertOperationConfiguration, grantSupports, type GrantRequirement } from "./provider-auth.ts"; +import { grantCoolingDown, grantQuotaExhausted, grantRuntimeFresh } from "./grant-quota.ts"; +import type { Env, GrantRoutingPolicy, GrantRuntimeState, UpstreamGrant } from "./types"; + +const DEFAULT_GRANT_PRIORITY = 100; +const DEFAULT_GRANT_WEIGHT = 1; +export const DEFAULT_GRANT_ROUTING: GrantRoutingPolicy = { + strategy: "most_remaining", + stickiness: "none", + failover: true, + environmentFallback: false, + includedQuotaReservePercent: 0, + staleState: "allow", + staleAfterSeconds: 300, + switchAtUsedPercent: 90, + hysteresisPercent: 10, + eligibleGrants: {}, +}; + +export function grantRoutingPolicy(value: GrantRoutingPolicy | null | undefined): GrantRoutingPolicy { + if (!value) return { ...DEFAULT_GRANT_ROUTING, eligibleGrants: {} }; + const eligibleGrants: Record = {}; + if (value.eligibleGrants && typeof value.eligibleGrants === "object" && !Array.isArray(value.eligibleGrants)) { + for (const [providerId, refs] of Object.entries(value.eligibleGrants)) { + if (!validGrantSegment(providerId) || !Array.isArray(refs)) continue; + eligibleGrants[providerId] = [...new Set(refs.filter((ref): ref is string => typeof ref === "string" && validGrantSegment(ref)))].slice(0, 32).sort(); + } + } + const strategy = ["priority", "round_robin", "least_used", "most_remaining", "threshold", "weighted_random"].includes(value.strategy) ? value.strategy : DEFAULT_GRANT_ROUTING.strategy; + return { + strategy, + stickiness: strategy === "threshold" ? "none" : ["none", "identity", "session"].includes(value.stickiness) ? value.stickiness : DEFAULT_GRANT_ROUTING.stickiness, + failover: value.failover !== false, + environmentFallback: value.environmentFallback === true, + includedQuotaReservePercent: Number.isInteger(value.includedQuotaReservePercent) && value.includedQuotaReservePercent! >= 0 && value.includedQuotaReservePercent! <= 50 ? value.includedQuotaReservePercent : 0, + staleState: value.staleState === "deny" ? "deny" : "allow", + staleAfterSeconds: Number.isSafeInteger(value.staleAfterSeconds) && value.staleAfterSeconds >= 30 && value.staleAfterSeconds <= 86_400 ? value.staleAfterSeconds : DEFAULT_GRANT_ROUTING.staleAfterSeconds, + switchAtUsedPercent: finitePercent(value.switchAtUsedPercent, DEFAULT_GRANT_ROUTING.switchAtUsedPercent), + hysteresisPercent: finitePercent(value.hysteresisPercent, DEFAULT_GRANT_ROUTING.hysteresisPercent), + eligibleGrants, + }; +} + +export interface SelectedGrant { + authorityRevision?: number; + key: string; + grant: UpstreamGrant; + runtimeState: GrantRuntimeState | null; +} + +export interface GrantSelectionResult { + selected: SelectedGrant | null; + hasConfiguredGrant: boolean; +} +export type PinnedGrant = { key: string | null; revision: string | null } | { key: string; lineage: string }; + +export function grantUsable(grant: UpstreamGrant): boolean { + if (grant.credentialStore === "durable_object") { + return grant.credentialStatus !== "reauth_required" && !!(grant.hasCredential || grant.hasAccessToken); + } + const scalarCredentials = [grant.credential, grant.accessToken, grant.refreshToken]; + if (scalarCredentials.some((value) => value != null && (typeof value !== "string" || value.trim().length === 0)) || !validCredentialBundle(grant.credentials)) return false; + const bundled = grant.credentials ? Object.values(grant.credentials) : []; + return [grant.credential, grant.accessToken, ...bundled].some((value) => typeof value === "string" && value.trim().length > 0); +} + +export function validCredentialBundle(value: UpstreamGrant["credentials"]): boolean { + return value == null || (!Array.isArray(value) && typeof value === "object" && Object.entries(value).every(([name, secret]) => /^[A-Za-z0-9_.-]{1,128}$/.test(name) && typeof secret === "string" && secret.trim().length > 0)); +} + +export interface GrantCandidates { available: SelectedGrant[]; hasConfiguredGrant: boolean } + +export function activeOperationCandidates(available: SelectedGrant[], env: Env, requirement?: GrantRequirement): SelectedGrant[] { + // Qualification follows policy choice. Preserve the original pool-presence + // fact so an unusable configured pool cannot fall back to environment auth. + const configured = requirement ? available.filter(({ grant }) => { + try { assertOperationConfiguration(requirement, grant, env); return true; } + catch (error) { if (error instanceof HttpError) return false; throw error; } + }) : available; + const priority = configured.length ? Math.min(...configured.map(({ grant }) => grantPriority(grant))) : null; + return configured.filter(({ grant }) => grantPriority(grant) === priority); +} + +export function filterGrantCandidates(providerId: string, candidates: Array<{ key: string; grant: UpstreamGrant | null; runtimeState: GrantRuntimeState | null }>, excludedKeys: ReadonlySet, routing: GrantRoutingPolicy, pinnedKey: string | null | undefined, requirement: GrantRequirement | undefined, nowMs: number): GrantCandidates { + const configured = candidates.filter((entry): entry is SelectedGrant => !!entry.grant && entry.grant.enabled !== false && (!entry.grant.provider || entry.grant.provider === providerId) && grantUsable(entry.grant)); + const eligibilityRestricted = Object.prototype.hasOwnProperty.call(routing.eligibleGrants, providerId); + const eligibleRefs = routing.eligibleGrants[providerId] ?? []; + const available = configured.filter((entry) => { + if (requirement && !grantSupports(requirement, entry.grant)) return false; + if (pinnedKey !== undefined && entry.key !== pinnedKey) return false; + if (excludedKeys.has(entry.key) || grantCoolingDown(entry.runtimeState, nowMs)) return false; + if (grantQuotaExhausted(entry.runtimeState, requirement?.model, entry.grant.kind === "subscription" ? routing.includedQuotaReservePercent : 0, entry.grant.kind === "subscription")) return false; + const scope = parseGrantScope(entry.key); + if (eligibilityRestricted && (!scope || !eligibleRefs.includes(scope.tokenRef))) return false; + return routing.staleState !== "deny" || grantRuntimeFresh(entry.runtimeState, routing.staleAfterSeconds * 1_000, nowMs, requirement?.model); + }); + // Transport filtering must not reopen environment auth for a configured pool. + return { available, hasConfiguredGrant: configured.length > 0 || eligibilityRestricted || routing.staleState === "deny" || typeof pinnedKey === "string" }; +} + +function finitePercent(value: unknown, fallback: number): number { + return typeof value === "number" && Number.isFinite(value) && value >= 0 && value <= 100 ? value : fallback; +} + +export function currentGrantRuntime(grant: UpstreamGrant, state: GrantRuntimeState | null | undefined): GrantRuntimeState | null { + if (!state) return null; + return state.grantRevision === grantRevision(grant) ? state : null; +} + +export function grantRevision(grant: UpstreamGrant): string | null { return grant.updatedAt ?? grant.createdAt ?? null; } + +export function grantPriority(grant: UpstreamGrant): number { + return Number.isInteger(grant.priority) && grant.priority! >= 0 && grant.priority! <= 1_000_000 ? grant.priority! : DEFAULT_GRANT_PRIORITY; +} + +export function grantWeight(grant: UpstreamGrant): number { + return typeof grant.weight === "number" && Number.isFinite(grant.weight) && grant.weight > 0 && grant.weight <= 1_000_000 ? grant.weight : DEFAULT_GRANT_WEIGHT; +} + +export function validGrantSegment(value: string): boolean { + return value.length > 0 && value.length <= 256 && !value.includes("/") && !/[\u0000-\u001f\u007f]/.test(value); +} + +function parseGrantScope(key: string): { scope: "policies" | "tenants"; scopeId: string; tokenRef: string } | null { + const parts = key.split("/"); + if (parts[0] !== "oauth") return null; + if (parts[1] === "tenants" && parts.length === 4 && validGrantSegment(parts[2]) && validGrantSegment(parts[3])) return { scope: "tenants", scopeId: parts[2], tokenRef: parts[3] }; + if (parts.length === 3 && validGrantSegment(parts[1]) && validGrantSegment(parts[2])) return { scope: "policies", scopeId: parts[1], tokenRef: parts[2] }; + return null; +} diff --git a/worker/grant-projection.ts b/worker/grant-projection.ts new file mode 100644 index 0000000..9808821 --- /dev/null +++ b/worker/grant-projection.ts @@ -0,0 +1,22 @@ +import { HttpError } from "./utils.ts"; +import type { UpstreamGrant } from "./types"; + +// An allowlist, rather than object spread: neither legacy secrets nor future +// credential fields may enter the global authority's routing index. +export function routingProjection(grant: UpstreamGrant): UpstreamGrant { + const fields = ["version", "enabled", "priority", "weight", "kind", "provider", "label", "credentialStore", "credentialGeneration", "credentialLineage", "credentialStatus", "hasCredential", "credentialFields", "hasAccessToken", "hasRefreshToken", "tokenType", "expiresAt", "scopes", "accountId", "subscription", "createdAt", "updatedAt", "revokedAt"] as const; + return Object.fromEntries(fields.filter(field => grant[field] !== undefined).map(field => [field, grant[field]])); +} + +export function publishRoutingProjection(sql: SqlStorage, input: { key: string; generation: number; projection?: UpstreamGrant }): void { + if (!input.projection) { + // An overlapping old owner cannot leave a previously indexed generation + // eligible after publishing new lifecycle state without its projection. + sql.exec("INSERT INTO upstream_grant_projections (grant_key, projection_json, legacy) VALUES (?, NULL, 1) ON CONFLICT (grant_key) DO UPDATE SET projection_json = NULL, legacy = 1", input.key); + return; + } + const projection = routingProjection(input.projection); + if (projection.credentialGeneration !== input.generation || projection.credentialStore !== "durable_object") throw new HttpError(400, "invalid_authority_request", "projection generation must match its credential owner"); + sql.exec("INSERT INTO upstream_grant_projections (grant_key, projection_json, legacy) VALUES (?, ?, 0) ON CONFLICT (grant_key) DO UPDATE SET projection_json = excluded.projection_json, legacy = 0", input.key, JSON.stringify(projection)); +} + diff --git a/worker/grant-selection.ts b/worker/grant-selection.ts index a20b7ad..341caba 100644 --- a/worker/grant-selection.ts +++ b/worker/grant-selection.ts @@ -1,76 +1,10 @@ -import { HttpError } from "./utils.ts"; -import { assertOperationConfiguration, grantSupports, type GrantRequirement } from "./provider-auth.ts"; +import { grantSupports, type GrantRequirement } from "./provider-auth.ts"; import { authorityCall, type AdmissionPool } from "./authority.ts"; -import { grantCoolingDown, grantQuotaExhausted, grantQuotaRatio, grantRuntimeFresh } from "./grant-quota.ts"; +import { grantQuotaRatio } from "./grant-quota.ts"; +export * from "./grant-eligibility.ts"; +import { DEFAULT_GRANT_ROUTING, grantRoutingPolicy, activeOperationCandidates, filterGrantCandidates, currentGrantRuntime, grantWeight, type SelectedGrant, type GrantCandidates, type GrantSelectionResult } from "./grant-eligibility.ts"; import type { AccessPolicyEntry, Env, GrantRoutingPolicy, GrantRuntimeState, UpstreamGrant } from "./types"; -const DEFAULT_GRANT_PRIORITY = 100; -const DEFAULT_GRANT_WEIGHT = 1; -export const DEFAULT_GRANT_ROUTING: GrantRoutingPolicy = { - strategy: "most_remaining", - stickiness: "none", - failover: true, - environmentFallback: false, - includedQuotaReservePercent: 0, - staleState: "allow", - staleAfterSeconds: 300, - switchAtUsedPercent: 90, - hysteresisPercent: 10, - eligibleGrants: {}, -}; - -export function grantRoutingPolicy(value: GrantRoutingPolicy | null | undefined): GrantRoutingPolicy { - if (!value) return { ...DEFAULT_GRANT_ROUTING, eligibleGrants: {} }; - const eligibleGrants: Record = {}; - if (value.eligibleGrants && typeof value.eligibleGrants === "object" && !Array.isArray(value.eligibleGrants)) { - for (const [providerId, refs] of Object.entries(value.eligibleGrants)) { - if (!validGrantSegment(providerId) || !Array.isArray(refs)) continue; - eligibleGrants[providerId] = [...new Set(refs.filter((ref): ref is string => typeof ref === "string" && validGrantSegment(ref)))].slice(0, 32).sort(); - } - } - const strategy = ["priority", "round_robin", "least_used", "most_remaining", "threshold", "weighted_random"].includes(value.strategy) ? value.strategy : DEFAULT_GRANT_ROUTING.strategy; - return { - strategy, - stickiness: strategy === "threshold" ? "none" : ["none", "identity", "session"].includes(value.stickiness) ? value.stickiness : DEFAULT_GRANT_ROUTING.stickiness, - failover: value.failover !== false, - environmentFallback: value.environmentFallback === true, - includedQuotaReservePercent: Number.isInteger(value.includedQuotaReservePercent) && value.includedQuotaReservePercent! >= 0 && value.includedQuotaReservePercent! <= 50 ? value.includedQuotaReservePercent : 0, - staleState: value.staleState === "deny" ? "deny" : "allow", - staleAfterSeconds: Number.isSafeInteger(value.staleAfterSeconds) && value.staleAfterSeconds >= 30 && value.staleAfterSeconds <= 86_400 ? value.staleAfterSeconds : DEFAULT_GRANT_ROUTING.staleAfterSeconds, - switchAtUsedPercent: finitePercent(value.switchAtUsedPercent, DEFAULT_GRANT_ROUTING.switchAtUsedPercent), - hysteresisPercent: finitePercent(value.hysteresisPercent, DEFAULT_GRANT_ROUTING.hysteresisPercent), - eligibleGrants, - }; -} - -export interface SelectedGrant { - key: string; - grant: UpstreamGrant; - runtimeState: GrantRuntimeState | null; -} - -export interface GrantSelectionResult { - selected: SelectedGrant | null; - hasConfiguredGrant: boolean; -} -export type PinnedGrant = { key: string | null; revision: string | null } | { key: string; lineage: string }; - -export function grantUsable(grant: UpstreamGrant): boolean { - if (grant.credentialStore === "durable_object") { - return grant.credentialStatus !== "reauth_required" && !!(grant.hasCredential || grant.hasAccessToken); - } - const scalarCredentials = [grant.credential, grant.accessToken, grant.refreshToken]; - if (scalarCredentials.some((value) => value != null && (typeof value !== "string" || value.trim().length === 0)) || !validCredentialBundle(grant.credentials)) return false; - const bundled = grant.credentials ? Object.values(grant.credentials) : []; - return [grant.credential, grant.accessToken, ...bundled].some((value) => typeof value === "string" && value.trim().length > 0); -} - -export function validCredentialBundle(value: UpstreamGrant["credentials"]): boolean { - return value == null || (!Array.isArray(value) && typeof value === "object" && Object.entries(value).every(([name, secret]) => /^[A-Za-z0-9_.-]{1,128}$/.test(name) && typeof secret === "string" && secret.trim().length > 0)); -} - -export interface GrantCandidates { available: SelectedGrant[]; hasConfiguredGrant: boolean } - export function selectPolicyCandidates(pools: Array<{ entry: T; candidates: GrantCandidates }>, requirement?: GrantRequirement): { entry: T; candidates: GrantCandidates } | undefined { const compatible = pools.map(({ entry, candidates }) => ({ entry, @@ -108,17 +42,6 @@ export async function selectGrant( return (await resolveGrantSelection(providerId, policyId, tenantId, defaultTokenRef, env, excludedKeys, routing, stickyHash)).selected; } -export function activeOperationCandidates(available: SelectedGrant[], env: Env, requirement?: GrantRequirement): SelectedGrant[] { - // Qualification follows policy choice. Preserve the original pool-presence - // fact so an unusable configured pool cannot fall back to environment auth. - const configured = requirement ? available.filter(({ grant }) => { - try { assertOperationConfiguration(requirement, grant, env); return true; } - catch (error) { if (error instanceof HttpError) return false; throw error; } - }) : available; - const priority = configured.length ? Math.min(...configured.map(({ grant }) => grantPriority(grant))) : null; - return configured.filter(({ grant }) => grantPriority(grant) === priority); -} - export async function resolveGrantSelection( providerId: string, policyId: string, @@ -174,24 +97,7 @@ export async function resolveGrantCandidates( const grant = grants.get(key) ?? null; return { key, grant, runtimeState: grant ? currentGrantRuntime(grant, pool.states?.[key]) : null }; }); - const configured = candidates.filter((entry): entry is SelectedGrant => !!entry.grant && entry.grant.enabled !== false && (!entry.grant.provider || entry.grant.provider === providerId) && grantUsable(entry.grant)); - const eligibilityRestricted = Object.prototype.hasOwnProperty.call(routing.eligibleGrants, providerId); - const eligibleRefs = routing.eligibleGrants[providerId] ?? []; - const available = configured.filter((entry) => { - if (requirement && !grantSupports(requirement, entry.grant)) return false; - if (pinnedKey !== undefined && entry.key !== pinnedKey) return false; - if (excludedKeys.has(entry.key) || grantCoolingDown(entry.runtimeState, nowMs)) return false; - if (grantQuotaExhausted(entry.runtimeState, requirement?.model, entry.grant.kind === "subscription" ? routing.includedQuotaReservePercent : 0, entry.grant.kind === "subscription")) return false; - const scope = parseGrantScope(entry.key); - if (eligibilityRestricted && (!scope || !eligibleRefs.includes(scope.tokenRef))) return false; - return routing.staleState !== "deny" || grantRuntimeFresh(entry.runtimeState, routing.staleAfterSeconds * 1_000, nowMs, requirement?.model); - }); - // Transport filtering must not reopen environment auth for a configured pool. - return { available, hasConfiguredGrant: configured.length > 0 || eligibilityRestricted || routing.staleState === "deny" || typeof pinnedKey === "string" }; -} - -function finitePercent(value: unknown, fallback: number): number { - return typeof value === "number" && Number.isFinite(value) && value >= 0 && value <= 100 ? value : fallback; + return filterGrantCandidates(providerId, candidates, excludedKeys, routing, pinnedKey, requirement, nowMs); } export async function recordGrantRuntime(env: Env, key: string, state: GrantRuntimeState): Promise { @@ -209,30 +115,3 @@ export async function grantSelectionStats(env: Env, keys: string[]): Promise }>(env, "/grant-pools/stats", { keys: keys.slice(offset, offset + 66) })).stats); return stats; } - -export function currentGrantRuntime(grant: UpstreamGrant, state: GrantRuntimeState | null | undefined): GrantRuntimeState | null { - if (!state) return null; - return state.grantRevision === grantRevision(grant) ? state : null; -} - -export function grantRevision(grant: UpstreamGrant): string | null { return grant.updatedAt ?? grant.createdAt ?? null; } - -export function grantPriority(grant: UpstreamGrant): number { - return Number.isInteger(grant.priority) && grant.priority! >= 0 && grant.priority! <= 1_000_000 ? grant.priority! : DEFAULT_GRANT_PRIORITY; -} - -export function grantWeight(grant: UpstreamGrant): number { - return typeof grant.weight === "number" && Number.isFinite(grant.weight) && grant.weight > 0 && grant.weight <= 1_000_000 ? grant.weight : DEFAULT_GRANT_WEIGHT; -} - -export function validGrantSegment(value: string): boolean { - return value.length > 0 && value.length <= 256 && !value.includes("/") && !/[\u0000-\u001f\u007f]/.test(value); -} - -function parseGrantScope(key: string): { scope: "policies" | "tenants"; scopeId: string; tokenRef: string } | null { - const parts = key.split("/"); - if (parts[0] !== "oauth") return null; - if (parts[1] === "tenants" && parts.length === 4 && validGrantSegment(parts[2]) && validGrantSegment(parts[3])) return { scope: "tenants", scopeId: parts[2], tokenRef: parts[3] }; - if (parts.length === 3 && validGrantSegment(parts[1]) && validGrantSegment(parts[2])) return { scope: "policies", scopeId: parts[1], tokenRef: parts[2] }; - return null; -} diff --git a/worker/grant-sticky.ts b/worker/grant-sticky.ts new file mode 100644 index 0000000..79d1fa7 --- /dev/null +++ b/worker/grant-sticky.ts @@ -0,0 +1,11 @@ +import { grantRoutingPolicy } from "./grant-eligibility.ts"; +import type { AuthorizedIdentity } from "./types"; +import { sha256Hex } from "./utils.ts"; + +export function grantStickyInput(auth: AuthorizedIdentity, sessionHash?: string, sessionId?: string | null): Promise { + const routing = grantRoutingPolicy(auth.policy.grantRouting); + if (routing.stickiness === "none") return Promise.resolve(null); + const identity = auth.principalId ?? auth.credentialId ?? auth.policyId; + if (routing.stickiness === "identity") return sha256Hex(`identity:${identity}`); + return sessionHash ? Promise.resolve(sessionHash) : sha256Hex(`session:${sessionId ?? identity}`); +} diff --git a/worker/policy-models.ts b/worker/policy-models.ts index cf6bc05..247e88f 100644 --- a/worker/policy-models.ts +++ b/worker/policy-models.ts @@ -1,11 +1,11 @@ import { policyAllowsModel } from "../shared/policy-models.ts"; import type { AccessPolicy } from "./types"; import type { ProxySelection } from "./proxy-selection"; -import { HttpError } from "./utils"; +import { HttpError } from "./utils.ts"; export { policyAllowsModel }; -export function assertPolicyModel(policy: AccessPolicy, selection: ProxySelection): void { +export function assertPolicyModel(policy: AccessPolicy, selection: { model: Pick, "id"> | null; endpoint: ProxySelection["endpoint"]; capability: string }): void { if (selection.model) { if (policy.models != null && !policy.models.includes(selection.model.id)) throw new HttpError(403, "model_not_allowed", "model is not allowed by this policy"); } else if (policy.models != null && (selection.capability.startsWith("llm.") || selection.endpoint.request_format === "cloudflare_ai_gateway.universal")) { diff --git a/worker/providers.ts b/worker/providers.ts index 4d6703f..5a001c8 100644 --- a/worker/providers.ts +++ b/worker/providers.ts @@ -1,6 +1,7 @@ +import { grantStickyInput } from "./grant-sticky.ts"; import snapshotJson from "./generated/provider-snapshot.json" with { type: "json" }; import { listConnections, resolveConnection } from "./authority.ts"; -import { proxyAdmission } from "./proxy-auth.ts"; +import { admittedGrantSelection, proxyAdmission, proxySessionHash } from "./proxy-auth.ts"; import { grantRevision, grantUsable as canonicalGrantUsable, resolveGrantSelection, type PinnedGrant } from "./grant-selection.ts"; import { grantsVisibleToPolicies, type GrantRecord } from "./grant-scope.ts"; @@ -351,8 +352,9 @@ export async function listHealth(env: Env): Promise> async function grantFor(provider: CompiledProvider, auth: AuthorizedIdentity, env: Env, excludedKeys: ReadonlySet, stickyHash: string | null, recordSelection: boolean, pinned?: PinnedGrant, requirement?: GrantRequirement): Promise<{ selected: { key: string; grant: UpstreamGrant } | null; hasConfiguredGrant: boolean }> { const tokenRef = provider.auth.schemes.find((scheme) => scheme.type === "oauth")?.tokenRef ?? provider.id; - const tenant = auth.policy.tenantId ?? "default"; - const resolution = await resolveGrantSelection(provider.id, auth.policyId, tenant, tokenRef, env, excludedKeys, auth.policy.grantRouting, stickyHash, recordSelection, pinned?.key, requirement, excludedKeys.size ? undefined : proxyAdmission(auth)?.pools[provider.id]); + const admitted = recordSelection && !pinned && requirement ? await admittedGrantSelection(auth, env, requirement, excludedKeys) : undefined; + if (!admitted && excludedKeys.size && auth.authType === "proxy_key") stickyHash = await grantStickyInput(auth, proxySessionHash(auth)); + const resolution = admitted ?? await resolveGrantSelection(provider.id, auth.policyId, auth.policy.tenantId ?? "default", tokenRef, env, excludedKeys, auth.policy.grantRouting, stickyHash, recordSelection, pinned?.key, requirement, excludedKeys.size ? undefined : proxyAdmission(auth)?.pools[provider.id]); return { selected: resolution.selected ? { key: resolution.selected.key, grant: await refreshGrant(resolution.selected.key, resolution.selected.grant, provider, env, false) } : null, hasConfiguredGrant: resolution.hasConfiguredGrant }; } diff --git a/worker/proxy-auth.ts b/worker/proxy-auth.ts index 8af1f34..d4c8681 100644 --- a/worker/proxy-auth.ts +++ b/worker/proxy-auth.ts @@ -1,3 +1,7 @@ +import { assertPolicyModel } from "./policy-models.ts"; +import type { AdmissionSelection } from "./authority-admission.ts"; +import type { GrantRequirement } from "./provider-auth.ts"; +import type { GrantSelectionResult } from "./grant-eligibility.ts"; import type { AccessPolicyEntry, AuthorizedIdentity, Env, ProxyCredential } from "./types"; import { authorityCall, type AdmissionProvider, type AdmissionResult } from "./authority.ts"; import { errorResponse, HttpError, parseProxyKey, safeEqual, sha256Hex } from "./utils.ts"; @@ -12,14 +16,12 @@ export function proxyAdmission(identity: AuthorizedIdentity): AdmissionResult | export function proxySessionHash(identity: AuthorizedIdentity): string | undefined { return sessionHashes.get(identity); } export function proxyKeyVerification(identity: AuthorizedIdentity): ProxyKeyVerification | undefined { return verifiedKeys.get(identity); } -export async function authenticateProxyKey(headers: Headers, env: Env, providers: AdmissionProvider[] = [], allConnections = false, sessionId?: string | null): Promise { +export async function authenticateProxyKey(headers: Headers, env: Env, providers: AdmissionProvider[] = [], allConnections = false, sessionId?: string | null, selection?: AdmissionSelection): Promise { const parsed = proxyKeyFromHeaders(headers); if (!parsed) return errorResponse("invalid_proxy_key", "a valid ClawRouter proxy key is required", 401); const secretSha256 = await sha256Hex(parsed.secret); - const [admission, sessionHash] = await Promise.all([ - authorityCall(env, "/admission/resolve", { credentialId: parsed.kid, secretSha256, providers, allConnections }), - sessionId ? sha256Hex(`session:${sessionId}`) : undefined, - ]); + const sessionHash = sessionId ? await sha256Hex(`session:${sessionId}`) : undefined; + const admission = await authorityCall(env, "/admission/resolve", { credentialId: parsed.kid, secretSha256, providers, allConnections, selection: selection ? { ...selection, sessionHash } : undefined }); const credentialEntry = admission.credential; if (!credentialEntry) return errorResponse("unknown_proxy_key", "proxy key is not registered", 401); if (!safeEqual(secretSha256, credentialEntry.credential.secretSha256.toLowerCase())) return errorResponse("invalid_proxy_key", "proxy key secret is invalid", 401); @@ -43,10 +45,37 @@ export async function authenticateProxyKey(headers: Headers, env: Env, providers }; verifiedKeys.set(identity, Object.freeze({ kind: "proxy_key", credentialId: parsed.kid, principalId: identity.principalId, secretSha256 })); admissions.set(identity, admission); + if (selection) admissionRequests.set(identity, selection); + if (selection && admission.selection) selections.set(identity, { request: selection, result: admission.selection }); if (sessionHash) sessionHashes.set(identity, sessionHash); return identity; } +const admissionRequests = new WeakMap(); +const selections = new WeakMap(); + +export async function admittedGrantSelection(auth: AuthorizedIdentity, env: Env, requirement: GrantRequirement, excludedKeys: ReadonlySet): Promise { + const initial = selections.get(auth); + selections.delete(auth); + if (!excludedKeys.size) return initial && initial.request.providerId === requirement.provider.id && initial.request.endpointId === requirement.endpoint.id && initial.request.mode === requirement.mode && initial.request.model === requirement.model ? initial.result : undefined; + const verified = verifiedKeys.get(auth); + if (!verified) return undefined; + const request: AdmissionSelection = { providerId: requirement.provider.id, endpointId: requirement.endpoint.id, mode: requirement.mode, model: requirement.model, capability: requirement.provider.capabilities.find(item => item.endpoint === requirement.endpoint.id)?.id ?? requirement.endpoint.id, catalogModelId: admissionRequests.get(auth)?.catalogModelId ?? requirement.provider.models.find(item => item.upstream === requirement.model || item.id === requirement.model)?.id ?? null, excludedKeys: [...excludedKeys], sessionHash: sessionHashes.get(auth) }; + const admission = await authorityCall(env, "/admission/resolve", { credentialId: verified.credentialId, secretSha256: verified.secretSha256, providers: [{ id: requirement.provider.id, tokenRef: requirement.provider.auth.schemes.find(scheme => scheme.type === "oauth")?.tokenRef ?? requirement.provider.id }], selection: request }); + if (!admission.credential || !safeEqual(verified.secretSha256, admission.credential.credential.secretSha256)) throw new HttpError(401, "invalid_proxy_key", "proxy key is no longer valid"); + if (admission.policyUnavailable || admission.ownerUnavailable || admission.connectionsUnavailable) throw new Error("admission unavailable"); + const failure = proxyCredentialFailure(admission.credential.credential, admission.policy, admission.owner?.record.enabled); + if (failure) throw failure; + if (admission.policy!.policy.providers.length && !admission.policy!.policy.providers.includes(requirement.provider.id)) throw new HttpError(403, "provider_not_allowed", "policy does not allow provider"); + if (admission.connections.find(item => item.providerId === requirement.provider.id)?.enabled === false) throw new HttpError(503, "provider_disabled", "provider is disabled"); + assertPolicyModel(admission.policy!.policy, { model: request.catalogModelId ? { id: request.catalogModelId } : null, endpoint: requirement.endpoint, capability: request.capability }); + auth.principalId = admission.credential.credential.principalId ?? null; + auth.policyId = admission.policy!.policyId; + auth.policy = admission.policy!.policy; + admissions.set(auth, admission); + return admission.selection; +} + // Entry authentication and delayed control admission share these canonical // lifecycle rules, including unowned/unmaterialized service-key semantics. export function proxyCredentialFailure(credential: ProxyCredential, entry: AccessPolicyEntry | null | undefined, principalEnabled?: boolean | null): HttpError | null { diff --git a/worker/proxy.ts b/worker/proxy.ts index d60e381..a383d71 100644 --- a/worker/proxy.ts +++ b/worker/proxy.ts @@ -1,3 +1,4 @@ +import { grantStickyInput } from "./grant-sticky.ts"; import { assertPolicyModel } from "./policy-models.ts"; import { resolveTemplate } from "./provider-templates.ts"; import { assessModelRequest } from "../shared/model-request-parameters.ts"; @@ -34,7 +35,7 @@ import { normalizePreStreamError, observeUsage, proxyResponseHeaders } from "./p import type { AuthorizedIdentity, CompiledQuotaConfig, Env, ProviderConnection } from "./types"; import { decodePathSegment, - errorResponse, HttpError, randomId, readJson, sha256Hex, + errorResponse, HttpError, randomId, readJson, } from "./utils"; type AuthMode = "proxy_key" | "access"; @@ -71,13 +72,15 @@ export async function proxyOpenAi(request: Request, env: Env, context: Execution throw error; } const modelId = typeof body.model === "string" ? body.model : ""; - const preauthenticated = await preauthenticate(request, env, mode, mode === "proxy_key" ? modelRoute(modelId, capabilityForPath(path) ?? undefined)?.provider.id : undefined); + const earlySelection = concreteOpenAiSelection(path, body, env); + const preauthenticated = await preauthenticate(request, env, mode, mode === "proxy_key" ? modelRoute(modelId, capabilityForPath(path) ?? undefined)?.provider.id : undefined, undefined, isSelectionFailure(earlySelection) ? undefined : earlySelection); if (preauthenticated instanceof Response) return preauthenticated; if (!modelId) return errorResponse("model_required", "model is required", 400); if (modelId === FUSION_MODEL_ID) { if (path !== "/v1/chat/completions") return errorResponse("fusion_capability_unsupported", `${FUSION_MODEL_ID} supports only /v1/chat/completions`, 400); return proxyFusion(request, env, context, mode, body, preauthenticated); } + if (!isSelectionFailure(earlySelection)) return proxySelected(request, env, context, mode, earlySelection, {}, preauthenticated); return proxyConcreteOpenAi(request, env, context, path, mode, body, preauthenticated); } @@ -176,10 +179,11 @@ export async function proxyManifest(request: Request, env: Env, context: Executi if (!provider || !endpoint) return errorResponse("route_not_found", "manifest proxy route not found", 404); const session = mode === "access" ? await verifiedAccessSession(request, env) : null; if (mode === "access" && !session) return errorResponse("access_session_required", "a verified session is required", 401); - let preauthenticated = mode === "access" ? null : await preauthenticate(request, env, mode, provider.id, { provider, endpoint, mode: "http" }); - if (preauthenticated instanceof Response) return preauthenticated; + let preauthenticated: AuthorizedIdentity | Response | null = null; const control = responsesControl(provider, endpoint); if (control) { + preauthenticated = await preauthenticate(request, env, mode, provider.id, { provider, endpoint, mode: "http" }); + if (preauthenticated instanceof Response) return preauthenticated; if (request.method === "GET") { if (endpoint.method !== "GET") return errorResponse("method_not_allowed", "control method is not allowed", 405); await requireEmptyResponseControlBody(request); @@ -195,12 +199,18 @@ export async function proxyManifest(request: Request, env: Env, context: Executi } // Controls resolve their original owner; creation keeps ordinary policy // selection, including transports that cannot provide durable recovery. - const envelope = request.method === "GET" || request.method === "HEAD" - ? directManifestEnvelope(request, endpoint) - : manifestEnvelope(await readJson(request)); + let envelope, prepared; + try { + envelope = request.method === "GET" || request.method === "HEAD" ? directManifestEnvelope(request, endpoint) : manifestEnvelope(await readJson(request)); + prepared = prepareManifestRequest(provider, endpoint, envelope.body, envelope.pathParams, env); + } catch (error) { + const auth = await preauthenticate(request, env, mode, provider.id); + if (auth instanceof Response) return auth; + throw error; + } const method = (envelope.method ?? endpoint.method).toUpperCase(); - if (!endpoint.methods.includes(method)) return errorResponse("method_not_allowed", `endpoint does not allow ${method}`, 405); - const prepared = prepareManifestRequest(provider, endpoint, envelope.body, envelope.pathParams, env); + if (!endpoint.methods.includes(method)) { const auth = await preauthenticate(request, env, mode, provider.id); return auth instanceof Response ? auth : errorResponse("method_not_allowed", `endpoint does not allow ${method}`, 405); } + if (!session) preauthenticated = await preauthenticate(request, env, mode, provider.id, undefined, { provider, endpoint, model: prepared.model, capability: provider.capabilities.find(item => item.endpoint === endpoint.id)?.id ?? endpoint.id, body: prepared.body, pathParams: prepared.pathParams, method }); if (session) preauthenticated = await accessSessionIdentity(session, env, provider.id, { provider, endpoint, mode: "http", model: prepared.model?.id }); if (preauthenticated instanceof Response) return preauthenticated; const capability = provider.capabilities.find((item) => item.endpoint === endpoint.id)?.id ?? endpoint.id; @@ -212,21 +222,31 @@ export async function proxyNative(request: Request, env: Env, context: Execution if (!match) return errorResponse("route_not_found", "native proxy route not found", 404); const provider = providerById(decodePathSegment(match[1])); if (!provider) return errorResponse("provider_not_found", "provider not found", 404); - const preauthenticated = await preauthenticate(request, env, "proxy_key", provider.id); - if (preauthenticated instanceof Response) return preauthenticated; const endpoint = provider.endpoints.find((candidate) => nativeMatch(candidate, match[2])); - if (!endpoint || !endpoint.native_proxy) return errorResponse("route_not_found", "native provider route not found", 404); + if (!endpoint || !endpoint.native_proxy) { const auth = await preauthenticate(request, env, "proxy_key", provider.id); return auth instanceof Response ? auth : errorResponse("route_not_found", "native provider route not found", 404); } const method = request.method.toUpperCase(); - if (!endpoint.methods.includes(method)) return errorResponse("method_not_allowed", `endpoint does not allow ${method}`, 405); + if (!endpoint.methods.includes(method)) { const auth = await preauthenticate(request, env, "proxy_key", provider.id); return auth instanceof Response ? auth : errorResponse("method_not_allowed", `endpoint does not allow ${method}`, 405); } const control = responsesControl(provider, endpoint); if (control) { + const preauthenticated = await preauthenticate(request, env, "proxy_key", provider.id); + if (preauthenticated instanceof Response) return preauthenticated; await requireEmptyResponseControlBody(request); return proxyResponseControl(request, env, control.action, nativeParams(endpoint, match[2]).response_id, new URL(request.url).searchParams, "proxy_key", provider.id, control.create.id, preauthenticated ?? undefined); } - const body = request.method === "GET" || request.method === "HEAD" ? {} : await readJson(request); - const prepared = prepareNativeRequest(provider, endpoint, body, match[2], env); + let prepared; + try { + const body = request.method === "GET" || request.method === "HEAD" ? {} : await readJson(request); + prepared = prepareNativeRequest(provider, endpoint, body, match[2], env); + } catch (error) { + const auth = await preauthenticate(request, env, "proxy_key", provider.id); + if (auth instanceof Response) return auth; + throw error; + } const capability = provider.capabilities.find((item) => item.endpoint === endpoint.id)?.id ?? endpoint.id; - return proxySelected(request, env, context, "proxy_key", { provider, endpoint, model: prepared.model, capability, body: prepared.body, pathParams: prepared.pathParams, method }, searchParamsRecord(new URL(request.url).searchParams), preauthenticated); + const selection = { provider, endpoint, model: prepared.model, capability, body: prepared.body, pathParams: prepared.pathParams, method }; + const preauthenticated = await preauthenticate(request, env, "proxy_key", provider.id, undefined, selection); + if (preauthenticated instanceof Response) return preauthenticated; + return proxySelected(request, env, context, "proxy_key", selection, searchParamsRecord(new URL(request.url).searchParams), preauthenticated); } async function proxySelected(request: Request, env: Env, context: ExecutionContext, mode: AuthMode, selection: ProxySelection, queryInput: Record = {}, preauthenticated: AuthorizedIdentity | null = null, reservedBudget?: ReservedProxyBudget, compound?: CompoundRequestContext, auditAuth?: AuthorizedIdentity): Promise { @@ -469,14 +489,8 @@ export function captureGrantRuntime(context: ExecutionContext, env: Env, key: st if (state) context.waitUntil(recordGrantRuntime(env, key, { ...state, grantRevision: revision }).catch(() => undefined)); } -async function grantStickyHash(request: Request, auth: AuthorizedIdentity): Promise { - const routing = grantRoutingPolicy(auth.policy.grantRouting); - if (routing.stickiness === "none") return null; - const identity = auth.principalId ?? auth.credentialId ?? auth.policyId; - if (routing.stickiness === "identity") return sha256Hex(`identity:${identity}`); - const sessionId = correlationMetadata(request).sessionId; - if (sessionId && proxySessionHash(auth)) return proxySessionHash(auth)!; - return sha256Hex(`session:${sessionId ?? identity}`); +function grantStickyHash(request: Request, auth: AuthorizedIdentity): Promise { + return grantStickyInput(auth, proxySessionHash(auth), correlationMetadata(request).sessionId); } function selectedFailure(error: unknown): HttpError { @@ -491,10 +505,17 @@ async function auditSelectionFailure(request: Request, env: Env, context: Execut createProxyAccounting({ context, env, auth, selection, request, compound }).fail(statusCode, status); } -async function preauthenticate(request: Request, env: Env, mode: AuthMode, providerId?: string, requirement?: import("./provider-auth").GrantRequirement): Promise { +async function preauthenticate(request: Request, env: Env, mode: AuthMode, providerId?: string, requirement?: import("./provider-auth").GrantRequirement, selection?: ProxySelection): Promise { if (mode === "proxy_key") { + // Continuations select their recorded owner later. Invalid operations must + // retain their existing error precedence without incrementing counters. + if (selection) { + try { validateSelectedInput(selection); } + catch { selection = undefined; } + if (selection?.capability === "llm.responses" && (request.headers.has("x-codex-turn-state") || !Array.isArray(selection.body) && selection.body.previous_response_id)) selection = undefined; + } const provider = providerId ? providerById(providerId) : null; - return authenticateProxyKey(request.headers, env, provider ? [{ id: provider.id, tokenRef: provider.auth.schemes.find(scheme => scheme.type === "oauth")?.tokenRef ?? provider.id }] : [], false, correlationMetadata(request).sessionId); + return authenticateProxyKey(request.headers, env, provider ? [{ id: provider.id, tokenRef: provider.auth.schemes.find(scheme => scheme.type === "oauth")?.tokenRef ?? provider.id }] : [], false, correlationMetadata(request).sessionId, selection ? { providerId: selection.provider.id, endpointId: selection.endpoint.id, mode: "http", model: selectionUpstreamModel(selection), catalogModelId: selection.model?.id ?? null, capability: selection.capability } : undefined); } return providerId ? accessIdentity(request, env, providerId, requirement) : null; } diff --git a/worker/test/admission-fixture.mjs b/worker/test/admission-fixture.mjs new file mode 100644 index 0000000..4d709ac --- /dev/null +++ b/worker/test/admission-fixture.mjs @@ -0,0 +1,27 @@ +import { createGrantAuthority } from "./grant-authority-fixture.mjs"; +import { authorityCall } from "../authority.ts"; +import { sha256Hex } from "../utils.ts"; +const secret = 'synthetic-admission-secret'; +export const admissionHeaders = { authorization: `Bearer clawrouter-live-fixture-${secret}`, 'content-type': 'application/json' }; +export const admissionProvider = { id: 'anthropic', tokenRef: 'anthropic' }; +export async function admissionFixture() { + const records = new Map(), calls = [], pending = []; + let beforeRead; + const env = { + ANTHROPIC_API_KEY: 'synthetic-environment', + POLICY_KV: { async get(key, type) { const action = beforeRead; beforeRead = null; await action?.(key); const value = Array.isArray(key) ? new Map(key.map(k => [k, records.get(k) ?? null])) : records.get(key) ?? null; return type === "text" && value !== null ? JSON.stringify(value) : structuredClone(value); }, async put(key, value) { records.set(key, JSON.parse(value)); } }, + USAGE_QUEUE: { send: async () => {} }, + }; + const authority = createGrantAuthority(undefined, env); + env.ACCESS_CONTROL = { idFromName: name => name, get: () => ({ fetch(url, init) { calls.push(new URL(url).pathname); return authority.fetch(url, init); } }) }; + const policy = { enabled: true, generation: 'v1', providers: ['anthropic'], models: ['anthropic/claude-sonnet-5'], tenantId: 'fixture', monthlyBudgetMicros: null, retainRequestContent: false }; + const credential = { enabled: true, secretSha256: await sha256Hex(secret), policyId: 'fixture', policyGeneration: 'v1', principalId: 'owner@example.com' }; + const call = (path, body) => authorityCall(env, path, body); + await call('/policies/initialize-all', [{ policyId: 'fixture', policy }]); + await call('/credentials/initialize-all', [{ credentialId: 'fixture', credential }]); + await call('/users/initialize-all', [{ email: credential.principalId, record: { enabled: true } }]); + await call('/connections/initialize-all', [{ providerId: admissionProvider.id, enabled: true }]); + calls.length = 0; + return { env, records, authority, calls, pending, policy, credential, call, beforeRead(action) { beforeRead = action; }, context: { waitUntil(promise) { pending.push(promise); } } }; +} + diff --git a/worker/test/admission.test.mjs b/worker/test/admission.test.mjs index a285f29..a2b296a 100644 --- a/worker/test/admission.test.mjs +++ b/worker/test/admission.test.mjs @@ -1,35 +1,11 @@ import './typescript-setup.mjs'; import assert from 'node:assert/strict'; import test from 'node:test'; -import { createGrantAuthority } from './grant-authority-fixture.mjs'; -import { authorityCall, resolveCredentials, resolvePolicies, resolveUsers, resolveConnections } from '../authority.ts'; +import { resolveCredentials, resolvePolicies, resolveUsers, resolveConnections } from '../authority.ts'; import { authenticateProxyKey, proxyCredentialFailure } from '../proxy-auth.ts'; -import { sha256Hex } from '../utils.ts'; const { default: worker } = await import('../index.ts'); -const secret = 'synthetic-admission-secret'; -const headers = { authorization: `Bearer clawrouter-live-fixture-${secret}`, 'content-type': 'application/json' }; -const provider = { id: 'anthropic', tokenRef: 'anthropic' }; -async function fixture() { - const records = new Map(), calls = [], pending = []; - let beforeRead; - const env = { - ANTHROPIC_API_KEY: 'synthetic-environment', - POLICY_KV: { async get(key) { const action = beforeRead; beforeRead = null; await action?.(key); return structuredClone(Array.isArray(key) ? new Map(key.map(k => [k, records.get(k) ?? null])) : records.get(key) ?? null); } }, - USAGE_QUEUE: { send: async () => {} }, - }; - const authority = createGrantAuthority(undefined, env); - env.ACCESS_CONTROL = { idFromName: name => name, get: () => ({ fetch(url, init) { calls.push(new URL(url).pathname); return authority.fetch(url, init); } }) }; - const policy = { enabled: true, generation: 'v1', providers: ['anthropic'], models: ['anthropic/claude-sonnet-5'], tenantId: 'fixture', monthlyBudgetMicros: null, retainRequestContent: false }; - const credential = { enabled: true, secretSha256: await sha256Hex(secret), policyId: 'fixture', policyGeneration: 'v1', principalId: 'owner@example.com' }; - const call = (path, body) => authorityCall(env, path, body); - await call('/policies/initialize-all', [{ policyId: 'fixture', policy }]); - await call('/credentials/initialize-all', [{ credentialId: 'fixture', credential }]); - await call('/users/initialize-all', [{ email: credential.principalId, record: { enabled: true } }]); - await call('/connections/initialize-all', [{ providerId: provider.id, enabled: true }]); - calls.length = 0; - return { env, records, authority, calls, pending, policy, credential, call, beforeRead(action) { beforeRead = action; }, context: { waitUntil(promise) { pending.push(promise); } } }; -} +import { admissionFixture as fixture, admissionHeaders as headers, admissionProvider as provider } from "./admission-fixture.mjs"; for (const [name, change, expected] of [ ['valid owner', () => {}, null], @@ -102,7 +78,7 @@ for (const [name, setup, expected] of [ assert.deepEqual(f.calls, ['/admission/resolve']); }); -test('warmed subscription inference makes two authority RPCs and streams JSON before EOF/accounting', async t => { +test('legacy subscription inference retains two authority RPCs and streams JSON before EOF/accounting', async t => { const f = await fixture(), key = 'oauth/fixture/anthropic'; f.records.set(key, { provider: provider.id, kind: 'subscription', enabled: true, credentialStore: 'durable_object', hasAccessToken: true, credentialStatus: 'active', updatedAt: '2026-01-01T00:00:00Z' }); f.env.GRANT_CREDENTIALS = { idFromName: name => name, get: () => ({ fetch: async () => Response.json({ grant: { ...f.records.get(key), accessToken: 'synthetic-subscription' }, changed: false, migrated: false }) }) }; diff --git a/worker/test/http-continuation.test.mjs b/worker/test/http-continuation.test.mjs index 79ff6a6..875cf91 100644 --- a/worker/test/http-continuation.test.mjs +++ b/worker/test/http-continuation.test.mjs @@ -178,7 +178,7 @@ test("failed alternate selection preserves the original rejection body and zero let selections = 0, canceled = false; if (scenario === "unavailable") await revokeGrantCredentials(f.env, grantKeys[1]); f.env.ACCESS_CONTROL.beforeFetch = (_name, request) => { - if (new URL(request.url).pathname === "/grant-pools/select" && ++selections === 2 && scenario === "selection_error") throw new Error("fixture selection outage"); + if (new URL(request.url).pathname === "/admission/resolve" && ++selections === 2 && scenario === "selection_error") throw new Error("fixture selection outage"); }; const body = '{"error":{"code":"fixture_rejection"}}'; f.response = () => new Response(new ReadableStream({ @@ -213,7 +213,7 @@ for (const phase of ["selection", "body_cancel"]) { const caller = new AbortController(); let selections = 0, canceled = false; f.env.ACCESS_CONTROL.beforeFetch = (_name, request) => { - if (new URL(request.url).pathname === "/grant-pools/select" && ++selections === 2 && phase === "selection") caller.abort(); + if (new URL(request.url).pathname === "/admission/resolve" && ++selections === 2 && phase === "selection") caller.abort(); }; f.response = () => new Response(new ReadableStream({ cancel() { canceled = true; if (phase === "body_cancel") caller.abort(); } }, { highWaterMark: 0 }), { status: 429 }); const response = await f.request({ max_output_tokens: 32 }, {}, "/v1/responses", caller.signal); @@ -236,7 +236,7 @@ test("the request deadline during alternate selection prevents another dispatch" return setTimer(callback, delay, ...args); }); f.env.ACCESS_CONTROL.beforeFetch = (_name, request) => { - if (new URL(request.url).pathname === "/grant-pools/select" && ++selections === 2) { + if (new URL(request.url).pathname === "/admission/resolve" && ++selections === 2) { assert.equal(typeof deadline, "function"); deadline(); } }; @@ -802,7 +802,7 @@ test("rejection delivery retires the endpoint deadline while an earlier selectio const f = await fixture(t, phase === "selection", { limit: 1_000_000, fixedCost: 7 }), caller = new AbortController(); let selects = 0, cancels = 0; if (phase === "selection") f.env.ACCESS_CONTROL.beforeFetch = (_name, request) => { - if (new URL(request.url).pathname === "/grant-pools/select" && ++selects === 2) { deadline(); throw new Error("fixture selection rejected"); } + if (new URL(request.url).pathname === "/admission/resolve" && ++selects === 2) { deadline(); throw new Error("fixture selection rejected"); } }; f.response = () => new Response(new ReadableStream({ cancel() { cancels++; } }, { highWaterMark: 0 }), { status: 429 }); const response = await f.request({}, {}, "/v1/responses", caller.signal); diff --git a/worker/test/single-hop.test.mjs b/worker/test/single-hop.test.mjs new file mode 100644 index 0000000..60f06c9 --- /dev/null +++ b/worker/test/single-hop.test.mjs @@ -0,0 +1,264 @@ +import './typescript-setup.mjs'; +import assert from 'node:assert/strict'; +import test from 'node:test'; +import { admissionFixture, admissionHeaders } from './admission-fixture.mjs'; +import { attachGrantCredentialNamespace } from './grant-credential-mock.mjs'; +import { putGrantCredentials, revokeGrantCredentials, materializeGrantCredentials } from '../grant-credentials.ts'; +import { resolveGrantSelection } from '../grant-selection.ts'; +import { providerById } from '../providers.ts'; +import { sha256Hex } from '../utils.ts'; +const { default: worker } = await import('../index.ts'); +const provider = providerById('anthropic'); +const endpoint = provider.endpoints.find(item => item.id === 'messages'); +const model = 'claude-sonnet-5'; +const selection = { providerId: provider.id, endpointId: endpoint.id, mode: 'http', model, catalogModelId: 'anthropic/claude-sonnet-5', capability: 'llm.chat' }; +const keyA = 'oauth/fixture/anthropic', keyB = 'oauth/fixture/backup'; +const grant = (extra = {}) => ({ provider: provider.id, kind: 'subscription', enabled: true, accessToken: 'synthetic-token', updatedAt: '2026-01-01T00:00:00Z', ...extra }); +async function fixture() { + const f = await admissionFixture(); + attachGrantCredentialNamespace(f.env, { useExistingAuthority: true }); + f.put = (key, value) => putGrantCredentials(f.env, key, value); + f.admit = () => f.call('/admission/resolve', { credentialId: 'fixture', secretSha256: f.credential.secretSha256, providers: [{ id: provider.id, tokenRef: provider.id }], selection: f.selection ?? selection }); + f.request = () => worker.fetch(new Request('https://fixture/v1/messages', { method: 'POST', headers: admissionHeaders, body: JSON.stringify({ model: 'anthropic/claude-sonnet-5', max_tokens: 16, messages: [] }) }), f.env, f.context); + return f; +} +const observedAt = new Date().toISOString(), resetAt = new Date(Date.now() + 60_000).toISOString(); +function runtime(remaining, extra = {}) { + return { status: 'available', observedAt, source: 'provider_response', lastSignal: 'quota', cooldownUntil: null, grantRevision: '2026-01-01T00:00:00Z', windows: [{ id: 'week', kind: 'requests', remaining, limit: 100, resetAt }], ...extra }; +} + +for (const strategy of ['priority', 'round_robin', 'least_used', 'most_remaining', 'threshold', 'weighted_random']) { + for (const stickiness of ['none', 'identity', 'session']) test(`combined selection parity: ${strategy}/${stickiness}`, async t => { + const old = await fixture(), combined = await fixture(); + t.mock.method(crypto, 'getRandomValues', array => { array.fill(123456789); return array; }); + const routing = { strategy, stickiness, staleState: 'allow', includedQuotaReservePercent: 10 }; + for (const f of [old, combined]) { + await f.put(keyA, grant({ weight: 1 })); await f.put(keyB, grant({ weight: 7 })); + f.policy.grantRouting = routing; + await f.call('/policies/put', { policyId: 'fixture', policy: f.policy }); + await f.call('/grant-pools/feedback', { key: keyA, state: runtime(25) }); + await f.call('/grant-pools/feedback', { key: keyB, state: runtime(90) }); + } + const normalizedSticky = strategy === 'threshold' ? 'none' : stickiness; + const hash = normalizedSticky === 'identity' ? await sha256Hex('identity:owner@example.com') : normalizedSticky === 'session' ? await sha256Hex('session:fixture-session') : null; + combined.selection = { ...selection, sessionHash: await sha256Hex('session:fixture-session') }; + for (let n = 0; n < 4; n++) { + const expected = await resolveGrantSelection(provider.id, 'fixture', 'fixture', provider.id, old.env, new Set(), routing, hash, true, undefined, { provider, endpoint, mode: 'http', model }); + const actual = (await combined.admit()).selection; + assert.ok(actual); + assert.equal(actual.selected?.key, expected.selected?.key); + assert.equal(actual.hasConfiguredGrant, expected.hasConfiguredGrant); + assert.deepEqual(actual.selected?.runtimeState, expected.selected?.runtimeState); + } + assert.deepEqual((await combined.call('/grant-pools/stats', { keys: [keyA, keyB] })).stats && combined.authority.sql.exec('SELECT grant_key, selected_count FROM upstream_grant_selection ORDER BY grant_key'), old.authority.sql.exec('SELECT grant_key, selected_count FROM upstream_grant_selection ORDER BY grant_key')); + }); +} + +for (const scenario of ['priority', 'reserve', 'scoped-exhaustion', 'other-model', 'cooldown', 'stale', 'revision', 'exclusion', 'eligibility', 'environment-fallback']) test(`combined eligibility parity: ${scenario}`, async () => { + const old = await fixture(), combined = await fixture(); + const routing = { strategy: 'most_remaining', includedQuotaReservePercent: scenario === 'reserve' ? 20 : 0, staleState: scenario === 'stale' ? 'deny' : 'allow', staleAfterSeconds: 300, environmentFallback: scenario === 'environment-fallback', eligibleGrants: scenario === 'eligibility' || scenario === 'environment-fallback' ? { anthropic: [] } : {} }; + for (const f of [old, combined]) { + await f.put(keyA, grant({ priority: scenario === 'priority' ? 1 : 100 })); await f.put(keyB, grant()); + f.policy.grantRouting = routing; await f.call('/policies/put', { policyId: 'fixture', policy: f.policy }); + let state = runtime(scenario === 'reserve' ? 10 : 80); + if (scenario === 'scoped-exhaustion' || scenario === 'other-model') state.windows = [{ ...state.windows[0], remaining: 0, models: [scenario === 'scoped-exhaustion' ? 'sonnet' : 'opus'] }]; + if (scenario === 'cooldown') state = { ...state, status: 'cooldown', cooldownUntil: new Date(Date.now() + 60_000).toISOString() }; + if (scenario === 'stale') state.observedAt = '2020-01-01T00:00:00Z'; + if (scenario === 'revision') state = { ...state, grantRevision: 'old-revision', status: 'cooldown', cooldownUntil: new Date(Date.now() + 60_000).toISOString() }; + await f.call('/grant-pools/feedback', { key: keyA, state }); await f.call('/grant-pools/feedback', { key: keyB, state: runtime(50) }); + } + const exclusions = scenario === 'exclusion' ? [keyA] : []; + combined.selection = { ...selection, excludedKeys: exclusions }; + const expected = await resolveGrantSelection(provider.id, 'fixture', 'fixture', provider.id, old.env, new Set(exclusions), routing, null, true, undefined, { provider, endpoint, mode: 'http', model }); + const actual = (await combined.admit()).selection; + assert.ok(actual); + assert.equal(actual.selected?.key, expected.selected?.key); + assert.equal(actual.hasConfiguredGrant, expected.hasConfiguredGrant); +}); + +test('warmed HTTP uses one authority invocation, no KV projection read, and one credential-owner call', async t => { + const f = await fixture(); await f.put(keyA, grant()); await f.admit(); + let ownerCalls = 0; + const get = f.env.GRANT_CREDENTIALS.get; + f.env.GRANT_CREDENTIALS.get = id => { const stub = get(id); return { fetch(...args) { ownerCalls++; return stub.fetch(...args); } }; }; + t.mock.method(f.env.POLICY_KV, 'get', () => { throw new Error('warm inference must not read KV'); }); + t.mock.method(globalThis, 'fetch', async () => Response.json({ usage: { input_tokens: 1, output_tokens: 1 } })); + f.calls.length = 0; + const response = await f.request(); assert.equal(response.status, 200, await response.clone().text()); await response.text(); await Promise.all(f.pending); + assert.deepEqual(f.calls, ['/admission/resolve']); assert.equal(ownerCalls, 1); +}); + +test('failover excludes the failed key through a fresh combined admission', async t => { + const f = await fixture(); await f.put(keyA, grant()); await f.put(keyB, grant()); await f.admit(); + const tokens = []; + t.mock.method(globalThis, 'fetch', async (_url, init) => { tokens.push(new Headers(init.headers).get('authorization')); return tokens.length === 1 ? new Response('rejected', { status: 401 }) : Response.json({ usage: { input_tokens: 1, output_tokens: 1 } }); }); + f.calls.length = 0; const response = await f.request(); assert.equal(response.status, 200); await response.text(); await Promise.all(f.pending); + assert.equal(tokens.length, 2); assert.equal(response.headers.get('x-clawrouter-grant-failover'), '1'); + assert.equal(f.calls.filter(path => path === '/admission/resolve').length, 2); assert.ok(!f.calls.includes('/grant-pools/select') && !f.calls.includes('/grant-pools/resolve')); +}); + +for (const mutation of ['revoke', 'reauth_required', 'disable', 'rotate']) test(`fresh admission/materialization enforces ${mutation}`, async t => { + const f = await fixture(); f.policy.grantRouting = { eligibleGrants: { anthropic: ["anthropic"] } }; await f.call("/policies/put", { policyId: "fixture", policy: f.policy }); await f.put(keyA, grant()); const before = (await f.admit()).selection.selected; + let upstreamCalls = 0; + t.mock.method(globalThis, 'fetch', async (_url, init) => { upstreamCalls++; assert.equal(new Headers(init.headers).get('authorization'), 'Bearer synthetic-rotated'); return Response.json({ usage: { input_tokens: 1, output_tokens: 1 } }); }); + if (mutation === 'revoke') await revokeGrantCredentials(f.env, keyA); + if (mutation === 'disable') { f.policy.enabled = false; await f.call('/policies/put', { policyId: 'fixture', policy: f.policy }); } + if (mutation === 'rotate') await f.put(keyA, grant({ accessToken: 'synthetic-rotated' })); + if (mutation === 'reauth_required') { + const owner = f.env.GRANT_CREDENTIALS.objects.get(keyA); + const record = owner.values.get('credential'); record.status = 'reauth_required'; record.generation++; record.poolSyncPending = true; owner.values.set('credential', record); + await owner.object.fetch(new Request('https://fixture/reconcile', { method: 'POST', body: JSON.stringify({ key: keyA }) })); + } + const current = await f.admit(); + if (mutation === 'rotate') assert.ok(current.selection.selected.grant.credentialGeneration > before.grant.credentialGeneration); + else assert.equal(current.selection?.selected ?? null, null); + const response = await f.request(); await response.text(); await Promise.all(f.pending); + assert.equal(upstreamCalls, mutation === 'rotate' ? 1 : 0); +}); + +test('failed revoke publication demonstrates why authority-generation credential caching is unsafe', async () => { + const f = await fixture(); await f.put(keyA, grant()); await f.admit(); + const original = f.env.ACCESS_CONTROL.get; + f.env.ACCESS_CONTROL.get = id => { const stub = original(id); return { fetch(url, init) { if (new URL(url).pathname === '/grant-pools/publish') throw new Error('synthetic publication outage'); return stub.fetch(url, init); } }; }; + await assert.rejects(() => revokeGrantCredentials(f.env, keyA)); + const stale = (await f.admit()).selection.selected; + assert.equal(stale.key, keyA); assert.equal(stale.grant.credentialStatus, 'active'); + assert.equal(f.env.GRANT_CREDENTIALS.objects.get(keyA).values.get('credential').accessToken, undefined); + await assert.rejects(() => materializeGrantCredentials(f.env, keyA, stale.grant, provider.id, provider.auth.refresh, false)); +}); + +for (const [endpointId, mode] of [['responses', 'http'], ['chat_completions', 'http'], ['embeddings', 'http'], ['responses', 'websocket']]) test(`transport parity: OpenAI ${endpointId}/${mode}`, async () => { + const old = await fixture(), combined = await fixture(), openai = providerById('openai'); + const endpoint = openai.endpoints.find(item => item.id === endpointId); + for (const f of [old, combined]) { + f.policy.providers = ['openai']; f.policy.models = null; + await f.call('/policies/put', { policyId: 'fixture', policy: f.policy }); + await f.put('oauth/fixture/sub', grant({ provider: 'openai', accountId: 'synthetic-account', priority: 1 })); + await f.put('oauth/fixture/api', grant({ provider: 'openai', kind: 'api_key', priority: 100 })); + } + const expected = await resolveGrantSelection('openai', 'fixture', 'fixture', 'openai', old.env, new Set(), undefined, null, true, undefined, { provider: openai, endpoint, mode }); + const actual = await combined.call('/admission/resolve', { credentialId: 'fixture', secretSha256: combined.credential.secretSha256, providers: [{ id: 'openai', tokenRef: 'openai' }], selection: { providerId: 'openai', endpointId, mode, capability: `llm.${endpointId}` } }); + assert.equal(actual.selection.selected.key, expected.selected.key); +}); + +test('legacy projections use the old path until owner publication, without storing secrets', async t => { + const f = await fixture(); + f.records.set(keyA, grant()); + const first = await f.admit(); assert.equal(first.selection, undefined); + assert.deepEqual(f.authority.sql.exec('SELECT projection_json, legacy FROM upstream_grant_projections WHERE grant_key = ?', keyA).map(row => ({ ...row })), [{ projection_json: null, legacy: 1 }]); + let bulkReads = 0; const get = f.env.POLICY_KV.get; + t.mock.method(f.env.POLICY_KV, 'get', async (...args) => { if (Array.isArray(args[0])) bulkReads++; return get(...args); }); + await f.admit(); assert.equal(bulkReads, 0, 'import marker prevents repeating the legacy import'); + t.mock.method(globalThis, 'fetch', async () => Response.json({ usage: { input_tokens: 1, output_tokens: 1 } })); + f.calls.length = 0; + const response = await f.request(); assert.equal(response.status, 200); await response.text(); await Promise.all(f.pending); + assert.ok(f.calls.includes('/grant-pools/select')); + assert.ok((await f.admit()).selection.selected, 'materialization publishes the migrated owner projection'); + assert.ok(!JSON.stringify(f.authority.sql.exec('SELECT * FROM upstream_grant_projections')).includes('synthetic-token')); +}); + +test('bounded projection backfill imports current metadata once; null defaults stay fenced', async t => { + const f = await fixture(); await f.put(keyA, grant()); + f.authority.sql.exec('DELETE FROM upstream_grant_projections'); + let reads = 0; const get = f.env.POLICY_KV.get; + t.mock.method(f.env.POLICY_KV, 'get', async (...args) => { if (Array.isArray(args[0])) { reads++; assert.ok(args[0].length <= 66); } return get(...args); }); + assert.ok((await f.admit()).selection.selected); assert.equal(reads, 1); + await f.admit(); assert.equal(reads, 1); + // A late legacy KV row cannot resurrect a default-key miss after its marker. + f.records.set('oauth/tenants/fixture/anthropic', grant()); + await f.admit(); assert.equal(reads, 1); +}); + +for (const change of ['rotate', 'revoke']) test(`owner ${change} during backfill beats a delayed KV read`, async t => { + const f = await fixture(); await f.put(keyA, grant()); f.authority.sql.exec('DELETE FROM upstream_grant_projections'); + const get = f.env.POLICY_KV.get; let delayed = true; + t.mock.method(f.env.POLICY_KV, 'get', async (...args) => { + const stale = await get(...args); + if (delayed && Array.isArray(args[0])) { delayed = false; if (change === 'rotate') await f.put(keyA, grant({ accessToken: 'synthetic-replacement' })); else await revokeGrantCredentials(f.env, keyA); } + return stale; + }); + const admission = await f.admit(); + if (change === 'revoke') assert.equal(admission.selection.selected, null); + else assert.equal(admission.selection.selected.grant.credentialGeneration, 2); +}); + +test('policy tenant changes during backfill cannot reuse the previous tenant pool', async t => { + const f = await fixture(); await f.put('oauth/tenants/fixture/anthropic', grant()); f.authority.sql.exec('DELETE FROM upstream_grant_projections'); + const get = f.env.POLICY_KV.get; let delayed = true; + t.mock.method(f.env.POLICY_KV, 'get', async (...args) => { + const value = await get(...args); + if (delayed && Array.isArray(args[0])) { delayed = false; f.policy.tenantId = 'other'; await f.call('/policies/put', { policyId: 'fixture', policy: f.policy }); } + return value; + }); + const admission = await f.admit(); + assert.equal(admission.policy.policy.tenantId, 'other'); assert.equal(admission.selection, undefined); assert.deepEqual(admission.pools, {}); +}); + +test('legacy failover uses the freshly admitted tenant instead of the previous tenant', async t => { + const f = await fixture(); + await f.put('oauth/tenants/fixture/a', grant({ accessToken: 'synthetic-old-tenant' })); + await f.put('oauth/tenants/fixture/b', grant({ accessToken: 'synthetic-old-backup' })); + const nextKey = 'oauth/tenants/other/legacy'; + f.records.set(nextKey, grant({ accessToken: 'synthetic-new-tenant' })); f.authority.seedLegacy(nextKey, 'anthropic'); + const tokens = []; + t.mock.method(globalThis, 'fetch', async (_url, init) => { + tokens.push(new Headers(init.headers).get('authorization')); + if (tokens.length === 1) { + f.policy.tenantId = 'other'; await f.call('/policies/put', { policyId: 'fixture', policy: f.policy }); + return new Response('retry', { status: 401 }); + } + return Response.json({ usage: { input_tokens: 1, output_tokens: 1 } }); + }); + const response = await f.request(); assert.equal(response.status, 200); await response.text(); await Promise.all(f.pending); + assert.deepEqual(tokens, ['Bearer synthetic-old-tenant', 'Bearer synthetic-new-tenant']); +}); + +test('a transient KV miss for an attached owner does not install an absence marker', async () => { + const f = await fixture(); const projection = await f.put(keyA, grant()); + f.authority.sql.exec('DELETE FROM upstream_grant_projections'); f.records.delete(keyA); + assert.equal((await f.admit()).selection, undefined); + assert.equal(f.authority.sql.exec('SELECT 1 FROM upstream_grant_projections WHERE grant_key = ?', keyA).length, 0); + f.records.set(keyA, projection); + assert.equal((await f.admit()).selection.selected.key, keyA); +}); + +test('budget rejection preserves the existing pre-budget selection counter semantics', async t => { + const { sqlBudgetNamespace } = await import('./sql-budget-namespace.mjs'); + const outcomes = []; + for (const legacy of [false, true]) { + const f = await fixture(); await f.put(keyA, grant()); + f.policy.monthlyBudgetMicros = 0; f.policy.requestCostMicros = 1; + await f.call('/policies/put', { policyId: 'fixture', policy: f.policy }); + f.env.BUDGET_LEDGER = sqlBudgetNamespace(t); + if (legacy) { + const get = f.env.ACCESS_CONTROL.get; + f.env.ACCESS_CONTROL.get = id => { const stub = get(id); return { fetch(url, init) { + if (new URL(url).pathname === '/admission/resolve') { const body = JSON.parse(init.body); delete body.selection; init = { ...init, body: JSON.stringify(body) }; } + return stub.fetch(url, init); + } }; }; + } + const response = await f.request(); assert.equal(response.status, 402, await response.clone().text()); await response.text(); await Promise.all(f.pending); + outcomes.push((await f.call('/grant-pools/stats', { keys: [keyA] })).stats[keyA].selectedCount); + } + assert.deepEqual(outcomes, [1, 1]); +}); + +test('legacy failover recomputes sticky input after a concurrent routing-policy change', async t => { + const f = await fixture(); await f.put(keyA, grant()); + for (const key of [keyB, 'oauth/fixture/third']) { f.records.set(key, grant()); f.authority.seedLegacy(key, 'anthropic'); } + let sends = 0, retryHash; + const get = f.env.ACCESS_CONTROL.get; + f.env.ACCESS_CONTROL.get = id => { const stub = get(id); return { fetch(url, init) { + if (new URL(url).pathname === '/grant-pools/select' && sends === 1) retryHash = JSON.parse(init.body).stickyHash; + return stub.fetch(url, init); + } }; }; + t.mock.method(globalThis, 'fetch', async () => { + if (++sends === 1) { + f.policy.grantRouting = { stickiness: 'session' }; await f.call('/policies/put', { policyId: 'fixture', policy: f.policy }); + return new Response('retry', { status: 401 }); + } + return Response.json({ usage: { input_tokens: 1, output_tokens: 1 } }); + }); + const response = await f.request(); assert.equal(response.status, 200); await response.text(); await Promise.all(f.pending); + assert.equal(sends, 2); assert.equal(retryHash, await sha256Hex('session:owner@example.com')); +});