Diff Coverage

        ``layout.placements`` (a plain attribute) is passed only to satisfy the
        constructor's non-None check; ``__init_data__`` ignores it when ``layout``
        is supplied.
        """
        return DTensor(local_tensor, layout.mesh, layout.placements, layout)

    def _alias_placements(self) -> Sequence[Placement]:
        """Return alias_placements from layout, falling back to _placements."""
        if hasattr(self, '_layout') and self._layout:

        if has_dtensor:
            self._check_ce_op_without_loss_parallel_context(op_name, args)

            if not is_loss_parallel_op(op_name):
                raise RuntimeError(
                    f"Operator {op_name} does not contain parallel layout infer func. "
                    f"DTensor dispatch requires explicit layout inference registration. "
                    f"Please register a distributed operator for '{op_name}' or use local tensors."
                )

                logits = gathered_args[0]
                targets = gathered_args[1]
                if isinstance(logits, Tensor) and isinstance(targets, Tensor):
                    if logits.ndim > 2 and targets.ndim > 1 and targets.ndim == logits.ndim - 1:
                        vocab_size = logits.shape[-1]
                        gathered_args[0] = logits.reshape(-1, vocab_size)
                        gathered_args[1] = targets.reshape(-1)

            return op_call(*gathered_args, **gathered_kwargs)
        raise RuntimeError(f"Operator {op_name} does not contain parallel layout infer func.")

        # of (deprecation-bound) unpack_ops ever use the packed format, so gate the
        # whole normalization behind an O(1) membership test instead of paying two
        # function frames (_normalize_aclop_args + _is_aclop_packed) on every op.
        if op_name in getattr(self, 'unpack_ops', ()):
            packed_call, args = self._normalize_aclop_args(op_name, self.unpack_ops, args)
        else:
            packed_call = None

        result = distribute_op.preprocess(args, kwargs)

        # is_loss_parallel_active() contextvar reads — on every subsequent call.
        if op_name not in getattr(self, '_non_loss_ops', ()):
            self._check_decomposed_ce_op_in_loss_parallel(op_name, args, kwargs)
            if self._should_dispatch_loss_parallel(op_name):
                return self._dispatch_loss_parallel(op_call, args, kwargs)
            if not is_loss_parallel_op(op_name) and not is_decomposed_ce_op(op_name):
                if not hasattr(self, '_non_loss_ops'):
                    self._non_loss_ops = set()
                self._non_loss_ops.add(op_name)

        # ABCMeta isinstance(local_tensor, DTensorBase) check (MindSpore Tensor's
        # metaclass is ABCMeta, so that isinstance is ~10x a normal one), the three
        # None guards, and the device-placement guard — all pure per-output overhead.
        if layout is not None:
            t = Tensor._make_subclass(cls, local_tensor)
            t.__init_data__(local_tensor, device_mesh, placements, layout)
            return t

        npu_device = "Ascend"
        if isinstance(local_tensor, DTensorBase):
            src = local_tensor