From 620e2d823e32c66a3cd803245d2a514160313774 Mon Sep 17 00:00:00 2001 From: Jakub Chledowski Date: Wed, 11 Nov 2026 08:10:23 +0100 Subject: [PATCH 1/6] fix(server): using a banked usage reset continues threads waiting on that limit A thread stopped by a usage limit can be scheduled to continue at the limit's reset time and snoozed until then. Spending a banked reset under Usage -> Limits cleared the limit, but the thread kept waiting for the reset time the provider reported when the run failed. A redemption that returns `reset` now marks the armed recoveries of the redeemed account's provider instances as reset early. Their snooze until reset ends, and the recovery worker continues the ones with a scheduled continuation on its next sweep. A snooze the user set stays in place. Co-Authored-By: Claude Fable 5.2 --- .../src/orchestration-v2/Orchestrator.ts | 32 ++++- .../orchestration-v2/ProjectionStore.test.ts | 34 +++++ .../src/orchestration-v2/ProjectionStore.ts | 55 ++++++- .../ProviderTurnControlService.test.ts | 2 + .../UsageLimitRecoveryWorker.ts | 2 +- .../src/orchestration-v2/runtimeLayer.test.ts | 87 +++++++++++- apps/server/src/ws.test.ts | 134 ++++++++++++++++++ apps/server/src/ws.ts | 27 +++- docs/user/thread-sidebar.md | 7 + packages/contracts/src/orchestrationV2.ts | 3 + packages/shared/src/usageLimits.test.ts | 37 +++++ packages/shared/src/usageLimits.ts | 30 ++++ 12 files changed, 444 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 07c1a544798be..197755d37dbdb 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -385,6 +385,12 @@ export interface OrchestratorV2Shape { readonly threadId: ThreadId; readonly limitRecovery: OrchestrationV2LimitRecoveryUpdate; }) => Effect.Effect; + /** Marks recoveries waiting on these provider instances as reset early: their snooze ends and continuations start. */ + readonly recordLimitReset: (input: { + readonly providerInstanceIds: ReadonlyArray; + /** A run that failed later hit a limit the reset did not clear. */ + readonly failedBefore: DateTime.Utc; + }) => Effect.Effect; readonly searchThreadStream: ( input: OrchestrationV2SearchThreadInput, ) => Stream.Stream; @@ -4059,6 +4065,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio autoResume: command.limitRecovery.autoResume ?? previousRecovery?.autoResume ?? false, snooze: command.limitRecovery.snooze ?? previousRecovery?.snooze ?? false, + ...(previousRecovery?.resetEarly ? { resetEarly: true } : {}), requestId: command.commandId, }; return { @@ -5885,7 +5892,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio recovery.requestId !== command.usageLimitRecoveryRequestId || recovery.runId !== run.id || recovery.resetAt !== failure.resetAt || - Date.parse(recovery.resetAt) > DateTime.toEpochMillis(now) || + (recovery.resetEarly !== true && + Date.parse(recovery.resetAt) > DateTime.toEpochMillis(now)) || projection.thread.archivedAt !== null || projection.thread.deletedAt !== null || projection.thread.settledOverride === "settled" || @@ -14565,6 +14573,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ); }), + recordLimitReset: Effect.fn("OrchestratorV2.recordLimitReset")(function* (input) { + const waiting = yield* projectionStore + .getLimitRecoveriesAwaitingReset({ ...input, now: yield* DateTime.now }) + .pipe(Effect.mapError(shellProjectionError)); + yield* Effect.forEach( + waiting, + ({ threadId, limitRecovery: { runId, resetAt } }) => + dispatchWithReceipt({ + type: "thread.metadata.update", + commandId: CommandId.make(`limit-reset:${threadId}:${runId}:${Date.parse(resetAt)}`), + threadId, + // Ends only the snooze this recovery set; one the user chose stays. + limitRecovery: { runId, resetAt, snooze: false, resetEarly: true }, + }).pipe( + // A thread that moved on since the read is refused and keeps its own schedule. + Effect.catchCause((cause) => + Effect.logDebug("orchestration-v2.limit-recovery.reset-skipped", { threadId, cause }), + ), + ), + { discard: true }, + ); + }), searchThreadStream: (input) => projectionStore .searchThreadStream(input) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 40b7709870585..ba823624cecec 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6348,6 +6348,40 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { now: reset, })).find((row) => row.id === threadId)!; assert.deepEqual(due.limitRecovery, recovery); + const awaitingReset = store + .getLimitRecoveriesAwaitingReset({ + now, + providerInstanceIds: [original.providerInstanceId], + failedBefore: now, + }) + .pipe(Effect.map((rows) => rows.filter((row) => row.threadId === threadId))); + assert.deepEqual(yield* awaitingReset, [{ threadId, limitRecovery: recovery }]); + const [armedRow] = yield* sql<{ + payload_json: string; + }>`SELECT payload_json FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId}`; + // Nothing waits on a reset once it is recorded, cancelled, archived, settled or deleted. + for (const change of [ + sql`payload_json = json_set(payload_json, '$.limitRecovery.resetEarly', json('true'))`, + sql`payload_json = json_set(payload_json, '$.limitRecovery.autoResume', json('false'))`, + sql`payload_json = json_set(payload_json, '$.archivedAt', ${DateTime.formatIso(now)})`, + sql`payload_json = json_set(payload_json, '$.settledOverride', 'settled')`, + sql`deleted_at = ${DateTime.formatIso(now)}`, + ]) { + yield* sql`UPDATE orchestration_v2_projection_threads SET ${change} WHERE thread_id = ${threadId}`; + assert.deepEqual(yield* awaitingReset, []); + yield* sql`UPDATE orchestration_v2_projection_threads + SET payload_json = ${armedRow!.payload_json}, deleted_at = NULL WHERE thread_id = ${threadId}`; + } + yield* sql`UPDATE orchestration_v2_projection_threads + SET payload_json = json_set(payload_json, '$.limitRecovery.resetEarly', json('true')) + WHERE thread_id = ${threadId}`; + assert.isDefined( + (yield* store.getLimitRecoveryCandidates({ ...recoveryOptions, autoResume: true })).find( + (row) => row.id === threadId, + ), + ); + yield* sql`UPDATE orchestration_v2_projection_threads + SET payload_json = ${armedRow!.payload_json} WHERE thread_id = ${threadId}`; yield* sql`INSERT INTO orchestration_v2_projection_runtime_requests (runtime_request_id, thread_id, node_id, kind, status, created_at, payload_json) VALUES ('limit-shell:pending-request', ${threadId}, ${original.rootNodeId}, 'approval', 'pending', ${DateTime.formatIso(now)}, '{}')`; diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 0c4976dc90598..a94fe3a45bd6f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -29,6 +29,7 @@ import type { OrchestrationV2ConversationMessage, OrchestrationV2ContextTransfer, OrchestrationV2DomainEvent, + OrchestrationV2LimitRecovery, OrchestrationV2OlderSettledCount, OrchestrationV2OlderSettledCursor, OrchestrationV2ProjectedTurnItem, @@ -190,6 +191,11 @@ export type ProjectionLimitRecoveryCandidate = Pick< | "snoozedUntil" >; +export interface ProjectionAwaitedLimitRecovery { + readonly threadId: ThreadId; + readonly limitRecovery: OrchestrationV2LimitRecovery; +} + export type ProjectionThreadPullRequests = Pick< OrchestrationV2AppThread, "id" | "projectId" | "lineage" | "settledOverride" | "settledAt" | "pullRequests" @@ -558,6 +564,12 @@ export interface ProjectionStoreV2Shape { readonly getLimitRecoveryWakeAt: ( now: DateTime.Utc, ) => Effect.Effect; + /** Auto-resume or snooze recoveries not yet reset, for runs these instances failed by `failedBefore`. */ + readonly getLimitRecoveriesAwaitingReset: (options: { + readonly now: DateTime.Utc; + readonly providerInstanceIds: ReadonlyArray; + readonly failedBefore: DateTime.Utc; + }) => Effect.Effect, ProjectionStoreV2Error>; readonly getSettlementCandidates: ( threadId?: ThreadId, ) => Effect.Effect, ProjectionStoreV2Error>; @@ -4135,7 +4147,10 @@ export const layer: Layer.Layer = json_extract(t.payload_json, '$.limitRecovery.runId') IS r.run_id AND json_extract(t.payload_json, '$.limitRecovery.resetAt') IS json_extract(item.payload_json, '$.failure.resetAt') AND json_extract(t.payload_json, '$.limitRecovery.autoResume') = 1 - AND julianday(json_extract(item.payload_json, '$.failure.resetAt')) <= julianday(${DateTime.formatIso(options.now)}) + AND ( + json_extract(t.payload_json, '$.limitRecovery.resetEarly') = 1 + OR julianday(json_extract(item.payload_json, '$.failure.resetAt')) <= julianday(${DateTime.formatIso(options.now)}) + ) AND ( json_extract(t.payload_json, '$.snoozedUntil') IS NULL OR julianday(json_extract(t.payload_json, '$.snoozedUntil')) <= julianday(${DateTime.formatIso(options.now)}) @@ -4227,6 +4242,43 @@ export const layer: Layer.Layer = Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })), ); + const getLimitRecoveriesAwaitingReset = Effect.fn( + "ProjectionStore.getLimitRecoveriesAwaitingReset", + )( + function* ( + options: Parameters[0], + ) { + if (options.providerInstanceIds.length === 0) return []; + const rows = yield* sql<{ readonly payload_json: string }>` + SELECT t.payload_json + FROM orchestration_v2_projection_threads t + INNER JOIN orchestration_v2_projection_runs r + ON r.run_id = json_extract(t.payload_json, '$.limitRecovery.runId') + WHERE t.deleted_at IS NULL + AND json_extract(t.payload_json, '$.archivedAt') IS NULL + AND json_extract(t.payload_json, '$.settledOverride') IS NOT 'settled' + AND r.status = 'failed' + AND r.provider_instance_id IN ${sql.in(options.providerInstanceIds)} + AND julianday(COALESCE(r.completed_at, json_extract(t.payload_json, '$.updatedAt'))) <= julianday(${DateTime.formatIso(options.failedBefore)}) + AND julianday(json_extract(t.payload_json, '$.limitRecovery.resetAt')) > julianday(${DateTime.formatIso(options.now)}) + AND json_extract(t.payload_json, '$.limitRecovery.resetEarly') IS NOT 1 + AND ( + json_extract(t.payload_json, '$.limitRecovery.autoResume') = 1 + OR json_extract(t.payload_json, '$.limitRecovery.snooze') = 1 + ) + ORDER BY t.thread_id + `; + const waiting: Array = []; + for (const row of rows) { + const thread = yield* decodeThreadPayload(row.payload_json); + if (thread.limitRecovery != null) + waiting.push({ threadId: thread.id, limitRecovery: thread.limitRecovery }); + } + return waiting; + }, + Effect.mapError((cause) => new ProjectionStoreSetupError({ cause })), + ); + const getRecoveryThreadIds = Effect.fn("ProjectionStore.getRecoveryThreadIds")( function* ( kind: ProjectionRecoveryKind, @@ -7657,6 +7709,7 @@ export const layer: Layer.Layer = getProviderControlContext: readOnReader(getProviderControlContext), getLimitRecoveryCandidates: readOnReader(getLimitRecoveryCandidates), getLimitRecoveryWakeAt: readOnReader(getLimitRecoveryWakeAt), + getLimitRecoveriesAwaitingReset: readOnReader(getLimitRecoveriesAwaitingReset), getRecoveryThreadIds: readOnReader(getRecoveryThreadIds), getUnreadableThreadIds: readOnReader(getUnreadableThreadIds), getThreadSnapshot: readOnReader(getThreadSnapshot), diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 8b30a961c099d..a4489e1e0fe9b 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -211,6 +211,8 @@ it.effect( apply: () => Effect.void, getLimitRecoveryCandidates: () => Effect.die("unused getLimitRecoveryCandidates"), getLimitRecoveryWakeAt: () => Effect.die("unused getLimitRecoveryWakeAt"), + getLimitRecoveriesAwaitingReset: () => + Effect.die("unused getLimitRecoveriesAwaitingReset"), getOpenRemoteDelegatedTasks: Effect.die("unused getOpenRemoteDelegatedTasks"), getAgentAwarenessSnapshot: () => Effect.die("unused awareness snapshot"), getShellSnapshot: () => Effect.die("unused getShellSnapshot"), diff --git a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts index 985815bd73eb4..cf4c38d9c4d63 100644 --- a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts @@ -56,7 +56,7 @@ export function limitRecoveryCommand( } if ( !recovery.autoResume || - resetMs > nowMs || + (recovery.resetEarly !== true && resetMs > nowMs) || (thread.snoozedUntil != null && DateTime.toEpochMillis(thread.snoozedUntil) > nowMs) ) return null; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index a49e323770cc1..9fd9f0263f55d 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -8407,6 +8407,9 @@ it.layer(layerTest)("usage-limit recovery", (it) => { "stop", "stop-before-arm", "stop-before-arm-mismatched-recovery", + "early-reset", + "early-reset-snooze-only", + "early-reset-manual-snooze", ] as const)("guards a scheduled usage-limit continuation against %s", (scenario) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -8553,8 +8556,12 @@ it.layer(layerTest)("usage-limit recovery", (it) => { assert.isNull(current.thread.snoozedUntil); return; } - const snooze = scenario === "wake" || scenario === "cancel-resume-keep-snooze"; - const autoResume = scenario !== "wake"; + const snooze = + scenario === "wake" || + scenario === "cancel-resume-keep-snooze" || + scenario === "early-reset" || + scenario === "early-reset-snooze-only"; + const autoResume = scenario !== "wake" && scenario !== "early-reset-snooze-only"; const arm = limitRecoveryCommand(shell, autoResume, DateTime.toEpochMillis(now), snooze); assert(arm?.type === "thread.metadata.update" && arm.limitRecovery != null); const automaticArm = { @@ -8630,6 +8637,82 @@ it.layer(layerTest)("usage-limit recovery", (it) => { }); if (snooze) assert.equal(DateTime.toEpochMillis(armedShell.snoozedUntil!), Date.parse(resetAt)); + if ( + scenario === "early-reset" || + scenario === "early-reset-snooze-only" || + scenario === "early-reset-manual-snooze" + ) { + const manualWakeAt = DateTime.add(now, { seconds: 30 }); + if (scenario === "early-reset-manual-snooze") + yield* orchestrator.dispatch({ + type: "thread.snooze", + commandId: CommandId.make(`recovery:manual-snooze:${scenario}`), + threadId, + snoozedUntil: DateTime.formatIso(manualWakeAt), + }); + const waiting = yield* orchestrator.getThreadProjection(threadId); + // A reset of another account, or one from before this run failed, clears nothing here. + yield* orchestrator.recordLimitReset({ + providerInstanceIds: [alternateInstanceId], + failedBefore: now, + }); + yield* orchestrator.recordLimitReset({ + providerInstanceIds: [run.providerInstanceId], + failedBefore: DateTime.subtract(now, { seconds: 1 }), + }); + assert.deepEqual( + (yield* orchestrator.getThreadProjection(threadId)).thread, + waiting.thread, + ); + const recordReset = orchestrator.recordLimitReset({ + providerInstanceIds: [run.providerInstanceId], + failedBefore: now, + }); + yield* recordReset; + const reset = (yield* orchestrator.getThreadProjection(threadId)).thread; + assert.deepInclude(reset.limitRecovery, { autoResume, snooze: false, resetEarly: true }); + // The reset ends the snooze the recovery set, never one the user chose. + if (scenario === "early-reset-manual-snooze") + assert.equal( + DateTime.toEpochMillis(reset.snoozedUntil!), + DateTime.toEpochMillis(manualWakeAt), + ); + else assert.isNull(reset.snoozedUntil); + const projections = yield* ProjectionStore.ProjectionStoreV2; + const sweep = Effect.gen(function* () { + const sweptAt = yield* DateTime.now; + const candidate = (yield* projections.getLimitRecoveryCandidates({ + now: sweptAt, + autoResume: false, + snooze: false, + })).find((row) => row.id === threadId); + return candidate === undefined + ? null + : limitRecoveryCommand(candidate, false, DateTime.toEpochMillis(sweptAt)); + }); + if (scenario !== "early-reset") { + assert.isNull(yield* sweep); + if (scenario === "early-reset-manual-snooze") yield* TestClock.adjust("30 seconds"); + else + yield* orchestrator.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(`recovery:resume-after-reset:${scenario}`), + threadId, + limitRecovery: { runId: run.id, resetAt, autoResume: true }, + }); + } + const resume = yield* sweep; + assert(resume?.type === "message.dispatch"); + assert.isBelow(DateTime.toEpochMillis(yield* DateTime.now), Date.parse(resetAt)); + yield* orchestrator.dispatch(resume); + yield* recordReset; + yield* orchestrator.dispatch(resume); + const resumed = yield* orchestrator.getThreadProjection(threadId); + assert.lengthOf(resumed.runs, 2); + assert.equal(resumed.runs[1]!.status, "starting"); + assert.equal(resumed.runs[1]!.usageLimitContinuationOfRunId, run.id); + return; + } if (scenario === "cancel-resume-keep-snooze") { yield* TestClock.adjust("10 seconds"); yield* orchestrator.dispatch({ diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 6374204471964..747c24721cd7b 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -3,6 +3,7 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { + AuthProvidersManageScope, AuthSessionId, AuthSourceControlWriteScope, EnvironmentId, @@ -14,6 +15,7 @@ import { ProviderDriverKind, ProviderInstanceId, SourceControlRepositoryError, + UsageLimitSourceId, WS_METHODS, WsRpcGroup, WsServerRefreshProvidersRpc, @@ -1620,6 +1622,138 @@ it.effect("a publish whose push fails still refreshes the new remote", () => ), ); +it.effect.each([ + { redeemedOn: "an instance", outcome: "reset", recorded: true }, + { redeemedOn: "a hub", outcome: "reset", recorded: true }, + { redeemedOn: "an instance", outcome: "nothingToReset", recorded: false }, + { redeemedOn: "an instance", outcome: "reset", recorded: true, recordFails: true }, +] as const)( + "a reset credit redeemed on $redeemedOn with outcome $outcome ends limit waits: $recorded", + ({ redeemedOn, outcome, recorded, recordFails }) => + Effect.gen(function* () { + const provider = (id: string, email: string) => + ({ + instanceId: ProviderInstanceId.make(id), + driver: ProviderDriverKind.make("codex"), + enabled: true, + auth: { status: "authenticated", email }, + }) as ServerProvider; + const sourceId = UsageLimitSourceId.make("hub"); + const resets = + yield* Queue.unbounded< + Parameters[0] + >(); + const exits = yield* Queue.unbounded>(); + const redeem = TestClock.adjust("5 seconds").pipe(Effect.as(outcome)); + const services = Layer.mergeAll( + layerUnusedConfigRpcServices, + Layer.mock(Orchestrator.OrchestratorV2)({ + recordLimitReset: (input) => + Queue.offer(resets, input).pipe( + Effect.andThen(recordFails ? Effect.die("store unavailable") : Effect.void), + ), + }), + Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ + getInstance: () => + Effect.succeed({ enabled: true, consumeResetCredit: () => redeem } as never), + }), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed([ + provider("codex", "same@example.com"), + provider("codex-second-home", "same@example.com"), + provider("codex-other-account", "other@example.com"), + ]), + }), + Layer.mock(UsageLimitSources.UsageLimitSources)({ + consumeResetCredit: () => redeem.pipe(Effect.map((outcome) => ({ outcome }))), + current: Effect.succeed([ + { + id: sourceId, + kind: "cliproxy", + label: "Hub", + checkedAt: "2026-11-11T00:00:00.000Z", + accounts: [ + { + id: "pooled", + driver: ProviderDriverKind.make("codex"), + email: "same@example.com", + usageLimits: { checkedAt: "2026-11-11T00:00:00.000Z", windows: [] }, + }, + ], + }, + ]), + }), + Layer.mock(ExternalLauncher.ExternalLauncher)({}), + Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({}), + Layer.mock(DirectEndpoints.DirectEndpoints)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/unused" }), + Layer.mock(ScratchThreadPromotion.ScratchThreadPromotion)({}), + Layer.mock(ServerEnvironment.ServerEnvironment)({}), + Layer.mock(EnvironmentAuth.EnvironmentAuth)({}), + ); + const server = yield* RpcServer.makeNoSerialization(WsRpcGroup, { + ...WS_RPC_SERVER_OPTIONS, + onFromServer: (response) => + response._tag === "Exit" ? Queue.offer(exits, response.exit) : Effect.void, + }).pipe( + Effect.provide( + layerWsRpc( + { + sessionId: AuthSessionId.make("reset-credit-test"), + subject: "reset-credit-test", + method: "browser-session-cookie", + scopes: [AuthProvidersManageScope], + }, + {}, + yield* ServerBrowser.ServerBrowser, + () => Effect.void, + { secretReferences: true }, + ).pipe( + Layer.provide(services), + Layer.merge(RpcAuthorization.layer([AuthProvidersManageScope])), + ), + ), + ); + const startedAt = yield* DateTime.now; + yield* server.write(0, { + _tag: "Request", + id: RpcMessage.RequestId("redeem"), + tag: WS_METHODS.providerConsumeResetCredit, + payload: + redeemedOn === "a hub" + ? { sourceId, accountId: "pooled", creditId: "credit" } + : { instanceId: ProviderInstanceId.make("codex") }, + headers: Headers.empty, + }); + // The credit is spent either way, so the redemption reports its outcome. + assert.deepEqual(yield* Queue.take(exits), Exit.succeed({ outcome })); + assert.deepEqual( + yield* Queue.clear(resets), + recorded + ? [ + { + providerInstanceIds: [ + ProviderInstanceId.make("codex"), + ProviderInstanceId.make("codex-second-home"), + ], + // A run that failed while the provider was redeeming is not counted as reset. + failedBefore: startedAt, + }, + ] + : [], + ); + }).pipe( + Effect.provide( + Layer.mergeAll( + Layer.fresh(layerConfigSources), + Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), + Layer.mock(ServerBrowser.ServerBrowser)({}), + ), + ), + Effect.scoped, + ), +); + // Pruned history cannot be replayed: a cursor below the boundary must resync from a snapshot, // or, for a shell client that loads it over HTTP, be told to without the server building one. it.effect.each([ diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 5564b801d6b8a..ded6e5baabd49 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -77,6 +77,7 @@ import { ProjectSearchEntriesError, ProjectWriteFileError, ProjectMutationError, + type ProviderConsumeResetCreditInput, ProviderUploadFeedbackError, ProviderSetupError, RelayClientInstallFailedError, @@ -271,6 +272,7 @@ import { } from "@t3tools/shared/socketFrameParts"; import { isTelemetryIdHash } from "@t3tools/shared/telemetryId"; import { + resetCreditInstanceIds, sameUsageLimitCommandCoverage, withUsageLimitsCommands, } from "@t3tools/shared/usageLimits"; @@ -1899,6 +1901,23 @@ export const layerWsRpc = ( ), ); + // The credit is already spent, so a failure here must not fail its redemption. + const endLimitWaits = Effect.fn("ws.providers.endLimitWaits")( + function* (input: ProviderConsumeResetCreditInput, failedBefore: DateTime.Utc) { + yield* orchestrationEngine.recordLimitReset({ + providerInstanceIds: resetCreditInstanceIds( + input, + yield* providerRegistry.getProviders, + yield* usageLimitSources.current, + ), + failedBefore, + }); + }, + Effect.catchCause((cause) => + Effect.logWarning("limit recovery skipped after a usage reset", { cause }), + ), + ); + const loadServerConfig = Effect.fn("ws.loadServerConfig")(function* (options: { readonly usageLimitsCommand: boolean; readonly workspaceSnapshotsOnDemand: boolean; @@ -2633,7 +2652,12 @@ export const layerWsRpc = ( [WS_METHODS.providerConsumeResetCredit]: /* Request instrumentation already owns the span; keep this helper in that context. */ Effect.fnUntraced(function* (input) { - if ("sourceId" in input) return yield* usageLimitSources.consumeResetCredit(input); + const startedAt = yield* DateTime.now; + if ("sourceId" in input) { + const result = yield* usageLimitSources.consumeResetCredit(input); + if (result.outcome === "reset") yield* endLimitWaits(input, startedAt); + return result; + } const instance = yield* providerInstances.getInstance(input.instanceId); if (instance === undefined || !instance.enabled) { return yield* new ProviderSetupError({ @@ -2660,6 +2684,7 @@ export const layerWsRpc = ( }), ), ); + if (outcome === "reset") yield* endLimitWaits(input, startedAt); return { outcome }; }), [WS_METHODS.providerAuthStart]: (input) => providerAuth.start(input, currentSessionId), diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 09ce63e2a3e65..368a4dd80b3eb 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -268,6 +268,13 @@ the snooze. Enable **Snooze until limits reset** in thread behavior settings to snooze limit stops by default. Providers without a reset time offer manual retry and the normal snooze choices. +Using a banked reset from [**Usage → Limits**](./usage.md#track-subscription-limits) +counts as the reset for threads that account stopped in the same environment: +their snooze until reset ends and scheduled continuations start right away. A +thread you snoozed yourself stays snoozed and continues when it wakes. If the +reset did not refill the limit that stopped a thread, its continuation stops on +that limit again. + On web and desktop, use **Agents** to follow work delegated to subagents. On mobile, open a subagent from its conversation card or **Agents**. On iOS layouts with an inspector, its thread opens beside the parent. Maximize it to diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index b1fd02b3d3a07..a0944566707b4 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -404,6 +404,8 @@ export const OrchestrationV2LimitRecovery = Schema.Struct({ resetAt: IsoDateTime, autoResume: Schema.Boolean, snooze: Schema.optional(Schema.Boolean), + /** The provider reset the limit before `resetAt`, so a continuation no longer waits for it. */ + resetEarly: Schema.optional(Schema.Boolean), }); export type OrchestrationV2LimitRecovery = typeof OrchestrationV2LimitRecovery.Type; @@ -412,6 +414,7 @@ export const OrchestrationV2LimitRecoveryUpdate = Schema.Struct({ resetAt: IsoDateTime, autoResume: Schema.optional(Schema.Boolean), snooze: Schema.optional(Schema.Boolean), + resetEarly: Schema.optional(Schema.Literal(true)), }).check( Schema.makeFilter( (update) => diff --git a/packages/shared/src/usageLimits.test.ts b/packages/shared/src/usageLimits.test.ts index e9d5f6e604f31..e1cec5cc07b2c 100644 --- a/packages/shared/src/usageLimits.test.ts +++ b/packages/shared/src/usageLimits.test.ts @@ -21,6 +21,7 @@ import { exhaustedUsageWindow, paceOf, paceRemainingPercent, + resetCreditInstanceIds, usesChatGptSharing, resolveNewThreadModelSelection, } from "./usageLimits.ts"; @@ -1162,6 +1163,42 @@ describe("/usage-limits", () => { }); }); + it("finds the instances signed in to the account a reset credit was redeemed for", () => { + const instance = (id: string, email?: string, extra: Partial = {}) => + provider({ + instanceId: ProviderInstanceId.make(id), + auth: { status: "authenticated", ...(email ? { email } : {}) }, + ...extra, + }); + const providers = [ + selected, + instance("second-home", "same@example.com"), + instance("disabled", "same@example.com", { enabled: false }), + instance("other-account", "other@example.com"), + instance("claude", "same@example.com", { driver: ProviderDriverKind.make("claudeAgent") }), + instance("signed-out"), + ]; + const hub = (accountId: string) => ({ + sourceId: sources[0]!.id, + accountId, + creditId: "credit", + }); + expect(resetCreditInstanceIds({ instanceId: selected.instanceId }, providers, [])).toEqual([ + "codex", + "second-home", + ]); + // An account nothing else can be matched to still covers the instance it was redeemed on. + expect( + resetCreditInstanceIds({ instanceId: ProviderInstanceId.make("signed-out") }, providers, []), + ).toEqual(["signed-out"]); + expect(resetCreditInstanceIds(hub("duplicate"), providers, sources)).toEqual([ + "codex", + "second-home", + ]); + expect(resetCreditInstanceIds(hub("oss"), providers, sources)).toEqual([]); + expect(resetCreditInstanceIds(hub("removed"), providers, sources)).toEqual([]); + }); + it("gives a hub credit to no native row when its email is signed in to two workspaces", () => { const workspaces = ["ws-a", "ws-b"].map((workspaceId) => provider({ diff --git a/packages/shared/src/usageLimits.ts b/packages/shared/src/usageLimits.ts index 24403fe8a6c5b..7c8c047bd49c8 100644 --- a/packages/shared/src/usageLimits.ts +++ b/packages/shared/src/usageLimits.ts @@ -726,6 +726,36 @@ export function resetCreditChoices( ]; } +/** Enabled instances signed in to the account a reset credit belongs to. */ +export function resetCreditInstanceIds( + input: ProviderConsumeResetCreditInput, + providers: readonly ServerProvider[], + sources: UsageLimitSourceSnapshots, +): ReadonlyArray { + const enabled = providers.filter((provider) => provider.enabled); + const resolveKey = accountKeyResolver([ + ...enabled.map(nativeReport), + ...sources.flatMap((source) => source.accounts), + ]); + const signedInTo = (account: AccountReport | undefined) => { + const key = account ? resolveKey(account) : null; + return key === null + ? [] + : enabled.flatMap((provider) => + resolveKey(nativeReport(provider)) === key ? [provider.instanceId] : [], + ); + }; + if ("sourceId" in input) { + return signedInTo( + sources + .find((source) => source.id === input.sourceId) + ?.accounts.find((account) => account.id === input.accountId), + ); + } + const redeemedOn = enabled.find((provider) => provider.instanceId === input.instanceId); + return [...new Set([input.instanceId, ...signedInTo(redeemedOn && nativeReport(redeemedOn))])]; +} + export function formatResetsIn(window: ServerProviderUsageWindow, now: number): string | null { const resetsAt = resetMillis(window); if (resetsAt === null) return null; From 99b4119e5ec5253d79e29a6161dc0050a553ac56 Mon Sep 17 00:00:00 2001 From: Jakub Chledowski Date: Wed, 11 Nov 2026 08:39:23 +0100 Subject: [PATCH 2/6] fix(server): a redeemed reset continues only threads whose limit it refilled Review of the first commit found that a redemption continued every waiting thread of the account, whatever limit the credit refilled, and that a client could set the mark itself. - Redemption moves from the ws.ts handler into a ResetCredits service. It reads the account's limits after the redemption and leaves alone threads due at or after the reset of a limit that is still used up. An unconfirmed redemption continues nothing. A retry answered with alreadyRedeemed counts as the reset. A caller that disconnects once the credit is spent no longer interrupts the follow-up. - The mark is set by a server-only command, thread.limit-recovery.reset, so it is off the client's metadata update. - Each redemption uses its own command id, so a recovery refused once can be marked by a later redemption. A pending approval or question no longer refuses the mark. Co-Authored-By: Claude Fable 5.2 --- .../src/orchestration-v2/Orchestrator.ts | 81 ++++++- .../src/orchestration-v2/ProjectionStore.ts | 6 + .../src/orchestration-v2/runtimeLayer.test.ts | 30 ++- apps/server/src/server.ts | 2 + apps/server/src/usage/ResetCredits.test.ts | 214 ++++++++++++++++++ apps/server/src/usage/ResetCredits.ts | 128 +++++++++++ apps/server/src/ws.test.ts | 136 +---------- apps/server/src/ws.ts | 60 +---- docs/user/thread-sidebar.md | 6 +- packages/contracts/src/orchestrationV2.ts | 9 +- packages/shared/src/usageLimits.test.ts | 62 +++-- packages/shared/src/usageLimits.ts | 62 +++-- 12 files changed, 558 insertions(+), 238 deletions(-) create mode 100644 apps/server/src/usage/ResetCredits.test.ts create mode 100644 apps/server/src/usage/ResetCredits.ts diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 197755d37dbdb..a7874cc0dcf5e 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -390,6 +390,8 @@ export interface OrchestratorV2Shape { readonly providerInstanceIds: ReadonlyArray; /** A run that failed later hit a limit the reset did not clear. */ readonly failedBefore: DateTime.Utc; + /** A recovery that waits until this time or later waits on a limit the reset did not clear. */ + readonly resetBefore?: DateTime.Utc; }) => Effect.Effect; readonly searchThreadStream: ( input: OrchestrationV2SearchThreadInput, @@ -590,6 +592,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "thread.pull-request-watch.sync": case "thread.pull-request.sync": case "thread.title.regeneration.complete": + case "thread.limit-recovery.reset": case "thread.project.adopt-folder": case "thread.runtime-mode.set": case "thread.interaction-mode.set": @@ -4703,6 +4706,71 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, ); + const dispatchLimitRecoveryReset = Effect.fn("orchestrationV2.dispatch.limitRecoveryReset")( + function* ( + command: Extract< + OrchestrationV2ServerCommand, + { readonly type: "thread.limit-recovery.reset" } + >, + events: Ref.Ref>, + ) { + const projection = yield* projectionStore + .getThreadRecords(command.threadId, ["runs", "turnItems"], { turnItemTypes: ["error"] }) + .pipe( + Effect.mapError( + (cause) => new OrchestratorProjectionError({ threadId: command.threadId, cause }), + ), + ); + const thread = projection.thread; + const recovery = thread.limitRecovery; + const run = usageLimitBlockedRun(projection.runs, projection.turnItems, null); + const now = yield* DateTime.now; + if ( + recovery?.runId !== command.runId || + recovery.resetAt !== command.resetAt || + recovery.resetEarly === true || + !(recovery.autoResume || recovery.snooze) || + Date.parse(recovery.resetAt) <= DateTime.toEpochMillis(now) || + run?.id !== recovery.runId || + latestRootProviderFailure(run, projection.turnItems)?.resetAt !== recovery.resetAt || + thread.archivedAt !== null || + thread.deletedAt !== null || + thread.settledOverride === "settled" + ) { + return yield* new OrchestratorDispatchError({ + commandId: command.commandId, + commandType: command.type, + cause: `Thread ${command.threadId} no longer waits on that limit.`, + }); + } + yield* emit( + events, + command, + )({ + type: "thread.metadata-updated", + threadId: command.threadId, + providerInstanceId: thread.providerInstanceId, + occurredAt: now, + payload: { + ...thread, + limitRecovery: { + ...recovery, + snooze: false, + resetEarly: true, + requestId: command.commandId, + }, + // Ends only the snooze this recovery set; one the user chose stays. + ...(recovery.snooze && + thread.snoozedUntil != null && + DateTime.toEpochMillis(thread.snoozedUntil) === Date.parse(recovery.resetAt) + ? { snoozedUntil: null, snoozedAt: null, lastSnoozeWakeAt: now } + : {}), + updatedAt: now, + }, + }); + }, + ); + const dispatchThreadProjectAdoptFolder = Effect.fn( "orchestrationV2.dispatch.threadProjectAdoptFolder", )(function* ( @@ -13664,6 +13732,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "provider-session.detach": yield* dispatchProviderSessionDetach(command, events, effects); break; + case "thread.limit-recovery.reset": + yield* dispatchLimitRecoveryReset(command, events); + break; case "thread.project.adopt-folder": yield* dispatchThreadProjectAdoptFolder(command, events); break; @@ -14577,15 +14648,17 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const waiting = yield* projectionStore .getLimitRecoveriesAwaitingReset({ ...input, now: yield* DateTime.now }) .pipe(Effect.mapError(shellProjectionError)); + // A refusal is stored under its command id, so each reset tries under its own. + const reset = DateTime.toEpochMillis(input.failedBefore); yield* Effect.forEach( waiting, ({ threadId, limitRecovery: { runId, resetAt } }) => dispatchWithReceipt({ - type: "thread.metadata.update", - commandId: CommandId.make(`limit-reset:${threadId}:${runId}:${Date.parse(resetAt)}`), + type: "thread.limit-recovery.reset", + commandId: CommandId.make(`limit-reset:${threadId}:${runId}:${reset}`), threadId, - // Ends only the snooze this recovery set; one the user chose stays. - limitRecovery: { runId, resetAt, snooze: false, resetEarly: true }, + runId, + resetAt, }).pipe( // A thread that moved on since the read is refused and keeps its own schedule. Effect.catchCause((cause) => diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index a94fe3a45bd6f..4c087e4f1e75b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -569,6 +569,7 @@ export interface ProjectionStoreV2Shape { readonly now: DateTime.Utc; readonly providerInstanceIds: ReadonlyArray; readonly failedBefore: DateTime.Utc; + readonly resetBefore?: DateTime.Utc; }) => Effect.Effect, ProjectionStoreV2Error>; readonly getSettlementCandidates: ( threadId?: ThreadId, @@ -4249,6 +4250,7 @@ export const layer: Layer.Layer = options: Parameters[0], ) { if (options.providerInstanceIds.length === 0) return []; + const resetBefore = options.resetBefore ? DateTime.formatIso(options.resetBefore) : null; const rows = yield* sql<{ readonly payload_json: string }>` SELECT t.payload_json FROM orchestration_v2_projection_threads t @@ -4261,6 +4263,10 @@ export const layer: Layer.Layer = AND r.provider_instance_id IN ${sql.in(options.providerInstanceIds)} AND julianday(COALESCE(r.completed_at, json_extract(t.payload_json, '$.updatedAt'))) <= julianday(${DateTime.formatIso(options.failedBefore)}) AND julianday(json_extract(t.payload_json, '$.limitRecovery.resetAt')) > julianday(${DateTime.formatIso(options.now)}) + AND ( + ${resetBefore} IS NULL + OR julianday(json_extract(t.payload_json, '$.limitRecovery.resetAt')) < julianday(${resetBefore}) + ) AND json_extract(t.payload_json, '$.limitRecovery.resetEarly') IS NOT 1 AND ( json_extract(t.payload_json, '$.limitRecovery.autoResume') = 1 diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 9fd9f0263f55d..65975dc1f59ed 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -8651,24 +8651,41 @@ it.layer(layerTest)("usage-limit recovery", (it) => { snoozedUntil: DateTime.formatIso(manualWakeAt), }); const waiting = yield* orchestrator.getThreadProjection(threadId); - // A reset of another account, or one from before this run failed, clears nothing here. + const instances = { providerInstanceIds: [run.providerInstanceId] }; + // A reset of another account, one from before this run failed, or one that left this + // run's limit used up clears nothing here. yield* orchestrator.recordLimitReset({ providerInstanceIds: [alternateInstanceId], failedBefore: now, }); yield* orchestrator.recordLimitReset({ - providerInstanceIds: [run.providerInstanceId], + ...instances, failedBefore: DateTime.subtract(now, { seconds: 1 }), }); + yield* orchestrator.recordLimitReset({ + ...instances, + failedBefore: now, + resetBefore: DateTime.makeUnsafe(resetAt), + }); + // So does a reset of a limit this thread does not wait on. + yield* orchestrator + .dispatch({ + type: "thread.limit-recovery.reset", + commandId: CommandId.make(`recovery:other-limit:${scenario}`), + threadId, + runId: run.id, + resetAt: DateTime.formatIso(DateTime.add(DateTime.makeUnsafe(resetAt), { hours: 1 })), + }) + .pipe(Effect.flip); assert.deepEqual( (yield* orchestrator.getThreadProjection(threadId)).thread, waiting.thread, ); - const recordReset = orchestrator.recordLimitReset({ - providerInstanceIds: [run.providerInstanceId], + yield* orchestrator.recordLimitReset({ + ...instances, failedBefore: now, + resetBefore: DateTime.add(DateTime.makeUnsafe(resetAt), { seconds: 1 }), }); - yield* recordReset; const reset = (yield* orchestrator.getThreadProjection(threadId)).thread; assert.deepInclude(reset.limitRecovery, { autoResume, snooze: false, resetEarly: true }); // The reset ends the snooze the recovery set, never one the user chose. @@ -8705,7 +8722,8 @@ it.layer(layerTest)("usage-limit recovery", (it) => { assert(resume?.type === "message.dispatch"); assert.isBelow(DateTime.toEpochMillis(yield* DateTime.now), Date.parse(resetAt)); yield* orchestrator.dispatch(resume); - yield* recordReset; + // A second redemption, and a repeat of the continuation, start nothing more. + yield* orchestrator.recordLimitReset({ ...instances, failedBefore: yield* DateTime.now }); yield* orchestrator.dispatch(resume); const resumed = yield* orchestrator.getThreadProjection(threadId); assert.lengthOf(resumed.runs, 2); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 47190f60dcf51..986c1c549041e 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -114,6 +114,7 @@ import * as ProviderInstanceRegistry from "./provider/ProviderInstanceRegistry.t import * as ProviderAdapterRegistry from "./orchestration-v2/ProviderAdapterRegistry.ts"; import * as ProviderRegistry from "./provider/ProviderRegistry.ts"; import * as ProviderUsageLimitsIngestion from "./provider/ProviderUsageLimitsIngestion.ts"; +import * as ResetCredits from "./usage/ResetCredits.ts"; import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; import * as ProjectFaviconResolver from "./project/ProjectFaviconResolver.ts"; import * as T3ProjectFileLoader from "./project/T3ProjectFileLoader.ts"; @@ -682,6 +683,7 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( ProviderUsageLimitsIngestion.layer, layerProviderInstallationRefresh, ProviderInstallation.layer, + ResetCredits.layer, ReplayMarkers.layer, Layer.effectDiscard(ProjectSettingsOverrideSweep.sweep), ).pipe( diff --git a/apps/server/src/usage/ResetCredits.test.ts b/apps/server/src/usage/ResetCredits.test.ts new file mode 100644 index 0000000000000..fbb663858af0b --- /dev/null +++ b/apps/server/src/usage/ResetCredits.test.ts @@ -0,0 +1,214 @@ +import { assert, it } from "@effect/vitest"; +import { + type ProviderConsumeResetCreditOutcome, + ProviderDriverKind, + ProviderInstanceId, + type ServerProvider, + UsageLimitSourceId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as TestClock from "effect/testing/TestClock"; + +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as ResetCredits from "./ResetCredits.ts"; +import * as UsageLimitSources from "./UsageLimitSources.ts"; + +type LimitReset = Parameters[0]; + +const sourceId = UsageLimitSourceId.make("hub"); +const codex = ProviderInstanceId.make("codex"); +const weeklyReset = "1970-01-05T00:00:00.000Z"; +const usedUp = { id: "weekly", kind: "weekly", label: "Weekly", usedPercent: 100 } as const; + +/** What the redeemed account's limits look like when they are read after the redemption. */ +const LIMITS_AFTER = { + refilled: () => [], + // Monthly spend has no reset, so no thread waits on it. + weeklyUsedUp: () => [ + { ...usedUp, resetsAt: weeklyReset }, + { ...usedUp, id: "monthly_spend" }, + ], +} as const; + +const redeem = (input: { + readonly on: "instance" | "hub"; + readonly outcome: ProviderConsumeResetCreditOutcome; + readonly limits?: keyof typeof LIMITS_AFTER | "notReadAgain"; + readonly recordLimitReset: Orchestrator.OrchestratorV2["Service"]["recordLimitReset"]; +}) => + Effect.gen(function* () { + const limits = input.limits ?? "refilled"; + const startedAt = yield* DateTime.now; + const usageLimits = DateTime.now.pipe( + Effect.map((now) => ({ + checkedAt: DateTime.formatIso(limits === "notReadAgain" ? startedAt : now), + windows: limits === "notReadAgain" ? [] : LIMITS_AFTER[limits](), + })), + ); + const provider = (id: string, email: string, extra: Partial = {}) => + ({ + instanceId: ProviderInstanceId.make(id), + driver: ProviderDriverKind.make("codex"), + enabled: true, + auth: { status: "authenticated", email }, + ...extra, + }) as ServerProvider; + const consume = TestClock.adjust("5 seconds").pipe(Effect.as(input.outcome)); + const result = yield* Effect.gen(function* () { + return yield* (yield* ResetCredits.ResetCredits).redeem( + input.on === "hub" + ? { sourceId, accountId: "pooled", creditId: "credit" } + : { instanceId: codex }, + ); + }).pipe( + Effect.provide( + ResetCredits.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(Orchestrator.OrchestratorV2)({ + recordLimitReset: input.recordLimitReset, + }), + Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ + getInstance: () => + Effect.succeed({ + enabled: true, + consumeResetCredit: () => consume, + snapshot: { + getSnapshot: usageLimits.pipe( + Effect.map((usageLimits) => + provider("codex", "same@example.com", { usageLimits }), + ), + ), + }, + } as never), + }), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + // The registry still holds the redeemed instance's limits from before. + getProviders: Effect.succeed([ + provider("codex", "same@example.com", { + usageLimits: { + checkedAt: DateTime.formatIso(startedAt), + windows: [{ ...usedUp, resetsAt: weeklyReset }], + }, + }), + provider("codex-second-home", "same@example.com"), + provider("codex-other-account", "other@example.com"), + ]), + }), + Layer.mock(UsageLimitSources.UsageLimitSources)({ + consumeResetCredit: () => consume.pipe(Effect.map((outcome) => ({ outcome }))), + current: usageLimits.pipe( + Effect.map((usageLimits) => [ + { + id: sourceId, + kind: "cliproxy" as const, + label: "Hub", + checkedAt: usageLimits.checkedAt, + accounts: [ + { + id: "pooled", + driver: ProviderDriverKind.make("codex"), + email: "same@example.com", + usageLimits, + }, + ], + }, + ]), + ), + }), + ), + ), + ), + ), + ); + return { result, startedAt }; + }); + +it.effect.each([ + { on: "instance", outcome: "reset", ends: "all" }, + { on: "hub", outcome: "reset", ends: "all" }, + // A retry of a redemption whose answer was lost. + { on: "instance", outcome: "alreadyRedeemed", ends: "all" }, + { on: "instance", outcome: "reset", limits: "weeklyUsedUp", ends: "beforeWeeklyReset" }, + { on: "hub", outcome: "reset", limits: "weeklyUsedUp", ends: "beforeWeeklyReset" }, + { on: "instance", outcome: "reset", limits: "notReadAgain", ends: "none" }, + { on: "hub", outcome: "reset", limits: "notReadAgain", ends: "none" }, + { on: "instance", outcome: "nothingToReset", ends: "none" }, + { on: "instance", outcome: "noCredit", ends: "none" }, + { on: "instance", outcome: "reset", ends: "all", recordFails: true }, +] as const)( + "a reset credit redeemed on $on with outcome $outcome and limits $limits ends $ends limit waits", + ({ + ends, + recordFails, + ...input + }: { + readonly on: "instance" | "hub"; + readonly outcome: ProviderConsumeResetCreditOutcome; + readonly limits?: keyof typeof LIMITS_AFTER | "notReadAgain"; + readonly ends: "all" | "beforeWeeklyReset" | "none"; + readonly recordFails?: boolean; + }) => + Effect.gen(function* () { + const resets = yield* Queue.unbounded(); + const { result, startedAt } = yield* redeem({ + ...input, + recordLimitReset: (reset) => + Queue.offer(resets, reset).pipe( + Effect.andThen(recordFails ? Effect.die("store unavailable") : Effect.void), + ), + }); + // The credit is spent either way, so the redemption reports its outcome. + assert.deepEqual(result, { outcome: input.outcome }); + assert.deepEqual( + yield* Queue.clear(resets), + ends === "none" + ? [] + : [ + { + providerInstanceIds: [codex, ProviderInstanceId.make("codex-second-home")], + // A run that failed while the provider was redeeming is not counted as reset. + failedBefore: startedAt, + ...(ends === "beforeWeeklyReset" + ? { + resetBefore: DateTime.subtract(DateTime.makeUnsafe(weeklyReset), { + minutes: 1, + }), + } + : {}), + }, + ], + ); + }), +); + +it.effect("a caller that leaves once the credit is spent still ends the limit waits", () => + Effect.gen(function* () { + const recording = yield* Deferred.make(); + const release = yield* Deferred.make(); + const resets = yield* Queue.unbounded(); + const fiber = yield* redeem({ + on: "instance", + outcome: "reset", + recordLimitReset: (reset) => + Deferred.succeed(recording, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.andThen(Queue.offer(resets, reset)), + ), + }).pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(recording); + const leaving = yield* Fiber.interrupt(fiber).pipe( + Effect.forkChild({ startImmediately: true }), + ); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(leaving); + assert.lengthOf(yield* Queue.clear(resets), 1); + }), +); diff --git a/apps/server/src/usage/ResetCredits.ts b/apps/server/src/usage/ResetCredits.ts new file mode 100644 index 0000000000000..48d45d65a9485 --- /dev/null +++ b/apps/server/src/usage/ResetCredits.ts @@ -0,0 +1,128 @@ +import { + type ProviderConsumeResetCreditInput, + type ProviderConsumeResetCreditResult, + ProviderSetupError, + type ServerProvider, + type UsageLimitSourceError, +} from "@t3tools/contracts"; +import { resetCreditReach } from "@t3tools/shared/usageLimits"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; + +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as UsageLimitSources from "./UsageLimitSources.ts"; + +export class ResetCredits extends Context.Service< + ResetCredits, + { + /** Redeems a banked reset, then ends the waits of threads stopped by the limits it refilled. */ + readonly redeem: ( + input: ProviderConsumeResetCreditInput, + ) => Effect.Effect< + ProviderConsumeResetCreditResult, + ProviderSetupError | UsageLimitSourceError + >; + } +>()("t3/usage/ResetCredits") {} + +const make = Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const providerInstances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry; + const providerRegistry = yield* ProviderRegistry.ProviderRegistry; + const usageLimitSources = yield* UsageLimitSources.UsageLimitSources; + + const consume = Effect.fn("ResetCredits.consume")(function* ( + input: ProviderConsumeResetCreditInput, + ) { + if ("sourceId" in input) { + return { result: yield* usageLimitSources.consumeResetCredit(input) }; + } + const instance = yield* providerInstances.getInstance(input.instanceId); + if (instance === undefined || !instance.enabled) { + return yield* new ProviderSetupError({ + instanceId: input.instanceId, + operation: "consume-reset-credit", + detail: instance ? "This provider is disabled." : "Provider instance not found.", + }); + } + if (instance.consumeResetCredit === undefined) { + return yield* new ProviderSetupError({ + instanceId: input.instanceId, + operation: "consume-reset-credit", + detail: "This provider does not bank reset credits.", + }); + } + const outcome = yield* instance.consumeResetCredit(input.creditId).pipe( + Effect.mapError( + (error) => + new ProviderSetupError({ + instanceId: input.instanceId, + operation: "consume-reset-credit", + detail: error.detail, + cause: error, + }), + ), + ); + // The registry hears of the redemption's own limits read only later. + return { result: { outcome }, redeemedOn: yield* instance.snapshot.getSnapshot }; + }); + + // The credit is spent by now, so a failure here must not fail its redemption. + const endLimitWaits = Effect.fn("ResetCredits.endLimitWaits")( + function* ( + input: ProviderConsumeResetCreditInput, + since: DateTime.Utc, + redeemedOn: ServerProvider | undefined, + ) { + const providers = yield* providerRegistry.getProviders; + const reach = resetCreditReach( + input, + redeemedOn + ? [ + redeemedOn, + ...providers.filter((provider) => provider.instanceId !== redeemedOn.instanceId), + ] + : providers, + yield* usageLimitSources.current, + DateTime.toEpochMillis(since), + ); + if (reach === null) return; + yield* orchestrator.recordLimitReset({ + providerInstanceIds: reach.instanceIds, + failedBefore: since, + ...(reach.resetBefore === undefined + ? {} + : { resetBefore: DateTime.makeUnsafe(reach.resetBefore) }), + }); + }, + Effect.catchCause((cause) => + Effect.logWarning("limit recovery skipped after a usage reset", { cause }), + ), + ); + + const redeem: ResetCredits["Service"]["redeem"] = Effect.fn("ResetCredits.redeem")( + function* (input) { + const startedAt = yield* DateTime.now; + // A caller that goes away once the credit is spent must not leave its threads waiting. + return yield* Effect.uninterruptibleMask((restore) => + restore(consume(input)).pipe( + Effect.tap(({ result, redeemedOn }) => + // The second is a retry of a redemption whose answer was lost. + result.outcome === "reset" || result.outcome === "alreadyRedeemed" + ? endLimitWaits(input, startedAt, redeemedOn) + : Effect.void, + ), + Effect.map(({ result }) => result), + ), + ); + }, + ); + + return ResetCredits.of({ redeem }); +}); + +export const layer = Layer.effect(ResetCredits, make); diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 747c24721cd7b..d226fe2d1f35b 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -3,7 +3,6 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { - AuthProvidersManageScope, AuthSessionId, AuthSourceControlWriteScope, EnvironmentId, @@ -15,7 +14,6 @@ import { ProviderDriverKind, ProviderInstanceId, SourceControlRepositoryError, - UsageLimitSourceId, WS_METHODS, WsRpcGroup, WsServerRefreshProvidersRpc, @@ -115,6 +113,7 @@ import * as DatabaseStorage from "./persistence/DatabaseStorage.ts"; import * as TerminalManager from "./terminal/Manager.ts"; import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts"; import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; +import * as ResetCredits from "./usage/ResetCredits.ts"; import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; import * as UsageService from "./usage/UsageService.ts"; import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; @@ -824,6 +823,7 @@ const layerUnusedConfigRpcServices = Layer.mergeAll( Layer.mock(PullRequestSyncReactor.PullRequestSyncReactor)({}), Layer.mock(RelayClient.RelayClient)({}), Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({}), + Layer.mock(ResetCredits.ResetCredits)({}), Layer.mock(ResourceTelemetry.ResourceTelemetry)({}), Layer.mock(ReviewService.ReviewService)({}), Layer.mock(ScheduledTasks.ScheduledTaskService)({}), @@ -1622,138 +1622,6 @@ it.effect("a publish whose push fails still refreshes the new remote", () => ), ); -it.effect.each([ - { redeemedOn: "an instance", outcome: "reset", recorded: true }, - { redeemedOn: "a hub", outcome: "reset", recorded: true }, - { redeemedOn: "an instance", outcome: "nothingToReset", recorded: false }, - { redeemedOn: "an instance", outcome: "reset", recorded: true, recordFails: true }, -] as const)( - "a reset credit redeemed on $redeemedOn with outcome $outcome ends limit waits: $recorded", - ({ redeemedOn, outcome, recorded, recordFails }) => - Effect.gen(function* () { - const provider = (id: string, email: string) => - ({ - instanceId: ProviderInstanceId.make(id), - driver: ProviderDriverKind.make("codex"), - enabled: true, - auth: { status: "authenticated", email }, - }) as ServerProvider; - const sourceId = UsageLimitSourceId.make("hub"); - const resets = - yield* Queue.unbounded< - Parameters[0] - >(); - const exits = yield* Queue.unbounded>(); - const redeem = TestClock.adjust("5 seconds").pipe(Effect.as(outcome)); - const services = Layer.mergeAll( - layerUnusedConfigRpcServices, - Layer.mock(Orchestrator.OrchestratorV2)({ - recordLimitReset: (input) => - Queue.offer(resets, input).pipe( - Effect.andThen(recordFails ? Effect.die("store unavailable") : Effect.void), - ), - }), - Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ - getInstance: () => - Effect.succeed({ enabled: true, consumeResetCredit: () => redeem } as never), - }), - Layer.mock(ProviderRegistry.ProviderRegistry)({ - getProviders: Effect.succeed([ - provider("codex", "same@example.com"), - provider("codex-second-home", "same@example.com"), - provider("codex-other-account", "other@example.com"), - ]), - }), - Layer.mock(UsageLimitSources.UsageLimitSources)({ - consumeResetCredit: () => redeem.pipe(Effect.map((outcome) => ({ outcome }))), - current: Effect.succeed([ - { - id: sourceId, - kind: "cliproxy", - label: "Hub", - checkedAt: "2026-11-11T00:00:00.000Z", - accounts: [ - { - id: "pooled", - driver: ProviderDriverKind.make("codex"), - email: "same@example.com", - usageLimits: { checkedAt: "2026-11-11T00:00:00.000Z", windows: [] }, - }, - ], - }, - ]), - }), - Layer.mock(ExternalLauncher.ExternalLauncher)({}), - Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({}), - Layer.mock(DirectEndpoints.DirectEndpoints)({}), - Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/unused" }), - Layer.mock(ScratchThreadPromotion.ScratchThreadPromotion)({}), - Layer.mock(ServerEnvironment.ServerEnvironment)({}), - Layer.mock(EnvironmentAuth.EnvironmentAuth)({}), - ); - const server = yield* RpcServer.makeNoSerialization(WsRpcGroup, { - ...WS_RPC_SERVER_OPTIONS, - onFromServer: (response) => - response._tag === "Exit" ? Queue.offer(exits, response.exit) : Effect.void, - }).pipe( - Effect.provide( - layerWsRpc( - { - sessionId: AuthSessionId.make("reset-credit-test"), - subject: "reset-credit-test", - method: "browser-session-cookie", - scopes: [AuthProvidersManageScope], - }, - {}, - yield* ServerBrowser.ServerBrowser, - () => Effect.void, - { secretReferences: true }, - ).pipe( - Layer.provide(services), - Layer.merge(RpcAuthorization.layer([AuthProvidersManageScope])), - ), - ), - ); - const startedAt = yield* DateTime.now; - yield* server.write(0, { - _tag: "Request", - id: RpcMessage.RequestId("redeem"), - tag: WS_METHODS.providerConsumeResetCredit, - payload: - redeemedOn === "a hub" - ? { sourceId, accountId: "pooled", creditId: "credit" } - : { instanceId: ProviderInstanceId.make("codex") }, - headers: Headers.empty, - }); - // The credit is spent either way, so the redemption reports its outcome. - assert.deepEqual(yield* Queue.take(exits), Exit.succeed({ outcome })); - assert.deepEqual( - yield* Queue.clear(resets), - recorded - ? [ - { - providerInstanceIds: [ - ProviderInstanceId.make("codex"), - ProviderInstanceId.make("codex-second-home"), - ], - // A run that failed while the provider was redeeming is not counted as reset. - failedBefore: startedAt, - }, - ] - : [], - ); - }).pipe( - Effect.provide( - Layer.mergeAll( - Layer.fresh(layerConfigSources), - Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), - Layer.mock(ServerBrowser.ServerBrowser)({}), - ), - ), - Effect.scoped, - ), -); - // Pruned history cannot be replayed: a cursor below the boundary must resync from a snapshot, // or, for a shell client that loads it over HTTP, be told to without the server building one. it.effect.each([ diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index ded6e5baabd49..44ffd992b7d0d 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -77,7 +77,6 @@ import { ProjectSearchEntriesError, ProjectWriteFileError, ProjectMutationError, - type ProviderConsumeResetCreditInput, ProviderUploadFeedbackError, ProviderSetupError, RelayClientInstallFailedError, @@ -272,12 +271,12 @@ import { } from "@t3tools/shared/socketFrameParts"; import { isTelemetryIdHash } from "@t3tools/shared/telemetryId"; import { - resetCreditInstanceIds, sameUsageLimitCommandCoverage, withUsageLimitsCommands, } from "@t3tools/shared/usageLimits"; import * as AgentSessionScanner from "./project/AgentSessionScanner.ts"; import * as AgentSessionImporter from "./project/AgentSessionImporter.ts"; +import * as ResetCredits from "./usage/ResetCredits.ts"; import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; import { requestUrl } from "./requestUrl.ts"; @@ -1479,6 +1478,7 @@ export const layerWsRpc = ( ); const usage = yield* UsageService.UsageService; const usageLimitSources = yield* UsageLimitSources.UsageLimitSources; + const resetCredits = yield* ResetCredits.ResetCredits; const worktreeSetupTracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; const projectCloneTracker = yield* ProjectCloneTracker.ProjectCloneTracker; const repositoryIdentityResolver = @@ -1901,23 +1901,6 @@ export const layerWsRpc = ( ), ); - // The credit is already spent, so a failure here must not fail its redemption. - const endLimitWaits = Effect.fn("ws.providers.endLimitWaits")( - function* (input: ProviderConsumeResetCreditInput, failedBefore: DateTime.Utc) { - yield* orchestrationEngine.recordLimitReset({ - providerInstanceIds: resetCreditInstanceIds( - input, - yield* providerRegistry.getProviders, - yield* usageLimitSources.current, - ), - failedBefore, - }); - }, - Effect.catchCause((cause) => - Effect.logWarning("limit recovery skipped after a usage reset", { cause }), - ), - ); - const loadServerConfig = Effect.fn("ws.loadServerConfig")(function* (options: { readonly usageLimitsCommand: boolean; readonly workspaceSnapshotsOnDemand: boolean; @@ -2649,44 +2632,7 @@ export const layerWsRpc = ( ), [WS_METHODS.serverUpdateProvider]: (input) => providerMaintenanceRunner.updateProvider(input), - [WS_METHODS.providerConsumeResetCredit]: - /* Request instrumentation already owns the span; keep this helper in that context. */ - Effect.fnUntraced(function* (input) { - const startedAt = yield* DateTime.now; - if ("sourceId" in input) { - const result = yield* usageLimitSources.consumeResetCredit(input); - if (result.outcome === "reset") yield* endLimitWaits(input, startedAt); - return result; - } - const instance = yield* providerInstances.getInstance(input.instanceId); - if (instance === undefined || !instance.enabled) { - return yield* new ProviderSetupError({ - instanceId: input.instanceId, - operation: "consume-reset-credit", - detail: instance ? "This provider is disabled." : "Provider instance not found.", - }); - } - if (instance.consumeResetCredit === undefined) { - return yield* new ProviderSetupError({ - instanceId: input.instanceId, - operation: "consume-reset-credit", - detail: "This provider does not bank reset credits.", - }); - } - const outcome = yield* instance.consumeResetCredit(input.creditId).pipe( - Effect.mapError( - (error) => - new ProviderSetupError({ - instanceId: input.instanceId, - operation: "consume-reset-credit", - detail: error.detail, - cause: error, - }), - ), - ); - if (outcome === "reset") yield* endLimitWaits(input, startedAt); - return { outcome }; - }), + [WS_METHODS.providerConsumeResetCredit]: (input) => resetCredits.redeem(input), [WS_METHODS.providerAuthStart]: (input) => providerAuth.start(input, currentSessionId), [WS_METHODS.providerAuthRespond]: (input) => Effect.annotateCurrentSpan({ instanceId: input.instanceId }).pipe( diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 368a4dd80b3eb..c138e5145e984 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -271,9 +271,9 @@ retry and the normal snooze choices. Using a banked reset from [**Usage → Limits**](./usage.md#track-subscription-limits) counts as the reset for threads that account stopped in the same environment: their snooze until reset ends and scheduled continuations start right away. A -thread you snoozed yourself stays snoozed and continues when it wakes. If the -reset did not refill the limit that stopped a thread, its continuation stops on -that limit again. +thread you snoozed yourself stays snoozed and continues when it wakes. A thread +keeps waiting for its reset time when the limit it waits on is still used up +after the reset. On web and desktop, use **Agents** to follow work delegated to subagents. On mobile, open a subagent from its conversation card or **Agents**. On iOS diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index a0944566707b4..9e2da968886e4 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -414,7 +414,6 @@ export const OrchestrationV2LimitRecoveryUpdate = Schema.Struct({ resetAt: IsoDateTime, autoResume: Schema.optional(Schema.Boolean), snooze: Schema.optional(Schema.Boolean), - resetEarly: Schema.optional(Schema.Literal(true)), }).check( Schema.makeFilter( (update) => @@ -3444,6 +3443,14 @@ const OrchestrationV2InternalCommand = Schema.Union([ text: Schema.String, attachments: Schema.optional(Schema.Array(ChatAttachment)), }), + /** The provider reset this recovery's limit before `resetAt`. */ + Schema.Struct({ + type: Schema.Literal("thread.limit-recovery.reset"), + commandId: CommandId, + threadId: ThreadId, + runId: RunId, + resetAt: IsoDateTime, + }), /** Moves a thread to a project rooted at the folder it already works in, so its cwd never changes. */ Schema.Struct({ type: Schema.Literal("thread.project.adopt-folder"), diff --git a/packages/shared/src/usageLimits.test.ts b/packages/shared/src/usageLimits.test.ts index e1cec5cc07b2c..fe54f47ffd9b7 100644 --- a/packages/shared/src/usageLimits.test.ts +++ b/packages/shared/src/usageLimits.test.ts @@ -21,7 +21,7 @@ import { exhaustedUsageWindow, paceOf, paceRemainingPercent, - resetCreditInstanceIds, + resetCreditReach, usesChatGptSharing, resolveNewThreadModelSelection, } from "./usageLimits.ts"; @@ -1163,11 +1163,12 @@ describe("/usage-limits", () => { }); }); - it("finds the instances signed in to the account a reset credit was redeemed for", () => { + it("finds what a redeemed reset credit cleared", () => { const instance = (id: string, email?: string, extra: Partial = {}) => provider({ instanceId: ProviderInstanceId.make(id), auth: { status: "authenticated", ...(email ? { email } : {}) }, + usageLimits: limits, ...extra, }); const providers = [ @@ -1183,20 +1184,53 @@ describe("/usage-limits", () => { accountId, creditId: "credit", }); - expect(resetCreditInstanceIds({ instanceId: selected.instanceId }, providers, [])).toEqual([ - "codex", - "second-home", - ]); + const since = Date.parse(limits.checkedAt) - 1; + const sameAccount = { instanceIds: ["codex", "second-home"] }; + expect(resetCreditReach({ instanceId: selected.instanceId }, providers, [], since)).toEqual( + sameAccount, + ); // An account nothing else can be matched to still covers the instance it was redeemed on. expect( - resetCreditInstanceIds({ instanceId: ProviderInstanceId.make("signed-out") }, providers, []), - ).toEqual(["signed-out"]); - expect(resetCreditInstanceIds(hub("duplicate"), providers, sources)).toEqual([ - "codex", - "second-home", - ]); - expect(resetCreditInstanceIds(hub("oss"), providers, sources)).toEqual([]); - expect(resetCreditInstanceIds(hub("removed"), providers, sources)).toEqual([]); + resetCreditReach({ instanceId: ProviderInstanceId.make("signed-out") }, providers, [], since), + ).toEqual({ instanceIds: ["signed-out"] }); + expect(resetCreditReach(hub("duplicate"), providers, sources, since)).toEqual(sameAccount); + expect(resetCreditReach(hub("oss"), providers, sources, since)).toEqual({ instanceIds: [] }); + + // Nothing is known to be cleared until the account's limits are read after the redemption. + expect(resetCreditReach(hub("removed"), providers, sources, since)).toBeNull(); + expect( + resetCreditReach({ instanceId: selected.instanceId }, providers, [], since + 1), + ).toBeNull(); + const redeemedWith = (usageLimits: ServerProvider["usageLimits"]) => + resetCreditReach( + { instanceId: selected.instanceId }, + [{ ...selected, usageLimits }, ...providers.slice(1)], + [], + since, + ); + expect(redeemedWith(undefined)).toBeNull(); + expect(redeemedWith({ ...limits, unavailable: { reason: "probeFailed" } })).toBeNull(); + + // A limit that is still used up keeps the waits that last until it resets. + const usedUp = (resetsAt?: string) => ({ + ...window, + id: resetsAt ?? "monthly_spend", + usedPercent: 100, + resetsAt, + }); + expect( + redeemedWith({ + ...limits, + windows: [ + window, + usedUp("2026-09-06T12:00:00.000Z"), + usedUp("2026-09-04T12:00:00.000Z"), + // One that already reset, and one with no reset to wait for. + usedUp("2026-09-03T10:00:00.000Z"), + usedUp(), + ], + }), + ).toEqual({ ...sameAccount, resetBefore: Date.parse("2026-09-04T11:59:00.000Z") }); }); it("gives a hub credit to no native row when its email is signed in to two workspaces", () => { diff --git a/packages/shared/src/usageLimits.ts b/packages/shared/src/usageLimits.ts index 7c8c047bd49c8..6bc42353c239f 100644 --- a/packages/shared/src/usageLimits.ts +++ b/packages/shared/src/usageLimits.ts @@ -726,34 +726,58 @@ export function resetCreditChoices( ]; } -/** Enabled instances signed in to the account a reset credit belongs to. */ -export function resetCreditInstanceIds( +/** + * What a redeemed reset credit cleared: the enabled instances signed in to its account, and + * waits that end before `resetBefore`, where a limit still used up resets. Null until that + * account's limits are read again after `since`. + */ +export function resetCreditReach( input: ProviderConsumeResetCreditInput, providers: readonly ServerProvider[], sources: UsageLimitSourceSnapshots, -): ReadonlyArray { + since: number, +): { + readonly instanceIds: ReadonlyArray; + readonly resetBefore?: number; +} | null { const enabled = providers.filter((provider) => provider.enabled); + const redeemedOn = + "sourceId" in input + ? undefined + : enabled.find((provider) => provider.instanceId === input.instanceId); + const account: AccountReport | undefined = + "sourceId" in input + ? sources + .find((source) => source.id === input.sourceId) + ?.accounts.find((account) => account.id === input.accountId) + : redeemedOn && nativeReport(redeemedOn); + const limits = account?.usageLimits; + if (!account || !limits || limits.unavailable || !(Date.parse(limits.checkedAt) > since)) { + return null; + } const resolveKey = accountKeyResolver([ ...enabled.map(nativeReport), ...sources.flatMap((source) => source.accounts), ]); - const signedInTo = (account: AccountReport | undefined) => { - const key = account ? resolveKey(account) : null; - return key === null - ? [] - : enabled.flatMap((provider) => - resolveKey(nativeReport(provider)) === key ? [provider.instanceId] : [], - ); + const key = resolveKey(account); + const stillUsedUp = limits.windows.flatMap((window) => { + const resetsAt = resetMillis(window); + return window.usedPercent >= 100 && resetsAt !== null && resetsAt > since ? [resetsAt] : []; + }); + return { + instanceIds: [ + ...new Set([ + ...(redeemedOn ? [redeemedOn.instanceId] : []), + ...(key === null + ? [] + : enabled.flatMap((provider) => + resolveKey(nativeReport(provider)) === key ? [provider.instanceId] : [], + )), + ]), + ], + // A stop and a usage read report the same reset through different provider calls. + ...(stillUsedUp.length > 0 ? { resetBefore: Math.min(...stillUsedUp) - MINUTE } : {}), }; - if ("sourceId" in input) { - return signedInTo( - sources - .find((source) => source.id === input.sourceId) - ?.accounts.find((account) => account.id === input.accountId), - ); - } - const redeemedOn = enabled.find((provider) => provider.instanceId === input.instanceId); - return [...new Set([input.instanceId, ...signedInTo(redeemedOn && nativeReport(redeemedOn))])]; } export function formatResetsIn(window: ServerProviderUsageWindow, now: number): string | null { From dd2a9d857551eba66f3ca92b3eb1d6da196e45be Mon Sep 17 00:00:00 2001 From: Jakub Chledowski Date: Wed, 11 Nov 2026 09:08:56 +0100 Subject: [PATCH 3/6] fix(server): a redemption outlives its caller and marks only scheduled continuations Second review round on the reset-credit follow-up. - The redemption runs on a fiber the service owns. A caller that disconnects while the provider is still re-reading limits no longer skips the follow-up after the credit is spent. - Only outcome `reset` ends waits. `alreadyRedeemed` does not say whether this request refilled anything. - A recovery with no scheduled continuation is woken but not marked, and a recovery choice made after the reset goes by the reset time the thread still shows. The existing metadata reducer is unchanged again. - A failed mark is logged as a warning. Co-Authored-By: Claude Fable 5.2 --- .../src/orchestration-v2/Orchestrator.ts | 9 +- .../src/orchestration-v2/runtimeLayer.test.ts | 56 +++- apps/server/src/usage/ResetCredits.test.ts | 262 +++++++++--------- apps/server/src/usage/ResetCredits.ts | 20 +- docs/user/thread-sidebar.md | 6 +- packages/contracts/src/orchestrationV2.ts | 2 +- 6 files changed, 195 insertions(+), 160 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index a7874cc0dcf5e..3a0cb223d18c8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -4068,7 +4068,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio autoResume: command.limitRecovery.autoResume ?? previousRecovery?.autoResume ?? false, snooze: command.limitRecovery.snooze ?? previousRecovery?.snooze ?? false, - ...(previousRecovery?.resetEarly ? { resetEarly: true } : {}), requestId: command.commandId, }; return { @@ -4756,7 +4755,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio limitRecovery: { ...recovery, snooze: false, - resetEarly: true, + // Without a scheduled continuation the reset only ends the snooze. + ...(recovery.autoResume ? { resetEarly: true } : {}), requestId: command.commandId, }, // Ends only the snooze this recovery set; one the user chose stays. @@ -14662,7 +14662,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }).pipe( // A thread that moved on since the read is refused and keeps its own schedule. Effect.catchCause((cause) => - Effect.logDebug("orchestration-v2.limit-recovery.reset-skipped", { threadId, cause }), + Effect.logWarning("orchestration-v2.limit-recovery.reset-failed", { + threadId, + cause, + }), ), ), { discard: true }, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 65975dc1f59ed..05a866734ad5e 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -8410,6 +8410,9 @@ it.layer(layerTest)("usage-limit recovery", (it) => { "early-reset", "early-reset-snooze-only", "early-reset-manual-snooze", + "early-reset-then-snooze", + "early-reset-then-cancel", + "early-reset-then-stop", ] as const)("guards a scheduled usage-limit continuation against %s", (scenario) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -8637,13 +8640,12 @@ it.layer(layerTest)("usage-limit recovery", (it) => { }); if (snooze) assert.equal(DateTime.toEpochMillis(armedShell.snoozedUntil!), Date.parse(resetAt)); - if ( - scenario === "early-reset" || - scenario === "early-reset-snooze-only" || - scenario === "early-reset-manual-snooze" - ) { + if (scenario.startsWith("early-reset")) { const manualWakeAt = DateTime.add(now, { seconds: 30 }); - if (scenario === "early-reset-manual-snooze") + // Until a snooze the user set ends, the early reset is recorded and nothing continues. + const manualSnooze = + scenario === "early-reset-manual-snooze" || scenario.startsWith("early-reset-then"); + if (manualSnooze) yield* orchestrator.dispatch({ type: "thread.snooze", commandId: CommandId.make(`recovery:manual-snooze:${scenario}`), @@ -8687,9 +8689,11 @@ it.layer(layerTest)("usage-limit recovery", (it) => { resetBefore: DateTime.add(DateTime.makeUnsafe(resetAt), { seconds: 1 }), }); const reset = (yield* orchestrator.getThreadProjection(threadId)).thread; - assert.deepInclude(reset.limitRecovery, { autoResume, snooze: false, resetEarly: true }); + assert.deepInclude(reset.limitRecovery, { autoResume, snooze: false }); + // Without a scheduled continuation there is nothing to bring forward. + assert.equal(reset.limitRecovery?.resetEarly, autoResume ? true : undefined); // The reset ends the snooze the recovery set, never one the user chose. - if (scenario === "early-reset-manual-snooze") + if (manualSnooze) assert.equal( DateTime.toEpochMillis(reset.snoozedUntil!), DateTime.toEpochMillis(manualWakeAt), @@ -8707,17 +8711,39 @@ it.layer(layerTest)("usage-limit recovery", (it) => { ? null : limitRecoveryCommand(candidate, false, DateTime.toEpochMillis(sweptAt)); }); - if (scenario !== "early-reset") { - assert.isNull(yield* sweep); - if (scenario === "early-reset-manual-snooze") yield* TestClock.adjust("30 seconds"); - else + const choose = ( + choice: string, + limitRecovery: { autoResume?: boolean; snooze?: boolean }, + ) => + orchestrator.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(`recovery:${choice}:${scenario}`), + threadId, + limitRecovery: { runId: run.id, resetAt, ...limitRecovery }, + }); + if (scenario !== "early-reset") assert.isNull(yield* sweep); + if (scenario !== "early-reset" && scenario !== "early-reset-manual-snooze") { + // A choice made after the reset goes by the reset time the thread still shows. + if (scenario === "early-reset-then-snooze") yield* choose("snooze", { snooze: true }); + if (scenario === "early-reset-then-cancel") + yield* choose("cancel", { autoResume: false }); + if (scenario === "early-reset-then-stop") yield* orchestrator.dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make(`recovery:resume-after-reset:${scenario}`), + type: "thread.stop", + commandId: CommandId.make(`recovery:stop:${scenario}`), threadId, - limitRecovery: { runId: run.id, resetAt, autoResume: true }, }); + if (scenario !== "early-reset-then-snooze") yield* choose("resume", { autoResume: true }); + const chosen = (yield* orchestrator.getThreadProjection(threadId)).thread; + assert.deepInclude(chosen.limitRecovery, { autoResume: true }); + assert.isUndefined(chosen.limitRecovery?.resetEarly); + if (scenario === "early-reset-then-snooze") + assert.equal(DateTime.toEpochMillis(chosen.snoozedUntil!), Date.parse(resetAt)); + yield* TestClock.adjust("30 seconds"); + assert.isNull(yield* sweep); + return; } + if (manualSnooze) yield* TestClock.adjust("30 seconds"); const resume = yield* sweep; assert(resume?.type === "message.dispatch"); assert.isBelow(DateTime.toEpochMillis(yield* DateTime.now), Date.parse(resetAt)); diff --git a/apps/server/src/usage/ResetCredits.test.ts b/apps/server/src/usage/ResetCredits.test.ts index fbb663858af0b..aaa480940eda0 100644 --- a/apps/server/src/usage/ResetCredits.test.ts +++ b/apps/server/src/usage/ResetCredits.test.ts @@ -21,152 +21,154 @@ import * as ResetCredits from "./ResetCredits.ts"; import * as UsageLimitSources from "./UsageLimitSources.ts"; type LimitReset = Parameters[0]; +type LimitsAfter = "refilled" | "weeklyUsedUp" | "notReadAgain"; const sourceId = UsageLimitSourceId.make("hub"); const codex = ProviderInstanceId.make("codex"); const weeklyReset = "1970-01-05T00:00:00.000Z"; const usedUp = { id: "weekly", kind: "weekly", label: "Weekly", usedPercent: 100 } as const; +const provider = (id: string, email: string, extra: Partial = {}) => + ({ + instanceId: ProviderInstanceId.make(id), + driver: ProviderDriverKind.make("codex"), + enabled: true, + auth: { status: "authenticated", email }, + ...extra, + }) as ServerProvider; -/** What the redeemed account's limits look like when they are read after the redemption. */ -const LIMITS_AFTER = { - refilled: () => [], - // Monthly spend has no reset, so no thread waits on it. - weeklyUsedUp: () => [ - { ...usedUp, resetsAt: weeklyReset }, - { ...usedUp, id: "monthly_spend" }, - ], -} as const; - -const redeem = (input: { - readonly on: "instance" | "hub"; +/** The service over fake providers whose redemption takes five seconds. */ +const layerRedeeming = (input: { readonly outcome: ProviderConsumeResetCreditOutcome; - readonly limits?: keyof typeof LIMITS_AFTER | "notReadAgain"; + readonly limits?: LimitsAfter; + /** Runs once the provider has spent the credit, before it answers. */ + readonly afterSpend?: Effect.Effect; readonly recordLimitReset: Orchestrator.OrchestratorV2["Service"]["recordLimitReset"]; -}) => - Effect.gen(function* () { - const limits = input.limits ?? "refilled"; - const startedAt = yield* DateTime.now; - const usageLimits = DateTime.now.pipe( - Effect.map((now) => ({ - checkedAt: DateTime.formatIso(limits === "notReadAgain" ? startedAt : now), - windows: limits === "notReadAgain" ? [] : LIMITS_AFTER[limits](), - })), - ); - const provider = (id: string, email: string, extra: Partial = {}) => - ({ - instanceId: ProviderInstanceId.make(id), - driver: ProviderDriverKind.make("codex"), - enabled: true, - auth: { status: "authenticated", email }, - ...extra, - }) as ServerProvider; - const consume = TestClock.adjust("5 seconds").pipe(Effect.as(input.outcome)); - const result = yield* Effect.gen(function* () { - return yield* (yield* ResetCredits.ResetCredits).redeem( - input.on === "hub" - ? { sourceId, accountId: "pooled", creditId: "credit" } - : { instanceId: codex }, - ); - }).pipe( - Effect.provide( - ResetCredits.layer.pipe( - Layer.provide( - Layer.mergeAll( - Layer.mock(Orchestrator.OrchestratorV2)({ - recordLimitReset: input.recordLimitReset, - }), - Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ - getInstance: () => - Effect.succeed({ - enabled: true, - consumeResetCredit: () => consume, - snapshot: { - getSnapshot: usageLimits.pipe( - Effect.map((usageLimits) => - provider("codex", "same@example.com", { usageLimits }), - ), - ), - }, - } as never), - }), - Layer.mock(ProviderRegistry.ProviderRegistry)({ - // The registry still holds the redeemed instance's limits from before. - getProviders: Effect.succeed([ - provider("codex", "same@example.com", { - usageLimits: { - checkedAt: DateTime.formatIso(startedAt), - windows: [{ ...usedUp, resetsAt: weeklyReset }], - }, - }), - provider("codex-second-home", "same@example.com"), - provider("codex-other-account", "other@example.com"), - ]), - }), - Layer.mock(UsageLimitSources.UsageLimitSources)({ - consumeResetCredit: () => consume.pipe(Effect.map((outcome) => ({ outcome }))), - current: usageLimits.pipe( - Effect.map((usageLimits) => [ - { - id: sourceId, - kind: "cliproxy" as const, - label: "Hub", - checkedAt: usageLimits.checkedAt, - accounts: [ - { - id: "pooled", - driver: ProviderDriverKind.make("codex"), - email: "same@example.com", - usageLimits, - }, - ], - }, - ]), +}) => { + const consume = TestClock.adjust("5 seconds").pipe( + Effect.andThen(input.afterSpend ?? Effect.void), + Effect.as(input.outcome), + ); + // What the redeemed account's limits look like when they are read after the redemption. + const usageLimits = DateTime.now.pipe( + Effect.map((now) => ({ + checkedAt: DateTime.formatIso( + input.limits === "notReadAgain" ? DateTime.subtract(now, { seconds: 5 }) : now, + ), + windows: + input.limits === "weeklyUsedUp" + ? // Monthly spend has no reset, so no thread waits on it. + [ + { ...usedUp, resetsAt: weeklyReset }, + { ...usedUp, id: "monthly_spend" }, + ] + : [], + })), + ); + return ResetCredits.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(Orchestrator.OrchestratorV2)({ recordLimitReset: input.recordLimitReset }), + Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({ + getInstance: () => + Effect.succeed({ + enabled: true, + consumeResetCredit: () => consume, + snapshot: { + getSnapshot: usageLimits.pipe( + Effect.map((usageLimits) => + provider("codex", "same@example.com", { usageLimits }), + ), ), - }), - ), + }, + } as never), + }), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + // The registry still holds the redeemed instance's limits from before. + getProviders: Effect.succeed([ + provider("codex", "same@example.com", { + usageLimits: { + checkedAt: "1970-01-01T00:00:00.000Z", + windows: [{ ...usedUp, resetsAt: weeklyReset }], + }, + }), + provider("codex-second-home", "same@example.com"), + provider("codex-other-account", "other@example.com"), + ]), + }), + Layer.mock(UsageLimitSources.UsageLimitSources)({ + consumeResetCredit: () => consume.pipe(Effect.map((outcome) => ({ outcome }))), + current: usageLimits.pipe( + Effect.map((usageLimits) => [ + { + id: sourceId, + kind: "cliproxy" as const, + label: "Hub", + checkedAt: usageLimits.checkedAt, + accounts: [ + { + id: "pooled", + driver: ProviderDriverKind.make("codex"), + email: "same@example.com", + usageLimits, + }, + ], + }, + ]), ), - ), + }), ), - ); - return { result, startedAt }; - }); + ), + ); +}; + +const redeemOn = Effect.fn("test.redeemOn")(function* (on: "instance" | "hub") { + return yield* (yield* ResetCredits.ResetCredits).redeem( + on === "hub" ? { sourceId, accountId: "pooled", creditId: "credit" } : { instanceId: codex }, + ); +}); it.effect.each([ { on: "instance", outcome: "reset", ends: "all" }, { on: "hub", outcome: "reset", ends: "all" }, - // A retry of a redemption whose answer was lost. - { on: "instance", outcome: "alreadyRedeemed", ends: "all" }, { on: "instance", outcome: "reset", limits: "weeklyUsedUp", ends: "beforeWeeklyReset" }, { on: "hub", outcome: "reset", limits: "weeklyUsedUp", ends: "beforeWeeklyReset" }, { on: "instance", outcome: "reset", limits: "notReadAgain", ends: "none" }, { on: "hub", outcome: "reset", limits: "notReadAgain", ends: "none" }, { on: "instance", outcome: "nothingToReset", ends: "none" }, { on: "instance", outcome: "noCredit", ends: "none" }, + // The provider does not say whether this request's credit refilled anything. + { on: "instance", outcome: "alreadyRedeemed", ends: "none" }, { on: "instance", outcome: "reset", ends: "all", recordFails: true }, ] as const)( "a reset credit redeemed on $on with outcome $outcome and limits $limits ends $ends limit waits", ({ + on, ends, recordFails, - ...input + ...redemption }: { readonly on: "instance" | "hub"; readonly outcome: ProviderConsumeResetCreditOutcome; - readonly limits?: keyof typeof LIMITS_AFTER | "notReadAgain"; + readonly limits?: LimitsAfter; readonly ends: "all" | "beforeWeeklyReset" | "none"; readonly recordFails?: boolean; }) => Effect.gen(function* () { const resets = yield* Queue.unbounded(); - const { result, startedAt } = yield* redeem({ - ...input, - recordLimitReset: (reset) => - Queue.offer(resets, reset).pipe( - Effect.andThen(recordFails ? Effect.die("store unavailable") : Effect.void), - ), - }); + const startedAt = yield* DateTime.now; + const result = yield* redeemOn(on).pipe( + Effect.provide( + layerRedeeming({ + ...redemption, + recordLimitReset: (reset) => + Queue.offer(resets, reset).pipe( + Effect.andThen(recordFails ? Effect.die("store unavailable") : Effect.void), + ), + }), + ), + ); // The credit is spent either way, so the redemption reports its outcome. - assert.deepEqual(result, { outcome: input.outcome }); + assert.deepEqual(result, { outcome: redemption.outcome }); assert.deepEqual( yield* Queue.clear(resets), ends === "none" @@ -191,24 +193,28 @@ it.effect.each([ it.effect("a caller that leaves once the credit is spent still ends the limit waits", () => Effect.gen(function* () { - const recording = yield* Deferred.make(); - const release = yield* Deferred.make(); - const resets = yield* Queue.unbounded(); - const fiber = yield* redeem({ - on: "instance", - outcome: "reset", - recordLimitReset: (reset) => - Deferred.succeed(recording, undefined).pipe( - Effect.andThen(Deferred.await(release)), - Effect.andThen(Queue.offer(resets, reset)), - ), - }).pipe(Effect.forkChild({ startImmediately: true })); - yield* Deferred.await(recording); - const leaving = yield* Fiber.interrupt(fiber).pipe( - Effect.forkChild({ startImmediately: true }), + const spent = yield* Deferred.make(); + const answer = yield* Deferred.make(); + const recorded = yield* Deferred.make(); + yield* Effect.gen(function* () { + const caller = yield* redeemOn("instance").pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(spent); + yield* Fiber.interrupt(caller); + yield* Deferred.succeed(answer, undefined); + assert.deepEqual((yield* Deferred.await(recorded)).providerInstanceIds, [ + codex, + ProviderInstanceId.make("codex-second-home"), + ]); + }).pipe( + Effect.provide( + layerRedeeming({ + outcome: "reset", + afterSpend: Deferred.succeed(spent, undefined).pipe( + Effect.andThen(Deferred.await(answer)), + ), + recordLimitReset: (reset) => Deferred.succeed(recorded, reset), + }), + ), ); - yield* Deferred.succeed(release, undefined); - yield* Fiber.join(leaving); - assert.lengthOf(yield* Queue.clear(resets), 1); }), ); diff --git a/apps/server/src/usage/ResetCredits.ts b/apps/server/src/usage/ResetCredits.ts index 48d45d65a9485..5bb658eee627f 100644 --- a/apps/server/src/usage/ResetCredits.ts +++ b/apps/server/src/usage/ResetCredits.ts @@ -9,6 +9,7 @@ import { resetCreditReach } from "@t3tools/shared/usageLimits"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; @@ -34,6 +35,7 @@ const make = Effect.gen(function* () { const providerInstances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry; const providerRegistry = yield* ProviderRegistry.ProviderRegistry; const usageLimitSources = yield* UsageLimitSources.UsageLimitSources; + const scope = yield* Effect.scope; const consume = Effect.fn("ResetCredits.consume")(function* ( input: ProviderConsumeResetCreditInput, @@ -107,17 +109,15 @@ const make = Effect.gen(function* () { const redeem: ResetCredits["Service"]["redeem"] = Effect.fn("ResetCredits.redeem")( function* (input) { const startedAt = yield* DateTime.now; - // A caller that goes away once the credit is spent must not leave its threads waiting. - return yield* Effect.uninterruptibleMask((restore) => - restore(consume(input)).pipe( - Effect.tap(({ result, redeemedOn }) => - // The second is a retry of a redemption whose answer was lost. - result.outcome === "reset" || result.outcome === "alreadyRedeemed" - ? endLimitWaits(input, startedAt, redeemedOn) - : Effect.void, - ), - Effect.map(({ result }) => result), + return yield* consume(input).pipe( + Effect.tap(({ result, redeemedOn }) => + result.outcome === "reset" ? endLimitWaits(input, startedAt, redeemedOn) : Effect.void, ), + Effect.map(({ result }) => result), + // The credit may be spent before the caller hears of it. A caller that goes away must + // not leave the threads waiting, so only service shutdown ends this. + Effect.forkIn(scope), + Effect.flatMap(Fiber.join), ); }, ); diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index c138e5145e984..8b94167775acf 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -271,9 +271,9 @@ retry and the normal snooze choices. Using a banked reset from [**Usage → Limits**](./usage.md#track-subscription-limits) counts as the reset for threads that account stopped in the same environment: their snooze until reset ends and scheduled continuations start right away. A -thread you snoozed yourself stays snoozed and continues when it wakes. A thread -keeps waiting for its reset time when the limit it waits on is still used up -after the reset. +thread you snoozed yourself stays snoozed and continues when it wakes. Threads +keep waiting for their reset time if, after the reset, the account still has a +used-up limit that resets by then. On web and desktop, use **Agents** to follow work delegated to subagents. On mobile, open a subagent from its conversation card or **Agents**. On iOS diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 9e2da968886e4..58f7217994330 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -404,7 +404,7 @@ export const OrchestrationV2LimitRecovery = Schema.Struct({ resetAt: IsoDateTime, autoResume: Schema.Boolean, snooze: Schema.optional(Schema.Boolean), - /** The provider reset the limit before `resetAt`, so a continuation no longer waits for it. */ + /** The provider reset the limit before `resetAt`, so the scheduled continuation no longer waits for it. */ resetEarly: Schema.optional(Schema.Boolean), }); export type OrchestrationV2LimitRecovery = typeof OrchestrationV2LimitRecovery.Type; From bff275cd0fa911b21b8d0bc188f03ff0facc6ce3 Mon Sep 17 00:00:00 2001 From: Jakub Chledowski Date: Wed, 11 Nov 2026 09:17:35 +0100 Subject: [PATCH 4/6] test(server): a pending request holds back a recovery that was reset early Also say in the docs that a thread snoozed by hand starts its scheduled continuation on waking, since one without a schedule only wakes. Co-Authored-By: Claude Fable 5.2 --- .../src/orchestration-v2/ProjectionStore.test.ts | 15 ++++++++------- docs/user/thread-sidebar.md | 6 +++--- 2 files changed, 11 insertions(+), 10 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index ba823624cecec..8ecb54315ffe4 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -6375,16 +6375,17 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { yield* sql`UPDATE orchestration_v2_projection_threads SET payload_json = json_set(payload_json, '$.limitRecovery.resetEarly', json('true')) WHERE thread_id = ${threadId}`; - assert.isDefined( - (yield* store.getLimitRecoveryCandidates({ ...recoveryOptions, autoResume: true })).find( - (row) => row.id === threadId, - ), - ); - yield* sql`UPDATE orchestration_v2_projection_threads - SET payload_json = ${armedRow!.payload_json} WHERE thread_id = ${threadId}`; + const dueEarly = store + .getLimitRecoveryCandidates({ ...recoveryOptions, autoResume: true }) + .pipe(Effect.map((rows) => rows.find((row) => row.id === threadId))); + assert.isDefined(yield* dueEarly); yield* sql`INSERT INTO orchestration_v2_projection_runtime_requests (runtime_request_id, thread_id, node_id, kind, status, created_at, payload_json) VALUES ('limit-shell:pending-request', ${threadId}, ${original.rootNodeId}, 'approval', 'pending', ${DateTime.formatIso(now)}, '{}')`; + // A pending approval or question holds back a recovery that was reset early too. + assert.isUndefined(yield* dueEarly); + yield* sql`UPDATE orchestration_v2_projection_threads + SET payload_json = ${armedRow!.payload_json} WHERE thread_id = ${threadId}`; assert.isUndefined( (yield* store.getLimitRecoveryCandidates({ ...recoveryOptions, now: reset })).find( (row) => row.id === threadId, diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 8b94167775acf..f134e9aebc649 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -271,9 +271,9 @@ retry and the normal snooze choices. Using a banked reset from [**Usage → Limits**](./usage.md#track-subscription-limits) counts as the reset for threads that account stopped in the same environment: their snooze until reset ends and scheduled continuations start right away. A -thread you snoozed yourself stays snoozed and continues when it wakes. Threads -keep waiting for their reset time if, after the reset, the account still has a -used-up limit that resets by then. +thread you snoozed yourself stays snoozed, and its scheduled continuation starts +when it wakes. Threads keep waiting for their reset time if, after the reset, +the account still has a used-up limit that resets by then. On web and desktop, use **Agents** to follow work delegated to subagents. On mobile, open a subagent from its conversation card or **Agents**. On iOS From 7b838d7afc5cf50a00a10bb0619a8363a05de505 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 11 Nov 2026 09:19:42 +0000 Subject: [PATCH 5/6] fix(server): a reset that leaves the hub cooldown in place keeps threads waiting --- apps/server/src/usage/ResetCredits.test.ts | 17 +++++++++++++++-- apps/server/src/usage/ResetCredits.ts | 8 ++++++-- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/apps/server/src/usage/ResetCredits.test.ts b/apps/server/src/usage/ResetCredits.test.ts index aaa480940eda0..490b173433af0 100644 --- a/apps/server/src/usage/ResetCredits.test.ts +++ b/apps/server/src/usage/ResetCredits.test.ts @@ -40,6 +40,7 @@ const provider = (id: string, email: string, extra: Partial = {} const layerRedeeming = (input: { readonly outcome: ProviderConsumeResetCreditOutcome; readonly limits?: LimitsAfter; + readonly warning?: string; /** Runs once the provider has spent the credit, before it answers. */ readonly afterSpend?: Effect.Effect; readonly recordLimitReset: Orchestrator.OrchestratorV2["Service"]["recordLimitReset"]; @@ -96,7 +97,12 @@ const layerRedeeming = (input: { ]), }), Layer.mock(UsageLimitSources.UsageLimitSources)({ - consumeResetCredit: () => consume.pipe(Effect.map((outcome) => ({ outcome }))), + consumeResetCredit: () => + consume.pipe( + Effect.map((outcome) => + input.warning === undefined ? { outcome } : { outcome, warning: input.warning }, + ), + ), current: usageLimits.pipe( Effect.map((usageLimits) => [ { @@ -134,6 +140,7 @@ it.effect.each([ { on: "hub", outcome: "reset", limits: "weeklyUsedUp", ends: "beforeWeeklyReset" }, { on: "instance", outcome: "reset", limits: "notReadAgain", ends: "none" }, { on: "hub", outcome: "reset", limits: "notReadAgain", ends: "none" }, + { on: "hub", outcome: "reset", warning: "Cooldown not cleared.", ends: "none" }, { on: "instance", outcome: "nothingToReset", ends: "none" }, { on: "instance", outcome: "noCredit", ends: "none" }, // The provider does not say whether this request's credit refilled anything. @@ -150,6 +157,7 @@ it.effect.each([ readonly on: "instance" | "hub"; readonly outcome: ProviderConsumeResetCreditOutcome; readonly limits?: LimitsAfter; + readonly warning?: string; readonly ends: "all" | "beforeWeeklyReset" | "none"; readonly recordFails?: boolean; }) => @@ -168,7 +176,12 @@ it.effect.each([ ), ); // The credit is spent either way, so the redemption reports its outcome. - assert.deepEqual(result, { outcome: redemption.outcome }); + assert.deepEqual( + result, + redemption.warning === undefined + ? { outcome: redemption.outcome } + : { outcome: redemption.outcome, warning: redemption.warning }, + ); assert.deepEqual( yield* Queue.clear(resets), ends === "none" diff --git a/apps/server/src/usage/ResetCredits.ts b/apps/server/src/usage/ResetCredits.ts index 5bb658eee627f..895d863161938 100644 --- a/apps/server/src/usage/ResetCredits.ts +++ b/apps/server/src/usage/ResetCredits.ts @@ -70,7 +70,8 @@ const make = Effect.gen(function* () { ), ); // The registry hears of the redemption's own limits read only later. - return { result: { outcome }, redeemedOn: yield* instance.snapshot.getSnapshot }; + const result: ProviderConsumeResetCreditResult = { outcome }; + return { result, redeemedOn: yield* instance.snapshot.getSnapshot }; }); // The credit is spent by now, so a failure here must not fail its redemption. @@ -110,8 +111,11 @@ const make = Effect.gen(function* () { function* (input) { const startedAt = yield* DateTime.now; return yield* consume(input).pipe( + // A warning means the hub may still refuse until its cooldown ends. Effect.tap(({ result, redeemedOn }) => - result.outcome === "reset" ? endLimitWaits(input, startedAt, redeemedOn) : Effect.void, + result.outcome === "reset" && result.warning === undefined + ? endLimitWaits(input, startedAt, redeemedOn) + : Effect.void, ), Effect.map(({ result }) => result), // The credit may be spent before the caller hears of it. A caller that goes away must From d3d11c284a1ea5ccfb00ad2d9c10cb667dbefcc6 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Wed, 11 Nov 2026 09:48:34 +0000 Subject: [PATCH 6/6] fix(server): a limit reset read before a later recovery choice is refused --- apps/server/src/orchestration-v2/Orchestrator.ts | 4 +++- .../src/orchestration-v2/runtimeLayer.test.ts | 14 +++++++++++++- packages/contracts/src/orchestrationV2.ts | 2 ++ 3 files changed, 18 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index b7b46699d32fb..19421d579515d 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -4791,6 +4791,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( recovery?.runId !== command.runId || recovery.resetAt !== command.resetAt || + recovery.requestId !== command.requestId || recovery.resetEarly === true || !(recovery.autoResume || recovery.snooze) || Date.parse(recovery.resetAt) <= DateTime.toEpochMillis(now) || @@ -14719,13 +14720,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const reset = DateTime.toEpochMillis(input.failedBefore); yield* Effect.forEach( waiting, - ({ threadId, limitRecovery: { runId, resetAt } }) => + ({ threadId, limitRecovery: { runId, resetAt, requestId } }) => dispatchWithReceipt({ type: "thread.limit-recovery.reset", commandId: CommandId.make(`limit-reset:${threadId}:${runId}:${reset}`), threadId, runId, resetAt, + ...(requestId === undefined ? {} : { requestId }), }).pipe( // A thread that moved on since the read is refused and keeps its own schedule. Effect.catchCause((cause) => diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index d2fc2a030993d..d5f7b2eed7bf9 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -9000,7 +9000,8 @@ it.layer(layerTest)("usage-limit recovery", (it) => { failedBefore: now, resetBefore: DateTime.makeUnsafe(resetAt), }); - // So does a reset of a limit this thread does not wait on. + // So does a reset of a limit this thread does not wait on, or one read before a later + // recovery choice. yield* orchestrator .dispatch({ type: "thread.limit-recovery.reset", @@ -9008,6 +9009,17 @@ it.layer(layerTest)("usage-limit recovery", (it) => { threadId, runId: run.id, resetAt: DateTime.formatIso(DateTime.add(DateTime.makeUnsafe(resetAt), { hours: 1 })), + requestId: armCommand.commandId, + }) + .pipe(Effect.flip); + yield* orchestrator + .dispatch({ + type: "thread.limit-recovery.reset", + commandId: CommandId.make(`recovery:stale-choice:${scenario}`), + threadId, + runId: run.id, + resetAt, + requestId: CommandId.make(`recovery:earlier-choice:${scenario}`), }) .pipe(Effect.flip); assert.deepEqual( diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 58f7217994330..cb6c9adfcbfa0 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -3450,6 +3450,8 @@ const OrchestrationV2InternalCommand = Schema.Union([ threadId: ThreadId, runId: RunId, resetAt: IsoDateTime, + /** The recovery choice the reset was read from; a later choice refuses it. */ + requestId: Schema.optional(CommandId), }), /** Moves a thread to a project rooted at the folder it already works in, so its cwd never changes. */ Schema.Struct({