diff --git a/test/helpers/sqlite-sessions-transcripts-flip-proof.ts b/test/helpers/sqlite-sessions-transcripts-flip-proof.ts index b02104a68ae2..59576d2f842d 100644 --- a/test/helpers/sqlite-sessions-transcripts-flip-proof.ts +++ b/test/helpers/sqlite-sessions-transcripts-flip-proof.ts @@ -9,6 +9,7 @@ import { DatabaseSync, type SQLInputValue } from "node:sqlite"; import type { Readable } from "node:stream"; import { fileURLToPath } from "node:url"; import { expectDefined } from "@openclaw/normalization-core"; +import { asOptionalRecord as asRecord } from "@openclaw/normalization-core/record-coerce"; import { readSessionArchiveContentSync, stripSessionArchiveCompressionSuffix, @@ -44,204 +45,22 @@ type ProofChildProcess = ChildProcessByStdio; const SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS = 60_000; -type DoctorMigrationRunEvidence = { - failureReportJsonPath?: string; - failureReportMarkdownPath?: string; - manifestPath: string; - runId: string; -}; +type DoctorCommandEvidence = Awaited>; -type DoctorRestoreEvidence = { - conflicts: number; - manifestPaths: string[]; - restoredFiles: string[]; - skippedFiles: string[]; -}; +type FileInventoryEntry = Awaited>; -type DoctorCommandEvidence = { - code: number | null; - mode: DoctorMode; - migrationRun?: DoctorMigrationRunEvidence; - restore?: DoctorRestoreEvidence; - stderrTail: string; - stdoutTail: string; - totals?: Record; -}; +type ProofCheckpoint = Awaited>; +type PluginSdkConsumerEvidence = Awaited>; +type ManualCompactionEvidence = Awaited>; +type ScaleMigrationEvidence = ReturnType; +type DowngradeReupgradeEvidence = Awaited>; +type BusyContentionEvidence = Awaited>; +type SecondStartupAfterResetEvidence = Awaited>; +type RollbackRestoreEvidence = Awaited>; -type FileInventoryEntry = { - archiveReason?: "bak" | "deleted" | "reset"; - archiveSessionId?: string; - path: string; - bytes: number; - jsonlTypes?: string[]; - lines?: number; - messageRoles?: string[]; - messageTexts?: string[]; - textTail?: string; -}; - -type SqliteSessionEntryEvidence = { - entry?: Record; - sessionId: string; - sessionKey: string; - trajectoryEvents: number; - transcriptEvents: number; -}; - -type SqliteEvidence = { - exists: boolean; - path: string; - sessionEntries: number; - sessions: number; - trajectoryRuntimeEvents: number; - transcriptEvents: number; - trackedEntries: SqliteSessionEntryEvidence[]; -}; - -type ProofCheckpoint = { - activeJsonl: FileInventoryEntry[]; - archiveArtifacts: FileInventoryEntry[]; - doctor?: DoctorCommandEvidence; - gatewayLogTail?: string; - label: string; - legacyStateJsonl: FileInventoryEntry[]; - sqlite: SqliteEvidence; -}; - -type PluginSdkConsumerEvidence = { - activeJsonlForSessionExists: boolean; - activeTrajectoryPointerForSessionExists: boolean; - activeTrajectoryRuntimeSidecarForSessionExists: boolean; - activeTrajectorySessionSidecarForSessionExists: boolean; - appendedMessageId: string; - identityMemoryKey: string; - latestAssistantTextBeforeAppend: string; - latestAssistantTextAfterAppend: string; - listedSessionKeys: string[]; - sessionIdentity: string; - sessionId: string; - sessionKey: string; - storeTranscriptEvents: number; - transcriptEventsAfterAppend: number; - transcriptEventsBeforeAppend: number; -}; - -type ManualCompactionEvidence = { - checkpointCount: number; - compacted: boolean; - rowCountAfter: number; - rowCountBefore: number; - transcriptIdentity: string; - sessionId: string; - sessionKey: string; -}; - -type ScaleMigrationEvidence = { - importedSessionKeys: string[]; - minTranscriptEventsPerSession: number; - seededEvents: number; - seededSessions: number; - startupImportElapsedMs: number; -}; - -type DowngradeReupgradeEvidence = { - activeJsonlArchived: boolean; - doctorImportedEntries: number; - doctorImportedTranscriptEvents: number; - trajectoryPointerArchived: boolean; - trajectoryPointerSourceRemoved: boolean; - trajectorySidecarArchived: boolean; - trajectorySidecarSourceRemoved: boolean; - sessionId: string; - sessionKey: string; - transcriptEvents: number; -}; - -type BusyContentionEvidence = { - childExitCode: number | null; - childSignal: NodeJS.Signals | null; - elapsedMs: number; - holdMs: number; - sessionId: string; - sessionKey: string; - transcriptEvents: number; -}; - -type SecondStartupAfterResetEvidence = { - activeJsonlForSessionExists: boolean; - historyContainsPostResetAppend: boolean; - sessionId: string; - sessionKey: string; - transcriptEvents: number; -}; - -type RollbackRestoreEvidence = { - archivePath: string; - archivedBeforeRestore: boolean; - failedManifestIssueCode: string; - idempotentRestoreSkippedFiles: string[]; - manifestPath: string; - restoredFiles: string[]; - sourcePath: string; - sourceRestored: boolean; - sqliteStillExists: boolean; -}; - -type SqliteSessionsTranscriptsFlipProofReport = { - ok: boolean; - agentId: string; - checkpoints: ProofCheckpoint[]; - cleanupPruneSessionKey: string; - concurrentDeleteSessionKey: string; - concurrentResetSessionKey: string; - concurrentSendSessionKey: string; - deleteSessionKey: string; - failures: string[]; - fullTurnAssistantText: string; - fullTurnSessionKey: string; - gatewayEntrypoint: string[]; - legacySessionId: string; - manualCompaction?: ManualCompactionEvidence; - manualCompactionSessionKey: string; - mockOpenAiRequestLog: string; - oldStateSessionKeys: string[]; - pluginSdkConsumer?: PluginSdkConsumerEvidence; - pluginSdkSessionKey: string; - resetSessionKey: string; - rollbackRestore?: RollbackRestoreEvidence; - busyContention?: BusyContentionEvidence; - downgradeReupgrade?: DowngradeReupgradeEvidence; - scaleMigration?: ScaleMigrationEvidence; - secondStartupAfterReset?: SecondStartupAfterResetEvidence; - sharedSessionKeys: string[]; - stateDir: string; -}; - -type ProofContext = { - activeSessionsDir: string; - agentDbPath: string; - agentId: string; - archiveRoots: string[]; - cleanupPruneSessionKey: string; - concurrentDeleteSessionKey: string; - concurrentResetSessionKey: string; - concurrentSendSessionKey: string; - deleteSessionKey: string; - fullTurnAssistantText: string; - fullTurnSessionKey: string; - legacySessionsDir: string; - legacySessionId: string; - manualCompactionSessionKey: string; - mockOpenAiRequestLog: string; - oldStateSessionKeys: string[]; - pluginSdkAppendText: string; - pluginSdkSessionKey: string; - resetSessionKey: string; - sharedSessionKeys: string[]; - stateDir: string; - storePath: string; - trackedSessionKeys: string[]; -}; +type ProofContext = ReturnType; +type GatewayClient = Awaited>; +type OpenClawTestInstance = Awaited>; type RunOptions = { print?: boolean; @@ -288,9 +107,7 @@ const OLD_STATE_SESSION_KEYS = [ ] as const; /** Runs the isolated live gateway SQLite flip proof and returns structured evidence. */ -export async function runSqliteSessionsTranscriptsFlipProof( - options: RunOptions = {}, -): Promise { +export async function runSqliteSessionsTranscriptsFlipProof(options: RunOptions = {}) { const print = options.print ?? false; const mockOpenAiPort = await getFreeTcpPort(); const inst = await createOpenClawTestInstance({ @@ -362,13 +179,7 @@ export async function runSqliteSessionsTranscriptsFlipProof( rollbackRestore = await runRollbackRestoreProof(inst, context); await record("after-rollback-restore"); - const client = await connectGatewayClient({ - url: inst.url, - token: inst.gatewayToken, - clientDisplayName: "sqlite-sessions-transcripts-flip-proof", - requestTimeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, - timeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, - }); + const client = await connectProofClient(inst, "sqlite-sessions-transcripts-flip-proof"); try { await waitForHistoryContains(client, context.resetSessionKey, "legacy hello"); } finally { @@ -379,13 +190,10 @@ export async function runSqliteSessionsTranscriptsFlipProof( await inst.startGateway(); await record("after-gateway-restart"); - const restartedClient = await connectGatewayClient({ - url: inst.url, - token: inst.gatewayToken, - clientDisplayName: "sqlite-sessions-transcripts-flip-proof-restart", - requestTimeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, - timeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, - }); + const restartedClient = await connectProofClient( + inst, + "sqlite-sessions-transcripts-flip-proof-restart", + ); let restartedClientConnected = true; try { await waitForHistoryContains(restartedClient, context.resetSessionKey, "legacy hello"); @@ -486,13 +294,10 @@ export async function runSqliteSessionsTranscriptsFlipProof( await inst.startGateway(); await record("after-second-startup-after-reset"); - const postResetClient = await connectGatewayClient({ - url: inst.url, - token: inst.gatewayToken, - clientDisplayName: "sqlite-sessions-transcripts-flip-proof-post-reset-restart", - requestTimeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, - timeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, - }); + const postResetClient = await connectProofClient( + inst, + "sqlite-sessions-transcripts-flip-proof-post-reset-restart", + ); try { secondStartupAfterReset = await runSecondStartupAfterResetProof( postResetClient, @@ -538,7 +343,7 @@ export async function runSqliteSessionsTranscriptsFlipProof( await inst.cleanup(); } - const report: SqliteSessionsTranscriptsFlipProofReport = { + const report = { ok: failures.length === 0, agentId: context.agentId, checkpoints, @@ -578,7 +383,7 @@ function isBuiltCliEntrypoint(entrypoint: readonly string[]): boolean { return rest.length === 0 && (first === "dist/index.js" || first === "dist/index.mjs"); } -function buildProofContext(stateDir: string): ProofContext { +function buildProofContext(stateDir: string) { const agentDir = path.join(stateDir, "agents", AGENT_ID); const activeSessionsDir = path.join(agentDir, "sessions"); const legacySessionsDir = path.join(stateDir, "sessions"); @@ -688,6 +493,19 @@ async function getFreeTcpPort(): Promise { return addr.port; } +async function connectProofClient( + inst: OpenClawTestInstance, + clientDisplayName: string, +): Promise { + return await connectGatewayClient({ + url: inst.url, + token: inst.gatewayToken, + clientDisplayName, + requestTimeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, + timeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, + }); +} + async function startMockOpenAiServer(params: { port: number; requestLogPath: string; @@ -703,31 +521,34 @@ async function startMockOpenAiServer(params: { }, stdio: ["ignore", "pipe", "pipe"], }); - child.stdout.setEncoding("utf8"); - child.stderr.setEncoding("utf8"); - let output = ""; - child.stdout.on("data", (chunk) => { - output += String(chunk); - }); - child.stderr.on("data", (chunk) => { - output += String(chunk); - }); + const childOutput = captureChildOutput(child); const deadline = Date.now() + SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS; while (Date.now() < deadline) { if (child.exitCode !== null || child.signalCode !== null) { throw new Error( `mock OpenAI exited before listening (code=${String(child.exitCode)} signal=${String( child.signalCode, - )})\n${tail(output)}`, + )})\n${tail(childOutput())}`, ); } - if (output.includes("mock-openai listening")) { + if (childOutput().includes("mock-openai listening")) { return child; } await sleep(25); } await stopChildProcess(child); - throw new Error(`timeout waiting for mock OpenAI server\n${tail(output)}`); + throw new Error(`timeout waiting for mock OpenAI server\n${tail(childOutput())}`); +} + +function captureChildOutput(child: ProofChildProcess): () => string { + let output = ""; + for (const stream of [child.stdout, child.stderr]) { + stream.setEncoding("utf8"); + stream.on("data", (chunk) => { + output += String(chunk); + }); + } + return () => output; } async function stopChildProcess(child: ProofChildProcess | undefined): Promise { @@ -781,91 +602,75 @@ async function seedLegacySessionStore(context: ProofContext): Promise { for (const [index, sessionKey] of SCALE_SESSION_KEYS.entries()) { entries[sessionKey] = legacyEntry(scaleSessionId(index), now - 20_000 - index); } - await fs.writeFile(context.storePath, `${JSON.stringify(entries, null, 2)}\n`, { mode: 0o600 }); - await fs.writeFile( - path.join(context.legacySessionsDir, "sessions.json"), - `${JSON.stringify(oldStateEntries, null, 2)}\n`, - { mode: 0o600 }, + await writeJsonFile(context.storePath, entries, 2); + await writeJsonFile(path.join(context.legacySessionsDir, "sessions.json"), oldStateEntries, 2); + await writeJsonFile(path.join(context.stateDir, "agent", "old-settings.json"), { + source: "old-agent-layout", + }); + const legacyDir = context.legacySessionsDir; + const activeDir = context.activeSessionsDir; + await writeMessageTranscript(legacyDir, context.legacySessionId, "sqlite-user-1", "legacy hello"); + await writeMessageTranscript(legacyDir, "sqlite-old-direct", "sqlite-old-direct-1", "old dm"); + await writeMessageTranscript(legacyDir, "sqlite-old-group", "sqlite-old-group-1", "old group"); + await writeMessageTranscript(activeDir, "sqlite-delete-session", "sqlite-delete-1", "delete me"); + await writeMessageTranscript( + activeDir, + "sqlite-concurrent-reset", + "sqlite-concurrent-reset-1", + "concurrent reset seed", ); - await fs.writeFile( - path.join(context.stateDir, "agent", "old-settings.json"), - `${JSON.stringify({ source: "old-agent-layout" })}\n`, - { mode: 0o600 }, + await writeMessageTranscript( + activeDir, + "sqlite-concurrent-delete", + "sqlite-concurrent-delete-1", + "concurrent delete seed", + ); + await writeMessageTranscript( + activeDir, + "sqlite-shared-a", + "sqlite-shared-1", + "shared", + "sqlite-shared-session", + ); + await writeMessageTranscript( + activeDir, + "sqlite-shared-b", + "sqlite-shared-2", + "shared b", + "sqlite-shared-session", ); - await writeTranscript(context.legacySessionsDir, context.legacySessionId, [ - legacySessionEvent(context.legacySessionId), - { type: "message", id: "sqlite-user-1", message: { role: "user", content: "legacy hello" } }, - ]); - await writeTranscript(context.legacySessionsDir, "sqlite-old-direct", [ - legacySessionEvent("sqlite-old-direct"), - { type: "message", id: "sqlite-old-direct-1", message: { role: "user", content: "old dm" } }, - ]); - await writeTranscript(context.legacySessionsDir, "sqlite-old-group", [ - legacySessionEvent("sqlite-old-group"), - { type: "message", id: "sqlite-old-group-1", message: { role: "user", content: "old group" } }, - ]); - await writeTranscript(context.activeSessionsDir, "sqlite-delete-session", [ - legacySessionEvent("sqlite-delete-session"), - { type: "message", id: "sqlite-delete-1", message: { role: "user", content: "delete me" } }, - ]); - await writeTranscript(context.activeSessionsDir, "sqlite-concurrent-reset", [ - legacySessionEvent("sqlite-concurrent-reset"), - { - type: "message", - id: "sqlite-concurrent-reset-1", - message: { role: "user", content: "concurrent reset seed" }, - }, - ]); - await writeTranscript(context.activeSessionsDir, "sqlite-concurrent-delete", [ - legacySessionEvent("sqlite-concurrent-delete"), - { - type: "message", - id: "sqlite-concurrent-delete-1", - message: { role: "user", content: "concurrent delete seed" }, - }, - ]); - await writeTranscript(context.activeSessionsDir, "sqlite-shared-a", [ - legacySessionEvent("sqlite-shared-session"), - { type: "message", id: "sqlite-shared-1", message: { role: "user", content: "shared" } }, - ]); - await writeTranscript(context.activeSessionsDir, "sqlite-shared-b", [ - legacySessionEvent("sqlite-shared-session"), - { type: "message", id: "sqlite-shared-2", message: { role: "user", content: "shared b" } }, - ]); for (const [index] of SCALE_SESSION_KEYS.entries()) { const sessionId = scaleSessionId(index); await writeTranscript(context.activeSessionsDir, sessionId, [ legacySessionEvent(sessionId), - ...Array.from({ length: SCALE_EVENTS_PER_SESSION - 1 }, (_, eventIndex) => ({ - type: "message", - id: `${sessionId}-${eventIndex + 1}`, - message: { - role: eventIndex % 2 === 0 ? "user" : "assistant", - content: `sqlite scale ${index + 1} event ${eventIndex + 1}`, - }, - })), + ...Array.from({ length: SCALE_EVENTS_PER_SESSION - 1 }, (_, eventIndex) => + messageEvent( + `${sessionId}-${eventIndex + 1}`, + eventIndex % 2 === 0 ? "user" : "assistant", + `sqlite scale ${index + 1} event ${eventIndex + 1}`, + ), + ), ]); } - await fs.writeFile( + await writeJsonFile( path.join(context.legacySessionsDir, `${context.legacySessionId}.trajectory.jsonl`), - `${JSON.stringify({ type: "trajectory", sessionId: context.legacySessionId })}\n`, - { mode: 0o600 }, - ); - await fs.writeFile( - path.join(context.legacySessionsDir, "old-orphan.deleted.jsonl"), - `${JSON.stringify({ type: "event", id: "old-orphan" })}\n`, - { mode: 0o600 }, + { type: "trajectory", sessionId: context.legacySessionId }, ); + await writeJsonFile(path.join(context.legacySessionsDir, "old-orphan.deleted.jsonl"), { + type: "event", + id: "old-orphan", + }); const archiveDir = path.join(context.activeSessionsDir, "archive-fixture"); await fs.mkdir(archiveDir, { recursive: true }); - await fs.writeFile( - path.join(archiveDir, "cold-archive.jsonl"), - `${JSON.stringify({ type: "event", id: "cold-archive" })}\n`, - { mode: 0o600 }, - ); - await importSqliteSessionRows({ - agentId: context.agentId, - entry: { + await writeJsonFile(path.join(archiveDir, "cold-archive.jsonl"), { + type: "event", + id: "cold-archive", + }); + await importProofSession( + context, + "agent:main:partial-direct", + "sqlite-partial-import", + { ...legacyEntry("sqlite-partial-import", now - 7_000), sessionFile: formatSqliteSessionFileMarker({ agentId: context.agentId, @@ -873,17 +678,8 @@ async function seedLegacySessionStore(context: ProofContext): Promise { storePath: context.storePath, }), }, - readTranscriptEvents(append) { - append(legacySessionEvent("sqlite-partial-import")); - append({ - type: "message", - id: "sqlite-partial-import-1", - message: { role: "user", content: "already imported" }, - }); - }, - sessionKey: "agent:main:partial-direct", - storePath: context.storePath, - }); + [messageEvent("sqlite-partial-import-1", "user", "already imported")], + ); } function scaleSessionId(index: number): string { @@ -910,6 +706,27 @@ function legacySessionEvent(sessionId: string): TranscriptEvent { return { type: "session", sessionId }; } +function messageEvent(id: string, role: "assistant" | "user", content: string): TranscriptEvent { + return { type: "message", id, message: { role, content } }; +} + +async function writeJsonFile(filePath: string, value: unknown, indent?: number): Promise { + await fs.writeFile(filePath, `${JSON.stringify(value, null, indent)}\n`, { mode: 0o600 }); +} + +async function writeMessageTranscript( + sessionsDir: string, + fileId: string, + messageId: string, + content: string, + sessionId = fileId, +): Promise { + await writeTranscript(sessionsDir, fileId, [ + legacySessionEvent(sessionId), + messageEvent(messageId, "user", content), + ]); +} + async function writeTranscript( sessionsDir: string, sessionId: string, @@ -919,11 +736,26 @@ async function writeTranscript( await fs.writeFile(path.join(sessionsDir, `${sessionId}.jsonl`), `${body}\n`, { mode: 0o600 }); } -async function runDoctor( - inst: Awaited>, - mode: DoctorMode, - storePath: string, -): Promise { +async function importProofSession( + context: ProofContext, + sessionKey: string, + sessionId: string, + entry: SessionEntry, + events: TranscriptEvent[], +) { + return await importSqliteSessionRows({ + agentId: context.agentId, + entry, + readTranscriptEvents(append) { + append(legacySessionEvent(sessionId)); + events.forEach(append); + }, + sessionKey, + storePath: context.storePath, + }); +} + +async function runDoctor(inst: OpenClawTestInstance, mode: DoctorMode, storePath: string) { const result = await inst.cli( ["doctor", "--session-sqlite", mode, "--session-sqlite-store", storePath, "--json"], { timeoutMs: 60_000 }, @@ -942,9 +774,7 @@ async function runDoctor( }; } -function parseDoctorMigrationRun(parsed: Record): { - migrationRun?: DoctorMigrationRunEvidence; -} { +function parseDoctorMigrationRun(parsed: Record) { const migrationRun = asRecord(parsed.migrationRun); if ( !migrationRun || @@ -967,7 +797,7 @@ function parseDoctorMigrationRun(parsed: Record): { }; } -function parseDoctorRestore(parsed: Record): { restore?: DoctorRestoreEvidence } { +function parseDoctorRestore(parsed: Record) { const targets = Array.isArray(parsed.targets) ? parsed.targets : []; const restore = asRecord(asRecord(targets[0])?.restore); if (!restore) { @@ -983,10 +813,7 @@ function parseDoctorRestore(parsed: Record): { restore?: Doctor }; } -async function runRollbackRestoreProof( - inst: Awaited>, - context: ProofContext, -): Promise { +async function runRollbackRestoreProof(inst: OpenClawTestInstance, context: ProofContext) { const drillDir = path.join(context.stateDir, "rollback-drill"); const storePath = path.join(drillDir, "sessions.json"); const sessionId = "sqlite-rollback-restore"; @@ -994,19 +821,13 @@ async function runRollbackRestoreProof( const sourcePath = path.join(drillDir, `${sessionId}.jsonl`); const sqlitePath = path.join(drillDir, "openclaw-agent.sqlite"); await fs.mkdir(drillDir, { recursive: true }); - await fs.writeFile( - storePath, - `${JSON.stringify({ [sessionKey]: legacyEntry(sessionId, Date.now()) }, null, 2)}\n`, - { mode: 0o600 }, + await writeJsonFile(storePath, { [sessionKey]: legacyEntry(sessionId, Date.now()) }, 2); + await writeMessageTranscript( + drillDir, + sessionId, + "sqlite-rollback-restore-1", + "rollback restore me", ); - await writeTranscript(drillDir, sessionId, [ - legacySessionEvent(sessionId), - { - type: "message", - id: "sqlite-rollback-restore-1", - message: { role: "user", content: "rollback restore me" }, - }, - ]); const importDoctor = await runDoctor(inst, "import", storePath); if (importDoctor.code !== 0) { @@ -1131,7 +952,7 @@ async function markMigrationManifestFailed(manifestPath: string, issueCode: stri }, ]; } - await fs.writeFile(manifestPath, `${JSON.stringify(manifest, null, 2)}\n`, { mode: 0o600 }); + await writeJsonFile(manifestPath, manifest, 2); } async function appendProofMessage( @@ -1160,10 +981,7 @@ async function appendProofMessage( } } -async function runManualCompactionProof( - client: Awaited>, - context: ProofContext, -): Promise { +async function runManualCompactionProof(client: GatewayClient, context: ProofContext) { const runId = await sendGatewayUserMessage( client, context.manualCompactionSessionKey, @@ -1243,10 +1061,7 @@ async function runManualCompactionProof( }; } -async function runPluginSdkConsumerProbe( - context: ProofContext, - sessionId: string, -): Promise { +async function runPluginSdkConsumerProbe(context: ProofContext, sessionId: string) { const scope = { agentId: context.agentId, sessionId, @@ -1285,31 +1100,14 @@ async function runPluginSdkConsumerProbe( } const transcriptEventsBeforeAppend = (await readSessionTranscriptEvents(scope)).length; const storeTranscriptEvents = loadSdkTranscriptEventsSync(scope).length; - const activeJsonlPath = path.join(context.activeSessionsDir, `${sessionId}.jsonl`); - const activeTrajectorySessionSidecarPath = path.join( - context.activeSessionsDir, - `${sessionId}.trajectory.jsonl`, - ); - const activeTrajectoryPointerPath = path.join( - context.activeSessionsDir, - `${sessionId}.trajectory-path.json`, - ); - const activeTrajectoryRuntimeSidecarPath = path.join( - context.activeSessionsDir, - "trajectory", - `${sessionId}.jsonl`, - ); - const activeJsonlForSessionExists = fsSync.existsSync(activeJsonlPath); + const artifacts = sessionArtifactPaths(context.activeSessionsDir, sessionId); + const activeJsonlForSessionExists = fsSync.existsSync(artifacts.jsonl); if (activeJsonlForSessionExists) { - throw new Error(`SDK probe found active JSONL for SQLite session at ${activeJsonlPath}`); + throw new Error(`SDK probe found active JSONL for SQLite session at ${artifacts.jsonl}`); } - const activeTrajectorySessionSidecarForSessionExists = fsSync.existsSync( - activeTrajectorySessionSidecarPath, - ); - const activeTrajectoryPointerForSessionExists = fsSync.existsSync(activeTrajectoryPointerPath); - const activeTrajectoryRuntimeSidecarForSessionExists = fsSync.existsSync( - activeTrajectoryRuntimeSidecarPath, - ); + const activeTrajectorySessionSidecarForSessionExists = fsSync.existsSync(artifacts.trajectory); + const activeTrajectoryPointerForSessionExists = fsSync.existsSync(artifacts.pointer); + const activeTrajectoryRuntimeSidecarForSessionExists = fsSync.existsSync(artifacts.runtime); if ( activeTrajectorySessionSidecarForSessionExists || activeTrajectoryPointerForSessionExists || @@ -1317,9 +1115,9 @@ async function runPluginSdkConsumerProbe( ) { throw new Error( `SDK trajectory probe found active sidecar paths: ${JSON.stringify({ - pointer: activeTrajectoryPointerPath, - runtime: activeTrajectoryRuntimeSidecarPath, - session: activeTrajectorySessionSidecarPath, + pointer: artifacts.pointer, + runtime: artifacts.runtime, + session: artifacts.trajectory, })}`, ); } @@ -1369,14 +1167,25 @@ async function runPluginSdkConsumerProbe( }; } +function sessionArtifactPaths(sessionsDir: string, sessionId: string) { + return { + jsonl: path.join(sessionsDir, `${sessionId}.jsonl`), + pointer: path.join(sessionsDir, `${sessionId}.trajectory-path.json`), + runtime: path.join(sessionsDir, "trajectory", `${sessionId}.jsonl`), + trajectory: path.join(sessionsDir, `${sessionId}.trajectory.jsonl`), + }; +} + async function runGatewayCleanupPruningProof( - client: Awaited>, + client: GatewayClient, context: ProofContext, ): Promise { const updatedAt = Date.now() - 31 * 24 * 60 * 60 * 1000; - await importSqliteSessionRows({ - agentId: context.agentId, - entry: { + await importProofSession( + context, + context.cleanupPruneSessionKey, + CLEANUP_PRUNE_SESSION_ID, + { delivery: normalizeSessionDeliveryState({ context: { channel: "cli" } }), chatType: "direct", sessionFile: formatSqliteSessionFileMarker({ @@ -1388,17 +1197,8 @@ async function runGatewayCleanupPruningProof( sessionStartedAt: updatedAt - 500, updatedAt, }, - readTranscriptEvents(append) { - append(legacySessionEvent(CLEANUP_PRUNE_SESSION_ID)); - append({ - type: "message", - id: "sqlite-cleanup-prune-1", - message: { role: "user", content: CLEANUP_PRUNE_TEXT }, - }); - }, - sessionKey: context.cleanupPruneSessionKey, - storePath: context.storePath, - }); + [messageEvent("sqlite-cleanup-prune-1", "user", CLEANUP_PRUNE_TEXT)], + ); await waitForSqliteMessageContains( context.agentDbPath, CLEANUP_PRUNE_SESSION_ID, @@ -1419,7 +1219,7 @@ async function runGatewayCleanupPruningProof( } async function runDoctorIdempotenceProof( - inst: Awaited>, + inst: OpenClawTestInstance, context: ProofContext, ): Promise { const before = readSqliteEvidence(context.agentDbPath, context.trackedSessionKeys); @@ -1446,10 +1246,7 @@ async function runDoctorIdempotenceProof( return doctor; } -function requireScaleMigrationProof( - context: ProofContext, - startupImportElapsedMs: number, -): ScaleMigrationEvidence { +function requireScaleMigrationProof(context: ProofContext, startupImportElapsedMs: number) { const sqlite = readSqliteEvidence(context.agentDbPath, SCALE_SESSION_KEYS); const importedSessionKeys: string[] = []; for (const [index, sessionKey] of SCALE_SESSION_KEYS.entries()) { @@ -1474,57 +1271,36 @@ function requireScaleMigrationProof( }; } -async function runDowngradeReupgradeProof( - inst: Awaited>, - context: ProofContext, -): Promise { +async function runDowngradeReupgradeProof(inst: OpenClawTestInstance, context: ProofContext) { await fs.mkdir(context.activeSessionsDir, { recursive: true }); - await fs.writeFile( + await writeJsonFile( context.storePath, - `${JSON.stringify( - { - [DOWNGRADE_REUPGRADE_SESSION_KEY]: legacyEntry(DOWNGRADE_REUPGRADE_SESSION_ID, Date.now()), - }, - null, - 2, - )}\n`, - { mode: 0o600 }, - ); - const transcriptPath = path.join( - context.activeSessionsDir, - `${DOWNGRADE_REUPGRADE_SESSION_ID}.jsonl`, - ); - const trajectoryPath = path.join( - context.activeSessionsDir, - `${DOWNGRADE_REUPGRADE_SESSION_ID}.trajectory.jsonl`, - ); - const trajectoryPointerPath = path.join( - context.activeSessionsDir, - `${DOWNGRADE_REUPGRADE_SESSION_ID}.trajectory-path.json`, - ); - await writeTranscript(context.activeSessionsDir, DOWNGRADE_REUPGRADE_SESSION_ID, [ - legacySessionEvent(DOWNGRADE_REUPGRADE_SESSION_ID), { - type: "message", - id: "sqlite-downgrade-reupgrade-1", - message: { role: "user", content: DOWNGRADE_REUPGRADE_TEXT }, + [DOWNGRADE_REUPGRADE_SESSION_KEY]: legacyEntry(DOWNGRADE_REUPGRADE_SESSION_ID, Date.now()), }, - ]); - await fs.writeFile( - trajectoryPath, - `${JSON.stringify({ type: "trajectory", sessionId: DOWNGRADE_REUPGRADE_SESSION_ID })}\n`, - { mode: 0o600 }, + 2, ); - await fs.writeFile( - trajectoryPointerPath, - `${JSON.stringify({ - traceSchema: "openclaw-trajectory-pointer", - schemaVersion: 1, - sessionId: DOWNGRADE_REUPGRADE_SESSION_ID, - runtimeFile: trajectoryPath, - })}\n`, - { mode: 0o600 }, + const { + jsonl: transcriptPath, + pointer: trajectoryPointerPath, + trajectory: trajectoryPath, + } = sessionArtifactPaths(context.activeSessionsDir, DOWNGRADE_REUPGRADE_SESSION_ID); + await writeMessageTranscript( + context.activeSessionsDir, + DOWNGRADE_REUPGRADE_SESSION_ID, + "sqlite-downgrade-reupgrade-1", + DOWNGRADE_REUPGRADE_TEXT, ); + await writeJsonFile(trajectoryPath, { + type: "trajectory", + sessionId: DOWNGRADE_REUPGRADE_SESSION_ID, + }); + await writeJsonFile(trajectoryPointerPath, { + traceSchema: "openclaw-trajectory-pointer", + schemaVersion: 1, + sessionId: DOWNGRADE_REUPGRADE_SESSION_ID, + runtimeFile: trajectoryPath, + }); const doctor = await runDoctor(inst, "import", context.storePath); if (doctor.code !== 0) { @@ -1582,9 +1358,7 @@ async function runDowngradeReupgradeProof( }; } -async function runSqliteBusyContentionProof( - context: ProofContext, -): Promise { +async function runSqliteBusyContentionProof(context: ProofContext) { const holdMs = 500; const readyPath = path.join(context.stateDir, `sqlite-busy-${randomUUID()}.ready`); const child = spawn( @@ -1615,48 +1389,33 @@ async function runSqliteBusyContentionProof( stdio: ["ignore", "pipe", "pipe"], }, ); - let childOutput = ""; - child.stdout.setEncoding("utf8"); - child.stderr.setEncoding("utf8"); - child.stdout.on("data", (chunk) => { - childOutput += String(chunk); - }); - child.stderr.on("data", (chunk) => { - childOutput += String(chunk); - }); + const childOutput = captureChildOutput(child); await waitForFile(readyPath, SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, () => { if (child.exitCode !== null || child.signalCode !== null) { throw new Error( `SQLite busy child exited before acquiring lock code=${String( child.exitCode, - )} signal=${String(child.signalCode)} ${tail(childOutput)}`, + )} signal=${String(child.signalCode)} ${tail(childOutput())}`, ); } }); const startedAt = Date.now(); - const result = await importSqliteSessionRows({ - agentId: context.agentId, - entry: legacyEntry(SQLITE_BUSY_SESSION_ID, Date.now()), - readTranscriptEvents(append) { - append(legacySessionEvent(SQLITE_BUSY_SESSION_ID)); - append({ - type: "message", - id: "sqlite-busy-contention-1", - message: { role: "user", content: SQLITE_BUSY_TEXT }, - }); - }, - sessionKey: SQLITE_BUSY_SESSION_KEY, - storePath: context.storePath, - }); + const result = await importProofSession( + context, + SQLITE_BUSY_SESSION_KEY, + SQLITE_BUSY_SESSION_ID, + legacyEntry(SQLITE_BUSY_SESSION_ID, Date.now()), + [messageEvent("sqlite-busy-contention-1", "user", SQLITE_BUSY_TEXT)], + ); const elapsedMs = Date.now() - startedAt; const exit = await waitForChildExit(child, SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS); if (exit.code !== 0) { throw new Error( `SQLite busy child exited non-zero code=${String(exit.code)} signal=${String( exit.signal, - )} ${tail(childOutput)}`, + )} ${tail(childOutput())}`, ); } if (elapsedMs < Math.floor(holdMs / 2)) { @@ -1680,10 +1439,10 @@ async function runSqliteBusyContentionProof( } async function runSecondStartupAfterResetProof( - client: Awaited>, + client: GatewayClient, context: ProofContext, resetSessionId: string, -): Promise { +) { await appendProofMessage( context, resetSessionId, @@ -1704,24 +1463,18 @@ async function runSecondStartupAfterResetProof( } async function runConcurrentMultiClientLifecycle( - inst: Awaited>, + inst: OpenClawTestInstance, context: ProofContext, - primaryClient: Awaited>, + primaryClient: GatewayClient, ): Promise { - const historyClient = await connectGatewayClient({ - url: inst.url, - token: inst.gatewayToken, - clientDisplayName: "sqlite-sessions-transcripts-flip-proof-concurrent-history", - requestTimeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, - timeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, - }); - const lifecycleClient = await connectGatewayClient({ - url: inst.url, - token: inst.gatewayToken, - clientDisplayName: "sqlite-sessions-transcripts-flip-proof-concurrent-lifecycle", - requestTimeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, - timeoutMs: SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS, - }); + const historyClient = await connectProofClient( + inst, + "sqlite-sessions-transcripts-flip-proof-concurrent-history", + ); + const lifecycleClient = await connectProofClient( + inst, + "sqlite-sessions-transcripts-flip-proof-concurrent-lifecycle", + ); try { await requireHistoryContains( historyClient, @@ -1792,10 +1545,7 @@ async function runConcurrentMultiClientLifecycle( } } -async function resetSession( - client: Awaited>, - key: string, -): Promise { +async function resetSession(client: GatewayClient, key: string): Promise { const result: { ok?: boolean; entry?: { sessionId?: string } } = await client.request( "sessions.reset", { key, reason: "reset" }, @@ -1807,7 +1557,7 @@ async function resetSession( } async function sendGatewayUserMessage( - client: Awaited>, + client: GatewayClient, sessionKey: string, message: string, ): Promise { @@ -1826,10 +1576,7 @@ async function sendGatewayUserMessage( return result.runId; } -async function waitForAgentRunOk( - client: Awaited>, - runId: string, -): Promise { +async function waitForAgentRunOk(client: GatewayClient, runId: string): Promise { const result: { error?: unknown; status?: string } = await client.request( "agent.wait", { runId, timeoutMs: 60_000 }, @@ -1840,10 +1587,7 @@ async function waitForAgentRunOk( } } -async function waitForAgentRunSettled( - client: Awaited>, - runId: string, -): Promise { +async function waitForAgentRunSettled(client: GatewayClient, runId: string): Promise { const result: { endedAt?: number; status?: string; stopReason?: string } = await client.request( "agent.wait", { runId, timeoutMs: 60_000 }, @@ -1866,10 +1610,7 @@ async function waitForAgentRunSettled( throw new Error(`agent.wait did not settle acceptably for ${runId}: ${JSON.stringify(result)}`); } -async function deleteSession( - client: Awaited>, - key: string, -): Promise { +async function deleteSession(client: GatewayClient, key: string): Promise { const result: { ok?: boolean; deleted?: boolean } = await client.request("sessions.delete", { key, deleteTranscript: true, @@ -1879,10 +1620,7 @@ async function deleteSession( } } -async function requireTrackedSession( - client: Awaited>, - key: string, -): Promise { +async function requireTrackedSession(client: GatewayClient, key: string): Promise { const result: { sessions?: Array<{ key?: string }> } = await client.request("sessions.list", {}); if (!result.sessions?.some((session) => session.key === key)) { throw new Error(`expected session ${key} to remain listed`); @@ -1890,7 +1628,7 @@ async function requireTrackedSession( } async function requireHistoryContains( - client: Awaited>, + client: GatewayClient, sessionKey: string, expected: string, ): Promise { @@ -1910,7 +1648,7 @@ async function requireHistoryContains( } async function requireHistoryRoleContains( - client: Awaited>, + client: GatewayClient, sessionKey: string, role: string, expected: string, @@ -1938,38 +1676,55 @@ async function requireHistoryRoleContains( } async function waitForHistoryContains( - client: Awaited>, + client: GatewayClient, sessionKey: string, expected: string, ): Promise { - const deadline = Date.now() + 30_000; - while (Date.now() < deadline) { - try { - await requireHistoryContains(client, sessionKey, expected); - return; - } catch { - await sleep(50); - } - } - await requireHistoryContains(client, sessionKey, expected); + await waitForAssertion(() => requireHistoryContains(client, sessionKey, expected)); } async function waitForHistoryRoleContains( - client: Awaited>, + client: GatewayClient, sessionKey: string, role: string, expected: string, ): Promise { + await waitForAssertion(() => requireHistoryRoleContains(client, sessionKey, role, expected)); +} + +async function waitForAssertion(assertion: () => Promise): Promise { const deadline = Date.now() + 30_000; while (Date.now() < deadline) { try { - await requireHistoryRoleContains(client, sessionKey, role, expected); - return; + return await assertion(); } catch { await sleep(50); } } - await requireHistoryRoleContains(client, sessionKey, role, expected); + await assertion(); +} + +async function pollUntil( + read: () => T | Promise, + ready: (value: T) => boolean, + timeoutError: () => Error, + options: { intervalMs?: number; stableMs?: number; timeoutMs?: number } = {}, +): Promise { + const deadline = Date.now() + (options.timeoutMs ?? SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS); + let readySince: number | undefined; + while (Date.now() < deadline) { + const value = await read(); + if (ready(value)) { + readySince ??= Date.now(); + if (Date.now() - readySince >= (options.stableMs ?? 0)) { + return value; + } + } else { + readySince = undefined; + } + await sleep(options.intervalMs ?? 50); + } + throw timeoutError(); } async function waitForSqliteEvents( @@ -1977,30 +1732,24 @@ async function waitForSqliteEvents( sessionId: string, minEvents: number, ): Promise { - const deadline = Date.now() + SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS; - while (Date.now() < deadline) { - const sqlite = readSqliteEvidence(dbPath, []); - const row = sqlite.trackedEntries.find((entry) => entry.sessionId === sessionId); - if ((row?.transcriptEvents ?? 0) >= minEvents) { - return; - } - await sleep(50); - } - throw new Error(`timed out waiting for ${minEvents} SQLite events for ${sessionId}`); + await pollUntil( + () => + readSqliteEvidence(dbPath, []).trackedEntries.find((entry) => entry.sessionId === sessionId) + ?.transcriptEvents ?? 0, + (eventCount) => eventCount >= minEvents, + () => new Error(`timed out waiting for ${minEvents} SQLite events for ${sessionId}`), + ); } async function waitForSqliteSessionId(dbPath: string, sessionKey: string): Promise { - const deadline = Date.now() + SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS; - while (Date.now() < deadline) { - const row = readSqliteEvidence(dbPath, [sessionKey]).trackedEntries.find( - (entry) => entry.sessionKey === sessionKey && entry.sessionId, - ); - if (row?.sessionId) { - return row.sessionId; - } - await sleep(50); - } - throw new Error(`timed out waiting for SQLite session entry for ${sessionKey}`); + return await pollUntil( + () => + readSqliteEvidence(dbPath, [sessionKey]).trackedEntries.find( + (entry) => entry.sessionKey === sessionKey, + )?.sessionId ?? "", + Boolean, + () => new Error(`timed out waiting for SQLite session entry for ${sessionKey}`), + ); } async function waitForTrackedSessionId( @@ -2008,59 +1757,38 @@ async function waitForTrackedSessionId( sessionKey: string, expectedSessionId: string, ): Promise { - const deadline = Date.now() + SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS; - while (Date.now() < deadline) { - const row = readSqliteEvidence(dbPath, [sessionKey]).trackedEntries.find( - (entry) => entry.sessionKey === sessionKey, - ); - if (row?.sessionId === expectedSessionId) { - return; - } - await sleep(50); - } - throw new Error( - `timed out waiting for SQLite session entry ${sessionKey} to point at ${expectedSessionId}`, + await pollUntil( + () => + readSqliteEvidence(dbPath, [sessionKey]).trackedEntries.find( + (entry) => entry.sessionKey === sessionKey, + )?.sessionId, + (sessionId) => sessionId === expectedSessionId, + () => + new Error( + `timed out waiting for SQLite session entry ${sessionKey} to point at ${expectedSessionId}`, + ), ); } async function waitForSessionEntryAbsent(dbPath: string, sessionKey: string): Promise { - const deadline = Date.now() + SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS; - let absentSince: number | undefined; - while (Date.now() < deadline) { - const row = readSqliteEvidence(dbPath, [sessionKey]).trackedEntries.find( - (entry) => entry.sessionKey === sessionKey, - ); - if (!row) { - absentSince ??= Date.now(); - if (Date.now() - absentSince >= 1_500) { - return; - } - await sleep(50); - continue; - } - absentSince = undefined; - await sleep(50); - } - throw new Error(`timed out waiting for SQLite session entry deletion for ${sessionKey}`); + await pollUntil( + () => + readSqliteEvidence(dbPath, [sessionKey]).trackedEntries.some( + (entry) => entry.sessionKey === sessionKey, + ), + (exists) => !exists, + () => new Error(`timed out waiting for SQLite session entry deletion for ${sessionKey}`), + { stableMs: 1_500 }, + ); } async function waitForSqliteEventsAbsent(dbPath: string, sessionId: string): Promise { - const deadline = Date.now() + SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS; - let absentSince: number | undefined; - while (Date.now() < deadline) { - const count = countSqliteTranscriptEvents(dbPath, sessionId); - if (count === 0) { - absentSince ??= Date.now(); - if (Date.now() - absentSince >= 1_500) { - return; - } - await sleep(50); - continue; - } - absentSince = undefined; - await sleep(50); - } - throw new Error(`timed out waiting for SQLite transcript row deletion for ${sessionId}`); + await pollUntil( + () => countSqliteTranscriptEvents(dbPath, sessionId), + (eventCount) => eventCount === 0, + () => new Error(`timed out waiting for SQLite transcript row deletion for ${sessionId}`), + { stableMs: 1_500 }, + ); } async function waitForSqliteMessageContains( @@ -2069,35 +1797,31 @@ async function waitForSqliteMessageContains( role: "assistant" | "user", expected: string, ): Promise { - const deadline = Date.now() + SQLITE_FLIP_PROOF_OPERATION_TIMEOUT_MS; - while (Date.now() < deadline) { - const messages = readSqliteTranscriptMessages(dbPath, sessionId); - if ( + await pollUntil( + () => readSqliteTranscriptMessages(dbPath, sessionId), + (messages) => messages.some( (message) => message.role === role && JSON.stringify(message.content).includes(expected), - ) - ) { - return; - } - await sleep(50); - } - throw new Error( - `timed out waiting for SQLite ${role} transcript message containing ${JSON.stringify( - expected, - )} for ${sessionId}: ${JSON.stringify(readSqliteTranscriptMessages(dbPath, sessionId))}`, + ), + () => + new Error( + `timed out waiting for SQLite ${role} transcript message containing ${JSON.stringify( + expected, + )} for ${sessionId}: ${JSON.stringify(readSqliteTranscriptMessages(dbPath, sessionId))}`, + ), ); } async function waitForFile(filePath: string, timeoutMs: number, poll: () => void): Promise { - const deadline = Date.now() + timeoutMs; - while (Date.now() < deadline) { - poll(); - if (fsSync.existsSync(filePath)) { - return; - } - await sleep(25); - } - throw new Error(`timed out waiting for ${filePath}`); + await pollUntil( + () => { + poll(); + return fsSync.existsSync(filePath); + }, + Boolean, + () => new Error(`timed out waiting for ${filePath}`), + { intervalMs: 25, timeoutMs }, + ); } async function waitForChildExit( @@ -2139,11 +1863,7 @@ function readSqliteTranscriptMessages( dbPath: string, sessionId: string, ): Array<{ content?: unknown; role?: string }> { - if (!fsSync.existsSync(dbPath)) { - return []; - } - const db = new DatabaseSync(dbPath, { readOnly: true }); - try { + return withReadOnlyDatabase(dbPath, [], (db) => { const rows = db .prepare( `SELECT event_json AS eventJson @@ -2165,32 +1885,22 @@ function readSqliteTranscriptMessages( ? [{ role: message.role, content: message.content }] : []; }); - } finally { - db.close(); - } + }); } function countSqliteTranscriptEvents(dbPath: string, sessionId: string): number { - if (!fsSync.existsSync(dbPath)) { - return 0; - } - const db = new DatabaseSync(dbPath, { readOnly: true }); - try { - return scalarNumber( - db, - "SELECT COUNT(*) AS count FROM transcript_events WHERE session_id = ?", - [sessionId], - ); - } finally { - db.close(); - } + return withReadOnlyDatabase(dbPath, 0, (db) => + scalarNumber(db, "SELECT COUNT(*) AS count FROM transcript_events WHERE session_id = ?", [ + sessionId, + ]), + ); } async function captureCheckpoint( context: ProofContext, label: string, options: { doctor?: DoctorCommandEvidence; gatewayLogTail?: string }, -): Promise { +) { return { activeJsonl: await inventoryActiveJsonl(context.activeSessionsDir), archiveArtifacts: await inventoryArchiveArtifacts(context), @@ -2259,7 +1969,7 @@ async function walkFiles(root: string, visit: (filePath: string) => Promise { +async function inventoryFile(filePath: string, relativeRoot: string) { const stat = await fs.stat(filePath); const text = await Promise.resolve() .then(() => readSessionArchiveContentSync(filePath)) @@ -2281,7 +1991,7 @@ async function inventoryFile(filePath: string, relativeRoot: string): Promise | undefined { +): { archiveReason: "bak" | "deleted" | "reset"; archiveSessionId: string } | undefined { const normalized = stripSessionArchiveCompressionSuffix(fileName); for (const archiveReason of ["deleted", "reset", "bak"] as const) { const marker = `.jsonl.${archiveReason}.`; @@ -2332,9 +2042,10 @@ function summarizeJsonl(text: string): { }; } -function readSqliteEvidence(dbPath: string, trackedSessionKeys: readonly string[]): SqliteEvidence { - if (!fsSync.existsSync(dbPath)) { - return { +function readSqliteEvidence(dbPath: string, trackedSessionKeys: readonly string[]) { + return withReadOnlyDatabase( + dbPath, + { exists: false, path: dbPath, sessionEntries: 0, @@ -2342,12 +2053,8 @@ function readSqliteEvidence(dbPath: string, trackedSessionKeys: readonly string[ trackedEntries: [], trajectoryRuntimeEvents: 0, transcriptEvents: 0, - }; - } - const db = new DatabaseSync(dbPath, { readOnly: true }); - try { - const trackedEntries = readTrackedEntries(db, trackedSessionKeys); - return { + }, + (db) => ({ exists: true, path: dbPath, sessionEntries: scalarNumber(db, "SELECT COUNT(*) AS count FROM session_nodes"), @@ -2356,18 +2063,25 @@ function readSqliteEvidence(dbPath: string, trackedSessionKeys: readonly string[ db, "SELECT COUNT(*) AS count FROM trajectory_runtime_events", ), - trackedEntries, + trackedEntries: readTrackedEntries(db, trackedSessionKeys), transcriptEvents: scalarNumber(db, "SELECT COUNT(*) AS count FROM transcript_events"), - }; + }), + ); +} + +function withReadOnlyDatabase(dbPath: string, fallback: T, read: (db: DatabaseSync) => T): T { + if (!fsSync.existsSync(dbPath)) { + return fallback; + } + const db = new DatabaseSync(dbPath, { readOnly: true }); + try { + return read(db); } finally { db.close(); } } -function readTrackedEntries( - db: DatabaseSync, - trackedSessionKeys: readonly string[], -): SqliteSessionEntryEvidence[] { +function readTrackedEntries(db: DatabaseSync, trackedSessionKeys: readonly string[]) { const rows = db .prepare( `SELECT session_key AS sessionKey, current_session_id AS sessionId, entry_json AS entryJson @@ -2383,7 +2097,9 @@ function readTrackedEntries( ) .map((row) => { const sessionId = typeof row.sessionId === "string" ? row.sessionId : ""; - const result: SqliteSessionEntryEvidence = { + const entry = typeof row.entryJson === "string" ? parseEntryJson(row.entryJson) : undefined; + return { + ...(entry ? { entry } : {}), sessionId, sessionKey: typeof row.sessionKey === "string" ? row.sessionKey : "", trajectoryEvents: scalarNumber( @@ -2397,10 +2113,6 @@ function readTrackedEntries( [sessionId], ), }; - if (typeof row.entryJson === "string") { - result.entry = parseEntryJson(row.entryJson); - } - return result; }); } @@ -2428,19 +2140,19 @@ function validateCheckpointInvariants( checkpoint: ProofCheckpoint, failures: string[], ): void { - if (checkpoint.label !== "seeded-legacy-store" && checkpoint.activeJsonl.length > 0) { - failures.push( - `${checkpoint.label}: active sessions directory still has JSONL files: ${checkpoint.activeJsonl - .map((entry) => entry.path) - .join(", ")}`, - ); - } - if (checkpoint.label !== "seeded-legacy-store" && checkpoint.legacyStateJsonl.length > 0) { - failures.push( - `${checkpoint.label}: old sessions directory still has JSONL files: ${checkpoint.legacyStateJsonl - .map((entry) => entry.path) - .join(", ")}`, - ); + if (checkpoint.label !== "seeded-legacy-store") { + for (const [description, inventory] of [ + ["active sessions directory", checkpoint.activeJsonl], + ["old sessions directory", checkpoint.legacyStateJsonl], + ] as const) { + if (inventory.length > 0) { + failures.push( + `${checkpoint.label}: ${description} still has JSONL files: ${inventory + .map((entry) => entry.path) + .join(", ")}`, + ); + } + } } const doctor = checkpoint.doctor; if (checkpoint.label.startsWith("after-doctor") && doctor?.code !== 0) { @@ -2591,18 +2303,12 @@ function findArchiveArtifact( sessionId?: string; }, ): FileInventoryEntry | undefined { - return checkpoint.archiveArtifacts.find((artifact) => { - if (params.pathIncludes && !artifact.path.includes(params.pathIncludes)) { - return false; - } - if (params.reason && artifact.archiveReason !== params.reason) { - return false; - } - if (params.sessionId && artifact.archiveSessionId !== params.sessionId) { - return false; - } - return true; - }); + return checkpoint.archiveArtifacts.find( + (artifact) => + (!params.pathIncludes || artifact.path.includes(params.pathIncludes)) && + (!params.reason || artifact.archiveReason === params.reason) && + (!params.sessionId || artifact.archiveSessionId === params.sessionId), + ); } function parseJsonObject(text: string): Record | undefined { @@ -2621,12 +2327,6 @@ function parseJsonObject(text: string): Record | undefined { } } -function asRecord(value: unknown): Record | undefined { - return value && typeof value === "object" && !Array.isArray(value) - ? (value as Record) - : undefined; -} - function stringArray(value: unknown): string[] { return Array.isArray(value) ? value.filter((item): item is string => typeof item === "string")