fix: every surface raises the error that says what happened

The split into ChannelClosed, GatewayGone and ExecnetStateError reached
only the sync channel.  The async core behind execnet.trio, execnet.aio
and execnet.raw_trio still raised a bare OSError for the same events, so
the names those namespaces export were never raised there.

Connection gone: sends on a closed channel are ChannelClosed, sends and
opens on a finished connection GatewayGone, and a write that times out is
execnet.TimeoutError.  The reader and writer store a lost connection as
GatewayGone, which is what receive raises on every surface.  They also
stored the backend's own ClosedResource/BrokenResource before, which
receive then raised as-is; it is kept only as the cause now.

Misuse: a group used before start or started twice, blocking inside a
running event loop or on the engine's own loop thread, and using a closed
engine raise ExecnetStateError instead of plain RuntimeError, which it
subclasses.  Registering a second callback moves from OSError to
ExecnetStateError, the same break as the two channel misuse cases.

Environment and internal-invariant errors stay RuntimeError: a missing
uv, a monkey-patched process, an engine that cannot start.  So does the
deprecated RSync adapter's OSError, which keeps its 2.x behaviour.

Five of the six new tests fail against the previous code; the sixth
(sync send after close) was already ChannelClosed.

Co-Authored-By: Claude Opus 5.5 (1M context) via Claude Code <noreply@anthropic.com>

Author: RonnyPfannschmidt

CHANGELOG.rst

@@ -210,13 +210,22 @@ series, once the consumers that need them have released without them.
   keeps working -- pytest-xdist swallows exactly that around its shutdown send.
   ``GatewayGone`` is *also* an ``EOFError``, because that is how a broken
   connection has always surfaced; the distinction becomes available without
-  anything ceasing to be caught.
-
-  One deliberate break: two channel operations raised ``OSError`` for API
-  misuse -- closing a channel inside its own ``remote_exec``, and calling
-  ``receive()`` on a channel that has a callback registered. Both are
-  ``ExecnetStateError`` now, which is a ``RuntimeError`` and *not* an
-  ``OSError``. That is the point of the split.
+  anything ceasing to be caught. A connection that is lost rather than
+  terminated is reported as ``GatewayGone`` by ``receive`` too, on every
+  surface -- ``execnet``, ``execnet.gevent``, ``execnet.trio``,
+  ``execnet.aio`` and ``execnet.raw_trio`` alike.
+
+  ``ExecnetStateError`` is what calling something at the wrong time or place
+  raises: using a group that was never started or started twice, blocking
+  from inside a running event loop or from the engine's own loop thread, or
+  using an engine after it was closed. These were plain ``RuntimeError``,
+  which ``ExecnetStateError`` subclasses, so nothing that catches them changes.
+
+  One deliberate break: three channel operations raised ``OSError`` for API
+  misuse -- closing a channel inside its own ``remote_exec``, calling
+  ``receive()`` on a channel that has a callback registered, and registering
+  a second callback. All three are ``ExecnetStateError`` now, which is a
+  ``RuntimeError`` and *not* an ``OSError``. That is the point of the split.
 
 * **What crosses a channel is typed.** Values sent and received were ``object``
   and ``Any``; they now name execnet's wire format -- ``None``, ``bool``,

src/execnet/_bridge.py

@@ -41,6 +41,7 @@
 
 from ._async import current_async
 from ._async import for_backend
+from ._errors import ExecnetStateError
 from ._errors import LoopFinishedError
 from ._trio_gateway import AsyncGroup as _TrioGroup
 
@@ -276,7 +277,7 @@ async def runner() -> None:
                     result = await async_fn(*args)
             except self._aio.Cancelled:
                 # engine shutdown: the nursery cancel must propagate
-                carrier.resolve(None, RuntimeError(ENGINE_GONE))
+                carrier.resolve(None, ExecnetStateError(ENGINE_GONE))
                 raise
             except BaseException as exc:
                 carrier.resolve(None, exc)
@@ -289,14 +290,14 @@ def spawn() -> None:
             try:
                 self._engine.start_soon(runner)
             except BaseException as exc:
-                error = RuntimeError(ENGINE_GONE)
+                error = ExecnetStateError(ENGINE_GONE)
                 error.__cause__ = exc
                 carrier.resolve(None, error)
 
         try:
             self._engine.portal.post(spawn)
         except LoopFinishedError:
-            raise RuntimeError("the execnet engine is not running") from None
+            raise ExecnetStateError("the execnet engine is not running") from None
 
         def cancel_engine_side() -> None:
             with suppress(LoopFinishedError):

