diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index 3fecc1c7d2d61..7f1006f1c5ec5 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -46,6 +46,9 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; 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 ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; import { makeProviderEventRoutingState, @@ -68,8 +71,16 @@ const layerGit = GitVcsDriver.layer.pipe( Layer.provideMerge(NodeServices.layer), ); -const makeLayerTest = (ingestorEventSink: Layer.Layer) => +const layerEmptyInstances = Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ + // @effect-diagnostics-next-line effectSucceedWithVoid:off - Optional instances require undefined rather than void. + getInstance: () => Effect.succeed(undefined), +}); +const makeLayerTest = ( + ingestorEventSink: Layer.Layer, + instances = layerEmptyInstances, +) => Layer.mergeAll( + instances, layerTestDatabase, layerTestStores, layerTestEventSink, @@ -79,6 +90,7 @@ const makeLayerTest = (ingestorEventSink: Layer.Layer + Effect.scoped( + Effect.gen(function* () { + const refreshed = yield* Deferred.make(); + const adapter = yield* ProviderAdapter.ProviderAdapterV2.pipe( + Effect.provide( + Layer.mock(ProviderAdapter.ProviderAdapterV2)({ + instanceId: ProviderInstanceId.make("cursor-work"), + driver: ProviderDriverKind.make("cursor"), + }), + ), + ); + const instance: ProviderInstance = { + instanceId: ProviderInstanceId.make("cursor-work"), + driverKind: ProviderDriverKind.make("cursor"), + continuationIdentity: { + driverKind: ProviderDriverKind.make("cursor"), + continuationKey: "work", + }, + enabled: true, + displayName: undefined, + orchestrationAdapter: adapter, + textGeneration: { + generateBranchName: () => Effect.die("unused"), + generateCommitMessage: () => Effect.die("unused"), + generatePrContent: () => Effect.die("unused"), + generateThreadTitle: () => Effect.die("unused"), + generateReviewGuide: () => Effect.die("unused"), + }, + snapshot: { + refreshUsage: Deferred.succeed(refreshed, undefined).pipe(Effect.asVoid), + getSnapshot: Effect.die("unused"), + refresh: Effect.die("unused"), + resolveMaintenance: () => Effect.die("unused"), + applyUsageLimits: () => Effect.die("unused"), + streamChanges: Stream.empty, + }, + }; + const lookups: ProviderInstanceId[] = []; + const instances = Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ + getInstance: (id) => + Effect.sync(() => { + lookups.push(id); + return id === instance.instanceId ? instance : undefined; + }), + }); + yield* Effect.gen(function* () { + const now = yield* DateTime.now; + const eventSink = yield* EventSink.EventSinkV2; + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const threadEvent = yield* threadCreatedEvent(now); + yield* eventSink.write({ events: [threadEvent] }); + const input = { + providerSessionId: yield* idAllocator.allocate.providerSession({ + providerInstanceId: instance.instanceId, + threadId: threadEvent.threadId, + }), + providerInstanceId: instance.instanceId, + threadId: threadEvent.threadId, + }; + for (const failureClass of ["validation_error", "usage_limit"] as const) { + const stored = yield* ingestor.ingestNormalized({ + ...input, + event: { + type: "turn.terminal", + driver: instance.driverKind, + providerThreadId: idAllocator.derive.providerThread({ + driver: instance.driverKind, + nativeThreadId: "limited", + }), + providerTurnId: idAllocator.derive.providerTurn({ + driver: instance.driverKind, + nativeTurnId: failureClass, + }), + runOrdinal: 1, + failureItemOrdinal: 1, + status: "failed", + failure: makeProviderFailure({ + class: failureClass, + message: "Provider rejected turn.", + }), + threadDisposition: "reusable", + }, + }); + assert.equal(stored.length, 1); + if (failureClass === "validation_error") assert.deepEqual(lookups, []); + } + assert.deepEqual(lookups, [instance.instanceId]); + yield* Deferred.await(refreshed); + }).pipe(Effect.provide(makeLayerTest(layerTestEventSink, instances))); + }), + ), +); + it.effect("records accepted billed turn usage once without billing the context window", () => { const recorded: Array>> = []; const layerAnalytics = Layer.succeed(ProviderEventIngestor.ProviderTurnAnalytics, { diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 6beb4d0e55471..06f444776cb3c 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -34,6 +34,7 @@ import { isTemporaryWorktreeBranch } from "@t3tools/shared/git"; import { getModelSelectionStringOptionValue, modelSelectionsEqual } from "@t3tools/shared/model"; import * as AnalyticsService from "../telemetry/AnalyticsService.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; +import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts"; import * as EventSink from "./EventSink.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; @@ -326,6 +327,7 @@ export const layer: Layer.Layer< | IdAllocator.IdAllocatorV2 | ProjectionStore.ProjectionStoreV2 | ThreadCommandExecutor.ThreadCommandExecutor + | ProviderInstanceRegistry.ProviderInstanceRegistry > = Layer.effect( ProviderEventIngestorV2, Effect.gen(function* () { @@ -336,6 +338,8 @@ export const layer: Layer.Layer< const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor; const analytics = yield* ProviderTurnAnalytics; const agentScope = yield* AgentScope; + const instances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry; + const scope = yield* Effect.scope; const completedTurnAnalytics = new Set(); const makeDomainEvent = @@ -794,6 +798,23 @@ export const layer: Layer.Layer< }, (effect, input) => effect.pipe( + Effect.tap((storedEvents) => + Effect.gen(function* () { + if ( + storedEvents.length === 0 || + input.event.type !== "turn.terminal" || + input.event.status !== "failed" || + input.event.failure.class !== "usage_limit" + ) + return; + const instance = yield* instances.getInstance(input.providerInstanceId); + if (!instance?.snapshot.refreshUsage) return; + yield* instance.snapshot.refreshUsage.pipe( + Effect.ignoreCause({ log: true }), + Effect.forkIn(scope, { startImmediately: true }), + ); + }), + ), Effect.flatMap((storedEvents) => Effect.gen(function* () { if (input.event.type !== "app_thread.created") return storedEvents; diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 9ae9ddce81a99..cac8e4fbbb8e0 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -47,6 +47,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; +import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts"; import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; @@ -533,6 +534,10 @@ function layerTest(input: { const layerProviderEventIngestorTest = ProviderEventIngestor.layer.pipe( Layer.provide( Layer.mergeAll( + Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ + // @effect-diagnostics-next-line effectSucceedWithVoid:off - Optional instances require undefined rather than void. + getInstance: () => Effect.succeed(undefined), + }), layerConfiguredEventSink, IdAllocator.layer, layerTestStores, diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 4da359f74c920..afd11a9ab7ecb 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -40,6 +40,7 @@ import * as ProviderAuthService from "../../provider/ProviderAuthService.ts"; import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests"; import * as ProviderContinuationService from "../ProviderContinuationService.ts"; import * as ProviderEventIngestor from "../ProviderEventIngestor.ts"; +import * as ProviderInstanceRegistry from "../../provider/ProviderInstanceRegistry.ts"; import * as ProviderRuntimeRecoveryService from "../ProviderRuntimeRecoveryService.ts"; import * as ProviderSessionManager from "../ProviderSessionManager.ts"; import * as CommandOutputRecorder from "./CommandOutputRecorder.ts"; @@ -320,6 +321,10 @@ export function layerWithRegistry( const layerProviderEventIngestorProvided = ProviderEventIngestor.layer.pipe( Layer.provide( Layer.mergeAll( + Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ + // @effect-diagnostics-next-line effectSucceedWithVoid:off - Optional instances require undefined rather than void. + getInstance: () => Effect.succeed(undefined), + }), layerStores, layerEventSinkProvided, IdAllocator.layer, diff --git a/apps/server/src/provider/Drivers/ClaudeDriver.ts b/apps/server/src/provider/Drivers/ClaudeDriver.ts index 86832d53d3119..e1fbcd30080fd 100644 --- a/apps/server/src/provider/Drivers/ClaudeDriver.ts +++ b/apps/server/src/provider/Drivers/ClaudeDriver.ts @@ -299,6 +299,7 @@ export const ClaudeDriver: ProviderDriver resolveMaintenance().pipe( Effect.flatMap((maintenanceCapabilities) => diff --git a/packages/provider-core/src/server/managedProvider.test.ts b/packages/provider-core/src/server/managedProvider.test.ts index 668502f93e432..7794107824c9b 100644 --- a/packages/provider-core/src/server/managedProvider.test.ts +++ b/packages/provider-core/src/server/managedProvider.test.ts @@ -6,6 +6,7 @@ import { type ServerProvider, } from "@t3tools/contracts"; import { createModelCapabilities } from "@t3tools/shared/model"; +import * as Cache from "effect/Cache"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -138,6 +139,116 @@ const enrichedSnapshotSecond: ServerProvider = { }; describe("makeManagedServerProvider", () => { + it.effect("invalidates cached usage and queues one refresh behind an in-flight refresh", () => + Effect.scoped( + Effect.gen(function* () { + const calls = yield* Ref.make(0); + const probeStarted = yield* Deferred.make(); + const releaseProbe = yield* Deferred.make(); + const cache = yield* Cache.makeWith( + () => + Effect.gen(function* () { + const count = yield* Ref.updateAndGet(calls, (count) => count + 1); + if (count === 2) { + yield* Deferred.succeed(probeStarted, undefined); + yield* Deferred.await(releaseProbe); + } + return { + ...refreshedSnapshot, + usageLimits: { + checkedAt: refreshedSnapshot.checkedAt, + windows: [ + { + id: "five_hour", + kind: "session", + label: "Session", + usedPercent: count === 1 ? 20 : 100, + }, + ], + }, + } satisfies ServerProvider; + }), + { capacity: 1, timeToLive: () => "5 minutes" }, + ); + const warmed = yield* Cache.get(cache, "claude"); + const provider = yield* makeManagedServerProvider({ + resolveMaintenance: () => Effect.die("usage refresh must not resolve maintenance"), + getSettings: Effect.succeed({ enabled: true }), + streamSettings: Stream.empty, + haveSettingsChanged: () => false, + initialSnapshot: () => Effect.succeed(warmed), + checkProvider: Cache.get(cache, "claude"), + invalidateUsageCache: Cache.invalidateAll(cache), + refreshOnInterval: false, + }); + yield* provider.streamChanges.pipe(Stream.take(1), Stream.runDrain); + assert.strictEqual(yield* Ref.get(calls), 1); + const refresh = yield* provider.refreshUsage!.pipe(Effect.forkChild); + yield* Deferred.await(probeStarted); + const queued = yield* provider.refreshUsage!.pipe( + Effect.forkChild({ startImmediately: true }), + ); + yield* provider.refreshUsage!; + assert.strictEqual(yield* Ref.get(calls), 2); + assert.strictEqual((yield* provider.getSnapshot).usageLimits?.windows[0]?.usedPercent, 20); + yield* Deferred.succeed(releaseProbe, undefined); + yield* Fiber.join(refresh); + yield* Fiber.join(queued); + assert.strictEqual(yield* Ref.get(calls), 3); + assert.strictEqual((yield* provider.getSnapshot).usageLimits?.windows[0]?.usedPercent, 100); + }), + ).pipe(Effect.provide(layerAlwaysRunTest)), + ); + it.effect("keeps the waiting usage refresh's claim when the running one is interrupted", () => + Effect.scoped( + Effect.gen(function* () { + const calls = yield* Ref.make(0); + const usageProbeStarted = yield* Deferred.make(); + const statusProbeStarted = yield* Deferred.make(); + const releaseStatusProbe = yield* Deferred.make(); + const provider = yield* makeManagedServerProvider({ + resolveMaintenance: () => Effect.succeed(maintenanceCapabilities), + getSettings: Effect.succeed({ enabled: true }), + streamSettings: Stream.empty, + haveSettingsChanged: () => false, + initialSnapshot: () => Effect.succeed(initialSnapshot), + checkProvider: Effect.gen(function* () { + const count = yield* Ref.updateAndGet(calls, (count) => count + 1); + if (count === 2) { + yield* Deferred.succeed(usageProbeStarted, undefined); + return yield* Effect.never; + } + if (count === 3) { + yield* Deferred.succeed(statusProbeStarted, undefined); + yield* Deferred.await(releaseStatusProbe); + } + return { + ...refreshedSnapshot, + usageLimits: { checkedAt: refreshedSnapshot.checkedAt, windows: [] }, + } satisfies ServerProvider; + }), + refreshOnInterval: false, + }); + yield* provider.streamChanges.pipe(Stream.take(1), Stream.runDrain); + const running = yield* provider.refreshUsage!.pipe(Effect.forkChild); + yield* Deferred.await(usageProbeStarted); + const status = yield* provider.refresh.pipe(Effect.forkChild({ startImmediately: true })); + const waiting = yield* provider.refreshUsage!.pipe( + Effect.forkChild({ startImmediately: true }), + ); + yield* Fiber.interrupt(running); + yield* Deferred.await(statusProbeStarted); + const extra = yield* provider.refreshUsage!.pipe( + Effect.forkChild({ startImmediately: true }), + ); + yield* Deferred.succeed(releaseStatusProbe, undefined); + yield* Fiber.join(status); + yield* Fiber.join(waiting); + yield* Fiber.join(extra); + assert.strictEqual(yield* Ref.get(calls), 4); + }), + ).pipe(Effect.provide(layerAlwaysRunTest)), + ); it.effect( "runs the initial provider check in the background and streams the refreshed snapshot", () => diff --git a/packages/provider-core/src/server/managedProvider.ts b/packages/provider-core/src/server/managedProvider.ts index f40baee525c8d..41ff65b0f294b 100644 --- a/packages/provider-core/src/server/managedProvider.ts +++ b/packages/provider-core/src/server/managedProvider.ts @@ -44,6 +44,7 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( readonly haveSettingsChanged: (previous: Settings, next: Settings) => boolean; readonly initialSnapshot: (settings: Settings) => Effect.Effect; readonly checkProvider: Effect.Effect; + readonly invalidateUsageCache?: Effect.Effect; readonly enrichSnapshot?: (input: { readonly settings: Settings; readonly snapshot: ServerProvider; @@ -60,6 +61,7 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( > { const host = yield* ProviderHost.ProviderHost; const refreshSemaphore = yield* Semaphore.make(1); + const usageRefreshQueued = yield* Ref.make(null); const changesPubSub = yield* Effect.acquireRelease( PubSub.unbounded(), PubSub.shutdown, @@ -216,6 +218,38 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( return yield* applySnapshot(nextSettings, { forceRefresh: true }); }); + const refreshUsage = Effect.fn("ManagedServerProvider.refreshUsage")(function* () { + const { snapshot } = yield* Ref.get(snapshotStateRef); + if ( + !snapshot.enabled || + !snapshot.usageLimits || + snapshot.usageLimits.unavailable?.reason === "unsupported" + ) + return; + // An in-flight refresh may have read cached usage, so one refresh waits behind it. + const claim = Symbol(); + const release = Ref.update(usageRefreshQueued, (queued) => (queued === claim ? null : queued)); + yield* Effect.uninterruptibleMask((restore) => + Ref.modify(usageRefreshQueued, (queued) => [queued !== null, queued ?? claim]).pipe( + Effect.flatMap((held) => + held + ? Effect.void + : restore( + refreshSemaphore.withPermits(1)( + release.pipe( + Effect.andThen(input.invalidateUsageCache ?? Effect.void), + Effect.andThen(input.getSettings), + Effect.flatMap((settings) => + applySnapshotBase(settings, { forceRefresh: true }), + ), + ), + ), + ).pipe(Effect.onInterrupt(() => release)), + ), + ), + ); + }); + const shouldRefreshOnInterval = Effect.gen(function* () { const state = yield* Ref.get(snapshotStateRef); // A disabled instance has nothing to probe; a periodic check would only @@ -304,6 +338,7 @@ export const makeManagedServerProvider = Effect.fn("makeManagedServerProvider")( .resolveMaintenance({ fresh: true }) .pipe(Effect.andThen(refreshSnapshot), Effect.tapError(Effect.logError), Effect.orDie), applyUsageLimits, + refreshUsage: refreshUsage(), get streamChanges() { return Stream.fromPubSub(changesPubSub); }, diff --git a/packages/provider-core/src/server/snapshot.ts b/packages/provider-core/src/server/snapshot.ts index c9c877d29fe20..88f2c1f113848 100644 --- a/packages/provider-core/src/server/snapshot.ts +++ b/packages/provider-core/src/server/snapshot.ts @@ -1,4 +1,8 @@ -import type { ProviderUsageLimitsUpdate, ServerProvider } from "@t3tools/contracts"; +import type { + ProviderUsageLimitsUpdate, + ServerProvider, + ServerSettingsError, +} from "@t3tools/contracts"; import type * as Effect from "effect/Effect"; import type * as Stream from "effect/Stream"; import type { ProviderMaintenanceCapabilities } from "./maintenance.ts"; @@ -10,6 +14,7 @@ export interface ManagedServerProvider { }) => Effect.Effect; readonly getSnapshot: Effect.Effect; readonly refresh: Effect.Effect; + readonly refreshUsage?: Effect.Effect; readonly streamChanges: Stream.Stream; /** Merge windows by ID; updates with no usable window leave the snapshot untouched. */ readonly applyUsageLimits: ( diff --git a/packages/provider-cursor/src/server/CursorAgentSdk.ts b/packages/provider-cursor/src/server/CursorAgentSdk.ts index 7f0decb49af3f..5986d09506d09 100644 --- a/packages/provider-cursor/src/server/CursorAgentSdk.ts +++ b/packages/provider-cursor/src/server/CursorAgentSdk.ts @@ -43,6 +43,21 @@ export function cursorSdkFailureMessage(error: CursorAgentSdkRunnerError): strin : `Cursor Agent SDK ${error.method} failed: ${reason}`; } +// The SDK maps backend details to codes like PRO_USER_USAGE_LIMIT or RATE_LIMITED, without a status. +const CURSOR_USAGE_LIMIT_CODE = + /^(?:resource_exhausted|rate_limited(?:_changeable)?|\w*_usage_limit|\w*_rate_limit(?:_exceeded)?)$/i; + +export function isCursorUsageLimitError(cause: unknown): boolean { + const error = isCursorAgentSdkRunnerError(cause) ? cause.cause : cause; + if (typeof error !== "object" || error === null) return false; + const code = Reflect.get(error, "code"); + return ( + Reflect.get(error, "name") === "RateLimitError" || + Reflect.get(error, "status") === 429 || + (typeof code === "string" && CURSOR_USAGE_LIMIT_CODE.test(code)) + ); +} + const GENERIC_SDK_ERROR_CODES = new Set(["", "error", "unknown"]); function sdkErrorReason(cause: unknown): string | undefined { diff --git a/packages/provider-cursor/src/server/adapter.test.ts b/packages/provider-cursor/src/server/adapter.test.ts index d91d3bc339933..062089fffa0c3 100644 --- a/packages/provider-cursor/src/server/adapter.test.ts +++ b/packages/provider-cursor/src/server/adapter.test.ts @@ -214,6 +214,15 @@ describe("CursorAdapterV2", () => { ); it.effect.each([ + { + reason: "the SDK's detailed plan usage limit code", + wait: Effect.succeed({ + id: "native-cursor-run", + status: "error" as const, + error: { message: "You've hit your usage limit", code: "PRO_USER_USAGE_LIMIT" }, + }), + failure: { class: "usage_limit", message: "You've hit your usage limit" }, + }, { reason: "the error Cursor's run ended with", wait: Effect.succeed({ @@ -232,7 +241,7 @@ describe("CursorAdapterV2", () => { }, }, { - reason: "the SDK error that broke the run", + reason: "the SDK rate limit that broke the run", wait: Effect.fail( new CursorAgentSdk.CursorAgentSdkRunnerError({ method: "run.wait", @@ -243,7 +252,7 @@ describe("CursorAdapterV2", () => { }), ), failure: { - class: "transport_error", + class: "usage_limit", message: "Cursor Agent SDK run.wait failed: Too many requests (resource_exhausted, HTTP 429)", }, diff --git a/packages/provider-cursor/src/server/adapter.ts b/packages/provider-cursor/src/server/adapter.ts index 93652e492e031..72aca919b25f4 100644 --- a/packages/provider-cursor/src/server/adapter.ts +++ b/packages/provider-cursor/src/server/adapter.ts @@ -2506,7 +2506,9 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( cause: result.error, // makeProviderFailure never reads a cause's own text. message: result.error?.message, - class: "provider_error", + class: CursorAgentSdk.isCursorUsageLimitError(result.error) + ? "usage_limit" + : "provider_error", }) : makeProviderFailure({ cause: transportFailure, @@ -2532,7 +2534,9 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( message: CursorAgentSdk.isCursorAgentSdkRunnerError(cause) ? CursorAgentSdk.cursorSdkFailureMessage(cause) : undefined, - class: "transport_error", + class: CursorAgentSdk.isCursorUsageLimitError(cause) + ? "usage_limit" + : "transport_error", }), }), }); diff --git a/packages/provider-opencode/src/server/adapter.test.ts b/packages/provider-opencode/src/server/adapter.test.ts index 759545fff0d19..41da6bd0ff435 100644 --- a/packages/provider-opencode/src/server/adapter.test.ts +++ b/packages/provider-opencode/src/server/adapter.test.ts @@ -240,6 +240,51 @@ const makeOpenCodeRuntimeHarness = Effect.fn("makeOpenCodeRuntimeHarness")(funct }); describe("OpenCodeAdapterV2", () => { + it.effect("classifies APIError 429 as a usage limit", () => + Effect.gen(function* () { + const nativeEvents = asyncEventStream(); + const harness = yield* makeOpenCodeRuntimeHarness("usage-limit", "root", { + event: { subscribe: async () => ({ stream: nativeEvents.stream }) }, + session: { + create: async () => ({ data: { id: "root", time: { created: 1, updated: 1 } } }), + promptAsync: async () => ({ data: true }), + abort: async () => ({ data: true }), + children: async () => ({ data: [] }), + }, + }); + yield* harness.startTurn(); + const received = yield* harness.runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + Effect.forkScoped, + ); + yield* Effect.promise(() => + nativeEvents.push({ + type: "session.error", + properties: { + sessionID: "root", + error: { + name: "APIError", + data: { + message: "Provider rejected request", + statusCode: 429, + isRetryable: true, + }, + }, + }, + }), + ); + const terminal = (yield* Fiber.join(received)).find( + (event) => event.type === "turn.terminal", + ); + assert.equal(terminal?.status, "failed"); + assert.equal(terminal?.failure?.class, "usage_limit"); + }).pipe( + Effect.provide(Layer.merge(IdAllocator.layer, McpProviderSessions.layer)), + Effect.scoped, + ), + ); + it.effect("applies a changed runtime policy to the session on resume and turn start", () => Effect.gen(function* () { const updates: Array<{ sessionID: string; permission: unknown }> = []; diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index ef85a7e70b29b..2a851188d076e 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -2643,6 +2643,8 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function (state): state is OpenCodeThreadState => state !== undefined, ); const message = openCodeErrorMessage(event); + const error = event.properties.error; + const usageLimit = error?.name === "APIError" && error.data.statusCode === 429; if ( !isMessageAbortedError(event) && (event.properties.sessionID === undefined || @@ -2659,8 +2661,8 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function { failure: makeProviderFailure({ message, - code: event.properties.error?.name ?? null, - class: "provider_error", + code: error?.name ?? null, + class: usageLimit ? "usage_limit" : "provider_error", }), threadDisposition: event.properties.sessionID === undefined ? "broken" : "reusable",