<commit_before>package cromwell.engine.workflow.lifecycle.execution

import akka.actor._
import cats.data.NonEmptyList
import cromwell.backend.BackendJobExecutionActor.{AbortedResponse, JobFailedNonRetryableResponse, JobFailedRetryableResponse, JobSucceededResponse}
import cromwell.backend.BackendLifecycleActor.AbortJobCommand
import cromwell.backend.{AllBackendInitializationData, BackendJobDescriptorKey, JobExecutionMap}
import cromwell.core.Dispatcher._
import cromwell.core.ExecutionIndex._
import cromwell.core.ExecutionStatus._
import cromwell.core.WorkflowOptions.WorkflowFailureMode
import cromwell.core._
import cromwell.core.logging.WorkflowLogging
import cromwell.engine.backend.{BackendSingletonCollection, CromwellBackends}
import cromwell.engine.workflow.lifecycle.execution.WorkflowExecutionActor._
import cromwell.engine.workflow.lifecycle.{EngineLifecycleActorAbortCommand, EngineLifecycleActorAbortedResponse}
import cromwell.engine.{ContinueWhilePossible, EngineWorkflowDescriptor}
import cromwell.util.StopAndLogSupervisor
import cromwell.webservice.EngineStatsActor
import lenthall.exception.ThrowableAggregation
import lenthall.util.TryUtil
import org.apache.commons.lang3.StringUtils
import wdl4s.WdlExpression.ScopedLookupFunction
import wdl4s.expression.WdlFunctions
import wdl4s.values.{WdlArray, WdlBoolean, WdlOptionalValue, WdlString, WdlValue}
import wdl4s.{Scope, _}

import scala.concurrent.duration._
import scala.language.postfixOps
import scala.util.{Failure, Success, Try}

