Fix HITL operators failing when using notifiers

[amoghrajesh]

closes: #57437

Problem and Root cause

The root cause for this failure and resulting error like this:

[2025-10-29 19:17:32] ERROR - Task failed with exception source=task loc=task_runner.py:996
AirflowException: SMTP connection is not found.
File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 934 in run

File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/task_runner.py", line 1331 in _execute_task

File "/opt/airflow/task-sdk/src/airflow/sdk/bases/operator.py", line 416 in wrapper

File "/opt/airflow/providers/standard/src/airflow/providers/standard/operators/hitl.py", line 152 in execute

File "/opt/airflow/task-sdk/src/airflow/sdk/bases/notifier.py", line 138 in __call__

File "/opt/airflow/providers/smtp/src/airflow/providers/smtp/notifications/smtp.py", line 152 in notify

File "/opt/airflow/providers/smtp/src/airflow/providers/smtp/hooks/smtp.py", line 80 in __enter__

File "/opt/airflow/providers/smtp/src/airflow/providers/smtp/hooks/smtp.py", line 120 in get_conn

AirflowNotFoundException: The conn_id `smtp_default` isn't defined
File "/opt/airflow/providers/smtp/src/airflow/providers/smtp/hooks/smtp.py", line 118 in get_conn

File "/opt/airflow/task-sdk/src/airflow/sdk/bases/hook.py", line 61 in get_connection

File "/opt/airflow/task-sdk/src/airflow/sdk/definitions/connection.py", line 225 in get

File "/opt/airflow/task-sdk/src/airflow/sdk/execution_time/context.py", line 175 in _get_connection

was a type mismatch between API response and client expectations in SDK.

The client's add_response() method tried to parse API JSON directly as HITLDetailRequestResult:

But the API returns JSON without a type field:

{
  "ti_id": "...",
  "options": [...],
  "subject": "...",
  // NO "type" field
}

HITLDetailRequestResult is part of a discriminated union (ToTask) that requires a type discriminator field for Pydantic to deserialize correctly. Since the API doesn't include this field, validation failed.

This issue can be fixed by correcting the types sent across between client <=> supervisor. Client returns HITLDetailRequest, supervisor converts to HITLDetailRequestResult (similar pattern to XCom req/resp) by using a from_api_response defined on HITLDetailRequestResult

Testing

Used this DAG:

from airflow import DAG
from airflow.providers.smtp.notifications.smtp import send_smtp_notification
from datetime import datetime

from airflow.providers.standard.operators.empty import EmptyOperator
from airflow.providers.standard.operators.hitl import HITLBranchOperator

with DAG(
    dag_id="hitl_example",
    schedule=None,
    start_date=datetime(2024, 1, 1),
    catchup=False,
) as dag:

    send_invoices_export = EmptyOperator(
        task_id="send_invoices_export",
    )

    abort_export_creation = EmptyOperator(
        task_id="abort_export_creation",
    )

    review_export_task = HITLBranchOperator(
            task_id = "review_export",
            subject="Please check the export provided in S3 bucket. See the previous task for detailed errors.",
            options=["send_invoices_export", "abort_export_creation"],
            notifiers=[
                send_smtp_notification(
                    from_email="your_mail@example.com",
                    to="my_mail@example.com",
                    subject="Export needs to be checked",
                    html_content="Please check the export provided in S3 bucket"
                )
            ]
        )

    review_export_task >> [abort_export_creation, send_invoices_export]

Created an SMTP connection with host as: mailpit, port as 1025

Flow:

Notifier sent this email:

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[github-actions]

Backport successfully created: v3-1-test

Status	Branch	Result
✅	v3-1-test