diff --git a/client/ayon_core/pipeline/__init__.py b/client/ayon_core/pipeline/__init__.py
index 8e89029e7b..41bcd0dbd1 100644
--- a/client/ayon_core/pipeline/__init__.py
+++ b/client/ayon_core/pipeline/__init__.py
@@ -7,6 +7,10 @@ from .constants import (
 
 from .anatomy import Anatomy
 
+from .tempdir import get_temp_dir
+
+from .staging_dir import get_staging_dir_info
+
 from .create import (
     BaseCreator,
     Creator,
@@ -117,6 +121,12 @@ __all__ = (
     # --- Anatomy ---
     "Anatomy",
 
+    # --- Temp dir ---
+    "get_temp_dir",
+
+    # --- Staging dir ---
+    "get_staging_dir_info",
+
     # --- Create ---
     "BaseCreator",
     "Creator",
diff --git a/client/ayon_core/pipeline/create/creator_plugins.py b/client/ayon_core/pipeline/create/creator_plugins.py
index fe41d2fe65..6ccafe1bc7 100644
--- a/client/ayon_core/pipeline/create/creator_plugins.py
+++ b/client/ayon_core/pipeline/create/creator_plugins.py
@@ -1,4 +1,5 @@
 # -*- coding: utf-8 -*-
+import os
 import copy
 import collections
 from typing import TYPE_CHECKING, Optional, Dict, Any
@@ -6,7 +7,7 @@ from typing import TYPE_CHECKING, Optional, Dict, Any
 from abc import ABC, abstractmethod
 
 from ayon_core.settings import get_project_settings
-from ayon_core.lib import Logger
+from ayon_core.lib import Logger, get_version_from_path
 from ayon_core.pipeline.plugin_discover import (
     discover,
     register_plugin,
@@ -14,6 +15,7 @@ from ayon_core.pipeline.plugin_discover import (
     deregister_plugin,
     deregister_plugin_path
 )
+from ayon_core.pipeline import get_staging_dir_info
 
 from .constants import DEFAULT_VARIANT_VALUE
 from .product_name import get_product_name
@@ -831,6 +833,95 @@ class Creator(BaseCreator):
         """
         return self.pre_create_attr_defs
 
+    def get_staging_dir(self, instance):
+        """Return the staging dir and persistence from instance.
+
+        Args:
+            instance (CreatedInstance): Instance for which should be staging
+                dir gathered.
+
+        Returns:
+            Optional[namedtuple]: Staging dir path and persistence or None
+        """
+        create_ctx = self.create_context
+        product_name = instance.get("productName")
+        product_type = instance.get("productType")
+        folder_path = instance.get("folderPath")
+
+        # this can only work if product name and folder path are available
+        if not product_name or not folder_path:
+            return None
+
+        publish_settings = self.project_settings["core"]["publish"]
+        follow_workfile_version = (
+            publish_settings
+            ["CollectAnatomyInstanceData"]
+            ["follow_workfile_version"]
+        )
+
+        # Gather version number provided from the instance.
+        version = instance.get("version")
+
+        # If follow workfile, gather version from workfile path.
+        if version is None and follow_workfile_version:
+            current_workfile = self.create_context.get_current_workfile_path()
+            workfile_version = get_version_from_path(current_workfile)
+            version = int(workfile_version)
+
+        # Fill-up version with next version available.
+        elif version is None:
+            versions = self.get_next_versions_for_instances(
+                [instance]
+            )
+            version, = tuple(versions.values())
+
+        template_data = {"version": version}
+
+        staging_dir_info = get_staging_dir_info(
+            create_ctx.get_current_project_entity(),
+            create_ctx.get_folder_entity(folder_path),
+            create_ctx.get_task_entity(folder_path, instance.get("task")),
+            product_type,
+            product_name,
+            create_ctx.host_name,
+            anatomy=create_ctx.get_current_project_anatomy(),
+            project_settings=create_ctx.get_current_project_settings(),
+            always_return_path=False,
+            logger=self.log,
+            template_data=template_data,
+        )
+
+        return staging_dir_info or None
+
+    def apply_staging_dir(self, instance):
+        """Apply staging dir with persistence to instance's transient data.
+
+        Method is called on instance creation and on instance update.
+
+        Args:
+            instance (CreatedInstance): Instance for which should be staging
+                dir applied.
+
+        Returns:
+            Optional[str]: Staging dir path or None if not applied.
+        """
+        staging_dir_info = self.get_staging_dir(instance)
+        if staging_dir_info is None:
+            return None
+
+        # path might be already created by get_staging_dir_info
+        staging_dir_path = staging_dir_info.directory
+        os.makedirs(staging_dir_path, exist_ok=True)
+
+        instance.transient_data.update({
+            "stagingDir": staging_dir_path,
+            "stagingDir_persistent": staging_dir_info.persistent,
+        })
+
+        self.log.info(f"Applied staging dir to instance: {staging_dir_path}")
+
+        return staging_dir_path
+
     def _pre_create_attr_defs_changed(self):
         """Called when pre-create attribute definitions change.
 
diff --git a/client/ayon_core/pipeline/publish/constants.py b/client/ayon_core/pipeline/publish/constants.py
index f2f4e851a9..a33e8f9eed 100644
--- a/client/ayon_core/pipeline/publish/constants.py
+++ b/client/ayon_core/pipeline/publish/constants.py
@@ -8,6 +8,5 @@ ValidateMeshOrder = pyblish.api.ValidatorOrder + 0.3
 
 DEFAULT_PUBLISH_TEMPLATE = "default"
 DEFAULT_HERO_PUBLISH_TEMPLATE = "default"
-TRANSIENT_DIR_TEMPLATE = "default"
 
 FARM_JOB_ENV_DATA_KEY: str = "farmJobEnv"
diff --git a/client/ayon_core/pipeline/publish/lib.py b/client/ayon_core/pipeline/publish/lib.py
index dc2eef3bb9..2ba40d7687 100644
--- a/client/ayon_core/pipeline/publish/lib.py
+++ b/client/ayon_core/pipeline/publish/lib.py
@@ -2,7 +2,7 @@ import os
 import sys
 import inspect
 import copy
-import tempfile
+import warnings
 import xml.etree.ElementTree
 from typing import Optional, Union, List
 
@@ -18,15 +18,11 @@ from ayon_core.lib import (
 )
 from ayon_core.settings import get_project_settings
 from ayon_core.addon import AddonsManager
-from ayon_core.pipeline import (
-    tempdir,
-    Anatomy
-)
+from ayon_core.pipeline import get_staging_dir_info
 from ayon_core.pipeline.plugin_discover import DiscoverResult
 from .constants import (
     DEFAULT_PUBLISH_TEMPLATE,
     DEFAULT_HERO_PUBLISH_TEMPLATE,
-    TRANSIENT_DIR_TEMPLATE
 )
 
 
@@ -581,58 +577,6 @@ def context_plugin_should_run(plugin, context):
     return False
 
 
-def get_instance_staging_dir(instance):
-    """Unified way how staging dir is stored and created on instances.
-
-    First check if 'stagingDir' is already set in instance data.
-    In case there already is new tempdir will not be created.
-
-    It also supports `AYON_TMPDIR`, so studio can define own temp
-    shared repository per project or even per more granular context.
-    Template formatting is supported also with optional keys. Folder is
-    created in case it doesn't exists.
-
-    Available anatomy formatting keys:
-        - root[work | <root name key>]
-        - project[name | code]
-
-    Note:
-        Staging dir does not have to be necessarily in tempdir so be careful
-        about its usage.
-
-    Args:
-        instance (pyblish.lib.Instance): Instance for which we want to get
-            staging dir.
-
-    Returns:
-        str: Path to staging dir of instance.
-    """
-    staging_dir = instance.data.get('stagingDir')
-    if staging_dir:
-        return staging_dir
-
-    anatomy = instance.context.data.get("anatomy")
-
-    # get customized tempdir path from `AYON_TMPDIR` env var
-    custom_temp_dir = tempdir.create_custom_tempdir(
-        anatomy.project_name, anatomy)
-
-    if custom_temp_dir:
-        staging_dir = os.path.normpath(
-            tempfile.mkdtemp(
-                prefix="pyblish_tmp_",
-                dir=custom_temp_dir
-            )
-        )
-    else:
-        staging_dir = os.path.normpath(
-            tempfile.mkdtemp(prefix="pyblish_tmp_")
-        )
-    instance.data['stagingDir'] = staging_dir
-
-    return staging_dir
-
-
 def get_publish_repre_path(instance, repre, only_published=False):
     """Get representation path that can be used for integration.
 
@@ -685,6 +629,8 @@ def get_publish_repre_path(instance, repre, only_published=False):
     return None
 
 
+# deprecated: backward compatibility only (2024-09-12)
+# TODO: remove in the future
 def get_custom_staging_dir_info(
     project_name,
     host_name,
@@ -694,67 +640,86 @@ def get_custom_staging_dir_info(
     product_name,
     project_settings=None,
     anatomy=None,
-    log=None
+    log=None,
 ):
-    """Checks profiles if context should use special custom dir as staging.
+    from ayon_core.pipeline.staging_dir import get_staging_dir_config
+    warnings.warn(
+        (
+            "Function 'get_custom_staging_dir_info' in"
+            " 'ayon_core.pipeline.publish' is deprecated. Please use"
+            " 'get_custom_staging_dir_info'"
+            " in 'ayon_core.pipeline.stagingdir'."
+        ),
+        DeprecationWarning,
+    )
+    tr_data = get_staging_dir_config(
+        project_name,
+        task_type,
+        task_name,
+        product_type,
+        product_name,
+        host_name,
+        project_settings=project_settings,
+        anatomy=anatomy,
+        log=log,
+    )
 
-    Args:
-        project_name (str)
-        host_name (str)
-        product_type (str)
-        task_name (str)
-        task_type (str)
-        product_name (str)
-        project_settings(Dict[str, Any]): Prepared project settings.
-        anatomy (Dict[str, Any])
-        log (Logger) (optional)
+    if not tr_data:
+        return None, None
+
+    return tr_data["template"], tr_data["persistence"]
+
+
+def get_instance_staging_dir(instance):
+    """Unified way how staging dir is stored and created on instances.
+
+    First check if 'stagingDir' is already set in instance data.
+    In case there already is new tempdir will not be created.
 
     Returns:
-        (tuple)
-    Raises:
-        ValueError - if misconfigured template should be used
+        str: Path to staging dir
     """
-    settings = project_settings or get_project_settings(project_name)
-    custom_staging_dir_profiles = (settings["core"]
-                                           ["tools"]
-                                           ["publish"]
-                                           ["custom_staging_dir_profiles"])
-    if not custom_staging_dir_profiles:
-        return None, None
+    staging_dir = instance.data.get("stagingDir")
 
-    if not log:
-        log = Logger.get_logger("get_custom_staging_dir_info")
+    if staging_dir:
+        return staging_dir
 
-    filtering_criteria = {
-        "hosts": host_name,
-        "families": product_type,
-        "task_names": task_name,
-        "task_types": task_type,
-        "subsets": product_name
-    }
-    profile = filter_profiles(custom_staging_dir_profiles,
-                              filtering_criteria,
-                              logger=log)
+    anatomy_data = instance.data["anatomyData"]
+    template_data = copy.deepcopy(anatomy_data)
 
-    if not profile or not profile["active"]:
-        return None, None
+    # context data based variables
+    context = instance.context
 
-    if not anatomy:
-        anatomy = Anatomy(project_name)
+    # add current file as workfile name into formatting data
+    current_file = context.data.get("currentFile")
+    if current_file:
+        workfile = os.path.basename(current_file)
+        workfile_name, _ = os.path.splitext(workfile)
+        template_data["workfile_name"] = workfile_name
 
-    template_name = profile["template_name"] or TRANSIENT_DIR_TEMPLATE
-
-    custom_staging_dir = anatomy.get_template_item(
-        "staging", template_name, "directory", default=None
+    staging_dir_info = get_staging_dir_info(
+        context.data["projectEntity"],
+        instance.data.get("folderEntity"),
+        instance.data.get("taskEntity"),
+        instance.data["productType"],
+        instance.data["productName"],
+        context.data["hostName"],
+        anatomy=context.data["anatomy"],
+        project_settings=context.data["project_settings"],
+        template_data=template_data,
+        always_return_path=True,
     )
-    if custom_staging_dir is None:
-        raise ValueError((
-            "Anatomy of project \"{}\" does not have set"
-            " \"{}\" template key!"
-        ).format(project_name, template_name))
-    is_persistent = profile["custom_staging_dir_persistent"]
 
-    return custom_staging_dir.template, is_persistent
+    staging_dir_path = staging_dir_info.directory
+
+    # path might be already created by get_staging_dir_info
+    os.makedirs(staging_dir_path, exist_ok=True)
+    instance.data.update({
+        "stagingDir": staging_dir_path,
+        "stagingDir_persistent": staging_dir_info.persistent,
+    })
+
+    return staging_dir_path
 
 
 def get_published_workfile_instance(context):
diff --git a/client/ayon_core/pipeline/staging_dir.py b/client/ayon_core/pipeline/staging_dir.py
new file mode 100644
index 0000000000..ea22d99389
--- /dev/null
+++ b/client/ayon_core/pipeline/staging_dir.py
@@ -0,0 +1,222 @@
+from dataclasses import dataclass
+
+from ayon_core.lib import Logger, filter_profiles
+from ayon_core.settings import get_project_settings
+
+from .template_data import get_template_data
+from .anatomy import Anatomy
+from .tempdir import get_temp_dir
+
+
+@dataclass
+class StagingDir:
+    directory: str
+    persistent: bool
+
+
+def get_staging_dir_config(
+    project_name,
+    task_type,
+    task_name,
+    product_type,
+    product_name,
+    host_name,
+    project_settings=None,
+    anatomy=None,
+    log=None,
+):
+    """Get matching staging dir profile.
+
+    Args:
+        host_name (str): Name of host.
+        project_name (str): Name of project.
+        task_type (Optional[str]): Type of task.
+        task_name (Optional[str]): Name of task.
+        product_type (str): Type of product.
+        product_name (str): Name of product.
+        project_settings(Dict[str, Any]): Prepared project settings.
+        anatomy (Dict[str, Any])
+        log (Optional[logging.Logger])
+
+    Returns:
+        Dict or None: Data with directory template and is_persistent or None
+
+    Raises:
+        KeyError - if misconfigured template should be used
+
+    """
+    settings = project_settings or get_project_settings(project_name)
+
+    staging_dir_profiles = settings["core"]["tools"]["publish"][
+        "custom_staging_dir_profiles"
+    ]
+
+    if not staging_dir_profiles:
+        return None
+
+    if not log:
+        log = Logger.get_logger("get_staging_dir_config")
+
+    filtering_criteria = {
+        "hosts": host_name,
+        "task_types": task_type,
+        "task_names": task_name,
+        "product_types": product_type,
+        "product_names": product_name,
+    }
+    profile = filter_profiles(
+        staging_dir_profiles, filtering_criteria, logger=log)
+
+    if not profile or not profile["active"]:
+        return None
+
+    if not anatomy:
+        anatomy = Anatomy(project_name)
+
+    # get template from template name
+    template_name = profile["template_name"]
+    _validate_template_name(project_name, template_name, anatomy)
+
+    template = anatomy.get_template_item("staging", template_name)
+
+    if not template:
+        # template should always be found either from anatomy or from profile
+        raise KeyError(
+            f"Staging template '{template_name}' was not found."
+            "Check project anatomy or settings at: "
+            "'ayon+settings://core/tools/publish/custom_staging_dir_profiles'"
+        )
+
+    data_persistence = profile["custom_staging_dir_persistent"]
+
+    return {"template": template, "persistence": data_persistence}
+
+
+def _validate_template_name(project_name, template_name, anatomy):
+    """Check that staging dir section with appropriate template exist.
+
+    Raises:
+        ValueError - if misconfigured template
+    """
+    if template_name not in anatomy.templates["staging"]:
+        raise ValueError(
+            f'Anatomy of project "{project_name}" does not have set'
+            f' "{template_name}" template key at Staging Dir category!'
+        )
+
+
+def get_staging_dir_info(
+    project_entity,
+    folder_entity,
+    task_entity,
+    product_type,
+    product_name,
+    host_name,
+    anatomy=None,
+    project_settings=None,
+    template_data=None,
+    always_return_path=True,
+    force_tmp_dir=False,
+    logger=None,
+    prefix=None,
+    suffix=None,
+):
+    """Get staging dir info data.
+
+    If `force_temp` is set, staging dir will be created as tempdir.
+    If `always_get_some_dir` is set, staging dir will be created as tempdir if
+    no staging dir profile is found.
+    If `prefix` or `suffix` is not set, default values will be used.
+
+    Arguments:
+        project_entity (Dict[str, Any]): Project entity.
+        folder_entity (Optional[Dict[str, Any]]): Folder entity.
+        task_entity (Optional[Dict[str, Any]]): Task entity.
+        product_type (str): Type of product.
+        product_name (str): Name of product.
+        host_name (str): Name of host.
+        anatomy (Optional[Anatomy]): Anatomy object.
+        project_settings (Optional[Dict[str, Any]]): Prepared project settings.
+        template_data (Optional[Dict[str, Any]]): Additional data for
+            formatting staging dir template.
+        always_return_path (Optional[bool]): If True, staging dir will be
+            created as tempdir if no staging dir profile is found. Input value
+            False will return None if no staging dir profile is found.
+        force_tmp_dir (Optional[bool]): If True, staging dir will be created as
+            tempdir.
+        logger (Optional[logging.Logger]): Logger instance.
+        prefix (Optional[str]) Optional prefix for staging dir name.
+        suffix (Optional[str]): Optional suffix for staging dir name.
+
+    Returns:
+        Optional[StagingDir]: Staging dir info data
+
+    """
+    log = logger or Logger.get_logger("get_staging_dir_info")
+
+    if anatomy is None:
+        anatomy = Anatomy(
+            project_entity["name"], project_entity=project_entity
+        )
+
+    if force_tmp_dir:
+        return get_temp_dir(
+            project_name=project_entity["name"],
+            anatomy=anatomy,
+            prefix=prefix,
+            suffix=suffix,
+        )
+
+    # making few queries to database
+    ctx_data = get_template_data(
+        project_entity, folder_entity, task_entity, host_name
+    )
+
+    # add additional data
+    ctx_data["product"] = {
+        "type": product_type,
+        "name": product_name
+    }
+
+    # add additional template formatting data
+    if template_data:
+        ctx_data.update(template_data)
+
+    task_name = task_type = None
+    if task_entity:
+        task_name = task_entity["name"]
+        task_type = task_entity["taskType"]
+
+    # get staging dir config
+    staging_dir_config = get_staging_dir_config(
+        project_entity["name"],
+        task_type,
+        task_name ,
+        product_type,
+        product_name,
+        host_name,
+        project_settings=project_settings,
+        anatomy=anatomy,
+        log=log,
+    )
+
+    if staging_dir_config:
+        dir_template = staging_dir_config["template"]["directory"]
+        return StagingDir(
+            dir_template.format_strict(ctx_data),
+            staging_dir_config["persistence"],
+        )
+
+    # no config found but force an output
+    if always_return_path:
+        return StagingDir(
+            get_temp_dir(
+                project_name=project_entity["name"],
+                anatomy=anatomy,
+                prefix=prefix,
+                suffix=suffix,
+            ),
+            False,
+        )
+
+    return None
diff --git a/client/ayon_core/pipeline/tempdir.py b/client/ayon_core/pipeline/tempdir.py
index d8f42ea60a..fe057b7fc7 100644
--- a/client/ayon_core/pipeline/tempdir.py
+++ b/client/ayon_core/pipeline/tempdir.py
@@ -3,11 +3,74 @@ Temporary folder operations
 """
 
 import os
+import tempfile
+from pathlib import Path
+
 from ayon_core.lib import StringTemplate
 from ayon_core.pipeline import Anatomy
 
 
-def create_custom_tempdir(project_name, anatomy=None):
+def get_temp_dir(
+    project_name, anatomy=None, prefix=None, suffix=None, use_local_temp=False
+):
+    """Get temporary dir path.
+
+    If `use_local_temp` is set, tempdir will be created in local tempdir.
+    If `anatomy` is not set, default anatomy will be used.
+    If `prefix` or `suffix` is not set, default values will be used.
+
+    It also supports `AYON_TMPDIR`, so studio can define own temp
+    shared repository per project or even per more granular context.
+    Template formatting is supported also with optional keys. Folder is
+    created in case it doesn't exists.
+
+    Args:
+        project_name (str): Name of project.
+        anatomy (Optional[Anatomy]): Project Anatomy object.
+        suffix (Optional[str]): Suffix for tempdir.
+        prefix (Optional[str]): Prefix for tempdir.
+        use_local_temp (Optional[bool]): If True, temp dir will be created in
+            local tempdir.
+
+    Returns:
+        str: Path to staging dir of instance.
+
+    """
+    if prefix is None:
+        prefix = "ay_tmp_"
+    suffix = suffix or ""
+
+    if use_local_temp:
+        return _create_local_staging_dir(prefix, suffix)
+
+    # make sure anatomy is set
+    if not anatomy:
+        anatomy = Anatomy(project_name)
+
+    # get customized tempdir path from `OPENPYPE_TMPDIR` env var
+    custom_temp_dir = _create_custom_tempdir(anatomy.project_name, anatomy)
+
+    return _create_local_staging_dir(prefix, suffix, dirpath=custom_temp_dir)
+
+
+def _create_local_staging_dir(prefix, suffix, dirpath=None):
+    """Create local staging dir
+
+    Args:
+        prefix (str): prefix for tempdir
+        suffix (str): suffix for tempdir
+        dirpath (Optional[str]): path to tempdir
+
+    Returns:
+        str: path to tempdir
+    """
+    # use pathlib for creating tempdir
+    return tempfile.mkdtemp(
+        prefix=prefix, suffix=suffix, dir=dirpath
+    )
+
+
+def _create_custom_tempdir(project_name, anatomy):
     """ Create custom tempdir
 
     Template path formatting is supporting:
@@ -18,42 +81,35 @@ def create_custom_tempdir(project_name, anatomy=None):
 
     Args:
         project_name (str): project name
-        anatomy (ayon_core.pipeline.Anatomy)[optional]: Anatomy object
+        anatomy (ayon_core.pipeline.Anatomy): Anatomy object
 
     Returns:
         str | None: formatted path or None
     """
     env_tmpdir = os.getenv("AYON_TMPDIR")
     if not env_tmpdir:
-        return
+        return None
 
     custom_tempdir = None
     if "{" in env_tmpdir:
-        if anatomy is None:
-            anatomy = Anatomy(project_name)
         # create base formate data
-        data = {
+        template_data = {
             "root": anatomy.roots,
             "project": {
                 "name": anatomy.project_name,
                 "code": anatomy.project_code,
-            }
+            },
         }
         # path is anatomy template
         custom_tempdir = StringTemplate.format_template(
-            env_tmpdir, data).normalized()
+            env_tmpdir, template_data)
+
+        custom_tempdir_path = Path(custom_tempdir)
 
     else:
         # path is absolute
-        custom_tempdir = env_tmpdir
+        custom_tempdir_path = Path(env_tmpdir)
 
-    # create the dir path if it doesn't exists
-    if not os.path.exists(custom_tempdir):
-        try:
-            # create it if it doesn't exists
-            os.makedirs(custom_tempdir)
-        except IOError as error:
-            raise IOError(
-                "Path couldn't be created: {}".format(error))
+    custom_tempdir_path.mkdir(parents=True, exist_ok=True)
 
-    return custom_tempdir
+    return custom_tempdir_path.as_posix()
diff --git a/client/ayon_core/plugins/publish/collect_custom_staging_dir.py b/client/ayon_core/plugins/publish/collect_custom_staging_dir.py
deleted file mode 100644
index 49c3a98dd2..0000000000
--- a/client/ayon_core/plugins/publish/collect_custom_staging_dir.py
+++ /dev/null
@@ -1,76 +0,0 @@
-"""
-Requires:
-    anatomy
-
-
-Provides:
-    instance.data     -> stagingDir (folder path)
-                      -> stagingDir_persistent (bool)
-"""
-import copy
-import os.path
-
-import pyblish.api
-
-from ayon_core.pipeline.publish.lib import get_custom_staging_dir_info
-
-
-class CollectCustomStagingDir(pyblish.api.InstancePlugin):
-    """Looks through profiles if stagingDir should be persistent and in special
-    location.
-
-    Transient staging dir could be useful in specific use cases where is
-    desirable to have temporary renders in specific, persistent folders, could
-    be on disks optimized for speed for example.
-
-    It is studio responsibility to clean up obsolete folders with data.
-
-    Location of the folder is configured in `project_anatomy/templates/others`.
-    ('transient' key is expected, with 'folder' key)
-
-    Which family/task type/product is applicable is configured in:
-    `project_settings/global/tools/publish/custom_staging_dir_profiles`
-
-    """
-    label = "Collect Custom Staging Directory"
-    order = pyblish.api.CollectorOrder + 0.4990
-
-    template_key = "transient"
-
-    def process(self, instance):
-        product_type = instance.data["productType"]
-        product_name = instance.data["productName"]
-        host_name = instance.context.data["hostName"]
-        project_name = instance.context.data["projectName"]
-        project_settings = instance.context.data["project_settings"]
-        anatomy = instance.context.data["anatomy"]
-        task = instance.data["anatomyData"].get("task", {})
-
-        transient_tml, is_persistent = get_custom_staging_dir_info(
-            project_name,
-            host_name,
-            product_type,
-            product_name,
-            task.get("name"),
-            task.get("type"),
-            project_settings=project_settings,
-            anatomy=anatomy,
-            log=self.log)
-
-        if transient_tml:
-            anatomy_data = copy.deepcopy(instance.data["anatomyData"])
-            anatomy_data["root"] = anatomy.roots
-            scene_name = instance.context.data.get("currentFile")
-            if scene_name:
-                anatomy_data["scene_name"] = os.path.basename(scene_name)
-            transient_dir = transient_tml.format(**anatomy_data)
-            instance.data["stagingDir"] = transient_dir
-
-            instance.data["stagingDir_persistent"] = is_persistent
-            result_str = "Adding '{}' as".format(transient_dir)
-        else:
-            result_str = "Not adding"
-
-        self.log.debug("{} custom staging dir for instance with '{}'".format(
-            result_str, product_type
-        ))
diff --git a/client/ayon_core/plugins/publish/collect_managed_staging_dir.py b/client/ayon_core/plugins/publish/collect_managed_staging_dir.py
new file mode 100644
index 0000000000..1034b9a716
--- /dev/null
+++ b/client/ayon_core/plugins/publish/collect_managed_staging_dir.py
@@ -0,0 +1,47 @@
+"""
+Requires:
+    anatomy
+
+
+Provides:
+    instance.data     -> stagingDir (folder path)
+                      -> stagingDir_persistent (bool)
+"""
+
+import pyblish.api
+
+from ayon_core.pipeline.publish import get_instance_staging_dir
+
+
+class CollectManagedStagingDir(pyblish.api.InstancePlugin):
+    """Apply matching Staging Dir profile to a instance.
+
+    Apply Staging dir via profiles could be useful in specific use cases
+    where is desirable to have temporary renders in specific,
+    persistent folders, could be on disks optimized for speed for example.
+
+    It is studio's responsibility to clean up obsolete folders with data.
+
+    Location of the folder is configured in:
+        `ayon+anatomy://_/templates/staging`.
+
+    Which family/task type/subset is applicable is configured in:
+        `ayon+settings://core/tools/publish/custom_staging_dir_profiles`
+    """
+
+    label = "Collect Managed Staging Directory"
+    order = pyblish.api.CollectorOrder + 0.4990
+
+    def process(self, instance):
+        """ Collect the staging data and stores it to the instance.
+
+        Args:
+            instance (object): The instance to inspect.
+        """
+        staging_dir_path = get_instance_staging_dir(instance)
+        persistance = instance.data.get("stagingDir_persistent", False)
+
+        self.log.info((
+            f"Instance staging dir was set to `{staging_dir_path}` "
+            f"and persistence is set to `{persistance}`"
+        ))
diff --git a/client/ayon_core/plugins/publish/extract_burnin.py b/client/ayon_core/plugins/publish/extract_burnin.py
index 2007240d3d..8e8764fc33 100644
--- a/client/ayon_core/plugins/publish/extract_burnin.py
+++ b/client/ayon_core/plugins/publish/extract_burnin.py
@@ -9,11 +9,13 @@ import clique
 import pyblish.api
 
 from ayon_core import resources, AYON_CORE_ROOT
-from ayon_core.pipeline import publish
+from ayon_core.pipeline import (
+    publish,
+    get_temp_dir
+)
 from ayon_core.lib import (
     run_ayon_launcher_process,
 
-    get_transcode_temp_directory,
     convert_input_paths_for_ffmpeg,
     should_convert_for_ffmpeg
 )
@@ -250,7 +252,10 @@ class ExtractBurnin(publish.Extractor):
             #   - change staging dir of source representation
             #   - must be set back after output definitions processing
             if do_convert:
-                new_staging_dir = get_transcode_temp_directory()
+                new_staging_dir = get_temp_dir(
+                    project_name=instance.context.data["projectName"],
+                    use_local_temp=True,
+                )
                 repre["stagingDir"] = new_staging_dir
 
                 convert_input_paths_for_ffmpeg(
diff --git a/client/ayon_core/plugins/publish/extract_color_transcode.py b/client/ayon_core/plugins/publish/extract_color_transcode.py
index 56d5d33ea4..3c11a016ec 100644
--- a/client/ayon_core/plugins/publish/extract_color_transcode.py
+++ b/client/ayon_core/plugins/publish/extract_color_transcode.py
@@ -3,14 +3,15 @@ import copy
 import clique
 import pyblish.api
 
-from ayon_core.pipeline import publish
+from ayon_core.pipeline import (
+    publish,
+    get_temp_dir
+)
 from ayon_core.lib import (
     is_oiio_supported,
 )
-
 from ayon_core.lib.transcoding import (
     convert_colorspace,
-    get_transcode_temp_directory,
 )
 
 from ayon_core.lib.profiles_filtering import filter_profiles
@@ -103,7 +104,10 @@ class ExtractOIIOTranscode(publish.Extractor):
                 new_repre = copy.deepcopy(repre)
 
                 original_staging_dir = new_repre["stagingDir"]
-                new_staging_dir = get_transcode_temp_directory()
+                new_staging_dir = get_temp_dir(
+                    project_name=instance.context.data["projectName"],
+                    use_local_temp=True,
+                )
                 new_repre["stagingDir"] = new_staging_dir
 
                 if isinstance(new_repre["files"], list):
@@ -265,7 +269,7 @@ class ExtractOIIOTranscode(publish.Extractor):
             (list) of [file.1001-1010#.exr] or [fileA.exr, fileB.exr]
         """
         pattern = [clique.PATTERNS["frames"]]
-        collections, remainder = clique.assemble(
+        collections, _ = clique.assemble(
             files_to_convert, patterns=pattern,
             assume_padded_when_ambiguous=True)
 
diff --git a/client/ayon_core/plugins/publish/extract_review.py b/client/ayon_core/plugins/publish/extract_review.py
index 06b451bfbe..7c38b0453b 100644
--- a/client/ayon_core/plugins/publish/extract_review.py
+++ b/client/ayon_core/plugins/publish/extract_review.py
@@ -22,8 +22,8 @@ from ayon_core.lib.transcoding import (
     should_convert_for_ffmpeg,
     get_review_layer_name,
     convert_input_paths_for_ffmpeg,
-    get_transcode_temp_directory,
 )
+from ayon_core.pipeline import get_temp_dir
 from ayon_core.pipeline.publish import (
     KnownPublishError,
     get_publish_instance_label,
@@ -310,7 +310,10 @@ class ExtractReview(pyblish.api.InstancePlugin):
             #   - change staging dir of source representation
             #   - must be set back after output definitions processing
             if do_convert:
-                new_staging_dir = get_transcode_temp_directory()
+                new_staging_dir = get_temp_dir(
+                    project_name=instance.context.data["projectName"],
+                    use_local_temp=True,
+                )
                 repre["stagingDir"] = new_staging_dir
 
                 convert_input_paths_for_ffmpeg(