diff --git a/src/core/src/CoreMain.py b/src/core/src/CoreMain.py
index 6b78f608..619141da 100644
--- a/src/core/src/CoreMain.py
+++ b/src/core/src/CoreMain.py
@@ -13,6 +13,7 @@
 # limitations under the License.
 #
 # Requires Python 2.7+
+import os
 
 from core.src.bootstrap.Bootstrapper import Bootstrapper
 from core.src.bootstrap.Constants import Constants
@@ -27,7 +28,7 @@ def __init__(self, argv):
         composite_logger = bootstrapper.composite_logger
         stdout_file_mirror = bootstrapper.stdout_file_mirror
         telemetry_writer = bootstrapper.telemetry_writer
-        lifecycle_manager = status_handler = None
+        lifecycle_manager = status_handler = execution_config = None
 
         # Init operation statuses
         patch_operation_requested = Constants.UNKNOWN
@@ -59,6 +60,12 @@ def __init__(self, argv):
             telemetry_writer.set_task_name(Constants.TelemetryTaskName.AUTO_ASSESSMENT if execution_config.exec_auto_assess_only else Constants.TelemetryTaskName.EXEC)
             patch_operation_requested = execution_config.operation.lower()
 
+            # clean up temp folder before any operation execution begins from Core
+            if os.path.exists(execution_config.temp_folder):
+                composite_logger.log_debug("Deleting all files of certain format from temp folder [FileFormat={0}][TempFolderLocation={1}]"
+                                           .format(Constants.TEMP_FOLDER_CLEANUP_ARTIFACT_LIST, str(execution_config.temp_folder)))
+                bootstrapper.env_layer.file_system.delete_files_from_dir(execution_config.temp_folder, Constants.TEMP_FOLDER_CLEANUP_ARTIFACT_LIST)
+
             patch_assessor = container.get('patch_assessor')
             package_manager = container.get('package_manager')
             configure_patching_processor = container.get('configure_patching_processor')
@@ -115,6 +122,12 @@ def __init__(self, argv):
             composite_logger.log_debug("Completed exception handling.\n")
 
         finally:
+            # clean up temp folder of files created by Core after execution completes
+            if self.is_temp_folder_available(bootstrapper.env_layer, execution_config):
+                composite_logger.log_debug("Deleting all files of certain format from temp folder [FileFormat={0}][TempFolderLocation={1}]"
+                                           .format(Constants.TEMP_FOLDER_CLEANUP_ARTIFACT_LIST, str(execution_config.temp_folder)))
+                bootstrapper.env_layer.file_system.delete_files_from_dir(execution_config.temp_folder, Constants.TEMP_FOLDER_CLEANUP_ARTIFACT_LIST)
+
             if lifecycle_manager is not None:
                 lifecycle_manager.update_core_sequence(completed=True)
 
