From f94d7e4aee5678b4287c5b86fe20835183a92eee Mon Sep 17 00:00:00 2001 From: flexi767 Date: Thu, 12 Nov 2026 14:39:20 +0100 Subject: [PATCH] feat(web): show each thread's session cost in the sidebar Threads give no sign of what a session has cost. The usage service can now attribute priced native-session records to the T3 thread that owns them, and each sidebar row shows that thread's API-equivalent cost, with a per-category breakdown on hover. Usage refreshes once a minute only while a thread in the environment is working. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/usage/UsageService.test.ts | 55 +++++++ apps/server/src/usage/UsageService.ts | 39 +++++ .../server/src/usage/usageAggregation.test.ts | 52 +++++++ apps/server/src/usage/usageAggregation.ts | 57 ++++++- .../components/sidebar/SessionUsageBadge.tsx | 139 ++++++++++++++++++ .../components/sidebar/SidebarThreadRow.tsx | 41 ++++-- apps/web/src/lib/sessionUsage.test.ts | 30 ++++ apps/web/src/lib/sessionUsage.ts | 33 +++++ apps/web/src/state/sessionUsage.ts | 20 +++ apps/web/src/state/workingUsageQuery.test.ts | 35 +++++ apps/web/src/state/workingUsageQuery.ts | 23 +++ packages/client-runtime/src/state/server.ts | 24 +++ .../src/state/serverUsage.test.ts | 17 +++ packages/contracts/src/usage.ts | 12 ++ 14 files changed, 561 insertions(+), 16 deletions(-) create mode 100644 apps/web/src/components/sidebar/SessionUsageBadge.tsx create mode 100644 apps/web/src/lib/sessionUsage.test.ts create mode 100644 apps/web/src/lib/sessionUsage.ts create mode 100644 apps/web/src/state/sessionUsage.ts create mode 100644 apps/web/src/state/workingUsageQuery.test.ts create mode 100644 apps/web/src/state/workingUsageQuery.ts diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index ae0126d438254..cedb5893f950c 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -520,6 +520,61 @@ 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" }, + }), + })}`; + 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); + }).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..a3f83002cb355 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -26,6 +26,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"; @@ -156,6 +158,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 +766,41 @@ export const make = Effect.gen(function* () { { concurrency: 2 }, ); + let threadIdsBySession: Map | undefined; + if (input.includeThreadUsage === true) { + 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 +964,7 @@ export const make = Effect.gen(function* () { sinceDay: input.sinceDay, untilDay: input.untilDay, buckets: aggregated.buckets, + ...(aggregated.threadUsage === undefined ? {} : { threadUsage: aggregated.threadUsage }), sources, pricing: pricing(), scanDurationMs: Math.max(0, finishedAtMs - startedAtMs), @@ -947,6 +985,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/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..a9c212e2fb331 100644 --- a/apps/web/src/components/sidebar/SidebarThreadRow.tsx +++ b/apps/web/src/components/sidebar/SidebarThreadRow.tsx @@ -1,6 +1,7 @@ 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 { ThreadTerminalStatusButton } from "../ThreadStatusIndicators"; import { discardComposerDraft } from "../../lib/discardComposerDraft"; import { requestCustomSnooze } from "../CustomSnoozeDialog"; @@ -1156,6 +1157,13 @@ 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 +1201,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 +1235,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 +1431,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..03adb7c5ac217 100644 --- a/packages/contracts/src/usage.ts +++ b/packages/contracts/src/usage.ts @@ -50,6 +50,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 +137,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 +151,7 @@ export const UsageSummaryInput = Schema.Struct({ export type UsageSummaryInput = typeof UsageSummaryInput.Type; export const UsageSummary = Schema.Struct({ + threadUsage: Schema.optional(Schema.Array(UsageThreadTotals)), contractVersion: Schema.Number, readAt: Schema.String, timeZone: TrimmedNonEmptyString,