From 2dfc9449bb1a92b19b814a3de1feb01eee7c5191 Mon Sep 17 00:00:00 2001
From: 1fanwang <1fannnw@gmail.com>
Date: Wed, 12 Aug 2026 14:37:31 -0700
Subject: [PATCH 1/2] Let a link opt into keeping one row per attempt

Approach B for #71471: the link decides, rather than core keeping a row per attempt
for every link. BaseOperatorLink gains keeps_a_link_per_attempt and xcom_key_for_try,
and only links that set the flag get extra rows.

The flag has to survive DAG serialization, since the api-server reads links through
XComOperatorLink, which is rebuilt from {name: xcom_key} and knows nothing about the
original class. That mapping now carries {key, per_attempt} for opted-in links and
stays a bare string otherwise.

Signed-off-by: 1fanwang <1fannnw@gmail.com>
---
 airflow-core/newsfragments/71471.bugfix.rst   |  1 +
 .../execution_api/routes/task_instances.py    |  8 +++++++
 .../serialization/definitions/operatorlink.py |  9 ++++++-
 .../serialization/serialized_objects.py       | 18 +++++++++++---
 .../src/airflow/sdk/bases/operatorlink.py     | 24 +++++++++++++++++++
 .../airflow/sdk/execution_time/task_runner.py |  3 ++-
 6 files changed, 58 insertions(+), 5 deletions(-)
 create mode 100644 airflow-core/newsfragments/71471.bugfix.rst

diff --git a/airflow-core/newsfragments/71471.bugfix.rst b/airflow-core/newsfragments/71471.bugfix.rst
new file mode 100644
index 0000000000000..022853ec95754
--- /dev/null
+++ b/airflow-core/newsfragments/71471.bugfix.rst
@@ -0,0 +1 @@
+Return each attempt's own URL from operator extra links, for links that opt in.
diff --git a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index 02d723a82185e..988e0a398ef59 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -94,6 +94,7 @@
 from airflow.models.taskreschedule import TaskReschedule
 from airflow.models.trigger import Trigger, handle_event_submit
 from airflow.models.xcom import XComModel
+from airflow.sdk.bases.operatorlink import ATTEMPT_LINK_XCOM_KEY_PREFIX
 from airflow.serialization.definitions.assets import SerializedAsset, SerializedAssetUniqueKey
 from airflow.state import get_state_backend
 from airflow.triggers.base import TriggerEvent