@@ -139,3 +152,10 @@ def update_patch_substatus_if_pending(patch_operation_requested, overall_patch_i
             status_handler.set_configure_patching_substatus_json(status=Constants.STATUS_ERROR)
             composite_logger.log_debug('  -- Persisted failed configure patching substatus.')
 
+    @staticmethod
+    def is_temp_folder_available(env_layer, execution_config):
+        return env_layer is not None \
+               and execution_config is not None \
+               and execution_config.temp_folder is not None \
+               and os.path.exists(execution_config.temp_folder)
+
diff --git a/src/core/src/bootstrap/Constants.py b/src/core/src/bootstrap/Constants.py
index e941b730..f60c8a44 100644
--- a/src/core/src/bootstrap/Constants.py
+++ b/src/core/src/bootstrap/Constants.py
@@ -59,6 +59,7 @@ class EnvSettings(EnumBackport):
         CONFIG_FOLDER = "configFolder"
         STATUS_FOLDER = "statusFolder"
         EVENTS_FOLDER = "eventsFolder"
+        TEMP_FOLDER = "tempFolder"
         TELEMETRY_SUPPORTED = "telemetrySupported"
 
     class ConfigSettings(EnumBackport):
@@ -76,6 +77,9 @@ class ConfigSettings(EnumBackport):
         ASSESSMENT_MODE = 'assessmentMode'
         MAXIMUM_ASSESSMENT_INTERVAL = 'maximumAssessmentInterval'
 
+    TEMP_FOLDER_DIR_NAME = "tmp"
+    TEMP_FOLDER_CLEANUP_ARTIFACT_LIST = ["*.list"]
+
     # File to save default settings for auto OS updates
     IMAGE_DEFAULT_PATCH_CONFIGURATION_BACKUP_PATH = "ImageDefaultPatchConfiguration.bak"
 
diff --git a/src/core/src/bootstrap/EnvLayer.py b/src/core/src/bootstrap/EnvLayer.py
index c7535c36..7ef73cb8 100644
--- a/src/core/src/bootstrap/EnvLayer.py
+++ b/src/core/src/bootstrap/EnvLayer.py
@@ -17,6 +17,7 @@
 from __future__ import print_function
 import base64
 import datetime
+import glob
 import json
 import os
 import re
@@ -428,6 +429,27 @@ def write_with_retry_using_temp_file(file_path, data, mode='w'):
                     else:
                         raise Exception("Unable to write to {0} (retries exhausted). Error: {1}.".format(str(file_path), repr(error)))
 
+        @staticmethod
+        def delete_files_from_dir(dir_name, file_identifier_list, raise_if_delete_failed=False):
+            """ Clears all files from given dir. NOTE: Uses file_identifier_list to determine the content to delete """
+            for file_identifier in file_identifier_list:
+                files_to_delete = glob.glob(str(dir_name) + "/" + str(file_identifier))
+
+                for file_to_delete in files_to_delete:
+                    try:
+                        os.remove(file_to_delete)
+                    except Exception as error:
+                        error_message = "Unable to delete files from directory [Dir={0}][File={1}][Error={2}][RaiseIfDeleteFailed={3}].".format(
+                            str(dir_name),
+                            str(file_to_delete),
+                            repr(error),
+                            str(raise_if_delete_failed))
+
+                        if raise_if_delete_failed:
+                            raise Exception(error_message)
+                        else:
+                            print(error_message)
+                            return None
 # endregion - File system emulation and extensions
 
 # region - DateTime emulation and extensions
diff --git a/src/core/src/core_logic/ExecutionConfig.py b/src/core/src/core_logic/ExecutionConfig.py
index fcca7b4c..ce34954b 100644
--- a/src/core/src/core_logic/ExecutionConfig.py
+++ b/src/core/src/core_logic/ExecutionConfig.py
@@ -44,6 +44,9 @@ def __init__(self, env_layer, composite_logger, execution_parameters):
         self.config_folder = self.environment_settings[Constants.EnvSettings.CONFIG_FOLDER]
         self.status_folder = self.environment_settings[Constants.EnvSettings.STATUS_FOLDER]
         self.events_folder = self.environment_settings[Constants.EnvSettings.EVENTS_FOLDER]
+        self.temp_folder = self.environment_settings[Constants.EnvSettings.TEMP_FOLDER]
+        self.__check_and_create_temp_folder_if_not_exists()
+
         self.telemetry_supported = self.environment_settings[Constants.EnvSettings.TELEMETRY_SUPPORTED]
 
         # Config Settings
@@ -162,3 +165,16 @@ def __extract_most_significant_unit_from_duration(duration_portion, unit_delimit
         else:  # bad data
             raise Exception("Invalid duration portion: {0}".format(str(duration_portion)))
         return most_significant_unit, remaining_duration_portion
+
+    def __check_and_create_temp_folder_if_not_exists(self):
+        """Verifies temp folder exists, creates new one if not found"""
+        if self.temp_folder is None:
+            par_dir = os.path.dirname(self.config_folder)
+            if not os.path.exists(par_dir):
+                raise Exception("Parent directory for all extension artifacts such as config folder, status folder, etc. not found at [{0}].".format(repr(par_dir)))
+            self.temp_folder = os.path.join(par_dir, Constants.TEMP_FOLDER_DIR_NAME)
+
+        if not os.path.exists(self.temp_folder):
+            self.composite_logger.log_debug("Temp folder does not exist, creating one from extension core. [Path={0}]".format(str(self.temp_folder)))
+            os.mkdir(self.temp_folder)
+
diff --git a/src/core/src/package_managers/AptitudePackageManager.py b/src/core/src/package_managers/AptitudePackageManager.py
index 21e4a596..4aa02e24 100644
--- a/src/core/src/package_managers/AptitudePackageManager.py
+++ b/src/core/src/package_managers/AptitudePackageManager.py
@@ -18,6 +18,8 @@
 import json
 import os
 import re
+import uuid
+
 from core.src.package_managers.PackageManager import PackageManager
 from core.src.bootstrap.Constants import Constants
 
@@ -28,12 +30,14 @@ class AptitudePackageManager(PackageManager):
     # For more details, try `man apt-get` on any Debian/Ubuntu based box.
     def __init__(self, env_layer, execution_config, composite_logger, telemetry_writer, status_handler):
         super(AptitudePackageManager, self).__init__(env_layer, execution_config, composite_logger, telemetry_writer, status_handler)
+
+        security_list_guid = str(uuid.uuid4())
         # Repo refresh
         self.repo_refresh = 'sudo apt-get -q update'
 
         # Support to get updates and their dependencies
-        self.security_sources_list = '/tmp/az-update-security.list'
-        self.prep_security_sources_list_cmd = 'sudo grep security /etc/apt/sources.list > ' + self.security_sources_list
+        self.security_sources_list = os.path.join(execution_config.temp_folder, 'msft-patch-security-{0}.list'.format(security_list_guid))
+        self.prep_security_sources_list_cmd = 'sudo grep security /etc/apt/sources.list > ' + os.path.normpath(self.security_sources_list)
         self.dist_upgrade_simulation_cmd_template = 'LANG=en_US.UTF8 sudo apt-get -s dist-upgrade <SOURCES> '  # Dist-upgrade simulation template - <SOURCES> needs to be replaced before use; sudo is used as sometimes the sources list needs sudo to be readable
         self.single_package_check_versions = 'apt-cache madison <PACKAGE-NAME>'
         self.single_package_find_installed_dpkg = 'sudo dpkg -s <PACKAGE-NAME>'
diff --git a/src/core/tests/Test_CoreMain.py b/src/core/tests/Test_CoreMain.py
index d0036c8d..e819f68d 100644
--- a/src/core/tests/Test_CoreMain.py
+++ b/src/core/tests/Test_CoreMain.py
@@ -14,9 +14,11 @@
 #
 # Requires Python 2.7+
 import datetime
+import glob
 import json
 import os
 import re
+import shutil
 import time
 import unittest
 import uuid
@@ -46,6 +48,12 @@ def mock_linux_distribution_to_return_centos(self):
     def mock_linux_distribution_to_return_redhat(self):
         return ['Red Hat Enterprise Linux Server', '7.5', 'Maipo']
 
+    def mock_os_remove(self, file_to_remove):
+        raise Exception("File could not be deleted")
+
+    def mock_os_path_exists(self, patch_to_validate):
+        return False
+
     def test_operation_fail_for_non_autopatching_request(self):
         # Test for non auto patching request
         argument_composer = ArgumentComposer()
@@ -880,7 +888,7 @@ def test_assessment_superseded(self):
         scratch_path = os.path.join(os.path.curdir, "scratch")
 
         # Step 2: Set 1.status to Transitioning
-        with open(os.path.join(scratch_path, "1.status"), 'r+') as f:
+        with open(os.path.join(scratch_path, "status", "1.status"), 'r+') as f:
             status = json.load(f)
             status[0]["status"]["status"] = "transitioning"
             status[0]["status"]["substatus"][0]["status"] = "transitioning"
@@ -919,6 +927,98 @@ def test_assessment_superseded(self):
 
         runtime.stop()
 
+    def test_temp_folder_created_during_execution_config_init(self):
+        # temp_folder is set with a path in environment settings but the dir does not exist
+        argument_composer = ArgumentComposer()
+        shutil.rmtree(argument_composer.temp_folder)
+        argument_composer.operation = Constants.ASSESSMENT
+        runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT)
+        # validate temp_folder is created
+        self.assertTrue(runtime.execution_config.temp_folder is not None)
+        self.assertTrue(os.path.exists(runtime.execution_config.temp_folder))
+        runtime.stop()
+
+        # temp_folder is set to None in ExecutionConfig with a valid config_folder location
+        argument_composer = ArgumentComposer()
+        shutil.rmtree(argument_composer.temp_folder)
+        argument_composer.temp_folder = None
+        argument_composer.operation = Constants.ASSESSMENT
+        runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT)
+        # validate temp_folder is created
+        self.assertTrue(runtime.execution_config.temp_folder is not None)
+        self.assertTrue(os.path.exists(runtime.execution_config.temp_folder))
+        runtime.stop()
+
+        # temp_folder is set to None in ExecutionConfig with an invalid config_folder location, throws exception
+        argument_composer = ArgumentComposer()
+        shutil.rmtree(argument_composer.temp_folder)
+        argument_composer.temp_folder = None
+        argument_composer.operation = Constants.ASSESSMENT
+        # mock path exists check to return False on config_folder exists check
+        backup_os_path_exists = os.path.exists
+        os.path.exists = self.mock_os_path_exists
+        self.assertRaises(Exception, lambda: RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT))
+        # validate temp_folder is not created
+        self.assertFalse(os.path.exists(os.path.join(os.path.curdir, "scratch", "tmp")))
+        os.path.exists = backup_os_path_exists
+        runtime.stop()
+
+    def test_delete_temp_folder_contents_success(self):
+        argument_composer = ArgumentComposer()
+        self.assertTrue(argument_composer.temp_folder is not None)
+        self.assertEqual(argument_composer.temp_folder, os.path.abspath(os.path.join(os.path.curdir, "scratch", "tmp")))
+
+        # delete temp content
+        argument_composer.operation = Constants.ASSESSMENT
+        runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT)
+        runtime.set_legacy_test_type('HappyPath')
+        CoreMain(argument_composer.get_composed_arguments())
+
+        # validate files are deleted
+        self.assertTrue(argument_composer.temp_folder is not None)
+        files_matched = glob.glob(str(argument_composer.temp_folder) + "/" + str(Constants.TEMP_FOLDER_CLEANUP_ARTIFACT_LIST))
+        self.assertTrue(len(files_matched) == 0)
+        runtime.stop()
+
+    def test_delete_temp_folder_contents_when_none_exists(self):
+        argument_composer = ArgumentComposer()
+        argument_composer.operation = Constants.ASSESSMENT
+        runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT)
+        shutil.rmtree(runtime.execution_config.temp_folder)
+
+        # attempt to delete temp content
+        runtime.env_layer.file_system.delete_files_from_dir(runtime.execution_config.temp_folder, Constants.TEMP_FOLDER_CLEANUP_ARTIFACT_LIST)
+
+        # validate files are deleted
+        self.assertTrue(runtime.execution_config.temp_folder is not None)
+        files_matched = glob.glob(str(runtime.execution_config.temp_folder) + "/" + str(Constants.TEMP_FOLDER_CLEANUP_ARTIFACT_LIST))
+        self.assertTrue(len(files_matched) == 0)
+        runtime.stop()
+
+    def test_delete_temp_folder_contents_failure(self):
+        argument_composer = ArgumentComposer()
+        self.assertTrue(argument_composer.temp_folder is not None)
+        self.assertEqual(argument_composer.temp_folder, os.path.abspath(os.path.join(os.path.curdir, "scratch", "tmp")))
+
+        # mock os.remove()
+        self.backup_os_remove = os.remove
+        os.remove = self.mock_os_remove
+
+        argument_composer.operation = Constants.ASSESSMENT
+        runtime = RuntimeCompositor(argument_composer.get_composed_arguments(), True, Constants.APT)
+
+        # delete temp content attempt #1, throws exception
+        self.assertRaises(Exception, lambda: runtime.env_layer.file_system.delete_files_from_dir(runtime.execution_config.temp_folder, Constants.TEMP_FOLDER_CLEANUP_ARTIFACT_LIST, raise_if_delete_failed=True))
+        self.assertTrue(os.path.isfile(os.path.join(runtime.execution_config.temp_folder, "temp1.list")))
+
+        # delete temp content attempt #2, does not throws exception
+        runtime.env_layer.file_system.delete_files_from_dir(runtime.execution_config.temp_folder, Constants.TEMP_FOLDER_CLEANUP_ARTIFACT_LIST)
+        self.assertTrue(os.path.isfile(os.path.join(runtime.execution_config.temp_folder, "temp1.list")))
+
+        # reset os.remove() mock
+        os.remove = self.backup_os_remove
+        runtime.stop()
+
     def __check_telemetry_events(self, runtime):
         all_events = os.listdir(runtime.telemetry_writer.events_folder_path)
         self.assertTrue(len(all_events) > 0)
