diff --git a/src/core/src/core_logic/ServiceManager.py b/src/core/src/core_logic/ServiceManager.py
index 9c8dde78..c79b4ce0 100644
--- a/src/core/src/core_logic/ServiceManager.py
+++ b/src/core/src/core_logic/ServiceManager.py
@@ -89,8 +89,8 @@ def is_service_enabled(self):
     # endregion
 
     # region - Service Unit Management
-    def create_service_unit_file(self, exec_start, desc, after="network.target", service_type="forking", wanted_by="multi-user.target"):
-        """ Note: Service type defaults to forking because of sh to py process fork """
+    def create_service_unit_file(self, exec_start, desc, after="network.target", service_type="simple", wanted_by="multi-user.target"):
+        """ Note: Service type is simple because the shell wrapper runs Core in the foreground."""
         service_unit_content_template = "\n[Unit]" + \
                                "\nDescription={0}" + \
                                "\nAfter={1}\n" + \
diff --git a/src/core/tests/Test_ServiceManager.py b/src/core/tests/Test_ServiceManager.py
index 0eb406ff..258adc4d 100644
--- a/src/core/tests/Test_ServiceManager.py
+++ b/src/core/tests/Test_ServiceManager.py
@@ -28,6 +28,8 @@ def setUp(self):
         self.service_manager = ServiceManager(self.runtime.env_layer, self.runtime.execution_config, self.runtime.composite_logger, self.runtime.telemetry_writer,ServiceInfo("AutoAssessment", "Auto assessment service", "path"))
         self.service_manager.service_name = "test_service"
         self.mock_systemd_service_unit_path = "/etc/systemd/system/{0}.service"
+        self.written_service_unit_path = None
+        self.written_service_unit_content = None
 
     def tearDown(self):
         self.runtime.stop()
@@ -38,6 +40,8 @@ def mock_run_command_to_set_service_file_permission(self, cmd, no_output=False,
             return 0, "permissions set"
 
     def mock_write_with_retry_valid(self, file_path_or_handle, data, mode='a+'):
+        self.written_service_unit_path = file_path_or_handle
+        self.written_service_unit_content = data
         return
 
     def mock_invoke_systemctl(self, command, description):
@@ -63,6 +67,23 @@ def test_create_service_unit_file(self):
         self.service_manager.env_layer.file_system.write_with_retry = self.mock_write_with_retry_valid
         self.service_manager.create_service_unit_file(exec_start="/bin/bash " + self.service_manager.service_exec_path, desc="Microsoft Azure Linux Patch Extension - Auto Assessment")
 
+        self.assertEqual("/etc/systemd/system/test_service.service", self.written_service_unit_path)
+        self.assertIn("\nType=simple\n", self.written_service_unit_content)
+        self.assertNotIn("Type=forking", self.written_service_unit_content)
+
+    def test_create_and_set_service_idem_generates_simple_unit(self):
+        """ Guards the production creation path, not just the helper, so a future default flip is caught """
+        self.service_manager.env_layer.run_command_output = self.mock_run_command_to_set_service_file_permission
+        self.service_manager.env_layer.file_system.write_with_retry = self.mock_write_with_retry_valid
+        self.service_manager.invoke_systemctl = self.mock_invoke_systemctl
+        self.service_manager.systemctl_daemon_reload = lambda: None
+
+        self.service_manager.create_and_set_service_idem()
+
+        self.assertIn("\nType=simple\n", self.written_service_unit_content)
+        self.assertNotIn("Type=forking", self.written_service_unit_content)
+        self.assertNotIn("Type=notify", self.written_service_unit_content)
+
     def test_start_service(self):
         # Set method calls
         self.service_manager.invoke_systemctl_called = False
diff --git a/src/extension/src/Constants.py b/src/extension/src/Constants.py
index de5bda73..535f2452 100644
--- a/src/extension/src/Constants.py
+++ b/src/extension/src/Constants.py
@@ -59,6 +59,13 @@ def __iter__(self):
     ENABLE_MAX_RUNTIME = 3
     DISABLE_MAX_RUNTIME = 13
 
+    # Bug 28537460: Under Type=simple a hung assessment keeps the unit active forever and every
+    # later timer fire becomes a no-op, so the run must be externally bounded.
+    # Maximum Runtime + grace time must stay under the hourly auto-assessment timer interval
+    # (core AUTO_ASSESSMENT_CRON_INTERVAL = PT1H) so process is killed within 1 hour if still running.
+    AUTO_ASSESSMENT_MAX_RUNTIME_IN_SECS = 3000      # 50m
+    AUTO_ASSESSMENT_KILL_GRACE_IN_SECS = 300        # 5m. Kill process after 55 minutes.
+
     # Telemetry Settings
     # Note: these limits are based on number of characters as confirmed with agent team
     TELEMETRY_MSG_SIZE_LIMIT_IN_CHARS = 3072
diff --git a/src/extension/src/ProcessHandler.py b/src/extension/src/ProcessHandler.py
index a0bc9ee0..2795f4fd 100644
--- a/src/extension/src/ProcessHandler.py
+++ b/src/extension/src/ProcessHandler.py
@@ -119,10 +119,11 @@ def stage_auto_assess_sh_safely(self, core_process_command):
                                   .format(cmd_core_py_path, exec_dir, core_py_path, auto_assess_sh_path, core_process_command))
 
             # generating exec script
