diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 062131170..1aae2d006 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -5,13 +5,13 @@ on:
   push:
     branches:
       - master
+      - "gsoc26-*"
   pull_request:
     branches:
       - master
       - "1.1"
       - "1.2"
-      - "gsoc26-x509-certificate-generator-templates"
-      - "gsoc26-mass-commands"
+      - "gsoc26-*"
 
 jobs:
   build:
@@ -100,7 +100,7 @@ jobs:
         with:
           parallel: true
           format: cobertura
-          flag-name: python-${{ matrix.env.env }}
+          flag-name: python-${{ matrix.python-version }}-${{ matrix.django-version }}
           github-token: ${{ secrets.GITHUB_TOKEN }}
           fail-on-error: false
 
diff --git a/docs/user/rest-api.rst b/docs/user/rest-api.rst
index c9702920d..a023c957b 100644
--- a/docs/user/rest-api.rst
+++ b/docs/user/rest-api.rst
@@ -481,6 +481,113 @@ Get Command Details
 
     GET /api/v1/controller/device/{device_id}/command/{command_id}/
 
+.. _controller_batch_command_api:
+
+Dry-Run Batch Command
+~~~~~~~~~~~~~~~~~~~~~
+
+.. code-block:: text
+
+    GET /api/v1/controller/batch-command/execute/
+
+Returns the list of devices that would be targeted without executing
+anything. Useful for previewing which devices are affected.
+
+**Query Parameters:**
+
+================ =================================================
+Parameter        Description
+================ =================================================
+``organization`` Organization UUID (optional)
+``type``         Command type (optional for dry-run)
+``input``        Input data for the command (optional for dry-run)
+``devices``      Comma-separated device UUIDs (optional)
+``group``        Device group UUID (optional)
+``location``     Location UUID (optional)
+``execute_all``  Set to ``true`` to target all devices (optional)
+================ =================================================
+
+Execute a Batch Command
+~~~~~~~~~~~~~~~~~~~~~~~
+
+.. code-block:: text
+
+    POST /api/v1/controller/batch-command/execute/
+
+Creates and executes a batch command on the targeted devices.
+
+**Request Parameters:**
+
+================ ================================================
+Parameter        Description
+================ ================================================
+``organization`` Organization UUID (optional for superusers)
+``type``         Type of command to execute (**required**)
+``input``        Input data for the command (**required**)
+``label``        A short label to identify this batch command
+                 (**required**)
+``notes``        Optional notes (optional)
+``devices``      List of device UUIDs (optional)
+``group``        Device group UUID (optional)
+``location``     Location UUID (optional)
+``execute_all``  Set to ``true`` to target all devices (optional)
+================ ================================================
+
+**Available Command Types:**
+
+See :ref:`controller_execute_command_api` for available command types and
+input formats.
+
+**Example payload:**
+
+.. code-block:: json
+
+    {
+        "organization": "org-uuid",
+        "type": "custom",
+        "input": {"command": "uptime"},
+        "label": "Check uptime",
+        "execute_all": true
+    }
+
+**Example request:**
+
+.. code-block:: shell
+
+    curl -X POST \
+        http://127.0.0.1:8000/api/v1/controller/batch-command/execute/ \
+        -H 'authorization: Bearer yoursecretauthtoken' \
+        -H 'content-type: application/json' \
+        -d '{
+                "organization": "org-uuid",
+                "type": "custom",
+                "input": {"command": "uptime"},
+                "label": "Check uptime",
+                "execute_all": true
+            }'
+
+**Response:** ``201 Created`` with the batch command UUID.
+
+List Batch Commands
+~~~~~~~~~~~~~~~~~~~
+
+.. code-block:: text
+
+    GET /api/v1/controller/batch-command/
+
+Returns a paginated list of batch commands with device count and skipped
+device information.
+
+Get Batch Command Detail
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+.. code-block:: text
+
+    GET /api/v1/controller/batch-command/{id}/
+
+Returns detailed information about a batch command, including the list of
+targeted devices.
+
 List Device Groups
 ~~~~~~~~~~~~~~~~~~
 
diff --git a/openwisp_controller/connection/admin.py b/openwisp_controller/connection/admin.py
index f15c98d58..2409c6595 100644
--- a/openwisp_controller/connection/admin.py
+++ b/openwisp_controller/connection/admin.py
@@ -1,26 +1,31 @@
+import json
 from datetime import timedelta
 
 import reversion
 import swapper
 from django import forms
 from django.contrib import admin
+from django.core.paginator import EmptyPage, PageNotAnInteger, Paginator
 from django.http import HttpResponseForbidden, JsonResponse
 from django.urls import path, resolve
-from django.utils.html import format_html
+from django.utils.html import format_html, format_html_join
+from django.utils.safestring import mark_safe
 from django.utils.timezone import localtime
 from django.utils.translation import gettext_lazy as _
 
 from openwisp_users.multitenancy import MultitenantOrgFilter
-from openwisp_utils.admin import TimeReadonlyAdminMixin
+from openwisp_utils.admin import ReadOnlyAdmin, TimeReadonlyAdminMixin
 
 from ..admin import MultitenantAdminMixin
 from ..config.admin import DeactivatedDeviceReadOnlyMixin, DeviceAdmin
+from .filters import GroupFilter, LocationFilter
 from .schema import schema
 from .widgets import CommandSchemaWidget, CredentialsSchemaWidget
 
 Credentials = swapper.load_model("connection", "Credentials")
 DeviceConnection = swapper.load_model("connection", "DeviceConnection")
 Command = swapper.load_model("connection", "Command")
+BatchCommand = swapper.load_model("connection", "BatchCommand")
 
 
 class CredentialsForm(forms.ModelForm):
@@ -215,3 +220,244 @@ def schema_view(self, request):
     CommandInline,
 ]
 DeviceAdmin.add_reversion_following(follow=["deviceconnection_set"])
+
+
+class BatchCommandAdmin(MultitenantAdminMixin, ReadOnlyAdmin):
+    list_display = [
+        "label",
+        "organization_display",
+        "colored_status",
+        "type",
+        "affected_devices",
+        "created",
+    ]
+    ordering = ("-created",)
+    list_filter = [
+        MultitenantOrgFilter,
+        "status",
+        "type",
+        GroupFilter,
+        LocationFilter,
+    ]
+    list_select_related = ("organization",)
+    search_fields = [
+        "label",
+        "notes",
+        "organization__name",
+        "devices__name",
+        "location__name",
+        "group__name",
+    ]
+    change_form_template = (
+        "admin/connection/batch_command/batch_command_change_form.html"
+    )
+    device_commands_per_page = 20
+    exclude = ("devices",)
+    fields = [
+        "organization_display",
+        "label",
+        "notes",
+        "colored_status",
+        "type",
+        "formatted_input",
+        "affected_devices",
+        "group",
+        "location",
+        "display_skipped_devices",
+        "created",
+        "modified",
+    ]
+    readonly_fields = [
+        "organization_display",
+        "colored_status",
+        "type",
+        "formatted_input",
+        "affected_devices",
+        "display_skipped_devices",
+        "group",
+        "location",
+        "created",
+        "modified",
+    ]
+
+    class Media:
+        css = {
+            "all": [
+                "admin/css/changelists.css",
+                "admin/css/ow-filters.css",
+                "connection/css/batch-command.css",
+            ]
+        }
+
+    def get_readonly_fields(self, request, obj=None):
+        fields = super().get_readonly_fields(request, obj)
+        return fields + list(self.__class__.readonly_fields)
+
+    def _get_commands(self, request, obj):
+        qs = Command.objects.filter(batch_command=obj).select_related("device")
+        if not request.user.is_superuser:
+            qs = qs.filter(
+                device__organization_id__in=request.user.organizations_managed
+            )
+        return qs
+
+    def organization_display(self, obj):
+        if obj.organization:
+            return obj.organization.name
+        return _("All")
+
+    organization_display.short_description = _("organization")
+    organization_display.admin_order_field = "organization"
+
+    def colored_status(self, obj):
+        css_class = f"command-status {obj.status}"
+        return format_html(
+            '<span class="{0}">{1}</span>',
+            css_class,
+            obj.get_status_display(),
+        )
+
+    colored_status.short_description = _("status")
+
+    def formatted_input(self, obj):
+        if not obj.input:
+            return "-"
+        return obj.input.get("command", obj.input)
+
+    formatted_input.short_description = _("input")
+
+    def affected_devices(self, obj):
+        return obj.affected_devices
+
+    affected_devices.short_description = _("affected devices")
+
+    def display_skipped_devices(self, obj):
+        if not obj.skipped_devices:
+            return "-"
+        Device = swapper.load_model("config", "Device")
+        pks = list(obj.skipped_devices.keys())
+        devices = {str(d.pk): d for d in Device.objects.filter(pk__in=pks)}
+        count = len(pks)
+        lines = [str(count)]
+        for pk_str, errors in obj.skipped_devices.items():
+            device = devices.get(pk_str)
+            name = device.name if device else _("Deleted ({})").format(pk_str)
+            lines.append(format_html("{}: {}", name, ", ".join(errors)))
+        return format_html(
+            '<div class="skipped-devices-list">{}</div>',
+            format_html_join(mark_safe("<br>"), "{}", ((line,) for line in lines)),
+        )
+
+    display_skipped_devices.short_description = _("Skipped devices")
+
+    def _build_filter_specs(self, request, obj, current_status):
+        filter_specs = []
+        params = request.GET.copy()
+
+        def _make_choice(current_value, display, param_name, value):
+            q = params.copy()
+            q.pop(param_name, None)
+            if value:
+                q[param_name] = value
+            qs = q.urlencode()
+            query_string = f"?{qs}" if qs else ""
+            return {
+                "display": display,
+                "selected": current_value == value,
+                "query_string": query_string,
+            }
+
+        status_choices = []
+        for status_value, display_name in (
+            (("", _("All")),) + Command.STATUS_CHOICES + (("skipped", _("Skipped")),)
+        ):
+            status_choices.append(
+                _make_choice(current_status, display_name, "status", status_value)
+            )
+
+        class StatusFilter:
+            title = _("status")
+            choices = status_choices
+
+        filter_specs.append(StatusFilter())
+        return filter_specs
+
+    def _paginate_commands(self, items, page_param, per_page=None):
+        per_page = per_page or self.device_commands_per_page
+        paginator = Paginator(list(items), per_page)
+        page_number = page_param or 1
+        try:
+            page_obj = paginator.page(page_number)
+        except (PageNotAnInteger, EmptyPage):
+            page_obj = paginator.page(1)
+        return page_obj, paginator, page_obj.object_list
+
+    def change_view(self, request, object_id, form_url="", extra_context=None):
+        extra_context = extra_context or {}
+        obj = self.get_object(request, object_id)
+        if obj:
+            Device = swapper.load_model("config", "Device")
+            commands_qs = self._get_commands(request, obj)
+            search_query = request.GET.get("q", "")
+            if search_query:
+                commands_qs = commands_qs.filter(device__name__icontains=search_query)
+            current_status = request.GET.get("status", "")
+            if current_status and current_status != "skipped":
+                commands_qs = commands_qs.filter(status=current_status)
+            rows = []
+            for cmd in commands_qs:
+                rows.append(
+                    {
+                        "device_name": cmd.device.name,
+                        "device_pk": cmd.device.pk,
+                        "status": cmd.status,
+                        "status_display": cmd.get_status_display(),
+                        "output": (cmd.output or "").lstrip(),
+                        "created": cmd.created,
+                        "is_skipped": False,
+                    }
+                )
+            if obj.skipped_devices and current_status in ("", "skipped"):
+                pks = list(obj.skipped_devices.keys())
+                devices = {str(d.pk): d for d in Device.objects.filter(pk__in=pks)}
+                for pk_str, errors in obj.skipped_devices.items():
+                    device = devices.get(pk_str)
+                    name = device.name if device else _("Deleted ({})").format(pk_str)
+                    if search_query and search_query.lower() not in name.lower():
+                        continue
+                    rows.append(
+                        {
+                            "device_name": name,
+                            "device_pk": pk_str,
+                            "status": "skipped",
+                            "status_display": _("Skipped"),
+                            "output": ", ".join(errors),
+                            "created": None,
+                            "is_skipped": True,
+                        }
+                    )
+
+            def _sort_key(row):
+                priority = {"success": 0, "failed": 1, "skipped": 2}
+                return (priority.get(row["status"], 99), row["device_name"].lower())
+
+            rows.sort(key=_sort_key)
+            filter_specs = self._build_filter_specs(request, obj, current_status)
+            page_obj, paginator, commands = self._paginate_commands(
+                rows, request.GET.get("page", 1)
+            )
+            extra_context.update(
+                {
+                    "commands": commands,
+                    "page_obj": page_obj,
+                    "paginator": paginator,
+                    "filter_specs": filter_specs,
+                    "has_active_filters": any(
+                        request.GET.get(param) for param in ["status"]
+                    ),
+                }
+            )
+        return super().change_view(request, object_id, extra_context=extra_context)
+
+
+admin.site.register(BatchCommand, BatchCommandAdmin)
diff --git a/openwisp_controller/connection/api/serializers.py b/openwisp_controller/connection/api/serializers.py
index 142c1c30a..82b67ffc3 100644
--- a/openwisp_controller/connection/api/serializers.py
+++ b/openwisp_controller/connection/api/serializers.py
@@ -12,6 +12,7 @@
 DeviceConnection = load_model("connection", "DeviceConnection")
 Credentials = load_model("connection", "Credentials")
 Device = load_model("config", "Device")
+BatchCommand = load_model("connection", "BatchCommand")
 
 
 class ValidatedDeviceFieldSerializer(ValidatedModelSerializer):
@@ -43,6 +44,10 @@ class CommandSerializer(ValidatedDeviceFieldSerializer):
         required=False,
         pk_field=serializers.UUIDField(format="hex_verbose"),
     )
+    batch_command = serializers.PrimaryKeyRelatedField(
+        read_only=True,
+        pk_field=serializers.UUIDField(format="hex_verbose"),
+    )
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
@@ -115,3 +120,117 @@ class Meta:
             "is_working": {"read_only": True},
         }
         read_only_fields = ("created", "modified")
