Flip a last `MultiError` to a beg, add todo on `@stream` func

Such that the top level `maybe_enable_greenback` from
`open_root_actor()` can toggle the entire actor tree's usage.
Read the rtv in `._rpc` tasks and only enable if set.

Also, rigor up the `._rpc.process_messages()` loop to handle `Error()`
and `case _:` separately such that we now raise an explicit rte for
unknown / invalid msgs. Use "parent" / "child" for side descriptions in
loop comments and put a fat comment before the `StartAck` in `_invoke()`.

examples/debugging/multi_subactors.py

@@ -38,6 +38,7 @@ async def main():
     """
     async with tractor.open_nursery(
         debug_mode=True,
+        # loglevel='runtime',
     ) as n:
 
         # Spawn both actors, don't bother with collecting results

examples/debugging/subactor_breakpoint.py

@@ -3,17 +3,20 @@ import tractor
 
 
 async def breakpoint_forever():
-    """Indefinitely re-enter debugger in child actor.
+    '''
-    """
+    Indefinitely re-enter debugger in child actor.
+
+    '''
     while True:
         await trio.sleep(0.1)
-        await tractor.breakpoint()
+        await tractor.pause()
 
 
 async def main():
 
     async with tractor.open_nursery(
         debug_mode=True,
+        loglevel='cancel',
     ) as n:
 
         portal = await n.run_in_actor(

tests/test_context_stream_semantics.py

@@ -246,10 +246,10 @@ def test_simple_context(
             trio.run(main)
         except error_parent:
             pass
-        except trio.MultiError as me:
+        except BaseExceptionGroup as beg:
             # XXX: on windows it seems we may have to expect the group error
             from tractor._exceptions import is_multi_cancelled
-            assert is_multi_cancelled(me)
+            assert is_multi_cancelled(beg)
     else:
         trio.run(main)
 

tests/test_legacy_one_way_streaming.py

@@ -38,10 +38,13 @@ async def async_gen_stream(sequence):
     assert cs.cancelled_caught
 
 
+# TODO: deprecated either remove entirely
+# or re-impl in terms of `MsgStream` one-sides
+# wrapper, but at least remove `Portal.open_stream_from()`
 @tractor.stream
 async def context_stream(
     ctx: tractor.Context,
-    sequence
+    sequence: list[int],
 ):
     for i in sequence:
         await ctx.send_yield(i)

tractor/_child.py

@@ -36,6 +36,7 @@ def parse_ipaddr(arg):
 
 
 if __name__ == "__main__":
+    __tracebackhide__: bool = True
 
     parser = argparse.ArgumentParser()
     parser.add_argument("--uid", type=parse_uid)

tractor/_context.py

@@ -47,6 +47,7 @@ import trio
 from ._exceptions import (
     ContextCancelled,
     InternalError,
+    MsgTypeError,
     RemoteActorError,
     StreamOverrun,
     pack_from_raise,
@@ -59,12 +60,14 @@ from .msg import (
     MsgType,
     MsgCodec,
     NamespacePath,
+    PayloadT,
     Return,
     Started,
     Stop,
     Yield,
     current_codec,
     pretty_struct,
+    types as msgtypes,
 )
 from ._ipc import Channel
 from ._streaming import MsgStream
@@ -88,7 +91,10 @@ async def _drain_to_final_msg(
     hide_tb: bool = True,
     msg_limit: int = 6,
 
-) -> list[dict]:
+) -> tuple[
+    Return|None,
+    list[MsgType]
+]:
     '''
     Drain IPC msgs delivered to the underlying rx-mem-chan
     `Context._recv_chan` from the runtime in search for a final
@@ -109,6 +115,7 @@ async def _drain_to_final_msg(
     # basically ignoring) any bi-dir-stream msgs still in transit
     # from the far end.
     pre_result_drained: list[MsgType] = []
+    return_msg: Return|None = None
     while not (
         ctx.maybe_error
         and not ctx._final_result_is_set()
@@ -169,8 +176,6 @@ async def _drain_to_final_msg(
             # pray to the `trio` gawds that we're corrent with this
             # msg: dict = await ctx._recv_chan.receive()
             msg: MsgType = await ctx._recv_chan.receive()
-            # always capture unexpected/non-result msgs
-            pre_result_drained.append(msg)
 
         # NOTE: we get here if the far end was
         # `ContextCancelled` in 2 cases:
@@ -207,11 +212,13 @@ async def _drain_to_final_msg(
                 # if ctx._recv_chan:
                 #     await ctx._recv_chan.aclose()
                 # TODO: ^ we don't need it right?
+                return_msg = msg
                 break
 
             # far end task is still streaming to us so discard
             # and report depending on local ctx state.
             case Yield():
+                pre_result_drained.append(msg)
                 if (
                     (ctx._stream.closed
                      and (reason := 'stream was already closed')
@@ -236,7 +243,10 @@ async def _drain_to_final_msg(
 
                         f'{pformat(msg)}\n'
                     )
-                    return pre_result_drained
+                    return (
+                        return_msg,
+                        pre_result_drained,
+                    )
 
                 # drain up to the `msg_limit` hoping to get
                 # a final result or error/ctxc.
@@ -260,6 +270,7 @@ async def _drain_to_final_msg(
             # -[ ] should be a runtime error if a stream is open right?
             # Stop()
             case Stop():
+                pre_result_drained.append(msg)
                 log.cancel(
                     'Remote stream terminated due to "stop" msg:\n\n'
                     f'{pformat(msg)}\n'
@@ -269,7 +280,6 @@ async def _drain_to_final_msg(
             # remote error msg, likely already handled inside
             # `Context._deliver_msg()`
             case Error():
-
                 # TODO: can we replace this with `ctx.maybe_raise()`?
                 # -[ ]  would this be handier for this case maybe?
                 #     async with maybe_raise_on_exit() as raises:
@@ -336,6 +346,7 @@ async def _drain_to_final_msg(
             # XXX should pretty much never get here unless someone
             # overrides the default `MsgType` spec.
             case _:
+                pre_result_drained.append(msg)
                 # It's definitely an internal error if any other
                 # msg type without a`'cid'` field arrives here!
                 if not msg.cid:
@@ -352,7 +363,10 @@ async def _drain_to_final_msg(
             f'{ctx.outcome}\n'
         )
 
-    return pre_result_drained
+    return (
+        return_msg,
+        pre_result_drained,
+    )
 
 
 class Unresolved:
@@ -719,21 +733,36 @@ class Context:
         Return string indicating which task this instance is wrapping.
 
         '''
-        return 'caller' if self._portal else 'callee'
+        return 'parent' if self._portal else 'child'
 
+    @staticmethod
+    def peer_side(side: str) -> str:
+        match side:
+            case 'child':
+                return 'parent'
+            case 'parent':
+                return 'child'
+
+    # TODO: remove stat!
+    # -[ ] re-implement the `.experiemental._pubsub` stuff
+    #     with `MsgStream` and that should be last usage?
+    # -[ ] remove from `tests/legacy_one_way_streaming.py`!
     async def send_yield(
         self,
         data: Any,
-
     ) -> None:
+        '''
+        Deprecated method for what now is implemented in `MsgStream`.
 
+        We need to rework / remove some stuff tho, see above.
+
+        '''
         warnings.warn(
             "`Context.send_yield()` is now deprecated. "
             "Use ``MessageStream.send()``. ",
             DeprecationWarning,
             stacklevel=2,
         )
-        # await self.chan.send({'yield': data, 'cid': self.cid})
         await self.chan.send(
             Yield(
                 cid=self.cid,
@@ -742,12 +771,11 @@ class Context:
         )
 
     async def send_stop(self) -> None:
-        # await pause()
+        '''
-        # await self.chan.send({
+        Terminate a `MsgStream` dialog-phase by sending the IPC
-        #     # Stop(
+        equiv of a `StopIteration`.
-        #     'stop': True,
+
-        #     'cid': self.cid
+        '''
-        # })
         await self.chan.send(
             Stop(cid=self.cid)
         )
@@ -843,6 +871,7 @@ class Context:
 
         # self-cancel (ack) or,
         # peer propagated remote cancellation.
+        msgtyperr: bool = False
         if isinstance(error, ContextCancelled):
 
             whom: str = (
@@ -854,6 +883,16 @@ class Context:
                 f'{error}'
             )
 
+        elif isinstance(error, MsgTypeError):
+            msgtyperr = True
+            peer_side: str = self.peer_side(self.side)
+            log.error(
+                f'IPC dialog error due to msg-type caused by {peer_side!r} side\n\n'
+
+                f'{error}\n'
+                f'{pformat(self)}\n'
+            )
+
         else:
             log.error(
                 f'Remote context error:\n\n'
@@ -894,9 +933,9 @@ class Context:
             # if `._cancel_called` then `.cancel_acked and .cancel_called`
             # always should be set.
             and not self._is_self_cancelled()
-
             and not cs.cancel_called
             and not cs.cancelled_caught
+            and not msgtyperr
         ):
             # TODO: it'd sure be handy to inject our own
             # `trio.Cancelled` subtype here ;)
@@ -1001,7 +1040,7 @@ class Context:
         # when the runtime finally receives it during teardown
         # (normally in `.result()` called from
         # `Portal.open_context().__aexit__()`)
-        if side == 'caller':
+        if side == 'parent':
             if not self._portal:
                 raise InternalError(
                     'No portal found!?\n'
@@ -1423,7 +1462,10 @@ class Context:
             # wait for a final context result/error by "draining"
             # (by more or less ignoring) any bi-dir-stream "yield"
             # msgs still in transit from the far end.
-            drained_msgs: list[dict] = await _drain_to_final_msg(
+            (
+                return_msg,
+                drained_msgs,
+            ) = await _drain_to_final_msg(
                 ctx=self,
                 hide_tb=hide_tb,
             )
@@ -1441,7 +1483,10 @@ class Context:
 
             log.cancel(
                 'Ctx drained pre-result msgs:\n'
-                f'{pformat(drained_msgs)}'
+                f'{pformat(drained_msgs)}\n\n'
+
+                f'Final return msg:\n'
+                f'{return_msg}\n'
             )
 
         self.maybe_raise(
@@ -1608,7 +1653,13 @@ class Context:
 
     async def started(
         self,
-        value: Any | None = None
+
+        # TODO: how to type this so that it's the
+        # same as the payload type? Is this enough?
+        value: PayloadT|None = None,
+
+        strict_parity: bool = False,
+        complain_no_parity: bool = True,
 
     ) -> None:
         '''
@@ -1629,7 +1680,7 @@ class Context:
                 f'called `.started()` twice on context with {self.chan.uid}'
             )
 
-        started = Started(
+        started_msg = Started(
             cid=self.cid,
             pld=value,
         )
@@ -1650,28 +1701,54 @@ class Context:
         # https://zguide.zeromq.org/docs/chapter7/#The-Cheap-or-Nasty-Pattern
         #
         codec: MsgCodec = current_codec()
-        msg_bytes: bytes = codec.encode(started)
+        msg_bytes: bytes = codec.encode(started_msg)
         try:
             # be a "cheap" dialog (see above!)
-            rt_started = codec.decode(msg_bytes)
+            if (
-            if rt_started != started:
+                strict_parity
+                or
+                complain_no_parity
+            ):
+                rt_started: Started = codec.decode(msg_bytes)
 
+                # XXX something is prolly totes cucked with the
+                # codec state!
+                if isinstance(rt_started, dict):
+                    rt_started = msgtypes.from_dict_msg(
+                        dict_msg=rt_started,
+                    )
+                    raise RuntimeError(
+                        'Failed to roundtrip `Started` msg?\n'
+                        f'{pformat(rt_started)}\n'
+                    )
+
+                if rt_started != started_msg:
                     # TODO: break these methods out from the struct subtype?
-                diff = pretty_struct.Struct.__sub__(rt_started, started)
 
+                    diff = pretty_struct.Struct.__sub__(
+                        rt_started,
+                        started_msg,
+                    )
                     complaint: str = (
                         'Started value does not match after codec rountrip?\n\n'
                         f'{diff}'
                     )
+
                     # TODO: rn this will pretty much always fail with
                     # any other sequence type embeded in the
                     # payload...
-                if self._strict_started:
+                    if (
+                        self._strict_started
+                        or
+                        strict_parity
+                    ):
                         raise ValueError(complaint)
                     else:
                         log.warning(complaint)
 
-            await self.chan.send(rt_started)
+                # started_msg = rt_started
+
+            await self.chan.send(started_msg)
 
         # raise any msg type error NO MATTER WHAT!
         except msgspec.ValidationError as verr:
@@ -1682,7 +1759,7 @@ class Context:
                 src_validation_error=verr,
                 verb_header='Trying to send payload'
                 # > 'invalid `Started IPC msgs\n'
-            )
+            ) from verr
 
         self._started_called = True
 
@@ -1783,13 +1860,17 @@ class Context:
             else:
                 log_meth = log.runtime
 
-            log_meth(
+            side: str = self.side
-                f'Delivering error-msg to caller\n\n'
 
-                f'<= peer: {from_uid}\n'
+            peer_side: str = self.peer_side(side)
+
+            log_meth(
+                f'Delivering IPC ctx error from {peer_side!r} to {side!r} task\n\n'
+
+                f'<= peer {peer_side!r}: {from_uid}\n'
                 f'  |_ {nsf}()\n\n'
 
-                f'=> cid: {cid}\n'
+                f'=> {side!r} cid: {cid}\n'
                 f'  |_{self._task}\n\n'
 
                 f'{pformat(re)}\n'
@@ -1804,6 +1885,7 @@ class Context:
             self._maybe_cancel_and_set_remote_error(re)
 
         # XXX only case where returning early is fine!
+        structfmt = pretty_struct.Struct.pformat
         if self._in_overrun:
             log.warning(
                 f'Queueing OVERRUN msg on caller task:\n'
@@ -1813,7 +1895,7 @@ class Context:
                 f'=> cid: {cid}\n'
                 f'  |_{self._task}\n\n'
 
-                f'{pformat(msg)}\n'
+                f'{structfmt(msg)}\n'
             )
             self._overflow_q.append(msg)
             return False
@@ -1827,7 +1909,7 @@ class Context:
                 f'=> {self._task}\n'
                 f'  |_cid={self.cid}\n\n'
 
-                f'{pformat(msg)}\n'
+                f'{structfmt(msg)}\n'
             )
 
             # NOTE: if an error is deteced we should always still
@@ -2047,6 +2129,9 @@ async def open_context_from_portal(
         # place..
         allow_overruns=allow_overruns,
     )
+    # ASAP, so that `Context.side: str` can be determined for
+    # logging / tracing / debug!
+    ctx._portal: Portal = portal
 
     assert ctx._remote_func_type == 'context'
     msg: Started = await ctx._recv_chan.receive()
@@ -2065,10 +2150,9 @@ async def open_context_from_portal(
             msg=msg,
             src_err=src_error,
             log=log,
-            expect_key='started',
+            expect_msg=Started,
         )
 
-    ctx._portal: Portal = portal
     uid: tuple = portal.channel.uid
     cid: str = ctx.cid
 

tractor/_entry.py

@@ -106,6 +106,7 @@ def _trio_main(
     Entry point for a `trio_run_in_process` subactor.
 
     '''
+    __tracebackhide__: bool = True
     _state._current_actor = actor
     trio_main = partial(
         async_main,

tractor/_exceptions.py

@@ -43,9 +43,12 @@ from tractor.msg import (
     MsgType,
     Stop,
     Yield,
-    pretty_struct,
     types as msgtypes,
 )
+from tractor.msg.pretty_struct import (
+    iter_fields,
+    Struct,
+)
 
 if TYPE_CHECKING:
     from ._context import Context
@@ -82,7 +85,7 @@ class InternalError(RuntimeError):
 _ipcmsg_keys: list[str] = [
     fi.name
     for fi, k, v
-    in pretty_struct.iter_fields(Error)
+    in iter_fields(Error)
 
 ]
 
@@ -321,7 +324,7 @@ class RemoteActorError(Exception):
             assert self.boxed_type is boxed_type
 
     @property
-    def ipc_msg(self) -> pretty_struct.Struct:
+    def ipc_msg(self) -> Struct:
         '''
         Re-render the underlying `._ipc_msg: Msg` as
         a `pretty_struct.Struct` for introspection such that the
@@ -334,12 +337,12 @@ class RemoteActorError(Exception):
         msg_type: MsgType = type(self._ipc_msg)
         fields: dict[str, Any] = {
             k: v for _, k, v in
-            pretty_struct.iter_fields(self._ipc_msg)
+            iter_fields(self._ipc_msg)
         }
         return defstruct(
             msg_type.__name__,
             fields=fields.keys(),
-            bases=(msg_type, pretty_struct.Struct),
+            bases=(msg_type, Struct),
         )(**fields)
 
     @property
@@ -641,11 +644,11 @@ class MsgTypeError(
 
     '''
     reprol_fields: list[str] = [
-        'payload_msg',
+        'expected_msg_type',
     ]
     extra_body_fields: list[str] = [
         'cid',
-        'payload_msg',
+        'expected_msg',
     ]
 
     @property
@@ -661,9 +664,7 @@ class MsgTypeError(
         return self.msgdata.get('_msg_dict')
 
     @property
-    def payload_msg(
+    def expected_msg(self) -> MsgType|None:
-        self,
-    ) -> MsgType|None:
         '''
         Attempt to construct what would have been the original
         `MsgType`-with-payload subtype (i.e. an instance from the set
@@ -674,9 +675,17 @@ class MsgTypeError(
         if msg_dict := self.msg_dict.copy():
             return msgtypes.from_dict_msg(
                 dict_msg=msg_dict,
+                # use_pretty=True,
+                # ^-TODO-^ would luv to use this BUT then the
+                # `field_prefix` in `pformat_boxed_tb()` cucks it
+                # all up.. XD
             )
         return None
 
+    @property
+    def expected_msg_type(self) -> Type[MsgType]|None:
+        return type(self.expected_msg)
+
     @property
     def cid(self) -> str:
         # pre-packed using `.from_decode()` constructor
@@ -929,7 +938,6 @@ def _raise_from_no_key_in_msg(
     src_err: KeyError,
     log: StackLevelAdapter,  # caller specific `log` obj
 
-    expect_key: str = 'yield',
     expect_msg: str = Yield,
     stream: MsgStream | None = None,
 
@@ -1044,7 +1052,7 @@ def _raise_from_no_key_in_msg(
     # is activated above.
     _type: str = 'Stream' if stream else 'Context'
     raise MessagingError(
-        f"{_type} was expecting a '{expect_key.upper()}' message"
+        f"{_type} was expecting a {expect_msg} message"
         " BUT received a non-error msg:\n"
         f'{pformat(msg)}'
     ) from src_err

tractor/_ipc.py

@@ -130,6 +130,8 @@ def _mk_msg_type_err(
 
 ) -> MsgTypeError:
 
+    import textwrap
+
     # `Channel.send()` case
     if src_validation_error is None:  # send-side
 
@@ -209,10 +211,24 @@ def _mk_msg_type_err(
         msg, _, maybe_field = msgspec_msg.rpartition('$.')
         obj = object()
         if (field_val := msg_dict.get(maybe_field, obj)) is not obj:
+            field_name_expr: str = (
+                f' |_{maybe_field}: {codec.pld_spec_str} = '
+            )
+            fmt_val_lines: list[str] = pformat(field_val).splitlines()
+            fmt_val: str = (
+                f'{fmt_val_lines[0]}\n'
+                +
+                textwrap.indent(
+                    '\n'.join(fmt_val_lines[1:]),
+                    prefix=' '*len(field_name_expr),
+                )
+            )
             message += (
                 f'{msg.rstrip("`")}\n\n'
-                f'{msg_type}\n'
+                f'<{msg_type.__qualname__}(\n'
-                f' |_.{maybe_field}: {codec.pld_spec_str} = {field_val!r}\n'
+                # f'{".".join([msg_type.__module__, msg_type.__qualname__])}\n'
+                f'{field_name_expr}{fmt_val}\n'
+                f')>'
             )
 
         msgtyperr = MsgTypeError.from_decode(
@@ -338,7 +354,7 @@ class MsgpackTCPStream(MsgTransport):
                     # self._task = task
                     self._codec = codec
                     log.runtime(
-                        'Using new codec in {self}.recv()\n'
+                        f'Using new codec in {self}.recv()\n'
                         f'codec: {self._codec}\n\n'
                         f'msg_bytes: {msg_bytes}\n'
                     )
@@ -420,7 +436,7 @@ class MsgpackTCPStream(MsgTransport):
             if self._codec.pld_spec != codec.pld_spec:
                 self._codec = codec
                 log.runtime(
-                    'Using new codec in {self}.send()\n'
+                    f'Using new codec in {self}.send()\n'
                     f'codec: {self._codec}\n\n'
                     f'msg: {msg}\n'
                 )

tractor/_root.py

@@ -79,6 +79,7 @@ async def open_root_actor(
 
     # enables the multi-process debugger support
     debug_mode: bool = False,
+    maybe_enable_greenback: bool = False,  # `.pause_from_sync()/breakpoint()` support
 
     # internal logging
     loglevel: str|None = None,
@@ -107,14 +108,16 @@ async def open_root_actor(
     )
     if (
         debug_mode
-        and
+        and maybe_enable_greenback
-        await _debug.maybe_init_greenback(
+        and await _debug.maybe_init_greenback(
             raise_not_found=False,
         )
     ):
         os.environ['PYTHONBREAKPOINT'] = (
             'tractor.devx._debug.pause_from_sync'
         )
+        _state._runtime_vars['use_greenback'] = True
+
     else:
         # TODO: disable `breakpoint()` by default (without
         # `greenback`) since it will break any multi-actor
@@ -385,14 +388,20 @@ async def open_root_actor(
         _state._last_actor_terminated = actor
 
         # restore built-in `breakpoint()` hook state
-        if debug_mode:
+        if (
+            debug_mode
+            and
+            maybe_enable_greenback
+        ):
             if builtin_bp_handler is not None:
                 sys.breakpointhook = builtin_bp_handler
+
             if orig_bp_path is not None:
                 os.environ['PYTHONBREAKPOINT'] = orig_bp_path
+
             else:
                 # clear env back to having no entry
-                os.environ.pop('PYTHONBREAKPOINT')
+                os.environ.pop('PYTHONBREAKPOINT', None)
 
         logger.runtime("Root actor terminated")
 

tractor/_rpc.py

@@ -41,7 +41,6 @@ from trio import (
     TaskStatus,
 )
 
-from .msg import NamespacePath
 from ._ipc import Channel
 from ._context import (
     Context,
@@ -61,6 +60,11 @@ from .devx import (
 )
 from . import _state
 from .log import get_logger
+from .msg import (
+    current_codec,
+    MsgCodec,
+    NamespacePath,
+)
 from tractor.msg.types import (
     CancelAck,
     Error,
@@ -98,6 +102,7 @@ async def _invoke_non_context(
         Context | BaseException
     ] = trio.TASK_STATUS_IGNORED,
 ):
+    __tracebackhide__: bool = True
 
     # TODO: can we unify this with the `context=True` impl below?
     if inspect.isasyncgen(coro):
@@ -398,7 +403,11 @@ async def _invoke(
     __tracebackhide__: bool = hide_tb
     treat_as_gen: bool = False
 
-    if _state.debug_mode():
+    if (
+        _state.debug_mode()
+        and
+        _state._runtime_vars['use_greenback']
+    ):
         # XXX for .pause_from_sync()` usage we need to make sure
         # `greenback` is boostrapped in the subactor!
         await _debug.maybe_init_greenback()
@@ -512,10 +521,22 @@ async def _invoke(
         #     wrapper that calls `Context.started()` and then does
         #     the `await coro()`?
 
-        # a "context" endpoint type is the most general and
+        # ------ - ------
-        # "least sugary" type of RPC ep with support for
+        # a "context" endpoint is the most general and
+        # "least sugary" type of RPC with support for
         # bi-dir streaming B)
-        # StartAck
+        #
+        # the concurrency relation is simlar to a task nursery
+        # wherein a "parent" task (the one that enters
+        # `trio.open_nursery()` in some actor "opens" (via
+        # `Portal.open_context()`) an IPC ctx to another peer
+        # (which is maybe a sub-) actor who then schedules (aka
+        # `trio.Nursery.start()`s) a new "child" task to execute
+        # the `@context` annotated func; that is this func we're
+        # running directly below!
+        # ------ - ------
+        #
+        # StartAck: respond immediately with endpoint info
         await chan.send(
             StartAck(
                 cid=cid,
@@ -524,11 +545,11 @@ async def _invoke(
         )
 
         # TODO: should we also use an `.open_context()` equiv
-        # for this callee side by factoring the impl from
+        # for this child side by factoring the impl from
         # `Portal.open_context()` into a common helper?
         #
         # NOTE: there are many different ctx state details
-        # in a callee side instance according to current impl:
+        # in a child side instance according to current impl:
         # - `.cancelled_caught` can never be `True`.
         #  -> the below scope is never exposed to the
         #     `@context` marked RPC function.
@@ -554,7 +575,7 @@ async def _invoke(
 
             # NOTE: this happens IFF `ctx._scope.cancel()` is
             # called by any of,
-            # - *this* callee task manually calling `ctx.cancel()`.
+            # - *this* child task manually calling `ctx.cancel()`.
             # - the runtime calling `ctx._deliver_msg()` which
             #   itself calls `ctx._maybe_cancel_and_set_remote_error()`
             #   which cancels the scope presuming the input error
@@ -631,10 +652,11 @@ async def _invoke(
                         # f'  |_{ctx}'
                     )
 
-                    # task-contex was either cancelled by request using
+                    # task-contex was either cancelled by request
-                    # ``Portal.cancel_actor()`` or ``Context.cancel()``
+                    # using ``Portal.cancel_actor()`` or
-                    # on the far end, or it was cancelled by the local
+                    # ``Context.cancel()`` on the far end, or it
-                    # (callee) task, so relay this cancel signal to the
+                    # was cancelled by the local child (or callee)
+                    # task, so relay this cancel signal to the
                     # other side.
                     ctxc = ContextCancelled(
                         message=msg,
@@ -655,7 +677,7 @@ async def _invoke(
 
         ) as scope_error:
 
-            # always set this (callee) side's exception as the
+            # always set this (child) side's exception as the
             # local error on the context
             ctx._local_error: BaseException = scope_error
 
@@ -1024,9 +1046,8 @@ async def process_messages(
                                 trio.Event(),
                             )
 
-                    # XXX remote (runtime scoped) error or uknown
+                    # runtime-scoped remote error (since no `.cid`)
-                    # msg (type).
+                    case Error():
-                    case Error() | _:
                         # NOTE: this is the non-rpc error case,
                         # that is, an error **not** raised inside
                         # a call to ``_invoke()`` (i.e. no cid was
@@ -1034,10 +1055,6 @@ async def process_messages(
                         # this error to all local channel
                         # consumers (normally portals) by marking
                         # the channel as errored
-                        log.exception(
-                            f'Unhandled IPC msg:\n\n'
-                            f'{msg}\n'
-                        )
                         # assert chan.uid
                         chan._exc: Exception = unpack_error(
                             msg,
@@ -1045,6 +1062,17 @@ async def process_messages(
                         )
                         raise chan._exc
 
+                    # unknown/invalid msg type?
+                    case _:
+                        codec: MsgCodec = current_codec()
+                        message: str = (
+                            f'Unhandled IPC msg for codec?\n\n'
+                            f'|_{codec}\n\n'
+                            f'{msg}\n'
+                        )
+                        log.exception(message)
+                        raise RuntimeError(message)
+
                 log.runtime(
                     'Waiting on next IPC msg from\n'
                     f'peer: {chan.uid}\n'

tractor/_spawn.py

@@ -513,7 +513,7 @@ async def trio_proc(
         # })
 
         # track subactor in current nursery
-        curr_actor = current_actor()
+        curr_actor: Actor = current_actor()
         curr_actor._actoruid2nursery[subactor.uid] = actor_nursery
 
         # resume caller at next checkpoint now that child is up

tractor/_streaming.py

@@ -44,6 +44,7 @@ from .trionics import (
     BroadcastReceiver,
 )
 from tractor.msg import (
+    Return,
     Stop,
     Yield,
 )
@@ -96,36 +97,26 @@ class MsgStream(trio.abc.Channel):
     # delegate directly to underlying mem channel
     def receive_nowait(
         self,
-        allow_msg_keys: list[str] = ['yield'],
+        allow_msgs: list[str] = Yield,
     ):
-        # msg: dict = self._rx_chan.receive_nowait()
         msg: Yield|Stop = self._rx_chan.receive_nowait()
-        for (
+        # TODO: replace msg equiv of this or does the `.pld`
-            i,
+        # interface read already satisfy it? I think so, yes?
-            key,
-        ) in enumerate(allow_msg_keys):
         try:
-                # return msg[key]
             return msg.pld
-            # except KeyError as kerr:
         except AttributeError as attrerr:
-                if i < (len(allow_msg_keys) - 1):
-                    continue
-
             _raise_from_no_key_in_msg(
                 ctx=self._ctx,
                 msg=msg,
-                    # src_err=kerr,
                 src_err=attrerr,
                 log=log,
-                    expect_key=key,
                 stream=self,
             )
 
     async def receive(
         self,
 
-        hide_tb: bool = True,
+        hide_tb: bool = False,
     ):
         '''
         Receive a single msg from the IPC transport, the next in
@@ -157,10 +148,9 @@ class MsgStream(trio.abc.Channel):
         try:
             try:
                 msg: Yield = await self._rx_chan.receive()
-                # return msg['yield']
                 return msg.pld
 
-            # except KeyError as kerr:
+            # TODO: implement with match: instead?
             except AttributeError as attrerr:
                 # src_err = kerr
                 src_err = attrerr
@@ -170,10 +160,8 @@ class MsgStream(trio.abc.Channel):
                 _raise_from_no_key_in_msg(
                     ctx=self._ctx,
                     msg=msg,
-                    # src_err=kerr,
                     src_err=attrerr,
                     log=log,
-                    expect_key='yield',
                     stream=self,
                 )
 
@@ -304,7 +292,7 @@ class MsgStream(trio.abc.Channel):
         while not drained:
             try:
                 maybe_final_msg = self.receive_nowait(
-                    allow_msg_keys=['yield', 'return'],
+                    allow_msgs=[Yield, Return],
                 )
                 if maybe_final_msg:
                     log.debug(

tractor/msg/_codec.py

@@ -420,7 +420,7 @@ def mk_codec(
 
 # instance of the default `msgspec.msgpack` codec settings, i.e.
 # no custom structs, hooks or other special types.
-_def_msgspec_codec: MsgCodec = mk_codec(ipc_msg_spec=Any)
+_def_msgspec_codec: MsgCodec = mk_codec(ipc_pld_spec=Any)
 
 # The built-in IPC `Msg` spec.
 # Our composing "shuttle" protocol which allows `tractor`-app code

tractor/msg/types.py

@@ -451,7 +451,8 @@ def from_dict_msg(
     dict_msg: dict,
 
     msgT: MsgType|None = None,
-    tag_field: str = 'msg_type'
+    tag_field: str = 'msg_type',
+    use_pretty: bool = False,
 
 ) -> MsgType:
     '''
@@ -468,6 +469,19 @@ def from_dict_msg(
     # XXX ensure tag field is removed
     msgT_name: str = dict_msg.pop(msg_type_tag_field)
     msgT: MsgType = _msg_table[msgT_name]
+    if use_pretty:
+        msgT = defstruct(
+            name=msgT_name,
+            fields=[
+                (key, fi.type)
+                for fi, key, _
+                in pretty_struct.iter_fields(msgT)
+            ],
+            bases=(
+                pretty_struct.Struct,
+                msgT,
+            ),
+        )
     return msgT(**dict_msg)
 
 # TODO: should be make a msg version of `ContextCancelled?`