"""
Dictionary-related variable tracking classes for PyTorch Dynamo.

This module implements variable tracking for different types of dictionary-like objects:
- Regular Python dictionaries (dict)
- Ordered dictionaries (collections.OrderedDict)
- Default dictionaries (collections.defaultdict)
- Dictionary views (keys and values)

These classes are responsible for tracking dictionary operations during graph compilation,
maintaining proper guards for dictionary mutations and key existence checks. They handle
dictionary creation, modification, key/value access, and view operations while ensuring
correct behavior in the compiled code through appropriate guard installation.

The implementation uses a special HashableTracker wrapper to handle
dictionary keys while preserving proper aliasing semantics. Set-related classes live
in sets.py.
"""

import collections
import functools
import types
from collections.abc import Callable, Iterator
from typing import Any, TYPE_CHECKING, Union

from torch.utils._pytree import MappingKey

from .. import graph_break_hints, polyfills, variables
from ..bytecode_transformation import (
    create_call_function,
    create_call_method,
    create_dup_top,
    create_instruction,
)
from ..exc import raise_observed_exception, raise_type_error, unimplemented
from ..guards import GuardBuilder, install_guard
from ..source import (
    AttrSource,
    DictGetItemSource,
    is_constant_source,
    is_from_local_source,
)
from ..utils import (
    _item_debug_repr,
    cmp_name_to_op_mapping,
    dict_items,
    dict_keys,
    dict_values,
    istype,
    raise_args_mismatch,
    tracked_repr,
    unpack_iterable,
)
from .base import (
    AttributeMutationExisting,
    AttributeMutationNew,
    AttrMutationKind,
    NO_SUCH_SUBOBJ,
    ValueMutationNew,
    VariableTracker,
)
from .constant import ConstantVariable
from .hashable import HashableTracker, is_hashable, raise_unhashable
from .object_protocol import vt_getitem
from .sets import SetVariable


if TYPE_CHECKING:
    from torch._dynamo.codegen import PyCodegen
    from torch._dynamo.symbolic_convert import InstructionTranslatorBase

    from .functions import UserFunctionVariable


# [Adding a new supported class within the keys of ConstDictVariable]
# - Implement hash_impl(self, tx) on the VariableTracker subclass (or rely on the
#   base class default which uses get_real_python_backed_value())
# - Implement is_python_equal() for key equality


def pydict_check(obj: VariableTracker) -> bool:
    # This is a simplified version of the CPython's PyDict_Check function:
    return issubclass(obj.python_type(), dict)


def _is_set_or_dictview(obj: VariableTracker) -> bool:
    """PyAnySet_Check(other) || PyDictViewSet_Check(other)"""
    try:
        t = obj.python_type()
    except NotImplementedError:
        return False
    return issubclass(t, (set, frozenset, dict_keys, dict_items))


