Add a `MsgDec` for receive-only decoding

In prep for a "payload receiver" abstraction that will wrap
`MsgType.pld`-IO delivery from `Context` and `MsgStream`, adds a small
`msgspec.msgpack.Decoder` shim which delegates an API similar to
`MsgCodec` and is offered via a `.msg._codec.mk_dec()` factory.

Detalles:
- move over the TODOs/comments from `.msg.types.Start` to to
  `MsgDec.spec` since it's probably the ideal spot to start thinking
  about it from a consumer code PoV.
- move codec reversion assert and log emit into `finally:` block.
- flip default `.types._tractor_codec = mk_codec_ipc_pld(ipc_pld_spec=Raw)`
  in prep for always doing payload-delayed decodes.
- make `MsgCodec._dec` private with public property getter.
- change `CancelAck` to NOT derive from `Return` so it's mutex in
  `match/case:` handling.

tractor/_exceptions.py

@@ -24,6 +24,7 @@ import importlib
 from pprint import pformat
 from typing import (
     Any,
+    Callable,
     Type,
     TYPE_CHECKING,
 )
@@ -32,8 +33,11 @@ import traceback
 
 import trio
 from msgspec import (
-    structs,
     defstruct,
+    msgpack,
+    Raw,
+    structs,
+    ValidationError,
 )
 
 from tractor._state import current_actor
