fix(server): limited threads through CLIProxyAPI resume when the hub refills

apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts

@@ -47,6 +47,7 @@ import * as ProjectionStore from "./ProjectionStore.ts";
 import * as ProviderEventIngestor from "./ProviderEventIngestor.ts";
 import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
 import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts";
+import * as UsageLimitSources from "../usage/UsageLimitSources.ts";
 import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
 import type { ProviderInstance } from "@t3tools/provider-core/server/driver";
 import { makeProviderFailure } from "@t3tools/provider-core/server/failure";
@@ -233,6 +234,17 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit
   Effect.scoped(
     Effect.gen(function* () {
       const refreshed = yield* Deferred.make<void>();
+      const hubResetAt = "2026-10-30T05:00:00.000Z";
+      const nativeResetAt = "2026-10-30T03:00:00.000Z";
+      const fallbackLookups: Array<{ instanceId: ProviderInstanceId; driver: ProviderDriverKind }> =
+        [];
+      const layerResetFallback = Layer.succeed(UsageLimitSources.UsageLimitResetFallback, {
+        resetAt: (lookup) =>
+          Effect.sync(() => {
+            fallbackLookups.push(lookup);
+            return hubResetAt;
+          }),
+      });
       const adapter = yield* ProviderAdapter.ProviderAdapterV2.pipe(
         Effect.provide(
           Layer.mock(ProviderAdapter.ProviderAdapterV2)({
@@ -290,7 +302,11 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit
           providerInstanceId: instance.instanceId,
           threadId: threadEvent.threadId,
         };
-        for (const failureClass of ["validation_error", "usage_limit"] as const) {
+        for (const [failureClass, resetAt, expectedResetAt] of [
+          ["validation_error", undefined, undefined],
+          ["usage_limit", undefined, hubResetAt],
+          ["usage_limit", nativeResetAt, nativeResetAt],
+        ] as const) {
           const stored = yield* ingestor.ingestNormalized({
             ...input,
             event: {
@@ -302,24 +318,35 @@ it.effect("refreshes the failing instance snapshot after a committed usage_limit
               }),
               providerTurnId: idAllocator.derive.providerTurn({
                 driver: instance.driverKind,
-                nativeTurnId: failureClass,
+                nativeTurnId: `${failureClass}:${resetAt ?? "unknown"}`,
               }),
               runOrdinal: 1,
               failureItemOrdinal: 1,
               status: "failed",
               failure: makeProviderFailure({
                 class: failureClass,
                 message: "Provider rejected turn.",
+                ...(resetAt === undefined ? {} : { resetAt }),
               }),
               threadDisposition: "reusable",
             },
           });
           assert.equal(stored.length, 1);
+          const event = stored[0]?.event;
+          const item = event?.type === "turn-item.updated" ? event.payload : undefined;
+          assert.equal(item?.type === "error" ? item.failure.resetAt : null, expectedResetAt);
           if (failureClass === "validation_error") assert.deepEqual(lookups, []);
         }
-        assert.deepEqual(lookups, [instance.instanceId]);
+        assert.deepEqual(lookups, [instance.instanceId, instance.instanceId]);
+        assert.deepEqual(fallbackLookups, [
+          { instanceId: instance.instanceId, driver: instance.driverKind },
+        ]);
         yield* Deferred.await(refreshed);
-      }).pipe(Effect.provide(makeLayerTest(layerTestEventSink, instances)));
+      }).pipe(
+        Effect.provide(
+          makeLayerTest(layerTestEventSink, instances).pipe(Layer.provide(layerResetFallback)),
+        ),
+      );
     }),
   ),
 );

apps/server/src/orchestration-v2/ProviderEventIngestor.ts

@@ -35,6 +35,7 @@ import { getModelSelectionStringOptionValue, modelSelectionsEqual } from "@t3too
 import * as AnalyticsService from "../telemetry/AnalyticsService.ts";
 import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
 import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts";
