diff --git a/config/assertion-safety-baseline.txt b/config/assertion-safety-baseline.txt index 08d8a11c8b5f..789b95b2b7e8 100644 --- a/config/assertion-safety-baseline.txt +++ b/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 diff --git a/src/gateway/session-transcript-derived-readers.ts b/src/gateway/session-transcript-derived-readers.ts index 0dc66d9b6ff5..1758154b3704 100644 --- a/src/gateway/session-transcript-derived-readers.ts +++ b/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((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; } diff --git a/src/gateway/session-transcript-readers.ts b/src/gateway/session-transcript-readers.ts index ecf265d171f6..b9eed986f8d9 100644 --- a/src/gateway/session-transcript-readers.ts +++ b/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. */ diff --git a/src/gateway/session-utils.fs.test.ts b/src/gateway/session-utils.fs.test.ts index 1383725c9959..b102df4c213e 100644 --- a/src/gateway/session-utils.fs.test.ts +++ b/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 = { diff --git a/src/gateway/session-utils.fs.ts b/src/gateway/session-utils.fs.ts index 81457eb95d75..f1a7994deb54 100644 --- a/src/gateway/session-utils.fs.ts +++ b/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; - 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; - const message = - parsed.message && typeof parsed.message === "object" && !Array.isArray(parsed.message) - ? (parsed.message as Record) - : 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; - const message = - parsed.message && typeof parsed.message === "object" && !Array.isArray(parsed.message) - ? (parsed.message as Record) - : 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, -): 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; + if ( + !record.message || + typeof record.message !== "object" || + Array.isArray(record.message) + ) { + continue; + } + const message = record.message as Record; + 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; }