feat(usage): break usage down by project, thread, account and environment

apps/server/src/provider/Drivers/claudeUsage.ts

@@ -9,6 +9,7 @@ import {
   type ProviderUsageReader,
   type TranscriptUsageFormat,
   type UsageRecord,
+  workingDirectoryHook,
 } from "@t3tools/provider-core/server/usage";
 import * as HostProcess from "@t3tools/shared/HostProcess";
 
@@ -115,12 +116,16 @@ export const claudeUsageFormat: TranscriptUsageFormat<void> = {
     timestamp: true,
     requestId: true,
     sessionId: true,
+    cwd: true,
     costUSD: true,
     message: { id: true, model: true, usage: true },
   },
   usageMarkers: ['"usage"'],
   parseLine: (line) => parseClaudeLine(line),
   parseProjected: (projected) => parseClaudeRecord(projected),
+  // Every record names the folder, but the first usage line may come after
+  // records without usage, so any record naming it is read until one does.
+  workingDirectory: workingDirectoryHook(['"cwd"'], (record) => record["cwd"]),
 };
 
 export const claudeUsageReader: ProviderUsageReader<ClaudeSettings, Path.Path> = {

apps/server/src/provider/Drivers/codexUsage.ts

@@ -9,6 +9,7 @@ import {
   type TranscriptUsageFormat,
   type UsageRecord,
   type UsageSpeed,
+  workingDirectoryHook,
 } from "@t3tools/provider-core/server/usage";
 import type { UsageTokenTotals } from "@t3tools/contracts";
 
@@ -416,6 +417,7 @@ export const codexUsageFormat: TranscriptUsageFormat<CodexScanState> = {
       type: true,
       id: true,
       session_id: true,
+      cwd: true,
       model_provider: true,
       turn_id: true,
       model: true,
@@ -440,6 +442,15 @@ export const codexUsageFormat: TranscriptUsageFormat<CodexScanState> = {
   readMalformedRecords: (state) => state.malformedRecords,
   // A rollout that moved after it was read leaves a copy in both places.
   sharedSessionsAcrossFiles: true,
+  // The rollout's first `session_meta` is its own session; a fork repeats its
+  // ancestors' metas after it, and the engine keeps the first folder.
+  workingDirectory: workingDirectoryHook(['"session_meta"'], (record, state: CodexScanState) => {
+    if (state.sawSessionMeta) return null;
+    const payload = record["payload"];
+    return record["type"] === "session_meta" && typeof payload === "object" && payload !== null
+      ? (payload as Record<string, unknown>)["cwd"]
+      : null;
+  }),
 };
 
 export const codexUsageReader: ProviderUsageReader<

apps/server/src/server.ts

@@ -198,6 +198,7 @@ import { MuseUsageHome } from "@t3tools/provider-muse/server";
 import * as CursorAccountReader from "@t3tools/provider-cursor/server/CursorAccountReader";
 import * as CursorUsageAccounts from "@t3tools/provider-cursor/server/CursorUsageAccounts";
 import * as UsageService from "./usage/UsageService.ts";
+import * as UsageAttribution from "./usage/UsageAttribution.ts";
 import * as CodexUsageHistory from "./usage/CodexUsageHistory.ts";
 import * as RuntimeLayer from "./orchestration-v2/runtimeLayer.ts";
 import * as ProjectStore from "./orchestration-v2/ProjectStore.ts";
@@ -300,6 +301,7 @@ const layerUsage = UsageService.layer.pipe(
       CursorUsageAccounts.layer.pipe(Layer.provide(CursorAccountReader.layer)),
     ).pipe(Layer.provide(ProviderHostLive.layer.pipe(Layer.provide(ServerSecretStore.layer)))),
   ),
+  Layer.provide(UsageAttribution.layer.pipe(Layer.provide(ProjectStore.layer))),
   Layer.provide(layerServerSettings),
 );
 

apps/server/src/usage/UsageAttribution.ts

