From 8aa67634ca03e9dc9343af70e1c5cbd9682e6375 Mon Sep 17 00:00:00 2001 From: Galin Iliev Date: Tue, 28 Jul 2026 21:47:51 +0300 Subject: [PATCH] fix: agent stops replying when a session transcript has no header row (#115080) * fix(sessions): stop rejecting header-less persisted transcripts as legacy Sessions whose SQLite transcript has no session header row were treated as version 1 and hard-failed every run with "Persisted legacy session transcripts require doctor/import migration before runtime use". Only an actual header now declares a legacy version. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(sessions): keep entry ids stable for header-less transcripts Rebuild the header at the current version instead of inferring v1, and route header-less transcripts that still hold legacy-shaped entries to doctor. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(sessions): repair headerless transcripts in doctor --------- Co-authored-by: Galin Iliev Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- docs/cli/doctor.md | 7 + src/agents/sessions/session-manager-core.ts | 4 +- src/commands/doctor-session-sqlite-readers.ts | 58 +++ .../doctor-session-transcript-headers.test.ts | 363 ++++++++++++++++++ .../doctor-session-transcript-headers.ts | 329 ++++++++++++++++ .../sessions/session-accessor.sqlite-read.ts | 25 ++ ...ession-accessor.sqlite-transcript-store.ts | 33 +- ...octor-health-contribution-runners.state.ts | 12 + .../doctor-health-contributions-initial.ts | 6 + 9 files changed, 832 insertions(+), 5 deletions(-) create mode 100644 src/commands/doctor-session-transcript-headers.test.ts create mode 100644 src/commands/doctor-session-transcript-headers.ts diff --git a/docs/cli/doctor.md b/docs/cli/doctor.md index 255847464024..6193966dae59 100644 --- a/docs/cli/doctor.md +++ b/docs/cli/doctor.md @@ -269,6 +269,13 @@ imported and archived out of the active sessions directory after successful import; archive-tier JSONL files remain support artifacts, not runtime fallbacks. +The regular `openclaw doctor` pass also reports canonical SQLite transcripts +whose initial session header was never persisted. `openclaw doctor --fix` +prepends a current header and rebuilds the transcript indexes in one +transaction while preserving existing event IDs, parent links, row timestamps, +and session-list recency. Headerless legacy or malformed transcripts remain +rejected until their owning migration can validate them. + Modes: | Mode | Behavior | diff --git a/src/agents/sessions/session-manager-core.ts b/src/agents/sessions/session-manager-core.ts index ccadf89bfdf9..65a6733ceb10 100644 --- a/src/agents/sessions/session-manager-core.ts +++ b/src/agents/sessions/session-manager-core.ts @@ -73,7 +73,9 @@ export class SessionManagerCore { entries: FileEntry[], ): void { const partitioned = partitionSessionFileEntries(entries); - if (partitioned.fileEntries.length === 0) { + // Only a physically empty transcript may initialize lazily. Opaque persisted rows still need + // a canonical header, or runtime would silently replace malformed history with a fresh session. + if (partitioned.fileEntries.length === 0 && partitioned.opaqueEntries.length === 0) { this.persistenceTarget = target ? { ...target } : undefined; this.initializeSession({ id: target?.sessionId }); this.persistenceHeaderPending = target !== undefined; diff --git a/src/commands/doctor-session-sqlite-readers.ts b/src/commands/doctor-session-sqlite-readers.ts index be18c7a8ac5d..228086680587 100644 --- a/src/commands/doctor-session-sqlite-readers.ts +++ b/src/commands/doctor-session-sqlite-readers.ts @@ -5,6 +5,7 @@ import { TextDecoder } from "node:util"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; import { normalizeLoadedFileEntry, type FileEntry } from "../agents/sessions/session-manager.js"; import type { TranscriptEvent } from "../config/sessions/session-accessor.js"; +import type { SqliteTranscriptStorageRow } from "../config/sessions/session-accessor.sqlite-read.js"; import { resolveSqliteTargetFromSessionStorePath } from "../config/sessions/session-sqlite-target.js"; import type { SessionStoreTarget } from "../config/sessions/targets.js"; import type { SessionEntry } from "../config/sessions/types.js"; @@ -432,3 +433,60 @@ export function readOnlySqliteTranscriptSnapshot( database?.close(); } } + +/** Reads exact row metadata for a guarded transcript replacement without opening a writer. */ +export function readOnlySqliteTranscriptStorageSnapshot( + sqlitePath: string, + sessionId: string, +): + | { ok: true; rows: SqliteTranscriptStorageRow[]; sessionKey?: string } + | { ok: false; error: unknown } { + if (!fs.existsSync(sqlitePath)) { + return { ok: false, error: new Error(`SQLite database not found: ${sqlitePath}`) }; + } + let database: DatabaseSync | undefined; + try { + database = openNodeSqliteDatabase(sqlitePath, { readOnly: true }); + const rows = database + .prepare( + "SELECT created_at, event_json, seq FROM transcript_events WHERE session_id = ? ORDER BY seq ASC", + ) + .all(sessionId) as Array<{ + created_at?: unknown; + event_json?: unknown; + seq?: unknown; + }>; + const sessionKeyRow = database + .prepare("SELECT session_key FROM session_windows WHERE session_id = ? LIMIT 1") + .get(sessionId) as { session_key?: unknown } | undefined; + const storageRows: SqliteTranscriptStorageRow[] = []; + for (const row of rows) { + if ( + typeof row.created_at !== "number" || + typeof row.event_json !== "string" || + typeof row.seq !== "number" + ) { + return { + ok: false, + error: new Error(`Invalid transcript row metadata for session ${sessionId}`), + }; + } + storageRows.push({ + createdAt: row.created_at, + eventJson: row.event_json, + seq: row.seq, + }); + } + return { + ok: true, + rows: storageRows, + ...(typeof sessionKeyRow?.session_key === "string" + ? { sessionKey: sessionKeyRow.session_key } + : {}), + }; + } catch (error) { + return { ok: false, error }; + } finally { + database?.close(); + } +} diff --git a/src/commands/doctor-session-transcript-headers.test.ts b/src/commands/doctor-session-transcript-headers.test.ts new file mode 100644 index 000000000000..bf9b25b12ec3 --- /dev/null +++ b/src/commands/doctor-session-transcript-headers.test.ts @@ -0,0 +1,363 @@ +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { SessionManager } from "../agents/sessions/session-manager.js"; +import { + loadTranscriptEventsSync, + replaceTranscriptEventsSync, + upsertSessionEntry, +} from "../config/sessions/session-accessor.js"; +import { readSqliteTranscriptStorageRows } from "../config/sessions/session-accessor.sqlite-read.js"; +import { CURRENT_SESSION_VERSION } from "../config/sessions/version.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, + runOpenClawAgentWriteTransaction, +} from "../state/openclaw-agent-db.js"; +import { closeOpenClawStateDatabaseForTest } from "../state/openclaw-state-db.js"; +import { + createOpenClawTestState, + type OpenClawTestState, +} from "../test-utils/openclaw-test-state.js"; + +const note = vi.hoisted(() => vi.fn()); + +vi.mock("../../packages/terminal-core/src/note.js", () => ({ note })); + +import { noteSessionTranscriptHeaderHealth } from "./doctor-session-transcript-headers.js"; + +const AGENT_ID = "main"; +const SESSION_ID = "headerless-session"; +const SESSION_KEY = "agent:main:headerless-session"; +const SPAWNED_CWD = "/workspace/headerless-child"; + +describe("doctor SQLite session transcript header repair", () => { + let state: OpenClawTestState; + let cfg: OpenClawConfig; + let scope: { + agentId: string; + env: NodeJS.ProcessEnv; + sessionId: string; + sessionKey: string; + storePath: string; + }; + + beforeEach(async () => { + note.mockClear(); + state = await createOpenClawTestState({ + layout: "state-only", + prefix: "openclaw-doctor-transcript-headers-", + }); + cfg = { + agents: { list: [{ id: AGENT_ID, workspace: state.workspaceDir }] }, + }; + scope = { + agentId: AGENT_ID, + env: state.env, + sessionId: SESSION_ID, + sessionKey: SESSION_KEY, + storePath: path.join(state.sessionsDir(AGENT_ID), "sessions.json"), + }; + }); + + afterEach(async () => { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + await state.cleanup(); + }); + + async function seedHeaderlessTranscript( + events: readonly unknown[], + options: { spawnedCwd?: string } = { spawnedCwd: SPAWNED_CWD }, + ): Promise { + await upsertSessionEntry(scope, { + sessionId: SESSION_ID, + ...(options.spawnedCwd ? { spawnedCwd: options.spawnedCwd } : {}), + updatedAt: 10, + }); + expect(replaceTranscriptEventsSync(scope, [...events])).toBe(true); + } + + it("detects read-only, repairs atomically, and keeps event identities stable", async () => { + await seedHeaderlessTranscript([ + { + type: "model_change", + id: "model-1", + parentId: null, + timestamp: "2026-07-15T21:23:03.632Z", + provider: "github-copilot", + modelId: "claude-opus-4.8", + }, + { + type: "message", + id: "user-1", + parentId: "model-1", + timestamp: "2026-07-15T21:23:03.698Z", + message: { role: "user", content: "Is all good?" }, + }, + { + type: "leaf", + id: "leaf-1", + parentId: "user-1", + targetId: "user-1", + timestamp: "2026-07-15T21:23:03.699Z", + }, + ]); + const databaseOptions = { agentId: AGENT_ID, env: state.env }; + const database = openOpenClawAgentDatabase(databaseOptions); + const beforeRows = readSqliteTranscriptStorageRows(database, SESSION_ID); + runOpenClawAgentWriteTransaction((transactionDatabase) => { + transactionDatabase.db + .prepare( + "UPDATE session_windows SET transcript_updated_at = ?, transcript_observed_at = ? WHERE session_id = ?", + ) + .run(1_000_000, 999_000, SESSION_ID); + }, databaseOptions); + + expect(() => SessionManager.open(scope, state.workspaceDir)).toThrow( + "require doctor/import migration before runtime use", + ); + + await expect( + noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: false }), + ).resolves.toEqual({ found: 1, repaired: 0 }); + expect(readSqliteTranscriptStorageRows(database, SESSION_ID)).toEqual(beforeRows); + expect(note).toHaveBeenCalledWith( + '- Found 1 canonical session transcript without a header.\n- Run "openclaw doctor --fix" to repair it before resuming the session.', + "Session transcript headers", + ); + + note.mockClear(); + await expect( + noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }), + ).resolves.toEqual({ found: 1, repaired: 1 }); + + const repairedRows = readSqliteTranscriptStorageRows(database, SESSION_ID); + const repairedEvents = repairedRows.map((row) => JSON.parse(row.eventJson)); + expect(repairedEvents[0]).toMatchObject({ + type: "session", + version: CURRENT_SESSION_VERSION, + id: SESSION_ID, + cwd: SPAWNED_CWD, + timestamp: new Date(beforeRows[0]?.createdAt ?? 0).toISOString(), + }); + expect( + repairedEvents.slice(1).map((event) => ({ + id: event.id, + parentId: event.parentId, + targetId: event.targetId, + })), + ).toEqual([ + { id: "model-1", parentId: null, targetId: undefined }, + { id: "user-1", parentId: "model-1", targetId: undefined }, + { id: "leaf-1", parentId: "user-1", targetId: "user-1" }, + ]); + expect(repairedRows.slice(1).map((row) => row.createdAt)).toEqual( + beforeRows.map((row) => row.createdAt), + ); + expect(repairedRows.map((row) => row.seq)).toEqual([0, 1, 2, 3]); + + const identities = database.db + .prepare( + "SELECT event_id, parent_id, seq FROM transcript_event_identities WHERE session_id = ? ORDER BY seq ASC", + ) + .all(SESSION_ID); + expect(identities).toEqual([ + { event_id: SESSION_ID, parent_id: null, seq: 0 }, + { event_id: "model-1", parent_id: null, seq: 1 }, + { event_id: "user-1", parent_id: "model-1", seq: 2 }, + { event_id: "leaf-1", parent_id: "user-1", seq: 3 }, + ]); + expect( + database.db + .prepare( + "SELECT event_seq FROM session_transcript_active_events WHERE session_id = ? ORDER BY active_position ASC", + ) + .all(SESSION_ID), + ).toEqual([{ event_seq: 1 }, { event_seq: 2 }]); + expect( + database.db + .prepare("SELECT transcript_updated_at AS value FROM session_windows WHERE session_id = ?") + .get(SESSION_ID), + ).toEqual({ value: 1_000_001 }); + const repairedManager = SessionManager.open(scope, state.workspaceDir); + expect(repairedManager.getEntries().map((entry) => entry.id)).toEqual(["model-1", "user-1"]); + expect(repairedManager.buildSessionContext().messages).toEqual([ + { role: "user", content: "Is all good?" }, + ]); + expect(note).toHaveBeenCalledWith( + "- Prepended current headers to 1 session transcript.", + "Session transcript headers", + ); + + note.mockClear(); + const afterFirstRepair = readSqliteTranscriptStorageRows(database, SESSION_ID); + await expect( + noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }), + ).resolves.toEqual({ found: 0, repaired: 0 }); + expect(readSqliteTranscriptStorageRows(database, SESSION_ID)).toEqual(afterFirstRepair); + expect(note).not.toHaveBeenCalled(); + }); + + it("does not admit legacy headerless rows into the current runtime shape", async () => { + await seedHeaderlessTranscript([ + { type: "message", message: { role: "user", content: "legacy message" } }, + { type: "message", message: { role: "hookMessage", content: "legacy hook" } }, + ]); + const database = openOpenClawAgentDatabase({ agentId: AGENT_ID, env: state.env }); + const before = readSqliteTranscriptStorageRows(database, SESSION_ID); + + await expect( + noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }), + ).resolves.toEqual({ found: 0, repaired: 0 }); + + expect(readSqliteTranscriptStorageRows(database, SESSION_ID)).toEqual(before); + expect(() => SessionManager.open(scope, state.workspaceDir)).toThrow( + "require doctor/import migration before runtime use", + ); + expect(loadTranscriptEventsSync(scope)).toHaveLength(2); + expect(note).not.toHaveBeenCalled(); + }); + + it.each([ + { + name: "a malformed current entry", + events: [ + { + type: "message", + id: "malformed-message", + parentId: null, + timestamp: "2026-07-15T21:23:03.698Z", + message: { role: "invalid", content: "not a current message" }, + }, + ], + }, + { + name: "an invalid leaf control", + events: [ + { + type: "message", + id: "user-1", + parentId: null, + timestamp: "2026-07-15T21:23:03.698Z", + message: { role: "user", content: "hello" }, + }, + { + type: "leaf", + id: "invalid-leaf", + parentId: "user-1", + }, + ], + }, + ])("does not repair $name", async ({ events }) => { + await seedHeaderlessTranscript(events); + const database = openOpenClawAgentDatabase({ agentId: AGENT_ID, env: state.env }); + const before = readSqliteTranscriptStorageRows(database, SESSION_ID); + + await expect( + noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }), + ).resolves.toEqual({ found: 0, repaired: 0 }); + + expect(readSqliteTranscriptStorageRows(database, SESSION_ID)).toEqual(before); + expect(() => SessionManager.open(scope, state.workspaceDir)).toThrow( + "require doctor/import migration before runtime use", + ); + expect(note).not.toHaveBeenCalled(); + }); + + it("does not repair duplicate event identities", async () => { + await seedHeaderlessTranscript([ + { + type: "message", + id: "user-1", + parentId: null, + timestamp: "2026-07-15T21:23:03.698Z", + message: { role: "user", content: "first" }, + }, + { + type: "message", + id: "user-2", + parentId: "user-1", + timestamp: "2026-07-15T21:23:03.699Z", + message: { role: "user", content: "second" }, + }, + ]); + const databaseOptions = { agentId: AGENT_ID, env: state.env }; + const database = openOpenClawAgentDatabase(databaseOptions); + runOpenClawAgentWriteTransaction((transactionDatabase) => { + const duplicate = { + type: "message", + id: "user-1", + parentId: "user-1", + timestamp: "2026-07-15T21:23:03.699Z", + message: { role: "user", content: "second" }, + }; + transactionDatabase.db + .prepare("UPDATE transcript_events SET event_json = ? WHERE session_id = ? AND seq = 1") + .run(JSON.stringify(duplicate), SESSION_ID); + }, databaseOptions); + const before = readSqliteTranscriptStorageRows(database, SESSION_ID); + + await expect( + noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }), + ).resolves.toEqual({ found: 0, repaired: 0 }); + + expect(readSqliteTranscriptStorageRows(database, SESSION_ID)).toEqual(before); + expect(note).not.toHaveBeenCalled(); + }); + + it("uses the configured agent workspace when the session has no spawned cwd", async () => { + await seedHeaderlessTranscript( + [ + { + type: "message", + id: "user-1", + parentId: null, + timestamp: "2026-07-15T21:23:03.698Z", + message: { role: "user", content: "hello" }, + }, + ], + {}, + ); + + await expect( + noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }), + ).resolves.toEqual({ found: 1, repaired: 1 }); + + const database = openOpenClawAgentDatabase({ agentId: AGENT_ID, env: state.env }); + const header = JSON.parse(readSqliteTranscriptStorageRows(database, SESSION_ID)[0]!.eventJson); + expect(header).toMatchObject({ type: "session", cwd: state.workspaceDir }); + }); + + it("does not borrow the spawned cwd from a newer session on the same key", async () => { + await seedHeaderlessTranscript([ + { + type: "message", + id: "user-1", + parentId: null, + timestamp: "2026-07-15T21:23:03.698Z", + message: { role: "user", content: "hello" }, + }, + ]); + const databaseOptions = { agentId: AGENT_ID, env: state.env }; + runOpenClawAgentWriteTransaction((database) => { + database.db + .prepare( + "UPDATE session_nodes SET current_session_id = ?, entry_json = ? WHERE session_key = ?", + ) + .run( + "newer-session", + JSON.stringify({ sessionId: "newer-session", spawnedCwd: "/workspace/newer" }), + SESSION_KEY, + ); + }, databaseOptions); + + await expect( + noteSessionTranscriptHeaderHealth({ cfg, env: state.env, shouldRepair: true }), + ).resolves.toEqual({ found: 1, repaired: 1 }); + + const database = openOpenClawAgentDatabase(databaseOptions); + const header = JSON.parse(readSqliteTranscriptStorageRows(database, SESSION_ID)[0]!.eventJson); + expect(header).toMatchObject({ type: "session", cwd: state.workspaceDir }); + }); +}); diff --git a/src/commands/doctor-session-transcript-headers.ts b/src/commands/doctor-session-transcript-headers.ts new file mode 100644 index 000000000000..23166cbb6226 --- /dev/null +++ b/src/commands/doctor-session-transcript-headers.ts @@ -0,0 +1,329 @@ +import fs from "node:fs"; +import { note } from "../../packages/terminal-core/src/note.js"; +import { resolveAgentWorkspaceDir } from "../agents/agent-scope.js"; +import { isIndexedSessionEntry } from "../agents/sessions/session-manager-codec.js"; +import type { TranscriptEvent } from "../config/sessions/session-accessor.js"; +import { + readSqliteTranscriptStorageRows, + type SqliteTranscriptStorageRow, +} from "../config/sessions/session-accessor.sqlite-read.js"; +import { getSessionKysely } from "../config/sessions/session-accessor.sqlite-scope.js"; +import { replaceSqliteTranscriptEventsInTransaction } from "../config/sessions/session-accessor.sqlite-transcript-store.js"; +import { resolveAllAgentSessionStoreTargetsSync } from "../config/sessions/targets.js"; +import { createSessionTranscriptHeader } from "../config/sessions/transcript-header.js"; +import { + isCanonicalSessionTranscriptEntry, + isSessionTranscriptLeafControl, +} from "../config/sessions/transcript-tree.js"; +import type { OpenClawConfig } from "../config/types.openclaw.js"; +import { formatErrorMessage } from "../infra/errors.js"; +import { executeSqliteQueryTakeFirstSync } from "../infra/kysely-sync.js"; +import { parseAgentSessionKey } from "../routing/session-key.js"; +import { + runOpenClawAgentWriteTransaction, + type OpenClawAgentDatabase, +} from "../state/openclaw-agent-db.js"; +import { + readOnlySqliteTranscriptSessionIds, + readOnlySqliteTranscriptStorageSnapshot, + resolveTargetSqlitePath, +} from "./doctor-session-sqlite-readers.js"; + +const NOTE_TITLE = "Session transcript headers"; + +type HeaderRepairContext = { + sessionKey: string; + spawnedCwd?: string; +}; + +type HeaderRepairReport = { + found: number; + repaired: number; +}; + +function parseCanonicalHeaderlessEvents( + rows: readonly SqliteTranscriptStorageRow[], + sessionId: string, +): TranscriptEvent[] | undefined { + if (rows.length === 0) { + return undefined; + } + const events: TranscriptEvent[] = []; + const eventIds = new Set([sessionId]); + let indexedEntries = 0; + for (const row of rows) { + let event: TranscriptEvent; + try { + event = JSON.parse(row.eventJson) as TranscriptEvent; + } catch { + return undefined; + } + if (!event || typeof event !== "object" || Array.isArray(event)) { + return undefined; + } + const record = event as Record; + if (record.type === "session") { + return undefined; + } + // Known transcript entries must already satisfy the current runtime schema. Opaque plugin + // rows remain uninterpreted, but malformed leaf controls and duplicate identities would make + // a delete-and-reinsert repair lossy or ambiguous, so fail closed on those shapes. + if (isCanonicalSessionTranscriptEntry(record)) { + if (!isIndexedSessionEntry(event)) { + return undefined; + } + indexedEntries += 1; + } else if (record.type === "leaf" && !isSessionTranscriptLeafControl(record)) { + return undefined; + } + if (typeof record.id === "string") { + const eventId = record.id.trim(); + if (!eventId || eventIds.has(eventId)) { + return undefined; + } + eventIds.add(eventId); + } + events.push(event); + } + return indexedEntries > 0 ? events : undefined; +} + +function snapshotsMatch( + expected: readonly SqliteTranscriptStorageRow[], + current: readonly SqliteTranscriptStorageRow[], +): boolean { + return ( + expected.length === current.length && + expected.every( + (row, index) => + row.seq === current[index]?.seq && + row.createdAt === current[index]?.createdAt && + row.eventJson === current[index]?.eventJson, + ) + ); +} + +function readHeaderRepairContext( + database: OpenClawAgentDatabase, + sessionId: string, +): HeaderRepairContext | undefined { + const db = getSessionKysely(database.db); + const window = executeSqliteQueryTakeFirstSync( + database.db, + db + .selectFrom("session_windows") + .select("session_key") + .where("session_id", "=", sessionId) + .limit(1), + ); + if (!window?.session_key) { + return undefined; + } + const node = executeSqliteQueryTakeFirstSync( + database.db, + db + .selectFrom("session_nodes") + .select(["current_session_id", "entry_json"]) + .where("session_key", "=", window.session_key) + .limit(1), + ); + let spawnedCwd: string | undefined; + // Historical windows can share a key; only the node's current session owns entry_json. + if (node?.current_session_id === sessionId && node.entry_json) { + try { + const entry = JSON.parse(node.entry_json) as { + sessionId?: unknown; + spawnedCwd?: unknown; + }; + if ( + entry.sessionId === sessionId && + typeof entry.spawnedCwd === "string" && + entry.spawnedCwd.trim() + ) { + spawnedCwd = entry.spawnedCwd.trim(); + } + } catch { + // The transcript can still be repaired with the configured agent workspace. + } + } + return { sessionKey: window.session_key, ...(spawnedCwd ? { spawnedCwd } : {}) }; +} + +function formatHeaderTimestamp(createdAt: number): string | undefined { + if (!Number.isFinite(createdAt)) { + return undefined; + } + try { + return new Date(createdAt).toISOString(); + } catch { + return undefined; + } +} + +function assertRepairPreservedEvents(params: { + before: readonly SqliteTranscriptStorageRow[]; + database: OpenClawAgentDatabase; + sessionId: string; +}): void { + const after = readSqliteTranscriptStorageRows(params.database, params.sessionId); + if (after.length !== params.before.length + 1) { + throw new Error(`header repair changed the event count for ${params.sessionId}`); + } + for (const [index, beforeRow] of params.before.entries()) { + const afterRow = after[index + 1]; + if (!afterRow || afterRow.createdAt !== beforeRow.createdAt) { + throw new Error(`header repair changed row timestamps for ${params.sessionId}`); + } + const beforeEvent = JSON.parse(beforeRow.eventJson) as Record; + const afterEvent = JSON.parse(afterRow.eventJson) as Record; + if ( + beforeEvent.id !== afterEvent.id || + beforeEvent.parentId !== afterEvent.parentId || + beforeEvent.targetId !== afterEvent.targetId || + beforeEvent.appendParentId !== afterEvent.appendParentId + ) { + throw new Error(`header repair changed event identity for ${params.sessionId}`); + } + } +} + +function formatCount(count: number, singular: string): string { + return `${count} ${singular}${count === 1 ? "" : "s"}`; +} + +/** Reports or repairs canonical SQLite transcripts whose first header was never persisted. */ +export async function noteSessionTranscriptHeaderHealth(params: { + cfg: OpenClawConfig; + env?: NodeJS.ProcessEnv; + shouldRepair: boolean; +}): Promise { + const env = params.env ?? process.env; + let found = 0; + let repaired = 0; + + const targetsBySqlitePath = new Map(); + for (const target of resolveAllAgentSessionStoreTargetsSync(params.cfg, { env })) { + const sqlitePath = resolveTargetSqlitePath(target); + if (!targetsBySqlitePath.has(sqlitePath)) { + targetsBySqlitePath.set(sqlitePath, target); + } + } + + for (const [sqlitePath, target] of targetsBySqlitePath) { + if (!fs.existsSync(sqlitePath)) { + continue; + } + const databaseOptions = { agentId: target.agentId, env, path: sqlitePath }; + try { + for (const sessionId of readOnlySqliteTranscriptSessionIds(sqlitePath)) { + const snapshot = readOnlySqliteTranscriptStorageSnapshot(sqlitePath, sessionId); + if (!snapshot.ok) { + const detail = formatErrorMessage(snapshot.error).replace(/\s+/g, " ").trim(); + note( + `- Failed to read transcript ${sessionId} (${target.agentId}): ${detail}`, + NOTE_TITLE, + ); + continue; + } + if (!snapshot.sessionKey || !parseCanonicalHeaderlessEvents(snapshot.rows, sessionId)) { + continue; + } + const headerTimestamp = formatHeaderTimestamp(snapshot.rows[0]?.createdAt ?? Number.NaN); + if (!headerTimestamp) { + note( + `- Failed to repair transcript ${sessionId} (${target.agentId}): invalid first-row timestamp`, + NOTE_TITLE, + ); + continue; + } + found += 1; + if (!params.shouldRepair) { + continue; + } + + const logicalAgentId = parseAgentSessionKey(snapshot.sessionKey)?.agentId ?? target.agentId; + const workspaceCwd = resolveAgentWorkspaceDir(params.cfg, logicalAgentId, env); + try { + runOpenClawAgentWriteTransaction( + (database) => { + const currentRows = readSqliteTranscriptStorageRows(database, sessionId); + if (!snapshotsMatch(snapshot.rows, currentRows)) { + throw new Error( + `transcript changed while preparing header repair for ${sessionId}`, + ); + } + const events = parseCanonicalHeaderlessEvents(currentRows, sessionId); + if (!events) { + throw new Error( + `transcript is no longer a canonical headerless session: ${sessionId}`, + ); + } + const context = readHeaderRepairContext(database, sessionId); + if (!context || context.sessionKey !== snapshot.sessionKey) { + throw new Error( + `session binding changed while preparing header repair for ${sessionId}`, + ); + } + const header = createSessionTranscriptHeader({ + cwd: context.spawnedCwd ?? workspaceCwd, + sessionId, + timestamp: headerTimestamp, + }); + replaceSqliteTranscriptEventsInTransaction( + database, + { + agentId: target.agentId, + env, + path: sqlitePath, + sessionId, + sessionKey: context.sessionKey, + }, + [header, ...events], + { + createdAtByIndex: [ + currentRows[0]?.createdAt ?? Date.parse(headerTimestamp), + ...currentRows.map((row) => row.createdAt), + ], + preserveSessionWindowRecency: true, + }, + ); + assertRepairPreservedEvents({ before: currentRows, database, sessionId }); + }, + databaseOptions, + { operationLabel: "doctor.session-transcript-headers" }, + ); + repaired += 1; + } catch (error) { + const detail = formatErrorMessage(error).replace(/\s+/g, " ").trim(); + note( + `- Failed to repair transcript ${sessionId} (${target.agentId}): ${detail}`, + NOTE_TITLE, + ); + } + } + } catch (error) { + const detail = formatErrorMessage(error).replace(/\s+/g, " ").trim(); + note( + `- Failed to inspect transcript headers for ${target.agentId} (${sqlitePath}): ${detail}`, + NOTE_TITLE, + ); + } + } + + if (params.shouldRepair && repaired > 0) { + note( + `- Prepended current headers to ${formatCount(repaired, "session transcript")}.`, + NOTE_TITLE, + ); + } else if (!params.shouldRepair && found > 0) { + note( + [ + `- Found ${formatCount(found, "canonical session transcript")} without a header.`, + `- Run "openclaw doctor --fix" to repair ${found === 1 ? "it" : "them"} before resuming the session.`, + ].join("\n"), + NOTE_TITLE, + ); + } + + return { found, repaired }; +} diff --git a/src/config/sessions/session-accessor.sqlite-read.ts b/src/config/sessions/session-accessor.sqlite-read.ts index 2ffc5fe9c764..30f39e3ff67e 100644 --- a/src/config/sessions/session-accessor.sqlite-read.ts +++ b/src/config/sessions/session-accessor.sqlite-read.ts @@ -30,6 +30,10 @@ export type SqliteTranscriptSnapshotRow = { seq: number; }; +export type SqliteTranscriptStorageRow = SqliteTranscriptSnapshotRow & { + createdAt: number; +}; + /** Loads raw transcript events from the additive SQLite transcript store. */ export async function loadSqliteTranscriptEvents( scope: SessionTranscriptReadScope, @@ -193,6 +197,27 @@ export function readSqliteTranscriptEventRows( })); } +/** Reads exact transcript storage rows for guarded doctor rewrites. */ +export function readSqliteTranscriptStorageRows( + database: OpenClawAgentDatabase, + sessionId: string, +): SqliteTranscriptStorageRow[] { + const db = getSessionKysely(database.db); + const rows = executeSqliteQuerySync( + database.db, + db + .selectFrom("transcript_events") + .select(["created_at", "event_json", "seq"]) + .where("session_id", "=", sessionId) + .orderBy("seq", "asc"), + ).rows; + return rows.map((row) => ({ + createdAt: normalizeSqliteNumber(row.created_at), + eventJson: row.event_json, + seq: normalizeSqliteNumber(row.seq), + })); +} + function sqliteTranscriptJsonlByteSize() { return /* kysely-allow-raw: JSONL size includes event bytes plus newline separators. */ sql`COALESCE(SUM(LENGTH(CAST(event_json AS BLOB))), 0) + CASE WHEN COUNT(*) > 0 THEN COUNT(*) - 1 ELSE 0 END`.as("size_bytes"); diff --git a/src/config/sessions/session-accessor.sqlite-transcript-store.ts b/src/config/sessions/session-accessor.sqlite-transcript-store.ts index 135372358371..1fbba6987ada 100644 --- a/src/config/sessions/session-accessor.sqlite-transcript-store.ts +++ b/src/config/sessions/session-accessor.sqlite-transcript-store.ts @@ -322,19 +322,28 @@ export function replaceSqliteTranscriptEventsInTransaction( events: readonly TranscriptEvent[], options: { createdAtByIndex?: readonly number[]; - preserveSessionWindowTimestamps?: boolean; + /** Keep maintenance rewrites at their existing recency while invalidating stale projections. */ + preserveSessionWindowRecency?: boolean; } = {}, ): void { + const preservedTranscriptUpdatedAt = + options.preserveSessionWindowRecency === true + ? readTranscriptMutationStateInTransaction(database, resolved.sessionId).updatedAt + : undefined; const previousGeneration = readTranscriptGenerationInTransaction(database, resolved.sessionId); const deleted = deleteSqliteTranscriptEventsInTransaction(database, resolved.sessionId); if (events.length === 0) { if (deleted || previousGeneration) { rotateTranscriptGenerationInTransaction(database, resolved.sessionId); - touchTranscriptMutationInTransaction(database, resolved.sessionId); + recordTranscriptReplacementMutation( + database, + resolved.sessionId, + preservedTranscriptUpdatedAt, + ); } return; } - if (!deleted || options.preserveSessionWindowTimestamps !== true) { + if (!deleted || options.preserveSessionWindowRecency !== true) { ensureTranscriptSessionRoot(database, resolved, readEventTimestamp(events[0]) ?? Date.now()); } if (deleted || previousGeneration) { @@ -363,11 +372,27 @@ export function replaceSqliteTranscriptEventsInTransaction( } } if (deleted || seq > 0) { - touchTranscriptMutationInTransaction(database, resolved.sessionId); + recordTranscriptReplacementMutation(database, resolved.sessionId, preservedTranscriptUpdatedAt); reconcileSessionTranscriptIndexInTransaction(database.db, resolved.sessionId); } } +function recordTranscriptReplacementMutation( + database: OpenClawAgentDatabase, + sessionId: string, + preservedUpdatedAt: number | null | undefined, +): void { + if (preservedUpdatedAt === undefined || preservedUpdatedAt === null) { + touchTranscriptMutationInTransaction(database, sessionId); + return; + } + // Maintenance rewrites must invalidate in-flight projections without making an old session + // look newly active. A one-tick advance preserves ordering while changing the snapshot key. + advanceTranscriptMutationAtInTransaction(database, sessionId, preservedUpdatedAt, { + strictly: true, + }); +} + /** Rewrite existing transcript rows exactly, without append-time deduplication. */ export function rewriteSqliteTranscriptEventRowsInTransaction( database: OpenClawAgentDatabase, diff --git a/src/flows/doctor-health-contribution-runners.state.ts b/src/flows/doctor-health-contribution-runners.state.ts index c1ffb10ab993..9eb0b7237929 100644 --- a/src/flows/doctor-health-contribution-runners.state.ts +++ b/src/flows/doctor-health-contribution-runners.state.ts @@ -128,6 +128,18 @@ export async function runSessionTranscriptsHealth(ctx: DoctorHealthFlowContext): }); } +export async function runSessionTranscriptHeadersHealth( + ctx: DoctorHealthFlowContext, +): Promise { + const { noteSessionTranscriptHeaderHealth } = + await import("../commands/doctor-session-transcript-headers.js"); + await noteSessionTranscriptHeaderHealth({ + cfg: ctx.cfg, + env: ctx.env ?? process.env, + shouldRepair: ctx.prompter.shouldRepair, + }); +} + export async function runSessionTranscriptLabelsHealth( ctx: DoctorHealthFlowContext, ): Promise { diff --git a/src/flows/doctor-health-contributions-initial.ts b/src/flows/doctor-health-contributions-initial.ts index ea51282d11b1..571a3077c792 100644 --- a/src/flows/doctor-health-contributions-initial.ts +++ b/src/flows/doctor-health-contributions-initial.ts @@ -17,6 +17,7 @@ import { runSandboxHealth, runSessionLocksHealth, runSessionSnapshotsHealth, + runSessionTranscriptHeadersHealth, runSessionTranscriptLabelsHealth, runSessionTranscriptsHealth, runStateIntegrityHealth, @@ -325,6 +326,11 @@ export function resolveInitialDoctorHealthContributions(params: { }, run: runSessionTranscriptsHealth, }), + createDoctorHealthContribution({ + id: "doctor:session-transcript-headers", + label: "Session transcript headers", + run: runSessionTranscriptHeadersHealth, + }), createDoctorHealthContribution({ id: "doctor:session-transcript-labels", label: "Session transcript labels",