diff --git a/packages/storage/src/__tests__/session-metadata-maintenance.test.ts b/packages/storage/src/__tests__/session-metadata-maintenance.test.ts new file mode 100644 index 00000000000..cb2f420430f --- /dev/null +++ b/packages/storage/src/__tests__/session-metadata-maintenance.test.ts @@ -0,0 +1,126 @@ +import assert from 'node:assert/strict'; +import { mkdtemp, readFile, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, test } from 'node:test'; +import type { CreateSessionInput } from '@maka/core'; +import { + backupSessionMetadataDatabase, + exportLegacySessionTree, + SESSION_METADATA_EXPORT_FORMAT, +} from '../session-metadata-maintenance.js'; +import { + createLegacyFileSessionStore, + createSessionStore, + SQLITE_SESSION_METADATA_DATABASE_NAME, +} from '../session-store.js'; +import { createSqliteSessionMetadataStore } from '../sqlite-session-metadata-store.js'; + +describe('session metadata migration maintenance', () => { + test('exports a legacy-compatible tree and creates an online SQLite backup', async () => { + const container = await mkdtemp(join(tmpdir(), 'maka-session-metadata-maintenance-')); + const workspaceRoot = join(container, 'workspace'); + const exportRoot = join(container, 'legacy-export'); + const backupPath = join(container, 'backups', 'sessions.sqlite'); + const store = createSessionStore(workspaceRoot); + try { + const created = await store.create(makeInput()); + await store.appendMessage(created.id, { + type: 'user', + id: 'user-1', + turnId: 'turn-1', + ts: 10, + text: 'portable transcript', + }); + await store.rename(created.id, 'Canonical SQLite title'); + + const exported = await exportLegacySessionTree({ + workspaceRoot, + destinationRoot: exportRoot, + now: () => 123, + }); + assert.equal(exported.sessionsExported, 1); + + const legacy = createLegacyFileSessionStore(exportRoot); + assert.equal((await legacy.readHeader(created.id)).name, 'Canonical SQLite title'); + assert.equal((await legacy.readMessages(created.id))[0]?.type, 'user'); + + const manifest = JSON.parse(await readFile(exported.manifestPath, 'utf8')) as { + format: string; + exportedAt: number; + sessions: Array<{ header: { id: string; name: string } }>; + }; + assert.equal(manifest.format, SESSION_METADATA_EXPORT_FORMAT); + assert.equal(manifest.exportedAt, 123); + assert.deepEqual( + manifest.sessions.map((record) => record.header.id), + [created.id], + ); + assert.equal(manifest.sessions[0]?.header.name, 'Canonical SQLite title'); + + const backup = await backupSessionMetadataDatabase({ + workspaceRoot, + destinationPath: backupPath, + }); + assert.equal(backup.destinationPath, backupPath); + assert.ok(backup.pagesCopied > 0); + const restored = createSqliteSessionMetadataStore(backupPath); + try { + assert.equal((await restored.read(created.id)).header.name, 'Canonical SQLite title'); + } finally { + restored.close(); + } + + const liveTranscript = await readFile( + join(workspaceRoot, 'sessions', created.id, 'session.jsonl'), + 'utf8', + ); + assert.equal(JSON.parse(liveTranscript.split('\n')[0]!).type, 'session_transcript'); + await assert.rejects( + () => exportLegacySessionTree({ workspaceRoot, destinationRoot: exportRoot }), + /already exists/, + ); + await assert.rejects( + () => backupSessionMetadataDatabase({ workspaceRoot, destinationPath: backupPath }), + /already exists/, + ); + } finally { + store.close?.(); + await rm(container, { recursive: true, force: true }); + } + }); + + test('refuses to export over the live workspace', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-session-metadata-export-overlap-')); + const store = createSessionStore(root); + try { + await store.create(makeInput()); + await assert.rejects( + () => exportLegacySessionTree({ workspaceRoot: root, destinationRoot: root }), + /overlaps/, + ); + await assert.rejects( + () => + backupSessionMetadataDatabase({ + workspaceRoot: root, + destinationPath: join(root, SQLITE_SESSION_METADATA_DATABASE_NAME), + }), + /differ from the source/, + ); + } finally { + store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); +}); + +function makeInput(): CreateSessionInput { + return { + cwd: '/tmp/cwd', + backend: 'fake', + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask', + name: 'Initial title', + }; +} diff --git a/packages/storage/src/__tests__/settings-store-usage.test.ts b/packages/storage/src/__tests__/settings-store-usage.test.ts index e29a046e88b..e7cd3b82512 100644 --- a/packages/storage/src/__tests__/settings-store-usage.test.ts +++ b/packages/storage/src/__tests__/settings-store-usage.test.ts @@ -4,6 +4,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { describe, it } from 'node:test'; import type { SessionHeader, StoredMessage } from '@maka/core/session'; +import { createSessionStore } from '../session-store.js'; import { createSettingsStore } from '../settings-store.js'; function makeHeader(overrides: Partial = {}): SessionHeader { @@ -44,6 +45,47 @@ async function seedSession( } describe('SettingsStore.usageStats request logs', () => { + it('uses canonical SQLite metadata for transcript-marker sessions', async () => { + const workspaceRoot = await mkdtemp(join(tmpdir(), 'maka-settings-usage-sqlite-')); + const sessions = createSessionStore(workspaceRoot); + try { + const header = await sessions.create({ + cwd: '/tmp/maka-workspace', + backend: 'ai-sdk', + llmConnectionSlug: 'sqlite-provider', + model: 'sqlite-default-model', + permissionMode: 'ask', + }); + await sessions.appendMessages(header.id, [ + { + type: 'assistant', + id: 'assistant-1', + turnId: 'turn-1', + ts: 10, + text: 'tracked', + modelId: 'sqlite-runtime-model', + }, + { + type: 'token_usage', + id: 'usage-1', + turnId: 'turn-1', + ts: 20, + input: 8, + output: 2, + }, + ]); + + const stats = await createSettingsStore(workspaceRoot).usageStats('all'); + + assert.equal(stats.summary.totalTokens, 10); + assert.equal(stats.logs[0]?.provider, 'sqlite-provider'); + assert.equal(stats.logs[0]?.model, 'sqlite-runtime-model'); + } finally { + sessions.close?.(); + await rm(workspaceRoot, { recursive: true, force: true }); + } + }); + it('includes tool invocation rows without inflating model usage totals', async () => { const workspaceRoot = await mkdtemp(join(tmpdir(), 'maka-settings-usage-')); try { diff --git a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts index 2c5b4cb5bed..e6bdddbc06f 100644 --- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts @@ -265,6 +265,8 @@ describe('SqliteSessionMetadataStore', () => { await store.create(fullHeader()); assert.equal(await store.remove('session-1'), true); assert.equal(await store.remove('session-1'), false); + assert.equal(await store.has('session-1'), false); + assert.equal(await store.isTombstoned('session-1'), true); assert.deepEqual(await store.list({ labelSlug: 'alpha' }), []); assert.deepEqual( await store.importEntries([ diff --git a/packages/storage/src/__tests__/sqlite-session-store.test.ts b/packages/storage/src/__tests__/sqlite-session-store.test.ts index 5f673e7e1c1..8891e0c8928 100644 --- a/packages/storage/src/__tests__/sqlite-session-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-store.test.ts @@ -35,14 +35,28 @@ describe('default SQLite session metadata store', () => { assert.equal((await store.readMessages(created.id))[0]?.type, 'user'); const transcriptPath = join(root, 'sessions', created.id, 'session.jsonl'); - const [legacyHeader, message] = (await readFile(transcriptPath, 'utf8')) + const [marker, message] = (await readFile(transcriptPath, 'utf8')) .trim() .split('\n') .map((line) => JSON.parse(line) as Record); - assert.equal(legacyHeader?.name, 'Initial title'); + assert.deepEqual(marker, { + type: 'session_transcript', + sessionId: created.id, + schemaVersion: 1, + }); + assert.equal('name' in (marker ?? {}), false); assert.equal(message?.type, 'user'); await stat(join(root, SQLITE_SESSION_METADATA_DATABASE_NAME)); await assert.rejects(() => stat(join(root, 'runtime.sqlite')), { code: 'ENOENT' }); + + store.close?.(); + const reopened = createSessionStore(root); + try { + assert.equal((await reopened.readHeader(created.id)).name, 'SQLite title'); + assert.equal((await reopened.readMessages(created.id)).length, 1); + } finally { + reopened.close?.(); + } } finally { store.close?.(); await rm(root, { recursive: true, force: true }); @@ -149,6 +163,30 @@ describe('default SQLite session metadata store', () => { await rm(root, { recursive: true, force: true }); } }); + + test('fails closed when a transcript marker has no canonical SQLite metadata', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-default-session-orphan-marker-')); + const sessionId = 'orphan-session'; + const sessionDir = join(root, 'sessions', sessionId); + await mkdir(sessionDir, { recursive: true }); + await writeFile( + join(sessionDir, 'session.jsonl'), + `${JSON.stringify({ + type: 'session_transcript', + sessionId, + schemaVersion: 1, + })}\n`, + 'utf8', + ); + + const store = createSessionStore(root); + try { + await assert.rejects(() => store.list(), /has no SQLite metadata/); + } finally { + store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); }); function makeInput(overrides: Partial = {}): CreateSessionInput { diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index d8514aca4b0..98b54348a0c 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -1,6 +1,8 @@ export * from './session-store.js'; +export * from './session-transcript.js'; export * from './sqlite-session-metadata-store.js'; export * from './session-metadata-transfer.js'; +export * from './session-metadata-maintenance.js'; export * from './agent-run-store.js'; export * from './shell-run-store.js'; export * from './connection-store.js'; diff --git a/packages/storage/src/session-metadata-maintenance.ts b/packages/storage/src/session-metadata-maintenance.ts new file mode 100644 index 00000000000..0c6b56f3e0f --- /dev/null +++ b/packages/storage/src/session-metadata-maintenance.ts @@ -0,0 +1,153 @@ +import { randomUUID } from 'node:crypto'; +import { mkdir, readFile, rename, rm, stat, writeFile } from 'node:fs/promises'; +import { dirname, join, relative, resolve } from 'node:path'; +import type { SessionHeader } from '@maka/core'; +import { decodeSessionHeader, SQLITE_SESSION_METADATA_DATABASE_NAME } from './session-store.js'; +import { + createSqliteSessionMetadataStore, + type SessionMetadataRecord, +} from './sqlite-session-metadata-store.js'; +import { decodeSessionTranscriptMarker, isSessionTranscriptMarker } from './session-transcript.js'; + +export const SESSION_METADATA_EXPORT_FORMAT = 'maka-session-metadata-export'; +export const SESSION_METADATA_EXPORT_SCHEMA_VERSION = 1; +export const SESSION_METADATA_EXPORT_MANIFEST_NAME = 'session-metadata.json'; + +export interface SessionMetadataExportManifest { + format: typeof SESSION_METADATA_EXPORT_FORMAT; + schemaVersion: typeof SESSION_METADATA_EXPORT_SCHEMA_VERSION; + exportedAt: number; + sessions: SessionMetadataRecord[]; +} + +export interface LegacySessionTreeExportReport { + destinationRoot: string; + sessionsExported: number; + manifestPath: string; +} + +/** + * Export a self-contained legacy-compatible session tree without changing the + * canonical SQLite database or the live transcript files. + */ +export async function exportLegacySessionTree(input: { + workspaceRoot: string; + destinationRoot: string; + now?: () => number; +}): Promise { + const workspaceRoot = resolve(input.workspaceRoot); + const destinationRoot = resolve(input.destinationRoot); + const sourceSessionsRoot = join(workspaceRoot, 'sessions'); + const destinationSessionsRoot = join(destinationRoot, 'sessions'); + if ( + destinationRoot === workspaceRoot || + isInsideOrSamePath(sourceSessionsRoot, destinationRoot) || + destinationSessionsRoot === sourceSessionsRoot + ) { + throw new Error('Session metadata export destination overlaps the live session tree'); + } + await assertPathMissing(destinationRoot, 'Session metadata export destination'); + + const databasePath = join(workspaceRoot, SQLITE_SESSION_METADATA_DATABASE_NAME); + await assertFileExists(databasePath, 'SQLite session metadata database'); + const metadata = createSqliteSessionMetadataStore(databasePath); + const stagingRoot = `${destinationRoot}.${process.pid}.${randomUUID()}.tmp`; + try { + const records = (await metadata.list()).sort((a, b) => a.header.id.localeCompare(b.header.id)); + await mkdir(join(stagingRoot, 'sessions'), { recursive: true }); + for (const record of records) { + const sourcePath = join(sourceSessionsRoot, record.header.id, 'session.jsonl'); + const transcript = await readFile(sourcePath, 'utf8'); + const body = legacyCompatibleTranscriptBody(transcript, record.header.id, record.header); + const destinationPath = join(stagingRoot, 'sessions', record.header.id, 'session.jsonl'); + await mkdir(dirname(destinationPath), { recursive: true }); + await writeFile(destinationPath, body, 'utf8'); + } + const manifest: SessionMetadataExportManifest = { + format: SESSION_METADATA_EXPORT_FORMAT, + schemaVersion: SESSION_METADATA_EXPORT_SCHEMA_VERSION, + exportedAt: (input.now ?? Date.now)(), + sessions: records, + }; + await writeFile( + join(stagingRoot, SESSION_METADATA_EXPORT_MANIFEST_NAME), + `${JSON.stringify(manifest, null, 2)}\n`, + 'utf8', + ); + await mkdir(dirname(destinationRoot), { recursive: true }); + await rename(stagingRoot, destinationRoot); + return { + destinationRoot, + sessionsExported: records.length, + manifestPath: join(destinationRoot, SESSION_METADATA_EXPORT_MANIFEST_NAME), + }; + } finally { + metadata.close(); + await rm(stagingRoot, { recursive: true, force: true }).catch(() => {}); + } +} + +export async function backupSessionMetadataDatabase(input: { + workspaceRoot: string; + destinationPath: string; +}): Promise<{ destinationPath: string; pagesCopied: number }> { + const databasePath = join(resolve(input.workspaceRoot), SQLITE_SESSION_METADATA_DATABASE_NAME); + const destinationPath = resolve(input.destinationPath); + await assertFileExists(databasePath, 'SQLite session metadata database'); + const metadata = createSqliteSessionMetadataStore(databasePath); + try { + return { + destinationPath, + pagesCopied: await metadata.backup(destinationPath), + }; + } finally { + metadata.close(); + } +} + +function legacyCompatibleTranscriptBody( + transcript: string, + sessionId: string, + header: SessionHeader, +): string { + const firstNewline = transcript.indexOf('\n'); + if (firstNewline < 0) { + throw new Error(`Session ${sessionId}: cannot find first JSONL record`); + } + const firstLine = transcript.slice(0, firstNewline); + let firstRecord: unknown; + try { + firstRecord = JSON.parse(firstLine) as unknown; + if (isSessionTranscriptMarker(firstRecord)) { + decodeSessionTranscriptMarker(firstRecord, sessionId); + } else { + decodeSessionHeader(firstRecord, sessionId); + } + } catch (error) { + throw new Error(`Session ${sessionId}: invalid first JSONL record`, { cause: error }); + } + return `${JSON.stringify(header)}\n${transcript.slice(firstNewline + 1)}`; +} + +async function assertPathMissing(path: string, label: string): Promise { + try { + await stat(path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return; + throw error; + } + throw new Error(`${label} already exists: ${path}`); +} + +async function assertFileExists(path: string, label: string): Promise { + const info = await stat(path).catch((error: NodeJS.ErrnoException) => { + if (error.code === 'ENOENT') throw new Error(`${label} does not exist: ${path}`); + throw error; + }); + if (!info.isFile()) throw new Error(`${label} is not a file: ${path}`); +} + +function isInsideOrSamePath(root: string, candidate: string): boolean { + const path = relative(root, candidate); + return path === '' || (!path.startsWith('..') && !path.includes(':')); +} diff --git a/packages/storage/src/session-metadata-transfer.ts b/packages/storage/src/session-metadata-transfer.ts index 5027795a490..677cff914f9 100644 --- a/packages/storage/src/session-metadata-transfer.ts +++ b/packages/storage/src/session-metadata-transfer.ts @@ -2,6 +2,7 @@ import { createHash } from 'node:crypto'; import { open, readdir } from 'node:fs/promises'; import { join } from 'node:path'; import { decodeSessionHeader, isSafeSessionId } from './session-store.js'; +import { decodeSessionTranscriptMarker, isSessionTranscriptMarker } from './session-transcript.js'; import type { SessionMetadataImportEntry, SqliteSessionMetadataStore, @@ -31,11 +32,17 @@ export async function importLegacySessionMetadataTree(input: { }): Promise { const sessionsRoot = join(input.workspaceRoot, 'sessions'); const entries: SessionMetadataImportEntry[] = []; + const transcriptMarkerSessionIds: string[] = []; const directories = await sessionDirectoryNames(sessionsRoot); for (const directory of directories) { const sourcePath = join(sessionsRoot, directory, 'session.jsonl'); try { - entries.push(await readLegacySessionMetadataEntry(sourcePath, directory)); + const entry = await readLegacySessionMetadataEntry(sourcePath, directory); + if (entry) { + entries.push(entry); + } else { + transcriptMarkerSessionIds.push(directory); + } } catch (error) { if (!isNotFound(error)) throw error; const canonicalStateExists = @@ -45,6 +52,14 @@ export async function importLegacySessionMetadataTree(input: { throw error; } } + for (const sessionId of transcriptMarkerSessionIds) { + if ( + !(await input.destination.has(sessionId)) && + !(await input.destination.isTombstoned(sessionId)) + ) { + throw new Error(`Session transcript marker has no SQLite metadata: ${sessionId}`); + } + } const result = await input.destination.importEntries(entries); const headersImported = result.created.filter(Boolean).length; return { @@ -60,12 +75,16 @@ export async function importLegacySessionMetadataTree(input: { export async function readLegacySessionMetadataEntry( sourcePath: string, sessionId: string, -): Promise { +): Promise { let value: unknown; let headerLine: string; try { headerLine = await readFirstJsonlRecord(sourcePath); value = JSON.parse(headerLine) as unknown; + if (isSessionTranscriptMarker(value)) { + decodeSessionTranscriptMarker(value, sessionId); + return null; + } return { header: decodeSessionHeader(value, sessionId), source: { diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index cdcb2d1d230..186fcd0c579 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -8,14 +8,16 @@ import { } from './execution-record-codec.js'; import { appendJsonl } from './jsonl-append.js'; import { classifyJsonRecord } from './json-prefix.js'; -import { - importLegacySessionMetadataTree, - readLegacySessionMetadataEntry, -} from './session-metadata-transfer.js'; +import { importLegacySessionMetadataTree } from './session-metadata-transfer.js'; import { createSqliteSessionMetadataStore, type SqliteSessionMetadataStore, } from './sqlite-session-metadata-store.js'; +import { + createSessionTranscriptMarker, + decodeSessionTranscriptMarker, + isSessionTranscriptMarker, +} from './session-transcript.js'; import { chainWrite } from './write-queue.js'; import { DEFAULT_SESSION_NAME, @@ -82,7 +84,7 @@ class SqliteSessionStore implements SessionStore { private readonly metadata: SqliteSessionMetadataStore; private readonly ready: Promise; - constructor(private readonly workspaceRoot: string) { + constructor(workspaceRoot: string) { this.files = new FileSessionStore(workspaceRoot); this.metadata = createSqliteSessionMetadataStore( join(workspaceRoot, SQLITE_SESSION_METADATA_DATABASE_NAME), @@ -95,12 +97,9 @@ class SqliteSessionStore implements SessionStore { async create(input: CreateSessionInput): Promise { await this.ensureReady(); - const staged = await this.files.create(input); + const staged = await this.files.createTranscript(input); try { - const sourcePath = this.sessionPath(staged.id); - const entry = await readLegacySessionMetadataEntry(sourcePath, staged.id); - await this.metadata.importEntries([entry]); - return (await this.metadata.read(staged.id)).header; + return (await this.metadata.create(staged)).header; } catch (error) { await this.files.remove(staged.id).catch(() => {}); throw error; @@ -138,9 +137,9 @@ class SqliteSessionStore implements SessionStore { const { header, previewMessages } = withPreviews[index]!; let messages = previewMessages.slice(-10); if (index < 3) { - messages = (await this.files.readMessagesSnapshot(header.id).catch(() => messages)).slice( - -10, - ); + messages = ( + await this.files.readTranscriptMessagesSnapshot(header.id, header).catch(() => messages) + ).slice(-10); } summaries.push(toSummary(header, messages)); } @@ -153,7 +152,7 @@ class SqliteSessionStore implements SessionStore { .map((record) => record.header) .sort((a, b) => a.id.localeCompare(b.id)); for (const header of headers) { - await this.files.readMessagesForRecovery(header.id); + await this.files.readTranscriptMessagesForRecovery(header.id, header); } return headers; } @@ -165,12 +164,14 @@ class SqliteSessionStore implements SessionStore { async readMessagesSnapshot(sessionId: string): Promise { await this.ensureReady(); - return this.files.readMessagesSnapshot(sessionId); + const header = (await this.metadata.read(sessionId)).header; + return this.files.readTranscriptMessagesSnapshot(sessionId, header); } async readMessagesForRecovery(sessionId: string): Promise { await this.ensureReady(); - return this.files.readMessagesForRecovery(sessionId); + const header = (await this.metadata.read(sessionId)).header; + return this.files.readTranscriptMessagesForRecovery(sessionId, header); } async listTurnsSnapshot(sessionId: string): Promise { @@ -286,7 +287,8 @@ class SqliteSessionStore implements SessionStore { knownMessages?: StoredMessage[], ): Promise { if (header.connectionLocked) return header; - const messages = knownMessages ?? (await this.files.readMessagesSnapshot(header.id)); + const messages = + knownMessages ?? (await this.files.readTranscriptMessagesSnapshot(header.id, header)); if (!messages.some((message) => message.type === 'user')) return header; return this.updateHeader(header.id, { connectionLocked: true }); } @@ -294,11 +296,6 @@ class SqliteSessionStore implements SessionStore { private async ensureReady(): Promise { await this.ready; } - - private sessionPath(sessionId: string): string { - assertSafeSessionId(sessionId); - return join(this.workspaceRoot, 'sessions', sessionId, 'session.jsonl'); - } } class FileSessionStore implements SessionStore { @@ -313,6 +310,17 @@ class FileSessionStore implements SessionStore { } async create(input: CreateSessionInput): Promise { + return this.createWithInitialRecord(input, 'legacy-header'); + } + + async createTranscript(input: CreateSessionInput): Promise { + return this.createWithInitialRecord(input, 'transcript-marker'); + } + + private async createWithInitialRecord( + input: CreateSessionInput, + initialRecord: 'legacy-header' | 'transcript-marker', + ): Promise { const now = Date.now(); const id = randomUUID(); // PR-UI-IPC-2 (@kenji msg 0474c3fe + @xuan msg 88d96a87): @@ -376,7 +384,9 @@ class FileSessionStore implements SessionStore { await this.withQueue(id, async () => { await mkdir(this.sessionDir(id), { recursive: true }); - await writeFile(this.sessionPath(id), JSON.stringify(header) + '\n', 'utf8'); + const firstRecord = + initialRecord === 'legacy-header' ? header : createSessionTranscriptMarker(header.id); + await writeFile(this.sessionPath(id), JSON.stringify(firstRecord) + '\n', 'utf8'); }); return header; @@ -502,6 +512,20 @@ class FileSessionStore implements SessionStore { return this.readTailPreviewMessages(sessionId); } + async readTranscriptMessagesSnapshot( + sessionId: string, + header: SessionHeader, + ): Promise { + return this.readTranscriptMessagesUnlocked(sessionId, header); + } + + async readTranscriptMessagesForRecovery( + sessionId: string, + header: SessionHeader, + ): Promise { + return this.readTranscriptMessagesUnlocked(sessionId, header, true); + } + async readMessagesForRecovery(sessionId: string): Promise { return (await this.readFilePartsUnlocked(sessionId, true)).messages; } @@ -759,6 +783,72 @@ class FileSessionStore implements SessionStore { return { header, messages }; } + private async readTranscriptMessagesUnlocked( + sessionId: string, + header: SessionHeader, + strict = false, + ): Promise { + const text = await readFile(this.sessionPath(sessionId), 'utf8'); + const rawLines = text.split('\n'); + const endsWithNewline = text.endsWith('\n'); + const lines = rawLines + .map((line, index) => ({ line, lineNumber: index + 1 })) + .filter((entry) => entry.line.trim().length > 0); + if (lines.length === 0 || !lines[0]) throw new Error(`Session ${sessionId} is empty`); + + let firstRecord: unknown; + try { + firstRecord = JSON.parse(lines[0].line) as unknown; + if (isSessionTranscriptMarker(firstRecord)) { + decodeSessionTranscriptMarker(firstRecord, sessionId); + } else { + decodeSessionHeader(firstRecord, sessionId); + } + } catch (error) { + const detail = error instanceof Error ? error.message : String(error); + throw new Error(`Session ${sessionId} has an invalid first JSONL record: ${detail}`); + } + + const messages: StoredMessage[] = []; + const lastLineNumber = lines.at(-1)?.lineNumber; + for (const entry of lines.slice(1)) { + let parsed: unknown; + try { + parsed = JSON.parse(entry.line); + } catch (error) { + if ( + !endsWithNewline && + entry.lineNumber === lastLineNumber && + classifyJsonRecord(entry.line) === 'incomplete-prefix' + ) { + continue; + } + if (strict) { + const detail = error instanceof Error ? error.message : String(error); + throw new Error( + `Session ${sessionId} has a corrupt JSONL record at line ${entry.lineNumber}: ${detail}`, + ); + } + messages.push(createJsonlCorruptionNote(header, entry.lineNumber, error)); + continue; + } + try { + messages.push( + strict ? decodeStoredMessageForRecovery(parsed) : decodeStoredMessageForRead(parsed), + ); + } catch (error) { + if (strict) { + const detail = error instanceof Error ? error.message : String(error); + throw new Error( + `Session ${sessionId} has a corrupt JSONL record at line ${entry.lineNumber}: ${detail}`, + ); + } + messages.push(createJsonlCorruptionNote(header, entry.lineNumber, error)); + } + } + return messages; + } + private async writeAtomic(path: string, content: string): Promise { await mkdir(dirname(path), { recursive: true }); const tempPath = `${path}.${process.pid}.${Date.now()}.${randomUUID()}.tmp`; diff --git a/packages/storage/src/session-transcript.ts b/packages/storage/src/session-transcript.ts new file mode 100644 index 00000000000..32b81419897 --- /dev/null +++ b/packages/storage/src/session-transcript.ts @@ -0,0 +1,38 @@ +export const SESSION_TRANSCRIPT_RECORD_TYPE = 'session_transcript'; +export const SESSION_TRANSCRIPT_SCHEMA_VERSION = 1; + +export interface SessionTranscriptMarker { + type: typeof SESSION_TRANSCRIPT_RECORD_TYPE; + sessionId: string; + schemaVersion: typeof SESSION_TRANSCRIPT_SCHEMA_VERSION; +} + +export function createSessionTranscriptMarker(sessionId: string): SessionTranscriptMarker { + return { + type: SESSION_TRANSCRIPT_RECORD_TYPE, + sessionId, + schemaVersion: SESSION_TRANSCRIPT_SCHEMA_VERSION, + }; +} + +export function isSessionTranscriptMarker(value: unknown): value is SessionTranscriptMarker { + if (!value || typeof value !== 'object' || Array.isArray(value)) return false; + const record = value as Record; + return record.type === SESSION_TRANSCRIPT_RECORD_TYPE; +} + +export function decodeSessionTranscriptMarker( + value: unknown, + expectedSessionId: string, +): SessionTranscriptMarker { + if (!isSessionTranscriptMarker(value)) { + throw new Error(`Session ${expectedSessionId}: missing transcript marker`); + } + if ( + value.sessionId !== expectedSessionId || + value.schemaVersion !== SESSION_TRANSCRIPT_SCHEMA_VERSION + ) { + throw new Error(`Session ${expectedSessionId}: invalid transcript marker`); + } + return value; +} diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index be4ed5389be..fb369914d5e 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -1,6 +1,6 @@ import { createRequire } from 'node:module'; -import { dirname } from 'node:path'; -import { mkdirSync } from 'node:fs'; +import { dirname, resolve } from 'node:path'; +import { existsSync, mkdirSync } from 'node:fs'; import { isDeepStrictEqual } from 'node:util'; import type { DatabaseSync } from 'node:sqlite'; import type { SessionHeader, SessionListFilter } from '@maka/core'; @@ -15,8 +15,23 @@ export { SQLITE_SESSION_METADATA_SCHEMA_VERSION } from './sqlite-session-metadat const require = createRequire(import.meta.url); -function loadDatabaseSync(): typeof import('node:sqlite').DatabaseSync { - return (require('node:sqlite') as typeof import('node:sqlite')).DatabaseSync; +function loadSqliteModule(): typeof import('node:sqlite') { + const emitWarning = process.emitWarning; + process.emitWarning = ((warning: string | Error, ...args: unknown[]) => { + const warningType = typeof args[0] === 'string' ? args[0] : undefined; + if ( + warningType === 'ExperimentalWarning' && + String(warning).startsWith('SQLite is an experimental feature') + ) { + return; + } + Reflect.apply(emitWarning, process, [warning, ...args]); + }) as typeof process.emitWarning; + try { + return require('node:sqlite') as typeof import('node:sqlite'); + } finally { + process.emitWarning = emitWarning; + } } export type SqliteSessionMetadataStoreFailpoint = @@ -66,11 +81,11 @@ export class SqliteSessionMetadataStore { private closed = false; constructor( - path: string, + private readonly path: string, private readonly options: SqliteSessionMetadataStoreOptions = {}, ) { if (path !== ':memory:') mkdirSync(dirname(path), { recursive: true }); - const DatabaseSync = loadDatabaseSync(); + const { DatabaseSync } = loadSqliteModule(); this.db = new DatabaseSync(path); configureSqliteSessionMetadataDatabase(this.db); migrateSqliteSessionMetadataDatabase(this.db); @@ -96,6 +111,19 @@ export class SqliteSessionMetadataStore { this.db.close(); } + async backup(destinationPath: string): Promise { + this.assertOpen(); + if (!destinationPath) throw new Error('Session metadata backup destination is required'); + if (this.path !== ':memory:' && resolve(destinationPath) === resolve(this.path)) { + throw new Error('Session metadata backup destination must differ from the source database'); + } + if (existsSync(destinationPath)) { + throw new Error(`Session metadata backup destination already exists: ${destinationPath}`); + } + mkdirSync(dirname(destinationPath), { recursive: true }); + return loadSqliteModule().backup(this.db, destinationPath); + } + async create(header: SessionHeader): Promise { this.assertOpen(); const normalized = normalizeSessionHeader(header); diff --git a/packages/storage/src/usage-stats-store.ts b/packages/storage/src/usage-stats-store.ts index 584c793955a..4284352989f 100644 --- a/packages/storage/src/usage-stats-store.ts +++ b/packages/storage/src/usage-stats-store.ts @@ -1,7 +1,9 @@ -import { readFile } from 'node:fs/promises'; +import { readFile, readdir, stat } from 'node:fs/promises'; import { join } from 'node:path'; import type { UsageRange, UsageStats } from '@maka/core'; import type { SessionHeader } from '@maka/core/session'; +import { SQLITE_SESSION_METADATA_DATABASE_NAME } from './session-store.js'; +import { createSqliteSessionMetadataStore } from './sqlite-session-metadata-store.js'; type UsageSessionHeader = Pick; @@ -54,7 +56,7 @@ export async function readUsageStats( range: UsageRange, ): Promise { const since = rangeToSince(range); - const sessions = await readStoredSessions(join(workspaceRoot, 'sessions')); + const sessions = await readStoredSessions(workspaceRoot); const modelLogs = sessions.flatMap(({ header, messages }) => { const assistantByTurn = new Map( messages @@ -116,11 +118,12 @@ export async function readUsageStats( } async function readStoredSessions( - sessionsRoot: string, + workspaceRoot: string, ): Promise> { - const fs = await import('node:fs/promises'); + const sessionsRoot = join(workspaceRoot, 'sessions'); try { - const entries = await fs.readdir(sessionsRoot, { withFileTypes: true }); + const canonicalHeaders = await readCanonicalUsageHeaders(workspaceRoot); + const entries = await readdir(sessionsRoot, { withFileTypes: true }); const sessions: Array<{ header: UsageSessionHeader; messages: UsageMessage[] }> = []; for (const entry of entries) { if (!entry.isDirectory()) continue; @@ -128,7 +131,10 @@ async function readStoredSessions( const text = await readFile(join(sessionsRoot, entry.name, 'session.jsonl'), 'utf8'); const lines = text.split('\n').filter((line) => line.trim()); if (!lines[0]) continue; - const header = normalizeUsageSessionHeader(JSON.parse(lines[0]), entry.name); + const header = + canonicalHeaders === null + ? normalizeUsageSessionHeader(JSON.parse(lines[0]), entry.name) + : canonicalHeaders.get(entry.name); if (!header) continue; const messages: UsageMessage[] = []; for (const line of lines.slice(1)) { @@ -153,6 +159,33 @@ async function readStoredSessions( } } +async function readCanonicalUsageHeaders( + workspaceRoot: string, +): Promise | null> { + const path = join(workspaceRoot, SQLITE_SESSION_METADATA_DATABASE_NAME); + try { + await stat(path); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === 'ENOENT') return null; + throw error; + } + const metadata = createSqliteSessionMetadataStore(path); + try { + return new Map( + (await metadata.list()).map(({ header }) => [ + header.id, + { + id: header.id, + llmConnectionSlug: header.llmConnectionSlug, + model: header.model, + }, + ]), + ); + } finally { + metadata.close(); + } +} + function normalizeUsageSessionHeader(value: unknown, sessionId: string): UsageSessionHeader | null { if (!isRecord(value)) return null; if (value.id !== sessionId) return null;