diff --git a/apps/mobile/src/features/settings/SettingsThreadsRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsThreadsRouteScreen.tsx index 22eef1d18ac76..c802d77550b02 100644 --- a/apps/mobile/src/features/settings/SettingsThreadsRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsThreadsRouteScreen.tsx @@ -181,7 +181,7 @@ function AutoSettleSettingsRows() { writeToAll({ stopAtSubscriptionLimit: value })} diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index d2d2e7a15053f..83e48bb489c42 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -179,6 +179,7 @@ import { makeProviderFailure, makeProviderRetryTurnItem, } from "@t3tools/provider-core/server/failure"; +import * as SubscriptionLimitPolicy from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; import { turnScopedSelectionTransition } from "@t3tools/provider-core/server/selectionTransition"; import { providerMessageTextWithAttachmentPaths } from "@t3tools/provider-core/server/attachmentPrompt"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; @@ -1727,6 +1728,9 @@ const getNativeThreadId = Effect.fnUntraced(function* ( return nativeThreadId; }); +// Extra usage starts once the session or weekly allowance runs out. +const CLAUDE_SUBSCRIPTION_WINDOW_IDS = ["five_hour", "seven_day"]; + const isSyntheticClaudeTurnId = (nativeTurnId: string): boolean => nativeTurnId.startsWith("turn:"); const isTerminalProviderTurn = (turn: OrchestrationV2ProviderTurn): boolean => @@ -4193,41 +4197,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); const subscriptionLimitFailure = Effect.fn("ClaudeAdapterV2.checkSubscriptionLimit")( function* () { - const limits = yield* adapterOptions.checkUsageLimits ?? Effect.succeed(undefined); - const nowMs = DateTime.toEpochMillis(yield* DateTime.now); - const subscriptionWindows = - limits?.windows.filter( - (window) => window.id === "five_hour" || window.id === "seven_day", - ) ?? []; - const exhausted = subscriptionWindows.filter( - (window) => - window.usedPercent >= 100 && - (window.resetsAt === undefined || Date.parse(window.resetsAt) > nowMs), - ); - if ( - limits !== undefined && - limits.unavailable === undefined && - subscriptionWindows.length > 0 && - exhausted.length === 0 - ) - return null; - const resetTimes = exhausted.map((window) => window.resetsAt); - const resetAt = - resetTimes.length > 0 && resetTimes.every((time) => time !== undefined) - ? resetTimes.reduce((latest, time) => (time! > latest ? time! : latest), "") - : null; - return { - ...makeProviderFailure({ - class: exhausted.length > 0 ? "usage_limit" : "provider_error", - retryable: false, - code: "subscription_limit_guard", - message: - exhausted.length > 0 - ? "Claude subscription limit reached. This thread stopped because Stop at subscription limit is enabled." - : "Claude subscription usage could not be checked. Retry the turn, or disable Stop at subscription limit in application and thread settings to continue without this check.", - }), - resetAt, - }; + return SubscriptionLimitPolicy.subscriptionLimitFailure({ + limits: yield* adapterOptions.checkUsageLimits ?? Effect.succeed(undefined), + nowMs: DateTime.toEpochMillis(yield* DateTime.now), + provider: "Claude", + windowIds: CLAUDE_SUBSCRIPTION_WINDOW_IDS, + }); }, ); diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index b22adf5c40d5a..61550906e5927 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -61,6 +61,7 @@ import * as ServerConfig from "../../config.ts"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as ProviderEventLoggers from "@t3tools/provider-core/server/ProviderEventLoggers"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; +import * as SubscriptionLimitPolicy from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; import * as EffectWorker from "../EffectWorker.ts"; import * as Orchestrator from "../Orchestrator.ts"; import * as ProviderReplayHarness from "../testkit/ProviderReplayHarness.ts"; @@ -2020,6 +2021,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { Effect.void, onRequest: (method: string, params: unknown) => Effect.Effect = () => Effect.void, readChildMetadata?: Parameters[2], + subscriptionLimitPolicy?: SubscriptionLimitPolicy.SubscriptionLimitPolicy, ) => Effect.gen(function* () { const fileSystem = yield* FileSystem.FileSystem; @@ -2073,6 +2075,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { continuationRequests.push(request); }), }, + ...(subscriptionLimitPolicy === undefined ? {} : { subscriptionLimitPolicy }), }); const threadId = ThreadId.make(`thread-${transcript.scenario}`); const runtime = yield* adapter.openSession({ @@ -7336,6 +7339,152 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); + const guardedPolicy: SubscriptionLimitPolicy.SubscriptionLimitPolicy = { + enabled: () => Effect.succeed(true), + enabledChanges: () => Stream.never, + }; + const rateLimitsRead = (id: number, usedPercent: number) => + [ + { + type: "expect_outbound", + label: "account/rateLimits/read", + frame: { id, method: "account/rateLimits/read", params: null }, + }, + { + type: "emit_inbound", + label: "account/rateLimits/read", + frame: { + id, + result: { + rateLimits: { + limitId: "codex", + primary: { usedPercent, resetsAt: 2000100000, windowDurationMins: 300 }, + }, + }, + }, + }, + ] satisfies Array; + + it.effect("refuses a guarded turn before Codex starts it once the allowance is used up", () => + Effect.gen(function* () { + const preamble = codexReplayPreamble({ + nativeThreadId: "native-guard-refused", + nativeTurnId: "turn-guard-refused", + prompt: "Continue.", + }); + const transcript = makeCodexReplayTranscript({ + scenario: "codex-subscription-guard-refused", + entries: [...preamble.slice(0, 5), ...rateLimitsRead(3, 100)], + }); + const harness = yield* makeCodexReplayHarness( + transcript, + undefined, + undefined, + undefined, + guardedPolicy, + ); + const error = yield* harness.runtime + .startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + text: "Continue.", + attemptId: RunAttemptId.make("attempt-guard-refused"), + }), + ) + .pipe(Effect.flip); + const failure = SubscriptionLimitPolicy.findSubscriptionLimitStop(error); + assert.equal(failure?.class, "usage_limit"); + assert.equal(failure?.code, "subscription_limit_guard"); + assert.equal(failure?.resetAt, "2033-05-19T07:20:00.000Z"); + assert.lengthOf(harness.terminalEvents(), 0); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + + it.effect("stops a running guarded turn when Codex reports the allowance used up", () => + Effect.gen(function* () { + const nativeThreadId = "native-guard-live"; + const nativeTurnId = "turn-guard-live"; + const preamble = codexReplayPreamble({ nativeThreadId, nativeTurnId, prompt: "Continue." }); + const transcript = makeCodexReplayTranscript({ + scenario: "codex-subscription-guard-live", + entries: [ + ...preamble.slice(0, 5), + ...rateLimitsRead(3, 40), + ...preamble.slice(5).map((entry) => withReplayRequestId(entry, 4)), + { + type: "emit_inbound", + label: "account/rateLimits/updated", + frame: { + method: "account/rateLimits/updated", + params: { + rateLimits: { + limitId: "codex", + primary: { usedPercent: 100, resetsAt: 2000100000, windowDurationMins: 300 }, + }, + }, + }, + }, + { + type: "expect_outbound", + label: "turn/interrupt", + frame: { + id: 5, + method: "turn/interrupt", + params: { threadId: nativeThreadId, turnId: nativeTurnId }, + }, + }, + { type: "emit_inbound", label: "turn/interrupt", frame: { id: 5, result: {} } }, + { + type: "emit_inbound", + label: "turn/completed", + frame: { + method: "turn/completed", + params: { + threadId: nativeThreadId, + turn: makeCodexReplayTurn({ id: nativeTurnId, status: "interrupted" }), + }, + }, + }, + ], + }); + const harness = yield* makeCodexReplayHarness( + transcript, + undefined, + undefined, + undefined, + guardedPolicy, + ); + yield* harness.runtime.startTurn( + makeCodexTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + text: "Continue.", + attemptId: RunAttemptId.make("attempt-guard-live"), + }), + ); + yield* harness.firstTerminal; + const terminal = harness.terminalEvents()[0]; + assert.equal(terminal?.status, "failed"); + if (terminal?.status !== "failed") return; + assert.equal(terminal.failure.class, "usage_limit"); + assert.equal(terminal.failure.code, "subscription_limit_guard"); + assert.equal(terminal.failure.resetAt, "2033-05-19T07:20:00.000Z"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + it.effect.each([ { notificationStatus: 500, diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index 07ad030180d8a..4e9327339ff6a 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -20,6 +20,7 @@ import { codexDefaultReasoningEffortOverride, } from "../../provider/CodexProvider.ts"; import { + codexRateLimitsToLimits, codexRateLimitsToUpdate, mergeCodexRateLimits, codexUsageLimitResetAt, @@ -71,6 +72,7 @@ import type { ProviderInstanceId, RuntimeMode, RuntimeRequestId, + ServerProviderUsageLimits, ThreadId, } from "@t3tools/contracts"; import { createPatch } from "diff"; @@ -140,6 +142,7 @@ import { makeProviderFailureTurnItem, makeProviderRetryTurnItem, } from "@t3tools/provider-core/server/failure"; +import * as SubscriptionLimitPolicy from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; import { turnScopedSelectionTransition } from "@t3tools/provider-core/server/selectionTransition"; import { isProviderNativeImageAttachment, @@ -1245,6 +1248,8 @@ function codexUsageLimitError( interface ActiveCodexTurnContext { closedCommandItemIds?: ReadonlySet; + /** Set when the subscription-limit guard interrupts this turn, which then fails with it. */ + subscriptionLimitStop?: OrchestrationV2ProviderFailure; latestProviderFailure?: { readonly nativeMessage: string; readonly failure: OrchestrationV2ProviderFailure; @@ -1707,7 +1712,10 @@ export type CodexAdapterV2DriverEnv = export const createCodexAdapterV2 = Effect.fn("CodexAdapterV2.createCodexAdapterV2")(function* ( { instanceId, environment, enabled, config }: ProviderAdapterDriverCreateInput, - hooks: Pick = {}, + hooks: Pick< + CodexAdapterV2Options, + "onUsageLimits" | "resolveRuntime" | "subscriptionLimitPolicy" + > = {}, ) { const clientFactory = yield* CodexAppServerClientFactory; const continuationRequests = yield* ProviderContinuationRequests.ProviderContinuationRequests; @@ -1798,6 +1806,7 @@ export interface CodexAdapterV2Options { readonly environment: NodeJS.ProcessEnv; readonly clientFactory: CodexAppServerClientFactoryShape; readonly onUsageLimits?: ManagedServerProvider["applyUsageLimits"]; + readonly subscriptionLimitPolicy?: SubscriptionLimitPolicy.SubscriptionLimitPolicy; readonly resolveRuntime?: Effect.Effect; readonly crypto: Crypto.Crypto; readonly fileSystem: FileSystem.FileSystem; @@ -1933,6 +1942,8 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( }); const events = yield* Queue.unbounded(); const rateLimitSnapshot = yield* Ref.make(undefined); + // Only the latest update matters: each check reads the merged snapshot. + const rateLimitUpdates = yield* Queue.sliding(1); const limitedTurnItems = yield* Ref.make( new Map>(), ); @@ -4523,6 +4534,7 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( const checkedAt = DateTime.formatIso(yield* DateTime.now); yield* adapterOptions.onUsageLimits({ ...update, checkedAt }); } + yield* Queue.offer(rateLimitUpdates, undefined); }).pipe(Effect.orDie), ); @@ -5765,6 +5777,25 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( readonly providerRetry?: ActiveCodexProviderRetry; }): Effect.fn.Return { const terminalStatus = providerTurnStatusToTerminal(input.status); + if ( + terminalStatus === "interrupted" && + input.context.subscriptionLimitStop !== undefined + ) { + return { + type: "turn.terminal", + driver: CODEX_PROVIDER, + providerThreadId: input.context.providerThread.id, + providerTurnId: input.context.providerTurnId, + runOrdinal: input.context.input.runOrdinal, + failureItemOrdinal: yield* resolveItemOrdinal( + input.context, + `terminal-failure:${input.context.providerTurnId}`, + ), + status: "failed", + failure: input.context.subscriptionLimitStop, + threadDisposition: "reusable", + }; + } if (terminalStatus === "failed") { const previousFailure = input.context.latestProviderFailure ?? input.providerRetry; const failureCode = @@ -6447,6 +6478,55 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ); }); + const subscriptionLimitPolicy = adapterOptions.subscriptionLimitPolicy; + const subscriptionLimitFailure = Effect.fnUntraced(function* ( + limits: ServerProviderUsageLimits | undefined, + ) { + return SubscriptionLimitPolicy.subscriptionLimitFailure({ + limits, + nowMs: DateTime.toEpochMillis(yield* DateTime.now), + provider: "Codex", + }); + }); + // The usage notifications only arrive while Codex works, so a turn needs a fresh read. + const readSubscriptionLimit = Effect.fn("CodexAdapterV2.readSubscriptionLimit")( + function* () { + const response = yield* client + .request("account/rateLimits/read", null) + .pipe(Effect.timeout("10 seconds"), Effect.option); + if (Option.isNone(response)) return yield* subscriptionLimitFailure(undefined); + yield* Ref.update(rateLimitSnapshot, (previous) => + mergeCodexRateLimits(previous, response.value.rateLimits), + ); + return yield* subscriptionLimitFailure( + codexRateLimitsToLimits({ + snapshot: response.value.rateLimits, + rateLimitsByLimitId: response.value.rateLimitsByLimitId, + checkedAt: DateTime.formatIso(yield* DateTime.now), + }), + ); + }, + ); + const refuseAtSubscriptionLimit = Effect.fnUntraced(function* ( + turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, + ) { + if ( + subscriptionLimitPolicy === undefined || + !(yield* subscriptionLimitPolicy.enabled(turnInput.threadId, turnInput.modelSelection)) + ) { + return; + } + const failure = yield* readSubscriptionLimit(); + if (failure === null) return; + return yield* new ProviderAdapter.ProviderAdapterTurnStartError({ + driver: CODEX_PROVIDER, + threadId: turnInput.threadId, + providerThreadId: turnInput.providerThread.id, + runId: turnInput.runId, + cause: new SubscriptionLimitPolicy.SubscriptionLimitStopError({ failure }), + }); + }); + const startNativeTurn = Effect.fn("CodexAdapterV2.startNativeTurn")( function* ( turnInput: ProviderAdapter.ProviderAdapterV2TurnInput, @@ -6900,8 +6980,13 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( turnInput.restartContinuationOfRunId === undefined ? parseCodexGoalCommand(turnInput.message.text) : null; - if (goalCommand !== null) return runGoalCommand(turnInput, goalCommand); + if (goalCommand !== null) { + return refuseAtSubscriptionLimit(turnInput).pipe( + Effect.andThen(runGoalCommand(turnInput, goalCommand)), + ); + } return Effect.gen(function* () { + yield* refuseAtSubscriptionLimit(turnInput); yield* startNativeTurn( turnInput, turnInput.restartContinuationOfRunId === undefined @@ -7640,6 +7725,60 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ), ), }; + if (subscriptionLimitPolicy !== undefined) { + // A usage update is read as it arrives; switching the guard on needs a fresh read. + const stopGuardedTurns = Effect.fnUntraced(function* (fresh: boolean) { + const guarded: Array = []; + for (const context of (yield* Ref.get(activeTurns)).values()) { + if (context.subagent !== null || context.subscriptionLimitStop !== undefined) + continue; + if ( + yield* subscriptionLimitPolicy.enabled( + context.input.threadId, + context.input.modelSelection, + ) + ) { + guarded.push(context); + } + } + if (guarded.length === 0) return; + const snapshot = yield* Ref.get(rateLimitSnapshot); + const failure = + fresh || snapshot === undefined + ? yield* readSubscriptionLimit() + : yield* subscriptionLimitFailure( + codexRateLimitsToLimits({ + snapshot, + checkedAt: DateTime.formatIso(yield* DateTime.now), + }), + ); + // These turns passed the check before they started, so a failed read leaves them + // running until a later update can tell. + if (failure?.class !== "usage_limit") return; + for (const context of guarded) { + context.subscriptionLimitStop = failure; + yield* runtime + .interruptTurn({ + providerThread: context.providerThread, + providerTurnId: context.providerTurnId, + }) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("orchestration-v2.codex-subscription-limit-stop-failed", { + providerTurnId: context.providerTurnId, + cause, + }), + ), + Effect.forkIn(scope), + ); + } + }); + yield* Stream.merge( + Stream.fromEffectRepeat(Queue.take(rateLimitUpdates)).pipe(Stream.as(false)), + subscriptionLimitPolicy.enabledChanges(input.threadId).pipe(Stream.as(true)), + ).pipe(Stream.runForEach(stopGuardedTurns), Effect.forkIn(scope)); + } + return runtime; }, (effect, input) => diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index dd69432077826..121f298e1ddbb 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -72,9 +72,11 @@ const isProviderTurnStartError = Schema.is(ProviderTurnStartError); /** * Claude refuses these before it reads the prompt: replacing a process running background work, - * or launch arguments whose appended system prompt cannot be read. + * or launch arguments whose appended system prompt cannot be read. Any guarded provider refuses a + * turn at its subscription limit. */ const PROMPT_REFUSAL_TAGS = [ + "SubscriptionLimitStopError", "ClaudeBackgroundWorkBlocksQueryReplacementError", "ClaudeAppendSystemPromptFileError", "ClaudeAppendSystemPromptMissingValueError", diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 38bc09c25cbb6..305de1666bcac 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -53,6 +53,8 @@ import type { PendingOrchestrationEffectV2 } from "./EffectOutbox.ts"; import * as EventSink from "./EventSink.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; +import { SubscriptionLimitStopError } from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; import * as McpAppModelContext from "../mcpApps/McpAppModelContext.ts"; @@ -4093,6 +4095,44 @@ it.effect("refreshes pull requests only once when startup failure closes its eve }), ); +it.effect("fails a run refused at the subscription limit with the guard's usage limit", () => + Effect.gen(function* () { + const ingestionStarted = yield* Deferred.make(); + const { observed, written } = yield* captureRootRunTermination({ + key: "subscription-limit-guard", + shouldFinalizeRun: () => Effect.succeed(true), + events: () => + Stream.unwrap(Deferred.succeed(ingestionStarted, undefined).pipe(Effect.as(Stream.never))), + startTurn: (input) => + Deferred.await(ingestionStarted).pipe( + Effect.andThen( + Effect.fail( + new ProviderAdapter.ProviderAdapterTurnStartError({ + driver, + threadId: input.threadId, + providerThreadId: input.providerThread.id, + runId: input.runId, + cause: new SubscriptionLimitStopError({ + failure: makeProviderFailure({ + class: "usage_limit", + code: "subscription_limit_guard", + message: "Codex subscription limit reached.", + resetAt: "2026-10-25T00:00:00.000Z", + }), + }), + }), + ), + ), + ), + }); + assert.equal(observed[0], "run:failed"); + const error = written.find((item) => item.type === "error"); + assert.equal(error?.failure.class, "usage_limit"); + assert.equal(error?.failure.code, "subscription_limit_guard"); + assert.equal(error?.failure.resetAt, "2026-10-25T00:00:00.000Z"); + }), +); + it.effect("keeps completed runs completed when pull request refresh fails", () => Effect.gen(function* () { const { observed } = yield* captureRootRunTermination({ diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 2ec4157a3baf4..7453f6a77f2fa 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -50,6 +50,7 @@ import { makeProviderFailure, makeProviderFailureTurnItem, } from "@t3tools/provider-core/server/failure"; +import { findSubscriptionLimitStop } from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; import * as RunFinalizationService from "./RunFinalizationService.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; @@ -1609,10 +1610,11 @@ export const layer: Layer.Layer< }, openRunOwnedSubagents: openSubagents, terminal: makeFailedTerminalEvent( - makeProviderFailure({ - cause: Cause.squash(cause), - class: Exit.isFailure(shouldStart) ? "unknown" : "provider_error", - }), + findSubscriptionLimitStop(Cause.squash(cause)) ?? + makeProviderFailure({ + cause: Cause.squash(cause), + class: Exit.isFailure(shouldStart) ? "unknown" : "provider_error", + }), latestItemOrdinal + 1, ), failureItemPersisted: false, diff --git a/apps/server/src/provider/Drivers/CodexDriver.ts b/apps/server/src/provider/Drivers/CodexDriver.ts index 5c0bb47c2b0cc..41b05b8374a8f 100644 --- a/apps/server/src/provider/Drivers/CodexDriver.ts +++ b/apps/server/src/provider/Drivers/CodexDriver.ts @@ -86,6 +86,8 @@ import * as CodexInstallation from "../CodexInstallation.ts"; import * as ServerSecretStore from "../../auth/ServerSecretStore.ts"; import * as ServerEnvironment from "../../environment/ServerEnvironment.ts"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as SubscriptionLimitGuards from "@t3tools/provider-core/server/SubscriptionLimitGuards"; +import { makeSubscriptionLimitPolicy } from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; const decodeCodexSettings = Schema.decodeSync(CodexSettings); const DRIVER_KIND = ProviderDriverKind.make("codex"); @@ -284,7 +286,13 @@ export const CodexDriver: ProviderDriver< enabled, config, }, - { onUsageLimits: (update) => snapshot.applyUsageLimits(update) }, + { + onUsageLimits: (update) => snapshot.applyUsageLimits(update), + subscriptionLimitPolicy: makeSubscriptionLimitPolicy({ + settings: host.settings, + guards: yield* Effect.serviceOption(SubscriptionLimitGuards.SubscriptionLimitGuards), + }), + }, ).pipe( Effect.mapError( (cause) => diff --git a/apps/server/src/provider/Drivers/CodexManagedProvider.ts b/apps/server/src/provider/Drivers/CodexManagedProvider.ts index af227ce473ba4..e4c1aea42fdb5 100644 --- a/apps/server/src/provider/Drivers/CodexManagedProvider.ts +++ b/apps/server/src/provider/Drivers/CodexManagedProvider.ts @@ -7,6 +7,9 @@ import * as Option from "effect/Option"; import { ChildProcessSpawner } from "effect/process"; import { makeCodexTextGeneration } from "../../textGeneration/CodexTextGeneration.ts"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; +import * as SubscriptionLimitGuards from "@t3tools/provider-core/server/SubscriptionLimitGuards"; +import { makeSubscriptionLimitPolicy } from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; +import { withCodexSubscriptionLimitOption } from "../codexModelCatalog.ts"; import { chatGptModels } from "../CodexChatGptModels.ts"; import { makeCodexManagedRuntime } from "../CodexManagedRuntime.ts"; import { ProviderDriverError } from "../Errors.ts"; @@ -141,19 +144,21 @@ export const makeManagedCodexProvider = Effect.fn("makeManagedCodexProvider")(fu auth: managedAuth, version: draft.version ?? executable.value.version, usageLimits, - models: draft.models.map((model) => ({ - ...model, - ...(model.capabilities - ? { - capabilities: { - ...model.capabilities, - optionDescriptors: (model.capabilities.optionDescriptors ?? []).filter( - (option) => option.id !== "serviceTier", - ), - }, - } - : {}), - })), + models: draft.models.map((model) => + withCodexSubscriptionLimitOption({ + ...model, + ...(model.capabilities + ? { + capabilities: { + ...model.capabilities, + optionDescriptors: (model.capabilities.optionDescriptors ?? []).filter( + (option) => option.id !== "serviceTier", + ), + }, + } + : {}), + }), + ), setup, runtimePaths, ...(draft.slashCommands @@ -230,6 +235,10 @@ export const makeManagedCodexProvider = Effect.fn("makeManagedCodexProvider")(fu const orchestrationAdapter = yield* createCodexAdapterV2(input, { onUsageLimits: (update) => snapshot.applyUsageLimits(update), resolveRuntime: runtime.resolve, + subscriptionLimitPolicy: makeSubscriptionLimitPolicy({ + settings: host.settings, + guards: yield* Effect.serviceOption(SubscriptionLimitGuards.SubscriptionLimitGuards), + }), }).pipe( Effect.mapError( (cause) => diff --git a/apps/server/src/provider/codexModelCatalog.ts b/apps/server/src/provider/codexModelCatalog.ts index 037dad8b0d9bc..4b802b8713d5c 100644 --- a/apps/server/src/provider/codexModelCatalog.ts +++ b/apps/server/src/provider/codexModelCatalog.ts @@ -1,5 +1,6 @@ -import type { ServerProviderUpdateRequiredModel } from "@t3tools/contracts"; +import type { ServerProviderModel, ServerProviderUpdateRequiredModel } from "@t3tools/contracts"; import * as ModelCatalog from "@t3tools/provider-core/server/ModelCatalog"; +import { withSubscriptionLimitOption } from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; import type { ServerProviderDraft } from "@t3tools/provider-core/server/snapshotProbe"; import { codexModelFamily } from "@t3tools/shared/model"; import { compareSemverVersions, parseSemver } from "@t3tools/shared/semver"; @@ -47,8 +48,17 @@ export function applyCodexModelCatalog( catalog: ModelCatalog.ProviderCatalog | undefined, ): ServerProviderDraft { const updateRequiredModels = codexUpdateRequiredModels(catalog, draft); + const applied = ModelCatalog.applyModelCatalog(draft, catalog, CODEX_MATCHING); return { - ...ModelCatalog.applyModelCatalog(draft, catalog, CODEX_MATCHING), + ...applied, + models: applied.models.map(withCodexSubscriptionLimitOption), ...(updateRequiredModels.length > 0 ? { updateRequiredModels } : {}), }; } + +export function withCodexSubscriptionLimitOption(model: ServerProviderModel): ServerProviderModel { + return withSubscriptionLimitOption( + model, + "Stop when Codex reports a usage limit used up, even if credits could continue the work. Reports can lag; this is not a billing cap.", + ); +} diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index 700d45cfe14f0..e0a3d038abca0 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -2433,7 +2433,7 @@ export function GeneralSettingsPanel() { General > Stop at subscription limit** (mobile: **Settings > Threads > -Usage limits**) protects all Claude threads on the selected environments, including -delegated work. It overrides a thread's Off option and immediately checks running -sessions when enabled. With the application setting off, threads can opt in through -the same-named Claude model option, also offered in the thread's context menu on web, -desktop, and mobile. Changing that option updates a Claude session that is already -running, and the application setting still overrides Off. Both settings default -to off. Settings remain on the server, so the guard also applies when clients -disconnect overnight. - -The server checks fresh subscription usage before protected turns, refuses to send -a prompt if usage is unavailable or the session/weekly allowance is exhausted, and -closes the Claude process when streamed events report exhaustion or active overage. -Closing stops its native background work too. Active turns use the existing Limited -status and usage-limit recovery controls, including optional resume after reset; -resuming does not bypass the guard. - -Agents use `stopAtSubscriptionLimit` through model options. Without the application -setting, delegated threads follow normal same-provider-and-model option inheritance. -Other providers do not advertise or enforce this guard yet. Provider reports may -lag, so it is not a hard billing cap. - -Code: [Claude model catalog](../apps/server/src/provider/ClaudeModelCatalog.ts), -[Claude driver](../apps/server/src/provider/Drivers/ClaudeDriver.ts), and -[Claude adapter](../apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts), +## Stop threads at subscription limits + +**Stop at subscription limit** stops Claude, Codex, and Cursor threads at their +subscription allowance instead of letting them continue on paid usage (Claude extra +usage, Codex credits, Cursor on-demand usage). **Settings > General** (mobile: +**Settings > Threads > Usage limits**) protects every such thread on the selected +environments, including delegated work, overrides a thread's Off option, and +rechecks running sessions when enabled. With it off, threads opt in through the +same-named model option, also in the thread's context menu on web, desktop, and +mobile; changing it applies to a running turn. Both default to off and live on +the server, so the guard holds when clients disconnect overnight. + +Before a protected turn the adapter reads fresh usage and refuses the prompt when +an allowance is used up or usage cannot be read; the run fails as a usage limit, so +Limited recovery and resume-after-reset apply and resuming does not bypass the +guard. During a turn, Claude stops on streamed exhaustion or overage events (closing +its process and native background work), Codex on its rate-limit notifications, and +Cursor on a one-minute usage poll. Agents use `stopAtSubscriptionLimit` through +model options. Other providers do not advertise or enforce it, and provider +reports can lag, so it is not a hard billing cap. + +Code: [shared policy](../packages/provider-core/src/server/SubscriptionLimitPolicy.ts), [live thread guard](../apps/server/src/orchestration-v2/SubscriptionLimitGuardSync.ts), -and [thread menu toggle](../packages/client-runtime/src/state/subscriptionLimitGuard.ts). -User guide: [Claude usage limits](./user/providers-claude.md#usage-limits). +[Claude adapter](../apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts), +[Codex adapter](../apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts), +[Cursor adapter](../packages/provider-cursor/src/server/adapter.ts), and +[thread menu toggle](../packages/client-runtime/src/state/subscriptionLimitGuard.ts). +User guide: [Stop at subscription limits](./user/usage.md#stop-at-subscription-limits). ## Usage-limit recovery diff --git a/docs/user/providers-claude.md b/docs/user/providers-claude.md index 52ed8818b9d6b..9c4136ffb37c9 100644 --- a/docs/user/providers-claude.md +++ b/docs/user/providers-claude.md @@ -71,34 +71,12 @@ reset, even with no app open; **Cancel auto-resume** turns it off), **Resume now**, and, on web and desktop, **Continue in new tab** on another account or model. Automatic resume is only offered when Claude reported a reset time. -For unattended work, enable **Stop at subscription limit** in **Settings > General** -(**Settings > Threads > Usage limits** on mobile). This applies to all Claude threads -on the selected environments, including delegated threads and threads whose own -option is Off. Enabling it also checks running Claude sessions immediately. -Disabling it removes the application-wide restriction; individual thread options -still apply. Other providers are not covered yet. - -For selected threads only, leave the application setting off and enable **Stop at -subscription limit** in that thread's Claude model options, or from the thread's -right-click menu (long-press on mobile). Changing a thread's own option also -applies to a Claude session that is already running: turning it on checks usage -and stops the session when the allowance is exhausted, and turning it off lets -that session continue. The application setting still overrides a thread's Off. -Agents can set `stopAtSubscriptionLimit: true` through model options; delegated -threads inherit that option under the normal same-provider-and-model rules. - -T3 checks subscription usage before protected turns and closes a Claude session -when it reports exhausted usage, even with paid extra usage enabled. Native -background work in that session also stops. A stopped turn appears as **Limited** -and offers the same **Resume now**, **Resume at reset**, and **Snooze until reset** -controls as an ordinary provider limit. **Resume now** keeps the guard: use another -provider or disable both the application and thread restrictions to allow paid -extra usage. Automatic resume checks the allowance again after reset. - -If usage cannot be checked, the protected turn stops before sending its prompt. -API-key and router accounts without subscription usage reporting cannot use this -option. Provider reports can arrive late, so this is a guard against unattended -spending, not a guarantee of zero extra charges. +For unattended work, **Stop at subscription limit** stops a thread at the +allowance instead of continuing on paid extra usage. See +[Stop at subscription limits](./usage.md#stop-at-subscription-limits). When it +stops a Claude session, native background work in that session stops too. +Router accounts using `ANTHROPIC_AUTH_TOKEN` do not report subscription usage and +cannot use it. ## Skills diff --git a/docs/user/providers-codex.md b/docs/user/providers-codex.md index 5f229945313ba..20577de208456 100644 --- a/docs/user/providers-codex.md +++ b/docs/user/providers-codex.md @@ -117,6 +117,9 @@ The thread keeps its work and offers a way to pick it back up: another account or model, carrying this conversation over. See [Switch accounts in an existing thread](#switch-accounts-in-an-existing-thread). +To stop at the allowance instead of continuing on credits, use **Stop at +subscription limit**. See [Stop at subscription limits](./usage.md#stop-at-subscription-limits). + Scheduling and cancelling an automatic resume are noted in the stopped turn. Sending any new message also moves the thread past the stop. See [Inspect agent work](./thread-sidebar.md#inspect-agent-work) for the settings diff --git a/docs/user/usage.md b/docs/user/usage.md index 0f305fe8fe6f6..e57fc2566251e 100644 --- a/docs/user/usage.md +++ b/docs/user/usage.md @@ -137,6 +137,33 @@ or endpoint configurations do not report subscription limits. API-key accounts may not report subscription limits. This also applies to Claude connections using a proxy through `ANTHROPIC_AUTH_TOKEN`. +## Stop at subscription limits + +Claude, Codex, and Cursor can keep working past a used-up subscription allowance on paid usage: +Claude's extra usage, Codex credits, or Cursor on-demand usage. For unattended work, **Stop at +subscription limit** stops a thread at the allowance instead. + +- To protect every Claude, Codex, and Cursor thread on the selected environments, including + delegated threads, enable it in **Settings > General** (**Settings > Threads > Usage limits** on + mobile). It overrides a thread's own Off, and enabling it checks running sessions immediately. +- For selected threads only, leave that setting off and enable **Stop at subscription limit** in the + thread's model options or its right-click menu (long-press on mobile). Changing it also applies + to a turn that is already running. Agents can set `stopAtSubscriptionLimit: true` through model + options; delegated threads inherit it under the normal same-provider-and-model rules. + +T3 checks fresh usage before each protected turn and refuses to send the prompt if an allowance is +used up or usage cannot be read. While a turn runs, it stops the turn when the provider reports the +allowance used up: Claude and Codex report this as they work, and Cursor is rechecked about once a +minute. A stopped turn appears as **Limited** with the usual **Resume now**, **Resume at reset**, and +**Snooze until reset** controls. Resuming keeps the guard; switch provider or turn the guard off to +allow paid usage. + +The guard needs subscription usage from the account, so API-key connections that do not report +limits cannot use it. For Cursor, that includes a configured `CURSOR_API_KEY`; on macOS a CLI login +also needs Cursor usage enabled. Cursor stops when either its Auto or API pool is used up, even if +the other has room. Provider reports can arrive late, so this guards against unattended spending +but does not guarantee zero paid usage. + ## Connect a CLIProxyAPI hub To see pooled accounts, open **Settings → Providers → Usage providers → Add hub**. Choose the diff --git a/packages/provider-core/package.json b/packages/provider-core/package.json index 9f8a68933f894..ed3775b478b24 100644 --- a/packages/provider-core/package.json +++ b/packages/provider-core/package.json @@ -75,6 +75,10 @@ "types": "./src/server/SubscriptionLimitGuards.ts", "import": "./src/server/SubscriptionLimitGuards.ts" }, + "./server/SubscriptionLimitPolicy": { + "types": "./src/server/SubscriptionLimitPolicy.ts", + "import": "./src/server/SubscriptionLimitPolicy.ts" + }, "./server/mcpSession": { "types": "./src/server/mcpSession.ts", "import": "./src/server/mcpSession.ts" diff --git a/packages/provider-core/src/server/SubscriptionLimitPolicy.test.ts b/packages/provider-core/src/server/SubscriptionLimitPolicy.test.ts new file mode 100644 index 0000000000000..04c412da3b12a --- /dev/null +++ b/packages/provider-core/src/server/SubscriptionLimitPolicy.test.ts @@ -0,0 +1,125 @@ +import { assert, describe, it } from "@effect/vitest"; +import { + DEFAULT_SERVER_SETTINGS, + ProviderInstanceId, + ThreadId, + type ServerProviderUsageLimits, +} from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; + +import * as SubscriptionLimitGuards from "./SubscriptionLimitGuards.ts"; +import { + SubscriptionLimitStopError, + findSubscriptionLimitStop, + makeSubscriptionLimitPolicy, + subscriptionLimitFailure, +} from "./SubscriptionLimitPolicy.ts"; + +const nowMs = Date.parse("2026-10-21T08:00:00.000Z"); +const limits = ( + windows: ReadonlyArray<{ id: string; usedPercent: number; resetsAt?: string }>, +): ServerProviderUsageLimits => ({ + checkedAt: "2026-10-21T08:00:00.000Z", + windows: windows.map((window) => ({ ...window, kind: "weekly", label: window.id })), +}); + +describe("subscriptionLimitFailure", () => { + it("lets a turn run while every counted window has room", () => { + const failure = subscriptionLimitFailure({ + limits: limits([ + { id: "five_hour", usedPercent: 99 }, + { id: "seven_day_opus", usedPercent: 100, resetsAt: "2026-10-22T00:00:00.000Z" }, + ]), + nowMs, + provider: "Claude", + windowIds: ["five_hour"], + }); + assert.isNull(failure); + }); + + it("stops at the latest reset of the exhausted windows, ignoring ones already reset", () => { + const failure = subscriptionLimitFailure({ + limits: limits([ + { id: "primary", usedPercent: 100, resetsAt: "2026-10-21T10:00:00.000Z" }, + { id: "secondary", usedPercent: 100, resetsAt: "2026-10-25T00:00:00.000Z" }, + { id: "individual", usedPercent: 100, resetsAt: "2026-10-21T07:00:00.000Z" }, + ]), + nowMs, + provider: "Codex", + }); + assert.equal(failure?.class, "usage_limit"); + assert.equal(failure?.code, "subscription_limit_guard"); + assert.equal(failure?.resetAt, "2026-10-25T00:00:00.000Z"); + assert.include(failure?.message ?? "", "Codex subscription limit reached"); + }); + + it("fails closed with the reader's reason when usage is unavailable", () => { + const failure = subscriptionLimitFailure({ + limits: { + ...limits([]), + unavailable: { reason: "unsupported", message: "Enable Cursor account usage" }, + }, + nowMs, + provider: "Cursor", + }); + assert.equal(failure?.class, "provider_error"); + assert.include(failure?.message ?? "", "(Enable Cursor account usage)"); + assert.equal( + subscriptionLimitFailure({ limits: undefined, nowMs, provider: "Cursor" })?.class, + "provider_error", + ); + }); +}); + +it("finds the guard's failure through wrapping start errors", () => { + const failure = subscriptionLimitFailure({ + limits: limits([{ id: "primary", usedPercent: 100 }]), + nowMs, + provider: "Codex", + }); + assert.isNotNull(failure); + if (failure === null) return; + const wrapped = { + _tag: "ProviderAdapterTurnStartError", + cause: new SubscriptionLimitStopError({ failure }), + }; + assert.strictEqual(findSubscriptionLimitStop(wrapped), failure); + assert.isUndefined(findSubscriptionLimitStop({ cause: new Error("other") })); +}); + +describe("makeSubscriptionLimitPolicy", () => { + const threadId = ThreadId.make("thread-policy"); + const selection = (enabled: boolean) => ({ + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.4", + options: [{ id: "stopAtSubscriptionLimit", value: enabled }], + }); + + it.effect("applies the application setting over a thread's Off", () => + Effect.gen(function* () { + const policy = makeSubscriptionLimitPolicy({ + settings: { + get: Effect.succeed({ ...DEFAULT_SERVER_SETTINGS, stopAtSubscriptionLimit: true }), + changes: Stream.empty, + }, + guards: Option.none(), + }); + assert.isTrue(yield* policy.enabled(threadId, selection(false))); + }), + ); + + it.effect("prefers the thread's latest choice to the selection its turn started with", () => + Effect.gen(function* () { + const guards = yield* SubscriptionLimitGuards.SubscriptionLimitGuards; + const policy = makeSubscriptionLimitPolicy({ + settings: { get: Effect.succeed(DEFAULT_SERVER_SETTINGS), changes: Stream.empty }, + guards: Option.some(guards), + }); + assert.isTrue(yield* policy.enabled(threadId, selection(true))); + yield* guards.set(threadId, false); + assert.isFalse(yield* policy.enabled(threadId, selection(true))); + }).pipe(Effect.provide(SubscriptionLimitGuards.layer)), + ); +}); diff --git a/packages/provider-core/src/server/SubscriptionLimitPolicy.ts b/packages/provider-core/src/server/SubscriptionLimitPolicy.ts new file mode 100644 index 0000000000000..f17b1a6bc496a --- /dev/null +++ b/packages/provider-core/src/server/SubscriptionLimitPolicy.ts @@ -0,0 +1,151 @@ +import type { + ModelSelection, + OrchestrationV2ProviderFailure, + ServerProviderModel, + ServerProviderUsageLimits, + ThreadId, +} from "@t3tools/contracts"; +import { getModelSelectionBooleanOptionValue } from "@t3tools/shared/model"; +import * as Data from "effect/Data"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; +import * as Predicate from "effect/Predicate"; +import * as Stream from "effect/Stream"; + +import { makeProviderFailure } from "./failure.ts"; +import type * as ProviderHost from "./ProviderHost.ts"; +import type * as SubscriptionLimitGuards from "./SubscriptionLimitGuards.ts"; + +export const SUBSCRIPTION_LIMIT_OPTION_ID = "stopAtSubscriptionLimit"; + +/** Adds the thread option a provider shows once its adapter enforces the guard. */ +export function withSubscriptionLimitOption( + model: ServerProviderModel, + description: string, +): ServerProviderModel { + return { + ...model, + capabilities: { + ...model.capabilities, + optionDescriptors: [ + ...(model.capabilities?.optionDescriptors ?? []).filter( + (option) => option.id !== SUBSCRIPTION_LIMIT_OPTION_ID, + ), + { + id: SUBSCRIPTION_LIMIT_OPTION_ID, + label: "Stop at subscription limit", + type: "boolean", + description, + }, + ], + }, + }; +} + +/** + * A guarded turn refused before its prompt reached the provider. The run fails + * with this usage-limit failure, so it gets the usual Limited recovery. + */ +export class SubscriptionLimitStopError extends Data.TaggedError("SubscriptionLimitStopError")<{ + readonly failure: OrchestrationV2ProviderFailure; +}> {} + +/** The guard's failure anywhere in a start error's cause chain. */ +export function findSubscriptionLimitStop( + error: unknown, +): OrchestrationV2ProviderFailure | undefined { + if (Predicate.isTagged(error, "SubscriptionLimitStopError")) { + return (error as SubscriptionLimitStopError).failure; + } + return Predicate.hasProperty(error, "cause") ? findSubscriptionLimitStop(error.cause) : undefined; +} + +/** + * Why a guarded turn must stop, or null when the allowance has room. Usage that + * cannot be read stops the turn too: a spending guard fails closed. + */ +export function subscriptionLimitFailure(input: { + readonly limits: ServerProviderUsageLimits | undefined; + readonly nowMs: number; + readonly provider: string; + /** The windows that gate paid usage; every reported window when omitted. */ + readonly windowIds?: ReadonlyArray; +}): OrchestrationV2ProviderFailure | null { + const { limits, windowIds } = input; + const windows = + limits?.windows.filter((window) => windowIds === undefined || windowIds.includes(window.id)) ?? + []; + const exhausted = windows.filter( + (window) => + window.usedPercent >= 100 && + (window.resetsAt === undefined || Date.parse(window.resetsAt) > input.nowMs), + ); + if ( + limits !== undefined && + limits.unavailable === undefined && + windows.length > 0 && + exhausted.length === 0 + ) { + return null; + } + if (exhausted.length === 0) { + const detail = limits?.unavailable?.message; + return makeProviderFailure({ + class: "provider_error", + retryable: false, + code: "subscription_limit_guard", + message: `${input.provider} subscription usage could not be checked${detail === undefined ? "" : ` (${detail})`}. Retry the turn, or disable Stop at subscription limit in application and thread settings to continue without this check.`, + }); + } + const resetTimes = exhausted.map((window) => window.resetsAt); + return makeProviderFailure({ + class: "usage_limit", + retryable: false, + code: "subscription_limit_guard", + message: `${input.provider} subscription limit reached. This thread stopped because Stop at subscription limit is enabled.`, + resetAt: resetTimes.every((time) => time !== undefined) + ? resetTimes.reduce((latest, time) => (time > latest ? time : latest)) + : null, + }); +} + +/** Whether the guard applies to a thread, and when that may have changed. */ +export interface SubscriptionLimitPolicy { + /** `selection` is the one the turn started with; a newer thread choice wins over it. */ + readonly enabled: ( + threadId: ThreadId, + selection: ModelSelection | undefined, + ) => Effect.Effect; + /** Emits each time the guard is switched on for the thread or the application. */ + readonly enabledChanges: (threadId: ThreadId) => Stream.Stream; +} + +export function makeSubscriptionLimitPolicy(input: { + readonly settings: Pick; + readonly guards: Option.Option; +}): SubscriptionLimitPolicy { + const appEnabled = input.settings.get.pipe( + Effect.map((settings) => settings.stopAtSubscriptionLimit), + // Unreadable settings must not let a guarded thread spend. + Effect.orElseSucceed(() => true), + ); + return { + enabled: Effect.fnUntraced(function* (threadId, selection) { + if (yield* appEnabled) return true; + if (Option.isSome(input.guards)) { + const live = yield* input.guards.value.read(threadId); + if (live !== undefined) return live; + } + return getModelSelectionBooleanOptionValue(selection, SUBSCRIPTION_LIMIT_OPTION_ID) === true; + }), + enabledChanges: (threadId) => + Stream.merge( + input.settings.changes.pipe(Stream.map((settings) => settings.stopAtSubscriptionLimit)), + Option.isSome(input.guards) ? input.guards.value.changes(threadId) : Stream.empty, + ).pipe( + Stream.changes, + Stream.filter((enabled) => enabled), + Stream.as(undefined), + ), + }; +} diff --git a/packages/provider-cursor/src/server/adapter.test.ts b/packages/provider-cursor/src/server/adapter.test.ts index 9405bfeba4d41..cc051c1e7584c 100644 --- a/packages/provider-cursor/src/server/adapter.test.ts +++ b/packages/provider-cursor/src/server/adapter.test.ts @@ -20,6 +20,8 @@ import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; @@ -29,6 +31,7 @@ import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; +import { findSubscriptionLimitStop } from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; import { CursorProviderCapabilitiesV2, cursorMcpServers, @@ -1180,6 +1183,8 @@ describe("CursorAdapterV2", () => { { id: "thinking", value: "high" }, { id: "contextWindow", value: "1m" }, { id: "fastMode", value: true }, + // T3's own thread option never reaches Cursor. + { id: "stopAtSubscriptionLimit", value: true }, ], }), { @@ -1683,3 +1688,179 @@ describe("CursorAdapterV2", () => { ), ); }); + +describe("CursorAdapterV2 subscription-limit guard", () => { + const makeGuardedSession = Effect.fn(function* (input: { + readonly name: string; + readonly usedPercent: Ref.Ref; + readonly guardSwitchedOn: Queue.Queue; + readonly cancelled: Deferred.Deferred; + }) { + const fileSystem = yield* FileSystem.FileSystem; + const workspace = yield* fileSystem.makeTempDirectoryScoped({ prefix: "cursor-v2-guard-" }); + const instanceId = ProviderInstanceId.make("cursor"); + const threadId = ThreadId.make(`cursor-guard-${input.name}`); + const modelSelection = { instanceId, model: "composer-2.5" }; + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: workspace, + }); + const sends = yield* Ref.make(0); + const adapter = yield* makeCursorAdapterV2({ + instanceId, + settings: yield* decodeCursorSettings({}), + environment: { HOME: workspace }, + subscriptionLimit: { + policy: { + enabled: () => Effect.succeed(true), + enabledChanges: () => Stream.fromQueue(input.guardSwitchedOn), + }, + checkUsage: Effect.gen(function* () { + return { + checkedAt: "2026-10-21T08:00:00.000Z", + windows: [ + { + id: "apiPercentUsed", + kind: "monthly" as const, + label: "API", + usedPercent: yield* Ref.get(input.usedPercent), + resetsAt: "2033-05-19T07:20:00.000Z", + }, + ], + }; + }), + }, + }).pipe( + Effect.provideService(CursorAgentSdk.CursorAgentSdkRunner, { + assertComplete: Effect.void, + open: () => + Effect.succeed({ + agentId: `native-cursor-guard-${input.name}`, + listMessages: Effect.succeed([]), + close: Effect.void, + send: () => + Ref.update(sends, (count) => count + 1).pipe( + Effect.as({ + agentId: `native-cursor-guard-${input.name}`, + runId: `native-cursor-guard-run-${input.name}`, + wait: Deferred.await(input.cancelled).pipe( + Effect.as({ + id: `native-cursor-guard-run-${input.name}`, + requestId: "native-request", + status: "cancelled" as const, + model: { id: "composer-2.5" }, + durationMs: 1, + }), + ), + cancel: Deferred.succeed(input.cancelled, undefined).pipe(Effect.asVoid), + steer: () => Effect.succeed("complete_delivered" as const), + }), + ), + }), + }), + Effect.provide(TestProviderHost.layer({ cwd: workspace })), + ); + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make(`cursor-guard-session-${input.name}`), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy }); + const now = yield* DateTime.now; + const startTurn = runtime.startTurn({ + threadId, + providerThread, + modelSelection, + runtimePolicy, + runId: RunId.make(`cursor-guard-run-${input.name}`), + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: RunAttemptId.make(`cursor-guard-attempt-${input.name}`), + rootNodeId: NodeId.make(`cursor-guard-root-${input.name}`), + appThread: { + id: threadId, + projectId: ProjectId.make("cursor-guard-project"), + createdBy: "user", + creationSource: "web", + title: "Cursor guard", + providerInstanceId: instanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: providerThread.id, + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId }, + forkedFrom: null, + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }, + message: { + messageId: MessageId.make(`cursor-guard-message-${input.name}`), + createdBy: "user", + creationSource: "web", + text: "Keep going.", + attachments: [], + }, + }); + return { runtime, startTurn, sends }; + }); + + it.effect("refuses a guarded turn before Cursor runs it once an allowance is used up", () => + Effect.gen(function* () { + const session = yield* makeGuardedSession({ + name: "refused", + usedPercent: yield* Ref.make(100), + guardSwitchedOn: yield* Queue.unbounded(), + cancelled: yield* Deferred.make(), + }); + const error = yield* session.startTurn.pipe(Effect.flip); + const failure = findSubscriptionLimitStop(error); + assert.equal(failure?.class, "usage_limit"); + assert.equal(failure?.resetAt, "2033-05-19T07:20:00.000Z"); + assert.equal(yield* Ref.get(session.sends), 0); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(NodeServices.layer, IdAllocator.layer, McpProviderSessions.layer), + ), + ), + ); + + it.effect("cancels a running guarded turn once a recheck finds the allowance used up", () => + Effect.gen(function* () { + const usedPercent = yield* Ref.make(40); + const guardSwitchedOn = yield* Queue.unbounded(); + const session = yield* makeGuardedSession({ + name: "live", + usedPercent, + guardSwitchedOn, + cancelled: yield* Deferred.make(), + }); + yield* session.startTurn; + yield* Ref.set(usedPercent, 100); + yield* Queue.offer(guardSwitchedOn, undefined); + const terminal = yield* session.runtime.events.pipe( + Stream.filter((event) => event.type === "turn.terminal"), + Stream.runHead, + ); + const event = Option.getOrUndefined(terminal); + assert.equal(event?.type === "turn.terminal" ? event.status : undefined, "failed"); + if (event?.type !== "turn.terminal" || event.status !== "failed") return; + assert.equal(event.failure.class, "usage_limit"); + assert.equal(event.failure.code, "subscription_limit_guard"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(NodeServices.layer, IdAllocator.layer, McpProviderSessions.layer), + ), + ), + ); +}); diff --git a/packages/provider-cursor/src/server/adapter.ts b/packages/provider-cursor/src/server/adapter.ts index 3ecaff1d68253..f764399b4b4fe 100644 --- a/packages/provider-cursor/src/server/adapter.ts +++ b/packages/provider-cursor/src/server/adapter.ts @@ -29,6 +29,7 @@ import { type OrchestrationV2TurnItem, type ProviderInstanceId, type ServerProviderModel, + type ServerProviderUsageLimits, type ThreadId, } from "@t3tools/contracts"; import { CursorSettings } from "../settings.ts"; @@ -59,6 +60,7 @@ import { t3OrchestrationPromptForFirstRun } from "@t3tools/provider-core/server/ import { buildRuntimeInstructions } from "@t3tools/provider-core/server/runtimeInstructions"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; +import * as SubscriptionLimitPolicy from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; import { turnScopedSelectionTransition } from "@t3tools/provider-core/server/selectionTransition"; import { isProviderNativeImageAttachment, @@ -940,6 +942,8 @@ interface ActiveCursorTurn { interrupted: boolean; finalized: boolean; usage?: TokenUsage; + /** Set when the subscription-limit guard cancels this run, which then fails with it. */ + subscriptionLimitStop?: OrchestrationV2ProviderFailure; } interface CursorLiveAgent { @@ -952,8 +956,16 @@ export interface CursorAdapterV2Options { readonly settings: CursorSettings; readonly environment: NodeJS.ProcessEnv; readonly getModels?: () => ReadonlyArray; + readonly subscriptionLimit?: { + readonly policy: SubscriptionLimitPolicy.SubscriptionLimitPolicy; + /** A fresh read: the provider snapshot's limits can be minutes old. */ + readonly checkUsage: Effect.Effect; + }; } +// Cursor reports no usage while a run streams, so a guarded run polls for it. +const CURSOR_SUBSCRIPTION_LIMIT_POLL_INTERVAL = "1 minute"; + export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( options: CursorAdapterV2Options, ) { @@ -2115,7 +2127,7 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( completedAt: input.completedAt, }); - const finalizeTurn = Effect.fnUntraced(function* (input: { + const finalizeTurn = Effect.fnUntraced(function* (requested: { readonly context: ActiveCursorTurn; readonly status: Extract< OrchestrationV2ProviderTurn["status"], @@ -2126,6 +2138,12 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( /** The finished run's summed usage; absent when the run returned no result. */ readonly usage?: TokenUsage; }) { + const stop = requested.context.subscriptionLimitStop; + // The guard's cancel ends the run as cancelled or failed; report why it stopped. + const input = + stop === undefined || requested.status === "completed" + ? requested + : { ...requested, status: "failed" as const, failure: stop }; if (input.context.finalized) { return; } @@ -2380,6 +2398,42 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( } satisfies SDKUserMessage); }); + const subscriptionLimit = adapterOptions.subscriptionLimit; + const readSubscriptionLimit = Effect.fnUntraced(function* () { + if (subscriptionLimit === undefined) return null; + return SubscriptionLimitPolicy.subscriptionLimitFailure({ + limits: yield* subscriptionLimit.checkUsage, + nowMs: DateTime.toEpochMillis(yield* DateTime.now), + provider: "Cursor", + }); + }); + if (subscriptionLimit !== undefined) { + const stopGuardedRun = Effect.fnUntraced(function* () { + const context = yield* Ref.get(activeTurn); + if ( + context === null || + context.finalized || + context.subscriptionLimitStop !== undefined || + !(yield* subscriptionLimit.policy.enabled( + context.input.threadId, + context.input.modelSelection, + )) + ) { + return; + } + const failure = yield* readSubscriptionLimit(); + // The run already passed the check before it started, so a failed read leaves it + // running until a later poll can tell. + if (failure?.class !== "usage_limit" || context.finalized) return; + context.subscriptionLimitStop = failure; + yield* context.run.cancel.pipe(Effect.ignore); + }); + yield* Stream.merge( + Stream.tick(CURSOR_SUBSCRIPTION_LIMIT_POLL_INTERVAL), + subscriptionLimit.policy.enabledChanges(input.threadId), + ).pipe(Stream.runForEach(stopGuardedRun), Effect.forkIn(sessionScope)); + } + const startTurn = Effect.fn("CursorAdapterV2.startTurn")( function* (turnInput: ProviderAdapter.ProviderAdapterV2TurnInput) { const current = yield* Ref.get(activeTurn); @@ -2390,6 +2444,18 @@ export const makeCursorAdapterV2 = Effect.fn("makeCursorAdapterV2")(function* ( reason: "previous_turn_active", }); } + if ( + subscriptionLimit !== undefined && + (yield* subscriptionLimit.policy.enabled( + turnInput.threadId, + turnInput.modelSelection, + )) + ) { + const failure = yield* readSubscriptionLimit(); + if (failure !== null) { + return yield* new SubscriptionLimitPolicy.SubscriptionLimitStopError({ failure }); + } + } const agentId = nativeThreadId(turnInput.providerThread); const agent = yield* openAgent({ operation: "resume", diff --git a/packages/provider-cursor/src/server/driver.ts b/packages/provider-cursor/src/server/driver.ts index 44ea3c4362434..11e61325018e7 100644 --- a/packages/provider-cursor/src/server/driver.ts +++ b/packages/provider-cursor/src/server/driver.ts @@ -20,9 +20,14 @@ import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as HttpClient from "effect/http/HttpClient"; -import { readCursorUsageLimits } from "./usageLimits.ts"; +import { readCursorUsageLimits, type CursorSdkCredential } from "./usageLimits.ts"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; +import * as SubscriptionLimitGuards from "@t3tools/provider-core/server/SubscriptionLimitGuards"; +import { + makeSubscriptionLimitPolicy, + withSubscriptionLimitOption, +} from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; import { makeCursorTextGeneration } from "./textGeneration.ts"; import { makeCursorAdapterV2, type CursorAdapterV2DriverEnv } from "./adapter.ts"; import { ProviderDriverError } from "@t3tools/provider-core/server/errors"; @@ -170,6 +175,12 @@ export const CursorDriver: ProviderDriver< const stampSnapshot: typeof stampIdentity = (draft) => stampIdentity({ ...draft, + models: draft.models.map((model) => + withSubscriptionLimitOption( + model, + "Stop when Cursor reports an included usage pool used up, even if on-demand usage is enabled. Usage is checked about once a minute; this is not a billing cap.", + ), + ), setup: { canAuthenticate: !auth.usesApiKey, canInstall: false }, auth: { ...draft.auth, @@ -211,12 +222,40 @@ export const CursorDriver: ProviderDriver< ), }); + // Each instance reads its own account's limits with its own key; only the default + // instance may fall back to the host's shared CLI login. + const readUsageLimits = (credential: CursorSdkCredential | undefined) => + host.settings.get.pipe( + Effect.flatMap((settings) => + readCursorUsageLimits(effectiveConfig, processEnv, { + allowKeychain: settings.cursorKeychainUsageEnabled, + sdkCredential: credential, + sharedLogin: instanceId === defaultInstanceIdForDriver(DRIVER_KIND), + }), + ), + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + Effect.provideService(CursorKeychain.CursorKeychain, keychain), + ); + let catalogModels: ReadonlyArray = []; const orchestrationAdapter = yield* makeCursorAdapterV2({ instanceId, settings: effectiveConfig, environment: processEnv, getModels: () => catalogModels, + subscriptionLimit: { + policy: makeSubscriptionLimitPolicy({ + settings: host.settings, + guards: yield* Effect.serviceOption(SubscriptionLimitGuards.SubscriptionLimitGuards), + }), + checkUsage: auth.readCredential.pipe( + Effect.orElseSucceed(() => undefined), + Effect.flatMap(readUsageLimits), + Effect.orElseSucceed(() => undefined), + ), + }, }).pipe( Effect.provideService(CursorAgentSdk.CursorAgentSdkRunner, { ...sdkRunner, @@ -279,16 +318,7 @@ export const CursorDriver: ProviderDriver< effectiveConfig.enabled && snapshot.installed && snapshot.auth.status === "authenticated" - ? host.settings.get.pipe( - Effect.flatMap((settings) => - // Each instance reads its own account's limits with its own key; only the - // default instance may fall back to the host's shared CLI login. - readCursorUsageLimits(effectiveConfig, processEnv, { - allowKeychain: settings.cursorKeychainUsageEnabled, - sdkCredential: credential, - sharedLogin: instanceId === defaultInstanceIdForDriver(DRIVER_KIND), - }), - ), + ? readUsageLimits(credential).pipe( Effect.map((usageLimits) => ({ ...snapshot, usageLimits })), ) : Effect.succeed(snapshot), diff --git a/packages/provider-cursor/src/server/sdkModel.ts b/packages/provider-cursor/src/server/sdkModel.ts index 2a2cb12daaae4..99b3fc89b8b9e 100644 --- a/packages/provider-cursor/src/server/sdkModel.ts +++ b/packages/provider-cursor/src/server/sdkModel.ts @@ -5,6 +5,7 @@ import type { ServerProviderModel, } from "@t3tools/contracts"; import { getProviderOptionCurrentValue } from "@t3tools/shared/model"; +import { SUBSCRIPTION_LIMIT_OPTION_ID } from "@t3tools/provider-core/server/SubscriptionLimitPolicy"; const CURSOR_SDK_PARAMETER_TO_PROVIDER_OPTION: Readonly> = { context: "contextWindow", @@ -60,12 +61,16 @@ export function cursorContextWindowTokens( } export function cursorSdkModelSelection(modelSelection: ModelSelection): CursorSdkModelSelection { + // A T3 thread option, not a Cursor model parameter: T3 enforces the guard itself. + const options = (modelSelection.options ?? []).filter( + (option) => option.id !== SUBSCRIPTION_LIMIT_OPTION_ID, + ); return { id: modelSelection.model === "auto" ? "default" : modelSelection.model, - ...(modelSelection.options === undefined || modelSelection.options.length === 0 + ...(options.length === 0 ? {} : { - params: modelSelection.options.map((option): ModelParameterValue => ({ + params: options.map((option): ModelParameterValue => ({ id: cursorSdkParameterId(option.id), value: String(option.value), })), @@ -87,6 +92,7 @@ export function withCursorDefaultParameters( const params = selection.params ?? []; const present = new Set(params.map((parameter) => parameter.id)); const defaults = (descriptors ?? []).flatMap((descriptor): Array => { + if (descriptor.id === SUBSCRIPTION_LIMIT_OPTION_ID) return []; const id = cursorSdkParameterId(descriptor.id); const value = descriptor.id === "fastMode" ? false : getProviderOptionCurrentValue(descriptor); return present.has(id) || value === undefined ? [] : [{ id, value: String(value) }];