+
+
+class BatchCommandExecuteSerializer(
+    FilterSerializerByOrgManaged, serializers.ModelSerializer
+):
+    input = serializers.JSONField(allow_null=True, required=False)
+    devices = serializers.PrimaryKeyRelatedField(
+        many=True,
+        queryset=Device.objects.all(),
+        required=False,
+        allow_empty=True,
+        pk_field=serializers.UUIDField(format="hex_verbose"),
+    )
+    execute_all = serializers.BooleanField(required=False, default=True)
+
+    def __init__(self, *args, dry_run=False, **kwargs):
+        super().__init__(*args, **kwargs)
+        if dry_run:
+            self._skip_target_validation = True
+            self.fields["type"].required = False
+            self.fields["label"].required = False
+        else:
+            self._skip_target_validation = False
+
+    class Meta:
+        model = BatchCommand
+        fields = (
+            "organization",
+            "type",
+            "input",
+            "label",
+            "notes",
+            "devices",
+            "group",
+            "location",
+            "execute_all",
+        )
+        extra_kwargs = {
+            "organization": {"required": False, "allow_null": True},
+        }
+
+    def validate(self, data):
+        org = data.get("organization")
+        execute_all = data.get("execute_all", False)
+        devices = data.get("devices")
+        group = data.get("group")
+        location = data.get("location")
+        if not org and not self.context["request"].user.is_superuser:
+            raise serializers.ValidationError(
+                _("Only superusers can execute batch commands without an organization.")
+            )
+        if not self._skip_target_validation:
+            if (
+                not execute_all
+                and not org
+                and not devices
+                and not group
+                and not location
+            ):
+                raise serializers.ValidationError(
+                    _(
+                        "Specify at least one targeting option "
+                        "or set execute_all to true."
+                    )
+                )
+        if devices:
+            for device in devices:
+                if org and device.organization_id != org.id:
+                    raise serializers.ValidationError(
+                        {
+                            "devices": _(
+                                "All devices must belong to the same organization."
+                            )
+                        }
+                    )
+        return data
+
+
+class BatchCommandSerializer(BaseSerializer):
+    device_count = serializers.IntegerField(read_only=True)
+    skipped_devices = serializers.JSONField(read_only=True)
+
+    class Meta:
+        model = BatchCommand
+        fields = (
+            "id",
+            "organization",
+            "status",
+            "type",
+            "input",
+            "label",
+            "notes",
+            "group",
+            "location",
+            "device_count",
+            "skipped_devices",
+            "created",
+            "modified",
+        )
+        read_only_fields = (
+            "created",
+            "modified",
+        )
+
+
+class BatchCommandDetailSerializer(BatchCommandSerializer):
+    devices = serializers.PrimaryKeyRelatedField(
+        many=True,
+        read_only=True,
+        pk_field=serializers.UUIDField(format="hex_verbose"),
+    )
+
+    class Meta(BatchCommandSerializer.Meta):
+        fields = BatchCommandSerializer.Meta.fields + ("devices",)
diff --git a/openwisp_controller/connection/api/urls.py b/openwisp_controller/connection/api/urls.py
index 4ec3e70ab..4a94d171d 100644
--- a/openwisp_controller/connection/api/urls.py
+++ b/openwisp_controller/connection/api/urls.py
@@ -40,6 +40,21 @@ def get_api_urls(api_views):
             api_views.deviceconnection_detail_view,
             name="deviceconnection_detail",
         ),
+        path(
+            "api/v1/controller/batch-command/",
+            api_views.batch_command_list_view,
+            name="batch_command_list",
+        ),
+        path(
+            "api/v1/controller/batch-command/<uuid:pk>/",
+            api_views.batch_command_detail_view,
+            name="batch_command_detail",
+        ),
+        path(
+            "api/v1/controller/batch-command/execute/",
+            api_views.batch_command_execute_view,
+            name="batch_command_execute",
+        ),
     ]
 
 
diff --git a/openwisp_controller/connection/api/views.py b/openwisp_controller/connection/api/views.py
index 6af1270c7..41625ad68 100644
--- a/openwisp_controller/connection/api/views.py
+++ b/openwisp_controller/connection/api/views.py
@@ -1,12 +1,18 @@
+from django.core.exceptions import ValidationError
+from django.db.models import Count
 from django.utils.translation import gettext_lazy as _
 from drf_yasg import openapi
 from drf_yasg.utils import swagger_auto_schema
+from rest_framework import status
 from rest_framework.generics import (
+    GenericAPIView,
+    ListAPIView,
     ListCreateAPIView,
     RetrieveAPIView,
     RetrieveUpdateDestroyAPIView,
     get_object_or_404,
 )
+from rest_framework.response import Response
 from swapper import load_model
 
 from openwisp_utils.api.pagination import OpenWispPagination
@@ -17,6 +23,9 @@
     RelatedDeviceProtectedAPIMixin,
 )
 from .serializers import (
+    BatchCommandDetailSerializer,
+    BatchCommandExecuteSerializer,
+    BatchCommandSerializer,
     CommandSerializer,
     CredentialSerializer,
     DeviceConnectionSerializer,
@@ -26,6 +35,7 @@
 Device = load_model("config", "Device")
 Credentials = load_model("connection", "Credentials")
 DeviceConnection = load_model("connection", "DeviceConnection")
+BatchCommand = load_model("connection", "BatchCommand")
 
 
 class BaseCommandView(RelatedDeviceProtectedAPIMixin):
@@ -138,6 +148,55 @@ class DeviceConnectionListCreateView(BaseDeviceConnection, ListCreateAPIView):
 DeviceConnenctionListCreateView = DeviceConnectionListCreateView
 
 
+class BatchCommandExecuteView(ProtectedAPIMixin, GenericAPIView):
+    model = BatchCommand
+    queryset = BatchCommand.objects.all()
+    serializer_class = BatchCommandExecuteSerializer
+
+    def post(self, request):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        serializer.validated_data.pop("execute_all", None)
+        try:
+            batch = BatchCommand.execute(**serializer.validated_data)
+        except ValidationError as e:
+            return Response(
+                getattr(e, "message_dict", e.messages),
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        return Response({"batch": str(batch.pk)}, status=201)
+
+    def get(self, request):
+        serializer = self.get_serializer(
+            data=request.query_params,
+            dry_run=True,
+        )
+        serializer.is_valid(raise_exception=True)
+        serializer.validated_data.pop("execute_all", None)
+        try:
+            data = BatchCommand.dry_run(**serializer.validated_data)
+        except ValidationError as e:
+            return Response(
+                getattr(e, "message_dict", e.messages),
+                status=status.HTTP_400_BAD_REQUEST,
+            )
+        data["devices"] = [str(d.pk) for d in data["devices"]]
+        return Response(data)
+
+
+class BatchCommandListView(ProtectedAPIMixin, ListAPIView):
+    queryset = BatchCommand.objects.annotate(device_count=Count("devices")).order_by(
+        "-created"
+    )
+    serializer_class = BatchCommandSerializer
+    pagination_class = OpenWispPagination
+
+
+class BatchCommandDetailView(ProtectedAPIMixin, RetrieveAPIView):
+    queryset = BatchCommand.objects.annotate(device_count=Count("devices"))
+    serializer_class = BatchCommandDetailSerializer
+
+
 class DeviceConnectionDetailView(BaseDeviceConnection, RetrieveUpdateDestroyAPIView):
     def get_object(self):
         queryset = self.filter_queryset(self.get_queryset())
@@ -149,6 +208,9 @@ def get_object(self):
         return obj
 
 
+batch_command_execute_view = BatchCommandExecuteView.as_view()
+batch_command_list_view = BatchCommandListView.as_view()
+batch_command_detail_view = BatchCommandDetailView.as_view()
 command_list_create_view = CommandListCreateView.as_view()
 command_details_view = CommandDetailsView.as_view()
 credential_list_create_view = CredentialListCreateView.as_view()
diff --git a/openwisp_controller/connection/base/models.py b/openwisp_controller/connection/base/models.py
index 17a06f7bb..aa680c76f 100644
--- a/openwisp_controller/connection/base/models.py
+++ b/openwisp_controller/connection/base/models.py
@@ -15,6 +15,7 @@
 from swapper import get_model_name, load_model
 
 from openwisp_controller.config.base.base import BaseModel
+from openwisp_users.mixins import ValidateOrgMixin
 from openwisp_utils.base import TimeStampedEditableModel
 
 from ...base import ShareableOrgMixinUniqueName
@@ -30,7 +31,11 @@
 )
 from ..exceptions import NoWorkingDeviceConnectionError
 from ..signals import is_working_changed
-from ..tasks import auto_add_credentials_to_devices, launch_command
+from ..tasks import (
+    auto_add_credentials_to_devices,
+    launch_batch_command,
+    launch_command,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -467,6 +472,13 @@ class AbstractCommand(TimeStampedEditableModel):
         encoder=DjangoJSONEncoder,
     )
     output = models.TextField(blank=True)
+    batch_command = models.ForeignKey(
+        get_model_name("connection", "BatchCommand"),
+        on_delete=models.SET_NULL,
+        blank=True,
+        null=True,
+        related_name="batch_commands",
+    )
 
     class Meta:
         verbose_name = _("Command")
@@ -557,6 +569,8 @@ def save(self, *args, **kwargs):
         output = super().save(*args, **kwargs)
         if adding:
             self._schedule_command()
+        if self.batch_command_id and self.status != "in-progress":
+            self.batch_command.calculate_and_update_status()
         return output
 
     def _save_without_resurrecting(self):
@@ -719,3 +733,306 @@ def _enforce_not_custom(self):
                 f"arguments property is not applicable in "
                 f'command instance of type "{self.type}"'
             )
+
+
+class AbstractBatchCommand(ValidateOrgMixin, TimeStampedEditableModel):
+    STATUS_CHOICES = (
+        ("idle", _("idle")),
+        ("in-progress", _("in progress")),
+        ("success", _("success")),
+        ("failed", _("failed")),
+    )
+
+    organization = models.ForeignKey(
+        get_model_name("openwisp_users", "Organization"),
+        on_delete=models.CASCADE,
+        blank=True,
+        null=True,
+    )
+    status = models.CharField(
+        max_length=12, choices=STATUS_CHOICES, default=STATUS_CHOICES[0][0]
+    )
+    type = models.CharField(
+        max_length=16,
+        choices=(COMMAND_CHOICES if django.VERSION < (5, 0) else get_command_choices),
+    )
+    input = JSONField(blank=True, null=True, encoder=DjangoJSONEncoder)
+    group = models.ForeignKey(
+        get_model_name("config", "DeviceGroup"),
+        on_delete=models.SET_NULL,
+        blank=True,
+        null=True,
+        verbose_name=_("device group"),
+    )
+    location = models.ForeignKey(
+        get_model_name("geo", "Location"),
+        on_delete=models.SET_NULL,
+        blank=True,
+        null=True,
+        verbose_name=_("location"),
+    )
+    label = models.CharField(
+        max_length=64,
+        verbose_name=_("label"),
+        help_text=_("A short label to identify this batch command."),
+    )
+    notes = models.TextField(
+        blank=True,
+        default="",
+        verbose_name=_("notes"),
+        help_text=_("Optional notes about this batch command."),
+    )
+    devices = models.ManyToManyField(
+        get_model_name("config", "Device"),
+        blank=True,
+        verbose_name=_("devices"),
+    )
+    skipped_devices = models.JSONField(
+        blank=True,
+        null=True,
+        default=dict,
+        verbose_name=_("Skipped devices"),
+        help_text=_(
+            "Maps device UUIDs to validation error messages for devices "
+            "that were skipped during command creation."
+        ),
+    )
+
+    class Meta:
+        abstract = True
+        verbose_name = _("Batch command")
+        verbose_name_plural = _("Batch commands")
+
+    def __str__(self):
+        return self.label
+
+    @cached_property
+    def total_devices(self):
+        return self.batch_commands.count() + len(self.skipped_devices or {})
+
+    @cached_property
+    def affected_devices(self):
+        return self.batch_commands.count()
+
+    @property
+    def successful(self):
+        return self.batch_commands.filter(status="success").count()
+
+    @property
+    def failed(self):
+        return self.batch_commands.filter(status="failed").count()
+
+    @staticmethod
+    def _validate_device_org(device, organization_id):
+        if organization_id and device.organization_id != organization_id:
+            raise ValidationError(
+                {
+                    "devices": _(
+                        "All devices must belong to the same "
+                        "organization as the batch command."
+                    )
+                }
+            )
+
+    def _validate_org_relations(self):
+        if not self.organization_id:
+            return
+        self._validate_org_relation("group", field_error="group")
+        self._validate_org_relation("location", field_error="location")
+        if self.pk and self.devices.exists():
+            org_mismatch = self.devices.exclude(organization=self.organization).exists()
+            if org_mismatch:
+                raise ValidationError(
+                    {
+                        "devices": _(
+                            "All devices must belong to the same "
+                            "organization as the batch command."
+                        )
+                    }
+                )
+
+    def clean(self):
+        super().clean()
+        self._validate_org_relations()
+        Command = load_model("connection", "Command")
+        allowed = dict(
+            Command.get_org_allowed_commands(organization_id=self.organization_id)
+        )
+        if self.type not in allowed:
+            raise ValidationError(
+                {
+                    "type": _(
+                        '"{command}" command is not available for this organization'
+                    ).format(command=self.type)
+                }
+            )
+        try:
+            jsonschema.Draft4Validator(get_command_schema(self.type)).validate(
+                self.input
+            )
+        except SchemaError as e:
+            raise ValidationError({"input": e.message})
+
+    def resolve_devices(self):
+        """
+        Returns an iterator of devices targeted by this batch command,
+        resolved from explicit M2M devices or filtered by organization,
+        group, and location. Returns an empty iterator if no devices match.
+        """
+        if self.pk and self.devices.exists():
+            return self.devices.iterator()
+        Device = load_model("config", "Device")
+        qs = Device.objects.all()
+        if self.organization_id:
+            qs = qs.filter(organization=self.organization)
+        if self.group:
+            qs = qs.filter(group=self.group)
+        if self.location:
+            qs = qs.filter(devicelocation__location=self.location)
+        return qs.iterator()
+
+    @classmethod
+    def execute(cls, **kwargs):
+        """
+        Creates, validates, and persists the batch command, then schedules
+        execution via a background task. Raises ValidationError and deletes
+        the batch if no devices match the criteria.
+        """
+        devices_list = kwargs.pop("devices", None)
+        batch = cls(**kwargs)
+        with transaction.atomic():
+            batch.full_clean()
+            batch.save()
+            if devices_list:
+                batch.devices.set(devices_list)
+                batch._validate_org_relations()
+                if not batch.devices.exists():
+                    raise ValidationError(
+                        _("No devices match the specified criteria."),
+                    )
+            elif not any(batch.resolve_devices()):
+                raise ValidationError(
+                    _("No devices match the specified criteria."),
+                )
+            batch.status = "in-progress"
+            batch.save(update_fields=["status"])
+        transaction.on_commit(lambda: launch_batch_command.delay(batch.pk))
+        return batch
+
+    @classmethod
+    def dry_run(cls, **kwargs):
+        """
+        Returns the devices that would be targeted by this batch command
+        without executing it. Skips full validation when command type is
+        not provided case for GET request.
+        """
+        devices_list = kwargs.pop("devices", None)
+        cmd_type = kwargs.pop("type", None)
+        kwargs.pop("label", None)
+        kwargs.pop("notes", None)
+        batch = cls(**kwargs)
+        if cmd_type:
+            batch.type = cmd_type
+            if not batch.label:
+                batch.label = "dry-run"
+            batch.full_clean()
+        else:
+            batch._validate_org_relations()
+        if devices_list:
+            for device in devices_list:
+                cls._validate_device_org(device, batch.organization_id)
+            return {"devices": list(devices_list)}
+        return {"devices": list(batch.resolve_devices())}
+
+    def create_commands(self):
+        """
+        Creates individual Command instances for each device targeted by
+        this batch command. Returns early if commands already exist
+        (idempotent guard). Devices that fail validation are recorded
+        in skipped_devices and logged.
+        """
+        if self.batch_commands.exists():
+            return
+        Command = load_model("connection", "Command")
+        self.skipped_devices = {}
+        self.status = "in-progress"
+        self.save()
+        for device in self.resolve_devices():
+            command = Command(
+                device=device,
+                type=self.type,
+                input=self.input,
+                batch_command=self,
+            )
+            try:
+                # Validate before the atomic block so errors like
+                # ValidationError don't create/rollback
+                command.full_clean()
+                with transaction.atomic():
+                    command.save()
+            except Exception as e:
+                self.skipped_devices[str(device.pk)] = (
+                    e.messages if hasattr(e, "messages") else [str(e)]
+                )
+                logger.warning(
+                    "Skipping device %s for batch %s: %s",
+                    device.pk,
+                    self.pk,
+                    e,
+                )
+        if self.skipped_devices:
+            self.save(update_fields=["skipped_devices"])
+        self.calculate_and_update_status()
+
+    def calculate_and_update_status(self):
+        """
+        Calculate batch status based on individual command statuses and update if
+        changed.
+        - No commands exist: status set to "idle".
+        - Commands still running: status set to "in-progress".
+        - Some commands failed: status set to "failed".
+        - All commands completed successfully: status set to "success".
+        - Status unchanged: no database write performed.
+        """
+        stats = self.batch_commands.aggregate(
+            total_operations=models.Count("id"),
+            in_progress=models.Count(
+                models.Case(
+                    models.When(status="in-progress", then=1),
+                    output_field=models.IntegerField(),
+                )
+            ),
+            completed=models.Count(
+                models.Case(
+                    models.When(~models.Q(status="in-progress"), then=1),
+                    output_field=models.IntegerField(),
+                )
+            ),
+            successful=models.Count(
+                models.Case(
+                    models.When(status="success", then=1),
+                    output_field=models.IntegerField(),
+                )
+            ),
+            failed=models.Count(
+                models.Case(
+                    models.When(status="failed", then=1),
+                    output_field=models.IntegerField(),
+                )
+            ),
+        )
+        if stats["total_operations"] == 0:
+            new_status = "idle"
+        elif stats["in_progress"] > 0:
+            new_status = "in-progress"
+        elif stats["failed"] > 0:
+            new_status = "failed"
+        elif (
+            stats["successful"] > 0 and stats["completed"] == stats["total_operations"]
+        ):
+            new_status = "success"
+        else:
+            new_status = self.status
+        if self.status != new_status:
+            self.status = new_status
+            self.save(update_fields=["status"])
diff --git a/openwisp_controller/connection/filters.py b/openwisp_controller/connection/filters.py
new file mode 100644
index 000000000..b0623c387
--- /dev/null
+++ b/openwisp_controller/connection/filters.py
@@ -0,0 +1,15 @@
+from django.utils.translation import gettext_lazy as _
+
+from openwisp_users.multitenancy import MultitenantRelatedOrgFilter
+
+
+class GroupFilter(MultitenantRelatedOrgFilter):
+    field_name = "group"
+    parameter_name = "group_id"
+    title = _("group")
+
+
+class LocationFilter(MultitenantRelatedOrgFilter):
+    field_name = "location"
+    parameter_name = "location_id"
+    title = _("location")
diff --git a/openwisp_controller/connection/migrations/0011_batchcommand_command_batch_command.py b/openwisp_controller/connection/migrations/0011_batchcommand_command_batch_command.py
new file mode 100644
index 000000000..f45dbf74a
--- /dev/null
+++ b/openwisp_controller/connection/migrations/0011_batchcommand_command_batch_command.py
@@ -0,0 +1,181 @@
+# Generated by Django 5.2.15 on 2026-06-23 13:40
+
+import uuid
+
+import django
+import django.core.serializers.json
+import django.db.migrations.operations.special
+import django.db.models.deletion
+import django.utils.timezone
+import model_utils.fields
+from django.conf import settings
+from django.db import migrations, models
+
+import openwisp_controller.connection.commands
+
+from . import assign_batchcommand_permissions_to_groups
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("connection", "0010_replace_jsonfield_with_django_builtin"),
+        ("openwisp_users", "0024_apikey"),
+        migrations.swappable_dependency(settings.CONFIG_DEVICEGROUP_MODEL),
+        migrations.swappable_dependency(settings.CONFIG_DEVICE_MODEL),
+        migrations.swappable_dependency(settings.GEO_LOCATION_MODEL),
+        ("config", "0036_device_group"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="BatchCommand",
+            fields=[
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                (
+                    "created",
+                    model_utils.fields.AutoCreatedField(
+                        default=django.utils.timezone.now,
+                        editable=False,
+                        verbose_name="created",
+                    ),
+                ),
+                (
+                    "modified",
+                    model_utils.fields.AutoLastModifiedField(
+                        default=django.utils.timezone.now,
+                        editable=False,
+                        verbose_name="modified",
+                    ),
+                ),
+                (
+                    "status",
+                    models.CharField(
+                        choices=[
+                            ("idle", "idle"),
+                            ("in-progress", "in progress"),
+                            ("success", "success"),
+                            ("failed", "failed"),
+                        ],
+                        default="idle",
+                        max_length=12,
+                    ),
+                ),
+                (
+                    "type",
+                    models.CharField(
+                        choices=(
+                            openwisp_controller.connection.commands.COMMAND_CHOICES
+                            if django.VERSION < (5, 0)
+                            else openwisp_controller.connection.commands.get_command_choices  # noqa: E501
+                        ),
+                        max_length=16,
+                    ),
+                ),
+                (
+                    "input",
+                    models.JSONField(
+                        blank=True,
+                        encoder=django.core.serializers.json.DjangoJSONEncoder,
+                        null=True,
+                    ),
+                ),
+                (
+                    "skipped_devices",
+                    models.JSONField(
+                        blank=True,
+                        default=dict,
+                        help_text=(
+                            "Maps device UUIDs to validation error messages for "
+                            "devices that were skipped during command creation."
+                        ),
+                        null=True,
+                        verbose_name="Skipped devices",
+                    ),
+                ),
+                (
+                    "devices",
+                    models.ManyToManyField(
+                        blank=True,
+                        to=settings.CONFIG_DEVICE_MODEL,
+                        verbose_name="devices",
+                    ),
+                ),
+                (
+                    "group",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        to=settings.CONFIG_DEVICEGROUP_MODEL,
+                        verbose_name="device group",
+                    ),
+                ),
+                (
+                    "location",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        to=settings.GEO_LOCATION_MODEL,
+                        verbose_name="location",
+                    ),
+                ),
+                (
+                    "label",
+                    models.CharField(
+                        help_text=("A short label to identify this batch command."),
+                        max_length=64,
+                        verbose_name="label",
+                    ),
+                ),
+                (
+                    "notes",
+                    models.TextField(
+                        blank=True,
+                        default="",
+                        verbose_name="notes",
+                        help_text="Optional notes about this batch command.",
+                    ),
+                ),
+                (
+                    "organization",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to="openwisp_users.organization",
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "Batch command",
+                "verbose_name_plural": "Batch commands",
+                "abstract": False,
+                "swappable": "CONNECTION_BATCHCOMMAND_MODEL",
+            },
+        ),
+        migrations.AddField(
+            model_name="command",
+            name="batch_command",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="batch_commands",
+                to=settings.CONNECTION_BATCHCOMMAND_MODEL,
+            ),
+        ),
+        migrations.RunPython(
+            code=assign_batchcommand_permissions_to_groups,
+            reverse_code=django.db.migrations.operations.special.RunPython.noop,
+        ),
+    ]
diff --git a/openwisp_controller/connection/migrations/__init__.py b/openwisp_controller/connection/migrations/__init__.py
index 5b5853d2b..67a2c20e7 100644
--- a/openwisp_controller/connection/migrations/__init__.py
+++ b/openwisp_controller/connection/migrations/__init__.py
@@ -60,3 +60,30 @@ def assign_command_permissions_to_groups(apps, schema_editor):
         admin.permissions.add(
             Permission.objects.get(codename="{}_{}".format(operation, "command")).pk
         )
