diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index 53cb64334ad2c..db330912b7de9 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -102,6 +102,15 @@ describe("RPC authorization scopes", () => { ); }); + it("requires diagnostics read for the usage cohort report, like the usage summary", () => { + expect(requiredScopeForRpcMethod(WS_METHODS.serverGetUsageCohortReport)).toBe( + requiredScopeForRpcMethod(WS_METHODS.serverGetUsageSummary), + ); + expect(requiredScopeForRpcMethod(WS_METHODS.serverGetUsageCohortReport)).toBe( + AuthDiagnosticsReadScope, + ); + }); + it("requires thread read permission for file history, which joins files to threads", () => { expect(requiredScopeForRpcMethod(WS_METHODS.projectCodeFileHistory)).toBe( AuthOrchestrationReadScope, diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 7b8e1a5a96453..60b3ca1bc36d5 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -242,6 +242,7 @@ export const make = Effect.gen(function* () { projectCodeFileHistory: true, projectCodeIndexEvents: true, requiredWorktreeBootstrap: true, + usageCohortReports: 1, threadSettlement: true, threadAutoSettlement: true, storageCleanup: true, diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index cf3cd017aa61b..ddf858ad2a474 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -108,6 +108,7 @@ const RPC_AGGREGATES = { [WS_METHODS.serverRefreshSkillsLibrary]: "server", [WS_METHODS.serverPrepareTriage]: "server", [WS_METHODS.serverGetUsageSummary]: "server", + [WS_METHODS.serverGetUsageCohortReport]: "server", [WS_METHODS.serverRefreshUsageRates]: "server", [WS_METHODS.serverSignalProcess]: "server", [WS_METHODS.serverReportClientActivity]: "server", diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 483644d16353f..eb8aab152732a 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -38,6 +38,7 @@ import { OrchestrationV2DomainEvent, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as SqlClient from "effect/sql/SqlClient"; import * as Deferred from "effect/Deferred"; @@ -69,6 +70,7 @@ import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderTurnStart from "./ProviderTurnStartService.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as ProviderUsageAnalytics from "../telemetry/ProviderUsageAnalytics.ts"; +import * as UsageAttemptStamps from "../usage/UsageAttemptStamps.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; const isDomainEvent = Schema.is(OrchestrationV2DomainEvent); @@ -310,6 +312,7 @@ function makeLocalCommandHarness(input: { readonly interruptRunBeforeOpenFailure?: boolean; readonly writeFailure?: unknown; readonly failReadsAfterRunning?: boolean; + readonly attemptStamps?: Context.Service.Shape; readonly previousTurn?: { readonly runStatus: "failed" | "completed"; readonly turnStatus: "running" | "completed" | "interrupted"; @@ -1138,6 +1141,9 @@ function makeLocalCommandHarness(input: { turnCompleted: () => Effect.void, runSettled: (report) => Effect.sync(() => void settledRuns.push(report.status)), }), + input.attemptStamps === undefined + ? Layer.empty + : Layer.succeed(UsageAttemptStamps.UsageAttemptStamps, input.attemptStamps), Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: resolveRuntimePolicy, }), @@ -1758,6 +1764,56 @@ effectIt.effect("starts another provider in the same checkout while a turn is pe }), ); +effectIt.effect("stamps the attempt once, before it commits to running", () => + Effect.gen(function* () { + const stamp = vi.fn((_: UsageAttemptStamps.UsageAttemptStampInput) => Effect.void); + const harness = makeLocalCommandHarness({ + text: "Continue", + failReadsAfterRunning: true, + attemptStamps: { stamp }, + }); + harness.writeIfRunCurrent.mockImplementationOnce((write) => { + expect(stamp).toHaveBeenCalledOnce(); + return harness.writeIfRunCurrent.getMockImplementation()!(write); + }); + yield* harness.start; + yield* harness.start; + + expect(stamp).toHaveBeenCalledOnce(); + expect(stamp).toHaveBeenCalledWith( + expect.objectContaining({ + attemptId: harness.attemptId, + threadId: harness.projection().thread.id, + projectId: harness.projection().thread.projectId, + message: expect.objectContaining({ createdBy: "user" }), + }), + ); + }), +); + +effectIt.effect("starts the turn when the stamp cannot be written", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`DROP TABLE usage_attempt_stamps`; + const harness = makeLocalCommandHarness({ + text: "Continue", + failReadsAfterRunning: true, + attemptStamps: yield* UsageAttemptStamps.UsageAttemptStamps, + }); + yield* harness.start; + + expect(harness.startRootRun).toHaveBeenCalledOnce(); + expect(harness.projection().runs.at(-1)?.status).toBe("running"); + }).pipe( + Effect.provide( + UsageAttemptStamps.layer.pipe( + Layer.provideMerge(SqlitePersistence.layerMemory), + Layer.provideMerge(ServerSettings.layerTest()), + ), + ), + ), +); + effectIt.effect("terminalizes a starting run when its provider session cannot open", () => Effect.gen(function* () { const harness = makeLocalCommandHarness({ diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 954cfce326e53..4b75bb6bb52c3 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -39,6 +39,7 @@ import * as ProviderAuthService from "../provider/ProviderAuthService.ts"; import * as ServerSettings from "../serverSettings.ts"; import { makeIsManagedWorktreePath } from "../worktreesDirectory.ts"; import * as EventSink from "./EventSink.ts"; +import * as UsageAttemptStamps from "../usage/UsageAttemptStamps.ts"; import * as ProviderUsageAnalytics from "../telemetry/ProviderUsageAnalytics.ts"; import * as ContextHandoffService from "./ContextHandoffService.ts"; import { isActiveRun } from "./CommandPolicy.ts"; @@ -161,6 +162,7 @@ export const layer: Layer.Layer< const pstack = yield* PstackPack.PstackPack; const eventSink = yield* EventSink.EventSinkV2; const turnAnalytics = yield* ProviderUsageAnalytics.ProviderTurnAnalytics; + const attemptStamps = yield* UsageAttemptStamps.UsageAttemptStamps; const contextHandoffService = yield* ContextHandoffService.ContextHandoffServiceV2; const idAllocator = yield* IdAllocator.IdAllocatorV2; const fileSystem = yield* FileSystem.FileSystem; @@ -1545,6 +1547,13 @@ export const layer: Layer.Layer< payload: runningRootNode, }, ]; + // Retries return at the status gate once this commits, so a stamp written after it could be lost for good. + yield* attemptStamps.stamp({ + attemptId: attempt.id, + threadId: projection.thread.id, + projectId: projection.thread.projectId, + message, + }); const runningWrite = yield* eventSink.writeIfRunCurrent({ threadId: projection.thread.id, runId: run.id, diff --git a/apps/server/src/persistence/Migrations.test.ts b/apps/server/src/persistence/Migrations.test.ts index 3c5055707d951..f92489fdca6ab 100644 --- a/apps/server/src/persistence/Migrations.test.ts +++ b/apps/server/src/persistence/Migrations.test.ts @@ -21,6 +21,7 @@ const settledOverrideMigration = [86, "RepairBooleanSettledOverride"] as const; const correlationIndexMigration = [87, "DropOrchestrationEventCorrelationIndex"] as const; const modelEvaluationMigration = [88, "ModelEvaluations"] as const; const delegatedRecoveryMigration = [89, "DelegatedRecoveryIndexes"] as const; +const usageAttemptStampsMigration = [90, "UsageAttemptStamps"] as const; it.effect("upgrades attempt receipts atomically and preserves distinct steering messages", () => Effect.gen(function* () { @@ -45,6 +46,7 @@ it.effect("upgrades attempt receipts atomically and preserves distinct steering correlationIndexMigration, modelEvaluationMigration, delegatedRecoveryMigration, + usageAttemptStampsMigration, ]); assert.deepStrictEqual(yield* sql`SELECT * FROM project_knowledge_deliveries`, before); yield* sql`INSERT INTO project_knowledge_deliveries ( @@ -290,6 +292,7 @@ describe("runMigrations", () => { correlationIndexMigration, modelEvaluationMigration, delegatedRecoveryMigration, + usageAttemptStampsMigration, ]); assert.deepStrictEqual((yield* history).slice(0, before.length), before); const indexes = yield* sql` @@ -364,6 +367,7 @@ describe("runMigrations", () => { correlationIndexMigration, modelEvaluationMigration, delegatedRecoveryMigration, + usageAttemptStampsMigration, ], ); yield* assertConsolidatedSchema; @@ -404,6 +408,7 @@ describe("runMigrations", () => { correlationIndexMigration, modelEvaluationMigration, delegatedRecoveryMigration, + usageAttemptStampsMigration, ], ); assert.deepStrictEqual(yield* runMigrations(), [ @@ -417,6 +422,7 @@ describe("runMigrations", () => { correlationIndexMigration, modelEvaluationMigration, delegatedRecoveryMigration, + usageAttemptStampsMigration, ]); yield* assertConsolidatedSchema; assert.deepStrictEqual( @@ -452,6 +458,7 @@ describe("runMigrations", () => { correlationIndexMigration, modelEvaluationMigration, delegatedRecoveryMigration, + usageAttemptStampsMigration, ]); yield* assertConsolidatedSchema; @@ -557,6 +564,7 @@ describe("runMigrations", () => { correlationIndexMigration, modelEvaluationMigration, delegatedRecoveryMigration, + usageAttemptStampsMigration, ]); yield* assertConsolidatedSchema; assert.deepStrictEqual(yield* sql`SELECT secret_name, fenced FROM peer_environment_links`, [ @@ -611,6 +619,7 @@ describe("runMigrations", () => { correlationIndexMigration, modelEvaluationMigration, delegatedRecoveryMigration, + usageAttemptStampsMigration, ]); assert.deepStrictEqual(yield* history, migrationManifest); assert.deepStrictEqual( @@ -670,6 +679,7 @@ describe("runMigrations", () => { correlationIndexMigration, modelEvaluationMigration, delegatedRecoveryMigration, + usageAttemptStampsMigration, ]); assert.deepStrictEqual(yield* history, migrationManifest); assert.deepStrictEqual(yield* sql`SELECT * FROM cross_environment_launches`, routes); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 0a4aaa299689b..69dcb366aa419 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -94,6 +94,7 @@ import Migration0086 from "./Migrations/086_RepairBooleanSettledOverride.ts"; import Migration0087 from "./Migrations/087_DropOrchestrationEventCorrelationIndex.ts"; import Migration0088 from "./Migrations/088_ModelEvaluations.ts"; import Migration0089 from "./Migrations/089_DelegatedRecoveryIndexes.ts"; +import Migration0090 from "./Migrations/090_UsageAttemptStamps.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -186,6 +187,7 @@ const migrationEntries = [ [87, "DropOrchestrationEventCorrelationIndex", Migration0087], [88, "ModelEvaluations", Migration0088], [89, "DelegatedRecoveryIndexes", Migration0089], + [90, "UsageAttemptStamps", Migration0090], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/090_UsageAttemptStamps.ts b/apps/server/src/persistence/Migrations/090_UsageAttemptStamps.ts new file mode 100644 index 0000000000000..e654c972b5712 --- /dev/null +++ b/apps/server/src/persistence/Migrations/090_UsageAttemptStamps.ts @@ -0,0 +1,20 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql` + CREATE TABLE usage_attempt_stamps ( + attempt_id TEXT PRIMARY KEY NOT NULL, + thread_id TEXT NOT NULL, + origin TEXT NOT NULL, + t3_version TEXT NOT NULL, + settings_json TEXT NOT NULL + ) WITHOUT ROWID + `; + yield* sql` + CREATE INDEX orchestration_v2_projection_run_attempts_started_at_idx + ON orchestration_v2_projection_run_attempts (json_extract(payload_json, '$.startedAt')) + WHERE json_extract(payload_json, '$.startedAt') IS NOT NULL + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index f491d857192c6..cc52ad53fad70 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -69,6 +69,7 @@ describe("V2 preview upgrade", () => { [87, "DropOrchestrationEventCorrelationIndex"], [88, "ModelEvaluations"], [89, "DelegatedRecoveryIndexes"], + [90, "UsageAttemptStamps"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -181,6 +182,7 @@ describe("V2 preview upgrade", () => { [87, "DropOrchestrationEventCorrelationIndex"], [88, "ModelEvaluations"], [89, "DelegatedRecoveryIndexes"], + [90, "UsageAttemptStamps"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 4fd938d2c0398..ff704e15cb5d3 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -206,6 +206,8 @@ import { MuseUsageHome } from "@t3tools/provider-muse/server"; import * as CursorAccountReader from "@t3tools/provider-cursor/server/CursorAccountReader"; import * as CursorKeychain from "@t3tools/provider-cursor/server/CursorKeychain"; import * as CursorUsageAccounts from "@t3tools/provider-cursor/server/CursorUsageAccounts"; +import * as UsageAttemptStamps from "./usage/UsageAttemptStamps.ts"; +import * as UsageCohortReports from "./usage/UsageCohortReports.ts"; import * as UsageService from "./usage/UsageService.ts"; import * as CodexUsageHistory from "./usage/CodexUsageHistory.ts"; import * as RuntimeLayer from "./orchestration-v2/runtimeLayer.ts"; @@ -585,6 +587,7 @@ const layerScheduledTaskWebhookOrigin = Layer.effect( const layerOrchestrationV2Runtime = RuntimeLayer.layerProduction.pipe( Layer.provide(layerScheduledTaskWebhookOrigin), Layer.provide(ProviderUsageAnalytics.layer), + Layer.provide(UsageAttemptStamps.layer), Layer.provide(layerCheckpointStore), Layer.provide(layerGitWorkflow), Layer.provide( @@ -787,6 +790,7 @@ const layerCodeIntelligence = Layer.mergeAll(CodeIntelligence.layer, FileHistory const layerRuntimeDependencies = layerRuntimeCoreDependencies.pipe( Layer.provideMerge(layerCodeIntelligence), + Layer.provideMerge(UsageCohortReports.layer), Layer.provideMerge(layerUsage), Layer.provideMerge(layerBackground), Layer.provideMerge(layerResourceDiagnostics), diff --git a/apps/server/src/usage/UsageAttemptStamps.test.ts b/apps/server/src/usage/UsageAttemptStamps.test.ts new file mode 100644 index 0000000000000..fa4981cc4dca1 --- /dev/null +++ b/apps/server/src/usage/UsageAttemptStamps.test.ts @@ -0,0 +1,123 @@ +import { assert, it } from "@effect/vitest"; +import { ProjectId, RunAttemptId, ScheduledTaskId, ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as UsageAttemptStamps from "./UsageAttemptStamps.ts"; + +const projectId = ProjectId.make("project-stamps"); +const attemptId = RunAttemptId.make("attempt-stamps"); +const threadId = ThreadId.make("thread-stamps"); +const input = { + attemptId, + threadId, + projectId, + message: { createdBy: "user", creationSource: "web" }, +} as const satisfies UsageAttemptStamps.UsageAttemptStampInput; + +const layer = UsageAttemptStamps.layer.pipe( + Layer.provideMerge(SqlitePersistence.layerMemory), + Layer.provideMerge( + ServerSettings.layerTest({ + enableModelEvaluations: true, + projectSettingsOverrides: { [projectId]: { enableAssistedLearning: true } }, + }), + ), +); + +const rows = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + return yield* sql<{ + readonly origin: string; + readonly settings_json: string; + }>`SELECT origin, settings_json FROM usage_attempt_stamps WHERE attempt_id = ${attemptId}`; +}); + +it.effect("overwrites the stamp when the same attempt is stamped again", () => + Effect.gen(function* () { + const stamps = yield* UsageAttemptStamps.UsageAttemptStamps; + yield* stamps.stamp(input); + yield* stamps.stamp({ + ...input, + message: { + createdBy: "user", + creationSource: "web", + scheduledTaskId: ScheduledTaskId.make("task-1"), + }, + }); + + const stored = yield* rows; + assert.deepStrictEqual( + stored.map((row) => row.origin), + ["scheduled"], + ); + }).pipe(Effect.provide(layer)), +); + +it.effect("stores only booleans, resolved for the project", () => + Effect.gen(function* () { + yield* (yield* UsageAttemptStamps.UsageAttemptStamps).stamp(input); + + const [row] = yield* rows; + const settings = JSON.parse(row!.settings_json) as Record; + assert.isTrue(Object.values(settings).every((value) => typeof value === "boolean")); + assert.strictEqual(settings.enableModelEvaluations, true); + assert.strictEqual(settings.enableAssistedLearning, true); + assert.strictEqual(settings.pstackEnabled, false); + assert.notProperty(settings, "textGenerationModelSelection"); + }).pipe(Effect.provide(layer)), +); + +it.effect("does not fail the caller when the write fails", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`DROP TABLE usage_attempt_stamps`; + + yield* (yield* UsageAttemptStamps.UsageAttemptStamps).stamp(input); + }).pipe(Effect.provide(layer)), +); + +it.effect("does not fail the caller when reading settings dies", () => + Effect.gen(function* () { + yield* (yield* UsageAttemptStamps.UsageAttemptStamps).stamp(input); + + assert.deepStrictEqual(yield* rows, []); + }).pipe( + Effect.provide( + UsageAttemptStamps.layer.pipe( + Layer.provideMerge(SqlitePersistence.layerMemory), + Layer.provideMerge( + Layer.effect( + ServerSettings.ServerSettingsService, + Effect.gen(function* () { + const settings = yield* ServerSettings.ServerSettingsService; + return { ...settings, getSettings: Effect.die("settings blew up") }; + }), + ).pipe(Layer.provide(ServerSettings.layerTest({}))), + ), + ), + ), + ), +); + +it("derives the origin a comparison uses", () => { + const origin = UsageAttemptStamps.originOf; + assert.strictEqual(origin({ createdBy: "user", creationSource: "web" }), "user"); + assert.strictEqual(origin({ createdBy: "agent", creationSource: "mcp" }), "agent"); + assert.strictEqual(origin({ createdBy: "agent", creationSource: "server" }), "server_dispatched"); + assert.strictEqual( + origin({ createdBy: "system", creationSource: "provider" }), + "server_dispatched", + ); + assert.strictEqual( + origin({ + createdBy: "user", + creationSource: "server", + scheduledTaskId: ScheduledTaskId.make("t"), + }), + "scheduled", + ); +}); diff --git a/apps/server/src/usage/UsageAttemptStamps.ts b/apps/server/src/usage/UsageAttemptStamps.ts new file mode 100644 index 0000000000000..70ab55dc3072f --- /dev/null +++ b/apps/server/src/usage/UsageAttemptStamps.ts @@ -0,0 +1,164 @@ +import type { + OrchestrationV2ConversationMessage, + ProjectId, + RunAttemptId, + ThreadId, +} from "@t3tools/contracts"; +import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/sql/SqlClient"; + +import packageJson from "../../package.json" with { type: "json" }; +import * as ServerSettings from "../serverSettings.ts"; +import { UsageOrigin, type AttemptRow } from "./usageCohortJoin.ts"; + +export interface UsageAttemptStampInput { + readonly attemptId: RunAttemptId; + readonly threadId: ThreadId; + readonly projectId: ProjectId; + readonly message: Pick< + OrchestrationV2ConversationMessage, + "createdBy" | "creationSource" | "scheduledTaskId" + >; +} + +const encodeSettings = Schema.encodeEffect( + Schema.fromJsonString(Schema.Record(Schema.String, Schema.Boolean)), +); + +/** Observational, so the default does nothing and turn start tests do not need a layer. */ +export class UsageAttemptStamps extends Context.Reference<{ + readonly stamp: (input: UsageAttemptStampInput) => Effect.Effect; +}>("t3/usage/UsageAttemptStamps", { + defaultValue: () => ({ stamp: () => Effect.void }), +}) {} + +export function originOf(message: UsageAttemptStampInput["message"]): UsageOrigin { + if (message.scheduledTaskId !== undefined) return "scheduled"; + if (message.createdBy === "system" || message.creationSource === "server") { + return "server_dispatched"; + } + return message.createdBy; +} + +export const layer = Layer.effect( + UsageAttemptStamps, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const serverSettings = yield* ServerSettings.ServerSettingsService; + + const stamp = Effect.fn("UsageAttemptStamps.stamp")( + function* (input: UsageAttemptStampInput) { + const { settings } = resolveProjectSettings( + yield* serverSettings.getSettings, + input.projectId, + ); + const booleans: Record = {}; + for (const [key, value] of Object.entries(settings)) { + if (typeof value === "boolean") booleans[key] = value; + } + const settingsJson = yield* encodeSettings(booleans); + yield* sql` + INSERT INTO usage_attempt_stamps (attempt_id, thread_id, origin, t3_version, settings_json) + VALUES (${input.attemptId}, ${input.threadId}, ${originOf(input.message)}, ${packageJson.version}, ${settingsJson}) + ON CONFLICT (attempt_id) DO UPDATE SET + thread_id = excluded.thread_id, + origin = excluded.origin, + t3_version = excluded.t3_version, + settings_json = excluded.settings_json + `; + }, + (effect, input) => + effect.pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logWarning("usage.attempt-stamp.failed", { + attemptId: input.attemptId, + cause: Cause.pretty(cause), + }), + ), + ), + ); + + return { stamp }; + }), +); + +const AttemptSqlRow = Schema.Struct({ + attempt_id: Schema.String, + run_id: Schema.String, + thread_id: Schema.String, + provider_instance_id: Schema.String, + session_id: Schema.NullOr(Schema.String), + started_at: Schema.String, + completed_at: Schema.NullOr(Schema.String), + origin: Schema.NullOr(UsageOrigin), + t3_version: Schema.NullOr(Schema.String), + settings_json: Schema.NullOr(Schema.fromJsonString(Schema.Record(Schema.String, Schema.Boolean))), +}); +const decodeAttemptRows = Schema.decodeUnknownEffect(Schema.Array(AttemptSqlRow)); + +const FirstSeenSqlRow = Schema.Struct({ t3_version: Schema.String, first_seen: Schema.String }); +const decodeFirstSeenRows = Schema.decodeUnknownEffect(Schema.Array(FirstSeenSqlRow)); + +const toMs = (iso: string) => DateTime.toEpochMillis(DateTime.makeUnsafe(iso)); + +/** Attempts that started in the range, with their stamp when they have one. */ +export const readAttempts = Effect.fn("UsageAttemptStamps.readAttempts")(function* ( + sql: SqlClient.SqlClient, + range: { readonly sinceMs: number; readonly untilMs: number }, +) { + const sinceIso = DateTime.formatIso(DateTime.makeUnsafe(range.sinceMs)); + const untilIso = DateTime.formatIso(DateTime.makeUnsafe(range.untilMs)); + const rows = yield* sql` + SELECT + a.attempt_id AS attempt_id, + a.run_id AS run_id, + a.thread_id AS thread_id, + COALESCE(a.provider_instance_id, a.provider) AS provider_instance_id, + json_extract(a.payload_json, '$.nativeThreadId') AS session_id, + json_extract(a.payload_json, '$.startedAt') AS started_at, + json_extract(a.payload_json, '$.completedAt') AS completed_at, + s.origin AS origin, + s.t3_version AS t3_version, + s.settings_json AS settings_json + FROM orchestration_v2_projection_run_attempts a + LEFT JOIN usage_attempt_stamps s ON s.attempt_id = a.attempt_id + WHERE json_extract(a.payload_json, '$.startedAt') IS NOT NULL + AND json_extract(a.payload_json, '$.startedAt') >= ${sinceIso} + AND json_extract(a.payload_json, '$.startedAt') <= ${untilIso} + `.pipe(Effect.flatMap(decodeAttemptRows)); + return rows.map((row): AttemptRow => ({ + attemptId: row.attempt_id, + runId: row.run_id, + threadId: row.thread_id, + sessionId: row.session_id ?? "", + providerInstanceId: row.provider_instance_id, + startedAtMs: toMs(row.started_at), + endedAtMs: row.completed_at === null ? null : toMs(row.completed_at), + stamp: + row.origin === null || row.t3_version === null || row.settings_json === null + ? null + : { origin: row.origin, t3Version: row.t3_version, settings: row.settings_json }, + })); +}); + +/** When this environment first ran each build, over every stamp it holds. */ +export const readBuildFirstSeen = Effect.fn("UsageAttemptStamps.readBuildFirstSeen")(function* ( + sql: SqlClient.SqlClient, +) { + const rows = yield* sql` + SELECT s.t3_version AS t3_version, MIN(json_extract(a.payload_json, '$.startedAt')) AS first_seen + FROM usage_attempt_stamps s + JOIN orchestration_v2_projection_run_attempts a ON a.attempt_id = s.attempt_id + WHERE json_extract(a.payload_json, '$.startedAt') IS NOT NULL + GROUP BY s.t3_version + `.pipe(Effect.flatMap(decodeFirstSeenRows)); + return new Map(rows.map((row) => [row.t3_version, toMs(row.first_seen)] as const)); +}); diff --git a/apps/server/src/usage/UsageCohortReports.test.ts b/apps/server/src/usage/UsageCohortReports.test.ts new file mode 100644 index 0000000000000..f7edfefc34253 --- /dev/null +++ b/apps/server/src/usage/UsageCohortReports.test.ts @@ -0,0 +1,181 @@ +import { assert, it } from "@effect/vitest"; +import { UsageDay, USAGE_CONTRACT_VERSION, type UsageSummary } from "@t3tools/contracts"; +import { EMPTY_TOTALS, type UsageRecord } from "@t3tools/provider-core/server/usage"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import { UsageAggregator, type PricedUsage } from "./usageAggregation.ts"; +import * as UsageCohortReports from "./UsageCohortReports.ts"; +import * as UsageService from "./UsageService.ts"; + +const INPUT = { + sinceDay: UsageDay.make("2026-11-01"), + untilDay: UsageDay.make("2026-11-02"), + timeZone: "UTC", +} as const; +const AT = Date.parse("2026-11-01T00:30:00Z"); + +const record = ( + sessionId: string, + costUsd: number, + overrides: Partial = {}, +): UsageRecord => ({ + provider: "codex", + timestampMs: AT, + model: "gpt-6.1-sol", + sessionId, + totals: { ...EMPTY_TOTALS, uncachedInputTokens: 100, outputTokens: 10 }, + reportedCostUsd: costUsd, + speed: "standard", + dedupeKey: null, + ...overrides, +}); + +const scanned: ReadonlyArray = [ + [record("s1", 1), undefined], + [record("s1", 1), undefined], + [record("s2", 3), undefined], + [record("s3", 5, { provider: "claude", model: "claude-opus-5-5" }), undefined], + [record("nobody", 7), undefined], + [record("s1", 11, { timestampMs: AT + 3 * 3_600_000 }), undefined], + [record("s1", 13, { provider: "claude" }), "/home/u/.claude/projects/p/s1/subagents/a.jsonl"], + [record("", 17), undefined], +]; + +const layerScanner = Layer.mock(UsageService.UsageService)({ + scanPriced: (input, onPriced: (priced: PricedUsage) => void) => + Effect.sync(() => { + const aggregator = new UsageAggregator({ + timeZone: input.timeZone, + sinceDay: input.sinceDay, + untilDay: input.untilDay, + rates: new Map(), + onPriced, + }); + for (const [item, path] of scanned) aggregator.add(item, "/dir", path); + return { + contractVersion: USAGE_CONTRACT_VERSION, + readAt: "2026-11-02T00:00:00.000Z", + timeZone: input.timeZone, + sinceDay: input.sinceDay, + untilDay: input.untilDay, + buckets: aggregator.finish().buckets, + sources: [], + pricing: { status: "fresh", source: "test", fetchedAt: null, knownModels: 0 }, + scanDurationMs: 0, + } satisfies UsageSummary; + }), +}); + +const layer = UsageCohortReports.layer.pipe( + Layer.provideMerge(layerScanner), + Layer.provideMerge(SqlitePersistence.layerMemory), +); + +const attempt = ( + id: string, + options: { + readonly session: string; + readonly startedAt: string | null; + readonly instance?: string; + }, +) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`INSERT INTO orchestration_v2_projection_run_attempts ${sql.insert({ + attempt_id: id, + thread_id: `thread-${id}`, + run_id: `run-${id}`, + attempt_ordinal: 1, + root_node_id: `node-${id}`, + provider: "codex", + provider_instance_id: options.instance ?? "codex", + provider_thread_id: `pt-${id}`, + provider_turn_id: null, + status: "completed", + payload_json: JSON.stringify({ + nativeThreadId: options.session, + startedAt: options.startedAt, + completedAt: options.startedAt === null ? null : "2026-11-01T01:00:00.000Z", + }), + })}`; + }); + +const stamp = (id: string, version: string, settings: Record) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`INSERT INTO usage_attempt_stamps ${sql.insert({ + attempt_id: id, + thread_id: `thread-${id}`, + origin: "user", + t3_version: version, + settings_json: JSON.stringify(settings), + })}`; + }); + +it.effect("accounts for every dollar of a scan exactly once", () => + Effect.gen(function* () { + yield* attempt("a1", { session: "s1", startedAt: "2026-11-01T00:00:00.000Z" }); + yield* attempt("a2", { session: "s2", startedAt: "2026-11-01T00:10:00.000Z" }); + yield* attempt("a3", { session: "s3", startedAt: "2026-11-01T00:20:00.000Z" }); + yield* attempt("never-started", { session: "s9", startedAt: null }); + yield* stamp("a1", "0.1.7-nightly.1", { pstackEnabled: false }); + yield* stamp("a2", "0.1.7-nightly.2", { pstackEnabled: true }); + yield* stamp("never-started", "0.1.7-nightly.2", { pstackEnabled: true }); + yield* stamp("orphan", "0.1.7-nightly.3", { pstackEnabled: true }); + + const report = yield* (yield* UsageCohortReports.UsageCohortReports).readReport({ + ...INPUT, + boundaryBuild: "0.1.7-nightly.2", + }); + const { ledger } = report; + + const unmatchedUsd = ledger.unmatched.reduce((sum, row) => sum + row.usd, 0); + assert.strictEqual(ledger.totalUsd, 58); + assert.strictEqual(ledger.attributedUsd + ledger.childWorkUsd + unmatchedUsd, ledger.totalUsd); + assert.deepStrictEqual( + { + attributed: ledger.attributedUsd, + before: ledger.beforeTrackingUsd, + child: ledger.childWorkUsd, + }, + { attributed: 10, before: 5, child: 13 }, + ); + assert.deepStrictEqual( + ledger.unmatched.map((row) => [row.reason, row.usd]), + [ + ["no_session_id", 17], + ["outside_attempt", 11], + ["no_attempt_here", 7], + ], + ); + assert.strictEqual(ledger.stampingSince, "2026-11-01T00:00:00.000Z"); + assert.deepStrictEqual( + report.builds.map((row) => row.build), + ["0.1.7-nightly.1", "0.1.7-nightly.2", null], + ); + const pstack = report.comparisons.find( + (comparison) => + comparison.subject._tag === "experiment" && comparison.subject.key === "pstackEnabled", + ); + assert.deepStrictEqual( + [pstack?.base.threads, pstack?.test.threads, pstack?.usdPerRun._tag], + [1, 1, "not_enough_data"], + ); + const build = report.comparisons.find((comparison) => comparison.subject._tag === "build"); + assert.strictEqual(build?.assignment, "self_selected"); + assert.deepStrictEqual([build?.base.usd, build?.test.usd], [2, 3]); + }).pipe(Effect.provide(layer)), +); + +it.effect("rejects a window longer than the scan cache keeps", () => + Effect.gen(function* () { + const error = yield* (yield* UsageCohortReports.UsageCohortReports) + .readReport({ ...INPUT, sinceDay: UsageDay.make("2026-07-01") }) + .pipe(Effect.flip); + + assert.deepStrictEqual([error._tag, error.reason], ["UsageReadError", "invalidWindow"]); + }).pipe(Effect.provide(layer)), +); diff --git a/apps/server/src/usage/UsageCohortReports.ts b/apps/server/src/usage/UsageCohortReports.ts new file mode 100644 index 0000000000000..5ea58d4e3a3e8 --- /dev/null +++ b/apps/server/src/usage/UsageCohortReports.ts @@ -0,0 +1,183 @@ +import { + USAGE_COHORT_MAX_WINDOW_DAYS, + UsageReadError, + type UsageCohortComparison, + type UsageCohortReport, + type UsageCohortReportInput, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Semaphore from "effect/Semaphore"; +import * as SqlClient from "effect/sql/SqlClient"; + +import { readAttempts, readBuildFirstSeen } from "./UsageAttemptStamps.ts"; +import * as UsageService from "./UsageService.ts"; +import { compareArms, type CompareInput } from "./usageCohortCompare.ts"; +import { + buildRows, + comparisonCells, + makeCohortScan, + type Arm, + type AttemptStamp, +} from "./usageCohortJoin.ts"; +import { USAGE_EXPERIMENTS, type UsageExperimentDefinition } from "./usageExperiments.ts"; + +const DAY_MS = 24 * 60 * 60 * 1000; +// An attempt older than this loses its records to "no_attempt_here", which the ledger shows. +const ATTEMPT_LOOKBACK_MS = 30 * DAY_MS; + +export class UsageCohortReports extends Context.Service< + UsageCohortReports, + { + readonly readReport: ( + input: UsageCohortReportInput, + ) => Effect.Effect; + } +>()("t3/usage/UsageCohortReports") {} + +const invalidWindow = (detail: string) => new UsageReadError({ reason: "invalidWindow", detail }); + +const dayStartMs = (day: string) => + Option.map(DateTime.make(`${day}T00:00:00Z`), DateTime.toEpochMillis); + +const compareCohort = Effect.fn("UsageCohortReports.compare")(function* (input: CompareInput) { + const steps = compareArms(input); + let step = steps.next(); + while (!step.done) { + yield* Effect.yieldNow; + step = steps.next(); + } + const result = step.value; + const empty = + result.base.threads === 0 && result.test.threads === 0 && result.threadsSpanningArms === 0; + return empty ? null : result; +}); + +export const layer = Layer.effect( + UsageCohortReports, + Effect.gen(function* () { + const usage = yield* UsageService.UsageService; + const sql = yield* SqlClient.SqlClient; + const oneAtATime = yield* Semaphore.make(1); + + const readReport = Effect.fn("UsageCohortReports.readReport")( + function* (input: UsageCohortReportInput) { + const sinceMs = dayStartMs(input.sinceDay); + const untilMs = dayStartMs(input.untilDay); + if (Option.isNone(sinceMs) || Option.isNone(untilMs) || sinceMs.value > untilMs.value) { + return yield* invalidWindow("Cohort report needs valid days, in order."); + } + const windowDays = Math.round((untilMs.value - sinceMs.value) / DAY_MS) + 1; + if (windowDays > USAGE_COHORT_MAX_WINDOW_DAYS) { + return yield* invalidWindow( + `Cohort report window must be at most ${USAGE_COHORT_MAX_WINDOW_DAYS} days.`, + ); + } + + const history = yield* Effect.all([ + readAttempts(sql, { + sinceMs: sinceMs.value - DAY_MS - ATTEMPT_LOOKBACK_MS, + untilMs: untilMs.value + 2 * DAY_MS, + }), + readBuildFirstSeen(sql), + ]).pipe( + Effect.mapError( + (cause) => + new UsageReadError({ + reason: "scanFailed", + detail: "Run attempt history could not be read.", + cause, + }), + ), + ); + const [attempts, firstSeenAtMs] = history; + + const scan = makeCohortScan(attempts); + const summary = yield* usage.scanPriced( + { + sinceDay: input.sinceDay, + untilDay: input.untilDay, + timeZone: input.timeZone, + awaitRefresh: true, + }, + scan.add, + ); + const { ledger, cells } = scan.finish(); + + const compare = (key: string, armOf: (stamp: AttemptStamp) => Arm | null) => + compareCohort({ key, windowDays, cells: comparisonCells(cells, armOf) }); + + const comparisons: UsageCohortComparison[] = []; + for (const [key, definition] of Object.entries( + USAGE_EXPERIMENTS, + )) { + const result = yield* compare(`experiment:${key}`, ({ settings }) => { + const value = settings[key]; + return value === undefined ? null : value ? "test" : "base"; + }); + if (result === null) continue; + comparisons.push({ + subject: { + _tag: "experiment", + key, + label: definition.label, + spendsOutsideTranscripts: definition.spendsOutsideTranscripts === true, + }, + assignment: "self_selected", + baseLabel: "Off", + testLabel: "On", + ...result, + }); + } + const boundaryMs = + input.boundaryBuild === undefined ? undefined : firstSeenAtMs.get(input.boundaryBuild); + if (input.boundaryBuild !== undefined && boundaryMs !== undefined) { + const result = yield* compare(`build:${input.boundaryBuild}`, ({ t3Version }) => { + const seen = firstSeenAtMs.get(t3Version); + return seen === undefined ? null : seen < boundaryMs ? "base" : "test"; + }); + if (result !== null) { + comparisons.push({ + subject: { _tag: "build", boundary: input.boundaryBuild }, + assignment: "self_selected", + baseLabel: `Before ${input.boundaryBuild}`, + testLabel: `${input.boundaryBuild} and later`, + ...result, + }); + } + } + + const stampingSince = firstSeenAtMs.size === 0 ? null : Math.min(...firstSeenAtMs.values()); + return { + readAt: summary.readAt, + sinceDay: input.sinceDay, + untilDay: input.untilDay, + timeZone: input.timeZone, + scanDurationMs: summary.scanDurationMs, + intervalConfidence: 0.998, + ledger: { + ...ledger, + incompleteSources: summary.sources.filter( + (source) => + source.status === "partial" || + source.status === "failed" || + source.refreshing === true, + ).length, + stampingSince: + stampingSince === null + ? null + : DateTime.formatIso(DateTime.makeUnsafe(stampingSince)), + }, + builds: buildRows(cells, firstSeenAtMs), + comparisons, + } satisfies UsageCohortReport; + }, + (effect) => effect.pipe(oneAtATime.withPermits(1)), + ); + + return { readReport }; + }), +); diff --git a/apps/server/src/usage/UsageService.test.ts b/apps/server/src/usage/UsageService.test.ts index 869b6aa9cefbf..b2aa4b84dfc43 100644 --- a/apps/server/src/usage/UsageService.test.ts +++ b/apps/server/src/usage/UsageService.test.ts @@ -554,6 +554,45 @@ describe("UsageService", () => { }).pipe(Effect.scoped), ); + it.live("scanPriced hands every counted record to the visitor with its transcript path", () => + Effect.gen(function* () { + const { transcript, settings, home } = yield* setup; + const subagent = NodePath.join( + NodePath.dirname(transcript), + "session-1", + "subagents", + "agent.jsonl", + ); + yield* Effect.promise(async () => { + await NodeFSP.mkdir(NodePath.dirname(subagent), { recursive: true }); + await NodeFSP.writeFile(transcript, claudeLine(1, 5) + claudeLine(1, 5)); + await NodeFSP.writeFile(subagent, claudeLine(2, 7)); + }); + const seen: Array<{ readonly path: string | undefined; readonly output: number }> = []; + const summary = yield* Effect.gen(function* () { + const service = yield* UsageService.make; + return yield* service.scanPriced(WINDOW, (priced) => { + seen.push({ path: priced.transcriptPath, output: priced.record.totals.outputTokens }); + }); + }).pipe( + Effect.scoped, + Effect.provide(layerService({ prefix: "usage-scan-priced", home, settings })), + ); + + assert.deepStrictEqual( + seen.toSorted((a, b) => a.output - b.output), + [ + { path: transcript, output: 5 }, + { path: subagent, output: 7 }, + ], + ); + assert.strictEqual( + seen.reduce((sum, item) => sum + item.output, 0), + totalOutputTokens(summary), + ); + }).pipe(Effect.scoped), + ); + it.live("keeps the first host identity across recreation and explicit override removal", () => 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 7e2e8586ed95f..ee5187fcbe0ee 100644 --- a/apps/server/src/usage/UsageService.ts +++ b/apps/server/src/usage/UsageService.ts @@ -39,7 +39,12 @@ import type { TranscriptUsageFormat, UsageRecord, } from "@t3tools/provider-core/server/usage"; -import { canonicalModelFilter, resolveModelAliases, UsageAggregator } from "./usageAggregation.ts"; +import { + canonicalModelFilter, + resolveModelAliases, + UsageAggregator, + type PricedUsage, +} from "./usageAggregation.ts"; import { createOverrideRateTable, parseRateTable, type RateTable } from "./usagePricing.ts"; import { listTranscriptFiles, @@ -143,6 +148,11 @@ export class UsageService extends Context.Service< UsageService, { readonly readSummary: (input: UsageSummaryInput) => Effect.Effect; + /** Scans without joining an in-flight read, since a shared result has no visitor for `onPriced`. */ + readonly scanPriced: ( + input: UsageSummaryInput, + onPriced: (priced: PricedUsage) => void, + ) => Effect.Effect; readonly refreshRates: Effect.Effect; } >()("t3/usage/UsageService") {} @@ -660,6 +670,7 @@ export const make = Effect.gen(function* () { const scanSummary = Effect.fn("UsageService.scanSummary")(function* ( input: UsageSummaryInput, settings: ServerSettingsValue, + onPriced?: (priced: PricedUsage) => void, ) { if (input.sinceDay > input.untilDay) { return yield* new UsageReadError({ @@ -725,6 +736,7 @@ export const make = Effect.gen(function* () { priceOverrides: createOverrideRateTable(settings.usagePriceOverrides), modelAliases: resolveModelAliases(settings.usageModelAliases), ...(input.modelFilter === undefined ? {} : { modelFilter: input.modelFilter }), + ...(onPriced === undefined ? {} : { onPriced }), }); const sources: UsageSource[] = []; @@ -833,7 +845,7 @@ export const make = Effect.gen(function* () { eventOccurrences.set(key, occurrence); usageRecord = { ...record, dedupeKey: key + ":" + occurrence }; } - if (aggregator.add(usageRecord, dir) && record.sessionId.length > 0) { + if (aggregator.add(usageRecord, dir, file.path) && record.sessionId.length > 0) { sessionIds.add(record.sessionId); } } @@ -935,7 +947,18 @@ export const make = Effect.gen(function* () { return yield* Deferred.await(deferred); }); - return { readSummary, refreshRates, awaitPersisted } as const; + const scanPriced = Effect.fn("UsageService.scanPriced")(function* ( + input: UsageSummaryInput, + onPriced: (priced: PricedUsage) => void, + ) { + const summary = yield* readSettings.pipe( + Effect.flatMap((settings) => scanSummary(input, settings, onPriced)), + ); + yield* schedulePersist; + return summary; + }); + + return { readSummary, scanPriced, refreshRates, awaitPersisted } as const; }); export const layer = Layer.effect(UsageService, make); diff --git a/apps/server/src/usage/usageAggregation.test.ts b/apps/server/src/usage/usageAggregation.test.ts index 08b3f3ae1b8c8..d65d3dcec648e 100644 --- a/apps/server/src/usage/usageAggregation.test.ts +++ b/apps/server/src/usage/usageAggregation.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from "@effect/vitest"; -import { resolveModelAliases, UsageAggregator } from "./usageAggregation.ts"; +import { resolveModelAliases, UsageAggregator, type PricedUsage } from "./usageAggregation.ts"; import type { RateTable } from "./usagePricing.ts"; import type { UsageRecord } from "@t3tools/provider-core/server/usage"; @@ -47,6 +47,7 @@ function aggregate( records: readonly UsageRecord[], timeZone = "UTC", resolution: "day" | "hour" = "day", + onPriced?: (priced: PricedUsage) => void, ) { const hourlyBounds = resolution === "hour" @@ -62,8 +63,10 @@ function aggregate( resolution, ...hourlyBounds, rates, + ...(onPriced === undefined ? {} : { onPriced }), }); - for (const item of records) aggregator.add(item); + for (const item of records) + aggregator.add(item, undefined, `/transcripts/${item.sessionId}.jsonl`); return aggregator.finish(); } @@ -98,6 +101,35 @@ describe("UsageAggregator", () => { expect(result.buckets[0]?.totals.outputTokens).toBe(50); }); + it("reports each counted record once, with the cost its bucket received", () => { + const seen: PricedUsage[] = []; + const result = aggregate( + [ + record({ dedupeKey: "msg_1:" }), + record({ dedupeKey: "msg_1:" }), + record({ model: "claude-fable-5", speed: "fast", sessionId: "session-b" }), + record({ timestampMs: Date.parse("2026-09-07T00:00:00Z") }), + record({ model: "no-rate-model" }), + ], + "UTC", + "day", + (priced) => seen.push(priced), + ); + + expect(seen.map((item) => item.record.sessionId)).toEqual([ + "session-a", + "session-b", + "session-a", + ]); + expect(seen.map((item) => item.unpriced)).toEqual([false, false, true]); + expect(seen[1]?.transcriptPath).toBe("/transcripts/session-b.jsonl"); + expect(seen.reduce((sum, item) => sum + item.costUsd, 0)).toBeCloseTo( + result.buckets.reduce((sum, bucket) => sum + bucket.costUsd, 0), + ); + expect(result.duplicatesDropped).toBe(1); + expect(result.outOfWindow).toBe(1); + }); + it("buckets by the day in the requested time zone", () => { const utc = aggregate([record()], "UTC"); const losAngeles = aggregate([record()], "America/Los_Angeles"); diff --git a/apps/server/src/usage/usageAggregation.ts b/apps/server/src/usage/usageAggregation.ts index 8f94c84a3c470..ac2fa6cf83a30 100644 --- a/apps/server/src/usage/usageAggregation.ts +++ b/apps/server/src/usage/usageAggregation.ts @@ -62,6 +62,14 @@ interface MutableBucket { sessions: Set; } +/** One record as the aggregator counted it, after filter, alias, dedupe, window, and pricing. */ +export interface PricedUsage { + readonly record: UsageRecord; + readonly costUsd: number; + readonly unpriced: boolean; + readonly transcriptPath: string | undefined; +} + export interface AggregateOptions { readonly timeZone: string; readonly sinceDay: string; @@ -73,6 +81,8 @@ export interface AggregateOptions { readonly resolution?: UsageResolution; readonly sinceTimeMs?: number; readonly untilTimeMs?: number; + /** Called once per record the aggregator counts. */ + readonly onPriced?: (priced: PricedUsage) => void; } export function resolveModelAliases( @@ -163,7 +173,7 @@ export class UsageAggregator { } } - add(input: UsageRecord, sourcePath?: string): boolean { + add(input: UsageRecord, sourcePath?: string, transcriptPath?: string): boolean { if (!keepsUsageRecord(this.#options.modelFilter, input)) return false; const record = this.#mapModel(input); if (record.dedupeKey !== null) { @@ -232,6 +242,12 @@ export class UsageAggregator { if (priced.costSource === "unpriced") bucket.unpricedRecords += 1; if (priced.costSource === "providerReported") bucket.providerReportedRecords += 1; if (record.sessionId.length > 0) bucket.sessions.add(record.sessionId); + this.#options.onPriced?.({ + record, + costUsd: priced.costUsd, + unpriced: priced.costSource === "unpriced", + transcriptPath, + }); return true; } diff --git a/apps/server/src/usage/usageCohortCompare.test.ts b/apps/server/src/usage/usageCohortCompare.test.ts new file mode 100644 index 0000000000000..8009871d4dca9 --- /dev/null +++ b/apps/server/src/usage/usageCohortCompare.test.ts @@ -0,0 +1,222 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { compareArms, type CompareInput } from "./usageCohortCompare.ts"; +import type { Arm, CompareCell } from "./usageCohortJoin.ts"; + +const run = (input: CompareInput) => { + const steps = compareArms(input); + for (let step = steps.next(); ; step = steps.next()) { + if (step.done) return step.value; + } +}; + +const cell = (arm: Arm, index: number, overrides: Partial = {}): CompareCell => ({ + threadId: `${arm}-${index}`, + arm, + providerInstanceId: "codex", + model: "gpt-6.1-sol", + speed: "standard", + runIds: [`run-${arm}-${index}`], + modelCalls: 10, + promptTokens: 1000, + outputTokens: 50, + usd: 1, + ...overrides, +}); + +const cells = (arm: Arm, count: number, at: (index: number) => Partial = () => ({})) => + Array.from({ length: count }, (_, index) => cell(arm, index, at(index))); + +const base = { key: "experiment:flag", windowDays: 14 }; +const noisy = (index: number) => 1 + (index % 5) * 0.1; + +describe("compareArms verdicts", () => { + it("is not enough data below the thread floor, with the arm totals still visible", () => { + const result = run({ ...base, cells: [...cells("base", 29), ...cells("test", 40)] }); + + expect(result.promptTokensPerCall).toMatchObject({ + _tag: "not_enough_data", + reason: "too_few_threads", + }); + expect(result.usdPerRun).toMatchObject({ _tag: "not_enough_data", reason: "too_few_threads" }); + expect(result.base).toMatchObject({ threads: 29, usd: 29 }); + }); + + it.each([ + [ + "most threads ran a model the other side never did", + (arm: Arm, i: number) => (i === 0 ? "shared" : `${arm}-only`), + ], + [ + "each model sits almost entirely on one side", + (arm: Arm, i: number) => ((i === 0) === (arm === "base") ? "model-a" : "model-b"), + ], + ])("is not enough data when %s", (_name, modelAt) => { + const side = (arm: Arm, usd: number) => + cells(arm, 30, (i) => ({ model: modelAt(arm, i), usd })); + const result = run({ ...base, cells: [...side("base", 1), ...side("test", 2)] }); + + expect(result.usdPerRun).toMatchObject({ _tag: "not_enough_data", reason: "too_few_threads" }); + expect(result.base.threads).toBe(30); + }); + + it("does not let one thread speak for a model the other side ran heavily", () => { + const balanced = (arm: Arm) => cells(arm, 200, () => ({ model: "balanced" })); + const heavy = cells("base", 200, (i) => ({ model: "thin", threadId: `heavy-${i}` })); + const thin = cell("test", 0, { model: "thin", threadId: "thin", usd: 8 }); + const result = run({ + ...base, + cells: [...balanced("base"), ...balanced("test"), ...heavy, thin], + }); + + expect(result.usdPerRun).toMatchObject({ + _tag: "no_detectable_change", + ratio: expect.closeTo(8 ** (1 / 201), 5), + }); + }); + + it("counts a run on two models once in the arm totals", () => { + const result = run({ + ...base, + cells: [ + cell("base", 0, { runIds: ["shared-run"] }), + cell("base", 0, { runIds: ["shared-run"], model: "gpt-6-luna" }), + ], + }); + + expect(result.base).toMatchObject({ threads: 1, runs: 1, usd: 2 }); + }); + + it("gives null estimates when an arm is empty", () => { + const result = run({ ...base, cells: cells("test", 40) }); + + expect(result.usdPerRun).toEqual({ + _tag: "not_enough_data", + reason: "too_few_threads", + threadsPerArmFor15Percent: null, + daysAtCurrentRate: null, + }); + }); + + it("reports a clear shift as changed with an interval around the ratio", () => { + const result = run({ + ...base, + cells: [ + ...cells("base", 40, (i) => ({ usd: noisy(i) })), + ...cells("test", 40, (i) => ({ usd: 2 * noisy(i) })), + ], + }); + + expect(result.usdPerRun).toMatchObject({ _tag: "changed", ratio: expect.closeTo(2, 5) }); + const verdict = result.usdPerRun; + if (verdict._tag !== "changed") throw new Error("expected changed"); + expect(verdict.low).toBeGreaterThan(1.5); + expect(verdict.high).toBeLessThan(2.6); + expect(result.promptTokensPerCall._tag).toBe("no_detectable_change"); + }); + + it("reports tight identical arms as no detectable change", () => { + const result = run({ ...base, cells: [...cells("base", 40), ...cells("test", 40)] }); + + expect(result.usdPerRun).toEqual({ _tag: "no_detectable_change", ratio: 1, low: 1, high: 1 }); + }); + + it("sizes the data it would need when the interval straddles the material band", () => { + const heavyTail = (shift: number) => (i: number) => ({ usd: i % 10 === 0 ? 30 * shift : 1 }); + const result = run({ + ...base, + cells: [...cells("base", 40, heavyTail(1)), ...cells("test", 40, heavyTail(1.1))], + }); + const verdict = result.usdPerRun; + + expect(verdict._tag).toBe("not_enough_data"); + if (verdict._tag !== "not_enough_data") return; + expect(verdict.reason).toBe("interval_too_wide"); + expect(verdict.threadsPerArmFor15Percent).toBeGreaterThan(40); + expect(verdict.daysAtCurrentRate).toBeGreaterThan(14); + expect(verdict).not.toHaveProperty("ratio"); + }); +}); + +describe("compareArms policy", () => { + it("returns a ratio of 1 for equal per-stratum values at different stratum proportions", () => { + const a = { model: "model-a", usd: 1 }; + const b = { model: "model-b", usd: 10 }; + const mix = (arm: Arm, aThreads: number, bThreads: number) => [ + ...cells(arm, aThreads, () => a), + ...cells(arm, bThreads, (i) => ({ ...b, threadId: `${arm}-b-${i}` })), + ]; + const result = run({ ...base, cells: [...mix("base", 90, 20), ...mix("test", 20, 90)] }); + + expect(result.usdPerRun).toMatchObject({ _tag: "no_detectable_change", ratio: 1 }); + expect(result.test.usd).toBeGreaterThan(result.base.usd * 3); + }); + + it("excludes a thread that appears in both arms and counts it", () => { + const spanning = [ + cell("base", 0, { threadId: "both", usd: 500 }), + cell("test", 0, { threadId: "both", usd: 500 }), + ]; + const result = run({ + ...base, + cells: [...spanning, ...cells("base", 35), ...cells("test", 35)], + }); + + expect(result.threadsSpanningArms).toBe(1); + expect(result.base).toMatchObject({ threads: 35, usd: 35 }); + expect(result.test).toMatchObject({ threads: 35, usd: 35 }); + }); + + it("returns the same verdict for the same input", () => { + const input = { + ...base, + cells: [ + ...cells("base", 40, (i) => ({ usd: noisy(i) })), + ...cells("test", 40, (i) => ({ usd: 1.05 * noisy(i) + (i % 7) * 0.05 })), + ], + }; + + expect(run(input)).toEqual(run(input)); + }); + + it("resamples a thread with cells in two strata as a whole", () => { + // A random source stuck on 0 always picks each arm's first thread. Resampling whole threads + // keeps both of its strata, so the ratio is the mean of 2x and 8x in log space, which is 4. + const stratum = (model: string, usd: number) => ({ model, usd }); + const spread = (arm: Arm, index: number, aUsd: number, bUsd: number) => [ + cell(arm, index, stratum("model-a", aUsd)), + cell(arm, index, stratum("model-b", bUsd)), + ]; + const input: CompareInput = { + ...base, + random: () => 0, + cells: [ + ...spread("base", 0, 1, 1), + ...spread("test", 0, 2, 8), + ...Array.from({ length: 34 }, (_, i) => spread("base", i + 1, 1, 1)).flat(), + ...Array.from({ length: 34 }, (_, i) => spread("test", i + 1, 3, 3)).flat(), + ], + }; + const verdict = run(input).usdPerRun; + + expect(verdict).toMatchObject({ + _tag: "changed", + low: expect.closeTo(4, 5), + high: expect.closeTo(4, 5), + }); + }); + + it("caps descriptive segments by spend", () => { + const models = Array.from({ length: 15 }, (_, index) => `model-${index}`); + const result = run({ + ...base, + cells: models.flatMap((model, index) => [ + cell("base", index, { model, usd: index + 1 }), + cell("test", index, { model, usd: index + 1 }), + ]), + }); + + expect(result.segments).toHaveLength(12); + expect(result.segments[0]).toMatchObject({ model: "model-14" }); + }); +}); diff --git a/apps/server/src/usage/usageCohortCompare.ts b/apps/server/src/usage/usageCohortCompare.ts new file mode 100644 index 0000000000000..d5d8e90135c91 --- /dev/null +++ b/apps/server/src/usage/usageCohortCompare.ts @@ -0,0 +1,300 @@ +import type { + UsageCohortComparison, + UsageCohortTotals, + UsageCohortVerdict, +} from "@t3tools/contracts"; + +import type { Arm, CompareCell } from "./usageCohortJoin.ts"; + +const MIN_THREADS_PER_ARM = 30; +const BOOTSTRAP_RESAMPLES = 4000; +const MAX_SEGMENTS_PER_COMPARISON = 12; +const RESAMPLES_PER_YIELD = 50; +const MATERIAL_RATIO = 1.15; +const LOG_MATERIAL = Math.log(MATERIAL_RATIO); +// Spend is heavy-tailed, so the bootstrap runs narrow. A nominal 99% interval flagged 2.75% of +// random splits of real threads, and a report holds up to 14 verdicts. This tail flagged 0.57%. +const TAIL = 0.001; + +export interface CompareInput { + /** Seeds the bootstrap, so the same comparison always returns the same verdict. */ + readonly key: string; + readonly cells: readonly CompareCell[]; + readonly windowDays: number; + readonly random?: () => number; +} + +export type CompareResult = Pick< + UsageCohortComparison, + "base" | "test" | "threadsSpanningArms" | "promptTokensPerCall" | "usdPerRun" | "segments" +>; + +interface Metric { + readonly numerator: "promptTokens" | "usd"; + readonly denominator: "modelCalls" | "runs"; +} + +const METRICS = { + promptTokensPerCall: { numerator: "promptTokens", denominator: "modelCalls" }, + usdPerRun: { numerator: "usd", denominator: "runs" }, +} as const satisfies Record; + +type MetricName = keyof typeof METRICS; + +interface Quantities { + promptTokens: number; + modelCalls: number; + usd: number; + runs: number; +} + +const zeroQuantities = (): Quantities => ({ promptTokens: 0, modelCalls: 0, usd: 0, runs: 0 }); + +interface ThreadCell extends Quantities { + readonly stratum: number; +} + +type Thread = readonly ThreadCell[]; + +// FNV-1a into mulberry32: small, dependency free, and stable across runtimes. +function seededRandom(key: string): () => number { + let hash = 2166136261; + for (let index = 0; index < key.length; index += 1) { + hash = Math.imul(hash ^ key.charCodeAt(index), 16777619); + } + let state = hash >>> 0; + return () => { + state = (state + 0x6d2b79f5) >>> 0; + let t = state; + t = Math.imul(t ^ (t >>> 15), t | 1); + t ^= t + Math.imul(t ^ (t >>> 7), t | 61); + return ((t ^ (t >>> 14)) >>> 0) / 4294967296; + }; +} + +/** Weighted mean of per-stratum log ratios over strata present in both arms, or null when none. */ +function stratifiedRatio( + metric: Metric, + base: readonly Quantities[], + test: readonly Quantities[], + baseThreads: ArrayLike, + testThreads: ArrayLike, +): { readonly logRatio: number; readonly comparableThreads: number } | null { + let weighted = 0; + let comparableThreads = 0; + for (let stratum = 0; stratum < base.length; stratum += 1) { + const b = base[stratum]!; + const t = test[stratum]!; + const baseDenominator = b[metric.denominator]; + const testDenominator = t[metric.denominator]; + const baseNumerator = b[metric.numerator]; + const testNumerator = t[metric.numerator]; + if (baseDenominator <= 0 || testDenominator <= 0 || baseNumerator <= 0 || testNumerator <= 0) { + continue; + } + // A stratum is only as precise as its smaller side, however many threads the other holds. + const weight = Math.min(baseThreads[stratum]!, testThreads[stratum]!); + weighted += + weight * Math.log(testNumerator / testDenominator / (baseNumerator / baseDenominator)); + comparableThreads += weight; + } + return comparableThreads === 0 + ? null + : { logRatio: weighted / comparableThreads, comparableThreads }; +} + +function sumArm(threads: readonly Thread[], strata: number) { + const sums = Array.from({ length: strata }, zeroQuantities); + const present = new Float64Array(strata); + for (const thread of threads) { + for (const cell of thread) { + const sum = sums[cell.stratum]!; + sum.promptTokens += cell.promptTokens; + sum.modelCalls += cell.modelCalls; + sum.usd += cell.usd; + sum.runs += cell.runs; + } + for (const stratum of new Set(thread.map((cell) => cell.stratum))) present[stratum]! += 1; + } + return { sums, present }; +} + +function quantile(sorted: readonly number[], p: number): number { + return sorted[Math.min(sorted.length - 1, Math.max(0, Math.floor(p * sorted.length)))]!; +} + +function verdictFrom( + point: number | null, + interval: { readonly low: number; readonly high: number } | null, + comparableThreads: number, + windowDays: number, +): UsageCohortVerdict { + const estimate = (): Pick< + Extract, + "threadsPerArmFor15Percent" | "daysAtCurrentRate" + > => { + if (interval === null || comparableThreads === 0) { + return { threadsPerArmFor15Percent: null, daysAtCurrentRate: null }; + } + const halfWidth = (Math.log(interval.high) - Math.log(interval.low)) / 2; + const needed = Math.max( + MIN_THREADS_PER_ARM, + Math.ceil(comparableThreads * (halfWidth / LOG_MATERIAL) ** 2), + ); + return { + threadsPerArmFor15Percent: needed, + daysAtCurrentRate: Math.ceil((windowDays * needed) / comparableThreads), + }; + }; + if (comparableThreads < MIN_THREADS_PER_ARM) { + return { _tag: "not_enough_data", reason: "too_few_threads", ...estimate() }; + } + if (point === null || interval === null) { + return { + _tag: "not_enough_data", + reason: "interval_too_wide", + threadsPerArmFor15Percent: null, + daysAtCurrentRate: null, + }; + } + const ratio = Math.exp(point); + if (interval.low > 1 || interval.high < 1) { + return { _tag: "changed", ratio, ...interval }; + } + if (interval.low >= 1 / MATERIAL_RATIO && interval.high <= MATERIAL_RATIO) { + return { _tag: "no_detectable_change", ratio, ...interval }; + } + return { _tag: "not_enough_data", reason: "interval_too_wide", ...estimate() }; +} + +const totalsOf = (cells: readonly CompareCell[]): UsageCohortTotals => ({ + threads: new Set(cells.map((cell) => cell.threadId)).size, + runs: new Set(cells.flatMap((cell) => cell.runIds)).size, + modelCalls: cells.reduce((sum, cell) => sum + cell.modelCalls, 0), + promptTokens: cells.reduce((sum, cell) => sum + cell.promptTokens, 0), + outputTokens: cells.reduce((sum, cell) => sum + cell.outputTokens, 0), + usd: cells.reduce((sum, cell) => sum + cell.usd, 0), +}); + +/** + * Compares two arms. A thread is the resampling unit and keeps all its cells, so cells that + * belong to one thread stay dependent. Yields between batches of resamples so a caller can + * give the event loop back. + */ +export function* compareArms(input: CompareInput): Generator { + const armsByThread = new Map>(); + for (const cell of input.cells) { + const arms = armsByThread.get(cell.threadId) ?? new Set(); + arms.add(cell.arm); + armsByThread.set(cell.threadId, arms); + } + const spanning = new Set([...armsByThread].filter(([, arms]) => arms.size > 1).map(([id]) => id)); + const cells = input.cells.filter((cell) => !spanning.has(cell.threadId)); + + const stratumIndex = new Map(); + const strataCells: CompareCell[] = []; + const grouped: Record> = { base: new Map(), test: new Map() }; + for (const cell of cells) { + const stratumKey = [cell.providerInstanceId, cell.model, cell.speed].join("\u0000"); + let stratum = stratumIndex.get(stratumKey); + if (stratum === undefined) { + stratum = stratumIndex.size; + stratumIndex.set(stratumKey, stratum); + strataCells.push(cell); + } + const thread = grouped[cell.arm].get(cell.threadId) ?? []; + thread.push({ + stratum, + promptTokens: cell.promptTokens, + modelCalls: cell.modelCalls, + usd: cell.usd, + runs: cell.runIds.length, + }); + grouped[cell.arm].set(cell.threadId, thread); + } + const strata = stratumIndex.size; + const baseThreads = [...grouped.base.values()]; + const testThreads = [...grouped.test.values()]; + + const baseSum = sumArm(baseThreads, strata); + const testSum = sumArm(testThreads, strata); + const metricNames = Object.keys(METRICS) as MetricName[]; + const points = Object.fromEntries( + metricNames.map((name) => [ + name, + stratifiedRatio(METRICS[name], baseSum.sums, testSum.sums, baseSum.present, testSum.present), + ]), + ) as Record>; + + const samples: Record = { promptTokensPerCall: [], usdPerRun: [] }; + if (baseThreads.length > 0 && testThreads.length > 0) { + const random = input.random ?? seededRandom(input.key); + const draw = (threads: readonly Thread[]) => + sumArm( + threads.map(() => threads[Math.floor(random() * threads.length)]!), + strata, + ); + for (let resample = 0; resample < BOOTSTRAP_RESAMPLES; resample += 1) { + const base = draw(baseThreads); + const test = draw(testThreads); + for (const name of metricNames) { + const value = stratifiedRatio( + METRICS[name], + base.sums, + test.sums, + base.present, + test.present, + )?.logRatio; + if (value !== undefined && Number.isFinite(value)) samples[name].push(value); + } + if ((resample + 1) % RESAMPLES_PER_YIELD === 0) yield; + } + } + + const verdict = (name: MetricName): UsageCohortVerdict => { + const sorted = samples[name].toSorted((a, b) => a - b); + const interval = + sorted.length < 2 + ? null + : { low: Math.exp(quantile(sorted, TAIL)), high: Math.exp(quantile(sorted, 1 - TAIL)) }; + const point = points[name]; + return verdictFrom( + point?.logRatio ?? null, + interval, + point?.comparableThreads ?? 0, + input.windowDays, + ); + }; + + const byArm = (arm: Arm) => cells.filter((cell) => cell.arm === arm); + const segments = strataCells + .map((sample) => { + const inSegment = (arm: Arm) => + byArm(arm).filter( + (cell) => + cell.providerInstanceId === sample.providerInstanceId && + cell.model === sample.model && + cell.speed === sample.speed, + ); + const base = totalsOf(inSegment("base")); + const test = totalsOf(inSegment("test")); + return { + providerInstanceId: sample.providerInstanceId, + model: sample.model, + speed: sample.speed, + base, + test, + }; + }) + .toSorted((a, b) => b.base.usd + b.test.usd - (a.base.usd + a.test.usd)) + .slice(0, MAX_SEGMENTS_PER_COMPARISON); + + return { + base: totalsOf(byArm("base")), + test: totalsOf(byArm("test")), + threadsSpanningArms: spanning.size, + promptTokensPerCall: verdict("promptTokensPerCall"), + usdPerRun: verdict("usdPerRun"), + segments, + }; +} diff --git a/apps/server/src/usage/usageCohortJoin.test.ts b/apps/server/src/usage/usageCohortJoin.test.ts new file mode 100644 index 0000000000000..cfa6835ab755e --- /dev/null +++ b/apps/server/src/usage/usageCohortJoin.test.ts @@ -0,0 +1,277 @@ +import { describe, expect, it } from "@effect/vitest"; +import { EMPTY_TOTALS, type UsageRecord } from "@t3tools/provider-core/server/usage"; + +import { + attribute, + buildRows, + comparisonCells, + indexAttemptsBySession, + makeCohortScan, + type AttemptRow, + type AttemptStamp, +} from "./usageCohortJoin.ts"; + +const T0 = Date.parse("2026-11-01T00:00:00Z"); +const HOUR = 3_600_000; + +const stamp = (overrides: Partial = {}): AttemptStamp => ({ + origin: "user", + t3Version: "0.1.7-nightly.20261102.72", + settings: {}, + ...overrides, +}); + +const attempt = (overrides: Partial = {}): AttemptRow => ({ + attemptId: "attempt-1", + runId: "run-1", + threadId: "thread-1", + sessionId: "session-a", + providerInstanceId: "codex", + startedAtMs: T0, + endedAtMs: T0 + HOUR, + stamp: stamp(), + ...overrides, +}); + +const usage = (overrides: Partial = {}): UsageRecord => ({ + provider: "codex", + timestampMs: T0 + HOUR / 2, + model: "gpt-6.1-sol", + sessionId: "session-a", + totals: { ...EMPTY_TOTALS, uncachedInputTokens: 100, cachedInputTokens: 900, outputTokens: 10 }, + reportedCostUsd: null, + speed: "standard", + dedupeKey: null, + ...overrides, +}); + +const priced = (record: UsageRecord, costUsd = 1, transcriptPath?: string, unpriced = false) => ({ + record, + costUsd, + unpriced, + transcriptPath, +}); + +describe("attribute", () => { + const first = attempt({ attemptId: "first", startedAtMs: T0, endedAtMs: T0 + HOUR }); + const second = attempt({ + attemptId: "second", + startedAtMs: T0 + 2 * HOUR, + endedAtMs: null, + stamp: null, + }); + const index = indexAttemptsBySession([second, first]); + + const cases: ReadonlyArray<{ + readonly name: string; + readonly record: Partial; + readonly path?: string; + readonly expected: unknown; + }> = [ + { + name: "a running attempt with no end owns later records", + record: { timestampMs: T0 + 90 * HOUR }, + expected: { _tag: "attempt", attempt: second }, + }, + { + name: "a record inside a finished attempt belongs to it", + record: { timestampMs: T0 + HOUR }, + expected: { _tag: "attempt", attempt: first }, + }, + { + name: "a record after an attempt ended and before the next started is outside", + record: { timestampMs: T0 + HOUR + 1 }, + expected: { _tag: "unmatched", reason: "outside_attempt" }, + }, + { + name: "a record before the first attempt on its session is outside", + record: { timestampMs: T0 - 1 }, + expected: { _tag: "unmatched", reason: "outside_attempt" }, + }, + { + name: "a session this environment never ran has no attempt here", + record: { sessionId: "other" }, + expected: { _tag: "unmatched", reason: "no_attempt_here" }, + }, + { + name: "an empty session id cannot be matched", + record: { sessionId: "" }, + expected: { _tag: "unmatched", reason: "no_session_id" }, + }, + { + name: "a Claude subagent transcript is child work even when its session matches", + record: { provider: "claude" }, + path: "/home/u/.claude/projects/p/session-a/subagents/agent-1.jsonl", + expected: { _tag: "child_work" }, + }, + { + name: "a Claude subagent transcript on Windows is child work", + record: { provider: "claude" }, + path: "C:\\Users\\u\\.claude\\projects\\p\\session-a\\subagents\\agent-1.jsonl", + expected: { _tag: "child_work" }, + }, + { + name: "a Codex transcript under a directory named subagents is still matched", + record: {}, + path: "/home/u/.codex/sessions/subagents/rollout.jsonl", + expected: { _tag: "attempt", attempt: first }, + }, + ]; + + it.each(cases)("$name", ({ record, path, expected }) => { + expect(attribute(index, usage(record), path)).toEqual(expected); + }); +}); + +describe("makeCohortScan", () => { + it("keeps a null stamp null after a stamped attempt on the same session", () => { + const stamped = attempt({ attemptId: "stamped", endedAtMs: T0 + HOUR }); + const unstamped = attempt({ + attemptId: "unstamped", + startedAtMs: T0 + 2 * HOUR, + endedAtMs: null, + stamp: null, + }); + const scan = makeCohortScan([stamped, unstamped]); + scan.add(priced(usage({ timestampMs: T0 + 3 * HOUR }), 4)); + const { ledger, cells } = scan.finish(); + + expect(cells.map((cell) => cell.attempt.attemptId)).toEqual(["unstamped"]); + expect(ledger).toMatchObject({ attributedUsd: 4, beforeTrackingUsd: 4 }); + }); + + it("puts every record in exactly one ledger bucket", () => { + const scheduled = attempt({ + attemptId: "scheduled", + sessionId: "session-s", + stamp: stamp({ origin: "scheduled" }), + }); + const scan = makeCohortScan([attempt(), scheduled]); + scan.add(priced(usage(), 1)); + scan.add(priced(usage({ sessionId: "session-s" }), 2)); + scan.add(priced(usage({ sessionId: "nobody" }), 3, undefined, true)); + scan.add(priced(usage({ timestampMs: T0 + 5 * HOUR }), 5)); + scan.add(priced(usage({ provider: "claude" }), 7, "/x/subagents/a.jsonl")); + const { ledger } = scan.finish(); + + expect(ledger).toMatchObject({ + totalUsd: 18, + attributedUsd: 3, + childWorkUsd: 7, + excludedOriginUsd: 2, + beforeTrackingUsd: 0, + unpricedRecords: 1, + }); + expect(ledger.unmatched).toEqual([ + { reason: "outside_attempt", provider: "codex", records: 1, usd: 5 }, + { reason: "no_attempt_here", provider: "codex", records: 1, usd: 3 }, + ]); + const unmatchedUsd = ledger.unmatched.reduce((sum, row) => sum + row.usd, 0); + expect(ledger.attributedUsd + ledger.childWorkUsd + unmatchedUsd).toBe(ledger.totalUsd); + }); +}); + +describe("comparisonCells", () => { + const scanOf = (attempts: readonly AttemptRow[], records: readonly UsageRecord[]) => { + const scan = makeCohortScan(attempts); + for (const record of records) scan.add(priced(record, 1)); + return scan.finish().cells; + }; + const enabled = (settings: AttemptStamp["settings"]) => (settings.flag ? "test" : "base"); + + it("keeps only user and agent origins on Claude and Codex, and counts runs once", () => { + const cells = scanOf( + [ + attempt({ attemptId: "a1", runId: "r1" }), + attempt({ attemptId: "a2", runId: "r1", startedAtMs: T0 + 2 * HOUR, endedAtMs: null }), + attempt({ + attemptId: "ghost", + sessionId: "s-ghost", + stamp: stamp({ origin: "scheduled" }), + }), + attempt({ attemptId: "old", sessionId: "s-old", stamp: null }), + attempt({ attemptId: "oc", sessionId: "s-oc" }), + ], + [ + usage({ timestampMs: T0 + 1 }), + usage({ timestampMs: T0 + 3 * HOUR }), + usage({ sessionId: "s-ghost" }), + usage({ sessionId: "s-old" }), + usage({ sessionId: "s-oc", provider: "opencode" }), + ], + ); + const compared = comparisonCells(cells, (s) => enabled(s.settings)); + + expect(compared).toHaveLength(1); + expect(compared[0]).toMatchObject({ modelCalls: 2, usd: 2, arm: "base" }); + expect(compared[0]!.runIds).toHaveLength(1); + }); + + it("assigns arms from the stamp and leaves out attempts that belong to neither", () => { + const cells = scanOf( + [ + attempt({ attemptId: "on", threadId: "t-on", stamp: stamp({ settings: { flag: true } }) }), + attempt({ + attemptId: "off", + threadId: "t-off", + sessionId: "s-off", + stamp: stamp({ settings: { flag: false } }), + }), + attempt({ attemptId: "none", threadId: "t-none", sessionId: "s-none" }), + ], + [usage(), usage({ sessionId: "s-off" }), usage({ sessionId: "s-none" })], + ); + const compared = comparisonCells(cells, ({ settings }) => + settings.flag === undefined ? null : settings.flag ? "test" : "base", + ); + + expect(compared.map((cell) => [cell.threadId, cell.arm]).toSorted()).toEqual([ + ["t-off", "base"], + ["t-on", "test"], + ]); + }); +}); + +describe("buildRows", () => { + it("orders builds by first seen, puts untracked spend last, and counts threads and runs", () => { + const attempts = [ + attempt({ + attemptId: "a", + threadId: "t1", + runId: "r1", + stamp: stamp({ t3Version: "0.1.8-b" }), + }), + attempt({ + attemptId: "b", + sessionId: "s2", + threadId: "t2", + runId: "r2", + stamp: stamp({ t3Version: "0.1.7-z" }), + }), + attempt({ + attemptId: "c", + sessionId: "s3", + threadId: "t2", + runId: "r3", + stamp: stamp({ t3Version: "0.1.7-z" }), + }), + attempt({ attemptId: "d", sessionId: "s4", threadId: "t4", stamp: null }), + ]; + const scan = makeCohortScan(attempts); + for (const sessionId of ["session-a", "s2", "s3", "s4"]) { + scan.add(priced(usage({ sessionId }), 2)); + } + const rows = buildRows( + scan.finish().cells, + new Map([ + ["0.1.8-b", T0 + 5 * HOUR], + ["0.1.7-z", T0 + HOUR], + ]), + ); + + expect(rows.map((row) => row.build)).toEqual(["0.1.7-z", "0.1.8-b", null]); + expect(rows[0]).toMatchObject({ firstSeenAt: "2026-11-01T01:00:00.000Z" }); + expect(rows[0]!.segments[0]).toMatchObject({ threads: 1, runs: 2, usd: 4, modelCalls: 2 }); + expect(rows[2]).toMatchObject({ firstSeenAt: null }); + }); +}); diff --git a/apps/server/src/usage/usageCohortJoin.ts b/apps/server/src/usage/usageCohortJoin.ts new file mode 100644 index 0000000000000..db1cc37bff2f4 --- /dev/null +++ b/apps/server/src/usage/usageCohortJoin.ts @@ -0,0 +1,308 @@ +import type { + UsageCohortReport, + UsageCohortSegment, + UsageCohortUnmatched, + UsageProviderKind, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Schema from "effect/Schema"; +import type { UsageRecord, UsageSpeed } from "@t3tools/provider-core/server/usage"; + +import type { PricedUsage } from "./usageAggregation.ts"; + +export const UsageOrigin = Schema.Literals(["user", "agent", "server_dispatched", "scheduled"]); +export type UsageOrigin = typeof UsageOrigin.Type; + +export interface AttemptStamp { + readonly origin: UsageOrigin; + readonly t3Version: string; + readonly settings: Readonly>; +} + +/** An attempt as the projection knows it, with the stamp left-joined; a null stamp means "before tracking". */ +export interface AttemptRow { + readonly attemptId: string; + readonly runId: string; + readonly threadId: string; + readonly sessionId: string; + readonly providerInstanceId: string; + readonly startedAtMs: number; + readonly endedAtMs: number | null; + readonly stamp: AttemptStamp | null; +} + +export type UnmatchedReason = UsageCohortUnmatched["reason"]; + +export type Attribution = + | { readonly _tag: "child_work" } + | { readonly _tag: "attempt"; readonly attempt: AttemptRow } + | { readonly _tag: "unmatched"; readonly reason: UnmatchedReason }; + +export type AttemptIndex = ReadonlyMap; + +export function indexAttemptsBySession(attempts: readonly AttemptRow[]): AttemptIndex { + const bySession = new Map(); + for (const attempt of attempts) { + if (attempt.sessionId.length === 0) continue; + const list = bySession.get(attempt.sessionId); + if (list === undefined) bySession.set(attempt.sessionId, [attempt]); + else list.push(attempt); + } + for (const list of bySession.values()) list.sort((a, b) => a.startedAtMs - b.startedAtMs); + return bySession; +} + +// Claude subagent transcripts cross parent attempts and carry no parent identity. +const isClaudeChildTranscript = (provider: UsageProviderKind, transcriptPath: string | undefined) => + provider === "claude" && transcriptPath?.split(/[\\/]/).includes("subagents") === true; + +export function attribute( + index: AttemptIndex, + record: Pick, + transcriptPath: string | undefined, +): Attribution { + if (isClaudeChildTranscript(record.provider, transcriptPath)) return { _tag: "child_work" }; + if (record.sessionId.length === 0) return { _tag: "unmatched", reason: "no_session_id" }; + const attempts = index.get(record.sessionId); + if (attempts === undefined) return { _tag: "unmatched", reason: "no_attempt_here" }; + const owner = attempts.findLast((attempt) => attempt.startedAtMs <= record.timestampMs); + if (owner === undefined || (owner.endedAtMs !== null && owner.endedAtMs < record.timestampMs)) { + return { _tag: "unmatched", reason: "outside_attempt" }; + } + return { _tag: "attempt", attempt: owner }; +} + +const COMPARABLE_PROVIDERS: ReadonlySet = new Set(["claude", "codex"]); +const COMPARABLE_ORIGINS: ReadonlySet = new Set(["user", "agent"]); + +/** Spend of one attempt in one provider, model, and speed. */ +export interface AttemptCell { + readonly attempt: AttemptRow; + readonly provider: UsageProviderKind; + readonly model: string; + readonly speed: UsageSpeed; + modelCalls: number; + promptTokens: number; + outputTokens: number; + usd: number; +} + +export type CohortLedger = Omit< + UsageCohortReport["ledger"], + "incompleteSources" | "stampingSince" | "unmatched" +> & { readonly unmatched: readonly UsageCohortUnmatched[] }; + +export interface CohortScan { + readonly ledger: CohortLedger; + readonly cells: readonly AttemptCell[]; +} + +/** Collects the scan's priced records into a ledger and per-attempt cells. Every record lands in exactly one ledger bucket. */ +export function makeCohortScan(attempts: readonly AttemptRow[]) { + const index = indexAttemptsBySession(attempts); + const cells = new Map(); + const unmatched = new Map< + string, + { -readonly [K in keyof UsageCohortUnmatched]: UsageCohortUnmatched[K] } + >(); + const sums = { + totalUsd: 0, + attributedUsd: 0, + beforeTrackingUsd: 0, + childWorkUsd: 0, + excludedOriginUsd: 0, + unpricedRecords: 0, + }; + + const add = ({ record, costUsd, unpriced, transcriptPath }: PricedUsage) => { + sums.totalUsd += costUsd; + if (unpriced) sums.unpricedRecords += 1; + const owner = attribute(index, record, transcriptPath); + if (owner._tag === "child_work") { + sums.childWorkUsd += costUsd; + return; + } + if (owner._tag === "unmatched") { + const key = `${owner.reason}\u0000${record.provider}`; + const row = unmatched.get(key) ?? { + reason: owner.reason, + provider: record.provider, + records: 0, + usd: 0, + }; + row.records += 1; + row.usd += costUsd; + unmatched.set(key, row); + return; + } + const { attempt } = owner; + sums.attributedUsd += costUsd; + if (attempt.stamp === null) sums.beforeTrackingUsd += costUsd; + else if (!COMPARABLE_ORIGINS.has(attempt.stamp.origin)) sums.excludedOriginUsd += costUsd; + + const key = `${attempt.attemptId}\u0000${record.provider}\u0000${record.model}\u0000${record.speed}`; + let cell = cells.get(key); + if (cell === undefined) { + cell = { + attempt, + provider: record.provider, + model: record.model, + speed: record.speed, + modelCalls: 0, + promptTokens: 0, + outputTokens: 0, + usd: 0, + }; + cells.set(key, cell); + } + cell.modelCalls += 1; + cell.promptTokens += + record.totals.uncachedInputTokens + + record.totals.cachedInputTokens + + record.totals.cacheCreationTokens; + cell.outputTokens += record.totals.outputTokens; + cell.usd += costUsd; + }; + + const finish = (): CohortScan => ({ + ledger: { ...sums, unmatched: [...unmatched.values()].sort((a, b) => b.usd - a.usd) }, + cells: [...cells.values()], + }); + + return { add, finish }; +} + +export type Arm = "base" | "test"; + +/** One thread's spend in one arm and segment; the unit the comparison resamples. */ +export interface CompareCell { + readonly threadId: string; + readonly arm: Arm; + readonly providerInstanceId: string; + readonly model: string; + readonly speed: UsageSpeed; + readonly runIds: readonly string[]; + readonly modelCalls: number; + readonly promptTokens: number; + readonly outputTokens: number; + readonly usd: number; +} + +/** Cells for one comparison. `armOf` returns null for an attempt that belongs to neither arm. */ +export function comparisonCells( + cells: readonly AttemptCell[], + armOf: (stamp: AttemptStamp) => Arm | null, +): CompareCell[] { + const merged = new Map< + string, + { -readonly [K in keyof Omit]: CompareCell[K] } & { + readonly runIds: Set; + } + >(); + for (const cell of cells) { + const { stamp } = cell.attempt; + if ( + stamp === null || + !COMPARABLE_PROVIDERS.has(cell.provider) || + !COMPARABLE_ORIGINS.has(stamp.origin) + ) { + continue; + } + const arm = armOf(stamp); + if (arm === null) continue; + const { threadId, providerInstanceId } = cell.attempt; + const key = [threadId, arm, providerInstanceId, cell.model, cell.speed].join("\u0000"); + let target = merged.get(key); + if (target === undefined) { + target = { + threadId, + arm, + providerInstanceId, + model: cell.model, + speed: cell.speed, + modelCalls: 0, + promptTokens: 0, + outputTokens: 0, + usd: 0, + runIds: new Set(), + }; + merged.set(key, target); + } + target.runIds.add(cell.attempt.runId); + target.modelCalls += cell.modelCalls; + target.promptTokens += cell.promptTokens; + target.outputTokens += cell.outputTokens; + target.usd += cell.usd; + } + return [...merged.values()].map(({ runIds, ...rest }) => ({ ...rest, runIds: [...runIds] })); +} + +type BuildRow = UsageCohortReport["builds"][number]; + +/** Builds in first-seen order, with the untracked remainder last. */ +export function buildRows( + cells: readonly AttemptCell[], + firstSeenAtMs: ReadonlyMap, +): BuildRow[] { + type Group = { + readonly threads: Set; + readonly runs: Set; + modelCalls: number; + promptTokens: number; + outputTokens: number; + usd: number; + }; + const builds = new Map< + string | null, + Map> + >(); + for (const cell of cells) { + const build = cell.attempt.stamp?.t3Version ?? null; + const segments = builds.get(build) ?? new Map(); + builds.set(build, segments); + const { providerInstanceId } = cell.attempt; + const key = [providerInstanceId, cell.model, cell.speed].join("\u0000"); + let group = segments.get(key); + if (group === undefined) { + group = { + providerInstanceId, + model: cell.model, + speed: cell.speed, + threads: new Set(), + runs: new Set(), + modelCalls: 0, + promptTokens: 0, + outputTokens: 0, + usd: 0, + }; + segments.set(key, group); + } + group.threads.add(cell.attempt.threadId); + group.runs.add(cell.attempt.runId); + group.modelCalls += cell.modelCalls; + group.promptTokens += cell.promptTokens; + group.outputTokens += cell.outputTokens; + group.usd += cell.usd; + } + const rank = (build: string | null) => + build === null + ? Number.POSITIVE_INFINITY + : (firstSeenAtMs.get(build) ?? Number.POSITIVE_INFINITY); + return [...builds] + .sort(([a], [b]) => rank(a) - rank(b) || String(a).localeCompare(String(b))) + .map(([build, segments]) => { + const firstSeen = build === null ? undefined : firstSeenAtMs.get(build); + return { + build, + firstSeenAt: + firstSeen === undefined ? null : DateTime.formatIso(DateTime.makeUnsafe(firstSeen)), + segments: [...segments.values()] + .sort((a, b) => b.usd - a.usd) + .map(({ threads, runs, ...group }) => ({ + ...group, + threads: threads.size, + runs: runs.size, + })), + }; + }); +} diff --git a/apps/server/src/usage/usageExperiments.ts b/apps/server/src/usage/usageExperiments.ts new file mode 100644 index 0000000000000..3a44be356aba6 --- /dev/null +++ b/apps/server/src/usage/usageExperiments.ts @@ -0,0 +1,24 @@ +import type { ServerSettings } from "@t3tools/contracts"; + +export type BooleanSettingKey = { + [K in keyof ServerSettings]-?: ServerSettings[K] extends boolean ? K : never; +}[keyof ServerSettings]; + +export interface UsageExperimentDefinition { + readonly label: string; + /** The flag makes model calls that no provider transcript records, so transcript cost understates it. */ + readonly spendsOutsideTranscripts?: true; +} + +/** + * Flags the cost report compares. A flag's consumer must read the settings the stamp resolves, + * not a project file, or the stamp would not match what ran. + */ +export const USAGE_EXPERIMENTS = { + enableAssistedLearning: { label: "Assisted learning", spendsOutsideTranscripts: true }, + enableConversationSummaries: { label: "Conversation summaries", spendsOutsideTranscripts: true }, + enableModelEvaluations: { label: "Model evaluations", spendsOutsideTranscripts: true }, + pstackEnabled: { label: "Pstack" }, + enableAgentGitHubReadGate: { label: "Agent GitHub read gate" }, + threadSplitSuggestionsEnabled: { label: "Thread split suggestions" }, +} as const satisfies Partial>; diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 4880d99e1d78f..cf0656491ef02 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -118,6 +118,7 @@ import * as TerminalManager from "./terminal/Manager.ts"; import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts"; import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; +import * as UsageCohortReports from "./usage/UsageCohortReports.ts"; import * as UsageService from "./usage/UsageService.ts"; import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts"; @@ -748,6 +749,7 @@ const layerUnusedConfigRpcServices = Layer.mergeAll( Layer.mock(ExternalHistory.ExternalHistory)({}), Layer.mock(ProjectLoom.ProjectLoom)({}), Layer.mock(UsageService.UsageService)({}), + Layer.mock(UsageCohortReports.UsageCohortReports)({}), Layer.mock(VcsProvisioningService.VcsProvisioningService)({}), Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({}), Layer.mock(WorkspaceEntries.WorkspaceEntries)({}), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 20b0d6d01edc4..3712bba84fbbb 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -265,6 +265,7 @@ import * as HostResources from "./resourceTelemetry/HostResources.ts"; import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; import { commandFeatureUsage, featureUsage } from "./telemetry/FeatureUsage.ts"; import type { TelemetryOwner } from "./telemetry/Identify.ts"; +import * as UsageCohortReports from "./usage/UsageCohortReports.ts"; import * as UsageService from "./usage/UsageService.ts"; import * as TraceDiagnostics from "./diagnostics/TraceDiagnostics.ts"; import * as TriageService from "./triage/TriageService.ts"; @@ -1607,6 +1608,7 @@ export const layerWsRpc = ( Effect.orElseSucceed(() => null), ); const usage = yield* UsageService.UsageService; + const usageCohortReports = yield* UsageCohortReports.UsageCohortReports; const usageLimitSources = yield* UsageLimitSources.UsageLimitSources; const worktreeSetupTracker = yield* WorktreeSetupTracker.WorktreeSetupTracker; const projectCloneTracker = yield* ProjectCloneTracker.ProjectCloneTracker; @@ -2860,6 +2862,7 @@ export const layerWsRpc = ( [WS_METHODS.serverGetResourceTelemetryHistory]: (input) => resourceTelemetry.readHistory(input), [WS_METHODS.serverGetUsageSummary]: (input) => usage.readSummary(input), + [WS_METHODS.serverGetUsageCohortReport]: (input) => usageCohortReports.readReport(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/components/settings/ZackSettings.tsx b/apps/web/src/components/settings/ZackSettings.tsx index ed41da6afbfee..29706b4883d48 100644 --- a/apps/web/src/components/settings/ZackSettings.tsx +++ b/apps/web/src/components/settings/ZackSettings.tsx @@ -32,6 +32,7 @@ export function ZackSettings() { + @@ -279,3 +280,26 @@ function AccountPoolsRow() { /> ); } + +/** The report lives in Usage because it reads one environment's transcripts and runs. */ +function UsageCostRow() { + const { environments } = useEnvironments(); + if ( + !environments.some( + (environment) => environment.serverConfig?.environment.capabilities.usageCohortReports === 1, + ) + ) { + return null; + } + return ( + }> + Open Usage + + } + /> + ); +} diff --git a/apps/web/src/components/usage/UsageCohortSection.tsx b/apps/web/src/components/usage/UsageCohortSection.tsx new file mode 100644 index 0000000000000..f82568478ebb5 --- /dev/null +++ b/apps/web/src/components/usage/UsageCohortSection.tsx @@ -0,0 +1,363 @@ +import { useAtomRefresh, useAtomValue } from "@effect/atom-react"; +import { + USAGE_COHORT_MAX_WINDOW_DAYS, + type EnvironmentId, + type UsageCohortComparison, + type UsageCohortReport, + type UsageCohortSegment, + type UsageCohortVerdict, + type UsageSummaryInput, +} from "@t3tools/contracts"; +import { formatCount, formatDayShort, formatTokens, formatUsd } from "@t3tools/shared/usageFormat"; +import { AsyncResult } from "effect/reactivity"; +import { useState } from "react"; + +import { environmentPresentations } from "../../state/presentation"; +import { serverEnvironment } from "../../state/server"; +import type { EnvironmentUsageStatus } from "../../state/usage"; +import { Badge } from "../ui/badge"; +import { Button } from "../ui/button"; +import { CollapsibleSectionHeader } from "../ui/collapsible-section-header"; +import { Select, SelectItem, SelectPopup, SelectTrigger, SelectValue } from "../ui/select"; +import { Skeleton } from "../ui/skeleton"; +import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "../ui/table"; + +const UNMATCHED_LABELS = { + no_session_id: "Record has no session id", + no_attempt_here: "No run on this environment matches the session", + outside_attempt: "Record falls outside every attempt on its session", +} as const; + +const NO_BOUNDARY = "none"; +const NO_BOUNDARY_LABEL = "No build comparison"; + +const segmentLabel = ( + segment: Pick, +) => + `${segment.providerInstanceId} · ${segment.model}${segment.speed === "standard" ? "" : ` (${segment.speed})`}`; + +const ratioText = (ratio: number) => `${ratio.toFixed(2)}x`; + +function VerdictCell({ verdict }: { readonly verdict: UsageCohortVerdict }) { + switch (verdict._tag) { + case "not_enough_data": + return ( + + Not enough data + + {verdict.reason === "too_few_threads" + ? "Each side needs at least 30 threads on the same provider, model, and speed." + : "The range is too wide to call a 15% change."} + {verdict.threadsPerArmFor15Percent !== null && verdict.daysAtCurrentRate !== null + ? ` About ${formatCount(verdict.threadsPerArmFor15Percent)} comparable threads per side, ${ + verdict.daysAtCurrentRate > USAGE_COHORT_MAX_WINDOW_DAYS + ? `more than the ${USAGE_COHORT_MAX_WINDOW_DAYS} days this report can cover` + : `about ${formatCount(verdict.daysAtCurrentRate)} days at this rate` + }.` + : null} + + + ); + case "no_detectable_change": + return ( + + No detectable change + + {ratioText(verdict.ratio)} ({ratioText(verdict.low)} to {ratioText(verdict.high)}) + + + ); + case "changed": + return ( + + Changed + + {ratioText(verdict.ratio)} ({ratioText(verdict.low)} to {ratioText(verdict.high)}) + + + ); + } +} + +function ArmTotals({ + label, + totals, +}: { + readonly label: string; + readonly totals: UsageCohortComparison["base"]; +}) { + return ( +

+ {label}: {formatCount(totals.threads)}{" "} + threads, {formatCount(totals.runs)} runs, {formatCount(totals.modelCalls)} calls,{" "} + {formatTokens(totals.promptTokens)} prompt tokens, {formatUsd(totals.usd)} +

+ ); +} + +function ComparisonCard({ comparison }: { readonly comparison: UsageCohortComparison }) { + const { subject } = comparison; + return ( +
+
+

+ {subject._tag === "experiment" ? subject.label : `Builds around ${subject.boundary}`} +

+ Self-selected +
+ {subject._tag === "experiment" && subject.spendsOutsideTranscripts ? ( +

+ This setting also makes model calls that provider transcripts do not record, so dollars + here understate its cost. +

+ ) : null} +
+ + + {comparison.threadsSpanningArms > 0 ? ( +

+ {formatCount(comparison.threadsSpanningArms)} threads span both sides and are left out. +

+ ) : null} +
+
+
Prompt tokens per call
+
+ +
+
Dollars per run
+
+ +
+
+ {comparison.segments.length > 0 ? ( + + + + Same provider, model, and speed + {comparison.baseLabel} tok/call + {comparison.testLabel} tok/call + {comparison.baseLabel} $/run + {comparison.testLabel} $/run + + + + {comparison.segments.map((segment) => ( + + {segmentLabel(segment)} + {perCall(segment.base)} + {perCall(segment.test)} + {perRun(segment.base)} + {perRun(segment.test)} + + ))} + +
+ ) : null} +
+ ); +} + +const perCall = (totals: UsageCohortComparison["base"]) => + totals.modelCalls === 0 ? "none" : formatTokens(totals.promptTokens / totals.modelCalls); +const perRun = (totals: UsageCohortComparison["base"]) => + totals.runs === 0 ? "none" : formatUsd(totals.usd / totals.runs); + +function Ledger({ ledger }: { readonly ledger: UsageCohortReport["ledger"] }) { + const unmatchedUsd = ledger.unmatched.reduce((sum, row) => sum + row.usd, 0); + return ( +
+
+
Total priced
+
{formatUsd(ledger.totalUsd)}
+
+
Matched to a run attempt
+
{formatUsd(ledger.attributedUsd)}
+
+ {formatUsd(ledger.beforeTrackingUsd)} is from before tracking started + {ledger.excludedOriginUsd > 0 + ? `, ${formatUsd(ledger.excludedOriginUsd)} from scheduled or server-started runs, left out of comparisons` + : null} +
+
Native subagent work
+
{formatUsd(ledger.childWorkUsd)}
+
Owner uncertain, left out of comparisons
+
Unmatched
+
{formatUsd(unmatchedUsd)}
+
+
+ {ledger.unmatched.length > 0 ? ( +
    + {ledger.unmatched.map((row) => ( +
  • + {UNMATCHED_LABELS[row.reason]}, {row.provider}: {formatUsd(row.usd)} ( + {formatCount(row.records)} records) +
    • + ))} +
+ ) : null} +

+ {ledger.stampingSince === null + ? "No run attempts have been recorded with a build yet." + : `Builds are recorded from ${formatDayShort(ledger.stampingSince.slice(0, 10))}.`} + {ledger.incompleteSources > 0 + ? ` ${formatCount(ledger.incompleteSources)} usage sources were incomplete, so totals may be short.` + : null} + {ledger.unpricedRecords > 0 + ? ` ${formatCount(ledger.unpricedRecords)} records have no known price.` + : null} +

+
+ ); +} + +function BuildsTable({ builds }: { readonly builds: UsageCohortReport["builds"] }) { + return ( + + + + Build + Provider, model, speed + Threads + Runs + Calls + Prompt tok/call + Cost + + + + {builds.flatMap((row) => + row.segments.map((segment) => ( + + {row.build ?? "Before tracking"} + {segmentLabel(segment)} + {formatCount(segment.threads)} + {formatCount(segment.runs)} + {formatCount(segment.modelCalls)} + {perCall(segment)} + {formatUsd(segment.usd)} + + )), + )} + +
+ ); +} + +function EnvironmentReport({ + environmentId, + window, +}: { + readonly environmentId: EnvironmentId; + readonly window: UsageSummaryInput; +}) { + const [boundary, setBoundary] = useState(null); + const reportAtom = serverEnvironment.usageCohortReport({ + environmentId, + input: { + sinceDay: window.sinceDay, + untilDay: window.untilDay, + timeZone: window.timeZone, + ...(boundary === null ? {} : { boundaryBuild: boundary }), + }, + }); + const result = useAtomValue(reportAtom); + const refresh = useAtomRefresh(reportAtom); + const report = AsyncResult.getOrElse(result, () => null); + if (report === null) { + return result._tag === "Failure" ? ( +
+ This environment could not build the report. + +
+ ) : ( + + ); + } + const builds = report.builds.flatMap((row) => (row.build === null ? [] : [row.build])); + return ( +
+

+ This environment, {formatDayShort(report.sinceDay)} to {formatDayShort(report.untilDay)}. + List-price equivalents, not invoices. +

+ + + {builds.length > 1 ? ( +
+ Compare builds before + + against it and later +
+ ) : null} + {report.comparisons.length > 0 ? ( +

+ Nobody assigned these sides at random, so a difference can come from who turned a setting + on or when a build was used. It is not proof of cause. +

+ ) : null} + {report.comparisons.map((comparison) => ( + + ))} +
+ ); +} + +/** Cost and token burn per build and Labs setting for one environment, loaded when opened. */ +export function UsageCohortSection({ + environments, + window, +}: { + readonly environments: readonly EnvironmentUsageStatus[]; + readonly window: UsageSummaryInput; +}) { + const presentations = useAtomValue(environmentPresentations.presentationsAtom); + const [open, setOpen] = useState(false); + const available = environments.filter( + (environment) => + environment.canReadDiagnostics && + environment.isConnected && + presentations.get(environment.environmentId)?.serverConfig?.environment.capabilities + .usageCohortReports === 1, + ); + const [only] = available; + if (only === undefined) return null; + return ( +
+ setOpen(!open)} tone="emphasized"> + Cost by build and setting + + {!open ? null : available.length > 1 ? ( +

+ Choose one environment above. Each environment reports on its own runs. +

+ ) : ( + + )} +
+ ); +} diff --git a/apps/web/src/components/usage/UsagePage.tsx b/apps/web/src/components/usage/UsagePage.tsx index b4ec17a8940cf..01a22f5de4d86 100644 --- a/apps/web/src/components/usage/UsagePage.tsx +++ b/apps/web/src/components/usage/UsagePage.tsx @@ -82,6 +82,7 @@ import { WorkspaceBreadcrumbText, } from "../WorkspaceBreadcrumb"; import { WorkspacePageContainer } from "../WorkspacePageContainer"; +import { UsageCohortSection } from "./UsageCohortSection"; import { UsageLimitsSection } from "./UsageLimits"; import { UsagePriceOverrides } from "./UsagePriceOverrides"; import { UsageProviderChart } from "./UsageProviderChart"; @@ -549,25 +550,28 @@ export function UsagePage() { ))} ) : ( - { - void refresh(); - void refreshLimits(false, true); - }} - onSelectModel={selectUsageModel} - /> + <> + { + void refresh(); + void refreshLimits(false, true); + }} + onSelectModel={selectUsageModel} + /> + + )} diff --git a/docs/user/usage.md b/docs/user/usage.md index 351639c1da81f..728a7fc2c880a 100644 --- a/docs/user/usage.md +++ b/docs/user/usage.md @@ -112,6 +112,25 @@ show every model again. If the setting is invalid, Usage ignores it and shows every model, and the next save of any other setting removes it from `settings.json`. +## See cost by build and setting + +On web and desktop, the **Cost** and **Tokens** views of **Usage** end with **Cost by build and +setting** for one environment. Choose one environment in the dropdown, then expand it. It reads +that environment's runs and transcripts, so it can take a few seconds. + +It first shows where the money went: priced spend matched to a run's attempt, native subagent work +(left out of comparisons because its owner is uncertain), and spend T3 could not match to any run +here. Spend from before T3 began recording builds shows as **Before tracking**. Recording starts +with the version that includes this report, so older history has no build or setting. + +Each Labs setting, and a build you choose, is compared with its other side on prompt tokens per +call and dollars per run. Only Claude Code and Codex runs started by you or an agent count, and +only against runs on the same provider, model, and speed. Nobody assigns the sides at random, so +a result is a difference between people and times, not proof of cause. A verdict says **Changed** +or **No detectable change** only when the threads recorded support it. Otherwise it says **Not +enough data**, with roughly how many threads and days it would take. Assisted learning, summaries, +and evaluations also call models that leave no transcript, so their dollars here are understated. + ## Track subscription limits **Usage → Limits** pools every subscription account it can see per provider, so with several Codex diff --git a/packages/client-runtime/src/state/commandPermissions.test.ts b/packages/client-runtime/src/state/commandPermissions.test.ts index c788baeb69de6..260876c714c35 100644 --- a/packages/client-runtime/src/state/commandPermissions.test.ts +++ b/packages/client-runtime/src/state/commandPermissions.test.ts @@ -9,6 +9,7 @@ import { vi } from "vite-plus/test"; import { AuthAccessReadScope, AuthAccessWriteScope, + AuthDiagnosticsReadScope, AuthOrchestrationOperateScope, AuthOrchestrationReadScope, AuthPreviewOperateScope, @@ -128,6 +129,30 @@ describe("command permissions", () => { ), ); + it.effect("requires the destination diagnostics grant to read the usage cohort report", () => + Effect.scoped( + Effect.gen(function* () { + const registry = yield* setup; + const report = createCommandPermissions(runtime, WS_METHODS.serverGetUsageCohortReport); + registry.set(sessions(env), AsyncResult.success(grant(true))); + expect(registry.get(report.permissionAtom(env))).toBe(false); + expect((yield* report.authorize(registry, env).pipe(Effect.flip))._tag).toBe( + "EnvironmentAuthorizationError", + ); + registry.set( + sessions(env), + AsyncResult.success({ + ...grant(false), + scopes: [AuthDiagnosticsReadScope], + permissions: [AuthDiagnosticsReadScope], + }), + ); + expect(registry.get(report.permissionAtom(env))).toBe(true); + yield* report.authorize(registry, env); + }), + ), + ); + it.effect("requires the destination preview grant to close a browser tab", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 93783f2f528b6..70f06dc25fba4 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -33,6 +33,7 @@ import { createAtomCommandScheduler, createEnvironmentRpcCommand, createEnvironmentQueryAtomFamily, + createEnvironmentGuardedRpcQueryAtomFamily, createEnvironmentRpcQueryAtomFamily, createEnvironmentRpcSubscriptionAtomFamily, createEnvironmentSubscriptionAtomFamily, @@ -1406,6 +1407,13 @@ export function createServerEnvironmentAtoms( ), ), }), + // One scan plus a bootstrap per comparison, so a reopened section reuses the last report. + usageCohortReport: createEnvironmentGuardedRpcQueryAtomFamily(runtime, { + label: "environment-data:server:usage-cohort-report", + tag: WS_METHODS.serverGetUsageCohortReport, + staleTimeMs: 5 * 60_000, + execute: (input) => requestGuarded(WS_METHODS.serverGetUsageCohortReport, input), + }), resourceTelemetry: createEnvironmentRpcSubscriptionAtomFamily(runtime, { label: "environment-data:server:resource-telemetry", tag: WS_METHODS.subscribeResourceTelemetry, diff --git a/packages/contracts/src/clientRpcPermissions.ts b/packages/contracts/src/clientRpcPermissions.ts index 994b4a70d77fb..67da83ea982e0 100644 --- a/packages/contracts/src/clientRpcPermissions.ts +++ b/packages/contracts/src/clientRpcPermissions.ts @@ -2,6 +2,7 @@ import * as Schema from "effect/Schema"; import { GitPreparePullRequestThreadInput, GitRunStackedActionInput } from "./git.ts"; import { AuthAccessWriteScope, + AuthDiagnosticsReadScope, AuthOrchestrationOperateScope, AuthFilesystemReadScope, AuthProvidersManageScope, @@ -21,6 +22,7 @@ export const CLIENT_GUARDED_RPC_SCOPES = { [WS_METHODS.projectCodeQuery]: AuthFilesystemReadScope, [WS_METHODS.projectCodeFileHistory]: AuthOrchestrationReadScope, [WS_METHODS.projectKnowledgeMutate]: AuthOrchestrationOperateScope, + [WS_METHODS.serverGetUsageCohortReport]: AuthDiagnosticsReadScope, [WS_METHODS.modelEvaluationsClear]: AuthSettingsWriteScope, [WS_METHODS.previewClose]: AuthPreviewOperateScope, // Stopping a server signals its processes, as closing a terminal does. diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 1d181886fe8e6..a795dc5382e18 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -195,6 +195,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ triage: Schema.optionalKey(Schema.Boolean), /** Server accepts `{ type: "once" }` scheduled tasks; older servers reject them on save. */ oneTimeSchedules: Schema.optionalKey(Schema.Boolean), + /** Server answers server.getUsageCohortReport. */ + usageCohortReports: Schema.optionalKey(Schema.Literal(1)), }); export type ExecutionEnvironmentCapabilities = typeof ExecutionEnvironmentCapabilities.Type; diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 43ca16d9e2d42..91ef19d03bb3b 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -70,6 +70,7 @@ export * from "./preview.ts"; export * from "./previewAutomation.ts"; export * from "./previewCapture.ts"; export * from "./usage.ts"; +export * from "./usageCohorts.ts"; export * from "./scheduledTask.ts"; export * from "./worktreeMcp.ts"; export * from "./resourceTelemetry.ts"; diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 3e66b29f9dde2..4c2930615ca4b 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -412,6 +412,7 @@ import { ProviderConsumeResetCreditResult, } from "./providerUsageLimits.ts"; import { UsagePricing, UsageReadError, UsageSummary, UsageSummaryInput } from "./usage.ts"; +import { UsageCohortReport, UsageCohortReportInput } from "./usageCohorts.ts"; import { StorageCleanupReport, DatabaseStorageStatus, @@ -622,6 +623,7 @@ export const WS_METHODS = { serverSignalProcess: "server.signalProcess", serverReportClientActivity: "server.reportClientActivity", serverGetUsageSummary: "server.getUsageSummary", + serverGetUsageCohortReport: "server.getUsageCohortReport", serverRefreshUsageRates: "server.refreshUsageRates", scheduledTasksSubscribe: "scheduledTasks.subscribe", @@ -1079,6 +1081,12 @@ const WsServerGetUsageSummaryRpc = Rpc.make(WS_METHODS.serverGetUsageSummary, { error: Schema.Union([EnvironmentAuthorizationError, UsageReadError]), }); +const WsServerGetUsageCohortReportRpc = Rpc.make(WS_METHODS.serverGetUsageCohortReport, { + payload: UsageCohortReportInput, + success: UsageCohortReport, + error: Schema.Union([EnvironmentAuthorizationError, UsageReadError]), +}); + const WsServerRefreshUsageRatesRpc = Rpc.make(WS_METHODS.serverRefreshUsageRates, { payload: Schema.Struct({}), success: UsagePricing, @@ -2306,6 +2314,7 @@ export const WsRpcGroup = RpcGroup.make( WsServerRefreshSkillsLibraryRpc, WsServerPrepareTriageRpc, WsServerGetUsageSummaryRpc, + WsServerGetUsageCohortReportRpc, WsServerRefreshUsageRatesRpc, WsServerSignalProcessRpc, WsScheduledTasksSubscribeRpc, diff --git a/packages/contracts/src/usageCohorts.test.ts b/packages/contracts/src/usageCohorts.test.ts new file mode 100644 index 0000000000000..90d62021dbb21 --- /dev/null +++ b/packages/contracts/src/usageCohorts.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it } from "vite-plus/test"; +import * as Schema from "effect/Schema"; + +import { UsageCohortReport } from "./usageCohorts.ts"; + +const totals = { threads: 1, runs: 1, modelCalls: 1, promptTokens: 1, outputTokens: 1, usd: 1 }; +const notEnough = { + _tag: "not_enough_data", + reason: "too_few_threads", + threadsPerArmFor15Percent: null, + daysAtCurrentRate: null, +}; +const comparison = { + subject: { _tag: "build", boundary: "0.1.8" }, + assignment: "self_selected", + baseLabel: "Before", + testLabel: "After", + base: totals, + test: totals, + threadsSpanningArms: 0, + promptTokensPerCall: notEnough, + usdPerRun: notEnough, + segments: [], +}; +const report = { + readAt: "2026-11-05T00:00:00.000Z", + sinceDay: "2026-11-01", + untilDay: "2026-11-05", + timeZone: "UTC", + scanDurationMs: 1, + intervalConfidence: 0.998, + ledger: { + totalUsd: 1, + attributedUsd: 1, + beforeTrackingUsd: 0, + childWorkUsd: 0, + excludedOriginUsd: 0, + unpricedRecords: 0, + incompleteSources: 0, + stampingSince: null, + unmatched: [], + }, + builds: [], + comparisons: [comparison], +}; + +const decode = Schema.decodeUnknownSync(UsageCohortReport); + +describe("UsageCohortReport across client and server versions", () => { + it("accepts a server that retuned its interval", () => { + expect(decode({ ...report, intervalConfidence: 0.95 }).comparisons).toHaveLength(1); + }); + + it.each([ + ["assignment", { ...comparison, assignment: "randomized" }], + ["verdict", { ...comparison, usdPerRun: { _tag: "some_new_verdict" } }], + ])("keeps the report when one comparison has an unknown %s", (_name, unknown) => { + expect(decode({ ...report, comparisons: [unknown, comparison] }).comparisons).toHaveLength(1); + }); + + it("keeps the report when one unmatched row has an unknown reason", () => { + const row = { provider: "codex", records: 1, usd: 1 }; + const decoded = decode({ + ...report, + ledger: { + ...report.ledger, + unmatched: [ + { ...row, reason: "some_new_reason" }, + { ...row, reason: "outside_attempt" }, + ], + }, + }); + expect(decoded.ledger.unmatched).toHaveLength(1); + }); +}); diff --git a/packages/contracts/src/usageCohorts.ts b/packages/contracts/src/usageCohorts.ts new file mode 100644 index 0000000000000..9aa0ac3da231f --- /dev/null +++ b/packages/contracts/src/usageCohorts.ts @@ -0,0 +1,122 @@ +import * as Schema from "effect/Schema"; + +import { ForwardCompatibleArray, NonNegativeInt, TrimmedNonEmptyString } from "./baseSchemas.ts"; +import { UsageDay } from "./usage.ts"; + +/** Matches the usage scan cache's retention, past which transcripts are gone. */ +export const USAGE_COHORT_MAX_WINDOW_DAYS = 92; + +export const UsageCohortReportInput = Schema.Struct({ + sinceDay: UsageDay, + untilDay: UsageDay, + /** IANA zone that turns the day bounds into instants. */ + timeZone: TrimmedNonEmptyString, + /** Builds first seen before this one are the base arm of the build comparison. */ + boundaryBuild: Schema.optional(TrimmedNonEmptyString), +}); +export type UsageCohortReportInput = typeof UsageCohortReportInput.Type; + +export const UsageCohortVerdict = Schema.Union([ + Schema.TaggedStruct("not_enough_data", { + reason: Schema.Literals(["too_few_threads", "interval_too_wide"]), + threadsPerArmFor15Percent: Schema.NullOr(NonNegativeInt), + daysAtCurrentRate: Schema.NullOr(NonNegativeInt), + }), + Schema.TaggedStruct("no_detectable_change", { + ratio: Schema.Number, + low: Schema.Number, + high: Schema.Number, + }), + Schema.TaggedStruct("changed", { + ratio: Schema.Number, + low: Schema.Number, + high: Schema.Number, + }), +]); +export type UsageCohortVerdict = typeof UsageCohortVerdict.Type; + +export const UsageCohortTotals = Schema.Struct({ + threads: NonNegativeInt, + runs: NonNegativeInt, + modelCalls: NonNegativeInt, + promptTokens: NonNegativeInt, + outputTokens: NonNegativeInt, + usd: Schema.Number, +}); +export type UsageCohortTotals = typeof UsageCohortTotals.Type; + +const UsageCohortSegmentKey = { + providerInstanceId: TrimmedNonEmptyString, + model: TrimmedNonEmptyString, + speed: Schema.Literals(["standard", "fast", "ultrafast"]), +}; + +export const UsageCohortSegment = Schema.Struct({ + ...UsageCohortSegmentKey, + ...UsageCohortTotals.fields, +}); +export type UsageCohortSegment = typeof UsageCohortSegment.Type; + +export const UsageCohortUnmatched = Schema.Struct({ + reason: Schema.Literals(["no_session_id", "no_attempt_here", "outside_attempt"]), + provider: TrimmedNonEmptyString, + records: NonNegativeInt, + usd: Schema.Number, +}); +export type UsageCohortUnmatched = typeof UsageCohortUnmatched.Type; + +export const UsageCohortComparison = Schema.Struct({ + subject: Schema.Union([ + Schema.TaggedStruct("experiment", { + key: TrimmedNonEmptyString, + label: TrimmedNonEmptyString, + spendsOutsideTranscripts: Schema.Boolean, + }), + Schema.TaggedStruct("build", { boundary: TrimmedNonEmptyString }), + ]), + assignment: Schema.Literal("self_selected"), + baseLabel: Schema.String, + testLabel: Schema.String, + base: UsageCohortTotals, + test: UsageCohortTotals, + threadsSpanningArms: NonNegativeInt, + promptTokensPerCall: UsageCohortVerdict, + usdPerRun: UsageCohortVerdict, + segments: ForwardCompatibleArray( + Schema.Struct({ + ...UsageCohortSegmentKey, + base: UsageCohortTotals, + test: UsageCohortTotals, + }), + ), +}); +export type UsageCohortComparison = typeof UsageCohortComparison.Type; + +export const UsageCohortReport = Schema.Struct({ + readAt: Schema.String, + sinceDay: UsageDay, + untilDay: UsageDay, + timeZone: TrimmedNonEmptyString, + scanDurationMs: NonNegativeInt, + intervalConfidence: Schema.Number, + ledger: Schema.Struct({ + totalUsd: Schema.Number, + attributedUsd: Schema.Number, + beforeTrackingUsd: Schema.Number, + childWorkUsd: Schema.Number, + excludedOriginUsd: Schema.Number, + unpricedRecords: NonNegativeInt, + incompleteSources: NonNegativeInt, + stampingSince: Schema.NullOr(Schema.String), + unmatched: ForwardCompatibleArray(UsageCohortUnmatched), + }), + builds: ForwardCompatibleArray( + Schema.Struct({ + build: Schema.NullOr(TrimmedNonEmptyString), + firstSeenAt: Schema.NullOr(Schema.String), + segments: ForwardCompatibleArray(UsageCohortSegment), + }), + ), + comparisons: ForwardCompatibleArray(UsageCohortComparison), +}); +export type UsageCohortReport = typeof UsageCohortReport.Type;