ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-12 21:53:00 -06:00
Code Issues Packages Projects Releases Wiki Activity

refactor(sessions): consolidate transcript and state helpers (#119435)

Browse Source
This commit is contained in:
Peter Steinberger
2026-08-04 20:45:41 -07:00
committed by GitHub
parent 2692954c72
commit 39bbd8afd3
12 changed files with 117 additions and 223 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/context-engine/quarantine-health.ts
+5 -8
View File
@@ -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<PersistedContextEngineRuntimeQuarantine, "failedAt">;
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<PersistedContextEngineQuarantineRecord>({
@@ -29,9 +26,9 @@ const quarantineStore = createRuntimeHealthStore<PersistedContextEngineQuarantin
maxEntries: 64,
normalizeRecord: (value) => {
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<PersistedContextEngineQuarantin
processId: value.processId,
processToken: value.processToken,
processStartTime: value.processStartTime,
...(isNonEmptyString(value.owner) ? { owner: value.owner } : {}),
...(hasNonEmptyString(value.owner) ? { owner: value.owner } : {}),
};
},
displayKey: (record) => record.engineId,
src/sessions/conversation-turns.ts
+11 -15
View File
@@ -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<ConversationTurnReplyClaim | undefined> {
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;
}
src/sessions/session-chat-type.ts
+3 -18
View File
@@ -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<ReturnType<typeof getBootstrapChannelPlugin>>["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;
}
src/sessions/transcript-events.ts
+12 -25
View File
@@ -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 } : {}),
};
}
src/sessions/user-turn-transcript.media-normalize.ts
+17 -25
View File
@@ -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<NonNullable<MediaFactInput["kind"]>>([
]);
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<MediaFactInput["kind"]>)
? (kind as NonNullable<MediaFactInput["kind"]>)
: 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 } : {}),
src/sessions/user-turn-transcript.ts
+14 -30
View File
@@ -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<string, string> | 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<string, unknown> | undefined {
const meta = (message as unknown as Record<string, unknown>)["__openclaw"];
return meta && typeof meta === "object" && !Array.isArray(meta)
? (meta as Record<string, unknown>)
: undefined;
return asOptionalRecord((message as unknown as Record<string, unknown>)["__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 } : {}),
src/state/openclaw-agent-db-session-migrations.ts
+34 -47
View File
@@ -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<string, unknown>;
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) {
src/state/openclaw-agent-db-session-provenance.ts
+8 -16
View File
@@ -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<string, unknown> | undefined {
if (typeof value === "string") {
try {
const parsed: unknown = JSON.parse(value);
return parsed && typeof parsed === "object" && !Array.isArray(parsed)
? (parsed as Record<string, unknown>)
: undefined;
return asOptionalRecord(JSON.parse(value));
} catch {
return undefined;
}
}
return value && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: 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,
);
src/transcripts/provider-registry.ts
+2 -15
View File
@@ -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<string, TranscriptSourceProvider>;
aliases: Map<string, TranscriptSourceProvider>;
} {
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. */
src/transcripts/store-export-ownership.ts
+1 -7
View File
@@ -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);
src/transcripts/store-sqlite.ts
+2 -4
View File
@@ -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<string, unknown>
if (!value) {
return undefined;
}
const parsed = JSON.parse(value) as unknown;
return parsed && typeof parsed === "object" && !Array.isArray(parsed)
? (parsed as Record<string, unknown>)
: undefined;
return asOptionalRecord(JSON.parse(value));
}
export function sessionFromRow(row: MeetingTranscriptSessionRow): TranscriptSessionDescriptor {
src/transcripts/store.ts
+8 -13
View File
@@ -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,