+
+
+def assign_batchcommand_permissions_to_groups(apps, schema_editor):
+    create_default_permissions(apps, schema_editor)
+    admin_operations = ["add", "change", "delete", "view"]
+    operator_operations = ["add", "view"]
+    Group = get_swapped_model(apps, "openwisp_users", "Group")
+
+    try:
+        admin = Group.objects.get(name="Administrator")
+        operator = Group.objects.get(name="Operator")
+    except Group.DoesNotExist:
+        return
+
+    for operation in operator_operations:
+        permission = Permission.objects.get(
+            codename="{}_{}".format(operation, "batchcommand")
+        )
+        admin.permissions.add(permission.pk)
+        operator.permissions.add(permission.pk)
+
+    for operation in admin_operations:
+        admin.permissions.add(
+            Permission.objects.get(
+                codename="{}_{}".format(operation, "batchcommand")
+            ).pk
+        )
diff --git a/openwisp_controller/connection/models.py b/openwisp_controller/connection/models.py
index 39d485b57..533ad6c4d 100644
--- a/openwisp_controller/connection/models.py
+++ b/openwisp_controller/connection/models.py
@@ -1,6 +1,11 @@
 import swapper
 
-from .base.models import AbstractCommand, AbstractCredentials, AbstractDeviceConnection
+from .base.models import (
+    AbstractBatchCommand,
+    AbstractCommand,
+    AbstractCredentials,
+    AbstractDeviceConnection,
+)
 
 
 class Credentials(AbstractCredentials):
@@ -19,3 +24,9 @@ class Command(AbstractCommand):
     class Meta(AbstractCommand.Meta):
         abstract = False
         swappable = swapper.swappable_setting("connection", "Command")
+
+
+class BatchCommand(AbstractBatchCommand):
+    class Meta(AbstractBatchCommand.Meta):
+        abstract = False
+        swappable = swapper.swappable_setting("connection", "BatchCommand")
diff --git a/openwisp_controller/connection/static/connection/css/batch-command.css b/openwisp_controller/connection/static/connection/css/batch-command.css
new file mode 100644
index 000000000..cff9510f1
--- /dev/null
+++ b/openwisp_controller/connection/static/connection/css/batch-command.css
@@ -0,0 +1,144 @@
+#batchcommand_form .submit-row {
+  display: none;
+}
+
+.commands-title {
+  font-size: 22px;
+  font-weight: 300;
+  margin: 0;
+  padding: 0;
+}
+
+.search-section {
+  padding: 20px;
+}
+
+.search-form {
+  display: flex;
+  align-items: center;
+}
+
+#main #content .search-icon {
+  width: 20px;
+  height: 20px;
+  margin-right: 15px;
+  font-size: 16px;
+  margin-top: -3px;
+}
+
+#main #content .search-input {
+  padding: 10px 15px;
+}
+
+#main #content .search-button {
+  padding: 10px 20px;
+  margin-left: 15px;
+}
+
+.filter-clear-link:hover {
+  color: var(--ow-color-fg-darker);
+}
+
+.results-table {
+  width: 100%;
+  border-collapse: collapse;
+}
+
+#main #content .device-link {
+  color: var(--ow-color-primary);
+  font-weight: bold;
+}
+
+.device-name-disabled {
+  color: var(--body-quiet-color);
+  font-style: italic;
+}
+
+.empty-results {
+  padding: 40px;
+  text-align: center;
+  color: var(--body-quiet-color);
+  font-style: italic;
+}
+
+.pagination {
+  padding: 15px 20px;
+  text-align: right;
+  border-top: 2px solid var(--hairline-color);
+  background: var(--darkened-bg);
+}
+
+.pagination a {
+  color: var(--body-fg);
+  text-decoration: none;
+  margin: 0 5px;
+}
+
+.pagination .current-page {
+  margin: 0 10px;
+  color: var(--body-quiet-color);
+}
+
+.paginator {
+  color: var(--body-quiet-color);
+  padding: 10px 20px;
+  border-bottom: 1px solid var(--hairline-color);
+  margin: 0;
+}
+
+.command-status {
+  font-weight: bold;
+}
+
+.command-status.success {
+  color: var(--ow-color-success);
+}
+
+.command-status.failed {
+  color: var(--error-fg);
+}
+
+.command-status.in-progress {
+  color: var(--body-quiet-color);
+}
+
+.command-status.skipped {
+  color: var(--body-quiet-color);
+  opacity: 0.7;
+}
+
+.command-output {
+  padding: 0;
+}
+
+.command-output pre {
+  white-space: pre-wrap;
+  word-wrap: break-word;
+  margin: 0;
+  padding: 0;
+  font: inherit;
+  color: inherit;
+  background: transparent;
+}
+
+.skipped-devices-list {
+  line-height: 1.7;
+}
+
+.field-display_skipped_devices .readonly.readonly {
+  padding: 0;
+}
+
+/* Adjustments for list filters */
+#main #content .left-arrow {
+  left: -1.125rem;
+}
+#main #content .right-arrow {
+  right: -1.125rem;
+}
+#main #content #ow-changelist-filter {
+  padding: 1.25rem 0rem;
+}
+#main #content .filters-top {
+  margin-bottom: 0.5rem;
+}
diff --git a/openwisp_controller/connection/tasks.py b/openwisp_controller/connection/tasks.py
index d6ea2828b..22574e6ed 100644
--- a/openwisp_controller/connection/tasks.py
+++ b/openwisp_controller/connection/tasks.py
@@ -98,6 +98,24 @@ def launch_command(command_id):
         command._save_without_resurrecting()
 
 
+@shared_task(bind=True)
+def launch_batch_command(self, batch_id):
+    BatchCommand = load_model("connection", "BatchCommand")
+    try:
+        batch = BatchCommand.objects.get(pk=batch_id)
+    except BatchCommand.DoesNotExist:
+        logger.warning(f"The BatchCommand object with id {batch_id} has been deleted")
+        return
+    try:
+        batch.create_commands()
+    except Exception as e:
+        batch.status = "failed"
+        batch.save(update_fields=["status"])
+        logger.exception(
+            f"An exception was raised while executing batch " f"command {batch_id}: {e}"
+        )
+
+
 @shared_task(soft_time_limit=3600)
 def auto_add_credentials_to_devices(credential_id, organization_id):
     Credentials = load_model("connection", "Credentials")
