refactor(gateway): share transcript usage aggregation (#129199)

* refactor(gateway): share transcript usage aggregation

* refactor(gateway): preserve transcript usage type ownership

config/assertion-safety-baseline.txt

@@ -3095,7 +3095,7 @@ src/gateway/session-transcript-title-reader.ts	1
 src/gateway/session-utils-core.ts	1
 src/gateway/session-utils-row.ts	1
 src/gateway/session-utils-search.ts	1
-src/gateway/session-utils.fs.ts	16
+src/gateway/session-utils.fs.ts	10
 src/gateway/sessions-history-http.ts	1
 src/gateway/stale-install.ts	1
 src/gateway/system-ca-warmup.ts	1

src/gateway/session-transcript-derived-readers.ts

@@ -1,12 +1,40 @@
+import {
+  estimateStringChars,
+  estimateTokensFromChars,
+} from "@openclaw/normalization-core/cjk-chars";
+import {
+  asNonNegativeFiniteNumber,
+  asPositiveFiniteNumber,
+} from "@openclaw/normalization-core/number-coercion";
+import { isRecord } from "@openclaw/normalization-core/record-coerce";
 import {
   deriveSessionTotalTokens,
   hasNonzeroUsage,
   normalizeUsage,
+  type ContextUsage,
   type UsageLike,
 } from "../agents/usage.js";
-import type { SessionTranscriptUsageSnapshot } from "./session-utils.fs.js";
 
-function extractSqliteUsageSnapshot(message: unknown): SessionTranscriptUsageSnapshot | null {
+export type SessionTranscriptUsageSnapshot = {
+  modelProvider?: string;
+  model?: string;
+  inputTokens?: number;
+  outputTokens?: number;
+  cacheRead?: number;
+  cacheWrite?: number;
+  contextUsage?: ContextUsage;
+  trailingBytes?: number;
+  totalTokens?: number;
+  totalTokensFresh?: boolean;
+  costUsd?: number;
+};
+
+type TranscriptUsageSource = "sqlite" | "artifact";
+
+function extractTranscriptUsageSnapshot(
+  message: unknown,
+  source: TranscriptUsageSource,
+): SessionTranscriptUsageSnapshot | null {
   if (!message || typeof message !== "object" || Array.isArray(message)) {
     return null;
   }
@@ -14,26 +42,38 @@ function extractSqliteUsageSnapshot(message: unknown): SessionTranscriptUsageSna
     api?: unknown;
     model?: unknown;
     provider?: unknown;
+    role?: unknown;
     usage?: unknown;
   };
+  if (source === "artifact" && typeof record.role === "string" && record.role !== "assistant") {
+    return null;
+  }
   const usageRaw =
     record.usage && typeof record.usage === "object" && !Array.isArray(record.usage)
       ? (record.usage as UsageLike & { cost?: { total?: unknown }; costUsd?: unknown })
       : undefined;
   const usage = normalizeUsage(usageRaw);
   const normalizedUsage = usage ?? {};
-  const legacyCliUsage = record.api === "cli" && usageRaw && usageRaw.contextUsage === undefined;
-  const totalTokens = legacyCliUsage ? undefined : deriveSessionTotalTokens({ usage });
+  const api =
+    source === "artifact" && typeof record.api === "string" ? record.api.trim() : record.api;
+  const legacyCliUsage = api === "cli" && usageRaw && usageRaw.contextUsage === undefined;
+  const derivedTotalTokens = legacyCliUsage ? undefined : deriveSessionTotalTokens({ usage });
+  const totalTokens =
+    source === "artifact" ? asPositiveFiniteNumber(derivedTotalTokens) : derivedTotalTokens;
   const modelProvider = typeof record.provider === "string" ? record.provider.trim() : undefined;
   const model = typeof record.model === "string" ? record.model.trim() : undefined;
   const costUsd =
-    typeof usageRaw?.cost?.total === "number" && Number.isFinite(usageRaw.cost.total)
-      ? usageRaw.cost.total
-      : usageRaw?.costUsd;
+    source === "artifact"
+      ? asNonNegativeFiniteNumber(usageRaw?.cost?.total)
+      : typeof usageRaw?.cost?.total === "number" && Number.isFinite(usageRaw.cost.total)
+        ? usageRaw.cost.total
+        : usageRaw?.costUsd;
   const hasMeaningfulUsage =
     hasNonzeroUsage(usage) ||
     typeof totalTokens === "number" ||
-    (typeof costUsd === "number" && Number.isFinite(costUsd) && costUsd > 0);
+    (typeof costUsd === "number" &&
+      Number.isFinite(costUsd) &&
+      (source === "artifact" || costUsd > 0));
   const isDeliveryMirror = modelProvider === "openclaw" && model === "delivery-mirror";
   if (!hasMeaningfulUsage && !modelProvider && !model) {
     return null;
@@ -62,8 +102,39 @@ function extractSqliteUsageSnapshot(message: unknown): SessionTranscriptUsageSna
   };
 }
 
-export function aggregateSqliteUsageSnapshots(
+function estimateTranscriptMessageChars(message: unknown): number {
+  if (!isRecord(message)) {
+    return 0;
+  }
+  const content = message.content;
+  if (typeof content === "string") {
+    return content.trim() ? estimateStringChars(content.trim()) : 0;
+  }
+  if (!Array.isArray(content)) {
+    return 0;
+  }
+  return content.reduce<number>((total, part) => {
+    if (!isRecord(part)) {
+      return total;
+    }
+    const { text, type } = part;
+    if (
+      typeof text !== "string" ||
+      (typeof type === "string" &&
+        type !== "text" &&
+        type !== "output_text" &&
+        type !== "input_text")
+    ) {
+      return total;
+    }
+    const normalized = text.trim();
+    return normalized ? total + estimateStringChars(normalized) : total;
+  }, 0);
+}
+
+export function aggregateSessionTranscriptUsage(
   messages: unknown[],
+  source: TranscriptUsageSource = "sqlite",
 ): SessionTranscriptUsageSnapshot | null {
   const aggregate: SessionTranscriptUsageSnapshot = {};
   let sawUsage = false;
@@ -77,8 +148,23 @@ export function aggregateSqliteUsageSnapshots(
   let sawCacheRead = false;
   let sawCacheWrite = false;
   let sawCost = false;
+  let estimatedTranscriptChars = 0;
+  let sawEstimateModelIdentity = false;
   for (const message of messages) {
-    const snapshot = extractSqliteUsageSnapshot(message);
+    if (source === "artifact" && isRecord(message)) {
+      const provider = typeof message.provider === "string" ? message.provider.trim() : undefined;
+      const model = typeof message.model === "string" ? message.model.trim() : undefined;
+      if (
+        (message.role === "user" || message.role === "assistant") &&
+        !(message.role === "assistant" && provider === "openclaw" && model === "delivery-mirror")
+      ) {
+        const estimatedChars = estimateTranscriptMessageChars(message);
+        estimatedTranscriptChars += estimatedChars;
+        sawEstimateModelIdentity ||=
+          message.role === "assistant" && estimatedChars > 0 && Boolean(provider || model);
+      }
+    }
+    const snapshot = extractTranscriptUsageSnapshot(message, source);
     if (!snapshot) {
       continue;
     }
@@ -142,5 +228,18 @@ export function aggregateSqliteUsageSnapshots(
   if (sawCost) {
     aggregate.costUsd = costUsd;
   }
+  if (
+    source === "artifact" &&
+    typeof aggregate.totalTokens !== "number" &&
+    aggregate.contextUsage?.state !== "unavailable" &&
+    estimatedTranscriptChars > 0 &&
+    sawEstimateModelIdentity
+  ) {
+    const estimatedTotalTokens = estimateTokensFromChars(estimatedTranscriptChars);
+    if (estimatedTotalTokens > 0) {
+      aggregate.totalTokens = estimatedTotalTokens;
+      aggregate.totalTokensFresh = true;
+    }
+  }
   return aggregate;
 }

src/gateway/session-transcript-readers.ts

@@ -19,7 +19,7 @@ import {
   readSessionTranscriptHistoryEvents,
 } from "../config/sessions/session-accessor.sqlite-history-events.js";
 import { resolveAgentIdFromSessionKey } from "../routing/session-key.js";
-import { aggregateSqliteUsageSnapshots } from "./session-transcript-derived-readers.js";
+import { aggregateSessionTranscriptUsage } from "./session-transcript-derived-readers.js";
 import {
   attachOpenClawTranscriptMeta,
   projectTranscriptEntryMessage,
@@ -238,7 +238,7 @@ export function sqliteMessageEventWithSeq(entry: SessionTranscriptMessageEvent):
 function readSqliteAggregateUsageSnapshot(
   target: ResolvedTranscriptReadTarget,
 ): SessionTranscriptUsageSnapshot | null {
-  return aggregateSqliteUsageSnapshots(readSqliteMessagesSync(target));
+  return aggregateSessionTranscriptUsage(readSqliteMessagesSync(target));
 }
 
 function buildSqlitePreviewItems(
@@ -444,7 +444,7 @@ export function readRecentSessionUsageFromTranscript(
   maxBytes: number,
 ): SessionTranscriptUsageSnapshot | null {
   const target = resolveTranscriptReadTarget(scope);
-  return aggregateSqliteUsageSnapshots(readRecentSqliteUsageMessages(target, maxBytes));
+  return aggregateSessionTranscriptUsage(readRecentSqliteUsageMessages(target, maxBytes));
 }
 
 /** Reads compact session preview items through the reader seam. */

src/gateway/session-utils.fs.test.ts

@@ -1534,6 +1534,23 @@ describe("readLatestSessionUsageFromTranscript", () => {
     }
   });
 
+  test.each([
+    { name: "an unattributed assistant", identity: {} },
+    {
+      name: "a delivery mirror",
+      identity: { provider: "openclaw", model: "delivery-mirror" },
+    },
+  ])("retains a meaningful zero-cost artifact snapshot for $name", async ({ identity }) => {
+    const sessionId = "usage-zero-cost-artifact";
+    writeTranscript(tmpDir, sessionId, [
+      { message: { role: "assistant", ...identity, usage: { cost: { total: 0 } } } },
+    ]);
+
+    await expect(
+      readLatestSessionUsageFromTranscriptFileAsync(sessionId, storePath),
+    ).resolves.toEqual({ costUsd: 0 });
+  });
+
   test("treats unavailable JSONL context as terminal until a later valid snapshot", async () => {
     const sessionId = "usage-unavailable-upgrade-sequence";
     const oldCumulative = {

src/gateway/session-utils.fs.ts

@@ -4,22 +4,9 @@ import fs from "node:fs";
 import readline from "node:readline";
 import { expectDefined } from "@openclaw/normalization-core";
 import {
-  estimateStringChars,
-  estimateTokensFromChars,
-} from "@openclaw/normalization-core/cjk-chars";
-import {
-  asNonNegativeFiniteNumber,
-  asPositiveFiniteNumber as resolvePositiveUsageNumber,
   resolveIntegerOption,
   resolveNonNegativeIntegerOption,
 } from "@openclaw/normalization-core/number-coercion";
-import {
-  deriveSessionTotalTokens,
-  hasNonzeroUsage,
-  normalizeUsage,
-  type ContextUsage,
-  type UsageLike,
-} from "../agents/usage.js";
 import { materializeSessionArchiveForRead } from "../config/sessions/archive-compression.js";
 import type { TranscriptEvent } from "../config/sessions/session-accessor.js";
 import { streamSessionTranscriptLines } from "../config/sessions/transcript-stream.js";
@@ -28,6 +15,10 @@ import { readFileWindowFully } from "../infra/file-read.js";
 import { jsonUtf8Bytes } from "../infra/json-utf8-bytes.js";
 import { pruneMapToMaxSize } from "../infra/map-size.js";
 import { projectSessionDisplayMessage } from "./session-display-projection.js";
+import {
+  aggregateSessionTranscriptUsage,
+  type SessionTranscriptUsageSnapshot,
+} from "./session-transcript-derived-readers.js";
 import {
   resolveSessionTranscriptCandidates,
   resolveSessionTranscriptResetArchiveCandidatesAsync,
@@ -44,6 +35,8 @@ import {
 } from "./session-transcript-message.js";
 import type { SessionPreviewItem } from "./session-utils.types.js";
 
+export type { SessionTranscriptUsageSnapshot } from "./session-transcript-derived-readers.js";
+
 export type ReadRecentSessionMessagesOptions = {
   maxMessages: number;
   maxBytes?: number;
@@ -880,315 +873,6 @@ export async function resolveSessionHistoryTranscriptPathAsync(
   });
 }
 
-export type SessionTranscriptUsageSnapshot = {
-  modelProvider?: string;
-  model?: string;
-  inputTokens?: number;
-  outputTokens?: number;
-  cacheRead?: number;
-  cacheWrite?: number;
-  contextUsage?: ContextUsage;
-  trailingBytes?: number;
-  totalTokens?: number;
-  totalTokensFresh?: boolean;
-  costUsd?: number;
-};
-
-function extractTranscriptUsageCost(raw: unknown): number | undefined {
-  if (!raw || typeof raw !== "object" || Array.isArray(raw)) {
-    return undefined;
-  }
-  const cost = (raw as { cost?: unknown }).cost;
-  if (!cost || typeof cost !== "object" || Array.isArray(cost)) {
-    return undefined;
-  }
-  const total = (cost as { total?: unknown }).total;
-  return asNonNegativeFiniteNumber(total);
-}
-
-function extractTranscriptContentEstimatedChars(content: unknown): number {
-  if (typeof content === "string") {
-    const normalized = content.trim();
-    return normalized ? estimateStringChars(normalized) : 0;
-  }
-  if (!Array.isArray(content)) {
-    return 0;
-  }
-  let chars = 0;
-  for (const part of content) {
-    if (!part || typeof part !== "object" || Array.isArray(part)) {
-      continue;
-    }
-    const record = part as Record<string, unknown>;
-    if (typeof record.text !== "string") {
-      continue;
-    }
-    const type = typeof record.type === "string" ? record.type : "text";
-    if (type !== "text" && type !== "output_text" && type !== "input_text") {
-      continue;
-    }
-    const normalized = record.text.trim();
-    if (normalized) {
-      chars += estimateStringChars(normalized);
-    }
-  }
-  return chars;
-}
-
-function extractTranscriptTokenEstimateFromLine(line: string): {
-  estimatedChars: number;
-  hasModelIdentity: boolean;
-} | null {
-  if (isOversizedTranscriptLine(line)) {
-    return null;
-  }
-  try {
-    const parsed = JSON.parse(line) as Record<string, unknown>;
-    const message =
-      parsed.message && typeof parsed.message === "object" && !Array.isArray(parsed.message)
-        ? (parsed.message as Record<string, unknown>)
-        : undefined;
-    if (!message) {
-      return null;
-    }
-    const role = typeof message.role === "string" ? message.role : undefined;
-    if (role !== "user" && role !== "assistant") {
-      return null;
-    }
-    const modelProvider =
-      typeof message.provider === "string"
-        ? message.provider.trim()
-        : typeof parsed.provider === "string"
-          ? parsed.provider.trim()
-          : undefined;
-    const model =
-      typeof message.model === "string"
-        ? message.model.trim()
-        : typeof parsed.model === "string"
-          ? parsed.model.trim()
-          : undefined;
-    const isDeliveryMirror =
-      role === "assistant" && modelProvider === "openclaw" && model === "delivery-mirror";
-    if (isDeliveryMirror) {
-      return null;
-    }
-    const contentChars = extractTranscriptContentEstimatedChars(message.content);
-    if (contentChars <= 0) {
-      return null;
-    }
-    return {
-      estimatedChars: contentChars,
-      hasModelIdentity: role === "assistant" && Boolean(modelProvider || model),
-    };
-  } catch {
-    return null;
-  }
-}
-
-function extractUsageSnapshotFromTranscriptLine(
-  line: string,
-): SessionTranscriptUsageSnapshot | null {
-  if (isOversizedTranscriptLine(line)) {
-    return null;
-  }
-  try {
-    const parsed = JSON.parse(line) as Record<string, unknown>;
-    const message =
-      parsed.message && typeof parsed.message === "object" && !Array.isArray(parsed.message)
-        ? (parsed.message as Record<string, unknown>)
-        : undefined;
-    if (!message) {
-      return null;
-    }
-    const role = typeof message.role === "string" ? message.role : undefined;
-    if (role && role !== "assistant") {
-      return null;
-    }
-    const usageRaw =
-      message.usage && typeof message.usage === "object" && !Array.isArray(message.usage)
-        ? message.usage
-        : parsed.usage && typeof parsed.usage === "object" && !Array.isArray(parsed.usage)
-          ? parsed.usage
-          : undefined;
-    const usageRecord = usageRaw as UsageLike | undefined;
-    const usage = normalizeUsage(usageRecord);
-    const api = typeof message.api === "string" ? message.api.trim() : undefined;
-    const legacyCliUsage =
-      api === "cli" && usageRecord !== undefined && usageRecord.contextUsage === undefined;
-    const totalTokens = legacyCliUsage
-      ? undefined
-      : resolvePositiveUsageNumber(deriveSessionTotalTokens({ usage }));
-    const costUsd = extractTranscriptUsageCost(usageRaw);
-    const modelProvider =
-      typeof message.provider === "string"
-        ? message.provider.trim()
-        : typeof parsed.provider === "string"
-          ? parsed.provider.trim()
-          : undefined;
-    const model =
-      typeof message.model === "string"
-        ? message.model.trim()
-        : typeof parsed.model === "string"
-          ? parsed.model.trim()
-          : undefined;
-    const isDeliveryMirror = modelProvider === "openclaw" && model === "delivery-mirror";
-    const hasMeaningfulUsage =
-      hasNonzeroUsage(usage) ||
-      typeof totalTokens === "number" ||
-      (typeof costUsd === "number" && Number.isFinite(costUsd));
-    const hasModelIdentity = Boolean(modelProvider || model);
-    if (!hasMeaningfulUsage && !hasModelIdentity) {
-      return null;
-    }
-    if (isDeliveryMirror && !hasMeaningfulUsage) {
-      return null;
-    }
-
-    const snapshot: SessionTranscriptUsageSnapshot = {};
-    if (!isDeliveryMirror) {
-      if (modelProvider) {
-        snapshot.modelProvider = modelProvider;
-      }
-      if (model) {
-        snapshot.model = model;
-      }
-    }
-    if (typeof usage?.input === "number" && Number.isFinite(usage.input)) {
-      snapshot.inputTokens = usage.input;
-    }
-    if (typeof usage?.output === "number" && Number.isFinite(usage.output)) {
-      snapshot.outputTokens = usage.output;
-    }
-    if (typeof usage?.cacheRead === "number" && Number.isFinite(usage.cacheRead)) {
-      snapshot.cacheRead = usage.cacheRead;
-    }
-    if (typeof usage?.cacheWrite === "number" && Number.isFinite(usage.cacheWrite)) {
-      snapshot.cacheWrite = usage.cacheWrite;
-    }
-    if (legacyCliUsage) {
-      snapshot.contextUsage = { state: "unavailable" };
-    } else if (usage?.contextUsage) {
-      snapshot.contextUsage = usage.contextUsage;
-    }
-    if (typeof totalTokens === "number") {
-      snapshot.totalTokens = totalTokens;
-      snapshot.totalTokensFresh = true;
-    }
-    if (typeof costUsd === "number" && Number.isFinite(costUsd)) {
-      snapshot.costUsd = costUsd;
-    }
-    return snapshot;
-  } catch {
-    return null;
-  }
-}
-
-function extractAggregateUsageFromTranscriptLines(
-  lines: Iterable<string>,
-): SessionTranscriptUsageSnapshot | null {
-  const snapshot: SessionTranscriptUsageSnapshot = {};
-  let sawSnapshot = false;
-  let inputTokens = 0;
-  let outputTokens = 0;
-  let cacheRead = 0;
-  let cacheWrite = 0;
-  let sawInputTokens = false;
-  let sawOutputTokens = false;
-  let sawCacheRead = false;
-  let sawCacheWrite = false;
-  let costUsdTotal = 0;
-  let sawCost = false;
-  let estimatedTranscriptChars = 0;
-  let sawEstimatedTranscriptContent = false;
-  let sawEstimateModelIdentity = false;
-
-  for (const line of lines) {
-    const estimate = extractTranscriptTokenEstimateFromLine(line);
-    if (estimate) {
-      estimatedTranscriptChars += estimate.estimatedChars;
-      sawEstimatedTranscriptContent = true;
-      sawEstimateModelIdentity ||= estimate.hasModelIdentity;
-    }
-    const current = extractUsageSnapshotFromTranscriptLine(line);
-    if (!current) {
-      continue;
-    }
-    sawSnapshot = true;
-    if (current.modelProvider) {
-      snapshot.modelProvider = current.modelProvider;
-    }
-    if (current.model) {
-      snapshot.model = current.model;
-    }
-    if (typeof current.inputTokens === "number") {
-      inputTokens += current.inputTokens;
-      sawInputTokens = true;
-    }
-    if (typeof current.outputTokens === "number") {
-      outputTokens += current.outputTokens;
-      sawOutputTokens = true;
-    }
-    if (typeof current.cacheRead === "number") {
-      cacheRead += current.cacheRead;
-      sawCacheRead = true;
-    }
-    if (typeof current.cacheWrite === "number") {
-      cacheWrite += current.cacheWrite;
-      sawCacheWrite = true;
-    }
-    if (current.contextUsage) {
-      snapshot.contextUsage = current.contextUsage;
-    } else if (typeof current.totalTokens === "number") {
-      delete snapshot.contextUsage;
-    }
-    if (current.contextUsage?.state === "unavailable") {
-      // Unavailable invalidates every older total; only a later numeric snapshot
-      // may restore freshness as the forward scan continues.
-      delete snapshot.totalTokens;
-      delete snapshot.totalTokensFresh;
-    } else if (typeof current.totalTokens === "number") {
-      snapshot.totalTokens = current.totalTokens;
-      snapshot.totalTokensFresh = true;
-    }
-    if (typeof current.costUsd === "number" && Number.isFinite(current.costUsd)) {
-      costUsdTotal += current.costUsd;
-      sawCost = true;
-    }
-  }
-
-  if (!sawSnapshot) {
-    return null;
-  }
-  if (sawInputTokens) {
-    snapshot.inputTokens = inputTokens;
-  }
-  if (sawOutputTokens) {
-    snapshot.outputTokens = outputTokens;
-  }
-  if (sawCacheRead) {
-    snapshot.cacheRead = cacheRead;
-  }
-  if (sawCacheWrite) {
-    snapshot.cacheWrite = cacheWrite;
-  }
-  if (sawCost) {
-    snapshot.costUsd = costUsdTotal;
-  }
-  if (
-    typeof snapshot.totalTokens !== "number" &&
-    snapshot.contextUsage?.state !== "unavailable" &&
-    sawEstimatedTranscriptContent &&
-    sawEstimateModelIdentity
-  ) {
-    const estimatedTotalTokens = estimateTokensFromChars(estimatedTranscriptChars);
-    if (estimatedTotalTokens > 0) {
-      snapshot.totalTokens = estimatedTotalTokens;
-      snapshot.totalTokensFresh = true;
-    }
-  }
-  return snapshot;
-}
-
 export async function readLatestSessionUsageFromTranscriptFileAsync(
   sessionId: string,
   storePath: string | undefined,
@@ -1205,11 +889,40 @@ export async function readLatestSessionUsageFromTranscriptFileAsync(
     if (stat.size === 0) {
       return null;
     }
-    const lines: string[] = [];
+    const messages: unknown[] = [];
     for await (const line of streamSessionTranscriptLines(filePath)) {
-      lines.push(line);
+      if (isOversizedTranscriptLine(line)) {
+        continue;
+      }
+      try {
+        const record = JSON.parse(line) as Record<string, unknown>;
+        if (
+          !record.message ||
+          typeof record.message !== "object" ||
+          Array.isArray(record.message)
+        ) {
+          continue;
+        }
+        const message = record.message as Record<string, unknown>;
+        const usage =
+          message.usage && typeof message.usage === "object" && !Array.isArray(message.usage)
+            ? message.usage
+            : record.usage;
+        messages.push({
+          ...message,
+          ...(typeof message.provider !== "string" && typeof record.provider === "string"
+            ? { provider: record.provider }
+            : {}),
+          ...(typeof message.model !== "string" && typeof record.model === "string"
+            ? { model: record.model }
+            : {}),
+          ...(usage && typeof usage === "object" && !Array.isArray(usage) ? { usage } : {}),
+        });
+      } catch {
+        continue;
+      }
     }
-    return extractAggregateUsageFromTranscriptLines(lines);
+    return aggregateSessionTranscriptUsage(messages, "artifact");
   } catch {
     return null;
   }