diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index fc46e2b100232..7c3903d8151ba 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -15455,13 +15455,111 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); - it.effect.each([true, false])( - "carries a rejected wake rate limit into the continuation failure (event=%s)", - (standaloneEvent) => + it.effect.each( + [ + { standaloneEvent: true, arrivedAfterReset: false, expired: false, newerTurn: false }, + { standaloneEvent: false, arrivedAfterReset: false, expired: false, newerTurn: false }, + { standaloneEvent: true, arrivedAfterReset: false, expired: true, newerTurn: false }, + { standaloneEvent: false, arrivedAfterReset: false, expired: true, newerTurn: false }, + { standaloneEvent: true, arrivedAfterReset: false, expired: true, newerTurn: true }, + { standaloneEvent: true, arrivedAfterReset: false, expired: true, newerRootOpen: true }, + { + standaloneEvent: true, + arrivedAfterReset: false, + expired: true, + newerRootOpen: true, + emptyBeforeNewRoot: true, + }, + { + standaloneEvent: true, + arrivedAfterReset: false, + expired: true, + newerRootOpen: true, + absorbedByUser: true, + }, + { + standaloneEvent: true, + arrivedAfterReset: false, + expired: true, + newerRootOpen: true, + emptyBeforeNewRoot: true, + absorbedByUser: true, + }, + { + standaloneEvent: true, + arrivedAfterReset: false, + expired: true, + newerRootOpen: true, + twoWakes: true, + }, + { standaloneEvent: true, arrivedAfterReset: true, expired: false, newerTurn: false }, + { standaloneEvent: true, arrivedAfterReset: false, expired: true, weeklyStillActive: true }, + { + standaloneEvent: true, + arrivedAfterReset: false, + expired: true, + weeklyStillActive: true, + overageIndicator: "status", + }, + { + standaloneEvent: true, + arrivedAfterReset: false, + expired: true, + weeklyStillActive: true, + overageIndicator: "using", + }, + { + standaloneEvent: true, + arrivedAfterReset: false, + expired: true, + weeklyStillActive: true, + overageIndicator: "inUse", + }, + { + standaloneEvent: false, + arrivedAfterReset: false, + expired: true, + overageIndicator: "status", + }, + { standaloneEvent: true, arrivedAfterReset: false, expired: true, twoWakes: true }, + { + standaloneEvent: true, + arrivedAfterReset: false, + expired: true, + twoWakes: true, + secondWakeStillActive: true, + }, + ].map((row) => ({ + newerTurn: false, + newerRootOpen: false, + emptyBeforeNewRoot: false, + absorbedByUser: false, + weeklyStillActive: false, + twoWakes: false, + secondWakeStillActive: false, + overageIndicator: "none", + ...row, + })), + )( + "replays a rejected wake rate limit only while it is current (event=$standaloneEvent arrivedAfterReset=$arrivedAfterReset expired=$expired newerTurn=$newerTurn rootOpen=$newerRootOpen emptyBeforeRoot=$emptyBeforeNewRoot absorbed=$absorbedByUser weekly=$weeklyStillActive twoWakes=$twoWakes secondActive=$secondWakeStillActive overage=$overageIndicator)", + ({ + standaloneEvent, + arrivedAfterReset, + expired, + newerTurn, + newerRootOpen, + emptyBeforeNewRoot, + absorbedByUser, + weeklyStillActive, + twoWakes, + secondWakeStillActive, + overageIndicator, + }) => Effect.gen(function* () { const harness = yield* makeWakeHarness; const now = yield* DateTime.now; const resetsAt = 1790298600; + const weeklyResetsAt = resetsAt + 604_800; yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ @@ -15476,7 +15574,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* Queue.offer(harness.sdkMessages, wakeTaskStarted); yield* Queue.offer(harness.sdkMessages, turnOneResult); yield* Queue.take(harness.terminalReceipts); - + if (arrivedAfterReset) { + yield* TestClock.setTime(resetsAt * 1_000 + 1_000); + } yield* Queue.offer(harness.sdkMessages, wakeNotification); if (standaloneEvent) { yield* harness.offerAndWait( @@ -15501,9 +15601,16 @@ describe("ClaudeAdapterV2 background wake turns", () => { apiErrorParams: { rate_limit_info: { status: "rejected", - rateLimitType: "five_hour", + rateLimitType: weeklyStillActive ? "seven_day" : "five_hour", // A conflicting embedded time must not replace the event for this window. - resetsAt: standaloneEvent ? resetsAt + 3_600 : resetsAt, + resetsAt: weeklyStillActive + ? weeklyResetsAt + : standaloneEvent + ? resetsAt + 3_600 + : resetsAt, + ...(overageIndicator === "status" ? { overageStatus: "allowed" } : {}), + ...(overageIndicator === "using" ? { isUsingOverage: true } : {}), + ...(overageIndicator === "inUse" ? { overageInUse: true } : {}), }, }, }), @@ -15513,6 +15620,16 @@ describe("ClaudeAdapterV2 background wake turns", () => { makeResultFrame({ uuid: "00000000-0000-4000-8000-000000000632", result: "You've hit your session limit ยท resets 11:10am (Australia/Sydney)", + ...(twoWakes && newerRootOpen + ? { + usage: { + input_tokens: 2, + output_tokens: 3, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + } + : {}), isError: true, apiErrorStatus: 429, terminalReason: "api_error", @@ -15521,42 +15638,255 @@ describe("ClaudeAdapterV2 background wake turns", () => { ); assert.lengthOf(harness.continuationRequests, 1); + if (twoWakes) { + // A second rejected background wake completes while the first + // continuation remains queued. Both rejections are expired by replay. + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "rate_limit_event", + rate_limit_info: { + status: "rejected", + rateLimitType: secondWakeStillActive ? "seven_day" : "five_hour", + resetsAt: secondWakeStillActive ? weeklyResetsAt : resetsAt, + overageStatus: "rejected", + }, + uuid: "00000000-0000-4000-8000-000000000637", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* harness.offerAndWait( + makeAssistantErrorFrame({ + uuid: "00000000-0000-4000-8000-000000000638", + error: "rate_limit", + apiErrorParams: { + rate_limit_info: { + status: "rejected", + rateLimitType: secondWakeStillActive ? "seven_day" : "five_hour", + resetsAt: secondWakeStillActive ? weeklyResetsAt : resetsAt, + }, + }, + }), + ); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000639", + result: "The second background wake also hit the limit.", + ...(newerRootOpen + ? { + usage: { + input_tokens: 4, + output_tokens: 5, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + } + : {}), + isError: true, + apiErrorStatus: 429, + terminalReason: "api_error", + origin: { kind: "task-notification" }, + }), + ); + assert.lengthOf(harness.continuationRequests, 1); + } + // An empty session-task lifecycle result arriving before T3 opens the // wake run must not replace the root's terminal usage-limit result. yield* harness.offerAndWait(wakeTurnInit); - yield* harness.offerAndWait( - makeResultFrame({ - uuid: "00000000-0000-4000-8000-000000000633", - result: "", - numTurns: 0, - origin: { kind: "task-notification" }, - terminalReason: null, - }), - ); + if (!newerRootOpen || emptyBeforeNewRoot) { + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000633", + result: "", + numTurns: 0, + origin: { kind: "task-notification" }, + terminalReason: null, + }), + ); + } + if (expired) { + yield* TestClock.setTime(resetsAt * 1_000 + 60_000); + } + if (newerRootOpen) { + // The earlier 429 has expired, but another background wake has + // started and must be allowed to finish this same continuation. + yield* harness.offerAndWait( + claudeSdkFrame({ + ...wakeTurnInit, + uuid: "00000000-0000-4000-8000-000000000643", + }), + ); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000000644", + text: "A later wake is still running.", + }), + ); + } + if (absorbedByUser) { + // A normal user message can consume the newer root's output, + // so no terminal SDK result will ever follow for the continuation. + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-claude-wake-limit-absorber"), + text: "Did the new background task finish?", + attachments: [], + providerTurnOrdinal: 2, + }), + ); + yield* harness.until(() => harness.offeredMessages.length === 2); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000646", + result: "The background task was absorbed into this answer.", + }), + ); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + } + if (newerTurn) { + // A queued continuation may run behind an ordinary turn served after reset. + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("attempt-claude-wake-limit-intervening"), + text: "Check whether the limit cleared.", + attachments: [], + providerTurnOrdinal: 2, + }), + ); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000636", + result: "The new turn succeeded.", + }), + ); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "completed"); + } yield* harness.runtime.startTurn( makeClaudeTestTurnInput({ threadId: harness.threadId, providerThread: harness.providerThread, - now, + now: yield* DateTime.now, attemptId: RunAttemptId.make("attempt-claude-wake-limit-2"), text: "Background task completed.", attachments: [], - providerTurnOrdinal: 2, + providerTurnOrdinal: newerTurn || absorbedByUser ? 3 : 2, messageCreatedBy: "agent", messageCreationSource: "provider", }), ); + if (newerRootOpen) { + // An unabsorbed root must not settle until its own result arrives. + // An absorbed root has no result left to deliver and must settle now. + yield* yieldTicks(20); + assert.lengthOf(harness.terminalEvents(), absorbedByUser ? 3 : 1); + if (!absorbedByUser) { + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "00000000-0000-4000-8000-000000000645", + result: "The newer background wake completed.", + ...(twoWakes + ? { + usage: { + input_tokens: 7, + output_tokens: 11, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + } + : {}), + origin: { kind: "task-notification" }, + }), + ); + } + const finished = absorbedByUser + ? harness.terminalEvents().at(-1) + : yield* Queue.take(harness.terminalReceipts); + assert.equal(finished?.status, "completed"); + const completed = harness.events.findLast( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "completed", + ); + assert.equal(completed?.type, "provider_turn.updated"); + if (completed?.type === "provider_turn.updated") { + // Retain every charged rejected wake exactly once, even when a + // newer root is absorbed by a user turn and emits no own result. + assert.equal( + completed.providerTurn.turnTokenUsage?.inputTokens, + twoWakes ? 13 : absorbedByUser ? 1 : 2, + ); + assert.equal( + completed.providerTurn.turnTokenUsage?.outputTokens, + twoWakes ? 19 : absorbedByUser ? 1 : 2, + ); + } + assert.isFalse( + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "system_notice" && + event.turnItem.message.includes("usage limit reached"), + ), + ); + return; + } const terminal = yield* Queue.take(harness.terminalReceipts); - assert.equal(terminal?.status, "failed"); - if (terminal === undefined || terminal.status !== "failed") return; + if ( + expired && + !secondWakeStillActive && + (overageIndicator !== "none" ? standaloneEvent : !weeklyStillActive) + ) { + assert.equal(terminal.status, "completed"); + assert.isFalse( + harness.events.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "system_notice" && + event.turnItem.message.includes("usage limit reached"), + ), + ); + const completed = harness.events.findLast( + (event) => event.type === "provider_turn.updated", + ); + assert.equal(completed?.type, "provider_turn.updated"); + if (completed?.type === "provider_turn.updated") { + assert.equal(completed.providerTurn.turnTokenUsage?.inputTokens, twoWakes ? 2 : 1); + assert.equal(completed.providerTurn.turnTokenUsage?.outputTokens, twoWakes ? 2 : 1); + } + return; + } + assert.equal(terminal.status, "failed"); + if (terminal.status !== "failed") return; assert.equal(terminal.failure.class, "usage_limit"); - assert.equal(terminal.failure.resetAt, "2026-09-25T01:10:00.000Z"); + if (!standaloneEvent && overageIndicator !== "none") { + // With no blocking reset evidence, preserve the actual 429 failure, + // but do not infer a reset timestamp from overage-allowed metadata. + assert.isNotOk(terminal.failure.resetAt); + } else { + assert.equal( + terminal.failure.resetAt, + weeklyStillActive || secondWakeStillActive + ? "2026-10-02T01:10:00.000Z" + : "2026-09-25T01:10:00.000Z", + ); + } if (standaloneEvent) { const pause = yield* Queue.take(harness.systemNoticeReceipts); assert.equal(pause.turnItem.type, "system_notice"); if (pause.turnItem.type !== "system_notice") return; - assert.include(pause.turnItem.message, "This turn is paused until the 5-hour limit"); + assert.include( + pause.turnItem.message, + secondWakeStillActive + ? "This turn is paused until the 7-day limit" + : "This turn is paused until the 5-hour limit", + ); } }).pipe( Effect.provide( diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 17a9ca8486a38..fa96545ba4eda 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -4451,6 +4451,85 @@ const isClaudeBlockedAssistantRateLimit = Schema.is( }), ); +// An idle wake can be rejected before a reset, then replayed after a newer turn succeeds. +// Only discard the old rejection when its explicitly reported reset has already passed. +// A live 429 or one with an unknown/future reset still needs normal recovery. +function expiredBufferedWakeRateLimitFrames( + frames: ReadonlyArray, + result: SDKMessage, + receivedAtMs: number, + nowMs: number, +): ReadonlySet | null { + if ( + result.type !== "result" || + result.subtype !== "success" || + !result.is_error || + result.api_error_status !== 429 + ) { + return null; + } + const resultIndex = frames.indexOf(result); + if (resultIndex < 0) return null; + const beforeResult = frames.slice(0, resultIndex); + const lastRootStart = beforeResult.findLastIndex(isClaudeTurnStartMessage); + const lastResult = beforeResult.findLastIndex((frame) => frame.type === "result"); + const segment = frames.slice(Math.max(lastRootStart, lastResult) + 1, resultIndex); + const directLimits = segment.flatMap((frame) => + frame.type === "rate_limit_event" && + frame.rate_limit_info?.status === "rejected" && + frame.rate_limit_info.overageStatus !== "allowed" && + frame.rate_limit_info.overageStatus !== "allowed_warning" && + frame.rate_limit_info.isUsingOverage !== true && + frame.rate_limit_info.overageInUse !== true + ? [frame.rate_limit_info] + : [], + ); + const directTypes = new Set( + directLimits.map((limit) => limit.rateLimitType ?? "unknown"), + ); + // Like result handling, a standalone event wins for its own limit type only. + const embeddedResets = segment.flatMap((frame) => { + if (!isClaudeBlockedAssistantRateLimit(frame)) return []; + const info = frame.api_error_params.rate_limit_info; + if ( + info.overageStatus === "allowed" || + info.overageStatus === "allowed_warning" || + info.isUsingOverage === true || + info.overageInUse === true + ) { + return []; + } + const { rateLimitType, resetsAt } = info; + return directLimits.length === 0 || + (rateLimitType != null && !directTypes.has(String(rateLimitType))) + ? [resetsAt] + : []; + }); + const resets = [...directLimits.map((limit) => limit.resetsAt), ...embeddedResets]; + // Without a blocking reset, an overage-allowed payload cannot prove a 429 expired. + if ( + resets.length === 0 || + !resets.every( + (reset) => + typeof reset === "number" && + Number.isFinite(reset) && + reset > 0 && + reset * 1_000 > receivedAtMs && + reset * 1_000 <= nowMs, + ) + ) { + return null; + } + return new Set( + segment.filter( + (frame) => + (frame.type === "rate_limit_event" && frame.rate_limit_info?.status === "rejected") || + (frame.type === "assistant" && + (frame.error === "rate_limit" || isClaudeBlockedAssistantRateLimit(frame))), + ), + ); +} + function claudeUsageLimitResetAt(resetsAt: unknown): string | null { const resetMs = typeof resetsAt === "number" ? resetsAt * 1000 : NaN; return Number.isFinite(resetMs) && resetMs > 0 && resetMs < 8.64e15 @@ -12774,6 +12853,35 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ); const lastResult = drained[replayResultIndex === -1 ? lastResultIndex : replayResultIndex]; + const lastUnsettledRootIndex = drained.findLastIndex(isClaudeTurnStartMessage); + const newerRootStillOpen = + lastResult !== undefined && + lastUnsettledRootIndex > drained.indexOf(lastResult) && + !drained + .slice(lastUnsettledRootIndex + 1) + .some((entry) => entry.type === "result"); + // More than one rejected wake can be buffered behind a continuation. + // Suppress expired evidence from every completed wake, but only the + // selected terminal result may complete this continuation. + const expiredWakeFrames = new Set(); + const expiredResults: SDKResultMessage[] = []; + let lastResultExpired = false; + const replayedAtMs = DateTime.toEpochMillis(yield* DateTime.now); + for (const entry of drained) { + if (entry.type !== "result") continue; + const receivedAt = bufferedResultArrivals.get(entry); + if (receivedAt === undefined) continue; + const expired = expiredBufferedWakeRateLimitFrames( + drained, + entry, + DateTime.toEpochMillis(receivedAt), + replayedAtMs, + ); + if (expired === null) continue; + for (const frame of expired) expiredWakeFrames.add(frame); + expiredResults.push(entry); + if (entry === lastResult) lastResultExpired = true; + } const opaqueReplayTombstones = taskIdSetForNativeThread( yield* Ref.get(opaqueBackgroundTaskReplayTombstonesByNativeThread), nativeThreadId, @@ -12790,7 +12898,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ) { continue; } - if (entry.type !== "result") { + if (entry.type !== "result" && !expiredWakeFrames.has(entry)) { yield* handleSdkMessage({ query: querySession.query, message: entry }); } yield* Ref.update(wakeBuffers, (current) => { @@ -12808,6 +12916,40 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return next; }); } + // Every expired result was consumed from the wake buffer. Retain its + // charged usage once, except when the selected result is about to + // supply that usage through finalizeActiveTurn itself. + const selectedResultWillFinalize = + lastResultExpired && + (wakeAbsorbed || (!emptyTurnBeforeNextTurn && !newerRootStillOpen)); + for (const entry of expiredResults) { + if (selectedResultWillFinalize && entry === lastResult) continue; + context.steeredSegmentUsage.push(entry.usage); + context.steeredLedgerUsage = addClaudeLedgerUsage( + context.steeredLedgerUsage, + resultLedgerUsage.get(entry), + ); + } + if (lastResultExpired && newerRootStillOpen && !wakeAbsorbed) { + // A newer root still owns this continuation. Do not dispatch + // the expired result or settle the root before its own result. + if (leavesLaterWake) yield* offerBufferedWake(nativeThreadId); + return; + } + if (lastResultExpired && (!emptyTurnBeforeNextTurn || wakeAbsorbed)) { + // The selected wake ended before the limit reset. It is no longer a + // rejection of this continuation, so do not fail it or show stale errors. + const completedAt = yield* DateTime.now; + yield* finalizeActiveTurn({ + context, + status: "completed", + completedAt, + // The old wake may have incurred usage even though its rejection has expired. + ...(lastResult?.type === "result" ? { result: lastResult } : {}), + }); + if (leavesLaterWake) yield* offerBufferedWake(nativeThreadId); + return; + } if (lastResult !== undefined && !emptyTurnBeforeNextTurn) { yield* handleSdkMessage({ query: querySession.query, message: lastResult }); if (leavesLaterWake) yield* offerBufferedWake(nativeThreadId);