diff --git a/docs/workflow-runtime.md b/docs/workflow-runtime.md new file mode 100644 index 00000000..efbdf95b --- /dev/null +++ b/docs/workflow-runtime.md @@ -0,0 +1,75 @@ +# Restricted JavaScript workflow runtime + +DevSpace can run a bounded JavaScript workflow that coordinates the same durable local-agent workflows exposed through CLI and MCP. The script is orchestration code only: provider execution, retries, policy enforcement, worktrees, persistence, and cancellation remain owned by the shared workflow store and supervisor. + +## Run a script + +```bash +devspace workflows script ./review.workflow.js \ + --args-json '{"topic":"authentication"}' \ + --idempotency-key review-auth-v1 \ + --json +``` + +The script must be inside the canonical `DEVSPACE_WORKSPACE_ROOT` and use a `.js` or `.mjs` extension. `--args-json` must contain a JSON object. The command emits one versioned JSON envelope and waits for the script to reach a terminal state. + +## Script contract + +```js +// @devspace-workflow {"version":1,"name":"review","maxAgentCalls":4,"maxConcurrency":2,"timeoutMs":600000} +export default async function ({ agent, parallel, pipeline, phase, log, args, budget }) { + const reviews = await phase("review", () => parallel([ + () => agent({ target: "security", prompt: `Review ${args.topic}` }), + () => agent({ target: "tests", prompt: `Test ${args.topic}` }) + ])); + + const summary = await pipeline([ + (items) => agent({ + target: "reviewer", + prompt: `Synthesize: ${JSON.stringify(items)}` + }) + ], reviews); + + log("review complete", { workflowId: summary.workflowId }); + return { reviews, summary }; +} +``` + +The optional first non-empty line is strict JSON metadata. Supported fields are: + +- `version`: must be `1`. +- `name`: at most 128 characters. +- `description`: at most 1,024 characters. +- `maxAgentCalls`: 1–64; default 16. +- `maxConcurrency`: 1–16; default 4. +- `timeoutMs`: 1–86,400,000; default 15 minutes. + +Unknown metadata and `agent()` fields are rejected. + +## Primitives + +- `agent(options)`: submits one durable local-agent workflow and returns `{ workflowId, status, finalResponse }`. Options support `target`, `prompt`, `model`, `thinking`, `access`, `timeoutMs`, and the existing bounded read-only retry policy. +- `parallel(tasks)`: runs task functions concurrently. Actual agent dispatch remains bounded by `maxConcurrency` and supervisor limits. +- `pipeline(tasks, initialValue)`: awaits task functions in sequence, passing each result to the next task. +- `phase(name, task)`: emits durable `phase.started` and `phase.completed` runtime events. +- `log(message, data)`: emits a bounded durable log event. +- `args`: deeply frozen JSON supplied by `--args-json`. +- `budget`: deeply frozen effective `maxAgentCalls`, `maxConcurrency`, and `timeoutMs` values. + +Every `agent()` promise must be awaited, directly or through `parallel`, `pipeline`, or another awaited operation. Returning while calls are outstanding fails the runtime and requests cancellation of active children. + +## Isolation and limits + +The workflow body runs in a separate Node process with the permission model enabled, an empty environment, no inherited stdio, and a 64 MiB old-space limit. Before evaluating workflow code, the child applies SES `lockdown()` and creates a fresh authority-free `Compartment`. Only a hardened orchestration API and bounded JSON values cross into that compartment; Node APIs, imports, `process`, `require`, ambient environment variables, filesystem, network, database handles, and the orchestrator are not endowed. + +The child can read only the installed SES runtime packages. Node permissions provide a second boundary around filesystem, subprocess, worker, addon, and—where supported by the running Node release—network access. Source, arguments, logs, event payloads, results, call counts, concurrency, and wall-clock duration are bounded. Scripts can only cause agent execution through `agent()`. + +Keep Node and the pinned SES dependency current. For hostile multi-tenant execution requiring protection beyond the JavaScript object-capability boundary, run DevSpace inside an operator-managed OS or container sandbox as an additional layer. + +## Durability and replay + +Runtime runs, events, and each indexed `agent()` call are journaled in SQLite. Reusing an idempotency key with identical source, arguments, metadata, budget, and workspace returns the existing successful result. Reusing it with different input is rejected. + +After a failed or interrupted runtime, invoking the same request re-executes the orchestration function from the beginning. A completed call prefix is replayed only when the call index and normalized request hash match exactly; divergence fails closed. In-flight child workflow IDs are reused rather than submitted twice. + +Replay does not make arbitrary side effects safe. The restricted runtime intentionally offers no raw filesystem, network, shell, or database primitive. `workspace_write` agents use the workflow scheduler's isolated managed-worktree semantics and are not automatically retried. diff --git a/package-lock.json b/package-lock.json index 0d7decc4..d9214e34 100644 --- a/package-lock.json +++ b/package-lock.json @@ -27,6 +27,7 @@ "react": "^19.2.6", "react-dom": "^19.2.6", "semver": "^7.8.4", + "ses": "^2.2.0", "yaml": "^2.9.0", "zod": "^4.4.3" }, @@ -2101,6 +2102,24 @@ "tslib": "^2.4.0" } }, + "node_modules/@endo/cache-map": { + "version": "1.1.0", + "resolved": "https://registry.npmjs.org/@endo/cache-map/-/cache-map-1.1.0.tgz", + "integrity": "sha512-owFGshs/97PDw9oguZqU/px8Lv1d0KjAUtDUiPwKHNXRVUE/jyettEbRoTbNJR1OaI8biMn6bHr9kVJsOh6dXw==", + "license": "Apache-2.0" + }, + "node_modules/@endo/env-options": { + "version": "1.1.11", + "resolved": "https://registry.npmjs.org/@endo/env-options/-/env-options-1.1.11.tgz", + "integrity": "sha512-p9OnAPsdqoX4YJsE98e3NBVhIr2iW9gNZxHhAI2/Ul5TdRfoOViItzHzTqrgUVopw6XxA1u1uS6CykLMDUxarA==", + "license": "Apache-2.0" + }, + "node_modules/@endo/immutable-arraybuffer": { + "version": "1.1.2", + "resolved": "https://registry.npmjs.org/@endo/immutable-arraybuffer/-/immutable-arraybuffer-1.1.2.tgz", + "integrity": "sha512-u+NaYB2aqEugQ3u7w3c5QNkPogf8q/xGgsPaqdY6pUiGWtYiTiFspKFcha6+oeZhWXWQ23rf0KrUq0kfuzqYyQ==", + "license": "Apache-2.0" + }, "node_modules/@esbuild/aix-ppc64": { "version": "0.28.1", "resolved": "https://registry.npmjs.org/@esbuild/aix-ppc64/-/aix-ppc64-0.28.1.tgz", @@ -5534,6 +5553,17 @@ "url": "https://opencollective.com/express" } }, + "node_modules/ses": { + "version": "2.2.0", + "resolved": "https://registry.npmjs.org/ses/-/ses-2.2.0.tgz", + "integrity": "sha512-mXZfO9O2bhE9E3INX5Dbqq+eo1Dj6Yeo+r7jas7GYTXRS6fzcZFYf0UbSHcXO7xJuPbhztBskSuSqJLMMoYMVQ==", + "license": "Apache-2.0", + "dependencies": { + "@endo/cache-map": "^1.1.0", + "@endo/env-options": "^1.1.11", + "@endo/immutable-arraybuffer": "^1.1.2" + } + }, "node_modules/setprototypeof": { "version": "1.2.0", "resolved": "https://registry.npmjs.org/setprototypeof/-/setprototypeof-1.2.0.tgz", diff --git a/package.json b/package.json index 274635d3..061618e1 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/dag-scheduler.test.ts && tsx src/workflows/mcp.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/runtime.test.ts && tsx src/workflows/cli.test.ts", "typecheck": "tsc -p tsconfig.json --noEmit" }, "keywords": [], @@ -53,6 +53,7 @@ "react": "^19.2.6", "react-dom": "^19.2.6", "semver": "^7.8.4", + "ses": "^2.2.0", "yaml": "^2.9.0", "zod": "^4.4.3" }, diff --git a/src/db/migrations.ts b/src/db/migrations.ts index 83630ffb..3ff631c1 100644 --- a/src/db/migrations.ts +++ b/src/db/migrations.ts @@ -37,6 +37,11 @@ const migrations: Migration[] = [ name: "workflow-dag-scheduler", up: migrateWorkflowDagScheduler, }, + { + version: 7, + name: "workflow-js-runtime", + up: migrateWorkflowJsRuntime, + }, ]; export function migrateDatabase(sqlite: Database.Database): void { @@ -380,6 +385,62 @@ function migrateWorkflowDagScheduler(sqlite: Database.Database): void { `); } +function migrateWorkflowJsRuntime(sqlite: Database.Database): void { + sqlite.exec(` + create table if not exists workflow_runtime_runs ( + id text primary key, + workspace_id text not null, + workspace_root text not null, + source_hash text not null, + args_json text not null, + metadata_json text not null, + budget_json text not null, + idempotency_key text, + request_hash text not null, + status text not null check (status in ('running', 'succeeded', 'failed', 'cancelled')), + result_json text, + error_json text, + created_at text not null, + updated_at text not null, + completed_at text, + unique (workspace_id, workspace_root, idempotency_key) + ); + + create table if not exists workflow_runtime_calls ( + runtime_run_id text not null, + call_index integer not null, + request_hash text not null, + request_json text not null, + workflow_run_id text, + status text not null check (status in ('pending', 'running', 'succeeded', 'failed')), + result_json text, + error_json text, + created_at text not null, + updated_at text not null, + completed_at text, + primary key (runtime_run_id, call_index), + foreign key (runtime_run_id) references workflow_runtime_runs(id) on delete cascade, + foreign key (workflow_run_id) references workflow_runs(id) on delete set null + ); + + create table if not exists workflow_runtime_events ( + runtime_run_id text not null, + sequence integer not null, + event_type text not null, + payload_json text not null, + created_at text not null, + primary key (runtime_run_id, sequence), + foreign key (runtime_run_id) references workflow_runtime_runs(id) on delete cascade + ); + + create index if not exists workflow_runtime_runs_workspace_idx + on workflow_runtime_runs(workspace_id, workspace_root, created_at); + + create index if not exists workflow_runtime_calls_workflow_idx + on workflow_runtime_calls(workflow_run_id); + `); +} + 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 216edc19..e37afe68 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -275,6 +275,76 @@ export const workflowProviderEvents = sqliteTable( ], ); +export const workflowRuntimeRuns = sqliteTable( + "workflow_runtime_runs", + { + id: text("id").primaryKey(), + workspaceId: text("workspace_id").notNull(), + workspaceRoot: text("workspace_root").notNull(), + sourceHash: text("source_hash").notNull(), + argsJson: text("args_json").notNull(), + metadataJson: text("metadata_json").notNull(), + budgetJson: text("budget_json").notNull(), + idempotencyKey: text("idempotency_key"), + requestHash: text("request_hash").notNull(), + status: text("status").notNull(), + resultJson: text("result_json"), + errorJson: text("error_json"), + createdAt: text("created_at").notNull(), + updatedAt: text("updated_at").notNull(), + completedAt: text("completed_at"), + }, + (table) => [ + uniqueIndex("workflow_runtime_runs_idempotency_idx").on( + table.workspaceId, + table.workspaceRoot, + table.idempotencyKey, + ), + index("workflow_runtime_runs_workspace_idx").on( + table.workspaceId, + table.workspaceRoot, + table.createdAt, + ), + ], +); + +export const workflowRuntimeCalls = sqliteTable( + "workflow_runtime_calls", + { + runtimeRunId: text("runtime_run_id") + .notNull() + .references(() => workflowRuntimeRuns.id, { onDelete: "cascade" }), + callIndex: integer("call_index").notNull(), + requestHash: text("request_hash").notNull(), + requestJson: text("request_json").notNull(), + workflowRunId: text("workflow_run_id").references(() => workflowRuns.id, { onDelete: "set null" }), + status: text("status").notNull(), + resultJson: text("result_json"), + errorJson: text("error_json"), + createdAt: text("created_at").notNull(), + updatedAt: text("updated_at").notNull(), + completedAt: text("completed_at"), + }, + (table) => [ + primaryKey({ columns: [table.runtimeRunId, table.callIndex] }), + index("workflow_runtime_calls_workflow_idx").on(table.workflowRunId), + ], +); + +export const workflowRuntimeEvents = sqliteTable( + "workflow_runtime_events", + { + runtimeRunId: text("runtime_run_id") + .notNull() + .references(() => workflowRuntimeRuns.id, { onDelete: "cascade" }), + sequence: integer("sequence").notNull(), + eventType: text("event_type").notNull(), + payloadJson: text("payload_json").notNull(), + createdAt: text("created_at").notNull(), + }, + (table) => [primaryKey({ columns: [table.runtimeRunId, table.sequence] })], +); + export const workflowWorktrees = sqliteTable( "workflow_worktrees", { diff --git a/src/oauth-store.test.ts b/src/oauth-store.test.ts index 59f414e2..1435d44e 100644 --- a/src/oauth-store.test.ts +++ b/src/oauth-store.test.ts @@ -47,6 +47,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise { { version: 4, name: "durable-workflows" }, { version: 5, name: "workflow-supervisor" }, { version: 6, name: "workflow-dag-scheduler" }, + { version: 7, name: "workflow-js-runtime" }, ]); } finally { database.close(); diff --git a/src/workflows/cli.test.ts b/src/workflows/cli.test.ts index b4d87f7d..b7c475fc 100644 --- a/src/workflows/cli.test.ts +++ b/src/workflows/cli.test.ts @@ -17,7 +17,15 @@ mkdirSync(stateDir, { recursive: true }); mkdirSync(agentsDir, { recursive: true }); const profilePath = join(agentsDir, "reviewer.md"); +const workflowScriptPath = join(project, "example.workflow.js"); writeProfile(profilePath, "Original immutable profile."); +writeFileSync(workflowScriptPath, ` +// @devspace-workflow {"version":1,"name":"cli-test","maxAgentCalls":2,"maxConcurrency":1,"timeoutMs":5000} +export default function ({ args, budget, log }) { + log("script executed", { value: args.value }); + return { value: args.value, maxAgentCalls: budget.maxAgentCalls }; +} +`); const baseEnv = { ...process.env, @@ -117,8 +125,39 @@ try { ], baseEnv); assert.notEqual(conflict.status, 0); assert.equal((JSON.parse(conflict.stdout) as Envelope).error!.code, "idempotency_conflict"); + + const scripted = runCli([ + "workflows", "script", workflowScriptPath, "--args-json", JSON.stringify({ value: "from-cli" }), + "--idempotency-key", "script-shell-parent", "--json", + ], baseEnv); + assert.equal(scripted.status, 0, `${scripted.stderr}\n${scripted.stdout}`); + const scriptEnvelope = JSON.parse(scripted.stdout) as Envelope; + assert.equal(scriptEnvelope.ok, true); + assert.equal(scriptEnvelope.created, true); + assert.equal(scriptEnvelope.runtime!.status, "succeeded"); + assert.deepEqual(scriptEnvelope.runtime!.result, { value: "from-cli", maxAgentCalls: 2 }); + + const scriptReplay = runCli([ + "workflows", "script", workflowScriptPath, "--args-json", JSON.stringify({ value: "from-cli" }), + "--idempotency-key", "script-shell-parent", "--json", + ], baseEnv); + assert.equal(scriptReplay.status, 0, scriptReplay.stderr); + assert.equal((JSON.parse(scriptReplay.stdout) as Envelope).created, false); + + const scriptConflict = runCli([ + "workflows", "script", workflowScriptPath, "--args-json", JSON.stringify({ value: "different" }), + "--idempotency-key", "script-shell-parent", "--json", + ], baseEnv); + assert.notEqual(scriptConflict.status, 0); + assert.equal((JSON.parse(scriptConflict.stdout) as Envelope).error!.code, "idempotency_conflict"); + + const invalidScriptArgs = runCli([ + "workflows", "script", workflowScriptPath, "--args-json", "[]", "--json", + ], baseEnv); + assert.notEqual(invalidScriptArgs.status, 0); + assert.equal((JSON.parse(invalidScriptArgs.stdout) as Envelope).error!.code, "invalid_input"); } finally { - rmSync(root, { recursive: true, force: true }); + rmSync(root, { recursive: true, force: true, maxRetries: 50, retryDelay: 100 }); } function runCli(args: string[], env: NodeJS.ProcessEnv) { @@ -157,5 +196,6 @@ interface Envelope { definition: { nodes: Array<{ config: { profileBody: string } }> }; }; events?: Array<{ type: string; sequence: number }>; + runtime?: { status: string; result?: unknown }; error?: { code: string; message: string }; } diff --git a/src/workflows/cli.ts b/src/workflows/cli.ts index e2db7147..f42cfc0b 100644 --- a/src/workflows/cli.ts +++ b/src/workflows/cli.ts @@ -1,4 +1,4 @@ -import { realpath } from "node:fs/promises"; +import { readFile, realpath } from "node:fs/promises"; import { homedir } from "node:os"; import { isAbsolute, join, relative, resolve } from "node:path"; import type { Writable } from "node:stream"; @@ -7,6 +7,7 @@ import { loadLocalAgentProfiles } from "../local-agent-profiles.js"; import { expandHomePath } from "../roots.js"; import { createWorkflowSubmission } from "./submission.js"; import { WorkflowOrchestrator } from "./orchestrator.js"; +import { executeWorkflowRuntime } from "./runtime.js"; import { WorkflowIdempotencyConflictError, WorkflowNotFoundError, @@ -15,6 +16,7 @@ import { import { ensureSupervisor } from "./supervisor-launch.js"; import { runWorkflowSupervisor } from "./supervisor.js"; import type { + JsonObject, WorkflowRunRecord, WorkflowWorkspaceScope, } from "./types.js"; @@ -82,6 +84,8 @@ async function dispatchWorkflowCommand( switch (command) { case "run": return await runCommand(parsed.positionals[0]!, args, context, config, scope, orchestrator); + case "script": + return await scriptCommand(parsed.positionals[0]!, args, context, config, scope, orchestrator); case "status": { const workflowId = parsed.positionals[0]!; return { workflow: orchestrator.getForWorkspace(workflowId, scope) ?? notFound(workflowId) }; @@ -184,6 +188,49 @@ async function runCommand( }; } +async function scriptCommand( + scriptPath: string, + args: string[], + context: WorkflowsCliContext, + config: WorkflowCliConfig, + scope: WorkflowWorkspaceScope, + orchestrator: WorkflowOrchestrator, +): Promise> { + const canonicalScript = await canonicalPath(resolve(context.cwd, scriptPath), "workflow script"); + if (!isInside(canonicalScript, scope.workspaceRoot)) { + throw new WorkspaceDeniedError("Workflow script is outside DEVSPACE_WORKSPACE_ROOT."); + } + if (!canonicalScript.endsWith(".js") && !canonicalScript.endsWith(".mjs")) { + throw new CliInputError("Workflow script must use a .js or .mjs extension."); + } + const source = await readFile(canonicalScript, "utf8"); + const runtimeArgs = parseJsonObjectOption(args, "--args-json"); + const profiles = await loadLocalAgentProfiles(config, scope.workspaceRoot); + const result = await executeWorkflowRuntime({ + stateDir: config.stateDir, + worktreeRoot: config.worktreeRoot, + source, + args: runtimeArgs, + workspace: scope, + profiles, + idempotencyKey: optionalOption(args, "--idempotency-key"), + environment: context.env, + cliEntrypoint: context.cliEntrypoint, + orchestrator, + }); + if (result.run.status !== "succeeded") { + const message = typeof result.run.error?.message === "string" + ? result.run.error.message + : `Workflow runtime ${result.run.id} failed`; + throw new CliRuntimeError(`Workflow runtime ${result.run.id} failed: ${message}`); + } + return { + runtime: result.run, + created: result.created, + replayedCalls: result.replayedCalls, + }; +} + async function resolveWorkspaceScope( context: Pick, allowedRoots: string[], @@ -239,6 +286,10 @@ function validateWorkflowArguments( "--idempotency-key", ]), }, + script: { + positionals: 1, + valueOptions: new Set(["--args-json", "--idempotency-key"]), + }, status: { positionals: 1, valueOptions: new Set() }, events: { positionals: 1, valueOptions: new Set(["--after"]) }, wait: { positionals: 1, valueOptions: new Set(["--timeout-ms", "--after"]) }, @@ -298,6 +349,22 @@ function optionalOption(args: string[], name: string): string | undefined { return undefined; } +function parseJsonObjectOption(args: string[], name: string): JsonObject { + const raw = optionalOption(args, name); + if (raw === undefined) return {}; + try { + const parsed = JSON.parse(raw) as unknown; + if (!parsed || typeof parsed !== "object" || Array.isArray(parsed)) { + throw new Error("value must be a JSON object"); + } + return parsed as JsonObject; + } catch (error) { + throw new CliInputError( + `${name} must be valid JSON object: ${error instanceof Error ? error.message : String(error)}`, + ); + } +} + function integerOption(args: string[], name: string): number | undefined { const raw = optionalOption(args, name); if (raw === undefined) return undefined; @@ -314,6 +381,7 @@ function normalizeCliError(error: unknown): { code: string; message: string; exi if (error instanceof WorkspaceDeniedError) return { code: "workspace_denied", message: error.message, exitCode: 3 }; if (error instanceof WorkflowNotFoundError) return { code: "not_found", message: error.message, exitCode: 4 }; if (error instanceof WorkflowIdempotencyConflictError) return { code: "idempotency_conflict", message: error.message, exitCode: 5 }; + if (error instanceof CliRuntimeError) return { code: "runtime_failed", message: error.message, exitCode: 6 }; if (error instanceof CliInputError || error instanceof WorkflowValidationError) { return { code: "invalid_input", message: error.message, exitCode: 2 }; } @@ -351,6 +419,7 @@ function workflowsHelp(): string { "", "Usage:", " devspace workflows run --prompt --json [--idempotency-key ]", + " devspace workflows script --json [--args-json ] [--idempotency-key ]", " devspace workflows status --json", " devspace workflows wait --json [--timeout-ms ] [--after ]", " devspace workflows events --json [--after ]", @@ -360,4 +429,5 @@ function workflowsHelp(): string { } class CliInputError extends Error {} +class CliRuntimeError extends Error {} class WorkspaceDeniedError extends Error {} diff --git a/src/workflows/migrations.test.ts b/src/workflows/migrations.test.ts index 6178d546..9ecd888e 100644 --- a/src/workflows/migrations.test.ts +++ b/src/workflows/migrations.test.ts @@ -27,6 +27,7 @@ function testFreshSchema(stateDir: string): void { { version: 4, name: "durable-workflows" }, { version: 5, name: "workflow-supervisor" }, { version: 6, name: "workflow-dag-scheduler" }, + { version: 7, name: "workflow-js-runtime" }, ], ); const tables = database.sqlite @@ -44,6 +45,9 @@ function testFreshSchema(stateDir: string): void { "workflow_nodes", "workflow_provider_events", "workflow_runs", + "workflow_runtime_calls", + "workflow_runtime_events", + "workflow_runtime_runs", "workflow_supervisor", "workflow_worktrees", ]); @@ -100,7 +104,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, 6], + [1, 2, 3, 4, 5, 6, 7], ); assert.deepEqual(migrated.sqlite.prepare("select * from workspace_sessions").all(), [ { @@ -173,7 +177,7 @@ function testExistingMigrationCompatibility(stateDir: string): void { .prepare("select count(*) from sqlite_master where type = 'table' and name like 'workflow_%'") .pluck() .get(), - 8, + 11, ); } finally { migrated.close(); @@ -224,6 +228,9 @@ function createVersion3Fixture(stateDir: string): void { const initialized = openDatabase(stateDir); try { initialized.sqlite.exec(` + drop table workflow_runtime_events; + drop table workflow_runtime_calls; + drop table workflow_runtime_runs; drop table workflow_worktrees; drop table workflow_provider_events; drop table workflow_node_attempts; @@ -232,7 +239,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, 6); + delete from devspace_schema_migrations where version in (4, 5, 6, 7); 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/runtime-journal.ts b/src/workflows/runtime-journal.ts new file mode 100644 index 00000000..7f9cd825 --- /dev/null +++ b/src/workflows/runtime-journal.ts @@ -0,0 +1,377 @@ +import { createHash, randomUUID } from "node:crypto"; +import { openDatabase, type DatabaseHandle } from "../db/client.js"; +import type { JsonObject, JsonValue, WorkflowWorkspaceScope } from "./types.js"; +import { WorkflowIdempotencyConflictError, WorkflowValidationError } from "./store.js"; + +export type WorkflowRuntimeStatus = "running" | "succeeded" | "failed" | "cancelled"; +export type WorkflowRuntimeCallStatus = "pending" | "running" | "succeeded" | "failed"; + +export interface WorkflowRuntimeMetadata { + version: 1; + name?: string; + description?: string; +} + +export interface WorkflowRuntimeBudget { + maxAgentCalls: number; + maxConcurrency: number; + timeoutMs: number; +} + +export interface WorkflowRuntimeRun { + id: string; + workspaceId: string; + workspaceRoot: string; + sourceHash: string; + args: JsonObject; + metadata: WorkflowRuntimeMetadata; + budget: WorkflowRuntimeBudget; + idempotencyKey?: string; + requestHash: string; + status: WorkflowRuntimeStatus; + result?: JsonValue; + error?: JsonObject; + createdAt: string; + updatedAt: string; + completedAt?: string; +} + +export interface WorkflowRuntimeCall { + runtimeRunId: string; + callIndex: number; + requestHash: string; + request: JsonObject; + workflowRunId?: string; + status: WorkflowRuntimeCallStatus; + result?: JsonValue; + error?: JsonObject; + createdAt: string; + updatedAt: string; + completedAt?: string; +} + +interface RuntimeRunRow { + id: string; + workspace_id: string; + workspace_root: string; + source_hash: string; + args_json: string; + metadata_json: string; + budget_json: string; + idempotency_key: string | null; + request_hash: string; + status: WorkflowRuntimeStatus; + result_json: string | null; + error_json: string | null; + created_at: string; + updated_at: string; + completed_at: string | null; +} + +interface RuntimeCallRow { + runtime_run_id: string; + call_index: number; + request_hash: string; + request_json: string; + workflow_run_id: string | null; + status: WorkflowRuntimeCallStatus; + result_json: string | null; + error_json: string | null; + created_at: string; + updated_at: string; + completed_at: string | null; +} + +export class WorkflowRuntimeJournal { + private readonly database: DatabaseHandle; + + constructor(stateDir: string) { + this.database = openDatabase(stateDir); + } + + submit(input: { + sourceHash: string; + args: JsonObject; + metadata: WorkflowRuntimeMetadata; + budget: WorkflowRuntimeBudget; + workspace: WorkflowWorkspaceScope; + idempotencyKey?: string; + }): { run: WorkflowRuntimeRun; created: boolean } { + const requestHash = hashJson({ + sourceHash: input.sourceHash, + args: input.args, + metadata: input.metadata, + budget: input.budget, + workspace: input.workspace, + }); + const idempotencyKey = input.idempotencyKey?.trim() || undefined; + const submit = this.database.sqlite.transaction(() => { + if (idempotencyKey) { + const existing = this.database.sqlite + .prepare( + `select * from workflow_runtime_runs + where workspace_id = ? and workspace_root = ? and idempotency_key = ?`, + ) + .get(input.workspace.workspaceId, input.workspace.workspaceRoot, idempotencyKey) as RuntimeRunRow | undefined; + if (existing) { + if (existing.request_hash !== requestHash) { + throw new WorkflowIdempotencyConflictError(idempotencyKey); + } + return { run: rowToRun(existing), created: false }; + } + } + + const id = randomUUID(); + const now = new Date().toISOString(); + this.database.sqlite + .prepare( + `insert into workflow_runtime_runs ( + id, workspace_id, workspace_root, source_hash, args_json, metadata_json, + budget_json, idempotency_key, request_hash, status, created_at, updated_at + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, 'running', ?, ?)`, + ) + .run( + id, + input.workspace.workspaceId, + input.workspace.workspaceRoot, + input.sourceHash, + serialize(input.args), + serialize(input.metadata), + serialize(input.budget), + idempotencyKey ?? null, + requestHash, + now, + now, + ); + return { run: this.require(id), created: true }; + }); + const result = submit.immediate(); + if (result.created) { + this.appendEvent(result.run.id, "runtime.started", { sourceHash: input.sourceHash }); + } + return result; + } + + require(runId: string): WorkflowRuntimeRun { + const row = this.database.sqlite + .prepare("select * from workflow_runtime_runs where id = ?") + .get(runId) as RuntimeRunRow | undefined; + if (!row) throw new WorkflowValidationError(`Unknown workflow runtime run: ${runId}`); + return rowToRun(row); + } + + resume(runId: string): WorkflowRuntimeRun { + const current = this.require(runId); + if (current.status === "succeeded") return current; + const now = new Date().toISOString(); + this.database.sqlite + .prepare( + `update workflow_runtime_runs + set status = 'running', result_json = null, error_json = null, + completed_at = null, updated_at = ? where id = ?`, + ) + .run(now, runId); + this.appendEvent(runId, "runtime.resumed", {}); + return this.require(runId); + } + + beginCall(input: { + runId: string; + callIndex: number; + request: JsonObject; + }): { call: WorkflowRuntimeCall; replayed: boolean } { + if (!Number.isSafeInteger(input.callIndex) || input.callIndex < 0) { + throw new WorkflowValidationError("Workflow runtime call index must be a non-negative integer"); + } + const requestHash = hashJson(input.request); + const begin = this.database.sqlite.transaction(() => { + const existing = this.getCallRow(input.runId, input.callIndex); + if (existing) { + if (existing.request_hash !== requestHash) { + throw new WorkflowValidationError( + `Workflow runtime replay diverged at agent call ${input.callIndex}`, + ); + } + return { call: rowToCall(existing), replayed: true }; + } + const now = new Date().toISOString(); + this.database.sqlite + .prepare( + `insert into workflow_runtime_calls ( + runtime_run_id, call_index, request_hash, request_json, status, created_at, updated_at + ) values (?, ?, ?, ?, 'pending', ?, ?)`, + ) + .run(input.runId, input.callIndex, requestHash, serialize(input.request), now, now); + return { call: rowToCall(this.getCallRow(input.runId, input.callIndex)!), replayed: false }; + }); + const result = begin.immediate(); + if (!result.replayed) { + this.appendEvent(input.runId, "agent.requested", { callIndex: input.callIndex, requestHash }); + } + return result; + } + + markCallRunning(runId: string, callIndex: number, workflowRunId: string): WorkflowRuntimeCall { + const now = new Date().toISOString(); + const result = this.database.sqlite + .prepare( + `update workflow_runtime_calls + set status = 'running', workflow_run_id = ?, updated_at = ? + where runtime_run_id = ? and call_index = ? and status in ('pending', 'running')`, + ) + .run(workflowRunId, now, runId, callIndex); + if (result.changes !== 1) throw new WorkflowValidationError("Workflow runtime call is not runnable"); + this.appendEvent(runId, "agent.started", { callIndex, workflowRunId }); + return this.requireCall(runId, callIndex); + } + + completeCall( + runId: string, + callIndex: number, + status: "succeeded" | "failed", + value: JsonValue | JsonObject, + ): WorkflowRuntimeCall { + const now = new Date().toISOString(); + const resultJson = status === "succeeded" ? serialize(value) : null; + const errorJson = status === "failed" ? serialize(value) : null; + const result = this.database.sqlite + .prepare( + `update workflow_runtime_calls + set status = ?, result_json = ?, error_json = ?, updated_at = ?, completed_at = ? + where runtime_run_id = ? and call_index = ? and status in ('pending', 'running')`, + ) + .run(status, resultJson, errorJson, now, now, runId, callIndex); + if (result.changes !== 1) { + const current = this.requireCall(runId, callIndex); + if (current.status === status) return current; + throw new WorkflowValidationError("Workflow runtime call is already terminal"); + } + this.appendEvent(runId, `agent.${status}`, { callIndex }); + return this.requireCall(runId, callIndex); + } + + requireCall(runId: string, callIndex: number): WorkflowRuntimeCall { + const row = this.getCallRow(runId, callIndex); + if (!row) throw new WorkflowValidationError(`Unknown workflow runtime call: ${callIndex}`); + return rowToCall(row); + } + + appendEvent(runId: string, eventType: string, payload: JsonObject): number { + const append = this.database.sqlite.transaction(() => { + this.require(runId); + const row = this.database.sqlite + .prepare( + "select coalesce(max(sequence), 0) + 1 as sequence from workflow_runtime_events where runtime_run_id = ?", + ) + .get(runId) as { sequence: number }; + this.database.sqlite + .prepare( + `insert into workflow_runtime_events + (runtime_run_id, sequence, event_type, payload_json, created_at) + values (?, ?, ?, ?, ?)`, + ) + .run(runId, row.sequence, eventType, serialize(payload), new Date().toISOString()); + return row.sequence; + }); + return append.immediate(); + } + + completeRun( + runId: string, + status: "succeeded" | "failed" | "cancelled", + value: JsonValue | JsonObject, + ): WorkflowRuntimeRun { + const now = new Date().toISOString(); + const resultJson = status === "succeeded" ? serialize(value) : null; + const errorJson = status === "succeeded" ? null : serialize(value); + const updated = this.database.sqlite + .prepare( + `update workflow_runtime_runs + set status = ?, result_json = ?, error_json = ?, updated_at = ?, completed_at = ? + where id = ? and status = 'running'`, + ) + .run(status, resultJson, errorJson, now, now, runId); + if (updated.changes !== 1) { + const current = this.require(runId); + if (current.status === status) return current; + throw new WorkflowValidationError("Workflow runtime run is already terminal"); + } + this.appendEvent(runId, `runtime.${status}`, {}); + return this.require(runId); + } + + close(): void { + this.database.close(); + } + + private getCallRow(runId: string, callIndex: number): RuntimeCallRow | undefined { + return this.database.sqlite + .prepare( + "select * from workflow_runtime_calls where runtime_run_id = ? and call_index = ?", + ) + .get(runId, callIndex) as RuntimeCallRow | undefined; + } +} + +export function workflowRuntimeSourceHash(source: string): string { + return createHash("sha256").update(source).digest("hex"); +} + +function rowToRun(row: RuntimeRunRow): WorkflowRuntimeRun { + return { + id: row.id, + workspaceId: row.workspace_id, + workspaceRoot: row.workspace_root, + sourceHash: row.source_hash, + args: parseJson(row.args_json), + metadata: parseJson(row.metadata_json), + budget: parseJson(row.budget_json), + idempotencyKey: row.idempotency_key ?? undefined, + requestHash: row.request_hash, + status: row.status, + result: parseOptional(row.result_json), + error: parseOptional(row.error_json) as JsonObject | undefined, + createdAt: row.created_at, + updatedAt: row.updated_at, + completedAt: row.completed_at ?? undefined, + }; +} + +function rowToCall(row: RuntimeCallRow): WorkflowRuntimeCall { + return { + runtimeRunId: row.runtime_run_id, + callIndex: row.call_index, + requestHash: row.request_hash, + request: parseJson(row.request_json), + workflowRunId: row.workflow_run_id ?? undefined, + status: row.status, + result: parseOptional(row.result_json), + error: parseOptional(row.error_json) as JsonObject | undefined, + createdAt: row.created_at, + updatedAt: row.updated_at, + completedAt: row.completed_at ?? undefined, + }; +} + +function hashJson(value: unknown): string { + return createHash("sha256").update(canonicalJson(value)).digest("hex"); +} + +function canonicalJson(value: unknown): string { + if (value === null || typeof value !== "object") return JSON.stringify(value); + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(",")}]`; + const record = value as Record; + return `{${Object.keys(record).sort().map((key) => `${JSON.stringify(key)}:${canonicalJson(record[key])}`).join(",")}}`; +} + +function serialize(value: unknown): string { + return JSON.stringify(value); +} + +function parseJson(value: string): T { + return JSON.parse(value) as T; +} + +function parseOptional(value: string | null): JsonValue | undefined { + return value === null ? undefined : parseJson(value); +} diff --git a/src/workflows/runtime.test.ts b/src/workflows/runtime.test.ts new file mode 100644 index 00000000..daae3d46 --- /dev/null +++ b/src/workflows/runtime.test.ts @@ -0,0 +1,290 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { LocalAgentRunController, type LocalAgentRunHandle } from "../local-agent-runtime.js"; +import { WorkflowOrchestrator } from "./orchestrator.js"; +import { executeWorkflowRuntime, parseWorkflowRuntimeSource } from "./runtime.js"; +import { runWorkflowSupervisor } from "./supervisor.js"; + +const root = mkdtempSync(join(tmpdir(), "devspace-workflow-runtime-test-")); +try { + await testParallelPipelineRuntime(join(root, "parallel")); + await testPrefixReplay(join(root, "replay")); + await testRestrictedGlobalsAndBudget(join(root, "sandbox")); + await testDynamicImportRejected(join(root, "dynamic-import")); + await testUnawaitedAgentFails(join(root, "unawaited")); + testMetadataValidation(); +} finally { + rmSync(root, { recursive: true, force: true }); +} + +async function testParallelPipelineRuntime(stateDir: string): Promise { + const orchestrator = new WorkflowOrchestrator(stateDir); + let active = 0; + let maximum = 0; + const prompts: string[] = []; + const handleFactory = async (_provider: string, input: { prompt: string }): Promise => { + active += 1; + maximum = Math.max(maximum, active); + prompts.push(input.prompt); + const controller = new LocalAgentRunController("fake"); + const timer = setTimeout(() => { + active -= 1; + controller.succeed({ + provider: "fake", + providerSessionId: `runtime-${input.prompt}`, + finalResponse: `result:${input.prompt}`, + items: [], + }); + }, 30); + controller.setLifecycle({ + cancel: () => { + clearTimeout(timer); + active = Math.max(0, active - 1); + }, + dispose: () => clearTimeout(timer), + }); + return controller; + }; + const wakeSupervisor = async () => { + await runWorkflowSupervisor(stateDir, { + handleFactory, + globalConcurrency: 4, + heartbeatMs: 5, + nodeLeaseMs: 500, + supervisorLeaseMs: 500, + idleMs: 0, + }); + }; + const source = ` +// @devspace-workflow {"version":1,"name":"parallel-test","maxAgentCalls":3,"maxConcurrency":2,"timeoutMs":5000} +export default async function ({ agent, parallel, pipeline, phase, log, args }) { + const pair = await phase("research", () => parallel([ + () => agent({ target: "fake", prompt: args.first }), + () => agent({ target: "fake", prompt: args.second }) + ])); + log("research complete", { count: pair.length }); + const final = await pipeline([ + (value) => agent({ target: "fake", prompt: value[0].finalResponse + "+summary" }) + ], pair); + return { pair, final }; +} +`; + try { + const result = await executeWorkflowRuntime({ + stateDir, + worktreeRoot: join(stateDir, "worktrees"), + source, + args: { first: "alpha", second: "beta" }, + workspace: { workspaceId: "workspace", workspaceRoot: root }, + profiles: [], + idempotencyKey: "parallel-runtime", + environment: { ...process.env, DEVSPACE_WORKFLOW_FAKE_PROVIDER: "1" }, + orchestrator, + wakeSupervisor, + }); + assert.equal(result.run.status, "succeeded", JSON.stringify(result.run.error)); + assert.equal(maximum, 2); + assert.deepEqual(prompts.slice(0, 2).sort(), ["alpha", "beta"]); + assert.equal(prompts[2], "result:alpha+summary"); + const output = result.run.result as { final: { finalResponse: string } }; + assert.equal(output.final.finalResponse, "result:result:alpha+summary"); + } finally { + orchestrator.close(); + } +} + +async function testPrefixReplay(stateDir: string): Promise { + const orchestrator = new WorkflowOrchestrator(stateDir); + let starts = 0; + const handleFactory = async (_provider: string, input: { prompt: string }): Promise => { + starts += 1; + const controller = new LocalAgentRunController("fake"); + queueMicrotask(() => controller.succeed({ + provider: "fake", + providerSessionId: `replay-${starts}`, + finalResponse: input.prompt, + items: [], + })); + return controller; + }; + const wakeSupervisor = async () => { + await runWorkflowSupervisor(stateDir, { + handleFactory, + heartbeatMs: 5, + nodeLeaseMs: 500, + supervisorLeaseMs: 500, + idleMs: 0, + }); + }; + const source = ` +// @devspace-workflow {"version":1,"maxAgentCalls":1,"timeoutMs":5000} +export default async function ({ agent }) { + await agent({ target: "fake", prompt: "once" }); + throw new Error("intentional failure after durable prefix"); +} +`; + try { + const input = { + stateDir, + worktreeRoot: join(stateDir, "worktrees"), + source, + args: {}, + workspace: { workspaceId: "workspace", workspaceRoot: root }, + profiles: [], + idempotencyKey: "prefix-replay", + environment: { ...process.env, DEVSPACE_WORKFLOW_FAKE_PROVIDER: "1" }, + orchestrator, + wakeSupervisor, + }; + const first = await executeWorkflowRuntime(input); + assert.equal(first.run.status, "failed"); + assert.equal(starts, 1); + const replay = await executeWorkflowRuntime(input); + assert.equal(replay.run.status, "failed"); + assert.equal(replay.replayedCalls, 1); + assert.equal(starts, 1, "completed prefix must not execute the provider twice"); + } finally { + orchestrator.close(); + } +} + +async function testRestrictedGlobalsAndBudget(stateDir: string): Promise { + const source = ` +// @devspace-workflow {"version":1,"maxAgentCalls":7,"maxConcurrency":3,"timeoutMs":5000} +export default function (api) { + const probes = [ + () => api.agent.constructor("return typeof process")(), + () => api.agent.constructor.constructor("return typeof process")(), + () => api.constructor.constructor("return typeof process")(), + () => Object.getPrototypeOf(api.agent).constructor("return typeof process")(), + () => (async function () {}).constructor("return typeof process")(), + () => (function* () {}).constructor("return typeof process")(), + () => (0, eval)("typeof process") + ].map((probe) => { + try { + return probe(); + } catch { + return "blocked"; + } + }); + return { + args: api.args, + budget: api.budget, + probes, + globals: { + process: typeof process, + require: typeof require, + fetch: typeof fetch, + Buffer: typeof Buffer, + WebSocket: typeof WebSocket + } + }; +} +`; + const result = await executeWorkflowRuntime({ + stateDir, + worktreeRoot: join(stateDir, "worktrees"), + source, + args: { value: "safe" }, + workspace: { workspaceId: "workspace", workspaceRoot: root }, + profiles: [], + }); + assert.equal(result.run.status, "succeeded", JSON.stringify(result.run.error)); + const output = result.run.result as { + args: { value: string }; + budget: { maxAgentCalls: number; maxConcurrency: number; timeoutMs: number }; + probes: string[]; + globals: Record; + }; + assert.deepEqual(output.args, { value: "safe" }); + assert.deepEqual(output.budget, { maxAgentCalls: 7, maxConcurrency: 3, timeoutMs: 5000 }); + assert.ok(output.probes.every((probe) => probe === "undefined" || probe === "blocked")); + assert.deepEqual(output.globals, { + process: "undefined", + require: "undefined", + fetch: "undefined", + Buffer: "undefined", + WebSocket: "undefined", + }); +} + +async function testDynamicImportRejected(stateDir: string): Promise { + const result = await executeWorkflowRuntime({ + stateDir, + worktreeRoot: join(stateDir, "worktrees"), + source: ` +// @devspace-workflow {"version":1,"timeoutMs":5000} +export default async function () { + await import("node:net"); + return "unreachable"; +} +`, + args: {}, + workspace: { workspaceId: "workspace", workspaceRoot: root }, + profiles: [], + }); + assert.equal(result.run.status, "failed"); + assert.match(String(result.run.error?.message), /import expression rejected/i); +} + +async function testUnawaitedAgentFails(stateDir: string): Promise { + const orchestrator = new WorkflowOrchestrator(stateDir); + const handleFactory = async (): Promise => { + const controller = new LocalAgentRunController("fake"); + const timer = setTimeout(() => controller.succeed({ + provider: "fake", + providerSessionId: "late-session", + finalResponse: "late", + items: [], + }), 5_000); + controller.setLifecycle({ + cancel: () => { + clearTimeout(timer); + }, + dispose: () => clearTimeout(timer), + }); + return controller; + }; + try { + const result = await executeWorkflowRuntime({ + stateDir, + worktreeRoot: join(stateDir, "worktrees"), + source: ` +// @devspace-workflow {"version":1,"maxAgentCalls":1,"timeoutMs":5000} +export default function ({ agent }) { + agent({ target: "fake", prompt: "must await" }); + return { premature: true }; +} +`, + args: {}, + workspace: { workspaceId: "workspace", workspaceRoot: root }, + profiles: [], + environment: { ...process.env, DEVSPACE_WORKFLOW_FAKE_PROVIDER: "1" }, + orchestrator, + wakeSupervisor: async () => runWorkflowSupervisor(stateDir, { + handleFactory, + heartbeatMs: 5, + nodeLeaseMs: 500, + supervisorLeaseMs: 500, + idleMs: 0, + }), + }); + assert.equal(result.run.status, "failed"); + assert.match(String(result.run.error?.message), /before all agent\(\) calls were awaited/); + } finally { + orchestrator.close(); + } +} + +function testMetadataValidation(): void { + assert.throws( + () => parseWorkflowRuntimeSource(`// @devspace-workflow {"version":1,"unknown":true}\nexport default function () {}`), + /Unknown workflow metadata field/, + ); + assert.throws( + () => parseWorkflowRuntimeSource("export const workflow = () => null"), + /default function declaration/, + ); +} diff --git a/src/workflows/runtime.ts b/src/workflows/runtime.ts new file mode 100644 index 00000000..f211d651 --- /dev/null +++ b/src/workflows/runtime.ts @@ -0,0 +1,699 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { readFileSync, realpathSync } from "node:fs"; +import { dirname, join } from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import type { LocalAgentProfile } from "../local-agent-profiles.js"; +import { createWorkflowSubmission, type WorkflowAgentIntent } from "./submission.js"; +import { WorkflowOrchestrator } from "./orchestrator.js"; +import { ensureSupervisor } from "./supervisor-launch.js"; +import { + WorkflowRuntimeJournal, + workflowRuntimeSourceHash, + type WorkflowRuntimeBudget, + type WorkflowRuntimeMetadata, + type WorkflowRuntimeRun, +} from "./runtime-journal.js"; +import { WorkflowValidationError } from "./store.js"; +import type { JsonObject, JsonValue, WorkflowRunRecord, WorkflowWorkspaceScope } from "./types.js"; + +const MAX_SOURCE_BYTES = 256 * 1024; +const MAX_ARGS_BYTES = 64 * 1024; +const DEFAULT_MAX_AGENT_CALLS = 16; +const MAX_AGENT_CALLS = 64; +const DEFAULT_MAX_CONCURRENCY = 4; +const MAX_CONCURRENCY = 16; +const DEFAULT_RUNTIME_TIMEOUT_MS = 15 * 60_000; +const MAX_RUNTIME_TIMEOUT_MS = 24 * 60 * 60_000; +const MAX_LOG_CHARS = 16_384; +const MAX_EVENT_DATA_BYTES = 64 * 1024; +const CHILD_WORKFLOW_WAIT_MS = 300_000; + +export interface ParsedWorkflowRuntimeSource { + metadata: WorkflowRuntimeMetadata; + budget: WorkflowRuntimeBudget; + code: string; +} + +export interface ExecuteWorkflowRuntimeInput { + stateDir: string; + worktreeRoot: string; + source: string; + args: JsonObject; + workspace: WorkflowWorkspaceScope; + profiles: LocalAgentProfile[]; + idempotencyKey?: string; + environment?: NodeJS.ProcessEnv; + cliEntrypoint?: string; + orchestrator?: WorkflowOrchestrator; + wakeSupervisor?: () => Promise; +} + +export interface ExecuteWorkflowRuntimeResult { + run: WorkflowRuntimeRun; + created: boolean; + replayedCalls: number; +} + +interface RuntimeAgentMessage { + type: "agent"; + requestId: number; + callIndex: number; + request: unknown; +} + +interface RuntimeEventMessage { + type: "event"; + eventType: unknown; + payload: unknown; +} + +interface RuntimeResultMessage { + type: "result"; + result: unknown; +} + +interface RuntimeFailureMessage { + type: "failure"; + error: unknown; +} + +type RuntimeChildMessage = RuntimeAgentMessage | RuntimeEventMessage | RuntimeResultMessage | RuntimeFailureMessage; + +export function parseWorkflowRuntimeSource(source: string): ParsedWorkflowRuntimeSource { + if (Buffer.byteLength(source, "utf8") > MAX_SOURCE_BYTES) { + throw new WorkflowValidationError(`Workflow script exceeds ${MAX_SOURCE_BYTES} bytes`); + } + const lines = source.replace(/^/, "").split(/\r?\n/); + const firstContentIndex = lines.findIndex((line) => line.trim().length > 0); + let header: Record = {}; + if (firstContentIndex >= 0 && lines[firstContentIndex]!.trim().startsWith("// @devspace-workflow")) { + const line = lines[firstContentIndex]!.trim(); + const raw = line.slice("// @devspace-workflow".length).trim(); + if (!raw) throw new WorkflowValidationError("Workflow metadata comment must contain JSON"); + try { + const parsed = JSON.parse(raw) as unknown; + if (!isPlainObject(parsed)) throw new Error("metadata must be an object"); + header = parsed; + } catch (error) { + throw new WorkflowValidationError( + `Invalid workflow metadata JSON: ${error instanceof Error ? error.message : String(error)}`, + ); + } + lines.splice(firstContentIndex, 1); + } + + const allowed = new Set([ + "version", + "name", + "description", + "maxAgentCalls", + "maxConcurrency", + "timeoutMs", + ]); + for (const key of Object.keys(header)) { + if (!allowed.has(key)) throw new WorkflowValidationError(`Unknown workflow metadata field: ${key}`); + } + const version = header.version ?? 1; + if (version !== 1) throw new WorkflowValidationError("Workflow metadata version must be 1"); + const name = optionalBoundedString(header.name, "Workflow name", 128); + const description = optionalBoundedString(header.description, "Workflow description", 1_024); + const metadata: WorkflowRuntimeMetadata = { + version: 1, + ...(name ? { name } : {}), + ...(description ? { description } : {}), + }; + const budget: WorkflowRuntimeBudget = { + maxAgentCalls: boundedInteger( + header.maxAgentCalls, + DEFAULT_MAX_AGENT_CALLS, + 1, + MAX_AGENT_CALLS, + "Workflow maxAgentCalls", + ), + maxConcurrency: boundedInteger( + header.maxConcurrency, + DEFAULT_MAX_CONCURRENCY, + 1, + MAX_CONCURRENCY, + "Workflow maxConcurrency", + ), + timeoutMs: boundedInteger( + header.timeoutMs, + DEFAULT_RUNTIME_TIMEOUT_MS, + 1, + MAX_RUNTIME_TIMEOUT_MS, + "Workflow timeoutMs", + ), + }; + const code = lines.join("\n").trim(); + if (!/^export\s+default\s+(?:async\s+)?function\b/.test(code)) { + throw new WorkflowValidationError( + "Workflow script must export a default function declaration", + ); + } + return { metadata, budget, code }; +} + +export async function executeWorkflowRuntime( + input: ExecuteWorkflowRuntimeInput, +): Promise { + validateJsonObject(input.args, "Workflow args", MAX_ARGS_BYTES); + const parsed = parseWorkflowRuntimeSource(input.source); + const journal = new WorkflowRuntimeJournal(input.stateDir); + const orchestrator = input.orchestrator ?? new WorkflowOrchestrator(input.stateDir); + const ownsOrchestrator = !input.orchestrator; + const submitted = journal.submit({ + sourceHash: workflowRuntimeSourceHash(parsed.code), + args: input.args, + metadata: parsed.metadata, + budget: parsed.budget, + workspace: input.workspace, + idempotencyKey: input.idempotencyKey, + }); + if (!submitted.created && submitted.run.status === "succeeded") { + journal.close(); + if (ownsOrchestrator) orchestrator.close(); + return { run: submitted.run, created: false, replayedCalls: 0 }; + } + const run = submitted.created ? submitted.run : journal.resume(submitted.run.id); + const activeWorkflowIds = new Set(); + const scheduledOperations = new Set>(); + let aborting = false; + let replayedCalls = 0; + let child: ChildProcess | undefined; + let timeout: NodeJS.Timeout | undefined; + try { + const finalResult = await new Promise((resolve, reject) => { + let settled = false; + let activeAgents = 0; + const queuedAgents: Array<() => void> = []; + const settle = (action: () => void) => { + if (settled) return; + settled = true; + if (timeout) clearTimeout(timeout); + action(); + }; + const runQueued = () => { + while (!settled && activeAgents < parsed.budget.maxConcurrency && queuedAgents.length > 0) { + activeAgents += 1; + queuedAgents.shift()!(); + } + }; + const scheduleAgent = (operation: () => Promise) => { + queuedAgents.push(() => { + const scheduled = operation().finally(() => { + activeAgents -= 1; + scheduledOperations.delete(scheduled); + runQueued(); + }); + scheduledOperations.add(scheduled); + void scheduled; + }); + runQueued(); + }; + + child = spawnRuntimeChild(); + child.once("error", (error) => settle(() => reject(error))); + child.once("exit", (code, signal) => { + if (!settled) { + settle(() => reject(new Error( + `Workflow runtime child exited before returning a result (${signal ? `signal ${signal}` : `exit code ${code ?? "unknown"}`})`, + ))); + } + }); + child.on("message", (raw: unknown) => { + void handleChildMessage(raw).catch((error) => settle(() => reject(error))); + }); + timeout = setTimeout(() => { + settle(() => reject(new WorkflowValidationError( + `Workflow runtime exceeded ${parsed.budget.timeoutMs} milliseconds`, + ))); + }, parsed.budget.timeoutMs); + + child.send({ + type: "start", + source: parsed.code, + args: input.args, + budget: parsed.budget, + }); + + async function handleChildMessage(raw: unknown): Promise { + if (!isPlainObject(raw) || typeof raw.type !== "string") { + throw new WorkflowValidationError("Workflow runtime child sent an invalid message"); + } + const message = raw as unknown as RuntimeChildMessage; + if (message.type === "agent") { + if (!Number.isSafeInteger(message.requestId) || Number(message.requestId) < 0) { + throw new WorkflowValidationError("Workflow runtime request ID is invalid"); + } + if (!Number.isSafeInteger(message.callIndex) || Number(message.callIndex) < 0) { + throw new WorkflowValidationError("Workflow runtime call index is invalid"); + } + if (message.callIndex >= parsed.budget.maxAgentCalls) { + sendChild(child, { + type: "agent_result", + requestId: message.requestId, + ok: false, + error: { code: "budget_exceeded", message: "Workflow agent-call budget was exceeded" }, + }); + return; + } + scheduleAgent(async () => { + try { + const result = await executeAgentCall(message.callIndex, message.request); + sendChild(child, { type: "agent_result", requestId: message.requestId, ok: true, result }); + } catch (error) { + sendChild(child, { + type: "agent_result", + requestId: message.requestId, + ok: false, + error: normalizeError(error), + }); + } + }); + return; + } + if (message.type === "event") { + const eventType = requiredBoundedString(message.eventType, "Runtime event type", 64); + if (eventType !== "phase.started" && eventType !== "phase.completed" && eventType !== "log") { + throw new WorkflowValidationError(`Unsupported workflow runtime event: ${eventType}`); + } + const payload = normalizeEventPayload(message.payload); + journal.appendEvent(run.id, eventType, payload); + return; + } + if (message.type === "result") { + const result = normalizeJsonValue(message.result, "Workflow result"); + settle(() => resolve(result)); + return; + } + if (message.type === "failure") { + const error = normalizeChildError(message.error); + settle(() => reject(new WorkflowValidationError(error.message))); + return; + } + throw new WorkflowValidationError("Workflow runtime child sent an unsupported message"); + } + + async function executeAgentCall(callIndex: number, rawRequest: unknown): Promise { + const request = normalizeAgentRequest(rawRequest); + const journaled = journal.beginCall({ runId: run.id, callIndex, request: request as unknown as JsonObject }); + if (journaled.replayed) replayedCalls += 1; + if (journaled.call.status === "succeeded") return journaled.call.result ?? null; + if (journaled.call.status === "failed") { + throw new WorkflowValidationError( + typeof journaled.call.error?.message === "string" + ? journaled.call.error.message + : `Workflow agent call ${callIndex} previously failed`, + ); + } + + let workflow: WorkflowRunRecord; + if (journaled.call.workflowRunId) { + workflow = orchestrator.getForWorkspace(journaled.call.workflowRunId, input.workspace) + ?? (() => { throw new WorkflowValidationError("Journaled child workflow is unavailable"); })(); + } else { + const submission = await createWorkflowSubmission({ + intent: { + single: request, + idempotencyKey: `runtime:${run.id}:agent:${callIndex}`, + }, + workspace: input.workspace, + profiles: input.profiles, + worktreeRoot: input.worktreeRoot, + environment: input.environment, + }); + const childWorkflow = orchestrator.submitDetailed(submission).workflow; + journal.markCallRunning(run.id, callIndex, childWorkflow.id); + workflow = childWorkflow; + await wakeSupervisor(input); + } + activeWorkflowIds.add(workflow.id); + if (aborting && !isTerminal(workflow)) { + orchestrator.cancelForWorkspace(workflow.id, input.workspace); + await wakeSupervisor(input); + } + while (!isTerminal(workflow)) { + workflow = await orchestrator.waitForWorkspace(workflow.id, input.workspace, { + timeoutMs: Math.min(CHILD_WORKFLOW_WAIT_MS, parsed.budget.timeoutMs), + }); + } + activeWorkflowIds.delete(workflow.id); + if (workflow.status !== "succeeded") { + const error = { + code: stringField(workflow.error, "code") ?? workflow.status, + message: stringField(workflow.error, "message") ?? `Child workflow ${workflow.id} ${workflow.status}`, + workflowId: workflow.id, + } satisfies JsonObject; + journal.completeCall(run.id, callIndex, "failed", error); + throw new WorkflowValidationError(String(error.message)); + } + const result = { + workflowId: workflow.id, + status: workflow.status, + finalResponse: stringField(workflow.result as JsonObject | undefined, "finalResponse") ?? "", + } satisfies JsonObject; + journal.completeCall(run.id, callIndex, "succeeded", result); + return result; + } + }); + child?.kill("SIGTERM"); + return { + run: journal.completeRun(run.id, "succeeded", finalResult), + created: submitted.created, + replayedCalls, + }; + } catch (error) { + aborting = true; + child?.kill("SIGKILL"); + for (const workflowId of activeWorkflowIds) { + try { + orchestrator.cancelForWorkspace(workflowId, input.workspace); + } catch { + // The child may have terminalized while the runtime was aborting. + } + } + if (activeWorkflowIds.size > 0 || scheduledOperations.size > 0) { + await wakeSupervisor(input).catch(() => undefined); + } + await Promise.allSettled([...scheduledOperations]); + const normalized = normalizeError(error); + return { + run: journal.completeRun(run.id, "failed", normalized), + created: submitted.created, + replayedCalls, + }; + } finally { + if (timeout) clearTimeout(timeout); + journal.close(); + if (ownsOrchestrator) orchestrator.close(); + } +} + +async function wakeSupervisor(input: ExecuteWorkflowRuntimeInput): Promise { + if (input.wakeSupervisor) return input.wakeSupervisor(); + return ensureSupervisor({ + stateDir: input.stateDir, + cliEntrypoint: input.cliEntrypoint, + env: input.environment, + }); +} + +function spawnRuntimeChild(): ChildProcess { + const permissionArgs = runtimeDependencyRoots().map((root) => `--allow-fs-read=${root}`); + const sesEntrypoint = pathToFileURL(realpathSync(fileURLToPath(import.meta.resolve("ses")))).href; + const childSource = RUNTIME_CHILD_SOURCE.replace( + "DEVSPACE_SES_ENTRYPOINT", + JSON.stringify(sesEntrypoint), + ); + return spawn( + process.execPath, + [ + "--permission", + ...permissionArgs, + "--max-old-space-size=64", + "--input-type=module", + "--eval", + childSource, + ], + { + stdio: ["ignore", "ignore", "ignore", "ipc"], + env: {}, + shell: false, + windowsHide: true, + }, + ); +} + +function runtimeDependencyRoots(): string[] { + return [ + "ses", + "@endo/cache-map", + "@endo/env-options", + "@endo/immutable-arraybuffer", + ].map(findPackageRoot); +} + +function findPackageRoot(specifier: string): string { + let current = dirname(fileURLToPath(import.meta.resolve(specifier))); + while (true) { + try { + const manifest = JSON.parse(readFileSync(join(current, "package.json"), "utf8")) as { + name?: unknown; + }; + if (manifest.name === specifier) return realpathSync(current); + } catch { + // Continue through package-internal directories until the package root is found. + } + const parent = dirname(current); + if (parent === current) throw new Error(`Unable to resolve runtime dependency '${specifier}'.`); + current = parent; + } +} + +function sendChild(child: ChildProcess | undefined, message: JsonObject): void { + if (!child?.connected) return; + child.send(message); +} + +function normalizeAgentRequest(value: unknown): WorkflowAgentIntent { + if (!isPlainObject(value)) throw new WorkflowValidationError("agent() requires an options object"); + const allowed = new Set(["target", "prompt", "model", "thinking", "access", "timeoutMs", "retry"]); + for (const key of Object.keys(value)) { + if (!allowed.has(key)) throw new WorkflowValidationError(`Unknown agent() option: ${key}`); + } + const target = requiredBoundedString(value.target, "agent target", 128); + const prompt = requiredBoundedString(value.prompt, "agent prompt", 200_000); + const model = optionalBoundedString(value.model, "agent model", 256); + const thinking = optionalBoundedString(value.thinking, "agent thinking", 64); + const access = value.access === undefined ? undefined : requiredBoundedString(value.access, "agent access", 32); + if (access !== undefined && access !== "read_only" && access !== "workspace_write") { + throw new WorkflowValidationError("agent access must be read_only or workspace_write"); + } + const timeoutMs = value.timeoutMs === undefined + ? undefined + : boundedInteger(value.timeoutMs, 1, 1, MAX_RUNTIME_TIMEOUT_MS, "agent timeoutMs"); + let retry: WorkflowAgentIntent["retry"]; + if (value.retry !== undefined) { + if (!isPlainObject(value.retry)) throw new WorkflowValidationError("agent retry must be an object"); + const retryAllowed = new Set(["maxAttempts", "retryOn", "backoffMs"]); + for (const key of Object.keys(value.retry)) { + if (!retryAllowed.has(key)) throw new WorkflowValidationError(`Unknown agent retry option: ${key}`); + } + const retryOn = value.retry.retryOn; + if (retryOn !== undefined && (!Array.isArray(retryOn) || retryOn.some( + (entry) => entry !== "provider_failed" && entry !== "timed_out", + ))) { + throw new WorkflowValidationError("agent retryOn contains an unsupported failure class"); + } + retry = { + maxAttempts: value.retry.maxAttempts === undefined + ? undefined + : boundedInteger(value.retry.maxAttempts, 1, 1, 10, "agent retry maxAttempts"), + retryOn: retryOn as Array<"provider_failed" | "timed_out"> | undefined, + backoffMs: value.retry.backoffMs === undefined + ? undefined + : boundedInteger(value.retry.backoffMs, 0, 0, 60_000, "agent retry backoffMs"), + }; + } + return { target, prompt, model, thinking, access, timeoutMs, retry }; +} + +function normalizeEventPayload(value: unknown): JsonObject { + if (!isPlainObject(value)) throw new WorkflowValidationError("Workflow runtime event payload must be an object"); + const serialized = JSON.stringify(value); + if (Buffer.byteLength(serialized, "utf8") > MAX_EVENT_DATA_BYTES) { + throw new WorkflowValidationError("Workflow runtime event payload is too large"); + } + if (typeof value.message === "string" && value.message.length > MAX_LOG_CHARS) { + throw new WorkflowValidationError(`Workflow runtime log exceeds ${MAX_LOG_CHARS} characters`); + } + return JSON.parse(serialized) as JsonObject; +} + +function normalizeJsonValue(value: unknown, label: string): JsonValue { + const serialized = JSON.stringify(value); + if (serialized === undefined) return null; + if (Buffer.byteLength(serialized, "utf8") > MAX_EVENT_DATA_BYTES) { + throw new WorkflowValidationError(`${label} exceeds ${MAX_EVENT_DATA_BYTES} bytes`); + } + return JSON.parse(serialized) as JsonValue; +} + +function validateJsonObject(value: JsonObject, label: string, maxBytes: number): void { + if (!isPlainObject(value)) throw new WorkflowValidationError(`${label} must be an object`); + const serialized = JSON.stringify(value); + if (Buffer.byteLength(serialized, "utf8") > maxBytes) { + throw new WorkflowValidationError(`${label} exceeds ${maxBytes} bytes`); + } +} + +function normalizeError(error: unknown): JsonObject { + return { + code: error instanceof WorkflowValidationError ? "invalid_workflow" : "runtime_failed", + message: error instanceof Error ? error.message.slice(0, MAX_LOG_CHARS) : String(error).slice(0, MAX_LOG_CHARS), + }; +} + +function normalizeChildError(error: unknown): { message: string } { + if (!isPlainObject(error) || typeof error.message !== "string") { + return { message: "Workflow script failed" }; + } + return { message: error.message.slice(0, MAX_LOG_CHARS) }; +} + +function requiredBoundedString(value: unknown, label: string, maximum: number): string { + if (typeof value !== "string" || !value.trim()) throw new WorkflowValidationError(`${label} is required`); + const normalized = value.trim(); + if (normalized.length > maximum) throw new WorkflowValidationError(`${label} exceeds ${maximum} characters`); + return normalized; +} + +function optionalBoundedString(value: unknown, label: string, maximum: number): string | undefined { + if (value === undefined || value === null) return undefined; + return requiredBoundedString(value, label, maximum); +} + +function boundedInteger( + value: unknown, + fallback: number, + minimum: number, + maximum: number, + label: string, +): 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 isPlainObject(value: unknown): value is Record { + return Boolean(value) && typeof value === "object" && !Array.isArray(value); +} + +function isTerminal(workflow: WorkflowRunRecord): boolean { + return workflow.status === "succeeded" || workflow.status === "failed" || workflow.status === "cancelled"; +} + +function stringField(value: JsonObject | undefined, key: string): string | undefined { + const field = value?.[key]; + return typeof field === "string" ? field : undefined; +} + +const RUNTIME_CHILD_SOURCE = String.raw` +import DEVSPACE_SES_ENTRYPOINT; + +lockdown(); +if (typeof Compartment !== "function" || typeof harden !== "function") { + throw new Error("SES lockdown did not initialize the workflow runtime"); +} + +let nextRequestId = 0; +let nextCallIndex = 0; +const pending = new Map(); + +process.on("message", (message) => { + if (!message || typeof message !== "object") return; + if (message.type === "agent_result") { + const entry = pending.get(message.requestId); + if (!entry) return; + pending.delete(message.requestId); + if (message.ok) { + entry.resolve(copyJson(message.result, "Agent result")); + } else { + entry.reject(new Error( + typeof message.error?.message === "string" ? message.error.message : "Agent call failed", + )); + } + return; + } + if (message.type === "start") void start(message); +}); + +function rpcAgent(request) { + const requestId = nextRequestId++; + const callIndex = nextCallIndex++; + const safeRequest = copyJson(request, "Agent request"); + return new Promise((resolve, reject) => { + pending.set(requestId, { resolve, reject }); + process.send({ type: "agent", requestId, callIndex, request: safeRequest }); + }); +} + +function emit(eventType, payload) { + process.send({ type: "event", eventType, payload: copyJson(payload, "Runtime event") }); +} + +async function start(message) { + try { + const transformed = message.source.replace( + /^export\s+default\s+/, + "globalThis.__devspaceWorkflow = ", + ); + const compartment = new Compartment(); + compartment.evaluate('"use strict";\n' + transformed); + const workflow = compartment.globalThis.__devspaceWorkflow; + delete compartment.globalThis.__devspaceWorkflow; + if (typeof workflow !== "function") throw new Error("Default export must be a function"); + + const api = harden({ + args: copyJson(message.args, "Workflow args"), + budget: copyJson(message.budget, "Workflow budget"), + agent: (request) => rpcAgent(request), + parallel: async (tasks) => { + if (!Array.isArray(tasks)) throw new Error("parallel() requires an array of functions"); + return Promise.all(tasks.map((task) => { + if (typeof task !== "function") throw new Error("parallel() entries must be functions"); + return task(); + })); + }, + pipeline: async (tasks, initialValue) => { + if (!Array.isArray(tasks)) throw new Error("pipeline() requires an array of functions"); + let value = initialValue; + for (const task of tasks) { + if (typeof task !== "function") throw new Error("pipeline() entries must be functions"); + value = await task(value); + } + return value; + }, + phase: async (name, task) => { + if (typeof name !== "string" || !name.trim()) throw new Error("phase() requires a name"); + if (typeof task !== "function") throw new Error("phase() requires a function"); + emit("phase.started", { name: name.trim() }); + const result = await task(); + emit("phase.completed", { name: name.trim() }); + return result; + }, + log: (message, data = null) => { + if (typeof message !== "string") throw new Error("log() requires a string message"); + emit("log", { message, data }); + }, + }); + const result = await workflow(api); + if (pending.size > 0) { + throw new Error("Workflow returned before all agent() calls were awaited"); + } + process.send({ type: "result", result: copyJson(result === undefined ? null : result, "Workflow result") }); + } catch (error) { + process.send({ + type: "failure", + error: { + name: typeof error?.name === "string" ? error.name : "Error", + message: typeof error?.message === "string" ? error.message : String(error), + }, + }); + } +} + +function copyJson(value, label) { + let serialized; + try { + serialized = JSON.stringify(value); + } catch { + throw new Error(label + " must be JSON serializable"); + } + if (serialized === undefined) return null; + return harden(JSON.parse(serialized)); +} +`;