Keep UDS post-kill cleanup best-effort

`unlink_uds_bind_addrs()` reconstructs self-assigned socket paths
after a hard kill. An over-budget bindspace can make
`UDSAddress.get_sockname()` raise before the guarded `os.unlink()`,
replacing the original supervision outcome after the child is gone.

Catch and report reconstruction failures, then skip cleanup without
raising. Cover the overflow path and prove no unlink is attempted.

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/ipc/test_each_tpt.py

@@ -3,14 +3,19 @@ Unit-ish tests for specific IPC transport protocol backends.
 
 '''
 from __future__ import annotations
+import os
 from pathlib import Path
+import stat
+import sys
+from types import SimpleNamespace
+from unittest.mock import Mock
 
 import pytest
 import trio
 import tractor
 from tractor import Actor
-from tractor.runtime import _state
 from tractor.discovery import _addr
+from tractor.runtime import _state
 
 
 @pytest.fixture
@@ -31,6 +36,244 @@ def bindspace_dir_str() -> str:
         bs_dir.rmdir()
 
 
+def test_macos_rt_dir_fits_uds_path_limit(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+):
+    '''
+    Keep the default Darwin UDS bindpath below its 104-byte limit.
+
+    `platformdirs` normally places the runtime directory below the
+    long `~/Library/Caches/TemporaryItems` path. Pytest also assigns
+    a deeply nested temporary home, so appending a registry socket
+    name made every macOS UDS listener fail with `AF_UNIX path too
+    long`. This test simulates Darwin and an intentionally long
+    platformdirs result, then proves `get_rt_dir()` uses the short
+    system temporary directory and leaves room for the socket name.
+
+    '''
+    long_rt_dir: Path = tmp_path / ('long' * 30)
+    monkeypatch.setattr(sys, 'platform', 'darwin')
+    monkeypatch.setattr(
+        'platformdirs.user_runtime_dir',
+        lambda appname: str(long_rt_dir / appname),
+    )
+    monkeypatch.setattr(_state, '_DARWIN_TMPDIR', tmp_path)
+    rt_dir: Path = _state.get_rt_dir()
+    sockpath: Path = (
+        Path('/tmp')
+        / f'tractor-{os.getuid()}'
+        / 'registry@1616.sock'
+    )
+
+    assert rt_dir == tmp_path / f'tractor-{os.getuid()}'
+    assert len(os.fsencode(sockpath)) < 104
+    assert stat.S_IMODE(rt_dir.stat().st_mode) == 0o700
+
+
+def test_macos_rt_dir_rejects_symlink(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+):
+    '''
+    Reject a pre-created symlink at the Darwin runtime path.
+
+    Darwin uses the predictable `/tmp/tractor-<uid>` path to stay
+    below its `AF_UNIX` limit. A hostile local user could otherwise
+    point that path at a victim-owned directory and make
+    `get_rt_dir()` chmod or place sockets in the symlink target. The
+    test replaces `/tmp` with a controlled directory, installs the
+    malicious link, and proves non-following validation rejects it.
+
+    '''
+    runtime_link: Path = tmp_path / f'tractor-{os.getuid()}'
+    target_dir: Path = tmp_path / 'target'
+    target_dir.mkdir(mode=0o755)
+    runtime_link.symlink_to(target_dir, target_is_directory=True)
+    monkeypatch.setattr(sys, 'platform', 'darwin')
+    monkeypatch.setattr(_state, '_DARWIN_TMPDIR', tmp_path)
+
+    with pytest.raises(PermissionError, match='Unsafe Darwin'):
+        _state.get_rt_dir()
+
+    assert stat.S_IMODE(target_dir.stat().st_mode) == 0o755
+
+
+def test_rt_dir_rejects_non_directory(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+):
+    '''
+    Preserve the non-Darwin runtime-directory type contract.
+
+    Replacing `Path.is_dir()` with unguarded `lstat()` briefly made
+    existing files look like valid runtime directories on Linux.
+    This test points `platformdirs` at a regular file and proves
+    `get_rt_dir()` rejects it during initialization.
+
+    '''
+    rt_file: Path = tmp_path / 'runtime-file'
+    rt_file.touch()
+    monkeypatch.setattr(sys, 'platform', 'linux')
+    monkeypatch.setattr(
+        'platformdirs.user_runtime_dir',
+        lambda appname: str(rt_file),
+    )
+
+    with pytest.raises(FileExistsError):
+        _state.get_rt_dir()
+
+    new_rt_dir: Path = tmp_path / 'new-runtime-dir'
+    monkeypatch.setattr(
+        'platformdirs.user_runtime_dir',
+        lambda appname: str(new_rt_dir),
+    )
+    assert _state.get_rt_dir() == new_rt_dir
+    assert stat.S_IMODE(new_rt_dir.stat().st_mode) == 0o700
+
+
+def test_macos_rt_dir_rejects_intermediate_symlink(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+):
+    '''
+    Reject symlinks in nested Darwin runtime subdirectories.
+
+    The earlier final-component check allowed `link/child` to follow
+    an intermediate symlink and create `child` outside the secured
+    runtime root. This test installs that link and proves traversal
+    stops before anything is created in its target.
+
+    '''
+    rt_root: Path = tmp_path / f'tractor-{os.getuid()}'
+    target_dir: Path = tmp_path / 'target'
+    rt_root.mkdir(mode=0o700)
+    target_dir.mkdir()
+    (rt_root / 'link').symlink_to(
+        target_dir,
+        target_is_directory=True,
+    )
+    monkeypatch.setattr(sys, 'platform', 'darwin')
+    monkeypatch.setattr(_state, '_DARWIN_TMPDIR', tmp_path)
+
+    with pytest.raises(PermissionError, match='Unsafe Darwin'):
+        _state.get_rt_dir(subdir='link/child')
+
+    assert not (target_dir / 'child').exists()
+
+
+@pytest.mark.parametrize(
+    ('platform_name', 'path_limit'),
+    [
+        ('darwin', 104),
+        ('linux', 108),
+    ],
+)
+def test_uds_sockname_compaction(
+    monkeypatch: pytest.MonkeyPatch,
+    platform_name: str,
+    path_limit: int,
+):
+    '''
+    Keep generated actor sockets safe and below Darwin's byte limit.
+
+    Actor names are unrestricted identity strings. A long, multibyte,
+    or path-like name previously produced overlong or escaping socket
+    paths. These cases prove `UDSAddress.get_sockname()` preserves a
+    short legacy name, deterministically compacts unsafe names, keeps
+    the reaper's `@pid.sock` suffix, and stays within Darwin's byte
+    limit.
+
+    '''
+    from tractor.ipc._uds import UDSAddress
+
+    bindspace: Path = Path('/tmp/tractor-501')
+    pid: int = 12345
+    from tractor.ipc import _uds
+
+    monkeypatch.setattr(sys, 'platform', platform_name)
+    monkeypatch.setattr(_uds, '_SUN_PATH_LIMIT', path_limit)
+
+    short: Path = UDSAddress.get_sockname(
+        name='worker',
+        pid=pid,
+        bindspace=bindspace,
+    )
+    long_name: str = 'actor-' + ('\u00e9' * 100)
+    compact: Path = UDSAddress.get_sockname(
+        name=long_name,
+        pid=pid,
+        bindspace=bindspace,
+    )
+    unsafe: Path = UDSAddress.get_sockname(
+        name='../worker',
+        pid=pid,
+        bindspace=bindspace,
+    )
+
+    assert short == Path(f'worker@{pid}.sock')
+    assert compact == UDSAddress.get_sockname(
+        name=long_name,
+        pid=pid,
+        bindspace=bindspace,
+    )
+    assert compact.name.endswith(f'@{pid}.sock')
+    assert unsafe.parent == Path('.')
+    assert '..' not in unsafe.name
+    assert len(os.fsencode(bindspace / compact)) < path_limit
+
+    with pytest.raises(ValueError) as exc_info:
+        UDSAddress.get_sockname(
+            name=long_name,
+            pid=pid,
+            bindspace=Path('/tmp') / ('x' * 90),
+        )
+
+    errmsg: str = str(exc_info.value)
+    assert 'leaves no room' in errmsg
+    assert 'name was unsafe: False' in errmsg
+    assert 'name was over budget: True' in errmsg
+    assert f'AF_UNIX path limit: {path_limit}' in errmsg
+
+
+def test_uds_reaper_ignores_unreconstructable_path(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    '''
+    Keep post-kill UDS cleanup best-effort on path overflow.
+
+    `unlink_uds_bind_addrs()` reconstructs a self-assigned socket from
+    the dead actor's name and PID. An over-budget bindspace makes that
+    naming helper raise before `os.unlink()`; propagating the error
+    would replace the original supervision outcome after the child was
+    already killed. This test forces overflow and proves cleanup skips
+    reconstruction without attempting an unlink or raising.
+
+    '''
+    from tractor.ipc import _uds
+    from tractor.spawn import _reap
+
+    long_bindspace: Path = Path('/tmp') / ('x' * 120)
+    proc = SimpleNamespace(pid=12345)
+    subactor = SimpleNamespace(
+        aid=SimpleNamespace(name='worker'),
+    )
+    unlink = Mock()
+    monkeypatch.setattr(
+        _uds.UDSAddress,
+        'def_bindspace',
+        long_bindspace,
+    )
+    monkeypatch.setattr(_reap.os, 'unlink', unlink)
+
+    _reap.unlink_uds_bind_addrs(
+        proc=proc,
+        subactor=subactor,
+    )
+
+    unlink.assert_not_called()
+
+
 def test_uds_bindspace_created_implicitly(
     debug_mode: bool,
     bindspace_dir_str: str,

tests/ipc/test_server.py

@@ -3,6 +3,7 @@ High-level `.ipc._server` unit tests.
 
 '''
 from __future__ import annotations
