Make the HITLOperator summary extension point public

providers/standard/src/airflow/providers/standard/operators/hitl.py

@@ -133,15 +133,17 @@ def __init__(
         self.validate_defaults()
 
         # Runtime/subclass additions to the summary; config-derived entries live in the property.
-        self._hitl_summary_extra: dict[str, Any] = {}
+        self.hitl_summary_extra: dict[str, Any] = {}
 
     @property
     def hitl_summary(self) -> dict[str, Any]:
         """
         Summary of the Human-in-the-loop request, for listeners/observability.
 
         A property so the ``subject``/``body`` template fields are read after rendering, not
-        captured as un-rendered Jinja in ``__init__``.
+        captured as un-rendered Jinja in ``__init__``. Each access builds a fresh snapshot, so
+        mutating the returned dict has no effect — runtime code and subclasses extend the summary
+        by adding entries to ``hitl_summary_extra`` instead.
         """
         return {
             "subject": self.subject,
@@ -151,7 +153,7 @@ def hitl_summary(self) -> dict[str, Any]:
             "multiple": self.multiple,
             "assigned_users": self.assigned_users,
             "serialized_params": self.serialized_params or None,
-            **self._hitl_summary_extra,
+            **self.hitl_summary_extra,
         }
 
     def validate_options(self) -> None:
@@ -221,7 +223,7 @@ def execute(self, context: Context):
             timeout_datetime = None
 
         # Enrich summary with runtime info
-        self._hitl_summary_extra["timeout_datetime"] = (
+        self.hitl_summary_extra["timeout_datetime"] = (
             timeout_datetime.isoformat() if timeout_datetime else None
         )
 
@@ -259,15 +261,15 @@ def serialized_params(self) -> dict[str, dict[str, Any]]:
 
     def execute_complete(self, context: Context, event: dict[str, Any]) -> Any:
         if "error" in event:
-            self._hitl_summary_extra["error_type"] = event["error_type"]
+            self.hitl_summary_extra["error_type"] = event["error_type"]
             self.process_trigger_event_error(event)
 
         chosen_options = event["chosen_options"]
         params_input = event["params_input"] or {}
         self.validate_chosen_options(chosen_options)
         self.validate_params_input(params_input)
 
-        self._hitl_summary_extra.update(
+        self.hitl_summary_extra.update(
             {
                 "chosen_options": chosen_options,
                 "params_input": params_input,
@@ -445,14 +447,14 @@ def __init__(
             **kwargs,
         )
 
-        self._hitl_summary_extra["ignore_downstream_trigger_rules"] = self.ignore_downstream_trigger_rules
-        self._hitl_summary_extra["fail_on_reject"] = self.fail_on_reject
+        self.hitl_summary_extra["ignore_downstream_trigger_rules"] = self.ignore_downstream_trigger_rules
+        self.hitl_summary_extra["fail_on_reject"] = self.fail_on_reject
 
     def execute_complete(self, context: Context, event: dict[str, Any]) -> Any:
         ret = super().execute_complete(context=context, event=event)
 
         chosen_option = ret["chosen_options"][0]
-        self._hitl_summary_extra["approved"] = chosen_option == self.APPROVE
+        self.hitl_summary_extra["approved"] = chosen_option == self.APPROVE
         if chosen_option == self.APPROVE:
             self.log.info("Approved. Proceeding with downstream tasks...")
             return ret
@@ -506,7 +508,7 @@ def __init__(self, *, options_mapping: dict[str, str] | None = None, **kwargs) -
         super().__init__(**kwargs)
         self.options_mapping = options_mapping or {}
         self.validate_options_mapping()
-        self._hitl_summary_extra["options_mapping"] = self.options_mapping
+        self.hitl_summary_extra["options_mapping"] = self.options_mapping
 
     def validate_options_mapping(self) -> None:
         """
@@ -541,7 +543,7 @@ def execute_complete(self, context: Context, event: dict[str, Any]) -> Any:
 
         # Map options to task IDs using the mapping, fallback to original option
         chosen_options = [self.options_mapping.get(option, option) for option in chosen_options]
-        self._hitl_summary_extra["branches_to_execute"] = chosen_options
+        self.hitl_summary_extra["branches_to_execute"] = chosen_options
         return self.do_branch(context=context, branches_to_execute=chosen_options)
 
 

providers/standard/tests/unit/standard/operators/test_hitl.py

@@ -989,6 +989,17 @@ def test_summary_reflects_rendered_subject_body(self) -> None:
         assert op.hitl_summary["subject"] == "Review for 2020-01-01"
         assert op.hitl_summary["body"] == "Deploy 2020-01-01?"
 
+    def test_summary_extension_via_hitl_summary_extra(self) -> None:
+        """Subclasses and runtime code extend the summary through the public hitl_summary_extra dict."""
+        op = HITLOperator(
+            task_id="test",
+            subject="Review",
+            options=["A", "B"],
+        )
+        op.hitl_summary_extra["team"] = "data-eng"
+
+        assert op.hitl_summary["team"] == "data-eng"
+
     def test_approval_operator_init_summary(self) -> None:
         """ApprovalOperator hitl_summary includes base + approval-specific fields."""
         op = ApprovalOperator(