Skip `test_ringbuf` at collection off-linux

`tests/test_ringbuf.py` imports `tractor.ipc._ringbuf` at module
top, which pulls in `tractor.ipc._linux` whose module-level
`ffi.dlopen(None)` raises `OSError` on Windows (and any non-linux
host). That fires at COLLECTION, before the module's existing
`pytestmark = pytest.mark.skip` can apply, so it aborts the whole
pytest session — the new `windows-latest` CI leg never gets past
collection.

- guard the module with `pytest.skip(allow_module_level=True)`
  gated on `platform.system() != 'Linux'`, placed before the
  crashing import — same idiom as `tests/devx/test_debugger.py`.
- the `eventfd`-based ringbuf backend is linux-only by design, so
  macOS skips cleanly too (previously it only skipped incidentally
  via the absent `cffi` optional dep).

(this patch was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

.github/workflows/ci.yml

@@ -91,6 +91,11 @@ jobs:
     name: '${{ matrix.os }} Python${{ matrix.python-version }} spawn_backend=${{ matrix.spawn_backend }} tpt_proto=${{ matrix.tpt_proto }}'
     timeout-minutes: 16
     runs-on: ${{ matrix.os }}
+    # Windows support is nascent: keep its legs informational so a
+    # known-incomplete area doesn't block merges. The `import
+    # tractor` smoke step below is the hard signal for this job;
+    # promote the whole leg to required once the suite is green.
+    continue-on-error: ${{ matrix.os == 'windows-latest' }}
 
     strategy:
       fail-fast: false
@@ -98,6 +103,7 @@ jobs:
         os: [
           ubuntu-latest,
           macos-latest,
+          windows-latest,
         ]
         python-version: [
           '3.13',
@@ -123,6 +129,10 @@ jobs:
           # don't do UDS run on macOS (for now)
           - os: macos-latest
             tpt_proto: 'uds'
+          # UDS is POSIX-only; Windows has no `AF_UNIX` so the
+          # backend is intentionally unavailable there.
+          - os: windows-latest
+            tpt_proto: 'uds'
 
     steps:
       - uses: actions/checkout@v4
@@ -150,6 +160,12 @@ jobs:
       - name: List deps tree
         run: uv tree
 
+      # hard signal for the Windows import-safety fix: `import
+      # tractor` must succeed everywhere, and `HAS_UDS` reflects
+      # platform capability (False on Windows, True on POSIX).
+      - name: 'Smoke: import tractor'
+        run: uv run python -c "import tractor; from tractor.ipc._uds import HAS_UDS; print('import tractor OK | HAS_UDS=', HAS_UDS)"
+
       - name: Run tests
         run: >
           uv run

tests/test_advanced_streaming.py

@@ -189,11 +189,16 @@ def test_dynamic_pub_sub(
     # sits forever until external SIGINT. The `afk_alarm_w_trace`
     # outer guard below is the AFK-safety counterpart (SIGALRM
     # raises in the main thread regardless of trio scope state).
-    fail_after_s: int = (
+    # scale the budget for slow/noisy CI runners (esp. macOS, where
+    # `cpu_scaling_factor()` applies a 3x CI bump) so a sluggish
+    # runner doesn't trip the deadline and inject a `TooSlowError`
+    # that collides with the `expect_cancel_exc=TooSlowError` param.
+    from .conftest import cpu_scaling_factor
+    fail_after_s: float = (
         8
         if is_forking_spawner
         else 20
-    )
+    ) * cpu_scaling_factor()
 
     async def main():
         # bug-class-3 breadcrumb: tag each level of the cancel path

tests/test_ringbuf.py

@@ -1,10 +1,22 @@
 import time
+import platform
 
 import trio
 import pytest
 
 import tractor
 
+# `tractor.ipc._ringbuf` is built on linux `eventfd(2)`; importing
+# it pulls in `tractor.ipc._linux` whose module-level
+# `ffi.dlopen(None)` raises on non-linux. Skip the whole module at
+# COLLECTION before that crashing import runs (a `pytestmark` skip
+# is too late — markers apply only after the import succeeds).
+if platform.system() != 'Linux':
+    pytest.skip(
+        'ringbuf (eventfd) IPC is linux-only',
+        allow_module_level=True,
+    )
+
 # XXX `cffi` dun build on py3.14 yet..
 pytest.importorskip("cffi")
 

tractor/devx/_stackscope.py

@@ -31,12 +31,21 @@ from threading import (
     RLock,
 )
 import multiprocessing as mp
+
+import platform
+
 from signal import (
     signal,
     getsignal,
-    SIGUSR1,
     SIGINT,
 )
+
+
+if platform.system() != "Windows":
+    from signal import SIGUSR1
+else:
+    SIGUSR1 = None
+
 # import traceback
 from types import ModuleType
 from typing import (
@@ -347,8 +356,8 @@ def dump_tree_on_sig(
 
 
 def enable_stack_on_sig(
-    sig: int = SIGUSR1,
+    sig: int|None = SIGUSR1,
-) -> ModuleType:
+) -> ModuleType|None:
     '''
     Enable `stackscope` tracing on reception of a signal; by
     default this is SIGUSR1.
@@ -367,6 +376,16 @@ def enable_stack_on_sig(
     >> pkill --signal SIGUSR1 -f <part-of-cmd: str>
 
     '''
+    # no `SIGUSR1` on this platform (e.g. Windows) -> nothing to
+    # wire up; degrade gracefully instead of crashing callers that
+    # only guard against a missing `stackscope` (`ImportError`).
+    if sig is None:
+        log.warning(
+            'No `SIGUSR1` on this platform;\n'
+            'skipping `stackscope` trace-on-signal setup!\n'
+        )
+        return None
+
     try:
         # NOTE, `stackscope._glue` does intentional async-gen type
         # introspection at import-time which trips

tractor/discovery/_addr.py

@@ -32,14 +32,16 @@ from ..runtime._state import (
     _def_tpt_proto,
 )
 from ..ipc._tcp import TCPAddress
-from ..ipc._uds import UDSAddress
+from ..ipc._uds import (
+    UDSAddress,
+    HAS_UDS,
+)
 
 if TYPE_CHECKING:
     from ..runtime._runtime import Actor
 
 log = get_logger()
 
-
 # TODO, maybe breakout the netns key to a struct?
 # class NetNs(Struct)[str, int]:
 #     ...
@@ -172,19 +174,18 @@ class Address(Protocol):
 
 _address_types: bidict[str, Type[Address]] = {
     'tcp': TCPAddress,
-    'uds': UDSAddress
 }
+if HAS_UDS:
+    _address_types['uds'] = UDSAddress
 
 
 # TODO! really these are discovery sys default addrs ONLY useful for
 # when none is provided to a root actor on first boot.
-_default_lo_addrs: dict[
+_default_lo_addrs: dict[str, UnwrappedAddress] = {
-    str,
-    UnwrappedAddress
-] = {
     'tcp': TCPAddress.get_root().unwrap(),
-    'uds': UDSAddress.get_root().unwrap(),
 }
+if HAS_UDS:
+    _default_lo_addrs['uds'] = UDSAddress.get_root().unwrap()
 
 
 def get_address_cls(name: str) -> Type[Address]:

tractor/ipc/_server.py

@@ -62,16 +62,17 @@ from .. import log
 from ..discovery._addr import Address
 from ._chan import Channel
 from ._transport import MsgTransport
-from ._uds import UDSAddress
+
-from ._tcp import TCPAddress
 
 if TYPE_CHECKING:
     from ..runtime._runtime import Actor
     from ..runtime._supervise import ActorNursery
 
 
-log = log.get_logger()
+from ._tcp import TCPAddress
+from ._uds import UDSAddress
 
+log = log.get_logger()
 
 async def maybe_wait_on_canced_subs(
     uid: tuple[str, str],

tractor/ipc/_types.py

@@ -18,17 +18,13 @@
 IPC subsys type-lookup helpers?
 
 '''
-from typing import (
+from typing import Type
-    Type,
-    # TYPE_CHECKING,
-)
-
-import trio
 import socket
+import trio
 
 from tractor.ipc._transport import (
     MsgTransportKey,
-    MsgTransport
+    MsgTransport,
 )
 from tractor.ipc._tcp import (
     TCPAddress,
@@ -37,87 +33,86 @@ from tractor.ipc._tcp import (
 from tractor.ipc._uds import (
     UDSAddress,
     MsgpackUDSStream,
+    HAS_UDS,
 )
 
-# if TYPE_CHECKING:
+# the UDS backend is importable everywhere but only *usable* where
-#     from tractor._addr import Address
+# `trio` reports `has_unix` (i.e. POSIX). On Windows / no-`AF_UNIX`
-
+# hosts `HAS_UDS` is `False` and the runtime registers TCP only.
-
 Address = TCPAddress|UDSAddress
 
 # manually updated list of all supported msg transport types
-_msg_transports = [
+_msg_transports: list[Type[MsgTransport]] = [
     MsgpackTCPStream,
-    MsgpackUDSStream
 ]
+if HAS_UDS:
+    _msg_transports.append(MsgpackUDSStream)
 
-
+# map a `MsgTransportKey` to its `MsgTransport` type
-# convert a MsgTransportKey to the corresponding transport type
+_key_to_transport: dict[MsgTransportKey, Type[MsgTransport]] = {
-_key_to_transport: dict[
-    MsgTransportKey,
-    Type[MsgTransport],
-] = {
     ('msgpack', 'tcp'): MsgpackTCPStream,
-    ('msgpack', 'uds'): MsgpackUDSStream,
 }
+if HAS_UDS:
+    _key_to_transport[('msgpack', 'uds')] = MsgpackUDSStream
 
-# convert an Address wrapper to its corresponding transport type
+# map an `Address`-wrapper to its `MsgTransport` type
-_addr_to_transport: dict[
+_addr_to_transport: dict[Type[Address], Type[MsgTransport]] = {
-    Type[TCPAddress|UDSAddress],
-    Type[MsgTransport]
-] = {
     TCPAddress: MsgpackTCPStream,
-    UDSAddress: MsgpackUDSStream,
 }
+if HAS_UDS:
+    _addr_to_transport[UDSAddress] = MsgpackUDSStream
 
 
+# ------------------------------------------------------------
+# Helpers
+# ------------------------------------------------------------
 def transport_from_addr(
     addr: Address,
-    codec_key: str = 'msgpack',
+    codec_key: str = "msgpack",
 ) -> Type[MsgTransport]:
-    '''
+    """
     Given a destination address and a desired codec, find the
     corresponding `MsgTransport` type.
-
+    """
-    '''
     try:
-        return _addr_to_transport[type(addr)]
+        return _addr_to_transport[type(addr)]  # type: ignore[call-arg]
-
     except KeyError:
         raise NotImplementedError(
-            f'No known transport for address {repr(addr)}'
+            f"No known transport for address {repr(addr)}"
         )
 
 
 def transport_from_stream(
     stream: trio.abc.Stream,
-    codec_key: str = 'msgpack'
+    codec_key: str = "msgpack",
 ) -> Type[MsgTransport]:
-    '''
+    """
     Given an arbitrary `trio.abc.Stream` and a desired codec,
     find the corresponding `MsgTransport` type.
-
+    """
-    '''
     transport = None
+
     if isinstance(stream, trio.SocketStream):
         sock: socket.socket = stream.socket
-        match sock.family:
+        fam = sock.family
-            case socket.AF_INET | socket.AF_INET6:
-                transport = 'tcp'
 
-            case socket.AF_UNIX:
+        if fam in (socket.AF_INET, getattr(socket, "AF_INET6", None)):
-                transport = 'uds'
+            transport = "tcp"
 
-            case _:
+        # only consider `AF_UNIX` when the UDS backend is active;
-                raise NotImplementedError(
+        # `HAS_UDS` short-circuits before `socket.AF_UNIX` so this
-                    f'Unsupported socket family: {sock.family}'
+        # stays safe on hosts where that constant is absent.
-                )
+        if transport is None and HAS_UDS and fam == socket.AF_UNIX:  # type: ignore[attr-defined]
+            transport = "uds"
+
+        if transport is None:
+            raise NotImplementedError(f"Unsupported socket family: {fam}")
 
     if not transport:
         raise NotImplementedError(
-            f'Could not figure out transport type for stream type {type(stream)}'
+            f"Could not figure out transport type for stream type {type(stream)}"
         )
 
     key = (codec_key, transport)
-
     return _key_to_transport[key]
+

tractor/ipc/_uds.py

@@ -25,11 +25,21 @@ from pathlib import Path
 import os
 import sys
 from socket import (
-    AF_UNIX,
     SOCK_STREAM,
     SOL_SOCKET,
     error as socket_error,
 )
+# NOTE, `AF_UNIX` is absent on Windows / any CPython built without
+# unix-domain-socket support. Keep this module importable
+# everywhere (so `UDSAddress` stays referenceable for type and
+# `isinstance()` checks plus registry lookups); the `AF_UNIX`-using
+# code paths below are runtime-only and are never reached when the
+# UDS backend is unusable (gated on `trio`'s `has_unix`, see
+# `HAS_UDS`).
+try:
+    from socket import AF_UNIX
+except ImportError:
+    AF_UNIX = None
 import struct
 from typing import (
     Type,
@@ -90,6 +100,13 @@ else:
 log = get_logger()
 
 
+# single source of truth for "is the UDS backend usable on this
+# host?" — reuse `trio`'s `has_unix` (the same predicate that gates
+# `trio.open_unix_socket()`) rather than re-deriving an `AF_UNIX`
+# probe in every consumer module.
+HAS_UDS: bool = has_unix
+
+
 def unwrap_sockpath(
     sockpath: Path,
 ) -> tuple[Path, Path]: