Add typed MCP workflow tools and bounded DAG scheduling

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.

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": [],

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<typeof createDrizzleDatabase>;
 
@@ -23,18 +27,39 @@ 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);
-
-  return {
-    sqlite,
-    db: createDrizzleDatabase(sqlite),
-    close: () => sqlite.close(),
-  };
+  try {
+    chmodSync(path, 0o600);
+    sqlite.pragma("busy_timeout = 5000");
+    enableWriteAheadLogging(sqlite);
+    sqlite.pragma("synchronous = NORMAL");
+    sqlite.pragma("foreign_keys = ON");
+    migrateDatabase(sqlite);
+
+    return {
+      sqlite,
+      db: createDrizzleDatabase(sqlite),
+      close: () => sqlite.close(),
+    };
+  } catch (error) {
+    sqlite.close();
+    throw error;
+  }
+}
+
+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) {

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",

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;

src/oauth-store.test.ts

@@ -46,6 +46,7 @@ async function testDatabaseConfiguration(stateDir: string): Promise<void> {
       { 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();

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<void>;
   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")));

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 });
+}