diff --git a/src/core/tests/library/ArgumentComposer.py b/src/core/tests/library/ArgumentComposer.py
index 6ba6e201..2b79290c 100644
--- a/src/core/tests/library/ArgumentComposer.py
+++ b/src/core/tests/library/ArgumentComposer.py
@@ -31,14 +31,22 @@ def __init__(self):
         self.__TESTS_FOLDER = "tests"
         self.__SCRATCH_FOLDER = "scratch"
         self.__ARG_TEMPLATE = "{0} {1} {2} {3} \'{4}\' {5} \'{6}\' {7} {8}"
+        self.__CONFIG_FOLDER = "config"
+        self.__STATUS_FOLDER = "status"
+        self.__LOG_FOLDER = "log"
         self.__EVENTS_FOLDER = "events"
+        self.__TEMP_FOLDER = "tmp"
 
         # sequence number
         self.sequence_number = 1
 
         # environment settings
-        self.__log_folder = self.__config_folder = self.__status_folder = self.__get_scratch_folder()
-        self.events_folder = self.__get_events_folder(self.__log_folder)
+        scratch_folder = self.__get_scratch_folder()
+        self.__log_folder = self.__get_custom_folder(scratch_folder, self.__LOG_FOLDER)
+        self.__config_folder = self.__get_custom_folder(scratch_folder, self.__CONFIG_FOLDER)
+        self.__status_folder = self.__get_custom_folder(scratch_folder, self.__STATUS_FOLDER)
+        self.events_folder = self.__get_custom_folder(self.__log_folder, self.__EVENTS_FOLDER)
+        self.temp_folder = self.__get_custom_folder(scratch_folder, self.__TEMP_FOLDER)
 
         # config settings
         self.operation = Constants.INSTALLATION
@@ -67,6 +75,7 @@ def get_composed_arguments(self, env_settings={}):
             "configFolder": self.__config_folder,
             "statusFolder": self.__status_folder,
             "eventsFolder": self.events_folder,
+            "tempFolder": self.temp_folder,
             "telemetrySupported": True
         }
 
@@ -111,14 +120,15 @@ def __get_scratch_folder(self):
             os.mkdir(scratch_folder)
         return scratch_folder
 
-    def __get_events_folder(self, scratch_folder):
-        """ Returns a predetermined events folder and guarantees it exists and is empty. """
-        events_folder = os.path.join(scratch_folder, self.__EVENTS_FOLDER)
-        if os.path.exists(events_folder):
-            shutil.rmtree(events_folder, ignore_errors=True)
-        if not os.path.exists(events_folder):
-            os.mkdir(events_folder)
-        return events_folder
+    @staticmethod
+    def __get_custom_folder(par_dir, custom_folder_name):
+        """ Returns a predetermined custom folder, and guarantees it exists and is empty. """
+        custom_folder = os.path.join(par_dir, custom_folder_name)
+        if os.path.exists(custom_folder):
+            shutil.rmtree(custom_folder, ignore_errors=True)
+        if not os.path.exists(custom_folder):
+            os.mkdir(custom_folder)
+        return custom_folder
 
     def __try_get_tests_folder(self, path=os.getcwd()):
         """ Returns the current working directory if there's no folder with tests in its name in the absolute path
diff --git a/src/core/tests/library/LegacyEnvLayerExtensions.py b/src/core/tests/library/LegacyEnvLayerExtensions.py
index 24abb8cb..95ac5af2 100644
--- a/src/core/tests/library/LegacyEnvLayerExtensions.py
+++ b/src/core/tests/library/LegacyEnvLayerExtensions.py
@@ -13,7 +13,7 @@
 # limitations under the License.
 #
 # Requires Python 2.7+
-
+import os
 import sys
 from core.src.bootstrap.Constants import Constants
 
@@ -22,6 +22,7 @@ class LegacyEnvLayerExtensions():
     def __init__(self, package_manager_name):
         self.legacy_package_manager_name = package_manager_name
         self.legacy_test_type = "HappyPath"
+        self.temp_folder_path = ""
 
     class LegacyPlatform(object):
         def linux_distribution(self):
@@ -43,6 +44,9 @@ def get_package_manager(self):
         """return passed in package manager name"""
         return self.legacy_package_manager_name
 
+    def set_temp_folder_path(self, temp_folder_path):
+        self.temp_folder_path = temp_folder_path
+
     @staticmethod
     def get_python_major_version():
         if hasattr(sys.version_info, 'major'):
@@ -522,6 +526,10 @@ def run_command_output(self, cmd, no_output=False, chk_err=True):
                         output = "      bash | 4.3-14ubuntu1.3 | http://us.archive.ubuntu.com/ubuntu xenial-updates/main amd64 Packages\n" + \
                                  "      bash | 4.3-14ubuntu1.2 | http://security.ubuntu.com/ubuntu xenial-security/main amd64 Packages\n" + \
                                  "      bash | 4.3-14ubuntu1 | http://us.archive.ubuntu.com/ubuntu xenial/main amd64 Packages"
+                    elif cmd.find('sudo grep security /etc/apt/sources.list >') > -1:
+                        self.write_to_file(os.path.join(self.temp_folder_path, "temp2.list"), "test temp file 2")
+                        code = 0
+                        output = "tmp file created"
             elif self.legacy_test_type == 'SadPath':
                 if cmd.find("cat /proc/cpuinfo | grep name") > -1:
                     code = 0
@@ -953,8 +961,6 @@ def run_command_output(self, cmd, no_output=False, chk_err=True):
                         code = 0
                         output = "Error: Cannot retrieve repository metadata (repomd.xml) for repository: addons. Please verify its path and try again"
 
-
-
             major_version = self.get_python_major_version()
             if major_version == 2:
                 return code, output.decode('utf8', 'ignore').encode('ascii', 'ignore')
@@ -962,3 +968,8 @@ def run_command_output(self, cmd, no_output=False, chk_err=True):
                 return code, output.encode('ascii', 'ignore').decode('ascii', 'ignore')
             else:
                 raise Exception("Unknown version of python encountered.")
+
+    @staticmethod
+    def write_to_file(path, data):
+        with open(path, "w+") as file_handle:
+            file_handle.write(data)
diff --git a/src/core/tests/library/RuntimeCompositor.py b/src/core/tests/library/RuntimeCompositor.py
index 98180c71..f6b7aa66 100644
--- a/src/core/tests/library/RuntimeCompositor.py
+++ b/src/core/tests/library/RuntimeCompositor.py
@@ -30,14 +30,15 @@
 
 # Todo: find a different way to import these
 try:
-    import urllib2 as urlreq   #Python 2.x
+    import urllib2 as urlreq   # Python 2.x
 except:
-    import urllib.request as urlreq   #Python 3.x
+    import urllib.request as urlreq   # Python 3.x
 
 try:
-    from StringIO import StringIO ## for Python 2
+    from StringIO import StringIO # for Python 2
 except ImportError:
-    from io import StringIO ## for Python 3
+    from io import StringIO # for Python 3
+
 
 class RuntimeCompositor(object):
     def __init__(self, argv=Constants.DEFAULT_UNSPECIFIED_VALUE, legacy_mode=False, package_manager_name=Constants.APT, vm_cloud_type=Constants.VMCloudType.AZURE):
@@ -88,6 +89,7 @@ def mkdtemp_runner():
 
         # Business logic components
         self.execution_config = self.container.get('execution_config')
+        self.legacy_env_layer_extensions.set_temp_folder_path(self.execution_config.temp_folder)
         self.package_manager = self.container.get('package_manager')
         self.backup_get_current_auto_os_patch_state = None
         self.reconfigure_package_manager()
@@ -102,6 +104,9 @@ def mkdtemp_runner():
         # Extension handler dependency
         self.write_ext_state_file(self.lifecycle_manager.ext_state_file_path, self.execution_config.sequence_number, datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ"), self.execution_config.operation)
 
+        # Write file to temp dir
+        self.write_to_file(os.path.join(self.execution_config.temp_folder, "temp1.list"), "test temp file")
+
         # Mock service and timer creation and removal used for Auto Assessment
         self.backup_create_and_set_service_idem = self.configure_patching_processor.auto_assess_service_manager.create_and_set_service_idem
         self.configure_patching_processor.auto_assess_service_manager.create_and_set_service_idem = self.mock_create_and_set_service_idem
diff --git a/src/extension/src/ActionHandler.py b/src/extension/src/ActionHandler.py
index ad8050ba..c2791831 100644
--- a/src/extension/src/ActionHandler.py
+++ b/src/extension/src/ActionHandler.py
@@ -238,6 +238,9 @@ def update(self):
             # copy all required files from preceding version to current
             self.copy_config_files(preceding_version_path, new_version_config_folder)
 
+            # Delete temp_folder
+            self.ext_env_handler.delete_temp_folder()
+
             self.logger.log("All update actions from extension handler completed.")
             self.ext_output_status_handler.write_status_file("", self.seq_no, status=Constants.Status.Success.lower())
             return Constants.ExitCode.Okay
@@ -293,6 +296,10 @@ def copy_config_files(self, src, dst, raise_if_not_copied=False):
     def uninstall(self):
         try:
             self.setup(action=Constants.UNINSTALL, log_message="Extension uninstalled")
+
+            # Delete temp_folder
+            self.ext_env_handler.delete_temp_folder()
+
             self.ext_output_status_handler.write_status_file("", self.seq_no, status=Constants.Status.Success.lower())
             return Constants.ExitCode.Okay
 
@@ -340,6 +347,10 @@ def disable(self):
                                       "\r\n printf \"Auto-assessment was paused by the Azure Linux Patch Extension.\""
                 self.env_layer.file_system.write_with_retry(auto_assess_sh_path, auto_assess_sh_data)
                 self.env_layer.run_command_output("chmod a+x " + auto_assess_sh_path)
+
+                # Clear temp folder
+                self.ext_env_handler.delete_temp_folder_contents()
+
             except Exception as error:
                 self.logger.log_error("Error occurred during auto-assessment disable. [Error={0}]".format(repr(error)))
             # End of temporary auto-assessment disablement
@@ -360,6 +371,10 @@ def reset(self):
             self.setup(action=Constants.RESET, log_message="Reset triggered on extension, deleting CoreState and ExtState files")
             self.utility.delete_file(self.core_state_handler.dir_path, self.core_state_handler.file, raise_if_not_found=False)
             self.utility.delete_file(self.ext_state_handler.dir_path, self.ext_state_handler.file, raise_if_not_found=False)
+
+            # Clear temp folder
+            self.ext_env_handler.delete_temp_folder_contents()
+
             self.ext_output_status_handler.write_status_file("", self.seq_no, status=Constants.Status.Success.lower())
             return Constants.ExitCode.Okay
 
diff --git a/src/extension/src/Constants.py b/src/extension/src/Constants.py
index e6001807..b509fffc 100644
--- a/src/extension/src/Constants.py
+++ b/src/extension/src/Constants.py
@@ -136,9 +136,12 @@ class EnvSettingsFields(EnumBackport):
         config_folder = "configFolder"
         status_folder = "statusFolder"
         events_folder = "eventsFolder"
+        temp_folder = "tempFolder"
         events_folder_preview = "eventsFolder_preview"
         telemetry_supported = "telemetrySupported"
 
+    TEMP_FOLDER_DIR_NAME = "tmp"
+
     # Config Settings json keys
     RUNTIME_SETTINGS = "runtimeSettings"
     HANDLER_SETTINGS = "handlerSettings"
diff --git a/src/extension/src/EnableCommandHandler.py b/src/extension/src/EnableCommandHandler.py
index 2f6ecd34..22dd404f 100644
--- a/src/extension/src/EnableCommandHandler.py
+++ b/src/extension/src/EnableCommandHandler.py
@@ -68,6 +68,9 @@ def execute_handler_action(self):
             self.ext_state_handler.create_file(self.seq_no, operation, prev_patch_max_end_time)
             core_state_content = self.core_state_handler.read_file()
 
+            # log tmp folder size
+            self.ext_env_handler.log_temp_folder_details()
+
             # if NoOperation is requested, terminate all running processes from previous operation and update status file
             if operation == Constants.NOOPERATION:
                 self.process_nooperation(config_settings, core_state_content)
@@ -115,6 +118,9 @@ def process_reenable_request(self, config_settings, core_state_content):
 
     def launch_new_process(self, config_settings, create_status_output_file):
         """ Creates <sequence number>.status to report the current request's status and launches core code to handle the requested operation """
