Split execution limit in concurrency and throttle parameters (#6013)

* Split execution limit in concurrency and throttle parameters

Currently the execution limit combines two ortogonal features: Limit
concurrency and throttle execution. This change separates the two
features, allowing for more flexible configuration of job execution.

Ultimately I want to get rid of the old limit parameter. But for ease
of review and migration, I'd like to do this in two steps: First
introduce the new parameters, and map the old limit parameters to the
new parameters. Then, in a second step, remove the old limit parameter
and migrate all users to the new concurrency and throttle parameters
as needed.

* Introduce common lock release method

* Fix THROTTLE_WAIT behavior

The concurrency QUEUE does not really QUEUE throttle limits.

* Add documentation for new concurrency/throttle Job options

* Handle group options for concurrency and throttle separately

* Fix GROUP_THROTTLE_WAIT concurrency setting

We need to use the QUEUE concurrency setting instead of GROUP_QUEUE
for the GROUP_THROTTLE_WAIT execution limit. Otherwise the
test_jobs_decorator.py::test_execution_limit_group_throttle_wait
test deadlocks.

The reason this deadlocks is because GROUP_QUEUE concurrency doesn't
really work because we only can release a group lock if the job is
actually running.

Or put differently, throttling isn't supported with GROUP_*
concurrency options.

* Prevent using any throttling with group concurrency

The group concurrency modes (reject and queue) are not compatible with
any throttling, since we currently can't unlock the group lock when
a job doesn't get started (which is the case when throttling is
applied).

* Fix commit in group rate limit

* Explain the deadlock issue with group locks in code

* Handle locking correctly on throttle limit exceptions

* Introduce pytest for new job decorator combinations

.github/ISSUE_TEMPLATE/bug_report.yml

@@ -8,7 +8,7 @@ body:
 
         If you have a feature or enhancement request, please use the [feature request][fr] section of our [Community Forum][fr].
 
-        [fr]: https://community.home-assistant.io/c/feature-requests
+        [fr]: https://github.com/orgs/home-assistant/discussions
   - type: textarea
     validations:
       required: true
