fix(core): make workflow runs durable across resumes

.github/workflows/ci.yml

@@ -0,0 +1,34 @@
+name: CI
+
+on:
+  pull_request:
+
+permissions:
+  contents: read
+
+jobs:
+  test:
+    name: Test and typecheck
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Setup Node.js
+        uses: actions/setup-node@v4
+        with:
+          node-version: '22'
+          cache: npm
+
+      - name: Install dependencies
+        run: npm ci
+
+      - name: Build primitives
+        run: npm run build:primitives
+
+      - name: Test
+        run: npm test
+
+      - name: Typecheck
+        run: npm run typecheck

package.json

@@ -14,7 +14,7 @@
   "scripts": {
     "build:primitives": "npm run build --workspace=packages/github-primitive --workspace=packages/slack-primitive --workspace=packages/browser-primitive",
     "build": "npm run build:primitives && npm run build --workspace=packages/core && npm run build --workspace=packages/cli",
-    "typecheck": "npm run build:primitives && npm run typecheck --workspace=packages/core && npm run typecheck --workspace=packages/cli",
+    "typecheck": "npm run build:primitives && npm run build --workspace=packages/core && npm run typecheck --workspace=packages/core && npm run typecheck --workspace=packages/cli",
     "test": "npm run test --workspace=packages/core"
   },
   "devDependencies": {

packages/core/src/__tests__/resume-fallback.test.ts

@@ -262,6 +262,33 @@ describe('resume fallback to step-output cache', () => {
     expect(startedSteps).toContain('step-c');
   });
 
+  it('should reset stale running steps to pending and re-execute them', async () => {
+    const runId = 'resume-stale-running-run';
+    const config = makeResumeConfig();
+
+    await db.insertRun(makeRunRow(runId, config, 'running'));
+    await db.insertStep(makeStepRow(runId, 'step-a', 'Do step A', [], 'running'));
+    await db.insertStep(makeStepRow(runId, 'step-b', 'Do step B', ['step-a'], 'pending'));
+    await db.insertStep(makeStepRow(runId, 'step-c', 'Do step C', ['step-b'], 'pending'));
+
+    const events: Array<{ type: string; stepName?: string }> = [];
+    runner.on((event) => {
+      if ('stepName' in event) {
+        events.push({ type: event.type, stepName: event.stepName });
+      }
+    });
+
+    const run = await runner.resume(runId, undefined, undefined, { resetRunningSteps: true });
+    expect(run.status, run.error).toBe('completed');
+
+    expect(db.updateStep).toHaveBeenCalledWith(
+      `${runId}-step-a`,
+      expect.objectContaining({ status: 'pending', error: undefined, completionReason: undefined })
+    );
+    const startedSteps = events.filter((event) => event.type === 'step:started').map((event) => event.stepName);
+    expect(startedSteps).toContain('step-a');
+  });
+
   it('should handle empty step-output directory gracefully', async () => {
     const runId = 'resume-empty-cache';
     const config = makeResumeConfig();

packages/core/src/__tests__/run-persistence.test.ts

@@ -0,0 +1,81 @@
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { mkdtempSync, rmSync, writeFileSync } from 'node:fs';
+import os from 'node:os';
+import path from 'node:path';
+import { workflow } from '../builder.js';
+import { JsonFileWorkflowDb } from '../file-db.js';
+import { InMemoryWorkflowDb } from '../memory-db.js';
+import { runWorkflow } from '../run.js';
+import { WorkflowRunner } from '../runner.js';
+import type { WorkflowRunRow } from '../types.js';
+
+describe('workflow run persistence', () => {
+  const tmpDirs: string[] = [];
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+    for (const tmpDir of tmpDirs.splice(0)) {
+      rmSync(tmpDir, { recursive: true, force: true });
+    }
+  });
+
+  it('constructs runWorkflow with the cwd JSONL database', async () => {
+    const tmpDir = mkdtempSync(path.join(os.tmpdir(), 'run-persistence-'));
+    tmpDirs.push(tmpDir);
+    const yamlPath = path.join(tmpDir, 'relay.yaml');
+    writeFileSync(
+      yamlPath,
+      [
+        'version: "1"',
+        'name: run-persistence-test',
+        'swarm:',
+        '  pattern: sequential',
+        'agents: []',
+        'workflows:',
+        '  - name: default',
+        '    steps:',
+        '      - name: noop',
+        '        type: deterministic',
+        '        command: "true"',
+      ].join('\n')
+    );
+    const dryRunSpy = vi.spyOn(WorkflowRunner.prototype, 'dryRun');
+    vi.spyOn(console, 'log').mockImplementation(() => {});
+
+    await runWorkflow(yamlPath, { cwd: tmpDir, dryRun: true });
+
+    const runner = dryRunSpy.mock.instances[0] as unknown as { db: unknown };
+    expect(runner.db).toBeInstanceOf(JsonFileWorkflowDb);
+    expect((runner.db as JsonFileWorkflowDb).getStoragePath()).toBe(
+      path.join(tmpDir, '.agent-relay', 'workflow-runs.jsonl')
+    );
+    expect(runner.db).not.toBeInstanceOf(InMemoryWorkflowDb);
+  });
+
+  it('honors WorkflowRunOptions.resume before executing a new run', async () => {
+    const tmpDir = mkdtempSync(path.join(os.tmpdir(), 'builder-resume-'));
+    tmpDirs.push(tmpDir);
+    const resumedRun = { id: 'resume-id', status: 'completed' } as WorkflowRunRow;
+    const resumeSpy = vi.spyOn(WorkflowRunner.prototype, 'resume').mockResolvedValue(resumedRun);
+    const executeSpy = vi.spyOn(WorkflowRunner.prototype, 'execute').mockResolvedValue(resumedRun);
+
+    const result = await workflow('builder-resume-test')
+      .agent('agent-a', { cli: 'claude' })
+      .step('step-a', { agent: 'agent-a', task: 'Do step A' })
+      .run({ cwd: tmpDir, renderer: false, resume: 'resume-id' });
+
+    expect(result).toBe(resumedRun);
+    // The third arg is the parsed config: resume() feeds it to
+    // reconstructRunFromCache() when workflow-runs.jsonl is absent, so dropping
+    // it silently disables the cached-step-output fallback.
+    expect(resumeSpy).toHaveBeenCalledWith(
+      'resume-id',
+      undefined,
+      expect.objectContaining({ name: 'builder-resume-test' }),
+      // User-facing resume means "the previous process is gone", so the builder
+      // opts in to requeueing steps left running. The library default is off.
+      expect.objectContaining({ resetRunningSteps: true })
+    );
+    expect(executeSpy).not.toHaveBeenCalled();
+  });
+});

packages/core/src/builder.ts

@@ -165,6 +165,8 @@ export interface WorkflowRunOptions {
   dryRun?: boolean;
   /** External step executor (e.g. Daytona sandbox backend). */
   executor?: RunnerStepExecutor;
+  /** Resume a failed run by its ID instead of starting fresh. */
+  resume?: string;
   /** Start from a specific step, skipping all predecessors. */
   startFrom?: string;
   /** Previous run ID whose cached outputs are used with startFrom. */
@@ -616,7 +618,7 @@ export class WorkflowBuilder {
     }
 
     // Auto-detect RESUME_RUN_ID env var for resuming failed runs
-    const resumeRunId = process.env.RESUME_RUN_ID;
+    const resumeRunId = options.resume ?? process.env.RESUME_RUN_ID;
 
     const startFrom = this._startFrom ?? options.startFrom ?? process.env.START_FROM;
     const previousRunId = this._previousRunId ?? options.previousRunId ?? process.env.PREVIOUS_RUN_ID;
@@ -632,7 +634,7 @@ export class WorkflowBuilder {
       runner.on(renderer.onEvent);
 
       const runPromise = resumeRunId
-        ? runner.resume(resumeRunId, options.vars, config)
+        ? runner.resume(resumeRunId, options.vars, config, { resetRunningSteps: true })
         : runner.execute(config, options.workflow, options.vars, executeOptions);
 
       try {
@@ -644,7 +646,7 @@ export class WorkflowBuilder {
     }
 
     if (resumeRunId) {
-      return runner.resume(resumeRunId, options.vars, config);
+      return runner.resume(resumeRunId, options.vars, config, { resetRunningSteps: true });
     }
 
     return runner.execute(config, options.workflow, options.vars, executeOptions);

packages/core/src/run.ts

@@ -1,5 +1,7 @@
+import path from 'node:path';
 import type { RuntimeSpawnOptions } from '@agent-relay/harness-driver';
 import type { DryRunReport, TrajectoryConfig, WorkflowRunRow } from './types.js';
+import { JsonFileWorkflowDb } from './file-db.js';
 import { WorkflowRunner, type WorkflowEventListener } from './runner.js';
 import { createDefaultEventLogger } from './default-logger.js';
 import { formatDryRunReport } from './dry-run-format.js';
@@ -51,9 +53,12 @@ export async function runWorkflow(
   yamlPath: string,
   options: RunWorkflowOptions = {}
 ): Promise<WorkflowRunRow | DryRunReport> {
+  const dbPath = path.join(options.cwd ?? process.cwd(), '.agent-relay', 'workflow-runs.jsonl');
+  const db = new JsonFileWorkflowDb(dbPath);
   const runner = new WorkflowRunner({
     cwd: options.cwd,
     relay: options.relay,
+    db,
   });
 
   const config = await runner.parseYamlFile(yamlPath);
@@ -83,7 +88,7 @@ export async function runWorkflow(
   // Resume a previous run if requested
   const resumeRunId = options.resume ?? process.env.RESUME_RUN_ID;
   if (resumeRunId) {
-    return runner.resume(resumeRunId, options.vars);
+    return runner.resume(resumeRunId, options.vars, config, { resetRunningSteps: true });
   }
 
   const startFrom = options.startFrom ?? process.env.START_FROM;

packages/core/src/runner.ts

@@ -122,7 +122,7 @@ import type {
   WorkflowStepStatus,
   ProcessBackend,
   RunnerStepExecutor,
-} from './types.js';
+  ResumeOptions,} from './types.js';
 import { WorkflowTrajectory, type StepOutcome } from './trajectory.js';
 import {
   activateWorkflowPersona,
@@ -295,6 +295,16 @@ function sleepMs(ms: number): Promise<void> {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
 
+function closeWriteStream(stream: WriteStream): Promise<void> {
+  if (stream.closed) return Promise.resolve();
+
+  return new Promise((resolve) => {
+    const settle = () => resolve();
+    stream.once('error', settle);
+    stream.end(settle);
+  });
+}
+
 // ── DB adapter interface ────────────────────────────────────────────────────
 
 /** Minimal DB adapter so the runner is not coupled to a specific driver. */
@@ -3976,7 +3986,13 @@ export class WorkflowRunner {
   }
 
   /** Resume a previously paused or partially completed run. */
-  async resume(runId: string, vars?: VariableContext, config?: RelayYamlConfig): Promise<WorkflowRunRow> {
+  async resume(
+    runId: string,
+    vars?: VariableContext,
+    config?: RelayYamlConfig,
+    options?: ResumeOptions
+  ): Promise<WorkflowRunRow> {
+    const resetRunningSteps = options?.resetRunningSteps ?? false;
     // Set up abort controller early so callers can abort() even during setup
     this.abortController = new AbortController();
     this.paused = false;
@@ -4026,16 +4042,28 @@ export class WorkflowRunner {
       }
     }
 
-    // Reset failed steps to pending for retry
+    // Reset steps to pending so they are retried.
+    //
+    // `failed` is always safe to requeue. `running` is only safe when no other
+    // process is still executing the step: there is no lease/heartbeat on runs
+    // today, so we cannot detect a live owner. Requeueing blindly would let a
+    // second `resume` re-run steps concurrently with the original process and
+    // duplicate non-idempotent side effects. It is therefore opt-in via
+    // `resetRunningSteps`, which the user-facing resume paths set because
+    // `--resume` explicitly means "the previous process is gone".
     for (const [, state] of stepStates) {
-      if (state.row.status === 'failed') {
+      const isFailed = state.row.status === 'failed';
+      const isStaleRunning = state.row.status === 'running' && resetRunningSteps;
+      if (isFailed || isStaleRunning) {
         state.row.status = 'pending';
         state.row.error = undefined;
         state.row.completionReason = undefined;
+        state.row.retryCount = 0;
         await this.db.updateStep(state.row.id, {
           status: 'pending',
           error: undefined,
           completionReason: undefined,
+          retryCount: 0,
           updatedAt: new Date().toISOString(),
         });
       }
@@ -7646,7 +7674,7 @@ export class WorkflowRunner {
         combined: combinedOutput,
       });
       stopHeartbeat?.();
-      logStream.end();
+      await closeWriteStream(logStream);
       this.unregisterWorker(agentName);
     }
   }
@@ -7848,7 +7876,7 @@ export class WorkflowRunner {
         const newLogPath = path.join(logsDir, `${agent.name}.log`);
         const oldLogStream = this.ptyLogStreams.get(oldName);
         if (oldLogStream) {
-          oldLogStream.end();
+          await closeWriteStream(oldLogStream);
           this.ptyLogStreams.delete(oldName);
           try {
             renameSync(oldLogPath, newLogPath);
@@ -8036,7 +8064,7 @@ export class WorkflowRunner {
       this.ptyListeners.delete(agentName);
       const stream = this.ptyLogStreams.get(agentName);
       if (stream) {
-        stream.end();
+        await closeWriteStream(stream);
         this.ptyLogStreams.delete(agentName);
       }
       this.unregisterWorker(agentName);

packages/core/src/types.ts

@@ -278,6 +278,20 @@ export interface PreflightCheck {
   description?: string;
 }
 
+/** Options for {@link WorkflowRunner.resume}. */
+export interface ResumeOptions {
+  /**
+   * Requeue steps left in `running` when the run stopped.
+   *
+   * Off by default. Runs carry no lease or heartbeat, so a live owner cannot be
+   * detected; requeueing blindly lets a second resume re-run steps alongside the
+   * original process and duplicate non-idempotent side effects. The user-facing
+   * resume paths set this because `--resume` explicitly means the previous
+   * process is gone.
+   */
+  resetRunningSteps?: boolean;
+}
+
 /** A named workflow composed of sequential or parallel steps. */
 export interface WorkflowDefinition {
   name: string;