def before_task_start(self, task_ex, task_spec):
        super(WaitBeforePolicy, self).before_task_start(task_ex, task_spec)

        context_key = "wait_before_policy"

        runtime_context = _ensure_context_has_key(task_ex.runtime_context, context_key)

        task_ex.runtime_context = runtime_context

        policy_context = runtime_context[context_key]

        if policy_context.get("skip"):
            # Unset state 'DELAYED'.
            wf_trace.info(task_ex, "Task '%s' [%s -> %s]" % (task_ex.name, states.DELAYED, states.RUNNING))

            task_ex.state = states.RUNNING

            return

        if task_ex.state != states.IDLE:
            policy_context.update({"skip": True})
            _log_task_delay(task_ex, self.delay)

            task_ex.state = states.DELAYED

            scheduler.schedule_call(None, _RUN_EXISTING_TASK_PATH, self.delay, task_ex_id=task_ex.id)

    def set_state(self, state, state_info, processed=None):
        """Sets task state without executing post completion logic.

        :param state: New task state.
        :param state_info: New state information (i.e. error message).
        :param processed: New "processed" flag value.
        """

        assert self.task_ex

        if (self.task_ex.state != state or
                self.task_ex.state_info != state_info):
            wf_trace.info(
                self.task_ex.workflow_execution,
                "Task '%s' (%s) [%s -> %s, msg=%s]" %
                (self.task_ex.name,
                 self.task_ex.id,
                 self.task_ex.state,
                 state,
                 state_info)
            )

            self.state_changed = True

        self.task_ex.state = state
        self.task_ex.state_info = state_info

        if processed is not None:
            self.task_ex.processed = processed

    def start(self, input_dict, desc='', params=None):
        """Start workflow.

        :param input_dict: Workflow input.
        :param desc: Workflow execution description.
        :param params: Workflow type specific parameters.
        """

        assert not self.wf_ex

        wf_trace.info(self.wf_ex, "Starting workflow: %s" % self.wf_def)

        # TODO(rakhmerov): This call implicitly changes input_dict! Fix it!
        # After fix we need to move validation after adding risky fields.
        eng_utils.validate_input(self.wf_def, input_dict, self.wf_spec)

        self._create_execution(input_dict, desc, params)

        self.set_state(states.RUNNING)

        wf_ctrl = wf_base.get_controller(self.wf_ex, self.wf_spec)

        cmds = wf_ctrl.continue_workflow()

        dispatcher.dispatch_workflow_commands(self.wf_ex, cmds)

    def on_task_state_change(self, task_ex_id, state):
        with db_api.transaction():
            task_ex = db_api.get_task_execution(task_ex_id)
            # TODO(rakhmerov): The method is mostly needed for policy and
            # we are supposed to get the same action execution as when the
            # policy worked. But by the moment this method is called the
            # last execution object may have changed. It's a race condition.
            execution = task_ex.executions[-1]

            wf_ex_id = task_ex.workflow_execution_id

            # Must be before loading the object itself (see method doc).
            self._lock_workflow_execution(wf_ex_id)

            wf_ex = task_ex.workflow_execution

            wf_trace.info(
                task_ex,
                "Task '%s' [%s -> %s]"
                % (task_ex.name, task_ex.state, state)
            )

            task_ex.state = state

            self._on_task_state_change(task_ex, wf_ex, action_ex=execution)

