Partially revert #50825 on database level

airflow-core/docs/img/airflow_erd.sha256

@@ -1 +1 @@
-625f362919679fe85b2bfb3f9e053261248ac6ec7a974eee51012e55a8105b94
+b31700355c6c8e073896b9daba68b09c252bf1a05f535ef40aa0cb927d12e9ce

airflow-core/docs/migrations-ref.rst

@@ -47,9 +47,7 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``09fa89ba1710``        | ``40f7c30a228b`` | ``3.1.0``         | Add trigger_id to deadline.                                  |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
-| ``40f7c30a228b``        | ``5d3072c51bac`` | ``3.1.0``         | Add Human In the Loop Detail table.                          |
-+-------------------------+------------------+-------------------+--------------------------------------------------------------+
-| ``5d3072c51bac``        | ``ffdb0566c7c0`` | ``3.1.0``         | Make dag_version_id non-nullable in TaskInstance.            |
+| ``40f7c30a228b``        | ``ffdb0566c7c0`` | ``3.1.0``         | Add Human In the Loop Detail table.                          |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``ffdb0566c7c0``        | ``66a7743fe20e`` | ``3.1.0``         | Add dag_favorite table.                                      |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

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

@@ -44,7 +44,6 @@ class TaskInstanceResponse(BaseModel):
     id: str
     task_id: str
     dag_id: str
-    dag_version: DagVersionResponse
     run_id: str = Field(alias="dag_run_id")
     map_index: int
     logical_date: datetime | None
@@ -77,6 +76,7 @@ class TaskInstanceResponse(BaseModel):
     )
     trigger: TriggerResponse | None
     queued_by_job: JobResponse | None = Field(alias="triggerer_job")
+    dag_version: DagVersionResponse | None
 
 
 class TaskInstanceCollectionResponse(BaseModel):

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -11202,8 +11202,6 @@ components:
         dag_id:
           type: string
           title: Dag Id
-        dag_version:
-          $ref: '#/components/schemas/DagVersionResponse'
         dag_run_id:
           type: string
           title: Dag Run Id
@@ -11331,12 +11329,15 @@ components:
           anyOf:
           - $ref: '#/components/schemas/JobResponse'
           - type: 'null'
+        dag_version:
+          anyOf:
+          - $ref: '#/components/schemas/DagVersionResponse'
+          - type: 'null'
       type: object
       required:
       - id
       - task_id
       - dag_id
-      - dag_version
       - dag_run_id
       - map_index
       - logical_date
@@ -11365,6 +11366,7 @@ components:
       - rendered_map_index
       - trigger
       - triggerer_job
+      - dag_version
       title: TaskInstanceResponse
       description: TaskInstance serializer for responses.
     TaskInstanceState:

airflow-core/src/airflow/migrations/versions/0077_3_1_0_add_human_in_the_loop_response.py → airflow-core/src/airflow/migrations/versions/0076_3_1_0_add_human_in_the_loop_response.py

@@ -20,7 +20,7 @@
 Add Human In the Loop Detail table.
 
 Revision ID: 40f7c30a228b
