fix(workflow): supervise cancellation and stale workers

Author: Waishnav

src/server.ts

@@ -48,6 +48,7 @@ import { createWorkspaceStore } from "./workspace-store.js";
 import { formatAgentsPath, WorkspaceRegistry } from "./workspaces.js";
 import { summarizeLocalAgentProfile } from "./local-agent-profiles.js";
 import { registerWorkflowTools } from "./workflow-tools.js";
+import { startWorkflowReaper } from "./workflow-lifecycle.js";
 import { createWorkflowStore } from "./workflow-store.js";
 import { loadActiveWorkflowSummaries } from "./workflow-ui.js";
 import {
@@ -1659,6 +1660,15 @@ export function createServer(config = loadConfig()): RunningServer {
   const localAgentProviders = config.subagents
     ? getLocalAgentProviderAvailabilitySnapshot()
     : [];
+  const workflowReaper = config.subagents
+    ? startWorkflowReaper(config, {
+        onError: (error) => {
+          logEvent(config.logging, "warn", "workflow_reaper_failed", {
+            error: error instanceof Error ? error.message : String(error),
+          });
+        },
+      })
+    : undefined;
 
   const logSessionCloseResults = (
     reason: "idle_timeout" | "server_shutdown",
@@ -1846,6 +1856,7 @@ export function createServer(config = loadConfig()): RunningServer {
     close: () => {
       closePromise ??= (async () => {
         clearInterval(sessionCleanupTimer);
+        workflowReaper?.close();
         const results = await transports.closeAll();
         logSessionCloseResults("server_shutdown", results);
         processSessions.shutdown();

src/workflow-cli.ts

@@ -21,10 +21,13 @@ import {
   resolveWorkflowScriptFromPathOrNameResult,
 } from "./workflow-files.js";
 import { createWorkflowReplay } from "./workflow-replay.js";
+import {
+  cancelWorkflowRun,
+  reapStaleWorkflows,
+} from "./workflow-lifecycle.js";
 import { parseWorkflowScript } from "./workflow-script.js";
 import { createWorkflowStore, type WorkflowStore } from "./workflow-store.js";
 import {
-  WORKFLOW_CANCEL_HARD_MS,
   WORKFLOW_HEARTBEAT_MS,
   WORKFLOW_LIMITS,
   resolveWorkflowConcurrency,
@@ -234,6 +237,7 @@ async function runWorkflowStatus(args: string[], config: ServerConfig): Promise<
 
   const store = createWorkflowStore(config);
   try {
+    reapStaleWorkflows(store);
     const runResult = store.getRunResult(runId);
     if (runResult.isErr()) throw runResult.error;
     const run = runResult.value;
@@ -256,36 +260,8 @@ async function runWorkflowCancel(args: string[], config: ServerConfig): Promise<
   if (!runId) throw new Error("Usage: devspace workflow cancel <runId>");
   const store = createWorkflowStore(config);
   try {
-    const requested = store.requestCancelResult(runId);
-    if (requested.isErr()) throw requested.error;
-    const run = requested.value;
-    console.log(formatRunLine(run));
-    if (run.pid && (run.status === "running" || run.status === "starting")) {
-      try {
-        process.kill(run.pid, "SIGTERM");
-      } catch {
-        // already dead
-      }
-      await sleep(WORKFLOW_CANCEL_HARD_MS);
-      const again = store.getRun(runId);
-      if (again && (again.status === "running" || again.status === "starting") && again.pid) {
-        try {
-          process.kill(-again.pid, "SIGKILL");
-        } catch {
-          try {
-            process.kill(again.pid, "SIGKILL");
-          } catch {
-            // gone
-          }
-        }
-        const latest = store.getRun(runId);
-        if (latest && (latest.status === "running" || latest.status === "starting")) {
-          const cancelled = store.cancelRunResult(runId, "cancelled (hard kill)");
-          if (cancelled.isErr()) throw cancelled.error;
-        }
-      }
-    }
-    console.log(formatRunLine(store.getRun(runId)!));
+    reapStaleWorkflows(store);
+    console.log(formatRunLine(await cancelWorkflowRun(store, runId)));
   } finally {
     store.close();
   }
@@ -294,6 +270,7 @@ async function runWorkflowCancel(args: string[], config: ServerConfig): Promise<
 async function runWorkflowList(config: ServerConfig): Promise<void> {
   const store = createWorkflowStore(config);
   try {
+    reapStaleWorkflows(store);
     const runs = store.listRuns(50);
     if (runs.length === 0) {
       console.log("No workflow runs.");

src/workflow-store.test.ts

@@ -193,6 +193,25 @@ try {
   assert.equal(store.getRun(run3.id)?.errorKind, "heartbeat");
   assert.equal(store.listEvents(run3.id).at(-1)?.type, "run_failed");
 
+  const runStarting = store.createRun({
+    name: "never-started",
+    source: "inline",
+    scriptPath: join(root, "never.js"),
+    scriptHash: "never",
+    workspaceRoot: join(root, "project"),
+  });
+  const staleStartingDb = openDatabase(root);
+  try {
+    staleStartingDb.sqlite
+      .prepare(`update workflow_runs set updated_at = ? where id = ?`)
+      .run(new Date(Date.now() - 120_000).toISOString(), runStarting.id);
+  } finally {
+    staleStartingDb.close();
+  }
+  const reapedStarting = store.reapStale(60_000);
+  assert.ok(reapedStarting.some((entry) => entry.id === runStarting.id));
+  assert.equal(store.getRun(runStarting.id)?.status, "failed");
+
   const run4 = store.createRun({
     name: "seq",
     source: "inline",

src/workflow-store.ts

@@ -749,27 +749,26 @@ export class WorkflowStore {
   }
 
   /**
-   * Mark running runs with a dead worker as failed.
-   * staleBeforeMs: heartbeat older than this AND pid not alive.
+   * Mark abandoned starting runs and running runs with a dead worker as failed.
+   * staleBeforeMs: start/update or heartbeat older than this and no live pid.
    */
   reapStale(staleBeforeMs = 60_000, nowMs = Date.now()): WorkflowRunRecord[] {
     const cutoff = new Date(nowMs - staleBeforeMs).toISOString();
     const candidates = this.database.sqlite
       .prepare(
         `select * from workflow_runs
-         where status = 'running'
-           and heartbeat_at is not null
-           and heartbeat_at < ?`,
+         where (status = 'running' and heartbeat_at is not null and heartbeat_at < ?)
+            or (status = 'starting' and updated_at < ?)`,
       )
-      .all(cutoff) as WorkflowRunRow[];
+      .all(cutoff, cutoff) as WorkflowRunRow[];
 
     const reaped: WorkflowRunRecord[] = [];
     for (const row of candidates) {
       if (row.pid !== null && isPidAlive(row.pid)) continue;
       const latest = this.getRun(row.id);
-      if (!latest || latest.status !== "running") continue;
+      if (!latest || (latest.status !== "running" && latest.status !== "starting")) continue;
       const failed = this.failRun(row.id, {
-        error: "worker heartbeat lost",
+        error: latest.status === "starting" ? "workflow worker failed to start" : "worker heartbeat lost",
         errorKind: "heartbeat",
       });
       if (failed.status === "failed" && failed.errorKind === "heartbeat") {

src/workflow-tools.ts

@@ -21,6 +21,7 @@ import {
 } from "./workflow-types.js";
 import { resolveWorkspaceHead } from "./workflow-worktrees.js";
 import { spawnWorkflowWorkerFromCli } from "./workflow-cli.js";
+import { cancelWorkflowRun } from "./workflow-lifecycle.js";
 import { getLocalAgentProviderAvailabilitySnapshot } from "./local-agent-availability.js";
 import {
   isLocalAgentProvider,
@@ -268,17 +269,7 @@ export function registerWorkflowTools(
     async ({ runId }) => {
       const store = createWorkflowStore(config);
       try {
-        const requested = store.requestCancelResult(runId);
-        if (requested.isErr()) throw requested.error;
-        const run = requested.value;
-        if (run.pid && (run.status === "running" || run.status === "starting")) {
-          try {
-            process.kill(run.pid, "SIGTERM");
-          } catch {
-            // already gone
-          }
-        }
-        const latest = store.getRun(runId)!;
+        const latest = await cancelWorkflowRun(store, runId);
         return {
           content: [{ type: "text" as const, text: JSON.stringify({ runId, status: latest.status }) }],
           structuredContent: { runId, status: latest.status },