@@ -44,6 +48,8 @@ from tractor.msg import (
     Stop,
     Yield,
     types as msgtypes,
+    MsgCodec,
+    MsgDec,
 )
 from tractor.msg.pretty_struct import (
     iter_fields,
@@ -170,7 +176,7 @@ def pformat_boxed_tb(
         f' ------ - ------\n\n'
         # f'{tb_str}\n'
         f'{tb_body}'
-        f'  ------ - ------\n'
+        f' ------ - ------\n'
         f'_|\n'
     )
     tb_box_indent: str = (
@@ -932,7 +938,7 @@ def is_multi_cancelled(exc: BaseException) -> bool:
     return False
 
 
-def _raise_from_no_key_in_msg(
+def _raise_from_unexpected_msg(
     ctx: Context,
     msg: MsgType,
     src_err: AttributeError,
@@ -972,8 +978,6 @@ def _raise_from_no_key_in_msg(
     # an internal error should never get here
     try:
         cid: str = msg.cid
-        # cid: str = msg['cid']
-    # except KeyError as src_err:
     except AttributeError as src_err:
         raise MessagingError(
             f'IPC `Context` rx-ed msg without a ctx-id (cid)!?\n'
@@ -985,7 +989,6 @@ def _raise_from_no_key_in_msg(
     # TODO: test that shows stream raising an expected error!!!
 
     # raise the error message in a boxed exception type!
-    # if msg.get('error'):
     if isinstance(msg, Error):
     # match msg:
     #     case Error():
@@ -1001,7 +1004,6 @@ def _raise_from_no_key_in_msg(
     # the stream._eoc outside this in the calleer always?
         # case Stop():
     elif (
-        # msg.get('stop')
         isinstance(msg, Stop)
         or (
             stream
@@ -1036,7 +1038,6 @@ def _raise_from_no_key_in_msg(
         # that arrived which is probably the source of this stream
         # closure
         ctx.maybe_raise()
-
         raise eoc from src_err
 
     if (
@@ -1056,3 +1057,128 @@ def _raise_from_no_key_in_msg(
         " BUT received a non-error msg:\n"
         f'{pformat(msg)}'
     ) from src_err
+
+
+_raise_from_no_key_in_msg = _raise_from_unexpected_msg
+
+
+def _mk_msg_type_err(
+    msg: Any|bytes|Raw,
+    codec: MsgCodec|MsgDec,
+
+    message: str|None = None,
+    verb_header: str = '',
+
+    src_validation_error: ValidationError|None = None,
+    src_type_error: TypeError|None = None,
+
+) -> MsgTypeError:
+    '''
+    Compose a `MsgTypeError` from an input runtime context.
+
+    '''
+    # `Channel.send()` case
+    if src_validation_error is None:
+
+        if isinstance(codec, MsgDec):
+            raise RuntimeError(
+                '`codec` must be a `MsgCodec` for send-side errors?'
+            )
+
+        # no src error from `msgspec.msgpack.Decoder.decode()` so
+        # prolly a manual type-check on our part.
+        if message is None:
+            fmt_spec: str = codec.pformat_msg_spec()
+            fmt_stack: str = (
+                '\n'.join(traceback.format_stack(limit=3))
+            )
+            tb_fmt: str = pformat_boxed_tb(
+                tb_str=fmt_stack,
+                # fields_str=header,
+                field_prefix='  ',
+                indent='',
+            )
+            message: str = (
+                f'invalid msg -> {msg}: {type(msg)}\n\n'
+                f'{tb_fmt}\n'
+                f'Valid IPC msgs are:\n\n'
+                # f'  ------ - ------\n'
+                f'{fmt_spec}\n',
+            )
+        elif src_type_error:
+            src_message: str = str(src_type_error)
+            patt: str = 'type '
+            type_idx: int = src_message.find('type ')
+            invalid_type: str = src_message[type_idx + len(patt):].split()[0]
+
+            enc_hook: Callable|None = codec.enc.enc_hook
+            if enc_hook is None:
+                message += (
+                    '\n\n'
+
+                    f"The current IPC-msg codec can't encode type `{invalid_type}` !\n"
+                    f'Maybe a `msgpack.Encoder.enc_hook()` extension is needed?\n\n'
+
+                    f'Check the `msgspec` docs for ad-hoc type extending:\n'
+                    '|_ https://jcristharif.com/msgspec/extending.html\n'
+                    '|_ https://jcristharif.com/msgspec/extending.html#defining-a-custom-extension-messagepack-only\n'
+                )
+
+
+        msgtyperr = MsgTypeError(
+            message=message,
+            ipc_msg=msg,
+        )
+        # ya, might be `None`
+        msgtyperr.__cause__ = src_type_error
+        return msgtyperr
+
+    # `Channel.recv()` case
+    else:
+        # decode the msg-bytes using the std msgpack
+        # interchange-prot (i.e. without any
+        # `msgspec.Struct` handling) so that we can
+        # determine what `.msg.types.Msg` is the culprit
+        # by reporting the received value.
+        msg_dict: dict = msgpack.decode(msg)
+        msg_type_name: str = msg_dict['msg_type']
+        msg_type = getattr(msgtypes, msg_type_name)
+        message: str = (
+            f'invalid `{msg_type_name}` IPC msg\n\n'
+        )
+        if verb_header:
+            message = f'{verb_header} ' + message
+
+        # XXX see if we can determine the exact invalid field
+        # such that we can comprehensively report the
+        # specific field's type problem
+        msgspec_msg: str = src_validation_error.args[0].rstrip('`')
+        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.__qualname__}(\n'
+                # f'{".".join([msg_type.__module__, msg_type.__qualname__])}\n'
+                f'{field_name_expr}{fmt_val}\n'
+                f')>'
+            )
+
+        msgtyperr = MsgTypeError.from_decode(
+            message=message,
+            msgdict=msg_dict,
+        )
+        msgtyperr.__cause__ = src_validation_error
+        return msgtyperr

tractor/_ipc.py

@@ -49,6 +49,7 @@ from tractor._exceptions import (
     MsgTypeError,
     pack_from_raise,
     TransportClosed,
+    _mk_msg_type_err,
 )
 from tractor.msg import (
     _ctxvar_MsgCodec,
@@ -118,127 +119,6 @@ class MsgTransport(Protocol[MsgType]):
         ...
 
 
-def _mk_msg_type_err(
-    msg: Any|bytes,
-    codec: MsgCodec,
-
-    message: str|None = None,
-    verb_header: str = '',
-
-    src_validation_error: msgspec.ValidationError|None = None,
-    src_type_error: TypeError|None = None,
-
-) -> MsgTypeError:
-
-    import textwrap
-
-    # `Channel.send()` case
-    if src_validation_error is None:  # send-side
-
-        # no src error from `msgspec.msgpack.Decoder.decode()` so
-        # prolly a manual type-check on our part.
-        if message is None:
-            import traceback
-            from tractor._exceptions import pformat_boxed_tb
-
-            fmt_spec: str = '\n'.join(
-                map(str, codec.msg_spec.__args__)
-            )
-            fmt_stack: str = (
-                '\n'.join(traceback.format_stack(limit=3))
-            )
-            tb_fmt: str = pformat_boxed_tb(
-                tb_str=fmt_stack,
-                # fields_str=header,
-                field_prefix='  ',
-                indent='',
-            )
-            message: str = (
-                f'invalid msg -> {msg}: {type(msg)}\n\n'
-                f'{tb_fmt}\n'
-                f'Valid IPC msgs are:\n\n'
-                # f'  ------ - ------\n'
-                f'{fmt_spec}\n',
-            )
-        elif src_type_error:
-            src_message: str = str(src_type_error)
-            patt: str = 'type '
-            type_idx: int = src_message.find('type ')
-            invalid_type: str = src_message[type_idx + len(patt):].split()[0]
-
-            enc_hook: Callable|None = codec.enc.enc_hook
-            if enc_hook is None:
-                message += (
-                    '\n\n'
-
-                    f"The current IPC-msg codec can't encode type `{invalid_type}` !\n"
-                    f'Maybe a `msgpack.Encoder.enc_hook()` extension is needed?\n\n'
-
-                    f'Check the `msgspec` docs for ad-hoc type extending:\n'
-                    '|_ https://jcristharif.com/msgspec/extending.html\n'
-                    '|_ https://jcristharif.com/msgspec/extending.html#defining-a-custom-extension-messagepack-only\n'
-                )
-
-
-        msgtyperr = MsgTypeError(
-            message=message,
-            ipc_msg=msg,
-        )
-        # ya, might be `None`
-        msgtyperr.__cause__ = src_type_error
-        return msgtyperr
-
-    # `Channel.recv()` case
-    else:
-        # decode the msg-bytes using the std msgpack
-        # interchange-prot (i.e. without any
-        # `msgspec.Struct` handling) so that we can
-        # determine what `.msg.types.Msg` is the culprit
-        # by reporting the received value.
-        msg_dict: dict = msgspec.msgpack.decode(msg)
-        msg_type_name: str = msg_dict['msg_type']
-        msg_type = getattr(msgtypes, msg_type_name)
-        message: str = (
-            f'invalid `{msg_type_name}` IPC msg\n\n'
-        )
-        if verb_header:
-            message = f'{verb_header} ' + message
-
-        # XXX see if we can determine the exact invalid field
-        # such that we can comprehensively report the
-        # specific field's type problem
-        msgspec_msg: str = src_validation_error.args[0].rstrip('`')
-        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.__qualname__}(\n'
-                # f'{".".join([msg_type.__module__, msg_type.__qualname__])}\n'
-                f'{field_name_expr}{fmt_val}\n'
-                f')>'
-            )
-
-        msgtyperr = MsgTypeError.from_decode(
-            message=message,
-            msgdict=msg_dict,
-        )
-        msgtyperr.__cause__ = src_validation_error
-        return msgtyperr
-
-
 # TODO: not sure why we have to inherit here, but it seems to be an
 # issue with ``get_msg_transport()`` returning a ``Type[Protocol]``;
 # probably should make a `mypy` issue?
@@ -299,10 +179,10 @@ class MsgpackTCPStream(MsgTransport):
             _codec._ctxvar_MsgCodec.get()
         )
         # TODO: mask out before release?
-        log.runtime(
+        # log.runtime(
-            f'New {self} created with codec\n'
+        #     f'New {self} created with codec\n'
-            f'codec: {self._codec}\n'
+        #     f'codec: {self._codec}\n'
-        )
+        # )
 
     async def _iter_packets(self) -> AsyncGenerator[dict, None]:
         '''

tractor/msg/__init__.py

@@ -34,6 +34,7 @@ from ._codec import (
     apply_codec as apply_codec,
     mk_codec as mk_codec,
     MsgCodec as MsgCodec,
+    MsgDec as MsgDec,
     current_codec as current_codec,
 )
 
@@ -50,6 +51,7 @@ from .types import (
     Yield as Yield,
     Stop as Stop,
     Return as Return,
+    CancelAck as CancelAck,
 
     Error as Error,
 

tractor/msg/_codec.py

@@ -71,6 +71,108 @@ if TYPE_CHECKING:
 
 log = get_logger(__name__)
 
+
+# TODO: unify with `MsgCodec` by making `._dec` part this?
+class MsgDec(Struct):
+    '''
+    An IPC msg decoder.
+
+    Normally used to decode only a payload: `MsgType.pld:
+    PayloadT` field before delivery to IPC consumer code.
+
+    '''
+    _dec: msgpack.Decoder
+
+    @property
+    def dec(self) -> msgpack.Decoder:
+        return self._dec
+
+    # struct type unions
+    # https://jcristharif.com/msgspec/structs.html#tagged-unions
+    #
+    # ^-TODO-^: make a wrapper type for this such that alt
+    # backends can be represented easily without a `Union` needed,
+    # AND so that we have better support for wire transport.
+    #
+    # -[ ] maybe `FieldSpec` is a good name since msg-spec
+    #   better applies to a `MsgType[FieldSpec]`?
+    #
+    # -[ ] both as part of the `.open_context()` call AND as part of the
+    #     immediate ack-reponse (see similar below)
+    #     we should do spec matching and fail if anything is awry?
+    #
+    # -[ ] eventually spec should be generated/parsed from the
+    #     type-annots as # desired in GH issue:
+    #     https://github.com/goodboy/tractor/issues/365
+    #
+    # -[ ] semantics of the mismatch case
+    #   - when caller-callee specs we should raise
+    #    a `MsgTypeError` or `MsgSpecError` or similar?
+    #
+    # -[ ] wrapper types for both spec types such that we can easily
+    #     IPC transport them?
+    #     - `TypeSpec: Union[Type]`
+    #      * also a `.__contains__()` for doing `None in
+    #      TypeSpec[None|int]` since rn you need to do it on
+    #      `.__args__` for unions..
+    #     - `MsgSpec: Union[Type[Msg]]
+    #
+    # -[ ] auto-genning this from new (in 3.12) type parameter lists Bo
+    # |_ https://docs.python.org/3/reference/compound_stmts.html#type-params
+    # |_ historical pep 695: https://peps.python.org/pep-0695/
+    # |_ full lang spec: https://typing.readthedocs.io/en/latest/spec/
+    # |_ on annotation scopes:
+    #    https://docs.python.org/3/reference/executionmodel.html#annotation-scopes
+    # |_ 3.13 will have subscriptable funcs Bo
+    #    https://peps.python.org/pep-0718/
+    @property
+    def spec(self) -> Union[Type[Struct]]:
+        # NOTE: defined and applied inside `mk_codec()`
+        return self._dec.type
+
+    # no difference, as compared to a `MsgCodec` which defines the
+    # `MsgType.pld: PayloadT` part of its spec separately
+    pld_spec = spec
+
+    # TODO: would get moved into `FieldSpec.__str__()` right?
+    @property
+    def spec_str(self) -> str:
+
+        # TODO: could also use match: instead?
+        spec: Union[Type]|Type = self.spec
+
+        # `typing.Union` case
+        if getattr(spec, '__args__', False):
+            return str(spec)
+
+        # just a single type
+        else:
+            return spec.__name__
+
+    pld_spec_str = spec_str
+
+    def decode(
+        self,
+        raw: Raw|bytes,
+    ) -> Any:
+        return self._dec.decode(raw)
+
+    @property
+    def hook(self) -> Callable|None:
+        return self._dec.dec_hook
+
+
+def mk_dec(
+    spec: Union[Type[Struct]]|Any = Any,
+    dec_hook: Callable|None = None,
+
+) -> MsgDec:
+
+    return msgpack.Decoder(
+        type=spec,  # like `Msg[Any]`
+        dec_hook=dec_hook,
+    )
+
 # TODO: overall IPC msg-spec features (i.e. in this mod)!
 #
 # -[ ] API changes towards being interchange lib agnostic!
@@ -94,8 +196,7 @@ class MsgCodec(Struct):
     '''
     _enc: msgpack.Encoder
     _dec: msgpack.Decoder
-
+    _pld_spec: Type[Struct]|Raw|Any
-    pld_spec: Union[Type[Struct]]|None
 
     def __repr__(self) -> str:
         speclines: str = textwrap.indent(
@@ -118,14 +219,21 @@ class MsgCodec(Struct):
             ')>'
         )
 
+    @property
+    def pld_spec(self) -> Type[Struct]|Raw|Any:
+        return self._pld_spec
+
     @property
     def pld_spec_str(self) -> str:
-        spec: Union[Type]|Type = self.pld_spec
 
         # TODO: could also use match: instead?
+        spec: Union[Type]|Type = self.pld_spec
+
+        # `typing.Union` case
         if getattr(spec, '__args__', False):
-            # `typing.Union` case
             return str(spec)
+
+        # just a single type
         else:
             return spec.__name__
 
@@ -133,6 +241,7 @@ class MsgCodec(Struct):
     # https://jcristharif.com/msgspec/structs.html#tagged-unions
     @property
     def msg_spec(self) -> Union[Type[Struct]]:
+        # NOTE: defined and applied inside `mk_codec()`
         return self._dec.type
 
     def msg_spec_items(
@@ -157,8 +266,9 @@ class MsgCodec(Struct):
     def pformat_msg_spec(
         self,
         msg: MsgType|None = None,
+        join_char: str = '\n',
     ) -> str:
-        return '\n'.join(
+        return join_char.join(
             self.msg_spec_items(msg=msg).values()
         )
 
@@ -405,18 +515,25 @@ def mk_codec(
     assert len(ipc_msg_spec.__args__) == len(msg_types)
     assert ipc_msg_spec
 
+    # TODO: use this shim instead?
+    # bc.. unification, err somethin?
+    # dec: MsgDec = mk_dec(
+    #     spec=ipc_msg_spec,
+    #     dec_hook=dec_hook,
+    # )
+
+    dec = msgpack.Decoder(
+        type=ipc_msg_spec,
+        dec_hook=dec_hook,
+    )
     enc = msgpack.Encoder(
        enc_hook=enc_hook,
     )
-    dec = msgpack.Decoder(
-        type=ipc_msg_spec,  # like `Msg[Any]`
-        dec_hook=dec_hook,
-    )
 
     codec = MsgCodec(
         _enc=enc,
         _dec=dec,
-        pld_spec=ipc_pld_spec,
+        _pld_spec=ipc_pld_spec,
     )
 
     # sanity on expected backend support
@@ -435,10 +552,9 @@ _def_msgspec_codec: MsgCodec = mk_codec(ipc_pld_spec=Any)
 # https://jcristharif.com/msgspec/supported-types.html
 #
 _def_tractor_codec: MsgCodec = mk_codec(
-    ipc_pld_spec=Any,
-
     # TODO: use this for debug mode locking prot?
-    # ipc_pld_spec=Raw,
+    # ipc_pld_spec=Any,
+    ipc_pld_spec=Raw,
 )
 # TODO: IDEALLY provides for per-`trio.Task` specificity of the
 # IPC msging codec used by the transport layer when doing
@@ -538,12 +654,12 @@ def apply_codec(
         yield var.get()
     finally:
         var.reset(token)
+        log.info(
+            'Reverted to last msg-spec codec\n\n'
+            f'{orig}\n'
+        )
+        assert var.get() is orig
 
-    assert var.get() is orig
-    log.info(
-        'Reverted to last msg-spec codec\n\n'
-        f'{orig}\n'
-    )
 
 def current_codec() -> MsgCodec:
     '''
@@ -574,7 +690,7 @@ def limit_msg_spec(
 
     '''
     __tracebackhide__: bool = True
-    curr_codec = current_codec()
+    curr_codec: MsgCodec = current_codec()
     msgspec_codec: MsgCodec = mk_codec(
         ipc_pld_spec=payload_spec,
         **codec_kwargs,

tractor/msg/types.py

@@ -38,6 +38,7 @@ from typing import (
 from msgspec import (
     defstruct,
     # field,
+    Raw,
     Struct,
     # UNSET,
     # UnsetType,
@@ -105,7 +106,7 @@ class Msg(
     # TODO: could also be set to `msgspec.Raw` if the sub-decoders
     # approach is preferred over the generic parameterization 
     # approach as take by `mk_msg_spec()` below.
-    pld: PayloadT
+    pld: PayloadT|Raw
 
 
 class Aid(
@@ -265,35 +266,7 @@ class Start(
 
     # TODO: enforcing a msg-spec in terms `Msg.pld`
     # parameterizable msgs to be used in the appls IPC dialog.
-    #
+    # => SEE `._codec.MsgDec` for more <=
-    # -[ ] both as part of the `.open_context()` call AND as part of the
-    #     immediate ack-reponse (see similar below)
-    #     we should do spec matching and fail if anything is awry?
-    #
-    # -[ ] eventually spec should be generated/parsed from the
-    #     type-annots as # desired in GH issue:
-    #     https://github.com/goodboy/tractor/issues/365
-    #
-    # -[ ] semantics of the mismatch case
-    #   - when caller-callee specs we should raise
-    #    a `MsgTypeError` or `MsgSpecError` or similar?
-    #
-    # -[ ] wrapper types for both spec types such that we can easily
-    #     IPC transport them?
-    #     - `TypeSpec: Union[Type]`
-    #      * also a `.__contains__()` for doing `None in
-    #      TypeSpec[None|int]` since rn you need to do it on
-    #      `.__args__` for unions..
-    #     - `MsgSpec: Union[Type[Msg]]
-    #
-    # -[ ] auto-genning this from new (in 3.12) type parameter lists Bo
-    # |_ https://docs.python.org/3/reference/compound_stmts.html#type-params
-    # |_ historical pep 695: https://peps.python.org/pep-0695/
-    # |_ full lang spec: https://typing.readthedocs.io/en/latest/spec/
-    # |_ on annotation scopes:
-    #    https://docs.python.org/3/reference/executionmodel.html#annotation-scopes
-    # |_ 3.13 will have subscriptable funcs Bo
-    #    https://peps.python.org/pep-0718/
     pld_spec: str = str(Any)
 
 
@@ -332,7 +305,7 @@ class Started(
     decorated IPC endpoint.
 
     '''
-    pld: PayloadT
+    pld: PayloadT|Raw
 
 
 # TODO: instead of using our existing `Start`
@@ -349,7 +322,7 @@ class Yield(
     Per IPC transmission of a value from `await MsgStream.send(<value>)`.
 
     '''
-    pld: PayloadT
+    pld: PayloadT|Raw
 
 
 class Stop(
@@ -377,11 +350,12 @@ class Return(
     func-as-`trio.Task`.
 
     '''
-    pld: PayloadT
+    pld: PayloadT|Raw
 
 
 class CancelAck(
-    Return,
+    Msg,
+    Generic[PayloadT],
 ):
     '''
     Deliver the `bool` return-value from a cancellation `Actor`
@@ -710,7 +684,9 @@ def mk_msg_spec(
     )
     return (
         ipc_spec,
-        msgtypes_table[spec_build_method] + ipc_msg_types,
+        msgtypes_table[spec_build_method]
+        +
+        ipc_msg_types,
     )