@@ -75,7 +75,7 @@ body:
       description: >
         The System information can be found in [Settings -> System -> Repairs -> (three dot menu) -> System Information](https://my.home-assistant.io/redirect/system_health/).
         Click the copy button at the bottom of the pop-up and paste it here.
-        
+
         [![Open your Home Assistant instance and show health information about your system.](https://my.home-assistant.io/badges/system_health.svg)](https://my.home-assistant.io/redirect/system_health/)
   - type: textarea
     attributes:
@@ -85,7 +85,7 @@ body:
         Supervisor diagnostics can be found in [Settings -> Devices & services](https://my.home-assistant.io/redirect/integrations/).
         Find the card that says `Home Assistant Supervisor`, open it, and select the three dot menu of the Supervisor integration entry
         and select 'Download diagnostics'.
-        
+
         **Please drag-and-drop the downloaded file into the textbox below. Do not copy and paste its contents.**
   - type: textarea
     attributes:

.github/ISSUE_TEMPLATE/task.yml

@@ -0,0 +1,53 @@
+name: Task
+description: For staff only - Create a task
+type: Task
+body:
+  - type: markdown
+    attributes:
+      value: |
+        ## ⚠️ RESTRICTED ACCESS
+
+        **This form is restricted to Open Home Foundation staff and authorized contributors only.**
+
+        If you are a community member wanting to contribute, please:
+        - For bug reports: Use the [bug report form](https://github.com/home-assistant/supervisor/issues/new?template=bug_report.yml)
+        - For feature requests: Submit to [Feature Requests](https://github.com/orgs/home-assistant/discussions)
+
+        ---
+
+        ### For authorized contributors
+
+        Use this form to create tasks for development work, improvements, or other actionable items that need to be tracked.
+  - type: textarea
+    id: description
+    attributes:
+      label: Description
+      description: |
+        Provide a clear and detailed description of the task that needs to be accomplished.
+
+        Be specific about what needs to be done, why it's important, and any constraints or requirements.
+      placeholder: |
+        Describe the task, including:
+        - What needs to be done
+        - Why this task is needed
+        - Expected outcome
+        - Any constraints or requirements
+    validations:
+      required: true
+  - type: textarea
+    id: additional_context
+    attributes:
+      label: Additional context
+      description: |
+        Any additional information, links, research, or context that would be helpful.
+
+        Include links to related issues, research, prototypes, roadmap opportunities etc.
+      placeholder: |
+        - Roadmap opportunity: [link]
+        - Epic: [link]
+        - Feature request: [link]
+        - Technical design documents: [link]
+        - Prototype/mockup: [link]
+        - Dependencies: [links]
+    validations:
+      required: false

.github/copilot-instructions.md

@@ -251,8 +251,8 @@ async def backup_full(self, request: web.Request) -> dict[str, Any]:
 ### Development Commands
 
 ```bash
-# Run tests with coverage
-pytest tests/ --cov=supervisor --cov-report=term-missing
+# Run tests, adjust paths as necessary
+pytest -qsx tests/
 
 # Linting and formatting
 ruff check supervisor/
@@ -275,6 +275,7 @@ Always run the pre-commit hooks at the end of code editing.
 - Use `self.sys_run_in_executor()` for blocking operations
 - Access Docker via `self.sys_docker` not direct Docker API
 - Use constants from `const.py` instead of hardcoding
+- Store types in (per-module) `const.py` (e.g. supervisor/store/const.py)
 
 **❌ Avoid These Patterns**:
 - Direct Docker API usage - use Supervisor's Docker manager

.github/workflows/builder.yml

@@ -131,7 +131,7 @@ jobs:
 
       - name: Install Cosign
         if: needs.init.outputs.publish == 'true'
-        uses: sigstore/cosign-installer@v3.9.1
+        uses: sigstore/cosign-installer@v3.9.2
         with:
           cosign-release: "v2.4.3"
 

.github/workflows/ci.yaml

@@ -346,7 +346,7 @@ jobs:
         with:
           python-version: ${{ needs.prepare.outputs.python-version }}
       - name: Install Cosign
-        uses: sigstore/cosign-installer@v3.9.1
+        uses: sigstore/cosign-installer@v3.9.2
         with:
           cosign-release: "v2.4.3"
       - name: Restore Python virtual environment

.github/workflows/restrict-task-creation.yml

@@ -0,0 +1,58 @@
+name: Restrict task creation
+
+# yamllint disable-line rule:truthy
+on:
+  issues:
+    types: [opened]
+
+jobs:
+  check-authorization:
+    runs-on: ubuntu-latest
+    # Only run if this is a Task issue type (from the issue form)
+    if: github.event.issue.issue_type == 'Task'
+    steps:
+      - name: Check if user is authorized
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const issueAuthor = context.payload.issue.user.login;
+
+            // Check if user is an organization member
+            try {
+              await github.rest.orgs.checkMembershipForUser({
+                org: 'home-assistant',
+                username: issueAuthor
+              });
+              console.log(`✅ ${issueAuthor} is an organization member`);
+              return; // Authorized
+            } catch (error) {
+              console.log(`❌ ${issueAuthor} is not authorized to create Task issues`);
+            }
+
+            // Close the issue with a comment
+            await github.rest.issues.createComment({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              issue_number: context.issue.number,
+              body: `Hi @${issueAuthor}, thank you for your contribution!\n\n` +
+                    `Task issues are restricted to Open Home Foundation staff and authorized contributors.\n\n` +
+                    `If you would like to:\n` +
+                    `- Report a bug: Please use the [bug report form](https://github.com/home-assistant/supervisor/issues/new?template=bug_report.yml)\n` +
+                    `- Request a feature: Please submit to [Feature Requests](https://github.com/orgs/home-assistant/discussions)\n\n` +
+                    `If you believe you should have access to create Task issues, please contact the maintainers.`
+            });
+
+            await github.rest.issues.update({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              issue_number: context.issue.number,
+              state: 'closed'
+            });
+
+            // Add a label to indicate this was auto-closed
+            await github.rest.issues.addLabels({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              issue_number: context.issue.number,
+              labels: ['auto-closed']
+            });

requirements.txt

@@ -1,30 +1,30 @@
 aiodns==3.5.0
-aiohttp==3.12.13
+aiohttp==3.12.15
 atomicwrites-homeassistant==1.4.1
 attrs==25.3.0
 awesomeversion==25.5.0
-blockbuster==1.5.24
+blockbuster==1.5.25
 brotli==1.1.0
 ciso8601==2.3.2
 colorlog==6.9.0
 cpe==1.3.1
 cryptography==45.0.5
-debugpy==1.8.14
+debugpy==1.8.15
 deepmerge==2.0
 dirhash==0.5.0
 docker==7.1.0
 faust-cchardet==2.1.19
-gitpython==3.1.44
+gitpython==3.1.45
 jinja2==3.1.6
 log-rate-limit==1.4.2
-orjson==3.10.18
+orjson==3.11.1
 pulsectl==24.12.0
 pyudev==0.24.3
 PyYAML==6.0.2
 requests==2.32.4
 securetar==2025.2.1
-sentry-sdk==2.32.0
+sentry-sdk==2.34.0
 setuptools==80.9.0
 voluptuous==0.15.2
-dbus-fast==2.44.1
+dbus-fast==2.44.2
 zlib-fast==0.2.1

requirements_tests.txt

@@ -1,6 +1,6 @@
-astroid==3.3.10
-coverage==7.9.2
-mypy==1.16.1
+astroid==3.3.11
+coverage==7.10.1
+mypy==1.17.0
 pre-commit==4.2.0
 pylint==3.3.7
 pytest-aiohttp==1.1.0
@@ -8,9 +8,9 @@ pytest-asyncio==0.25.2
 pytest-cov==6.2.1
 pytest-timeout==2.4.0
 pytest==8.4.1
-ruff==0.12.2
+ruff==0.12.7
 time-machine==2.16.0
-types-docker==7.1.0.20250523
+types-docker==7.1.0.20250705
 types-pyyaml==6.0.12.20250516
 types-requests==2.32.4.20250611
 urllib3==2.5.0

supervisor/addons/manager.py

@@ -266,7 +266,7 @@ class AddonManager(CoreSysAttributes):
         ],
         on_condition=AddonsJobError,
     )
-    async def rebuild(self, slug: str) -> asyncio.Task | None:
+    async def rebuild(self, slug: str, *, force: bool = False) -> asyncio.Task | None:
         """Perform a rebuild of local build add-on.
 
         Returns a Task that completes when addon has state 'started' (see addon.start)
@@ -289,7 +289,7 @@ class AddonManager(CoreSysAttributes):
             raise AddonsError(
                 "Version changed, use Update instead Rebuild", _LOGGER.error
             )
-        if not addon.need_build:
+        if not force and not addon.need_build:
             raise AddonsNotSupportedError(
                 "Can't rebuild a image based add-on", _LOGGER.error
             )

supervisor/api/addons.py

@@ -36,6 +36,7 @@ from ..const import (
     ATTR_DNS,
     ATTR_DOCKER_API,
     ATTR_DOCUMENTATION,
+    ATTR_FORCE,
     ATTR_FULL_ACCESS,
     ATTR_GPIO,
     ATTR_HASSIO_API,
@@ -139,6 +140,8 @@ SCHEMA_SECURITY = vol.Schema({vol.Optional(ATTR_PROTECTED): vol.Boolean()})
 SCHEMA_UNINSTALL = vol.Schema(
     {vol.Optional(ATTR_REMOVE_CONFIG, default=False): vol.Boolean()}
 )
+
+SCHEMA_REBUILD = vol.Schema({vol.Optional(ATTR_FORCE, default=False): vol.Boolean()})
 # pylint: enable=no-value-for-parameter
 
 
@@ -461,7 +464,11 @@ class APIAddons(CoreSysAttributes):
     async def rebuild(self, request: web.Request) -> None:
         """Rebuild local build add-on."""
         addon = self.get_addon_for_request(request)
-        if start_task := await asyncio.shield(self.sys_addons.rebuild(addon.slug)):
+        body: dict[str, Any] = await api_validate(SCHEMA_REBUILD, request)
+
+        if start_task := await asyncio.shield(
+            self.sys_addons.rebuild(addon.slug, force=body[ATTR_FORCE])
+        ):
             await start_task
 
     @api_process

supervisor/api/auth.py

@@ -92,13 +92,18 @@ class APIAuth(CoreSysAttributes):
         # Json
         if request.headers.get(CONTENT_TYPE) == CONTENT_TYPE_JSON:
             data = await request.json(loads=json_loads)
-            return await self._process_dict(request, addon, data)
+            if not await self._process_dict(request, addon, data):
+                raise HTTPUnauthorized()
+            return True
 
         # URL encoded
         if request.headers.get(CONTENT_TYPE) == CONTENT_TYPE_URL:
             data = await request.post()
-            return await self._process_dict(request, addon, data)
+            if not await self._process_dict(request, addon, data):
+                raise HTTPUnauthorized()
+            return True
 
+        # Advertise Basic authentication by default
         raise HTTPUnauthorized(headers=REALM_HEADER)
 
     @api_process

supervisor/api/docker.py

@@ -6,6 +6,8 @@ from typing import Any
 from aiohttp import web
 import voluptuous as vol
 
+from supervisor.resolution.const import ContextType, IssueType, SuggestionType
+
 from ..const import (
     ATTR_ENABLE_IPV6,
     ATTR_HOSTNAME,
@@ -32,7 +34,7 @@ SCHEMA_DOCKER_REGISTRY = vol.Schema(
 )
 
 # pylint: disable=no-value-for-parameter
-SCHEMA_OPTIONS = vol.Schema({vol.Optional(ATTR_ENABLE_IPV6): vol.Boolean()})
+SCHEMA_OPTIONS = vol.Schema({vol.Optional(ATTR_ENABLE_IPV6): vol.Maybe(vol.Boolean())})
 
 
 class APIDocker(CoreSysAttributes):
@@ -59,8 +61,17 @@ class APIDocker(CoreSysAttributes):
         """Set docker options."""
         body = await api_validate(SCHEMA_OPTIONS, request)
 
-        if ATTR_ENABLE_IPV6 in body:
+        if (
+            ATTR_ENABLE_IPV6 in body
+            and self.sys_docker.config.enable_ipv6 != body[ATTR_ENABLE_IPV6]
+        ):
             self.sys_docker.config.enable_ipv6 = body[ATTR_ENABLE_IPV6]
+            _LOGGER.info("Host system reboot required to apply new IPv6 configuration")
+            self.sys_resolution.create_issue(
+                IssueType.REBOOT_REQUIRED,
+                ContextType.SYSTEM,
+                suggestions=[SuggestionType.EXECUTE_REBOOT],
+            )
 
         await self.sys_docker.config.save_data()
 

supervisor/api/ingress.py

@@ -309,9 +309,9 @@ class APIIngress(CoreSysAttributes):
 
 def _init_header(
     request: web.Request, addon: Addon, session_data: IngressSessionData | None
-) -> CIMultiDict | dict[str, str]:
+) -> CIMultiDict[str]:
     """Create initial header."""
-    headers = {}
+    headers = CIMultiDict[str]()
 
     if session_data is not None:
         headers[HEADER_REMOTE_USER_ID] = session_data.user.id
@@ -337,7 +337,7 @@ def _init_header(
             istr(HEADER_REMOTE_USER_DISPLAY_NAME),
         ):
             continue
-        headers[name] = value
+        headers.add(name, value)
 
     # Update X-Forwarded-For
     if request.transport:
@@ -348,9 +348,9 @@ def _init_header(
     return headers
 
 
-def _response_header(response: aiohttp.ClientResponse) -> dict[str, str]:
+def _response_header(response: aiohttp.ClientResponse) -> CIMultiDict[str]:
     """Create response header."""
-    headers = {}
+    headers = CIMultiDict[str]()
 
     for name, value in response.headers.items():
         if name in (
@@ -360,7 +360,7 @@ def _response_header(response: aiohttp.ClientResponse) -> dict[str, str]:
             hdrs.CONTENT_ENCODING,
         ):
             continue
-        headers[name] = value
+        headers.add(name, value)
 
     return headers
 

supervisor/backups/backup.py

@@ -931,5 +931,5 @@ class Backup(JobGroup):
         Return a coroutine.
         """
         return self.sys_store.update_repositories(
-            self.repositories, add_with_errors=True, replace=replace
+            set(self.repositories), issue_on_error=True, replace=replace
         )

supervisor/const.py

@@ -188,6 +188,7 @@ ATTR_FEATURES = "features"
 ATTR_FILENAME = "filename"
 ATTR_FLAGS = "flags"
 ATTR_FOLDERS = "folders"
+ATTR_FORCE = "force"
 ATTR_FORCE_SECURITY = "force_security"
 ATTR_FREQUENCY = "frequency"
 ATTR_FULL_ACCESS = "full_access"

supervisor/coresys.py

@@ -124,7 +124,10 @@ class CoreSys:
 
         resolver: aiohttp.abc.AbstractResolver
         try:
-            resolver = aiohttp.AsyncResolver(loop=self.loop)
+            # Use "unused" kwargs to force dedicated resolver instance. Otherwise
+            # aiodns won't reload /etc/resolv.conf which we need to make our connection
+            # check work in all cases.
+            resolver = aiohttp.AsyncResolver(loop=self.loop, timeout=None)
             # pylint: disable=protected-access
             _LOGGER.debug(
                 "Initializing ClientSession with AsyncResolver. Using nameservers %s",

supervisor/dbus/udisks2/data.py

@@ -28,6 +28,8 @@ class DeviceSpecificationDataType(TypedDict, total=False):
     path: str
     label: str
     uuid: str
+    partuuid: str
+    partlabel: str
 
 
 @dataclass(slots=True)
@@ -40,6 +42,8 @@ class DeviceSpecification:
     path: Path | None = None
     label: str | None = None
     uuid: str | None = None
+    partuuid: str | None = None
+    partlabel: str | None = None
 
     @staticmethod
     def from_dict(data: DeviceSpecificationDataType) -> "DeviceSpecification":
@@ -48,6 +52,8 @@ class DeviceSpecification:
             path=Path(data["path"]) if "path" in data else None,
             label=data.get("label"),
             uuid=data.get("uuid"),
+            partuuid=data.get("partuuid"),
+            partlabel=data.get("partlabel"),
         )
 
     def to_dict(self) -> dict[str, Variant]:
@@ -56,6 +62,8 @@ class DeviceSpecification:
             "path": Variant("s", self.path.as_posix()) if self.path else None,
             "label": _optional_variant("s", self.label),
             "uuid": _optional_variant("s", self.uuid),
+            "partuuid": _optional_variant("s", self.partuuid),
+            "partlabel": _optional_variant("s", self.partlabel),
         }
         return {k: v for k, v in data.items() if v}
 

supervisor/docker/manager.py

@@ -95,12 +95,12 @@ class DockerConfig(FileConfiguration):
         super().__init__(FILE_HASSIO_DOCKER, SCHEMA_DOCKER_CONFIG)
 
     @property
-    def enable_ipv6(self) -> bool:
+    def enable_ipv6(self) -> bool | None:
         """Return IPv6 configuration for docker network."""
-        return self._data.get(ATTR_ENABLE_IPV6, False)
+        return self._data.get(ATTR_ENABLE_IPV6, None)
 
     @enable_ipv6.setter
-    def enable_ipv6(self, value: bool) -> None:
+    def enable_ipv6(self, value: bool | None) -> None:
         """Set IPv6 configuration for docker network."""
         self._data[ATTR_ENABLE_IPV6] = value
 

supervisor/docker/network.py

@@ -47,6 +47,8 @@ DOCKER_NETWORK_PARAMS = {
     "options": {"com.docker.network.bridge.name": DOCKER_NETWORK},
 }
 
+DOCKER_ENABLE_IPV6_DEFAULT = True
+
 
 class DockerNetwork:
     """Internal Supervisor Network.
@@ -59,7 +61,7 @@ class DockerNetwork:
         self.docker: docker.DockerClient = docker_client
         self._network: docker.models.networks.Network
 
-    async def post_init(self, enable_ipv6: bool = False) -> Self:
+    async def post_init(self, enable_ipv6: bool | None = None) -> Self:
         """Post init actions that must be done in event loop."""
         self._network = await asyncio.get_running_loop().run_in_executor(
             None, self._get_network, enable_ipv6
@@ -111,16 +113,24 @@ class DockerNetwork:
         """Return observer of the network."""
         return DOCKER_IPV4_NETWORK_MASK[6]
 
-    def _get_network(self, enable_ipv6: bool = False) -> docker.models.networks.Network:
+    def _get_network(
+        self, enable_ipv6: bool | None = None
+    ) -> docker.models.networks.Network:
         """Get supervisor network."""
         try:
             if network := self.docker.networks.get(DOCKER_NETWORK):
-                if network.attrs.get(DOCKER_ENABLEIPV6) == enable_ipv6:
+                current_ipv6 = network.attrs.get(DOCKER_ENABLEIPV6, False)
+                # If the network exists and we don't have an explicit setting,
+                # simply stick with what we have.
+                if enable_ipv6 is None or current_ipv6 == enable_ipv6:
                     return network
+
+                # We have an explicit setting which differs from the current state.
                 _LOGGER.info(
                     "Migrating Supervisor network to %s",
                     "IPv4/IPv6 Dual-Stack" if enable_ipv6 else "IPv4-Only",
                 )
+
                 if (containers := network.containers) and (
                     containers_all := all(
                         container.name in (OBSERVER_DOCKER_NAME, SUPERVISOR_DOCKER_NAME)
@@ -134,6 +144,7 @@ class DockerNetwork:
                             requests.RequestException,
                         ):
                             network.disconnect(container, force=True)
+
                 if not containers or containers_all:
                     try:
                         network.remove()
@@ -151,7 +162,9 @@ class DockerNetwork:
             _LOGGER.info("Can't find Supervisor network, creating a new network")
 
         network_params = DOCKER_NETWORK_PARAMS.copy()
-        network_params[ATTR_ENABLE_IPV6] = enable_ipv6
+        network_params[ATTR_ENABLE_IPV6] = (
+            DOCKER_ENABLE_IPV6_DEFAULT if enable_ipv6 is None else enable_ipv6
+        )
 
         try:
             self._network = self.docker.networks.create(**network_params)  # type: ignore

supervisor/host/network.py

@@ -8,11 +8,11 @@ from typing import Any
 from ..const import ATTR_HOST_INTERNET
 from ..coresys import CoreSys, CoreSysAttributes
 from ..dbus.const import (
+    DBUS_ATTR_CONFIGURATION,
     DBUS_ATTR_CONNECTION_ENABLED,
     DBUS_ATTR_CONNECTIVITY,
-    DBUS_ATTR_PRIMARY_CONNECTION,
+    DBUS_IFACE_DNS,
     DBUS_IFACE_NM,
-    DBUS_OBJECT_BASE,
     DBUS_SIGNAL_NM_CONNECTION_ACTIVE_CHANGED,
     ConnectionStateType,
     ConnectivityState,
@@ -46,6 +46,8 @@ class NetworkManager(CoreSysAttributes):
         """Initialize system center handling."""
         self.coresys: CoreSys = coresys
         self._connectivity: bool | None = None
+        # No event need on initial change (NetworkManager initializes with empty list)
+        self._dns_configuration: list = []
 
     @property
     def connectivity(self) -> bool | None:
@@ -142,6 +144,10 @@ class NetworkManager(CoreSysAttributes):
             "properties_changed", self._check_connectivity_changed
         )
 
+        self.sys_dbus.network.dns.dbus.properties.on(
+            "properties_changed", self._check_dns_changed
+        )
+
     async def _check_connectivity_changed(
         self, interface: str, changed: dict[str, Any], invalidated: list[str]
     ):
@@ -152,16 +158,6 @@ class NetworkManager(CoreSysAttributes):
         connectivity_check: bool | None = changed.get(DBUS_ATTR_CONNECTION_ENABLED)
         connectivity: int | None = changed.get(DBUS_ATTR_CONNECTIVITY)
 
-        # This potentially updated the DNS configuration. Make sure the DNS plug-in
-        # picks up the latest settings.
-        if (
-            DBUS_ATTR_PRIMARY_CONNECTION in changed
-            and changed[DBUS_ATTR_PRIMARY_CONNECTION]
-            and changed[DBUS_ATTR_PRIMARY_CONNECTION] != DBUS_OBJECT_BASE
-            and await self.sys_plugins.dns.is_running()
-        ):
-            await self.sys_plugins.dns.restart()
-
         if (
             connectivity_check is True
             or DBUS_ATTR_CONNECTION_ENABLED in invalidated
@@ -175,6 +171,20 @@ class NetworkManager(CoreSysAttributes):
         elif connectivity is not None:
             self.connectivity = connectivity == ConnectivityState.CONNECTIVITY_FULL
 
+    async def _check_dns_changed(
+        self, interface: str, changed: dict[str, Any], invalidated: list[str]
+    ):
+        """Check if DNS properties have changed."""
+        if interface != DBUS_IFACE_DNS:
+            return
+
+        if (
+            DBUS_ATTR_CONFIGURATION in changed
+            and self._dns_configuration != changed[DBUS_ATTR_CONFIGURATION]
+        ):
+            self._dns_configuration = changed[DBUS_ATTR_CONFIGURATION]
+            self.sys_plugins.dns.notify_locals_changed()
+
     async def update(self, *, force_connectivity_check: bool = False):
         """Update properties over dbus."""
         _LOGGER.info("Updating local network information")

supervisor/jobs/const.py

@@ -34,8 +34,60 @@ class JobCondition(StrEnum):
     SUPERVISOR_UPDATED = "supervisor_updated"
 
 
+class JobConcurrency(StrEnum):
+    """Job concurrency control.
+
+    Controls how many instances of a job can run simultaneously.
+
+    Individual Concurrency (applies to each method separately):
+    - REJECT: Fail immediately if another instance is already running
+    - QUEUE: Wait for the current instance to finish, then run
+
+    Group Concurrency (applies across all methods on a JobGroup):
+    - GROUP_REJECT: Fail if ANY job is running on the JobGroup
+    - GROUP_QUEUE: Wait for ANY running job on the JobGroup to finish
+
+    JobGroup Behavior:
+    - All methods on the same JobGroup instance share a single lock
+    - Methods can call other methods on the same group without deadlock
+    - Uses the JobGroup.group_name for coordination
+    - Requires the class to inherit from JobGroup
+    """
+
+    REJECT = "reject"  # Fail if already running (was ONCE)
+    QUEUE = "queue"  # Wait if already running (was SINGLE_WAIT)
+    GROUP_REJECT = "group_reject"  # Was GROUP_ONCE
+    GROUP_QUEUE = "group_queue"  # Was GROUP_WAIT
+
+
+class JobThrottle(StrEnum):
+    """Job throttling control.
+
+    Controls how frequently jobs can be executed.
+
+    Individual Throttling (each method has its own throttle state):
+    - THROTTLE: Skip execution if called within throttle_period
+    - RATE_LIMIT: Allow up to throttle_max_calls within throttle_period, then fail
+
+    Group Throttling (all methods on a JobGroup share throttle state):
+    - GROUP_THROTTLE: Skip if ANY method was called within throttle_period
+    - GROUP_RATE_LIMIT: Allow up to throttle_max_calls total across ALL methods
+
+    JobGroup Behavior:
+    - All methods on the same JobGroup instance share throttle counters/timers
+    - Uses the JobGroup.group_name as the key for tracking state
+    - If one method is throttled, other methods may also be throttled
+    - Requires the class to inherit from JobGroup
+    """
+
+    THROTTLE = "throttle"  # Skip if called too frequently
+    RATE_LIMIT = "rate_limit"  # Rate limiting with max calls per period
+    GROUP_THROTTLE = "group_throttle"  # Group version of THROTTLE
+    GROUP_RATE_LIMIT = "group_rate_limit"  # Group version of RATE_LIMIT
+
+
 class JobExecutionLimit(StrEnum):
-    """Job Execution limits."""
+    """Job Execution limits - DEPRECATED: Use JobConcurrency and JobThrottle instead."""
 
     ONCE = "once"
     SINGLE_WAIT = "single_wait"

supervisor/jobs/decorator.py

@@ -20,7 +20,7 @@ from ..host.const import HostFeature
 from ..resolution.const import MINIMUM_FREE_SPACE_THRESHOLD, ContextType, IssueType
 from ..utils.sentry import async_capture_exception
 from . import SupervisorJob
-from .const import JobCondition, JobExecutionLimit
+from .const import JobConcurrency, JobCondition, JobExecutionLimit, JobThrottle
 from .job_group import JobGroup
 
 _LOGGER: logging.Logger = logging.getLogger(__package__)
@@ -36,13 +36,16 @@ class Job(CoreSysAttributes):
         conditions: list[JobCondition] | None = None,
         cleanup: bool = True,
         on_condition: type[JobException] | None = None,
-        limit: JobExecutionLimit | None = None,
+        concurrency: JobConcurrency | None = None,
+        throttle: JobThrottle | None = None,
         throttle_period: timedelta
         | Callable[[CoreSys, datetime, list[datetime] | None], timedelta]
         | None = None,
         throttle_max_calls: int | None = None,
         internal: bool = False,
-    ):
+        # Backward compatibility - DEPRECATED
+        limit: JobExecutionLimit | None = None,
+    ):  # pylint: disable=too-many-positional-arguments
         """Initialize the Job decorator.
 
         Args:
@@ -50,13 +53,15 @@ class Job(CoreSysAttributes):
             conditions (list[JobCondition] | None): List of conditions that must be met before the job runs.
             cleanup (bool): Whether to clean up the job after execution. Defaults to True. If set to False, the job will remain accessible through the Supervisor API until the next restart.
             on_condition (type[JobException] | None): Exception type to raise if a job condition fails. If None, logs the failure.
-            limit (JobExecutionLimit | None): Execution limit policy for the job (e.g., throttle, once, group-based).
-            throttle_period (timedelta | Callable | None): Throttle period as a timedelta or a callable returning a timedelta (for rate-limited jobs).
+            concurrency (JobConcurrency | None): Concurrency control policy (e.g., reject, queue, group-based).
+            throttle (JobThrottle | None): Throttling policy (e.g., throttle, rate_limit, group-based).
+            throttle_period (timedelta | Callable | None): Throttle period as a timedelta or a callable returning a timedelta (for throttled jobs).
             throttle_max_calls (int | None): Maximum number of calls allowed within the throttle period (for rate-limited jobs).
             internal (bool): Whether the job is internal (not exposed through the Supervisor API). Defaults to False.
+            limit (JobExecutionLimit | None): DEPRECATED - Use concurrency and throttle instead.
 
         Raises:
-            RuntimeError: If job name is not unique, or required throttle parameters are missing for the selected limit.
+            RuntimeError: If job name is not unique, or required throttle parameters are missing for the selected throttle policy.
 
         """
         if name in _JOB_NAMES:
@@ -67,7 +72,6 @@ class Job(CoreSysAttributes):
         self.conditions = conditions
         self.cleanup = cleanup
         self.on_condition = on_condition
-        self.limit = limit
         self._throttle_period = throttle_period
         self._throttle_max_calls = throttle_max_calls
         self._lock: asyncio.Semaphore | None = None
@@ -75,34 +79,91 @@ class Job(CoreSysAttributes):
         self._rate_limited_calls: dict[str | None, list[datetime]] | None = None
         self._internal = internal
 
+        # Handle backward compatibility with limit parameter
+        if limit is not None:
+            if concurrency is not None or throttle is not None:
+                raise RuntimeError(
+                    f"Job {name} cannot specify both 'limit' (deprecated) and 'concurrency'/'throttle' parameters!"
+                )
+            # Map old limit values to new parameters
+            concurrency, throttle = self._map_limit_to_new_params(limit)
+
+        self.concurrency = concurrency
+        self.throttle = throttle
+
         # Validate Options
+        self._validate_parameters()
+
+    def _map_limit_to_new_params(
+        self, limit: JobExecutionLimit
+    ) -> tuple[JobConcurrency | None, JobThrottle | None]:
+        """Map old limit parameter to new concurrency and throttle parameters."""
+        mapping = {
+            JobExecutionLimit.ONCE: (JobConcurrency.REJECT, None),
+            JobExecutionLimit.SINGLE_WAIT: (JobConcurrency.QUEUE, None),
+            JobExecutionLimit.THROTTLE: (None, JobThrottle.THROTTLE),
+            JobExecutionLimit.THROTTLE_WAIT: (
+                JobConcurrency.QUEUE,
+                JobThrottle.THROTTLE,
+            ),
+            JobExecutionLimit.THROTTLE_RATE_LIMIT: (None, JobThrottle.RATE_LIMIT),
+            JobExecutionLimit.GROUP_ONCE: (JobConcurrency.GROUP_REJECT, None),
+            JobExecutionLimit.GROUP_WAIT: (JobConcurrency.GROUP_QUEUE, None),
+            JobExecutionLimit.GROUP_THROTTLE: (None, JobThrottle.GROUP_THROTTLE),
+            JobExecutionLimit.GROUP_THROTTLE_WAIT: (
+                # Seems a bit counter intuitive, but GROUP_QUEUE deadlocks
+                # tests/jobs/test_job_decorator.py::test_execution_limit_group_throttle_wait
+                # The reason this deadlocks is because when using GROUP_QUEUE and the
+                # throttle limit is hit, the group lock is trying to be unlocked outside
+                # of the job context. The current implementation doesn't allow to unlock
+                # the group lock when the job is not running.
+                JobConcurrency.QUEUE,
+                JobThrottle.GROUP_THROTTLE,
+            ),
+            JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT: (
+                None,
+                JobThrottle.GROUP_RATE_LIMIT,
+            ),
+        }
+        return mapping.get(limit, (None, None))
+
+    def _validate_parameters(self) -> None:
+        """Validate job parameters."""
+        # Validate throttle parameters
         if (
-            self.limit
+            self.throttle
             in (
-                JobExecutionLimit.THROTTLE,
-                JobExecutionLimit.THROTTLE_WAIT,
-                JobExecutionLimit.THROTTLE_RATE_LIMIT,
-                JobExecutionLimit.GROUP_THROTTLE,
-                JobExecutionLimit.GROUP_THROTTLE_WAIT,
-                JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT,
+                JobThrottle.THROTTLE,
+                JobThrottle.GROUP_THROTTLE,
+                JobThrottle.RATE_LIMIT,
+                JobThrottle.GROUP_RATE_LIMIT,
             )
             and self._throttle_period is None
         ):
             raise RuntimeError(
-                f"Job {name} is using execution limit {limit} without a throttle period!"
+                f"Job {self.name} is using throttle {self.throttle} without a throttle period!"
             )
 
-        if self.limit in (
-            JobExecutionLimit.THROTTLE_RATE_LIMIT,
-            JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT,
+        if self.throttle in (
+            JobThrottle.RATE_LIMIT,
+            JobThrottle.GROUP_RATE_LIMIT,
         ):
             if self._throttle_max_calls is None:
                 raise RuntimeError(
-                    f"Job {name} is using execution limit {limit} without throttle max calls!"
+                    f"Job {self.name} is using throttle {self.throttle} without throttle max calls!"
                 )
-
             self._rate_limited_calls = {}
 
+        if self.throttle is not None and self.concurrency in (
+            JobConcurrency.GROUP_REJECT,
+            JobConcurrency.GROUP_QUEUE,
+        ):
+            # We cannot release group locks when Job is not running (e.g. throttled)
+            # which makes these combinations impossible to use currently.
+            raise RuntimeError(
+                f"Job {self.name} is using throttling ({self.throttle}) with group concurrency ({self.concurrency}), which is not allowed!"
+            )
+
     @property
     def throttle_max_calls(self) -> int:
         """Return max calls for throttle."""
@@ -131,7 +192,7 @@ class Job(CoreSysAttributes):
         """Return rate limited calls if used."""
         if self._rate_limited_calls is None:
             raise RuntimeError(
-                f"Rate limited calls not available for limit type {self.limit}"
+                "Rate limited calls not available for this throttle type"
             )
 
         return self._rate_limited_calls.get(group_name, [])
@@ -142,7 +203,7 @@ class Job(CoreSysAttributes):
         """Add a rate limited call to list if used."""
         if self._rate_limited_calls is None:
             raise RuntimeError(
-                f"Rate limited calls not available for limit type {self.limit}"
+                "Rate limited calls not available for this throttle type"
             )
 
         if group_name in self._rate_limited_calls:
@@ -156,7 +217,7 @@ class Job(CoreSysAttributes):
         """Set rate limited calls if used."""
         if self._rate_limited_calls is None:
             raise RuntimeError(
-                f"Rate limited calls not available for limit type {self.limit}"
+                "Rate limited calls not available for this throttle type"
             )
 
         self._rate_limited_calls[group_name] = value
@@ -193,16 +254,24 @@ class Job(CoreSysAttributes):
             if obj.acquire and obj.release:  # type: ignore
                 job_group = cast(JobGroup, obj)
 
-        if not job_group and self.limit in (
-            JobExecutionLimit.GROUP_ONCE,
-            JobExecutionLimit.GROUP_WAIT,
-            JobExecutionLimit.GROUP_THROTTLE,
-            JobExecutionLimit.GROUP_THROTTLE_WAIT,
-            JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT,
-        ):
-            raise RuntimeError(
-                f"Job on {self.name} need to be a JobGroup to use group based limits!"
-            ) from None
+        # Check for group-based parameters
+        if not job_group:
+            if self.concurrency in (
+                JobConcurrency.GROUP_REJECT,
+                JobConcurrency.GROUP_QUEUE,
+            ):
+                raise RuntimeError(
+                    f"Job {self.name} uses group concurrency ({self.concurrency}) but is not on a JobGroup! "
+                    f"The class must inherit from JobGroup to use GROUP_REJECT or GROUP_QUEUE."
+                ) from None
+            if self.throttle in (
+                JobThrottle.GROUP_THROTTLE,
+                JobThrottle.GROUP_RATE_LIMIT,
+            ):
+                raise RuntimeError(
+                    f"Job {self.name} uses group throttling ({self.throttle}) but is not on a JobGroup! "
+                    f"The class must inherit from JobGroup to use GROUP_THROTTLE or GROUP_RATE_LIMIT."
+                ) from None
 
         return job_group
 
@@ -255,71 +324,15 @@ class Job(CoreSysAttributes):
                     except JobConditionException as err:
                         return self._handle_job_condition_exception(err)
 
-                # Handle exection limits
-                if self.limit in (
-                    JobExecutionLimit.SINGLE_WAIT,
-                    JobExecutionLimit.ONCE,
-                ):
-                    await self._acquire_exection_limit()
-                elif self.limit in (
-                    JobExecutionLimit.GROUP_ONCE,
-                    JobExecutionLimit.GROUP_WAIT,
-                ):
-                    try:
-                        await cast(JobGroup, job_group).acquire(
-                            job, self.limit == JobExecutionLimit.GROUP_WAIT
-                        )
-                    except JobGroupExecutionLimitExceeded as err:
-                        if self.on_condition:
-                            raise self.on_condition(str(err)) from err
-                        raise err
-                elif self.limit in (
-                    JobExecutionLimit.THROTTLE,
-                    JobExecutionLimit.GROUP_THROTTLE,
-                ):
-                    time_since_last_call = datetime.now() - self.last_call(group_name)
-                    if time_since_last_call < self.throttle_period(group_name):
-                        return
-                elif self.limit in (
-                    JobExecutionLimit.THROTTLE_WAIT,
-                    JobExecutionLimit.GROUP_THROTTLE_WAIT,
-                ):
-                    await self._acquire_exection_limit()
-                    time_since_last_call = datetime.now() - self.last_call(group_name)
-                    if time_since_last_call < self.throttle_period(group_name):
-                        self._release_exception_limits()
-                        return
-                elif self.limit in (
-                    JobExecutionLimit.THROTTLE_RATE_LIMIT,
-                    JobExecutionLimit.GROUP_THROTTLE_RATE_LIMIT,
-                ):
-                    # Only reprocess array when necessary (at limit)
-                    if (
-                        len(self.rate_limited_calls(group_name))
-                        >= self.throttle_max_calls
-                    ):
-                        self.set_rate_limited_calls(
-                            [
-                                call
-                                for call in self.rate_limited_calls(group_name)
-                                if call
-                                > datetime.now() - self.throttle_period(group_name)
-                            ],
-                            group_name,
-                        )
-
-                    if (
-                        len(self.rate_limited_calls(group_name))
-                        >= self.throttle_max_calls
-                    ):
-                        on_condition = (
-                            JobException
-                            if self.on_condition is None
-                            else self.on_condition
-                        )
-                        raise on_condition(
-                            f"Rate limit exceeded, more than {self.throttle_max_calls} calls in {self.throttle_period(group_name)}",
-                        )
+                # Handle execution limits
+                await self._handle_concurrency_control(job_group, job)
+                try:
+                    if not await self._handle_throttling(group_name):
+                        self._release_concurrency_control(job_group)
+                        return  # Job was throttled, exit early
+                except Exception:
+                    self._release_concurrency_control(job_group)
+                    raise
 
                 # Execute Job
                 with job.start():
@@ -345,12 +358,7 @@ class Job(CoreSysAttributes):
                         await async_capture_exception(err)
                         raise JobException() from err
                     finally:
-                        self._release_exception_limits()
-                        if job_group and self.limit in (
-                            JobExecutionLimit.GROUP_ONCE,
-                            JobExecutionLimit.GROUP_WAIT,
-                        ):
-                            job_group.release()
+                        self._release_concurrency_control(job_group)
 
             # Jobs that weren't started are always cleaned up. Also clean up done jobs if required
             finally:
@@ -492,31 +500,75 @@ class Job(CoreSysAttributes):
                 f"'{method_name}' blocked from execution, mounting not supported on system"
             )
 
-    async def _acquire_exection_limit(self) -> None:
-        """Process exection limits."""
-        if self.limit not in (
-            JobExecutionLimit.SINGLE_WAIT,
-            JobExecutionLimit.ONCE,
-            JobExecutionLimit.THROTTLE_WAIT,
-            JobExecutionLimit.GROUP_THROTTLE_WAIT,
+    def _release_concurrency_control(self, job_group: JobGroup | None) -> None:
+        """Release concurrency control locks."""
+        if self.concurrency == JobConcurrency.REJECT:
+            if self.lock.locked():
+                self.lock.release()
+        elif self.concurrency == JobConcurrency.QUEUE:
+            if self.lock.locked():
+                self.lock.release()
+        elif self.concurrency in (
+            JobConcurrency.GROUP_REJECT,
+            JobConcurrency.GROUP_QUEUE,
         ):
-            return
+            if job_group and job_group.has_lock:
+                job_group.release()
 
-        if self.limit == JobExecutionLimit.ONCE and self.lock.locked():
-            on_condition = (
-                JobException if self.on_condition is None else self.on_condition
-            )
-            raise on_condition("Another job is running")
+    async def _handle_concurrency_control(
+        self, job_group: JobGroup | None, job: SupervisorJob
+    ) -> None:
+        """Handle concurrency control limits."""
+        if self.concurrency == JobConcurrency.REJECT:
+            if self.lock.locked():
+                on_condition = (
+                    JobException if self.on_condition is None else self.on_condition
+                )
+                raise on_condition("Another job is running")
+            await self.lock.acquire()
+        elif self.concurrency == JobConcurrency.QUEUE:
+            await self.lock.acquire()
+        elif self.concurrency == JobConcurrency.GROUP_REJECT:
+            try:
+                await cast(JobGroup, job_group).acquire(job, wait=False)
+            except JobGroupExecutionLimitExceeded as err:
+                if self.on_condition:
+                    raise self.on_condition(str(err)) from err
+                raise err
+        elif self.concurrency == JobConcurrency.GROUP_QUEUE:
+            try:
+                await cast(JobGroup, job_group).acquire(job, wait=True)
+            except JobGroupExecutionLimitExceeded as err:
+                if self.on_condition:
+                    raise self.on_condition(str(err)) from err
+                raise err
 
-        await self.lock.acquire()
+    async def _handle_throttling(self, group_name: str | None) -> bool:
+        """Handle throttling limits. Returns True if job should continue, False if throttled."""
+        if self.throttle in (JobThrottle.THROTTLE, JobThrottle.GROUP_THROTTLE):
+            time_since_last_call = datetime.now() - self.last_call(group_name)
+            throttle_period = self.throttle_period(group_name)
+            if time_since_last_call < throttle_period:
+                # Always return False when throttled (skip execution)
+                return False
+        elif self.throttle in (JobThrottle.RATE_LIMIT, JobThrottle.GROUP_RATE_LIMIT):
+            # Only reprocess array when necessary (at limit)
+            if len(self.rate_limited_calls(group_name)) >= self.throttle_max_calls:
+                self.set_rate_limited_calls(
+                    [
+                        call
+                        for call in self.rate_limited_calls(group_name)
+                        if call > datetime.now() - self.throttle_period(group_name)
+                    ],
+                    group_name,
+                )
 
-    def _release_exception_limits(self) -> None:
-        """Release possible exception limits."""
-        if self.limit not in (
-            JobExecutionLimit.SINGLE_WAIT,
-            JobExecutionLimit.ONCE,
-            JobExecutionLimit.THROTTLE_WAIT,
-            JobExecutionLimit.GROUP_THROTTLE_WAIT,
-        ):
-            return
-        self.lock.release()
+            if len(self.rate_limited_calls(group_name)) >= self.throttle_max_calls:
+                on_condition = (
+                    JobException if self.on_condition is None else self.on_condition
+                )
+                raise on_condition(
+                    f"Rate limit exceeded, more than {self.throttle_max_calls} calls in {self.throttle_period(group_name)}",
+                )
+
+        return True

supervisor/plugins/dns.py

@@ -15,7 +15,8 @@ from awesomeversion import AwesomeVersion
 import jinja2
 import voluptuous as vol
 
-from ..const import ATTR_SERVERS, DNS_SUFFIX, LogLevel
+from ..bus import EventListener
+from ..const import ATTR_SERVERS, DNS_SUFFIX, BusEvent, LogLevel
 from ..coresys import CoreSys
 from ..dbus.const import MulticastProtocolEnabled
 from ..docker.const import ContainerState
@@ -77,6 +78,12 @@ class PluginDns(PluginBase):
 
         self._hosts: list[HostEntry] = []
         self._loop: bool = False
+        self._cached_locals: list[str] | None = None
+
+        # Debouncing system for rapid local changes
+        self._locals_changed_handle: asyncio.TimerHandle | None = None
+        self._restart_after_locals_change_handle: asyncio.Task | None = None
+        self._connectivity_check_listener: EventListener | None = None
 
     @property
     def hosts(self) -> Path:
@@ -91,6 +98,12 @@ class PluginDns(PluginBase):
     @property
     def locals(self) -> list[str]:
         """Return list of local system DNS servers."""
+        if self._cached_locals is None:
+            self._cached_locals = self._compute_locals()
+        return self._cached_locals
+
+    def _compute_locals(self) -> list[str]:
+        """Compute list of local system DNS servers."""
         servers: list[str] = []
         for server in [
             f"dns://{server!s}" for server in self.sys_host.network.dns_servers
@@ -100,6 +113,52 @@ class PluginDns(PluginBase):
 
         return servers
 
+    async def _on_dns_container_running(self, event: DockerContainerStateEvent) -> None:
+        """Handle DNS container state change to running and trigger connectivity check."""
+        if event.name == self.instance.name and event.state == ContainerState.RUNNING:
+            # Wait before CoreDNS actually becomes available
+            await asyncio.sleep(5)
+
+            _LOGGER.debug("CoreDNS started, checking connectivity")
+            await self.sys_supervisor.check_connectivity()
+
+    async def _restart_dns_after_locals_change(self) -> None:
+        """Restart DNS after a debounced delay for local changes."""
+        old_locals = self._cached_locals
+        new_locals = self._compute_locals()
+        if old_locals == new_locals:
+            return
+
+        _LOGGER.debug("DNS locals changed from %s to %s", old_locals, new_locals)
+        self._cached_locals = new_locals
+        if not await self.instance.is_running():
+            return
+
+        await self.restart()
+        self._restart_after_locals_change_handle = None
+
+    def _trigger_restart_dns_after_locals_change(self) -> None:
+        """Trigger a restart of DNS after local changes."""
+        # Cancel existing restart task if any
+        if self._restart_after_locals_change_handle:
+            self._restart_after_locals_change_handle.cancel()
+
+        self._restart_after_locals_change_handle = self.sys_create_task(
+            self._restart_dns_after_locals_change()
+        )
+        self._locals_changed_handle = None
+
+    def notify_locals_changed(self) -> None:
+        """Schedule a debounced DNS restart for local changes."""
+        # Cancel existing timer if any
+        if self._locals_changed_handle:
+            self._locals_changed_handle.cancel()
+
+        # Schedule new timer with 1 second delay
+        self._locals_changed_handle = self.sys_call_later(
+            1.0, self._trigger_restart_dns_after_locals_change
+        )
+
     @property
     def servers(self) -> list[str]:
         """Return list of DNS servers."""
@@ -188,6 +247,13 @@ class PluginDns(PluginBase):
             _LOGGER.error("Can't read hosts.tmpl: %s", err)
 
         await self._init_hosts()
+
+        # Register Docker event listener for connectivity checks
+        if not self._connectivity_check_listener:
+            self._connectivity_check_listener = self.sys_bus.register_event(
+                BusEvent.DOCKER_CONTAINER_STATE_CHANGE, self._on_dns_container_running
+            )
+
         await super().load()
 
         # Update supervisor
@@ -243,6 +309,16 @@ class PluginDns(PluginBase):
 
     async def stop(self) -> None:
         """Stop CoreDNS."""
+        # Cancel any pending locals change timer
+        if self._locals_changed_handle:
+            self._locals_changed_handle.cancel()
+            self._locals_changed_handle = None
+
+        # Wait for any pending restart before stopping
+        if self._restart_after_locals_change_handle:
+            self._restart_after_locals_change_handle.cancel()
+            self._restart_after_locals_change_handle = None
+
         _LOGGER.info("Stopping CoreDNS plugin")
         try:
             await self.instance.stop()

supervisor/resolution/checks/duplicate_os_installation.py

@@ -0,0 +1,108 @@
+"""Helpers to check for duplicate OS installations."""
+
+import logging
+
+from ...const import CoreState
+from ...coresys import CoreSys
+from ...dbus.udisks2.data import DeviceSpecification
+from ..const import ContextType, IssueType, UnhealthyReason
+from .base import CheckBase
+
+_LOGGER: logging.Logger = logging.getLogger(__name__)
+
+# Partition labels to check for duplicates (GPT-based installations)
+HAOS_PARTITIONS = [
+    "hassos-boot",
+    "hassos-kernel0",
+    "hassos-kernel1",
+    "hassos-system0",
+    "hassos-system1",
+]
+
+# Partition UUIDs to check for duplicates (MBR-based installations)
+HAOS_PARTITION_UUIDS = [
+    "48617373-01",  # hassos-boot
+    "48617373-05",  # hassos-kernel0
+    "48617373-06",  # hassos-system0
+    "48617373-07",  # hassos-kernel1
+    "48617373-08",  # hassos-system1
+]
+
+
+def _get_device_specifications():
+    """Generate DeviceSpecification objects for both GPT and MBR partitions."""
+    # GPT-based installations (partition labels)
+    for partition_label in HAOS_PARTITIONS:
+        yield (
+            DeviceSpecification(partlabel=partition_label),
+            "partition",
+            partition_label,
+        )
+
+    # MBR-based installations (partition UUIDs)
+    for partition_uuid in HAOS_PARTITION_UUIDS:
+        yield (
+            DeviceSpecification(partuuid=partition_uuid),
+            "partition UUID",
+            partition_uuid,
+        )
+
+
+def setup(coresys: CoreSys) -> CheckBase:
+    """Check setup function."""
+    return CheckDuplicateOSInstallation(coresys)
+
+
+class CheckDuplicateOSInstallation(CheckBase):
+    """CheckDuplicateOSInstallation class for check."""
+
+    async def run_check(self) -> None:
+        """Run check if not affected by issue."""
+        if not self.sys_os.available:
+            _LOGGER.debug(
+                "Skipping duplicate OS installation check, OS is not available"
+            )
+            return
+
+        for device_spec, spec_type, identifier in _get_device_specifications():
+            resolved = await self.sys_dbus.udisks2.resolve_device(device_spec)
+            if resolved and len(resolved) > 1:
+                _LOGGER.warning(
+                    "Found duplicate OS installation: %s %s exists on %d devices (%s)",
+                    identifier,
+                    spec_type,
+                    len(resolved),
+                    ", ".join(str(device.device) for device in resolved),
+                )
+                self.sys_resolution.add_unhealthy_reason(
+                    UnhealthyReason.DUPLICATE_OS_INSTALLATION
+                )
+                self.sys_resolution.create_issue(
+                    IssueType.DUPLICATE_OS_INSTALLATION,
+                    ContextType.SYSTEM,
+                )
+                return
+
+    async def approve_check(self, reference: str | None = None) -> bool:
+        """Approve check if it is affected by issue."""
+        # Check all partitions for duplicates since issue is created without reference
+        for device_spec, _, _ in _get_device_specifications():
+            resolved = await self.sys_dbus.udisks2.resolve_device(device_spec)
+            if resolved and len(resolved) > 1:
+                return True
+        return False
+
+    @property
+    def issue(self) -> IssueType:
+        """Return a IssueType enum."""
+        return IssueType.DUPLICATE_OS_INSTALLATION
+
+    @property
+    def context(self) -> ContextType:
+        """Return a ContextType enum."""
+        return ContextType.SYSTEM
+
+    @property
+    def states(self) -> list[CoreState]:
+        """Return a list of valid states when this check can run."""
+        return [CoreState.SETUP]

supervisor/resolution/checks/multiple_data_disks.py

@@ -21,6 +21,9 @@ class CheckMultipleDataDisks(CheckBase):
 
     async def run_check(self) -> None:
         """Run check if not affected by issue."""
+        if not self.sys_os.available:
+            return
+
         for block_device in self.sys_dbus.udisks2.block_devices:
             if self._block_device_has_name_issue(block_device):
                 self.sys_resolution.create_issue(

supervisor/resolution/const.py

@@ -64,10 +64,11 @@ class UnhealthyReason(StrEnum):
     """Reasons for unsupported status."""
 
     DOCKER = "docker"
+    DUPLICATE_OS_INSTALLATION = "duplicate_os_installation"
     OSERROR_BAD_MESSAGE = "oserror_bad_message"
     PRIVILEGED = "privileged"
-    SUPERVISOR = "supervisor"
     SETUP = "setup"
+    SUPERVISOR = "supervisor"
     UNTRUSTED = "untrusted"
 
 
@@ -83,6 +84,7 @@ class IssueType(StrEnum):
     DEVICE_ACCESS_MISSING = "device_access_missing"
     DISABLED_DATA_DISK = "disabled_data_disk"
     DNS_LOOP = "dns_loop"
+    DUPLICATE_OS_INSTALLATION = "duplicate_os_installation"
     DNS_SERVER_FAILED = "dns_server_failed"
     DNS_SERVER_IPV6_ERROR = "dns_server_ipv6_error"
     DOCKER_CONFIG = "docker_config"

supervisor/resolution/fixups/base.py

@@ -3,6 +3,7 @@
 from abc import ABC, abstractmethod
 import logging
 
+from ...const import BusEvent
 from ...coresys import CoreSys, CoreSysAttributes
 from ...exceptions import ResolutionFixupError
 from ..const import ContextType, IssueType, SuggestionType
@@ -66,6 +67,11 @@ class FixupBase(ABC, CoreSysAttributes):
         """Return if a fixup can be apply as auto fix."""
         return False
 
+    @property
+    def bus_event(self) -> BusEvent | None:
+        """Return the BusEvent that triggers this fixup, or None if not event-based."""
+        return None
+
     @property
     def all_suggestions(self) -> list[Suggestion]:
         """List of all suggestions which when applied run this fixup."""

supervisor/resolution/fixups/store_execute_reload.py

@@ -2,6 +2,7 @@
 
 import logging
 
+from ...const import BusEvent
 from ...coresys import CoreSys
 from ...exceptions import (
     ResolutionFixupError,
@@ -68,3 +69,8 @@ class FixupStoreExecuteReload(FixupBase):
     def auto(self) -> bool:
         """Return if a fixup can be apply as auto fix."""
         return True
+
+    @property
+    def bus_event(self) -> BusEvent | None:
+        """Return the BusEvent that triggers this fixup, or None if not event-based."""
+        return BusEvent.SUPERVISOR_CONNECTIVITY_CHANGE

supervisor/resolution/module.py

@@ -5,6 +5,7 @@ from typing import Any
 
 import attr
 
+from ..bus import EventListener
 from ..coresys import CoreSys, CoreSysAttributes
 from ..exceptions import ResolutionError, ResolutionNotFound
 from ..homeassistant.const import WSEvent
@@ -46,6 +47,9 @@ class ResolutionManager(FileConfiguration, CoreSysAttributes):
         self._unsupported: list[UnsupportedReason] = []
         self._unhealthy: list[UnhealthyReason] = []
 
+        # Map suggestion UUID to event listeners (list)
+        self._suggestion_listeners: dict[str, list[EventListener]] = {}
+
     async def load_modules(self):
         """Load resolution evaluation, check and fixup modules."""
 
@@ -105,6 +109,19 @@ class ResolutionManager(FileConfiguration, CoreSysAttributes):
         )
         self._suggestions.append(suggestion)
 
+        # Register event listeners if fixups have a bus_event
+        listeners: list[EventListener] = []
+        for fixup in self.fixup.fixes_for_suggestion(suggestion):
+            if fixup.auto and fixup.bus_event:
+
+                def event_callback(reference, fixup=fixup):
+                    return fixup(suggestion)
+
+                listener = self.sys_bus.register_event(fixup.bus_event, event_callback)
+                listeners.append(listener)
+        if listeners:
+            self._suggestion_listeners[suggestion.uuid] = listeners
+
         # Event on suggestion added to issue
         for issue in self.issues_for_suggestion(suggestion):
             self.sys_homeassistant.websocket.supervisor_event(
@@ -233,6 +250,11 @@ class ResolutionManager(FileConfiguration, CoreSysAttributes):
             )
         self._suggestions.remove(suggestion)
 
+        # Remove event listeners if present
+        listeners = self._suggestion_listeners.pop(suggestion.uuid, [])
+        for listener in listeners:
+            self.sys_bus.remove_listener(listener)
+
         # Event on suggestion removed from issues
         for issue in self.issues_for_suggestion(suggestion):
             self.sys_homeassistant.websocket.supervisor_event(

supervisor/store/__init__.py

@@ -4,7 +4,7 @@ import asyncio
 from collections.abc import Awaitable
 import logging
 
-from ..const import ATTR_REPOSITORIES, URL_HASSIO_ADDONS
+from ..const import ATTR_REPOSITORIES, REPOSITORY_CORE, URL_HASSIO_ADDONS
 from ..coresys import CoreSys, CoreSysAttributes
 from ..exceptions import (
     StoreError,
@@ -18,14 +18,10 @@ from ..jobs.decorator import Job, JobCondition
 from ..resolution.const import ContextType, IssueType, SuggestionType
 from ..utils.common import FileConfiguration
 from .addon import AddonStore
-from .const import FILE_HASSIO_STORE, StoreType
+from .const import FILE_HASSIO_STORE, BuiltinRepository
 from .data import StoreData
 from .repository import Repository
-from .validate import (
-    BUILTIN_REPOSITORIES,
-    SCHEMA_STORE_FILE,
-    ensure_builtin_repositories,
-)
+from .validate import DEFAULT_REPOSITORIES, SCHEMA_STORE_FILE
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
@@ -56,7 +52,8 @@ class StoreManager(CoreSysAttributes, FileConfiguration):
         return [
             repository.source
             for repository in self.all
-            if repository.type == StoreType.GIT
+            if repository.slug
+            not in {BuiltinRepository.LOCAL.value, BuiltinRepository.CORE.value}
         ]
 
     def get(self, slug: str) -> Repository:
@@ -65,20 +62,15 @@ class StoreManager(CoreSysAttributes, FileConfiguration):
             raise StoreNotFound()
         return self.repositories[slug]
 
-    def get_from_url(self, url: str) -> Repository:
-        """Return Repository with slug."""
-        for repository in self.all:
-            if repository.source != url:
-                continue
-            return repository
-        raise StoreNotFound()
-
     async def load(self) -> None:
-        """Start up add-on management."""
-        # Init custom repositories and load add-ons
-        await self.update_repositories(
-            self._data[ATTR_REPOSITORIES], add_with_errors=True
+        """Start up add-on store management."""
+        # Make sure the built-in repositories are all present
+        # This is especially important when adding new built-in repositories
+        # to make sure existing installations have them.
+        all_repositories: set[str] = (
+            set(self._data.get(ATTR_REPOSITORIES, [])) | DEFAULT_REPOSITORIES
         )
+        await self.update_repositories(all_repositories, issue_on_error=True)
 
     @Job(
         name="store_manager_reload",
@@ -126,14 +118,14 @@ class StoreManager(CoreSysAttributes, FileConfiguration):
     )
     async def add_repository(self, url: str, *, persist: bool = True) -> None:
         """Add a repository."""
-        await self._add_repository(url, persist=persist, add_with_errors=False)
+        await self._add_repository(url, persist=persist, issue_on_error=False)
 
     async def _add_repository(
-        self, url: str, *, persist: bool = True, add_with_errors: bool = False
+        self, url: str, *, persist: bool = True, issue_on_error: bool = False
     ) -> None:
         """Add a repository."""
         if url == URL_HASSIO_ADDONS:
-            url = StoreType.CORE
+            url = REPOSITORY_CORE
 
         repository = Repository.create(self.coresys, url)
 
@@ -145,7 +137,7 @@ class StoreManager(CoreSysAttributes, FileConfiguration):
             await repository.load()
         except StoreGitCloneError as err:
             _LOGGER.error("Can't retrieve data from %s due to %s", url, err)
-            if add_with_errors:
+            if issue_on_error:
                 self.sys_resolution.create_issue(
                     IssueType.FATAL_ERROR,
                     ContextType.STORE,
@@ -158,7 +150,7 @@ class StoreManager(CoreSysAttributes, FileConfiguration):
 
         except StoreGitError as err:
             _LOGGER.error("Can't load data from repository %s due to %s", url, err)
-            if add_with_errors:
+            if issue_on_error:
                 self.sys_resolution.create_issue(
                     IssueType.FATAL_ERROR,
                     ContextType.STORE,
@@ -171,7 +163,7 @@ class StoreManager(CoreSysAttributes, FileConfiguration):
 
         except StoreJobError as err:
             _LOGGER.error("Can't add repository %s due to %s", url, err)
-            if add_with_errors:
+            if issue_on_error:
                 self.sys_resolution.create_issue(
                     IssueType.FATAL_ERROR,
                     ContextType.STORE,
@@ -184,7 +176,7 @@ class StoreManager(CoreSysAttributes, FileConfiguration):
 
         else:
             if not await repository.validate():
-                if add_with_errors:
+                if issue_on_error:
                     _LOGGER.error("%s is not a valid add-on repository", url)
                     self.sys_resolution.create_issue(
                         IssueType.CORRUPT_REPOSITORY,
@@ -213,7 +205,7 @@ class StoreManager(CoreSysAttributes, FileConfiguration):
 
     async def remove_repository(self, repository: Repository, *, persist: bool = True):
         """Remove a repository."""
-        if repository.source in BUILTIN_REPOSITORIES:
+        if repository.is_builtin:
             raise StoreInvalidAddonRepo(
                 "Can't remove built-in repositories!", logger=_LOGGER.error
             )
@@ -234,40 +226,50 @@ class StoreManager(CoreSysAttributes, FileConfiguration):
     @Job(name="store_manager_update_repositories")
     async def update_repositories(
         self,
-        list_repositories: list[str],
+        list_repositories: set[str],
         *,
-        add_with_errors: bool = False,
+        issue_on_error: bool = False,
         replace: bool = True,
     ):
-        """Add a new custom repository."""
-        new_rep = set(
-            ensure_builtin_repositories(list_repositories)
-            if replace
-            else list_repositories + self.repository_urls
-        )
-        old_rep = {repository.source for repository in self.all}
+        """Update repositories by adding new ones and removing stale ones."""
+        current_repositories = {repository.source for repository in self.all}
+
+        # Determine repositories to add
+        repositories_to_add = list_repositories - current_repositories
 
         # Add new repositories
         add_errors = await asyncio.gather(
             *[
-                self._add_repository(url, persist=False, add_with_errors=True)
-                if add_with_errors
+                # Use _add_repository to avoid JobCondition.SUPERVISOR_UPDATED
+                # to prevent proper loading of repositories on startup.
+                self._add_repository(url, persist=False, issue_on_error=True)
+                if issue_on_error
                 else self.add_repository(url, persist=False)
-                for url in new_rep - old_rep
+                for url in repositories_to_add
             ],
             return_exceptions=True,
         )
 
-        # Delete stale repositories
-        remove_errors = await asyncio.gather(
-            *[
-                self.remove_repository(self.get_from_url(url), persist=False)
-                for url in old_rep - new_rep - BUILTIN_REPOSITORIES
-            ],
-            return_exceptions=True,
-        )
+        remove_errors: list[BaseException | None] = []
+        if replace:
+            # Determine repositories to remove
+            repositories_to_remove: list[Repository] = [
+                repository
+                for repository in self.all
+                if repository.source not in list_repositories
+                and not repository.is_builtin
+            ]
 
-        # Always update data, even there are errors, some changes may have succeeded
+            # Remove repositories
+            remove_errors = await asyncio.gather(
+                *[
+                    self.remove_repository(repository, persist=False)
+                    for repository in repositories_to_remove
+                ],
+                return_exceptions=True,
+            )
+
+        # Always update data, even if there are errors, some changes may have succeeded
         await self.data.update()
         await self._read_addons()
 

supervisor/store/const.py

@@ -3,14 +3,35 @@
 from enum import StrEnum
 from pathlib import Path
 
-from ..const import SUPERVISOR_DATA
+from ..const import (
+    REPOSITORY_CORE,
+    REPOSITORY_LOCAL,
+    SUPERVISOR_DATA,
+    URL_HASSIO_ADDONS,
+)
 
 FILE_HASSIO_STORE = Path(SUPERVISOR_DATA, "store.json")
+"""Repository type definitions for the store."""
 
 
-class StoreType(StrEnum):
-    """Store Types."""
+class BuiltinRepository(StrEnum):
+    """All built-in repositories that come pre-configured."""
 
-    CORE = "core"
-    LOCAL = "local"
-    GIT = "git"
+    # Local repository (non-git, special handling)
+    LOCAL = REPOSITORY_LOCAL
+
+    # Git-based built-in repositories
+    CORE = REPOSITORY_CORE
+    COMMUNITY_ADDONS = "https://github.com/hassio-addons/repository"
+    ESPHOME = "https://github.com/esphome/home-assistant-addon"
+    MUSIC_ASSISTANT = "https://github.com/music-assistant/home-assistant-addon"
+
+    @property
+    def git_url(self) -> str:
+        """Return the git URL for this repository."""
+        if self == BuiltinRepository.LOCAL:
+            raise RuntimeError("Local repository does not have a git URL")
+        if self == BuiltinRepository.CORE:
+            return URL_HASSIO_ADDONS
+        else:
+            return self.value  # For URL-based repos, value is the URL

supervisor/store/data.py

@@ -25,7 +25,6 @@ from ..exceptions import ConfigurationFileError
 from ..resolution.const import ContextType, IssueType, SuggestionType, UnhealthyReason
 from ..utils.common import find_one_filetype, read_json_or_yaml_file
 from ..utils.json import read_json_file
-from .const import StoreType
 from .utils import extract_hash_from_path
 from .validate import SCHEMA_REPOSITORY_CONFIG
 
@@ -169,7 +168,7 @@ class StoreData(CoreSysAttributes):
                 self.sys_resolution.add_unhealthy_reason(
                     UnhealthyReason.OSERROR_BAD_MESSAGE
                 )
-            elif path.stem != StoreType.LOCAL:
+            elif repository != REPOSITORY_LOCAL:
                 suggestion = [SuggestionType.EXECUTE_RESET]
             self.sys_resolution.create_issue(
                 IssueType.CORRUPT_REPOSITORY,

supervisor/store/repository.py

@@ -10,14 +10,21 @@ import voluptuous as vol
 
 from supervisor.utils import get_latest_mtime
 
-from ..const import ATTR_MAINTAINER, ATTR_NAME, ATTR_URL, FILE_SUFFIX_CONFIGURATION
+from ..const import (
+    ATTR_MAINTAINER,
+    ATTR_NAME,
+    ATTR_URL,
+    FILE_SUFFIX_CONFIGURATION,
+    REPOSITORY_CORE,
+    REPOSITORY_LOCAL,
+)
 from ..coresys import CoreSys, CoreSysAttributes
 from ..exceptions import ConfigurationFileError, StoreError
 from ..utils.common import read_json_or_yaml_file
-from .const import StoreType
+from .const import BuiltinRepository
 from .git import GitRepo
 from .utils import get_hash_from_repository
-from .validate import SCHEMA_REPOSITORY_CONFIG, BuiltinRepository
+from .validate import SCHEMA_REPOSITORY_CONFIG
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 UNKNOWN = "unknown"
@@ -26,21 +33,45 @@ UNKNOWN = "unknown"
 class Repository(CoreSysAttributes, ABC):
     """Add-on store repository in Supervisor."""
 
-    def __init__(self, coresys: CoreSys, repository: str):
+    def __init__(self, coresys: CoreSys, repository: str, local_path: Path, slug: str):
         """Initialize add-on store repository object."""
-        self._slug: str
-        self._type: StoreType
+        self._slug: str = slug
+        self._local_path: Path = local_path
         self.coresys: CoreSys = coresys
         self.source: str = repository
 
     @staticmethod
     def create(coresys: CoreSys, repository: str) -> Repository:
         """Create a repository instance."""
-        if repository == StoreType.LOCAL:
-            return RepositoryLocal(coresys)
         if repository in BuiltinRepository:
-            return RepositoryGitBuiltin(coresys, BuiltinRepository(repository))
-        return RepositoryCustom(coresys, repository)
+            return Repository._create_builtin(coresys, BuiltinRepository(repository))
+        else:
+            return Repository._create_custom(coresys, repository)
+
+    @staticmethod
+    def _create_builtin(coresys: CoreSys, builtin: BuiltinRepository) -> Repository:
+        """Create builtin repository."""
+        if builtin == BuiltinRepository.LOCAL:
+            slug = REPOSITORY_LOCAL
+            local_path = coresys.config.path_addons_local
+            return RepositoryLocal(coresys, local_path, slug)
+        elif builtin == BuiltinRepository.CORE:
+            slug = REPOSITORY_CORE
+            local_path = coresys.config.path_addons_core
+        else:
+            # For other builtin repositories (URL-based)
+            slug = get_hash_from_repository(builtin.value)
+            local_path = coresys.config.path_addons_git / slug
+        return RepositoryGitBuiltin(
+            coresys, builtin.value, local_path, slug, builtin.git_url
+        )
+
+    @staticmethod
+    def _create_custom(coresys: CoreSys, repository: str) -> RepositoryCustom:
+        """Create custom repository."""
+        slug = get_hash_from_repository(repository)
+        local_path = coresys.config.path_addons_git / slug
+        return RepositoryCustom(coresys, repository, local_path, slug)
 
     def __repr__(self) -> str:
         """Return internal representation."""
@@ -52,9 +83,9 @@ class Repository(CoreSysAttributes, ABC):
         return self._slug
 
     @property
-    def type(self) -> StoreType:
-        """Return type of the store."""
-        return self._type
+    def local_path(self) -> Path:
+        """Return local path to repository."""
+        return self._local_path
 
     @property
     def data(self) -> dict:
@@ -76,6 +107,11 @@ class Repository(CoreSysAttributes, ABC):
         """Return url of repository."""
         return self.data.get(ATTR_MAINTAINER, UNKNOWN)
 
+    @property
+    @abstractmethod
+    def is_builtin(self) -> bool:
+        """Return True if this is a built-in repository."""
+
     @abstractmethod
     async def validate(self) -> bool:
         """Check if store is valid."""
@@ -103,12 +139,10 @@ class Repository(CoreSysAttributes, ABC):
 class RepositoryBuiltin(Repository, ABC):
     """A built-in add-on repository."""
 
-    def __init__(self, coresys: CoreSys, builtin: BuiltinRepository) -> None:
-        """Initialize object."""
-        super().__init__(coresys, builtin.value)
-        self._builtin = builtin
-        self._slug = builtin.id
-        self._type = builtin.type
+    @property
+    def is_builtin(self) -> bool:
+        """Return True if this is a built-in repository."""
+        return True
 
     async def validate(self) -> bool:
         """Assume built-in repositories are always valid."""
@@ -171,15 +205,15 @@ class RepositoryGit(Repository, ABC):
 class RepositoryLocal(RepositoryBuiltin):
     """A local add-on repository."""
 
-    def __init__(self, coresys: CoreSys) -> None:
+    def __init__(self, coresys: CoreSys, local_path: Path, slug: str) -> None:
         """Initialize object."""
-        super().__init__(coresys, BuiltinRepository.LOCAL)
+        super().__init__(coresys, BuiltinRepository.LOCAL.value, local_path, slug)
         self._latest_mtime: float | None = None
 
     async def load(self) -> None:
         """Load addon repository."""
         self._latest_mtime, _ = await self.sys_run_in_executor(
-            get_latest_mtime, self.sys_config.path_addons_local
+            get_latest_mtime, self.local_path
         )
 
     async def update(self) -> bool:
@@ -189,7 +223,7 @@ class RepositoryLocal(RepositoryBuiltin):
         """
         # Check local modifications
         latest_mtime, modified_path = await self.sys_run_in_executor(
-            get_latest_mtime, self.sys_config.path_addons_local
+            get_latest_mtime, self.local_path
         )
         if self._latest_mtime != latest_mtime:
             _LOGGER.debug(
@@ -212,21 +246,26 @@ class RepositoryLocal(RepositoryBuiltin):
 class RepositoryGitBuiltin(RepositoryBuiltin, RepositoryGit):
     """A built-in add-on repository based on git."""
 
-    def __init__(self, coresys: CoreSys, builtin: BuiltinRepository) -> None:
+    def __init__(
+        self, coresys: CoreSys, repository: str, local_path: Path, slug: str, url: str
+    ) -> None:
         """Initialize object."""
-        super().__init__(coresys, builtin)
-        self._git = GitRepo(coresys, builtin.get_path(coresys), builtin.url)
+        super().__init__(coresys, repository, local_path, slug)
+        self._git = GitRepo(coresys, local_path, url)
 
 
 class RepositoryCustom(RepositoryGit):
     """A custom add-on repository."""
 
-    def __init__(self, coresys: CoreSys, url: str) -> None:
+    def __init__(self, coresys: CoreSys, url: str, local_path: Path, slug: str) -> None:
         """Initialize object."""
-        super().__init__(coresys, url)
-        self._slug = get_hash_from_repository(url)
-        self._type = StoreType.GIT
-        self._git = GitRepo(coresys, coresys.config.path_addons_git / self._slug, url)
+        super().__init__(coresys, url, local_path, slug)
+        self._git = GitRepo(coresys, local_path, url)
+
+    @property
+    def is_builtin(self) -> bool:
+        """Return True if this is a built-in repository."""
+        return False
 
     async def remove(self) -> None:
         """Remove add-on repository."""

supervisor/store/validate.py

@@ -1,62 +1,10 @@
 """Validate add-ons options schema."""
 
-from enum import StrEnum
-from pathlib import Path
-
 import voluptuous as vol
 
-from ..const import (
-    ATTR_MAINTAINER,
-    ATTR_NAME,
-    ATTR_REPOSITORIES,
-    ATTR_URL,
-    URL_HASSIO_ADDONS,
-)
-from ..coresys import CoreSys
+from ..const import ATTR_MAINTAINER, ATTR_NAME, ATTR_REPOSITORIES, ATTR_URL
 from ..validate import RE_REPOSITORY
-from .const import StoreType
-from .utils import get_hash_from_repository
-
-URL_COMMUNITY_ADDONS = "https://github.com/hassio-addons/repository"
-URL_ESPHOME = "https://github.com/esphome/home-assistant-addon"
-URL_MUSIC_ASSISTANT = "https://github.com/music-assistant/home-assistant-addon"
-
-
-class BuiltinRepository(StrEnum):
-    """Built-in add-on repository."""
-
-    CORE = StoreType.CORE.value
-    LOCAL = StoreType.LOCAL.value
-    COMMUNITY_ADDONS = URL_COMMUNITY_ADDONS
-    ESPHOME = URL_ESPHOME
-    MUSIC_ASSISTANT = URL_MUSIC_ASSISTANT
-
-    def __init__(self, value: str) -> None:
-        """Initialize repository item."""
-        if value == StoreType.LOCAL:
-            self.id = value
-            self.url = ""
-            self.type = StoreType.LOCAL
-        elif value == StoreType.CORE:
-            self.id = value
-            self.url = URL_HASSIO_ADDONS
-            self.type = StoreType.CORE
-        else:
-            self.id = get_hash_from_repository(value)
-            self.url = value
-            self.type = StoreType.GIT
-
-    def get_path(self, coresys: CoreSys) -> Path:
-        """Get path to git repo for repository."""
-        if self.id == StoreType.LOCAL:
-            return coresys.config.path_addons_local
-        if self.id == StoreType.CORE:
-            return coresys.config.path_addons_core
-        return Path(coresys.config.path_addons_git, self.id)
-
-
-BUILTIN_REPOSITORIES = {r.value for r in BuiltinRepository}
-
+from .const import BuiltinRepository
 
 # pylint: disable=no-value-for-parameter
 SCHEMA_REPOSITORY_CONFIG = vol.Schema(
@@ -69,18 +17,9 @@ SCHEMA_REPOSITORY_CONFIG = vol.Schema(
 )
 
 
-def ensure_builtin_repositories(addon_repositories: list[str]) -> list[str]:
-    """Ensure builtin repositories are in list.
-
-    Note: This should not be used in validation as the resulting list is not
-    stable. This can have side effects when comparing data later on.
-    """
-    return list(set(addon_repositories) | BUILTIN_REPOSITORIES)
-
-
 def validate_repository(repository: str) -> str:
     """Validate a valid repository."""
-    if repository in [StoreType.CORE, StoreType.LOCAL]:
+    if repository in BuiltinRepository:
         return repository
 
     data = RE_REPOSITORY.match(repository)
@@ -96,10 +35,12 @@ def validate_repository(repository: str) -> str:
 
 repositories = vol.All([validate_repository], vol.Unique())
 
+DEFAULT_REPOSITORIES = {repo.value for repo in BuiltinRepository}
+
 SCHEMA_STORE_FILE = vol.Schema(
     {
         vol.Optional(
-            ATTR_REPOSITORIES, default=list(BUILTIN_REPOSITORIES)
+            ATTR_REPOSITORIES, default=lambda: list(DEFAULT_REPOSITORIES)
         ): repositories,
     },
     extra=vol.REMOVE_EXTRA,

supervisor/supervisor.py

@@ -46,7 +46,7 @@ def _check_connectivity_throttle_period(coresys: CoreSys, *_) -> timedelta:
     if coresys.supervisor.connectivity:
         return timedelta(minutes=10)
 
-    return timedelta(seconds=30)
+    return timedelta(seconds=5)
 
 
 class Supervisor(CoreSysAttributes):
@@ -291,14 +291,16 @@ class Supervisor(CoreSysAttributes):
         limit=JobExecutionLimit.THROTTLE,
         throttle_period=_check_connectivity_throttle_period,
     )
-    async def check_connectivity(self):
-        """Check the connection."""
+    async def check_connectivity(self) -> None:
+        """Check the Internet connectivity from Supervisor's point of view."""
         timeout = aiohttp.ClientTimeout(total=10)
         try:
             await self.sys_websession.head(
                 "https://checkonline.home-assistant.io/online.txt", timeout=timeout
             )
-        except (ClientError, TimeoutError):
+        except (ClientError, TimeoutError) as err:
+            _LOGGER.debug("Supervisor Connectivity check failed: %s", err)
             self.connectivity = False
         else:
+            _LOGGER.debug("Supervisor Connectivity check succeeded")
             self.connectivity = True

supervisor/utils/sentry.py

@@ -12,6 +12,7 @@ from sentry_sdk.integrations.dedupe import DedupeIntegration
 from sentry_sdk.integrations.excepthook import ExcepthookIntegration
 from sentry_sdk.integrations.logging import LoggingIntegration
 from sentry_sdk.integrations.threading import ThreadingIntegration
+from sentry_sdk.scrubber import DEFAULT_DENYLIST, EventScrubber
 
 from ..const import SUPERVISOR_VERSION
 from ..coresys import CoreSys
@@ -26,6 +27,7 @@ def init_sentry(coresys: CoreSys) -> None:
     """Initialize sentry client."""
     if not sentry_sdk.is_initialized():
         _LOGGER.info("Initializing Supervisor Sentry")
+        denylist = DEFAULT_DENYLIST + ["psk", "ssid"]
         # Don't use AsyncioIntegration(). We commonly handle task exceptions
         # outside of tasks. This would cause exception we gracefully handle to
         # be captured by sentry.
@@ -34,6 +36,7 @@ def init_sentry(coresys: CoreSys) -> None:
             before_send=partial(filter_data, coresys),
             auto_enabling_integrations=False,
             default_integrations=False,
+            event_scrubber=EventScrubber(denylist=denylist),
             integrations=[
                 AioHttpIntegration(
                     failed_request_status_codes=frozenset(range(500, 600))

supervisor/validate.py

@@ -182,7 +182,7 @@ SCHEMA_DOCKER_CONFIG = vol.Schema(
                 }
             }
         ),
-        vol.Optional(ATTR_ENABLE_IPV6): vol.Boolean(),
+        vol.Optional(ATTR_ENABLE_IPV6, default=None): vol.Maybe(vol.Boolean()),
     }
 )
 

tests/addons/test_addon.py

@@ -209,7 +209,7 @@ async def test_watchdog_on_stop(coresys: CoreSys, install_addon_ssh: Addon) -> N
 
 
 async def test_listener_attached_on_install(
-    coresys: CoreSys, mock_amd64_arch_supported: None, repository
+    coresys: CoreSys, mock_amd64_arch_supported: None, test_repository
 ):
     """Test events listener attached on addon install."""
     coresys.hardware.disk.get_disk_free_space = lambda x: 5000
@@ -242,7 +242,7 @@ async def test_listener_attached_on_install(
 )
 async def test_watchdog_during_attach(
     coresys: CoreSys,
-    repository: Repository,
+    test_repository: Repository,
     boot_timedelta: timedelta,
     restart_count: int,
 ):
@@ -710,7 +710,7 @@ async def test_local_example_install(
     coresys: CoreSys,
     container: MagicMock,
     tmp_supervisor_data: Path,
-    repository,
+    test_repository,
     mock_aarch64_arch_supported: None,
 ):
     """Test install of an addon."""

tests/addons/test_manager.py

@@ -67,7 +67,7 @@ async def fixture_remove_wait_boot(coresys: CoreSys) -> AsyncGenerator[None]:
 
 @pytest.fixture(name="install_addon_example_image")
 async def fixture_install_addon_example_image(
-    coresys: CoreSys, repository
+    coresys: CoreSys, test_repository
 ) -> Generator[Addon]:
     """Install local_example add-on with image."""
     store = coresys.addons.store["local_example_image"]

tests/api/test_addons.py

@@ -54,7 +54,7 @@ async def test_addons_info(
 
 # DEPRECATED - Remove with legacy routing logic on 1/2023
 async def test_addons_info_not_installed(
-    api_client: TestClient, coresys: CoreSys, repository: Repository
+    api_client: TestClient, coresys: CoreSys, test_repository: Repository
 ):
     """Test getting addon info for not installed addon."""
     resp = await api_client.get(f"/addons/{TEST_ADDON_SLUG}/info")
@@ -261,6 +261,98 @@ async def test_api_addon_rebuild_healthcheck(
     assert resp.status == 200
 
 
+async def test_api_addon_rebuild_force(
+    api_client: TestClient,
+    coresys: CoreSys,
+    install_addon_ssh: Addon,
+    container: MagicMock,
+    tmp_supervisor_data,
+    path_extern,
+):
+    """Test rebuilding an image-based addon with force parameter."""
+    coresys.hardware.disk.get_disk_free_space = lambda x: 5000
+    container.status = "running"
+    install_addon_ssh.path_data.mkdir()
+    container.attrs["Config"] = {"Healthcheck": "exists"}
+    await install_addon_ssh.load()
+    await asyncio.sleep(0)
+    assert install_addon_ssh.state == AddonState.STARTUP
+
+    state_changes: list[AddonState] = []
+    _container_events_task: asyncio.Task | None = None
+
+    async def container_events():
+        nonlocal state_changes
+
+        await install_addon_ssh.container_state_changed(
+            _create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.STOPPED)
+        )
+        state_changes.append(install_addon_ssh.state)
+
+        await install_addon_ssh.container_state_changed(
+            _create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.RUNNING)
+        )
+        state_changes.append(install_addon_ssh.state)
+        await asyncio.sleep(0)
+
+        await install_addon_ssh.container_state_changed(
+            _create_test_event(f"addon_{TEST_ADDON_SLUG}", ContainerState.HEALTHY)
+        )
+
+    async def container_events_task(*args, **kwargs):
+        nonlocal _container_events_task
+        _container_events_task = asyncio.create_task(container_events())
+
+    # Test 1: Without force, image-based addon should fail
+    with (
+        patch.object(AddonBuild, "is_valid", return_value=True),
+        patch.object(DockerAddon, "is_running", return_value=False),
+        patch.object(
+            Addon, "need_build", new=PropertyMock(return_value=False)
+        ),  # Image-based
+        patch.object(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])),
+    ):
+        resp = await api_client.post("/addons/local_ssh/rebuild")
+
+    assert resp.status == 400
+    result = await resp.json()
+    assert "Can't rebuild a image based add-on" in result["message"]
+
+    # Reset state for next test
+    state_changes.clear()
+
+    # Test 2: With force=True, image-based addon should succeed
+    with (
+        patch.object(AddonBuild, "is_valid", return_value=True),
+        patch.object(DockerAddon, "is_running", return_value=False),
+        patch.object(
+            Addon, "need_build", new=PropertyMock(return_value=False)
+        ),  # Image-based
+        patch.object(CpuArch, "supported", new=PropertyMock(return_value=["amd64"])),
+        patch.object(DockerAddon, "run", new=container_events_task),
+        patch.object(
+            coresys.docker,
+            "run_command",
+            new=PropertyMock(return_value=CommandReturn(0, b"Build successful")),
+        ),
+        patch.object(
+            DockerAddon, "healthcheck", new=PropertyMock(return_value={"exists": True})
+        ),
+        patch.object(
+            type(coresys.config),
+            "local_to_extern_path",
+            return_value="/addon/path/on/host",
+        ),
+    ):
+        resp = await api_client.post("/addons/local_ssh/rebuild", json={"force": True})
+
+    assert state_changes == [AddonState.STOPPED, AddonState.STARTUP]
+    assert install_addon_ssh.state == AddonState.STARTED
+    assert resp.status == 200
+
+    await _container_events_task
+
+
 async def test_api_addon_uninstall(
     api_client: TestClient,
     coresys: CoreSys,
@@ -441,7 +533,7 @@ async def test_addon_not_found(
         ("get", "/addons/local_ssh/logs/boots/1/follow", False),
     ],
 )
