diff --git a/package.json b/package.json index 25f21059..831aaab5 100644 --- a/package.json +++ b/package.json @@ -28,7 +28,8 @@ "dev": "node scripts/dev-server.mjs", "postinstall": "node scripts/fix-node-pty-permissions.mjs", "start": "node dist/cli.js serve", - "test": "tsx src/config.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && tsx src/cli.test.ts", + "test": "tsx src/config.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && npm run test:workflows && tsx src/cli.test.ts", + "test:workflows": "tsx src/workflows/migrations.test.ts && tsx src/workflows/workflows.test.ts", "typecheck": "tsc -p tsconfig.json --noEmit" }, "keywords": [], diff --git a/src/db/migrations.ts b/src/db/migrations.ts index 2bda20c2..9fe3956c 100644 --- a/src/db/migrations.ts +++ b/src/db/migrations.ts @@ -22,6 +22,11 @@ const migrations: Migration[] = [ name: "local-agent-sessions", up: migrateLocalAgentSessions, }, + { + version: 4, + name: "durable-workflows", + up: migrateDurableWorkflows, + }, ]; export function migrateDatabase(sqlite: Database.Database): void { @@ -174,6 +179,91 @@ function migrateLocalAgentSessions(sqlite: Database.Database): void { addColumnIfMissing(sqlite, "local_agent_sessions", "thinking", "text"); } +function migrateDurableWorkflows(sqlite: Database.Database): void { + sqlite.exec(` + create table if not exists workflow_runs ( + id text primary key, + definition_version integer not null, + status text not null, + definition_json text not null, + input_json text not null, + policy_json text not null, + idempotency_key text unique, + request_hash text not null, + result_json text, + error_json text, + cancellation_requested_at text, + event_sequence integer not null default 0, + created_at text not null, + updated_at text not null, + started_at text, + completed_at text + ); + + create index if not exists workflow_runs_status_idx + on workflow_runs(status, created_at); + + create table if not exists workflow_nodes ( + id text primary key, + workflow_run_id text not null, + node_key text not null, + node_type text not null, + status text not null, + definition_json text not null, + attempt integer not null default 0, + claim_token text, + claimed_at text, + claim_expires_at text, + result_json text, + error_json text, + created_at text not null, + updated_at text not null, + completed_at text, + foreign key (workflow_run_id) references workflow_runs(id) on delete cascade + ); + + create unique index if not exists workflow_nodes_run_key_idx + on workflow_nodes(workflow_run_id, node_key); + + create unique index if not exists workflow_nodes_run_id_idx + on workflow_nodes(workflow_run_id, id); + + create index if not exists workflow_nodes_status_idx + on workflow_nodes(workflow_run_id, status, created_at); + + create table if not exists workflow_edges ( + workflow_run_id text not null, + from_node_id text not null, + to_node_id text not null, + primary key (workflow_run_id, from_node_id, to_node_id), + foreign key (workflow_run_id) references workflow_runs(id) on delete cascade, + foreign key (workflow_run_id, from_node_id) + references workflow_nodes(workflow_run_id, id) on delete cascade, + foreign key (workflow_run_id, to_node_id) + references workflow_nodes(workflow_run_id, id) on delete cascade + ); + + create index if not exists workflow_edges_to_node_idx + on workflow_edges(workflow_run_id, to_node_id); + + create table if not exists workflow_events ( + workflow_run_id text not null, + sequence integer not null, + event_type text not null, + node_id text, + payload_json text not null, + created_at text not null, + primary key (workflow_run_id, sequence), + foreign key (workflow_run_id) references workflow_runs(id) on delete cascade, + foreign key (workflow_run_id, node_id) + references workflow_nodes(workflow_run_id, id) + ); + + create index if not exists workflow_events_cursor_idx + on workflow_events(workflow_run_id, sequence); + `); +} + function addColumnIfMissing( sqlite: Database.Database, table: "workspace_sessions" | "local_agent_sessions", diff --git a/src/db/schema.ts b/src/db/schema.ts index 01d13fac..631e6d8c 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -1,4 +1,12 @@ -import { index, integer, primaryKey, sqliteTable, text } from "drizzle-orm/sqlite-core"; +import { + foreignKey, + index, + integer, + primaryKey, + sqliteTable, + text, + uniqueIndex, +} from "drizzle-orm/sqlite-core"; export const workspaceSessions = sqliteTable( "workspace_sessions", @@ -97,9 +105,113 @@ export const localAgentSessions = sqliteTable( ], ); +export const workflowRuns = sqliteTable( + "workflow_runs", + { + id: text("id").primaryKey(), + definitionVersion: integer("definition_version").notNull(), + status: text("status").notNull(), + definitionJson: text("definition_json").notNull(), + inputJson: text("input_json").notNull(), + policyJson: text("policy_json").notNull(), + idempotencyKey: text("idempotency_key").unique(), + requestHash: text("request_hash").notNull(), + resultJson: text("result_json"), + errorJson: text("error_json"), + cancellationRequestedAt: text("cancellation_requested_at"), + eventSequence: integer("event_sequence").notNull().default(0), + createdAt: text("created_at").notNull(), + updatedAt: text("updated_at").notNull(), + startedAt: text("started_at"), + completedAt: text("completed_at"), + }, + (table) => [index("workflow_runs_status_idx").on(table.status, table.createdAt)], +); + +export const workflowNodes = sqliteTable( + "workflow_nodes", + { + id: text("id").primaryKey(), + workflowRunId: text("workflow_run_id") + .notNull() + .references(() => workflowRuns.id, { onDelete: "cascade" }), + nodeKey: text("node_key").notNull(), + nodeType: text("node_type").notNull(), + status: text("status").notNull(), + definitionJson: text("definition_json").notNull(), + attempt: integer("attempt").notNull().default(0), + claimToken: text("claim_token"), + claimedAt: text("claimed_at"), + claimExpiresAt: text("claim_expires_at"), + resultJson: text("result_json"), + errorJson: text("error_json"), + createdAt: text("created_at").notNull(), + updatedAt: text("updated_at").notNull(), + completedAt: text("completed_at"), + }, + (table) => [ + uniqueIndex("workflow_nodes_run_key_idx").on(table.workflowRunId, table.nodeKey), + uniqueIndex("workflow_nodes_run_id_idx").on(table.workflowRunId, table.id), + index("workflow_nodes_status_idx").on(table.workflowRunId, table.status, table.createdAt), + ], +); + +export const workflowEdges = sqliteTable( + "workflow_edges", + { + workflowRunId: text("workflow_run_id") + .notNull() + .references(() => workflowRuns.id, { onDelete: "cascade" }), + fromNodeId: text("from_node_id").notNull(), + toNodeId: text("to_node_id").notNull(), + }, + (table) => [ + primaryKey({ columns: [table.workflowRunId, table.fromNodeId, table.toNodeId] }), + foreignKey({ + columns: [table.workflowRunId, table.fromNodeId], + foreignColumns: [workflowNodes.workflowRunId, workflowNodes.id], + }).onDelete("cascade"), + foreignKey({ + columns: [table.workflowRunId, table.toNodeId], + foreignColumns: [workflowNodes.workflowRunId, workflowNodes.id], + }).onDelete("cascade"), + index("workflow_edges_to_node_idx").on(table.workflowRunId, table.toNodeId), + ], +); + +export const workflowEvents = sqliteTable( + "workflow_events", + { + workflowRunId: text("workflow_run_id") + .notNull() + .references(() => workflowRuns.id, { onDelete: "cascade" }), + sequence: integer("sequence").notNull(), + eventType: text("event_type").notNull(), + nodeId: text("node_id"), + payloadJson: text("payload_json").notNull(), + createdAt: text("created_at").notNull(), + }, + (table) => [ + primaryKey({ columns: [table.workflowRunId, table.sequence] }), + foreignKey({ + columns: [table.workflowRunId, table.nodeId], + foreignColumns: [workflowNodes.workflowRunId, workflowNodes.id], + }), + index("workflow_events_cursor_idx").on(table.workflowRunId, table.sequence), + ], +); + export type WorkspaceSessionRow = typeof workspaceSessions.$inferSelect; export type NewWorkspaceSessionRow = typeof workspaceSessions.$inferInsert; export type LoadedAgentFileRow = typeof loadedAgentFiles.$inferSelect; export type NewLoadedAgentFileRow = typeof loadedAgentFiles.$inferInsert; export type LocalAgentSessionRow = typeof localAgentSessions.$inferSelect; export type NewLocalAgentSessionRow = typeof localAgentSessions.$inferInsert; +export type WorkflowRunRow = typeof workflowRuns.$inferSelect; +export type NewWorkflowRunRow = typeof workflowRuns.$inferInsert; +export type WorkflowNodeRow = typeof workflowNodes.$inferSelect; +export type NewWorkflowNodeRow = typeof workflowNodes.$inferInsert; +export type WorkflowEdgeRow = typeof workflowEdges.$inferSelect; +export type NewWorkflowEdgeRow = typeof workflowEdges.$inferInsert; +export type WorkflowEventRow = typeof workflowEvents.$inferSelect; +export type NewWorkflowEventRow = typeof workflowEvents.$inferInsert; diff --git a/src/oauth-store.test.ts b/src/oauth-store.test.ts index e1c00338..de3c116b 100644 --- a/src/oauth-store.test.ts +++ b/src/oauth-store.test.ts @@ -44,6 +44,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise { { version: 1, name: "workspace-state" }, { version: 2, name: "oauth-state" }, { version: 3, name: "local-agent-sessions" }, + { version: 4, name: "durable-workflows" }, ]); } finally { database.close(); diff --git a/src/workflows/migrations.test.ts b/src/workflows/migrations.test.ts new file mode 100644 index 00000000..30671448 --- /dev/null +++ b/src/workflows/migrations.test.ts @@ -0,0 +1,271 @@ +import assert from "node:assert/strict"; +import { chmodSync, mkdirSync, mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import Database from "better-sqlite3"; +import { databasePath, openDatabase } from "../db/client.js"; + +const root = mkdtempSync(join(tmpdir(), "devspace-workflow-migrations-test-")); + +try { + testFreshSchema(join(root, "fresh")); + testExistingMigrationCompatibility(join(root, "existing")); + testMigrationRollback(join(root, "rollback")); +} finally { + rmSync(root, { recursive: true, force: true }); +} + +function testFreshSchema(stateDir: string): void { + const database = openDatabase(stateDir); + try { + assert.deepEqual( + database.sqlite.prepare("select version, name from devspace_schema_migrations order by version").all(), + [ + { version: 1, name: "workspace-state" }, + { version: 2, name: "oauth-state" }, + { version: 3, name: "local-agent-sessions" }, + { version: 4, name: "durable-workflows" }, + ], + ); + const tables = database.sqlite + .prepare( + `select name from sqlite_master + where type = 'table' and name like 'workflow_%' + order by name`, + ) + .pluck() + .all(); + assert.deepEqual(tables, ["workflow_edges", "workflow_events", "workflow_nodes", "workflow_runs"]); + + const edgeForeignKeys = database.sqlite.prepare("pragma foreign_key_list(workflow_edges)").all() as Array<{ + id: number; + seq: number; + table: string; + from: string; + to: string; + }>; + assert.deepEqual( + edgeForeignKeys + .filter((row) => row.table === "workflow_nodes") + .map((row) => [row.id, row.seq, row.from, row.to]) + .sort(), + [ + [0, 0, "workflow_run_id", "workflow_run_id"], + [0, 1, "to_node_id", "id"], + [1, 0, "workflow_run_id", "workflow_run_id"], + [1, 1, "from_node_id", "id"], + ], + ); + + insertWorkflowFixture(database.sqlite); + assert.throws( + () => + database.sqlite + .prepare( + "insert into workflow_edges (workflow_run_id, from_node_id, to_node_id) values ('run-a', 'node-a', 'node-b')", + ) + .run(), + /FOREIGN KEY constraint failed/, + ); + assert.throws( + () => + database.sqlite + .prepare( + `insert into workflow_events + (workflow_run_id, sequence, event_type, node_id, payload_json, created_at) + values ('run-a', 1, 'invalid', 'node-b', '{}', '2026-01-01T00:00:00.000Z')`, + ) + .run(), + /FOREIGN KEY constraint failed/, + ); + } finally { + database.close(); + } +} + +function testExistingMigrationCompatibility(stateDir: string): void { + createVersion3Fixture(stateDir); + const migrated = openDatabase(stateDir); + try { + assert.deepEqual( + migrated.sqlite.prepare("select version from devspace_schema_migrations order by version").pluck().all(), + [1, 2, 3, 4], + ); + assert.deepEqual(migrated.sqlite.prepare("select * from workspace_sessions").all(), [ + { + id: "workspace-1", + root: "/tmp/workspace", + status: "active", + mode: "checkout", + source_root: null, + base_ref: null, + base_sha: null, + managed: "false", + created_at: "2026-01-01T00:00:00.000Z", + last_used_at: "2026-01-01T00:00:00.000Z", + }, + ]); + assert.equal( + migrated.sqlite.prepare("select content from loaded_agent_files where path = 'CLAUDE.md'").pluck().get(), + "instructions", + ); + assert.equal(migrated.sqlite.prepare("select client_json from oauth_clients").pluck().get(), "{}"); + assert.equal(migrated.sqlite.prepare("select client_id from oauth_access_tokens").pluck().get(), "client-1"); + assert.equal(migrated.sqlite.prepare("select client_id from oauth_refresh_tokens").pluck().get(), "client-1"); + assert.equal(migrated.sqlite.prepare("select status from local_agent_sessions").pluck().get(), "active"); + + assert.deepEqual( + migrated.sqlite + .prepare("pragma table_info(local_agent_sessions)") + .all() + .map((row) => (row as { name: string }).name), + [ + "id", + "workspace_id", + "workspace_root", + "profile_name", + "provider", + "model", + "thinking", + "provider_session_id", + "status", + "latest_response", + "error", + "created_at", + "updated_at", + ], + ); + for (const index of [ + "workspace_sessions_root_idx", + "workspace_sessions_status_idx", + "loaded_agent_files_path_idx", + "oauth_clients_issued_at_idx", + "oauth_access_tokens_client_id_idx", + "oauth_access_tokens_expires_at_idx", + "oauth_refresh_tokens_client_id_idx", + "oauth_refresh_tokens_expires_at_idx", + "local_agent_sessions_workspace_id_idx", + "local_agent_sessions_workspace_root_idx", + "local_agent_sessions_provider_session_id_idx", + ]) { + assert.equal( + migrated.sqlite + .prepare("select count(*) from sqlite_master where type = 'index' and name = ?") + .pluck() + .get(index), + 1, + `missing pre-v4 index ${index}`, + ); + } + assert.equal( + migrated.sqlite + .prepare("select count(*) from sqlite_master where type = 'table' and name like 'workflow_%'") + .pluck() + .get(), + 4, + ); + } finally { + migrated.close(); + } +} + +function testMigrationRollback(stateDir: string): void { + createVersion3Fixture(stateDir); + const incompatible = new Database(databasePath(stateDir)); + try { + incompatible.exec("create table workflow_events (legacy_value text not null)"); + } finally { + incompatible.close(); + } + + assert.throws(() => openDatabase(stateDir), /no such column: workflow_run_id/); + + const sqlite = new Database(databasePath(stateDir)); + try { + assert.deepEqual( + sqlite.prepare("select version from devspace_schema_migrations order by version").pluck().all(), + [1, 2, 3], + ); + assert.equal(sqlite.prepare("select count(*) from workspace_sessions").pluck().get(), 1); + assert.equal(sqlite.prepare("select count(*) from local_agent_sessions").pluck().get(), 1); + assert.deepEqual( + sqlite + .prepare("select name from sqlite_master where type = 'table' and name like 'workflow_%' order by name") + .pluck() + .all(), + ["workflow_events"], + ); + assert.deepEqual( + sqlite + .prepare("pragma table_info(workflow_events)") + .all() + .map((row) => (row as { name: string }).name), + ["legacy_value"], + ); + } finally { + sqlite.close(); + } +} + +function createVersion3Fixture(stateDir: string): void { + mkdirSync(stateDir, { recursive: true, mode: 0o700 }); + chmodSync(stateDir, 0o700); + const initialized = openDatabase(stateDir); + try { + initialized.sqlite.exec(` + drop table workflow_edges; + drop table workflow_events; + drop table workflow_nodes; + drop table workflow_runs; + delete from devspace_schema_migrations where version = 4; + + insert into workspace_sessions ( + id, root, status, mode, source_root, base_ref, base_sha, managed, created_at, last_used_at + ) values ( + 'workspace-1', '/tmp/workspace', 'active', 'checkout', null, null, null, 'false', + '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z' + ); + insert into loaded_agent_files ( + workspace_session_id, path, content_hash, content, loaded_at, last_seen_at + ) values ( + 'workspace-1', 'CLAUDE.md', 'hash', 'instructions', + '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z' + ); + insert into oauth_clients (client_id, client_json, issued_at) values ('client-1', '{}', 1); + insert into oauth_access_tokens ( + token_hash, client_id, scopes_json, expires_at, resource + ) values ('access-1', 'client-1', '[]', 2, null); + insert into oauth_refresh_tokens ( + token_hash, client_id, scopes_json, expires_at, resource + ) values ('refresh-1', 'client-1', '[]', 3, null); + insert into local_agent_sessions ( + id, workspace_id, workspace_root, profile_name, provider, model, thinking, + provider_session_id, status, latest_response, error, created_at, updated_at + ) values ( + 'agent-1', 'workspace-1', '/tmp/workspace', 'default', 'claude', null, null, + null, 'active', null, null, '2026-01-01T00:00:00.000Z', '2026-01-01T00:00:00.000Z' + ); + `); + } finally { + initialized.close(); + } +} + +function insertWorkflowFixture(sqlite: Database.Database): void { + const insertRun = sqlite.prepare( + `insert into workflow_runs ( + id, definition_version, status, definition_json, input_json, policy_json, + request_hash, created_at, updated_at + ) values (?, 1, 'queued', '{}', '{}', '{}', 'hash', ?, ?)`, + ); + const now = "2026-01-01T00:00:00.000Z"; + insertRun.run("run-a", now, now); + insertRun.run("run-b", now, now); + const insertNode = sqlite.prepare( + `insert into workflow_nodes ( + id, workflow_run_id, node_key, node_type, status, definition_json, created_at, updated_at + ) values (?, ?, 'agent', 'agent', 'ready', '{}', ?, ?)`, + ); + insertNode.run("node-a", "run-a", now, now); + insertNode.run("node-b", "run-b", now, now); +} diff --git a/src/workflows/orchestrator.ts b/src/workflows/orchestrator.ts new file mode 100644 index 00000000..c2ab164b --- /dev/null +++ b/src/workflows/orchestrator.ts @@ -0,0 +1,125 @@ +import { performance } from "node:perf_hooks"; +import type { + SubmitWorkflowRequest, + WorkflowEventPage, + WorkflowEventReadOptions, + WorkflowRunRecord, + WorkflowWaitOptions, +} from "./types.js"; +import { WorkflowStore, WorkflowValidationError } from "./store.js"; + +const DEFAULT_WAIT_TIMEOUT_MS = 30_000; +const MAX_WAIT_TIMEOUT_MS = 60_000; +const DEFAULT_POLL_INTERVAL_MS = 50; +const MAX_POLL_INTERVAL_MS = 1_000; + +export class WorkflowOrchestrator { + private readonly store: WorkflowStore; + private readonly waiters = new Map void>>(); + private closed = false; + + constructor(stateDir: string) { + this.store = new WorkflowStore(stateDir); + } + + submit(request: SubmitWorkflowRequest): WorkflowRunRecord { + const workflow = this.store.submit(request).workflow; + this.notify(workflow.id); + return workflow; + } + + get(workflowId: string): WorkflowRunRecord | undefined { + return this.store.get(workflowId); + } + + async wait(workflowId: string, options: WorkflowWaitOptions = {}): Promise { + const timeoutMs = options.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS; + const pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; + validateWaitOptions(timeoutMs, pollIntervalMs); + + let workflow = this.store.require(workflowId); + if (isTerminal(workflow) || timeoutMs === 0) return workflow; + + const deadline = performance.now() + timeoutMs; + while (!isTerminal(workflow)) { + const remaining = deadline - performance.now(); + if (remaining <= 0) return workflow; + await this.waitForPollOrNotification( + workflowId, + Math.min(remaining, pollIntervalMs), + ); + if (this.closed) return workflow; + workflow = this.store.require(workflowId); + } + return workflow; + } + + events(workflowId: string, options: WorkflowEventReadOptions = {}): WorkflowEventPage { + return this.store.readEvents(workflowId, options); + } + + cancel(workflowId: string): WorkflowRunRecord { + const workflow = this.store.requestCancellation(workflowId); + this.notify(workflowId); + return workflow; + } + + close(): void { + if (this.closed) return; + this.closed = true; + for (const callbacks of this.waiters.values()) { + for (const callback of callbacks) callback(); + } + this.waiters.clear(); + this.store.close(); + } + + private waitForPollOrNotification(workflowId: string, delayMs: number): Promise { + return new Promise((resolve) => { + let settled = false; + const callbacks = this.waiters.get(workflowId) ?? new Set<() => void>(); + const finish = () => { + if (settled) return; + settled = true; + clearTimeout(timer); + callbacks.delete(finish); + if (callbacks.size === 0) this.waiters.delete(workflowId); + resolve(); + }; + callbacks.add(finish); + this.waiters.set(workflowId, callbacks); + const timer = setTimeout(finish, delayMs); + }); + } + + private notify(workflowId: string): void { + const callbacks = this.waiters.get(workflowId); + if (!callbacks) return; + for (const callback of [...callbacks]) callback(); + } +} + +function validateWaitOptions(timeoutMs: number, pollIntervalMs: number): void { + if (!Number.isFinite(timeoutMs) || timeoutMs < 0 || timeoutMs > MAX_WAIT_TIMEOUT_MS) { + throw new WorkflowValidationError( + `Workflow wait timeout must be between 0 and ${MAX_WAIT_TIMEOUT_MS} milliseconds`, + ); + } + if ( + !Number.isFinite(pollIntervalMs) || + pollIntervalMs < 1 || + pollIntervalMs > MAX_POLL_INTERVAL_MS + ) { + throw new WorkflowValidationError( + `Workflow wait poll interval must be between 1 and ${MAX_POLL_INTERVAL_MS} milliseconds`, + ); + } +} + +function isTerminal(workflow: WorkflowRunRecord): boolean { + return ( + workflow.status === "succeeded" || + workflow.status === "failed" || + workflow.status === "cancelled" + ); +} diff --git a/src/workflows/store.ts b/src/workflows/store.ts new file mode 100644 index 00000000..78099d7f --- /dev/null +++ b/src/workflows/store.ts @@ -0,0 +1,891 @@ +import { createHash, randomUUID } from "node:crypto"; +import { Buffer } from "node:buffer"; +import { openDatabase, type DatabaseHandle } from "../db/client.js"; +import { + WORKFLOW_DEFINITION_VERSION, + WORKFLOW_POLICY_VERSION, + type JsonObject, + type JsonValue, + type SubmitWorkflowRequest, + type SubmitWorkflowResult, + type WorkflowDefinition, + type WorkflowEdgeRecord, + type WorkflowEvent, + type WorkflowEventPage, + type WorkflowEventReadOptions, + type WorkflowNodeClaim, + type WorkflowNodeDefinitionV1, + type WorkflowNodeRecord, + type WorkflowNodeStatus, + type WorkflowNodeTransition, + type WorkflowPolicy, + type WorkflowRunRecord, + type WorkflowStatus, + type WorkflowTransition, +} from "./types.js"; + +const MAX_EVENT_PAYLOAD_BYTES = 64 * 1024; +const DEFAULT_EVENT_LIMIT = 100; +const MAX_EVENT_LIMIT = 1_000; +const DEFAULT_CLAIM_LEASE_MS = 5 * 60_000; +const MAX_CLAIM_LEASE_MS = 24 * 60 * 60_000; + +const TERMINAL_WORKFLOW_STATUSES = new Set([ + "succeeded", + "failed", + "cancelled", +]); +const TERMINAL_NODE_STATUSES = new Set([ + "succeeded", + "failed", + "cancelled", + "skipped", +]); + +const WORKFLOW_TRANSITIONS: Readonly>> = { + queued: new Set(["running", "cancelling", "failed", "cancelled"]), + running: new Set(["cancelling", "succeeded", "failed", "cancelled"]), + cancelling: new Set(["succeeded", "failed", "cancelled"]), + succeeded: new Set(), + failed: new Set(), + cancelled: new Set(), +}; + +const NODE_TRANSITIONS: Readonly>> = { + pending: new Set(["ready", "cancelled", "skipped"]), + ready: new Set(["running", "cancelled", "skipped"]), + running: new Set(["succeeded", "failed", "cancelled"]), + succeeded: new Set(), + failed: new Set(), + cancelled: new Set(), + skipped: new Set(), +}; + +interface WorkflowRunRow { + id: string; + definition_version: number; + status: string; + definition_json: string; + input_json: string; + policy_json: string; + idempotency_key: string | null; + request_hash: string; + result_json: string | null; + error_json: string | null; + cancellation_requested_at: string | null; + created_at: string; + updated_at: string; + started_at: string | null; + completed_at: string | null; +} + +interface WorkflowNodeRow { + id: string; + workflow_run_id: string; + node_key: string; + node_type: string; + status: string; + definition_json: string; + attempt: number; + claim_token: string | null; + claimed_at: string | null; + claim_expires_at: string | null; + result_json: string | null; + error_json: string | null; + created_at: string; + updated_at: string; + completed_at: string | null; +} + +interface WorkflowEdgeRow { + workflow_run_id: string; + from_node_id: string; + to_node_id: string; + from_key: string; + to_key: string; +} + +interface WorkflowEventRow { + workflow_run_id: string; + sequence: number; + event_type: string; + node_id: string | null; + payload_json: string; + created_at: string; +} + +export class WorkflowNotFoundError extends Error { + constructor(workflowId: string) { + super(`Unknown workflow: ${workflowId}`); + this.name = "WorkflowNotFoundError"; + } +} + +export class WorkflowIdempotencyConflictError extends Error { + constructor(idempotencyKey: string) { + super(`Idempotency key was already used for a different workflow request: ${idempotencyKey}`); + this.name = "WorkflowIdempotencyConflictError"; + } +} + +export class WorkflowTransitionError extends Error { + constructor(entity: "workflow" | "node", from: string, to: string) { + super(`Illegal ${entity} status transition: ${from} -> ${to}`); + this.name = "WorkflowTransitionError"; + } +} + +export class WorkflowValidationError extends Error { + constructor(message: string) { + super(message); + this.name = "WorkflowValidationError"; + } +} + +export class WorkflowStore { + private readonly database: DatabaseHandle; + + constructor(stateDir: string) { + this.database = openDatabase(stateDir); + } + + submit(request: SubmitWorkflowRequest): SubmitWorkflowResult { + const normalized = normalizeSubmission(request); + const now = new Date().toISOString(); + const workflowId = createId("wf_"); + const nodeIds = new Map(normalized.definition.nodes.map((node) => [node.key, createId("wfn_")])); + + const save = this.database.sqlite.transaction(() => { + if (normalized.idempotencyKey) { + const existing = this.database.sqlite + .prepare("select id, request_hash from workflow_runs where idempotency_key = ?") + .get(normalized.idempotencyKey) as { id: string; request_hash: string } | undefined; + if (existing) { + if (existing.request_hash !== normalized.requestHash) { + throw new WorkflowIdempotencyConflictError(normalized.idempotencyKey); + } + return { workflowId: existing.id, created: false }; + } + } + + this.database.sqlite + .prepare( + `insert into workflow_runs ( + id, definition_version, status, definition_json, input_json, policy_json, + idempotency_key, request_hash, created_at, updated_at + ) values (?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + workflowId, + WORKFLOW_DEFINITION_VERSION, + normalized.definitionJson, + normalized.inputJson, + normalized.policyJson, + normalized.idempotencyKey ?? null, + normalized.requestHash, + now, + now, + ); + + const incoming = new Map(normalized.definition.nodes.map((node) => [node.key, 0])); + for (const edge of normalized.definition.edges ?? []) { + incoming.set(edge.to, (incoming.get(edge.to) ?? 0) + 1); + } + + const insertNode = this.database.sqlite.prepare( + `insert into workflow_nodes ( + id, workflow_run_id, node_key, node_type, status, definition_json, + created_at, updated_at + ) values (?, ?, ?, ?, ?, ?, ?, ?)`, + ); + for (const node of normalized.definition.nodes) { + insertNode.run( + nodeIds.get(node.key), + workflowId, + node.key, + node.type, + incoming.get(node.key) === 0 ? "ready" : "pending", + canonicalJson(node), + now, + now, + ); + } + + const insertEdge = this.database.sqlite.prepare( + `insert into workflow_edges (workflow_run_id, from_node_id, to_node_id) + values (?, ?, ?)`, + ); + for (const edge of normalized.definition.edges ?? []) { + insertEdge.run(workflowId, nodeIds.get(edge.from), nodeIds.get(edge.to)); + } + + this.insertEvent(workflowId, "workflow.submitted", undefined, { status: "queued" }, now); + return { workflowId, created: true }; + }); + + const saved = save.immediate(); + return { workflow: this.require(saved.workflowId), created: saved.created }; + } + + get(workflowId: string): WorkflowRunRecord | undefined { + const read = this.database.sqlite.transaction(() => { + const row = this.getWorkflowRow(workflowId); + return row ? this.hydrateWorkflow(row) : undefined; + }); + return read.deferred(); + } + + require(workflowId: string): WorkflowRunRecord { + const workflow = this.get(workflowId); + if (!workflow) throw new WorkflowNotFoundError(workflowId); + return workflow; + } + + claimNode(claim: WorkflowNodeClaim): WorkflowNodeRecord | undefined { + if (!claim.claimToken) throw new WorkflowValidationError("Node claim token must not be empty"); + const leaseMs = claim.leaseMs ?? DEFAULT_CLAIM_LEASE_MS; + if (!Number.isSafeInteger(leaseMs) || leaseMs < 1 || leaseMs > MAX_CLAIM_LEASE_MS) { + throw new WorkflowValidationError( + `Node claim lease must be between 1 and ${MAX_CLAIM_LEASE_MS} milliseconds`, + ); + } + const nowDate = new Date(); + const now = nowDate.toISOString(); + const expiresAt = new Date(nowDate.getTime() + leaseMs).toISOString(); + const claimReady = this.database.sqlite.transaction(() => { + const row = this.getNodeRow(claim.workflowId, claim.nodeKey); + if (!row) return undefined; + + const workflow = this.getWorkflowRow(claim.workflowId); + if (!workflow) throw new WorkflowNotFoundError(claim.workflowId); + const workflowStatus = readWorkflowStatus(workflow.status); + if (workflowStatus !== "queued" && workflowStatus !== "running") return undefined; + + if (row.status === "running" && row.claim_token === claim.claimToken) { + this.database.sqlite + .prepare( + `update workflow_nodes + set claim_expires_at = ?, updated_at = ? + where id = ? and status = 'running' and claim_token = ?`, + ) + .run(expiresAt, now, row.id, claim.claimToken); + return this.getNodeRow(claim.workflowId, claim.nodeKey); + } + + const reclaiming = row.status === "running"; + const updated = this.database.sqlite + .prepare( + `update workflow_nodes + set status = 'running', claim_token = ?, claimed_at = ?, claim_expires_at = ?, + attempt = attempt + 1, updated_at = ? + where id = ? + and ( + (status = 'ready' and claim_token is null) + or (status = 'running' and claim_expires_at <= ?) + )`, + ) + .run(claim.claimToken, now, expiresAt, now, row.id, now); + if (updated.changes !== 1) return undefined; + + if (workflowStatus === "queued") { + this.database.sqlite + .prepare( + `update workflow_runs + set status = 'running', started_at = coalesce(started_at, ?), updated_at = ? + where id = ? and status = 'queued'`, + ) + .run(now, now, claim.workflowId); + this.insertEvent(claim.workflowId, "workflow.running", undefined, { status: "running" }, now); + } + this.insertEvent( + claim.workflowId, + reclaiming ? "node.reclaimed" : "node.running", + row.id, + { nodeKey: row.node_key, status: "running", claimToken: claim.claimToken }, + now, + ); + return this.getNodeRow(claim.workflowId, claim.nodeKey); + }); + + const node = claimReady.immediate(); + return node ? rowToWorkflowNode(node) : undefined; + } + + claimReadyNode(claim: WorkflowNodeClaim): WorkflowNodeRecord | undefined { + return this.claimNode(claim); + } + + transitionNode(transition: WorkflowNodeTransition): WorkflowNodeRecord { + const now = new Date().toISOString(); + const update = this.database.sqlite.transaction(() => { + const current = this.getNodeRow(transition.workflowId, transition.nodeKey); + if (!current) { + this.assertWorkflowExists(transition.workflowId); + throw new WorkflowValidationError(`Unknown workflow node: ${transition.nodeKey}`); + } + const workflow = this.getWorkflowRow(transition.workflowId); + if (!workflow) throw new WorkflowNotFoundError(transition.workflowId); + const workflowStatus = readWorkflowStatus(workflow.status); + if (TERMINAL_WORKFLOW_STATUSES.has(workflowStatus)) { + throw new WorkflowValidationError( + `Cannot transition node after workflow reached terminal status: ${workflowStatus}`, + ); + } + + const currentStatus = readNodeStatus(current.status); + if (!NODE_TRANSITIONS[currentStatus].has(transition.status)) { + throw new WorkflowTransitionError("node", currentStatus, transition.status); + } + if ( + currentStatus === "running" && + (!transition.claimToken || transition.claimToken !== current.claim_token) + ) { + throw new WorkflowValidationError("Running node transition requires the active claim token"); + } + + const terminal = TERMINAL_NODE_STATUSES.has(transition.status); + const resultJson = serializeOptionalJson(transition.result); + const errorJson = serializeOptionalObject(transition.error); + this.database.sqlite + .prepare( + `update workflow_nodes + set status = ?, result_json = ?, error_json = ?, updated_at = ?, completed_at = ?, + claim_expires_at = case when ? then null else claim_expires_at end + where id = ? and status = ?`, + ) + .run( + transition.status, + resultJson, + errorJson, + now, + terminal ? now : null, + terminal ? 1 : 0, + current.id, + currentStatus, + ); + this.insertEvent( + transition.workflowId, + `node.${transition.status}`, + current.id, + transition.eventPayload ?? { nodeKey: current.node_key, status: transition.status }, + now, + ); + return current.id; + }); + + return this.getNodeById(update.immediate())!; + } + + transitionWorkflow(transition: WorkflowTransition): WorkflowRunRecord { + const now = new Date().toISOString(); + const update = this.database.sqlite.transaction(() => { + const current = this.getWorkflowRow(transition.workflowId); + if (!current) throw new WorkflowNotFoundError(transition.workflowId); + const currentStatus = readWorkflowStatus(current.status); + if (!WORKFLOW_TRANSITIONS[currentStatus].has(transition.status)) { + throw new WorkflowTransitionError("workflow", currentStatus, transition.status); + } + + const terminal = TERMINAL_WORKFLOW_STATUSES.has(transition.status); + if (transition.status === "succeeded") { + const nonSuccessful = this.database.sqlite + .prepare( + `select node_key, status from workflow_nodes + where workflow_run_id = ? and status not in ('succeeded', 'skipped') + order by node_key + limit 1`, + ) + .get(transition.workflowId) as { node_key: string; status: string } | undefined; + if (nonSuccessful) { + throw new WorkflowValidationError( + `Cannot succeed workflow while node ${nonSuccessful.node_key} is ${nonSuccessful.status}`, + ); + } + } else if (transition.status === "failed" || transition.status === "cancelled") { + this.terminalizeOpenNodes(transition.workflowId, transition.status, now); + } + + this.database.sqlite + .prepare( + `update workflow_runs + set status = ?, result_json = ?, error_json = ?, updated_at = ?, + started_at = case when ? = 'running' then coalesce(started_at, ?) else started_at end, + completed_at = ? + where id = ? and status = ?`, + ) + .run( + transition.status, + serializeOptionalJson(transition.result), + serializeOptionalObject(transition.error), + now, + transition.status, + now, + terminal ? now : null, + transition.workflowId, + currentStatus, + ); + this.insertEvent( + transition.workflowId, + `workflow.${transition.status}`, + undefined, + transition.eventPayload ?? { status: transition.status }, + now, + ); + }); + + update.immediate(); + return this.require(transition.workflowId); + } + + requestCancellation(workflowId: string): WorkflowRunRecord { + const now = new Date().toISOString(); + const cancel = this.database.sqlite.transaction(() => { + const current = this.getWorkflowRow(workflowId); + if (!current) throw new WorkflowNotFoundError(workflowId); + const status = readWorkflowStatus(current.status); + if (TERMINAL_WORKFLOW_STATUSES.has(status) || status === "cancelling") return; + + this.database.sqlite + .prepare( + `update workflow_runs + set status = 'cancelling', cancellation_requested_at = ?, updated_at = ? + where id = ? and status = ?`, + ) + .run(now, now, workflowId, status); + this.insertEvent( + workflowId, + "workflow.cancellation_requested", + undefined, + { status: "cancelling" }, + now, + ); + }); + + cancel.immediate(); + return this.require(workflowId); + } + + appendEvent( + workflowId: string, + type: string, + payload: JsonObject, + nodeId?: string, + ): WorkflowEvent { + if (!type.trim()) throw new WorkflowValidationError("Workflow event type must not be empty"); + const now = new Date().toISOString(); + const append = this.database.sqlite.transaction(() => { + this.assertWorkflowExists(workflowId); + if (nodeId) this.assertNodeBelongsToWorkflow(workflowId, nodeId); + return this.insertEvent(workflowId, type, nodeId, payload, now); + }); + const sequence = append.immediate(); + return this.readEvents(workflowId, { after: sequence - 1, limit: 1 }).events[0]!; + } + + readEvents(workflowId: string, options: WorkflowEventReadOptions = {}): WorkflowEventPage { + this.assertWorkflowExists(workflowId); + const after = options.after ?? 0; + const limit = options.limit ?? DEFAULT_EVENT_LIMIT; + if (!Number.isSafeInteger(after) || after < 0) { + throw new WorkflowValidationError("Workflow event cursor must be a non-negative integer"); + } + if (!Number.isSafeInteger(limit) || limit < 1 || limit > MAX_EVENT_LIMIT) { + throw new WorkflowValidationError(`Workflow event limit must be between 1 and ${MAX_EVENT_LIMIT}`); + } + + const rows = this.database.sqlite + .prepare( + `select * from workflow_events + where workflow_run_id = ? and sequence > ? + order by sequence asc + limit ?`, + ) + .all(workflowId, after, limit) as WorkflowEventRow[]; + const events = rows.map(rowToWorkflowEvent); + return { events, nextCursor: events.at(-1)?.sequence ?? after }; + } + + close(): void { + this.database.close(); + } + + private hydrateWorkflow(row: WorkflowRunRow): WorkflowRunRecord { + const nodes = this.database.sqlite + .prepare("select * from workflow_nodes where workflow_run_id = ? order by created_at, node_key") + .all(row.id) as WorkflowNodeRow[]; + const edges = this.database.sqlite + .prepare( + `select e.workflow_run_id, e.from_node_id, e.to_node_id, + source.node_key as from_key, target.node_key as to_key + from workflow_edges e + join workflow_nodes source on source.id = e.from_node_id + join workflow_nodes target on target.id = e.to_node_id + where e.workflow_run_id = ? + order by source.node_key, target.node_key`, + ) + .all(row.id) as WorkflowEdgeRow[]; + + return { + id: row.id, + definitionVersion: readDefinitionVersion(row.definition_version), + status: readWorkflowStatus(row.status), + definition: parseJson(row.definition_json), + input: parseJson(row.input_json), + policy: parseJson(row.policy_json), + idempotencyKey: row.idempotency_key ?? undefined, + requestHash: row.request_hash, + result: parseOptionalJson(row.result_json), + error: parseOptionalJson(row.error_json), + cancellationRequestedAt: row.cancellation_requested_at ?? undefined, + createdAt: row.created_at, + updatedAt: row.updated_at, + startedAt: row.started_at ?? undefined, + completedAt: row.completed_at ?? undefined, + nodes: nodes.map(rowToWorkflowNode), + edges: edges.map(rowToWorkflowEdge), + }; + } + + private getWorkflowRow(workflowId: string): WorkflowRunRow | undefined { + return this.database.sqlite + .prepare("select * from workflow_runs where id = ?") + .get(workflowId) as WorkflowRunRow | undefined; + } + + private getNodeRow(workflowId: string, nodeKey: string): WorkflowNodeRow | undefined { + return this.database.sqlite + .prepare("select * from workflow_nodes where workflow_run_id = ? and node_key = ?") + .get(workflowId, nodeKey) as WorkflowNodeRow | undefined; + } + + private getNodeById(nodeId: string): WorkflowNodeRecord | undefined { + const row = this.database.sqlite + .prepare("select * from workflow_nodes where id = ?") + .get(nodeId) as WorkflowNodeRow | undefined; + return row ? rowToWorkflowNode(row) : undefined; + } + + private assertWorkflowExists(workflowId: string): void { + if (!this.getWorkflowRow(workflowId)) throw new WorkflowNotFoundError(workflowId); + } + + private assertNodeBelongsToWorkflow(workflowId: string, nodeId: string): void { + const row = this.database.sqlite + .prepare("select 1 from workflow_nodes where id = ? and workflow_run_id = ?") + .get(nodeId, workflowId); + if (!row) throw new WorkflowValidationError(`Node ${nodeId} does not belong to ${workflowId}`); + } + + private terminalizeOpenNodes( + workflowId: string, + workflowStatus: "failed" | "cancelled", + now: string, + ): void { + const rows = this.database.sqlite + .prepare( + `select * from workflow_nodes + where workflow_run_id = ? and status in ('pending', 'ready', 'running') + order by created_at, node_key`, + ) + .all(workflowId) as WorkflowNodeRow[]; + const update = this.database.sqlite.prepare( + `update workflow_nodes + set status = ?, claim_expires_at = null, updated_at = ?, completed_at = ? + where id = ? and status = ?`, + ); + for (const row of rows) { + const nodeStatus: WorkflowNodeStatus = + workflowStatus === "failed" && row.status !== "running" ? "skipped" : "cancelled"; + const changed = update.run(nodeStatus, now, now, row.id, row.status); + if (changed.changes !== 1) { + throw new WorkflowValidationError(`Workflow node changed during terminal transition: ${row.node_key}`); + } + this.insertEvent( + workflowId, + `node.${nodeStatus}`, + row.id, + { nodeKey: row.node_key, status: nodeStatus }, + now, + ); + } + } + + private insertEvent( + workflowId: string, + type: string, + nodeId: string | undefined, + payload: JsonObject, + now: string, + ): number { + const payloadJson = serializeEventPayload(payload); + const sequenceRow = this.database.sqlite + .prepare( + `update workflow_runs + set event_sequence = event_sequence + 1, updated_at = ? + where id = ? + returning event_sequence`, + ) + .get(now, workflowId) as { event_sequence: number } | undefined; + if (!sequenceRow) throw new WorkflowNotFoundError(workflowId); + + this.database.sqlite + .prepare( + `insert into workflow_events ( + workflow_run_id, sequence, event_type, node_id, payload_json, created_at + ) values (?, ?, ?, ?, ?, ?)`, + ) + .run(workflowId, sequenceRow.event_sequence, type, nodeId ?? null, payloadJson, now); + return sequenceRow.event_sequence; + } +} + +function normalizeSubmission(request: SubmitWorkflowRequest): { + definition: WorkflowDefinition; + definitionJson: string; + inputJson: string; + policyJson: string; + idempotencyKey?: string; + requestHash: string; +} { + const definition = normalizeDefinition(request.definition); + const input = normalizeObject(request.input ?? {}, "Workflow input"); + const policy = normalizePolicy(request.policy ?? { version: WORKFLOW_POLICY_VERSION }); + const definitionJson = canonicalJson(definition); + const inputJson = canonicalJson(input); + const policyJson = canonicalJson(policy); + const idempotencyKey = request.idempotencyKey?.trim(); + if (request.idempotencyKey !== undefined && !idempotencyKey) { + throw new WorkflowValidationError("Idempotency key must not be empty"); + } + const requestHash = createHash("sha256") + .update(canonicalJson({ definition, input, policy })) + .digest("hex"); + return { definition, definitionJson, inputJson, policyJson, idempotencyKey, requestHash }; +} + +function normalizeDefinition(definition: WorkflowDefinition): WorkflowDefinition { + if (!isObject(definition) || definition.version !== WORKFLOW_DEFINITION_VERSION) { + throw new WorkflowValidationError( + `Unsupported workflow definition version; expected ${WORKFLOW_DEFINITION_VERSION}`, + ); + } + if (!Array.isArray(definition.nodes) || definition.nodes.length === 0) { + throw new WorkflowValidationError("Workflow definition must contain at least one node"); + } + if (definition.edges !== undefined && !Array.isArray(definition.edges)) { + throw new WorkflowValidationError("Workflow definition edges must be an array"); + } + + const keys = new Set(); + const nodes = definition.nodes.map((node, index): WorkflowNodeDefinitionV1 => { + if (!isObject(node) || node.type !== "agent" || typeof node.key !== "string") { + throw new WorkflowValidationError(`Workflow node at index ${index} is not a valid agent node`); + } + const key = node.key.trim(); + if (!key) throw new WorkflowValidationError(`Workflow node at index ${index} has an empty key`); + if (keys.has(key)) throw new WorkflowValidationError(`Duplicate workflow node key: ${key}`); + keys.add(key); + return { key, type: "agent", config: normalizeObject(node.config ?? {}, `Node ${key} config`) }; + }); + + const edgeKeys = new Set(); + const edges = (definition.edges ?? []).map((edge, index) => { + if (!isObject(edge) || typeof edge.from !== "string" || typeof edge.to !== "string") { + throw new WorkflowValidationError(`Workflow edge at index ${index} is invalid`); + } + const from = edge.from.trim(); + const to = edge.to.trim(); + if (!keys.has(from)) throw new WorkflowValidationError(`Workflow edge references missing node: ${from}`); + if (!keys.has(to)) throw new WorkflowValidationError(`Workflow edge references missing node: ${to}`); + const edgeKey = canonicalJson([from, to]); + if (edgeKeys.has(edgeKey)) { + throw new WorkflowValidationError(`Duplicate workflow edge: ${from} -> ${to}`); + } + edgeKeys.add(edgeKey); + return { from, to }; + }); + + assertAcyclic(nodes.map((node) => node.key), edges); + return parseJson(canonicalJson({ version: WORKFLOW_DEFINITION_VERSION, nodes, edges })); +} + +function normalizePolicy(policy: WorkflowPolicy): WorkflowPolicy { + const normalized = normalizeObject(policy, "Workflow policy"); + if (normalized.version !== WORKFLOW_POLICY_VERSION) { + throw new WorkflowValidationError( + `Unsupported workflow policy version; expected ${WORKFLOW_POLICY_VERSION}`, + ); + } + return normalized as WorkflowPolicy; +} + +function assertAcyclic(nodeKeys: string[], edges: Array<{ from: string; to: string }>): void { + const incoming = new Map(nodeKeys.map((key) => [key, 0])); + const outgoing = new Map(nodeKeys.map((key) => [key, [] as string[]])); + for (const edge of edges) { + incoming.set(edge.to, incoming.get(edge.to)! + 1); + outgoing.get(edge.from)!.push(edge.to); + } + const ready = nodeKeys.filter((key) => incoming.get(key) === 0); + let visited = 0; + while (ready.length > 0) { + const key = ready.pop()!; + visited += 1; + for (const target of outgoing.get(key)!) { + const count = incoming.get(target)! - 1; + incoming.set(target, count); + if (count === 0) ready.push(target); + } + } + if (visited !== nodeKeys.length) throw new WorkflowValidationError("Workflow definition contains a cycle"); +} + +function normalizeObject(value: unknown, label: string): JsonObject { + if (!isObject(value)) throw new WorkflowValidationError(`${label} must be a JSON object`); + try { + return parseJson(canonicalJson(value)); + } catch (error) { + if (error instanceof WorkflowValidationError) throw error; + throw new WorkflowValidationError(`${label} must contain only JSON values`); + } +} + +function canonicalJson(value: unknown): string { + return JSON.stringify(normalizeJsonValue(value)); +} + +function normalizeJsonValue(value: unknown): JsonValue { + if (value === null || typeof value === "string" || typeof value === "boolean") return value; + if (typeof value === "number") { + if (!Number.isFinite(value)) throw new WorkflowValidationError("JSON numbers must be finite"); + return value; + } + if (Array.isArray(value)) return value.map(normalizeJsonValue); + if (isObject(value)) { + const normalized = Object.create(null) as JsonObject; + for (const key of Object.keys(value).sort()) { + const item = value[key]; + if (item === undefined) throw new WorkflowValidationError("JSON object values must not be undefined"); + normalized[key] = normalizeJsonValue(item); + } + return normalized; + } + throw new WorkflowValidationError("Value must contain only JSON data"); +} + +function serializeEventPayload(payload: JsonObject): string { + if (!isObject(payload)) throw new WorkflowValidationError("Workflow event payload must be a JSON object"); + const serialized = canonicalJson(payload); + if (Buffer.byteLength(serialized, "utf8") > MAX_EVENT_PAYLOAD_BYTES) { + throw new WorkflowValidationError( + `Workflow event payload exceeds ${MAX_EVENT_PAYLOAD_BYTES} bytes`, + ); + } + return serialized; +} + +function serializeOptionalJson(value: JsonValue | undefined): string | null { + return value === undefined ? null : canonicalJson(value); +} + +function serializeOptionalObject(value: JsonObject | undefined): string | null { + return value === undefined ? null : canonicalJson(normalizeObject(value, "Workflow error")); +} + +function parseJson(value: string): T { + return JSON.parse(value) as T; +} + +function parseOptionalJson(value: string | null): T | undefined { + return value === null ? undefined : parseJson(value); +} + +function rowToWorkflowNode(row: WorkflowNodeRow): WorkflowNodeRecord { + return { + id: row.id, + workflowId: row.workflow_run_id, + key: row.node_key, + type: readNodeType(row.node_type), + status: readNodeStatus(row.status), + definition: parseJson(row.definition_json), + attempt: row.attempt, + claimToken: row.claim_token ?? undefined, + claimedAt: row.claimed_at ?? undefined, + claimExpiresAt: row.claim_expires_at ?? undefined, + result: parseOptionalJson(row.result_json), + error: parseOptionalJson(row.error_json), + createdAt: row.created_at, + updatedAt: row.updated_at, + completedAt: row.completed_at ?? undefined, + }; +} + +function rowToWorkflowEdge(row: WorkflowEdgeRow): WorkflowEdgeRecord { + return { + workflowId: row.workflow_run_id, + fromNodeId: row.from_node_id, + toNodeId: row.to_node_id, + from: row.from_key, + to: row.to_key, + }; +} + +function rowToWorkflowEvent(row: WorkflowEventRow): WorkflowEvent { + return { + workflowId: row.workflow_run_id, + sequence: row.sequence, + type: row.event_type, + nodeId: row.node_id ?? undefined, + payload: parseJson(row.payload_json), + createdAt: row.created_at, + }; +} + +function readDefinitionVersion(version: number): typeof WORKFLOW_DEFINITION_VERSION { + if (version !== WORKFLOW_DEFINITION_VERSION) { + throw new WorkflowValidationError(`Unsupported stored workflow definition version: ${version}`); + } + return version; +} + +function readWorkflowStatus(status: string): WorkflowStatus { + if ( + status === "queued" || + status === "running" || + status === "cancelling" || + status === "succeeded" || + status === "failed" || + status === "cancelled" + ) { + return status; + } + throw new WorkflowValidationError(`Unsupported stored workflow status: ${status}`); +} + +function readNodeStatus(status: string): WorkflowNodeStatus { + if ( + status === "pending" || + status === "ready" || + status === "running" || + status === "succeeded" || + status === "failed" || + status === "cancelled" || + status === "skipped" + ) { + return status; + } + throw new WorkflowValidationError(`Unsupported stored workflow node status: ${status}`); +} + +function readNodeType(type: string): WorkflowNodeDefinitionV1["type"] { + if (type === "agent") return type; + throw new WorkflowValidationError(`Unsupported stored workflow node type: ${type}`); +} + +function createId(prefix: "wf_" | "wfn_"): string { + return `${prefix}${randomUUID().replaceAll("-", "")}`; +} + +function isObject(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/src/workflows/types.ts b/src/workflows/types.ts new file mode 100644 index 00000000..fb9659cf --- /dev/null +++ b/src/workflows/types.ts @@ -0,0 +1,159 @@ +export type JsonPrimitive = string | number | boolean | null; +export type JsonValue = JsonPrimitive | JsonObject | JsonValue[]; +export interface JsonObject { + [key: string]: JsonValue; +} + +export const WORKFLOW_DEFINITION_VERSION = 1 as const; +export const WORKFLOW_POLICY_VERSION = 1 as const; + +export type WorkflowStatus = + | "queued" + | "running" + | "cancelling" + | "succeeded" + | "failed" + | "cancelled"; + +export type WorkflowNodeStatus = + | "pending" + | "ready" + | "running" + | "succeeded" + | "failed" + | "cancelled" + | "skipped"; + +export interface AgentWorkflowNodeDefinitionV1 { + key: string; + type: "agent"; + config?: JsonObject; +} + +export type WorkflowNodeDefinitionV1 = AgentWorkflowNodeDefinitionV1; + +export interface WorkflowEdgeDefinitionV1 { + from: string; + to: string; +} + +export interface WorkflowDefinitionV1 { + version: typeof WORKFLOW_DEFINITION_VERSION; + nodes: WorkflowNodeDefinitionV1[]; + edges?: WorkflowEdgeDefinitionV1[]; +} + +export type WorkflowDefinition = WorkflowDefinitionV1; + +export interface WorkflowPolicyV1 extends JsonObject { + version: typeof WORKFLOW_POLICY_VERSION; +} + +export type WorkflowPolicy = WorkflowPolicyV1; + +export interface SubmitWorkflowRequest { + definition: WorkflowDefinition; + input?: JsonObject; + policy?: WorkflowPolicy; + idempotencyKey?: string; +} + +export interface WorkflowNodeRecord { + id: string; + workflowId: string; + key: string; + type: WorkflowNodeDefinitionV1["type"]; + status: WorkflowNodeStatus; + definition: WorkflowNodeDefinitionV1; + attempt: number; + claimToken?: string; + claimedAt?: string; + claimExpiresAt?: string; + result?: JsonValue; + error?: JsonObject; + createdAt: string; + updatedAt: string; + completedAt?: string; +} + +export interface WorkflowEdgeRecord { + workflowId: string; + fromNodeId: string; + toNodeId: string; + from: string; + to: string; +} + +export interface WorkflowRunRecord { + id: string; + definitionVersion: typeof WORKFLOW_DEFINITION_VERSION; + status: WorkflowStatus; + definition: WorkflowDefinition; + input: JsonObject; + policy: WorkflowPolicy; + idempotencyKey?: string; + requestHash: string; + result?: JsonValue; + error?: JsonObject; + cancellationRequestedAt?: string; + createdAt: string; + updatedAt: string; + startedAt?: string; + completedAt?: string; + nodes: WorkflowNodeRecord[]; + edges: WorkflowEdgeRecord[]; +} + +export interface WorkflowEvent { + workflowId: string; + sequence: number; + type: string; + nodeId?: string; + payload: JsonObject; + createdAt: string; +} + +export interface WorkflowEventPage { + events: WorkflowEvent[]; + nextCursor: number; +} + +export interface SubmitWorkflowResult { + workflow: WorkflowRunRecord; + created: boolean; +} + +export interface WorkflowWaitOptions { + timeoutMs?: number; + pollIntervalMs?: number; +} + +export interface WorkflowEventReadOptions { + after?: number; + limit?: number; +} + +export interface WorkflowNodeClaim { + workflowId: string; + nodeKey: string; + claimToken: string; + leaseMs?: number; +} + +export interface WorkflowNodeTransition { + workflowId: string; + nodeKey: string; + claimToken?: string; + status: WorkflowNodeStatus; + result?: JsonValue; + error?: JsonObject; + eventPayload?: JsonObject; +} + +export interface WorkflowTransition { + workflowId: string; + status: WorkflowStatus; + result?: JsonValue; + error?: JsonObject; + eventPayload?: JsonObject; +} diff --git a/src/workflows/workflows.test.ts b/src/workflows/workflows.test.ts new file mode 100644 index 00000000..1f894ab7 --- /dev/null +++ b/src/workflows/workflows.test.ts @@ -0,0 +1,600 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { existsSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { openDatabase } from "../db/client.js"; +import { WorkflowOrchestrator } from "./orchestrator.js"; +import { + WorkflowIdempotencyConflictError, + WorkflowStore, + WorkflowTransitionError, + WorkflowValidationError, +} from "./store.js"; +import type { SubmitWorkflowRequest, WorkflowDefinitionV1 } from "./types.js"; + +interface WorkerSpec { + action: "append" | "claim" | "submit"; + stateDir: string; + readyPath: string; + barrierPath: string; + workflowId?: string; + claimToken?: string; + start?: number; + count?: number; + requestVariant?: "a" | "b" | "same"; +} + +const workerSpecJson = process.argv[2]; +if (workerSpecJson) { + await runWorker(JSON.parse(workerSpecJson) as WorkerSpec); +} else { + await runTests(); +} + +async function runTests(): Promise { + const root = mkdtempSync(join(tmpdir(), "devspace-workflows-test-")); + try { + testSubmissionAndIdempotency(join(root, "submission")); + await testConcurrentSubmission(join(root, "concurrent-submission")); + testMonotonicCursorReads(join(root, "events")); + await testConcurrentEventSequencing(join(root, "concurrent-events")); + testCancellationIdempotency(join(root, "cancellation")); + await testTransitionsAndClaims(join(root, "transitions")); + testTerminalInvariantsAndAtomicity(join(root, "terminal-invariants")); + testDagValidation(join(root, "dag")); + await testBoundedAndRepeatedWait(join(root, "wait")); + } finally { + rmSync(root, { recursive: true, force: true }); + } +} + +function testSubmissionAndIdempotency(stateDir: string): void { + const store = new WorkflowStore(stateDir); + try { + const request: SubmitWorkflowRequest = { + definition: singleAgentDefinition({ z: true, a: "first" }), + input: { z: 2, a: { nested: true } }, + policy: { version: 1, retries: 0 }, + idempotencyKey: " durable-request ", + }; + const first = store.submit(request); + assert.equal(first.created, true); + assert.match(first.workflow.id, /^wf_[a-f0-9]{32}$/); + assert.match(first.workflow.nodes[0]!.id, /^wfn_[a-f0-9]{32}$/); + assert.equal(first.workflow.status, "queued"); + assert.equal(first.workflow.nodes[0]!.status, "ready"); + assert.equal(first.workflow.idempotencyKey, "durable-request"); + + request.input!.a = { nested: false }; + request.definition.nodes[0]!.config!.a = "mutated"; + const persisted = store.require(first.workflow.id); + assert.deepEqual(persisted.input, { a: { nested: true }, z: 2 }); + assert.deepEqual(persisted.definition.nodes[0]!.config, { a: "first", z: true }); + + const replay = store.submit({ + definition: singleAgentDefinition({ a: "first", z: true }), + input: { a: { nested: true }, z: 2 }, + policy: { retries: 0, version: 1 }, + idempotencyKey: "durable-request", + }); + assert.equal(replay.created, false); + assert.equal(replay.workflow.id, first.workflow.id); + + const specialInput = JSON.parse('{"__proto__":{"preserved":true}}') as Record; + const special = store.submit({ definition: singleAgentDefinition(), input: specialInput }).workflow; + assert.equal(Object.hasOwn(special.input, "__proto__"), true); + assert.deepEqual(special.input["__proto__"], { preserved: true }); + + assert.throws( + () => + store.submit({ + definition: singleAgentDefinition(), + input: { changed: true }, + idempotencyKey: "durable-request", + }), + WorkflowIdempotencyConflictError, + ); + } finally { + store.close(); + } +} + +async function testConcurrentSubmission(stateDir: string): Promise { + const same = await runWorkers(stateDir, [ + { action: "submit", requestVariant: "same" }, + { action: "submit", requestVariant: "same" }, + ]); + assert.equal(same[0]!.workflowId, same[1]!.workflowId); + assert.deepEqual( + same.map((result) => result.created).sort(), + [false, true], + ); + assertDatabaseCounts(stateDir, { runs: 1, nodes: 1, edges: 0, events: 1 }); + + const conflictDir = `${stateDir}-conflict`; + const conflict = await runWorkers(conflictDir, [ + { action: "submit", requestVariant: "a" }, + { action: "submit", requestVariant: "b" }, + ]); + assert.equal(conflict.filter((result) => result.created === true).length, 1); + assert.deepEqual( + conflict.filter((result) => result.error).map((result) => result.error), + ["WorkflowIdempotencyConflictError"], + ); + assertDatabaseCounts(conflictDir, { runs: 1, nodes: 1, edges: 0, events: 1 }); +} + +function testMonotonicCursorReads(stateDir: string): void { + const store = new WorkflowStore(stateDir); + try { + const workflow = store.submit({ definition: singleAgentDefinition() }).workflow; + store.appendEvent(workflow.id, "custom.first", { value: 1 }); + store.appendEvent(workflow.id, "custom.second", { value: 2 }); + + const firstPage = store.readEvents(workflow.id, { after: 0, limit: 2 }); + assert.deepEqual(firstPage.events.map((event) => event.sequence), [1, 2]); + assert.equal(firstPage.nextCursor, 2); + const secondPage = store.readEvents(workflow.id, { after: firstPage.nextCursor, limit: 2 }); + assert.deepEqual(secondPage.events.map((event) => event.sequence), [3]); + assert.equal(secondPage.nextCursor, 3); + assert.deepEqual( + store.readEvents(workflow.id, { after: 0, limit: 10 }).events.map((event) => event.sequence), + [1, 2, 3], + ); + + assert.throws( + () => store.appendEvent(workflow.id, "oversized", { body: "x".repeat(65_536) }), + WorkflowValidationError, + ); + assert.throws( + () => store.appendEvent(workflow.id, "invalid", [] as unknown as Record), + WorkflowValidationError, + ); + } finally { + store.close(); + } +} + +async function testConcurrentEventSequencing(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + const workflow = store.submit({ definition: singleAgentDefinition() }).workflow; + store.close(); + + await runWorkers( + stateDir, + Array.from({ length: 4 }, (_, worker) => ({ + action: "append" as const, + workflowId: workflow.id, + start: worker * 10, + count: 10, + })), + ); + + const reader = new WorkflowStore(stateDir); + try { + const events = reader.readEvents(workflow.id, { limit: 100 }).events; + assert.deepEqual( + events.map((event) => event.sequence), + Array.from({ length: 41 }, (_, index) => index + 1), + ); + assert.deepEqual( + events.slice(1).map((event) => event.payload.value).sort((a, b) => Number(a) - Number(b)), + Array.from({ length: 40 }, (_, index) => index), + ); + const database = openDatabase(stateDir); + try { + assert.equal( + database.sqlite.prepare("select event_sequence from workflow_runs where id = ?").pluck().get(workflow.id), + 41, + ); + } finally { + database.close(); + } + } finally { + reader.close(); + } +} + +function testCancellationIdempotency(stateDir: string): void { + const store = new WorkflowStore(stateDir); + try { + const workflow = store.submit({ definition: singleAgentDefinition() }).workflow; + const first = store.requestCancellation(workflow.id); + const firstEvents = store.readEvents(workflow.id, { limit: 100 }).events; + const second = store.requestCancellation(workflow.id); + const secondEvents = store.readEvents(workflow.id, { limit: 100 }).events; + + assert.equal(first.status, "cancelling"); + assert.ok(first.cancellationRequestedAt); + assert.equal(second.cancellationRequestedAt, first.cancellationRequestedAt); + assert.equal(secondEvents.length, firstEvents.length); + assert.equal(secondEvents.at(-1)!.type, "workflow.cancellation_requested"); + + const cancelled = store.transitionWorkflow({ workflowId: workflow.id, status: "cancelled" }); + assert.equal(cancelled.nodes[0]!.status, "cancelled"); + assert.equal(store.requestCancellation(workflow.id).status, "cancelled"); + assert.equal(store.require(workflow.id).completedAt, cancelled.completedAt); + } finally { + store.close(); + } +} + +async function testTransitionsAndClaims(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + try { + const workflow = store.submit({ definition: singleAgentDefinition() }).workflow; + const raced = await runWorkers(stateDir, [ + { action: "claim", workflowId: workflow.id, claimToken: "claim-one" }, + { action: "claim", workflowId: workflow.id, claimToken: "claim-two" }, + ]); + const winner = raced.find((result) => result.claimed)!.claimToken as string; + const loser = winner === "claim-one" ? "claim-two" : "claim-one"; + assert.equal(raced.filter((result) => result.claimed).length, 1); + + const claimed = store.require(workflow.id).nodes[0]!; + assert.equal(claimed.status, "running"); + assert.equal(claimed.attempt, 1); + assert.equal(claimed.claimToken, winner); + const eventCount = store.readEvents(workflow.id, { limit: 100 }).events.length; + const replay = store.claimNode({ workflowId: workflow.id, nodeKey: "agent", claimToken: winner }); + assert.equal(replay!.attempt, 1); + assert.equal(store.readEvents(workflow.id, { limit: 100 }).events.length, eventCount); + + assert.throws( + () => + store.transitionNode({ + workflowId: workflow.id, + nodeKey: "agent", + claimToken: loser, + status: "succeeded", + }), + WorkflowValidationError, + ); + const completedNode = store.transitionNode({ + workflowId: workflow.id, + nodeKey: "agent", + claimToken: winner, + status: "succeeded", + result: { answer: 42 }, + }); + assert.deepEqual(completedNode.result, { answer: 42 }); + assert.ok(completedNode.completedAt); + assert.throws( + () => + store.transitionNode({ + workflowId: workflow.id, + nodeKey: "agent", + claimToken: winner, + status: "failed", + }), + WorkflowTransitionError, + ); + + const completed = store.transitionWorkflow({ + workflowId: workflow.id, + status: "succeeded", + result: { answer: 42 }, + }); + assert.deepEqual(completed.result, { answer: 42 }); + assert.ok(completed.completedAt); + assert.equal(store.readEvents(workflow.id, { limit: 100 }).events.at(-1)!.type, "workflow.succeeded"); + + const abandoned = store.submit({ definition: singleAgentDefinition() }).workflow; + const firstClaim = store.claimNode({ + workflowId: abandoned.id, + nodeKey: "agent", + claimToken: "abandoned", + leaseMs: 10, + }); + assert.equal(firstClaim!.attempt, 1); + await delay(20); + const recovered = store.claimNode({ + workflowId: abandoned.id, + nodeKey: "agent", + claimToken: "replacement", + }); + assert.equal(recovered!.attempt, 2); + assert.equal(recovered!.claimToken, "replacement"); + assert.equal(store.readEvents(abandoned.id, { limit: 100 }).events.at(-1)!.type, "node.reclaimed"); + } finally { + store.close(); + } +} + +function testTerminalInvariantsAndAtomicity(stateDir: string): void { + const store = new WorkflowStore(stateDir); + try { + const workflow = store.submit({ definition: singleAgentDefinition() }).workflow; + store.claimNode({ workflowId: workflow.id, nodeKey: "agent", claimToken: "claim" }); + assert.throws( + () => store.transitionWorkflow({ workflowId: workflow.id, status: "succeeded" }), + /Cannot succeed workflow while node agent is running/, + ); + + const beforeNodeFailure = store.require(workflow.id); + const beforeNodeEvents = store.readEvents(workflow.id, { limit: 100 }).events; + assert.throws( + () => + store.transitionNode({ + workflowId: workflow.id, + nodeKey: "agent", + claimToken: "claim", + status: "succeeded", + result: { persisted: false }, + eventPayload: { body: "x".repeat(65_536) }, + }), + WorkflowValidationError, + ); + assert.deepEqual(store.require(workflow.id), beforeNodeFailure); + assert.deepEqual(store.readEvents(workflow.id, { limit: 100 }).events, beforeNodeEvents); + + store.transitionNode({ + workflowId: workflow.id, + nodeKey: "agent", + claimToken: "claim", + status: "succeeded", + }); + const beforeWorkflowFailure = store.require(workflow.id); + const beforeWorkflowEvents = store.readEvents(workflow.id, { limit: 100 }).events; + assert.throws( + () => + store.transitionWorkflow({ + workflowId: workflow.id, + status: "succeeded", + result: { persisted: false }, + eventPayload: { body: "x".repeat(65_536) }, + }), + WorkflowValidationError, + ); + assert.deepEqual(store.require(workflow.id), beforeWorkflowFailure); + assert.deepEqual(store.readEvents(workflow.id, { limit: 100 }).events, beforeWorkflowEvents); + + const failedWorkflow = store.submit({ definition: singleAgentDefinition() }).workflow; + store.claimNode({ workflowId: failedWorkflow.id, nodeKey: "agent", claimToken: "active" }); + const failed = store.transitionWorkflow({ workflowId: failedWorkflow.id, status: "failed" }); + assert.equal(failed.nodes[0]!.status, "cancelled"); + assert.throws( + () => + store.transitionNode({ + workflowId: failed.id, + nodeKey: "agent", + claimToken: "active", + status: "failed", + }), + /Cannot transition node after workflow reached terminal status/, + ); + } finally { + store.close(); + } +} + +function testDagValidation(stateDir: string): void { + const store = new WorkflowStore(stateDir); + try { + assert.throws( + () => + store.submit({ + definition: { version: 1, nodes: [agentNode("same"), agentNode("same")], edges: [] }, + }), + /Duplicate workflow node key/, + ); + assert.throws( + () => + store.submit({ + definition: { + version: 1, + nodes: [agentNode("one")], + edges: [{ from: "one", to: "missing" }], + }, + }), + /missing node/, + ); + assert.throws( + () => + store.submit({ + definition: { + version: 1, + nodes: [agentNode("one"), agentNode("two")], + edges: [ + { from: "one", to: "two" }, + { from: "two", to: "one" }, + ], + }, + }), + /contains a cycle/, + ); + + const dag = store.submit({ + definition: { + version: 1, + nodes: [agentNode("one"), agentNode("two"), agentNode("three")], + edges: [ + { from: "one", to: "two" }, + { from: "one", to: "three" }, + ], + }, + }).workflow; + assert.deepEqual(Object.fromEntries(dag.nodes.map((node) => [node.key, node.status])), { + one: "ready", + three: "pending", + two: "pending", + }); + + const nulKeys = store.submit({ + definition: { + version: 1, + nodes: [agentNode("a\0b"), agentNode("a"), agentNode("b\0c"), agentNode("c")], + edges: [ + { from: "a\0b", to: "c" }, + { from: "a", to: "b\0c" }, + ], + }, + }).workflow; + assert.equal(nulKeys.edges.length, 2); + } finally { + store.close(); + } +} + +async function testBoundedAndRepeatedWait(stateDir: string): Promise { + const orchestrator = new WorkflowOrchestrator(stateDir); + const workerStore = new WorkflowStore(stateDir); + try { + const workflow = orchestrator.submit({ definition: singleAgentDefinition() }); + const timeoutStart = performance.now(); + const timedOut = await orchestrator.wait(workflow.id, { timeoutMs: 100, pollIntervalMs: 10 }); + const timeoutElapsed = performance.now() - timeoutStart; + assert.equal(timedOut.status, "queued"); + assert.ok(timeoutElapsed >= 70, `bounded wait returned early after ${timeoutElapsed}ms`); + assert.ok(timeoutElapsed < 500, `bounded wait took ${timeoutElapsed}ms`); + + const realDateNow = Date.now; + const jumpTimer = setTimeout(() => { + Date.now = () => realDateNow() + 60 * 60_000; + }, 20); + try { + const jumpStart = performance.now(); + await orchestrator.wait(workflow.id, { timeoutMs: 100, pollIntervalMs: 10 }); + assert.ok(performance.now() - jumpStart >= 70, "wall-clock jump ended wait early"); + } finally { + clearTimeout(jumpTimer); + Date.now = realDateNow; + } + + setTimeout(() => { + workerStore.transitionWorkflow({ workflowId: workflow.id, status: "failed" }); + }, 15); + const completed = await orchestrator.wait(workflow.id, { timeoutMs: 500, pollIntervalMs: 5 }); + assert.equal(completed.status, "failed"); + + const repeatStart = performance.now(); + const repeated = await orchestrator.wait(workflow.id, { timeoutMs: 500, pollIntervalMs: 5 }); + assert.equal(repeated.status, "failed"); + assert.ok(performance.now() - repeatStart < 100); + } finally { + workerStore.close(); + orchestrator.close(); + } + + const closingOrchestrator = new WorkflowOrchestrator(`${stateDir}-closing`); + const closingWorkflow = closingOrchestrator.submit({ definition: singleAgentDefinition() }); + const pendingWait = closingOrchestrator.wait(closingWorkflow.id, { + timeoutMs: 500, + pollIntervalMs: 50, + }); + closingOrchestrator.close(); + assert.equal((await pendingWait).status, "queued"); + closingOrchestrator.close(); +} + +async function runWorker(spec: WorkerSpec): Promise { + const store = new WorkflowStore(spec.stateDir); + try { + writeFileSync(spec.readyPath, "ready"); + while (!existsSync(spec.barrierPath)) await delay(2); + if (spec.action === "submit") { + const input = spec.requestVariant === "b" ? { variant: "b" } : { variant: "a" }; + const request: SubmitWorkflowRequest = { + definition: singleAgentDefinition(), + input, + idempotencyKey: spec.requestVariant === "same" ? "same-request" : "conflicting-request", + }; + if (spec.requestVariant === "same") request.input = { variant: "same" }; + try { + const result = store.submit(request); + process.stdout.write(JSON.stringify({ workflowId: result.workflow.id, created: result.created })); + } catch (error) { + process.stdout.write(JSON.stringify({ error: (error as Error).name })); + } + return; + } + if (spec.action === "claim") { + const claimed = store.claimNode({ + workflowId: spec.workflowId!, + nodeKey: "agent", + claimToken: spec.claimToken!, + }); + process.stdout.write(JSON.stringify({ claimed: Boolean(claimed), claimToken: spec.claimToken })); + return; + } + for (let index = 0; index < spec.count!; index += 1) { + const value = spec.start! + index; + store.appendEvent(spec.workflowId!, `concurrent.${value}`, { value }); + } + process.stdout.write(JSON.stringify({ appended: spec.count })); + } finally { + store.close(); + } +} + +async function runWorkers( + stateDir: string, + workers: Array>, +): Promise>> { + const coordinationDir = mkdtempSync(join(tmpdir(), "devspace-workflow-race-")); + const barrierPath = join(coordinationDir, "start"); + const running = workers.map((worker, index) => { + const readyPath = join(coordinationDir, `ready-${index}`); + const spec: WorkerSpec = { ...worker, stateDir, readyPath, barrierPath }; + const child = spawn(process.execPath, ["--import", "tsx", fileURLToPath(import.meta.url), JSON.stringify(spec)], { + stdio: ["ignore", "pipe", "pipe"], + }); + let stdout = ""; + let stderr = ""; + child.stdout.setEncoding("utf8").on("data", (chunk) => (stdout += chunk)); + child.stderr.setEncoding("utf8").on("data", (chunk) => (stderr += chunk)); + const result = new Promise>((resolve, reject) => { + child.on("error", reject); + child.on("exit", (code) => { + if (code !== 0) reject(new Error(`Workflow worker exited ${code}: ${stderr}`)); + else resolve(JSON.parse(stdout) as Record); + }); + }); + return { readyPath, result }; + }); + try { + const deadline = performance.now() + 10_000; + while (running.some(({ readyPath }) => !existsSync(readyPath))) { + if (performance.now() >= deadline) throw new Error("Timed out waiting for workflow workers"); + await delay(5); + } + writeFileSync(barrierPath, "start"); + return await Promise.all(running.map(({ result }) => result)); + } finally { + rmSync(coordinationDir, { recursive: true, force: true }); + } +} + +function assertDatabaseCounts( + stateDir: string, + expected: { runs: number; nodes: number; edges: number; events: number }, +): void { + const database = openDatabase(stateDir); + try { + assert.deepEqual( + { + runs: database.sqlite.prepare("select count(*) from workflow_runs").pluck().get(), + nodes: database.sqlite.prepare("select count(*) from workflow_nodes").pluck().get(), + edges: database.sqlite.prepare("select count(*) from workflow_edges").pluck().get(), + events: database.sqlite.prepare("select count(*) from workflow_events").pluck().get(), + }, + expected, + ); + } finally { + database.close(); + } +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +function singleAgentDefinition(config: Record = {}): WorkflowDefinitionV1 { + return { version: 1, nodes: [{ key: "agent", type: "agent", config }], edges: [] }; +} + +function agentNode(key: string): WorkflowDefinitionV1["nodes"][number] { + return { key, type: "agent", config: {} }; +}