Use Unix socket for Supervisor to Core communication

Switch internal Supervisor→Core HTTP/WebSocket communication to use a
Unix socket instead of TCP when the installed Core version supports it.

The existing /run/supervisor directory on the host (mounted at /run/os
inside Supervisor) is bind-mounted into the Core container as
/run/supervisor. Core receives the SUPERVISOR_CORE_API_SOCKET environment
variable telling it where to create the socket. Supervisor connects to
it via aiohttp.UnixConnector at /run/os/core.sock.

All session and URL management lives in HomeAssistantAPI as private
properties (_session, _api_url, _ws_url) with a version-gated fallback
to the existing TCP path for older Core versions.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

supervisor/api/proxy.py

@@ -179,10 +179,10 @@ class APIProxy(CoreSysAttributes):
 
     async def _websocket_client(self) -> ClientWebSocketResponse:
         """Initialize a WebSocket API connection."""
-        url = f"{self.sys_homeassistant.api_url}/api/websocket"
+        url = f"{self.sys_homeassistant.api._api_url}/api/websocket"
 
         try:
-            client = await self.sys_websession.ws_connect(
+            client = await self.sys_homeassistant.api._session.ws_connect(
                 url, heartbeat=30, ssl=False, max_msg_size=MAX_MESSAGE_SIZE_FROM_CORE
             )
 

supervisor/const.py

@@ -39,9 +39,10 @@ FILE_HASSIO_SECURITY = Path(SUPERVISOR_DATA, "security.json")
 FILE_SUFFIX_CONFIGURATION = [".yaml", ".yml", ".json"]
 
 MACHINE_ID = Path("/etc/machine-id")
+RUN_SUPERVISOR_STATE = Path("/run/supervisor")
+SOCKET_CORE = Path("/run/os/core.sock")
 SOCKET_DBUS = Path("/run/dbus/system_bus_socket")
 SOCKET_DOCKER = Path("/run/docker.sock")
-RUN_SUPERVISOR_STATE = Path("/run/supervisor")
 SYSTEMD_JOURNAL_PERSISTENT = Path("/var/log/journal")
 SYSTEMD_JOURNAL_VOLATILE = Path("/run/log/journal")
 

supervisor/core.py

@@ -337,6 +337,7 @@ class Core(CoreSysAttributes):
                         self.sys_create_task(coro)
                         for coro in (
                             self.sys_websession.close(),
+                            self.sys_homeassistant.api.close(),
                             self.sys_ingress.unload(),
                             self.sys_hardware.unload(),
                             self.sys_dbus.unload(),

supervisor/docker/const.py

@@ -89,6 +89,7 @@ class MountBindOptions:
 
     propagation: PropagationMode | None = None
     read_only_non_recursive: bool | None = None
+    create_mountpoint: bool | None = None
 
     def to_dict(self) -> dict[str, Any]:
         """To dictionary representation."""
@@ -97,6 +98,8 @@ class MountBindOptions:
             out["Propagation"] = self.propagation.value
         if self.read_only_non_recursive is not None:
             out["ReadOnlyNonRecursive"] = self.read_only_non_recursive
+        if self.create_mountpoint is not None:
+            out["CreateMountpoint"] = self.create_mountpoint
         return out
 
 
@@ -140,6 +143,7 @@ class Ulimit:
         }
 
 
+ENV_CORE_API_SOCKET = "SUPERVISOR_CORE_API_SOCKET"
 ENV_DUPLICATE_LOG_FILE = "HA_DUPLICATE_LOG_FILE"
 ENV_TIME = "TZ"
 ENV_TOKEN = "SUPERVISOR_TOKEN"
@@ -169,6 +173,12 @@ MOUNT_MACHINE_ID = DockerMount(
     target=MACHINE_ID.as_posix(),
     read_only=True,
 )
+MOUNT_CORE_RUN = DockerMount(
+    type=MountType.BIND,
+    source="/run/supervisor",
+    target="/run/supervisor",
+    read_only=False,
+)
 MOUNT_UDEV = DockerMount(
     type=MountType.BIND, source="/run/udev", target="/run/udev", read_only=True
 )

supervisor/docker/homeassistant.py

@@ -13,10 +13,12 @@ from ..homeassistant.const import LANDINGPAGE
 from ..jobs.const import JobConcurrency
 from ..jobs.decorator import Job
 from .const import (
+    ENV_CORE_API_SOCKET,
     ENV_DUPLICATE_LOG_FILE,
     ENV_TIME,
     ENV_TOKEN,
     ENV_TOKEN_OLD,
+    MOUNT_CORE_RUN,
     MOUNT_DBUS,
     MOUNT_DEV,
     MOUNT_MACHINE_ID,
@@ -136,6 +138,8 @@ class DockerHomeAssistant(DockerInterface):
                             propagation=PropagationMode.RSLAVE
                         ),
                     ),
