Nested flows steps possibly returns differently if using `apply` ?

  def run(input, _opts, _context) do
    Logger.info("Creating resource #{input.resource}")

    created_resource =
      input.resource
      |> Changeset.for_create(
        :create,
        input.attributes |> Map.merge(additional_attributes(input))
      )
      |> MmsBiztalk.create()

    case created_resource do
      {:ok, resource_record} ->
        resource_record |> input.resource.to_syncing!() |> input.resource.to_synced()

      # NOTE this skips the invalid records allowing all the valid ones to
      # be saved, one error in the flow will rollback everything
      {:error, error} ->
        Logger.warn(error)
        {:ok, nil}
    end
  end

  def run(input, _opts, _context) do
    Logger.info("Creating resource #{input.resource}")

    created_resource =
      input.resource
      |> Changeset.for_create(
        :create,
        input.attributes |> Map.merge(additional_attributes(input))
      )
      |> MmsBiztalk.create()

    case created_resource do
      {:ok, resource_record} ->
        resource_record |> input.resource.to_syncing!() |> input.resource.to_synced()

      # NOTE this skips the invalid records allowing all the valid ones to
      # be saved, one error in the flow will rollback everything
      {:error, error} ->
        Logger.warn(error)
        {:ok, nil}
    end
  end

  def run(input, _opts, _context) do
    Logger.info("Updating resource #{input.resource}")

    resource_to_update =
      input.resource
      |> Ash.Query.for_read(String.to_atom("by_#{input.update_key}"), %{
        input.update_key => input.attributes[input.update_key]
      })
      |> MmsBiztalk.read!()
      |> List.first()

    updated_resource =
      resource_to_update
      |> Changeset.for_update(
        :update,
        input.attributes
      )
      |> MmsBiztalk.update()

    case updated_resource do
      {:ok, resource_record} ->
        transitioned_status_name = "to_#{input.data_type}_synced" |> String.to_atom()
        apply(input.resource, transitioned_status_name, [resource_record])

      # NOTE this skips the invalid records allowing all the valid ones to
      # be saved, one error in the flow will rollback everything
      {:error, error} ->
        Logger.warn(error)
        {:ok, nil}
    end
  end

  def run(input, _opts, _context) do
    Logger.info("Updating resource #{input.resource}")

    resource_to_update =
      input.resource
      |> Ash.Query.for_read(String.to_atom("by_#{input.update_key}"), %{
        input.update_key => input.attributes[input.update_key]
      })
      |> MmsBiztalk.read!()
      |> List.first()

    updated_resource =
      resource_to_update
      |> Changeset.for_update(
        :update,
        input.attributes
      )
      |> MmsBiztalk.update()

    case updated_resource do
      {:ok, resource_record} ->
        transitioned_status_name = "to_#{input.data_type}_synced" |> String.to_atom()
        apply(input.resource, transitioned_status_name, [resource_record])

      # NOTE this skips the invalid records allowing all the valid ones to
      # be saved, one error in the flow will rollback everything
      {:error, error} ->
        Logger.warn(error)
        {:ok, nil}
    end
  end