Do not require dag_version_id in workload TI model

airflow-core/src/airflow/executors/workloads.py

@@ -55,7 +55,6 @@ class TaskInstance(BaseModel):
     """Schema for TaskInstance with minimal required fields needed for Executors and Task SDK."""
 
     id: uuid.UUID
-    dag_version_id: uuid.UUID
     task_id: str
     dag_id: str
     run_id: str
@@ -69,6 +68,7 @@ class TaskInstance(BaseModel):
 
     parent_context_carrier: dict | None = None
     context_carrier: dict | None = None
+    dag_version_id: uuid.UUID | None = None
 
     # TODO: Task-SDK: Can we replace TastInstanceKey with just the uuid across the codebase?
     @property

airflow-core/tests/unit/executors/test_workloads.py

@@ -0,0 +1,58 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import uuid
+
+from airflow.executors.workloads import TaskInstance
+
+
+def test_taskinstance_dag_version_id_is_optional_in_model_construction():
+    # Should construct without providing dag_version_id
+    ti = TaskInstance(
+        id=uuid.uuid4(),
+        task_id="t1",
+        dag_id="d1",
+        run_id="r1",
+        try_number=1,
+        pool_slots=1,
+        queue="default",
+        priority_weight=1,
+    )
+    assert getattr(ti, "dag_version_id", None) is None
+
+
+def test_taskinstance_json_schema_marks_dag_version_id_as_nullable_optional():
+    schema = TaskInstance.model_json_schema()
+
+    # Field should exist in properties
+    assert "dag_version_id" in schema["properties"], "dag_version_id should be present in properties"
+
+    # It should not be in required
+    assert "required" in schema
+    assert "dag_version_id" not in schema["required"], "dag_version_id should not be required"
+
+    # Validate the OpenAPI/JSON Schema for optional nullable uuid
+    dag_version_schema = schema["properties"]["dag_version_id"]
+
+    # pydantic v2 emits anyOf: [{type: string, format: uuid}, {type: 'null'}]
+    assert "anyOf" in dag_version_schema, "dag_version_id should be nullable (anyOf with null)"
+    any_of = dag_version_schema["anyOf"]
+    assert any(item.get("type") == "string" and item.get("format") == "uuid" for item in any_of), (
+        "one variant must be uuid string"
+    )
+    assert any(item.get("type") == "null" for item in any_of), "one variant must be null"

providers/edge3/src/airflow/providers/edge3/openapi/v2-edge-generated.yaml

@@ -1066,10 +1066,6 @@ components:
           type: string
           format: uuid
           title: Id
-        dag_version_id:
-          type: string
-          format: uuid
-          title: Dag Version Id
         task_id:
           type: string
           title: Task Id
@@ -1107,10 +1103,15 @@ components:
             type: object
           - type: 'null'
           title: Context Carrier
+        dag_version_id:
+          anyOf:
+          - type: string
+            format: uuid
+          - type: 'null'
+          title: Dag Version Id
       type: object
       required:
       - id
-      - dag_version_id
       - task_id
       - dag_id
       - run_id

providers/edge3/src/airflow/providers/edge3/plugins/www/openapi-gen/requests/schemas.gen.ts

@@ -293,11 +293,6 @@ export const $TaskInstance = {
             format: 'uuid',
             title: 'Id'
         },
-        dag_version_id: {
-            type: 'string',
-            format: 'uuid',
-            title: 'Dag Version Id'
-        },
         task_id: {
             type: 'string',
             title: 'Task Id'
@@ -354,10 +349,22 @@ export const $TaskInstance = {
                 }
             ],
             title: 'Context Carrier'
+        },
+        dag_version_id: {
+            anyOf: [
+                {
+                    type: 'string',
+                    format: 'uuid'
+                },
+                {
+                    type: 'null'
+                }
+            ],
+            title: 'Dag Version Id'
         }
     },
     type: 'object',
-    required: ['id', 'dag_version_id', 'task_id', 'dag_id', 'run_id', 'try_number', 'pool_slots', 'queue', 'priority_weight'],
+    required: ['id', 'task_id', 'dag_id', 'run_id', 'try_number', 'pool_slots', 'queue', 'priority_weight'],
     title: 'TaskInstance',
     description: 'Schema for TaskInstance with minimal required fields needed for Executors and Task SDK.'
 } as const;

providers/edge3/src/airflow/providers/edge3/plugins/www/openapi-gen/requests/types.gen.ts

@@ -155,7 +155,6 @@ export type PushLogsBody = {
  */
 export type TaskInstance = {
     id: string;
-    dag_version_id: string;
     task_id: string;
     dag_id: string;
     run_id: string;
@@ -170,6 +169,7 @@ export type TaskInstance = {
     context_carrier?: {
     [key: string]: unknown;
 } | null;
+    dag_version_id?: string | null;
 };
 
 /**

providers/edge3/www-hash.txt

@@ -1 +1 @@
-05e9a75c13c19b0018c43055b669b2db9ef0464c1f5179a20e34add6c59ad53b
+e2cc56278007e2de0289d9dc6f22e2b5457ab65561431c18129973bd64f6c89c