diff --git a/apps/backend/components/collections/common/script_content.py b/apps/backend/components/collections/common/script_content.py
index 693125742..b94051e98 100644
--- a/apps/backend/components/collections/common/script_content.py
+++ b/apps/backend/components/collections/common/script_content.py
@@ -37,13 +37,26 @@
 """
 
 INITIALIZE_SCRIPT = """
-setup_path="$@"
-cd $setup_path
+setup_path="$2"
+subscription_tmp_dir="$1"
+
+if [ ! -d "$subscription_tmp_dir" ]; then
+  exit 0
+fi
+
+if [ ! -d "$setup_path" ]; then
+  mkdir -p "$setup_path"
+fi
+
+cd "$subscription_tmp_dir"
 for file in $(ls); do
     if echo $file | grep '.*sh$'; then
-        chmod +x $setup_path/${file}
+        chmod +x $subscription_tmp_dir/${file}
+        cp -p $subscription_tmp_dir/${file} $setup_path/${file}
     fi
 done
+
+rm -rf $subscription_tmp_dir
 """
 
 ENVIRON_SH_TEMPLATE = """
diff --git a/apps/backend/components/collections/plugin.py b/apps/backend/components/collections/plugin.py
index d74c59745..fe7fc979c 100644
--- a/apps/backend/components/collections/plugin.py
+++ b/apps/backend/components/collections/plugin.py
@@ -1557,10 +1557,15 @@ def host_id__proc_status_map(self, process_statuses: List[models.ProcessStatus])
         return host_id__proc_status_map
 
     def get_file_target_path(self, data, common_data: PluginCommonData, host: models.Host) -> str:
+        if host.os_type != constants.OsType.WINDOWS:
+            # 非 windows 下发到临时文件夹中
+            return f"/tmp/plugin_scripts_sub_{common_data.subscription.id}"
+
+        # windows 直接下发到目标路径
         host_id__proc_status_map = self.host_id__proc_status_map(common_data.process_statuses)
         process_status = host_id__proc_status_map[host.bk_host_id]
         agent_config = self.get_agent_config_by_process_status(process_status, common_data)
-        path_sep: str = (constants.LINUX_SEP, constants.WINDOWS_SEP)[host.os_type == constants.OsType.WINDOWS]
+        path_sep: str = constants.WINDOWS_SEP
         file_target_path = path_sep.join([agent_config["setup_path"], "plugins", "bin"])
         return file_target_path
 
