diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 226b7d703d0b1..15bdcf845a18a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -417,6 +417,14 @@ 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; + /** 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, ) => Stream.Stream; @@ -649,6 +657,7 @@ function commandThreadId(command: OrchestrationV2ServerCommand): ThreadId { case "thread.pull-request-watch.checks-reopen": 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": @@ -5489,6 +5498,73 @@ 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.requestId !== command.requestId || + 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, + // 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. + ...(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* ( @@ -6787,7 +6863,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" || @@ -15033,6 +15110,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; @@ -16124,6 +16204,34 @@ 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)); + // 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, 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) => + Effect.logWarning("orchestration-v2.limit-recovery.reset-failed", { + 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 50f2a0a12d1ae..df74e1472a76b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -8295,9 +8295,44 @@ 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}`; + 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/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index b5e01ea08b52b..d01bcc591b502 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -30,6 +30,7 @@ import type { OrchestrationV2ConversationMessage, OrchestrationV2ContextTransfer, OrchestrationV2DomainEvent, + OrchestrationV2LimitRecovery, OrchestrationV2OlderSettledCount, OrchestrationV2OlderSettledCursor, OrchestrationV2ProjectedTurnItem, @@ -196,6 +197,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" @@ -622,6 +628,13 @@ 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; + readonly resetBefore?: DateTime.Utc; + }) => Effect.Effect, ProjectionStoreV2Error>; readonly getSettlementCandidates: ( threadId?: ThreadId, ) => Effect.Effect, ProjectionStoreV2Error>; @@ -4636,7 +4649,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)}) @@ -4728,6 +4744,48 @@ 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 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 + 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 ( + ${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 + 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, @@ -9128,6 +9186,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 c3d1096cb3c93..0c8a668f7ab1f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -212,6 +212,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 3992d46e6bf9a..0d1b3392243aa 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 96add5d1fc48a..364d069958fde 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -11151,6 +11151,12 @@ 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", + "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; @@ -11329,8 +11335,12 @@ it.layer(layerTest)("usage-limit recovery", (it) => { ); 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 = { @@ -11406,6 +11416,135 @@ it.layer(layerTest)("usage-limit recovery", (it) => { }); if (snooze) assert.equal(DateTime.toEpochMillis(armedShell.snoozedUntil!), Date.parse(resetAt)); + if (scenario.startsWith("early-reset")) { + const manualWakeAt = DateTime.add(now, { seconds: 30 }); + // 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}`), + threadId, + snoozedUntil: DateTime.formatIso(manualWakeAt), + }); + const waiting = yield* orchestrator.getThreadProjection(threadId); + 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({ + ...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, or one read before a later + // recovery choice. + 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 })), + 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( + (yield* orchestrator.getThreadProjection(threadId)).thread, + waiting.thread, + ); + yield* orchestrator.recordLimitReset({ + ...instances, + failedBefore: now, + resetBefore: DateTime.add(DateTime.makeUnsafe(resetAt), { seconds: 1 }), + }); + const reset = (yield* orchestrator.getThreadProjection(threadId)).thread; + 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 (manualSnooze) + 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)); + }); + 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.stop", + commandId: CommandId.make(`recovery:stop:${scenario}`), + threadId, + }); + 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)); + yield* orchestrator.dispatch(resume); + // 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); + 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/server.ts b/apps/server/src/server.ts index 30f003bf12891..91a1258dfa100 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -122,6 +122,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"; @@ -713,6 +714,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..490b173433af0 --- /dev/null +++ b/apps/server/src/usage/ResetCredits.test.ts @@ -0,0 +1,233 @@ +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]; +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; + +/** The service over fake providers whose redemption takes five seconds. */ +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"]; +}) => { + 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) => + input.warning === undefined ? { outcome } : { outcome, warning: input.warning }, + ), + ), + 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 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" }, + { 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: "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. + { 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, + ...redemption + }: { + readonly on: "instance" | "hub"; + readonly outcome: ProviderConsumeResetCreditOutcome; + readonly limits?: LimitsAfter; + readonly warning?: string; + readonly ends: "all" | "beforeWeeklyReset" | "none"; + readonly recordFails?: boolean; + }) => + Effect.gen(function* () { + const resets = yield* Queue.unbounded(); + 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, + redemption.warning === undefined + ? { outcome: redemption.outcome } + : { outcome: redemption.outcome, warning: redemption.warning }, + ); + 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 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), + }), + ), + ); + }), +); diff --git a/apps/server/src/usage/ResetCredits.ts b/apps/server/src/usage/ResetCredits.ts new file mode 100644 index 0000000000000..895d863161938 --- /dev/null +++ b/apps/server/src/usage/ResetCredits.ts @@ -0,0 +1,132 @@ +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 Fiber from "effect/Fiber"; +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 scope = yield* Effect.scope; + + 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. + 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. + 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; + return yield* consume(input).pipe( + // A warning means the hub may still refuse until its cooldown ends. + Effect.tap(({ result, redeemedOn }) => + 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 + // not leave the threads waiting, so only service shutdown ends this. + Effect.forkIn(scope), + Effect.flatMap(Fiber.join), + ); + }, + ); + + 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 660548c1dd68b..aeeac066b7481 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -124,6 +124,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"; @@ -847,6 +848,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)({}), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 09c6d86c050fd..bab39e3f7b28f 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -279,6 +279,7 @@ import { } 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"; @@ -1499,6 +1500,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 = @@ -2576,38 +2578,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) { - if ("sourceId" in input) return 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, - }), - ), - ); - 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 4c02f1c79baee..80b1c6940f0f4 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -269,6 +269,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 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 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 44e9aa8290e9a..3c6df4b634659 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -419,6 +419,8 @@ export const OrchestrationV2LimitRecovery = Schema.Struct({ resetAt: IsoDateTime, autoResume: Schema.Boolean, snooze: Schema.optional(Schema.Boolean), + /** 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; @@ -3605,6 +3607,16 @@ 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, + /** 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({ type: Schema.Literal("thread.project.adopt-folder"), diff --git a/packages/shared/src/usageLimits.test.ts b/packages/shared/src/usageLimits.test.ts index b66001ca19244..6a563ffc607ee 100644 --- a/packages/shared/src/usageLimits.test.ts +++ b/packages/shared/src/usageLimits.test.ts @@ -23,6 +23,7 @@ import { exhaustedUsageWindow, paceOf, paceRemainingPercent, + resetCreditReach, usesChatGptSharing, resolveNewThreadModelSelection, } from "./usageLimits.ts"; @@ -1426,6 +1427,76 @@ describe("/usage-limits", () => { }); }); + 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 = [ + 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", + }); + 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( + 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", () => { const workspaces = ["ws-a", "ws-b"].map((workspaceId) => provider({ diff --git a/packages/shared/src/usageLimits.ts b/packages/shared/src/usageLimits.ts index 6fe6c70a89a4f..aa92dd55ed60d 100644 --- a/packages/shared/src/usageLimits.ts +++ b/packages/shared/src/usageLimits.ts @@ -816,6 +816,60 @@ export function resetCreditChoices( ]; } +/** + * 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, + 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 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 } : {}), + }; +} + export function formatResetsIn(window: ServerProviderUsageWindow, now: number): string | null { const resetsAt = resetMillis(window); if (resetsAt === null) return null;