From e846c212352e5436b715bb31e47bccd894f04067 Mon Sep 17 00:00:00 2001 From: Alex Flekkas Date: Wed, 26 Aug 2026 15:01:13 +0300 Subject: [PATCH] fix(usage): resolve account limit review findings --- .../Layers/ProviderRuntimeIngestion.ts | 4 +- .../src/provider/Layers/ClaudeAdapter.test.ts | 32 ++++++++++- .../src/provider/Layers/ClaudeAdapter.ts | 27 ++++++--- .../src/usage/AccountLimitsService.test.ts | 57 +++++++++++++++++++ apps/server/src/usage/AccountLimitsService.ts | 13 ++--- .../src/usage/accountLimitsNormalize.test.ts | 9 ++- .../src/usage/accountLimitsTranscripts.ts | 5 +- 7 files changed, 121 insertions(+), 26 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 7a757876a92b..9a8159d3a10c 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -43,7 +43,7 @@ import { } from "../Services/ProviderRuntimeIngestion.ts"; import { forkParked } from "../../serverActivation.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; -import { AccountLimitsService } from "../../usage/AccountLimitsService.ts"; +import * as AccountLimitsService from "../../usage/AccountLimitsService.ts"; const providerTurnKey = (threadId: ThreadId, turnId: TurnId) => `${threadId}:${turnId}`; const providerTaskKey = (threadId: ThreadId, taskId: string) => `${threadId}:${taskId}`; @@ -876,7 +876,7 @@ const make = Effect.gen(function* () { const providerService = yield* ProviderService; const projectionTurnRepository = yield* ProjectionTurnRepository; const serverSettingsService = yield* ServerSettingsService; - const accountLimits = yield* AccountLimitsService; + const accountLimits = yield* AccountLimitsService.AccountLimitsService; const providerCommandId = (event: ProviderRuntimeEvent, tag: string) => crypto.randomUUIDv4.pipe( Effect.map((uuid) => CommandId.make(`provider:${event.eventId}:${tag}:${uuid}`)), diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts index 711b0f6f6aa3..a16203f3dbaf 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.test.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.test.ts @@ -8,6 +8,7 @@ import type { Options as ClaudeQueryOptions, PermissionMode, PermissionResult, + SDKControlGetUsageResponse, SDKMessage, SDKUserMessage, } from "@anthropic-ai/claude-agent-sdk"; @@ -37,7 +38,11 @@ import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; import { ProviderAdapterProcessError, ProviderAdapterValidationError } from "../Errors.ts"; import type { ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; -import { makeClaudeAdapter, type ClaudeAdapterLiveOptions } from "./ClaudeAdapter.ts"; +import { + makeClaudeAdapter, + readClaudeAccountUsage, + type ClaudeAdapterLiveOptions, +} from "./ClaudeAdapter.ts"; const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings); // Test-local service tag so the rest of the file can keep using `yield* ClaudeAdapter`. @@ -273,6 +278,31 @@ const THREAD_ID = ThreadId.make("thread-claude-1"); const RESUME_THREAD_ID = ThreadId.make("thread-claude-resume"); describe("ClaudeAdapterLive", () => { + it.effect("abandons a hung account-usage request after three seconds", () => + Effect.gen(function* () { + let resolveUsage: (response: SDKControlGetUsageResponse) => void = () => {}; + const usageResponse = new Promise((resolve) => { + resolveUsage = resolve; + }); + let signalUsageRequested: () => void = () => {}; + const usageRequested = new Promise((resolve) => { + signalUsageRequested = resolve; + }); + + const usageFiber = yield* readClaudeAccountUsage({ + usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET: () => { + signalUsageRequested(); + return usageResponse; + }, + }).pipe(Effect.forkChild); + yield* Effect.promise(() => usageRequested); + yield* TestClock.adjust("3 seconds"); + + assert.equal(yield* Fiber.join(usageFiber), undefined); + resolveUsage({} as SDKControlGetUsageResponse); + }), + ); + it.effect("returns validation error for non-claude provider on startSession", () => { const harness = makeHarness(); return Effect.gen(function* () { diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index bb3e0cfa0e71..b82ce10da2dc 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -273,6 +273,21 @@ interface ClaudeQueryRuntime extends AsyncIterable { readonly close: () => void; } +/** Reads Claude's experimental account-usage snapshot without allowing a hung control call to leak a fiber. */ +export function readClaudeAccountUsage( + runtime: Pick, +) { + return Effect.promise(async () => { + try { + // Called through the query object so the SDK method keeps its receiver; + // an extracted reference loses `this` and throws. + return await runtime.usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET?.(); + } catch { + return undefined; + } + }).pipe(Effect.timeoutOption("3 seconds"), Effect.map(Option.getOrUndefined)); +} + export interface ClaudeAdapterLiveOptions { readonly instanceId?: ProviderInstanceId; readonly environment?: NodeJS.ProcessEnv; @@ -3430,6 +3445,8 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( * `account.rate-limits.updated`. The streamed `rate_limit_event` only ever * names the single window currently binding, and Claude limits never reach * disk, so this pull is the only source that shows every window at once. + * The throttle is shared by sessions owned by this adapter; concurrent + * initialization can race into one extra request, which is harmless. */ const emitAccountUsageSnapshot = Effect.fn("emitAccountUsageSnapshot")(function* ( context: ClaudeSessionContext, @@ -3442,15 +3459,7 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( if (elapsed >= 0 && elapsed < ACCOUNT_USAGE_MIN_INTERVAL_MS) return; lastAccountUsageFetchAtMs = now; - const usage = yield* Effect.promise(async () => { - try { - // Called through the query object so the SDK method keeps its - // receiver; an extracted reference loses `this` and throws. - return await context.query.usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET?.(); - } catch { - return undefined; - } - }); + const usage = yield* readClaudeAccountUsage(context.query); if (!usage || usage.rate_limits === null || usage.rate_limits === undefined) return; const stamp = yield* makeEventStamp(); diff --git a/apps/server/src/usage/AccountLimitsService.test.ts b/apps/server/src/usage/AccountLimitsService.test.ts index db6c38fc9367..7ed773f7bece 100644 --- a/apps/server/src/usage/AccountLimitsService.test.ts +++ b/apps/server/src/usage/AccountLimitsService.test.ts @@ -12,6 +12,7 @@ import * as Effect from "effect/Effect"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as TestClock from "effect/testing/TestClock"; import * as ServerConfig from "../config.ts"; import * as ServerSettingsModule from "../serverSettings.ts"; @@ -311,6 +312,62 @@ it.layer(NodeServices.layer)("account limits service", (it) => { ), ), ); + + it.effect("retries transcript seeding immediately after the wall clock moves backward", () => + Effect.gen(function* () { + const codexHome = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-seed-clock-step-")); + const sessionsDir = NodePath.join(codexHome, "sessions"); + NodeFS.mkdirSync(sessionsDir, { recursive: true }); + try { + yield* Effect.gen(function* () { + const beforeStepMs = 1_800_000_000_000; + yield* TestClock.setTime(beforeStepMs); + const service = yield* AccountLimitsServiceModule.AccountLimitsService; + + // The empty scan records the throttle floor. + expect((yield* service.readSummary()).snapshots).toEqual([]); + + const afterStepMs = beforeStepMs - 1_000; + const transcriptPath = NodePath.join(sessionsDir, "rollout-1.jsonl"); + NodeFS.writeFileSync( + transcriptPath, + // @effect-diagnostics-next-line preferSchemaOverJson:off - fabricates one raw transcript line. + `${JSON.stringify({ + timestamp: DateTime.formatIso(DateTime.makeUnsafe(afterStepMs)), + payload: { rate_limits: codexPayload(64) }, + })}\n`, + ); + NodeFS.utimesSync(transcriptPath, afterStepMs / 1_000, afterStepMs / 1_000); + yield* TestClock.setTime(afterStepMs); + + const summary = yield* service.readSummary(); + expect( + summary.snapshots.map((snapshot) => [ + snapshot.instanceId, + snapshot.windows[0]?.usedPercent, + ]), + ).toEqual([["codex_clock", 64]]); + }).pipe( + Effect.provide( + makeLayer({ + providerInstances: { + [asInstanceId("codex")]: { + driver: asDriver("codex"), + config: { homePath: "/nonexistent/t3-test-codex-default" }, + }, + [asInstanceId("codex_clock")]: { + driver: asDriver("codex"), + config: { homePath: codexHome }, + }, + }, + }), + ), + ); + } finally { + NodeFS.rmSync(codexHome, { recursive: true, force: true }); + } + }), + ); }); // Plain `it`: the seed consults the real clock for its retry floor and diff --git a/apps/server/src/usage/AccountLimitsService.ts b/apps/server/src/usage/AccountLimitsService.ts index 333667a6f6bc..49af0dcf1d79 100644 --- a/apps/server/src/usage/AccountLimitsService.ts +++ b/apps/server/src/usage/AccountLimitsService.ts @@ -65,12 +65,8 @@ const CODEX_SEED_MIN_INTERVAL_MS = 60_000; /** On-disk shape of the snapshot cache: the contract array, JSON-encoded. */ const LimitsCacheFile = Schema.Array(AccountLimitsSnapshot); -const decodeLimitsCache = Schema.decodeUnknownEffect( - Schema.fromJsonString(LimitsCacheFile as unknown as Schema.Codec), -); -const encodeLimitsCache = Schema.encodeEffect( - Schema.fromJsonString(LimitsCacheFile as unknown as Schema.Codec), -); +const decodeLimitsCache = Schema.decodeUnknownEffect(Schema.fromJsonString(LimitsCacheFile)); +const encodeLimitsCache = Schema.encodeEffect(Schema.fromJsonString(LimitsCacheFile)); const decodeCodexSettings = Schema.decodeUnknownEffect(CodexSettings); export interface AccountLimitsIngestInput { @@ -378,7 +374,10 @@ export const make = Effect.gen(function* () { configMap: ProviderInstanceConfigMap | null, ) { if (configMap === null) return; - if (nowMs - lastCodexSeedAttemptAtMs < CODEX_SEED_MIN_INTERVAL_MS) return; + const elapsedMs = nowMs - lastCodexSeedAttemptAtMs; + // A backward host clock must not park transcript recovery until the old + // timestamp comes around again. One extra scan after an NTP step is safe. + if (elapsedMs >= 0 && elapsedMs < CODEX_SEED_MIN_INTERVAL_MS) return; lastCodexSeedAttemptAtMs = nowMs; const targets: CodexSeedTarget[] = []; diff --git a/apps/server/src/usage/accountLimitsNormalize.test.ts b/apps/server/src/usage/accountLimitsNormalize.test.ts index 9b41b2b891d6..d0afe47cb913 100644 --- a/apps/server/src/usage/accountLimitsNormalize.test.ts +++ b/apps/server/src/usage/accountLimitsNormalize.test.ts @@ -1,5 +1,4 @@ import { describe, expect, it } from "@effect/vitest"; -import * as DateTime from "effect/DateTime"; import { claudeUsageSnapshotFromUnknown, @@ -32,11 +31,11 @@ describe("claudeUsageSnapshotFromUnknown", () => { expect(snapshot?.windows[2]).toMatchObject({ label: "Fable", usedPercent: 30 }); }); - it("reads the newer limits array, including a Fable-scoped weekly", () => { + it("prefers the newer limits array, including a Fable-scoped weekly", () => { const snapshot = claudeUsageSnapshotFromUnknown({ subscription_type: "max", rate_limits: { - five_hour: null, + five_hour: { utilization: 88, resets_at: "2026-08-09T00:00:00.000Z" }, limits: [ { kind: "session", percent: 10, resets_at: "2026-08-08T23:00:00.000Z" }, { kind: "weekly_all", percent: 20, resets_at: "2026-08-11T17:00:00.000Z" }, @@ -114,7 +113,7 @@ describe("claudeWindowFromRateLimitEvent", () => { id: "five_hour", label: "5h", usedPercent: 87.5, - resetsAt: DateTime.formatIso(DateTime.makeUnsafe(1_786_600_800_000)), + resetsAt: "2026-08-13T06:00:00.000Z", windowMinutes: 300, }); }); @@ -146,7 +145,7 @@ describe("codexSnapshotFromUnknown", () => { id: "seven_day", label: "Week", usedPercent: 14, - resetsAt: DateTime.formatIso(DateTime.makeUnsafe(1_786_677_720_000)), + resetsAt: "2026-08-14T03:22:00.000Z", windowMinutes: 10080, }, ]); diff --git a/apps/server/src/usage/accountLimitsTranscripts.ts b/apps/server/src/usage/accountLimitsTranscripts.ts index 59024cedbe08..90c525374f70 100644 --- a/apps/server/src/usage/accountLimitsTranscripts.ts +++ b/apps/server/src/usage/accountLimitsTranscripts.ts @@ -82,11 +82,12 @@ async function readTailRateLimits( const length = stat.size - start; if (length <= 0) return null; const buffer = Buffer.alloc(length); - await handle.read(buffer, 0, length, start); + const { bytesRead } = await handle.read(buffer, 0, length, start); + if (bytesRead <= 0) return null; // The first line may be cut mid-record by the tail offset; JSON.parse // rejects it and the scan moves on. - const lines = buffer.toString("utf8").split("\n"); + const lines = buffer.subarray(0, bytesRead).toString("utf8").split("\n"); for (let index = lines.length - 1; index >= 0; index -= 1) { const line = lines[index]; if (!line || !line.includes('"rate_limits"')) continue;