Fixing bad cadwyn migration for upstream map indexes (#52797)

(cherry picked from commit 2bb5d01)

Author: amoghrajesh

RELEASE_NOTES.rst

@@ -77,6 +77,7 @@ Bug Fixes
 - Add ti information to re-queue logs (#49995)
 - Task SDK: Fix ``AssetEventOperations.get`` to use ``alias_name`` when specified (#52324)
 - Ensure trigger kwargs are properly deserialized during trigger execution (#52721)
+- Fixing bad cadwyn migration for upstream map indexes (#52797)
 
 Miscellaneous
 """""""""""""

airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_05_20.py

@@ -30,7 +30,7 @@ class DowngradeUpstreamMapIndexes(VersionChange):
     description = __doc__
 
     instructions_to_migrate_to_previous_version = (
-        schema(TIRunContext).field("upstream_map_indexes").had(type=Optional[dict[str, int]]),  # type: ignore
+        schema(TIRunContext).field("upstream_map_indexes").had(type=Optional[dict[str, Optional[int]]]),
     )
 
     @convert_response_to_previous_version_for(TIRunContext)  # type: ignore[arg-type]
@@ -42,13 +42,14 @@ def downgrade_upstream_map_indexes(response: ResponseInfo = None) -> None:  # ty
         """
         resp = response.body.get("upstream_map_indexes")
         if isinstance(resp, dict):
-            downgraded = {}
+            downgraded: dict[str, int | list | None] = {}
             for k, v in resp.items():
                 if isinstance(v, int):
                     downgraded[k] = v
                 elif isinstance(v, list) and v and all(isinstance(i, int) for i in v):
                     downgraded[k] = v[0]
                 else:
-                    # for cases like None, make it -1
-                    downgraded[k] = -1
+                    # Keep values like None as is — the Task SDK expects them unchanged during mapped task expansion,
+                    # and modifying them can cause unexpected failures.
+                    downgraded[k] = None
             response.body["upstream_map_indexes"] = downgraded

airflow-core/tests/unit/api_fastapi/execution_api/versions/v2025_04_28/test_task_instances.py

@@ -62,9 +62,16 @@ def teardown_method(self):
                 {"task_a": 3, "task_b": 9},
                 id="list of ints",
             ),
+            pytest.param(
+                [
+                    ("task_a", None),
+                ],
+                {"task_a": None},
+                id="task has no upstreams",
+            ),
             pytest.param(
                 [("task_a", None), ("task_b", [6, 7]), ("task_c", 2)],
-                {"task_a": -1, "task_b": 6, "task_c": 2},
+                {"task_a": None, "task_b": 6, "task_c": 2},
                 id="mixed types",
             ),
         ],

reproducible_build.yaml

@@ -1,2 +1,2 @@
-release-notes-hash: 3a16d9fdbc3cac54b23e0cac1e92ff10
-source-date-epoch: 1751573977
+release-notes-hash: de8144e966c67986caa98ae835947858
+source-date-epoch: 1751574576