#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