The subclass of dict is not respected on >=2024.11.0

[emmaai]

Describe the issue:
The subclass is not respected. I noticed the issue on subclass of dict, not sure if it's more general. See below for details.

Minimal Complete Verifiable Example:

from dask.distributed import Client, as_completed

class TestDict(dict):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def __getattr__(self, key: str):
        """ Allow access to items as attributes. """
        v = self.get(key, self)
        if v is self:
            raise AttributeError("has no attribute '{}'".format(key))
        return v

    def __repr__(self) -> str:
        return "(TestDict{})".format(super().__repr__())


def test_subclass(test_dict):
    print(f'input type {type(test_dict)}')
    print(f'get val {test_dict.a}')
    return 0


def main():
    client = Client()
    test_dict = TestDict(a=1)
    futures = [client.submit(test_subclass, test_dict)]
    for f in as_completed(futures):
        res = f.result()
        print(f"result {res}")


if __name__ == "__main__":
    main()

Anything else we need to know?:

The test code works as expected on <2024.11.0, which outputs

input type <class '__main__.TestDict'>
get val 1
result 0

on >=2024.11.0 outputs

input type <class 'dict'>
2024-11-13 05:01:36,954 - distributed.worker - ERROR - Compute Failed
Key:       test_subclass-e8dd185fef44ec112f057d9b352f913f
State:     executing
Task:  <Task 'test_subclass-e8dd185fef44ec112f057d9b352f913f' test_subclass(...)>
Exception: 'AttributeError("\'dict\' object has no attribute \'a\'")'
Traceback: '  File "/home/ubuntu/data/home/ubuntu/test_dask.py", line 20, in test_subclass\n    print(f\'get val {test_dict.a}\')\n'

Traceback (most recent call last):
  File "/home/ubuntu/data/home/ubuntu/test_dask.py", line 34, in <module>
    main()
  File "/home/ubuntu/data/home/ubuntu/test_dask.py", line 29, in main
    res = f.result()
  File "/home/ubuntu/.local/lib/python3.10/site-packages/distributed/client.py", line 402, in result
    return self.client.sync(self._result, callback_timeout=timeout)
  File "/home/ubuntu/data/home/ubuntu/test_dask.py", line 20, in test_subclass
    print(f'get val {test_dict.a}')

Environment:

• Dask version: 2024.11.0
• Python version: 3.10
• Operating System: ubuntu 22.04
• Install method (conda, pip, source): pip