Currently we use execution-date when generating the command to send via executor in https://github.com/apache/airflow/blob/9c94b72d440b18a9e42123d20d48b951712038f9/airflow/models/taskinstance.py#L420-L436

We should replace the execution_date in their to run_id instead and handle the corresponding changes needs on the executor and worker.
airflow/models/taskinstance.py
return self._executor_config
    @provide_session
    def construct_task_instance(self, session=None, lock_for_update=False) -> TaskInstance:
        """
        Construct a TaskInstance from the database based on the primary key
        :param session: DB session.
        :param lock_for_update: if True, indicates that the database should
            lock the TaskInstance (issuing a FOR UPDATE clause) until the
            session is committed.
        :return: the task instance constructed
        """
        qry = session.query(TaskInstance).filter(
            TaskInstance.dag_id == self._dag_id,
            TaskInstance.task_id == self._task_id,
            TaskInstance.execution_date == self._execution_date,
        )
        if lock_for_update:
            ti = qry.with_for_update().first()
        else:
            ti = qry.first()
        return ti
STATICA_HACK = True
globals()['kcah_acitats'[::-1].upper()] = False
if STATICA_HACK:  
    from airflow.jobs.base_job import BaseJob
    from airflow.models.dagrun import DagRun
    TaskInstance.dag_run = relationship(DagRun)
    TaskInstance.queued_by_job = relationship(BaseJob)