class ConstDictVariable(VariableTracker):
    # PyDict_Type: https://github.com/python/cpython/blob/v3.13.0/Objects/dictobject.c#L4825
    _cpython_type = dict

    CONTAINS_GUARD = GuardBuilder.DICT_CONTAINS
    NOT_CONTAINS_GUARD = GuardBuilder.DICT_NOT_CONTAINS

    _nonvar_fields = {
        "user_cls",
        *VariableTracker._nonvar_fields,
    }

    def __init__(
        self,
        items: dict[VariableTracker, VariableTracker],
        user_cls: type = dict,
        **kwargs: Any,
    ) -> None:
        # .clone() pass these arguments in kwargs but they're recreated a few
        # lines below
        if "original_items" in kwargs:
            kwargs.pop("original_items")
        if "should_reconstruct_all" in kwargs:
            kwargs.pop("should_reconstruct_all")

        super().__init__(**kwargs)

        Hashable = HashableTracker

        # Keys will just be HashableTrackers when cloning, in any other case they'll be VariableTrackers
        if not all(
            isinstance(x, (VariableTracker, Hashable))
            and isinstance(v, VariableTracker)
            for x, v in items.items()
        ):
            raise AssertionError(
                "All keys must be VariableTracker or HashableTracker "
                "and all values must be VariableTracker"
            )

        def make_hashable(
            key: Union[VariableTracker, "HashableTracker"],
        ) -> "HashableTracker":
            return key if isinstance(key, Hashable) else Hashable(key)

        dict_cls = self._get_dict_cls_from_user_cls(user_cls)
        self.items = dict_cls({make_hashable(x): v for x, v in items.items()})
        # need to reconstruct everything if the dictionary is an intermediate value
        # or if a pop/delitem was executed
        self.should_reconstruct_all = (
            not is_from_local_source(self.source) if self.source else True
        )
        self.original_items = items.copy()
        self.user_cls = user_cls

    def _get_dict_cls_from_user_cls(self, user_cls: type) -> type:
        accepted_dict_types = (dict, collections.OrderedDict, collections.defaultdict)

        # avoid executing user code if user_cls is a dict subclass
        if user_cls in accepted_dict_types:
            dict_cls = user_cls
        else:
            # <Subclass, ..., dict, object>
            dict_cls = next(
                base for base in user_cls.__mro__ if base in accepted_dict_types
            )
        if dict_cls not in accepted_dict_types:
            raise AssertionError(f"Unexpected dict_cls: {dict_cls}")

        # Use a dict instead as the call "defaultdict({make_hashable(x): v ..})"
        # would fail as defaultdict expects a callable as first argument
        if dict_cls is collections.defaultdict:
            dict_cls = dict
        return dict_cls

    def as_proxy(self) -> dict[Any, Any]:
        return {k.vt.as_proxy(): v.as_proxy() for k, v in self.items.items()}

    def debug_repr(self) -> str:
        items: list[str] = []
        for k, v in self.items.items():
            key_str = _item_debug_repr(k.vt)
            val_str = _item_debug_repr(v)
            items.append(f"{key_str}: {val_str}")
        return "{" + ", ".join(items) + "}"

    def as_python_constant(self) -> dict[Any, Any]:
        return {
            k.vt.as_python_constant(): v.as_python_constant()
            for k, v in self.items.items()
        }

    def repr_impl(self, tx: "InstructionTranslatorBase") -> VariableTracker:
        items = [
            f"{tracked_repr(tx, key.vt)}: {tracked_repr(tx, value)}"
            for key, value in self.items.items()
        ]
        return VariableTracker.build(tx, "{" + ", ".join(items) + "}")

    def keys_as_python_constant(self) -> dict[Any, VariableTracker]:
        self.install_dict_keys_match_guard()
        return {k.vt.as_python_constant(): v for k, v in self.items.items()}

    def python_type(self) -> type:
        return self.user_cls

    def __contains__(self, vt: VariableTracker) -> bool:
        if not isinstance(vt, VariableTracker):
            raise AssertionError(f"Expected VariableTracker, got {type(vt)}")
        # Use is_hashable as a side-effect-free pre-check.  We can't catch
        # ObservedTypeError from HashableTracker because it modifies
        # tx.exn_vt_stack as a side effect.
        if not is_hashable(vt):
            return False
        key = HashableTracker(vt)
        return key in self.items and not isinstance(
            self.items[key], variables.DeletedVariable
        )

    def call_tree_map_branch(
        self,
        tx: "InstructionTranslatorBase",
        tree_map_fn: "UserFunctionVariable",
        map_fn: VariableTracker,
        rest: list[VariableTracker],
        tree_map_kwargs: dict[str, VariableTracker],
    ) -> VariableTracker:
        other_dicts: list[ConstDictVariable] = []
        for candidate in rest:
            candidate = candidate.realize()
            if not isinstance(candidate, ConstDictVariable) or len(
                candidate.items
            ) != len(self.items):
                return self._tree_map_fallback(
                    tx, tree_map_fn, map_fn, rest, tree_map_kwargs
                )
            other_dicts.append(candidate)

        new_items_hashed = type(self.items)()
        for key_tracker, value in self.items.items():
            sibling_leaves: list[VariableTracker] = []
            for candidate in other_dicts:
                try:
                    sibling_leaves.append(candidate.items[key_tracker])
                except KeyError:
                    return self._tree_map_fallback(
                        tx, tree_map_fn, map_fn, rest, tree_map_kwargs
                    )
            new_items_hashed[key_tracker] = value.call_tree_map(
                tx,
                tree_map_fn,
                map_fn,
                sibling_leaves,
                tree_map_kwargs,
            )

        updated_original_items = {
            key_tracker.vt: new_items_hashed[key_tracker]
            for key_tracker in new_items_hashed
        }

        return self.clone(
            items=new_items_hashed,
            original_items=updated_original_items,
            should_reconstruct_all=True,
            source=None,
            mutation_type=ValueMutationNew(),
        )

    def call_tree_map_with_path_branch(
        self,
        tx: "InstructionTranslatorBase",
        tree_map_fn: "UserFunctionVariable",
        map_fn: VariableTracker,
        rest: list[VariableTracker],
        tree_map_kwargs: dict[str, VariableTracker],
        keypath: tuple[Any, ...],
    ) -> VariableTracker:
        other_dicts: list[ConstDictVariable] = []
        for candidate in rest:
            candidate = candidate.realize()
            if not isinstance(candidate, ConstDictVariable) or len(
                candidate.items
            ) != len(self.items):
                return self._tree_map_with_path_fallback(
                    tx, tree_map_fn, map_fn, rest, tree_map_kwargs, keypath
                )
            other_dicts.append(candidate)

        new_items_hashed = type(self.items)()
        for key_tracker, value in self.items.items():
            sibling_leaves: list[VariableTracker] = []
            for candidate in other_dicts:
                try:
                    sibling_leaves.append(candidate.items[key_tracker])
                except KeyError:
                    return self._tree_map_with_path_fallback(
                        tx, tree_map_fn, map_fn, rest, tree_map_kwargs, keypath
                    )
            key_const = key_tracker.vt.as_python_constant()
            child_keypath = keypath + (MappingKey(key_const),)
            new_items_hashed[key_tracker] = value.call_tree_map_with_path(
                tx,
                tree_map_fn,
                map_fn,
                sibling_leaves,
                tree_map_kwargs,
                child_keypath,
            )

        updated_original_items = {
            key_tracker.vt: new_items_hashed[key_tracker]
            for key_tracker in new_items_hashed
        }

        return self.clone(
            items=new_items_hashed,
            original_items=updated_original_items,
            should_reconstruct_all=True,
            source=None,
            mutation_type=ValueMutationNew(),
        )

    def len(self) -> int:
        return sum(
            not isinstance(x, variables.DeletedVariable) for x in self.items.values()
        )

    def has_new_items(self) -> bool:
        return self.should_reconstruct_all or any(
            self.is_new_item(self.original_items.get(key.vt), value)
            for key, value in self.items.items()
        )

    def is_new_item(
        self, value: VariableTracker | None, other: VariableTracker
    ) -> bool:
        # compare the id of the realized values if both values are not lazy VTs
        if value and value.is_realized() and other.is_realized():
            return id(value.realize()) != id(other.realize())
        return id(value) != id(other)

    def reconstruct_kvs_into_new_dict(self, codegen: "PyCodegen") -> None:
        # Build a dictionary that contains the keys and values.
        num_args = 0
        for key, value in self.items.items():
            # We can safely call realize() here as it won't introduce any new guards
            item = self.original_items.get(key.vt)
            if self.is_new_item(item, value) or self.should_reconstruct_all:
                codegen(key.vt)
                codegen(value)
                num_args += 1
        codegen.append_output(create_instruction("BUILD_MAP", arg=num_args))

    def reconstruct(self, codegen: "PyCodegen") -> None:
        if self.user_cls is collections.OrderedDict:
            # emit `OrderedDict(constructed_dict)`
            codegen.add_push_null(
                lambda: codegen.extend_output(
                    [
                        codegen.create_load_python_module(collections),
                        codegen.create_load_attr("OrderedDict"),
                    ]
                )
            )
            if self._contains_self_reference():
                codegen.extend_output(
                    [
                        *create_call_function(0, False),
                        create_dup_top(),
                    ]
                )
                codegen.add_cache(self)

                codegen.append_output(create_dup_top())
                codegen.load_method("update")
                self.reconstruct_kvs_into_new_dict(codegen)
                codegen.extend_output(
                    [
                        *create_call_method(1),
                        create_instruction("POP_TOP"),
                    ]
                )
            else:
                self.reconstruct_kvs_into_new_dict(codegen)
                codegen.extend_output(create_call_function(1, False))
        else:
            if self._contains_self_reference():
                codegen.extend_output(
                    [
                        create_instruction("BUILD_MAP", arg=0),
                        create_dup_top(),
                    ]
                )
                codegen.add_cache(self)
                self.reconstruct_kvs_into_new_dict(codegen)
                codegen.append_output(create_instruction("DICT_UPDATE", arg=1))
            else:
                # Non-self-referential: use simple codegen
                self.reconstruct_kvs_into_new_dict(codegen)

    def reconstruct_pycode(self, codegen) -> str:
        if (
            issubclass(self.user_cls, collections.OrderedDict)
            or self._contains_self_reference()
        ):
            raise NotImplementedError
        return f"{{{', '.join(f'{k.vt.reconstruct_pycode(codegen)}: {v.reconstruct_pycode(codegen)}' for k, v in self.items.items())}}}"

    def getitem_const_raise_exception_if_absent(
        self, tx: "InstructionTranslatorBase", arg: VariableTracker
    ) -> VariableTracker:
        key = HashableTracker(arg)
        if key not in self.items:
            raise_observed_exception(KeyError, tx, args=[arg])
        return self.items[key]

    def getitem_const(
        self, tx: "InstructionTranslatorBase", arg: VariableTracker
    ) -> VariableTracker:
        key = HashableTracker(arg)
        if key not in self.items:
            msg = f"Dictionary key {arg.value} not found during tracing"  # type: ignore[attr-defined]
            unimplemented(
                gb_type="key not found in dict",
                context=f"Key {arg.value}",  # type: ignore[attr-defined]
                explanation=msg,
                hints=[
                    "Check if the key exists in the dictionary before accessing it.",
                    *graph_break_hints.USER_ERROR,
                ],
            )
        return self.items[key]

    def maybe_getitem_const(self, arg: VariableTracker) -> VariableTracker | None:
        key = HashableTracker(arg)
        if key not in self.items:
            return None
        return self.items[key]

    def realize_key_vt(self, arg: VariableTracker) -> None:
        # Realize the LazyVT on a particular index
        if arg not in self:
            raise AssertionError(f"Key {arg} not found in dict")
        key = HashableTracker(arg)
        index = tuple(self.items.keys()).index(key)
        original_key_vt = tuple(self.original_items.keys())[index]
        if isinstance(original_key_vt, variables.LazyVariableTracker):
            original_key_vt.realize()

    def install_dict_keys_match_guard(self) -> None:
        if self.source:
            install_guard(self.make_guard(GuardBuilder.DICT_KEYS_MATCH))

    def install_dict_contains_guard(
        self, tx: "InstructionTranslatorBase", args: list[VariableTracker]
    ) -> None:
        # Key guarding - These are the cases to consider
        # 1) The dict has been mutated. In this case, we would have already
        # inserted a DICT_KEYS_MATCH guard, so we can skip.
        #
        # 2) args[0].source is None. This happens for const keys. Here, we
        # have to insert the DICT_CONTAINS guard.
        #
        # 3) args[0].source is not None. This can happen for non-const VTs.
        #   3a) contains=True. In this case, we can access the lazyVT from
        #   original_items and selectively realize it.
        #   3b) contains=False. There is no easy way to selectively apply this
        #   DICT_NOT_CONTAINS guard because our guard are represented via trees.
        #   Be conservative and add DICT_KEYS_MATCH guard.

        if not self.source:
            return

        if tx.output.side_effects.is_modified(self):
            return

        contains = args[0] in self
        if args[0].source is None and args[0].is_python_constant():
            guard_fn = (
                type(self).CONTAINS_GUARD if contains else type(self).NOT_CONTAINS_GUARD
            )
            install_guard(
                self.make_guard(
                    functools.partial(
                        guard_fn,
                        key=args[0].as_python_constant(),
                    )
                )
            )
        elif args[0].source:
            if contains:
                self.realize_key_vt(args[0])
            else:
                self.install_dict_keys_match_guard()

    def mp_subscript_impl(
        self,
        tx: "InstructionTranslatorBase",
        key: VariableTracker,
    ) -> VariableTracker:
        # dict_subscript: https://github.com/python/cpython/blob/62a6e898e01/Objects/dictobject.c#L3673-L3706
        # Unhashable key check happens inside HashableTracker (hash_impl → TypeError).
        return self.getitem_const_raise_exception_if_absent(tx, key)

    def sq_contains(
        self, tx: "InstructionTranslatorBase", item: VariableTracker
    ) -> VariableTracker:
        # ref: https://github.com/python/cpython/blob/v3.13.0/Objects/dictobject.c#L4657-L4668
        if not is_hashable(item):
            raise_type_error(tx, f"unhashable type: '{item.python_type_name()}'")
        self.install_dict_contains_guard(tx, [item])
        contains = item in self
        return VariableTracker.build(tx, contains)

    def tp_iter_impl(self, tx: "InstructionTranslatorBase") -> VariableTracker:
        from .iter import DictIterator

        if self.source and not is_constant_source(self.source):
            install_guard(self.make_guard(GuardBuilder.DICT_KEYS_MATCH))
            tx.output.guard_on_key_order.add(self.source)
        return DictIterator(self.items.keys())

    def call_method(
        self,
        tx: "InstructionTranslatorBase",
        name: str,
        args: list[VariableTracker],
        kwargs: dict[str, VariableTracker],
    ) -> VariableTracker:
        # NB - Both key and value are LazyVariableTrackers in the beginning. So,
        # we have to insert guards when a dict method is accessed. For this to
        # be simple, we are conservative and overguard. We skip guard only for
        # get/__getitem__ because the key guard will be inserted by the
        # corresponding value VT. For __contains__, we add a DICT_CONTAINS
        # guard. But for all the other methods, we insert the DICT_KEYS_MATCH
        # guard to be conservative.
        from . import DictBuiltinVariable

        Hashable = HashableTracker

        if name == "__init__":
            temp_dict_vt = DictBuiltinVariable.call_custom_dict(
                tx, dict, *args, **kwargs
            )
            tx.output.side_effects.mutation(self)
            self.items.update(temp_dict_vt.items)  # type: ignore[attr-defined]
            return ConstantVariable.create(None)
        elif name == "items":
            if args or kwargs:
                raise_args_mismatch(
                    tx,
                    name,
                    "0 args and 0 kwargs",
                    f"{len(args)} args and {len(kwargs)} kwargs",
                )
            self.install_dict_keys_match_guard()
            if self.source:
                tx.output.guard_on_key_order.add(self.source)
            return DictItemsVariable(self)
        elif name == "keys":
            if len(args):
                raise_args_mismatch(tx, name, "0 args", f"{len(args)} args")
            self.install_dict_keys_match_guard()
            if self.source:
                tx.output.guard_on_key_order.add(self.source)
            return DictKeysVariable(self)
        elif name == "__reversed__":
            # dict.__reversed__: reverse insertion-order key iterator.
            # ref: https://github.com/python/cpython/blob/v3.13.0/Objects/dictobject.c (dict___reversed___impl)
            if args or kwargs:
                raise_args_mismatch(
                    tx,
                    name,
                    "0 args and 0 kwargs",
                    f"{len(args)} args and {len(kwargs)} kwargs",
                )
            self.install_dict_keys_match_guard()
            if self.source:
                tx.output.guard_on_key_order.add(self.source)
            reversed_keys = [k.vt for k in reversed(list(self.items.keys()))]
            return variables.ListIteratorVariable(
                reversed_keys, mutation_type=ValueMutationNew()
            )
        elif name == "values":
            if args or kwargs:
                raise_args_mismatch(
                    tx,
                    name,
                    "0 args and 0 kwargs",
                    f"{len(args)} args and {len(kwargs)} kwargs",
                )
            self.install_dict_keys_match_guard()
            if self.source:
                tx.output.guard_on_key_order.add(self.source)
            if args or kwargs:
                raise_observed_exception(TypeError, tx)
            return DictValuesVariable(self)
        elif name == "copy":
            self.install_dict_keys_match_guard()
            if args or kwargs:
                raise_args_mismatch(
                    tx,
                    name,
                    "0 args and 0 kwargs",
                    f"{len(args)} args and {len(kwargs)} kwargs",
                )
            return self.clone(
                items=self.items.copy(), mutation_type=ValueMutationNew(), source=None
            )
        elif name == "get":
            if len(args) not in (1, 2):
                raise_args_mismatch(tx, name, "1 or 2 args", f"{len(args)} args")

            if args[0] not in self:
                self.install_dict_contains_guard(tx, args)
                if len(args) == 1:
                    # if default is not given, return None
                    return ConstantVariable.create(None)
                return args[1]
            # Key guarding - Nothing to do.
            return self.getitem_const(tx, args[0])
        elif name == "pop" and self.is_mutable():
            if len(args) not in (1, 2):
                raise_args_mismatch(tx, name, "1 or 2 args", f"{len(args)} args")

            if args[0] not in self:
                # missing item, return the default value. Install no DICT_CONTAINS guard.
                self.install_dict_contains_guard(tx, args)
                if len(args) == 1:
                    # if default is not given, raise KeyError
                    raise_observed_exception(KeyError, tx)
                return args[1]

            self.should_reconstruct_all = True
            tx.output.side_effects.mutation(self)
            return self.items.pop(Hashable(args[0]))
        elif name == "popitem" and self.is_mutable():
            # dict.popitem() takes no args. OrderedDict.popitem(last=) is
            # handled by OrderedDictVariable.call_method.
            if len(args):
                raise_args_mismatch(tx, name)

            if not self.items:
                raise_observed_exception(
                    KeyError,
                    tx,
                    args=[
                        "popitem(): dictionary is empty",
                    ],
                )

            k, v = self.items.popitem()
            self.should_reconstruct_all = True
            tx.output.side_effects.mutation(self)

            return variables.TupleVariable([k.vt, v])
        elif name == "clear":
            if args or kwargs:
                raise_args_mismatch(
                    tx,
                    name,
                    "0 args and 0 kwargs",
                    f"{len(args)} args and {len(kwargs)} kwargs",
                )
            self.should_reconstruct_all = True
            tx.output.side_effects.mutation(self)
            self.items.clear()
            return ConstantVariable.create(None)
        elif name == "update" and self.is_mutable():
            # Mirrors CPython PyDict_Merge: if arg has keys(), iterate keys and
            # __getitem__; else treat arg as iterable of (key, value) pairs.
            # kwargs are always merged on top.
            # ref: https://github.com/python/cpython/blob/60403a5409ff2c3f3b07dd2ca91a7a3e096839c7/Objects/dictobject.c#L3571
            self.install_dict_keys_match_guard()
            if len(args) > 1:
                raise_args_mismatch(tx, name, "at most 1 args", f"{len(args)} args")
            if args or kwargs:
                tx.output.side_effects.mutation(self)
            if args:
                other = args[0]
                if isinstance(other, ConstDictVariable):
                    # NB - Guard on all the keys of the other dict to ensure
                    # correctness.
                    other.install_dict_keys_match_guard()
                    self.items.update(other.items)
                elif (
                    isinstance(
                        other,
                        (variables.UserDefinedObjectVariable, MappingProxyVariable),
                    )
                    and other.call_obj_hasattr(tx, "keys").as_python_constant()
                ):
                    keys = other.call_method(tx, "keys", [], {})
                    for key in unpack_iterable(tx, keys):
                        self.items[Hashable(key)] = vt_getitem(tx, other, key)
                else:
                    for idx, item in enumerate(unpack_iterable(tx, other)):
                        pair = unpack_iterable(tx, item)
                        if len(pair) != 2:
                            raise_observed_exception(
                                ValueError,
                                tx,
                                args=[
                                    "dictionary update sequence element "
                                    f"#{idx} has length {len(pair)}; 2 is required"
                                ],
                            )
                        self.items[Hashable(pair[0])] = pair[1]
            for k, v in kwargs.items():
                self.items[Hashable(VariableTracker.build(tx, k))] = v
            return ConstantVariable.create(None)
        elif name == "setdefault" and self.is_mutable():
            if len(args) not in (1, 2):
                raise_args_mismatch(
                    tx,
                    name,
                    "1 or 2 args and 0 kwargs",
                    f"{len(args)} args and {len(kwargs)} kwargs",
                )

            self.install_dict_keys_match_guard()
            if kwargs or len(args) > 2:
                raise_args_mismatch(
                    tx,
                    name,
                    "at most 2 args and 0 kwargs",
                    f"{len(args)} args and {len(kwargs)} kwargs",
                )
            value = self.maybe_getitem_const(args[0])
            if value is not None:
                return value
            else:
                if len(args) == 1:
                    x = ConstantVariable.create(None)
                else:
                    x = args[1]
                tx.output.side_effects.mutation(self)
                self.items[Hashable(args[0])] = x
                return x
        else:
            return super().call_method(tx, name, args, kwargs)

    def unpack_var_sequence(
        self, tx: "InstructionTranslatorBase"
    ) -> list[VariableTracker]:
        self.install_dict_keys_match_guard()
        return [x.vt for x in self.items]

    def nb_or_impl(
        self,
        tx: Any,
        other: VariableTracker,
        reverse: bool = False,
    ) -> VariableTracker:
        # ref: https://github.com/python/cpython/blob/3.13/Objects/dictobject.c#L4643-L4658
        self_, other_ = (other, self) if reverse else (self, other)
        if pydict_check(self_) and pydict_check(other_):
            new = self_.call_method(tx, "copy", [], {})
            new.call_method(tx, "update", [other_], {})
            return new
        return ConstantVariable.create(NotImplemented)

    def nb_inplace_or_impl(
        self,
        tx: Any,
        other: VariableTracker,
    ) -> VariableTracker:
        # ref: https://github.com/python/cpython/blob/3.13/Objects/dictobject.c#L4660-L4667
        self.call_method(tx, "update", [other], {})
        return self

    def mp_ass_subscript_impl(
        self,
        tx: "InstructionTranslatorBase",
        key: VariableTracker,
        value: VariableTracker | None,
    ) -> VariableTracker:
        # ref: https://github.com/python/cpython/blob/v3.13.0/Objects/dictobject.c (dict_ass_sub)
        if not self.is_mutable():
            return super().mp_ass_subscript_impl(tx, key, value)

        # value=None signals delete (CPython NULL sentinel).
        if not is_hashable(key):
            raise_unhashable(key, tx)
        self.install_dict_keys_match_guard()
        hkey = HashableTracker(key)
        tx.output.side_effects.mutation(self)
        if value is None:
            if hkey not in self.items:
                raise_observed_exception(KeyError, tx, args=[key.as_python_constant()])
            self.should_reconstruct_all = True
            del self.items[hkey]
        else:
            self.items[hkey] = value
        return ConstantVariable.create(None)

    def mp_length(self, tx: "InstructionTranslatorBase") -> VariableTracker:
        """Mapping length for dict objects."""
        self.install_dict_keys_match_guard()
        return VariableTracker.build(tx, len(self.items))

    def call_obj_hasattr(
        self, tx: "InstructionTranslatorBase", name: str
    ) -> ConstantVariable:
        # dict not allow setting arbitrary attributes.  OrderedDict and
        # defaultdict allow arbitrary setattr, but not deletion of default attrs
        if any(
            self.user_cls is t
            for t in (dict, collections.OrderedDict, collections.defaultdict)
        ):
            if hasattr(self.user_cls, name):
                return ConstantVariable.create(True)
            if self.user_cls is dict:
                return ConstantVariable.create(False)

        msg = f"hasattr on {self.user_cls} is not supported"
        unimplemented(
            gb_type="unsupported hasattr operation",
            context=f"Class {self.user_cls}",
            explanation=msg,
            hints=[
                "Consider using a regular dictionary instead",
                *graph_break_hints.SUPPORTABLE,
            ],
        )

    def clone(self, **kwargs: Any) -> VariableTracker:
        self.install_dict_keys_match_guard()
        return super().clone(**kwargs)

    def is_hashable(self) -> bool:
        return False

    def hash_impl(self, tx: "InstructionTranslatorBase") -> tuple[int, bool]:
        from ..exc import raise_type_error

        raise_type_error(tx, f"unhashable type: '{self.python_type_name()}'")

    def richcompare_impl(
        self, tx: "InstructionTranslatorBase", other: VariableTracker, op: str
    ) -> VariableTracker:
        # dict_richcompare: https://github.com/python/cpython/blob/e76aa128fe/Objects/dictobject.c#L4198
        # Only supports eq/ne; returns NotImplemented for ordering.
        from .builder import SourcelessBuilder
        from .user_defined import UserDefinedDictVariable

        if op not in ("__eq__", "__ne__"):
            return ConstantVariable.create(NotImplemented)
        if not isinstance(other, ConstDictVariable):
            # Unwrap UserDefinedDictVariable to its base ConstDictVariable.
            # This is correct because CPython's dict_equal operates on the
            # internal C struct directly (ma_used, dk_entries, _Py_dict_lookup)
            # -- it never calls __getitem__ or __len__ on dict subclasses.
            # https://github.com/python/cpython/blob/e76aa128fe/Objects/dictobject.c#L4125-L4185
            if isinstance(other, UserDefinedDictVariable):
                if other._base_vt is None:
                    raise AssertionError("expected _base_vt to be set")
                other = other._base_vt
            else:
                return ConstantVariable.create(NotImplemented)
        eq_result = SourcelessBuilder.create(tx, polyfills.dict___eq__).call_function(
            tx, [self, other], {}
        )
        if op == "__ne__":
            return VariableTracker.build(tx, not eq_result.as_python_constant())
        return eq_result

    def var_getattr(self, tx: "InstructionTranslatorBase", name: str):
        if name == "__class__":
            return VariableTracker.build(tx, self.python_type())
        return super().var_getattr(tx, name)