def _log_action_result(action_ex, from_state, to_state, result):
    def _result_msg():
        if action_ex.state == states.ERROR:
            return "error = %s" % utils.cut(result.error)

        return "result = %s" % utils.cut(result.data)

    wf_trace.info(None, "Action execution '%s' [%s -> %s, %s]" % (action_ex.name, from_state, to_state, _result_msg()))

    def after_task_complete(self, task_ex, task_spec):
        """Possible Cases:

        1. state = SUCCESS
           No need to move to next iteration.
        2. retry:count = 5, current:count = 2, state = ERROR,
           state = IDLE/DELAYED, current:count = 3
        3. retry:count = 5, current:count = 4, state = ERROR
        Iterations complete therefore state = #{state}, current:count = 4.
        """
        super(RetryPolicy, self).after_task_complete(task_ex, task_spec)

        context_key = 'retry_task_policy'

        runtime_context = _ensure_context_has_key(
            task_ex.runtime_context,
            context_key
        )

        task_ex.runtime_context = runtime_context

        state = task_ex.state

        if state != states.ERROR:
            return

        wf_trace.info(
            task_ex,
            "Task '%s' [%s -> ERROR]"
            % (task_ex.name, task_ex.state)
        )

        policy_context = runtime_context[context_key]

        retry_no = 0

        if 'retry_no' in policy_context:
            retry_no = policy_context['retry_no']
            del policy_context['retry_no']

        retries_remain = retry_no + 1 < self.count

        if not retries_remain or self.break_on:
            return

        _log_task_delay(task_ex, self.delay)

        task_ex.state = states.DELAYED

        policy_context['retry_no'] = retry_no + 1
        runtime_context[context_key] = policy_context

        scheduler.schedule_call(
            None,
            _RUN_EXISTING_TASK_PATH,
            self.delay,
            task_ex_id=task_ex.id,
        )

def _set_task_state(task_ex, state):
    # TODO(rakhmerov): How do we log task result?
    wf_trace.info(
        task_ex.workflow_execution,
        "Task execution '%s' [%s -> %s]" %
        (task_ex.name, task_ex.state, state)
    )

    task_ex.state = state

def fail_task_if_incomplete(task_ex_id, timeout):
    task_ex = db_api.get_task_execution(task_ex_id)

    if not states.is_completed(task_ex.state):
        msg = "Task timed out [id=%s, timeout(s)=%s]." % (task_ex_id, timeout)

        wf_trace.info(task_ex, msg)

        wf_trace.info(task_ex, "Task '%s' [%s -> ERROR]" % (task_ex.name, task_ex.state))

        rpc.get_engine_client().on_task_state_change(task_ex_id, states.ERROR)

def set_task_state(task_ex, state, state_info, processed=None):
    wf_trace.info(
        task_ex.workflow_execution,
        "Task execution '%s' [%s -> %s]" %
        (task_ex.name, task_ex.state, state)
    )

    task_ex.state = state
    task_ex.state_info = state_info

    if processed is not None:
        task_ex.processed = processed

    def before_task_start(self, task_ex, task_spec):
        super(PauseBeforePolicy, self).before_task_start(task_ex, task_spec)

        if not self.expr:
            return

        wf_trace.info(
            task_ex,
            "Workflow paused before task '%s' [%s -> %s]" %
            (task_ex.name, task_ex.workflow_execution.state, states.PAUSED)
        )

        task_ex.workflow_execution.state = states.PAUSED
        task_ex.state = states.IDLE

    def set_state(self, state, state_info, processed=None):
        """Sets task state without executing post completion logic.

        :param state: New task state.
        :param state_info: New state information (i.e. error message).
        :param processed: New "processed" flag value.
        :return: True if the state was changed as a result of this call,
            False otherwise.
        """

        assert self.task_ex

        cur_state = self.task_ex.state

        # Set initial started_at in case of waiting => running.
        # We can't set this just in run_existing, because task retries
        # will update started_at, which is incorrect.
        if cur_state == states.WAITING and state == states.RUNNING:
            self.save_started_time()

        if cur_state != state or self.task_ex.state_info != state_info:
            task_ex = db_api.update_task_execution_state(
                id=self.task_ex.id,
                cur_state=cur_state,
                state=state
            )

            if task_ex is None:
                # Do nothing because the update query did not change the DB.
                return False

            self.task_ex = task_ex
            self.task_ex.state_info = json.dumps(state_info) \
                if isinstance(state_info, dict) else state_info
            self.state_changed = True

            if processed is not None:
                self.task_ex.processed = processed

            wf_trace.info(
                self.task_ex.workflow_execution,
                "Task '%s' (%s) [%s -> %s, msg=%s]" %
                (self.task_ex.name,
                 self.task_ex.id,
                 cur_state,
                 state,
                 self.task_ex.state_info)
            )

        return True

    def _log_result(self, prev_state, result):
        state = self.action_ex.state

        def _result_msg():
            if state == states.ERROR:
                return "error = %s" % utils.cut(result.error)

            return "result = %s" % utils.cut(result.data)

        wf_trace.info(
            None,
            "Action execution '%s' [%s -> %s, %s]" %
            (self.action_ex.name, prev_state, state, _result_msg())
        )

    def start(self, wf_def, wf_ex_id, input_dict, desc='', params=None):
        """Start workflow.

        :param wf_def: Workflow definition.
        :param wf_ex_id: Workflow execution id.
        :param input_dict: Workflow input.
        :param desc: Workflow execution description.
        :param params: Workflow type specific parameters.

        :raises
        """

        assert not self.wf_ex

        # New workflow execution.
        self.wf_spec = spec_parser.get_workflow_spec_by_definition_id(
            wf_def.id,
            wf_def.updated_at
        )

        wf_trace.info(
            self.wf_ex,
            'Starting workflow [name=%s, input=%s]' %
            (wf_def.name, utils.cut(input_dict))
        )

        self.validate_input(input_dict)

        self._create_execution(
            wf_def,
            wf_ex_id,
            self.prepare_input(input_dict),
            desc,
            params
        )

        self.set_state(states.RUNNING)

        # Publish event as soon as state is set to running.
        self.notify(events.WORKFLOW_LAUNCHED)

        wf_ctrl = wf_base.get_controller(self.wf_ex, self.wf_spec)

        dispatcher.dispatch_workflow_commands(
            self.wf_ex,
            wf_ctrl.continue_workflow()
        )

