feat(usage): thread details card shows the hub account each thread was routed to

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

@@ -2,6 +2,8 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
 import { expect, it } from "@effect/vitest";
 import { SecretProviderId, ServerConfigStreamEvent, UsageLimitSourceId } from "@t3tools/contracts";
 import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretResolver";
+import * as Deferred from "effect/Deferred";
+import * as TestClock from "effect/testing/TestClock";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Schema from "effect/Schema";
@@ -157,3 +159,179 @@ it.effect("reads a management key from its secret provider", () => {
     Effect.scoped,
   );
 });
+
+it.effect("reads routed sessions from opted-in hubs, even with no client open", () =>
+  Effect.gen(function* () {
+    const queueReads: string[] = [];
+    const http = HttpClient.make((request) =>
+      Effect.sync(() => {
+        expect(request.headers.authorization).toBe("Bearer key");
+        const url = new URL(request.url);
+        if (url.pathname.endsWith("/usage-queue")) {
+          queueReads.push(url.host);
+          return HttpClientResponse.fromWeb(
+            request,
+            Response.json([{ session_id: "thread-session", auth_index: "a" }]),
+          );
+        }
+        if (url.pathname.endsWith("/auth-files")) {
+          return HttpClientResponse.fromWeb(
+            request,
+            Response.json({
+              files: [{ id: "claude-a.json", auth_index: "a", provider: "claude" }],
+            }),
+          );
+        }
+        return HttpClientResponse.fromWeb(
+          request,
+          Response.json({ status_code: 200, body: JSON.stringify({}) }),
+        );
+      }),
+    );
+    const sources = yield* UsageLimitSources.make.pipe(
+      Effect.provideService(HttpClient.HttpClient, http),
+    );
+    yield* sources.refresh;
+    yield* TestClock.adjust("30 seconds");
+
+    const snapshots = yield* sources.current;
+    expect(queueReads).toEqual(["routed.test"]);
+    expect(
+      snapshots.map((source) => [source.id, source.accounts.map((account) => account.sessionIds)]),
+    ).toEqual([
+      ["routed", [["thread-session"]]],
+      ["observed", [undefined]],
+    ]);
+
+    // Switching Thread accounts off drops the routes on that refresh, not on the next queue read.
+    const settings = yield* ServerSettings.ServerSettingsService;
+    yield* settings.updateSettings({
+      usageLimitSources: {
+        [UsageLimitSourceId.make("routed")]: {
+          kind: "cliproxy",
+          url: "https://routed.test",
+          managementKey: "key",
+          enabled: true,
+          threadAccounts: false,
+        },
+      },
+    });
+    yield* sources.refresh;
+    const afterOff = yield* sources.current;
+    expect(afterOff.find((source) => source.id === "routed")?.accounts[0]?.sessionIds).toBe(
+      undefined,
+    );
+  }).pipe(
+    Effect.provide([
+      Layer.succeed(SecretResolver.SecretResolver, {
+        read: (references) => Effect.succeed(references.map(() => "key")),
+        invalidate: () => Effect.void,
+        label: (provider) => Effect.succeed(provider),
+      }),
+      ServerSettings.layerTest({
+        usageLimitSources: {
+          [UsageLimitSourceId.make("routed")]: {
+            kind: "cliproxy",
+            url: "https://routed.test",
+            managementKey: {
+              provider: SecretProviderId.make("homelab"),
+              reference: "cliproxy#key",
+            },
+            enabled: true,
+            threadAccounts: true,
+          },
+          [UsageLimitSourceId.make("observed")]: {
+            kind: "cliproxy",
+            url: "https://observed.test",
+            managementKey: "key",
+            enabled: true,
+          },
+        },
+      }),
+      // No client lease: a turn that ends while nobody watches must still keep its route.
+      Layer.mock(BackgroundPolicy.BackgroundPolicy)({
+        shouldRunScopeWork: () => Effect.succeed(false),
+      }),
+      NodeCrypto.layer,
+    ]),
+    Effect.scoped,
+  ),
+);
+
+it.effect("drops a queue read that lands after Thread accounts was switched off", () =>
+  Effect.gen(function* () {
+    const release = yield* Deferred.make<void>();
+    const http = HttpClient.make((request) => {
+      const url = new URL(request.url);
+      if (url.pathname.endsWith("/usage-queue")) {
+        return Deferred.await(release).pipe(
+          Effect.as(
+            HttpClientResponse.fromWeb(
+              request,
+              Response.json([{ session_id: "thread-session", auth_index: "a" }]),
+            ),
+          ),
+        );
+      }
+      return Effect.succeed(
+        HttpClientResponse.fromWeb(
+          request,
+          url.pathname.endsWith("/auth-files")
+            ? Response.json({
+                files: [{ id: "claude-a.json", auth_index: "a", provider: "claude" }],
+              })
+            : Response.json({ status_code: 200, body: JSON.stringify({}) }),
+        ),
+      );
+    });
+    const sources = yield* UsageLimitSources.make.pipe(
+      Effect.provideService(HttpClient.HttpClient, http),
+    );
+    yield* sources.refresh;
+    // The 30-second queue read starts and waits on the hub.
+    yield* TestClock.adjust("30 seconds");
+
+    const settings = yield* ServerSettings.ServerSettingsService;
+    yield* settings.updateSettings({
+      usageLimitSources: {
+        [UsageLimitSourceId.make("routed")]: {
+          kind: "cliproxy",
+          url: "https://routed.test",
+          managementKey: "key",
+          enabled: true,
+          threadAccounts: false,
+        },
+      },
+    });
+    yield* sources.refresh;
+    yield* Deferred.succeed(release, undefined);
+    yield* TestClock.adjust("1 second");
+
+    const snapshots = yield* sources.current;
+    expect(snapshots[0]?.accounts[0]?.sessionIds).toBe(undefined);
+  }).pipe(
+    Effect.provide([
+      Layer.succeed(SecretResolver.SecretResolver, {
+        read: (references) => Effect.succeed(references.map(() => undefined)),
+        invalidate: () => Effect.void,
+        label: (provider) => Effect.succeed(provider),
+      }),
+      ServerSettings.layerTest({
+        usageLimitSources: {
+          [UsageLimitSourceId.make("routed")]: {
+            kind: "cliproxy",
+            url: "https://routed.test",
+            managementKey: "key",
+            enabled: true,
+            threadAccounts: true,
+          },
+        },
+      }),
+      Layer.mock(BackgroundPolicy.BackgroundPolicy)({
+        shouldRunScopeWork: () => Effect.succeed(false),
+      }),
+      NodeCrypto.layer,
+    ]),
+    Effect.scoped,
+  ),
+);

