Traverse `BaseExceptionGroup` peer-close errors

`_peer_closed_errno()` followed cause and context links but did not
descend through grouped exceptions. A reset below a group could
therefore escape as `trio.BrokenResourceError` instead of the
normalized `TransportClosed` boundary.

Walk the exception tree with cycle protection, requiring every
group branch to represent peer closure before normalization. Extend
the `MsgpackTransport.send()` regression to prove all-transport and
mixed-failure behavior.

Review: PR #480 (goodboy)
https://github.com/goodboy/tractor/pull/480

(this patch was generated in some part by `opencode` using
`gpt-5.6-sol` (`openai`))

tests/discovery/test_daemon_fixture.py

@@ -1,6 +1,10 @@
 '''
 Discovery daemon fixture regressions.
 
+This module imports private helpers from the sibling
+`tests.discovery.conftest` plugin to exercise that fixture machinery
+directly, rather than testing a production `tractor` API.
+
 '''
 from unittest.mock import (
     call,

tests/discovery/test_tpt_bind_addrs.py

@@ -2,7 +2,8 @@
 `open_root_actor(tpt_bind_addrs=...)` test suite.
 
 Verify all three runtime code paths for explicit IPC-server
-bind-address selection in `_root.py`:
+bind-address selection in `_root.py` and registry probing in
+`discovery._api`:
 
 1. Non-registrar, no explicit bind -> random addrs from registry proto
 2. Registrar, no explicit bind -> binds to registry_addrs
@@ -19,11 +20,12 @@ from unittest.mock import (
 import pytest
 import trio
 import tractor
-from tractor import _root
+from tractor.discovery import _api
 from tractor.discovery._addr import (
     wrap_address,
 )
 from tractor.discovery._multiaddr import mk_maddr
+from tractor.ipc import _connect_chan
 from tractor._testing.addr import get_rando_addr
 
 
@@ -68,11 +70,11 @@ def test_registry_probe_retries_transient_handshake(
             closed.append(chan)
 
     sleep = AsyncMock()
-    monkeypatch.setattr(_root, '_connect_chan', connect_chan)
-    monkeypatch.setattr(_root.trio, 'sleep', sleep)
+    monkeypatch.setattr(_api, '_connect_chan', connect_chan)
+    monkeypatch.setattr(_api.trio, 'sleep', sleep)
 
     async def main():
-        status = await _root._probe_registry(
+        status = await _api._probe_registry(
             addr=wrap_address(('127.0.0.1', 1616)),
             timeout=.3,
             attempt_timeout=.1,
@@ -114,7 +116,7 @@ def test_probe_channel_close_is_bounded(
 
     async def main():
         with trio.fail_after(.5):
-            async with _root._connect_chan(
+            async with _connect_chan(
                 ('127.0.0.1', 1616),
                 close_timeout=.01,
             ):
@@ -193,7 +195,7 @@ def test_registry_probe_preserves_no_peers_state(
             actor = tractor.current_actor()
             server = actor.ipc_server
 
-            probe_status = await _root._probe_registry(
+            probe_status = await _api._probe_registry(
                 addr=wrap_address(reg_addr),
             )
             assert probe_status == 'registrar'

tests/ipc/test_each_tpt.py

@@ -194,7 +194,10 @@ def test_rt_dir_rejects_non_directory(
         lambda appname: str(rt_file),
     )
 
-    with pytest.raises(FileExistsError):
+    with pytest.raises(
+        PermissionError,
+        match='Unsafe POSIX',
+    ):
         _state.get_rt_dir()
 
     new_rt_dir: Path = tmp_path / 'new-runtime-dir'
@@ -206,6 +209,68 @@ def test_rt_dir_rejects_non_directory(
     assert stat.S_IMODE(new_rt_dir.stat().st_mode) == 0o700
 
 
+def test_linux_rt_dir_secures_existing_path(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+):
+    '''
+    Enforce owner-only access on an existing Linux runtime directory.
+
+    Linux previously accepted any existing directory returned by
+    `platformdirs`, without checking ownership or correcting a
+    traversable mode. This test creates an owner-controlled `0o755`
+    directory and proves `get_rt_dir()` normalizes the managed
+    bindspace to `0o700` before returning it.
+
+    '''
+    rt_dir: Path = tmp_path / 'tractor'
+    rt_dir.mkdir(mode=0o755)
+    monkeypatch.setattr(sys, 'platform', 'linux')
+    monkeypatch.setattr(
+        'platformdirs.user_runtime_dir',
+        lambda appname: str(rt_dir),
+    )
+
+    assert _state.get_rt_dir() == rt_dir
+    assert stat.S_IMODE(rt_dir.stat().st_mode) == 0o700
+
+
+def test_linux_rt_dir_rejects_foreign_owner(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+):
+    '''
+    Reject an existing Linux runtime directory owned by another UID.
+
+    A pre-created bindspace must never be made private with `chmod`
+    until ownership is verified. This test makes the current process
+    appear to have a different UID and proves `get_rt_dir()` rejects
+    the directory without changing its original mode.
+
+    '''
+    rt_dir: Path = tmp_path / 'tractor'
+    rt_dir.mkdir(mode=0o755)
+    original_mode: int = stat.S_IMODE(rt_dir.stat().st_mode)
+    monkeypatch.setattr(sys, 'platform', 'linux')
+    monkeypatch.setattr(
+        'platformdirs.user_runtime_dir',
+        lambda appname: str(rt_dir),
+    )
+    monkeypatch.setattr(
+        os,
+        'getuid',
+        lambda: rt_dir.stat().st_uid + 1,
+    )
+
+    with pytest.raises(
+        PermissionError,
+        match='Unsafe POSIX',
+    ):
+        _state.get_rt_dir()
+
+    assert stat.S_IMODE(rt_dir.stat().st_mode) == original_mode
+
+
 def test_macos_rt_dir_rejects_intermediate_symlink(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,

tests/ipc/test_server.py

@@ -34,31 +34,49 @@ from tractor.msg.types import Aid
 # from ._tcp import TCPAddress
 
 
-def test_send_normalizes_peer_reset():
+def test_send_normalizes_only_grouped_peer_resets():
     '''
-    Normalize Darwin's pre-handshake peer reset as transport closure.
+    Normalize only all-peer-close grouped transport failures.
 
     A UDS peer may disconnect before completing the actor handshake.
     Darwin can report the server's first handshake write as
-    `ECONNRESET`, wrapped by `trio.BrokenResourceError`; allowing
-    that raw error to escape cancels the daemon's shared IPC nursery.
-    This fake stream reproduces the exact exception chain and proves
-    `.send()` raises the expected `TransportClosed` boundary instead.
+    `ECONNRESET`, wrapped by `trio.BrokenResourceError` and potentially
+    nested in an `ExceptionGroup`. This fake stream first groups reset
+    and broken-pipe branches, proving `.send()` normalizes a complete
+    peer-close tree to `TransportClosed`. It then groups a reset with
+    an unrelated `ValueError`, proving the mixed failure remains a
+    `trio.BrokenResourceError` instead of hiding the application error.
 
     '''
-    class ResetStream:
-        async def send_all(self, data: bytes) -> None:
+    def broken_resource(err_no: int) -> trio.BrokenResourceError:
+        try:
+            raise OSError(
+                err_no,
+                'Peer closed',
+            )
+        except OSError as peer_err:
             try:
-                raise OSError(
-                    errno.ECONNRESET,
-                    'Connection reset by peer',
-                )
-            except OSError as reset_err:
-                raise trio.BrokenResourceError from reset_err
+                raise trio.BrokenResourceError from peer_err
+            except trio.BrokenResourceError as broken_err:
+                return broken_err
+
+    class GroupedFailureStream:
+        def __init__(self, exceptions: list[Exception]) -> None:
+            self.exceptions = exceptions
+
+        async def send_all(self, data: bytes) -> None:
+            grouped_err = ExceptionGroup(
+                'concurrent send failures',
+                self.exceptions,
+            )
+            raise trio.BrokenResourceError from grouped_err
 
     async def main():
         transport = object.__new__(MsgpackTransport)
-        transport.stream = ResetStream()
+        transport.stream = GroupedFailureStream([
+            broken_resource(errno.ECONNRESET),
+            broken_resource(errno.EPIPE),
+        ])
         transport._send_lock = trio.StrictFIFOLock()
         transport._laddr = 'local'
         transport._raddr = 'remote'
@@ -70,9 +88,23 @@ def test_send_normalizes_peer_reset():
                 strict_types=False,
             )
 
-        assert exc_info.value.src_exc.__cause__.errno == (
-            errno.ECONNRESET
-        )
+        grouped_err = exc_info.value.src_exc.__cause__
+        assert isinstance(grouped_err, ExceptionGroup)
+        assert len(grouped_err.exceptions) == 2
+
+        transport.stream = GroupedFailureStream([
+            ValueError('unrelated failure'),
+            broken_resource(errno.ECONNRESET),
+        ])
+        with pytest.raises(trio.BrokenResourceError) as exc_info:
+            await transport.send(
+                {'probe': True},
+                strict_types=False,
+            )
+
+        grouped_err = exc_info.value.__cause__
+        assert isinstance(grouped_err, ExceptionGroup)
+        assert isinstance(grouped_err.exceptions[0], ValueError)
 
     trio.run(main)
 

tractor/_root.py

@@ -31,7 +31,6 @@ import sys
 from typing import (
     Any,
     Callable,
-    Literal,
 )
 import warnings
 
@@ -48,9 +47,7 @@ from .devx import (
 from .spawn import _spawn
 from .runtime import _state
 from . import log
-from .ipc import (
-    _connect_chan,
-)
+from .discovery._api import _probe_registry_addrs
 from .discovery._addr import (
     Address,
     UnwrappedAddress,
@@ -64,9 +61,7 @@ from .trionics import (
 )
 from ._exceptions import (
     RuntimeFailure,
-    TransportClosed,
 )
-from .msg.types import Aid
 
 
 logger = log.get_logger('tractor')
@@ -86,68 +81,6 @@ _DEBUG_COMPATIBLE_BACKENDS: tuple[str, ...] = (
 )
 
 
-async def _probe_registry(
-    addr: Address,
-    timeout: float = 3,
-    attempt_timeout: float = 1,
-    max_attempts: int = 3,
-    retry_delay: float = .05,
-    close_timeout: float = .2,
-) -> Literal[
-    'absent',
-    'occupied',
-    'registrar',
-]:
-    '''
-    Confirm an address serves the Tractor actor handshake.
-
-    Connection and handshake work share `timeout`; each attempt gets
-    `attempt_timeout`. Shielded cleanup may add up to `close_timeout`
-    per attempted channel.
-
-    '''
-    connected_once: bool = False
-    with trio.move_on_after(timeout):
-        for attempt in range(max_attempts):
-            try:
-                with trio.move_on_after(attempt_timeout) as attempt_cs:
-                    async with _connect_chan(
-                        addr.unwrap(),
-                        close_timeout=close_timeout,
-                    ) as chan:
-                        connected_once = True
-                        peer_aid: Aid = await chan._do_handshake(
-                            aid=Aid(
-                                name='registry-probe',
-                                uuid=mk_uuid(),
-                                pid=os.getpid(),
-                                is_probe=True,
-                            ),
-                            timeout=attempt_timeout,
-                        )
-                        if peer_aid.is_registrar is not False:
-                            return 'registrar'
-                        return 'occupied'
-
-                if attempt_cs.cancelled_caught:
-                    if not connected_once:
-                        return 'absent'
-
-            except OSError:
-                return (
-                    'occupied'
-                    if connected_once
-                    else 'absent'
-                )
-            except TransportClosed:
-                pass
-
-            if attempt + 1 < max_attempts:
-                await trio.sleep(retry_delay * (attempt + 1))
-
-    return 'occupied'
-
-
 # TODO: stick this in a `@acm` defined in `devx.debug`?
 # -[ ] also maybe consider making this a `wrapt`-deco to
 #     save an indent level?
@@ -516,46 +449,12 @@ async def open_root_actor(
             from .devx._stackscope import enable_stack_on_sig
             enable_stack_on_sig()
 
-        # closed into below ping task-func
-        ponged_addrs: list[Address] = []
-        occupied_addrs: list[Address] = []
-
-        async def ping_tpt_socket(
-            addr: Address,
-            timeout: float = 3,
-        ) -> None:
-            '''
-            Probe with a bounded Tractor actor handshake.
-
-            Classify the address as a registrar, occupied by a
-            non-registrar, or absent.
-
-            '''
-            probe_status = await _probe_registry(
-                addr=addr,
-                timeout=timeout,
-            )
-            if probe_status == 'registrar':
-                ponged_addrs.append(addr)
-            elif probe_status == 'occupied':
-                occupied_addrs.append(addr)
-            else:
-                # ?TODO, make this a "discovery" log level?
-                logger.info(
-                    f'No root-actor registry found @ {addr!r}\n'
-                )
-
-        # !TODO, this is basically just another (abstract)
-        # happy-eyeballs, so we should try for formalize it somewhere
-        # in a `.[_]discovery` ya?
-        #
-        async with trio.open_nursery() as tn:
-            for uw_addr in uw_reg_addrs:
-                addr: Address = wrap_address(uw_addr)
-                tn.start_soon(
-                    ping_tpt_socket,
-                    addr,
-                )
+        ponged_addrs: list[Address]
+        occupied_addrs: list[Address]
+        (
+            ponged_addrs,
+            occupied_addrs,
+        ) = await _probe_registry_addrs(uw_reg_addrs)
 
         if (
             not ponged_addrs

tractor/discovery/_api.py

@@ -21,14 +21,18 @@ management of (service) actors.
 """
 from __future__ import annotations
 import ipaddress
+import os
 import socket
 from typing import (
     AsyncGenerator,
     AsyncContextManager,
+    Literal,
     TYPE_CHECKING,
 )
 from contextlib import asynccontextmanager as acm
 
+import trio
+
 from tractor.log import get_logger
 from ..trionics import (
     gather_contexts,
@@ -40,6 +44,7 @@ from ..ipc._uds import UDSAddress
 from ._addr import (
     UnwrappedAddress,
     Address,
+    mk_uuid,
     wrap_address,
 )
 from ..runtime._portal import (
@@ -52,6 +57,7 @@ from ..runtime._state import (
     _runtime_vars,
     _def_tpt_proto,
 )
+from ..msg.types import Aid
 
 if TYPE_CHECKING:
     from ..runtime._runtime import Actor
@@ -60,6 +66,115 @@ if TYPE_CHECKING:
 log = get_logger()
 
 
+async def _probe_registry(
+    addr: Address,
+    timeout: float = 3,
+    attempt_timeout: float = 1,
+    max_attempts: int = 3,
+    retry_delay: float = .05,
+    close_timeout: float = .2,
+) -> Literal[
+    'absent',
+    'occupied',
+    'registrar',
+]:
+    '''
+    Confirm an address serves the Tractor actor handshake.
+
+    Connection and handshake work share `timeout`; each attempt gets
+    `attempt_timeout`. Shielded cleanup may add up to `close_timeout`
+    per attempted channel.
+
+    '''
+    from .._exceptions import TransportClosed
+
+    connected_once: bool = False
+    with trio.move_on_after(timeout):
+        for attempt in range(max_attempts):
+            try:
+                with trio.move_on_after(attempt_timeout) as attempt_cs:
+                    async with _connect_chan(
+                        addr.unwrap(),
+                        close_timeout=close_timeout,
+                    ) as chan:
+                        connected_once = True
+                        peer_aid: Aid = await chan._do_handshake(
+                            aid=Aid(
+                                name='registry-probe',
+                                uuid=mk_uuid(),
+                                pid=os.getpid(),
+                                is_probe=True,
+                            ),
+                            timeout=attempt_timeout,
+                        )
+                        if peer_aid.is_registrar is not False:
+                            return 'registrar'
+                        return 'occupied'
+
+                if attempt_cs.cancelled_caught:
+                    if not connected_once:
+                        return 'absent'
+
+            except OSError:
+                return (
+                    'occupied'
+                    if connected_once
+                    else 'absent'
+                )
+            except TransportClosed:
+                pass
+
+            if attempt + 1 < max_attempts:
+                await trio.sleep(retry_delay * (attempt + 1))
+
+    return 'occupied'
+
+
+async def _probe_registry_addrs(
+    addrs: list[UnwrappedAddress],
+    timeout: float = 3,
+) -> tuple[
+    list[Address],
+    list[Address],
+]:
+    '''
+    Concurrently classify candidate registrar addresses.
+
+    Return confirmed registrar addresses followed by addresses occupied
+    by non-registrar or unresponsive Tractor peers.
+
+    '''
+    registrar_addrs: list[Address] = []
+    occupied_addrs: list[Address] = []
+
+    async def probe_addr(addr: Address) -> None:
+        probe_status = await _probe_registry(
+            addr=addr,
+            timeout=timeout,
+        )
+        if probe_status == 'registrar':
+            registrar_addrs.append(addr)
+        elif probe_status == 'occupied':
+            occupied_addrs.append(addr)
+        else:
+            # ?TODO, make this a "discovery" log level?
+            log.info(
+                f'No root-actor registry found @ {addr!r}\n'
+            )
+
+    async with trio.open_nursery() as nursery:
+        for unwrapped_addr in addrs:
+            nursery.start_soon(
+                probe_addr,
+                wrap_address(unwrapped_addr),
+            )
+
+    return (
+        registrar_addrs,
+        occupied_addrs,
+    )
+
+
 def _is_local_addr(addr: Address) -> bool:
     '''
     Determine whether `addr` is reachable on the

tractor/ipc/_transport.py

@@ -64,29 +64,64 @@ log = get_logger()
 
 def _peer_closed_errno(exc: BaseException) -> int|None:
     '''
-    Find a peer-close errno in a transport exception chain.
+    Classify a complete transport exception tree as peer closure.
+
+    Follow explicit cause/context links. For a `BaseExceptionGroup`,
+    require every child branch to resolve to a peer-close errno so an
+    unrelated concurrent failure is never hidden as `TransportClosed`.
 
     '''
-    seen: set[int] = set()
-    while (
-        exc
-        and
-        id(exc) not in seen
-    ):
-        seen.add(id(exc))
+    def find_peer_errno(
+        current_exc: BaseException,
+        ancestors: set[int],
+    ) -> int|None:
+        exc_id: int = id(current_exc)
+        if exc_id in ancestors:
+            return None
+
+        ancestors = ancestors | {exc_id}
         if (
-            isinstance(exc, OSError)
+            isinstance(current_exc, OSError)
             and
-            exc.errno in {
+            current_exc.errno in {
                 errno.ECONNRESET,
                 errno.EPIPE,
             }
         ):
-            return exc.errno
+            return current_exc.errno
 
-        exc = exc.__cause__ or exc.__context__
+        if isinstance(current_exc, BaseExceptionGroup):
+            child_errnos: list[int|None] = [
+                find_peer_errno(
+                    child_exc,
+                    ancestors,
+                )
+                for child_exc in current_exc.exceptions
+            ]
+            if all(
+                child_errno is not None
+                for child_errno in child_errnos
+            ):
+                return child_errnos[0]
+            return None
 
-    return None
+        chained_exc: BaseException|None = (
+            current_exc.__cause__
+            or
+            current_exc.__context__
+        )
+        if chained_exc is not None:
+            return find_peer_errno(
+                chained_exc,
+                ancestors,
+            )
+
+        return None
+
+    return find_peer_errno(
+        exc,
+        set(),
+    )
 
 
 # (codec, transport)

tractor/runtime/_state.py

@@ -323,6 +323,60 @@ def current_ipc_ctx(
 
 
 
+def _ensure_owner_only_posix_dir(
+    path: Path,
+    *,
+    parents: bool = False,
+) -> None:
+    '''
+    Create or validate a UID-owned POSIX runtime directory.
+
+    Pre-existing directories are accepted only when owned by the
+    current user. Their mode is normalized to `0o700` because runtime
+    directories hold IPC sockets and are private bindspaces.
+
+    '''
+    # TODO: https://github.com/goodboy/tractor/issues/494
+    # Research having the actor-tree root process choose and create
+    # this bindspace, then propagate it to every subactor. On
+    # Linux, a private mount namespace could isolate it while letting
+    # spawned subactors inherit access; independently launched
+    # discovery clients would need an explicit join or fallback path.
+    # POSIX metadata alone records UID/GID ownership, so other systems
+    # still need explicit runtime metadata and lifecycle management.
+    try:
+        dir_stat: os.stat_result = path.lstat()
+    except FileNotFoundError:
+        try:
+            path.mkdir(
+                mode=0o700,
+                parents=parents,
+            )
+        except FileExistsError:
+            pass
+        dir_stat = path.lstat()
+
+    if (
+        not stat.S_ISDIR(dir_stat.st_mode)
+        or
+        dir_stat.st_uid != os.getuid()
+    ):
+        platform_name: str = (
+            'Darwin'
+            if sys.platform == 'darwin'
+            else 'POSIX'
+        )
+        raise PermissionError(
+            f'Unsafe {platform_name} runtime directory!\n'
+            f'path: {path}\n'
+            f'owner uid: {dir_stat.st_uid}\n'
+            f'mode: {stat.filemode(dir_stat.st_mode)}\n'
+        )
+
+    if stat.S_IMODE(dir_stat.st_mode) != 0o700:
+        path.chmod(0o700)
+
+
 def get_rt_dir(
     subdir: str|Path|None = None,
     appname: str = 'tractor',
@@ -332,9 +386,9 @@ def get_rt_dir(
     userspace apps stick their IPC and cache related system
     util-files.
 
-    Linux uses `${XDG_RUNTIME_DIR}/tractor/`; Darwin uses a short,
-    owner-only `/tmp/tractor-<uid>` path; other platforms use the
-    lovely `platformdirs` lib.
+    Linux uses an owner-only `${XDG_RUNTIME_DIR}/tractor/`; Darwin
+    uses a short, owner-only `/tmp/tractor-<uid>` path; other
+    platforms use the lovely `platformdirs` lib.
 
     '''
     # lazy-imported to keep it off the eager
@@ -349,29 +403,6 @@ def get_rt_dir(
             _DARWIN_TMPDIR
             / f'{appname}-{os.getuid()}'
         )
-        try:
-            rt_stat: os.stat_result = rt_root.lstat()
-        except FileNotFoundError:
-            try:
-                rt_root.mkdir(mode=0o700)
-            except FileExistsError:
-                pass
-            rt_stat = rt_root.lstat()
-
-        if (
-            not stat.S_ISDIR(rt_stat.st_mode)
-            or
-            rt_stat.st_uid != os.getuid()
-        ):
-            raise PermissionError(
-                f'Unsafe Darwin runtime directory!\n'
-                f'path: {rt_root}\n'
-                f'owner uid: {rt_stat.st_uid}\n'
-                f'mode: {stat.filemode(rt_stat.st_mode)}\n'
-            )
-        if stat.S_IMODE(rt_stat.st_mode) != 0o700:
-            rt_root.chmod(0o700)
-
         rt_dir: Path = rt_root
     else:
         rt_dir = Path(
@@ -401,7 +432,7 @@ def get_rt_dir(
                 f'{subdir!r}\n'
             )
 
-    if rt_root is None:
+    if os.name != 'posix':
         if subdir_path is not None:
             rt_dir = rt_dir / subdir_path
         if not rt_dir.is_dir():
@@ -414,33 +445,15 @@ def get_rt_dir(
             )
         return rt_dir
 
+    _ensure_owner_only_posix_dir(
+        rt_dir,
+        parents=(rt_root is None),
+    )
+
     if subdir_path is not None:
         for part in subdir_path.parts:
             rt_dir = rt_dir / part
-            try:
-                dir_stat: os.stat_result = rt_dir.lstat()
-            except FileNotFoundError:
-                try:
-                    # Every Darwin component is private so no other
-                    # user can replace descendants below `rt_root`.
-                    rt_dir.mkdir(mode=0o700)
-                except FileExistsError:
-                    pass
-                dir_stat = rt_dir.lstat()
-
-            if (
-                not stat.S_ISDIR(dir_stat.st_mode)
-                or
-                dir_stat.st_uid != os.getuid()
-            ):
-                raise PermissionError(
-                    f'Unsafe Darwin runtime directory!\n'
-                    f'path: {rt_dir}\n'
-                    f'owner uid: {dir_stat.st_uid}\n'
-                    f'mode: {stat.filemode(dir_stat.st_mode)}\n'
-                )
-            if stat.S_IMODE(dir_stat.st_mode) != 0o700:
-                rt_dir.chmod(0o700)
+            _ensure_owner_only_posix_dir(rt_dir)
 
     return rt_dir