From 842d703032f22042cc879ddc2dc98dfbb43254e6 Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Fri, 28 Aug 2026 18:30:11 +0300 Subject: [PATCH 01/11] Say that an interrupt tore a queue's connection An interrupt that lands inside a manager call leaves that connection holding half a message, and the next read comes back as something the queue never carried. The reader took it for a message and raised `AttributeError` over the real reason the run ended. --- pimm/world.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/pimm/world.py b/pimm/world.py index fcd47c495..034967dbb 100644 --- a/pimm/world.py +++ b/pimm/world.py @@ -275,6 +275,10 @@ def _read_queue(self) -> Message[T] | None: except Empty: message = None else: + if message is None: + # An interrupt that lands inside a manager call leaves that connection holding half a + # message, and every read after it comes back as something the queue never carried. + raise ConnectionError('the queue was read after an interrupt tore its connection') self._last_queue_message = Message(message.data, message.ts, True) if self._mode is TransportMode.UNDECIDED: self._mode = TransportMode.QUEUE From 6b3220ffe7305f541c4aaa8374bcaab1c88e2a42 Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Fri, 28 Aug 2026 19:46:57 +0300 Subject: [PATCH 02/11] Stop a process that took an interrupt from reading its neighbours' messages An interrupt lands in every process of a run at once, and it can land inside a call to the manager. That connection then holds half a message, and the next call over it returns what another one asked for: the recorder read a grip value off the channel that carries its commands and raised `AttributeError` over the real reason the run ended. A process that has taken an interrupt now neither sends nor reads. What it would have carried is going nowhere anyway: every other process is stopping. --- pimm/tests/test_world.py | 18 ++++++++++++++++++ pimm/world.py | 18 ++++++++++++++++++ 2 files changed, 36 insertions(+) diff --git a/pimm/tests/test_world.py b/pimm/tests/test_world.py index 9c7dbe92c..81e4bd480 100644 --- a/pimm/tests/test_world.py +++ b/pimm/tests/test_world.py @@ -9,6 +9,7 @@ import pytest +import pimm.world from pimm.core import ( ControlSystem, ControlSystemEmitter, @@ -99,6 +100,23 @@ def read_from_buffer(self, buffer: memoryview | bytes) -> None: self.value = struct.unpack('d', buffer[:8])[0] +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() + 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 + + class TestQueueEmitter: """Test the QueueEmitter class.""" diff --git a/pimm/world.py b/pimm/world.py index 034967dbb..522f64ffc 100644 --- a/pimm/world.py +++ b/pimm/world.py @@ -6,6 +6,7 @@ import multiprocessing as mp import multiprocessing.shared_memory import os +import signal import sys import time import traceback @@ -141,6 +142,8 @@ def _ensure_mode(self, data: T) -> TransportMode: return self._mode def _emit_queue(self, data: T, ts: int) -> bool: + if _interrupted: + return False msg = Message(data, ts) success = False @@ -220,6 +223,18 @@ def __del__(self): self.close() +# 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 + + +def _note_interrupt(signum, frame): + global _interrupted + _interrupted = True + raise KeyboardInterrupt + + class MultiprocessReceiver(SignalReceiver[T]): """Signal receiver companion for :class:`MultiprocessEmitter`. @@ -270,6 +285,8 @@ def uses_shared_memory(self) -> bool: return self.transport_mode is TransportMode.SHARED_MEMORY def _read_queue(self) -> Message[T] | None: + if _interrupted: + return None try: message = self._queue.get_nowait() except Empty: @@ -470,6 +487,7 @@ def __call__(self, should_stop: SignalReceiver, clock: Clock) -> Iterator[Comman def _bg_wrapper( run_func: ControlLoop, stop_event: EventClass, clock: Clock, name: str, parent_component_levels: Mapping[str, int] ): + signal.signal(signal.SIGINT, _note_interrupt) try: # A freshly spawned subprocess carries no logging configuration, so set one up. It is inside # the `try` because a failure here must still reach the `finally` that stops the World. From a2baca0936aff12d281377518d49a0d218243978 Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Fri, 28 Aug 2026 17:50:33 +0300 Subject: [PATCH 03/11] Let a vendor that cannot load stand in for one that is absent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `_optional_import` caught only a missing package. A vendor that is installed but whose own native library is not — `pyzed` without the ZED SDK — raises plain `ImportError`, and that ended the collection of the whole module. A station without the component is in the same place either way. --- positronic/tests/test_components.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/positronic/tests/test_components.py b/positronic/tests/test_components.py index 2f9d0a645..29e23ef38 100644 --- a/positronic/tests/test_components.py +++ b/positronic/tests/test_components.py @@ -12,7 +12,9 @@ def _optional_import(module: str, symbol: str) -> Any | None: try: return getattr(import_module(module), symbol) - except ModuleNotFoundError: # pragma: no cover - optional dependency + # 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. + except ImportError: # pragma: no cover - optional dependency return None From b2fb0377263fac20b8a9540cef1ac243b88e8ff8 Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Fri, 28 Aug 2026 17:59:54 +0300 Subject: [PATCH 04/11] Let a connection wrapper change the type it carries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `pimm.map` maps a `SignalReceiver[T]` to a `SignalReceiver[U]` — its docstring says so, and data collection uses it to hand the headset the array inside a camera frame. `World.connect` typed the wrapper as type-preserving, so the call read as an error as soon as the camera port carried a type of its own. --- pimm/world.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/pimm/world.py b/pimm/world.py index 522f64ffc..eb2c79960 100644 --- a/pimm/world.py +++ b/pimm/world.py @@ -43,6 +43,7 @@ logger = logging.getLogger(__name__) T = TypeVar('T') +U = TypeVar('U') Req = TypeVar('Req') Res = TypeVar('Res') @@ -667,10 +668,10 @@ def interleave(self, *loops: ControlLoop) -> Iterator[Command]: 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. From d5e4fa226c5c9682a26e611dca56aa243ef4cee1 Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Thu, 3 Sep 2026 19:25:14 +0300 Subject: [PATCH 05/11] Note an interrupt where a connection is torn, not where a handler was installed Four things the review of this branch found. The guard sat inside `_emit_queue` and `_read_queue`, which the public `emit` and `read` reach only after `_ensure_mode` and `transport_mode` have already called the manager -- and the shared memory path, which holds a manager lock for every frame, never reached it at all. Both public methods check first now. Tracking the interrupt from a handler `_bg_wrapper` installs left every main-process control system out: those reach the same transports, and their interrupt went unrecorded. A connection is torn only by an interrupt that lands inside a call over it, so the transport records its own -- `emit` and `read` re-raise what they catch, and no process depends on a handler being there. The handler and the `signal` import go. `World.connect` typed the receiver with a free `U`, which let an emitter of one type feed a receiver of another whenever no wrapper was passed. Overloads now ask for the emitter's own type unless a `receiver_wrapper` is given. That made `wire` say what it has always required -- `connect` asserts a `ControlSystemEmitter`, where the signature promised any `SignalEmitter` -- and six errors leave the basedpyright baseline. `_optional_import` caught every `ImportError`, which turns a broken project import into a skipped test. A failure inside `positronic` or `pimm` is raised now; a vendor's is logged at ERROR before the component is given up. --- pimm/tests/test_world.py | 26 +++++++++ pimm/world.py | 88 ++++++++++++++++++++--------- positronic/tests/test_components.py | 18 +++++- positronic/wire.py | 4 +- 4 files changed, 104 insertions(+), 32 deletions(-) diff --git a/pimm/tests/test_world.py b/pimm/tests/test_world.py index 81e4bd480..beb8e4cac 100644 --- a/pimm/tests/test_world.py +++ b/pimm/tests/test_world.py @@ -100,6 +100,31 @@ 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. @@ -108,6 +133,7 @@ def test_a_process_that_took_an_interrupt_stops_talking_to_the_manager(monkeypat """ 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 diff --git a/pimm/world.py b/pimm/world.py index eb2c79960..8bd797e26 100644 --- a/pimm/world.py +++ b/pimm/world.py @@ -6,12 +6,12 @@ import multiprocessing as mp import multiprocessing.shared_memory import os -import signal import sys import time 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 @@ -143,8 +143,6 @@ def _ensure_mode(self, data: T) -> TransportMode: return self._mode def _emit_queue(self, data: T, ts: int) -> bool: - if _interrupted: - return False msg = Message(data, ts) success = False @@ -193,16 +191,19 @@ def _emit_shared_memory(self, data: SMCompliant, ts: int) -> bool: return True def emit(self, data: T, ts: int = -1): - ts = ts if ts >= 0 else self._clock.now_ns() - mode = self._ensure_mode(data) - - if mode is TransportMode.SHARED_MEMORY: - if not isinstance(data, SMCompliant): - raise TypeError('Shared memory transport selected; data must implement SMCompliant') - self._emit_shared_memory(data, ts) + if _interrupted: return + with _noting_interrupt(): + ts = ts if ts >= 0 else self._clock.now_ns() + 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): + raise TypeError('Shared memory transport selected; data must implement SMCompliant') + self._emit_shared_memory(data, ts) + return - self._emit_queue(data, ts) + self._emit_queue(data, ts) def close(self) -> None: if self._closed: @@ -230,10 +231,20 @@ def __del__(self): _interrupted = False -def _note_interrupt(signum, frame): +@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 - _interrupted = True - raise KeyboardInterrupt + try: + yield + except KeyboardInterrupt: + _interrupted = True + raise class MultiprocessReceiver(SignalReceiver[T]): @@ -286,8 +297,6 @@ def uses_shared_memory(self) -> bool: return self.transport_mode is TransportMode.SHARED_MEMORY def _read_queue(self) -> Message[T] | None: - if _interrupted: - return None try: message = self._queue.get_nowait() except Empty: @@ -355,19 +364,22 @@ def _read_shared_memory(self) -> Message[T] | None: return Message(data=self._out_value, ts=self._ts_value.value, updated=updated) # instead of True def read(self) -> Message[T] | None: - mode = self.transport_mode + if _interrupted: + return None + with _noting_interrupt(): + 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() + if mode is TransportMode.SHARED_MEMORY: + return self._read_shared_memory() - message = self._read_queue() - if message is not None: - return message + message = self._read_queue() + if message is not None: + return message - if mode is TransportMode.UNDECIDED: - # No data yet; underlying transport still undecided. + if mode is TransportMode.UNDECIDED: + # No data yet; underlying transport still undecided. + return None return None - return None def close(self) -> None: if self._closed: @@ -488,7 +500,6 @@ def __call__(self, should_stop: SignalReceiver, clock: Clock) -> Iterator[Comman def _bg_wrapper( run_func: ControlLoop, stop_event: EventClass, clock: Clock, name: str, parent_component_levels: Mapping[str, int] ): - signal.signal(signal.SIGINT, _note_interrupt) try: # A freshly spawned subprocess carries no logging configuration, so set one up. It is inside # the `try` because a failure here must still reach the `finally` that stops the World. @@ -665,6 +676,28 @@ 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], @@ -686,7 +719,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, diff --git a/positronic/tests/test_components.py b/positronic/tests/test_components.py index 29e23ef38..5bb438da2 100644 --- a/positronic/tests/test_components.py +++ b/positronic/tests/test_components.py @@ -1,3 +1,4 @@ +import logging import pickle from collections.abc import Callable from importlib import import_module @@ -8,13 +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) - # 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. - except ImportError: # 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 diff --git a/positronic/wire.py b/positronic/wire.py index d49225f94..965eb2b9a 100644 --- a/positronic/wire.py +++ b/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. From 004a5a88fb8993af5231141231e1d09c35b19893 Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Mon, 7 Sep 2026 19:26:21 +0300 Subject: [PATCH 06/11] Put the interrupt guard on the transport methods, above their first user `_noting_interrupt` is a `contextmanager`, so it is a `ContextDecorator` too: `emit` and `read` wear it instead of indenting their whole bodies under it. Both it and `_interrupted` were defined between the emitter that first uses them and the receiver that uses them next; they sit above the emitter now. --- pimm/world.py | 82 +++++++++++++++++++++++++-------------------------- 1 file changed, 41 insertions(+), 41 deletions(-) diff --git a/pimm/world.py b/pimm/world.py index 8bd797e26..ccdff070a 100644 --- a/pimm/world.py +++ b/pimm/world.py @@ -77,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. @@ -190,20 +212,20 @@ 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 - with _noting_interrupt(): - ts = ts if ts >= 0 else self._clock.now_ns() - mode = self._ensure_mode(data) # itself a call to the manager, so it sits inside the guard + ts = ts if ts >= 0 else self._clock.now_ns() + 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): - raise TypeError('Shared memory transport selected; data must implement SMCompliant') - self._emit_shared_memory(data, ts) - return + if mode is TransportMode.SHARED_MEMORY: + if not isinstance(data, SMCompliant): + raise TypeError('Shared memory transport selected; data must implement SMCompliant') + self._emit_shared_memory(data, ts) + return - self._emit_queue(data, ts) + self._emit_queue(data, ts) def close(self) -> None: if self._closed: @@ -225,28 +247,6 @@ def __del__(self): self.close() -# 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 MultiprocessReceiver(SignalReceiver[T]): """Signal receiver companion for :class:`MultiprocessEmitter`. @@ -363,23 +363,23 @@ 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: if _interrupted: return None - with _noting_interrupt(): - mode = self.transport_mode # itself a call to the manager, so it sits inside the guard + 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() + if mode is TransportMode.SHARED_MEMORY: + return self._read_shared_memory() - message = self._read_queue() - if message is not None: - return message + message = self._read_queue() + if message is not None: + return message - if mode is TransportMode.UNDECIDED: - # No data yet; underlying transport still undecided. - return None + if mode is TransportMode.UNDECIDED: + # No data yet; underlying transport still undecided. return None + return None def close(self) -> None: if self._closed: From 851da7a7d1804fe516a7359ee0b66b0f98e042ef Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Mon, 7 Sep 2026 20:13:08 +0300 Subject: [PATCH 07/11] Say in the test that a call takes no connection wrapper `World.connect` gained overloads on this branch, and none of them takes a wrapper beside a caller and a handler. The test that asks for one and expects the assertion now says the type refuses it too. --- pimm/tests/test_calls.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pimm/tests/test_calls.py b/pimm/tests/test_calls.py index 09187272e..45c4699c8 100644 --- a/pimm/tests/test_calls.py +++ b/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) From df49ba8d83931c99171edd14bf90d5d0817dbe3f Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Mon, 7 Sep 2026 20:13:36 +0300 Subject: [PATCH 08/11] Take the type errors this branch removed out of the baseline The overloads on `World.connect` end five errors the baseline grandfathered. The ratchet only refuses growth, so CI would prune them on its own; the branch carries the pruned file instead. --- .basedpyright/baseline.json | 42 +------------------------------------ 1 file changed, 1 insertion(+), 41 deletions(-) diff --git a/.basedpyright/baseline.json b/.basedpyright/baseline.json index 06a39d376..f4b52256b 100644 --- a/.basedpyright/baseline.json +++ b/.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,14 +6927,6 @@ "lineCount": 1 } }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 26, - "endColumn": 33, - "lineCount": 1 - } - }, { "code": "reportAttributeAccessIssue", "range": { @@ -6966,15 +6934,7 @@ "endColumn": 46, "lineCount": 1 } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 22, - "endColumn": 26, - "lineCount": 1 - } } ] } -} +} \ No newline at end of file From f09172d90d11c1118d3acda217a2aa917b19ff97 Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Tue, 8 Sep 2026 09:07:11 +0300 Subject: [PATCH 09/11] Refuse every answer a torn connection gives, not only a missing one The guard asked whether the queue answered with `None`. A connection torn by an interrupt answers with what another call asked for, of whatever type that call wanted -- the incident this came from produced a float, and `.data` off it raises `AttributeError` and hides the interrupt. Anything that is not a `Message` says the connection is torn, and the answer is in the error. --- pimm/tests/test_world.py | 14 ++++++++++++++ pimm/world.py | 6 +++--- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/pimm/tests/test_world.py b/pimm/tests/test_world.py index beb8e4cac..0dcd0b1c4 100644 --- a/pimm/tests/test_world.py +++ b/pimm/tests/test_world.py @@ -143,6 +143,20 @@ def test_a_process_that_took_an_interrupt_stops_talking_to_the_manager(monkeypat 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 it was: + the incident that named this produced a float. Reading `.data` off it would hide the interrupt.""" + with World() as world: + emitter, receiver = world.mp_pipes() + assert not isinstance(receiver, list) + 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.""" diff --git a/pimm/world.py b/pimm/world.py index ccdff070a..6bdc036fa 100644 --- a/pimm/world.py +++ b/pimm/world.py @@ -302,10 +302,10 @@ def _read_queue(self) -> Message[T] | None: except Empty: message = None else: - if message is None: + 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 something the queue never carried. - raise ConnectionError('the queue was read after an interrupt tore its connection') + # 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 From ee97552f2072480c16d9439769f2b5702bcccb86 Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Tue, 8 Sep 2026 09:17:29 +0300 Subject: [PATCH 10/11] Name the receiver the torn-connection test reaches into `mp_pipes` answers with a `SignalReceiver`, which carries no queue; the test says which receiver it has. --- pimm/tests/test_world.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/pimm/tests/test_world.py b/pimm/tests/test_world.py index 0dcd0b1c4..b1848be52 100644 --- a/pimm/tests/test_world.py +++ b/pimm/tests/test_world.py @@ -27,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): @@ -148,7 +156,7 @@ def test_a_queue_that_answers_with_anything_but_a_message_says_the_connection_is the incident that named this produced a float. Reading `.data` off it would hide the interrupt.""" with World() as world: emitter, receiver = world.mp_pipes() - assert not isinstance(receiver, list) + 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 From 89c3831838cfecb313303a8109f562c55eae4af6 Mon Sep 17 00:00:00 2001 From: DarksaCY Date: Tue, 8 Sep 2026 09:38:59 +0300 Subject: [PATCH 11/11] Record an interrupt taken while a channel reads its mode `transport_mode` reads the manager when the mode is still undecided, and that read sits outside `emit` and `read`. An interrupt landing in it left the latch false, and the teardown after it used the torn connection. The test that names a torn connection also told the incident it came from rather than what has to hold. --- pimm/tests/test_world.py | 4 ++-- pimm/world.py | 6 ++++-- 2 files changed, 6 insertions(+), 4 deletions(-) diff --git a/pimm/tests/test_world.py b/pimm/tests/test_world.py index b1848be52..f2d53e93f 100644 --- a/pimm/tests/test_world.py +++ b/pimm/tests/test_world.py @@ -152,8 +152,8 @@ def test_a_process_that_took_an_interrupt_stops_talking_to_the_manager(monkeypat 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 it was: - the incident that named this produced a float. Reading `.data` off it would hide the interrupt.""" + """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) diff --git a/pimm/world.py b/pimm/world.py index 6bdc036fa..daaf3c5da 100644 --- a/pimm/world.py +++ b/pimm/world.py @@ -142,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 @@ -289,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