+        # Clear temp folder since a new Core process is to be launched
+        self.ext_env_handler.delete_temp_folder_contents()
+
         # create Status file
         if create_status_output_file:
             self.ext_output_status_handler.write_status_file(config_settings.__getattribute__(self.config_public_settings.operation), self.seq_no, status=self.status.Transitioning.lower())
diff --git a/src/extension/src/EnvLayer.py b/src/extension/src/EnvLayer.py
index 22403ca6..de85ffd6 100644
--- a/src/extension/src/EnvLayer.py
+++ b/src/extension/src/EnvLayer.py
@@ -16,8 +16,10 @@
 
 from __future__ import print_function
 import getpass
+import glob
 import os
 import re
+import shutil
 import subprocess
 import sys
 import time
@@ -244,4 +246,39 @@ def write_with_retry(self, file_path_or_handle, data, mode='a+'):
             if was_path:  # what was passed in was not a file handle, so close the handle that was init here
                 file_handle.close()
 
+        @staticmethod
+        def delete_files_from_dir(dir_name, file_identifier_list, raise_if_delete_failed=False):
+            """ Clears all files from given dir. NOTE: Uses file_identifier_list to determine the content to delete """
+            for file_identifier in file_identifier_list:
+                files_to_delete = glob.glob(str(dir_name) + "/" + str(file_identifier))
+
+                for file_to_delete in files_to_delete:
+                    try:
+                        os.remove(file_to_delete)
+                    except Exception as error:
+                        error_message = "Unable to delete files from directory [Dir={0}][File={1}][Error={2}][RaiseIfDeleteFailed={3}].".format(
+                            str(dir_name), str(file_to_delete), repr(error), str(raise_if_delete_failed))
+                        if raise_if_delete_failed:
+                            raise Exception(error_message)
+                        else:
+                            print(error_message)
+                            return None
+
+        @staticmethod
+        def remove_dir(dir_name, raise_if_delete_failed=False):
+            """ Deletes given directory and all of it's contents """
+            try:
+                shutil.rmtree(dir_name)
+            except Exception as error:
+                error_message = "Unable to delete directory [Dir={0}][Error={1}][RaiseIfDeleteFailed={2}].".format(
+                    str(dir_name),
+                    repr(error),
+                    str(raise_if_delete_failed))
+
+                if raise_if_delete_failed:
+                    raise Exception(error_message)
+                else:
+                    print(error_message)
+                    return None
+
 # endregion - File system
diff --git a/src/extension/src/ProcessHandler.py b/src/extension/src/ProcessHandler.py
index 85762787..5794c314 100644
--- a/src/extension/src/ProcessHandler.py
+++ b/src/extension/src/ProcessHandler.py
@@ -63,6 +63,7 @@ def get_env_settings(ext_env_handler):
             env_settings.update({env_settings_keys.config_folder: ext_env_handler.config_folder})
             env_settings.update({env_settings_keys.status_folder: ext_env_handler.status_folder})
             env_settings.update({env_settings_keys.events_folder: ext_env_handler.events_folder})
+            env_settings.update({env_settings_keys.temp_folder: ext_env_handler.temp_folder})
             env_settings.update({env_settings_keys.telemetry_supported: ext_env_handler.telemetry_supported})
         return env_settings
 
@@ -95,6 +96,9 @@ def start_daemon(self, seq_no, config_settings, ext_env_handler):
             self.logger.log("New shell process launched successfully. [Process ID (PID)={0}]".format(str(process.pid)))
             did_process_start = self.__check_process_state(process, seq_no)
             return process if did_process_start else None
+
+        # Clear temp folder since core process launch failed
+        ext_env_handler.delete_temp_folder_contents()
         self.logger.log_error("Error launching process for given sequence. [sequence={0}]".format(seq_no))
 
     def stage_auto_assess_sh_safely(self, core_process_command):
diff --git a/src/extension/src/__main__.py b/src/extension/src/__main__.py
index c1742af4..b178f21b 100644
--- a/src/extension/src/__main__.py
+++ b/src/extension/src/__main__.py
@@ -49,7 +49,7 @@ def main(argv):
         utility = Utility(logger)
         runtime_context_handler = RuntimeContextHandler(logger)
         json_file_handler = JsonFileHandler(logger)
-        ext_env_handler = ExtEnvHandler(json_file_handler)
+        ext_env_handler = ExtEnvHandler(logger, env_layer, json_file_handler)
         env_health_manager = EnvHealthManager(env_layer)
         if ext_env_handler.handler_environment_json is not None and ext_env_handler.config_folder is not None:
             config_folder = ext_env_handler.config_folder
