boumenot: LGTM.
diff --git a/azurelinuxagent/common/errorstate.py b/azurelinuxagent/common/errorstate.py
index 38aaa1f9..052db075 100644
--- a/azurelinuxagent/common/errorstate.py
+++ b/azurelinuxagent/common/errorstate.py
@@ -31,3 +31,15 @@ class ErrorState(object):
             return True
 
         return False
+
+    @property
+    def fail_time(self):
+        if self.timestamp is None:
+            return 'unknown'
+
+        delta = round((datetime.utcnow() - self.timestamp).seconds / 60.0, 2)
+        if delta < 60:
+            return '{0} min'.format(delta)
+
+        delta_hr = round(delta / 60.0, 2)
+        return '{0} hr'.format(delta_hr)
diff --git a/azurelinuxagent/common/event.py b/azurelinuxagent/common/event.py
index 71723d45..4852bb37 100644
--- a/azurelinuxagent/common/event.py
+++ b/azurelinuxagent/common/event.py
@@ -58,6 +58,7 @@ class WALAEventOperation:
     HeartBeat = "HeartBeat"
     HostPlugin = "HostPlugin"
     HostPluginHeartbeat = "HostPluginHeartbeat"
+    HostPluginHeartbeatExtended = "HostPluginHeartbeatExtended"
     HttpErrors = "HttpErrors"
     ImdsHeartbeat = "ImdsHeartbeat"
     Install = "Install"
diff --git a/azurelinuxagent/ga/monitor.py b/azurelinuxagent/ga/monitor.py
index e28d7321..c1215806 100644
--- a/azurelinuxagent/ga/monitor.py
+++ b/azurelinuxagent/ga/monitor.py
@@ -336,6 +336,15 @@ class MonitorHandler(object):
 
                 self.health_service.report_host_plugin_heartbeat(is_healthy)
 