@@ -296,6 +297,13 @@ def ti_run(
             if map_index is not None:
                 xcom_query = xcom_query.where(XComModel.map_index == map_index)
 
+            # A link that keeps one row per attempt describes an attempt that already ran,
+            # so those rows outlive the clear. The underscores are LIKE single-character
+            # wildcards and must be escaped, or unrelated keys are spared too.
+            xcom_query = xcom_query.where(
+                ~XComModel.key.like(ATTEMPT_LINK_XCOM_KEY_PREFIX.replace("_", r"\_") + "%", escape="\\")
+            )
+
             xcom_keys = list(session.scalars(xcom_query))
         task_reschedule_count = (
             session.scalar(
diff --git a/airflow-core/src/airflow/serialization/definitions/operatorlink.py b/airflow-core/src/airflow/serialization/definitions/operatorlink.py
index b7a6eef11b84f..753ece1300c31 100644
--- a/airflow-core/src/airflow/serialization/definitions/operatorlink.py
+++ b/airflow-core/src/airflow/serialization/definitions/operatorlink.py
@@ -24,6 +24,7 @@
 import attrs
 
 from airflow.models.xcom import XComModel
+from airflow.sdk.bases.operatorlink import attempt_link_xcom_key
 from airflow.utils.log.logging_mixin import LoggingMixin
 from airflow.utils.session import create_session
 
@@ -42,6 +43,12 @@ class XComOperatorLink(LoggingMixin):
 
     name: str
     xcom_key: str
+    per_attempt: bool = False
+
+    def xcom_key_for_try(self, try_number: int) -> str:
+        if not self.per_attempt:
+            return self.xcom_key
+        return attempt_link_xcom_key(self.xcom_key, try_number)
 
     def get_link(self, operator: Operator, *, ti_key: TaskInstanceKey) -> str:
         """
@@ -57,7 +64,7 @@ def get_link(self, operator: Operator, *, ti_key: TaskInstanceKey) -> str:
         with create_session() as session:
             result = session.execute(
                 XComModel.get_many(
-                    key=self.xcom_key,
+                    key=self.xcom_key_for_try(ti_key.try_number),
                     run_id=ti_key.run_id,
                     dag_ids=ti_key.dag_id,
                     task_ids=ti_key.task_id,
diff --git a/airflow-core/src/airflow/serialization/serialized_objects.py b/airflow-core/src/airflow/serialization/serialized_objects.py
index 3a0a9278ae2e4..5a8be3ef5d699 100644
--- a/airflow-core/src/airflow/serialization/serialized_objects.py
+++ b/airflow-core/src/airflow/serialization/serialized_objects.py
@@ -1512,7 +1512,7 @@ def _is_excluded(cls, var: Any, attrname: str, op: SDKDAGNode) -> bool:
 
     @classmethod
     def _deserialize_operator_extra_links(
-        cls, encoded_op_links: dict[str, str]
+        cls, encoded_op_links: dict[str, str | dict[str, Any]]
     ) -> dict[str, XComOperatorLink]:
         """
         Deserialize Operator Links if the Classes are registered in Airflow Plugins.
@@ -1540,7 +1540,14 @@ def _deserialize_operator_extra_links(
             #     'raise_error': 'key'
             # }
 
-            op_predefined_extra_link = XComOperatorLink(name=name, xcom_key=xcom_key)
+            # A link that keeps one row per attempt serializes as a mapping; everything
+            # else stays the bare key string it has always been.
+            if isinstance(xcom_key, dict):
+                op_predefined_extra_link = XComOperatorLink(
+                    name=name, xcom_key=xcom_key["key"], per_attempt=xcom_key.get("per_attempt", False)
+                )
+            else:
+                op_predefined_extra_link = XComOperatorLink(name=name, xcom_key=xcom_key)
             op_predefined_extra_links.update({op_predefined_extra_link.name: op_predefined_extra_link})
 
         return op_predefined_extra_links
@@ -1560,7 +1567,12 @@ def _serialize_operator_extra_links(
         :param operator_extra_links: Operator Link
         :return: Serialized Operator Link
         """
-        return {link.name: link.xcom_key for link in operator_extra_links}
+        return {
+            link.name: {"key": link.xcom_key, "per_attempt": link.keeps_a_link_per_attempt}
+            if link.keeps_a_link_per_attempt
+            else link.xcom_key
+            for link in operator_extra_links
+        }
 
     @classmethod
     def serialize(cls, var: Any, *, strict: bool = False) -> Any:
diff --git a/task-sdk/src/airflow/sdk/bases/operatorlink.py b/task-sdk/src/airflow/sdk/bases/operatorlink.py
index ebc5a4dcd04db..bcd64acc020e1 100644
--- a/task-sdk/src/airflow/sdk/bases/operatorlink.py
+++ b/task-sdk/src/airflow/sdk/bases/operatorlink.py
@@ -27,6 +27,15 @@
     from airflow.sdk.types import TaskInstanceKey
 
 
+ATTEMPT_LINK_XCOM_KEY_PREFIX = "_link_attempt_"
+"""Prefix marking the XCom rows that hold one attempt's rendered operator link."""
+
+
+def attempt_link_xcom_key(xcom_key: str, try_number: int) -> str:
+    """Return the XCom key holding ``xcom_key``'s link as rendered for ``try_number``."""
+    return f"{ATTEMPT_LINK_XCOM_KEY_PREFIX}{try_number}_{xcom_key}"
+
+
 @attrs.define()
 class BaseOperatorLink(metaclass=ABCMeta):
     """Abstract base class that defines how we get an operator link."""
@@ -55,6 +64,21 @@ def xcom_key(self) -> str:
         """
         return f"_link_{self.__class__.__name__}"
 
+    keeps_a_link_per_attempt: ClassVar[bool] = False
+    """
+    Keep one row per attempt rather than a single row holding whichever ran last.
+
+    Set on links whose URL cannot be recomputed, such as one carrying a job id the remote
+    service mints per submission. A task's XComs are cleared before each attempt; the rows
+    behind a link that sets this are kept.
+    """
+
+    def xcom_key_for_try(self, try_number: int) -> str:
+        """Return the XCom key holding this link as rendered for ``try_number``."""
+        if not self.keeps_a_link_per_attempt:
+            return self.xcom_key
+        return attempt_link_xcom_key(self.xcom_key, try_number)
+
     @abstractmethod
     def get_link(self, operator: BaseOperator, *, ti_key: TaskInstanceKey) -> str:
         """
diff --git a/task-sdk/src/airflow/sdk/execution_time/task_runner.py b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
index 1ce848128d663..aa35ecbef0fe4 100644
--- a/task-sdk/src/airflow/sdk/execution_time/task_runner.py
+++ b/task-sdk/src/airflow/sdk/execution_time/task_runner.py
@@ -2326,7 +2326,8 @@ def finalize(
     # Pushing xcom for each operator extra links defined on the operator only.
     for oe in task.operator_extra_links:
         try:
-            link, xcom_key = oe.get_link(operator=task, ti_key=ti), oe.xcom_key  # type: ignore[arg-type]
+            link = oe.get_link(operator=task, ti_key=ti)  # type: ignore[arg-type]
+            xcom_key = oe.xcom_key_for_try(ti.try_number)
             log.debug("Setting xcom for operator extra link", link=link, xcom_key=xcom_key)
             _xcom_push_to_db(ti, key=xcom_key, value=link)
         except Exception:

From 55e9c56188fb419a4932869efe8b19ecc679909a Mon Sep 17 00:00:00 2001
From: 1fanwang <1fannnw@gmail.com>
Date: Wed, 12 Aug 2026 16:44:55 -0700
Subject: [PATCH 2/2] Drop the newsfragment while this is a draft

Signed-off-by: 1fanwang <1fannnw@gmail.com>
---
 airflow-core/newsfragments/71471.bugfix.rst | 1 -
 1 file changed, 1 deletion(-)
 delete mode 100644 airflow-core/newsfragments/71471.bugfix.rst

diff --git a/airflow-core/newsfragments/71471.bugfix.rst b/airflow-core/newsfragments/71471.bugfix.rst
deleted file mode 100644
index 022853ec95754..0000000000000
--- a/airflow-core/newsfragments/71471.bugfix.rst
+++ /dev/null
@@ -1 +0,0 @@
-Return each attempt's own URL from operator extra links, for links that opt in.