diff --git a/openwisp_controller/connection/templates/admin/connection/batch_command/batch_command_change_form.html b/openwisp_controller/connection/templates/admin/connection/batch_command/batch_command_change_form.html
new file mode 100644
index 000000000..43ae2a83c
--- /dev/null
+++ b/openwisp_controller/connection/templates/admin/connection/batch_command/batch_command_change_form.html
@@ -0,0 +1,173 @@
+{% extends "admin/change_form.html" %}
+{% load i18n admin_urls static admin_list ow_tags %}
+
+{% block extrahead %}
+{{ block.super }}
+<link rel="stylesheet" type="text/css" href="{% static "admin/css/changelists.css" %}">
+<link rel="stylesheet" type="text/css" href="{% static "admin/css/ow-filters.css" %}">
+<link rel="stylesheet" type="text/css" href="{% static "connection/css/batch-command.css" %}">
+{% endblock %}
+
+{% block content %}
+{{ block.super }}
+
+<!-- Commands Section -->
+<h1 class="commands-title">{% trans "Commands" %}</h1>
+
+<!-- OpenWISP Filter Section -->
+{% if filter_specs %}
+<div id="ow-changelist-filter">
+  <div class="filters-bottom">
+    <span class="left-arrow">
+      <img src="{% static "ui/openwisp/images/left-arrow.svg" %}" alt="{% trans 'left' %}">
+    </span>
+    <span class="right-arrow">
+      <img src="{% static "ui/openwisp/images/right-arrow.svg" %}" alt="{% trans 'right' %}">
+    </span>
+    <div aria-label="{% trans 'changelist filters region' %}" tabindex="0" class="ow-filter-slider">
+      <div class="slider">
+        {% for spec in filter_specs %}
+        <div class="ow-filter {{ spec.title|join_string }}">
+          {% for choice in spec.choices %}
+          {% if choice.selected %}
+          <div id="{{ spec.title|join_string }}" aria-label="{% trans 'by' %} {{ spec.title }}" tabindex="0"
+            aria-controls="choices-{{ spec.title|join_string }}" role="button" aria-expanded="false"
+            class="filter-title">
+            <h3>
+              {% blocktrans with filter_title=spec.title %} By {{ filter_title }} {% endblocktrans %}
+            </h3>
+            <div class="selected-option" tabindex="0" title="{{ choice.display }}">
+              {{ choice.display }}
+            </div>
+          </div>
+          {% endif %}
+          {% endfor %}
+          <div id="choices-{{ spec.title|join_string }}" role="region" aria-labelledby="{{ spec.title|join_string }}"
+            class="filter-options">
+            <ul>
+              {% for choice in spec.choices %}
+              <li>
+                <a {% if choice.selected %} class="selected" {% endif %}
+                  title="{{ choice.display | default_if_none:'------'}}" href="{{ choice.query_string|iriencode }}">
+                  {{ choice.display | default_if_none:'------' }}
+                </a>
+              </li>
+              {% endfor %}
+            </ul>
+          </div>
+        </div>
+        {% endfor %}
+      </div>
+    </div>
+  </div>
+  <div class="filters-top">
+    <h2>{% trans 'Filter' %}</h2>
+    <div class="filters-control">
+      {% if has_active_filters %}
+      <h3 id="changelist-filter-clear" class="filter-clear-heading">
+        <a href="?{% if request.GET.q %}q={{ request.GET.q|urlencode }}{% endif %}" class="filter-clear-link">&#10006; {% trans "Clear all filters" %}</a>
+      </h3>
+      {% endif %}
+      {% if filter_specs|length > 4 %}
+      <button id="ow-apply-filter" class="danger-btn">{% trans "Apply Filters" %}</button>
+      {% endif %}
+    </div>
+  </div>
+</div>
+{% endif %}
+
+<!-- Search Bar -->
+<div class="search-section">
+  <form method="get" class="search-form">
+    <svg class="search-icon" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2">
+      <circle cx="11" cy="11" r="8"></circle>
+      <path d="m21 21-4.35-4.35"></path>
+    </svg>
+    <input type="text" id="searchbar" name="q" aria-label="{% trans 'Search devices...' %}"
+      placeholder="{% trans 'Search devices...' %}" value="{{ request.GET.q }}" class="search-input">
+    <button type="submit" class="search-button">{% trans "Search" %}</button>
+    {% for param, value in request.GET.items %}
+    {% if param != 'q' and param != 'page' %}
+    <input type="hidden" name="{{ param }}" value="{{ value }}">
+    {% endif %}
+    {% endfor %}
+  </form>
+</div>
+
+<!-- Results Table -->
+<div class="results-container">
+  <table id="result_list" class="results-table">
+    <thead>
+      <tr>
+        <th>{% trans "Device" %}</th>
+        <th>{% trans "Status" %}</th>
+        <th>{% trans "Output" %}</th>
+        <th>{% trans "Timestamp" %}</th>
+      </tr>
+    </thead>
+    <tbody>
+      {% for command in commands %}
+      <tr class="{% cycle 'row1' 'row2' %}">
+        <td>
+          {% if command.is_skipped %}
+          <span class="device-name-disabled">{{ command.device_name }}</span>
+          {% else %}
+          <a href="{% url 'admin:config_device_change' command.device_pk %}#command_set-2-group"
+            class="device-link" aria-label="{% trans 'View recent commands' %} {{ command.device_name }}">
+            {{ command.device_name }}
+          </a>
+          {% endif %}
+        </td>
+        <td>
+          <span class="command-status {{ command.status }}">{{ command.status_display }}</span>
+        </td>
+        <td class="command-output">
+          <pre>{{ command.output|default:"-" }}</pre>
+        </td>
+        <td>{{ command.created|date|default:"-" }}</td>
+      </tr>
+      {% empty %}
+      <tr>
+        <td colspan="4" class="empty-results">{% trans "No commands found." %}</td>
+      </tr>
+      {% endfor %}
+    </tbody>
+  </table>
+
+  <!-- Results Count -->
+  {% if paginator %}
+  <p class="paginator">
+    {% blocktrans count counter=paginator.count %}
+      {{ counter }} command
+      {% plural %}{{ counter }} commands
+    {% endblocktrans %}
+  </p>
+  {% endif %}
+
+  <!-- Pagination -->
+  {% if page_obj.has_other_pages %}
+  <div class="pagination">
+    <span class="step-links">
+      {% if page_obj.has_previous %}
+      <a href="?{% for key,value in request.GET.items %}{% if key != 'page' %}{{ key }}={{ value|urlencode }}&{% endif %}{% endfor %}page={{ page_obj.previous_page_number }}">
+        {% trans 'Previous' %}
+      </a>
+      {% endif %}
+      <span class="current-page">
+        {% trans 'Page' %} {{ page_obj.number }} {% trans 'of' %} {{ page_obj.paginator.num_pages }}
+      </span>
+      {% if page_obj.has_next %}
+      <a href="?{% for key,value in request.GET.items %}{% if key != 'page' %}{{ key }}={{ value|urlencode }}&{% endif %}{% endfor %}page={{ page_obj.next_page_number }}">
+        {% trans 'Next' %}
+      </a>
+      {% endif %}
+    </span>
+  </div>
+  {% endif %}
+</div>
+{% endblock %}
+
+{% block footer %}
+{{ block.super }}
+<script src="{% static "admin/js/ow-filter.js" %}"></script>
+{% endblock %}
diff --git a/openwisp_controller/connection/tests/pytest.py b/openwisp_controller/connection/tests/pytest.py
index 220604cef..d7f7f5659 100644
--- a/openwisp_controller/connection/tests/pytest.py
+++ b/openwisp_controller/connection/tests/pytest.py
@@ -65,6 +65,7 @@ def _get_expected_response(self, command):
                 "output": command.output,
                 "device": str(command.device_id),
                 "connection": str(command.connection_id),
+                "batch_command": None,
             },
         }
 
diff --git a/openwisp_controller/connection/tests/test_api.py b/openwisp_controller/connection/tests/test_api.py
index 6e3a827ab..f52001af7 100644
--- a/openwisp_controller/connection/tests/test_api.py
+++ b/openwisp_controller/connection/tests/test_api.py
@@ -15,15 +15,19 @@
 from openwisp_users.tests.test_api import AuthenticationMixin
 
 from .. import settings as app_settings
-from ..api.views import CommandListCreateView
+from ..api.views import BatchCommandListView, CommandListCreateView
 from ..commands import ORGANIZATION_ENABLED_COMMANDS
 from .utils import CreateCommandMixin, CreateConnectionsMixin
 
 Command = load_model("connection", "Command")
 DeviceConnection = load_model("connection", "DeviceConnection")
+BatchCommand = load_model("connection", "BatchCommand")
 command_qs = Command.objects.order_by("-created")
 OrganizationUser = load_model("openwisp_users", "OrganizationUser")
 Group = load_model("openwisp_users", "Group")
+DeviceGroup = load_model("config", "DeviceGroup")
+Location = load_model("geo", "Location")
+DeviceLocation = load_model("geo", "DeviceLocation")
 
 
 class TestCommandsAPI(TestCase, AuthenticationMixin, CreateCommandMixin):
@@ -860,3 +864,1149 @@ def test_deviceconnection_unauthenticated_user(self):
                 "delete": 401,
             },
         )