diff --git a/src/extension/src/file_handlers/ExtEnvHandler.py b/src/extension/src/file_handlers/ExtEnvHandler.py
index fb1e7487..9cbec33c 100644
--- a/src/extension/src/file_handlers/ExtEnvHandler.py
+++ b/src/extension/src/file_handlers/ExtEnvHandler.py
@@ -13,6 +13,8 @@
 # limitations under the License.
 #
 # Requires Python 2.7+
+import glob
+import os
 
 from extension.src.Constants import Constants
 
@@ -35,7 +37,9 @@
 
 class ExtEnvHandler(object):
     """ Responsible for all operations with HandlerEnvironment.json file and other environment config """
-    def __init__(self, json_file_handler, handler_env_file=Constants.HANDLER_ENVIRONMENT_FILE, handler_env_file_path=Constants.HANDLER_ENVIRONMENT_FILE_PATH):
+    def __init__(self, logger, env_layer, json_file_handler, handler_env_file=Constants.HANDLER_ENVIRONMENT_FILE, handler_env_file_path=Constants.HANDLER_ENVIRONMENT_FILE_PATH):
+        self.logger = logger
+        self.env_layer = env_layer
         json_file_handler = json_file_handler
         self.env_settings_all_keys = Constants.EnvSettingsFields
 
