fix(core): Suspend jobs that register after worker shutdown asked for suspension

packages/cli/src/scaling/__tests__/job-processor.service.test.ts

@@ -327,6 +327,50 @@ describe('JobProcessor', () => {
 			await processPromise;
 		});
 
+		it('should suspend a suspendable job that registers after suspension was requested', async () => {
+			const executionPersistence = mock<ExecutionPersistence>();
+			let resolveExecution!: (execution: IExecutionResponse) => void;
+			executionPersistence.findSingleExecution.mockReturnValue(
+				new Promise<IExecutionResponse>((r) => (resolveExecution = r)),
+			);
+			vi.spyOn(WorkflowExecuteAdditionalData, 'getBase').mockResolvedValue(
+				mock<IWorkflowExecuteAdditionalData>(),
+			);
+			const jobProcessor = createJobProcessor(executionPersistence);
+
+			let resolveRun!: (run: IRun) => void;
+			processRunExecutionDataMock.mockReturnValue(new Promise<IRun>((r) => (resolveRun = r)));
+
+			const job = mock<Job>({
+				id: 'job-1',
+				data: {
+					executionId: 'exec-1',
+					loadStaticData: false,
+					streamingEnabled: false,
+					isMcpExecution: false,
+				},
+			});
+
+			const processPromise = jobProcessor.processJob(job);
+
+			// Shutdown arrives while the job is still in its preflight reads.
+			jobProcessor.suspendRunningJobs();
+			expect(workflowExecuteSuspendMock).not.toHaveBeenCalled();
+
+			resolveExecution(
+				mock<IExecutionResponse>({
+					mode: 'webhook',
+					workflowData: { nodes: [], staticData: {} },
+					data: mock<IRunExecutionData>(),
+				}),
+			);
+			await vi.waitFor(() => expect(jobProcessor.getRunningJobIds()).toEqual(['job-1']));
+			expect(workflowExecuteSuspendMock).toHaveBeenCalledTimes(1);
+
+			resolveRun(successRun());
+			await processPromise;
+		});
+
 		it('should not attach a suspend handle to a non-suspendable job', async () => {
 			const executionPersistence = mock<ExecutionPersistence>();
 			executionPersistence.findSingleExecution.mockResolvedValue(

packages/cli/src/scaling/job-processor.ts

@@ -369,6 +369,9 @@ export class JobProcessor {
 		if (workflowExecute && this.isJobSuspendable(job, execution)) {
 			const suspendable = workflowExecute;
 			runningJob.suspend = () => suspendable.suspend();
+			// A job that was still in its preflight reads when shutdown asked the
+			// running jobs to suspend would otherwise run to completion unasked.
+			if (this.suspensionRequested) runningJob.suspend();
 		}
 
 		this.runningJobs[job.id] = runningJob;
@@ -589,6 +592,9 @@ export class JobProcessor {
 	 * streaming responses cannot migrate mid-stream, and MCP executions are
 	 * pinned to their session.
 	 */
+	/** Set once at shutdown; jobs that register afterwards suspend right away. */
+	private suspensionRequested = false;
+
 	private isJobSuspendable(job: Job, execution: IExecutionResponse): boolean {
 		const isProductionMode = ['webhook', 'trigger', 'retry'].includes(execution.mode);
 
@@ -602,6 +608,7 @@ export class JobProcessor {
 
 	/** Ask every suspendable running job to stop at its next node boundary. */
 	suspendRunningJobs() {
+		this.suspensionRequested = true;
 		const executionIds: string[] = [];
 
 		for (const runningJob of Object.values(this.runningJobs)) {