From ae174c4c72a9ab3068afedfb93657757bdf60197 Mon Sep 17 00:00:00 2001 From: Vedant Mahajan Date: Wed, 28 Oct 2026 18:38:48 +0530 Subject: [PATCH] feat(usage): show per-thread and per-project API spend Adds a priced spend read (dollars + tokens) alongside the existing usage scan, and surfaces it in the app: thread spend in the chat header and project spend next to the project name, both with a hover breakdown. Co-authored-by: T3 Code --- apps/server/src/auth/RpcAuthorization.test.ts | 59 ++++ apps/server/src/auth/RpcAuthorization.ts | 10 + .../src/observability/RpcInstrumentation.ts | 1 + .../Orchestrator.control-reads.test.ts | 58 +++ .../src/orchestration-v2/Orchestrator.ts | 2 + .../ThreadManagementService.ts | 1 + apps/server/src/usage/UsageService.ts | 91 ++++- .../server/src/usage/usageAggregation.test.ts | 37 ++ apps/server/src/usage/usageAggregation.ts | 131 +++++-- .../src/usage/usageSessionSpendInput.test.ts | 104 ++++++ .../src/usage/usageSessionSpendInput.ts | 62 ++++ apps/server/src/ws.ts | 52 +++ apps/web/src/components/chat/ChatHeader.tsx | 11 + .../src/components/chat/ProjectSpendLabel.tsx | 158 +++++++++ .../components/chat/ThreadSpendControl.tsx | 227 ++++++++++++ .../web/src/components/chat/spendSessions.tsx | 163 +++++++++ .../src/components/chat/threadSpend.test.ts | 329 +++++++++++++++++ apps/web/src/components/chat/threadSpend.ts | 330 ++++++++++++++++++ apps/web/src/rpc/requestLatencyState.ts | 1 + packages/client-runtime/src/state/server.ts | 24 ++ packages/contracts/src/rpc.ts | 17 +- packages/contracts/src/usage.ts | 65 +++- packages/shared/package.json | 4 + packages/shared/src/usageMerge.ts | 16 +- packages/shared/src/usageSessionRefs.ts | 46 +++ 25 files changed, 1954 insertions(+), 45 deletions(-) create mode 100644 apps/server/src/usage/usageSessionSpendInput.test.ts create mode 100644 apps/server/src/usage/usageSessionSpendInput.ts create mode 100644 apps/web/src/components/chat/ProjectSpendLabel.tsx create mode 100644 apps/web/src/components/chat/ThreadSpendControl.tsx create mode 100644 apps/web/src/components/chat/spendSessions.tsx create mode 100644 apps/web/src/components/chat/threadSpend.test.ts create mode 100644 apps/web/src/components/chat/threadSpend.ts create mode 100644 packages/shared/src/usageSessionRefs.ts diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index 0a9f97cc3914..2394fa53f912 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -7,6 +7,7 @@ import { DEFAULT_SERVER_SETTINGS, ProviderInstanceId, ThreadId, + UsageDay, AuthOrchestrationOperateScope, AuthOrchestrationReadScope, AuthSourceControlWriteScope, @@ -333,6 +334,64 @@ it.effect("separates host file URLs from readable attachment URLs", () => }).pipe(Effect.scoped), ); +it.effect("requires orchestration read only when a spend read names threads", () => + Effect.gen(function* () { + const group = WsRpcGroup.omit( + ...[...WsRpcGroup.requests.keys()].filter( + ( + tag, + ): tag is Exclude< + keyof typeof RPC_REQUIRED_SCOPES, + typeof WS_METHODS.serverGetSessionSpend + > => tag !== WS_METHODS.serverGetSessionSpend, + ), + ); + let handled = 0; + const client = yield* RpcTest.makeClient(group).pipe( + Effect.provide( + Layer.mergeAll( + group.toLayerHandler(WS_METHODS.serverGetSessionSpend, () => + Effect.sync(() => { + handled++; + return { + readAt: "2026-10-28T00:00:00.000Z", + sinceDay: UsageDay.make("2026-07-28"), + untilDay: UsageDay.make("2026-10-28"), + sources: [], + pricing: { + status: "cached" as const, + source: "test", + fetchedAt: null, + knownModels: 0, + }, + sessions: [], + }; + }), + ), + RpcAuthorization.layer([AuthDiagnosticsReadScope]), + ), + ), + ); + + yield* client[WS_METHODS.serverGetSessionSpend]({ + sinceDay: UsageDay.make("2026-07-28"), + untilDay: UsageDay.make("2026-10-28"), + timeZone: "UTC", + sessions: [], + }); + expect( + yield* client[WS_METHODS.serverGetSessionSpend]({ + sinceDay: UsageDay.make("2026-07-28"), + untilDay: UsageDay.make("2026-10-28"), + timeZone: "UTC", + sessions: [], + threadIds: [ThreadId.make("thread")], + }).pipe(Effect.flip), + ).toMatchObject({ requiredScope: AuthOrchestrationReadScope }); + expect(handled).toBe(1); + }).pipe(Effect.scoped), +); + it.effect("denies manual cleanup before the handler without settings permission", () => Effect.gen(function* () { const group = WsRpcGroup.omit( diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index f127f94fdba3..faa209dfa544 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -27,6 +27,7 @@ import { type AuthEnvironmentScope, EnvironmentAuthorizationError, RpcScopeAuthorization, + UsageSessionSpendInput, WS_METHODS, WsRpcGroup, } from "@t3tools/contracts"; @@ -105,6 +106,7 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.serverGetResourceTelemetryHistory]: AuthDiagnosticsReadScope, [WS_METHODS.serverRetryResourceTelemetry]: AuthDiagnosticsReadScope, [WS_METHODS.serverGetUsageSummary]: AuthDiagnosticsReadScope, + [WS_METHODS.serverGetSessionSpend]: AuthDiagnosticsReadScope, [WS_METHODS.serverRefreshUsageRates]: AuthDiagnosticsReadScope, [WS_METHODS.serverSignalProcess]: AuthEnvironmentMaintainScope, [WS_METHODS.serverReportClientActivity]: AuthOrchestrationReadScope, @@ -239,6 +241,7 @@ const requiredScopesForSettingsUpdate = (payload: unknown) => { const decodeProjectMutation = Schema.decodeUnknownSync(ProjectMutation); const decodeAssetCreateUrlInput = Schema.decodeUnknownSync(AssetCreateUrlInput); +const decodeUsageSessionSpendInput = Schema.decodeUnknownSync(UsageSessionSpendInput); const requiredScopesForRpcCall = ( method: string, @@ -247,6 +250,13 @@ const requiredScopesForRpcCall = ( if (method === WS_METHODS.serverRetryResourceTelemetry) { return [AuthEnvironmentMaintainScope, AuthDiagnosticsReadScope]; } + // Pricing named threads reads those threads and answers with their provider session ids. + if (method === WS_METHODS.serverGetSessionSpend) { + const { threadIds } = decodeUsageSessionSpendInput(payload); + return threadIds === undefined || threadIds.length === 0 + ? [AuthDiagnosticsReadScope] + : [AuthDiagnosticsReadScope, AuthOrchestrationReadScope]; + } if (method === WS_METHODS.assetsCreateUrl) { const { resource } = decodeAssetCreateUrlInput(payload); return [ diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index 504e533b37e0..e11e18e6cc0e 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -82,6 +82,7 @@ const RPC_AGGREGATES = { [WS_METHODS.serverCheckOtlpEndpoint]: "server", [WS_METHODS.serverPrepareTriage]: "server", [WS_METHODS.serverGetUsageSummary]: "server", + [WS_METHODS.serverGetSessionSpend]: "server", [WS_METHODS.serverRefreshUsageRates]: "server", [WS_METHODS.serverSignalProcess]: "server", [WS_METHODS.serverReportClientActivity]: "server", diff --git a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts index c36dfc365af6..27ad9a0ffeba 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.control-reads.test.ts @@ -39,6 +39,7 @@ import * as Orchestrator from "./Orchestrator.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts"; +import { sessionRefsForProviderThreads } from "../usage/usageSessionSpendInput.ts"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; const instanceId = ProviderInstanceId.make("codex"); @@ -596,6 +597,63 @@ it.effect("implements a proposed plan that the command projection leaves out", ( }).pipe(Effect.provide(layerTest)), ); +it.effect("resolves a thread's usage sessions from its thread id", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const threadId = ThreadId.make("thread:spend-refs"); + const now = yield* DateTime.now; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("create-spend-refs"), + threadId, + projectId: ProjectId.make("project:spend-refs"), + title: "Spend refs", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* projections.apply({ + id: EventId.make("spend-refs:provider-thread"), + type: "provider-thread.updated", + threadId, + occurredAt: now, + payload: { + id: ProviderThreadId.make("provider-thread:spend-refs"), + driver: adapter.driver, + providerInstanceId: instanceId, + providerSessionId: null, + appThreadId: threadId, + ownerNodeId: null, + nativeThreadRef: { + driver: adapter.driver, + nativeId: "session:spend-refs", + strength: "strong", + }, + nativeConversationHeadRef: null, + status: "idle", + firstRunOrdinal: 1, + lastRunOrdinal: 1, + handoffIds: [], + forkedFrom: null, + createdAt: now, + updatedAt: now, + }, + }); + + // The spend RPC reads threads by id; a read that returned no provider threads here would + // price nothing while still reporting success. + const { providerThreads } = yield* projections.getThreadRecords(threadId, ["providerThreads"]); + assert.deepStrictEqual(sessionRefsForProviderThreads(providerThreads), [ + { provider: "codex", sessionId: "session:spend-refs" }, + ]); + }).pipe(Effect.provide(layerTest)), +); + it.effect("settles only the stopped run's background work, once", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 91acb881637c..a6518f595db8 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -366,6 +366,7 @@ export interface OrchestratorV2Shape { readonly getCheckpointContext: ( threadId: ThreadId, ) => Effect.Effect; + readonly getThreadSnapshot: (threadId: ThreadId) => Effect.Effect< { readonly schemaVersion: number; @@ -13324,6 +13325,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .getCheckpointContext(threadId) .pipe(Effect.mapError((cause) => new OrchestratorProjectionError({ threadId, cause }))); }), + getThreadSnapshot: Effect.fn("OrchestratorV2.getThreadSnapshot")(function (threadId) { return projectionStore .getThreadSnapshot(threadId) diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index cfbbd4a68a51..20d3ef724855 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -376,6 +376,7 @@ export interface ThreadManagementServiceShape { readonly getThreadProjection: ( threadId: ThreadId, ) => Effect.Effect; + readonly getCheckpointContext: Orchestrator.OrchestratorV2["Service"]["getCheckpointContext"]; readonly getThreadSnapshot: Orchestrator.OrchestratorV2["Service"]["getThreadSnapshot"]; readonly getThreadTranscript: ( diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 6150ab1d3cc3..db21089c061e 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -8,6 +8,8 @@ import { type UsageProviderKind, type UsageSource, type UsagePricing, + type UsageSessionSpendInput, + type UsageSessionSpendResult, type UsageSummary, type UsageSummaryInput, UsageReadError, @@ -32,6 +34,7 @@ import { HttpClient, HttpClientResponse } from "effect/http"; import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite"; import * as ServerConfig from "../config.ts"; import * as ServerSettings from "../serverSettings.ts"; +import { sessionSpendSourcesIncomplete } from "./usageSessionSpendInput.ts"; import { BUILT_IN_USAGE_DRIVERS, type BuiltInUsageReadersEnv } from "../provider/builtInDrivers.ts"; import type { ProviderDriver } from "@t3tools/provider-core/server/driver"; import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment"; @@ -72,6 +75,7 @@ const MTIME_SLACK_MS = 36 * 60 * 60 * 1000; const MAX_HOURLY_WINDOW_MS = 24 * 60 * 60 * 1000; const CACHE_RETENTION_DAYS = 92; +export const MAX_SESSION_SPEND_KEYS = 200; const TRANSCRIPT_READ_CONCURRENCY = 4; @@ -144,6 +148,9 @@ export class UsageService extends Context.Service< UsageService, { readonly readSummary: (input: UsageSummaryInput) => Effect.Effect; + readonly readSessionSpend: ( + input: UsageSessionSpendInput, + ) => Effect.Effect; readonly refreshRates: Effect.Effect; } >()("t3/usage/UsageService") {} @@ -629,6 +636,7 @@ export const make = Effect.gen(function* () { const scanSummary = Effect.fn("UsageService.scanSummary")(function* ( input: UsageSummaryInput, settings: ServerSettingsValue, + collectSessions = false, ) { if (input.sinceDay > input.untilDay) { return yield* new UsageReadError({ @@ -694,6 +702,7 @@ export const make = Effect.gen(function* () { rates, priceOverrides: createOverrideRateTable(settings.usagePriceOverrides), modelAliases: resolveModelAliases(settings.usageModelAliases), + collectSessions, }); const sources: UsageSource[] = []; @@ -845,7 +854,7 @@ export const make = Effect.gen(function* () { const readAt = yield* DateTime.now; const finishedAtMs = yield* Clock.currentTimeMillis; - return { + const summary = { contractVersion: USAGE_CONTRACT_VERSION, readAt: DateTime.formatIso(readAt), timeZone: input.timeZone, @@ -856,9 +865,15 @@ export const make = Effect.gen(function* () { pricing: pricing(), scanDurationMs: Math.max(0, finishedAtMs - startedAtMs), } satisfies UsageSummary; + return { summary, sessions: aggregator.finishSessions() }; }); - const inflightScans = new Map>(); + interface ScanBody { + readonly summary: UsageSummary; + readonly sessions: ReturnType; + } + + const inflightScans = new Map>(); const scanKey = (input: UsageSummaryInput, settings: ServerSettingsValue): string => JSON.stringify([ @@ -872,20 +887,23 @@ export const make = Effect.gen(function* () { input.awaitRefresh === true, ]); - const readSummary = Effect.fn("UsageService.readSummary")(function* (input: UsageSummaryInput) { + const readBody = Effect.fn("UsageService.readBody")(function* ( + input: UsageSummaryInput, + collectSessions: boolean, + ) { const settings = yield* readSettings; - const key = scanKey(input, settings); + const key = `${scanKey(input, settings)}\0${collectSessions ? "sessions" : "summary"}`; const deferred = yield* Effect.uninterruptible( Effect.gen(function* () { const existing = inflightScans.get(key); if (existing !== undefined) return existing; // Register and fork atomically so cancellation cannot leave a Deferred without a scan. - const created = Deferred.makeUnsafe(); + const created = Deferred.makeUnsafe(); inflightScans.set(key, created); // Detach so caller departure cannot cancel shared scans; finished scans warm the cache. // Register writes before releasing waiters, but start writing after delivering their summary. - yield* scanSummary(input, settings).pipe( + yield* scanSummary(input, settings, collectSessions).pipe( Effect.onExit((exit) => Effect.sync(() => inflightScans.delete(key)).pipe( Effect.andThen(Exit.isSuccess(exit) ? schedulePersist : Effect.void), @@ -900,7 +918,66 @@ export const make = Effect.gen(function* () { return yield* Deferred.await(deferred); }); - return { readSummary, refreshRates, awaitPersisted } as const; + const readSummary = Effect.fn("UsageService.readSummary")(function* (input: UsageSummaryInput) { + const body = yield* readBody(input, false); + return body.summary; + }); + + const readSessionSpend = Effect.fn("UsageService.readSessionSpend")(function* ( + input: UsageSessionSpendInput, + ) { + if (input.sinceDay > input.untilDay) { + return yield* new UsageReadError({ + reason: "invalidWindow", + detail: `sinceDay '${input.sinceDay}' is after untilDay '${input.untilDay}'`, + }); + } + if (input.sessions.length > MAX_SESSION_SPEND_KEYS) { + return yield* new UsageReadError({ + reason: "invalidWindow", + detail: `At most ${MAX_SESSION_SPEND_KEYS} sessions can be priced in one read`, + }); + } + if (input.sessions.length === 0) { + const readAt = yield* DateTime.now; + return { + readAt: DateTime.formatIso(readAt), + sinceDay: input.sinceDay, + untilDay: input.untilDay, + sources: [], + pricing: pricing(), + sessions: [], + } satisfies UsageSessionSpendResult; + } + + const body = yield* readBody( + { + sinceDay: input.sinceDay, + untilDay: input.untilDay, + timeZone: input.timeZone, + ...(input.awaitRefresh === undefined ? {} : { awaitRefresh: input.awaitRefresh }), + }, + true, + ); + const wanted = new Set( + input.sessions.map((session) => `${session.provider}\u0000${session.sessionId}`), + ); + return { + readAt: body.summary.readAt, + sinceDay: body.summary.sinceDay, + untilDay: body.summary.untilDay, + sources: body.summary.sources, + pricing: body.summary.pricing, + sessions: body.sessions.filter((session) => + wanted.has(`${session.provider}\u0000${session.sessionId}`), + ), + ...(sessionSpendSourcesIncomplete(body.summary.sources, input.sessions) + ? { incomplete: true } + : {}), + } satisfies UsageSessionSpendResult; + }); + + return { readSummary, readSessionSpend, refreshRates, awaitPersisted } as const; }); export const layer = Layer.effect(UsageService, make); diff --git a/apps/server/src/usage/usageAggregation.test.ts b/apps/server/src/usage/usageAggregation.test.ts index 08b3f3ae1b8c..279d437920d5 100644 --- a/apps/server/src/usage/usageAggregation.test.ts +++ b/apps/server/src/usage/usageAggregation.test.ts @@ -208,6 +208,43 @@ describe("UsageAggregator", () => { expect(result.buckets[0]?.costUsd).toBeCloseTo(0.00925, 9); expect(result.buckets[0]?.unpricedRecords).toBe(0); }); + + it("prices each session once and keeps the partition equal to the buckets", () => { + const aggregator = new UsageAggregator({ + timeZone: "UTC", + sinceDay: "2026-08-01", + untilDay: "2026-08-31", + rates, + collectSessions: true, + }); + aggregator.add(record({ sessionId: "parent", dedupeKey: "message-1" })); + aggregator.add( + record({ + sessionId: "fork-copy", + dedupeKey: "message-1", + timestampMs: Date.parse("2026-08-07T05:00:00.000Z"), + }), + ); + aggregator.add(record({ sessionId: "child", dedupeKey: "message-2" })); + aggregator.add(record({ sessionId: "", dedupeKey: "message-3" })); + aggregator.add( + record({ + sessionId: "parent", + dedupeKey: "old", + timestampMs: Date.parse("2026-07-01T00:00:00.000Z"), + }), + ); + const result = aggregator.finish(); + const sessions = aggregator.finishSessions(); + + expect(sessions.map((session) => session.sessionId)).toEqual(["child", "parent"]); + expect(sessions.find((session) => session.sessionId === "parent")?.records).toBe(1); + const sessionCost = sessions.reduce((sum, session) => sum + session.costUsd, 0); + const bucketCost = result.buckets.reduce((sum, bucket) => sum + bucket.costUsd, 0); + // The unnamed session stays in the dashboard buckets and cannot be assigned. + expect(bucketCost - sessionCost).toBeCloseTo(0.004625, 9); + expect(sessionCost).toBeCloseTo(0.00925, 9); + }); }); describe("resolveModelAliases", () => { diff --git a/apps/server/src/usage/usageAggregation.ts b/apps/server/src/usage/usageAggregation.ts index 17903cd062ea..27d8074aaccf 100644 --- a/apps/server/src/usage/usageAggregation.ts +++ b/apps/server/src/usage/usageAggregation.ts @@ -3,7 +3,9 @@ import type { UsageBucket, UsageCategoryCost, UsageDay, + UsageProviderKind, UsageResolution, + UsageSessionSpend, UsageTokenTotals, } from "@t3tools/contracts"; @@ -46,20 +48,28 @@ function makeDayFormatter(timeZone: string): (timestampMs: number) => string { const QUARTER_HOUR_MS = 15 * 60 * 1000; const HOUR_MS = 60 * 60 * 1000; -interface MutableBucket { +interface PricedAccumulator { totals: { -readonly [K in keyof UsageTokenTotals]: number }; costUsd: number; - cacheSavingsUsd: number; categoryCostUsd: UsageCategoryCost | null; - fastCostUsd: number; - ultrafastCostUsd: number; - speedPremiumUsd: number; records: number; unpricedRecords: number; providerReportedRecords: number; +} + +interface MutableBucket extends PricedAccumulator { + cacheSavingsUsd: number; + fastCostUsd: number; + ultrafastCostUsd: number; + speedPremiumUsd: number; sessions: Set; } +interface MutableSession extends PricedAccumulator { + provider: UsageProviderKind; + sessionId: string; +} + export interface AggregateOptions { readonly timeZone: string; readonly sinceDay: string; @@ -70,6 +80,8 @@ export interface AggregateOptions { readonly resolution?: UsageResolution; readonly sinceTimeMs?: number; readonly untilTimeMs?: number; + /** Dashboard scans leave this off so they do not retain a session map. */ + readonly collectSessions?: boolean; } export function resolveModelAliases( @@ -97,6 +109,7 @@ export interface AggregateResult { /** Deduplicates across the whole scan; Claude copies records into resumed/forked transcripts. */ export class UsageAggregator { readonly #buckets = new Map(); + readonly #sessions: Map | null; readonly #seen = new Set(); readonly #toDay: (timestampMs: number) => string; readonly #hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null; @@ -114,6 +127,7 @@ export class UsageAggregator { constructor(options: AggregateOptions) { this.#options = options; + this.#sessions = options.collectSessions === true ? new Map() : null; this.#toDay = makeDayFormatter(options.timeZone); if (options.resolution === "hour") { if (options.sinceTimeMs === undefined || options.untilTimeMs === undefined) { @@ -163,27 +177,9 @@ export class UsageAggregator { const bucket = this.#bucketFor(day, hourIndex, record.provider, record.model, sourcePath ?? ""); const priced = priceUsage(this.#options.rates, record, this.#options.priceOverrides); - - const totals = bucket.totals; - totals.uncachedInputTokens += record.totals.uncachedInputTokens; - totals.cachedInputTokens += record.totals.cachedInputTokens; - totals.cacheCreationTokens += record.totals.cacheCreationTokens; - totals.outputTokens += record.totals.outputTokens; - totals.reasoningTokens += record.totals.reasoningTokens; - bucket.costUsd += priced.costUsd; - if (priced.categoryCostUsd !== null) { - const sum = bucket.categoryCostUsd; - const add = priced.categoryCostUsd; - bucket.categoryCostUsd = - sum === null - ? add - : { - input: sum.input + add.input, - cacheRead: sum.cacheRead + add.cacheRead, - cacheWrite: sum.cacheWrite + add.cacheWrite, - output: sum.output + add.output, - }; - } + accumulatePriced(bucket, record.totals, priced.costUsd, priced.categoryCostUsd); + if (priced.costSource === "unpriced") bucket.unpricedRecords += 1; + if (priced.costSource === "providerReported") bucket.providerReportedRecords += 1; if (record.speed === "fast") bucket.fastCostUsd += priced.costUsd; if (record.speed === "ultrafast") bucket.ultrafastCostUsd += priced.costUsd; bucket.speedPremiumUsd += priced.speedPremiumUsd; @@ -192,10 +188,27 @@ export class UsageAggregator { record, this.#options.priceOverrides, ); - bucket.records += 1; - if (priced.costSource === "unpriced") bucket.unpricedRecords += 1; - if (priced.costSource === "providerReported") bucket.providerReportedRecords += 1; if (record.sessionId.length > 0) bucket.sessions.add(record.sessionId); + if (this.#sessions !== null && record.sessionId.length > 0) { + const key = `${record.provider}\u0000${record.sessionId}`; + let session = this.#sessions.get(key); + if (session === undefined) { + session = { + provider: record.provider, + sessionId: record.sessionId, + totals: { ...EMPTY_TOTALS }, + costUsd: 0, + categoryCostUsd: null, + records: 0, + unpricedRecords: 0, + providerReportedRecords: 0, + }; + this.#sessions.set(key, session); + } + accumulatePriced(session, record.totals, priced.costUsd, priced.categoryCostUsd); + if (priced.costSource === "unpriced") session.unpricedRecords += 1; + if (priced.costSource === "providerReported") session.providerReportedRecords += 1; + } return true; } @@ -302,13 +315,71 @@ export class UsageAggregator { outOfWindow: this.#outOfWindow, }; } + + /** Empty unless `collectSessions` was set. Omits records that never named a session. */ + finishSessions(): readonly UsageSessionSpend[] { + if (this.#sessions === null) return []; + const sessions: UsageSessionSpend[] = []; + for (const session of this.#sessions.values()) { + const category = session.categoryCostUsd; + sessions.push({ + provider: session.provider, + sessionId: session.sessionId, + totals: { ...session.totals }, + costUsd: session.costUsd, + ...(category === null + ? {} + : { + categoryCostUsd: { + input: roundUsd(category.input), + cacheRead: roundUsd(category.cacheRead), + cacheWrite: roundUsd(category.cacheWrite), + output: roundUsd(category.output), + }, + }), + costSource: resolveCostSource(session), + records: session.records, + unpricedRecords: session.unpricedRecords, + }); + } + sessions.sort( + (a, b) => a.provider.localeCompare(b.provider) || a.sessionId.localeCompare(b.sessionId), + ); + return sessions; + } } function roundUsd(value: number): number { return Math.round(value * 1e6) / 1e6; } -function resolveCostSource(bucket: MutableBucket): UsageBucket["costSource"] { +function accumulatePriced( + target: PricedAccumulator, + totals: UsageTokenTotals, + costUsd: number, + categoryCostUsd: UsageCategoryCost | null, +): void { + target.totals.uncachedInputTokens += totals.uncachedInputTokens; + target.totals.cachedInputTokens += totals.cachedInputTokens; + target.totals.cacheCreationTokens += totals.cacheCreationTokens; + target.totals.outputTokens += totals.outputTokens; + target.totals.reasoningTokens += totals.reasoningTokens; + target.costUsd += costUsd; + target.records += 1; + if (categoryCostUsd === null) return; + const current = target.categoryCostUsd; + target.categoryCostUsd = + current === null + ? categoryCostUsd + : { + input: current.input + categoryCostUsd.input, + cacheRead: current.cacheRead + categoryCostUsd.cacheRead, + cacheWrite: current.cacheWrite + categoryCostUsd.cacheWrite, + output: current.output + categoryCostUsd.output, + }; +} + +function resolveCostSource(bucket: PricedAccumulator): UsageBucket["costSource"] { if (bucket.unpricedRecords === bucket.records) return "unpriced"; if (bucket.providerReportedRecords === bucket.records) return "providerReported"; return "modelPriced"; diff --git a/apps/server/src/usage/usageSessionSpendInput.test.ts b/apps/server/src/usage/usageSessionSpendInput.test.ts new file mode 100644 index 000000000000..1a364428aba2 --- /dev/null +++ b/apps/server/src/usage/usageSessionSpendInput.test.ts @@ -0,0 +1,104 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { + mergeSessionSpendRefs, + sessionRefsForProviderThreads, + sessionSpendSourcesIncomplete, +} from "./usageSessionSpendInput.ts"; +import { MAX_SESSION_SPEND_KEYS } from "./UsageService.ts"; + +function providerThread( + driver: string, + nativeId: string | null, + strength: "strong" | "weak" | "none" = "strong", +) { + return { driver, nativeThreadRef: nativeId === null ? null : { nativeId, strength } }; +} + +describe("sessionRefsForProviderThreads", () => { + it("keeps joinable sessions and skips ids that are not usage sessions", () => { + expect( + sessionRefsForProviderThreads([ + providerThread("codex", "session-1"), + providerThread("claudeAgent", "session-2"), + providerThread("claudeAgent", "parent:task:task-1"), + providerThread("cursor", "conversation-1"), + providerThread("pi", "C:\\sessions\\pi.jsonl"), + providerThread("grok", "session-3", "weak"), + providerThread("opencode", null), + ]), + ).toEqual([ + { provider: "codex", sessionId: "session-1" }, + { provider: "claude", sessionId: "session-2" }, + ]); + }); +}); + +describe("mergeSessionSpendRefs", () => { + it("counts a session once when a thread resolves a session the caller listed", () => { + const merged = mergeSessionSpendRefs( + [{ provider: "codex", sessionId: "shared" }], + [ + { provider: "codex", sessionId: "shared" }, + { provider: "grok", sessionId: "other" }, + ], + ); + expect(merged).toEqual({ + sessions: [ + { provider: "codex", sessionId: "shared" }, + { provider: "grok", sessionId: "other" }, + ], + truncatedSessions: 0, + }); + }); + + it("reports sessions dropped past the per-read cap", () => { + const resolved = Array.from({ length: MAX_SESSION_SPEND_KEYS + 3 }, (_, index) => ({ + provider: "codex" as const, + sessionId: `session-${index}`, + })); + const merged = mergeSessionSpendRefs([], resolved); + expect(merged.sessions).toHaveLength(MAX_SESSION_SPEND_KEYS); + expect(merged.truncatedSessions).toBe(3); + }); +}); + +describe("sessionSpendSourcesIncomplete", () => { + const source = (provider: string, status: "ok" | "missing" | "partial" | "failed") => ({ + status, + fingerprint: { provider }, + }); + + it("flags a partial or failed source even when it returned no rows", () => { + expect( + sessionSpendSourcesIncomplete( + [source("codex", "ok"), source("grok", "failed")], + [{ provider: "grok", sessionId: "session-1" }], + ), + ).toBe(true); + expect( + sessionSpendSourcesIncomplete( + [source("grok", "partial")], + [ + { provider: "grok", sessionId: "session-1" }, + { provider: "codex", sessionId: "session-2" }, + ], + ), + ).toBe(true); + }); + + it("stays complete when unrelated providers are partial", () => { + expect( + sessionSpendSourcesIncomplete( + [source("grok", "failed"), source("codex", "ok")], + [{ provider: "codex", sessionId: "session-1" }], + ), + ).toBe(false); + expect( + sessionSpendSourcesIncomplete( + [source("codex", "missing")], + [{ provider: "codex", sessionId: "session-1" }], + ), + ).toBe(false); + }); +}); diff --git a/apps/server/src/usage/usageSessionSpendInput.ts b/apps/server/src/usage/usageSessionSpendInput.ts new file mode 100644 index 000000000000..d4241f96b8b1 --- /dev/null +++ b/apps/server/src/usage/usageSessionSpendInput.ts @@ -0,0 +1,62 @@ +import type { UsageSessionRef } from "@t3tools/contracts"; +import { joinableUsageSessionRef, uniqueUsageSessionRefs } from "@t3tools/shared/usageSessionRefs"; + +import { MAX_SESSION_SPEND_KEYS } from "./UsageService.ts"; + +export interface ProviderThreadSessionSource { + readonly driver: string; + readonly nativeThreadRef: { + readonly nativeId: string | null; + readonly strength: "strong" | "weak" | "none"; + } | null; +} + +/** Provider sessions a T3 thread's provider threads can be priced by. */ +export function sessionRefsForProviderThreads( + providerThreads: readonly ProviderThreadSessionSource[], +): UsageSessionRef[] { + const refs: UsageSessionRef[] = []; + for (const providerThread of providerThreads) { + const ref = joinableUsageSessionRef( + providerThread.driver, + providerThread.nativeThreadRef?.nativeId ?? null, + providerThread.nativeThreadRef?.strength ?? "none", + ); + if (ref !== null) refs.push(ref); + } + return refs; +} + +/** + * Explicit sessions first, then thread-resolved ones. One read prices at most + * `MAX_SESSION_SPEND_KEYS` sessions, so the caller learns what it dropped instead of quietly + * showing a short total. + */ +export function mergeSessionSpendRefs( + explicit: readonly UsageSessionRef[], + resolved: readonly UsageSessionRef[], +): { readonly sessions: readonly UsageSessionRef[]; readonly truncatedSessions: number } { + const unique = uniqueUsageSessionRefs([...explicit, ...resolved]); + const sessions = unique.slice(0, MAX_SESSION_SPEND_KEYS); + return { sessions, truncatedSessions: unique.length - sessions.length }; +} + +/** + * True when a source that would have covered one of the requested sessions is partial or failed. + * Matching on the request rather than on returned rows keeps a provider whose scan failed, and so + * returned nothing, counted as incomplete instead of looking like a complete total. + */ +export function sessionSpendSourcesIncomplete( + sources: readonly { + readonly status: "ok" | "missing" | "partial" | "failed"; + readonly fingerprint: { readonly provider: string }; + }[], + sessions: readonly UsageSessionRef[], +): boolean { + const wanted = new Set(sessions.map((session) => session.provider)); + return sources.some( + (source) => + wanted.has(source.fingerprint.provider) && + (source.status === "partial" || source.status === "failed"), + ); +} diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index e1d6303b2350..55fdec8c0d72 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -93,6 +93,8 @@ import { type ProjectId, type ProviderDriverKind, type ProviderInstanceId, + type UsageSessionRef, + type UsageSessionSpendInput, ThreadId, type TerminalMetadataStreamEvent, type PullRequestRef, @@ -164,6 +166,10 @@ import { projectThreadProjectionForWire, } from "./orchestration-v2/WireProjection.ts"; import { dedupeSnapshotBodies } from "@t3tools/shared/orchestrationV2DedupedBodies"; +import { + mergeSessionSpendRefs, + sessionRefsForProviderThreads, +} from "./usage/usageSessionSpendInput.ts"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; @@ -1858,6 +1864,51 @@ export const layerWsRpc = ( return result; }); + /** + * Prices a scope by T3 thread. The server resolves each thread's provider sessions, so a + * caller holding a thread list does not have to open every thread to price it. + */ + const readSessionSpend = Effect.fn("ws.usage.readSessionSpend")(function* ( + input: UsageSessionSpendInput, + ) { + const threadIds = input.threadIds ?? []; + const resolved: UsageSessionRef[] = []; + let unresolvedThreads = 0; + if (threadIds.length > 0) { + const providerThreads = yield* Effect.forEach( + threadIds, + (threadId) => + threadManagement.getThreadRecords(threadId, ["providerThreads"]).pipe( + Effect.map((records) => records.providerThreads), + Effect.orElseSucceed(() => null), + ), + { concurrency: 8 }, + ); + for (const providerThreadList of providerThreads) { + if (providerThreadList === null) { + unresolvedThreads += 1; + continue; + } + resolved.push(...sessionRefsForProviderThreads(providerThreadList)); + } + } + const merged = mergeSessionSpendRefs(input.sessions, resolved); + const result = yield* usage.readSessionSpend({ + sinceDay: input.sinceDay, + untilDay: input.untilDay, + timeZone: input.timeZone, + ...(input.awaitRefresh === undefined ? {} : { awaitRefresh: input.awaitRefresh }), + sessions: merged.sessions, + }); + return { + ...result, + ...(merged.truncatedSessions === 0 + ? {} + : { truncatedSessions: merged.truncatedSessions }), + ...(unresolvedThreads === 0 ? {} : { unresolvedThreads }), + }; + }); + const handlers = ServerWsRpcGroup.of({ [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command) => Effect.annotateCurrentSpan({ @@ -2523,6 +2574,7 @@ export const layerWsRpc = ( [WS_METHODS.serverGetResourceTelemetryHistory]: (input) => resourceTelemetry.readHistory(input), [WS_METHODS.serverGetUsageSummary]: (input) => usage.readSummary(input), + [WS_METHODS.serverGetSessionSpend]: (input) => readSessionSpend(input), [WS_METHODS.serverRefreshUsageRates]: (_input) => usage.refreshRates, [WS_METHODS.serverRetryResourceTelemetry]: (_input) => resourceTelemetry.retry, [WS_METHODS.serverCheckOtlpEndpoint]: (input) => otlpEndpointCheck.check(input), diff --git a/apps/web/src/components/chat/ChatHeader.tsx b/apps/web/src/components/chat/ChatHeader.tsx index ec92540efd1b..e8e86830dd1f 100644 --- a/apps/web/src/components/chat/ChatHeader.tsx +++ b/apps/web/src/components/chat/ChatHeader.tsx @@ -41,6 +41,8 @@ import { cn } from "~/lib/utils"; import { textDirection } from "~/textDirection"; import { useClientSettings } from "../../hooks/useSettings"; import { GuestShareIndicator } from "./GuestShareControls"; +import { ProjectSpendLabel } from "./ProjectSpendLabel"; +import { ThreadSpendControl } from "./ThreadSpendControl"; interface ChatHeaderProps { activeThreadEnvironmentId: EnvironmentId; @@ -307,6 +309,12 @@ export const ChatHeader = memo(function ChatHeader({ New thread in {activeProjectName} + + + / @@ -423,6 +431,9 @@ export const ChatHeader = memo(function ChatHeader({ ) : null} + {isServerThread ? ( + + ) : null} {isServerThread ? : null} ); diff --git a/apps/web/src/components/chat/ProjectSpendLabel.tsx b/apps/web/src/components/chat/ProjectSpendLabel.tsx new file mode 100644 index 000000000000..81da70fb9d57 --- /dev/null +++ b/apps/web/src/components/chat/ProjectSpendLabel.tsx @@ -0,0 +1,158 @@ +import { useAtomValue } from "@effect/atom-react"; +import { + AuthDiagnosticsReadScope, + MAX_SESSION_SPEND_THREADS, + type EnvironmentId, + type ProjectId, + type UsageSessionSpendResult, +} from "@t3tools/contracts"; +import { makeWindow } from "@t3tools/shared/usageFormat"; +import { useNavigate } from "@tanstack/react-router"; +import * as Option from "effect/Option"; +import { AsyncResult, Atom } from "effect/reactivity"; + +import { cn } from "~/lib/utils"; +import { useThreadShellsForProjectRefs } from "../../state/entities"; +import { serverEnvironment } from "../../state/server"; +import { useEnvironmentScope } from "../../state/session"; +import { + buildProjectSpendView, + formatSpendLabel, + formatSpendMoney, + spendResultIncomplete, + THREAD_SPEND_WINDOW_DAYS, +} from "./threadSpend"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; + +const idleSpendAtom = Atom.make(AsyncResult.success(null)).pipe( + Atom.withLabel("project-spend:idle"), +); + +/** One read prices at most this many sessions; past it the label stays a lower bound. */ +const MAX_QUERIED_THREADS = MAX_SESSION_SPEND_THREADS; + +/** Project API spend, shown in the chat header next to the project name. */ +export function ProjectSpendLabel(props: { + environmentId: EnvironmentId | null; + projectId: ProjectId | null; +}) { + const navigate = useNavigate(); + const shells = useThreadShellsForProjectRefs( + props.environmentId === null || props.projectId === null + ? [] + : [{ environmentId: props.environmentId, projectId: props.projectId }], + ); + const environmentId = props.environmentId ?? shells[0]?.environmentId ?? null; + const canRead = useEnvironmentScope(environmentId, AuthDiagnosticsReadScope); + // The server resolves each thread to its provider sessions, so the header never has to open + // every thread's stream. Newest threads first so a capped read keeps the busiest work. + const threadIds = + environmentId === null + ? [] + : shells + .filter((shell) => shell.environmentId === environmentId && shell.deletedAt === null) + .sort((a, b) => b.updatedAt.localeCompare(a.updatedAt)) + .slice(0, MAX_QUERIED_THREADS) + .map((shell) => shell.id); + const spendWindow = makeWindow(THREAD_SPEND_WINDOW_DAYS); + const spendAtom = + !canRead || environmentId === null || threadIds.length === 0 + ? idleSpendAtom + : serverEnvironment.usageSessionSpend({ + environmentId, + input: { + sinceDay: spendWindow.sinceDay, + untilDay: spendWindow.untilDay, + timeZone: spendWindow.timeZone, + sessions: [], + threadIds, + }, + }); + const result = useAtomValue( + spendAtom as Atom.Atom>, + ); + if (!canRead || environmentId === null || props.projectId === null) return null; + + const value = Option.getOrNull(AsyncResult.value(result)); + const failed = result._tag === "Failure"; + const sources = value?.sources ?? []; + const rows = value?.sessions ?? null; + const projectThreadCount = + environmentId === null + ? 0 + : shells.filter((shell) => shell.environmentId === environmentId && shell.deletedAt === null) + .length; + // Threads past the cap are never requested, so the server cannot report them as dropped. + const uncountedThreads = Math.max(0, projectThreadCount - threadIds.length); + const view = buildProjectSpendView({ + priced: failed ? [] : rows, + threads: threadIds.length, + partial: spendResultIncomplete(value) || uncountedThreads > 0, + refreshing: + sources.some((source) => source.refreshing === true) || (result.waiting && value !== null), + sinceDay: value?.sinceDay ?? null, + untilDay: value?.untilDay ?? null, + }); + const label = failed + ? "Usage unavailable" + : view.kind === "loading" + ? "project spend" + : view.kind === "empty" + ? view.refreshing + ? "project spend scanning" + : view.partial + ? "project usage incomplete" + : "no project usage" + : formatSpendLabel(view.money, view.partial); + const formatted = view.kind === "ready" ? formatSpendMoney(view.money) : null; + const detail = [ + "Project spend: API estimate for T3 threads in this project, not the subscription bill.", + view.kind === "ready" && view.money.unpriced + ? "Only tokens are shown; no price for these models." + : null, + view.kind === "ready" && view.money.unpricedRecords > 0 && !view.money.unpriced + ? "Some records have no price." + : null, + view.kind === "ready" && view.partial ? "Some usage could not be read." : null, + view.kind === "ready" && uncountedThreads > 0 + ? `Only the newest ${threadIds.length} threads are priced; ${uncountedThreads} are not counted.` + : null, + view.kind === "empty" && view.partial + ? "No usage was read, but some sources were incomplete, so usage may exist." + : null, + view.kind === "empty" && view.refreshing ? "The usage scan is still running." : null, + view.kind === "ready" && view.sinceDay !== null && view.untilDay !== null + ? `${view.sinceDay} to ${view.untilDay}` + : null, + ] + .filter((line) => line !== null) + .join(" "); + const dim = !failed && view.kind === "ready" && view.refreshing; + + return ( + + void navigate({ to: "/usage" })} + /> + } + > + {label} + + + {detail} + + + ); +} diff --git a/apps/web/src/components/chat/ThreadSpendControl.tsx b/apps/web/src/components/chat/ThreadSpendControl.tsx new file mode 100644 index 000000000000..f275811da5e9 --- /dev/null +++ b/apps/web/src/components/chat/ThreadSpendControl.tsx @@ -0,0 +1,227 @@ +import { useAtomValue } from "@effect/atom-react"; +import { + AuthDiagnosticsReadScope, + type EnvironmentId, + type ThreadId, + type UsageSessionSpendResult, +} from "@t3tools/contracts"; +import { formatUsd, makeWindow } from "@t3tools/shared/usageFormat"; +import { useNavigate } from "@tanstack/react-router"; +import * as Option from "effect/Option"; +import { AsyncResult, Atom } from "effect/reactivity"; + +import { cn } from "~/lib/utils"; +import { useThreadShells } from "../../state/entities"; +import { serverEnvironment } from "../../state/server"; +import { useEnvironmentScope } from "../../state/session"; +import { Popover, PopoverPopup, PopoverTrigger } from "../ui/popover"; +import { useJoinedSpendSessions } from "./spendSessions"; +import { + buildThreadSpendView, + formatSpendLabel, + formatSpendMoney, + spendResultIncomplete, + THREAD_SPEND_WINDOW_DAYS, + type SpendMoney, + type ThreadSpendView, +} from "./threadSpend"; + +const idleSpendAtom = Atom.make(AsyncResult.success(null)).pipe( + Atom.withLabel("thread-spend:idle"), +); + +function MoneyLine(props: { label: string; money: SpendMoney }) { + const formatted = formatSpendMoney(props.money); + return ( +
+ {props.label} + + {formatted.dollars} + · {formatted.tokens} + +
+ ); +} + +function viewLabel(view: ThreadSpendView, failed: boolean): string { + if (failed) return "Usage unavailable"; + if (view.kind === "loading") return "Spend"; + if (view.kind === "empty") { + if (view.refreshing) return "Scanning"; + return view.partial ? "Usage unknown" : "No usage yet"; + } + if (view.kind === "unsplit") return "In parent"; + return formatSpendLabel(view.own, view.partial); +} + +/** Thread API spend, shown in the chat header next to the thread title. */ +export function ThreadSpendControl(props: { + environmentId: EnvironmentId | null; + threadId: ThreadId | null; +}) { + const navigate = useNavigate(); + const canRead = useEnvironmentScope(props.environmentId, AuthDiagnosticsReadScope); + const shells = useThreadShells(canRead && props.threadId !== null); + const joined = useJoinedSpendSessions({ + enabled: canRead, + environmentId: props.environmentId, + shells, + focusThreadId: props.threadId, + }); + const nodes = joined.nodes; + const thread = nodes.find((candidate) => candidate.id === props.threadId) ?? null; + const sessions = joined.sessions; + const spendWindow = makeWindow(THREAD_SPEND_WINDOW_DAYS); + const spendAtom = + !canRead || props.environmentId === null || thread?.sessions == null || sessions.length === 0 + ? idleSpendAtom + : serverEnvironment.usageSessionSpend({ + environmentId: props.environmentId, + input: { + sinceDay: spendWindow.sinceDay, + untilDay: spendWindow.untilDay, + timeZone: spendWindow.timeZone, + sessions, + }, + }); + const result = useAtomValue( + spendAtom as Atom.Atom>, + ); + if (!canRead || props.environmentId === null) return null; + + const value = Option.getOrNull(AsyncResult.value(result)); + const failed = result._tag === "Failure"; + const requested = thread?.sessions != null && thread.sessions.length > 0; + const sources = value?.sources ?? []; + const view = buildThreadSpendView({ + thread: props.threadId === null ? null : thread, + threads: nodes, + priced: failed || !requested ? [] : (value?.sessions ?? null), + refreshing: + sources.some((source) => source.refreshing === true) || (result.waiting && value !== null), + partial: spendResultIncomplete(value), + sinceDay: value?.sinceDay ?? null, + untilDay: value?.untilDay ?? null, + }); + const label = joined.overCap ? "Too many sessions" : viewLabel(view, failed); + const own = view.kind === "ready" ? formatSpendMoney(view.own) : null; + const dim = !failed && view.kind === "ready" && view.refreshing; + + return ( + <> + {joined.loaders} + + + } + > + {label} + + +
+
Thread spend
+ {failed ?

Usage could not be read.

: null} + {!failed && joined.overCap ? ( +

+ This thread has more sessions than can be priced at once. +

+ ) : null} + {!failed && !joined.overCap && view.kind === "loading" ? ( +

Reading usage…

+ ) : null} + {!failed && view.kind === "empty" ? ( +

+ {view.refreshing + ? "No usage priced yet; the scan is still running." + : view.partial + ? "No usage could be read for this thread." + : "No usage yet for this thread."} +

+ ) : null} + {!failed && view.kind === "unsplit" ? ( +

+ This subagent's spend is inside {view.parentTitle ?? "the parent thread"} and + is not split. +

+ ) : null} + {!failed && !joined.overCap && view.kind === "ready" ? ( + <> + + {view.includingSubagents !== null ? ( + + ) : null} + {view.subagentsNotIncluded ? ( +

Subagent spend is not included.

+ ) : null} + {view.isSubagent && view.parentTitle !== null ? ( +

Subagent of {view.parentTitle}

+ ) : null} + {view.own.unpriced ? ( +

+ No price for these models, so only tokens are shown. +

+ ) : view.own.unpricedRecords > 0 ? ( +

+ {view.own.unpricedRecords} record + {view.own.unpricedRecords === 1 ? " has" : "s have"} no price; the dollar figure + is a lower bound. +

+ ) : null} + {view.partial ? ( +

+ The usage scan is incomplete, so this is a lower bound. +

+ ) : null} + {(value?.unresolvedThreads ?? 0) > 0 ? ( +

+ {value?.unresolvedThreads} thread + {(value?.unresolvedThreads ?? 0) === 1 ? " could" : "s could"} not be resolved. +

+ ) : null} + {view.own.category !== null ? ( +
+ Input {formatUsd(view.own.category.input)} + Cache read {formatUsd(view.own.category.cacheRead)} + Cache write {formatUsd(view.own.category.cacheWrite)} + Output {formatUsd(view.own.category.output)} +
+ ) : null} + {view.sinceDay !== null && view.untilDay !== null ? ( +

+ {view.sinceDay} to {view.untilDay} + {view.refreshing ? " · Still scanning" : ""} +

+ ) : null} + + ) : null} +

API estimate, not the subscription bill.

+ +
+ + + + ); +} diff --git a/apps/web/src/components/chat/spendSessions.tsx b/apps/web/src/components/chat/spendSessions.tsx new file mode 100644 index 000000000000..49a2942c6c11 --- /dev/null +++ b/apps/web/src/components/chat/spendSessions.tsx @@ -0,0 +1,163 @@ +import type { EnvironmentId, ThreadId, UsageSessionRef } from "@t3tools/contracts"; +import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/models"; +import * as Option from "effect/Option"; +import { useEffect, useState, type ReactNode } from "react"; + +import { useStableCallback } from "../../hooks/useStableCallback"; +import { useEnvironmentThread } from "../../state/threads"; +import { + collectSessionRefs, + descendantSubagentIds, + joinableSessionRef, + sessionListKey, + type SpendThreadNode, +} from "./threadSpend"; + +/** Live thread streams are expensive. Descendants of the open thread stay under this cap. */ +const MAX_LOADED_DESCENDANTS = 24; +/** Matches UsageService. A larger set is not priced, so a short total is never shown. */ +const MAX_PRICED_SESSIONS = 200; + +function sessionsFromState( + requested: boolean, + state: ReturnType, +): readonly UsageSessionRef[] | null { + if (!requested) return null; + const projection = Option.getOrNull(state.data); + if (projection === null) { + return state.status === "live" || state.status === "cached" || state.status === "deleted" + ? [] + : null; + } + const refs: UsageSessionRef[] = []; + for (const providerThread of projection.providerThreads) { + const ref = joinableSessionRef( + providerThread.driver, + providerThread.nativeThreadRef?.nativeId ?? null, + providerThread.nativeThreadRef?.strength ?? "none", + ); + if (ref !== null) refs.push(ref); + } + return refs; +} + +function JoinedSessionLoader(props: { + environmentId: EnvironmentId; + threadId: ThreadId; + onSessions: (threadId: string, sessions: readonly UsageSessionRef[] | null) => void; +}) { + const state = useEnvironmentThread(props.environmentId, props.threadId); + const sessions = sessionsFromState(true, state); + const report = useStableCallback(props.onSessions); + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- reports a child thread's sessions, which arrive on that thread's own live stream, to the parent's state on every stream update. + useEffect(() => { + report(props.threadId, sessions); + }, [report, props.threadId, sessions]); + return null; +} + +function shellNode( + shell: EnvironmentThreadShell, + sessions: readonly UsageSessionRef[] | null, +): SpendThreadNode { + return { + id: shell.id, + title: shell.title, + environmentId: shell.environmentId, + deleted: shell.deletedAt !== null, + relationshipToParent: shell.lineage.relationshipToParent, + parentThreadId: shell.lineage.parentThreadId, + activeProviderThreadId: shell.activeProviderThreadId, + sessions, + }; +} + +/** + * Sessions for the open thread, plus its subagent descendants. + * Other project threads stay unloaded so each one does not open a live snapshot stream. + */ +export function useJoinedSpendSessions(input: { + readonly enabled: boolean; + readonly environmentId: EnvironmentId | null; + readonly shells: readonly EnvironmentThreadShell[]; + readonly focusThreadId: ThreadId | null; +}): { + readonly nodes: readonly SpendThreadNode[]; + readonly sessionKey: string; + readonly sessions: readonly UsageSessionRef[]; + readonly overCap: boolean; + readonly loaders: ReactNode; +} { + const focusState = useEnvironmentThread( + input.enabled ? input.environmentId : null, + input.enabled ? input.focusThreadId : null, + ); + const focusSessions = sessionsFromState( + input.enabled && input.focusThreadId !== null, + focusState, + ); + const descendantIds = + !input.enabled || input.focusThreadId === null || input.environmentId === null + ? [] + : descendantSubagentIds( + input.focusThreadId, + input.shells.map((shell) => shellNode(shell, null)), + ) + .filter((id) => + input.shells.some( + (shell) => + shell.id === id && + shell.environmentId === input.environmentId && + shell.deletedAt === null, + ), + ) + .sort() + .slice(0, MAX_LOADED_DESCENDANTS); + const [loaded, setLoaded] = useState>( + () => new Map(), + ); + const onSessions = useStableCallback( + (threadId: string, sessions: readonly UsageSessionRef[] | null) => { + setLoaded((current) => { + if ( + sessionListKey(current.get(threadId)) === sessionListKey(sessions) && + current.has(threadId) + ) { + return current; + } + const next = new Map(current); + next.set(threadId, sessions); + return next; + }); + }, + ); + const nodes = input.shells.map((shell) => { + if (shell.id === input.focusThreadId) return shellNode(shell, focusSessions); + if (descendantIds.includes(shell.id)) { + return shellNode(shell, loaded.get(shell.id) ?? null); + } + return shellNode(shell, null); + }); + const sessions = collectSessionRefs( + nodes.filter( + (node) => + !node.deleted && + (input.environmentId === null || node.environmentId === input.environmentId), + ), + ); + const overCap = sessions.length > MAX_PRICED_SESSIONS; + const pricedSessions = overCap ? [] : sessions; + const key = pricedSessions.map((ref) => `${ref.provider}:${ref.sessionId}`).join("\n"); + const loaders = + input.enabled && input.environmentId !== null + ? descendantIds.map((threadId) => ( + + )) + : null; + return { nodes, sessionKey: key, sessions: pricedSessions, overCap, loaders }; +} diff --git a/apps/web/src/components/chat/threadSpend.test.ts b/apps/web/src/components/chat/threadSpend.test.ts new file mode 100644 index 000000000000..bd66344561e6 --- /dev/null +++ b/apps/web/src/components/chat/threadSpend.test.ts @@ -0,0 +1,329 @@ +import { describe, expect, it } from "@effect/vitest"; + +import type { UsageSessionSpend } from "@t3tools/contracts"; + +import { + buildProjectSpendView, + buildThreadSpendView, + descendantSubagentIds, + formatSpendLabel, + joinableSessionRef, + sessionListKey, + spendResultIncomplete, + type SpendThreadNode, +} from "./threadSpend.ts"; + +function node(overrides: Partial & Pick): SpendThreadNode { + return { + title: overrides.id, + environmentId: "env-1", + deleted: false, + relationshipToParent: null, + parentThreadId: null, + activeProviderThreadId: "provider-thread", + sessions: [], + ...overrides, + }; +} + +function priced( + sessionId: string, + costUsd: number, + tokens: number, + unpricedRecords = 0, +): UsageSessionSpend { + return { + provider: "codex", + sessionId, + totals: { + uncachedInputTokens: tokens, + cachedInputTokens: 0, + cacheCreationTokens: 0, + outputTokens: 0, + reasoningTokens: 0, + }, + costUsd, + costSource: unpricedRecords > 0 ? "unpriced" : "modelPriced", + records: 1, + unpricedRecords, + }; +} + +describe("joinableSessionRef", () => { + it("joins known native ids and skips ids that are not usage sessions", () => { + expect(joinableSessionRef("codex", "session-1")).toEqual({ + provider: "codex", + sessionId: "session-1", + }); + expect(joinableSessionRef("claudeAgent", "parent:task:task-1")).toBeNull(); + expect(joinableSessionRef("cursor", "conversation-1")).toBeNull(); + expect(joinableSessionRef("antigravity", "conversation-1")).toBeNull(); + expect(joinableSessionRef("pi", "C:\\sessions\\pi.jsonl")).toBeNull(); + expect(joinableSessionRef("pi", "session-from-file")).toEqual({ + provider: "pi", + sessionId: "session-from-file", + }); + expect(joinableSessionRef("codex", "session-1", "weak")).toBeNull(); + expect(joinableSessionRef("codex", "session-1", "none")).toBeNull(); + }); +}); + +describe("buildThreadSpendView", () => { + const parent = node({ + id: "parent", + title: "Parent", + sessions: [{ provider: "codex", sessionId: "parent-session" }], + }); + const child = node({ + id: "child", + title: "Audit", + relationshipToParent: "subagent", + parentThreadId: "parent", + sessions: [{ provider: "codex", sessionId: "child-session" }], + }); + const fork = node({ + id: "fork", + relationshipToParent: "fork", + parentThreadId: "parent", + sessions: [{ provider: "codex", sessionId: "fork-session" }], + }); + const unsplit = node({ + id: "native-child", + title: "Native child", + relationshipToParent: "subagent", + parentThreadId: "parent", + activeProviderThreadId: null, + sessions: [], + }); + + it("keeps forks out and adds a loaded subagent once", () => { + expect(new Set(descendantSubagentIds("parent", [parent, child, fork, unsplit]))).toEqual( + new Set(["child", "native-child"]), + ); + const view = buildThreadSpendView({ + thread: parent, + threads: [parent, child, fork], + priced: [priced("parent-session", 1.5, 1000), priced("child-session", 0.5, 400)], + refreshing: false, + partial: false, + sinceDay: "2026-07-01", + untilDay: "2026-08-07", + }); + expect(view).toMatchObject({ + kind: "ready", + own: { costUsd: 1.5, tokens: 1000 }, + includingSubagents: { costUsd: 2, tokens: 1400 }, + subagentsNotIncluded: false, + }); + }); + + it("does not treat an unjoined provider thread as zero spend", () => { + const hidden = node({ + id: "child", + relationshipToParent: "subagent", + parentThreadId: "parent", + activeProviderThreadId: "provider-thread", + sessions: [], + }); + const view = buildThreadSpendView({ + thread: parent, + threads: [parent, hidden], + priced: [priced("parent-session", 1.5, 1000)], + refreshing: false, + partial: false, + sinceDay: null, + untilDay: null, + }); + expect(view).toMatchObject({ + kind: "ready", + own: { costUsd: 1.5, tokens: 1000 }, + includingSubagents: null, + subagentsNotIncluded: true, + }); + }); + + it("does not invent $0 when a subagent session is not loaded", () => { + const unloadedChild = node({ + id: "child", + relationshipToParent: "subagent", + parentThreadId: "parent", + sessions: null, + }); + const view = buildThreadSpendView({ + thread: parent, + threads: [parent, unloadedChild], + priced: [priced("parent-session", 1.5, 1000)], + refreshing: false, + partial: false, + sinceDay: null, + untilDay: null, + }); + expect(view).toMatchObject({ + kind: "ready", + includingSubagents: null, + subagentsNotIncluded: true, + }); + }); + + it("says an unsplit native subagent is inside the parent", () => { + const view = buildThreadSpendView({ + thread: unsplit, + threads: [parent, unsplit], + priced: [], + refreshing: false, + partial: false, + sinceDay: null, + untilDay: null, + }); + expect(view).toEqual({ kind: "unsplit", parentTitle: "Parent" }); + }); + + it("says there is no usage when the session has no priced records", () => { + const view = buildThreadSpendView({ + thread: parent, + threads: [parent], + priced: [], + refreshing: false, + partial: false, + sinceDay: null, + untilDay: null, + }); + expect(view).toEqual({ kind: "empty", partial: false, refreshing: false }); + }); + + it("carries scan state on an empty read so it cannot read as settled", () => { + const view = buildThreadSpendView({ + thread: parent, + threads: [parent], + priced: [], + refreshing: true, + partial: true, + sinceDay: null, + untilDay: null, + }); + expect(view).toEqual({ kind: "empty", partial: true, refreshing: true }); + }); + + it("keeps a priced zero", () => { + const view = buildThreadSpendView({ + thread: parent, + threads: [parent], + priced: [priced("parent-session", 0, 10)], + refreshing: true, + partial: true, + sinceDay: "2026-07-01", + untilDay: "2026-08-07", + }); + expect(view).toMatchObject({ + kind: "ready", + own: { costUsd: 0, tokens: 10, unpriced: false }, + refreshing: true, + partial: true, + }); + }); +}); + +describe("buildProjectSpendView", () => { + it("counts a session once when the read repeats a row", () => { + const view = buildProjectSpendView({ + priced: [priced("shared", 3, 90), { ...priced("shared", 3, 90) }], + threads: 2, + partial: false, + refreshing: false, + sinceDay: "2026-07-01", + untilDay: "2026-08-07", + }); + expect(view).toMatchObject({ + kind: "ready", + money: { costUsd: 3, tokens: 90 }, + threads: 2, + partial: false, + }); + }); + + it("stays empty instead of zero when nothing has been priced", () => { + const view = buildProjectSpendView({ + priced: [], + threads: 3, + partial: true, + refreshing: false, + sinceDay: null, + untilDay: null, + }); + expect(view).toEqual({ kind: "empty", partial: true, refreshing: false }); + }); + + it("carries scan state on an empty project read", () => { + const view = buildProjectSpendView({ + priced: [], + threads: 3, + partial: false, + refreshing: true, + sinceDay: null, + untilDay: null, + }); + expect(view).toEqual({ kind: "empty", partial: false, refreshing: true }); + }); + + it("stays loading until a read answers", () => { + expect( + buildProjectSpendView({ + priced: null, + threads: 4, + partial: false, + refreshing: false, + sinceDay: null, + untilDay: null, + }), + ).toEqual({ kind: "loading" }); + }); +}); + +describe("formatSpendLabel", () => { + const money = { + tokens: 1_500_000, + costUsd: 2.5, + unpriced: false, + unpricedRecords: 0, + category: null, + }; + + it("shows dollars and tokens", () => { + expect(formatSpendLabel(money, false)).toBe("$2.50 · 1.50M"); + }); + + it("marks totals that can only be a lower bound", () => { + expect(formatSpendLabel({ ...money, unpricedRecords: 2 }, false)).toBe("≥ $2.50 · 1.50M"); + expect(formatSpendLabel(money, true)).toBe("≥ $2.50 · 1.50M"); + }); + + it("keeps the token total when nothing could be priced", () => { + expect(formatSpendLabel({ ...money, unpriced: true, unpricedRecords: 4 }, true)).toBe( + "unpriced · 1.50M", + ); + }); +}); + +describe("spendResultIncomplete", () => { + it("treats a dropped session, an unresolved thread or a failed source as a lower bound", () => { + expect(spendResultIncomplete({ incomplete: true })).toBe(true); + expect(spendResultIncomplete({ truncatedSessions: 1 })).toBe(true); + expect(spendResultIncomplete({ unresolvedThreads: 2 })).toBe(true); + }); + + it("stays exact for a complete read and for no read at all", () => { + expect(spendResultIncomplete({ incomplete: false, truncatedSessions: 0 })).toBe(false); + expect(spendResultIncomplete(null)).toBe(false); + expect(spendResultIncomplete(undefined)).toBe(false); + }); +}); + +describe("sessionListKey", () => { + it("separates a thread with no joinable sessions from one that has not loaded", () => { + expect(sessionListKey([])).not.toBe(sessionListKey(null)); + expect(sessionListKey(null)).toBe(sessionListKey(undefined)); + expect(sessionListKey([{ provider: "codex", sessionId: "session-1" }])).toBe( + sessionListKey([{ provider: "codex", sessionId: "session-1" }]), + ); + }); +}); diff --git a/apps/web/src/components/chat/threadSpend.ts b/apps/web/src/components/chat/threadSpend.ts new file mode 100644 index 000000000000..8d0f70562cec --- /dev/null +++ b/apps/web/src/components/chat/threadSpend.ts @@ -0,0 +1,330 @@ +import type { UsageCategoryCost, UsageSessionRef, UsageSessionSpend } from "@t3tools/contracts"; +import { formatTokens, formatUsd } from "@t3tools/shared/usageFormat"; +import { usageTotalTokens } from "@t3tools/shared/usageMerge"; +import { + joinableUsageSessionRef, + uniqueUsageSessionRefs, + usageSessionKey, +} from "@t3tools/shared/usageSessionRefs"; + +export { usageTotalTokens }; + +/** Matches the usage scan retention in UsageService. */ +export const THREAD_SPEND_WINDOW_DAYS = 92; + +/** Kept for callers that name the provider-side join rule. */ +export const joinableSessionRef = joinableUsageSessionRef; +export const sessionKey = usageSessionKey; + +export interface SpendThreadNode { + readonly id: string; + readonly title: string; + readonly environmentId: string; + readonly deleted: boolean; + readonly relationshipToParent: "fork" | "subagent" | null; + readonly parentThreadId: string | null; + readonly activeProviderThreadId: string | null; + /** null until that thread's provider sessions are loaded. */ + readonly sessions: readonly UsageSessionRef[] | null; +} + +export interface SpendMoney { + readonly tokens: number; + readonly costUsd: number; + /** Every record is unpriced, so there is no dollar figure at all. */ + readonly unpriced: boolean; + /** Records without a price; the dollar figure is a lower bound while this is above zero. */ + readonly unpricedRecords: number; + readonly category: UsageCategoryCost | null; +} + +/** + * What a thread's spend read can say. `empty` still carries scan state: no priced rows yet is not + * the same as no usage, so a first or partial scan must not read as a settled answer. + */ +export type ThreadSpendView = + | { readonly kind: "empty"; readonly partial: boolean; readonly refreshing: boolean } + | { readonly kind: "unsplit"; readonly parentTitle: string | null } + | { readonly kind: "loading" } + | { + readonly kind: "ready"; + readonly own: SpendMoney; + readonly includingSubagents: SpendMoney | null; + readonly subagentsNotIncluded: boolean; + readonly parentTitle: string | null; + readonly isSubagent: boolean; + readonly refreshing: boolean; + /** A usage source was partial, so every total here is a lower bound. */ + readonly partial: boolean; + readonly sinceDay: string | null; + readonly untilDay: string | null; + }; + +export type ProjectSpendView = + | { readonly kind: "empty"; readonly partial: boolean; readonly refreshing: boolean } + | { readonly kind: "loading" } + | { + readonly kind: "ready"; + readonly money: SpendMoney; + /** Threads the read covered. */ + readonly threads: number; + /** A source was partial or the read dropped sessions, so totals are a lower bound. */ + readonly partial: boolean; + readonly refreshing: boolean; + readonly sinceDay: string | null; + readonly untilDay: string | null; + }; + +export function descendantSubagentIds( + rootId: string, + threads: readonly SpendThreadNode[], +): readonly string[] { + const children = new Map(); + for (const thread of threads) { + if (thread.deleted || thread.relationshipToParent !== "subagent") continue; + if (thread.parentThreadId === null) continue; + const list = children.get(thread.parentThreadId); + if (list === undefined) children.set(thread.parentThreadId, [thread.id]); + else list.push(thread.id); + } + const out: string[] = []; + const seen = new Set(); + const stack = [...(children.get(rootId) ?? [])]; + while (stack.length > 0) { + const id = stack.pop(); + if (id === undefined || seen.has(id)) continue; + seen.add(id); + out.push(id); + for (const child of children.get(id) ?? []) stack.push(child); + } + return out; +} + +/** + * Sums deduped session rows. The server already dedupes a thread-resolved read, but a caller can + * pass explicit sessions alongside thread ids, so the same session can arrive twice. + */ +function sumRows(rows: readonly UsageSessionSpend[]): { + readonly matched: number; + readonly money: SpendMoney; +} { + const seen = new Set(); + let tokens = 0; + let costUsd = 0; + let records = 0; + let unpricedRecords = 0; + let category: UsageCategoryCost | null = null; + let categoryComplete = true; + let matched = 0; + for (const row of rows) { + const key = usageSessionKey(row); + if (seen.has(key)) continue; + seen.add(key); + matched += 1; + tokens += usageTotalTokens(row.totals); + costUsd += row.costUsd; + records += row.records; + unpricedRecords += row.unpricedRecords; + const add = row.categoryCostUsd; + if (add === undefined) { + categoryComplete = false; + continue; + } + category = + category === null + ? add + : { + input: category.input + add.input, + cacheRead: category.cacheRead + add.cacheRead, + cacheWrite: category.cacheWrite + add.cacheWrite, + output: category.output + add.output, + }; + } + if (!categoryComplete) category = null; + return { + matched, + money: { + tokens, + costUsd, + unpriced: records > 0 && unpricedRecords === records, + unpricedRecords, + category, + }, + }; +} + +function sumRefs( + refs: readonly UsageSessionRef[], + priced: ReadonlyMap, +): { readonly matched: number; readonly money: SpendMoney } { + const rows: UsageSessionSpend[] = []; + for (const ref of uniqueUsageSessionRefs(refs)) { + const row = priced.get(usageSessionKey(ref)); + if (row !== undefined) rows.push(row); + } + return sumRows(rows); +} + +function pricedMap(rows: readonly UsageSessionSpend[]): Map { + return new Map(rows.map((row) => [usageSessionKey(row), row])); +} + +export function buildThreadSpendView(input: { + readonly thread: SpendThreadNode | null; + readonly threads: readonly SpendThreadNode[]; + readonly priced: readonly UsageSessionSpend[] | null; + readonly refreshing: boolean; + readonly partial: boolean; + readonly sinceDay: string | null; + readonly untilDay: string | null; +}): ThreadSpendView { + const thread = input.thread; + if (thread === null || thread.deleted) + return { kind: "empty", partial: false, refreshing: false }; + + const parent = + thread.parentThreadId === null + ? null + : (input.threads.find((candidate) => candidate.id === thread.parentThreadId) ?? null); + const parentTitle = parent?.title ?? null; + const isSubagent = thread.relationshipToParent === "subagent"; + if ( + isSubagent && + thread.sessions !== null && + thread.sessions.length === 0 && + thread.activeProviderThreadId === null + ) { + return { kind: "unsplit", parentTitle }; + } + if (thread.sessions === null) return { kind: "loading" }; + if (thread.sessions.length === 0) + return { kind: "empty", partial: input.partial, refreshing: input.refreshing }; + if (input.priced === null) return { kind: "loading" }; + + const priced = pricedMap(input.priced); + const own = sumRefs(thread.sessions, priced); + if (own.matched === 0) + return { kind: "empty", partial: input.partial, refreshing: input.refreshing }; + + const byId = new Map(input.threads.map((candidate) => [candidate.id, candidate])); + const descendants = descendantSubagentIds(thread.id, input.threads); + let subagentsNotIncluded = false; + let includingSubagents: SpendMoney | null = null; + if (descendants.length > 0) { + const childRefs: UsageSessionRef[] = []; + for (const id of descendants) { + const child = byId.get(id); + if (child === undefined || child.sessions === null) { + subagentsNotIncluded = true; + continue; + } + // A provider thread with no joinable usage id is not a zero-cost child. + if (child.sessions.length === 0 && child.activeProviderThreadId !== null) { + subagentsNotIncluded = true; + continue; + } + childRefs.push(...child.sessions); + } + if (!subagentsNotIncluded) { + includingSubagents = sumRefs([...thread.sessions, ...childRefs], priced).money; + } + } + + return { + kind: "ready", + own: own.money, + includingSubagents, + subagentsNotIncluded, + parentTitle, + isSubagent, + refreshing: input.refreshing, + partial: input.partial, + sinceDay: input.sinceDay, + untilDay: input.untilDay, + }; +} + +export function buildProjectSpendView(input: { + readonly priced: readonly UsageSessionSpend[] | null; + readonly threads: number; + readonly partial: boolean; + readonly refreshing: boolean; + readonly sinceDay: string | null; + readonly untilDay: string | null; +}): ProjectSpendView { + if (input.priced === null) return { kind: "loading" }; + const summed = sumRows(input.priced); + if (summed.matched === 0) + return { kind: "empty", partial: input.partial, refreshing: input.refreshing }; + return { + kind: "ready", + money: summed.money, + threads: input.threads, + partial: input.partial, + refreshing: input.refreshing, + sinceDay: input.sinceDay, + untilDay: input.untilDay, + }; +} + +export function formatSpendMoney(money: SpendMoney): { + readonly dollars: string; + readonly tokens: string; +} { + return { + dollars: money.unpriced ? "unpriced" : formatUsd(money.costUsd), + tokens: formatTokens(money.tokens), + }; +} + +/** True when only part of the usage could be counted, so the total is a lower bound. */ +export function spendIsPartial(money: SpendMoney, partial: boolean): boolean { + return partial || (money.unpricedRecords > 0 && !money.unpriced); +} + +export function formatSpendLabel(money: SpendMoney, partial: boolean): string { + const formatted = formatSpendMoney(money); + // A missing dollar figure is already its own signal; do not stack a lower-bound marker on it. + const prefix = !money.unpriced && spendIsPartial(money, partial) ? "≥ " : ""; + return `${prefix}${formatted.dollars} · ${formatted.tokens}`; +} + +/** + * True when the read is anything less than the whole picture for the scope it was asked about: + * an incomplete source, sessions the server dropped at the cap, or threads it could not resolve. + * The server decides the first case because only it knows which sessions a request reaches. + */ +export function spendResultIncomplete( + result: + | { + readonly incomplete?: boolean | undefined; + readonly truncatedSessions?: number | undefined; + readonly unresolvedThreads?: number | undefined; + } + | null + | undefined, +): boolean { + if (result == null) return false; + return ( + result.incomplete === true || + (result.truncatedSessions ?? 0) > 0 || + (result.unresolvedThreads ?? 0) > 0 + ); +} + +/** + * Identity for a loaded session list. Distinguishes "not loaded yet" (null) from "loaded, no + * joinable sessions" ([]), so a subagent that resolves to nothing is not left looking unloaded. + */ +export function sessionListKey(sessions: readonly UsageSessionRef[] | null | undefined): string { + if (sessions == null) return "\u0000unloaded"; + return sessions.map((ref) => usageSessionKey(ref)).join("\n"); +} + +export function collectSessionRefs(threads: readonly SpendThreadNode[]): UsageSessionRef[] { + return uniqueUsageSessionRefs( + threads.flatMap((thread) => + thread.deleted || thread.sessions === null ? [] : thread.sessions, + ), + ); +} diff --git a/apps/web/src/rpc/requestLatencyState.ts b/apps/web/src/rpc/requestLatencyState.ts index c95ddae2afcd..1f81c0be4864 100644 --- a/apps/web/src/rpc/requestLatencyState.ts +++ b/apps/web/src/rpc/requestLatencyState.ts @@ -31,6 +31,7 @@ const pendingRpcAckRequests = new Map(); // lasts as long as the user takes on the forge. const untrackedRpcAckMethods = new Set([ WS_METHODS.serverGetUsageSummary, + WS_METHODS.serverGetSessionSpend, WS_METHODS.serverReportClientActivity, WS_METHODS.sourceControlSignInWait, ]); diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index c8ce8b387874..7f0482ea9576 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -1293,6 +1293,30 @@ export function createServerEnvironmentAtoms( ), ), }), + usageSessionSpend: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:server:usage-session-spend", + tag: WS_METHODS.serverGetSessionSpend, + staleTimeMs: 60_000, + refreshTrigger: ({ environmentId }) => usageScanSettingsAtom(environmentId), + execute: (input, emit) => + request(WS_METHODS.serverGetSessionSpend, input).pipe( + Effect.filterOrElse( + (result) => !result.sources.some((source) => source.refreshing === true), + (result) => + emit(result).pipe( + Effect.andThen( + request(WS_METHODS.serverGetSessionSpend, { ...input, awaitRefresh: true }), + ), + Effect.catch((error) => + Effect.logWarning("Could not refresh slow usage sources.").pipe( + Effect.annotateLogs({ ...safeErrorLogAttributes(error) }), + Effect.as(result), + ), + ), + ), + ), + ), + }), resourceTelemetry: createEnvironmentRpcSubscriptionAtomFamily(runtime, { label: "environment-data:server:resource-telemetry", tag: WS_METHODS.subscribeResourceTelemetry, diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index e4823f30272b..9cd0da695fbd 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -339,7 +339,14 @@ import { ProviderConsumeResetCreditInput, ProviderConsumeResetCreditResult, } from "./providerUsageLimits.ts"; -import { UsagePricing, UsageReadError, UsageSummary, UsageSummaryInput } from "./usage.ts"; +import { + UsagePricing, + UsageReadError, + UsageSessionSpendInput, + UsageSessionSpendResult, + UsageSummary, + UsageSummaryInput, +} from "./usage.ts"; import { StorageCleanupReport, DatabaseStorageStatus, @@ -519,6 +526,7 @@ export const WS_METHODS = { serverSignalProcess: "server.signalProcess", serverReportClientActivity: "server.reportClientActivity", serverGetUsageSummary: "server.getUsageSummary", + serverGetSessionSpend: "server.getSessionSpend", serverRefreshUsageRates: "server.refreshUsageRates", scheduledTasksSubscribe: "scheduledTasks.subscribe", @@ -937,6 +945,12 @@ const WsServerGetUsageSummaryRpc = Rpc.make(WS_METHODS.serverGetUsageSummary, { error: Schema.Union([EnvironmentAuthorizationError, UsageReadError]), }); +const WsServerGetSessionSpendRpc = Rpc.make(WS_METHODS.serverGetSessionSpend, { + payload: UsageSessionSpendInput, + success: UsageSessionSpendResult, + error: Schema.Union([EnvironmentAuthorizationError, UsageReadError]), +}); + const WsServerRefreshUsageRatesRpc = Rpc.make(WS_METHODS.serverRefreshUsageRates, { payload: Schema.Struct({}), success: UsagePricing, @@ -1967,6 +1981,7 @@ export const WsRpcGroup = RpcGroup.make( WsServerCheckOtlpEndpointRpc, WsServerPrepareTriageRpc, WsServerGetUsageSummaryRpc, + WsServerGetSessionSpendRpc, WsServerRefreshUsageRatesRpc, WsServerSignalProcessRpc, WsScheduledTasksSubscribeRpc, diff --git a/packages/contracts/src/usage.ts b/packages/contracts/src/usage.ts index 8e34c10eb885..751b12aa75ef 100644 --- a/packages/contracts/src/usage.ts +++ b/packages/contracts/src/usage.ts @@ -1,6 +1,11 @@ import * as Schema from "effect/Schema"; -import { ForwardCompatibleArray, NonNegativeInt, TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { + ForwardCompatibleArray, + NonNegativeInt, + ThreadId, + TrimmedNonEmptyString, +} from "./baseSchemas.ts"; /** Bump for incompatible UsageSummary changes; skipped unknown array variants need no bump. */ export const USAGE_CONTRACT_VERSION = 6 as const; @@ -133,6 +138,64 @@ export const UsageSummary = Schema.Struct({ }); export type UsageSummary = typeof UsageSummary.Type; +/** A provider transcript session. This is not a T3 thread id. */ +export const UsageSessionRef = Schema.Struct({ + provider: UsageProviderKind, + sessionId: TrimmedNonEmptyString, +}); +export type UsageSessionRef = typeof UsageSessionRef.Type; + +export const UsageSessionSpend = Schema.Struct({ + provider: UsageProviderKind, + sessionId: TrimmedNonEmptyString, + totals: UsageTokenTotals, + costUsd: Schema.Number, + categoryCostUsd: Schema.optional(UsageCategoryCost), + costSource: UsageCostSource, + records: NonNegativeInt, + unpricedRecords: NonNegativeInt, +}); +export type UsageSessionSpend = typeof UsageSessionSpend.Type; + +/** Threads one spend request may name. The web prices the newest threads of a project under it. */ +export const MAX_SESSION_SPEND_THREADS = 150 as const; + +/** Same day window as a summary. `sessions` selects rows after the scan is deduped. */ +export const UsageSessionSpendInput = Schema.Struct({ + sinceDay: UsageDay, + untilDay: UsageDay, + timeZone: TrimmedNonEmptyString, + awaitRefresh: Schema.optional(Schema.Boolean), + sessions: Schema.Array(UsageSessionRef), + /** + * T3 threads this environment resolves to provider sessions, unioned with `sessions`. Callers + * that hold every thread of a scope do not have to open each thread to price it. + */ + threadIds: Schema.optional( + Schema.Array(ThreadId).check(Schema.isMaxLength(MAX_SESSION_SPEND_THREADS)), + ), +}); +export type UsageSessionSpendInput = typeof UsageSessionSpendInput.Type; + +export const UsageSessionSpendResult = Schema.Struct({ + readAt: Schema.String, + sinceDay: UsageDay, + untilDay: UsageDay, + sources: ForwardCompatibleArray(UsageSource), + pricing: UsagePricing, + sessions: ForwardCompatibleArray(UsageSessionSpend), + /** Sessions dropped because the request named more than the server prices at once. */ + truncatedSessions: Schema.optional(NonNegativeInt), + /** + * A usage source covering a priced session was partial or failed, so every total here is a + * lower bound. The server answers this because it resolves which sessions a request reaches. + */ + incomplete: Schema.optional(Schema.Boolean), + /** T3 threads in the request whose provider sessions could not be resolved. */ + unresolvedThreads: Schema.optional(NonNegativeInt), +}); +export type UsageSessionSpendResult = typeof UsageSessionSpendResult.Type; + export class UsageReadError extends Schema.TaggedError()("UsageReadError", { reason: Schema.Literals(["scanFailed", "invalidWindow"]), detail: TrimmedNonEmptyString, diff --git a/packages/shared/package.json b/packages/shared/package.json index cc113b00fe4d..533e532cf34d 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -363,6 +363,10 @@ "types": "./src/usageFormat.ts", "import": "./src/usageFormat.ts" }, + "./usageSessionRefs": { + "types": "./src/usageSessionRefs.ts", + "import": "./src/usageSessionRefs.ts" + }, "./usageLimits": { "types": "./src/usageLimits.ts", "import": "./src/usageLimits.ts" diff --git a/packages/shared/src/usageMerge.ts b/packages/shared/src/usageMerge.ts index 994ae01d6ffe..7633a58c64f8 100644 --- a/packages/shared/src/usageMerge.ts +++ b/packages/shared/src/usageMerge.ts @@ -255,16 +255,20 @@ function ownedContribution( }; } -function bucketTokens(bucket: UsageBucket): number { - // reasoningTokens is a subset of outputTokens and must not be added again. +/** Same total the Usage dashboard sums. Reasoning is already inside output. */ +export function usageTotalTokens(totals: UsageTokenTotals): number { return ( - bucket.totals.uncachedInputTokens + - bucket.totals.cachedInputTokens + - bucket.totals.cacheCreationTokens + - bucket.totals.outputTokens + totals.uncachedInputTokens + + totals.cachedInputTokens + + totals.cacheCreationTokens + + totals.outputTokens ); } +function bucketTokens(bucket: UsageBucket): number { + return usageTotalTokens(bucket.totals); +} + export function isCompatibleUsageContractVersion(version: number, expected: number): boolean { return version >= USAGE_MERGE_COMPATIBLE_SINCE && version <= expected; } diff --git a/packages/shared/src/usageSessionRefs.ts b/packages/shared/src/usageSessionRefs.ts new file mode 100644 index 000000000000..735a5cab220f --- /dev/null +++ b/packages/shared/src/usageSessionRefs.ts @@ -0,0 +1,46 @@ +import type { UsageProviderKind, UsageSessionRef } from "@t3tools/contracts"; + +/** + * Providers whose native thread id is also the usage session id. Cursor's conversation id is + * not the agent id, Antigravity's native id is unconfirmed, and Pi often stores a file path. + */ +const JOINABLE_DRIVERS: Readonly> = { + claudeAgent: "claude", + codex: "codex", + grok: "grok", + opencode: "opencode", + muse: "muse", + pi: "pi", +}; + +export function joinableUsageSessionRef( + driver: string, + nativeId: string | null | undefined, + strength: "strong" | "weak" | "none" | null = "strong", +): UsageSessionRef | null { + if (strength !== "strong") return null; + const provider = JOINABLE_DRIVERS[driver]; + if (provider === undefined || nativeId === null || nativeId === undefined || nativeId === "") { + return null; + } + // An ACP `:task:` key is not the billable session. + if (nativeId.includes(":task:")) return null; + if (provider === "pi" && (nativeId.includes("/") || nativeId.includes("\\"))) return null; + return { provider, sessionId: nativeId }; +} + +export function usageSessionKey(ref: Pick): string { + return `${ref.provider}\u0000${ref.sessionId}`; +} + +export function uniqueUsageSessionRefs(refs: readonly UsageSessionRef[]): UsageSessionRef[] { + const seen = new Set(); + const unique: UsageSessionRef[] = []; + for (const ref of refs) { + const key = usageSessionKey(ref); + if (seen.has(key)) continue; + seen.add(key); + unique.push(ref); + } + return unique; +}