From 39bbd8afd338dde0831dea3214cf17f80594ea44 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 4 Aug 2026 20:45:41 -0700 Subject: [PATCH] refactor(sessions): consolidate transcript and state helpers (#119435) --- src/context-engine/quarantine-health.ts | 13 ++- src/sessions/conversation-turns.ts | 26 +++--- src/sessions/session-chat-type.ts | 21 +---- src/sessions/transcript-events.ts | 37 +++------ .../user-turn-transcript.media-normalize.ts | 42 ++++------ src/sessions/user-turn-transcript.ts | 44 ++++------ .../openclaw-agent-db-session-migrations.ts | 81 ++++++++----------- .../openclaw-agent-db-session-provenance.ts | 24 ++---- src/transcripts/provider-registry.ts | 17 +--- src/transcripts/store-export-ownership.ts | 8 +- src/transcripts/store-sqlite.ts | 6 +- src/transcripts/store.ts | 21 ++--- 12 files changed, 117 insertions(+), 223 deletions(-) diff --git a/src/context-engine/quarantine-health.ts b/src/context-engine/quarantine-health.ts index 389956298fe4..24f437c88b27 100644 --- a/src/context-engine/quarantine-health.ts +++ b/src/context-engine/quarantine-health.ts @@ -1,5 +1,6 @@ // Persists context-engine runtime quarantines so health surfaces can see // failures recorded in sibling runtime processes. +import { hasNonEmptyString } from "@openclaw/normalization-core/string-coerce"; import { createRuntimeHealthRecordEnvelope, createRuntimeHealthStore, @@ -17,10 +18,6 @@ type PersistedContextEngineRuntimeQuarantine = { type PersistedContextEngineQuarantineRecord = RuntimeHealthRecordEnvelope & Omit; -function isNonEmptyString(value: unknown): value is string { - return typeof value === "string" && value.trim().length > 0; -} - // No TTL: a quarantine is recorded once per failure and stays valid for the // recorder's lifetime, so process liveness alone owns expiry here. const quarantineStore = createRuntimeHealthStore({ @@ -29,9 +26,9 @@ const quarantineStore = createRuntimeHealthStore { if ( - !isNonEmptyString(value.engineId) || - !isNonEmptyString(value.operation) || - !isNonEmptyString(value.reason) + !hasNonEmptyString(value.engineId) || + !hasNonEmptyString(value.operation) || + !hasNonEmptyString(value.reason) ) { return undefined; } @@ -43,7 +40,7 @@ const quarantineStore = createRuntimeHealthStore record.engineId, diff --git a/src/sessions/conversation-turns.ts b/src/sessions/conversation-turns.ts index 1bb236042357..057e1efbcebb 100644 --- a/src/sessions/conversation-turns.ts +++ b/src/sessions/conversation-turns.ts @@ -1,4 +1,5 @@ import crypto from "node:crypto"; +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { resolveGlobalSingleton } from "../shared/global-singleton.js"; type ConversationTurnReply = { @@ -54,11 +55,6 @@ const pendingTurns = resolveGlobalSingleton( } }, ); -function normalize(value: string | undefined): string | undefined { - const normalized = value?.trim(); - return normalized || undefined; -} - function pendingTurnKey(agentId: string, id: string): string { return JSON.stringify([agentId, id]); } @@ -73,11 +69,11 @@ export function registerPendingConversationTurn(params: { timeoutMs: number; signal?: AbortSignal; }): PendingConversationTurnHandle { - const agentId = normalize(params.agentId); + const agentId = normalizeOptionalString(params.agentId); if (!agentId) { throw new Error("conversation turn requires an agent id"); } - const id = normalize(params.id) ?? crypto.randomUUID(); + const id = normalizeOptionalString(params.id) ?? crypto.randomUUID(); const key = pendingTurnKey(agentId, id); if (pendingTurns.has(key)) { throw new Error(`conversation turn already pending for ${agentId}: ${id}`); @@ -128,7 +124,7 @@ export function registerPendingConversationTurn(params: { id, conversationRef: params.conversationRef, sessionId: params.sessionId, - threadId: normalize(params.threadId), + threadId: normalizeOptionalString(params.threadId), createdAt, correlationReady, markCorrelationReady, @@ -149,7 +145,7 @@ export function registerPendingConversationTurn(params: { if (pendingTurns.get(key) !== pending) { return; } - pending.outboundMessageId = normalize(messageId); + pending.outboundMessageId = normalizeOptionalString(messageId); if (!pending.outboundMessageId) { pending.settle(undefined); } @@ -171,8 +167,8 @@ export function registerPendingConversationTurn(params: { /** Cancels one Gateway-owned turn so a late reply follows ordinary inbound dispatch. */ export function cancelPendingConversationTurn(params: { agentId: string; id: string }): boolean { - const agentId = normalize(params.agentId); - const id = normalize(params.id); + const agentId = normalizeOptionalString(params.agentId); + const id = normalizeOptionalString(params.id); const pending = agentId && id ? pendingTurns.get(pendingTurnKey(agentId, id)) : undefined; if (!pending) { return false; @@ -193,13 +189,13 @@ export async function claimPendingConversationTurnReply(params: { text: string; timestamp?: number; }): Promise { - const replyToId = normalize(params.replyToId); + const replyToId = normalizeOptionalString(params.replyToId); if (!replyToId) { return undefined; } - const threadId = normalize(params.threadId); - const parentConversationRef = normalize(params.parentConversationRef); - const agentId = normalize(params.agentId); + const threadId = normalizeOptionalString(params.threadId); + const parentConversationRef = normalizeOptionalString(params.parentConversationRef); + const agentId = normalizeOptionalString(params.agentId); if (!agentId) { return undefined; } diff --git a/src/sessions/session-chat-type.ts b/src/sessions/session-chat-type.ts index d2d3dba0e198..3e2f363c3595 100644 --- a/src/sessions/session-chat-type.ts +++ b/src/sessions/session-chat-type.ts @@ -9,10 +9,6 @@ import { parseAgentSessionKey } from "./session-key-utils.js"; // Session chat-type derivation first uses generic key parsing, then falls back // to bootstrap channel plugins for legacy platform-specific session keys. -type LegacySessionChatTypeDeriver = NonNullable< - NonNullable>["messaging"] ->["deriveLegacySessionChatType"]; - function resolveScopedSessionKey(sessionKey: string | undefined | null): string { const raw = normalizeLowercaseStringOrEmpty(sessionKey); if (!raw) { @@ -34,16 +30,6 @@ function collectLegacyChatTypeCandidatePluginIds(scopedSessionKey: string): stri return Array.from(ids); } -function derivePluginLegacySessionChatType( - scopedSessionKey: string, - deriveLegacySessionChatType: LegacySessionChatTypeDeriver, -): SessionKeyChatType | undefined { - if (!deriveLegacySessionChatType) { - return undefined; - } - return deriveLegacySessionChatType(scopedSessionKey); -} - export function deriveSessionChatType(sessionKey: string | undefined | null): SessionKeyChatType { const builtInType = deriveSessionChatTypeFromKey(sessionKey); if (builtInType !== "unknown") { @@ -52,10 +38,9 @@ export function deriveSessionChatType(sessionKey: string | undefined | null): Se const scopedSessionKey = resolveScopedSessionKey(sessionKey); for (const pluginId of collectLegacyChatTypeCandidatePluginIds(scopedSessionKey)) { - const derived = derivePluginLegacySessionChatType( - scopedSessionKey, - getBootstrapChannelPlugin(pluginId)?.messaging?.deriveLegacySessionChatType, - ); + const deriveLegacySessionChatType = + getBootstrapChannelPlugin(pluginId)?.messaging?.deriveLegacySessionChatType; + const derived = deriveLegacySessionChatType?.(scopedSessionKey); if (derived) { return derived; } diff --git a/src/sessions/transcript-events.ts b/src/sessions/transcript-events.ts index 9ebc4dddef45..cf18c68669c0 100644 --- a/src/sessions/transcript-events.ts +++ b/src/sessions/transcript-events.ts @@ -68,7 +68,7 @@ export function onInternalSessionTranscriptUpdate( /** Emits a normalized transcript update to all registered listeners. */ export function emitSessionTranscriptUpdate(update: InternalSessionTranscriptUpdate): void { - const nextUpdate = normalizeSessionTranscriptUpdate(update, { allowIdentityOnly: true }); + const nextUpdate = normalizeSessionTranscriptUpdate(update); if (!nextUpdate) { return; } @@ -81,29 +81,18 @@ export function emitSessionTranscriptUpdate(update: InternalSessionTranscriptUpd function normalizeSessionTranscriptUpdate( update: InternalSessionTranscriptUpdate, - options: { allowIdentityOnly: boolean }, ): InternalSessionTranscriptUpdate | undefined { - const normalized = { - sessionFile: update.sessionFile, - target: update.target, - sessionKey: update.sessionKey, - agentId: update.agentId, - sessionId: update.sessionId, - lifecycleRevision: update.lifecycleRevision, - message: update.message, - messageId: update.messageId, - messageSeq: update.messageSeq, - }; - const trimmed = normalizeOptionalString(normalized.sessionFile); - const target = normalizeUpdateTarget(normalized); - if (!trimmed && (!options.allowIdentityOnly || !target)) { + const trimmed = normalizeOptionalString(update.sessionFile); + const target = normalizeUpdateTarget(update); + if (!trimmed && !target) { return undefined; } - const messageSeq = asPositiveSafeInteger(normalized.messageSeq); - const sessionKey = normalizeOptionalString(normalized.sessionKey) ?? target?.sessionKey; - const agentId = normalizeOptionalString(normalized.agentId) ?? target?.agentId; - const sessionId = normalizeOptionalString(normalized.sessionId) ?? target?.sessionId; - const lifecycleRevision = normalizeOptionalString(normalized.lifecycleRevision); + const messageSeq = asPositiveSafeInteger(update.messageSeq); + const sessionKey = normalizeOptionalString(update.sessionKey) ?? target?.sessionKey; + const agentId = normalizeOptionalString(update.agentId) ?? target?.agentId; + const sessionId = normalizeOptionalString(update.sessionId) ?? target?.sessionId; + const lifecycleRevision = normalizeOptionalString(update.lifecycleRevision); + const messageId = normalizeOptionalString(update.messageId); return { ...(trimmed ? { sessionFile: trimmed } : {}), ...(target ? { target } : {}), @@ -111,10 +100,8 @@ function normalizeSessionTranscriptUpdate( ...(agentId ? { agentId } : {}), ...(sessionId ? { sessionId } : {}), ...(lifecycleRevision ? { lifecycleRevision } : {}), - ...(normalized.message !== undefined ? { message: normalized.message } : {}), - ...(normalizeOptionalString(normalized.messageId) - ? { messageId: normalizeOptionalString(normalized.messageId) } - : {}), + ...(update.message !== undefined ? { message: update.message } : {}), + ...(messageId ? { messageId } : {}), ...(messageSeq !== undefined ? { messageSeq } : {}), }; } diff --git a/src/sessions/user-turn-transcript.media-normalize.ts b/src/sessions/user-turn-transcript.media-normalize.ts index 212801735bd9..3dd73606cf10 100644 --- a/src/sessions/user-turn-transcript.media-normalize.ts +++ b/src/sessions/user-turn-transcript.media-normalize.ts @@ -1,5 +1,10 @@ import path from "node:path"; import { mimeTypeFromFilePath } from "@openclaw/media-core/mime"; +import { + asFiniteNumberInRange, + asPositiveSafeInteger, +} from "@openclaw/normalization-core/number-coercion"; +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import type { MediaFactInput } from "../media/media-facts.js"; import type { PersistedUserTurnMediaInput } from "./user-turn-transcript.types.js"; @@ -14,26 +19,13 @@ const STRUCTURED_MEDIA_KINDS = new Set>([ ]); const MIME_TYPE_PATTERN = /^[a-z0-9!#$&^_.+-]+\/[a-z0-9!#$&^_.+-]+$/iu; -function normalizeOptionalText(value: string | null | undefined): string | undefined { - const normalized = value?.trim(); - return normalized ? normalized : undefined; -} - -function normalizeNonNegativeNumber(value: number | null | undefined): number | undefined { - return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : undefined; -} - function normalizeStructuredMediaKind(value: string | null | undefined): MediaFactInput["kind"] { - const kind = normalizeOptionalText(value); + const kind = normalizeOptionalString(value); return kind && STRUCTURED_MEDIA_KINDS.has(kind as NonNullable) ? (kind as NonNullable) : undefined; } -function normalizePositiveInteger(value: number | null | undefined): number | undefined { - return typeof value === "number" && Number.isSafeInteger(value) && value > 0 ? value : undefined; -} - export function resolveTranscriptMediaPath( pathValue: string, workspaceDir: string | undefined, @@ -49,21 +41,21 @@ export function resolveTranscriptMediaPath( export function normalizeStructuredMediaEntryForTranscript( media: PersistedUserTurnMediaInput, ): MediaFactInput { - const workspaceDir = normalizeOptionalText(media.workspaceDir); - const mediaPath = normalizeOptionalText(media.path); - const mediaUrl = normalizeOptionalText(media.url); + const workspaceDir = normalizeOptionalString(media.workspaceDir); + const mediaPath = normalizeOptionalString(media.path); + const mediaUrl = normalizeOptionalString(media.url); const kind = normalizeStructuredMediaKind(media.kind); - const legacyKind = normalizeOptionalText(media.kind); - const messageId = normalizeOptionalText(media.messageId); + const legacyKind = normalizeOptionalString(media.kind); + const messageId = normalizeOptionalString(media.messageId); const contentType = - normalizeOptionalText(media.contentType) ?? + normalizeOptionalString(media.contentType) ?? (kind || !legacyKind || !MIME_TYPE_PATTERN.test(legacyKind) ? undefined : legacyKind) ?? mimeTypeFromFilePath(mediaPath ?? mediaUrl); - const durationMs = normalizePositiveInteger(media.durationMs); - const width = normalizePositiveInteger(media.width); - const height = normalizePositiveInteger(media.height); - const fileName = normalizeOptionalText(media.fileName); - const sizeBytes = normalizeNonNegativeNumber(media.sizeBytes); + const durationMs = asPositiveSafeInteger(media.durationMs); + const width = asPositiveSafeInteger(media.width); + const height = asPositiveSafeInteger(media.height); + const fileName = normalizeOptionalString(media.fileName); + const sizeBytes = asFiniteNumberInRange(media.sizeBytes, { min: 0 }); return { ...(mediaPath ? { path: resolveTranscriptMediaPath(mediaPath, workspaceDir) } : {}), ...(mediaUrl ? { url: mediaUrl } : {}), diff --git a/src/sessions/user-turn-transcript.ts b/src/sessions/user-turn-transcript.ts index 539bcf1fd043..bc8eaab48d5b 100644 --- a/src/sessions/user-turn-transcript.ts +++ b/src/sessions/user-turn-transcript.ts @@ -1,5 +1,7 @@ // User turn transcript helpers extract user-turn text from session transcripts. import { mimeTypeFromFilePath } from "@openclaw/media-core/mime"; +import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce"; +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import type { AgentMessage } from "../../packages/agent-core/src/types.js"; import { persistSessionTranscriptTurn, @@ -35,22 +37,9 @@ export function buildRunUserTurnIdempotencyKey(runId: string): string { return `${runId}:user`; } -function normalizeOptionalText(value: string | null | undefined): string | undefined { - const normalized = value?.trim(); - return normalized ? normalized : undefined; -} - -function normalizeTranscriptText(value: string | null | undefined): string { - return value ?? ""; -} - // Select normalized text for persisted user turns. export function resolvePersistedUserTurnText(value: string | null | undefined): string | undefined { - const normalized = normalizeOptionalText(value); - if (!normalized) { - return undefined; - } - return normalized; + return normalizeOptionalString(value); } function resolveTranscriptMediaType(params: { @@ -70,16 +59,16 @@ export function buildPersistedUserTurnMediaInputsFromFields( const facts = readPersistedMediaFacts(fields) ?? []; const normalizedMedia = facts.map((fact) => { - const rawPath = normalizeOptionalText(fact.path); + const rawPath = normalizeOptionalString(fact.path); const mediaPath = rawPath - ? resolveTranscriptMediaPath(rawPath, normalizeOptionalText(fact.workspaceDir)) + ? resolveTranscriptMediaPath(rawPath, normalizeOptionalString(fact.workspaceDir)) : undefined; - const url = normalizeOptionalText(fact.url); + const url = normalizeOptionalString(fact.url); if (!mediaPath && !url) { return {}; } const contentType = resolveTranscriptMediaType({ - explicitType: normalizeOptionalText(fact.contentType), + explicitType: normalizeOptionalString(fact.contentType), mediaPath, mediaUrl: url, }); @@ -131,9 +120,9 @@ export function buildLateMediaAttachedProjection(message: AgentMessage): { function buildUserTurnSenderMeta( sender: UserTurnInput["sender"], ): Record | undefined { - const senderId = normalizeOptionalText(sender?.id); - const senderName = normalizeOptionalText(sender?.name); - const senderUsername = normalizeOptionalText(sender?.username); + const senderId = normalizeOptionalString(sender?.id); + const senderName = normalizeOptionalString(sender?.name); + const senderUsername = normalizeOptionalString(sender?.username); if (!senderId && !senderName && !senderUsername) { return undefined; } @@ -145,15 +134,12 @@ function buildUserTurnSenderMeta( } function readOpenClawMessageMeta(message: AgentMessage): Record | undefined { - const meta = (message as unknown as Record)["__openclaw"]; - return meta && typeof meta === "object" && !Array.isArray(meta) - ? (meta as Record) - : undefined; + return asOptionalRecord((message as unknown as Record)["__openclaw"]); } export function buildPersistedUserTurnMessage(params: UserTurnInput): PersistedUserTurnMessage { const normalizedMedia = (params.media ?? []).map(normalizeStructuredMediaEntryForTranscript); - const text = normalizeTranscriptText(params.text); + const text = params.text ?? ""; // Storage is BARE (no timestamp prefix). The per-message timestamp is added // at the single LLM-boundary stamping site (normalizeMessagesForLlmBoundary), // derived from each message's own `timestamp` field, so the current turn and @@ -338,10 +324,8 @@ export function preparePersistedUserTurnMessageForTranscriptWrite( originalMediaImageLayout === undefined ? undefined : structuredClone(originalMediaImageLayout); // Hooks receive the original message object and may mutate nested metadata in // place. Snapshot transport correlation before handing them that reference. - const transport = - originalTransport && typeof originalTransport === "object" && !Array.isArray(originalTransport) - ? { ...originalTransport } - : undefined; + const originalTransportRecord = asOptionalRecord(originalTransport); + const transport = originalTransportRecord ? { ...originalTransportRecord } : undefined; const nextMessage = params.beforeMessageWrite({ message, ...(params.agentId ? { agentId: params.agentId } : {}), diff --git a/src/state/openclaw-agent-db-session-migrations.ts b/src/state/openclaw-agent-db-session-migrations.ts index 8b04ab9031eb..b8b838950d6f 100644 --- a/src/state/openclaw-agent-db-session-migrations.ts +++ b/src/state/openclaw-agent-db-session-migrations.ts @@ -1,4 +1,6 @@ import type { DatabaseSync } from "node:sqlite"; +import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce"; +import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; import { normalizeChatType, type ChatType } from "../channels/chat-type.js"; import { parseSqliteSessionEntryRecord } from "../config/sessions/session-entry-json.js"; import { normalizeAccountId } from "../routing/account-id.js"; @@ -7,29 +9,12 @@ import { deriveSessionChatTypeFromKey } from "../sessions/session-chat-type-shar type MigratedConversationEntry = Record; -function migratedObject( - entry: MigratedConversationEntry, - key: string, -): MigratedConversationEntry | undefined { - const value = entry[key]; - return value && typeof value === "object" && !Array.isArray(value) - ? (value as MigratedConversationEntry) - : undefined; -} - -function migratedText(value: unknown): string | undefined { - return typeof value === "string" && value.trim() ? value.trim() : undefined; -} - function parseConversationEntry(value: unknown): MigratedConversationEntry | undefined { if (typeof value !== "string") { return undefined; } try { - const parsed = JSON.parse(value) as unknown; - return parsed && typeof parsed === "object" && !Array.isArray(parsed) - ? (parsed as MigratedConversationEntry) - : undefined; + return asOptionalRecord(JSON.parse(value)); } catch { return undefined; } @@ -42,8 +27,8 @@ function inferMigratedChatType(params: { deliveryTarget?: string; }): ChatType { const explicit = - normalizeChatType(migratedText(params.entry.chatType)) ?? - normalizeChatType(migratedText(params.persistedChatType)); + normalizeChatType(normalizeOptionalString(params.entry.chatType)) ?? + normalizeChatType(normalizeOptionalString(params.persistedChatType)); if (explicit) { return explicit; } @@ -55,7 +40,10 @@ function inferMigratedChatType(params: { if (target?.startsWith("channel:") || /^[^:]+:channel:/u.test(target ?? "")) { return "channel"; } - if (/^(?:[^:]+:)?(?:group|room):/u.test(target ?? "") || migratedText(params.entry.groupId)) { + if ( + /^(?:[^:]+:)?(?:group|room):/u.test(target ?? "") || + normalizeOptionalString(params.entry.groupId) + ) { return "group"; } return "direct"; @@ -66,44 +54,43 @@ function migratedConversation( persistedChatType?: string, sessionKey?: string, ) { - const canonicalDelivery = migratedObject(entry, "delivery"); + const canonicalDelivery = asOptionalRecord(entry.delivery); const delivery = - migratedObject(canonicalDelivery ?? {}, "context") ?? migratedObject(entry, "deliveryContext"); - const origin = - migratedObject(canonicalDelivery ?? {}, "origin") ?? migratedObject(entry, "origin"); - const deliveryRouteTarget = migratedText(delivery?.to); + asOptionalRecord(canonicalDelivery?.context) ?? asOptionalRecord(entry.deliveryContext); + const origin = asOptionalRecord(canonicalDelivery?.origin) ?? asOptionalRecord(entry.origin); + const deliveryRouteTarget = normalizeOptionalString(delivery?.to); const kind = inferMigratedChatType({ entry, persistedChatType, sessionKey, - deliveryTarget: deliveryRouteTarget ?? migratedText(origin?.from), + deliveryTarget: deliveryRouteTarget ?? normalizeOptionalString(origin?.from), }); const deliveryTarget = - deliveryRouteTarget ?? (kind === "direct" ? migratedText(origin?.from) : undefined); + deliveryRouteTarget ?? (kind === "direct" ? normalizeOptionalString(origin?.from) : undefined); if (!deliveryTarget) { return undefined; } const routeOwnsTarget = Boolean(deliveryRouteTarget); const channel = ( routeOwnsTarget - ? (migratedText(delivery?.channel) ?? - migratedText(entry.channel) ?? - migratedText(entry.lastChannel) ?? - migratedText(origin?.provider)) - : migratedText(origin?.provider) + ? (normalizeOptionalString(delivery?.channel) ?? + normalizeOptionalString(entry.channel) ?? + normalizeOptionalString(entry.lastChannel) ?? + normalizeOptionalString(origin?.provider)) + : normalizeOptionalString(origin?.provider) )?.toLowerCase(); const accountId = normalizeAccountId( routeOwnsTarget - ? (migratedText(delivery?.accountId) ?? - migratedText(entry.lastAccountId) ?? - migratedText(origin?.accountId)) - : migratedText(origin?.accountId), + ? (normalizeOptionalString(delivery?.accountId) ?? + normalizeOptionalString(entry.lastAccountId) ?? + normalizeOptionalString(origin?.accountId)) + : normalizeOptionalString(origin?.accountId), ); const threadIdRaw = routeOwnsTarget ? delivery?.threadId : origin?.threadId; const threadId = typeof threadIdRaw === "number" && Number.isFinite(threadIdRaw) ? String(threadIdRaw) - : migratedText(threadIdRaw); + : normalizeOptionalString(threadIdRaw); // The routable target is authoritative for both identity and delivery. Stale // native metadata must never label one peer while sending to another. const peerId = channel ? normalizeConversationPeerId(channel, deliveryTarget) : undefined; @@ -121,13 +108,13 @@ function migratedConversation( peerId, deliveryTarget, threadId, - nativeChannelId: migratedText(origin?.nativeChannelId), - nativeDirectUserId: migratedText(origin?.nativeDirectUserId), + nativeChannelId: normalizeOptionalString(origin?.nativeChannelId), + nativeDirectUserId: normalizeOptionalString(origin?.nativeDirectUserId), label: - migratedText(entry.displayName) ?? - migratedText(entry.label) ?? - migratedText(entry.subject) ?? - migratedText(entry.groupId), + normalizeOptionalString(entry.displayName) ?? + normalizeOptionalString(entry.label) ?? + normalizeOptionalString(entry.subject) ?? + normalizeOptionalString(entry.groupId), }; } @@ -228,14 +215,14 @@ export function backfillSessionConversations(db: DatabaseSync): void { "UPDATE sessions SET primary_conversation_id = ? WHERE session_id = ?", ); for (const row of rows) { - const sessionId = migratedText(row.session_id); + const sessionId = normalizeOptionalString(row.session_id); const entry = parseConversationEntry(row.entry_json); const updatedAt = typeof row.updated_at === "number" ? row.updated_at : Date.now(); const conversation = entry ? migratedConversation( entry, - migratedText(row.persisted_chat_type), - migratedText(row.session_key), + normalizeOptionalString(row.persisted_chat_type), + normalizeOptionalString(row.session_key), ) : undefined; if (!sessionId || !conversation) { diff --git a/src/state/openclaw-agent-db-session-provenance.ts b/src/state/openclaw-agent-db-session-provenance.ts index 6d0aa9e69461..2b50fd146869 100644 --- a/src/state/openclaw-agent-db-session-provenance.ts +++ b/src/state/openclaw-agent-db-session-provenance.ts @@ -1,23 +1,16 @@ import type { DatabaseSync } from "node:sqlite"; +import { asOptionalRecord, isRecord } from "@openclaw/normalization-core/record-coerce"; +import { normalizeNullableString } from "@openclaw/normalization-core/string-coerce"; function readMigratedEntry(value: unknown): Record | undefined { if (typeof value === "string") { try { - const parsed: unknown = JSON.parse(value); - return parsed && typeof parsed === "object" && !Array.isArray(parsed) - ? (parsed as Record) - : undefined; + return asOptionalRecord(JSON.parse(value)); } catch { return undefined; } } - return value && typeof value === "object" && !Array.isArray(value) - ? (value as Record) - : undefined; -} - -function normalizedText(value: unknown): string | null { - return typeof value === "string" && value.trim() ? value.trim() : null; + return asOptionalRecord(value); } export function addSessionProvenanceColumns( @@ -72,16 +65,15 @@ export function backfillSessionEntryProvenance(db: DatabaseSync, previousVersion WHERE session_id = ?; `); for (const row of rows) { - const sessionId = normalizedText(row.session_id); + const sessionId = normalizeNullableString(row.session_id); const entry = readMigratedEntry(row.entry_json); if (!sessionId || !entry) { continue; } - const hookSource = normalizedText(entry.hookExternalContentSource); - const acp = entry.acp; + const hookSource = normalizeNullableString(entry.hookExternalContentSource); update.run( - acp && typeof acp === "object" && !Array.isArray(acp) ? 1 : 0, - normalizedText(entry.pluginOwnerId), + isRecord(entry.acp) ? 1 : 0, + normalizeNullableString(entry.pluginOwnerId), hookSource === "gmail" || hookSource === "webhook" ? hookSource : null, sessionId, ); diff --git a/src/transcripts/provider-registry.ts b/src/transcripts/provider-registry.ts index ca3957ed34cf..4fbcd522661a 100644 --- a/src/transcripts/provider-registry.ts +++ b/src/transcripts/provider-registry.ts @@ -23,23 +23,10 @@ export function normalizeTranscriptSourceProviderId( return normalizeCapabilityProviderId(providerId); } -function resolveTranscriptsSourceProviderEntries(cfg?: OpenClawConfig): TranscriptSourceProvider[] { - return resolvePluginCapabilityProviders({ - key: "transcriptSourceProviders", - cfg, - }); -} - -function buildProviderMaps(cfg?: OpenClawConfig): { - canonical: Map; - aliases: Map; -} { - return buildCapabilityProviderMaps(resolveTranscriptsSourceProviderEntries(cfg)); -} - /** List canonical transcript source providers for a config snapshot. */ export function listTranscriptSourceProviders(cfg?: OpenClawConfig): TranscriptSourceProvider[] { - return [...buildProviderMaps(cfg).canonical.values()]; + const providers = resolvePluginCapabilityProviders({ key: "transcriptSourceProviders", cfg }); + return [...buildCapabilityProviderMaps(providers).canonical.values()]; } /** Resolve a transcript provider by canonical id or alias. */ diff --git a/src/transcripts/store-export-ownership.ts b/src/transcripts/store-export-ownership.ts index 933082976f66..08c0a07a4041 100644 --- a/src/transcripts/store-export-ownership.ts +++ b/src/transcripts/store-export-ownership.ts @@ -12,6 +12,7 @@ import type { TranscriptSessionDescriptor } from "./provider-types.js"; import { ensureMeetingTranscriptsSchema } from "./sqlite-schema.js"; import { isCaseSensitiveDirectory, + TRANSCRIPT_EXPORT_FILE_NAMES, transcriptSessionExportKey, transcriptSessionSelector, } from "./store-artifacts.js"; @@ -23,13 +24,6 @@ type ExportOwnershipParams = { databaseOptions: OpenClawStateDatabaseOptions; }; -const TRANSCRIPT_EXPORT_FILE_NAMES = new Set([ - "metadata.json", - "summary.json", - "summary.md", - "transcript.jsonl", -]); - function database(options: OpenClawStateDatabaseOptions) { ensureMeetingTranscriptsSchema(options); return openOpenClawStateDatabase(options); diff --git a/src/transcripts/store-sqlite.ts b/src/transcripts/store-sqlite.ts index 9c743656914c..26037abcabbe 100644 --- a/src/transcripts/store-sqlite.ts +++ b/src/transcripts/store-sqlite.ts @@ -1,4 +1,5 @@ import type { DatabaseSync } from "node:sqlite"; +import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce"; import type { Selectable } from "kysely"; import { executeSqliteQuerySync, @@ -121,10 +122,7 @@ function parseOptionalJsonRecord(value: string | null): Record if (!value) { return undefined; } - const parsed = JSON.parse(value) as unknown; - return parsed && typeof parsed === "object" && !Array.isArray(parsed) - ? (parsed as Record) - : undefined; + return asOptionalRecord(JSON.parse(value)); } export function sessionFromRow(row: MeetingTranscriptSessionRow): TranscriptSessionDescriptor { diff --git a/src/transcripts/store.ts b/src/transcripts/store.ts index d9b9c065eb01..b8ae50759a37 100644 --- a/src/transcripts/store.ts +++ b/src/transcripts/store.ts @@ -605,9 +605,6 @@ export class TranscriptsStore { const transcriptPath = path.join(sessionDir, "transcript.jsonl"); const summaryJsonPath = path.join(sessionDir, "summary.json"); const summaryPath = path.join(sessionDir, "summary.md"); - // Every export starts with identity metadata, so even an interrupted partial - // materialization remains inspectable by Doctor without guessing its owner. - const includeMetadata = true; const includeTranscript = kind === "all" || kind === "transcript"; const includeSummary = kind === "all" || kind === "summary"; const storedSummary = includeSummary ? await this.readSummary(session) : {}; @@ -632,13 +629,13 @@ export class TranscriptsStore { if (!ensured.ok) { throw ensured.error; } - if (includeMetadata) { - exportedHashes["metadata.json"] = await writeTranscriptArtifact( - sessionDir, - "metadata.json", - `${JSON.stringify(session, null, 2)}\n`, - ); - } + // Every export starts with identity metadata, so even an interrupted partial + // materialization remains inspectable by Doctor without guessing its owner. + exportedHashes["metadata.json"] = await writeTranscriptArtifact( + sessionDir, + "metadata.json", + `${JSON.stringify(session, null, 2)}\n`, + ); if (includeTranscript) { exportedHashes["transcript.jsonl"] = await writeTranscriptJsonlArtifact({ sessionDir, @@ -668,9 +665,7 @@ export class TranscriptsStore { removedExports.add("summary.md"); } } - if (Object.keys(exportedHashes).length > 0 || removedExports.size > 0) { - this.updateExportManifest(session, exportedHashes, removedExports); - } + this.updateExportManifest(session, exportedHashes, removedExports); return { sessionDir, metadataPath,