case class WorkflowExecutionActor(workflowDescriptor: EngineWorkflowDescriptor,
                                  ioActor: ActorRef,
                                  serviceRegistryActor: ActorRef,
                                  jobStoreActor: ActorRef,
                                  subWorkflowStoreActor: ActorRef,
                                  callCacheReadActor: ActorRef,
                                  dockerHashActor: ActorRef,
                                  jobTokenDispenserActor: ActorRef,
                                  backendSingletonCollection: BackendSingletonCollection,
                                  initializationData: AllBackendInitializationData,
                                  restarting: Boolean)
  extends LoggingFSM[WorkflowExecutionActorState, WorkflowExecutionActorData] with WorkflowLogging with CallMetadataHelper with StopAndLogSupervisor {
  
  implicit val ec = context.dispatcher
  
  override val workflowIdForLogging = workflowDescriptor.id
  override val workflowIdForCallMetadata = workflowDescriptor.id

  private val tag = s"WorkflowExecutionActor [UUID(${workflowDescriptor.id.shortString})]"

  private val backendFactories = TryUtil.sequenceMap(workflowDescriptor.backendAssignments.values.toSet[String] map { backendName =>
    backendName -> CromwellBackends.backendLifecycleFactoryActorByName(backendName)
  } toMap) recover {
    case e => throw new RuntimeException("Could not instantiate backend factories", e)
  } get

  startWith(
    WorkflowExecutionPendingState,
    WorkflowExecutionActorData(
      workflowDescriptor,
      executionStore = ExecutionStore(workflowDescriptor.backendDescriptor.workflow, workflowDescriptor.knownValues),
      backendJobExecutionActors = Map.empty,
      engineCallExecutionActors = Map.empty,
      subWorkflowExecutionActors = Map.empty,
      downstreamExecutionMap = Map.empty,
      outputStore = OutputStore.empty
    )
  )

  when(WorkflowExecutionPendingState) {
    case Event(ExecuteWorkflowCommand, stateData) =>
      scheduleStartRunnableCalls()
      goto(WorkflowExecutionInProgressState)
  }

  when(WorkflowExecutionInProgressState) {
    case Event(CheckRunnable, data) => handleCheckRunnable(data)

    case Event(JobStarting(jobKey), stateData) =>
      pushStartingCallMetadata(jobKey)
      stay() using stateData
        .mergeExecutionDiff(WorkflowExecutionDiff(Map(jobKey -> ExecutionStatus.Starting)))
    case Event(JobRunning(key, inputs, callExecutionActor), stateData) =>
      pushRunningCallMetadata(key, inputs)
      stay() using stateData
        .addCallExecutionActor(key, callExecutionActor)
        .mergeExecutionDiff(WorkflowExecutionDiff(Map(key -> ExecutionStatus.Running)))
      
      //Success
        // Job
    case Event(JobSucceededResponse(jobKey, returnCode, callOutputs, _, _), stateData) =>
      pushSuccessfulCallMetadata(jobKey, returnCode, callOutputs)
      handleCallSuccessful(jobKey, callOutputs, stateData, Map.empty)
        // Sub Workflow
    case Event(SubWorkflowSucceededResponse(jobKey, descendantJobKeys, callOutputs), stateData) =>
      pushSuccessfulCallMetadata(jobKey, None, callOutputs)
      handleCallSuccessful(jobKey, callOutputs, stateData, descendantJobKeys)
        // Scatter
    case Event(ScatterCollectionSucceededResponse(jobKey, callOutputs), stateData) =>
      handleCallSuccessful(jobKey, callOutputs, stateData, Map.empty)
        // Declaration
    case Event(DeclarationEvaluationSucceededResponse(jobKey, callOutputs), stateData) =>
      handleDeclarationEvaluationSuccessful(jobKey, callOutputs, stateData)
        // Conditional
    case Event(BypassedCallResults(callOutputs), stateData) =>
      handleCallBypassed(callOutputs, stateData)
    case Event(BypassedDeclaration(declKey), stateData) =>
      handleDeclarationEvaluationSuccessful(declKey, WdlOptionalValue.none(declKey.scope.wdlType), stateData)

      // Failure
        // Initialization
    case Event(JobInitializationFailed(jobKey, reason), stateData) =>
      pushFailedCallMetadata(jobKey, None, reason, retryableFailure = false)
      handleNonRetryableFailure(stateData, jobKey, reason, Map.empty)
      // Job Non Retryable
    case Event(JobFailedNonRetryableResponse(jobKey, reason, returnCode), stateData) =>
      pushFailedCallMetadata(jobKey, returnCode, reason, retryableFailure = false)
      handleNonRetryableFailure(stateData, jobKey, reason, Map.empty)
      // Job Retryable
    case Event(JobFailedRetryableResponse(jobKey, reason, returnCode), _) =>
      pushFailedCallMetadata(jobKey, None, reason, retryableFailure = true)
      handleRetryableFailure(jobKey, reason, returnCode)
      // Sub Workflow - sub workflow failures are always non retryable
    case Event(SubWorkflowFailedResponse(jobKey, descendantJobKeys, reason), stateData) =>
      pushFailedCallMetadata(jobKey, None, reason, retryableFailure = false)
      handleNonRetryableFailure(stateData, jobKey, reason, descendantJobKeys)
    case Event(DeclarationEvaluationFailedResponse(jobKey, reason), stateData) =>
      handleDeclarationEvaluationFailure(jobKey, reason, stateData)
  }

  when(WorkflowExecutionAbortingState) {
    case Event(AbortedResponse(jobKey), stateData) =>
      pushAbortedCallMetadata(jobKey)
      handleCallAborted(stateData, jobKey, Map.empty)
    case Event(SubWorkflowAbortedResponse(jobKey, executedKeys), stateData) =>
      pushAbortedCallMetadata(jobKey)
      handleCallAborted(stateData, jobKey, executedKeys)
    case Event(SubWorkflowSucceededResponse(subKey, executedKeys, _), stateData) =>
      pushAbortedCallMetadata(subKey)
      handleCallAborted(stateData, subKey, executedKeys)
    case Event(JobSucceededResponse(jobKey, _, _, _, _), stateData) =>
      pushAbortedCallMetadata(jobKey)
      handleCallAborted(stateData, jobKey, Map.empty)
  }
  
  when(WorkflowExecutionSuccessfulState) {
    FSM.NullFunction
  }
  when(WorkflowExecutionFailedState) {
    alreadyFailedMopUp
  }
  when(WorkflowExecutionAbortedState) {
    alreadyFailedMopUp
  }

  private def scheduleStartRunnableCalls() = {
    context.system.scheduler.scheduleOnce(SweepInterval, self, CheckRunnable)
  }

  /**
    * Mop up function to handle a set of incoming results if this workflow has already failed:
    */
  private def alreadyFailedMopUp: StateFunction = {
    case Event(JobInitializationFailed(jobKey, reason), _) =>
      pushFailedCallMetadata(jobKey, None, reason, retryableFailure = false)
      stay
    case Event(JobFailedNonRetryableResponse(jobKey, reason, returnCode), _) =>
      pushFailedCallMetadata(jobKey, returnCode, reason, retryableFailure = false)
      stay
    case Event(JobFailedRetryableResponse(jobKey, reason, returnCode), _) =>
      pushFailedCallMetadata(jobKey, returnCode, reason, retryableFailure = true)
      stay
    case Event(JobSucceededResponse(jobKey, returnCode, callOutputs, _, _), _) =>
      pushSuccessfulCallMetadata(jobKey, returnCode, callOutputs)
      stay
  }


  def handleTerminated(actorRef: ActorRef) = {
    // Both of these Should Never Happen (tm), assuming the state data is set correctly on EJEA creation.
    // If they do, it's a big programmer error and the workflow execution fails.
    val jobKey = stateData.engineCallExecutionActors.getOrElse(actorRef, throw new RuntimeException("Programmer Error: An EJEA has terminated but was not assigned a jobKey"))
    val jobStatus = stateData.executionStore.jobStatus(jobKey).getOrElse(throw new RuntimeException("Programmer Error: An EJEA representing a jobKey which this workflow is not running has sent up a terminated message."))

    if (!jobStatus.isTerminal) {
      val terminationException = getFailureCause(actorRef) match {
        case Some(e) => new RuntimeException("Unexpected failure in EJEA.", e)
        case None => new RuntimeException("Unexpected failure in EJEA (root cause not captured).")
      }
      self ! JobFailedNonRetryableResponse(jobKey, terminationException, None)
    }

    stay
  }

  whenUnhandled {
    case Event(CheckRunnable, _) => stay()
    case Event(Terminated(actorRef), stateData) => handleTerminated(actorRef) using stateData.removeEngineJobExecutionActor(actorRef)
    case Event(EngineLifecycleActorAbortCommand, stateData) =>
      if (stateData.hasRunningActors) {
        log.info(s"$tag: Abort received. " +
          s"Aborting ${stateData.backendJobExecutionActors.size} Job Execution Actors" +
          s" and ${stateData.subWorkflowExecutionActors.size} Sub Workflow Execution Actors"
        )
        stateData.backendJobExecutionActors.values foreach { _ ! AbortJobCommand }
        stateData.subWorkflowExecutionActors.values foreach { _ ! EngineLifecycleActorAbortCommand }
        goto(WorkflowExecutionAbortingState)
      } else {
        goto(WorkflowExecutionAbortedState)
      }
    case Event(EngineStatsActor.JobCountQuery, data) =>
      sender ! EngineStatsActor.JobCount(data.backendJobExecutionActors.size)
      data.subWorkflowExecutionActors.values foreach { _ forward EngineStatsActor.JobCountQuery }
      stay()
    case unhandledMessage =>
      workflowLogger.warn(s"$tag received an unhandled message: ${unhandledMessage.event} in state: $stateName")
      stay()
  }

  onTransition {
    case fromState -> toState if toState.terminal =>
      workflowLogger.debug(s"$tag transitioning from $fromState to $toState. Stopping self.")
      context.stop(self)
    case fromState -> toState =>
      workflowLogger.debug(s"$tag transitioning from $fromState to $toState.")
  }

  onTransition {
    case _ -> WorkflowExecutionAbortedState =>
      context.parent ! WorkflowExecutionAbortedResponse(nextStateData.jobExecutionMap)
  }

  private def handleNonRetryableFailure(stateData: WorkflowExecutionActorData, failedJobKey: JobKey, reason: Throwable, jobExecutionMap: JobExecutionMap) = {
    val newData = stateData
      .removeCallExecutionActor(failedJobKey)
      .addExecutions(jobExecutionMap)
    
    handleExecutionFailure(failedJobKey, newData, reason, jobExecutionMap)
  }
  
  private def handleDeclarationEvaluationFailure(declarationKey: DeclarationKey, reason: Throwable, stateData: WorkflowExecutionActorData) = {
    handleExecutionFailure(declarationKey, stateData, reason, Map.empty)
  }
  
  private def handleExecutionFailure(failedJobKey: JobKey, data: WorkflowExecutionActorData, reason: Throwable, jobExecutionMap: JobExecutionMap) = {
    val newData = data.executionFailed(failedJobKey)
    
    if (workflowDescriptor.getWorkflowOption(WorkflowFailureMode).contains(ContinueWhilePossible.toString)) {
      newData.workflowCompletionStatus match {
        case Some(completionStatus) if completionStatus == Failed =>
          context.parent ! WorkflowExecutionFailedResponse(newData.jobExecutionMap, reason)
          goto(WorkflowExecutionFailedState) using newData
        case _ =><commit_msg>use the right failureMode value (#2239)<commit_after>package cromwell.engine.workflow.lifecycle.execution

import akka.actor._
import cats.data.NonEmptyList
import cromwell.backend.BackendJobExecutionActor.{AbortedResponse, JobFailedNonRetryableResponse, JobFailedRetryableResponse, JobSucceededResponse}
import cromwell.backend.BackendLifecycleActor.AbortJobCommand
import cromwell.backend.{AllBackendInitializationData, BackendJobDescriptorKey, JobExecutionMap}
import cromwell.core.Dispatcher._
import cromwell.core.ExecutionIndex._
import cromwell.core.ExecutionStatus._
import cromwell.core._
import cromwell.core.logging.WorkflowLogging
import cromwell.engine.backend.{BackendSingletonCollection, CromwellBackends}
import cromwell.engine.workflow.lifecycle.execution.WorkflowExecutionActor._
import cromwell.engine.workflow.lifecycle.{EngineLifecycleActorAbortCommand, EngineLifecycleActorAbortedResponse}
import cromwell.engine.{ContinueWhilePossible, EngineWorkflowDescriptor}
import cromwell.util.StopAndLogSupervisor
import cromwell.webservice.EngineStatsActor
import lenthall.exception.ThrowableAggregation
import lenthall.util.TryUtil
import org.apache.commons.lang3.StringUtils
import wdl4s.WdlExpression.ScopedLookupFunction
import wdl4s.expression.WdlFunctions
import wdl4s.values.{WdlArray, WdlBoolean, WdlOptionalValue, WdlString, WdlValue}
import wdl4s.{Scope, _}

import scala.concurrent.duration._
import scala.language.postfixOps
import scala.util.{Failure, Success, Try}

case class WorkflowExecutionActor(workflowDescriptor: EngineWorkflowDescriptor,
                                  ioActor: ActorRef,
                                  serviceRegistryActor: ActorRef,
                                  jobStoreActor: ActorRef,
                                  subWorkflowStoreActor: ActorRef,
                                  callCacheReadActor: ActorRef,
                                  dockerHashActor: ActorRef,
                                  jobTokenDispenserActor: ActorRef,
                                  backendSingletonCollection: BackendSingletonCollection,
                                  initializationData: AllBackendInitializationData,
                                  restarting: Boolean)
  extends LoggingFSM[WorkflowExecutionActorState, WorkflowExecutionActorData] with WorkflowLogging with CallMetadataHelper with StopAndLogSupervisor {
  
  implicit val ec = context.dispatcher
  
  override val workflowIdForLogging = workflowDescriptor.id
  override val workflowIdForCallMetadata = workflowDescriptor.id

  private val tag = s"WorkflowExecutionActor [UUID(${workflowDescriptor.id.shortString})]"

  private val backendFactories = TryUtil.sequenceMap(workflowDescriptor.backendAssignments.values.toSet[String] map { backendName =>
    backendName -> CromwellBackends.backendLifecycleFactoryActorByName(backendName)
  } toMap) recover {
    case e => throw new RuntimeException("Could not instantiate backend factories", e)
  } get

  startWith(
    WorkflowExecutionPendingState,
    WorkflowExecutionActorData(
      workflowDescriptor,
      executionStore = ExecutionStore(workflowDescriptor.backendDescriptor.workflow, workflowDescriptor.knownValues),
      backendJobExecutionActors = Map.empty,
      engineCallExecutionActors = Map.empty,
      subWorkflowExecutionActors = Map.empty,
      downstreamExecutionMap = Map.empty,
      outputStore = OutputStore.empty
    )
  )

  when(WorkflowExecutionPendingState) {
    case Event(ExecuteWorkflowCommand, stateData) =>
      scheduleStartRunnableCalls()
      goto(WorkflowExecutionInProgressState)
  }

  when(WorkflowExecutionInProgressState) {
    case Event(CheckRunnable, data) => handleCheckRunnable(data)

    case Event(JobStarting(jobKey), stateData) =>
      pushStartingCallMetadata(jobKey)
      stay() using stateData
        .mergeExecutionDiff(WorkflowExecutionDiff(Map(jobKey -> ExecutionStatus.Starting)))
    case Event(JobRunning(key, inputs, callExecutionActor), stateData) =>
      pushRunningCallMetadata(key, inputs)
      stay() using stateData
        .addCallExecutionActor(key, callExecutionActor)
        .mergeExecutionDiff(WorkflowExecutionDiff(Map(key -> ExecutionStatus.Running)))
      
      //Success
        // Job
    case Event(JobSucceededResponse(jobKey, returnCode, callOutputs, _, _), stateData) =>
      pushSuccessfulCallMetadata(jobKey, returnCode, callOutputs)
      handleCallSuccessful(jobKey, callOutputs, stateData, Map.empty)
        // Sub Workflow
    case Event(SubWorkflowSucceededResponse(jobKey, descendantJobKeys, callOutputs), stateData) =>
      pushSuccessfulCallMetadata(jobKey, None, callOutputs)
      handleCallSuccessful(jobKey, callOutputs, stateData, descendantJobKeys)
        // Scatter
    case Event(ScatterCollectionSucceededResponse(jobKey, callOutputs), stateData) =>
      handleCallSuccessful(jobKey, callOutputs, stateData, Map.empty)
        // Declaration
    case Event(DeclarationEvaluationSucceededResponse(jobKey, callOutputs), stateData) =>
      handleDeclarationEvaluationSuccessful(jobKey, callOutputs, stateData)
        // Conditional
    case Event(BypassedCallResults(callOutputs), stateData) =>
      handleCallBypassed(callOutputs, stateData)
    case Event(BypassedDeclaration(declKey), stateData) =>
      handleDeclarationEvaluationSuccessful(declKey, WdlOptionalValue.none(declKey.scope.wdlType), stateData)

      // Failure
        // Initialization
    case Event(JobInitializationFailed(jobKey, reason), stateData) =>
      pushFailedCallMetadata(jobKey, None, reason, retryableFailure = false)
      handleNonRetryableFailure(stateData, jobKey, reason, Map.empty)
      // Job Non Retryable
    case Event(JobFailedNonRetryableResponse(jobKey, reason, returnCode), stateData) =>
      pushFailedCallMetadata(jobKey, returnCode, reason, retryableFailure = false)
      handleNonRetryableFailure(stateData, jobKey, reason, Map.empty)
      // Job Retryable
    case Event(JobFailedRetryableResponse(jobKey, reason, returnCode), _) =>
      pushFailedCallMetadata(jobKey, None, reason, retryableFailure = true)
      handleRetryableFailure(jobKey, reason, returnCode)
      // Sub Workflow - sub workflow failures are always non retryable
    case Event(SubWorkflowFailedResponse(jobKey, descendantJobKeys, reason), stateData) =>
      pushFailedCallMetadata(jobKey, None, reason, retryableFailure = false)
      handleNonRetryableFailure(stateData, jobKey, reason, descendantJobKeys)
    case Event(DeclarationEvaluationFailedResponse(jobKey, reason), stateData) =>
      handleDeclarationEvaluationFailure(jobKey, reason, stateData)
  }

  when(WorkflowExecutionAbortingState) {
    case Event(AbortedResponse(jobKey), stateData) =>
      pushAbortedCallMetadata(jobKey)
      handleCallAborted(stateData, jobKey, Map.empty)
    case Event(SubWorkflowAbortedResponse(jobKey, executedKeys), stateData) =>
      pushAbortedCallMetadata(jobKey)
      handleCallAborted(stateData, jobKey, executedKeys)
    case Event(SubWorkflowSucceededResponse(subKey, executedKeys, _), stateData) =>
      pushAbortedCallMetadata(subKey)
      handleCallAborted(stateData, subKey, executedKeys)
    case Event(JobSucceededResponse(jobKey, _, _, _, _), stateData) =>
      pushAbortedCallMetadata(jobKey)
      handleCallAborted(stateData, jobKey, Map.empty)
  }
  
  when(WorkflowExecutionSuccessfulState) {
    FSM.NullFunction
  }
  when(WorkflowExecutionFailedState) {
    alreadyFailedMopUp
  }
  when(WorkflowExecutionAbortedState) {
    alreadyFailedMopUp
  }

  private def scheduleStartRunnableCalls() = {
    context.system.scheduler.scheduleOnce(SweepInterval, self, CheckRunnable)
  }

  /**
    * Mop up function to handle a set of incoming results if this workflow has already failed:
    */
  private def alreadyFailedMopUp: StateFunction = {
    case Event(JobInitializationFailed(jobKey, reason), _) =>
      pushFailedCallMetadata(jobKey, None, reason, retryableFailure = false)
      stay
    case Event(JobFailedNonRetryableResponse(jobKey, reason, returnCode), _) =>
      pushFailedCallMetadata(jobKey, returnCode, reason, retryableFailure = false)
      stay
    case Event(JobFailedRetryableResponse(jobKey, reason, returnCode), _) =>
      pushFailedCallMetadata(jobKey, returnCode, reason, retryableFailure = true)
      stay
    case Event(JobSucceededResponse(jobKey, returnCode, callOutputs, _, _), _) =>
      pushSuccessfulCallMetadata(jobKey, returnCode, callOutputs)
      stay
  }


  def handleTerminated(actorRef: ActorRef) = {
    // Both of these Should Never Happen (tm), assuming the state data is set correctly on EJEA creation.
    // If they do, it's a big programmer error and the workflow execution fails.
    val jobKey = stateData.engineCallExecutionActors.getOrElse(actorRef, throw new RuntimeException("Programmer Error: An EJEA has terminated but was not assigned a jobKey"))
    val jobStatus = stateData.executionStore.jobStatus(jobKey).getOrElse(throw new RuntimeException("Programmer Error: An EJEA representing a jobKey which this workflow is not running has sent up a terminated message."))

    if (!jobStatus.isTerminal) {
      val terminationException = getFailureCause(actorRef) match {
        case Some(e) => new RuntimeException("Unexpected failure in EJEA.", e)
        case None => new RuntimeException("Unexpected failure in EJEA (root cause not captured).")
      }
      self ! JobFailedNonRetryableResponse(jobKey, terminationException, None)
    }

    stay
  }

  whenUnhandled {
    case Event(CheckRunnable, _) => stay()
    case Event(Terminated(actorRef), stateData) => handleTerminated(actorRef) using stateData.removeEngineJobExecutionActor(actorRef)
    case Event(EngineLifecycleActorAbortCommand, stateData) =>
      if (stateData.hasRunningActors) {
        log.info(s"$tag: Abort received. " +
          s"Aborting ${stateData.backendJobExecutionActors.size} Job Execution Actors" +
          s" and ${stateData.subWorkflowExecutionActors.size} Sub Workflow Execution Actors"
        )
        stateData.backendJobExecutionActors.values foreach { _ ! AbortJobCommand }
        stateData.subWorkflowExecutionActors.values foreach { _ ! EngineLifecycleActorAbortCommand }
        goto(WorkflowExecutionAbortingState)
      } else {
        goto(WorkflowExecutionAbortedState)
      }
    case Event(EngineStatsActor.JobCountQuery, data) =>
      sender ! EngineStatsActor.JobCount(data.backendJobExecutionActors.size)
      data.subWorkflowExecutionActors.values foreach { _ forward EngineStatsActor.JobCountQuery }
      stay()
    case unhandledMessage =>
      workflowLogger.warn(s"$tag received an unhandled message: ${unhandledMessage.event} in state: $stateName")
      stay()
  }

  onTransition {
    case fromState -> toState if toState.terminal =>
      workflowLogger.debug(s"$tag transitioning from $fromState to $toState. Stopping self.")
      context.stop(self)
    case fromState -> toState =>
      workflowLogger.debug(s"$tag transitioning from $fromState to $toState.")
  }

  onTransition {
    case _ -> WorkflowExecutionAbortedState =>
      context.parent ! WorkflowExecutionAbortedResponse(nextStateData.jobExecutionMap)
  }

  private def handleNonRetryableFailure(stateData: WorkflowExecutionActorData, failedJobKey: JobKey, reason: Throwable, jobExecutionMap: JobExecutionMap) = {
    val newData = stateData
      .removeCallExecutionActor(failedJobKey)
      .addExecutions(jobExecutionMap)
    
    handleExecutionFailure(failedJobKey, newData, reason, jobExecutionMap)
  }
  
  private def handleDeclarationEvaluationFailure(declarationKey: DeclarationKey, reason: Throwable, stateData: WorkflowExecutionActorData) = {
    handleExecutionFailure(declarationKey, stateData, reason, Map.empty)
  }
  
  private def handleExecutionFailure(failedJobKey: JobKey, data: WorkflowExecutionActorData, reason: Throwable, jobExecutionMap: JobExecutionMap) = {
    val newData = data.executionFailed(failedJobKey)
    
    if (workflowDescriptor.failureMode == ContinueWhilePossible) {
      newData.workflowCompletionStatus match {
        case Some(completionStatus) if completionStatus == Failed =>
          context.parent ! WorkflowExecutionFailedResponse(newData.jobExecutionMap, reason)
          goto(WorkflowExecutionFailedState) using newData
        case _ =>
{'fraction_non_alphanumeric': 0.07705948734483314, 'fraction_numerical': 0.0005642431081734644, 'mean_word_length': 4.300790429395429, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 89, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5111875', 'n_tokens_mistral': 6830, 'n_tokens_neox': 6669, 'n_words': 1360}