You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by as...@apache.org on 2026/08/29 09:47:59 UTC
(maka) branch main updated: refactor(storage): retire unread workflow projection tables (#3937)
This is an automated email from the ASF dual-hosted git repository.
Astro-Han pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/maka.git
The following commit(s) were added to refs/heads/main by this push:
new 319b7ed0e refactor(storage): retire unread workflow projection tables (#3937)
319b7ed0e is described below
commit 319b7ed0e30710ef29a62ec29d636bd94070d083
Author: ooopSok <81...@users.noreply.github.com>
AuthorDate: Sat Aug 29 17:47:53 2026 +0800
refactor(storage): retire unread workflow projection tables (#3937)
Remove unread Task Ledger and Plan projection tables so their event ledgers remain the sole durable authority.
Advance workflow schema 9 to 10 and fail closed unless released projection shapes are recognized before atomic retirement. Preserve event-only migration, backup/restore, and older-reader rejection behavior.
Generated-by: Codex
---
.../src/__tests__/sqlite-workflow-store.test.ts | 462 ++++++++++++++++++++-
.../storage/src/__tests__/work-board-store.test.ts | 15 +-
packages/storage/src/operational-target-schema.ts | 2 +-
packages/storage/src/plan-store.ts | 28 +-
packages/storage/src/sqlite-workflow-schema.ts | 81 +++-
packages/storage/src/task-ledger-store.ts | 37 +-
6 files changed, 546 insertions(+), 79 deletions(-)
diff --git a/packages/storage/src/__tests__/sqlite-workflow-store.test.ts b/packages/storage/src/__tests__/sqlite-workflow-store.test.ts
index dacb94261..23fb82b95 100644
--- a/packages/storage/src/__tests__/sqlite-workflow-store.test.ts
+++ b/packages/storage/src/__tests__/sqlite-workflow-store.test.ts
@@ -18,15 +18,20 @@
*/
import assert from 'node:assert/strict';
-import { mkdtemp, rm } from 'node:fs/promises';
+import { mkdir, mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { after, describe, test } from 'node:test';
import { DatabaseSync } from 'node:sqlite';
import { createSqliteDeepResearchStore } from '../deep-research-store.js';
+import {
+ createOperationalStateBackup,
+ restoreOperationalStateBackup,
+} from '../operational-state-backup.js';
import { openInteractiveScheduledTaskStoreForWrite } from '../scheduled-task-store.js';
import { createSqlitePlanStore } from '../plan-store.js';
import { createSqliteTaskLedgerStore } from '../task-ledger-store.js';
+import { SQLITE_WORKFLOW_SCHEMA_VERSION } from '../sqlite-workflow-schema.js';
import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '../root-authority.js';
import {
removeTrackedControlDirectories,
@@ -40,7 +45,7 @@ after(removeTrackedControlDirectories);
const SESSION_ID = 'session-workflow';
describe('SQLite workflow stores', () => {
- test('persists Task Ledger events and projections', async () => {
+ test('persists Task Ledger exclusively through events', async () => {
await withRoot(async (root) => {
const store = createSqliteTaskLedgerStore(root);
const { created } = await store.create(SESSION_ID, [{ subject: 'Implement SQLite' }]);
@@ -53,10 +58,18 @@ describe('SQLite workflow stores', () => {
} finally {
reopened.close();
}
+
+ const database = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(rowCount(database, 'workflow_task_ledger_events'), 1);
+ assert.equal(tableExists(database, 'workflow_task_ledger_projections'), false);
+ } finally {
+ database.close();
+ }
});
});
- test('persists Plan events and their projection', async () => {
+ test('persists Plan exclusively through events', async () => {
await withRoot(async (root) => {
const store = createSqlitePlanStore(root, {
newId: (() => {
@@ -82,9 +95,370 @@ describe('SQLite workflow stores', () => {
} finally {
reopened.close();
}
+
+ const database = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(rowCount(database, 'workflow_plan_events'), 1);
+ assert.equal(tableExists(database, 'workflow_plan_projections'), false);
+ } finally {
+ database.close();
+ }
+ });
+ });
+
+ test('migrates released workflow schema 9 projections to event-only schema 10', async () => {
+ await withRoot(async (root) => {
+ const taskStore = createSqliteTaskLedgerStore(root);
+ await taskStore.create(SESSION_ID, [{ subject: 'Preserve event authority' }]);
+ taskStore.close();
+
+ const planStore = createSqlitePlanStore(root, { newId: () => 'proposal-1', now: () => 100 });
+ const submitted = await planStore.submitProposal({
+ sessionId: SESSION_ID,
+ turnId: 'turn-1',
+ title: 'Preserve Plan events',
+ steps: [{ id: 'one', title: 'Replay', description: 'Ignore stale projection bytes' }],
+ });
+ planStore.close();
+
+ const released = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ installReleasedProjectionTables(released);
+ released
+ .prepare(
+ 'INSERT INTO workflow_task_ledger_projections(session_id, record_json) VALUES (?, ?)',
+ )
+ .run(SESSION_ID, '{not-json');
+ released
+ .prepare(
+ 'INSERT INTO workflow_plan_projections(session_id, store_version, record_json) VALUES (?, ?, ?)',
+ )
+ .run(SESSION_ID, 999, '{not-json');
+ setWorkflowSchemaVersion(released, 9);
+ } finally {
+ released.close();
+ }
+
+ const migratedTasks = createSqliteTaskLedgerStore(root);
+ try {
+ assert.equal(
+ (await migratedTasks.list(SESSION_ID))[0]?.subject,
+ 'Preserve event authority',
+ );
+ } finally {
+ migratedTasks.close();
+ }
+ const migratedPlan = createSqlitePlanStore(root);
+ try {
+ assert.equal(
+ (await migratedPlan.readState(SESSION_ID)).latestProposalId,
+ submitted.state.latestProposalId,
+ );
+ } finally {
+ migratedPlan.close();
+ }
+
+ const verified = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(workflowSchemaVersion(verified), SQLITE_WORKFLOW_SCHEMA_VERSION);
+ assert.equal(tableExists(verified, 'workflow_task_ledger_projections'), false);
+ assert.equal(tableExists(verified, 'workflow_plan_projections'), false);
+ assert.equal(rowCount(verified, 'workflow_task_ledger_events'), 1);
+ assert.equal(rowCount(verified, 'workflow_plan_events'), 1);
+ } finally {
+ verified.close();
+ }
+ });
+ });
+
+ test('preserves every released projection when one table has unfamiliar DDL', async () => {
+ await withRoot(async (root) => {
+ createSqliteTaskLedgerStore(root).close();
+ const released = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ installReleasedProjectionTables(released, { planVersionFloor: -1 });
+ released
+ .prepare(
+ 'INSERT INTO workflow_task_ledger_projections(session_id, record_json) VALUES (?, ?)',
+ )
+ .run(SESSION_ID, 'task-sentinel');
+ released
+ .prepare(
+ 'INSERT INTO workflow_plan_projections(session_id, store_version, record_json) VALUES (?, ?, ?)',
+ )
+ .run(SESSION_ID, 0, 'plan-sentinel');
+ setWorkflowSchemaVersion(released, 9);
+ } finally {
+ released.close();
+ }
+
+ assert.throws(
+ () => createSqliteTaskLedgerStore(root),
+ (error: unknown) =>
+ error instanceof Error &&
+ (error as { code?: unknown }).code === 'operational_state_migration_blocked' &&
+ /unfamiliar released shape/u.test(error.message),
+ );
+
+ assertReleasedProjectionStatePreserved(root);
+ });
+ });
+
+ test('preserves every released projection when one table carries an extra trigger', async () => {
+ await withRoot(async (root) => {
+ createSqliteTaskLedgerStore(root).close();
+ const released = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ installReleasedProjectionTables(released);
+ released.exec(`
+ CREATE TRIGGER workflow_task_ledger_projection_guard
+ AFTER INSERT ON workflow_task_ledger_projections
+ BEGIN
+ SELECT 1;
+ END;
+ `);
+ released
+ .prepare(
+ 'INSERT INTO workflow_task_ledger_projections(session_id, record_json) VALUES (?, ?)',
+ )
+ .run(SESSION_ID, 'task-sentinel');
+ released
+ .prepare(
+ 'INSERT INTO workflow_plan_projections(session_id, store_version, record_json) VALUES (?, ?, ?)',
+ )
+ .run(SESSION_ID, 0, 'plan-sentinel');
+ setWorkflowSchemaVersion(released, 9);
+ } finally {
+ released.close();
+ }
+
+ assert.throws(
+ () => createSqliteTaskLedgerStore(root),
+ (error: unknown) =>
+ error instanceof Error &&
+ (error as { code?: unknown }).code === 'operational_state_migration_blocked' &&
+ /unexpected object/u.test(error.message),
+ );
+
+ assertReleasedProjectionStatePreserved(root);
+ });
+ });
+
+ test('preserves an unfamiliar projection whose table name differs only by case', async () => {
+ await withRoot(async (root) => {
+ createSqliteTaskLedgerStore(root).close();
+ const released = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ installReleasedProjectionTables(released, {
+ taskTableName: 'WORKFLOW_TASK_LEDGER_PROJECTIONS',
+ });
+ released
+ .prepare(
+ 'INSERT INTO WORKFLOW_TASK_LEDGER_PROJECTIONS(session_id, record_json) VALUES (?, ?)',
+ )
+ .run(SESSION_ID, 'task-sentinel');
+ released
+ .prepare(
+ 'INSERT INTO workflow_plan_projections(session_id, store_version, record_json) VALUES (?, ?, ?)',
+ )
+ .run(SESSION_ID, 0, 'plan-sentinel');
+ setWorkflowSchemaVersion(released, 9);
+ } finally {
+ released.close();
+ }
+
+ assert.throws(
+ () => createSqliteTaskLedgerStore(root),
+ (error: unknown) =>
+ error instanceof Error &&
+ (error as { code?: unknown }).code === 'operational_state_migration_blocked' &&
+ /unfamiliar released shape/u.test(error.message),
+ );
+
+ assertReleasedProjectionStatePreserved(root);
+ });
+ });
+
+ test('preserves released projections with a sqliteX-prefixed trigger', async () => {
+ await withRoot(async (root) => {
+ createSqliteTaskLedgerStore(root).close();
+ const released = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ installReleasedProjectionTables(released);
+ released.exec(`
+ CREATE TRIGGER sqliteX_projection_guard
+ AFTER INSERT ON workflow_task_ledger_projections
+ BEGIN
+ SELECT 1;
+ END;
+ `);
+ released
+ .prepare(
+ 'INSERT INTO workflow_task_ledger_projections(session_id, record_json) VALUES (?, ?)',
+ )
+ .run(SESSION_ID, 'task-sentinel');
+ released
+ .prepare(
+ 'INSERT INTO workflow_plan_projections(session_id, store_version, record_json) VALUES (?, ?, ?)',
+ )
+ .run(SESSION_ID, 0, 'plan-sentinel');
+ setWorkflowSchemaVersion(released, 9);
+ } finally {
+ released.close();
+ }
+
+ assert.throws(
+ () => createSqliteTaskLedgerStore(root),
+ (error: unknown) =>
+ error instanceof Error &&
+ (error as { code?: unknown }).code === 'operational_state_migration_blocked' &&
+ /unexpected object/u.test(error.message),
+ );
+
+ assertReleasedProjectionStatePreserved(root);
+ });
+ });
+
+ test('preserves released projections with a sqliteX-prefixed dependent view', async () => {
+ await withRoot(async (root) => {
+ createSqliteTaskLedgerStore(root).close();
+ const released = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ installReleasedProjectionTables(released);
+ released.exec(`
+ CREATE VIEW sqliteX_projection_guard AS
+ SELECT session_id, record_json
+ FROM workflow_task_ledger_projections;
+ `);
+ released
+ .prepare(
+ 'INSERT INTO workflow_task_ledger_projections(session_id, record_json) VALUES (?, ?)',
+ )
+ .run(SESSION_ID, 'task-sentinel');
+ released
+ .prepare(
+ 'INSERT INTO workflow_plan_projections(session_id, store_version, record_json) VALUES (?, ?, ?)',
+ )
+ .run(SESSION_ID, 0, 'plan-sentinel');
+ setWorkflowSchemaVersion(released, 9);
+ } finally {
+ released.close();
+ }
+
+ assert.throws(
+ () => createSqliteTaskLedgerStore(root),
+ (error: unknown) =>
+ error instanceof Error &&
+ (error as { code?: unknown }).code === 'operational_state_migration_blocked' &&
+ /unexpected schema object view:sqliteX_projection_guard/u.test(error.message),
+ );
+
+ assertReleasedProjectionStatePreserved(root);
+ const preserved = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(tableExists(preserved, 'sqliteX_projection_guard', 'view'), true);
+ } finally {
+ preserved.close();
+ }
+ });
+ });
+
+ test('an older workflow reader rejects the newer schema without changing it', async () => {
+ await withRoot(async (root) => {
+ const store = createSqliteTaskLedgerStore(root);
+ await store.create(SESSION_ID, [{ subject: 'Preserve newer workflow state' }]);
+ store.close();
+
+ const newer = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ setWorkflowSchemaVersion(newer, SQLITE_WORKFLOW_SCHEMA_VERSION + 1);
+ newer.exec(`
+ CREATE TABLE workflow_future_sentinel (value TEXT NOT NULL);
+ INSERT INTO workflow_future_sentinel(value) VALUES ('preserved');
+ `);
+ } finally {
+ newer.close();
+ }
+
+ assert.throws(
+ () => createSqliteTaskLedgerStore(root),
+ /Operational schema workflow is newer than supported/u,
+ );
+
+ const preserved = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(workflowSchemaVersion(preserved), SQLITE_WORKFLOW_SCHEMA_VERSION + 1);
+ assert.equal(rowCount(preserved, 'workflow_task_ledger_events'), 1);
+ assert.equal(
+ (
+ preserved.prepare('SELECT value FROM workflow_future_sentinel').get() as {
+ value?: unknown;
+ }
+ ).value,
+ 'preserved',
+ );
+ } finally {
+ preserved.close();
+ }
});
});
+ test('backs up and restores event-only Task Ledger and Plan state', async () => {
+ const base = await mkdtemp(join(tmpdir(), 'maka-workflow-backup-'));
+ const stateRoot = join(base, 'state');
+ const backupRoot = join(base, 'backup');
+ const restoreRoot = join(base, 'restore');
+ await mkdir(stateRoot);
+ try {
+ const taskStore = createSqliteTaskLedgerStore(stateRoot);
+ await taskStore.create(SESSION_ID, [{ subject: 'Restore Task event' }]);
+ taskStore.close();
+
+ const planStore = createSqlitePlanStore(stateRoot, {
+ newId: () => 'backup-proposal',
+ now: () => 100,
+ });
+ const submitted = await planStore.submitProposal({
+ sessionId: SESSION_ID,
+ turnId: 'turn-backup',
+ title: 'Restore Plan event',
+ steps: [{ id: 'restore', title: 'Restore', description: 'Replay the event ledger' }],
+ });
+ planStore.close();
+
+ await createOperationalStateBackup({ stateRoot, destinationRoot: backupRoot, now: () => 10 });
+ await restoreOperationalStateBackup({ backupRoot, destinationRoot: restoreRoot });
+
+ const restoredTasks = createSqliteTaskLedgerStore(restoreRoot);
+ try {
+ assert.equal((await restoredTasks.list(SESSION_ID))[0]?.subject, 'Restore Task event');
+ } finally {
+ restoredTasks.close();
+ }
+ const restoredPlan = createSqlitePlanStore(restoreRoot);
+ try {
+ assert.equal(
+ (await restoredPlan.readState(SESSION_ID)).latestProposalId,
+ submitted.state.latestProposalId,
+ );
+ } finally {
+ restoredPlan.close();
+ }
+
+ const restored = new DatabaseSync(join(restoreRoot, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(tableExists(restored, 'workflow_task_ledger_projections'), false);
+ assert.equal(tableExists(restored, 'workflow_plan_projections'), false);
+ assert.equal(rowCount(restored, 'workflow_task_ledger_events'), 1);
+ assert.equal(rowCount(restored, 'workflow_plan_events'), 1);
+ } finally {
+ restored.close();
+ }
+ } finally {
+ await rm(base, { recursive: true, force: true });
+ }
+ });
+
test('reconciles exact Plan retries through durable operation identity', async () => {
await withRoot(async (root) => {
const store = createSqlitePlanStore(root, {
@@ -256,7 +630,27 @@ describe('SQLite workflow stores', () => {
});
});
- test('purges Plan events and projections for retired Sessions', async () => {
+ test('purges Task Ledger events for retired Sessions', async () => {
+ await withRoot(async (root) => {
+ const store = createSqliteTaskLedgerStore(root);
+ try {
+ await store.create(SESSION_ID, [{ subject: 'Disposable task' }]);
+ await store.purgeConversationTaskLedger(SESSION_ID);
+ assert.deepEqual(await store.list(SESSION_ID), []);
+ } finally {
+ store.close();
+ }
+
+ const database = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(rowCount(database, 'workflow_task_ledger_events'), 0);
+ } finally {
+ database.close();
+ }
+ });
+ });
+
+ test('purges Plan events for retired Sessions', async () => {
await withRoot(async (root) => {
const store = createSqlitePlanStore(root);
try {
@@ -543,3 +937,63 @@ async function withRoot(run: (root: string) => Promise<void>): Promise<void> {
await rm(root, { recursive: true, force: true });
}
}
+
+function tableExists(database: DatabaseSync, name: string, type = 'table'): boolean {
+ return (
+ database
+ .prepare('SELECT 1 AS present FROM sqlite_schema WHERE type = ? AND name = ?')
+ .get(type, name) !== undefined
+ );
+}
+
+function rowCount(database: DatabaseSync, table: string): number {
+ const row = database.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get() as {
+ count?: unknown;
+ };
+ assert.equal(typeof row.count, 'number');
+ return row.count as number;
+}
+
+function installReleasedProjectionTables(
+ database: DatabaseSync,
+ options: { planVersionFloor?: number; taskTableName?: string } = {},
+): void {
+ const taskTableName = options.taskTableName ?? 'workflow_task_ledger_projections';
+ database.exec(`
+ CREATE TABLE ${taskTableName} (
+ session_id TEXT PRIMARY KEY,
+ record_json TEXT NOT NULL
+ );
+ CREATE TABLE workflow_plan_projections (
+ session_id TEXT PRIMARY KEY,
+ store_version INTEGER NOT NULL CHECK (store_version >= ${options.planVersionFloor ?? 0}),
+ record_json TEXT NOT NULL
+ );
+ `);
+}
+
+function setWorkflowSchemaVersion(database: DatabaseSync, version: number): void {
+ database
+ .prepare("UPDATE operational_schema_migrations SET version = ? WHERE scope = 'workflow'")
+ .run(version);
+}
+
+function workflowSchemaVersion(database: DatabaseSync): number {
+ const row = database
+ .prepare("SELECT version FROM operational_schema_migrations WHERE scope = 'workflow'")
+ .get() as { version?: unknown } | undefined;
+ const version = row?.version;
+ assert.equal(typeof version, 'number');
+ return version as number;
+}
+
+function assertReleasedProjectionStatePreserved(root: string): void {
+ const preserved = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(workflowSchemaVersion(preserved), 9);
+ assert.equal(rowCount(preserved, 'workflow_task_ledger_projections'), 1);
+ assert.equal(rowCount(preserved, 'workflow_plan_projections'), 1);
+ } finally {
+ preserved.close();
+ }
+}
diff --git a/packages/storage/src/__tests__/work-board-store.test.ts b/packages/storage/src/__tests__/work-board-store.test.ts
index 8b58318bf..f24d9f6c9 100644
--- a/packages/storage/src/__tests__/work-board-store.test.ts
+++ b/packages/storage/src/__tests__/work-board-store.test.ts
@@ -39,6 +39,7 @@ import {
WORK_BOARD_DEFAULT_PAGE_SIZE,
WORK_BOARD_PROJECT_ID_MAX_CHARS,
} from '@maka/core/work-board';
+import { SQLITE_WORKFLOW_SCHEMA_VERSION } from '../sqlite-workflow-schema.js';
describe('Work Board store', () => {
test('persists items across reopen', async () => {
@@ -664,7 +665,7 @@ describe('Work Board store', () => {
});
});
- test('migrates a real workflow schema 8 database to version 9 with the board table', async () => {
+ test('migrates a real workflow schema 8 database through event-only version 10', async () => {
const base = await mkdtemp(join(tmpdir(), 'maka-work-board-schema8-'));
const stateRoot = join(base, 'state');
const databasePath = join(stateRoot, 'runtime.sqlite');
@@ -683,7 +684,7 @@ describe('Work Board store', () => {
),
);
// The released v0.1.6 fixture ships an Automation definition; clearing
- // it keeps the fixture focused on the workflow 8 -> 9 upgrade while
+ // it keeps the fixture focused on the workflow 8 -> current upgrade while
// leaving every released table and row otherwise intact.
v8.exec('DELETE FROM automation_definitions; DELETE FROM automation_pending_fires;');
v8.prepare(
@@ -699,12 +700,20 @@ describe('Work Board store', () => {
const version = database
.prepare("SELECT version FROM operational_schema_migrations WHERE scope = 'workflow'")
.get() as { version?: unknown } | undefined;
- assert.equal(version?.version, 9);
+ assert.equal(version?.version, SQLITE_WORKFLOW_SCHEMA_VERSION);
assert.ok(
database
.prepare("SELECT 1 FROM sqlite_schema WHERE name = 'workflow_work_board_items'")
.get(),
);
+ assert.equal(
+ database
+ .prepare(
+ "SELECT 1 FROM sqlite_schema WHERE name IN ('workflow_task_ledger_projections', 'workflow_plan_projections') LIMIT 1",
+ )
+ .get(),
+ undefined,
+ );
assert.ok(
database
.prepare("SELECT 1 FROM sqlite_schema WHERE name = 'workflow_goal_authority'")
diff --git a/packages/storage/src/operational-target-schema.ts b/packages/storage/src/operational-target-schema.ts
index b4434e21a..244850d36 100644
--- a/packages/storage/src/operational-target-schema.ts
+++ b/packages/storage/src/operational-target-schema.ts
@@ -90,7 +90,7 @@ function readSchemaObjects(
.prepare(`
SELECT type, name, tbl_name, sql
FROM sqlite_schema
- WHERE name NOT LIKE 'sqlite_%' AND type IN ('table', 'index', 'trigger', 'view') AND sql IS NOT NULL
+ WHERE name NOT GLOB 'sqlite_*' AND type IN ('table', 'index', 'trigger', 'view') AND sql IS NOT NULL
ORDER BY type, name
`)
.all() as Array<{ type: string; name: string; tbl_name: string; sql: string }>;
diff --git a/packages/storage/src/plan-store.ts b/packages/storage/src/plan-store.ts
index 1ba8c2956..8ed7a4e9b 100644
--- a/packages/storage/src/plan-store.ts
+++ b/packages/storage/src/plan-store.ts
@@ -436,7 +436,7 @@ class SqlitePlanStoreImpl implements SqlitePlanStore {
if (fingerprint) event.operationFingerprint = fingerprint;
const state = applyPlanEvent(ledger.state, event);
assertPlanProjectionWithinLimit(state);
- await this.appendCanonicalEvent(sessionId, event, state);
+ await this.appendCanonicalEvent(event);
result = { event, state };
});
return result;
@@ -449,9 +449,6 @@ class SqlitePlanStoreImpl implements SqlitePlanStore {
this.#lease.database
.prepare('DELETE FROM workflow_plan_events WHERE session_id = ?')
.run(sessionId);
- this.#lease.database
- .prepare('DELETE FROM workflow_plan_projections WHERE session_id = ?')
- .run(sessionId);
});
});
}
@@ -463,14 +460,9 @@ class SqlitePlanStoreImpl implements SqlitePlanStore {
return readSqlitePlanLedger(this.#lease.database, sessionId);
}
- private async appendCanonicalEvent(
- sessionId: string,
- event: PlanEvent,
- state: PlanSessionState,
- ): Promise<void> {
+ private async appendCanonicalEvent(event: PlanEvent): Promise<void> {
this.#lease.transaction('write', () => {
insertPlanEvent(this.#lease.database, event);
- writePlanProjection(this.#lease.database, sessionId, state);
});
}
}
@@ -570,22 +562,6 @@ function insertPlanEvent(database: DatabaseSync, event: PlanEvent): void {
.run(event.sessionId, row.sequence, event.id, event.storeVersion, JSON.stringify(event));
}
-function writePlanProjection(
- database: DatabaseSync,
- sessionId: string,
- state: PlanSessionState,
-): void {
- database
- .prepare(`
- INSERT INTO workflow_plan_projections(session_id, store_version, record_json)
- VALUES (?, ?, ?)
- ON CONFLICT(session_id) DO UPDATE SET
- store_version = excluded.store_version,
- record_json = excluded.record_json
- `)
- .run(sessionId, state.storeVersion, JSON.stringify(state));
-}
-
export function applyPlanEvent(state: PlanSessionState, event: PlanEvent): PlanSessionState {
if (event.sessionId !== state.sessionId) throw new Error('Plan event session mismatch');
if (event.storeVersion !== state.storeVersion + 1) {
diff --git a/packages/storage/src/sqlite-workflow-schema.ts b/packages/storage/src/sqlite-workflow-schema.ts
index 4ddb48330..ab3486f07 100644
--- a/packages/storage/src/sqlite-workflow-schema.ts
+++ b/packages/storage/src/sqlite-workflow-schema.ts
@@ -19,9 +19,28 @@
import type { DatabaseSync } from 'node:sqlite';
-export const SQLITE_WORKFLOW_SCHEMA_VERSION = 9;
+export const SQLITE_WORKFLOW_SCHEMA_VERSION = 10;
+
+const RELEASED_WORKFLOW_PROJECTION_TABLES = [
+ {
+ name: 'workflow_task_ledger_projections',
+ sql: `CREATE TABLE workflow_task_ledger_projections (
+ session_id TEXT PRIMARY KEY,
+ record_json TEXT NOT NULL
+ )`,
+ },
+ {
+ name: 'workflow_plan_projections',
+ sql: `CREATE TABLE workflow_plan_projections (
+ session_id TEXT PRIMARY KEY,
+ store_version INTEGER NOT NULL CHECK (store_version >= 0),
+ record_json TEXT NOT NULL
+ )`,
+ },
+] as const;
export function migrateSqliteWorkflowDatabase(db: DatabaseSync): void {
+ retireReleasedWorkflowProjections(db);
db.exec(`
DROP INDEX IF EXISTS workflow_plan_reminders_order;
DROP TABLE IF EXISTS workflow_plan_reminders;
@@ -35,11 +54,6 @@ export function migrateSqliteWorkflowDatabase(db: DatabaseSync): void {
UNIQUE (session_id, event_id)
);
- CREATE TABLE IF NOT EXISTS workflow_task_ledger_projections (
- session_id TEXT PRIMARY KEY,
- record_json TEXT NOT NULL
- );
-
CREATE TABLE IF NOT EXISTS workflow_plan_events (
session_id TEXT NOT NULL,
sequence INTEGER NOT NULL CHECK (sequence >= 0),
@@ -51,12 +65,6 @@ export function migrateSqliteWorkflowDatabase(db: DatabaseSync): void {
UNIQUE (session_id, store_version)
);
- CREATE TABLE IF NOT EXISTS workflow_plan_projections (
- session_id TEXT PRIMARY KEY,
- store_version INTEGER NOT NULL CHECK (store_version >= 0),
- record_json TEXT NOT NULL
- );
-
CREATE TABLE IF NOT EXISTS workflow_deep_research_events (
session_id TEXT NOT NULL,
sequence INTEGER NOT NULL CHECK (sequence >= 0),
@@ -174,3 +182,52 @@ export function migrateSqliteWorkflowDatabase(db: DatabaseSync): void {
WHERE record_json IS NULL
`).run();
}
+
+function retireReleasedWorkflowProjections(db: DatabaseSync): void {
+ for (const table of RELEASED_WORKFLOW_PROJECTION_TABLES) {
+ assertReleasedWorkflowProjectionShape(db, table);
+ }
+ db.exec(`
+ DROP TABLE IF EXISTS workflow_task_ledger_projections;
+ DROP TABLE IF EXISTS workflow_plan_projections;
+ `);
+}
+
+function assertReleasedWorkflowProjectionShape(
+ db: DatabaseSync,
+ table: (typeof RELEASED_WORKFLOW_PROJECTION_TABLES)[number],
+): void {
+ const objects = db
+ .prepare(`
+ SELECT type, name, sql
+ FROM sqlite_schema
+ WHERE tbl_name COLLATE NOCASE = ?
+ AND type IN ('table', 'index', 'trigger', 'view')
+ AND sql IS NOT NULL
+ ORDER BY type, name
+ `)
+ .all(table.name) as Array<{ type: string; name: string; sql: string }>;
+ if (objects.length === 0) return;
+
+ const releasedTable = objects.find(
+ (object) => object.type === 'table' && object.name === table.name,
+ );
+ if (!releasedTable || normalizeSql(releasedTable.sql) !== normalizeSql(table.sql)) {
+ throw new Error(`Workflow projection table ${table.name} has an unfamiliar released shape`);
+ }
+ const unexpected = objects.find((object) => object !== releasedTable);
+ if (unexpected) {
+ throw new Error(
+ `Workflow projection table ${table.name} carries an unexpected object ` +
+ `${unexpected.type}:${unexpected.name}`,
+ );
+ }
+}
+
+function normalizeSql(value: string): string {
+ return value
+ .replace(/\s+/gu, ' ')
+ .replace(/\s*([(),;])\s*/gu, '$1')
+ .trim()
+ .toUpperCase();
+}
diff --git a/packages/storage/src/task-ledger-store.ts b/packages/storage/src/task-ledger-store.ts
index ceb62d1df..0b2c6a485 100644
--- a/packages/storage/src/task-ledger-store.ts
+++ b/packages/storage/src/task-ledger-store.ts
@@ -193,7 +193,7 @@ class SqliteTaskLedgerStoreImpl implements SqliteTaskLedgerStore {
}
await chainWrite(this.writeQueues, input.targetSessionId, async () => {
- this.copyConversationLedger(input.targetSessionId, selected, projection.tasks);
+ this.copyConversationLedger(input.targetSessionId, selected);
});
}
@@ -204,9 +204,6 @@ class SqliteTaskLedgerStoreImpl implements SqliteTaskLedgerStore {
this.#lease.database
.prepare('DELETE FROM workflow_task_ledger_events WHERE session_id = ?')
.run(sessionId);
- this.#lease.database
- .prepare('DELETE FROM workflow_task_ledger_projections WHERE session_id = ?')
- .run(sessionId);
});
});
}
@@ -601,7 +598,6 @@ class SqliteTaskLedgerStoreImpl implements SqliteTaskLedgerStore {
taskIds: [...new Set(mutationEvents.map((event) => event.taskId))],
at: Date.now(),
});
- await this.write(sessionId, next);
});
return next;
}
@@ -613,30 +609,15 @@ class SqliteTaskLedgerStoreImpl implements SqliteTaskLedgerStore {
});
}
- private async write(sessionId: string, tasks: Task[]): Promise<void> {
- this.#lease.transaction('write', () => {
- writeTaskLedgerProjection(this.#lease.database, sessionId, tasks);
- });
- }
-
- private copyConversationLedger(
- sessionId: string,
- events: readonly TaskLedgerEvent[],
- tasks: readonly Task[],
- ): void {
+ private copyConversationLedger(sessionId: string, events: readonly TaskLedgerEvent[]): void {
this.#lease.transaction('write', () => {
const existing = this.#lease.database
- .prepare(`
- SELECT
- (SELECT COUNT(*) FROM workflow_task_ledger_events WHERE session_id = ?) +
- (SELECT COUNT(*) FROM workflow_task_ledger_projections WHERE session_id = ?) AS count
- `)
- .get(sessionId, sessionId) as { count?: unknown };
+ .prepare('SELECT COUNT(*) AS count FROM workflow_task_ledger_events WHERE session_id = ?')
+ .get(sessionId) as { count?: unknown };
if (existing.count !== 0) {
throw new Error('Task Ledger conversation-copy target already exists');
}
for (const event of events) insertTaskLedgerEvent(this.#lease.database, sessionId, event);
- writeTaskLedgerProjection(this.#lease.database, sessionId, [...tasks]);
});
}
@@ -718,16 +699,6 @@ function insertTaskLedgerEvent(
.run(sessionId, row.sequence, event.eventId, JSON.stringify(event));
}
-function writeTaskLedgerProjection(database: DatabaseSync, sessionId: string, tasks: Task[]): void {
- database
- .prepare(`
- INSERT INTO workflow_task_ledger_projections(session_id, record_json)
- VALUES (?, ?)
- ON CONFLICT(session_id) DO UPDATE SET record_json = excluded.record_json
- `)
- .run(sessionId, JSON.stringify(tasks));
-}
-
function nextTaskKey(tasks: readonly Task[], parent: Task | undefined): string {
const siblings = tasks.filter((task) => task.parentId === parent?.id);
const prefix = parent ? `${parent.key}.` : 'T';
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org