feat(hitl): include task_instance detail in hitl detail response

Author: Lee-W

airflow-core/src/airflow/api_fastapi/core_api/datamodels/hitl.py

@@ -23,6 +23,7 @@
 from pydantic import Field, field_validator
 
 from airflow.api_fastapi.core_api.base import BaseModel
+from airflow.api_fastapi.core_api.datamodels.task_instances import TaskInstanceResponse
 from airflow.sdk import Param
 
 
@@ -45,7 +46,7 @@ class HITLDetailResponse(BaseModel):
 class HITLDetail(BaseModel):
     """Schema for Human-in-the-loop detail."""
 
-    ti_id: str
+    task_instance: TaskInstanceResponse
 
     # User Request Detail
     options: list[str]

airflow-core/src/airflow/api_fastapi/core_api/routes/public/hitl.py

@@ -19,6 +19,7 @@
 import structlog
 from fastapi import Depends, HTTPException, status
 from sqlalchemy import select
+from sqlalchemy.orm import joinedload
 
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
 from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
@@ -132,7 +133,11 @@ def _get_hitl_detail(
         )
 
     ti_id_str = str(task_instance.id)
-    hitl_detail_model = session.scalar(select(HITLDetailModel).where(HITLDetailModel.ti_id == ti_id_str))
+    hitl_detail_model = session.scalar(
+        select(HITLDetailModel)
+        .where(HITLDetailModel.ti_id == ti_id_str)
+        .options(joinedload(HITLDetailModel.task_instance))
+    )
     if not hitl_detail_model:
         log.error("Human-in-the-loop detail not found")
         raise HTTPException(

airflow-core/src/airflow/models/hitl.py

@@ -20,6 +20,7 @@
 from sqlalchemy import Boolean, Column, ForeignKeyConstraint, String, Text
 from sqlalchemy.dialects import postgresql
 from sqlalchemy.ext.hybrid import hybrid_property
+from sqlalchemy.orm import relationship
 
 from airflow.models.base import Base
 from airflow.settings import json
@@ -53,6 +54,11 @@ class HITLDetail(Base):
         default=None,
     )
     params_input = Column(sqlalchemy_jsonfield.JSONField(json=json), nullable=False, default={})
+    task_instance = relationship(
+        "TaskInstance",
+        lazy="joined",
+        back_populates="hitl_detail",
+    )
 
     __table_args__ = (
         ForeignKeyConstraint(

airflow-core/src/airflow/models/taskinstance.py

@@ -555,6 +555,8 @@ class TaskInstance(Base, LoggingMixin):
     triggerer_job = association_proxy("trigger", "triggerer_job")
     dag_run = relationship("DagRun", back_populates="task_instances", lazy="joined", innerjoin=True)
     rendered_task_instance_fields = relationship("RenderedTaskInstanceFields", lazy="noload", uselist=False)
+    hitl_detail = relationship("HITLDetail", lazy="noload", uselist=False)
+
     run_after = association_proxy("dag_run", "run_after")
     logical_date = association_proxy("dag_run", "logical_date")
     task_instance_note = relationship(

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl.py

@@ -16,6 +16,8 @@
 # under the License.
 from __future__ import annotations
 
+from unittest import mock
+
 import pytest
 from sqlalchemy.orm import Session
 
@@ -111,8 +113,51 @@ def expected_sample_hitl_detail_dict(sample_ti: TaskInstance) -> dict[str, Any]:
         "chosen_options": None,
         "response_received": False,
         "subject": "This is subject",
-        "ti_id": sample_ti.id,
         "user_id": None,
+        "task_instance": {
+            "dag_display_name": "dag",
+            "dag_id": "dag",
+            "dag_run_id": "test",
+            "dag_version": {
+                "bundle_name": "dag_maker",
+                "bundle_url": None,
+                "bundle_version": None,
+                "created_at": mock.ANY,
+                "dag_display_name": "dag",
+                "dag_id": "dag",
+                "id": mock.ANY,
+                "version_number": 1,
+            },
+            "duration": None,
+            "end_date": None,
+            "executor": None,
+            "executor_config": "{}",
+            "hostname": "",
+            "id": sample_ti.id,
+            "logical_date": mock.ANY,
+            "map_index": -1,
+            "max_tries": 0,
+            "note": None,
+            "operator": "EmptyOperator",
+            "pid": None,
+            "pool": "default_pool",
+            "pool_slots": 1,
+            "priority_weight": 1,
+            "queue": "default",
+            "queued_when": None,
+            "rendered_fields": {},
+            "rendered_map_index": None,
+            "run_after": mock.ANY,
+            "scheduled_when": None,
+            "start_date": None,
+            "state": None,
+            "task_display_name": "op1",
+            "task_id": "op1",
+            "trigger": None,
+            "triggerer_job": None,
+            "try_number": 0,
+            "unixname": "root",
+        },
     }