https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
task: BaseOperator | None = None
        try:
            if getattr(ti, "task", None) and context:
                task = ti.task.unmap((context, session))
        except Exception:
            cls.logger().error("Unable to unmap task to determine if we need to send an alert email")
        if force_fail or not ti.is_eligible_to_retry():
            ti.state = TaskInstanceState.FAILED
            email_for_state = operator.attrgetter("email_on_failure")
            callbacks = task.on_failure_callback if task else None
            if task and task.dag and task.dag.fail_stop:
                _stop_remaining_tasks(task_instance=ti, session=session)
        else:
            if ti.state == TaskInstanceState.QUEUED:
                
                ti._try_number += 1
            ti.state = State.UP_FOR_RETRY
            email_for_state = operator.attrgetter("email_on_retry")
            callbacks = task.on_retry_callback if task else None
        return {
            "ti": ti,
            "email_for_state": email_for_state,
            "task": task,
            "callbacks": callbacks,
            "context": context,
        }
    @staticmethod
    @internal_api_call
    @provide_session
    def save_to_db(ti: TaskInstance | TaskInstancePydantic, session: Session = NEW_SESSION):