feat(server): mint session-scoped member tokens, release 0.0.4

README.md

@@ -252,7 +252,7 @@ Anyone building on Reactor who needs to meter live access to a model:
  ┌─────────────────────┐  WebSocket  ┌────────────────────────────┐  REST   ┌──────────────────────┐
  │ @reactor-team/queue │◀───────────▶│ @reactor-team/queue/server │────────▶│ POST /tokens         │
  │  • partysocket      │   queue +   │  • FIFO queue + session cap │         │ POST /sessions       │
- │  • getJwt() resolver│   tokens    │  • mints 60s Reactor JWTs  │         │ GET  /sessions/{id}  │
+ │  • getJwt() resolver│   tokens    │  • mints scoped Reactor JWTs│         │ GET  /sessions/{id}  │
  │  • sessionId on claim│             │  • creates sessions on claim│         │ DELETE /sessions/{id}│
  │  • zustand store    │             │  • per-user session timer  │         └──────────────────────┘
  └──────────┬──────────┘             │  • stops + reaps sessions  │
@@ -273,10 +273,15 @@ The queue server is the single source of truth. It:
    every connection; the client only adopts them. Creating nothing during grace
    means an abandoned admission never orphans a GPU session.
 3. Mints the Reactor JWT server-side (the API key is a server secret) and sends it
-   only to admitted users.
-4. Issues short-lived tokens (default 60s). The client refreshes them on demand
-   over the WebSocket via a `request_token` command, exposed as a standard `getJwt`
-   resolver for the Reactor SDK.
+   only to admitted users. Each slot gets its own **session-scoped** token
+   (`authorization_details` on `POST /tokens`): it can create one session for the
+   configured model and act only on that session — a leaked token exposes nothing
+   else on the account. The slot's session is created _with_ that token, which is
+   what binds the two together.
+4. Issues each member their slot's token. A scoped token is the session bond and
+   cannot be refreshed mid-session, so it is minted to outlive the slot (grace +
+   session budget; `tokenTtlSeconds` only raises that floor). `request_token` /
+   the `getJwt` resolver re-deliver the stored token.
 5. Gives each admitted user a bounded session (default 120s), then calls
    `DELETE /sessions/{id}` to stop the GPU session when time runs out.
 6. Frees a slot the instant a member leaves — via an explicit `session_ended`
@@ -425,7 +430,7 @@ retuned without a code change. Values resolve **default → `createReactorQueueS
 | `RQ_SESSION_DURATION_MS`         | `sessionDurationMs`         | `120000`                  | Session budget after claim                                                                                         |
 | `RQ_ADMISSION_GRACE_MS`          | `admissionGraceMs`          | `45000`                   | Time to claim a reserved slot                                                                                      |
 | `RQ_WARNING_BEFORE_MS`           | `warningBeforeMs`           | `30000`                   | Lead time for `time_warning`                                                                                       |
-| `RQ_TOKEN_TTL_SECONDS`           | `tokenTtlSeconds`           | `60`                      | Minted JWT lifetime                                                                                                |
+| `RQ_TOKEN_TTL_SECONDS`           | `tokenTtlSeconds`           | `60`                      | Minted JWT lifetime floor (scoped member tokens always cover grace + session)                                      |
 | `RQ_POLL_INTERVAL_MS`            | `pollIntervalMs`            | `15000`                   | Session reconciliation cadence                                                                                     |
 | `RQ_COORDINATOR_URL`             | `coordinatorUrl`            | `https://api.reactor.inc` | Reactor API base URL                                                                                               |
 | `RQ_API_VERSION`                 | `apiVersion`                | `1`                       | `Reactor-API-Version` header                                                                                       |
@@ -610,6 +615,12 @@ mints the JWT, so a custom-acquired session must belong to the same Reactor
 account as the queue's API key. If another client shares the session with the
 queued user, the platform must allow more than one connection per session.
 
+With an `acquireSession` override the queue hands out **unscoped** member
+tokens: the externally acquired session is not bound to any token the queue
+holds, and a session-scoped token can only act on sessions its own grant
+created, so scoping would lock members out. The default session source is what
+enables scoped tokens.
+
 ## Configuration (client)
 
 `ReactorQueueClientOptions` / `<ReactorQueueProvider>` props: `host` (required),

packages/queue/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@reactor-team/queue",
-  "version": "0.0.3",
+  "version": "0.0.4",
   "description": "Drop-in waiting room for Reactor demos: a browser client (framework-agnostic + React) and a PartyKit server that gate access to a capacity-limited app and admit waiting users in order.",
   "license": "Apache-2.0",
   "homepage": "https://github.com/reactor-team/reactor-queue#readme",

packages/queue/src/server/config.ts

