Let a run end on an interrupt without naming another cause

.basedpyright/baseline.json

@@ -1211,14 +1211,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 73,
-                    "endColumn": 99,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportAttributeAccessIssue",
                 "range": {
@@ -6895,14 +6887,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 22,
-                    "endColumn": 29,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportAttributeAccessIssue",
                 "range": {
@@ -6911,14 +6895,6 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 26,
-                    "endColumn": 33,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportAttributeAccessIssue",
                 "range": {
@@ -6951,30 +6927,14 @@
                     "lineCount": 1
                 }
             },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 26,
-                    "endColumn": 33,
-                    "lineCount": 1
-                }
-            },
             {
                 "code": "reportAttributeAccessIssue",
                 "range": {
                     "startColumn": 39,
                     "endColumn": 46,
                     "lineCount": 1
                 }
-            },
-            {
-                "code": "reportArgumentType",
-                "range": {
-                    "startColumn": 22,
-                    "endColumn": 26,
-                    "lineCount": 1
-                }
             }
         ]
     }
-}
+}

pimm/tests/test_calls.py

@@ -241,7 +241,8 @@ def test_wrappers_do_not_apply_to_calls(self):
         caller, handler = ControlSystemCaller(Passive()), ControlSystemHandler(Passive())
         with World() as world:
             with pytest.raises(AssertionError):
-                world.connect(caller, handler, emitter_wrapper=lambda e: e)
+                # A call takes no wrapper, so `connect` has no overload for one: the assertion is what says so.
+                world.connect(caller, handler, emitter_wrapper=lambda e: e)  # pyright: ignore[reportArgumentType]
 
     def test_in_process(self):
         client, adder = Client([(1, 2), (-1, 2), (3, 4)]), Adder(defer=2)

pimm/tests/test_world.py

@@ -9,6 +9,7 @@
 
 import pytest
 
