You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@maka.apache.org by zh...@apache.org on 2026/08/31 14:11:00 UTC
(maka) branch main updated: feat(storage): add dormant SessionTodo foundation (#4340)
This is an automated email from the ASF dual-hosted git repository.
me2seeks 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 d0e238ca09 feat(storage): add dormant SessionTodo foundation (#4340)
d0e238ca09 is described below
commit d0e238ca09eafe4f99289f57902ec0b47a7d8097
Author: Zhang <96...@users.noreply.github.com>
AuthorDate: Mon Aug 31 22:10:55 2026 +0800
feat(storage): add dormant SessionTodo foundation (#4340)
* feat(storage): add dormant SessionTodo foundation
Add the bounded current-document contract and an internal SQLite authority with one-time active Task bootstrap. Persist initialized-empty state, keep explicit whole replacements independent from legacy replay, and cover schema migration plus backup recovery without exposing the feature to Runtime Host or surfaces yet.\n\nRefs #4338\n\nGenerated-by: OpenAI Codex
* fix(storage): preserve blocked tasks during todo bootstrap
Map legacy blocked Tasks to pending SessionTodo items while preserving the subject and omitting workflow-only metadata. Cover both production replay and the schema 10 to 11 migration path.
Refs #4338
Generated-by: OpenAI Codex
---
packages/core/package.json | 1 +
packages/core/src/__tests__/session-todo.test.ts | 82 +++++++
packages/core/src/session-todo.ts | 106 +++++++++
.../src/__tests__/session-todo-authority.test.ts | 97 ++++++++
.../src/__tests__/session-todo-store.test.ts | 258 +++++++++++++++++++++
.../src/__tests__/sqlite-workflow-store.test.ts | 62 ++++-
packages/storage/src/session-todo-authority.ts | 131 +++++++++++
packages/storage/src/session-todo-store.ts | 244 +++++++++++++++++++
packages/storage/src/sqlite-workflow-schema.ts | 7 +-
9 files changed, 985 insertions(+), 3 deletions(-)
diff --git a/packages/core/package.json b/packages/core/package.json
index 6584b561b9..9de20f57e8 100644
--- a/packages/core/package.json
+++ b/packages/core/package.json
@@ -69,6 +69,7 @@
"./agent-graph-client-projection": "./dist/agent-graph-client-projection.js",
"./agent-graph-supervisor-wake": "./dist/agent-graph-supervisor-wake.js",
"./task-ledger": "./dist/task-ledger.js",
+ "./session-todo": "./dist/session-todo.js",
"./foreign-session": "./dist/foreign-session.js",
"./external-session": "./dist/external-session.js",
"./deep-research-run": "./dist/deep-research-run.js",
diff --git a/packages/core/src/__tests__/session-todo.test.ts b/packages/core/src/__tests__/session-todo.test.ts
new file mode 100644
index 0000000000..cbc5900ce1
--- /dev/null
+++ b/packages/core/src/__tests__/session-todo.test.ts
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { describe, test } from 'node:test';
+import {
+ SESSION_TODO_CONTENT_MAX_CHARS,
+ SESSION_TODO_MAX_ITEMS,
+ normalizeSessionTodoItems,
+} from '../session-todo.js';
+
+describe('SessionTodo document', () => {
+ test('normalizes the complete ordered list and permits initialized-empty state', () => {
+ assert.deepEqual(normalizeSessionTodoItems([]), { ok: true, value: { items: [] } });
+ assert.deepEqual(
+ normalizeSessionTodoItems([
+ { content: ' inspect\n the owner ', status: 'pending' },
+ { content: 'ship', status: 'completed' },
+ ]),
+ {
+ ok: true,
+ value: {
+ items: [
+ { content: 'inspect the owner', status: 'pending' },
+ { content: 'ship', status: 'completed' },
+ ],
+ },
+ },
+ );
+ });
+
+ test('enforces the legacy-compatible item and content bounds', () => {
+ assert.equal(
+ normalizeSessionTodoItems(
+ Array.from({ length: SESSION_TODO_MAX_ITEMS }, () => ({
+ content: 'x'.repeat(SESSION_TODO_CONTENT_MAX_CHARS),
+ status: 'pending',
+ })),
+ ).ok,
+ true,
+ );
+ assert.equal(
+ normalizeSessionTodoItems(
+ Array.from({ length: SESSION_TODO_MAX_ITEMS + 1 }, () => ({
+ content: 'x',
+ status: 'pending',
+ })),
+ ).ok,
+ false,
+ );
+ assert.equal(
+ normalizeSessionTodoItems([
+ { content: '😀'.repeat(SESSION_TODO_CONTENT_MAX_CHARS + 1), status: 'pending' },
+ ]).ok,
+ false,
+ );
+ });
+
+ test('rejects invalid statuses and unknown fields', () => {
+ assert.equal(normalizeSessionTodoItems([{ content: 'x', status: 'blocked' }]).ok, false);
+ assert.equal(
+ normalizeSessionTodoItems([{ content: 'x', status: 'pending', revision: 1 }]).ok,
+ false,
+ );
+ });
+});
diff --git a/packages/core/src/session-todo.ts b/packages/core/src/session-todo.ts
new file mode 100644
index 0000000000..0a2176c551
--- /dev/null
+++ b/packages/core/src/session-todo.ts
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { serializedByteLength } from './serialized-byte-length.js';
+
+/**
+ * SessionTodo is a current-state document, not an event ledger. Its bounds
+ * deliberately preserve every valid legacy Task subject/count while keeping a
+ * full document comfortably below the Runtime Host message ceiling.
+ */
+export const SESSION_TODO_CONTENT_MAX_CHARS = 200;
+export const SESSION_TODO_MAX_ITEMS = 200;
+export const SESSION_TODO_DOCUMENT_MAX_BYTES = 256 * 1024;
+
+export const SESSION_TODO_STATUSES = ['pending', 'in_progress', 'completed'] as const;
+export type SessionTodoStatus = (typeof SESSION_TODO_STATUSES)[number];
+
+export interface SessionTodoItem {
+ content: string;
+ status: SessionTodoStatus;
+}
+
+export interface SessionTodoSnapshot {
+ items: SessionTodoItem[];
+}
+
+export type SessionTodoNormalizeResult =
+ | { ok: true; value: SessionTodoSnapshot }
+ | { ok: false; message: string };
+
+export function normalizeSessionTodoItems(input: unknown): SessionTodoNormalizeResult {
+ if (!Array.isArray(input)) return invalid('Todo items must be an array');
+ if (input.length > SESSION_TODO_MAX_ITEMS) {
+ return invalid(`Todo list may contain at most ${SESSION_TODO_MAX_ITEMS} items`);
+ }
+
+ const items: SessionTodoItem[] = [];
+ for (const [index, raw] of input.entries()) {
+ if (!isPlainRecord(raw)) {
+ return invalid(`Todo item ${index + 1} must be an object`);
+ }
+ const item = raw;
+ if (!hasExactKeys(item, ['content', 'status'])) {
+ return invalid(`Todo item ${index + 1} must contain only content and status`);
+ }
+ if (typeof item.content !== 'string') {
+ return invalid(`Todo item ${index + 1} content must be a string`);
+ }
+ const content = item.content.normalize('NFC').replace(/\s+/gu, ' ').trim();
+ if (content.length === 0) return invalid(`Todo item ${index + 1} content cannot be empty`);
+ if ([...content].length > SESSION_TODO_CONTENT_MAX_CHARS) {
+ return invalid(
+ `Todo item ${index + 1} content must be ${SESSION_TODO_CONTENT_MAX_CHARS} characters or fewer`,
+ );
+ }
+ if (!isSessionTodoStatus(item.status)) {
+ return invalid(
+ `Todo item ${index + 1} status must be one of ${SESSION_TODO_STATUSES.join(', ')}`,
+ );
+ }
+ items.push({ content, status: item.status });
+ }
+
+ const value = { items };
+ if (
+ serializedByteLength(value, SESSION_TODO_DOCUMENT_MAX_BYTES) > SESSION_TODO_DOCUMENT_MAX_BYTES
+ ) {
+ return invalid(`Todo document exceeds ${SESSION_TODO_DOCUMENT_MAX_BYTES} encoded bytes`);
+ }
+ return { ok: true, value };
+}
+
+export function isSessionTodoStatus(value: unknown): value is SessionTodoStatus {
+ return typeof value === 'string' && (SESSION_TODO_STATUSES as readonly string[]).includes(value);
+}
+
+function invalid(message: string): SessionTodoNormalizeResult {
+ return { ok: false, message };
+}
+
+function isPlainRecord(value: unknown): value is Record<string, unknown> {
+ if (typeof value !== 'object' || value === null || Array.isArray(value)) return false;
+ const prototype = Object.getPrototypeOf(value);
+ return prototype === Object.prototype || prototype === null;
+}
+
+function hasExactKeys(record: Record<string, unknown>, expected: readonly string[]): boolean {
+ const keys = Object.keys(record).sort();
+ return keys.length === expected.length && keys.every((key, index) => key === expected[index]);
+}
diff --git a/packages/storage/src/__tests__/session-todo-authority.test.ts b/packages/storage/src/__tests__/session-todo-authority.test.ts
new file mode 100644
index 0000000000..929a15a097
--- /dev/null
+++ b/packages/storage/src/__tests__/session-todo-authority.test.ts
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { after, describe, test } from 'node:test';
+import {
+ authenticateInteractiveSessionTodoWriter,
+ openInteractiveSessionTodoStoreForWrite,
+ type InteractiveSessionTodoWriter,
+} from '../session-todo-authority.js';
+import {
+ resolveStorageRoot,
+ StorageRootAuthorityError,
+ tryAcquireInteractiveRootOwner,
+ type StorageRootLease,
+} from '../root-authority.js';
+import {
+ removeTrackedControlDirectories,
+ trackControlDirectory,
+} from './fixtures/control-directory-hygiene.js';
+
+after(removeTrackedControlDirectories);
+
+describe('interactive SessionTodo authority', () => {
+ test('single-flights opens and invalidates the facade when closed', async () => {
+ await withInteractiveRoot(async (capability) => {
+ const owner = await tryAcquireInteractiveRootOwner(capability);
+ assert.ok(owner);
+ if (!owner) return;
+ try {
+ const [first, second] = await Promise.all([
+ openInteractiveSessionTodoStoreForWrite(owner.lease),
+ openInteractiveSessionTodoStoreForWrite(owner.lease),
+ ]);
+ assert.equal(first, second);
+ assert.equal(authenticateInteractiveSessionTodoWriter(first), first);
+ await first.replaceAll('authority-session', [{ content: 'owned', status: 'pending' }]);
+ assert.deepEqual(await second.readOrBootstrap('authority-session'), {
+ items: [{ content: 'owned', status: 'pending' }],
+ });
+ first.close();
+ assert.throws(() => authenticateInteractiveSessionTodoWriter(first), isInvalidLease);
+ await assert.rejects(() => first.readOrBootstrap('authority-session'), isInvalidLease);
+ } finally {
+ if (!owner.closed) await owner.close();
+ }
+ });
+ });
+
+ test('rejects forged leases and facades', async () => {
+ await assert.rejects(
+ () => openInteractiveSessionTodoStoreForWrite({} as StorageRootLease<'interactive', 'write'>),
+ isInvalidLease,
+ );
+ assert.throws(
+ () => authenticateInteractiveSessionTodoWriter({} as InteractiveSessionTodoWriter),
+ isInvalidLease,
+ );
+ });
+});
+
+async function withInteractiveRoot(
+ run: (capability: Awaited<ReturnType<typeof resolveStorageRoot<'interactive'>>>) => Promise<void>,
+): Promise<void> {
+ const base = await mkdtemp(join(tmpdir(), 'maka-session-todo-authority-'));
+ try {
+ const capability = trackControlDirectory(
+ await resolveStorageRoot({ path: join(base, 'interactive'), kind: 'interactive' }),
+ );
+ await run(capability);
+ } finally {
+ await rm(base, { recursive: true, force: true });
+ }
+}
+
+function isInvalidLease(error: unknown): boolean {
+ return error instanceof StorageRootAuthorityError && error.code === 'invalid_lease';
+}
diff --git a/packages/storage/src/__tests__/session-todo-store.test.ts b/packages/storage/src/__tests__/session-todo-store.test.ts
new file mode 100644
index 0000000000..a643482453
--- /dev/null
+++ b/packages/storage/src/__tests__/session-todo-store.test.ts
@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { describe, test } from 'node:test';
+import { DatabaseSync } from 'node:sqlite';
+import { createSqliteSessionTodoStore } from '../session-todo-store.js';
+import { createSqliteTaskLedgerStore } from '../task-ledger-store.js';
+
+const SESSION_ID = 'session-todo';
+
+describe('SQLite SessionTodo store', () => {
+ test('bootstraps active legacy Tasks once and persists the document', async () => {
+ await withRoot(async (root) => {
+ const tasks = createSqliteTaskLedgerStore(root);
+ const created = await tasks.create(SESSION_ID, [
+ { subject: 'pending legacy work' },
+ { subject: 'completed legacy work' },
+ ]);
+ await tasks.update(SESSION_ID, created.created[1]!.id, { status: 'in_progress' });
+ await tasks.update(SESSION_ID, created.created[1]!.id, {
+ status: 'completed',
+ completionEvidence: 'done',
+ });
+ tasks.close();
+
+ const todos = createSqliteSessionTodoStore(root);
+ assert.deepEqual(await todos.readOrBootstrap(SESSION_ID), {
+ items: [{ content: 'pending legacy work', status: 'pending' }],
+ });
+ todos.close();
+
+ const database = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(
+ database
+ .prepare(
+ 'SELECT COUNT(*) AS count FROM workflow_session_todo_documents WHERE session_id = ?',
+ )
+ .get(SESSION_ID)!.count,
+ 1,
+ );
+ } finally {
+ database.close();
+ }
+ });
+ });
+
+ test('keeps a blocked legacy Task visible as pending without importing workflow metadata', async () => {
+ await withRoot(async (root) => {
+ const tasks = createSqliteTaskLedgerStore(root);
+ const created = await tasks.create(SESSION_ID, [{ subject: 'waiting for approval' }]);
+ const taskId = created.created[0]!.id;
+ await tasks.update(SESSION_ID, taskId, { status: 'in_progress' });
+ await tasks.update(SESSION_ID, taskId, {
+ status: 'blocked',
+ blockedReason: 'approval has not arrived',
+ });
+ tasks.close();
+
+ const todos = createSqliteSessionTodoStore(root);
+ assert.deepEqual(await todos.readOrBootstrap(SESSION_ID), {
+ items: [{ content: 'waiting for approval', status: 'pending' }],
+ });
+ todos.close();
+ });
+ });
+
+ test('persists initialized-empty and never revives later legacy Tasks', async () => {
+ await withRoot(async (root) => {
+ const todos = createSqliteSessionTodoStore(root);
+ assert.deepEqual(await todos.readOrBootstrap(SESSION_ID), { items: [] });
+
+ const tasks = createSqliteTaskLedgerStore(root);
+ await tasks.create(SESSION_ID, [{ subject: 'too late for bootstrap' }]);
+ assert.deepEqual(await todos.readOrBootstrap(SESSION_ID), { items: [] });
+ tasks.close();
+ todos.close();
+ });
+ });
+
+ test('lets the first explicit replacement win without reading legacy state', async () => {
+ await withRoot(async (root) => {
+ const tasks = createSqliteTaskLedgerStore(root);
+ await tasks.create(SESSION_ID, [{ subject: 'legacy work' }]);
+ tasks.close();
+
+ const todos = createSqliteSessionTodoStore(root);
+ assert.deepEqual(await todos.replaceAll(SESSION_ID, []), { items: [] });
+ assert.deepEqual(await todos.readOrBootstrap(SESSION_ID), { items: [] });
+ todos.close();
+ });
+ });
+
+ test('serializes a first explicit replacement ahead of a following bootstrap read', async () => {
+ await withRoot(async (root) => {
+ const tasks = createSqliteTaskLedgerStore(root);
+ await tasks.create(SESSION_ID, [{ subject: 'legacy work' }]);
+ tasks.close();
+
+ const todos = createSqliteSessionTodoStore(root);
+ const [written, read] = await Promise.all([
+ todos.replaceAll(SESSION_ID, [{ content: 'explicit work', status: 'in_progress' }]),
+ todos.readOrBootstrap(SESSION_ID),
+ ]);
+ assert.deepEqual(written, {
+ items: [{ content: 'explicit work', status: 'in_progress' }],
+ });
+ assert.deepEqual(read, written);
+ todos.close();
+ });
+ });
+
+ test('persists replacement order across reopen and purge restores uninitialized state', async () => {
+ await withRoot(async (root) => {
+ const first = createSqliteSessionTodoStore(root);
+ await first.replaceAll(SESSION_ID, [
+ { content: 'second', status: 'in_progress' },
+ { content: 'first', status: 'pending' },
+ ]);
+ first.close();
+
+ const reopened = createSqliteSessionTodoStore(root);
+ assert.deepEqual(await reopened.readOrBootstrap(SESSION_ID), {
+ items: [
+ { content: 'second', status: 'in_progress' },
+ { content: 'first', status: 'pending' },
+ ],
+ });
+ await reopened.purge(SESSION_ID);
+ assert.deepEqual(await reopened.readOrBootstrap(SESSION_ID), { items: [] });
+ reopened.close();
+ });
+ });
+
+ test('fails closed on corrupt legacy events without writing an initialized marker', async () => {
+ await withRoot(async (root) => {
+ createSqliteTaskLedgerStore(root).close();
+ const database = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ database
+ .prepare(`
+ INSERT INTO workflow_task_ledger_events(session_id, sequence, event_id, record_json)
+ VALUES (?, 0, 'bad-event', '{not-json')
+ `)
+ .run(SESSION_ID);
+ } finally {
+ database.close();
+ }
+
+ const todos = createSqliteSessionTodoStore(root);
+ await assert.rejects(
+ () => todos.readOrBootstrap(SESSION_ID),
+ /Invalid legacy Task event JSON/,
+ );
+ todos.close();
+
+ const verified = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(
+ verified
+ .prepare(
+ 'SELECT COUNT(*) AS count FROM workflow_session_todo_documents WHERE session_id = ?',
+ )
+ .get(SESSION_ID)!.count,
+ 0,
+ );
+ } finally {
+ verified.close();
+ }
+ });
+ });
+
+ test('explicit replacement recovers without decoding corrupt legacy state', async () => {
+ await withRoot(async (root) => {
+ createSqliteTaskLedgerStore(root).close();
+ const database = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ database
+ .prepare(`
+ INSERT INTO workflow_task_ledger_events(session_id, sequence, event_id, record_json)
+ VALUES (?, 0, 'bad-event', '{not-json')
+ `)
+ .run(SESSION_ID);
+ } finally {
+ database.close();
+ }
+
+ const todos = createSqliteSessionTodoStore(root);
+ assert.deepEqual(
+ await todos.replaceAll(SESSION_ID, [{ content: 'explicit recovery', status: 'pending' }]),
+ { items: [{ content: 'explicit recovery', status: 'pending' }] },
+ );
+ assert.deepEqual(await todos.readOrBootstrap(SESSION_ID), {
+ items: [{ content: 'explicit recovery', status: 'pending' }],
+ });
+ todos.close();
+ });
+ });
+
+ test('fails closed on a corrupt current document but permits explicit replacement recovery', async () => {
+ await withRoot(async (root) => {
+ const initialized = createSqliteSessionTodoStore(root);
+ await initialized.replaceAll(SESSION_ID, []);
+ initialized.close();
+
+ const database = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ database
+ .prepare(
+ 'UPDATE workflow_session_todo_documents SET record_json = ? WHERE session_id = ?',
+ )
+ .run('{not-json', SESSION_ID);
+ } finally {
+ database.close();
+ }
+
+ const todos = createSqliteSessionTodoStore(root);
+ await assert.rejects(() => todos.readOrBootstrap(SESSION_ID), /Invalid SessionTodo document/);
+ assert.deepEqual(
+ await todos.replaceAll(SESSION_ID, [{ content: 'recovered', status: 'pending' }]),
+ { items: [{ content: 'recovered', status: 'pending' }] },
+ );
+ assert.deepEqual(await todos.readOrBootstrap(SESSION_ID), {
+ items: [{ content: 'recovered', status: 'pending' }],
+ });
+ todos.close();
+ });
+ });
+});
+
+async function withRoot(run: (root: string) => Promise<void>): Promise<void> {
+ const root = await mkdtemp(join(tmpdir(), 'maka-session-todo-'));
+ try {
+ await run(root);
+ } finally {
+ await rm(root, { recursive: true, force: true });
+ }
+}
diff --git a/packages/storage/src/__tests__/sqlite-workflow-store.test.ts b/packages/storage/src/__tests__/sqlite-workflow-store.test.ts
index 23fb82b95a..6489c1493e 100644
--- a/packages/storage/src/__tests__/sqlite-workflow-store.test.ts
+++ b/packages/storage/src/__tests__/sqlite-workflow-store.test.ts
@@ -31,6 +31,7 @@ import {
import { openInteractiveScheduledTaskStoreForWrite } from '../scheduled-task-store.js';
import { createSqlitePlanStore } from '../plan-store.js';
import { createSqliteTaskLedgerStore } from '../task-ledger-store.js';
+import { createSqliteSessionTodoStore } from '../session-todo-store.js';
import { SQLITE_WORKFLOW_SCHEMA_VERSION } from '../sqlite-workflow-schema.js';
import { resolveStorageRoot, tryAcquireInteractiveRootOwner } from '../root-authority.js';
import {
@@ -106,7 +107,7 @@ describe('SQLite workflow stores', () => {
});
});
- test('migrates released workflow schema 9 projections to event-only schema 10', async () => {
+ test('migrates released workflow schema 9 projections to the current workflow schema', async () => {
await withRoot(async (root) => {
const taskStore = createSqliteTaskLedgerStore(root);
await taskStore.create(SESSION_ID, [{ subject: 'Preserve event authority' }]);
@@ -171,6 +172,46 @@ describe('SQLite workflow stores', () => {
});
});
+ test('adds SessionTodo storage to workflow schema 10 without changing Task events', async () => {
+ await withRoot(async (root) => {
+ const tasks = createSqliteTaskLedgerStore(root);
+ const created = await tasks.create(SESSION_ID, [{ subject: 'Preserve schema 10 Task' }]);
+ const taskId = created.created[0]!.id;
+ await tasks.update(SESSION_ID, taskId, { status: 'in_progress' });
+ await tasks.update(SESSION_ID, taskId, {
+ status: 'blocked',
+ blockedReason: 'waiting across the upgrade',
+ });
+ tasks.close();
+
+ const released = new DatabaseSync(join(root, 'runtime.sqlite'));
+ try {
+ released.exec('DROP TABLE workflow_session_todo_documents');
+ setWorkflowSchemaVersion(released, 10);
+ } finally {
+ released.close();
+ }
+
+ const migrated = createSqliteSessionTodoStore(root);
+ try {
+ assert.deepEqual(await migrated.readOrBootstrap(SESSION_ID), {
+ items: [{ content: 'Preserve schema 10 Task', status: 'pending' }],
+ });
+ } finally {
+ migrated.close();
+ }
+
+ const verified = new DatabaseSync(join(root, 'runtime.sqlite'), { readOnly: true });
+ try {
+ assert.equal(workflowSchemaVersion(verified), SQLITE_WORKFLOW_SCHEMA_VERSION);
+ assert.equal(rowCount(verified, 'workflow_task_ledger_events'), 3);
+ assert.equal(rowCount(verified, 'workflow_session_todo_documents'), 1);
+ } finally {
+ verified.close();
+ }
+ });
+ });
+
test('preserves every released projection when one table has unfamiliar DDL', async () => {
await withRoot(async (root) => {
createSqliteTaskLedgerStore(root).close();
@@ -403,7 +444,7 @@ describe('SQLite workflow stores', () => {
});
});
- test('backs up and restores event-only Task Ledger and Plan state', async () => {
+ test('backs up and restores Task, Plan, and initialized SessionTodo state', async () => {
const base = await mkdtemp(join(tmpdir(), 'maka-workflow-backup-'));
const stateRoot = join(base, 'state');
const backupRoot = join(base, 'backup');
@@ -426,6 +467,13 @@ describe('SQLite workflow stores', () => {
});
planStore.close();
+ const todoStore = createSqliteSessionTodoStore(stateRoot);
+ await todoStore.replaceAll('todo-non-empty', [
+ { content: 'Restore current Todo', status: 'in_progress' },
+ ]);
+ await todoStore.replaceAll('todo-empty', []);
+ todoStore.close();
+
await createOperationalStateBackup({ stateRoot, destinationRoot: backupRoot, now: () => 10 });
await restoreOperationalStateBackup({ backupRoot, destinationRoot: restoreRoot });
@@ -444,6 +492,15 @@ describe('SQLite workflow stores', () => {
} finally {
restoredPlan.close();
}
+ const restoredTodos = createSqliteSessionTodoStore(restoreRoot);
+ try {
+ assert.deepEqual(await restoredTodos.readOrBootstrap('todo-non-empty'), {
+ items: [{ content: 'Restore current Todo', status: 'in_progress' }],
+ });
+ assert.deepEqual(await restoredTodos.readOrBootstrap('todo-empty'), { items: [] });
+ } finally {
+ restoredTodos.close();
+ }
const restored = new DatabaseSync(join(restoreRoot, 'runtime.sqlite'), { readOnly: true });
try {
@@ -451,6 +508,7 @@ describe('SQLite workflow stores', () => {
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);
+ assert.equal(rowCount(restored, 'workflow_session_todo_documents'), 2);
} finally {
restored.close();
}
diff --git a/packages/storage/src/session-todo-authority.ts b/packages/storage/src/session-todo-authority.ts
new file mode 100644
index 0000000000..fb0fb6b855
--- /dev/null
+++ b/packages/storage/src/session-todo-authority.ts
@@ -0,0 +1,131 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import {
+ assertStorageRootLease,
+ runWithStorageRootLease,
+ StorageRootAuthorityError,
+ type StorageRootLease,
+} from './root-authority.js';
+import {
+ createSqliteSessionTodoStore,
+ type SessionTodoStore,
+ type SqliteSessionTodoStore,
+} from './session-todo-store.js';
+
+const writerBrand: unique symbol = Symbol('InteractiveSessionTodoWriter');
+const writers = new WeakSet<object>();
+const writerByLease = new WeakMap<object, InteractiveSessionTodoWriter>();
+const writerOpeningByLease = new WeakMap<object, Promise<InteractiveSessionTodoWriter>>();
+
+export interface InteractiveSessionTodoWriter extends SessionTodoStore {
+ readonly kind: 'interactive';
+ readonly access: 'write';
+ readonly [writerBrand]: true;
+ close(): void;
+}
+
+export function authenticateInteractiveSessionTodoWriter(
+ writer: InteractiveSessionTodoWriter,
+): InteractiveSessionTodoWriter {
+ if (!writers.has(writer)) {
+ throw new StorageRootAuthorityError(
+ 'invalid_lease',
+ 'Expected an authentic interactive SessionTodo writer',
+ );
+ }
+ return writer;
+}
+
+export async function openInteractiveSessionTodoStoreForWrite(
+ lease: StorageRootLease<'interactive', 'write'>,
+): Promise<InteractiveSessionTodoWriter> {
+ await assertStorageRootLease(lease, 'interactive', 'write');
+ const existing = writerByLease.get(lease);
+ if (existing) return existing;
+ const opening = writerOpeningByLease.get(lease);
+ if (opening) return opening;
+
+ const pending = Promise.resolve().then(async () => {
+ let store: SqliteSessionTodoStore | undefined;
+ try {
+ store = await runWithStorageRootLease(lease, 'interactive', 'write', async (root) => {
+ const opened = createSqliteSessionTodoStore(root);
+ try {
+ await opened.ready();
+ return opened;
+ } catch (error) {
+ opened.close();
+ throw error;
+ }
+ });
+ await assertStorageRootLease(lease, 'interactive', 'write');
+ const recoveredExisting = writerByLease.get(lease);
+ if (recoveredExisting) {
+ store.close();
+ return recoveredExisting;
+ }
+ const writer = createWriterFacade(lease, store);
+ writers.add(writer);
+ writerByLease.set(lease, writer);
+ return writer;
+ } catch (error) {
+ store?.close();
+ throw error;
+ }
+ });
+ writerOpeningByLease.set(lease, pending);
+ try {
+ return await pending;
+ } finally {
+ if (writerOpeningByLease.get(lease) === pending) writerOpeningByLease.delete(lease);
+ }
+}
+
+function createWriterFacade(
+ lease: StorageRootLease<'interactive', 'write'>,
+ store: SqliteSessionTodoStore,
+): InteractiveSessionTodoWriter {
+ let closed = false;
+ const run = <T>(operation: () => Promise<T>) => {
+ if (closed) {
+ return Promise.reject(
+ new StorageRootAuthorityError('invalid_lease', 'SessionTodo writer is closed'),
+ );
+ }
+ return runWithStorageRootLease(lease, 'interactive', 'write', async () => operation());
+ };
+ const writer: InteractiveSessionTodoWriter = {
+ kind: 'interactive',
+ access: 'write',
+ [writerBrand]: true,
+ readOrBootstrap: (sessionId) => run(() => store.readOrBootstrap(sessionId)),
+ replaceAll: (sessionId, items) => run(() => store.replaceAll(sessionId, items)),
+ purge: (sessionId) => run(() => store.purge(sessionId)),
+ close: () => {
+ if (closed) return;
+ closed = true;
+ if (writerByLease.get(lease) === writer) writerByLease.delete(lease);
+ writers.delete(writer);
+ store.close();
+ },
+ };
+ Object.freeze(writer);
+ return writer;
+}
diff --git a/packages/storage/src/session-todo-store.ts b/packages/storage/src/session-todo-store.ts
new file mode 100644
index 0000000000..781fe373ab
--- /dev/null
+++ b/packages/storage/src/session-todo-store.ts
@@ -0,0 +1,244 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { resolve } from 'node:path';
+import type { DatabaseSync } from 'node:sqlite';
+import {
+ isTaskLedgerEvent,
+ projectTaskLedgerEvents,
+ type TaskLedgerEvent,
+} from '@maka/core/task-ledger';
+import {
+ normalizeSessionTodoItems,
+ type SessionTodoItem,
+ type SessionTodoSnapshot,
+} from '@maka/core/session-todo';
+import {
+ acquireOperationalStateDatabase,
+ type OperationalStateDatabaseLease,
+} from './operational-state-store.js';
+import { assertSafeSessionId } from './session-store.js';
+import { chainWrite } from './write-queue.js';
+
+const SESSION_TODO_DOCUMENT_SCHEMA_VERSION = 1;
+
+interface StoredSessionTodoDocument {
+ schemaVersion: typeof SESSION_TODO_DOCUMENT_SCHEMA_VERSION;
+ items: SessionTodoItem[];
+}
+
+export interface SessionTodoStore {
+ /**
+ * Return the initialized current document. On the first read only, bootstrap
+ * pending/in-progress legacy Tasks, map blocked Tasks to pending, and persist
+ * even an empty result.
+ */
+ readOrBootstrap(sessionId: string): Promise<SessionTodoSnapshot>;
+ /** Replace the complete document without consulting legacy Task state. */
+ replaceAll(sessionId: string, items: unknown): Promise<SessionTodoSnapshot>;
+ purge(sessionId: string): Promise<void>;
+}
+
+export interface SqliteSessionTodoStore extends SessionTodoStore {
+ ready(): Promise<void>;
+ close(): void;
+}
+
+export function createSqliteSessionTodoStore(workspaceRoot: string): SqliteSessionTodoStore {
+ return new SqliteSessionTodoStoreImpl(workspaceRoot);
+}
+
+class SqliteSessionTodoStoreImpl implements SqliteSessionTodoStore {
+ readonly #lease: OperationalStateDatabaseLease;
+ private readonly writeQueues = new Map<string, Promise<void>>();
+
+ constructor(workspaceRoot: string) {
+ this.#lease = acquireOperationalStateDatabase(resolve(workspaceRoot));
+ }
+
+ ready(): Promise<void> {
+ return Promise.resolve();
+ }
+
+ close(): void {
+ this.#lease.close();
+ }
+
+ async readOrBootstrap(sessionId: string): Promise<SessionTodoSnapshot> {
+ assertSafeSessionId(sessionId);
+ let snapshot: SessionTodoSnapshot | undefined;
+ await chainWrite(this.writeQueues, sessionId, async () => {
+ snapshot = this.#lease.transaction('write', () => {
+ const existing = readStoredDocument(this.#lease.database, sessionId);
+ if (existing) return snapshotFromDocument(existing);
+
+ const bootstrapped = bootstrapLegacyTasks(this.#lease.database, sessionId);
+ insertDocument(this.#lease.database, sessionId, bootstrapped);
+ return snapshotFromDocument(bootstrapped);
+ });
+ });
+ return snapshot!;
+ }
+
+ async replaceAll(sessionId: string, items: unknown): Promise<SessionTodoSnapshot> {
+ assertSafeSessionId(sessionId);
+ const normalized = normalizeSessionTodoItems(items);
+ if (!normalized.ok) throw new Error(normalized.message);
+ const document: StoredSessionTodoDocument = {
+ schemaVersion: SESSION_TODO_DOCUMENT_SCHEMA_VERSION,
+ items: normalized.value.items,
+ };
+ await chainWrite(this.writeQueues, sessionId, async () => {
+ this.#lease.transaction('write', () =>
+ upsertDocument(this.#lease.database, sessionId, document),
+ );
+ });
+ return snapshotFromDocument(document);
+ }
+
+ async purge(sessionId: string): Promise<void> {
+ assertSafeSessionId(sessionId);
+ await chainWrite(this.writeQueues, sessionId, async () => {
+ this.#lease.transaction('write', () => {
+ this.#lease.database
+ .prepare('DELETE FROM workflow_session_todo_documents WHERE session_id = ?')
+ .run(sessionId);
+ });
+ });
+ }
+}
+
+function readStoredDocument(
+ database: DatabaseSync,
+ sessionId: string,
+): StoredSessionTodoDocument | undefined {
+ const row = database
+ .prepare('SELECT record_json FROM workflow_session_todo_documents WHERE session_id = ?')
+ .get(sessionId) as { record_json?: unknown } | undefined;
+ if (!row) return undefined;
+ if (typeof row.record_json !== 'string') throw new Error('Invalid SessionTodo document record');
+ let parsed: unknown;
+ try {
+ parsed = JSON.parse(row.record_json);
+ } catch {
+ throw new Error('Invalid SessionTodo document JSON');
+ }
+ if (typeof parsed !== 'object' || parsed === null || Array.isArray(parsed)) {
+ throw new Error('Invalid SessionTodo document shape');
+ }
+ const record = parsed as Record<string, unknown>;
+ const keys = Object.keys(record).sort();
+ if (keys.length !== 2 || keys[0] !== 'items' || keys[1] !== 'schemaVersion') {
+ throw new Error('Invalid SessionTodo document fields');
+ }
+ if (record.schemaVersion !== SESSION_TODO_DOCUMENT_SCHEMA_VERSION) {
+ throw new Error(`Unsupported SessionTodo document schema: ${String(record.schemaVersion)}`);
+ }
+ const normalized = normalizeSessionTodoItems(record.items);
+ if (!normalized.ok) throw new Error(`Invalid SessionTodo document: ${normalized.message}`);
+ return {
+ schemaVersion: SESSION_TODO_DOCUMENT_SCHEMA_VERSION,
+ items: normalized.value.items,
+ };
+}
+
+function bootstrapLegacyTasks(
+ database: DatabaseSync,
+ sessionId: string,
+): StoredSessionTodoDocument {
+ const rows = database
+ .prepare(`
+ SELECT record_json
+ FROM workflow_task_ledger_events
+ WHERE session_id = ?
+ ORDER BY sequence
+ `)
+ .all(sessionId) as Array<{ record_json?: unknown }>;
+ const events = rows.map((row, index) => {
+ if (typeof row.record_json !== 'string') {
+ throw new Error(`Invalid legacy Task event at sequence ${index}`);
+ }
+ let parsed: unknown;
+ try {
+ parsed = JSON.parse(row.record_json);
+ } catch {
+ throw new Error(`Invalid legacy Task event JSON at sequence ${index}`);
+ }
+ if (!isTaskLedgerEvent(parsed) || parsed.sessionId !== sessionId) {
+ throw new Error(`Invalid legacy Task event at sequence ${index}`);
+ }
+ return parsed;
+ }) as TaskLedgerEvent[];
+ const projected = projectTaskLedgerEvents(events);
+ if (projected.diagnostics.length > 0) {
+ throw new Error(`Legacy Task ledger is not projectable: ${projected.diagnostics.join('; ')}`);
+ }
+ const items = projected.tasks.flatMap((task) => {
+ switch (task.status) {
+ case 'pending':
+ case 'in_progress':
+ return [{ content: task.subject, status: task.status }];
+ case 'blocked':
+ // SessionTodo deliberately has no workflow-specific blocked state or
+ // reason field. Keep the unfinished subject visible for replanning.
+ return [{ content: task.subject, status: 'pending' as const }];
+ case 'completed':
+ case 'failed':
+ case 'cancelled':
+ return [];
+ }
+ });
+ const normalized = normalizeSessionTodoItems(items);
+ if (!normalized.ok) throw new Error(`Legacy Task bootstrap failed: ${normalized.message}`);
+ return {
+ schemaVersion: SESSION_TODO_DOCUMENT_SCHEMA_VERSION,
+ items: normalized.value.items,
+ };
+}
+
+function insertDocument(
+ database: DatabaseSync,
+ sessionId: string,
+ document: StoredSessionTodoDocument,
+): void {
+ database
+ .prepare(`
+ INSERT INTO workflow_session_todo_documents(session_id, record_json)
+ VALUES (?, ?)
+ `)
+ .run(sessionId, JSON.stringify(document));
+}
+
+function upsertDocument(
+ database: DatabaseSync,
+ sessionId: string,
+ document: StoredSessionTodoDocument,
+): void {
+ database
+ .prepare(`
+ INSERT INTO workflow_session_todo_documents(session_id, record_json)
+ VALUES (?, ?)
+ ON CONFLICT(session_id) DO UPDATE SET record_json = excluded.record_json
+ `)
+ .run(sessionId, JSON.stringify(document));
+}
+
+function snapshotFromDocument(document: StoredSessionTodoDocument): SessionTodoSnapshot {
+ return { items: document.items.map((item) => ({ ...item })) };
+}
diff --git a/packages/storage/src/sqlite-workflow-schema.ts b/packages/storage/src/sqlite-workflow-schema.ts
index ab3486f07e..545ad69619 100644
--- a/packages/storage/src/sqlite-workflow-schema.ts
+++ b/packages/storage/src/sqlite-workflow-schema.ts
@@ -19,7 +19,7 @@
import type { DatabaseSync } from 'node:sqlite';
-export const SQLITE_WORKFLOW_SCHEMA_VERSION = 10;
+export const SQLITE_WORKFLOW_SCHEMA_VERSION = 11;
const RELEASED_WORKFLOW_PROJECTION_TABLES = [
{
@@ -54,6 +54,11 @@ export function migrateSqliteWorkflowDatabase(db: DatabaseSync): void {
UNIQUE (session_id, event_id)
);
+ CREATE TABLE IF NOT EXISTS workflow_session_todo_documents (
+ 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),
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@maka.apache.org
For additional commands, e-mail: commits-help@maka.apache.org