@@ -1582,7 +1587,8 @@ def get_script_param(self, data, common_data: PluginCommonData, host: models.Hos
         process_status = host_id__proc_status_map[host.bk_host_id]
         agent_config = self.get_agent_config_by_process_status(process_status, common_data)
         file_target_path = os.path.join(agent_config["setup_path"], "plugins", "bin")
-        return file_target_path
+        tmp_file_path = f"/tmp/plugin_scripts_sub_{common_data.subscription.id}"
+        return f"{tmp_file_path} {file_target_path}"
 
     @cache.class_member_cache()
     def host_id__proc_status_map(self, process_statuses: List[models.ProcessStatus]) -> Dict[int, models.ProcessStatus]:
diff --git a/apps/backend/subscription/steps/plugin.py b/apps/backend/subscription/steps/plugin.py
index 58280f8d6..52956e2fe 100644
--- a/apps/backend/subscription/steps/plugin.py
+++ b/apps/backend/subscription/steps/plugin.py
@@ -26,7 +26,6 @@
 from apps.backend.subscription.constants import MAX_RETRY_TIME
 from apps.backend.subscription.steps import adapter
 from apps.backend.subscription.steps.base import Action, Step
-from apps.core.files.storage import CustomBKRepoStorage, get_storage
 from apps.core.tag import targets
 from apps.core.tag.models import Tag
 from apps.node_man import constants, models
@@ -957,22 +956,16 @@ def generate_activities(
             # 初始化进程状态
             activities = [plugin_manager.init_process_status(action=self.ACTION_NAME)]
 
-        storage = get_storage()
-
         # 插入生成的流程
         _activities, pipeline_data = self._generate_activities(plugin_manager)
         for _activity in _activities:
             # 排除特殊条件下，_activity为None的场景
-            if _activity:
-                if (
-                    _activity.component.code == plugin.TransferScriptComponent.code
-                    and storage.__class__ is CustomBKRepoStorage
-                ):
-                    # 只有在使用对象存储是，才下发初始化脚本的命令
-                    activities.append(_activity)
-                    activities.append(plugin_manager.init_proc_script())
-                else:
-                    activities.append(_activity)
+            if not _activity:
+                continue
+
+            activities.append(_activity)
+            if _activity.component.code == plugin.TransferScriptComponent.code:
+                activities.append(plugin_manager.init_proc_script())
 
         if self.NEED_RESET_RETRY_TIMES:
             # 插入重置重试次数及结束事件
diff --git a/apps/backend/tests/components/collections/plugin/test_init_proc_script.py b/apps/backend/tests/components/collections/plugin/test_init_proc_script.py
index a0a5f8f91..ae5a4ff4a 100644
--- a/apps/backend/tests/components/collections/plugin/test_init_proc_script.py
+++ b/apps/backend/tests/components/collections/plugin/test_init_proc_script.py
@@ -8,11 +8,17 @@
 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.
 """
+import os
 
+from mock.mock import patch
+
+from apps.backend.api.job import process_parms
+from apps.backend.components.collections.common.script_content import INITIALIZE_SCRIPT
 from apps.backend.components.collections.plugin import InitProcOperateScriptComponent
 from apps.backend.tests.components.collections.plugin.test_install_package import (
     InstallPackageTest,
 )
+from apps.backend.tests.components.collections.plugin.utils import JOB_INSTANCE_ID
 from pipeline.component_framework.test import (
     ComponentTestCase,
     ExecuteAssertion,
@@ -46,3 +52,28 @@ def cases(self):
                 execute_call_assertion=None,
             )
         ]
+
+
+class InitProcOperateInTmpDir(InitProcOperateScriptTest):
+    def test_component(self):
+        with patch(
+            "apps.backend.tests.components.collections.plugin.utils.JobMockClient.fast_execute_script"
+        ) as fast_execute_script:
+            fast_execute_script.return_value = {
+                "job_instance_name": "API Quick execution script1521100521303",
+                "job_instance_id": JOB_INSTANCE_ID,
+            }
+            super().test_component()
+
+            # 验证脚本参数
+            file_target_path = os.path.join("/usr/local/gse", "plugins", "bin")
+            self.assertEqual(
+                process_parms(f"/tmp/plugin_scripts_sub_{self.ids['subscription_id']} {file_target_path}"),
+                fast_execute_script.call_args[0][0]["script_param"],
+            )
+
+            # 验证脚本内容
+            self.assertEqual(
+                process_parms(INITIALIZE_SCRIPT),
+                fast_execute_script.call_args[0][0]["script_content"],
+            )
diff --git a/apps/backend/tests/components/collections/plugin/test_transfer_script.py b/apps/backend/tests/components/collections/plugin/test_transfer_script.py
index b1dc260b6..b4eea9191 100644
--- a/apps/backend/tests/components/collections/plugin/test_transfer_script.py
+++ b/apps/backend/tests/components/collections/plugin/test_transfer_script.py
@@ -8,10 +8,13 @@
 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 mock.mock import patch
+
 from apps.backend.components.collections.plugin import TransferScriptComponent
 from apps.backend.tests.components.collections.plugin.test_transfer_package import (
     TransferPackageTest,
 )
+from apps.backend.tests.components.collections.plugin.utils import JOB_INSTANCE_ID
 from apps.node_man import constants, models
 from pipeline.component_framework.test import (
     ComponentTestCase,
@@ -58,3 +61,43 @@ class TransferWindowsScriptTest(TransferLinuxScriptTest):
     def setUp(self):
         super().setUp()
         models.Host.objects.all().update(os_type=constants.OsType.WINDOWS)
+
+
+class TransferFileToTmpDirWithLinux(TransferLinuxScriptTest):
+    def test_component(self):
+        with patch(
+            "apps.backend.tests.components.collections.plugin.utils.JobMockClient.fast_transfer_file"
+        ) as fast_transfer_file:
+            fast_transfer_file.return_value = {
+                "job_instance_name": "API Quick Distribution File1521101427176",
+                "job_instance_id": JOB_INSTANCE_ID,
+            }
+            super().test_component()
+            self.assertEqual(
+                f"/tmp/plugin_scripts_sub_{self.ids['subscription_id']}",
+                fast_transfer_file.call_args[0][0]["file_target_path"],
+            )
+
+
+class TransferFileToTmpDirWithAix(TransferAixScriptTest, TransferFileToTmpDirWithLinux):
+    def setUp(self):
+        super(TransferAixScriptTest, self).setUp()
+
+    def test_component(self):
+        super(TransferFileToTmpDirWithLinux, self).test_component()
+
+
+class TransferFileToTmpDirWithWindows(TransferWindowsScriptTest):
+    def test_component(self):
+        with patch(
+            "apps.backend.tests.components.collections.plugin.utils.JobMockClient.fast_transfer_file"
+        ) as fast_transfer_file:
+            fast_transfer_file.return_value = {
+                "job_instance_name": "API Quick Distribution File1521101427176",
+                "job_instance_id": JOB_INSTANCE_ID,
+            }
+            super().test_component()
+            self.assertEqual(
+                "c:\\gse\\plugins\\bin",
+                fast_transfer_file.call_args[0][0]["file_target_path"],
+            )