From 2157c98e7b5c6c2fb1d25760889419992804acac Mon Sep 17 00:00:00 2001 From: Jakub Chledowski Date: Wed, 11 Nov 2026 13:23:32 +0100 Subject: [PATCH 1/3] feat(web): thread details can show what a thread has cost The Usage tab prices model usage per day and model, but nothing says what one thread has cost. Thread details gains an optional Cost row, off by default like Usage limits and Context window. It shows the thread's usage at API prices and, on hover, the cost by token type. The figure is the Usage tab's own estimate narrowed to the thread, so the two always agree. Usage summaries take a session filter: the provider sessions the thread has held, from the moment it was created, so a fork does not count the history its transcript repeats. Servers announce the filter with a capability, and the row stays hidden on servers without it. Co-Authored-By: Claude Fable 5.2 --- .../src/environment/ServerEnvironment.ts | 1 + apps/server/src/usage/UsageService.test.ts | 48 +++++ apps/server/src/usage/UsageService.ts | 24 +++ .../server/src/usage/usageAggregation.test.ts | 20 ++ apps/server/src/usage/usageAggregation.ts | 9 + .../components/chat/ThreadDetailsPanel.tsx | 14 +- .../components/chat/ThreadDetailsUsage.tsx | 184 ++++++++++++++---- .../src/components/chat/threadDetailsItems.ts | 1 + apps/web/src/state/usage.ts | 64 +++++- docs/user/usage.md | 6 + .../client-runtime/src/state/usage.test.ts | 52 ++++- packages/client-runtime/src/state/usage.ts | 36 +++- packages/contracts/src/environment.ts | 2 + packages/contracts/src/usage.ts | 9 + 14 files changed, 420 insertions(+), 50 deletions(-) diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 813efbe53ea30..12c9cce697122 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, + usageSessionFilter: true, threadPinning: true, threadPinReorder: true, threadActiveReorder: true, diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 38847bb6c8d5e..8a1b578ffe4b7 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -2824,6 +2824,54 @@ describe("UsageService", () => { }), ); + it.effect("narrows a summary to one conversation's sessions from when it began", () => + 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 forkFilter = { sessionIds: ["fork", "unknown"], sinceTime: "2026-08-01T11:00:00Z" }; + const outputTokens = (summary: UsageSummary) => + summary.buckets.reduce((sum, bucket) => sum + bucket.totals.outputTokens, 0); + + yield* Effect.gen(function* () { + const service = yield* UsageService.make; + // Same service, concurrent reads: the in-flight key must keep these scans apart. + const [all, source, fork, invalid] = yield* Effect.all( + [ + service.readSummary(WINDOW), + service.readSummary({ ...WINDOW, sessionFilter: { sessionIds: ["source"] } }), + service.readSummary({ ...WINDOW, sessionFilter: forkFilter }), + service + .readSummary({ + ...WINDOW, + sessionFilter: { sessionIds: ["fork"], sinceTime: "yesterday" }, + }) + .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(invalid.reason, "invalidWindow"); + }).pipe( + Effect.scoped, + Effect.provide( + layerService({ prefix: "usage-service-session-filter-test", home, settings }), + ), + ); + }).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..10e8fa86f60fc 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -854,6 +854,18 @@ export const make = Effect.gen(function* () { hourlyWindow = { sinceTimeMs, untilTimeMs }; } + const sessionFilter = input.sessionFilter; + const sessionsSince = + sessionFilter?.sinceTime === undefined + ? Option.none() + : DateTime.make(sessionFilter.sinceTime); + if (sessionFilter?.sinceTime !== undefined && Option.isNone(sessionsSince)) { + return yield* new UsageReadError({ + reason: "invalidWindow", + detail: "Session usage requires a valid sinceTime instant", + }); + } + const startedAtMs = yield* Clock.currentTimeMillis; yield* ensureScanCacheLoaded; @@ -999,6 +1011,17 @@ export const make = Effect.gen(function* () { priceOverrides, modelAliases, ...(input.modelFilter === undefined ? {} : { modelFilter: input.modelFilter }), + ...(sessionFilter === undefined + ? {} + : { + sessions: { + ids: new Set(sessionFilter.sessionIds), + sinceTimeMs: Option.match(sessionsSince, { + onNone: () => 0, + onSome: DateTime.toEpochMillis, + }), + }, + }), }); const windowEndMs = @@ -1196,6 +1219,7 @@ export const make = Effect.gen(function* () { input.awaitRefresh === true, provisional, input.modelFilter ?? null, + input.sessionFilter ?? null, ]); 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 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..163aa3f8f42d8 100644 --- a/apps/server/src/usage/usageAggregation.ts +++ b/apps/server/src/usage/usageAggregation.ts @@ -71,6 +71,8 @@ export interface AggregateOptions { readonly priceOverrides?: RateTable; readonly modelAliases?: ReadonlyMap; readonly modelFilter?: UsageModelFilter; + /** Keeps only these sessions' records at or after `sinceTimeMs`. */ + readonly sessions?: { readonly ids: ReadonlySet; readonly sinceTimeMs: number }; readonly resolution?: UsageResolution; readonly sinceTimeMs?: number; readonly untilTimeMs?: number; @@ -212,6 +214,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..e8e9a0ac24a03 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"; @@ -286,8 +286,16 @@ export function ThreadDetailsPanel(props: ThreadDetailsPanelProps) { {visible.usage ? : null} - {visible.context && props.usage.contextWindow ? ( - + {(visible.context && props.usage.contextWindow) || + (visible.cost && !props.draftId) ? ( + ) : 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..451824571b445 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..226bf1bc9a6a1 100644 --- a/apps/web/src/state/usage.ts +++ b/apps/web/src/state/usage.ts @@ -3,12 +3,17 @@ 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 { + refreshUsage, + threadUsageSummaryInput, + usageSummaryInput, +} from "@t3tools/client-runtime/state/usage"; import { resolveUsageAccess } from "@t3tools/client-runtime/state/usage-access"; import { AsyncResult, Atom } from "effect/reactivity"; import { useState } from "react"; @@ -17,7 +22,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 +250,57 @@ export function useUsage( refresh, }; } + +const threadUsageAtom = Atom.family((key: string) => { + const ref = JSON.parse(key) as ScopedThreadRef; + const { environmentId } = ref; + // The settled turns the answer in hand covers. + let pricedTurns: number | null = null; + return Atom.make((get): MergedUsage | null => { + // An older server ignores the session filter and would answer with every session's usage. + if ( + get(environmentPresentations.presentationAtom(environmentId))?.serverConfig?.environment + .capabilities.usageSessionFilter !== true || + !sessionHasScope( + get(environmentSession.sessionStateAtom(environmentId)), + AuthDiagnosticsReadScope, + ) + ) { + return null; + } + const projection = get(environmentThreadDetails.quietThreadAtom(ref))?.projection; + const input = + projection === undefined + ? null + : threadUsageSummaryInput({ + providerThreads: projection.providerThreads, + createdAt: projection.thread.createdAt, + }); + if (projection === undefined || input === null) return null; + const query = serverEnvironment.usageSummary({ + environmentId, + input: usageSummaryInput(input, get(serverEnvironment.usageModelFilterAtom(environmentId))), + }); + // A turn's last records reach the transcript as it ends, so price the thread again then. + const settledTurns = projection.providerTurns.filter( + (turn) => turn.completedAt !== null, + ).length; + if (pricedTurns !== null && pricedTurns !== settledTurns) get.refresh(query); + pricedTurns = settledTurns; + const summary = AsyncResult.getOrElse(get(query), () => null); + if (summary === null) return null; + const merged = mergeUsage([{ environmentId, label: "", summary }], USAGE_CONTRACT_VERSION); + // A summary from another contract version merges to nothing, which is not a zero cost. + return merged.contractMismatches.length === 0 ? merged : null; + }).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(JSON.stringify({ environmentId: ref.environmentId, threadId: ref.threadId })), + ); +} diff --git a/docs/user/usage.md b/docs/user/usage.md index e6b3a46d5a14e..34cc39e834bcd 100644 --- a/docs/user/usage.md +++ b/docs/user/usage.md @@ -150,6 +150,12 @@ 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. + 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..c0a41aa468b20 100644 --- a/packages/client-runtime/src/state/usage.test.ts +++ b/packages/client-runtime/src/state/usage.test.ts @@ -1,17 +1,19 @@ import { EnvironmentId, + ProviderDriverKind, 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 { refreshUsage, refreshUsageLimits, threadUsageSummaryInput } from "./usage.ts"; const input = { sinceDay: UsageDay.make("2026-09-05"), @@ -205,3 +207,51 @@ describe("limits refresh cooldown", () => { expect(newCheck).not.toHaveBeenCalled(); }); }); + +describe("threadUsageSummaryInput", () => { + const providerThread = (nativeId: string | null) => ({ + nativeThreadRef: + nativeId === null + ? null + : { driver: ProviderDriverKind.make("codex"), nativeId, strength: "strong" as const }, + }); + const createdAt = DateTime.makeUnsafe("2026-09-03T08:15:00.000Z"); + const now = DateTime.makeUnsafe("2026-09-05T12:00:00Z"); + + it("asks for every conversation the thread has held, from its creation", () => { + expect( + threadUsageSummaryInput({ + providerThreads: [ + providerThread("session-b"), + providerThread(null), + providerThread("session-a"), + providerThread("session-b"), + ], + createdAt, + now, + }), + ).toMatchObject({ + sinceDay: "2026-09-03", + untilDay: "2026-09-05", + sessionFilter: { + sessionIds: ["session-a", "session-b"], + sinceTime: "2026-09-03T08:15:00.000Z", + }, + }); + }); + + it("has nothing to ask before a provider starts a conversation", () => { + expect( + threadUsageSummaryInput({ providerThreads: [providerThread(null)], createdAt, now }), + ).toBeNull(); + }); + + it("never opens a window that starts after it ends", () => { + const input = threadUsageSummaryInput({ + providerThreads: [providerThread("session-a")], + createdAt, + now: DateTime.makeUnsafe("2026-09-01T12:00:00Z"), + }); + expect(input).toMatchObject({ sinceDay: input?.untilDay }); + }); +}); diff --git a/packages/client-runtime/src/state/usage.ts b/packages/client-runtime/src/state/usage.ts index f302ffbdfd85c..14803de353318 100644 --- a/packages/client-runtime/src/state/usage.ts +++ b/packages/client-runtime/src/state/usage.ts @@ -1,4 +1,11 @@ -import type { EnvironmentId, UsageModelFilter, UsageSummaryInput } from "@t3tools/contracts"; +import type { + EnvironmentId, + OrchestrationV2ProviderThread, + 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 +24,33 @@ export function usageSummaryInput( return modelFilter === undefined ? input : { ...input, modelFilter }; } +/** + * The usage query for one thread: every provider conversation it has held, from when it was + * created. Null until a provider has started one. Ask only servers with `usageSessionFilter`. + */ +export function threadUsageSummaryInput(input: { + readonly providerThreads: ReadonlyArray>; + readonly createdAt: DateTime.Utc; + readonly now?: DateTime.Utc; +}): UsageSummaryInput | null { + const sessionIds = [ + ...new Set(input.providerThreads.flatMap((thread) => thread.nativeThreadRef?.nativeId ?? [])), + ].sort(); + if (sessionIds.length === 0) return null; + const { untilDay, timeZone } = makeWindow( + 1, + input.now === undefined ? undefined : DateTime.toDateUtc(input.now), + ); + const createdDay = makeWindow(1, DateTime.toDateUtc(input.createdAt)).untilDay; + return { + // A viewer whose clock trails the server's can see a thread created "tomorrow". + sinceDay: createdDay < untilDay ? createdDay : untilDay, + untilDay, + timeZone, + sessionFilter: { sessionIds, sinceTime: DateTime.formatIso(input.createdAt) }, + }; +} + const limitsRefreshAfter = new Map(); const limitsRefreshes = new Map>(); diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 603b6c6cc862d..60d54470a2e77 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 `sessionFilter`; older servers ignore it and return every session. */ + usageSessionFilter: 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..8e8ea2f0cfacf 100644 --- a/packages/contracts/src/usage.ts +++ b/packages/contracts/src/usage.ts @@ -133,6 +133,14 @@ export const UsageModelFilter = Schema.Struct({ }); export type UsageModelFilter = typeof UsageModelFilter.Type; +/** Narrows usage to one conversation, such as a thread: its provider sessions, from when it began. */ +export const UsageSessionFilter = Schema.Struct({ + sessionIds: Schema.Array(TrimmedNonEmptyString), + /** A forked session repeats its source's earlier records; this instant leaves them out. */ + sinceTime: Schema.optional(TrimmedNonEmptyString), +}); +export type UsageSessionFilter = typeof UsageSessionFilter.Type; + export const UsageSummaryInput = Schema.Struct({ sinceDay: UsageDay, untilDay: UsageDay, @@ -145,6 +153,7 @@ 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), + sessionFilter: Schema.optional(UsageSessionFilter), }); export type UsageSummaryInput = typeof UsageSummaryInput.Type; From d94d88c973c4e28f48df2212ce17daeff2d583f8 Mon Sep 17 00:00:00 2001 From: Jakub Chledowski Date: Wed, 11 Nov 2026 13:46:38 +0100 Subject: [PATCH 2/3] fix(usage): resolve a thread's sessions on the server The client built the session list from the paged chat projection, which drops idle provider threads once their items leave the window, so a long thread could lose part of its cost. The summary request now carries the thread id and the server reads every provider thread the thread holds. A turn ending while a summary was still scanning joined that scan and kept its stale answer. The client now follows such a read with one more. The row also keeps its last answer while a changed query, such as a new day's, has none yet. Co-Authored-By: Claude Fable 5.2 --- .../src/environment/ServerEnvironment.ts | 2 +- apps/server/src/usage/UsageService.test.ts | 34 ++++++--- apps/server/src/usage/UsageService.ts | 76 ++++++++++--------- apps/server/src/usage/usageAggregation.ts | 10 ++- apps/server/src/ws.ts | 18 ++++- .../components/chat/ThreadDetailsPanel.tsx | 9 ++- .../components/chat/ThreadDetailsUsage.tsx | 2 +- apps/web/src/state/usage.ts | 45 ++++++----- docs/user/usage.md | 5 +- .../client-runtime/src/state/usage.test.ts | 65 +++++++++------- packages/client-runtime/src/state/usage.ts | 36 ++++++--- packages/contracts/src/environment.ts | 4 +- packages/contracts/src/usage.ts | 18 ++--- 13 files changed, 199 insertions(+), 125 deletions(-) diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 12c9cce697122..99e58679fd55d 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -255,7 +255,7 @@ export const make = Effect.gen(function* () { modelCatalogRefresh: true, usagePriceOverrides: true, usageModelAliases: true, - usageSessionFilter: true, + usageThreadFilter: true, threadPinning: true, threadPinReorder: true, threadActiveReorder: true, diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 8a1b578ffe4b7..d9e7c3b8942a7 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -23,6 +23,7 @@ import { 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"; @@ -2824,7 +2825,7 @@ describe("UsageService", () => { }), ); - it.effect("narrows a summary to one conversation's sessions from when it began", () => + 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. @@ -2836,24 +2837,33 @@ describe("UsageService", () => { claudeLine(3, 7, "orbit-1", "fork").replace("T10:00:00Z", "T12:00:00Z"), ), ); - const forkFilter = { sessionIds: ["fork", "unknown"], sinceTime: "2026-08-01T11:00:00Z" }; + // The thread was forked at 11:00 and has started no conversation on its second provider. + const forkSessions = UsageService.threadUsageSessions({ + thread: { createdAt: DateTime.makeUnsafe("2026-08-01T11:00:00Z") }, + providerThreads: [ + { + nativeThreadRef: { + driver: ProviderDriverKind.make("claudeAgent"), + nativeId: "fork", + strength: "strong", + }, + }, + { nativeThreadRef: null }, + ], + }); const outputTokens = (summary: UsageSummary) => summary.buckets.reduce((sum, bucket) => sum + bucket.totals.outputTokens, 0); yield* Effect.gen(function* () { const service = yield* UsageService.make; // Same service, concurrent reads: the in-flight key must keep these scans apart. - const [all, source, fork, invalid] = yield* Effect.all( + const [all, source, fork, none] = yield* Effect.all( [ service.readSummary(WINDOW), - service.readSummary({ ...WINDOW, sessionFilter: { sessionIds: ["source"] } }), - service.readSummary({ ...WINDOW, sessionFilter: forkFilter }), - service - .readSummary({ - ...WINDOW, - sessionFilter: { sessionIds: ["fork"], sinceTime: "yesterday" }, - }) - .pipe(Effect.flip), + service.readSummary(WINDOW, { ids: new Set(["source"]), sinceTimeMs: 0 }), + service.readSummary(WINDOW, forkSessions), + // A thread with no conversation yet has no usage, not everyone's. + service.readSummary(WINDOW, { ids: new Set(), sinceTimeMs: 0 }), ], { concurrency: "unbounded" }, ); @@ -2862,7 +2872,7 @@ describe("UsageService", () => { assert.strictEqual(outputTokens(source), 3); assert.strictEqual(outputTokens(fork), 7); assert.strictEqual(fork.sources[0]?.distinctSessions, 1); - assert.strictEqual(invalid.reason, "invalidWindow"); + assert.strictEqual(outputTokens(none), 0); }).pipe( Effect.scoped, Effect.provide( diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index 10e8fa86f60fc..ec922e6b7ef53 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, @@ -39,7 +41,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,10 +158,29 @@ const decodeCachedSources = Schema.decodeUnknownOption( }), ); +/** A summary request once its thread, if any, is resolved to sessions. */ +export type UsageScanInput = Omit; + +/** Where a thread's usage is recorded: the native conversations of its provider threads. */ +export 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, { - readonly readSummary: (input: UsageSummaryInput) => Effect.Effect; + readonly readSummary: ( + input: UsageScanInput, + sessions?: UsageSessions, + ) => Effect.Effect; readonly refreshRates: Effect.Effect; } >()("t3/usage/UsageService") {} @@ -819,9 +845,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({ @@ -854,18 +881,6 @@ export const make = Effect.gen(function* () { hourlyWindow = { sinceTimeMs, untilTimeMs }; } - const sessionFilter = input.sessionFilter; - const sessionsSince = - sessionFilter?.sinceTime === undefined - ? Option.none() - : DateTime.make(sessionFilter.sinceTime); - if (sessionFilter?.sinceTime !== undefined && Option.isNone(sessionsSince)) { - return yield* new UsageReadError({ - reason: "invalidWindow", - detail: "Session usage requires a valid sinceTime instant", - }); - } - const startedAtMs = yield* Clock.currentTimeMillis; yield* ensureScanCacheLoaded; @@ -1011,17 +1026,7 @@ export const make = Effect.gen(function* () { priceOverrides, modelAliases, ...(input.modelFilter === undefined ? {} : { modelFilter: input.modelFilter }), - ...(sessionFilter === undefined - ? {} - : { - sessions: { - ids: new Set(sessionFilter.sessionIds), - sinceTimeMs: Option.match(sessionsSince, { - onNone: () => 0, - onSome: DateTime.toEpochMillis, - }), - }, - }), + ...(sessions === undefined ? {} : { sessions }), }); const windowEndMs = @@ -1190,23 +1195,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, @@ -1219,18 +1226,19 @@ export const make = Effect.gen(function* () { input.awaitRefresh === true, provisional, input.modelFilter ?? null, - input.sessionFilter ?? null, + sessions === undefined ? null : [[...sessions.ids].sort(), sessions.sinceTimeMs], ]); const readSummary = Effect.fn("UsageService.readSummary")(function* ( - requested: UsageSummaryInput, + requested: UsageScanInput, + sessions?: UsageSessions, ) { const { modelFilter: requestedFilter, ...unfiltered } = requested; 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); @@ -1241,7 +1249,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.ts b/apps/server/src/usage/usageAggregation.ts index 163aa3f8f42d8..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,8 +78,7 @@ export interface AggregateOptions { readonly priceOverrides?: RateTable; readonly modelAliases?: ReadonlyMap; readonly modelFilter?: UsageModelFilter; - /** Keeps only these sessions' records at or after `sinceTimeMs`. */ - readonly sessions?: { readonly ids: ReadonlySet; readonly sinceTimeMs: number }; + readonly sessions?: UsageSessions; readonly resolution?: UsageResolution; readonly sinceTimeMs?: number; readonly untilTimeMs?: number; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 09c6d86c050fd..a41bf375bfe20 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -98,6 +98,7 @@ import { ThreadId, type TerminalMetadataStreamEvent, type PullRequestRef, + UsageReadError, WS_METHODS, WsRpcGroup, } from "@t3tools/contracts"; @@ -2688,7 +2689,22 @@ export const layerWsRpc = ( processResourceMonitor.readHistory(input), [WS_METHODS.serverGetResourceTelemetryHistory]: (input) => resourceTelemetry.readHistory(input), - [WS_METHODS.serverGetUsageSummary]: (input) => usage.readSummary(input), + [WS_METHODS.serverGetUsageSummary]: ({ threadId, ...input }) => + threadId === undefined + ? usage.readSummary(input) + : threadManagement.getThreadRecords(threadId, ["providerThreads"]).pipe( + Effect.mapError( + (cause) => + new UsageReadError({ + reason: "scanFailed", + detail: "Could not read the thread's provider sessions", + cause, + }), + ), + Effect.flatMap((records) => + usage.readSummary(input, UsageService.threadUsageSessions(records)), + ), + ), [WS_METHODS.serverRefreshUsageRates]: (_input) => usage.refreshRates, [WS_METHODS.serverRetryResourceTelemetry]: (_input) => resourceTelemetry.retry, [WS_METHODS.serverCheckOtlpEndpoint]: (input) => otlpEndpointCheck.check(input), diff --git a/apps/web/src/components/chat/ThreadDetailsPanel.tsx b/apps/web/src/components/chat/ThreadDetailsPanel.tsx index e8e9a0ac24a03..0ffc64d11309c 100644 --- a/apps/web/src/components/chat/ThreadDetailsPanel.tsx +++ b/apps/web/src/components/chat/ThreadDetailsPanel.tsx @@ -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,12 +288,11 @@ export function ThreadDetailsPanel(props: ThreadDetailsPanelProps) { {visible.usage ? : null} - {(visible.context && props.usage.contextWindow) || - (visible.cost && !props.draftId) ? ( + {show.context || show.cost ? (
{contextWindow ? : null} - {cost ? : null} + {cost ? : null}
); diff --git a/apps/web/src/state/usage.ts b/apps/web/src/state/usage.ts index 226bf1bc9a6a1..8adfef2f8fd21 100644 --- a/apps/web/src/state/usage.ts +++ b/apps/web/src/state/usage.ts @@ -9,12 +9,15 @@ import { type UsageProviderKind, type UsageSummaryInput, } from "@t3tools/contracts"; +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"; @@ -252,15 +255,13 @@ export function useUsage( } const threadUsageAtom = Atom.family((key: string) => { - const ref = JSON.parse(key) as ScopedThreadRef; - const { environmentId } = ref; - // The settled turns the answer in hand covers. - let pricedTurns: number | null = null; + const ref = parseThreadKey(key); + const { environmentId, threadId } = ref; + const shouldRefresh = makeTurnEndRefresh(); + // Stays on screen while a changed query, such as a new day's, has no answer yet. + let answered: MergedUsage | null = null; return Atom.make((get): MergedUsage | null => { - // An older server ignores the session filter and would answer with every session's usage. if ( - get(environmentPresentations.presentationAtom(environmentId))?.serverConfig?.environment - .capabilities.usageSessionFilter !== true || !sessionHasScope( get(environmentSession.sessionStateAtom(environmentId)), AuthDiagnosticsReadScope, @@ -273,25 +274,31 @@ const threadUsageAtom = Atom.family((key: string) => { projection === undefined ? null : threadUsageSummaryInput({ - providerThreads: projection.providerThreads, + 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))), }); - // A turn's last records reach the transcript as it ends, so price the thread again then. - const settledTurns = projection.providerTurns.filter( - (turn) => turn.completedAt !== null, - ).length; - if (pricedTurns !== null && pricedTurns !== settledTurns) get.refresh(query); - pricedTurns = settledTurns; - const summary = AsyncResult.getOrElse(get(query), () => null); - if (summary === null) return null; + 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); + 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. - return merged.contractMismatches.length === 0 ? merged : null; + answered = merged.contractMismatches.length === 0 ? merged : null; + return answered; }).pipe(Atom.withLabel(`web-usage:thread:${key}`)); }); @@ -300,7 +307,5 @@ const threadUsageAtom = Atom.family((key: string) => { * server that cannot narrow usage to a thread. */ export function useThreadUsage(ref: ScopedThreadRef): MergedUsage | null { - return useAtomValue( - threadUsageAtom(JSON.stringify({ environmentId: ref.environmentId, threadId: ref.threadId })), - ); + return useAtomValue(threadUsageAtom(threadKey(ref))); } diff --git a/docs/user/usage.md b/docs/user/usage.md index 34cc39e834bcd..fe8ad7b3845b8 100644 --- a/docs/user/usage.md +++ b/docs/user/usage.md @@ -154,7 +154,10 @@ Turn on **Cost** to show what the thread's model usage so far would cost at API 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. +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, 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 diff --git a/packages/client-runtime/src/state/usage.test.ts b/packages/client-runtime/src/state/usage.test.ts index c0a41aa468b20..789d42a887ac5 100644 --- a/packages/client-runtime/src/state/usage.test.ts +++ b/packages/client-runtime/src/state/usage.test.ts @@ -1,6 +1,6 @@ import { EnvironmentId, - ProviderDriverKind, + ThreadId, UsageDay, USAGE_CONTRACT_VERSION, type UsageModelFilter, @@ -13,7 +13,12 @@ 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, threadUsageSummaryInput } from "./usage.ts"; +import { + makeTurnEndRefresh, + refreshUsage, + refreshUsageLimits, + threadUsageSummaryInput, +} from "./usage.ts"; const input = { sinceDay: UsageDay.make("2026-09-05"), @@ -209,49 +214,53 @@ describe("limits refresh cooldown", () => { }); describe("threadUsageSummaryInput", () => { - const providerThread = (nativeId: string | null) => ({ - nativeThreadRef: - nativeId === null - ? null - : { driver: ProviderDriverKind.make("codex"), nativeId, strength: "strong" as const }, - }); + 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 every conversation the thread has held, from its creation", () => { - expect( - threadUsageSummaryInput({ - providerThreads: [ - providerThread("session-b"), - providerThread(null), - providerThread("session-a"), - providerThread("session-b"), - ], - createdAt, - now, - }), - ).toMatchObject({ + 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", - sessionFilter: { - sessionIds: ["session-a", "session-b"], - sinceTime: "2026-09-03T08:15:00.000Z", - }, + threadId, }); }); - it("has nothing to ask before a provider starts a conversation", () => { + it("asks nothing of a server that would answer with every thread's usage", () => { + expect(threadUsageSummaryInput({ threadId, createdAt, capabilities: {}, now })).toBeNull(); expect( - threadUsageSummaryInput({ providerThreads: [providerThread(null)], createdAt, now }), + threadUsageSummaryInput({ threadId, createdAt, capabilities: undefined, now }), ).toBeNull(); }); it("never opens a window that starts after it ends", () => { const input = threadUsageSummaryInput({ - providerThreads: [providerThread("session-a")], + threadId, createdAt, + capabilities, now: DateTime.makeUnsafe("2026-09-01T12:00:00Z"), }); expect(input).toMatchObject({ sinceDay: input?.untilDay }); }); }); + +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 14803de353318..b87a882bef23d 100644 --- a/packages/client-runtime/src/state/usage.ts +++ b/packages/client-runtime/src/state/usage.ts @@ -1,6 +1,7 @@ import type { EnvironmentId, - OrchestrationV2ProviderThread, + ExecutionEnvironmentCapabilities, + ThreadId, UsageModelFilter, UsageSummaryInput, } from "@t3tools/contracts"; @@ -25,18 +26,16 @@ export function usageSummaryInput( } /** - * The usage query for one thread: every provider conversation it has held, from when it was - * created. Null until a provider has started one. Ask only servers with `usageSessionFilter`. + * 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 providerThreads: ReadonlyArray>; + readonly threadId: ThreadId; readonly createdAt: DateTime.Utc; + readonly capabilities: Pick | undefined; readonly now?: DateTime.Utc; }): UsageSummaryInput | null { - const sessionIds = [ - ...new Set(input.providerThreads.flatMap((thread) => thread.nativeThreadRef?.nativeId ?? [])), - ].sort(); - if (sessionIds.length === 0) return null; + if (input.capabilities?.usageThreadFilter !== true) return null; const { untilDay, timeZone } = makeWindow( 1, input.now === undefined ? undefined : DateTime.toDateUtc(input.now), @@ -47,7 +46,26 @@ export function threadUsageSummaryInput(input: { sinceDay: createdDay < untilDay ? createdDay : untilDay, untilDay, timeZone, - sessionFilter: { sessionIds, sinceTime: DateTime.formatIso(input.createdAt) }, + 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; }; } diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 60d54470a2e77..cfb905e86b405 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -131,8 +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 `sessionFilter`; older servers ignore it and return every session. */ - usageSessionFilter: 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 8e8ea2f0cfacf..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; @@ -133,14 +138,6 @@ export const UsageModelFilter = Schema.Struct({ }); export type UsageModelFilter = typeof UsageModelFilter.Type; -/** Narrows usage to one conversation, such as a thread: its provider sessions, from when it began. */ -export const UsageSessionFilter = Schema.Struct({ - sessionIds: Schema.Array(TrimmedNonEmptyString), - /** A forked session repeats its source's earlier records; this instant leaves them out. */ - sinceTime: Schema.optional(TrimmedNonEmptyString), -}); -export type UsageSessionFilter = typeof UsageSessionFilter.Type; - export const UsageSummaryInput = Schema.Struct({ sinceDay: UsageDay, untilDay: UsageDay, @@ -153,7 +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), - sessionFilter: Schema.optional(UsageSessionFilter), + /** Narrows usage to one thread: its provider sessions, from when it was created. */ + threadId: Schema.optional(ThreadId), }); export type UsageSummaryInput = typeof UsageSummaryInput.Type; From 95638af6969fdf71b9cd81de3eb1026f9aeca5f0 Mon Sep 17 00:00:00 2001 From: Jakub Chledowski Date: Wed, 11 Nov 2026 15:28:39 +0100 Subject: [PATCH 3/3] fix(usage): resolve thread sessions in the usage service Review feedback on the PR: - The RPC handler resolved a thread's sessions between two service calls. UsageService.readSummary now takes the thread id and does it, so the handler is one call again. - A viewer whose clock trails the server's asked for a window that ended before the thread's creation day. The window now always holds that day. - After a failed read the row kept showing the figure before it. It now shows none. Co-Authored-By: Claude Fable 5.2 --- apps/server/src/server.ts | 1 + apps/server/src/usage/UsageService.test.ts | 74 ++++++++++++++----- apps/server/src/usage/UsageService.ts | 30 +++++--- apps/server/src/ws.ts | 18 +---- apps/web/src/state/usage.ts | 7 +- docs/user/usage.md | 2 +- .../client-runtime/src/state/usage.test.ts | 17 +++-- packages/client-runtime/src/state/usage.ts | 6 +- 8 files changed, 97 insertions(+), 58 deletions(-) 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 d9e7c3b8942a7..a11cb08272f18 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -18,6 +18,7 @@ import { EnvironmentId, ProviderDriverKind, ProviderInstanceId, + ThreadId, UsageDay, type UsageSource, type UsageSummary, @@ -48,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"; @@ -196,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), @@ -203,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, @@ -2837,33 +2841,50 @@ describe("UsageService", () => { claudeLine(3, 7, "orbit-1", "fork").replace("T10:00:00Z", "T12:00:00Z"), ), ); - // The thread was forked at 11:00 and has started no conversation on its second provider. - const forkSessions = UsageService.threadUsageSessions({ - thread: { createdAt: DateTime.makeUnsafe("2026-08-01T11:00:00Z") }, - providerThreads: [ - { - nativeThreadRef: { - driver: ProviderDriverKind.make("claudeAgent"), - nativeId: "fork", - strength: "strong", - }, - }, - { nativeThreadRef: null }, - ], + 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, none] = yield* Effect.all( + const [all, source, fork, unstarted, missing] = yield* Effect.all( [ service.readSummary(WINDOW), - service.readSummary(WINDOW, { ids: new Set(["source"]), sinceTimeMs: 0 }), - service.readSummary(WINDOW, forkSessions), + ofThread("source"), + ofThread("fork"), // A thread with no conversation yet has no usage, not everyone's. - service.readSummary(WINDOW, { ids: new Set(), sinceTimeMs: 0 }), + ofThread("unstarted"), + ofThread("deleted").pipe(Effect.flip), ], { concurrency: "unbounded" }, ); @@ -2872,11 +2893,26 @@ describe("UsageService", () => { assert.strictEqual(outputTokens(source), 3); assert.strictEqual(outputTokens(fork), 7); assert.strictEqual(fork.sources[0]?.distinctSessions, 1); - assert.strictEqual(outputTokens(none), 0); + 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-session-filter-test", home, settings }), + layerService({ + prefix: "usage-service-thread-filter-test", + home, + settings, + projectionStore: { getThreadRecords }, + }), ), ); }).pipe(Effect.scoped), diff --git a/apps/server/src/usage/UsageService.ts b/apps/server/src/usage/UsageService.ts index ec922e6b7ef53..d0b9c5cf42ac7 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -32,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"; @@ -159,10 +160,10 @@ const decodeCachedSources = Schema.decodeUnknownOption( ); /** A summary request once its thread, if any, is resolved to sessions. */ -export type UsageScanInput = Omit; +type UsageScanInput = Omit; /** Where a thread's usage is recorded: the native conversations of its provider threads. */ -export function threadUsageSessions(records: { +function threadUsageSessions(records: { readonly thread: Pick; readonly providerThreads: ReadonlyArray>; }): UsageSessions { @@ -177,10 +178,7 @@ export function threadUsageSessions(records: { export class UsageService extends Context.Service< UsageService, { - readonly readSummary: ( - input: UsageScanInput, - sessions?: UsageSessions, - ) => Effect.Effect; + readonly readSummary: (input: UsageSummaryInput) => Effect.Effect; readonly refreshRates: Effect.Effect; } >()("t3/usage/UsageService") {} @@ -211,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(); @@ -1230,10 +1229,23 @@ export const make = Effect.gen(function* () { ]); const readSummary = Effect.fn("UsageService.readSummary")(function* ( - requested: UsageScanInput, - sessions?: UsageSessions, + 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; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index a41bf375bfe20..09c6d86c050fd 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -98,7 +98,6 @@ import { ThreadId, type TerminalMetadataStreamEvent, type PullRequestRef, - UsageReadError, WS_METHODS, WsRpcGroup, } from "@t3tools/contracts"; @@ -2689,22 +2688,7 @@ export const layerWsRpc = ( processResourceMonitor.readHistory(input), [WS_METHODS.serverGetResourceTelemetryHistory]: (input) => resourceTelemetry.readHistory(input), - [WS_METHODS.serverGetUsageSummary]: ({ threadId, ...input }) => - threadId === undefined - ? usage.readSummary(input) - : threadManagement.getThreadRecords(threadId, ["providerThreads"]).pipe( - Effect.mapError( - (cause) => - new UsageReadError({ - reason: "scanFailed", - detail: "Could not read the thread's provider sessions", - cause, - }), - ), - Effect.flatMap((records) => - usage.readSummary(input, UsageService.threadUsageSessions(records)), - ), - ), + [WS_METHODS.serverGetUsageSummary]: (input) => usage.readSummary(input), [WS_METHODS.serverRefreshUsageRates]: (_input) => usage.refreshRates, [WS_METHODS.serverRetryResourceTelemetry]: (_input) => resourceTelemetry.retry, [WS_METHODS.serverCheckOtlpEndpoint]: (input) => otlpEndpointCheck.check(input), diff --git a/apps/web/src/state/usage.ts b/apps/web/src/state/usage.ts index 8adfef2f8fd21..1fb9c82af4d7a 100644 --- a/apps/web/src/state/usage.ts +++ b/apps/web/src/state/usage.ts @@ -258,7 +258,7 @@ 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, has no answer yet. + // 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 ( @@ -293,6 +293,11 @@ const threadUsageAtom = Atom.family((key: string) => { 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); diff --git a/docs/user/usage.md b/docs/user/usage.md index fe8ad7b3845b8..f11f2db2efb71 100644 --- a/docs/user/usage.md +++ b/docs/user/usage.md @@ -156,7 +156,7 @@ Hover the row to see the cost by token type. It covers the thread's own conversa 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, when your +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 diff --git a/packages/client-runtime/src/state/usage.test.ts b/packages/client-runtime/src/state/usage.test.ts index 789d42a887ac5..a3f6c2ff99202 100644 --- a/packages/client-runtime/src/state/usage.test.ts +++ b/packages/client-runtime/src/state/usage.test.ts @@ -234,14 +234,15 @@ describe("threadUsageSummaryInput", () => { ).toBeNull(); }); - it("never opens a window that starts after it ends", () => { - const input = threadUsageSummaryInput({ - threadId, - createdAt, - capabilities, - now: DateTime.makeUnsafe("2026-09-01T12:00:00Z"), - }); - expect(input).toMatchObject({ sinceDay: input?.untilDay }); + 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" }); }); }); diff --git a/packages/client-runtime/src/state/usage.ts b/packages/client-runtime/src/state/usage.ts index b87a882bef23d..30ce002c87308 100644 --- a/packages/client-runtime/src/state/usage.ts +++ b/packages/client-runtime/src/state/usage.ts @@ -36,15 +36,15 @@ export function threadUsageSummaryInput(input: { readonly now?: DateTime.Utc; }): UsageSummaryInput | null { if (input.capabilities?.usageThreadFilter !== true) return null; - const { untilDay, timeZone } = makeWindow( + 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". - sinceDay: createdDay < untilDay ? createdDay : untilDay, - untilDay, + untilDay: createdDay < today ? today : createdDay, timeZone, threadId: input.threadId, };