+                    # Supervisor <-> Core communication socket
+                    MOUNT_CORE_RUN,
                     # Configuration audio
                     DockerMount(
                         type=MountType.BIND,
@@ -180,6 +184,8 @@ class DockerHomeAssistant(DockerInterface):
         }
         if restore_job_id:
             environment[ENV_RESTORE_JOB_ID] = restore_job_id
+        if self.sys_homeassistant.api.use_unix_socket:
+            environment[ENV_CORE_API_SOCKET] = "/run/supervisor/core.sock"
         if self.sys_homeassistant.duplicate_log_file:
             environment[ENV_DUPLICATE_LOG_FILE] = "1"
         await self._run(

supervisor/homeassistant/api.py

@@ -13,6 +13,7 @@ from aiohttp import hdrs
 from awesomeversion import AwesomeVersion
 from multidict import MultiMapping
 
+from ..const import SOCKET_CORE
 from ..coresys import CoreSys, CoreSysAttributes
 from ..exceptions import HomeAssistantAPIError, HomeAssistantAuthError
 from ..utils import version_is_new_enough
@@ -20,6 +21,9 @@ from .const import LANDINGPAGE
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
 
+CORE_UNIX_SOCKET_MIN_VERSION: AwesomeVersion = AwesomeVersion(
+    "2026.3.0.dev202602160311"
+)
 GET_CORE_STATE_MIN_VERSION: AwesomeVersion = AwesomeVersion("2023.8.0.dev20230720")
 
 
@@ -42,6 +46,54 @@ class HomeAssistantAPI(CoreSysAttributes):
         self.access_token: str | None = None
         self._access_token_expires: datetime | None = None
         self._token_lock: asyncio.Lock = asyncio.Lock()
+        self._unix_session: aiohttp.ClientSession | None = None
+
+    @property
+    def use_unix_socket(self) -> bool:
+        """Return True if Core supports Unix socket communication."""
+        return (
+            self.sys_homeassistant.version is not None
+            and self.sys_homeassistant.version != LANDINGPAGE
+            and version_is_new_enough(
+                self.sys_homeassistant.version, CORE_UNIX_SOCKET_MIN_VERSION
+            )
+        )
+
+    @property
+    def _session(self) -> aiohttp.ClientSession:
+        """Return session for Core communication.
+
+        Uses a Unix socket session when the installed Core version supports it,
+        otherwise falls back to the default TCP websession.
+        """
+        if not self.use_unix_socket:
+            return self.sys_websession
+
+        if self._unix_session is None or self._unix_session.closed:
+            self._unix_session = aiohttp.ClientSession(
+                connector=aiohttp.UnixConnector(path=str(SOCKET_CORE))
+            )
+        return self._unix_session
+
+    @property
+    def _api_url(self) -> str:
+        """Return API base url for internal Supervisor to Core communication."""
+        if self.use_unix_socket:
+            return "http://localhost"
+        return self.sys_homeassistant.api_url
+
+    @property
+    def _ws_url(self) -> str:
+        """Return WebSocket url for internal Supervisor to Core communication."""
+        if self.use_unix_socket:
+            return "ws://localhost/api/websocket"
+        return self.sys_homeassistant.ws_url
+
+    async def close(self) -> None:
+        """Close the Unix socket session."""
+        if self._unix_session and not self._unix_session.closed:
+            await self._unix_session.close()
+            self._unix_session = None
 
     async def ensure_access_token(self) -> None:
         """Ensure there is a valid access token.
@@ -70,8 +122,8 @@ class HomeAssistantAPI(CoreSysAttributes):
             ):
                 return
 
-            async with self.sys_websession.post(
-                f"{self.sys_homeassistant.api_url}/auth/token",
+            async with self._session.post(
+                f"{self._api_url}/auth/token",
                 timeout=aiohttp.ClientTimeout(total=30),
                 data={
                     "grant_type": "refresh_token",
@@ -133,7 +185,7 @@ class HomeAssistantAPI(CoreSysAttributes):
                 network errors, timeouts, or connection failures
 
         """
-        url = f"{self.sys_homeassistant.api_url}/{path}"
+        url = f"{self._api_url}/{path}"
         headers = headers or {}
         client_timeout = aiohttp.ClientTimeout(total=timeout)
 
@@ -145,7 +197,7 @@ class HomeAssistantAPI(CoreSysAttributes):
             try:
                 await self.ensure_access_token()
                 headers[hdrs.AUTHORIZATION] = f"Bearer {self.access_token}"
-                async with self.sys_websession.request(
+                async with self._session.request(
                     method,
                     url,
                     data=data,

supervisor/homeassistant/websocket.py

@@ -189,8 +189,8 @@ class HomeAssistantWebSocket(CoreSysAttributes):
             with suppress(asyncio.TimeoutError, aiohttp.ClientError):
                 await self.sys_homeassistant.api.ensure_access_token()
             client = await WSClient.connect_with_auth(
-                self.sys_websession,
-                self.sys_homeassistant.ws_url,
+                self.sys_homeassistant.api._session,
+                self.sys_homeassistant.api._ws_url,
                 cast(str, self.sys_homeassistant.api.access_token),
             )
 

tests/docker/test_homeassistant.py

@@ -9,6 +9,7 @@ import pytest
 
 from supervisor.coresys import CoreSys
 from supervisor.docker.const import (
+    MOUNT_CORE_RUN,
     DockerMount,
     MountBindOptions,
     MountType,
@@ -93,6 +94,7 @@ async def test_homeassistant_start(coresys: CoreSys, container: DockerContainer)
                 read_only=False,
                 bind_options=MountBindOptions(propagation=PropagationMode.RSLAVE),
             ),
+            MOUNT_CORE_RUN,
             DockerMount(
                 type=MountType.BIND,
                 source=coresys.homeassistant.path_extern_pulse.as_posix(),
@@ -144,6 +146,28 @@ async def test_homeassistant_start_with_duplicate_log_file(
         assert env["HA_DUPLICATE_LOG_FILE"] == "1"
 
 
+@pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
+async def test_homeassistant_start_with_unix_socket(
+    coresys: CoreSys, container: DockerContainer
+):
+    """Test starting homeassistant with unix socket env var for supported version."""
+    coresys.homeassistant.version = AwesomeVersion("2026.4.0")
+
+    with (
+        patch.object(DockerAPI, "run", return_value=container.show.return_value) as run,
+        patch.object(
+            DockerHomeAssistant, "is_running", side_effect=[False, False, True]
+        ),
+        patch("supervisor.homeassistant.core.asyncio.sleep"),
+    ):
+        await coresys.homeassistant.core.start()
+
+        run.assert_called_once()
+        env = run.call_args.kwargs["environment"]
+        assert "SUPERVISOR_CORE_API_SOCKET" in env
+        assert env["SUPERVISOR_CORE_API_SOCKET"] == "/run/supervisor/core.sock"
+
+
 @pytest.mark.usefixtures("tmp_supervisor_data", "path_extern")
 async def test_landingpage_start(coresys: CoreSys, container: DockerContainer):
     """Test starting landingpage."""