apps/server/src/usage/UsageLimitSources.ts

@@ -40,6 +40,31 @@ export class UsageLimitSources extends Context.Service<
   }
 >()("t3/usage/UsageLimitSources") {}
 
+const ROUTE_READ_INTERVAL = Duration.seconds(30);
+const MAX_ROUTES_PER_SOURCE = 200;
+
+function withSessionIds(
+  snapshots: ReadonlyArray<UsageLimitSourceSnapshot>,
+  routes: ReadonlyMap<UsageLimitSourceId, ReadonlyMap<string, string>>,
+): ReadonlyArray<UsageLimitSourceSnapshot> {
+  return snapshots.map((source) => {
+    const sessionIdsByAccount = new Map<string, string[]>();
+    for (const [sessionId, accountId] of routes.get(source.id) ?? []) {
+      const sessionIds = sessionIdsByAccount.get(accountId);
+      if (sessionIds) sessionIds.push(sessionId);
+      else sessionIdsByAccount.set(accountId, [sessionId]);
+    }
+    if (sessionIdsByAccount.size === 0) return source;
+    return {
+      ...source,
+      accounts: source.accounts.map((account) => {
+        const sessionIds = sessionIdsByAccount.get(account.id);
+        return sessionIds ? { ...account, sessionIds } : account;
+      }),
+    };
+  });
+}
+
 function sourceLabel(id: string, config: UsageLimitSourceConfig): string {
   if (config.label) return config.label;
   const fallback = id.trim() || "Usage limit source";
@@ -62,6 +87,11 @@ export const make = Effect.gen(function* () {
   const settingsService = yield* Settings.ServerSettingsService;
   const secrets = yield* SecretResolver.SecretResolver;
   const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;
+  const readsRef = yield* Ref.make<ReadonlyArray<UsageLimitSourceSnapshot>>([]);
+  // Per source, provider session id to hub account id, oldest route first.
+  const routesRef = yield* Ref.make<ReadonlyMap<UsageLimitSourceId, ReadonlyMap<string, string>>>(
+    new Map(),
+  );
   const readManagementKey = Effect.fn("UsageLimitSources.readManagementKey")(function* (
     config: UsageLimitSourceConfig,
   ) {
@@ -94,14 +124,28 @@ export const make = Effect.gen(function* () {
     return { ...base, accounts: accounts.success };
   });
 
+  // Its own lock, so a route read never waits on a slow quota refresh yet never publishes stale reads.
+  const publishLock = yield* Semaphore.make(1);
   const publish = Effect.fn("UsageLimitSources.publish")(function* (
-    next: ReadonlyArray<UsageLimitSourceSnapshot>,
+    reads: ReadonlyArray<UsageLimitSourceSnapshot> | undefined,
   ) {
+    if (reads) yield* Ref.set(readsRef, reads);
+    // Current settings, so no publish names routes for a hub switched out of Thread accounts,
+    // even when a queue read started before the switch lands after it.
+    const settings = yield* settingsService.getSettings.pipe(
+      Effect.orElseSucceed((): ServerSettings | null => null),
+    );
+    const routes = new Map(
+      [...(yield* Ref.get(routesRef))].filter(
+        ([id]) => settings?.usageLimitSources[id]?.threadAccounts,
+      ),
+    );
+    const next = withSessionIds(yield* Ref.get(readsRef), routes);
     const changed = yield* Ref.modify(stateRef, (previous) =>
       Equal.equals(previous, next) ? [false, previous] : [true, next],
     );
     if (changed) yield* PubSub.publish(changes, next);
-  });
+  }, publishLock.withPermits(1));
 
   // Serialize refreshes so slow old-settings reads cannot resurrect removed sources after newer refreshes.
   const refreshLock = yield* Semaphore.make(1);
@@ -143,11 +187,48 @@ export const make = Effect.gen(function* () {
       input.creditId,
     );
     const snapshot = yield* readSource(input.sourceId, config);
-    const previous = yield* Ref.get(stateRef);
+    const previous = yield* Ref.get(readsRef);
     yield* publish(previous.map((source) => (source.id === input.sourceId ? snapshot : source)));
     return result;
   }, refreshLock.withPermits(1));
 
+  // Opted-in hubs only; a failed read keeps the routes already known for that hub.
+  const readRoutes = Effect.gen(function* () {
+    const settings = yield* settingsService.getSettings.pipe(
+      Effect.orElseSucceed((): ServerSettings | null => null),
+    );
+    const entries = Object.entries(settings?.usageLimitSources ?? {}).filter(
+      ([, config]) => config.enabled && config.threadAccounts,
+    );
+    const reads = yield* Effect.forEach(
+      entries,
+      ([id, config]) =>
+        readManagementKey(config).pipe(
+          Effect.flatMap((managementKey) =>
+            managementKey.length > 0
+              ? api.readSessionAccounts({ url: config.url, managementKey })
+              : Effect.succeed([]),
+          ),
+          Effect.orElseSucceed(() => []),
+          Effect.map((routes) => [id as UsageLimitSourceId, routes] as const),
+        ),
+      { concurrency: 4 },
+    );
+    yield* Ref.update(routesRef, (previous) => {
+      const next = new Map<UsageLimitSourceId, ReadonlyMap<string, string>>();
+      for (const [id, routes] of reads) {
+        const routed = new Map(previous.get(id));
+        for (const { sessionId, accountId } of routes) {
+          routed.delete(sessionId);
+          routed.set(sessionId, accountId);
+        }
+        next.set(id, new Map([...routed].slice(-MAX_ROUTES_PER_SOURCE)));
+      }
+      return next;
+    });
+    yield* publish(undefined);
+  }).pipe(Effect.ignoreCause({ log: true }));
+
   yield* settingsService.streamChanges.pipe(
     Stream.map((settings) => settings.usageLimitSources),
     Stream.changes,
@@ -175,6 +256,12 @@ export const make = Effect.gen(function* () {
     ),
   ).pipe(Effect.forkScoped);
 
+  // CLIProxyAPI keeps queued usage for 60 seconds by default. Not gated on client activity:
+  // a turn that ends while no client is open would otherwise lose its route for good.
+  yield* Effect.forever(Effect.sleep(ROUTE_READ_INTERVAL).pipe(Effect.andThen(readRoutes))).pipe(
+    Effect.forkScoped,
+  );
+
   yield* refresh.pipe(Effect.forkScoped);
 
   return {

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

@@ -60,6 +60,8 @@ function fixture(
     }>;
     upstream?: (request: RequestBody) => { status: number; body: unknown };
     cooldownStatus?: number;
+    usageQueue?: ReadonlyArray<unknown>;
+    authFilesStatus?: number;
   } = {},
 ) {
   const requests: Array<{ path: string; body?: RequestBody }> = [];
@@ -75,8 +77,13 @@ function fixture(
       if (path.endsWith("/auth-files"))
         return HttpClientResponse.fromWeb(
           request,
-          Response.json({ files: options.accounts ?? accounts }),
+          Response.json(
+            { files: options.accounts ?? accounts },
+            { status: options.authFilesStatus ?? 200 },
+          ),
         );
+      if (path.endsWith("/usage-queue"))
+        return HttpClientResponse.fromWeb(request, Response.json(options.usageQueue ?? []));
       if (path.endsWith("/reset-quota"))
         return HttpClientResponse.fromWeb(
           request,
@@ -437,6 +444,40 @@ describe("CLIProxyAPI built-in management API", () => {
     }),
   );
 
+  it.effect("leaves the usage queue unread when the hub's accounts cannot be listed", () =>
+    Effect.gen(function* () {
+      const test = fixture({
+        authFilesStatus: 503,
+        usageQueue: [{ session_id: "session-1", auth_index: "a" }],
+      });
+      const api = yield* test.api;
+      expect((yield* api.readSessionAccounts(config).pipe(Effect.result))._tag).toBe("Failure");
+      // Reading the queue pops it, so the records must still be there for the next read.
+      expect(test.requests.map((request) => request.path)).not.toContainEqual(
+        expect.stringMatching(/usage-queue$/),
+      );
+    }),
+  );
+
+  it.effect("names the account that served each queued session by its auth index", () =>
+    Effect.gen(function* () {
+      const test = fixture({
+        usageQueue: [
+          { session_id: "session-1", auth_index: "b" },
+          { session_id: "session-2", auth_index: "unknown" },
+          { auth_index: "a" },
+          "not a record",
+          { session_id: "session-3", auth_index: "a" },
+        ],
+      });
+      const api = yield* test.api;
+      expect(yield* api.readSessionAccounts(config)).toEqual([
+        { sessionId: "session-1", accountId: "second.json" },
+        { sessionId: "session-3", accountId: "first.json" },
+      ]);
+    }),
+  );
+
   const antigravitySummary = {
     buckets: [
       {