feat(usage): see token burn and cost by T3 build and Labs setting

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,

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,

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",

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<typeof UsageAttemptStamps.UsageAttemptStamps>;
   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({

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,

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);

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);

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
+  `;
+});

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:" }))),
   );

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),