# Files changed · PR \#19820 · pingdotgg/t3code

[View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/19820/diffs)

Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code)

Visibility: public

[fix\(server\): preserve hub usage readings after provider rate limits](https://git.cafe/pingdotgg/t3code/pull/19820.md)

Version: 11

Head commit: df2da49b4f2482cb82e134a6b645f90c3d42029a

Base commit: d36b7da2485bc889b320a285deea181f637d760a

Comparison base: d36b7da2485bc889b320a285deea181f637d760a

Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing.

## apps/server/src/usage/cliproxyApi.test.ts

Status: modified; +237; −4

```diff
@@ -1,10 +1,11 @@
 import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
 import { describe, expect, it } from "@effect/vitest";
+import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as Fiber from "effect/Fiber";
 import * as Schema from "effect/Schema";
 import * as TestClock from "effect/testing/TestClock";
-import { HttpClient, HttpClientError, HttpClientResponse } from "effect/http";
+import { HttpClient, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/http";
 
 import { makeCliproxyApi } from "./cliproxyApi.ts";
 
@@ -57,11 +58,13 @@
     >;
     upstream?: (request: RequestBody) => { status: number; body: unknown };
     cooldownStatus?: number;
+    beforeCooldownReset?: Effect.Effect<void>;
+    beforeUpstreamResponse?: (request: RequestBody) => Effect.Effect<void>;
   } = {},
 ) {
   const requests: Array<{ path: string; body?: RequestBody }> = [];
-  const http = HttpClient.make((request) =>
-    Effect.sync(() => {
+  const respond = (request: HttpClientRequest.HttpClientRequest) =>
+    Effect.gen(function* () {
       expect(request.headers.authorization).toBe("Bearer management-secret");
       const path = new URL(request.url).pathname;
       const body =
@@ -105,11 +108,19 @@
                 },
               },
       };
+      if (body && options.beforeUpstreamResponse) yield* options.beforeUpstreamResponse(body);
       return HttpClientResponse.fromWeb(
         request,
         Response.json({ status_code: upstream.status, body: encodeJson(upstream.body) }),
       );
+    });
+  const http = HttpClient.make((request) =>
+    Effect.andThen(
+      request.url.endsWith("/reset-quota")
+        ? (options.beforeCooldownReset ?? Effect.void)
+        : Effect.void,
-    }),
+      respond(request),
+    ),
   );
   return {
     requests,
@@ -379,6 +390,228 @@
     }),
   );
 
+  it.effect(
+    "paces usage probes, retains recent readings through 429s, and expires them during backoff",
+    () =>
+      Effect.gen(function* () {
+        let limited = false;
+        let probes = 0;
+        const test = fixture({
+          accounts: [{ ...accounts[0]!, provider: "claude" }],
+          upstream: () => {
+            probes++;
+            return limited
+              ? { status: 429, body: {} }
+              : { status: 200, body: { five_hour: { utilization: probes * 10, resets_at: null } } };
+          },
+        });
+        const api = yield* test.api;
+        const [initial] = yield* api.readAccounts(config);
+        expect(initial?.usageLimits.windows[0]?.usedPercent).toBe(10);
+        limited = true;
+        for (const minutes of [1, 2, 4, 8]) {
+          const previousProbes = probes;
+          yield* TestClock.adjust(minutes * 60_000 - 1);
+          expect((yield* api.readAccounts(config))[0]?.usageLimits).toEqual(initial?.usageLimits);
+          expect(probes).toBe(previousProbes);
+          yield* TestClock.adjust(1);
+          const [account] = yield* api.readAccounts(config);
+          expect(probes).toBe(previousProbes + 1);
+          if (minutes < 8) {
+            expect(account?.usageLimits).toEqual(initial?.usageLimits);
+          } else {
+            expect(account?.usageLimits.windows).toEqual([]);
+            expect(account?.usageLimits.unavailable).toMatchObject({
+              reason: "probeFailed",
+              message: expect.stringContaining("rate limiting"),
+            });
+          }
+        }
+        limited = false;
+        yield* TestClock.adjust(8 * 60_000 - 1);
+        expect((yield* api.readAccounts(config))[0]?.usageLimits.unavailable?.message).toContain(
+          "rate limiting",
+        );
+        expect(probes).toBe(5);
+        yield* TestClock.adjust(1);
+        expect((yield* api.readAccounts(config))[0]?.usageLimits.windows[0]?.usedPercent).toBe(60);
+        yield* TestClock.adjust("1 minute");
+        expect((yield* api.readAccounts(config))[0]?.usageLimits.windows[0]?.usedPercent).toBe(70);
+      }),
+  );
+
+  it.effect("isolates cached readings by hub, file, provider, email, and workspace", () =>
+    Effect.gen(function* () {
+      const original = accounts[0]!;
+      const listed = [{ ...original }];
+      let probes = 0;
+      const test = fixture({
+        accounts: listed,
+        upstream: (request) => {
+          if (request.url?.endsWith("/rate-limit-reset-credits"))
+            return { status: 200, body: { credits: [] } };
+          const percent = ++probes * 10;
+          return {
+            status: 200,
+            body: {
+              five_hour: { utilization: percent, resets_at: null },
+              rate_limit: { primary_window: { used_percent: percent } },
+            },
+          };
+        },
+      });
+      const api = yield* test.api;
+      const baseline = yield* api.readAccounts(config);
+      const identities = [
+        { url: "http://other-hub.test:8317", account: original },
+        { url: config.url, account: { ...original, id: "replacement.json" } },
+        { url: config.url, account: { ...original, provider: "claude" } },
+        { url: config.url, account: { ...original, email: "replacement@example.com" } },
+        {
+          url: config.url,
+          account: { ...original, id_token: { chatgpt_account_id: "other-workspace" } },
+        },
+      ];
+      for (const [index, identity] of identities.entries()) {
+        listed[0] = identity.account;
+        const [account] = yield* api.readAccounts({ ...config, url: identity.url });
+        expect(account).toMatchObject({
+          id: identity.account.id,
+          email: identity.account.email,
+          workspaceId: identity.account.id_token.chatgpt_account_id,
+          driver: identity.account.provider === "claude" ? "claudeAgent" : "codex",
+        });
+        expect(account?.usageLimits.windows[0]?.usedPercent).toBe((index + 2) * 10);
+      }
+      listed[0] = original;
+      expect(yield* api.readAccounts(config)).toEqual(baseline);
+      expect(probes).toBe(6);
+    }),
+  );
+
+  it.effect(
+    "shares concurrent reads without blocking other identities or retaining a pre-reset reading",
+    () =>
+      Effect.gen(function* () {
+        const started = yield* Deferred.make<void>();
+        const release = yield* Deferred.make<void>();
+        const spent = yield* Deferred.make<void>();
+        let probes = 0;
+        const test = fixture({
+          accounts: [accounts[0]!],
+          upstream: (request) => ({
+            status: 200,
+            body: request.url?.endsWith("/consume")
+              ? { code: "reset" }
+              : request.url?.endsWith("/rate-limit-reset-credits")
+                ? { credits: [] }
+                : { rate_limit: { primary_window: { used_percent: ++probes * 10 } } },
+          }),
+          beforeUpstreamResponse: (request) =>
+            request.url?.endsWith("/wham/usage") && probes === 1
+              ? Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(release)))
+              : Effect.void,
+          beforeCooldownReset: Deferred.succeed(spent, undefined).pipe(Effect.asVoid),
+        });
+        const api = yield* test.api;
+        const first = yield* Effect.forkChild(api.readAccounts(config));
+        yield* Deferred.await(started);
+        const second = yield* Effect.forkChild(api.readAccounts(config), {
+          startImmediately: true,
+        });
+        const [other] = yield* api.readAccounts({ ...config, url: "http://other-hub.test:8317" });
+        expect(other?.usageLimits.windows[0]?.usedPercent).toBe(20);
+        const redeem = yield* Effect.forkChild(api.consume(config, "first.json", "credit"));
+        yield* Deferred.await(spent);
+        yield* Deferred.succeed(release, undefined);
+        expect((yield* Fiber.join(first))[0]?.usageLimits.windows[0]?.usedPercent).toBe(10);
+        expect((yield* Fiber.join(second))[0]?.usageLimits.windows[0]?.usedPercent).toBe(10);
+        expect((yield* Fiber.join(redeem)).outcome).toBe("reset");
+        const [fresh] = yield* api.readAccounts(config);
+        expect(fresh?.usageLimits.windows[0]?.usedPercent).toBe(30);
+        expect(probes).toBe(3);
+      }),
+  );
+
+  it.effect(
+    "retains pacing after a rejected reset but invalidates only the attempted account after success or uncertain failure",
+    () =>
+      Effect.gen(function* () {
+        let resetStatus = 429;
+        let limited = false;
+        let percent = 12;
+        const test = fixture({
+          upstream: (request) =>
+            request.url?.endsWith("/consume")
+              ? { status: resetStatus, body: { code: "reset" } }
+              : limited
+                ? { status: 429, body: {} }
+                : {
+                    status: 200,
+                    body: request.url?.endsWith("/rate-limit-reset-credits")
+                      ? { credits: [] }
+                      : { rate_limit: { primary_window: { used_percent: percent } } },
+                  },
+        });
+        const api = yield* test.api;
+        const usageProbes = () =>
+          test.requests.filter((request) => request.body?.url?.endsWith("/wham/usage")).length;
+        const initial = yield* api.readAccounts(config);
+        limited = true;
+        yield* TestClock.adjust("1 minute");
+        expect(yield* api.readAccounts(config)).toEqual(initial);
+        expect(usageProbes()).toBe(4);
+        expect((yield* Effect.flip(api.consume(config, "first.json", "credit"))).detail).toContain(
+          "rate limiting",
+        );
+        yield* TestClock.adjust("1 minute");
+        expect(yield* api.readAccounts(config)).toEqual(initial);
+        expect(usageProbes()).toBe(4);
+
+        resetStatus = 200;
+        limited = false;
+        percent = 0;
+        expect((yield* api.consume(config, "first.json", "credit")).outcome).toBe("reset");
+        const afterReset = yield* api.readAccounts(config);
+        expect(afterReset.map((account) => account.usageLimits.windows[0]?.usedPercent)).toEqual([
+          0, 12,
+        ]);
+        expect(usageProbes()).toBe(5);
+
+        resetStatus = 503;
+        limited = true;
+        expect((yield* Effect.result(api.consume(config, "first.json", "next-credit")))._tag).toBe(
+          "Failure",
+        );
+        const afterFailure = yield* api.readAccounts(config);
+        expect(afterFailure[0]?.usageLimits.windows).toEqual([]);
+        expect(afterFailure[0]?.usageLimits.unavailable?.message).toContain("rate limiting");
+        expect(afterFailure[1]).toEqual(initial[1]);
+        expect(usageProbes()).toBe(6);
+      }),
+  );
+
+  it.effect("invalidates an account when reset is interrupted after the credit is spent", () =>
+    Effect.gen(function* () {
+      const spent = yield* Deferred.make<void>();
+      const test = fixture({
+        beforeCooldownReset: Deferred.succeed(spent, undefined).pipe(Effect.andThen(Effect.never)),
+      });
+      const api = yield* test.api;
+      yield* api.readAccounts(config);
+      const redeem = yield* Effect.forkChild(api.consume(config, "second.json", "credit-b"));
+      yield* Deferred.await(spent);
+      yield* Fiber.interrupt(redeem);
+      yield* api.readAccounts(config);
+      expect(
+        test.requests
+          .filter((request) => request.body?.url?.endsWith("/wham/usage"))
+          .map((request) => request.body?.auth_index)
+          .sort(),
+      ).toEqual(["a", "b", "b"]);
+    }),
+  );
+
   it.effect("pins redemption to the displayed credit and clears only that account's cooldown", () =>
     Effect.gen(function* () {
       const test = fixture();
```

## apps/server/src/usage/cliproxyApi.ts

Status: modified; +113; −17

```diff
@@ -5,11 +5,16 @@
   type UsageLimitSourceAccount,
   type UsageLimitSourceConfig,
 } from "@t3tools/contracts";
+import * as KeyedLock from "@t3tools/shared/KeyedLock";
 import * as Cause from "effect/Cause";
+import * as Clock from "effect/Clock";
 import * as Crypto from "effect/Crypto";
 import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
 import * as Hex from "effect/encoding/Hex";
+import * as Ref from "effect/Ref";
+import * as Result from "effect/Result";
 import * as Schema from "effect/Schema";
 import { HttpClient, HttpClientError, HttpClientRequest, HttpClientResponse } from "effect/http";
 
@@ -18,6 +23,27 @@
 import { claudeUsageResponseToLimits } from "../provider/claudeUsageLimits.ts";
 import { makeUnavailableUsageLimits } from "@t3tools/provider-core/server/usageLimits";
 
+class HubProviderRateLimitedError extends Schema.TaggedError<HubProviderRateLimitedError>()(
+  "HubProviderRateLimitedError",
+  {},
+) {
+  override get message(): string {
+    return "The provider is rate limiting requests from the hub. Try again in a few minutes.";
+  }
+}
+const isHubProviderRateLimitedError = Schema.is(HubProviderRateLimitedError);
+
+const USAGE_READ_INTERVAL_MS = 60_000;
+const USAGE_READ_MAX_BACKOFF_MS = 8 * 60_000;
+const USAGE_READ_RETENTION_MS = 15 * 60_000;
+
+interface UsageReadState {
+  readonly nextReadAt: number;
+  readonly intervalMs: number;
+  readonly last: { readonly value: UsageLimitSourceAccount; readonly at: number } | undefined;
+  readonly message: string;
+}
+
 const AuthFile = Schema.Struct({
   id: Schema.String,
   auth_index: Schema.String,
@@ -223,6 +249,7 @@
       ...(data === undefined ? {} : { data: yield* encodeJson(data) }),
     });
     const response = yield* decodeApiResponse(raw);
+    if (response.status_code === 429) return yield* new HubProviderRateLimitedError();
     if (response.status_code < 200 || response.status_code >= 300) {
       return yield* new UsageLimitSourceError({
         detail: `The provider refused the hub request (HTTP ${response.status_code}).`,
@@ -248,6 +275,17 @@
       .toSorted((a, b) => Date.parse(a.expires_at) - Date.parse(b.expires_at));
   });
 
+  const usageReads = yield* Ref.make<ReadonlyMap<string, UsageReadState>>(new Map());
+  const usageReadLock = yield* KeyedLock.make<string>();
+  const usageReadKey = (config: UsageLimitSourceConfig, account: typeof AuthFile.Type) =>
+    JSON.stringify([
+      config.url,
+      account.id,
+      account.provider,
+      account.email ?? "",
+      account.id_token?.chatgpt_account_id ?? "",
+    ]);
+
   const readAccount = Effect.fn("CliproxyApi.readAccount")(function* (
     config: UsageLimitSourceConfig,
     account: typeof AuthFile.Type,
@@ -336,15 +374,50 @@
         },
       };
     });
-    return yield* read.pipe(
-      Effect.orElseSucceed(() => ({
-        ...base,
-        usageLimits: makeUnavailableUsageLimits({
-          checkedAt,
-          reason: "probeFailed",
-          message: "The hub could not read this account's usage.",
-        }),
-      })),
+    const key = usageReadKey(config, account);
+    return yield* usageReadLock.withLock(
+      key,
+      Effect.gen(function* () {
+        const now = yield* Clock.currentTimeMillis;
+        const previous = (yield* Ref.get(usageReads)).get(key);
+        let entry = previous;
+        if (!entry || now >= entry.nextReadAt) {
+          const result = yield* Effect.result(read);
+          const intervalMs = Result.isSuccess(result)
+            ? USAGE_READ_INTERVAL_MS
+            : Math.min(
+                (entry?.intervalMs ?? USAGE_READ_INTERVAL_MS) * 2,
+                USAGE_READ_MAX_BACKOFF_MS,
+              );
+          entry = {
+            nextReadAt: now + intervalMs,
+            intervalMs,
+            last: Result.isSuccess(result) ? { value: result.success, at: now } : entry?.last,
+            message:
+              Result.isFailure(result) && isHubProviderRateLimitedError(result.failure)
+                ? result.failure.message
+                : "The hub could not read this account's usage.",
+          };
+        }
+        // A slow failed probe can cross the retention boundary.
+        const finishedAt = yield* Clock.currentTimeMillis;
+        if (entry.last && finishedAt - entry.last.at >= USAGE_READ_RETENTION_MS) {
+          entry = { ...entry, last: undefined };
+        }
+        if (entry !== previous) {
+          const next = entry;
+          yield* Ref.update(usageReads, (current) => new Map(current).set(key, next));
+        }
+        if (entry.last) return entry.last.value;
+        return {
+          ...base,
+          usageLimits: makeUnavailableUsageLimits({
+            checkedAt,
+            reason: "probeFailed",
+            message: entry.message,
+          }),
+        };
+      }),
     );
   });
 
@@ -373,13 +446,7 @@
     accountId: string,
     creditId: string,
   ): Effect.fn.Return<ProviderConsumeResetCreditResult, UsageLimitSourceError> {
-    const operation = Effect.gen(function* () {
-      const account = (yield* authFiles(config)).find((account) => account.id === accountId);
-      if (!account || account.disabled || account.provider !== "codex") {
-        return yield* new UsageLimitSourceError({
-          detail: "The Codex hub account is missing or disabled.",
-        });
+    const redeem = Effect.fnUntraced(function* (account: typeof AuthFile.Type) {
-      }
       const body = yield* apiCall(config, account, `${CREDIT_URL}/consume`, {
         redeem_request_id: yield* creditRedeemRequestId(
           account.id_token?.chatgpt_account_id ?? account.id,
@@ -410,12 +477,41 @@
           : {}),
       } as const;
     });
+    const operation = Effect.gen(function* () {
+      const account = (yield* authFiles(config)).find((account) => account.id === accountId);
+      if (!account || account.disabled || account.provider !== "codex") {
+        return yield* new UsageLimitSourceError({
+          detail: "The Codex hub account is missing or disabled.",
+        });
+      }
+      const key = usageReadKey(config, account);
+      return yield* redeem(account).pipe(
+        // Only an explicit provider 429 guarantees that no credit was spent.
+        Effect.onExit((exit) => {
+          const failure = Exit.findError(exit);
+          if (Result.isSuccess(failure) && isHubProviderRateLimitedError(failure.success)) {
+            return Effect.void;
+          }
+          // Wait for older probes before dropping their result.
+          return usageReadLock.withLock(
+            key,
+            Ref.update(usageReads, (current) => {
+              const next = new Map(current);
+              next.delete(key);
+              return next;
+            }),
+          );
+        }),
+      );
+    });
     return yield* operation.pipe(
       Effect.mapError((error) =>
         isUsageLimitSourceError(error)
           ? error
           : new UsageLimitSourceError({
-              detail: "The hub returned an unexpected reset-credit response.",
+              detail: isHubProviderRateLimitedError(error)
+                ? error.message
+                : "The hub returned an unexpected reset-credit response.",
             }),
       ),
     );
```

## docs/user/usage.md

Status: modified; +5; −2

```diff
@@ -104,7 +104,8 @@
 
 Opening Limits checks the selected connected environments automatically. Each client waits at
 least five minutes between automatic checks of an environment, including after a failed check.
-If a window still looks stale, refresh Limits to re-check every provider and hub.
+If a window still looks stale, refresh Limits to re-check every provider and hub. Hub accounts may
+reuse their last reading while a refresh interval or failure backoff is active.
 
 Pick `/usage-limits` from the composer's command menu, or send it as a message, to check the
 current model's limits without leaving the conversation. The result opens above the composer and
@@ -135,7 +136,9 @@
 environment that will connect to the hub and enter its URL and management key.
 
 The accounts appear under **Usage → Limits**. Codex accounts show banked reset credits; select an
+account and choose **Use reset** to redeem one. No hub plugin is required. Hub readings refresh at
+most once a minute, with longer pauses after failures; if a read fails, the last successful reading
+stays visible until it is fifteen minutes old.
-account and choose **Use reset** to redeem one. No hub plugin is required.
 
 This connection supplies usage information; configure
 the provider separately to send agent requests through the hub. Remove the hub from the same
```