feat(web): show each thread's context window use in the sidebar

apps/server/src/usage/UsageService.test.ts

@@ -520,6 +520,69 @@ describe("Codex usage history", () => {
 });
 
 describe("UsageService", () => {
+  it.live("matches priced native session records to their T3 thread", () =>
+    Effect.gen(function* () {
+      const { transcript, home, settings } = yield* setup;
+      yield* Effect.promise(() => NodeFSP.writeFile(transcript, claudeLine(1, 25)));
+      yield* Effect.gen(function* () {
+        const sql = yield* SqlClient.SqlClient;
+        yield* sql`INSERT INTO orchestration_v2_projection_provider_threads ${sql.insert({
+          provider_thread_id: "provider-thread-usage",
+          thread_id: "thread-usage",
+          provider: "claudeAgent",
+          driver: "claudeAgent",
+          status: "ready",
+          updated_at: "2026-08-01T10:00:00Z",
+          payload_json: encodeUnknownJsonString({
+            nativeThreadRef: { driver: "claudeAgent", nativeId: "session-1", strength: "strong" },
+            contextUsage: { usedTokens: 42000, maxTokens: 200000 },
+          }),
+        })}`;
+        yield* sql`INSERT INTO orchestration_v2_projection_threads ${sql.insert({
+          thread_id: "thread-usage",
+          project_id: "project-usage",
+          title: "Usage",
+          default_provider: "claudeAgent",
+          provider_instance_id: "claudeAgent",
+          runtime_mode: "full-access",
+          interaction_mode: "default",
+          active_provider_thread_id: "provider-thread-usage",
+          created_at: "2026-08-01T10:00:00Z",
+          updated_at: "2026-08-01T10:00:00Z",
+          payload_json: "{}",
+        })}`;
+        const service = yield* UsageService.make;
+        const summary = yield* service.readSummary({ ...WINDOW, includeThreadUsage: true });
+        assert.deepEqual(
+          summary.threadUsage?.map((usage) => ({
+            threadId: usage.threadId,
+            input: usage.totals.uncachedInputTokens,
+            output: usage.totals.outputTokens,
+          })),
+          [{ threadId: "thread-usage", input: 10, output: 25 }],
+        );
+        assert.isAbove(summary.threadUsage?.[0]?.costUsd ?? 0, 0);
+        assert.deepEqual(summary.threadContexts, [
+          {
+            threadId: "thread-usage",
+            usage: { usedTokens: 42000, maxTokens: 200000 },
+            updatedAt: "2026-08-01T10:00:00Z",
+          },
+        ]);
+      }).pipe(
+        Effect.provide(
+          layerService({
+            prefix: "thread-usage",
+            home,
+            settings,
+            ratesDocument: {
+              "claude-fable-5": { input_cost_per_token: 0.000003, output_cost_per_token: 0.000015 },
+            },
+          }),
+        ),
+      );
+    }).pipe(Effect.scoped),
+  );
   it.live("counts compressed Codex rollouts across refresh, rewrites, and cache restarts", () =>
     Effect.gen(function* () {
       const { home, settings } = yield* setup;

apps/server/src/usage/UsageService.ts

@@ -9,6 +9,7 @@ import {
   type UsageSummary,
   type UsageSummaryInput,
   UsageReadError,
+  ThreadTokenUsageSnapshot,
 } from "@t3tools/contracts";
 import * as HostProcess from "@t3tools/shared/HostProcess";
 import * as Cause from "effect/Cause";
@@ -26,6 +27,8 @@ import * as Path from "effect/Path";
 import * as Schema from "effect/Schema";
 import * as Semaphore from "effect/Semaphore";
 import { HttpClient, HttpClientResponse } from "effect/http";
+import * as SqlClient from "effect/sql/SqlClient";
+import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
 
 import { writeFileStringAtomically } from "@t3tools/shared/atomicWrite";
 import { formatMalformedUsageWarning } from "@t3tools/shared/usageFormat";
@@ -105,6 +108,15 @@ const encodeRatesCache = Schema.encodeEffect(Schema.fromJsonString(RatesCacheFil
 const ScanCacheJson = Schema.fromJsonString(Schema.Unknown);
 const decodeScanCacheFile = Schema.decodeUnknownEffect(ScanCacheJson);
 const encodeUsageRecordKey = Schema.encodeSync(ScanCacheJson);
+const decodeThreadContext = Schema.decodeUnknownOption(
+  Schema.fromJsonString(
+    Schema.Struct({
+      ...ThreadTokenUsageSnapshot.fields,
+      updatedAt: Schema.optional(Schema.String),
+      maxTokens: Schema.optional(Schema.NullOr(Schema.Number)),
+    }),
+  ),
+);
 const CachedSource = Schema.Struct({ dir: Schema.String, volumeId: Schema.String });
 
 function isLaterRead(a: CachedFile, b: CachedFile): boolean {
@@ -156,6 +168,7 @@ export const make = Effect.gen(function* () {
   const httpClient = yield* HttpClient.HttpClient;
   const hostEnvironment = yield* HostProcess.Environment;
   const hostname = yield* HostProcess.Hostname;
+  const database = yield* SqlClient.SqlClient;
   const readerContext = yield* Effect.context<BuiltInUsageReadersEnv>();
 
   const fileCache: ScanCache = new Map();
@@ -763,7 +776,78 @@ export const make = Effect.gen(function* () {
       { concurrency: 2 },
     );
 
+    let threadIdsBySession: Map<string, readonly string[]> | undefined;
+    const threadContexts: NonNullable<UsageSummary["threadContexts"]>[number][] = [];
+    if (input.includeThreadUsage === true) {
+      // Context size for idle threads, from each thread's latest reported provider turn.
+      const contexts = yield* database<{
+        readonly threadId: string;
+        readonly usage: string | null;
+        readonly updatedAt: string;
+      }>`
+        SELECT t.thread_id AS threadId, p.updated_at AS updatedAt,
+          COALESCE((SELECT json_extract(pt.payload_json, '$.tokenUsage')
+            FROM orchestration_v2_projection_provider_turns pt
+            WHERE pt.provider_thread_id = p.provider_thread_id
+              AND json_extract(pt.payload_json, '$.tokenUsage.usedTokens') IS NOT NULL
+            ORDER BY pt.ordinal DESC LIMIT 1), json_extract(p.payload_json, '$.contextUsage')) AS usage
+        FROM orchestration_v2_projection_threads t
+        JOIN orchestration_v2_projection_provider_threads p ON p.provider_thread_id = t.active_provider_thread_id
+        WHERE t.deleted_at IS NULL
+      `.pipe(
+        NodeSqliteClient.readOnly,
+        Effect.mapError(
+          (cause) =>
+            new UsageReadError({
+              reason: "scanFailed",
+              detail: "Could not read session context usage.",
+              cause,
+            }),
+        ),
+      );
+      for (const row of contexts) {
+        const decoded = row.usage === null ? Option.none() : decodeThreadContext(row.usage);
+        if (Option.isNone(decoded)) continue;
+        const { maxTokens, updatedAt, ...usage } = decoded.value;
+        threadContexts.push({
+          threadId: row.threadId,
+          usage: { ...usage, ...(maxTokens != null && maxTokens > 0 ? { maxTokens } : {}) },
+          updatedAt: updatedAt ?? row.updatedAt,
+        });
+      }
+      const links = yield* database<{
+        readonly threadId: string;
+        readonly driver: string;
+        readonly sessionId: string;
+      }>`
+        SELECT DISTINCT thread_id AS threadId, driver,
+          json_extract(payload_json, '$.nativeThreadRef.nativeId') AS sessionId
+        FROM orchestration_v2_projection_provider_threads
+        WHERE json_extract(payload_json, '$.nativeThreadRef.nativeId') IS NOT NULL
+      `.pipe(
+        NodeSqliteClient.readOnly,
+        Effect.mapError(
+          (cause) =>
+            new UsageReadError({
+              reason: "scanFailed",
+              detail: "Could not match native usage to sessions.",
+              cause,
+            }),
+        ),
+      );
+      threadIdsBySession = new Map();
+      for (const link of links) {
+        const provider = BUILT_IN_USAGE_DRIVERS.find((driver) => driver.driverKind === link.driver)
+          ?.usage?.provider;
+        if (!provider) continue;
+        const key = `${provider}:${link.sessionId}`;
+        const previous = threadIdsBySession.get(key) ?? [];
+        if (!previous.includes(link.threadId))
+          threadIdsBySession.set(key, [...previous, link.threadId]);
+      }
+    }
     const aggregator = new UsageAggregator({
+      ...(threadIdsBySession === undefined ? {} : { threadIdsBySession }),
       timeZone: input.timeZone,
       sinceDay: input.sinceDay,
       untilDay: input.untilDay,
@@ -927,6 +1011,8 @@ export const make = Effect.gen(function* () {
       sinceDay: input.sinceDay,
       untilDay: input.untilDay,
       buckets: aggregated.buckets,
+      ...(aggregated.threadUsage === undefined ? {} : { threadUsage: aggregated.threadUsage }),
+      ...(input.includeThreadUsage ? { threadContexts } : {}),
       sources,
       pricing: pricing(),
       scanDurationMs: Math.max(0, finishedAtMs - startedAtMs),
@@ -947,6 +1033,7 @@ export const make = Effect.gen(function* () {
       settings,
       input.awaitRefresh === true,
       input.modelFilter ?? null,
+      input.includeThreadUsage ?? false,
     ]);
 
   const readSummary = Effect.fn("UsageService.readSummary")(function* (

apps/server/src/usage/usageAggregation.test.ts

@@ -69,6 +69,58 @@ function aggregate(
 }
 
 describe("UsageAggregator", () => {
+  it("prices thread usage separately across models and deduplicates native session links", () => {
+    const aggregator = new UsageAggregator({
+      timeZone: "UTC",
+      sinceDay: "2026-08-01",
+      untilDay: "2026-08-31",
+      rates,
+      threadIdsBySession: new Map([
+        ["claude:session-a", ["thread-a"]],
+        ["claude:session-b", ["thread-b"]],
+      ]),
+    });
+    aggregator.add(record({ dedupeKey: "a" }));
+    aggregator.add(record({ dedupeKey: "a" }));
+    aggregator.add(record({ sessionId: "session-b", model: "unknown" }));
+    const usage = aggregator.finish().threadUsage;
+    expect(usage?.find((entry) => entry.threadId === "thread-a")).toMatchObject({
+      totals: {
+        uncachedInputTokens: 100,
+        cachedInputTokens: 1000,
+        cacheCreationTokens: 10,
+        outputTokens: 50,
+      },
+      unpricedRecords: 0,
+    });
+    expect(usage?.find((entry) => entry.threadId === "thread-a")?.costUsd).toBeCloseTo(0.004625, 8);
+    expect(usage?.find((entry) => entry.threadId === "thread-b")).toMatchObject({
+      unpricedRecords: 1,
+    });
+  });
+  it("attributes a copied inference when its linked session is scanned second", () => {
+    const aggregator = new UsageAggregator({
+      timeZone: "UTC",
+      sinceDay: "2026-08-01",
+      untilDay: "2026-08-31",
+      rates,
+      threadIdsBySession: new Map([["claude:session-linked", ["thread-a"]]]),
+    });
+    aggregator.add(record({ sessionId: "session-unlinked", dedupeKey: "copied-inference" }));
+    aggregator.add(record({ sessionId: "session-linked", dedupeKey: "copied-inference" }));
+    aggregator.add(record({ sessionId: "session-linked", dedupeKey: "copied-inference" }));
+    const result = aggregator.finish();
+
+    expect(result.buckets).toHaveLength(1);
+    expect(result.buckets[0]?.records).toBe(1);
+    expect(result.duplicatesDropped).toBe(2);
+    expect(result.threadUsage).toEqual([
+      expect.objectContaining({
+        threadId: "thread-a",
+        totals: expect.objectContaining({ outputTokens: 50 }),
+      }),
+    ]);
+  });
   it("splits a bucket's cost by category and speed", () => {
     const [bucket] = aggregate([record(), record({ speed: "fast" })]).buckets;
 

apps/server/src/usage/usageAggregation.ts

@@ -7,6 +7,7 @@ import type {
   UsageModelSelector,
   UsageResolution,
   UsageTokenTotals,
+  UsageThreadTotals,
 } from "@t3tools/contracts";
 
 import { EMPTY_TOTALS, type UsageRecord } from "@t3tools/provider-core/server/usage";
@@ -64,6 +65,7 @@ interface MutableBucket {
 }
 
 export interface AggregateOptions {
+  readonly threadIdsBySession?: ReadonlyMap<string, readonly string[]>;
   readonly timeZone: string;
   readonly sinceDay: string;
   readonly untilDay: string;
@@ -125,6 +127,7 @@ export function canonicalModelFilter(
 }
 
 export interface AggregateResult {
+  readonly threadUsage?: readonly UsageThreadTotals[];
   readonly buckets: readonly UsageBucket[];
   readonly duplicatesDropped: number;
   readonly outOfWindow: number;
@@ -172,8 +175,10 @@ interface Contribution {
 
 /** Deduplicates across the whole scan; Claude copies records into resumed/forked transcripts. */
 export class UsageAggregator {
+  readonly #threadUsage = new Map<string, UsageThreadTotals>();
   readonly #buckets = new Map<string, MutableBucket>();
   readonly #seen = new Map<string, Contribution>();
+  readonly #threadSeen = new Map<string, Set<string>>();
   readonly #toDay: (timestampMs: number) => string;
   readonly #hourlyWindow: { readonly sinceTimeMs: number; readonly untilTimeMs: number } | null;
   readonly #options: AggregateOptions;
@@ -207,6 +212,8 @@ export class UsageAggregator {
   add(input: UsageRecord, sourcePath?: string): boolean {
     if (!keepsUsageRecord(this.#options.modelFilter, input)) return false;
     const record = this.#mapModel(input);
+    // Counted before the scan-wide dedupe: a copy in another thread's transcript still belongs to it.
+    this.#addThreadUsage(record);
     if (record.dedupeKey !== null) {
       const held = this.#seen.get(record.dedupeKey);
       if (held !== undefined) {
@@ -228,7 +235,46 @@ export class UsageAggregator {
     return true;
   }
 
-  #placeInWindow(record: UsageRecord, source: string): MutableBucket | null {
+  #addThreadUsage(record: UsageRecord): void {
+    const threadIds =
+      this.#options.threadIdsBySession?.get(`${record.provider}:${record.sessionId}`) ?? [];
+    if (threadIds.length === 0 || this.#dayInWindow(record) === null) return;
+    const priced = priceUsage(this.#options.rates, record, this.#options.priceOverrides);
+    const costs = priced.categoryCostUsd;
+    for (const threadId of threadIds) {
+      if (record.dedupeKey !== null) {
+        const seen = this.#threadSeen.get(threadId);
+        if (seen?.has(record.dedupeKey)) continue;
+        if (seen) seen.add(record.dedupeKey);
+        else this.#threadSeen.set(threadId, new Set([record.dedupeKey]));
+      }
+      const previous = this.#threadUsage.get(threadId);
+      const add = <K extends keyof UsageTokenTotals>(key: K) =>
+        (previous?.totals[key] ?? 0) + record.totals[key];
+      this.#threadUsage.set(threadId, {
+        threadId,
+        totals: {
+          uncachedInputTokens: add("uncachedInputTokens"),
+          cachedInputTokens: add("cachedInputTokens"),
+          cacheCreationTokens: add("cacheCreationTokens"),
+          outputTokens: add("outputTokens"),
+          reasoningTokens: add("reasoningTokens"),
+        },
+        costUsd: (previous?.costUsd ?? 0) + priced.costUsd,
+        categoryCostUsd: {
+          input: (previous?.categoryCostUsd.input ?? 0) + (costs?.input ?? 0),
+          cacheRead: (previous?.categoryCostUsd.cacheRead ?? 0) + (costs?.cacheRead ?? 0),
+          cacheWrite: (previous?.categoryCostUsd.cacheWrite ?? 0) + (costs?.cacheWrite ?? 0),
+          output: (previous?.categoryCostUsd.output ?? 0) + (costs?.output ?? 0),
+        },
+        unsplitCostUsd: (previous?.unsplitCostUsd ?? 0) + (costs === null ? priced.costUsd : 0),
+        unpricedRecords:
+          (previous?.unpricedRecords ?? 0) + (priced.costSource === "unpriced" ? 1 : 0),
+      });
+    }
+  }
+
+  #dayInWindow(record: UsageRecord): string | null {
     if (
       this.#hourlyWindow !== null &&
       (record.timestampMs < this.#hourlyWindow.sinceTimeMs ||
@@ -244,6 +290,12 @@ export class UsageAggregator {
     ) {
       return null;
     }
+    return day;
+  }
+
+  #placeInWindow(record: UsageRecord, source: string): MutableBucket | null {
+    const day = this.#dayInWindow(record);
+    if (day === null) return null;
 
     const hourIndex =
       this.#hourlyWindow === null
@@ -400,6 +452,9 @@ export class UsageAggregator {
 
     return {
       buckets,
+      ...(this.#options.threadIdsBySession === undefined
+        ? {}
+        : { threadUsage: [...this.#threadUsage.values()] }),
       duplicatesDropped: this.#duplicatesDropped,
       outOfWindow: this.#outOfWindow,
     };