+import errno
 
 import pytest
 import trio
@@ -14,6 +15,8 @@ from tractor import (
 from tractor._testing.addr import (
     get_rando_addr,
 )
+from tractor._exceptions import TransportClosed
+from tractor.ipc._transport import MsgpackTransport
 # TODO, use/check-roundtripping with some of these wrapper types?
 #
 # from .._addr import Address
@@ -23,6 +26,49 @@ from tractor._testing.addr import (
 # from ._tcp import TCPAddress
 
 
+def test_send_normalizes_peer_reset():
+    '''
+    Normalize Darwin's pre-handshake peer reset as transport closure.
+
+    A raw UDS readiness client connects and immediately disconnects.
+    Darwin reports 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.
+
+    '''
+    class ResetStream:
+        async def send_all(self, data: bytes) -> None:
+            try:
+                raise OSError(
+                    errno.ECONNRESET,
+                    'Connection reset by peer',
+                )
+            except OSError as reset_err:
+                raise trio.BrokenResourceError from reset_err
+
+    async def main():
+        transport = object.__new__(MsgpackTransport)
+        transport.stream = ResetStream()
+        transport._send_lock = trio.StrictFIFOLock()
+        transport._laddr = 'local'
+        transport._raddr = 'remote'
+        transport._task = trio.lowlevel.current_task()
+
+        with pytest.raises(TransportClosed) as exc_info:
+            await transport.send(
+                {'probe': True},
+                strict_types=False,
+            )
+
+        assert exc_info.value.src_exc.__cause__.errno == (
+            errno.ECONNRESET
+        )
+
+    trio.run(main)
+
+
 @pytest.mark.parametrize(
     '_tpt_proto',
     ['uds', 'tcp']

tests/test_docs_examples.py

@@ -5,11 +5,13 @@ Let's make sure them docs work yah?
 from contextlib import contextmanager
 import itertools
 import os
+import signal
 import sys
 import subprocess
 import platform
 import shutil
 from typing import Callable
+from unittest.mock import Mock
 
 import pytest
 import tractor
@@ -21,6 +23,183 @@ _non_linux: bool = platform.system() != 'Linux'
 _friggin_macos: bool = platform.system() == 'Darwin'
 
 
+def _kill_proc_tree(proc: subprocess.Popen) -> None:
+    '''
+    Terminate an example process and its POSIX descendants.
+
+    '''
+    try:
+        if platform.system() == 'Windows':
+            proc.kill()
+        else:
+            os.killpg(proc.pid, signal.SIGKILL)
+    except ProcessLookupError:
+        pass
+
+
+def _reap_killed_proc(
+    proc: subprocess.Popen,
+) -> tuple[bytes, bytes]:
+    '''
+    Reap a killed process without waiting on Windows descendants.
+
+    '''
+    if platform.system() != 'Windows':
+        return proc.communicate()
+
+    proc.wait(timeout=5)
+    if proc.stdin:
+        proc.stdin.close()
+    if proc.stdout:
+        proc.stdout.close()
+    if proc.stderr:
+        proc.stderr.close()
+    return b'', b''
+
+
+def _wait_for_proc(
+    proc: subprocess.Popen,
+    timeout: float,
+    test_log: tractor.log.StackLevelAdapter,
+) -> None:
+    '''
+    Wait for an example process and surface its captured output.
+
+    '''
+    try:
+        out, err = proc.communicate(timeout=timeout)
+
+    except subprocess.TimeoutExpired as timeout_exc:
+        test_log.exception(
+            f'Example failed to finish within {timeout}s ??\n'
+        )
+        _kill_proc_tree(proc)
+        out, err = _reap_killed_proc(proc)
+        if platform.system() == 'Windows':
+            out = timeout_exc.output or b''
+            err = timeout_exc.stderr or b''
+
+    errmsg: str = err.decode(errors='replace')
+
+    # XXX, ALWAYS surface the subproc's full stderr
+    # whenever it exits non-zero!
+    #
+    # The prior impl only raised when the LAST stderr
+    # line contained 'Error', swallowing any crash whose
+    # traceback ends in a non-`XxxError:` line; in
+    # particular EVERY `tractor` root-actor crash ends
+    # with the strict-EG collapse note,
+    # '( ^^^ this exc was collapsed from a group ^^^ )',
+    # so ALL such failures were reduced to a bare
+    # `assert 1 == 0` in CI logs.. see GH #473.
+    rc: int|None = proc.returncode
+    if rc:
+        outmsg: str = out.decode(errors='replace')
+        raise Exception(
+            f'Example script exited with rc={rc} !?\n'
+            f'\n'
+            f'stdout:\n'
+            f'{outmsg}\n'
+            f'\n'
+            f'stderr:\n'
+            f'{errmsg}\n'
+        )
+
+    # if we get some gnarly output let's aggregate and raise
+    if errmsg:
+        errlines = errmsg.splitlines()
+        last_error = errlines[-1]
+        if (
+            'Error' in last_error
+
+            # XXX: currently we print this to console, but maybe
+            # shouldn't eventually once we figure out what's
+            # a better way to be explicit about aio side
+            # cancels?
+            and
+            'asyncio.exceptions.CancelledError' not in last_error
+        ):
+            raise Exception(errmsg)
+
+    assert proc.returncode == 0
+
+
+def test_wait_for_failed_example_captures_output():
+    '''
+    Preserve diagnostics from a subprocess which already exited.
+
+    The previous `poll()` guard skipped `communicate()` when a fast
+    failure returned a non-zero status before the parent checked it.
+    Its stdout and stderr were therefore reported as empty. This
+    fake process begins with `returncode=1` and returns non-UTF-8
+    output, proving the helper always drains both pipes and replaces
+    undecodable bytes without hiding the original process failure.
+
+    '''
+    proc = Mock()
+    proc.returncode = 1
+    proc.communicate.return_value = (
+        b'stdout\xff',
+        b'stderr\xff',
+    )
+
+    with pytest.raises(Exception) as exc_info:
+        _wait_for_proc(
+            proc=proc,
+            timeout=1,
+            test_log=Mock(),
+        )
+
+    proc.communicate.assert_called_once_with(timeout=1)
+    errmsg: str = str(exc_info.value)
+    assert 'stdout\ufffd' in errmsg
+    assert 'stderr\ufffd' in errmsg
+
+
+@pytest.mark.skipif(
+    platform.system() == 'Windows',
+    reason='POSIX process groups are unavailable on Windows',
+)
+def test_wait_for_timed_out_example_reaps_group(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    '''
+    Kill the example process group and reap its leader on timeout.
+
+    The old timeout branch killed only the immediate process and
+    never drained it. Actor descendants could retain the capture
+    pipes while the leader remained unreaped, hanging CI until its
+    job timeout. This fake process raises `TimeoutExpired` on the
+    timed wait and completes on the second `communicate()` call;
+    the assertions prove group-directed `SIGKILL` precedes that
+    final drain and leaves a concrete non-zero return code.
+
+    '''
+    proc = Mock()
+    proc.pid = 1234
+
+    def communicate(timeout=None):
+        if timeout is not None:
+            raise subprocess.TimeoutExpired('example', timeout)
+        proc.returncode = -signal.SIGKILL
+        return b'', b'timed out'
+
+    proc.communicate.side_effect = communicate
+    killpg = Mock()
+    monkeypatch.setattr(os, 'killpg', killpg)
+
+    with pytest.raises(Exception, match='timed out'):
+        _wait_for_proc(
+            proc=proc,
+            timeout=.01,
+            test_log=Mock(),
+        )
+
+    killpg.assert_called_once_with(1234, signal.SIGKILL)
+    assert proc.communicate.call_count == 2
+    assert proc.returncode == -signal.SIGKILL
+
+
 @pytest.fixture
 def run_example_in_subproc(
     loglevel: str,
@@ -61,6 +240,7 @@ def run_example_in_subproc(
             ]
         else:
             script_file = testdir.makefile('.py', script_code)
+            kwargs['start_new_session'] = True
             cmdargs = [
                 sys.executable,
                 str(script_file),
@@ -77,9 +257,20 @@ def run_example_in_subproc(
             **kwargs,
         )
         assert not proc.returncode
-        yield proc
+        try:
-        proc.wait()
+            yield proc
-        assert proc.returncode == 0
+        except BaseException:
+            if proc.poll() is None:
+                try:
+                    _kill_proc_tree(proc)
+                    _reap_killed_proc(proc)
+                except Exception:
+                    pass
+            raise
+        else:
+            if proc.poll() is None:
+                _kill_proc_tree(proc)
+                _reap_killed_proc(proc)
 
     yield run
 
@@ -163,59 +354,8 @@ def test_example(
         code = ex.read()
 
         with run_example_in_subproc(code) as proc:
-            out = None
+            _wait_for_proc(
-            err = None
+                proc=proc,
-            try:
+                timeout=timeout,
-                if not proc.poll():
+                test_log=test_log,
-                    out, err = proc.communicate(timeout=timeout)
+            )
-
-            except subprocess.TimeoutExpired as e:
-                test_log.exception(
-                    f'Example failed to finish within {timeout}s ??\n'
-                )
-                proc.kill()
-                err = e.stderr
-
-            errmsg: str = err.decode() if err else ''
-
-            # XXX, ALWAYS surface the subproc's full stderr
-            # whenever it exits non-zero!
-            #
-            # The prior impl only raised when the LAST stderr
-            # line contained 'Error', swallowing any crash whose
-            # traceback ends in a non-`XxxError:` line; in
-            # particular EVERY `tractor` root-actor crash ends
-            # with the strict-EG collapse note,
-            # '( ^^^ this exc was collapsed from a group ^^^ )',
-            # so ALL such failures were reduced to a bare
-            # `assert 1 == 0` in CI logs.. see GH #473.
-            rc: int|None = proc.returncode
-            if rc:
-                outmsg: str = out.decode() if out else ''
-                raise Exception(
-                    f'Example script exited with rc={rc} !?\n'
-                    f'\n'
-                    f'stdout:\n'
-                    f'{outmsg}\n'
-                    f'\n'
-                    f'stderr:\n'
-                    f'{errmsg}\n'
-                )
-
-            # if we get some gnarly output let's aggregate and raise
-            if errmsg:
-                errlines = errmsg.splitlines()
-                last_error = errlines[-1]
-                if (
-                    'Error' in last_error
-
-                    # XXX: currently we print this to console, but maybe
-                    # shouldn't eventually once we figure out what's
-                    # a better way to be explicit about aio side
-                    # cancels?
-                    and
-                    'asyncio.exceptions.CancelledError' not in last_error
-                ):
-                    raise Exception(errmsg)
-
-            assert proc.returncode == 0

tractor/ipc/_transport.py

@@ -33,6 +33,7 @@ from collections.abc import (
     AsyncGenerator,
     AsyncIterator,
 )
+import errno
 import struct
 
 import trio
@@ -61,6 +62,33 @@ if TYPE_CHECKING:
 log = get_logger()
 
 
+def _peer_closed_errno(exc: BaseException) -> int|None:
+    '''
+    Find a peer-close errno in a transport exception chain.
+
+    '''
+    seen: set[int] = set()
+    while (
+        exc
+        and
+        id(exc) not in seen
+    ):
+        seen.add(id(exc))
+        if (
+            isinstance(exc, OSError)
+            and
+            exc.errno in {
+                errno.ECONNRESET,
+                errno.EPIPE,
+            }
+        ):
+            return exc.errno
+
+        exc = exc.__cause__ or exc.__context__
+
+    return None
+
+
 # (codec, transport)
 MsgTransportKey = tuple[str, str]
 
@@ -443,23 +471,23 @@ class MsgpackTransport(MsgTransport):
                 trans_err = _re
                 tpt_name: str = f'{type(self).__name__!r}'
 
-                trans_err_msg: str = trans_err.args[0]
+                trans_err_msg: str = (
+                    str(trans_err.args[0])
+                    if trans_err.args
+                    else ''
+                )
                 by_whom: str = {
                     'another task closed this fd': 'locally',
                     'this socket was already closed': 'by peer',
                 }.get(trans_err_msg)
                 match trans_err:
 
-                    # XXX, specifc to UDS transport and its,
+                    # UDS peers can disconnect before handshake.
-                    # well, "speediness".. XD
+                    # Linux normally reports `EPIPE`; Darwin reports
-                    # |_ likely todo with races related to how fast
+                    # `ECONNRESET` for the same expected closure.
-                    #    the socket is setup/torn-down on linux
-                    #    as it pertains to rando pings from the
-                    #    `.discovery` subsys and protos.
                     case trio.BrokenResourceError() if (
-                        '[Errno 32] Broken pipe'
+                        _peer_closed_errno(trans_err)
-                        in
+                        is not None
-                        trans_err_msg
                     ):
                         tpt_closed = TransportClosed.from_src_exc(
                             message=(

tractor/ipc/_uds.py

@@ -21,6 +21,7 @@ from __future__ import annotations
 from contextlib import (
     contextmanager as cm,
 )
+import hashlib
 from pathlib import Path
 import os
 import sys
@@ -96,6 +97,12 @@ else:
 
 log = get_logger()
 
+_SUN_PATH_LIMIT: int = (
+    108
+    if sys.platform == 'linux'
+    else 104
+)
+
 
 def unwrap_sockpath(
     sockpath: Path,
@@ -196,7 +203,7 @@ class UDSAddress(
             err_on_no_runtime=False,
         )
         if actor:
-            sockname: str = f'{actor.aid.name}@{pid}'
+            sockname: str = actor.aid.name
             # XXX, orig version which broke both macOS (file-name
             # length) and `multiaddrs` ('::' invalid separator).
             # sockname: str = '::'.join(actor.uid) + f'@{pid}'
@@ -222,15 +229,72 @@ class UDSAddress(
             # `(?P<name>.+)@(?P<pid>\d+)\.sock` regex, and the
             # `spawn._reap` `{name}@{pid}.sock` reconstruction.
             token: str = uuid4().hex[:8]
-            sockname: str = f'{prefix}.{token}@{pid}'
+            sockname = f'{prefix}.{token}'
 
-        sockpath: Path = Path(f'{sockname}.sock')
+        sockpath: Path = cls.get_sockname(
+            name=sockname,
+            pid=pid,
+            bindspace=filedir,
+        )
         return UDSAddress(
             filedir=filedir,
             filename=sockpath,
             maybe_pid=pid,
         )
 
+    @classmethod
+    def get_sockname(
+        cls,
+        name: str,
+        pid: int,
+        bindspace: Path,
+    ) -> Path:
+        '''
+        Build a safe, deterministic UDS socket filename.
+
+        '''
+        suffix: str = f'@{pid}.sock'
+        filename: str = f'{name}{suffix}'
+        unsafe: bool = (
+            '\0' in name
+            or
+            '/' in name
+            or
+            bool(os.altsep and os.altsep in name)
+            or
+            Path(filename).is_absolute()
+        )
+        too_long: bool = (
+            len(os.fsencode(bindspace / filename))
+            >= _SUN_PATH_LIMIT
+        )
+        if (
+            unsafe
+            or
+            too_long
+        ):
+            digest: str = hashlib.blake2s(
+                os.fsencode(name),
+                digest_size=16,
+            ).hexdigest()
+            filename = f'actor.{digest}{suffix}'
+
+        sockpath: Path = bindspace / filename
+        path_nbytes: int = len(os.fsencode(sockpath))
+        if path_nbytes >= _SUN_PATH_LIMIT:
+            raise ValueError(
+                f'UDS bindspace leaves no room for an AF_UNIX '
+                f'socket filename!\n'
+                f'bindspace: {bindspace}\n'
+                f'name was unsafe: {unsafe}\n'
+                f'name was over budget: {too_long}\n'
+                f'compacted filename: {filename}\n'
+                f'encoded path bytes: {path_nbytes}\n'
+                f'AF_UNIX path limit: {_SUN_PATH_LIMIT}\n'
+            )
+
+        return Path(filename)
+
     @classmethod
     def get_root(cls) -> UDSAddress:
         def_uds_filename: Path = 'registry@1616.sock'

tractor/runtime/_state.py

@@ -22,7 +22,10 @@ from __future__ import annotations
 from contextvars import (
     ContextVar,
 )
+import os
 from pathlib import Path
+import stat
+import sys
 from typing import (
     Any,
     Callable,
@@ -42,6 +45,9 @@ if TYPE_CHECKING:
     from .._context import Context
 
 
+_DARWIN_TMPDIR: Path = Path('/tmp')
+
+
 # default IPC transport protocol settings
 TransportProtocolKey = Literal[
     'tcp',
@@ -335,15 +341,49 @@ def get_rt_dir(
     # `import tractor` path (gh #470).
     import platformdirs
 
-    rt_dir: Path = Path(
+    rt_root: Path|None = None
-        platformdirs.user_runtime_dir(
+    if sys.platform == 'darwin':
-            appname=appname,
+        # Darwin's AF_UNIX path limit is 104 bytes. The standard
-        ),
+        # platformdirs path can consume that before the sock name.
-    )
+        rt_root = (
+            _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(
+            platformdirs.user_runtime_dir(
+                appname=appname,
+            ),
+        )
 
     # Normalize and validate that `subdir` is a relative path
     # without any parent-directory ("..") components, to prevent
     # escaping the runtime directory.
+    subdir_path: Path|None = None
     if subdir:
         subdir_path = (
             subdir
@@ -361,13 +401,46 @@ def get_rt_dir(
                 f'{subdir!r}\n'
             )
 
-        rt_dir: Path = rt_dir / subdir_path
+    if rt_root is None:
+        if subdir_path is not None:
+            rt_dir = rt_dir / subdir_path
+        if not rt_dir.is_dir():
+            rt_dir.mkdir(
+                # Runtime dirs hold IPC sockets; owner-only access
+                # prevents other users from traversing the bindspace.
+                mode=0o700,
+                parents=True,
+                exist_ok=True,
+            )
+        return rt_dir
 
-    if not rt_dir.is_dir():
+    if subdir_path is not None:
-        rt_dir.mkdir(
+        for part in subdir_path.parts:
-            parents=True,
+            rt_dir = rt_dir / part
-            exist_ok=True,  # avoid `FileExistsError` from conc calls
+            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)
 
     return rt_dir
 

tractor/spawn/_reap.py

@@ -71,6 +71,7 @@ fd leak. Different bug class but same broader theme of
 from __future__ import annotations
 
 import os
+from pathlib import Path
 from typing import TYPE_CHECKING
 
 import trio
@@ -154,7 +155,21 @@ def unlink_uds_bind_addrs(
         and subactor is not None
         and proc.pid is not None
     ):
-        sockname: str = f'{subactor.aid.name}@{proc.pid}.sock'
+        try:
+            sockname: Path = UDSAddress.get_sockname(
+                name=subactor.aid.name,
+                pid=proc.pid,
+                bindspace=UDSAddress.def_bindspace,
+            )
+        except Exception:
+            log.exception(
+                f'Failed to reconstruct UDS sock-file for '
+                f'post-kill cleanup — skipping\n'
+                f' |_{proc}\n'
+                f' |_{subactor.aid}\n'
+            )
+            return
+
         sockpath: str = str(
             UDSAddress.def_bindspace / sockname
         )