class MappingProxyVariable(VariableTracker):
    # PyDictProxy_Type: https://github.com/python/cpython/blob/v3.13.0/Objects/descrobject.c#L1995
    _cpython_type = types.MappingProxyType

    # proxies to the original dict_vt
    def __init__(self, dv_dict: ConstDictVariable, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        if not isinstance(dv_dict, ConstDictVariable):
            raise AssertionError(f"Expected ConstDictVariable, got {type(dv_dict)}")
        self.dv_dict = dv_dict

    def python_type(self) -> type:
        return types.MappingProxyType

    def hash_impl(self, tx: "InstructionTranslatorBase") -> tuple[int, bool]:
        # mappingproxy.__hash__ delegates to the underlying dict, which
        # raises TypeError. Mirror that behavior.
        return self.dv_dict.hash_impl(tx)

    def unpack_var_sequence(
        self, tx: "InstructionTranslatorBase"
    ) -> list[VariableTracker]:
        return self.dv_dict.unpack_var_sequence(tx)

    def reconstruct(self, codegen: "PyCodegen") -> None:
        # load types.MappingProxyType
        if self.source:
            msg = (
                f"Preexisting MappingProxyVariable (source: {self.source}) cannot be reconstructed "
                "because the connection to the original dict will be lost."
            )
            unimplemented(
                gb_type="mapping proxy cannot be reconstructed",
                context=f"Source: {self.source}",
                explanation=msg,
                hints=[
                    "Use a mapping proxy constructed in the same `torch.compile` region.",
                    *graph_break_hints.SUPPORTABLE,
                ],
            )
        codegen.add_push_null(
            lambda: codegen.extend_output(
                [
                    codegen.create_load_python_module(types),
                    codegen.create_load_attr("MappingProxyType"),
                ]
            )
        )
        codegen(self.dv_dict)
        codegen.extend_output(create_call_function(1, False))

    def _check_mutation_guard(self, tx: "InstructionTranslatorBase") -> None:
        if self.source and tx.output.side_effects.has_existing_dict_mutation():
            msg = (
                "A dict has been modified while we have an existing mappingproxy object. "
                "A mapping proxy object, as the name suggest, proxies a mapping "
                "object (usually a dict). If the original dict object mutates, it "
                "is reflected in the proxy object as well. For an existing proxy "
                "object, we do not know the original dict it points to. Therefore, "
                "for correctness we graph break when there is dict mutation and we "
                "are trying to access a proxy object."
            )

            unimplemented(
                gb_type="mapping proxy affected by dictionary mutation",
                context=f"Source: {self.source}, Dict mutation detected",
                explanation=msg,
                hints=[
                    "Avoid modifying dictionaries that might be referenced by mapping proxy objects",
                    "Or avoid using the mapping proxy objects after modifying its underlying dictionary",
                ],
            )

    def mp_subscript_impl(
        self,
        tx: "InstructionTranslatorBase",
        key: VariableTracker,
    ) -> VariableTracker:
        # mappingproxy_getitem: https://github.com/python/cpython/blob/62a6e898e01/Objects/descrobject.c#L1052-L1056
        # TODO(follow-up): add tests for invalid key type, missing key
        self._check_mutation_guard(tx)
        return self.dv_dict.mp_subscript_impl(tx, key)

    def call_method(
        self,
        tx: "InstructionTranslatorBase",
        name: str,
        args: list[VariableTracker],
        kwargs: dict[str, VariableTracker],
    ) -> VariableTracker:
        self._check_mutation_guard(tx)
        return self.dv_dict.call_method(tx, name, args, kwargs)

    def tp_iter_impl(self, tx: "InstructionTranslatorBase") -> VariableTracker:
        return self.dv_dict.tp_iter_impl(tx)

    def sq_contains(
        self, tx: "InstructionTranslatorBase", item: VariableTracker
    ) -> VariableTracker:
        # https://github.com/python/cpython/blob/60403a5409ff/Objects/descrobject.c#L1087-L1095
        return self.dv_dict.sq_contains(tx, item)

    def mp_length(self, tx: "InstructionTranslatorBase") -> VariableTracker:
        return self.dv_dict.mp_length(tx)

    def richcompare_impl(
        self, tx: "InstructionTranslatorBase", other: VariableTracker, op: str
    ) -> VariableTracker:
        # mappingproxy_richcompare delegates to the underlying mapping:
        # https://github.com/python/cpython/blob/60403a5409ff/Objects/descrobject.c#L1231-L1236
        from .object_protocol import generic_richcompare

        return generic_richcompare(tx, self.dv_dict, other, op)

    def call_obj_hasattr(
        self, tx: "InstructionTranslatorBase", name: str
    ) -> ConstantVariable:
        if self.python_type() is types.MappingProxyType:
            return VariableTracker.build(tx, name in types.MappingProxyType.__dict__)
        return super().call_obj_hasattr(tx, name)


class NNModuleHooksDictVariable(ConstDictVariable):
    # Special class to avoid adding any guards on the nn module hook ids.
    def install_dict_keys_match_guard(self) -> None:
        pass

    def install_dict_contains_guard(
        self, tx: "InstructionTranslatorBase", args: list[VariableTracker]
    ) -> None:
        pass


class DictViewVariable(VariableTracker):
    """
    Models _PyDictViewObject

    This is an "abstract" class. Subclasses will override kv and the items method
    """

    kv: str | None = None

    def __init__(self, dv_dict: ConstDictVariable, **kwargs: Any) -> None:
        super().__init__(**kwargs)
        if self.kv not in ("keys", "values", "items"):
            raise AssertionError(
                f"Expected kv to be 'keys', 'values', or 'items', got {self.kv!r}"
            )
        if not isinstance(dv_dict, ConstDictVariable):
            raise AssertionError(f"Expected ConstDictVariable, got {type(dv_dict)}")
        self.dv_dict = dv_dict

    @property
    def view_items(self) -> Any:
        if self.kv is None:
            raise AssertionError("kv must not be None")
        return getattr(self.dv_dict.items, self.kv)()

    def is_hashable(self) -> bool:
        return False

    def hash_impl(self, tx: "InstructionTranslatorBase") -> tuple[int, bool]:
        from ..exc import raise_type_error

        raise_type_error(tx, f"unhashable type: '{self.python_type_name()}'")

    @property
    def view_items_vt(self) -> list[VariableTracker]:
        # Returns an iterable of the unpacked items
        # Implement in the subclasses
        raise NotImplementedError

    def unpack_var_sequence(
        self, tx: "InstructionTranslatorBase"
    ) -> list[VariableTracker]:
        return self.view_items_vt

    def reconstruct(self, codegen: "PyCodegen") -> None:
        if self.kv is None:
            raise AssertionError("kv must not be None for reconstruct")
        codegen(self.dv_dict)
        codegen.load_method(self.kv)
        codegen.call_method(0)

    def call_obj_hasattr(
        self, tx: "InstructionTranslatorBase", name: str
    ) -> ConstantVariable:
        if self.kv is None:
            raise AssertionError("kv must not be None for call_obj_hasattr")
        if name in self.python_type().__dict__:
            return ConstantVariable.create(True)
        return ConstantVariable.create(False)

    def repr_impl(self, tx: "InstructionTranslatorBase") -> VariableTracker:
        if self.kv == "keys":
            items = ", ".join(tracked_repr(tx, key.vt) for key in self.view_items)
            return VariableTracker.build(tx, f"dict_keys([{items}])")
        if self.kv == "values":
            items = ", ".join(tracked_repr(tx, value) for value in self.view_items)
            return VariableTracker.build(tx, f"dict_values([{items}])")
        items = ", ".join(
            f"({tracked_repr(tx, key.vt)}, {tracked_repr(tx, value)})"
            for key, value in self.view_items
        )
        return VariableTracker.build(tx, f"dict_items([{items}])")

    def call_method(
        self,
        tx: "InstructionTranslatorBase",
        name: str,
        args: list[VariableTracker],
        kwargs: dict[str, VariableTracker],
    ) -> VariableTracker:
        if name == "__reversed__":
            # dict_keys/values/items __reversed__: reverse insertion order.
            if args or kwargs:
                raise_args_mismatch(
                    tx,
                    name,
                    "0 args and 0 kwargs",
                    f"{len(args)} args and {len(kwargs)} kwargs",
                )
            if self.dv_dict.source and not is_constant_source(self.dv_dict.source):
                tx.output.guard_on_key_order.add(self.dv_dict.source)
            return variables.ListIteratorVariable(
                list(reversed(self.view_items_vt)),
                mutation_type=ValueMutationNew(),
            )
        return super().call_method(tx, name, args, kwargs)

    def sq_length(self, tx: "InstructionTranslatorBase") -> VariableTracker:
        """Sequence length for dict view objects."""
        return VariableTracker.build(tx, len(self.view_items))

    def nb_or_impl(
        self,
        tx: "InstructionTranslatorBase",
        other: VariableTracker,
        reverse: bool = False,
    ) -> VariableTracker:
        # ref: https://github.com/python/cpython/blob/v3.13.0/Objects/dictobject.c#L6142-L6155 (dictviews_or)
        s = VariableTracker.build(tx, set).call_function(tx, [self], {})
        s.call_method(tx, "update", [other], {})
        return s

    def nb_subtract_impl(
        self,
        tx: "InstructionTranslatorBase",
        other: VariableTracker,
        reverse: bool = False,
    ) -> VariableTracker:
        # ref: https://github.com/python/cpython/blob/v3.13.0/Objects/dictobject.c#L6036 (dictviews_sub)
        self_, other_ = (other, self) if reverse else (self, other)
        s = VariableTracker.build(tx, set).call_function(tx, [self_], {})
        s.call_method(tx, "difference_update", [other_], {})
        return s

    def nb_and_impl(
        self,
        tx: "InstructionTranslatorBase",
        other: VariableTracker,
        reverse: bool = False,
    ) -> VariableTracker:
        # ref: https://github.com/python/cpython/blob/3.13/Objects/dictobject.c#L6105-L6188 (_PyDictView_Intersect)
        s = VariableTracker.build(tx, set).call_function(tx, [self], {})
        s.call_method(tx, "intersection_update", [other], {})
        return s

    def nb_xor_impl(
        self,
        tx: "InstructionTranslatorBase",
        other: VariableTracker,
        reverse: bool = False,
    ) -> VariableTracker:
        # ref: https://github.com/python/cpython/blob/3.13/Objects/dictobject.c#L6305-L6325 (dictviews_xor)
        s = VariableTracker.build(tx, set).call_function(tx, [self], {})
        s.call_method(tx, "symmetric_difference_update", [other], {})
        return s


class DictKeysVariable(DictViewVariable):
    # PyDictKeys_Type: https://github.com/python/cpython/blob/v3.13.0/Objects/dictobject.c#L6365
    _cpython_type = dict_keys

    kv = "keys"

    @property
    def set_items(self) -> set[VariableTracker]:
        return set(self.view_items)

    @property
    def view_items_vt(self) -> list[VariableTracker]:
        # Returns an iterable of the unpacked items
        return [x.vt for x in self.view_items]

    def python_type(self) -> type:
        return dict_keys

    def tp_iter_impl(self, tx: "InstructionTranslatorBase") -> VariableTracker:
        from .iter import DictKeysIterator

        if self.dv_dict.source and not is_constant_source(self.dv_dict.source):
            tx.output.guard_on_key_order.add(self.dv_dict.source)
        return DictKeysIterator(self.dv_dict.items)

    def debug_repr(self) -> str:
        if not self.view_items:
            return "dict_keys([])"
        else:
            items: list[str] = []
            for k in self.view_items:
                key_str = _item_debug_repr(k.vt)
                items.append(key_str)
            return "dict_keys([" + ", ".join(items) + "])"

    def sq_contains(
        self, tx: "InstructionTranslatorBase", item: VariableTracker
    ) -> VariableTracker:
        # ref: https://github.com/python/cpython/blob/v3.13.0/Objects/dictobject.c#L5998-L6005
        return self.dv_dict.sq_contains(tx, item)

    def richcompare_impl(
        self, tx: "InstructionTranslatorBase", other: VariableTracker, op: str
    ) -> VariableTracker:
        # dictview_richcompare: accepts set/frozenset and dict_keys/dict_items.
        # https://github.com/python/cpython/blob/e76aa128fe/Objects/dictobject.c#L5952-L6010
        # Uses a polyfill with len() and ``in`` so that Dynamo traces through
        # sq_length/sq_contains, matching CPython's use of PyObject_Size and
        # PySequence_Contains.
        from .builder import SourcelessBuilder

        if not _is_set_or_dictview(other):
            return ConstantVariable.create(NotImplemented)
        return SourcelessBuilder.create(
            tx, polyfills.dictview_richcompare
        ).call_function(
            tx,
            [
                SourcelessBuilder.create(tx, cmp_name_to_op_mapping[op]),
                self,
                other,
            ],
            {},
        )

    def call_method(
        self,
        tx: "InstructionTranslatorBase",
        name: str,
        args: list[VariableTracker],
        kwargs: dict[str, VariableTracker],
    ) -> VariableTracker:
        if name in (
            "__and__",
            "__iand__",
            "__xor__",
            "__ixor__",
        ):
            # These methods always returns a set
            m = getattr(self.set_items, name)
            r = m(args[0].set_items)  # type: ignore[attr-defined]
            return SetVariable(r)
        return super().call_method(tx, name, args, kwargs)


class DictValuesVariable(DictViewVariable):
    # PyDictValues_Type: https://github.com/python/cpython/blob/e76aa128fe/Objects/dictobject.c#L6615
    _cpython_type = dict_values

    # dict_values is hashable (identity hash), unlike dict_keys/dict_items.
    #
    # CPython only inherits tp_hash from the base when BOTH tp_richcompare
    # AND tp_hash are NULL on the type
    # (https://github.com/python/cpython/blob/e76aa128fe/Objects/typeobject.c#L7665-L7679).
    # dict_keys/dict_items set tp_richcompare (for set-like comparisons),
    # so their tp_hash is NOT inherited — it stays NULL, and
    # type_ready_set_hash sets it to PyObject_HashNotImplemented (unhashable)
    # (https://github.com/python/cpython/blob/e76aa128fe/Objects/typeobject.c#L8066-L8085).
    # dict_values does NOT set tp_richcompare
    # (https://github.com/python/cpython/blob/e76aa128fe/Objects/dictobject.c#L6630),
    # so it inherits object's tp_hash (PyObject_GenericHash) — hashable.
    #
    # Override DictViewVariable.hash_impl to restore the base identity hash.
    kv = "values"

    # dict.values() do not implement tp_as_number
    nb_or_impl = None  # type: ignore[bad-override]
    nb_inplace_or_impl = None  # type: ignore[bad-override]
    nb_subtract_impl = None  # type: ignore[bad-override]
    nb_inplace_subtract_impl = None  # type: ignore[bad-override]
    nb_and_impl = None  # type: ignore[bad-override]
    nb_inplace_and_impl = None  # type: ignore[bad-override]
    nb_xor_impl = None  # type: ignore[bad-override]
    nb_inplace_xor_impl = None  # type: ignore[bad-override]

    def is_hashable(self) -> bool:
        return True

    def hash_impl(self, tx: "InstructionTranslatorBase") -> tuple[int, bool]:
        return super(DictViewVariable, self).hash_impl(tx)

    @property
    def view_items_vt(self) -> list[VariableTracker]:
        return list(self.view_items)

    def python_type(self) -> type:
        return dict_values

    def tp_iter_impl(self, tx: "InstructionTranslatorBase") -> VariableTracker:
        from .iter import DictValuesIterator

        if self.dv_dict.source and not is_constant_source(self.dv_dict.source):
            tx.output.guard_on_key_order.add(self.dv_dict.source)
        return DictValuesIterator(self.dv_dict.items)

    def richcompare_impl(
        self, tx: "InstructionTranslatorBase", other: VariableTracker, op: str
    ) -> VariableTracker:
        # dict_values has no tp_richcompare (inherits object's).
        from .object_protocol import object_richcompare

        return object_richcompare(self, tx, other, op)

    def debug_repr(self) -> str:
        if not self.view_items:
            return "dict_values([])"
        else:
            items: list[str] = []
            for v in self.view_items:
                val_str = _item_debug_repr(v)
                items.append(val_str)
            return "dict_values([" + ", ".join(items) + "])"


class DictItemsVariable(DictViewVariable):
    # PyDictItems_Type: https://github.com/python/cpython/blob/v3.13.0/Objects/dictobject.c#L6477
    _cpython_type = dict_items

    kv = "items"

    @property
    def set_items(self) -> set["HashableTracker"]:
        return {
            HashableTracker(variables.TupleVariable([k.vt, v]))
            for k, v in self.view_items
        }

    @property
    def view_items_vt(self) -> list[VariableTracker]:
        # Returns an iterable of the unpacked items
        return [variables.TupleVariable([k.vt, v]) for k, v in self.view_items]

    def python_type(self) -> type:
        return dict_items

    def debug_repr(self) -> str:
        if not self.view_items:
            return "dict_items([])"
        else:
            items: list[str] = []
            for k, v in self.view_items:
                key_str = _item_debug_repr(k.vt)
                val_str = _item_debug_repr(v)
                items.append(f"({key_str}, {val_str})")
            return "dict_items([" + ", ".join(items) + "])"

    def sq_contains(
        self, tx: "InstructionTranslatorBase", item: VariableTracker
    ) -> VariableTracker:
        # ref: https://github.com/python/cpython/blob/v3.13.0/Objects/dictobject.c#L6433-L6451
        from ..utils import iter_contains

        if not is_hashable(item):
            raise_type_error(tx, f"unhashable type: '{item.python_type_name()}'")

        # Fast path: if item is a known (key, value) pair, use O(1) dict lookup.
        if isinstance(item, variables.TupleVariable) and len(item.items) == 2:
            key, val = item.items
            key_ht = HashableTracker(key)
            if key_ht not in self.dv_dict.items:
                return VariableTracker.build(tx, False)
            stored = self.dv_dict.items[key_ht]
            return VariableTracker.build(tx, val.is_python_equal(stored))

        return iter_contains(self.view_items_vt, item, tx)

    def richcompare_impl(
        self, tx: "InstructionTranslatorBase", other: VariableTracker, op: str
    ) -> VariableTracker:
        # dictview_richcompare: accepts set/frozenset and dict_keys/dict_items.
        # https://github.com/python/cpython/blob/e76aa128fe/Objects/dictobject.c#L5952-L6010
        # Uses a polyfill with len() and ``in`` so that Dynamo traces through
        # sq_length/sq_contains, matching CPython's use of PyObject_Size and
        # PySequence_Contains.
        from .builder import SourcelessBuilder

        if not _is_set_or_dictview(other):
            return ConstantVariable.create(NotImplemented)
        return SourcelessBuilder.create(
            tx, polyfills.dictview_richcompare
        ).call_function(
            tx,
            [
                SourcelessBuilder.create(tx, cmp_name_to_op_mapping[op]),
                self,
                other,
            ],
            {},
        )

    def tp_iter_impl(self, tx: "InstructionTranslatorBase") -> VariableTracker:
        from .iter import DictItemsIterator

        if self.dv_dict.source and not is_constant_source(self.dv_dict.source):
            tx.output.guard_on_key_order.add(self.dv_dict.source)
        return DictItemsIterator(self.dv_dict.items)

    def call_method(
        self,
        tx: "InstructionTranslatorBase",
        name: str,
        args: list[VariableTracker],
        kwargs: dict[str, VariableTracker],
    ) -> VariableTracker:
        if name in (
            "__and__",
            "__iand__",
            "__xor__",
            "__ixor__",
        ):
            # These methods always returns a set
            fn_hdl = getattr(self.set_items, name)
            ret_val = fn_hdl(args[0].set_items)  # type: ignore[attr-defined]
            return SetVariable(ret_val)
        return super().call_method(tx, name, args, kwargs)


kV = HashableTracker | str


class SideEffectsProxyDict(collections.abc.MutableMapping[kV, VariableTracker]):
    """
    A proxy dict that allows us to track mutations to the dict using side
    effects table as storage.
    """

    @staticmethod
    def get_example_value_dict(vt: VariableTracker) -> dict[str, object]:
        if istype(vt, variables.NestedUserFunctionVariable):
            # NestedUserFunctionVariable is created with MAKE_FUNCTION and its
            # __dict__ starts empty. Any mutation will actually be recorded in
            # the side effects table.
            return {}
        elif isinstance(vt, variables.LocalGeneratorFunctionVariable):
            return SideEffectsProxyDict.get_example_value_dict(vt.vt)
        else:
            value = vt.get_real_python_backed_value()
            if value is not NO_SUCH_SUBOBJ:
                if isinstance(vt, variables.UserDefinedObjectVariable):
                    return vt._getattr_static("__dict__")  # type: ignore[bad-return]
                else:
                    return object.__getattribute__(value, "__dict__")
            else:
                unimplemented(
                    gb_type="unsupported variable type for __dict__ access",
                    context=f"VariableTracker type: {type(vt)}",
                    explanation=f"Dynamo does not know how to get __dict__ from {type(vt)}",
                    hints=[
                        *graph_break_hints.DYNAMO_BUG,
                    ],
                )

    @staticmethod
    def get_value___dict__(
        tx: "InstructionTranslatorBase", vt: VariableTracker
    ) -> dict[str, VariableTracker]:
        # GENERIC_SETATTR on "__dict__" means the whole instance dict was
        # replaced through the __dict__ getset descriptor (obj.__dict__ = ...).
        # Per-key obj.__dict__[name] mutations are tracked as INSTANCE_DICT.
        if isinstance(
            vt, variables.UserDefinedObjectVariable
        ) and tx.output.side_effects.has_pending_mutation_of_attr(
            vt, "__dict__", AttrMutationKind.GENERIC_SETATTR
        ):
            dict_vt = tx.output.side_effects.load_attr(vt, "__dict__")
            if isinstance(dict_vt, ConstDictVariable):
                result = {}
                for key, value in dict_vt.items.items():
                    try:
                        key_value = key.vt.as_python_constant()
                    except NotImplementedError:
                        unimplemented(
                            gb_type="non-constant key in object __dict__",
                            context=f"key={key.vt}",
                            explanation="Dynamo expects object __dict__ replacement keys to be constants.",
                            hints=[*graph_break_hints.SUPPORTABLE],
                        )
                    if not isinstance(key_value, str):
                        unimplemented(
                            gb_type="non-string key in object __dict__",
                            context=f"key={key_value!r}",
                            explanation="Dynamo expects object __dict__ keys to be strings.",
                            hints=[*graph_break_hints.USER_ERROR],
                        )
                    result[key_value] = value
                return result
            # Other replacement forms are not materialized here yet. Leave
            # them on the normal example-value path until Dynamo models their
            # contents explicitly.

        example_value_dict = SideEffectsProxyDict.get_example_value_dict(vt)

        return {
            key: VariableTracker.build(
                tx,
                value,
                source=vt.source
                and DictGetItemSource(AttrSource(vt.source, "__dict__"), key),
            )
            for key, value in example_value_dict.items()
        }

    def __init__(self, item: VariableTracker, tx: "InstructionTranslatorBase") -> None:
        self.item = item
        self.side_effects = tx.output.side_effects
        self.item_dict = self.get_value___dict__(tx, item)

    def _maybe_unwrap_key(self, key: kV) -> str:
        Hasher = HashableTracker
        return key.vt.as_python_constant() if istype(key, Hasher) else key

    def side_effects_table(self) -> dict[str, VariableTracker]:
        return {
            name: value
            for name, value in self.side_effects.store_attr_mutations.get(
                self.item, {}
            ).items()
            if self.side_effects.has_pending_mutation_of_attr(
                self.item, name, AttrMutationKind.INSTANCE_DICT
            )
        }

    def __getitem__(self, key: kV) -> VariableTracker:
        name = self._maybe_unwrap_key(key)
        if self.side_effects.has_pending_mutation_of_attr(
            self.item, name, AttrMutationKind.INSTANCE_DICT
        ):
            return self.side_effects.load_attr(self.item, name, deleted_ok=True)
        return self.item_dict[name]

    def __setitem__(self, key: kV, value: VariableTracker) -> None:
        # Find a way to not hash the key using HashableTracker
        name = self._maybe_unwrap_key(key)
        if not istype(name, str):
            raise AssertionError(f"Expected str key, got {type(name)}")
        self.side_effects.store_instance_dict_attr(self.item, name, value)

    def __delitem__(self, key: kV) -> None:
        name = self._maybe_unwrap_key(key)
        self.side_effects.store_instance_dict_attr(
            self.item, name, variables.DeletedVariable()
        )

    def __contains__(self, key: kV) -> bool:  # type: ignore[bad-override]
        name = self._maybe_unwrap_key(key)
        if self.side_effects.has_pending_mutation_of_attr(
            self.item, name, AttrMutationKind.INSTANCE_DICT
        ):
            value = self.side_effects.load_attr(self.item, name, deleted_ok=True)
            return not isinstance(value, variables.DeletedVariable)
        return name in self.item_dict

    def __len__(self) -> int:
        return sum(1 for _ in self)

    def __iter__(self) -> Iterator[HashableTracker]:
        Hasher = HashableTracker
        d = self.side_effects_table()
        for k, v in d.items():
            if isinstance(v, variables.DeletedVariable):
                continue
            yield Hasher(ConstantVariable.create(k))

        for k, v in self.item_dict.items():
            if k not in d:
                yield Hasher(ConstantVariable.create(k))


class DunderDictVariable(ConstDictVariable):
    """represents object.__dict__"""

    @classmethod
    def create(
        cls,
        tx: "InstructionTranslatorBase",
        vt: VariableTracker,
    ) -> "DunderDictVariable":
        mutation = AttributeMutationExisting() if vt.source else AttributeMutationNew()
        source = vt.source and AttrSource(vt.source, "__dict__")

        return cls(
            vt,
            tx=tx,
            mutation_type=mutation,
            source=source,
        )

    def __init__(
        self,
        vt: VariableTracker,
        tx: "InstructionTranslatorBase",
        **kwargs: Any,
    ) -> None:
        super().__init__({}, **kwargs)
        self.items = SideEffectsProxyDict(vt, tx)

    def setitem(self, name: str, value: VariableTracker) -> None:
        self.items[name] = value

    def delitem(self, name: str) -> None:
        if not self.contains(name):
            raise KeyError(name)
        del self.items[name]

    def getitem(self, name: str) -> VariableTracker:
        return self.items[name]

    def contains(self, name: str) -> bool:
        return name in self.items

    def getitem_or_default(
        self,
        name: str,
        default: Callable[[], VariableTracker],
    ) -> VariableTracker:
        if self.contains(name):
            return self.getitem(name)
        else:
            value = default()
            self.items[name] = value
            return value

    def call_method(
        self,
        tx: "InstructionTranslatorBase",
        name: str,
        args: list[VariableTracker],
        kwargs: dict[str, VariableTracker],
    ) -> VariableTracker:
        if name == "copy":
            if args or kwargs:
                raise_args_mismatch(
                    tx,
                    name,
                    "0 args and 0 kwargs",
                    f"{len(args)} args and {len(kwargs)} kwargs",
                )
            return ConstDictVariable(
                dict(self.items), mutation_type=ValueMutationNew(), source=None
            )
        return super().call_method(tx, name, args, kwargs)

    # Mutations to __dict__ are tracked through side effects (SideEffectsProxyDict),
    # so we don't need to install guards. Guard installation is overridden to no-op.
    def install_dict_keys_match_guard(self) -> None:
        pass

    def install_dict_contains_guard(
        self, tx: "InstructionTranslatorBase", args: list[VariableTracker]
    ) -> None:
        pass