+            auto_assess_core_command = core_process_command + " -" + Constants.AUTO_ASSESS_ONLY + " True"
             auto_assess_sh_data = "#!/usr/bin/env bash" +\
                                   "\n# Copyright 2021 Microsoft Corporation." + \
                                   "\ncd \"$(dirname \"$0\")\"" + \
-                                  "\n" + core_process_command + " -" + Constants.AUTO_ASSESS_ONLY + " True"
+                                  "\nexec timeout -s TERM -k " + str(Constants.AUTO_ASSESSMENT_KILL_GRACE_IN_SECS) + " " + str(Constants.AUTO_ASSESSMENT_MAX_RUNTIME_IN_SECS) + " " + auto_assess_core_command
 
             # stage exec script
             if os.path.exists(auto_assess_sh_path):
diff --git a/src/extension/tests/Test_ProcessHandler.py b/src/extension/tests/Test_ProcessHandler.py
index 98b7fb0a..732ac4f6 100644
--- a/src/extension/tests/Test_ProcessHandler.py
+++ b/src/extension/tests/Test_ProcessHandler.py
@@ -45,6 +45,8 @@ def setUp(self):
         self.proc_cmdline_path = os.path.join(self.test_dir, "proc_cmdline")
         self.ext_output_status_handler = ExtOutputStatusHandler(self.logger, self.utility, self.json_file_handler, dir_path)
         self.process = subprocess.Popen(["echo", "Hello World!"], shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        self.written_auto_assess_sh_path = None
+        self.written_auto_assess_sh_content = None
 
     def tearDown(self):
         VirtualTerminal().print_lowlight("\n----------------- tear down test runner -----------------")
@@ -81,6 +83,10 @@ def mock_file_system_open_raises_exception(self, path, mode):
     def mock_run_command_to_set_auto_assess_shell_file_permission(self, cmd, no_output=False, chk_err=False):
         return 0, "permissions set"
 
+    def mock_write_with_retry_valid(self, file_path_or_handle, data, mode='a+'):
+        self.written_auto_assess_sh_path = file_path_or_handle
+        self.written_auto_assess_sh_content = data
+
     def mock_subprocess_popen_process_not_running_after_launch(self, command, shell, stdout, stderr):
         self.process.pid = 1
         self.process.poll = self.mock_process_poll_return_Not_None
@@ -223,6 +229,26 @@ def test_start_daemon(self):
         process_handler.env_layer.run_command_output = run_command_output_backup
         ExtEnvHandler.get_temp_folder = ext_env_handler_get_temp_folder_backup
     
+    def test_auto_assess_sh_is_bounded_by_timeout(self):
+        process_handler = ProcessHandler(self.logger, self.env_layer, self.ext_output_status_handler)
+        write_backup = process_handler.env_layer.file_system.write_with_retry
+        run_backup = process_handler.env_layer.run_command_output
+        process_handler.env_layer.file_system.write_with_retry = self.mock_write_with_retry_valid
+        process_handler.env_layer.run_command_output = self.mock_run_command_to_set_auto_assess_shell_file_permission
+
+        process_handler.stage_auto_assess_sh_safely("/usr/bin/python3 /tmp/MsftLinuxPatchCore.py -sequenceNumber 1")
+
+        self.assertIn(Constants.CORE_AUTO_ASSESS_SH_FILE_NAME, self.written_auto_assess_sh_path)
+        self.assertIn("exec timeout -s TERM -k " + str(Constants.AUTO_ASSESSMENT_KILL_GRACE_IN_SECS)
+                      + " " + str(Constants.AUTO_ASSESSMENT_MAX_RUNTIME_IN_SECS), self.written_auto_assess_sh_content)
+        self.assertIn("-" + Constants.AUTO_ASSESS_ONLY + " True", self.written_auto_assess_sh_content)
+        self.assertEqual(1, self.written_auto_assess_sh_content.count("exec "))
+        script_lines = [line for line in self.written_auto_assess_sh_content.split("\n") if line.strip()]
+        self.assertTrue(script_lines[-1].startswith("exec timeout -s TERM -k "))
+
+        process_handler.env_layer.file_system.write_with_retry = write_backup
+        process_handler.env_layer.run_command_output = run_backup
+
     def test_is_process_patching_operation(self):
         # setting mocks
         backup_file_system_open = self.env_layer.file_system.open