Add durable workflow supervisor and CLI

docs/workflow-cli.md

@@ -0,0 +1,66 @@
+# Workflow CLI
+
+DevSpace workflows are durable, single-agent runs intended for shell parent processes. Submission stores an immutable execution snapshot and wakes a detached supervisor. The workflow continues after the submitting shell exits.
+
+## Workspace identity
+
+Every command requires a workspace identity and canonical workspace root. The root must be inside `DEVSPACE_ALLOWED_ROOTS`.
+
+```sh
+export DEVSPACE_WORKSPACE_ID="checkout-42"
+export DEVSPACE_WORKSPACE_ROOT="$(git rev-parse --show-toplevel)"
+export DEVSPACE_ALLOWED_ROOTS="$HOME/src"
+```
+
+A workflow ID does not grant access by itself. Status, events, wait, and cancellation must use the same workspace ID and canonical root used at submission.
+
+## Submit and wait
+
+`run --json` writes exactly one versioned JSON object to stdout after the request is durable and supervisor wakeup has been requested.
+
+```sh
+accepted="$({
+  devspace workflows run reviewer \
+    --prompt "Review the current change" \
+    --idempotency-key "review-$GITHUB_SHA" \
+    --json
+})"
+workflow_id="$(node -e 'process.stdout.write(JSON.parse(process.argv[1]).workflow.id)' "$accepted")"
+
+# The submitting process may exit here. Another process can wait later.
+result="$(devspace workflows wait "$workflow_id" --timeout-ms 300000 --after 0 --json)"
+status="$(node -e 'process.stdout.write(JSON.parse(process.argv[1]).workflow.status)' "$result")"
+```
+
+`wait` is bounded, repeatable, non-destructive, and exits zero when the timeout expires or when the workflow reaches `failed` or `cancelled`. Inspect `timedOut` and `workflow.status` in the JSON response.
+
+The maximum wait timeout is 300000 milliseconds. Use the returned `cursor` with `--after` to replay only newer events:
+
+```sh
+devspace workflows events "$workflow_id" --after "$cursor" --json
+devspace workflows wait "$workflow_id" --after "$cursor" --timeout-ms 300000 --json
+```
+
+## Other commands
+
+```sh
+devspace workflows status "$workflow_id" --json
+devspace workflows cancel "$workflow_id" --json
+```
+
+Cancellation is idempotent. Pending work is cancelled without dispatch; active providers receive cooperative cancellation and process cleanup through their existing provider handles.
+
+## Submission options
+
+```text
+--prompt <task>                 required; no positional prompt is accepted
+--idempotency-key <key>         optional durable deduplication key
+--model <model>                 optional provider model override
+--thinking <setting>            optional provider thinking override
+--access read_only|workspace_write
+--timeout-ms <milliseconds>     optional node execution timeout
+```
+
+The persisted snapshot includes the resolved profile body and hash, profile name, provider, model, thinking setting, effective access/environment policy, canonical workspace root, prompt, and timeout. Later profile-file changes do not affect an accepted workflow.
+
+JSON errors use the same versioned envelope and a nonzero exit code. Diagnostics go to stderr; prompts and provider output are never written to diagnostic logs.

package.json

@@ -29,7 +29,7 @@
     "postinstall": "node scripts/fix-node-pty-permissions.mjs",
     "start": "node dist/cli.js serve",
     "test": "tsx src/config.test.ts && tsx src/ui/card-types.test.ts && tsx src/ui/patch-display.test.ts && tsx src/ui/tool-display.test.ts && tsx src/apply-patch.test.ts && tsx src/process-platform.test.ts && tsx src/process-sessions.test.ts && tsx src/mcp-sessions.test.ts && tsx src/server-shutdown.test.ts && tsx src/local-agent-runtime.test.ts && tsx src/local-agent-adapters.test.ts && tsx src/local-agent-availability.test.ts && tsx src/local-agent-profiles.test.ts && tsx src/local-agent-targets.test.ts && tsx src/local-agent-store.test.ts && tsx src/roots.test.ts && tsx src/skills.test.ts && tsx src/workspaces.test.ts && tsx src/review-checkpoints.test.ts && tsx src/oauth-store.test.ts && npm run test:workflows && tsx src/cli.test.ts",