@@ -44,7 +44,13 @@ export interface ReactorQueueServerConfig {
   admissionGraceMs?: number;
   /** Lead time for the `time_warning`, in ms. Env: `RQ_WARNING_BEFORE_MS`. */
   warningBeforeMs?: number;
-  /** Requested lifetime for each minted Reactor JWT, in seconds. Env: `RQ_TOKEN_TTL_SECONDS`. */
+  /**
+   * Requested lifetime floor for each minted Reactor JWT, in seconds. Member
+   * tokens are session-scoped and cannot be refreshed mid-session, so they are
+   * minted to cover at least the admission grace plus the full session budget
+   * regardless of this value; it only raises that floor. Env:
+   * `RQ_TOKEN_TTL_SECONDS`.
+   */
   tokenTtlSeconds?: number;
   /** How often to reconcile tracked sessions with Reactor, in ms. Env: `RQ_POLL_INTERVAL_MS`. */
   pollIntervalMs?: number;

packages/queue/src/server/coordinator.ts

@@ -20,17 +20,34 @@ export class CoordinatorError extends Error {
   }
 }
 
+/**
+ * Session authorization scope for a minted JWT. When passed to
+ * {@link CoordinatorClient.mintToken}, the JWT is restricted to creating
+ * sessions for the named model (at most `maxSessions` of them) and to
+ * operating the sessions it created — nothing else on the account.
+ */
+export interface TokenScope {
+  /** Model the token may create sessions for (fully-qualified `org/model`). */
+  model: string;
+  /** How many sessions the token may ever create (spawned, not concurrent). */
+  maxSessions: number;
+}
+
 /**
  * Thin server-side client for the Reactor Coordinator REST API. From inside the
  * trusted PartyKit server it:
  *
  *  1. mints short-lived client JWTs from the API key (`POST /tokens`),
+ *     optionally scoped to one model via `authorization_details`,
  *  2. creates sessions (`POST /sessions`),
  *  3. reads a session's state (`GET /sessions/{id}/runtime`), and
  *  4. stops a session (`DELETE /sessions/{id}`).
  *
- * (2)–(4) need a Bearer JWT, so the client keeps its own cached "server JWT"
- * (minted with a longer TTL) and reuses it across calls.
+ * (2)–(4) need a Bearer JWT. By default the client keeps its own cached
+ * "server JWT" (unscoped, minted with a longer TTL) and reuses it across
+ * calls; `createSession`/`createConnection` also accept an explicit `jwt` so
+ * a session can be created *by* a scoped token, binding it to that token's
+ * grant.
  */
 export class CoordinatorClient {
   private readonly baseUrl: string;
@@ -64,17 +81,37 @@ export class CoordinatorClient {
 
   /**
    * Exchange the API key for a JWT. `ttlSeconds` is passed as `expires_after`;
-   * the Coordinator caps it at its server maximum.
+   * the Coordinator caps it at its server maximum. With a `scope`, the JWT
+   * carries session `authorization_details`: it can only create sessions for
+   * `scope.model` (at most `scope.maxSessions`) and act on the sessions it
+   * created. Bound sessions live on the token's server-side grant, so a
+   * scoped token cannot be re-minted for an existing session.
    */
-  async mintToken(ttlSeconds: number): Promise<{ jwt: string; expiresAt: number }> {
+  async mintToken(
+    ttlSeconds: number,
+    scope?: TokenScope
+  ): Promise<{ jwt: string; expiresAt: number }> {
     const res = await fetch(`${this.baseUrl}/tokens`, {
       method: "POST",
       headers: {
         "Content-Type": "application/json",
         "Reactor-API-Key": this.apiKey,
         ...this.versionHeaders(),
       },
-      body: JSON.stringify({ expires_after: Math.max(1, Math.floor(ttlSeconds)) }),
+      body: JSON.stringify({
+        expires_after: Math.max(1, Math.floor(ttlSeconds)),
+        ...(scope
+          ? {
+              authorization_details: [
+                {
+                  type: "session",
+                  resources: { models: { match: [scope.model] } },
+                  constraints: { max_sessions: scope.maxSessions },
+                },
+              ],
+            }
+          : {}),
+      }),
     });
 
     if (!res.ok) {
@@ -131,9 +168,15 @@ export class CoordinatorClient {
   /**
    * Create a Reactor session for the configured model. Returns the new
    * `session_id`. Runs billing/quota checks against the server's API key.
+   * With `opts.jwt` the session is created by that token — for a scoped
+   * token, this is the step that binds the session to its grant.
    */
-  async createSession(opts: { model: string; webrtcVersion: string }): Promise<string> {
-    const jwt = await this.getServerJwt();
+  async createSession(opts: {
+    model: string;
+    webrtcVersion: string;
+    jwt?: string;
+  }): Promise<string> {
+    const jwt = opts.jwt ?? (await this.getServerJwt());
     const res = await fetch(`${this.baseUrl}/sessions`, {
       method: "POST",
       headers: {
@@ -169,14 +212,15 @@ export class CoordinatorClient {
    * Register a WebRTC connection under an existing session and return the
    * server-minted `connection_id`. This is a transport call (carries
    * `Reactor-WebRTC-Version`, not the API-version headers) and must use a JWT
-   * for the session's owning user — the server JWT, minted from the same API
-   * key that created the session, satisfies that.
+   * allowed to act on the session: pass `opts.jwt` for the scoped token whose
+   * grant owns the session, or omit it to use the server JWT (same-user
+   * ownership, minted from the same API key).
    *
    * A {@link CoordinatorError} with `status === 429` means the session hit its
    * `connections_per_session` cap; the caller falls back to another/new session.
    */
-  async createConnection(sessionId: string): Promise<number> {
-    const jwt = await this.getServerJwt();
+  async createConnection(sessionId: string, opts: { jwt?: string } = {}): Promise<number> {
+    const jwt = opts.jwt ?? (await this.getServerJwt());
     const endpoint = `/sessions/${encodeURIComponent(sessionId)}/transport/webrtc/connections`;
     const res = await fetch(`${this.baseUrl}${endpoint}`, {
       method: "POST",

packages/queue/src/server/server.ts

@@ -73,6 +73,18 @@ interface SlotRecord {
   members: string[];
   createdAt: number;
   lastPollAt?: number;
+  /**
+   * The slot's session-scoped JWT: minted once per slot, used to create the
+   * slot's session (binding the session to the token's grant), and handed to
+   * every member seated here — it is the only token the Coordinator lets act
+   * on this session besides same-user unscoped ones. Never re-minted: a fresh
+   * scoped token would have an empty grant and could not attach. Absent on
+   * slots created before scoped tokens existed and when `acquireSession`
+   * sources sessions externally; those slots fall back to unscoped tokens.
+   */
+  jwt?: string | null;
+  /** Unix seconds when `jwt` expires (from the mint response). */
+  jwtExpiresAt?: number | null;
 }
 
 /** Maps a stable browser id → its current connection, for duplicate-tab eviction. */
@@ -137,6 +149,55 @@ export function createReactorQueueServer(
       return this.api.createSession({ model, webrtcVersion: this.config.webrtcVersion });
     }
 
+    /**
+     * Whether members get session-scoped tokens (the default). Requires the
+     * default session source: an `acquireSession` override creates sessions
+     * under a grant this server does not hold, which a scoped token could
+     * never attach to — so those deployments keep unscoped member tokens.
+     */
+    private get scopedTokens(): boolean {
+      return !this.config.acquireSession;
+    }
+
+    /**
+     * TTL for a slot's scoped token. The token is the session bond and cannot
+     * be refreshed (a re-mint starts a new, empty grant), so it must outlive
+     * the slot: admission grace + the full session budget + a margin. The
+     * configured `tokenTtlSeconds` still acts as a floor.
+     */
+    private slotTokenTtlSeconds(): number {
+      const coverMs = this.config.admissionGraceMs + this.config.sessionDurationMs;
+      return Math.max(this.config.tokenTtlSeconds, Math.ceil(coverMs / 1000) + 120);
+    }
+
+    /**
+     * The slot's scoped token, minted and persisted on first use. Returns null
+     * when the slot vanished mid-mint, or for a legacy slot whose session
+     * already exists without a stored grant (created before scoped tokens):
+     * that session only accepts unscoped same-user tokens, so callers fall
+     * back to the unscoped mint.
+     */
+    private async ensureSlotToken(
+      slotId: string
+    ): Promise<{ jwt: string; expiresAt: number } | null> {
+      const slot = await this.getSlot(slotId);
+      if (!slot) return null;
+      if (slot.jwt) return { jwt: slot.jwt, expiresAt: slot.jwtExpiresAt ?? 0 };
+      if (slot.sessionId) return null;
+
+      const minted = await this.api.mintToken(this.slotTokenTtlSeconds(), {
+        model: this.config.model,
+        maxSessions: 1,
+      });
+      // Re-read before persisting: the mint is a network call and the input
+      // gate was open, so the slot may have been mutated or deleted.
+      const fresh = await this.getSlot(slotId);
+      if (!fresh) return null;
+      if (fresh.jwt) return { jwt: fresh.jwt, expiresAt: fresh.jwtExpiresAt ?? 0 };
+      await this.setSlot({ ...fresh, jwt: minted.jwt, jwtExpiresAt: minted.expiresAt });
+      return minted;
+    }
+
     /**
      * A user left a session. Uses the configured `releaseSession` override, or by
      * default deletes the session via the Reactor API once the last member leaves
@@ -764,7 +825,7 @@ export function createReactorQueueServer(
 
           case "request_token": {
             const member = await this.getMember(sender.id);
-            if (member) await this.mintAndSend(sender);
+            if (member) await this.sendMemberToken(sender);
             break;
           }
 
@@ -969,10 +1030,12 @@ export function createReactorQueueServer(
               connId,
             }
           );
-          // Token can be minted now (it's not session-scoped); the SDK uses it
-          // after claim. getJwt/request_token keep it fresh.
+          // Hand the member their slot's token right away so the SDK has it
+          // before claim. In scoped mode this is the token that will create
+          // (and therefore own) the slot's session; if the claim spills them
+          // to a different slot, claimMember re-sends that slot's token.
           const conn = this.room.getConnection(connId);
-          if (conn) await this.mintAndSend(conn);
+          if (conn) await this.sendMemberToken(conn);
           admittedAny = true;
         }
       }
@@ -1021,6 +1084,23 @@ export function createReactorQueueServer(
         claiming: false,
       });
 
+      // The claim may have seated the member on a different slot than the one
+      // whose token they received at admission (spill / new slot). The attach
+      // token must be the one whose grant owns the session, so re-send the
+      // final slot's token before session_ready. Reads the stored token — no
+      // extra mint — and skips grant-less (legacy/override) sessions, whose
+      // members keep their unscoped admission token.
+      if (this.scopedTokens) {
+        const finalSlot = await this.getSlot(slotId);
+        if (finalSlot?.jwt) {
+          this.sendTo(connId, {
+            type: "token",
+            jwt: finalSlot.jwt,
+            expiresAt: finalSlot.jwtExpiresAt ?? 0,
+          });
+        }
+      }
+
       this.sendTo(connId, {
         type: "session_ready",
         sessionId,
@@ -1103,9 +1183,24 @@ export function createReactorQueueServer(
       if (!slot) return null;
 
       let sessionId = slot.sessionId;
+      let slotJwt = slot.jwt ?? null;
       if (!sessionId) {
         try {
-          sessionId = await this.runAcquire(this.config.model);
+          if (this.scopedTokens) {
+            // Create the session *with the slot's scoped token* — this is the
+            // binding step: the Coordinator adds the session to that token's
+            // grant, making it the token members attach with.
+            const token = await this.ensureSlotToken(slotId);
+            if (!token) return null;
+            slotJwt = token.jwt;
+            sessionId = await this.api.createSession({
+              model: this.config.model,
+              webrtcVersion: this.config.webrtcVersion,
+              jwt: slotJwt,
+            });
+          } else {
+            sessionId = await this.runAcquire(this.config.model);
+          }
         } catch (err) {
           this.reportError("session_create_failed", err, { connId });
           return null;
@@ -1119,7 +1214,12 @@ export function createReactorQueueServer(
 
       for (let attempt = 1; attempt <= CONNECTION_MINT_ATTEMPTS; attempt++) {
         try {
-          const connectionId = await this.api.createConnection(sessionId);
+          // A session created by the slot token must also be operated with it;
+          // grant-less (legacy/override) sessions use the server JWT as before.
+          const connectionId = await this.api.createConnection(
+            sessionId,
+            slotJwt ? { jwt: slotJwt } : {}
+          );
           return { sessionId, connectionId };
         } catch (err) {
           if (err instanceof CoordinatorError && err.status === 429) {
@@ -1187,8 +1287,26 @@ export function createReactorQueueServer(
       return new Promise((resolve) => setTimeout(resolve, ms));
     }
 
-    private async mintAndSend(conn: Party.Connection): Promise<void> {
+    /**
+     * Send a member the token their slot's session accepts. Scoped mode hands
+     * out the slot's own token (minting it on first use); a legacy slot with a
+     * grant-less session, and deployments with an `acquireSession` override,
+     * get the old unscoped mint instead.
+     */
+    private async sendMemberToken(conn: Party.Connection): Promise<void> {
       try {
+        if (this.scopedTokens) {
+          const member = await this.getMember(conn.id);
+          if (!member) return;
+          const token = await this.ensureSlotToken(member.slotId);
+          if (token) {
+            this.send(conn, { type: "token", jwt: token.jwt, expiresAt: token.expiresAt });
+            return;
+          }
+          const slot = await this.getSlot(member.slotId);
+          if (!slot) return;
+          // Legacy slot: fall through to the unscoped mint below.
+        }
         const { jwt, expiresAt } = await this.api.mintToken(this.config.tokenTtlSeconds);
         this.send(conn, { type: "token", jwt, expiresAt });
       } catch (err) {