diff --git a/awx/main/models/credential.py b/awx/main/models/credential.py
index d96c32eaf2..e646d9b5bc 100644
--- a/awx/main/models/credential.py
+++ b/awx/main/models/credential.py
@@ -118,15 +118,19 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique):
         return self.kind == 'ssh' and self.password == 'ASK'
 
     @property
-    def needs_ssh_key_unlock(self):
-        ssh_key_data = ''
-        if self.kind == 'ssh' and self.ssh_key_unlock == 'ASK':
-            if self.pk:
-                ssh_key_data = decrypt_field(self, 'ssh_key_data')
-            else:
-                ssh_key_data = self.ssh_key_data
+    def has_encrypted_ssh_key_data(self):
+        if self.pk:
+            ssh_key_data = decrypt_field(self, 'ssh_key_data')
+        else:
+            ssh_key_data = self.ssh_key_data
         return 'ENCRYPTED' in ssh_key_data
 
+    @property
+    def needs_ssh_key_unlock(self):
+        if self.kind == 'ssh' and self.ssh_key_unlock in ('ASK', ''):
+            return self.has_encrypted_ssh_key_data
+        return False
+
     @property
     def needs_sudo_password(self):
         return self.kind == 'ssh' and self.sudo_password == 'ASK'
@@ -211,11 +215,7 @@ class Credential(PasswordFieldsModel, CommonModelNameNotUnique):
         return self.ssh_key_data # No need to return decrypted version here.
 
     def clean_ssh_key_unlock(self):
-        if self.pk:
-            ssh_key_data = decrypt_field(self, 'ssh_key_data')
-        else:
-            ssh_key_data = self.ssh_key_data
-        if 'ENCRYPTED' in ssh_key_data and not self.ssh_key_unlock:
+        if self.has_encrypted_ssh_key_data and not self.ssh_key_unlock:
             raise ValidationError('SSH key unlock must be set when SSH key '
                                   'is encrypted')
         return self.ssh_key_unlock
diff --git a/awx/main/tasks.py b/awx/main/tasks.py
index 43d5074787..97e5d3ca7e 100644
--- a/awx/main/tasks.py
+++ b/awx/main/tasks.py
@@ -403,9 +403,7 @@ class RunJob(BaseTask):
                     value = kwargs.get(field, decrypt_field(creds, 'password'))
                 else:
                     value = kwargs.get(field, decrypt_field(creds, field))
-                if field == 'ssh_key_unlock' and value != 'ASK':
-                    passwords[field] = value
-                elif value not in ('', 'ASK'):
+                if value not in ('', 'ASK'):
                     passwords[field] = value
         return passwords
 
@@ -489,8 +487,7 @@ class RunJob(BaseTask):
 
         # If private key isn't encrypted, pass the path on the command line.
         ssh_key_path = kwargs.get('private_data_file', '')
-        ssh_key_unlock = kwargs.get('passwords', {}).get('ssh_key_unlock', None)
-        use_ssh_agent = bool(ssh_key_unlock is not None)
+        use_ssh_agent = bool(creds and creds.has_encrypted_ssh_key_data)
         if ssh_key_path and not use_ssh_agent:
             args.append('--private-key=%s' % ssh_key_path)