From da506997583959ad93011ddd9d26b0bd4a16c83b Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 02:13:28 +0530 Subject: [PATCH 01/10] feat(workflows): add durable DAG scheduling Co-Authored-By: Claude --- src/db/migrations.ts | 38 +++++ src/db/schema.ts | 26 ++++ src/workflows/cli.ts | 97 +++--------- src/workflows/store.ts | 225 ++++++++++++++++++++++++++- src/workflows/submission.ts | 236 +++++++++++++++++++++++++++++ src/workflows/supervisor-launch.ts | 18 ++- src/workflows/supervisor.ts | 86 +++++++++-- src/workflows/types.ts | 25 +++ src/workflows/worktrees.ts | 176 +++++++++++++++++++++ 9 files changed, 831 insertions(+), 96 deletions(-) create mode 100644 src/workflows/submission.ts create mode 100644 src/workflows/worktrees.ts diff --git a/src/db/migrations.ts b/src/db/migrations.ts index 727ca990..83630ffb 100644 --- a/src/db/migrations.ts +++ b/src/db/migrations.ts @@ -32,6 +32,11 @@ const migrations: Migration[] = [ name: "workflow-supervisor", up: migrateWorkflowSupervisor, }, + { + version: 6, + name: "workflow-dag-scheduler", + up: migrateWorkflowDagScheduler, + }, ]; export function migrateDatabase(sqlite: Database.Database): void { @@ -342,6 +347,39 @@ function migrateWorkflowSupervisor(sqlite: Database.Database): void { `); } +function migrateWorkflowDagScheduler(sqlite: Database.Database): void { + addColumnIfMissing(sqlite, "workflow_runs", "max_concurrency", "integer not null default 1"); + addColumnIfMissing(sqlite, "workflow_runs", "last_dispatched_at", "text"); + addColumnIfMissing(sqlite, "workflow_nodes", "next_eligible_at", "text"); + + sqlite.exec(` + create table if not exists workflow_worktrees ( + workflow_run_id text not null, + node_key text not null, + attempt integer not null, + path text not null unique, + source_root text not null, + base_sha text not null, + state text not null check (state in ('allocated', 'active', 'preserved', 'removed', 'cleanup_failed')), + retain_until text, + cleanup_error text, + created_at text not null, + updated_at text not null, + primary key (workflow_run_id, node_key, attempt), + foreign key (workflow_run_id) references workflow_runs(id) on delete cascade + ); + + create index if not exists workflow_runs_dispatch_idx + on workflow_runs(status, last_dispatched_at, created_at); + + create index if not exists workflow_nodes_retry_idx + on workflow_nodes(workflow_run_id, status, next_eligible_at); + + create index if not exists workflow_worktrees_cleanup_idx + on workflow_worktrees(state, retain_until); + `); +} + function addColumnIfMissing( sqlite: Database.Database, table: "workspace_sessions" | "local_agent_sessions" | "workflow_runs" | "workflow_nodes", diff --git a/src/db/schema.ts b/src/db/schema.ts index a95107c3..216edc19 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -118,6 +118,8 @@ export const workflowRuns = sqliteTable( requestHash: text("request_hash").notNull(), workspaceId: text("workspace_id"), workspaceRoot: text("workspace_root"), + maxConcurrency: integer("max_concurrency").notNull().default(1), + lastDispatchedAt: text("last_dispatched_at"), resultJson: text("result_json"), errorJson: text("error_json"), cancellationRequestedAt: text("cancellation_requested_at"), @@ -145,6 +147,7 @@ export const workflowNodes = sqliteTable( claimToken: text("claim_token"), claimedAt: text("claimed_at"), claimExpiresAt: text("claim_expires_at"), + nextEligibleAt: text("next_eligible_at"), supervisorOwnerToken: text("supervisor_owner_token"), supervisorOwnerEpoch: integer("supervisor_owner_epoch"), heartbeatAt: text("heartbeat_at"), @@ -272,6 +275,29 @@ export const workflowProviderEvents = sqliteTable( ], ); +export const workflowWorktrees = sqliteTable( + "workflow_worktrees", + { + workflowRunId: text("workflow_run_id") + .notNull() + .references(() => workflowRuns.id, { onDelete: "cascade" }), + nodeKey: text("node_key").notNull(), + attempt: integer("attempt").notNull(), + path: text("path").notNull().unique(), + sourceRoot: text("source_root").notNull(), + baseSha: text("base_sha").notNull(), + state: text("state").notNull(), + retainUntil: text("retain_until"), + cleanupError: text("cleanup_error"), + createdAt: text("created_at").notNull(), + updatedAt: text("updated_at").notNull(), + }, + (table) => [ + primaryKey({ columns: [table.workflowRunId, table.nodeKey, table.attempt] }), + index("workflow_worktrees_cleanup_idx").on(table.state, table.retainUntil), + ], +); + export type WorkspaceSessionRow = typeof workspaceSessions.$inferSelect; export type NewWorkspaceSessionRow = typeof workspaceSessions.$inferInsert; export type LoadedAgentFileRow = typeof loadedAgentFiles.$inferSelect; diff --git a/src/workflows/cli.ts b/src/workflows/cli.ts index 339e85e5..e2db7147 100644 --- a/src/workflows/cli.ts +++ b/src/workflows/cli.ts @@ -1,19 +1,11 @@ -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 { loadLocalAgentProfiles } from "../local-agent-profiles.js"; import { expandHomePath } from "../roots.js"; -import { resolveWorkflowNodePolicy } from "./policy.js"; +import { createWorkflowSubmission } from "./submission.js"; import { WorkflowOrchestrator } from "./orchestrator.js"; import { WorkflowIdempotencyConflictError, @@ -23,7 +15,6 @@ import { import { ensureSupervisor } from "./supervisor-launch.js"; import { runWorkflowSupervisor } from "./supervisor.js"; import type { - JsonObject, WorkflowRunRecord, WorkflowWorkspaceScope, } from "./types.js"; @@ -43,6 +34,7 @@ export interface WorkflowsCliContext { interface WorkflowCliConfig { stateDir: string; + worktreeRoot: string; allowedRoots: string[]; devspaceAgentsDir: string; subagents: boolean; @@ -158,54 +150,24 @@ async function runCommand( } 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: [], + const request = await createWorkflowSubmission({ + intent: { + single: { + target: targetName, + prompt, + model, + thinking, + access, + timeoutMs, + }, + idempotencyKey, }, - input: { prompt }, - policy: workflowPolicy, - idempotencyKey, workspace: scope, + profiles, + worktreeRoot: config.worktreeRoot, + environment: context.env, }); + const submitted = orchestrator.submitDetailed(request); const supervisor = await ensureSupervisor({ stateDir: config.stateDir, cliEntrypoint: context.cliEntrypoint, @@ -251,32 +213,15 @@ function loadWorkflowCliConfig(env: NodeJS.ProcessEnv, cwd: string): WorkflowCli : files.config.allowedRoots ?? [cwd]; return { stateDir, + worktreeRoot: resolve(expandHomePath( + env.DEVSPACE_WORKTREE_ROOT ?? files.config.worktreeRoot ?? join(homedir(), ".devspace", "worktrees"), + )), 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[], diff --git a/src/workflows/store.ts b/src/workflows/store.ts index bdb3cd14..f00b0c38 100644 --- a/src/workflows/store.ts +++ b/src/workflows/store.ts @@ -28,6 +28,7 @@ import { type WorkflowSupervisorRecord, type WorkflowTransition, type WorkflowWorkspaceScope, + type WorkflowWorktreeRecord, } from "./types.js"; const MAX_EVENT_PAYLOAD_BYTES = 64 * 1024; @@ -35,6 +36,10 @@ const DEFAULT_EVENT_LIMIT = 100; const MAX_EVENT_LIMIT = 1_000; const DEFAULT_CLAIM_LEASE_MS = 5 * 60_000; const MAX_CLAIM_LEASE_MS = 24 * 60 * 60_000; +const MAX_WORKFLOW_NODES = 64; +const MAX_WORKFLOW_EDGES = 256; +const MAX_RUN_CONCURRENCY = 16; +const SAFE_NODE_KEY = /^[a-zA-Z0-9][a-zA-Z0-9._-]{0,63}$/; const TERMINAL_WORKFLOW_STATUSES = new Set([ "succeeded", @@ -78,6 +83,8 @@ interface WorkflowRunRow { request_hash: string; workspace_id: string | null; workspace_root: string | null; + max_concurrency: number; + last_dispatched_at: string | null; result_json: string | null; error_json: string | null; cancellation_requested_at: string | null; @@ -98,6 +105,7 @@ interface WorkflowNodeRow { claim_token: string | null; claimed_at: string | null; claim_expires_at: string | null; + next_eligible_at: string | null; supervisor_owner_token: string | null; supervisor_owner_epoch: number | null; heartbeat_at: string | null; @@ -136,6 +144,20 @@ interface WorkflowSupervisorRow { started_at: string | null; } +interface WorkflowWorktreeRow { + workflow_run_id: string; + node_key: string; + attempt: number; + path: string; + source_root: string; + base_sha: string; + state: WorkflowWorktreeRecord["state"]; + retain_until: string | null; + cleanup_error: string | null; + created_at: string; + updated_at: string; +} + interface WorkflowAttemptRow { node_id: string; workflow_run_id: string; @@ -215,8 +237,9 @@ export class WorkflowStore { .prepare( `insert into workflow_runs ( id, definition_version, status, definition_json, input_json, policy_json, - idempotency_key, request_hash, workspace_id, workspace_root, created_at, updated_at - ) values (?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + idempotency_key, request_hash, workspace_id, workspace_root, max_concurrency, + created_at, updated_at + ) values (?, ?, 'queued', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ) .run( workflowId, @@ -228,6 +251,7 @@ export class WorkflowStore { normalized.requestHash, normalized.workspace?.workspaceId ?? null, normalized.workspace?.workspaceRoot ?? null, + normalized.maxConcurrency, now, now, ); @@ -663,16 +687,23 @@ export class WorkflowStore { `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 (n.next_eligible_at is null or n.next_eligible_at <= ?) 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`, + and ( + select count(*) from workflow_nodes active + where active.workflow_run_id = r.id and active.status = 'running' + ) < r.max_concurrency + order by case when r.last_dispatched_at is null then 0 else 1 end, + r.last_dispatched_at, r.created_at, n.created_at, n.node_key limit 1`, ) - .get() as WorkflowNodeRow | undefined; + .get(now) 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 = ?, + next_eligible_at = null, attempt = attempt + 1, + supervisor_owner_token = ?, supervisor_owner_epoch = ?, heartbeat_at = ?, updated_at = ? where id = ? and status = 'ready' and claim_token is null`, ) @@ -713,6 +744,9 @@ export class WorkflowStore { now, now, ); + this.database.sqlite + .prepare("update workflow_runs set last_dispatched_at = ?, updated_at = ? where id = ?") + .run(now, now, node.workflow_run_id); const workflow = this.getWorkflowRow(node.workflow_run_id)!; if (workflow.status === "queued") { this.database.sqlite @@ -890,10 +924,44 @@ export class WorkflowStore { const status = workflow.status === "cancelling" ? "cancelled" : input.status; const resultJson = serializeOptionalJson(input.result); const errorJson = serializeOptionalObject(input.error); + const retryAt = status === "failed" + ? retryEligibleAt(parseJson(node.definition_json), input.attempt, input.error, now) + : undefined; + if (retryAt) { + const nodeUpdate = this.database.sqlite + .prepare( + `update workflow_nodes + set status = 'ready', result_json = null, error_json = ?, claim_token = null, + claimed_at = null, claim_expires_at = null, next_eligible_at = ?, + supervisor_owner_token = null, supervisor_owner_epoch = null, + heartbeat_at = ?, updated_at = ?, completed_at = null + where id = ? and status = 'running' and attempt = ? and claim_token = ?`, + ) + .run(errorJson, retryAt, now, now, node.id, input.attempt, input.claimToken); + if (nodeUpdate.changes !== 1) return false; + this.database.sqlite + .prepare( + `update workflow_node_attempts + set phase = 'terminal', terminal_status = 'failed', error_json = ?, + heartbeat_at = ?, updated_at = ?, completed_at = ? + where node_id = ? and attempt = ? and claim_token = ? and phase != 'terminal'`, + ) + .run(errorJson, now, now, now, node.id, input.attempt, input.claimToken); + this.insertEvent( + input.workflowId, + "node.retry_scheduled", + node.id, + { nodeKey: node.node_key, attempt: input.attempt, nextEligibleAt: retryAt }, + now, + ); + return true; + } const nodeUpdate = this.database.sqlite .prepare( `update workflow_nodes - set status = ?, result_json = ?, error_json = ?, claim_expires_at = null, + set status = ?, result_json = ?, error_json = ?, claim_token = null, + claimed_at = null, claim_expires_at = null, next_eligible_at = null, + supervisor_owner_token = null, supervisor_owner_epoch = null, heartbeat_at = ?, updated_at = ?, completed_at = ? where id = ? and status = 'running' and attempt = ? and claim_token = ?`, ) @@ -1064,10 +1132,76 @@ export class WorkflowStore { return row?.status === "cancelling"; } + hasPendingWork(): boolean { + return Boolean( + this.database.sqlite + .prepare("select 1 from workflow_runs where status in ('queued', 'running', 'cancelling') limit 1") + .get(), + ); + } + + recordWorktree(input: Omit): WorkflowWorktreeRecord { + const now = new Date().toISOString(); + this.database.sqlite + .prepare( + `insert into workflow_worktrees ( + workflow_run_id, node_key, attempt, path, source_root, base_sha, state, + retain_until, created_at, updated_at + ) values (?, ?, ?, ?, ?, ?, 'allocated', ?, ?, ?) + on conflict(workflow_run_id, node_key, attempt) do nothing`, + ) + .run( + input.workflowId, + input.nodeKey, + input.attempt, + input.path, + input.sourceRoot, + input.baseSha, + input.retainUntil ?? null, + now, + now, + ); + return this.requireWorktree(input.workflowId, input.nodeKey, input.attempt); + } + + getWorktree(workflowId: string, nodeKey: string, attempt: number): WorkflowWorktreeRecord | undefined { + const row = this.database.sqlite + .prepare( + `select * from workflow_worktrees + where workflow_run_id = ? and node_key = ? and attempt = ?`, + ) + .get(workflowId, nodeKey, attempt) as WorkflowWorktreeRow | undefined; + return row ? rowToWorktree(row) : undefined; + } + + updateWorktreeState( + workflowId: string, + nodeKey: string, + attempt: number, + state: WorkflowWorktreeRecord["state"], + cleanupError?: string, + ): WorkflowWorktreeRecord { + const now = new Date().toISOString(); + const changed = this.database.sqlite + .prepare( + `update workflow_worktrees set state = ?, cleanup_error = ?, updated_at = ? + where workflow_run_id = ? and node_key = ? and attempt = ?`, + ) + .run(state, cleanupError ?? null, now, workflowId, nodeKey, attempt); + if (changed.changes !== 1) throw new WorkflowValidationError("Unknown workflow worktree allocation"); + return this.requireWorktree(workflowId, nodeKey, attempt); + } + close(): void { this.database.close(); } + private requireWorktree(workflowId: string, nodeKey: string, attempt: number): WorkflowWorktreeRecord { + const record = this.getWorktree(workflowId, nodeKey, attempt); + if (!record) throw new WorkflowValidationError("Unknown workflow worktree allocation"); + return record; + } + private assertActiveSupervisor(identity: WorkflowSupervisorIdentity, now: string): void { const row = this.database.sqlite .prepare( @@ -1145,6 +1279,8 @@ export class WorkflowStore { requestHash: row.request_hash, workspaceId: row.workspace_id ?? undefined, workspaceRoot: row.workspace_root ?? undefined, + maxConcurrency: row.max_concurrency, + lastDispatchedAt: row.last_dispatched_at ?? undefined, result: parseOptionalJson(row.result_json), error: parseOptionalJson(row.error_json), cancellationRequestedAt: row.cancellation_requested_at ?? undefined, @@ -1296,6 +1432,7 @@ function normalizeSubmission(request: SubmitWorkflowRequest): { policyJson: string; idempotencyKey?: string; workspace?: WorkflowWorkspaceScope; + maxConcurrency: number; requestHash: string; } { const definition = normalizeDefinition(request.definition); @@ -1314,10 +1451,20 @@ function normalizeSubmission(request: SubmitWorkflowRequest): { workspaceRoot: requireNonEmptyString(request.workspace.workspaceRoot, "Workspace root"), } : undefined; + const maxConcurrency = readBoundedInteger(policy.maxConcurrency, 1, MAX_RUN_CONCURRENCY, "Workflow maxConcurrency"); const requestHash = createHash("sha256") .update(canonicalJson({ definition, input, policy, workspace: workspace ?? null })) .digest("hex"); - return { definition, definitionJson, inputJson, policyJson, idempotencyKey, workspace, requestHash }; + return { + definition, + definitionJson, + inputJson, + policyJson, + idempotencyKey, + workspace, + maxConcurrency, + requestHash, + }; } function normalizeDefinition(definition: WorkflowDefinition): WorkflowDefinition { @@ -1329,9 +1476,15 @@ function normalizeDefinition(definition: WorkflowDefinition): WorkflowDefinition if (!Array.isArray(definition.nodes) || definition.nodes.length === 0) { throw new WorkflowValidationError("Workflow definition must contain at least one node"); } + if (definition.nodes.length > MAX_WORKFLOW_NODES) { + throw new WorkflowValidationError(`Workflow definition exceeds ${MAX_WORKFLOW_NODES} nodes`); + } if (definition.edges !== undefined && !Array.isArray(definition.edges)) { throw new WorkflowValidationError("Workflow definition edges must be an array"); } + if ((definition.edges?.length ?? 0) > MAX_WORKFLOW_EDGES) { + throw new WorkflowValidationError(`Workflow definition exceeds ${MAX_WORKFLOW_EDGES} edges`); + } const keys = new Set(); const nodes = definition.nodes.map((node, index): WorkflowNodeDefinitionV1 => { @@ -1340,6 +1493,9 @@ function normalizeDefinition(definition: WorkflowDefinition): WorkflowDefinition } const key = node.key.trim(); if (!key) throw new WorkflowValidationError(`Workflow node at index ${index} has an empty key`); + if (!SAFE_NODE_KEY.test(key)) { + throw new WorkflowValidationError(`Workflow node key is unsafe: ${key}`); + } if (keys.has(key)) throw new WorkflowValidationError(`Duplicate workflow node key: ${key}`); keys.add(key); return { key, type: "agent", config: normalizeObject(node.config ?? {}, `Node ${key} config`) }; @@ -1469,6 +1625,7 @@ function rowToWorkflowNode(row: WorkflowNodeRow): WorkflowNodeRecord { claimToken: row.claim_token ?? undefined, claimedAt: row.claimed_at ?? undefined, claimExpiresAt: row.claim_expires_at ?? undefined, + nextEligibleAt: row.next_eligible_at ?? undefined, result: parseOptionalJson(row.result_json), error: parseOptionalJson(row.error_json), createdAt: row.created_at, @@ -1477,6 +1634,22 @@ function rowToWorkflowNode(row: WorkflowNodeRow): WorkflowNodeRecord { }; } +function rowToWorktree(row: WorkflowWorktreeRow): WorkflowWorktreeRecord { + return { + workflowId: row.workflow_run_id, + nodeKey: row.node_key, + attempt: row.attempt, + path: row.path, + sourceRoot: row.source_root, + baseSha: row.base_sha, + state: row.state, + retainUntil: row.retain_until ?? undefined, + cleanupError: row.cleanup_error ?? undefined, + createdAt: row.created_at, + updatedAt: row.updated_at, + }; +} + function rowToAttempt(row: WorkflowAttemptRow): WorkflowNodeAttemptRecord { return { nodeId: row.node_id, @@ -1583,6 +1756,44 @@ function validateLease(leaseMs: number, label: string): void { } } +function retryEligibleAt( + definition: WorkflowNodeDefinitionV1, + attempt: number, + error: JsonObject | undefined, + now: string, +): string | undefined { + const config = definition.config ?? {}; + const effectivePolicy = isObject(config.effectivePolicy) ? config.effectivePolicy : undefined; + if (effectivePolicy?.access !== "read_only") return undefined; + const retry = isObject(config.retry) ? config.retry : undefined; + if (!retry) return undefined; + const maxAttempts = readBoundedInteger(retry.maxAttempts, 1, 10, "Retry maxAttempts"); + if (attempt >= maxAttempts) return undefined; + const code = typeof error?.code === "string" ? error.code : undefined; + const retryOn = Array.isArray(retry.retryOn) + ? retry.retryOn.filter((value): value is string => typeof value === "string") + : []; + if (!code || !retryOn.includes(code)) return undefined; + const backoffMs = retry.backoffMs === undefined + ? 0 + : readBoundedInteger(retry.backoffMs, 0, 60_000, "Retry backoffMs", 0); + return new Date(new Date(now).getTime() + backoffMs * attempt).toISOString(); +} + +function readBoundedInteger( + value: JsonValue | undefined, + fallback: number, + maximum: number, + label: string, + minimum = 1, +): number { + if (value === undefined || value === null) return fallback; + if (!Number.isSafeInteger(value) || Number(value) < minimum || Number(value) > maximum) { + throw new WorkflowValidationError(`${label} must be between ${minimum} and ${maximum}`); + } + return Number(value); +} + function requireNonEmptyString(value: string, label: string): string { const normalized = value.trim(); if (!normalized) throw new WorkflowValidationError(`${label} must not be empty`); diff --git a/src/workflows/submission.ts b/src/workflows/submission.ts new file mode 100644 index 00000000..c1285a19 --- /dev/null +++ b/src/workflows/submission.ts @@ -0,0 +1,236 @@ +import { createHash } from "node:crypto"; +import { execFile } from "node:child_process"; +import { realpath } from "node:fs/promises"; +import { promisify } from "node:util"; +import type { LocalAgentProfile } from "../local-agent-profiles.js"; +import { formatAvailableLocalAgentTargets, resolveLocalAgentTarget } from "../local-agent-targets.js"; +import { resolveWorkflowNodePolicy, type WorkflowAgentAccess } from "./policy.js"; +import { WorkflowValidationError } from "./store.js"; +import type { + JsonObject, + SubmitWorkflowRequest, + WorkflowEdgeDefinitionV1, + WorkflowNodeDefinitionV1, + WorkflowRetryClass, + WorkflowWorkspaceScope, +} from "./types.js"; + +const execFileAsync = promisify(execFile); +const MAX_NODE_TIMEOUT_MS = 24 * 60 * 60_000; +const MAX_ATTEMPTS = 10; +const MAX_RETRY_BACKOFF_MS = 60_000; + +export interface WorkflowAgentIntent { + key?: string; + target: string; + prompt: string; + model?: string; + thinking?: string; + access?: WorkflowAgentAccess; + timeoutMs?: number; + retry?: { + maxAttempts?: number; + retryOn?: WorkflowRetryClass[]; + backoffMs?: number; + }; +} + +export interface WorkflowDagIntent { + version: 1; + nodes: Array; + edges?: WorkflowEdgeDefinitionV1[]; + maxConcurrency?: number; + access?: WorkflowAgentAccess; +} + +export interface WorkflowSubmissionIntent { + single?: WorkflowAgentIntent; + dag?: WorkflowDagIntent; + idempotencyKey?: string; +} + +export async function createWorkflowSubmission(input: { + intent: WorkflowSubmissionIntent; + workspace: WorkflowWorkspaceScope; + profiles: LocalAgentProfile[]; + worktreeRoot: string; + environment?: NodeJS.ProcessEnv; +}): Promise { + if (Boolean(input.intent.single) === Boolean(input.intent.dag)) { + throw new WorkflowValidationError("Provide exactly one of single or dag"); + } + const environment = input.environment ?? process.env; + const dag: WorkflowDagIntent = input.intent.dag ?? { + version: 1, + nodes: [{ ...input.intent.single!, key: input.intent.single!.key ?? "agent" }], + edges: [], + maxConcurrency: 1, + access: input.intent.single!.access ?? "read_only", + }; + if (dag.version !== 1) throw new WorkflowValidationError("Unsupported workflow DAG version"); + const workflowAccess = dag.access ?? "read_only"; + const hasWriteNode = dag.nodes.some((node) => (node.access ?? workflowAccess) === "workspace_write"); + const baseSha = hasWriteNode ? await resolveBaseSha(input.workspace.workspaceRoot) : undefined; + const nodes: WorkflowNodeDefinitionV1[] = dag.nodes.map((node) => ({ + key: node.key, + type: "agent", + config: snapshotNode({ + node, + workflowAccess, + workspaceRoot: input.workspace.workspaceRoot, + worktreeRoot: input.worktreeRoot, + baseSha, + profiles: input.profiles, + environment, + }), + })); + + return { + definition: { version: 1, nodes, edges: dag.edges ?? [] }, + input: { kind: input.intent.dag ? "dag" : "single" }, + policy: { + version: 1, + access: workflowAccess, + maxConcurrency: dag.maxConcurrency ?? 1, + }, + idempotencyKey: input.intent.idempotencyKey, + workspace: input.workspace, + }; +} + +function snapshotNode(input: { + node: WorkflowAgentIntent; + workflowAccess: WorkflowAgentAccess; + workspaceRoot: string; + worktreeRoot: string; + baseSha?: string; + profiles: LocalAgentProfile[]; + environment: NodeJS.ProcessEnv; +}): JsonObject { + const prompt = input.node.prompt.trim(); + if (!prompt) throw new WorkflowValidationError(`Workflow node ${input.node.key ?? "agent"} prompt is empty`); + const timeoutMs = input.node.timeoutMs; + if (timeoutMs !== undefined && (!Number.isSafeInteger(timeoutMs) || timeoutMs < 1 || timeoutMs > MAX_NODE_TIMEOUT_MS)) { + throw new WorkflowValidationError(`Workflow node timeoutMs must be between 1 and ${MAX_NODE_TIMEOUT_MS}`); + } + const target = resolveTarget( + input.node.target, + input.profiles, + input.node.model, + input.node.thinking, + input.environment, + ); + if (!target) { + throw new WorkflowValidationError( + `Unknown workflow profile or provider: ${input.node.target}. Available ${formatAvailableLocalAgentTargets(input.profiles)}`, + ); + } + const profile = target.kind === "profile" ? target.profile : undefined; + const profileBody = profile?.body.trim() ?? ""; + const profileName = profile?.name ?? target.name; + const effectivePolicy = resolveWorkflowNodePolicy({ + workflowPolicy: { access: input.workflowAccess }, + nodeConfig: { access: input.node.access ?? input.workflowAccess }, + environment: input.environment, + }); + const retry = normalizeRetry(input.node.retry); + if (effectivePolicy.access === "workspace_write" && retry.maxAttempts > 1) { + throw new WorkflowValidationError("workspace_write nodes cannot be retried automatically"); + } + if (effectivePolicy.access === "workspace_write" && !input.baseSha) { + throw new WorkflowValidationError("workspace_write nodes require a pinned Git base SHA"); + } + const profileHash = createHash("sha256") + .update(JSON.stringify({ + name: profileName, + provider: target.provider, + model: target.model ?? null, + thinking: target.thinking ?? null, + body: profileBody, + })) + .digest("hex"); + + return { + profileBody, + profileName, + profileHash, + provider: target.provider, + model: target.model ?? null, + thinking: target.thinking ?? null, + effectivePolicy: effectivePolicy as unknown as JsonObject, + workspaceRoot: input.workspaceRoot, + worktreeRoot: input.worktreeRoot, + baseSha: input.baseSha ?? null, + prompt, + timeoutMs: timeoutMs ?? null, + retry: retry as unknown as JsonObject, + environmentPolicy: effectivePolicy.environment as JsonObject, + }; +} + +function normalizeRetry(retry: WorkflowAgentIntent["retry"]): { + maxAttempts: number; + retryOn: WorkflowRetryClass[]; + backoffMs: number; +} { + const maxAttempts = retry?.maxAttempts ?? 1; + const backoffMs = retry?.backoffMs ?? 0; + const retryOn = [...new Set(retry?.retryOn ?? [])]; + if (!Number.isSafeInteger(maxAttempts) || maxAttempts < 1 || maxAttempts > MAX_ATTEMPTS) { + throw new WorkflowValidationError(`Retry maxAttempts must be between 1 and ${MAX_ATTEMPTS}`); + } + if (!Number.isSafeInteger(backoffMs) || backoffMs < 0 || backoffMs > MAX_RETRY_BACKOFF_MS) { + throw new WorkflowValidationError(`Retry backoffMs must be between 0 and ${MAX_RETRY_BACKOFF_MS}`); + } + for (const value of retryOn) { + if (value !== "provider_failed" && value !== "timed_out") { + throw new WorkflowValidationError(`Unsupported retry class: ${value}`); + } + } + if (maxAttempts > 1 && retryOn.length === 0) { + throw new WorkflowValidationError("Retry policy must name at least one retryable failure class"); + } + return { maxAttempts, retryOn, backoffMs }; +} + +function resolveTarget( + name: string, + profiles: LocalAgentProfile[], + model: string | undefined, + thinking: string | undefined, + environment: NodeJS.ProcessEnv, +): ReturnType | { + kind: "provider"; + name: "fake"; + provider: "fake"; + model?: string; + thinking?: string; + profile?: undefined; +} | undefined { + if (name === "fake" && environment.DEVSPACE_WORKFLOW_FAKE_PROVIDER === "1") { + return { kind: "provider", name: "fake", provider: "fake", model, thinking }; + } + return resolveLocalAgentTarget(name, profiles, model, thinking); +} + +async function resolveBaseSha(workspaceRoot: string): Promise { + try { + const { stdout: root } = await execFileAsync("git", ["rev-parse", "--show-toplevel"], { + cwd: workspaceRoot, + maxBuffer: 1024 * 1024, + }); + const { stdout: sha } = await execFileAsync("git", ["rev-parse", "--verify", "HEAD^{commit}"], { + cwd: workspaceRoot, + maxBuffer: 1024 * 1024, + }); + if (!root.trim() || !sha.trim()) throw new Error("missing Git root or commit"); + if ((await realpath(root.trim())) !== (await realpath(workspaceRoot))) { + throw new Error("workspace root must be the Git repository root"); + } + return sha.trim(); + } catch (error) { + throw new WorkflowValidationError( + `workspace_write workflows require a Git repository with a commit: ${error instanceof Error ? error.message : String(error)}`, + ); + } +} diff --git a/src/workflows/supervisor-launch.ts b/src/workflows/supervisor-launch.ts index a1a24a45..209bcea6 100644 --- a/src/workflows/supervisor-launch.ts +++ b/src/workflows/supervisor-launch.ts @@ -63,13 +63,17 @@ export async function ensureSupervisor(input: { windowsHide: true, }, ); + let childExit: { code: number | null; signal: NodeJS.Signals | null } | undefined; + child.once("exit", (code, signal) => { + childExit = { code, signal }; + }); await new Promise((resolve, reject) => { child.once("spawn", resolve); child.once("error", reject); }); child.unref(); - const deadline = Date.now() + (input.startupTimeoutMs ?? 2_000); + const deadline = Date.now() + (input.startupTimeoutMs ?? 5_000); while (Date.now() < deadline) { const supervisor = store.getSupervisor(); if (supervisor?.leaseExpiresAt && supervisor.leaseExpiresAt > new Date().toISOString()) { @@ -79,14 +83,24 @@ export async function ensureSupervisor(input: { ownerEpoch: supervisor.ownerEpoch, }; } + if (childExit) { + throw new Error( + `Workflow supervisor exited before acquiring its durable lease (${formatChildExit(childExit)}).`, + ); + } await delay(20); } - throw new Error("Workflow supervisor did not acquire its durable lease."); + throw new Error("Workflow supervisor did not acquire its durable lease before the startup timeout."); } finally { store.close(); } } +function formatChildExit(exit: { code: number | null; signal: NodeJS.Signals | null }): string { + if (exit.signal) return `signal ${exit.signal}`; + return `exit code ${exit.code ?? "unknown"}`; +} + function isProcessRunning(pid: number): boolean { if (!Number.isSafeInteger(pid) || pid < 1) return false; try { diff --git a/src/workflows/supervisor.ts b/src/workflows/supervisor.ts index e34d412d..ef1a24af 100644 --- a/src/workflows/supervisor.ts +++ b/src/workflows/supervisor.ts @@ -9,6 +9,11 @@ import { } from "../local-agent-runtime.js"; import { isLocalAgentProvider } from "../local-agent-profiles.js"; import type { EffectiveLocalAgentPolicy } from "./policy.js"; +import { + allocateWorkflowWorktree, + cleanupWorkflowWorktree, + preserveWorkflowWorktree, +} from "./worktrees.js"; import { WorkflowStore, WorkflowValidationError } from "./store.js"; import type { JsonObject, @@ -22,6 +27,8 @@ 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 DEFAULT_GLOBAL_CONCURRENCY = 4; +const MAX_GLOBAL_CONCURRENCY = 32; const MAX_ERROR_CHARS = 16_384; const MAX_TIMEOUT_MS = 24 * 60 * 60_000; @@ -37,6 +44,7 @@ export interface WorkflowSupervisorOptions { idleMs?: number; ownerToken?: string; ownerPid?: number; + globalConcurrency?: number; handleFactory?: WorkflowHandleFactory; } @@ -49,7 +57,8 @@ export async function runWorkflowSupervisor( 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 globalConcurrency = options.globalConcurrency ?? DEFAULT_GLOBAL_CONCURRENCY; + validateSupervisorTiming(supervisorLeaseMs, nodeLeaseMs, heartbeatMs, idleMs, globalConcurrency); const acquired = store.acquireSupervisor({ ownerToken: options.ownerToken ?? randomUUID(), ownerPid: options.ownerPid ?? process.pid, @@ -61,6 +70,7 @@ export async function runWorkflowSupervisor( } const identity: WorkflowSupervisorIdentity = acquired; + const active = new Set>(); let lastWorkAt = Date.now(); try { store.reconcileExpiredClaims(); @@ -68,27 +78,43 @@ export async function runWorkflowSupervisor( while (store.heartbeatSupervisor(identity, supervisorLeaseMs)) { store.reconcileExpiredClaims(); store.convergeCancellations(); - const claim = store.claimNextAgentNode({ - supervisor: identity, - claimToken: randomUUID(), - leaseMs: nodeLeaseMs, - }); - if (claim) { + let claimed = false; + while (active.size < globalConcurrency) { + const claim = store.claimNextAgentNode({ + supervisor: identity, + claimToken: randomUUID(), + leaseMs: nodeLeaseMs, + }); + if (!claim) break; + claimed = true; lastWorkAt = Date.now(); - await executeClaim(store, identity, claim, { + let execution!: Promise; + execution = executeClaim(store, identity, claim, { supervisorLeaseMs, nodeLeaseMs, heartbeatMs, handleFactory: options.handleFactory ?? defaultHandleFactory, + }).finally(() => { + active.delete(execution); + lastWorkAt = Date.now(); }); - lastWorkAt = Date.now(); - continue; + active.add(execution); } + if (claimed) continue; const supervisor = store.getSupervisor(); if (!supervisor || supervisor.ownerToken !== identity.ownerToken || supervisor.ownerEpoch !== identity.ownerEpoch) { + await Promise.allSettled(active); return false; } + if (active.size > 0) { + await Promise.race([...active, delay(heartbeatMs)]); + continue; + } + if (store.hasPendingWork()) { + await delay(heartbeatMs); + continue; + } if (Date.now() - lastWorkAt >= idleMs) { if (store.releaseSupervisor(identity, supervisor.wakeGeneration)) return true; lastWorkAt = Date.now(); @@ -96,6 +122,7 @@ export async function runWorkflowSupervisor( } await delay(Math.min(heartbeatMs, Math.max(10, idleMs - (Date.now() - lastWorkAt)))); } + await Promise.allSettled(active); return false; } finally { store.releaseSupervisor(identity); @@ -126,6 +153,9 @@ async function executeClaim( let timeout: NodeJS.Timeout | undefined; let cancellationObserved = false; let timeoutObserved = false; + let worktreeAllocated = false; + let allocatedWorktreeRoot: string | undefined; + let completedSuccessfully = false; let leaseFailure: Error | undefined; let tickRunning = false; @@ -134,6 +164,18 @@ async function executeClaim( const fullPrompt = execution.profileBody ? `${execution.profileBody}\n\nTask:\n${execution.prompt}` : execution.prompt; + if (execution.effectivePolicy.access === "workspace_write") { + const allocation = await allocateWorkflowWorktree({ + store, + identity, + sourceRoot: execution.workspaceRoot, + worktreeRoot: execution.worktreeRoot, + baseSha: execution.baseSha, + }); + execution.workspaceRoot = allocation.path; + allocatedWorktreeRoot = execution.worktreeRoot; + worktreeAllocated = true; + } if (!store.markNodeDispatching(identity)) { throw new Error("Workflow node attempt was lost before dispatch."); } @@ -201,7 +243,7 @@ async function executeClaim( store.recordNodeProviderSession(identity, result.providerSessionId); } const cancelled = cancellationObserved || store.isCancellationRequested(identity.workflowId); - store.completeAgentNode({ + const completed = store.completeAgentNode({ ...identity, status: cancelled ? "cancelled" : "succeeded", result: { @@ -210,6 +252,7 @@ async function executeClaim( finalResponse: result.finalResponse, }, }); + completedSuccessfully = Boolean(completed && !cancelled); } catch (error) { const cancelled = !timeoutObserved && ( cancellationObserved || store.isCancellationRequested(identity.workflowId) || isAbortError(error) @@ -223,6 +266,17 @@ async function executeClaim( if (heartbeat) clearInterval(heartbeat); if (timeout) clearTimeout(timeout); await handle?.dispose().catch(() => undefined); + if (worktreeAllocated) { + if (completedSuccessfully && allocatedWorktreeRoot) { + await cleanupWorkflowWorktree({ + store, + identity, + worktreeRoot: allocatedWorktreeRoot, + }).catch(() => undefined); + } else { + preserveWorkflowWorktree(store, identity); + } + } store.convergeCancellations(); } } @@ -261,6 +315,8 @@ function readExecutionSnapshot(config: JsonObject): { profileBody: string; prompt: string; workspaceRoot: string; + worktreeRoot: string; + baseSha: string; timeoutMs?: number; effectivePolicy: EffectiveLocalAgentPolicy; } { @@ -283,6 +339,8 @@ function readExecutionSnapshot(config: JsonObject): { profileBody: optionalString(config.profileBody) ?? "", prompt: requiredString(config.prompt, "prompt"), workspaceRoot: requiredString(config.workspaceRoot, "workspaceRoot"), + worktreeRoot: optionalString(config.worktreeRoot) ?? "", + baseSha: optionalString(config.baseSha) ?? "", timeoutMs, effectivePolicy: policy as unknown as EffectiveLocalAgentPolicy, }; @@ -364,6 +422,7 @@ function validateSupervisorTiming( nodeLeaseMs: number, heartbeatMs: number, idleMs: number, + globalConcurrency: number, ): void { for (const [label, value] of [ ["supervisor lease", supervisorLeaseMs], @@ -377,6 +436,11 @@ function validateSupervisorTiming( if (!Number.isSafeInteger(idleMs) || idleMs < 0) { throw new WorkflowValidationError("Workflow supervisor idle timeout must be a non-negative integer."); } + if (!Number.isSafeInteger(globalConcurrency) || globalConcurrency < 1 || globalConcurrency > MAX_GLOBAL_CONCURRENCY) { + throw new WorkflowValidationError( + `Workflow global concurrency must be between 1 and ${MAX_GLOBAL_CONCURRENCY}.`, + ); + } if (heartbeatMs >= Math.min(supervisorLeaseMs, nodeLeaseMs)) { throw new WorkflowValidationError("Workflow heartbeat interval must be shorter than both leases."); } diff --git a/src/workflows/types.ts b/src/workflows/types.ts index 1f0597c9..903bb587 100644 --- a/src/workflows/types.ts +++ b/src/workflows/types.ts @@ -45,6 +45,14 @@ export interface WorkflowDefinitionV1 { export type WorkflowDefinition = WorkflowDefinitionV1; +export type WorkflowRetryClass = "provider_failed" | "timed_out"; + +export interface WorkflowRetryPolicy extends JsonObject { + maxAttempts: number; + retryOn: WorkflowRetryClass[]; + backoffMs: number; +} + export interface WorkflowPolicyV1 extends JsonObject { version: typeof WORKFLOW_POLICY_VERSION; } @@ -70,6 +78,7 @@ export interface WorkflowNodeRecord { claimToken?: string; claimedAt?: string; claimExpiresAt?: string; + nextEligibleAt?: string; result?: JsonValue; error?: JsonObject; createdAt: string; @@ -96,6 +105,8 @@ export interface WorkflowRunRecord { requestHash: string; workspaceId?: string; workspaceRoot?: string; + maxConcurrency: number; + lastDispatchedAt?: string; result?: JsonValue; error?: JsonObject; cancellationRequestedAt?: string; @@ -185,6 +196,20 @@ export interface WorkflowEventReadOptions { limit?: number; } +export interface WorkflowWorktreeRecord { + workflowId: string; + nodeKey: string; + attempt: number; + path: string; + sourceRoot: string; + baseSha: string; + state: "allocated" | "active" | "preserved" | "removed" | "cleanup_failed"; + createdAt: string; + updatedAt: string; + retainUntil?: string; + cleanupError?: string; +} + export interface WorkflowNodeClaim { workflowId: string; nodeKey: string; diff --git a/src/workflows/worktrees.ts b/src/workflows/worktrees.ts new file mode 100644 index 00000000..5bd98d35 --- /dev/null +++ b/src/workflows/worktrees.ts @@ -0,0 +1,176 @@ +import { execFile } from "node:child_process"; +import { mkdir, realpath } from "node:fs/promises"; +import { basename, join, relative, resolve } from "node:path"; +import { promisify } from "node:util"; +import { WorkflowStore, WorkflowValidationError } from "./store.js"; +import type { WorkflowAttemptIdentity, WorkflowWorktreeRecord } from "./types.js"; + +const execFileAsync = promisify(execFile); +const FAILED_WORKTREE_RETENTION_MS = 7 * 24 * 60 * 60_000; + +export async function allocateWorkflowWorktree(input: { + store: WorkflowStore; + identity: WorkflowAttemptIdentity; + sourceRoot: string; + worktreeRoot: string; + baseSha: string; +}): Promise { + const existing = input.store.getWorktree( + input.identity.workflowId, + input.identity.nodeKey, + input.identity.attempt, + ); + if (existing) { + if (existing.state === "active") return existing; + throw new WorkflowValidationError( + `Workflow worktree attempt already has allocation state ${existing.state}`, + ); + } + + const sourceRoot = await canonicalDirectory(input.sourceRoot, "workflow source root"); + await mkdir(input.worktreeRoot, { recursive: true }); + const worktreeRoot = await canonicalDirectory(input.worktreeRoot, "workflow worktree root"); + const gitRoot = await git(["rev-parse", "--show-toplevel"], sourceRoot); + if ((await realpath(gitRoot.trim())) !== sourceRoot) { + throw new WorkflowValidationError("Workflow workspace_write source must be a Git repository root"); + } + const baseSha = (await git(["rev-parse", "--verify", `${input.baseSha}^{commit}`], sourceRoot)).trim(); + if (baseSha !== input.baseSha) { + throw new WorkflowValidationError("Workflow worktree base SHA no longer resolves to the pinned commit"); + } + + const path = join( + worktreeRoot, + `${safeSegment(basename(sourceRoot))}-${safeSegment(input.identity.workflowId)}-${safeSegment(input.identity.nodeKey)}-a${input.identity.attempt}`, + ); + assertContained(path, worktreeRoot); + const record = input.store.recordWorktree({ + workflowId: input.identity.workflowId, + nodeKey: input.identity.nodeKey, + attempt: input.identity.attempt, + path, + sourceRoot, + baseSha, + retainUntil: new Date(Date.now() + FAILED_WORKTREE_RETENTION_MS).toISOString(), + }); + try { + await git(["worktree", "add", "--detach", path, baseSha], sourceRoot); + return input.store.updateWorktreeState( + record.workflowId, + record.nodeKey, + record.attempt, + "active", + ); + } catch (error) { + input.store.updateWorktreeState( + record.workflowId, + record.nodeKey, + record.attempt, + "cleanup_failed", + error instanceof Error ? error.message.slice(0, 4_096) : String(error).slice(0, 4_096), + ); + throw error; + } +} + +export function preserveWorkflowWorktree( + store: WorkflowStore, + identity: WorkflowAttemptIdentity, +): WorkflowWorktreeRecord | undefined { + const record = store.getWorktree(identity.workflowId, identity.nodeKey, identity.attempt); + if (!record || record.state === "removed") return record; + return store.updateWorktreeState(identity.workflowId, identity.nodeKey, identity.attempt, "preserved"); +} + +export async function cleanupWorkflowWorktree(input: { + store: WorkflowStore; + identity: WorkflowAttemptIdentity; + worktreeRoot: string; +}): Promise { + const record = input.store.getWorktree( + input.identity.workflowId, + input.identity.nodeKey, + input.identity.attempt, + ); + if (!record || record.state === "removed") return record; + + try { + const worktreeRoot = await canonicalDirectory(input.worktreeRoot, "workflow worktree root"); + assertContained(record.path, worktreeRoot); + const canonicalWorktreePath = await canonicalDirectory(record.path, "workflow worktree path"); + assertContained(canonicalWorktreePath, worktreeRoot); + const registered = await registeredWorktrees(record.sourceRoot); + const match = registered.find((entry) => resolve(entry.path) === resolve(record.path)); + if (!match || match.head !== record.baseSha) { + throw new WorkflowValidationError("Workflow worktree Git registration does not match persisted ownership"); + } + await git(["worktree", "remove", "--force", record.path], record.sourceRoot); + return input.store.updateWorktreeState( + record.workflowId, + record.nodeKey, + record.attempt, + "removed", + ); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + input.store.updateWorktreeState( + record.workflowId, + record.nodeKey, + record.attempt, + "cleanup_failed", + message.slice(0, 4_096), + ); + throw error; + } +} + +async function registeredWorktrees(sourceRoot: string): Promise> { + const output = await git(["worktree", "list", "--porcelain"], sourceRoot); + const records: Array<{ path: string; head: string }> = []; + let path: string | undefined; + let head: string | undefined; + for (const line of output.split("\n")) { + if (line.startsWith("worktree ")) path = line.slice("worktree ".length); + else if (line.startsWith("HEAD ")) head = line.slice("HEAD ".length); + else if (!line && path && head) { + records.push({ path, head }); + path = undefined; + head = undefined; + } + } + if (path && head) records.push({ path, head }); + return records; +} + +async function canonicalDirectory(path: string, label: string): Promise { + try { + return await realpath(path); + } catch { + throw new WorkflowValidationError(`Invalid ${label}: ${path}`); + } +} + +function assertContained(path: string, root: string): void { + const relationship = relative(root, resolve(path)); + if (relationship === "" || relationship === ".." || relationship.startsWith("../") || relationship.startsWith("..\\")) { + throw new WorkflowValidationError("Workflow worktree path escapes the managed root"); + } +} + +function safeSegment(value: string): string { + const segment = value.replace(/[^a-zA-Z0-9._-]+/g, "-").replace(/^-+|-+$/g, "").slice(0, 72); + if (!segment) throw new WorkflowValidationError("Workflow worktree identity is invalid"); + return segment; +} + +async function git(args: string[], cwd: string): Promise { + try { + const { stdout } = await execFileAsync("git", args, { cwd, maxBuffer: 10 * 1024 * 1024 }); + return stdout; + } catch (error) { + const stderr = typeof error === "object" && error && "stderr" in error + ? String((error as { stderr?: unknown }).stderr ?? "").trim() + : ""; + throw new WorkflowValidationError(stderr || (error instanceof Error ? error.message : String(error))); + } +} From 8df9d4d8cf773ea613fe0daa65eb46c34235f7e1 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 02:13:33 +0530 Subject: [PATCH 02/10] feat(mcp): expose durable workflow tools Co-Authored-By: Claude --- src/server.ts | 15 ++ src/workflows/mcp.ts | 318 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 333 insertions(+) create mode 100644 src/workflows/mcp.ts diff --git a/src/server.ts b/src/server.ts index 7dd9629e..c11e6f39 100644 --- a/src/server.ts +++ b/src/server.ts @@ -47,6 +47,8 @@ import { formatPathForPrompt } from "./skills.js"; import { createWorkspaceStore } from "./workspace-store.js"; import { formatAgentsPath, WorkspaceRegistry } from "./workspaces.js"; import { summarizeLocalAgentProfile } from "./local-agent-profiles.js"; +import { WorkflowOrchestrator } from "./workflows/orchestrator.js"; +import { registerWorkflowTools } from "./workflows/mcp.js"; import { formatLocalAgentProviderAvailabilitySummary, getLocalAgentProviderAvailabilitySnapshot, @@ -688,6 +690,7 @@ function createMcpServer( reviewCheckpoints: ReturnType, processSessions: ProcessSessionManager, localAgentProviders: LocalAgentProviderAvailability[], + workflowOrchestrator: WorkflowOrchestrator, ): McpServer { const server = new McpServer( { @@ -1594,6 +1597,15 @@ function createMcpServer( registerCodexProcessTools(server, config, workspaces, processSessions); } + if (config.subagents) { + registerWorkflowTools({ + server, + config, + workspaces, + orchestrator: workflowOrchestrator, + }); + } + return server; } @@ -1618,6 +1630,7 @@ export function createServer(config = loadConfig()): RunningServer { const workspaces = new WorkspaceRegistry(config, workspaceStore); const reviewCheckpoints = createReviewCheckpointManager(); const processSessions = new ProcessSessionManager(); + const workflowOrchestrator = new WorkflowOrchestrator(config.stateDir); const localAgentProviders = config.subagents ? getLocalAgentProviderAvailabilitySnapshot() : []; @@ -1781,6 +1794,7 @@ export function createServer(config = loadConfig()): RunningServer { reviewCheckpoints, processSessions, localAgentProviders, + workflowOrchestrator, ); await server.connect(transport); } else { @@ -1811,6 +1825,7 @@ export function createServer(config = loadConfig()): RunningServer { const results = await transports.closeAll(); logSessionCloseResults("server_shutdown", results); processSessions.shutdown(); + workflowOrchestrator.close(); oauthProvider.close(); workspaceStore.close?.(); })(); diff --git a/src/workflows/mcp.ts b/src/workflows/mcp.ts new file mode 100644 index 00000000..eede6470 --- /dev/null +++ b/src/workflows/mcp.ts @@ -0,0 +1,318 @@ +import type { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import * as z from "zod/v4"; +import type { ServerConfig } from "../config.js"; +import { loadLocalAgentProfiles } from "../local-agent-profiles.js"; +import type { WorkspaceRegistry } from "../workspaces.js"; +import { WorkflowOrchestrator } from "./orchestrator.js"; +import { createWorkflowSubmission } from "./submission.js"; +import { ensureSupervisor } from "./supervisor-launch.js"; +import type { WorkflowEvent, WorkflowRunRecord, WorkflowWorkspaceScope } from "./types.js"; + +const CONTRACT_VERSION = 1 as const; +const MAX_WAIT_MS = 300_000; +const MAX_EVENT_LIMIT = 1_000; + +const retrySchema = z.object({ + maxAttempts: z.number().int().min(1).max(10).optional(), + retryOn: z.array(z.enum(["provider_failed", "timed_out"])).max(2).optional(), + backoffMs: z.number().int().min(0).max(60_000).optional(), +}).strict(); + +const agentNodeSchema = z.object({ + key: z.string().min(1).max(64), + target: z.string().min(1).max(128), + prompt: z.string().min(1).max(200_000), + model: z.string().min(1).max(256).optional(), + thinking: z.string().min(1).max(64).optional(), + access: z.enum(["read_only", "workspace_write"]).optional(), + timeoutMs: z.number().int().min(1).max(86_400_000).optional(), + retry: retrySchema.optional(), +}).strict(); + +const dagSchema = z.object({ + version: z.literal(1), + nodes: z.array(agentNodeSchema).min(1).max(64), + edges: z.array(z.object({ + from: z.string().min(1).max(64), + to: z.string().min(1).max(64), + }).strict()).max(256).optional(), + maxConcurrency: z.number().int().min(1).max(16).optional(), + access: z.enum(["read_only", "workspace_write"]).optional(), +}).strict(); + +const nodeOutputSchema = z.object({ + key: z.string(), + status: z.enum(["pending", "ready", "running", "succeeded", "failed", "cancelled", "skipped"]), + attempt: z.number().int(), + completedAt: z.string().optional(), + errorCode: z.string().optional(), +}); + +const workflowOutputSchema = z.object({ + id: z.string(), + status: z.enum(["queued", "running", "cancelling", "succeeded", "failed", "cancelled"]), + maxConcurrency: z.number().int(), + createdAt: z.string(), + startedAt: z.string().optional(), + completedAt: z.string().optional(), + cancellationRequestedAt: z.string().optional(), + errorCode: z.string().optional(), + nodes: z.array(nodeOutputSchema), +}); + +const eventOutputSchema = z.object({ + sequence: z.number().int(), + type: z.string(), + nodeKey: z.string().optional(), + createdAt: z.string(), +}); + +const commonOutput = { + version: z.literal(CONTRACT_VERSION), + result: z.string(), +}; + +export function registerWorkflowTools(input: { + server: McpServer; + config: ServerConfig; + workspaces: WorkspaceRegistry; + orchestrator: WorkflowOrchestrator; +}): void { + const { server, config, workspaces, orchestrator } = input; + + server.registerTool( + "workflow_run", + { + title: "Run workflow", + description: "Submit a durable local-agent workflow. Provide either target/prompt for one agent or a versioned DAG.", + inputSchema: { + workspaceId: z.string().min(1), + target: z.string().min(1).max(128).optional(), + prompt: z.string().min(1).max(200_000).optional(), + model: z.string().min(1).max(256).optional(), + thinking: z.string().min(1).max(64).optional(), + access: z.enum(["read_only", "workspace_write"]).optional(), + timeoutMs: z.number().int().min(1).max(86_400_000).optional(), + retry: retrySchema.optional(), + dag: dagSchema.optional(), + idempotencyKey: z.string().min(1).max(256).optional(), + }, + outputSchema: { + ...commonOutput, + created: z.boolean(), + workflow: workflowOutputSchema, + }, + annotations: { + readOnlyHint: false, + destructiveHint: true, + idempotentHint: true, + openWorldHint: false, + }, + }, + async (args) => { + const hasSingleFields = [ + args.target, + args.prompt, + args.model, + args.thinking, + args.access, + args.timeoutMs, + args.retry, + ].some((value) => value !== undefined); + if (args.dag && hasSingleFields) { + throw new Error("Provide either dag or single-agent fields, not both"); + } + if (!args.dag && (!args.target || !args.prompt)) { + throw new Error("Single-agent workflows require target and prompt"); + } + const scope = workspaceScope(workspaces, args.workspaceId); + const workspace = workspaces.getWorkspace(args.workspaceId); + const profiles = workspace.agentProfiles.length > 0 + ? workspace.agentProfiles + : await loadLocalAgentProfiles(config, workspace.root); + const request = await createWorkflowSubmission({ + intent: { + single: args.dag ? undefined : { + target: args.target ?? "", + prompt: args.prompt ?? "", + model: args.model, + thinking: args.thinking, + access: args.access, + timeoutMs: args.timeoutMs, + retry: args.retry, + }, + dag: args.dag, + idempotencyKey: args.idempotencyKey, + }, + workspace: scope, + profiles, + worktreeRoot: config.worktreeRoot, + }); + const submitted = orchestrator.submitDetailed(request); + await ensureSupervisor({ stateDir: config.stateDir }); + const workflow = publicWorkflow(submitted.workflow); + const result = submitted.created + ? `Workflow ${workflow.id} was submitted with status ${workflow.status}.` + : `Workflow ${workflow.id} already exists for this idempotency key.`; + return response({ version: CONTRACT_VERSION, result, created: submitted.created, workflow }); + }, + ); + + server.registerTool( + "workflow_status", + { + title: "Get workflow status", + description: "Read the current durable workflow and node states for a workspace.", + inputSchema: { workspaceId: z.string().min(1), workflowId: z.string().min(1) }, + outputSchema: { ...commonOutput, workflow: workflowOutputSchema }, + annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, openWorldHint: false }, + }, + async ({ workspaceId, workflowId }) => { + const workflow = publicWorkflow(requireOwned(orchestrator, workflowId, workspaceScope(workspaces, workspaceId))); + const result = `Workflow ${workflow.id} is ${workflow.status}.`; + return response({ version: CONTRACT_VERSION, result, workflow }); + }, + ); + + server.registerTool( + "workflow_wait", + { + title: "Wait for workflow", + description: "Wait for a bounded interval, then return the latest durable workflow state.", + inputSchema: { + workspaceId: z.string().min(1), + workflowId: z.string().min(1), + timeoutMs: z.number().int().min(0).max(MAX_WAIT_MS).optional(), + }, + outputSchema: { ...commonOutput, timedOut: z.boolean(), workflow: workflowOutputSchema }, + annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, openWorldHint: false }, + }, + async ({ workspaceId, workflowId, timeoutMs }) => { + const scope = workspaceScope(workspaces, workspaceId); + const current = requireOwned(orchestrator, workflowId, scope); + const waited = await orchestrator.waitForWorkspace(workflowId, scope, { timeoutMs }); + const workflow = publicWorkflow(waited); + const timedOut = !isTerminal(waited) && !isTerminal(current); + const result = timedOut + ? `Workflow ${workflow.id} is still ${workflow.status} after the bounded wait.` + : `Workflow ${workflow.id} reached ${workflow.status}.`; + return response({ version: CONTRACT_VERSION, result, timedOut, workflow }); + }, + ); + + server.registerTool( + "workflow_events", + { + title: "Read workflow events", + description: "Read an ordered page of redacted durable workflow lifecycle events.", + inputSchema: { + workspaceId: z.string().min(1), + workflowId: z.string().min(1), + after: z.number().int().min(0).optional(), + limit: z.number().int().min(1).max(MAX_EVENT_LIMIT).optional(), + }, + outputSchema: { + ...commonOutput, + workflowId: z.string(), + events: z.array(eventOutputSchema), + cursor: z.number().int(), + }, + annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, openWorldHint: false }, + }, + async ({ workspaceId, workflowId, after, limit }) => { + const scope = workspaceScope(workspaces, workspaceId); + const workflow = requireOwned(orchestrator, workflowId, scope); + const page = orchestrator.eventsForWorkspace(workflowId, scope, { after, limit }); + const events = page.events.map((event) => publicEvent(event, workflow)); + const result = `Read ${events.length} event${events.length === 1 ? "" : "s"} for workflow ${workflowId}.`; + return response({ + version: CONTRACT_VERSION, + result, + workflowId, + events, + cursor: page.nextCursor, + }); + }, + ); + + server.registerTool( + "workflow_cancel", + { + title: "Cancel workflow", + description: "Request durable cancellation of a workflow and its active local agents.", + inputSchema: { workspaceId: z.string().min(1), workflowId: z.string().min(1) }, + outputSchema: { ...commonOutput, workflow: workflowOutputSchema }, + annotations: { readOnlyHint: false, destructiveHint: true, idempotentHint: true, openWorldHint: false }, + }, + async ({ workspaceId, workflowId }) => { + const scope = workspaceScope(workspaces, workspaceId); + requireOwned(orchestrator, workflowId, scope); + const cancelled = orchestrator.cancelForWorkspace(workflowId, scope); + await ensureSupervisor({ stateDir: config.stateDir }); + const workflow = publicWorkflow(cancelled); + const result = `Cancellation was requested for workflow ${workflow.id}; current status is ${workflow.status}.`; + return response({ version: CONTRACT_VERSION, result, workflow }); + }, + ); +} + +function workspaceScope(workspaces: WorkspaceRegistry, workspaceId: string): WorkflowWorkspaceScope { + const workspace = workspaces.getWorkspace(workspaceId); + return { workspaceId: workspace.id, workspaceRoot: workspace.root }; +} + +function requireOwned( + orchestrator: WorkflowOrchestrator, + workflowId: string, + scope: WorkflowWorkspaceScope, +): WorkflowRunRecord { + const workflow = orchestrator.getForWorkspace(workflowId, scope); + if (!workflow) throw new Error(`Unknown workflow: ${workflowId}`); + return workflow; +} + +function publicWorkflow(workflow: WorkflowRunRecord) { + return { + id: workflow.id, + status: workflow.status, + maxConcurrency: workflow.maxConcurrency, + createdAt: workflow.createdAt, + startedAt: workflow.startedAt, + completedAt: workflow.completedAt, + cancellationRequestedAt: workflow.cancellationRequestedAt, + errorCode: stringField(workflow.error, "code"), + nodes: workflow.nodes.map((node) => ({ + key: node.key, + status: node.status, + attempt: node.attempt, + completedAt: node.completedAt, + errorCode: stringField(node.error, "code"), + })), + }; +} + +function publicEvent(event: WorkflowEvent, workflow: WorkflowRunRecord) { + const node = event.nodeId ? workflow.nodes.find((candidate) => candidate.id === event.nodeId) : undefined; + return { + sequence: event.sequence, + type: event.type, + nodeKey: node?.key, + createdAt: event.createdAt, + }; +} + +function stringField(value: Record | undefined, key: string): string | undefined { + const field = value?.[key]; + return typeof field === "string" ? field : undefined; +} + +function response(structuredContent: T) { + return { + content: [{ type: "text" as const, text: structuredContent.result }], + structuredContent, + }; +} + +function isTerminal(workflow: WorkflowRunRecord): boolean { + return workflow.status === "succeeded" || workflow.status === "failed" || workflow.status === "cancelled"; +} From bafab267cf631853647979030ac46b3055931a33 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 02:13:48 +0530 Subject: [PATCH 03/10] test(workflows): cover DAG and MCP contracts Co-Authored-By: Claude --- docs/workflow-mcp.md | 98 ++++++++++ package.json | 2 +- src/oauth-store.test.ts | 1 + src/workflows/dag-scheduler.test.ts | 284 ++++++++++++++++++++++++++++ src/workflows/mcp.test.ts | 105 ++++++++++ src/workflows/migrations.test.ts | 9 +- src/workflows/workflows.test.ts | 22 +-- 7 files changed, 506 insertions(+), 15 deletions(-) create mode 100644 docs/workflow-mcp.md create mode 100644 src/workflows/dag-scheduler.test.ts create mode 100644 src/workflows/mcp.test.ts diff --git a/docs/workflow-mcp.md b/docs/workflow-mcp.md new file mode 100644 index 00000000..f9f79798 --- /dev/null +++ b/docs/workflow-mcp.md @@ -0,0 +1,98 @@ +# Durable workflow MCP tools + +DevSpace exposes durable local-agent workflows when subagents are enabled (`DEVSPACE_SUBAGENTS=1`). Workflow execution is owned by the DevSpace process and its detached supervisor, not by an individual MCP transport. Closing or reconnecting an MCP session does not cancel submitted work. + +## Tools + +All tools require `workspaceId`. A workflow ID alone is never authorization: the workflow must belong to the same workspace ID and current canonical workspace root. + +### `workflow_run` + +Submit either one local agent or a versioned DAG. + +Single-agent example: + +```json +{ + "workspaceId": "ws_...", + "target": "reviewer", + "prompt": "Review the authentication changes", + "access": "read_only", + "timeoutMs": 120000, + "retry": { + "maxAttempts": 2, + "retryOn": ["provider_failed"], + "backoffMs": 1000 + }, + "idempotencyKey": "auth-review-v1" +} +``` + +DAG example: + +```json +{ + "workspaceId": "ws_...", + "dag": { + "version": 1, + "access": "read_only", + "maxConcurrency": 3, + "nodes": [ + { "key": "tests", "target": "qa", "prompt": "Run focused tests" }, + { "key": "security", "target": "reviewer", "prompt": "Audit security" }, + { "key": "summary", "target": "claude", "prompt": "Synthesize the findings" } + ], + "edges": [ + { "from": "tests", "to": "summary" }, + { "from": "security", "to": "summary" } + ] + } +} +``` + +The DAG is rejected before persistence when it exceeds 64 nodes or 256 edges, uses duplicate or unsafe keys, references missing endpoints, contains duplicate edges or a cycle, requests an unsupported target or access mode, or exceeds timeout/retry/concurrency bounds. + +### `workflow_status` + +Returns the current redacted run and node states. + +### `workflow_wait` + +Waits for at most 300 seconds and returns the latest state. A timeout is not cancellation and does not destroy the result; callers may wait again. + +### `workflow_events` + +Reads ordered lifecycle events using an `after` cursor and a limit of at most 1,000. MCP event projections omit prompts, roots, profile bodies, environment data, claim tokens, process IDs, leases, and provider session IDs. + +### `workflow_cancel` + +Durably requests cancellation. Repeated requests are idempotent. Ready and pending nodes are terminalized, while active provider handles receive cancellation through the supervisor. + +## Scheduling semantics + +- Ready nodes are selected fairly across runs using persisted last-dispatch order. +- Both a process-wide supervisor limit and each run's persisted `maxConcurrency` are enforced. +- A node becomes ready only after all predecessors succeed. +- Exhausted failure skips unopened dependents and cancels active siblings. +- A run succeeds only after every node succeeds or is validly skipped; failures and user cancellation converge deterministically. +- Claims and heartbeats are lease-fenced. Stale attempts cannot append events or complete a replacement attempt. + +## Retry semantics + +Retries are opt-in and default to one attempt. Only `provider_failed` and `timed_out` may be configured as retryable. Each retry creates a new durable attempt and emits `node.retry_scheduled`. + +Automatic retries are limited to `read_only` nodes. DevSpace does not retry worker-loss/unknown-outcome failures or `workspace_write` side effects automatically. + +## Worktree isolation + +Every `workspace_write` attempt receives a unique detached Git worktree at the base SHA pinned when the workflow was submitted. + +- Changes are never merged, copied, committed, rebased, cherry-picked, pushed, or applied to the source checkout automatically. +- A successful attempt is durably completed before guarded cleanup. +- Failed or cancelled attempt worktrees are preserved for diagnosis with a seven-day retention timestamp. +- Cleanup verifies managed-root containment, persisted workflow/node/attempt ownership, Git worktree registration, and the expected base SHA. A failed verification preserves the directory and records `cleanup_failed`; there is no unverified recursive-delete fallback. +- Dependent nodes do not automatically inherit predecessor filesystem changes. Handoffs must use durable results/artifacts rather than an implicit shared working tree. + +## CLI compatibility + +Existing `devspace workflows run|status|wait|events|cancel` commands use the same submission snapshots and durable store. MCP and CLI can observe the same workflow when they use the same workspace ID and root. The CLI remains the shell-compatible fallback for parent harnesses that cannot call MCP tools directly. diff --git a/package.json b/package.json index 9d0b9dd4..274635d3 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 && tsx src/workflows/supervisor.test.ts && tsx src/workflows/cli.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/dag-scheduler.test.ts && tsx src/workflows/mcp.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 b1e36d5c..59f414e2 100644 --- a/src/oauth-store.test.ts +++ b/src/oauth-store.test.ts @@ -46,6 +46,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise { { version: 3, name: "local-agent-sessions" }, { version: 4, name: "durable-workflows" }, { version: 5, name: "workflow-supervisor" }, + { version: 6, name: "workflow-dag-scheduler" }, ]); } finally { database.close(); diff --git a/src/workflows/dag-scheduler.test.ts b/src/workflows/dag-scheduler.test.ts new file mode 100644 index 00000000..b495b8f0 --- /dev/null +++ b/src/workflows/dag-scheduler.test.ts @@ -0,0 +1,284 @@ +import assert from "node:assert/strict"; +import { execFileSync } from "node:child_process"; +import { mkdirSync, mkdtempSync, readFileSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { LocalAgentRunController, type LocalAgentRunHandle } from "../local-agent-runtime.js"; +import { runWorkflowSupervisor } from "./supervisor.js"; +import { WorkflowStore } from "./store.js"; +import { allocateWorkflowWorktree, cleanupWorkflowWorktree } from "./worktrees.js"; +import type { JsonObject, WorkflowDefinitionV1 } from "./types.js"; + +const root = mkdtempSync(join(tmpdir(), "devspace-workflow-dag-test-")); +try { + await testParallelBoundedScheduling(join(root, "parallel")); + await testFairSchedulingAcrossRuns(join(root, "fairness")); + await testExplicitReadOnlyRetry(join(root, "retry")); + await testManagedWriteWorktree(join(root, "worktree")); + await testWorktreeCleanupRejectsRootSubstitution(join(root, "cleanup-guard")); +} finally { + rmSync(root, { recursive: true, force: true }); +} + +async function testParallelBoundedScheduling(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + const workflow = store.submit({ + definition: definition(["a", "b", "c"], { access: "read_only" }), + policy: { version: 1, maxConcurrency: 2 }, + }).workflow; + store.close(); + + let active = 0; + let maximum = 0; + const started: string[] = []; + const handleFactory = async (_provider: string, input: { prompt: string }): Promise => { + active += 1; + maximum = Math.max(maximum, active); + started.push(input.prompt); + const controller = new LocalAgentRunController("fake"); + const timer = setTimeout(() => { + active -= 1; + controller.succeed({ provider: "fake", providerSessionId: `session-${input.prompt}`, finalResponse: input.prompt, items: [] }); + }, 40); + controller.setLifecycle({ + cancel: () => { + clearTimeout(timer); + active = Math.max(0, active - 1); + }, + dispose: () => clearTimeout(timer), + }); + return controller; + }; + + assert.equal(await runWorkflowSupervisor(stateDir, { + handleFactory, + globalConcurrency: 2, + heartbeatMs: 10, + nodeLeaseMs: 500, + supervisorLeaseMs: 500, + idleMs: 0, + }), true); + assert.equal(maximum, 2); + assert.deepEqual(started.sort(), ["a", "b", "c"]); + const reader = new WorkflowStore(stateDir); + try { + assert.equal(reader.require(workflow.id).status, "succeeded"); + } finally { + reader.close(); + } +} + +async function testFairSchedulingAcrossRuns(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + const first = store.submit({ + definition: definition(["first-a", "first-b"], { access: "read_only" }), + policy: { version: 1, maxConcurrency: 2 }, + }).workflow; + const second = store.submit({ + definition: definition(["second-a", "second-b"], { access: "read_only" }), + policy: { version: 1, maxConcurrency: 2 }, + }).workflow; + store.close(); + + const started: string[] = []; + const handleFactory = async (_provider: string, input: { prompt: string }): Promise => { + started.push(input.prompt); + const controller = new LocalAgentRunController("fake"); + const timer = setTimeout(() => { + controller.succeed({ + provider: "fake", + providerSessionId: `fair-${input.prompt}`, + finalResponse: input.prompt, + items: [], + }); + }, 40); + controller.setLifecycle({ + cancel: () => clearTimeout(timer), + dispose: () => clearTimeout(timer), + }); + return controller; + }; + + await runWorkflowSupervisor(stateDir, { + handleFactory, + globalConcurrency: 2, + heartbeatMs: 10, + nodeLeaseMs: 500, + supervisorLeaseMs: 500, + idleMs: 0, + }); + assert.equal(started.length, 4); + assert.equal(started.slice(0, 2).filter((prompt) => prompt.startsWith("first-")).length, 1); + assert.equal(started.slice(0, 2).filter((prompt) => prompt.startsWith("second-")).length, 1); + const reader = new WorkflowStore(stateDir); + try { + assert.equal(reader.require(first.id).status, "succeeded"); + assert.equal(reader.require(second.id).status, "succeeded"); + } finally { + reader.close(); + } +} + +async function testExplicitReadOnlyRetry(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + const workflow = store.submit({ + definition: definition(["retry"], { + access: "read_only", + retry: { maxAttempts: 2, retryOn: ["provider_failed"], backoffMs: 0 }, + }), + policy: { version: 1, maxConcurrency: 1 }, + }).workflow; + store.close(); + let starts = 0; + const handleFactory = async (): Promise => { + starts += 1; + if (starts === 1) throw new Error("first attempt fails"); + const controller = new LocalAgentRunController("fake"); + queueMicrotask(() => controller.succeed({ provider: "fake", providerSessionId: "retry-session", finalResponse: "ok", items: [] })); + return controller; + }; + + await runWorkflowSupervisor(stateDir, { + handleFactory, + globalConcurrency: 1, + heartbeatMs: 10, + nodeLeaseMs: 500, + supervisorLeaseMs: 500, + idleMs: 0, + }); + const reader = new WorkflowStore(stateDir); + try { + const completed = reader.require(workflow.id); + assert.equal(completed.status, "succeeded"); + assert.equal(completed.nodes[0]!.attempt, 2); + assert.ok(reader.readEvents(workflow.id, { limit: 100 }).events.some((event) => event.type === "node.retry_scheduled")); + } finally { + reader.close(); + } +} + +async function testManagedWriteWorktree(stateDir: string): Promise { + const repo = join(stateDir, "repo"); + const worktreeRoot = join(stateDir, "managed"); + execFileSync("mkdir", ["-p", repo]); + execFileSync("git", ["init", "-q"], { cwd: repo }); + execFileSync("git", ["config", "user.email", "test@example.com"], { cwd: repo }); + execFileSync("git", ["config", "user.name", "Test"], { cwd: repo }); + writeFileSync(join(repo, "source.txt"), "source\n"); + execFileSync("git", ["add", "source.txt"], { cwd: repo }); + execFileSync("git", ["commit", "-qm", "initial"], { cwd: repo }); + const baseSha = execFileSync("git", ["rev-parse", "HEAD"], { cwd: repo, encoding: "utf8" }).trim(); + + const store = new WorkflowStore(stateDir); + const workflow = store.submit({ + definition: definition(["writer"], { + access: "workspace_write", + workspaceRoot: repo, + worktreeRoot, + baseSha, + }), + policy: { version: 1, maxConcurrency: 1 }, + }).workflow; + store.close(); + let executionRoot = ""; + const handleFactory = async (_provider: string, input: { workspace: string }): Promise => { + executionRoot = input.workspace; + writeFileSync(join(input.workspace, "source.txt"), "isolated\n"); + const controller = new LocalAgentRunController("fake"); + queueMicrotask(() => controller.succeed({ provider: "fake", providerSessionId: "writer-session", finalResponse: "changed", items: [] })); + return controller; + }; + + await runWorkflowSupervisor(stateDir, { + handleFactory, + globalConcurrency: 1, + heartbeatMs: 10, + nodeLeaseMs: 1_000, + supervisorLeaseMs: 1_000, + idleMs: 0, + }); + assert.notEqual(executionRoot, repo); + assert.equal(readFileSync(join(repo, "source.txt"), "utf8"), "source\n"); + assert.equal(execFileSync("git", ["status", "--porcelain"], { cwd: repo, encoding: "utf8" }), ""); + const reader = new WorkflowStore(stateDir); + try { + assert.equal(reader.require(workflow.id).status, "succeeded"); + assert.equal(reader.getWorktree(workflow.id, "writer", 1)?.state, "removed"); + } finally { + reader.close(); + } +} + +async function testWorktreeCleanupRejectsRootSubstitution(stateDir: string): Promise { + const repo = join(stateDir, "repo"); + const worktreeRoot = join(stateDir, "managed"); + const outside = join(stateDir, "outside"); + mkdirSync(repo, { recursive: true }); + mkdirSync(outside, { recursive: true }); + execFileSync("git", ["init", "-q"], { cwd: repo }); + execFileSync("git", ["config", "user.email", "test@example.com"], { cwd: repo }); + execFileSync("git", ["config", "user.name", "Test"], { cwd: repo }); + writeFileSync(join(repo, "source.txt"), "source\n"); + writeFileSync(join(outside, "marker.txt"), "keep\n"); + execFileSync("git", ["add", "source.txt"], { cwd: repo }); + execFileSync("git", ["commit", "-qm", "initial"], { cwd: repo }); + const baseSha = execFileSync("git", ["rev-parse", "HEAD"], { cwd: repo, encoding: "utf8" }).trim(); + + const store = new WorkflowStore(stateDir); + const workflow = store.submit({ + definition: definition(["writer"], { access: "workspace_write", workspaceRoot: repo, worktreeRoot, baseSha }), + policy: { version: 1, maxConcurrency: 1 }, + }).workflow; + const identity = { + workflowId: workflow.id, + nodeKey: "writer", + attempt: 1, + claimToken: "cleanup-claim", + supervisorOwnerToken: "cleanup-owner", + supervisorOwnerEpoch: 1, + }; + const allocation = await allocateWorkflowWorktree({ + store, + identity, + sourceRoot: repo, + worktreeRoot, + baseSha, + }); + execFileSync("git", ["worktree", "remove", "--force", allocation.path], { cwd: repo }); + symlinkSync(outside, allocation.path, "dir"); + + await assert.rejects( + cleanupWorkflowWorktree({ store, identity, worktreeRoot }), + /escapes the managed root/, + ); + assert.equal(readFileSync(join(outside, "marker.txt"), "utf8"), "keep\n"); + assert.equal(store.getWorktree(workflow.id, "writer", 1)?.state, "cleanup_failed"); + store.close(); +} + +function definition(keys: string[], overrides: JsonObject): WorkflowDefinitionV1 { + return { + version: 1, + nodes: keys.map((key) => ({ + key, + type: "agent" as const, + config: { + provider: "fake", + prompt: key, + profileBody: "", + workspaceRoot: String(overrides.workspaceRoot ?? process.cwd()), + worktreeRoot: String(overrides.worktreeRoot ?? ""), + baseSha: String(overrides.baseSha ?? ""), + timeoutMs: null, + effectivePolicy: { + version: 1, + mode: "workflow", + access: String(overrides.access ?? "read_only"), + environment: {}, + }, + retry: (overrides.retry ?? { maxAttempts: 1, retryOn: [], backoffMs: 0 }) as JsonObject, + }, + })), + edges: [], + }; +} diff --git a/src/workflows/mcp.test.ts b/src/workflows/mcp.test.ts new file mode 100644 index 00000000..3a613ab1 --- /dev/null +++ b/src/workflows/mcp.test.ts @@ -0,0 +1,105 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; +import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; +import type { ServerConfig } from "../config.js"; +import { WorkspaceRegistry } from "../workspaces.js"; +import { registerWorkflowTools } from "./mcp.js"; +import { WorkflowOrchestrator } from "./orchestrator.js"; + +const root = mkdtempSync(join(tmpdir(), "devspace-workflow-mcp-test-")); +const stateDir = join(root, "state"); +const config = { + stateDir, + worktreeRoot: join(root, "worktrees"), + allowedRoots: [root], + subagents: true, + devspaceAgentsDir: join(root, "agents"), + skillsEnabled: false, + skillPaths: [], + devspaceSkillsDir: join(root, "skills"), + agentDir: join(root, "agent-dir"), +} as unknown as ServerConfig; +const workspaces = new WorkspaceRegistry(config); +const { workspace } = await workspaces.openWorkspace(root); +const orchestrator = new WorkflowOrchestrator(stateDir); +const workflow = orchestrator.submit({ + definition: { + version: 1, + nodes: [{ + key: "agent", + type: "agent", + config: { + provider: "fake", + prompt: "secret prompt", + profileBody: "secret profile", + workspaceRoot: root, + effectivePolicy: { version: 1, mode: "workflow", access: "read_only", environment: {} }, + }, + }], + edges: [], + }, + workspace: { workspaceId: workspace.id, workspaceRoot: workspace.root }, +}); + +const server = new McpServer({ name: "workflow-test", version: "1.0.0" }); +registerWorkflowTools({ server, config, workspaces, orchestrator }); +const client = new Client({ name: "workflow-test-client", version: "1.0.0" }); +const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); +try { + await Promise.all([server.connect(serverTransport), client.connect(clientTransport)]); + const listed = await client.listTools(); + assert.deepEqual( + listed.tools.map((tool) => tool.name).sort(), + ["workflow_cancel", "workflow_events", "workflow_run", "workflow_status", "workflow_wait"], + ); + const runTool = listed.tools.find((tool) => tool.name === "workflow_run")!; + assert.equal(runTool.annotations?.destructiveHint, true); + assert.equal(runTool.annotations?.idempotentHint, true); + const statusTool = listed.tools.find((tool) => tool.name === "workflow_status")!; + assert.equal(statusTool.annotations?.readOnlyHint, true); + + const ambiguous = await client.callTool({ + name: "workflow_run", + arguments: { + workspaceId: workspace.id, + target: "fake", + prompt: "single", + dag: { + version: 1, + nodes: [{ key: "dag", target: "fake", prompt: "dag" }], + }, + }, + }); + assert.equal(ambiguous.isError, true); + + const result = await client.callTool({ + name: "workflow_status", + arguments: { workspaceId: workspace.id, workflowId: workflow.id }, + }); + const structured = result.structuredContent as { + version: number; + result: string; + workflow: { id: string; status: string; nodes: Array<{ key: string }> }; + }; + assert.equal(structured.version, 1); + assert.equal(structured.workflow.id, workflow.id); + assert.equal(structured.workflow.status, "queued"); + assert.equal((result.content as Array<{ text: string }>)[0]!.text, structured.result); + const serialized = JSON.stringify(structured); + assert.doesNotMatch(serialized, /secret prompt|secret profile|workspaceRoot|claimToken|providerSession/); + + const denied = await client.callTool({ + name: "workflow_status", + arguments: { workspaceId: "unknown", workflowId: workflow.id }, + }); + assert.equal(denied.isError, true); +} finally { + await client.close(); + await server.close(); + orchestrator.close(); + rmSync(root, { recursive: true, force: true }); +} diff --git a/src/workflows/migrations.test.ts b/src/workflows/migrations.test.ts index 6ddb3304..6178d546 100644 --- a/src/workflows/migrations.test.ts +++ b/src/workflows/migrations.test.ts @@ -26,6 +26,7 @@ function testFreshSchema(stateDir: string): void { { version: 3, name: "local-agent-sessions" }, { version: 4, name: "durable-workflows" }, { version: 5, name: "workflow-supervisor" }, + { version: 6, name: "workflow-dag-scheduler" }, ], ); const tables = database.sqlite @@ -44,6 +45,7 @@ function testFreshSchema(stateDir: string): void { "workflow_provider_events", "workflow_runs", "workflow_supervisor", + "workflow_worktrees", ]); const edgeForeignKeys = database.sqlite.prepare("pragma foreign_key_list(workflow_edges)").all() as Array<{ @@ -98,7 +100,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, 5], + [1, 2, 3, 4, 5, 6], ); assert.deepEqual(migrated.sqlite.prepare("select * from workspace_sessions").all(), [ { @@ -171,7 +173,7 @@ function testExistingMigrationCompatibility(stateDir: string): void { .prepare("select count(*) from sqlite_master where type = 'table' and name like 'workflow_%'") .pluck() .get(), - 7, + 8, ); } finally { migrated.close(); @@ -222,6 +224,7 @@ function createVersion3Fixture(stateDir: string): void { const initialized = openDatabase(stateDir); try { initialized.sqlite.exec(` + drop table workflow_worktrees; drop table workflow_provider_events; drop table workflow_node_attempts; drop table workflow_supervisor; @@ -229,7 +232,7 @@ function createVersion3Fixture(stateDir: string): void { drop table workflow_events; drop table workflow_nodes; drop table workflow_runs; - delete from devspace_schema_migrations where version in (4, 5); + delete from devspace_schema_migrations where version in (4, 5, 6); 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/workflows.test.ts b/src/workflows/workflows.test.ts index e99850f6..1c4fb475 100644 --- a/src/workflows/workflows.test.ts +++ b/src/workflows/workflows.test.ts @@ -486,17 +486,17 @@ function testDagValidation(stateDir: string): void { two: "pending", }); - const nulKeys = store.submit({ - definition: { - version: 1, - nodes: [agentNode("a\0b"), agentNode("a"), agentNode("b\0c"), agentNode("c")], - edges: [ - { from: "a\0b", to: "c" }, - { from: "a", to: "b\0c" }, - ], - }, - }).workflow; - assert.equal(nulKeys.edges.length, 2); + assert.throws( + () => + store.submit({ + definition: { + version: 1, + nodes: [agentNode("a\0b")], + edges: [], + }, + }), + /node key is unsafe/, + ); } finally { store.close(); } From d8b15fa8b877f7a2a24dca420ef809a67094a48c Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 22:39:10 +0530 Subject: [PATCH 04/10] fix(workflows): harden MCP DAG lifecycle Co-Authored-By: Claude --- src/server-shutdown.test.ts | 24 ++++ src/server.ts | 9 ++ src/workflows/cli.test.ts | 2 +- src/workflows/dag-scheduler.test.ts | 177 +++++++++++++++++++++++++++- src/workflows/mcp.test.ts | 120 ++++++++++++++++++- src/workflows/mcp.ts | 125 ++++++++++++++++---- src/workflows/orchestrator.ts | 3 +- src/workflows/store.ts | 176 ++++++++++++++++++++------- src/workflows/supervisor.test.ts | 99 +++++++++++++++- src/workflows/supervisor.ts | 66 ++++++----- src/workflows/workflows.test.ts | 48 +++++++- src/workflows/worktrees.ts | 36 +++++- 12 files changed, 777 insertions(+), 108 deletions(-) diff --git a/src/server-shutdown.test.ts b/src/server-shutdown.test.ts index 7dca2eea..db68c370 100644 --- a/src/server-shutdown.test.ts +++ b/src/server-shutdown.test.ts @@ -1,5 +1,9 @@ import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { shutdownHttpServer } from "./server-shutdown.js"; +import { WorkflowOrchestrator } from "./workflows/orchestrator.js"; let finishHttpClose: (() => void) | undefined; let applicationCloseStarted = false; @@ -95,3 +99,23 @@ await assert.rejects( ), httpCloseError, ); + +const workflowState = mkdtempSync(join(tmpdir(), "devspace-shutdown-workflow-")); +try { + const orchestrator = new WorkflowOrchestrator(workflowState); + const workflow = orchestrator.submit({ + definition: { version: 1, nodes: [{ key: "agent", type: "agent" }], edges: [] }, + workspace: { workspaceId: "workspace", workspaceRoot: "/tmp/workspace" }, + }); + const waiting = orchestrator.waitForWorkspace( + workflow.id, + { workspaceId: "workspace", workspaceRoot: "/tmp/workspace" }, + { timeoutMs: 5_000, pollIntervalMs: 1_000 }, + ); + orchestrator.close(); + const closedSnapshot = await waiting; + assert.equal(closedSnapshot.id, workflow.id); + assert.equal(closedSnapshot.status, "queued"); +} finally { + rmSync(workflowState, { recursive: true, force: true }); +} diff --git a/src/server.ts b/src/server.ts index c11e6f39..c2da9091 100644 --- a/src/server.ts +++ b/src/server.ts @@ -171,6 +171,7 @@ const toolNames = { interface ToolLogFields { tool: string; workspaceId?: string; + workflowId?: string; path?: string; workingDirectory?: string; command?: string; @@ -1603,6 +1604,14 @@ function createMcpServer( config, workspaces, orchestrator: workflowOrchestrator, + audit: (event) => logToolCall(config, { + tool: event.tool, + workspaceId: event.workspaceId, + workflowId: event.workflowId, + success: event.success, + durationMs: event.durationMs, + error: event.errorCode, + }), }); } diff --git a/src/workflows/cli.test.ts b/src/workflows/cli.test.ts index ef506697..43d0e949 100644 --- a/src/workflows/cli.test.ts +++ b/src/workflows/cli.test.ts @@ -56,7 +56,7 @@ try { ], baseEnv); assert.equal(timed.status, 0, timed.stderr); const timedEnvelope = JSON.parse(timed.stdout) as Envelope; - assert.equal(timedEnvelope.timedOut, true); + assert.equal(timedEnvelope.timedOut, true, JSON.stringify(timedEnvelope)); const waited = runCli([ "workflows", "wait", workflowId, "--timeout-ms", "10000", "--after", "0", "--json", diff --git a/src/workflows/dag-scheduler.test.ts b/src/workflows/dag-scheduler.test.ts index b495b8f0..366032f1 100644 --- a/src/workflows/dag-scheduler.test.ts +++ b/src/workflows/dag-scheduler.test.ts @@ -1,20 +1,29 @@ import assert from "node:assert/strict"; import { execFileSync } from "node:child_process"; -import { mkdirSync, mkdtempSync, readFileSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, symlinkSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; +import { openDatabase } from "../db/client.js"; import { LocalAgentRunController, type LocalAgentRunHandle } from "../local-agent-runtime.js"; import { runWorkflowSupervisor } from "./supervisor.js"; import { WorkflowStore } from "./store.js"; -import { allocateWorkflowWorktree, cleanupWorkflowWorktree } from "./worktrees.js"; +import { + allocateWorkflowWorktree, + cleanupExpiredWorkflowWorktrees, + cleanupWorkflowWorktree, + preserveWorkflowWorktree, +} from "./worktrees.js"; import type { JsonObject, WorkflowDefinitionV1 } from "./types.js"; const root = mkdtempSync(join(tmpdir(), "devspace-workflow-dag-test-")); try { await testParallelBoundedScheduling(join(root, "parallel")); await testFairSchedulingAcrossRuns(join(root, "fairness")); + testFairSchedulingWithTiedTimestamps(join(root, "fairness-tied")); await testExplicitReadOnlyRetry(join(root, "retry")); + await testRetryFailureClasses(join(root, "retry-classes")); await testManagedWriteWorktree(join(root, "worktree")); + await testExpiredWorktreeCleanup(join(root, "worktree-retention")); await testWorktreeCleanupRejectsRootSubstitution(join(root, "cleanup-guard")); } finally { rmSync(root, { recursive: true, force: true }); @@ -119,6 +128,31 @@ async function testFairSchedulingAcrossRuns(stateDir: string): Promise { } } +function testFairSchedulingWithTiedTimestamps(stateDir: string): void { + const store = new WorkflowStore(stateDir); + const first = store.submit({ + definition: definition(["first-a", "first-b"], { access: "read_only" }), + policy: { version: 1, maxConcurrency: 2 }, + }).workflow; + const second = store.submit({ + definition: definition(["second-a", "second-b"], { access: "read_only" }), + policy: { version: 1, maxConcurrency: 2 }, + }).workflow; + const supervisor = store.acquireSupervisor({ ownerToken: "fair-tie", ownerPid: 1, leaseMs: 1_000 })!; + const firstClaim = store.claimNextAgentNode({ supervisor, claimToken: "first", leaseMs: 1_000 })!; + assert.equal(firstClaim.workflow.id, first.id); + + const database = openDatabase(stateDir); + database.sqlite.prepare("update workflow_runs set last_dispatched_at = ? where id in (?, ?)") + .run("2026-01-01T00:00:00.000Z", first.id, second.id); + database.close(); + + const secondClaim = store.claimNextAgentNode({ supervisor, claimToken: "second", leaseMs: 1_000 })!; + assert.equal(secondClaim.workflow.id, second.id); + store.releaseSupervisor(supervisor); + store.close(); +} + async function testExplicitReadOnlyRetry(stateDir: string): Promise { const store = new WorkflowStore(stateDir); const workflow = store.submit({ @@ -157,6 +191,104 @@ async function testExplicitReadOnlyRetry(stateDir: string): Promise { } } +async function testRetryFailureClasses(stateDir: string): Promise { + const resultStore = new WorkflowStore(stateDir); + const resultRetry = resultStore.submit({ + definition: definition(["result-retry"], { + access: "read_only", + retry: { maxAttempts: 2, retryOn: ["provider_failed"], backoffMs: 0 }, + }), + policy: { version: 1, maxConcurrency: 1 }, + }).workflow; + resultStore.close(); + let resultStarts = 0; + await runWorkflowSupervisor(stateDir, { + globalConcurrency: 1, + heartbeatMs: 5, + nodeLeaseMs: 200, + supervisorLeaseMs: 200, + idleMs: 0, + handleFactory: async () => { + resultStarts += 1; + const controller = new LocalAgentRunController("fake"); + queueMicrotask(() => { + if (resultStarts === 1) controller.fail(new Error("result rejected")); + else controller.succeed({ provider: "fake", providerSessionId: "result-retry", finalResponse: "ok", items: [] }); + }); + return controller; + }, + }); + const resultReader = new WorkflowStore(stateDir); + assert.equal(resultReader.require(resultRetry.id).status, "succeeded"); + assert.equal(resultReader.require(resultRetry.id).nodes[0]!.attempt, 2); + resultReader.close(); + + const timeoutStore = new WorkflowStore(stateDir); + const timeoutRetry = timeoutStore.submit({ + definition: definition(["timeout-retry"], { + access: "read_only", + timeoutMs: 10, + retry: { maxAttempts: 2, retryOn: ["timed_out"], backoffMs: 20 }, + }), + policy: { version: 1, maxConcurrency: 1 }, + }).workflow; + timeoutStore.close(); + const timeoutStarts: number[] = []; + await runWorkflowSupervisor(stateDir, { + globalConcurrency: 1, + heartbeatMs: 5, + nodeLeaseMs: 200, + supervisorLeaseMs: 200, + idleMs: 0, + handleFactory: async () => { + timeoutStarts.push(Date.now()); + const controller = new LocalAgentRunController("fake"); + const timer = setTimeout(() => controller.succeed({ + provider: "fake", + providerSessionId: "late-timeout", + finalResponse: "late", + items: [], + }), 30); + controller.setLifecycle({ cancel: () => undefined, dispose: () => clearTimeout(timer) }); + return controller; + }, + }); + const timeoutReader = new WorkflowStore(stateDir); + const timedOut = timeoutReader.require(timeoutRetry.id); + assert.equal(timedOut.status, "failed"); + assert.equal(timedOut.nodes[0]!.attempt, 2); + assert.equal((timedOut.error as JsonObject).code, "timed_out"); + assert.ok(timeoutStarts[1]! - timeoutStarts[0]! >= 15, "retry backoff must delay the second attempt"); + timeoutReader.close(); + + const nonRetryStore = new WorkflowStore(stateDir); + const nonRetry = nonRetryStore.submit({ + definition: definition(["non-retryable"], { + access: "read_only", + retry: { maxAttempts: 3, retryOn: ["timed_out"], backoffMs: 0 }, + }), + policy: { version: 1, maxConcurrency: 1 }, + }).workflow; + nonRetryStore.close(); + let nonRetryStarts = 0; + await runWorkflowSupervisor(stateDir, { + globalConcurrency: 1, + heartbeatMs: 5, + nodeLeaseMs: 200, + supervisorLeaseMs: 200, + idleMs: 0, + handleFactory: async () => { + nonRetryStarts += 1; + throw new Error("not retryable by policy"); + }, + }); + const nonRetryReader = new WorkflowStore(stateDir); + assert.equal(nonRetryReader.require(nonRetry.id).status, "failed"); + assert.equal(nonRetryReader.require(nonRetry.id).nodes[0]!.attempt, 1); + assert.equal(nonRetryStarts, 1); + nonRetryReader.close(); +} + async function testManagedWriteWorktree(stateDir: string): Promise { const repo = join(stateDir, "repo"); const worktreeRoot = join(stateDir, "managed"); @@ -184,6 +316,8 @@ async function testManagedWriteWorktree(stateDir: string): Promise { const handleFactory = async (_provider: string, input: { workspace: string }): Promise => { executionRoot = input.workspace; writeFileSync(join(input.workspace, "source.txt"), "isolated\n"); + execFileSync("git", ["add", "source.txt"], { cwd: input.workspace }); + execFileSync("git", ["commit", "-qm", "agent change"], { cwd: input.workspace }); const controller = new LocalAgentRunController("fake"); queueMicrotask(() => controller.succeed({ provider: "fake", providerSessionId: "writer-session", finalResponse: "changed", items: [] })); return controller; @@ -209,6 +343,43 @@ async function testManagedWriteWorktree(stateDir: string): Promise { } } +async function testExpiredWorktreeCleanup(stateDir: string): Promise { + const repo = join(stateDir, "repo"); + const worktreeRoot = join(stateDir, "managed"); + mkdirSync(repo, { recursive: true }); + execFileSync("git", ["init", "-q"], { cwd: repo }); + execFileSync("git", ["config", "user.email", "test@example.com"], { cwd: repo }); + execFileSync("git", ["config", "user.name", "Test"], { cwd: repo }); + writeFileSync(join(repo, "source.txt"), "source\n"); + execFileSync("git", ["add", "source.txt"], { cwd: repo }); + execFileSync("git", ["commit", "-qm", "initial"], { cwd: repo }); + const baseSha = execFileSync("git", ["rev-parse", "HEAD"], { cwd: repo, encoding: "utf8" }).trim(); + + const store = new WorkflowStore(stateDir); + const workflow = store.submit({ + definition: definition(["writer"], { access: "workspace_write", workspaceRoot: repo, worktreeRoot, baseSha }), + policy: { version: 1, maxConcurrency: 1 }, + }).workflow; + const identity = { + workflowId: workflow.id, + nodeKey: "writer", + attempt: 1, + claimToken: "retention-claim", + }; + const allocation = await allocateWorkflowWorktree({ store, identity, sourceRoot: repo, worktreeRoot, baseSha }); + preserveWorkflowWorktree(store, identity); + const database = openDatabase(stateDir); + database.sqlite.prepare( + "update workflow_worktrees set retain_until = ? where workflow_run_id = ? and node_key = ? and attempt = ?", + ).run("2020-01-01T00:00:00.000Z", workflow.id, "writer", 1); + database.close(); + + assert.equal(await cleanupExpiredWorkflowWorktrees({ store }), 1); + assert.equal(store.getWorktree(workflow.id, "writer", 1)?.state, "removed"); + assert.equal(existsSync(allocation.path), false); + store.close(); +} + async function testWorktreeCleanupRejectsRootSubstitution(stateDir: string): Promise { const repo = join(stateDir, "repo"); const worktreeRoot = join(stateDir, "managed"); @@ -269,7 +440,7 @@ function definition(keys: string[], overrides: JsonObject): WorkflowDefinitionV1 workspaceRoot: String(overrides.workspaceRoot ?? process.cwd()), worktreeRoot: String(overrides.worktreeRoot ?? ""), baseSha: String(overrides.baseSha ?? ""), - timeoutMs: null, + timeoutMs: overrides.timeoutMs ?? null, effectivePolicy: { version: 1, mode: "workflow", diff --git a/src/workflows/mcp.test.ts b/src/workflows/mcp.test.ts index 3a613ab1..dcdbdbe1 100644 --- a/src/workflows/mcp.test.ts +++ b/src/workflows/mcp.test.ts @@ -7,10 +7,14 @@ import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import type { ServerConfig } from "../config.js"; import { WorkspaceRegistry } from "../workspaces.js"; -import { registerWorkflowTools } from "./mcp.js"; +import { registerWorkflowTools, type WorkflowToolAuditEvent } from "./mcp.js"; import { WorkflowOrchestrator } from "./orchestrator.js"; +import { WorkflowStore } from "./store.js"; +import { runWorkflowSupervisor } from "./supervisor.js"; const root = mkdtempSync(join(tmpdir(), "devspace-workflow-mcp-test-")); +const originalFakeProvider = process.env.DEVSPACE_WORKFLOW_FAKE_PROVIDER; +process.env.DEVSPACE_WORKFLOW_FAKE_PROVIDER = "1"; const stateDir = join(root, "state"); const config = { stateDir, @@ -45,8 +49,40 @@ const workflow = orchestrator.submit({ workspace: { workspaceId: workspace.id, workspaceRoot: workspace.root }, }); +const completionStore = new WorkflowStore(stateDir); +const supervisor = completionStore.acquireSupervisor({ ownerToken: "mcp-test", ownerPid: 1, leaseMs: 5_000 })!; +const claim = completionStore.claimNextAgentNode({ supervisor, claimToken: "mcp-claim", leaseMs: 5_000 })!; +completionStore.completeAgentNode({ + workflowId: workflow.id, + nodeKey: "agent", + attempt: claim.node.attempt, + claimToken: claim.node.claimToken!, + status: "succeeded", + result: { provider: "fake", providerSessionId: "secret-session", finalResponse: "safe final response" }, +}); +completionStore.releaseSupervisor(supervisor); +completionStore.close(); +const audits: WorkflowToolAuditEvent[] = []; +let failSupervisorLaunch = false; const server = new McpServer({ name: "workflow-test", version: "1.0.0" }); -registerWorkflowTools({ server, config, workspaces, orchestrator }); +registerWorkflowTools({ + server, + config, + workspaces, + orchestrator, + launchSupervisor: async () => { + if (failSupervisorLaunch) throw new Error("startup details must stay private"); + await runWorkflowSupervisor(stateDir, { + globalConcurrency: 2, + heartbeatMs: 5, + nodeLeaseMs: 500, + supervisorLeaseMs: 500, + idleMs: 0, + }); + return { requestedWakeGeneration: 1, spawned: true }; + }, + audit: (event) => audits.push(event), +}); const client = new Client({ name: "workflow-test-client", version: "1.0.0" }); const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); try { @@ -58,7 +94,7 @@ try { ); const runTool = listed.tools.find((tool) => tool.name === "workflow_run")!; assert.equal(runTool.annotations?.destructiveHint, true); - assert.equal(runTool.annotations?.idempotentHint, true); + assert.equal(runTool.annotations?.idempotentHint, false); const statusTool = listed.tools.find((tool) => tool.name === "workflow_status")!; assert.equal(statusTool.annotations?.readOnlyHint, true); @@ -76,6 +112,43 @@ try { }); assert.equal(ambiguous.isError, true); + const dagRun = await client.callTool({ + name: "workflow_run", + arguments: { + workspaceId: workspace.id, + dag: { + version: 1, + maxConcurrency: 2, + nodes: [ + { key: "first", target: "fake", prompt: "first deterministic task" }, + { key: "second", target: "fake", prompt: "second deterministic task" }, + ], + edges: [{ from: "first", to: "second" }], + }, + idempotencyKey: "mcp-dag-e2e", + }, + }); + assert.equal(dagRun.isError, undefined); + const dagWorkflowId = (dagRun.structuredContent as { workflow: { id: string } }).workflow.id; + const dagWait = await client.callTool({ + name: "workflow_wait", + arguments: { workspaceId: workspace.id, workflowId: dagWorkflowId, timeoutMs: 1_000 }, + }); + const dagResult = dagWait.structuredContent as { + timedOut: boolean; + workflow: { status: string; finalResponse?: string; nodes: Array<{ status: string; finalResponse?: string }> }; + }; + assert.equal(dagResult.timedOut, false); + assert.equal(dagResult.workflow.status, "succeeded"); + assert.deepEqual(dagResult.workflow.nodes.map((node) => node.status), ["succeeded", "succeeded"]); + assert.equal(dagResult.workflow.finalResponse, "fake result"); + const dagEvents = await client.callTool({ + name: "workflow_events", + arguments: { workspaceId: workspace.id, workflowId: dagWorkflowId, after: 0, limit: 100 }, + }); + assert.equal(dagEvents.isError, undefined); + assert.ok(((dagEvents.structuredContent as { events: unknown[] }).events).length > 0); + const result = await client.callTool({ name: "workflow_status", arguments: { workspaceId: workspace.id, workflowId: workflow.id }, @@ -83,11 +156,18 @@ try { const structured = result.structuredContent as { version: number; result: string; - workflow: { id: string; status: string; nodes: Array<{ key: string }> }; + workflow: { + id: string; + status: string; + finalResponse?: string; + nodes: Array<{ key: string; finalResponse?: string }>; + }; }; assert.equal(structured.version, 1); assert.equal(structured.workflow.id, workflow.id); - assert.equal(structured.workflow.status, "queued"); + assert.equal(structured.workflow.status, "succeeded"); + assert.equal(structured.workflow.finalResponse, "safe final response"); + assert.equal(structured.workflow.nodes[0]!.finalResponse, "safe final response"); assert.equal((result.content as Array<{ text: string }>)[0]!.text, structured.result); const serialized = JSON.stringify(structured); assert.doesNotMatch(serialized, /secret prompt|secret profile|workspaceRoot|claimToken|providerSession/); @@ -97,9 +177,39 @@ try { arguments: { workspaceId: "unknown", workflowId: workflow.id }, }); assert.equal(denied.isError, true); + + const cancellable = orchestrator.submit({ + definition: { version: 1, nodes: [{ key: "agent", type: "agent" }], edges: [] }, + workspace: { workspaceId: workspace.id, workspaceRoot: workspace.root }, + }); + failSupervisorLaunch = true; + const cancellation = await client.callTool({ + name: "workflow_cancel", + arguments: { workspaceId: workspace.id, workflowId: cancellable.id }, + }); + assert.equal(cancellation.isError, undefined); + const cancellationResult = cancellation.structuredContent as { + supervisor: { started: boolean; errorCode?: string }; + workflow: { id: string; status: string }; + }; + assert.deepEqual(cancellationResult.supervisor, { + started: false, + errorCode: "supervisor_start_failed", + }); + assert.equal(cancellationResult.workflow.id, cancellable.id); + assert.equal(orchestrator.get(cancellable.id)?.status, "cancelling"); + assert.ok(audits.some((event) => event.tool === "workflow_wait" && event.success)); + assert.ok(audits.some((event) => event.tool === "workflow_events" && event.success)); + assert.ok(audits.some((event) => event.tool === "workflow_status" && event.success)); + assert.ok(audits.some((event) => event.tool === "workflow_status" && !event.success)); + assert.ok(audits.some((event) => event.tool === "workflow_run" && !event.success)); + assert.ok(audits.some((event) => event.tool === "workflow_cancel" && event.success)); + assert.doesNotMatch(JSON.stringify(audits), /secret prompt|secret profile|startup details|secret-session/); } finally { await client.close(); await server.close(); orchestrator.close(); + if (originalFakeProvider === undefined) delete process.env.DEVSPACE_WORKFLOW_FAKE_PROVIDER; + else process.env.DEVSPACE_WORKFLOW_FAKE_PROVIDER = originalFakeProvider; rmSync(root, { recursive: true, force: true }); } diff --git a/src/workflows/mcp.ts b/src/workflows/mcp.ts index eede6470..472e4c72 100644 --- a/src/workflows/mcp.ts +++ b/src/workflows/mcp.ts @@ -46,6 +46,7 @@ const nodeOutputSchema = z.object({ attempt: z.number().int(), completedAt: z.string().optional(), errorCode: z.string().optional(), + finalResponse: z.string().optional(), }); const workflowOutputSchema = z.object({ @@ -57,9 +58,15 @@ const workflowOutputSchema = z.object({ completedAt: z.string().optional(), cancellationRequestedAt: z.string().optional(), errorCode: z.string().optional(), + finalResponse: z.string().optional(), nodes: z.array(nodeOutputSchema), }); +const supervisorOutputSchema = z.object({ + started: z.boolean(), + errorCode: z.string().optional(), +}); + const eventOutputSchema = z.object({ sequence: z.number().int(), type: z.string(), @@ -72,13 +79,25 @@ const commonOutput = { result: z.string(), }; +export interface WorkflowToolAuditEvent { + tool: string; + workspaceId?: string; + workflowId?: string; + success: boolean; + durationMs: number; + errorCode?: string; +} + export function registerWorkflowTools(input: { server: McpServer; config: ServerConfig; workspaces: WorkspaceRegistry; orchestrator: WorkflowOrchestrator; + launchSupervisor?: typeof ensureSupervisor; + audit?: (event: WorkflowToolAuditEvent) => void; }): void { const { server, config, workspaces, orchestrator } = input; + const launchSupervisor = input.launchSupervisor ?? ensureSupervisor; server.registerTool( "workflow_run", @@ -100,16 +119,17 @@ export function registerWorkflowTools(input: { outputSchema: { ...commonOutput, created: z.boolean(), + supervisor: supervisorOutputSchema, workflow: workflowOutputSchema, }, annotations: { readOnlyHint: false, destructiveHint: true, - idempotentHint: true, + idempotentHint: false, openWorldHint: false, }, }, - async (args) => { + async (args) => audited(input.audit, "workflow_run", args.workspaceId, undefined, async () => { const hasSingleFields = [ args.target, args.prompt, @@ -149,13 +169,15 @@ export function registerWorkflowTools(input: { worktreeRoot: config.worktreeRoot, }); const submitted = orchestrator.submitDetailed(request); - await ensureSupervisor({ stateDir: config.stateDir }); + const supervisor = await supervisorStartState(() => launchSupervisor({ stateDir: config.stateDir })); const workflow = publicWorkflow(submitted.workflow); - const result = submitted.created - ? `Workflow ${workflow.id} was submitted with status ${workflow.status}.` - : `Workflow ${workflow.id} already exists for this idempotency key.`; - return response({ version: CONTRACT_VERSION, result, created: submitted.created, workflow }); - }, + const result = supervisor.started + ? submitted.created + ? `Workflow ${workflow.id} was submitted with status ${workflow.status}.` + : `Workflow ${workflow.id} already exists for this idempotency key.` + : `Workflow ${workflow.id} was durably submitted, but its supervisor could not be started.`; + return response({ version: CONTRACT_VERSION, result, created: submitted.created, supervisor, workflow }); + }), ); server.registerTool( @@ -167,11 +189,11 @@ export function registerWorkflowTools(input: { outputSchema: { ...commonOutput, workflow: workflowOutputSchema }, annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, openWorldHint: false }, }, - async ({ workspaceId, workflowId }) => { + async ({ workspaceId, workflowId }) => audited(input.audit, "workflow_status", workspaceId, workflowId, async () => { const workflow = publicWorkflow(requireOwned(orchestrator, workflowId, workspaceScope(workspaces, workspaceId))); const result = `Workflow ${workflow.id} is ${workflow.status}.`; return response({ version: CONTRACT_VERSION, result, workflow }); - }, + }), ); server.registerTool( @@ -187,7 +209,7 @@ export function registerWorkflowTools(input: { outputSchema: { ...commonOutput, timedOut: z.boolean(), workflow: workflowOutputSchema }, annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, openWorldHint: false }, }, - async ({ workspaceId, workflowId, timeoutMs }) => { + async ({ workspaceId, workflowId, timeoutMs }) => audited(input.audit, "workflow_wait", workspaceId, workflowId, async () => { const scope = workspaceScope(workspaces, workspaceId); const current = requireOwned(orchestrator, workflowId, scope); const waited = await orchestrator.waitForWorkspace(workflowId, scope, { timeoutMs }); @@ -197,7 +219,7 @@ export function registerWorkflowTools(input: { ? `Workflow ${workflow.id} is still ${workflow.status} after the bounded wait.` : `Workflow ${workflow.id} reached ${workflow.status}.`; return response({ version: CONTRACT_VERSION, result, timedOut, workflow }); - }, + }), ); server.registerTool( @@ -219,7 +241,7 @@ export function registerWorkflowTools(input: { }, annotations: { readOnlyHint: true, destructiveHint: false, idempotentHint: true, openWorldHint: false }, }, - async ({ workspaceId, workflowId, after, limit }) => { + async ({ workspaceId, workflowId, after, limit }) => audited(input.audit, "workflow_events", workspaceId, workflowId, async () => { const scope = workspaceScope(workspaces, workspaceId); const workflow = requireOwned(orchestrator, workflowId, scope); const page = orchestrator.eventsForWorkspace(workflowId, scope, { after, limit }); @@ -232,7 +254,7 @@ export function registerWorkflowTools(input: { events, cursor: page.nextCursor, }); - }, + }), ); server.registerTool( @@ -241,18 +263,20 @@ export function registerWorkflowTools(input: { title: "Cancel workflow", description: "Request durable cancellation of a workflow and its active local agents.", inputSchema: { workspaceId: z.string().min(1), workflowId: z.string().min(1) }, - outputSchema: { ...commonOutput, workflow: workflowOutputSchema }, + outputSchema: { ...commonOutput, supervisor: supervisorOutputSchema, workflow: workflowOutputSchema }, annotations: { readOnlyHint: false, destructiveHint: true, idempotentHint: true, openWorldHint: false }, }, - async ({ workspaceId, workflowId }) => { + async ({ workspaceId, workflowId }) => audited(input.audit, "workflow_cancel", workspaceId, workflowId, async () => { const scope = workspaceScope(workspaces, workspaceId); requireOwned(orchestrator, workflowId, scope); const cancelled = orchestrator.cancelForWorkspace(workflowId, scope); - await ensureSupervisor({ stateDir: config.stateDir }); + const supervisor = await supervisorStartState(() => launchSupervisor({ stateDir: config.stateDir })); const workflow = publicWorkflow(cancelled); - const result = `Cancellation was requested for workflow ${workflow.id}; current status is ${workflow.status}.`; - return response({ version: CONTRACT_VERSION, result, workflow }); - }, + const result = supervisor.started + ? `Cancellation was requested for workflow ${workflow.id}; current status is ${workflow.status}.` + : `Cancellation was durably requested for workflow ${workflow.id}, but its supervisor could not be started.`; + return response({ version: CONTRACT_VERSION, result, supervisor, workflow }); + }), ); } @@ -281,12 +305,14 @@ function publicWorkflow(workflow: WorkflowRunRecord) { completedAt: workflow.completedAt, cancellationRequestedAt: workflow.cancellationRequestedAt, errorCode: stringField(workflow.error, "code"), + finalResponse: stringField(workflow.result, "finalResponse"), nodes: workflow.nodes.map((node) => ({ key: node.key, status: node.status, attempt: node.attempt, completedAt: node.completedAt, errorCode: stringField(node.error, "code"), + finalResponse: stringField(node.result, "finalResponse"), })), }; } @@ -301,11 +327,66 @@ function publicEvent(event: WorkflowEvent, workflow: WorkflowRunRecord) { }; } -function stringField(value: Record | undefined, key: string): string | undefined { - const field = value?.[key]; +function stringField(value: unknown, key: string): string | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) return undefined; + const field = (value as Record)[key]; return typeof field === "string" ? field : undefined; } +async function supervisorStartState(start: () => Promise): Promise<{ + started: boolean; + errorCode?: string; +}> { + try { + await start(); + return { started: true }; + } catch { + return { started: false, errorCode: "supervisor_start_failed" }; + } +} + +async function audited( + audit: ((event: WorkflowToolAuditEvent) => void) | undefined, + tool: string, + workspaceId: string | undefined, + workflowId: string | undefined, + operation: () => Promise, +): Promise { + const startedAt = performance.now(); + try { + const result = await operation(); + emitAudit(audit, { + tool, + workspaceId, + workflowId, + success: true, + durationMs: Math.round(performance.now() - startedAt), + }); + return result; + } catch (error) { + emitAudit(audit, { + tool, + workspaceId, + workflowId, + success: false, + durationMs: Math.round(performance.now() - startedAt), + errorCode: error instanceof Error ? error.name : "Error", + }); + throw error; + } +} + +function emitAudit( + audit: ((event: WorkflowToolAuditEvent) => void) | undefined, + event: WorkflowToolAuditEvent, +): void { + try { + audit?.(event); + } catch { + // Audit failures must not change workflow tool behavior. + } +} + function response(structuredContent: T) { return { content: [{ type: "text" as const, text: structuredContent.result }], diff --git a/src/workflows/orchestrator.ts b/src/workflows/orchestrator.ts index bca3df1e..8bc8b250 100644 --- a/src/workflows/orchestrator.ts +++ b/src/workflows/orchestrator.ts @@ -70,8 +70,7 @@ export class WorkflowOrchestrator { options: WorkflowWaitOptions = {}, ): Promise { this.store.requireForWorkspace(workflowId, scope); - const workflow = await this.wait(workflowId, options); - return this.store.requireForWorkspace(workflow.id, scope); + return this.wait(workflowId, options); } events(workflowId: string, options: WorkflowEventReadOptions = {}): WorkflowEventPage { diff --git a/src/workflows/store.ts b/src/workflows/store.ts index f00b0c38..5854614f 100644 --- a/src/workflows/store.ts +++ b/src/workflows/store.ts @@ -226,7 +226,7 @@ export class WorkflowStore { .prepare("select id, request_hash from workflow_runs where idempotency_key = ?") .get(normalized.idempotencyKey) as { id: string; request_hash: string } | undefined; if (existing) { - if (existing.request_hash !== normalized.requestHash) { + if (!normalized.compatibleRequestHashes.has(existing.request_hash)) { throw new WorkflowIdempotencyConflictError(normalized.idempotencyKey); } return { workflowId: existing.id, created: false }; @@ -694,7 +694,10 @@ export class WorkflowStore { where active.workflow_run_id = r.id and active.status = 'running' ) < r.max_concurrency order by case when r.last_dispatched_at is null then 0 else 1 end, - r.last_dispatched_at, r.created_at, n.created_at, n.node_key limit 1`, + r.last_dispatched_at, + (select count(*) from workflow_node_attempts attempts + where attempts.workflow_run_id = r.id), + r.created_at, n.created_at, n.node_key limit 1`, ) .get(now) as WorkflowNodeRow | undefined; if (!candidate) return undefined; @@ -810,14 +813,18 @@ export class WorkflowStore { 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; + const update = this.database.sqlite.transaction(() => { + const attempt = this.getActiveAttempt(identity, now); + if (!attempt) return undefined; + const result = this.database.sqlite + .prepare( + `update workflow_node_attempts set phase = 'cancelling', cancellation_requested_at = ?, updated_at = ? + where node_id = ? and attempt = ? and claim_token = ? and phase != 'terminal'`, + ) + .run(now, now, attempt.nodeId, attempt.attempt, identity.claimToken); + return result.changes === 1 ? this.getAttempt(identity) : undefined; + }); + return update.immediate(); } recordNodeProviderSession( @@ -825,21 +832,18 @@ export class WorkflowStore { 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; + const update = this.database.sqlite.transaction(() => { + const attempt = this.getActiveAttempt(identity, now); + if (!attempt) return undefined; + const result = this.database.sqlite + .prepare( + `update workflow_node_attempts set provider_session_id = ?, updated_at = ? + where node_id = ? and attempt = ? and claim_token = ? and phase != 'terminal'`, + ) + .run(providerSessionId, now, attempt.nodeId, attempt.attempt, identity.claimToken); + return result.changes === 1 ? this.getAttempt(identity) : undefined; + }); + return update.immediate(); } appendNodeExecutionEvent(input: { @@ -850,8 +854,8 @@ export class WorkflowStore { }): { 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 attempt = this.getActiveAttempt(input.identity, now); + if (!attempt) return undefined; const existing = this.database.sqlite .prepare( `select workflow_sequence from workflow_provider_events @@ -1174,6 +1178,18 @@ export class WorkflowStore { return row ? rowToWorktree(row) : undefined; } + listExpiredWorktrees(now = new Date().toISOString()): WorkflowWorktreeRecord[] { + const rows = this.database.sqlite + .prepare( + `select * from workflow_worktrees + where state in ('preserved', 'cleanup_failed') + and retain_until is not null and retain_until <= ? + order by retain_until, created_at`, + ) + .all(now) as WorkflowWorktreeRow[]; + return rows.map(rowToWorktree); + } + updateWorktreeState( workflowId: string, nodeKey: string, @@ -1230,26 +1246,54 @@ export class WorkflowStore { return row ? rowToAttempt(row) : undefined; } - private updateAttemptPhase( + private getActiveAttempt( identity: WorkflowAttemptIdentity, - phase: "dispatching" | "running", + now: string, ): WorkflowNodeAttemptRecord | undefined { - const now = new Date().toISOString(); - const result = this.database.sqlite + const row = 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'`, + `select attempts.* from workflow_node_attempts attempts + join workflow_nodes nodes on nodes.id = attempts.node_id + join workflow_supervisor supervisor on supervisor.id = 1 + where attempts.workflow_run_id = ? and attempts.node_key = ? + and attempts.attempt = ? and attempts.claim_token = ? + and attempts.phase != 'terminal' + and nodes.status = 'running' and nodes.attempt = attempts.attempt + and nodes.claim_token = attempts.claim_token and nodes.claim_expires_at > ? + and nodes.supervisor_owner_token = attempts.supervisor_owner_token + and nodes.supervisor_owner_epoch = attempts.supervisor_owner_epoch + and supervisor.owner_token = attempts.supervisor_owner_token + and supervisor.owner_epoch = attempts.supervisor_owner_epoch + and supervisor.lease_expires_at > ?`, ) - .run( - phase, - now, + .get( identity.workflowId, identity.nodeKey, identity.attempt, identity.claimToken, - ); - return result.changes === 1 ? this.getAttempt(identity) : undefined; + now, + now, + ) as WorkflowAttemptRow | undefined; + return row ? rowToAttempt(row) : undefined; + } + + private updateAttemptPhase( + identity: WorkflowAttemptIdentity, + phase: "dispatching" | "running", + ): WorkflowNodeAttemptRecord | undefined { + const now = new Date().toISOString(); + const update = this.database.sqlite.transaction(() => { + const attempt = this.getActiveAttempt(identity, now); + if (!attempt) return undefined; + const result = this.database.sqlite + .prepare( + `update workflow_node_attempts set phase = ?, updated_at = ? + where node_id = ? and attempt = ? and claim_token = ? and phase != 'terminal'`, + ) + .run(phase, now, attempt.nodeId, attempt.attempt, identity.claimToken); + return result.changes === 1 ? this.getAttempt(identity) : undefined; + }); + return update.immediate(); } private hydrateWorkflow(row: WorkflowRunRow): WorkflowRunRecord { @@ -1434,6 +1478,7 @@ function normalizeSubmission(request: SubmitWorkflowRequest): { workspace?: WorkflowWorkspaceScope; maxConcurrency: number; requestHash: string; + compatibleRequestHashes: ReadonlySet; } { const definition = normalizeDefinition(request.definition); const input = normalizeObject(request.input ?? {}, "Workflow input"); @@ -1452,9 +1497,10 @@ function normalizeSubmission(request: SubmitWorkflowRequest): { } : undefined; const maxConcurrency = readBoundedInteger(policy.maxConcurrency, 1, MAX_RUN_CONCURRENCY, "Workflow maxConcurrency"); - const requestHash = createHash("sha256") - .update(canonicalJson({ definition, input, policy, workspace: workspace ?? null })) - .digest("hex"); + const requestHash = hashWorkflowRequest({ definition, input, policy, workspace: workspace ?? null }); + const compatibleRequestHashes = new Set([requestHash]); + const legacyRequest = legacySingleAgentRequest(definition, input, policy, workspace); + if (legacyRequest) compatibleRequestHashes.add(hashWorkflowRequest(legacyRequest)); return { definition, definitionJson, @@ -1464,6 +1510,52 @@ function normalizeSubmission(request: SubmitWorkflowRequest): { workspace, maxConcurrency, requestHash, + compatibleRequestHashes, + }; +} + +function hashWorkflowRequest(value: unknown): string { + return createHash("sha256").update(canonicalJson(value)).digest("hex"); +} + +function legacySingleAgentRequest( + definition: WorkflowDefinition, + input: JsonObject, + policy: WorkflowPolicy, + workspace: WorkflowWorkspaceScope | undefined, +): JsonObject | undefined { + if ( + input.kind !== "single" || + definition.nodes.length !== 1 || + (definition.edges?.length ?? 0) !== 0 || + policy.maxConcurrency !== 1 || + typeof policy.access !== "string" + ) return undefined; + const node = definition.nodes[0]!; + const config = { ...(node.config ?? {}) } as JsonObject; + const prompt = config.prompt; + const retry = config.retry; + if ( + typeof prompt !== "string" || + !isObject(retry) || + retry.maxAttempts !== 1 || + !Array.isArray(retry.retryOn) || + retry.retryOn.length !== 0 || + retry.backoffMs !== 0 + ) return undefined; + delete config.worktreeRoot; + delete config.baseSha; + delete config.retry; + const legacyDefinition: WorkflowDefinition = { + version: WORKFLOW_DEFINITION_VERSION, + nodes: [{ key: node.key, type: "agent", config }], + edges: [], + }; + return { + definition: legacyDefinition as unknown as JsonObject, + input: { prompt }, + policy: { version: WORKFLOW_POLICY_VERSION, access: policy.access }, + workspace: workspace ? { ...workspace } : null, }; } diff --git a/src/workflows/supervisor.test.ts b/src/workflows/supervisor.test.ts index 09732b8f..5c221022 100644 --- a/src/workflows/supervisor.test.ts +++ b/src/workflows/supervisor.test.ts @@ -16,6 +16,8 @@ try { await testStaleSupervisorProcessRecovery(join(root, "stale-supervisor")); await testAtomicClaimAndLeaseRecovery(join(root, "claims")); await testDispatchHeartbeatsProtectLeases(join(root, "dispatch-heartbeats")); + await testPreparationHeartbeatsProtectLeases(join(root, "preparation-heartbeats")); + await testTimeoutWinsOverLateSuccess(join(root, "timeout-race")); await testCancellationBeforeAndDuringExecution(join(root, "cancellation")); } finally { rmSync(root, { recursive: true, force: true }); @@ -80,7 +82,7 @@ async function testStaleSupervisorProcessRecovery(stateDir: string): Promise stale.ownerEpoch); @@ -112,6 +114,23 @@ async function testAtomicClaimAndLeaseRecovery(stateDir: string): Promise await delay(15); assert.equal(store.reconcileExpiredClaims(), 0); await delay(25); + const expiredIdentity = { + workflowId: first.id, + nodeKey: "agent", + attempt: 1, + claimToken: "first", + }; + assert.equal(store.markNodeDispatching(expiredIdentity), undefined); + assert.equal(store.recordNodeProviderSession(expiredIdentity, "late-session"), undefined); + assert.throws( + () => store.appendNodeExecutionEvent({ + identity: expiredIdentity, + sourceSequence: 1, + type: "provider.output", + payload: { delta: "late" }, + }), + /no longer active/, + ); assert.equal(store.reconcileExpiredClaims(), 1); const failed = store.require(first.id); assert.equal(failed.status, "failed"); @@ -165,6 +184,84 @@ async function testDispatchHeartbeatsProtectLeases(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + const workflow = store.submit(workflowRequest(executionSnapshot({ + effectivePolicy: { version: 1, mode: "workflow", access: "workspace_write", environment: {} }, + worktreeRoot: join(stateDir, "worktrees"), + baseSha: "base", + }))).workflow; + store.close(); + + const ran = await runWorkflowSupervisor(stateDir, { + idleMs: 20, + heartbeatMs: 5, + supervisorLeaseMs: 40, + nodeLeaseMs: 40, + worktreeAllocator: async ({ store: allocatorStore, identity }) => { + await delay(90); + const record = allocatorStore.recordWorktree({ + workflowId: identity.workflowId, + nodeKey: identity.nodeKey, + attempt: identity.attempt, + path: join(stateDir, "worktrees", "prepared"), + sourceRoot: stateDir, + baseSha: "base", + retainUntil: new Date(Date.now() + 60_000).toISOString(), + }); + return allocatorStore.updateWorktreeState(record.workflowId, record.nodeKey, record.attempt, "active"); + }, + handleFactory: successfulHandleFactory, + }); + assert.equal(ran, true); + + const reader = new WorkflowStore(stateDir); + try { + assert.equal(reader.require(workflow.id).status, "succeeded"); + } finally { + reader.close(); + } +} + +async function testTimeoutWinsOverLateSuccess(stateDir: string): Promise { + const store = new WorkflowStore(stateDir); + const workflow = store.submit(workflowRequest(executionSnapshot({ timeoutMs: 20 }))).workflow; + store.close(); + + await runWorkflowSupervisor(stateDir, { + idleMs: 20, + heartbeatMs: 5, + supervisorLeaseMs: 200, + nodeLeaseMs: 200, + handleFactory: async () => { + const controller = new LocalAgentRunController("fake"); + const timer = setTimeout(() => { + controller.succeed({ + provider: "fake", + providerSessionId: "late-success", + finalResponse: "too late", + items: [], + }); + }, 60); + controller.setLifecycle({ + cancel: () => undefined, + dispose: () => clearTimeout(timer), + }); + return controller; + }, + }); + + const reader = new WorkflowStore(stateDir); + try { + const completed = reader.require(workflow.id); + assert.equal(completed.status, "failed"); + assert.equal((completed.error as JsonObject).code, "timed_out"); + assert.equal(completed.nodes[0]!.status, "failed"); + } finally { + reader.close(); + } +} + async function testCancellationBeforeAndDuringExecution(stateDir: string): Promise { const beforeStore = new WorkflowStore(stateDir); const before = beforeStore.submit(workflowRequest(executionSnapshot())).workflow; diff --git a/src/workflows/supervisor.ts b/src/workflows/supervisor.ts index ef1a24af..8d0acd5c 100644 --- a/src/workflows/supervisor.ts +++ b/src/workflows/supervisor.ts @@ -11,6 +11,7 @@ import { isLocalAgentProvider } from "../local-agent-profiles.js"; import type { EffectiveLocalAgentPolicy } from "./policy.js"; import { allocateWorkflowWorktree, + cleanupExpiredWorkflowWorktrees, cleanupWorkflowWorktree, preserveWorkflowWorktree, } from "./worktrees.js"; @@ -46,6 +47,7 @@ export interface WorkflowSupervisorOptions { ownerPid?: number; globalConcurrency?: number; handleFactory?: WorkflowHandleFactory; + worktreeAllocator?: typeof allocateWorkflowWorktree; } export async function runWorkflowSupervisor( @@ -75,6 +77,10 @@ export async function runWorkflowSupervisor( try { store.reconcileExpiredClaims(); store.convergeCancellations(); + await cleanupExpiredWorkflowWorktrees({ + store, + beforeCleanup: () => store.heartbeatSupervisor(identity, supervisorLeaseMs), + }); while (store.heartbeatSupervisor(identity, supervisorLeaseMs)) { store.reconcileExpiredClaims(); store.convergeCancellations(); @@ -94,6 +100,7 @@ export async function runWorkflowSupervisor( nodeLeaseMs, heartbeatMs, handleFactory: options.handleFactory ?? defaultHandleFactory, + worktreeAllocator: options.worktreeAllocator ?? allocateWorkflowWorktree, }).finally(() => { active.delete(execution); lastWorkAt = Date.now(); @@ -139,6 +146,7 @@ async function executeClaim( nodeLeaseMs: number; heartbeatMs: number; handleFactory: WorkflowHandleFactory; + worktreeAllocator: typeof allocateWorkflowWorktree; }, ): Promise { const identity: WorkflowAttemptIdentity = { @@ -159,13 +167,41 @@ async function executeClaim( let leaseFailure: Error | undefined; let tickRunning = false; + 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; + } + }; + try { + await tick(); + if (leaseFailure) throw leaseFailure; + heartbeat = setInterval(() => void tick().catch(() => undefined), options.heartbeatMs); + const execution = readExecutionSnapshot(config); const fullPrompt = execution.profileBody ? `${execution.profileBody}\n\nTask:\n${execution.prompt}` : execution.prompt; if (execution.effectivePolicy.access === "workspace_write") { - const allocation = await allocateWorkflowWorktree({ + const allocation = await options.worktreeAllocator({ store, identity, sourceRoot: execution.workspaceRoot, @@ -180,33 +216,6 @@ async function executeClaim( 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, @@ -239,6 +248,7 @@ async function executeClaim( } finally { await drain; } + if (timeoutObserved) throw new Error("Workflow node timed out."); if (result.providerSessionId) { store.recordNodeProviderSession(identity, result.providerSessionId); } diff --git a/src/workflows/workflows.test.ts b/src/workflows/workflows.test.ts index 1c4fb475..3ead272d 100644 --- a/src/workflows/workflows.test.ts +++ b/src/workflows/workflows.test.ts @@ -12,7 +12,7 @@ import { WorkflowTransitionError, WorkflowValidationError, } from "./store.js"; -import type { SubmitWorkflowRequest, WorkflowDefinitionV1 } from "./types.js"; +import type { JsonObject, SubmitWorkflowRequest, WorkflowDefinitionV1 } from "./types.js"; interface WorkerSpec { action: "append" | "claim" | "submit"; @@ -37,6 +37,7 @@ async function runTests(): Promise { const root = mkdtempSync(join(tmpdir(), "devspace-workflows-test-")); try { testSubmissionAndIdempotency(join(root, "submission")); + testLegacySingleAgentIdempotencyCompatibility(join(root, "legacy-idempotency")); await testConcurrentSubmission(join(root, "concurrent-submission")); testMonotonicCursorReads(join(root, "events")); await testConcurrentEventSequencing(join(root, "concurrent-events")); @@ -102,6 +103,49 @@ function testSubmissionAndIdempotency(stateDir: string): void { } } +function testLegacySingleAgentIdempotencyCompatibility(stateDir: string): void { + const store = new WorkflowStore(stateDir); + try { + const workspace = { workspaceId: "workspace", workspaceRoot: "/tmp/workspace" }; + const legacyConfig: JsonObject = { + profileBody: "", + profileName: "fake", + profileHash: "hash", + provider: "fake", + model: null, + thinking: null, + effectivePolicy: { version: 1, mode: "workflow", access: "read_only", environment: {} }, + workspaceRoot: workspace.workspaceRoot, + prompt: "same prompt", + timeoutMs: null, + environmentPolicy: {}, + }; + const legacy = store.submit({ + definition: singleAgentDefinition(legacyConfig), + input: { prompt: "same prompt" }, + policy: { version: 1, access: "read_only" }, + idempotencyKey: "legacy-key", + workspace, + }); + const replay = store.submit({ + definition: singleAgentDefinition({ + ...legacyConfig, + worktreeRoot: "/tmp/worktrees", + baseSha: null, + retry: { maxAttempts: 1, retryOn: [], backoffMs: 0 }, + }), + input: { kind: "single" }, + policy: { version: 1, access: "read_only", maxConcurrency: 1 }, + idempotencyKey: "legacy-key", + workspace, + }); + assert.equal(replay.created, false); + assert.equal(replay.workflow.id, legacy.workflow.id); + } finally { + store.close(); + } +} + async function testConcurrentSubmission(stateDir: string): Promise { const same = await runWorkers(stateDir, [ { action: "submit", requestVariant: "same" }, @@ -655,7 +699,7 @@ function delay(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } -function singleAgentDefinition(config: Record = {}): WorkflowDefinitionV1 { +function singleAgentDefinition(config: JsonObject = {}): WorkflowDefinitionV1 { return { version: 1, nodes: [{ key: "agent", type: "agent", config }], edges: [] }; } diff --git a/src/workflows/worktrees.ts b/src/workflows/worktrees.ts index 5bd98d35..99dbe773 100644 --- a/src/workflows/worktrees.ts +++ b/src/workflows/worktrees.ts @@ -1,6 +1,6 @@ import { execFile } from "node:child_process"; import { mkdir, realpath } from "node:fs/promises"; -import { basename, join, relative, resolve } from "node:path"; +import { basename, dirname, join, relative, resolve } from "node:path"; import { promisify } from "node:util"; import { WorkflowStore, WorkflowValidationError } from "./store.js"; import type { WorkflowAttemptIdentity, WorkflowWorktreeRecord } from "./types.js"; @@ -82,6 +82,33 @@ export function preserveWorkflowWorktree( return store.updateWorktreeState(identity.workflowId, identity.nodeKey, identity.attempt, "preserved"); } +export async function cleanupExpiredWorkflowWorktrees(input: { + store: WorkflowStore; + now?: string; + beforeCleanup?: () => boolean; +}): Promise { + let removed = 0; + for (const record of input.store.listExpiredWorktrees(input.now)) { + if (input.beforeCleanup && !input.beforeCleanup()) break; + try { + await cleanupWorkflowWorktree({ + store: input.store, + identity: { + workflowId: record.workflowId, + nodeKey: record.nodeKey, + attempt: record.attempt, + claimToken: "retention-cleanup", + }, + worktreeRoot: dirname(record.path), + }); + removed += 1; + } catch { + // cleanupWorkflowWorktree persists the failure for a later retry. + } + } + return removed; +} + export async function cleanupWorkflowWorktree(input: { store: WorkflowStore; identity: WorkflowAttemptIdentity; @@ -95,15 +122,20 @@ export async function cleanupWorkflowWorktree(input: { if (!record || record.state === "removed") return record; try { + const expectedName = `${safeSegment(basename(record.sourceRoot))}-${safeSegment(record.workflowId)}-${safeSegment(record.nodeKey)}-a${record.attempt}`; + if (basename(record.path) !== expectedName) { + throw new WorkflowValidationError("Workflow worktree path does not match persisted ownership"); + } const worktreeRoot = await canonicalDirectory(input.worktreeRoot, "workflow worktree root"); assertContained(record.path, worktreeRoot); const canonicalWorktreePath = await canonicalDirectory(record.path, "workflow worktree path"); assertContained(canonicalWorktreePath, worktreeRoot); const registered = await registeredWorktrees(record.sourceRoot); const match = registered.find((entry) => resolve(entry.path) === resolve(record.path)); - if (!match || match.head !== record.baseSha) { + if (!match) { throw new WorkflowValidationError("Workflow worktree Git registration does not match persisted ownership"); } + await git(["merge-base", "--is-ancestor", record.baseSha, match.head], record.sourceRoot); await git(["worktree", "remove", "--force", record.path], record.sourceRoot); return input.store.updateWorktreeState( record.workflowId, From 7fa59181be232c5d3ff88dbdf73c5a7f8d9fb516 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 22:59:14 +0530 Subject: [PATCH 05/10] fix(process): escalate Windows interrupt cleanup Co-Authored-By: Claude --- src/process-sessions.ts | 21 +++++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/src/process-sessions.ts b/src/process-sessions.ts index f414df19..5730c3e8 100644 --- a/src/process-sessions.ts +++ b/src/process-sessions.ts @@ -1,5 +1,9 @@ import { spawn } from "node:child_process"; -import { resolveShellCommand, terminateProcessTree } from "./process-platform.js"; +import { + resolveShellCommand, + terminateProcessTree, + terminateProcessTreeGracefully, +} from "./process-platform.js"; const DEFAULT_EXEC_YIELD_MS = 10_000; const DEFAULT_INTERACTIVE_YIELD_MS = 250; @@ -46,7 +50,7 @@ export interface ProcessSnapshot { interface ManagedProcess { write(data: string): void; - kill(signal?: NodeJS.Signals): void; + kill(signal?: NodeJS.Signals): void | Promise; resize?(columns: number, rows: number): void; } @@ -259,7 +263,7 @@ export class ProcessSessionManager { const interruptRequested = chars.includes("\u0003") && session.running; if (interruptRequested) { - session.process?.kill("SIGINT"); + await session.process?.kill("SIGINT"); } const writableChars = chars.replaceAll("\u0003", ""); if (writableChars && session.running) session.process?.write(writableChars); @@ -339,7 +343,16 @@ export class ProcessSessionManager { session.process = { write: (data) => child.stdin.write(data), - kill: (signal = "SIGTERM") => terminateProcessTree(child, signal, detached), + kill: (signal = "SIGTERM") => { + if (process.platform === "win32" && signal === "SIGINT") { + terminateProcessTree(child, "SIGKILL", detached); + return; + } + if (process.platform === "win32" && signal !== "SIGKILL") { + return terminateProcessTreeGracefully(child, detached); + } + terminateProcessTree(child, signal, detached); + }, resize: input.tty ? () => undefined : undefined, }; child.stdout.on("data", (data: Buffer) => this.append(session, data.toString("utf8"))); From d136508f76e3212c387fed484e460090fe404b8f Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 23:11:42 +0530 Subject: [PATCH 06/10] fix(db): close failed workflow database initialization Co-Authored-By: Claude --- src/db/client.ts | 27 ++++++++++++++++----------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/src/db/client.ts b/src/db/client.ts index e2264091..88f5253f 100644 --- a/src/db/client.ts +++ b/src/db/client.ts @@ -23,18 +23,23 @@ export function openDatabase(stateDir: string): DatabaseHandle { chmodSync(stateDir, 0o700); 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("foreign_keys = ON"); - migrateDatabase(sqlite); + try { + chmodSync(path, 0o600); + sqlite.pragma("busy_timeout = 5000"); + sqlite.pragma("journal_mode = WAL"); + sqlite.pragma("synchronous = NORMAL"); + sqlite.pragma("foreign_keys = ON"); + migrateDatabase(sqlite); - return { - sqlite, - db: createDrizzleDatabase(sqlite), - close: () => sqlite.close(), - }; + return { + sqlite, + db: createDrizzleDatabase(sqlite), + close: () => sqlite.close(), + }; + } catch (error) { + sqlite.close(); + throw error; + } } function createDrizzleDatabase(sqlite: SqliteDatabase) { From e4566a60bd713ba5412283ce7c3c8810b33bf141 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 23:17:04 +0530 Subject: [PATCH 07/10] test(workflows): use portable loader URL Co-Authored-By: Claude --- src/workflows/cli.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/workflows/cli.test.ts b/src/workflows/cli.test.ts index 43d0e949..9597c4e4 100644 --- a/src/workflows/cli.test.ts +++ b/src/workflows/cli.test.ts @@ -11,7 +11,7 @@ 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)); +const loaderUrl = new URL("../../node_modules/tsx/dist/loader.mjs", import.meta.url).href; mkdirSync(project, { recursive: true }); mkdirSync(stateDir, { recursive: true }); mkdirSync(agentsDir, { recursive: true }); @@ -122,7 +122,7 @@ try { } function runCli(args: string[], env: NodeJS.ProcessEnv) { - return spawnSync(process.execPath, ["--import", loaderPath, cliPath, ...args], { + return spawnSync(process.execPath, ["--import", loaderUrl, cliPath, ...args], { cwd: project, env, encoding: "utf8", From 71e762d1eccd6f52b3d7bc69b46adc81b22e31bc Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 23:22:00 +0530 Subject: [PATCH 08/10] fix(workflows): detach supervisor from caller directory Co-Authored-By: Claude --- src/workflows/supervisor-launch.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/workflows/supervisor-launch.ts b/src/workflows/supervisor-launch.ts index 209bcea6..09de35ae 100644 --- a/src/workflows/supervisor-launch.ts +++ b/src/workflows/supervisor-launch.ts @@ -1,4 +1,5 @@ import { spawn } from "node:child_process"; +import { dirname } from "node:path"; import { fileURLToPath } from "node:url"; import { filterWorkflowEnvironment } from "./policy.js"; import { WorkflowStore } from "./store.js"; @@ -56,6 +57,7 @@ export async function ensureSupervisor(input: { input.stateDir, ], { + cwd: dirname(cliEntrypoint), detached: true, stdio: "ignore", env: environment, From 57b1c22f95ebd3b8d93e8ea7e3b583a9c20fc7a3 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 23:36:37 +0530 Subject: [PATCH 09/10] fix(db): retry concurrent WAL initialization Co-Authored-By: Claude --- src/db/client.ts | 22 +++++++++++++++++++++- 1 file changed, 21 insertions(+), 1 deletion(-) diff --git a/src/db/client.ts b/src/db/client.ts index 88f5253f..4eeca0c2 100644 --- a/src/db/client.ts +++ b/src/db/client.ts @@ -5,6 +5,10 @@ import { drizzle } from "drizzle-orm/better-sqlite3"; import * as schema from "./schema.js"; import { migrateDatabase } from "./migrations.js"; +const JOURNAL_MODE_RETRY_TIMEOUT_MS = 5_000; +const JOURNAL_MODE_RETRY_DELAY_MS = 25; +const journalModeRetrySignal = new Int32Array(new SharedArrayBuffer(4)); + export type SqliteDatabase = Database.Database; export type AppDatabase = ReturnType; @@ -26,7 +30,7 @@ export function openDatabase(stateDir: string): DatabaseHandle { try { chmodSync(path, 0o600); sqlite.pragma("busy_timeout = 5000"); - sqlite.pragma("journal_mode = WAL"); + enableWriteAheadLogging(sqlite); sqlite.pragma("synchronous = NORMAL"); sqlite.pragma("foreign_keys = ON"); migrateDatabase(sqlite); @@ -42,6 +46,22 @@ export function openDatabase(stateDir: string): DatabaseHandle { } } +function enableWriteAheadLogging(sqlite: SqliteDatabase): void { + const deadline = Date.now() + JOURNAL_MODE_RETRY_TIMEOUT_MS; + while (true) { + try { + sqlite.pragma("journal_mode = WAL"); + return; + } catch (error) { + const code = (error as { code?: string }).code; + if ((code !== "SQLITE_BUSY" && code !== "SQLITE_LOCKED") || Date.now() >= deadline) { + throw error; + } + Atomics.wait(journalModeRetrySignal, 0, 0, JOURNAL_MODE_RETRY_DELAY_MS); + } + } +} + function createDrizzleDatabase(sqlite: SqliteDatabase) { return drizzle(sqlite, { schema }); } From 7392b939b3a88cde9ed1fc186ba1707fac01a0d7 Mon Sep 17 00:00:00 2001 From: Waishnav Date: Fri, 17 Jul 2026 23:48:01 +0530 Subject: [PATCH 10/10] test(workflows): avoid terminal wait race Co-Authored-By: Claude --- src/workflows/cli.test.ts | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/workflows/cli.test.ts b/src/workflows/cli.test.ts index 9597c4e4..b4d87f7d 100644 --- a/src/workflows/cli.test.ts +++ b/src/workflows/cli.test.ts @@ -43,6 +43,13 @@ try { assert.equal(runEnvelope.version, 1); assert.equal(runEnvelope.ok, true); const workflowId = runEnvelope.workflow!.id; + 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, JSON.stringify(timedEnvelope)); + const replay = runCli([ "workflows", "run", "reviewer", "--prompt", "private task", "--json", "--idempotency-key", "shell-parent", @@ -51,13 +58,6 @@ try { 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, JSON.stringify(timedEnvelope)); - const waited = runCli([ "workflows", "wait", workflowId, "--timeout-ms", "10000", "--after", "0", "--json", ], baseEnv);