diff --git a/packages/core/src/__tests__/resume-fallback.test.ts b/packages/core/src/__tests__/resume-fallback.test.ts index 657e1e6..213288b 100644 --- a/packages/core/src/__tests__/resume-fallback.test.ts +++ b/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(); diff --git a/packages/core/src/__tests__/run-persistence.test.ts b/packages/core/src/__tests__/run-persistence.test.ts new file mode 100644 index 0000000..da515f9 --- /dev/null +++ b/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(); + }); +}); diff --git a/packages/core/src/builder.ts b/packages/core/src/builder.ts index c560911..32b6c12 100644 --- a/packages/core/src/builder.ts +++ b/packages/core/src/builder.ts @@ -152,6 +152,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. */ @@ -571,7 +573,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; @@ -587,7 +589,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 { @@ -599,7 +601,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); diff --git a/packages/core/src/run.ts b/packages/core/src/run.ts index b01276e..6f00805 100644 --- a/packages/core/src/run.ts +++ b/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 { + 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; diff --git a/packages/core/src/runner.ts b/packages/core/src/runner.ts index ee17545..f85588e 100644 --- a/packages/core/src/runner.ts +++ b/packages/core/src/runner.ts @@ -116,7 +116,7 @@ import type { WorkflowStepStatus, ProcessBackend, RunnerStepExecutor, -} from './types.js'; + ResumeOptions,} from './types.js'; import { WorkflowTrajectory, type StepOutcome } from './trajectory.js'; import { runVerification, @@ -3635,7 +3635,13 @@ export class WorkflowRunner { } /** Resume a previously paused or partially completed run. */ - async resume(runId: string, vars?: VariableContext, config?: RelayYamlConfig): Promise { + async resume( + runId: string, + vars?: VariableContext, + config?: RelayYamlConfig, + options?: ResumeOptions + ): Promise { + const resetRunningSteps = options?.resetRunningSteps ?? false; // Set up abort controller early so callers can abort() even during setup this.abortController = new AbortController(); this.paused = false; @@ -3685,16 +3691,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(), }); } diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index 0879a8e..59a5195 100644 --- a/packages/core/src/types.ts +++ b/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;