From 6f6139899fa857627272f37a42e6c020c948c7e2 Mon Sep 17 00:00:00 2001
From: zzstoatzz <thrast36@gmail.com>
Date: Mon, 21 Oct 2024 16:29:22 -0500
Subject: [PATCH] add test

---
 src/prefect/client/orchestration.py           | 10 ++---
 .../flows/test_flow_with_mapped_tasks.py      | 42 +++++++++++++++++++
 2 files changed, 47 insertions(+), 5 deletions(-)
 create mode 100644 tests/public/flows/test_flow_with_mapped_tasks.py

diff --git a/src/prefect/client/orchestration.py b/src/prefect/client/orchestration.py
index 7b43e428bf6d..748e19205697 100644
--- a/src/prefect/client/orchestration.py
+++ b/src/prefect/client/orchestration.py
@@ -2303,11 +2303,11 @@ async def read_task_run(self, task_run_id: UUID) -> TaskRun:
     async def read_task_runs(
         self,
         *,
-        flow_filter: FlowFilter = None,
-        flow_run_filter: FlowRunFilter = None,
-        task_run_filter: TaskRunFilter = None,
-        deployment_filter: DeploymentFilter = None,
-        sort: TaskRunSort = None,
+        flow_filter: Optional[FlowFilter] = None,
+        flow_run_filter: Optional[FlowRunFilter] = None,
+        task_run_filter: Optional[TaskRunFilter] = None,
+        deployment_filter: Optional[DeploymentFilter] = None,
+        sort: Optional[TaskRunSort] = None,
         limit: Optional[int] = None,
         offset: int = 0,
     ) -> List[TaskRun]:
diff --git a/tests/public/flows/test_flow_with_mapped_tasks.py b/tests/public/flows/test_flow_with_mapped_tasks.py
new file mode 100644
index 000000000000..4a92ab0a99e9
--- /dev/null
+++ b/tests/public/flows/test_flow_with_mapped_tasks.py
@@ -0,0 +1,42 @@
+"""This is a regression test for https://github.com/PrefectHQ/prefect/issues/15747"""
+
+from typing import Tuple
+
+from prefect import flow, task
+from prefect.client.schemas.objects import FlowRun
+from prefect.context import get_run_context
+from prefect.runtime import task_run
+
+
+def generate_task_run_name() -> str:
+    return f'{task_run.task_name} - input: {task_run.parameters["input"]["number"]}'
+
+
+@task(log_prints=True, task_run_name=generate_task_run_name)
+def increment_number(input: dict) -> dict:
+    input["number"] += 1
+    input["name"] = get_run_context().task_run.name
+    print(f"increment_number - result: {input['number']}")
+    return input
+
+
+@flow
+def double_increment_flow() -> Tuple[list, FlowRun]:
+    inputs = [
+        {"number": 1, "is_even": False},
+        {"number": 2, "is_even": True},
+    ]
+
+    first_increment = increment_number.map(input=inputs)
+    second_increment = increment_number.map(input=first_increment)
+    final_results = second_increment.result()
+    print(f"Final results: {final_results}")
+    return final_results
+
+
+async def test_flow_with_mapped_tasks():
+    results = double_increment_flow()
+    assert results == [
+        {"number": 3, "is_even": False, "name": "increment_number - input: 2"},
+        {"number": 4, "is_even": True, "name": "increment_number - input: 3"},
+    ]