+import pimm.world
 from pimm.core import (
     ControlSystem,
     ControlSystemEmitter,
@@ -26,7 +27,15 @@
 from pimm.logging import LOG_LEVEL_ENV
 from pimm.shared_memory import SMCompliant
 from pimm.tests.testing import MockClock
-from pimm.world import EventReceiver, LocalQueueEmitter, QueueEmitter, SystemClock, VirtualClock, World
+from pimm.world import (
+    EventReceiver,
+    LocalQueueEmitter,
+    MultiprocessReceiver,
+    QueueEmitter,
+    SystemClock,
+    VirtualClock,
+    World,
+)
 
 
 def dummy_process(stop_reader, clock):
@@ -99,6 +108,63 @@ def read_from_buffer(self, buffer: memoryview | bytes) -> None:
         self.value = struct.unpack('d', buffer[:8])[0]
 
 
+def test_an_interrupt_taken_inside_a_send_is_what_stops_the_process(monkeypatch):
+    """Nothing installs a handler in a process for it: the transport it interrupts is what records it.
+
+    A main-process control system reaches the same transports a background one does, so an interrupt that
+    tears a connection has to be noted where the tearing happens.
+    """
+    monkeypatch.setattr(pimm.world, '_interrupted', False)
+
+    def torn(data, ts):
+        raise KeyboardInterrupt
+
+    with World() as world:
+        emitter, receiver = world.mp_pipes()
+        assert not isinstance(receiver, list)
+        emitter.emit('before', ts=1)
+        assert receiver.read() is not None
+
+        monkeypatch.setattr(emitter, '_emit_queue', torn)
+        with pytest.raises(KeyboardInterrupt):
+            emitter.emit('torn', ts=2)
+
+        assert pimm.world._interrupted, 'the interrupt went unrecorded'
+        assert receiver.read() is None
+
+
+def test_a_process_that_took_an_interrupt_stops_talking_to_the_manager(monkeypatch):
+    """An interrupt can land inside a call to the manager, and that connection then holds half a message.
+
+    Reading it again returns what another call asked for, so a reader takes a value off a channel it never
+    subscribed to. A process that has taken one neither reads nor sends after it.
+    """
+    with World() as world:
+        emitter, receiver = world.mp_pipes()
+        assert not isinstance(receiver, list)
+        emitter.emit('before', ts=1)
+        assert receiver.read() is not None
+
+        monkeypatch.setattr(pimm.world, '_interrupted', True)
+
+        emitter.emit('after', ts=2)  # dropped, not sent
+        assert receiver.read() is None
+
+
+def test_a_queue_that_answers_with_anything_but_a_message_says_the_connection_is_torn():
+    """A connection torn by an interrupt answers with what another call asked for, of whatever type that
+    call wanted -- a float as readily as nothing. Reading `.data` off it would hide the interrupt."""
+    with World() as world:
+        emitter, receiver = world.mp_pipes()
+        assert isinstance(receiver, MultiprocessReceiver)
+        emitter.emit('before', ts=1)  # settles the channel on the queue transport
+        assert receiver.read() is not None
+        receiver._queue.put(0.5)  # what the torn connection hands back
+
+        with pytest.raises(ConnectionError, match='tore its connection'):
+            receiver.read()
+
+
 class TestQueueEmitter:
     """Test the QueueEmitter class."""
 

pimm/world.py

@@ -11,6 +11,7 @@
 import traceback
 from collections import Counter, defaultdict, deque
 from collections.abc import Callable, Iterator, Mapping
+from contextlib import contextmanager
 from enum import IntEnum
 from multiprocessing import resource_tracker
 from multiprocessing.managers import ValueProxy
@@ -42,6 +43,7 @@
 logger = logging.getLogger(__name__)
 
 T = TypeVar('T')
+U = TypeVar('U')
 Req = TypeVar('Req')
 Res = TypeVar('Res')
 
@@ -75,6 +77,28 @@ def emit(self, data: T, ts: int = -1):
                 pass
 
 
+# Set in a process that has taken an interrupt. An interrupt can land inside a call to the manager, and
+# that connection then holds half a message: the next call over it returns what another one asked for, so
+# a reader takes a value from a channel it never subscribed to. Nothing may be sent or read after it.
+_interrupted = False
+
+
+@contextmanager
+def _noting_interrupt() -> Iterator[None]:
+    """Record an interrupt taken inside the block, and let it go on.
+
+    A connection is torn by an interrupt that lands in the middle of a call over it, so every process that
+    reaches a transport records its own -- there is nowhere else the tearing can happen, and no process has
+    to have had a handler installed for it.
+    """
+    global _interrupted
+    try:
+        yield
+    except KeyboardInterrupt:
+        _interrupted = True
+        raise
+
+
 class MultiprocessEmitter(SignalEmitter[T]):
     """Signal emitter that transparently bridges processes.
 
@@ -118,7 +142,8 @@ def __init__(
     @property
     def transport_mode(self) -> TransportMode:
         if self._mode is TransportMode.UNDECIDED:
-            self._mode = TransportMode(self._mode_value.value)
+            with _noting_interrupt():  # reading the manager is where a connection is torn
+                self._mode = TransportMode(self._mode_value.value)
         return self._mode
 
     @property
@@ -188,9 +213,12 @@ def _emit_shared_memory(self, data: SMCompliant, ts: int) -> bool:
 
         return True
 
+    @_noting_interrupt()
     def emit(self, data: T, ts: int = -1):
+        if _interrupted:
+            return
         ts = ts if ts >= 0 else self._clock.now_ns()
-        mode = self._ensure_mode(data)
+        mode = self._ensure_mode(data)  # itself a call to the manager, so it sits inside the guard
 
         if mode is TransportMode.SHARED_MEMORY:
             if not isinstance(data, SMCompliant):
@@ -262,7 +290,8 @@ def __init__(
     @property
     def transport_mode(self) -> TransportMode:
         if self._mode is TransportMode.UNDECIDED:
-            self._mode = TransportMode(self._mode_value.value)
+            with _noting_interrupt():  # reading the manager is where a connection is torn
+                self._mode = TransportMode(self._mode_value.value)
         return self._mode
 
     @property
@@ -275,6 +304,10 @@ def _read_queue(self) -> Message[T] | None:
         except Empty:
             message = None
         else:
+            if not isinstance(message, Message):
+                # An interrupt that lands inside a manager call leaves that connection holding half a
+                # message, and every read after it comes back as whatever another call asked for.
+                raise ConnectionError(f'the queue was read after an interrupt tore its connection: {message!r}')
             self._last_queue_message = Message(message.data, message.ts, True)
             if self._mode is TransportMode.UNDECIDED:
                 self._mode = TransportMode.QUEUE
@@ -332,8 +365,11 @@ def _read_shared_memory(self) -> Message[T] | None:
             self._up_value.value = False
             return Message(data=self._out_value, ts=self._ts_value.value, updated=updated)  # instead of True
 
+    @_noting_interrupt()
     def read(self) -> Message[T] | None:
-        mode = self.transport_mode
+        if _interrupted:
+            return None
+        mode = self.transport_mode  # itself a call to the manager, so it sits inside the guard
 
         if mode is TransportMode.SHARED_MEMORY:
             return self._read_shared_memory()
@@ -642,13 +678,35 @@ def interleave(self, *loops: ControlLoop) -> Iterator[Command]:
             self._advance_to(target_ns)
             yield Sleep(wait_ns / 1e9) if wait_ns else Yield()
 
+    @overload
+    def connect(self, source: ControlSystemCaller[Req, Res], target: ControlSystemHandler[Req, Res]) -> None: ...
+
+    @overload
+    def connect(
+        self,
+        source: ControlSystemEmitter[T],
+        target: ControlSystemReceiver[T],
+        *,
+        emitter_wrapper: Callable[[SignalEmitter[T]], SignalEmitter[T]] = ...,
+    ) -> None: ...
+
+    @overload
+    def connect(
+        self,
+        source: ControlSystemEmitter[T],
+        target: ControlSystemReceiver[U],
+        *,
+        emitter_wrapper: Callable[[SignalEmitter[T]], SignalEmitter[T]] = ...,
+        receiver_wrapper: Callable[[SignalReceiver[T]], SignalReceiver[U]],
+    ) -> None: ...
+
     def connect(
         self,
         source: ControlSystemEmitter[T] | ControlSystemCaller[Req, Res],
-        target: ControlSystemReceiver[T] | ControlSystemHandler[Req, Res],
+        target: ControlSystemReceiver[U] | ControlSystemHandler[Req, Res],
         *,
         emitter_wrapper: Callable[[SignalEmitter[T]], SignalEmitter[T]] = identity,
-        receiver_wrapper: Callable[[SignalReceiver[T]], SignalReceiver[T]] = identity,
+        receiver_wrapper: Callable[[SignalReceiver[T]], SignalReceiver[U]] = identity,
     ) -> None:
         """Declare a logical connection: an Emitter feeding a Receiver, or a Caller invoking a Handler.
 
@@ -663,7 +721,8 @@ def connect(
             emitter_wrapper: Optional function to wrap the underlying SignalEmitter
                            before binding. Defaults to identity function.
             receiver_wrapper: Optional function to wrap the underlying SignalReceiver
-                            before binding. Defaults to identity function.
+                            before binding. Defaults to identity function. It is the only way the
+                            receiver may carry a type other than the emitter's.
 
         The wrapper functions allow for transformation or decoration of the
         underlying signal transport mechanisms, such as adding logging,

positronic/tests/test_components.py

@@ -1,3 +1,4 @@
+import logging
 import pickle
 from collections.abc import Callable
 from importlib import import_module
@@ -8,11 +9,24 @@
 import pimm
 from pimm.core import ControlSystem
 
+logger = logging.getLogger(__name__)
+
+_OURS = ('positronic', 'pimm')
+
 
 def _optional_import(module: str, symbol: str) -> Any | None:
+    """``symbol``, or ``None`` where the vendor package it needs is not usable on this machine.
+
+    A vendor that is installed but cannot load — its own native library is missing — leaves this station
+    without the component just as an absent package does. A project module that fails to import is a broken
+    build rather than a missing vendor, and is not one of those.
+    """
     try:
         return getattr(import_module(module), symbol)
-    except ModuleNotFoundError:  # pragma: no cover - optional dependency
+    except ImportError as e:  # pragma: no cover - optional dependency
+        if (e.name or '').split('.')[0] in _OURS:
+            raise
+        logger.error('%s is not available: %s', module, e)
         return None
 
 

positronic/wire.py

@@ -16,7 +16,7 @@ def wire(  # noqa: C901
     world: pimm.World,
     harness: pimm.ControlSystem,
     dataset_factory: DatasetFactory | None,
-    cameras: Mapping[str, pimm.SignalEmitter],
+    cameras: Mapping[str, pimm.ControlSystemEmitter],
     robot_arm: pimm.ControlSystem | None,
     gripper: pimm.ControlSystem | None,
     gui: pimm.ControlSystem | None,
@@ -101,7 +101,7 @@ def wire_embodiment(
     *,
     record: bool = True,
     privileged: dict[str, Observation] | None = None,
-    done: pimm.SignalEmitter | None = None,
+    done: pimm.ControlSystemEmitter | None = None,
 ):
     """Wire an embodiment to the Harness for the inference path.