diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 043f3b8a756d9..eabec272c3be3 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -99,6 +99,7 @@ import { WorktreeToolkit } from "../../mcp/toolkits/worktree/tools.ts"; import { ThreadToolkit } from "../../mcp/toolkits/thread/tools.ts"; import { OrchestratorToolkit } from "../../mcp/toolkits/orchestrator/tools.ts"; import { ClaudeExecutableFileCheck } from "../../provider/Drivers/ClaudeExecutable.ts"; +import { claudeSignedOutMessage } from "../../provider/Drivers/ClaudeHome.ts"; import { BUNDLED_CLAUDE_MODEL_CATALOG, type ClaudeModelCatalog, @@ -16637,6 +16638,438 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + const replayWake = ( + id: string, + text: string, + result: Omit[0], "uuid">, + ) => [ + claudeSdkFrame({ ...wakeTurnInit, uuid: `${id}-init` }), + makeAssistantTextFrame({ uuid: `${id}-assistant`, text }), + makeResultFrame({ ...result, uuid: `${id}-result` }), + ]; + const earlierWake = replayWake("earlier-a", "Earlier A.", { + result: "Earlier A.", + origin: { kind: "task-notification" }, + usage: { + input_tokens: 10, + output_tokens: 20, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }); + const earlierErrorWake = replayWake("earlier-error", "Earlier A.", { + result: "Earlier A.", + origin: { kind: "task-notification" }, + isError: true, + terminalReason: "api_error", + errors: ["Tool execution failed: EACCES"], + usage: { + input_tokens: 10, + output_tokens: 20, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }); + const emptyWakeSuccess = makeResultFrame({ + uuid: "empty-wake-result", + result: "", + numTurns: 0, + origin: { kind: "task-notification" }, + terminalReason: null, + usage: { + input_tokens: 7, + output_tokens: 7, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }); + // Claude's modelUsage is a running total for the whole CLI process. + const ledgerTotal = (inputTokens: number, outputTokens: number) => ({ + "claude-sonnet-4-6": { + inputTokens, + outputTokens, + cacheReadInputTokens: 0, + cacheCreationInputTokens: 0, + webSearchRequests: 0, + costUSD: 0, + contextWindow: 0, + maxOutputTokens: 32_000, + }, + }); + it.effect.each([ + { + label: "stopping the successor preserves earlier billed usage", + before: earlierWake, + after: [wakeTurnInit], + beforeCheck: [wakeAssistant], + live: [], + texts: ["Earlier A.", WAKE_ASSISTANT_TEXT], + inputTokens: 10, + outputTokens: 20, + notices: [], + openOnAttach: true, + stop: true, + }, + { + label: "older results preserve the active successor and its approval", + before: [wakeTurnInit, emptyWakeSuccess, ...earlierWake, ...earlierErrorWake], + after: [wakeTurnInit], + beforeCheck: [wakeAssistant], + live: [wakeResult], + texts: ["Earlier A.", "Earlier A.", WAKE_ASSISTANT_TEXT], + inputTokens: 21, + outputTokens: 41, + notices: ["Tool execution failed: EACCES"], + openOnAttach: true, + }, + { + label: "fully buffered successor without assistant text uses its own result", + before: replayWake("silent-older", "", { + result: "Earlier A.", + origin: { kind: "task-notification" }, + usage: { + input_tokens: 10, + output_tokens: 20, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }), + after: [ + wakeTurnInit, + makeResultFrame({ + uuid: "successor-result", + result: "Successor text.", + origin: { kind: "task-notification" }, + }), + ], + beforeCheck: [], + live: [], + texts: ["Successor text."], + inputTokens: 11, + outputTokens: 21, + notices: [], + openOnAttach: false, + }, + { + label: "the earlier wake's ledger growth stays in the successor's other usage", + // Totals 100/200, 115/225, 120/230, 126/238: the earlier wake grew 15/25 and the + // successor 6/8, against root usage 10/20 and 1/1, so other work is 10/12. + ledger: { + turnOne: ledgerTotal(100, 200), + prompt: ledgerTotal(120, 230), + otherUsage: { inputTokens: 10, outputTokens: 12 }, + }, + before: replayWake("ledger-older", "", { + result: "Earlier A.", + origin: { kind: "task-notification" }, + usage: { + input_tokens: 10, + output_tokens: 20, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + modelUsage: ledgerTotal(115, 225), + }), + after: [ + wakeTurnInit, + makeResultFrame({ + uuid: "ledger-successor-result", + result: "Successor text.", + origin: { kind: "task-notification" }, + modelUsage: ledgerTotal(126, 238), + }), + ], + beforeCheck: [], + live: [], + texts: ["Successor text."], + inputTokens: 11, + outputTokens: 21, + notices: [], + openOnAttach: false, + }, + { + label: "each earlier wake's failure notice follows that wake's own sign-in state", + before: [ + claudeSdkFrame({ ...wakeTurnInit, uuid: "signed-out-init" }), + makeAssistantErrorFrame({ uuid: "signed-out-assistant", error: "authentication_failed" }), + makeResultFrame({ + uuid: "signed-out-result", + result: "", + origin: { kind: "task-notification" }, + isError: true, + terminalReason: "api_error", + }), + claudeSdkFrame({ ...wakeTurnInit, uuid: "api-error-init" }), + makeResultFrame({ + uuid: "api-error-result", + result: "", + origin: { kind: "task-notification" }, + isError: true, + terminalReason: "api_error", + }), + claudeSdkFrame({ ...wakeTurnInit, uuid: "denied-init" }), + makeAssistantErrorFrame({ uuid: "denied-assistant", error: "authentication_failed" }), + makeResultFrame({ + uuid: "denied-result", + result: + "Failed to authenticate. API Error: 403 Your organization does not have access to this model.", + origin: { kind: "task-notification" }, + isError: true, + apiErrorStatus: 403, + terminalReason: "api_error", + }), + ], + after: [ + wakeTurnInit, + makeAssistantTextFrame({ uuid: "signed-in-successor-assistant", text: "Successor text." }), + makeResultFrame({ + uuid: "signed-in-successor-result", + result: "Successor text.", + origin: { kind: "task-notification" }, + }), + ], + beforeCheck: [], + live: [], + texts: [ + "Claude could not complete this request.", + "Claude could not complete this request.", + "Successor text.", + ], + inputTokens: 4, + outputTokens: 4, + notices: [ + claudeSignedOutMessage({ configDir: undefined, cwd: "/workspace" }), + "Claude gave up after repeated API errors.", + "Failed to authenticate. API Error: 403 Your organization does not have access to this model.", + ], + openOnAttach: false, + }, + { + label: "two older results and latest empty successes keep base text fallback", + before: [ + ...replayWake("fallback-x", "", { + result: "Earlier X.", + origin: { kind: "task-notification" }, + usage: { + input_tokens: 10, + output_tokens: 20, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }), + ...replayWake("fallback-a", "", { + result: "Earlier A.", + origin: { kind: "task-notification" }, + usage: { + input_tokens: 100, + output_tokens: 200, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + }, + }), + ], + after: [ + wakeTurnInit, + makeAssistantTextFrame({ uuid: "empty-assistant", text: "" }), + emptyWakeSuccess, + claudeSdkFrame({ ...emptyWakeSuccess, uuid: "second-empty-result" }), + ], + beforeCheck: [], + live: [], + texts: ["Earlier A."], + inputTokens: 110, + outputTokens: 220, + notices: [], + openOnAttach: false, + }, + ])("replays wakes: $label", (scenario) => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeWakeHarnessWithOptions({ + close: (sdkMessages) => Queue.shutdown(sdkMessages), + }); + const now = yield* DateTime.now; + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "approval-required", + interactionMode: "default", + cwd: "/workspace", + }); + const wakeAttempt = RunAttemptId.make("attempt-claude-replay-3"); + const wakeRun = RunId.make(`run-${wakeAttempt}`); + const ledger = "ledger" in scenario ? scenario.ledger : undefined; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + runtimePolicy, + attemptId: RunAttemptId.make("attempt-claude-replay-1"), + text: "Run the build in the background.", + attachments: [], + }), + ); + yield* harness.offerAndWait( + ledger === undefined + ? turnOneResult + : claudeSdkFrame({ ...turnOneResult, modelUsage: ledger.turnOne }), + ); + yield* harness.awaitTerminal(1); + for (const frame of scenario.before) yield* harness.offerAndWait(frame); + yield* Queue.take(harness.continuationReceipts); + // The queued user prompt runs before the buffered continuation attaches. + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + runtimePolicy, + attemptId: RunAttemptId.make("attempt-claude-replay-2"), + text: "Check the build.", + attachments: [], + providerTurnOrdinal: 2, + }), + ); + yield* harness.offerAndWait(claudeSdkFrame({ ...wakeTurnInit, uuid: "prompt-init" })); + yield* harness.offerAndWait( + makeAssistantTextFrame({ uuid: "prompt-assistant", text: "Checked the build." }), + ); + yield* harness.offerAndWait( + claudeSdkFrame({ + ...makeResultFrame({ + uuid: "prompt-result", + result: "Checked the build.", + ...(ledger === undefined ? {} : { modelUsage: ledger.prompt }), + }), + user_message_uuid: harness.promptUuid(1), + }), + ); + yield* harness.awaitTerminal(2); + for (const frame of scenario.after) yield* harness.offerAndWait(frame); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + runtimePolicy, + attemptId: wakeAttempt, + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 3, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + for (const frame of scenario.beforeCheck) { + yield* harness.offerAndWait(frame); + yield* harness.until( + () => + harness.terminalEvents().length >= 3 || + harness.events.some( + (event) => + event.type === "message.updated" && + event.message.runId === wakeRun && + event.message.text === WAKE_ASSISTANT_TEXT, + ), + ); + } + if (!scenario.openOnAttach) yield* harness.awaitTerminal(3); + const terminalCountOnAttach = harness.terminalEvents().length; + if (scenario.openOnAttach) { + assert.equal(terminalCountOnAttach, 2); + const permission = yield* Effect.promise(() => + harness.getOpenedOptions()!.canUseTool!( + "Bash", + { command: "git status" }, + { + signal: new AbortController().signal, + toolUseID: "toolu_replayed_wake_approval", + requestId: "request-replayed-wake", + }, + ), + ).pipe(Effect.forkScoped); + const request = yield* Queue.take(harness.runtimeRequestReceipts); + const wakeTurn = harness.events.findLast( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.runAttemptId === wakeAttempt, + ); + assert.equal( + request.runtimeRequest.providerTurnId, + wakeTurn?.type === "provider_turn.updated" ? wakeTurn.providerTurn.id : undefined, + ); + yield* harness.runtime.respondToRuntimeRequest({ + requestId: request.runtimeRequest.id, + decision: "accept", + }); + assert.equal((yield* Fiber.join(permission))?.behavior, "allow"); + } + for (const frame of scenario.live) yield* harness.offerAndWait(frame); + if ("stop" in scenario && scenario.stop) { + yield* harness.runtime.interruptTurn({ + providerThread: harness.providerThread, + providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeTurnId: `turn:${wakeAttempt}`, + }), + }); + } + yield* harness.awaitTerminal(3); + const completed = harness.events.findLast( + (event) => event.type === "provider_turn.updated", + ); + assert.deepEqual( + { + terminalCountOnAttach, + texts: harness.events.flatMap((event) => + event.type === "message.updated" && event.message.runId === wakeRun + ? [event.message.text] + : [], + ), + status: harness.terminalEvents()[2]?.status, + usage: + completed?.type === "provider_turn.updated" + ? completed.providerTurn.turnTokenUsage + : undefined, + notices: harness.events.flatMap((event) => + event.type === "turn_item.updated" && + event.turnItem.runId === wakeRun && + event.turnItem.type === "system_notice" + ? [event.turnItem.message] + : [], + ), + }, + { + terminalCountOnAttach: scenario.openOnAttach ? 2 : 3, + texts: scenario.texts, + status: "stop" in scenario ? "interrupted" : "completed", + usage: { + usageScope: "main_agent", + usageStatus: "stop" in scenario ? "partial" : "complete", + hasSubagents: false, + inputTokens: scenario.inputTokens, + outputTokens: scenario.outputTokens, + cachedInputTokens: 0, + cacheCreationTokens: 0, + models: ["claude-sonnet-4-6"], + ...(ledger?.otherUsage === undefined ? {} : { otherUsage: ledger.otherUsage }), + }, + notices: scenario.notices, + }, + ); + assert.lengthOf(harness.offeredMessages, 2); + }).pipe( + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ), + ); + it.effect.each(["before", "after"] as const)( "offers a later wake when a wake a user turn absorbed is cancelled %s that turn ends", (cancelled) => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index e0659b0167f57..0332a55ec22df 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -4568,6 +4568,19 @@ function isClaudeEmptyTaskNotificationSuccess(message: SDKMessage): boolean { ); } +// Whether the wake that ends with the result at `resultIndex` reported a sign-in failure. +function claudeWakeSignedOut(frames: ReadonlyArray, resultIndex: number): boolean { + const wakeStart = frames.slice(0, resultIndex).findLastIndex(isClaudeTurnStartMessage); + return frames + .slice(wakeStart + 1, resultIndex) + .some( + (frame) => + frame.type === "assistant" && + frame.parent_tool_use_id === null && + frame.error === "authentication_failed", + ); +} + // The CLI throttles unchanged rate_limit_event frames, but repeats this // undeclared payload on each blocked assistant response. const isClaudeBlockedAssistantRateLimit = Schema.is( @@ -4939,13 +4952,13 @@ interface ActiveClaudeTurnContext { compactionFailure: { readonly message: string | undefined } | null; // The title a `session_title_changed` frame reported; a `/rename` result applies it to the thread. nativeSessionTitle: string | null; - // Steers end segments whose usage is excluded from the final result; bill them too, oldest first. - readonly steeredSegmentUsage: Array; + // The final result excludes earlier steered segments and buffered wakes; bill them too. + readonly earlierSegmentUsage: Array; // Set as finalization starts: a steer admitted later would stay pending // forever, so it fails and the orchestrator follows up instead. steersClosed: boolean; - // Ledger growth of the steered segments; undefined once any segment's growth is unknown. - steeredLedgerUsage: ClaudeLedgerUsage | undefined; + // Ledger growth of those earlier results; undefined once any result's growth is unknown. + earlierLedgerUsage: ClaudeLedgerUsage | undefined; readonly rootModels: Set; readonly subagentsByTaskId: Map; readonly subagentsByToolUseId: Map; @@ -9287,14 +9300,14 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( input.context.subagentsByTaskId.size > 0 || input.context.subagentsByToolUseId.size > 0, input.status, - input.context.steeredSegmentUsage, + input.context.earlierSegmentUsage, { rootModels: input.context.rootModels, ...(input.result === undefined ? {} : { ledger: addClaudeLedgerUsage( - input.context.steeredLedgerUsage, + input.context.earlierLedgerUsage, resultLedgerUsage.get(input.result), ), }), @@ -9496,6 +9509,66 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); }); + // Assistant frames carry the earlier wake's text; its result only adds usage and errors. + const recordEarlierWakeResult = Effect.fn("ClaudeAdapter.recordEarlierWakeResult")( + function* ( + context: ActiveClaudeTurnContext, + result: SDKResultMessage, + signedOut: boolean, + ) { + if (isClaudeEmptyTaskNotificationSuccess(result)) return; + context.earlierSegmentUsage.push(result.usage); + context.earlierLedgerUsage = addClaudeLedgerUsage( + context.earlierLedgerUsage, + resultLedgerUsage.get(result), + ); + const now = yield* DateTime.now; + // As for a live result: a 403 shows its text, and this wake's sign-in failure gives sign-in advice. + const permissionDenied = + result.subtype === "success" && result.api_error_status === 403; + const failure = providerFailureFromResult( + result, + now, + permissionDenied + ? result.result + : signedOut + ? context.authenticationFailureMessage + : undefined, + ); + if (failure !== null) { + yield* emitProviderEvent({ + type: "turn_item.updated", + driver: CLAUDE_PROVIDER, + turnItem: { + id: idAllocator.derive.turnItemFromProviderItem({ + driver: CLAUDE_PROVIDER, + nativeItemId: result.uuid, + }), + threadId: context.input.threadId, + runId: context.input.runId, + nodeId: context.input.rootNodeId, + providerThreadId: context.input.providerThread.id, + providerTurnId: context.providerTurnId, + nativeItemRef: { + driver: CLAUDE_PROVIDER, + nativeId: result.uuid, + strength: "strong", + }, + parentItemId: null, + ordinal: yield* resolveItemOrdinal(context, result.uuid), + type: "system_notice", + status: "completed", + title: failure.message, + message: failure.message, + startedAt: now, + completedAt: now, + updatedAt: now, + }, + }); + } + }, + ); + const finalizeActiveTurnAfterQueryExit = Effect.fnUntraced(function* ( closedQuery: ClaudeLiveQueryContext, cause?: Cause.Cause, @@ -11599,9 +11672,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( completedAt, }); } - context.steeredSegmentUsage.push(message.usage); - context.steeredLedgerUsage = addClaudeLedgerUsage( - context.steeredLedgerUsage, + context.earlierSegmentUsage.push(message.usage); + context.earlierLedgerUsage = addClaudeLedgerUsage( + context.earlierLedgerUsage, resultLedgerUsage.get(message), ); return; @@ -13019,9 +13092,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( latestAssistantRateLimit: null, compactionFailure: null, nativeSessionTitle: null, - steeredSegmentUsage: [], + earlierSegmentUsage: [], steersClosed: false, - steeredLedgerUsage: { inputTokens: 0, outputTokens: 0 }, + earlierLedgerUsage: { inputTokens: 0, outputTokens: 0 }, rootModels: new Set(), subagentsByTaskId: new Map(), subagentsByToolUseId: new Map(), @@ -13165,25 +13238,19 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* finalizeActiveTurn({ context, status: "completed", completedAt }); return; } - // Replay results last to avoid rebuffering later frames. An empty turn's result must not settle a newer running turn. + // Replay results last to avoid rebuffering later frames; a newer root init owns settlement. + const lastTurnStartIndex = drained.findLastIndex(isClaudeTurnStartMessage); const lastResultIndex = drained.findLastIndex((entry) => entry.type === "result"); - const tailResult = drained[lastResultIndex]; - const emptyTurnBeforeNextTurn = - tailResult?.type === "result" && - tailResult.num_turns === 0 && - !tailResult.is_error && - terminalStatusFromResult(tailResult) === "completed" && - drained - .slice(lastResultIndex + 1) - .some((entry) => isClaudeTurnStartMessage(entry)); - // The CLI can append empty task successes after a failure. Filter - // those for replay without changing the raw-tail successor guard. + // Preserve the base fallback when the latest wake has only empty + // task successes: an earlier non-empty result still owns failure. const replayResultIndex = drained.findLastIndex( (entry) => entry.type === "result" && !isClaudeEmptyTaskNotificationSuccess(entry), ); const lastResult = - drained[replayResultIndex === -1 ? lastResultIndex : replayResultIndex]; + lastResultIndex > lastTurnStartIndex + ? drained[replayResultIndex === -1 ? lastResultIndex : replayResultIndex] + : undefined; const opaqueReplayTombstones = taskIdSetForNativeThread( yield* Ref.get(opaqueBackgroundTaskReplayTombstonesByNativeThread), nativeThreadId, @@ -13194,7 +13261,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( entry.subtype === "task_notification" && opaqueReplayTombstones.has(entry.task_id), ); - for (const entry of drained) { + for (const [index, entry] of drained.entries()) { if ( !(yield* Ref.get(wakeBuffers)).get(nativeThreadId)?.messages.includes(entry) ) { @@ -13202,6 +13269,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } if (entry.type !== "result") { yield* handleSdkMessage({ query: querySession.query, message: entry }); + } else if (index < lastTurnStartIndex && entry !== lastResult) { + yield* recordEarlierWakeResult( + context, + entry, + claudeWakeSignedOut(drained, index), + ); } yield* Ref.update(wakeBuffers, (current) => { const buffered = current.get(nativeThreadId); @@ -13218,7 +13291,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return next; }); } - if (lastResult !== undefined && !emptyTurnBeforeNextTurn) { + if (lastResult !== undefined) { yield* handleSdkMessage({ query: querySession.query, message: lastResult }); if (leavesLaterWake) yield* offerBufferedWake(nativeThreadId); return;