From 34954c111f35b82db08a8b57a7791f203998b8f9 Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 29 Aug 2026 10:24:20 -0400 Subject: [PATCH 01/12] feat(opencode): surface context-window usage and session cost on the meter --- .../provider/Layers/OpenCodeAdapter.test.ts | 734 ++++++++++++++++++ .../src/provider/Layers/OpenCodeAdapter.ts | 273 ++++++- .../chat/ContextWindowMeter.test.tsx | 9 +- .../components/chat/ContextWindowMeter.tsx | 16 +- apps/web/src/lib/contextWindow.ts | 11 + packages/contracts/src/providerRuntime.ts | 5 + 6 files changed, 1032 insertions(+), 16 deletions(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 8fa4de72f6c5..d59f4fc7b956 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -47,6 +47,8 @@ import { isSameOpenCodeDirectory, makeOpenCodeAdapter, mergeOpenCodeAssistantText, + normalizeOpenCodeTokenCounts, + normalizeOpenCodeTokenUsage, } from "./OpenCodeAdapter.ts"; import { symlinksSupported } from "@t3tools/shared/testing/symlinks"; @@ -129,6 +131,9 @@ const runtimeMock = { questionListImplementation: null as (() => Promise>) | null, sessionUpdateCalls: [] as Array<{ sessionID: string; permission: unknown }>, forkCalls: [] as Array<{ sessionID: string; directory?: string }>, + modelListCalls: 0, + modelListResults: [] as Array>, + modelListError: null as Error | null, }, reset() { this.state.startCalls.length = 0; @@ -184,6 +189,9 @@ const runtimeMock = { this.state.questionListImplementation = null; this.state.sessionUpdateCalls.length = 0; this.state.forkCalls.length = 0; + this.state.modelListCalls = 0; + this.state.modelListResults = []; + this.state.modelListError = null; }, }; @@ -497,6 +505,15 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { ); }, }, + provider: { + list: async () => { + runtimeMock.state.modelListCalls += 1; + if (runtimeMock.state.modelListError) { + throw runtimeMock.state.modelListError; + } + return { data: { all: runtimeMock.state.modelListResults } }; + }, + }, }) as unknown as ReturnType, loadOpenCodeInventory: () => Effect.fail( @@ -6500,6 +6517,113 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("clamps partial, non-finite, and negative token counts to zero", () => + Effect.sync(() => { + const counts = normalizeOpenCodeTokenCounts({ + input: NaN, + output: -40, + reasoning: Number.POSITIVE_INFINITY, + cache: { read: 11.9, write: undefined }, + }); + NodeAssert.deepEqual(counts, { + input: 0, + output: 0, + reasoning: 0, + cacheRead: 12, + cacheWrite: 0, + }); + + // A payload missing the whole cache object must not throw. + const withoutCache = normalizeOpenCodeTokenCounts({ + input: 3, + output: 4, + reasoning: 0, + cache: undefined, + }); + NodeAssert.deepEqual(withoutCache, { + input: 3, + output: 4, + reasoning: 0, + cacheRead: 0, + cacheWrite: 0, + }); + }), + ); + + it.effect("maps OpenCode token counts into a context-window usage snapshot", () => + Effect.sync(() => { + const counts = normalizeOpenCodeTokenCounts({ + input: 100, + output: 40, + reasoning: 7, + cache: { read: 11, write: 13 }, + }); + NodeAssert.deepEqual(counts, { + input: 100, + output: 40, + reasoning: 7, + cacheRead: 11, + cacheWrite: 13, + }); + + const usage = normalizeOpenCodeTokenUsage({ + messageTokens: counts, + cumulativeTokens: counts, + maxTokens: 200_000, + cumulativeSessionCost: 0.42, + }); + NodeAssert.deepEqual(usage, { + usedTokens: 124, + totalProcessedTokens: 171, + maxTokens: 200_000, + cost: 0.42, + inputTokens: 124, + cachedInputTokens: 24, + outputTokens: 40, + reasoningOutputTokens: 7, + lastUsedTokens: 124, + lastInputTokens: 124, + lastCachedInputTokens: 24, + lastOutputTokens: 40, + lastReasoningOutputTokens: 7, + }); + }), + ); + + it.effect("omits cumulative and breakdown fields OpenCode did not report", () => + Effect.sync(() => { + const usage = normalizeOpenCodeTokenUsage({ + messageTokens: { input: 5, output: 0, reasoning: 0, cacheRead: 0, cacheWrite: 0 }, + cumulativeSessionCost: 0, + }); + NodeAssert.deepEqual(usage, { + usedTokens: 5, + inputTokens: 5, + cachedInputTokens: 0, + lastUsedTokens: 5, + lastInputTokens: 5, + lastCachedInputTokens: 0, + }); + }), + ); + + it.effect("returns no usage until an assistant message reports tokens", () => + Effect.sync(() => { + NodeAssert.equal( + normalizeOpenCodeTokenUsage({ + cumulativeTokens: { input: 50, output: 10, reasoning: 0, cacheRead: 20, cacheWrite: 0 }, + }), + undefined, + ); + NodeAssert.equal( + normalizeOpenCodeTokenUsage({ + messageTokens: { input: 0, output: 0, reasoning: 0, cacheRead: 0, cacheWrite: 0 }, + }), + undefined, + ); + }), + ); + it.effect("does not strip coincidental prefix overlap from OpenCode part deltas", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; @@ -7059,6 +7183,616 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("emits context-window usage from assistant messages and session totals", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage"); + runtimeMock.state.modelListResults = [ + { + id: "anthropic", + models: { + "claude-sonnet-4-5": { limit: { context: 200_000 } }, + }, + }, + ]; + runtimeMock.state.subscribedEvents = [ + { + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + title: "Usage thread", + model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + cost: 0.42, + }, + }, + }, + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-usage-1", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + const usageEvent = events[0]; + NodeAssert.equal(usageEvent?.type, "thread.token-usage.updated"); + if (usageEvent?.type !== "thread.token-usage.updated") { + return; + } + // Context usage = input + cache read/write of the latest assistant + // message; total processed folds output and reasoning in on top of the + // cumulative session totals; maxTokens comes from the catalog; cost is + // the cumulative session cost in USD. + NodeAssert.deepEqual(usageEvent.payload.usage, { + usedTokens: 70, + totalProcessedTokens: 85, + maxTokens: 200_000, + cost: 0.42, + inputTokens: 70, + cachedInputTokens: 20, + outputTokens: 10, + reasoningOutputTokens: 5, + lastUsedTokens: 70, + lastInputTokens: 70, + lastCachedInputTokens: 20, + lastOutputTokens: 10, + lastReasoningOutputTokens: 5, + }); + // The catalog is resolved once per model, not per event. + NodeAssert.equal(runtimeMock.state.modelListCalls, 1); + }), + ); + + it.effect("does not re-emit unchanged token usage for repeated updates", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage-dedupe"); + const assistantMessage = { + id: "msg-usage-dedupe", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }; + runtimeMock.state.subscribedEvents = [ + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: assistantMessage, + }, + }, + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { ...assistantMessage, id: "msg-usage-dedupe-2" }, + }, + }, + // Drain marker: its metadata event is emitted only after both + // messages have been fully processed, so the collected window is + // complete regardless of how many ambient events they produce. + { + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + title: "dedupe-marker", + }, + }, + }, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil( + (event) => + event.type === "thread.metadata.updated" && event.payload.name === "dedupe-marker", + ), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + const usageEvents = events.filter((event) => event.type === "thread.token-usage.updated"); + NodeAssert.equal(usageEvents.length, 1); + }), + ); + + it.effect("re-emits usage when output tokens grow on a later message", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage-output-growth"); + runtimeMock.state.subscribedEvents = [ + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-output-growth-1", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-output-growth-2", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 50, output: 90, reasoning: 30, cache: { read: 20, write: 0 } }, + }, + }, + }, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + NodeAssert.equal(events.length, 2); + const lastUsage = events[1]; + NodeAssert.equal(lastUsage?.type, "thread.token-usage.updated"); + if (lastUsage?.type !== "thread.token-usage.updated") { + return; + } + // Context usage is unchanged (same input and cache), but the grown + // output and reasoning must still re-emit rather than dedupe away. + NodeAssert.equal(lastUsage.payload.usage.usedTokens, 70); + NodeAssert.equal(lastUsage.payload.usage.outputTokens, 90); + NodeAssert.equal(lastUsage.payload.usage.reasoningOutputTokens, 30); + }), + ); + + it.effect("retries maxTokens resolution after a transient catalog failure", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage-catalog-retry"); + runtimeMock.state.modelListError = new Error("catalog offline"); + const secondSessionUpdate = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + { + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-usage-catalog-retry", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + secondSessionUpdate.promise, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + const first = events[0]; + NodeAssert.equal(first?.type, "thread.token-usage.updated"); + if (first?.type !== "thread.token-usage.updated") { + return; + } + NodeAssert.equal("maxTokens" in first.payload.usage, false); + NodeAssert.equal(runtimeMock.state.modelListCalls, 1); + + // The catalog comes back: a later session.updated must retry the + // resolution instead of giving up for the session's lifetime. + runtimeMock.state.modelListError = null; + runtimeMock.state.modelListResults = [ + { + id: "anthropic", + models: { + "claude-sonnet-4-5": { limit: { context: 200_000 } }, + }, + }, + ]; + const secondEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + secondSessionUpdate.resolve({ + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, + tokens: { input: 60, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }); + + const secondEvents = Array.from( + yield* Fiber.join(secondEventsFiber).pipe(Effect.timeout("1 second")), + ); + const second = secondEvents[0]; + NodeAssert.equal(second?.type, "thread.token-usage.updated"); + if (second?.type !== "thread.token-usage.updated") { + return; + } + NodeAssert.equal(second.payload.usage.maxTokens, 200_000); + NodeAssert.equal(runtimeMock.state.modelListCalls, 2); + }), + ); + + it.effect("keeps per-message usage when a zero-token message arrives", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage-zero-token"); + runtimeMock.state.subscribedEvents = [ + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-usage-real", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-usage-zero", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + }, + }, + }, + { + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, + tokens: { input: 60, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + { + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + title: "zero-token-marker", + }, + }, + }, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil( + (event) => + event.type === "thread.metadata.updated" && event.payload.name === "zero-token-marker", + ), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + const usageEvents = events.filter((event) => event.type === "thread.token-usage.updated"); + // The zero-token message must not clobber per-message usage: the + // session update after it still re-emits against the real message. + NodeAssert.equal(usageEvents.length, 2); + const lastUsage = usageEvents[usageEvents.length - 1]; + NodeAssert.equal(lastUsage?.type, "thread.token-usage.updated"); + if (lastUsage?.type !== "thread.token-usage.updated") { + return; + } + NodeAssert.equal(lastUsage.payload.usage.usedTokens, 70); + }), + ); + + it.effect("emits usage without maxTokens when the model catalog is unavailable", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage-no-catalog"); + runtimeMock.state.modelListError = new Error("catalog offline"); + runtimeMock.state.subscribedEvents = [ + { + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-usage-no-catalog", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + const usageEvent = events[0]; + NodeAssert.equal(usageEvent?.type, "thread.token-usage.updated"); + if (usageEvent?.type !== "thread.token-usage.updated") { + return; + } + NodeAssert.equal("maxTokens" in usageEvent.payload.usage, false); + NodeAssert.equal(usageEvent.payload.usage.usedTokens, 70); + }), + ); + + it.effect("re-emits usage once the catalog resolves maxTokens after a message", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage-late-catalog"); + runtimeMock.state.modelListResults = [ + { + id: "anthropic", + models: { + "claude-sonnet-4-5": { limit: { context: 200_000 } }, + }, + }, + ]; + runtimeMock.state.subscribedEvents = [ + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-usage-late-catalog", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + { + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + NodeAssert.equal(events.length, 2); + const [first, second] = events; + NodeAssert.equal(first?.type, "thread.token-usage.updated"); + NodeAssert.equal(second?.type, "thread.token-usage.updated"); + if ( + first?.type !== "thread.token-usage.updated" || + second?.type !== "thread.token-usage.updated" + ) { + return; + } + NodeAssert.equal("maxTokens" in first.payload.usage, false); + NodeAssert.equal(second.payload.usage.maxTokens, 200_000); + }), + ); + + it.effect("re-resolves maxTokens when the session model changes", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage-model-switch"); + runtimeMock.state.modelListResults = [ + { + id: "anthropic", + models: { + "claude-sonnet-4-5": { limit: { context: 200_000 } }, + }, + }, + { + id: "openai", + models: { + "gpt-5": { limit: { context: 128_000 } }, + }, + }, + ]; + runtimeMock.state.subscribedEvents = [ + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-usage-model-switch", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + { + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + { + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + model: { id: "gpt-5", providerID: "openai" }, + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(3), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + const maxTokensByEvent = events.map((event) => { + if (event.type !== "thread.token-usage.updated") { + return null; + } + return event.payload.usage.maxTokens ?? null; + }); + NodeAssert.deepEqual(maxTokensByEvent, [null, 200_000, 128_000]); + // Resolved once per model, not per event. + NodeAssert.equal(runtimeMock.state.modelListCalls, 2); + }), + ); + it.effect("writes provider-native observability records using the session thread id", () => Effect.gen(function* () { const nativeEvents: Array<{ diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index b10a0c14b2b6..ad4f3fbb58ce 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -8,6 +8,7 @@ import { RuntimeItemId, RuntimeRequestId, ThreadId, + type ThreadTokenUsageSnapshot, type ToolLifecycleItemType, type TurnTokenUsage, TurnId, @@ -88,6 +89,102 @@ function parseOpenCodeResume(raw: unknown): { readonly sessionId: string } | und return { sessionId: record.sessionId.trim() }; } +/** + * Normalized, clamped token counts shared by the two OpenCode usage sources: + * per-message counts on assistant `message.updated` events and cumulative + * per-session counts on `session.updated` events (both carry the same + * non-overlapping shape: `input` excludes cache, `reasoning` is split out of + * `output`). + */ +export interface OpenCodeTokenCounts { + readonly input: number; + readonly output: number; + readonly reasoning: number; + readonly cacheRead: number; + readonly cacheWrite: number; +} + +export function normalizeOpenCodeTokenCounts(tokens: { + readonly input?: number | undefined; + readonly output?: number | undefined; + readonly reasoning?: number | undefined; + readonly cache?: { readonly read?: number | undefined; readonly write?: number | undefined } | undefined; +}): OpenCodeTokenCounts { + // Guard against partial payloads: a missing or non-finite count clamps to + // zero instead of leaking NaN into the usage snapshot (mirrors OpenCode's + // own `safe()` in session.ts getUsage). + const clamp = (value: number | undefined) => { + const n = typeof value === "number" && Number.isFinite(value) ? value : 0; + return Math.max(0, Math.round(n)); + }; + return { + input: clamp(tokens.input), + output: clamp(tokens.output), + reasoning: clamp(tokens.reasoning), + cacheRead: clamp(tokens.cache?.read), + cacheWrite: clamp(tokens.cache?.write), + }; +} + +/** + * Build the token-usage snapshot the context-window meter renders. OpenCode + * reports non-overlapping counts (input excludes cache), so the + * context-window equivalent of its own `contextTokens` helper is + * `input + cache.read + cache.write` on the latest assistant message; the + * cumulative session totals (which keep growing across compactions) become + * `totalProcessedTokens`, and the cumulative session cost (USD) is surfaced + * as-is. Returns undefined until an assistant message reports usage — a + * session with nothing processed has no meter. + */ +export function normalizeOpenCodeTokenUsage(input: { + readonly messageTokens?: OpenCodeTokenCounts | undefined; + readonly cumulativeTokens?: OpenCodeTokenCounts | undefined; + readonly maxTokens?: number | undefined; + readonly cumulativeSessionCost?: number | undefined; +}): ThreadTokenUsageSnapshot | undefined { + const { messageTokens, cumulativeTokens, maxTokens, cumulativeSessionCost } = input; + if (!messageTokens) { + return undefined; + } + + const usedTokens = messageTokens.input + messageTokens.cacheRead + messageTokens.cacheWrite; + if (usedTokens <= 0) { + return undefined; + } + + const inputTokens = usedTokens; + const cachedInputTokens = messageTokens.cacheRead + messageTokens.cacheWrite; + const outputTokens = messageTokens.output; + const reasoningOutputTokens = messageTokens.reasoning; + const totalProcessedTokens = cumulativeTokens + ? cumulativeTokens.input + + cumulativeTokens.output + + cumulativeTokens.reasoning + + cumulativeTokens.cacheRead + + cumulativeTokens.cacheWrite + : undefined; + + return { + usedTokens, + ...(totalProcessedTokens !== undefined && totalProcessedTokens > usedTokens + ? { totalProcessedTokens } + : {}), + ...(maxTokens !== undefined && maxTokens > 0 ? { maxTokens } : {}), + ...(cumulativeSessionCost !== undefined && cumulativeSessionCost > 0 + ? { cost: cumulativeSessionCost } + : {}), + inputTokens, + cachedInputTokens, + ...(outputTokens > 0 ? { outputTokens } : {}), + ...(reasoningOutputTokens > 0 ? { reasoningOutputTokens } : {}), + lastUsedTokens: usedTokens, + lastInputTokens: inputTokens, + lastCachedInputTokens: cachedInputTokens, + ...(outputTokens > 0 ? { lastOutputTokens: outputTokens } : {}), + ...(reasoningOutputTokens > 0 ? { lastReasoningOutputTokens: reasoningOutputTokens } : {}), + }; +} + /** * Whether an error definitively reports a missing session. Only a confirmed * miss may silently start a fresh session; any other failure (the SDK client @@ -339,6 +436,33 @@ interface OpenCodeSessionContext { readonly partById: Map; readonly emittedTextByPartId: Map; readonly completedAssistantPartIds: Set; + /** + * Token counts from the latest assistant `message.updated`. The current + * context-window usage derives from this (matches OpenCode's own + * `contextTokens` definition for the latest assistant message). + */ + lastMessageTokenCounts: OpenCodeTokenCounts | undefined; + /** + * Cumulative session token counts from `session.updated`. Folded into the + * usage snapshot as `totalProcessedTokens`, which keeps growing across + * compactions. + */ + cumulativeTokenCounts: OpenCodeTokenCounts | undefined; + /** + * Cumulative session cost in USD from `session.updated`, surfaced on the + * context-window meter next to the token counts. + */ + cumulativeSessionCost: number | undefined; + /** + * Context-window size resolved once per model from the catalog + * (`v2/model` list). Unset until resolved or when the catalog is + * unavailable — the meter then renders the raw token count. + */ + maxTokens: number | undefined; + /** Model key the resolved `maxTokens` belongs to; re-resolve on change. */ + maxTokensModelKey: string | undefined; + /** Dedupe key of the last emitted `thread.token-usage.updated`. */ + lastEmittedTokenUsageKey: string | undefined; turnTokenUsage: OpenCodeTurnTokenUsageAccumulator | undefined; activeTurnId: TurnId | undefined; activeAgent: string | undefined; @@ -1632,6 +1756,107 @@ export function makeOpenCodeAdapter( } }); + /** + * Resolve the session model's context-window size from the provider + * catalog once per model, best-effort: an unknown model leaves `maxTokens` + * unset and the meter renders the raw token count. A failed catalog call + * does not count as resolved, so a transient failure retries on the next + * `session.updated`. Uses the same v1 `provider.list` endpoint the runtime + * already relies on for inventory, so the response shape is proven. + */ + const resolveOpenCodeMaxTokens = Effect.fn("resolveOpenCodeMaxTokens")(function* ( + context: OpenCodeSessionContext, + providerID: string | undefined, + modelID: string | undefined, + ) { + if (!providerID || !modelID) { + return; + } + const modelKey = `${providerID}/${modelID}`; + if (context.maxTokensModelKey === modelKey) { + return; + } + const outcome = yield* runOpenCodeSdk("provider.list", () => + context.client.provider.list(), + ).pipe( + Effect.map((response) => { + const providers = response.data?.all; + if (!Array.isArray(providers)) { + return undefined; + } + const provider = providers.find((entry) => entry.id === providerID); + const limit = provider?.models?.[modelID]?.limit?.context; + return typeof limit === "number" && Number.isFinite(limit) && limit > 0 + ? Math.round(limit) + : undefined; + }), + Effect.exit, + ); + if (Exit.isSuccess(outcome)) { + context.maxTokensModelKey = modelKey; + context.maxTokens = outcome.value; + } + }); + + /** + * Emit `thread.token-usage.updated` from the latest per-message and + * cumulative session token counts. Deduped so unchanged `session.updated` + * replays don't spam the event log; the key covers every input that + * shapes the snapshot (per-message and cumulative counts, resolved + * maxTokens, session cost), so any change to the rendered snapshot + * re-emits. The context-window meter reads the newest activity, so a + * re-emission after `maxTokens` resolves simply refreshes the snapshot. + */ + const emitOpenCodeTokenUsage = Effect.fn("emitOpenCodeTokenUsage")(function* ( + context: OpenCodeSessionContext, + options?: { readonly turnId?: TurnId | undefined; readonly raw?: unknown }, + ) { + const messageTokens = context.lastMessageTokenCounts; + if (!messageTokens) { + return; + } + const usage = normalizeOpenCodeTokenUsage({ + messageTokens, + cumulativeTokens: context.cumulativeTokenCounts, + maxTokens: context.maxTokens, + cumulativeSessionCost: context.cumulativeSessionCost, + }); + if (!usage) { + return; + } + const cumulative = context.cumulativeTokenCounts; + // Key over every input that shapes the snapshot. -1 marks an absent + // value, so "not reported" and a literal zero are distinct (a session + // whose cost report goes away must re-emit rather than be suppressed). + const dedupeKey = [ + messageTokens.input, + messageTokens.output, + messageTokens.reasoning, + messageTokens.cacheRead, + messageTokens.cacheWrite, + cumulative?.input ?? -1, + cumulative?.output ?? -1, + cumulative?.reasoning ?? -1, + cumulative?.cacheRead ?? -1, + cumulative?.cacheWrite ?? -1, + context.maxTokens ?? -1, + context.cumulativeSessionCost ?? -1, + ].join(","); + if (dedupeKey === context.lastEmittedTokenUsageKey) { + return; + } + context.lastEmittedTokenUsageKey = dedupeKey; + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + ...(options?.turnId ? { turnId: options.turnId } : {}), + ...(options?.raw !== undefined ? { raw: options.raw } : {}), + })), + type: "thread.token-usage.updated", + payload: { usage }, + }); + }); + // Records a child session of this thread. A child seen during a live turn // means that turn used subagents, whether the relation came from a // `session.created` event or a later ancestry lookup after reconnect. @@ -2261,6 +2486,7 @@ export function makeOpenCodeAdapter( switch (event.type) { case "session.updated": { + const info = event.properties.info; const title = openCodeEventSessionTitle(event); if (title) { yield* emit({ @@ -2277,6 +2503,14 @@ export function makeOpenCodeAdapter( }, }); } + if (info.tokens) { + context.cumulativeTokenCounts = normalizeOpenCodeTokenCounts(info.tokens); + } + if (typeof info.cost === "number" && Number.isFinite(info.cost)) { + context.cumulativeSessionCost = info.cost; + } + yield* resolveOpenCodeMaxTokens(context, info.model?.providerID, info.model?.id); + yield* emitOpenCodeTokenUsage(context, { raw: event }); break; } case "session.compacted": { @@ -2296,11 +2530,9 @@ export function makeOpenCodeAdapter( } case "message.updated": { + const info = event.properties.info; const promptAdmission = context.promptAdmission; - if ( - event.properties.info.role === "user" && - promptAdmission?.messageId === event.properties.info.id - ) { + if (info.role === "user" && promptAdmission?.messageId === info.id) { promptAdmission.messageObserved = true; if (promptAdmission.accepted) { const idle = promptAdmission.idleDuringAdmission; @@ -2314,13 +2546,22 @@ export function makeOpenCodeAdapter( } } } - context.messageRoleById.set(event.properties.info.id, event.properties.info.role); - if (event.properties.info.role === "assistant") { + context.messageRoleById.set(info.id, info.role); + if (info.role === "assistant") { + if (info.tokens) { + const counts = normalizeOpenCodeTokenCounts(info.tokens); + // Zero-token updates (e.g. compaction summary messages) must + // not clobber the meter's per-message usage with nothing; keep + // the last assistant message that actually reported usage. + if (counts.input + counts.cacheRead + counts.cacheWrite > 0) { + context.lastMessageTokenCounts = counts; + yield* emitOpenCodeTokenUsage(context, { turnId, raw: event }); + } + } const usage = context.turnTokenUsage; const parentMessageId = - typeof event.properties.info.parentID === "string" && - event.properties.info.parentID.trim().length > 0 - ? event.properties.info.parentID + typeof info.parentID === "string" && info.parentID.trim().length > 0 + ? info.parentID : undefined; const observedOwnership = parentMessageId === undefined @@ -2328,18 +2569,16 @@ export function makeOpenCodeAdapter( : usage?.promptMessageIds.has(parentMessageId) ? "owned" : "other"; - const priorOwnership = usage?.assistantOwnershipByMessageId.get( - event.properties.info.id, - ); + const priorOwnership = usage?.assistantOwnershipByMessageId.get(info.id); const ownership = priorOwnership === undefined || priorOwnership === "unknown" ? observedOwnership : priorOwnership; if (usage) { - usage.assistantOwnershipByMessageId.set(event.properties.info.id, ownership); + usage.assistantOwnershipByMessageId.set(info.id, ownership); } for (const part of context.partById.values()) { - if (part.messageID !== event.properties.info.id) { + if (part.messageID !== info.id) { continue; } if (usage && part.type === "step-finish") { @@ -2967,6 +3206,12 @@ export function makeOpenCodeAdapter( emittedTextByPartId: new Map(), messageRoleById: new Map(), completedAssistantPartIds: new Set(), + lastMessageTokenCounts: undefined, + cumulativeTokenCounts: undefined, + cumulativeSessionCost: undefined, + maxTokens: undefined, + maxTokensModelKey: undefined, + lastEmittedTokenUsageKey: undefined, turnTokenUsage: undefined, activeTurnId: undefined, activeAgent: undefined, diff --git a/apps/web/src/components/chat/ContextWindowMeter.test.tsx b/apps/web/src/components/chat/ContextWindowMeter.test.tsx index 90617ca63af1..d9ad3de7997e 100644 --- a/apps/web/src/components/chat/ContextWindowMeter.test.tsx +++ b/apps/web/src/components/chat/ContextWindowMeter.test.tsx @@ -20,7 +20,7 @@ const usage = deriveLatestContextWindowSnapshot([ tone: "info", kind: "context-window.updated", summary: "Context updated", - payload: { usedTokens: 100_000, maxTokens: 1_000_000 }, + payload: { usedTokens: 100_000, maxTokens: 1_000_000, cost: 0.42 }, turnId: TurnId.make("turn-1"), createdAt: "2026-08-24T12:00:00.000Z", }, @@ -59,4 +59,11 @@ describe("ContextWindowMeter", () => { expect(markup).toContain(">Send or clear your draft before compacting<"); expect(markup).not.toContain('aria-label="Send or clear your draft before compacting"'); }); + + it("shows the cumulative session cost when reported", () => { + const markup = renderToStaticMarkup(); + + expect(markup).toContain(">Cost"); + expect(markup).toContain(">$0.42"); + }); }); diff --git a/apps/web/src/components/chat/ContextWindowMeter.tsx b/apps/web/src/components/chat/ContextWindowMeter.tsx index a11f400c0627..ccc21fff61e0 100644 --- a/apps/web/src/components/chat/ContextWindowMeter.tsx +++ b/apps/web/src/components/chat/ContextWindowMeter.tsx @@ -1,5 +1,9 @@ import { Button } from "../ui/button"; -import { type ContextWindowSnapshot, formatContextWindowTokens } from "~/lib/contextWindow"; +import { + type ContextWindowSnapshot, + formatContextWindowCost, + formatContextWindowTokens, +} from "~/lib/contextWindow"; import { Popover, PopoverPopup, PopoverTrigger } from "../ui/popover"; import { formatContextWindowCompactionMessage } from "./ContextWindowMeter.logic"; import { Minimize2Icon } from "lucide-react"; @@ -30,6 +34,8 @@ export function ContextWindowMeter(props: { const dashOffset = circumference * (1 - normalizedPercentage / 100); const totalProcessedTokens = usage.totalProcessedTokens ?? null; const showTotalProcessed = totalProcessedTokens !== null && totalProcessedTokens > 0; + const cost = usage.cost ?? null; + const showCost = cost !== null && cost > 0; const isOverloaded = normalizedPercentage > 90; const usageColor = isOverloaded ? "var(--color-error)" @@ -132,6 +138,14 @@ export function ContextWindowMeter(props: { ) : null} + {showCost ? ( +
+ Cost + + {formatContextWindowCost(cost)} + +
+ ) : null} {usage.compactsAutomatically ? (
{formatContextWindowCompactionMessage(modelDisplayName, usage.autoCompactThreshold)} diff --git a/apps/web/src/lib/contextWindow.ts b/apps/web/src/lib/contextWindow.ts index 4e2ee139cd85..44ffc9258f1e 100644 --- a/apps/web/src/lib/contextWindow.ts +++ b/apps/web/src/lib/contextWindow.ts @@ -63,6 +63,7 @@ export function deriveLatestContextWindowSnapshot( lastCachedInputTokens: asFiniteNumber(payload?.lastCachedInputTokens), lastOutputTokens: asFiniteNumber(payload?.lastOutputTokens), lastReasoningOutputTokens: asFiniteNumber(payload?.lastReasoningOutputTokens), + cost: asFiniteNumber(payload?.cost), toolUses: asFiniteNumber(payload?.toolUses), durationMs: asFiniteNumber(payload?.durationMs), compactsAutomatically: asBoolean(payload?.compactsAutomatically) ?? false, @@ -89,3 +90,13 @@ export function formatContextWindowTokens(value: number | null): string { } return `${(value / 1_000_000).toFixed(1).replace(/\.0$/, "")}m`; } + +export function formatContextWindowCost(value: number | null): string { + if (value === null || !Number.isFinite(value) || value <= 0) { + return "$0.00"; + } + if (value < 0.01) { + return `$${value.toFixed(4)}`; + } + return `$${value.toFixed(2)}`; +} diff --git a/packages/contracts/src/providerRuntime.ts b/packages/contracts/src/providerRuntime.ts index af1baac74f9d..a51596b3c6d0 100644 --- a/packages/contracts/src/providerRuntime.ts +++ b/packages/contracts/src/providerRuntime.ts @@ -329,6 +329,11 @@ export const ThreadTokenUsageSnapshot = Schema.Struct({ durationMs: Schema.optional(NonNegativeInt), compactsAutomatically: Schema.optional(Schema.Boolean), autoCompactThreshold: Schema.optional(PositiveInt), + /** + * Cumulative session cost in USD. Only OpenCode reports it today + * (session cost is not part of the Codex or Claude notification shapes). + */ + cost: Schema.optional(Schema.Finite.check(Schema.isGreaterThanOrEqualTo(0))), }); export type ThreadTokenUsageSnapshot = typeof ThreadTokenUsageSnapshot.Type; From 641c58ce916f52225e7d906e1337466b0ea3caaf Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 29 Aug 2026 13:20:26 -0400 Subject: [PATCH 02/12] fix(opencode): bound provider.list catalog lookup with a 5s timeout --- .../server/src/provider/Layers/OpenCodeAdapter.ts | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index ad4f3fbb58ce..0756f13652c1 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -1759,10 +1759,13 @@ export function makeOpenCodeAdapter( /** * Resolve the session model's context-window size from the provider * catalog once per model, best-effort: an unknown model leaves `maxTokens` - * unset and the meter renders the raw token count. A failed catalog call - * does not count as resolved, so a transient failure retries on the next - * `session.updated`. Uses the same v1 `provider.list` endpoint the runtime - * already relies on for inventory, so the response shape is proven. + * unset and the meter renders the raw token count. A failed or timed-out + * catalog call does not count as resolved, so a transient failure retries + * on the next `session.updated`. Uses the same v1 `provider.list` endpoint + * the runtime already relies on for inventory, so the response shape is + * proven. Bounded by a short timeout because it runs on the event-pump + * path: a wedged catalog request must leave `maxTokens` unresolved rather + * than stall assistant deltas, requests, and turn completions. */ const resolveOpenCodeMaxTokens = Effect.fn("resolveOpenCodeMaxTokens")(function* ( context: OpenCodeSessionContext, @@ -1790,6 +1793,10 @@ export function makeOpenCodeAdapter( ? Math.round(limit) : undefined; }), + // Event-pump path: never let an unresolved catalog request hold the + // pump hostage. The timeout surfaces as a failure, which leaves + // `maxTokensModelKey` unset so the next `session.updated` retries. + Effect.timeout("5 seconds"), Effect.exit, ); if (Exit.isSuccess(outcome)) { From d26f87325e9e78c5afd94b08ed1de1989f7652e2 Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 29 Aug 2026 13:29:52 -0400 Subject: [PATCH 03/12] fix(opencode): drop stale maxTokens on model switch when catalog lookup fails --- .../provider/Layers/OpenCodeAdapter.test.ts | 139 ++++++++++++++++++ .../src/provider/Layers/OpenCodeAdapter.ts | 20 ++- 2 files changed, 154 insertions(+), 5 deletions(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index d59f4fc7b956..e7f38b1c9f30 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -7793,6 +7793,145 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect( + "drops the previous model's maxTokens when a switch fails to resolve, and re-resolves on switch-back", + () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage-model-switch-failure"); + runtimeMock.state.modelListResults = [ + { + id: "anthropic", + models: { + "claude-sonnet-4-5": { limit: { context: 200_000 } }, + }, + }, + ]; + const switchedSessionUpdate = promiseWithResolvers(); + const switchBackSessionUpdate = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + { + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg-usage-switch-failure", + role: "assistant", + modelID: "claude-sonnet-4-5", + providerID: "anthropic", + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + { + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }, + switchedSessionUpdate.promise, + switchBackSessionUpdate.promise, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + NodeAssert.equal(events.length, 2); + const resolved = events[1]; + NodeAssert.equal(resolved?.type, "thread.token-usage.updated"); + if (resolved?.type !== "thread.token-usage.updated") { + return; + } + NodeAssert.equal(resolved.payload.usage.maxTokens, 200_000); + + // Switch to model B while the catalog is down: the emitted snapshot + // must drop model A's capacity rather than keep reporting it. + runtimeMock.state.modelListError = new Error("catalog offline"); + const failureEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + switchedSessionUpdate.resolve({ + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + model: { id: "gpt-5", providerID: "openai" }, + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }); + + const failureEvents = Array.from( + yield* Fiber.join(failureEventsFiber).pipe(Effect.timeout("1 second")), + ); + const failed = failureEvents[0]; + NodeAssert.equal(failed?.type, "thread.token-usage.updated"); + if (failed?.type !== "thread.token-usage.updated") { + return; + } + NodeAssert.equal("maxTokens" in failed.payload.usage, false); + NodeAssert.equal(runtimeMock.state.modelListCalls, 2); + + // Switching back to model A must re-resolve instead of trusting the + // stale model key from the failed switch. + runtimeMock.state.modelListError = null; + const switchBackEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + switchBackSessionUpdate.resolve({ + type: "session.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "http://127.0.0.1:9999/session", + model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, + tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }, + }, + }); + + const switchBackEvents = Array.from( + yield* Fiber.join(switchBackEventsFiber).pipe(Effect.timeout("1 second")), + ); + const switchBack = switchBackEvents[0]; + NodeAssert.equal(switchBack?.type, "thread.token-usage.updated"); + if (switchBack?.type !== "thread.token-usage.updated") { + return; + } + NodeAssert.equal(switchBack.payload.usage.maxTokens, 200_000); + NodeAssert.equal(runtimeMock.state.modelListCalls, 3); + }), + ); + it.effect("writes provider-native observability records using the session thread id", () => Effect.gen(function* () { const nativeEvents: Array<{ diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 0756f13652c1..1f72625fc9ad 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -455,8 +455,9 @@ interface OpenCodeSessionContext { cumulativeSessionCost: number | undefined; /** * Context-window size resolved once per model from the catalog - * (`v2/model` list). Unset until resolved or when the catalog is - * unavailable — the meter then renders the raw token count. + * (`v2/model` list). Unset until resolved, when the catalog is + * unavailable, or after a model switch until the new model's limit is + * resolved — the meter then renders the raw token count. */ maxTokens: number | undefined; /** Model key the resolved `maxTokens` belongs to; re-resolve on change. */ @@ -1765,7 +1766,10 @@ export function makeOpenCodeAdapter( * the runtime already relies on for inventory, so the response shape is * proven. Bounded by a short timeout because it runs on the event-pump * path: a wedged catalog request must leave `maxTokens` unresolved rather - * than stall assistant deltas, requests, and turn completions. + * than stall assistant deltas, requests, and turn completions. A model + * switch drops the previously resolved limit (and its model key) up + * front, so a failed lookup reports no capacity rather than the previous + * model's, and switching back to the old model re-resolves it. */ const resolveOpenCodeMaxTokens = Effect.fn("resolveOpenCodeMaxTokens")(function* ( context: OpenCodeSessionContext, @@ -1779,6 +1783,12 @@ export function makeOpenCodeAdapter( if (context.maxTokensModelKey === modelKey) { return; } + // A model switch invalidates the previous resolution: the old limit is + // another model's capacity and must not leak into the next emitted + // snapshot. Dropping the model key too keeps the switch-back case + // honest — it re-resolves instead of early-returning on a stale key. + context.maxTokens = undefined; + context.maxTokensModelKey = undefined; const outcome = yield* runOpenCodeSdk("provider.list", () => context.client.provider.list(), ).pipe( @@ -1794,8 +1804,8 @@ export function makeOpenCodeAdapter( : undefined; }), // Event-pump path: never let an unresolved catalog request hold the - // pump hostage. The timeout surfaces as a failure, which leaves - // `maxTokensModelKey` unset so the next `session.updated` retries. + // pump hostage. The timeout surfaces as a failure, which leaves the + // model unresolved so the next `session.updated` retries. Effect.timeout("5 seconds"), Effect.exit, ); From ea8606919265fcd86f77c421e907fb30988be2f6 Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 29 Aug 2026 13:49:33 -0400 Subject: [PATCH 04/12] test(opencode): consolidate usage test fixtures into helpers, merge redundant scenarios --- .../provider/Layers/OpenCodeAdapter.test.ts | 676 +++++------------- 1 file changed, 162 insertions(+), 514 deletions(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index e7f38b1c9f30..2bfe4578510d 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -7183,61 +7183,88 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + // ---- Context-window usage fixtures -------------------------------------- + // + // The usage integration tests all drive the same two OpenCode events with + // small variations, then collect `thread.token-usage.updated` emissions. + // These helpers keep each scenario down to its actual variations. + + const usageSessionId = "http://127.0.0.1:9999/session"; + const usageModel = { id: "claude-sonnet-4-5", providerID: "anthropic" }; + const usageTokens = { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }; + const usageCatalog = [ + { id: "anthropic", models: { "claude-sonnet-4-5": { limit: { context: 200_000 } } } }, + ]; + + const usageAssistantMessage = (id: string, tokens = usageTokens) => ({ + type: "message.updated", + properties: { + sessionID: usageSessionId, + info: { + id, + role: "assistant", + modelID: usageModel.id, + providerID: usageModel.providerID, + tokens, + }, + }, + }); + + const usageSessionUpdate = ({ + model = usageModel, + tokens = usageTokens, + ...rest + }: { + readonly model?: Record | undefined; + readonly tokens?: Record | undefined; + } & Record = {}) => ({ + type: "session.updated", + properties: { + sessionID: usageSessionId, + info: { id: usageSessionId, model, tokens, ...rest }, + }, + }); + + // Session update carrying only a title — used as a drain marker whose + // metadata event proves earlier events finished processing. + const usageMarkerSessionUpdate = (title: string) => ({ + type: "session.updated", + properties: { + sessionID: usageSessionId, + info: { id: usageSessionId, title }, + }, + }); + + const startUsageSession = (adapter: OpenCodeAdapterShape, threadId: ThreadId) => + adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const collectUsageEvents = (adapter: OpenCodeAdapterShape, threadId: ThreadId, take: number) => + adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(take), + Stream.runCollect, + Effect.forkChild, + ); + it.effect("emits context-window usage from assistant messages and session totals", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; const threadId = asThreadId("thread-opencode-usage"); - runtimeMock.state.modelListResults = [ - { - id: "anthropic", - models: { - "claude-sonnet-4-5": { limit: { context: 200_000 } }, - }, - }, - ]; + runtimeMock.state.modelListResults = usageCatalog; runtimeMock.state.subscribedEvents = [ - { - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - title: "Usage thread", - model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - cost: 0.42, - }, - }, - }, - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-usage-1", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, + usageSessionUpdate({ cost: 0.42 }), + usageAssistantMessage("msg-usage-1"), ]; - const eventsFiber = yield* adapter.streamEvents.pipe( - Stream.filter( - (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", - ), - Stream.take(1), - Stream.runCollect, - Effect.forkChild, - ); + const eventsFiber = yield* collectUsageEvents(adapter, threadId, 1); - yield* adapter.startSession({ - provider: ProviderDriverKind.make("opencode"), - threadId, - runtimeMode: "full-access", - }); + yield* startUsageSession(adapter, threadId); const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); const usageEvent = events[0]; @@ -7273,41 +7300,13 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; const threadId = asThreadId("thread-opencode-usage-dedupe"); - const assistantMessage = { - id: "msg-usage-dedupe", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }; runtimeMock.state.subscribedEvents = [ - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: assistantMessage, - }, - }, - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { ...assistantMessage, id: "msg-usage-dedupe-2" }, - }, - }, + usageAssistantMessage("msg-usage-dedupe"), + usageAssistantMessage("msg-usage-dedupe-2"), // Drain marker: its metadata event is emitted only after both // messages have been fully processed, so the collected window is // complete regardless of how many ambient events they produce. - { - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - title: "dedupe-marker", - }, - }, - }, + usageMarkerSessionUpdate("dedupe-marker"), ]; const eventsFiber = yield* adapter.streamEvents.pipe( @@ -7337,48 +7336,18 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const adapter = yield* OpenCodeAdapter; const threadId = asThreadId("thread-opencode-usage-output-growth"); runtimeMock.state.subscribedEvents = [ - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-output-growth-1", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-output-growth-2", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 50, output: 90, reasoning: 30, cache: { read: 20, write: 0 } }, - }, - }, - }, + usageAssistantMessage("msg-output-growth-1"), + usageAssistantMessage("msg-output-growth-2", { + input: 50, + output: 90, + reasoning: 30, + cache: { read: 20, write: 0 }, + }), ]; - const eventsFiber = yield* adapter.streamEvents.pipe( - Stream.filter( - (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", - ), - Stream.take(2), - Stream.runCollect, - Effect.forkChild, - ); + const eventsFiber = yield* collectUsageEvents(adapter, threadId, 2); - yield* adapter.startSession({ - provider: ProviderDriverKind.make("opencode"), - threadId, - runtimeMode: "full-access", - }); + yield* startUsageSession(adapter, threadId); const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); NodeAssert.equal(events.length, 2); @@ -7395,106 +7364,60 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); - it.effect("retries maxTokens resolution after a transient catalog failure", () => - Effect.gen(function* () { - const adapter = yield* OpenCodeAdapter; - const threadId = asThreadId("thread-opencode-usage-catalog-retry"); - runtimeMock.state.modelListError = new Error("catalog offline"); - const secondSessionUpdate = promiseWithResolvers(); - runtimeMock.state.subscribedEvents = [ - { - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-usage-catalog-retry", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - secondSessionUpdate.promise, - ]; - - const eventsFiber = yield* adapter.streamEvents.pipe( - Stream.filter( - (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", - ), - Stream.take(1), - Stream.runCollect, - Effect.forkChild, - ); + it.effect( + "emits usage without maxTokens while the catalog is unavailable, then retries once it resolves", + () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage-catalog-retry"); + runtimeMock.state.modelListError = new Error("catalog offline"); + const recoveredSessionUpdate = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + usageAssistantMessage("msg-usage-catalog-retry"), + usageSessionUpdate(), + recoveredSessionUpdate.promise, + ]; - yield* adapter.startSession({ - provider: ProviderDriverKind.make("opencode"), - threadId, - runtimeMode: "full-access", - }); + const eventsFiber = yield* collectUsageEvents(adapter, threadId, 1); - const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); - const first = events[0]; - NodeAssert.equal(first?.type, "thread.token-usage.updated"); - if (first?.type !== "thread.token-usage.updated") { - return; - } - NodeAssert.equal("maxTokens" in first.payload.usage, false); - NodeAssert.equal(runtimeMock.state.modelListCalls, 1); + yield* startUsageSession(adapter, threadId); - // The catalog comes back: a later session.updated must retry the - // resolution instead of giving up for the session's lifetime. - runtimeMock.state.modelListError = null; - runtimeMock.state.modelListResults = [ - { - id: "anthropic", - models: { - "claude-sonnet-4-5": { limit: { context: 200_000 } }, - }, - }, - ]; - const secondEventsFiber = yield* adapter.streamEvents.pipe( - Stream.filter( - (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", - ), - Stream.take(1), - Stream.runCollect, - Effect.forkChild, - ); - secondSessionUpdate.resolve({ - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + const first = events[0]; + NodeAssert.equal(first?.type, "thread.token-usage.updated"); + if (first?.type !== "thread.token-usage.updated") { + return; + } + // The meter still renders the raw token count while the catalog is + // down. (The message-before-session ordering that delays the first + // resolution with a healthy catalog is covered by the model-switch + // test below.) + NodeAssert.equal("maxTokens" in first.payload.usage, false); + NodeAssert.equal(first.payload.usage.usedTokens, 70); + NodeAssert.equal(runtimeMock.state.modelListCalls, 1); + + // The catalog comes back: a later session.updated must retry the + // resolution instead of giving up for the session's lifetime. + runtimeMock.state.modelListError = null; + runtimeMock.state.modelListResults = usageCatalog; + const secondEventsFiber = yield* collectUsageEvents(adapter, threadId, 1); + recoveredSessionUpdate.resolve( + usageSessionUpdate({ tokens: { input: 60, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }); + }), + ); - const secondEvents = Array.from( - yield* Fiber.join(secondEventsFiber).pipe(Effect.timeout("1 second")), - ); - const second = secondEvents[0]; - NodeAssert.equal(second?.type, "thread.token-usage.updated"); - if (second?.type !== "thread.token-usage.updated") { - return; - } - NodeAssert.equal(second.payload.usage.maxTokens, 200_000); - NodeAssert.equal(runtimeMock.state.modelListCalls, 2); - }), + const secondEvents = Array.from( + yield* Fiber.join(secondEventsFiber).pipe(Effect.timeout("1 second")), + ); + const second = secondEvents[0]; + NodeAssert.equal(second?.type, "thread.token-usage.updated"); + if (second?.type !== "thread.token-usage.updated") { + return; + } + NodeAssert.equal(second.payload.usage.maxTokens, 200_000); + NodeAssert.equal(runtimeMock.state.modelListCalls, 2); + }), ); it.effect("keeps per-message usage when a zero-token message arrives", () => @@ -7502,53 +7425,17 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const adapter = yield* OpenCodeAdapter; const threadId = asThreadId("thread-opencode-usage-zero-token"); runtimeMock.state.subscribedEvents = [ - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-usage-real", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-usage-zero", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, - }, - }, - }, - { - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, - tokens: { input: 60, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - { - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - title: "zero-token-marker", - }, - }, - }, + usageAssistantMessage("msg-usage-real"), + usageAssistantMessage("msg-usage-zero", { + input: 0, + output: 0, + reasoning: 0, + cache: { read: 0, write: 0 }, + }), + usageSessionUpdate({ + tokens: { input: 60, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }), + usageMarkerSessionUpdate("zero-token-marker"), ]; const eventsFiber = yield* adapter.streamEvents.pipe( @@ -7581,204 +7468,23 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); - it.effect("emits usage without maxTokens when the model catalog is unavailable", () => - Effect.gen(function* () { - const adapter = yield* OpenCodeAdapter; - const threadId = asThreadId("thread-opencode-usage-no-catalog"); - runtimeMock.state.modelListError = new Error("catalog offline"); - runtimeMock.state.subscribedEvents = [ - { - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-usage-no-catalog", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - ]; - - const eventsFiber = yield* adapter.streamEvents.pipe( - Stream.filter( - (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", - ), - Stream.take(1), - Stream.runCollect, - Effect.forkChild, - ); - - yield* adapter.startSession({ - provider: ProviderDriverKind.make("opencode"), - threadId, - runtimeMode: "full-access", - }); - - const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); - const usageEvent = events[0]; - NodeAssert.equal(usageEvent?.type, "thread.token-usage.updated"); - if (usageEvent?.type !== "thread.token-usage.updated") { - return; - } - NodeAssert.equal("maxTokens" in usageEvent.payload.usage, false); - NodeAssert.equal(usageEvent.payload.usage.usedTokens, 70); - }), - ); - - it.effect("re-emits usage once the catalog resolves maxTokens after a message", () => - Effect.gen(function* () { - const adapter = yield* OpenCodeAdapter; - const threadId = asThreadId("thread-opencode-usage-late-catalog"); - runtimeMock.state.modelListResults = [ - { - id: "anthropic", - models: { - "claude-sonnet-4-5": { limit: { context: 200_000 } }, - }, - }, - ]; - runtimeMock.state.subscribedEvents = [ - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-usage-late-catalog", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - { - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - ]; - - const eventsFiber = yield* adapter.streamEvents.pipe( - Stream.filter( - (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", - ), - Stream.take(2), - Stream.runCollect, - Effect.forkChild, - ); - - yield* adapter.startSession({ - provider: ProviderDriverKind.make("opencode"), - threadId, - runtimeMode: "full-access", - }); - - const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); - NodeAssert.equal(events.length, 2); - const [first, second] = events; - NodeAssert.equal(first?.type, "thread.token-usage.updated"); - NodeAssert.equal(second?.type, "thread.token-usage.updated"); - if ( - first?.type !== "thread.token-usage.updated" || - second?.type !== "thread.token-usage.updated" - ) { - return; - } - NodeAssert.equal("maxTokens" in first.payload.usage, false); - NodeAssert.equal(second.payload.usage.maxTokens, 200_000); - }), - ); - it.effect("re-resolves maxTokens when the session model changes", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; const threadId = asThreadId("thread-opencode-usage-model-switch"); runtimeMock.state.modelListResults = [ - { - id: "anthropic", - models: { - "claude-sonnet-4-5": { limit: { context: 200_000 } }, - }, - }, - { - id: "openai", - models: { - "gpt-5": { limit: { context: 128_000 } }, - }, - }, + ...usageCatalog, + { id: "openai", models: { "gpt-5": { limit: { context: 128_000 } } } }, ]; runtimeMock.state.subscribedEvents = [ - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-usage-model-switch", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - { - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - { - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - model: { id: "gpt-5", providerID: "openai" }, - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, + usageAssistantMessage("msg-usage-model-switch"), + usageSessionUpdate(), + usageSessionUpdate({ model: { id: "gpt-5", providerID: "openai" } }), ]; - const eventsFiber = yield* adapter.streamEvents.pipe( - Stream.filter( - (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", - ), - Stream.take(3), - Stream.runCollect, - Effect.forkChild, - ); + const eventsFiber = yield* collectUsageEvents(adapter, threadId, 3); - yield* adapter.startSession({ - provider: ProviderDriverKind.make("opencode"), - threadId, - runtimeMode: "full-access", - }); + yield* startUsageSession(adapter, threadId); const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); const maxTokensByEvent = events.map((event) => { @@ -7799,59 +7505,19 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; const threadId = asThreadId("thread-opencode-usage-model-switch-failure"); - runtimeMock.state.modelListResults = [ - { - id: "anthropic", - models: { - "claude-sonnet-4-5": { limit: { context: 200_000 } }, - }, - }, - ]; + runtimeMock.state.modelListResults = usageCatalog; const switchedSessionUpdate = promiseWithResolvers(); const switchBackSessionUpdate = promiseWithResolvers(); runtimeMock.state.subscribedEvents = [ - { - type: "message.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "msg-usage-switch-failure", - role: "assistant", - modelID: "claude-sonnet-4-5", - providerID: "anthropic", - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, - { - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }, + usageAssistantMessage("msg-usage-switch-failure"), + usageSessionUpdate(), switchedSessionUpdate.promise, switchBackSessionUpdate.promise, ]; - const eventsFiber = yield* adapter.streamEvents.pipe( - Stream.filter( - (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", - ), - Stream.take(2), - Stream.runCollect, - Effect.forkChild, - ); + const eventsFiber = yield* collectUsageEvents(adapter, threadId, 2); - yield* adapter.startSession({ - provider: ProviderDriverKind.make("opencode"), - threadId, - runtimeMode: "full-access", - }); + yield* startUsageSession(adapter, threadId); const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); NodeAssert.equal(events.length, 2); @@ -7873,17 +7539,9 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { Stream.runCollect, Effect.forkChild, ); - switchedSessionUpdate.resolve({ - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - model: { id: "gpt-5", providerID: "openai" }, - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }); + switchedSessionUpdate.resolve( + usageSessionUpdate({ model: { id: "gpt-5", providerID: "openai" } }), + ); const failureEvents = Array.from( yield* Fiber.join(failureEventsFiber).pipe(Effect.timeout("1 second")), @@ -7907,17 +7565,7 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { Stream.runCollect, Effect.forkChild, ); - switchBackSessionUpdate.resolve({ - type: "session.updated", - properties: { - sessionID: "http://127.0.0.1:9999/session", - info: { - id: "http://127.0.0.1:9999/session", - model: { id: "claude-sonnet-4-5", providerID: "anthropic" }, - tokens: { input: 50, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, - }, - }, - }); + switchBackSessionUpdate.resolve(usageSessionUpdate()); const switchBackEvents = Array.from( yield* Fiber.join(switchBackEventsFiber).pipe(Effect.timeout("1 second")), From ebf5d7b841766e18544caa05d6aa378ff9156a5a Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 5 Sep 2026 08:47:38 -0400 Subject: [PATCH 05/12] fix(opencode): cool down context-window catalog retries - Gate same-model provider.list retries behind 30s cooldown so replays don't stall pump - Pass abort signal so 5s timeout interrupts wedged catalog fetch --- .../provider/Layers/OpenCodeAdapter.test.ts | 84 ++++++++++++++++++- .../src/provider/Layers/OpenCodeAdapter.ts | 75 ++++++++++++++--- 2 files changed, 142 insertions(+), 17 deletions(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 2bfe4578510d..21c18cf72d4b 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -134,6 +134,7 @@ const runtimeMock = { modelListCalls: 0, modelListResults: [] as Array>, modelListError: null as Error | null, + modelListSignals: [] as Array, }, reset() { this.state.startCalls.length = 0; @@ -192,6 +193,7 @@ const runtimeMock = { this.state.modelListCalls = 0; this.state.modelListResults = []; this.state.modelListError = null; + this.state.modelListSignals.length = 0; }, }; @@ -506,8 +508,9 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { }, }, provider: { - list: async () => { + list: async (_params?: unknown, options?: { signal?: AbortSignal }) => { runtimeMock.state.modelListCalls += 1; + runtimeMock.state.modelListSignals.push(options?.signal); if (runtimeMock.state.modelListError) { throw runtimeMock.state.modelListError; } @@ -7395,11 +7398,19 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { NodeAssert.equal("maxTokens" in first.payload.usage, false); NodeAssert.equal(first.payload.usage.usedTokens, 70); NodeAssert.equal(runtimeMock.state.modelListCalls, 1); - - // The catalog comes back: a later session.updated must retry the - // resolution instead of giving up for the session's lifetime. + // The catalog lookup runs on the sequential event pump, so it must + // carry the abort signal — the 5s timeout interrupts the fetch rather + // than leaving a wedged request behind the next retry. + NodeAssert.ok(runtimeMock.state.modelListSignals[0] instanceof AbortSignal); + + // The catalog comes back: a later session.updated after the retry + // cooldown must retry the resolution instead of giving up for the + // session's lifetime. Same-model failures inside the cooldown window + // skip the bounded lookup so an outage (or a burst of unchanged + // replays) stalls the pump at most once per window. runtimeMock.state.modelListError = null; runtimeMock.state.modelListResults = usageCatalog; + yield* TestClock.adjust("31 seconds"); const secondEventsFiber = yield* collectUsageEvents(adapter, threadId, 1); recoveredSessionUpdate.resolve( usageSessionUpdate({ @@ -7420,6 +7431,71 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("cools down same-model catalog retries so replays do not stall the pump", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-usage-catalog-cooldown"); + runtimeMock.state.modelListError = new Error("catalog offline"); + const replayedSessionUpdate = promiseWithResolvers(); + const recoveredSessionUpdate = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + usageAssistantMessage("msg-usage-catalog-cooldown"), + usageSessionUpdate(), + replayedSessionUpdate.promise, + usageMarkerSessionUpdate("cooldown-marker"), + recoveredSessionUpdate.promise, + ]; + + const eventsFiber = yield* collectUsageEvents(adapter, threadId, 1); + yield* startUsageSession(adapter, threadId); + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + NodeAssert.equal(events.length, 1); + NodeAssert.equal(runtimeMock.state.modelListCalls, 1); + + // An unchanged replay inside the cooldown must not pay another bounded + // lookup on the sequential pump. Drain through the marker (its metadata + // event proves the replay finished processing): no extra catalog call. + const drainFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil( + (event) => + event.type === "thread.metadata.updated" && event.payload.name === "cooldown-marker", + ), + Stream.runCollect, + Effect.forkChild, + ); + replayedSessionUpdate.resolve(usageSessionUpdate()); + const drained = Array.from(yield* Fiber.join(drainFiber).pipe(Effect.timeout("1 second"))); + const drainedUsage = drained.filter((event) => event.type === "thread.token-usage.updated"); + // The replay carries identical tokens, so the deduped meter emits + // nothing new — and the cooldown means no second catalog call either. + NodeAssert.equal(drainedUsage.length, 0); + NodeAssert.equal(runtimeMock.state.modelListCalls, 1); + + // After the cooldown the next session.updated retries and resolves. + runtimeMock.state.modelListError = null; + runtimeMock.state.modelListResults = usageCatalog; + yield* TestClock.adjust("31 seconds"); + const recoveredEventsFiber = yield* collectUsageEvents(adapter, threadId, 1); + recoveredSessionUpdate.resolve( + usageSessionUpdate({ + tokens: { input: 60, output: 10, reasoning: 5, cache: { read: 20, write: 0 } }, + }), + ); + const recoveredEvents = Array.from( + yield* Fiber.join(recoveredEventsFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(recoveredEvents.length, 1); + const recovered = recoveredEvents[0]; + NodeAssert.equal(recovered?.type, "thread.token-usage.updated"); + if (recovered?.type !== "thread.token-usage.updated") { + return; + } + NodeAssert.equal(recovered.payload.usage.maxTokens, 200_000); + NodeAssert.equal(runtimeMock.state.modelListCalls, 2); + }), + ); + it.effect("keeps per-message usage when a zero-token message arrives", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 1f72625fc9ad..2669ccfda238 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -15,6 +15,7 @@ import { type UserInputQuestion, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; @@ -69,6 +70,17 @@ const PROVIDER = ProviderDriverKind.make("opencode"); */ const OPENCODE_RESUME_VERSION = 1 as const; +/** + * Cooldown between failed context-window catalog (`provider.list`) attempts + * for the same model. The lookup runs on the sequential event pump, so an + * outage must not stall deltas, permission asks, and turn completions on + * every `session.updated` replay — one bounded attempt per window is enough, + * and the meter keeps rendering the raw token count meanwhile. A model + * switch bypasses the cooldown because it is a different lookup, not a + * retry of the failed one. + */ +const OPENCODE_CATALOG_RETRY_COOLDOWN_MS = 30_000; + /** * Decode a persisted resume cursor into the upstream `ses_…` id. Anything * that isn't a current-version cursor with a non-empty id means "no resume" @@ -108,7 +120,9 @@ export function normalizeOpenCodeTokenCounts(tokens: { readonly input?: number | undefined; readonly output?: number | undefined; readonly reasoning?: number | undefined; - readonly cache?: { readonly read?: number | undefined; readonly write?: number | undefined } | undefined; + readonly cache?: + | { readonly read?: number | undefined; readonly write?: number | undefined } + | undefined; }): OpenCodeTokenCounts { // Guard against partial payloads: a missing or non-finite count clamps to // zero instead of leaking NaN into the usage snapshot (mirrors OpenCode's @@ -462,6 +476,16 @@ interface OpenCodeSessionContext { maxTokens: number | undefined; /** Model key the resolved `maxTokens` belongs to; re-resolve on change. */ maxTokensModelKey: string | undefined; + /** + * Model key of the last failed catalog attempt, with its wall-clock time. + * Gates retries for the same model behind + * {@link OPENCODE_CATALOG_RETRY_COOLDOWN_MS} so a catalog outage stalls the + * sequential event pump at most once per window instead of once per + * `session.updated` replay. Cleared on success (the resolved key takes + * over) and bypassed on model switch (different model, not a retry). + */ + maxTokensLastAttemptModelKey: string | undefined; + maxTokensLastAttemptMs: number | undefined; /** Dedupe key of the last emitted `thread.token-usage.updated`. */ lastEmittedTokenUsageKey: string | undefined; turnTokenUsage: OpenCodeTurnTokenUsageAccumulator | undefined; @@ -1762,14 +1786,18 @@ export function makeOpenCodeAdapter( * catalog once per model, best-effort: an unknown model leaves `maxTokens` * unset and the meter renders the raw token count. A failed or timed-out * catalog call does not count as resolved, so a transient failure retries - * on the next `session.updated`. Uses the same v1 `provider.list` endpoint - * the runtime already relies on for inventory, so the response shape is - * proven. Bounded by a short timeout because it runs on the event-pump - * path: a wedged catalog request must leave `maxTokens` unresolved rather - * than stall assistant deltas, requests, and turn completions. A model - * switch drops the previously resolved limit (and its model key) up - * front, so a failed lookup reports no capacity rather than the previous - * model's, and switching back to the old model re-resolves it. + * on a later `session.updated` once the cooldown lapses. Uses the same v1 + * `provider.list` endpoint the runtime already relies on for inventory, so + * the response shape is proven. Bounded by a short timeout because it runs + * on the event-pump path: a wedged catalog request must leave `maxTokens` + * unresolved rather than stall assistant deltas, requests, and turn + * completions. Failed attempts for the same model are gated behind a + * cooldown so an outage (or a burst of unchanged `session.updated` + * replays) stalls the sequential pump at most once per window instead of + * once per event. A model switch bypasses the cooldown and drops the + * previously resolved limit (and its model key) up front, so a failed + * lookup reports no capacity rather than the previous model's, and + * switching back to the old model re-resolves it. */ const resolveOpenCodeMaxTokens = Effect.fn("resolveOpenCodeMaxTokens")(function* ( context: OpenCodeSessionContext, @@ -1783,14 +1811,29 @@ export function makeOpenCodeAdapter( if (context.maxTokensModelKey === modelKey) { return; } + const now = yield* Clock.currentTimeMillis; + const lastAttemptMs = context.maxTokensLastAttemptMs; + if ( + context.maxTokensLastAttemptModelKey === modelKey && + lastAttemptMs !== undefined && + now >= lastAttemptMs && + now - lastAttemptMs < OPENCODE_CATALOG_RETRY_COOLDOWN_MS + ) { + // Same model failed recently: leave `maxTokens` unresolved without + // paying another bounded stall on the pump. The next `session.updated` + // after the cooldown retries. + return; + } // A model switch invalidates the previous resolution: the old limit is // another model's capacity and must not leak into the next emitted // snapshot. Dropping the model key too keeps the switch-back case // honest — it re-resolves instead of early-returning on a stale key. context.maxTokens = undefined; context.maxTokensModelKey = undefined; - const outcome = yield* runOpenCodeSdk("provider.list", () => - context.client.provider.list(), + context.maxTokensLastAttemptModelKey = modelKey; + context.maxTokensLastAttemptMs = now; + const outcome = yield* runOpenCodeSdk("provider.list", (signal) => + context.client.provider.list(undefined, { signal }), ).pipe( Effect.map((response) => { const providers = response.data?.all; @@ -1804,14 +1847,18 @@ export function makeOpenCodeAdapter( : undefined; }), // Event-pump path: never let an unresolved catalog request hold the - // pump hostage. The timeout surfaces as a failure, which leaves the - // model unresolved so the next `session.updated` retries. + // pump hostage. The timeout interrupts the SDK fetch via the abort + // signal (so a timed-out request cannot linger and pile up behind the + // next retry) and surfaces as a failure, which leaves the model + // unresolved so a later `session.updated` retries after the cooldown. Effect.timeout("5 seconds"), Effect.exit, ); if (Exit.isSuccess(outcome)) { context.maxTokensModelKey = modelKey; context.maxTokens = outcome.value; + context.maxTokensLastAttemptModelKey = undefined; + context.maxTokensLastAttemptMs = undefined; } }); @@ -3228,6 +3275,8 @@ export function makeOpenCodeAdapter( cumulativeSessionCost: undefined, maxTokens: undefined, maxTokensModelKey: undefined, + maxTokensLastAttemptModelKey: undefined, + maxTokensLastAttemptMs: undefined, lastEmittedTokenUsageKey: undefined, turnTokenUsage: undefined, activeTurnId: undefined, From 9f840eddac3a9100d2ae066b829146b757728fa8 Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 5 Sep 2026 18:11:43 -0400 Subject: [PATCH 06/12] feat(opencode): track per-turn token contributions for usage - Merge per-message token reports with same-id replace and cross-turn fencing - Emit usage, modelUsage, and totalCostUsd on turn.completed - Drop contribution ledger on failed, aborted, or interrupted turns --- .../src/provider/Layers/OpenCodeAdapter.ts | 255 ++++++++++++++++++ 1 file changed, 255 insertions(+) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 2669ccfda238..05838103de55 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -199,6 +199,93 @@ export function normalizeOpenCodeTokenUsage(input: { }; } +/** + * Per-source token report feeding the per-turn contribution ledger (ported + * from #8918's live-usage accounting). Unlike {@link OpenCodeTokenCounts} + * (flat, clamped, meter-oriented), this keeps the nested cache shape and an + * optional `total` so merged turn totals stay faithful to what OpenCode + * reported per message/part. + */ +interface OpenCodeTurnTokens { + readonly input: number; + readonly output: number; + readonly reasoning: number; + readonly cache: { readonly read: number; readonly write: number }; + readonly total?: number | undefined; +} + +function readOpenCodeTurnTokens(value: unknown): OpenCodeTurnTokens | null { + if (typeof value !== "object" || value === null) return null; + const record = value as Record; + const input = typeof record.input === "number" ? record.input : null; + const output = typeof record.output === "number" ? record.output : null; + if (input === null || output === null) return null; + const reasoning = typeof record.reasoning === "number" ? record.reasoning : 0; + const cacheRaw = record.cache; + let cacheRead = 0; + let cacheWrite = 0; + if (typeof cacheRaw === "object" && cacheRaw !== null) { + const cache = cacheRaw as Record; + cacheRead = typeof cache.read === "number" ? cache.read : 0; + cacheWrite = typeof cache.write === "number" ? cache.write : 0; + } + const total = typeof record.total === "number" ? record.total : undefined; + return { + input, + output, + reasoning, + cache: { read: cacheRead, write: cacheWrite }, + ...(total !== undefined ? { total } : {}), + }; +} + +function mergeOpenCodeTurnTokens(a: OpenCodeTurnTokens, b: OpenCodeTurnTokens): OpenCodeTurnTokens { + const totalA = a.total ?? a.input + a.output + a.cache.read + a.cache.write; + const totalB = b.total ?? b.input + b.output + b.cache.read + b.cache.write; + const hasTotal = a.total !== undefined || b.total !== undefined; + return { + input: a.input + b.input, + output: a.output + b.output, + reasoning: a.reasoning + b.reasoning, + cache: { read: a.cache.read + b.cache.read, write: a.cache.write + b.cache.write }, + ...(hasTotal ? { total: totalA + totalB } : {}), + }; +} + +/** + * Model key (`provider/model`) for a token report's per-model bucket. Accepts + * the string and object model shapes OpenCode emits on messages and parts; + * defaults an absent provider to `opencode`. Returns null when no model id is + * present — the report still counts toward turn totals, just not toward any + * per-model bucket. + */ +function openCodeContributionModelKey(model: unknown, providerId: unknown): string | null { + if (typeof model === "string" && model.trim().length > 0) { + const provider = + typeof providerId === "string" && providerId.trim().length > 0 + ? providerId.trim() + : "opencode"; + return `${provider}/${model.trim()}`; + } + if (typeof model === "object" && model !== null) { + const record = model as Record; + const id = + typeof record.id === "string" + ? record.id + : typeof record.modelID === "string" + ? record.modelID + : null; + const provider = + typeof record.providerID === "string" + ? record.providerID + : typeof record.provider === "string" + ? record.provider + : "opencode"; + if (id) return `${provider}/${id}`; + } + return null; +} + /** * Whether an error definitively reports a missing session. Only a confirmed * miss may silently start a fresh session; any other failure (the SDK client @@ -488,6 +575,32 @@ interface OpenCodeSessionContext { maxTokensLastAttemptMs: number | undefined; /** Dedupe key of the last emitted `thread.token-usage.updated`. */ lastEmittedTokenUsageKey: string | undefined; + /** + * Per-turn token contribution ledger (ported from #8918's live-usage + * accounting). Each message/part id contributes one report keyed + * `msg:` / `part:`; a same-id re-emission carries that source's + * updated totals and *replaces* its entry, while distinct ids (tool steps) + * each contribute — totals are merged across entries. This is the + * cumulative-processed view that complements the meter's latest-message + * context-occupancy view; the two are kept distinct on purpose. + */ + turnContributionEntries: Map< + string, + { + readonly tokens: OpenCodeTurnTokens; + readonly cost: number | null; + readonly modelKey: string | null; + } + >; + /** + * Owning turn per contribution key. Late events from a prior turn must not + * contaminate the new turn's totals, so a key already seen under a + * different turn is fenced out. Retained across turns (bounded by + * message/part count) — clearing it would re-admit stale reports as new. + */ + contributionTurnMap: Map; + /** Suffix source for contributions without a stable id. */ + contributionCounter: number; turnTokenUsage: OpenCodeTurnTokenUsageAccumulator | undefined; activeTurnId: TurnId | undefined; activeAgent: string | undefined; @@ -1253,6 +1366,10 @@ export function makeOpenCodeAdapter( context.pendingIdleReconciliation = undefined; } const tokenUsage = takeOpenCodeTurnTokenUsage(context, true); + // Per-turn contribution ledger (ported from #8918): merged message + // reports plus per-model buckets and summed cost for the usage page. + // Empty ledgers (e.g. zero-token-only turns) attach nothing. + const contributions = takeOpenCodeTurnContributions(context); context.activeTurnId = undefined; context.activeAgent = undefined; context.activeVariant = undefined; @@ -1283,6 +1400,15 @@ export function makeOpenCodeAdapter( payload: { state: "completed", tokenUsage, + ...(contributions.usage + ? { + usage: contributions.usage, + ...(contributions.modelUsage ? { modelUsage: contributions.modelUsage } : {}), + } + : {}), + ...(contributions.totalCostUsd !== undefined + ? { totalCostUsd: contributions.totalCostUsd } + : {}), }, }); }); @@ -1421,6 +1547,9 @@ export function makeOpenCodeAdapter( return; } const tokenUsage = takeOpenCodeTurnTokenUsage(context, false); + // The turn ends failed: drop its contribution ledger with it so a + // later turn never inherits these reports. + clearOpenCodeTurnContributions(context); context.promptAdmission = undefined; context.activeTurnId = undefined; context.activeAgent = undefined; @@ -1627,6 +1756,8 @@ export function makeOpenCodeAdapter( }; if (context.activeTurnId === turnId) { tokenUsage = takeOpenCodeTurnTokenUsage(context, false); + // The turn ends interrupted: drop its contribution ledger with it. + clearOpenCodeTurnContributions(context); context.activeTurnId = undefined; context.activeAgent = undefined; context.activeVariant = undefined; @@ -1921,6 +2052,103 @@ export function makeOpenCodeAdapter( }); }); + /** + * Record one per-source token report into the active turn's contribution + * ledger (ported from #8918). Zero-token reports carry no usage and are + * skipped, so e.g. compaction summaries never populate the ledger. A + * report whose key was already seen under a *different* turn is a late + * event from a prior turn and is fenced out instead of contaminating the + * new turn's totals. Reports for any other turn (or no turn) are ignored: + * only the active turn accumulates. Part-level reports feed turn + * accounting only — the context-window meter keeps reflecting + * latest-message occupancy plus session cumulative totals. + */ + const recordOpenCodeTurnContribution = ( + context: OpenCodeSessionContext, + input: { + readonly tokens: OpenCodeTurnTokens; + readonly cost: number | null; + readonly modelKey: string | null; + readonly dedupeKey: string | undefined; + readonly turnId: TurnId | undefined; + }, + ): void => { + const { tokens, cost, modelKey, dedupeKey, turnId } = input; + if ( + tokens.input + tokens.output + tokens.cache.read + tokens.cache.write === 0 && + tokens.total === undefined + ) { + return; + } + if (dedupeKey !== undefined) { + const owningTurn = context.contributionTurnMap.get(dedupeKey); + if (owningTurn !== undefined) { + if (turnId === undefined || owningTurn !== turnId) { + return; + } + } else if (turnId !== undefined) { + context.contributionTurnMap.set(dedupeKey, turnId); + } + } + if (turnId === undefined || turnId !== context.activeTurnId) { + return; + } + const key = dedupeKey ?? `generic:${context.contributionCounter++}`; + context.turnContributionEntries.set(key, { tokens, cost, modelKey }); + }; + + /** + * Take the active turn's merged contribution totals for `turn.completed` + * and reset the per-turn ledger (the cross-turn fence map is retained so + * late events stay fenced). Same-id re-emissions were replaced on record, + * so merging across entries sums each source once; cost sums the same + * way, and per-model buckets merge independently. Absent cost/model data + * stays absent rather than becoming zero/unknown. + */ + const takeOpenCodeTurnContributions = ( + context: OpenCodeSessionContext, + ): { + readonly usage: OpenCodeTurnTokens | undefined; + readonly modelUsage: Record | undefined; + readonly totalCostUsd: number | undefined; + } => { + let merged: OpenCodeTurnTokens | null = null; + let costSum = 0; + let hasCost = false; + const modelUsage = new Map(); + for (const entry of context.turnContributionEntries.values()) { + merged = merged === null ? entry.tokens : mergeOpenCodeTurnTokens(merged, entry.tokens); + if (entry.cost !== null) { + costSum += entry.cost; + hasCost = true; + } + if (entry.modelKey !== null) { + const existing = modelUsage.get(entry.modelKey); + modelUsage.set( + entry.modelKey, + existing ? mergeOpenCodeTurnTokens(existing, entry.tokens) : entry.tokens, + ); + } + } + context.turnContributionEntries.clear(); + context.contributionCounter = 0; + return { + usage: merged ?? undefined, + modelUsage: modelUsage.size > 0 ? Object.fromEntries(modelUsage) : undefined, + totalCostUsd: hasCost ? costSum : undefined, + }; + }; + + /** + * Drop the per-turn contribution ledger without emitting (turns that end + * failed, aborted, or interrupted carry no contribution payload; the + * cross-turn fence map is retained). + */ + const clearOpenCodeTurnContributions = (context: OpenCodeSessionContext): void => { + context.turnContributionEntries.clear(); + context.contributionCounter = 0; + }; + // Records a child session of this thread. A child seen during a live turn // means that turn used subagents, whether the relation came from a // `session.created` event or a later ancestry lookup after reconnect. @@ -2622,6 +2850,24 @@ export function makeOpenCodeAdapter( yield* emitOpenCodeTokenUsage(context, { turnId, raw: event }); } } + // Per-turn contribution ledger (ported from #8918): same-id + // re-emissions replace that message's report, distinct ids + // (tool steps) each contribute; merged on turn completion into + // `usage`/`modelUsage`/`totalCostUsd`. Deliberately message-level + // only: part-level reports have unproven overlap semantics with + // their parent message totals, and none of the preserved cases + // exercise them. + const contributionTokens = readOpenCodeTurnTokens(info.tokens); + if (contributionTokens) { + recordOpenCodeTurnContribution(context, { + tokens: contributionTokens, + cost: + typeof info.cost === "number" && Number.isFinite(info.cost) ? info.cost : null, + modelKey: openCodeContributionModelKey(info.modelID, info.providerID), + dedupeKey: typeof info.id === "string" ? `msg:${info.id}` : undefined, + turnId, + }); + } const usage = context.turnTokenUsage; const parentMessageId = typeof info.parentID === "string" && info.parentID.trim().length > 0 @@ -2926,6 +3172,8 @@ export function makeOpenCodeAdapter( terminalCancellation.acknowledged = true; } const tokenUsage = activeTurnId ? takeOpenCodeTurnTokenUsage(context, false) : undefined; + // The turn ends failed: drop its contribution ledger with it. + clearOpenCodeTurnContributions(context); context.activeTurnId = undefined; context.activeAgent = undefined; context.activeVariant = undefined; @@ -3278,6 +3526,9 @@ export function makeOpenCodeAdapter( maxTokensLastAttemptModelKey: undefined, maxTokensLastAttemptMs: undefined, lastEmittedTokenUsageKey: undefined, + turnContributionEntries: new Map(), + contributionTurnMap: new Map(), + contributionCounter: 0, turnTokenUsage: undefined, activeTurnId: undefined, activeAgent: undefined, @@ -3544,6 +3795,8 @@ export function makeOpenCodeAdapter( return; } const tokenUsage = takeOpenCodeTurnTokenUsage(context, false); + // The turn ends aborted: drop its contribution ledger with it. + clearOpenCodeTurnContributions(context); context.promptAdmission = undefined; context.activeTurnId = undefined; context.activeAgent = undefined; @@ -3592,6 +3845,8 @@ export function makeOpenCodeAdapter( return; } const tokenUsage = takeOpenCodeTurnTokenUsage(context, false); + // The turn ends aborted: drop its contribution ledger with it. + clearOpenCodeTurnContributions(context); context.promptAdmission = undefined; context.activeTurnId = undefined; context.activeAgent = undefined; From fae9e44db649a33173de34b205c2749a2dcc50c6 Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 5 Sep 2026 18:18:14 -0400 Subject: [PATCH 07/12] test(opencode): port live turn-usage cases from #8918 - Accumulate per-message reports with same-id replace across tool steps - Carry usage, modelUsage, and totalCostUsd on idle turn.completed - Fence late events from a prior turn out of the new turn's totals --- .../provider/Layers/OpenCodeAdapter.test.ts | 580 ++++++++++++++++++ 1 file changed, 580 insertions(+) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 21c18cf72d4b..3706c10d9f04 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -7496,6 +7496,586 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + // ---- Live turn-usage accounting (ported from #8918) ----------------------- + // + // #8918 recorded these live-usage cases before it was closed unmerged; they + // are preserved here against the current turn machinery. The per-turn + // contribution ledger (same-id reports replace, distinct ids accumulate, + // late events are fenced to their owning turn) feeds `usage`/`modelUsage`/ + // `totalCostUsd` on idle `turn.completed`. The context-window meter keeps + // its own latest-message occupancy semantics on top (so `inputTokens` below + // folds cache in, unlike #8918's raw snapshot). + + it.effect("emits thread.token-usage.updated on message.updated with tokens", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-token-usage"); + const messageUpdatedEvent = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [messageUpdatedEvent.promise]; + + const tokenUsageFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.token-usage.updated", + ), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId, + input: "test tokens", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/mimo-test", + ), + }); + + messageUpdatedEvent.resolve({ + id: "evt-token-usage", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_token", + role: "assistant", + tokens: { input: 100, output: 50, reasoning: 10, cache: { read: 20, write: 5 } }, + cost: 0.002, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + const events = Array.from( + yield* Fiber.join(tokenUsageFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(events.length, 1); + const usage = (events[0] as { payload: { usage: Record } }).payload.usage; + // Meter semantics: input folds cache in (occupancy), unlike the raw + // per-model turn usage below. + NodeAssert.equal(usage.inputTokens as number, 125); + NodeAssert.equal(usage.cachedInputTokens as number, 25); + NodeAssert.equal(usage.outputTokens as number, 50); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("does not emit token usage on zero-token payload", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-zero-token"); + const messageUpdatedEvent = promiseWithResolvers(); + const idleEvent = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [messageUpdatedEvent.promise, idleEvent.promise]; + + const turnCompletedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "zero tokens", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/mimo-test", + ), + }); + + messageUpdatedEvent.resolve({ + id: "evt-zero-token", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_zero", + role: "assistant", + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + cost: 0, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + yield* Effect.yieldNow; + + idleEvent.resolve({ + id: "evt-zero-idle", + type: "session.status", + properties: { + sessionID: "http://127.0.0.1:9999/session", + status: { type: "idle" }, + }, + }); + + const events = Array.from( + yield* Fiber.join(turnCompletedFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(events.length, 1); + const payload = (events[0] as { payload: Record }).payload; + NodeAssert.equal(payload.usage, undefined); + NodeAssert.equal(String((events[0] as { turnId: unknown }).turnId), String(turn.turnId)); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("idle turn completion carries usage and modelUsage", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-idle-usage"); + const messageUpdatedEvent = promiseWithResolvers(); + const idleEvent = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [messageUpdatedEvent.promise, idleEvent.promise]; + + const turnCompletedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "test idle usage", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/mimo-test", + ), + }); + + messageUpdatedEvent.resolve({ + id: "evt-idle-usage-msg", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_idle", + role: "assistant", + tokens: { input: 100, output: 20, reasoning: 5, cache: { read: 10, write: 2 } }, + cost: 0.001, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + // Yield to let token emission be processed + yield* Effect.yieldNow; + + idleEvent.resolve({ + id: "evt-idle-usage-idle", + type: "session.status", + properties: { + sessionID: "http://127.0.0.1:9999/session", + status: { type: "idle" }, + }, + }); + + const events = Array.from( + yield* Fiber.join(turnCompletedFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(events.length, 1); + const payload = (events[0] as { payload: Record }).payload; + NodeAssert.equal(payload.state, "completed"); + NodeAssert.ok(payload.usage); + NodeAssert.equal(payload.totalCostUsd as number, 0.001); + const modelUsage = payload.modelUsage as Record; + NodeAssert.ok(modelUsage["opencode/mimo-test"]); + + // Ensure turnId matches + NodeAssert.equal(String((events[0] as { turnId: unknown }).turnId), String(turn.turnId)); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("accumulates usage and cost across multiple tool steps", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-accumulated-usage"); + const firstMessage = promiseWithResolvers(); + const secondMessage = promiseWithResolvers(); + const idleEvent = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + firstMessage.promise, + secondMessage.promise, + idleEvent.promise, + ]; + + const turnCompletedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "test accumulation", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/mimo-test", + ), + }); + + firstMessage.resolve({ + id: "evt-accum-1", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_accum_1", + role: "assistant", + tokens: { input: 100, output: 20, reasoning: 5, cache: { read: 10, write: 2 } }, + cost: 0.001, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + yield* Effect.yieldNow; + + secondMessage.resolve({ + id: "evt-accum-2", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_accum_2", + role: "assistant", + tokens: { input: 50, output: 30, reasoning: 10, cache: { read: 5, write: 1 } }, + cost: 0.002, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + yield* Effect.yieldNow; + + idleEvent.resolve({ + id: "evt-accum-idle", + type: "session.status", + properties: { + sessionID: "http://127.0.0.1:9999/session", + status: { type: "idle" }, + }, + }); + + const events = Array.from( + yield* Fiber.join(turnCompletedFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(events.length, 1); + const payload = (events[0] as { payload: Record }).payload; + const usage = payload.usage as { + input: number; + output: number; + reasoning: number; + cache: { read: number; write: number }; + }; + // Input 100+50, output 20+30, cache read 10+5, write 2+1 + NodeAssert.equal(usage.input, 150); + NodeAssert.equal(usage.output, 50); + NodeAssert.equal(usage.cache.read, 15); + NodeAssert.equal(usage.cache.write, 3); + // Cost summed + NodeAssert.equal(payload.totalCostUsd as number, 0.003); + const modelUsage = payload.modelUsage as Record; + NodeAssert.ok(modelUsage["opencode/mimo-test"]); + NodeAssert.equal(modelUsage["opencode/mimo-test"].input, 150); + NodeAssert.equal(String((events[0] as { turnId: unknown }).turnId), String(turn.turnId)); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("does not double count duplicate message.updated for same id", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-dedupe-usage"); + const firstMessage = promiseWithResolvers(); + const secondMessage = promiseWithResolvers(); + const idleEvent = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + firstMessage.promise, + secondMessage.promise, + idleEvent.promise, + ]; + + const turnCompletedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "test dedupe", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/mimo-test", + ), + }); + + firstMessage.resolve({ + id: "evt-dedupe-1", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_dedupe", + role: "assistant", + tokens: { input: 100, output: 20, reasoning: 5, cache: { read: 10, write: 2 } }, + cost: 0.001, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + yield* Effect.yieldNow; + + // Same message id, updated tokens (cumulative) — should replace, not sum + secondMessage.resolve({ + id: "evt-dedupe-2", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_dedupe", + role: "assistant", + tokens: { input: 150, output: 30, reasoning: 10, cache: { read: 15, write: 3 } }, + cost: 0.002, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + yield* Effect.yieldNow; + + idleEvent.resolve({ + id: "evt-dedupe-idle", + type: "session.status", + properties: { + sessionID: "http://127.0.0.1:9999/session", + status: { type: "idle" }, + }, + }); + + const events = Array.from( + yield* Fiber.join(turnCompletedFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(events.length, 1); + const payload = (events[0] as { payload: Record }).payload; + const usage = payload.usage as { + input: number; + output: number; + cache: { read: number; write: number }; + }; + // Should be latest snapshot, not sum of both (100+150) + NodeAssert.equal(usage.input, 150); + NodeAssert.equal(usage.output, 30); + NodeAssert.equal(usage.cache.read, 15); + // Cost should be latest as well (deduped) — 0.002 not 0.003 + // For deduped message, cost is replaced, so total is 0.002 + NodeAssert.equal(payload.totalCostUsd as number, 0.002); + NodeAssert.equal(String((events[0] as { turnId: unknown }).turnId), String(turn.turnId)); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("fences late events from a prior turn out of the new turn's totals", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-fenced-usage"); + const firstMessage = promiseWithResolvers(); + const firstIdle = promiseWithResolvers(); + const lateDuplicate = promiseWithResolvers(); + const secondMessage = promiseWithResolvers(); + const secondIdle = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + firstMessage.promise, + firstIdle.promise, + lateDuplicate.promise, + secondMessage.promise, + secondIdle.promise, + ]; + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const firstTurn = yield* adapter.sendTurn({ + threadId, + input: "first turn", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/mimo-test", + ), + }); + const firstCompletedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + firstMessage.resolve({ + id: "evt-fence-1", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_fence", + role: "assistant", + tokens: { input: 10, output: 5, reasoning: 1, cache: { read: 2, write: 0 } }, + cost: 0.001, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + yield* Effect.yieldNow; + + firstIdle.resolve({ + id: "evt-fence-idle-1", + type: "session.status", + properties: { + sessionID: "http://127.0.0.1:9999/session", + status: { type: "idle" }, + }, + }); + + const firstEvents = Array.from( + yield* Fiber.join(firstCompletedFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(firstEvents.length, 1); + const firstPayload = (firstEvents[0] as { payload: Record }).payload; + NodeAssert.equal((firstPayload.usage as { input: number }).input, 10); + + const secondTurn = yield* adapter.sendTurn({ + threadId, + input: "second turn", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/mimo-test", + ), + }); + const secondCompletedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + // Late re-emission of the first turn's message while the second turn is + // active: same id, much larger totals. It must not contaminate the new + // turn — the contribution key is already owned by the first turn. + lateDuplicate.resolve({ + id: "evt-fence-late", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_fence", + role: "assistant", + tokens: { input: 100, output: 50, reasoning: 10, cache: { read: 20, write: 5 } }, + cost: 0.009, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + yield* Effect.yieldNow; + + secondMessage.resolve({ + id: "evt-fence-2", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_fence_2", + role: "assistant", + tokens: { input: 7, output: 3, reasoning: 1, cache: { read: 1, write: 0 } }, + cost: 0.002, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + yield* Effect.yieldNow; + + secondIdle.resolve({ + id: "evt-fence-idle-2", + type: "session.status", + properties: { + sessionID: "http://127.0.0.1:9999/session", + status: { type: "idle" }, + }, + }); + + const secondEvents = Array.from( + yield* Fiber.join(secondCompletedFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(secondEvents.length, 1); + const secondPayload = (secondEvents[0] as { payload: Record }).payload; + const secondUsage = secondPayload.usage as { input: number; output: number }; + // Only the second turn's own message — not 10+100+7. + NodeAssert.equal(secondUsage.input, 7); + NodeAssert.equal(secondUsage.output, 3); + NodeAssert.equal(secondPayload.totalCostUsd as number, 0.002); + NodeAssert.equal( + String((secondEvents[0] as { turnId: unknown }).turnId), + String(secondTurn.turnId), + ); + NodeAssert.equal( + String((firstEvents[0] as { turnId: unknown }).turnId), + String(firstTurn.turnId), + ); + + yield* adapter.stopSession(threadId); + }), + ); + it.effect("keeps per-message usage when a zero-token message arrives", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; From b4a6f13e133848b0e050b69f38bfdeff68b2714a Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 5 Sep 2026 18:40:09 -0400 Subject: [PATCH 08/12] fix(opencode): bound the turn-contribution fence map - Evict oldest contributionTurnMap entries past 1,000 so long-lived sessions cannot accumulate message ids without bound, while recent keys stay fenced to their owning turn --- .../provider/Layers/OpenCodeAdapter.test.ts | 36 +++++++++++++++++++ .../src/provider/Layers/OpenCodeAdapter.ts | 35 ++++++++++++++++-- 2 files changed, 69 insertions(+), 2 deletions(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 3706c10d9f04..78432c94eb44 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -30,6 +30,7 @@ import { ProviderDriverKind, ProviderInstanceId, ThreadId, + TurnId, } from "@t3tools/contracts"; import { createModelSelection } from "@t3tools/shared/model"; import { ServerConfig } from "../../config.ts"; @@ -49,6 +50,7 @@ import { mergeOpenCodeAssistantText, normalizeOpenCodeTokenCounts, normalizeOpenCodeTokenUsage, + trimContributionTurnMap, } from "./OpenCodeAdapter.ts"; import { symlinksSupported } from "@t3tools/shared/testing/symlinks"; @@ -8076,6 +8078,40 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("bounds the contribution fence map with oldest-first eviction", () => + Effect.gen(function* () { + const entries = new Map(); + const turn = TurnId.make("turn-fence-bound"); + // Past the 1,000-entry bound, the oldest keys go first. + for (let index = 0; index < 1_003; index += 1) { + entries.set(`msg:${index}`, turn); + } + NodeAssert.equal(entries.size, 1_003); + trimContributionTurnMap(entries); + NodeAssert.equal(entries.size, 1_000); + NodeAssert.equal(entries.has("msg:0"), false); + NodeAssert.equal(entries.has("msg:1"), false); + NodeAssert.equal(entries.has("msg:2"), false); + NodeAssert.equal(entries.get("msg:3"), turn); + NodeAssert.equal(entries.get("msg:1002"), turn); + }), + ); + + it.effect("leaves the contribution fence map alone below the bound", () => + Effect.gen(function* () { + const entries = new Map(); + const turn = TurnId.make("turn-fence-small"); + for (let index = 0; index < 500; index += 1) { + entries.set(`msg:${index}`, turn); + } + trimContributionTurnMap(entries); + NodeAssert.equal(entries.size, 500); + NodeAssert.equal(entries.get("msg:0"), turn); + NodeAssert.equal(entries.get("msg:499"), turn); + trimContributionTurnMap(new Map()); + }), + ); + it.effect("keeps per-message usage when a zero-token message arrives", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 05838103de55..baba000d305d 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -81,6 +81,17 @@ const OPENCODE_RESUME_VERSION = 1 as const; */ const OPENCODE_CATALOG_RETRY_COOLDOWN_MS = 30_000; +/** + * Bound for the per-turn contribution fence map (`contributionTurnMap`). + * Every recorded message/part id is retained so late re-deliveries stay + * fenced to their owning turn, which would otherwise grow the map without + * bound over a long-lived session. Past this many distinct keys the oldest + * entries are evicted first: fencing only matters for recent keys (late + * events surface within a turn or two of their original), so eviction trades + * a vanishing residual risk for strictly bounded memory. + */ +const OPENCODE_CONTRIBUTION_FENCE_LIMIT = 1_000; + /** * Decode a persisted resume cursor into the upstream `ses_…` id. Anything * that isn't a current-version cursor with a non-empty id means "no resume" @@ -286,6 +297,22 @@ function openCodeContributionModelKey(model: unknown, providerId: unknown): stri return null; } +/** + * Evict the oldest fence entries past {@link OPENCODE_CONTRIBUTION_FENCE_LIMIT}. + * Exported for unit tests; the record path is its only production caller. + * `Map` preserves insertion order, so iterating from the front evicts + * least-recently-recorded keys first. + */ +export function trimContributionTurnMap(entries: Map): void { + while (entries.size > OPENCODE_CONTRIBUTION_FENCE_LIMIT) { + const oldest = entries.keys().next(); + if (oldest.done) { + return; + } + entries.delete(oldest.value); + } +} + /** * Whether an error definitively reports a missing session. Only a confirmed * miss may silently start a fresh session; any other failure (the SDK client @@ -595,8 +622,9 @@ interface OpenCodeSessionContext { /** * Owning turn per contribution key. Late events from a prior turn must not * contaminate the new turn's totals, so a key already seen under a - * different turn is fenced out. Retained across turns (bounded by - * message/part count) — clearing it would re-admit stale reports as new. + * different turn is fenced out. Retained across turns, but bounded by + * {@link OPENCODE_CONTRIBUTION_FENCE_LIMIT} with oldest-first eviction — + * fencing only matters for recent keys. */ contributionTurnMap: Map; /** Suffix source for contributions without a stable id. */ @@ -2088,6 +2116,9 @@ export function makeOpenCodeAdapter( } } else if (turnId !== undefined) { context.contributionTurnMap.set(dedupeKey, turnId); + // The fence map outlives turns by design; keep it bounded so + // long-lived sessions cannot accumulate ids without limit. + trimContributionTurnMap(context.contributionTurnMap); } } if (turnId === undefined || turnId !== context.activeTurnId) { From 6712dc58073d7e1971e728ac01286c10f0bf330b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ra=C3=BAl=20Alejandro=20P=C3=A9rez=20Acosta?= Date: Sat, 5 Sep 2026 18:44:02 -0400 Subject: [PATCH 09/12] Update apps/server/src/provider/Layers/OpenCodeAdapter.ts Co-authored-by: macroscopeapp[bot] <170038800+macroscopeapp[bot]@users.noreply.github.com> --- apps/server/src/provider/Layers/OpenCodeAdapter.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index baba000d305d..4d0ff94b772f 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -251,8 +251,8 @@ function readOpenCodeTurnTokens(value: unknown): OpenCodeTurnTokens | null { } function mergeOpenCodeTurnTokens(a: OpenCodeTurnTokens, b: OpenCodeTurnTokens): OpenCodeTurnTokens { - const totalA = a.total ?? a.input + a.output + a.cache.read + a.cache.write; - const totalB = b.total ?? b.input + b.output + b.cache.read + b.cache.write; + const totalA = a.total ?? a.input + a.output + a.reasoning + a.cache.read + a.cache.write; + const totalB = b.total ?? b.input + b.output + b.reasoning + b.cache.read + b.cache.write; const hasTotal = a.total !== undefined || b.total !== undefined; return { input: a.input + b.input, From ef71de3cd59d91b028a39c6cd808a889e11972d1 Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 5 Sep 2026 19:13:05 -0400 Subject: [PATCH 10/12] test(opencode): drop per-turn usage on failed turns - Consolidate token normalization cases into one context-window snapshot test - Cover failed turns omitting usage, modelUsage, and totalCostUsd - Fold small contribution fence-map bound check into trim test --- .../provider/Layers/OpenCodeAdapter.test.ts | 223 +++++++++++------- 1 file changed, 138 insertions(+), 85 deletions(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 78432c94eb44..4f8320002ef6 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -6522,41 +6522,25 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); - it.effect("clamps partial, non-finite, and negative token counts to zero", () => + it.effect("normalizes OpenCode token reports into context-window snapshots", () => Effect.sync(() => { - const counts = normalizeOpenCodeTokenCounts({ - input: NaN, - output: -40, - reasoning: Number.POSITIVE_INFINITY, - cache: { read: 11.9, write: undefined }, - }); - NodeAssert.deepEqual(counts, { - input: 0, - output: 0, - reasoning: 0, - cacheRead: 12, - cacheWrite: 0, - }); + // Non-finite, negative, and fractional counts clamp to whole zeros. + NodeAssert.deepEqual( + normalizeOpenCodeTokenCounts({ + input: NaN, + output: -40, + reasoning: Number.POSITIVE_INFINITY, + cache: { read: 11.9, write: undefined }, + }), + { input: 0, output: 0, reasoning: 0, cacheRead: 12, cacheWrite: 0 }, + ); // A payload missing the whole cache object must not throw. - const withoutCache = normalizeOpenCodeTokenCounts({ - input: 3, - output: 4, - reasoning: 0, - cache: undefined, - }); - NodeAssert.deepEqual(withoutCache, { - input: 3, - output: 4, - reasoning: 0, - cacheRead: 0, - cacheWrite: 0, - }); - }), - ); + NodeAssert.deepEqual( + normalizeOpenCodeTokenCounts({ input: 3, output: 4, reasoning: 0, cache: undefined }), + { input: 3, output: 4, reasoning: 0, cacheRead: 0, cacheWrite: 0 }, + ); - it.effect("maps OpenCode token counts into a context-window usage snapshot", () => - Effect.sync(() => { const counts = normalizeOpenCodeTokenCounts({ input: 100, output: 40, @@ -6571,49 +6555,47 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { cacheWrite: 13, }); - const usage = normalizeOpenCodeTokenUsage({ - messageTokens: counts, - cumulativeTokens: counts, - maxTokens: 200_000, - cumulativeSessionCost: 0.42, - }); - NodeAssert.deepEqual(usage, { - usedTokens: 124, - totalProcessedTokens: 171, - maxTokens: 200_000, - cost: 0.42, - inputTokens: 124, - cachedInputTokens: 24, - outputTokens: 40, - reasoningOutputTokens: 7, - lastUsedTokens: 124, - lastInputTokens: 124, - lastCachedInputTokens: 24, - lastOutputTokens: 40, - lastReasoningOutputTokens: 7, - }); - }), - ); - - it.effect("omits cumulative and breakdown fields OpenCode did not report", () => - Effect.sync(() => { - const usage = normalizeOpenCodeTokenUsage({ - messageTokens: { input: 5, output: 0, reasoning: 0, cacheRead: 0, cacheWrite: 0 }, - cumulativeSessionCost: 0, - }); - NodeAssert.deepEqual(usage, { - usedTokens: 5, - inputTokens: 5, - cachedInputTokens: 0, - lastUsedTokens: 5, - lastInputTokens: 5, - lastCachedInputTokens: 0, - }); - }), - ); + NodeAssert.deepEqual( + normalizeOpenCodeTokenUsage({ + messageTokens: counts, + cumulativeTokens: counts, + maxTokens: 200_000, + cumulativeSessionCost: 0.42, + }), + { + usedTokens: 124, + totalProcessedTokens: 171, + maxTokens: 200_000, + cost: 0.42, + inputTokens: 124, + cachedInputTokens: 24, + outputTokens: 40, + reasoningOutputTokens: 7, + lastUsedTokens: 124, + lastInputTokens: 124, + lastCachedInputTokens: 24, + lastOutputTokens: 40, + lastReasoningOutputTokens: 7, + }, + ); + + // Fields OpenCode did not report stay absent instead of zero-filled. + NodeAssert.deepEqual( + normalizeOpenCodeTokenUsage({ + messageTokens: { input: 5, output: 0, reasoning: 0, cacheRead: 0, cacheWrite: 0 }, + cumulativeSessionCost: 0, + }), + { + usedTokens: 5, + inputTokens: 5, + cachedInputTokens: 0, + lastUsedTokens: 5, + lastInputTokens: 5, + lastCachedInputTokens: 0, + }, + ); - it.effect("returns no usage until an assistant message reports tokens", () => - Effect.sync(() => { + // Reports without message tokens mean no usage yet. NodeAssert.equal( normalizeOpenCodeTokenUsage({ cumulativeTokens: { input: 50, output: 10, reasoning: 0, cacheRead: 20, cacheWrite: 0 }, @@ -7712,6 +7694,82 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("drops per-turn usage when the turn ends failed", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-failed-usage"); + const messageUpdatedEvent = promiseWithResolvers(); + const errorEvent = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [messageUpdatedEvent.promise, errorEvent.promise]; + + const turnCompletedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.take(1), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "test failed usage", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/mimo-test", + ), + }); + + messageUpdatedEvent.resolve({ + id: "evt-failed-usage-msg", + type: "message.updated", + properties: { + sessionID: "http://127.0.0.1:9999/session", + info: { + id: "msg_failed", + role: "assistant", + tokens: { input: 100, output: 20, reasoning: 5, cache: { read: 10, write: 2 } }, + cost: 0.001, + modelID: "mimo-test", + providerID: "opencode", + }, + }, + }); + + yield* Effect.yieldNow; + + errorEvent.resolve({ + id: "evt-failed-usage-error", + type: "session.error", + properties: { + sessionID: "http://127.0.0.1:9999/session", + error: { + name: "APIError", + data: { message: "Upstream failed", isRetryable: false }, + }, + }, + }); + + const events = Array.from( + yield* Fiber.join(turnCompletedFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(events.length, 1); + const payload = (events[0] as { payload: Record }).payload; + NodeAssert.equal(payload.state, "failed"); + // The failed turn carries no contribution payload: usage, per-model + // buckets, and summed cost are dropped with the ledger. + NodeAssert.equal("usage" in payload, false); + NodeAssert.equal("modelUsage" in payload, false); + NodeAssert.equal("totalCostUsd" in payload, false); + NodeAssert.equal(String((events[0] as { turnId: unknown }).turnId), String(turn.turnId)); + + yield* adapter.stopSession(threadId); + }), + ); + it.effect("accumulates usage and cost across multiple tool steps", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; @@ -8094,21 +8152,16 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { NodeAssert.equal(entries.has("msg:2"), false); NodeAssert.equal(entries.get("msg:3"), turn); NodeAssert.equal(entries.get("msg:1002"), turn); - }), - ); - it.effect("leaves the contribution fence map alone below the bound", () => - Effect.gen(function* () { - const entries = new Map(); - const turn = TurnId.make("turn-fence-small"); + // At or below the bound the map is left alone. + const small = new Map(); for (let index = 0; index < 500; index += 1) { - entries.set(`msg:${index}`, turn); + small.set(`msg:${index}`, turn); } - trimContributionTurnMap(entries); - NodeAssert.equal(entries.size, 500); - NodeAssert.equal(entries.get("msg:0"), turn); - NodeAssert.equal(entries.get("msg:499"), turn); - trimContributionTurnMap(new Map()); + trimContributionTurnMap(small); + NodeAssert.equal(small.size, 500); + NodeAssert.equal(small.get("msg:0"), turn); + NodeAssert.equal(small.get("msg:499"), turn); }), ); From d4b2eaa10ffbebf53cae8242c565176dce37ee30 Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 5 Sep 2026 19:34:01 -0400 Subject: [PATCH 11/12] fix(opencode): include reasoning tokens in usage guard --- apps/server/src/provider/Layers/OpenCodeAdapter.ts | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 4d0ff94b772f..e00c71318858 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -2103,7 +2103,8 @@ export function makeOpenCodeAdapter( ): void => { const { tokens, cost, modelKey, dedupeKey, turnId } = input; if ( - tokens.input + tokens.output + tokens.cache.read + tokens.cache.write === 0 && + tokens.input + tokens.output + tokens.reasoning + tokens.cache.read + tokens.cache.write === + 0 && tokens.total === undefined ) { return; From bf3d9ab3ffa18e0fb8048b797a478d90d4517505 Mon Sep 17 00:00:00 2001 From: ardevdevts Date: Sat, 5 Sep 2026 19:46:07 -0400 Subject: [PATCH 12/12] fix(opencode): ignore prior-turn updates in token totals - Skip `other`-owned messages when recording turn contributions - Prevents late message.updated events from inflating current usage totals --- .../src/provider/Layers/OpenCodeAdapter.ts | 40 ++++++++++--------- 1 file changed, 22 insertions(+), 18 deletions(-) diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index e00c71318858..48865e1c74d5 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -2882,24 +2882,6 @@ export function makeOpenCodeAdapter( yield* emitOpenCodeTokenUsage(context, { turnId, raw: event }); } } - // Per-turn contribution ledger (ported from #8918): same-id - // re-emissions replace that message's report, distinct ids - // (tool steps) each contribute; merged on turn completion into - // `usage`/`modelUsage`/`totalCostUsd`. Deliberately message-level - // only: part-level reports have unproven overlap semantics with - // their parent message totals, and none of the preserved cases - // exercise them. - const contributionTokens = readOpenCodeTurnTokens(info.tokens); - if (contributionTokens) { - recordOpenCodeTurnContribution(context, { - tokens: contributionTokens, - cost: - typeof info.cost === "number" && Number.isFinite(info.cost) ? info.cost : null, - modelKey: openCodeContributionModelKey(info.modelID, info.providerID), - dedupeKey: typeof info.id === "string" ? `msg:${info.id}` : undefined, - turnId, - }); - } const usage = context.turnTokenUsage; const parentMessageId = typeof info.parentID === "string" && info.parentID.trim().length > 0 @@ -2919,6 +2901,28 @@ export function makeOpenCodeAdapter( if (usage) { usage.assistantOwnershipByMessageId.set(info.id, ownership); } + // Per-turn contribution ledger (ported from #8918): same-id + // re-emissions replace that message's report, distinct ids + // (tool steps) each contribute; merged on turn completion into + // `usage`/`modelUsage`/`totalCostUsd`. Deliberately message-level + // only: part-level reports have unproven overlap semantics with + // their parent message totals, and none of the preserved cases + // exercise them. Late `message.updated` events from a prior turn + // resolve to ownership `"other"` via `parentID` above and are + // skipped so they cannot contaminate the current turn's totals. + if (ownership !== "other") { + const contributionTokens = readOpenCodeTurnTokens(info.tokens); + if (contributionTokens) { + recordOpenCodeTurnContribution(context, { + tokens: contributionTokens, + cost: + typeof info.cost === "number" && Number.isFinite(info.cost) ? info.cost : null, + modelKey: openCodeContributionModelKey(info.modelID, info.providerID), + dedupeKey: typeof info.id === "string" ? `msg:${info.id}` : undefined, + turnId, + }); + } + } for (const part of context.partById.values()) { if (part.messageID !== info.id) { continue;