Read HITLOperator subject/body into the summary lazily

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

@@ -132,15 +132,26 @@ def __init__(
         self.validate_params()
         self.validate_defaults()
 
-        # HITL summary for the use of listeners; subclasses can extend it.
-        self.hitl_summary: dict[str, Any] = {
+        # Runtime/subclass additions to the summary; config-derived entries live in the property.
+        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__``.
+        """
+        return {
             "subject": self.subject,
             "body": self.body,
             "options": self.options,
             "defaults": self.defaults,
             "multiple": self.multiple,
             "assigned_users": self.assigned_users,
             "serialized_params": self.serialized_params or None,
+            **self._hitl_summary_extra,
         }
 
     def validate_options(self) -> None:
@@ -210,7 +221,9 @@ def execute(self, context: Context):
             timeout_datetime = None
 
         # Enrich summary with runtime info
-        self.hitl_summary["timeout_datetime"] = timeout_datetime.isoformat() if timeout_datetime else None
+        self._hitl_summary_extra["timeout_datetime"] = (
+            timeout_datetime.isoformat() if timeout_datetime else None
+        )
 
         self.log.info("Waiting for response")
         for notifier in self.notifiers:
@@ -246,15 +259,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["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.update(
+        self._hitl_summary_extra.update(
             {
                 "chosen_options": chosen_options,
                 "params_input": params_input,
@@ -432,14 +445,14 @@ def __init__(
             **kwargs,
         )
 
-        self.hitl_summary["ignore_downstream_trigger_rules"] = self.ignore_downstream_trigger_rules
-        self.hitl_summary["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["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
@@ -493,7 +506,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["options_mapping"] = self.options_mapping
+        self._hitl_summary_extra["options_mapping"] = self.options_mapping
 
     def validate_options_mapping(self) -> None:
         """
@@ -528,7 +541,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["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

@@ -974,6 +974,21 @@ def test_hitl_operator_init_minimal(self) -> None:
             "serialized_params": None,
         }
 
+    def test_summary_reflects_rendered_subject_body(self) -> None:
+        """The summary reads subject/body live, so it reflects rendered values (guards #70296)."""
+        op = HITLOperator(
+            task_id="test",
+            subject="Review for {{ ds }}",
+            body="Deploy {{ ds }}?",
+            options=["Yes", "No"],
+        )
+        # Airflow renders template fields in place before execute.
+        op.subject = "Review for 2020-01-01"
+        op.body = "Deploy 2020-01-01?"
+
+        assert op.hitl_summary["subject"] == "Review for 2020-01-01"
+        assert op.hitl_summary["body"] == "Deploy 2020-01-01?"
+
     def test_approval_operator_init_summary(self) -> None:
         """ApprovalOperator hitl_summary includes base + approval-specific fields."""
         op = ApprovalOperator(
@@ -1382,7 +1397,8 @@ def test_full_lifecycle_approval(self) -> None:
             },
         )
 
-        assert s == {
+        # hitl_summary is a property, so re-read it to see the execute_complete additions.
+        assert op.hitl_summary == {
             "subject": "Release v2.0?",
             "body": "Please approve the production deployment.",
             "options": ["Approve", "Reject"],

scripts/ci/prek/validate_operators_init_exemptions.txt

@@ -84,7 +84,6 @@ providers/snowflake/src/airflow/providers/snowflake/operators/snowpark_container
 providers/ssh/src/airflow/providers/ssh/operators/ssh.py::SSHOperator
 providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py::SSHRemoteJobOperator
 providers/standard/src/airflow/providers/standard/operators/bash.py::BashOperator
-providers/standard/src/airflow/providers/standard/operators/hitl.py::HITLOperator
 providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py::TriggerDagRunOperator
 providers/standard/src/airflow/providers/standard/sensors/date_time.py::DateTimeSensor
 providers/teradata/src/airflow/providers/teradata/transfers/teradata_to_teradata.py::TeradataToTeradataOperator