@@ -48,6 +52,8 @@ def __init__(self, json_file_handler, handler_env_file=Constants.HANDLER_ENVIRON
             if self.events_folder is None:
                 self.events_folder = self.get_ext_env_config_value_safely(self.env_settings_all_keys.events_folder_preview, raise_if_not_found=False)
 
+            self.temp_folder = self.get_temp_folder()
+
         self.telemetry_supported = False
 
     def get_ext_env_config_value_safely(self, key, raise_if_not_found=True):
@@ -64,3 +70,49 @@ def get_ext_env_config_value_safely(self, key, raise_if_not_found=True):
                 else:
                     return None
         return None
+
+    def get_temp_folder(self):
+        """ Returns path to the temp folder, if one exists. If not, creates a temp folder and returns it's path """
+        par_dir = os.path.dirname(self.config_folder)
+        temp_folder_path = os.path.join(par_dir, Constants.TEMP_FOLDER_DIR_NAME)
+        if not os.path.exists(par_dir):
+            raise Exception("Parent directory for all extension artifacts such as config folder, status folder, etc. not found at [{0}].".format(repr(par_dir)))
+
+        if not os.path.exists(temp_folder_path):
+            os.mkdir(temp_folder_path)
+        return temp_folder_path
+
+    def delete_temp_folder_contents(self, raise_if_delete_failed=False):
+        """ Clears all artifacts from temp_folder."""
+        # clean up temp folder after all operation execution is finished from Core
+        if self.env_layer is not None \
+                and self.temp_folder is not None \
+                and os.path.exists(self.temp_folder):
+            self.logger.log_debug("Deleting all files of certain format from temp folder [FileFormat={0}][TempFolderLocation={1}]".format("*", str(self.temp_folder)))
+            self.env_layer.file_system.delete_files_from_dir(self.temp_folder, ["*"], raise_if_delete_failed=raise_if_delete_failed)
+        else:
+            self.logger.log_debug("Temp folder not found")
+
+    def delete_temp_folder(self, raise_if_delete_failed=False):
+        """ Deletes temp_folder and all of it's contents """
+        if self.env_layer is not None \
+                and self.temp_folder is not None \
+                and os.path.exists(self.temp_folder):
+            self.logger.log_debug("Deleting all files of certain format from temp folder [FileFormat={0}][TempFolderLocation={1}]".format("*", str(self.temp_folder)))
+            self.env_layer.file_system.remove_dir(self.temp_folder, raise_if_delete_failed=raise_if_delete_failed)
+        else:
+            self.logger.log_debug("Temp folder not found")
+
+    def log_temp_folder_details(self):
+        """ Computes size of temp folder from all files in it. NOTE: Does not include dirs within temp folder for this calculation """
+        if self.temp_folder is not None and os.path.exists(self.temp_folder):
+            size = 0
+            file_count = 0
+            for path, dirs, files in os.walk(self.temp_folder):
+                for f in files:
+                    fp = os.path.join(path, f)
+                    size += os.stat(fp).st_size
+                    file_count += 1
+            self.logger.log_debug("Temp folder details: [Location={0}][TotalSizeOfAllFiles={1}][TotalNumberOfFiles-{2}]".format(str(self.temp_folder), str(size), str(file_count)))
+        else:
+            self.logger.log_debug("Temp folder not found")
diff --git a/src/extension/tests/Test_ActionHandler.py b/src/extension/tests/Test_ActionHandler.py
index 5d3d514e..d494c1fa 100644
--- a/src/extension/tests/Test_ActionHandler.py
+++ b/src/extension/tests/Test_ActionHandler.py
@@ -43,7 +43,12 @@ def setUp(self):
 
         self.runtime = RuntimeComposer()
         runtime_context_handler = RuntimeContextHandler(self.runtime.logger)
-        self.ext_env_handler = ExtEnvHandler(self.runtime.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
+
+        # Mock temp folder setup in ExtEnvHandler
+        self.ext_env_handler_get_temp_folder_backup = ExtEnvHandler.get_temp_folder
+        ExtEnvHandler.get_temp_folder = self.mock_get_temp_folder
+
+        self.ext_env_handler = ExtEnvHandler(self.runtime.logger, self.runtime.env_layer, self.runtime.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
         self.ext_env_handler.telemetry_supported = True
         self.setup_files_and_folders(self.temp_dir)
 
@@ -69,6 +74,10 @@ def tearDown(self):
         VirtualTerminal().print_lowlight("\n----------------- tear down test runner -----------------")
         self.ext_config_settings_handler.get_seq_no_from_env_var = self.backup_get_seq_no_from_env_var
         os.path.realpath = self.backup_mock_os_path_realpath
+
+        # reset temp folder mock from ExtEnvHandler
+        ExtEnvHandler.get_temp_folder = self.ext_env_handler_get_temp_folder_backup
+
         # delete tempdir
         shutil.rmtree(self.temp_dir)
 
@@ -83,11 +92,13 @@ def setup_files_and_folders(self, temp_dir):
         status_folder_complete_path = os.path.join(temp_dir, status_folder)
         log_folder_complete_path = os.path.join(temp_dir, log_folder)
         events_folder_complete_path = os.path.join(temp_dir, log_folder, events_folder)
+        temp_folder_complete_path = os.path.join(os.path.dirname(config_folder_complete_path), Constants.TEMP_FOLDER_DIR_NAME)
 
         os.mkdir(config_folder_complete_path)
         os.mkdir(status_folder_complete_path)
         os.mkdir(log_folder_complete_path)
         os.mkdir(events_folder_complete_path)
+        os.mkdir(temp_folder_complete_path)
 
         # copying a sample version of the <seqno>.settings file from the helpers folder to the temp directory
         shutil.copy(os.path.join("helpers", "1234.settings"), config_folder_complete_path)
@@ -97,6 +108,7 @@ def setup_files_and_folders(self, temp_dir):
         self.ext_env_handler.status_folder = status_folder_complete_path
         self.ext_env_handler.log_folder = log_folder_complete_path
         self.ext_env_handler.events_folder = events_folder_complete_path
+        self.ext_env_handler.temp_folder = temp_folder_complete_path
 
     def mock_get_seq_no_from_env_var(self):
         return 1234
@@ -126,6 +138,9 @@ def mock_setup_throws_exception(self, action, log_message):
     def mock_validate_os_type(self):
         return True
 
+    def mock_get_temp_folder(self):
+        return "testTempFolder"
+
     @staticmethod
     def create_latest_extension_dir(version, test_dir):
         latest_extension_version = version
diff --git a/src/extension/tests/Test_EnableCommandHandler.py b/src/extension/tests/Test_EnableCommandHandler.py
index ad4d1dbe..cd530e80 100644
--- a/src/extension/tests/Test_EnableCommandHandler.py
+++ b/src/extension/tests/Test_EnableCommandHandler.py
@@ -41,6 +41,11 @@ def setUp(self):
         # create tempdir which will have all the required files
         self.temp_dir = tempfile.mkdtemp()
         runtime = RuntimeComposer()
+
+        # Mock temp folder setup in ExtEnvHandler
+        self.ext_env_handler_get_temp_folder_backup = ExtEnvHandler.get_temp_folder
+        ExtEnvHandler.get_temp_folder = self.mock_get_temp_folder
+
         self.logger = runtime.logger
         self.telemetry_writer = runtime.telemetry_writer
         self.logger.telemetry_writer = self.telemetry_writer
@@ -48,7 +53,7 @@ def setUp(self):
         self.env_health_manager = runtime.env_health_manager
         self.json_file_handler = runtime.json_file_handler
         self.runtime_context_handler = RuntimeContextHandler(self.logger)
-        self.ext_env_handler = ExtEnvHandler(self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
+        self.ext_env_handler = ExtEnvHandler(self.logger, runtime.env_layer, self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
         self.ext_env_handler.telemetry_supported = True
         self.config_folder = self.ext_env_handler.config_folder
         self.ext_config_settings_handler = ExtConfigSettingsHandler(self.logger, self.json_file_handler, self.config_folder)
@@ -65,6 +70,10 @@ def tearDown(self):
         VirtualTerminal().print_lowlight("\n----------------- tear down test runner -----------------")
         # reseting mocks to their original definition
         ProcessHandler.start_daemon = self.start_daemon_backup
+
+        # reset temp folder mock from ExtEnvHandler
+        ExtEnvHandler.get_temp_folder = self.ext_env_handler_get_temp_folder_backup
+
         self.logger.file_logger.close()
         # delete tempdir
         shutil.rmtree(self.temp_dir)
@@ -75,6 +84,9 @@ def mock_start_daemon_to_return_true(self, seq_no, config_settings, ext_env_hand
     def mock_patch_complete_time_check_to_return_true(self, time_for_prev_patch_to_complete, core_state_last_heartbeat, core_state_handler):
         return True
 
+    def mock_get_temp_folder(self):
+        return "testTempFolder"
+
     def test_enable_command_first_request(self):
         self.check_if_patch_completes_in_time_backup = RuntimeContextHandler.check_if_patch_completes_in_time
         RuntimeContextHandler.check_if_patch_completes_in_time = self.mock_patch_complete_time_check_to_return_true
@@ -237,10 +249,12 @@ def setup_for_enable_handler(self, dir_path):
         config_folder_complete_path = os.path.join(dir_path, config_folder)
         status_folder_complete_path = os.path.join(dir_path, status_folder)
         log_folder_complete_path = os.path.join(dir_path, log_folder)
+        temp_folder_complete_path = os.path.join(os.path.dirname(config_folder_complete_path), Constants.TEMP_FOLDER_DIR_NAME)
 
         os.mkdir(config_folder_complete_path)
         os.mkdir(status_folder_complete_path)
         os.mkdir(log_folder_complete_path)
+        os.mkdir(temp_folder_complete_path)
 
         # copying a sample version of the <seqno>.settings file from the helpers folder to the temp directory
         shutil.copy(os.path.join("helpers", "1234.settings"), config_folder_complete_path)
@@ -261,6 +275,7 @@ def setup_for_enable_handler(self, dir_path):
         self.ext_env_handler.config_folder = config_folder_complete_path
         self.ext_env_handler.status_folder = status_folder_complete_path
         self.ext_env_handler.log_folder = log_folder_complete_path
+        self.ext_env_handler.temp_folder = temp_folder_complete_path
 
         self.enable_command_handler.ext_env_handler = self.ext_env_handler
 
diff --git a/src/extension/tests/Test_ExtEnvHandler.py b/src/extension/tests/Test_ExtEnvHandler.py
index b89c7c30..bead1170 100644
--- a/src/extension/tests/Test_ExtEnvHandler.py
+++ b/src/extension/tests/Test_ExtEnvHandler.py
@@ -20,6 +20,8 @@
 import unittest
 from extension.src.Constants import Constants
 from extension.src.file_handlers.ExtEnvHandler import ExtEnvHandler
+from extension.src.local_loggers.FileLogger import FileLogger
+from extension.src.local_loggers.Logger import Logger
 from extension.tests.helpers.RuntimeComposer import RuntimeComposer
 from extension.tests.helpers.VirtualTerminal import VirtualTerminal
 
@@ -31,21 +33,65 @@ def setUp(self):
         self.json_file_handler = self.runtime.json_file_handler
         self.env_settings_fields = Constants.EnvSettingsFields
 
+        self.backup_pathexists = os.path.exists
+        os.path.exists = self.mock_os_pathexists
+
     def tearDown(self):
         VirtualTerminal().print_lowlight("\n----------------- tear down test runner -----------------")
+        os.path.exists = self.backup_pathexists
+
+    def mock_os_pathexists(self, path):
+        return True
+
+    def mock_os_remove(self, file_to_remove):
+        raise Exception("File could not be deleted")
+
+    def mock_shutil_rmtree(self, dir_to_remove):
+        raise Exception("Directory could not be deleted")
+
+    def __create_ext_env_handler_and_validate_tmp_folder(self, test_dir):
+        # Reset os.pathexists that was mocked in setup()
+        os.path.exists = self.backup_pathexists
+
+        # create temp folder
+        ext_env_settings = [{
+            Constants.EnvSettingsFields.version: "1.0",
+            Constants.EnvSettingsFields.settings_parent_key: {
+                Constants.EnvSettingsFields.log_folder: os.path.join(test_dir, "testLog"),
+                Constants.EnvSettingsFields.config_folder: os.path.join(test_dir, "testConfig"),
+                Constants.EnvSettingsFields.status_folder: os.path.join(test_dir, "testStatus"),
+                Constants.EnvSettingsFields.events_folder_preview: os.path.join(test_dir, "testEventsPreview")
+            }
+        }]
+        file_name = Constants.HANDLER_ENVIRONMENT_FILE
+        self.runtime.create_temp_file(test_dir, file_name, content=json.dumps(ext_env_settings))
+        ext_env_handler = ExtEnvHandler(self.runtime.logger, self.runtime.env_layer, self.json_file_handler, handler_env_file_path=test_dir)
+        self.assertTrue(ext_env_handler.config_folder is not None)
+        self.assertTrue(ext_env_handler.temp_folder is not None)
+        self.assertEqual(ext_env_handler.temp_folder, os.path.join(test_dir, "tmp"))
+
+        # add files to tmp folder
+        self.runtime.create_temp_file(ext_env_handler.temp_folder, "Test1.list", content='')
+        self.runtime.create_temp_file(ext_env_handler.temp_folder, "Test2.list", content='')
+        self.assertTrue(os.path.isfile(os.path.join(ext_env_handler.temp_folder, "Test1.list")))
+        self.assertTrue(os.path.isfile(os.path.join(ext_env_handler.temp_folder, "Test2.list")))
+
+        return ext_env_handler
 
     def test_file_read_success(self):
-        ext_env_handler = ExtEnvHandler(self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
+        ext_env_handler = ExtEnvHandler(self.runtime.logger, self.runtime.env_layer, self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
         self.assertTrue(ext_env_handler.log_folder is not None)
         self.assertEqual(ext_env_handler.log_folder, "mockLog")
         self.assertTrue(ext_env_handler.status_folder is not None)
+        self.assertTrue(ext_env_handler.temp_folder is not None)
+        self.assertEqual(ext_env_handler.temp_folder, "tmp")
 
     def test_file_read_failure(self):
         # empty file
         test_dir = tempfile.mkdtemp()
         file_name = "test_handler_env.json"
         self.runtime.create_temp_file(test_dir, file_name, content=None)
-        self.assertRaises(Exception, ExtEnvHandler, self.json_file_handler, handler_env_file=file_name, handler_env_file_path=test_dir)
+        self.assertRaises(Exception, ExtEnvHandler, self.runtime.logger, self.runtime.env_layer, self.json_file_handler, handler_env_file=file_name, handler_env_file_path=test_dir)
         shutil.rmtree(test_dir)
 
         # invalid file content
@@ -53,7 +99,7 @@ def test_file_read_failure(self):
         test_dir = tempfile.mkdtemp()
         file_name = "test_handler_env.json"
         self.runtime.create_temp_file(test_dir, file_name, str(json_content))
-        self.assertRaises(Exception, ExtEnvHandler, self.json_file_handler, handler_env_file=file_name, handler_env_file_path=test_dir)
+        self.assertRaises(Exception, ExtEnvHandler, self.runtime.logger, self.runtime.env_layer, self.json_file_handler, handler_env_file=file_name, handler_env_file_path=test_dir)
         shutil.rmtree(test_dir)
 
         # invalid file content
@@ -61,7 +107,7 @@ def test_file_read_failure(self):
         test_dir = tempfile.mkdtemp()
         file_name = "test_handler_env.json"
         self.runtime.create_temp_file(test_dir, file_name, str(json_content))
-        self.assertRaises(Exception, ExtEnvHandler, self.json_file_handler, handler_env_file=file_name, handler_env_file_path=test_dir)
+        self.assertRaises(Exception, ExtEnvHandler, self.runtime.logger, self.runtime.env_layer, self.json_file_handler, handler_env_file=file_name, handler_env_file_path=test_dir)
         shutil.rmtree(test_dir)
 
     def test_read_event_folder_preview(self):
@@ -77,8 +123,174 @@ def test_read_event_folder_preview(self):
         test_dir = tempfile.mkdtemp()
         file_name = Constants.HANDLER_ENVIRONMENT_FILE
         self.runtime.create_temp_file(test_dir, file_name, content=json.dumps(ext_env_settings))
-        ext_env_handler = ExtEnvHandler(self.json_file_handler, handler_env_file_path=test_dir)
+        ext_env_handler = ExtEnvHandler(self.runtime.logger, self.runtime.env_layer, self.json_file_handler, handler_env_file_path=test_dir)
         self.assertTrue(ext_env_handler.log_folder is not None)
         self.assertEqual(ext_env_handler.events_folder, "testEventsPreview")
         shutil.rmtree(test_dir)
 
+    def test_temp_folder_creation_success(self):
+        test_dir = tempfile.mkdtemp()
+        ext_env_handler = self.__create_ext_env_handler_and_validate_tmp_folder(test_dir)
+        shutil.rmtree(test_dir)
+
+    def test_delete_temp_folder_contents_success(self):
+        test_dir = tempfile.mkdtemp()
+        ext_env_handler = self.__create_ext_env_handler_and_validate_tmp_folder(test_dir)
+
+        # delete temp content
+        ext_env_handler.delete_temp_folder_contents()
+
+        # validate files are deleted
+        self.assertFalse(os.path.isfile(os.path.join(ext_env_handler.temp_folder, "Test1.list")))
+        self.assertFalse(os.path.isfile(os.path.join(ext_env_handler.temp_folder, "Test2.list")))
+        shutil.rmtree(test_dir)
+
+    def test_delete_temp_folder_contents_when_none_exists(self):
+        ext_env_handler = ExtEnvHandler(self.runtime.logger, self.runtime.env_layer, self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
+        self.assertTrue(ext_env_handler.log_folder is not None)
+        self.assertEqual(ext_env_handler.log_folder, "mockLog")
+        self.assertTrue(ext_env_handler.status_folder is not None)
+        self.assertTrue(ext_env_handler.temp_folder is not None)
+        self.assertEqual(ext_env_handler.temp_folder, "tmp")
+
+        # Reset os.pathexists that was mocked in setup()
+        os.path.exists = self.backup_pathexists
+        # delete temp content
+        ext_env_handler.delete_temp_folder_contents()
+
+    def test_delete_temp_folder_contents_failure(self):
+        test_dir = tempfile.mkdtemp()
+        ext_env_handler = self.__create_ext_env_handler_and_validate_tmp_folder(test_dir)
+
+        # mock os.remove()
+        self.backup_os_remove = os.remove
+        os.remove = self.mock_os_remove
+
+        # delete temp content attempt #1, throws exception
+        self.assertRaises(Exception, lambda: ext_env_handler.delete_temp_folder_contents(raise_if_delete_failed=True))
+        self.assertTrue(os.path.isfile(os.path.join(ext_env_handler.temp_folder, "Test1.list")))
+        self.assertTrue(os.path.isfile(os.path.join(ext_env_handler.temp_folder, "Test2.list")))
+
+        # delete temp content attempt #2, does not throws exception
+        ext_env_handler.delete_temp_folder_contents()
+        self.assertTrue(os.path.isfile(os.path.join(ext_env_handler.temp_folder, "Test1.list")))
+        self.assertTrue(os.path.isfile(os.path.join(ext_env_handler.temp_folder, "Test2.list")))
+
+        # reset os.remove() mock
+        os.remove = self.backup_os_remove
+
+        shutil.rmtree(test_dir)
+
+    def test_delete_temp_folder_success(self):
+        test_dir = tempfile.mkdtemp()
+        ext_env_handler = self.__create_ext_env_handler_and_validate_tmp_folder(test_dir)
+        ext_env_handler.delete_temp_folder()
+        self.assertFalse(os.path.isdir(os.path.join(ext_env_handler.temp_folder)))
+        shutil.rmtree(test_dir)
+
+    def test_delete_temp_folder_when_none_exists(self):
+        ext_env_handler = ExtEnvHandler(self.runtime.logger, self.runtime.env_layer, self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
+        self.assertTrue(ext_env_handler.log_folder is not None)
+        self.assertEqual(ext_env_handler.log_folder, "mockLog")
+        self.assertTrue(ext_env_handler.status_folder is not None)
+        self.assertTrue(ext_env_handler.temp_folder is not None)
+        self.assertEqual(ext_env_handler.temp_folder, "tmp")
+
+        # Reset os.pathexists that was mocked in setup()
+        os.path.exists = self.backup_pathexists
+        # delete temp content
+        ext_env_handler.delete_temp_folder_contents()
+        self.assertFalse(os.path.isdir(os.path.join(ext_env_handler.temp_folder)))
+
+    def test_delete_temp_folder_failure(self):
+        test_dir = tempfile.mkdtemp()
+        ext_env_handler = self.__create_ext_env_handler_and_validate_tmp_folder(test_dir)
+
+        # mock shutil.rmtree()
+        self.backup_shutil_rmtree = shutil.rmtree
+        shutil.rmtree = self.mock_shutil_rmtree
+
+        # delete temp content attempt #1, throws exception
+        self.assertRaises(Exception, lambda: ext_env_handler.delete_temp_folder(raise_if_delete_failed=True))
+        self.assertTrue(os.path.isdir(os.path.join(ext_env_handler.temp_folder)))
+
+        # delete temp content attempt #2, does not throws exception
+        ext_env_handler.delete_temp_folder()
+        self.assertTrue(os.path.isdir(os.path.join(ext_env_handler.temp_folder)))
+
+        # reset shutil.rmtree() mock
+        shutil.rmtree = self.backup_shutil_rmtree
+
+        shutil.rmtree(test_dir)
+
+    def test_get_temp_folder_success(self):
+        test_dir = tempfile.mkdtemp()
+        ext_env_handler = self.__create_ext_env_handler_and_validate_tmp_folder(test_dir)
+
+        # get temp content
+        temp_folder_path = ext_env_handler.get_temp_folder()
+
+        # validate path
+        self.assertEquals(temp_folder_path, ext_env_handler.temp_folder)
+
+        shutil.rmtree(test_dir)
+
+    def test_get_temp_folder_failure(self):
+        ext_env_handler = ExtEnvHandler(self.runtime.logger, self.runtime.env_layer, self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
+        self.assertTrue(ext_env_handler.log_folder is not None)
+        self.assertEqual(ext_env_handler.log_folder, "mockLog")
+        self.assertTrue(ext_env_handler.status_folder is not None)
+        self.assertTrue(ext_env_handler.temp_folder is not None)
+        self.assertEqual(ext_env_handler.temp_folder, "tmp")
+
+        # Reset os.pathexists that was mocked in setup()
+        os.path.exists = self.backup_pathexists
+        # get temp content
+        self.assertRaises(Exception, lambda: ext_env_handler.get_temp_folder())
+
+    def test_log_temp_folder_success(self):
+        test_dir = tempfile.mkdtemp()
+        ext_env_handler = self.__create_ext_env_handler_and_validate_tmp_folder(test_dir)
+        log_file_path = os.path.join(test_dir, 'test.log')
+        file_logger = FileLogger(test_dir, 'test.log')
+        ext_env_handler.logger = Logger(file_logger)
+
+        # log temp content
+        ext_env_handler.log_temp_folder_details()
+        file_logger.close()
+
+        # validate
+        file_read = open(log_file_path, "r")
+        self.assertTrue(file_read is not None)
+        self.assertTrue("Temp folder details: " in file_read.readlines()[1])
+        file_read.close()
+
+        shutil.rmtree(test_dir)
+
+    def test_log_temp_folder_failure(self):
+        test_dir = tempfile.mkdtemp()
+        ext_env_handler = ExtEnvHandler(self.runtime.logger, self.runtime.env_layer, self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
+        log_file_path = os.path.join(test_dir, 'test.log')
+        file_logger = FileLogger(test_dir, 'test.log')
+        ext_env_handler.logger = Logger(file_logger)
+        self.assertTrue(ext_env_handler.log_folder is not None)
+        self.assertEqual(ext_env_handler.log_folder, "mockLog")
+        self.assertTrue(ext_env_handler.status_folder is not None)
+        self.assertTrue(ext_env_handler.temp_folder is not None)
+        self.assertEqual(ext_env_handler.temp_folder, "tmp")
+
+        # Reset os.pathexists that was mocked in setup()
+        os.path.exists = self.backup_pathexists
+        # log temp content
+        ext_env_handler.temp_folder = None
+        ext_env_handler.log_temp_folder_details()
+        file_logger.close()
+
+        # validate
+        file_read = open(log_file_path, "r")
+        self.assertTrue(file_read is not None)
+        self.assertTrue("Temp folder not found" in file_read.readlines()[1])
+        file_read.close()
+
+        shutil.rmtree(test_dir)
+
diff --git a/src/extension/tests/Test_InstallCommandHandler.py b/src/extension/tests/Test_InstallCommandHandler.py
index 1f76760b..d19fe68d 100644
--- a/src/extension/tests/Test_InstallCommandHandler.py
+++ b/src/extension/tests/Test_InstallCommandHandler.py
@@ -30,28 +30,38 @@ def setUp(self):
         VirtualTerminal().print_lowlight("\n----------------- setup test runner -----------------")
         runtime = RuntimeComposer()
         self.logger = runtime.logger
+        self.env_layer = runtime.env_layer
         self.telemetry_writer = runtime.telemetry_writer
         self.logger.telemetry_writer = self.telemetry_writer
         self.json_file_handler = runtime.json_file_handler
         self.get_json_file_content_backup = self.json_file_handler.get_json_file_content
         self.json_file_handler.get_json_file_content = self.mock_get_json_file_content_to_return_none
 
+        # Mock temp folder setup in ExtEnvHandler
+        self.ext_env_handler_get_temp_folder_backup = ExtEnvHandler.get_temp_folder
+        ExtEnvHandler.get_temp_folder = self.mock_get_temp_folder
+
     def tearDown(self):
         VirtualTerminal().print_lowlight("\n----------------- tear down test runner -----------------")
-        # reseting mocks
+        # resetting mocks
         self.json_file_handler.get_json_file_content = self.get_json_file_content_backup
+        # reset temp folder mock from ExtEnvHandler
+        ExtEnvHandler.get_temp_folder = self.ext_env_handler_get_temp_folder_backup
 
     def mock_get_json_file_content_to_return_none(self, file_name, dir_path, raise_if_not_found=False):
         return None
 
+    def mock_get_temp_folder(self):
+        return "testTempFolder"
+
     def test_validate_os_type_is_linux(self):
-        ext_env_handler = ExtEnvHandler(self.json_file_handler)
+        ext_env_handler = ExtEnvHandler(self.logger, self.env_layer, self.json_file_handler)
         install_command_handler = InstallCommandHandler(self.logger, ext_env_handler)
         sys.platform = 'linux'
         self.assertTrue(install_command_handler.validate_os_type())
 
     def test_validate_os_type_not_linux(self):
-        ext_env_handler = ExtEnvHandler(self.json_file_handler)
+        ext_env_handler = ExtEnvHandler(self.logger, self.env_layer, self.json_file_handler)
         install_command_handler = InstallCommandHandler(self.logger, ext_env_handler)
         sys.platform = 'win32'
         self.assertRaises(Exception, install_command_handler.validate_os_type)
@@ -60,7 +70,7 @@ def test_validate_environment(self):
         config_type = 'handlerEnvironment'
 
         # file has no content
-        ext_env_handler = ExtEnvHandler(self.json_file_handler)
+        ext_env_handler = ExtEnvHandler(self.logger, self.env_layer, self.json_file_handler)
         install_command_handler = InstallCommandHandler(self.logger, ext_env_handler)
         self.assertRaises(Exception, install_command_handler.validate_environment)
 
@@ -79,7 +89,7 @@ def test_validate_environment(self):
         # Validating HandlerEnvironment.json file
         # reseting mock to original func def
         self.json_file_handler.get_json_file_content = self.get_json_file_content_backup
-        ext_env_handler = ExtEnvHandler(self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
+        ext_env_handler = ExtEnvHandler(self.logger, self.env_layer, self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
         install_command_handler = InstallCommandHandler(self.logger, ext_env_handler)
         install_command_handler.validate_environment()
 
@@ -98,7 +108,7 @@ def test_execute_action_handler(self):
         sys.platform = 'linux'
         # reseting mock to original func def
         self.json_file_handler.get_json_file_content = self.get_json_file_content_backup
-        ext_env_handler = ExtEnvHandler(self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
+        ext_env_handler = ExtEnvHandler(self.logger, self.env_layer, self.json_file_handler, handler_env_file_path=os.path.join(os.path.pardir, "tests", "helpers"))
         install_command_handler = InstallCommandHandler(self.logger, ext_env_handler)
         self.assertEqual(install_command_handler.execute_handler_action(), Constants.ExitCode.Okay)
 
diff --git a/src/extension/tests/Test_ProcessHandler.py b/src/extension/tests/Test_ProcessHandler.py
index b0148f7c..2bab46a3 100644
--- a/src/extension/tests/Test_ProcessHandler.py
+++ b/src/extension/tests/Test_ProcessHandler.py
@@ -88,6 +88,9 @@ def mock_process_poll_return_None(self):
     def mock_process_poll_return_Not_None(self):
         return 0
 
+    def mock_get_temp_folder(self):
+        return "testTempFolder"
+
     def test_get_public_config_settings(self):
         ext_config_settings_handler = ExtConfigSettingsHandler(self.logger, self.json_file_handler, os.path.join(os.path.pardir, "tests", "helpers"))
         seq_no = "1234"
@@ -100,13 +103,20 @@ def test_get_public_config_settings(self):
         self.assertEqual(public_config_settings.get(Constants.ConfigPublicSettingsFields.patch_mode), "AutomaticByPlatform")
 
     def test_get_env_settings(self):
+        # Mock temp folder setup in ExtEnvHandler
+        ext_env_handler_get_temp_folder_backup = ExtEnvHandler.get_temp_folder
+        ExtEnvHandler.get_temp_folder = self.mock_get_temp_folder
+
         handler_env_file_path = os.path.join(os.path.pardir, "tests", "helpers")
-        ext_env_handler = ExtEnvHandler(self.json_file_handler, handler_env_file_path=handler_env_file_path)
+        ext_env_handler = ExtEnvHandler(self.logger, self.env_layer, self.json_file_handler, handler_env_file_path=handler_env_file_path)
         process_handler = ProcessHandler(self.logger, self.env_layer, self.ext_output_status_handler)
         env_settings = process_handler.get_env_settings(ext_env_handler)
         self.assertTrue(env_settings is not None)
         self.assertEqual(env_settings.get(Constants.EnvSettingsFields.log_folder), "mockLog")
 
+        # reset temp folder mock from ExtEnvHandler
+        ExtEnvHandler.get_temp_folder = ext_env_handler_get_temp_folder_backup
+
     def test_kill_process(self):
         # setting mocks
         is_process_running_backup = ProcessHandler.is_process_running
@@ -148,13 +158,15 @@ def test_start_daemon(self):
         get_python_cmd_backup = ProcessHandler.get_python_cmd
         ProcessHandler.get_python_cmd = self.mock_get_python_cmd
         subprocess_popen_backup = subprocess.Popen
+        ext_env_handler_get_temp_folder_backup = ExtEnvHandler.get_temp_folder
+        ExtEnvHandler.get_temp_folder = self.mock_get_temp_folder
 
         # Initializing config env
         ext_config_settings_handler = ExtConfigSettingsHandler(self.logger, self.json_file_handler, os.path.join(os.path.pardir, "tests", "helpers"))
         seq_no = "1234"
         config_settings = ext_config_settings_handler.read_file(seq_no)
         handler_env_file_path = os.path.join(os.path.pardir, "tests", "helpers")
-        ext_env_handler = ExtEnvHandler(self.json_file_handler, handler_env_file_path=handler_env_file_path)
+        ext_env_handler = ExtEnvHandler(self.logger, self.env_layer, self.json_file_handler, handler_env_file_path=handler_env_file_path)
 
         # process was not launched
         subprocess.Popen = self.mock_subprocess_popen_process_not_launched
@@ -184,6 +196,7 @@ def test_start_daemon(self):
         ProcessHandler.get_python_cmd = get_python_cmd_backup
         subprocess.Popen = subprocess_popen_backup
         process_handler.env_layer.run_command_output = run_command_output_backup
+        ExtEnvHandler.get_temp_folder = ext_env_handler_get_temp_folder_backup
 
 
 if __name__ == '__main__':