-    "test:workflows": "tsx src/workflows/migrations.test.ts && tsx src/workflows/policy.test.ts && tsx src/workflows/workflows.test.ts",
+    "test:workflows": "tsx src/workflows/migrations.test.ts && tsx src/workflows/policy.test.ts && tsx src/workflows/workflows.test.ts && tsx src/workflows/supervisor.test.ts && tsx src/workflows/cli.test.ts",
     "typecheck": "tsc -p tsconfig.json --noEmit"
   },
   "keywords": [],

src/cli.ts

@@ -1,6 +1,6 @@
 #!/usr/bin/env node
 import { createRequire } from "node:module";
-import { stdin as input, stdout as output } from "node:process";
+import { stderr as errorOutput, stdin as input, stdout as output } from "node:process";
 import { spawn } from "node:child_process";
 import { mkdtempSync, writeFileSync } from "node:fs";
 import { readFile } from "node:fs/promises";
@@ -39,8 +39,9 @@ import {
 } from "./user-config.js";
 import { expandHomePath } from "./roots.js";
 import { shutdownHttpServer } from "./server-shutdown.js";
+import { runWorkflowsCli } from "./workflows/cli.js";
 
-type Command = "serve" | "init" | "doctor" | "config" | "agents" | "help" | "version";
+type Command = "serve" | "init" | "doctor" | "config" | "agents" | "workflows" | "help" | "version";
 const require = createRequire(import.meta.url);
 const SUPPORTED_NODE_RANGE = ">=20.12 <27";
 