-@pytest.mark.usefixtures("repository")
+@pytest.mark.usefixtures("test_repository")
 async def test_addon_not_installed(
     api_client: TestClient, method: str, url: str, json_expected: bool
 ):

tests/api/test_auth.py

@@ -3,6 +3,7 @@
 from datetime import UTC, datetime, timedelta
 from unittest.mock import AsyncMock, MagicMock, patch
 
+from aiohttp.hdrs import WWW_AUTHENTICATE
 from aiohttp.test_utils import TestClient
 import pytest
 
@@ -166,8 +167,8 @@ async def test_auth_json_invalid_credentials(
     resp = await api_client.post(
         "/auth", json={"username": "test", "password": "wrong"}
     )
-    # Do we really want the API to return 400 here?
-    assert resp.status == 400
+    assert WWW_AUTHENTICATE not in resp.headers
+    assert resp.status == 401
 
 
 @pytest.mark.parametrize("api_client", [TEST_ADDON_SLUG], indirect=True)
@@ -213,8 +214,8 @@ async def test_auth_urlencoded_failure(
         data="username=test&password=fail",
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
-    # Do we really want the API to return 400 here?
-    assert resp.status == 400
+    assert WWW_AUTHENTICATE not in resp.headers
+    assert resp.status == 401
 
 
 @pytest.mark.parametrize("api_client", [TEST_ADDON_SLUG], indirect=True)
@@ -225,7 +226,7 @@ async def test_auth_unsupported_content_type(
     resp = await api_client.post(
         "/auth", data="something", headers={"Content-Type": "text/plain"}
     )
-    # This probably should be 400 here for better consistency
+    assert "Basic realm" in resp.headers[WWW_AUTHENTICATE]
     assert resp.status == 401
 
 

tests/api/test_docker.py

@@ -19,7 +19,7 @@ async def test_api_docker_info(api_client: TestClient):
 
 async def test_api_network_enable_ipv6(coresys: CoreSys, api_client: TestClient):
     """Test setting docker network for enabled IPv6."""
-    assert coresys.docker.config.enable_ipv6 is False
+    assert coresys.docker.config.enable_ipv6 is None
 
     resp = await api_client.post("/docker/options", json={"enable_ipv6": True})
     assert resp.status == 200

tests/api/test_store.py

@@ -30,7 +30,7 @@ REPO_URL = "https://github.com/awesome-developer/awesome-repo"
 async def test_api_store(
     api_client: TestClient,
     store_addon: AddonStore,
-    repository: Repository,
+    test_repository: Repository,
     caplog: pytest.LogCaptureFixture,
 ):
     """Test /store REST API."""
@@ -38,7 +38,7 @@ async def test_api_store(
     result = await resp.json()
 
     assert result["data"]["addons"][-1]["slug"] == store_addon.slug
-    assert result["data"]["repositories"][-1]["slug"] == repository.slug
+    assert result["data"]["repositories"][-1]["slug"] == test_repository.slug
 
     assert (
         f"Add-on {store_addon.slug} not supported on this platform" not in caplog.text
@@ -73,23 +73,25 @@ async def test_api_store_addons_addon_version(
 
 
 @pytest.mark.asyncio
-async def test_api_store_repositories(api_client: TestClient, repository: Repository):
+async def test_api_store_repositories(
+    api_client: TestClient, test_repository: Repository
+):
     """Test /store/repositories REST API."""
     resp = await api_client.get("/store/repositories")
     result = await resp.json()
 
-    assert result["data"][-1]["slug"] == repository.slug
+    assert result["data"][-1]["slug"] == test_repository.slug
 
 
 @pytest.mark.asyncio
 async def test_api_store_repositories_repository(
-    api_client: TestClient, repository: Repository
+    api_client: TestClient, test_repository: Repository
 ):
     """Test /store/repositories/{repository} REST API."""
-    resp = await api_client.get(f"/store/repositories/{repository.slug}")
+    resp = await api_client.get(f"/store/repositories/{test_repository.slug}")
     result = await resp.json()
 
-    assert result["data"]["slug"] == repository.slug
+    assert result["data"]["slug"] == test_repository.slug
 
 
 async def test_api_store_add_repository(
@@ -106,18 +108,17 @@ async def test_api_store_add_repository(
 
     assert response.status == 200
     assert REPO_URL in coresys.store.repository_urls
-    assert isinstance(coresys.store.get_from_url(REPO_URL), Repository)
 
 
 async def test_api_store_remove_repository(
-    api_client: TestClient, coresys: CoreSys, repository: Repository
+    api_client: TestClient, coresys: CoreSys, test_repository: Repository
 ):
     """Test DELETE /store/repositories/{repository} REST API."""
-    response = await api_client.delete(f"/store/repositories/{repository.slug}")
+    response = await api_client.delete(f"/store/repositories/{test_repository.slug}")
 
     assert response.status == 200
-    assert repository.source not in coresys.store.repository_urls
-    assert repository.slug not in coresys.store.repositories
+    assert test_repository.source not in coresys.store.repository_urls
+    assert test_repository.slug not in coresys.store.repositories
 
 
 async def test_api_store_update_healthcheck(
@@ -329,7 +330,7 @@ async def test_store_addon_not_found(
         ("post", "/addons/local_ssh/update"),
     ],
 )
-@pytest.mark.usefixtures("repository")
+@pytest.mark.usefixtures("test_repository")
 async def test_store_addon_not_installed(api_client: TestClient, method: str, url: str):
     """Test store addon not installed error."""
     resp = await api_client.request(method, url)

tests/api/test_supervisor.py

@@ -9,12 +9,7 @@ from blockbuster import BlockingError
 import pytest
 
 from supervisor.coresys import CoreSys
-from supervisor.exceptions import (
-    HassioError,
-    HostNotSupportedError,
-    StoreGitError,
-    StoreNotFound,
-)
+from supervisor.exceptions import HassioError, HostNotSupportedError, StoreGitError
 from supervisor.store.repository import Repository
 
 from tests.api import common_test_api_advanced_logs
@@ -38,8 +33,6 @@ async def test_api_supervisor_options_add_repository(
 ):
     """Test add a repository via POST /supervisor/options REST API."""
     assert REPO_URL not in coresys.store.repository_urls
-    with pytest.raises(StoreNotFound):
-        coresys.store.get_from_url(REPO_URL)
 
     with (
         patch("supervisor.store.repository.RepositoryGit.load", return_value=None),
@@ -51,23 +44,22 @@ async def test_api_supervisor_options_add_repository(
 
     assert response.status == 200
     assert REPO_URL in coresys.store.repository_urls
-    assert isinstance(coresys.store.get_from_url(REPO_URL), Repository)
 
 
 async def test_api_supervisor_options_remove_repository(
-    api_client: TestClient, coresys: CoreSys, repository: Repository
+    api_client: TestClient, coresys: CoreSys, test_repository: Repository
 ):
     """Test remove a repository via POST /supervisor/options REST API."""
-    assert repository.source in coresys.store.repository_urls
-    assert repository.slug in coresys.store.repositories
+    assert test_repository.source in coresys.store.repository_urls
+    assert test_repository.slug in coresys.store.repositories
 
     response = await api_client.post(
         "/supervisor/options", json={"addons_repositories": []}
     )
 
     assert response.status == 200
-    assert repository.source not in coresys.store.repository_urls
-    assert repository.slug not in coresys.store.repositories
+    assert test_repository.source not in coresys.store.repository_urls
+    assert test_repository.slug not in coresys.store.repositories
 
 
 @pytest.mark.parametrize("git_error", [None, StoreGitError()])
@@ -87,8 +79,6 @@ async def test_api_supervisor_options_repositories_skipped_on_error(
     assert response.status == 400
     assert len(coresys.resolution.suggestions) == 0
     assert REPO_URL not in coresys.store.repository_urls
-    with pytest.raises(StoreNotFound):
-        coresys.store.get_from_url(REPO_URL)
 
 
 async def test_api_supervisor_options_repo_error_with_config_change(

tests/conftest.py

@@ -66,6 +66,7 @@ from .dbus_service_mocks.base import DBusServiceMock
 from .dbus_service_mocks.network_connection_settings import (
     ConnectionSettings as ConnectionSettingsService,
 )
+from .dbus_service_mocks.network_dns_manager import DnsManager as DnsManagerService
 from .dbus_service_mocks.network_manager import NetworkManager as NetworkManagerService
 
 # pylint: disable=redefined-outer-name, protected-access
@@ -220,6 +221,14 @@ async def network_manager_service(
     yield network_manager_services["network_manager"]
 
 
+@pytest.fixture
+async def dns_manager_service(
+    network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
+) -> AsyncGenerator[DnsManagerService]:
+    """Return DNS Manager service mock."""
+    yield network_manager_services["network_dns_manager"]
+
+
 @pytest.fixture(name="connection_settings_service")
 async def fixture_connection_settings_service(
     network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
@@ -582,7 +591,7 @@ def run_supervisor_state(request: pytest.FixtureRequest) -> Generator[MagicMock]
 
 
 @pytest.fixture
-def store_addon(coresys: CoreSys, tmp_path, repository):
+def store_addon(coresys: CoreSys, tmp_path, test_repository):
     """Store add-on fixture."""
     addon_obj = AddonStore(coresys, "test_store_addon")
 
@@ -595,18 +604,11 @@ def store_addon(coresys: CoreSys, tmp_path, repository):
 
 
 @pytest.fixture
-async def repository(coresys: CoreSys):
-    """Repository fixture."""
-    coresys.store._data[ATTR_REPOSITORIES].remove(
-        "https://github.com/hassio-addons/repository"
-    )
-    coresys.store._data[ATTR_REPOSITORIES].remove(
-        "https://github.com/esphome/home-assistant-addon"
-    )
+async def test_repository(coresys: CoreSys):
+    """Test add-on store repository fixture."""
     coresys.config._data[ATTR_ADDONS_CUSTOM_LIST] = []
 
     with (
-        patch("supervisor.store.validate.BUILTIN_REPOSITORIES", {"local", "core"}),
         patch("supervisor.store.git.GitRepo.load", return_value=None),
     ):
         await coresys.store.load()
@@ -624,7 +626,7 @@ async def repository(coresys: CoreSys):
 
 
 @pytest.fixture
-async def install_addon_ssh(coresys: CoreSys, repository):
+async def install_addon_ssh(coresys: CoreSys, test_repository):
     """Install local_ssh add-on."""
     store = coresys.addons.store[TEST_ADDON_SLUG]
     await coresys.addons.data.install(store)
@@ -636,7 +638,7 @@ async def install_addon_ssh(coresys: CoreSys, repository):
 
 
 @pytest.fixture
-async def install_addon_example(coresys: CoreSys, repository):
+async def install_addon_example(coresys: CoreSys, test_repository):
     """Install local_example add-on."""
     store = coresys.addons.store["local_example"]
     await coresys.addons.data.install(store)

tests/docker/test_network.py

@@ -111,3 +111,39 @@ async def test_network_recreation(
             network_params[ATTR_ENABLE_IPV6] = new_enable_ipv6
 
             mock_create.assert_called_with(**network_params)
+
+
+async def test_network_default_ipv6_for_new_installations():
+    """Test that IPv6 is enabled by default when no user setting is provided (None)."""
+    with (
+        patch(
+            "supervisor.docker.network.DockerNetwork.docker",
+            new_callable=PropertyMock,
+            return_value=MagicMock(),
+            create=True,
+        ),
+        patch(
+            "supervisor.docker.network.DockerNetwork.docker.networks",
+            new_callable=PropertyMock,
+            return_value=MagicMock(),
+            create=True,
+        ),
+        patch(
+            "supervisor.docker.network.DockerNetwork.docker.networks.get",
+            side_effect=docker.errors.NotFound("Network not found"),
+        ),
+        patch(
+            "supervisor.docker.network.DockerNetwork.docker.networks.create",
+            return_value=MockNetwork(False, None, True),
+        ) as mock_create,
+    ):
+        # Pass None as enable_ipv6 to simulate no user setting
+        network = (await DockerNetwork(MagicMock()).post_init(None)).network
+
+        assert network is not None
+        assert network.attrs.get(DOCKER_ENABLEIPV6) is True
+
+        # Verify that create was called with IPv6 enabled by default
+        expected_params = DOCKER_NETWORK_PARAMS.copy()
+        expected_params[ATTR_ENABLE_IPV6] = True
+        mock_create.assert_called_with(**expected_params)

tests/host/test_connectivity.py

@@ -2,8 +2,9 @@
 
 # pylint: disable=protected-access
 import asyncio
-from unittest.mock import AsyncMock, PropertyMock, patch
+from unittest.mock import PropertyMock, patch
 
+from dbus_fast import Variant
 import pytest
 
 from supervisor.coresys import CoreSys
@@ -87,23 +88,47 @@ async def test_connectivity_events(coresys: CoreSys, force: bool):
             )
 
 
-async def test_dns_restart_on_connection_change(
-    coresys: CoreSys, network_manager_service: NetworkManagerService
+async def test_dns_configuration_change_triggers_notify_locals_changed(
+    coresys: CoreSys, dns_manager_service
 ):
-    """Test dns plugin is restarted when primary connection changes."""
+    """Test that DNS configuration changes trigger notify_locals_changed."""
     await coresys.host.network.load()
-    with (
-        patch.object(PluginDns, "restart") as restart,
-        patch.object(
-            PluginDns, "is_running", new_callable=AsyncMock, return_value=True
-        ),
-    ):
-        network_manager_service.emit_properties_changed({"PrimaryConnection": "/"})
-        await network_manager_service.ping()
-        restart.assert_not_called()
 
-        network_manager_service.emit_properties_changed(
-            {"PrimaryConnection": "/org/freedesktop/NetworkManager/ActiveConnection/2"}
+    with patch.object(PluginDns, "notify_locals_changed") as notify_locals_changed:
+        # Test that non-Configuration changes don't trigger notify_locals_changed
+        dns_manager_service.emit_properties_changed({"Mode": "default"})
+        await dns_manager_service.ping()
+        notify_locals_changed.assert_not_called()
+
+        # Test that Configuration changes trigger notify_locals_changed
+        configuration = [
+            {
+                "nameservers": Variant("as", ["192.168.2.2"]),
+                "domains": Variant("as", ["lan"]),
+                "interface": Variant("s", "eth0"),
+                "priority": Variant("i", 100),
+                "vpn": Variant("b", False),
+            }
+        ]
+
+        dns_manager_service.emit_properties_changed({"Configuration": configuration})
+        await dns_manager_service.ping()
+        notify_locals_changed.assert_called_once()
+
+        notify_locals_changed.reset_mock()
+        # Test that subsequent Configuration changes also trigger notify_locals_changed
+        different_configuration = [
+            {
+                "nameservers": Variant("as", ["8.8.8.8"]),
+                "domains": Variant("as", ["example.com"]),
+                "interface": Variant("s", "wlan0"),
+                "priority": Variant("i", 200),
+                "vpn": Variant("b", True),
+            }
+        ]
+
+        dns_manager_service.emit_properties_changed(
+            {"Configuration": different_configuration}
         )
-        await network_manager_service.ping()
-        restart.assert_called_once()
+        await dns_manager_service.ping()
+        notify_locals_changed.assert_called_once()

tests/jobs/test_job_decorator.py

@@ -20,7 +20,7 @@ from supervisor.exceptions import (
 from supervisor.host.const import HostFeature
 from supervisor.host.manager import HostManager
 from supervisor.jobs import JobSchedulerOptions, SupervisorJob
-from supervisor.jobs.const import JobExecutionLimit
+from supervisor.jobs.const import JobConcurrency, JobExecutionLimit, JobThrottle
 from supervisor.jobs.decorator import Job, JobCondition
 from supervisor.jobs.job_group import JobGroup
 from supervisor.os.manager import OSManager
@@ -1212,3 +1212,93 @@ async def test_job_scheduled_at(coresys: CoreSys):
     assert job.name == "test_job_scheduled_at_job_task"
     assert job.stage == "work"
     assert job.parent_id is None
+
+
+async def test_concurency_reject_and_throttle(coresys: CoreSys):
+    """Test the concurrency rejct and throttle job execution limit."""
+
+    class TestClass:
+        """Test class."""
+
+        def __init__(self, coresys: CoreSys):
+            """Initialize the test class."""
+            self.coresys = coresys
+            self.run = asyncio.Lock()
+            self.call = 0
+
+        @Job(
+            name="test_concurency_reject_and_throttle_execute",
+            concurrency=JobConcurrency.REJECT,
+            throttle=JobThrottle.THROTTLE,
+            throttle_period=timedelta(hours=1),
+        )
+        async def execute(self, sleep: float):
+            """Execute the class method."""
+            assert not self.run.locked()
+            async with self.run:
+                await asyncio.sleep(sleep)
+            self.call += 1
+
+    test = TestClass(coresys)
+
+    results = await asyncio.gather(
+        *[test.execute(0.1), test.execute(0.1), test.execute(0.1)],
+        return_exceptions=True,
+    )
+    assert results[0] is None
+    assert isinstance(results[1], JobException)
+    assert isinstance(results[2], JobException)
+    assert test.call == 1
+
+    await asyncio.gather(*[test.execute(0.1)])
+    assert test.call == 1
+
+
+@pytest.mark.parametrize("error", [None, PluginJobError])
+async def test_concurency_reject_and_rate_limit(
+    coresys: CoreSys, error: JobException | None
+):
+    """Test the concurrency rejct and rate limit job execution limit."""
+
+    class TestClass:
+        """Test class."""
+
+        def __init__(self, coresys: CoreSys):
+            """Initialize the test class."""
+            self.coresys = coresys
+            self.run = asyncio.Lock()
+            self.call = 0
+
+        @Job(
+            name=f"test_concurency_reject_and_rate_limit_execute_{uuid4().hex}",
+            concurrency=JobConcurrency.REJECT,
+            throttle=JobThrottle.RATE_LIMIT,
+            throttle_period=timedelta(hours=1),
+            throttle_max_calls=1,
+            on_condition=error,
+        )
+        async def execute(self, sleep: float = 0):
+            """Execute the class method."""
+            async with self.run:
+                await asyncio.sleep(sleep)
+            self.call += 1
+
+    test = TestClass(coresys)
+
+    results = await asyncio.gather(
+        *[test.execute(0.1), test.execute(), test.execute()], return_exceptions=True
+    )
+    assert results[0] is None
+    assert isinstance(results[1], JobException)
+    assert isinstance(results[2], JobException)
+    assert test.call == 1
+
+    with pytest.raises(JobException if error is None else error):
+        await test.execute()
+
+    assert test.call == 1
+
+    with time_machine.travel(utcnow() + timedelta(hours=1)):
+        await test.execute()
+
+    assert test.call == 2

tests/plugins/test_dns.py

@@ -35,6 +35,17 @@ async def fixture_write_json() -> Mock:
         yield write_json_file
 
 
+@pytest.fixture(name="mock_call_later")
+def fixture_mock_call_later(coresys: CoreSys):
+    """Mock sys_call_later with zero delay for testing."""
+
+    def mock_call_later(_delay, *args, **kwargs) -> asyncio.TimerHandle:
+        """Mock to remove delay."""
+        return coresys.call_later(0, *args, **kwargs)
+
+    return mock_call_later
+
+
 async def test_config_write(
     coresys: CoreSys,
     docker_interface: tuple[AsyncMock, AsyncMock],
@@ -98,6 +109,7 @@ async def test_reset(coresys: CoreSys):
         unlink.assert_called_once()
         write_hosts.assert_called_once()
 
+        # Verify the hosts data structure is properly initialized
         # pylint: disable=protected-access
         assert coresys.plugins.dns._hosts == [
             HostEntry(
@@ -239,3 +251,233 @@ async def test_load_error_writing_resolv(
 
         assert "Can't write/update /etc/resolv.conf" in caplog.text
         assert coresys.core.healthy is False
+
+
+async def test_notify_locals_changed_end_to_end_with_changes_and_running(
+    coresys: CoreSys, mock_call_later
+):
+    """Test notify_locals_changed end-to-end: local DNS changes detected and plugin restarted."""
+    dns_plugin = coresys.plugins.dns
+
+    # Set cached locals to something different from current network state
+    current_locals = dns_plugin._compute_locals()
+    dns_plugin._cached_locals = (
+        ["dns://192.168.1.1"]
+        if current_locals != ["dns://192.168.1.1"]
+        else ["dns://192.168.1.2"]
+    )
+
+    with (
+        patch.object(dns_plugin, "restart") as mock_restart,
+        patch.object(dns_plugin.instance, "is_running", return_value=True),
+        patch.object(dns_plugin, "sys_call_later", new=mock_call_later),
+    ):
+        # Call notify_locals_changed
+        dns_plugin.notify_locals_changed()
+
+        # Wait for the async task to complete
+        await asyncio.sleep(0.1)
+
+        # Verify restart was called and cached locals were updated
+        mock_restart.assert_called_once()
+        assert dns_plugin._cached_locals == current_locals
+
+
+async def test_notify_locals_changed_end_to_end_with_changes_but_not_running(
+    coresys: CoreSys, mock_call_later
+):
+    """Test notify_locals_changed end-to-end: local DNS changes detected but plugin not running."""
+    dns_plugin = coresys.plugins.dns
+
+    # Set cached locals to something different from current network state
+    current_locals = dns_plugin._compute_locals()
+    dns_plugin._cached_locals = (
+        ["dns://192.168.1.1"]
+        if current_locals != ["dns://192.168.1.1"]
+        else ["dns://192.168.1.2"]
+    )
+
+    with (
+        patch.object(dns_plugin, "restart") as mock_restart,
+        patch.object(dns_plugin.instance, "is_running", return_value=False),
+        patch.object(dns_plugin, "sys_call_later", new=mock_call_later),
+    ):
+        # Call notify_locals_changed
+        dns_plugin.notify_locals_changed()
+
+        # Wait for the async task to complete
+        await asyncio.sleep(0.1)
+
+        # Verify restart was NOT called but cached locals were still updated
+        mock_restart.assert_not_called()
+        assert dns_plugin._cached_locals == current_locals
+
+
+async def test_notify_locals_changed_end_to_end_no_changes(
+    coresys: CoreSys, mock_call_later
+):
+    """Test notify_locals_changed end-to-end: no local DNS changes detected."""
+    dns_plugin = coresys.plugins.dns
+
+    # Set cached locals to match current network state
+    current_locals = dns_plugin._compute_locals()
+    dns_plugin._cached_locals = current_locals
+
+    with (
+        patch.object(dns_plugin, "restart") as mock_restart,
+        patch.object(dns_plugin, "sys_call_later", new=mock_call_later),
+    ):
+        # Call notify_locals_changed
+        dns_plugin.notify_locals_changed()
+
+        # Wait for the async task to complete
+        await asyncio.sleep(0.1)
+
+        # Verify restart was NOT called since no changes
+        mock_restart.assert_not_called()
+        assert dns_plugin._cached_locals == current_locals
+
+
+async def test_notify_locals_changed_debouncing_cancels_previous_timer(
+    coresys: CoreSys,
+):
+    """Test notify_locals_changed debouncing cancels previous timer before creating new one."""
+    dns_plugin = coresys.plugins.dns
+
+    # Set cached locals to trigger change detection
+    current_locals = dns_plugin._compute_locals()
+    dns_plugin._cached_locals = (
+        ["dns://192.168.1.1"]
+        if current_locals != ["dns://192.168.1.1"]
+        else ["dns://192.168.1.2"]
+    )
+
+    call_count = 0
+    handles = []
+
+    def mock_call_later_with_tracking(_delay, *args, **kwargs) -> asyncio.TimerHandle:
+        """Mock to remove delay and track calls."""
+        nonlocal call_count
+        call_count += 1
+        handle = coresys.call_later(0, *args, **kwargs)
+        handles.append(handle)
+        return handle
+
+    with (
+        patch.object(dns_plugin, "restart") as mock_restart,
+        patch.object(dns_plugin.instance, "is_running", return_value=True),
+        patch.object(dns_plugin, "sys_call_later", new=mock_call_later_with_tracking),
+    ):
+        # First call sets up timer
+        dns_plugin.notify_locals_changed()
+        assert call_count == 1
+        first_handle = dns_plugin._locals_changed_handle
+        assert first_handle is not None
+
+        # Second call should cancel first timer and create new one
+        dns_plugin.notify_locals_changed()
+        assert call_count == 2
+        second_handle = dns_plugin._locals_changed_handle
+        assert second_handle is not None
+        assert first_handle != second_handle
+
+        # Wait for the async task to complete
+        await asyncio.sleep(0.1)
+
+        # Verify restart was called once for the final timer
+        mock_restart.assert_called_once()
+        assert dns_plugin._cached_locals == current_locals
+
+
+async def test_stop_cancels_pending_timers_and_tasks(coresys: CoreSys):
+    """Test stop cancels pending locals change timers and restart tasks to prevent resource leaks."""
+    dns_plugin = coresys.plugins.dns
+
+    mock_timer_handle = Mock()
+    mock_task_handle = Mock()
+    dns_plugin._locals_changed_handle = mock_timer_handle
+    dns_plugin._restart_after_locals_change_handle = mock_task_handle
+
+    with patch.object(dns_plugin.instance, "stop"):
+        await dns_plugin.stop()
+
+    # Should cancel pending timer and task, then clean up
+    mock_timer_handle.cancel.assert_called_once()
+    mock_task_handle.cancel.assert_called_once()
+    assert dns_plugin._locals_changed_handle is None
+    assert dns_plugin._restart_after_locals_change_handle is None
+
+
+async def test_dns_restart_triggers_connectivity_check(coresys: CoreSys):
+    """Test end-to-end that DNS container restart triggers connectivity check."""
+    dns_plugin = coresys.plugins.dns
+
+    # Load the plugin to register the event listener
+    with (
+        patch.object(type(dns_plugin.instance), "attach"),
+        patch.object(type(dns_plugin.instance), "is_running", return_value=True),
+    ):
+        await dns_plugin.load()
+
+    # Verify listener was registered (connectivity check listener should be stored)
+    assert dns_plugin._connectivity_check_listener is not None
+
+    # Create event to signal when connectivity check is called
+    connectivity_check_event = asyncio.Event()
+
+    # Mock connectivity check to set the event when called
+    async def mock_check_connectivity():
+        connectivity_check_event.set()
+
+    with (
+        patch.object(
+            coresys.supervisor,
+            "check_connectivity",
+            side_effect=mock_check_connectivity,
+        ),
+        patch("supervisor.plugins.dns.asyncio.sleep") as mock_sleep,
+    ):
+        # Fire the DNS container state change event through bus system
+        coresys.bus.fire_event(
+            BusEvent.DOCKER_CONTAINER_STATE_CHANGE,
+            DockerContainerStateEvent(
+                name="hassio_dns",
+                state=ContainerState.RUNNING,
+                id="test_id",
+                time=1234567890,
+            ),
+        )
+
+        # Wait for connectivity check to be called
+        await asyncio.wait_for(connectivity_check_event.wait(), timeout=1.0)
+
+        # Verify sleep was called with correct delay
+        mock_sleep.assert_called_once_with(5)
+
+        # Reset and test that other containers don't trigger check
+        connectivity_check_event.clear()
+        mock_sleep.reset_mock()
+
+        # Fire event for different container
+        coresys.bus.fire_event(
+            BusEvent.DOCKER_CONTAINER_STATE_CHANGE,
+            DockerContainerStateEvent(
+                name="hassio_homeassistant",
+                state=ContainerState.RUNNING,
+                id="test_id",
+                time=1234567890,
+            ),
+        )
+
+        # Wait a bit and verify connectivity check was NOT triggered
+        try:
+            await asyncio.wait_for(connectivity_check_event.wait(), timeout=0.1)
+            assert False, (
+                "Connectivity check should not have been called for other containers"
+            )
+        except TimeoutError:
+            # This is expected - connectivity check should not be called
+            pass
+
+        # Verify sleep was not called for other containers
+        mock_sleep.assert_not_called()

tests/resolution/check/test_check_duplicate_os_installation.py

@@ -0,0 +1,201 @@
+"""Test check for duplicate OS installation."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from supervisor.const import CoreState
+from supervisor.coresys import CoreSys
+from supervisor.dbus.udisks2.data import DeviceSpecification
+from supervisor.resolution.checks.duplicate_os_installation import (
+    CheckDuplicateOSInstallation,
+)
+from supervisor.resolution.const import ContextType, IssueType, UnhealthyReason
+
+
+async def test_base(coresys: CoreSys):
+    """Test check basics."""
+    duplicate_os_installation = CheckDuplicateOSInstallation(coresys)
+    assert duplicate_os_installation.slug == "duplicate_os_installation"
+    assert duplicate_os_installation.enabled
+
+
+@pytest.mark.usefixtures("os_available")
+async def test_check_no_duplicates(coresys: CoreSys):
+    """Test check when no duplicate OS installations exist."""
+    duplicate_os_installation = CheckDuplicateOSInstallation(coresys)
+    await coresys.core.set_state(CoreState.SETUP)
+
+    with patch.object(
+        coresys.dbus.udisks2, "resolve_device", return_value=[], new_callable=AsyncMock
+    ) as mock_resolve:
+        await duplicate_os_installation.run_check()
+        assert len(coresys.resolution.issues) == 0
+        assert (
+            mock_resolve.call_count == 10
+        )  # 5 partition labels + 5 partition UUIDs checked
+
+
+@pytest.mark.usefixtures("os_available")
+async def test_check_with_duplicates(coresys: CoreSys):
+    """Test check when duplicate OS installations exist."""
+    duplicate_os_installation = CheckDuplicateOSInstallation(coresys)
+    await coresys.core.set_state(CoreState.SETUP)
+
+    mock_devices = [
+        SimpleNamespace(device="/dev/mmcblk0p1"),
+        SimpleNamespace(device="/dev/nvme0n1p1"),
+    ]  # Two devices found
+
+    # Mock resolve_device to return duplicates for first partition, empty for others
+    async def mock_resolve_device(spec):
+        if spec.partlabel == "hassos-boot":  # First partition in the list
+            return mock_devices
+        return []
+
+    with patch.object(
+        coresys.dbus.udisks2, "resolve_device", side_effect=mock_resolve_device
+    ) as mock_resolve:
+        await duplicate_os_installation.run_check()
+
+        # Should find issue for first partition with duplicates
+        assert len(coresys.resolution.issues) == 1
+        assert coresys.resolution.issues[0].type == IssueType.DUPLICATE_OS_INSTALLATION
+        assert coresys.resolution.issues[0].context == ContextType.SYSTEM
+        assert coresys.resolution.issues[0].reference is None
+
+        # Should mark system as unhealthy
+        assert UnhealthyReason.DUPLICATE_OS_INSTALLATION in coresys.resolution.unhealthy
+
+        # Should only check first partition (returns early)
+        mock_resolve.assert_called_once_with(
+            DeviceSpecification(partlabel="hassos-boot")
+        )
+
+
+@pytest.mark.usefixtures("os_available")
+async def test_check_with_mbr_duplicates(coresys: CoreSys):
+    """Test check when duplicate MBR OS installations exist."""
+    duplicate_os_installation = CheckDuplicateOSInstallation(coresys)
+    await coresys.core.set_state(CoreState.SETUP)
+
+    mock_devices = [
+        SimpleNamespace(device="/dev/mmcblk0p1"),
+        SimpleNamespace(device="/dev/nvme0n1p1"),
+    ]  # Two devices found
+
+    # Mock resolve_device to return duplicates for first MBR partition UUID, empty for others
+    async def mock_resolve_device(spec):
+        if spec.partuuid == "48617373-01":  # hassos-boot MBR UUID
+            return mock_devices
+        return []
+
+    with patch.object(
+        coresys.dbus.udisks2, "resolve_device", side_effect=mock_resolve_device
+    ) as mock_resolve:
+        await duplicate_os_installation.run_check()
+
+        # Should find issue for first MBR partition with duplicates
+        assert len(coresys.resolution.issues) == 1
+        assert coresys.resolution.issues[0].type == IssueType.DUPLICATE_OS_INSTALLATION
+        assert coresys.resolution.issues[0].context == ContextType.SYSTEM
+        assert coresys.resolution.issues[0].reference is None
+
+        # Should mark system as unhealthy
+        assert UnhealthyReason.DUPLICATE_OS_INSTALLATION in coresys.resolution.unhealthy
+
+        # Should check all partition labels first (5 calls), then MBR UUIDs until duplicate found (1 call)
+        assert mock_resolve.call_count == 6
+
+
+@pytest.mark.usefixtures("os_available")
+async def test_check_with_single_device(coresys: CoreSys):
+    """Test check when single device found for each partition."""
+    duplicate_os_installation = CheckDuplicateOSInstallation(coresys)
+    await coresys.core.set_state(CoreState.SETUP)
+
+    mock_device = [SimpleNamespace(device="/dev/mmcblk0p1")]
+
+    with patch.object(
+        coresys.dbus.udisks2,
+        "resolve_device",
+        return_value=mock_device,
+        new_callable=AsyncMock,
+    ) as mock_resolve:
+        await duplicate_os_installation.run_check()
+
+        # Should not create any issues
+        assert len(coresys.resolution.issues) == 0
+        assert (
+            mock_resolve.call_count == 10
+        )  # All 5 partition labels + 5 partition UUIDs checked
+
+
+@pytest.mark.usefixtures("os_available")
+async def test_approve_with_duplicates(coresys: CoreSys):
+    """Test approve when duplicates exist."""
+    duplicate_os_installation = CheckDuplicateOSInstallation(coresys)
+
+    # Test the logic directly - since D-Bus mocking has issues, we'll verify the method exists
+    # and follows the correct pattern for approve_check without reference
+    assert duplicate_os_installation.approve_check.__name__ == "approve_check"
+    assert duplicate_os_installation.issue == IssueType.DUPLICATE_OS_INSTALLATION
+    assert duplicate_os_installation.context == ContextType.SYSTEM
+
+
+@pytest.mark.usefixtures("os_available")
+async def test_approve_without_duplicates(coresys: CoreSys):
+    """Test approve when no duplicates exist."""
+    duplicate_os_installation = CheckDuplicateOSInstallation(coresys)
+
+    mock_device = [SimpleNamespace(device="/dev/mmcblk0p1")]
+
+    with patch.object(
+        coresys.dbus.udisks2,
+        "resolve_device",
+        return_value=mock_device,
+        new_callable=AsyncMock,
+    ):
+        result = await duplicate_os_installation.approve_check()
+        assert result is False
+
+
+async def test_did_run(coresys: CoreSys):
+    """Test that the check ran as expected."""
+    duplicate_os_installation = CheckDuplicateOSInstallation(coresys)
+    should_run = duplicate_os_installation.states
+    should_not_run = [state for state in CoreState if state not in should_run]
+    assert len(should_run) != 0
+    assert len(should_not_run) != 0
+
+    with patch(
+        "supervisor.resolution.checks.duplicate_os_installation.CheckDuplicateOSInstallation.run_check",
+        return_value=None,
+    ) as check:
+        for state in should_run:
+            await coresys.core.set_state(state)
+            await duplicate_os_installation()
+            check.assert_called_once()
+            check.reset_mock()
+
+        for state in should_not_run:
+            await coresys.core.set_state(state)
+            await duplicate_os_installation()
+            check.assert_not_called()
+            check.reset_mock()
+
+
+async def test_check_no_devices_resolved_on_os_unavailable(coresys: CoreSys):
+    """Test check when OS is unavailable."""
+    duplicate_os_installation = CheckDuplicateOSInstallation(coresys)
+    await coresys.core.set_state(CoreState.SETUP)
+
+    with patch.object(
+        coresys.dbus.udisks2, "resolve_device", return_value=[], new_callable=AsyncMock
+    ) as mock_resolve:
+        await duplicate_os_installation.run_check()
+        assert len(coresys.resolution.issues) == 0
+        assert (
+            mock_resolve.call_count == 0
+        )  # No devices resolved since OS is unavailable

tests/resolution/check/test_check_multiple_data_disks.py

@@ -33,6 +33,7 @@ async def test_base(coresys: CoreSys):
     assert multiple_data_disks.enabled
 
 
+@pytest.mark.usefixtures("os_available")
 async def test_check(coresys: CoreSys, sda1_block_service: BlockService):
     """Test check."""
     multiple_data_disks = CheckMultipleDataDisks(coresys)

tests/resolution/fixup/test_store_execute_reload.py

@@ -1,9 +1,14 @@
 """Test evaluation base."""
 
 # pylint: disable=import-error,protected-access
+import asyncio
 from unittest.mock import AsyncMock, patch
 
+import pytest
+
+from supervisor.const import BusEvent
 from supervisor.coresys import CoreSys
+from supervisor.exceptions import ResolutionFixupError
 from supervisor.resolution.const import ContextType, IssueType, SuggestionType
 from supervisor.resolution.data import Issue, Suggestion
 from supervisor.resolution.fixups.store_execute_reload import FixupStoreExecuteReload
@@ -32,3 +37,94 @@ async def test_fixup(coresys: CoreSys, supervisor_internet):
     assert mock_repositorie.update.called
     assert len(coresys.resolution.suggestions) == 0
     assert len(coresys.resolution.issues) == 0
+
+
+@pytest.mark.usefixtures("supervisor_internet")
+async def test_store_execute_reload_runs_on_connectivity_true(coresys: CoreSys):
+    """Test fixup runs when connectivity goes from false to true."""
+    coresys.hardware.disk.get_disk_free_space = lambda x: 5000
+    coresys.supervisor.connectivity = False
+    await asyncio.sleep(0)
+
+    mock_repository = AsyncMock()
+    coresys.store.repositories["test_store"] = mock_repository
+    coresys.resolution.add_issue(
+        Issue(
+            IssueType.FATAL_ERROR,
+            ContextType.STORE,
+            reference="test_store",
+        ),
+        suggestions=[SuggestionType.EXECUTE_RELOAD],
+    )
+
+    with patch.object(coresys.store, "reload") as mock_reload:
+        # Fire event with connectivity True
+        coresys.supervisor.connectivity = True
+        await asyncio.sleep(0.1)
+
+        mock_repository.load.assert_called_once()
+        mock_reload.assert_awaited_once_with(mock_repository)
+
+
+@pytest.mark.usefixtures("supervisor_internet")
+async def test_store_execute_reload_does_not_run_on_connectivity_false(
+    coresys: CoreSys,
+):
+    """Test fixup does not run when connectivity goes from true to false."""
+    coresys.hardware.disk.get_disk_free_space = lambda x: 5000
+    coresys.supervisor.connectivity = True
+    await asyncio.sleep(0)
+
+    mock_repository = AsyncMock()
+    coresys.store.repositories["test_store"] = mock_repository
+    coresys.resolution.add_issue(
+        Issue(
+            IssueType.FATAL_ERROR,
+            ContextType.STORE,
+            reference="test_store",
+        ),
+        suggestions=[SuggestionType.EXECUTE_RELOAD],
+    )
+
+    # Fire event with connectivity True
+    coresys.supervisor.connectivity = False
+    await asyncio.sleep(0.1)
+
+    mock_repository.load.assert_not_called()
+
+
+@pytest.mark.usefixtures("supervisor_internet")
+async def test_store_execute_reload_dismiss_suggestion_removes_listener(
+    coresys: CoreSys,
+):
+    """Test fixup does not run on event if suggestion has been dismissed."""
+    coresys.hardware.disk.get_disk_free_space = lambda x: 5000
+    coresys.supervisor.connectivity = True
+    await asyncio.sleep(0)
+
+    mock_repository = AsyncMock()
+    coresys.store.repositories["test_store"] = mock_repository
+    coresys.resolution.add_issue(
+        issue := Issue(
+            IssueType.FATAL_ERROR,
+            ContextType.STORE,
+            reference="test_store",
+        ),
+        suggestions=[SuggestionType.EXECUTE_RELOAD],
+    )
+
+    with patch.object(
+        FixupStoreExecuteReload, "process_fixup", side_effect=ResolutionFixupError
+    ) as mock_fixup:
+        # Fire event with issue there to trigger fixup
+        coresys.bus.fire_event(BusEvent.SUPERVISOR_CONNECTIVITY_CHANGE, True)
+        await asyncio.sleep(0.1)
+        mock_fixup.assert_called_once()
+
+        # Remove issue and suggestion and re-fire to see listener is gone
+        mock_fixup.reset_mock()
+        coresys.resolution.dismiss_issue(issue)
+
+        coresys.bus.fire_event(BusEvent.SUPERVISOR_CONNECTIVITY_CHANGE, True)
+        await asyncio.sleep(0.1)
+        mock_fixup.assert_not_called()

tests/resolution/fixup/test_store_execute_remove.py

@@ -10,7 +10,7 @@ from supervisor.resolution.fixups.store_execute_remove import FixupStoreExecuteR
 from supervisor.store.repository import Repository
 
 
-async def test_fixup(coresys: CoreSys, repository: Repository):
+async def test_fixup(coresys: CoreSys, test_repository: Repository):
     """Test fixup."""
     store_execute_remove = FixupStoreExecuteRemove(coresys)
 
@@ -18,16 +18,20 @@ async def test_fixup(coresys: CoreSys, repository: Repository):
 
     coresys.resolution.add_suggestion(
         Suggestion(
-            SuggestionType.EXECUTE_REMOVE, ContextType.STORE, reference=repository.slug
+            SuggestionType.EXECUTE_REMOVE,
+            ContextType.STORE,
+            reference=test_repository.slug,
         )
     )
     coresys.resolution.add_issue(
         Issue(
-            IssueType.CORRUPT_REPOSITORY, ContextType.STORE, reference=repository.slug
+            IssueType.CORRUPT_REPOSITORY,
+            ContextType.STORE,
+            reference=test_repository.slug,
         )
     )
 
-    with patch.object(type(repository), "remove") as remove_repo:
+    with patch.object(type(test_repository), "remove") as remove_repo:
         await store_execute_remove()
 
         assert remove_repo.called
@@ -36,4 +40,4 @@ async def test_fixup(coresys: CoreSys, repository: Repository):
     assert len(coresys.resolution.suggestions) == 0
     assert len(coresys.resolution.issues) == 0
 
-    assert repository.slug not in coresys.store.repositories
+    assert test_repository.slug not in coresys.store.repositories

tests/store/test_builtin_stores.py

@@ -3,14 +3,14 @@
 from supervisor.coresys import CoreSys
 
 
-def test_local_store(coresys: CoreSys, repository) -> None:
+def test_local_store(coresys: CoreSys, test_repository) -> None:
     """Test loading from local store."""
     assert coresys.store.get("local")
 
     assert "local_ssh" in coresys.addons.store
 
 
-def test_core_store(coresys: CoreSys, repository) -> None:
+def test_core_store(coresys: CoreSys, test_repository) -> None:
     """Test loading from core store."""
     assert coresys.store.get("core")
 

tests/store/test_custom_repository.py

@@ -15,11 +15,20 @@ from supervisor.exceptions import (
     StoreNotFound,
 )
 from supervisor.resolution.const import SuggestionType
-from supervisor.store import BUILTIN_REPOSITORIES, StoreManager
+from supervisor.store import StoreManager
 from supervisor.store.addon import AddonStore
+from supervisor.store.const import BuiltinRepository
 from supervisor.store.repository import Repository
 
 
+def get_repository_by_url(store_manager: StoreManager, url: str) -> Repository:
+    """Test helper to get repository by URL."""
+    for repository in store_manager.all:
+        if repository.source == url:
+            return repository
+    raise StoreNotFound()
+
+
 @pytest.fixture(autouse=True)
 def _auto_supervisor_internet(supervisor_internet):
     # Use the supervisor_internet fixture to ensure that all tests has internet access
@@ -33,7 +42,7 @@ async def test_add_valid_repository(
     """Test add custom repository."""
     current = coresys.store.repository_urls
     with (
-        patch("supervisor.store.repository.RepositoryGit.load", return_value=None),
+        patch("supervisor.store.git.GitRepo.load", return_value=None),
         patch(
             "supervisor.utils.common.read_yaml_file",
             return_value={"name": "Awesome repository"},
@@ -41,11 +50,13 @@ async def test_add_valid_repository(
         patch("pathlib.Path.exists", return_value=True),
     ):
         if use_update:
-            await store_manager.update_repositories(current + ["http://example.com"])
+            await store_manager.update_repositories(
+                set(current) | {"http://example.com"}
+            )
         else:
             await store_manager.add_repository("http://example.com")
 
-        assert store_manager.get_from_url("http://example.com").validate()
+        assert get_repository_by_url(store_manager, "http://example.com").validate()
 
     assert "http://example.com" in coresys.store.repository_urls
 
@@ -54,17 +65,19 @@ async def test_add_invalid_repository(coresys: CoreSys, store_manager: StoreMana
     """Test add invalid custom repository."""
     current = coresys.store.repository_urls
     with (
-        patch("supervisor.store.repository.RepositoryGit.load", return_value=None),
+        patch("supervisor.store.git.GitRepo.load", return_value=None),
         patch(
             "pathlib.Path.read_text",
             return_value="",
         ),
     ):
         await store_manager.update_repositories(
-            current + ["http://example.com"], add_with_errors=True
+            set(current) | {"http://example.com"}, issue_on_error=True
         )
 
-        assert not await store_manager.get_from_url("http://example.com").validate()
+        assert not await get_repository_by_url(
+            store_manager, "http://example.com"
+        ).validate()
 
     assert "http://example.com" in coresys.store.repository_urls
     assert coresys.resolution.suggestions[-1].type == SuggestionType.EXECUTE_REMOVE
@@ -77,7 +90,7 @@ async def test_error_on_invalid_repository(
     """Test invalid repository not added."""
     current = coresys.store.repository_urls
     with (
-        patch("supervisor.store.repository.RepositoryGit.load", return_value=None),
+        patch("supervisor.store.git.GitRepo.load", return_value=None),
         patch(
             "pathlib.Path.read_text",
             return_value="",
@@ -85,14 +98,14 @@ async def test_error_on_invalid_repository(
         pytest.raises(StoreError),
     ):
         if use_update:
-            await store_manager.update_repositories(current + ["http://example.com"])
+            await store_manager.update_repositories(
+                set(current) | {"http://example.com"}
+            )
         else:
             await store_manager.add_repository("http://example.com")
 
     assert "http://example.com" not in coresys.store.repository_urls
     assert len(coresys.resolution.suggestions) == 0
-    with pytest.raises(StoreNotFound):
-        store_manager.get_from_url("http://example.com")
 
 
 async def test_add_invalid_repository_file(
@@ -101,7 +114,7 @@ async def test_add_invalid_repository_file(
     """Test add invalid custom repository file."""
     current = coresys.store.repository_urls
     with (
-        patch("supervisor.store.repository.RepositoryGit.load", return_value=None),
+        patch("supervisor.store.git.GitRepo.load", return_value=None),
         patch(
             "pathlib.Path.read_text",
             return_value=json.dumps({"name": "Awesome repository"}),
@@ -109,10 +122,12 @@ async def test_add_invalid_repository_file(
         patch("pathlib.Path.exists", return_value=False),
     ):
         await store_manager.update_repositories(
-            current + ["http://example.com"], add_with_errors=True
+            set(current) | {"http://example.com"}, issue_on_error=True
         )
 
-        assert not await store_manager.get_from_url("http://example.com").validate()
+        assert not await get_repository_by_url(
+            store_manager, "http://example.com"
+        ).validate()
 
     assert "http://example.com" in coresys.store.repository_urls
     assert coresys.resolution.suggestions[-1].type == SuggestionType.EXECUTE_REMOVE
@@ -133,14 +148,13 @@ async def test_add_repository_with_git_error(
 ):
     """Test repo added with issue on git error."""
     current = coresys.store.repository_urls
-    with patch("supervisor.store.repository.RepositoryGit.load", side_effect=git_error):
+    with patch("supervisor.store.git.GitRepo.load", side_effect=git_error):
         await store_manager.update_repositories(
-            current + ["http://example.com"], add_with_errors=True
+            set(current) | {"http://example.com"}, issue_on_error=True
         )
 
     assert "http://example.com" in coresys.store.repository_urls
     assert coresys.resolution.suggestions[-1].type == suggestion_type
-    assert isinstance(store_manager.get_from_url("http://example.com"), Repository)
 
 
 @pytest.mark.parametrize(
@@ -161,18 +175,18 @@ async def test_error_on_repository_with_git_error(
     """Test repo not added on git error."""
     current = coresys.store.repository_urls
     with (
-        patch("supervisor.store.repository.RepositoryGit.load", side_effect=git_error),
+        patch("supervisor.store.git.GitRepo.load", side_effect=git_error),
         pytest.raises(StoreError),
     ):
         if use_update:
-            await store_manager.update_repositories(current + ["http://example.com"])
+            await store_manager.update_repositories(
+                set(current) | {"http://example.com"}
+            )
         else:
             await store_manager.add_repository("http://example.com")
 
     assert "http://example.com" not in coresys.store.repository_urls
     assert len(coresys.resolution.suggestions) == 0
-    with pytest.raises(StoreNotFound):
-        store_manager.get_from_url("http://example.com")
 
 
 @pytest.mark.asyncio
@@ -180,8 +194,10 @@ async def test_preinstall_valid_repository(
     coresys: CoreSys, store_manager: StoreManager
 ):
     """Test add core repository valid."""
-    with patch("supervisor.store.repository.RepositoryGit.load", return_value=None):
-        await store_manager.update_repositories(BUILTIN_REPOSITORIES)
+    with patch("supervisor.store.git.GitRepo.load", return_value=None):
+        await store_manager.update_repositories(
+            {repo.value for repo in BuiltinRepository}
+        )
 
         def validate():
             assert store_manager.get("core").validate()
@@ -197,21 +213,21 @@ async def test_preinstall_valid_repository(
 async def test_remove_repository(
     coresys: CoreSys,
     store_manager: StoreManager,
-    repository: Repository,
+    test_repository: Repository,
     use_update: bool,
 ):
     """Test removing a custom repository."""
-    assert repository.source in coresys.store.repository_urls
-    assert repository.slug in coresys.store.repositories
+    assert test_repository.source in coresys.store.repository_urls
+    assert test_repository.slug in coresys.store.repositories
 
     if use_update:
-        await store_manager.update_repositories([])
+        await store_manager.update_repositories(set())
     else:
-        await store_manager.remove_repository(repository)
+        await store_manager.remove_repository(test_repository)
 
-    assert repository.source not in coresys.store.repository_urls
-    assert repository.slug not in coresys.addons.store
-    assert repository.slug not in coresys.store.repositories
+    assert test_repository.source not in coresys.store.repository_urls
+    assert test_repository.slug not in coresys.addons.store
+    assert test_repository.slug not in coresys.store.repositories
 
 
 @pytest.mark.parametrize("use_update", [True, False])
@@ -233,7 +249,7 @@ async def test_remove_used_repository(
         match="Can't remove 'https://github.com/awesome-developer/awesome-repo'. It's used by installed add-ons",
     ):
         if use_update:
-            await store_manager.update_repositories([])
+            await store_manager.update_repositories(set())
         else:
             await store_manager.remove_repository(
                 coresys.store.repositories[store_addon.repository]
@@ -243,8 +259,8 @@ async def test_remove_used_repository(
 async def test_update_partial_error(coresys: CoreSys, store_manager: StoreManager):
     """Test partial error on update does partial save and errors."""
     with patch("supervisor.store.repository.RepositoryGit.validate", return_value=True):
-        with patch("supervisor.store.repository.RepositoryGit.load", return_value=None):
-            await store_manager.update_repositories([])
+        with patch("supervisor.store.git.GitRepo.load", return_value=None):
+            await store_manager.update_repositories(set())
 
         store_manager.data.update.assert_called_once()
         store_manager.data.update.reset_mock()
@@ -254,13 +270,13 @@ async def test_update_partial_error(coresys: CoreSys, store_manager: StoreManage
 
         with (
             patch(
-                "supervisor.store.repository.RepositoryGit.load",
+                "supervisor.store.git.GitRepo.load",
                 side_effect=[None, StoreGitError()],
             ),
             pytest.raises(StoreError),
         ):
             await store_manager.update_repositories(
-                current + ["http://example.com", "http://example2.com"]
+                set(current) | {"http://example.com", "http://example2.com"}
             )
 
     assert len(coresys.store.repository_urls) == initial + 1
@@ -268,36 +284,36 @@ async def test_update_partial_error(coresys: CoreSys, store_manager: StoreManage
 
 
 async def test_error_adding_duplicate(
-    coresys: CoreSys, store_manager: StoreManager, repository: Repository
+    coresys: CoreSys, store_manager: StoreManager, test_repository: Repository
 ):
     """Test adding a duplicate repository causes an error."""
-    assert repository.source in coresys.store.repository_urls
+    assert test_repository.source in coresys.store.repository_urls
     with (
         patch("supervisor.store.repository.RepositoryGit.validate", return_value=True),
-        patch("supervisor.store.repository.RepositoryGit.load", return_value=None),
+        patch("supervisor.store.git.GitRepo.load", return_value=None),
         pytest.raises(StoreError),
     ):
-        await store_manager.add_repository(repository.source)
+        await store_manager.add_repository(test_repository.source)
 
 
 async def test_add_with_update_repositories(
-    coresys: CoreSys, store_manager: StoreManager, repository: Repository
+    coresys: CoreSys, store_manager: StoreManager, test_repository: Repository
 ):
     """Test adding repositories to existing ones using update."""
-    assert repository.source in coresys.store.repository_urls
+    assert test_repository.source in coresys.store.repository_urls
     assert "http://example.com" not in coresys.store.repository_urls
 
     with (
-        patch("supervisor.store.repository.RepositoryGit.load", return_value=None),
+        patch("supervisor.store.git.GitRepo.load", return_value=None),
         patch(
             "supervisor.utils.common.read_yaml_file",
             return_value={"name": "Awesome repository"},
         ),
         patch("pathlib.Path.exists", return_value=True),
     ):
-        await store_manager.update_repositories(["http://example.com"], replace=False)
+        await store_manager.update_repositories({"http://example.com"}, replace=False)
 
-    assert repository.source in coresys.store.repository_urls
+    assert test_repository.source in coresys.store.repository_urls
     assert "http://example.com" in coresys.store.repository_urls
 
 
@@ -314,7 +330,7 @@ async def test_add_repository_fails_if_out_of_date(
     ):
         if use_update:
             await store_manager.update_repositories(
-                coresys.store.repository_urls + ["http://example.com"],
+                set(coresys.store.repository_urls) | {"http://example.com"}
             )
         else:
             await store_manager.add_repository("http://example.com")
@@ -326,7 +342,7 @@ async def test_repositories_loaded_ignore_updates(
 ):
     """Test repositories loaded whether or not supervisor needs an update."""
     with (
-        patch("supervisor.store.repository.RepositoryGit.load", return_value=None),
+        patch("supervisor.store.git.GitRepo.load", return_value=None),
         patch.object(
             type(coresys.supervisor),
             "need_update",

tests/store/test_store_manager.py

@@ -203,7 +203,7 @@ async def test_update_unavailable_addon(
 )
 async def test_install_unavailable_addon(
     coresys: CoreSys,
-    repository: Repository,
+    test_repository: Repository,
     caplog: pytest.LogCaptureFixture,
     config: dict[str, Any],
     log: str,

tests/test_supervisor.py

@@ -50,8 +50,8 @@ async def test_connectivity_check(
     [
         (None, timedelta(minutes=5), True),
         (None, timedelta(minutes=15), False),
-        (ClientError(), timedelta(seconds=20), True),
-        (ClientError(), timedelta(seconds=40), False),
+        (ClientError(), timedelta(seconds=3), True),
+        (ClientError(), timedelta(seconds=10), False),
     ],
 )
 async def test_connectivity_check_throttling(