Go back to `ContextVar` for codec mgmt

Turns out we do want per-task inheritance particularly if there's to be
per `Context` dynamic mutation of the spec; we don't want mutation in
some task to affect any parent/global setting.

Turns out since we use a common "feeder task" in the rpc loop, we need to
offer a per `Context` payload decoder sys anyway in order to enable
per-task controls for inter-actor multi-task-ctx scenarios.

tractor/_exceptions.py

@@ -24,7 +24,6 @@ import importlib
 from pprint import pformat
 from typing import (
     Any,
-    Callable,
     Type,
     TYPE_CHECKING,
 )
@@ -33,11 +32,8 @@ import traceback
 
 import trio
 from msgspec import (
-    defstruct,
-    msgpack,
-    Raw,
     structs,
-    ValidationError,
+    defstruct,
 )
 
 from tractor._state import current_actor
@@ -48,8 +44,6 @@ from tractor.msg import (
     Stop,
     Yield,
     types as msgtypes,
-    MsgCodec,
-    MsgDec,
 )
 from tractor.msg.pretty_struct import (
     iter_fields,
@@ -176,7 +170,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 = (
@@ -938,7 +932,7 @@ def is_multi_cancelled(exc: BaseException) -> bool:
     return False
 
 
-def _raise_from_unexpected_msg(
+def _raise_from_no_key_in_msg(
     ctx: Context,
     msg: MsgType,
     src_err: AttributeError,
@@ -978,6 +972,8 @@ def _raise_from_unexpected_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'
@@ -989,6 +985,7 @@ def _raise_from_unexpected_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():
@@ -1004,6 +1001,7 @@ def _raise_from_unexpected_msg(
     # the stream._eoc outside this in the calleer always?
         # case Stop():
     elif (
+        # msg.get('stop')
         isinstance(msg, Stop)
         or (
             stream
@@ -1038,6 +1036,7 @@ def _raise_from_unexpected_msg(
         # that arrived which is probably the source of this stream
         # closure
         ctx.maybe_raise()
+
         raise eoc from src_err
 
     if (
@@ -1057,128 +1056,3 @@ def _raise_from_unexpected_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,7 +49,6 @@ from tractor._exceptions import (
     MsgTypeError,
     pack_from_raise,
     TransportClosed,
-    _mk_msg_type_err,
 )
 from tractor.msg import (
     _ctxvar_MsgCodec,
@@ -119,6 +118,127 @@ 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?
@@ -179,10 +299,10 @@ class MsgpackTCPStream(MsgTransport):
             _codec._ctxvar_MsgCodec.get()
         )
         # TODO: mask out before release?
-        # log.runtime(
-        #     f'New {self} created with codec\n'
-        #     f'codec: {self._codec}\n'
-        # )
+        log.runtime(
+            f'New {self} created with codec\n'
+            f'codec: {self._codec}\n'
+        )
 
     async def _iter_packets(self) -> AsyncGenerator[dict, None]:
         '''

tractor/msg/__init__.py

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

tractor/msg/_codec.py

@@ -71,108 +71,6 @@ 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!
@@ -196,7 +94,8 @@ 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(
@@ -219,21 +118,14 @@ class MsgCodec(Struct):
             ')>'
         )
 
-    @property
-    def pld_spec(self) -> Type[Struct]|Raw|Any:
-        return self._pld_spec
-
     @property
     def pld_spec_str(self) -> str:
-
-        # TODO: could also use match: instead?
         spec: Union[Type]|Type = self.pld_spec
 
-        # `typing.Union` case
+        # TODO: could also use match: instead?
         if getattr(spec, '__args__', False):
+            # `typing.Union` case
             return str(spec)
-
-        # just a single type
         else:
             return spec.__name__
 
@@ -241,7 +133,6 @@ 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(
@@ -266,9 +157,8 @@ class MsgCodec(Struct):
     def pformat_msg_spec(
         self,
         msg: MsgType|None = None,
-        join_char: str = '\n',
     ) -> str:
-        return join_char.join(
+        return '\n'.join(
             self.msg_spec_items(msg=msg).values()
         )
 
@@ -515,25 +405,18 @@ 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
@@ -552,9 +435,10 @@ _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=Any,
-    ipc_pld_spec=Raw,
+    # ipc_pld_spec=Raw,
 )
 # TODO: IDEALLY provides for per-`trio.Task` specificity of the
 # IPC msging codec used by the transport layer when doing
@@ -654,12 +538,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:
     '''
@@ -690,7 +574,7 @@ def limit_msg_spec(
 
     '''
     __tracebackhide__: bool = True
-    curr_codec: MsgCodec = current_codec()
+    curr_codec = current_codec()
     msgspec_codec: MsgCodec = mk_codec(
         ipc_pld_spec=payload_spec,
         **codec_kwargs,

tractor/msg/types.py

@@ -38,7 +38,6 @@ from typing import (
 from msgspec import (
     defstruct,
     # field,
-    Raw,
     Struct,
     # UNSET,
     # UnsetType,
@@ -106,7 +105,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|Raw
+    pld: PayloadT
 
 
 class Aid(
@@ -266,7 +265,35 @@ 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)
 
 
@@ -305,7 +332,7 @@ class Started(
     decorated IPC endpoint.
 
     '''
-    pld: PayloadT|Raw
+    pld: PayloadT
 
 
 # TODO: instead of using our existing `Start`
@@ -322,7 +349,7 @@ class Yield(
     Per IPC transmission of a value from `await MsgStream.send(<value>)`.
 
     '''
-    pld: PayloadT|Raw
+    pld: PayloadT
 
 
 class Stop(
@@ -350,12 +377,11 @@ class Return(
     func-as-`trio.Task`.
 
     '''
-    pld: PayloadT|Raw
+    pld: PayloadT
 
 
 class CancelAck(
-    Msg,
-    Generic[PayloadT],
+    Return,
 ):
     '''
     Deliver the `bool` return-value from a cancellation `Actor`
@@ -684,9 +710,7 @@ def mk_msg_spec(
     )
     return (
         ipc_spec,
-        msgtypes_table[spec_build_method]
-        +
-        ipc_msg_types,
+        msgtypes_table[spec_build_method] + ipc_msg_types,
     )