Fix InProcessSupervisorComms not awaiting deferred SetXCom response

When SetXCom is offloaded to the thread pool, _handle_request() returns
immediately without calling send_msg(). The InProcessSupervisorComms.send()
then calls _get_response() on an empty deque, causing IndexError.

Wait for any pending futures and drain them before popping the response.

Signed-off-by: Haneul Yeom <nyeom@nyeom.dev>

Author: skymensch

task-sdk/src/airflow/sdk/execution_time/supervisor.py

@@ -1611,6 +1611,16 @@ def send(self, msg: BaseModel):
         with set_supervisor_comms(None):
             self.supervisor._handle_request(msg, log, 0)  # type: ignore[arg-type]
 
+        # Some requests (e.g. SetXCom) are offloaded to the supervisor's thread pool to
+        # avoid blocking its event loop. In the in-process path there is no event loop
+        # calling _drain_pending_requests(), so we wait for any in-flight futures here
+        # and drain them so that the response is available before we try to pop it.
+        if self.supervisor._pending_requests:
+            from concurrent.futures import wait as futures_wait
+
+            futures_wait([f for f, _ in self.supervisor._pending_requests])
+            self.supervisor._drain_pending_requests()
+
         return self._get_response()