+import { UsageLimitResetFallback } from "../usage/UsageLimitSources.ts";
 import * as EventSink from "./EventSink.ts";
 import * as ProjectionStore from "./ProjectionStore.ts";
 import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
@@ -339,6 +340,7 @@ export const layer: Layer.Layer<
     const analytics = yield* ProviderTurnAnalytics;
     const agentScope = yield* AgentScope;
     const instances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry;
+    const usageLimitResets = yield* UsageLimitResetFallback;
     const scope = yield* Effect.scope;
     const completedTurnAnalytics = new Set<string>();
 
@@ -711,9 +713,18 @@ export const layer: Layer.Layer<
                 return dismissed;
               }
               const occurredAt = yield* DateTime.now;
-              const failure = (yield* agentScope.oomKilled(input.threadId))
+              const reported = (yield* agentScope.oomKilled(input.threadId))
                 ? { ...input.event.failure, message: "Killed: out of memory.", code: "oom_kill" }
                 : input.event.failure;
+              const fallbackResetAt =
+                reported.class === "usage_limit" && reported.resetAt == null
+                  ? yield* usageLimitResets.resetAt({
+                      instanceId: input.providerInstanceId,
+                      driver: input.event.driver,
+                    })
+                  : null;
+              const failure =
+                fallbackResetAt === null ? reported : { ...reported, resetAt: fallbackResetAt };
               return [
                 ...dismissed,
                 yield* makeDomainEvent(input, {

apps/server/src/provider/claudeUsageLimits.ts

@@ -47,8 +47,15 @@ export const makeClaudeScopedLimitNames = Ref.make<ClaudeScopedLimitNames>({
   overageIncluded: undefined,
 });
 
+const SCOPED_WINDOW_PREFIX = "seven_day_";
+
 function scopedWindowId(displayName: string): string {
-  return `seven_day_${displayName.toLowerCase().replace(/[^a-z0-9]+/g, "_")}`;
+  return `${SCOPED_WINDOW_PREFIX}${displayName.toLowerCase().replace(/[^a-z0-9]+/g, "_")}`;
+}
+
+/** Model-scoped allowances leave the account usable for other models. */
+export function isClaudeModelScopedWindow(window: Pick<ServerProviderUsageWindow, "id">): boolean {
+  return window.id.startsWith(SCOPED_WINDOW_PREFIX);
 }
 
 function scopedWindow(

apps/server/src/server.ts

@@ -521,6 +521,7 @@ const layerScheduledTaskWebhookOrigin = Layer.effect(
 const layerOrchestrationV2Runtime = RuntimeLayer.layerProduction.pipe(
   Layer.provide(layerScheduledTaskWebhookOrigin),
   Layer.provide(ProviderEventIngestor.layerAnalytics),
+  Layer.provide(UsageLimitSources.layerResetFallback),
   Layer.provide(layerCheckpointStore),
   Layer.provide(layerGitWorkflow),
   Layer.provide(

apps/server/src/usage/UsageLimitSources.test.ts

@@ -1,10 +1,21 @@
 import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
-import { expect, it } from "@effect/vitest";
-import { ServerConfigStreamEvent, UsageLimitSourceId } from "@t3tools/contracts";
+import { describe, expect, it } from "@effect/vitest";
+import {
+  ProviderDriverKind,
+  ProviderInstanceId,
+  ServerConfigStreamEvent,
+  UsageLimitSourceId,
+  type ServerProviderUsageWindow,
+  type UsageLimitSourceSnapshot,
+} from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
 import * as Effect from "effect/Effect";
+import * as Deferred from "effect/Deferred";
+import * as Fiber from "effect/Fiber";
+import * as TestClock from "effect/testing/TestClock";
 import * as Layer from "effect/Layer";
 import * as Schema from "effect/Schema";
-import { HttpClient } from "effect/http";
+import { HttpClient, HttpClientResponse } from "effect/http";
 
 import * as BackgroundPolicy from "../background/BackgroundPolicy.ts";
 import * as ServerSettings from "../serverSettings.ts";
@@ -83,3 +94,205 @@ it.effect("keeps malformed source URLs encodable for config subscribers", () =>
     Effect.scoped,
   ),
 );
+
+it.effect("reads the selected hub while an unrelated refresh is still waiting", () =>
+  Effect.gen(function* () {
+    const stalled = yield* Deferred.make<void>();
+    const release = yield* Deferred.make<void>();
+    const http = HttpClient.make((request) =>
+      Effect.gen(function* () {
+        if (new URL(request.url).hostname === "slow.test") {
+          yield* Deferred.succeed(stalled, undefined);
+          yield* Deferred.await(release);
+        }
+        const body = request.url.endsWith("/auth-files")
+          ? { files: [{ id: "account", auth_index: "one", provider: "claude" }] }
+          : {
+              status_code: 200,
+              body: JSON.stringify({
+                five_hour: { utilization: 100, resets_at: "2099-01-01T00:00:00.000Z" },
+              }),
+            };
+        return HttpClientResponse.fromWeb(request, Response.json(body));
+      }),
+    );
+    const sources = yield* UsageLimitSources.make.pipe(
+      Effect.provideService(HttpClient.HttpClient, http),
+    );
+    yield* Deferred.await(stalled);
+    const fallback = yield* UsageLimitSources.UsageLimitResetFallback.pipe(
+      Effect.provide(UsageLimitSources.layerResetFallback),
+      Effect.provideService(UsageLimitSources.UsageLimitSources, sources),
+    );
+    const result = yield* fallback
+      .resetAt({
+        instanceId: ProviderInstanceId.make("proxy"),
+        driver: ProviderDriverKind.make("claudeAgent"),
+      })
+      .pipe(Effect.forkChild);
+    yield* TestClock.adjust("5 seconds");
+    expect(yield* Fiber.join(result)).toBe("2099-01-01T00:00:00.000Z");
+    expect(yield* Deferred.isDone(release)).toBe(false);
+  }).pipe(
+    Effect.provide([
+      ServerSettings.layerTest({
+        providerInstances: {
+          [ProviderInstanceId.make("proxy")]: {
+            driver: ProviderDriverKind.make("claudeAgent"),
+            environment: [
+              { name: "ANTHROPIC_BASE_URL", value: "https://fast.test", sensitive: false },
+            ],
+          },
+        },
+        usageLimitSources: Object.fromEntries(
+          ["fast", "slow"].map((id) => [
+            id,
+            {
+              kind: "cliproxy" as const,
+              url: `https://${id}.test`,
+              managementKey: "key",
+              enabled: true,
+            },
+          ]),
+        ),
+      }),
+      Layer.mock(BackgroundPolicy.BackgroundPolicy)({
+        shouldRunScopeWork: () => Effect.succeed(false),
+      }),
+      NodeCrypto.layer,
+    ]),
+    Effect.scoped,
+  ),
+);
+
+describe("hubPoolResetAt", () => {
+  const now = Date.parse("2026-10-30T00:00:00.000Z");
+  const hours = (count: number) => DateTime.formatIso(DateTime.makeUnsafe(now + count * 3_600_000));
+  const window = (
+    usedPercent: number,
+    resetsAt?: string,
+    id = resetsAt ?? `window-${usedPercent}`,
+  ): ServerProviderUsageWindow => ({
+    id,
+    kind: "session",
+    label: "5h",
+    usedPercent,
+    ...(resetsAt === undefined ? {} : { resetsAt }),
+  });
+  const hub = (
+    ...accounts: ReadonlyArray<readonly [string, ReadonlyArray<ServerProviderUsageWindow>]>
+  ): UsageLimitSourceSnapshot => ({
+    id: UsageLimitSourceId.make("hub"),
+    kind: "cliproxy",
+    label: "hub",
+    checkedAt: hours(0),
+    accounts: accounts.map(([driver, windows], index) => ({
+      id: `account-${index}`,
+      driver: ProviderDriverKind.make(driver),
+      usageLimits: { checkedAt: hours(0), windows },
+    })),
+  });
+
+  it.each([
+    {
+      name: "the first spent account to recover, after all its spent windows reset",
+      sources: [
+        hub(
+          ["codex", [window(100, hours(2))]],
+          ["codex", [window(100, hours(1)), window(100, hours(3))]],
+        ),
+        hub(["claudeAgent", [window(100, hours(0.5))]]),
+      ],
+      expected: hours(2),
+    },
+    {
+      name: "nothing while an account still has quota",
+      sources: [hub(["codex", [window(100, hours(2))]], ["codex", [window(40, hours(1))]])],
+      expected: null,
+    },
+    {
+      name: "nothing once a spent window's reset has passed",
+      sources: [hub(["codex", [window(100, hours(2))]], ["codex", [window(100, hours(-1))]])],
+      expected: null,
+    },
+    {
+      name: "a recovery time while skipping accounts without readings or reset times",
+      sources: [hub(["codex", []], ["codex", [window(100)]], ["codex", [window(100, hours(4))]])],
+      expected: hours(4),
+    },
+    {
+      name: "a recovery time that ignores spent model-scoped allowances",
+      driver: "claudeAgent",
+      sources: [
+        hub([
+          "claudeAgent",
+          [window(100, hours(2), "five_hour"), window(100, hours(90), "seven_day_fable")],
+        ]),
+      ],
+      expected: hours(2),
+    },
+    {
+      name: "nothing when the hub pools none of the driver's accounts",
+      sources: [hub(["claudeAgent", [window(100, hours(1))]])],
+      expected: null,
+    },
+  ])("returns $name", ({ sources, expected, driver = "codex" }) => {
+    expect(UsageLimitSources.hubPoolResetAt(sources, ProviderDriverKind.make(driver), now)).toBe(
+      expected,
+    );
+  });
+});
+
+describe("instanceHubIds", () => {
+  const source = (url: string, enabled = true) => ({
+    kind: "cliproxy" as const,
+    url,
+    managementKey: "key",
+    enabled,
+  });
+  const settings = {
+    usageLimitSources: {
+      [UsageLimitSourceId.make("fleet")]: source("https://hub.test:8443/"),
+      [UsageLimitSourceId.make("local")]: source("http://127.0.0.1:8317"),
+      [UsageLimitSourceId.make("off")]: source("https://off.test", false),
+      [UsageLimitSourceId.make("plain")]: source("https://plain.test"),
+    },
+    providerInstances: {
+      [ProviderInstanceId.make("claudeProxy")]: {
+        driver: ProviderDriverKind.make("claudeAgent"),
+        environment: [
+          { name: "ANTHROPIC_BASE_URL", value: "https://HUB.test:8443", sensitive: false },
+        ],
+      },
+      [ProviderInstanceId.make("codexProxy")]: {
+        driver: ProviderDriverKind.make("codex"),
+        config: {
+          launchArgs: `-c 'model_providers.p={base_url="http://127.0.0.1:8317/v1"}' https://off.test`,
+        },
+      },
+      [ProviderInstanceId.make("defaultPort")]: {
+        driver: ProviderDriverKind.make("claudeAgent"),
+        environment: [
+          { name: "ANTHROPIC_BASE_URL", value: "https://plain.test:443/v1", sensitive: false },
+          { name: "OTHER_URL", value: "http://hub.test:8443", sensitive: false },
+        ],
+      },
+      [ProviderInstanceId.make("otherPort")]: {
+        driver: ProviderDriverKind.make("codex"),
+        config: { launchArgs: `-c 'model_providers.p={base_url="http://127.0.0.1:83170/v1"}'` },
+      },
+    },
+  };
+
+  it.each([
+    ["claudeProxy", ["fleet"]],
+    ["codexProxy", ["local"]],
+    ["defaultPort", ["plain"]],
+    ["otherPort", []],
+    ["missing", []],
+  ])("matches %s to the enabled hubs its settings name", (instanceId, expected) => {
+    expect([
+      ...UsageLimitSources.instanceHubIds(settings, ProviderInstanceId.make(instanceId)),
+    ]).toEqual(expected);
+  });
+});