+
+
+class TestBatchCommandsAPI(
+    TestAdminMixin, AuthenticationMixin, TestCase, CreateConnectionsMixin
+):
+    url_namespace = "connection_api"
+
+    def setUp(self):
+        super().setUp()
+        self._login()
+
+    def _create_batch_command(self, organization, **kwargs):
+        opts = dict(
+            organization=organization,
+            type="custom",
+            input={"command": "echo test"},
+            label="test-label",
+        )
+        devices = kwargs.pop("devices", None)
+        opts.update(kwargs)
+        batch = BatchCommand(**opts)
+        batch.full_clean()
+        batch.save()
+        if devices is not None:
+            if not isinstance(devices, (list, tuple)):
+                devices = [devices]
+            batch.devices.set(devices)
+        return batch
+
+    def test_batch_command_list(self):
+        org = self._get_org()
+        url = reverse("connection_api:batch_command_list")
+        for _ in range(3):
+            self._create_batch_command(organization=org)
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data["count"], 3)
+        self.assertEqual(len(response.data["results"]), 3)
+        created_list = [cmd["created"] for cmd in response.data["results"]]
+        sorted_created = sorted(created_list, reverse=True)
+        self.assertEqual(created_list, sorted_created)
+        result = response.data["results"][0]
+        self.assertIn("id", result)
+        self.assertIn("status", result)
+        self.assertIn("type", result)
+        self.assertIn("input", result)
+        self.assertIn("device_count", result)
+        self.assertIn("created", result)
+        self.assertEqual(result["device_count"], 0)
+
+        with patch.object(BatchCommandListView, "pagination_page_size", 2, create=True):
+            with self.subTest("pagination page 1"):
+                for _ in range(2):
+                    self._create_batch_command(organization=org)
+                response = self.client.get(url + "?page=1")
+                self.assertEqual(response.status_code, 200)
+                self.assertEqual(len(response.data["results"]), 2)
+
+            with self.subTest("pagination page 2"):
+                response = self.client.get(url + "?page=2")
+                self.assertEqual(response.status_code, 200)
+                self.assertEqual(len(response.data["results"]), 2)
+
+            with self.subTest("pagination page 3 (partial)"):
+                response = self.client.get(url + "?page=3")
+                self.assertEqual(response.status_code, 200)
+                self.assertEqual(len(response.data["results"]), 1)
+
+    def test_batch_command_detail(self):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+        batch = self._create_batch_command(organization=org, devices=[device])
+        url = reverse("connection_api:batch_command_detail", args=[batch.pk])
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual(response.data["id"], str(batch.pk))
+        self.assertEqual(response.data["status"], batch.status)
+        self.assertEqual(response.data["type"], batch.type)
+        self.assertEqual(response.data["input"], batch.input)
+        self.assertIn("devices", response.data)
+        self.assertEqual(response.data["devices"], [str(device.pk)])
+        self.assertEqual(response.data["device_count"], 1)
+
+    def test_batch_command_dry_run_endpoint(self):
+        org = self._get_org()
+        device1 = self._create_device(
+            name="dryf-dev1",
+            mac_address="00:11:22:33:44:d1",
+            organization=org,
+        )
+        self._create_config(device=device1)
+        device2 = self._create_device(
+            name="dryf-dev2",
+            mac_address="00:11:22:33:44:d2",
+            organization=org,
+        )
+        self._create_config(device=device2)
+        group = DeviceGroup.objects.create(name="dryf-group", organization=org)
+        device1.group = group
+        device1.save()
+        location = Location.objects.create(
+            name="dryf-loc",
+            type="indoor",
+            organization=org,
+        )
+        DeviceLocation.objects.create(content_object=device2, location=location)
+
+        base_url = reverse("connection_api:batch_command_execute")
+
+        with self.subTest("dry run with explicit devices"):
+            url = "{0}?organization={1}&devices={2}".format(
+                base_url,
+                str(org.pk),
+                str(device1.pk),
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.data["devices"], [str(device1.pk)])
+
+        with self.subTest("dry run with group"):
+            url = "{0}?organization={1}&group={2}".format(
+                base_url,
+                str(org.pk),
+                str(group.pk),
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+            self.assertIn(str(device1.pk), response.data["devices"])
+            self.assertNotIn(str(device2.pk), response.data["devices"])
+
+        with self.subTest("dry run with location"):
+            url = "{0}?organization={1}&location={2}".format(
+                base_url,
+                str(org.pk),
+                str(location.pk),
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+            self.assertIn(str(device2.pk), response.data["devices"])
+            self.assertNotIn(str(device1.pk), response.data["devices"])
+
+        with self.subTest("dry run with group and location"):
+            DeviceLocation.objects.create(content_object=device1, location=location)
+            url = "{0}?organization={1}&group={2}&location={3}".format(
+                base_url,
+                str(org.pk),
+                str(group.pk),
+                str(location.pk),
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+            self.assertIn(str(device1.pk), response.data["devices"])
+            self.assertNotIn(str(device2.pk), response.data["devices"])
+
+        with self.subTest("dry run org-wide"):
+            url = "{0}?organization={1}".format(base_url, str(org.pk))
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+            self.assertIn(str(device1.pk), response.data["devices"])
+            self.assertIn(str(device2.pk), response.data["devices"])
+
+    def test_batch_command_execute(self):
+        org = self._get_org()
+        cred = self._create_credentials(name="exec-cred", organization=org)
+        device1 = self._create_device(
+            name="exec-dev1",
+            mac_address="00:11:22:33:44:e1",
+            organization=org,
+        )
+        self._create_config(device=device1)
+        self._create_device_connection(device=device1, credentials=cred)
+        device2 = self._create_device(
+            name="exec-dev2",
+            mac_address="00:11:22:33:44:e2",
+            organization=org,
+        )
+        self._create_config(device=device2)
+        self._create_device_connection(device=device2, credentials=cred)
+        group = DeviceGroup.objects.create(name="exec-group", organization=org)
+        device1.group = group
+        device1.save()
+        location = Location.objects.create(
+            name="exec-loc",
+            type="indoor",
+            organization=org,
+        )
+        DeviceLocation.objects.create(content_object=device2, location=location)
+        url = reverse("connection_api:batch_command_execute")
+
+        with self.subTest("execute with explicit devices"):
+            response = self.client.post(
+                url,
+                data=json.dumps(
+                    {
+                        "organization": str(org.pk),
+                        "type": "custom",
+                        "input": {"command": "echo test"},
+                        "label": "test-label",
+                        "devices": [str(device1.pk)],
+                    }
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            batch = BatchCommand.objects.get(pk=response.data["batch"])
+            batch.create_commands()
+            batch.refresh_from_db()
+            command = Command.objects.get(batch_command=batch, device=device1)
+            self.assertEqual(command.device.pk, device1.pk)
+            self.assertEqual(batch.skipped_devices, {})
+
+        with self.subTest("execute with group"):
+            response = self.client.post(
+                url,
+                data=json.dumps(
+                    {
+                        "organization": str(org.pk),
+                        "type": "custom",
+                        "input": {"command": "echo test"},
+                        "label": "test-label",
+                        "group": str(group.pk),
+                    }
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            batch = BatchCommand.objects.get(pk=response.data["batch"])
+            batch.create_commands()
+            batch.refresh_from_db()
+            command = Command.objects.get(batch_command=batch, device=device1)
+            self.assertEqual(command.device.pk, device1.pk)
+            self.assertEqual(batch.skipped_devices, {})
+
+        with self.subTest("execute with location"):
+            response = self.client.post(
+                url,
+                data=json.dumps(
+                    {
+                        "organization": str(org.pk),
+                        "type": "custom",
+                        "input": {"command": "echo test"},
+                        "label": "test-label",
+                        "location": str(location.pk),
+                    }
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            batch = BatchCommand.objects.get(pk=response.data["batch"])
+            batch.create_commands()
+            batch.refresh_from_db()
+            command = Command.objects.get(batch_command=batch, device=device2)
+            self.assertEqual(command.device.pk, device2.pk)
+            self.assertEqual(batch.skipped_devices, {})
+
+        with self.subTest("execute with group and location"):
+            DeviceLocation.objects.create(content_object=device1, location=location)
+            response = self.client.post(
+                url,
+                data=json.dumps(
+                    {
+                        "organization": str(org.pk),
+                        "type": "custom",
+                        "input": {"command": "echo test"},
+                        "label": "test-label",
+                        "group": str(group.pk),
+                        "location": str(location.pk),
+                    }
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            batch = BatchCommand.objects.get(pk=response.data["batch"])
+            batch.create_commands()
+            batch.refresh_from_db()
+            command = Command.objects.get(batch_command=batch, device=device1)
+            self.assertEqual(command.device.pk, device1.pk)
+            self.assertEqual(batch.skipped_devices, {})
+
+        with self.subTest("execute org-wide"):
+            response = self.client.post(
+                url,
+                data=json.dumps(
+                    {
+                        "organization": str(org.pk),
+                        "type": "custom",
+                        "input": {"command": "echo test"},
+                        "label": "test-label",
+                        "execute_all": True,
+                    }
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            batch = BatchCommand.objects.get(pk=response.data["batch"])
+            batch.create_commands()
+            batch.refresh_from_db()
+            self.assertEqual(
+                Command.objects.filter(batch_command=batch).count(),
+                2,
+            )
+            self.assertEqual(batch.skipped_devices, {})
+
+    def test_batch_command_endpoints_no_of_queries(self):
+        with self.subTest("list queries"):
+            org = self._get_org()
+            devices = []
+            for i in range(2):
+                d = self._create_device(
+                    name=f"q-dev-{i}",
+                    mac_address=f"00:11:22:33:44:{i:02x}",
+                    organization=org,
+                )
+                self._create_config(device=d)
+                devices.append(d)
+            self._create_batch_command(organization=org, devices=devices)
+            url = reverse("connection_api:batch_command_list")
+            with self.assertNumQueries(3):
+                response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.data["count"], 1)
+
+        with self.subTest("detail queries"):
+            url = reverse(
+                "connection_api:batch_command_detail",
+                args=[response.data["results"][0]["id"]],
+            )
+            with self.assertNumQueries(3):
+                response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest("execute queries"):
+            org = self._get_org()
+            devices = []
+            for i in range(3):
+                d = self._create_device(
+                    name=f"q-exec-{i}",
+                    mac_address=f"00:11:22:33:44:{i + 0x10:02x}",
+                    organization=org,
+                )
+                self._create_config(device=d)
+                devices.append(d)
+            payload = {
+                "organization": str(org.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "devices": [str(d.pk) for d in devices],
+            }
+            url = reverse("connection_api:batch_command_execute")
+            with self.assertNumQueries(17):
+                response = self.client.post(
+                    url,
+                    data=json.dumps(payload),
+                    content_type="application/json",
+                )
+            self.assertEqual(response.status_code, 201)
+            self.assertIn("batch", response.data)
+            batch = BatchCommand.objects.get(pk=response.data["batch"])
+            self.assertEqual(batch.devices.count(), 3)
+            self.assertCountEqual(
+                batch.devices.values_list("pk", flat=True),
+                [d.pk for d in devices],
+            )
+
+    def test_batch_command_execute_org_has_no_devices(self):
+        org = self._get_org()
+        payload = {
+            "organization": str(org.pk),
+            "type": "custom",
+            "input": {"command": "echo test"},
+            "label": "test-label",
+            "execute_all": True,
+        }
+        url = reverse("connection_api:batch_command_execute")
+        response = self.client.post(
+            url,
+            data=json.dumps(payload),
+            content_type="application/json",
+        )
+        self.assertEqual(response.status_code, 400)
+        self.assertEqual(
+            response.data,
+            ["No devices match the specified criteria."],
+        )
+
+    def test_batch_command_execute_disallowed_type(self):
+        org = self._get_org()
+        url = reverse("connection_api:batch_command_execute")
+        with patch.dict(
+            ORGANIZATION_ENABLED_COMMANDS,
+            {str(org.pk): ("reboot",)},
+        ):
+            payload = {
+                "organization": str(org.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "execute_all": True,
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 400)
+            self.assertIn(
+                '"custom" command is not available for this organization',
+                response.data["type"][0],
+            )
+
+    def test_batch_command_no_org_only_allowed_to_superuser(self):
+        org = self._get_org()
+        self.client.logout()
+        operator = self._create_operator(organizations=[org])
+        add_perm = Permission.objects.get(codename="add_batchcommand")
+        operator.user_permissions.add(add_perm)
+        self.client.force_login(operator)
+        payload = {
+            "type": "custom",
+            "input": {"command": "echo test"},
+            "label": "test-label",
+            "execute_all": True,
+        }
+        url = reverse("connection_api:batch_command_execute")
+        with self.subTest("POST execute without org"):
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 400)
+            self.assertIn(
+                "Only superusers",
+                str(response.data),
+            )
+
+        with self.subTest("GET dry-run without org"):
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 400)
+            self.assertIn(
+                "Only superusers",
+                str(response.data),
+            )
+
+    def test_batch_command_operator_endpoints_on_managed_org(self):
+        org = self._get_org()
+        self._create_credentials(name="op-cred", organization=org)
+        device = self._create_device(
+            name="op-dev",
+            mac_address="00:11:22:33:44:01",
+            organization=org,
+        )
+        self._create_config(device=device)
+        self._create_device_connection(device=device)
+        self.client.logout()
+        operator = self._create_operator(organizations=[org])
+        operator.user_permissions.add(
+            Permission.objects.get(codename="add_batchcommand"),
+            Permission.objects.get(codename="view_batchcommand"),
+        )
+        self.client.force_login(operator)
+        url = reverse("connection_api:batch_command_execute")
+
+        with self.subTest("execute"):
+            payload = {
+                "organization": str(org.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "devices": [str(device.pk)],
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            self.assertIn("batch", response.data)
+
+        with self.subTest("dry-run"):
+            response = self.client.get(
+                url,
+                data={"organization": str(org.pk)},
+            )
+            self.assertEqual(response.status_code, 200)
+            self.assertIn("devices", response.data)
+
+        with self.subTest("execute_all"):
+            payload = {
+                "organization": str(org.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "execute_all": True,
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            self.assertIn("batch", response.data)
+
+    def test_batch_command_administrator_endpoints_on_managed_org(self):
+        org = self._get_org()
+        self._create_credentials(name="admin-cred", organization=org)
+        device = self._create_device(
+            name="admin-dev",
+            mac_address="00:11:22:33:44:02",
+            organization=org,
+        )
+        self._create_config(device=device)
+        self._create_device_connection(device=device)
+        self.client.logout()
+        administrator = self._create_administrator(organizations=[org])
+        self.client.force_login(administrator)
+        url = reverse("connection_api:batch_command_execute")
+
+        with self.subTest("execute"):
+            payload = {
+                "organization": str(org.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "devices": [str(device.pk)],
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            self.assertIn("batch", response.data)
+
+        with self.subTest("dry-run"):
+            response = self.client.get(
+                url,
+                data={"organization": str(org.pk)},
+            )
+            self.assertEqual(response.status_code, 200)
+            self.assertIn("devices", response.data)
+
+        with self.subTest("execute_all"):
+            payload = {
+                "organization": str(org.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "execute_all": True,
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            self.assertIn("batch", response.data)
+
+    def test_batch_command_operator_endpoints_on_non_managed_org(self):
+        org1 = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        self._create_credentials(name="op-nm-cred", organization=org1)
+        device = self._create_device(
+            name="op-nm-dev",
+            mac_address="00:11:22:33:44:03",
+            organization=org1,
+        )
+        self._create_config(device=device)
+        self._create_device_connection(device=device)
+        self.client.logout()
+        operator = self._create_operator(organizations=[org1])
+        operator.user_permissions.add(
+            Permission.objects.get(codename="add_batchcommand"),
+            Permission.objects.get(codename="view_batchcommand"),
+        )
+        self.client.force_login(operator)
+        url = reverse("connection_api:batch_command_execute")
+
+        with self.subTest("execute"):
+            payload = {
+                "organization": str(org2.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "devices": [str(device.pk)],
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 400)
+
+        with self.subTest("dry-run"):
+            response = self.client.get(
+                url,
+                data={"organization": str(org2.pk)},
+            )
+            self.assertEqual(response.status_code, 400)
+
+        with self.subTest("execute_all"):
+            payload = {
+                "organization": str(org2.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "execute_all": True,
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 400)
+
+    def test_batch_command_administrator_endpoints_on_non_managed_org(self):
+        org1 = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        self._create_credentials(name="admin-nm-cred", organization=org1)
+        device = self._create_device(
+            name="admin-nm-dev",
+            mac_address="00:11:22:33:44:04",
+            organization=org1,
+        )
+        self._create_config(device=device)
+        self._create_device_connection(device=device)
+        self.client.logout()
+        administrator = self._create_administrator(organizations=[org1])
+        self.client.force_login(administrator)
+        url = reverse("connection_api:batch_command_execute")
+
+        with self.subTest("execute"):
+            payload = {
+                "organization": str(org2.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "devices": [str(device.pk)],
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 400)
+
+        with self.subTest("dry-run"):
+            response = self.client.get(
+                url,
+                data={"organization": str(org2.pk)},
+            )
+            self.assertEqual(response.status_code, 400)
+
+        with self.subTest("execute_all"):
+            payload = {
+                "organization": str(org2.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "execute_all": True,
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 400)
+
+    def test_batch_command_endpoints_organization_scoped(self):
+        org = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        batch_org1 = self._create_batch_command(organization=org)
+        batch_org2 = self._create_batch_command(organization=org2)
+        self.client.logout()
+        operator = self._create_operator(organizations=[org])
+        view_perm = Permission.objects.get(codename="view_batchcommand")
+        operator.user_permissions.add(view_perm)
+        self.client.force_login(operator)
+
+        with self.subTest("list scoped to own org"):
+            url = reverse("connection_api:batch_command_list")
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.data["count"], 1)
+
+        with self.subTest("detail cross-org"):
+            url = reverse(
+                "connection_api:batch_command_detail",
+                args=[batch_org2.pk],
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 404)
+
+        with self.subTest("detail own org"):
+            url = reverse(
+                "connection_api:batch_command_detail",
+                args=[batch_org1.pk],
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+
+    def test_batch_command_endpoints_unauthorized(self):
+        self.client.logout()
+        execute_url = reverse("connection_api:batch_command_execute")
+
+        with self.subTest("List"):
+            url = reverse("connection_api:batch_command_list")
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 401)
+
+        with self.subTest("Detail"):
+            url = reverse(
+                "connection_api:batch_command_detail",
+                args=[uuid.uuid4()],
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 401)
+
+        with self.subTest("Dry run"):
+            response = self.client.get(execute_url)
+            self.assertEqual(response.status_code, 401)
+
+        with self.subTest("Execute"):
+            response = self.client.post(
+                execute_url,
+                data=json.dumps({"type": "custom"}),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 401)
+
+    def test_batch_command_bearer_authentication(self):
+        self.client.logout()
+        org = self._get_org()
+        batch = self._create_batch_command(organization=org)
+        device = self._create_device(
+            name="bearer-dev",
+            mac_address="00:11:22:33:44:be",
+            organization=org,
+        )
+        self._create_config(device=device)
+        self._create_device_connection(device=device)
+        token = self._obtain_auth_token(username="admin", password="tester")
+        auth = {"HTTP_AUTHORIZATION": f"Bearer {token}"}
+
+        with self.subTest("list"):
+            url = reverse("connection_api:batch_command_list")
+            response = self.client.get(url, **auth)
+            self.assertEqual(response.status_code, 200)
+            self.assertIn("results", response.data)
+
+        with self.subTest("detail"):
+            url = reverse(
+                "connection_api:batch_command_detail",
+                args=[batch.pk],
+            )
+            response = self.client.get(url, **auth)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.data["id"], str(batch.pk))
+
+        with self.subTest("dry-run"):
+            url = reverse("connection_api:batch_command_execute")
+            response = self.client.get(
+                url,
+                data={"organization": str(org.pk)},
+                **auth,
+            )
+            self.assertEqual(response.status_code, 200)
+
+        with self.subTest("execute"):
+            url = reverse("connection_api:batch_command_execute")
+            payload = {
+                "organization": str(org.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "devices": [str(device.pk)],
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+                **auth,
+            )
+            self.assertEqual(response.status_code, 201)
+            self.assertIn("batch", response.data)
+
+    def test_batch_command_detail_404(self):
+        org = self._get_org()
+        self._create_batch_command(organization=org)
+        url = reverse(
+            "connection_api:batch_command_detail",
+            args=[uuid.uuid4()],
+        )
+        response = self.client.get(url)
+        self.assertEqual(response.status_code, 404)
+
+    def test_batch_command_endpoints_operator_different_org(self):
+        org = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        batch_org1 = self._create_batch_command(organization=org)
+        self.client.logout()
+        operator = self._create_operator(organizations=[org2])
+        self.client.force_login(operator)
+
+        with self.subTest("list"):
+            url = reverse("connection_api:batch_command_list")
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.data["count"], 0)
+
+        with self.subTest("detail"):
+            url = reverse(
+                "connection_api:batch_command_detail",
+                args=[batch_org1.pk],
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 404)
+
+        with self.subTest("dry-run"):
+            url = reverse("connection_api:batch_command_execute")
+            response = self.client.get(
+                url,
+                data={"organization": str(org.pk)},
+            )
+            self.assertEqual(response.status_code, 400)
+
+        with self.subTest("execute"):
+            url = reverse("connection_api:batch_command_execute")
+            payload = {
+                "organization": str(org.pk),
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "execute_all": True,
+            }
+            response = self.client.post(
+                url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 400)
+
+    def test_batch_command_execute_skipped_devices(self):
+        org = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        device_a = self._create_device(
+            name="device-a",
+            mac_address="00:11:22:33:44:aa",
+            organization=org,
+        )
+        self._create_config(device=device_a)
+        self._create_device_connection(device=device_a)
+        device_b = self._create_device(
+            name="device-b",
+            mac_address="00:11:22:33:44:bb",
+            organization=org2,
+        )
+        self._create_config(device=device_b)
+        self._create_device_connection(device=device_b)
+        execute_url = reverse("connection_api:batch_command_execute")
+
+        with patch.dict(
+            ORGANIZATION_ENABLED_COMMANDS,
+            {str(org2.pk): ("reboot",)},
+        ):
+            payload = {
+                "type": "custom",
+                "input": {"command": "echo test"},
+                "label": "test-label",
+                "devices": [str(device_a.pk), str(device_b.pk)],
+            }
+            response = self.client.post(
+                execute_url,
+                data=json.dumps(payload),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            batch = BatchCommand.objects.get(pk=response.data["batch"])
+            # transaction.on_commit doesn't fire in TestCase;
+            # trigger create_commands() manually
+            batch.create_commands()
+            batch.refresh_from_db()
+            self.assertIn(str(device_b.pk), batch.skipped_devices)
+            self.assertIn(
+                '"custom" command is not available for this organization',
+                batch.skipped_devices[str(device_b.pk)][0],
+            )
+            command_qs = Command.objects.filter(batch_command=batch)
+            self.assertTrue(command_qs.filter(device=device_a).exists())
+            self.assertFalse(command_qs.filter(device=device_b).exists())
+            url = reverse(
+                "connection_api:device_command_list",
+                args=[device_a.pk],
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 200)
+            cmd_data = response.data["results"][0]
+            self.assertIn("type", cmd_data)
+            self.assertIn("input", cmd_data)
+
+        with self.subTest("skipped: no credentials"):
+            device = self._create_device(
+                name="skip-nc-dev",
+                mac_address="00:11:22:33:44:a1",
+                organization=org,
+            )
+            self._create_config(device=device)
+            response = self.client.post(
+                execute_url,
+                data=json.dumps(
+                    {
+                        "organization": str(org.pk),
+                        "type": "custom",
+                        "input": {"command": "echo test"},
+                        "label": "test-label",
+                        "devices": [str(device.pk)],
+                    }
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            batch = BatchCommand.objects.get(pk=response.data["batch"])
+            batch.create_commands()
+            batch.refresh_from_db()
+            self.assertIn(str(device.pk), batch.skipped_devices)
+            self.assertIn(
+                "Device has no credentials assigned",
+                batch.skipped_devices[str(device.pk)][0],
+            )
+            detail_url = reverse(
+                "connection_api:batch_command_detail",
+                args=[batch.pk],
+            )
+            detail_response = self.client.get(detail_url)
+            self.assertEqual(detail_response.status_code, 200)
+            self.assertIn(str(device.pk), detail_response.data["skipped_devices"])
+
+        with self.subTest("skipped: deactivated device"):
+            device = self._create_device(
+                name="skip-dd-dev",
+                mac_address="00:11:22:33:44:a2",
+                organization=org,
+            )
+            self._create_config(device=device)
+            dd_cred = self._create_credentials(
+                name="skip-dd-cred",
+                organization=org,
+            )
+            self._create_device_connection(device=device, credentials=dd_cred)
+            device.deactivate()
+            response = self.client.post(
+                execute_url,
+                data=json.dumps(
+                    {
+                        "organization": str(org.pk),
+                        "type": "custom",
+                        "input": {"command": "echo test"},
+                        "label": "test-label",
+                        "devices": [str(device.pk)],
+                    }
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 201)
+            batch = BatchCommand.objects.get(pk=response.data["batch"])
+            batch.create_commands()
+            batch.refresh_from_db()
+            self.assertIn(str(device.pk), batch.skipped_devices)
+            self.assertIn(
+                "Device is deactivated",
+                batch.skipped_devices[str(device.pk)][0],
+            )
+            detail_url = reverse(
+                "connection_api:batch_command_detail",
+                args=[batch.pk],
+            )
+            detail_response = self.client.get(detail_url)
+            self.assertEqual(detail_response.status_code, 200)
+            self.assertIn(str(device.pk), detail_response.data["skipped_devices"])
+
+    def test_batch_command_execute_org_mismatched_data_provided(self):
+        org = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        device_org2 = self._create_device(
+            name="api-mm-dev",
+            mac_address="00:11:22:33:44:88",
+            organization=org2,
+        )
+        url = reverse("connection_api:batch_command_execute")
+
+        with self.subTest("device org mismatch"):
+            response = self.client.post(
+                url,
+                data=json.dumps(
+                    {
+                        "organization": str(org.pk),
+                        "type": "custom",
+                        "input": {"command": "echo test"},
+                        "label": "test-label",
+                        "devices": [str(device_org2.pk)],
+                    }
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(
+                response.data,
+                {"devices": ["All devices must belong to the same organization."]},
+            )
+
+        with self.subTest("group org mismatch"):
+            group_org2 = DeviceGroup.objects.create(
+                name="api-mm-group",
+                organization=org2,
+            )
+            response = self.client.post(
+                url,
+                data=json.dumps(
+                    {
+                        "organization": str(org.pk),
+                        "type": "custom",
+                        "input": {"command": "echo test"},
+                        "label": "test-label",
+                        "group": str(group_org2.pk),
+                    }
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(
+                response.data,
+                {
+                    "group": [
+                        (
+                            "Please ensure that the organization of this Batch command "
+                            "and the organization of the related Device Group match."
+                        )
+                    ]
+                },
+            )
+
+        with self.subTest("location org mismatch"):
+            location_org2 = Location.objects.create(
+                name="api-mm-loc",
+                type="indoor",
+                organization=org2,
+            )
+            response = self.client.post(
+                url,
+                data=json.dumps(
+                    {
+                        "organization": str(org.pk),
+                        "type": "custom",
+                        "input": {"command": "echo test"},
+                        "label": "test-label",
+                        "location": str(location_org2.pk),
+                    }
+                ),
+                content_type="application/json",
+            )
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(
+                response.data,
+                {
+                    "location": [
+                        (
+                            "Please ensure that the organization of this Batch command "
+                            "and the organization of the related location match."
+                        )
+                    ]
+                },
+            )
+
+    def test_batch_command_dry_run_org_mismatched_data_provided(self):
+        org = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        device_org2 = self._create_device(
+            name="dry-mm-dev",
+            mac_address="00:11:22:33:44:78",
+            organization=org2,
+        )
+        base_url = reverse("connection_api:batch_command_execute")
+
+        with self.subTest("device org mismatch"):
+            url = "{0}?organization={1}&devices={2}".format(
+                base_url,
+                str(org.pk),
+                str(device_org2.pk),
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(
+                response.data,
+                {"devices": ["All devices must belong to the same organization."]},
+            )
+
+        with self.subTest("group org mismatch"):
+            group_org2 = DeviceGroup.objects.create(
+                name="dry-mm-group",
+                organization=org2,
+            )
+            url = "{0}?organization={1}&group={2}".format(
+                base_url,
+                str(org.pk),
+                str(group_org2.pk),
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(
+                response.data,
+                {
+                    "group": [
+                        (
+                            "Please ensure that the organization of this Batch command "
+                            "and the organization of the related Device Group match."
+                        )
+                    ]
+                },
+            )
+
+        with self.subTest("location org mismatch"):
+            location_org2 = Location.objects.create(
+                name="dry-mm-loc",
+                type="indoor",
+                organization=org2,
+            )
+            url = "{0}?organization={1}&location={2}".format(
+                base_url,
+                str(org.pk),
+                str(location_org2.pk),
+            )
+            response = self.client.get(url)
+            self.assertEqual(response.status_code, 400)
+            self.assertEqual(
+                response.data,
+                {
+                    "location": [
+                        (
+                            "Please ensure that the organization of this Batch command "
+                            "and the organization of the related location match."
+                        )
+                    ]
+                },
+            )
diff --git a/openwisp_controller/connection/tests/test_models.py b/openwisp_controller/connection/tests/test_models.py
index e3ed6bb97..bd176f582 100644
--- a/openwisp_controller/connection/tests/test_models.py
+++ b/openwisp_controller/connection/tests/test_models.py
@@ -32,6 +32,10 @@
 Group = load_model("openwisp_users", "Group")
 Organization = load_model("openwisp_users", "Organization")
 Command = load_model("connection", "Command")
+BatchCommand = load_model("connection", "BatchCommand")
+DeviceGroup = load_model("config", "DeviceGroup")
+Location = load_model("geo", "Location")
+DeviceLocation = load_model("geo", "DeviceLocation")
 
 _connect_path = "paramiko.SSHClient.connect"
 _exec_command_path = "paramiko.SSHClient.exec_command"
@@ -460,14 +464,14 @@ def test_operator_group_permissions(self):
         permissions = group.permissions.filter(
             content_type__app_label=f"{self.app_label}"
         )
-        self.assertEqual(permissions.count(), 6)
+        self.assertEqual(permissions.count(), 8)
 
     def test_administrator_group_permissions(self):
         group = Group.objects.get(name="Administrator")
         permissions = group.permissions.filter(
             content_type__app_label=f"{self.app_label}"
         )
-        self.assertEqual(permissions.count(), 12)
+        self.assertEqual(permissions.count(), 16)
 
     def test_device_connection_set_connector(self):
         dc = self._create_device_connection()
@@ -968,6 +972,797 @@ def test_command_multiple_connections(self, connect_mocked):
             command.refresh_from_db()
             self.assertIn(command.connection, [dc1, dc2])
 
+    def _create_batch_command(self, organization, **kwargs):
+        opts = dict(
+            organization=organization,
+            type="custom",
+            input={"command": "echo test"},
+            label="test-label",
+        )
+        devices = kwargs.pop("devices", None)
+        opts.update(kwargs)
+        batch = BatchCommand(**opts)
+        batch.full_clean()
+        batch.save()
+        if devices is not None:
+            if not isinstance(devices, (list, tuple)):
+                devices = [devices]
+            batch.devices.set(devices)
+        return batch
+
+    def test_batch_command_str(self):
+        org = self._get_org()
+        batch = self._create_batch_command(organization=org)
+        self.assertEqual(str(batch), "test-label")
+
+    def test_batch_command_total_devices_successful_failed(self):
+        org = self._get_org()
+        device1 = self._create_device(organization=org)
+        self._create_config(device=device1)
+        dc1 = self._create_device_connection(device=device1)
+        device2 = self._create_device(
+            name="device2",
+            mac_address="00:11:22:33:44:02",
+            organization=org,
+        )
+        self._create_config(device=device2)
+        dc2 = self._create_device_connection(
+            device=device2,
+            credentials=self._create_credentials(
+                name="Test credentials 2",
+                organization=org,
+            ),
+        )
+        batch = self._create_batch_command(organization=org)
+        Command.objects.create(
+            batch_command=batch,
+            device=device1,
+            connection=dc1,
+            type=batch.type,
+            input={"command": "echo test"},
+            status="success",
+        )
+        Command.objects.create(
+            batch_command=batch,
+            device=device2,
+            connection=dc2,
+            type=batch.type,
+            input={"command": "echo test"},
+            status="failed",
+        )
+        self.assertEqual(batch.total_devices, 2)
+        self.assertEqual(batch.successful, 1)
+        self.assertEqual(batch.failed, 1)
+        self.assertEqual(
+            batch.batch_commands.filter(status="success").first().device, device1
+        )
+        self.assertEqual(
+            batch.batch_commands.filter(status="failed").first().device, device2
+        )
+        self.assertFalse(
+            batch.batch_commands.filter(status="success", device=device2).exists()
+        )
+        self.assertFalse(
+            batch.batch_commands.filter(status="failed", device=device1).exists()
+        )
+
+    def test_batch_command_clean_validation(self):
+        org = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        device_org2 = self._create_device(organization=org2)
+
+        with self.subTest("devices from different org"):
+            batch = self._create_batch_command(organization=org)
+            batch.devices.add(device_org2)
+            with self.assertRaises(ValidationError) as ctx:
+                batch.clean()
+            self.assertIn("devices", ctx.exception.message_dict)
+            self.assertIn(
+                "must belong to the same organization",
+                ctx.exception.message_dict["devices"][0],
+            )
+
+        with self.subTest("invalid command type for org"):
+            with mock.patch.dict(
+                ORGANIZATION_ENABLED_COMMANDS,
+                {str(org.pk): ("reboot",)},
+            ):
+                batch = BatchCommand(
+                    organization=org,
+                    type="custom",
+                    input={"command": "echo test"},
+                    label="test-label",
+                )
+                with self.assertRaises(ValidationError) as ctx:
+                    batch.clean()
+                self.assertIn("type", ctx.exception.message_dict)
+                self.assertIn(
+                    "not available for this organization",
+                    ctx.exception.message_dict["type"][0],
+                )
+
+        with self.subTest("invalid JSON schema"):
+            batch = BatchCommand(
+                organization=org,
+                type="change_password",
+                input="not_an_object",
+                label="test-label",
+            )
+            with self.assertRaises(ValidationError) as ctx:
+                batch.clean()
+            self.assertIn("input", ctx.exception.message_dict)
+
+        with self.subTest("group org mismatch"):
+            group = DeviceGroup.objects.create(name="test-group", organization=org2)
+            batch = BatchCommand(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+                label="test-label",
+                group=group,
+            )
+            with self.assertRaises(ValidationError) as ctx:
+                batch.clean()
+                self.assertIn("group", ctx.exception.message_dict)
+            self.assertIn(
+                "Please ensure that the organization of this Batch command "
+                "and the organization of the related Device Group match",
+                ctx.exception.message_dict["group"][0],
+            )
+
+        with self.subTest("location org mismatch"):
+            location = Location.objects.create(
+                name="test-location",
+                type="indoor",
+                organization=org2,
+            )
+            batch = BatchCommand(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+                label="test-label",
+                location=location,
+            )
+            with self.assertRaises(ValidationError) as ctx:
+                batch.clean()
+                self.assertIn("location", ctx.exception.message_dict)
+            self.assertIn(
+                "Please ensure that the organization of this Batch command "
+                "and the organization of the related location match",
+                ctx.exception.message_dict["location"][0],
+            )
+
+    def test_batch_command_create_commands_deactivated_device(self):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+        self._create_device_connection(device=device)
+
+        with self.subTest("deactivating device does not block create_commands"):
+            device._is_deactivated = True
+            device.save(update_fields=["_is_deactivated"])
+            device.config.set_status_deactivating()
+            device = Device.objects.get(pk=device.pk)
+            batch = self._create_batch_command(organization=org)
+            batch.devices.add(device)
+            batch.create_commands()
+            self.assertEqual(batch.batch_commands.count(), 1)
+            self.assertEqual(batch.skipped_devices, {})
+
+        with self.subTest("fully deactivated device skipped in create_commands"):
+            device.config.set_status_deactivated()
+            device = Device.objects.get(pk=device.pk)
+            batch = self._create_batch_command(organization=org)
+            batch.devices.add(device)
+            batch.create_commands()
+            self.assertEqual(batch.batch_commands.count(), 0)
+            self.assertIn(str(device.pk), batch.skipped_devices)
+            self.assertIn(
+                "Device is deactivated",
+                batch.skipped_devices[str(device.pk)][0],
+            )
+
+    def test_batch_command_create_commands_no_credentials(self):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+        batch = self._create_batch_command(
+            organization=org,
+            devices=[device],
+        )
+        batch.create_commands()
+        batch.refresh_from_db()
+        self.assertIn(str(device.pk), batch.skipped_devices)
+        self.assertIn(
+            "Device has no credentials assigned",
+            batch.skipped_devices[str(device.pk)][0],
+        )
+
+    def test_batch_command_create_commands_skip_scenarios(self):
+        org = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+
+        with self.subTest("org command not allowed"):
+            device_a = self._create_device(
+                name="device-a",
+                mac_address="00:11:22:33:44:aa",
+                organization=org,
+            )
+            self._create_config(device=device_a)
+            self._create_device_connection(device=device_a)
+            device_b = self._create_device(
+                name="device-b",
+                mac_address="00:11:22:33:44:bb",
+                organization=org2,
+            )
+            self._create_config(device=device_b)
+            self._create_device_connection(device=device_b)
+            with mock.patch.dict(
+                ORGANIZATION_ENABLED_COMMANDS,
+                {str(org2.pk): ("reboot",)},
+            ):
+                batch = self._create_batch_command(
+                    organization=org,
+                    devices=[device_a, device_b],
+                )
+                batch.create_commands()
+                batch.refresh_from_db()
+                self.assertIn(str(device_b.pk), batch.skipped_devices)
+                self.assertIn(
+                    '"custom" command is not available for this organization',
+                    batch.skipped_devices[str(device_b.pk)][0],
+                )
+                db_batch = BatchCommand.objects.get(pk=batch.pk)
+                self.assertEqual(batch.skipped_devices, db_batch.skipped_devices)
+                command_qs = Command.objects.filter(batch_command=batch)
+                self.assertTrue(command_qs.filter(device=device_a).exists())
+                self.assertFalse(command_qs.filter(device=device_b).exists())
+
+        with self.subTest("mixed skip scenario"):
+            device_ok = self._create_device(
+                name="device-ok",
+                mac_address="00:11:22:33:44:01",
+                organization=org,
+            )
+            self._create_config(device=device_ok)
+            ok_cred = self._create_credentials(name="device-ok-cred", organization=org)
+            self._create_device_connection(device=device_ok, credentials=ok_cred)
+            device_no_creds = self._create_device(
+                name="device-no-creds",
+                mac_address="00:11:22:33:44:02",
+                organization=org,
+            )
+            self._create_config(device=device_no_creds)
+            device_deactivated = self._create_device(
+                name="device-deactivated",
+                mac_address="00:11:22:33:44:03",
+                organization=org,
+            )
+            self._create_config(device=device_deactivated)
+            dd_cred = self._create_credentials(name="device-dd-cred", organization=org)
+            self._create_device_connection(
+                device=device_deactivated, credentials=dd_cred
+            )
+            device_deactivated.deactivate()
+            batch = self._create_batch_command(
+                organization=org,
+                devices=[device_ok, device_no_creds, device_deactivated],
+            )
+            batch.create_commands()
+            batch.refresh_from_db()
+            command_qs = Command.objects.filter(batch_command=batch)
+            self.assertEqual(command_qs.count(), 1)
+            self.assertTrue(command_qs.filter(device=device_ok).exists())
+            self.assertIn(str(device_no_creds.pk), batch.skipped_devices)
+            self.assertIn(
+                "Device has no credentials assigned",
+                batch.skipped_devices[str(device_no_creds.pk)][0],
+            )
+            self.assertIn(str(device_deactivated.pk), batch.skipped_devices)
+            self.assertIn(
+                "Device is deactivated",
+                batch.skipped_devices[str(device_deactivated.pk)][0],
+            )
+            self.assertNotIn(str(device_ok.pk), batch.skipped_devices)
+            db_batch = BatchCommand.objects.get(pk=batch.pk)
+            self.assertEqual(batch.skipped_devices, db_batch.skipped_devices)
+
+    def test_batch_command_resolve_devices(self):
+        org = self._get_org()
+        device1 = self._create_device(
+            name="device1",
+            mac_address="00:11:22:33:44:01",
+            organization=org,
+        )
+        device2 = self._create_device(
+            name="device2",
+            mac_address="00:11:22:33:44:02",
+            organization=org,
+        )
+
+        with self.subTest("explicit devices via M2M"):
+            batch = self._create_batch_command(
+                organization=org,
+                devices=[device1],
+            )
+            resolved = list(batch.resolve_devices())
+            self.assertEqual(resolved, [device1])
+
+        with self.subTest("organization-scoped (no explicit devices)"):
+            batch = self._create_batch_command(organization=org)
+            resolved = list(batch.resolve_devices())
+            self.assertIn(device1, resolved)
+            self.assertIn(device2, resolved)
+
+        with self.subTest("group filtering"):
+            group = DeviceGroup.objects.create(name="test-group", organization=org)
+            device1.group = group
+            device1.save()
+            batch = self._create_batch_command(organization=org, group=group)
+            resolved = list(batch.resolve_devices())
+            self.assertIn(device1, resolved)
+            self.assertNotIn(device2, resolved)
+
+        with self.subTest("location filtering"):
+            location = Location.objects.create(
+                name="test-location",
+                type="indoor",
+                organization=org,
+            )
+            DeviceLocation.objects.create(content_object=device2, location=location)
+            batch = self._create_batch_command(organization=org, location=location)
+            resolved = list(batch.resolve_devices())
+            self.assertIn(device2, resolved)
+            self.assertNotIn(device1, resolved)
+
+        with self.subTest("group and location combined"):
+            DeviceLocation.objects.create(content_object=device1, location=location)
+            batch = self._create_batch_command(
+                organization=org,
+                group=device1.group,
+                location=location,
+            )
+            resolved = list(batch.resolve_devices())
+            self.assertIn(device1, resolved)
+            self.assertNotIn(device2, resolved)
+
+        with self.subTest("no devices match"):
+            empty_group = DeviceGroup.objects.create(
+                name="empty-group",
+                organization=org,
+            )
+            batch = self._create_batch_command(
+                organization=org,
+                group=empty_group,
+            )
+            resolved = list(batch.resolve_devices())
+            self.assertEqual(resolved, [])
+
+    def test_batch_command_dry_run_method(self):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+
+        with self.subTest("dry_run with type"):
+            result = BatchCommand.dry_run(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+            )
+            self.assertIn("devices", result)
+            self.assertIn(device, result["devices"])
+
+        with self.subTest("dry_run without type"):
+            result = BatchCommand.dry_run(organization=org)
+            self.assertIn("devices", result)
+            self.assertIn(device, result["devices"])
+
+        with self.subTest("dry_run with explicit devices"):
+            result = BatchCommand.dry_run(
+                organization=org,
+                devices=[device],
+            )
+            self.assertEqual(result["devices"], [device])
+
+        with self.subTest("dry_run with group"):
+            group = DeviceGroup.objects.create(name="dry-run-group", organization=org)
+            device.group = group
+            device.save()
+            result = BatchCommand.dry_run(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+                group=group,
+            )
+            self.assertIn("devices", result)
+            self.assertIn(device, result["devices"])
+
+        with self.subTest("dry_run with location"):
+            location = Location.objects.create(
+                name="dry-run-loc",
+                type="indoor",
+                organization=org,
+            )
+            DeviceLocation.objects.create(content_object=device, location=location)
+            result = BatchCommand.dry_run(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+                location=location,
+            )
+            self.assertIn("devices", result)
+            self.assertIn(device, result["devices"])
+
+        with self.subTest("dry_run with group and location"):
+            result = BatchCommand.dry_run(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+                group=group,
+                location=location,
+            )
+            self.assertIn("devices", result)
+            self.assertIn(device, result["devices"])
+
+        with self.subTest("dry run org-wide"):
+            device2 = self._create_device(
+                name="dry-org-dev2",
+                mac_address="00:11:22:33:44:77",
+                organization=org,
+            )
+            result = BatchCommand.dry_run(organization=org)
+            self.assertIn("devices", result)
+            self.assertIn(device, result["devices"])
+            self.assertIn(device2, result["devices"])
+
+    def test_batch_command_execute_method(self):
+        org = self._get_org()
+        empty_org = self._create_org(name="empty-org", slug="empty-org")
+
+        with self.subTest("execute with no devices"):
+            with self.assertRaises(ValidationError) as ctx:
+                BatchCommand.execute(
+                    organization=empty_org,
+                    type="custom",
+                    input={"command": "echo test"},
+                    label="test-label",
+                )
+            self.assertIn(
+                "No devices match",
+                str(ctx.exception),
+            )
+
+        cred = self._create_credentials(
+            name="exec-cred",
+            organization=org,
+        )
+        device1 = self._create_device(
+            name="exec-dev1",
+            mac_address="00:11:22:33:44:e1",
+            organization=org,
+        )
+        self._create_config(device=device1)
+        self._create_device_connection(device=device1, credentials=cred)
+        device2 = self._create_device(
+            name="exec-dev2",
+            mac_address="00:11:22:33:44:e2",
+            organization=org,
+        )
+        self._create_config(device=device2)
+        self._create_device_connection(device=device2, credentials=cred)
+        group = DeviceGroup.objects.create(name="exec-group", organization=org)
+        device1.group = group
+        device1.save()
+        location = Location.objects.create(
+            name="exec-loc",
+            type="indoor",
+            organization=org,
+        )
+        DeviceLocation.objects.create(content_object=device2, location=location)
+
+        with self.subTest("execute with explicit devices"):
+            batch = BatchCommand.execute(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+                label="test-label",
+                devices=[device1],
+            )
+            batch.create_commands()
+            self.assertEqual(batch.batch_commands.count(), 1)
+            self.assertEqual(batch.batch_commands.first().device, device1)
+
+        with self.subTest("execute with group"):
+            batch = BatchCommand.execute(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+                label="test-label",
+                group=group,
+            )
+            batch.create_commands()
+            self.assertEqual(batch.batch_commands.count(), 1)
+            self.assertEqual(batch.batch_commands.first().device, device1)
+
+        with self.subTest("execute with location"):
+            batch = BatchCommand.execute(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+                label="test-label",
+                location=location,
+            )
+            batch.create_commands()
+            self.assertEqual(batch.batch_commands.count(), 1)
+            self.assertEqual(batch.batch_commands.first().device, device2)
+
+        with self.subTest("execute with group and location"):
+            DeviceLocation.objects.create(content_object=device1, location=location)
+            batch = BatchCommand.execute(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+                label="test-label",
+                group=group,
+                location=location,
+            )
+            batch.create_commands()
+            self.assertEqual(batch.batch_commands.count(), 1)
+            self.assertEqual(batch.batch_commands.first().device, device1)
+
+        with self.subTest("execute org-wide"):
+            batch = BatchCommand.execute(
+                organization=org,
+                type="custom",
+                input={"command": "echo test"},
+                label="test-label",
+            )
+            batch.create_commands()
+            self.assertEqual(batch.batch_commands.count(), 2)
+            cmd_devices = [c.device for c in batch.batch_commands.all()]
+            self.assertIn(device1, cmd_devices)
+            self.assertIn(device2, cmd_devices)
+
+    def test_batch_command_execute_org_mismatch(self):
+        org = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        device_org2 = self._create_device(
+            name="exec-mm-dev",
+            mac_address="00:11:22:33:44:99",
+            organization=org2,
+        )
+        self._create_config(device=device_org2)
+        self._create_device_connection(device=device_org2)
+
+        with self.subTest("device org mismatch"):
+            with self.assertRaises(ValidationError) as ctx:
+                BatchCommand.execute(
+                    organization=org,
+                    type="custom",
+                    input={"command": "echo test"},
+                    label="test-label",
+                    devices=[device_org2],
+                )
+            self.assertIn("devices", ctx.exception.message_dict)
+            self.assertIn(
+                "must belong to the same organization",
+                ctx.exception.message_dict["devices"][0],
+            )
+
+        with self.subTest("group org mismatch"):
+            group_org2 = DeviceGroup.objects.create(
+                name="exec-mm-group",
+                organization=org2,
+            )
+            with self.assertRaises(ValidationError) as ctx:
+                BatchCommand.execute(
+                    organization=org,
+                    type="custom",
+                    input={"command": "echo test"},
+                    label="test-label",
+                    group=group_org2,
+                )
+            self.assertIn("group", ctx.exception.message_dict)
+            self.assertIn(
+                "Please ensure that the organization of this Batch command "
+                "and the organization of the related Device Group match",
+                ctx.exception.message_dict["group"][0],
+            )
+
+        with self.subTest("location org mismatch"):
+            location_org2 = Location.objects.create(
+                name="exec-mm-loc",
+                type="indoor",
+                organization=org2,
+            )
+            with self.assertRaises(ValidationError) as ctx:
+                BatchCommand.execute(
+                    organization=org,
+                    type="custom",
+                    input={"command": "echo test"},
+                    label="test-label",
+                    location=location_org2,
+                )
+            self.assertIn("location", ctx.exception.message_dict)
+            self.assertIn(
+                "Please ensure that the organization of this Batch command "
+                "and the organization of the related location match",
+                ctx.exception.message_dict["location"][0],
+            )
+
+    def test_batch_command_dry_run_org_mismatch(self):
+        org = self._get_org()
+        org2 = self._create_org(name="org2", slug="org2")
+        device_org2 = self._create_device(
+            name="dry-mm-dev",
+            mac_address="00:11:22:33:44:98",
+            organization=org2,
+        )
+
+        with self.subTest("device org mismatch"):
+            with self.assertRaises(ValidationError) as ctx:
+                BatchCommand.dry_run(
+                    organization=org,
+                    devices=[device_org2],
+                )
+            self.assertIn("devices", ctx.exception.message_dict)
+            self.assertIn(
+                "must belong to the same organization",
+                ctx.exception.message_dict["devices"][0],
+            )
+
+        with self.subTest("group org mismatch"):
+            group_org2 = DeviceGroup.objects.create(
+                name="dry-mm-group",
+                organization=org2,
+            )
+            with self.assertRaises(ValidationError) as ctx:
+                BatchCommand.dry_run(
+                    organization=org,
+                    type="custom",
+                    input={"command": "echo test"},
+                    group=group_org2,
+                )
+            self.assertIn("group", ctx.exception.message_dict)
+            self.assertIn(
+                "Please ensure that the organization of this Batch command "
+                "and the organization of the related Device Group match",
+                ctx.exception.message_dict["group"][0],
+            )
+
+        with self.subTest("location org mismatch"):
+            location_org2 = Location.objects.create(
+                name="dry-mm-loc",
+                type="indoor",
+                organization=org2,
+            )
+            with self.assertRaises(ValidationError) as ctx:
+                BatchCommand.dry_run(
+                    organization=org,
+                    type="custom",
+                    input={"command": "echo test"},
+                    location=location_org2,
+                )
+            self.assertIn("location", ctx.exception.message_dict)
+            self.assertIn(
+                "Please ensure that the organization of this Batch command "
+                "and the organization of the related location match",
+                ctx.exception.message_dict["location"][0],
+            )
+
+    def test_batch_command_create_commands_idempotent(self):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+        self._create_device_connection(device=device)
+        batch = self._create_batch_command(
+            organization=org,
+            devices=[device],
+        )
+        batch.create_commands()
+        self.assertEqual(Command.objects.filter(batch_command=batch).count(), 1)
+        batch.create_commands()
+        self.assertEqual(
+            Command.objects.filter(batch_command=batch).count(),
+            1,
+            "create_commands must be idempotent",
+        )
+
+    def test_batch_command_calculate_and_update_status(self):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+        dc = self._create_device_connection(device=device)
+        batch = self._create_batch_command(organization=org)
+
+        with self.subTest("no commands shows idle"):
+            batch.status = "in-progress"
+            batch.save(update_fields=["status"])
+            batch.calculate_and_update_status()
+            batch.refresh_from_db()
+            self.assertEqual(batch.status, "idle")
+
+        with self.subTest("all in-progress shows in-progress"):
+            Command.objects.create(
+                batch_command=batch,
+                device=device,
+                connection=dc,
+                type=batch.type,
+                input={"command": "echo test"},
+                status="in-progress",
+            )
+            batch.calculate_and_update_status()
+            batch.refresh_from_db()
+            self.assertEqual(batch.status, "in-progress")
+
+        with self.subTest("some failed shows failed"):
+            Command.objects.filter(batch_command=batch).update(status="success")
+            device2 = self._create_device(
+                name="device2",
+                mac_address="00:11:22:33:44:02",
+                organization=org,
+            )
+            self._create_config(device=device2)
+            dc2 = self._create_device_connection(
+                device=device2,
+                credentials=self._create_credentials(
+                    name="Test credentials 2",
+                    organization=org,
+                ),
+            )
+            Command.objects.create(
+                batch_command=batch,
+                device=device2,
+                connection=dc2,
+                type=batch.type,
+                input={"command": "echo test"},
+                status="failed",
+            )
+            batch.calculate_and_update_status()
+            batch.refresh_from_db()
+            self.assertEqual(batch.status, "failed")
+
+        with self.subTest("all success shows success"):
+            batch2 = self._create_batch_command(organization=org)
+            Command.objects.create(
+                batch_command=batch2,
+                device=device,
+                connection=dc,
+                type=batch2.type,
+                input={"command": "echo test"},
+                status="success",
+            )
+            batch2.calculate_and_update_status()
+            batch2.refresh_from_db()
+            self.assertEqual(batch2.status, "success")
+
+        with self.subTest("no change shows no extra save"):
+            initial_modified = batch2.modified
+            batch2.calculate_and_update_status()
+            batch2.refresh_from_db()
+            self.assertEqual(batch2.modified, initial_modified)
+
+    def test_batch_command_permissions(self):
+        ct = ContentType.objects.get_by_natural_key(
+            app_label=self.app_label, model="batchcommand"
+        )
+        operator_group = Group.objects.get(name="Operator")
+        admin_group = Group.objects.get(name="Administrator")
+        operator_permissions = operator_group.permissions.filter(content_type=ct)
+        admin_permissions = admin_group.permissions.filter(content_type=ct)
+
+        with self.subTest("operator permissions"):
+            self.assertEqual(operator_permissions.count(), 2)
+            self.assertTrue(
+                operator_permissions.filter(codename="add_batchcommand").exists()
+            )
+            self.assertTrue(
+                operator_permissions.filter(codename="view_batchcommand").exists()
+            )
+
+        with self.subTest("administrator permissions"):
+            self.assertEqual(admin_permissions.count(), 4)
+
 
 class TestModelsTransaction(BaseTestModels, TransactionTestCase):
     def _prepare_conf_object(self, organization=None):
diff --git a/openwisp_controller/connection/tests/test_selenium.py b/openwisp_controller/connection/tests/test_selenium.py
index e75af19d3..644b0154f 100644
--- a/openwisp_controller/connection/tests/test_selenium.py
+++ b/openwisp_controller/connection/tests/test_selenium.py
@@ -1,3 +1,5 @@
+from time import sleep
+
 from django.contrib.staticfiles.testing import StaticLiveServerTestCase
 from django.test import tag
 from django.urls import reverse
@@ -53,11 +55,9 @@ def test_command_widget_on_device(self):
             by=By.CSS_SELECTOR, value='button.ow-command-btn[data-command="reboot"]'
         ).click()
         self.find_element(by=By.CSS_SELECTOR, value="#ow-command-confirm-yes").click()
-
-        self.assertEqual(Command.objects.count(), 1)
-        # TODO: Selenium tests do not support websocket connections.
-        # Thus, we need to refresh the page. Remove this when support for
-        # websockets is added.
+        # Wait for the redirect triggered by command submission to complete.
+        # Navigating away immediately can race with the redirect
+        sleep(0.3)
         self.open(path)
         self.wait_for_visibility(
             By.CSS_SELECTOR,
diff --git a/openwisp_controller/connection/tests/test_tasks.py b/openwisp_controller/connection/tests/test_tasks.py
index 587c08209..68acf8b62 100644
--- a/openwisp_controller/connection/tests/test_tasks.py
+++ b/openwisp_controller/connection/tests/test_tasks.py
@@ -16,6 +16,7 @@
 Command = load_model("connection", "Command")
 DeviceConnection = load_model("connection", "DeviceConnection")
 OrganizationConfigSettings = load_model("config", "OrganizationConfigSettings")
+BatchCommand = load_model("connection", "BatchCommand")
 
 
 class TestTasks(CreateConnectionsMixin, TestCase):
@@ -315,3 +316,165 @@ def _delete_then_raise(self):
         with mock.patch.object(Command, "execute", _delete_then_raise):
             tasks.launch_command(command.pk)
         self.assertFalse(Command.objects.filter(pk=command.pk).exists())
+
+    @mock.patch("logging.Logger.warning")
+    def test_launch_batch_command_skips_deleted_batch(self, mocked_warning):
+        batch_id = uuid.uuid4()
+        tasks.launch_batch_command(batch_id=batch_id)
+        mocked_warning.assert_called_with(
+            f"The BatchCommand object with id {batch_id} has been deleted"
+        )
+
+    @mock.patch("openwisp_controller.connection.tasks.launch_command.delay")
+    def test_launch_batch_command_creates_commands(self, mocked_delay):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+        self._create_device_connection(device=device)
+        batch = BatchCommand(
+            organization=org,
+            type="custom",
+            input={"command": "echo 'test'"},
+            label="test-label",
+        )
+        batch.full_clean()
+        batch.save()
+        batch.devices.set([device])
+        tasks.launch_batch_command(batch_id=batch.pk)
+        batch.refresh_from_db()
+        self.assertEqual(batch.batch_commands.count(), 1)
+        command = batch.batch_commands.first()
+        self.assertEqual(command.device, device)
+        self.assertEqual(command.type, batch.type)
+        self.assertEqual(command.status, "in-progress")
+        mocked_delay.assert_called_once_with(command.pk)
+
+    @mock.patch("openwisp_controller.connection.tasks.launch_command.delay")
+    def test_launch_batch_command_creates_commands_for_multiple_devices(
+        self, mocked_delay
+    ):
+        org = self._get_org()
+        cred = self._create_credentials(organization=org, name="Multi device cred")
+        devices = []
+        for i in range(2):
+            d = self._create_device(
+                name=f"task-dev-{i}",
+                mac_address=f"00:11:22:33:44:{i + 0x50:02x}",
+                organization=org,
+            )
+            self._create_config(device=d)
+            self._create_device_connection(device=d, credentials=cred)
+            devices.append(d)
+        batch = BatchCommand(
+            organization=org,
+            type="custom",
+            input={"command": "echo test"},
+            label="test-label",
+        )
+        batch.full_clean()
+        batch.save()
+        batch.devices.set(devices)
+        tasks.launch_batch_command(batch_id=batch.pk)
+        batch.refresh_from_db()
+        self.assertEqual(batch.batch_commands.count(), 2)
+        cmd_devices = [c.device for c in batch.batch_commands.all()]
+        for d in devices:
+            self.assertIn(d, cmd_devices)
+        self.assertEqual(batch.status, "in-progress")
+        self.assertEqual(mocked_delay.call_count, 2)
+
+    @mock.patch(
+        "openwisp_controller.connection.base.models.AbstractBatchCommand"
+        ".create_commands",
+        side_effect=SoftTimeLimitExceeded(),
+    )
+    def test_launch_batch_command_timeout(self, mocked_create_commands):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+        batch = BatchCommand(
+            organization=org,
+            type="custom",
+            input={"command": "echo test"},
+            label="test-label",
+        )
+        batch.full_clean()
+        batch.save()
+        tasks.launch_batch_command(batch_id=batch.pk)
+        batch.refresh_from_db()
+        self.assertEqual(batch.status, "failed")
+
+    @mock.patch(
+        "openwisp_controller.connection.base.models.AbstractBatchCommand"
+        ".create_commands",
+        side_effect=RuntimeError("test error"),
+    )
+    def test_launch_batch_command_exception(self, mocked_create_commands):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+        batch = BatchCommand(
+            organization=org,
+            type="custom",
+            input={"command": "echo test"},
+            label="test-label",
+        )
+        batch.full_clean()
+        batch.save()
+        with redirect_stderr(StringIO()) as stderr:
+            tasks.launch_batch_command(batch_id=batch.pk)
+            self.assertIn(
+                f"An exception was raised while executing batch " f"command {batch.pk}",
+                stderr.getvalue(),
+            )
+        batch.refresh_from_db()
+        self.assertEqual(batch.status, "failed")
+
+    @mock.patch("openwisp_controller.connection.tasks.launch_command.delay")
+    def test_launch_batch_command_all_devices_skipped(self, mocked_delay):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+        device.deactivate()
+        device.config.set_status_deactivated()
+        batch = BatchCommand(
+            organization=org,
+            type="custom",
+            input={"command": "echo test"},
+            label="test-label",
+        )
+        batch.full_clean()
+        batch.save()
+        batch.devices.set([device])
+        tasks.launch_batch_command(batch_id=batch.pk)
+        batch.refresh_from_db()
+        self.assertEqual(batch.batch_commands.count(), 0)
+        self.assertIn(str(device.pk), batch.skipped_devices)
+        self.assertEqual(batch.status, "idle")
+        mocked_delay.assert_not_called()
+
+    @mock.patch("openwisp_controller.connection.tasks.launch_command.delay")
+    def test_launch_batch_command_already_processed(self, mocked_delay):
+        org = self._get_org()
+        device = self._create_device(organization=org)
+        self._create_config(device=device)
+        self._create_device_connection(device=device)
+        batch = BatchCommand(
+            organization=org,
+            type="custom",
+            input={"command": "echo test"},
+            label="test-label",
+        )
+        batch.full_clean()
+        batch.save()
+        batch.devices.set([device])
+        # First call — creates commands
+        tasks.launch_batch_command(batch_id=batch.pk)
+        batch.refresh_from_db()
+        self.assertEqual(batch.batch_commands.count(), 1)
+        first_call_count = mocked_delay.call_count
+        # Second call — should be a no-op (idempotency guard)
+        tasks.launch_batch_command(batch_id=batch.pk)
+        batch.refresh_from_db()
+        self.assertEqual(batch.batch_commands.count(), 1)
+        self.assertEqual(mocked_delay.call_count, first_call_count)
diff --git a/openwisp_controller/geo/estimated_location/tests/tests.py b/openwisp_controller/geo/estimated_location/tests/tests.py
index c037106dd..fee63ae94 100644
--- a/openwisp_controller/geo/estimated_location/tests/tests.py
+++ b/openwisp_controller/geo/estimated_location/tests/tests.py
@@ -566,7 +566,7 @@ def _verify_location_details(device, mocked_response):
             device2.save()
             # 3 queries related to notifications cleanup
             device2.refresh_from_db()
-            with self.assertNumQueries(15):
+            with self.assertNumQueries(16):
                 manage_estimated_locations(device2.pk, device2.last_ip)
             mock_info.assert_called_once_with(
                 f"Estimated location saved successfully for {device2.pk}"
diff --git a/openwisp_controller/geo/tests/test_api.py b/openwisp_controller/geo/tests/test_api.py
index 156550692..df46460cf 100644
--- a/openwisp_controller/geo/tests/test_api.py
+++ b/openwisp_controller/geo/tests/test_api.py
@@ -694,7 +694,7 @@ def test_change_location_type_to_outdoor_api(self):
     def test_delete_location_detail(self):
         l1 = self._create_location()
         path = reverse("geo_api:detail_location", args=[l1.pk])
-        with self.assertNumQueries(5):
+        with self.assertNumQueries(6):
             response = self.client.delete(path)
         self.assertEqual(response.status_code, 204)
 
diff --git a/tests/openwisp2/sample_connection/api/views.py b/tests/openwisp2/sample_connection/api/views.py
index fd2207cbc..22d04a924 100644
--- a/tests/openwisp2/sample_connection/api/views.py
+++ b/tests/openwisp2/sample_connection/api/views.py
@@ -1,3 +1,12 @@
+from openwisp_controller.connection.api.views import (
+    BatchCommandDetailView as BaseBatchCommandDetailView,
+)
+from openwisp_controller.connection.api.views import (
+    BatchCommandExecuteView as BaseBatchCommandExecuteView,
+)
+from openwisp_controller.connection.api.views import (
+    BatchCommandListView as BaseBatchCommandListView,
+)
 from openwisp_controller.connection.api.views import (
     CommandDetailsView as BaseCommandDetailsView,
 )
@@ -42,9 +51,24 @@ class DeviceConnectionDetailView(BaseDeviceConnectionDetailView):
     pass
 
 
+class BatchCommandExecuteView(BaseBatchCommandExecuteView):
+    pass
+
+
+class BatchCommandListView(BaseBatchCommandListView):
+    pass
+
+
+class BatchCommandDetailView(BaseBatchCommandDetailView):
+    pass
+
+
 command_list_create_view = CommandListCreateView.as_view()
 command_details_view = CommandDetailsView.as_view()
 credential_list_create_view = CredentialListCreateView.as_view()
 credential_detail_view = CredentialDetailView.as_view()
 deviceconnection_list_create_view = DeviceConnectionListCreateView.as_view()
 deviceconnection_detail_view = DeviceConnectionDetailView.as_view()
+batch_command_execute_view = BatchCommandExecuteView.as_view()
+batch_command_list_view = BatchCommandListView.as_view()
+batch_command_detail_view = BatchCommandDetailView.as_view()
diff --git a/tests/openwisp2/sample_connection/migrations/0005_batchcommand_command_batch_command.py b/tests/openwisp2/sample_connection/migrations/0005_batchcommand_command_batch_command.py
new file mode 100644
index 000000000..dd03617cc
--- /dev/null
+++ b/tests/openwisp2/sample_connection/migrations/0005_batchcommand_command_batch_command.py
@@ -0,0 +1,176 @@
+# Generated by Django 5.2.15 on 2026-06-15 18:15
+
+import uuid
+
+import django
+import django.core.serializers.json
+import django.db.migrations.operations.special
+import django.db.models.deletion
+import django.utils.timezone
+import model_utils.fields
+from django.db import migrations, models
+
+from openwisp_controller import connection as connection_config
+from openwisp_controller.connection.migrations import (
+    assign_batchcommand_permissions_to_groups,
+)
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("sample_config", "0009_replace_jsonfield_with_django_builtin"),
+        ("sample_connection", "0004_replace_jsonfield_with_django_builtin"),
+        ("sample_geo", "0005_organizationgeosettings"),
+        ("sample_users", "0005_user_expiration_date_user_user_active_expiry_idx"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="BatchCommand",
+            fields=[
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                (
+                    "created",
+                    model_utils.fields.AutoCreatedField(
+                        default=django.utils.timezone.now,
+                        editable=False,
+                        verbose_name="created",
+                    ),
+                ),
+                (
+                    "modified",
+                    model_utils.fields.AutoLastModifiedField(
+                        default=django.utils.timezone.now,
+                        editable=False,
+                        verbose_name="modified",
+                    ),
+                ),
+                (
+                    "status",
+                    models.CharField(
+                        choices=[
+                            ("idle", "idle"),
+                            ("in-progress", "in progress"),
+                            ("success", "success"),
+                            ("failed", "failed"),
+                        ],
+                        default="idle",
+                        max_length=12,
+                    ),
+                ),
+                (
+                    "type",
+                    models.CharField(
+                        max_length=16,
+                        choices=(
+                            connection_config.commands.COMMAND_CHOICES
+                            if django.VERSION < (5, 0)
+                            else connection_config.commands.get_command_choices
+                        ),
+                    ),
+                ),
+                (
+                    "input",
+                    models.JSONField(
+                        blank=True,
+                        encoder=django.core.serializers.json.DjangoJSONEncoder,
+                        null=True,
+                    ),
+                ),
+                (
+                    "devices",
+                    models.ManyToManyField(
+                        blank=True, to="sample_config.device", verbose_name="devices"
+                    ),
+                ),
+                (
+                    "skipped_devices",
+                    models.JSONField(
+                        blank=True,
+                        null=True,
+                        default=dict,
+                        verbose_name="Skipped devices",
+                        help_text=(
+                            "Maps device UUIDs to validation error messages for "
+                            "devices that were skipped during command creation."
+                        ),
+                    ),
+                ),
+                (
+                    "group",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        to="sample_config.devicegroup",
+                        verbose_name="device group",
+                    ),
+                ),
+                (
+                    "location",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.SET_NULL,
+                        to="sample_geo.location",
+                        verbose_name="location",
+                    ),
+                ),
+                (
+                    "label",
+                    models.CharField(
+                        max_length=64,
+                        verbose_name="label",
+                        help_text="A short label to identify this batch command.",
+                    ),
+                ),
+                (
+                    "notes",
+                    models.TextField(
+                        blank=True,
+                        default="",
+                        verbose_name="notes",
+                        help_text="Optional notes about this batch command.",
+                    ),
+                ),
+                (
+                    "organization",
+                    models.ForeignKey(
+                        blank=True,
+                        null=True,
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to="sample_users.organization",
+                    ),
+                ),
+            ],
+            options={
+                "verbose_name": "Batch command",
+                "verbose_name_plural": "Batch commands",
+                "abstract": False,
+            },
+        ),
+        migrations.AddField(
+            model_name="command",
+            name="batch_command",
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.SET_NULL,
+                related_name="batch_commands",
+                to="sample_connection.batchcommand",
+            ),
+        ),
+        migrations.RunPython(
+            code=assign_batchcommand_permissions_to_groups,
+            reverse_code=django.db.migrations.operations.special.RunPython.noop,
+        ),
+    ]
diff --git a/tests/openwisp2/sample_connection/models.py b/tests/openwisp2/sample_connection/models.py
index 7964c5471..4e96065e8 100644
--- a/tests/openwisp2/sample_connection/models.py
+++ b/tests/openwisp2/sample_connection/models.py
@@ -1,6 +1,7 @@
 from django.db import models
 
 from openwisp_controller.connection.base.models import (
+    AbstractBatchCommand,
     AbstractCommand,
     AbstractCredentials,
     AbstractDeviceConnection,
@@ -27,3 +28,8 @@ class Meta(AbstractDeviceConnection.Meta):
 class Command(AbstractCommand):
     class Meta(AbstractCommand.Meta):
         abstract = False
+
+
+class BatchCommand(AbstractBatchCommand):
+    class Meta(AbstractBatchCommand.Meta):
+        abstract = False
diff --git a/tests/openwisp2/settings.py b/tests/openwisp2/settings.py
index c45eb5537..0e27adfde 100644
--- a/tests/openwisp2/settings.py
+++ b/tests/openwisp2/settings.py
@@ -293,6 +293,7 @@
     CONNECTION_CREDENTIALS_MODEL = "sample_connection.Credentials"
     CONNECTION_DEVICECONNECTION_MODEL = "sample_connection.DeviceConnection"
     CONNECTION_COMMAND_MODEL = "sample_connection.Command"
+    CONNECTION_BATCHCOMMAND_MODEL = "sample_connection.BatchCommand"
     SUBNET_DIVISION_SUBNETDIVISIONRULE_MODEL = (
         "sample_subnet_division.SubnetDivisionRule"
     )