diff --git a/supervisor/jobs/const.py b/supervisor/jobs/const.py
index cd06f446b..bb67c6575 100644
--- a/supervisor/jobs/const.py
+++ b/supervisor/jobs/const.py
@@ -20,6 +20,7 @@ class JobCondition(str, Enum):
     HAOS = "haos"
     OS_AGENT = "os_agent"
     HOST_NETWORK = "host_network"
+    SUPERVISOR_UPDATED = "supervisor_updated"
 
 
 class JobExecutionLimit(str, Enum):
diff --git a/supervisor/jobs/decorator.py b/supervisor/jobs/decorator.py
index e8a64e000..9d3b49bbe 100644
--- a/supervisor/jobs/decorator.py
+++ b/supervisor/jobs/decorator.py
@@ -189,6 +189,14 @@ class Job(CoreSysAttributes):
                 f"'{self._method.__qualname__}' blocked from execution, host Network Manager not available"
             )
 
+        if (
+            JobCondition.SUPERVISOR_UPDATED in self.conditions
+            and self.sys_supervisor.need_update
+        ):
+            raise JobConditionException(
+                f"'{self._method.__qualname__}' blocked from execution, supervisor needs to be updated first"
+            )
+
     async def _acquire_exection_limit(self) -> None:
         """Process exection limits."""
         if self.limit not in (
diff --git a/supervisor/misc/tasks.py b/supervisor/misc/tasks.py
index 265ce6efb..1a1ed0be2 100644
--- a/supervisor/misc/tasks.py
+++ b/supervisor/misc/tasks.py
@@ -88,6 +88,7 @@ class Tasks(CoreSysAttributes):
             JobCondition.FREE_SPACE,
             JobCondition.INTERNET_HOST,
             JobCondition.RUNNING,
+            JobCondition.SUPERVISOR_UPDATED,
         ]
     )
     async def _update_addons(self):
@@ -172,7 +173,7 @@ class Tasks(CoreSysAttributes):
         finally:
             self._cache[HASS_WATCHDOG_API] = 0
 
-    @Job(conditions=JobCondition.RUNNING)
+    @Job(conditions=[JobCondition.RUNNING, JobCondition.SUPERVISOR_UPDATED])
     async def _update_cli(self):
         """Check and run update of cli."""
         if not self.sys_plugins.cli.need_update:
@@ -183,7 +184,7 @@ class Tasks(CoreSysAttributes):
         )
         await self.sys_plugins.cli.update()
 
-    @Job(conditions=JobCondition.RUNNING)
+    @Job(conditions=[JobCondition.RUNNING, JobCondition.SUPERVISOR_UPDATED])
     async def _update_dns(self):
         """Check and run update of CoreDNS plugin."""
         if not self.sys_plugins.dns.need_update:
@@ -195,7 +196,7 @@ class Tasks(CoreSysAttributes):
         )
         await self.sys_plugins.dns.update()
 
-    @Job(conditions=JobCondition.RUNNING)
+    @Job(conditions=[JobCondition.RUNNING, JobCondition.SUPERVISOR_UPDATED])
     async def _update_audio(self):
         """Check and run update of PulseAudio plugin."""
         if not self.sys_plugins.audio.need_update:
@@ -207,7 +208,7 @@ class Tasks(CoreSysAttributes):
         )
         await self.sys_plugins.audio.update()
 
-    @Job(conditions=JobCondition.RUNNING)
+    @Job(conditions=[JobCondition.RUNNING, JobCondition.SUPERVISOR_UPDATED])
     async def _update_observer(self):
         """Check and run update of Observer plugin."""
         if not self.sys_plugins.observer.need_update:
@@ -219,7 +220,7 @@ class Tasks(CoreSysAttributes):
         )
         await self.sys_plugins.observer.update()
 
-    @Job(conditions=JobCondition.RUNNING)
+    @Job(conditions=[JobCondition.RUNNING, JobCondition.SUPERVISOR_UPDATED])
     async def _update_multicast(self):
         """Check and run update of multicast."""
         if not self.sys_plugins.multicast.need_update:
diff --git a/tests/jobs/test_job_decorator.py b/tests/jobs/test_job_decorator.py
index 3bb5bc920..e3a94cc5d 100644
--- a/tests/jobs/test_job_decorator.py
+++ b/tests/jobs/test_job_decorator.py
@@ -2,7 +2,7 @@
 # pylint: disable=protected-access,import-error
 import asyncio
 from datetime import timedelta
-from unittest.mock import patch
+from unittest.mock import PropertyMock, patch
 
 import pytest
 
@@ -355,3 +355,28 @@ async def test_exectution_limit_once(coresys: CoreSys, loop: asyncio.BaseEventLo
         await test.execute(0.1)
 
     await run_task
+
+
+async def test_supervisor_updated(coresys: CoreSys):
+    """Test the supervisor updated decorator."""
+
+    class TestClass:
+        """Test class."""
+
+        def __init__(self, coresys: CoreSys):
+            """Initialize the test class."""
+            self.coresys = coresys
+
+        @Job(conditions=JobCondition.SUPERVISOR_UPDATED)
+        async def execute(self) -> bool:
+            """Execute the class method."""
+            return True
+
+    test = TestClass(coresys)
+    assert not coresys.supervisor.need_update
+    assert await test.execute()
+
+    with patch.object(
+        type(coresys.supervisor), "need_update", new=PropertyMock(return_value=True)
+    ):
+        assert not await test.execute()