Diff Coverage

        module_name = f"{_DISTRIBUTED_OPS_PACKAGE}.{module_file}"
        spec = importlib.util.find_spec(module_name)
        if spec is None or spec.origin is None:
            raise ImportError(f"Distributed op module cannot be resolved: {module_name}")

        module_origin = os.path.realpath(spec.origin)
        try:
            common_path = os.path.commonpath((_DISTRIBUTED_OPS_DIR, module_origin))
        except ValueError as exc:
            raise ImportError(f"Distributed op module has an untrusted source: {module_name}") from exc
        if common_path != _DISTRIBUTED_OPS_DIR:
            raise ImportError(f"Distributed op module has an untrusted source: {module_name}")
        return module_name

            raise ValueError("distributed_op_module is not supported; use distributed_op_file")

        class_name = config["distributed_op_class"]
        if not isinstance(class_name, str) or not class_name.isidentifier():
            raise ValueError(f"Invalid distributed op class name: {class_name!r}")

        module_file = config["distributed_op_file"]
        module_name = self._resolve_distributed_op_module(module_file)
        module = importlib.import_module(module_name)

    @staticmethod
    def forward(ctx: Any, *args: Any, **kwargs: Any) -> Any:
        """Override in subclass to define the forward computation on local tensors."""
        raise NotImplementedError("Subclasses must implement forward()")

    @staticmethod
    def backward(ctx: Any, *grad_outputs: Any) -> Any:
        """Override in subclass to define the backward computation on local tensors."""
        raise NotImplementedError("Subclasses must implement backward()")

    @classmethod
    def apply(cls, *args: Any, **kwargs: Any) -> Any:
        """Execute the function, routing to distributed dispatch when DTensor inputs are present.

                raise ValueError(
                    f"{cls.__name__} received DTensor inputs but '_op_name' is not set. "
                    "Set '_op_name' on the subclass and register a matching DistributedOp."
                )
            return _OP_DISPATCHER.dispatch(cls._get_local_callable(), args, kwargs)
        return super().apply(*args, **kwargs)

    @classmethod
    def _get_local_callable(cls) -> _LocalCallable:

        if '_local_callable' not in cls.__dict__:
            def _local_fn(*a, **kw):
                # Called by _OP_DISPATCHER with extracted local (non-DTensor) tensors.
                # has_dtensor will be False here, so super().apply() is taken directly.
                return cls.apply(*a, **kw)

            cls._local_callable = _LocalCallable(_local_fn, cls._op_name)
        return cls._local_callable