@@ -67,6 +68,16 @@ async function main(argv: string[]): Promise<void> {
     case "agents":
       await runAgentsCommand(args);
       return;
+    case "workflows":
+      process.exitCode = await runWorkflowsCli({
+        argv: args,
+        cwd: process.cwd(),
+        env: process.env,
+        stdout: output,
+        stderr: errorOutput,
+        cliEntrypoint: fileURLToPath(import.meta.url),
+      });
+      return;
     case "help":
       printHelp();
       return;
@@ -78,7 +89,7 @@ async function main(argv: string[]): Promise<void> {
 
 function normalizeCommand(command: string | undefined): Command {
   if (!command || command === "serve" || command === "start") return "serve";
-  if (command === "init" || command === "doctor" || command === "config" || command === "agents") return command;
+  if (command === "init" || command === "doctor" || command === "config" || command === "agents" || command === "workflows") return command;
   if (command === "help" || command === "--help" || command === "-h") return "help";
   if (command === "version" || command === "--version" || command === "-v") return "version";
   throw new Error(`Unknown command: ${command}`);
@@ -312,6 +323,8 @@ function printHelp(): void {
       "  devspace agents ls       List subagent sessions",
       "  devspace agents run <profile-or-provider-or-id> [--model <model>] <prompt>",
       "  devspace agents show <id>",
+      "  devspace workflows run <profile-or-provider> --prompt <task> --json",
+      "  devspace workflows help",
       "  devspace -v, --version   Print the installed version",
       "",
       "For temporary tunnels:",

src/db/client.ts

@@ -24,9 +24,9 @@ export function openDatabase(stateDir: string): DatabaseHandle {
   const path = databasePath(stateDir);
   const sqlite = new Database(path);
   chmodSync(path, 0o600);
+  sqlite.pragma("busy_timeout = 5000");
   sqlite.pragma("journal_mode = WAL");
   sqlite.pragma("synchronous = NORMAL");
-  sqlite.pragma("busy_timeout = 5000");
   sqlite.pragma("foreign_keys = ON");
   migrateDatabase(sqlite);
 

src/db/migrations.ts

@@ -27,6 +27,11 @@ const migrations: Migration[] = [
     name: "durable-workflows",
     up: migrateDurableWorkflows,
   },
+  {
+    version: 5,
+    name: "workflow-supervisor",
+    up: migrateWorkflowSupervisor,
+  },
 ];
 
 export function migrateDatabase(sqlite: Database.Database): void {
@@ -264,9 +269,82 @@ function migrateDurableWorkflows(sqlite: Database.Database): void {
   `);
 }
 
+function migrateWorkflowSupervisor(sqlite: Database.Database): void {
+  addColumnIfMissing(sqlite, "workflow_runs", "workspace_id", "text");
+  addColumnIfMissing(sqlite, "workflow_runs", "workspace_root", "text");
+  addColumnIfMissing(sqlite, "workflow_nodes", "supervisor_owner_token", "text");
+  addColumnIfMissing(sqlite, "workflow_nodes", "supervisor_owner_epoch", "integer");
+  addColumnIfMissing(sqlite, "workflow_nodes", "heartbeat_at", "text");
+
+  sqlite.exec(`
+    create table if not exists workflow_supervisor (
+      id integer primary key check (id = 1),
+      owner_token text,
+      owner_epoch integer not null default 0,
+      owner_pid integer,
+      status text not null default 'stopped'
+        check (status in ('stopped', 'starting', 'running', 'stopping')),
+      lease_expires_at text,
+      heartbeat_at text,
+      wake_generation integer not null default 0,
+      started_at text,
+      last_error text
+    );
+
+    insert or ignore into workflow_supervisor (id) values (1);
+
+    create table if not exists workflow_node_attempts (
+      node_id text not null,
+      workflow_run_id text not null,
+      node_key text not null,
+      attempt integer not null,
+      claim_token text not null,
+      supervisor_owner_token text not null,
+      supervisor_owner_epoch integer not null,
+      provider text not null,
+      phase text not null check (phase in ('claimed', 'dispatching', 'running', 'cancelling', 'terminal')),
+      provider_session_id text,
+      heartbeat_at text,
+      cancellation_requested_at text,
+      terminal_status text check (terminal_status in ('succeeded', 'failed', 'cancelled')),
+      result_json text,
+      error_json text,
+      created_at text not null,
+      updated_at text not null,
+      completed_at text,
+      primary key (node_id, attempt),
+      unique (node_id, attempt, claim_token),
+      foreign key (workflow_run_id, node_id)
+        references workflow_nodes(workflow_run_id, id) on delete cascade
+    );
+
+    create index if not exists workflow_node_attempts_run_idx
+      on workflow_node_attempts(workflow_run_id, node_key, attempt);
+
+    create table if not exists workflow_provider_events (
+      node_id text not null,
+      attempt integer not null,
+      source_sequence integer not null,
+      workflow_sequence integer not null,
+      event_type text not null,
+      payload_json text not null,
+      created_at text not null,
+      primary key (node_id, attempt, source_sequence),
+      foreign key (node_id, attempt)
+        references workflow_node_attempts(node_id, attempt) on delete cascade
+    );
+
+    create index if not exists workflow_nodes_claimable_idx
+      on workflow_nodes(status, claim_expires_at, created_at);
+
+    create index if not exists workflow_runs_workspace_idx
+      on workflow_runs(workspace_id, workspace_root, created_at);
+  `);
+}
+
 function addColumnIfMissing(
   sqlite: Database.Database,
-  table: "workspace_sessions" | "local_agent_sessions",
+  table: "workspace_sessions" | "local_agent_sessions" | "workflow_runs" | "workflow_nodes",
   column: string,
   definition: string,
 ): void {

src/db/schema.ts

@@ -116,6 +116,8 @@ export const workflowRuns = sqliteTable(
     policyJson: text("policy_json").notNull(),
     idempotencyKey: text("idempotency_key").unique(),
     requestHash: text("request_hash").notNull(),
+    workspaceId: text("workspace_id"),
+    workspaceRoot: text("workspace_root"),
     resultJson: text("result_json"),
     errorJson: text("error_json"),
     cancellationRequestedAt: text("cancellation_requested_at"),
@@ -143,6 +145,9 @@ export const workflowNodes = sqliteTable(
     claimToken: text("claim_token"),
     claimedAt: text("claimed_at"),
     claimExpiresAt: text("claim_expires_at"),
+    supervisorOwnerToken: text("supervisor_owner_token"),
+    supervisorOwnerEpoch: integer("supervisor_owner_epoch"),
+    heartbeatAt: text("heartbeat_at"),
     resultJson: text("result_json"),
     errorJson: text("error_json"),
     createdAt: text("created_at").notNull(),
@@ -201,6 +206,72 @@ export const workflowEvents = sqliteTable(
   ],
 );
 
+export const workflowSupervisor = sqliteTable("workflow_supervisor", {
+  id: integer("id").primaryKey(),
+  ownerToken: text("owner_token"),
+  ownerEpoch: integer("owner_epoch").notNull().default(0),
+  ownerPid: integer("owner_pid"),
+  status: text("status").notNull().default("stopped"),
+  leaseExpiresAt: text("lease_expires_at"),
+  heartbeatAt: text("heartbeat_at"),
+  wakeGeneration: integer("wake_generation").notNull().default(0),
+  startedAt: text("started_at"),
+  lastError: text("last_error"),
+});
+
+export const workflowNodeAttempts = sqliteTable(
+  "workflow_node_attempts",
+  {
+    nodeId: text("node_id").notNull(),
+    workflowRunId: text("workflow_run_id").notNull(),
+    nodeKey: text("node_key").notNull(),
+    attempt: integer("attempt").notNull(),
+    claimToken: text("claim_token").notNull(),
+    supervisorOwnerToken: text("supervisor_owner_token").notNull(),
+    supervisorOwnerEpoch: integer("supervisor_owner_epoch").notNull(),
+    provider: text("provider").notNull(),
+    phase: text("phase").notNull(),
+    providerSessionId: text("provider_session_id"),
+    heartbeatAt: text("heartbeat_at"),
+    cancellationRequestedAt: text("cancellation_requested_at"),
+    terminalStatus: text("terminal_status"),
+    resultJson: text("result_json"),
+    errorJson: text("error_json"),
+    createdAt: text("created_at").notNull(),
+    updatedAt: text("updated_at").notNull(),
+    completedAt: text("completed_at"),
+  },
+  (table) => [
+    primaryKey({ columns: [table.nodeId, table.attempt] }),
+    uniqueIndex("workflow_node_attempts_claim_idx").on(table.nodeId, table.attempt, table.claimToken),
+    foreignKey({
+      columns: [table.workflowRunId, table.nodeId],
+      foreignColumns: [workflowNodes.workflowRunId, workflowNodes.id],
+    }).onDelete("cascade"),
+    index("workflow_node_attempts_run_idx").on(table.workflowRunId, table.nodeKey, table.attempt),
+  ],
+);
+
+export const workflowProviderEvents = sqliteTable(
+  "workflow_provider_events",
+  {
+    nodeId: text("node_id").notNull(),
+    attempt: integer("attempt").notNull(),
+    sourceSequence: integer("source_sequence").notNull(),
+    workflowSequence: integer("workflow_sequence").notNull(),
+    eventType: text("event_type").notNull(),
+    payloadJson: text("payload_json").notNull(),
+    createdAt: text("created_at").notNull(),
+  },
+  (table) => [
+    primaryKey({ columns: [table.nodeId, table.attempt, table.sourceSequence] }),
+    foreignKey({
+      columns: [table.nodeId, table.attempt],
+      foreignColumns: [workflowNodeAttempts.nodeId, workflowNodeAttempts.attempt],
+    }).onDelete("cascade"),
+  ],
+);
+
 export type WorkspaceSessionRow = typeof workspaceSessions.$inferSelect;
 export type NewWorkspaceSessionRow = typeof workspaceSessions.$inferInsert;
 export type LoadedAgentFileRow = typeof loadedAgentFiles.$inferSelect;

src/local-agent-profiles.ts

@@ -43,7 +43,7 @@ const FRONTMATTER_DELIMITER = "---";
 const PROVIDERS = new Set<LocalAgentProvider>(LOCAL_AGENT_PROVIDERS);
 
 export async function loadLocalAgentProfiles(
-  config: ServerConfig,
+  config: Pick<ServerConfig, "subagents" | "devspaceAgentsDir">,
   workspaceRoot: string,
 ): Promise<LocalAgentProfile[]> {
   if (!config.subagents) return [];

src/oauth-store.test.ts

@@ -45,6 +45,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise<void> {
       { version: 2, name: "oauth-state" },
       { version: 3, name: "local-agent-sessions" },
       { version: 4, name: "durable-workflows" },
+      { version: 5, name: "workflow-supervisor" },
     ]);
   } finally {
     database.close();