diff --git a/README.md b/README.md index 89a0fa7fc5..0881543777 100644 --- a/README.md +++ b/README.md @@ -191,6 +191,15 @@ write, Maka batch-idempotently imports legacy RuntimeEvent JSONL without rewriting it. Legacy-only workspaces remain available to read-only inspection until that first write. +Session metadata and Agent Graph control tables now use the same process-local +operational database owner and the same `runtime.sqlite` transaction authority. +The first operational open copies a WAL-consistent `sessions.sqlite` source +into `runtime.sqlite`, validates every source row, and records the source digest +and result in `cutover_journal`. An interrupted copy resumes without partial +rows; a legacy database changed after cutover fails closed. The old database is +retained as migration evidence but is no longer a production writer. Session +transcript bodies remain append-only JSONL. + Runtime continuation remains opt-in: - `MAKA_RUNTIME_SAFE_BOUNDARY_RESUME=1` enables the Desktop interrupted-turn diff --git a/packages/storage/src/__tests__/operational-state-store.test.ts b/packages/storage/src/__tests__/operational-state-store.test.ts new file mode 100644 index 0000000000..890941ac56 --- /dev/null +++ b/packages/storage/src/__tests__/operational-state-store.test.ts @@ -0,0 +1,258 @@ +import assert from 'node:assert/strict'; +import { mkdtemp, rm, stat, writeFile } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; +import { describe, test } from 'node:test'; +import type { SessionHeader } from '@maka/core'; +import { + acquireOperationalStateDatabase, + type OperationalStateCutoverFailpoint, +} from '../operational-state-store.js'; +import { createSqliteRuntimeStore } from '../sqlite-runtime-store.js'; +import { createSqliteSessionMetadataStore } from '../sqlite-session-metadata-store.js'; + +describe('operational state database cutover', () => { + test('imports sessions.sqlite into runtime.sqlite once and rejects later legacy changes', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-operational-cutover-')); + const legacyPath = join(root, 'sessions.sqlite'); + const runtimePath = join(root, 'runtime.sqlite'); + try { + const legacy = createSqliteSessionMetadataStore(legacyPath, { now: () => 10 }); + await legacy.create(sessionHeader()); + legacy.close(); + + const lease = acquireOperationalStateDatabase(root, { now: () => 20 }); + const secondLease = acquireOperationalStateDatabase(root); + assert.equal(secondLease.database, lease.database); + secondLease.close(); + const metadata = createSqliteSessionMetadataStore(runtimePath, { + databaseLease: lease, + now: () => 30, + }); + assert.equal((await metadata.read('session-1')).header.name, 'Legacy session'); + assert.deepEqual( + lease.database + .prepare(`SELECT scope, version FROM operational_schema_migrations ORDER BY scope`) + .all() + .map((row) => ({ ...row })), + [ + { scope: 'operational', version: 1 }, + { scope: 'runtime', version: 5 }, + { scope: 'session_metadata', version: 14 }, + ], + ); + assert.deepEqual( + { + ...(lease.database + .prepare(` + SELECT state, source_path, validation_json + FROM cutover_journal + WHERE store_name = 'session_metadata' + `) + .get() as Record), + }, + { + state: 'completed', + source_path: legacyPath, + validation_json: JSON.stringify({ + session_metadata: 1, + session_metadata_labels: 1, + session_metadata_import_sources: 0, + session_metadata_tombstones: 0, + subagent_spawns: 0, + agent_graph_intent_claims: 0, + agent_graph_schedule_updates: 0, + agent_graph_operator_provisions: 0, + agent_graph_client_projections: 0, + agent_graph_client_operator_projections: 0, + agent_graph_client_terminal_activity: 0, + agent_graph_client_applied_records: 0, + agent_graph_supervisor_wakes: 0, + agent_graph_supervisor_wake_attempts: 0, + sandbox_boundary_log: 1, + }), + }, + ); + metadata.close(); + assert.ok((await stat(legacyPath)).isFile(), 'legacy database remains as cutover evidence'); + + const reopenedLease = acquireOperationalStateDatabase(root); + reopenedLease.close(); + + const changedLegacy = createSqliteSessionMetadataStore(legacyPath); + await changedLegacy.update('session-1', { name: 'Changed by an old binary' }); + changedLegacy.close(); + assert.throws( + () => acquireOperationalStateDatabase(root), + /changed after session metadata cutover completed/, + ); + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + + test('fails closed instead of merging conflicting canonical and legacy rows', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-operational-cutover-conflict-')); + const legacyPath = join(root, 'sessions.sqlite'); + const runtimePath = join(root, 'runtime.sqlite'); + try { + const legacy = createSqliteSessionMetadataStore(legacyPath); + await legacy.create(sessionHeader()); + legacy.close(); + + const runtime = createSqliteRuntimeStore(runtimePath); + runtime.close(); + const canonical = createSqliteSessionMetadataStore(runtimePath); + await canonical.create(sessionHeader({ name: 'Canonical session' })); + canonical.close(); + + assert.throws( + () => acquireOperationalStateDatabase(root), + /Session metadata cutover conflict in table session_metadata/, + ); + const verified = createSqliteSessionMetadataStore(runtimePath); + try { + assert.equal((await verified.read('session-1')).header.name, 'Canonical session'); + } finally { + verified.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + + test('rejects an empty legacy database instead of recording an empty cutover', async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-operational-cutover-empty-')); + try { + await writeFile(join(root, 'sessions.sqlite'), ''); + assert.throws( + () => acquireOperationalStateDatabase(root), + /not a non-empty regular database file/, + ); + const target = new DatabaseSync(join(root, 'runtime.sqlite')); + try { + assert.equal( + ( + target.prepare(`SELECT COUNT(*) AS count FROM cutover_journal`).get() as { + count: number; + } + ).count, + 0, + ); + } finally { + target.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + + for (const failpoint of [ + 'after_cutover_started', + 'after_cutover_rows_copied', + 'after_cutover_validated', + ] satisfies OperationalStateCutoverFailpoint[]) { + test(`resumes atomically after ${failpoint}`, async () => { + const root = await mkdtemp(join(tmpdir(), 'maka-operational-cutover-crash-')); + const legacyPath = join(root, 'sessions.sqlite'); + const runtimePath = join(root, 'runtime.sqlite'); + try { + const legacy = createSqliteSessionMetadataStore(legacyPath); + await legacy.create(sessionHeader()); + legacy.close(); + + assert.throws( + () => + acquireOperationalStateDatabase(root, { + failpoint: (point) => { + if (point === failpoint) throw new Error(`failpoint:${point}`); + }, + }), + new RegExp(`failpoint:${failpoint}`), + ); + + const interrupted = new DatabaseSync(runtimePath); + try { + assert.deepEqual( + { + ...(interrupted + .prepare(` + SELECT state, completed_at + FROM cutover_journal + WHERE store_name = 'session_metadata' + `) + .get() as Record), + }, + { state: 'started', completed_at: null }, + ); + assert.equal( + ( + interrupted.prepare(`SELECT COUNT(*) AS count FROM session_metadata`).get() as { + count: number; + } + ).count, + 0, + ); + } finally { + interrupted.close(); + } + + const resumed = acquireOperationalStateDatabase(root); + try { + assert.equal( + ( + resumed.database.prepare(`SELECT COUNT(*) AS count FROM session_metadata`).get() as { + count: number; + } + ).count, + 1, + ); + assert.equal( + ( + resumed.database + .prepare(` + SELECT state + FROM cutover_journal + WHERE store_name = 'session_metadata' + `) + .get() as { state: string } + ).state, + 'completed', + ); + } finally { + resumed.close(); + } + } finally { + await rm(root, { recursive: true, force: true }); + } + }); + } +}); + +function sessionHeader(overrides: Partial = {}): SessionHeader { + return { + id: 'session-1', + workspaceRoot: '/workspace', + cwd: '/workspace', + createdAt: 1, + lastUsedAt: 2, + name: 'Legacy session', + titleIsManual: true, + isFlagged: false, + labels: ['legacy'], + isArchived: false, + status: 'active', + hasUnread: false, + backend: 'fake', + llmConnectionSlug: 'test', + connectionLocked: true, + model: 'test-model', + thinkingLevel: 'medium', + permissionMode: 'ask', + collaborationMode: 'agent', + orchestrationMode: 'swarm', + schemaVersion: 1, + ...overrides, + }; +} diff --git a/packages/storage/src/__tests__/session-bundle-policy.test.ts b/packages/storage/src/__tests__/session-bundle-policy.test.ts index 0ef8c2c694..527f4a324a 100644 --- a/packages/storage/src/__tests__/session-bundle-policy.test.ts +++ b/packages/storage/src/__tests__/session-bundle-policy.test.ts @@ -114,7 +114,6 @@ test('exports one session only and excludes credential/config canaries', async ( '.maka_cli_claude_device_id', 'credentials.json', 'llm-connections.json', - 'sessions.sqlite', `artifacts/${other.id}`, `sessions/${other.id}`, ].sort(), @@ -182,7 +181,7 @@ test('exports one session only and excludes credential/config canaries', async ( }); }); -test('exports while the session metadata WAL remains open and protects its sidecars', async () => { +test('exports while the operational DB remains open and protects its sidecars', async () => { await withBundleRoots(async ({ stateRoot, configRoot, destinationRoot }) => { const sessions = createSessionStore(stateRoot); try { @@ -195,7 +194,7 @@ test('exports while the session metadata WAL remains open and protects its sidec text: 'selected transcript', }); - const sidecars = ['sessions.sqlite-wal', 'sessions.sqlite-shm']; + const sidecars = ['runtime.sqlite-wal', 'runtime.sqlite-shm']; const liveEntries = await readdir(stateRoot); for (const sidecar of sidecars) { assert.ok(liveEntries.includes(sidecar), liveEntries.join(', ')); diff --git a/packages/storage/src/__tests__/sqlite-session-store.test.ts b/packages/storage/src/__tests__/sqlite-session-store.test.ts index 7b3e4daef8..765f975771 100644 --- a/packages/storage/src/__tests__/sqlite-session-store.test.ts +++ b/packages/storage/src/__tests__/sqlite-session-store.test.ts @@ -97,7 +97,7 @@ describe('default SQLite session metadata store', () => { 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' }); + await assert.rejects(() => stat(join(root, 'sessions.sqlite')), { code: 'ENOENT' }); await store.close?.(); const reopened = createSessionStore(root); diff --git a/packages/storage/src/agent-graph-control-store.ts b/packages/storage/src/agent-graph-control-store.ts index 3f25b279ae..36da21bc7c 100644 --- a/packages/storage/src/agent-graph-control-store.ts +++ b/packages/storage/src/agent-graph-control-store.ts @@ -1,17 +1,21 @@ import { join } from 'node:path'; -import { SQLITE_SESSION_METADATA_DATABASE_NAME } from './session-store.js'; +import { + acquireOperationalStateDatabase, + OPERATIONAL_STATE_DATABASE_NAME, +} from './operational-state-store.js'; import { createSqliteSessionMetadataStore, type SqliteSessionMetadataStore, } from './sqlite-session-metadata-store.js'; /** - * Open the SQLite metadata control plane used by one host-owned agent graph - * coordinator. Session JSONL remains the transcript authority; graph - * schedule/topology/claim relationships are queried from SQLite. + * Open the metadata repository owned by the operational database. Session + * JSONL remains the transcript-body authority; graph schedule/topology/claim + * relationships are canonical in runtime.sqlite. */ export function createAgentGraphControlStore(workspaceRoot: string): SqliteSessionMetadataStore { - return createSqliteSessionMetadataStore( - join(workspaceRoot, SQLITE_SESSION_METADATA_DATABASE_NAME), - ); + const databaseLease = acquireOperationalStateDatabase(workspaceRoot); + return createSqliteSessionMetadataStore(join(workspaceRoot, OPERATIONAL_STATE_DATABASE_NAME), { + databaseLease, + }); } diff --git a/packages/storage/src/index.ts b/packages/storage/src/index.ts index 73114eb286..377bd2af33 100644 --- a/packages/storage/src/index.ts +++ b/packages/storage/src/index.ts @@ -50,6 +50,7 @@ export * from './config-transfer.js'; export * from './automation-store.js'; export * from './sqlite-runtime-store.js'; export * from './runtime-event-transfer.js'; +export * from './operational-state-store.js'; export * from './mcp-config-store.js'; export * from './workspace-identity.js'; export * from './memory-bundle-store.js'; diff --git a/packages/storage/src/operational-state-store.ts b/packages/storage/src/operational-state-store.ts new file mode 100644 index 0000000000..192874e63b --- /dev/null +++ b/packages/storage/src/operational-state-store.ts @@ -0,0 +1,484 @@ +import { createHash } from 'node:crypto'; +import { existsSync, lstatSync, mkdirSync } from 'node:fs'; +import { dirname, join, resolve } from 'node:path'; +import { createRequire } from 'node:module'; +import type { DatabaseSync } from 'node:sqlite'; +import { + configureSqliteRuntimeDatabase, + migrateSqliteRuntimeDatabase, + SQLITE_RUNTIME_SCHEMA_VERSION, +} from './sqlite-runtime-schema.js'; +import { + migrateSqliteSessionMetadataDatabase, + SQLITE_SESSION_METADATA_SCHEMA_VERSION, +} from './sqlite-session-metadata-schema.js'; + +export const OPERATIONAL_STATE_DATABASE_NAME = 'runtime.sqlite'; +export const LEGACY_SESSION_METADATA_DATABASE_NAME = 'sessions.sqlite'; +export const OPERATIONAL_STATE_SCHEMA_VERSION = 1; + +const SESSION_METADATA_TABLES = [ + 'session_metadata', + 'session_metadata_labels', + 'session_metadata_import_sources', + 'session_metadata_tombstones', + 'subagent_spawns', + 'agent_graph_intent_claims', + 'agent_graph_schedule_updates', + 'agent_graph_operator_provisions', + 'agent_graph_client_projections', + 'agent_graph_client_operator_projections', + 'agent_graph_client_terminal_activity', + 'agent_graph_client_applied_records', + 'agent_graph_supervisor_wakes', + 'agent_graph_supervisor_wake_attempts', + 'sandbox_boundary_log', +] as const; + +const require = createRequire(import.meta.url); +const owners = new Map(); + +export type OperationalStateCutoverFailpoint = + | 'after_cutover_started' + | 'after_cutover_rows_copied' + | 'after_cutover_validated'; + +export interface OperationalStateDatabaseOptions { + now?: () => number; + failpoint?: (point: OperationalStateCutoverFailpoint) => void; +} + +export interface OperationalStateDatabaseLease { + readonly database: DatabaseSync; + readonly databasePath: string; + transaction(mode: 'read' | 'write', operation: () => T): T; + close(): void; +} + +interface CutoverJournalRow { + source_fingerprint: string; + state: 'started' | 'completed'; +} + +interface TableColumn { + name: string; + pk: number; +} + +/** + * Acquire the process-local owner for the operational SQLite authority. + * + * Repositories receive leases instead of opening independent connections. + * The last lease closes the connection, while transaction boundaries remain + * centralized on the owner for the lifetime of the workspace. + */ +export function acquireOperationalStateDatabase( + workspaceRoot: string, + options: OperationalStateDatabaseOptions = {}, +): OperationalStateDatabaseLease { + const databasePath = resolve(workspaceRoot, OPERATIONAL_STATE_DATABASE_NAME); + let owner = owners.get(databasePath); + if (!owner) { + owner = new OperationalStateDatabaseOwner(databasePath, options); + owners.set(databasePath, owner); + } + return owner.acquire(); +} + +class OperationalStateDatabaseOwner { + readonly database: DatabaseSync; + private references = 0; + private closed = false; + private transactionDepth = 0; + + constructor( + readonly databasePath: string, + options: OperationalStateDatabaseOptions, + ) { + mkdirSync(dirname(databasePath), { recursive: true }); + const Database = loadDatabaseSync(); + this.database = new Database(databasePath); + try { + configureSqliteRuntimeDatabase(this.database); + migrateSqliteRuntimeDatabase(this.database); + migrateSqliteSessionMetadataDatabase(this.database); + migrateOperationalStateDatabase(this.database, options.now ?? Date.now); + cutoverLegacySessionMetadata({ + destination: this.database, + destinationPath: databasePath, + sourcePath: join(dirname(databasePath), LEGACY_SESSION_METADATA_DATABASE_NAME), + now: options.now ?? Date.now, + failpoint: options.failpoint, + }); + } catch (error) { + this.database.close(); + this.closed = true; + throw error; + } + } + + acquire(): OperationalStateDatabaseLease { + if (this.closed) throw new Error('Operational state database is closed'); + this.references += 1; + let released = false; + return { + database: this.database, + databasePath: this.databasePath, + transaction: (mode, operation) => this.transaction(mode, operation), + close: () => { + if (released) return; + released = true; + this.references -= 1; + if (this.references !== 0) return; + this.closed = true; + owners.delete(this.databasePath); + this.database.close(); + }, + }; + } + + private transaction(mode: 'read' | 'write', operation: () => T): T { + if (this.closed) throw new Error('Operational state database is closed'); + if (this.transactionDepth > 0) return operation(); + this.database.exec(mode === 'write' ? 'BEGIN IMMEDIATE' : 'BEGIN'); + this.transactionDepth += 1; + try { + const result = operation(); + this.database.exec('COMMIT'); + return result; + } catch (error) { + rollback(this.database); + throw error; + } finally { + this.transactionDepth -= 1; + } + } +} + +function migrateOperationalStateDatabase(db: DatabaseSync, now: () => number): void { + db.exec('BEGIN IMMEDIATE'); + try { + db.exec(` + CREATE TABLE IF NOT EXISTS operational_schema_migrations ( + scope TEXT PRIMARY KEY, + version INTEGER NOT NULL CHECK (version >= 0), + applied_at INTEGER NOT NULL CHECK (applied_at >= 0) + ); + + CREATE TABLE IF NOT EXISTS cutover_journal ( + store_name TEXT PRIMARY KEY, + source_path TEXT NOT NULL, + source_fingerprint TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('started', 'completed')), + started_at INTEGER NOT NULL CHECK (started_at >= 0), + completed_at INTEGER, + validation_json TEXT + ); + `); + const appliedAt = now(); + registerSchema(db, 'runtime', SQLITE_RUNTIME_SCHEMA_VERSION, appliedAt); + registerSchema(db, 'session_metadata', SQLITE_SESSION_METADATA_SCHEMA_VERSION, appliedAt); + registerSchema(db, 'operational', OPERATIONAL_STATE_SCHEMA_VERSION, appliedAt); + db.exec('COMMIT'); + } catch (error) { + rollback(db); + throw error; + } +} + +function registerSchema(db: DatabaseSync, scope: string, version: number, appliedAt: number): void { + const existing = db + .prepare('SELECT version FROM operational_schema_migrations WHERE scope = ?') + .get(scope) as { version?: unknown } | undefined; + if ( + existing && + (typeof existing.version !== 'number' || + !Number.isSafeInteger(existing.version) || + existing.version > version) + ) { + throw new Error(`Operational schema ${scope} is newer than supported version ${version}`); + } + db.prepare(` + INSERT INTO operational_schema_migrations(scope, version, applied_at) + VALUES (?, ?, ?) + ON CONFLICT(scope) DO UPDATE SET + version = excluded.version, + applied_at = CASE + WHEN operational_schema_migrations.version = excluded.version + THEN operational_schema_migrations.applied_at + ELSE excluded.applied_at + END + `).run(scope, version, appliedAt); +} + +function cutoverLegacySessionMetadata(input: { + destination: DatabaseSync; + destinationPath: string; + sourcePath: string; + now: () => number; + failpoint?: (point: OperationalStateCutoverFailpoint) => void; +}): void { + if ( + !existsSync(input.sourcePath) || + resolve(input.sourcePath) === resolve(input.destinationPath) + ) { + return; + } + const sourceStat = lstatSync(input.sourcePath); + if (!sourceStat.isFile() || sourceStat.isSymbolicLink() || sourceStat.size === 0) { + throw new Error('Legacy sessions.sqlite is not a non-empty regular database file'); + } + + const Database = loadDatabaseSync(); + const source = new Database(input.sourcePath); + try { + configureSqliteRuntimeDatabase(source); + migrateSqliteSessionMetadataDatabase(source); + source.exec('BEGIN IMMEDIATE'); + try { + const fingerprint = fingerprintSessionMetadata(source); + const journal = input.destination + .prepare(` + SELECT source_fingerprint, state + FROM cutover_journal + WHERE store_name = 'session_metadata' + `) + .get() as CutoverJournalRow | undefined; + if (journal?.state === 'completed') { + if (journal.source_fingerprint !== fingerprint) { + throw new Error( + 'Legacy sessions.sqlite changed after session metadata cutover completed', + ); + } + source.exec('COMMIT'); + return; + } + if (journal && journal.source_fingerprint !== fingerprint) { + throw new Error('Legacy sessions.sqlite changed after session metadata cutover started'); + } + if (!journal) { + input.destination.exec('BEGIN IMMEDIATE'); + try { + input.destination + .prepare(` + INSERT INTO cutover_journal( + store_name, + source_path, + source_fingerprint, + state, + started_at + ) VALUES ('session_metadata', ?, ?, 'started', ?) + `) + .run(input.sourcePath, fingerprint, input.now()); + input.destination.exec('COMMIT'); + } catch (error) { + rollback(input.destination); + throw error; + } + } + input.failpoint?.('after_cutover_started'); + + input.destination.exec('BEGIN IMMEDIATE'); + let attached = false; + try { + input.destination.exec( + `ATTACH DATABASE ${quoteString(input.sourcePath)} AS legacy_sessions`, + ); + attached = true; + const validation: Record = {}; + for (const table of SESSION_METADATA_TABLES) { + const columns = readTableColumns(source, table); + assertNoCutoverConflicts(input.destination, table, columns); + const names = columns.map((column) => quoteIdentifier(column.name)).join(', '); + input.destination.exec(` + INSERT OR IGNORE INTO main.${quoteIdentifier(table)} (${names}) + SELECT ${names} FROM legacy_sessions.${quoteIdentifier(table)} + `); + } + input.failpoint?.('after_cutover_rows_copied'); + for (const table of SESSION_METADATA_TABLES) { + const columns = readTableColumns(source, table); + assertAllSourceRowsCopied(input.destination, table, columns); + validation[table] = readRowCount(source, table); + } + input.failpoint?.('after_cutover_validated'); + input.destination + .prepare(` + UPDATE cutover_journal + SET state = 'completed', completed_at = ?, validation_json = ? + WHERE store_name = 'session_metadata' + AND source_fingerprint = ? + AND state = 'started' + `) + .run(input.now(), JSON.stringify(validation), fingerprint); + input.destination.exec('COMMIT'); + } catch (error) { + rollback(input.destination); + throw error; + } finally { + if (attached) { + input.destination.exec('DETACH DATABASE legacy_sessions'); + } + } + source.exec('COMMIT'); + } catch (error) { + rollback(source); + throw error; + } + } finally { + source.close(); + } +} + +function fingerprintSessionMetadata(db: DatabaseSync): string { + const hash = createHash('sha256'); + hash.update(`session_metadata_schema:${SQLITE_SESSION_METADATA_SCHEMA_VERSION}\n`); + for (const table of SESSION_METADATA_TABLES) { + const columns = readTableColumns(db, table); + const order = primaryKeyColumns(columns); + const rows = db + .prepare( + `SELECT * FROM ${quoteIdentifier(table)} ORDER BY ${order + .map((column) => quoteIdentifier(column)) + .join(', ')}`, + ) + .all() as Record[]; + hash.update(`${table}:${rows.length}\n`); + for (const row of rows) { + hash.update(JSON.stringify(columns.map((column) => row[column.name]))); + hash.update('\n'); + } + } + return `sha256:${hash.digest('hex')}`; +} + +function assertNoCutoverConflicts( + db: DatabaseSync, + table: string, + columns: readonly TableColumn[], +): void { + const primaryKey = primaryKeyColumns(columns); + const identity = primaryKey + .map( + (column) => + `main.${quoteIdentifier(table)}.${quoteIdentifier(column)} IS legacy.${quoteIdentifier(column)}`, + ) + .join(' AND '); + const equality = columns + .map( + (column) => + `main.${quoteIdentifier(table)}.${quoteIdentifier(column.name)} IS legacy.${quoteIdentifier(column.name)}`, + ) + .join(' AND '); + const row = db + .prepare(` + SELECT COUNT(*) AS count + FROM legacy_sessions.${quoteIdentifier(table)} AS legacy + JOIN main.${quoteIdentifier(table)} + ON ${identity} + WHERE NOT (${equality}) + `) + .get() as { count?: unknown } | undefined; + if (row?.count !== 0) { + throw new Error(`Session metadata cutover conflict in table ${table}`); + } +} + +function assertAllSourceRowsCopied( + db: DatabaseSync, + table: string, + columns: readonly TableColumn[], +): void { + const equality = columns + .map( + (column) => + `main.${quoteIdentifier(table)}.${quoteIdentifier(column.name)} IS legacy.${quoteIdentifier(column.name)}`, + ) + .join(' AND '); + const row = db + .prepare(` + SELECT COUNT(*) AS count + FROM legacy_sessions.${quoteIdentifier(table)} AS legacy + WHERE NOT EXISTS ( + SELECT 1 FROM main.${quoteIdentifier(table)} + WHERE ${equality} + ) + `) + .get() as { count?: unknown } | undefined; + if (row?.count !== 0) { + throw new Error(`Session metadata cutover validation failed for table ${table}`); + } +} + +function readTableColumns(db: DatabaseSync, table: string): TableColumn[] { + const rows = db.prepare(`PRAGMA table_info(${quoteIdentifier(table)})`).all() as Array<{ + name?: unknown; + pk?: unknown; + }>; + const columns = rows.map((row) => { + if ( + typeof row.name !== 'string' || + typeof row.pk !== 'number' || + !Number.isSafeInteger(row.pk) + ) { + throw new Error(`Invalid SQLite table metadata for ${table}`); + } + return { name: row.name, pk: row.pk }; + }); + if (columns.length === 0 || columns.every((column) => column.pk === 0)) { + throw new Error(`Session metadata cutover table ${table} has no primary key`); + } + return columns; +} + +function primaryKeyColumns(columns: readonly TableColumn[]): string[] { + return columns + .filter((column) => column.pk > 0) + .sort((left, right) => left.pk - right.pk) + .map((column) => column.name); +} + +function readRowCount(db: DatabaseSync, table: string): number { + const row = db.prepare(`SELECT COUNT(*) AS count FROM ${quoteIdentifier(table)}`).get() as + | { count?: unknown } + | undefined; + if (typeof row?.count !== 'number' || !Number.isSafeInteger(row.count) || row.count < 0) { + throw new Error(`Invalid row count for ${table}`); + } + return row.count; +} + +function quoteIdentifier(value: string): string { + return `"${value.replaceAll('"', '""')}"`; +} + +function quoteString(value: string): string { + return `'${value.replaceAll("'", "''")}'`; +} + +function loadDatabaseSync(): typeof import('node:sqlite').DatabaseSync { + 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')).DatabaseSync; + } finally { + process.emitWarning = emitWarning; + } +} + +function rollback(db: DatabaseSync): void { + try { + db.exec('ROLLBACK'); + } catch { + // Preserve the failure that triggered rollback. + } +} diff --git a/packages/storage/src/runtime-event-transfer.ts b/packages/storage/src/runtime-event-transfer.ts index f65634b344..7ae17e457f 100644 --- a/packages/storage/src/runtime-event-transfer.ts +++ b/packages/storage/src/runtime-event-transfer.ts @@ -5,8 +5,12 @@ import { createRuntimeEventStore, type DurableRuntimeEventStore } from './agent- import { classifyJsonRecord } from './json-prefix.js'; import type { SqliteRuntimeStore } from './sqlite-runtime-store.js'; import { createSqliteRuntimeStore } from './sqlite-runtime-store.js'; +import { + acquireOperationalStateDatabase, + OPERATIONAL_STATE_DATABASE_NAME, +} from './operational-state-store.js'; -export const SQLITE_RUNTIME_DATABASE_NAME = 'runtime.sqlite'; +export const SQLITE_RUNTIME_DATABASE_NAME = OPERATIONAL_STATE_DATABASE_NAME; const SAFE_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/; export type RuntimeEventPersistence = { @@ -37,7 +41,8 @@ export async function openRuntimeEventPersistence(input: { workspaceRoot: string; }): Promise { const databasePath = join(input.workspaceRoot, SQLITE_RUNTIME_DATABASE_NAME); - const store = createSqliteRuntimeStore(databasePath); + const databaseLease = acquireOperationalStateDatabase(input.workspaceRoot); + const store = createSqliteRuntimeStore(databasePath, { databaseLease }); try { const importReport = await importLegacyRuntimeEventJsonlTree({ workspaceRoot: input.workspaceRoot, diff --git a/packages/storage/src/session-bundle-policy.ts b/packages/storage/src/session-bundle-policy.ts index 16ae0b3ec3..f65e7b67e3 100644 --- a/packages/storage/src/session-bundle-policy.ts +++ b/packages/storage/src/session-bundle-policy.ts @@ -26,7 +26,10 @@ import { encodeExecutionBoundaryTransfer, EXECUTION_BOUNDARY_TRANSFER_FILE, } from './session-metadata-transfer.js'; -import { SQLITE_SESSION_METADATA_DATABASE_NAME } from './session-store.js'; +import { + LEGACY_SESSION_METADATA_DATABASE_NAME, + OPERATIONAL_STATE_DATABASE_NAME, +} from './operational-state-store.js'; import { createSqliteSessionMetadataStore } from './sqlite-session-metadata-store.js'; import type { SessionAuthoritySnapshot } from './sqlite-session-metadata-store.js'; import { createSqliteRuntimeStore } from './sqlite-runtime-store.js'; @@ -72,10 +75,10 @@ export const SESSION_BUNDLE_PROTECTED_ENTRIES = [ 'activation-input.json', '.maka', ARTIFACT_WRITER_LOCK_FILE, - SQLITE_SESSION_METADATA_DATABASE_NAME, - `${SQLITE_SESSION_METADATA_DATABASE_NAME}-wal`, - `${SQLITE_SESSION_METADATA_DATABASE_NAME}-shm`, - `${SQLITE_SESSION_METADATA_DATABASE_NAME}-journal`, + LEGACY_SESSION_METADATA_DATABASE_NAME, + `${LEGACY_SESSION_METADATA_DATABASE_NAME}-wal`, + `${LEGACY_SESSION_METADATA_DATABASE_NAME}-shm`, + `${LEGACY_SESSION_METADATA_DATABASE_NAME}-journal`, 'runtime.sqlite-wal', 'runtime.sqlite-shm', 'runtime.sqlite-journal', @@ -534,7 +537,7 @@ async function readSelectedSessionAuthoritySnapshot( plan: SessionBundleExportPlan, ): Promise { const metadata = createSqliteSessionMetadataStore( - resolve(plan.stateRoot, SQLITE_SESSION_METADATA_DATABASE_NAME), + resolve(plan.stateRoot, OPERATIONAL_STATE_DATABASE_NAME), ); try { return await metadata.readSessionAuthoritySnapshot(plan.sessionId); diff --git a/packages/storage/src/session-metadata-maintenance.ts b/packages/storage/src/session-metadata-maintenance.ts index 573fd37190..92ffaab660 100644 --- a/packages/storage/src/session-metadata-maintenance.ts +++ b/packages/storage/src/session-metadata-maintenance.ts @@ -6,7 +6,8 @@ import { encodeExecutionBoundaryTransfer, EXECUTION_BOUNDARY_TRANSFER_FILE, } from './session-metadata-transfer.js'; -import { decodeSessionHeader, SQLITE_SESSION_METADATA_DATABASE_NAME } from './session-store.js'; +import { decodeSessionHeader } from './session-store.js'; +import { OPERATIONAL_STATE_DATABASE_NAME } from './operational-state-store.js'; import { createSqliteSessionMetadataStore, type SessionMetadataRecord, @@ -42,7 +43,7 @@ export async function exportLegacySessionTree(input: { now?: () => number; }): Promise { const workspaceRoot = resolve(input.workspaceRoot); - const databasePath = join(workspaceRoot, SQLITE_SESSION_METADATA_DATABASE_NAME); + const databasePath = join(workspaceRoot, OPERATIONAL_STATE_DATABASE_NAME); await assertFileExists(databasePath, 'SQLite session metadata database'); const metadata = createSqliteSessionMetadataStore(databasePath); try { @@ -154,7 +155,7 @@ 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 databasePath = join(resolve(input.workspaceRoot), OPERATIONAL_STATE_DATABASE_NAME); const destinationPath = resolve(input.destinationPath); await assertFileExists(databasePath, 'SQLite session metadata database'); const metadata = createSqliteSessionMetadataStore(databasePath); diff --git a/packages/storage/src/session-store.ts b/packages/storage/src/session-store.ts index 34c411dcc3..a4e8755fc1 100644 --- a/packages/storage/src/session-store.ts +++ b/packages/storage/src/session-store.ts @@ -19,6 +19,10 @@ import { isSessionTranscriptMarker, } from './session-transcript.js'; import { chainWrite } from './write-queue.js'; +import { + acquireOperationalStateDatabase, + OPERATIONAL_STATE_DATABASE_NAME, +} from './operational-state-store.js'; import { DEFAULT_SESSION_NAME, deriveTurnRecords, @@ -52,7 +56,8 @@ import type { } from '@maka/core'; const SESSION_ID_PATTERN = /^[A-Za-z0-9_-]{1,128}$/; -export const SQLITE_SESSION_METADATA_DATABASE_NAME = 'sessions.sqlite'; +/** @deprecated Session metadata is canonical in the operational runtime.sqlite database. */ +export const SQLITE_SESSION_METADATA_DATABASE_NAME = OPERATIONAL_STATE_DATABASE_NAME; export class SessionNotFoundError extends Error { readonly name = 'SessionNotFoundError'; @@ -142,8 +147,10 @@ class SqliteSessionStore implements SessionStore { constructor(workspaceRoot: string) { this.files = new FileSessionStore(workspaceRoot); + const databaseLease = acquireOperationalStateDatabase(workspaceRoot); this.metadata = createSqliteSessionMetadataStore( - join(workspaceRoot, SQLITE_SESSION_METADATA_DATABASE_NAME), + join(workspaceRoot, OPERATIONAL_STATE_DATABASE_NAME), + { databaseLease }, ); this.ready = importLegacySessionMetadataTree({ workspaceRoot, diff --git a/packages/storage/src/sqlite-runtime-store.ts b/packages/storage/src/sqlite-runtime-store.ts index 33f4dea271..38ec74c36c 100644 --- a/packages/storage/src/sqlite-runtime-store.ts +++ b/packages/storage/src/sqlite-runtime-store.ts @@ -34,6 +34,7 @@ import { SQLITE_RUNTIME_SCHEMA_VERSION, } from './sqlite-runtime-schema.js'; import type { ImmutableSteeringMessageProof } from './agent-run-store.js'; +import type { OperationalStateDatabaseLease } from './operational-state-store.js'; import { immutableSteeringMessageId, isRuntimeStorageSafeId } from './runtime-event-invariants.js'; export { SQLITE_RUNTIME_SCHEMA_VERSION } from './sqlite-runtime-schema.js'; @@ -69,6 +70,8 @@ export type SqliteRuntimeStoreFailpoint = export interface SqliteRuntimeStoreOptions { failpoint?: (point: SqliteRuntimeStoreFailpoint) => void; readOnly?: boolean; + /** @internal Repository connection supplied by the operational DB owner. */ + databaseLease?: OperationalStateDatabaseLease; } export interface CommitToolPreparedInput { @@ -148,13 +151,23 @@ export class SqliteRuntimeStore implements RuntimeRecoveryBundleStore { readonly toolBoundaryProtocol = 't1_after_preflight_v1' as const; readonly recoveryBundleCapability = TOOL_RECOVERY_BUNDLE_CAPABILITY_V1; private readonly db: DatabaseSync; + private readonly databaseLease?: OperationalStateDatabaseLease; private closed = false; constructor( path: string, private readonly options: SqliteRuntimeStoreOptions = {}, ) { + if (options.readOnly && options.databaseLease) { + throw new Error('Operational state database leases cannot be opened read-only'); + } if (path !== ':memory:' && !options.readOnly) mkdirSync(dirname(path), { recursive: true }); + if (options.databaseLease) { + this.databaseLease = options.databaseLease; + this.db = options.databaseLease.database; + assertRecoveryAuthorityCapability(this.db); + return; + } const DatabaseSync = loadDatabaseSync(); this.db = options.readOnly ? new DatabaseSync(path, { readOnly: true }) @@ -201,7 +214,8 @@ export class SqliteRuntimeStore implements RuntimeRecoveryBundleStore { close(): void { if (this.closed) return; this.closed = true; - this.db.close(); + if (this.databaseLease) this.databaseLease.close(); + else this.db.close(); } async appendRuntimeEvent( @@ -917,6 +931,7 @@ export class SqliteRuntimeStore implements RuntimeRecoveryBundleStore { } private transaction(operation: () => T): T { + if (this.databaseLease) return this.databaseLease.transaction('write', operation); this.db.exec('BEGIN IMMEDIATE'); try { const result = operation(); diff --git a/packages/storage/src/sqlite-session-metadata-store.ts b/packages/storage/src/sqlite-session-metadata-store.ts index df27456e79..f6f6778236 100644 --- a/packages/storage/src/sqlite-session-metadata-store.ts +++ b/packages/storage/src/sqlite-session-metadata-store.ts @@ -68,6 +68,7 @@ import { migrateSqliteSessionMetadataDatabase, readSqliteSessionMetadataSchemaVersion, } from './sqlite-session-metadata-schema.js'; +import type { OperationalStateDatabaseLease } from './operational-state-store.js'; export { SQLITE_SESSION_METADATA_SCHEMA_VERSION } from './sqlite-session-metadata-schema.js'; @@ -104,6 +105,8 @@ export type SqliteSessionMetadataStoreFailpoint = export interface SqliteSessionMetadataStoreOptions { now?: () => number; failpoint?: (point: SqliteSessionMetadataStoreFailpoint) => void; + /** @internal Repository connection supplied by the operational DB owner. */ + databaseLease?: OperationalStateDatabaseLease; } export interface SessionMetadataRecord { @@ -163,6 +166,7 @@ export function createSqliteSessionMetadataStore( export class SqliteSessionMetadataStore { private readonly db: DatabaseSync; + private readonly databaseLease?: OperationalStateDatabaseLease; private readonly now: () => number; private closed = false; @@ -171,6 +175,12 @@ export class SqliteSessionMetadataStore { private readonly options: SqliteSessionMetadataStoreOptions = {}, ) { if (path !== ':memory:') mkdirSync(dirname(path), { recursive: true }); + if (options.databaseLease) { + this.databaseLease = options.databaseLease; + this.db = options.databaseLease.database; + this.now = options.now ?? Date.now; + return; + } const { DatabaseSync } = loadSqliteModule(); this.db = new DatabaseSync(path); configureSqliteSessionMetadataDatabase(this.db); @@ -194,7 +204,8 @@ export class SqliteSessionMetadataStore { close(): void { if (this.closed) return; this.closed = true; - this.db.close(); + if (this.databaseLease) this.databaseLease.close(); + else this.db.close(); } async backup(destinationPath: string): Promise { @@ -2675,6 +2686,7 @@ export class SqliteSessionMetadataStore { } private transaction(operation: () => T): T { + if (this.databaseLease) return this.databaseLease.transaction('write', operation); this.db.exec('BEGIN IMMEDIATE'); try { const result = operation(); @@ -2691,6 +2703,7 @@ export class SqliteSessionMetadataStore { } private readTransaction(operation: () => T): T { + if (this.databaseLease) return this.databaseLease.transaction('read', operation); this.db.exec('BEGIN'); try { const result = operation(); diff --git a/packages/storage/src/usage-stats-store.ts b/packages/storage/src/usage-stats-store.ts index 4284352989..d5b0da829f 100644 --- a/packages/storage/src/usage-stats-store.ts +++ b/packages/storage/src/usage-stats-store.ts @@ -2,7 +2,7 @@ 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 { OPERATIONAL_STATE_DATABASE_NAME } from './operational-state-store.js'; import { createSqliteSessionMetadataStore } from './sqlite-session-metadata-store.js'; type UsageSessionHeader = Pick; @@ -162,7 +162,7 @@ async function readStoredSessions( async function readCanonicalUsageHeaders( workspaceRoot: string, ): Promise | null> { - const path = join(workspaceRoot, SQLITE_SESSION_METADATA_DATABASE_NAME); + const path = join(workspaceRoot, OPERATIONAL_STATE_DATABASE_NAME); try { await stat(path); } catch (error) {