diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 813efbe53ea30..99e58679fd55d 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -255,6 +255,7 @@ export const make = Effect.gen(function* () { modelCatalogRefresh: true, usagePriceOverrides: true, usageModelAliases: true, + usageThreadFilter: true, threadPinning: true, threadPinReorder: true, threadActiveReorder: true, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 30f003bf12891..661f7393cd637 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -311,6 +311,7 @@ const layerUsage = UsageService.layer.pipe( CursorUsageAccounts.layer.pipe(Layer.provide(CursorAccountReader.layer)), ).pipe(Layer.provide(ProviderHostLive.layer.pipe(Layer.provide(ServerSecretStore.layer)))), ), + Layer.provide(ProjectionStoreV2.layer), Layer.provide(layerServerSettings), ); diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 38847bb6c8d5e..a11cb08272f18 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -18,11 +18,13 @@ import { EnvironmentId, ProviderDriverKind, ProviderInstanceId, + ThreadId, UsageDay, type UsageSource, type UsageSummary, type UsageSummaryInput, } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Clock from "effect/Clock"; import * as Deferred from "effect/Deferred"; @@ -47,6 +49,7 @@ import * as CursorAccountReader from "@t3tools/provider-cursor/server/CursorAcco import * as CursorUsageAccounts from "@t3tools/provider-cursor/server/CursorUsageAccounts"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; +import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; import * as AntigravityUsage from "../provider/Drivers/AntigravityUsage.ts"; import * as ProviderHostLive from "../provider/ProviderHostLive.ts"; import { totalTokens, type UsageRecord } from "@t3tools/provider-core/server/usage"; @@ -195,6 +198,7 @@ const layerService = (input: { readonly platform?: NodeJS.Platform; readonly hostname?: string; readonly secretStore?: Partial; + readonly projectionStore?: Partial; }) => layerCursorUsageAccounts(CursorAccountReader.layer).pipe( Layer.provideMerge(AntigravityUsage.layer), @@ -202,6 +206,7 @@ const layerService = (input: { Layer.provideMerge(ProviderHostLive.layer), Layer.provideMerge(CodexUsageHistory.layer.pipe(Layer.provideMerge(Sqlite.layerMemory))), Layer.provideMerge(Layer.mock(BackgroundPolicy.BackgroundPolicy)({})), + Layer.provideMerge(Layer.mock(ProjectionStore.ProjectionStoreV2)({ ...input.projectionStore })), Layer.provideMerge( Layer.mock(ServerSecretStore.ServerSecretStore)({ get: () => Effect.succeedNone, @@ -2824,6 +2829,95 @@ describe("UsageService", () => { }), ); + it.effect("narrows a summary to a thread's sessions from when it was created", () => + Effect.gen(function* () { + const { transcript, settings, home } = yield* setup; + // A fork's transcript repeats the source's history under the fork's id, then adds its own turn. + yield* Effect.promise(() => + NodeFSP.writeFile( + transcript, + claudeLine(1, 3, "orbit-1", "source") + + claudeLine(2, 5, "orbit-1", "fork") + + claudeLine(3, 7, "orbit-1", "fork").replace("T10:00:00Z", "T12:00:00Z"), + ), + ); + const claudeThread = (nativeId: string | null) => ({ + nativeThreadRef: + nativeId === null + ? null + : { + driver: ProviderDriverKind.make("claudeAgent"), + nativeId, + strength: "strong" as const, + }, + }); + const threads: Record = { + source: { createdAt: "2026-08-01T09:00:00Z", providerThreads: [claudeThread("source")] }, + // Forked at 11:00, with no conversation yet on its second provider. + fork: { + createdAt: "2026-08-01T11:00:00Z", + providerThreads: [claudeThread("fork"), claudeThread(null)], + }, + unstarted: { createdAt: "2026-08-01T09:00:00Z", providerThreads: [] }, + }; + const getThreadRecords = ((threadId: ThreadId) => { + const thread = threads[threadId]; + return thread === undefined + ? Effect.fail(new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId })) + : Effect.succeed({ + thread: { createdAt: DateTime.makeUnsafe(thread.createdAt) }, + providerThreads: thread.providerThreads, + }); + }) as ProjectionStore.ProjectionStoreV2["Service"]["getThreadRecords"]; + const outputTokens = (summary: UsageSummary) => + summary.buckets.reduce((sum, bucket) => sum + bucket.totals.outputTokens, 0); + + yield* Effect.gen(function* () { + const service = yield* UsageService.make; + const ofThread = (id: string) => + service.readSummary({ ...WINDOW, threadId: ThreadId.make(id) }); + // Same service, concurrent reads: the in-flight key must keep these scans apart. + const [all, source, fork, unstarted, missing] = yield* Effect.all( + [ + service.readSummary(WINDOW), + ofThread("source"), + ofThread("fork"), + // A thread with no conversation yet has no usage, not everyone's. + ofThread("unstarted"), + ofThread("deleted").pipe(Effect.flip), + ], + { concurrency: "unbounded" }, + ); + + assert.strictEqual(outputTokens(all), 15); + assert.strictEqual(outputTokens(source), 3); + assert.strictEqual(outputTokens(fork), 7); + assert.strictEqual(fork.sources[0]?.distinctSessions, 1); + assert.strictEqual(outputTokens(unstarted), 0); + assert.strictEqual(missing.reason, "scanFailed"); + + // A first load asks for the last scan's answer, which is narrowed the same way. + const saved = yield* service.readSummary({ + ...WINDOW, + provisional: true, + threadId: ThreadId.make("fork"), + }); + assert.strictEqual(outputTokens(saved), 7); + assert.isTrue(saved.sources[0]?.refreshing); + }).pipe( + Effect.scoped, + Effect.provide( + layerService({ + prefix: "usage-service-thread-filter-test", + home, + settings, + projectionStore: { getThreadRecords }, + }), + ), + ); + }).pipe(Effect.scoped), + ); + it.effect("reprices unchanged transcripts when custom prices are added, edited, or removed", () => Effect.gen(function* () { const { transcript, settings, home } = yield* setup; diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index d84de15a59ed9..d0b9c5cf42ac7 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -1,4 +1,6 @@ import { + type OrchestrationV2ProviderThread, + type OrchestrationV2ThreadProjection, type ProviderInstanceConfig, ProviderInstanceId, USAGE_CONTRACT_VERSION, @@ -30,6 +32,7 @@ import { HttpClient, HttpClientResponse } from "effect/http"; import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite"; import { formatMalformedUsageWarning } from "@t3tools/shared/usageFormat"; import * as ServerConfig from "../config.ts"; +import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; import * as ServerSettings from "../serverSettings.ts"; import { BUILT_IN_USAGE_DRIVERS, type BuiltInUsageReadersEnv } from "../provider/builtInDrivers.ts"; import type { ProviderDriver } from "@t3tools/provider-core/server/driver"; @@ -39,7 +42,12 @@ import type { TranscriptUsageFormat, UsageRecord, } from "@t3tools/provider-core/server/usage"; -import { canonicalModelFilter, resolveModelAliases, UsageAggregator } from "./usageAggregation.ts"; +import { + canonicalModelFilter, + resolveModelAliases, + UsageAggregator, + type UsageSessions, +} from "./usageAggregation.ts"; import { createOverrideRateTable, isUnlistedModel, @@ -151,6 +159,22 @@ const decodeCachedSources = Schema.decodeUnknownOption( }), ); +/** A summary request once its thread, if any, is resolved to sessions. */ +type UsageScanInput = Omit; + +/** Where a thread's usage is recorded: the native conversations of its provider threads. */ +function threadUsageSessions(records: { + readonly thread: Pick; + readonly providerThreads: ReadonlyArray>; +}): UsageSessions { + return { + ids: new Set( + records.providerThreads.flatMap((thread) => thread.nativeThreadRef?.nativeId ?? []), + ), + sinceTimeMs: DateTime.toEpochMillis(records.thread.createdAt), + }; +} + export class UsageService extends Context.Service< UsageService, { @@ -185,6 +209,7 @@ export const make = Effect.gen(function* () { return id; }); const hostname = yield* HostProcess.Hostname; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; const readerContext = yield* Effect.context(); const fileCache: ScanCache = new Map(); @@ -819,9 +844,10 @@ export const make = Effect.gen(function* () { }); const scanSummary = Effect.fn("UsageService.scanSummary")(function* ( - input: UsageSummaryInput, + input: UsageScanInput, settings: ServerSettingsValue, provisional: boolean, + sessions: UsageSessions | undefined, ) { if (input.sinceDay > input.untilDay) { return yield* new UsageReadError({ @@ -999,6 +1025,7 @@ export const make = Effect.gen(function* () { priceOverrides, modelAliases, ...(input.modelFilter === undefined ? {} : { modelFilter: input.modelFilter }), + ...(sessions === undefined ? {} : { sessions }), }); const windowEndMs = @@ -1167,23 +1194,25 @@ export const make = Effect.gen(function* () { }); const readWindow = Effect.fn("UsageService.readWindow")(function* ( - input: UsageSummaryInput, + input: UsageScanInput, settings: ServerSettingsValue, provisional: boolean, + sessions: UsageSessions | undefined, ) { - const { summary, fromMemory } = yield* scanSummary(input, settings, provisional); + const { summary, fromMemory } = yield* scanSummary(input, settings, provisional, sessions); // With nothing marked refreshing the client would keep a saved answer as final. return fromMemory && !summary.sources.some((source) => source.refreshing === true) - ? (yield* scanSummary(input, settings, false)).summary + ? (yield* scanSummary(input, settings, false, sessions)).summary : summary; }); const inflightScans = new Map>(); const scanKey = ( - input: UsageSummaryInput, + input: UsageScanInput, settings: ServerSettingsValue, provisional: boolean, + sessions: UsageSessions | undefined, ): string => JSON.stringify([ input.timeZone, @@ -1196,17 +1225,32 @@ export const make = Effect.gen(function* () { input.awaitRefresh === true, provisional, input.modelFilter ?? null, + sessions === undefined ? null : [[...sessions.ids].sort(), sessions.sinceTimeMs], ]); const readSummary = Effect.fn("UsageService.readSummary")(function* ( requested: UsageSummaryInput, ) { - const { modelFilter: requestedFilter, ...unfiltered } = requested; + const { modelFilter: requestedFilter, threadId, ...unfiltered } = requested; + const sessions = + threadId === undefined + ? undefined + : yield* projectionStore.getThreadRecords(threadId, ["providerThreads"]).pipe( + Effect.map(threadUsageSessions), + Effect.mapError( + (cause) => + new UsageReadError({ + reason: "scanFailed", + detail: "Could not read the thread's provider sessions", + cause, + }), + ), + ); const modelFilter = canonicalModelFilter(requestedFilter); const input = modelFilter === undefined ? unfiltered : { ...unfiltered, modelFilter }; const settings = yield* readSettings; const provisional = input.provisional === true && input.awaitRefresh !== true; - const key = scanKey(input, settings, provisional); + const key = scanKey(input, settings, provisional, sessions); const deferred = yield* Effect.uninterruptible( Effect.gen(function* () { const existing = inflightScans.get(key); @@ -1217,7 +1261,7 @@ export const make = Effect.gen(function* () { 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* readWindow(input, settings, provisional).pipe( + yield* readWindow(input, settings, provisional, sessions).pipe( Effect.onExit((exit) => Effect.sync(() => inflightScans.delete(key)).pipe( Effect.andThen(Exit.isSuccess(exit) ? schedulePersist : Effect.void), diff --git a/apps/server/src/usage/usageAggregation.test.ts b/apps/server/src/usage/usageAggregation.test.ts index f99615085bd2a..616755985428b 100644 --- a/apps/server/src/usage/usageAggregation.test.ts +++ b/apps/server/src/usage/usageAggregation.test.ts @@ -88,6 +88,26 @@ describe("UsageAggregator", () => { expect(bucket).not.toHaveProperty("ultrafastCostUsd"); }); + it("keeps only the chosen sessions' records, before deduplicating", () => { + const aggregator = new UsageAggregator({ + timeZone: "UTC", + sinceDay: "2026-08-01", + untilDay: "2026-08-31", + rates, + sessions: { ids: new Set(["session-fork"]), sinceTimeMs: Date.parse("2026-08-08T00:00:00Z") }, + }); + const forkTurn = { sessionId: "session-fork", timestampMs: Date.parse("2026-08-08T12:00:00Z") }; + // Another session's copy comes first and must not claim the key the fork's record shares. + aggregator.add(record({ dedupeKey: "msg_2:", timestampMs: forkTurn.timestampMs }), "a.jsonl"); + // History the fork copied from its source predates the fork. + aggregator.add(record({ dedupeKey: "msg_1:", sessionId: "session-fork" }), "b.jsonl"); + aggregator.add(record({ dedupeKey: "msg_2:", ...forkTurn }), "b.jsonl"); + + expect(aggregator.finish().buckets).toMatchObject([ + { day: "2026-08-08", sourcePath: "b.jsonl", records: 1, sessions: 1 }, + ]); + }); + it("merges partial copies of a key by channel maximum in the first copy's bucket", () => { const zeroTotals = { uncachedInputTokens: 0, diff --git a/apps/server/src/usage/usageAggregation.ts b/apps/server/src/usage/usageAggregation.ts index 9ff9cb9369462..9f853e78e8c58 100644 --- a/apps/server/src/usage/usageAggregation.ts +++ b/apps/server/src/usage/usageAggregation.ts @@ -63,6 +63,13 @@ interface MutableBucket { sessions: Set; } +/** Keeps only these sessions' records at or after `sinceTimeMs`. */ +export interface UsageSessions { + readonly ids: ReadonlySet; + /** A forked session repeats its source's earlier records; this instant leaves them out. */ + readonly sinceTimeMs: number; +} + export interface AggregateOptions { readonly timeZone: string; readonly sinceDay: string; @@ -71,6 +78,7 @@ export interface AggregateOptions { readonly priceOverrides?: RateTable; readonly modelAliases?: ReadonlyMap; readonly modelFilter?: UsageModelFilter; + readonly sessions?: UsageSessions; readonly resolution?: UsageResolution; readonly sinceTimeMs?: number; readonly untilTimeMs?: number; @@ -212,6 +220,13 @@ export class UsageAggregator { add(input: UsageRecord, sourcePath?: string): boolean { if (!keepsUsageRecord(this.#options.modelFilter, input)) return false; + const sessions = this.#options.sessions; + if ( + sessions !== undefined && + (!sessions.ids.has(input.sessionId) || input.timestampMs < sessions.sinceTimeMs) + ) { + return false; + } const record = this.#mapModel(input); if (record.dedupeKey !== null) { const held = this.#seen.get(record.dedupeKey); diff --git a/apps/web/src/components/chat/ThreadDetailsPanel.tsx b/apps/web/src/components/chat/ThreadDetailsPanel.tsx index 4942080dbb2d2..0ffc64d11309c 100644 --- a/apps/web/src/components/chat/ThreadDetailsPanel.tsx +++ b/apps/web/src/components/chat/ThreadDetailsPanel.tsx @@ -27,7 +27,7 @@ import { ThreadAutomationsPanel } from "./ThreadAutomationsPanel"; import { ThreadRelationshipsPanel } from "./ThreadRelationshipsControl"; import { ThreadServersPanel } from "./ThreadServersPanel"; import { - ThreadDetailsContext, + ThreadDetailsStats, ThreadDetailsUsage, type ThreadDetailsUsageInput, } from "./ThreadDetailsUsage"; @@ -176,6 +176,8 @@ export function ThreadDetailsPanel(props: ThreadDetailsPanelProps) { offersGit && props.activeProjectName !== undefined && (visible.gitActions || (visible.changes && props.isGitRepo)), + context: visible.context ? props.usage.contextWindow : null, + cost: visible.cost && !props.draftId, }; return ( <> @@ -286,8 +288,15 @@ export function ThreadDetailsPanel(props: ThreadDetailsPanelProps) { {visible.usage ? : null} - {visible.context && props.usage.contextWindow ? ( - + {show.context || show.cost ? ( + ) : null} {/* Shown only when no section above has content, so an emptied card can be refilled. */} diff --git a/apps/web/src/components/chat/ThreadDetailsUsage.tsx b/apps/web/src/components/chat/ThreadDetailsUsage.tsx index 0d6c0129484ed..9d972f41d0389 100644 --- a/apps/web/src/components/chat/ThreadDetailsUsage.tsx +++ b/apps/web/src/components/chat/ThreadDetailsUsage.tsx @@ -1,9 +1,12 @@ import type { + EnvironmentId, ProviderInstanceId, ServerProvider, ServerProviderUsageWindow, + ThreadId, UsageLimitSourceSnapshots, } from "@t3tools/contracts"; +import { formatPercent, formatUsd } from "@t3tools/shared/usageFormat"; import { collectProviderUsageLimits, formatDuration, @@ -12,15 +15,17 @@ import { paceOf, remainingPercent, } from "@t3tools/shared/usageLimits"; -import type { ReactNode } from "react"; +import { Fragment, type ReactNode } from "react"; import { usePrimarySettings } from "../../hooks/useSettings"; import { useNowMinute } from "../../hooks/useNowMinute"; import type { ContextWindowSnapshot } from "@t3tools/client-runtime/state/context-window"; import { formatContextWindowTokens } from "../../lib/contextWindow"; import { cn } from "../../lib/utils"; +import { useThreadUsage } from "../../state/usage"; import { formatUpcomingTimestamp } from "../../timestampFormat"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; +import { costTypeSegments } from "../usage/usageBreakdown"; import { formatContextWindowCompactionMessage } from "./compaction.logic"; import { ThreadDetailsSection } from "./ThreadDetailsSection"; @@ -82,8 +87,31 @@ export function ThreadDetailsUsage({ ); } -/** The thread's context window. The bar fills as context is used; the tick marks auto-compaction. */ -export function ThreadDetailsContext({ usage }: { readonly usage: ContextWindowSnapshot }) { +/** The thread's own figures. They share one grid, so their labels and values line up. */ +export function ThreadDetailsStats({ + contextWindow, + cost, +}: { + readonly contextWindow: ContextWindowSnapshot | null; + /** The thread to price. Null hides the cost row. */ + readonly cost: { readonly environmentId: EnvironmentId; readonly threadId: ThreadId } | null; +}) { + return ( + +
+ {contextWindow ? : null} + {cost ? : null} +
+ + ); +} + +/** The bar fills as context is used; the tick marks auto-compaction. */ +function ContextRow({ usage }: { readonly usage: ContextWindowSnapshot }) { const maxTokens = usage.maxTokens != null && usage.maxTokens > 0 ? usage.maxTokens : null; const used = maxTokens !== null && usage.usedPercentage !== null @@ -96,48 +124,115 @@ export function ThreadDetailsContext({ usage }: { readonly usage: ContextWindowS : null; const usedTokens = formatContextWindowTokens(usage.usedTokens); return ( - -
- = 90 ? "error" : used >= 80 ? "warning" : null} - tick={tick} - muted - value={ - maxTokens !== null - ? `${usedTokens} / ${formatContextWindowTokens(maxTokens)}` - : usedTokens - } - ariaValue={used !== null ? `${used}% used` : `${usedTokens} tokens used`} - tooltip={ - <> - - {used !== null ? `Context: ${used}% used` : `Context: ${usedTokens} tokens used`} - - {maxTokens !== null ? ( - - {usage.usedTokens.toLocaleString("en-US")} of {maxTokens.toLocaleString("en-US")}{" "} - tokens - - ) : null} - {threshold !== null && threshold > 0 ? ( - - {formatContextWindowCompactionMessage(threshold)} - - ) : null} - - } - /> -
- + = 90 ? "error" : used >= 80 ? "warning" : null} + tick={tick} + muted + value={ + maxTokens !== null ? `${usedTokens} / ${formatContextWindowTokens(maxTokens)}` : usedTokens + } + ariaValue={used !== null ? `${used}% used` : `${usedTokens} tokens used`} + tooltip={ + <> + + {used !== null ? `Context: ${used}% used` : `Context: ${usedTokens} tokens used`} + + {maxTokens !== null ? ( + + {usage.usedTokens.toLocaleString("en-US")} of {maxTokens.toLocaleString("en-US")}{" "} + tokens + + ) : null} + {threshold !== null && threshold > 0 ? ( + + {formatContextWindowCompactionMessage(threshold)} + + ) : null} + + } + /> + ); +} + +/** + * What the thread's model usage would cost at API prices, as the Usage tab prices it. A figure, + * not a meter, so it has no bar. Renders nothing until the server reports usage for the thread. + */ +function CostRow({ + environmentId, + threadId, +}: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; +}) { + const usage = useThreadUsage({ environmentId, threadId }); + // No records means the provider's usage is not read here, which is not a cost of zero. + if (usage === null || usage.records === 0) return null; + const { unpricedShare } = usage.costQuality; + if (unpricedShare >= 1) { + return ( + + Cost: no known price for this model + + {formatContextWindowTokens(usage.totalTokens)} tokens + + + } + /> + ); + } + const value = formatCost(usage.costUsd); + // A part that rounds to no cents says nothing next to the total. + const parts = costTypeSegments(usage.categoryCost).filter((part) => part.value >= 0.005); + return ( + + Cost: {value} ยท API estimate + {parts.length > 0 ? ( + + {parts.map((part) => ( + + {part.label} + + {formatUsd(part.value)} + + + ))} + + ) : null} + + {formatContextWindowTokens(usage.totalTokens)} tokens + + {unpricedShare > 0 ? ( + + Excludes {formatPercent(unpricedShare)} unpriced records + + ) : null} + + } + /> ); } +/** A cost that rounds to no cents still cost something. */ +function formatCost(usd: number): string { + return usd > 0 && usd < 0.005 ? "<$0.01" : formatUsd(usd); +} + function LimitRow({ window, now, @@ -184,6 +279,7 @@ function UsageRow({ ariaValue = value, tooltip, muted = false, + dimValue, }: { readonly label: string; /** Bar fill from 0 to 100. Null hides the bar. */ @@ -195,6 +291,8 @@ function UsageRow({ readonly ariaValue?: string; readonly tooltip: ReactNode; readonly muted?: boolean; + /** A row without a bar dims its value, as a notice. False keeps it a figure. */ + readonly dimValue?: boolean; }) { return ( @@ -243,7 +341,7 @@ function UsageRow({ ? "text-error-foreground" : tone === "warning" ? "text-warning-foreground" - : fill === null + : (dimValue ?? fill === null) ? "text-muted-foreground" : "text-foreground/80", )} diff --git a/apps/web/src/components/chat/threadDetailsItems.ts b/apps/web/src/components/chat/threadDetailsItems.ts index e6f94cc4e11c8..2dd423b49d361 100644 --- a/apps/web/src/components/chat/threadDetailsItems.ts +++ b/apps/web/src/components/chat/threadDetailsItems.ts @@ -14,6 +14,7 @@ export const THREAD_DETAILS_ITEMS = [ { id: "sideChats", label: "Side chats", defaultVisible: true }, { id: "usage", label: "Usage limits", defaultVisible: false }, { id: "context", label: "Context window", defaultVisible: false }, + { id: "cost", label: "Cost", defaultVisible: false }, ] as const; export type ThreadDetailsItemId = (typeof THREAD_DETAILS_ITEMS)[number]["id"]; diff --git a/apps/web/src/state/usage.ts b/apps/web/src/state/usage.ts index bf5de2e330a3f..1fb9c82af4d7a 100644 --- a/apps/web/src/state/usage.ts +++ b/apps/web/src/state/usage.ts @@ -3,13 +3,21 @@ import { AuthDiagnosticsReadScope, USAGE_CONTRACT_VERSION, type EnvironmentId, + type ScopedThreadRef, type UsageBucket, type UsageSummary, type UsageProviderKind, type UsageSummaryInput, } from "@t3tools/contracts"; -import { refreshUsage, usageSummaryInput } from "@t3tools/client-runtime/state/usage"; +import { parseThreadKey, threadKey } from "@t3tools/client-runtime/state/entities"; +import { + makeTurnEndRefresh, + refreshUsage, + threadUsageSummaryInput, + usageSummaryInput, +} from "@t3tools/client-runtime/state/usage"; import { resolveUsageAccess } from "@t3tools/client-runtime/state/usage-access"; +import * as DateTime from "effect/DateTime"; import { AsyncResult, Atom } from "effect/reactivity"; import { useState } from "react"; @@ -17,7 +25,8 @@ import { mergeUsage, type EnvironmentUsage, type MergedUsage } from "@t3tools/sh import { appAtomRegistry } from "../rpc/atomRegistry"; import { environmentPresentations } from "./presentation"; import { serverEnvironment } from "./server"; -import { environmentSession, readEnvironmentScope } from "./session"; +import { environmentSession, readEnvironmentScope, sessionHasScope } from "./session"; +import { environmentThreadDetails } from "./threads"; export interface EnvironmentUsageStatus { readonly environmentId: EnvironmentId; @@ -244,3 +253,64 @@ export function useUsage( refresh, }; } + +const threadUsageAtom = Atom.family((key: string) => { + const ref = parseThreadKey(key); + const { environmentId, threadId } = ref; + const shouldRefresh = makeTurnEndRefresh(); + // Stays on screen while a changed query, such as a new day's, is still being read. + let answered: MergedUsage | null = null; + return Atom.make((get): MergedUsage | null => { + if ( + !sessionHasScope( + get(environmentSession.sessionStateAtom(environmentId)), + AuthDiagnosticsReadScope, + ) + ) { + return null; + } + const projection = get(environmentThreadDetails.quietThreadAtom(ref))?.projection; + const input = + projection === undefined + ? null + : threadUsageSummaryInput({ + threadId, + createdAt: projection.thread.createdAt, + capabilities: get(environmentPresentations.presentationAtom(environmentId)) + ?.serverConfig?.environment.capabilities, + }); + if (projection === undefined || input === null) return null; + const query = serverEnvironment.usageSummary({ + environmentId, + input: usageSummaryInput(input, get(serverEnvironment.usageModelFilterAtom(environmentId))), + }); + const result = get(query); + const lastTurnEndMs = projection.providerTurns.reduce( + (latest, turn) => + turn.completedAt === null + ? latest + : Math.max(latest, DateTime.toEpochMillis(turn.completedAt)), + 0, + ); + if (shouldRefresh(lastTurnEndMs, result.waiting)) get.refresh(query); + // A failed read has no figure; the one before it would pass for current. + if (AsyncResult.isFailure(result) && !AsyncResult.isInterrupted(result)) { + answered = null; + return null; + } + const summary = AsyncResult.getOrElse(result, () => null); + if (summary === null) return answered; + const merged = mergeUsage([{ environmentId, label: "", summary }], USAGE_CONTRACT_VERSION); + // A summary from another contract version merges to nothing, which is not a zero cost. + answered = merged.contractMismatches.length === 0 ? merged : null; + return answered; + }).pipe(Atom.withLabel(`web-usage:thread:${key}`)); +}); + +/** + * One thread's usage so far, priced like the Usage tab. Null until the server answers, and on a + * server that cannot narrow usage to a thread. + */ +export function useThreadUsage(ref: ScopedThreadRef): MergedUsage | null { + return useAtomValue(threadUsageAtom(threadKey(ref))); +} diff --git a/docs/user/usage.md b/docs/user/usage.md index e6b3a46d5a14e..f11f2db2efb71 100644 --- a/docs/user/usage.md +++ b/docs/user/usage.md @@ -150,6 +150,15 @@ row to see when it resets and your pace. Turn on **Context window** to also show context use against the model's limit. A tick on the bar marks where the provider compacts automatically, when it reports that point. +Turn on **Cost** to show what the thread's model usage so far would cost at API prices. It is the +same estimate as the Usage tab, narrowed to the thread, so on a subscription it is not a charge. +Hover the row to see the cost by token type. It covers the thread's own conversations, including +subagents the provider runs inside them, but not other threads it started. The row appears once +the provider's usage for the thread is recorded, and updates when a turn ends. If the provider had +to start a fresh conversation in the thread, such as after a failed resume, usage before it is left +out. The row stays hidden when the saved usage does not name the thread's conversation, as with Pi, when your +session cannot read usage, and on a server too old to narrow usage to a thread. + OpenCode Go reports its session, weekly, and monthly allowance when OpenCode runs locally in the environment. T3 cannot report limits for external OpenCode servers because their credentials belong to the remote server. Limits need an OpenCode Go API key, or an OpenCode Console connected diff --git a/packages/client-runtime/src/state/usage.test.ts b/packages/client-runtime/src/state/usage.test.ts index 380e2de731b5d..a3f6c2ff99202 100644 --- a/packages/client-runtime/src/state/usage.test.ts +++ b/packages/client-runtime/src/state/usage.test.ts @@ -1,17 +1,24 @@ import { EnvironmentId, + ThreadId, UsageDay, USAGE_CONTRACT_VERSION, type UsageModelFilter, type UsageSummary, type UsageSummaryInput, } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity"; import { afterEach, describe, expect, it, vi } from "vite-plus/test"; import type { EnvironmentPresentation } from "../connection/presentation.ts"; -import { refreshUsage, refreshUsageLimits } from "./usage.ts"; +import { + makeTurnEndRefresh, + refreshUsage, + refreshUsageLimits, + threadUsageSummaryInput, +} from "./usage.ts"; const input = { sinceDay: UsageDay.make("2026-09-05"), @@ -205,3 +212,56 @@ describe("limits refresh cooldown", () => { expect(newCheck).not.toHaveBeenCalled(); }); }); + +describe("threadUsageSummaryInput", () => { + const threadId = ThreadId.make("thread-1"); + const capabilities = { usageThreadFilter: true }; + const createdAt = DateTime.makeUnsafe("2026-09-03T08:15:00.000Z"); + const now = DateTime.makeUnsafe("2026-09-05T12:00:00Z"); + + it("asks for the thread's usage from the day it was created", () => { + expect(threadUsageSummaryInput({ threadId, createdAt, capabilities, now })).toMatchObject({ + sinceDay: "2026-09-03", + untilDay: "2026-09-05", + threadId, + }); + }); + + it("asks nothing of a server that would answer with every thread's usage", () => { + expect(threadUsageSummaryInput({ threadId, createdAt, capabilities: {}, now })).toBeNull(); + expect( + threadUsageSummaryInput({ threadId, createdAt, capabilities: undefined, now }), + ).toBeNull(); + }); + + it("keeps the creation day in the window when the viewer's clock trails the server's", () => { + expect( + threadUsageSummaryInput({ + threadId, + createdAt, + capabilities, + now: DateTime.makeUnsafe("2026-09-01T12:00:00Z"), + }), + ).toMatchObject({ sinceDay: "2026-09-03", untilDay: "2026-09-03" }); + }); +}); + +describe("makeTurnEndRefresh", () => { + it("reads again when a turn ends after the answer in hand", () => { + const shouldRefresh = makeTurnEndRefresh(); + expect(shouldRefresh(100, true)).toBe(false); + expect(shouldRefresh(100, false)).toBe(false); + expect(shouldRefresh(200, false)).toBe(true); + expect(shouldRefresh(200, true)).toBe(false); + expect(shouldRefresh(200, false)).toBe(false); + }); + + it("follows a read that a turn end overtook with one more", () => { + const shouldRefresh = makeTurnEndRefresh(); + expect(shouldRefresh(100, true)).toBe(false); + // The turn ends while the first read is still scanning. + expect(shouldRefresh(200, true)).toBe(false); + expect(shouldRefresh(200, false)).toBe(true); + expect(shouldRefresh(200, false)).toBe(false); + }); +}); diff --git a/packages/client-runtime/src/state/usage.ts b/packages/client-runtime/src/state/usage.ts index f302ffbdfd85c..30ce002c87308 100644 --- a/packages/client-runtime/src/state/usage.ts +++ b/packages/client-runtime/src/state/usage.ts @@ -1,4 +1,12 @@ -import type { EnvironmentId, UsageModelFilter, UsageSummaryInput } from "@t3tools/contracts"; +import type { + EnvironmentId, + ExecutionEnvironmentCapabilities, + ThreadId, + UsageModelFilter, + UsageSummaryInput, +} from "@t3tools/contracts"; +import { makeWindow } from "@t3tools/shared/usageFormat"; +import * as DateTime from "effect/DateTime"; import * as Schema from "effect/Schema"; import type { AtomRegistry } from "effect/reactivity"; @@ -17,6 +25,50 @@ export function usageSummaryInput( return modelFilter === undefined ? input : { ...input, modelFilter }; } +/** + * The usage query for one thread, from the day it was created. Null for a server that would + * ignore `threadId` and answer with every thread's usage. + */ +export function threadUsageSummaryInput(input: { + readonly threadId: ThreadId; + readonly createdAt: DateTime.Utc; + readonly capabilities: Pick | undefined; + readonly now?: DateTime.Utc; +}): UsageSummaryInput | null { + if (input.capabilities?.usageThreadFilter !== true) return null; + const { untilDay: today, timeZone } = makeWindow( + 1, + input.now === undefined ? undefined : DateTime.toDateUtc(input.now), + ); + const createdDay = makeWindow(1, DateTime.toDateUtc(input.createdAt)).untilDay; + return { + sinceDay: createdDay, + // A viewer whose clock trails the server's can see a thread created "tomorrow". + untilDay: createdDay < today ? today : createdDay, + timeZone, + threadId: input.threadId, + }; +} + +/** + * Decides when to read a thread's usage again. A turn's last records reach the transcript as it + * ends, and a read in flight at that moment may have scanned before them, so that read's answer + * is followed by another. + */ +export function makeTurnEndRefresh(): (lastTurnEndMs: number, waiting: boolean) => boolean { + // The latest turn end that the read in flight, or the answer in hand, began after. + let readAfter: number | null = null; + return (lastTurnEndMs, waiting) => { + if (readAfter === null) { + readAfter = lastTurnEndMs; + return false; + } + if (waiting || readAfter === lastTurnEndMs) return false; + readAfter = lastTurnEndMs; + return true; + }; +} + const limitsRefreshAfter = new Map(); const limitsRefreshes = new Map>(); diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 603b6c6cc862d..cfb905e86b405 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -131,6 +131,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ modelCatalogRefresh: Schema.optionalKey(Schema.Boolean), usagePriceOverrides: Schema.optionalKey(Schema.Boolean), usageModelAliases: Schema.optionalKey(Schema.Boolean), + /** Usage summaries honor `threadId`; older servers ignore it and return every thread's usage. */ + usageThreadFilter: Schema.optionalKey(Schema.Boolean), threadPinning: Schema.optionalKey(Schema.Boolean), threadPinReorder: Schema.optionalKey(Schema.Boolean), threadActiveReorder: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/usage.ts b/packages/contracts/src/usage.ts index 427d6c93af5c2..b0486850d881a 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; @@ -145,6 +150,8 @@ export const UsageSummaryInput = Schema.Struct({ /** Accept an immediate answer from the last scan; its sources are marked refreshing. */ provisional: Schema.optional(Schema.Boolean), modelFilter: Schema.optional(UsageModelFilter), + /** Narrows usage to one thread: its provider sessions, from when it was created. */ + threadId: Schema.optional(ThreadId), }); export type UsageSummaryInput = typeof UsageSummaryInput.Type;