def _run_action_or_workflow(task_ex, task_spec, input_dict, index):
    t_name = task_ex.name

    if task_spec.get_action_name():
        wf_trace.info(
            task_ex,
            "Task '%s' is RUNNING [action_name = %s]" %
            (t_name, task_spec.get_action_name())
        )

        _schedule_run_action(task_ex, task_spec, input_dict, index)
    elif task_spec.get_workflow_name():
        wf_trace.info(
            task_ex,
            "Task '%s' is RUNNING [workflow_name = %s]" %
            (t_name, task_spec.get_workflow_name()))

        _schedule_run_workflow(task_ex, task_spec, input_dict, index)

    def set_state(self, state, state_info=None, recursive=False):
        assert self.wf_ex

        cur_state = self.wf_ex.state

        if states.is_valid_transition(cur_state, state):
            self.wf_ex.state = state
            self.wf_ex.state_info = state_info

            wf_trace.info(
                self.wf_ex,
                "Execution of workflow '%s' [%s -> %s]"
                % (self.wf_ex.workflow_name, cur_state, state)
            )
        else:
            msg = ("Can't change workflow execution state from %s to %s. "
                   "[workflow=%s, execution_id=%s]" %
                   (cur_state, state, self.wf_ex.name, self.wf_ex.id))

            raise exc.WorkflowException(msg)

        # Workflow result should be accepted by parent workflows (if any)
        # only if it completed successfully or failed.
        self.wf_ex.accepted = states.is_completed(state)

        if recursive and self.wf_ex.task_execution_id:
            parent_task_ex = db_api.get_task_execution(
                self.wf_ex.task_execution_id
            )

            parent_wf = Workflow(
                db_api.get_workflow_definition(parent_task_ex.workflow_id),
                parent_task_ex.workflow_execution
            )

            parent_wf.lock()

            parent_wf.set_state(state, recursive=recursive)

            # TODO(rakhmerov): It'd be better to use instance of Task here.
            parent_task_ex.state = state
            parent_task_ex.state_info = None
            parent_task_ex.processed = False

    def before_task_start(self, task_ex, task_spec):
        super(WaitBeforePolicy, self).before_task_start(task_ex, task_spec)

        # No need to wait for a task if delay is 0
        if self.delay == 0:
            return

        context_key = 'wait_before_policy'

        runtime_context = _ensure_context_has_key(
            task_ex.runtime_context,
            context_key
        )

        task_ex.runtime_context = runtime_context

        policy_context = runtime_context[context_key]

        if policy_context.get('skip'):
            # Unset state 'RUNNING_DELAYED'.
            wf_trace.info(
                task_ex,
                "Task '%s' [%s -> %s]"
                % (task_ex.name, states.RUNNING_DELAYED, states.RUNNING)
            )

            task_ex.state = states.RUNNING

            return

        if task_ex.state != states.IDLE:
            policy_context.update({'skip': True})

            _log_task_delay(task_ex, self.delay)

            task_ex.state = states.RUNNING_DELAYED

            scheduler.schedule_call(
                None,
                _CONTINUE_TASK_PATH,
                self.delay,
                task_ex_id=task_ex.id,
            )

    def set_state(self, state, state_info=None):
        assert self.wf_ex

        cur_state = self.wf_ex.state

        if states.is_valid_transition(cur_state, state):
            wf_ex = db_api.update_workflow_execution_state(
                id=self.wf_ex.id,
                cur_state=cur_state,
                state=state
            )

            if wf_ex is None:
                # Do nothing because the state was updated previously.
                return False

            self.wf_ex = wf_ex
            self.wf_ex.state_info = json.dumps(state_info) \
                if isinstance(state_info, dict) else state_info

            wf_trace.info(
                self.wf_ex,
                "Workflow '%s' [%s -> %s, msg=%s]" %
                (self.wf_ex.workflow_name,
                 cur_state,
                 state,
                 self.wf_ex.state_info)
            )
        else:
            msg = ("Can't change workflow execution state from %s to %s. "
                   "[workflow=%s, execution_id=%s]" %
                   (cur_state, state, self.wf_ex.name, self.wf_ex.id))

            raise exc.WorkflowException(msg)

        # Workflow result should be accepted by parent workflows (if any)
        # only if it completed successfully or failed.
        self.wf_ex.accepted = states.is_completed(state)

        if states.is_completed(state):
            triggers.on_workflow_complete(self.wf_ex)

        return True