src/execnet/_deploy/_facade.py

@@ -18,6 +18,7 @@
 from typing import TypeVar
 
 from .._engine import check_not_in_event_loop
+from .._errors import GatewayGone
 from .._services import ServiceTarget
 
 if TYPE_CHECKING:
@@ -49,7 +50,7 @@ def run_blocking(
 
     sessions = [gateway._trio_session for gateway in gateways]
     if any(session is None for session in sessions):
-        raise OSError("a gateway with no connection cannot be worked on")
+        raise GatewayGone("a gateway with no connection cannot be worked on")
     engines = {id(session.engine) for session in sessions}
     if len(engines) > 1:
         raise ValueError(

src/execnet/_engine.py

@@ -35,6 +35,7 @@
 from typing import Any
 
 from ._errors import ActiveGroupsWarning
+from ._errors import ExecnetStateError
 from ._errors import forked_error
 
 if TYPE_CHECKING:
@@ -152,7 +153,7 @@ def check_not_in_event_loop(what: str) -> None:
     loop = _running_event_loop()
     if loop is None:
         return
-    raise RuntimeError(
+    raise ExecnetStateError(
         f"{what} blocks the calling thread and you are inside a running"
         f" {loop} event loop, which would stall every task on it."
         f" Use {_SURFACE_FOR_LOOP[loop]} instead, or run this in a"
@@ -246,7 +247,7 @@ def _ensure_started(self) -> Any:
         """The started :class:`~execnet._trio_engine.TrioEngine` (internal)."""
         with self._lock:
             if self._closed:
-                raise RuntimeError(
+                raise ExecnetStateError(
                     f"{self!r} was closed: the loop thread is gone, and with it"
                     " every gateway and channel it served. Closing is final --"
                     " build a new ProtocolEngine (and a new Group on it)"
@@ -338,7 +339,7 @@ def _check_not_engine_thread(self, what: str) -> None:
         """Refuse an operation that would have the loop wait for itself."""
         trio_engine = self._trio_engine
         if trio_engine is not None and trio_engine._on_engine_thread():
-            raise RuntimeError(
+            raise ExecnetStateError(
                 f"{what} was called from {self!r}'s own loop thread, where it"
                 " would wait for that loop to finish work it is itself"
                 " running. Call it from the thread that owns the engine."

src/execnet/_gateway_base.py

@@ -135,7 +135,7 @@ def _start_channel_consumer(
         """
         session = self._trio_session
         if session is None:
-            raise OSError(f"cannot set callback on {channel!r}: no active session")
+            raise GatewayGone(f"cannot set callback on {channel!r}: no active session")
         session.attach_consumer(channel, callback, endmarker)
 
     def _terminate_execution(self) -> None:

src/execnet/_portal.py

@@ -35,6 +35,7 @@
 from ._boundary import OneShot
 from ._boundary import ThreadWakener
 from ._boundary import Wakener
+from ._errors import ExecnetStateError
 from ._errors import LoopFinishedError
 from ._errors import forked_error
 
@@ -164,7 +165,7 @@ def run(self, async_fn: Callable[..., Awaitable[T]], *args: Any) -> T:
         """Run ``await async_fn(*args)`` on the loop, blocking this thread."""
         self._check_process()
         if self.is_loop_thread():
-            raise RuntimeError(
+            raise ExecnetStateError(
                 "this is a blocking function; call it from a thread that is"
                 " not running this loop"
             )

src/execnet/_services.py

@@ -33,6 +33,7 @@
 from typing import TYPE_CHECKING
 from typing import Any
 
+from ._errors import GatewayGone
 from ._message import Message
 from ._serialize import dumps_internal
 
@@ -140,7 +141,7 @@ def from_sync(cls, gateway: Any) -> ServiceTarget:
         """A target for a blocking-surface ``Gateway``."""
         session = gateway._trio_session
         if session is None:
-            raise OSError(f"{gateway!r} has no connection to run a service on")
+            raise GatewayGone(f"{gateway!r} has no connection to run a service on")
         return cls(session, gateway._channelfactory.allocate_id)
 
     def __repr__(self) -> str:

src/execnet/_trio_gateway.py

@@ -48,6 +48,9 @@
 
     from ._xspec import XSpec
 
+from ._errors import ChannelClosed
+from ._errors import ExecnetStateError
+from ._errors import GatewayGone
 from ._errors import GatewayReceivedTerminate
 from ._errors import HostNotFound
 from ._errors import RemoteError
@@ -89,6 +92,17 @@ async def provision_sync(fn: Callable[..., T], *args: Any, **kwargs: Any) -> T:
     return result
 
 
+def _gone(message: str, cause: BaseException) -> GatewayGone:
+    """The connection ended: what a receive or send on it reports.
+
+    The backend's own exception stays attached as the cause, but never
+    reaches a caller as the error itself.
+    """
+    error = GatewayGone(message)
+    error.__cause__ = cause
+    return error
+
+
 class ByteStream(Protocol):
     """Neutral bidirectional byte-stream protocol for gateway transports.
 
@@ -300,7 +314,7 @@ async def send_bytes(self, data: bytes) -> None:
         the sync ``Channel.send`` contract.
         """
         if self._closed or self._sent_eof:
-            raise OSError(f"cannot send to {self!r}")
+            raise ChannelClosed(f"cannot send to {self!r}")
         await self.gateway._send(Message.CHANNEL_DATA, self.id, data)
 
     async def receive_bytes(self) -> bytes:
@@ -318,7 +332,7 @@ async def receive_bytes(self) -> bytes:
     async def send_eof(self) -> None:
         """Signal that no more payloads follow (peer keeps its send side)."""
         if self._closed or self._sent_eof:
-            raise OSError(f"cannot send EOF to {self!r}")
+            raise ChannelClosed(f"cannot send EOF to {self!r}")
         self._sent_eof = True
         await self.gateway._send(Message.CHANNEL_LAST_MESSAGE, self.id)
 
@@ -501,7 +515,7 @@ async def send(self, item: SendPayload) -> None:
         channel or gateway is closed.
         """
         if self.isclosed():
-            raise OSError(f"cannot send to {self!r}")
+            raise ChannelClosed(f"cannot send to {self!r}")
         await self._raw.send_bytes(dumps_internal(item))
 
     async def receive(self, timeout: float | None = None) -> Payload[AsyncChannel]:
@@ -628,7 +642,7 @@ def _open_raw_channel(self, id: int | None = None) -> RawChannel:
         dispatch loop already routed data to it.
         """
         if self._closed:
-            raise OSError(f"connection already closed: {self!r}")
+            raise GatewayGone(f"connection already closed: {self!r}")
         if id is None:
             id = self._count
             self._count += 2
@@ -734,14 +748,12 @@ async def _reader(self) -> None:
                     # callers test for EOFError, and an endmarker callback
                     # must fire either way.
                     if not self._closed:
-                        error = EOFError(f"connection closed: {exc}")
-                        error.__cause__ = exc
-                        self._error = error
+                        self._error = _gone(f"connection closed: {exc}", exc)
                     return
                 except self._aio.ClosedResource as exc:
                     # we closed it; not the peer going away
                     if not self._closed:
-                        self._error = exc
+                        self._error = _gone("connection closed by this side", exc)
                     return
                 if not data:
                     decoder.close()  # raises EOFError on a mid-frame EOF
@@ -753,7 +765,7 @@ async def _reader(self) -> None:
             self._trace("received GATEWAY_TERMINATE")
         except EOFError as exc:
             self._trace("EOF without prior gateway termination message")
-            self._error = exc
+            self._error = exc if isinstance(exc, GatewayGone) else _gone(str(exc), exc)
 
     async def _writer(self) -> None:
         error: BaseException | None = None
@@ -771,7 +783,7 @@ async def _writer(self) -> None:
         except (*self._aio.STREAM_GONE, OSError) as exc:
             self._trace("writer failed", exc)
             if self._error is None:
-                self._error = exc
+                self._error = _gone(f"connection lost while writing: {exc}", exc)
         else:
             # Queue closed and drained: signal write-EOF to the peer.
             with suppress(Exception):
@@ -784,7 +796,9 @@ async def _writer(self) -> None:
             self._writer_done.set()
 
     def _fail_pending_writes(self, error: BaseException | None) -> None:
-        exc = error if error is not None else OSError("cannot send (already closed?)")
+        exc = (
+            error if error is not None else GatewayGone("cannot send (already closed?)")
+        )
         while True:
             try:
                 _frame, on_written = self._outbound.receive_nowait()
@@ -884,7 +898,7 @@ async def _send(self, msgcode: int, channelid: int = 0, data: bytes = b"") -> No
                 (Message(msgcode, channelid, data).pack(), None)
             )
         except self._aio.STREAM_GONE as exc:
-            raise OSError("cannot send (already closed?)") from exc
+            raise GatewayGone("cannot send (already closed?)") from exc
 
     def _enqueue_frame(
         self,
@@ -900,7 +914,7 @@ def _enqueue_frame(
         try:
             self._outbound_send.send_nowait((frame, on_written))
         except self._aio.STREAM_GONE as exc:
-            raise OSError("cannot send (already closed?)") from exc
+            raise GatewayGone("cannot send (already closed?)") from exc
 
     def _send_nowait(self, msgcode: int, channelid: int = 0, data: bytes = b"") -> None:
         """Enqueue a frame from a dispatch handler (sync, inline on the loop)."""
@@ -1369,7 +1383,7 @@ async def makegateway(self, spec: str | XSpec = "popen") -> AsyncGateway:
         from ._xspec import XSpec
 
         if self._scope is None:
-            raise RuntimeError(f"{self!r} is not entered")
+            raise ExecnetStateError(f"{self!r} is not entered")
         if not isinstance(spec, XSpec):
             spec = XSpec(spec)
         if spec.profile is None:

src/execnet/_trio_host.py

@@ -24,9 +24,12 @@
 from ._channel import ENDMARKER
 from ._channel import NO_ENDMARKER_WANTED
 from ._channel import Endmarker
+from ._errors import ExecnetStateError
+from ._errors import GatewayGone
 from ._errors import GatewayReceivedTerminate
 from ._errors import LoopFinishedError
 from ._errors import RemoteError
+from ._errors import TimeoutError
 from ._execmodel import ExecModel
 from ._execmodel import get_execmodel
 from ._message import FrameDecoder
@@ -337,13 +340,13 @@ def switch() -> None:
                 # Fail before touching the channel -- a half-switched channel
                 # loses its buffered items, refuses receive(), and leaves
                 # waitclose() waiting for a consumer that will never run.
-                raise OSError(
+                raise GatewayGone(
                     f"cannot set callback on {channel!r}: the engine loop has"
                     " stopped, so nothing can deliver to it"
                 )
             mailbox = channel._mailbox
             if mailbox is None:
-                raise OSError(f"{channel!r} has callback already registered")
+                raise ExecnetStateError(f"{channel!r} has callback already registered")
             inbox_send, inbox_recv = self._aio.queue()
 
             def feed(data: bytes) -> None:
@@ -488,21 +491,21 @@ def post() -> None:
 
         with self._send_lock:
             if self._send_closed:
-                raise OSError("cannot send (already closed?)")
+                raise GatewayGone("cannot send (already closed?)")
             try:
                 # Through the portal even from the engine thread so every
                 # send lands in one global FIFO order.
                 self.engine.portal.post(post)
             except LoopFinishedError:
-                raise OSError("cannot send (already closed?)") from None
+                raise GatewayGone("cannot send (already closed?)") from None
         if ack is None:
             return
         try:
             error = ack.wait(timeout=120.0)
         except TimeoutError:
-            raise OSError("cannot send (write timed out)") from None
+            raise TimeoutError("cannot send (write timed out)") from None
         if error is not None:
-            raise OSError("cannot send (already closed?)") from error
+            raise GatewayGone("cannot send (already closed?)") from error
 
     def post_message(self, message: Message) -> None:
         """Best-effort non-waiting send (Channel.__del__ during GC)."""
@@ -515,7 +518,7 @@ def post() -> None:
         try:
             self.engine.portal.post(post)
         except LoopFinishedError:
-            raise OSError("cannot send (already closed?)") from None
+            raise GatewayGone("cannot send (already closed?)") from None
 
     def request_close_write(self) -> None:
         with self._send_lock:

src/execnet/aio.py

@@ -267,7 +267,7 @@ def engine(self) -> ProtocolEngine:
     async def start(self) -> None:
         """Bring the engine up and start the group task on it."""
         if self._group is not None:
-            raise RuntimeError(f"{self!r} is already started")
+            raise ExecnetStateError(f"{self!r} is already started")
         trio_engine = await start_engine(self._engine, AsyncioCarrier())
         bridge = AsyncioBridge(trio_engine)
 
@@ -315,7 +315,7 @@ async def makegateway(self, spec: str | XSpec = "popen") -> AsyncGateway:
         """
         group, bridge = self._group, self._bridge
         if group is None or bridge is None:
-            raise RuntimeError(f"{self!r} is not started")
+            raise ExecnetStateError(f"{self!r} is not started")
         gateway = await bridge.call(group.makegateway, spec)
         return AsyncGateway(bridge, gateway)