@@ -22,6 +22,7 @@ package org.apache.airflow.sdk.execution
 import kotlinx.coroutines.runBlocking
 import org.apache.airflow.sdk.execution.comm.ConnectionResult
 import org.apache.airflow.sdk.execution.comm.DeleteVariable
+import org.apache.airflow.sdk.execution.comm.DeleteXCom
 import org.apache.airflow.sdk.execution.comm.GetConnection
 import org.apache.airflow.sdk.execution.comm.GetVariable
 import org.apache.airflow.sdk.execution.comm.GetXCom
@@ -74,6 +75,14 @@ interface Client {
     runId: String,
     mapIndex: Int,
   )
+
+  fun deleteXCom(
+    key: String,
+    dagId: String,
+    taskId: String,
+    runId: String,
+    mapIndex: Int,
+  )
 }
 
 /**
@@ -157,4 +166,22 @@ class CoordinatorClient(
       }
     return runBlocking { exec.communicate<XComResult>(message) }
   }
+
+  override fun deleteXCom(
+    key: String,
+    dagId: String,
+    taskId: String,
+    runId: String,
+    mapIndex: Int,
+  ) {
+    val message =
+      DeleteXCom().also {
+        it.key = key
+        it.dagId = dagId
+        it.taskId = taskId
+        it.runId = runId
+        it.mapIndex = mapIndex
+      }
+    runBlocking { exec.communicate<Unit>(message) }
+  }
 }
@@ -77,6 +77,27 @@ internal object TaskRunner {
     val taskDef =
       bundle.taskDef(request.ti.dagId, request.ti.taskId)
         ?: return TaskResult.of(TaskState.State.REMOVED)
+    // When a try starts, the runtime has to delete the XComs in xcom_keys_to_clear, because Airflow
+    // does not delete them. The Python task runner deletes them before it renders templates. Here
+    // they are deleted before the task class is instantiated, so they are already gone if the
+    // constructor throws during this try.
+    for (key in request.tiContext.xcomKeysToClear.orEmpty()) {
+      try {
+        client.impl.deleteXCom(
+          key = key,
+          dagId = request.ti.dagId,
+          taskId = request.ti.taskId,
+          runId = request.ti.runId,
+          mapIndex = request.ti.mapIndex ?: -1,
+        )
+      } catch (e: Throwable) {
+        logger.error(
+          "Error clearing XCom",
+          mapOf("ti" to request.ti, "key" to key, "error" to e, "trace" to e.stackTraceToString()),
+        )
+        return TaskResult.failure(request.tiContext.shouldRetry)
+      }
+    }
     val definition = taskDef.definition
     val instance =
       try {
 
@@ -67,6 +67,14 @@ internal class FakeXComTransport(
     runId: String,
     mapIndex: Int,
   ) = throw NotImplementedError()
+
+  override fun deleteXCom(
+    key: String,
+    dagId: String,
+    taskId: String,
+    runId: String,
+    mapIndex: Int,
+  ) = throw NotImplementedError()
 }
 
 internal fun startupDetails(argBindings: List<Map<String, Any?>>?): StartupDetails =
 
@@ -59,6 +59,14 @@ private class FakeTransport(
     runId: String,
     mapIndex: Int,
   ) = throw NotImplementedError()
+
+  override fun deleteXCom(
+    key: String,
+    dagId: String,
+    taskId: String,
+    runId: String,
+    mapIndex: Int,
+  ) = throw NotImplementedError()
 }
 
 class ClientTest {
 
@@ -33,6 +33,7 @@ import org.junit.jupiter.api.DisplayName
 import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.Timeout
 import org.msgpack.core.MessagePack
+import org.msgpack.core.buffer.ArrayBufferInput
 import java.io.ByteArrayOutputStream
 import java.util.concurrent.ArrayBlockingQueue
 import java.util.concurrent.TimeUnit
@@ -60,6 +61,111 @@ class ServerTest {
     writeByteArray(payload)
   }
 
+  private fun errorResponseFrame(id: Int): ByteArray {
+    val out = ByteArrayOutputStream()
+    MessagePack.newDefaultPacker(out).use { packer ->
+      packer.packArrayHeader(3)
+      packer.packInt(id)
+      packer.packNil()
+      packer.packMapHeader(3)
+      packer.packString("type")
+      packer.packString("ErrorResponse")
+      packer.packString("error")
+      packer.packString("API_SERVER_ERROR")
+      packer.packString("detail")
+      packer.packMapHeader(1)
+      packer.packString("status_code")
+      packer.packInt(500)
+    }
+    return out.toByteArray()
+  }
+
+  private fun startupFrame(
+    mapIndex: Int?,
+    tiContext: Map<String, Any?>,
+  ): ByteArray {
+    val body =
+      mapOf(
+        "type" to "StartupDetails",
+        "ti" to
+          mapOf(
+            "id" to "4d828a62-a417-4936-a7a6-2b3fabacecab",
+            "task_id" to "extract",
+            "dag_id" to "dag1",
+            "run_id" to "run1",
+            "try_number" to 2,
+            "map_index" to mapIndex,
+            "dag_version_id" to "4d828a62-a417-4936-a7a6-2b3fabacecab",
+          ),
+        "ti_context" to mapOf("dag_run" to mapOf("dag_id" to "dag1", "run_id" to "run1"), "max_tries" to 1) + tiContext,
+        "dag_rel_path" to "/dev/null",
+        "bundle_info" to mapOf("name" to "any-name", "version" to "any-version"),
+        "start_date" to "2024-12-01T01:00:00Z",
+        "sentry_integration" to "",
+      )
+    return Frame.encodeRequest(0, body).fold(ByteArray(0)) { acc, buffer -> acc + buffer.toByteArray() }
+  }
+
+  /**
+   * Runs [taskClass] as task "extract" of Dag "dag1" through [Server.dispatchTask]. The fake
+   * supervisor first sends StartupDetails, with [mapIndex] as the map_index of the task instance and
+   * [tiContext] added to ti_context. Until the task reports its final state, [answer] builds the
+   * reply to each request from the request ID. Returns the bodies of those requests in arrival
+   * order, and the body of the final state.
+   */
+  private fun serveTask(
+    taskClass: Class<out Task>,
+    mapIndex: Int?,
+    tiContext: Map<String, Any?>,
+    answer: (Int) -> ByteArray = ::ackFrame,
+  ): Pair<List<Map<*, *>>, Map<*, *>> {
+    val toServer = ByteChannel(autoFlush = true)
+    val fromServer = ByteChannel(autoFlush = true)
+    val comm = CoordinatorComm(toServer, fromServer)
+    val server = Server(InetSocketAddress("localhost", 0), InetSocketAddress("localhost", 0))
+    val bundle = Bundle(listOf(DagDef("dag1").addTask(TaskDef("extract", taskClass))))
+
+    val requests = mutableListOf<Map<*, *>>()
+    var terminal: Map<*, *>? = null
+    val supervisor =
+      Thread {
+        runBlocking {
+          toServer.writeFrame(startupFrame(mapIndex, tiContext))
+          while (terminal == null) {
+            val prefix = fromServer.readByteArray(4)
+            val payload = fromServer.readByteArray(Frame.parseLengthPrefix(prefix).toInt())
+            val request = Frame.decodeRaw(ArrayBufferInput(payload))
+            val body = request.rawBody as Map<*, *>
+            if (body["type"] in setOf("SucceedTask", "TaskState", "RetryTask")) {
+              terminal = body
+              toServer.writeFrame(ackFrame(request.id))
+            } else {
+              requests += body
+              toServer.writeFrame(answer(request.id))
+            }
+          }
+        }
+      }
+    supervisor.start()
+
+    runBlocking { server.dispatchTask(bundle, comm) }
+    supervisor.join()
+    comm.close()
+    return requests to checkNotNull(terminal)
+  }
+
+  private fun deleteXComRequest(
+    key: String,
+    mapIndex: Int,
+  ) = mapOf(
+    "type" to "DeleteXCom",
+    "key" to key,
+    "dag_id" to "dag1",
+    "run_id" to "run1",
+    "task_id" to "extract",
+    "map_index" to mapIndex.toLong(),
+  )
+
   @Test
   @DisplayName("Should run the task and report the result as a normal awaited request")
   @Timeout(value = 30, unit = TimeUnit.SECONDS)
@@ -112,6 +218,64 @@ class ServerTest {
     comm.close()
   }
 
+  @Test
+  @DisplayName("Should delete each listed XCom before the task sends anything")
+  @Timeout(value = 30, unit = TimeUnit.SECONDS)
+  fun deletesListedXComsBeforeTheTaskSendsAnything() {
+    val cases =
+      listOf(
+        Triple(-1, listOf("return_value", "summary"), -1),
+        Triple(0, listOf("return_value", "summary"), 0),
+        Triple(2, listOf("return_value", "summary"), 2),
+        Triple(null, listOf("return_value"), -1),
+        Triple(-1, emptyList(), -1),
+        Triple(-1, null, -1),
+      )
+    for ((mapIndex, keys, deletedMapIndex) in cases) {
+      val (requests, terminal) = serveTask(PushingTask::class.java, mapIndex, mapOf("xcom_keys_to_clear" to keys))
+
+      Assertions.assertEquals(keys.orEmpty().map { deleteXComRequest(it, deletedMapIndex) }, requests.dropLast(1)) {
+        "unexpected requests for map_index $mapIndex and keys $keys: $requests"
+      }
+      Assertions.assertEquals("SetXCom", requests.last()["type"])
+      Assertions.assertEquals("SucceedTask", terminal["type"])
+    }
+  }
+
+  @Test
+  @DisplayName("Should fail without running the task when an XCom cannot be deleted")
+  @Timeout(value = 30, unit = TimeUnit.SECONDS)
+  fun failsWithoutRunningTheTaskWhenAnXComCannotBeDeleted() {
+    for (shouldRetry in listOf(true, false)) {
+      val (requests, terminal) =
+        serveTask(
+          PushingTask::class.java,
+          -1,
+          mapOf("xcom_keys_to_clear" to listOf("return_value", "summary"), "should_retry" to shouldRetry),
+          ::errorResponseFrame,
+        )
+
+      Assertions.assertEquals(listOf(deleteXComRequest("return_value", -1)), requests) {
+        "the runtime should stop at the first XCom it cannot delete and not run the task"
+      }
+      if (shouldRetry) {
+        Assertions.assertEquals("RetryTask", terminal["type"])
+      } else {
+        Assertions.assertEquals("TaskState", terminal["type"])
+        Assertions.assertEquals("failed", terminal["state"])
+      }
+    }
+  }
+
+  class PushingTask : Task {
+    override fun execute(
+      context: Context,
+      client: Client,
+    ) {
+      client.setXCom(value = "rows")
+    }
+  }
+
   private companion object {
     // [2, msg, null] with msg coming from
     // https://github.com/astronomer/airflow/blob/f39c8da8/task-sdk/tests/task_sdk/execution_time/test_comms.py#L73-L108