diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.ts b/apps/server/src/provider/Layers/ClaudeAdapter.ts index 6096fd7cbef7..184c9aaa0008 100644 --- a/apps/server/src/provider/Layers/ClaudeAdapter.ts +++ b/apps/server/src/provider/Layers/ClaudeAdapter.ts @@ -2648,6 +2648,10 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* ( context.lastKnownContextWindow, context.lastKnownTotalProcessedTokens, ); + if (snapshot !== undefined && context.turnState) { + context.turnState.latestAssistantUsage = event.usage; + context.turnState.compactedSinceLatestAssistantUsage = false; + } yield* emitThreadTokenUsage(context, snapshot, { rawMethod: "claude/stream_event/message_delta", rawPayload: message, diff --git a/apps/server/src/provider/Layers/ClaudeAdapter.usageRegression.test.ts b/apps/server/src/provider/Layers/ClaudeAdapter.usageRegression.test.ts new file mode 100644 index 000000000000..63702b1789ce --- /dev/null +++ b/apps/server/src/provider/Layers/ClaudeAdapter.usageRegression.test.ts @@ -0,0 +1,315 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import type { + Options as ClaudeQueryOptions, + SDKMessage, + SDKUserMessage, +} from "@anthropic-ai/claude-agent-sdk"; +import { ClaudeSettings, ProviderDriverKind, ThreadId } from "@t3tools/contracts"; +import { assert, describe, it } from "@effect/vitest"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Random from "effect/Random"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; + +import { ServerConfig } from "../../config.ts"; +import { ServerSettingsService } from "../../serverSettings.ts"; +import type { ClaudeAdapterShape } from "../Services/ClaudeAdapter.ts"; +import { makeClaudeAdapter } from "./ClaudeAdapter.ts"; + +const decodeClaudeSettings = Schema.decodeSync(ClaudeSettings); +const THREAD_ID = ThreadId.make("thread-claude-usage-regression"); + +class ClaudeUsageRegressionAdapter extends Context.Service< + ClaudeUsageRegressionAdapter, + ClaudeAdapterShape +>()("t3/provider/Layers/ClaudeAdapter.usageRegression.test/ClaudeUsageRegressionAdapter") {} + +class FakeClaudeQuery implements AsyncIterable { + private readonly queue: Array = []; + private readonly waiters: Array<(value: IteratorResult) => void> = []; + private done = false; + + emit(message: SDKMessage): void { + const waiter = this.waiters.shift(); + if (waiter) { + waiter({ done: false, value: message }); + return; + } + this.queue.push(message); + } + + finish(): void { + this.done = true; + for (const waiter of this.waiters.splice(0)) { + waiter({ done: true, value: undefined }); + } + } + + readonly setModel = async (_model?: string): Promise => {}; + readonly setPermissionMode = async (_mode: string): Promise => {}; + readonly setMaxThinkingTokens = async (_tokens: number | null): Promise => {}; + readonly close = (): void => this.finish(); + + [Symbol.asyncIterator](): AsyncIterator { + return { + next: () => { + const value = this.queue.shift(); + if (value) { + return Promise.resolve({ done: false, value }); + } + if (this.done) { + return Promise.resolve({ done: true, value: undefined }); + } + return new Promise((resolve) => this.waiters.push(resolve)); + }, + }; + } +} + +/** Builds the minimal live adapter layer needed to drive SDK messages through completeTurn. */ +function makeHarness() { + const query = new FakeClaudeQuery(); + let createInput: + | { + readonly prompt: AsyncIterable; + readonly options: ClaudeQueryOptions; + } + | undefined; + + const layer = Layer.effect( + ClaudeUsageRegressionAdapter, + Effect.gen(function* () { + return yield* makeClaudeAdapter(decodeClaudeSettings({}), { + createQuery: (input) => { + createInput = input; + return query; + }, + }); + }), + ).pipe( + Layer.provideMerge(ServerConfig.layerTest("/tmp/claude-usage-regression", "/tmp")), + Layer.provideMerge(ServerSettingsService.layerTest()), + Layer.provideMerge(NodeServices.layer), + ); + + return { layer, query, getCreateInput: () => createInput }; +} + +/** Returns a deterministic Random service so runtime event ids are stable in the test harness. */ +function makeDeterministicRandomService(seed = 0x1234_5678): { + nextIntUnsafe: () => number; + nextDoubleUnsafe: () => number; +} { + let state = seed >>> 0; + const nextIntUnsafe = (): number => { + state = (Math.imul(1_664_525, state) + 1_013_904_223) >>> 0; + return state; + }; + return { + nextIntUnsafe, + nextDoubleUnsafe: () => nextIntUnsafe() / 0x1_0000_0000, + }; +} + +describe("ClaudeAdapter cumulative usage regression", () => { + it.effect("preserves parent message_delta context over cumulative result usage (#8594)", () => { + const harness = makeHarness(); + + return Effect.gen(function* () { + const adapter = yield* ClaudeUsageRegressionAdapter; + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId: THREAD_ID, + input: "hello", + attachments: [], + }); + + // This is the parent-session path from the issue: message_delta carries + // the per-request active context. It deliberately does not use + // task_progress, which is the separate subagent-meter bug (#4650). + harness.query.emit({ + type: "stream_event", + session_id: "sdk-session-8594", + uuid: "message-delta-8594", + parent_tool_use_id: null, + event: { + type: "message_delta", + delta: { + stop_reason: null, + stop_sequence: null, + }, + usage: { + input_tokens: 100000, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 12994, + output_tokens: 0, + }, + }, + } as unknown as SDKMessage); + + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + duration_ms: 1234, + duration_api_ms: 1200, + num_turns: 1, + result: "done", + stop_reason: "end_turn", + session_id: "sdk-session-8594", + usage: { + input_tokens: 100000, + cache_creation_input_tokens: 50000, + cache_read_input_tokens: 2034473, + output_tokens: 18487, + total_tokens: 2202960, + iterations: [], + }, + modelUsage: { + "claude-opus-4-6": { + contextWindow: 1000000, + maxOutputTokens: 64000, + }, + }, + } as unknown as SDKMessage); + harness.query.finish(); + + const events = Array.from(yield* Fiber.join(eventsFiber)); + const finalUsage = events.findLast((event) => event.type === "thread.token-usage.updated"); + assert.equal(finalUsage?.type, "thread.token-usage.updated"); + if (finalUsage?.type === "thread.token-usage.updated") { + assert.deepEqual(finalUsage.payload.usage, { + usedTokens: 112994, + lastUsedTokens: 112994, + totalProcessedTokens: 2202960, + inputTokens: 112994, + maxTokens: 1000000, + }); + } + + // Keep the prompt reference live so this harness matches the real SDK + // setup shape and catches accidental query construction regressions. + assert.isDefined(harness.getCreateInput()); + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); + + it.effect("uses the current result when a later turn has no message_delta", () => { + const harness = makeHarness(); + + return Effect.gen(function* () { + const adapter = yield* ClaudeUsageRegressionAdapter; + yield* adapter.startSession({ + threadId: THREAD_ID, + provider: ProviderDriverKind.make("claudeAgent"), + runtimeMode: "full-access", + }); + + const firstTurnEvents = yield* adapter.streamEvents.pipe( + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + yield* adapter.sendTurn({ threadId: THREAD_ID, input: "first", attachments: [] }); + harness.query.emit({ + type: "stream_event", + session_id: "sdk-session-stale-usage", + uuid: "message-delta-first-turn", + parent_tool_use_id: null, + event: { + type: "message_delta", + delta: { stop_reason: null, stop_sequence: null }, + usage: { + input_tokens: 100000, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 12994, + output_tokens: 0, + }, + }, + } as unknown as SDKMessage); + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + duration_ms: 100, + duration_api_ms: 90, + num_turns: 1, + result: "first done", + stop_reason: "end_turn", + session_id: "sdk-session-stale-usage", + usage: { + input_tokens: 100000, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 12994, + output_tokens: 0, + total_tokens: 112994, + iterations: [], + }, + modelUsage: { + "claude-opus-4-6": { contextWindow: 1000000, maxOutputTokens: 64000 }, + }, + } as unknown as SDKMessage); + yield* Fiber.join(firstTurnEvents); + + const secondTurnEvents = yield* adapter.streamEvents.pipe( + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + yield* adapter.sendTurn({ threadId: THREAD_ID, input: "second", attachments: [] }); + // No message_delta or assistant snapshot is emitted for this turn. The + // result is therefore the only current-turn active-usage reading and + // must beat the session-wide lastKnownTokenUsage from the prior turn. + harness.query.emit({ + type: "result", + subtype: "success", + is_error: false, + duration_ms: 100, + duration_api_ms: 90, + num_turns: 1, + result: "second done", + stop_reason: "end_turn", + session_id: "sdk-session-stale-usage", + usage: { + input_tokens: 200000, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 10000, + output_tokens: 5000, + total_tokens: 215000, + iterations: [], + }, + modelUsage: { + "claude-opus-4-6": { contextWindow: 1000000, maxOutputTokens: 64000 }, + }, + } as unknown as SDKMessage); + harness.query.finish(); + + const events = Array.from(yield* Fiber.join(secondTurnEvents)); + const finalUsage = events.findLast((event) => event.type === "thread.token-usage.updated"); + assert.equal(finalUsage?.type, "thread.token-usage.updated"); + if (finalUsage?.type === "thread.token-usage.updated") { + assert.equal(finalUsage.payload.usage.usedTokens, 215000); + assert.equal(finalUsage.payload.usage.inputTokens, 210000); + assert.equal(finalUsage.payload.usage.outputTokens, 5000); + } + }).pipe( + Effect.provideService(Random.Random, makeDeterministicRandomService()), + Effect.provide(harness.layer), + ); + }); +});