def set_execution_state(wf_ex, state, state_info=None, set_upstream=False):
    cur_state = wf_ex.state

    if states.is_valid_transition(cur_state, state):
        wf_ex.state = state
        wf_ex.state_info = state_info

        wf_trace.info(
            wf_ex,
            "Execution of workflow '%s' [%s -> %s]"
            % (wf_ex.workflow_name, cur_state, state)
        )
    else:
        msg = ("Can't change workflow execution state from %s to %s. "
               "[workflow=%s, execution_id=%s]" %
               (cur_state, state, wf_ex.name, wf_ex.id))
        raise exc.WorkflowException(msg)

    # Workflow result should be accepted by parent workflows (if any)
    # only if it completed successfully or failed.
    wf_ex.accepted = wf_ex.state in (states.SUCCESS, states.ERROR)

    # If specified, then recursively set the state of the parent workflow
    # executions to the same state. Only changing state to RUNNING is
    # supported.
    if set_upstream and state == states.RUNNING and wf_ex.task_execution_id:
        task_ex = db_api.get_task_execution(wf_ex.task_execution_id)

        parent_wf_ex = lock_workflow_execution(task_ex.workflow_execution_id)

        set_execution_state(
            parent_wf_ex,
            state,
            state_info=state_info,
            set_upstream=set_upstream
        )

        task_handler.set_task_state(
            task_ex,
            state,
            state_info=None,
            processed=False
        )

def create_workflow_execution(wf_identifier, wf_input, description, params):
    params = canonize_workflow_params(params)

    wf_def = db_api.get_workflow_definition(wf_identifier)
    wf_spec = spec_parser.get_workflow_spec(wf_def.spec)

    eng_utils.validate_input(wf_def, wf_input, wf_spec)

    wf_ex = _create_workflow_execution(
        wf_def,
        wf_spec,
        wf_input,
        description,
        params
    )

    wf_trace.info(wf_ex, "Starting workflow: '%s'" % wf_identifier)

    return wf_ex.id

    def set_state(self, state, state_info, processed=None):
        """Sets task state without executing post completion logic.

        :param state: New task state.
        :param state_info: New state information (i.e. error message).
        :param processed: New "processed" flag value.
        """

        wf_trace.info(
            self.task_ex.workflow_execution,
            "Task execution '%s' [%s -> %s]: %s" %
            (self.task_ex.id, self.task_ex.state, state, state_info)
        )

        self.task_ex.state = state
        self.task_ex.state_info = state_info

        if processed is not None:
            self.task_ex.processed = processed