Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 55 additions & 0 deletions apps/server/src/usage/UsageService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -520,6 +520,61 @@ describe("Codex usage history", () => {
});

describe("UsageService", () => {
it.live("matches priced native session records to their T3 thread", () =>
Effect.gen(function* () {
const { transcript, home, settings } = yield* setup;
yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 25)));
yield* Effect.gen(function* () {
const sql = yield* SqlClient.SqlClient;
yield* sql`INSERT INTO orchestration_v2_projection_provider_threads ${sql.insert({
provider_thread_id: "provider-thread-usage",
thread_id: "thread-usage",
provider: "claudeAgent",
driver: "claudeAgent",
status: "ready",
updated_at: "2026-08-01T10:00:00Z",
payload_json: encodeUnknownJsonString({
nativeThreadRef: { driver: "claudeAgent", nativeId: "session-1", strength: "strong" },
}),
})}`;
yield* sql`INSERT INTO orchestration_v2_projection_threads ${sql.insert({
thread_id: "thread-usage",
project_id: "project-usage",
title: "Usage",
default_provider: "claudeAgent",
provider_instance_id: "claudeAgent",
runtime_mode: "full-access",
interaction_mode: "default",
active_provider_thread_id: "provider-thread-usage",
created_at: "2026-08-01T10:00:00Z",
updated_at: "2026-08-01T10:00:00Z",
payload_json: "{}",
})}`;
const service = yield* UsageService.make;
const summary = yield* service.readSummary({ ...WINDOW, includeThreadUsage: true });
assert.deepEqual(
summary.threadUsage?.map((usage) => ({
threadId: usage.threadId,
input: usage.totals.uncachedInputTokens,
output: usage.totals.outputTokens,
})),
[{ threadId: "thread-usage", input: 10, output: 25 }],
);
assert.isAbove(summary.threadUsage?.[0]?.costUsd ?? 0, 0);
}).pipe(
Effect.provide(
layerService({
prefix: "thread-usage",
home,
settings,
ratesDocument: {
"claude-fable-5": { input_cost_per_token: 0.000003, output_cost_per_token: 0.000015 },
},
}),
),
);
}).pipe(Effect.scoped),
);
it.live("counts compressed Codex rollouts across refresh, rewrites, and cache restarts", () =>
Effect.gen(function* () {
const { home, settings } = yield* setup;
Expand Down
39 changes: 39 additions & 0 deletions apps/server/src/usage/UsageService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,8 @@ import * as Path from "effect/Path";
import * as Schema from "effect/Schema";
import * as Semaphore from "effect/Semaphore";
import { HttpClient, HttpClientResponse } from "effect/http";
import * as SqlClient from "effect/sql/SqlClient";
import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";

import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite";
import { formatMalformedUsageWarning } from "@t3tools/shared/usageFormat";
Expand Down Expand Up @@ -156,6 +158,7 @@ export const make = Effect.gen(function* () {
const httpClient = yield* HttpClient.HttpClient;
const hostEnvironment = yield* HostProcess.Environment;
const hostname = yield* HostProcess.Hostname;
const database = yield* SqlClient.SqlClient;
const readerContext = yield* Effect.context<BuiltInUsageReadersEnv>();

const fileCache: ScanCache = new Map();
Expand Down Expand Up @@ -763,7 +766,41 @@ export const make = Effect.gen(function* () {
{ concurrency: 2 },
);

let threadIdsBySession: Map<string, readonly string[]> | undefined;
if (input.includeThreadUsage === true) {
const links = yield* database<{
readonly threadId: string;
readonly driver: string;
readonly sessionId: string;
}>`
SELECT DISTINCT thread_id AS threadId, driver,
json_extract(payload_json, '$.nativeThreadRef.nativeId') AS sessionId
FROM orchestration_v2_projection_provider_threads
WHERE json_extract(payload_json, '$.nativeThreadRef.nativeId') IS NOT NULL
`.pipe(
NodeSqliteClient.readOnly,
Effect.mapError(
(cause) =>
new UsageReadError({
reason: "scanFailed",
detail: "Could not match native usage to sessions.",
cause,
}),
),
);
threadIdsBySession = new Map();
for (const link of links) {
const provider = BUILT_IN_USAGE_DRIVERS.find((driver) => driver.driverKind === link.driver)
?.usage?.provider;
if (!provider) continue;
const key = `${provider}:${link.sessionId}`;
const previous = threadIdsBySession.get(key) ?? [];
if (!previous.includes(link.threadId))
threadIdsBySession.set(key, [...previous, link.threadId]);
}
}
const aggregator = new UsageAggregator({
...(threadIdsBySession === undefined ? {} : { threadIdsBySession }),
timeZone: input.timeZone,
sinceDay: input.sinceDay,
untilDay: input.untilDay,
Expand Down Expand Up @@ -927,6 +964,7 @@ export const make = Effect.gen(function* () {
sinceDay: input.sinceDay,
untilDay: input.untilDay,
buckets: aggregated.buckets,
...(aggregated.threadUsage === undefined ? {} : { threadUsage: aggregated.threadUsage }),
sources,
pricing: pricing(),
scanDurationMs: Math.max(0, finishedAtMs - startedAtMs),
Expand All @@ -947,6 +985,7 @@ export const make = Effect.gen(function* () {
settings,
input.awaitRefresh === true,
input.modelFilter ?? null,
input.includeThreadUsage ?? false,
]);

const readSummary = Effect.fn("UsageService.readSummary")(function* (
Expand Down
52 changes: 52 additions & 0 deletions apps/server/src/usage/usageAggregation.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,58 @@ function aggregate(
}

describe("UsageAggregator", () => {
it("prices thread usage separately across models and deduplicates native session links", () => {
const aggregator = new UsageAggregator({
timeZone: "UTC",
sinceDay: "2026-08-01",
untilDay: "2026-08-31",
rates,
threadIdsBySession: new Map([
["claude:session-a", ["thread-a"]],
["claude:session-b", ["thread-b"]],
]),
});
aggregator.add(record({ dedupeKey: "a" }));
aggregator.add(record({ dedupeKey: "a" }));
aggregator.add(record({ sessionId: "session-b", model: "unknown" }));
const usage = aggregator.finish().threadUsage;
expect(usage?.find((entry) => entry.threadId === "thread-a")).toMatchObject({
totals: {
uncachedInputTokens: 100,
cachedInputTokens: 1000,
cacheCreationTokens: 10,
outputTokens: 50,
},
unpricedRecords: 0,
});
expect(usage?.find((entry) => entry.threadId === "thread-a")?.costUsd).toBeCloseTo(0.004625, 8);
expect(usage?.find((entry) => entry.threadId === "thread-b")).toMatchObject({
unpricedRecords: 1,
});
});
it("attributes a copied inference when its linked session is scanned second", () => {
const aggregator = new UsageAggregator({
timeZone: "UTC",
sinceDay: "2026-08-01",
untilDay: "2026-08-31",
rates,
threadIdsBySession: new Map([["claude:session-linked", ["thread-a"]]]),
});
aggregator.add(record({ sessionId: "session-unlinked", dedupeKey: "copied-inference" }));
aggregator.add(record({ sessionId: "session-linked", dedupeKey: "copied-inference" }));
aggregator.add(record({ sessionId: "session-linked", dedupeKey: "copied-inference" }));
const result = aggregator.finish();

expect(result.buckets).toHaveLength(1);
expect(result.buckets[0]?.records).toBe(1);
expect(result.duplicatesDropped).toBe(2);
expect(result.threadUsage).toEqual([
expect.objectContaining({
threadId: "thread-a",
totals: expect.objectContaining({ outputTokens: 50 }),
}),
]);
});
it("splits a bucket's cost by category and speed", () => {
const [bucket] = aggregate([record(), record({ speed: "fast" })]).buckets;

Expand Down
57 changes: 56 additions & 1 deletion apps/server/src/usage/usageAggregation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ import type {
UsageModelSelector,
UsageResolution,
UsageTokenTotals,
UsageThreadTotals,
} from "@t3tools/contracts";

import { EMPTY_TOTALS, type UsageRecord } from "@t3tools/provider-core/server/usage";
Expand Down Expand Up @@ -64,6 +65,7 @@ interface MutableBucket {
}

export interface AggregateOptions {
readonly threadIdsBySession?: ReadonlyMap<string, readonly string[]>;
readonly timeZone: string;
readonly sinceDay: string;
readonly untilDay: string;
Expand Down Expand Up @@ -125,6 +127,7 @@ export function canonicalModelFilter(
}

export interface AggregateResult {
readonly threadUsage?: readonly UsageThreadTotals[];
readonly buckets: readonly UsageBucket[];
readonly duplicatesDropped: number;
readonly outOfWindow: number;
Expand Down Expand Up @@ -172,8 +175,10 @@ interface Contribution {

/** Deduplicates across the whole scan; Claude copies records into resumed/forked transcripts. */
export class UsageAggregator {
readonly #threadUsage = new Map<string, UsageThreadTotals>();
readonly #buckets = new Map<string, MutableBucket>();
readonly #seen = new Map<string, Contribution>();
readonly #threadSeen = new Map<string, Set<string>>();
readonly #toDay: (timestampMs: number) => string;
readonly #hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null;
readonly #options: AggregateOptions;
Expand Down Expand Up @@ -207,6 +212,8 @@ export class UsageAggregator {
add(input: UsageRecord, sourcePath?: string): boolean {
if (!keepsUsageRecord(this.#options.modelFilter, input)) return false;
const record = this.#mapModel(input);
// Counted before the scan-wide dedupe: a copy in another thread's transcript still belongs to it.
this.#addThreadUsage(record);
if (record.dedupeKey !== null) {
const held = this.#seen.get(record.dedupeKey);
if (held !== undefined) {
Expand All @@ -228,7 +235,46 @@ export class UsageAggregator {
return true;
}

#placeInWindow(record: UsageRecord, source: string): MutableBucket | null {
#addThreadUsage(record: UsageRecord): void {
const threadIds =
this.#options.threadIdsBySession?.get(`${record.provider}:${record.sessionId}`) ?? [];
if (threadIds.length === 0 || this.#dayInWindow(record) === null) return;
const priced = priceUsage(this.#options.rates, record, this.#options.priceOverrides);
const costs = priced.categoryCostUsd;
for (const threadId of threadIds) {
if (record.dedupeKey !== null) {
const seen = this.#threadSeen.get(threadId);
if (seen?.has(record.dedupeKey)) continue;

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟠 High usage/usageAggregation.ts:247

threadUsage keeps the first copy's token totals and cost when a later copy has the same dedupeKey, so committed partial usage followed by the unterminated final record is undercounted. The continue skips the later record instead of merging it as the main buckets do; retain per-thread records by key and merge duplicate copies before pricing them.

🤖 Copy this AI Prompt to have your agent fix this:
In file @apps/server/src/usage/usageAggregation.ts around line 247:

`threadUsage` keeps the first copy's token totals and cost when a later copy has the same `dedupeKey`, so committed partial usage followed by the unterminated final record is undercounted. The `continue` skips the later record instead of merging it as the main buckets do; retain per-thread records by key and merge duplicate copies before pricing them.

if (seen) seen.add(record.dedupeKey);
else this.#threadSeen.set(threadId, new Set([record.dedupeKey]));
}
const previous = this.#threadUsage.get(threadId);
const add = <K extends keyof UsageTokenTotals>(key: K) =>
(previous?.totals[key] ?? 0) + record.totals[key];
this.#threadUsage.set(threadId, {
threadId,
totals: {
uncachedInputTokens: add("uncachedInputTokens"),
cachedInputTokens: add("cachedInputTokens"),
cacheCreationTokens: add("cacheCreationTokens"),
outputTokens: add("outputTokens"),
reasoningTokens: add("reasoningTokens"),
},
costUsd: (previous?.costUsd ?? 0) + priced.costUsd,
categoryCostUsd: {
input: (previous?.categoryCostUsd.input ?? 0) + (costs?.input ?? 0),
cacheRead: (previous?.categoryCostUsd.cacheRead ?? 0) + (costs?.cacheRead ?? 0),
cacheWrite: (previous?.categoryCostUsd.cacheWrite ?? 0) + (costs?.cacheWrite ?? 0),
output: (previous?.categoryCostUsd.output ?? 0) + (costs?.output ?? 0),
},
unsplitCostUsd: (previous?.unsplitCostUsd ?? 0) + (costs === null ? priced.costUsd : 0),
unpricedRecords:
(previous?.unpricedRecords ?? 0) + (priced.costSource === "unpriced" ? 1 : 0),
});
}
}

#dayInWindow(record: UsageRecord): string | null {
if (
this.#hourlyWindow !== null &&
(record.timestampMs < this.#hourlyWindow.sinceTimeMs ||
Expand All @@ -244,6 +290,12 @@ export class UsageAggregator {
) {
return null;
}
return day;
}

#placeInWindow(record: UsageRecord, source: string): MutableBucket | null {
const day = this.#dayInWindow(record);
if (day === null) return null;

const hourIndex =
this.#hourlyWindow === null
Expand Down Expand Up @@ -400,6 +452,9 @@ export class UsageAggregator {

return {
buckets,
...(this.#options.threadIdsBySession === undefined
? {}
: { threadUsage: [...this.#threadUsage.values()] }),
duplicatesDropped: this.#duplicatesDropped,
outOfWindow: this.#outOfWindow,
};
Expand Down
Loading
Loading