diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index dfcf279a0392e..eed0ff4e4c6f5 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -33,6 +33,7 @@ import { type ModelSelection, OrchestrationV2Command, type OrchestrationV2InternalCommand, + type OrchestrationV2LimitRecovery, type OrchestrationV2LimitRecoveryUpdate, type OrchestrationV2Notification, type OrchestrationV2ServerCommand, @@ -1169,6 +1170,32 @@ function queuedRunCancellationEvents( return events; } +function limitRecoveryWaitsForReset( + previous: OrchestrationV2LimitRecovery | null | undefined, + runId: RunId, +): boolean { + return previous?.runId === runId && previous.waitForReset === true; +} + +type LimitRecoverySnooze = Pick; + +function limitRecoveryOwnsSnooze(thread: LimitRecoverySnooze): boolean { + return ( + thread.limitRecovery?.snooze === true && + thread.snoozedUntil != null && + DateTime.toEpochMillis(thread.snoozedUntil) === Date.parse(thread.limitRecovery.resetAt) + ); +} + +// The recovery's own snooze ends when the thread resumes early; a snooze the user set stays. +export function snoozeHoldsLimitRecovery(thread: LimitRecoverySnooze, nowMs: number): boolean { + return ( + thread.snoozedUntil != null && + DateTime.toEpochMillis(thread.snoozedUntil) > nowMs && + !limitRecoveryOwnsSnooze(thread) + ); +} + function latestStableRun( projection: Pick, ): OrchestrationV2Run | null { @@ -4440,7 +4467,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio thread.settledOverride === "settled" || run?.id !== command.limitRecovery.runId || failure?.class !== "usage_limit" || - failure.resetAt !== command.limitRecovery.resetAt || + (failure.resetAt ?? + (thread.limitRecovery?.runId === run.id && thread.limitRecovery.waitForReset === true + ? thread.limitRecovery.resetAt + : null)) !== command.limitRecovery.resetAt || resetMs <= DateTime.toEpochMillis(run.completedAt ?? run.requestedAt) || projection.runtimeRequests.some((request) => request.status === "pending") ) { @@ -4692,6 +4722,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command.limitRecovery.autoResume ?? previousRecovery?.autoResume ?? false, snooze: command.limitRecovery.snooze ?? previousRecovery?.snooze ?? false, requestId: command.commandId, + ...(limitRecoveryWaitsForReset( + thread.limitRecovery, + command.limitRecovery.runId, + ) + ? { waitForReset: true } + : {}), }; return { ...thread, @@ -4705,11 +4741,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // Recovery updates acknowledge the same stopped run; their timestamps must not trigger fresh failure wakes. snoozedAt: now, } - : command.limitRecovery !== undefined && - thread.limitRecovery?.snooze && - thread.snoozedUntil != null && - DateTime.toEpochMillis(thread.snoozedUntil) === - Date.parse(thread.limitRecovery.resetAt) + : command.limitRecovery !== undefined && limitRecoveryOwnsSnooze(thread) ? { snoozedUntil: null, snoozedAt: null, lastSnoozeWakeAt: now } : {}), ...(command.branch === undefined ? {} : { branch: command.branch }), @@ -6682,15 +6714,19 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio !recovery?.autoResume || recovery.requestId !== command.usageLimitRecoveryRequestId || recovery.runId !== run.id || - recovery.resetAt !== failure.resetAt || - Date.parse(recovery.resetAt) > DateTime.toEpochMillis(now) || + recovery.resetAt !== + (failure.resetAt ?? (recovery.waitForReset === true ? recovery.resetAt : null)) || + (Date.parse(recovery.resetAt) > DateTime.toEpochMillis(now) && + (recovery.waitForReset === true || + command.usageLimitClearedAt === undefined || + Date.parse(command.usageLimitClearedAt) <= + DateTime.toEpochMillis(run.completedAt ?? run.requestedAt))) || projection.thread.archivedAt !== null || projection.thread.deletedAt !== null || projection.thread.settledOverride === "settled" || projection.thread.providerInstanceId !== run.providerInstanceId || projection.runtimeRequests.some((request) => request.status === "pending") || - (projection.thread.snoozedUntil != null && - DateTime.toEpochMillis(projection.thread.snoozedUntil) > DateTime.toEpochMillis(now)) + snoozeHoldsLimitRecovery(projection.thread, DateTime.toEpochMillis(now)) ) { yield* emit( events, @@ -6704,14 +6740,36 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(recovery?.autoResume && recovery.requestId === command.usageLimitRecoveryRequestId && recovery.runId === command.usageLimitContinuationOfRunId && - projection.thread.snoozedUntil != null && - DateTime.toEpochMillis(projection.thread.snoozedUntil) > DateTime.toEpochMillis(now) + (projection.runtimeRequests.some((request) => request.status === "pending") || + snoozeHoldsLimitRecovery(projection.thread, DateTime.toEpochMillis(now))) ? { limitRecovery: { ...recovery, requestId: command.commandId } } : {}), }, }); return; } + if (Date.parse(recovery.resetAt) > DateTime.toEpochMillis(now)) { + const event = yield* emit( + events, + command, + )({ + type: "thread.metadata-updated", + threadId: command.threadId, + occurredAt: now, + payload: { + ...projection.thread, + limitRecovery: { + ...recovery, + runId: idAllocator.derive.run({ + threadId: command.threadId, + ordinal: nextRunOrdinal(projection), + }), + waitForReset: true, + }, + }, + }); + projection = applyToProjection(projection, event); + } } if (command.restartContinuationOfRunId !== undefined) { @@ -12374,6 +12432,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio resetAt: failure.resetAt, autoResume: false, snooze: false, + ...(limitRecoveryWaitsForReset(recovery, latest.id) ? { waitForReset: true } : {}), }; } } diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 8b967f648165f..38234dd0f67e4 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -2498,6 +2498,104 @@ it.effect("SQL projection keeps background thread updates out of updatedAt", () backgroundThreadUpdatesKeepTimestamp.pipe(Effect.provide(layerTest)), ); +it.effect("SQL limit recovery loads armed threads before their reset", () => + Effect.gen(function* () { + const store = yield* ProjectionStore.ProjectionStoreV2; + const threadId = yield* addRolledBackRecoveryCandidate("armed-before-reset"); + const projection = yield* store.getThreadProjection(threadId); + const run = projection.runs[0]!; + const now = yield* DateTime.now; + const resetAt = "2099-01-01T00:00:00.000Z"; + yield* store.apply({ + id: EventId.make("event:armed-before-reset:error"), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make("armed-before-reset:error"), + threadId, + runId: run.id, + nodeId: run.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "failed", + title: "Usage limit reached", + startedAt: now, + completedAt: now, + updatedAt: now, + type: "error", + failure: { + class: "usage_limit", + message: "Plan limit reached.", + resetAt, + code: "usageLimitExceeded", + retryable: null, + }, + }, + }); + yield* store.apply({ + id: EventId.make("event:armed-before-reset:failed"), + type: "run.updated", + threadId, + occurredAt: now, + payload: { ...run, status: "failed" }, + }); + const recovery = { + runId: run.id, + resetAt, + autoResume: true, + requestId: CommandId.make("recovery:armed-before-reset"), + }; + const updateThread = (suffix: string, patch: Partial) => + store.apply({ + id: EventId.make(`event:armed-before-reset:${suffix}`), + type: "thread.metadata-updated", + threadId, + occurredAt: now, + payload: { ...projection.thread, limitRecovery: recovery, ...patch }, + }); + const candidate = Effect.map( + store.getLimitRecoveryCandidates({ now, autoResume: false, snooze: false }), + (rows) => rows.find((row) => row.id === threadId), + ); + + yield* updateThread("armed", {}); + assert.deepEqual((yield* candidate)?.limitRecovery, recovery); + const retry = { ...recovery, resetAt: "2099-01-02T00:00:00.000Z", waitForReset: true }; + yield* updateThread("retry", { limitRecovery: retry }); + assert.deepEqual((yield* candidate)?.limitRecovery, retry); + const failure = (yield* store.getThreadProjection(threadId)).turnItems.find( + (item) => item.id === TurnItemId.make("armed-before-reset:error"), + ); + assert(failure?.type === "error"); + yield* store.apply({ + id: EventId.make("event:armed-before-reset:no-reset"), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { ...failure, failure: { ...failure.failure!, resetAt: null } }, + }); + assert.equal((yield* candidate)?.usageLimitResetAt, retry.resetAt); + assert.equal((yield* store.getThreadShell(threadId))?.usageLimitResetAt, retry.resetAt); + assert.equal( + ProjectionStore.threadShellFromProjection(yield* store.getThreadProjection(threadId)) + .usageLimitResetAt, + retry.resetAt, + ); + yield* updateThread("no-reset-unarmed", {}); + assert.isUndefined(yield* candidate); + yield* updateThread("retry-cancelled", { limitRecovery: { ...retry, autoResume: false } }); + assert.isUndefined(yield* candidate); + yield* updateThread("snoozed", { snoozedUntil: DateTime.makeUnsafe(resetAt) }); + assert.isUndefined(yield* candidate); + yield* updateThread("cancelled", { limitRecovery: { ...recovery, autoResume: false } }); + assert.isUndefined(yield* candidate); + }).pipe(Effect.provide(layerTest)), +); + const runlessNativeSubagentNeedsRecovery = Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; const threadId = yield* addRolledBackRecoveryCandidate("runless-native-recovery"); @@ -8105,6 +8203,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { if (lastErrorClass === "usage_limit") { assert.deepEqual(candidate, { id: sqlShell.id, + providerInstanceId: sqlShell.providerInstanceId, status: sqlShell.status, lastErrorClass: sqlShell.lastErrorClass, usageLimitResetAt: sqlShell.usageLimitResetAt, @@ -8283,10 +8382,12 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { yield* sql`UPDATE orchestration_v2_projection_threads SET payload_json = json_set(payload_json, '$.limitRecovery', json(${encodeUnknownJsonString(recovery)})) WHERE thread_id = ${threadId}`; - assert.isUndefined( + // Armed rows load before the reset too, so the sweep can see an allowance restored early. + assert.deepEqual( (yield* store.getLimitRecoveryCandidates({ ...recoveryOptions, autoResume: true })).find( (row) => row.id === threadId, - ), + )?.limitRecovery, + recovery, ); assert.strictEqual(yield* store.getLimitRecoveryWakeAt(now), DateTime.toEpochMillis(reset)); assert.isNull(yield* store.getLimitRecoveryWakeAt(reset)); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 78422b56535af..183ca0b91aa15 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -181,6 +181,7 @@ export type ProjectionRecoveryKind = export type ProjectionLimitRecoveryCandidate = Pick< OrchestrationV2ThreadShell, | "id" + | "providerInstanceId" | "status" | "lastErrorClass" | "latestRunId" @@ -1779,6 +1780,8 @@ export function threadShellFromProjection( ...threadErrorSummary( latestRootProviderFailure(latestRun, projection.turnItems), providerSession?.lastError ?? null, + projection.thread.limitRecovery, + latestRun?.id ?? null, ), pendingRuntimeRequest, pendingGuestMessageCount: pendingGuestMessageCount(projection.thread.guestShare), @@ -4549,9 +4552,16 @@ export const layer: Layer.Layer = readonly completed_at: string | null; readonly failure_payload_json: string; readonly last_error: string | null; + readonly recovery_reset_at: string | null; }>` SELECT t.payload_json, r.run_id, r.completed_at, item.payload_json AS failure_payload_json, + COALESCE(json_extract(item.payload_json, '$.failure.resetAt'), CASE + WHEN json_extract(t.payload_json, '$.limitRecovery.runId') IS r.run_id + AND json_extract(t.payload_json, '$.limitRecovery.autoResume') = 1 + AND json_extract(t.payload_json, '$.limitRecovery.waitForReset') = 1 + THEN json_extract(t.payload_json, '$.limitRecovery.resetAt') + END) AS recovery_reset_at, ( SELECT json_extract(session.payload_json, '$.lastError') -- Keep bindings outermost: one provider instance can hold every session. @@ -4597,27 +4607,34 @@ export const layer: Layer.Layer = AND json_extract(t.payload_json, '$.archivedAt') IS NULL AND json_extract(t.payload_json, '$.settledOverride') IS NOT 'settled' AND json_extract(item.payload_json, '$.failure.class') = 'usage_limit' - AND json_extract(item.payload_json, '$.failure.resetAt') IS NOT NULL - AND julianday(json_extract(item.payload_json, '$.failure.resetAt')) > julianday(COALESCE(r.completed_at, json_extract(t.payload_json, '$.updatedAt'))) + AND recovery_reset_at IS NOT NULL + AND julianday(recovery_reset_at) > julianday(COALESCE(r.completed_at, json_extract(t.payload_json, '$.updatedAt'))) AND ( ( 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.resetAt') IS recovery_reset_at + OR json_extract(t.payload_json, '$.limitRecovery.waitForReset') = 1 + ) 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, '$.snoozedUntil') IS NULL OR julianday(json_extract(t.payload_json, '$.snoozedUntil')) <= julianday(${DateTime.formatIso(options.now)}) + -- snoozeHoldsLimitRecovery: the recovery's own snooze yields to an early resume. + OR ( + json_extract(t.payload_json, '$.limitRecovery.snooze') = 1 + AND julianday(json_extract(t.payload_json, '$.snoozedUntil')) = julianday(json_extract(t.payload_json, '$.limitRecovery.resetAt')) + ) ) ) OR ( ( json_extract(t.payload_json, '$.limitRecovery.runId') IS NOT r.run_id - OR json_extract(t.payload_json, '$.limitRecovery.resetAt') IS NOT json_extract(item.payload_json, '$.failure.resetAt') + OR json_extract(t.payload_json, '$.limitRecovery.resetAt') IS NOT recovery_reset_at ) AND ( ${booleanInt(options.autoResume)} - OR (${booleanInt(options.snooze)} AND julianday(json_extract(item.payload_json, '$.failure.resetAt')) > julianday(${DateTime.formatIso(options.now)})) + OR (${booleanInt(options.snooze)} AND julianday(recovery_reset_at) > julianday(${DateTime.formatIso(options.now)})) ) -- A subagent of a settled, archived or deleted parent is abandoned work. AND NOT EXISTS ( @@ -4649,9 +4666,10 @@ export const layer: Layer.Layer = if (summary.lastErrorClass !== "usage_limit") continue; candidates.push({ id: thread.id, + providerInstanceId: thread.providerInstanceId, status: "failed", lastErrorClass: summary.lastErrorClass, - usageLimitResetAt: summary.usageLimitResetAt, + usageLimitResetAt: row.recovery_reset_at, latestRunId: RunId.make(row.run_id), latestRunCompletedAt: row.completed_at === null ? null : DateTime.makeUnsafe(row.completed_at), @@ -8409,6 +8427,8 @@ export const layer: Layer.Layer = ...threadErrorSummary( terminalFailureItem?.type === "error" ? terminalFailureItem.failure : null, row.last_error, + thread.limitRecovery, + latestRunId, ), pendingRuntimeRequest, latestUserMessageAt: diff --git a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts index 430018adcb06a..321e5c775bf88 100644 --- a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.test.ts @@ -1,5 +1,10 @@ import { assert, describe, it } from "@effect/vitest"; -import { RunId, ThreadId, type OrchestrationV2DomainEvent } from "@t3tools/contracts"; +import { + ProviderInstanceId, + RunId, + ThreadId, + type OrchestrationV2DomainEvent, +} from "@t3tools/contracts"; import * as Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -8,6 +13,7 @@ import * as Queue from "effect/Queue"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as Orchestrator from "./Orchestrator.ts"; @@ -21,11 +27,16 @@ it.effect("idle sweeps skip SQL until the next wake time or a relevant event", ( yield* TestClock.setTime(Date.parse("2026-07-01T09:00:00.000Z")); const startMs = Date.parse("2026-07-01T09:00:00.000Z"); const queries = yield* Queue.unbounded(); + const providerChanges = yield* Queue.unbounded(); const events = yield* Queue.unbounded(); const layer = UsageLimitRecoveryWorker.layer.pipe( Layer.provide( Layer.mergeAll( Scheduler.layer, + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed([]), + streamChanges: Stream.fromQueue(providerChanges), + }), Layer.mock(ProjectionStore.ProjectionStoreV2)({ getLimitRecoveryCandidates: () => Clock.currentTimeMillis.pipe( @@ -66,6 +77,9 @@ it.effect("idle sweeps skip SQL until the next wake time or a relevant event", ( yield* TestClock.adjust("5 seconds"); // A streaming item at 90s does not wake it; the failed run does on the 95s tick. assert.equal(yield* Queue.take(queries), 95_000); + yield* Queue.offer(providerChanges, []); + yield* TestClock.adjust("5 seconds"); + assert.equal(yield* Queue.take(queries), 100_000); }).pipe(Effect.provide(layer)); }), ); @@ -81,6 +95,7 @@ const failedThread = (failedAgoMs: number, resetAgoMs: number) => { lastErrorClass: "usage_limit", usageLimitResetAt: DateTime.formatIso(DateTime.makeUnsafe(nowMs - resetAgoMs)), latestRunId: RunId.make("run:limited"), + providerInstanceId: ProviderInstanceId.make("codex"), latestRunCompletedAt: failedAt, updatedAt: failedAt, archivedAt: null, diff --git a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts index 3992d46e6bf9a..c7fd9f4ae9ffa 100644 --- a/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts +++ b/apps/server/src/orchestration-v2/UsageLimitRecoveryWorker.ts @@ -1,9 +1,15 @@ -import { CommandId, MessageId, type OrchestrationV2Command } from "@t3tools/contracts"; +import { + CommandId, + MessageId, + type OrchestrationV2Command, + type ServerProviderUsageLimits, +} from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; import type { OrchestrationV2DomainEvent } from "@t3tools/contracts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; import * as DueHint from "../scheduling/DueHint.ts"; import * as Scheduler from "../scheduling/Scheduler.ts"; import { forkParked } from "../serverActivation.ts"; @@ -19,6 +25,7 @@ export function limitRecoveryCommand( autoResume: boolean, nowMs: number, snooze = false, + usageLimits?: ServerProviderUsageLimits, ): OrchestrationV2Command | null { if ( thread.status !== "failed" || @@ -31,16 +38,17 @@ export function limitRecoveryCommand( ) return null; const resetMs = Date.parse(thread.usageLimitResetAt); + const failedAtMs = DateTime.toEpochMillis(thread.latestRunCompletedAt ?? thread.updatedAt); // An already-expired window reported with a fresh failure cannot start a retry loop. - if ( - !Number.isFinite(resetMs) || - resetMs <= DateTime.toEpochMillis(thread.latestRunCompletedAt ?? thread.updatedAt) - ) - return null; + if (!Number.isFinite(resetMs) || resetMs <= failedAtMs) return null; const identity = `${thread.id}:${thread.latestRunId}:${resetMs}`; const recovery = thread.limitRecovery; if (recovery?.runId !== thread.latestRunId || recovery.resetAt !== thread.usageLimitResetAt) { if (nowMs - resetMs > MAX_OVERDUE_ARM_MS) return null; + autoResume ||= + recovery?.runId === thread.latestRunId && + recovery.autoResume && + recovery.waitForReset === true; if (!autoResume && (!snooze || resetMs <= nowMs)) return null; return { type: "thread.metadata.update", @@ -54,13 +62,29 @@ export function limitRecoveryCommand( }, }; } + // A reset credit or provider refresh can restore the allowance before the + // reported reset. Only a reading taken after the failure counts. A reading can + // omit the limit that stopped the run, so once an early run hits the limit + // again the server sets waitForReset and the thread waits for the reset. + const clearedAt = + resetMs > nowMs && + recovery.waitForReset !== true && + usageLimits !== undefined && + usageLimits.unavailable === undefined && + usageLimits.windows.length > 0 && + usageLimits.windows.every((window) => window.usedPercent < 100) && + Date.parse(usageLimits.checkedAt) > failedAtMs + ? usageLimits.checkedAt + : undefined; if ( !recovery.autoResume || - resetMs > nowMs || - (thread.snoozedUntil != null && DateTime.toEpochMillis(thread.snoozedUntil) > nowMs) + (resetMs > nowMs && clearedAt === undefined) || + Orchestrator.snoozeHoldsLimitRecovery(thread, nowMs) ) return null; - const deliveryIdentity = `${identity}:${recovery.requestId ?? "legacy"}`; + // A separate id for early delivery keeps a rejected one from claiming the + // receipt of the delivery at the reset. + const deliveryIdentity = `${identity}:${recovery.requestId ?? "legacy"}${clearedAt === undefined ? "" : ":early"}`; return { type: "message.dispatch", commandId: CommandId.make(`limit-resume:${deliveryIdentity}`), @@ -70,6 +94,7 @@ export function limitRecoveryCommand( ...(recovery.requestId === undefined ? {} : { usageLimitRecoveryRequestId: recovery.requestId }), + ...(clearedAt === undefined ? {} : { usageLimitClearedAt: clearedAt }), text: "Continue where you left off.", attachments: [], dispatchMode: { type: "start_immediately" }, @@ -92,6 +117,7 @@ const makeSweep = Effect.gen(function* () { const threads = yield* ThreadManagement.ThreadManagementService; const orchestrator = yield* Orchestrator.OrchestratorV2; const settings = yield* ServerSettings.ServerSettingsService; + const providerRegistry = yield* ProviderRegistry.ProviderRegistry; const due = DueHint.make(); yield* forkParked( Stream.runForEach(orchestrator.streamDomainEvents, (event) => @@ -99,6 +125,7 @@ const makeSweep = Effect.gen(function* () { ), ); yield* forkParked(Stream.runForEach(settings.streamChanges, () => due.invalidate)); + yield* forkParked(Stream.runForEach(providerRegistry.streamChanges, () => due.invalidate)); return Effect.fn("UsageLimitRecoveryWorker.sweep")(function* () { const now = yield* DateTime.now; const nowMs = DateTime.toEpochMillis(now); @@ -110,6 +137,7 @@ const makeSweep = Effect.gen(function* () { autoResume: preferences.autoResumeLimitedThreads, snooze: preferences.snoozeLimitedThreads, }); + const providers = candidates.length === 0 ? [] : yield* providerRegistry.getProviders; let acted = false; for (const thread of candidates) { const command = limitRecoveryCommand( @@ -117,6 +145,8 @@ const makeSweep = Effect.gen(function* () { preferences.autoResumeLimitedThreads, nowMs, preferences.snoozeLimitedThreads, + providers.find((provider) => provider.instanceId === thread.providerInstanceId) + ?.usageLimits, ); if (command === null) continue; acted = true; diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index c7cc43ee5ddea..0d63d182d2210 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -1,6 +1,8 @@ import { PullRequestProviderError } from "@t3tools/source-control-core/server/PullRequestProvider"; import { SourceControlProviderKind } from "@t3tools/contracts"; import { limitRecoveryCommand } from "./UsageLimitRecoveryWorker.ts"; +import * as UsageLimitRecoveryWorker from "./UsageLimitRecoveryWorker.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; import { createOrchestrationV2TurnItemVisibility } from "@t3tools/shared/orchestrationV2Timeline"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as SourceControlProviderRegistry from "../sourceControl/SourceControlProviderRegistry.ts"; @@ -22,6 +24,7 @@ import { TurnItemId, type ModelSelection, type OrchestrationV2Run, + type ServerProviderUsageLimits, type OrchestrationV2ShellSnapshot, type OrchestrationV2RuntimeRequest, ProjectId, @@ -10822,6 +10825,341 @@ it.layer(layerTest)("usage-limit recovery", (it) => { }), ); + it.effect.each([true, false])( + "preserves early recovery with global auto-resume off when retry reports reset: %s", + (retryReportsReset) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const events = yield* EventSink.EventSinkV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threadId = ThreadId.make(`recovery:early-reset:${retryReportsReset}`); + const projectId = ProjectId.make(`recovery:project:early-reset:${retryReportsReset}`); + const now = yield* DateTime.now; + yield* seedProject({ + projectId, + title: "Recovery project", + workspaceRoot: process.cwd(), + defaultModelSelection: modelSelection, + createdAt: DateTime.formatIso(now), + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`recovery:create:early-reset:${retryReportsReset}`), + threadId, + projectId, + title: "Limited thread", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`recovery:message:early-reset:${retryReportsReset}`), + threadId, + messageId: MessageId.make(`recovery:message:early-reset:${retryReportsReset}`), + text: "Work on this.", + attachments: [], + dispatchMode: { type: "defer_start" }, + createdBy: "user", + creationSource: "web", + }); + const run = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + const resetAt = DateTime.formatIso(DateTime.add(now, { hours: 3 })); + yield* events.write({ + events: [ + { + id: EventId.make(`recovery:run:early-reset:${retryReportsReset}`), + type: "run.updated", + threadId, + occurredAt: now, + payload: { ...run, status: "failed", completedAt: now }, + }, + { + id: EventId.make(`recovery:error:early-reset:${retryReportsReset}`), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make(`recovery:error:early-reset:${retryReportsReset}`), + type: "error", + threadId, + runId: run.id, + nodeId: run.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "failed", + title: "Usage limit reached", + startedAt: now, + completedAt: now, + updatedAt: now, + failure: { + class: "usage_limit", + message: "Plan limit reached.", + code: "usageLimitExceeded", + retryable: null, + resetAt, + }, + }, + }, + ], + }); + const shell = (yield* orchestrator.getShellSnapshot()).threads.find( + (thread) => thread.id === threadId, + )!; + // The case without a retry reset also arms with a snooze until the reset. + const arm = limitRecoveryCommand( + shell, + true, + DateTime.toEpochMillis(now), + !retryReportsReset, + ); + assert(arm?.type === "thread.metadata.update" && arm.limitRecovery != null); + yield* orchestrator.autoArmLimitRecovery({ + commandId: arm.commandId, + threadId, + limitRecovery: arm.limitRecovery, + }); + yield* TestClock.adjust("1 minute"); + const checkedNow = yield* DateTime.now; + const reading = (checkedAt: DateTime.Utc, usedPercent: number) => ({ + checkedAt: DateTime.formatIso(checkedAt), + windows: [{ id: "five_hour", kind: "session" as const, label: "5h", usedPercent }], + }); + const sweep = ( + usageLimits: ServerProviderUsageLimits, + instanceId = shell.providerInstanceId, + ) => + Effect.scoped( + Layer.build( + UsageLimitRecoveryWorker.layer.pipe( + Layer.provide([ + Layer.succeed(ProjectionStore.ProjectionStoreV2, projections), + Layer.succeed(Orchestrator.OrchestratorV2, orchestrator), + Layer.mock(ThreadManagementService.ThreadManagementService)({ + dispatch: orchestrator.dispatch, + }), + ServerSettings.layerTest(), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + streamChanges: Stream.empty, + getProviders: Effect.succeed([ + { + instanceId, + driver, + enabled: true, + installed: true, + version: null, + status: "ready", + auth: { status: "authenticated" }, + checkedAt: usageLimits.checkedAt, + models: [], + slashCommands: [], + skills: [], + usageLimits, + }, + ]), + }), + Layer.mock(Scheduler.Scheduler)({ + register: (_name, work) => Effect.orDie(work), + }), + ]), + ), + ), + ); + const armed = (yield* projections.getLimitRecoveryCandidates({ + now: checkedNow, + autoResume: true, + snooze: false, + })).find((row) => row.id === threadId); + assert.isDefined(armed); + const nowMs = DateTime.toEpochMillis(checkedNow); + // Readings from before the failure, or ones that still show the limit, keep waiting. + assert.isNull(limitRecoveryCommand(armed!, true, nowMs, false, reading(now, 0))); + assert.isNull(limitRecoveryCommand(armed!, true, nowMs, false, reading(checkedNow, 100))); + yield* sweep(reading(now, 0)); + yield* sweep(reading(checkedNow, 100)); + yield* sweep(reading(checkedNow, 0), alternateInstanceId); + assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 1); + if (!retryReportsReset) { + // A sidebar snooze holds the thread even when it ends at the reset. + yield* orchestrator.dispatch({ + type: "thread.snooze", + commandId: CommandId.make("recovery:user-snooze:early-reset"), + threadId, + snoozedUntil: resetAt, + }); + yield* sweep(reading(checkedNow, 0)); + assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 1); + yield* orchestrator.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make("recovery:resnooze:early-reset"), + threadId, + limitRecovery: { runId: run.id, resetAt, snooze: true }, + }); + } + const staleResume = limitRecoveryCommand( + armed!, + true, + nowMs, + false, + reading(checkedNow, 0), + ); + assert(staleResume?.type === "message.dispatch"); + yield* orchestrator.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(`recovery:cancel:early-reset:${retryReportsReset}`), + threadId, + limitRecovery: { runId: run.id, resetAt, autoResume: false }, + }); + yield* orchestrator.dispatch(staleResume); + yield* sweep(reading(checkedNow, 0)); + assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 1); + yield* orchestrator.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make(`recovery:rearm:early-reset:${retryReportsReset}`), + threadId, + limitRecovery: { runId: run.id, resetAt, autoResume: true }, + }); + const ready = (yield* projections.getLimitRecoveryCandidates({ + now: checkedNow, + autoResume: true, + snooze: false, + })).find((row) => row.id === threadId)!; + const racedResume = limitRecoveryCommand(ready, true, nowMs, false, reading(checkedNow, 0)); + assert(racedResume?.type === "message.dispatch"); + const nodeId = run.rootNodeId; + assert(nodeId !== null); + const request = { + id: RuntimeRequestId.make(`recovery:pending-race:${retryReportsReset}`), + nodeId, + providerTurnId: null, + nativeRequestRef: null, + kind: "user_input" as const, + status: "pending" as const, + responseCapability: { type: "message" as const }, + createdAt: checkedNow, + resolvedAt: null, + }; + yield* events.write({ + events: [ + { + id: EventId.make(`recovery:pending-race:${retryReportsReset}`), + type: "runtime-request.updated", + threadId, + nodeId, + occurredAt: checkedNow, + payload: request, + }, + ], + }); + yield* orchestrator.dispatch(racedResume); + assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 1); + yield* events.write({ + events: [ + { + id: EventId.make(`recovery:pending-race:resolved:${retryReportsReset}`), + type: "runtime-request.updated", + threadId, + nodeId, + occurredAt: checkedNow, + payload: { ...request, status: "resolved", resolvedAt: checkedNow }, + }, + ], + }); + if (!retryReportsReset) { + const snoozedUntil = (yield* orchestrator.getThreadProjection(threadId)).thread + .snoozedUntil; + assert.equal(DateTime.toEpochMillis(snoozedUntil!), Date.parse(resetAt)); + } + yield* sweep(reading(checkedNow, 0)); + const resumed = yield* orchestrator.getThreadProjection(threadId); + assert.lengthOf(resumed.runs, 2); + assert.equal(resumed.runs[1]?.status, "starting"); + assert.isNull(resumed.thread.snoozedUntil); + + // The reading can omit the limit that stopped the run, such as a + // per-model weekly limit. The early run then fails on the same limit. + const earlyRun = resumed.runs[1]!; + const retryResetAt = retryReportsReset + ? DateTime.formatIso(DateTime.add(now, { hours: 4 })) + : resetAt; + yield* events.write({ + events: [ + { + id: EventId.make(`recovery:run:early-reset:again:${retryReportsReset}`), + type: "run.updated", + threadId, + occurredAt: checkedNow, + payload: { ...earlyRun, status: "failed", completedAt: checkedNow }, + }, + { + id: EventId.make(`recovery:error:early-reset:again:${retryReportsReset}`), + type: "turn-item.updated", + threadId, + occurredAt: checkedNow, + payload: { + id: TurnItemId.make(`recovery:error:early-reset:again:${retryReportsReset}`), + type: "error", + threadId, + runId: earlyRun.id, + nodeId: earlyRun.rootNodeId, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "failed", + title: "Usage limit reached", + startedAt: checkedNow, + completedAt: checkedNow, + updatedAt: checkedNow, + failure: { + class: "usage_limit", + message: "Plan limit reached.", + code: "usageLimitExceeded", + retryable: null, + resetAt: retryReportsReset ? retryResetAt : null, + }, + }, + }, + ], + }); + yield* TestClock.adjust("1 minute"); + const laterNow = yield* DateTime.now; + yield* sweep(reading(laterNow, 0)); + const rearmed = (yield* projections.getLimitRecoveryCandidates({ + now: laterNow, + autoResume: false, + snooze: false, + })).find((row) => row.id === threadId)!; + assert.equal(rearmed.limitRecovery?.waitForReset, true); + assert.equal(rearmed.limitRecovery?.resetAt, retryResetAt); + yield* sweep(reading(laterNow, 0)); + assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 2); + assert.isNull( + limitRecoveryCommand( + rearmed, + true, + DateTime.toEpochMillis(laterNow), + false, + reading(laterNow, 0), + ), + ); + const onTime = limitRecoveryCommand(rearmed, false, Date.parse(retryResetAt)); + assert(onTime?.type === "message.dispatch"); + assert.isUndefined(onTime.usageLimitClearedAt); + yield* TestClock.setTime(Date.parse(retryResetAt)); + yield* orchestrator.dispatch(onTime); + assert.lengthOf((yield* orchestrator.getThreadProjection(threadId)).runs, 3); + }), + ); + it.effect.each([ "resume", "resume-started", @@ -10840,6 +11178,7 @@ it.layer(layerTest)("usage-limit recovery", (it) => { "stop", "stop-before-arm", "stop-before-arm-mismatched-recovery", + "unrelated-recovery", ] as const)("guards a scheduled usage-limit continuation against %s", (scenario) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -10995,6 +11334,28 @@ it.layer(layerTest)("usage-limit recovery", (it) => { threadId, limitRecovery: arm.limitRecovery, }; + if (scenario === "unrelated-recovery") { + const current = yield* orchestrator.getThreadProjection(threadId); + yield* events.write({ + events: [ + { + id: EventId.make("recovery:unrelated-window"), + type: "thread.metadata-updated", + threadId, + occurredAt: now, + payload: { + ...current.thread, + limitRecovery: { + runId: RunId.make("recovery:old-provider-run"), + resetAt: DateTime.formatIso(DateTime.add(now, { hours: 4 })), + autoResume: true, + waitForReset: true, + }, + }, + }, + ], + }); + } const stopBeforeArm = scenario === "stop-before-arm" || scenario === "stop-before-arm-mismatched-recovery"; if (stopBeforeArm) { @@ -11399,6 +11760,7 @@ it.layer(layerTest)("usage-limit recovery", (it) => { after.runs, before.runs.length + (scenario === "resume" || + scenario === "unrelated-recovery" || scenario === "resume-started" || scenario === "resume-untagged" || stopBeforeArm || @@ -11412,6 +11774,7 @@ it.layer(layerTest)("usage-limit recovery", (it) => { after.messages, before.messages.length + (scenario === "resume" || + scenario === "unrelated-recovery" || scenario === "resume-started" || scenario === "resume-untagged" || stopBeforeArm || diff --git a/apps/server/src/provider/ClaudeCapabilitiesProbe.test.ts b/apps/server/src/provider/ClaudeCapabilitiesProbe.test.ts index e2622bc9c2a65..e4b4844e02681 100644 --- a/apps/server/src/provider/ClaudeCapabilitiesProbe.test.ts +++ b/apps/server/src/provider/ClaudeCapabilitiesProbe.test.ts @@ -1,12 +1,16 @@ -// @effect-diagnostics nodeBuiltinImport:off - cleanup uses Node's retrying rm, which the FileSystem service does not expose. +// @effect-diagnostics nodeBuiltinImport:off - an SDK mock rewrites a cache file while the probe awaits it, outside any Effect. +import * as NodeFS from "node:fs"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { ClaudeSettings, type ServerProvider } from "@t3tools/contracts"; import { vi } from "vite-plus/test"; import * as ClaudeSdk from "@anthropic-ai/claude-agent-sdk"; import { assert, it } from "@effect/vitest"; import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; +import * as PlatformError from "effect/PlatformError"; import * as Schema from "effect/Schema"; import * as TestClock from "effect/testing/TestClock"; @@ -74,6 +78,103 @@ it.effect("reports a deleted worktree's scan as finished, not pending", () => }).pipe(Effect.provide(NodeServices.layer)), ); +const cachedReading = { five_hour: { utilization: 2, resets_at: "2026-11-12T01:00:00.000Z" } }; +const cacheEntry = (fetchedAtMs: number, utilization: unknown) => + JSON.stringify({ cachedUsageUtilization: { fetchedAtMs, utilization } }); + +// The clock reads 00:10 when the probe starts; Claude Code fetched the cached reading at 00:01. +it.effect.each([ + { name: "the cache entry it answered from", layout: "configDir", expected: "01" }, + { name: "no cache entry at Claude Code's path", layout: "none", expected: "10" }, + { + name: "a cache another session replaced during the call", + layout: "replaced", + expected: "01", + }, + { name: "a config dir relative to the probe's cwd", layout: "relative", expected: "01" }, + { name: "the HOME the probe runs with", layout: "home", expected: "01" }, + { name: "nothing when the cache cannot be read", layout: "unreadable", expected: undefined }, + { name: "nothing when the cache read does not answer", layout: "stalled", expected: undefined }, +] as const)("dates a Claude usage reading by $name", ({ layout, expected }) => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const root = yield* fileSystem.makeTempDirectoryScoped(); + const cacheFile = + layout === "home" ? `${root}/home/.claude.json` : `${root}/claude-config/.claude.json`; + yield* fileSystem.makeDirectory(`${root}/home`); + yield* fileSystem.makeDirectory(`${root}/claude-config`); + yield* fileSystem.makeDirectory(`${root}/host-home`); + if (layout !== "none") + yield* fileSystem.writeFileString(cacheFile, cacheEntry(60_000, cachedReading)); + const query = vi.spyOn(ClaudeSdk, "query").mockImplementation( + () => + ({ + initializationResult: async () => ({ + account: { subscriptionType: "max", tokenSource: "oauth" }, + commands: [], + }), + usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET: async () => { + if (layout === "replaced") { + NodeFS.writeFileSync( + cacheFile, + cacheEntry(540_000, { five_hour: { utilization: 100 } }), + ); + } + return { rate_limits_available: true, rate_limits: cachedReading }; + }, + }) as unknown as ReturnType, + ); + yield* Effect.addFinalizer(() => Effect.sync(() => query.mockRestore())); + const { PATH } = yield* HostProcess.Environment; + const environment = + layout === "home" + ? { PATH, HOME: `${root}/home` } + : { + PATH, + CLAUDE_CONFIG_DIR: layout === "relative" ? "claude-config" : `${root}/claude-config`, + }; + yield* TestClock.adjust("10 minutes"); + const stallStarted = yield* Deferred.make(); + let stalls = layout === "stalled" ? 1 : 0; + const readFileString: FileSystem.FileSystem["readFileString"] = (path, encoding) => { + if (path === cacheFile && layout === "unreadable") { + return Effect.fail( + PlatformError.systemError({ + _tag: "PermissionDenied", + module: "FileSystem", + method: "readFileString", + pathOrDescriptor: path, + }), + ); + } + if (path === cacheFile && stalls > 0) { + stalls -= 1; + return Deferred.succeed(stallStarted, undefined).pipe(Effect.andThen(Effect.never)); + } + return fileSystem.readFileString(path, encoding); + }; + const probe = probeClaudeCapabilities( + yield* decodeClaudeSettings({ binaryPath: "claude" }), + environment, + layout === "relative" ? root : undefined, + ).pipe( + Effect.provideService(HostProcess.HomeDirectory, `${root}/host-home`), + Effect.provideService(FileSystem.FileSystem, { ...fileSystem, readFileString }), + ); + const fiber = yield* Effect.forkChild(probe); + if (layout === "stalled") { + yield* Deferred.await(stallStarted); + yield* TestClock.adjust("2 seconds"); + } + const capabilities = yield* Fiber.join(fiber); + assert.equal( + capabilities.usageCheckedAt, + expected === undefined ? undefined : `1970-01-01T00:${expected}:00.000Z`, + ); + if (expected === undefined) assert.equal(capabilities.usage, undefined); + }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), +); + it.effect("keeps the probe subprocess alive while gateway model discovery can finish", () => Effect.gen(function* () { let abortSignal: AbortSignal | undefined; diff --git a/apps/server/src/provider/ClaudeProvider.ts b/apps/server/src/provider/ClaudeProvider.ts index 0fb7ea8b2e1fd..76f4d68107de0 100644 --- a/apps/server/src/provider/ClaudeProvider.ts +++ b/apps/server/src/provider/ClaudeProvider.ts @@ -469,6 +469,36 @@ class ClaudeCapabilitiesProbeSdkError extends Schema.TaggedError + error.reason._tag === "NotFound" ? Effect.succeed("{}") : Effect.fail(error), + }), + ); + const entry = (yield* decodeClaudeUsageCacheFile(text)).cachedUsageUtilization; + if (entry === undefined) return undefined; + const fetchedAt = DateTime.make(entry.fetchedAtMs); + return Option.isSome(fetchedAt) + ? { fetchedAt: DateTime.formatIso(fetchedAt.value), utilization: entry.utilization } + : ("unreadable" as const); + }, + Effect.timeout("2 seconds"), + Effect.orElseSucceed(() => "unreadable" as const), +); + /** A never-yielding prompt allows local initialization IPC without sending an Anthropic request. */ const probeClaudeCapabilities = ( claudeSettings: ClaudeSettings, @@ -487,6 +517,16 @@ const probeClaudeCapabilities = ( claudeEnvironment, ); const managedMcpConfig = yield* hasClaudeManagedMcpConfig(); + const path = yield* Path.Path; + const configDir = claudeEnvironment.CLAUDE_CONFIG_DIR; + const home = + (yield* HostProcess.Platform) === "win32" + ? claudeEnvironment.USERPROFILE + : claudeEnvironment.HOME; + // Where the CLI itself keeps .claude.json: its config dir resolved against its cwd, else its home. + const usageCacheFile = configDir + ? path.join(path.resolve(cwd ?? ".", configDir), ".claude.json") + : path.join(home ?? (yield* HostProcess.HomeDirectory), ".claude.json"); // Workspace probes (no usage) discard models; the machine probe refreshes the shared cache. const graceMs = includeStatus ? claudeGatewayDiscoveryGraceMs({ @@ -513,7 +553,7 @@ const probeClaudeCapabilities = ( }), }); const init = await q.initializationResult(); - return { q, init, graceMs }; + return { q, init, graceMs, usageCacheFile }; }, catch: (cause) => new ClaudeCapabilitiesProbeSdkError({ operation: "initialize", cause }), }); @@ -521,13 +561,14 @@ const probeClaudeCapabilities = ( Effect.timeout( resolveProviderProbeTimeoutMs("claude", CAPABILITIES_PROBE_TIMEOUT_MS, environment), ), - Effect.flatMap(({ q, init, graceMs }) => + Effect.flatMap(({ q, init, graceMs, usageCacheFile }) => Effect.gen(function* () { // Usage has its own deadline so a slow optional request cannot discard initialization. // Only the rate limits are read, so skip the local transcript scan that fills // `behaviors`: with a few GB of transcripts it outlasts the deadline. // The effort caps are read in parallel under the same deadline; workspace probes skip both. const usageCheckedAt = DateTime.formatIso(yield* DateTime.now); + const cacheBefore = includeStatus ? yield* readClaudeUsageCache(usageCacheFile) : undefined; const { getSettings } = q as ClaudeSettingsQuery; const [usageResult, settingsResult] = yield* Effect.all( [ @@ -551,7 +592,7 @@ const probeClaudeCapabilities = ( ], { concurrency: "unbounded" }, ); - const usage = + const usageResponse = usageResult && Result.isSuccess(usageResult) ? { rate_limits_available: usageResult.success.rate_limits_available, @@ -561,6 +602,19 @@ const probeClaudeCapabilities = ( const effortCapSources = parseClaudeEffortCapSources( settingsResult && Result.isSuccess(settingsResult) ? settingsResult.success : undefined, ); + const cacheAfter = usageResponse ? yield* readClaudeUsageCache(usageCacheFile) : undefined; + // An unreadable cache cannot date the reading, so this probe reports none. + const usage = + cacheBefore === "unreadable" || cacheAfter === "unreadable" ? undefined : usageResponse; + // Date the reading by the cache entry it equals. Without a match it is no older than the + // entry cached before the call. No entry at the CLI's path means the CLI fetched during this + // call, so the probe's start is a lower bound; a CLI caching elsewhere needs that path here. + const servedFrom = + [cacheAfter, cacheBefore] + .filter((entry) => typeof entry === "object") + .find( + (entry) => JSON.stringify(entry.utilization) === JSON.stringify(usage?.rate_limits), + ) ?? (typeof cacheBefore === "object" ? cacheBefore : undefined); const account = init.account as | { readonly email?: string; @@ -578,7 +632,7 @@ const probeClaudeCapabilities = ( apiProvider: account?.apiProvider, slashCommands: parseClaudeInitializationCommands(init.commands), ...(init.models?.length ? { models: init.models } : {}), - ...(usage ? { usage, usageCheckedAt } : {}), + ...(usage ? { usage, usageCheckedAt: servedFrom?.fetchedAt ?? usageCheckedAt } : {}), ...(effortCapSources ? { effortCapSources } : {}), } satisfies ClaudeCapabilitiesProbe; }).pipe( diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index e5fe950e140a2..6ae1bc2a28b5c 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -258,6 +258,9 @@ the continuation finishes. If the queue was held by a restart, resume it then. When the provider reports a reset time, choose **Resume at reset** to schedule a continuation. You can cancel it from the thread. Enable **Resume after usage limits** in **Settings → Threads** to schedule limit stops by default. +If the provider's usage shows the allowance back before the reset, for example +after you redeem a reset credit, the continuation starts then. A thread that hits +the limit again before the reset waits for the reset. The environment must be running when the reset arrives; it resumes overdue continuations after a restart. Stopping the thread, sending a new message, archiving, or settling it prevents a pending continuation from starting. diff --git a/packages/client-runtime/src/state/threadExecution.test.ts b/packages/client-runtime/src/state/threadExecution.test.ts index ea6de929b4cb7..ef4ec08b89da8 100644 --- a/packages/client-runtime/src/state/threadExecution.test.ts +++ b/packages/client-runtime/src/state/threadExecution.test.ts @@ -181,6 +181,31 @@ describe("thread execution presentation", () => { lastErrorClass: "usage_limit", }); expect(deriveResumableRunId(projection, deriveThreadRuntime(projection))).toBe(failed.id); + const resetAt = "2026-07-28T12:00:00.000Z"; + const recovery = { runId: failed.id, resetAt, autoResume: true, waitForReset: true }; + const recovered = { ...projection, thread: { ...projection.thread, limitRecovery: recovery } }; + expect(deriveThreadRuntime(recovered)?.usageLimitResetAt).toBe(resetAt); + expect( + deriveThreadRuntime({ + ...recovered, + turnItems: [{ ...item, failure: { ...item.failure, resetAt: "2026-07-28T13:00:00.000Z" } }], + })?.usageLimitResetAt, + ).toBe("2026-07-28T13:00:00.000Z"); + expect( + deriveThreadRuntime({ + ...recovered, + thread: { + ...recovered.thread, + limitRecovery: { ...recovery, runId: RunId.make("other-run") }, + }, + })?.usageLimitResetAt, + ).toBeNull(); + expect( + deriveThreadRuntime({ + ...recovered, + thread: { ...recovered.thread, limitRecovery: { ...recovery, waitForReset: false } }, + })?.usageLimitResetAt, + ).toBeNull(); expect( deriveThreadRuntime({ ...projection, diff --git a/packages/client-runtime/src/state/threadExecution.ts b/packages/client-runtime/src/state/threadExecution.ts index 622db98a2903a..10230acb01916 100644 --- a/packages/client-runtime/src/state/threadExecution.ts +++ b/packages/client-runtime/src/state/threadExecution.ts @@ -385,6 +385,8 @@ export function deriveThreadStructureRuntime( ...threadErrorSummary( latestRootProviderFailure(latestRunProjection, projection.turnItems), providerSession?.lastError ?? null, + projection.thread.limitRecovery, + latestRunProjection?.id ?? null, ), }; } diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 69338865e55a6..1873d00e11531 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), + /** Set by the server when this run hit the limit again before the previous reset. */ + waitForReset: Schema.optional(Schema.Boolean), }); export type OrchestrationV2LimitRecovery = typeof OrchestrationV2LimitRecovery.Type; @@ -3331,6 +3333,8 @@ export const OrchestrationV2Command = Schema.Union([ steerFollowUpOfRunId: Schema.optional(RunId), steerFollowUpAfterRunOrdinal: Schema.optional(NonNegativeInt), usageLimitRecoveryRequestId: Schema.optional(CommandId), + /** A usage reading after the failure showed the limits cleared before the reset. */ + usageLimitClearedAt: Schema.optional(IsoDateTime), deliveryIntent: Schema.optional(Schema.Literals(["auto", "steer", "restart"])), delegatedCompletion: Schema.optional( Schema.Struct({ diff --git a/packages/shared/src/orchestrationV2ThreadError.ts b/packages/shared/src/orchestrationV2ThreadError.ts index 141d73f402540..72ef59c4c4a14 100644 --- a/packages/shared/src/orchestrationV2ThreadError.ts +++ b/packages/shared/src/orchestrationV2ThreadError.ts @@ -1,5 +1,7 @@ import type { OrchestrationV2ProviderFailure, + OrchestrationV2LimitRecovery, + RunId, OrchestrationV2Run, OrchestrationV2TurnItem, } from "@t3tools/contracts"; @@ -40,12 +42,19 @@ export function latestRootProviderFailure( export function threadErrorSummary( failure: OrchestrationV2ProviderFailure | null, sessionError: string | null, + recovery?: OrchestrationV2LimitRecovery | null, + runId?: RunId | null, ) { const currentFailure = sessionError !== null && sessionError !== failure?.message ? null : failure; return { usageLimitResetAt: - currentFailure?.class === "usage_limit" ? (currentFailure.resetAt ?? null) : null, + currentFailure?.class === "usage_limit" + ? (currentFailure.resetAt ?? + (recovery != null && recovery.runId === runId && recovery.waitForReset === true + ? recovery.resetAt + : null)) + : null, lastError: sessionError ?? failure?.message ?? null, lastErrorClass: currentFailure?.class ?? null, lastErrorCode: currentFailure?.code ?? null,