From bf86c909d8c9efbe4d74872f2b1f72f70c767030 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 01:31:16 +0530 Subject: [PATCH 1/4] feat(workflows): add supervisor persistence primitives Co-Authored-By: Claude --- src/db/client.ts | 2 +- src/db/migrations.ts | 80 +++- src/db/schema.ts | 71 ++++ src/workflows/orchestrator.ts | 42 +- src/workflows/store.ts | 715 +++++++++++++++++++++++++++++++++- src/workflows/types.ts | 52 +++ 6 files changed, 952 insertions(+), 10 deletions(-) diff --git a/src/db/client.ts b/src/db/client.ts index fd129fbe..e2264091 100644 --- a/src/db/client.ts +++ b/src/db/client.ts @@ -24,9 +24,9 @@ export function openDatabase(stateDir: string): DatabaseHandle { const path = databasePath(stateDir); const sqlite = new Database(path); chmodSync(path, 0o600); + sqlite.pragma("busy_timeout = 5000"); sqlite.pragma("journal_mode = WAL"); sqlite.pragma("synchronous = NORMAL"); - sqlite.pragma("busy_timeout = 5000"); sqlite.pragma("foreign_keys = ON"); migrateDatabase(sqlite); diff --git a/src/db/migrations.ts b/src/db/migrations.ts index 9fe3956c..727ca990 100644 --- a/src/db/migrations.ts +++ b/src/db/migrations.ts @@ -27,6 +27,11 @@ const migrations: Migration[] = [ name: "durable-workflows", up: migrateDurableWorkflows, }, + { + version: 5, + name: "workflow-supervisor", + up: migrateWorkflowSupervisor, + }, ]; export function migrateDatabase(sqlite: Database.Database): void { @@ -264,9 +269,82 @@ function migrateDurableWorkflows(sqlite: Database.Database): void { `); } +function migrateWorkflowSupervisor(sqlite: Database.Database): void { + addColumnIfMissing(sqlite, "workflow_runs", "workspace_id", "text"); + addColumnIfMissing(sqlite, "workflow_runs", "workspace_root", "text"); + addColumnIfMissing(sqlite, "workflow_nodes", "supervisor_owner_token", "text"); + addColumnIfMissing(sqlite, "workflow_nodes", "supervisor_owner_epoch", "integer"); + addColumnIfMissing(sqlite, "workflow_nodes", "heartbeat_at", "text"); + + sqlite.exec(` + create table if not exists workflow_supervisor ( + id integer primary key check (id = 1), + owner_token text, + owner_epoch integer not null default 0, + owner_pid integer, + status text not null default 'stopped' + check (status in ('stopped', 'starting', 'running', 'stopping')), + lease_expires_at text, + heartbeat_at text, + wake_generation integer not null default 0, + started_at text, + last_error text + ); + + insert or ignore into workflow_supervisor (id) values (1); + + create table if not exists workflow_node_attempts ( + node_id text not null, + workflow_run_id text not null, + node_key text not null, + attempt integer not null, + claim_token text not null, + supervisor_owner_token text not null, + supervisor_owner_epoch integer not null, + provider text not null, + phase text not null check (phase in ('claimed', 'dispatching', 'running', 'cancelling', 'terminal')), + provider_session_id text, + heartbeat_at text, + cancellation_requested_at text, + terminal_status text check (terminal_status in ('succeeded', 'failed', 'cancelled')), + result_json text, + error_json text, + created_at text not null, + updated_at text not null, + completed_at text, + primary key (node_id, attempt), + unique (node_id, attempt, claim_token), + foreign key (workflow_run_id, node_id) + references workflow_nodes(workflow_run_id, id) on delete cascade + ); + + create index if not exists workflow_node_attempts_run_idx + on workflow_node_attempts(workflow_run_id, node_key, attempt); + + create table if not exists workflow_provider_events ( + node_id text not null, + attempt integer not null, + source_sequence integer not null, + workflow_sequence integer not null, + event_type text not null, + payload_json text not null, + created_at text not null, + primary key (node_id, attempt, source_sequence), + foreign key (node_id, attempt) + references workflow_node_attempts(node_id, attempt) on delete cascade + ); + + create index if not exists workflow_nodes_claimable_idx + on workflow_nodes(status, claim_expires_at, created_at); + + create index if not exists workflow_runs_workspace_idx + on workflow_runs(workspace_id, workspace_root, created_at); + `); +} + function addColumnIfMissing( sqlite: Database.Database, - table: "workspace_sessions" | "local_agent_sessions", + table: "workspace_sessions" | "local_agent_sessions" | "workflow_runs" | "workflow_nodes", column: string, definition: string, ): void { diff --git a/src/db/schema.ts b/src/db/schema.ts index 631e6d8c..a95107c3 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -116,6 +116,8 @@ export const workflowRuns = sqliteTable( policyJson: text("policy_json").notNull(), idempotencyKey: text("idempotency_key").unique(), requestHash: text("request_hash").notNull(), + workspaceId: text("workspace_id"), + workspaceRoot: text("workspace_root"), resultJson: text("result_json"), errorJson: text("error_json"), cancellationRequestedAt: text("cancellation_requested_at"), @@ -143,6 +145,9 @@ export const workflowNodes = sqliteTable( claimToken: text("claim_token"), claimedAt: text("claimed_at"), claimExpiresAt: text("claim_expires_at"), + supervisorOwnerToken: text("supervisor_owner_token"), + supervisorOwnerEpoch: integer("supervisor_owner_epoch"), + heartbeatAt: text("heartbeat_at"), resultJson: text("result_json"), errorJson: text("error_json"), createdAt: text("created_at").notNull(), @@ -201,6 +206,72 @@ export const workflowEvents = sqliteTable( ], ); +export const workflowSupervisor = sqliteTable("workflow_supervisor", { + id: integer("id").primaryKey(), + ownerToken: text("owner_token"), + ownerEpoch: integer("owner_epoch").notNull().default(0), + ownerPid: integer("owner_pid"), + status: text("status").notNull().default("stopped"), + leaseExpiresAt: text("lease_expires_at"), + heartbeatAt: text("heartbeat_at"), + wakeGeneration: integer("wake_generation").notNull().default(0), + startedAt: text("started_at"), + lastError: text("last_error"), +}); + +export const workflowNodeAttempts = sqliteTable( + "workflow_node_attempts", + { + nodeId: text("node_id").notNull(), + workflowRunId: text("workflow_run_id").notNull(), + nodeKey: text("node_key").notNull(), + attempt: integer("attempt").notNull(), + claimToken: text("claim_token").notNull(), + supervisorOwnerToken: text("supervisor_owner_token").notNull(), + supervisorOwnerEpoch: integer("supervisor_owner_epoch").notNull(), + provider: text("provider").notNull(), + phase: text("phase").notNull(), + providerSessionId: text("provider_session_id"), + heartbeatAt: text("heartbeat_at"), + cancellationRequestedAt: text("cancellation_requested_at"), + terminalStatus: text("terminal_status"), + resultJson: text("result_json"), + errorJson: text("error_json"), + createdAt: text("created_at").notNull(), + updatedAt: text("updated_at").notNull(), + completedAt: text("completed_at"), + }, + (table) => [ + primaryKey({ columns: [table.nodeId, table.attempt] }), + uniqueIndex("workflow_node_attempts_claim_idx").on(table.nodeId, table.attempt, table.claimToken), + foreignKey({ + columns: [table.workflowRunId, table.nodeId], + foreignColumns: [workflowNodes.workflowRunId, workflowNodes.id], + }).onDelete("cascade"), + index("workflow_node_attempts_run_idx").on(table.workflowRunId, table.nodeKey, table.attempt), + ], +); + +export const workflowProviderEvents = sqliteTable( + "workflow_provider_events", + { + nodeId: text("node_id").notNull(), + attempt: integer("attempt").notNull(), + sourceSequence: integer("source_sequence").notNull(), + workflowSequence: integer("workflow_sequence").notNull(), + eventType: text("event_type").notNull(), + payloadJson: text("payload_json").notNull(), + createdAt: text("created_at").notNull(), + }, + (table) => [ + primaryKey({ columns: [table.nodeId, table.attempt, table.sourceSequence] }), + foreignKey({ + columns: [table.nodeId, table.attempt], + foreignColumns: [workflowNodeAttempts.nodeId, workflowNodeAttempts.attempt], + }).onDelete("cascade"), + ], +); + export type WorkspaceSessionRow = typeof workspaceSessions.$inferSelect; export type NewWorkspaceSessionRow = typeof workspaceSessions.$inferInsert; export type LoadedAgentFileRow = typeof loadedAgentFiles.$inferSelect; diff --git a/src/workflows/orchestrator.ts b/src/workflows/orchestrator.ts index c2ab164b..bca3df1e 100644 --- a/src/workflows/orchestrator.ts +++ b/src/workflows/orchestrator.ts @@ -3,13 +3,15 @@ import type { SubmitWorkflowRequest, WorkflowEventPage, WorkflowEventReadOptions, + SubmitWorkflowResult, WorkflowRunRecord, WorkflowWaitOptions, + WorkflowWorkspaceScope, } from "./types.js"; import { WorkflowStore, WorkflowValidationError } from "./store.js"; const DEFAULT_WAIT_TIMEOUT_MS = 30_000; -const MAX_WAIT_TIMEOUT_MS = 60_000; +const MAX_WAIT_TIMEOUT_MS = 300_000; const DEFAULT_POLL_INTERVAL_MS = 50; const MAX_POLL_INTERVAL_MS = 1_000; @@ -23,15 +25,23 @@ export class WorkflowOrchestrator { } submit(request: SubmitWorkflowRequest): WorkflowRunRecord { - const workflow = this.store.submit(request).workflow; - this.notify(workflow.id); - return workflow; + return this.submitDetailed(request).workflow; + } + + submitDetailed(request: SubmitWorkflowRequest): SubmitWorkflowResult { + const result = this.store.submit(request); + this.notify(result.workflow.id); + return result; } get(workflowId: string): WorkflowRunRecord | undefined { return this.store.get(workflowId); } + getForWorkspace(workflowId: string, scope: WorkflowWorkspaceScope): WorkflowRunRecord | undefined { + return this.store.getForWorkspace(workflowId, scope); + } + async wait(workflowId: string, options: WorkflowWaitOptions = {}): Promise { const timeoutMs = options.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS; const pollIntervalMs = options.pollIntervalMs ?? DEFAULT_POLL_INTERVAL_MS; @@ -54,16 +64,40 @@ export class WorkflowOrchestrator { return workflow; } + async waitForWorkspace( + workflowId: string, + scope: WorkflowWorkspaceScope, + options: WorkflowWaitOptions = {}, + ): Promise { + this.store.requireForWorkspace(workflowId, scope); + const workflow = await this.wait(workflowId, options); + return this.store.requireForWorkspace(workflow.id, scope); + } + events(workflowId: string, options: WorkflowEventReadOptions = {}): WorkflowEventPage { return this.store.readEvents(workflowId, options); } + eventsForWorkspace( + workflowId: string, + scope: WorkflowWorkspaceScope, + options: WorkflowEventReadOptions = {}, + ): WorkflowEventPage { + this.store.requireForWorkspace(workflowId, scope); + return this.store.readEvents(workflowId, options); + } + cancel(workflowId: string): WorkflowRunRecord { const workflow = this.store.requestCancellation(workflowId); this.notify(workflowId); return workflow; } + cancelForWorkspace(workflowId: string, scope: WorkflowWorkspaceScope): WorkflowRunRecord { + this.store.requireForWorkspace(workflowId, scope); + return this.cancel(workflowId); + } + close(): void { if (this.closed) return; this.closed = true; diff --git a/src/workflows/store.ts b/src/workflows/store.ts index 78099d7f..bdb3cd14 100644 --- a/src/workflows/store.ts +++ b/src/workflows/store.ts @@ -13,7 +13,10 @@ import { type WorkflowEvent, type WorkflowEventPage, type WorkflowEventReadOptions, + type WorkflowAttemptIdentity, + type WorkflowNodeAttemptRecord, type WorkflowNodeClaim, + type WorkflowNodeClaimResult, type WorkflowNodeDefinitionV1, type WorkflowNodeRecord, type WorkflowNodeStatus, @@ -21,7 +24,10 @@ import { type WorkflowPolicy, type WorkflowRunRecord, type WorkflowStatus, + type WorkflowSupervisorIdentity, + type WorkflowSupervisorRecord, type WorkflowTransition, + type WorkflowWorkspaceScope, } from "./types.js"; const MAX_EVENT_PAYLOAD_BYTES = 64 * 1024; @@ -70,6 +76,8 @@ interface WorkflowRunRow { policy_json: string; idempotency_key: string | null; request_hash: string; + workspace_id: string | null; + workspace_root: string | null; result_json: string | null; error_json: string | null; cancellation_requested_at: string | null; @@ -90,6 +98,9 @@ interface WorkflowNodeRow { claim_token: string | null; claimed_at: string | null; claim_expires_at: string | null; + supervisor_owner_token: string | null; + supervisor_owner_epoch: number | null; + heartbeat_at: string | null; result_json: string | null; error_json: string | null; created_at: string; @@ -114,6 +125,38 @@ interface WorkflowEventRow { created_at: string; } +interface WorkflowSupervisorRow { + owner_token: string | null; + owner_epoch: number; + owner_pid: number | null; + status: WorkflowSupervisorRecord["status"]; + lease_expires_at: string | null; + heartbeat_at: string | null; + wake_generation: number; + started_at: string | null; +} + +interface WorkflowAttemptRow { + node_id: string; + workflow_run_id: string; + node_key: string; + attempt: number; + claim_token: string; + supervisor_owner_token: string; + supervisor_owner_epoch: number; + provider: string; + phase: WorkflowNodeAttemptRecord["phase"]; + provider_session_id: string | null; + heartbeat_at: string | null; + cancellation_requested_at: string | null; + terminal_status: WorkflowNodeAttemptRecord["terminalStatus"] | null; + result_json: string | null; + error_json: string | null; + created_at: string; + updated_at: string; + completed_at: string | null; +} + export class WorkflowNotFoundError extends Error { constructor(workflowId: string) { super(`Unknown workflow: ${workflowId}`); @@ -172,8 +215,8 @@ export class WorkflowStore { .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', ?, ?, ?, ?, ?, ?, ?)`, + idempotency_key, request_hash, workspace_id, workspace_root, created_at, updated_at + ) values (?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ) .run( workflowId, @@ -183,6 +226,8 @@ export class WorkflowStore { normalized.policyJson, normalized.idempotencyKey ?? null, normalized.requestHash, + normalized.workspace?.workspaceId ?? null, + normalized.workspace?.workspaceRoot ?? null, now, now, ); @@ -241,6 +286,21 @@ export class WorkflowStore { return workflow; } + getForWorkspace(workflowId: string, scope: WorkflowWorkspaceScope): WorkflowRunRecord | undefined { + const workflow = this.get(workflowId); + if (!workflow) return undefined; + if (workflow.workspaceId !== scope.workspaceId || workflow.workspaceRoot !== scope.workspaceRoot) { + return undefined; + } + return workflow; + } + + requireForWorkspace(workflowId: string, scope: WorkflowWorkspaceScope): WorkflowRunRecord { + const workflow = this.getForWorkspace(workflowId, scope); + 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; @@ -505,10 +565,559 @@ export class WorkflowStore { return { events, nextCursor: events.at(-1)?.sequence ?? after }; } + requestSupervisorWake(): number { + const row = this.database.sqlite + .prepare( + `update workflow_supervisor set wake_generation = wake_generation + 1 where id = 1 + returning wake_generation`, + ) + .get() as { wake_generation: number }; + return row.wake_generation; + } + + getSupervisor(): WorkflowSupervisorRecord | undefined { + const row = this.database.sqlite + .prepare("select * from workflow_supervisor where id = 1") + .get() as WorkflowSupervisorRow | undefined; + return row?.owner_token ? rowToSupervisor(row) : undefined; + } + + acquireSupervisor(input: { + ownerToken: string; + ownerPid: number; + leaseMs: number; + }): WorkflowSupervisorRecord | undefined { + validateLease(input.leaseMs, "Supervisor"); + if (!input.ownerToken) throw new WorkflowValidationError("Supervisor owner token must not be empty"); + const nowDate = new Date(); + const now = nowDate.toISOString(); + const expiresAt = new Date(nowDate.getTime() + input.leaseMs).toISOString(); + const acquire = this.database.sqlite.transaction(() => { + const current = this.database.sqlite + .prepare("select * from workflow_supervisor where id = 1") + .get() as WorkflowSupervisorRow; + if (current.owner_token && current.lease_expires_at && current.lease_expires_at > now) { + return undefined; + } + const epoch = current.owner_epoch + 1; + this.database.sqlite + .prepare( + `update workflow_supervisor + set owner_token = ?, owner_epoch = ?, owner_pid = ?, status = 'running', + lease_expires_at = ?, heartbeat_at = ?, started_at = ?, last_error = null + where id = 1`, + ) + .run(input.ownerToken, epoch, input.ownerPid, expiresAt, now, now); + return this.getSupervisor(); + }); + return acquire.immediate(); + } + + heartbeatSupervisor(identity: WorkflowSupervisorIdentity, leaseMs: number): boolean { + validateLease(leaseMs, "Supervisor"); + const nowDate = new Date(); + const now = nowDate.toISOString(); + const expiresAt = new Date(nowDate.getTime() + leaseMs).toISOString(); + const result = this.database.sqlite + .prepare( + `update workflow_supervisor set heartbeat_at = ?, lease_expires_at = ?, status = 'running' + where id = 1 and owner_token = ? and owner_epoch = ? and lease_expires_at > ?`, + ) + .run(now, expiresAt, identity.ownerToken, identity.ownerEpoch, now); + return result.changes === 1; + } + + releaseSupervisor(identity: WorkflowSupervisorIdentity, expectedWakeGeneration?: number): boolean { + const now = new Date().toISOString(); + const wakeFence = expectedWakeGeneration === undefined ? "" : " and wake_generation = ?"; + const result = this.database.sqlite + .prepare( + `update workflow_supervisor + set owner_token = null, owner_pid = null, status = 'stopped', lease_expires_at = null, + heartbeat_at = ?, started_at = null + where id = 1 and owner_token = ? and owner_epoch = ?${wakeFence}`, + ) + .run( + now, + identity.ownerToken, + identity.ownerEpoch, + ...(expectedWakeGeneration === undefined ? [] : [expectedWakeGeneration]), + ); + return result.changes === 1; + } + + claimNextAgentNode(input: { + supervisor: WorkflowSupervisorIdentity; + claimToken: string; + leaseMs: number; + }): WorkflowNodeClaimResult | undefined { + validateLease(input.leaseMs, "Node claim"); + if (!input.claimToken) throw new WorkflowValidationError("Node claim token must not be empty"); + const nowDate = new Date(); + const now = nowDate.toISOString(); + const expiresAt = new Date(nowDate.getTime() + input.leaseMs).toISOString(); + const claim = this.database.sqlite.transaction(() => { + this.assertActiveSupervisor(input.supervisor, now); + const candidate = this.database.sqlite + .prepare( + `select n.* from workflow_nodes n + join workflow_runs r on r.id = n.workflow_run_id + where n.status = 'ready' and n.claim_token is null + and r.status in ('queued', 'running') and r.cancellation_requested_at is null + order by r.created_at, n.created_at, n.node_key limit 1`, + ) + .get() as WorkflowNodeRow | undefined; + if (!candidate) return undefined; + const updated = this.database.sqlite + .prepare( + `update workflow_nodes + set status = 'running', claim_token = ?, claimed_at = ?, claim_expires_at = ?, + attempt = attempt + 1, supervisor_owner_token = ?, supervisor_owner_epoch = ?, + heartbeat_at = ?, updated_at = ? + where id = ? and status = 'ready' and claim_token is null`, + ) + .run( + input.claimToken, + now, + expiresAt, + input.supervisor.ownerToken, + input.supervisor.ownerEpoch, + now, + now, + candidate.id, + ); + if (updated.changes !== 1) return undefined; + const node = this.database.sqlite + .prepare("select * from workflow_nodes where id = ?") + .get(candidate.id) as WorkflowNodeRow; + const definition = parseJson(node.definition_json); + const provider = typeof definition.config?.provider === "string" ? definition.config.provider : "unknown"; + this.database.sqlite + .prepare( + `insert into workflow_node_attempts ( + node_id, workflow_run_id, node_key, attempt, claim_token, + supervisor_owner_token, supervisor_owner_epoch, provider, phase, + heartbeat_at, created_at, updated_at + ) values (?, ?, ?, ?, ?, ?, ?, ?, 'claimed', ?, ?, ?)`, + ) + .run( + node.id, + node.workflow_run_id, + node.node_key, + node.attempt, + input.claimToken, + input.supervisor.ownerToken, + input.supervisor.ownerEpoch, + provider, + now, + now, + now, + ); + const workflow = this.getWorkflowRow(node.workflow_run_id)!; + if (workflow.status === "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, node.workflow_run_id); + this.insertEvent(node.workflow_run_id, "workflow.running", undefined, { status: "running" }, now); + } + this.insertEvent( + node.workflow_run_id, + "node.running", + node.id, + { nodeKey: node.node_key, status: "running", attempt: node.attempt }, + now, + ); + return { + workflow: this.hydrateWorkflow(this.getWorkflowRow(node.workflow_run_id)!), + node: rowToWorkflowNode(node), + attempt: rowToAttempt(this.getAttemptRow(node.id, node.attempt)!), + }; + }); + return claim.immediate(); + } + + heartbeatNode(input: WorkflowAttemptIdentity & { leaseMs: number }): WorkflowNodeRecord | undefined { + validateLease(input.leaseMs, "Node claim"); + const nowDate = new Date(); + const now = nowDate.toISOString(); + const expiresAt = new Date(nowDate.getTime() + input.leaseMs).toISOString(); + const update = this.database.sqlite.transaction(() => { + const result = this.database.sqlite + .prepare( + `update workflow_nodes set heartbeat_at = ?, claim_expires_at = ?, updated_at = ? + where workflow_run_id = ? and node_key = ? and status = 'running' + and attempt = ? and claim_token = ? and claim_expires_at > ?`, + ) + .run(now, expiresAt, now, input.workflowId, input.nodeKey, input.attempt, input.claimToken, now); + if (result.changes !== 1) return undefined; + this.database.sqlite + .prepare( + `update workflow_node_attempts set heartbeat_at = ?, updated_at = ? + where workflow_run_id = ? and node_key = ? and attempt = ? and claim_token = ? + and phase != 'terminal'`, + ) + .run(now, now, input.workflowId, input.nodeKey, input.attempt, input.claimToken); + return this.getNodeRow(input.workflowId, input.nodeKey); + }); + const row = update.immediate(); + return row ? rowToWorkflowNode(row) : undefined; + } + + markNodeDispatching(identity: WorkflowAttemptIdentity): WorkflowNodeAttemptRecord | undefined { + return this.updateAttemptPhase(identity, "dispatching"); + } + + markNodeRunning(identity: WorkflowAttemptIdentity): WorkflowNodeAttemptRecord | undefined { + return this.updateAttemptPhase(identity, "running"); + } + + markNodeCancelling(identity: WorkflowAttemptIdentity): WorkflowNodeAttemptRecord | undefined { + const now = new Date().toISOString(); + const result = this.database.sqlite + .prepare( + `update workflow_node_attempts set phase = 'cancelling', cancellation_requested_at = ?, updated_at = ? + where workflow_run_id = ? and node_key = ? and attempt = ? and claim_token = ? + and phase != 'terminal'`, + ) + .run(now, now, identity.workflowId, identity.nodeKey, identity.attempt, identity.claimToken); + return result.changes === 1 ? this.getAttempt(identity) : undefined; + } + + recordNodeProviderSession( + identity: WorkflowAttemptIdentity, + providerSessionId: string, + ): WorkflowNodeAttemptRecord | undefined { + const now = new Date().toISOString(); + const result = this.database.sqlite + .prepare( + `update workflow_node_attempts set provider_session_id = ?, updated_at = ? + where workflow_run_id = ? and node_key = ? and attempt = ? and claim_token = ? + and phase != 'terminal'`, + ) + .run( + providerSessionId, + now, + identity.workflowId, + identity.nodeKey, + identity.attempt, + identity.claimToken, + ); + return result.changes === 1 ? this.getAttempt(identity) : undefined; + } + + appendNodeExecutionEvent(input: { + identity: WorkflowAttemptIdentity; + sourceSequence: number; + type: string; + payload: JsonObject; + }): { event: WorkflowEvent; created: boolean } { + const now = new Date().toISOString(); + const append = this.database.sqlite.transaction(() => { + const attempt = this.getAttempt(input.identity); + if (!attempt || attempt.phase === "terminal") return undefined; + const existing = this.database.sqlite + .prepare( + `select workflow_sequence from workflow_provider_events + where node_id = ? and attempt = ? and source_sequence = ?`, + ) + .get(attempt.nodeId, attempt.attempt, input.sourceSequence) as + | { workflow_sequence: number } + | undefined; + if (existing) return { sequence: existing.workflow_sequence, created: false }; + const sequence = this.insertEvent( + input.identity.workflowId, + input.type, + attempt.nodeId, + input.payload, + now, + ); + this.database.sqlite + .prepare( + `insert into workflow_provider_events ( + node_id, attempt, source_sequence, workflow_sequence, event_type, payload_json, created_at + ) values (?, ?, ?, ?, ?, ?, ?)`, + ) + .run( + attempt.nodeId, + attempt.attempt, + input.sourceSequence, + sequence, + input.type, + serializeEventPayload(input.payload), + now, + ); + return { sequence, created: true }; + }); + const saved = append.immediate(); + if (!saved) throw new WorkflowValidationError("Workflow attempt is no longer active"); + return { + event: this.readEvents(input.identity.workflowId, { after: saved.sequence - 1, limit: 1 }).events[0]!, + created: saved.created, + }; + } + + completeAgentNode(input: WorkflowAttemptIdentity & { + status: "succeeded" | "failed" | "cancelled"; + result?: JsonValue; + error?: JsonObject; + }): WorkflowRunRecord | undefined { + const now = new Date().toISOString(); + const complete = this.database.sqlite.transaction(() => { + const node = this.getNodeRow(input.workflowId, input.nodeKey); + if ( + !node || + node.status !== "running" || + node.attempt !== input.attempt || + node.claim_token !== input.claimToken || + !node.claim_expires_at || + node.claim_expires_at <= now || + !node.supervisor_owner_token || + node.supervisor_owner_epoch === null + ) return false; + const activeSupervisor = this.database.sqlite + .prepare( + `select 1 from workflow_supervisor + where id = 1 and owner_token = ? and owner_epoch = ? and lease_expires_at > ?`, + ) + .get(node.supervisor_owner_token, node.supervisor_owner_epoch, now); + if (!activeSupervisor) return false; + + const workflow = this.getWorkflowRow(input.workflowId); + if (!workflow || (workflow.status !== "running" && workflow.status !== "cancelling")) return false; + const status = workflow.status === "cancelling" ? "cancelled" : input.status; + const resultJson = serializeOptionalJson(input.result); + const errorJson = serializeOptionalObject(input.error); + const nodeUpdate = this.database.sqlite + .prepare( + `update workflow_nodes + set status = ?, result_json = ?, error_json = ?, claim_expires_at = null, + heartbeat_at = ?, updated_at = ?, completed_at = ? + where id = ? and status = 'running' and attempt = ? and claim_token = ?`, + ) + .run(status, resultJson, errorJson, now, now, now, node.id, input.attempt, input.claimToken); + if (nodeUpdate.changes !== 1) return false; + const attemptUpdate = this.database.sqlite + .prepare( + `update workflow_node_attempts + set phase = 'terminal', terminal_status = ?, result_json = ?, error_json = ?, + heartbeat_at = ?, updated_at = ?, completed_at = ? + where node_id = ? and attempt = ? and claim_token = ? and phase != 'terminal'`, + ) + .run(status, resultJson, errorJson, now, now, now, node.id, input.attempt, input.claimToken); + if (attemptUpdate.changes !== 1) { + throw new WorkflowValidationError(`Workflow attempt changed during completion: ${node.node_key}`); + } + this.insertEvent( + input.workflowId, + `node.${status}`, + node.id, + { nodeKey: node.node_key, status, attempt: input.attempt }, + now, + ); + + let workflowStatus: "succeeded" | "failed" | "cancelled" | undefined; + if (status === "succeeded") { + this.promoteReadyNodes(input.workflowId, node.id, now); + const remaining = this.database.sqlite + .prepare( + `select count(*) from workflow_nodes + where workflow_run_id = ? and status not in ('succeeded', 'skipped')`, + ) + .pluck() + .get(input.workflowId) as number; + if (remaining === 0) workflowStatus = "succeeded"; + } else { + workflowStatus = status; + this.terminalizeOpenNodes(input.workflowId, status, now); + } + if (!workflowStatus) return true; + + const workflowUpdate = this.database.sqlite + .prepare( + `update workflow_runs set status = ?, result_json = ?, error_json = ?, updated_at = ?, completed_at = ? + where id = ? and status in ('running', 'cancelling')`, + ) + .run(workflowStatus, resultJson, errorJson, now, now, input.workflowId); + if (workflowUpdate.changes !== 1) { + throw new WorkflowValidationError(`Workflow changed during node completion: ${input.workflowId}`); + } + this.insertEvent( + input.workflowId, + `workflow.${workflowStatus}`, + undefined, + { status: workflowStatus }, + now, + ); + return true; + }); + return complete.immediate() ? this.require(input.workflowId) : undefined; + } + + reconcileExpiredClaims(): number { + const now = new Date().toISOString(); + const reconcile = this.database.sqlite.transaction(() => { + const expired = this.database.sqlite + .prepare( + `select * from workflow_nodes + where status = 'running' and claim_expires_at is not null and claim_expires_at <= ? + order by created_at`, + ) + .all(now) as WorkflowNodeRow[]; + let reconciled = 0; + for (const node of expired) { + const error = { code: "worker_lost", message: "Workflow worker lease expired before completion." }; + const nodeUpdate = this.database.sqlite + .prepare( + `update workflow_nodes set status = 'failed', error_json = ?, claim_expires_at = null, + updated_at = ?, completed_at = ? where id = ? and status = 'running'`, + ) + .run(canonicalJson(error), now, now, node.id); + if (nodeUpdate.changes !== 1) continue; + reconciled += 1; + this.database.sqlite + .prepare( + `update workflow_node_attempts + set phase = 'terminal', terminal_status = 'failed', error_json = ?, updated_at = ?, completed_at = ? + where node_id = ? and attempt = ? and phase != 'terminal'`, + ) + .run(canonicalJson(error), now, now, node.id, node.attempt); + this.insertEvent( + node.workflow_run_id, + "node.failed", + node.id, + { nodeKey: node.node_key, status: "failed", code: "worker_lost" }, + now, + ); + this.terminalizeOpenNodes(node.workflow_run_id, "failed", now); + const changed = this.database.sqlite + .prepare( + `update workflow_runs set status = 'failed', error_json = ?, updated_at = ?, completed_at = ? + where id = ? and status in ('running', 'cancelling')`, + ) + .run(canonicalJson(error), now, now, node.workflow_run_id); + if (changed.changes === 1) { + this.insertEvent( + node.workflow_run_id, + "workflow.failed", + undefined, + { status: "failed", code: "worker_lost" }, + now, + ); + } + } + return reconciled; + }); + return reconcile.immediate(); + } + + convergeCancellations(): number { + const now = new Date().toISOString(); + const converge = this.database.sqlite.transaction(() => { + const runs = this.database.sqlite + .prepare("select id from workflow_runs where status = 'cancelling'") + .all() as Array<{ id: string }>; + let completed = 0; + for (const run of runs) { + const pending = this.database.sqlite + .prepare( + `select * from workflow_nodes where workflow_run_id = ? and status in ('pending', 'ready')`, + ) + .all(run.id) as WorkflowNodeRow[]; + for (const node of pending) { + this.database.sqlite + .prepare( + `update workflow_nodes set status = 'cancelled', updated_at = ?, completed_at = ? + where id = ? and status in ('pending', 'ready')`, + ) + .run(now, now, node.id); + this.insertEvent(run.id, "node.cancelled", node.id, { nodeKey: node.node_key, status: "cancelled" }, now); + } + const open = this.database.sqlite + .prepare( + `select count(*) from workflow_nodes + where workflow_run_id = ? and status in ('pending', 'ready', 'running')`, + ) + .pluck() + .get(run.id) as number; + if (open !== 0) continue; + this.database.sqlite + .prepare( + `update workflow_runs set status = 'cancelled', updated_at = ?, completed_at = ? + where id = ? and status = 'cancelling'`, + ) + .run(now, now, run.id); + this.insertEvent(run.id, "workflow.cancelled", undefined, { status: "cancelled" }, now); + completed += 1; + } + return completed; + }); + return converge.immediate(); + } + + isCancellationRequested(workflowId: string): boolean { + const row = this.database.sqlite + .prepare("select status from workflow_runs where id = ?") + .get(workflowId) as { status: string } | undefined; + return row?.status === "cancelling"; + } + close(): void { this.database.close(); } + private assertActiveSupervisor(identity: WorkflowSupervisorIdentity, now: string): void { + const row = this.database.sqlite + .prepare( + `select 1 from workflow_supervisor + where id = 1 and owner_token = ? and owner_epoch = ? and lease_expires_at > ?`, + ) + .get(identity.ownerToken, identity.ownerEpoch, now); + if (!row) throw new WorkflowValidationError("Workflow supervisor lease is not active"); + } + + private getAttemptRow(nodeId: string, attempt: number): WorkflowAttemptRow | undefined { + return this.database.sqlite + .prepare("select * from workflow_node_attempts where node_id = ? and attempt = ?") + .get(nodeId, attempt) as WorkflowAttemptRow | undefined; + } + + private getAttempt(identity: WorkflowAttemptIdentity): WorkflowNodeAttemptRecord | undefined { + const row = this.database.sqlite + .prepare( + `select * from workflow_node_attempts + where workflow_run_id = ? and node_key = ? and attempt = ? and claim_token = ?`, + ) + .get(identity.workflowId, identity.nodeKey, identity.attempt, identity.claimToken) as + | WorkflowAttemptRow + | undefined; + return row ? rowToAttempt(row) : undefined; + } + + private updateAttemptPhase( + identity: WorkflowAttemptIdentity, + phase: "dispatching" | "running", + ): WorkflowNodeAttemptRecord | undefined { + const now = new Date().toISOString(); + const result = this.database.sqlite + .prepare( + `update workflow_node_attempts set phase = ?, updated_at = ? + where workflow_run_id = ? and node_key = ? and attempt = ? and claim_token = ? + and phase != 'terminal'`, + ) + .run( + phase, + now, + identity.workflowId, + identity.nodeKey, + identity.attempt, + identity.claimToken, + ); + return result.changes === 1 ? this.getAttempt(identity) : undefined; + } + 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") @@ -534,6 +1143,8 @@ export class WorkflowStore { policy: parseJson(row.policy_json), idempotencyKey: row.idempotency_key ?? undefined, requestHash: row.request_hash, + workspaceId: row.workspace_id ?? undefined, + workspaceRoot: row.workspace_root ?? undefined, result: parseOptionalJson(row.result_json), error: parseOptionalJson(row.error_json), cancellationRequestedAt: row.cancellation_requested_at ?? undefined, @@ -576,6 +1187,36 @@ export class WorkflowStore { if (!row) throw new WorkflowValidationError(`Node ${nodeId} does not belong to ${workflowId}`); } + private promoteReadyNodes(workflowId: string, completedNodeId: string, now: string): void { + const rows = this.database.sqlite + .prepare( + `select target.* from workflow_edges edge + join workflow_nodes target + on target.workflow_run_id = edge.workflow_run_id and target.id = edge.to_node_id + where edge.workflow_run_id = ? and edge.from_node_id = ? and target.status = 'pending' + and not exists ( + select 1 from workflow_edges incoming + join workflow_nodes source + on source.workflow_run_id = incoming.workflow_run_id and source.id = incoming.from_node_id + where incoming.workflow_run_id = edge.workflow_run_id + and incoming.to_node_id = edge.to_node_id + and source.status not in ('succeeded', 'skipped') + ) + order by target.created_at, target.node_key`, + ) + .all(workflowId, completedNodeId) as WorkflowNodeRow[]; + const update = this.database.sqlite.prepare( + `update workflow_nodes set status = 'ready', updated_at = ? where id = ? and status = 'pending'`, + ); + for (const row of rows) { + const changed = update.run(now, row.id); + if (changed.changes !== 1) { + throw new WorkflowValidationError(`Workflow node changed during dependency promotion: ${row.node_key}`); + } + this.insertEvent(workflowId, "node.ready", row.id, { nodeKey: row.node_key, status: "ready" }, now); + } + } + private terminalizeOpenNodes( workflowId: string, workflowStatus: "failed" | "cancelled", @@ -600,6 +1241,15 @@ export class WorkflowStore { if (changed.changes !== 1) { throw new WorkflowValidationError(`Workflow node changed during terminal transition: ${row.node_key}`); } + if (row.status === "running") { + this.database.sqlite + .prepare( + `update workflow_node_attempts + set phase = 'terminal', terminal_status = ?, updated_at = ?, completed_at = ? + where node_id = ? and attempt = ? and phase != 'terminal'`, + ) + .run(nodeStatus, now, now, row.id, row.attempt); + } this.insertEvent( workflowId, `node.${nodeStatus}`, @@ -645,6 +1295,7 @@ function normalizeSubmission(request: SubmitWorkflowRequest): { inputJson: string; policyJson: string; idempotencyKey?: string; + workspace?: WorkflowWorkspaceScope; requestHash: string; } { const definition = normalizeDefinition(request.definition); @@ -657,10 +1308,16 @@ function normalizeSubmission(request: SubmitWorkflowRequest): { if (request.idempotencyKey !== undefined && !idempotencyKey) { throw new WorkflowValidationError("Idempotency key must not be empty"); } + const workspace = request.workspace + ? { + workspaceId: requireNonEmptyString(request.workspace.workspaceId, "Workspace id"), + workspaceRoot: requireNonEmptyString(request.workspace.workspaceRoot, "Workspace root"), + } + : undefined; const requestHash = createHash("sha256") - .update(canonicalJson({ definition, input, policy })) + .update(canonicalJson({ definition, input, policy, workspace: workspace ?? null })) .digest("hex"); - return { definition, definitionJson, inputJson, policyJson, idempotencyKey, requestHash }; + return { definition, definitionJson, inputJson, policyJson, idempotencyKey, workspace, requestHash }; } function normalizeDefinition(definition: WorkflowDefinition): WorkflowDefinition { @@ -820,6 +1477,42 @@ function rowToWorkflowNode(row: WorkflowNodeRow): WorkflowNodeRecord { }; } +function rowToAttempt(row: WorkflowAttemptRow): WorkflowNodeAttemptRecord { + return { + nodeId: row.node_id, + workflowId: row.workflow_run_id, + nodeKey: row.node_key, + attempt: row.attempt, + claimToken: row.claim_token, + supervisorOwnerToken: row.supervisor_owner_token, + supervisorOwnerEpoch: row.supervisor_owner_epoch, + provider: row.provider, + phase: row.phase, + providerSessionId: row.provider_session_id ?? undefined, + heartbeatAt: row.heartbeat_at ?? undefined, + cancellationRequestedAt: row.cancellation_requested_at ?? undefined, + terminalStatus: row.terminal_status ?? 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 rowToSupervisor(row: WorkflowSupervisorRow): WorkflowSupervisorRecord { + return { + ownerToken: row.owner_token!, + ownerEpoch: row.owner_epoch, + ownerPid: row.owner_pid ?? undefined, + status: row.status, + leaseExpiresAt: row.lease_expires_at ?? undefined, + heartbeatAt: row.heartbeat_at ?? undefined, + wakeGeneration: row.wake_generation, + startedAt: row.started_at ?? undefined, + }; +} + function rowToWorkflowEdge(row: WorkflowEdgeRow): WorkflowEdgeRecord { return { workflowId: row.workflow_run_id, @@ -882,6 +1575,20 @@ function readNodeType(type: string): WorkflowNodeDefinitionV1["type"] { throw new WorkflowValidationError(`Unsupported stored workflow node type: ${type}`); } +function validateLease(leaseMs: number, label: string): void { + if (!Number.isSafeInteger(leaseMs) || leaseMs < 1 || leaseMs > MAX_CLAIM_LEASE_MS) { + throw new WorkflowValidationError( + `${label} lease must be between 1 and ${MAX_CLAIM_LEASE_MS} milliseconds`, + ); + } +} + +function requireNonEmptyString(value: string, label: string): string { + const normalized = value.trim(); + if (!normalized) throw new WorkflowValidationError(`${label} must not be empty`); + return normalized; +} + function createId(prefix: "wf_" | "wfn_"): string { return `${prefix}${randomUUID().replaceAll("-", "")}`; } diff --git a/src/workflows/types.ts b/src/workflows/types.ts index fb9659cf..1f0597c9 100644 --- a/src/workflows/types.ts +++ b/src/workflows/types.ts @@ -56,6 +56,7 @@ export interface SubmitWorkflowRequest { input?: JsonObject; policy?: WorkflowPolicy; idempotencyKey?: string; + workspace?: WorkflowWorkspaceScope; } export interface WorkflowNodeRecord { @@ -93,6 +94,8 @@ export interface WorkflowRunRecord { policy: WorkflowPolicy; idempotencyKey?: string; requestHash: string; + workspaceId?: string; + workspaceRoot?: string; result?: JsonValue; error?: JsonObject; cancellationRequestedAt?: string; @@ -128,6 +131,55 @@ export interface WorkflowWaitOptions { pollIntervalMs?: number; } +export interface WorkflowWorkspaceScope { + workspaceId: string; + workspaceRoot: string; +} + +export interface WorkflowSupervisorIdentity { + ownerToken: string; + ownerEpoch: number; +} + +export interface WorkflowSupervisorRecord extends WorkflowSupervisorIdentity { + ownerPid?: number; + status: "starting" | "running" | "stopping" | "stopped"; + leaseExpiresAt?: string; + heartbeatAt?: string; + wakeGeneration: number; + startedAt?: string; +} + +export interface WorkflowAttemptIdentity { + workflowId: string; + nodeKey: string; + attempt: number; + claimToken: string; +} + +export interface WorkflowNodeAttemptRecord extends WorkflowAttemptIdentity { + nodeId: string; + supervisorOwnerToken: string; + supervisorOwnerEpoch: number; + provider: string; + phase: "claimed" | "dispatching" | "running" | "cancelling" | "terminal"; + providerSessionId?: string; + heartbeatAt?: string; + cancellationRequestedAt?: string; + terminalStatus?: "succeeded" | "failed" | "cancelled"; + result?: JsonValue; + error?: JsonObject; + createdAt: string; + updatedAt: string; + completedAt?: string; +} + +export interface WorkflowNodeClaimResult { + workflow: WorkflowRunRecord; + node: WorkflowNodeRecord; + attempt: WorkflowNodeAttemptRecord; +} + export interface WorkflowEventReadOptions { after?: number; limit?: number; From c5837dbaae49490fa732e20b5246b1e31e152070 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 01:31:21 +0530 Subject: [PATCH 2/4] feat(workflows): add durable workflow supervisor Co-Authored-By: Claude --- src/workflows/supervisor-launch.ts | 102 ++++++++ src/workflows/supervisor.ts | 387 +++++++++++++++++++++++++++++ 2 files changed, 489 insertions(+) create mode 100644 src/workflows/supervisor-launch.ts create mode 100644 src/workflows/supervisor.ts diff --git a/src/workflows/supervisor-launch.ts b/src/workflows/supervisor-launch.ts new file mode 100644 index 00000000..a1a24a45 --- /dev/null +++ b/src/workflows/supervisor-launch.ts @@ -0,0 +1,102 @@ +import { spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { filterWorkflowEnvironment } from "./policy.js"; +import { WorkflowStore } from "./store.js"; + +export interface EnsureSupervisorResult { + requestedWakeGeneration: number; + spawned: boolean; + ownerEpoch?: number; +} + +export async function ensureSupervisor(input: { + stateDir: string; + cliEntrypoint?: string; + env?: NodeJS.ProcessEnv; + startupTimeoutMs?: number; +}): Promise { + const store = new WorkflowStore(input.stateDir); + try { + const requestedWakeGeneration = store.requestSupervisorWake(); + const current = store.getSupervisor(); + if (current?.leaseExpiresAt && current.leaseExpiresAt > new Date().toISOString()) { + if (current.ownerPid !== undefined && !isProcessRunning(current.ownerPid)) { + store.releaseSupervisor(current); + } else { + return { + requestedWakeGeneration, + spawned: false, + ownerEpoch: current.ownerEpoch, + }; + } + } + + const cliEntrypoint = input.cliEntrypoint ?? fileURLToPath(new URL("../cli.js", import.meta.url)); + const sourceEnvironment = input.env ?? process.env; + const environment: NodeJS.ProcessEnv = { + ...filterWorkflowEnvironment(sourceEnvironment), + DEVSPACE_STATE_DIR: input.stateDir, + }; + for (const key of [ + "DEVSPACE_WORKFLOW_FAKE_PROVIDER", + "DEVSPACE_WORKFLOW_FAKE_BEHAVIOR", + "DEVSPACE_WORKFLOW_FAKE_DELAY_MS", + ]) { + if (sourceEnvironment[key]) environment[key] = sourceEnvironment[key]; + } + + const child = spawn( + process.execPath, + [ + ...process.execArgv, + cliEntrypoint, + "workflows", + "__supervisor", + "--state-dir", + input.stateDir, + ], + { + detached: true, + stdio: "ignore", + env: environment, + shell: false, + windowsHide: true, + }, + ); + await new Promise((resolve, reject) => { + child.once("spawn", resolve); + child.once("error", reject); + }); + child.unref(); + + const deadline = Date.now() + (input.startupTimeoutMs ?? 2_000); + while (Date.now() < deadline) { + const supervisor = store.getSupervisor(); + if (supervisor?.leaseExpiresAt && supervisor.leaseExpiresAt > new Date().toISOString()) { + return { + requestedWakeGeneration, + spawned: true, + ownerEpoch: supervisor.ownerEpoch, + }; + } + await delay(20); + } + throw new Error("Workflow supervisor did not acquire its durable lease."); + } finally { + store.close(); + } +} + +function isProcessRunning(pid: number): boolean { + if (!Number.isSafeInteger(pid) || pid < 1) return false; + try { + process.kill(pid, 0); + return true; + } catch (error) { + return (error as NodeJS.ErrnoException).code === "EPERM"; + } +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} diff --git a/src/workflows/supervisor.ts b/src/workflows/supervisor.ts new file mode 100644 index 00000000..e34d412d --- /dev/null +++ b/src/workflows/supervisor.ts @@ -0,0 +1,387 @@ +import { randomUUID } from "node:crypto"; +import { createLocalAgentAdapter } from "../local-agent-adapters.js"; +import { + LocalAgentRunController, + type LocalAgentEvent, + type LocalAgentRunHandle, + type LocalAgentRunInput, + type LocalAgentRunResult, +} from "../local-agent-runtime.js"; +import { isLocalAgentProvider } from "../local-agent-profiles.js"; +import type { EffectiveLocalAgentPolicy } from "./policy.js"; +import { WorkflowStore, WorkflowValidationError } from "./store.js"; +import type { + JsonObject, + JsonValue, + WorkflowAttemptIdentity, + WorkflowNodeClaimResult, + WorkflowSupervisorIdentity, +} from "./types.js"; + +const DEFAULT_SUPERVISOR_LEASE_MS = 5_000; +const DEFAULT_NODE_LEASE_MS = 5_000; +const DEFAULT_HEARTBEAT_MS = 1_000; +const DEFAULT_IDLE_MS = 1_000; +const MAX_ERROR_CHARS = 16_384; +const MAX_TIMEOUT_MS = 24 * 60 * 60_000; + +export type WorkflowHandleFactory = ( + provider: string, + input: LocalAgentRunInput, +) => Promise; + +export interface WorkflowSupervisorOptions { + supervisorLeaseMs?: number; + nodeLeaseMs?: number; + heartbeatMs?: number; + idleMs?: number; + ownerToken?: string; + ownerPid?: number; + handleFactory?: WorkflowHandleFactory; +} + +export async function runWorkflowSupervisor( + stateDir: string, + options: WorkflowSupervisorOptions = {}, +): Promise { + const store = new WorkflowStore(stateDir); + const supervisorLeaseMs = options.supervisorLeaseMs ?? DEFAULT_SUPERVISOR_LEASE_MS; + const nodeLeaseMs = options.nodeLeaseMs ?? DEFAULT_NODE_LEASE_MS; + const heartbeatMs = options.heartbeatMs ?? DEFAULT_HEARTBEAT_MS; + const idleMs = options.idleMs ?? DEFAULT_IDLE_MS; + validateSupervisorTiming(supervisorLeaseMs, nodeLeaseMs, heartbeatMs, idleMs); + const acquired = store.acquireSupervisor({ + ownerToken: options.ownerToken ?? randomUUID(), + ownerPid: options.ownerPid ?? process.pid, + leaseMs: supervisorLeaseMs, + }); + if (!acquired) { + store.close(); + return false; + } + + const identity: WorkflowSupervisorIdentity = acquired; + let lastWorkAt = Date.now(); + try { + store.reconcileExpiredClaims(); + store.convergeCancellations(); + while (store.heartbeatSupervisor(identity, supervisorLeaseMs)) { + store.reconcileExpiredClaims(); + store.convergeCancellations(); + const claim = store.claimNextAgentNode({ + supervisor: identity, + claimToken: randomUUID(), + leaseMs: nodeLeaseMs, + }); + if (claim) { + lastWorkAt = Date.now(); + await executeClaim(store, identity, claim, { + supervisorLeaseMs, + nodeLeaseMs, + heartbeatMs, + handleFactory: options.handleFactory ?? defaultHandleFactory, + }); + lastWorkAt = Date.now(); + continue; + } + + const supervisor = store.getSupervisor(); + if (!supervisor || supervisor.ownerToken !== identity.ownerToken || supervisor.ownerEpoch !== identity.ownerEpoch) { + return false; + } + if (Date.now() - lastWorkAt >= idleMs) { + if (store.releaseSupervisor(identity, supervisor.wakeGeneration)) return true; + lastWorkAt = Date.now(); + continue; + } + await delay(Math.min(heartbeatMs, Math.max(10, idleMs - (Date.now() - lastWorkAt)))); + } + return false; + } finally { + store.releaseSupervisor(identity); + store.close(); + } +} + +async function executeClaim( + store: WorkflowStore, + supervisor: WorkflowSupervisorIdentity, + claim: WorkflowNodeClaimResult, + options: { + supervisorLeaseMs: number; + nodeLeaseMs: number; + heartbeatMs: number; + handleFactory: WorkflowHandleFactory; + }, +): Promise { + const identity: WorkflowAttemptIdentity = { + workflowId: claim.workflow.id, + nodeKey: claim.node.key, + attempt: claim.node.attempt, + claimToken: claim.node.claimToken!, + }; + const config = claim.node.definition.config ?? {}; + let handle: LocalAgentRunHandle | undefined; + let heartbeat: NodeJS.Timeout | undefined; + let timeout: NodeJS.Timeout | undefined; + let cancellationObserved = false; + let timeoutObserved = false; + let leaseFailure: Error | undefined; + let tickRunning = false; + + try { + const execution = readExecutionSnapshot(config); + const fullPrompt = execution.profileBody + ? `${execution.profileBody}\n\nTask:\n${execution.prompt}` + : execution.prompt; + if (!store.markNodeDispatching(identity)) { + throw new Error("Workflow node attempt was lost before dispatch."); + } + + const tick = async () => { + if (tickRunning || leaseFailure) return; + tickRunning = true; + try { + if (!store.heartbeatSupervisor(supervisor, options.supervisorLeaseMs)) { + leaseFailure = new Error("Workflow supervisor lease lost."); + await handle?.cancel(leaseFailure); + return; + } + if (!store.heartbeatNode({ ...identity, leaseMs: options.nodeLeaseMs })) { + leaseFailure = new Error("Workflow node claim lost."); + await handle?.cancel(leaseFailure); + return; + } + if (!cancellationObserved && store.isCancellationRequested(identity.workflowId)) { + cancellationObserved = true; + store.markNodeCancelling(identity); + await handle?.cancel(new Error("Workflow cancellation requested.")); + } + } finally { + tickRunning = false; + } + }; + await tick(); + if (leaseFailure) throw leaseFailure; + heartbeat = setInterval(() => void tick().catch(() => undefined), options.heartbeatMs); + + handle = await options.handleFactory(execution.provider, { + prompt: fullPrompt, + workspace: execution.workspaceRoot, + model: execution.model, + thinking: execution.thinking, + policy: execution.effectivePolicy, + }); + if (leaseFailure) { + await handle.cancel(leaseFailure); + throw leaseFailure; + } + if (cancellationObserved) { + await handle.cancel(new Error("Workflow cancellation requested.")); + } + if (!store.markNodeRunning(identity)) { + throw new Error("Workflow node attempt was lost during dispatch."); + } + + if (execution.timeoutMs !== undefined) { + timeout = setTimeout(() => { + timeoutObserved = true; + void handle?.cancel(new Error("Workflow node timed out.")); + }, execution.timeoutMs); + } + + const drain = drainEvents(store, identity, handle); + let result: LocalAgentRunResult; + try { + result = await handle.result(); + } finally { + await drain; + } + if (result.providerSessionId) { + store.recordNodeProviderSession(identity, result.providerSessionId); + } + const cancelled = cancellationObserved || store.isCancellationRequested(identity.workflowId); + store.completeAgentNode({ + ...identity, + status: cancelled ? "cancelled" : "succeeded", + result: { + provider: result.provider, + providerSessionId: result.providerSessionId, + finalResponse: result.finalResponse, + }, + }); + } catch (error) { + const cancelled = !timeoutObserved && ( + cancellationObserved || store.isCancellationRequested(identity.workflowId) || isAbortError(error) + ); + store.completeAgentNode({ + ...identity, + status: cancelled ? "cancelled" : "failed", + error: normalizedError(timeoutObserved ? "timed_out" : cancelled ? "cancelled" : "provider_failed", error), + }); + } finally { + if (heartbeat) clearInterval(heartbeat); + if (timeout) clearTimeout(timeout); + await handle?.dispose().catch(() => undefined); + store.convergeCancellations(); + } +} + +async function drainEvents( + store: WorkflowStore, + identity: WorkflowAttemptIdentity, + handle: LocalAgentRunHandle, +): Promise { + for await (const event of handle.events()) { + if (event.type === "session") { + store.recordNodeProviderSession(identity, event.providerSessionId); + } + try { + store.appendNodeExecutionEvent({ + identity, + sourceSequence: event.sequence, + type: `provider.${event.type}`, + payload: eventPayload(event), + }); + } catch (error) { + if (!(error instanceof WorkflowValidationError)) throw error; + break; + } + } +} + +function eventPayload(event: LocalAgentEvent): JsonObject { + return JSON.parse(JSON.stringify(event)) as JsonObject; +} + +function readExecutionSnapshot(config: JsonObject): { + provider: string; + model?: string; + thinking?: string; + profileBody: string; + prompt: string; + workspaceRoot: string; + timeoutMs?: number; + effectivePolicy: EffectiveLocalAgentPolicy; +} { + const provider = requiredString(config.provider, "provider"); + if (!isLocalAgentProvider(provider) && provider !== "fake") { + throw new Error(`Unsupported workflow provider: ${provider}`); + } + const timeoutMs = optionalInteger(config.timeoutMs, "timeoutMs"); + if (timeoutMs !== undefined && (timeoutMs < 1 || timeoutMs > MAX_TIMEOUT_MS)) { + throw new Error(`Workflow timeoutMs must be between 1 and ${MAX_TIMEOUT_MS}.`); + } + const policy = config.effectivePolicy; + if (!policy || typeof policy !== "object" || Array.isArray(policy)) { + throw new Error("Workflow execution snapshot is missing effectivePolicy."); + } + return { + provider, + model: optionalString(config.model), + thinking: optionalString(config.thinking), + profileBody: optionalString(config.profileBody) ?? "", + prompt: requiredString(config.prompt, "prompt"), + workspaceRoot: requiredString(config.workspaceRoot, "workspaceRoot"), + timeoutMs, + effectivePolicy: policy as unknown as EffectiveLocalAgentPolicy, + }; +} + +async function defaultHandleFactory( + provider: string, + input: LocalAgentRunInput, +): Promise { + if (provider === "fake" || process.env.DEVSPACE_WORKFLOW_FAKE_PROVIDER === "1") { + return createFakeHandle(input); + } + if (!isLocalAgentProvider(provider)) throw new Error(`Unsupported workflow provider: ${provider}`); + return createLocalAgentAdapter(provider).start(input); +} + +function createFakeHandle(input: LocalAgentRunInput): LocalAgentRunHandle { + const controller = new LocalAgentRunController("fake", input.signal); + const behavior = process.env.DEVSPACE_WORKFLOW_FAKE_BEHAVIOR ?? "success"; + const delayMs = Number(process.env.DEVSPACE_WORKFLOW_FAKE_DELAY_MS ?? "25"); + let timer: NodeJS.Timeout | undefined; + let finishPump!: () => void; + const pump = new Promise((resolve) => { + finishPump = resolve; + timer = setTimeout(() => { + controller.emit({ type: "session", providerSessionId: "fake-session", resumed: false }); + controller.emit({ type: "output", stream: "assistant", delta: "fake result" }); + if (behavior === "fail") controller.fail(new Error("Deterministic fake provider failure.")); + else controller.succeed({ + provider: "fake", + providerSessionId: "fake-session", + finalResponse: "fake result", + items: [], + }); + finishPump(); + }, Number.isFinite(delayMs) && delayMs >= 0 ? delayMs : 25); + }); + controller.setLifecycle({ + cancel: () => { + if (timer) clearTimeout(timer); + finishPump(); + }, + dispose: () => { + if (timer) clearTimeout(timer); + finishPump(); + }, + pump, + }); + return controller; +} + +function normalizedError(code: string, error: unknown): JsonObject { + const message = error instanceof Error ? error.message : String(error); + return { code, message: message.slice(0, MAX_ERROR_CHARS) }; +} + +function requiredString(value: JsonValue | undefined, name: string): string { + const text = optionalString(value); + if (!text) throw new Error(`Workflow execution snapshot is missing ${name}.`); + return text; +} + +function optionalString(value: JsonValue | undefined): string | undefined { + return typeof value === "string" && value.trim() ? value : undefined; +} + +function optionalInteger(value: JsonValue | undefined, name: string): number | undefined { + if (value === undefined || value === null) return undefined; + if (!Number.isSafeInteger(value)) throw new Error(`Workflow ${name} must be an integer.`); + return value as number; +} + +function isAbortError(error: unknown): boolean { + return error instanceof Error && error.name === "AbortError"; +} + +function validateSupervisorTiming( + supervisorLeaseMs: number, + nodeLeaseMs: number, + heartbeatMs: number, + idleMs: number, +): void { + for (const [label, value] of [ + ["supervisor lease", supervisorLeaseMs], + ["node lease", nodeLeaseMs], + ["heartbeat", heartbeatMs], + ] as const) { + if (!Number.isSafeInteger(value) || value < 1) { + throw new WorkflowValidationError(`Workflow ${label} must be a positive integer.`); + } + } + if (!Number.isSafeInteger(idleMs) || idleMs < 0) { + throw new WorkflowValidationError("Workflow supervisor idle timeout must be a non-negative integer."); + } + if (heartbeatMs >= Math.min(supervisorLeaseMs, nodeLeaseMs)) { + throw new WorkflowValidationError("Workflow heartbeat interval must be shorter than both leases."); + } +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} From 33378dc9722bf467f4b2710c3b95851e4e58647e Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 01:31:29 +0530 Subject: [PATCH 3/4] feat(cli): add durable workflow commands Co-Authored-By: Claude --- docs/workflow-cli.md | 66 ++++++ src/cli.ts | 19 +- src/local-agent-profiles.ts | 2 +- src/workflows/cli.ts | 418 ++++++++++++++++++++++++++++++++++++ 4 files changed, 501 insertions(+), 4 deletions(-) create mode 100644 docs/workflow-cli.md create mode 100644 src/workflows/cli.ts diff --git a/docs/workflow-cli.md b/docs/workflow-cli.md new file mode 100644 index 00000000..755c3399 --- /dev/null +++ b/docs/workflow-cli.md @@ -0,0 +1,66 @@ +# Workflow CLI + +DevSpace workflows are durable, single-agent runs intended for shell parent processes. Submission stores an immutable execution snapshot and wakes a detached supervisor. The workflow continues after the submitting shell exits. + +## Workspace identity + +Every command requires a workspace identity and canonical workspace root. The root must be inside `DEVSPACE_ALLOWED_ROOTS`. + +```sh +export DEVSPACE_WORKSPACE_ID="checkout-42" +export DEVSPACE_WORKSPACE_ROOT="$(git rev-parse --show-toplevel)" +export DEVSPACE_ALLOWED_ROOTS="$HOME/src" +``` + +A workflow ID does not grant access by itself. Status, events, wait, and cancellation must use the same workspace ID and canonical root used at submission. + +## Submit and wait + +`run --json` writes exactly one versioned JSON object to stdout after the request is durable and supervisor wakeup has been requested. + +```sh +accepted="$({ + devspace workflows run reviewer \ + --prompt "Review the current change" \ + --idempotency-key "review-$GITHUB_SHA" \ + --json +})" +workflow_id="$(node -e 'process.stdout.write(JSON.parse(process.argv[1]).workflow.id)' "$accepted")" + +# The submitting process may exit here. Another process can wait later. +result="$(devspace workflows wait "$workflow_id" --timeout-ms 300000 --after 0 --json)" +status="$(node -e 'process.stdout.write(JSON.parse(process.argv[1]).workflow.status)' "$result")" +``` + +`wait` is bounded, repeatable, non-destructive, and exits zero when the timeout expires or when the workflow reaches `failed` or `cancelled`. Inspect `timedOut` and `workflow.status` in the JSON response. + +The maximum wait timeout is 300000 milliseconds. Use the returned `cursor` with `--after` to replay only newer events: + +```sh +devspace workflows events "$workflow_id" --after "$cursor" --json +devspace workflows wait "$workflow_id" --after "$cursor" --timeout-ms 300000 --json +``` + +## Other commands + +```sh +devspace workflows status "$workflow_id" --json +devspace workflows cancel "$workflow_id" --json +``` + +Cancellation is idempotent. Pending work is cancelled without dispatch; active providers receive cooperative cancellation and process cleanup through their existing provider handles. + +## Submission options + +```text +--prompt required; no positional prompt is accepted +--idempotency-key optional durable deduplication key +--model optional provider model override +--thinking optional provider thinking override +--access read_only|workspace_write +--timeout-ms optional node execution timeout +``` + +The persisted snapshot includes the resolved profile body and hash, profile name, provider, model, thinking setting, effective access/environment policy, canonical workspace root, prompt, and timeout. Later profile-file changes do not affect an accepted workflow. + +JSON errors use the same versioned envelope and a nonzero exit code. Diagnostics go to stderr; prompts and provider output are never written to diagnostic logs. diff --git a/src/cli.ts b/src/cli.ts index 7a1ac63f..6c10a563 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -1,6 +1,6 @@ #!/usr/bin/env node import { createRequire } from "node:module"; -import { stdin as input, stdout as output } from "node:process"; +import { stderr as errorOutput, stdin as input, stdout as output } from "node:process"; import { spawn } from "node:child_process"; import { mkdtempSync, writeFileSync } from "node:fs"; import { readFile } from "node:fs/promises"; @@ -39,8 +39,9 @@ import { } from "./user-config.js"; import { expandHomePath } from "./roots.js"; import { shutdownHttpServer } from "./server-shutdown.js"; +import { runWorkflowsCli } from "./workflows/cli.js"; -type Command = "serve" | "init" | "doctor" | "config" | "agents" | "help" | "version"; +type Command = "serve" | "init" | "doctor" | "config" | "agents" | "workflows" | "help" | "version"; const require = createRequire(import.meta.url); const SUPPORTED_NODE_RANGE = ">=20.12 <27"; @@ -67,6 +68,16 @@ async function main(argv: string[]): Promise { case "agents": await runAgentsCommand(args); return; + case "workflows": + process.exitCode = await runWorkflowsCli({ + argv: args, + cwd: process.cwd(), + env: process.env, + stdout: output, + stderr: errorOutput, + cliEntrypoint: fileURLToPath(import.meta.url), + }); + return; case "help": printHelp(); return; @@ -78,7 +89,7 @@ async function main(argv: string[]): Promise { function normalizeCommand(command: string | undefined): Command { if (!command || command === "serve" || command === "start") return "serve"; - if (command === "init" || command === "doctor" || command === "config" || command === "agents") return command; + if (command === "init" || command === "doctor" || command === "config" || command === "agents" || command === "workflows") return command; if (command === "help" || command === "--help" || command === "-h") return "help"; if (command === "version" || command === "--version" || command === "-v") return "version"; throw new Error(`Unknown command: ${command}`); @@ -312,6 +323,8 @@ function printHelp(): void { " devspace agents ls List subagent sessions", " devspace agents run [--model ] ", " devspace agents show ", + " devspace workflows run --prompt --json", + " devspace workflows help", " devspace -v, --version Print the installed version", "", "For temporary tunnels:", diff --git a/src/local-agent-profiles.ts b/src/local-agent-profiles.ts index a7fa0d87..e9d1b242 100644 --- a/src/local-agent-profiles.ts +++ b/src/local-agent-profiles.ts @@ -43,7 +43,7 @@ const FRONTMATTER_DELIMITER = "---"; const PROVIDERS = new Set(LOCAL_AGENT_PROVIDERS); export async function loadLocalAgentProfiles( - config: ServerConfig, + config: Pick, workspaceRoot: string, ): Promise { if (!config.subagents) return []; diff --git a/src/workflows/cli.ts b/src/workflows/cli.ts new file mode 100644 index 00000000..339e85e5 --- /dev/null +++ b/src/workflows/cli.ts @@ -0,0 +1,418 @@ +import { createHash } from "node:crypto"; +import { realpath } from "node:fs/promises"; +import { homedir } from "node:os"; +import { isAbsolute, join, relative, resolve } from "node:path"; +import type { Writable } from "node:stream"; +import { devspaceAgentsDir, loadDevspaceFiles, resolveSubagentsFlag } from "../user-config.js"; +import { + formatAvailableLocalAgentTargets, + resolveLocalAgentTarget, +} from "../local-agent-targets.js"; +import { + loadLocalAgentProfiles, + type LocalAgentProfile, +} from "../local-agent-profiles.js"; +import { expandHomePath } from "../roots.js"; +import { resolveWorkflowNodePolicy } from "./policy.js"; +import { WorkflowOrchestrator } from "./orchestrator.js"; +import { + WorkflowIdempotencyConflictError, + WorkflowNotFoundError, + WorkflowValidationError, +} from "./store.js"; +import { ensureSupervisor } from "./supervisor-launch.js"; +import { runWorkflowSupervisor } from "./supervisor.js"; +import type { + JsonObject, + WorkflowRunRecord, + WorkflowWorkspaceScope, +} from "./types.js"; + +const ENVELOPE_VERSION = 1 as const; +const DEFAULT_WAIT_TIMEOUT_MS = 30_000; +const MAX_WAIT_TIMEOUT_MS = 300_000; + +export interface WorkflowsCliContext { + argv: string[]; + cwd: string; + env: NodeJS.ProcessEnv; + stdout: Writable; + stderr: Writable; + cliEntrypoint: string; +} + +interface WorkflowCliConfig { + stateDir: string; + allowedRoots: string[]; + devspaceAgentsDir: string; + subagents: boolean; +} + +export async function runWorkflowsCli(context: WorkflowsCliContext): Promise { + const [command, ...args] = context.argv; + if (command === "__supervisor") { + validateWorkflowArguments(command, args); + const stateDir = requiredOption(args, "--state-dir"); + await runWorkflowSupervisor(stateDir); + return 0; + } + if (!command || command === "help" || command === "--help" || command === "-h") { + context.stdout.write(workflowsHelp()); + return 0; + } + + const jsonMode = args.includes("--json"); + try { + const result = await dispatchWorkflowCommand(command, args, context); + writeEnvelope(context.stdout, { ok: true, command, ...result }); + return 0; + } catch (error) { + const normalized = normalizeCliError(error); + if (jsonMode) { + writeEnvelope(context.stdout, { ok: false, command, error: normalized }); + } else { + context.stderr.write(`${normalized.message}\n`); + } + return normalized.exitCode; + } +} + +async function dispatchWorkflowCommand( + command: string, + args: string[], + context: WorkflowsCliContext, +): Promise> { + const parsed = validateWorkflowArguments(command, args); + const config = loadWorkflowCliConfig(context.env, context.cwd); + const scope = await resolveWorkspaceScope(context, config.allowedRoots); + const orchestrator = new WorkflowOrchestrator(config.stateDir); + try { + switch (command) { + case "run": + return await runCommand(parsed.positionals[0]!, args, context, config, scope, orchestrator); + case "status": { + const workflowId = parsed.positionals[0]!; + return { workflow: orchestrator.getForWorkspace(workflowId, scope) ?? notFound(workflowId) }; + } + case "events": { + const workflowId = parsed.positionals[0]!; + const after = integerOption(args, "--after") ?? 0; + const page = orchestrator.eventsForWorkspace(workflowId, scope, { after, limit: 1_000 }); + return { workflowId, events: page.events, cursor: page.nextCursor }; + } + case "wait": { + const workflowId = parsed.positionals[0]!; + const timeoutMs = integerOption(args, "--timeout-ms") ?? DEFAULT_WAIT_TIMEOUT_MS; + if (timeoutMs < 0 || timeoutMs > MAX_WAIT_TIMEOUT_MS) { + throw new CliInputError(`--timeout-ms must be between 0 and ${MAX_WAIT_TIMEOUT_MS}.`); + } + const before = orchestrator.getForWorkspace(workflowId, scope) ?? notFound(workflowId); + const workflow = await orchestrator.waitForWorkspace(workflowId, scope, { timeoutMs }); + const after = integerOption(args, "--after"); + const page = after === undefined + ? undefined + : orchestrator.eventsForWorkspace(workflowId, scope, { after, limit: 1_000 }); + return { + workflow, + timedOut: !isTerminal(workflow) && !isTerminal(before), + ...(page ? { events: page.events, cursor: page.nextCursor } : {}), + }; + } + case "cancel": { + const workflowId = parsed.positionals[0]!; + const workflow = orchestrator.cancelForWorkspace(workflowId, scope); + await ensureSupervisor({ + stateDir: config.stateDir, + cliEntrypoint: context.cliEntrypoint, + env: context.env, + }); + return { workflow }; + } + default: + throw new CliInputError(`Unknown workflows command: ${command}`); + } + } finally { + orchestrator.close(); + } +} + +async function runCommand( + targetName: string, + args: string[], + context: WorkflowsCliContext, + config: WorkflowCliConfig, + scope: WorkflowWorkspaceScope, + orchestrator: WorkflowOrchestrator, +): Promise> { + const prompt = requiredOption(args, "--prompt"); + const model = optionalOption(args, "--model"); + const thinking = optionalOption(args, "--thinking"); + const timeoutMs = integerOption(args, "--timeout-ms"); + const access = optionalOption(args, "--access") ?? "read_only"; + const idempotencyKey = optionalOption(args, "--idempotency-key"); + if (timeoutMs !== undefined && (timeoutMs < 1 || timeoutMs > 24 * 60 * 60_000)) { + throw new CliInputError("--timeout-ms must be between 1 and 86400000."); + } + if (access !== "read_only" && access !== "workspace_write") { + throw new CliInputError("--access must be read_only or workspace_write."); + } + + const profiles = await loadLocalAgentProfiles(config, scope.workspaceRoot); + const target = resolveWorkflowTarget(targetName, profiles, model, thinking, context.env); + if (!target) { + throw new CliInputError( + `Unknown workflow profile or provider: ${targetName}. Available ${formatAvailableLocalAgentTargets(profiles)}`, + ); + } + const profile = target.kind === "profile" ? target.profile : undefined; + const profileBody = profile?.body.trim() ?? ""; + const profileName = profile?.name ?? target.name; + const profileHash = createHash("sha256") + .update(JSON.stringify({ + name: profileName, + provider: target.provider, + model: target.model ?? null, + thinking: target.thinking ?? null, + body: profileBody, + })) + .digest("hex"); + const workflowPolicy = { version: 1 as const, access }; + const effectivePolicy = resolveWorkflowNodePolicy({ + workflowPolicy, + nodeConfig: { access }, + environment: context.env, + }); + const snapshot: JsonObject = { + profileBody, + profileName, + profileHash, + provider: target.provider, + model: target.model ?? null, + thinking: target.thinking ?? null, + effectivePolicy: effectivePolicy as unknown as JsonObject, + workspaceRoot: scope.workspaceRoot, + prompt, + timeoutMs: timeoutMs ?? null, + environmentPolicy: effectivePolicy.environment as JsonObject, + }; + const submitted = orchestrator.submitDetailed({ + definition: { + version: 1, + nodes: [{ key: "agent", type: "agent", config: snapshot }], + edges: [], + }, + input: { prompt }, + policy: workflowPolicy, + idempotencyKey, + workspace: scope, + }); + const supervisor = await ensureSupervisor({ + stateDir: config.stateDir, + cliEntrypoint: context.cliEntrypoint, + env: context.env, + }); + return { + workflow: submitted.workflow, + created: submitted.created, + supervisor: { + wakeGeneration: supervisor.requestedWakeGeneration, + spawned: supervisor.spawned, + ownerEpoch: supervisor.ownerEpoch ?? null, + }, + }; +} + +async function resolveWorkspaceScope( + context: Pick, + allowedRoots: string[], +): Promise { + const workspaceId = context.env.DEVSPACE_WORKSPACE_ID?.trim(); + if (!workspaceId) throw new WorkspaceDeniedError("DEVSPACE_WORKSPACE_ID is required for workflow commands."); + const requestedRoot = resolve(context.env.DEVSPACE_WORKSPACE_ROOT || context.cwd); + const workspaceRoot = await canonicalPath(requestedRoot, "workspace root"); + const cwd = await canonicalPath(context.cwd, "current directory"); + if (!isInside(cwd, workspaceRoot)) { + throw new WorkspaceDeniedError("Current directory is outside DEVSPACE_WORKSPACE_ROOT."); + } + const canonicalAllowedRoots = await Promise.all(allowedRoots.map((root) => canonicalPath(root, "allowed root"))); + if (!canonicalAllowedRoots.some((root) => isInside(workspaceRoot, root))) { + throw new WorkspaceDeniedError("Workspace root is outside configured allowed roots."); + } + return { workspaceId, workspaceRoot }; +} + +function loadWorkflowCliConfig(env: NodeJS.ProcessEnv, cwd: string): WorkflowCliConfig { + const files = loadDevspaceFiles(env); + const stateDir = resolve(expandHomePath( + env.DEVSPACE_STATE_DIR ?? files.config.stateDir ?? join(homedir(), ".local", "share", "devspace"), + )); + const rawRoots = env.DEVSPACE_ALLOWED_ROOTS + ? env.DEVSPACE_ALLOWED_ROOTS.split(",") + : files.config.allowedRoots ?? [cwd]; + return { + stateDir, + allowedRoots: rawRoots.map((root) => resolve(expandHomePath(root.trim()))).filter(Boolean), + devspaceAgentsDir: devspaceAgentsDir(env), + subagents: resolveSubagentsFlag(files.config, env) === true, + }; +} + +function resolveWorkflowTarget( + name: string, + profiles: LocalAgentProfile[], + model: string | undefined, + thinking: string | undefined, + env: NodeJS.ProcessEnv, +): ReturnType | { + kind: "provider"; + name: "fake"; + provider: "fake"; + model?: string; + thinking?: string; + profile?: undefined; +} | undefined { + if (name === "fake" && env.DEVSPACE_WORKFLOW_FAKE_PROVIDER === "1") { + return { kind: "provider", name: "fake", provider: "fake", model, thinking }; + } + return resolveLocalAgentTarget(name, profiles, model, thinking); +} + +function validateWorkflowArguments( + command: string, + args: string[], +): { positionals: string[] } { + const specifications: Record }> = { + __supervisor: { positionals: 0, valueOptions: new Set(["--state-dir"]) }, + run: { + positionals: 1, + valueOptions: new Set([ + "--prompt", + "--model", + "--thinking", + "--timeout-ms", + "--access", + "--idempotency-key", + ]), + }, + status: { positionals: 1, valueOptions: new Set() }, + events: { positionals: 1, valueOptions: new Set(["--after"]) }, + wait: { positionals: 1, valueOptions: new Set(["--timeout-ms", "--after"]) }, + cancel: { positionals: 1, valueOptions: new Set() }, + }; + const specification = specifications[command]; + if (!specification) throw new CliInputError(`Unknown workflows command: ${command}`); + + const positionals: string[] = []; + const seenOptions = new Set(); + for (let index = 0; index < args.length; index += 1) { + const part = args[index]!; + if (!part.startsWith("--")) { + positionals.push(part); + continue; + } + const separator = part.indexOf("="); + const name = separator === -1 ? part : part.slice(0, separator); + if (name === "--json") { + if (separator !== -1) throw new CliInputError("--json does not accept a value."); + } else if (specification.valueOptions.has(name)) { + const value = separator === -1 ? args[index + 1] : part.slice(separator + 1); + if (!value || value.startsWith("--")) throw new CliInputError(`Missing value for ${name}.`); + if (separator === -1) index += 1; + } else { + throw new CliInputError(`Unknown option for workflows ${command}: ${name}`); + } + if (seenOptions.has(name)) throw new CliInputError(`Duplicate option ${name}.`); + seenOptions.add(name); + } + if (positionals.length !== specification.positionals) { + throw new CliInputError(`workflows ${command} expects ${specification.positionals} positional argument${specification.positionals === 1 ? "" : "s"}.`); + } + return { positionals }; +} + +function requiredOption(args: string[], name: string): string { + const value = optionalOption(args, name); + if (!value) throw new CliInputError(`Missing required option ${name}.`); + return value; +} + +function optionalOption(args: string[], name: string): string | undefined { + for (let index = 0; index < args.length; index += 1) { + const part = args[index]; + if (part === name) { + const value = args[index + 1]; + if (!value || value.startsWith("--")) throw new CliInputError(`Missing value for ${name}.`); + return value; + } + if (part?.startsWith(`${name}=`)) { + const value = part.slice(name.length + 1); + if (!value) throw new CliInputError(`Missing value for ${name}.`); + return value; + } + } + return undefined; +} + +function integerOption(args: string[], name: string): number | undefined { + const raw = optionalOption(args, name); + if (raw === undefined) return undefined; + const parsed = Number(raw); + if (!Number.isSafeInteger(parsed)) throw new CliInputError(`${name} must be an integer.`); + return parsed; +} + +function writeEnvelope(stdout: Writable, value: Record): void { + stdout.write(`${JSON.stringify({ version: ENVELOPE_VERSION, ...value })}\n`); +} + +function normalizeCliError(error: unknown): { code: string; message: string; exitCode: number } { + if (error instanceof WorkspaceDeniedError) return { code: "workspace_denied", message: error.message, exitCode: 3 }; + if (error instanceof WorkflowNotFoundError) return { code: "not_found", message: error.message, exitCode: 4 }; + if (error instanceof WorkflowIdempotencyConflictError) return { code: "idempotency_conflict", message: error.message, exitCode: 5 }; + if (error instanceof CliInputError || error instanceof WorkflowValidationError) { + return { code: "invalid_input", message: error.message, exitCode: 2 }; + } + return { + code: "internal_error", + message: error instanceof Error ? error.message : String(error), + exitCode: 1, + }; +} + +function notFound(workflowId: string): never { + throw new WorkflowNotFoundError(workflowId); +} + +function isTerminal(workflow: WorkflowRunRecord): boolean { + return workflow.status === "succeeded" || workflow.status === "failed" || workflow.status === "cancelled"; +} + +async function canonicalPath(path: string, label: string): Promise { + try { + return await realpath(path); + } catch (error) { + throw new WorkspaceDeniedError(`Unable to canonicalize ${label}: ${error instanceof Error ? error.message : String(error)}`); + } +} + +function isInside(path: string, root: string): boolean { + const relationship = relative(root, path); + return relationship === "" || (!isAbsolute(relationship) && relationship !== ".." && !relationship.startsWith(`..${process.platform === "win32" ? "\\" : "/"}`)); +} + +function workflowsHelp(): string { + return [ + "DevSpace workflows", + "", + "Usage:", + " devspace workflows run --prompt --json [--idempotency-key ]", + " devspace workflows status --json", + " devspace workflows wait --json [--timeout-ms ] [--after ]", + " devspace workflows events --json [--after ]", + " devspace workflows cancel --json", + "", + ].join("\n"); +} + +class CliInputError extends Error {} +class WorkspaceDeniedError extends Error {} From 0092d28b48f683343c788b21b77e42d871838f86 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 01:31:37 +0530 Subject: [PATCH 4/4] test(workflows): cover supervisor and CLI lifecycle Co-Authored-By: Claude --- package.json | 2 +- src/oauth-store.test.ts | 1 + src/workflows/cli.test.ts | 161 +++++++++++++++++ src/workflows/migrations.test.ts | 20 ++- src/workflows/supervisor.test.ts | 295 +++++++++++++++++++++++++++++++ src/workflows/workflows.test.ts | 64 +++++++ 6 files changed, 538 insertions(+), 5 deletions(-) create mode 100644 src/workflows/cli.test.ts create mode 100644 src/workflows/supervisor.test.ts diff --git a/package.json b/package.json index d6a10dfc..9d0b9dd4 100644 --- a/package.json +++ b/package.json @@ -29,7 +29,7 @@ "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 && npm run test:workflows && tsx src/cli.test.ts", - "test:workflows": "tsx src/workflows/migrations.test.ts && tsx src/workflows/policy.test.ts && tsx src/workflows/workflows.test.ts", + "test:workflows": "tsx src/workflows/migrations.test.ts && tsx src/workflows/policy.test.ts && tsx src/workflows/workflows.test.ts && tsx src/workflows/supervisor.test.ts && tsx src/workflows/cli.test.ts", "typecheck": "tsc -p tsconfig.json --noEmit" }, "keywords": [], diff --git a/src/oauth-store.test.ts b/src/oauth-store.test.ts index de3c116b..b1e36d5c 100644 --- a/src/oauth-store.test.ts +++ b/src/oauth-store.test.ts @@ -45,6 +45,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise { { version: 2, name: "oauth-state" }, { version: 3, name: "local-agent-sessions" }, { version: 4, name: "durable-workflows" }, + { version: 5, name: "workflow-supervisor" }, ]); } finally { database.close(); diff --git a/src/workflows/cli.test.ts b/src/workflows/cli.test.ts new file mode 100644 index 00000000..ef506697 --- /dev/null +++ b/src/workflows/cli.test.ts @@ -0,0 +1,161 @@ +import assert from "node:assert/strict"; +import { spawnSync } from "node:child_process"; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; + +const root = mkdtempSync(join(tmpdir(), "devspace-workflows-cli-test-")); +const project = join(root, "project"); +const stateDir = join(root, "state"); +const configDir = join(root, "config"); +const agentsDir = join(configDir, "agents"); +const cliPath = fileURLToPath(new URL("../cli.ts", import.meta.url)); +const loaderPath = fileURLToPath(new URL("../../node_modules/tsx/dist/loader.mjs", import.meta.url)); +mkdirSync(project, { recursive: true }); +mkdirSync(stateDir, { recursive: true }); +mkdirSync(agentsDir, { recursive: true }); + +const profilePath = join(agentsDir, "reviewer.md"); +writeProfile(profilePath, "Original immutable profile."); + +const baseEnv = { + ...process.env, + DEVSPACE_CONFIG_DIR: configDir, + DEVSPACE_STATE_DIR: stateDir, + DEVSPACE_ALLOWED_ROOTS: project, + DEVSPACE_WORKSPACE_ID: "workspace-a", + DEVSPACE_WORKSPACE_ROOT: project, + DEVSPACE_SUBAGENTS: "1", + DEVSPACE_WORKFLOW_FAKE_PROVIDER: "1", + DEVSPACE_WORKFLOW_FAKE_DELAY_MS: "5000", +}; + +try { + const submitted = runCli([ + "workflows", "run", "reviewer", "--prompt", "private task", "--json", + "--idempotency-key", "shell-parent", + ], baseEnv); + assert.equal(submitted.status, 0, `${submitted.stderr}\n${submitted.stdout}`); + assert.equal(submitted.stdout.trim().split("\n").length, 1, "run stdout must contain one JSON object"); + assert.equal(submitted.stderr, ""); + const runEnvelope = JSON.parse(submitted.stdout) as Envelope; + assert.equal(runEnvelope.version, 1); + assert.equal(runEnvelope.ok, true); + const workflowId = runEnvelope.workflow!.id; + const replay = runCli([ + "workflows", "run", "reviewer", "--prompt", "private task", "--json", + "--idempotency-key", "shell-parent", + ], baseEnv); + assert.equal(replay.status, 0, replay.stderr); + assert.equal((JSON.parse(replay.stdout) as Envelope).created, false); + + writeProfile(profilePath, "Mutated after durable submission."); + const timed = runCli([ + "workflows", "wait", workflowId, "--timeout-ms", "0", "--json", + ], baseEnv); + assert.equal(timed.status, 0, timed.stderr); + const timedEnvelope = JSON.parse(timed.stdout) as Envelope; + assert.equal(timedEnvelope.timedOut, true); + + const waited = runCli([ + "workflows", "wait", workflowId, "--timeout-ms", "10000", "--after", "0", "--json", + ], baseEnv); + assert.equal(waited.status, 0, waited.stderr); + const waitEnvelope = JSON.parse(waited.stdout) as Envelope; + assert.equal(waitEnvelope.workflow!.status, "succeeded"); + assert.equal( + waitEnvelope.workflow!.definition.nodes[0].config.profileBody, + "Original immutable profile.", + ); + assert.ok(waitEnvelope.events!.some((event) => event.type === "provider.session")); + assert.ok(waitEnvelope.events!.some((event) => event.type === "provider.output")); + assert.equal(waitEnvelope.cursor, waitEnvelope.events!.at(-1)!.sequence); + + const repeated = runCli([ + "workflows", "wait", workflowId, "--timeout-ms", "5000", "--json", + ], baseEnv); + assert.equal(repeated.status, 0, repeated.stderr); + assert.equal((JSON.parse(repeated.stdout) as Envelope).workflow!.status, "succeeded"); + + const denied = runCli([ + "workflows", "status", workflowId, "--json", + ], { ...baseEnv, DEVSPACE_WORKSPACE_ID: "workspace-b" }); + assert.notEqual(denied.status, 0); + const deniedEnvelope = JSON.parse(denied.stdout) as Envelope; + assert.equal(deniedEnvelope.ok, false); + assert.equal(deniedEnvelope.error!.code, "not_found"); + + const invalid = runCli([ + "workflows", "run", "fake", "--json", + ], baseEnv); + assert.notEqual(invalid.status, 0); + assert.equal((JSON.parse(invalid.stdout) as Envelope).error!.code, "invalid_input"); + assert.equal(invalid.stdout.trim().split("\n").length, 1); + + const optionValueAsPositional = runCli([ + "workflows", "run", "--prompt", "reviewer", "--json", + ], baseEnv); + assert.notEqual(optionValueAsPositional.status, 0); + assert.equal((JSON.parse(optionValueAsPositional.stdout) as Envelope).error!.code, "invalid_input"); + + const unknownOption = runCli([ + "workflows", "status", workflowId, "--unknown", "value", "--json", + ], baseEnv); + assert.notEqual(unknownOption.status, 0); + assert.equal((JSON.parse(unknownOption.stdout) as Envelope).error!.code, "invalid_input"); + + const extraPositional = runCli([ + "workflows", "cancel", workflowId, "extra", "--json", + ], baseEnv); + assert.notEqual(extraPositional.status, 0); + assert.equal((JSON.parse(extraPositional.stdout) as Envelope).error!.code, "invalid_input"); + + const conflict = runCli([ + "workflows", "run", "reviewer", "--prompt", "different task", "--json", + "--idempotency-key", "shell-parent", + ], baseEnv); + assert.notEqual(conflict.status, 0); + assert.equal((JSON.parse(conflict.stdout) as Envelope).error!.code, "idempotency_conflict"); +} finally { + rmSync(root, { recursive: true, force: true }); +} + +function runCli(args: string[], env: NodeJS.ProcessEnv) { + return spawnSync(process.execPath, ["--import", loaderPath, cliPath, ...args], { + cwd: project, + env, + encoding: "utf8", + timeout: 15_000, + }); +} + +function writeProfile(path: string, body: string): void { + writeFileSync(path, [ + "---", + "name: reviewer", + "description: Workflow test reviewer.", + "provider: codex", + "model: test-model", + "thinking: high", + "---", + "", + body, + "", + ].join("\n")); +} + +interface Envelope { + version: number; + ok: boolean; + created?: boolean; + timedOut?: boolean; + cursor?: number; + workflow?: { + id: string; + status: string; + definition: { nodes: Array<{ config: { profileBody: string } }> }; + }; + events?: Array<{ type: string; sequence: number }>; + error?: { code: string; message: string }; +} diff --git a/src/workflows/migrations.test.ts b/src/workflows/migrations.test.ts index 30671448..6ddb3304 100644 --- a/src/workflows/migrations.test.ts +++ b/src/workflows/migrations.test.ts @@ -25,6 +25,7 @@ function testFreshSchema(stateDir: string): void { { version: 2, name: "oauth-state" }, { version: 3, name: "local-agent-sessions" }, { version: 4, name: "durable-workflows" }, + { version: 5, name: "workflow-supervisor" }, ], ); const tables = database.sqlite @@ -35,7 +36,15 @@ function testFreshSchema(stateDir: string): void { ) .pluck() .all(); - assert.deepEqual(tables, ["workflow_edges", "workflow_events", "workflow_nodes", "workflow_runs"]); + assert.deepEqual(tables, [ + "workflow_edges", + "workflow_events", + "workflow_node_attempts", + "workflow_nodes", + "workflow_provider_events", + "workflow_runs", + "workflow_supervisor", + ]); const edgeForeignKeys = database.sqlite.prepare("pragma foreign_key_list(workflow_edges)").all() as Array<{ id: number; @@ -89,7 +98,7 @@ function testExistingMigrationCompatibility(stateDir: string): void { try { assert.deepEqual( migrated.sqlite.prepare("select version from devspace_schema_migrations order by version").pluck().all(), - [1, 2, 3, 4], + [1, 2, 3, 4, 5], ); assert.deepEqual(migrated.sqlite.prepare("select * from workspace_sessions").all(), [ { @@ -162,7 +171,7 @@ function testExistingMigrationCompatibility(stateDir: string): void { .prepare("select count(*) from sqlite_master where type = 'table' and name like 'workflow_%'") .pluck() .get(), - 4, + 7, ); } finally { migrated.close(); @@ -213,11 +222,14 @@ function createVersion3Fixture(stateDir: string): void { const initialized = openDatabase(stateDir); try { initialized.sqlite.exec(` + drop table workflow_provider_events; + drop table workflow_node_attempts; + drop table workflow_supervisor; drop table workflow_edges; drop table workflow_events; drop table workflow_nodes; drop table workflow_runs; - delete from devspace_schema_migrations where version = 4; + delete from devspace_schema_migrations where version in (4, 5); insert into workspace_sessions ( id, root, status, mode, source_root, base_ref, base_sha, managed, created_at, last_used_at diff --git a/src/workflows/supervisor.test.ts b/src/workflows/supervisor.test.ts new file mode 100644 index 00000000..09732b8f --- /dev/null +++ b/src/workflows/supervisor.test.ts @@ -0,0 +1,295 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { LocalAgentRunController, type LocalAgentRunHandle } from "../local-agent-runtime.js"; +import { ensureSupervisor } from "./supervisor-launch.js"; +import { runWorkflowSupervisor } from "./supervisor.js"; +import { WorkflowStore } from "./store.js"; +import type { JsonObject, SubmitWorkflowRequest } from "./types.js"; + +const root = mkdtempSync(join(tmpdir(), "devspace-supervisor-test-")); +try { + await testExecutionAndEventReplay(join(root, "execution")); + await testDuplicateSupervisorPrevention(join(root, "singleton")); + await testStaleSupervisorProcessRecovery(join(root, "stale-supervisor")); + await testAtomicClaimAndLeaseRecovery(join(root, "claims")); + await testDispatchHeartbeatsProtectLeases(join(root, "dispatch-heartbeats")); + await testCancellationBeforeAndDuringExecution(join(root, "cancellation")); +} finally { + rmSync(root, { recursive: true, force: true }); +} + +async function testExecutionAndEventReplay(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + const snapshot = executionSnapshot({ profileBody: "Original profile" }); + const submitted = store.submit(workflowRequest(snapshot)).workflow; + snapshot.profileBody = "Mutated after submit"; + store.close(); + + const ran = await runWorkflowSupervisor(stateDir, { + idleMs: 20, + heartbeatMs: 5, + supervisorLeaseMs: 100, + nodeLeaseMs: 100, + handleFactory: successfulHandleFactory, + }); + assert.equal(ran, true); + + const reader = new WorkflowStore(stateDir); + try { + const workflow = reader.require(submitted.id); + assert.equal(workflow.status, "succeeded"); + assert.equal(workflow.nodes[0]!.attempt, 1); + assert.equal(workflow.nodes[0]!.definition.config!.profileBody, "Original profile"); + assert.equal((workflow.result as JsonObject).finalResponse, "completed"); + const first = reader.readEvents(workflow.id, { after: 0, limit: 4 }); + const second = reader.readEvents(workflow.id, { after: first.nextCursor, limit: 100 }); + assert.ok(second.events.some((event) => event.type === "provider.session")); + assert.ok(second.events.some((event) => event.type === "provider.output")); + assert.deepEqual( + [...first.events, ...second.events].map((event) => event.sequence), + Array.from({ length: first.events.length + second.events.length }, (_, index) => index + 1), + ); + } finally { + reader.close(); + } +} + +async function testDuplicateSupervisorPrevention(stateDir: string): Promise { + const owner = new WorkflowStore(stateDir); + const acquired = owner.acquireSupervisor({ ownerToken: "owner", ownerPid: 1, leaseMs: 1_000 }); + assert.ok(acquired); + assert.equal(await runWorkflowSupervisor(stateDir, { ownerToken: "other", idleMs: 5 }), false); + assert.equal(owner.getSupervisor()!.ownerToken, "owner"); + owner.releaseSupervisor(acquired!); + owner.close(); +} + +async function testStaleSupervisorProcessRecovery(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + const stale = store.acquireSupervisor({ + ownerToken: "stale-owner", + ownerPid: 2_147_483_647, + leaseMs: 60_000, + })!; + store.close(); + + const launched = await ensureSupervisor({ + stateDir, + cliEntrypoint: fileURLToPath(new URL("../cli.ts", import.meta.url)), + env: process.env, + startupTimeoutMs: 2_000, + }); + assert.equal(launched.spawned, true); + assert.ok((launched.ownerEpoch ?? 0) > stale.ownerEpoch); + + const cleanup = new WorkflowStore(stateDir); + const current = cleanup.getSupervisor(); + if (current) cleanup.releaseSupervisor(current); + cleanup.close(); +} + +async function testAtomicClaimAndLeaseRecovery(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + const first = store.submit(workflowRequest(executionSnapshot())).workflow; + const supervisor = store.acquireSupervisor({ ownerToken: "claims", ownerPid: 2, leaseMs: 1_000 })!; + const claim = store.claimNextAgentNode({ supervisor, claimToken: "first", leaseMs: 30 }); + assert.equal(claim?.workflow.id, first.id); + assert.equal( + store.claimNextAgentNode({ supervisor, claimToken: "duplicate", leaseMs: 30 }), + undefined, + ); + await delay(10); + assert.ok(store.heartbeatNode({ + workflowId: first.id, + nodeKey: "agent", + attempt: 1, + claimToken: "first", + leaseMs: 30, + })); + await delay(15); + assert.equal(store.reconcileExpiredClaims(), 0); + await delay(25); + assert.equal(store.reconcileExpiredClaims(), 1); + const failed = store.require(first.id); + assert.equal(failed.status, "failed"); + assert.equal((failed.error as JsonObject).code, "worker_lost"); + assert.equal(failed.nodes[0]!.attempt, 1, "worker loss must not retry by default"); + + const parallel = store.submit({ + definition: { + version: 1, + nodes: [ + { key: "first", type: "agent", config: executionSnapshot() }, + { key: "second", type: "agent", config: executionSnapshot() }, + ], + edges: [], + }, + }).workflow; + store.claimNode({ workflowId: parallel.id, nodeKey: "first", claimToken: "parallel-first", leaseMs: 10 }); + store.claimNode({ workflowId: parallel.id, nodeKey: "second", claimToken: "parallel-second", leaseMs: 10 }); + await delay(20); + assert.equal(store.reconcileExpiredClaims(), 1); + assert.deepEqual( + store.require(parallel.id).nodes.map((node) => node.status).sort(), + ["cancelled", "failed"], + ); + store.releaseSupervisor(supervisor); + store.close(); +} + +async function testDispatchHeartbeatsProtectLeases(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + const workflow = store.submit(workflowRequest(executionSnapshot())).workflow; + store.close(); + + const ran = await runWorkflowSupervisor(stateDir, { + idleMs: 20, + heartbeatMs: 5, + supervisorLeaseMs: 40, + nodeLeaseMs: 40, + handleFactory: async () => { + await delay(90); + return successfulHandleFactory(); + }, + }); + assert.equal(ran, true); + + const reader = new WorkflowStore(stateDir); + try { + assert.equal(reader.require(workflow.id).status, "succeeded"); + } finally { + reader.close(); + } +} + +async function testCancellationBeforeAndDuringExecution(stateDir: string): Promise { + const beforeStore = new WorkflowStore(stateDir); + const before = beforeStore.submit(workflowRequest(executionSnapshot())).workflow; + beforeStore.requestCancellation(before.id); + beforeStore.close(); + let starts = 0; + await runWorkflowSupervisor(stateDir, { + idleMs: 20, + heartbeatMs: 5, + supervisorLeaseMs: 100, + nodeLeaseMs: 100, + handleFactory: async (...args) => { + starts += 1; + return successfulHandleFactory(); + }, + }); + const afterBefore = new WorkflowStore(stateDir); + assert.equal(afterBefore.require(before.id).status, "cancelled"); + assert.equal(starts, 0); + + const during = afterBefore.submit(workflowRequest(executionSnapshot())).workflow; + afterBefore.close(); + const supervisorPromise = runWorkflowSupervisor(stateDir, { + idleMs: 20, + heartbeatMs: 5, + supervisorLeaseMs: 100, + nodeLeaseMs: 100, + handleFactory: delayedHandleFactory, + }); + await waitForStatus(stateDir, during.id, "running"); + const canceller = new WorkflowStore(stateDir); + canceller.requestCancellation(during.id); + canceller.requestSupervisorWake(); + canceller.close(); + await supervisorPromise; + const finalStore = new WorkflowStore(stateDir); + try { + assert.equal(finalStore.require(during.id).status, "cancelled"); + assert.equal(finalStore.require(during.id).nodes[0]!.status, "cancelled"); + } finally { + finalStore.close(); + } +} + +async function successfulHandleFactory(): Promise { + return completedHandle(5); +} + +async function delayedHandleFactory(): Promise { + return completedHandle(500); +} + +function completedHandle(delayMs: number): LocalAgentRunHandle { + const controller = new LocalAgentRunController("fake"); + let timer: NodeJS.Timeout | undefined; + let finish!: () => void; + const pump = new Promise((resolve) => { + finish = resolve; + timer = setTimeout(() => { + controller.emit({ type: "session", providerSessionId: "session-1", resumed: false }); + controller.emit({ type: "output", stream: "assistant", delta: "completed" }); + controller.succeed({ + provider: "fake", + providerSessionId: "session-1", + finalResponse: "completed", + items: [], + }); + finish(); + }, delayMs); + }); + controller.setLifecycle({ + cancel: () => { + if (timer) clearTimeout(timer); + finish(); + }, + dispose: () => { + if (timer) clearTimeout(timer); + finish(); + }, + pump, + }); + return controller; +} + +function workflowRequest(config: JsonObject): SubmitWorkflowRequest { + return { + definition: { version: 1, nodes: [{ key: "agent", type: "agent", config }], edges: [] }, + workspace: { workspaceId: "workspace", workspaceRoot: "/tmp/workspace" }, + }; +} + +function executionSnapshot(overrides: JsonObject = {}): JsonObject { + return { + profileBody: "", + profileName: "fake", + profileHash: "hash", + provider: "fake", + model: null, + thinking: null, + effectivePolicy: { + version: 1, + mode: "workflow", + access: "read_only", + environment: {}, + }, + workspaceRoot: "/tmp/workspace", + prompt: "task", + timeoutMs: null, + environmentPolicy: {}, + ...overrides, + }; +} + +async function waitForStatus(stateDir: string, workflowId: string, status: string): Promise { + const deadline = Date.now() + 2_000; + while (Date.now() < deadline) { + const store = new WorkflowStore(stateDir); + const current = store.require(workflowId).status; + store.close(); + if (current === status) return; + await delay(5); + } + throw new Error(`Timed out waiting for ${workflowId} to become ${status}`); +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} diff --git a/src/workflows/workflows.test.ts b/src/workflows/workflows.test.ts index 1f894ab7..e99850f6 100644 --- a/src/workflows/workflows.test.ts +++ b/src/workflows/workflows.test.ts @@ -44,6 +44,7 @@ async function runTests(): Promise { await testTransitionsAndClaims(join(root, "transitions")); testTerminalInvariantsAndAtomicity(join(root, "terminal-invariants")); testDagValidation(join(root, "dag")); + testAgentCompletionAdvancesDag(join(root, "dag-completion")); await testBoundedAndRepeatedWait(join(root, "wait")); } finally { rmSync(root, { recursive: true, force: true }); @@ -370,6 +371,69 @@ function testTerminalInvariantsAndAtomicity(stateDir: string): void { } } +function testAgentCompletionAdvancesDag(stateDir: string): void { + const store = new WorkflowStore(stateDir); + try { + const workflow = store.submit({ + definition: { + version: 1, + nodes: [agentNode("first"), agentNode("second")], + edges: [{ from: "first", to: "second" }], + }, + }).workflow; + const supervisor = store.acquireSupervisor({ ownerToken: "dag", ownerPid: 1, leaseMs: 1_000 })!; + const first = store.claimNextAgentNode({ supervisor, claimToken: "first-claim", leaseMs: 1_000 })!; + const afterFirst = store.completeAgentNode({ + workflowId: workflow.id, + nodeKey: "first", + attempt: first.node.attempt, + claimToken: first.node.claimToken!, + status: "succeeded", + result: { step: 1 }, + })!; + assert.equal(afterFirst.status, "running"); + assert.deepEqual(Object.fromEntries(afterFirst.nodes.map((node) => [node.key, node.status])), { + first: "succeeded", + second: "ready", + }); + + const second = store.claimNextAgentNode({ supervisor, claimToken: "second-claim", leaseMs: 1_000 })!; + assert.equal(second.node.key, "second"); + const completed = store.completeAgentNode({ + workflowId: workflow.id, + nodeKey: "second", + attempt: second.node.attempt, + claimToken: second.node.claimToken!, + status: "succeeded", + result: { step: 2 }, + })!; + assert.equal(completed.status, "succeeded"); + assert.deepEqual(completed.result, { step: 2 }); + + const failedWorkflow = store.submit({ + definition: { + version: 1, + nodes: [agentNode("first"), agentNode("second")], + edges: [{ from: "first", to: "second" }], + }, + }).workflow; + const failedClaim = store.claimNextAgentNode({ supervisor, claimToken: "failed-claim", leaseMs: 1_000 })!; + const failed = store.completeAgentNode({ + workflowId: failedWorkflow.id, + nodeKey: "first", + attempt: failedClaim.node.attempt, + claimToken: failedClaim.node.claimToken!, + status: "failed", + error: { code: "provider_failed" }, + })!; + assert.equal(failed.status, "failed"); + assert.equal(failed.nodes.find((node) => node.key === "second")!.status, "skipped"); + store.releaseSupervisor(supervisor); + } finally { + store.close(); + } +} + function testDagValidation(stateDir: string): void { const store = new WorkflowStore(stateDir); try {