From e0221a09e166676cd995a0dc31e74c071ef917d5 Mon Sep 17 00:00:00 2001 From: likun666661 <90952590+likun666661@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:10:54 +0800 Subject: [PATCH 1/2] Make SQLite the default session metadata store --- apps/desktop/src/main/app-lifecycle.ts | 4 + apps/desktop/src/main/main.ts | 1 + packages/cli/src/runtime-bootstrap.ts | 1 + .../session-metadata-transfer.test.ts | 6 +- .../src/__tests__/session-store.test.ts | 2 +- .../sqlite-session-metadata-store.test.ts | 18 +- .../__tests__/sqlite-session-store.test.ts | 165 ++++++++++++ .../storage/src/session-metadata-transfer.ts | 37 ++- packages/storage/src/session-store.ts | 243 ++++++++++++++++++ .../src/sqlite-session-metadata-schema.ts | 11 +- .../src/sqlite-session-metadata-store.ts | 37 ++- 11 files changed, 500 insertions(+), 25 deletions(-) create mode 100644 packages/storage/src/__tests__/sqlite-session-store.test.ts diff --git a/apps/desktop/src/main/app-lifecycle.ts b/apps/desktop/src/main/app-lifecycle.ts index 6e9e81ecc0..6fc87cccd1 100644 --- a/apps/desktop/src/main/app-lifecycle.ts +++ b/apps/desktop/src/main/app-lifecycle.ts @@ -6,6 +6,7 @@ import type { BotRegistry, SessionManager, ShellRunProcessManager } from '@maka/ import type { McpClientManager } from '@maka/mcp'; import type { createConnectionStore, + createSessionStore, createSettingsStore, createTelemetryRepo, openRuntimeEventPersistence, @@ -35,6 +36,7 @@ export interface AppLifecycleDeps { isIsolatedE2e: boolean; e2eFixture: ReturnType; workspaceRoot: string; + sessionStore: ReturnType; credentialStore: ReturnType; connectionStore: ReturnType; settingsStore: ReturnType; @@ -84,6 +86,7 @@ export function wireAppLifecycle(deps: AppLifecycleDeps): void { isIsolatedE2e, e2eFixture, workspaceRoot, + sessionStore, credentialStore, connectionStore, settingsStore, @@ -331,5 +334,6 @@ export function wireAppLifecycle(deps: AppLifecycleDeps): void { if (result.status === 'rejected') console.error('[shutdown] cleanup failed:', result.reason); } runtimePersistence.close(); + sessionStore.close?.(); } } diff --git a/apps/desktop/src/main/main.ts b/apps/desktop/src/main/main.ts index 9918d583db..4b4d002e7c 100644 --- a/apps/desktop/src/main/main.ts +++ b/apps/desktop/src/main/main.ts @@ -1214,6 +1214,7 @@ wireAppLifecycle({ isIsolatedE2e, e2eFixture, workspaceRoot, + sessionStore: store, credentialStore, connectionStore, settingsStore, diff --git a/packages/cli/src/runtime-bootstrap.ts b/packages/cli/src/runtime-bootstrap.ts index e1127a6a9f..595e2defb4 100644 --- a/packages/cli/src/runtime-bootstrap.ts +++ b/packages/cli/src/runtime-bootstrap.ts @@ -889,6 +889,7 @@ export async function createMakaCliRuntimeContext( goalManager.dispose(); await shellRuns.terminateAll(); shellRunListeners.clear(); + store.close?.(); runtimePersistence.close(); }, }; diff --git a/packages/storage/src/__tests__/session-metadata-transfer.test.ts b/packages/storage/src/__tests__/session-metadata-transfer.test.ts index ba2bef2895..5e3243e291 100644 --- a/packages/storage/src/__tests__/session-metadata-transfer.test.ts +++ b/packages/storage/src/__tests__/session-metadata-transfer.test.ts @@ -5,7 +5,7 @@ import { join } from 'node:path'; import { describe, test } from 'node:test'; import type { CreateSessionInput } from '@maka/core'; import { importLegacySessionMetadataTree } from '../session-metadata-transfer.js'; -import { createSessionStore } from '../session-store.js'; +import { createLegacyFileSessionStore as createSessionStore } from '../session-store.js'; import { createSqliteSessionMetadataStore } from '../sqlite-session-metadata-store.js'; describe('legacy session metadata transfer', () => { @@ -39,6 +39,7 @@ describe('legacy session metadata transfer', () => { headersImported: 2, headersExisting: 0, sourcesAlreadyImported: 0, + sourcesTombstoned: 0, }); assert.deepEqual((await sqlite.list()).map((record) => record.header.name).sort(), [ 'First', @@ -72,6 +73,7 @@ describe('legacy session metadata transfer', () => { headersImported: 0, headersExisting: 0, sourcesAlreadyImported: 2, + sourcesTombstoned: 0, }); assert.equal((await sqlite.read(first.id)).header.name, 'SQLite is canonical now'); } finally { @@ -135,7 +137,7 @@ describe('legacy session metadata transfer', () => { await assert.rejects( () => importLegacySessionMetadataTree({ workspaceRoot: root, destination: sqlite }), - /Invalid session header/, + /Invalid legacy session header/, ); await assert.rejects(() => sqlite.read(valid.id), /not found/); assert.deepEqual(await sqlite.list(), []); diff --git a/packages/storage/src/__tests__/session-store.test.ts b/packages/storage/src/__tests__/session-store.test.ts index d9e850ef18..15c791bafc 100644 --- a/packages/storage/src/__tests__/session-store.test.ts +++ b/packages/storage/src/__tests__/session-store.test.ts @@ -4,7 +4,7 @@ import { join } from 'node:path'; import { tmpdir } from 'node:os'; import { describe, test } from 'node:test'; import type { CreateSessionInput, SessionHeader, StoredMessage } from '@maka/core'; -import { createSessionStore } from '../session-store.js'; +import { createLegacyFileSessionStore as createSessionStore } from '../session-store.js'; describe('FileSessionStore CRUD', () => { test('list on a missing workspace is observational and does not create session storage', async () => { 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 8eecf5cbac..2c5b4cb5be 100644 --- a/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-metadata-store.test.ts @@ -18,7 +18,7 @@ describe('SqliteSessionMetadataStore', () => { try { const store = createSqliteSessionMetadataStore(path, { now: () => 100 }); const header = fullHeader(); - assert.equal(store.schemaVersion(), 1); + assert.equal(store.schemaVersion(), 2); assert.equal(store.journalMode(), 'wal'); assert.deepEqual(await store.create(header), { header, @@ -29,7 +29,7 @@ describe('SqliteSessionMetadataStore', () => { const reopened = createSqliteSessionMetadataStore(path, { now: () => 200 }); try { - assert.equal(reopened.schemaVersion(), 1); + assert.equal(reopened.schemaVersion(), 2); assert.deepEqual(await reopened.read(header.id), { header, metadataVersion: 1, @@ -50,7 +50,7 @@ describe('SqliteSessionMetadataStore', () => { const metadata = createSqliteSessionMetadataStore(path); try { assert.equal(runtime.schemaVersion(), 4); - assert.equal(metadata.schemaVersion(), 1); + assert.equal(metadata.schemaVersion(), 2); await metadata.create(fullHeader()); await runtime.appendRuntimeEvent('session-1', 'run-1', { id: 'event-1', @@ -205,10 +205,12 @@ describe('SqliteSessionMetadataStore', () => { assert.deepEqual(await store.importEntries([entry]), { created: [true], sourcesAlreadyImported: 0, + sourcesTombstoned: 0, }); assert.deepEqual(await store.importEntries([entry]), { created: [], sourcesAlreadyImported: 1, + sourcesTombstoned: 0, }); await assert.rejects( () => @@ -264,6 +266,16 @@ describe('SqliteSessionMetadataStore', () => { assert.equal(await store.remove('session-1'), true); assert.equal(await store.remove('session-1'), false); assert.deepEqual(await store.list({ labelSlug: 'alpha' }), []); + assert.deepEqual( + await store.importEntries([ + { + header: fullHeader(), + source: { path: '/session-1.jsonl', fingerprint: '1:1' }, + }, + ]), + { created: [], sourcesAlreadyImported: 0, sourcesTombstoned: 1 }, + ); + await assert.rejects(() => store.create(fullHeader()), /tombstoned/); } finally { store.close(); } diff --git a/packages/storage/src/__tests__/sqlite-session-store.test.ts b/packages/storage/src/__tests__/sqlite-session-store.test.ts new file mode 100644 index 0000000000..5f673e7e1c --- /dev/null +++ b/packages/storage/src/__tests__/sqlite-session-store.test.ts @@ -0,0 +1,165 @@ +import assert from 'node:assert/strict'; +import { mkdir, mkdtemp, readFile, rm, stat, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { describe, test } from 'node:test'; +import type { CreateSessionInput, SessionHeader } from '@maka/core'; +import { + createLegacyFileSessionStore, + createSessionStore, + SQLITE_SESSION_METADATA_DATABASE_NAME, +} from '../session-store.js'; +import { createSqliteSessionMetadataStore } from '../sqlite-session-metadata-store.js'; + +describe('default SQLite session metadata store', () => { + test('uses SQLite as canonical metadata while keeping transcript bodies in JSONL', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-default-session-store-')); + const store = createSessionStore(root); + try { + const created = await store.create(makeInput({ name: 'Initial title' })); + await store.appendMessage(created.id, { + type: 'user', + id: 'user-1', + turnId: 'turn-1', + ts: 10, + text: 'hello', + }); + await store.rename(created.id, 'SQLite title'); + await store.updateHeader(created.id, { + hasUnread: true, + lastMessageAt: 10, + }); + + assert.equal((await store.readHeader(created.id)).name, 'SQLite title'); + assert.equal((await store.list())[0]?.name, 'SQLite title'); + 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')) + .trim() + .split('\n') + .map((line) => JSON.parse(line) as Record); + assert.equal(legacyHeader?.name, 'Initial title'); + assert.equal(message?.type, 'user'); + await stat(join(root, SQLITE_SESSION_METADATA_DATABASE_NAME)); + await assert.rejects(() => stat(join(root, 'runtime.sqlite')), { code: 'ENOENT' }); + } finally { + store.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('imports an existing JSONL catalog once and preserves later SQLite-only updates', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-default-session-migration-')); + const legacy = createLegacyFileSessionStore(root); + const created = await legacy.create(makeInput({ name: 'Legacy title', labels: ['legacy'] })); + await legacy.updateHeader(created.id, { + status: 'blocked', + blockedReason: 'permission_required', + hasUnread: true, + }); + await legacy.appendMessage(created.id, { + type: 'user', + id: 'user-1', + turnId: 'turn-1', + ts: 10, + text: 'legacy message', + }); + + const first = createSessionStore(root); + try { + assert.equal((await first.readHeader(created.id)).name, 'Legacy title'); + await first.rename(created.id, 'SQLite title'); + await first.appendMessage(created.id, { + type: 'assistant', + id: 'assistant-1', + turnId: 'turn-1', + ts: 11, + text: 'new message', + modelId: 'fake-model', + }); + } finally { + first.close?.(); + } + + const reopened = createSessionStore(root); + try { + assert.equal((await reopened.readHeader(created.id)).name, 'SQLite title'); + assert.equal((await reopened.readMessages(created.id)).length, 2); + assert.deepEqual( + (await reopened.list({ labelSlug: 'legacy' })).map((item) => item.id), + [created.id], + ); + } finally { + reopened.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('uses tombstones to prevent a deleted legacy transcript from resurrecting', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-default-session-delete-')); + const store = createSessionStore(root); + let header!: SessionHeader; + try { + header = await store.create(makeInput({ name: 'Delete me' })); + await store.remove(header.id); + } finally { + store.close?.(); + } + + const sessionDir = join(root, 'sessions', header.id); + await mkdir(sessionDir, { recursive: true }); + await writeFile(join(sessionDir, 'session.jsonl'), `${JSON.stringify(header)}\n`, 'utf8'); + + const reopened = createSessionStore(root); + try { + assert.deepEqual(await reopened.list(), []); + await assert.rejects(() => reopened.readHeader(header.id), /not found/); + } finally { + reopened.close?.(); + await rm(root, { recursive: true, force: true }); + } + }); + + test('fails the whole startup migration before exposing a partial catalog', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-default-session-invalid-')); + const legacy = createLegacyFileSessionStore(root); + const valid = await legacy.create(makeInput({ name: 'Valid' })); + const invalid = await legacy.create(makeInput({ name: 'Invalid' })); + const invalidPath = join(root, 'sessions', invalid.id, 'session.jsonl'); + const lines = (await readFile(invalidPath, 'utf8')).split('\n'); + lines[0] = JSON.stringify({ ...JSON.parse(lines[0]!), labels: 'invalid' }); + await writeFile(invalidPath, lines.join('\n'), 'utf8'); + + const store = createSessionStore(root); + try { + await assert.rejects(() => store.list(), /Invalid legacy session header/); + } finally { + store.close?.(); + } + + const metadata = createSqliteSessionMetadataStore( + join(root, SQLITE_SESSION_METADATA_DATABASE_NAME), + ); + try { + await assert.rejects(() => metadata.read(valid.id), /not found/); + assert.deepEqual(await metadata.list(), []); + } finally { + metadata.close(); + await rm(root, { recursive: true, force: true }); + } + }); +}); + +function makeInput(overrides: Partial = {}): CreateSessionInput { + return { + cwd: '/tmp/cwd', + backend: 'fake', + llmConnectionSlug: 'fake', + model: 'fake-model', + permissionMode: 'ask', + name: 'Session', + labels: [], + ...overrides, + }; +} diff --git a/packages/storage/src/session-metadata-transfer.ts b/packages/storage/src/session-metadata-transfer.ts index 4c8f543d57..a74c362f5e 100644 --- a/packages/storage/src/session-metadata-transfer.ts +++ b/packages/storage/src/session-metadata-transfer.ts @@ -16,6 +16,7 @@ export interface LegacySessionMetadataImportReport { headersImported: number; headersExisting: number; sourcesAlreadyImported: number; + sourcesTombstoned: number; } /** @@ -32,19 +33,7 @@ export async function importLegacySessionMetadataTree(input: { const entries: SessionMetadataImportEntry[] = []; for (const directory of await sessionDirectoryNames(sessionsRoot)) { const sourcePath = join(sessionsRoot, directory, 'session.jsonl'); - let value: unknown; - let headerLine: string; - try { - headerLine = await readFirstJsonlRecord(sourcePath); - value = JSON.parse(headerLine) as unknown; - } catch (error) { - throw new Error(`Invalid legacy session header at ${sourcePath}`, { cause: error }); - } - const fingerprint = createHash('sha256').update(headerLine).digest('hex'); - entries.push({ - header: decodeSessionHeader(value, directory), - source: { path: sourcePath, fingerprint }, - }); + entries.push(await readLegacySessionMetadataEntry(sourcePath, directory)); } const result = await input.destination.importEntries(entries); const headersImported = result.created.filter(Boolean).length; @@ -54,9 +43,31 @@ export async function importLegacySessionMetadataTree(input: { headersImported, headersExisting: result.created.length - headersImported, sourcesAlreadyImported: result.sourcesAlreadyImported, + sourcesTombstoned: result.sourcesTombstoned, }; } +export async function readLegacySessionMetadataEntry( + sourcePath: string, + sessionId: string, +): Promise { + let value: unknown; + let headerLine: string; + try { + headerLine = await readFirstJsonlRecord(sourcePath); + value = JSON.parse(headerLine) as unknown; + return { + header: decodeSessionHeader(value, sessionId), + source: { + path: sourcePath, + fingerprint: createHash('sha256').update(headerLine).digest('hex'), + }, + }; + } catch (error) { + throw new Error(`Invalid legacy session header at ${sourcePath}`, { cause: error }); + } +} + async function sessionDirectoryNames(root: string): Promise { let entries; try { diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 25639a6714..cdcb2d1d23 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -8,6 +8,14 @@ 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 { + createSqliteSessionMetadataStore, + type SqliteSessionMetadataStore, +} from './sqlite-session-metadata-store.js'; import { chainWrite } from './write-queue.js'; import { DEFAULT_SESSION_NAME, @@ -30,6 +38,7 @@ import type { } from '@maka/core'; const SESSION_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/; +export const SQLITE_SESSION_METADATA_DATABASE_NAME = 'sessions.sqlite'; export interface SessionStore { create(input: CreateSessionInput): Promise; @@ -56,12 +65,242 @@ export interface SessionStore { rename(sessionId: string, name: string): Promise; setGeneratedTitleIfAbsent(sessionId: string, title: string): Promise; remove(sessionId: string): Promise; + close?(): void; } export function createSessionStore(workspaceRoot: string): SessionStore { + return new SqliteSessionStore(workspaceRoot); +} + +/** Legacy JSONL-header store retained only for migration and compatibility tests. */ +export function createLegacyFileSessionStore(workspaceRoot: string): SessionStore { return new FileSessionStore(workspaceRoot); } +class SqliteSessionStore implements SessionStore { + private readonly files: FileSessionStore; + private readonly metadata: SqliteSessionMetadataStore; + private readonly ready: Promise; + + constructor(private readonly workspaceRoot: string) { + this.files = new FileSessionStore(workspaceRoot); + this.metadata = createSqliteSessionMetadataStore( + join(workspaceRoot, SQLITE_SESSION_METADATA_DATABASE_NAME), + ); + this.ready = importLegacySessionMetadataTree({ + workspaceRoot, + destination: this.metadata, + }).then(() => {}); + } + + async create(input: CreateSessionInput): Promise { + await this.ensureReady(); + const staged = await this.files.create(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; + } catch (error) { + await this.files.remove(staged.id).catch(() => {}); + throw error; + } + } + + async list(filter?: SessionListFilter): Promise { + await this.ensureReady(); + const records = await this.metadata.list(filter); + const withPreviews: Array<{ + header: SessionHeader; + previewMessages: StoredMessage[]; + }> = []; + for (const record of records) { + const previewMessages = await this.files + .readPreviewMessages(record.header.id) + .catch(() => []); + withPreviews.push({ header: record.header, previewMessages }); + } + withPreviews.sort((a, b) => { + const aLastMessageAt = maxTimestamp( + a.header.lastMessageAt, + latestVisibleMessageAt(a.previewMessages), + ); + const bLastMessageAt = maxTimestamp( + b.header.lastMessageAt, + latestVisibleMessageAt(b.previewMessages), + ); + const tsDelta = (bLastMessageAt ?? 0) - (aLastMessageAt ?? 0); + return tsDelta !== 0 ? tsDelta : a.header.id.localeCompare(b.header.id); + }); + + const summaries: SessionSummary[] = []; + for (let index = 0; index < withPreviews.length; index += 1) { + const { header, previewMessages } = withPreviews[index]!; + let messages = previewMessages.slice(-10); + if (index < 3) { + messages = (await this.files.readMessagesSnapshot(header.id).catch(() => messages)).slice( + -10, + ); + } + summaries.push(toSummary(header, messages)); + } + return summaries; + } + + async listForRecovery(): Promise { + await this.ensureReady(); + const headers = (await this.metadata.list()) + .map((record) => record.header) + .sort((a, b) => a.id.localeCompare(b.id)); + for (const header of headers) { + await this.files.readMessagesForRecovery(header.id); + } + return headers; + } + + async readHeaderSnapshot(sessionId: string): Promise { + await this.ensureReady(); + return (await this.metadata.read(sessionId)).header; + } + + async readMessagesSnapshot(sessionId: string): Promise { + await this.ensureReady(); + return this.files.readMessagesSnapshot(sessionId); + } + + async readMessagesForRecovery(sessionId: string): Promise { + await this.ensureReady(); + return this.files.readMessagesForRecovery(sessionId); + } + + async listTurnsSnapshot(sessionId: string): Promise { + return deriveTurnRecords(await this.readMessagesSnapshot(sessionId)); + } + + async readHeader(sessionId: string): Promise { + const header = await this.readHeaderSnapshot(sessionId); + return this.lockConnectionAfterFirstUserMessage(header); + } + + async readMessages(sessionId: string): Promise { + const messages = await this.readMessagesSnapshot(sessionId); + const header = (await this.metadata.read(sessionId)).header; + await this.lockConnectionAfterFirstUserMessage(header, messages); + return messages; + } + + async listTurns(sessionId: string): Promise { + return deriveTurnRecords(await this.readMessages(sessionId)); + } + + async appendMessage(sessionId: string, message: StoredMessage): Promise { + await this.appendMessages(sessionId, [message]); + } + + async appendMessages(sessionId: string, messages: StoredMessage[]): Promise { + await this.ensureReady(); + await this.files.appendMessages(sessionId, messages); + } + + async updateHeader(sessionId: string, patch: Partial): Promise { + await this.ensureReady(); + return (await this.metadata.update(sessionId, patch)).header; + } + + async markSessionReadThrough(sessionId: string, readThroughTs: number): Promise { + const header = await this.readHeaderSnapshot(sessionId); + const messages = await this.readMessagesSnapshot(sessionId); + const effectiveLastMessageAt = maxTimestamp( + header.lastMessageAt, + latestVisibleMessageAt(messages), + ); + if ( + !Number.isFinite(readThroughTs) || + !header.hasUnread || + (effectiveLastMessageAt !== undefined && effectiveLastMessageAt > readThroughTs) + ) { + return header; + } + return this.updateHeader(sessionId, { hasUnread: false }); + } + + async archive(sessionId: string): Promise { + const now = Date.now(); + await this.updateHeader(sessionId, { + isArchived: true, + archivedAt: now, + status: 'archived', + statusUpdatedAt: now, + }); + } + + async unarchive(sessionId: string): Promise { + await this.updateHeader(sessionId, { + isArchived: false, + archivedAt: undefined, + status: 'active', + blockedReason: undefined, + statusUpdatedAt: Date.now(), + }); + } + + async setFlagged(sessionId: string, isFlagged: boolean): Promise { + await this.updateHeader(sessionId, { isFlagged }); + } + + async rename(sessionId: string, name: string): Promise { + const normalized = normalizeUserSessionName(name); + if (!normalized.ok) throw new Error(normalized.error); + await this.updateHeader(sessionId, { + name: normalized.value, + titleIsManual: true, + }); + } + + async setGeneratedTitleIfAbsent(sessionId: string, title: string): Promise { + const normalized = normalizeUserSessionName(title); + if (!normalized.ok) return null; + const current = await this.readHeaderSnapshot(sessionId); + if ( + current.titleIsManual || + current.name !== DEFAULT_SESSION_NAME || + normalized.value === current.name + ) { + return null; + } + return this.updateHeader(sessionId, { name: normalized.value }); + } + + async remove(sessionId: string): Promise { + await this.ensureReady(); + await this.metadata.remove(sessionId); + await this.files.remove(sessionId); + } + + close(): void { + this.metadata.close(); + } + + private async lockConnectionAfterFirstUserMessage( + header: SessionHeader, + knownMessages?: StoredMessage[], + ): Promise { + if (header.connectionLocked) return header; + const messages = knownMessages ?? (await this.files.readMessagesSnapshot(header.id)); + if (!messages.some((message) => message.type === 'user')) return header; + return this.updateHeader(header.id, { connectionLocked: true }); + } + + 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 { private static readonly HEADER_BUDGET = 8192; private static readonly MAX_HEADER_BYTES = 1024 * 1024; @@ -259,6 +498,10 @@ class FileSessionStore implements SessionStore { return (await this.readFileParts(sessionId)).messages; } + async readPreviewMessages(sessionId: string): Promise { + return this.readTailPreviewMessages(sessionId); + } + async readMessagesForRecovery(sessionId: string): Promise { return (await this.readFilePartsUnlocked(sessionId, true)).messages; } diff --git a/packages/storage/src/sqlite-session-metadata-schema.ts b/packages/storage/src/sqlite-session-metadata-schema.ts index 2d08d07f34..64a24206f1 100644 --- a/packages/storage/src/sqlite-session-metadata-schema.ts +++ b/packages/storage/src/sqlite-session-metadata-schema.ts @@ -1,6 +1,6 @@ import type { DatabaseSync } from 'node:sqlite'; -export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 1; +export const SQLITE_SESSION_METADATA_SCHEMA_VERSION = 2; const MIGRATIONS: ReadonlyMap = new Map([ [ @@ -63,6 +63,15 @@ const MIGRATIONS: ReadonlyMap = new Map([ ); `, ], + [ + 2, + ` + CREATE TABLE session_metadata_tombstones ( + session_id TEXT PRIMARY KEY, + deleted_at INTEGER NOT NULL + ); + `, + ], ]); export function configureSqliteSessionMetadataDatabase(db: DatabaseSync): void { diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index ac135485ac..a2d9176e25 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -46,6 +46,7 @@ export interface SessionMetadataImportEntry { export interface SessionMetadataImportResult { created: boolean[]; sourcesAlreadyImported: number; + sourcesTombstoned: number; } export class SessionMetadataConflictError extends Error { @@ -100,6 +101,11 @@ export class SqliteSessionMetadataStore { const normalized = normalizeSessionHeader(header); assertSafeSessionId(normalized.id); return this.transaction(() => { + if (this.hasTombstone(normalized.id)) { + throw new SessionMetadataConflictError( + `Session metadata id is tombstoned: ${normalized.id}`, + ); + } if (this.readRecordSync(normalized.id)) { throw new SessionMetadataConflictError(`Session metadata already exists: ${normalized.id}`); } @@ -236,11 +242,19 @@ export class SqliteSessionMetadataStore { async remove(sessionId: string): Promise { this.assertOpen(); assertSafeSessionId(sessionId); - return this.transaction( - () => + return this.transaction(() => { + const deleted = this.db.prepare('DELETE FROM session_metadata WHERE session_id = ?').run(sessionId) - .changes === 1, - ); + .changes === 1; + this.db + .prepare(` + INSERT INTO session_metadata_tombstones(session_id, deleted_at) + VALUES (?, ?) + ON CONFLICT(session_id) DO NOTHING + `) + .run(sessionId, this.now()); + return deleted; + }); } async importEntries( @@ -263,7 +277,12 @@ export class SqliteSessionMetadataStore { return this.transaction(() => { const created: boolean[] = []; let sourcesAlreadyImported = 0; + let sourcesTombstoned = 0; for (const entry of normalized) { + if (this.hasTombstone(entry.header.id)) { + sourcesTombstoned += 1; + continue; + } const source = this.db .prepare(` SELECT fingerprint @@ -306,7 +325,7 @@ export class SqliteSessionMetadataStore { .run(entry.source.path, entry.source.fingerprint, entry.header.id, this.now()); this.options.failpoint?.('after_session_import_marker_write'); } - return { created, sourcesAlreadyImported }; + return { created, sourcesAlreadyImported, sourcesTombstoned }; }); } @@ -388,6 +407,14 @@ export class SqliteSessionMetadataStore { return row ? decodeRecord(row) : undefined; } + private hasTombstone(sessionId: string): boolean { + return ( + this.db + .prepare('SELECT 1 AS found FROM session_metadata_tombstones WHERE session_id = ?') + .get(sessionId) !== undefined + ); + } + private transaction(operation: () => T): T { this.db.exec('BEGIN IMMEDIATE'); try { From 93fd480d5b8917e12af281869c778441f75acb25 Mon Sep 17 00:00:00 2001 From: likun666661 <90952590+likun666661@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:40:12 +0800 Subject: [PATCH 2/2] Fix SQLite session metadata CI regressions --- apps/desktop/src/main/app-lifecycle.ts | 15 ++-------- apps/desktop/src/main/main.ts | 11 +++++-- .../headless-storage-boundary.test.ts | 8 +++-- .../session-metadata-transfer.test.ts | 29 +++++++++++++++++++ .../storage/src/session-metadata-transfer.ts | 25 ++++++++++++++-- .../src/sqlite-session-metadata-store.ts | 12 ++++++++ 6 files changed, 80 insertions(+), 20 deletions(-) diff --git a/apps/desktop/src/main/app-lifecycle.ts b/apps/desktop/src/main/app-lifecycle.ts index 6fc87cccd1..2ff1fc92c3 100644 --- a/apps/desktop/src/main/app-lifecycle.ts +++ b/apps/desktop/src/main/app-lifecycle.ts @@ -16,7 +16,6 @@ import type { createFileCredentialStore } from './credential-store.js'; import { startConfigFileWatcher, type ConfigFileWatcher } from './config-file-watcher.js'; import { toContractNetworkSettings } from './network-settings-main.js'; import { importLegacyOAuthTokenFiles } from './oauth/shared-credential-bridge.js'; -import { seedE2eFixture } from './e2e-fixture.js'; import type { resolveE2eFixture } from './e2e-fixture.js'; import type { OpenGatewayService } from './open-gateway.js'; import type { KeepSystemAwakeController } from './keep-system-awake.js'; @@ -204,17 +203,9 @@ export function wireAppLifecycle(deps: AppLifecycleDeps): void { // settled. Any state that background startup mutates is pushed to the // renderer via the existing `sessions:changed` / `connections:event` // / `settings:bots:statusChanged` channels, so the UI converges lazily. - // E2e-fixture mode wipes and reseeds the whole workspace - // (`rm -rf` first). That wipe must finish BEFORE the window opens and - // before background startup touches the workspace: createWindow reads - // the settings store, and a concurrent wipe lands inside the store's - // read-or-create write (mkdir → tmp → rename), rejecting createWindow - // so the window never appears. Fixture runs trade first-paint latency - // for determinism by definition; production launches skip this await. - if (e2eFixture) { - console.log(`[e2e-fixture] scenario=${e2eFixture.scenario} workspace=${workspaceRoot}`); - await seedE2eFixture({ workspaceRoot, fixture: e2eFixture, credentialStore }); - } + // E2E fixture workspaces are wiped and seeded before stores open in + // main.ts. SQLite keeps live file handles, so resetting the workspace + // here after store construction would detach the canonical database. await runCredentialStartup(); app.on('second-instance', focusOrCreateMainWindow); app.on('activate', focusOrCreateMainWindow); diff --git a/apps/desktop/src/main/main.ts b/apps/desktop/src/main/main.ts index 4b4d002e7c..535b67fc56 100644 --- a/apps/desktop/src/main/main.ts +++ b/apps/desktop/src/main/main.ts @@ -83,7 +83,7 @@ import { bindOnboardingDeps, createOnboardingService } from './onboarding-servic import { handleQuickChatStart as runQuickChatStart, type QuickChatResult } from './quick-chat.js'; import { createDailyReviewArchiveStore } from './daily-review-archive-store.js'; import { resolveDefaultPermissionMode } from './permission-mode-default.js'; -import { resolveE2eFixture } from './e2e-fixture.js'; +import { resolveE2eFixture, seedE2eFixture } from './e2e-fixture.js'; import { resolveBuildInfo } from './build-info.js'; import { OpenGatewayService } from './open-gateway.js'; import { LocalMemoryService } from './local-memory-service.js'; @@ -196,7 +196,13 @@ try { throw error; } const workspaceRoot = join(app.getPath('userData'), 'workspaces', e2eFixture?.workspaceName ?? 'default'); -await resolveStorageRoot({ path: workspaceRoot, kind: 'interactive' }); +const credentialStore = createFileCredentialStore(workspaceRoot); +if (e2eFixture) { + console.log(`[e2e-fixture] scenario=${e2eFixture.scenario} workspace=${workspaceRoot}`); + await seedE2eFixture({ workspaceRoot, fixture: e2eFixture, credentialStore }); +} else { + await resolveStorageRoot({ path: workspaceRoot, kind: 'interactive' }); +} // 保持系统唤醒 (settings.system.keepSystemAwake): holds an Electron // `powerSaveBlocker` so in-process scheduled tasks keep firing while the // machine would otherwise sleep. Injected with electron's blocker; the @@ -233,7 +239,6 @@ const artifactStore = createArtifactStore(workspaceRoot); const deepResearchStore = createDeepResearchStore(workspaceRoot); const storeReadImage = createReadImageSnapshotter(artifactStore); const attachmentApprovals = createAttachmentApprovalRegistry(); -const credentialStore = createFileCredentialStore(workspaceRoot); // PR-OAUTH-SUBSCRIPTION-0: Claude subscription OAuth service. // Lives in main process only; renderer accesses via IPC. Tokens // never cross the IPC boundary (xuan G-X3). Cloak path is dynamic- diff --git a/packages/headless/src/__tests__/headless-storage-boundary.test.ts b/packages/headless/src/__tests__/headless-storage-boundary.test.ts index 1814a9be2c..1e52d0f0f6 100644 --- a/packages/headless/src/__tests__/headless-storage-boundary.test.ts +++ b/packages/headless/src/__tests__/headless-storage-boundary.test.ts @@ -227,14 +227,18 @@ async function snapshotTree(path: string, label: string): Promise { await rm(root, { recursive: true, force: true }); } }); + + test('keeps canonical metadata readable when its optional transcript is missing', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-session-transfer-missing-transcript-')); + const legacy = createSessionStore(root); + const sqlite = createSqliteSessionMetadataStore(join(root, 'state.sqlite')); + try { + const created = await legacy.create(makeInput({ name: 'Canonical metadata' })); + await importLegacySessionMetadataTree({ workspaceRoot: root, destination: sqlite }); + await rm(join(root, 'sessions', created.id, 'session.jsonl')); + + const report = await importLegacySessionMetadataTree({ + workspaceRoot: root, + destination: sqlite, + }); + + assert.deepEqual(report, { + filesScanned: 1, + headersRead: 0, + headersImported: 0, + headersExisting: 0, + sourcesAlreadyImported: 0, + sourcesTombstoned: 0, + }); + assert.equal((await sqlite.read(created.id)).header.name, 'Canonical metadata'); + } finally { + sqlite.close(); + await rm(root, { recursive: true, force: true }); + } + }); }); function makeInput(overrides: Partial = {}): CreateSessionInput { diff --git a/packages/storage/src/session-metadata-transfer.ts b/packages/storage/src/session-metadata-transfer.ts index a74c362f5e..5027795a49 100644 --- a/packages/storage/src/session-metadata-transfer.ts +++ b/packages/storage/src/session-metadata-transfer.ts @@ -31,14 +31,24 @@ export async function importLegacySessionMetadataTree(input: { }): Promise { const sessionsRoot = join(input.workspaceRoot, 'sessions'); const entries: SessionMetadataImportEntry[] = []; - for (const directory of await sessionDirectoryNames(sessionsRoot)) { + const directories = await sessionDirectoryNames(sessionsRoot); + for (const directory of directories) { const sourcePath = join(sessionsRoot, directory, 'session.jsonl'); - entries.push(await readLegacySessionMetadataEntry(sourcePath, directory)); + try { + entries.push(await readLegacySessionMetadataEntry(sourcePath, directory)); + } catch (error) { + if (!isNotFound(error)) throw error; + const canonicalStateExists = + (await input.destination.has(directory)) || + (await input.destination.isTombstoned(directory)); + if (canonicalStateExists) continue; + throw error; + } } const result = await input.destination.importEntries(entries); const headersImported = result.created.filter(Boolean).length; return { - filesScanned: entries.length, + filesScanned: directories.length, headersRead: entries.length, headersImported, headersExisting: result.created.length - headersImported, @@ -68,6 +78,15 @@ export async function readLegacySessionMetadataEntry( } } +function isNotFound(error: unknown): boolean { + let current = error; + while (current && typeof current === 'object') { + if ('code' in current && current.code === 'ENOENT') return true; + current = 'cause' in current ? current.cause : undefined; + } + return false; +} + async function sessionDirectoryNames(root: string): Promise { let entries; try { diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index a2d9176e25..be4ed5389b 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -121,6 +121,18 @@ export class SqliteSessionMetadataStore { return record; } + async has(sessionId: string): Promise { + this.assertOpen(); + assertSafeSessionId(sessionId); + return this.readRecordSync(sessionId) !== undefined; + } + + async isTombstoned(sessionId: string): Promise { + this.assertOpen(); + assertSafeSessionId(sessionId); + return this.hasTombstone(sessionId); + } + async list(filter: SessionListFilter = {}): Promise { this.assertOpen(); const where: string[] = [];