Fix run_deployment execution graph display when called from tasks (#19361)

Co-authored-by: Claude <[email protected]>

Author: desertaxle

src/prefect/deployments/flow_runs.py

@@ -7,7 +7,7 @@
 
 import prefect
 from prefect._result_records import ResultRecordMetadata
-from prefect.client.schemas import FlowRun
+from prefect.client.schemas import FlowRun, TaskRunResult
 from prefect.client.utilities import inject_client
 from prefect.context import FlowRunContext, TaskRunContext
 from prefect.logging import get_logger
@@ -136,6 +136,21 @@ async def run_deployment(
             k: await collect_task_run_inputs(v) for k, v in parameters.items()
         }
 
+        # Track parent task if this is being called from within a task
+        # This enables the execution graph to properly display the deployment
+        # flow run as nested under the calling task
+        if task_run_ctx:
+            # The task run is only considered a parent if it is in the same
+            # flow run (otherwise the child is in a subflow, so the subflow
+            # serves as the parent) or if there is no flow run
+            if not flow_run_ctx or (
+                task_run_ctx.task_run.flow_run_id
+                == getattr(flow_run_ctx.flow_run, "id", None)
+            ):
+                task_inputs["__parents__"] = [
+                    TaskRunResult(id=task_run_ctx.task_run.id)
+                ]
+
         if deployment_id:
             flow = await client.read_flow(deployment.flow_id)
             deployment_name = f"{flow.name}/{deployment.name}"

tests/deployment/test_flow_runs.py

@@ -445,6 +445,53 @@ async def foo():
             ]
         }
 
+    async def test_tracks_parent_task_when_called_from_task(
+        self, test_deployment, use_hosted_api_server, prefect_client
+    ):
+        """
+        Test that when run_deployment is called from within a task,
+        the wrapper task run tracks the calling task as a parent
+        in task_inputs["__parents__"].
+
+        This is important for the execution graph to correctly display
+        the deployment flow run as nested under the calling task.
+
+        Regression test for https://github.com/PrefectHQ/prefect/issues/19359
+        """
+        deployment = test_deployment
+
+        @task
+        async def trigger_deployment():
+            return await run_deployment(
+                f"foo/{deployment.name}",
+                timeout=0,
+                poll_interval=0,
+            )
+
+        @flow
+        async def parent_flow():
+            return await trigger_deployment(return_state=True)
+
+        parent_state = await parent_flow(return_state=True)
+        calling_task_state = await parent_state.result()
+        child_flow_run = await calling_task_state.result()
+
+        # The wrapper task run should exist and be linked to the child flow run
+        assert child_flow_run.parent_task_run_id is not None
+        wrapper_task_run = await prefect_client.read_task_run(
+            child_flow_run.parent_task_run_id
+        )
+
+        # The wrapper task run should have the calling task as its parent
+        # in the __parents__ field, which enables proper execution graph display
+        assert "__parents__" in wrapper_task_run.task_inputs
+        assert wrapper_task_run.task_inputs["__parents__"] == [
+            TaskRunResult(
+                input_type="task_run",
+                id=calling_task_state.state_details.task_run_id,
+            )
+        ]
+
     async def test_propagates_otel_trace_to_deployment_flow_run(
         self,
         test_deployment: DeploymentResponse,