+                if not is_healthy:
+                    add_event(
+                        name=AGENT_NAME,
+                        version=CURRENT_VERSION,
+                        op=WALAEventOperation.HostPluginHeartbeatExtended,
+                        is_success=False,
+                        message='{0} since successful heartbeat'.format(self.host_plugin_errorstate.fail_time),
+                        log_event=False)
+
             except Exception as e:
                 msg = "Exception sending host plugin heartbeat: {0}".format(ustr(e))
                 add_event(
diff --git a/azurelinuxagent/pa/deprovision/default.py b/azurelinuxagent/pa/deprovision/default.py
index 3ed18a5a..b90e0456 100644
--- a/azurelinuxagent/pa/deprovision/default.py
+++ b/azurelinuxagent/pa/deprovision/default.py
@@ -144,44 +144,6 @@ class DeprovisionHandler(object):
         if len(files) > 0:
             actions.append(DeprovisionAction(fileutil.rm_files, files))
 
-    def cloud_init_dirs(self, include_once=True):
-        dirs = [
-            "/var/lib/cloud/instance",
-            "/var/lib/cloud/instances/",
-            "/var/lib/cloud/data"
-        ]
-        if include_once:
-            dirs += [
-                "/var/lib/cloud/scripts/per-once"
-            ]
-        return dirs
-    
-    def cloud_init_files(self, include_once=True, deluser=False):
-        files = []
-        if deluser:
-            files += [
-                "/etc/sudoers.d/90-cloud-init-users"
-            ]
-        if include_once:
-            files += [
-                "/var/lib/cloud/sem/config_scripts_per_once.once"
-            ]
-        return files
-
-    def del_cloud_init(self, warnings, actions,
-            include_once=True, deluser=False):
-        dirs = [d for d in self.cloud_init_dirs(include_once=include_once) \
-                    if os.path.isdir(d)]
-        if len(dirs) > 0:
-            actions.append(DeprovisionAction(fileutil.rm_dirs, dirs))
-
-        files = [f for f in self.cloud_init_files(
-                                    include_once=include_once,
-                                    deluser=deluser) \
-                    if os.path.isfile(f)]
-        if len(files) > 0:
-            actions.append(DeprovisionAction(fileutil.rm_files, files))
-
     def reset_hostname(self, warnings, actions):
         localhost = ["localhost.localdomain"]
         actions.append(DeprovisionAction(self.osutil.set_hostname, 
@@ -203,7 +165,6 @@ class DeprovisionHandler(object):
         if conf.get_delete_root_password():
             self.del_root_password(warnings, actions)
 
-        self.del_cloud_init(warnings, actions, deluser=deluser)
         self.del_dirs(warnings, actions)
         self.del_files(warnings, actions)
         self.del_resolv(warnings, actions)
@@ -217,8 +178,6 @@ class DeprovisionHandler(object):
         warnings = []
         actions = []
 
-        self.del_cloud_init(warnings, actions,
-                    include_once=False, deluser=False)
         self.del_dhcp_lease(warnings, actions)
         self.del_lib_dir_files(warnings, actions)
 

Waagent deprovision should not be messing with the cloudinit directories
Deleting and changing the files under /var/lib/cloud upon deprovision fundamentally breaks how per-once and per-instance works.   You also lose the ability to track instances created from the original image.  At the minimum, there should be a flag to turn this functionality completely off.
diff --git a/tests/common/test_errorstate.py b/tests/common/test_errorstate.py
index 7513fe59..a0a7761d 100644
--- a/tests/common/test_errorstate.py
+++ b/tests/common/test_errorstate.py
@@ -12,6 +12,7 @@ class TestErrorState(unittest.TestCase):
         test_subject = ErrorState(timedelta(seconds=10000))
         self.assertFalse(test_subject.is_triggered())
         self.assertEqual(0, test_subject.count)
+        self.assertEqual('unknown', test_subject.fail_time)
 
     def test_errorstate01(self):
         """
@@ -21,6 +22,7 @@ class TestErrorState(unittest.TestCase):
         test_subject = ErrorState(timedelta(seconds=0))
         self.assertFalse(test_subject.is_triggered())
         self.assertEqual(0, test_subject.count)
+        self.assertEqual('unknown', test_subject.fail_time)
 
     def test_errorstate02(self):
         """
@@ -30,9 +32,9 @@ class TestErrorState(unittest.TestCase):
         test_subject = ErrorState(timedelta(seconds=0))
         test_subject.incr()
 
-
         self.assertTrue(test_subject.is_triggered())
         self.assertEqual(1, test_subject.count)
+        self.assertEqual('0.0 min', test_subject.fail_time)
 
     @patch('azurelinuxagent.common.errorstate.datetime')
     def test_errorstate03(self, mock_time):
@@ -52,6 +54,7 @@ class TestErrorState(unittest.TestCase):
         mock_time.utcnow = Mock(return_value=datetime.utcnow() + timedelta(minutes=30))
         test_subject.incr()
         self.assertTrue(test_subject.is_triggered())
+        self.assertEqual('29.0 min', test_subject.fail_time)
 
     def test_errorstate04(self):
         """
@@ -67,3 +70,35 @@ class TestErrorState(unittest.TestCase):
 
         test_subject.reset()
         self.assertTrue(test_subject.timestamp is None)
+
+    def test_errorstate05(self):
+        """
+        Test the fail_time for various scenarios
+        """
+
+        test_subject = ErrorState(timedelta(minutes=15))
+        self.assertEqual('unknown', test_subject.fail_time)
+
+        test_subject.incr()
+        self.assertEqual('0.0 min', test_subject.fail_time)
+
+        test_subject.timestamp = datetime.utcnow() - timedelta(seconds=60)
+        self.assertEqual('1.0 min', test_subject.fail_time)
+
+        test_subject.timestamp = datetime.utcnow() - timedelta(seconds=73)
+        self.assertEqual('1.22 min', test_subject.fail_time)
+
+        test_subject.timestamp = datetime.utcnow() - timedelta(seconds=120)
+        self.assertEqual('2.0 min', test_subject.fail_time)
+
+        test_subject.timestamp = datetime.utcnow() - timedelta(seconds=60 * 59)
+        self.assertEqual('59.0 min', test_subject.fail_time)
+
+        test_subject.timestamp = datetime.utcnow() - timedelta(seconds=60 * 60)
+        self.assertEqual('1.0 hr', test_subject.fail_time)
+
+        test_subject.timestamp = datetime.utcnow() - timedelta(seconds=60 * 95)
+        self.assertEqual('1.58 hr', test_subject.fail_time)
+
+        test_subject.timestamp = datetime.utcnow() - timedelta(seconds=60 * 60 * 3)
+        self.assertEqual('3.0 hr', test_subject.fail_time)
diff --git a/tests/ga/test_monitor.py b/tests/ga/test_monitor.py
index 5d53b1e7..4bcc1da4 100644
--- a/tests/ga/test_monitor.py
+++ b/tests/ga/test_monitor.py
@@ -201,4 +201,18 @@ class TestMonitor(AgentTestCase):
         monitor_handler.last_host_plugin_heartbeat = datetime.datetime.utcnow() - timedelta(hours=1)
         monitor_handler.send_host_plugin_heartbeat()
         self.assertEqual(1, patch_report_heartbeat.call_count)
+        self.assertEqual(0, args[5].call_count)
+        monitor_handler.stop()
+
+    @patch('azurelinuxagent.common.errorstate.ErrorState.is_triggered', return_value=True)
+    @patch("azurelinuxagent.common.protocol.healthservice.HealthService.report_host_plugin_heartbeat")
+    def test_failed_heartbeat_creates_telemetry(self, patch_report_heartbeat, _, *args):
+        monitor_handler = get_monitor_handler()
+        monitor_handler.init_protocols()
+        monitor_handler.last_host_plugin_heartbeat = datetime.datetime.utcnow() - timedelta(hours=1)
+        monitor_handler.send_host_plugin_heartbeat()
+        self.assertEqual(1, patch_report_heartbeat.call_count)
+        self.assertEqual(1, args[5].call_count)
+        self.assertEqual('HostPluginHeartbeatExtended', args[5].call_args[1]['op'])
+        self.assertEqual(False, args[5].call_args[1]['is_success'])
         monitor_handler.stop()
diff --git a/tests/pa/test_deprovision.py b/tests/pa/test_deprovision.py
index 0b911d91..531ad0a2 100644
--- a/tests/pa/test_deprovision.py
+++ b/tests/pa/test_deprovision.py
@@ -53,68 +53,6 @@ class TestDeprovision(AgentTestCase):
         dh.run(force=True)
         self.assertEqual(2, dh.do_actions.call_count)
 
-    @patch("azurelinuxagent.pa.deprovision.default.DeprovisionHandler.cloud_init_dirs")
-    @patch("azurelinuxagent.pa.deprovision.default.DeprovisionHandler.cloud_init_files")
-    def test_del_cloud_init_without_once(self,
-                mock_files,
-                mock_dirs):
-        deprovision_handler = get_deprovision_handler("","","")
-        deprovision_handler.del_cloud_init([], [],
-                            include_once=False, deluser=False)
-
-        mock_dirs.assert_called_with(include_once=False)
-        mock_files.assert_called_with(include_once=False, deluser=False)
-
-    @patch("signal.signal")
-    @patch("azurelinuxagent.common.protocol.get_protocol_util")
-    @patch("azurelinuxagent.common.osutil.get_osutil")
-    @patch("azurelinuxagent.pa.deprovision.default.DeprovisionHandler.cloud_init_dirs")
-    @patch("azurelinuxagent.pa.deprovision.default.DeprovisionHandler.cloud_init_files")
-    def test_del_cloud_init(self,
-                mock_files,
-                mock_dirs,
-                mock_osutil,
-                mock_util,
-                mock_signal):
-        try:
-            with tempfile.NamedTemporaryFile() as f:
-                warnings = []
-                actions = []
-
-                dirs = [tempfile.mkdtemp()]
-                mock_dirs.return_value = dirs
-
-                files = [f.name]
-                mock_files.return_value = files
-
-                deprovision_handler = get_deprovision_handler("","","")
-                deprovision_handler.del_cloud_init(warnings, actions,
-                        deluser=True)
-
-                mock_dirs.assert_called_with(include_once=True)
-                mock_files.assert_called_with(include_once=True, deluser=True)
-
-                self.assertEqual(len(warnings), 0)
-                self.assertEqual(len(actions), 2)
-                for da in actions:
-                    if da.func == fileutil.rm_dirs:
-                        self.assertEqual(da.args, dirs)
-                    elif da.func == fileutil.rm_files:
-                        self.assertEqual(da.args, files)
-                    else:
-                        self.assertTrue(False)
-
-                try:
-                    for da in actions:
-                        da.invoke()
-                    self.assertEqual(len([d for d in dirs if os.path.isdir(d)]), 0)
-                    self.assertEqual(len([f for f in files if os.path.isfile(f)]), 0)
-                except Exception as e:
-                    self.assertTrue(False, "Exception {0}".format(e))
-        except OSError:
-            # Ignore the error caused by removing the file within the "with"
-            pass
-    
     @distros("ubuntu")
     @patch('azurelinuxagent.common.conf.get_lib_dir')
     def test_del_lib_dir_files(self,

pip install -e .
parse_log_pytest
distro pytest pytest-cov pytest-xdist pytest-mock pytest-asyncio
apt-get update apt-get install -y gcc
3.9
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
coverage==7.8.0
distro==1.9.0
exceptiongroup==1.2.2
execnet==2.1.1
iniconfig==2.1.0
packaging==24.2
pluggy==1.5.0
pytest==8.3.5
pytest-asyncio==0.26.0
pytest-cov==6.0.0
pytest-mock==3.14.0
pytest-xdist==3.6.1
tomli==2.2.1
typing_extensions==4.13.0
-e git+https://github.com/Azure/WALinuxAgent.git@dd3daa66040258a22b994d8f139a0d9c9bab3e6a#egg=WALinuxAgent

name: WALinuxAgent
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=25.0=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - tzdata=2025a=h04d1e81_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - coverage==7.8.0
      - distro==1.9.0
      - exceptiongroup==1.2.2
      - execnet==2.1.1
      - iniconfig==2.1.0
      - packaging==24.2
      - pluggy==1.5.0
      - pytest==8.3.5
      - pytest-asyncio==0.26.0
      - pytest-cov==6.0.0
      - pytest-mock==3.14.0
      - pytest-xdist==3.6.1
      - tomli==2.2.1
      - typing-extensions==4.13.0
prefix: /opt/conda/envs/WALinuxAgent

tests/common/test_errorstate.py::TestErrorState::test_errorstate00 tests/common/test_errorstate.py::TestErrorState::test_errorstate01 tests/common/test_errorstate.py::TestErrorState::test_errorstate02 tests/common/test_errorstate.py::TestErrorState::test_errorstate03 tests/common/test_errorstate.py::TestErrorState::test_errorstate05 tests/ga/test_monitor.py::TestMonitor::test_failed_heartbeat_creates_telemetry
tests/common/test_errorstate.py::TestErrorState::test_errorstate04 tests/ga/test_monitor.py::TestMonitor::test_add_sysinfo tests/ga/test_monitor.py::TestMonitor::test_heartbeat_creates_signal tests/ga/test_monitor.py::TestMonitor::test_heartbeat_timings_no_updates_within_window tests/ga/test_monitor.py::TestMonitor::test_heartbeat_timings_updates_after_window tests/ga/test_monitor.py::TestMonitor::test_heartbeats tests/ga/test_monitor.py::TestMonitor::test_parse_xml_event tests/pa/test_deprovision.py::TestDeprovision::test_confirmation tests/pa/test_deprovision.py::TestDeprovision::test_del_lib_dir_files tests/pa/test_deprovision.py::TestDeprovision::test_deprovision tests/pa/test_deprovision.py::TestDeprovision::test_deprovision_ubuntu
azurelinuxagent/common/errorstate.py azurelinuxagent/pa/deprovision/default.py azurelinuxagent/ga/monitor.py azurelinuxagent/common/event.py
azurelinuxagent/common/errorstate.py azurelinuxagent/pa/deprovision/default.py azurelinuxagent/ga/monitor.py azurelinuxagent/common/event.py