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
self.log.info('Submitting %s to sensor service', self)
        self.state = State.SENSING
        self.start_date = timezone.utcnow()
        session.merge(self)
        session.commit()
        
        raise AirflowSmartSensorException("Task successfully registered in smart sensor.")
    def _execute_task(self, context, task_copy):
        """Executes Task (optionally with a Timeout) and pushes Xcom results"""
        
        
        
        execute_callable = task_copy.execute
        if self.next_method:
            
            
            if self.next_method == "__fail__":
                next_kwargs = self.next_kwargs or {}
                raise TaskDeferralError(next_kwargs.get("error", "Unknown"))
            
            execute_callable = getattr(task_copy, self.next_method)
            if self.next_kwargs: