diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index ae0126d438254..21eecb52a828b 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -520,6 +520,69 @@ describe("Codex usage history", () => { }); describe("UsageService", () => { + it.live("matches priced native session records to their T3 thread", () => + Effect.gen(function* () { + const { transcript, home, settings } = yield* setup; + yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 25))); + yield* Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`INSERT INTO orchestration_v2_projection_provider_threads ${sql.insert({ + provider_thread_id: "provider-thread-usage", + thread_id: "thread-usage", + provider: "claudeAgent", + driver: "claudeAgent", + status: "ready", + updated_at: "2026-08-01T10:00:00Z", + payload_json: encodeUnknownJsonString({ + nativeThreadRef: { driver: "claudeAgent", nativeId: "session-1", strength: "strong" }, + contextUsage: { usedTokens: 42000, maxTokens: 200000 }, + }), + })}`; + yield* sql`INSERT INTO orchestration_v2_projection_threads ${sql.insert({ + thread_id: "thread-usage", + project_id: "project-usage", + title: "Usage", + default_provider: "claudeAgent", + provider_instance_id: "claudeAgent", + runtime_mode: "full-access", + interaction_mode: "default", + active_provider_thread_id: "provider-thread-usage", + created_at: "2026-08-01T10:00:00Z", + updated_at: "2026-08-01T10:00:00Z", + payload_json: "{}", + })}`; + const service = yield* UsageService.make; + const summary = yield* service.readSummary({ ...WINDOW, includeThreadUsage: true }); + assert.deepEqual( + summary.threadUsage?.map((usage) => ({ + threadId: usage.threadId, + input: usage.totals.uncachedInputTokens, + output: usage.totals.outputTokens, + })), + [{ threadId: "thread-usage", input: 10, output: 25 }], + ); + assert.isAbove(summary.threadUsage?.[0]?.costUsd ?? 0, 0); + assert.deepEqual(summary.threadContexts, [ + { + threadId: "thread-usage", + usage: { usedTokens: 42000, maxTokens: 200000 }, + updatedAt: "2026-08-01T10:00:00Z", + }, + ]); + }).pipe( + Effect.provide( + layerService({ + prefix: "thread-usage", + home, + settings, + ratesDocument: { + "claude-fable-5": { input_cost_per_token: 0.000003, output_cost_per_token: 0.000015 }, + }, + }), + ), + ); + }).pipe(Effect.scoped), + ); it.live("counts compressed Codex rollouts across refresh, rewrites, and cache restarts", () => Effect.gen(function* () { const { home, settings } = yield* setup; diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 341b14d8c6280..8d50210b57bdb 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -9,6 +9,7 @@ import { type UsageSummary, type UsageSummaryInput, UsageReadError, + ThreadTokenUsageSnapshot, } from "@t3tools/contracts"; import * as HostProcess from "@t3tools/shared/HostProcess"; import * as Cause from "effect/Cause"; @@ -26,6 +27,8 @@ import * as Path from "effect/Path"; import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import { HttpClient, HttpClientResponse } from "effect/http"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite"; import { formatMalformedUsageWarning } from "@t3tools/shared/usageFormat"; @@ -105,6 +108,15 @@ const encodeRatesCache = Schema.encodeEffect(Schema.fromJsonString(RatesCacheFil const ScanCacheJson = Schema.fromJsonString(Schema.Unknown); const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson); const encodeUsageRecordKey = Schema.encodeSync(ScanCacheJson); +const decodeThreadContext = Schema.decodeUnknownOption( + Schema.fromJsonString( + Schema.Struct({ + ...ThreadTokenUsageSnapshot.fields, + updatedAt: Schema.optional(Schema.String), + maxTokens: Schema.optional(Schema.NullOr(Schema.Number)), + }), + ), +); const CachedSource = Schema.Struct({ dir: Schema.String, volumeId: Schema.String }); function isLaterRead(a: CachedFile, b: CachedFile): boolean { @@ -156,6 +168,7 @@ export const make = Effect.gen(function* () { const httpClient = yield* HttpClient.HttpClient; const hostEnvironment = yield* HostProcess.Environment; const hostname = yield* HostProcess.Hostname; + const database = yield* SqlClient.SqlClient; const readerContext = yield* Effect.context(); const fileCache: ScanCache = new Map(); @@ -763,7 +776,78 @@ export const make = Effect.gen(function* () { { concurrency: 2 }, ); + let threadIdsBySession: Map | undefined; + const threadContexts: NonNullable[number][] = []; + if (input.includeThreadUsage === true) { + // Context size for idle threads, from each thread's latest reported provider turn. + const contexts = yield* database<{ + readonly threadId: string; + readonly usage: string | null; + readonly updatedAt: string; + }>` + SELECT t.thread_id AS threadId, p.updated_at AS updatedAt, + COALESCE((SELECT json_extract(pt.payload_json, '$.tokenUsage') + FROM orchestration_v2_projection_provider_turns pt + WHERE pt.provider_thread_id = p.provider_thread_id + AND json_extract(pt.payload_json, '$.tokenUsage.usedTokens') IS NOT NULL + ORDER BY pt.ordinal DESC LIMIT 1), json_extract(p.payload_json, '$.contextUsage')) AS usage + FROM orchestration_v2_projection_threads t + JOIN orchestration_v2_projection_provider_threads p ON p.provider_thread_id = t.active_provider_thread_id + WHERE t.deleted_at IS NULL + `.pipe( + NodeSqliteClient.readOnly, + Effect.mapError( + (cause) => + new UsageReadError({ + reason: "scanFailed", + detail: "Could not read session context usage.", + cause, + }), + ), + ); + for (const row of contexts) { + const decoded = row.usage === null ? Option.none() : decodeThreadContext(row.usage); + if (Option.isNone(decoded)) continue; + const { maxTokens, updatedAt, ...usage } = decoded.value; + threadContexts.push({ + threadId: row.threadId, + usage: { ...usage, ...(maxTokens != null && maxTokens > 0 ? { maxTokens } : {}) }, + updatedAt: updatedAt ?? row.updatedAt, + }); + } + const links = yield* database<{ + readonly threadId: string; + readonly driver: string; + readonly sessionId: string; + }>` + SELECT DISTINCT thread_id AS threadId, driver, + json_extract(payload_json, '$.nativeThreadRef.nativeId') AS sessionId + FROM orchestration_v2_projection_provider_threads + WHERE json_extract(payload_json, '$.nativeThreadRef.nativeId') IS NOT NULL + `.pipe( + NodeSqliteClient.readOnly, + Effect.mapError( + (cause) => + new UsageReadError({ + reason: "scanFailed", + detail: "Could not match native usage to sessions.", + cause, + }), + ), + ); + threadIdsBySession = new Map(); + for (const link of links) { + const provider = BUILT_IN_USAGE_DRIVERS.find((driver) => driver.driverKind === link.driver) + ?.usage?.provider; + if (!provider) continue; + const key = `${provider}:${link.sessionId}`; + const previous = threadIdsBySession.get(key) ?? []; + if (!previous.includes(link.threadId)) + threadIdsBySession.set(key, [...previous, link.threadId]); + } + } const aggregator = new UsageAggregator({ + ...(threadIdsBySession === undefined ? {} : { threadIdsBySession }), timeZone: input.timeZone, sinceDay: input.sinceDay, untilDay: input.untilDay, @@ -927,6 +1011,8 @@ export const make = Effect.gen(function* () { sinceDay: input.sinceDay, untilDay: input.untilDay, buckets: aggregated.buckets, + ...(aggregated.threadUsage === undefined ? {} : { threadUsage: aggregated.threadUsage }), + ...(input.includeThreadUsage ? { threadContexts } : {}), sources, pricing: pricing(), scanDurationMs: Math.max(0, finishedAtMs - startedAtMs), @@ -947,6 +1033,7 @@ export const make = Effect.gen(function* () { settings, input.awaitRefresh === true, input.modelFilter ?? null, + input.includeThreadUsage ?? false, ]); const readSummary = Effect.fn("UsageService.readSummary")(function* ( diff --git a/apps/server/src/usage/usageAggregation.test.ts b/apps/server/src/usage/usageAggregation.test.ts index 625f7a9d63eca..cf3eb1f4454fc 100644 --- a/apps/server/src/usage/usageAggregation.test.ts +++ b/apps/server/src/usage/usageAggregation.test.ts @@ -69,6 +69,58 @@ function aggregate( } describe("UsageAggregator", () => { + it("prices thread usage separately across models and deduplicates native session links", () => { + const aggregator = new UsageAggregator({ + timeZone: "UTC", + sinceDay: "2026-08-01", + untilDay: "2026-08-31", + rates, + threadIdsBySession: new Map([ + ["claude:session-a", ["thread-a"]], + ["claude:session-b", ["thread-b"]], + ]), + }); + aggregator.add(record({ dedupeKey: "a" })); + aggregator.add(record({ dedupeKey: "a" })); + aggregator.add(record({ sessionId: "session-b", model: "unknown" })); + const usage = aggregator.finish().threadUsage; + expect(usage?.find((entry) => entry.threadId === "thread-a")).toMatchObject({ + totals: { + uncachedInputTokens: 100, + cachedInputTokens: 1000, + cacheCreationTokens: 10, + outputTokens: 50, + }, + unpricedRecords: 0, + }); + expect(usage?.find((entry) => entry.threadId === "thread-a")?.costUsd).toBeCloseTo(0.004625, 8); + expect(usage?.find((entry) => entry.threadId === "thread-b")).toMatchObject({ + unpricedRecords: 1, + }); + }); + it("attributes a copied inference when its linked session is scanned second", () => { + const aggregator = new UsageAggregator({ + timeZone: "UTC", + sinceDay: "2026-08-01", + untilDay: "2026-08-31", + rates, + threadIdsBySession: new Map([["claude:session-linked", ["thread-a"]]]), + }); + aggregator.add(record({ sessionId: "session-unlinked", dedupeKey: "copied-inference" })); + aggregator.add(record({ sessionId: "session-linked", dedupeKey: "copied-inference" })); + aggregator.add(record({ sessionId: "session-linked", dedupeKey: "copied-inference" })); + const result = aggregator.finish(); + + expect(result.buckets).toHaveLength(1); + expect(result.buckets[0]?.records).toBe(1); + expect(result.duplicatesDropped).toBe(2); + expect(result.threadUsage).toEqual([ + expect.objectContaining({ + threadId: "thread-a", + totals: expect.objectContaining({ outputTokens: 50 }), + }), + ]); + }); it("splits a bucket's cost by category and speed", () => { const [bucket] = aggregate([record(), record({ speed: "fast" })]).buckets; diff --git a/apps/server/src/usage/usageAggregation.ts b/apps/server/src/usage/usageAggregation.ts index 8d1d799859737..68bdbe3ec95fb 100644 --- a/apps/server/src/usage/usageAggregation.ts +++ b/apps/server/src/usage/usageAggregation.ts @@ -7,6 +7,7 @@ import type { UsageModelSelector, UsageResolution, UsageTokenTotals, + UsageThreadTotals, } from "@t3tools/contracts"; import { EMPTY_TOTALS, type UsageRecord } from "@t3tools/provider-core/server/usage"; @@ -64,6 +65,7 @@ interface MutableBucket { } export interface AggregateOptions { + readonly threadIdsBySession?: ReadonlyMap; readonly timeZone: string; readonly sinceDay: string; readonly untilDay: string; @@ -125,6 +127,7 @@ export function canonicalModelFilter( } export interface AggregateResult { + readonly threadUsage?: readonly UsageThreadTotals[]; readonly buckets: readonly UsageBucket[]; readonly duplicatesDropped: number; readonly outOfWindow: number; @@ -172,8 +175,10 @@ interface Contribution { /** Deduplicates across the whole scan; Claude copies records into resumed/forked transcripts. */ export class UsageAggregator { + readonly #threadUsage = new Map(); readonly #buckets = new Map(); readonly #seen = new Map(); + readonly #threadSeen = new Map>(); readonly #toDay: (timestampMs: number) => string; readonly #hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null; readonly #options: AggregateOptions; @@ -207,6 +212,8 @@ export class UsageAggregator { add(input: UsageRecord, sourcePath?: string): boolean { if (!keepsUsageRecord(this.#options.modelFilter, input)) return false; const record = this.#mapModel(input); + // Counted before the scan-wide dedupe: a copy in another thread's transcript still belongs to it. + this.#addThreadUsage(record); if (record.dedupeKey !== null) { const held = this.#seen.get(record.dedupeKey); if (held !== undefined) { @@ -228,7 +235,46 @@ export class UsageAggregator { return true; } - #placeInWindow(record: UsageRecord, source: string): MutableBucket | null { + #addThreadUsage(record: UsageRecord): void { + const threadIds = + this.#options.threadIdsBySession?.get(`${record.provider}:${record.sessionId}`) ?? []; + if (threadIds.length === 0 || this.#dayInWindow(record) === null) return; + const priced = priceUsage(this.#options.rates, record, this.#options.priceOverrides); + const costs = priced.categoryCostUsd; + for (const threadId of threadIds) { + if (record.dedupeKey !== null) { + const seen = this.#threadSeen.get(threadId); + if (seen?.has(record.dedupeKey)) continue; + if (seen) seen.add(record.dedupeKey); + else this.#threadSeen.set(threadId, new Set([record.dedupeKey])); + } + const previous = this.#threadUsage.get(threadId); + const add = (key: K) => + (previous?.totals[key] ?? 0) + record.totals[key]; + this.#threadUsage.set(threadId, { + threadId, + totals: { + uncachedInputTokens: add("uncachedInputTokens"), + cachedInputTokens: add("cachedInputTokens"), + cacheCreationTokens: add("cacheCreationTokens"), + outputTokens: add("outputTokens"), + reasoningTokens: add("reasoningTokens"), + }, + costUsd: (previous?.costUsd ?? 0) + priced.costUsd, + categoryCostUsd: { + input: (previous?.categoryCostUsd.input ?? 0) + (costs?.input ?? 0), + cacheRead: (previous?.categoryCostUsd.cacheRead ?? 0) + (costs?.cacheRead ?? 0), + cacheWrite: (previous?.categoryCostUsd.cacheWrite ?? 0) + (costs?.cacheWrite ?? 0), + output: (previous?.categoryCostUsd.output ?? 0) + (costs?.output ?? 0), + }, + unsplitCostUsd: (previous?.unsplitCostUsd ?? 0) + (costs === null ? priced.costUsd : 0), + unpricedRecords: + (previous?.unpricedRecords ?? 0) + (priced.costSource === "unpriced" ? 1 : 0), + }); + } + } + + #dayInWindow(record: UsageRecord): string | null { if ( this.#hourlyWindow !== null && (record.timestampMs < this.#hourlyWindow.sinceTimeMs || @@ -244,6 +290,12 @@ export class UsageAggregator { ) { return null; } + return day; + } + + #placeInWindow(record: UsageRecord, source: string): MutableBucket | null { + const day = this.#dayInWindow(record); + if (day === null) return null; const hourIndex = this.#hourlyWindow === null @@ -400,6 +452,9 @@ export class UsageAggregator { return { buckets, + ...(this.#options.threadIdsBySession === undefined + ? {} + : { threadUsage: [...this.#threadUsage.values()] }), duplicatesDropped: this.#duplicatesDropped, outOfWindow: this.#outOfWindow, }; diff --git a/apps/web/src/components/chat/ContextWindowMeter.tsx b/apps/web/src/components/chat/ContextWindowMeter.tsx new file mode 100644 index 0000000000000..27a8ca692e22f --- /dev/null +++ b/apps/web/src/components/chat/ContextWindowMeter.tsx @@ -0,0 +1,217 @@ +import { Button } from "../ui/button"; +import type { ContextWindowSnapshot } from "@t3tools/client-runtime/state/context-window"; +import { formatContextWindowTokens } from "~/lib/contextWindow"; +import { Popover, PopoverPopup, PopoverTrigger } from "../ui/popover"; +import { formatContextWindowCompactionMessage } from "./compaction.logic"; +import { Minimize2Icon } from "lucide-react"; +import { composerFloatingLayerProps } from "./composerEventScope"; + +function formatCost(cost: { readonly amount: number; readonly currency: string }): string { + const fractionDigits = Math.abs(cost.amount) > 0 && Math.abs(cost.amount) < 0.01 ? 4 : 2; + return `${cost.currency} ${cost.amount.toFixed(fractionDigits)}`; +} + +function formatPercentage(value: number | null): string | null { + if (value === null || !Number.isFinite(value)) { + return null; + } + if (value < 10) { + return `${value.toFixed(1).replace(/\.0$/, "")}%`; + } + return `${Math.round(value)}%`; +} + +export function ContextWindowMeter(props: { + usage: ContextWindowSnapshot | null; + onCompact?: (() => void) | undefined; + compactDisabled?: boolean | undefined; + compactDisabledReason?: string | null | undefined; + compact?: boolean; +}) { + const { onCompact, compactDisabled, compactDisabledReason } = props; + const missing = props.usage === null; + const usage = props.usage ?? { + usedPercentage: null, + maxTokens: null, + usedTokens: 0, + totalProcessedTokens: null, + cost: null, + compactsAutomatically: false, + autoCompactThreshold: null, + updatedAt: null, + }; + const usedPercentage = formatPercentage(usage.usedPercentage); + const normalizedPercentage = Math.max(0, Math.min(100, usage.usedPercentage ?? 0)); + const radius = 9.75; + const circumference = 2 * Math.PI * radius; + const dashOffset = circumference * (1 - normalizedPercentage / 100); + const totalProcessedTokens = usage.totalProcessedTokens ?? null; + const showTotalProcessed = totalProcessedTokens !== null && totalProcessedTokens > 0; + const isOverloaded = normalizedPercentage > 90; + const usageColor = missing + ? "color-mix(in oklab, var(--color-muted-foreground) 30%, transparent)" + : isOverloaded + ? "var(--color-error)" + : "color-mix(in oklab, var(--color-muted-foreground) 72%, transparent)"; + + return ( + + event.stopPropagation()} + aria-label={ + missing + ? "Context usage unavailable" + : usage.maxTokens !== null && usedPercentage + ? `Context window ${usedPercentage} used` + : `Context window ${formatContextWindowTokens(usage.usedTokens)} tokens used` + } + > + + + + + } + /> + +
+ {missing ? null : ( +
+ {usage.maxTokens !== null && usedPercentage ? ( +
+ {usedPercentage} + · + + {formatContextWindowTokens(usage.usedTokens)}/ + {formatContextWindowTokens(usage.maxTokens ?? null)} + +
+ ) : ( +
+ {formatContextWindowTokens(usage.usedTokens)} +
+ )} +
+ )} + {usage.maxTokens !== null ? ( +
+
+
+ ) : null} + {showTotalProcessed ? ( +
+ Total processed + + {formatContextWindowTokens(totalProcessedTokens)} + +
+ ) : null} + {usage.maxTokens != null ? ( +
+ Remaining + + {formatContextWindowTokens(Math.max(0, usage.maxTokens - usage.usedTokens))} + +
+ ) : null} + {usage.updatedAt ? ( +
+ Updated {new Date(usage.updatedAt).toLocaleString()} +
+ ) : null} + {usage.cost != null ? ( +
+ Cost + + {formatCost(usage.cost)} + +
+ ) : null} + {usage.compactsAutomatically ? ( +
+ {formatContextWindowCompactionMessage(usage.autoCompactThreshold)} +
+ ) : null} + {onCompact ? ( + <> + + {compactDisabled && compactDisabledReason ? ( +
+ {compactDisabledReason} +
+ ) : null} + + ) : null} +
+ + + ); +} diff --git a/apps/web/src/components/sidebar/SessionContextMeter.tsx b/apps/web/src/components/sidebar/SessionContextMeter.tsx new file mode 100644 index 0000000000000..645b0f2b63dff --- /dev/null +++ b/apps/web/src/components/sidebar/SessionContextMeter.tsx @@ -0,0 +1,106 @@ +import { useState } from "react"; +import * as DateTime from "effect/DateTime"; +import * as Option from "effect/Option"; +import { + AuthDiagnosticsReadScope, + AuthOrchestrationOperateScope, + type ProviderDriverKind, +} from "@t3tools/contracts"; +import { useEnvironmentQuery } from "../../state/query"; +import { sessionUsageAtom } from "../../state/sessionUsage"; +import { threadEnvironment } from "../../state/threads"; +import { useEnvironmentScope } from "../../state/session"; +import { useOrchestrationCommand } from "../../state/use-orchestration-command"; +import { + deriveLatestContextWindowSnapshot, + latestProviderTurnTokenUsage, +} from "@t3tools/client-runtime/state/context-window"; +import { useQuietThreadProjection } from "../../state/entities"; +import { useRetainedValue } from "../Sidebar.logic"; +import { newMessageId } from "../../lib/utils"; +import { ContextWindowMeter } from "../chat/ContextWindowMeter"; +import type { SidebarThreadSummary } from "../../types"; + +export function SessionContextMeter({ + thread, + driverKind, +}: { + thread: SidebarThreadSummary; + driverKind: ProviderDriverKind | null; +}) { + const canRead = useEnvironmentScope(thread.environmentId, AuthDiagnosticsReadScope); + const canOperate = useEnvironmentScope(thread.environmentId, AuthOrchestrationOperateScope); + const query = useEnvironmentQuery(canRead ? sessionUsageAtom(thread.environmentId) : null); + const context = query.data?.threadContexts?.find((entry) => entry.threadId === thread.id); + const [pending, setPending] = useState(false); + const send = useOrchestrationCommand(threadEnvironment.startTurn, "Failed to compact context"); + const busy = + thread.runtime?.activeRunId != null || thread.hasPendingApprovals || thread.hasPendingUserInput; + const supported = driverKind === "claudeAgent" || driverKind === "codex"; + const compact = async () => { + if (!canOperate || busy || pending || !supported) return; + setPending(true); + await send({ + environmentId: thread.environmentId, + input: { + threadId: thread.id, + message: { messageId: newMessageId(), role: "user", text: "/compact", attachments: [] }, + modelSelection: thread.modelSelection, + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + dispatchMode: "start", + }, + }).finally(() => setPending(false)); + }; + const projection = useQuietThreadProjection( + thread.runtime?.activeRunId != null + ? { environmentId: thread.environmentId, threadId: thread.id } + : null, + )?.projection; + const providerThread = projection?.providerThreads.find( + (entry) => entry.id === projection.thread.activeProviderThreadId, + ); + const contextKey = `${thread.environmentId}:${thread.id}:${thread.activeProviderThreadId}`; + const nativeUsage = projection + ? latestProviderTurnTokenUsage( + projection.providerTurns.filter((entry) => entry.providerThreadId === providerThread?.id), + projection, + providerThread, + ) + : null; + const retainedNativeUsage = useRetainedValue(contextKey, nativeUsage); + const retainedProvider = useRetainedValue(contextKey, providerThread ?? null); + const retainedLive = deriveLatestContextWindowSnapshot([], retainedNativeUsage, retainedProvider); + const reportedAt = context ? DateTime.make(context.updatedAt) : Option.none(); + const reported = + context && Option.isSome(reportedAt) + ? deriveLatestContextWindowSnapshot([], null, { + contextUsage: context.usage, + updatedAt: reportedAt.value, + }) + : null; + const usage = + retainedLive && + (!reported || Date.parse(retainedLive.updatedAt) > Date.parse(reported.updatedAt)) + ? retainedLive + : reported; + return ( + { + void compact(); + }} + compactDisabled={!canOperate || busy || pending || !supported} + compactDisabledReason={ + !canOperate + ? "Read-only connection" + : !supported + ? "Model cannot compact" + : pending + ? "Compacting…" + : "Finish current turn first" + } + /> + ); +} diff --git a/apps/web/src/components/sidebar/SessionUsageBadge.tsx b/apps/web/src/components/sidebar/SessionUsageBadge.tsx new file mode 100644 index 0000000000000..2f9564b0009db --- /dev/null +++ b/apps/web/src/components/sidebar/SessionUsageBadge.tsx @@ -0,0 +1,139 @@ +import { type EnvironmentId, type ThreadId } from "@t3tools/contracts"; +import { AuthDiagnosticsReadScope } from "@t3tools/contracts"; +import { useRef, useState } from "react"; +import { useEnvironmentQuery } from "../../state/query"; +import { sessionUsageAtom } from "../../state/sessionUsage"; +import { useEnvironmentScope } from "../../state/session"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; +import { formatSessionCost, sessionTokenCount, sessionSpendRate } from "../../lib/sessionUsage"; + +const money = new Intl.NumberFormat("en-US", { + style: "currency", + currency: "USD", + minimumFractionDigits: 2, + maximumFractionDigits: 4, +}); + +export function SessionUsageBadge({ + environmentId, + threadId, + working, +}: { + environmentId: EnvironmentId; + threadId: ThreadId; + working: boolean; +}) { + const rowAnchor = useRef(null); + const canRead = useEnvironmentScope(environmentId, AuthDiagnosticsReadScope); + const query = useEnvironmentQuery(canRead ? sessionUsageAtom(environmentId) : null); + const usage = query.data?.threadUsage?.find((entry) => entry.threadId === threadId); + const sampleKey = `${environmentId}:${threadId}:${working}`; + const [sample, setSample] = useState({ key: sampleKey, at: 0, cost: 0, rate: 0 }); + if (sample.key !== sampleKey || (usage && query.dataUpdatedAt !== sample.at)) { + setSample({ + key: sampleKey, + at: query.dataUpdatedAt, + cost: usage?.costUsd ?? 0, + rate: + working && sample.key === sampleKey && usage + ? sessionSpendRate(sample.cost, sample.at, usage.costUsd, query.dataUpdatedAt) + : 0, + }); + } + const burningFast = working && sample.key === sampleKey && sample.rate >= 1; + if (!canRead) return null; + const tokens = usage ? sessionTokenCount(usage.totals) : null; + const price = usage ? formatSessionCost(usage.costUsd, usage.unpricedRecords > 0) : "$?"; + const rows = usage + ? ([ + ["Tokens in", usage.totals.uncachedInputTokens, usage.categoryCostUsd.input], + ["Tokens out", usage.totals.outputTokens, usage.categoryCostUsd.output], + ["Cached tokens in", usage.totals.cachedInputTokens, usage.categoryCostUsd.cacheRead], + ["Cache writes", usage.totals.cacheCreationTokens, usage.categoryCostUsd.cacheWrite], + ] as const) + : []; + return ( + + { + rowAnchor.current = element?.closest("[data-sidebar-row]") ?? element; + }} + className="pointer-events-auto inline-flex shrink-0 items-center gap-1 tabular-nums" + tabIndex={0} + aria-label={ + tokens === null + ? "Session usage unavailable" + : `${tokens.toLocaleString("en-US")} tokens, estimated cost ${price}` + } + /> + } + > + {tokens === null ? null : ( + + {new Intl.NumberFormat("en-US", { + notation: "compact", + maximumFractionDigits: 1, + }).format(tokens)} + + )} + {price} + {burningFast ? ( + + ! + + ) : null} + + + {usage ? ( +
+ Session usage · API-equivalent USD estimate + {burningFast ? ( + + High spending: ${sample.rate.toFixed(2)}/min (warning at $1/min). + + ) : null} +
+ Category + Tokens + Cost + {rows.map(([label, count, cost]) => ( +
+ {label} + {count.toLocaleString("en-US")} + + {usage.unpricedRecords > 0 ? "Unavailable" : money.format(cost)} + +
+ ))} + Cached tokens out + Not reported + N/A + {usage.unsplitCostUsd > 0 ? ( + <> + Provider-reported cost without breakdown + {money.format(usage.unsplitCostUsd)} + + ) : null} + Total + + {usage.unpricedRecords > 0 + ? `At least ${money.format(usage.costUsd)}` + : money.format(usage.costUsd)} + +
+ + Cache writes save input for reuse. Refreshes every minute while working. + + {usage.unpricedRecords > 0 ? Some records have no known price. : null} +
+ ) : query.isPending ? ( + "Loading session usage…" + ) : ( + "Session usage is unavailable. No matching provider transcript, or this server needs updating." + )} + + + ); +} diff --git a/apps/web/src/components/sidebar/SidebarThreadRow.tsx b/apps/web/src/components/sidebar/SidebarThreadRow.tsx index 89ab26f51b9a0..83e1ad18a62c5 100644 --- a/apps/web/src/components/sidebar/SidebarThreadRow.tsx +++ b/apps/web/src/components/sidebar/SidebarThreadRow.tsx @@ -1,6 +1,8 @@ import { useAtomMount } from "@effect/atom-react"; import { asLaunchingThread } from "@t3tools/client-runtime/state/thread-inbox"; import { ThreadHoverCard, ThreadHoverCardPopup } from "../ThreadHoverCard"; +import { SessionUsageBadge } from "./SessionUsageBadge"; +import { SessionContextMeter } from "./SessionContextMeter"; import { ThreadTerminalStatusButton } from "../ThreadStatusIndicators"; import { discardComposerDraft } from "../../lib/discardComposerDraft"; import { requestCustomSnooze } from "../CustomSnoozeDialog"; @@ -1156,6 +1158,16 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP ) : null; + const usageBadge = ( + <> + + + + ); const diff = latestRunDiff(thread); // Only the trailing icons would remain, so the row moves them to the title line instead. const metaLineEmpty = @@ -1193,25 +1205,28 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP −{formatDiffCount(diff.deletions)} ) : null} - + + {usageBadge} + + {remoteMachineIcon} + + + +
+ ); + const titleLineIcons = metaLineEmpty ? ( + + {usageBadge} + {remoteMachineIcon} - - ); - const titleLineIcons = metaLineEmpty ? ( - - {remoteMachineIcon} - ) : null; @@ -1224,7 +1239,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP {...sortableRootProps} {...(fileDropHandlers ?? {})} className={cn( - "list-none [content-visibility:auto] [contain-intrinsic-size:auto_--spacing(9)]", + "@container/session list-none [content-visibility:auto] [contain-intrinsic-size:auto_--spacing(9)]", metaLine !== null && multiline && "[contain-intrinsic-size:auto_--spacing(13.5)]", sortable?.isDragging && "relative z-20", )} @@ -1420,7 +1435,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP {...sortableRootProps} {...(fileDropHandlers ?? {})} className={cn( - "list-none py-0.5 [content-visibility:auto] [contain-intrinsic-size:auto_78px]", + "@container/session list-none py-0.5 [content-visibility:auto] [contain-intrinsic-size:auto_78px]", metaLineEmpty && "[contain-intrinsic-size:auto_58px]", sortable?.isDragging && "relative z-20", )} diff --git a/apps/web/src/lib/sessionUsage.test.ts b/apps/web/src/lib/sessionUsage.test.ts new file mode 100644 index 0000000000000..67269617a7daa --- /dev/null +++ b/apps/web/src/lib/sessionUsage.test.ts @@ -0,0 +1,30 @@ +import { expect, it } from "vite-plus/test"; +import { formatSessionCost, sessionTokenCount, sessionSpendRate } from "./sessionUsage"; + +it("measures recent spend without treating history, gaps or corrections as a burst", () => { + expect(sessionSpendRate(10, 60_000, 12, 120_000)).toBe(2); + expect(sessionSpendRate(10, 60_000, 10, 120_000)).toBe(0); + expect(sessionSpendRate(10, 60_000, 9, 120_000)).toBe(0); + expect(sessionSpendRate(0, 0, 100, 120_000)).toBe(0); + expect(sessionSpendRate(10, 60_000, 100, 60_001)).toBe(0); + expect(sessionSpendRate(10, 60_000, 100, 300_000)).toBe(0); +}); + +it("counts each billed token once, without adding reasoning twice", () => { + expect( + sessionTokenCount({ + uncachedInputTokens: 100, + cachedInputTokens: 200, + cacheCreationTokens: 30, + outputTokens: 50, + reasoningTokens: 20, + }), + ).toBe(380); +}); + +it("rounds the session estimate to whole dollars and marks partial pricing", () => { + expect(formatSessionCost(12.49)).toBe("$12"); + expect(formatSessionCost(12.5)).toBe("$13"); + expect(formatSessionCost(0.12)).toBe("$0"); + expect(formatSessionCost(1234.8, true)).toBe("≥$1,235"); +}); diff --git a/apps/web/src/lib/sessionUsage.ts b/apps/web/src/lib/sessionUsage.ts new file mode 100644 index 0000000000000..c6124f05e881c --- /dev/null +++ b/apps/web/src/lib/sessionUsage.ts @@ -0,0 +1,33 @@ +import { UsageDay, type UsageTokenTotals } from "@t3tools/contracts"; + +export const SESSION_USAGE_WINDOW = { + sinceDay: UsageDay.make("2000-01-01"), + untilDay: UsageDay.make("9999-12-31"), + timeZone: "UTC", + includeThreadUsage: true, +}; + +export function sessionSpendRate( + previousCost: number, + previousAt: number, + cost: number, + at: number, +): number { + const elapsed = at - previousAt; + // Ignore initial history, reconnect gaps and tiny intervals that magnify rounding. + if (previousAt === 0 || elapsed < 30_000 || elapsed > 120_000) return 0; + return (Math.max(0, cost - previousCost) * 60_000) / elapsed; +} + +export function sessionTokenCount(tokens: UsageTokenTotals): number { + return ( + tokens.uncachedInputTokens + + tokens.cachedInputTokens + + tokens.cacheCreationTokens + + tokens.outputTokens + ); +} + +export function formatSessionCost(costUsd: number, partial = false): string { + return `${partial ? "≥" : ""}$${Math.round(costUsd).toLocaleString("en-US")}`; +} diff --git a/apps/web/src/state/sessionUsage.ts b/apps/web/src/state/sessionUsage.ts new file mode 100644 index 0000000000000..f63b26e41d6f1 --- /dev/null +++ b/apps/web/src/state/sessionUsage.ts @@ -0,0 +1,20 @@ +import { Atom } from "effect/reactivity"; +import type { EnvironmentId } from "@t3tools/contracts"; +import { serverEnvironment } from "./server"; +import { environmentThreadShells } from "./threads"; +import { SESSION_USAGE_WINDOW } from "../lib/sessionUsage"; +import { workingUsageQuery } from "./workingUsageQuery"; + +// One timer per environment, shared by every usage badge and context meter. +export const sessionUsageAtom = Atom.family((environmentId: EnvironmentId) => { + const working = Atom.make((get) => + get(environmentThreadShells.environmentThreadsAtom(environmentId)).some( + (thread) => thread.status === "running", + ), + ); + const query = serverEnvironment.threadUsageSummary({ + environmentId, + input: SESSION_USAGE_WINDOW, + }); + return workingUsageQuery(query, working); +}); diff --git a/apps/web/src/state/workingUsageQuery.test.ts b/apps/web/src/state/workingUsageQuery.test.ts new file mode 100644 index 0000000000000..5843df118099a --- /dev/null +++ b/apps/web/src/state/workingUsageQuery.test.ts @@ -0,0 +1,35 @@ +import { Atom, AtomRegistry } from "effect/reactivity"; +import { expect, it, vi } from "vite-plus/test"; +import { workingUsageQuery } from "./workingUsageQuery"; + +it("shares polling while working, captures completion once, and stops while idle", () => { + using stack = new DisposableStack(); + vi.useFakeTimers(); + stack.defer(() => vi.useRealTimers()); + const registry = AtomRegistry.make(); + stack.defer(() => registry.dispose()); + let fetches = 0; + const query = Atom.make(() => ++fetches); + const working = Atom.make(false); + const usage = workingUsageQuery(query, working); + const release = registry.mount(usage); + const releaseSecond = registry.mount(usage); + expect(registry.get(usage)).toBe(1); + vi.advanceTimersByTime(180_000); + expect(fetches).toBe(1); + registry.set(working, true); + expect(registry.get(usage)).toBe(2); + vi.advanceTimersByTime(60_000); + expect(registry.get(usage)).toBe(3); + registry.set(working, false); + expect(registry.get(usage)).toBe(4); + vi.advanceTimersByTime(180_000); + expect(fetches).toBe(4); + registry.set(working, true); + expect(registry.get(usage)).toBe(5); + releaseSecond(); + release(); + registry.dispose(); + vi.advanceTimersByTime(180_000); + expect(fetches).toBe(5); +}); diff --git a/apps/web/src/state/workingUsageQuery.ts b/apps/web/src/state/workingUsageQuery.ts new file mode 100644 index 0000000000000..5a1efb8c4fc19 --- /dev/null +++ b/apps/web/src/state/workingUsageQuery.ts @@ -0,0 +1,23 @@ +import { Atom } from "effect/reactivity"; + +export function workingUsageQuery( + query: Atom.Atom, + working: Atom.Atom, +): Atom.Atom { + return Atom.make((get) => { + let timer: ReturnType | undefined; + const update = (running: boolean) => { + clearInterval(timer); + timer = running ? setInterval(() => get.refresh(query), 60_000) : undefined; + }; + update(get.once(working)); + get.subscribe(working, (running) => { + update(running); + // Capture the beginning and final totals, without polling stopped sessions. + get.refresh(query); + }); + get.subscribe(query, (value) => get.setSelf(value)); + get.addFinalizer(() => clearInterval(timer)); + return get.once(query); + }); +} diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 5a42d07c9b34f..3bc918a05845b 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -1498,6 +1498,30 @@ export function createServerEnvironmentAtoms( }), // A cold transcript scan is measured in seconds, so keep the result around // long enough that switching windows or re-rendering does not rescan. + threadUsageSummary: createEnvironmentRpcQueryAtomFamily(runtime, { + label: "environment-data:server:thread-usage", + tag: WS_METHODS.serverGetUsageSummary, + staleTimeMs: 60_000, + refreshTrigger: ({ environmentId }) => usageScanSettingsAtom(environmentId), + execute: (input, emit) => + request(WS_METHODS.serverGetUsageSummary, input).pipe( + Effect.filterOrElse( + (summary) => !summary.sources.some((source) => source.refreshing === true), + (summary) => + emit(summary).pipe( + Effect.andThen( + request(WS_METHODS.serverGetUsageSummary, { ...input, awaitRefresh: true }), + ), + Effect.catch((error) => + Effect.logWarning("Could not refresh slow thread usage sources.").pipe( + Effect.annotateLogs({ ...safeErrorLogAttributes(error) }), + Effect.as(summary), + ), + ), + ), + ), + ), + }), usageSummary: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:server:usage-summary", tag: WS_METHODS.serverGetUsageSummary, diff --git a/packages/client-runtime/src/state/serverUsage.test.ts b/packages/client-runtime/src/state/serverUsage.test.ts index b16550c7c487c..fe03895582629 100644 --- a/packages/client-runtime/src/state/serverUsage.test.ts +++ b/packages/client-runtime/src/state/serverUsage.test.ts @@ -205,6 +205,8 @@ const makeHarness = Effect.fn("ServerUsageTest.makeHarness")(function* ( inputs: () => inputs, updateSettings, summary: (input = INPUT) => atoms.usageSummary({ environmentId: TARGET.environmentId, input }), + threadSummary: (input = INPUT) => + atoms.threadUsageSummary({ environmentId: TARGET.environmentId, input }), }; }); @@ -300,3 +302,18 @@ it.effect("keeps the cached usage summary when the awaited refresh fails", () => }), ), ); + +it.effect("settles thread usage only after a refreshing source completes", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeRefreshingHarness(Effect.void); + const summary = harness.threadSummary(); + const unmount = harness.registry.mount(summary); + expect((yield* settledSummary(harness.registry, summary)).sources).toEqual([ + cursorSource(false), + ]); + expect(harness.inputs()).toEqual([INPUT, { ...INPUT, awaitRefresh: true }]); + unmount(); + }), + ), +); diff --git a/packages/contracts/src/usage.ts b/packages/contracts/src/usage.ts index 3663687811729..03778914948e5 100644 --- a/packages/contracts/src/usage.ts +++ b/packages/contracts/src/usage.ts @@ -1,4 +1,5 @@ import * as Schema from "effect/Schema"; +import { ThreadTokenUsageSnapshot } from "./providerRuntime.ts"; import { ForwardCompatibleArray, NonNegativeInt, TrimmedNonEmptyString } from "./baseSchemas.ts"; @@ -50,6 +51,16 @@ export const UsageCategoryCost = Schema.Struct({ }); export type UsageCategoryCost = typeof UsageCategoryCost.Type; +export const UsageThreadTotals = Schema.Struct({ + threadId: TrimmedNonEmptyString, + totals: UsageTokenTotals, + costUsd: Schema.Number, + categoryCostUsd: UsageCategoryCost, + unsplitCostUsd: Schema.Number, + unpricedRecords: NonNegativeInt, +}); +export type UsageThreadTotals = typeof UsageThreadTotals.Type; + export const UsageBucket = Schema.Struct({ day: UsageDay, hourStart: Schema.optional(TrimmedNonEmptyString), @@ -127,6 +138,7 @@ export const UsageModelFilter = Schema.Struct({ export type UsageModelFilter = typeof UsageModelFilter.Type; export const UsageSummaryInput = Schema.Struct({ + includeThreadUsage: Schema.optional(Schema.Boolean), sinceDay: UsageDay, untilDay: UsageDay, /** IANA bucketing zone; offsets fail across DST boundaries. */ @@ -140,6 +152,16 @@ export const UsageSummaryInput = Schema.Struct({ export type UsageSummaryInput = typeof UsageSummaryInput.Type; export const UsageSummary = Schema.Struct({ + threadContexts: Schema.optional( + Schema.Array( + Schema.Struct({ + threadId: TrimmedNonEmptyString, + usage: ThreadTokenUsageSnapshot, + updatedAt: Schema.String, + }), + ), + ), + threadUsage: Schema.optional(Schema.Array(UsageThreadTotals)), contractVersion: Schema.Number, readAt: Schema.String, timeZone: TrimmedNonEmptyString,