#4057 [BUG] Custom type transformer in nested dynamic workflows fails Issue created by geekysuavo ### Describe the bug I have a workflow which calls a dynamic task, which itself calls a second dynamic task, that is inexplicably failing when executed remotely: @dynamic def outer(myobj: MyCustomType, key: str) -> float: results: dict[str, float] = myobj.call_some_flytekit_tasks() return results[key] @dynamic def inner(myobj: MyCustomType) -> float: return outer(myobj=myobj, key='foo') @workflow def wf(myobj: MyCustomType) -> float: return inner(myobj=myobj) where

MyCustomType

has two methods: • An instance method

pack : MyCustomType -> dict[str, FlyteFile]

• A class method

unpack : dict[str, FlyteFile] -> MyCustomType

and the following type transformer has been registered: class MyTypeTransformer(TypeTransformer[MyCustomType]): def __init__(self): super().__init__("MyCustomType", t=MyCustomType) self._base_type = dict[str, FlyteFile] self._base_impl = DictTransformer() def get_literal_type(self, t: Type[MyCustomType]) -> LiteralType: return self._base_impl.get_literal_type(t=self._base_type) def to_literal( self, ctx: FlyteContext, python_val: MyCustomType, python_type: Type[MyCustomType], expected: LiteralType, ) -> Literal: flyte_files = python_val.pack() return self._base_impl.to_literal( ctx, flyte_files, self._base_type, expected ) def to_python_value( self, ctx: FlyteContext, lv: Literal, expected_python_type: Type[MyCustomType], ) -> MyCustomType: flyte_files = self._base_impl.to_python_value(ctx, lv, self._base_type) return MyCustomType.unpack(flyte_files) When executing this workflow remotely, the

inner

dynamic task fails, but the execution waits in a running state for a while until it eventually fails with the following error from flytepropeller:

Workflow[flytetester:development:<http://workflows.wf|workflows.wf>] failed. RuntimeExecutionError: max number of system retry attempts [51/50] exhausted. Last known status message: Workflow[flytetester:development:<http://workflows.wf|workflows.wf>] failed. CausedByError: Failed to propagate Abort for workflow. Error: 0: 0: [User] malformed dynamic workflow, caused by: Collected Errors: 2
	Error 0: Code: MismatchingBindings, Node Id: dn0, Description: Input [myobj] on node [dn0] expects bindings of type [map_value_type:<blob:<> > ].  Received []
...

However, replacing

MyCustomType

with

dict[str, FlyteFile]

in

outer

and manually packing that argument when calling it from

inner

results in successful execution. ### Expected behavior The workflow executes without failure when

MyCustomType

is used in both dynamic tasks. ### Additional context to reproduce No response ### Screenshots No response ### Are you sure this issue hasn't been raised already? • Yes ### Have you read the Code of Conduct? • Yes flyteorg/flyte