From fe00da920e3b59bc8690c7efbee93365e1731e4b Mon Sep 17 00:00:00 2001 From: tim8es <108188608+tim8es@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:41:50 +0500 Subject: [PATCH 1/2] test: define durable persistence invariants --- .../persistence/canonical-domain-validator.ts | 69 +++ .../d1-runtime-persistence-invariants.test.ts | 416 ++++++++++++++++++ .../d1-runtime-persistence.test.ts | 374 ++++++++++++++++ test/persistence/d1-sqlite-harness.ts | 100 +++++ test/persistence/fixtures.ts | 230 ++++++++++ test/persistence/setup.ts | 23 + 6 files changed, 1212 insertions(+) create mode 100644 test/persistence/canonical-domain-validator.ts create mode 100644 test/persistence/d1-runtime-persistence-invariants.test.ts create mode 100644 test/persistence/d1-runtime-persistence.test.ts create mode 100644 test/persistence/d1-sqlite-harness.ts create mode 100644 test/persistence/fixtures.ts create mode 100644 test/persistence/setup.ts diff --git a/test/persistence/canonical-domain-validator.ts b/test/persistence/canonical-domain-validator.ts new file mode 100644 index 00000000..cb06444c --- /dev/null +++ b/test/persistence/canonical-domain-validator.ts @@ -0,0 +1,69 @@ +import { readdirSync, readFileSync } from 'node:fs'; +import { createRequire } from 'node:module'; +import { dirname, join } from 'node:path'; +import { fileURLToPath } from 'node:url'; + +import type { Ajv, AnySchemaObject, ErrorObject, Options, ValidateFunction } from 'ajv'; +import type { FormatsPlugin } from 'ajv-formats'; + +import type { + PersistenceDomainTarget, + PersistenceDomainValidator, +} from '../../src/persistence/ports.ts'; + +const require = createRequire(import.meta.url); +const Ajv2020 = require('ajv/dist/2020.js') as new (options?: Options) => Ajv; +const addFormats = require('ajv-formats') as FormatsPlugin; +const here = dirname(fileURLToPath(import.meta.url)); +const schemaDirectory = join(here, '../../schemas/domain/v1'); + +const schemaRefs: Record = { + Workspace: 'urn:mindrail:schema:domain:v1:workspace', + Agent: 'urn:mindrail:schema:domain:v1:agent', + Session: 'urn:mindrail:schema:domain:v1:session', + Goal: 'urn:mindrail:schema:domain:v1:goal', + Task: 'urn:mindrail:schema:domain:v1:task', + Lease: 'urn:mindrail:schema:domain:v1:lease', + Checkpoint: 'urn:mindrail:schema:domain:v1:checkpoint', + PermissionRequest: 'urn:mindrail:schema:domain:v1:permission-request', + PermissionDecision: 'urn:mindrail:schema:domain:v1:permission-decision', + AuditEvent: 'urn:mindrail:schema:domain:v1:audit-event', +}; + +const ajv = new Ajv2020({ + allErrors: true, + strict: true, + validateFormats: true, +}); +addFormats(ajv); + +for (const name of readdirSync(schemaDirectory) + .filter((entry) => entry.endsWith('.schema.json')) + .sort()) { + const schema = JSON.parse(readFileSync(join(schemaDirectory, name), 'utf8')) as AnySchemaObject; + ajv.addSchema(schema); +} + +const validators = new Map(); +for (const [target, ref] of Object.entries(schemaRefs) as [PersistenceDomainTarget, string][]) { + validators.set(target, ajv.getSchema(ref) ?? ajv.compile({ $ref: ref })); +} + +export const persistenceCanonicalValidator: PersistenceDomainValidator = (target, value) => { + const validate = validators.get(target); + if (!validate) { + throw new Error(`Canonical validator for ${target} was not configured.`); + } + if (validate(value)) { + return { valid: true }; + } + return { + valid: false, + errors: (validate.errors ?? []).map(formatValidationError), + }; +}; + +function formatValidationError(error: ErrorObject): string { + const location = error.instancePath.length === 0 ? '/' : error.instancePath; + return `${location} ${error.keyword}${error.message === undefined ? '' : ` ${error.message}`}`; +} diff --git a/test/persistence/d1-runtime-persistence-invariants.test.ts b/test/persistence/d1-runtime-persistence-invariants.test.ts new file mode 100644 index 00000000..d01e0d87 --- /dev/null +++ b/test/persistence/d1-runtime-persistence-invariants.test.ts @@ -0,0 +1,416 @@ +import { mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import type { Checkpoint, Goal, Lease, Task } from '@mindrail/contracts'; +import { describe, expect, it } from 'vitest'; + +import { PersistenceError } from '../../src/persistence/ports.ts'; +import { + agent, + auditEvent, + checkpoint, + goal, + humanDecision, + leaseCandidate, + permissionRequest, + policyDecision, + receipt, + session, + T0, + T1, + T2, + T3, + task, + workspace, +} from './fixtures.ts'; +import { openPersistence } from './setup.ts'; + +function databasePath(): string { + return join(mkdtempSync(join(tmpdir(), 'mindrail-persistence-invariants-')), 'runtime.sqlite'); +} + +function expectCode(error: unknown, code: PersistenceError['code']): void { + expect(error).toBeInstanceOf(PersistenceError); + expect((error as PersistenceError).code).toBe(code); +} + +async function seedClaimedTask( + persistence: Awaited>['persistence'], + workspaceId: string, + suffix: string, +) { + const workspaceRecord = workspace(workspaceId); + const goalRecord = goal(workspaceId, `goal-${suffix}`); + const taskRecord = task(workspaceId, goalRecord.id, `task-${suffix}`); + const agentRecord = agent(workspaceId, `agent-${suffix}`); + const sessionRecord = session(workspaceId, agentRecord.id, `session-${suffix}`); + await persistence.bootstrapWorkspace(workspaceRecord); + await persistence.createGoal({ goal: goalRecord }); + await persistence.createTask({ task: taskRecord }); + await persistence.createAgent({ agent: agentRecord }); + await persistence.createSession({ session: sessionRecord }); + const claim = await persistence.claimTask({ + workspaceId, + taskId: taskRecord.id, + sessionId: sessionRecord.id, + expectedTaskRevision: 1, + lease: leaseCandidate( + workspaceId, + taskRecord.id, + sessionRecord.id, + `lease-${suffix}`, + T3, + ), + now: T0, + }); + if (claim.kind !== 'committed') throw new Error('expected committed seed claim'); + return { + workspace: workspaceRecord, + goal: goalRecord, + originalTask: taskRecord, + agent: agentRecord, + session: sessionRecord, + task: claim.value.task, + lease: claim.value.lease, + }; +} + +function completionRecords( + currentTask: Task, + currentLease: Lease, + sessionId: string, + checkpointId: string, +): { task: Task; lease: Lease; checkpoint: Checkpoint } { + return { + task: { + ...currentTask, + revision: currentTask.revision + 1, + updatedAt: T1, + status: 'succeeded', + }, + lease: { + ...currentLease, + revision: currentLease.revision + 1, + updatedAt: T1, + status: 'released', + }, + checkpoint: { + ...checkpoint( + currentTask.workspaceId, + currentTask.id, + sessionId, + currentLease.id, + currentLease.fencingToken, + checkpointId, + T1, + ), + kind: 'result', + summary: 'Task completed.', + }, + }; +} + +describe('D1RuntimePersistence ordering, isolation, and history invariants', () => { + it('serializes CreateTask against Goal auto-terminalization in one deterministic Workspace order', async () => { + const path = databasePath(); + const { database, persistence } = await openPersistence(path); + + const first = await seedClaimedTask(persistence, 'ws-complete-first', 'complete-first'); + const completedFirst = completionRecords( + first.task, + first.lease, + first.session.id, + 'checkpoint-complete-first', + ); + const completionPromise = persistence.completeTask({ + workspaceId: first.workspace.id, + ...completedFirst, + expectedTaskRevision: first.task.revision, + now: T1, + }); + const lateTaskPromise = persistence.createTask({ + task: task(first.workspace.id, first.goal.id, 'task-too-late'), + }); + const [completionResult, lateTaskResult] = await Promise.allSettled([ + completionPromise, + lateTaskPromise, + ]); + expect(completionResult.status).toBe('fulfilled'); + expect(lateTaskResult.status).toBe('rejected'); + if (lateTaskResult.status !== 'rejected') throw new Error('expected late Task rejection'); + expectCode(lateTaskResult.reason, 'INVALID_STATE_TRANSITION'); + const completedFirstState = await persistence.loadWorkspaceState(first.workspace.id); + expect(completedFirstState?.goals[0]).toEqual(expect.objectContaining({ status: 'succeeded' })); + expect(completedFirstState?.tasks.map((item) => item.id)).toEqual([first.task.id]); + + const second = await seedClaimedTask(persistence, 'ws-create-first', 'create-first'); + const completedSecond = completionRecords( + second.task, + second.lease, + second.session.id, + 'checkpoint-create-first', + ); + const earlyTaskPromise = persistence.createTask({ + task: task(second.workspace.id, second.goal.id, 'task-created-before-terminalization'), + }); + const secondCompletionPromise = persistence.completeTask({ + workspaceId: second.workspace.id, + ...completedSecond, + expectedTaskRevision: second.task.revision, + now: T1, + }); + const [earlyTaskResult, secondCompletionResult] = await Promise.allSettled([ + earlyTaskPromise, + secondCompletionPromise, + ]); + expect(earlyTaskResult.status).toBe('fulfilled'); + expect(secondCompletionResult.status).toBe('fulfilled'); + const createFirstState = await persistence.loadWorkspaceState(second.workspace.id); + expect(createFirstState?.goals[0]).toEqual(expect.objectContaining({ status: 'active' })); + expect(createFirstState?.tasks.map((item) => item.id).sort()).toEqual( + [second.task.id, 'task-created-before-terminalization'].sort(), + ); + database.close(); + }); + + it('rejects cross-Workspace references before durable insertion', async () => { + const path = databasePath(); + const { database, persistence } = await openPersistence(path); + await persistence.bootstrapWorkspace(workspace('ws-a')); + await persistence.bootstrapWorkspace(workspace('ws-b')); + await persistence.createGoal({ goal: goal('ws-a', 'goal-a') }); + + await expect( + persistence.createTask({ task: task('ws-b', 'goal-a', 'task-cross-workspace') }), + ).rejects.toSatisfy((error: unknown) => { + expectCode(error, 'NOT_FOUND'); + return true; + }); + expect((await persistence.loadWorkspaceState('ws-b'))?.tasks).toHaveLength(0); + database.close(); + }); + + it('validates canonical records before commit and does not reserve a receipt for invalid input', async () => { + const path = databasePath(); + const { database, persistence } = await openPersistence(path); + await persistence.bootstrapWorkspace(workspace()); + await persistence.createGoal({ goal: goal() }); + const invalidTask = { ...task(), title: '' } as Task; + + await expect( + persistence.createTask({ + task: invalidTask, + receipt: receipt( + 'cmd-invalid-task', + 'fingerprint-invalid-task', + { error: 'must-not-be-stored' }, + 'ws-a', + 'CreateTask', + ), + }), + ).rejects.toSatisfy((error: unknown) => { + expectCode(error, 'INVALID_RECORD'); + return true; + }); + expect(await persistence.getCommandReceipt('ws-a', 'cmd-invalid-task')).toBeUndefined(); + expect((await persistence.loadWorkspaceState('ws-a'))?.tasks).toHaveLength(0); + database.close(); + }); + + it('keeps checkpoints and audit history ordered and immutable', async () => { + const path = databasePath(); + const { database, persistence } = await openPersistence(path); + const seeded = await seedClaimedTask(persistence, 'ws-history', 'history'); + const firstCheckpoint = checkpoint( + seeded.workspace.id, + seeded.task.id, + seeded.session.id, + seeded.lease.id, + seeded.lease.fencingToken, + 'checkpoint-1', + T1, + ); + const secondCheckpoint = { + ...checkpoint( + seeded.workspace.id, + seeded.task.id, + seeded.session.id, + seeded.lease.id, + seeded.lease.fencingToken, + 'checkpoint-2', + T2, + ), + summary: 'Later checkpoint.', + }; + await persistence.appendCheckpoint({ checkpoint: firstCheckpoint, now: T1 }); + await persistence.appendCheckpoint({ checkpoint: secondCheckpoint, now: T2 }); + await persistence.appendAuditEvent({ + auditEvent: auditEvent(seeded.workspace.id, 'audit-1', T1, seeded.task.id), + }); + await persistence.appendAuditEvent({ + auditEvent: auditEvent(seeded.workspace.id, 'audit-2', T2, seeded.task.id), + }); + + expect((await persistence.listTaskCheckpoints(seeded.workspace.id, seeded.task.id)).map((item) => item.id)).toEqual([ + 'checkpoint-1', + 'checkpoint-2', + ]); + expect((await persistence.listAuditEvents(seeded.workspace.id, 10)).map((item) => item.id)).toEqual([ + 'audit-1', + 'audit-2', + ]); + + await expect( + database.exec( + `UPDATE checkpoints SET record_json = '{}' WHERE workspace_id = '${seeded.workspace.id}' AND id = 'checkpoint-1';`, + ), + ).rejects.toThrow(/append-only/i); + await expect( + database.exec( + `DELETE FROM audit_events WHERE workspace_id = '${seeded.workspace.id}' AND id = 'audit-1';`, + ), + ).rejects.toThrow(/append-only/i); + expect((await persistence.listTaskCheckpoints(seeded.workspace.id, seeded.task.id))[0]).toEqual( + firstCheckpoint, + ); + database.close(); + }); + + it('supports pending-human permission reads and immutable decision-head advancement without evaluating policy', async () => { + const path = databasePath(); + const { database, persistence } = await openPersistence(path); + const seeded = await seedClaimedTask(persistence, 'ws-permission', 'permission'); + const request = permissionRequest( + seeded.workspace.id, + seeded.task.id, + seeded.session.id, + seeded.lease.id, + seeded.lease.fencingToken, + 'permission-request-a', + ); + const initialDecision = policyDecision(seeded.workspace.id, request.id, 'permission-decision-a'); + + await persistence.appendPermissionRequestWithInitialDecision({ + request, + decision: initialDecision, + receipt: receipt( + 'cmd-permission', + 'fingerprint-permission', + { requestId: request.id, decisionId: initialDecision.id }, + seeded.workspace.id, + 'RequestPermission', + ), + auditEvent: auditEvent(seeded.workspace.id, 'audit-permission', T1, seeded.task.id), + }); + + expect(await persistence.listPendingHumanPermissions(seeded.workspace.id, 10)).toEqual([ + { request, latestDecision: initialDecision }, + ]); + + const finalDecision = humanDecision( + seeded.workspace.id, + request.id, + initialDecision.id, + 'permission-decision-b', + ); + await persistence.appendPermissionDecision({ + decision: finalDecision, + expectedPreviousDecisionId: initialDecision.id, + }); + expect(await persistence.listPendingHumanPermissions(seeded.workspace.id, 10)).toEqual([]); + expect(await persistence.listPermissionDecisions(seeded.workspace.id, request.id)).toEqual([ + initialDecision, + finalDecision, + ]); + + await expect( + database.exec( + `UPDATE permission_decisions SET record_json = '{}' WHERE workspace_id = '${seeded.workspace.id}' AND id = '${initialDecision.id}';`, + ), + ).rejects.toThrow(/append-only/i); + database.close(); + }); + + it('exposes authoritative recovery reads for active leases past expiry and sessions past liveness cutoff', async () => { + const path = databasePath(); + const { database, persistence } = await openPersistence(path); + await persistence.bootstrapWorkspace(workspace()); + await persistence.createGoal({ goal: goal() }); + await persistence.createTask({ task: task() }); + await persistence.createAgent({ agent: agent() }); + await persistence.createSession({ session: session() }); + const claim = await persistence.claimTask({ + workspaceId: 'ws-a', + taskId: 'task-a', + sessionId: 'session-a', + expectedTaskRevision: 1, + lease: leaseCandidate('ws-a', 'task-a', 'session-a', 'lease-expiring', T1), + now: T0, + }); + if (claim.kind !== 'committed') throw new Error('expected committed claim'); + + expect(await persistence.listExpiredActiveLeases('ws-a', T1, 10)).toEqual([ + claim.value.lease, + ]); + expect(await persistence.listActiveSessionsLastSeenBefore('ws-a', T1, 10)).toEqual([ + session(), + ]); + database.close(); + }); + + it('keeps Workspace command receipts isolated even when commandId values match', async () => { + const path = databasePath(); + const { database, persistence } = await openPersistence(path); + await persistence.bootstrapWorkspace(workspace('ws-a')); + await persistence.bootstrapWorkspace(workspace('ws-b')); + const goalA = goal('ws-a', 'goal-a'); + const goalB = goal('ws-b', 'goal-b'); + + await persistence.createGoal({ + goal: goalA, + receipt: receipt('cmd-shared', 'fp-a', { goalId: goalA.id }, 'ws-a', 'CreateGoal'), + }); + await persistence.createGoal({ + goal: goalB, + receipt: receipt('cmd-shared', 'fp-b', { goalId: goalB.id }, 'ws-b', 'CreateGoal'), + }); + + expect((await persistence.getCommandReceipt('ws-a', 'cmd-shared'))?.semanticFingerprint).toBe( + 'fp-a', + ); + expect((await persistence.getCommandReceipt('ws-b', 'cmd-shared'))?.semanticFingerprint).toBe( + 'fp-b', + ); + database.close(); + }); + + it('does not expose mutable current state through receipt replay snapshots', async () => { + const path = databasePath(); + const { database, persistence } = await openPersistence(path); + await persistence.bootstrapWorkspace(workspace()); + const originalGoal = goal(); + await persistence.createGoal({ + goal: originalGoal, + receipt: receipt( + 'cmd-goal-snapshot', + 'fp-goal-snapshot', + { result: { goal: originalGoal } }, + 'ws-a', + 'CreateGoal', + ), + }); + const succeededGoal: Goal = { + ...originalGoal, + revision: 2, + updatedAt: T2, + status: 'succeeded', + }; + await persistence.updateGoal({ goal: succeededGoal, expectedRevision: 1 }); + + const stored = await persistence.getCommandReceipt('ws-a', 'cmd-goal-snapshot'); + expect(stored?.responseSnapshot).toEqual({ result: { goal: originalGoal } }); + expect((await persistence.loadWorkspaceState('ws-a'))?.goals[0]).toEqual(succeededGoal); + database.close(); + }); +}); diff --git a/test/persistence/d1-runtime-persistence.test.ts b/test/persistence/d1-runtime-persistence.test.ts new file mode 100644 index 00000000..c11877dc --- /dev/null +++ b/test/persistence/d1-runtime-persistence.test.ts @@ -0,0 +1,374 @@ +import { mkdtempSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import type { Task } from '@mindrail/contracts'; +import { describe, expect, it } from 'vitest'; + +import { PersistenceError } from '../../src/persistence/ports.ts'; +import { + agent, + auditEvent, + checkpoint, + goal, + leaseCandidate, + receipt, + session, + T0, + T1, + T2, + T3, + task, + workspace, +} from './fixtures.ts'; +import { openPersistence } from './setup.ts'; + +function databasePath(): string { + return join(mkdtempSync(join(tmpdir(), 'mindrail-persistence-')), 'runtime.sqlite'); +} + +async function seedExecution(path: string, options?: { secondSession?: boolean }) { + const opened = await openPersistence(path); + const { persistence } = opened; + await persistence.bootstrapWorkspace(workspace()); + await persistence.createGoal({ goal: goal() }); + await persistence.createTask({ task: task() }); + await persistence.createAgent({ agent: agent() }); + await persistence.createSession({ session: session() }); + if (options?.secondSession) { + await persistence.createAgent({ agent: agent('ws-a', 'agent-b') }); + await persistence.createSession({ session: session('ws-a', 'agent-b', 'session-b') }); + } + return opened; +} + +function expectCode(error: unknown, code: PersistenceError['code']): void { + expect(error).toBeInstanceOf(PersistenceError); + expect((error as PersistenceError).code).toBe(code); +} + +describe('D1RuntimePersistence durable semantics', () => { + it('reconstructs canonical state, fencing state, and receipts after adapter re-instantiation', async () => { + const path = databasePath(); + let { database, persistence } = await seedExecution(path); + const originalSnapshot = { + protocolVersion: '0.1', + commandId: 'cmd-claim-restart', + replayed: false, + result: { taskId: 'task-a', leaseId: 'lease-a', fencingToken: 1 }, + }; + + const claim = await persistence.claimTask({ + workspaceId: 'ws-a', + taskId: 'task-a', + sessionId: 'session-a', + expectedTaskRevision: 1, + lease: leaseCandidate(), + now: T0, + receipt: receipt( + 'cmd-claim-restart', + 'fingerprint-claim-restart', + originalSnapshot, + 'ws-a', + 'ClaimTask', + ), + auditEvent: auditEvent('ws-a', 'audit-claim-restart'), + }); + expect(claim.kind).toBe('committed'); + if (claim.kind !== 'committed') throw new Error('expected committed claim'); + expect(claim.value.lease.fencingToken).toBe(1); + + database.close(); + ({ database, persistence } = await openPersistence(path)); + + const state = await persistence.loadWorkspaceState('ws-a'); + expect(state?.workspace.id).toBe('ws-a'); + expect(state?.goals).toHaveLength(1); + expect(state?.tasks).toEqual([ + expect.objectContaining({ id: 'task-a', status: 'running', revision: 2 }), + ]); + expect(state?.leases).toEqual([ + expect.objectContaining({ id: 'lease-a', status: 'active', fencingToken: 1 }), + ]); + expect(state?.fencingCounters).toEqual({ 'task-a': 1 }); + + const storedReceipt = await persistence.getCommandReceipt('ws-a', 'cmd-claim-restart'); + expect(storedReceipt?.responseSnapshot).toEqual(originalSnapshot); + database.close(); + }); + + it('returns an immutable stored response on exact command replay without duplicating mutation', async () => { + const path = databasePath(); + const { database, persistence } = await openPersistence(path); + await persistence.bootstrapWorkspace(workspace()); + const originalSnapshot = { + protocolVersion: '0.1', + commandId: 'cmd-goal', + replayed: false, + result: { id: 'goal-a', revision: 1 }, + }; + const commandReceipt = receipt( + 'cmd-goal', + 'fingerprint-goal-a', + originalSnapshot, + 'ws-a', + 'CreateGoal', + ); + + const first = await persistence.createGoal({ goal: goal(), receipt: commandReceipt }); + const second = await persistence.createGoal({ goal: goal(), receipt: commandReceipt }); + + expect(first.kind).toBe('committed'); + expect(second.kind).toBe('replayed'); + if (second.kind !== 'replayed') throw new Error('expected replay'); + expect(second.receipt.responseSnapshot).toEqual(originalSnapshot); + + const mutableReplay = second.receipt.responseSnapshot as { result: { revision: number } }; + mutableReplay.result.revision = 99; + const storedAgain = await persistence.getCommandReceipt('ws-a', 'cmd-goal'); + expect(storedAgain?.responseSnapshot).toEqual(originalSnapshot); + + const state = await persistence.loadWorkspaceState('ws-a'); + expect(state?.goals).toHaveLength(1); + expect(state?.goals[0]).toEqual(expect.objectContaining({ id: 'goal-a', revision: 1 })); + database.close(); + }); + + it('rejects reuse of one workspace commandId with a different semantic fingerprint', async () => { + const path = databasePath(); + const { database, persistence } = await openPersistence(path); + await persistence.bootstrapWorkspace(workspace()); + await persistence.createGoal({ + goal: goal(), + receipt: receipt('cmd-same', 'fingerprint-a', { result: 'a' }, 'ws-a', 'CreateGoal'), + }); + + await expect( + persistence.createGoal({ + goal: goal('ws-a', 'goal-b'), + receipt: receipt('cmd-same', 'fingerprint-b', { result: 'b' }, 'ws-a', 'CreateGoal'), + }), + ).rejects.toSatisfy((error: unknown) => { + expectCode(error, 'IDEMPOTENCY_CONFLICT'); + return true; + }); + + expect((await persistence.loadWorkspaceState('ws-a'))?.goals.map((item) => item.id)).toEqual([ + 'goal-a', + ]); + database.close(); + }); + + it('serializes competing claims so only one session obtains effective ownership', async () => { + const path = databasePath(); + const { database, persistence } = await seedExecution(path, { secondSession: true }); + + const claims = await Promise.allSettled([ + persistence.claimTask({ + workspaceId: 'ws-a', + taskId: 'task-a', + sessionId: 'session-a', + expectedTaskRevision: 1, + lease: leaseCandidate('ws-a', 'task-a', 'session-a', 'lease-a', T3), + now: T0, + }), + persistence.claimTask({ + workspaceId: 'ws-a', + taskId: 'task-a', + sessionId: 'session-b', + expectedTaskRevision: 1, + lease: leaseCandidate('ws-a', 'task-a', 'session-b', 'lease-b', T3), + now: T0, + }), + ]); + + expect(claims.filter((result) => result.status === 'fulfilled')).toHaveLength(1); + expect(claims.filter((result) => result.status === 'rejected')).toHaveLength(1); + const rejection = claims.find((result) => result.status === 'rejected'); + if (!rejection || rejection.status !== 'rejected') throw new Error('expected rejected claim'); + expectCode(rejection.reason, 'CONFLICT'); + + const state = await persistence.loadWorkspaceState('ws-a'); + expect(state?.leases.filter((lease) => lease.status === 'active')).toHaveLength(1); + expect(state?.fencingCounters['task-a']).toBe(1); + database.close(); + }); + + it('allocates a strictly higher fencing token after expired-lease recovery', async () => { + const path = databasePath(); + const { database, persistence } = await seedExecution(path, { secondSession: true }); + + const first = await persistence.claimTask({ + workspaceId: 'ws-a', + taskId: 'task-a', + sessionId: 'session-a', + expectedTaskRevision: 1, + lease: leaseCandidate('ws-a', 'task-a', 'session-a', 'lease-a', T1), + now: T0, + }); + if (first.kind !== 'committed') throw new Error('expected first claim'); + + const second = await persistence.claimTask({ + workspaceId: 'ws-a', + taskId: 'task-a', + sessionId: 'session-b', + expectedTaskRevision: 2, + lease: leaseCandidate('ws-a', 'task-a', 'session-b', 'lease-b', T3), + now: T1, + }); + if (second.kind !== 'committed') throw new Error('expected recovery claim'); + + expect(first.value.lease.fencingToken).toBe(1); + expect(second.value.lease.fencingToken).toBeGreaterThan(first.value.lease.fencingToken); + expect(second.value.lease.fencingToken).toBe(2); + + const state = await persistence.loadWorkspaceState('ws-a'); + expect(state?.leases).toEqual( + expect.arrayContaining([ + expect.objectContaining({ id: 'lease-a', status: 'expired', fencingToken: 1 }), + expect.objectContaining({ id: 'lease-b', status: 'active', fencingToken: 2 }), + ]), + ); + expect(state?.fencingCounters['task-a']).toBe(2); + database.close(); + }); + + it('makes a stale expected Task revision lose deterministically', async () => { + const path = databasePath(); + const { database, persistence } = await seedExecution(path); + const current = task(); + const updated: Task = { + ...current, + revision: 2, + updatedAt: T1, + status: 'blocked', + statusReason: { code: 'task.waiting', summary: 'Waiting for deterministic input.' }, + }; + + await expect( + persistence.updateTask({ task: updated, expectedRevision: 2 }), + ).rejects.toSatisfy((error: unknown) => { + expectCode(error, 'REVISION_MISMATCH'); + return true; + }); + expect((await persistence.loadWorkspaceState('ws-a'))?.tasks[0]).toEqual(current); + database.close(); + }); + + it('rejects stale fences for checkpoint and completion after reassignment', async () => { + const path = databasePath(); + const { database, persistence } = await seedExecution(path, { secondSession: true }); + const first = await persistence.claimTask({ + workspaceId: 'ws-a', + taskId: 'task-a', + sessionId: 'session-a', + expectedTaskRevision: 1, + lease: leaseCandidate('ws-a', 'task-a', 'session-a', 'lease-a', T1), + now: T0, + }); + if (first.kind !== 'committed') throw new Error('expected first claim'); + const second = await persistence.claimTask({ + workspaceId: 'ws-a', + taskId: 'task-a', + sessionId: 'session-b', + expectedTaskRevision: 2, + lease: leaseCandidate('ws-a', 'task-a', 'session-b', 'lease-b', T3), + now: T1, + }); + if (second.kind !== 'committed') throw new Error('expected second claim'); + + await expect( + persistence.appendCheckpoint({ + checkpoint: checkpoint('ws-a', 'task-a', 'session-a', 'lease-a', 1, 'checkpoint-stale', T2), + now: T2, + }), + ).rejects.toSatisfy((error: unknown) => { + expectCode(error, 'STALE_AUTHORITY'); + return true; + }); + + const completedTask: Task = { + ...second.value.task, + revision: second.value.task.revision + 1, + updatedAt: T2, + status: 'succeeded', + }; + const staleReleasedLease = { + ...first.value.lease, + revision: first.value.lease.revision + 1, + updatedAt: T2, + status: 'released' as const, + }; + await expect( + persistence.completeTask({ + workspaceId: 'ws-a', + task: completedTask, + lease: staleReleasedLease, + checkpoint: { + ...checkpoint('ws-a', 'task-a', 'session-a', 'lease-a', 1, 'checkpoint-result-stale', T2), + kind: 'result', + }, + expectedTaskRevision: second.value.task.revision, + now: T2, + }), + ).rejects.toSatisfy((error: unknown) => { + expectCode(error, 'STALE_AUTHORITY'); + return true; + }); + + expect((await persistence.loadWorkspaceState('ws-a'))?.tasks[0]).toEqual( + expect.objectContaining({ status: 'running', revision: 2 }), + ); + database.close(); + }); + + it('rolls back Task, Lease, fencing, audit, and receipt state when a claim batch fails', async () => { + const path = databasePath(); + const { database, persistence } = await seedExecution(path); + const duplicateAudit = auditEvent('ws-a', 'audit-duplicate', T1); + await persistence.appendAuditEvent({ auditEvent: duplicateAudit }); + + await expect( + persistence.claimTask({ + workspaceId: 'ws-a', + taskId: 'task-a', + sessionId: 'session-a', + expectedTaskRevision: 1, + lease: leaseCandidate('ws-a', 'task-a', 'session-a', 'lease-failed', T3), + now: T0, + auditEvent: duplicateAudit, + receipt: receipt( + 'cmd-failed-claim', + 'fingerprint-failed-claim', + { result: 'must-not-exist' }, + 'ws-a', + 'ClaimTask', + ), + }), + ).rejects.toSatisfy((error: unknown) => { + expectCode(error, 'INTEGRITY_ERROR'); + return true; + }); + + let state = await persistence.loadWorkspaceState('ws-a'); + expect(state?.tasks[0]).toEqual(expect.objectContaining({ status: 'ready', revision: 1 })); + expect(state?.leases).toHaveLength(0); + expect(state?.fencingCounters['task-a']).toBe(0); + expect(await persistence.getCommandReceipt('ws-a', 'cmd-failed-claim')).toBeUndefined(); + + const recovered = await persistence.claimTask({ + workspaceId: 'ws-a', + taskId: 'task-a', + sessionId: 'session-a', + expectedTaskRevision: 1, + lease: leaseCandidate('ws-a', 'task-a', 'session-a', 'lease-good', T3), + now: T0, + auditEvent: auditEvent('ws-a', 'audit-good', T1), + }); + if (recovered.kind !== 'committed') throw new Error('expected recovered claim'); + expect(recovered.value.lease.fencingToken).toBe(1); + state = await persistence.loadWorkspaceState('ws-a'); + expect(state?.fencingCounters['task-a']).toBe(1); + database.close(); + }); +}); diff --git a/test/persistence/d1-sqlite-harness.ts b/test/persistence/d1-sqlite-harness.ts new file mode 100644 index 00000000..a2bf0ba0 --- /dev/null +++ b/test/persistence/d1-sqlite-harness.ts @@ -0,0 +1,100 @@ +import { DatabaseSync, type SQLInputValue } from 'node:sqlite'; + +import type { + D1DatabaseLike, + D1PreparedStatementLike, + D1ResultLike, +} from '../../src/persistence/cloudflare/d1-types.ts'; + +export class SqliteD1Database implements D1DatabaseLike { + private readonly database: DatabaseSync; + + constructor(path: string) { + this.database = new DatabaseSync(path); + this.database.exec('PRAGMA foreign_keys = ON;'); + } + + prepare(sql: string): D1PreparedStatementLike { + return new SqliteD1PreparedStatement(this.database, sql); + } + + async batch(statements: D1PreparedStatementLike[]): Promise { + this.database.exec('BEGIN IMMEDIATE;'); + try { + const results = statements.map((statement) => { + if (!(statement instanceof SqliteD1PreparedStatement)) { + throw new TypeError('SqliteD1Database can only batch its own prepared statements.'); + } + return statement.runSync(); + }); + this.database.exec('COMMIT;'); + return results; + } catch (error) { + this.database.exec('ROLLBACK;'); + throw error; + } + } + + async exec(sql: string): Promise { + this.database.exec(sql); + } + + close(): void { + this.database.close(); + } +} + +class SqliteD1PreparedStatement implements D1PreparedStatementLike { + constructor( + private readonly database: DatabaseSync, + private readonly sql: string, + private readonly parameters: readonly SQLInputValue[] = [], + ) {} + + bind(...values: unknown[]): D1PreparedStatementLike { + return new SqliteD1PreparedStatement( + this.database, + this.sql, + values.map(toSqlInputValue), + ); + } + + async first(): Promise { + const row = this.database.prepare(this.sql).get(...this.parameters); + return (row as T | undefined) ?? null; + } + + async all(): Promise> { + const results = this.database.prepare(this.sql).all(...this.parameters) as T[]; + return { success: true, results, meta: { changes: 0 } }; + } + + async run(): Promise> { + return this.runSync() as D1ResultLike; + } + + runSync(): D1ResultLike { + const result = this.database.prepare(this.sql).run(...this.parameters); + return { + success: true, + results: [], + meta: { changes: Number(result.changes) }, + }; + } +} + +function toSqlInputValue(value: unknown): SQLInputValue { + if ( + value === null || + typeof value === 'string' || + typeof value === 'number' || + typeof value === 'bigint' || + value instanceof Uint8Array + ) { + return value; + } + if (typeof value === 'boolean') { + return value ? 1 : 0; + } + throw new TypeError(`Unsupported SQLite bind value: ${String(value)}`); +} diff --git a/test/persistence/fixtures.ts b/test/persistence/fixtures.ts new file mode 100644 index 00000000..f11681ca --- /dev/null +++ b/test/persistence/fixtures.ts @@ -0,0 +1,230 @@ +import type { + Agent, + AuditEvent, + Checkpoint, + Goal, + Lease, + PermissionDecision, + PermissionRequest, + Session, + Task, + Workspace, +} from '@mindrail/contracts'; + +import type { CommandReceiptInput } from '../../src/persistence/ports.ts'; + +export const T0 = '2026-08-29T12:00:00.000Z'; +export const T1 = '2026-08-29T12:05:00.000Z'; +export const T2 = '2026-08-29T12:10:00.000Z'; +export const T3 = '2026-08-29T12:15:00.000Z'; + +export function workspace(id = 'ws-a'): Workspace { + return { + id, + revision: 1, + createdAt: T0, + updatedAt: T0, + name: `Workspace ${id}`, + status: 'active', + }; +} + +export function goal(workspaceId = 'ws-a', id = 'goal-a'): Goal { + return { + id, + workspaceId, + revision: 1, + createdAt: T0, + updatedAt: T0, + title: `Goal ${id}`, + objective: 'Prove durable persistence semantics.', + successCriteria: ['Persistence state is durable.'], + status: 'active', + }; +} + +export function task(workspaceId = 'ws-a', goalId = 'goal-a', id = 'task-a'): Task { + return { + id, + workspaceId, + goalId, + revision: 1, + createdAt: T0, + updatedAt: T0, + title: `Task ${id}`, + objective: 'Execute one durable task.', + acceptanceCriteria: ['The task can be recovered safely.'], + requiredCapabilities: ['repo.write'], + dependencyTaskIds: [], + status: 'ready', + }; +} + +export function agent(workspaceId = 'ws-a', id = 'agent-a'): Agent { + return { + id, + workspaceId, + revision: 1, + createdAt: T0, + updatedAt: T0, + displayName: `Agent ${id}`, + status: 'active', + capabilities: ['repo.write'], + }; +} + +export function session( + workspaceId = 'ws-a', + agentId = 'agent-a', + id = 'session-a', +): Session { + return { + id, + workspaceId, + agentId, + revision: 1, + createdAt: T0, + updatedAt: T0, + status: 'active', + lastSeenAt: T0, + }; +} + +export function leaseCandidate( + workspaceId = 'ws-a', + taskId = 'task-a', + sessionId = 'session-a', + id = 'lease-a', + expiresAt = T2, +): Omit { + return { + id, + workspaceId, + taskId, + sessionId, + revision: 1, + createdAt: T0, + updatedAt: T0, + status: 'active', + expiresAt, + }; +} + +export function checkpoint( + workspaceId = 'ws-a', + taskId = 'task-a', + sessionId = 'session-a', + leaseId = 'lease-a', + fencingToken = 1, + id = 'checkpoint-a', + createdAt = T1, +): Checkpoint { + return { + id, + workspaceId, + taskId, + sessionId, + leaseId, + fencingToken, + createdAt, + kind: 'progress', + summary: `Checkpoint ${id}`, + evidence: [], + progressPercent: 50, + }; +} + +export function auditEvent( + workspaceId = 'ws-a', + id = 'audit-a', + createdAt = T1, + subjectId = 'task-a', +): AuditEvent { + return { + id, + workspaceId, + createdAt, + eventType: 'task.claimed', + actor: { type: 'system', id: 'system-runtime' }, + subject: { type: 'task', id: subjectId }, + correlationId: `corr-${id}`, + }; +} + +export function permissionRequest( + workspaceId = 'ws-a', + taskId = 'task-a', + sessionId = 'session-a', + leaseId = 'lease-a', + fencingToken = 1, + id = 'permission-request-a', +): PermissionRequest { + return { + id, + workspaceId, + taskId, + sessionId, + leaseId, + fencingToken, + createdAt: T1, + permission: 'repo.write', + justification: 'The task requires a repository write.', + }; +} + +export function policyDecision( + workspaceId = 'ws-a', + requestId = 'permission-request-a', + id = 'permission-decision-a', +): PermissionDecision { + return { + id, + workspaceId, + requestId, + createdAt: T1, + sequence: 1, + outcome: 'HUMAN_REQUIRED', + basis: 'policy', + decidedBy: { type: 'system', id: 'system-policy' }, + reasonCode: 'policy.human_required', + policyRef: { id: 'policy-default', version: '1' }, + }; +} + +export function humanDecision( + workspaceId = 'ws-a', + requestId = 'permission-request-a', + supersedesDecisionId = 'permission-decision-a', + id = 'permission-decision-b', +): PermissionDecision { + return { + id, + workspaceId, + requestId, + createdAt: T2, + sequence: 2, + outcome: 'ALLOW', + basis: 'human', + decidedBy: { type: 'human', id: 'human-reviewer' }, + reasonCode: 'human.approved', + supersedesDecisionId, + }; +} + +export function receipt( + commandId: string, + semanticFingerprint: string, + responseSnapshot: unknown, + workspaceId = 'ws-a', + command = 'TestCommand', +): CommandReceiptInput { + return { + workspaceId, + commandId, + command, + semanticFingerprint, + outcomeKind: 'result', + responseSnapshot, + createdAt: T1, + }; +} diff --git a/test/persistence/setup.ts b/test/persistence/setup.ts new file mode 100644 index 00000000..bb8a61ab --- /dev/null +++ b/test/persistence/setup.ts @@ -0,0 +1,23 @@ +import { readFileSync } from 'node:fs'; + +import { D1RuntimePersistence } from '../../src/persistence/cloudflare/d1-runtime-persistence.ts'; +import { WorkspaceDurableObjectCoordinator } from '../../src/persistence/cloudflare/workspace-durable-object-coordinator.ts'; +import { persistenceCanonicalValidator } from './canonical-domain-validator.ts'; +import { SqliteD1Database } from './d1-sqlite-harness.ts'; + +const migrationUrl = new URL('../../migrations/0001_runtime_persistence.sql', import.meta.url); +const migrationSql = readFileSync(migrationUrl, 'utf8'); + +export async function openPersistence(path: string): Promise<{ + database: SqliteD1Database; + persistence: D1RuntimePersistence; +}> { + const database = new SqliteD1Database(path); + await database.exec(migrationSql); + const persistence = new D1RuntimePersistence({ + database, + coordinator: new WorkspaceDurableObjectCoordinator(), + validateCanonicalDomainRecord: persistenceCanonicalValidator, + }); + return { database, persistence }; +} From 74d0f7c1faa5c2b2acad42f1205f69562b25b9c6 Mon Sep 17 00:00:00 2001 From: tim8es <108188608+tim8es@users.noreply.github.com> Date: Sat, 29 Aug 2026 19:53:14 +0500 Subject: [PATCH 2/2] feat: add durable persistence ports and D1 reference --- migrations/0001_runtime_persistence.sql | 298 ++++ migrations/0002_lease_fencing_guards.sql | 36 + .../cloudflare/d1-runtime-persistence.ts | 1400 +++++++++++++++++ src/persistence/cloudflare/d1-types.ts | 20 + src/persistence/cloudflare/index.ts | 7 + .../workspace-durable-object-coordinator.ts | 39 + src/persistence/index.ts | 18 + src/persistence/ports.ts | 176 +++ test/persistence/d1-sqlite-harness.ts | 6 +- test/persistence/fixtures.ts | 6 +- test/persistence/setup.ts | 16 +- 11 files changed, 2008 insertions(+), 14 deletions(-) create mode 100644 migrations/0001_runtime_persistence.sql create mode 100644 migrations/0002_lease_fencing_guards.sql create mode 100644 src/persistence/cloudflare/d1-runtime-persistence.ts create mode 100644 src/persistence/cloudflare/d1-types.ts create mode 100644 src/persistence/cloudflare/index.ts create mode 100644 src/persistence/cloudflare/workspace-durable-object-coordinator.ts create mode 100644 src/persistence/index.ts create mode 100644 src/persistence/ports.ts diff --git a/migrations/0001_runtime_persistence.sql b/migrations/0001_runtime_persistence.sql new file mode 100644 index 00000000..2b7f8f19 --- /dev/null +++ b/migrations/0001_runtime_persistence.sql @@ -0,0 +1,298 @@ +PRAGMA foreign_keys = ON; + +CREATE TABLE IF NOT EXISTS schema_migrations ( + version TEXT PRIMARY KEY, + applied_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS workspaces ( + id TEXT PRIMARY KEY, + revision INTEGER NOT NULL CHECK (revision >= 1), + status TEXT NOT NULL CHECK (status IN ('active', 'archived')), + created_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)) +); + +CREATE TABLE IF NOT EXISTS goals ( + workspace_id TEXT NOT NULL, + id TEXT NOT NULL, + revision INTEGER NOT NULL CHECK (revision >= 1), + status TEXT NOT NULL CHECK (status IN ('active', 'succeeded', 'failed', 'cancelled')), + created_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + PRIMARY KEY (workspace_id, id), + UNIQUE (workspace_id, id, status), + FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS tasks ( + workspace_id TEXT NOT NULL, + id TEXT NOT NULL, + goal_id TEXT NOT NULL, + revision INTEGER NOT NULL CHECK (revision >= 1), + status TEXT NOT NULL CHECK ( + status IN ('pending', 'ready', 'running', 'blocked', 'succeeded', 'failed', 'cancelled') + ), + created_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + PRIMARY KEY (workspace_id, id), + UNIQUE (workspace_id, id, goal_id), + FOREIGN KEY (workspace_id, goal_id) REFERENCES goals(workspace_id, id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS task_dependencies ( + workspace_id TEXT NOT NULL, + goal_id TEXT NOT NULL, + task_id TEXT NOT NULL, + dependency_task_id TEXT NOT NULL, + PRIMARY KEY (workspace_id, task_id, dependency_task_id), + CHECK (task_id <> dependency_task_id), + FOREIGN KEY (workspace_id, task_id, goal_id) + REFERENCES tasks(workspace_id, id, goal_id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, dependency_task_id, goal_id) + REFERENCES tasks(workspace_id, id, goal_id) ON DELETE RESTRICT +); + +CREATE TABLE IF NOT EXISTS task_required_capabilities ( + workspace_id TEXT NOT NULL, + task_id TEXT NOT NULL, + capability TEXT NOT NULL, + PRIMARY KEY (workspace_id, task_id, capability), + FOREIGN KEY (workspace_id, task_id) REFERENCES tasks(workspace_id, id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS task_fencing_counters ( + workspace_id TEXT NOT NULL, + task_id TEXT NOT NULL, + last_fencing_token INTEGER NOT NULL CHECK (last_fencing_token >= 0), + PRIMARY KEY (workspace_id, task_id), + FOREIGN KEY (workspace_id, task_id) REFERENCES tasks(workspace_id, id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS agents ( + workspace_id TEXT NOT NULL, + id TEXT NOT NULL, + revision INTEGER NOT NULL CHECK (revision >= 1), + status TEXT NOT NULL CHECK (status IN ('active', 'disabled')), + created_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + PRIMARY KEY (workspace_id, id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS agent_capabilities ( + workspace_id TEXT NOT NULL, + agent_id TEXT NOT NULL, + capability TEXT NOT NULL, + PRIMARY KEY (workspace_id, agent_id, capability), + FOREIGN KEY (workspace_id, agent_id) REFERENCES agents(workspace_id, id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS sessions ( + workspace_id TEXT NOT NULL, + id TEXT NOT NULL, + agent_id TEXT NOT NULL, + revision INTEGER NOT NULL CHECK (revision >= 1), + status TEXT NOT NULL CHECK (status IN ('active', 'ended', 'expired')), + created_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + last_seen_at_ms INTEGER NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + PRIMARY KEY (workspace_id, id), + FOREIGN KEY (workspace_id, agent_id) REFERENCES agents(workspace_id, id) ON DELETE RESTRICT +); + +CREATE TABLE IF NOT EXISTS leases ( + workspace_id TEXT NOT NULL, + id TEXT NOT NULL, + task_id TEXT NOT NULL, + session_id TEXT NOT NULL, + revision INTEGER NOT NULL CHECK (revision >= 1), + status TEXT NOT NULL CHECK (status IN ('active', 'released', 'expired', 'revoked')), + fencing_token INTEGER NOT NULL CHECK (fencing_token >= 1), + created_at_ms INTEGER NOT NULL, + updated_at_ms INTEGER NOT NULL, + expires_at_ms INTEGER NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + PRIMARY KEY (workspace_id, id), + UNIQUE (workspace_id, id, task_id, session_id), + FOREIGN KEY (workspace_id, task_id) REFERENCES tasks(workspace_id, id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, session_id) REFERENCES sessions(workspace_id, id) ON DELETE RESTRICT +); + +CREATE UNIQUE INDEX IF NOT EXISTS uq_leases_one_active_per_task + ON leases(workspace_id, task_id) + WHERE status = 'active'; + +CREATE TABLE IF NOT EXISTS checkpoints ( + workspace_id TEXT NOT NULL, + id TEXT NOT NULL, + task_id TEXT NOT NULL, + session_id TEXT NOT NULL, + lease_id TEXT NOT NULL, + fencing_token INTEGER NOT NULL CHECK (fencing_token >= 1), + created_at_ms INTEGER NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + PRIMARY KEY (workspace_id, id), + FOREIGN KEY (workspace_id, task_id) REFERENCES tasks(workspace_id, id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, session_id) REFERENCES sessions(workspace_id, id) ON DELETE RESTRICT, + FOREIGN KEY (workspace_id, lease_id, task_id, session_id) + REFERENCES leases(workspace_id, id, task_id, session_id) ON DELETE RESTRICT +); + +CREATE TABLE IF NOT EXISTS permission_requests ( + workspace_id TEXT NOT NULL, + id TEXT NOT NULL, + task_id TEXT NOT NULL, + session_id TEXT NOT NULL, + lease_id TEXT NOT NULL, + fencing_token INTEGER NOT NULL CHECK (fencing_token >= 1), + permission TEXT NOT NULL, + created_at_ms INTEGER NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + PRIMARY KEY (workspace_id, id), + FOREIGN KEY (workspace_id, task_id) REFERENCES tasks(workspace_id, id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, session_id) REFERENCES sessions(workspace_id, id) ON DELETE RESTRICT, + FOREIGN KEY (workspace_id, lease_id, task_id, session_id) + REFERENCES leases(workspace_id, id, task_id, session_id) ON DELETE RESTRICT +); + +CREATE TABLE IF NOT EXISTS permission_decisions ( + workspace_id TEXT NOT NULL, + id TEXT NOT NULL, + request_id TEXT NOT NULL, + sequence INTEGER NOT NULL CHECK (sequence >= 1), + outcome TEXT NOT NULL CHECK (outcome IN ('ALLOW', 'DENY', 'HUMAN_REQUIRED')), + basis TEXT NOT NULL CHECK (basis IN ('policy', 'human')), + supersedes_decision_id TEXT, + created_at_ms INTEGER NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + PRIMARY KEY (workspace_id, id), + UNIQUE (workspace_id, request_id, sequence), + FOREIGN KEY (workspace_id, request_id) REFERENCES permission_requests(workspace_id, id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, supersedes_decision_id) + REFERENCES permission_decisions(workspace_id, id) ON DELETE RESTRICT +); + +CREATE TABLE IF NOT EXISTS permission_heads ( + workspace_id TEXT NOT NULL, + request_id TEXT NOT NULL, + latest_decision_id TEXT NOT NULL, + latest_sequence INTEGER NOT NULL CHECK (latest_sequence >= 1), + latest_outcome TEXT NOT NULL CHECK (latest_outcome IN ('ALLOW', 'DENY', 'HUMAN_REQUIRED')), + PRIMARY KEY (workspace_id, request_id), + FOREIGN KEY (workspace_id, request_id) REFERENCES permission_requests(workspace_id, id) ON DELETE CASCADE, + FOREIGN KEY (workspace_id, latest_decision_id) + REFERENCES permission_decisions(workspace_id, id) ON DELETE RESTRICT +); + +CREATE TABLE IF NOT EXISTS audit_events ( + workspace_id TEXT NOT NULL, + id TEXT NOT NULL, + event_type TEXT NOT NULL, + subject_type TEXT NOT NULL, + subject_id TEXT NOT NULL, + created_at_ms INTEGER NOT NULL, + record_json TEXT NOT NULL CHECK (json_valid(record_json)), + PRIMARY KEY (workspace_id, id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE +); + +CREATE TABLE IF NOT EXISTS command_receipts ( + workspace_id TEXT NOT NULL, + command_id TEXT NOT NULL, + command_discriminator TEXT NOT NULL, + semantic_fingerprint TEXT NOT NULL, + outcome_kind TEXT NOT NULL CHECK (outcome_kind IN ('result', 'error')), + response_snapshot_json TEXT NOT NULL CHECK (json_valid(response_snapshot_json)), + created_at_ms INTEGER NOT NULL, + expires_at_ms INTEGER, + PRIMARY KEY (workspace_id, command_id), + FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE +); + +CREATE INDEX IF NOT EXISTS idx_tasks_workspace_goal_status + ON tasks(workspace_id, goal_id, status, id); +CREATE INDEX IF NOT EXISTS idx_tasks_workspace_status + ON tasks(workspace_id, status, updated_at_ms, id); +CREATE INDEX IF NOT EXISTS idx_sessions_workspace_status_last_seen + ON sessions(workspace_id, status, last_seen_at_ms, id); +CREATE INDEX IF NOT EXISTS idx_leases_workspace_status_expiry + ON leases(workspace_id, status, expires_at_ms, id); +CREATE INDEX IF NOT EXISTS idx_checkpoints_task_order + ON checkpoints(workspace_id, task_id, created_at_ms, id); +CREATE INDEX IF NOT EXISTS idx_permission_requests_task_order + ON permission_requests(workspace_id, task_id, created_at_ms, id); +CREATE INDEX IF NOT EXISTS idx_permission_heads_pending + ON permission_heads(workspace_id, latest_outcome, request_id); +CREATE INDEX IF NOT EXISTS idx_audit_events_workspace_order + ON audit_events(workspace_id, created_at_ms, id); +CREATE INDEX IF NOT EXISTS idx_receipts_workspace_created + ON command_receipts(workspace_id, created_at_ms, command_id); + +CREATE TRIGGER IF NOT EXISTS checkpoints_append_only_update +BEFORE UPDATE ON checkpoints +BEGIN + SELECT RAISE(ABORT, 'checkpoints are append-only'); +END; + +CREATE TRIGGER IF NOT EXISTS checkpoints_append_only_delete +BEFORE DELETE ON checkpoints +BEGIN + SELECT RAISE(ABORT, 'checkpoints are append-only'); +END; + +CREATE TRIGGER IF NOT EXISTS permission_requests_append_only_update +BEFORE UPDATE ON permission_requests +BEGIN + SELECT RAISE(ABORT, 'permission requests are append-only'); +END; + +CREATE TRIGGER IF NOT EXISTS permission_requests_append_only_delete +BEFORE DELETE ON permission_requests +BEGIN + SELECT RAISE(ABORT, 'permission requests are append-only'); +END; + +CREATE TRIGGER IF NOT EXISTS permission_decisions_append_only_update +BEFORE UPDATE ON permission_decisions +BEGIN + SELECT RAISE(ABORT, 'permission decisions are append-only'); +END; + +CREATE TRIGGER IF NOT EXISTS permission_decisions_append_only_delete +BEFORE DELETE ON permission_decisions +BEGIN + SELECT RAISE(ABORT, 'permission decisions are append-only'); +END; + +CREATE TRIGGER IF NOT EXISTS audit_events_append_only_update +BEFORE UPDATE ON audit_events +BEGIN + SELECT RAISE(ABORT, 'audit events are append-only'); +END; + +CREATE TRIGGER IF NOT EXISTS audit_events_append_only_delete +BEFORE DELETE ON audit_events +BEGIN + SELECT RAISE(ABORT, 'audit events are append-only'); +END; + +CREATE TRIGGER IF NOT EXISTS command_receipts_immutable_update +BEFORE UPDATE ON command_receipts +BEGIN + SELECT RAISE(ABORT, 'command receipts are immutable'); +END; + +CREATE TRIGGER IF NOT EXISTS command_receipts_immutable_delete +BEFORE DELETE ON command_receipts +BEGIN + SELECT RAISE(ABORT, 'command receipts are immutable'); +END; + +INSERT OR IGNORE INTO schema_migrations(version, applied_at) +VALUES ('0001_runtime_persistence', datetime('now')); diff --git a/migrations/0002_lease_fencing_guards.sql b/migrations/0002_lease_fencing_guards.sql new file mode 100644 index 00000000..b9ab13bc --- /dev/null +++ b/migrations/0002_lease_fencing_guards.sql @@ -0,0 +1,36 @@ +PRAGMA foreign_keys = ON; + +CREATE TRIGGER IF NOT EXISTS leases_fencing_must_match_counter +BEFORE INSERT ON leases +WHEN NEW.status = 'active' +BEGIN + SELECT CASE + WHEN ( + SELECT last_fencing_token + FROM task_fencing_counters + WHERE workspace_id = NEW.workspace_id AND task_id = NEW.task_id + ) IS NULL + THEN RAISE(ABORT, 'active lease requires fencing allocation state') + END; + SELECT CASE + WHEN ( + SELECT last_fencing_token + FROM task_fencing_counters + WHERE workspace_id = NEW.workspace_id AND task_id = NEW.task_id + ) <> NEW.fencing_token + THEN RAISE(ABORT, 'active lease fencing token must match allocated counter') + END; + SELECT CASE + WHEN EXISTS ( + SELECT 1 + FROM leases + WHERE workspace_id = NEW.workspace_id + AND task_id = NEW.task_id + AND fencing_token >= NEW.fencing_token + ) + THEN RAISE(ABORT, 'lease fencing token must strictly increase') + END; +END; + +INSERT OR IGNORE INTO schema_migrations(version, applied_at) +VALUES ('0002_lease_fencing_guards', datetime('now')); diff --git a/src/persistence/cloudflare/d1-runtime-persistence.ts b/src/persistence/cloudflare/d1-runtime-persistence.ts new file mode 100644 index 00000000..92a969ce --- /dev/null +++ b/src/persistence/cloudflare/d1-runtime-persistence.ts @@ -0,0 +1,1400 @@ +import type { + Agent, + AuditEvent, + Checkpoint, + Goal, + Lease, + PermissionDecision, + PermissionRequest, + Session, + Task, + Workspace, +} from '@mindrail/contracts'; + +import { + PersistenceError, + type ClaimTaskCommitInput, + type ClaimTaskCommitValue, + type CommandReceiptInput, + type CompleteTaskCommitInput, + type CompleteTaskCommitValue, + type DurableRuntimePersistence, + type MutationCommitResult, + type PendingHumanPermission, + type PersistenceDomainTarget, + type PersistenceDomainValidator, + type StoredCommandReceipt, + type WorkspaceMutationCoordinator, + type WorkspaceStateSnapshot, +} from '../ports.ts'; +import type { D1DatabaseLike, D1PreparedStatementLike, D1ResultLike } from './d1-types.ts'; + +const MAX_RECEIPT_SNAPSHOT_BYTES = 64 * 1024; + +interface D1RuntimePersistenceOptions { + database: D1DatabaseLike; + coordinator: WorkspaceMutationCoordinator; + validateCanonicalDomainRecord: PersistenceDomainValidator; +} + +interface RecordRow { + record_json: string; +} + +interface CounterRow { + task_id: string; + last_fencing_token: number; +} + +interface ReceiptRow { + workspace_id: string; + command_id: string; + command_discriminator: string; + semantic_fingerprint: string; + outcome_kind: 'result' | 'error'; + response_snapshot_json: string; + created_at_ms: number; + expires_at_ms: number | null; +} + +interface PermissionHeadRow { + latest_decision_id: string; + latest_sequence: number; + latest_outcome: PermissionDecision['outcome']; +} + +interface PendingPermissionRow { + request_json: string; + decision_json: string; +} + +export class D1RuntimePersistence implements DurableRuntimePersistence { + private readonly database: D1DatabaseLike; + private readonly coordinator: WorkspaceMutationCoordinator; + private readonly validateCanonicalDomainRecord: PersistenceDomainValidator; + + constructor(options: D1RuntimePersistenceOptions) { + this.database = options.database; + this.coordinator = options.coordinator; + this.validateCanonicalDomainRecord = options.validateCanonicalDomainRecord; + } + + async bootstrapWorkspace(workspace: Workspace): Promise { + this.assertCanonical('Workspace', workspace); + await this.coordinator.runSerialized(workspace.id, async () => { + await this.run( + this.database + .prepare( + `INSERT INTO workspaces( + id, revision, status, created_at_ms, updated_at_ms, record_json + ) VALUES (?, ?, ?, ?, ?, ?)`, + ) + .bind( + workspace.id, + workspace.revision, + workspace.status, + timestampMs(workspace.createdAt, 'Workspace.createdAt'), + timestampMs(workspace.updatedAt, 'Workspace.updatedAt'), + serializeJson(workspace, 'Workspace'), + ), + 'bootstrap Workspace', + ); + }); + } + + async createAgent(input: { agent: Agent }): Promise { + const { agent } = input; + this.assertCanonical('Agent', agent); + await this.coordinator.runSerialized(agent.workspaceId, async () => { + await this.requireWorkspace(agent.workspaceId); + const statements: D1PreparedStatementLike[] = [ + this.database + .prepare( + `INSERT INTO agents( + workspace_id, id, revision, status, created_at_ms, updated_at_ms, record_json + ) VALUES (?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + agent.workspaceId, + agent.id, + agent.revision, + agent.status, + timestampMs(agent.createdAt, 'Agent.createdAt'), + timestampMs(agent.updatedAt, 'Agent.updatedAt'), + serializeJson(agent, 'Agent'), + ), + ...agent.capabilities.map((capability) => + this.database + .prepare( + `INSERT INTO agent_capabilities(workspace_id, agent_id, capability) + VALUES (?, ?, ?)`, + ) + .bind(agent.workspaceId, agent.id, capability), + ), + ]; + await this.batch(statements, 'create Agent'); + }); + } + + async createSession(input: { session: Session }): Promise { + const { session } = input; + this.assertCanonical('Session', session); + await this.coordinator.runSerialized(session.workspaceId, async () => { + await this.requireWorkspace(session.workspaceId); + const agent = await this.getAgent(session.workspaceId, session.agentId); + if (!agent) { + throw new PersistenceError('NOT_FOUND', `Agent ${session.agentId} was not found.`); + } + await this.run( + this.database + .prepare( + `INSERT INTO sessions( + workspace_id, id, agent_id, revision, status, created_at_ms, updated_at_ms, + last_seen_at_ms, record_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + session.workspaceId, + session.id, + session.agentId, + session.revision, + session.status, + timestampMs(session.createdAt, 'Session.createdAt'), + timestampMs(session.updatedAt, 'Session.updatedAt'), + timestampMs(session.lastSeenAt, 'Session.lastSeenAt'), + serializeJson(session, 'Session'), + ), + 'create Session', + ); + }); + } + + async createGoal(input: { + goal: Goal; + receipt?: CommandReceiptInput; + auditEvent?: AuditEvent; + }): Promise> { + const { goal } = input; + this.assertCanonical('Goal', goal); + this.assertRelatedAudit(goal.workspaceId, input.auditEvent); + this.assertReceipt(goal.workspaceId, input.receipt); + + return this.coordinator.runSerialized(goal.workspaceId, async () => { + const replay = await this.resolveReceipt(input.receipt); + if (replay) return replay; + await this.requireWorkspace(goal.workspaceId); + + const statements: D1PreparedStatementLike[] = [ + this.database + .prepare( + `INSERT INTO goals( + workspace_id, id, revision, status, created_at_ms, updated_at_ms, record_json + ) VALUES (?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + goal.workspaceId, + goal.id, + goal.revision, + goal.status, + timestampMs(goal.createdAt, 'Goal.createdAt'), + timestampMs(goal.updatedAt, 'Goal.updatedAt'), + serializeJson(goal, 'Goal'), + ), + ]; + this.pushAuditStatement(statements, input.auditEvent); + this.pushReceiptStatement(statements, input.receipt); + await this.batch(statements, 'create Goal'); + return { kind: 'committed', value: clone(goal) }; + }); + } + + async createTask(input: { + task: Task; + receipt?: CommandReceiptInput; + auditEvent?: AuditEvent; + }): Promise> { + const { task } = input; + this.assertCanonical('Task', task); + this.assertRelatedAudit(task.workspaceId, input.auditEvent); + this.assertReceipt(task.workspaceId, input.receipt); + + return this.coordinator.runSerialized(task.workspaceId, async () => { + const replay = await this.resolveReceipt(input.receipt); + if (replay) return replay; + await this.requireWorkspace(task.workspaceId); + const parentGoal = await this.getGoal(task.workspaceId, task.goalId); + if (!parentGoal) { + throw new PersistenceError('NOT_FOUND', `Goal ${task.goalId} was not found.`); + } + if (parentGoal.status !== 'active') { + throw new PersistenceError( + 'INVALID_STATE_TRANSITION', + `Goal ${parentGoal.id} is terminal and cannot accept Task ${task.id}.`, + ); + } + await this.assertDependencies(task); + + const statements: D1PreparedStatementLike[] = [ + this.database + .prepare( + `INSERT INTO tasks( + workspace_id, id, goal_id, revision, status, created_at_ms, updated_at_ms, record_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + task.workspaceId, + task.id, + task.goalId, + task.revision, + task.status, + timestampMs(task.createdAt, 'Task.createdAt'), + timestampMs(task.updatedAt, 'Task.updatedAt'), + serializeJson(task, 'Task'), + ), + ...task.requiredCapabilities.map((capability) => + this.database + .prepare( + `INSERT INTO task_required_capabilities(workspace_id, task_id, capability) + VALUES (?, ?, ?)`, + ) + .bind(task.workspaceId, task.id, capability), + ), + ...task.dependencyTaskIds.map((dependencyTaskId) => + this.database + .prepare( + `INSERT INTO task_dependencies( + workspace_id, goal_id, task_id, dependency_task_id + ) VALUES (?, ?, ?, ?)`, + ) + .bind(task.workspaceId, task.goalId, task.id, dependencyTaskId), + ), + this.database + .prepare( + `INSERT INTO task_fencing_counters(workspace_id, task_id, last_fencing_token) + VALUES (?, ?, 0)`, + ) + .bind(task.workspaceId, task.id), + ]; + this.pushAuditStatement(statements, input.auditEvent); + this.pushReceiptStatement(statements, input.receipt); + await this.batch(statements, 'create Task'); + return { kind: 'committed', value: clone(task) }; + }); + } + + async claimTask( + input: ClaimTaskCommitInput, + ): Promise> { + this.assertReceipt(input.workspaceId, input.receipt); + this.assertRelatedAudit(input.workspaceId, input.auditEvent); + const nowMs = timestampMs(input.now, 'claim now'); + + return this.coordinator.runSerialized(input.workspaceId, async () => { + const replay = await this.resolveReceipt(input.receipt); + if (replay) return replay; + await this.requireWorkspace(input.workspaceId); + + const task = await this.getTask(input.workspaceId, input.taskId); + if (!task) throw new PersistenceError('NOT_FOUND', `Task ${input.taskId} was not found.`); + const session = await this.getSession(input.workspaceId, input.sessionId); + if (!session) { + throw new PersistenceError('NOT_FOUND', `Session ${input.sessionId} was not found.`); + } + if (session.status !== 'active') { + throw new PersistenceError('CONFLICT', `Session ${session.id} is not active.`); + } + const agent = await this.getAgent(input.workspaceId, session.agentId); + if (!agent || agent.status !== 'active') { + throw new PersistenceError('CONFLICT', `Agent ${session.agentId} is not active.`); + } + if (!task.requiredCapabilities.every((capability) => agent.capabilities.includes(capability))) { + throw new PersistenceError( + 'CONFLICT', + `Session ${session.id} does not satisfy Task ${task.id} capabilities.`, + ); + } + + const activeLease = await this.getActiveLease(input.workspaceId, input.taskId); + const statements: D1PreparedStatementLike[] = []; + if (activeLease && timestampMs(activeLease.expiresAt, 'Lease.expiresAt') > nowMs) { + if (activeLease.sessionId === input.sessionId) { + this.pushReceiptStatement(statements, input.receipt); + if (statements.length > 0) await this.batch(statements, 'record duplicate claim receipt'); + return { + kind: 'committed', + value: { task: clone(task), lease: clone(activeLease) }, + }; + } + throw new PersistenceError( + 'CONFLICT', + `Task ${task.id} already has an effective Lease ${activeLease.id}.`, + ); + } + + if (task.revision !== input.expectedTaskRevision) { + throw new PersistenceError( + 'REVISION_MISMATCH', + `Task ${task.id} revision ${task.revision} does not match ${input.expectedTaskRevision}.`, + ); + } + if (task.status !== 'ready' && task.status !== 'running') { + throw new PersistenceError( + 'INVALID_STATE_TRANSITION', + `Task ${task.id} cannot be claimed from ${task.status}.`, + ); + } + + const counter = await this.getFencingCounter(input.workspaceId, input.taskId); + if (counter === undefined) { + throw new PersistenceError('INTEGRITY_ERROR', `Task ${task.id} has no fencing counter.`); + } + const nextFence = counter + 1; + const lease: Lease = { + ...input.lease, + workspaceId: input.workspaceId, + taskId: input.taskId, + sessionId: input.sessionId, + fencingToken: nextFence, + createdAt: input.now, + updatedAt: input.now, + status: 'active', + }; + if (timestampMs(lease.expiresAt, 'Lease.expiresAt') <= nowMs) { + throw new PersistenceError('INVALID_RECORD', 'A newly granted Lease must expire in the future.'); + } + this.assertCanonical('Lease', lease); + + if (activeLease) { + const expiredLease: Lease = { + ...activeLease, + revision: activeLease.revision + 1, + updatedAt: input.now, + status: 'expired', + }; + this.assertCanonical('Lease', expiredLease); + statements.push( + this.database + .prepare( + `UPDATE leases + SET revision = ?, status = ?, updated_at_ms = ?, record_json = ? + WHERE workspace_id = ? AND id = ? AND revision = ? AND status = 'active'`, + ) + .bind( + expiredLease.revision, + expiredLease.status, + nowMs, + serializeJson(expiredLease, 'Lease'), + input.workspaceId, + activeLease.id, + activeLease.revision, + ), + ); + } + + statements.push( + this.database + .prepare( + `UPDATE task_fencing_counters + SET last_fencing_token = ? + WHERE workspace_id = ? AND task_id = ? AND last_fencing_token = ?`, + ) + .bind(nextFence, input.workspaceId, input.taskId, counter), + ); + + let nextTask = task; + if (task.status === 'ready') { + nextTask = { + ...task, + revision: task.revision + 1, + updatedAt: input.now, + status: 'running', + }; + this.assertCanonical('Task', nextTask); + statements.push( + this.database + .prepare( + `UPDATE tasks + SET revision = ?, status = ?, updated_at_ms = ?, record_json = ? + WHERE workspace_id = ? AND id = ? AND revision = ? AND status = 'ready'`, + ) + .bind( + nextTask.revision, + nextTask.status, + nowMs, + serializeJson(nextTask, 'Task'), + input.workspaceId, + input.taskId, + task.revision, + ), + ); + } + + statements.push(this.insertLeaseStatement(lease)); + this.pushAuditStatement(statements, input.auditEvent); + this.pushReceiptStatement(statements, input.receipt); + await this.batch(statements, 'claim Task'); + return { kind: 'committed', value: { task: clone(nextTask), lease: clone(lease) } }; + }); + } + + async updateTask(input: { task: Task; expectedRevision: number }): Promise { + const { task } = input; + this.assertCanonical('Task', task); + return this.coordinator.runSerialized(task.workspaceId, async () => { + const current = await this.getTask(task.workspaceId, task.id); + if (!current) throw new PersistenceError('NOT_FOUND', `Task ${task.id} was not found.`); + if (current.revision !== input.expectedRevision) { + throw new PersistenceError( + 'REVISION_MISMATCH', + `Task ${task.id} revision ${current.revision} does not match ${input.expectedRevision}.`, + ); + } + if (task.revision !== input.expectedRevision + 1) { + throw new PersistenceError( + 'INVALID_RECORD', + `Task ${task.id} replacement revision must be ${input.expectedRevision + 1}.`, + ); + } + if (task.goalId !== current.goalId) { + throw new PersistenceError('INVALID_RECORD', `Task ${task.id} cannot change Goal ownership.`); + } + const result = await this.run( + this.database + .prepare( + `UPDATE tasks + SET revision = ?, status = ?, updated_at_ms = ?, record_json = ? + WHERE workspace_id = ? AND id = ? AND revision = ?`, + ) + .bind( + task.revision, + task.status, + timestampMs(task.updatedAt, 'Task.updatedAt'), + serializeJson(task, 'Task'), + task.workspaceId, + task.id, + input.expectedRevision, + ), + 'update Task revision', + ); + if (changes(result) !== 1) { + throw new PersistenceError('REVISION_MISMATCH', `Task ${task.id} lost its revision race.`); + } + return clone(task); + }); + } + + async updateGoal(input: { goal: Goal; expectedRevision: number }): Promise { + const { goal } = input; + this.assertCanonical('Goal', goal); + return this.coordinator.runSerialized(goal.workspaceId, async () => { + const current = await this.getGoal(goal.workspaceId, goal.id); + if (!current) throw new PersistenceError('NOT_FOUND', `Goal ${goal.id} was not found.`); + if (current.revision !== input.expectedRevision) { + throw new PersistenceError( + 'REVISION_MISMATCH', + `Goal ${goal.id} revision ${current.revision} does not match ${input.expectedRevision}.`, + ); + } + if (goal.revision !== input.expectedRevision + 1) { + throw new PersistenceError( + 'INVALID_RECORD', + `Goal ${goal.id} replacement revision must be ${input.expectedRevision + 1}.`, + ); + } + const result = await this.run( + this.database + .prepare( + `UPDATE goals + SET revision = ?, status = ?, updated_at_ms = ?, record_json = ? + WHERE workspace_id = ? AND id = ? AND revision = ?`, + ) + .bind( + goal.revision, + goal.status, + timestampMs(goal.updatedAt, 'Goal.updatedAt'), + serializeJson(goal, 'Goal'), + goal.workspaceId, + goal.id, + input.expectedRevision, + ), + 'update Goal revision', + ); + if (changes(result) !== 1) { + throw new PersistenceError('REVISION_MISMATCH', `Goal ${goal.id} lost its revision race.`); + } + return clone(goal); + }); + } + + async appendCheckpoint(input: { checkpoint: Checkpoint; now: string }): Promise { + const { checkpoint } = input; + this.assertCanonical('Checkpoint', checkpoint); + const nowMs = timestampMs(input.now, 'checkpoint now'); + return this.coordinator.runSerialized(checkpoint.workspaceId, async () => { + await this.assertCheckpointAuthority(checkpoint, nowMs); + await this.run(this.insertCheckpointStatement(checkpoint), 'append Checkpoint'); + return clone(checkpoint); + }); + } + + async completeTask( + input: CompleteTaskCommitInput, + ): Promise> { + this.assertCanonical('Task', input.task); + this.assertCanonical('Lease', input.lease); + this.assertCanonical('Checkpoint', input.checkpoint); + this.assertRelatedAudit(input.workspaceId, input.auditEvent); + this.assertReceipt(input.workspaceId, input.receipt); + const nowMs = timestampMs(input.now, 'completion now'); + + return this.coordinator.runSerialized(input.workspaceId, async () => { + const replay = await this.resolveReceipt(input.receipt); + if (replay) return replay; + const currentTask = await this.getTask(input.workspaceId, input.task.id); + if (!currentTask) { + throw new PersistenceError('NOT_FOUND', `Task ${input.task.id} was not found.`); + } + if (currentTask.revision !== input.expectedTaskRevision) { + throw new PersistenceError( + 'REVISION_MISMATCH', + `Task ${currentTask.id} revision ${currentTask.revision} does not match ${input.expectedTaskRevision}.`, + ); + } + const currentLease = await this.getLease(input.workspaceId, input.lease.id); + if (!currentLease) { + throw new PersistenceError('STALE_AUTHORITY', `Lease ${input.lease.id} is not authoritative.`); + } + this.assertCompletionAuthority(input, currentTask, currentLease, nowMs); + + const currentGoal = await this.getGoal(input.workspaceId, currentTask.goalId); + if (!currentGoal) { + throw new PersistenceError('INTEGRITY_ERROR', `Goal ${currentTask.goalId} was not found.`); + } + const goalTasks = await this.listGoalTasks(input.workspaceId, currentTask.goalId); + const shouldSucceedGoal = + currentGoal.status === 'active' && + goalTasks.length > 0 && + goalTasks.every((task) => task.id === currentTask.id || task.status === 'succeeded'); + const succeededGoal = shouldSucceedGoal + ? ({ + ...currentGoal, + revision: currentGoal.revision + 1, + updatedAt: input.now, + status: 'succeeded', + } satisfies Goal) + : undefined; + if (succeededGoal) this.assertCanonical('Goal', succeededGoal); + + const statements: D1PreparedStatementLike[] = [ + this.insertCheckpointStatement(input.checkpoint), + this.database + .prepare( + `UPDATE tasks + SET revision = ?, status = ?, updated_at_ms = ?, record_json = ? + WHERE workspace_id = ? AND id = ? AND revision = ? AND status = 'running'`, + ) + .bind( + input.task.revision, + input.task.status, + timestampMs(input.task.updatedAt, 'Task.updatedAt'), + serializeJson(input.task, 'Task'), + input.workspaceId, + input.task.id, + input.expectedTaskRevision, + ), + this.database + .prepare( + `UPDATE leases + SET revision = ?, status = ?, updated_at_ms = ?, record_json = ? + WHERE workspace_id = ? AND id = ? AND revision = ? AND status = 'active' + AND fencing_token = ?`, + ) + .bind( + input.lease.revision, + input.lease.status, + timestampMs(input.lease.updatedAt, 'Lease.updatedAt'), + serializeJson(input.lease, 'Lease'), + input.workspaceId, + input.lease.id, + currentLease.revision, + currentLease.fencingToken, + ), + ]; + if (succeededGoal) { + statements.push( + this.database + .prepare( + `UPDATE goals + SET revision = ?, status = ?, updated_at_ms = ?, record_json = ? + WHERE workspace_id = ? AND id = ? AND revision = ? AND status = 'active'`, + ) + .bind( + succeededGoal.revision, + succeededGoal.status, + nowMs, + serializeJson(succeededGoal, 'Goal'), + input.workspaceId, + succeededGoal.id, + currentGoal.revision, + ), + ); + } + this.pushAuditStatement(statements, input.auditEvent); + this.pushReceiptStatement(statements, input.receipt); + await this.batch(statements, 'complete Task'); + return { + kind: 'committed', + value: { + task: clone(input.task), + lease: clone(input.lease), + checkpoint: clone(input.checkpoint), + ...(succeededGoal === undefined ? {} : { goal: clone(succeededGoal) }), + }, + }; + }); + } + + async appendAuditEvent(input: { auditEvent: AuditEvent }): Promise { + this.assertCanonical('AuditEvent', input.auditEvent); + await this.coordinator.runSerialized(input.auditEvent.workspaceId, async () => { + await this.requireWorkspace(input.auditEvent.workspaceId); + await this.run(this.insertAuditStatement(input.auditEvent), 'append AuditEvent'); + }); + } + + async appendPermissionRequestWithInitialDecision(input: { + request: PermissionRequest; + decision: PermissionDecision; + receipt?: CommandReceiptInput; + auditEvent?: AuditEvent; + }): Promise> { + this.assertCanonical('PermissionRequest', input.request); + this.assertCanonical('PermissionDecision', input.decision); + this.assertRelatedAudit(input.request.workspaceId, input.auditEvent); + this.assertReceipt(input.request.workspaceId, input.receipt); + if ( + input.decision.workspaceId !== input.request.workspaceId || + input.decision.requestId !== input.request.id || + input.decision.sequence !== 1 || + input.decision.supersedesDecisionId !== undefined + ) { + throw new PersistenceError( + 'INVALID_RECORD', + 'Initial PermissionDecision must be sequence 1 for the same PermissionRequest.', + ); + } + + return this.coordinator.runSerialized(input.request.workspaceId, async () => { + const replay = await this.resolveReceipt(input.receipt); + if (replay) return replay; + await this.requireWorkspace(input.request.workspaceId); + const statements: D1PreparedStatementLike[] = [ + this.insertPermissionRequestStatement(input.request), + this.insertPermissionDecisionStatement(input.decision), + this.database + .prepare( + `INSERT INTO permission_heads( + workspace_id, request_id, latest_decision_id, latest_sequence, latest_outcome + ) VALUES (?, ?, ?, ?, ?)`, + ) + .bind( + input.request.workspaceId, + input.request.id, + input.decision.id, + input.decision.sequence, + input.decision.outcome, + ), + ]; + this.pushAuditStatement(statements, input.auditEvent); + this.pushReceiptStatement(statements, input.receipt); + await this.batch(statements, 'append PermissionRequest and initial decision'); + return { + kind: 'committed', + value: { request: clone(input.request), decision: clone(input.decision) }, + }; + }); + } + + async appendPermissionDecision(input: { + decision: PermissionDecision; + expectedPreviousDecisionId: string; + }): Promise { + this.assertCanonical('PermissionDecision', input.decision); + return this.coordinator.runSerialized(input.decision.workspaceId, async () => { + const head = await this.first( + `SELECT latest_decision_id, latest_sequence, latest_outcome + FROM permission_heads + WHERE workspace_id = ? AND request_id = ?`, + input.decision.workspaceId, + input.decision.requestId, + ); + if (!head) { + throw new PersistenceError( + 'NOT_FOUND', + `PermissionRequest ${input.decision.requestId} has no decision head.`, + ); + } + if (head.latest_decision_id !== input.expectedPreviousDecisionId) { + throw new PersistenceError( + 'REVISION_MISMATCH', + `Permission decision head changed from ${input.expectedPreviousDecisionId}.`, + ); + } + if ( + input.decision.sequence !== head.latest_sequence + 1 || + input.decision.supersedesDecisionId !== head.latest_decision_id + ) { + throw new PersistenceError( + 'INVALID_RECORD', + 'PermissionDecision sequence/supersession does not extend the current head.', + ); + } + + await this.batch( + [ + this.insertPermissionDecisionStatement(input.decision), + this.database + .prepare( + `UPDATE permission_heads + SET latest_decision_id = ?, latest_sequence = ?, latest_outcome = ? + WHERE workspace_id = ? AND request_id = ? + AND latest_decision_id = ? AND latest_sequence = ?`, + ) + .bind( + input.decision.id, + input.decision.sequence, + input.decision.outcome, + input.decision.workspaceId, + input.decision.requestId, + head.latest_decision_id, + head.latest_sequence, + ), + ], + 'append PermissionDecision', + ); + return clone(input.decision); + }); + } + + async getCommandReceipt( + workspaceId: string, + commandId: string, + ): Promise { + const row = await this.first( + `SELECT workspace_id, command_id, command_discriminator, semantic_fingerprint, + outcome_kind, response_snapshot_json, created_at_ms, expires_at_ms + FROM command_receipts + WHERE workspace_id = ? AND command_id = ?`, + workspaceId, + commandId, + ); + if (!row) return undefined; + return clone({ + workspaceId: row.workspace_id, + commandId: row.command_id, + command: row.command_discriminator, + semanticFingerprint: row.semantic_fingerprint, + outcomeKind: row.outcome_kind, + responseSnapshot: parseJson(row.response_snapshot_json, 'command receipt response snapshot'), + createdAt: new Date(row.created_at_ms).toISOString(), + ...(row.expires_at_ms === null + ? {} + : { expiresAt: new Date(row.expires_at_ms).toISOString() }), + }); + } + + async loadWorkspaceState(workspaceId: string): Promise { + const workspaceRecord = await this.getWorkspace(workspaceId); + if (!workspaceRecord) return undefined; + const [goals, tasks, agents, sessions, leases, checkpoints, requests, decisions, auditEvents] = + await Promise.all([ + this.readRecords( + `SELECT record_json FROM goals WHERE workspace_id = ? ORDER BY created_at_ms, id`, + workspaceId, + ), + this.readRecords( + `SELECT record_json FROM tasks WHERE workspace_id = ? ORDER BY created_at_ms, id`, + workspaceId, + ), + this.readRecords( + `SELECT record_json FROM agents WHERE workspace_id = ? ORDER BY created_at_ms, id`, + workspaceId, + ), + this.readRecords( + `SELECT record_json FROM sessions WHERE workspace_id = ? ORDER BY created_at_ms, id`, + workspaceId, + ), + this.readRecords( + `SELECT record_json FROM leases WHERE workspace_id = ? ORDER BY created_at_ms, id`, + workspaceId, + ), + this.readRecords( + `SELECT record_json FROM checkpoints WHERE workspace_id = ? ORDER BY created_at_ms, id`, + workspaceId, + ), + this.readRecords( + `SELECT record_json FROM permission_requests + WHERE workspace_id = ? ORDER BY created_at_ms, id`, + workspaceId, + ), + this.readRecords( + `SELECT record_json FROM permission_decisions + WHERE workspace_id = ? ORDER BY created_at_ms, request_id, sequence, id`, + workspaceId, + ), + this.readRecords( + `SELECT record_json FROM audit_events WHERE workspace_id = ? ORDER BY created_at_ms, id`, + workspaceId, + ), + ]); + const counterRows = await this.all( + `SELECT task_id, last_fencing_token + FROM task_fencing_counters WHERE workspace_id = ? ORDER BY task_id`, + workspaceId, + ); + return { + workspace: workspaceRecord, + goals, + tasks, + agents, + sessions, + leases, + checkpoints, + permissionRequests: requests, + permissionDecisions: decisions, + auditEvents, + fencingCounters: Object.fromEntries( + counterRows.map((row) => [row.task_id, Number(row.last_fencing_token)]), + ), + }; + } + + async listTaskCheckpoints(workspaceId: string, taskId: string): Promise { + return this.readRecords( + `SELECT record_json FROM checkpoints + WHERE workspace_id = ? AND task_id = ? ORDER BY created_at_ms, id`, + workspaceId, + taskId, + ); + } + + async listAuditEvents(workspaceId: string, limit: number): Promise { + return this.readRecords( + `SELECT record_json FROM audit_events + WHERE workspace_id = ? ORDER BY created_at_ms, id LIMIT ?`, + workspaceId, + boundedLimit(limit), + ); + } + + async listPermissionDecisions( + workspaceId: string, + requestId: string, + ): Promise { + return this.readRecords( + `SELECT record_json FROM permission_decisions + WHERE workspace_id = ? AND request_id = ? ORDER BY sequence, created_at_ms, id`, + workspaceId, + requestId, + ); + } + + async listPendingHumanPermissions( + workspaceId: string, + limit: number, + ): Promise { + const rows = await this.all( + `SELECT pr.record_json AS request_json, pd.record_json AS decision_json + FROM permission_heads ph + JOIN permission_requests pr + ON pr.workspace_id = ph.workspace_id AND pr.id = ph.request_id + JOIN permission_decisions pd + ON pd.workspace_id = ph.workspace_id AND pd.id = ph.latest_decision_id + WHERE ph.workspace_id = ? AND ph.latest_outcome = 'HUMAN_REQUIRED' + ORDER BY pr.created_at_ms, pr.id + LIMIT ?`, + workspaceId, + boundedLimit(limit), + ); + return rows.map((row) => ({ + request: parseJson(row.request_json, 'PermissionRequest'), + latestDecision: parseJson(row.decision_json, 'PermissionDecision'), + })); + } + + async listExpiredActiveLeases(workspaceId: string, now: string, limit: number): Promise { + return this.readRecords( + `SELECT record_json FROM leases + WHERE workspace_id = ? AND status = 'active' AND expires_at_ms <= ? + ORDER BY expires_at_ms, id LIMIT ?`, + workspaceId, + timestampMs(now, 'lease recovery now'), + boundedLimit(limit), + ); + } + + async listActiveSessionsLastSeenBefore( + workspaceId: string, + cutoff: string, + limit: number, + ): Promise { + return this.readRecords( + `SELECT record_json FROM sessions + WHERE workspace_id = ? AND status = 'active' AND last_seen_at_ms <= ? + ORDER BY last_seen_at_ms, id LIMIT ?`, + workspaceId, + timestampMs(cutoff, 'session recovery cutoff'), + boundedLimit(limit), + ); + } + + private async resolveReceipt( + receipt: CommandReceiptInput | undefined, + ): Promise<{ kind: 'replayed'; receipt: StoredCommandReceipt } | undefined> { + if (!receipt) return undefined; + const stored = await this.getCommandReceipt(receipt.workspaceId, receipt.commandId); + if (!stored) return undefined; + if ( + stored.command !== receipt.command || + stored.semanticFingerprint !== receipt.semanticFingerprint + ) { + throw new PersistenceError( + 'IDEMPOTENCY_CONFLICT', + `Command ${receipt.commandId} was already admitted with different semantics.`, + ); + } + return { kind: 'replayed', receipt: clone(stored) }; + } + + private async assertDependencies(task: Task): Promise { + for (const dependencyId of task.dependencyTaskIds) { + const dependency = await this.getTask(task.workspaceId, dependencyId); + if (!dependency) { + throw new PersistenceError('NOT_FOUND', `Dependency Task ${dependencyId} was not found.`); + } + if (dependency.goalId !== task.goalId) { + throw new PersistenceError( + 'CONFLICT', + `Dependency Task ${dependencyId} belongs to a different Goal.`, + ); + } + } + } + + private async assertCheckpointAuthority(checkpoint: Checkpoint, nowMs: number): Promise { + const task = await this.getTask(checkpoint.workspaceId, checkpoint.taskId); + const lease = await this.getLease(checkpoint.workspaceId, checkpoint.leaseId); + const session = await this.getSession(checkpoint.workspaceId, checkpoint.sessionId); + if ( + !task || + !lease || + !session || + task.status !== 'running' || + session.status !== 'active' || + lease.status !== 'active' || + lease.taskId !== task.id || + lease.sessionId !== session.id || + lease.fencingToken !== checkpoint.fencingToken || + timestampMs(lease.expiresAt, 'Lease.expiresAt') <= nowMs + ) { + throw new PersistenceError('STALE_AUTHORITY', 'Checkpoint authority is stale or inactive.'); + } + const effective = await this.getActiveLease(checkpoint.workspaceId, checkpoint.taskId); + if (!effective || effective.id !== lease.id || effective.fencingToken !== checkpoint.fencingToken) { + throw new PersistenceError('STALE_AUTHORITY', 'Checkpoint Lease is no longer effective.'); + } + } + + private assertCompletionAuthority( + input: CompleteTaskCommitInput, + currentTask: Task, + currentLease: Lease, + nowMs: number, + ): void { + const checkpoint = input.checkpoint; + if ( + currentTask.status !== 'running' || + input.task.status !== 'succeeded' || + input.task.revision !== input.expectedTaskRevision + 1 || + input.task.workspaceId !== input.workspaceId || + input.task.id !== currentTask.id || + input.task.goalId !== currentTask.goalId || + currentLease.status !== 'active' || + timestampMs(currentLease.expiresAt, 'Lease.expiresAt') <= nowMs || + currentLease.taskId !== currentTask.id || + input.lease.id !== currentLease.id || + input.lease.workspaceId !== input.workspaceId || + input.lease.taskId !== currentTask.id || + input.lease.sessionId !== currentLease.sessionId || + input.lease.fencingToken !== currentLease.fencingToken || + input.lease.status !== 'released' || + input.lease.revision !== currentLease.revision + 1 || + checkpoint.workspaceId !== input.workspaceId || + checkpoint.taskId !== currentTask.id || + checkpoint.leaseId !== currentLease.id || + checkpoint.sessionId !== currentLease.sessionId || + checkpoint.fencingToken !== currentLease.fencingToken || + checkpoint.kind !== 'result' + ) { + throw new PersistenceError('STALE_AUTHORITY', 'Task completion authority is stale or invalid.'); + } + } + + private assertCanonical(target: PersistenceDomainTarget, value: unknown): void { + const validation = this.validateCanonicalDomainRecord(target, value); + if (validation.valid) return; + const details = validation.errors?.slice(0, 3).join('; '); + throw new PersistenceError( + 'INVALID_RECORD', + details ? `${target} violates canonical schema: ${details}` : `${target} is invalid.`, + ); + } + + private assertRelatedAudit(workspaceId: string, auditEvent: AuditEvent | undefined): void { + if (!auditEvent) return; + this.assertCanonical('AuditEvent', auditEvent); + if (auditEvent.workspaceId !== workspaceId) { + throw new PersistenceError('INVALID_RECORD', 'AuditEvent Workspace does not match mutation.'); + } + } + + private assertReceipt(workspaceId: string, receipt: CommandReceiptInput | undefined): void { + if (!receipt) return; + if (receipt.workspaceId !== workspaceId) { + throw new PersistenceError('INVALID_RECORD', 'Command receipt Workspace does not match mutation.'); + } + if (!receipt.commandId || !receipt.command || !receipt.semanticFingerprint) { + throw new PersistenceError('INVALID_RECORD', 'Command receipt identity fields must be non-empty.'); + } + timestampMs(receipt.createdAt, 'CommandReceipt.createdAt'); + if (receipt.expiresAt !== undefined) timestampMs(receipt.expiresAt, 'CommandReceipt.expiresAt'); + serializeJson(receipt.responseSnapshot, 'CommandReceipt.responseSnapshot', MAX_RECEIPT_SNAPSHOT_BYTES); + } + + private pushReceiptStatement( + statements: D1PreparedStatementLike[], + receipt: CommandReceiptInput | undefined, + ): void { + if (!receipt) return; + statements.push( + this.database + .prepare( + `INSERT INTO command_receipts( + workspace_id, command_id, command_discriminator, semantic_fingerprint, outcome_kind, + response_snapshot_json, created_at_ms, expires_at_ms + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + receipt.workspaceId, + receipt.commandId, + receipt.command, + receipt.semanticFingerprint, + receipt.outcomeKind, + serializeJson( + receipt.responseSnapshot, + 'CommandReceipt.responseSnapshot', + MAX_RECEIPT_SNAPSHOT_BYTES, + ), + timestampMs(receipt.createdAt, 'CommandReceipt.createdAt'), + receipt.expiresAt === undefined + ? null + : timestampMs(receipt.expiresAt, 'CommandReceipt.expiresAt'), + ), + ); + } + + private pushAuditStatement( + statements: D1PreparedStatementLike[], + auditEvent: AuditEvent | undefined, + ): void { + if (auditEvent) statements.push(this.insertAuditStatement(auditEvent)); + } + + private insertAuditStatement(auditEvent: AuditEvent): D1PreparedStatementLike { + return this.database + .prepare( + `INSERT INTO audit_events( + workspace_id, id, event_type, subject_type, subject_id, created_at_ms, record_json + ) VALUES (?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + auditEvent.workspaceId, + auditEvent.id, + auditEvent.eventType, + auditEvent.subject.type, + auditEvent.subject.id, + timestampMs(auditEvent.createdAt, 'AuditEvent.createdAt'), + serializeJson(auditEvent, 'AuditEvent'), + ); + } + + private insertLeaseStatement(lease: Lease): D1PreparedStatementLike { + return this.database + .prepare( + `INSERT INTO leases( + workspace_id, id, task_id, session_id, revision, status, fencing_token, created_at_ms, + updated_at_ms, expires_at_ms, record_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + lease.workspaceId, + lease.id, + lease.taskId, + lease.sessionId, + lease.revision, + lease.status, + lease.fencingToken, + timestampMs(lease.createdAt, 'Lease.createdAt'), + timestampMs(lease.updatedAt, 'Lease.updatedAt'), + timestampMs(lease.expiresAt, 'Lease.expiresAt'), + serializeJson(lease, 'Lease'), + ); + } + + private insertCheckpointStatement(checkpoint: Checkpoint): D1PreparedStatementLike { + return this.database + .prepare( + `INSERT INTO checkpoints( + workspace_id, id, task_id, session_id, lease_id, fencing_token, created_at_ms, record_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + checkpoint.workspaceId, + checkpoint.id, + checkpoint.taskId, + checkpoint.sessionId, + checkpoint.leaseId, + checkpoint.fencingToken, + timestampMs(checkpoint.createdAt, 'Checkpoint.createdAt'), + serializeJson(checkpoint, 'Checkpoint'), + ); + } + + private insertPermissionRequestStatement(request: PermissionRequest): D1PreparedStatementLike { + return this.database + .prepare( + `INSERT INTO permission_requests( + workspace_id, id, task_id, session_id, lease_id, fencing_token, permission, created_at_ms, + record_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + request.workspaceId, + request.id, + request.taskId, + request.sessionId, + request.leaseId, + request.fencingToken, + request.permission, + timestampMs(request.createdAt, 'PermissionRequest.createdAt'), + serializeJson(request, 'PermissionRequest'), + ); + } + + private insertPermissionDecisionStatement(decision: PermissionDecision): D1PreparedStatementLike { + return this.database + .prepare( + `INSERT INTO permission_decisions( + workspace_id, id, request_id, sequence, outcome, basis, supersedes_decision_id, + created_at_ms, record_json + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + decision.workspaceId, + decision.id, + decision.requestId, + decision.sequence, + decision.outcome, + decision.basis, + decision.supersedesDecisionId ?? null, + timestampMs(decision.createdAt, 'PermissionDecision.createdAt'), + serializeJson(decision, 'PermissionDecision'), + ); + } + + private async requireWorkspace(workspaceId: string): Promise { + const workspace = await this.getWorkspace(workspaceId); + if (!workspace) throw new PersistenceError('NOT_FOUND', `Workspace ${workspaceId} was not found.`); + if (workspace.status !== 'active') { + throw new PersistenceError('INVALID_STATE_TRANSITION', `Workspace ${workspaceId} is archived.`); + } + return workspace; + } + + private async getWorkspace(workspaceId: string): Promise { + return this.readRecord( + `SELECT record_json FROM workspaces WHERE id = ?`, + workspaceId, + ); + } + + private async getGoal(workspaceId: string, goalId: string): Promise { + return this.readRecord( + `SELECT record_json FROM goals WHERE workspace_id = ? AND id = ?`, + workspaceId, + goalId, + ); + } + + private async getTask(workspaceId: string, taskId: string): Promise { + return this.readRecord( + `SELECT record_json FROM tasks WHERE workspace_id = ? AND id = ?`, + workspaceId, + taskId, + ); + } + + private async getAgent(workspaceId: string, agentId: string): Promise { + return this.readRecord( + `SELECT record_json FROM agents WHERE workspace_id = ? AND id = ?`, + workspaceId, + agentId, + ); + } + + private async getSession(workspaceId: string, sessionId: string): Promise { + return this.readRecord( + `SELECT record_json FROM sessions WHERE workspace_id = ? AND id = ?`, + workspaceId, + sessionId, + ); + } + + private async getLease(workspaceId: string, leaseId: string): Promise { + return this.readRecord( + `SELECT record_json FROM leases WHERE workspace_id = ? AND id = ?`, + workspaceId, + leaseId, + ); + } + + private async getActiveLease(workspaceId: string, taskId: string): Promise { + return this.readRecord( + `SELECT record_json FROM leases + WHERE workspace_id = ? AND task_id = ? AND status = 'active' + ORDER BY fencing_token DESC LIMIT 1`, + workspaceId, + taskId, + ); + } + + private async getFencingCounter(workspaceId: string, taskId: string): Promise { + const row = await this.first<{ last_fencing_token: number }>( + `SELECT last_fencing_token FROM task_fencing_counters + WHERE workspace_id = ? AND task_id = ?`, + workspaceId, + taskId, + ); + return row ? Number(row.last_fencing_token) : undefined; + } + + private async listGoalTasks(workspaceId: string, goalId: string): Promise { + return this.readRecords( + `SELECT record_json FROM tasks + WHERE workspace_id = ? AND goal_id = ? ORDER BY created_at_ms, id`, + workspaceId, + goalId, + ); + } + + private async readRecord(sql: string, ...values: unknown[]): Promise { + const row = await this.first(sql, ...values); + return row ? parseJson(row.record_json, 'canonical record') : undefined; + } + + private async readRecords(sql: string, ...values: unknown[]): Promise { + const rows = await this.all(sql, ...values); + return rows.map((row) => parseJson(row.record_json, 'canonical record')); + } + + private async first(sql: string, ...values: unknown[]): Promise { + const row = await this.database.prepare(sql).bind(...values).first(); + return row ?? undefined; + } + + private async all(sql: string, ...values: unknown[]): Promise { + const result = await this.database.prepare(sql).bind(...values).all(); + return [...(result.results ?? [])]; + } + + private async run(statement: D1PreparedStatementLike, context: string): Promise { + try { + return await statement.run(); + } catch (error) { + throw wrapDatabaseError(context, error); + } + } + + private async batch( + statements: D1PreparedStatementLike[], + context: string, + ): Promise { + try { + return await this.database.batch(statements); + } catch (error) { + throw wrapDatabaseError(context, error); + } + } +} + +function serializeJson(value: unknown, label: string, maxBytes?: number): string { + let json: string | undefined; + try { + json = JSON.stringify(value); + } catch (error) { + throw new PersistenceError( + 'INVALID_RECORD', + `${label} is not JSON serializable: ${errorMessage(error)}.`, + ); + } + if (json === undefined) { + throw new PersistenceError('INVALID_RECORD', `${label} is not JSON serializable.`); + } + if (maxBytes !== undefined && new TextEncoder().encode(json).byteLength > maxBytes) { + throw new PersistenceError( + 'INVALID_RECORD', + `${label} exceeds the ${maxBytes}-byte durable snapshot limit.`, + ); + } + return json; +} + +function parseJson(json: string, label: string): T { + try { + return JSON.parse(json) as T; + } catch (error) { + throw new PersistenceError('INTEGRITY_ERROR', `${label} is corrupt: ${errorMessage(error)}.`); + } +} + +function timestampMs(value: string, label: string): number { + const timestamp = Date.parse(value); + if (!Number.isFinite(timestamp)) { + throw new PersistenceError('INVALID_RECORD', `${label} must be a valid UTC timestamp.`); + } + return timestamp; +} + +function boundedLimit(limit: number): number { + if (!Number.isInteger(limit) || limit < 1) { + throw new PersistenceError('INVALID_RECORD', 'Query limit must be a positive integer.'); + } + return Math.min(limit, 1000); +} + +function changes(result: D1ResultLike): number { + return Number(result.meta?.changes ?? 0); +} + +function wrapDatabaseError(context: string, error: unknown): PersistenceError { + if (error instanceof PersistenceError) return error; + return new PersistenceError('INTEGRITY_ERROR', `${context} failed: ${errorMessage(error)}.`); +} + +function errorMessage(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} + +function clone(value: T): T { + return structuredClone(value); +} diff --git a/src/persistence/cloudflare/d1-types.ts b/src/persistence/cloudflare/d1-types.ts new file mode 100644 index 00000000..16ea18cc --- /dev/null +++ b/src/persistence/cloudflare/d1-types.ts @@ -0,0 +1,20 @@ +export interface D1ResultLike> { + readonly success: boolean; + readonly results?: readonly T[]; + readonly meta?: { + readonly changes?: number; + }; +} + +export interface D1PreparedStatementLike { + bind(...values: unknown[]): D1PreparedStatementLike; + first(): Promise; + all(): Promise>; + run>(): Promise>; +} + +export interface D1DatabaseLike { + prepare(sql: string): D1PreparedStatementLike; + batch(statements: D1PreparedStatementLike[]): Promise; + exec?(sql: string): Promise; +} diff --git a/src/persistence/cloudflare/index.ts b/src/persistence/cloudflare/index.ts new file mode 100644 index 00000000..e03a809f --- /dev/null +++ b/src/persistence/cloudflare/index.ts @@ -0,0 +1,7 @@ +export { D1RuntimePersistence } from './d1-runtime-persistence.ts'; +export type { + D1DatabaseLike, + D1PreparedStatementLike, + D1ResultLike, +} from './d1-types.ts'; +export { WorkspaceDurableObjectCoordinator } from './workspace-durable-object-coordinator.ts'; diff --git a/src/persistence/cloudflare/workspace-durable-object-coordinator.ts b/src/persistence/cloudflare/workspace-durable-object-coordinator.ts new file mode 100644 index 00000000..07ed261d --- /dev/null +++ b/src/persistence/cloudflare/workspace-durable-object-coordinator.ts @@ -0,0 +1,39 @@ +import { PersistenceError, type WorkspaceMutationCoordinator } from '../ports.ts'; + +/** + * Reference coordination primitive for the Workspace Durable Object boundary. + * + * A Cloudflare deployment should bind one instance to one Workspace Durable Object id. The optional + * constructor binding models that deployment rule without importing Workers runtime types. Tests may + * leave it unbound to simulate a Durable Object namespace in one process. + */ +export class WorkspaceDurableObjectCoordinator implements WorkspaceMutationCoordinator { + private readonly tails = new Map>(); + + constructor(private readonly boundWorkspaceId?: string) {} + + runSerialized(workspaceId: string, operation: () => Promise): Promise { + if (this.boundWorkspaceId !== undefined && workspaceId !== this.boundWorkspaceId) { + return Promise.reject( + new PersistenceError( + 'CONFLICT', + `Workspace Durable Object ${this.boundWorkspaceId} cannot coordinate ${workspaceId}.`, + ), + ); + } + + const previous = this.tails.get(workspaceId) ?? Promise.resolve(); + const result = previous.then(operation, operation); + const tail = result.then( + () => undefined, + () => undefined, + ); + this.tails.set(workspaceId, tail); + + return result.finally(() => { + if (this.tails.get(workspaceId) === tail) { + this.tails.delete(workspaceId); + } + }); + } +} diff --git a/src/persistence/index.ts b/src/persistence/index.ts new file mode 100644 index 00000000..49006567 --- /dev/null +++ b/src/persistence/index.ts @@ -0,0 +1,18 @@ +export { + PersistenceError, + type ClaimTaskCommitInput, + type ClaimTaskCommitValue, + type CommandReceiptInput, + type CompleteTaskCommitInput, + type CompleteTaskCommitValue, + type DurableRuntimePersistence, + type MutationCommitResult, + type PendingHumanPermission, + type PersistenceDomainTarget, + type PersistenceDomainValidationResult, + type PersistenceDomainValidator, + type PersistenceErrorCode, + type StoredCommandReceipt, + type WorkspaceMutationCoordinator, + type WorkspaceStateSnapshot, +} from './ports.ts'; diff --git a/src/persistence/ports.ts b/src/persistence/ports.ts new file mode 100644 index 00000000..1fe1bed3 --- /dev/null +++ b/src/persistence/ports.ts @@ -0,0 +1,176 @@ +import type { + Agent, + AuditEvent, + Checkpoint, + Goal, + Lease, + PermissionDecision, + PermissionRequest, + Session, + Task, + Workspace, +} from '@mindrail/contracts'; + +export type PersistenceDomainTarget = + | 'Workspace' + | 'Agent' + | 'Session' + | 'Goal' + | 'Task' + | 'Lease' + | 'Checkpoint' + | 'PermissionRequest' + | 'PermissionDecision' + | 'AuditEvent'; + +export interface PersistenceDomainValidationResult { + readonly valid: boolean; + readonly errors?: readonly string[]; +} + +export type PersistenceDomainValidator = ( + target: PersistenceDomainTarget, + value: unknown, +) => PersistenceDomainValidationResult; + +export type PersistenceErrorCode = + | 'NOT_FOUND' + | 'CONFLICT' + | 'REVISION_MISMATCH' + | 'STALE_AUTHORITY' + | 'IDEMPOTENCY_CONFLICT' + | 'INVALID_STATE_TRANSITION' + | 'INVALID_RECORD' + | 'INTEGRITY_ERROR'; + +export class PersistenceError extends Error { + readonly code: PersistenceErrorCode; + + constructor(code: PersistenceErrorCode, message: string) { + super(message); + this.name = 'PersistenceError'; + this.code = code; + } +} + +export interface CommandReceiptInput { + workspaceId: string; + commandId: string; + command: string; + semanticFingerprint: string; + outcomeKind: 'result' | 'error'; + responseSnapshot: unknown; + createdAt: string; + expiresAt?: string; +} + +export type StoredCommandReceipt = Readonly; + +export type MutationCommitResult = + | { + kind: 'committed'; + value: T; + } + | { + kind: 'replayed'; + receipt: StoredCommandReceipt; + }; + +export interface WorkspaceStateSnapshot { + workspace: Workspace; + goals: Goal[]; + tasks: Task[]; + agents: Agent[]; + sessions: Session[]; + leases: Lease[]; + checkpoints: Checkpoint[]; + permissionRequests: PermissionRequest[]; + permissionDecisions: PermissionDecision[]; + auditEvents: AuditEvent[]; + fencingCounters: Record; +} + +export interface PendingHumanPermission { + request: PermissionRequest; + latestDecision: PermissionDecision; +} + +export interface WorkspaceMutationCoordinator { + runSerialized(workspaceId: string, operation: () => Promise): Promise; +} + +export interface ClaimTaskCommitInput { + workspaceId: string; + taskId: string; + sessionId: string; + expectedTaskRevision: number; + lease: Omit; + now: string; + receipt?: CommandReceiptInput; + auditEvent?: AuditEvent; +} + +export interface ClaimTaskCommitValue { + task: Task; + lease: Lease; +} + +export interface CompleteTaskCommitInput { + workspaceId: string; + task: Task; + lease: Lease; + checkpoint: Checkpoint; + expectedTaskRevision: number; + now: string; + receipt?: CommandReceiptInput; + auditEvent?: AuditEvent; +} + +export interface CompleteTaskCommitValue { + task: Task; + lease: Lease; + checkpoint: Checkpoint; + goal?: Goal; +} + +export interface DurableRuntimePersistence { + bootstrapWorkspace(workspace: Workspace): Promise; + createAgent(input: { agent: Agent }): Promise; + createSession(input: { session: Session }): Promise; + createGoal(input: { + goal: Goal; + receipt?: CommandReceiptInput; + auditEvent?: AuditEvent; + }): Promise>; + createTask(input: { + task: Task; + receipt?: CommandReceiptInput; + auditEvent?: AuditEvent; + }): Promise>; + claimTask(input: ClaimTaskCommitInput): Promise>; + updateTask(input: { task: Task; expectedRevision: number }): Promise; + updateGoal(input: { goal: Goal; expectedRevision: number }): Promise; + appendCheckpoint(input: { checkpoint: Checkpoint; now: string }): Promise; + completeTask( + input: CompleteTaskCommitInput, + ): Promise>; + appendAuditEvent(input: { auditEvent: AuditEvent }): Promise; + appendPermissionRequestWithInitialDecision(input: { + request: PermissionRequest; + decision: PermissionDecision; + receipt?: CommandReceiptInput; + auditEvent?: AuditEvent; + }): Promise>; + appendPermissionDecision(input: { + decision: PermissionDecision; + expectedPreviousDecisionId: string; + }): Promise; + getCommandReceipt(workspaceId: string, commandId: string): Promise; + loadWorkspaceState(workspaceId: string): Promise; + listTaskCheckpoints(workspaceId: string, taskId: string): Promise; + listAuditEvents(workspaceId: string, limit: number): Promise; + listPermissionDecisions(workspaceId: string, requestId: string): Promise; + listPendingHumanPermissions(workspaceId: string, limit: number): Promise; + listExpiredActiveLeases(workspaceId: string, now: string, limit: number): Promise; + listActiveSessionsLastSeenBefore(workspaceId: string, cutoff: string, limit: number): Promise; +} diff --git a/test/persistence/d1-sqlite-harness.ts b/test/persistence/d1-sqlite-harness.ts index a2bf0ba0..e80800a0 100644 --- a/test/persistence/d1-sqlite-harness.ts +++ b/test/persistence/d1-sqlite-harness.ts @@ -52,11 +52,7 @@ class SqliteD1PreparedStatement implements D1PreparedStatementLike { ) {} bind(...values: unknown[]): D1PreparedStatementLike { - return new SqliteD1PreparedStatement( - this.database, - this.sql, - values.map(toSqlInputValue), - ); + return new SqliteD1PreparedStatement(this.database, this.sql, values.map(toSqlInputValue)); } async first(): Promise { diff --git a/test/persistence/fixtures.ts b/test/persistence/fixtures.ts index f11681ca..a237caf1 100644 --- a/test/persistence/fixtures.ts +++ b/test/persistence/fixtures.ts @@ -73,11 +73,7 @@ export function agent(workspaceId = 'ws-a', id = 'agent-a'): Agent { }; } -export function session( - workspaceId = 'ws-a', - agentId = 'agent-a', - id = 'session-a', -): Session { +export function session(workspaceId = 'ws-a', agentId = 'agent-a', id = 'session-a'): Session { return { id, workspaceId, diff --git a/test/persistence/setup.ts b/test/persistence/setup.ts index bb8a61ab..bdd46757 100644 --- a/test/persistence/setup.ts +++ b/test/persistence/setup.ts @@ -1,19 +1,27 @@ -import { readFileSync } from 'node:fs'; +import { readdirSync, readFileSync } from 'node:fs'; +import { dirname, join } from 'node:path'; +import { fileURLToPath } from 'node:url'; import { D1RuntimePersistence } from '../../src/persistence/cloudflare/d1-runtime-persistence.ts'; import { WorkspaceDurableObjectCoordinator } from '../../src/persistence/cloudflare/workspace-durable-object-coordinator.ts'; import { persistenceCanonicalValidator } from './canonical-domain-validator.ts'; import { SqliteD1Database } from './d1-sqlite-harness.ts'; -const migrationUrl = new URL('../../migrations/0001_runtime_persistence.sql', import.meta.url); -const migrationSql = readFileSync(migrationUrl, 'utf8'); +const here = dirname(fileURLToPath(import.meta.url)); +const migrationDirectory = join(here, '../../migrations'); +const migrations = readdirSync(migrationDirectory) + .filter((name) => /^\d+.*\.sql$/.test(name)) + .sort() + .map((name) => readFileSync(join(migrationDirectory, name), 'utf8')); export async function openPersistence(path: string): Promise<{ database: SqliteD1Database; persistence: D1RuntimePersistence; }> { const database = new SqliteD1Database(path); - await database.exec(migrationSql); + for (const migration of migrations) { + await database.exec(migration); + } const persistence = new D1RuntimePersistence({ database, coordinator: new WorkspaceDurableObjectCoordinator(),