@@ -0,0 +1,149 @@
+import { ProjectId, ThreadId } from "@t3tools/contracts";
+import * as Context from "effect/Context";
+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 * as SqlSchema from "effect/sql/SqlSchema";
+
+import * as ProjectStore from "../orchestration-v2/ProjectStore.ts";
+import type { NativeThreadRef, T3ThreadRef, UsageAttributionIndex } from "./usageThreadIndex.ts";
+
+export class UsageAttributionError extends Schema.TaggedError<UsageAttributionError>()(
+  "UsageAttributionError",
+  {
+    /** Which read failed. */
+    stage: Schema.Literals(["providerThreads", "subagents", "threads", "projects"]),
+    cause: Schema.Defect(),
+  },
+) {
+  override get message(): string {
+    return `Could not read ${this.stage} for usage attribution.`;
+  }
+}
+
+const failedAt =
+  (stage: UsageAttributionError["stage"]) =>
+  <A, E, R>(read: Effect.Effect<A, E, R>) =>
+    read.pipe(Effect.mapError((cause) => new UsageAttributionError({ stage, cause })));
+
+const NativeRow = Schema.Struct({
+  driver: Schema.String,
+  nativeId: Schema.String,
+  threadId: ThreadId,
+  instanceId: Schema.NullOr(Schema.String),
+});
+
+const ThreadRow = Schema.Struct({
+  threadId: ThreadId,
+  projectId: ProjectId,
+  title: Schema.NullOr(Schema.String),
+  parentThreadId: Schema.NullOr(ThreadId),
+});
+
+/**
+ * Reads which T3 thread each provider session and provider sub-agent belongs
+ * to, with thread lineage and every project, including archived and deleted
+ * ones so their past usage keeps its place.
+ */
+export class UsageAttribution extends Context.Service<
+  UsageAttribution,
+  { readonly read: Effect.Effect<UsageAttributionIndex, UsageAttributionError> }
+>()("t3/usage/UsageAttribution") {}
+
+const make = Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+  const projectStore = yield* ProjectStore.ProjectStoreV2;
+
+  const selectNativeThreads = SqlSchema.findAll({
+    Request: Schema.Void,
+    Result: NativeRow,
+    execute: () => sql`
+      SELECT
+        driver,
+        json_extract(payload_json, '$.nativeThreadRef.nativeId') AS "nativeId",
+        thread_id AS "threadId",
+        provider_instance_id AS "instanceId"
+      FROM orchestration_v2_projection_provider_threads
+      WHERE thread_id IS NOT NULL
+        AND driver IS NOT NULL
+        AND json_extract(payload_json, '$.nativeThreadRef.nativeId') IS NOT NULL
+    `,
+  });
+
+  const selectNativeSubagents = SqlSchema.findAll({
+    Request: Schema.Void,
+    Result: NativeRow,
+    execute: () => sql`
+      SELECT
+        driver,
+        json_extract(payload_json, '$.nativeTaskRef.nativeId') AS "nativeId",
+        child_thread_id AS "threadId",
+        provider_instance_id AS "instanceId"
+      FROM orchestration_v2_projection_subagents
+      WHERE child_thread_id IS NOT NULL
+        AND driver IS NOT NULL
+        AND json_extract(payload_json, '$.nativeTaskRef.nativeId') IS NOT NULL
+    `,
+  });
+
+  const selectThreads = SqlSchema.findAll({
+    Request: Schema.Void,
+    Result: ThreadRow,
+    execute: () => sql`
+      SELECT
+        thread_id AS "threadId",
+        project_id AS "projectId",
+        title,
+        json_extract(payload_json, '$.lineage.parentThreadId') AS "parentThreadId"
+      FROM orchestration_v2_projection_threads
+    `,
+  });
+
+  const byNativeId = (rows: ReadonlyArray<typeof NativeRow.Type>) => {
+    const map = new Map<string, NativeThreadRef[]>();
+    for (const row of rows) {
+      const key = `${row.driver}\u0000${row.nativeId}`;
+      const entries = map.get(key) ?? [];
+      entries.push({ threadId: row.threadId, instanceId: row.instanceId });
+      map.set(key, entries);
+    }
+    return map;
+  };
+
+  const read = Effect.all(
+    [
+      selectNativeThreads(undefined).pipe(failedAt("providerThreads")),
+      selectNativeSubagents(undefined).pipe(failedAt("subagents")),
+      selectThreads(undefined).pipe(failedAt("threads")),
+      projectStore.list({ includeDeleted: true }).pipe(failedAt("projects")),
+    ],
+    { concurrency: 1 },
+  ).pipe(
+    Effect.map(([nativeThreads, nativeSubagents, threads, projects]) => ({
+      nativeThreads: byNativeId(nativeThreads),
+      nativeSubagents: byNativeId(nativeSubagents),
+      threads: new Map<string, T3ThreadRef>(
+        threads.map((row) => [
+          row.threadId,
+          {
+            projectId: row.projectId,
+            title: row.title?.trim() || null,
+            parentThreadId: row.parentThreadId,
+          },
+        ]),
+      ),
+      projects: projects.map((project) => ({
+        projectId: project.projectId,
+        title: project.title,
+        workspaceRoot: project.workspaceRoot,
+        deleted: project.deletedAt !== null,
+      })),
+    })),
+    Effect.withSpan("UsageAttribution.read"),
+  );
+
+  return UsageAttribution.of({ read });
+});
+
+export const layer = Layer.effect(UsageAttribution, make);