-Revises: 5d3072c51bac
+Revises: ffdb0566c7c0
 Create Date: 2025-07-04 15:05:19.459197
 
 """
@@ -37,7 +37,7 @@
 
 # revision identifiers, used by Alembic.
 revision = "40f7c30a228b"
-down_revision = "5d3072c51bac"
+down_revision = "ffdb0566c7c0"
 branch_labels = None
 depends_on = None
 airflow_version = "3.1.0"

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

@@ -527,7 +527,8 @@ class TaskInstance(Base, LoggingMixin):
 
     _task_display_property_value = Column("task_display_name", String(2000), nullable=True)
     dag_version_id = Column(
-        UUIDType(binary=False), ForeignKey("dag_version.id", ondelete="RESTRICT"), nullable=False
+        UUIDType(binary=False),
+        ForeignKey("dag_version.id", ondelete="RESTRICT"),
     )
     dag_version = relationship("DagVersion", back_populates="task_instances")
 

airflow-core/src/airflow/ui/openapi-gen/requests/schemas.gen.ts

@@ -4752,9 +4752,6 @@ export const $TaskInstanceResponse = {
             type: 'string',
             title: 'Dag Id'
         },
-        dag_version: {
-            '$ref': '#/components/schemas/DagVersionResponse'
-        },
         dag_run_id: {
             type: 'string',
             title: 'Dag Run Id'
@@ -5000,10 +4997,20 @@ export const $TaskInstanceResponse = {
                     type: 'null'
                 }
             ]
+        },
+        dag_version: {
+            anyOf: [
+                {
+                    '$ref': '#/components/schemas/DagVersionResponse'
+                },
+                {
+                    type: 'null'
+                }
+            ]
         }
     },
     type: 'object',
-    required: ['id', 'task_id', 'dag_id', 'dag_version', 'dag_run_id', 'map_index', 'logical_date', 'run_after', 'start_date', 'end_date', 'duration', 'state', 'try_number', 'max_tries', 'task_display_name', 'dag_display_name', 'hostname', 'unixname', 'pool', 'pool_slots', 'queue', 'priority_weight', 'operator', 'queued_when', 'scheduled_when', 'pid', 'executor', 'executor_config', 'note', 'rendered_map_index', 'trigger', 'triggerer_job'],
+    required: ['id', 'task_id', 'dag_id', 'dag_run_id', 'map_index', 'logical_date', 'run_after', 'start_date', 'end_date', 'duration', 'state', 'try_number', 'max_tries', 'task_display_name', 'dag_display_name', 'hostname', 'unixname', 'pool', 'pool_slots', 'queue', 'priority_weight', 'operator', 'queued_when', 'scheduled_when', 'pid', 'executor', 'executor_config', 'note', 'rendered_map_index', 'trigger', 'triggerer_job', 'dag_version'],
     title: 'TaskInstanceResponse',
     description: 'TaskInstance serializer for responses.'
 } as const;

airflow-core/src/airflow/ui/openapi-gen/requests/types.gen.ts

@@ -1299,7 +1299,6 @@ export type TaskInstanceResponse = {
     id: string;
     task_id: string;
     dag_id: string;
-    dag_version: DagVersionResponse;
     dag_run_id: string;
     map_index: number;
     logical_date: string | null;
@@ -1331,6 +1330,7 @@ export type TaskInstanceResponse = {
     };
     trigger: TriggerResponse | null;
     triggerer_job: JobResponse | null;
+    dag_version: DagVersionResponse | null;
 };
 
 /**

airflow-core/src/airflow/ui/src/components/Clear/TaskInstance/ClearTaskInstanceDialog.tsx

@@ -98,7 +98,7 @@ const ClearTaskInstanceDialog = ({ onClose, open, taskInstance }: Props) => {
 
   // Check if bundle versions are different
   const currentDagBundleVersion = dagDetails?.bundle_version;
-  const taskInstanceDagVersionBundleVersion = taskInstance.dag_version.bundle_version;
+  const taskInstanceDagVersionBundleVersion = taskInstance.dag_version?.bundle_version;
   const bundleVersionsDiffer = currentDagBundleVersion !== taskInstanceDagVersionBundleVersion;
   const shouldShowBundleVersionOption =
     bundleVersionsDiffer &&

airflow-core/src/airflow/ui/src/hooks/useSelectedVersion.ts

@@ -82,7 +82,7 @@ const useSelectedVersion = (): number | undefined => {
 
   const selectedVersionNumber =
     selectedVersionUrl ??
-    (mappedTaskInstanceData ? mappedTaskInstanceData.dag_version.version_number : undefined) ??
+    mappedTaskInstanceData?.dag_version?.version_number ??
     (runData?.dag_versions ?? []).at(-1)?.version_number ??
     dagData?.latest_dag_version?.version_number;
 

airflow-ctl/src/airflowctl/api/datamodels/generated.py

@@ -1618,7 +1618,6 @@ class TaskInstanceResponse(BaseModel):
     id: Annotated[str, Field(title="Id")]
     task_id: Annotated[str, Field(title="Task Id")]
     dag_id: Annotated[str, Field(title="Dag Id")]
-    dag_version: DagVersionResponse
     dag_run_id: Annotated[str, Field(title="Dag Run Id")]
     map_index: Annotated[int, Field(title="Map Index")]
     logical_date: Annotated[datetime | None, Field(title="Logical Date")] = None
@@ -1648,6 +1647,7 @@ class TaskInstanceResponse(BaseModel):
     rendered_fields: Annotated[dict[str, Any] | None, Field(title="Rendered Fields")] = None
     trigger: TriggerResponse | None = None
     triggerer_job: JobResponse | None = None
+    dag_version: DagVersionResponse | None = None
 
 
 class TaskResponse(BaseModel):